fix(vector): address PR review round 8 — anyio convention + cosine-safe sentinel + dedup get_collection
Reviewer findings (1 blocking + 2 important): - 🔴 Replace `import asyncio` / `await asyncio.sleep(0)` with `import anyio` / `await anyio.sleep(0)` in the four async side-effect helpers (_scroll_raises, _upsert_raises, _get_collection_raises, _create_index). CLAUDE.md mandates anyio for all async operations; conftest pins the backend to asyncio so the asyncio.sleep call worked today, but the inconsistency would surface the moment that pin moves. - 🟡 Replace the sentinel's zero dense vector with a single non-zero element (`[1e-9] + [0.0] * (dimension - 1)`). Cosine distance is mathematically undefined for the zero vector and Qdrant Cloud strict mode rejects zero-vector upserts. The exact value doesn't matter (sentinel never participates in a search — no user_id/doc_id/doc_type payload) but the upsert itself must be valid. - 🟡 Avoid the duplicate `get_collection` round-trip on every restart. `_ensure_payload_indexes` now accepts an optional `existing_schema: dict | None` parameter; when None it fetches collection_info itself (and the get_collection-failure swallow still applies), but `get_qdrant_client` already fetches collection_info for dimension validation in the existing-collection branch — pass `collection_info.payload_schema or {}` through to skip the second call. The new-collection branch passes `existing_schema={}` explicitly since a freshly created collection has no payload schema. The 🟡 deck_card iteration-fallback finding doesn't apply: the `isdigit()` guard at context.py:612 returns early before either the fast-path or the iteration fallback runs, so non-numeric doc_ids cannot reach the inner `c.id == int(doc_id)` comparison. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.7
parent
9720a7e4fe
commit
d390b3a4b8
@@ -48,18 +48,28 @@ _qdrant_client: AsyncQdrantClient | None = None
|
|||||||
|
|
||||||
|
|
||||||
async def _ensure_payload_indexes(
|
async def _ensure_payload_indexes(
|
||||||
client: AsyncQdrantClient, collection_name: str
|
client: AsyncQdrantClient,
|
||||||
|
collection_name: str,
|
||||||
|
existing_schema: dict[str, Any] | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Create payload indexes for fields used in exact-match filters.
|
"""Create payload indexes for fields used in exact-match filters.
|
||||||
|
|
||||||
Each entry in ``_PAYLOAD_INDEX_FIELDS`` is created with its declared
|
Each entry in ``_PAYLOAD_INDEX_FIELDS`` is created with its declared
|
||||||
schema type (KEYWORD for string fields, BOOL for ``is_placeholder``).
|
schema type (KEYWORD for string fields, BOOL for ``is_placeholder``).
|
||||||
Pre-fetches the existing payload schema and skips fields that are
|
Skips fields that are already in ``existing_schema`` so routine
|
||||||
already indexed, so routine restarts make no Qdrant write round-trips
|
restarts make no Qdrant write round-trips and emit no INFO log lines.
|
||||||
and emit no INFO log lines. Schema conflicts (a pre-existing index
|
Schema conflicts (a pre-existing index with a different type) still
|
||||||
with a different type) still surface as a 400 — log loudly so
|
surface as a 400 — log loudly so operators can intervene, but keep
|
||||||
operators can intervene, but keep going so the remaining fields still
|
going so the remaining fields still get indexed.
|
||||||
get indexed.
|
|
||||||
|
Args:
|
||||||
|
client: Qdrant client instance.
|
||||||
|
collection_name: Target collection.
|
||||||
|
existing_schema: The collection's current ``payload_schema``. If
|
||||||
|
``None``, this function fetches it via ``get_collection``;
|
||||||
|
callers that have already fetched the collection info (e.g.
|
||||||
|
``get_qdrant_client``'s dimension-validation step) should pass
|
||||||
|
it through to avoid a duplicate round-trip.
|
||||||
"""
|
"""
|
||||||
# Mirror the broad swallow in `_backfill_doc_id_to_string`: the singleton
|
# Mirror the broad swallow in `_backfill_doc_id_to_string`: the singleton
|
||||||
# in `get_qdrant_client` is already assigned by the time this function
|
# in `get_qdrant_client` is already assigned by the time this function
|
||||||
@@ -67,6 +77,7 @@ async def _ensure_payload_indexes(
|
|||||||
# propagating out would leave the process holding a usable client with
|
# propagating out would leave the process holding a usable client with
|
||||||
# the migration silently skipped on every subsequent call. Log ERROR
|
# the migration silently skipped on every subsequent call. Log ERROR
|
||||||
# with exc_info and return; the next process restart retries from scratch.
|
# with exc_info and return; the next process restart retries from scratch.
|
||||||
|
if existing_schema is None:
|
||||||
try:
|
try:
|
||||||
collection_info = await client.get_collection(collection_name)
|
collection_info = await client.get_collection(collection_name)
|
||||||
except Exception:
|
except Exception:
|
||||||
@@ -301,16 +312,20 @@ async def _backfill_doc_id_to_string(
|
|||||||
|
|
||||||
# Data backfill succeeded — write the sentinel so a future restart can
|
# Data backfill succeeded — write the sentinel so a future restart can
|
||||||
# short-circuit. Empty sparse vector mirrors the placeholder.py
|
# short-circuit. Empty sparse vector mirrors the placeholder.py
|
||||||
# convention (vector/placeholder.py); zero dense vector is fine
|
# convention (vector/placeholder.py). The dense vector uses a single
|
||||||
# because the sentinel never participates in a search (no user_id /
|
# non-zero element instead of all zeros: cosine distance is undefined
|
||||||
# doc_id / doc_type payload to match). A failure here is non-fatal:
|
# for the zero vector and Qdrant Cloud's strict mode rejects zero-vector
|
||||||
# the data is correct; only the short-circuit marker is missing, so
|
# upserts. The sentinel still never participates in a search (no
|
||||||
# the next restart will re-scroll an already-clean collection (idempotent
|
# user_id / doc_id / doc_type payload to match), so the exact value
|
||||||
# zero-write) before retrying the upsert.
|
# doesn't matter — it just has to be normalisable.
|
||||||
|
# A failure here is non-fatal: the data is correct; only the short-circuit
|
||||||
|
# marker is missing, so the next restart will re-scroll an already-clean
|
||||||
|
# collection (idempotent zero-write) before retrying the upsert.
|
||||||
|
sentinel_dense = [1e-9] + [0.0] * (dimension - 1)
|
||||||
sentinel_point = PointStruct(
|
sentinel_point = PointStruct(
|
||||||
id=_DOC_ID_BACKFILL_SENTINEL_ID,
|
id=_DOC_ID_BACKFILL_SENTINEL_ID,
|
||||||
vector={
|
vector={
|
||||||
"dense": [0.0] * dimension,
|
"dense": sentinel_dense,
|
||||||
"sparse": models.SparseVector(indices=[], values=[]),
|
"sparse": models.SparseVector(indices=[], values=[]),
|
||||||
},
|
},
|
||||||
payload=dict(_DOC_ID_BACKFILL_SENTINEL_PAYLOAD),
|
payload=dict(_DOC_ID_BACKFILL_SENTINEL_PAYLOAD),
|
||||||
@@ -443,11 +458,17 @@ async def get_qdrant_client() -> AsyncQdrantClient:
|
|||||||
|
|
||||||
# Existing collections may pre-date the doc_id normalization /
|
# Existing collections may pre-date the doc_id normalization /
|
||||||
# payload-index work. Backfill before creating the index so the
|
# payload-index work. Backfill before creating the index so the
|
||||||
# index covers every point.
|
# index covers every point. Pass the already-fetched
|
||||||
|
# collection_info.payload_schema through to avoid a redundant
|
||||||
|
# get_collection round-trip on every restart.
|
||||||
await _backfill_doc_id_to_string(
|
await _backfill_doc_id_to_string(
|
||||||
_qdrant_client, collection_name, expected_dimension
|
_qdrant_client, collection_name, expected_dimension
|
||||||
)
|
)
|
||||||
await _ensure_payload_indexes(_qdrant_client, collection_name)
|
await _ensure_payload_indexes(
|
||||||
|
_qdrant_client,
|
||||||
|
collection_name,
|
||||||
|
existing_schema=collection_info.payload_schema or {},
|
||||||
|
)
|
||||||
|
|
||||||
else:
|
else:
|
||||||
# Collection doesn't exist - create it
|
# Collection doesn't exist - create it
|
||||||
@@ -480,6 +501,10 @@ async def get_qdrant_client() -> AsyncQdrantClient:
|
|||||||
f" Distance: COSINE\n"
|
f" Distance: COSINE\n"
|
||||||
f"Background sync will index all documents with dense + sparse vectors."
|
f"Background sync will index all documents with dense + sparse vectors."
|
||||||
)
|
)
|
||||||
await _ensure_payload_indexes(_qdrant_client, collection_name)
|
# Freshly created collection has no payload schema yet; pass {}
|
||||||
|
# explicitly to skip the otherwise-redundant get_collection call.
|
||||||
|
await _ensure_payload_indexes(
|
||||||
|
_qdrant_client, collection_name, existing_schema={}
|
||||||
|
)
|
||||||
|
|
||||||
return _qdrant_client
|
return _qdrant_client
|
||||||
|
|||||||
@@ -14,10 +14,10 @@ run at startup. Producer-side normalization is exercised by the existing
|
|||||||
scanner tests.
|
scanner tests.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import asyncio
|
|
||||||
from types import SimpleNamespace
|
from types import SimpleNamespace
|
||||||
from unittest.mock import call
|
from unittest.mock import call
|
||||||
|
|
||||||
|
import anyio
|
||||||
import httpx
|
import httpx
|
||||||
import pytest
|
import pytest
|
||||||
from qdrant_client.http.exceptions import UnexpectedResponse
|
from qdrant_client.http.exceptions import UnexpectedResponse
|
||||||
@@ -237,7 +237,7 @@ async def test_ensure_payload_indexes_logs_and_returns_when_get_collection_raise
|
|||||||
|
|
||||||
async def _get_collection_raises(*args, **kwargs):
|
async def _get_collection_raises(*args, **kwargs):
|
||||||
# See _scroll_raises in the backfill section for why this is async.
|
# See _scroll_raises in the backfill section for why this is async.
|
||||||
await asyncio.sleep(0)
|
await anyio.sleep(0)
|
||||||
raise RuntimeError("connection refused")
|
raise RuntimeError("connection refused")
|
||||||
|
|
||||||
client.get_collection.side_effect = _get_collection_raises
|
client.get_collection.side_effect = _get_collection_raises
|
||||||
@@ -481,11 +481,11 @@ async def test_backfill_logs_and_returns_when_scroll_raises(mocker, caplog):
|
|||||||
# An async-callable side_effect lets AsyncMock await the coroutine
|
# An async-callable side_effect lets AsyncMock await the coroutine
|
||||||
# before the exception propagates; assigning a bare exception class
|
# before the exception propagates; assigning a bare exception class
|
||||||
# leaks an un-awaited coroutine and trips RuntimeWarning at gc time.
|
# leaks an un-awaited coroutine and trips RuntimeWarning at gc time.
|
||||||
# The `await asyncio.sleep(0)` is a no-op event-loop yield that
|
# The `await anyio.sleep(0)` is a no-op event-loop yield that
|
||||||
# satisfies static analysis ("async function uses no async features")
|
# satisfies static analysis ("async function uses no async features")
|
||||||
# without changing observable behavior.
|
# without changing observable behavior.
|
||||||
async def _scroll_raises(*args, **kwargs):
|
async def _scroll_raises(*args, **kwargs):
|
||||||
await asyncio.sleep(0)
|
await anyio.sleep(0)
|
||||||
raise RuntimeError("boom")
|
raise RuntimeError("boom")
|
||||||
|
|
||||||
client.scroll.side_effect = _scroll_raises
|
client.scroll.side_effect = _scroll_raises
|
||||||
@@ -525,7 +525,7 @@ async def test_backfill_logs_warning_when_sentinel_upsert_fails(mocker, caplog):
|
|||||||
|
|
||||||
async def _upsert_raises(*args, **kwargs):
|
async def _upsert_raises(*args, **kwargs):
|
||||||
# See _scroll_raises above for why this is async + sleep(0).
|
# See _scroll_raises above for why this is async + sleep(0).
|
||||||
await asyncio.sleep(0)
|
await anyio.sleep(0)
|
||||||
raise RuntimeError("sentinel write blip")
|
raise RuntimeError("sentinel write blip")
|
||||||
|
|
||||||
client.upsert.side_effect = _upsert_raises
|
client.upsert.side_effect = _upsert_raises
|
||||||
@@ -602,7 +602,7 @@ async def test_ensure_payload_indexes_summarises_failed_fields(mocker, caplog):
|
|||||||
|
|
||||||
async def _create_index(*args, **kwargs):
|
async def _create_index(*args, **kwargs):
|
||||||
# See _scroll_raises above for why this is async + sleep(0).
|
# See _scroll_raises above for why this is async + sleep(0).
|
||||||
await asyncio.sleep(0)
|
await anyio.sleep(0)
|
||||||
call_count["n"] += 1
|
call_count["n"] += 1
|
||||||
if call_count["n"] != 2:
|
if call_count["n"] != 2:
|
||||||
raise _make_unexpected(500, b'{"status":{"error":"boom"}}')
|
raise _make_unexpected(500, b'{"status":{"error":"boom"}}')
|
||||||
|
|||||||
Reference in New Issue
Block a user