fix(daemon): reconcile blob publications periodically

polylogue/daemon/blob_gc_periodic.py

@@ -28,6 +28,8 @@
 
 BLOB_GC_INTERVAL_SECONDS = 900
 BLOB_GC_MAX_BATCH = 200
+BLOB_PUBLICATION_RECONCILIATION_INTERVAL_SECONDS = BLOB_GC_INTERVAL_SECONDS
+BLOB_PUBLICATION_RECONCILIATION_MAX_BATCH = BLOB_GC_MAX_BATCH
 
 
 async def periodic_blob_gc_check(*, catch_up_complete: asyncio.Event | None = None) -> None:
@@ -62,6 +64,39 @@ async def periodic_blob_gc_check(*, catch_up_complete: asyncio.Event | None = No
             logger.warning("blob gc: periodic reclaim failed", exc_info=True)
 
 
+async def periodic_blob_publication_reconciliation_check(*, catch_up_complete: asyncio.Event | None = None) -> None:
+    """Periodically clear only terminal publication reservations.
+
+    The storage reconciler retains unreferenced reservations whose blob is
+    still present. Those unresolved rows are intentionally left for explicit
+    abandonment policy. Referenced and blob-missing rows are safe to clear,
+    but only while the archive-wide publisher exclusion is held.
+    """
+    from polylogue.daemon.cli import _await_catch_up_gate, _reconcile_blob_publications
+
+    await _await_catch_up_gate(catch_up_complete, loop_name="blob publication reconciliation")
+    after_publication_id: str | None = None
+    while True:
+        await asyncio.sleep(BLOB_PUBLICATION_RECONCILIATION_INTERVAL_SECONDS)
+        try:
+            outcome = await _reconcile_blob_publications(
+                actor="maintenance.blob_publication_reconciliation",
+                max_count=BLOB_PUBLICATION_RECONCILIATION_MAX_BATCH,
+                after_publication_id=after_publication_id,
+            )
+            if outcome is None or outcome.scanned < BLOB_PUBLICATION_RECONCILIATION_MAX_BATCH:
+                after_publication_id = None
+            else:
+                after_publication_id = outcome.last_scanned_publication_id
+        except sqlite3.OperationalError as exc:
+            if is_transient_sqlite_lock(exc):
+                logger.info("blob publication reconciliation: archive busy; retrying on next tick: %s", exc)
+                continue
+            logger.warning("blob publication reconciliation: periodic pass failed", exc_info=True)
+        except Exception:
+            logger.warning("blob publication reconciliation: periodic pass failed", exc_info=True)
+
+
 def run_blob_gc_once(source_db_path_arg: Path, blob_dir: Path) -> BlobGCResult | None:
     """Run one bounded daemon blob-GC pass, or ``None`` if the blob store is absent."""
     from polylogue.storage.blob_gc import run_blob_gc_report
@@ -76,6 +111,9 @@ def run_blob_gc_once(source_db_path_arg: Path, blob_dir: Path) -> BlobGCResult |
 __all__ = [
     "BLOB_GC_INTERVAL_SECONDS",
     "BLOB_GC_MAX_BATCH",
+    "BLOB_PUBLICATION_RECONCILIATION_INTERVAL_SECONDS",
+    "BLOB_PUBLICATION_RECONCILIATION_MAX_BATCH",
     "periodic_blob_gc_check",
+    "periodic_blob_publication_reconciliation_check",
     "run_blob_gc_once",
 ]

polylogue/daemon/cli.py

@@ -76,6 +76,7 @@
     from polylogue.daemon.parse_prefetch import DaemonParseStage
     from polylogue.product.raw_authority import RawMaterializationCounts
     from polylogue.sources.revision_backfill import RawParsePrefetchCache
+    from polylogue.storage.blob_publication import BlobPublicationReconciliation
 
 logger = get_logger(__name__)
 _CONVERGENCE_DEBT_RETRY_INTERVAL_SECONDS = 60
@@ -330,6 +331,7 @@ async def _await_catch_up_gate(
     "fts identity drift recompute",
     "fts orphan audit",
     "blob gc check",
+    "blob publication reconciliation",
     "secret scan sweep",
 )
 _SCHEMA_BLOCKED_OPTIONAL_DRIVE_CATCHUP_LOOP_NAME = "drive source catch-up"
@@ -1190,23 +1192,31 @@ async def _bridge_catch_up_complete(
     target.set()
 
 
-async def _reconcile_blob_publications() -> None:
-    """Classify crash-left publication reservations before source catch-up."""
+async def _reconcile_blob_publications(
+    *,
+    actor: str = "startup.blob_publications",
+    max_count: int | None = None,
+    after_publication_id: str | None = None,
+) -> BlobPublicationReconciliation | None:
+    """Classify crash-left publication reservations against the active archive."""
     from polylogue.paths import archive_root
+    from polylogue.storage.archive_identity import resolve_active_index_path
     from polylogue.storage.blob_publication import reconcile_blob_publication_reservations_under_exclusion
 
     root = archive_root()
     if not (root / "source.db").exists():
-        return
+        return None
     # Reconciliation only clears rows under a live ArchiveWriterExclusion; the
     # `_under_exclusion` entry point acquires it itself so this startup call
     # cannot silently regress into a no-op reconciliation (polylogue-qs0a).
     outcome = await daemon_write_coordinator().run_sync(
-        "startup.blob_publications",
+        actor,
         reconcile_blob_publication_reservations_under_exclusion,
         root / "source.db",
         root / "blob",
-        index_db_path=root / "index.db",
+        index_db_path=resolve_active_index_path(root),
+        max_count=max_count,
+        after_publication_id=after_publication_id,
     )
     if (
         outcome.cleared_referenced
@@ -1230,6 +1240,7 @@ async def _reconcile_blob_publications() -> None:
             "blob publications: retained %d receipt(s) for inspection or explicit abandonment",
             retained,
         )
+    return outcome
 
 
 def _drain_raw_materialization_once(
@@ -2450,7 +2461,10 @@ async def run_daemon_services(
             from polylogue.daemon.antigravity_conversation_acquisition import (
                 periodic_antigravity_conversation_acquisition_check,
             )
-            from polylogue.daemon.blob_gc_periodic import periodic_blob_gc_check
+            from polylogue.daemon.blob_gc_periodic import (
+                periodic_blob_gc_check,
+                periodic_blob_publication_reconciliation_check,
+            )
             from polylogue.daemon.convergence import DaemonConverger
             from polylogue.daemon.convergence_stages import make_default_convergence_stages
             from polylogue.daemon.embedding_backlog import (
@@ -2499,6 +2513,7 @@ async def run_daemon_services(
                 periodic_fts_identity_drift_recompute(catch_up_complete=catch_up_complete_gate),
                 periodic_fts_orphan_audit(catch_up_complete=catch_up_complete_gate),
                 periodic_blob_gc_check(catch_up_complete=catch_up_complete_gate),
+                periodic_blob_publication_reconciliation_check(catch_up_complete=catch_up_complete_gate),
                 periodic_secret_scan_sweep(catch_up_complete=catch_up_complete_gate),
                 periodic_antigravity_conversation_acquisition_check(catch_up_complete=catch_up_complete_gate),
             ]

polylogue/storage/blob_publication.py

@@ -52,6 +52,8 @@ class BlobPublicationReconciliation:
     retained_referenced: int = 0
     retained_missing: int = 0
     unresolved: int = 0
+    scanned: int = 0
+    last_scanned_publication_id: str | None = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -279,10 +281,14 @@ def inspect_blob_publication_receipts(
     blob_root: Path,
     *,
     index_db_path: Path | None = None,
+    max_count: int | None = None,
+    after_publication_id: str | None = None,
 ) -> tuple[BlobPublicationInspection, ...]:
-    """Return every receipt with its current path/reference evidence."""
+    """Return receipt evidence, optionally bounded by a stable ID cursor."""
     from polylogue.storage.archive_identity import ArchiveLocation
 
+    if max_count is not None and max_count <= 0:
+        raise ValueError("max_count must be positive when provided")
     source_conn = sqlite3.connect(f"file:{source_db_path}?mode=ro", uri=True)
     index_conn: sqlite3.Connection | None = None
     try:
@@ -295,13 +301,34 @@ def inspect_blob_publication_receipts(
         store = BlobStore(blob_root)
         if not _table_exists(source_conn, "blob_publication_reservations"):
             return ()
-        rows = source_conn.execute(
-            """
-            SELECT publication_id, blob_hash, size_bytes, publisher_id, reserved_at_ms
-            FROM blob_publication_reservations
-            ORDER BY reserved_at_ms, publication_id
-            """
-        ).fetchall()
+        if max_count is None and after_publication_id is None:
+            rows = source_conn.execute(
+                """
+                SELECT publication_id, blob_hash, size_bytes, publisher_id, reserved_at_ms
+                FROM blob_publication_reservations
+                ORDER BY reserved_at_ms, publication_id
+                """
+            ).fetchall()
+        else:
+            predicates = ""
+            parameters: list[object] = []
+            if after_publication_id is not None:
+                predicates = "WHERE publication_id > ?"
+                parameters.append(after_publication_id)
+            limit = ""
+            if max_count is not None:
+                limit = "LIMIT ?"
+                parameters.append(max_count)
+            rows = source_conn.execute(
+                f"""
+                SELECT publication_id, blob_hash, size_bytes, publisher_id, reserved_at_ms
+                FROM blob_publication_reservations
+                {predicates}
+                ORDER BY publication_id
+                {limit}
+                """,
+                parameters,
+            ).fetchall()
         return tuple(
             BlobPublicationInspection(
                 publication_id=str(row["publication_id"]),
@@ -326,12 +353,16 @@ def reconcile_blob_publication_reservations(
     *,
     index_db_path: Path | None = None,
     writer_exclusion: ArchiveWriterExclusion | None = None,
+    max_count: int | None = None,
+    after_publication_id: str | None = None,
 ) -> BlobPublicationReconciliation:
     """Classify receipts; clear safe rows only with archive-wide exclusion."""
     inspections = inspect_blob_publication_receipts(
         source_db_path,
         blob_root,
         index_db_path=index_db_path,
+        max_count=max_count,
+        after_publication_id=after_publication_id,
     )
     may_clear = (
         writer_exclusion is not None
@@ -379,6 +410,8 @@ def reconcile_blob_publication_reservations(
         retained_referenced=retained_referenced,
         retained_missing=retained_missing,
         unresolved=unresolved,
+        scanned=len(inspections),
+        last_scanned_publication_id=inspections[-1].publication_id if inspections else None,
     )
 
 
@@ -387,6 +420,8 @@ def reconcile_blob_publication_reservations_under_exclusion(
     blob_root: Path,
     *,
     index_db_path: Path | None = None,
+    max_count: int | None = None,
+    after_publication_id: str | None = None,
 ) -> BlobPublicationReconciliation:
     """Reconcile receipts while holding archive-wide publisher exclusion.
 
@@ -403,6 +438,8 @@ def reconcile_blob_publication_reservations_under_exclusion(
             blob_root,
             index_db_path=index_db_path,
             writer_exclusion=exclusion,
+            max_count=max_count,
+            after_publication_id=after_publication_id,
         )