Compare commits
3 Commits
b25c15fefa
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 0b1932f716 | |||
| 73bdac0da2 | |||
| b58f4630f3 |
52
CLAUDE.md
52
CLAUDE.md
@@ -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
|
||||
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`)
|
||||
with its `Point` read model (`src/application/points/point.py`), the Qdrant adapter
|
||||
(`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
|
||||
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`
|
||||
holds **one** scenario suite run against both `FakePointRepository` (unit) and
|
||||
`QdrantPointRepository` (integration), so new repository behaviour belongs there
|
||||
rather than in one of the two runners — that is what keeps the fake from drifting
|
||||
more permissive than the real store.
|
||||
|
||||
Not built yet: plan 002 Phases 3-6 — soft delete with neighbour relinking,
|
||||
create/replace/patch, reorder and batch, and the
|
||||
`api_request_logs`/`point_audit_events` tables — and `src/agent/`.
|
||||
Not built yet: plan 002 Phases 4-6 — create/replace/patch, reorder and batch,
|
||||
the `api_request_logs`/`point_audit_events` tables, and the runbook section on
|
||||
inspecting and repairing a file's pointer chain — and `src/agent/`.
|
||||
|
||||
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`;
|
||||
|
||||
@@ -17,7 +17,11 @@ from src.application.auth.errors import (
|
||||
TenantInactiveError,
|
||||
)
|
||||
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 (
|
||||
ChunkLimitExceededError,
|
||||
DocumentParseError,
|
||||
@@ -27,6 +31,7 @@ from src.application.ingestion.errors import (
|
||||
PointIndexingError,
|
||||
UnsupportedSourceTypeError,
|
||||
)
|
||||
from src.application.points.errors import PointVersionConflictError
|
||||
from src.application.points.point import PointNotFoundError
|
||||
|
||||
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
|
||||
# nonexistent one, or the API becomes an existence oracle (ADR-0016).
|
||||
(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"),
|
||||
(DomainAlreadyExistsError, status.HTTP_409_CONFLICT, "conflict"),
|
||||
(DocumentParseError, status.HTTP_400_BAD_REQUEST, "validation_error"),
|
||||
|
||||
@@ -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,
|
||||
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 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.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.upload import upload_source_file
|
||||
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"))]
|
||||
_RequirePointsRead = Annotated[AuthContext, Depends(require_scope("points:read"))]
|
||||
_RequirePointsWrite = Annotated[AuthContext, Depends(require_scope("points:write"))]
|
||||
_SessionmakerDep = Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)]
|
||||
_ObjectStorageDep = Annotated[ObjectStorage, Depends(get_object_storage)]
|
||||
_PointStorageDep = Annotated[PointStorage, Depends(get_point_storage)]
|
||||
@@ -100,6 +102,35 @@ async def get_file(
|
||||
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")
|
||||
async def list_points_for_file(
|
||||
file_id: uuid.UUID,
|
||||
|
||||
@@ -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
|
||||
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,
|
||||
)
|
||||
from src.application.auth.context import AuthContext
|
||||
from src.application.points.deletion import soft_delete_point
|
||||
from src.application.points.queries import (
|
||||
count_points,
|
||||
get_point,
|
||||
@@ -43,6 +44,7 @@ from src.bootstrap.dependencies import get_point_repository
|
||||
router = APIRouter(prefix="/points", tags=["points"])
|
||||
|
||||
_RequirePointsRead = Annotated[AuthContext, Depends(require_scope("points:read"))]
|
||||
_RequirePointsWrite = Annotated[AuthContext, Depends(require_scope("points:write"))]
|
||||
_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
|
||||
)
|
||||
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)
|
||||
|
||||
@@ -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):
|
||||
file_id: uuid.UUID
|
||||
source_filename: str
|
||||
|
||||
81
src/application/files/deletion.py
Normal file
81
src/application/files/deletion.py
Normal 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
|
||||
@@ -15,3 +15,11 @@ class InvalidUploadError(FilesError):
|
||||
|
||||
class FileTooLargeError(FilesError):
|
||||
"""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`.
|
||||
"""
|
||||
|
||||
@@ -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
|
||||
stay internal, exported mainly for their own unit tests.
|
||||
|
||||
Direct `/v1/points` CRUD (single-point edits, reordering, keyword search) is
|
||||
plan 002's surface, not this package's — plan 001 scopes it to "the reusable
|
||||
service layer required by ingestion".
|
||||
The `/v1/points` surface lives here too, in its own modules with their own
|
||||
entry points: `queries.py` for the read paths and `deletion.py` for soft delete
|
||||
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
|
||||
|
||||
253
src/application/points/deletion.py
Normal file
253
src/application/points/deletion.py
Normal 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],
|
||||
)
|
||||
19
src/application/points/errors.py
Normal file
19
src/application/points/errors.py
Normal 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.
|
||||
"""
|
||||
135
src/application/points/relinking.py
Normal file
135
src/application/points/relinking.py
Normal 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,
|
||||
)
|
||||
@@ -7,6 +7,7 @@ signature error rather than a cross-tenant leak.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
@@ -67,3 +68,19 @@ def create(
|
||||
)
|
||||
session.add(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
|
||||
|
||||
276
tests/integration/qdrant/test_points_deletion_api.py
Normal file
276
tests/integration/qdrant/test_points_deletion_api.py
Normal 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
|
||||
@@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
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.source_file import SourceFile
|
||||
from src.infrastructure.postgres.models.tenant import Tenant
|
||||
from src.infrastructure.postgres.models.tenant_domain import TenantDomain
|
||||
|
||||
@@ -50,6 +51,36 @@ async def create_api_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(
|
||||
session: AsyncSession,
|
||||
*,
|
||||
|
||||
@@ -329,6 +329,51 @@ async def scenario_patch_cannot_cross_tenants(
|
||||
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.
|
||||
READ_SCENARIOS = (
|
||||
scenario_get_returns_point_within_tenant,
|
||||
@@ -350,4 +395,5 @@ WRITE_SCENARIOS = (
|
||||
scenario_patch_applies_when_version_matches,
|
||||
scenario_patch_is_a_noop_when_version_is_stale,
|
||||
scenario_patch_cannot_cross_tenants,
|
||||
scenario_patch_batch_applies_every_patch_in_one_call,
|
||||
)
|
||||
|
||||
379
tests/unit/application/points/test_deletion.py
Normal file
379
tests/unit/application/points/test_deletion.py
Normal 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)
|
||||
Reference in New Issue
Block a user