Compare commits

...

3 Commits

Author SHA1 Message Date
0b1932f716 docs(claude): record plan 002 phase 3 in the project status
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-22 17:13:05 +03:30
73bdac0da2 test(points): cover soft delete, relinking, and the file sweep
Why:
- The failure modes worth testing are races, and they are cheap to force
  against the fake and expensive to observe anywhere else.

Changes:
- Unit tests for both boundaries, the repeat delete, a missing neighbour, the
  traversal property after several deletes, and — via a repository that bumps a
  rival's version before each apply — both the partial-apply repair and the
  unconvergent 409.
- One new shared contract scenario (a multi-point batch applies every patch,
  including nulling a pointer) so it runs against the fake and real Qdrant.
- HTTP tests against real Postgres and Qdrant for relinking, soft-not-hard
  delete, cross-tenant 404, and scope enforcement on both routes.
- create_source_file factory for tests addressing a file without uploading.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-22 17:12:56 +03:30
b58f4630f3 feat(points): add soft delete with neighbour relinking
Why:
- Plan 002 Phase 3. ADR-0002 makes delete soft by default and treats a partial
  neighbour relink as a defect, since ADR-0003's context-window expansion walks
  the previous/next pointer chain.

Changes:
- relinking.py computes the patches still missing between the state just read
  and the desired end state, so a normal delete, a repeat delete, and recovery
  from a half-applied batch are one path.
- deletion.py re-plans and re-applies up to three times, verifying by read-back,
  because Qdrant has no multi-point transaction and reports success for a
  filtered set_payload that matched nothing; exhausting the retries raises
  PointVersionConflictError (409).
- DELETE /v1/points/{point_id} and DELETE /v1/files/{file_id}, both on
  points:write. The file route sweeps points first, then marks the source_files
  row soft_deleted in its own short transaction.
- Log events carry ADR-0011 duration_ms plus rounds.

Impact:
- A deleted file's source_files row leaves 'active', so re-uploading the same
  bytes now re-ingests instead of matching the duplicate path.
- No migration; no point is ever removed from Qdrant.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-22 17:12:43 +03:30
16 changed files with 1384 additions and 11 deletions

View File

@@ -37,7 +37,7 @@ driving `tests/e2e/test_compose_smoke.py` against a real uvicorn process, which
skips itself unless `SMOKE_BASE_URL` is set. That maps to plan 001 Phases 1-6 skips itself unless `SMOKE_BASE_URL` is set. That maps to plan 001 Phases 1-6
done. done.
Plan 002 (`/v1/points` CRUD and keyword search) is **Phases 1-2 done**. Phase 1 Plan 002 (`/v1/points` CRUD and keyword search) is **Phases 1-3 done**. Phase 1
landed the `PointRepository` port (`src/application/ports/point_repository.py`) landed the `PointRepository` port (`src/application/ports/point_repository.py`)
with its `Point` read model (`src/application/points/point.py`), the Qdrant adapter with its `Point` read model (`src/application/points/point.py`), the Qdrant adapter
(`src/infrastructure/qdrant/point_repository.py`), request/response schemas (`src/infrastructure/qdrant/point_repository.py`), request/response schemas
@@ -75,15 +75,59 @@ with `normalize_persian_text` before matching, because ingestion letter-folds
content and an unfolded Arabic-keyboard query would return an empty result set content and an unfolded Arabic-keyboard query would return an empty result set
silently rather than erroring (ADR-0002). silently rather than erroring (ADR-0002).
Phase 3 added **soft delete**: `DELETE /v1/points/{point_id}` and
`DELETE /v1/files/{file_id}`, over `src/application/points/deletion.py` (with
the pure relinking primitive in `src/application/points/relinking.py`) and
`src/application/files/deletion.py`. Both are gated on `points:write` — the
file route included, since the data it destroys is points. Nothing is ever
removed from Qdrant.
Four rules there are load-bearing, and three of them look like complications
until the concurrency is taken seriously:
- `patches_for_removal` computes **what is still missing between the state just
read and the desired end state**, not "the patches a delete implies". That is
what makes a normal delete, a second delete of an already-inactive point (a
no-op success, never `404`), and recovery from a half-applied batch one code
path. Rewriting it as a straight-line "deactivate, patch prev, patch next"
breaks all three.
- Qdrant has no multi-point transaction and reports success for a filtered
`set_payload` that matched nothing, so a batch whose second operation loses a
version race applies its first anyway. `soft_delete_point` therefore re-plans
and re-applies up to three times, verifying by read-back, and only then raises
`PointVersionConflictError` (`409`). A single-shot delete would be able to
leave a stale pointer, which ADR-0002 calls a defect.
- A soft-deleted point **keeps its own** `previous_chunk_id`/`next_chunk_id`;
only the surviving neighbours are rewritten. Those pointers are unreachable
rather than stale, they are the only record of where the point sat, and the
retry re-plans from them. The whole-file sweep follows from the same rule:
every point leaves at once, so no survivor can dangle and no pointer is
touched at all.
- `DELETE /v1/files/{file_id}` marks the `source_files` row `soft_deleted`
**after** the point sweep, in its own short transaction (no session is held
across the Qdrant work). Order matters: a half-finished sweep leaves the row
`active` and a retried `DELETE` finishes it, and retiring the row is what
makes a later re-upload of the same bytes re-ingest instead of matching
`find_active_by_content_hash` and returning a file whose points are gone.
Audit rows are still Phase 4/6 work; Phase 3 emits log events only
(`points.soft_deleted`, `files.soft_deleted`, `points.relink.neighbour_missing`,
and the two `*.conflict` warnings). The completion and conflict events carry
ADR-0011's `duration_ms` plus `rounds`, and the pair is what makes them
diagnostic: relinking itself is O(1) (that is what the adjacency pointers buy),
so a single-point delete costs a fixed ~5 Qdrant round trips and a `rounds`
above 1 means contention, not a slow store. The whole-file sweep is the one
whose cost scales — two round trips per 100-point page.
Also worth knowing before touching the points tests: `tests/support/point_contract.py` Also worth knowing before touching the points tests: `tests/support/point_contract.py`
holds **one** scenario suite run against both `FakePointRepository` (unit) and holds **one** scenario suite run against both `FakePointRepository` (unit) and
`QdrantPointRepository` (integration), so new repository behaviour belongs there `QdrantPointRepository` (integration), so new repository behaviour belongs there
rather than in one of the two runners — that is what keeps the fake from drifting rather than in one of the two runners — that is what keeps the fake from drifting
more permissive than the real store. more permissive than the real store.
Not built yet: plan 002 Phases 3-6 — soft delete with neighbour relinking, Not built yet: plan 002 Phases 4-6 — create/replace/patch, reorder and batch,
create/replace/patch, reorder and batch, and the the `api_request_logs`/`point_audit_events` tables, and the runbook section on
`api_request_logs`/`point_audit_events` tables — and `src/agent/`. inspecting and repairing a file's pointer chain — and `src/agent/`.
Architecture decisions live in `docs/adr/` (18 ADRs plus the 0000 template; 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`; 0001–0004 are `Accepted` — 0004 amended by 0018; 0014 is `Superseded by 0017`;

View File

@@ -17,7 +17,11 @@ from src.application.auth.errors import (
TenantInactiveError, TenantInactiveError,
) )
from src.application.domains.errors import DomainAlreadyExistsError, UnknownDomainError from src.application.domains.errors import DomainAlreadyExistsError, UnknownDomainError
from src.application.files.errors import FileTooLargeError, InvalidUploadError from src.application.files.errors import (
FileTooLargeError,
InvalidUploadError,
SourceFileNotFoundError,
)
from src.application.ingestion.errors import ( from src.application.ingestion.errors import (
ChunkLimitExceededError, ChunkLimitExceededError,
DocumentParseError, DocumentParseError,
@@ -27,6 +31,7 @@ from src.application.ingestion.errors import (
PointIndexingError, PointIndexingError,
UnsupportedSourceTypeError, UnsupportedSourceTypeError,
) )
from src.application.points.errors import PointVersionConflictError
from src.application.points.point import PointNotFoundError from src.application.points.point import PointNotFoundError
logger = structlog.get_logger(__name__) logger = structlog.get_logger(__name__)
@@ -45,6 +50,10 @@ _MAPPING: tuple[tuple[type[Exception], int, str], ...] = (
# 404, never 403: a cross-tenant point id must be indistinguishable from a # 404, never 403: a cross-tenant point id must be indistinguishable from a
# nonexistent one, or the API becomes an existence oracle (ADR-0016). # nonexistent one, or the API becomes an existence oracle (ADR-0016).
(PointNotFoundError, status.HTTP_404_NOT_FOUND, "not_found"), (PointNotFoundError, status.HTTP_404_NOT_FOUND, "not_found"),
(SourceFileNotFoundError, status.HTTP_404_NOT_FOUND, "not_found"),
# Not "the version guard fired once" — that is retried. This is the service
# giving up after repeated re-plans, i.e. a genuinely contended point.
(PointVersionConflictError, status.HTTP_409_CONFLICT, "conflict"),
(UnknownDomainError, status.HTTP_400_BAD_REQUEST, "unknown_domain"), (UnknownDomainError, status.HTTP_400_BAD_REQUEST, "unknown_domain"),
(DomainAlreadyExistsError, status.HTTP_409_CONFLICT, "conflict"), (DomainAlreadyExistsError, status.HTTP_409_CONFLICT, "conflict"),
(DocumentParseError, status.HTTP_400_BAD_REQUEST, "validation_error"), (DocumentParseError, status.HTTP_400_BAD_REQUEST, "validation_error"),

View File

@@ -1,4 +1,4 @@
"""`POST /v1/files`, `GET /v1/files/{file_id}` (ADR-0008). """`POST /v1/files`, `GET /v1/files/{file_id}`, `DELETE /v1/files/{file_id}` (ADR-0008).
Routes adapt HTTP to `application/files` calls; they do not parse, hash, Routes adapt HTTP to `application/files` calls; they do not parse, hash,
touch MinIO/Qdrant, or otherwise carry ingestion business logic (ADR-0015). touch MinIO/Qdrant, or otherwise carry ingestion business logic (ADR-0015).
@@ -13,9 +13,10 @@ from fastapi import APIRouter, Depends, Form, HTTPException, Response, UploadFil
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from src.api.dependencies.auth import require_scope from src.api.dependencies.auth import require_scope
from src.api.schemas.files import FileStatusResponse, FileUploadResponse from src.api.schemas.files import FileDeleteResponse, FileStatusResponse, FileUploadResponse
from src.api.schemas.points import DEFAULT_PAGE_SIZE, LimitQuery, PointListResponse from src.api.schemas.points import DEFAULT_PAGE_SIZE, LimitQuery, PointListResponse
from src.application.auth.context import AuthContext from src.application.auth.context import AuthContext
from src.application.files.deletion import delete_source_file
from src.application.files.status import get_file_status from src.application.files.status import get_file_status
from src.application.files.upload import upload_source_file from src.application.files.upload import upload_source_file
from src.application.points.queries import list_file_points from src.application.points.queries import list_file_points
@@ -40,6 +41,7 @@ router = APIRouter(prefix="/files", tags=["files"])
_RequireFilesWrite = Annotated[AuthContext, Depends(require_scope("files:write"))] _RequireFilesWrite = Annotated[AuthContext, Depends(require_scope("files:write"))]
_RequirePointsRead = Annotated[AuthContext, Depends(require_scope("points:read"))] _RequirePointsRead = Annotated[AuthContext, Depends(require_scope("points:read"))]
_RequirePointsWrite = Annotated[AuthContext, Depends(require_scope("points:write"))]
_SessionmakerDep = Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)] _SessionmakerDep = Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)]
_ObjectStorageDep = Annotated[ObjectStorage, Depends(get_object_storage)] _ObjectStorageDep = Annotated[ObjectStorage, Depends(get_object_storage)]
_PointStorageDep = Annotated[PointStorage, Depends(get_point_storage)] _PointStorageDep = Annotated[PointStorage, Depends(get_point_storage)]
@@ -100,6 +102,35 @@ async def get_file(
return FileStatusResponse.from_result(result) return FileStatusResponse.from_result(result)
@router.delete("/{file_id}")
async def delete_file(
file_id: uuid.UUID,
auth: _RequirePointsWrite,
sessionmaker: _SessionmakerDep,
repository: _PointRepositoryDep,
) -> FileDeleteResponse:
"""Soft-delete a file: every active point, then the `source_files` row.
Gated on `points:write` rather than `files:write` for the same reason as the
listing above — the data this destroys is points. Nothing is removed from
Qdrant (ADR-0002); the points are flagged inactive and the row is marked
`soft_deleted`, which is also what makes a later re-upload of the same bytes
ingest afresh instead of matching the duplicate path.
Deleting an already-deleted file is a success reporting `0` points.
"""
points_soft_deleted = await delete_source_file(
sessionmaker,
repository,
tenant_id=auth.tenant_id,
source_file_id=file_id,
actor=f"api_key:{auth.api_key_id}",
)
return FileDeleteResponse(
file_id=file_id, status="soft_deleted", points_soft_deleted=points_soft_deleted
)
@router.get("/{file_id}/points") @router.get("/{file_id}/points")
async def list_points_for_file( async def list_points_for_file(
file_id: uuid.UUID, file_id: uuid.UUID,

View File

@@ -1,4 +1,4 @@
"""`/v1/points` read paths (ADR-0002, ADR-0008). """`/v1/points` read and soft-delete paths (ADR-0002, ADR-0008).
Routes adapt HTTP to `application/points` calls. They build no Qdrant filters Routes adapt HTTP to `application/points` calls. They build no Qdrant filters
and hold no CRUD semantics (ADR-0015), and they never read a tenant from the and hold no CRUD semantics (ADR-0015), and they never read a tenant from the
@@ -31,6 +31,7 @@ from src.api.schemas.points import (
PointSearchResponse, PointSearchResponse,
) )
from src.application.auth.context import AuthContext from src.application.auth.context import AuthContext
from src.application.points.deletion import soft_delete_point
from src.application.points.queries import ( from src.application.points.queries import (
count_points, count_points,
get_point, get_point,
@@ -43,6 +44,7 @@ from src.bootstrap.dependencies import get_point_repository
router = APIRouter(prefix="/points", tags=["points"]) router = APIRouter(prefix="/points", tags=["points"])
_RequirePointsRead = Annotated[AuthContext, Depends(require_scope("points:read"))] _RequirePointsRead = Annotated[AuthContext, Depends(require_scope("points:read"))]
_RequirePointsWrite = Annotated[AuthContext, Depends(require_scope("points:write"))]
_PointRepositoryDep = Annotated[PointRepository, Depends(get_point_repository)] _PointRepositoryDep = Annotated[PointRepository, Depends(get_point_repository)]
@@ -132,3 +134,29 @@ async def get_tenant_point(
repository, tenant_id=auth.tenant_id, point_id=point_id, with_vectors=with_vectors repository, tenant_id=auth.tenant_id, point_id=point_id, with_vectors=with_vectors
) )
return PointResponse.from_point(point) return PointResponse.from_point(point)
@router.delete("/{point_id}")
async def delete_tenant_point(
point_id: uuid.UUID,
auth: _RequirePointsWrite,
repository: _PointRepositoryDep,
) -> PointResponse:
"""Soft-delete one point and relink its neighbours around the gap.
The point is never removed from Qdrant (ADR-0002): it is flagged
`is_active=false` with `deleted_at` set, and its old neighbours are pointed
at each other in the same batch, so context-window expansion never walks
into it.
Deleting an already-inactive point is a no-op success rather than a `404` —
the response is the point as it stands, so the resulting `version` and
`deleted_at` are visible either way.
"""
point = await soft_delete_point(
repository,
tenant_id=auth.tenant_id,
point_id=point_id,
actor=f"api_key:{auth.api_key_id}",
)
return PointResponse.from_point(point)

View File

@@ -28,6 +28,20 @@ class FileUploadResponse(BaseModel):
) )
class FileDeleteResponse(BaseModel):
"""What `DELETE /v1/files/{file_id}` did.
`points_soft_deleted` is reported rather than left implicit because the
delete is a soft one: nothing is removed from Qdrant, and the count is the
only way a caller can tell "deactivated 40 points" from "the file was
already deleted" — both of which are successes.
"""
file_id: uuid.UUID
status: str
points_soft_deleted: int
class FileStatusResponse(BaseModel): class FileStatusResponse(BaseModel):
file_id: uuid.UUID file_id: uuid.UUID
source_filename: str source_filename: str

View File

@@ -0,0 +1,81 @@
"""`DELETE /v1/files/{file_id}` — retire a file and deactivate its points.
Two stores have to agree here, and the phase boundaries are the same ones
ingestion uses (ADR-0017): a short Postgres transaction to authorize, then the
Qdrant work with **no session held**, then a short transaction to record the
outcome. Holding a session across the sweep would pin a pool connection for the
length of a multi-page delete.
The order — points first, Postgres second — is deliberate. If the sweep dies
half way, the row stays `active` and a retried `DELETE` finishes the job, since
the sweep only ever looks at points that are still active. The reverse order
would leave a row marked deleted while its points are still live and still
retrievable by the agent, which is the failure that actually matters.
"""
import uuid
from datetime import UTC, datetime
from time import perf_counter
import structlog
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from src.application.files.errors import SourceFileNotFoundError
from src.application.points.deletion import soft_delete_file_points
from src.application.ports.point_repository import PointRepository
from src.infrastructure.postgres.repositories import source_files as source_files_repo
logger = structlog.get_logger(__name__)
async def delete_source_file(
sessionmaker: async_sessionmaker[AsyncSession],
repository: PointRepository,
*,
tenant_id: uuid.UUID,
source_file_id: uuid.UUID,
actor: str,
) -> int:
"""Soft-delete a file: every active point, then the `source_files` row.
Returns how many points the sweep deactivated. Raises
`SourceFileNotFoundError` (`404`) when the file is not this tenant's — the
check happens before anything is written, so a probe for another tenant's
file id cannot deactivate a single point.
Idempotent: a second call finds no active points and a row already marked
`soft_deleted`, and returns `0`.
"""
started = perf_counter()
async with sessionmaker() as session:
source_file = await source_files_repo.get_by_id(
session, tenant_id=tenant_id, source_file_id=source_file_id
)
if source_file is None:
raise SourceFileNotFoundError(f"file {source_file_id} not found")
points_soft_deleted = await soft_delete_file_points(
repository, tenant_id=tenant_id, file_id=source_file_id, actor=actor
)
async with sessionmaker() as session:
source_file = await source_files_repo.get_by_id(
session, tenant_id=tenant_id, source_file_id=source_file_id
)
if source_file is None:
raise SourceFileNotFoundError(f"file {source_file_id} not found")
source_files_repo.mark_soft_deleted(source_file, deleted_at=datetime.now(UTC))
await session.commit()
logger.info(
"files.soft_deleted",
tenant_id=str(tenant_id),
file_id=str(source_file_id),
points_soft_deleted=points_soft_deleted,
actor=actor,
# End to end, including both Postgres transactions. Comparing it with
# the sweep's own `duration_ms` on `points.file_soft_deleted` is what
# separates a slow Qdrant from a slow database.
duration_ms=round((perf_counter() - started) * 1000, 2),
)
return points_soft_deleted

View File

@@ -15,3 +15,11 @@ class InvalidUploadError(FilesError):
class FileTooLargeError(FilesError): class FileTooLargeError(FilesError):
"""The upload exceeds `INGESTION_MAX_UPLOAD_SIZE_MB`. Maps to `413`.""" """The upload exceeds `INGESTION_MAX_UPLOAD_SIZE_MB`. Maps to `413`."""
class SourceFileNotFoundError(FilesError):
"""No such source file *within the requesting tenant*. Maps to `404`.
Same non-disclosure rule as points (ADR-0016): a cross-tenant file id and a
nonexistent one are indistinguishable to the caller, so this is never `403`.
"""

View File

@@ -5,9 +5,11 @@ dispatches payload construction, batching, bounded-concurrency upserts, and the
post-success soft-delete sweep. `build_chunk_payload` and the batching helpers post-success soft-delete sweep. `build_chunk_payload` and the batching helpers
stay internal, exported mainly for their own unit tests. stay internal, exported mainly for their own unit tests.
Direct `/v1/points` CRUD (single-point edits, reordering, keyword search) is The `/v1/points` surface lives here too, in its own modules with their own
plan 002's surface, not this package's — plan 001 scopes it to "the reusable entry points: `queries.py` for the read paths and `deletion.py` for soft delete
service layer required by ingestion". with neighbour relinking. They share this package because they share ADR-0001's
payload schema, not because they share a caller — `index_chunks` writes a whole
file at once, while those serve one admin edit at a time.
""" """
from src.application.points.indexing import IndexingResult, index_chunks from src.application.points.indexing import IndexingResult, index_chunks

View File

@@ -0,0 +1,253 @@
"""Soft delete for `/v1/points` and for a whole file's points (ADR-0002).
The caller-facing entry points are `soft_delete_point` and
`soft_delete_file_points`. Routers call these; `patches_for_removal` and the
planning helpers stay internal, because getting a delete right is exactly the
composition a caller should not have to reassemble: read the point, load its
neighbours, compute the patches still missing, send them in **one** batch,
verify they landed, and retry against fresh versions if they did not.
Why the retry exists. Qdrant has no multi-point transaction, so a batch whose
second operation loses a version race applies its first operation anyway — and
a filtered `set_payload` that matched nothing still reports success. That
combination means "did my write land?" is only answerable by reading back, and
a single-shot delete would be able to leave the deactivation applied and a
neighbour's pointer stale. Since `patches_for_removal` plans from current state
towards a fixed end state, simply re-planning emits precisely the patches that
did not land, so the loop converges instead of re-doing work. Only exhausting
the attempts raises `PointVersionConflictError` (`409`).
Deleting an already-inactive point falls out of the same machinery rather than
needing a special case: its neighbours were relinked by the first delete, so the
plan is empty and the call is a no-op success — not a `404`, and not a second
relink.
"""
import uuid
from collections.abc import Mapping, Sequence
from datetime import UTC, datetime
from time import perf_counter
import structlog
from src.application.points.errors import PointVersionConflictError
from src.application.points.point import Point, PointNotFoundError
from src.application.points.relinking import (
neighbour_ids,
patch_for_deactivation,
patches_for_removal,
)
from src.application.ports.point_repository import PayloadPatch, PointRepository
logger = structlog.get_logger(__name__)
# Three plan-apply rounds, then a final verifying plan. Each round only re-emits
# what a concurrent writer displaced, so a caller that legitimately needs more
# than this is contending on the same points continuously and deserves the
# `409` rather than an unbounded loop inside a request.
_MAX_ATTEMPTS = 3
# One sweep page. Matches ADR-0002's 100-operation batch cap, so a page of
# points is always expressible as a single `points/batch` request.
_SWEEP_BATCH_SIZE = 100
# A hard ceiling on sweep rounds, so a file being concurrently re-ingested while
# it is deleted cannot spin here for the life of the request.
_MAX_SWEEP_ROUNDS = 1_000
def _elapsed_ms(started: float) -> float:
"""Wall-clock milliseconds since `started` (ADR-0011's `duration_ms`).
Worth carrying on these events even though the relinking itself is O(1):
what a delete actually spends is Qdrant round trips, and the whole-file
sweep spends a number of them proportional to the file's length. Timing the
operation is the only way to tell a slow store from a contended one, which
`rounds` on the same event then disambiguates.
"""
return round((perf_counter() - started) * 1000, 2)
async def soft_delete_point(
repository: PointRepository,
*,
tenant_id: uuid.UUID,
point_id: uuid.UUID,
actor: str,
) -> Point:
"""Deactivate one point and relink its neighbours around the gap.
Returns the point as it now stands. Raises `PointNotFoundError` (`404`) if
it is not this tenant's — the same non-disclosure rule the read paths
follow — or `PointVersionConflictError` (`409`) if concurrent writers keep
displacing the plan.
"""
started = perf_counter()
rounds = 0
point, patches = await _plan_removal(
repository, tenant_id=tenant_id, point_id=point_id, actor=actor
)
for _ in range(_MAX_ATTEMPTS):
if not patches:
break
await repository.apply_patches(tenant_id=tenant_id, patches=patches)
rounds += 1
# The next plan doubles as verification: anything that did not land is
# still missing from the end state and comes back as a patch.
point, patches = await _plan_removal(
repository, tenant_id=tenant_id, point_id=point_id, actor=actor
)
if patches:
logger.warning(
"points.soft_delete.conflict",
tenant_id=str(tenant_id),
point_id=str(point_id),
file_id=str(point.file_id),
unsettled_points=[str(patch.point_id) for patch in patches],
rounds=rounds,
duration_ms=_elapsed_ms(started),
)
raise PointVersionConflictError(
f"point {point_id} could not be soft-deleted under concurrent modification"
)
if rounds:
logger.info(
"points.soft_deleted",
tenant_id=str(tenant_id),
point_id=str(point_id),
file_id=str(point.file_id),
version=point.version,
actor=actor,
# `rounds` is 1 unless a concurrent writer forced a re-plan, so a
# rising value here is contention, not slow relinking.
rounds=rounds,
duration_ms=_elapsed_ms(started),
)
return point
async def soft_delete_file_points(
repository: PointRepository,
*,
tenant_id: uuid.UUID,
file_id: uuid.UUID,
actor: str,
) -> int:
"""Deactivate every active point of one file, in batches.
No relinking: the whole file leaves the sequence at once, so no surviving
active point can be left pointing at a deactivated one, and the chain is
preserved intact for whoever reads the deleted file later.
Returns how many points were active when the sweep reached them. Each round
re-lists from the start rather than paging with a cursor — deactivated
points drop straight out of the default listing, so the listing itself is
the progress check, and a round that attempts the exact same ids as the one
before it made no progress and raises `PointVersionConflictError`.
"""
started = perf_counter()
swept: set[uuid.UUID] = set()
previous_attempt: frozenset[uuid.UUID] = frozenset()
rounds = 0
for _ in range(_MAX_SWEEP_ROUNDS):
page = await repository.list_by_file(
tenant_id=tenant_id, file_id=file_id, limit=_SWEEP_BATCH_SIZE
)
if not page.points:
if swept:
logger.info(
"points.file_soft_deleted",
tenant_id=str(tenant_id),
file_id=str(file_id),
points_soft_deleted=len(swept),
actor=actor,
# Two Qdrant round trips per round, so this is the delete
# path whose cost tracks the size of the file.
rounds=rounds,
duration_ms=_elapsed_ms(started),
)
return len(swept)
attempt = frozenset(point.point_id for point in page.points)
if attempt == previous_attempt:
logger.warning(
"points.file_soft_delete.conflict",
tenant_id=str(tenant_id),
file_id=str(file_id),
unsettled_points=[str(point_id) for point_id in sorted(attempt, key=str)],
rounds=rounds,
duration_ms=_elapsed_ms(started),
)
raise PointVersionConflictError(
f"file {file_id} could not be soft-deleted under concurrent modification"
)
previous_attempt = attempt
now = datetime.now(UTC)
await repository.apply_patches(
tenant_id=tenant_id,
patches=[patch_for_deactivation(point, actor=actor, now=now) for point in page.points],
)
swept |= attempt
rounds += 1
logger.warning(
"points.file_soft_delete.conflict",
tenant_id=str(tenant_id),
file_id=str(file_id),
reason="sweep_rounds_exhausted",
rounds=rounds,
duration_ms=_elapsed_ms(started),
)
raise PointVersionConflictError(f"file {file_id} still had active points after the sweep")
async def _plan_removal(
repository: PointRepository,
*,
tenant_id: uuid.UUID,
point_id: uuid.UUID,
actor: str,
) -> tuple[Point, tuple[PayloadPatch, ...]]:
point = await repository.get(tenant_id=tenant_id, point_id=point_id)
if point is None:
raise PointNotFoundError(f"point {point_id} not found")
neighbours = await _load_neighbours(repository, tenant_id=tenant_id, point=point)
_warn_on_missing_neighbours(point, neighbours, tenant_id=tenant_id)
patches = patches_for_removal(point, neighbours, actor=actor, now=datetime.now(UTC))
return point, patches
async def _load_neighbours(
repository: PointRepository, *, tenant_id: uuid.UUID, point: Point
) -> dict[uuid.UUID, Point]:
wanted: Sequence[uuid.UUID] = neighbour_ids(point)
if not wanted:
return {}
found = await repository.get_many(tenant_id=tenant_id, point_ids=wanted)
return {neighbour.point_id: neighbour for neighbour in found}
def _warn_on_missing_neighbours(
point: Point, neighbours: Mapping[uuid.UUID, Point], *, tenant_id: uuid.UUID
) -> None:
"""A pointer naming a point that is not there means the chain is already broken.
Worth a log line rather than an exception: the delete can still complete the
part of the relink that does exist, and refusing would leave the caller with
a point it cannot remove through any endpoint.
"""
missing = [pointer for pointer in neighbour_ids(point) if pointer not in neighbours]
if missing:
logger.warning(
"points.relink.neighbour_missing",
tenant_id=str(tenant_id),
point_id=str(point.point_id),
file_id=str(point.file_id),
missing_neighbours=[str(pointer) for pointer in missing],
)

View File

@@ -0,0 +1,19 @@
"""Mutation failures for the `/v1/points` write paths (ADR-0002).
No HTTP knowledge here — `src/api/errors.py` owns the status mapping. Absence
lives on `PointNotFoundError` in `point.py`, next to the model whose read paths
raise it; this module is for the failures only a *write* can produce.
"""
class PointVersionConflictError(Exception):
"""A version-guarded write could not be landed against a moving target.
Raised when the service has re-read, recomputed, and re-applied its patches
the allowed number of times and the desired state still has not settled —
something else is writing the same points concurrently. Maps to `409`.
This is not "the guard fired once": a single stale guard is expected and is
retried, because Qdrant reports success for a filtered `set_payload` that
matched nothing. It means the retries were exhausted.
"""

View File

@@ -0,0 +1,135 @@
"""Adjacency-pointer maintenance for a point leaving a file's sequence.
ADR-0001 keeps `previous_chunk_id`/`next_chunk_id` on every point so ADR-0003's
context-window expansion can walk a file in O(1) steps. ADR-0002 makes keeping
them correct an obligation of every operation that changes a point's position:
a partial relink is a defect, not a degraded-but-acceptable outcome.
The function below is the primitive that obligation reduces to. It is pure, and
it is written as **"what is still missing between the state I just read and the
state I want"** rather than "the patches a delete implies". That framing is what
makes the caller's retry loop correct: re-planning after a partial apply emits
exactly the patches that did not land, and re-planning after a completed delete
emits nothing at all. The three cases the plan calls out — a normal delete, a
second delete of an already-inactive point, and recovery from a half-applied
batch — are then one code path instead of three.
Note what is deliberately *not* patched: the departing point's own
`previous_chunk_id`/`next_chunk_id`. Nothing active points at it once its
neighbours are relinked, so those pointers are unreachable rather than stale,
and leaving them records where the point sat — which is what a later restore or
an audit reader would need. `src/application/points/deletion.py` relies on that
when it re-plans: the departing point's pointers are the only surviving record
of which two neighbours have to be joined.
"""
import uuid
from collections.abc import Mapping
from datetime import datetime
from src.application.points.point import Point
from src.application.ports.point_repository import PayloadPatch
def _optional_id(value: uuid.UUID | None) -> str | None:
return str(value) if value is not None else None
def _provenance(point: Point, *, actor: str, now: datetime) -> dict[str, object]:
"""The fields every mutation writes: who, when, and the next version.
Bumping `version` on a relinked *neighbour* is intentional. The neighbour's
payload really did change, so a concurrent editor holding the old version
must get a `409` rather than overwrite the pointer we just fixed.
"""
return {
"updated_at": now.isoformat(),
"updated_by": actor,
"version": point.version + 1,
}
def neighbour_ids(point: Point) -> tuple[uuid.UUID, ...]:
"""The ids `patches_for_removal` needs loaded, skipping the nulls."""
return tuple(
pointer for pointer in (point.previous_chunk_id, point.next_chunk_id) if pointer is not None
)
def patches_for_removal(
point: Point,
neighbours: Mapping[uuid.UUID, Point],
*,
actor: str,
now: datetime,
) -> tuple[PayloadPatch, ...]:
"""The patches still needed to remove `point` from its file's sequence.
Returns an empty tuple when the removal is already complete, which the
caller reads as both "converged" and "this was a no-op".
A neighbour absent from `neighbours` is skipped rather than patched blind:
its id came from the departing point's payload, so a missing one means the
chain was already broken, and inventing a patch for a point that is not
there would not fix it. The caller logs that case.
"""
patches: list[PayloadPatch] = []
if point.is_active:
patches.append(
PayloadPatch(
point_id=point.point_id,
payload={
"is_active": False,
"deleted_at": now.isoformat(),
**_provenance(point, actor=actor, now=now),
},
expected_version=point.version,
)
)
previous = neighbours.get(point.previous_chunk_id) if point.previous_chunk_id else None
if previous is not None and previous.next_chunk_id != point.next_chunk_id:
patches.append(
PayloadPatch(
point_id=previous.point_id,
payload={
"next_chunk_id": _optional_id(point.next_chunk_id),
**_provenance(previous, actor=actor, now=now),
},
expected_version=previous.version,
)
)
following = neighbours.get(point.next_chunk_id) if point.next_chunk_id else None
if following is not None and following.previous_chunk_id != point.previous_chunk_id:
patches.append(
PayloadPatch(
point_id=following.point_id,
payload={
"previous_chunk_id": _optional_id(point.previous_chunk_id),
**_provenance(following, actor=actor, now=now),
},
expected_version=following.version,
)
)
return tuple(patches)
def patch_for_deactivation(point: Point, *, actor: str, now: datetime) -> PayloadPatch:
"""Deactivate one point without touching any pointer.
Used by the whole-file sweep, where every point in the file leaves at once:
no active point survives to dangle, so there is no neighbour to relink and
the chain stays intact for a later reader of the deactivated file.
"""
return PayloadPatch(
point_id=point.point_id,
payload={
"is_active": False,
"deleted_at": now.isoformat(),
**_provenance(point, actor=actor, now=now),
},
expected_version=point.version,
)

View File

@@ -7,6 +7,7 @@ signature error rather than a cross-tenant leak.
""" """
import uuid import uuid
from datetime import datetime
from sqlalchemy import select from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
@@ -67,3 +68,19 @@ def create(
) )
session.add(source_file) session.add(source_file)
return source_file return source_file
def mark_soft_deleted(source_file: SourceFile, *, deleted_at: datetime) -> None:
"""Retire a file: `status='soft_deleted'` plus `deleted_at` (ADR-0009).
Takes the already-loaded row rather than an id, because the caller fetched
it under its tenant filter and re-fetching here would be a second place
that could forget that filter.
Retiring the row matters beyond bookkeeping: `find_active_by_content_hash`
matches only `active` files, so a re-upload of the same bytes after a delete
creates a fresh file and re-ingests it, instead of taking the duplicate path
and returning a file whose points have all been deactivated.
"""
source_file.status = "soft_deleted"
source_file.deleted_at = deleted_at

View File

@@ -0,0 +1,276 @@
"""Soft delete over HTTP, against real Postgres and real Qdrant (ADR-0002).
The unit tests in `tests/unit/application/points/test_deletion.py` decide
whether the relink logic is right. This file decides whether it is right
*through the stack*: real routing and scope checks, a tenant derived from a real
API key, and a real filtered `set_payload` batch — the last of which is the part
a fake can only approximate, since Qdrant reports success for a patch that
matched nothing.
`DELETE /v1/files/{file_id}` is here too rather than with the upload tests: it
spans both stores, and the assertion that matters is the one about points.
"""
import uuid
from collections.abc import AsyncIterator
import pytest
import pytest_asyncio
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient
from qdrant_client import AsyncQdrantClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from src.bootstrap.dependencies import get_sessionmaker
from src.config import QdrantSettings, Settings
from src.infrastructure.postgres.repositories import source_files as source_files_repo
from src.infrastructure.qdrant.collection import ensure_chunks_collection
from src.infrastructure.qdrant.points import QdrantPointStorage
from src.main import create_app
from tests.support.factories import create_api_key, create_source_file, create_tenant
from tests.support.point_contract import SeedSpec, chunk_point_for, point_id_for
pytestmark = [
pytest.mark.integration,
pytest.mark.qdrant,
pytest.mark.postgres,
pytest.mark.asyncio(loop_scope="session"),
]
FILE_ID = uuid.UUID("cccccccc-cccc-4ccc-8ccc-cccccccccccc")
@pytest.fixture
def api_settings(settings: Settings, qdrant_settings: QdrantSettings) -> Settings:
return settings.model_copy(update={"qdrant": qdrant_settings})
@pytest_asyncio.fixture(loop_scope="session")
async def api_client(
api_settings: Settings, db_sessionmaker: async_sessionmaker[AsyncSession]
) -> AsyncIterator[AsyncClient]:
app = create_app(api_settings)
app.dependency_overrides[get_sessionmaker] = lambda: db_sessionmaker
async with (
LifespanManager(app) as manager,
AsyncClient(transport=ASGITransport(app=manager.app), base_url="http://test") as client,
):
yield client
def _auth(token: str) -> dict[str, str]:
return {"Authorization": f"Bearer {token}"}
async def _tenant_with_key(
db_session: AsyncSession, *, scopes: list[str] | None = None
) -> tuple[uuid.UUID, str]:
tenant = await create_tenant(db_session)
_, token = await create_api_key(
db_session,
tenant=tenant,
scopes=scopes if scopes is not None else ["points:read", "points:write"],
)
await db_session.commit()
return tenant.id, token
async def _seed_chain(
qdrant_client: AsyncQdrantClient,
qdrant_settings: QdrantSettings,
*,
tenant_id: uuid.UUID,
file_id: uuid.UUID = FILE_ID,
length: int = 3,
) -> list[uuid.UUID]:
"""A linked run of points, written the way ingestion writes them."""
await ensure_chunks_collection(qdrant_client, collection=qdrant_settings.collection)
ids = [point_id_for(file_id, index) for index in range(length)]
points = []
for index in range(length):
chunk_point = chunk_point_for(
SeedSpec(tenant_id, file_id, index, f"chunk {index}", float(index + 1))
)
chunk_point.payload["previous_chunk_id"] = str(ids[index - 1]) if index else None
chunk_point.payload["next_chunk_id"] = str(ids[index + 1]) if index + 1 < length else None
points.append(chunk_point)
storage = QdrantPointStorage(qdrant_client, collection=qdrant_settings.collection)
await storage.upsert_points(points)
return ids
async def _read_point(
api_client: AsyncClient, token: str, point_id: uuid.UUID
) -> dict[str, object]:
response = await api_client.get(f"/v1/points/{point_id}", headers=_auth(token))
assert response.status_code == 200
payload: dict[str, object] = response.json()
return payload
async def test_delete_point_deactivates_it_and_relinks_its_neighbours(
api_client: AsyncClient,
db_session: AsyncSession,
qdrant_client: AsyncQdrantClient,
qdrant_settings: QdrantSettings,
) -> None:
tenant_id, token = await _tenant_with_key(db_session)
first, middle, last = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id)
response = await api_client.delete(f"/v1/points/{middle}", headers=_auth(token))
assert response.status_code == 200
assert response.json()["is_active"] is False
assert response.json()["deleted_at"] is not None
assert (await _read_point(api_client, token, first))["next_chunk_id"] == str(last)
assert (await _read_point(api_client, token, last))["previous_chunk_id"] == str(first)
async def test_delete_point_keeps_the_point_in_qdrant(
api_client: AsyncClient,
db_session: AsyncSession,
qdrant_client: AsyncQdrantClient,
qdrant_settings: QdrantSettings,
) -> None:
"""Soft delete means soft: the point is still there, just not listed."""
tenant_id, token = await _tenant_with_key(db_session)
_, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id)
await api_client.delete(f"/v1/points/{middle}", headers=_auth(token))
listed = await api_client.get(
"/v1/points", params={"file_id": str(FILE_ID)}, headers=_auth(token)
)
assert str(middle) not in [point["point_id"] for point in listed.json()["points"]]
assert (await _read_point(api_client, token, middle))["is_active"] is False
async def test_delete_point_is_a_noop_success_the_second_time(
api_client: AsyncClient,
db_session: AsyncSession,
qdrant_client: AsyncQdrantClient,
qdrant_settings: QdrantSettings,
) -> None:
tenant_id, token = await _tenant_with_key(db_session)
first, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id)
await api_client.delete(f"/v1/points/{middle}", headers=_auth(token))
version_after_first = (await _read_point(api_client, token, first))["version"]
again = await api_client.delete(f"/v1/points/{middle}", headers=_auth(token))
assert again.status_code == 200
assert again.json()["is_active"] is False
assert (await _read_point(api_client, token, first))["version"] == version_after_first
async def test_delete_point_returns_404_for_another_tenants_point(
api_client: AsyncClient,
db_session: AsyncSession,
qdrant_client: AsyncQdrantClient,
qdrant_settings: QdrantSettings,
) -> None:
"""`404`, not `403`, and nothing is deactivated on the way to saying so."""
owner_id, owner_token = await _tenant_with_key(db_session)
_, intruder_token = await _tenant_with_key(db_session)
_, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=owner_id)
response = await api_client.delete(f"/v1/points/{middle}", headers=_auth(intruder_token))
assert response.status_code == 404
assert response.json()["error"]["code"] == "not_found"
assert (await _read_point(api_client, owner_token, middle))["is_active"] is True
async def test_delete_point_requires_the_points_write_scope(
api_client: AsyncClient,
db_session: AsyncSession,
qdrant_client: AsyncQdrantClient,
qdrant_settings: QdrantSettings,
) -> None:
"""A read-only key can see a point but must not be able to remove it."""
tenant_id, token = await _tenant_with_key(db_session, scopes=["points:read"])
_, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id)
response = await api_client.delete(f"/v1/points/{middle}", headers=_auth(token))
assert response.status_code == 403
assert (await _read_point(api_client, token, middle))["is_active"] is True
async def test_delete_file_deactivates_every_point_and_retires_the_row(
api_client: AsyncClient,
db_session: AsyncSession,
db_sessionmaker: async_sessionmaker[AsyncSession],
qdrant_client: AsyncQdrantClient,
qdrant_settings: QdrantSettings,
) -> None:
tenant = await create_tenant(db_session)
_, token = await create_api_key(
db_session, tenant=tenant, scopes=["points:read", "points:write"]
)
file_id = uuid.uuid4()
await create_source_file(db_session, tenant=tenant, source_file_id=file_id)
await db_session.commit()
await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant.id, file_id=file_id)
response = await api_client.delete(f"/v1/files/{file_id}", headers=_auth(token))
assert response.status_code == 200
assert response.json()["points_soft_deleted"] == 3
listed = await api_client.get(
"/v1/points", params={"file_id": str(file_id)}, headers=_auth(token)
)
assert listed.json()["points"] == []
async with db_sessionmaker() as session:
row = await source_files_repo.get_by_id(
session, tenant_id=tenant.id, source_file_id=file_id
)
assert row is not None
assert row.status == "soft_deleted"
assert row.deleted_at is not None
async def test_delete_file_returns_404_for_another_tenants_file(
api_client: AsyncClient,
db_session: AsyncSession,
qdrant_client: AsyncQdrantClient,
qdrant_settings: QdrantSettings,
) -> None:
"""The authorization check runs before the sweep, so nothing is deactivated."""
owner = await create_tenant(db_session)
_, owner_token = await create_api_key(
db_session, tenant=owner, scopes=["points:read", "points:write"]
)
intruder = await create_tenant(db_session)
_, intruder_token = await create_api_key(
db_session, tenant=intruder, scopes=["points:read", "points:write"]
)
file_id = uuid.uuid4()
await create_source_file(db_session, tenant=owner, source_file_id=file_id)
await db_session.commit()
await _seed_chain(qdrant_client, qdrant_settings, tenant_id=owner.id, file_id=file_id)
response = await api_client.delete(f"/v1/files/{file_id}", headers=_auth(intruder_token))
assert response.status_code == 404
assert response.json()["error"]["code"] == "not_found"
listed = await api_client.get(
"/v1/points", params={"file_id": str(file_id)}, headers=_auth(owner_token)
)
assert len(listed.json()["points"]) == 3
async def test_delete_file_requires_the_points_write_scope(
api_client: AsyncClient, db_session: AsyncSession
) -> None:
tenant = await create_tenant(db_session)
_, token = await create_api_key(db_session, tenant=tenant, scopes=["files:write"])
file_id = uuid.uuid4()
await create_source_file(db_session, tenant=tenant, source_file_id=file_id)
await db_session.commit()
response = await api_client.delete(f"/v1/files/{file_id}", headers=_auth(token))
assert response.status_code == 403

View File

@@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from src.application.auth.keys import generate_api_key, hash_secret from src.application.auth.keys import generate_api_key, hash_secret
from src.infrastructure.postgres.models.api_key import ApiKey from src.infrastructure.postgres.models.api_key import ApiKey
from src.infrastructure.postgres.models.source_file import SourceFile
from src.infrastructure.postgres.models.tenant import Tenant from src.infrastructure.postgres.models.tenant import Tenant
from src.infrastructure.postgres.models.tenant_domain import TenantDomain from src.infrastructure.postgres.models.tenant_domain import TenantDomain
@@ -50,6 +51,36 @@ async def create_api_key(
return api_key, full_key return api_key, full_key
async def create_source_file(
session: AsyncSession,
*,
tenant: Tenant,
source_file_id: uuid.UUID | None = None,
domain: str = "general",
status: str = "active",
) -> SourceFile:
"""A `source_files` row for tests that address a file without uploading one.
`DELETE /v1/files/{file_id}` authorizes against this row before touching a
single point, so a delete test needs it even though the interesting state
lives in Qdrant.
"""
source_file = SourceFile(
id=source_file_id or uuid.uuid4(),
tenant_id=tenant.id,
domain=domain,
source_filename="handbook.docx",
source_type="docx",
content_sha256="0" * 64,
byte_size=1024,
storage_uri="s3://bucket/key",
status=status,
)
session.add(source_file)
await session.flush()
return source_file
async def create_tenant_domain( async def create_tenant_domain(
session: AsyncSession, session: AsyncSession,
*, *,

View File

@@ -329,6 +329,51 @@ async def scenario_patch_cannot_cross_tenants(
assert victim.content == "onboarding checklist for new staff" assert victim.content == "onboarding checklist for new staff"
async def scenario_patch_batch_applies_every_patch_in_one_call(
repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID
) -> None:
"""One `apply_patches` call carrying several points applies all of them.
This is the primitive soft delete is built on: a deactivation plus its two
neighbour relinks go out together, and a batch that silently applied only
its first operation would leave the pointer chain broken — the exact defect
ADR-0002 rules out. Nulling a pointer is included because `None` and
"absent" are different payload values, and only one of them clears a link.
"""
deleted, previous, following = (point_id_for(file_a, index) for index in (1, 0, 2))
await repository.apply_patches(
tenant_id=tenant_a,
patches=[
PayloadPatch(
point_id=deleted,
payload={"is_active": False, "version": 2},
expected_version=1,
),
PayloadPatch(
point_id=previous,
payload={"next_chunk_id": str(following), "version": 2},
expected_version=1,
),
PayloadPatch(
point_id=following,
payload={"previous_chunk_id": None, "version": 2},
expected_version=1,
),
],
)
applied = {
point.point_id: point
for point in await repository.get_many(
tenant_id=tenant_a, point_ids=[deleted, previous, following]
)
}
assert applied[deleted].is_active is False
assert applied[previous].next_chunk_id == following
assert applied[following].previous_chunk_id is None
assert [applied[point_id].version for point_id in (deleted, previous, following)] == [2, 2, 2]
# Scenarios that only read, safe to run in any order against one seeded corpus. # Scenarios that only read, safe to run in any order against one seeded corpus.
READ_SCENARIOS = ( READ_SCENARIOS = (
scenario_get_returns_point_within_tenant, scenario_get_returns_point_within_tenant,
@@ -350,4 +395,5 @@ WRITE_SCENARIOS = (
scenario_patch_applies_when_version_matches, scenario_patch_applies_when_version_matches,
scenario_patch_is_a_noop_when_version_is_stale, scenario_patch_is_a_noop_when_version_is_stale,
scenario_patch_cannot_cross_tenants, scenario_patch_cannot_cross_tenants,
scenario_patch_batch_applies_every_patch_in_one_call,
) )

View File

@@ -0,0 +1,379 @@
"""Soft delete and neighbour relinking against the fake repository (ADR-0016).
The point of these tests is the pointer chain, not the HTTP surface. ADR-0002
treats a partial relink as a defect, and the ways to produce one are all here:
deleting at either boundary, deleting the same point twice, losing a version
race half way through a batch, and pointing at a neighbour that is gone.
`tests/integration/qdrant/test_points_deletion.py` runs the two central cases
against real Qdrant. What only this file can do cheaply is force the races.
"""
import uuid
from collections.abc import Sequence
from dataclasses import dataclass
from typing import override
import pytest
import structlog.testing
from src.application.points.deletion import soft_delete_file_points, soft_delete_point
from src.application.points.errors import PointVersionConflictError
from src.application.points.point import Point, PointNotFoundError
from src.application.ports.point_repository import PayloadPatch
from tests.fakes import FakePointRepository
from tests.support.point_contract import SeedSpec, build_point
pytestmark = [pytest.mark.unit, pytest.mark.asyncio]
TENANT = uuid.UUID("11111111-1111-4111-8111-111111111111")
OTHER_TENANT = uuid.UUID("22222222-2222-4222-8222-222222222222")
FILE = uuid.UUID("33333333-3333-4333-8333-333333333333")
OTHER_FILE = uuid.UUID("44444444-4444-4444-8444-444444444444")
ACTOR = "api_key:test"
def _chain(count: int, *, tenant: uuid.UUID = TENANT, file_id: uuid.UUID = FILE) -> list[Point]:
"""`count` points of one file, linked head to tail in `order_id` order."""
points = [
build_point(SeedSpec(tenant, file_id, index, f"chunk {index}", float(index + 1)))
for index in range(count)
]
return [
point.model_copy(
update={
"previous_chunk_id": points[index - 1].point_id if index else None,
"next_chunk_id": (points[index + 1].point_id if index + 1 < len(points) else None),
}
)
for index, point in enumerate(points)
]
def _seeded(*points: Point) -> FakePointRepository:
repository = FakePointRepository()
for point in points:
repository.add(point)
return repository
def _stored(repository: FakePointRepository, point_id: uuid.UUID) -> Point:
return repository.points[str(point_id)]
def _walk(repository: FakePointRepository, head: uuid.UUID) -> list[uuid.UUID]:
"""Follow `next_chunk_id` from `head`, guarding against a cycle."""
visited: list[uuid.UUID] = []
current: uuid.UUID | None = head
while current is not None and len(visited) <= len(repository.points):
visited.append(current)
current = _stored(repository, current).next_chunk_id
return visited
async def test_soft_delete_point_relinks_the_neighbours_of_a_middle_point() -> None:
first, middle, last = _chain(3)
repository = _seeded(first, middle, last)
deleted = await soft_delete_point(
repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR
)
assert deleted.is_active is False
assert deleted.deleted_at is not None
assert _stored(repository, first.point_id).next_chunk_id == last.point_id
assert _stored(repository, last.point_id).previous_chunk_id == first.point_id
async def test_soft_delete_point_sends_the_deactivation_and_both_relinks_in_one_batch() -> None:
"""A partial relink is a defect, so the three patches must not be split up."""
first, middle, last = _chain(3)
repository = _seeded(first, middle, last)
await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
assert len(repository.patch_batches) == 1
assert {patch.point_id for patch in repository.patch_batches[0]} == {
middle.point_id,
first.point_id,
last.point_id,
}
async def test_soft_delete_point_bumps_the_version_of_every_point_it_touches() -> None:
"""A relinked neighbour really changed, so a stale editor of it must `409`."""
first, middle, last = _chain(3)
repository = _seeded(first, middle, last)
await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
assert [_stored(repository, point.point_id).version for point in (first, middle, last)] == [
2,
2,
2,
]
assert _stored(repository, first.point_id).updated_by == ACTOR
async def test_soft_delete_point_leaves_the_new_head_without_a_previous_pointer() -> None:
first, second, third = _chain(3)
repository = _seeded(first, second, third)
await soft_delete_point(repository, tenant_id=TENANT, point_id=first.point_id, actor=ACTOR)
assert _stored(repository, second.point_id).previous_chunk_id is None
async def test_soft_delete_point_leaves_the_new_tail_without_a_next_pointer() -> None:
first, second, third = _chain(3)
repository = _seeded(first, second, third)
await soft_delete_point(repository, tenant_id=TENANT, point_id=third.point_id, actor=ACTOR)
assert _stored(repository, second.point_id).next_chunk_id is None
async def test_soft_delete_point_keeps_the_deleted_points_own_pointers() -> None:
"""Nothing active points at it any more, so its pointers record where it sat.
That record is what the retry re-plans from, and what a later restore or an
audit reader would need to place the point back in the sequence.
"""
first, middle, last = _chain(3)
repository = _seeded(first, middle, last)
await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
stored = _stored(repository, middle.point_id)
assert stored.previous_chunk_id == first.point_id
assert stored.next_chunk_id == last.point_id
async def test_soft_delete_point_is_a_noop_for_an_already_inactive_point() -> None:
"""Not a `404`, and not a second relink — no patch is issued at all."""
first, middle, last = _chain(3)
repository = _seeded(first, middle, last)
await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
versions_after_first = {
point.point_id: _stored(repository, point.point_id).version
for point in (first, middle, last)
}
repository.patch_batches.clear()
again = await soft_delete_point(
repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR
)
assert again.is_active is False
assert repository.patch_batches == []
assert {
point.point_id: _stored(repository, point.point_id).version
for point in (first, middle, last)
} == versions_after_first
async def test_soft_delete_point_raises_not_found_for_another_tenants_point() -> None:
first, middle, last = _chain(3)
repository = _seeded(first, middle, last)
with pytest.raises(PointNotFoundError):
await soft_delete_point(
repository, tenant_id=OTHER_TENANT, point_id=middle.point_id, actor=ACTOR
)
assert repository.patch_batches == []
assert _stored(repository, middle.point_id).is_active is True
async def test_soft_delete_point_raises_not_found_for_an_unknown_point() -> None:
repository = _seeded(*_chain(2))
with pytest.raises(PointNotFoundError):
await soft_delete_point(repository, tenant_id=TENANT, point_id=uuid.uuid4(), actor=ACTOR)
async def test_soft_delete_point_skips_a_neighbour_that_is_not_there() -> None:
"""A pointer naming an absent point means the chain was already broken.
The delete completes the half of the relink that exists rather than
refusing, which would leave the point unremovable through any endpoint.
"""
first, middle, last = _chain(3)
repository = _seeded(first, middle) # `last` is never stored
deleted = await soft_delete_point(
repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR
)
assert deleted.is_active is False
assert _stored(repository, first.point_id).next_chunk_id == last.point_id
async def test_soft_delete_point_visits_every_active_point_exactly_once_after_several_deletes() -> (
None
):
"""The plan's traversal property, over a sequence of deletes.
Walking `next_chunk_id` from the head must reach every surviving point once
and never step into a deactivated one.
"""
points = _chain(5)
repository = _seeded(*points)
for index in (1, 3):
await soft_delete_point(
repository, tenant_id=TENANT, point_id=points[index].point_id, actor=ACTOR
)
walked = _walk(repository, points[0].point_id)
assert walked == [points[0].point_id, points[2].point_id, points[4].point_id]
assert all(_stored(repository, point_id).is_active for point_id in walked)
@dataclass
class _ContendedRepository(FakePointRepository):
"""Bumps one point's version just before an apply, as a rival writer would.
That makes the patch guarding on the old version match nothing while the
rest of the batch lands — Qdrant's real behaviour, and the partial apply the
service's retry exists to repair. `rounds` bounds how long the rival keeps
interfering.
"""
rival: uuid.UUID | None = None
rounds: int = 0
@override
async def apply_patches(self, *, tenant_id: uuid.UUID, patches: Sequence[PayloadPatch]) -> None:
if self.rounds > 0 and self.rival is not None:
self.rounds -= 1
victim = self.points[str(self.rival)]
self.points[str(self.rival)] = victim.model_copy(update={"version": victim.version + 1})
await super().apply_patches(tenant_id=tenant_id, patches=patches)
async def test_soft_delete_point_repairs_a_partially_applied_batch_on_retry() -> None:
first, middle, last = _chain(3)
repository = _ContendedRepository(rival=first.point_id, rounds=1)
for point in (first, middle, last):
repository.add(point)
await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
# The first batch left `first` stale; the second re-planned only what was
# still missing, rather than re-deactivating the already-inactive point.
assert len(repository.patch_batches) == 2
assert [patch.point_id for patch in repository.patch_batches[1]] == [first.point_id]
assert _stored(repository, first.point_id).next_chunk_id == last.point_id
assert _stored(repository, middle.point_id).is_active is False
async def test_soft_delete_point_conflicts_when_the_relink_never_settles() -> None:
first, middle, last = _chain(3)
repository = _ContendedRepository(rival=first.point_id, rounds=99)
for point in (first, middle, last):
repository.add(point)
with pytest.raises(PointVersionConflictError):
await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
async def test_soft_delete_file_points_deactivates_every_active_point_of_the_file() -> None:
points = _chain(3)
other_file = _chain(2, file_id=OTHER_FILE)
other_tenant = _chain(1, tenant=OTHER_TENANT, file_id=uuid.uuid4())
repository = _seeded(*points, *other_file, *other_tenant)
swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
assert swept == 3
assert all(not _stored(repository, point.point_id).is_active for point in points)
assert all(_stored(repository, point.point_id).is_active for point in other_file)
assert all(_stored(repository, point.point_id).is_active for point in other_tenant)
async def test_soft_delete_file_points_leaves_the_chain_intact() -> None:
"""No survivor can dangle, so the sweep rewrites no pointer at all."""
first, middle, last = _chain(3)
repository = _seeded(first, middle, last)
await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
assert _stored(repository, first.point_id).next_chunk_id == middle.point_id
assert _stored(repository, middle.point_id).previous_chunk_id == first.point_id
assert _stored(repository, last.point_id).previous_chunk_id == middle.point_id
async def test_soft_delete_file_points_pages_past_one_batch() -> None:
"""More points than one sweep page, so the re-listing loop has to run."""
points = _chain(230)
repository = _seeded(*points)
swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
assert swept == 230
assert all(not _stored(repository, point.point_id).is_active for point in points)
async def test_soft_delete_file_points_is_a_noop_the_second_time() -> None:
repository = _seeded(*_chain(3))
await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
repository.patch_batches.clear()
swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
assert swept == 0
assert repository.patch_batches == []
async def test_soft_delete_file_points_returns_zero_for_an_unknown_file() -> None:
repository = _seeded(*_chain(2))
assert (
await soft_delete_file_points(
repository, tenant_id=TENANT, file_id=uuid.uuid4(), actor=ACTOR
)
== 0
)
async def test_soft_delete_point_logs_its_duration_and_round_count() -> None:
"""ADR-0011's `duration_ms`, plus the field that explains a slow one.
Relinking is O(1), so a delete's cost is Qdrant round trips; `rounds` above
1 means a concurrent writer forced a re-plan rather than the store being
slow, and the two fields are only useful together.
"""
first, middle, last = _chain(3)
repository = _seeded(first, middle, last)
with structlog.testing.capture_logs() as logs:
await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
event = next(entry for entry in logs if entry["event"] == "points.soft_deleted")
assert event["rounds"] == 1
assert isinstance(event["duration_ms"], float)
async def test_soft_delete_file_points_logs_a_round_per_sweep_page() -> None:
"""The sweep is the delete path whose cost tracks the size of the file."""
repository = _seeded(*_chain(230))
with structlog.testing.capture_logs() as logs:
await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
event = next(entry for entry in logs if entry["event"] == "points.file_soft_deleted")
assert event["points_soft_deleted"] == 230
assert event["rounds"] == 3
assert isinstance(event["duration_ms"], float)
async def test_soft_delete_file_points_conflicts_when_a_round_makes_no_progress() -> None:
"""A sweep round that attempts the same ids as the one before is stuck."""
points = _chain(3)
repository = _ContendedRepository(rival=points[0].point_id, rounds=99)
for point in points:
repository.add(point)
with pytest.raises(PointVersionConflictError):
await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)