d00d436e5c
Why: - Ingested chunks need to become searchable Qdrant points before the upload response returns, with tenant/domain isolation and a safe re-ingestion story per ADR-0001/0017. Changes: - src/application/points/: index_chunks() is the sole entry point, owning payload construction, batched/bounded-concurrency upserts (upsert_concurrency semaphore), and a soft-delete sweep for points a shorter re-ingestion leaves behind. The sweep runs only after every upsert in the attempt succeeds, so a failed attempt can leave a stale prefix but never removes content from a working index. - PointStorage port (application/ports/) + QdrantPointStorage adapter (infrastructure/qdrant/points.py), keeping the qdrant_client SDK out of application code per ADR-0015. - FakePointStorage test double for exercising the ordering/idempotency guarantees without a real Qdrant.
139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
"""Hand-written fakes for narrow application-owned ports (ADR-0016)."""
|
|
|
|
import asyncio
|
|
import uuid
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
|
|
from src.application.ingestion.models import SparseVector
|
|
from src.application.points.models import ChunkPoint
|
|
|
|
|
|
@dataclass
|
|
class FakeObjectStorage:
|
|
"""In-memory `ObjectStorage`. `fail_next` simulates one upload failure."""
|
|
|
|
objects: dict[str, bytes] = field(default_factory=dict)
|
|
fail_next: bool = False
|
|
|
|
async def put_object(self, *, key: str, data: bytes, content_type: str) -> None:
|
|
if self.fail_next:
|
|
self.fail_next = False
|
|
raise OSError("simulated object storage failure")
|
|
self.objects[key] = data
|
|
|
|
|
|
@dataclass
|
|
class FakeDenseEmbedder:
|
|
"""A scripted `DenseEmbedder`. Returns a fixed-dimension zero vector per
|
|
text by default; `fail_next` simulates one batch failure.
|
|
"""
|
|
|
|
name: str
|
|
dimensions: int = 4
|
|
model_version: str = "fake-dense-v1"
|
|
calls: list[list[str]] = field(default_factory=list)
|
|
fail_next: bool = False
|
|
delay_seconds: float = 0.0
|
|
"""Simulates a slow provider call, e.g. to exercise timeout handling."""
|
|
|
|
async def embed_batch(self, texts: Sequence[str]) -> list[list[float]]:
|
|
self.calls.append(list(texts))
|
|
if self.delay_seconds:
|
|
await asyncio.sleep(self.delay_seconds)
|
|
if self.fail_next:
|
|
self.fail_next = False
|
|
raise RuntimeError("simulated embedder failure")
|
|
return [[0.0] * self.dimensions for _ in texts]
|
|
|
|
|
|
@dataclass
|
|
class FakeSparseEmbedder:
|
|
"""A scripted `SparseEmbedder`. Returns an empty sparse vector per text."""
|
|
|
|
name: str = "sparse"
|
|
model_version: str = "fake-sparse-v1"
|
|
calls: list[list[str]] = field(default_factory=list)
|
|
fail_next: bool = False
|
|
|
|
def embed_batch(self, texts: Sequence[str], *, query: bool = False) -> list[SparseVector]:
|
|
self.calls.append(list(texts))
|
|
if self.fail_next:
|
|
self.fail_next = False
|
|
raise RuntimeError("simulated embedder failure")
|
|
return [SparseVector(indices=[], values=[]) for _ in texts]
|
|
|
|
|
|
@dataclass
|
|
class FakePointStorage:
|
|
"""An in-memory `PointStorage`.
|
|
|
|
`points` is keyed by point id, so a re-upsert of the same deterministic id
|
|
overwrites rather than accumulating — the property a test asserting "a
|
|
retry produces no duplicate points" needs the fake to actually model.
|
|
|
|
`fail_on_batch` fails the Nth (0-based) upsert batch, which is how a test
|
|
checks that the soft-delete sweep never runs after a partial failure.
|
|
"""
|
|
|
|
points: dict[str, ChunkPoint] = field(default_factory=dict)
|
|
upsert_batches: list[int] = field(default_factory=list)
|
|
deactivate_calls: list[dict[str, object]] = field(default_factory=list)
|
|
fail_on_batch: int | None = None
|
|
fail_deactivate: bool = False
|
|
max_in_flight: int = 0
|
|
_in_flight: int = 0
|
|
|
|
async def upsert_points(self, points: Sequence[ChunkPoint]) -> None:
|
|
self._in_flight += 1
|
|
self.max_in_flight = max(self.max_in_flight, self._in_flight)
|
|
try:
|
|
# Yield so concurrent batches actually overlap; without this the
|
|
# in-flight ceiling is trivially 1 and the bound goes untested.
|
|
await asyncio.sleep(0)
|
|
index = len(self.upsert_batches)
|
|
self.upsert_batches.append(len(points))
|
|
if self.fail_on_batch is not None and index == self.fail_on_batch:
|
|
raise RuntimeError("simulated point storage failure")
|
|
for point in points:
|
|
self.points[str(point.point_id)] = point
|
|
finally:
|
|
self._in_flight -= 1
|
|
|
|
async def deactivate_points_from_index(
|
|
self,
|
|
*,
|
|
tenant_id: uuid.UUID,
|
|
file_id: uuid.UUID,
|
|
from_chunk_index: int,
|
|
deleted_at: datetime,
|
|
updated_by: str,
|
|
) -> int:
|
|
self.deactivate_calls.append(
|
|
{
|
|
"tenant_id": tenant_id,
|
|
"file_id": file_id,
|
|
"from_chunk_index": from_chunk_index,
|
|
"updated_by": updated_by,
|
|
}
|
|
)
|
|
if self.fail_deactivate:
|
|
raise RuntimeError("simulated deactivate failure")
|
|
|
|
def is_stale(point: ChunkPoint) -> bool:
|
|
chunk_index = point.payload.get("chunk_index")
|
|
return (
|
|
point.payload.get("file_id") == str(file_id)
|
|
and point.payload.get("tenant_id") == str(tenant_id)
|
|
and point.payload.get("is_active") is True
|
|
and isinstance(chunk_index, int)
|
|
and chunk_index >= from_chunk_index
|
|
)
|
|
|
|
stale = [point for point in self.points.values() if is_stale(point)]
|
|
for point in stale:
|
|
point.payload["is_active"] = False
|
|
point.payload["deleted_at"] = deleted_at.isoformat()
|
|
return len(stale)
|