From cc915f0f1a22b0a06363ddec6628bb2e9444482e Mon Sep 17 00:00:00 2001 From: Ali Zarinkolah Date: Thu, 20 Aug 2026 18:17:39 +0330 Subject: [PATCH] feat(ingestion): index embedded chunks into Qdrant on upload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: - POST /v1/files was reporting chunks_indexed=0/points_created=0 unconditionally — chunks were parsed and embedded but never written to Qdrant, so nothing was actually searchable after upload. Changes: - upload_source_file() now calls index_chunks() after embedding, inside the same INGESTION_TIMEOUT_SECONDS window, and marks the job failed (error_code=index_failed, 502) if it raises. - Job counters (points_created, points_soft_deleted) and the response's chunks_indexed now reflect the real indexing result instead of a hardcoded zero. - Wired PointStorage through AppResources/lifespan/the files router. Impact: - A successful upload is now searchable in Qdrant by the time 201 returns. --- CLAUDE.md | 22 ++- README.md | 17 ++ ...nchronous-ingestion-in-the-request-path.md | 21 ++- docs/plans/001-ingestion-vertical-slice.md | 14 +- src/api/errors.py | 2 + src/api/routers/files.py | 6 + src/application/files/upload.py | 57 +++++- src/application/ingestion/errors.py | 9 + src/bootstrap/dependencies.py | 6 + src/bootstrap/lifespan.py | 9 + .../postgres/test_upload_service.py | 165 +++++++++++++++++- 11 files changed, 304 insertions(+), 24 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index e848d11..0049058 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -18,10 +18,14 @@ the chunk-count ceiling (`413`). The embedding configuration is **ported from the `emet` evaluation lab** (`~/code/talie/emet`), which benchmarked these models and analyzers on the real Farsi corpus — the analyzer and BM25 weights are verified token-for-token against it, so treat them as a measured artifact -and re-benchmark rather than tune them in place (ADR-0005). Not built yet: Qdrant collection bootstrap, -Qdrant point upserts (so uploaded chunks are parsed/embedded but not yet -searchable), and `src/agent/`. That maps to plan 001 Phases 1-4 done, Phase 5 -not started. +and re-benchmark rather than tune them in place (ADR-0005). Also working: the +`chunks` collection bootstrap (`src/infrastructure/qdrant/collection.py`, run as +a deployment step via `uv run python -m src.cli.qdrant_bootstrap` — never at +startup) and tenant-scoped point upserts (`src/application/points/` behind the +`PointStorage` port), so an upload is searchable by the time `201` returns. Not +built yet: `/v1/points` CRUD and keyword search (plan 002), `tenant_domains` +validation of the `domain` field, and `src/agent/`. That maps to plan 001 +Phases 1-5 done. Architecture decisions live in `docs/adr/` (18 ADRs plus the 0000 template; 0001–0004 are `Accepted` — 0004 amended by 0018; 0014 is `Superseded by 0017`; @@ -140,9 +144,13 @@ as its only caller-facing entry point. It dispatches on source type and owns the `anyio.to_thread.run_sync` + `CapacityLimiter` offload ADR-0017 requires; `parse_docx`/`parse_csv`/`parse_xlsx`/`chunk_document` stay in the package, exported mainly for their own tests, not for outside callers to reach for -directly. Follow this pattern in `application/` as new packages are added -there — `points/`, `retrieval/`, `threads/` — rather than exposing their -internals as the primary surface. +directly. `src/application/points/` follows the same shape: `index_chunks` is +the only caller-facing entry point, owning payload construction, batching, +the `upsert_concurrency` semaphore, and the ordering rule that the soft-delete +sweep runs only after every upsert succeeds; `build_chunk_payload` stays +internal. Follow this pattern in `application/` as new packages are added +there — `retrieval/`, `threads/` — rather than exposing their internals as the +primary surface. ### Resource lifetime rules (ADR-0012) diff --git a/README.md b/README.md index f1ee9d9..9a8938b 100644 --- a/README.md +++ b/README.md @@ -3,6 +3,23 @@ Architecture decisions live in [`docs/adr`](docs/adr). The first implementation milestone is documented in the [ingestion vertical-slice plan](docs/plans/001-ingestion-vertical-slice.md). +## Provisioning the datastores + +Both schema steps run as explicit deployment steps. The application performs no +DDL at startup — not for Postgres (ADR-0009) and not for Qdrant (ADR-0001, +"Collection provisioning"). + +```bash +docker compose up -d # Postgres, MinIO, Qdrant +uv run alembic upgrade head # Postgres schema +uv run python -m src.cli.qdrant_bootstrap # the `chunks` collection +uv run fastapi dev src/main.py +``` + +Both commands are idempotent and safe to re-run. `qdrant_bootstrap` verifies an +existing collection against the pinned schema and exits non-zero on a mismatch, +rather than leaving a silently degraded sparse index in place. + ## Local Langfuse This repo includes a root-level development Compose file for Langfuse: diff --git a/docs/adr/0017-synchronous-ingestion-in-the-request-path.md b/docs/adr/0017-synchronous-ingestion-in-the-request-path.md index ef59aba..2c831c3 100644 --- a/docs/adr/0017-synchronous-ingestion-in-the-request-path.md +++ b/docs/adr/0017-synchronous-ingestion-in-the-request-path.md @@ -159,8 +159,25 @@ retry, and phase 2 has no transaction protecting it: return the existing file/job rather than re-ingesting (plan 001). - `tenant_id` comes from `AuthContext`, never from the request body. - A terminal job is never transitioned back to `running`. -- Qdrant points from a failed attempt do not replace the previous successful - index; replacement happens only after a successful attempt. +- A failed attempt never *removes* content from a working index. The + soft-delete sweep that retires a shortened file's leftover points runs only + after every upsert in the attempt has succeeded. + + This is deliberately weaker than "replacement happens only after a successful + attempt", which an earlier revision of this ADR claimed. That guarantee is not + achievable alongside ADR-0001's deterministic point ids: those ids are exactly + what makes a retry idempotent, and they also mean a re-ingestion overwrites + points **in place**, so a crash partway through leaves a prefix updated and the + remainder still on the old content. Buying literal atomicity would mean + generation-suffixed ids and an activation flip, which contradicts ADR-0001 and + ADR-0002's stable point ids. Staging the new points as `is_active=false` and + flipping them on success is strictly worse — the in-place overwrite would + deactivate the previously live points, silently emptying a working index if the + attempt were interrupted. + + What holds instead: the index is never emptied, never partially deleted, and a + retry converges — deterministic ids rewrite every point and the sweep re-runs, + reaching the exact correct state. ### Failures are HTTP failures diff --git a/docs/plans/001-ingestion-vertical-slice.md b/docs/plans/001-ingestion-vertical-slice.md index 1b2ba3b..4c6e387 100644 --- a/docs/plans/001-ingestion-vertical-slice.md +++ b/docs/plans/001-ingestion-vertical-slice.md @@ -125,9 +125,12 @@ them: - Use `(tenant_id, domain, content_sha256)` to recognize identical uploads. - An identical active upload should return the existing source-file/job reference rather than create a duplicate ingestion. -- A changed upload creates a new ingestion job. Existing active Qdrant points are - replaced only after the new job completes successfully, so a failed re-ingestion - does not remove a working index. +- A changed upload creates a new ingestion job. A failed re-ingestion never + removes a working index: the soft-delete sweep for a shortened file runs only + after every upsert has succeeded. Because ADR-0001's point ids are + deterministic, upserts overwrite in place, so an interrupted attempt can leave + a prefix updated — it cannot empty or partially delete the index, and a retry + converges. See ADR-0017, "Re-running an ingestion stays safe". - Preserve the original filename in Postgres metadata. MinIO object keys remain internal ID-based paths. @@ -272,6 +275,11 @@ code and a terminal job row. tenant-filtered upserts, terminal state persistence, retrying an upload, and parser/Qdrant failure handling. +The `chunks` collection itself is provisioned by a deployment step — +`uv run python -m src.cli.qdrant_bootstrap` — not by FastAPI startup, for the +same reason ADR-0009 keeps Alembic out of startup and ADR-0012 makes LangGraph's +`.setup()` a deployment step. See ADR-0001, "Collection provisioning". + **Exit criteria:** a successful upload returns `201` with a terminal status, and its points are retrievable only under the owning tenant's Qdrant filter. A forced failure mid-ingestion produces a `failed` job and the right HTTP status, and diff --git a/src/api/errors.py b/src/api/errors.py index a261b71..9832671 100644 --- a/src/api/errors.py +++ b/src/api/errors.py @@ -23,6 +23,7 @@ from src.application.ingestion.errors import ( EmbedderError, IngestionAtCapacityError, IngestionTimeoutError, + PointIndexingError, UnsupportedSourceTypeError, ) @@ -44,6 +45,7 @@ _MAPPING: tuple[tuple[type[Exception], int, str], ...] = ( (FileTooLargeError, status.HTTP_413_CONTENT_TOO_LARGE, "payload_too_large"), (ChunkLimitExceededError, status.HTTP_413_CONTENT_TOO_LARGE, "payload_too_large"), (EmbedderError, status.HTTP_502_BAD_GATEWAY, "embedder_error"), + (PointIndexingError, status.HTTP_502_BAD_GATEWAY, "index_error"), (IngestionTimeoutError, status.HTTP_504_GATEWAY_TIMEOUT, "ingestion_timeout"), ) diff --git a/src/api/routers/files.py b/src/api/routers/files.py index 13e2ecf..4f6e971 100644 --- a/src/api/routers/files.py +++ b/src/api/routers/files.py @@ -19,11 +19,13 @@ from src.application.files.status import get_file_status from src.application.files.upload import upload_source_file from src.application.ports.embedding import DenseEmbedder, SparseEmbedder from src.application.ports.object_storage import ObjectStorage +from src.application.ports.point_storage import PointStorage from src.bootstrap.dependencies import ( get_dense_embedders, get_ingestion_concurrency_limiter, get_ingestion_limiter, get_object_storage, + get_point_storage, get_sessionmaker, get_settings, get_sparse_embedder, @@ -35,6 +37,7 @@ router = APIRouter(prefix="/files", tags=["files"]) _RequireFilesWrite = Annotated[AuthContext, Depends(require_scope("files:write"))] _SessionmakerDep = Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)] _ObjectStorageDep = Annotated[ObjectStorage, Depends(get_object_storage)] +_PointStorageDep = Annotated[PointStorage, Depends(get_point_storage)] _SettingsDep = Annotated[Settings, Depends(get_settings)] _IngestionLimiterDep = Annotated[CapacityLimiter, Depends(get_ingestion_limiter)] _ConcurrencyLimiterDep = Annotated[Semaphore, Depends(get_ingestion_concurrency_limiter)] @@ -50,6 +53,7 @@ async def upload_file( auth: _RequireFilesWrite, sessionmaker: _SessionmakerDep, storage: _ObjectStorageDep, + point_storage: _PointStorageDep, settings: _SettingsDep, limiter: _IngestionLimiterDep, concurrency_limiter: _ConcurrencyLimiterDep, @@ -60,12 +64,14 @@ async def upload_file( result = await upload_source_file( sessionmaker=sessionmaker, storage=storage, + point_storage=point_storage, auth=auth, domain=domain, filename=file.filename or "", data=data, ingestion_settings=settings.ingestion, chunking_settings=settings.chunking, + qdrant_settings=settings.qdrant, thread_limiter=limiter, concurrency_limiter=concurrency_limiter, dense_embedders=dense_embedders, diff --git a/src/application/files/upload.py b/src/application/files/upload.py index 2efcd4d..36f59cd 100644 --- a/src/application/files/upload.py +++ b/src/application/files/upload.py @@ -16,9 +16,10 @@ job stuck in `running`. The whole request additionally holds one of phase 2 is bounded by `INGESTION_TIMEOUT_SECONDS` (`504`) (ADR-0017, plan 001 Phase 4). -Qdrant point upserts are Phase 5 work, not implemented here: this phase -parses, chunks, and embeds, so a successful job still reports -`chunks_indexed=0` — nothing is searchable yet. +Phase 2 ends by upserting the embedded chunks as tenant-scoped Qdrant points +(`src/application/points/`), so a successful upload is searchable by the time +the `201` returns. The collection those points land in is provisioned by a +deployment step, not by this path — see `src/cli/qdrant_bootstrap.py`. """ import uuid @@ -45,10 +46,13 @@ from src.application.ingestion.errors import ( ChunkLimitExceededError, EmbedderError, IngestionTimeoutError, + PointIndexingError, ) +from src.application.points import index_chunks from src.application.ports.embedding import DenseEmbedder, SparseEmbedder from src.application.ports.object_storage import ObjectStorage -from src.config import ChunkingSettings, IngestionSettings +from src.application.ports.point_storage import PointStorage +from src.config import ChunkingSettings, IngestionSettings, QdrantSettings from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo from src.infrastructure.postgres.repositories import source_files as source_files_repo @@ -88,12 +92,14 @@ async def upload_source_file( *, sessionmaker: async_sessionmaker[AsyncSession], storage: ObjectStorage, + point_storage: PointStorage, auth: AuthContext, domain: str, filename: str, data: bytes, ingestion_settings: IngestionSettings, chunking_settings: ChunkingSettings, + qdrant_settings: QdrantSettings, thread_limiter: CapacityLimiter, concurrency_limiter: Semaphore, dense_embedders: Sequence[DenseEmbedder], @@ -249,6 +255,31 @@ async def upload_source_file( error_message=str(exc), ) raise + + try: + indexed = await index_chunks( + embedded, + storage=point_storage, + tenant_id=auth.tenant_id, + domain=domain, + file_id=source_file_id, + source_filename=filename, + source_type=validated.source_type, + actor=f"api_key:{auth.api_key_id}", + dense_embedders=dense_embedders, + sparse_embedder=sparse_embedder, + settings=qdrant_settings, + thread_limiter=thread_limiter, + ) + except PointIndexingError as exc: + await _mark_job_failed( + sessionmaker, + tenant_id=auth.tenant_id, + ingestion_job_id=ingestion_job_id, + error_code="index_failed", + error_message=str(exc), + ) + raise except TimeoutError: logger.warning( "files.upload.timeout", @@ -273,7 +304,11 @@ async def upload_source_file( tenant_id=auth.tenant_id, ingestion_job_id=ingestion_job_id, status="succeeded", - points_created=0, + # An upsert with deterministic ids cannot tell an insert from + # an overwrite, so every written point is reported here and + # `points_updated` stays 0 rather than being guessed at. + points_created=indexed.points_upserted, + points_soft_deleted=indexed.points_soft_deleted, ) jobs_repo.append_event( session, @@ -281,8 +316,13 @@ async def upload_source_file( ingestion_job_id=ingestion_job_id, level="info", stage="completed", - message="chunks parsed and embedded; Qdrant indexing not yet implemented", - details={"chunks_parsed": len(chunks), "chunks_embedded": len(embedded)}, + message="chunks parsed, embedded, and indexed", + details={ + "chunks_parsed": len(chunks), + "chunks_embedded": len(embedded), + "points_upserted": indexed.points_upserted, + "points_soft_deleted": indexed.points_soft_deleted, + }, ) await session.commit() @@ -291,11 +331,12 @@ async def upload_source_file( tenant_id=str(auth.tenant_id), file_id=str(source_file_id), ingestion_job_id=str(ingestion_job_id), + points_indexed=indexed.points_upserted, ) return UploadResult( file_id=source_file_id, ingestion_job_id=ingestion_job_id, status="succeeded", - chunks_indexed=0, + chunks_indexed=indexed.points_upserted, is_new_attempt=True, ) diff --git a/src/application/ingestion/errors.py b/src/application/ingestion/errors.py index 421b875..835921b 100644 --- a/src/application/ingestion/errors.py +++ b/src/application/ingestion/errors.py @@ -49,6 +49,15 @@ class EmbedderError(IngestionError): """ +class PointIndexingError(IngestionError): + """Upserting or soft-deleting Qdrant points failed. + + Maps to `502` — like `EmbedderError`, this is an upstream dependency + failing, not a malformed request. Kept distinct from `EmbedderError` so the + job's `error_code` says which dependency broke. + """ + + class IngestionAtCapacityError(IngestionError): """`INGESTION_MAX_CONCURRENCY` in-process ingestions are already running. diff --git a/src/bootstrap/dependencies.py b/src/bootstrap/dependencies.py index 36e5891..b1a2ee0 100644 --- a/src/bootstrap/dependencies.py +++ b/src/bootstrap/dependencies.py @@ -9,6 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker from src.application.ports.embedding import DenseEmbedder, SparseEmbedder from src.application.ports.object_storage import ObjectStorage +from src.application.ports.point_storage import PointStorage from src.config import Settings @@ -20,6 +21,7 @@ class AppResources: minio_client: Minio qdrant_client: AsyncQdrantClient object_storage: ObjectStorage + point_storage: PointStorage ingestion_limiter: CapacityLimiter dense_embedders: Sequence[DenseEmbedder] sparse_embedder: SparseEmbedder @@ -46,6 +48,10 @@ def get_object_storage(request: Request) -> ObjectStorage: return _resources(request).object_storage +def get_point_storage(request: Request) -> PointStorage: + return _resources(request).point_storage + + def get_ingestion_limiter(request: Request) -> CapacityLimiter: return _resources(request).ingestion_limiter diff --git a/src/bootstrap/lifespan.py b/src/bootstrap/lifespan.py index 6b40fc8..0e21a32 100644 --- a/src/bootstrap/lifespan.py +++ b/src/bootstrap/lifespan.py @@ -20,6 +20,7 @@ from src.infrastructure.minio.storage import MinioObjectStorage from src.infrastructure.observability.logging import configure_logging from src.infrastructure.postgres.database import create_engine, create_sessionmaker from src.infrastructure.qdrant.client import create_client as create_qdrant_client +from src.infrastructure.qdrant.points import QdrantPointStorage logger = structlog.get_logger(__name__) @@ -77,6 +78,13 @@ def create_lifespan( logger.info("lifespan.minio.client.created") qdrant_client = create_qdrant_client(resolved_settings.qdrant) + # No collection DDL here: `ensure_chunks_collection` is a deployment + # step (`python -m src.cli.qdrant_bootstrap`), for the same reason + # ADR-0009 keeps Alembic out of startup and ADR-0012 makes LangGraph's + # `.setup()` a deployment step. + point_storage = QdrantPointStorage( + qdrant_client, collection=resolved_settings.qdrant.collection + ) logger.info("lifespan.qdrant.client.created") nomic_settings = resolved_settings.embedding.nomic @@ -136,6 +144,7 @@ def create_lifespan( minio_client=minio_client, qdrant_client=qdrant_client, object_storage=object_storage, + point_storage=point_storage, ingestion_limiter=ingestion_limiter, dense_embedders=dense_embedders, sparse_embedder=sparse_embedder, diff --git a/tests/integration/postgres/test_upload_service.py b/tests/integration/postgres/test_upload_service.py index c2543bb..64b982c 100644 --- a/tests/integration/postgres/test_upload_service.py +++ b/tests/integration/postgres/test_upload_service.py @@ -13,11 +13,21 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.application.auth.context import AuthContext from src.application.files.models import UploadResult from src.application.files.upload import upload_source_file -from src.application.ingestion.errors import IngestionAtCapacityError, IngestionTimeoutError +from src.application.ingestion.errors import ( + IngestionAtCapacityError, + IngestionTimeoutError, + PointIndexingError, +) from src.application.ports.object_storage import ObjectStorage -from src.config import ChunkingSettings, IngestionSettings +from src.application.ports.point_storage import PointStorage +from src.config import ChunkingSettings, IngestionSettings, QdrantSettings from src.infrastructure.postgres.models.ingestion_job import IngestionJob -from tests.fakes import FakeDenseEmbedder, FakeObjectStorage, FakeSparseEmbedder +from tests.fakes import ( + FakeDenseEmbedder, + FakeObjectStorage, + FakePointStorage, + FakeSparseEmbedder, +) from tests.support.factories import create_api_key, create_tenant pytestmark = [ @@ -34,6 +44,7 @@ async def _upload( sessionmaker: async_sessionmaker[AsyncSession], storage: ObjectStorage, auth: AuthContext, + point_storage: PointStorage | None = None, domain: str = "general", filename: str = "report.csv", data: bytes = _CSV_BYTES, @@ -41,12 +52,14 @@ async def _upload( return await upload_source_file( sessionmaker=sessionmaker, storage=storage, + point_storage=point_storage if point_storage is not None else FakePointStorage(), auth=auth, domain=domain, filename=filename, data=data, ingestion_settings=IngestionSettings(), chunking_settings=ChunkingSettings(), + qdrant_settings=QdrantSettings(), thread_limiter=CapacityLimiter(2), concurrency_limiter=Semaphore(2), dense_embedders=[ @@ -83,7 +96,7 @@ async def test_upload_source_file_commits_running_job_before_storage_write( result = await _upload(sessionmaker=db_sessionmaker, storage=storage, auth=auth) assert result.status == "succeeded" - assert result.chunks_indexed == 0 + assert result.chunks_indexed == 1 assert result.is_new_attempt async with db_sessionmaker() as verify_session: @@ -170,12 +183,14 @@ async def test_upload_source_file_timeout_writes_failed_job_and_raises( await upload_source_file( sessionmaker=db_sessionmaker, storage=FakeObjectStorage(), + point_storage=FakePointStorage(), auth=auth, domain="general", filename="report.csv", data=_CSV_BYTES, ingestion_settings=IngestionSettings(timeout_seconds=0.05), chunking_settings=ChunkingSettings(), + qdrant_settings=QdrantSettings(), thread_limiter=CapacityLimiter(2), concurrency_limiter=Semaphore(2), dense_embedders=[slow_embedder, FakeDenseEmbedder(name="dense_openai")], @@ -212,12 +227,14 @@ async def test_upload_source_file_at_capacity_rejects_before_any_job_row( await upload_source_file( sessionmaker=db_sessionmaker, storage=FakeObjectStorage(), + point_storage=FakePointStorage(), auth=auth, domain="general", filename="report.csv", data=_CSV_BYTES, ingestion_settings=IngestionSettings(), chunking_settings=ChunkingSettings(), + qdrant_settings=QdrantSettings(), thread_limiter=CapacityLimiter(2), concurrency_limiter=concurrency_limiter, dense_embedders=[ @@ -238,3 +255,143 @@ async def test_upload_source_file_at_capacity_rejects_before_any_job_row( .all() ) assert jobs == [] + + +async def test_upload_source_file_indexes_points_and_records_real_counters( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + """The Phase 5 payoff: a successful upload is searchable, and the job row + says how many points it wrote rather than a hardcoded zero. + """ + auth = await _auth_for(db_session) + point_storage = FakePointStorage() + + result = await _upload( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth, + point_storage=point_storage, + data=b"name,value\nfirst,1\nsecond,2\nthird,3\n", + ) + + assert result.status == "succeeded" + assert result.chunks_indexed == len(point_storage.points) + assert result.chunks_indexed > 0 + + async with db_sessionmaker() as verify_session: + job = await verify_session.get(IngestionJob, result.ingestion_job_id) + assert job is not None + assert job.points_created == result.chunks_indexed + # An upsert cannot tell an insert from an overwrite, so this stays 0. + assert job.points_updated == 0 + + +async def test_upload_source_file_indexes_points_under_the_authenticated_tenant( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + auth = await _auth_for(db_session) + point_storage = FakePointStorage() + + result = await _upload( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth, + point_storage=point_storage, + domain="fire", + ) + + payloads = [point.payload for point in point_storage.points.values()] + assert payloads + for payload in payloads: + assert payload["tenant_id"] == str(auth.tenant_id) + assert payload["domain"] == "fire" + assert payload["file_id"] == str(result.file_id) + assert payload["created_by"] == f"api_key:{auth.api_key_id}" + + +async def test_upload_source_file_index_failure_writes_terminal_failed_job( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + auth = await _auth_for(db_session) + point_storage = FakePointStorage(fail_on_batch=0) + + with pytest.raises(PointIndexingError): + await _upload( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth, + point_storage=point_storage, + ) + + async with db_sessionmaker() as verify_session: + jobs = (await verify_session.execute(select(IngestionJob))).scalars().all() + job = next(job for job in jobs if job.tenant_id == auth.tenant_id) + assert job.status == "failed" + assert job.error_code == "index_failed" + assert job.completed_at is not None + + +async def test_upload_source_file_failed_index_does_not_soft_delete_existing_points( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + """A failed attempt must never remove content from a working index.""" + auth = await _auth_for(db_session) + point_storage = FakePointStorage() + await _upload( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth, + point_storage=point_storage, + data=b"name,value\nfirst,1\nsecond,2\n", + ) + # fail_on_batch indexes into upsert_batches, which accumulates across + # uploads -- reset it so "batch 0" means the retry's first batch. + point_storage.deactivate_calls.clear() + point_storage.upsert_batches.clear() + point_storage.fail_on_batch = 0 + + with pytest.raises(PointIndexingError): + await _upload( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth, + point_storage=point_storage, + data=b"name,value\nonly,1\n", + ) + + assert point_storage.deactivate_calls == [] + assert all(point.payload["is_active"] is True for point in point_storage.points.values()) + + +async def test_upload_source_file_retry_after_index_failure_produces_no_duplicate_points( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + auth = await _auth_for(db_session) + point_storage = FakePointStorage(fail_on_batch=0) + + with pytest.raises(PointIndexingError): + await _upload( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth, + point_storage=point_storage, + ) + + point_storage.fail_on_batch = None + retry = await _upload( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth, + point_storage=point_storage, + ) + + assert retry.status == "succeeded" + assert len(point_storage.points) == retry.chunks_indexed + + async with db_sessionmaker() as verify_session: + jobs = (await verify_session.execute(select(IngestionJob))).scalars().all() + tenant_jobs = [job for job in jobs if job.tenant_id == auth.tenant_id] + # A terminal job never returns to `running` (ADR-0017); the retry is a new row. + assert len(tenant_jobs) == 2 + assert {job.status for job in tenant_jobs} == {"failed", "succeeded"} +