diff --git a/CHANGELOG.md b/CHANGELOG.md index ece58266..a40739be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ ### Fixed - SQLite ingester queue runs with a multi-connection pool (`pool_size=5, max_overflow=5`) instead of a single connection. API reads (`/stats`, `/jobs`) no longer time out with `QueuePool limit of size 1 reached` while workers hold the connection. +- Permanently-failed ingester documents (revisioned sources) record their revision in `sync_state`, so discovery no longer re-enqueues them every sweep; re-attempted only when the file's revision changes or via explicit retry/rebuild. ## [0.58.0] - 2026-06-15 diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py index 1e8e6079..a4dc8c76 100644 --- a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py +++ b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py @@ -461,10 +461,12 @@ class SyncStateRepo: async def get_revision_snapshot(self, source_id: str) -> dict[str, str]: """uri -> revision map for URIs that have a stored revision. Sources compare current revision against this map to decide UPSERT vs - UNCHANGED. Rows without a revision (HTTP without ETag, or a worker - that didn't complete) are excluded — they have no revision to - compare against; the closing-loop DELETE diff uses list_known_uris - instead.""" + UNCHANGED. A stored revision means the file was accounted for at that + revision — successfully ingested OR permanently failed; both suppress + re-enqueue until the revision changes. Rows without a revision (HTTP + without ETag, or a worker that didn't complete) are excluded — they + have no revision to compare against; the closing-loop DELETE diff uses + list_known_uris instead.""" query = sa.select(sync_state.c.uri, sync_state.c.revision).where( sync_state.c.source_id == source_id, sync_state.c.revision.is_not(None), diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py index 6eb26066..464be50e 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py @@ -218,8 +218,37 @@ class WorkerPool: logger.info("Job %s released back to queue on cancel", job.id) raise except PermanentError as e: - await self._jobs.mark_dead(job.id, str(e), worker_id) + if not await self._jobs.mark_dead(job.id, str(e), worker_id): + logger.warning( + "Job %s lost claim before mark_dead (likely reaper race); " + "letting the re-claiming worker drive", + job.id, + ) + return logger.info("Job %s dead (permanent): %s", job.id, e) + # Record the failed revision so discovery treats the unchanged file as + # accounted-for and stops re-enqueuing it every sweep. sync_state.revision + # means "last accounted-for revision" — ingested OR permanently failed. + # Revision-less sources (no ETag) can't be suppressed this way. + if job.revision is not None: + try: + await self._sync.upsert( + job.source_id, + job.uri, + revision=job.revision, + content_hash=job.content_hash, + ingested=False, + ) + except Exception: + # The job is already dead. A failed marker write only means + # the next sweep may re-enqueue this URI — not worth crashing + # the worker and shrinking the pool over. + logger.exception( + "Job %s dead but failure marker write failed for %s; " + "next sweep may re-enqueue", + job.id, + job.uri, + ) return except TransientError as e: breaker = self._breaker_for(job.source_id) diff --git a/tests/ingester/test_run_batch.py b/tests/ingester/test_run_batch.py index 0491916e..d9643c12 100644 --- a/tests/ingester/test_run_batch.py +++ b/tests/ingester/test_run_batch.py @@ -185,6 +185,10 @@ async def test_run_batch_recovered_doc_is_not_counted_as_dead(tmp_path, use_clie healthy = _mock_client() use_client(healthy) + # Bump the file's revision so discovery re-attempts it. A permanent failure + # records the revision in sync_state, so a plain re-run no longer retries an + # unchanged file — recovery needs the content (mtime) to change. + (tmp_path / "a.md").write_text("hello again") second = await IngesterApp(config=config, db_path=db_path).run_batch() assert second.dead == 0 assert second.succeeded == 1 diff --git a/tests/ingester/test_workers.py b/tests/ingester/test_workers.py index b157e071..cd0fd052 100644 --- a/tests/ingester/test_workers.py +++ b/tests/ingester/test_workers.py @@ -146,7 +146,10 @@ async def test_successful_delete_prunes_dead_jobs_for_same_uri(client, jobs, syn @pytest.mark.asyncio -async def test_permanent_error_marks_dead_no_reschedule(client, jobs, sync): +async def test_permanent_error_without_revision_writes_no_marker(client, jobs, sync): + """A permanent failure on a revision-less job (e.g. HTTP without ETag) writes + no suppression marker — get_revision_snapshot omits revision-less rows, so it + would re-enqueue on the next sweep. Documents the revision-less caveat.""" client.create_document_from_source.side_effect = PermanentError("unsupported") job = await jobs.enqueue("src", "https://x/y.bin", JobOp.UPSERT) assert job is not None @@ -158,7 +161,52 @@ async def test_permanent_error_marks_dead_no_reschedule(client, jobs, sync): assert refreshed is not None assert refreshed.status is JobStatus.DEAD assert refreshed.last_error == "unsupported" - # sync_state is NOT written on failure + assert await sync.get_revision_snapshot("src") == {} + + +@pytest.mark.asyncio +async def test_permanent_error_with_revision_records_marker(client, jobs, sync): + """A permanent failure on a revisioned job records the failed revision in + sync_state (ingested=False) so discovery sees it as UNCHANGED and stops + re-enqueuing it every sweep, until the file's revision changes.""" + client.create_document_from_source.side_effect = PermanentError("encrypted") + job = await jobs.enqueue("src", "file:///x/y.pdf", JobOp.UPSERT, revision="r0") + assert job is not None + + pool = _pool(client, jobs, sync) + await pool.drain_once() + + refreshed = await jobs.get_job(job.id) + assert refreshed is not None + assert refreshed.status is JobStatus.DEAD + + # Failed revision is recorded, so discovery treats the unchanged file as known. + assert await sync.get_revision_snapshot("src") == {"file:///x/y.pdf": "r0"} + # Recorded as a failure, not an ingestion. + row = await sync.get_row("src", "file:///x/y.pdf") + assert row is not None + assert row.revision == "r0" + assert row.last_ingested_at is None + + +@pytest.mark.asyncio +async def test_transient_exhausted_writes_no_marker(client, jobs, sync): + """A transient failure that exhausts max_attempts goes dead but records no + suppression marker, so it stays re-attemptable on the next sweep (transient = + keep retrying once the service recovers).""" + client.create_document_from_source.side_effect = TransientError("blip") + job = await jobs.enqueue( + "src", "file:///x/y.pdf", JobOp.UPSERT, revision="r0", max_attempts=1 + ) + assert job is not None + + pool = _pool(client, jobs, sync) + await pool.drain_once() + + refreshed = await jobs.get_job(job.id) + assert refreshed is not None + assert refreshed.status is JobStatus.DEAD + # No marker despite a revision being present — only PermanentError suppresses. assert await sync.get_revision_snapshot("src") == {} @@ -386,6 +434,36 @@ async def test_worker_loses_claim_to_reaper_does_not_write_sync_state( assert await sync.get_revision_snapshot("src") == {} +@pytest.mark.asyncio +async def test_permanent_error_loses_claim_to_reaper_writes_no_marker( + client, jobs, sync +): + """If the reaper resets the claim and another worker re-claims before a + permanent failure is recorded, the original worker's mark_dead is a no-op + and it writes no failure marker — the re-claiming worker drives the + outcome, so the stale worker must not stamp sync_state.""" + client.create_document_from_source.side_effect = PermanentError("encrypted") + job = await jobs.enqueue("src", "u", JobOp.UPSERT, revision="r0") + assert job is not None + claimed_by_a = await jobs.claim_next("worker-A") + assert claimed_by_a is not None + # Reaper resets A's claim, worker-B re-claims. + await jobs.reap_stale(claim_timeout_seconds=0) + await jobs.claim_next("worker-B") + + pool = _pool(client, jobs, sync, worker_count=1) + # Drive A's _process with A's (now stale) Job snapshot. + await pool._process(claimed_by_a) + + refreshed = await jobs.get_job(job.id) + assert refreshed is not None + # B still owns the claim — A's mark_dead was a no-op. + assert refreshed.status is JobStatus.CLAIMED + assert refreshed.claimed_by == "worker-B" + # No failure marker written despite a revision being present. + assert await sync.get_revision_snapshot("src") == {} + + @pytest.mark.asyncio async def test_cancel_cleanup_survives_second_cancel(client, jobs, sync, monkeypatch): """A second cancel arriving while the cancel-handler is awaiting @@ -689,6 +767,31 @@ async def test_sync_state_write_failure_does_not_crash_worker( assert len(listed) == 1 +@pytest.mark.asyncio +async def test_permanent_failure_marker_write_failure_does_not_crash_worker( + client, jobs, sync, monkeypatch +): + """If the permanent-failure sync_state marker write fails after mark_dead, + the worker logs and continues rather than crashing. The job is already dead; + a missing marker just means the file may re-enqueue on the next sweep.""" + client.create_document_from_source.side_effect = PermanentError("encrypted") + await jobs.enqueue("src", "file:///x/y.pdf", JobOp.UPSERT, revision="r0") + + async def _failing_upsert(*args, **kwargs): + raise OSError("disk full") + + monkeypatch.setattr(sync, "upsert", _failing_upsert) + + pool = _pool(client, jobs, sync) + # drain_once should complete without raising despite the marker write failing. + processed = await pool.drain_once() + assert processed == 1 + + # Job is still dead — the marker write failure must not undo that. + listed = await jobs.list_jobs(status=JobStatus.DEAD) + assert len(listed) == 1 + + # --- reaper ---