diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py index 64896ac4..25881ec9 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py +++ b/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 diff --git a/tests/ingester/test_workers.py b/tests/ingester/test_workers.py index 1ea0ea3a..37065f61 100644 --- a/tests/ingester/test_workers.py +++ b/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 ---