diff --git a/polylogue/storage/repair.py b/polylogue/storage/repair.py index 5bc43c249b..c8d7d425b7 100644 --- a/polylogue/storage/repair.py +++ b/polylogue/storage/repair.py @@ -103,6 +103,16 @@ RAW_MATERIALIZATION_COMMIT_BATCH_SIZE = 20 RAW_MATERIALIZATION_OUTCOME_SAMPLE_LIMIT = 8 _TRANSIENT_LOCK_PARSE_ERROR = "OperationalError: database is locked" +#: polylogue-5iz4: ``MembershipReplayConflictError``'s ``parse_error`` text +#: (``f"{type(exc).__name__}: {exc}"``, set by ``mark_raw_parse_failed`` in +#: ``storage/sqlite/archive_tiers/revision_governance.py``) always starts +#: with this exact type name regardless of the exception's own +#: human-readable message wording, which is not itself stable (it has +#: already drifted once since PR #2718 introduced this guard under +#: different phrasing). Matching on the type name here is what keeps a raw +#: that hit this guard retry-eligible even after the message wording +#: changes again. +_MEMBERSHIP_REPLAY_CONFLICT_ERROR_PREFIX = "MembershipReplayConflictError:" _QUARANTINED_ACCEPTED_RAW_REPAIR_DETAIL = "repair:accepted_quarantined_raw_exact_byte_and_semantic_proof" _QUARANTINED_ACCEPTED_RAW_REPAIR_LIMIT = 100 _QUARANTINED_ACCEPTED_RAW_REPAIR_BLOB_LIMIT_BYTES = 256 * 1024 * 1024 @@ -3910,6 +3920,19 @@ def _raw_materialization_candidate_ids( OR ( r.parse_error LIKE 'decode:%No such file or directory:%' ) + OR ( + -- polylogue-5iz4: MembershipReplayConflictError + -- (storage/sqlite/archive_tiers/revision_governance.py) is a + -- transient, retry-eligible refusal by construction -- a + -- later pass over the SAME durable raw bytes can succeed + -- once sibling evidence resolves or the accepted head + -- itself changes. Matching by exception TYPE name (stable) + -- rather than the human-readable message text (which has + -- already drifted twice since #2718 introduced this guard) + -- is what keeps this retry-eligible even after the guard's + -- own wording changes again. + r.parse_error LIKE '{_MEMBERSHIP_REPLAY_CONFLICT_ERROR_PREFIX}%' + ) ) AND NOT ( COALESCE(r.validation_status, '') = 'skipped' @@ -4143,8 +4166,19 @@ def _raw_materialization_component_stream_safe( def _raw_materialization_retryable_missing_blob_error(parse_error: object) -> bool: if not isinstance(parse_error, str): return False - return parse_error == _TRANSIENT_LOCK_PARSE_ERROR or ( - parse_error.startswith("decode:") and "No such file or directory" in parse_error + return ( + parse_error == _TRANSIENT_LOCK_PARSE_ERROR + or (parse_error.startswith("decode:") and "No such file or directory" in parse_error) + # polylogue-5iz4: MembershipReplayConflictError + # (storage/sqlite/archive_tiers/revision_governance.py) is a + # transient, retry-eligible refusal by construction -- see the SQL + # candidate query's matching clause above for the full rationale. + # This Python-side check re-validates every row the SQL WHERE + # clause already passed and is the true single source of truth for + # retry eligibility (the SQL clause is a pre-filter, not merely an + # optimization the SQL and this function must independently agree, + # or a row that clears the SQL gate is silently re-excluded here). + or parse_error.startswith(_MEMBERSHIP_REPLAY_CONFLICT_ERROR_PREFIX) ) diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 12bb991f21..56466cae27 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -159,6 +159,7 @@ from polylogue.storage.sqlite.archive_tiers.revision_governance import ( ActiveByteRevisionChainError, ArchiveRawParsedWriteResult, + MembershipReplayConflictError, _authorize_full_snapshot_fold, _flush_pending_raw_parse_states, _index_parsed_for_retained_raw, @@ -11590,4 +11591,5 @@ def _month_bucket_end_ms(bucket: str) -> int: "ArchiveStore", "ArchiveSessionSearchHit", "ArchiveSessionSummary", + "MembershipReplayConflictError", ] diff --git a/polylogue/storage/sqlite/archive_tiers/revision_governance.py b/polylogue/storage/sqlite/archive_tiers/revision_governance.py index 2881678ce8..6edf690345 100644 --- a/polylogue/storage/sqlite/archive_tiers/revision_governance.py +++ b/polylogue/storage/sqlite/archive_tiers/revision_governance.py @@ -163,6 +163,31 @@ class ActiveByteRevisionChainError(RuntimeError): """A byte-identical revision chain cannot admit a conflicting sibling.""" +class MembershipReplayConflictError(RuntimeError): + """Membership replay refused to move an accepted head this pass. + + Raised by ``apply_raw_membership_classification`` when it cannot safely + retire or replace the currently accepted ``raw_revision_heads`` row for a + logical identity (an unrelated accepted head, or a head with unresolved + byte-append evidence still hanging off it). This is a **transient, + retry-eligible** refusal, not proof the raw is unparseable: a later pass + over the same durable raw bytes can succeed once sibling evidence + resolves or the accepted head itself changes (polylogue-5iz4). + + A dedicated subclass exists so ``mark_raw_parse_failed``'s ``parse_error`` + text (``f"{type(exc).__name__}: {exc}"``) carries a stable, matchable + marker for retry-eligibility checks (``storage/repair.py``'s raw + materialization candidate query) independent of this class's own + human-readable message wording, which has already drifted twice (#2718's + original "unconvertible byte head" phrasing no longer appears anywhere in + this module) and will keep drifting as the guard is refined. A plain + ``RuntimeError`` gives the retry-candidate query nothing durable to match + on beyond the exact message text, which is how a raw that hit this guard + under old wording got silently excluded from every future rebuild even + after the guard's conditions no longer applied to it. + """ + + class RawRevisionGovernanceHost(Protocol): """The narrow slice of ``ArchiveStore`` this module is allowed to touch. @@ -2444,7 +2469,7 @@ def apply_raw_membership_classification( or persisted_session is None or (persisted_raw != existing_raw_id and persisted_raw not in classified_raw_ids) ): - raise RuntimeError( + raise MembershipReplayConflictError( "membership replay cannot retire an unrelated accepted head: " f"logical_source_key={logical_source_key!r} " f"existing_head(raw_id={existing_raw_id!r}, session_id={str(existing_head[3])!r}, " @@ -2504,7 +2529,7 @@ def apply_raw_membership_classification( (logical_source_key, existing_raw_id, *classified_raw_ids, existing_raw_id), ).fetchone() if dangling_append is not None: - raise RuntimeError( + raise MembershipReplayConflictError( "membership replay cannot replace a head with unresolved byte-append " f"evidence: logical_source_key={logical_source_key!r} " f"existing_head(raw_id={existing_raw_id!r})" diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index e82622843f..ee1fe52f0c 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -4570,6 +4570,15 @@ def session(native_id: str, *texts: str) -> ParsedSession: assert parse_error is None else: assert parse_error is not None + # polylogue-5iz4: this guard's refusal is transient/retry-eligible by + # construction (a later pass over the same durable bytes can succeed + # once sibling evidence resolves), but a plain RuntimeError leaves the + # retry-candidate query (storage/repair.py) nothing stable to match + # once the message text drifts -- exactly what happened to a real + # production session that hit this guard under #2718's original + # wording and was never retried again. MembershipReplayConflictError + # gives that query a message-text-independent marker to key on. + assert parse_error.startswith("MembershipReplayConflictError:") with sqlite3.connect(index_db) as conn: assert conn.execute("SELECT message_count FROM sessions WHERE native_id = 'shared'").fetchone() == (2,) head_after = conn.execute( diff --git a/tests/unit/storage/test_repair.py b/tests/unit/storage/test_repair.py index df8e9d605a..04d67808a3 100644 --- a/tests/unit/storage/test_repair.py +++ b/tests/unit/storage/test_repair.py @@ -416,6 +416,76 @@ def test_raw_materialization_retries_typed_transient_lock_failure(tmp_path: Path ).fetchone() == (1, None) +def test_raw_materialization_retries_membership_replay_conflict_failure(tmp_path: Path) -> None: + """polylogue-5iz4: a MembershipReplayConflictError parse_error is retryable. + + ``apply_raw_membership_classification``'s two head-conflict guards + (storage/sqlite/archive_tiers/revision_governance.py) are explicitly + transient/retry-eligible refusals, not proof a raw is unparseable -- a + later pass over the same durable bytes can succeed once sibling evidence + resolves or the accepted head changes. But the raw materialization + candidate query only recognizes retry-eligible parse_error text through a + literal allowlist; a plain ``RuntimeError`` message (the type both guards + raised before this fix, and the type a real production Codex session's + parse_error was frozen as under PR #2718's now-superseded wording) is + indistinguishable from a genuinely terminal parse failure and is silently + excluded from every future rebuild attempt forever. Anti-vacuity: a + sibling row carrying the OLD plain-``RuntimeError`` text must remain + excluded -- this test would pass vacuously (0 candidates either way) if + the candidate query's parse_error filter were simply deleted instead of + extended. + """ + config = _config(tmp_path) + initialize_archive_database(tmp_path / "source.db", ArchiveTier.SOURCE) + initialize_archive_database(tmp_path / "index.db", ArchiveTier.INDEX) + blob_store = BlobStore(tmp_path / "blob") + retryable_raw_id, retryable_size = blob_store.write_from_bytes(b'{"mapping":{}}') + stale_raw_id, stale_size = blob_store.write_from_bytes(b'{"mapping":{"stale":{}}}') + + with sqlite3.connect(tmp_path / "source.db") as source_conn: + source_conn.executemany( + """ + INSERT INTO raw_sessions ( + raw_id, origin, native_id, source_path, source_index, blob_hash, blob_size, + acquired_at_ms, parsed_at_ms, parse_error + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + [ + ( + retryable_raw_id, + "codex-session", + "native-membership-conflict", + "whale.jsonl", + 0, + bytes.fromhex(retryable_raw_id), + retryable_size, + 2, + 3, + "MembershipReplayConflictError: membership replay cannot replace a head " + "with unresolved byte-append evidence: logical_source_key='codex:whale'", + ), + ( + stale_raw_id, + "codex-session", + "native-stale-plain-runtime-error", + "stale.jsonl", + 0, + bytes.fromhex(stale_raw_id), + stale_size, + 1, + 4, + "RuntimeError: membership replay cannot replace an unconvertible byte head", + ), + ], + ) + source_conn.commit() + + result = repair_mod.repair_raw_materialization(config, dry_run=True) + + assert result.metrics["raw_materialization_candidate_count"] == 1.0 + assert result.metrics["raw_materialization_total_blob_bytes"] == float(retryable_size) + + def test_raw_materialization_split_root_classifies_parsed_sidecar_from_routed_blob(tmp_path: Path) -> None: configured_root = tmp_path / "configured" routed_root = tmp_path / "routed"