From 366a06b03d41761c39bb67934ad1be4b1cb98bba Mon Sep 17 00:00:00 2001 From: Matt Spinola Date: Sat, 1 Aug 2026 10:07:59 -0400 Subject: [PATCH] fix(vintage): --retry recovers skipped snapshots, not just failed ones `skipped` means no canonicaliser existed for that report type at ingest time, and it was exactly as terminal as `failed`: --pending selects neither, and nothing else ever wrote parse_status back. The design notes promise that adding a canonicaliser later just means re-marking these pending and re-running, and there was no way to do the re-marking. Nearly bit today. The disagg and TFF bytes captured 2026-07-30 sat unparsed for two days because no canonicaliser existed yet, and this morning's run recovered 51,735 observations from them. That only worked because the producer's install predated the drain that marks them `skipped`, so they were still `pending`. On a current install they would have been stranded with no command able to reach them. The live case is the weekly static: every retained copy is `skipped`, so all of them would be unreachable the day it gets a canonicaliser, which is listed as still-deferred work. --retry now resets both states. Re-skipping is harmless and quiet: a source that still has no canonicaliser simply drains again, and the alert fatigue the drain was added to prevent does not apply to a flag somebody typed on purpose. The output names the breakdown so an operator can see which kind was recovered. --retry-failed is kept as an alias. It shipped earlier the same day under the narrower name and the new behaviour is a superset, so nothing anyone already wrote breaks. 259 tests pass, ruff clean. Co-Authored-By: Claude Opus 5 --- docs/design/cot_vintage.md | 2 +- src/cotdata/vintage_cli.py | 47 +++++++++++++++++++++++++----------- tests/test_vintage_ingest.py | 43 +++++++++++++++++++++++++++++++++ 3 files changed, 77 insertions(+), 15 deletions(-) diff --git a/docs/design/cot_vintage.md b/docs/design/cot_vintage.md index 7428c93..6bc0c85 100644 --- a/docs/design/cot_vintage.md +++ b/docs/design/cot_vintage.md @@ -581,7 +581,7 @@ Four further findings, all fixed: Scheduler while the store gained nothing, with `failed` being terminal. Now non-zero, in a single combined message: an earlier draft raised on the failure first and silently swallowed a restatement suspect in the same run, which is the more serious of the two. - **`ingest --retry-failed`** is the way back, and it is the only one: nothing else ever + **`ingest --retry`** is the way back, and it is the only one: nothing else ever wrote `parse_status` back to `pending`, and two separate defects had stranded whole backlogs there. - **Disaggregated and TFF were fetched from 2006**, but cftc.gov serves 404 for diff --git a/src/cotdata/vintage_cli.py b/src/cotdata/vintage_cli.py index ac3390c..8334505 100644 --- a/src/cotdata/vintage_cli.py +++ b/src/cotdata/vintage_cli.py @@ -10,6 +10,7 @@ cotdata-schedule backfill """ import argparse +from collections import Counter from . import config @@ -88,18 +89,32 @@ def _cmd_ingest(args) -> int: snaps = vintage.read_snapshots() if args.snapshot: snaps = [s for s in snaps if s.get("snapshot_id") == args.snapshot] - elif args.retry_failed: - # The ONLY route back from parse_status="failed", which is otherwise terminal: - # --pending does not select it and nothing else ever writes the field back. Two - # separate defects have reached that state en masse (a stale lock, and a path that - # would not resolve), and in both cases the entire backlog was stranded with no - # command able to recover it. Raw bytes are retained, so a retry is always safe. - for s_ in [x for x in snaps if x.get("parse_status") == "failed"]: + elif args.retry: + # The ONLY route back from the two terminal states, since --pending selects + # neither and nothing else ever writes the field back: + # + # failed the parse raised. Retry after fixing the cause. Two separate defects + # have driven whole backlogs here (a stale lock, and a path that would + # not resolve), each time with no command able to recover them. + # skipped no canonicaliser existed for that report type at ingest time. This is + # the one that matters going forward: the design notes promise that + # adding a canonicaliser later just means re-marking these pending and + # re-running, and until now there was no way to do the re-marking. The + # weekly static is the live case, because every retained copy of it is + # skipped and would be stranded the day it gets a canonicaliser. + # + # Raw bytes are retained in both states, so a retry is always safe. Re-skipping is + # harmless and quiet: a source that still has no canonicaliser simply drains again. + stuck = [x for x in snaps if x.get("parse_status") in ("failed", "skipped")] + by_state = Counter(x.get("parse_status") for x in stuck) + for s_ in stuck: vintage.update_snapshot(s_["snapshot_id"], parse_status="pending", parse_error=None) - snaps = vintage.read_snapshots() - snaps = [x for x in snaps if x.get("parse_status") == "pending"] - print(f"vintage ingest: reset {len(snaps)} failed snapshot(s) to pending.") + snaps = [x for x in vintage.read_snapshots() + if x.get("parse_status") == "pending"] + detail = ", ".join(f"{n} {state}" for state, n in sorted(by_state.items())) + print(f"vintage ingest: reset {len(stuck)} snapshot(s) to pending" + + (f" ({detail})." if detail else ".")) elif not args.all_snapshots: # Default to pending even with no flag. A bare `ingest` used to select EVERY # snapshot ever recorded and re-ingest each one, which is not a no-op: replaying @@ -184,7 +199,7 @@ def _cmd_ingest(args) -> int: if n_failed: parts.append(f"{n_failed} snapshot(s) FAILED to parse (raw bytes retained, so " f"nothing is lost: fix the cause and re-run with " - f"'cotdata-vintage ingest --retry-failed', which is the only way back " + f"'cotdata-vintage ingest --retry', which is the only way back " f"since 'failed' is not 'pending')") raise SystemExit( "cotdata-vintage: " + "; ".join(parts) @@ -313,9 +328,13 @@ def main(argv=None) -> int: g.add_argument("--snapshot", default=None, help="Ingest one snapshot id.") g.add_argument("--pending", action="store_true", help="Ingest all parse_status=pending. This is also the default.") - g.add_argument("--retry-failed", action="store_true", dest="retry_failed", - help="Reset parse_status=failed snapshots to pending and re-ingest " - "them. The only way back from a failed parse.") + # --retry-failed kept as an alias: it shipped earlier the same day under that name, + # and the widened behaviour is a superset, so nothing anyone already wrote breaks. + g.add_argument("--retry", "--retry-failed", action="store_true", dest="retry", + help="Reset snapshots stuck in parse_status=failed OR =skipped back to " + "pending and re-ingest them. The only way back from either: " + "--pending selects neither. Use after fixing a parse failure, or " + "after adding a canonicaliser for a report type that was skipped.") g.add_argument("--all-snapshots", action="store_true", dest="all_snapshots", help="Replay EVERY recorded snapshot, including already-ingested ones. " "Only for rebuilding a store from retained raw bytes.") diff --git a/tests/test_vintage_ingest.py b/tests/test_vintage_ingest.py index def2f64..a3a7289 100644 --- a/tests/test_vintage_ingest.py +++ b/tests/test_vintage_ingest.py @@ -408,3 +408,46 @@ def test_a_run_where_every_snapshot_failed_does_not_report_success(store_env): "retrieved_at": "2026-07-31T00:00:00Z"} for i in range(3)]}) with _pytest.raises(SystemExit, match="3 snapshot"): vintage_cli.main(["ingest", "--pending"]) + + +def test_retry_also_recovers_skipped_snapshots(store_env): + """`skipped` means no canonicaliser existed for that report type at ingest time, and + it was as terminal as `failed`: --pending selects neither and nothing else wrote the + field back. The design notes promise that adding a canonicaliser later just means + re-marking these pending and re-running, and there was no way to do the re-marking. + + The live case is the weekly static: every retained copy of it is `skipped`, so all of + them would have been stranded the day it gets a canonicaliser.""" + import pytest as _pytest + + from cotdata import vintage, vintage_cli + vintage._write_manifest({"schema_version": 1, "snapshots": [ + {"snapshot_id": "w1", "report_type": "legacy", "source_kind": "weekly_static", + "local_path": "y", "parse_status": "pending", + "retrieved_at": "2026-07-31T00:00:00Z"}, + {"snapshot_id": "d1", "report_type": "legacy", "source_kind": "annual_zip", + "local_path": "gone.zip", "parse_status": "pending", + "retrieved_at": "2026-07-31T00:00:00Z"}, + ]}) + with _pytest.raises(SystemExit): # w1 skips, d1 fails + vintage_cli.main(["ingest", "--pending"]) + after = {s["snapshot_id"]: s["parse_status"] for s in vintage.read_snapshots()} + assert after == {"w1": "skipped", "d1": "failed"} + + assert vintage_cli.main(["ingest", "--pending"]) == 0 # both terminal + + with _pytest.raises(SystemExit): # --retry reaches both + vintage_cli.main(["ingest", "--retry"]) + again = {s["snapshot_id"]: s["parse_status"] for s in vintage.read_snapshots()} + assert again == {"w1": "skipped", "d1": "failed"} # re-skipping is quiet+safe + + +def test_retry_failed_still_works_as_an_alias(store_env): + """It shipped earlier the same day under the narrower name; the widened behaviour is a + superset, so nothing already written breaks.""" + from cotdata import vintage, vintage_cli + vintage._write_manifest({"schema_version": 1, "snapshots": [ + {"snapshot_id": "w1", "report_type": "legacy", "source_kind": "weekly_static", + "local_path": "y", "parse_status": "skipped", + "retrieved_at": "2026-07-31T00:00:00Z"}]}) + assert vintage_cli.main(["ingest", "--retry-failed"]) == 0