jpmschweitzer/library-desk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

perf: switch Qdrant to AsyncQdrantClient with explicit timeout

Browse Source 
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
This commit is contained in:
jpmschweitzer
2026-07-14 12:58:01 +02:00
co-authored by Claude Fable 5
parent 0b346d3a57
commit 406143e7ae
8 changed files with 53 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+4
View File
@@ -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
+31 -26
View File
@@ -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
+1
View File
@@ -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
+3 -2
View File
@@ -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
+2 -2
View File
@@ -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
+6 -14
View File
@@ -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
+1 -1
View File
@@ -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
+5 -3
View File
@@ -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"}
)
# =============================================================================