diff --git a/CHANGELOG.md b/CHANGELOG.md index 4aa3530..805bfe9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed (performance) +- **Document sync indexing fixed** - `DocumentSyncService._index_vectors` now awaits `ensure_collection` (the coroutine was created but never ran, so fresh tenants had no collection at upsert time), filters out `None` entries from `embed_batch` so one failed chunk embedding no longer aborts the whole document upsert (all-failed still reports failure), and routes the raw `client.delete`/`client.upsert` calls through the async wrapper (`delete_by_filter` / new batch `upsert_points`). Offline unit tests added. - **Async Qdrant client** - `QdrantClientWrapper` now uses `AsyncQdrantClient` with an explicit timeout (`QDRANT_TIMEOUT`, default 30s). Every vector call previously ran on the synchronous client inside async wrapper methods, blocking the FastAPI event loop for the duration of each Qdrant round-trip. The wrapper API is unchanged (all methods were already `async`), so call sites only gained real awaits. The HybridRAG document leg was moved off the deprecated raw `client.search` onto the wrapper's `search_vectors` (fixing a latent `AttributeError`: it called the nonexistent `ollama.embed_text`, so the leg always reported `failed`), and the health check awaits `get_collections`. ### Changed diff --git a/src/clients/qdrant_client.py b/src/clients/qdrant_client.py index 608cc15..e17e579 100644 --- a/src/clients/qdrant_client.py +++ b/src/clients/qdrant_client.py @@ -467,6 +467,43 @@ class QdrantClientWrapper: logger.error(f"Failed to upsert vector: {e}", exc_info=True) return False + async def upsert_points( + self, + collection_name: str, + points: List[Dict[str, Any]] + ) -> int: + """ + Upsert a batch of vector points in a single request. + + Args: + collection_name: Collection name + points: List of {"id": str, "vector": List[float], "payload": dict} + + Returns: + Number of points upserted + + Raises: + Exception: If the upsert fails (callers decide how to degrade) + """ + if not points: + return 0 + + structs = [ + PointStruct( + id=p["id"], + vector=p["vector"], + payload=p.get("payload", {}) + ) + for p in points + ] + + await self.client.upsert( + collection_name=collection_name, + points=structs + ) + logger.info(f"Upserted {len(structs)} points into {collection_name}") + return len(structs) + async def delete_by_filter( self, collection_name: str, diff --git a/src/services/document_sync_service.py b/src/services/document_sync_service.py index 870b332..3780cfd 100644 --- a/src/services/document_sync_service.py +++ b/src/services/document_sync_service.py @@ -194,19 +194,15 @@ class DocumentSyncService: ) -> int: """Create vector embeddings for document content.""" collection = get_qdrant_collection_name(user) - self.qdrant.ensure_collection(collection) + await self.qdrant.ensure_collection(collection) - # Delete existing chunks for this document + # Delete existing chunks for this document (via the async wrapper) try: - await self.qdrant.client.delete( + await self.qdrant.delete_by_filter( collection_name=collection, - points_selector={ - "filter": { - "must": [ - {"key": "doc_type", "match": {"value": "document"}}, - {"key": "paperless_id", "match": {"value": document_id}}, - ] - } + filter_conditions={ + "doc_type": "document", + "paperless_id": document_id, } ) except Exception as e: @@ -217,12 +213,22 @@ class DocumentSyncService: if not chunks: return 0 - # Generate embeddings + # Generate embeddings (embed_batch returns None for failed chunks) embeddings = await self.ollama.embed_batch(chunks) - # Build points + # Build points, skipping chunks whose embedding failed. Previously a + # single None embedding poisoned the batch and aborted the whole + # document upsert. points = [] + skipped = 0 for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): + if embedding is None: + skipped += 1 + logger.warning( + f"Skipping chunk {i} of document {document_id}: embedding failed" + ) + continue + point_id = str(uuid.uuid4()) content_hash = hashlib.md5(chunk.encode()).hexdigest() @@ -240,9 +246,19 @@ class DocumentSyncService: } }) - # Upsert to Qdrant + if skipped and not points: + raise RuntimeError( + f"All {skipped} chunk embeddings failed for document {document_id}" + ) + if skipped: + logger.warning( + f"Document {document_id}: {skipped}/{len(chunks)} chunks skipped " + f"(embedding failures); indexing the remaining {len(points)}" + ) + + # Upsert to Qdrant in one batch via the async wrapper if points: - await self.qdrant.client.upsert( + await self.qdrant.upsert_points( collection_name=collection, points=points ) diff --git a/tests/test_document_sync.py b/tests/test_document_sync.py new file mode 100644 index 0000000..f056445 --- /dev/null +++ b/tests/test_document_sync.py @@ -0,0 +1,135 @@ +""" +Offline unit tests for DocumentSyncService vector indexing fixes. + +Pins the Phase D fixes: +- ensure_collection is actually awaited (it used to be a bare coroutine + that never ran, so fresh tenants had no collection at upsert time) +- None entries from embed_batch are filtered out instead of poisoning + the whole batch upsert (one failed chunk aborted the document) +- raw client.delete/client.upsert calls are routed through the async + wrapper methods (delete_by_filter / upsert_points) +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from src.services.document_sync_service import DocumentSyncService + +TENANT = "llm_tester" +TENANT_COLLECTION = "library_desk_llm_tester" + + +@pytest.fixture +def mock_qdrant(): + qdrant = MagicMock() + qdrant.ensure_collection = AsyncMock() + qdrant.delete_by_filter = AsyncMock(return_value=0) + qdrant.upsert_points = AsyncMock(side_effect=lambda collection_name, points: len(points)) + return qdrant + + +@pytest.fixture +def mock_ollama(): + ollama = MagicMock() + ollama.embed_batch = AsyncMock( + side_effect=lambda texts: [[0.1] * 768 for _ in texts] + ) + return ollama + + +@pytest.fixture +def mock_neo4j(): + neo4j = MagicMock() + neo4j.execute_query = AsyncMock(return_value=[]) + return neo4j + + +@pytest.fixture +def mock_paperless(): + paperless = MagicMock() + paperless.get_custom_field_by_name = AsyncMock(return_value=None) + return paperless + + +@pytest.fixture +def sync_service(mock_paperless, mock_qdrant, mock_ollama, mock_neo4j): + return DocumentSyncService( + paperless_client=mock_paperless, + qdrant_client=mock_qdrant, + ollama_client=mock_ollama, + neo4j_client=mock_neo4j, + wiki_client=MagicMock(), + settings=MagicMock(), + ) + + +@pytest.mark.unit +class TestIndexVectors: + async def test_ensure_collection_is_awaited(self, sync_service, mock_qdrant): + result = await sync_service.index_document( + document_id=1, user=TENANT, content="hello world", title="Doc" + ) + + assert result.success is True + mock_qdrant.ensure_collection.assert_awaited_once_with(TENANT_COLLECTION) + + async def test_delete_routed_through_wrapper(self, sync_service, mock_qdrant): + await sync_service.index_document( + document_id=7, user=TENANT, content="hello world", title="Doc" + ) + + mock_qdrant.delete_by_filter.assert_awaited_once_with( + collection_name=TENANT_COLLECTION, + filter_conditions={"doc_type": "document", "paperless_id": 7}, + ) + + async def test_upsert_routed_through_wrapper(self, sync_service, mock_qdrant): + result = await sync_service.index_document( + document_id=1, user=TENANT, content="hello world", title="Doc" + ) + + assert result.chunks_created == 1 + kwargs = mock_qdrant.upsert_points.await_args.kwargs + assert kwargs["collection_name"] == TENANT_COLLECTION + payload = kwargs["points"][0]["payload"] + assert payload["doc_type"] == "document" + assert payload["paperless_id"] == 1 + + async def test_failed_chunk_embedding_is_skipped_not_fatal( + self, sync_service, mock_qdrant, mock_ollama + ): + # Three chunks; the middle embedding fails + long_content = " ".join(f"word{i}" for i in range(1200)) + mock_ollama.embed_batch = AsyncMock( + side_effect=lambda texts: [ + [0.1] * 768 if i != 1 else None for i in range(len(texts)) + ] + ) + + result = await sync_service.index_document( + document_id=2, user=TENANT, content=long_content, title="Doc" + ) + + assert result.success is True + points = mock_qdrant.upsert_points.await_args.kwargs["points"] + assert result.chunks_created == len(points) + # The failed chunk (index 1) is absent, the others kept their index + indices = [p["payload"]["chunk_index"] for p in points] + assert 1 not in indices + assert len(indices) >= 2 + + async def test_all_embeddings_failed_reports_failure( + self, sync_service, mock_qdrant, mock_ollama + ): + mock_ollama.embed_batch = AsyncMock( + side_effect=lambda texts: [None for _ in texts] + ) + + result = await sync_service.index_document( + document_id=3, user=TENANT, content="hello world", title="Doc" + ) + + assert result.success is False + assert "embed" in (result.error or "").lower() + mock_qdrant.upsert_points.assert_not_awaited()