fix: commit multi-step CRUD workflows atomically

src/app/api/api_v1/endpoints/alerts.py

@@ -502,15 +502,14 @@ async def unmatch_alert_sequence(
         .where(cast(Any, AlertSequence.sequence_id) == sequence_id)
     )
     await session.exec(delete_stmt)
-    await session.commit()
-
-    await refresh_alert_state(alert_id, session, alerts)
+    await refresh_alert_state(alert_id, session, alerts, commit=False)
 
     other_links_stmt: Any = (
         select(func.count()).select_from(AlertSequence).where(AlertSequence.sequence_id == sequence_id)
     )
     other_links = int((await session.exec(other_links_stmt)).one())
     if other_links > 0:
+        await session.commit()
         return None
 
     sequence = cast(Sequence, await sequences.get(sequence_id, strict=True))
@@ -527,7 +526,8 @@ async def unmatch_alert_sequence(
             last_seen_at=sequence.last_seen_at,
             lat=None,
             lon=None,
-        )
+        ),
+        commit=False,
     )
     session.add(AlertSequence(alert_id=new_alert.id, sequence_id=sequence_id))
     await session.commit()
@@ -552,6 +552,6 @@ async def delete_alert(
     # Delete associations
     delete_stmt: Any = delete(AlertSequence).where(AlertSequence.alert_id == cast(Any, alert_id))
     await session.exec(delete_stmt)
-    await session.commit()
     # Delete alert
-    await alerts.delete(alert_id)
+    await alerts.delete(alert_id, commit=False)
+    await session.commit()

src/app/api/api_v1/endpoints/detections.py

@@ -131,7 +131,8 @@ async def _create_continuity_detection(
             bbox=EMPTY_BBOXES,
             sequence_id=sequence_id,
             recorded_at=recorded_at,
-        )
+        ),
+        commit=False,
     )
 
 
@@ -283,6 +284,7 @@ async def _maybe_update_alert(
         await alerts.update(
             target_alert_id,
             AlertUpdate(lat=lat, lon=lon, started_at=new_start_at, last_seen_at=new_last_seen),
+            commit=False,
         )
 
 
@@ -343,7 +345,7 @@ async def _merge_alerts(
     await session.exec(delete_links_stmt)
     delete_alerts_stmt: Any = delete(Alert).where(cast(Any, Alert.id).in_(list(absorbed_ids)))
     await session.exec(delete_alerts_stmt)
-    await session.commit()
+    await session.flush()
 
 
 def _rewrite_after_merge(
@@ -421,7 +423,8 @@ async def _get_or_create_alert_id(
             lon=location[1] if isinstance(location, tuple) else None,
             started_at=start_at,
             last_seen_at=last_seen_at,
-        )
+        ),
+        commit=False,
     )
     return alert.id, set()
 
@@ -448,6 +451,8 @@ async def _attach_sequence_to_alert(
     sequences: SequenceCRUD,
     alerts: AlertCRUD,
     anchor_on_sequence: bool = False,
+    *,
+    commit: bool = True,
 ) -> Optional[int]:
     """Assign the given sequence to an alert based on cone/time overlap."""
     camera_by_id = await _get_camera_by_id(camera, cameras, sequence_.camera_id)
@@ -493,7 +498,10 @@ async def _attach_sequence_to_alert(
 
     if to_link:
         session.add_all(to_link)
+    if commit:
         await session.commit()
+    else:
+        await session.flush()
 
     return alert_id
 
@@ -572,9 +580,15 @@ async def create_detection(
         if not continuity_sequences:
             return Response(status_code=status.HTTP_204_NO_CONTENT)
         bucket_key = await upload_file(file, token_payload.organization_id, token_payload.sub)
+        await sequences.lock_pose(pose_id)
+        # Uploading and waiting may change which sequences still need continuity.
+        continuity_sequences = await _get_continuity_sequences(sequences, token_payload.sub, pose_id)
+        if not continuity_sequences:
+            return Response(status_code=status.HTTP_204_NO_CONTENT)
         continuity_dets = await _attach_continuity_detections(
             detections, sequences, continuity_sequences, token_payload.sub, pose_id, bucket_key, effective_recorded_at
         )
+        await detections.session.commit()
         return DetectionRead(**continuity_dets[0].model_dump())
 
     # Upload media
@@ -587,6 +601,8 @@ async def create_detection(
             crop, token_payload.organization_id, token_payload.sub, key_prefix=f"crop_{idx}_"
         )
 
+    # Keep media I/O outside the lock; serialize matching and every write for this pose.
+    await sequences.lock_pose(pose_id)
     created: List[Detection] = []
     camera = cast(Camera, await cameras.get(token_payload.sub, strict=True))
     # sequences touched by this request, to mark due for validation (DB-backed queue).
@@ -605,7 +621,8 @@ async def create_detection(
                 bbox=single_bboxes,
                 others_bboxes=others_bboxes,
                 recorded_at=effective_recorded_at,
-            )
+            ),
+            commit=False,
         )
 
         det_bbox = _parse_bbox(bbox_str)
@@ -633,8 +650,8 @@ async def create_detection(
                 break
 
         if matched_sequence is not None:
-            await sequences.update(matched_sequence.id, SequenceUpdate(last_seen_at=det.created_at))
-            det = await detections.update(det.id, DetectionSequence(sequence_id=matched_sequence.id))
+            await sequences.update(matched_sequence.id, SequenceUpdate(last_seen_at=det.created_at), commit=False)
+            det = await detections.update(det.id, DetectionSequence(sequence_id=matched_sequence.id), commit=False)
             # Only the primary bbox tracks the sequence; siblings in others_bboxes are unrelated detections.
             det_max_conf = max_conf_from_bboxes(det.bbox)
             if det_max_conf is not None:
@@ -682,10 +699,13 @@ async def create_detection(
                         started_at=first_det.recorded_at,
                         last_seen_at=det.created_at,
                         max_conf=seq_max_conf,
-                    )
+                    ),
+                    commit=False,
                 )
                 for det_ in overlapping_dets:
-                    updated = await detections.update(det_.id, DetectionSequence(sequence_id=sequence_.id))
+                    updated = await detections.update(
+                        det_.id, DetectionSequence(sequence_id=sequence_.id), commit=False
+                    )
                     if det_.id == det.id:
                         det = updated
                 affected_sequences.add(sequence_.id)
@@ -714,6 +734,9 @@ async def create_detection(
     for seq_id in affected_sequences:
         await sequences.enqueue_validation(seq_id)
 
+    # Detection rows, sequence membership/confidence and validation jobs become visible together.
+    await detections.session.commit()
+
     first_det = cast(Detection, await detections.get(created[0].id, strict=True))
     return DetectionRead(**first_det.model_dump())
 

src/app/api/api_v1/endpoints/sequences.py

@@ -9,7 +9,7 @@
 from typing import Any, List, Union, cast
 
 from fastapi import APIRouter, Depends, HTTPException, Path, Query, Response, Security, status
-from sqlmodel import delete, func, select
+from sqlmodel import delete, func, select, update
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from app.api.api_v1.endpoints.detections import _attach_sequence_to_alert
@@ -19,7 +19,7 @@
 from app.db import get_session
 from app.models import AlertSequence, AnnotationType, Camera, Detection, Sequence, UserRole
 from app.schemas.alerts import AlertCreate
-from app.schemas.detections import DetectionSequence, DetectionWithUrl
+from app.schemas.detections import DetectionWithUrl
 from app.schemas.login import TokenPayload
 from app.schemas.sequences import SequenceLabel, SequenceRead
 from app.services.alerts import refresh_alert_state
@@ -248,7 +248,6 @@ async def fetch_sequences_from_date(
 async def delete_sequence(
     sequence_id: int = Path(..., gt=0),
     sequences: SequenceCRUD = Depends(get_sequence_crud),
-    detections: DetectionCRUD = Depends(get_detection_crud),
     alerts: AlertCRUD = Depends(get_alert_crud),
     session: AsyncSession = Depends(get_session),
     token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN]),
@@ -257,18 +256,17 @@ async def delete_sequence(
     alert_ids_res = await session.exec(select(AlertSequence.alert_id).where(AlertSequence.sequence_id == sequence_id))
     alert_ids = list(alert_ids_res.all())
     # Unset the sequence_id in the detections
-    det_ids = await session.exec(select(Detection.id).where(Detection.sequence_id == sequence_id))
-    for det_id in det_ids.all():
-        await detections.update(det_id, DetectionSequence(sequence_id=None))
+    unlink: Any = update(Detection).where(cast(Any, Detection.sequence_id) == sequence_id).values(sequence_id=None)
+    await session.exec(unlink)
     # Drop alert links for this sequence to avoid FK issues
     delete_stmt: Any = delete(AlertSequence).where(cast(Any, AlertSequence.sequence_id) == sequence_id)
     await session.exec(delete_stmt)
-    await session.commit()
     # Delete the sequence
-    await sequences.delete(sequence_id)
+    await sequences.delete(sequence_id, commit=False)
     # Refresh affected alerts
     for aid in alert_ids:
-        await refresh_alert_state(aid, session, alerts)
+        await refresh_alert_state(aid, session, alerts, commit=False)
+    await session.commit()
 
 
 @router.patch("/{sequence_id}/label", status_code=status.HTTP_200_OK, summary="Label the nature of the sequence")
@@ -288,7 +286,7 @@ async def label_sequence(
         await verify_org_rights(token_payload.organization_id, sequence.camera_id, cameras)
 
     previous_label = sequence.is_wildfire
-    updated = await sequences.update(sequence_id, payload)
+    updated = await sequences.update(sequence_id, payload, commit=False)
 
     # Reverting a non-wildfire label: re-run cone matching. The sequence keeps its lonely alert
     # until the matching merges it into an overlapping one, so it is never left without an alert.
@@ -300,10 +298,14 @@ async def label_sequence(
     ):
         camera = cast(Camera, await cameras.get(sequence.camera_id, strict=True))
         async with _organization_alert_lock(camera.organization_id):
-            await _attach_sequence_to_alert(updated, camera, cameras, sequences, alerts, anchor_on_sequence=True)
+            await _attach_sequence_to_alert(
+                updated, camera, cameras, sequences, alerts, anchor_on_sequence=True, commit=False
+            )
+            await session.commit()
         return updated
 
     if payload.is_wildfire is None or payload.is_wildfire == AnnotationType.WILDFIRE_SMOKE:
+        await session.commit()
         return updated
 
     alert_ids_res = await session.exec(select(AlertSequence.alert_id).where(AlertSequence.sequence_id == sequence_id))
@@ -320,13 +322,13 @@ async def label_sequence(
         )
         siblings_res = await session.exec(siblings_stmt)
         if siblings_res.first() is None:
+            await session.commit()
             return updated
 
         delete_links: Any = delete(AlertSequence).where(cast(Any, AlertSequence.sequence_id) == sequence_id)
         await session.exec(delete_links)
-        await session.commit()
         for aid in alert_ids:
-            await refresh_alert_state(aid, session, alerts)
+            await refresh_alert_state(aid, session, alerts, commit=False)
 
     # Create a fresh alert for this sequence alone
     camera = cast(Camera, await cameras.get(sequence.camera_id, strict=True))
@@ -337,7 +339,8 @@ async def label_sequence(
             last_seen_at=sequence.last_seen_at,
             lat=None,
             lon=None,
-        )
+        ),
+        commit=False,
     )
     session.add(AlertSequence(alert_id=new_alert.id, sequence_id=sequence_id))
     await session.commit()

src/app/crud/base.py

@@ -23,20 +23,25 @@ def __init__(self, session: AsyncSession, model: Type[ModelType]) -> None:
         self.session = session
         self.model = model
 
-    async def create(self, payload: CreateSchemaType) -> ModelType:
+    async def _save(self, entry: ModelType, *, commit: bool) -> ModelType:
+        self.session.add(entry)
+        await self.session.flush()
+        if commit:
+            await self.session.commit()
+            await self.session.refresh(entry)
+        return entry
+
+    async def create(self, payload: CreateSchemaType, *, commit: bool = True) -> ModelType:
+        """Flush a new row; pass commit=False to include it in a larger transaction."""
         entry = self.model(**payload.model_dump())
         try:
-            self.session.add(entry)
-            await self.session.commit()
+            return await self._save(entry, commit=commit)
         except exc.IntegrityError as error:
             await self.session.rollback()
             raise HTTPException(
                 status_code=status.HTTP_409_CONFLICT,
                 detail=f"An entry with the same index already exists : {error!s}",
             )
-        await self.session.refresh(entry)
-
-        return entry
 
     async def get(self, entry_id: int, strict: bool = False) -> Union[ModelType, None]:
         entry: Union[ModelType, None] = await self.session.get(self.model, entry_id)
@@ -105,25 +110,22 @@ async def fetch_all(
         result = await self.session.exec(statement=statement)
         return [r for r in result]
 
-    async def update(self, entry_id: int, payload: UpdateSchemaType) -> ModelType:
+    async def update(self, entry_id: int, payload: UpdateSchemaType, *, commit: bool = True) -> ModelType:
         access = cast(ModelType, await self.get(entry_id, strict=True))
         values = payload.model_dump(exclude_unset=True)
 
         for k, v in values.items():
             setattr(access, k, v)
 
-        self.session.add(access)
-        await self.session.commit()
-        await self.session.refresh(access)
+        return await self._save(access, commit=commit)
 
-        return access
-
-    async def delete(self, entry_id: int) -> None:
+    async def delete(self, entry_id: int, *, commit: bool = True) -> None:
         await self.get(entry_id, strict=True)
         statement = delete(self.model).where(cast(Any, self.model).id == entry_id)  # ty: ignore[invalid-argument-type]
 
         await self.session.exec(statement=statement)  # type: ignore[call-overload]
-        await self.session.commit()
+        if commit:
+            await self.session.commit()
 
     async def get_in(self, list_: List[Any], field_name: str) -> List[ModelType]:
         statement: Any = select(self.model).where(getattr(self.model, field_name).in_(list_))

src/app/crud/crud_sequence.py

@@ -8,7 +8,7 @@
 from datetime import timedelta
 from typing import Any, Union, cast
 
-from sqlalchemy import case, distinct, func, null, or_, select, update
+from sqlalchemy import case, distinct, func, null, or_, select, text, update
 from sqlmodel import select as select_model
 from sqlmodel.ext.asyncio.session import AsyncSession
 
@@ -26,8 +26,13 @@ class SequenceCRUD(BaseCRUD[Sequence, Sequence, Union[SequenceUpdate, SequenceLa
     def __init__(self, session: AsyncSession) -> None:
         super().__init__(session, Sequence)
 
+    async def lock_pose(self, pose_id: int) -> None:
+        """Serialize sequence assembly for this pose until the caller commits or rolls back."""
+        # Advisory-lock namespace 7342 is reserved for ingestion (7341 is alert attachment).
+        await self.session.exec(text("SELECT pg_advisory_xact_lock(7342, :pose_id)"), params={"pose_id": pose_id})
+
     async def bump_max_conf(self, sequence_id: int, candidate: float) -> None:
-        """Atomically raise sequences.max_conf to candidate if higher (or set if NULL).
+        """Raise sequences.max_conf if higher (or set if NULL), in the caller's transaction.
 
         Uses a portable CASE expression so it runs on SQLite as well as Postgres.
         """
@@ -38,7 +43,6 @@ async def bump_max_conf(self, sequence_id: int, candidate: float) -> None:
         )
         stmt: Any = update(Sequence).where(cast(Any, Sequence.id) == sequence_id).values(max_conf=bumped)
         await self.session.exec(stmt)
-        await self.session.commit()
 
     async def set_temporal_score(
         self,
@@ -84,7 +88,7 @@ async def claim_validation(self, sequence_id: int, validation_status: Union[str,
         return bool(getattr(result, "rowcount", 0))
 
     async def enqueue_validation(self, sequence_id: int) -> None:
-        """Mark the sequence as due for temporal validation (the DB-backed queue).
+        """Mark the sequence due for validation (the DB-backed queue), in the caller's transaction.
 
         Idempotent and FIFO-preserving: ``COALESCE`` keeps the oldest due timestamp, so a
         sequence already queued is NOT re-queued (one entry per sequence, whichever worker
@@ -101,7 +105,6 @@ async def enqueue_validation(self, sequence_id: int) -> None:
             .values(validation_due_at=func.coalesce(due_col, utcnow()))
         )
         await self.session.exec(stmt)
-        await self.session.commit()
 
     async def claim_due_validation(self, lease_seconds: float) -> Union[Sequence, None]:
         """Claim the oldest due sequence for validation, or None when nothing is due.