diff --git a/cron/incidents.py b/cron/incidents.py index d16ce4710cd68..c8b7378646c63 100644 --- a/cron/incidents.py +++ b/cron/incidents.py @@ -7,7 +7,9 @@ until the error text changes and mints a new one. ``alerted`` means a failure ping actually reached the operator (``alerted_at`` = when the latest one did; the scheduler withholds repeats until ``cron.failure_repeat_alert_hours`` have passed). Incidents share ``cron/executions.db`` with -``cron.executions`` (one ledger file). +``cron.executions`` (one ledger file). ``reopen_count`` tracks recurrence: how many times a +resolved incident re-opened under the same signature. Rows predating the counter carry NULL +(*not measured*) — never a silent 0, which would falsely attest "never reopened". """ from __future__ import annotations @@ -88,11 +90,18 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: alerted_at TEXT, closed_at TEXT, error TEXT NOT NULL, - output_file TEXT + output_file TEXT, + reopen_count INTEGER )""" ) # Ledgers created before the alert-once gate lack ``alerted_at``; add it in place. add_column_if_missing(conn, "cron_incidents", "alerted_at", "alerted_at TEXT") + # Ledgers predating the reopen counter lack ``reopen_count``. The ddl is deliberately NULLABLE + # with NO DEFAULT: ``add_column_if_missing`` is a raw ALTER TABLE pass-through, and a DEFAULT + # would backfill every existing row with the default value — silently attesting "never + # reopened" for rows that were never counted. Existing rows land NULL (not measured); the + # write path sets 0 on INSERT and increments on reopen. + add_column_if_missing(conn, "cron_incidents", "reopen_count", "reopen_count INTEGER") conn.execute( "CREATE INDEX IF NOT EXISTS idx_cron_incidents_job " "ON cron_incidents(job_id)" @@ -158,10 +167,13 @@ def upsert_incident( output_file: Optional[str] = None, ) -> tuple[str, bool]: """Record (or refresh) the incident for ``job_id`` + ``error``; returns ``(incident_id, - is_new)``. An existing row for the signature refreshes + reopened)``, where ``reopened`` is True *only* when an existing ``resolved`` incident re-opened. + A first-ever INSERT and a plain refresh of an open incident both return False. An existing row + for the signature refreshes ``last_seen_at``/``error``/``output_file`` and keeps its state — a ``closed`` incident stays closed, while a ``resolved`` one (job recovered, then broke the same way again) re-opens as - ``detected`` so the operator is alerted once more. A changed error text mints a new incident.""" + ``detected``, clears its alert/close stamps, and increments ``reopen_count`` so the operator is + alerted once more with the recurrence visible. A changed error text mints a new incident.""" job_id = str(job_id or "") sig = _error_signature(job_id, error) stored_error = _redact_error(error) @@ -176,25 +188,33 @@ def upsert_incident( ).fetchone() if row is not None: reopen = row["state"] == "resolved" - conn.execute( - """UPDATE cron_incidents - SET last_seen_at=?, error=?, output_file=?, - state=CASE WHEN state='resolved' THEN 'detected' ELSE state END, - alerted_at=CASE WHEN state='resolved' THEN NULL ELSE alerted_at END, - closed_at=CASE WHEN state='resolved' THEN NULL ELSE closed_at END - WHERE id=?""", - (now, stored_error, output_file, incident_id), - ) + if reopen: + # COALESCE: a pre-counter row (NULL) starts its history at 1, not at NULL+1. + conn.execute( + """UPDATE cron_incidents + SET last_seen_at=?, error=?, output_file=?, state='detected', + alerted_at=NULL, closed_at=NULL, + reopen_count=COALESCE(reopen_count, 0) + 1 + WHERE id=?""", + (now, stored_error, output_file, incident_id), + ) + else: + conn.execute( + """UPDATE cron_incidents + SET last_seen_at=?, error=?, output_file=? + WHERE id=?""", + (now, stored_error, output_file, incident_id), + ) return incident_id, reopen conn.execute( """INSERT INTO cron_incidents (id, job_id, error_sig, state, failure_type, - first_seen_at, last_seen_at, error, output_file) - VALUES (?, ?, ?, 'detected', ?, ?, ?, ?, ?)""", + first_seen_at, last_seen_at, error, output_file, reopen_count) + VALUES (?, ?, ?, 'detected', ?, ?, ?, ?, ?, 0)""", (incident_id, job_id, sig, failure_type, now, now, stored_error, output_file), ) - return incident_id, True + return incident_id, False def set_incident_state(incident_id: str, state: str) -> bool: diff --git a/cron/scheduler.py b/cron/scheduler.py index 4082d766f35cc..c5c9ff1130737 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -397,7 +397,7 @@ def _upsert_incident_for_failure( try: from cron.incidents import get_incident, upsert_incident - incident_id, _is_new = upsert_incident( + incident_id, _reopened = upsert_incident( job["id"], str(error or ""), job_name=job.get("name"), output_file=output_file) incident = get_incident(incident_id) state = incident.get("state") if incident else None diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index f1403a40289bf..30a5819100c0e 100644 --- a/hermes_cli/cron.py +++ b/hermes_cli/cron.py @@ -383,9 +383,13 @@ def cron_incidents(args) -> int: error_text = re.sub(r"\s+", " ", inc.get("error") or "").strip() if len(error_text) > 160: error_text = error_text[:157].rstrip() + "..." + # A pre-counter row (NULL) says so; 0 is a measured truth, not an absence. + reopen_display = ("pre-counter" if inc.get("reopen_count") is None + else f"{inc['reopen_count']}x") rows = [("Job", inc["job_id"]), ("Type", inc.get("failure_type", "unknown")), ("First seen", inc.get("first_seen_at", "?")), - ("Last seen", inc.get("last_seen_at", "?")), ("Error", error_text), + ("Last seen", inc.get("last_seen_at", "?")), + ("Reopened", reopen_display), ("Error", error_text), ("Output", inc.get("output_file"))] print(f" {color(inc['id'], Colors.YELLOW)} {state_display}") for label, value in rows: diff --git a/hermes_cli/sqlite_util.py b/hermes_cli/sqlite_util.py index 952a49b9a7242..9fc328346af59 100644 --- a/hermes_cli/sqlite_util.py +++ b/hermes_cli/sqlite_util.py @@ -85,6 +85,13 @@ def add_column_if_missing(conn: sqlite3.Connection, table: str, column: str, ddl ``column`` is the human-readable name for the call site; ``ddl`` carries the actual definition. See #21708. + + CAUTION — this is a raw ``ALTER TABLE`` pass-through: the only guard is swallowing the + concurrent-migrator duplicate-column error. A ``DEFAULT`` in ``ddl`` is applied to EVERY + EXISTING row, so a default quietly attests its value about all pre-migration history (e.g. + ``... NOT NULL DEFAULT 0`` makes unmeasured rows read "zero"). When old rows must read as + unmeasured, pass a nullable ``ddl`` with NO DEFAULT and set values explicitly on the write + path; the review gate is the ddl string at the call site, not this function's name. """ try: conn.execute(f"ALTER TABLE {table} ADD COLUMN {ddl}") diff --git a/tests/cron/test_cron_incidents.py b/tests/cron/test_cron_incidents.py index cc027325e5ae5..47dad8c405567 100644 --- a/tests/cron/test_cron_incidents.py +++ b/tests/cron/test_cron_incidents.py @@ -81,9 +81,11 @@ def fake_deliver(jb, content, adapters=None, loop=None, **kwargs): def test_new_failure_creates_incident_and_is_new(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) - inc_id, is_new = inc.upsert_incident("job-1", "Provider timeout: read timed out") + # ``reopened`` is True only for a resolved incident re-opening; a first-ever row is a + # creation (proven by the row below) and returns False. + inc_id, reopened = inc.upsert_incident("job-1", "Provider timeout: read timed out") - assert is_new is True + assert reopened is False assert inc_id.startswith("job-1_") row = inc.get_incident(inc_id) assert row is not None @@ -116,12 +118,11 @@ def test_incidents_list_newest_instant_first_across_dst_fall_back(monkeypatch, t def test_same_signature_dedups_same_incident(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) - id1, new1 = inc.upsert_incident("job-1", "Provider timeout: read timed out") - id2, new2 = inc.upsert_incident("job-1", "PROVIDER TIMEOUT: read timed out ") + id1, reopened1 = inc.upsert_incident("job-1", "Provider timeout: read timed out") + id2, reopened2 = inc.upsert_incident("job-1", "PROVIDER TIMEOUT: read timed out ") assert id1 == id2, "normalized (case/whitespace) signature must dedup" - assert new1 is True - assert new2 is False + assert reopened1 is False and reopened2 is False, "neither write re-opened anything" assert inc.count_incidents() == 1 # Refresh updates last_seen but never resets an open state. assert inc.get_incident(id1)["state"] == "detected" @@ -131,10 +132,10 @@ def test_error_change_mints_new_incident(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) id1, _ = inc.upsert_incident("job-1", "provider timeout") - id2, new2 = inc.upsert_incident("job-1", "provider rate limit 429") + id2, reopened2 = inc.upsert_incident("job-1", "provider rate limit 429") assert id1 != id2 - assert new2 is True + assert reopened2 is False, "a freshly minted signature is a creation, not a re-open" assert inc.count_incidents() == 2 @@ -231,9 +232,9 @@ def test_acked_signature_stays_closed_on_refresh(monkeypatch, tmp_path): def test_missing_db_no_crash(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) - inc_id, is_new = inc.upsert_incident("job-1", "boom") + inc_id, reopened = inc.upsert_incident("job-1", "boom") - assert is_new is True + assert reopened is False, "creation is not a re-open; the row asserts creation" assert (tmp_path / "cron" / "executions.db").is_file() assert inc.list_incidents() == [inc.get_incident(inc_id)] assert inc.count_incidents() == 1 @@ -425,3 +426,77 @@ def test_alerted_gate_honours_cooldown_opt_out_and_legacy_rows(monkeypatch, tmp_ assert inc.ack_incident(inc_id) is True monkeypatch.setattr(sched, "_failure_repeat_alert_hours", lambda: 0) assert sched._upsert_incident_for_failure(job, "repeat boom") == (True, inc_id) + + + +# ── reopen counter (recurrence visibility) ───────────────────────────────── + + +def test_reopen_count_tracks_recurrence(monkeypatch, tmp_path): + inc = _point_db(monkeypatch, tmp_path) + + iid, reopened = inc.upsert_incident("job-rec", "chronic boom") + assert reopened is False and inc.get_incident(iid)["reopen_count"] == 0 + + # Refresh while open: not a re-open, counter stays. + iid2, reopened = inc.upsert_incident("job-rec", "chronic boom") + assert iid2 == iid and reopened is False + assert inc.get_incident(iid)["reopen_count"] == 0 + + # Recover, then break the same way again: the resolved row re-opens and counts. + assert inc.close_incidents_for_recovered_job("job-rec") == 1 + _, reopened = inc.upsert_incident("job-rec", "chronic boom") + assert reopened is True + row = inc.get_incident(iid) + assert row["state"] == "detected" and row["reopen_count"] == 1 + assert row["alerted_at"] is None and row["closed_at"] is None + + # Chronic incidents grow a history instead of flattening into one shape. + inc.close_incidents_for_recovered_job("job-rec") + _, reopened = inc.upsert_incident("job-rec", "chronic boom") + assert reopened is True and inc.get_incident(iid)["reopen_count"] == 2 + + +def test_pre_counter_rows_migrate_to_null_not_zero(monkeypatch, tmp_path): + """Old-shape ledgers (no ``reopen_count``) must read *not measured*, never a silent 0: a + DEFAULT on the ALTER would attest "never reopened" for rows that were never counted.""" + import sqlite3 + + inc = _point_db(monkeypatch, tmp_path) + path = inc.EXECUTIONS_FILE + path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(path) + conn.execute( + """CREATE TABLE cron_incidents ( + id TEXT PRIMARY KEY, job_id TEXT NOT NULL, error_sig TEXT NOT NULL, + state TEXT NOT NULL, failure_type TEXT NOT NULL DEFAULT 'unknown', + first_seen_at TEXT NOT NULL, last_seen_at TEXT NOT NULL, acked_at TEXT, + alerted_at TEXT, closed_at TEXT, error TEXT NOT NULL, output_file TEXT)""" + ) + conn.execute( + "INSERT INTO cron_incidents (id, job_id, error_sig, state, first_seen_at, " + "last_seen_at, error) VALUES ('old1', 'job-old', 'deadbeefcafe', 'closed', " + "'2026-01-01T00:00:00-05:00', '2026-01-02T00:00:00-05:00', 'ancient failure')") + conn.commit() + conn.close() + + row = inc.get_incident("old1") # opening the store runs _initialize_schema in place + assert row is not None and row["state"] == "closed" + assert row["reopen_count"] is None, "pre-counter history must not inherit a measured 0" + + +def test_reopen_from_null_pre_counter_counts_one(monkeypatch, tmp_path): + """The COALESCE path: a re-open where the counter is NULL starts at 1, not NULL+1.""" + import sqlite3 + + inc = _point_db(monkeypatch, tmp_path) + iid, _ = inc.upsert_incident("job-null", "boom again") + conn = sqlite3.connect(inc.EXECUTIONS_FILE) + conn.execute("UPDATE cron_incidents SET reopen_count=NULL WHERE id=?", (iid,)) + conn.commit() + conn.close() + + assert inc.close_incidents_for_recovered_job("job-null") == 1 + _, reopened = inc.upsert_incident("job-null", "boom again") + assert reopened is True + assert inc.get_incident(iid)["reopen_count"] == 1