Merge pull request #409 from mcdonc/fix/sync-state-write-crash

fix: sync_state write failure after mark_succeeded should not crash worker

haiku_rag_slim/haiku/rag/ingester/workers/pool.py

@@ -244,25 +244,37 @@ class WorkerPool:
         self._breaker.record_success()
         if was_open:
             logger.info("Worker pool breaker closed after successful probe")
-        if job.op is JobOp.DELETE:
-            await self._sync.delete(job.source_id, job.uri)
-            # A successful DELETE resolves any earlier UPSERT failures for the
-            # same (source_id, uri): the document is gone, the original error
-            # is no longer actionable, the DLQ entry is just visual noise.
-            pruned = await self._jobs.prune_dead(job.source_id, job.uri)
-            if pruned:
-                logger.info(
-                    "Pruned %d dead job(s) for %s after successful DELETE",
-                    pruned,
+        try:
+            if job.op is JobOp.DELETE:
+                await self._sync.delete(job.source_id, job.uri)
+                # A successful DELETE resolves any earlier UPSERT failures for
+                # the same (source_id, uri): the document is gone, the original
+                # error is no longer actionable, the DLQ entry is visual noise.
+                pruned = await self._jobs.prune_dead(job.source_id, job.uri)
+                if pruned:
+                    logger.info(
+                        "Pruned %d dead job(s) for %s after successful DELETE",
+                        pruned,
+                        job.uri,
+                    )
+            else:
+                await self._sync.upsert(
+                    job.source_id,
                     job.uri,
+                    revision=result.revision,
+                    content_hash=result.content_hash,
+                    ingested=True,
                 )
-        else:
-            await self._sync.upsert(
-                job.source_id,
+        except Exception:
+            # The job is already marked succeeded — the document was ingested
+            # correctly. A sync_state write failure means the next sweep may
+            # redundantly re-ingest this URI, but that's better than crashing
+            # the worker and blocking the rest of the queue.
+            logger.exception(
+                "Job %s succeeded but sync_state write failed for %s; "
+                "next sweep may re-ingest",
+                job.id,
                 job.uri,
-                revision=result.revision,
-                content_hash=result.content_hash,
-                ingested=True,
             )
         logger.info(
             "Job %s succeeded in %.2fs: %s", job.id, time.monotonic() - started, job.uri

tests/ingester/test_workers.py

@@ -645,6 +645,42 @@ async def test_breaker_ignores_permanent_errors(client, jobs, sync):
     assert pool.breaker_consecutive_failures == 0
 
 
+# --- sync_state write resilience ---
+
+
+@pytest.mark.asyncio
+async def test_sync_state_write_failure_does_not_crash_worker(
+    client, jobs, sync, monkeypatch
+):
+    """If the sync_state write fails after mark_succeeded, the worker should
+    log the error and continue rather than crashing. The job is already
+    marked succeeded — a stale sync_state just means a redundant re-ingest
+    on the next sweep."""
+    client.create_document_from_source.return_value = Document(
+        id="d", content="x", uri="u", metadata={"md5": "m", "source_revision": "r"}
+    )
+    await jobs.enqueue("src", "u", JobOp.UPSERT)
+
+    original_upsert = sync.upsert
+
+    async def _failing_upsert(*args, **kwargs):
+        # Only fail for ingested=True (the post-success write)
+        if kwargs.get("ingested"):
+            raise OSError("disk full")
+        return await original_upsert(*args, **kwargs)
+
+    monkeypatch.setattr(sync, "upsert", _failing_upsert)
+
+    pool = _pool(client, jobs, sync)
+    # drain_once should complete without raising
+    processed = await pool.drain_once()
+    assert processed == 1
+
+    # Job should still be marked succeeded
+    listed = await jobs.list_jobs(status=JobStatus.SUCCEEDED)
+    assert len(listed) == 1
+
+
 # --- reaper ---