perf: switch Qdrant to AsyncQdrantClient with explicit timeout

Every vector call ran on the sync QdrantClient inside async wrapper
methods, blocking the FastAPI event loop per Qdrant round-trip. The
wrapper now holds an AsyncQdrantClient (timeout via QDRANT_TIMEOUT,
default 30s) and awaits all client calls; the wrapper API is unchanged.

Call sites off the wrapper were fixed too: the HybridRAG document leg
now uses the async search_vectors wrapper instead of the deprecated raw
client.search (also fixing its call to the nonexistent ollama.embed_text
which made the leg permanently report 'failed'), the health check awaits
get_collections, and document_sync's raw delete/upsert calls are awaited
(routed through wrappers in the next commit).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QbFZyDvYksazX6nYQYZ67L

CHANGELOG.md

@@ -18,6 +18,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Weekly quality report** — `POST /maintenance/quality-report {user}` runs the duplicate scan, flags stale pages (not updated in N days AND ≤ M SearchQuery hits from the graph data), lists pages missing tags/description, folds in the latest integrity-check results (Redis-cached or run inline), and writes a dated report page to `users/{user}/system/quality-reports/YYYY-MM-DD` (same-day reruns update the same page — the page id is remembered in Redis because the Wiki.js listing lags page creation). Response returns the full report content + page path. Verified end-to-end against the local dev server as `llm_tester`.
 - **Nightly integrity check** — `POST /maintenance/integrity-check {user}` (read-only: reports, never auto-fixes) reports per tenant: wiki pages with ZERO vectors in Qdrant (silent-skip reindex victims), orphaned vectors whose wiki page no longer exists, unexpected Qdrant collections (test-tenant residue and unknown namespaces flagged; other services' collections counted as foreign), Neo4j Document nodes without wiki counterparts, plus counts and duration. The latest report is cached in Redis (30 days) so the weekly quality report can fold it in.
 
+### Changed (performance)
+
+- **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
 
 - **BREAKING: `user` is now required on every tenant-data endpoint** - The implicit `jpmschweitzer` default tenant (`DEFAULT_USER`) has been removed everywhere. All endpoints that read or write tenant data (`/query/*`, `/wiki/*`, `/vector/*`, `/graph/*`, `/ingest/*`, `/volatile/*`, `/documents/*`, `/stats`, `/rag/search`) now reject requests without an explicit, non-empty, non-whitespace `user` (HTTP 422), matching the existing `/maintenance/*` pattern. A shared validator (`require_user` dependency / `RequiredUser` model type) also rejects blank users. The Wiki.js change listener now skips changes whose notification email yields no user instead of attributing them to the production tenant. **Caller coordination required:** tatlock and the Scheduler ingest/prefetch/consolidation tasks must send an explicit `user` on every call — see the deploy checklist.

src/clients/qdrant_client.py

@@ -8,7 +8,7 @@ Provides async vector operations with:
 - Similarity queries
 """
 
-from qdrant_client import QdrantClient
+from qdrant_client import AsyncQdrantClient
 from qdrant_client.models import (
     Distance, VectorParams, PointStruct,
     Filter, FieldCondition, MatchValue, Range
@@ -30,17 +30,22 @@ class QdrantClientWrapper:
     Each user has isolated vector collection for their documents.
     """
 
-    def __init__(self, url: str, embedding_dim: int = 768):
+    def __init__(self, url: str, embedding_dim: int = 768, timeout: float = 30.0):
         """
         Initialize Qdrant client.
 
+        Uses AsyncQdrantClient so vector calls never block the FastAPI
+        event loop, with an explicit timeout so a hung Qdrant cannot
+        stall requests indefinitely.
+
         Args:
             url: Qdrant server URL (e.g., "http://qdrant:6333")
             embedding_dim: Vector embedding dimension (default 768 for nomic-embed-text)
+            timeout: Per-request timeout in seconds (default 30.0)
         """
-        self.client = QdrantClient(url=url)
+        self.client = AsyncQdrantClient(url=url, timeout=timeout)
         self.embedding_dim = embedding_dim
-        logger.info(f"Initialized Qdrant client: {url}")
+        logger.info(f"Initialized async Qdrant client: {url} (timeout={timeout}s)")
 
     def get_collection_name(self, user: str) -> str:
         """
@@ -62,12 +67,12 @@ class QdrantClientWrapper:
             collection_name: Collection name
         """
         try:
-            collections = self.client.get_collections()
+            collections = await self.client.get_collections()
             existing = [c.name for c in collections.collections]
 
             if collection_name not in existing:
                 logger.info(f"Creating Qdrant collection: {collection_name}")
-                self.client.create_collection(
+                await self.client.create_collection(
                     collection_name=collection_name,
                     vectors_config=VectorParams(
                         size=self.embedding_dim,
@@ -90,7 +95,7 @@ class QdrantClientWrapper:
             True if collection exists
         """
         try:
-            collections = self.client.get_collections()
+            collections = await self.client.get_collections()
             existing = [c.name for c in collections.collections]
             return collection_name in existing
         except Exception as e:
@@ -161,7 +166,7 @@ class QdrantClientWrapper:
             ))
 
         try:
-            self.client.upsert(
+            await self.client.upsert(
                 collection_name=collection_name,
                 points=points
             )
@@ -213,7 +218,7 @@ class QdrantClientWrapper:
             query_filter = Filter(must=conditions)
 
         try:
-            response = self.client.query_points(
+            response = await self.client.query_points(
                 collection_name=collection_name,
                 query=query_vector,
                 limit=limit,
@@ -251,7 +256,7 @@ class QdrantClientWrapper:
         collection_name = self.get_collection_name(user)
 
         try:
-            self.client.delete(
+            await self.client.delete(
                 collection_name=collection_name,
                 points_selector=Filter(
                     must=[
@@ -287,7 +292,7 @@ class QdrantClientWrapper:
 
         try:
             # Scroll through points with doc_id filter
-            points, _ = self.client.scroll(
+            points, _ = await self.client.scroll(
                 collection_name=collection_name,
                 scroll_filter=Filter(
                     must=[
@@ -329,7 +334,7 @@ class QdrantClientWrapper:
 
         try:
             # Get collection info
-            collection_info = self.client.get_collection(collection_name)
+            collection_info = await self.client.get_collection(collection_name)
             # This gives total points, not unique docs
             # For unique docs, would need to aggregate by doc_id
             return collection_info.points_count
@@ -352,7 +357,7 @@ class QdrantClientWrapper:
         collection_name = self.get_collection_name(user)
 
         try:
-            self.client.delete_collection(collection_name)
+            await self.client.delete_collection(collection_name)
             logger.warning(f"Deleted collection: {collection_name}")
             return True
         except Exception as e:
@@ -382,7 +387,7 @@ class QdrantClientWrapper:
 
         try:
             # Get source document chunks with vectors
-            source_points, _ = self.client.scroll(
+            source_points, _ = await self.client.scroll(
                 collection_name=collection_name,
                 scroll_filter=Filter(
                     must=[
@@ -404,7 +409,7 @@ class QdrantClientWrapper:
             # (could aggregate multiple chunks for better results)
             first_vector = source_points[0].vector
 
-            response = self.client.query_points(
+            response = await self.client.query_points(
                 collection_name=collection_name,
                 query=first_vector,
                 limit=limit * 2,  # Get more to filter out same doc
@@ -449,7 +454,7 @@ class QdrantClientWrapper:
             True if successful
         """
         try:
-            self.client.upsert(
+            await self.client.upsert(
                 collection_name=collection_name,
                 points=[PointStruct(
                     id=vector_id,
@@ -487,7 +492,7 @@ class QdrantClientWrapper:
             query_filter = Filter(must=conditions)
 
             # Delete points
-            result = self.client.delete(
+            result = await self.client.delete(
                 collection_name=collection_name,
                 points_selector=query_filter
             )
@@ -532,7 +537,7 @@ class QdrantClientWrapper:
             query_filter = Filter(must=conditions)
 
         try:
-            response = self.client.query_points(
+            response = await self.client.query_points(
                 collection_name=collection_name,
                 query=query_vector,
                 limit=limit,
@@ -589,7 +594,7 @@ class QdrantClientWrapper:
 
         try:
             while True:
-                points, next_offset = self.client.scroll(
+                points, next_offset = await self.client.scroll(
                     collection_name=collection_name,
                     scroll_filter=scroll_filter,
                     limit=batch_size,
@@ -636,7 +641,7 @@ class QdrantClientWrapper:
             return 0
 
         try:
-            self.client.delete(
+            await self.client.delete(
                 collection_name=collection_name,
                 points_selector=point_ids
             )
@@ -655,13 +660,13 @@ class QdrantClientWrapper:
             List of collection info dictionaries
         """
         try:
-            collections = self.client.get_collections()
+            collections = await self.client.get_collections()
             result = []
 
             for coll in collections.collections:
                 # Get detailed collection info
                 try:
-                    info = self.client.get_collection(coll.name)
+                    info = await self.client.get_collection(coll.name)
                     result.append({
                         "name": coll.name,
                         "vectors_count": info.vectors_count or 0,
@@ -717,7 +722,7 @@ class QdrantClientWrapper:
         )
 
         try:
-            response = self.client.query_points(
+            response = await self.client.query_points(
                 collection_name=collection_name,
                 query=query_vector,
                 limit=limit,
@@ -768,7 +773,7 @@ class QdrantClientWrapper:
             count = 0
             offset = None
             while True:
-                points, next_offset = self.client.scroll(
+                points, next_offset = await self.client.scroll(
                     collection_name=collection_name,
                     scroll_filter=expiry_filter,
                     limit=100,
@@ -784,7 +789,7 @@ class QdrantClientWrapper:
                 return 0
 
             # Delete expired points
-            self.client.delete(
+            await self.client.delete(
                 collection_name=collection_name,
                 points_selector=expiry_filter
             )
@@ -804,7 +809,7 @@ class QdrantClientWrapper:
             List of volatile collection names
         """
         try:
-            collections = self.client.get_collections()
+            collections = await self.client.get_collections()
             return [
                 c.name for c in collections.collections
                 if c.name.startswith("volatile_")

src/config.py

@@ -40,6 +40,7 @@ class Settings(BaseSettings):
     # Qdrant Configuration
     qdrant_host: str = Field(default="qdrant", description="Qdrant host")
     qdrant_port: int = Field(default=6333, description="Qdrant port")
+    qdrant_timeout: int = Field(default=30, ge=1, le=300, description="Qdrant client timeout in seconds")
 
     # Wiki.js Configuration
     wikijs_url: str = Field(default="http://wiki:3000", description="Wiki.js URL")

src/core/dependencies.py

@@ -101,7 +101,8 @@ def get_qdrant_client() -> QdrantClientWrapper:
     settings = get_settings()
     client = QdrantClientWrapper(
         url=settings.qdrant_url,
-        embedding_dim=768  # nomic-embed-text default
+        embedding_dim=768,  # nomic-embed-text default
+        timeout=settings.qdrant_timeout
     )
     logger.debug("Created Qdrant client instance")
     return client
@@ -591,7 +592,7 @@ async def check_service_health() -> dict:
     try:
         qdrant = get_qdrant_client()
         # Check if we can list collections
-        collections = qdrant.client.get_collections()
+        await qdrant.client.get_collections()
         health["qdrant"] = True
     except Exception as e:
         logger.error(f"Qdrant health check failed: {e}")

src/services/document_sync_service.py

@@ -198,7 +198,7 @@ class DocumentSyncService:
 
         # Delete existing chunks for this document
         try:
-            self.qdrant.client.delete(
+            await self.qdrant.client.delete(
                 collection_name=collection,
                 points_selector={
                     "filter": {
@@ -242,7 +242,7 @@ class DocumentSyncService:
 
         # Upsert to Qdrant
         if points:
-            self.qdrant.client.upsert(
+            await self.qdrant.client.upsert(
                 collection_name=collection,
                 points=points
             )

src/services/hybrid_rag_service.py

@@ -485,34 +485,26 @@ JSON:"""
                         return [], (time.time() - start) * 1000, None
 
                     # Get query embedding
-                    query_embedding = await self.vector.ollama.embed_text(query)
+                    query_embedding = await self.vector.ollama.embed(query)
 
-                    # Search with filter for doc_type=document
-                    from qdrant_client.models import Filter, FieldCondition, MatchValue
-                    search_results = self.vector.qdrant.client.search(
+                    # Search via the async wrapper with doc_type=document filter
+                    search_results = await self.vector.qdrant.search_vectors(
                         collection_name=collection_name,
                         query_vector=query_embedding,
                         limit=config.document_limit,
                         score_threshold=config.document_threshold,
-                        query_filter=Filter(
-                            must=[
-                                FieldCondition(
-                                    key="doc_type",
-                                    match=MatchValue(value="document")
-                                )
-                            ]
-                        )
+                        filter_conditions={"doc_type": "document"}
                     )
 
                     # Format results
                     formatted = []
                     for r in search_results:
-                        payload = r.payload or {}
+                        payload = r.get("payload") or {}
                         formatted.append({
                             "paperless_id": payload.get("paperless_id"),
                             "title": payload.get("title", "Untitled Document"),
                             "content": payload.get("chunk_text", ""),
-                            "score": r.score,
+                            "score": r["score"],
                             "correspondent": payload.get("correspondent"),
                             "document_type": payload.get("document_type"),
                             "tags": payload.get("tags", []),

tests/test_integration.py

@@ -148,7 +148,7 @@ class TestQdrantIntegration:
         collection_name = qdrant_client.get_collection_name(test_user)
         await qdrant_client.ensure_collection(collection_name)
 
-        collections = qdrant_client.client.get_collections()
+        collections = await qdrant_client.client.get_collections()
         collection_names = [c.name for c in collections.collections]
 
         assert collection_name in collection_names

tests/test_tenant_scoping.py

@@ -42,8 +42,6 @@ def mock_qdrant():
     qdrant.search_with_expiry_filter = AsyncMock(return_value=[])
     qdrant.upsert_vector = AsyncMock(return_value=True)
     qdrant.delete_by_filter = AsyncMock(return_value=0)
-    qdrant.client = MagicMock()
-    qdrant.client.search = MagicMock(return_value=[])
     return qdrant
 
 
@@ -307,9 +305,13 @@ class TestDocumentLegScoping:
         await hybrid_service._retrieve_parallel("q", TENANT, config, {})
 
         assert (
-            mock_qdrant.client.search.call_args.kwargs["collection_name"]
+            mock_qdrant.search_vectors.await_args.kwargs["collection_name"]
             == TENANT_COLLECTION
         )
+        assert (
+            mock_qdrant.search_vectors.await_args.kwargs["filter_conditions"]
+            == {"doc_type": "document"}
+        )
 
 
 # =============================================================================