From 38e394c2d9be7bc9d9a4c8feea6461da6f07ca5d Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 09:40:56 +0000 Subject: [PATCH 01/27] Log how each run ended, so a rebuild restores its status The raw run line is written before the details start, so it can only ever say "ok". A rebuild therefore restored every cut-short, partial, drifted or mid-run auth-failed run as ok, with no finish time, exit code or error summary. Store.finish_run now appends an end line first, carrying only what cannot be derived, and every exit path in poll goes through it. rebuild prefers the end line where there is one; older runs replay as before. The run table now round-trips exactly, which a new test holds. This reverses the gap notes/storms.md had accepted; the note says why. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/poll.py | 8 ++++---- esb_outages/store.py | 45 ++++++++++++++++++++++++++++++++++++++----- notes/storms.md | 20 +++++++++++++++++++ tests/test_poll.py | 7 +++++-- tests/test_rebuild.py | 33 +++++++++++++++++++++++++++++++ 5 files changed, 102 insertions(+), 11 deletions(-) diff --git a/esb_outages/poll.py b/esb_outages/poll.py index fbfe0d0..1d47ca3 100644 --- a/esb_outages/poll.py +++ b/esb_outages/poll.py @@ -144,7 +144,7 @@ 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), @@ -152,7 +152,7 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_ return alert.fail(alert.auth_banner(client.masked_key, str(exc)), alert.EXIT_AUTH) except (TransientError, ApiError) 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), @@ -199,7 +199,7 @@ 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), @@ -248,7 +248,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, diff --git a/esb_outages/store.py b/esb_outages/store.py index f8f977c..2cb7b5c 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -117,6 +117,10 @@ """ +# 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. @@ -228,6 +232,25 @@ def write_run_raw( }, ) + 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: @@ -513,7 +536,13 @@ 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) + run_records.sort(key=lambda r: r["started_at"]) seen_run_ids = set() for rec in run_records: @@ -534,16 +563,22 @@ def apply_observation(rec) -> int: 1 for o in run_obs if o.get("http_status") not in (200, 404) ) listed = len(items) if isinstance(items, list) 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 rec.get("status", "ok"), + exit_code=end.get("exit_code"), n_listed=listed, - n_detail_fetched=fetched, - n_detail_skipped=( + n_detail_fetched=fetched if listed is not None 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: diff --git a/notes/storms.md b/notes/storms.md index f1ff1f1..179945a 100644 --- a/notes/storms.md +++ b/notes/storms.md @@ -92,3 +92,23 @@ 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. diff --git a/tests/test_poll.py b/tests/test_poll.py index b5d2b29..0d83adb 100644 --- a/tests/test_poll.py +++ b/tests/test_poll.py @@ -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) diff --git a/tests/test_rebuild.py b/tests/test_rebuild.py index 1c75b43..d850d6d 100644 --- a/tests/test_rebuild.py +++ b/tests/test_rebuild.py @@ -205,6 +205,39 @@ 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_on_empty_data_dir_is_harmless(self): with Store(self.data_dir) as st: self.assertEqual( From 1cd2b164f3cfefae2295245bc3ab18d2f036501e Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 09:41:22 +0000 Subject: [PATCH 02/27] Never let compact overwrite a month's archive compact wrote each archive with "wb", truncating any existing one. A Pi with no clock battery boots in the past until NTP syncs, so a poll can append to a month that was already compacted, and the next compact replaced the whole archived month with those few late lines. The late lines now join the archive as a further gzip member, written to a staged file, fsynced and renamed into place before the source is removed, so a power cut in between can no longer leave an empty archive and no log. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 26 ++++++++++++++++++++++++-- tests/test_store.py | 11 +++++++++++ 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index 2cb7b5c..3464a77 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -14,6 +14,7 @@ import io import json import os +import shutil import sqlite3 import sys from datetime import UTC, datetime @@ -144,6 +145,15 @@ 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] @@ -680,8 +690,20 @@ 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") + 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 target.exists(): + out.write(target.read_bytes()) + with path.open("rb") as src, gzip.GzipFile(fileobj=out, mode="wb") as dst: + shutil.copyfileobj(src, dst) + out.flush() + os.fsync(out.fileno()) + os.replace(staged, target) + _fsync_dir(self.raw_dir) path.unlink() compacted.append(target.name) return sorted(compacted) diff --git a/tests/test_store.py b/tests/test_store.py index 406a4e3..738c479 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -213,6 +213,17 @@ 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()) + if __name__ == "__main__": unittest.main() From d22965c57a4418c8a502fe0a2d6043f0dc6130f8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 09:41:41 +0000 Subject: [PATCH 03/27] Make rebuild and compact wait for a running poll rebuild deletes esb.db and its -wal and -shm, and compact unlinks the log a poll appends to, and neither took the poll lock, so `esb rebuild` during a 24-minute storm run could corrupt the database or lose the details the poll committed after the replay read the logs. Both now take the same lock and exit 1 with a message while a poll holds it. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/__main__.py | 23 ++++++++++++++++++----- tests/test_rebuild.py | 13 +++++++++++++ 2 files changed, 31 insertions(+), 5 deletions(-) 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/tests/test_rebuild.py b/tests/test_rebuild.py index d850d6d..2dd1bbf 100644 --- a/tests/test_rebuild.py +++ b/tests/test_rebuild.py @@ -238,6 +238,19 @@ def runs(st): 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_rebuild_on_empty_data_dir_is_harmless(self): with Store(self.data_dir) as st: self.assertEqual( From 3787fda767fc513b2a535fb15deafc13e8803acb Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 09:42:18 +0000 Subject: [PATCH 04/27] Keep a cut-off response or a 404 list from killing the run IncompleteRead is an HTTPException, not an OSError, and a truncated gzip body raises EOFError or zlib.error from decompress, so one connection cut off mid-read escaped the client's taxonomy and took the whole run with it: no run record, no heartbeat, a traceback. They are now TransientError, and an undecodable body is an ApiError. A 404 on the list itself, which the client raises as NotFound, was not caught either; poll and check now treat it as unreachable. The client had no tests; it has three. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/client.py | 8 +++++++ esb_outages/poll.py | 4 ++-- tests/test_client.py | 53 +++++++++++++++++++++++++++++++++++++++++++ tests/test_poll.py | 7 +++++- 4 files changed, 69 insertions(+), 3 deletions(-) create mode 100644 tests/test_client.py 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/poll.py b/esb_outages/poll.py index 1d47ca3..83fa720 100644 --- a/esb_outages/poll.py +++ b/esb_outages/poll.py @@ -150,7 +150,7 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_ 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.finish_run( run_id=run_id, started_at_utc=started_at, finished_at_utc=utc_now_iso(), @@ -284,7 +284,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/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_poll.py b/tests/test_poll.py index 0d83adb..04e3265 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 @@ -85,6 +85,11 @@ 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_failures_are_still_recorded_in_the_run_table(self): self.poll(FakeClient(list_error=AuthError("401 rejected"))) with self.store() as st: From 4d4e14cd53e737fa335fbbcac4b7593bedeb18a0 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 09:42:42 +0000 Subject: [PATCH 05/27] Survive a list or detail that is not an object, live and on replay A list item or body that was not a dict raised AttributeError after its raw line was already written: the live run died with no record or heartbeat, and since rebuild replays through the same apply_list, every later rebuild died on that line too, with editing the log the only way out. Non-object items are now skipped where they are applied, a non-object list or detail body is reported as schema drift, and the run carries on. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/parse.py | 4 ++++ esb_outages/poll.py | 4 +++- esb_outages/store.py | 4 ++++ tests/test_rebuild.py | 18 ++++++++++++++++++ 4 files changed, 29 insertions(+), 1 deletion(-) diff --git a/esb_outages/parse.py b/esb_outages/parse.py index 2d2709e..87ae8f3 100644 --- a/esb_outages/parse.py +++ b/esb_outages/parse.py @@ -105,6 +105,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 +120,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"] diff --git a/esb_outages/poll.py b/esb_outages/poll.py index 83fa720..e0c4d86 100644 --- a/esb_outages/poll.py +++ b/esb_outages/poll.py @@ -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 = [] @@ -220,6 +220,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 diff --git a/esb_outages/store.py b/esb_outages/store.py index 3464a77..dee3df3 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -331,6 +331,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")) diff --git a/tests/test_rebuild.py b/tests/test_rebuild.py index 2dd1bbf..58199e6 100644 --- a/tests/test_rebuild.py +++ b/tests/test_rebuild.py @@ -251,6 +251,24 @@ def test_rebuild_and_compact_wait_for_a_running_poll(self): 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) + with Store(self.data_dir) as st: + before = st.snapshot() + st.rebuild() + self.assertEqual(st.snapshot(), before) + self.assertEqual( + [r[0] for r in st.conn.execute("SELECT outage_id FROM outage")], + [fault["outageId"]], + ) + def test_rebuild_on_empty_data_dir_is_harmless(self): with Store(self.data_dir) as st: self.assertEqual( From dc39be5c624a044947c7cfc7dddbbe6c1e6d8c90 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 09:43:07 +0000 Subject: [PATCH 06/27] Keep a Restored outage still missing its restore time first in line ids_needing_detail put anything listed Restored first only until its first fetch. A Restored body with restoreTime still "" is not final, so on later runs it fell to rank 3 or 4, or out of the queue once quiet for six hours, behind every live outage a storm run could reach: ESB could purge it before its restore time was ever captured. It now stays at rank 0 until final, which is the order notes/storms.md already describes. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 4 ++++ tests/test_store.py | 14 ++++++++++++++ 2 files changed, 18 insertions(+) diff --git a/esb_outages/store.py b/esb_outages/store.py index dee3df3..c5f8d64 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -435,6 +435,10 @@ def urgency(outage_id: str) -> int | None: return 0 if restored else 2 if row["is_final"]: return None + if restored: + # fetched once already, but with no restore time yet: still on + # the purge clock, and what it takes is the one field we want + return 0 if _hours_between(row["last_change"], now) < quiet_after_hours: return 3 # actively changing, keep watching closely if _hours_between(row["last_detail_utc"], now) >= recheck_hours: diff --git a/tests/test_store.py b/tests/test_store.py index 738c479..be386b7 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -90,6 +90,20 @@ 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_without_a_restore_time_stays_first_in_line(self): + settling = dict(detail("restored"), restoreTime="") + fault = detail("fault") + 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 + later = "2026-07-31T17:00:00Z" + self.store.apply_list(later, items) + self.assertEqual( + self.store.ids_needing_detail([fault["outageId"], settling["outageId"]], now=later), + [settling["outageId"], fault["outageId"]], + ) + def test_handles_empty_input(self): self.assertEqual(self.store.ids_needing_detail([]), []) From c2256e1aec260a52ca803687f7aa2862f7474a06 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 09:43:27 +0000 Subject: [PATCH 07/27] Only a restore time that parses makes an outage final is_final was set from the raw restoreTime being non-empty. Final means never fetched again, so a restoreTime ESB sends in a new or broken format would freeze the outage with no restore time, and the site would read it as restored at an unknown time until ESB purged it. It now needs the parsed time. No outage in the corpus to 24 September has an unparsed one. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/parse.py | 6 +++--- tests/test_parse.py | 6 ++++++ 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/esb_outages/parse.py b/esb_outages/parse.py index 87ae8f3..d4ad2b3 100644 --- a/esb_outages/parse.py +++ b/esb_outages/parse.py @@ -188,8 +188,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/tests/test_parse.py b/tests/test_parse.py index 16b493b..3e54e59 100644 --- a/tests/test_parse.py +++ b/tests/test_parse.py @@ -124,6 +124,12 @@ 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_empty_strings_become_null(self): n = normalize_detail(detail("restored")) self.assertIsNone(n["status_message"]) From 049ef352851a055264b9fc4037b9006f2ad77ba5 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 09:43:35 +0000 Subject: [PATCH 08/27] Say which way the October clock change errs fold=0 is the first 01:xx, still on summer time, so a time that was really the second is stored an hour early, not late; test_dst_fall_back already pins the value. Only the docstring was wrong. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/parse.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/esb_outages/parse.py b/esb_outages/parse.py index d4ad2b3..50d6017 100644 --- a/esb_outages/parse.py +++ b/esb_outages/parse.py @@ -48,7 +48,8 @@ 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. From ad4847ef51c7df895570beebe05aadd4775f2a04 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 09:44:11 +0000 Subject: [PATCH 09/27] Replay observations that lost their run at their place in time rebuild replayed observations with no run record after every run in history, and apply_detail overwrites state, so an old orphaned body rolled a later restore back to live and moved last_seen backwards. They now take their place in the replay by their earliest observed time. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 31 ++++++++++++++++++++----------- tests/test_rebuild.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 11 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index c5f8d64..f787f8c 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -560,12 +560,28 @@ def apply_observation(rec) -> int: ends[rec.get("run_id")] = rec else: run_records.append(rec) - run_records.sort(key=lambda r: r["started_at"]) - seen_run_ids = set() - for rec in run_records: + # 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. + run_ids = {rec["run_id"] for rec in run_records} + 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 + ] + timeline = sorted( + [(rec["started_at"], rec, None) for rec in run_records] + + [(at, None, records) for at, records in orphans], + key=lambda event: event[0], + ) + + 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): @@ -602,13 +618,6 @@ def apply_observation(rec) -> int: 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") diff --git a/tests/test_rebuild.py b/tests/test_rebuild.py index 58199e6..c59df47 100644 --- a/tests/test_rebuild.py +++ b/tests/test_rebuild.py @@ -148,6 +148,34 @@ 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 + write("a", "2026-07-31T20:00:00Z", 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)) + 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. From abe450aeab5ea9381cf611c9a4c5b970ef4ec2f9 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 09:45:05 +0000 Subject: [PATCH 10/27] Point CLAUDE.md at the run's end line Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- CLAUDE.md | 1 + 1 file changed, 1 insertion(+) 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` | From b6eb21b94f5a21e1f70898cea507ea2836980d49 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 10:28:10 +0000 Subject: [PATCH 11/27] Treat a timestamp that is not a string as no timestamp The PR made a non-object detail survivable, but an object whose startTime, estRestoreTime or restoreTime was a number or a list still raised on .strip(), live and on every rebuild after it. Found by review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/parse.py | 2 +- tests/test_parse.py | 6 ++++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/esb_outages/parse.py b/esb_outages/parse.py index 50d6017..32f59b7 100644 --- a/esb_outages/parse.py +++ b/esb_outages/parse.py @@ -56,7 +56,7 @@ def parse_esb_datetime(value: str | None) -> tuple[str | None, bool]: 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) diff --git a/tests/test_parse.py b/tests/test_parse.py index 3e54e59..183df8e 100644 --- a/tests/test_parse.py +++ b/tests/test_parse.py @@ -130,6 +130,12 @@ def test_an_unreadable_restore_time_is_not_final(self): 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"]) From 823b55a8575f3c22285ec73431587b4ee0f4b36b Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 10:28:46 +0000 Subject: [PATCH 12/27] Make an interrupted compact safe to repeat With compact now appending to an existing archive, a crash after the rename but before the source was removed made the next compact archive the same lines twice, and every rebuild after that counted those runs twice. iter_raw now drops a line identical to one already read: every line carries its own timestamps and run id, so a repeat is the same record, which is also what a `sort -u` merge assumes. The removal is fsynced, and the existing archive is streamed into the staged file rather than read whole into memory on the Pi. Found by review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 15 ++++++++++++++- tests/test_store.py | 10 ++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index f787f8c..b703d19 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -11,6 +11,7 @@ from __future__ import annotations import gzip +import hashlib import io import json import os @@ -290,12 +291,20 @@ 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() for path in self.raw_files(kind): 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.encode("utf-8"), digest_size=16).digest() + if digest in seen: + continue + seen.add(digest) try: yield json.loads(line) except json.JSONDecodeError as exc: @@ -714,13 +723,17 @@ def compact(self) -> list[str]: # its archive and gains the late lines as a further gzip # member, which every gzip reader concatenates. if target.exists(): - out.write(target.read_bytes()) + with target.open("rb") as archived: + shutil.copyfileobj(archived, out) with path.open("rb") as src, gzip.GzipFile(fileobj=out, mode="wb") as dst: 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/tests/test_store.py b/tests/test_store.py index be386b7..44fc553 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -238,6 +238,16 @@ def test_a_month_written_to_after_compaction_keeps_its_archive(self): 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() From d67d175f117fe51eb3a9c486e2c18ba94bf31a1c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 10:29:12 +0000 Subject: [PATCH 13/27] Put a captured Restored outage behind an uncaptured one The settling outages moved to rank 0 earlier in this PR tied with Restored outages never captured at all, and list order broke the tie, so a storm with more of them than a run can fetch re-fetched the same head for a restore time while the uncaptured tail, whose whole record a purge takes, was lost. They now have their own rank, just behind. Found by review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 17 +++++++++-------- tests/test_store.py | 12 ++++++++---- 2 files changed, 17 insertions(+), 12 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index b703d19..4957962 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -433,25 +433,26 @@ 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 + return 0 if restored else 2 # Cleared by apply_list on a type change: something happened. if row["last_detail_utc"] is None: - return 0 if restored else 2 + return 0 if restored else 3 if row["is_final"]: return None if restored: - # fetched once already, but with no restore time yet: still on - # the purge clock, and what it takes is the one field we want - return 0 + # Captured, but with no restore time yet. Still on the purge + # clock, behind the rank above: a purge here costs one field, + # there the whole record. + return 1 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 = [ diff --git a/tests/test_store.py b/tests/test_store.py index 44fc553..a3a261d 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -90,18 +90,22 @@ 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_without_a_restore_time_stays_first_in_line(self): + 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 + # 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([fault["outageId"], settling["outageId"]], now=later), - [settling["outageId"], fault["outageId"]], + self.store.ids_needing_detail(ids, now=later), + [uncaptured["outageId"], settling["outageId"], fault["outageId"]], ) def test_handles_empty_input(self): From 897f9af8c9195fa02034771440d62bea4aefb1d1 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 10:29:28 +0000 Subject: [PATCH 14/27] Count the errors before a key rejected mid-run The mid-run AuthError path recorded n_errors=1 and only the 401 as its summary, dropping the transient failures already collected. With rebuild now trusting the end line, that undercount would have been fixed into the log. It now counts and names them too. Found by review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/poll.py | 3 ++- tests/test_poll.py | 11 +++++++++++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/esb_outages/poll.py b/esb_outages/poll.py index e0c4d86..6176a4f 100644 --- a/esb_outages/poll.py +++ b/esb_outages/poll.py @@ -204,7 +204,8 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_ 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 diff --git a/tests/test_poll.py b/tests/test_poll.py index 04e3265..889e364 100644 --- a/tests/test_poll.py +++ b/tests/test_poll.py @@ -90,6 +90,17 @@ def test_a_404_on_the_list_is_unreachable_not_a_crash(self): 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: From a4b4e7d9792f952744e2538cb04c54151220e77f Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 10:29:58 +0000 Subject: [PATCH 15/27] Count listed and fetched on replay as the live run does For the malformed inputs this PR survives, rebuild counted every list item where poll counts only objects, left n_listed NULL for a 200 whose body was not an object where poll records 0 (which also moved the site's horizon, read from n_listed), and counted a non-object detail as fetched. The malformed-response test now holds the run table across a rebuild too. Found by review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 12 ++++++++++-- tests/test_rebuild.py | 6 +++++- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index 4957962..402ca67 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -601,12 +601,20 @@ 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, {}) diff --git a/tests/test_rebuild.py b/tests/test_rebuild.py index c59df47..dc296f7 100644 --- a/tests/test_rebuild.py +++ b/tests/test_rebuild.py @@ -288,10 +288,14 @@ def test_a_malformed_response_neither_kills_the_run_nor_the_rebuild(self): 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 = st.snapshot() + 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"]], From 72d1dfa11b10c867c13c83c8bb286db19b9c203b Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 10:30:17 +0000 Subject: [PATCH 16/27] Replay an orphan before a run that starts in the same second Runs were concatenated ahead of orphans before a stable sort, so on a tie the run replayed first and the orphan, whose own run started no later, overwrote it: the rollback the orphan fix set out to stop. Found by review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 9 +++++---- tests/test_rebuild.py | 20 ++++++++++++-------- 2 files changed, 17 insertions(+), 12 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index 402ca67..d44878b 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -580,13 +580,14 @@ def apply_observation(rec) -> int: 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"], rec, None) for rec in run_records] - + [(at, None, records) for at, records in orphans], - key=lambda event: event[0], + [(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], ) - for _, rec, orphaned in timeline: + for _, _, rec, orphaned in timeline: if orphaned is not None: for obs in orphaned: n_obs += apply_observation(obs) diff --git a/tests/test_rebuild.py b/tests/test_rebuild.py index dc296f7..c8426c6 100644 --- a/tests/test_rebuild.py +++ b/tests/test_rebuild.py @@ -167,14 +167,18 @@ def write(run_id, at, body, with_run=True): "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 - write("a", "2026-07-31T20:00:00Z", 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)) - self.assertEqual(row["last_seen_utc"], "2026-07-31T23:40:00Z") + # 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. From 4389ab651185b6124ab9b6e46305715b0566af89 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 10:30:56 +0000 Subject: [PATCH 17/27] Replay a run with only one of its two lines as what it was Once runs log their end, a start line without one is a run that died before closing itself out, and a rebuild restored it as a clean ok. It is now `unfinished`. The reverse, an end line whose start line was damaged, was silently dropped though its run id carries the start time; it now gets a row and the rebuild says so. Found by review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 32 +++++++++++++++++++++++++++++++- notes/storms.md | 10 ++++++++++ tests/test_rebuild.py | 25 +++++++++++++++++++++++++ 3 files changed, 66 insertions(+), 1 deletion(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index d44878b..80b5a4f 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -15,6 +15,7 @@ import io import json import os +import re import shutil import sqlite3 import sys @@ -119,6 +120,8 @@ """ +_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") @@ -571,6 +574,12 @@ def apply_observation(rec) -> int: else: run_records.append(rec) + # Once runs log their end, a start line without one is a run that died + # before closing itself out, not a clean one. + ends_from = min( + (r["started_at"] for r in run_records if r["run_id"] in ends), 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. @@ -623,7 +632,10 @@ def apply_observation(rec) -> int: run_id=run_id, started_at_utc=rec["started_at"], finished_at_utc=end.get("finished_at"), - status=end.get("status") or rec.get("status", "ok"), + status=end.get("status") or ( + "unfinished" if ends_from and rec["started_at"] > ends_from + else rec.get("status", "ok") + ), exit_code=end.get("exit_code"), n_listed=listed, n_detail_fetched=fetched if listed is not None else None, @@ -637,9 +649,27 @@ def apply_observation(rec) -> int: for obs in run_obs: n_obs += apply_observation(obs) + # An end line whose start line was lost still says how the run went, + # and its run id carries the start time. + lost_starts = 0 + for run_id, end in ends.items(): + started = str(run_id).rsplit("-", 1)[0] + if run_id in run_ids or not _UTC_STAMP.fullmatch(started): + continue + self.record_run( + run_id=run_id, + started_at_utc=started, + finished_at_utc=end.get("finished_at"), + **{k: end.get(k) for k in RUN_END_FIELDS}, + ) + lost_starts += 1 + 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 { diff --git a/notes/storms.md b/notes/storms.md index 179945a..b8a00a1 100644 --- a/notes/storms.md +++ b/notes/storms.md @@ -112,3 +112,13 @@ 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 with no end +line, in a log where earlier runs have one, 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`. An end line +whose start line was lost still gets a row, started at the time its run id +carries, and the rebuild says how many it recorded that way. 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_rebuild.py b/tests/test_rebuild.py index c8426c6..eb39b68 100644 --- a/tests/test_rebuild.py +++ b/tests/test_rebuild.py @@ -305,6 +305,31 @@ def runs(st): [fault["outageId"]], ) + def test_a_run_that_died_before_its_end_line_is_not_ok(self): + self.run_a_realistic_history() + with Store(self.data_dir) as st: + st.write_run_raw("2026-12-01T10:00:00Z-deadbeef", "2026-12-01T10:00:00Z", + 200, make_list(detail("fault"))) + st.rebuild() + status = st.conn.execute( + "SELECT status FROM run WHERE run_id = '2026-12-01T10:00:00Z-deadbeef'" + ).fetchone()[0] + self.assertEqual(status, "unfinished") + + 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: + 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"]) + def test_rebuild_on_empty_data_dir_is_harmless(self): with Store(self.data_dir) as st: self.assertEqual( From 1e72464a98a0f5fabc21e23c02c6b41210f6f2b0 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 10:31:24 +0000 Subject: [PATCH 18/27] Start every raw append on a fresh line Each run now ends with a write to runs-*.jsonl, so a power cut that tears that end line would leave the next run's start line, its only copy of the list, appended to the torn fragment, and iter_raw would skip the two as one unreadable line. An append after a file that does not end in a newline now starts with one, and loses only the torn fragment. Found by review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 11 +++++++++-- tests/test_store.py | 8 ++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index 80b5a4f..b9bcbb0 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -215,8 +215,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()) diff --git a/tests/test_store.py b/tests/test_store.py index a3a261d..f909a9c 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -212,6 +212,14 @@ 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_compact_gzips_old_months_and_keeps_current(self): from esb_outages.store import utc_now_iso From 4b9487db82a8f8a9716822d7e991de769a098670 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 11:04:52 +0000 Subject: [PATCH 19/27] Decide unfinished from the start line, and spare the newest run The rule marking a start line without an end line `unfinished` keyed on the earliest end line anywhere in the log, so a merged log from a host on the old code, or one early-dated run from a Pi booted without its clock, relabelled every older-style run after it. Start lines now say an end line follows, and only those can be unfinished. The newest one reads `in_progress`: backup-to-git.sh commits raw/ without the poll lock, so the Pages rebuild often sees a run still going. A run rebuilt from its end line now replays as a run with no list, so its observations stop being orphans. Found by a second review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 40 ++++++++++++++++++---------------------- notes/storms.md | 19 ++++++++++++------- tests/test_rebuild.py | 26 +++++++++++++++++++++----- 3 files changed, 51 insertions(+), 34 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index b9bcbb0..127bc89 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -250,6 +250,9 @@ 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, }, ) @@ -581,11 +584,19 @@ def apply_observation(rec) -> int: else: run_records.append(rec) - # Once runs log their end, a start line without one is a run that died - # before closing itself out, not a clean one. - ends_from = min( - (r["started_at"] for r in run_records if r["run_id"] in ends), default=None - ) + # 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. + starts = {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 starts and _UTC_STAMP.fullmatch(started): + run_records.append({"run_id": run_id, "started_at": started}) + 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. @@ -640,8 +651,8 @@ def apply_observation(rec) -> int: started_at_utc=rec["started_at"], finished_at_utc=end.get("finished_at"), status=end.get("status") or ( - "unfinished" if ends_from and rec["started_at"] > ends_from - else rec.get("status", "ok") + ("in_progress" if rec["started_at"] == newest else "unfinished") + if rec.get("ends_logged") else rec.get("status", "ok") ), exit_code=end.get("exit_code"), n_listed=listed, @@ -656,21 +667,6 @@ def apply_observation(rec) -> int: for obs in run_obs: n_obs += apply_observation(obs) - # An end line whose start line was lost still says how the run went, - # and its run id carries the start time. - lost_starts = 0 - for run_id, end in ends.items(): - started = str(run_id).rsplit("-", 1)[0] - if run_id in run_ids or not _UTC_STAMP.fullmatch(started): - continue - self.record_run( - run_id=run_id, - started_at_utc=started, - finished_at_utc=end.get("finished_at"), - **{k: end.get(k) for k in RUN_END_FIELDS}, - ) - lost_starts += 1 - self.conn.commit() if verbose: print(f"replayed {n_runs} runs and {n_obs} detail observations") diff --git a/notes/storms.md b/notes/storms.md index b8a00a1..b65c698 100644 --- a/notes/storms.md +++ b/notes/storms.md @@ -113,12 +113,17 @@ 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 with no end -line, in a log where earlier runs have one, 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`. An end line -whose start line was lost still gets a row, started at the time its run id -carries, and the rebuild says how many it recorded that way. Two copies of +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_rebuild.py b/tests/test_rebuild.py index eb39b68..e6b20cb 100644 --- a/tests/test_rebuild.py +++ b/tests/test_rebuild.py @@ -306,15 +306,31 @@ def runs(st): ) 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("2026-12-01T10:00:00Z-deadbeef", "2026-12-01T10:00:00Z", + st.write_run_raw("2099-01-01T10:00:00Z-5ca1ab1e", "2099-01-01T10:00:00Z", 200, make_list(detail("fault"))) st.rebuild() - status = st.conn.execute( - "SELECT status FROM run WHERE run_id = '2026-12-01T10:00:00Z-deadbeef'" - ).fetchone()[0] - self.assertEqual(status, "unfinished") + 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_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 From c2db3c9742447fe219a30dab46d0a5a93beb46d8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 11:05:12 +0000 Subject: [PATCH 20/27] Only drop a line that can be nothing but a copy iter_raw reads an identical line once, which is wrong only if something writes the same line twice for real: an id ESB lists twice was fetched twice, and both observations could land in the same second with the same body. ids_needing_detail now fetches a repeated id once. The set of lines seen is cleared at each month, since copies only ever share a month's files, so a rebuild on the Pi no longer holds a digest for every line of the whole history. Found by a second review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 8 ++++++++ tests/test_store.py | 6 ++++++ 2 files changed, 14 insertions(+) diff --git a/esb_outages/store.py b/esb_outages/store.py index 127bc89..5db7ac4 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -308,7 +308,12 @@ def iter_raw(self, kind: str): # 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 file, so the set never needs to span one. + if path.name.split(".", 1)[0] != month: + month = path.name.split(".", 1)[0] + seen.clear() with _open_maybe_gzip(path) as fh: for lineno, line in enumerate(fh, 1): line = line.strip() @@ -429,6 +434,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( diff --git a/tests/test_store.py b/tests/test_store.py index f909a9c..5f42bf2 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -108,6 +108,12 @@ def test_restored_outages_lead_uncaptured_before_settling(self): [uncaptured["outageId"], settling["outageId"], 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([]), []) From 6b1648b888bcd2995028f62dd54ad09c547cbedf Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 11:05:42 +0000 Subject: [PATCH 21/27] Keep a torn line to itself, mid-character or at an archive's end Lines are written with ensure_ascii=False, so a power cut inside a fada left bytes that strict UTF-8 decoding raised on, ending the whole rebuild rather than counting one bad line; the logs are now read with errors replaced. And compact appended late lines as a new gzip member straight after the archive's last byte, so a torn line there swallowed the first late record; the new member now starts on a fresh line. Found by a second review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 9 +++++++-- tests/test_store.py | 24 ++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index 5db7ac4..4cf56eb 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -163,9 +163,11 @@ def _month_of(iso_ts: str) -> str: def _open_maybe_gzip(path: Path): + # A line torn inside a multi-byte character (any fada) must stay one + # malformed line, not a decode error that ends the whole read. if path.suffix == ".gz": - return io.TextIOWrapper(gzip.open(path, "rb"), encoding="utf-8") - return path.open("r", encoding="utf-8") + return io.TextIOWrapper(gzip.open(path, "rb"), encoding="utf-8", errors="replace") + return path.open("r", encoding="utf-8", errors="replace") class Store: @@ -777,6 +779,9 @@ def compact(self) -> list[str]: with target.open("rb") as archived: shutil.copyfileobj(archived, out) with path.open("rb") as src, gzip.GzipFile(fileobj=out, mode="wb") as dst: + if target.exists(): + # 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()) diff --git a/tests/test_store.py b/tests/test_store.py index 5f42bf2..7d4f858 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -226,6 +226,30 @@ def test_a_torn_last_line_does_not_take_the_next_record_with_it(self): 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_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": []}) + # a late line written straight after the torn one, as a restored backup might + path = self.data_dir / "raw" / "runs-2020-01.jsonl" + path.write_text(path.read_text().lstrip("\n")) + 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 From 16136dd0297096e6935870cb6f9e4700de0aa12b Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 11:05:55 +0000 Subject: [PATCH 22/27] Rank a captured outage that flips to Restored with the settling ones A type change clears last_detail_utc, and a flip to Restored on an outage already captured still ranked 0, tied with Restored outages never captured at all. A purge costs it only its restore time, so it now shares rank 1 with the settling ones, behind the uncaptured. Found by a second review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 6 ++++-- tests/test_store.py | 13 +++++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index 4cf56eb..bf97ce3 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -462,9 +462,11 @@ def urgency(outage_id: str) -> int | None: restored = row["outage_type"] == "Restored" if not row["has_detail"]: return 0 if restored else 2 - # Cleared by apply_list on a type change: something happened. + # Cleared by apply_list on a type change: something happened. A + # captured record that has flipped to Restored loses only its + # restore time to a purge, like the settling ones below. if row["last_detail_utc"] is None: - return 0 if restored else 3 + return 1 if restored else 3 if row["is_final"]: return None if restored: diff --git a/tests/test_store.py b/tests/test_store.py index 7d4f858..6eed4bd 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -108,6 +108,19 @@ def test_restored_outages_lead_uncaptured_before_settling(self): [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"]) From 5015497b737e73e1910173065a0e07ac851476c0 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 11:32:30 +0000 Subject: [PATCH 23/27] Decode each log line strictly, so corruption is counted, not replayed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit read the logs with errors="replace" so a line torn inside a fada could not end a rebuild. That also turned a line corrupted on the card into a different valid record: "Dún" read back as "D�n", recorded as a change and never counted as malformed. Files are now read as bytes and each line decoded strictly; a decode error is one malformed line, like a JSON error. The dedupe digest is taken from the bytes. Found by a third review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 15 +++++++-------- tests/test_store.py | 8 ++++++++ 2 files changed, 15 insertions(+), 8 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index bf97ce3..ef9de4c 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -12,7 +12,6 @@ import gzip import hashlib -import io import json import os import re @@ -163,11 +162,11 @@ def _month_of(iso_ts: str) -> str: def _open_maybe_gzip(path: Path): - # A line torn inside a multi-byte character (any fada) must stay one - # malformed line, not a decode error that ends the whole read. + # 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", errors="replace") - return path.open("r", encoding="utf-8", errors="replace") + return gzip.open(path, "rb") + return path.open("rb") class Store: @@ -321,13 +320,13 @@ def iter_raw(self, kind: str): line = line.strip() if not line: continue - digest = hashlib.blake2b(line.encode("utf-8"), digest_size=16).digest() + 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 ------------------------------------------ diff --git a/tests/test_store.py b/tests/test_store.py index 6eed4bd..83ccc42 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -247,6 +247,14 @@ def test_a_line_torn_inside_a_fada_is_one_bad_line(self): 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_torn_archive_does_not_swallow_the_late_lines(self): import gzip From 738db7e6e8e311879f38a58029e76eb1c56cabeb Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 11:32:54 +0000 Subject: [PATCH 24/27] Keep a failure the start line logged when its end line is lost A run that failed at the list call logs auth_error or unreachable on its start line and flags that an end line follows; with that end line torn, the rebuild labelled it unfinished, or in_progress, and lost the failure it had already recorded. A non-ok start status now wins over the label. Found by a third review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 4 +++- tests/test_rebuild.py | 11 +++++++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index ef9de4c..3bdcbdf 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -662,8 +662,10 @@ def apply_observation(rec) -> int: started_at_utc=rec["started_at"], 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") else rec.get("status", "ok") + 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, diff --git a/tests/test_rebuild.py b/tests/test_rebuild.py index e6b20cb..9b5b9dd 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 @@ -319,6 +320,16 @@ def test_a_run_that_died_before_its_end_line_is_not_ok(self): 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: From 92061706aa8802ad649de1bd256bd0b2fca0d3b6 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 11:33:06 +0000 Subject: [PATCH 25/27] Key the dedupe reset on the file's month, not its whole stem The set of lines seen was reset whenever the part of the file name before the first dot changed, so a same-month copy under another name, such as runs-2026-01-pi2.jsonl from a second host, got a set of its own and its duplicates replayed twice. The month is now read from the name. Found by a third review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 9 ++++++--- tests/test_store.py | 7 +++++++ 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index 3bdcbdf..e753b73 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -119,6 +119,7 @@ """ +_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 @@ -311,9 +312,11 @@ def iter_raw(self, kind: str): seen: set[bytes] = set() month = None for path in self.raw_files(kind): - # Copies share a month file, so the set never needs to span one. - if path.name.split(".", 1)[0] != month: - month = path.name.split(".", 1)[0] + # 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): diff --git a/tests/test_store.py b/tests/test_store.py index 83ccc42..64366ff 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -255,6 +255,13 @@ def test_a_corrupted_byte_is_a_bad_line_not_a_changed_record(self): 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 From 553d4e50e597016f83bfdf660e4d088585242a84 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 11:33:20 +0000 Subject: [PATCH 26/27] Count a lost-start run's fetches from its own observations A run rebuilt from its end line has no list, so its fetched count was left NULL by the rule meant for runs that never got past the list call, though its observations were there and replayed as its own. It now counts them, and the test pins that the run replays as a run, not as orphans. Found by a third review of this PR. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 3 ++- tests/test_rebuild.py | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index e753b73..4ace897 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -672,7 +672,8 @@ def apply_observation(rec) -> int: ), exit_code=end.get("exit_code"), n_listed=listed, - n_detail_fetched=fetched if listed is not None else None, + # 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 ), diff --git a/tests/test_rebuild.py b/tests/test_rebuild.py index 9b5b9dd..07831ab 100644 --- a/tests/test_rebuild.py +++ b/tests/test_rebuild.py @@ -352,10 +352,12 @@ def test_a_run_whose_start_line_was_lost_keeps_its_end(self): 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: - st.rebuild() + 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: From ce30ff0bfc61c908883c7a5af2df0fb808fe0a76 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 24 Sep 2026 11:34:11 +0000 Subject: [PATCH 27/27] Tidy what the third review found redundant One set of run ids instead of two; the settling and flipped Restored ranks reached through one branch with one comment, which is equivalent because apply_list clears is_final with last_detail_utc; the archive's existence checked once in compact; and a no-op line and its misleading comment dropped from the torn-archive test. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UqWuH6frirpbQDKnF5rEdT --- esb_outages/store.py | 30 +++++++++++++++--------------- tests/test_store.py | 2 -- 2 files changed, 15 insertions(+), 17 deletions(-) diff --git a/esb_outages/store.py b/esb_outages/store.py index 4ace897..ec2098e 100644 --- a/esb_outages/store.py +++ b/esb_outages/store.py @@ -464,18 +464,17 @@ def urgency(outage_id: str) -> int | None: restored = row["outage_type"] == "Restored" if not row["has_detail"]: return 0 if restored else 2 - # Cleared by apply_list on a type change: something happened. A - # captured record that has flipped to Restored loses only its - # restore time to a purge, like the settling ones below. - if row["last_detail_utc"] is None: - return 1 if restored else 3 if row["is_final"]: return None if restored: - # Captured, but with no restore time yet. Still on the purge - # clock, behind the rank above: a purge here costs one field, - # there the whole record. + # 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 4 # actively changing, keep watching closely if _hours_between(row["last_detail_utc"], now) >= recheck_hours: @@ -601,12 +600,13 @@ def apply_observation(rec) -> int: # 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. - starts = {rec["run_id"] for rec in run_records} + 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 starts and _UTC_STAMP.fullmatch(started): + 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. @@ -615,7 +615,6 @@ def apply_observation(rec) -> int: # 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. - run_ids = {rec["run_id"] for rec in run_records} orphans = [ (min(o.get("observed_at") or "" for o in records), records) for run_id, records in observations.items() @@ -777,16 +776,17 @@ def compact(self) -> list[str]: continue target = Path(str(path) + ".gz") 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 target.exists(): - with target.open("rb") as archived: - shutil.copyfileobj(archived, out) + 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 target.exists(): + if archived: # the archive may end in a torn line; a blank one is skipped dst.write(b"\n") shutil.copyfileobj(src, dst) diff --git a/tests/test_store.py b/tests/test_store.py index 64366ff..a15de2a 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -270,9 +270,7 @@ def test_a_torn_archive_does_not_swallow_the_late_lines(self): fh.write('{"run_id": "torn", "sta') self.store.compact() self.store.write_run_raw("late", "2020-01-31T23:00:00Z", 200, {"outageMessage": []}) - # a late line written straight after the torn one, as a restored backup might path = self.data_dir / "raw" / "runs-2020-01.jsonl" - path.write_text(path.read_text().lstrip("\n")) 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: