diff --git a/CLAUDE.md b/CLAUDE.md index 892a799..9ad568a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -115,6 +115,7 @@ Every one of these has already cost someone an hour: | Peak customers means the most off while the outage was live | `notes/grading.md` § The peak is the highest count | | Part-observed days keep their colour and say so in the tooltip | `notes/grading.md` § Short days say so | | 2.5M customer denominator, and which DAPR figures are comparable | `notes/grading.md` § The customer denominator | +| Each run logs two lines to `runs-*.jsonl`: the list before any detail is fetched, and an `"event": "end"` line (`Store.finish_run`) carrying what a rebuild cannot derive - status, exit code, finish time, skipped count, errors. Older runs have no end line and replay from the start line alone | `notes/storms.md` § The run's end is logged after all (2026-09-24) | | The collector pings `ESB_HEARTBEAT_URL` after every run that reached the feed, drifted or partial ones included, and never after a rejected key, an unreachable feed or a skipped trigger; a dead-man's monitor alerts on silence (period 30 min, grace 90). `test-alert` proves both channels | `notes/alerting.md` § The heartbeat (2026-09-06) | | A storm can list more than a run can fetch: `ids_needing_detail` ranks by what a purge would take (listed Restored and still needing a fetch, then live never-fetched, then re-checks), every detail is committed as it lands, a run stops itself at `RUN_BUDGET_S` (24 min) and records `cut_short` (exit 0, heartbeat sent, no webhook), SIGTERM from the service unit's backstop ends the loop the same way, and the pause is 500 ms. In list order a cut-short run re-fetched the same head every run and never reached the tail | `notes/storms.md` (2026-09-06) | | The Pi pushes every six hours and `esb-data` dispatches the site build on each push; the two crons are a fallback only, because scheduled runs here have landed 4-10h behind their cron time. The stale banner trips at 10h - above the widest legitimate age (~7h), below a missed push (13h+) | `notes/publish-cadence.md`; `STALE_AFTER` in `esb_site/render.py` | diff --git a/esb_outages/__main__.py b/esb_outages/__main__.py index 1f1c95d..c26fc62 100644 --- a/esb_outages/__main__.py +++ b/esb_outages/__main__.py @@ -5,10 +5,11 @@ import argparse import os import sys +from pathlib import Path from . import __version__, alert from .client import EsbClient -from .poll import DEFAULT_DELAY_MS, run_check, run_poll +from .poll import DEFAULT_DELAY_MS, poll_lock, run_check, run_poll from .store import Store DEFAULT_DATA_DIR = os.environ.get("ESB_DATA_DIR", "/data") @@ -103,17 +104,29 @@ def cmd_test_alert(args) -> int: return alert.EXIT_OK +def _held_by_poll() -> int: + # Both delete files a poll writes to: esb.db and its journal, or the log. + print("a poll run holds the lock; try again when it has finished", file=sys.stderr) + return 1 + + def cmd_rebuild(args) -> int: - with Store(args.data_dir) as store: - result = store.rebuild(verbose=True) + with poll_lock(Path(args.data_dir)) as acquired: + if not acquired: + return _held_by_poll() + with Store(args.data_dir) as store: + result = store.rebuild(verbose=True) if result["runs"] == 0 and result["observations"] == 0: print("nothing to replay: no raw logs found", file=sys.stderr) return alert.EXIT_OK def cmd_compact(args) -> int: - with Store(args.data_dir) as store: - done = store.compact() + with poll_lock(Path(args.data_dir)) as acquired: + if not acquired: + return _held_by_poll() + with Store(args.data_dir) as store: + done = store.compact() print(f"compacted {len(done)} file(s): {', '.join(done) or 'none'}") return alert.EXIT_OK diff --git a/esb_outages/client.py b/esb_outages/client.py index 779a553..55e2cc3 100644 --- a/esb_outages/client.py +++ b/esb_outages/client.py @@ -9,12 +9,14 @@ from __future__ import annotations import gzip +import http.client import json import os import random import time import urllib.error import urllib.request +import zlib BASE_URL = "https://api.esb.ie/esbn/powercheck/v1.0" @@ -122,6 +124,12 @@ def _request(self, url: str) -> str: raise ApiError(f"{exc.code} from {url}: {body}") from exc except (urllib.error.URLError, TimeoutError, OSError) as exc: raise TransientError(f"network failure for {url}: {exc}") from exc + # A body cut off mid-read: IncompleteRead is no OSError, and a + # truncated gzip stream raises EOFError or zlib.error from decompress. + except (http.client.HTTPException, EOFError, zlib.error) as exc: + raise TransientError(f"incomplete response from {url}: {exc}") from exc + except UnicodeDecodeError as exc: + raise ApiError(f"undecodable body from {url}: {exc}") from exc def get_json(self, path: str) -> dict: """GET a path, retrying transient failures with exponential backoff.""" diff --git a/esb_outages/parse.py b/esb_outages/parse.py index 2d2709e..32f59b7 100644 --- a/esb_outages/parse.py +++ b/esb_outages/parse.py @@ -48,14 +48,15 @@ def parse_esb_datetime(value: str | None) -> tuple[str | None, bool]: time is ambiguous or impossible because of a DST transition: - Fall back (last Sunday of October): 01:00-01:59 happens twice. We take the - first occurrence (fold=0), so the value may be an hour late. + first occurrence (fold=0, still UTC+1), so a time that was really the + second is stored an hour early. - Spring forward (last Sunday of March): 01:00-01:59 never happens. Python resolves it, but the result is a fiction. Either way the raw string is retained in the database, so a flagged row can be revisited rather than silently trusted. """ - if not value or not value.strip(): + if not isinstance(value, str) or not value.strip(): return None, False try: naive = datetime.strptime(value.strip(), ESB_DATETIME_FORMAT) @@ -105,6 +106,8 @@ def parse_point(point) -> tuple[float | None, float | None, str | None]: def check_detail_schema(body: dict) -> list[str]: """Return human-readable descriptions of any drift from the known shape.""" + if not isinstance(body, dict): + return ["detail response is not an object"] keys = set(body) problems = [] unexpected = sorted(keys - DETAIL_FIELDS) @@ -118,6 +121,8 @@ def check_detail_schema(body: dict) -> list[str]: def check_list_schema(body: dict) -> list[str]: problems = [] + if not isinstance(body, dict): + return ["response is not an object"] if "outageMessage" not in body: return ["response has no 'outageMessage' key"] items = body["outageMessage"] @@ -184,8 +189,8 @@ def normalize_detail(body: dict) -> dict: "status_message": _text(body.get("statusMessage")), "planned_outage_reason": _text(body.get("plannedOutageReason")), # An outage is only immutable once it is restored *and* carries the - # actual restore time; "Restored" with an empty restoreTime is still - # settling and must be re-fetched. - "is_final": int(outage_type == "Restored" and restore_raw is not None), + # actual restore time; "Restored" with an empty or unreadable + # restoreTime is still settling and must be re-fetched. + "is_final": int(outage_type == "Restored" and restore_utc is not None), "tz_ambiguous": int(start_amb or est_amb or restore_amb), } diff --git a/esb_outages/poll.py b/esb_outages/poll.py index fbfe0d0..6176a4f 100644 --- a/esb_outages/poll.py +++ b/esb_outages/poll.py @@ -144,15 +144,15 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_ list_body = client.get_outage_list() except AuthError as exc: store.write_run_raw(run_id, started_at, 401, None, status="auth_error") - store.record_run( + store.finish_run( run_id=run_id, started_at_utc=started_at, finished_at_utc=utc_now_iso(), status="auth_error", exit_code=alert.EXIT_AUTH, n_errors=1, error_summary=str(exc), ) return alert.fail(alert.auth_banner(client.masked_key, str(exc)), alert.EXIT_AUTH) - except (TransientError, ApiError) as exc: + except (TransientError, ApiError, NotFound) as exc: store.write_run_raw(run_id, started_at, 0, None, status="unreachable") - store.record_run( + store.finish_run( run_id=run_id, started_at_utc=started_at, finished_at_utc=utc_now_iso(), status="unreachable", exit_code=alert.EXIT_UNREACHABLE, n_errors=1, error_summary=str(exc), @@ -163,7 +163,7 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_ store.write_run_raw(run_id, started_at, 200, list_body) drift = check_list_schema(list_body) - items = list_body.get("outageMessage") or [] + items = (list_body.get("outageMessage") if isinstance(list_body, dict) else None) or [] if not isinstance(items, list): items = [] @@ -199,12 +199,13 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_ except AuthError as exc: # The key died mid-run. Stop immediately rather than burning # through hundreds of guaranteed-failing requests. - store.record_run( + store.finish_run( run_id=run_id, started_at_utc=started_at, finished_at_utc=utc_now_iso(), status="auth_error", exit_code=alert.EXIT_AUTH, n_listed=len(listed_ids), n_detail_fetched=fetched, n_detail_skipped=skipped, - n_errors=1, error_summary=str(exc), + n_errors=len(errors) + 1, + error_summary="; ".join(errors[:9] + [str(exc)]), ) return alert.fail( alert.auth_banner(client.masked_key, str(exc)), alert.EXIT_AUTH @@ -220,6 +221,8 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_ drift.extend( f"outage {outage_id}: {p}" for p in check_detail_schema(body) ) + if not isinstance(body, dict): + continue store.apply_detail(observed_at, normalize_detail(body)) # Committed per detail so a run killed outright still leaves the # database knowing what it fetched; the next run then carries on @@ -248,7 +251,7 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_ else: status, code = "ok", alert.EXIT_OK - store.record_run( + store.finish_run( run_id=run_id, started_at_utc=started_at, finished_at_utc=utc_now_iso(), status=status, exit_code=code, n_listed=len(listed_ids), n_detail_fetched=fetched, n_detail_skipped=skipped, n_errors=failed, @@ -284,7 +287,7 @@ def run_check(client: EsbClient | None = None) -> int: body = client.get_outage_list() except AuthError as exc: return alert.fail(alert.auth_banner(client.masked_key, str(exc)), alert.EXIT_AUTH) - except (TransientError, ApiError) as exc: + except (TransientError, ApiError, NotFound) as exc: return alert.fail(alert.unreachable_banner(str(exc)), alert.EXIT_UNREACHABLE) problems = check_list_schema(body) diff --git a/esb_outages/store.py b/esb_outages/store.py index f8f977c..ec2098e 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -11,9 +11,11 @@ from __future__ import annotations import gzip -import io +import hashlib import json import os +import re +import shutil import sqlite3 import sys from datetime import UTC, datetime @@ -117,6 +119,13 @@ """ +_FILE_MONTH = re.compile(r"-(\d{4}-\d{2})") +_UTC_STAMP = re.compile(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ") + +# What a run's end record carries: everything a rebuild cannot derive from the +# list and the observations. +RUN_END_FIELDS = ("status", "exit_code", "n_detail_skipped", "n_errors", "error_summary") + # How long an outage must go unchanged before it is treated as dormant, and how # often to re-check it once it is. Expressed in hours rather than run counts so # the behaviour does not shift if the poll interval changes. @@ -140,14 +149,25 @@ def _hours_between(earlier: str | None, later: str) -> float: return delta.total_seconds() / 3600.0 +def _fsync_dir(path: Path) -> None: + """Make a rename in `path` durable before anything relies on it.""" + fd = os.open(path, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + + def _month_of(iso_ts: str) -> str: return iso_ts[:7] def _open_maybe_gzip(path: Path): + # Bytes, decoded a line at a time in iter_raw: a line torn inside a fada + # or corrupted on the card is one malformed line, never a changed record. if path.suffix == ".gz": - return io.TextIOWrapper(gzip.open(path, "rb"), encoding="utf-8") - return path.open("r", encoding="utf-8") + return gzip.open(path, "rb") + return path.open("rb") class Store: @@ -197,8 +217,15 @@ def conn(self) -> sqlite3.Connection: def _append_raw(self, kind: str, month: str, record: dict) -> None: path = self.raw_dir / f"{kind}-{month}.jsonl" - with path.open("a", encoding="utf-8") as fh: - fh.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n") + line = json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n" + with path.open("a+b") as fh: + # A line torn by a power cut would swallow this one too, and a run's + # start line is its only copy of the list. + if fh.seek(0, os.SEEK_END): + fh.seek(-1, os.SEEK_END) + if fh.read(1) != b"\n": + line = "\n" + line + fh.write(line.encode("utf-8")) fh.flush() # The point of this file is to survive the NAS losing power mid-run. os.fsync(fh.fileno()) @@ -225,9 +252,31 @@ def write_run_raw( "list_status": list_status, "list_body": list_body, "status": status, + # Says an end line follows, so a rebuild can tell a run that + # died from one logged before end lines existed. + "ends_logged": True, }, ) + def finish_run(self, **fields) -> None: + """Log how a run ended, then record it. + + The start line is written before the run knows how it will end, so + without this a rebuild restored every cut-short, partial or drifted + run as "ok". Only what a rebuild cannot derive is logged. + """ + self._append_raw( + "runs", + _month_of(fields["finished_at_utc"]), + { + "event": "end", + "run_id": fields["run_id"], + "finished_at": fields["finished_at_utc"], + **{k: fields.get(k) for k in RUN_END_FIELDS}, + }, + ) + self.record_run(**fields) + def write_observation_raw( self, run_id: str, observed_at: str, outage_id: str, http_status: int, body ) -> None: @@ -257,15 +306,30 @@ def iter_raw(self, kind: str): by a backup, and neither is a reason to refuse to read the other 99.99%. Skips are counted and reported rather than passing silently. """ + # Every line carries its own timestamps and run id, so an identical one + # is the same record read twice: from a `sort -u`-less merge, or a + # compact that crashed before removing what it had archived. + seen: set[bytes] = set() + month = None for path in self.raw_files(kind): + # Copies share a month, so the set never needs to span one; keyed + # on the month itself, a copy under another name shares it too. + found = _FILE_MONTH.search(path.name) + if (found.group(1) if found else path.name) != month: + month = found.group(1) if found else path.name + seen.clear() with _open_maybe_gzip(path) as fh: for lineno, line in enumerate(fh, 1): line = line.strip() if not line: continue + digest = hashlib.blake2b(line, digest_size=16).digest() + if digest in seen: + continue + seen.add(digest) try: - yield json.loads(line) - except json.JSONDecodeError as exc: + yield json.loads(line.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: self.malformed_lines.append(f"{path.name}:{lineno}: {exc}") # ---- applying observations ------------------------------------------ @@ -298,6 +362,10 @@ def apply_list(self, observed_at: str, items: list[dict]) -> None: from .parse import parse_point for item in items: + # Already on disk, so replay meets it too: skip it, or every + # rebuild dies on the same line of a log nobody may edit. + if not isinstance(item, dict): + continue outage_id = str(item.get("i")) list_type = item.get("t") lat, lon, point_raw = parse_point(item.get("p")) @@ -370,6 +438,9 @@ def ids_needing_detail( """ if not outage_ids: return [] + # ESB listing an id twice would fetch it twice, and the two identical + # observation lines would read as one copy (iter_raw). + outage_ids = list(dict.fromkeys(outage_ids)) now = now or utc_now_iso() placeholders = ",".join("?" * len(outage_ids)) rows = self.conn.execute( @@ -387,21 +458,27 @@ def ids_needing_detail( def urgency(outage_id: str) -> int | None: row = state.get(outage_id) if row is None: - return 1 + return 2 # apply_list has already written the list's type, so this is # whether the purge clock is running restored = row["outage_type"] == "Restored" if not row["has_detail"]: - return 0 if restored else 1 - # Cleared by apply_list on a type change: something happened. - if row["last_detail_utc"] is None: return 0 if restored else 2 if row["is_final"]: return None + if restored: + # Captured, but with no restore time yet, whether settling or + # just flipped. Still on the purge clock, behind the rank + # above: a purge here costs one field, there the whole record. + return 1 + # Cleared by apply_list on a type change, with is_final: something + # happened. + if row["last_detail_utc"] is None: + return 3 if _hours_between(row["last_change"], now) < quiet_after_hours: - return 3 # actively changing, keep watching closely + return 4 # actively changing, keep watching closely if _hours_between(row["last_detail_utc"], now) >= recheck_hours: - return 4 + return 5 return None ranked = [ @@ -513,12 +590,49 @@ def apply_observation(rec) -> int: # matters when logs from two hosts are concatenated during a migration, # where interleaved runs would otherwise replay out of order and skew # first_seen / last_seen. - run_records = sorted(self.iter_raw("runs"), key=lambda r: r["started_at"]) + run_records, ends = [], {} + for rec in self.iter_raw("runs"): + if rec.get("event") == "end": + ends[rec.get("run_id")] = rec + else: + run_records.append(rec) + + # An end line whose start line was lost still says how the run went, + # and its run id carries the start time: it replays as a run with no + # list, so its observations keep their place. + run_ids = {rec["run_id"] for rec in run_records} + lost_starts = 0 + for run_id in ends: + started = str(run_id).rsplit("-", 1)[0] + if run_id not in run_ids and _UTC_STAMP.fullmatch(started): + run_records.append({"run_id": run_id, "started_at": started}) + run_ids.add(run_id) + lost_starts += 1 + # A start line promising an end line that never came is a run that + # died, unless it is the newest: a backup can snapshot the log mid-run. + newest = max((rec["started_at"] for rec in run_records), default=None) + + # Observations whose run record never made it to disk (a crash between + # the two writes, or a damaged line) replay at their own place in time. + # Replayed after everything else, an old body rolled back newer state. + orphans = [ + (min(o.get("observed_at") or "" for o in records), records) + for run_id, records in observations.items() + if run_id not in run_ids + ] + # On a tie the orphan goes first: its run started no later than it. + timeline = sorted( + [(rec["started_at"], 1, rec, None) for rec in run_records] + + [(at, 0, None, records) for at, records in orphans], + key=lambda event: event[:2], + ) - seen_run_ids = set() - for rec in run_records: + for _, _, rec, orphaned in timeline: + if orphaned is not None: + for obs in orphaned: + n_obs += apply_observation(obs) + continue run_id = rec["run_id"] - seen_run_ids.add(run_id) body = rec.get("list_body") items = body.get("outageMessage") if isinstance(body, dict) else None if isinstance(items, list): @@ -528,37 +642,53 @@ def apply_observation(rec) -> int: # are all derivable from the raw log, and they are what tells you # whether the dormancy back-off is working. run_obs = observations.get(run_id, []) - fetched = sum(1 for o in run_obs if o.get("http_status") == 200) + fetched = sum( + 1 for o in run_obs + if o.get("http_status") == 200 and isinstance(o.get("body"), dict) + ) purged = sum(1 for o in run_obs if o.get("http_status") == 404) errors = sum( 1 for o in run_obs if o.get("http_status") not in (200, 404) ) - listed = len(items) if isinstance(items, list) else None + # as poll counts them: object items only, and a list call that + # answered 200 with an unusable body still reached the feed + if isinstance(items, list): + listed = sum(isinstance(item, dict) for item in items) + else: + listed = 0 if rec.get("list_status") == 200 else None + # Runs logged before the end record existed take the start line's + # status and derived counters, which is all a rebuild ever had. + end = ends.get(run_id, {}) self.record_run( run_id=run_id, started_at_utc=rec["started_at"], - status=rec.get("status", "ok"), + finished_at_utc=end.get("finished_at"), + status=end.get("status") or ( + # a failure the start line already logged outranks the guess + ("in_progress" if rec["started_at"] == newest else "unfinished") + if rec.get("ends_logged") and rec.get("status", "ok") == "ok" + else rec.get("status", "ok") + ), + exit_code=end.get("exit_code"), n_listed=listed, - n_detail_fetched=fetched, - n_detail_skipped=( + # NULL only for a run that never got past its list call + n_detail_fetched=fetched if listed is not None or run_obs else None, + n_detail_skipped=end["n_detail_skipped"] if "n_detail_skipped" in end else ( listed - fetched - purged - errors if listed is not None else None ), - n_errors=errors, + n_errors=end["n_errors"] if "n_errors" in end else errors, + error_summary=end.get("error_summary"), ) n_runs += 1 for obs in run_obs: n_obs += apply_observation(obs) - # Observations whose run record never made it to disk (a crash between - # the two writes). Rare, but they are still real data. - for run_id, records in observations.items(): - if run_id not in seen_run_ids: - for obs in records: - n_obs += apply_observation(obs) - self.conn.commit() if verbose: print(f"replayed {n_runs} runs and {n_obs} detail observations") + if lost_starts: + print(f" {lost_starts} run(s) had lost their start line; " + "recorded from their end line", file=sys.stderr) for problem in self.malformed_lines: print(f" skipped unreadable line {problem}", file=sys.stderr) return { @@ -645,8 +775,28 @@ def compact(self) -> list[str]: if month >= current: continue target = Path(str(path) + ".gz") - with path.open("rb") as src, gzip.open(target, "wb") as dst: - dst.write(src.read()) + staged = Path(str(target) + ".tmp") + archived = target.exists() + with staged.open("wb") as out: + # A month written to after it was compacted - a Pi with no + # clock battery boots in the past until NTP syncs - keeps + # its archive and gains the late lines as a further gzip + # member, which every gzip reader concatenates. + if archived: + with target.open("rb") as old: + shutil.copyfileobj(old, out) + with path.open("rb") as src, gzip.GzipFile(fileobj=out, mode="wb") as dst: + if archived: + # the archive may end in a torn line; a blank one is skipped + dst.write(b"\n") + shutil.copyfileobj(src, dst) + out.flush() + os.fsync(out.fileno()) + os.replace(staged, target) + _fsync_dir(self.raw_dir) + # A crash before this unlink compacts the same lines again next + # time; iter_raw drops the repeats. path.unlink() + _fsync_dir(self.raw_dir) compacted.append(target.name) return sorted(compacted) diff --git a/notes/storms.md b/notes/storms.md index f1ff1f1..b65c698 100644 --- a/notes/storms.md +++ b/notes/storms.md @@ -92,3 +92,38 @@ fewer observations rather than its status, and counts the unfetched tail as skipped, so `esb stats` on a rebuilt database overstates the back-off's "% avoided" across storm runs. Known and accepted: a second raw record kind for the run's end would change the log's shape for a number nobody grades on. + +### The run's end is logged after all (2026-09-24) + +The paragraph above accepted the gap; a retroactive review of the collector +raised it again and the owner reversed it. The gap was wider than the "% +avoided" figure: every non-`ok` outcome a run decides after its list line +(`cut_short`, `partial`, `schema_drift`, a key rejected mid-run) came back from +a rebuild as `ok`, with no finish time, exit code or error summary. On the +corpus to 24 September the raw runs said `ok` 2,470 times and nothing else, so +`esb stats` on the rebuilt database reads "runs cut short: 0" whatever +happened in a storm. + +`Store.finish_run` now appends a second line for every run, `"event": "end"`, +before writing the run row, carrying only what cannot be derived: +`finished_at`, `status`, `exit_code`, `n_detail_skipped`, `n_errors` and +`error_summary`. `rebuild` prefers it when present. Older runs have none and +replay as before, from the start line's status and derived counters. The line +is written with sorted keys like every other, so `sort -u` merges still hold, +and it goes in the month file of its own timestamp, so a run crossing midnight +on the last day ends in the next month's file. + +Two edges follow from having two lines per run. A start line now carries +`"ends_logged": true`, and one with that flag but no end line is a run that +died before closing itself out (an uncaught exception, a full disk, a kill +the SIGTERM handler never saw): a rebuild records it as `unfinished`, where +the live database has no row for it at all, rather than as a clean `ok`. The +newest such run reads `in_progress` instead, because the six-hourly backup +commits `raw/` without the poll lock and often catches a run mid-flight. +The flag, not the date of the first end line, is what decides: a merged log +from a host still on the old code, or a Pi that booted with a wrong clock, +would otherwise relabel every older-style run after it. An end line whose +start line was lost replays as a run with no list, started at the time its +run id carries, so its observations keep their place. Two copies of +the same line, from a merge without `sort -u` or a `compact` that crashed +before removing what it had archived, are read once. diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..bc813c5 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,53 @@ +import gzip +import http.client +import unittest +from unittest import mock + +from esb_outages.client import ApiError, EsbClient, TransientError + + +class _Response: + def __init__(self, read, encoding="gzip"): + self._read = read + self.headers = {"Content-Encoding": encoding} + + def read(self): + return self._read() + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + +def _raise(exc): + def read(): + raise exc + return read + + +class TestFailuresStayInTheTaxonomy(unittest.TestCase): + """Anything else escapes the run: no run record, no heartbeat, a traceback.""" + + def request(self, response): + client = EsbClient(retries=1, sleep=lambda s: None) + with mock.patch("urllib.request.urlopen", return_value=response): + return client.get_json("/outages") + + def test_a_body_cut_off_mid_read_is_transient(self): + with self.assertRaises(TransientError): + self.request(_Response(_raise(http.client.IncompleteRead(b"{")))) + + def test_a_truncated_gzip_body_is_transient(self): + cut = gzip.compress(b'{"outageMessage": []}')[:-8] + with self.assertRaises(TransientError): + self.request(_Response(lambda: cut)) + + def test_a_body_that_is_not_utf8_is_an_api_error(self): + with self.assertRaises(ApiError): + self.request(_Response(lambda: b"\xff\xfe", encoding="")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_parse.py b/tests/test_parse.py index 16b493b..183df8e 100644 --- a/tests/test_parse.py +++ b/tests/test_parse.py @@ -124,6 +124,18 @@ def test_restored_without_restore_time_is_not_final(self): body["restoreTime"] = "" self.assertEqual(normalize_detail(body)["is_final"], 0) + def test_an_unreadable_restore_time_is_not_final(self): + # final means never fetched again, so a format ESB changes must not + # freeze the outage with no restore time + body = dict(detail("restored"), restoreTime="31/10/2026 1:5 pm") + self.assertEqual(normalize_detail(body)["is_final"], 0) + + def test_a_timestamp_that_is_not_a_string_is_no_timestamp(self): + # applied live and on every rebuild, so it must not raise + body = dict(detail("fault"), startTime=202607311000, estRestoreTime=["x"]) + n = normalize_detail(body) + self.assertEqual((n["start_time_utc"], n["est_restore_time_utc"]), (None, None)) + def test_empty_strings_become_null(self): n = normalize_detail(detail("restored")) self.assertIsNone(n["status_message"]) diff --git a/tests/test_poll.py b/tests/test_poll.py index b5d2b29..889e364 100644 --- a/tests/test_poll.py +++ b/tests/test_poll.py @@ -8,7 +8,7 @@ from pathlib import Path from esb_outages import alert -from esb_outages.client import ApiError, AuthError, TransientError +from esb_outages.client import ApiError, AuthError, NotFound, TransientError from esb_outages.poll import poll_lock, run_check, run_poll from esb_outages.store import Store @@ -68,8 +68,11 @@ def test_raw_log_records_every_response(self): with self.store() as st: runs = list(st.iter_raw("runs")) obs = list(st.iter_raw("observations")) - self.assertEqual(len(runs), 1) - self.assertEqual(runs[0]["list_body"], make_list(detail("fault"), detail("restored"))) + start, end = runs + self.assertEqual(start["list_body"], make_list(detail("fault"), detail("restored"))) + self.assertEqual( + (end["event"], end["run_id"], end["status"]), ("end", start["run_id"], "ok") + ) self.assertEqual(len(obs), 2) @@ -82,6 +85,22 @@ def test_unreachable_api_exits_three(self): client = FakeClient(list_error=TransientError("connection refused")) self.assertEqual(self.poll(client), alert.EXIT_UNREACHABLE) + def test_a_404_on_the_list_is_unreachable_not_a_crash(self): + client = FakeClient(list_error=NotFound("404 for /outages")) + self.assertEqual(self.poll(client), alert.EXIT_UNREACHABLE) + self.assertEqual(run_check(client), alert.EXIT_UNREACHABLE) + + def test_a_key_rejected_mid_run_keeps_the_errors_before_it(self): + many = [dict(detail("fault"), outageId=str(4000000 + i)) for i in range(3)] + errors = {many[0]["outageId"]: TransientError("503"), + many[1]["outageId"]: TransientError("503"), + many[2]["outageId"]: AuthError("401 rejected")} + self.poll(FakeClient(list_body=make_list(*many), details={}, detail_errors=errors)) + with self.store() as st: + row = st.conn.execute("SELECT n_errors, error_summary FROM run").fetchone() + self.assertEqual(row["n_errors"], 3) + self.assertEqual(row["error_summary"].count("503"), 2) + def test_failures_are_still_recorded_in_the_run_table(self): self.poll(FakeClient(list_error=AuthError("401 rejected"))) with self.store() as st: diff --git a/tests/test_rebuild.py b/tests/test_rebuild.py index 1c75b43..07831ab 100644 --- a/tests/test_rebuild.py +++ b/tests/test_rebuild.py @@ -10,6 +10,7 @@ import unittest from pathlib import Path +from esb_outages.client import AuthError from esb_outages.poll import run_poll from esb_outages.store import Store @@ -148,6 +149,38 @@ def write(run_id, started_at, outage_type): self.assertEqual(row["first_seen_utc"], "2026-07-31T10:00:00Z") self.assertEqual(row["last_seen_utc"], "2026-07-31T12:00:00Z") + def test_observations_that_lost_their_run_replay_in_their_place(self): + fault = detail("fault") + done = dict(fault, outageType="Restored", restoreTime="31/07/2026 23:30") + raw = self.data_dir / "raw" + raw.mkdir(parents=True, exist_ok=True) + + def write(run_id, at, body, with_run=True): + if with_run: + with (raw / "runs-2026-07.jsonl").open("a") as fh: + fh.write(json.dumps({ + "run_id": run_id, "started_at": at, + "list_status": 200, "list_body": make_list(body), + }, sort_keys=True) + "\n") + with (raw / "observations-2026-07.jsonl").open("a") as fh: + fh.write(json.dumps({ + "run_id": run_id, "observed_at": at, + "outage_id": body["outageId"], "http_status": 200, "body": body, + }, sort_keys=True) + "\n") + + # the earlier run's own line was lost, so its observation is orphaned; + # the later run starting in the same second must still win + for at in ("2026-07-31T20:00:00Z", "2026-07-31T23:40:00Z"): + for f in raw.glob("*.jsonl"): + f.unlink() + write("a", at, fault, with_run=False) + write("b", "2026-07-31T23:40:00Z", done) + with Store(self.data_dir) as st: + st.rebuild() + row = st.conn.execute("SELECT * FROM outage").fetchone() + self.assertEqual((row["outage_type"], row["is_final"]), ("Restored", 1), at) + self.assertEqual(row["last_seen_utc"], "2026-07-31T23:40:00Z") + def test_a_truncated_final_line_does_not_destroy_the_history(self): """A power cut mid-append, or a backup snapshotting mid-write. @@ -205,6 +238,127 @@ def test_a_failed_run_is_not_lost_by_a_rebuild(self): statuses = [r[0] for r in st.conn.execute("SELECT status FROM run")] self.assertIn("auth_error", statuses) + def test_every_run_column_survives_a_rebuild(self): + """How a run ended is not derivable from its list and observations, so + the log carries it; without that a storm's cut-short runs came back ok.""" + from esb_outages.client import AuthError, TransientError + + self.run_a_realistic_history() + many = [dict(detail("fault"), outageId=str(3000000 + i)) for i in range(4)] + by_id = {d["outageId"]: d for d in many} + run_poll(self.data_dir, client=FakeClient(list_body=make_list(*many), + details=by_id), delay_ms=0, budget_s=0) + more = [dict(detail("fault"), outageId=str(4000000 + i)) for i in range(4)] + self.poll(FakeClient( + list_body=make_list(*more), + details={d["outageId"]: d for d in more}, + detail_errors={d["outageId"]: TransientError("503") for d in more[:3]}, + )) + last = dict(detail("fault"), outageId="5000000") + self.poll(FakeClient(list_body=make_list(last), details={}, + detail_errors={"5000000": AuthError("401")})) + self.poll(FakeClient(list_error=AuthError("401"))) + + def runs(st): + return [tuple(r) for r in st.conn.execute( + "SELECT * FROM run ORDER BY started_at_utc, run_id")] + + with Store(self.data_dir) as st: + before = runs(st) + st.rebuild() + after = runs(st) + statuses = {r[3] for r in before} + self.assertTrue({"cut_short", "partial", "auth_error"} <= statuses, statuses) + self.assertEqual(before, after) + + def test_rebuild_and_compact_wait_for_a_running_poll(self): + from esb_outages.__main__ import main + from esb_outages.poll import poll_lock + + self.run_a_realistic_history() + db = self.data_dir / "esb.db" + inode = db.stat().st_ino + with poll_lock(self.data_dir) as held: + self.assertTrue(held) + for command in ("rebuild", "compact"): + self.assertEqual(main(["--data-dir", str(self.data_dir), command]), 1) + self.assertEqual(db.stat().st_ino, inode) + + def test_a_malformed_response_neither_kills_the_run_nor_the_rebuild(self): + from esb_outages import alert + + fault = detail("fault") + listed = make_list(fault) + listed["outageMessage"].insert(0, "not an object") + code = self.poll(FakeClient(list_body=listed, details={fault["outageId"]: None})) + self.assertEqual(code, alert.EXIT_SCHEMA_DRIFT) + self.assertEqual(self.poll(FakeClient(list_body=[])), alert.EXIT_SCHEMA_DRIFT) + def runs(st): + return [tuple(r) for r in st.conn.execute("SELECT * FROM run ORDER BY run_id")] + + with Store(self.data_dir) as st: + before, runs_before = st.snapshot(), runs(st) + st.rebuild() + self.assertEqual(st.snapshot(), before) + self.assertEqual(runs(st), runs_before) + self.assertEqual( + [r[0] for r in st.conn.execute("SELECT outage_id FROM outage")], + [fault["outageId"]], + ) + + def test_a_run_that_died_before_its_end_line_is_not_ok(self): + with Store(self.data_dir) as st: + st.write_run_raw("2026-01-01T10:00:00Z-deadbeef", "2026-01-01T10:00:00Z", + 200, make_list(detail("fault"))) + self.run_a_realistic_history() + with Store(self.data_dir) as st: + st.write_run_raw("2099-01-01T10:00:00Z-5ca1ab1e", "2099-01-01T10:00:00Z", + 200, make_list(detail("fault"))) + st.rebuild() + status = dict(st.conn.execute("SELECT run_id, status FROM run").fetchall()) + # the newest is a run a backup caught before it had ended + self.assertEqual(status["2026-01-01T10:00:00Z-deadbeef"], "unfinished") + self.assertEqual(status["2099-01-01T10:00:00Z-5ca1ab1e"], "in_progress") + + def test_a_failure_the_start_line_logged_survives_a_lost_end_line(self): + self.poll(FakeClient(list_error=AuthError("401"))) + runs = next((self.data_dir / "raw").glob("runs-*.jsonl")) + lines = runs.read_text().splitlines() + runs.write_text("\n".join(line for line in lines if '"event": "end"' not in line) + "\n") + with Store(self.data_dir) as st: + st.rebuild() + status = st.conn.execute("SELECT status FROM run").fetchone()[0] + self.assertEqual(status, "auth_error") + + def test_start_lines_from_before_end_lines_still_replay_as_they_said(self): + self.run_a_realistic_history() + with (self.data_dir / "raw" / "runs-2026-01.jsonl").open("a") as fh: + # a second host still on the old code, merged in after this one's + fh.write(json.dumps({"run_id": "2099-01-01T10:00:00Z-01d01d01", + "started_at": "2099-01-01T10:00:00Z", "list_status": 200, + "list_body": make_list(detail("fault"))}, sort_keys=True)) + with Store(self.data_dir) as st: + st.rebuild() + status = st.conn.execute("SELECT status FROM run WHERE run_id LIKE '2099-01-01%'" + ).fetchone()[0] + self.assertEqual(status, "ok") + + def test_a_run_whose_start_line_was_lost_keeps_its_end(self): + from esb_outages import alert + + self.poll(FakeClient(list_body=make_list(detail("fault")), + details={detail("fault")["outageId"]: detail("fault")})) + runs = next((self.data_dir / "raw").glob("runs-*.jsonl")) + lines = runs.read_text().splitlines() + runs.write_text("\n".join(line for line in lines if '"event": "end"' in line) + "\n") + with Store(self.data_dir) as st: + result = st.rebuild() + row = st.conn.execute("SELECT * FROM run").fetchone() + self.assertEqual((row["status"], row["exit_code"]), ("ok", alert.EXIT_OK)) + self.assertEqual(row["run_id"][:20], row["started_at_utc"]) + # replayed as a run, so its own observation counts as its fetch + self.assertEqual((result["runs"], row["n_detail_fetched"]), (1, 1)) + def test_rebuild_on_empty_data_dir_is_harmless(self): with Store(self.data_dir) as st: self.assertEqual( diff --git a/tests/test_store.py b/tests/test_store.py index 406a4e3..a15de2a 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -90,6 +90,43 @@ def test_skips_only_finalised_outages(self): # Restored is immutable; the ongoing fault still needs watching. self.assertEqual(self.store.ids_needing_detail(ids), [fault["outageId"]]) + def test_restored_outages_lead_uncaptured_before_settling(self): + settling = dict(detail("restored"), restoreTime="") + fault = detail("fault") + uncaptured = dict(detail("restored"), outageId="9100000") + items = make_list(fault, settling)["outageMessage"] + self.store.apply_list("2026-07-31T10:00:00Z", items) + self.store.apply_detail("2026-07-31T10:00:01Z", normalize_detail(settling)) + # seven hours on it has gone quiet, but ESB could purge it any time; + # one never captured at all still goes ahead of it + later = "2026-07-31T17:00:00Z" + items = make_list(settling, fault, uncaptured)["outageMessage"] + self.store.apply_list(later, items) + ids = [settling["outageId"], fault["outageId"], uncaptured["outageId"]] + self.assertEqual( + self.store.ids_needing_detail(ids, now=later), + [uncaptured["outageId"], settling["outageId"], fault["outageId"]], + ) + + def test_a_captured_fault_that_flips_to_restored_waits_behind_an_uncaptured_one(self): + fault = detail("fault") + uncaptured = dict(detail("restored"), outageId="9100000") + self.store.apply_list("2026-07-31T10:00:00Z", make_list(fault)["outageMessage"]) + self.store.apply_detail("2026-07-31T10:00:01Z", normalize_detail(fault)) + flipped = dict(fault, outageType="Restored") + later = "2026-07-31T11:00:00Z" + self.store.apply_list(later, make_list(flipped, uncaptured)["outageMessage"]) + self.assertEqual( + self.store.ids_needing_detail([fault["outageId"], "9100000"], now=later), + ["9100000", fault["outageId"]], + ) + + def test_an_id_listed_twice_is_fetched_once(self): + fault = detail("fault") + self.store.apply_list("2026-07-31T10:00:00Z", make_list(fault)["outageMessage"]) + ids = [fault["outageId"], fault["outageId"]] + self.assertEqual(self.store.ids_needing_detail(ids), [fault["outageId"]]) + def test_handles_empty_input(self): self.assertEqual(self.store.ids_needing_detail([]), []) @@ -194,6 +231,51 @@ def test_raw_files_are_split_by_month(self): names = {p.name for p in self.store.raw_files("runs")} self.assertEqual(names, {"runs-2026-07.jsonl", "runs-2026-08.jsonl"}) + def test_a_torn_last_line_does_not_take_the_next_record_with_it(self): + self.store.write_run_raw("r1", "2026-08-01T10:00:00Z", 200, {"outageMessage": []}) + with (self.data_dir / "raw" / "runs-2026-08.jsonl").open("a") as fh: + fh.write('{"event": "end", "run_id": "r1", "fin') + self.store.write_run_raw("r2", "2026-08-01T10:30:00Z", 200, {"outageMessage": []}) + self.assertEqual([r["run_id"] for r in self.store.iter_raw("runs")], ["r1", "r2"]) + self.assertEqual(len(self.store.malformed_lines), 1) + + def test_a_line_torn_inside_a_fada_is_one_bad_line(self): + self.store.write_run_raw("r1", "2026-08-01T10:00:00Z", 200, {"outageMessage": []}) + with (self.data_dir / "raw" / "runs-2026-08.jsonl").open("ab") as fh: + fh.write('{"run_id": "r2", "location": "Dún'.encode()[:-2]) + self.store.write_run_raw("r3", "2026-08-01T11:00:00Z", 200, {"outageMessage": []}) + self.assertEqual([r["run_id"] for r in self.store.iter_raw("runs")], ["r1", "r3"]) + self.assertEqual(len(self.store.malformed_lines), 1) + + def test_a_corrupted_byte_is_a_bad_line_not_a_changed_record(self): + self.store.write_run_raw("r1", "2026-08-01T10:00:00Z", 200, {"outageMessage": []}) + with (self.data_dir / "raw" / "runs-2026-08.jsonl").open("ab") as fh: + # still valid JSON if the stray byte were replaced + fh.write(b'{"location": "D\xc3n", "run_id": "r2"}\n') + self.assertEqual([r["run_id"] for r in self.store.iter_raw("runs")], ["r1"]) + self.assertEqual(len(self.store.malformed_lines), 1) + + def test_a_copy_under_another_name_in_the_same_month_is_read_once(self): + self.store.write_run_raw("r1", "2026-01-15T10:00:00Z", 200, {"outageMessage": []}) + raw = self.data_dir / "raw" + (raw / "runs-2026-01-pi2.jsonl").write_bytes((raw / "runs-2026-01.jsonl").read_bytes()) + self.store.write_run_raw("r2", "2026-02-15T10:00:00Z", 200, {"outageMessage": []}) + self.assertEqual([r["run_id"] for r in self.store.iter_raw("runs")], ["r1", "r2"]) + + def test_a_torn_archive_does_not_swallow_the_late_lines(self): + import gzip + + self.store.write_run_raw("old", "2020-01-15T10:00:00Z", 200, {"outageMessage": []}) + with (self.data_dir / "raw" / "runs-2020-01.jsonl").open("a") as fh: + fh.write('{"run_id": "torn", "sta') + self.store.compact() + self.store.write_run_raw("late", "2020-01-31T23:00:00Z", 200, {"outageMessage": []}) + path = self.data_dir / "raw" / "runs-2020-01.jsonl" + self.store.compact() + self.assertIn("late", [r["run_id"] for r in self.store.iter_raw("runs")]) + with gzip.open(str(path) + ".gz", "rt") as fh: + self.assertIn("late", fh.read()) + def test_compact_gzips_old_months_and_keeps_current(self): from esb_outages.store import utc_now_iso @@ -213,6 +295,27 @@ def test_gzipped_logs_are_still_readable(self): records = list(self.store.iter_raw("runs")) self.assertEqual([r["run_id"] for r in records], ["old"]) + def test_a_month_written_to_after_compaction_keeps_its_archive(self): + self.store.write_run_raw("old", "2020-01-15T10:00:00Z", 200, {"outageMessage": []}) + self.store.compact() + # a poll on a clock that booted in January again + self.store.write_run_raw("late", "2020-01-31T23:00:00Z", 200, {"outageMessage": []}) + self.store.compact() + records = list(self.store.iter_raw("runs")) + self.assertEqual([r["run_id"] for r in records], ["old", "late"]) + self.assertFalse((self.data_dir / "raw" / "runs-2020-01.jsonl").exists()) + self.assertFalse((self.data_dir / "raw" / "runs-2020-01.jsonl.gz.tmp").exists()) + + def test_a_compact_interrupted_before_the_unlink_does_not_double_the_log(self): + from unittest import mock + + self.store.write_run_raw("old", "2020-01-15T10:00:00Z", 200, {"outageMessage": []}) + with mock.patch.object(Path, "unlink", side_effect=OSError("power cut")): + with self.assertRaises(OSError): + self.store.compact() + self.store.compact() + self.assertEqual([r["run_id"] for r in self.store.iter_raw("runs")], ["old"]) + if __name__ == "__main__": unittest.main()