From 0f4f15eb6bcb5f64d5b593070838a327e06a06bc Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 27 Jul 2026 20:27:35 +0200 Subject: [PATCH] test(storage): prove raw-authority fair scheduling and CAS-typed retry (hjpx AC3/AC4) Problem: polylogue-hjpx AC3 requires bounded raw-materialization scheduling that never starves large valid work behind cheap components, and AC4 requires transient failures to remain retryable under a stable plan id while CAS conflicts/incomparable authority stay typed, durable, and non-mutating. hjpx.1 (PR #2961) landed the fair-rotation and plan/outcome-conservation machinery, and later PRs (#3029/#3031/#3034/#3043) hardened it further, but no regression test proved (a) component ordering is size-agnostic rather than cheap-first, (b) a retryable plan keeps its plan_id across a fail-then-succeed cycle, or (c) a genuine CAS-conflict RuntimeError from revision_application.py surfaces through the reconciler as a typed/durable/non-mutating outcome rather than silently vanishing or partially applying. What changed: added three regression tests to tests/unit/storage/test_repair.py: - test_raw_materialization_ordering_is_size_agnostic_and_does_not_starve_large_work: constructs one large-but-executable component (oldest by acquisition) plus five small ones; proves fair age-based ordering selects the large component first, and a cheap-first mutation of _raw_materialization_ordered_components recreates the starvation AC3 names. - test_raw_materialization_transient_failure_retries_with_same_plan_id_then_succeeds: proves a retryable outcome keeps its plan_id across a fail-then-succeed retry and that no parse residue exists before the retry lands. - test_raw_materialization_cas_conflict_outcome_is_typed_durable_and_non_mutating: injects the exact "CAS rejected a conflicting accepted head" message and proves the resulting outcome is typed (retryable), durably recorded in raw_authority_census_plans (source.db), and leaves no parse/application/head residue in source.db or index.db. All three tests were validated anti-vacuously: temporarily mutating the production code under test (cheap-first ordering; a mutated plan_id in the RETRYABLE outcome branch) made the corresponding test fail for the expected reason, then the mutation was reverted (clean `git diff` on production code). No production code gap was found for AC3/AC4 -- the investigation confirms existing behavior in polylogue/storage/repair.py (_raw_materialization_ordered_components) and the generic exception handler around backfill_historical_revision_evidence already satisfy both invariants. AC5/AC6/AC7 status is recorded in the bead's notes, not re-litigated here. Verification: - devtools test tests/unit/storage/test_repair.py tests/unit/sources/test_revision_backfill.py -> 109 passed - devtools test tests/unit/sources/test_revision_backfill.py -> 44 passed - devtools test -k raw_materialization -> 118 passed - devtools test -k raw_authority -> 83 passed - devtools verify --quick -> 17/17 steps green, exit 0 - Not run: devtools verify --seed-testmon/full (per repo convention a focused test-only change needs the narrow gate, not a full-suite seed; a first attempt was aborted on the coordinator's direction since this change doesn't warrant it). Ref polylogue-hjpx Co-Authored-By: Claude --- tests/unit/storage/test_repair.py | 197 ++++++++++++++++++++++++++++++ 1 file changed, 197 insertions(+) diff --git a/tests/unit/storage/test_repair.py b/tests/unit/storage/test_repair.py index cba00920ea..d939729456 100644 --- a/tests/unit/storage/test_repair.py +++ b/tests/unit/storage/test_repair.py @@ -2249,6 +2249,79 @@ def acquisition_only_order(candidates: Any, *, archive_root: Path) -> list[tuple assert unfair_second == unfair_first +def test_raw_materialization_ordering_is_size_agnostic_and_does_not_starve_large_work( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """hjpx AC3: bounded scheduling must pick components by stable + fairness/age, not by cheapness -- a size-preferring order recreates the + exact starvation failure mode AC3 names ("repeatedly selecting the same + cheap components... starving large valid work"), even though every + component here is independently executable in one pass. + """ + from polylogue.core.enums import Provider + from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + def run(*, prefer_cheap: bool) -> tuple[tuple[str, ...], str]: + root = tmp_path / ("cheap-first" if prefer_cheap else "fair-order") + initialize_active_archive_root(root) + with ArchiveStore.open_existing(root, read_only=False) as archive: + # The large valid component is acquired FIRST (oldest), so fair + # age-based ordering must select it on the very first pass. + large_raw_id = archive.write_raw_payload( + provider=Provider.CODEX, + payload=b'{"type":"session_meta","payload":{"id":"large-valid"}}\n', + source_path="large-valid.jsonl", + acquired_at_ms=1, + ) + for index in range(5): + archive.write_raw_payload( + provider=Provider.CODEX, + payload=f'{{"type":"session_meta","payload":{{"id":"cheap-{index}"}}}}\n'.encode(), + source_path=f"cheap-{index}.jsonl", + acquired_at_ms=index + 2, + ) + with sqlite3.connect(root / "source.db") as source_conn: + # blob_size is scheduling metadata only (parsing reads the tiny + # real payload); this makes the large component "expensive but + # still executable" (well under the execute limit) without + # generating megabytes of fixture bytes. + source_conn.execute( + "UPDATE raw_sessions SET blob_size = ? WHERE raw_id = ?", + (repair_mod.RAW_MATERIALIZATION_EXECUTE_BLOB_LIMIT_BYTES // 2, large_raw_id), + ) + source_conn.commit() + + config = _config(root) + _complete_bounded_raw_census(config, limit=1) + with monkeypatch.context() as mutation: + if prefer_cheap: + + def cheap_first_order(candidates: Any, *, archive_root: Path) -> list[tuple[str, ...]]: + candidate_ids = set(candidates.raw_ids) + source_components = candidates.authority_components or tuple( + (raw_id,) for raw_id in candidates.raw_ids + ) + components = [c for c in source_components if candidate_ids.intersection(c)] + return sorted( + components, + key=lambda component: sum( + candidates.expanded_blob_bytes.get(rid, candidates.raw_blob_bytes.get(rid, 0)) + for rid in component + ), + ) + + mutation.setattr(repair_mod, "_raw_materialization_ordered_components", cheap_first_order) + result = repair_mod.repair_raw_materialization(config, raw_artifact_limit=1) + return result.plan_outcomes[0].input_raw_ids, large_raw_id + + fair_selected, fair_large_id = run(prefer_cheap=False) + cheap_selected, cheap_large_id = run(prefer_cheap=True) + + assert fair_selected == (fair_large_id,) + assert cheap_selected != (cheap_large_id,) + + def test_raw_materialization_isolates_failed_component_and_continues_batch( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -2285,6 +2358,130 @@ def fail_oldest(*args: Any, selected_raw_ids: list[str] | None = None, **kwargs: assert [outcome.status.value for outcome in result.plan_outcomes].count("executed") == 2 +def test_raw_materialization_transient_failure_retries_with_same_plan_id_then_succeeds( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """hjpx AC4: a transient interruption must remain retryable under the + *same* plan id and later succeed once -- not spawn a fresh plan id, and + not silently mutate anything before the retry lands. + """ + from polylogue.core.enums import Provider + from polylogue.sources import revision_backfill + from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + raw_id = archive.write_raw_payload( + provider=Provider.CODEX, + payload=b'{"type":"session_meta","payload":{"id":"transient-target"}}\n', + source_path="transient-target.jsonl", + acquired_at_ms=1, + ) + + original = revision_backfill.backfill_historical_revision_evidence + should_fail = True + + def fail_once(*args: Any, selected_raw_ids: list[str] | None = None, **kwargs: Any) -> Any: + if should_fail and selected_raw_ids == [raw_id]: + raise RuntimeError("OperationalError: database is locked") + return original(*args, selected_raw_ids=selected_raw_ids, **kwargs) + + monkeypatch.setattr(revision_backfill, "backfill_historical_revision_evidence", fail_once) + + config = _config(tmp_path) + first = repair_mod.repair_raw_materialization(config) + assert first.plan_outcomes[0].status.value == "retryable" + assert "database is locked" in first.plan_outcomes[0].reason + first_plan_id = first.plan_outcomes[0].plan_id + + # Non-mutating: the injected failure must not have left any parse/apply + # residue behind before the retry runs. + with sqlite3.connect(tmp_path / "source.db") as source_conn: + assert source_conn.execute("SELECT parsed_at_ms FROM raw_sessions WHERE raw_id = ?", (raw_id,)).fetchone() == ( + None, + ) + + should_fail = False + second = repair_mod.repair_raw_materialization(config) + + assert second.plan_outcomes[0].status.value == "executed" + assert second.plan_outcomes[0].plan_id == first_plan_id + with sqlite3.connect(tmp_path / "source.db") as source_conn: + assert source_conn.execute( + "SELECT parsed_at_ms IS NOT NULL FROM raw_sessions WHERE raw_id = ?", (raw_id,) + ).fetchone() == (1,) + + +def test_raw_materialization_cas_conflict_outcome_is_typed_durable_and_non_mutating( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """hjpx AC4: a CAS conflict/incomparable-authority rejection from the + revision-application layer must surface as a typed, durably-recorded, + non-mutating outcome through the reconciler -- it must not silently + vanish, apply partially, or lose its plan id. + """ + from polylogue.core.enums import Provider + from polylogue.sources import revision_backfill + from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore + from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + raw_id = archive.write_raw_payload( + provider=Provider.CODEX, + payload=b'{"type":"session_meta","payload":{"id":"cas-conflict-target"}}\n', + source_path="cas-conflict-target.jsonl", + acquired_at_ms=1, + ) + + cas_message = ( + "raw revision CAS rejected a conflicting accepted head: " + "logical_source_key='codex:cas-conflict-target' existing(session_id='cas-conflict-target', " + "accepted_raw_id='other-raw') incoming(session_id='cas-conflict-target', accepted_raw_id='" + raw_id + "')" + ) + + def raise_cas_conflict(*args: Any, selected_raw_ids: list[str] | None = None, **kwargs: Any) -> Any: + assert selected_raw_ids == [raw_id] + raise RuntimeError(cas_message) + + monkeypatch.setattr(revision_backfill, "backfill_historical_revision_evidence", raise_cas_conflict) + result = repair_mod.repair_raw_materialization(_config(tmp_path)) + + outcome = result.plan_outcomes[0] + assert outcome.status.value == "retryable" + assert "CAS rejected a conflicting accepted head" in outcome.reason + + assert result.census_receipt is not None + with sqlite3.connect(tmp_path / "source.db") as source_conn: + # Durable: the typed outcome is recorded against the exact plan id in + # the durable source-tier ledger, not only in the in-memory receipt. + recorded = source_conn.execute( + """ + SELECT outcome_status, reason + FROM raw_authority_census_plans + WHERE census_id = ? AND plan_id = ? + """, + (result.census_receipt.census_id, outcome.plan_id), + ).fetchone() + assert recorded == ("retryable", outcome.reason) + # Non-mutating: no parse residue exists for the raw the CAS + # rejection blocked. + assert source_conn.execute("SELECT parsed_at_ms FROM raw_sessions WHERE raw_id = ?", (raw_id,)).fetchone() == ( + None, + ) + with sqlite3.connect(tmp_path / "index.db") as index_conn: + # Non-mutating: no application/head state exists in the (rebuildable + # but still write-through) index tier either. + assert ( + index_conn.execute("SELECT COUNT(*) FROM raw_revision_applications WHERE raw_id = ?", (raw_id,)).fetchone()[ + 0 + ] + == 0 + ) + assert index_conn.execute("SELECT COUNT(*) FROM raw_revision_heads").fetchone()[0] == 0 + + def test_raw_materialization_fails_closed_on_plan_conservation_mismatch( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: