diff --git a/polylogue/daemon/cli.py b/polylogue/daemon/cli.py index 09e0b9860a..88a5d69eaa 100644 --- a/polylogue/daemon/cli.py +++ b/polylogue/daemon/cli.py @@ -2016,7 +2016,9 @@ def _browser_capture_spool_has_pending_files() -> bool: if not spool.exists(): return False now = time.time() - cursor_store = CursorStore(_active_index_db_path()) + # Read-only probe: the writer initialized the ops tier at startup, and + # re-initializing here would write against the writer lease. + cursor_store = CursorStore(_active_index_db_path(), initialize=False) for path in spool.rglob("*.json"): try: stat = path.stat() diff --git a/polylogue/daemon/status.py b/polylogue/daemon/status.py index 7f08fc5335..72a5d21867 100644 --- a/polylogue/daemon/status.py +++ b/polylogue/daemon/status.py @@ -310,6 +310,7 @@ class RawFrontierIntegrity(BaseModel): cursor_ahead_samples: list[dict[str, object]] = Field(default_factory=list) cursor_authority_gap_count: int = 0 cursor_authority_gap_samples: list[dict[str, object]] = Field(default_factory=list) + cursor_authority_deferred_count: int = 0 cursor_ahead_reason: str = "" diff --git a/polylogue/readiness/__init__.py b/polylogue/readiness/__init__.py index cd905d4e04..2585bd3f9e 100644 --- a/polylogue/readiness/__init__.py +++ b/polylogue/readiness/__init__.py @@ -644,6 +644,7 @@ def _raw_frontier_integrity_check(projection: RawFrontierIntegrityProjection) -> "cursor_head_comparison_count": projection.cursor_head_comparison_count, "cursor_ahead_comparison_count": projection.cursor_ahead_comparison_count, "cursor_authority_gap_count": projection.cursor_authority_gap_count, + "cursor_authority_deferred_count": projection.cursor_authority_deferred_count, } issue_count = ( projection.broken_head_count diff --git a/polylogue/readiness/capability.py b/polylogue/readiness/capability.py index 883a7f1182..9701798d00 100644 --- a/polylogue/readiness/capability.py +++ b/polylogue/readiness/capability.py @@ -318,6 +318,7 @@ def component_from_raw_frontier_integrity(payload: Mapping[str, Any] | None) -> "cursor_head_comparison_count": _raw_frontier_count(data.get("cursor_head_comparison_count")), "cursor_ahead_comparison_count": _raw_frontier_count(data.get("cursor_ahead_comparison_count")), "cursor_authority_gap_count": _raw_frontier_count(data.get("cursor_authority_gap_count")), + "cursor_authority_deferred_count": _raw_frontier_count(data.get("cursor_authority_deferred_count")), }, caveats=tuple(caveats), repair_hint=None if state == CapabilityReadinessState.READY else "polylogue ops status --full", @@ -377,6 +378,7 @@ def unknown_raw_frontier_integrity_projection(reason: str) -> RawFrontierIntegri "cursor_head_comparison_count", "cursor_ahead_comparison_count", "cursor_authority_gap_count", + "cursor_authority_deferred_count", ) _RAW_FRONTIER_SAMPLE_COUNT_KEYS = ( ("broken_head_samples", "broken_head_count"), @@ -401,6 +403,9 @@ def _validate_raw_frontier_projection( ) -> tuple[dict[str, Any] | None, str | None]: """Validate and aggregate one complete canonical frontier projection.""" + # The deferred count is informational: an older producer that omits it + # still describes a complete projection. + payload = {"cursor_authority_deferred_count": 0, **payload} missing = required_keys.difference(payload) if missing: return None, f"missing field(s): {', '.join(sorted(missing))}" @@ -498,9 +503,10 @@ def raw_frontier_source_selection_block_reason( Raw convergence and reindex both select source rows before they write an index. They therefore require the same complete, healthy frontier proof exposed by the diagnostic/readiness route. ``None`` means every authority - comparison was proven healthy. A deferred or incomparable cursor is a - deliberate terminal exception to the byte comparison, but it remains a - blocking unknown until its authority can be resolved. + comparison was proven healthy. An incomparable cursor is a blocking + unknown until its authority can be resolved; a deferred cursor (a durably + captured tail awaiting the quiet window) is a typed safe state and never + blocks selection of other paths. """ projection = raw_frontier_integrity_projection( diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 141ca7ab60..67161a6ee8 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -198,6 +198,7 @@ if TYPE_CHECKING: from polylogue.api import Polylogue + from polylogue.storage.raw_retention import RawFrontierBlockedPaths logger = get_logger(__name__) @@ -585,6 +586,7 @@ def __init__( sync_runner: LiveBatchSyncRunner | None = None, parse_stage: LiveParseStage | None = None, ) -> None: + self._refused_paths: frozenset[Path] = frozenset() self._polylogue = polylogue self._sources = tuple(sources) self._cursor = cursor @@ -700,8 +702,55 @@ def require_cursor_authority(self, paths: Iterable[Path] | None = None) -> Curso for path in paths ): return None + # The proof names the paths it refuses. Refusing only those keeps one + # anomalous file from stalling the other 45,000; a refusal that no + # path explains still blocks everything. + blocked = self._blocked_source_paths() + if blocked.unattributed_reason is None: + if paths is None: + logger.warning( + "live.watcher: cursor authority refuses %d source path(s); path-less route proceeds: %s", + len(blocked.source_paths), + reason, + ) + return None + selected = [Path(path) for path in paths] + refused = frozenset( + path + for path in selected + if str(path) in blocked.source_paths or str(path.resolve()) in blocked.source_paths + ) + if len(refused) < len(selected): + self._refused_paths = refused + if refused: + logger.warning( + "live.watcher: cursor authority refused %d of %d path(s) in this batch: %s", + len(refused), + len(selected), + reason, + ) + return None raise CursorAuthorityBlockedError(f"live watcher source-selection gate blocked: {reason}") + def admit_paths(self, paths: Iterable[Path]) -> list[Path]: + """The subset of ``paths`` the frontier proof admits, in order. + + Raises when nothing may proceed: every path is refused, or the refusal + is one no path explains. + """ + selected = list(paths) + self.require_cursor_authority(selected) + refused = self._refused_paths + self._refused_paths = frozenset() + return [path for path in selected if path not in refused] + + def _blocked_source_paths(self) -> RawFrontierBlockedPaths: + archive_root = Path(getattr(self._polylogue, "archive_root", self._cursor._db_path.parent)) + from polylogue.storage.archive_readiness import raw_materialization_readiness_snapshot + from polylogue.storage.raw_retention import raw_frontier_blocked_source_paths + + return raw_frontier_blocked_source_paths(archive_root, raw_materialization_readiness_snapshot(archive_root)) + async def ingest_files( self, paths: list[Path], @@ -713,6 +762,11 @@ async def ingest_files( ) -> LiveBatchMetrics: """Ingest files in batch, run post-ingest convergence, and return metrics.""" authorization = self.require_cursor_authority(paths) + refused_paths = self._refused_paths + self._refused_paths = frozenset() + if refused_paths: + paths = [path for path in paths if path not in refused_paths] + skipped_file_count += len(refused_paths) if is_fully_degraded(): # The daemon has been marked structurally unable to ingest (e.g. # schema mismatch detected at preflight or on the first batch). diff --git a/polylogue/sources/live/watcher.py b/polylogue/sources/live/watcher.py index ffbbe49947..6140135ded 100644 --- a/polylogue/sources/live/watcher.py +++ b/polylogue/sources/live/watcher.py @@ -266,6 +266,12 @@ class CatchUpPlan: needed_bytes: int +def _is_retryable_lock_error(exc: sqlite3.OperationalError) -> bool: + """SQLite lock contention, as opposed to a broken database.""" + message = str(exc).lower() + return "database is locked" in message or "database table is locked" in message or "busy" in message + + class LiveWatcher: """Async watcher that ingests grown JSONL files in batches. @@ -598,11 +604,13 @@ async def prepare_catch_up() -> None: # planning pass. ``_plan_catch_up`` can reconcile missing cursors # and rebase matching filesystem observations before ingestion has # a chance to apply its own gate. - self._batch_processor.require_cursor_authority() + admitted = self._batch_processor.admit_paths([candidate.path for candidate in candidates]) await self._run_writer_sync("watcher.catch_up.cursor_initialize", self._cursor.initialize) - self._batch_processor.require_cursor_authority() - logger.info("live.watcher: catch-up scan over %d file(s)", len(candidates)) - plan_holder.append(self._plan_catch_up(candidates)) + admitted = self._batch_processor.admit_paths(admitted) + admitted_set = set(admitted) + planned = tuple(candidate for candidate in candidates if candidate.path in admitted_set) + logger.info("live.watcher: catch-up scan over %d file(s)", len(planned)) + plan_holder.append(self._plan_catch_up(planned)) try: await self._run_coordinated("watcher.catch_up.prefilter", prepare_catch_up) @@ -688,7 +696,23 @@ async def ingest_chunk( ): self._defer_unaccounted_failed_retries(chunk_paths) - await self._run_coordinated("watcher.catch_up.chunk", ingest_chunk) + try: + await self._run_coordinated("watcher.catch_up.chunk", ingest_chunk) + except sqlite3.OperationalError as exc: + if not _is_retryable_lock_error(exc): + raise + # A write that lost a lock race is this chunk's failure, + # never the daemon's death: the cursor retry policy brings + # the chunk back. Rehearsal 2026-09-05 died here while the + # Drive catch-up held source.db for 112 s. + failed += len(chunk_paths) + logger.warning( + "live.watcher: catch-up chunk %d/%d deferred, archive write lost a lock race: %s", + chunk_index, + len(chunks), + exc, + ) + self._defer_unaccounted_failed_retries(chunk_paths) if self._stop.is_set(): await self._emit_catch_up_terminal( operation_id, "stopped", plan, attempted, ingested, failed, stage_timings_s, cycle_started @@ -1008,12 +1032,13 @@ async def _flush_pending(self) -> bool: self._forced_reparse_paths.difference_update(paths) async def flush_batch() -> None: + nonlocal paths # Filtering a changed-file batch invokes cursor reconciliation and # lifecycle actuators, so the source-selection proof must be # consumed before initialization or any stateful decision. - self._batch_processor.require_cursor_authority() + paths = self._batch_processor.admit_paths(paths) await self._run_writer_sync("watcher.live_batch.cursor_initialize", self._cursor.initialize) - self._batch_processor.require_cursor_authority() + paths = self._batch_processor.admit_paths(paths) # Filter to files that actually need work. cursor_records = self._cursor.get_records(paths) needed = [] @@ -1032,11 +1057,18 @@ async def flush_batch() -> None: return logger.info("live.watcher: batching %d changed file(s)", len(needed)) - metrics = await self._ingest_files( - needed, - queued_file_count=len(paths), - skipped_file_count=len(paths) - len(needed), - ) + try: + metrics = await self._ingest_files( + needed, + queued_file_count=len(paths), + skipped_file_count=len(paths) - len(needed), + ) + except sqlite3.OperationalError as exc: + if not _is_retryable_lock_error(exc): + raise + logger.warning("live.watcher: changed-file batch deferred, archive write lost a lock race: %s", exc) + self._defer_unaccounted_failed_retries(needed) + return if metrics is not None: _log_ingest_metrics("live.watcher: changed-file batch", metrics) if ( @@ -1671,7 +1703,7 @@ async def _ingest_files( skipped_file_count: int = 0, ) -> LiveBatchMetrics: """Ingest files through the reusable daemon live batch processor.""" - self._batch_processor.require_cursor_authority() + self._batch_processor.require_cursor_authority(paths) async with self._ingest_lock: async def ingest() -> LiveBatchMetrics: diff --git a/polylogue/storage/raw_retention.py b/polylogue/storage/raw_retention.py index 8326791e76..2194c363e6 100644 --- a/polylogue/storage/raw_retention.py +++ b/polylogue/storage/raw_retention.py @@ -1036,6 +1036,9 @@ class RawFrontierIntegritySnapshot: cursor_ahead_samples: tuple[CursorAheadSample, ...] cursor_authority_gap_count: int cursor_authority_gap_samples: tuple[CursorAuthorityGapSample, ...] + #: Cursors whose durably captured tail awaits the quiet window. A typed, + #: safe state: never a gap, never a reason to block other paths. + cursor_authority_deferred_count: int cursor_ahead_reason: str @property @@ -1066,6 +1069,7 @@ class RawFrontierIntegrityProjection: cursor_ahead_samples: tuple[CursorAheadSample, ...] cursor_authority_gap_count: int cursor_authority_gap_samples: tuple[CursorAuthorityGapSample, ...] + cursor_authority_deferred_count: int cursor_ahead_reason: str @property @@ -1110,6 +1114,7 @@ def to_dict(self) -> dict[str, object]: for sample in self.cursor_ahead_samples ], "cursor_authority_gap_count": self.cursor_authority_gap_count, + "cursor_authority_deferred_count": self.cursor_authority_deferred_count, "cursor_authority_gap_samples": [ { "state": sample.state, @@ -1265,10 +1270,81 @@ def raw_frontier_integrity_projection( cursor_ahead_samples=snapshot.cursor_ahead_samples, cursor_authority_gap_count=snapshot.cursor_authority_gap_count, cursor_authority_gap_samples=snapshot.cursor_authority_gap_samples, + cursor_authority_deferred_count=snapshot.cursor_authority_deferred_count, cursor_ahead_reason=snapshot.cursor_ahead_reason, ) +@dataclass(frozen=True) +class RawFrontierBlockedPaths: + """Which source paths the frontier proof refuses, and what it cannot attribute. + + ``source_paths`` holds every path a violation or authority gap names. A + batch may proceed with its other paths. ``unattributed_reason`` is set + when something blocks that no path explains (an unreadable tier, a raw + with no source path, missing source raws); that still blocks everything. + """ + + source_paths: frozenset[str] + unattributed_reason: str | None + + +def raw_frontier_blocked_source_paths( + archive_root: Path, + raw_materialization_readiness: Mapping[str, object], +) -> RawFrontierBlockedPaths: + """Attribute the frontier proof's refusals to exact source paths.""" + + projection = raw_frontier_integrity_projection( + archive_root, + raw_materialization_readiness, + sample_limit=1_000_000, + ) + if projection.available and projection.overall_status == "healthy": + return RawFrontierBlockedPaths(frozenset(), None) + paths: set[str] = set() + unattributed: list[str] = [] + if projection.missing_source_raw_count: + unattributed.append(projection.missing_source_raw_reason) + if projection.broken_head_status == "unknown" and not projection.broken_head_count: + unattributed.append(projection.broken_head_reason) + if projection.cursor_ahead_status == "unknown" and not ( + projection.cursor_ahead_count or projection.cursor_authority_gap_count + ): + unattributed.append(projection.cursor_ahead_reason) + for ahead in projection.cursor_ahead_samples: + paths.add(ahead.source_path) + for gap in projection.cursor_authority_gap_samples: + if gap.source_path is None: + unattributed.append(gap.reason) + else: + paths.add(gap.source_path) + if projection.broken_head_samples: + raw_ids = {sample.accepted_raw_id for sample in projection.broken_head_samples} + source_db_path = archive_root / "source.db" + try: + from polylogue.storage.sqlite.connection_profile import open_readonly_connection + + conn = open_readonly_connection(source_db_path, validate_schema=False) + except (OSError, sqlite3.Error) as exc: + unattributed.append(f"source tier is unreadable: {exc}") + else: + try: + by_raw = _source_paths_for_raw_ids(conn, raw_ids) + except sqlite3.Error as exc: + unattributed.append(f"source raw path lookup failed: {exc}") + by_raw = {} + finally: + conn.close() + for sample in projection.broken_head_samples: + path = by_raw.get(sample.accepted_raw_id) + if path is None: + unattributed.append(f"broken head {sample.accepted_raw_id} has no source path") + else: + paths.add(path) + return RawFrontierBlockedPaths(frozenset(paths), "; ".join(unattributed) or None) + + def unknown_raw_frontier_integrity_projection( reason: str, *, @@ -1306,6 +1382,7 @@ def unknown_raw_frontier_integrity_projection( cursor_ahead_samples=snapshot.cursor_ahead_samples, cursor_authority_gap_count=snapshot.cursor_authority_gap_count, cursor_authority_gap_samples=snapshot.cursor_authority_gap_samples, + cursor_authority_deferred_count=snapshot.cursor_authority_deferred_count, cursor_ahead_reason=snapshot.cursor_ahead_reason, ) @@ -1370,6 +1447,7 @@ def raw_frontier_integrity_snapshot( cursor_samples, cursor_gap_count, cursor_gap_samples, + cursor_deferred_count, cursor_reason, ) = _check_cursor_ahead_of_accepted(conn, ops_db_path, heads, sample_limit=sample_limit) return RawFrontierIntegritySnapshot( @@ -1386,6 +1464,7 @@ def raw_frontier_integrity_snapshot( cursor_ahead_samples=cursor_samples, cursor_authority_gap_count=cursor_gap_count, cursor_authority_gap_samples=cursor_gap_samples, + cursor_authority_deferred_count=cursor_deferred_count, cursor_ahead_reason=cursor_reason, ) finally: @@ -1407,6 +1486,7 @@ def _unavailable_frontier_integrity_snapshot(reason: str) -> RawFrontierIntegrit cursor_ahead_samples=(), cursor_authority_gap_count=0, cursor_authority_gap_samples=(), + cursor_authority_deferred_count=0, cursor_ahead_reason=reason, ) @@ -1514,18 +1594,19 @@ def _check_cursor_ahead_of_accepted( tuple[CursorAheadSample, ...], int, tuple[CursorAuthorityGapSample, ...], + int, str, ]: try: cursor_map = _ops_cursor_byte_offsets(ops_db_path) except RawRetentionSafetyError as exc: - return "unknown", 0, 0, 0, 0, (), 0, (), str(exc) + return "unknown", 0, 0, 0, 0, (), 0, (), 0, str(exc) try: source_path_by_raw_id = _source_paths_for_raw_ids(conn, {head.accepted_raw_id for head in heads}) except sqlite3.Error as exc: logger.warning("raw frontier integrity: source raw path lookup failed: %s", exc) - return "unknown", 0, 0, 0, 0, (), 0, (), f"source raw path lookup failed: {exc}" + return "unknown", 0, 0, 0, 0, (), 0, (), 0, f"source raw path lookup failed: {exc}" byte_heads_by_path: dict[str, list[_IndexRawRevisionHead]] = {} all_head_paths: set[str] = set() @@ -1560,29 +1641,21 @@ def _check_cursor_ahead_of_accepted( source_paths = _source_paths_for_paths(conn, set(cursor_map)) except sqlite3.Error as exc: logger.warning("raw frontier integrity: cursor source path lookup failed: %s", exc) - return "unknown", 0, 0, 0, 0, (), 0, (), f"cursor source path lookup failed: {exc}" + return "unknown", 0, 0, 0, 0, (), 0, (), 0, f"cursor source path lookup failed: {exc}" try: terminal_artifact_paths = _terminal_artifact_paths(conn, set(cursor_map)) except sqlite3.Error as exc: logger.warning("raw frontier integrity: terminal artifact authority lookup failed: %s", exc) - return "unknown", 0, 0, 0, 0, (), 0, (), f"terminal artifact authority is unreadable: {exc}" + return "unknown", 0, 0, 0, 0, (), 0, (), 0, f"terminal artifact authority is unreadable: {exc}" + deferred_count = 0 for path, cursor in cursor_map.items(): cursor_offset = cursor.byte_offset if cursor.is_deferred: - gap_count += 1 - if len(gaps) < sample_limit: - gaps.append( - CursorAuthorityGapSample( - state="deferred", - source_path=path, - logical_source_key=None, - cursor_byte_offset=cursor_offset, - reason=( - "ingest cursor has durably captured material awaiting authority resolution " - f"through byte {cursor.deferred_end_offset}" - ), - ) - ) + # A deferred cursor is the positive proof of a safe incomplete + # tail: the prefix is accepted, the captured range is recorded, + # and the quiet window resolves it. Counting it as a gap made + # every live host refuse its whole backlog while one file was hot. + deferred_count += 1 continue comparable_heads = byte_heads_by_path.get(path) if not comparable_heads: @@ -1638,6 +1711,8 @@ def _check_cursor_ahead_of_accepted( ) if gap_count: reasons.append(f"{gap_count} cursor/head authority row(s) could not be compared") + if deferred_count: + reasons.append(f"{deferred_count} ingest cursor(s) deferred awaiting the quiet window") return ( status, ahead_count, @@ -1647,6 +1722,7 @@ def _check_cursor_ahead_of_accepted( tuple(samples), gap_count, tuple(gaps), + deferred_count, "; ".join(reasons), ) diff --git a/polylogue/storage/sqlite/archive_tiers/ops_write.py b/polylogue/storage/sqlite/archive_tiers/ops_write.py index 819a307ffc..635003e65e 100644 --- a/polylogue/storage/sqlite/archive_tiers/ops_write.py +++ b/polylogue/storage/sqlite/archive_tiers/ops_write.py @@ -37,7 +37,12 @@ def _record_ops_schema_state(conn: sqlite3.Connection, schema_digest: str) -> None: - """Record the DDL digest used to converge this disposable OPS database.""" + """Record the DDL digest used to converge this disposable OPS database. + + Opening an already-current database must not write: every reader that + constructs a cursor store re-enters this path, and a write here contends + with the daemon's writer lease on the same file. + """ conn.execute( """ CREATE TABLE IF NOT EXISTS polylogue_ops_schema_state ( @@ -45,6 +50,9 @@ def _record_ops_schema_state(conn: sqlite3.Connection, schema_digest: str) -> No ) STRICT """ ) + recorded = [row[0] for row in conn.execute("SELECT schema_digest FROM polylogue_ops_schema_state")] + if recorded == [schema_digest]: + return conn.execute( "DELETE FROM polylogue_ops_schema_state WHERE schema_digest <> ?", (schema_digest,), diff --git a/tests/unit/sources/test_live_watcher.py b/tests/unit/sources/test_live_watcher.py index ff1965affa..1eb9f58bf8 100644 --- a/tests/unit/sources/test_live_watcher.py +++ b/tests/unit/sources/test_live_watcher.py @@ -393,6 +393,38 @@ async def test_cursor_authority_seam_blocks_normal_live_route_before_writes(tmp_ watcher.stop() +@pytest.mark.asyncio +async def test_cursor_authority_refuses_only_the_named_path(tmp_path: Path) -> None: + """One path's violation refuses that path; its healthy siblings still ingest. + + Anti-vacuity: restoring the global refusal (raising whenever the block + reason is non-None) makes this batch raise instead of ingesting the + sibling, and the sibling's raw never lands in source.db. + """ + _processor, watcher, _cursor, blocked_path = _seed_live_cursor_authority_case(tmp_path) + sibling = blocked_path.parent / "sibling.jsonl" + sibling.write_bytes( + json.dumps(_codex_session_meta("session-2")).encode() + + b"\n" + + json.dumps( + _codex_message(message_id="m9", role="user", text="sibling", timestamp="2026-05-01T00:00:00Z") + ).encode() + + b"\n" + ) + before = _live_archive_snapshot(tmp_path) + + metrics = await watcher._ingest_files([blocked_path, sibling]) + + assert metrics.skipped_file_count == 1 + assert metrics.succeeded_file_count == 1 + assert _live_archive_snapshot(tmp_path) != before + with sqlite3.connect(tmp_path / "source.db") as conn: + paths = {row[0] for row in conn.execute("SELECT source_path FROM raw_sessions")} + assert str(sibling) in paths + assert str(blocked_path) in paths # the seeded prefix raw, unchanged + watcher.stop() + + def test_live_ingest_metrics_log_separates_read_bytes_from_candidate_size( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -3750,7 +3782,7 @@ async def fake_ingest(paths: list[Path], **_kwargs: object) -> None: try: test_watcher = cast(Any, watcher) - test_watcher._batch_processor.require_cursor_authority = lambda: None + test_watcher._batch_processor.require_cursor_authority = lambda *args, **kwargs: None test_watcher._needs_work_from_state = lambda *args, **kwargs: False test_watcher._ingest_files = fake_ingest watcher._enqueue(sidecar) @@ -4293,3 +4325,37 @@ async def test_ingest_files_max_pass_seconds_bounds_one_pass_and_preserves_progr assert cursor.get_record(path) is not None with sqlite3.connect(tmp_path / "index.db") as conn: assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] == 3 + + +def test_lock_contention_is_retryable_and_corruption_is_not() -> None: + """Anti-vacuity: treating every OperationalError as retryable would hide a + malformed database behind a warning; treating none as retryable killed + the daemon on 2026-09-05.""" + from polylogue.sources.live.watcher import _is_retryable_lock_error + + assert _is_retryable_lock_error(sqlite3.OperationalError("database is locked")) + assert not _is_retryable_lock_error(sqlite3.OperationalError("database disk image is malformed")) + + +@pytest.mark.asyncio +async def test_catch_up_chunk_losing_a_lock_race_defers_instead_of_dying(tmp_path: Path) -> None: + """A locked archive write fails the chunk and the watcher continues.""" + _processor, watcher, _cursor, source_path = _seed_live_cursor_authority_case(tmp_path, exact_frontier=True) + calls: list[list[Path]] = [] + + async def locked_ingest(paths: list[Path], **_: object) -> None: + calls.append(list(paths)) + raise sqlite3.OperationalError("database is locked") + + watcher._ingest_files = locked_ingest # type: ignore[assignment, method-assign] + deferred: list[list[Path]] = [] + watcher._defer_unaccounted_failed_retries = lambda paths: deferred.append(list(paths)) # type: ignore[method-assign] + watcher._batch_processor.require_cursor_authority = lambda *args, **kwargs: None # type: ignore[method-assign] + watcher._needs_work_from_state = lambda *args, **kwargs: True # type: ignore[method-assign] + watcher._pending_paths.add(source_path) + + assert await watcher._flush_pending() is not None + + assert calls == [[source_path]] + assert deferred == [[source_path]] + watcher.stop() diff --git a/tests/unit/storage/test_archive_tiers_ops_write.py b/tests/unit/storage/test_archive_tiers_ops_write.py index f36ababc3d..35492be02c 100644 --- a/tests/unit/storage/test_archive_tiers_ops_write.py +++ b/tests/unit/storage/test_archive_tiers_ops_write.py @@ -761,3 +761,25 @@ def test_record_route_observation_caps_row_count(tmp_path: Path) -> None: newest = list_route_observations(conn, limit=1) assert newest[0].observation_id == f"obs-{seeded}" assert conn.execute("SELECT 1 FROM route_observations WHERE observation_id = 'obs-0'").fetchone() is None + + +def test_reopening_a_current_ops_db_writes_nothing(tmp_path: Path) -> None: + """A converged ops database is opened read-only by every later initializer. + + Anti-vacuity: restoring the unconditional DELETE/INSERT in + ``_record_ops_schema_state`` commits a transaction on reopen and moves + ``PRAGMA data_version`` as seen from the observer connection. + """ + ops_db = tmp_path / "ops.db" + initialize_archive_database(ops_db, ArchiveTier.OPS) + + observer = sqlite3.connect(ops_db) + try: + before = observer.execute("PRAGMA data_version").fetchone()[0] + observer.execute("SELECT count(*) FROM polylogue_ops_schema_state").fetchone() + initialize_archive_database(ops_db, ArchiveTier.OPS) + after = observer.execute("PRAGMA data_version").fetchone()[0] + finally: + observer.close() + + assert after == before diff --git a/tests/unit/storage/test_raw_retention.py b/tests/unit/storage/test_raw_retention.py index fdb320eb95..978ff66046 100644 --- a/tests/unit/storage/test_raw_retention.py +++ b/tests/unit/storage/test_raw_retention.py @@ -2524,15 +2524,56 @@ def test_raw_frontier_integrity_snapshot_classifies_deferred_cursor_separately_f with sqlite3.connect(source_db) as conn: snapshot = raw_frontier_integrity_snapshot(conn, index_db_path=index_db, ops_db_path=ops_db) - assert snapshot.cursor_ahead_status == "unknown" + assert snapshot.cursor_ahead_status == "healthy" assert snapshot.cursor_ahead_count == 0 assert snapshot.cursor_head_comparison_count == 0 - assert snapshot.cursor_authority_gap_count == 1 - sample = snapshot.cursor_authority_gap_samples[0] - assert sample.state == "deferred" - assert sample.cursor_byte_offset == 10 - assert "awaiting authority resolution" in sample.reason - assert snapshot.overall_status == "unknown" + assert snapshot.cursor_authority_gap_count == 0 + assert snapshot.cursor_authority_gap_samples == () + assert snapshot.cursor_authority_deferred_count == 1 + assert "deferred awaiting the quiet window" in snapshot.cursor_ahead_reason + assert snapshot.overall_status == "healthy" + + +def test_deferred_cursor_never_blocks_source_selection(tmp_path: Path) -> None: + """A hot file's deferred tail is a typed safe state, not a gate for the backlog. + + Anti-vacuity: counting the deferred cursor as an authority gap again turns + the block reason non-None and this test red. On a live host some session + file is always being appended, so the old law refused every catch-up. + """ + from polylogue.readiness.capability import raw_frontier_source_selection_block_reason + + source_db = tmp_path / "source.db" + index_db = tmp_path / "index.db" + ops_db = tmp_path / "ops.db" + source_path = tmp_path / "deferred.jsonl" + source_path.write_text("{}\n", encoding="utf-8") + initialize_archive_database(source_db, ArchiveTier.SOURCE) + initialize_archive_database(index_db, ArchiveTier.INDEX) + with sqlite3.connect(source_db) as conn: + _insert_revision_raw( + conn, + raw_id="raw-baseline", + source_path=source_path, + acquired_at_ms=1, + kind="full", + source_revision="revision-0", + generation=0, + blob_size=10, + ) + conn.commit() + _seed_index_authority( + index_db, + session_raw_id="raw-baseline", + accepted_raw_id="raw-baseline", + accepted_revision="revision-0", + generation=0, + frontier=10, + append_end_offset=None, + ) + _seed_ops_cursor(ops_db, source_path=source_path, byte_offset=10, deferred_end_offset=20) + + assert raw_frontier_source_selection_block_reason(tmp_path) is None def test_raw_frontier_integrity_projection_preserves_violation_when_sibling_is_unknown(tmp_path: Path) -> None: