Fix the collector bugs a retroactive review found

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` |

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
 

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."""

esb_outages/parse.py

@@ -48,14 +48,15 @@ def parse_esb_datetime(value: str | None) -> tuple[str | None, bool]:
     time is ambiguous or impossible because of a DST transition:
 
     - Fall back (last Sunday of October): 01:00-01:59 happens twice. We take the
-      first occurrence (fold=0), so the value may be an hour late.
+      first occurrence (fold=0, still UTC+1), so a time that was really the
+      second is stored an hour early.
     - Spring forward (last Sunday of March): 01:00-01:59 never happens. Python
       resolves it, but the result is a fiction.
 
     Either way the raw string is retained in the database, so a flagged row can
     be revisited rather than silently trusted.
     """
-    if not value or not value.strip():
+    if not isinstance(value, str) or not value.strip():
         return None, False
     try:
         naive = datetime.strptime(value.strip(), ESB_DATETIME_FORMAT)
@@ -105,6 +106,8 @@ def parse_point(point) -> tuple[float | None, float | None, str | None]:
 
 def check_detail_schema(body: dict) -> list[str]:
     """Return human-readable descriptions of any drift from the known shape."""
+    if not isinstance(body, dict):
+        return ["detail response is not an object"]
     keys = set(body)
     problems = []
     unexpected = sorted(keys - DETAIL_FIELDS)
@@ -118,6 +121,8 @@ def check_detail_schema(body: dict) -> list[str]:
 
 def check_list_schema(body: dict) -> list[str]:
     problems = []
+    if not isinstance(body, dict):
+        return ["response is not an object"]
     if "outageMessage" not in body:
         return ["response has no 'outageMessage' key"]
     items = body["outageMessage"]
@@ -184,8 +189,8 @@ def normalize_detail(body: dict) -> dict:
         "status_message": _text(body.get("statusMessage")),
         "planned_outage_reason": _text(body.get("plannedOutageReason")),
         # An outage is only immutable once it is restored *and* carries the
-        # actual restore time; "Restored" with an empty restoreTime is still
-        # settling and must be re-fetched.
-        "is_final": int(outage_type == "Restored" and restore_raw is not None),
+        # actual restore time; "Restored" with an empty or unreadable
+        # restoreTime is still settling and must be re-fetched.
+        "is_final": int(outage_type == "Restored" and restore_utc is not None),
         "tz_ambiguous": int(start_amb or est_amb or restore_amb),
     }

esb_outages/poll.py

@@ -144,15 +144,15 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_
         list_body = client.get_outage_list()
     except AuthError as exc:
         store.write_run_raw(run_id, started_at, 401, None, status="auth_error")
-        store.record_run(
+        store.finish_run(
             run_id=run_id, started_at_utc=started_at, finished_at_utc=utc_now_iso(),
             status="auth_error", exit_code=alert.EXIT_AUTH, n_errors=1,
             error_summary=str(exc),
         )
         return alert.fail(alert.auth_banner(client.masked_key, str(exc)), alert.EXIT_AUTH)
-    except (TransientError, ApiError) as exc:
+    except (TransientError, ApiError, NotFound) as exc:
         store.write_run_raw(run_id, started_at, 0, None, status="unreachable")
-        store.record_run(
+        store.finish_run(
             run_id=run_id, started_at_utc=started_at, finished_at_utc=utc_now_iso(),
             status="unreachable", exit_code=alert.EXIT_UNREACHABLE, n_errors=1,
             error_summary=str(exc),
@@ -163,7 +163,7 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_
     store.write_run_raw(run_id, started_at, 200, list_body)
 
     drift = check_list_schema(list_body)
-    items = list_body.get("outageMessage") or []
+    items = (list_body.get("outageMessage") if isinstance(list_body, dict) else None) or []
     if not isinstance(items, list):
         items = []
 
@@ -199,12 +199,13 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_
         except AuthError as exc:
             # The key died mid-run. Stop immediately rather than burning
             # through hundreds of guaranteed-failing requests.
-            store.record_run(
+            store.finish_run(
                 run_id=run_id, started_at_utc=started_at,
                 finished_at_utc=utc_now_iso(), status="auth_error",
                 exit_code=alert.EXIT_AUTH, n_listed=len(listed_ids),
                 n_detail_fetched=fetched, n_detail_skipped=skipped,
-                n_errors=1, error_summary=str(exc),
+                n_errors=len(errors) + 1,
+                error_summary="; ".join(errors[:9] + [str(exc)]),
             )
             return alert.fail(
                 alert.auth_banner(client.masked_key, str(exc)), alert.EXIT_AUTH
@@ -220,6 +221,8 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_
         drift.extend(
             f"outage {outage_id}: {p}" for p in check_detail_schema(body)
         )
+        if not isinstance(body, dict):
+            continue
         store.apply_detail(observed_at, normalize_detail(body))
         # Committed per detail so a run killed outright still leaves the
         # database knowing what it fetched; the next run then carries on
@@ -248,7 +251,7 @@ def _collect(store: Store, client: EsbClient, delay_ms: int, stop: list, budget_
     else:
         status, code = "ok", alert.EXIT_OK
 
-    store.record_run(
+    store.finish_run(
         run_id=run_id, started_at_utc=started_at, finished_at_utc=utc_now_iso(),
         status=status, exit_code=code, n_listed=len(listed_ids),
         n_detail_fetched=fetched, n_detail_skipped=skipped, n_errors=failed,
@@ -284,7 +287,7 @@ def run_check(client: EsbClient | None = None) -> int:
         body = client.get_outage_list()
     except AuthError as exc:
         return alert.fail(alert.auth_banner(client.masked_key, str(exc)), alert.EXIT_AUTH)
-    except (TransientError, ApiError) as exc:
+    except (TransientError, ApiError, NotFound) as exc:
         return alert.fail(alert.unreachable_banner(str(exc)), alert.EXIT_UNREACHABLE)
 
     problems = check_list_schema(body)