Merge pull request #443 from ggozad/fix/queue-reenqueue

Stop re-enqueuing permanently-failed ingester documents

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
 

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
+        UNCHANGED. A stored revision means the file was accounted for at that
-        that didn't complete) are excluded — they have no revision to
+        revision — successfully ingested OR permanently failed; both suppress
-        compare against; the closing-loop DELETE diff uses list_known_uris
+        re-enqueue until the revision changes. Rows without a revision (HTTP
-        instead."""
+        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),

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)

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

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 ---