Merge pull request #487 from ggozad/fix/duplicate-ingestion

Prevent duplicate documents from concurrent same-URI ingestion

CHANGELOG.md

@@ -6,6 +6,10 @@
 - `update_document` accepts a `uri` argument to change a document's URI.
 - docling-serve requests fail over to another instance on transport/5xx errors and skip instances whose circuit breaker is open; tune via `providers.docling_serve.max_attempts` and `providers.docling_serve.circuit_breaker`.
 
+### Fixed
+
+- Concurrent ingestion of the same URI no longer creates duplicate documents; the URI is re-checked under the write lock and a colliding create becomes an update.
+
 ## [0.63.2] - 2026-07-03
 
 ### Changed

haiku_rag_slim/haiku/rag/client/documents.py

@@ -123,26 +123,48 @@ async def _store_document_with_chunks(
     async with client.store._write_lock:
         versions = await client.store.current_table_versions()
 
-        created_doc = await client.document_repository.create(document)
+        # A concurrent ingestion of the same URI may have created the document
+        # while this one was converting/embedding outside the lock. LanceDB has
+        # no unique constraint on `uri`, so re-check under the lock and update in
+        # place rather than inserting a duplicate.
+        existing = (
+            await client.get_document_by_uri(document.uri)
+            if document.uri is not None
+            else None
+        )
 
         try:
-            assert created_doc.id is not None, (
+            if existing is not None:
-                "Document ID should not be None after creation"
+                document.id = existing.id
+                document.created_at = existing.created_at
+                stored_doc = await client.document_repository.update(document)
+            else:
+                stored_doc = await client.document_repository.create(document)
+
+            assert stored_doc.id is not None, (
+                "Document ID should not be None after storing"
             )
             for order, chunk in enumerate(chunks):
-                chunk.document_id = created_doc.id
+                chunk.document_id = stored_doc.id
                 chunk.order = order
-
-            await client.chunk_repository.create(chunks)
-
             for item in items:
-                item.document_id = created_doc.id
+                item.document_id = stored_doc.id
-            await client.document_item_repository.create_items(created_doc.id, items)
+
+            if existing is not None:
+                await client.chunk_repository.replace_for_document(
+                    stored_doc.id, chunks
+                )
+                await client.document_item_repository.replace_for_document(
+                    stored_doc.id, items
+                )
+            else:
+                await client.chunk_repository.create(chunks)
+                await client.document_item_repository.create_items(stored_doc.id, items)
 
             if client._config.storage.auto_vacuum:
                 client._schedule_vacuum()
 
-            return created_doc
+            return stored_doc
         except Exception:
             await client.store.restore_table_versions(versions)
             raise

tests/test_duplicate_ingestion.py

@@ -0,0 +1,42 @@
+import asyncio
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from haiku.rag.client import HaikuRAG
+from haiku.rag.store.models.document import Document
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.mark.vcr()
+async def test_concurrent_same_uri_ingestion_creates_single_document(temp_db_path):
+    """Two concurrent ingestions of the same URI must not create duplicates.
+
+    Both calls read `existing_doc=None` before either acquires the write lock;
+    the atomic re-check under the lock turns the loser into an update instead of
+    a second insert (LanceDB has no unique constraint on `uri`).
+    """
+    async with HaikuRAG(temp_db_path, create=True) as client:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            temp_path = Path(temp_dir) / "dup.txt"
+            temp_path.write_text("Duplicate ingestion regression content.")
+
+            results = await asyncio.gather(
+                client.create_document_from_source(temp_path),
+                client.create_document_from_source(temp_path),
+            )
+
+            for doc in results:
+                assert isinstance(doc, Document)
+            # Both concurrent calls resolve to one document, not two.
+            assert results[0].id == results[1].id
+            assert await client.count_documents() == 1
+
+            # The surviving document owns exactly one ingestion's chunks; the
+            # loser's create was collapsed into an update, leaving no orphans.
+            surviving = await client.get_document_by_uri(temp_path.as_uri())
+            assert surviving is not None
+            chunks = await client.chunk_repository.get_by_document_id(surviving.id)
+            assert len(chunks) >= 1