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>
This commit is contained in:
@@ -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"),
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
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):
|
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`.
|
||||||
|
"""
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
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
|
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
|
||||||
|
|||||||
Reference in New Issue
Block a user