From 9681a63757f340084674c5d926f67b4ebed07614 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Tue, 14 Jul 2026 15:19:03 +0200 Subject: [PATCH] fix: upsert document vectors before pruning stale chunks DocumentSyncService._index_vectors ran delete_by_filter on the document's existing chunks FIRST and only then embedded; if the embedding pass failed (Ollama down) the Paperless document was left with zero vectors until the next successful sync - the same zero-vector hazard already fixed for wiki pages in VectorService.update_from_page. Chunk ids are now deterministic uuid5 (document_{id}_chunk_{i}) so re-upserting overwrites in place; new points are upserted first, then stale points (including legacy random-uuid4 ones) are pruned via scroll + delete_by_ids, and only after a successful upsert. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L --- CHANGELOG.md | 1 + src/services/document_sync_service.py | 45 ++++++++++++++++-------- tests/test_document_sync.py | 50 ++++++++++++++++++++++++--- 3 files changed, 77 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3660903..c7b25a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed (Phase D review batch) - **Runtime prefetch registration is executable** - `SchedulerClient.register_volatile_fetch` (used by consolidation's `_register_prefetch`) registered tasks that were dead on arrival three ways: the JSON body sat under the ignored `body` key (`rest_api_executor` only reads `config["payload"]`), there was no `auth` block (the scheduled POST would 401 against library-desk's `verify_api_key`), and `user` was in the body while every `/volatile/fetch` endpoint requires it as a QUERY parameter (would 422). The config now puts `user` in the URL query string (URL-encoded), an empty `payload`, and `auth: {type: bearer, token: "${LIBRARY_API_KEY}"}` (substituted from the Scheduler's environment; never stored raw). `SchedulerClient` itself also sent no Authorization to the Scheduler API, so registration failed silently at consolidation time — it now sends `Authorization: Bearer ` (new `scheduler_api_key` setting; a client without a key logs a warning). +- **Document sync uses delete-last reindex order** - `DocumentSyncService._index_vectors` deleted the document's existing chunks (`delete_by_filter`) BEFORE embedding, so a failed embedding pass (e.g. Ollama down) left the Paperless document with zero vectors until the next successful sync — the exact hazard already fixed for wiki pages in `VectorService.update_from_page`. Chunk point ids are now deterministic (uuid5 of `document_{id}_chunk_{i}`, replacing random uuid4) so re-upserting overwrites in place; new points are upserted first and stale points (including legacy uuid4 ones) are pruned afterwards, only after a successful upsert. - **Registrar authenticates to the Scheduler API** - `scripts/register_scheduler_tasks.py --execute` sent no `Authorization` header while the Scheduler's task-management endpoints are Bearer-guarded (`verify_api_key`: 401 on missing key), so the existence probes 401'd (misread as "task absent") and every registration failed; only the public `/health` gate passed. `--execute` now requires `SCHEDULER_API_KEY` in the environment (refuses to run without it, key is never stored) and sends `Authorization: Bearer $SCHEDULER_API_KEY` on all of its own HTTP calls. Deploy note updated alongside the existing `LIBRARY_API_KEY` requirement. ### Fixed (hazards batch) diff --git a/src/services/document_sync_service.py b/src/services/document_sync_service.py index 050d8be..029a747 100644 --- a/src/services/document_sync_service.py +++ b/src/services/document_sync_service.py @@ -196,18 +196,6 @@ class DocumentSyncService: collection = get_qdrant_collection_name(user) await self.qdrant.ensure_collection(collection) - # Delete existing chunks for this document (via the async wrapper) - try: - await self.qdrant.delete_by_filter( - collection_name=collection, - filter_conditions={ - "doc_type": "document", - "paperless_id": document_id, - } - ) - except Exception as e: - logger.debug(f"No existing chunks to delete: {e}") - # Chunk content chunks = self._chunk_text(content) if not chunks: @@ -229,7 +217,11 @@ class DocumentSyncService: ) continue - point_id = str(uuid.uuid4()) + # Deterministic id: re-upserting the same document overwrites + # its previous chunks in place (enables delete-last below). + point_id = str( + uuid.uuid5(uuid.NAMESPACE_DNS, f"document_{document_id}_chunk_{i}") + ) content_hash = hashlib.md5(chunk.encode()).hexdigest() points.append({ @@ -256,13 +248,38 @@ class DocumentSyncService: f"(embedding failures); indexing the remaining {len(points)}" ) - # Upsert to Qdrant in one batch via the async wrapper + # Upsert BEFORE pruning stale chunks (same order as the wiki + # reindex fix in VectorService.update_from_page): the old + # delete-first order left the document with ZERO vectors until the + # next successful sync whenever the embedding pass failed after the + # delete (e.g. Ollama down). Deterministic uuid5 ids make the + # in-place overwrite safe. if points: await self.qdrant.upsert_points( collection_name=collection, points=points ) + # Prune chunks left over from a previous version of the document + # (indexes beyond the new count, or legacy random-uuid4 points). + # Only prune after a successful upsert - a fully failed embedding + # pass must not wipe the old vectors. + new_ids = {p["id"] for p in points} + existing = await self.qdrant.scroll_all_points( + collection_name=collection, + filter_conditions={ + "doc_type": "document", + "paperless_id": document_id, + }, + with_payload=False, + ) + stale_ids = [pt["id"] for pt in existing if pt["id"] not in new_ids] + if stale_ids: + await self.qdrant.delete_by_ids( + collection_name=collection, + point_ids=stale_ids, + ) + return len(points) async def _index_graph( diff --git a/tests/test_document_sync.py b/tests/test_document_sync.py index 155f549..b842cf0 100644 --- a/tests/test_document_sync.py +++ b/tests/test_document_sync.py @@ -6,10 +6,13 @@ Pins the Phase D fixes: 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) +- delete-LAST reindex order (same as the wiki fix): new points are + upserted with deterministic uuid5 ids BEFORE stale points are pruned, + so a failed embedding pass can no longer leave a document with zero + vectors (the old order ran delete_by_filter first) """ +import uuid from unittest.mock import AsyncMock, MagicMock import pytest @@ -24,8 +27,9 @@ TENANT_COLLECTION = "library_desk_llm_tester" 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)) + qdrant.scroll_all_points = AsyncMock(return_value=[]) + qdrant.delete_by_ids = AsyncMock(side_effect=lambda collection_name, point_ids: len(point_ids)) return qdrant @@ -75,14 +79,46 @@ class TestIndexVectors: 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): + async def test_point_ids_are_deterministic(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( + points = mock_qdrant.upsert_points.await_args.kwargs["points"] + expected = str(uuid.uuid5(uuid.NAMESPACE_DNS, "document_7_chunk_0")) + assert points[0]["id"] == expected + + async def test_stale_chunks_pruned_after_upsert(self, sync_service, mock_qdrant): + """Old points not in the new set are deleted AFTER the new upsert.""" + call_order = [] + mock_qdrant.upsert_points = AsyncMock( + side_effect=lambda collection_name, points: ( + call_order.append("upsert"), len(points))[1] + ) + stale_id = str(uuid.uuid4()) # legacy random-uuid4 point + kept_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, "document_7_chunk_0")) + mock_qdrant.scroll_all_points = AsyncMock( + return_value=[{"id": stale_id, "payload": {}}, + {"id": kept_id, "payload": {}}] + ) + mock_qdrant.delete_by_ids = AsyncMock( + side_effect=lambda collection_name, point_ids: ( + call_order.append("delete"), len(point_ids))[1] + ) + + await sync_service.index_document( + document_id=7, user=TENANT, content="hello world", title="Doc" + ) + + assert call_order == ["upsert", "delete"] + mock_qdrant.delete_by_ids.assert_awaited_once_with( + collection_name=TENANT_COLLECTION, + point_ids=[stale_id], + ) + mock_qdrant.scroll_all_points.assert_awaited_once_with( collection_name=TENANT_COLLECTION, filter_conditions={"doc_type": "document", "paperless_id": 7}, + with_payload=False, ) async def test_upsert_routed_through_wrapper(self, sync_service, mock_qdrant): @@ -134,3 +170,7 @@ class TestIndexVectors: assert result.success is False assert "embed" in (result.error or "").lower() mock_qdrant.upsert_points.assert_not_awaited() + # Delete-last: a fully failed embedding pass must leave the old + # vectors untouched (previously delete_by_filter ran first, leaving + # the document with zero vectors until the next successful sync) + mock_qdrant.delete_by_ids.assert_not_awaited()