From 5400085147fbd69c12814732562b30f738d41761 Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Wed, 27 May 2026 17:17:01 +0300 Subject: [PATCH] Auto-prune dead jobs when a sibling DELETE succeeds --- CHANGELOG.md | 4 ++ .../haiku/rag/ingester/queue/repository.py | 15 ++++++ .../haiku/rag/ingester/workers/pool.py | 10 ++++ tests/ingester/test_queue.py | 48 +++++++++++++++++++ tests/ingester/test_workers.py | 28 +++++++++++ 5 files changed, 105 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 521d8283..7040da22 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py index c8934cd5..1f1b2b31 100644 --- a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py +++ b/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 diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py index 4f204a27..a14afe96 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py +++ b/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, diff --git a/tests/ingester/test_queue.py b/tests/ingester/test_queue.py index 7404fa1f..4ca3ae6c 100644 --- a/tests/ingester/test_queue.py +++ b/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 --- diff --git a/tests/ingester/test_workers.py b/tests/ingester/test_workers.py index 1d32b35b..0df4cdca 100644 --- a/tests/ingester/test_workers.py +++ b/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")