diff --git a/CHANGELOG.md b/CHANGELOG.md index fd39560..083af5f 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) +- **Search persistence off the hot path, one atomic transaction** - HybridRAG Phase 6 (`_persist_search_for_librarian`) no longer gates the `/query/hybrid` response: the `search_id` is generated up front and returned immediately while the Neo4j write runs as a background task (strong task references held so tasks are not GC'd mid-flight). The write itself collapsed from ~21+ sequential auto-commit queries (SearchQuery node + per-document FOUND links + per-web-result WebResult nodes) into ONE UNWIND-based `execute_write` transaction, so a mid-way failure can no longer leave a partial SearchQuery graph behind. The persisted shape (SearchQuery properties incl. `processed: false`, tenant labels, `FOUND` relationship properties, WebResult properties) is unchanged and pinned by `tests/test_search_persistence.py` against exactly what the consolidation service queries. `timing.persistence_ms` now reports 0 (no longer on the request path). - **Batched embeddings + delete-last reindex** - `OllamaClient.embed_batch` now sends ONE batched `/api/embed` request (verified against the live Ollama; the old "batch" looped one `/api/embeddings` call per chunk) with a per-text fallback preserving partial-success semantics. `VectorService.update_from_page` embeds all chunks in that single call and upserts them in one Qdrant batch, and the reindex order is reversed: new points are upserted BEFORE stale points are pruned (deterministic uuid5 chunk ids make the overwrite safe), so a mid-way failure can no longer leave a page with zero vectors — the old order deleted everything first. The summary now reports `status` (`success`/`partial`/`failed`) and `chunks_skipped` instead of unconditional `success=True`; a fully failed embedding pass keeps the old vectors and reports failure. Measured on a real 7-chunk page ingest as `llm_tester` against the local server: ~375ms → ~181ms median (3 runs each). - **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`. diff --git a/src/services/hybrid_rag_service.py b/src/services/hybrid_rag_service.py index 75a0683..6a575c1 100644 --- a/src/services/hybrid_rag_service.py +++ b/src/services/hybrid_rag_service.py @@ -83,6 +83,9 @@ class HybridRAGService: self.settings = settings self.volatile = volatile_service self.reranker_model = settings.ollama_llm_model + # Strong references to fire-and-forget persistence tasks so they are + # not garbage-collected mid-flight (see Phase 6 in search()). + self._background_tasks: set = set() async def search( self, @@ -182,17 +185,25 @@ class HybridRAGService: timing["total_ms"] = (time.time() - start_time) * 1000 - # Phase 6: Persistence (async, non-blocking) - phase6_start = time.time() - search_id = await self._persist_search_for_librarian( - query=query, - user=user, - keywords_data=keywords_data, - raw_results=raw_results, - final_results=final_results, - timing=timing + # Phase 6: Persistence — genuinely off the hot path. The search_id is + # generated up front and returned immediately; the Neo4j write runs as + # a background task (one atomic transaction, see + # _persist_search_for_librarian) instead of gating the response. + search_id = str(uuid.uuid4()) + timing["persistence_ms"] = 0.0 # not on the request path anymore + persist_task = asyncio.create_task( + self._persist_search_for_librarian( + search_id=search_id, + query=query, + user=user, + keywords_data=keywords_data, + raw_results=raw_results, + final_results=final_results, + timing=timing + ) ) - timing["persistence_ms"] = (time.time() - phase6_start) * 1000 + self._background_tasks.add(persist_task) + persist_task.add_done_callback(self._background_tasks.discard) # Degradation signaling: a failed leg contributes no results, but the # response says so instead of silently pretending the leg was empty @@ -882,6 +893,7 @@ Ranking:""" async def _persist_search_for_librarian( self, + search_id: str, query: str, user: str, keywords_data: Dict[str, Any], @@ -892,10 +904,20 @@ Ranking:""" """ Phase 6: Store search query and results for Librarian processing. - Creates SearchQuery node in Neo4j with relationships to found documents - and web results for offline knowledge consolidation. + Creates the SearchQuery node, FOUND links to this tenant's Document + nodes, and WebResult nodes in ONE UNWIND-based write transaction + (previously ~21+ sequential auto-commit queries), so a mid-way + failure can never leave a partial SearchQuery graph behind. + + SHAPE CONTRACT: the consolidation service (consolidation_service.py) + consumes exactly this shape — SearchQuery {id, query, user, + timestamp, processed:false, total_results, web_count, keywords}, + (sq)-[f:FOUND {rank, rrf_score}]->(wr:WebResult {url, title, + content}) — do not change it without updating both sides + (pinned by tests/test_search_persistence.py). Args: + search_id: Pre-generated search ID (already returned to the caller) query: Search query user: User identifier keywords_data: Extracted keywords/synonyms @@ -904,15 +926,47 @@ Ranking:""" timing: Performance timing Returns: - Search ID for tracking + Search ID on success, None on failure """ try: user_base_label = get_neo4j_user_base_label(user) user_doc_label = get_neo4j_user_label(user) - search_id = str(uuid.uuid4()) - # Create SearchQuery node - create_query = f""" + # Links to found wiki documents (top 20). + # TENANT ISOLATION: matched against this tenant's Document label + # only — an unscoped (d:Document {page_id}) match would attach + # FOUND relationships to other tenants' documents that share the + # same Wiki.js page id. + doc_links = [] + for rank, result_data in enumerate(final_results[:20], start=1): + result = result_data.get("result", {}) + page_id = result.get("page_id") + if page_id: + doc_links.append({ + "page_id": page_id, + "source": result_data.get("source_type", "unknown"), + "rank": rank, + "rrf_score": result_data.get("rrf_score", 0), + "final_rank": result_data.get("final_rank", rank) + }) + + # Web results as WebResult nodes (top 10) + web_links = [] + web_results = [r for r in final_results[:10] if r.get("result", {}).get("url")] + for rank, result_data in enumerate(web_results, start=1): + result = result_data.get("result", {}) + web_links.append({ + "url": result.get("url"), + "title": result.get("title", ""), + "content": result.get("content", "")[:1000], # Truncate + "rank": rank, + "rrf_score": result_data.get("rrf_score", 0) + }) + + # Single atomic write: node + doc links + web results. The CALL + # subqueries aggregate so an empty UNWIND list cannot swallow the + # rest of the query. + persist_query = f""" CREATE (sq:{user_base_label}_SearchQuery:SearchQuery {{ id: $search_id, query: $query, @@ -927,10 +981,39 @@ Ranking:""" synonyms: $synonyms, timing_ms: $timing_ms }}) - RETURN sq.id as id + WITH sq + CALL {{ + WITH sq + UNWIND $doc_links AS link + MATCH (d:{user_doc_label}:Document {{page_id: link.page_id}}) + MERGE (sq)-[f:FOUND]->(d) + SET f.source = link.source, + f.rank = link.rank, + f.rrf_score = link.rrf_score, + f.final_rank = link.final_rank + RETURN count(*) AS docs_linked + }} + CALL {{ + WITH sq + UNWIND $web_links AS wl + CREATE (wr:{user_base_label}_WebResult:WebResult {{ + url: wl.url, + title: wl.title, + content: wl.content, + search_id: $search_id, + timestamp: datetime() + }}) + CREATE (sq)-[:FOUND {{ + source: "web", + rank: wl.rank, + rrf_score: wl.rrf_score + }}]->(wr) + RETURN count(*) AS web_created + }} + RETURN sq.id AS id, docs_linked, web_created """ - result = await self.graph.neo4j.execute_query(create_query, { + await self.graph.neo4j.execute_write(persist_query, { "search_id": search_id, "query": query, "user": user, @@ -940,67 +1023,11 @@ Ranking:""" "web_count": len(raw_results.get("web", [])), "keywords": keywords_data.get("core_keywords", []), "synonyms": json.dumps(keywords_data.get("synonyms", {})), - "timing_ms": timing.get("total_ms", 0) + "timing_ms": timing.get("total_ms", 0), + "doc_links": doc_links, + "web_links": web_links }) - # Link to found wiki documents (top 20) - for rank, result_data in enumerate(final_results[:20], start=1): - result = result_data.get("result", {}) - page_id = result.get("page_id") - - if page_id: - # TENANT ISOLATION: only link to this tenant's Document - # nodes. An unscoped (d:Document {page_id}) match would - # attach FOUND relationships to other tenants' documents - # that share the same Wiki.js page id. - link_doc_query = f""" - MATCH (sq:{user_base_label}_SearchQuery:SearchQuery {{id: $search_id}}) - MATCH (d:{user_doc_label}:Document {{page_id: $page_id}}) - MERGE (sq)-[f:FOUND]->(d) - SET f.source = $source, - f.rank = $rank, - f.rrf_score = $rrf_score, - f.final_rank = $final_rank - """ - - await self.graph.neo4j.execute_query(link_doc_query, { - "search_id": search_id, - "page_id": page_id, - "source": result_data.get("source_type", "unknown"), - "rank": rank, - "rrf_score": result_data.get("rrf_score", 0), - "final_rank": result_data.get("final_rank", rank) - }) - - # Store web results as WebResult nodes (top 10) - web_results = [r for r in final_results[:10] if r.get("result", {}).get("url")] - for rank, result_data in enumerate(web_results, start=1): - result = result_data.get("result", {}) - create_web_query = f""" - MATCH (sq:{user_base_label}_SearchQuery:SearchQuery {{id: $search_id}}) - CREATE (wr:{user_base_label}_WebResult:WebResult {{ - url: $url, - title: $title, - content: $content, - search_id: $search_id, - timestamp: datetime() - }}) - CREATE (sq)-[:FOUND {{ - source: "web", - rank: $rank, - rrf_score: $rrf_score - }}]->(wr) - """ - - await self.graph.neo4j.execute_query(create_web_query, { - "search_id": search_id, - "url": result.get("url"), - "title": result.get("title", ""), - "content": result.get("content", "")[:1000], # Truncate - "rank": rank, - "rrf_score": result_data.get("rrf_score", 0) - }) - logger.info(f"Persisted search {search_id} for Librarian processing") return search_id diff --git a/tests/test_hybrid_rag.py b/tests/test_hybrid_rag.py index d0ee5be..da46f54 100644 --- a/tests/test_hybrid_rag.py +++ b/tests/test_hybrid_rag.py @@ -506,7 +506,9 @@ class TestPhase6_Persistence: timing = {"total_ms": 1000} + import uuid as _uuid search_id = await hybrid_rag_service._persist_search_for_librarian( + search_id=str(_uuid.uuid4()), query="test query", user=TEST_USER, keywords_data=keywords_data, diff --git a/tests/test_hybrid_rag_degradation.py b/tests/test_hybrid_rag_degradation.py index 9de1c29..0dc2a62 100644 --- a/tests/test_hybrid_rag_degradation.py +++ b/tests/test_hybrid_rag_degradation.py @@ -49,6 +49,7 @@ def graph_service(): graph.search_documents = AsyncMock(return_value=[]) graph.get_related_documents = AsyncMock(return_value=[]) graph.neo4j.execute_query = AsyncMock(return_value=[{"id": "search-1"}]) + graph.neo4j.execute_write = AsyncMock(return_value=[{"id": "search-1"}]) return graph diff --git a/tests/test_search_persistence.py b/tests/test_search_persistence.py new file mode 100644 index 0000000..883c17f --- /dev/null +++ b/tests/test_search_persistence.py @@ -0,0 +1,225 @@ +""" +Offline tests for the Phase 6 persistence rewrite (single atomic write, +off the hot path) and its SHAPE CONTRACT with the consolidation service. + +The consolidation repair loop (consolidation_service.py) consumes the +persisted graph: + +- _find_unprocessed_searches: + MATCH (sq:SearchQuery {processed: false}) WHERE sq.timestamp > ... + RETURN sq.id, sq.query, sq.user, sq.timestamp, sq.total_results, + sq.web_count, sq.keywords +- _get_web_results: + MATCH (sq:SearchQuery {id: $search_id})-[f:FOUND]->(wr:WebResult) + RETURN wr.url, wr.title, wr.content, f.rank, f.rrf_score +- _mark_search_processed: + SET sq.processed = true + +These tests pin that the new UNWIND-based persistence still writes every +node property, label, and relationship property that consolidation reads. +""" + +import asyncio +import json +import uuid +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from src.models.hybrid_rag import HybridRAGConfig +from src.services.hybrid_rag_service import HybridRAGService + +TENANT = "llm_tester" +TENANT_BASE_LABEL = "User_Llm_Tester" +TENANT_DOC_LABEL = "User_Llm_Tester_Document" + + +@pytest.fixture +def mock_neo4j(): + neo4j = MagicMock() + neo4j.execute_query = AsyncMock(return_value=[]) + neo4j.execute_write = AsyncMock(return_value=[{"id": "sid"}]) + return neo4j + + +@pytest.fixture +def service(mock_neo4j): + vector = MagicMock() + response = MagicMock() + response.results = [] + vector.search = AsyncMock(return_value=response) + + graph = MagicMock() + graph.neo4j = mock_neo4j + graph.search_documents = AsyncMock(return_value=[]) + graph.get_related_documents = AsyncMock(return_value=[]) + + ollama = MagicMock() + ollama.generate_text = AsyncMock( + return_value='{"core_keywords": ["q"], "synonyms": {}}' + ) + + settings = MagicMock() + settings.ollama_llm_model = "test-model" + settings.vector_similarity_threshold = 0.7 + + return HybridRAGService( + vector_service=vector, + graph_service=graph, + searxng_client=MagicMock(), + ollama_client=ollama, + content_extractor=MagicMock(), + settings=settings, + ) + + +FINAL_RESULTS = [ + { + "result": {"page_id": 42, "title": "wiki hit"}, + "source_type": "wiki", "sources": ["vector"], + "rrf_score": 0.5, "final_rank": 1, + }, + { + "result": { + "url": "http://example.com", + "title": "web hit", + "content": "x" * 5000, # must be truncated to 1000 + }, + "source_type": "web", "sources": ["web"], + "rrf_score": 0.4, "final_rank": 2, + }, +] + + +async def _persist(service, **overrides): + kwargs = dict( + search_id="sid-123", + query="test query", + user=TENANT, + keywords_data={"core_keywords": ["docker"], "synonyms": {"k8s": ["kubernetes"]}}, + raw_results={"vector": [1], "graph": [], "web": [1, 2]}, + final_results=FINAL_RESULTS, + timing={"total_ms": 123.0}, + ) + kwargs.update(overrides) + return await service._persist_search_for_librarian(**kwargs) + + +@pytest.mark.unit +class TestPersistenceIsAtomic: + async def test_single_write_transaction_no_autocommit_queries( + self, service, mock_neo4j + ): + result = await _persist(service) + + assert result == "sid-123" + mock_neo4j.execute_write.assert_awaited_once() + mock_neo4j.execute_query.assert_not_awaited() + + async def test_failure_returns_none(self, service, mock_neo4j): + mock_neo4j.execute_write = AsyncMock(side_effect=RuntimeError("boom")) + + assert await _persist(service) is None + + +@pytest.mark.unit +class TestConsolidationShapeContract: + """Every property/label/relationship consolidation reads must be written.""" + + async def test_searchquery_node_shape(self, service, mock_neo4j): + await _persist(service) + cypher, params = mock_neo4j.execute_write.await_args.args[:2] + + # Interoperable label + tenant label (consolidation matches bare + # :SearchQuery, tenant scoping needs the prefixed label) + assert f":{TENANT_BASE_LABEL}_SearchQuery:SearchQuery" in cypher + # _find_unprocessed_searches filters on these + assert "processed: false" in cypher + assert "timestamp: datetime()" in cypher + # ... and returns these properties + for prop in ("id", "query", "user", "total_results", "web_count", "keywords"): + assert f"{prop}: ${'search_id' if prop == 'id' else prop}" in cypher, prop + assert params["search_id"] == "sid-123" + assert params["query"] == "test query" + assert params["user"] == TENANT + assert params["total_results"] == 2 + assert params["web_count"] == 2 + assert params["keywords"] == ["docker"] + assert json.loads(params["synonyms"]) == {"k8s": ["kubernetes"]} + + async def test_webresult_shape_and_found_relationship(self, service, mock_neo4j): + await _persist(service) + cypher, params = mock_neo4j.execute_write.await_args.args[:2] + + # _get_web_results traverses (sq)-[f:FOUND]->(wr:WebResult) and + # reads wr.url, wr.title, wr.content, f.rank, f.rrf_score + assert f":{TENANT_BASE_LABEL}_WebResult:WebResult" in cypher + for fragment in ("url: wl.url", "title: wl.title", "content: wl.content"): + assert fragment in cypher, fragment + assert "rank: wl.rank" in cypher + assert "rrf_score: wl.rrf_score" in cypher + assert 'source: "web"' in cypher + + web = params["web_links"] + assert len(web) == 1 + assert web[0]["url"] == "http://example.com" + assert web[0]["rank"] == 1 + assert web[0]["rrf_score"] == 0.4 + assert len(web[0]["content"]) == 1000 # truncation preserved + + async def test_document_links_tenant_scoped(self, service, mock_neo4j): + await _persist(service) + cypher, params = mock_neo4j.execute_write.await_args.args[:2] + + assert f"MATCH (d:{TENANT_DOC_LABEL}:Document {{page_id: link.page_id}})" in cypher + assert "MERGE (sq)-[f:FOUND]->(d)" in cypher + for fragment in ("f.source = link.source", "f.rank = link.rank", + "f.rrf_score = link.rrf_score", + "f.final_rank = link.final_rank"): + assert fragment in cypher, fragment + + docs = params["doc_links"] + assert docs == [{ + "page_id": 42, "source": "wiki", "rank": 1, + "rrf_score": 0.5, "final_rank": 1, + }] + + async def test_empty_doc_links_cannot_swallow_web_results( + self, service, mock_neo4j + ): + """UNWIND [] yields no rows; the CALL subqueries must isolate that.""" + await _persist(service, final_results=[FINAL_RESULTS[1]]) + cypher, params = mock_neo4j.execute_write.await_args.args[:2] + + assert params["doc_links"] == [] + assert len(params["web_links"]) == 1 + # Both UNWINDs live in aggregating CALL subqueries + assert cypher.count("CALL {") == 2 + + +@pytest.mark.unit +class TestPersistenceOffHotPath: + async def test_search_returns_upfront_id_and_persists_in_background( + self, service, mock_neo4j + ): + config = HybridRAGConfig( + enable_vector=True, enable_graph=False, enable_web=False, + enable_volatile=False, enable_documents=False, + enable_reranking=False, enable_enrichment=False, + ) + + response = await service.search("q", TENANT, config) + + # search_id is generated up front and returned immediately + assert response.search_id + uuid.UUID(response.search_id) # valid uuid4 + assert response.timing.persistence_ms == 0.0 + + # The write happens in a background task, not on the request path + pending = list(service._background_tasks) + assert len(pending) == 1 + await asyncio.gather(*pending) + + mock_neo4j.execute_write.assert_awaited_once() + params = mock_neo4j.execute_write.await_args.args[1] + assert params["search_id"] == response.search_id diff --git a/tests/test_tenant_scoping.py b/tests/test_tenant_scoping.py index d1c8b9b..8cd83db 100644 --- a/tests/test_tenant_scoping.py +++ b/tests/test_tenant_scoping.py @@ -65,6 +65,7 @@ def mock_neo4j(): neo4j = MagicMock() neo4j.execute_query = AsyncMock(return_value=[]) neo4j.execute_read = AsyncMock(return_value=[]) + neo4j.execute_write = AsyncMock(return_value=[]) return neo4j @@ -362,18 +363,20 @@ class TestPersistenceScoping: ] await hybrid_service._persist_search_for_librarian( - query="q", user=TENANT, keywords_data={}, + search_id="sid-1", query="q", user=TENANT, keywords_data={}, raw_results={}, final_results=final_results, timing={}, ) - cypher = _all_cypher(mock_neo4j) + # Persistence is now ONE atomic write transaction + mock_neo4j.execute_write.assert_awaited_once() + cypher = str(mock_neo4j.execute_write.await_args.args[0]) # SearchQuery + WebResult nodes carry the tenant label assert f"{TENANT_BASE_LABEL}_SearchQuery" in cypher assert f"{TENANT_BASE_LABEL}_WebResult" in cypher # FOUND link matches only this tenant's Document nodes - assert f"MATCH (d:{TENANT_DOC_LABEL}:Document {{page_id: $page_id}})" in cypher + assert f"MATCH (d:{TENANT_DOC_LABEL}:Document {{page_id: link.page_id}})" in cypher # The old cross-tenant match must be gone - assert "MATCH (d:Document {page_id: $page_id})" not in cypher + assert "MATCH (d:Document {page_id:" not in cypher # =============================================================================