Auto-prune dead jobs when a sibling DELETE succeeds

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 ## [Unreleased]
 
+### Changed
+
+- A successful DELETE job auto-prunes dead jobs with the same `(source_id, uri)`. New `JobRepo.prune_dead(source_id, uri)`.
+
 ## [0.50.0] - 2026-05-27
 
 ### Added

haiku_rag_slim/haiku/rag/ingester/queue/repository.py

@@ -363,6 +363,21 @@ class JobRepo:
             await self._conn.commit()
         return row is not None
 
+    async def prune_dead(self, source_id: str, uri: str) -> int:
+        """Delete dead jobs for the given (source_id, uri). Called after a
+        successful DELETE to clear stale UPSERT failures for the same URI —
+        the document is gone, so a "couldn't ingest this" entry is no longer
+        actionable. Returns the number of rows removed."""
+        async with self._lock:
+            cursor = await self._conn.execute(
+                "DELETE FROM jobs WHERE source_id=? AND uri=? AND status='dead'",
+                (source_id, uri),
+            )
+            rowcount = cursor.rowcount or 0
+            await cursor.close()
+            await self._conn.commit()
+        return rowcount
+
     async def reap_stale(self, claim_timeout_seconds: int) -> int:
         """Reset claimed jobs whose claimed_at is older than the timeout
         back to `queued`. Decrements `attempts` to undo the increment from

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

@@ -235,6 +235,16 @@ class WorkerPool:
             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,
+                    job.uri,
+                )
         else:
             await self._sync.upsert(
                 job.source_id,

tests/ingester/test_queue.py

@@ -519,6 +519,54 @@ async def test_reap_stale_leaves_fresh_claims_alone(jobs):
     assert refreshed.status is JobStatus.CLAIMED
 
 
+# --- prune_dead ---
+
+
+@pytest.mark.asyncio
+async def test_prune_dead_removes_matching_dead_rows(jobs):
+    """A dead UPSERT becomes stale once a sibling DELETE has resolved the URI;
+    prune_dead() removes it so the DLQ stops showing resolved entries."""
+    job = await jobs.enqueue("s", "u", JobOp.UPSERT)
+    assert job is not None
+    claimed = await jobs.claim_next("w")
+    assert claimed is not None
+    await jobs.mark_dead(claimed.id, "boom", "w")
+
+    pruned = await jobs.prune_dead("s", "u")
+    assert pruned == 1
+    assert await jobs.get_job(job.id) is None
+
+
+@pytest.mark.asyncio
+async def test_prune_dead_leaves_non_dead_rows_alone(jobs):
+    """Queued/claimed/succeeded rows for the same (source, uri) are not
+    touched — only `dead` is purged."""
+    queued = await jobs.enqueue("s", "u", JobOp.UPSERT)
+    assert queued is not None
+
+    pruned = await jobs.prune_dead("s", "u")
+    assert pruned == 0
+    refreshed = await jobs.get_job(queued.id)
+    assert refreshed is not None and refreshed.status is JobStatus.QUEUED
+
+
+@pytest.mark.asyncio
+async def test_prune_dead_scoped_to_matching_uri(jobs):
+    """Dead rows for other URIs (and other sources) survive."""
+    j1 = await jobs.enqueue("s", "u1", JobOp.UPSERT)
+    assert j1 is not None
+    await jobs.mark_dead((await jobs.claim_next("w")).id, "err", "w")
+
+    j2 = await jobs.enqueue("s", "u2", JobOp.UPSERT)
+    assert j2 is not None
+    await jobs.mark_dead((await jobs.claim_next("w")).id, "err", "w")
+
+    pruned = await jobs.prune_dead("s", "u1")
+    assert pruned == 1
+    assert await jobs.get_job(j1.id) is None
+    assert await jobs.get_job(j2.id) is not None
+
+
 # --- release_if_claimed ---
 
 

tests/ingester/test_workers.py

@@ -105,6 +105,34 @@ async def test_drain_delete_op_removes_sync_state(client, jobs, sync):
     assert snapshot == {}
 
 
+@pytest.mark.asyncio
+async def test_successful_delete_prunes_dead_jobs_for_same_uri(client, jobs, sync):
+    """Once a DELETE resolves a URI, any earlier UPSERT failure for the same
+    (source_id, uri) is stale — auto-prune keeps the DLQ free of resolved
+    entries."""
+    # Stage a prior dead UPSERT (file-not-found style).
+    upsert = await jobs.enqueue("src", "file:///gone.md", JobOp.UPSERT)
+    assert upsert is not None
+    claimed = await jobs.claim_next("prev-worker")
+    assert claimed is not None
+    await jobs.mark_dead(claimed.id, "File does not exist", "prev-worker")
+    assert (await jobs.get_job(upsert.id)).status is JobStatus.DEAD
+
+    # Now run a DELETE for the same URI.
+    client.get_document_by_uri.return_value = Document(
+        id="doc-9", content="", uri="file:///gone.md"
+    )
+    delete = await jobs.enqueue("src", "file:///gone.md", JobOp.DELETE)
+    assert delete is not None
+
+    pool = _pool(client, jobs, sync)
+    await pool.drain_once()
+
+    assert (await jobs.get_job(delete.id)).status is JobStatus.SUCCEEDED
+    # The stale dead UPSERT for the same URI is gone.
+    assert await jobs.get_job(upsert.id) is None
+
+
 @pytest.mark.asyncio
 async def test_permanent_error_marks_dead_no_reschedule(client, jobs, sync):
     client.create_document_from_source.side_effect = PermanentError("unsupported")