From 922d1d567da7034eb5a4bfa2422532d35be21565 Mon Sep 17 00:00:00 2001
From: Yiorgis Gozadinos <ggozadinos@gmail.com>
Date: Wed, 27 May 2026 13:30:46 +0300
Subject: [PATCH] Prevent DELETE/UPSERT race for the same URI

---
 .../haiku/rag/ingester/queue/schema.py        | 11 +++++---
 tests/ingester/test_queue.py                  | 25 ++++++++++++++++++-
 2 files changed, 31 insertions(+), 5 deletions(-)

diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/schema.py b/haiku_rag_slim/haiku/rag/ingester/queue/schema.py
index 6d818831..04bb67e2 100644
--- a/haiku_rag_slim/haiku/rag/ingester/queue/schema.py
+++ b/haiku_rag_slim/haiku/rag/ingester/queue/schema.py
@@ -22,12 +22,15 @@ CREATE TABLE IF NOT EXISTS jobs (
 )
 """
 
-# Partial unique index: a (source_id, uri, op) triple can only have one live
-# job (queued or claimed) at a time. Once succeeded or dead, the row no
-# longer satisfies the WHERE clause and a re-enqueue is allowed.
+# Partial unique index: a (source_id, uri) pair can only have one live job
+# (queued or claimed) at a time, regardless of op. Live UPSERT and DELETE
+# for the same URI can't both exist — preventing a DELETE worker from
+# removing a document a sibling UPSERT just ingested. Once succeeded or
+# dead, the row no longer satisfies the WHERE clause and a re-enqueue is
+# allowed.
 JOBS_LIVE_INDEX = """
 CREATE UNIQUE INDEX IF NOT EXISTS uq_jobs_live
-ON jobs(source_id, uri, op)
+ON jobs(source_id, uri)
 WHERE status IN ('queued', 'claimed')
 """
 
diff --git a/tests/ingester/test_queue.py b/tests/ingester/test_queue.py
index e7b50b9e..77380b44 100644
--- a/tests/ingester/test_queue.py
+++ b/tests/ingester/test_queue.py
@@ -175,11 +175,34 @@ async def test_has_pending_is_per_source(jobs):
 
 
 @pytest.mark.asyncio
-async def test_enqueue_different_ops_coexist(jobs):
+async def test_enqueue_drops_delete_when_upsert_is_live(jobs):
+    """Stops a DELETE worker from removing a document a sibling UPSERT
+    just ingested."""
     upsert = await jobs.enqueue("s", "u", JobOp.UPSERT)
     delete = await jobs.enqueue("s", "u", JobOp.DELETE)
     assert upsert is not None
+    assert delete is None
+
+
+@pytest.mark.asyncio
+async def test_enqueue_drops_upsert_when_delete_is_live(jobs):
+    """Symmetric: live DELETE for a URI blocks a fresh UPSERT for the
+    same URI. The next sweep after DELETE completes will re-emit the
+    UPSERT if the file is back."""
+    delete = await jobs.enqueue("s", "u", JobOp.DELETE)
+    upsert = await jobs.enqueue("s", "u", JobOp.UPSERT)
     assert delete is not None
+    assert upsert is None
+
+
+@pytest.mark.asyncio
+async def test_enqueue_different_uris_independent_of_op(jobs):
+    """Uniqueness is per-(source_id, uri), not per-(source_id, uri, op).
+    Different URIs can be queued regardless of which op each one is."""
+    a = await jobs.enqueue("s", "u-a", JobOp.UPSERT)
+    b = await jobs.enqueue("s", "u-b", JobOp.DELETE)
+    assert a is not None
+    assert b is not None
 
 
 # --- claim_next ---