improve: qdrant upsert fallback

Author: ACI Bot

src/aci/infrastructure/vector_store/qdrant.py

@@ -31,6 +31,9 @@ def __init__(
         vector_size: int = 1536,
         api_key: str | None = None,
         url: str | None = None,
+        timeout_seconds: float = 60.0,
+        write_retry_attempts: int = 3,
+        write_retry_backoff_seconds: float = 0.5,
     ):
         url = (url or "").strip()
         if not url and host.startswith(("http://", "https://")):
@@ -41,14 +44,20 @@ def __init__(
         self._collection_name = collection_name
         self._vector_size = vector_size
         self._api_key = api_key or None
+        self._timeout_seconds = timeout_seconds
+        self._write_retry_attempts = max(1, write_retry_attempts)
+        self._write_retry_backoff_seconds = max(0.0, write_retry_backoff_seconds)
         self._client: AsyncQdrantClient | None = None
         self._initialized_collections: set[str] = set()
         self._init_locks: dict[str, asyncio.Lock] = {}
 
     async def _get_client(self) -> AsyncQdrantClient:
         """Get or create the Qdrant client."""
         if self._client is None:
-            client_kwargs: dict = {"api_key": self._api_key}
+            client_kwargs: dict = {
+                "api_key": self._api_key,
+                "timeout": self._timeout_seconds,
+            }
             if self._url:
                 client_kwargs["url"] = self._url
             else:
@@ -57,6 +66,33 @@ async def _get_client(self) -> AsyncQdrantClient:
             self._client = AsyncQdrantClient(**client_kwargs)
         return self._client
 
+    def _format_exception_chain(self, exc: Exception) -> str:
+        """Return a readable error string including nested causes."""
+        parts: list[str] = []
+        current: Exception | None = exc
+        depth = 0
+        while current is not None and depth < 5:
+            message = str(current).strip()
+            if not message:
+                message = repr(current)
+            parts.append(f"{type(current).__name__}: {message}")
+            current = current.__cause__
+            depth += 1
+        return " <- ".join(parts)
+
+    def _is_retryable_exception(self, exc: Exception) -> bool:
+        """Return True for transient timeout-style failures."""
+        current: Exception | None = exc
+        depth = 0
+        while current is not None and depth < 5:
+            name = type(current).__name__.lower()
+            message = str(current).lower()
+            if "timeout" in name or "timed out" in message or "timeout" in message:
+                return True
+            current = current.__cause__
+            depth += 1
+        return False
+
     def set_collection(self, collection_name: str) -> None:
         """Switch to a different collection."""
         self._collection_name = collection_name
@@ -143,15 +179,38 @@ async def upsert(
         if "artifact_type" not in payload:
             payload = {**payload, "artifact_type": "chunk"}
 
-        try:
-            await client.upsert(
-                collection_name=target_collection,
-                points=[
-                    models.PointStruct(id=chunk_id, vector=vector, payload=payload)
-                ],
-            )
-        except Exception as e:
-            raise VectorStoreError(f"Failed to upsert vector: {e}") from e
+        last_error: Exception | None = None
+        for attempt in range(1, self._write_retry_attempts + 1):
+            try:
+                await client.upsert(
+                    collection_name=target_collection,
+                    points=[
+                        models.PointStruct(id=chunk_id, vector=vector, payload=payload)
+                    ],
+                )
+                return
+            except Exception as e:
+                last_error = e
+                if attempt < self._write_retry_attempts and self._is_retryable_exception(e):
+                    delay = self._write_retry_backoff_seconds * (2 ** (attempt - 1))
+                    logger.warning(
+                        "Qdrant upsert timeout (attempt %s/%s) for chunk '%s' in collection '%s'; retrying in %.2fs",
+                        attempt,
+                        self._write_retry_attempts,
+                        chunk_id,
+                        target_collection,
+                        delay,
+                    )
+                    if delay > 0:
+                        await asyncio.sleep(delay)
+                    continue
+                break
+
+        details = self._format_exception_chain(last_error or Exception("unknown error"))
+        raise VectorStoreError(
+            f"Failed to upsert vector in collection '{target_collection}' after "
+            f"{self._write_retry_attempts} attempt(s): {details}"
+        ) from last_error
 
     async def upsert_batch(
         self,
@@ -163,24 +222,48 @@ async def upsert_batch(
         await self.initialize(target_collection)
         client = await self._get_client()
 
-        try:
-            qdrant_points = [
-                models.PointStruct(
-                    id=chunk_id,
-                    vector=vector,
-                    payload=(
-                        payload if "artifact_type" in payload
-                        else {**payload, "artifact_type": "chunk"}
-                    ),
-                )
-                for chunk_id, vector, payload in points
-            ]
-            await client.upsert(
-                collection_name=target_collection,
-                points=qdrant_points,
+        qdrant_points = [
+            models.PointStruct(
+                id=chunk_id,
+                vector=vector,
+                payload=(
+                    payload if "artifact_type" in payload
+                    else {**payload, "artifact_type": "chunk"}
+                ),
             )
-        except Exception as e:
-            raise VectorStoreError(f"Failed to batch upsert vectors: {e}") from e
+            for chunk_id, vector, payload in points
+        ]
+
+        last_error: Exception | None = None
+        for attempt in range(1, self._write_retry_attempts + 1):
+            try:
+                await client.upsert(
+                    collection_name=target_collection,
+                    points=qdrant_points,
+                )
+                return
+            except Exception as e:
+                last_error = e
+                if attempt < self._write_retry_attempts and self._is_retryable_exception(e):
+                    delay = self._write_retry_backoff_seconds * (2 ** (attempt - 1))
+                    logger.warning(
+                        "Qdrant batch upsert timeout (attempt %s/%s) for %s point(s) in collection '%s'; retrying in %.2fs",
+                        attempt,
+                        self._write_retry_attempts,
+                        len(points),
+                        target_collection,
+                        delay,
+                    )
+                    if delay > 0:
+                        await asyncio.sleep(delay)
+                    continue
+                break
+
+        details = self._format_exception_chain(last_error or Exception("unknown error"))
+        raise VectorStoreError(
+            f"Failed to batch upsert {len(points)} vector(s) in collection "
+            f"'{target_collection}' after {self._write_retry_attempts} attempt(s): {details}"
+        ) from last_error
 
     async def delete_by_file(self, file_path: str, collection_name: str | None = None) -> int:
         """Delete all vectors for a file, return count deleted."""

src/aci/services/indexing_service.py

@@ -723,6 +723,7 @@ async def _embed_and_store_chunks(
 
                 # Collect file info from this batch before writing vectors
                 batch_file_infos: dict[str, dict] = {}
+                batch_points: list[tuple[str, list[float], dict]] = []
 
                 # Time Qdrant upsert operations
                 qdrant_start = time.time()
@@ -739,17 +740,17 @@ async def _embed_and_store_chunks(
                         "artifact_type": ArtifactType.CHUNK.value,
                         **chunk.metadata,
                     }
-                    await self._vector_store.upsert(
-                        chunk.chunk_id,
-                        embedding,
-                        payload,
-                        collection_name=collection_name,
-                    )
+                    batch_points.append((chunk.chunk_id, embedding, payload))
 
                     file_path = pending_info["file_path"] if pending_info else None
                     if file_path and file_path not in persisted_files:
                         batch_file_infos[file_path] = pending_info
 
+                await self._upsert_points(
+                    batch_points,
+                    collection_name=collection_name,
+                )
+
                 qdrant_duration_ms = (time.time() - qdrant_start) * 1000
 
                 # Log Qdrant operation with structured fields
@@ -818,6 +819,7 @@ async def _embed_and_store_chunks(
                         actual=len(embeddings),
                     )
 
+                summary_points: list[tuple[str, list[float], dict]] = []
                 for summary, embedding in zip(batch, embeddings, strict=False):
                     payload = {
                         "file_path": summary.file_path,
@@ -828,19 +830,47 @@ async def _embed_and_store_chunks(
                         "name": summary.name,
                         **summary.metadata,
                     }
-                    await self._vector_store.upsert(
-                        summary.artifact_id,
-                        embedding,
-                        payload,
-                        collection_name=collection_name,
-                    )
+                    summary_points.append((summary.artifact_id, embedding, payload))
+
+                await self._upsert_points(
+                    summary_points,
+                    collection_name=collection_name,
+                )
 
                 self._report_progress(
                     total_chunks + min(i + self._batch_size, total_summaries),
                     total_items,
                     f"Embedded {min(i + self._batch_size, total_summaries)} summaries",
                 )
 
+    async def _upsert_points(
+        self,
+        points: list[tuple[str, list[float], dict]],
+        *,
+        collection_name: str | None,
+    ) -> None:
+        """
+        Upsert points with best-effort batching.
+
+        Uses vector-store batch upsert when available, and falls back to
+        per-point upsert for compatibility with lightweight test doubles.
+        """
+        if not points:
+            return
+
+        upsert_batch = getattr(self._vector_store, "upsert_batch", None)
+        if callable(upsert_batch):
+            await upsert_batch(points, collection_name=collection_name)
+            return
+
+        for chunk_id, embedding, payload in points:
+            await self._vector_store.upsert(
+                chunk_id,
+                embedding,
+                payload,
+                collection_name=collection_name,
+            )
+
     async def update_incremental(
         self, root_path: Path, *, max_workers: int | None = None
     ) -> IndexingResult: