diff --git a/src/app/api/api_v1/endpoints/alerts.py b/src/app/api/api_v1/endpoints/alerts.py index 855fdd4b..2c6c0a60 100644 --- a/src/app/api/api_v1/endpoints/alerts.py +++ b/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() diff --git a/src/app/api/api_v1/endpoints/detections.py b/src/app/api/api_v1/endpoints/detections.py index 9d5c2d40..363fda38 100644 --- a/src/app/api/api_v1/endpoints/detections.py +++ b/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()) diff --git a/src/app/api/api_v1/endpoints/sequences.py b/src/app/api/api_v1/endpoints/sequences.py index 6cc8a3c6..3697d9dc 100644 --- a/src/app/api/api_v1/endpoints/sequences.py +++ b/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() diff --git a/src/app/crud/base.py b/src/app/crud/base.py index e7a79482..9a5d8a86 100644 --- a/src/app/crud/base.py +++ b/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_)) diff --git a/src/app/crud/crud_sequence.py b/src/app/crud/crud_sequence.py index 8e8a70b6..37567ca5 100644 --- a/src/app/crud/crud_sequence.py +++ b/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. diff --git a/src/app/services/alerts.py b/src/app/services/alerts.py index 952b71ac..de60f103 100644 --- a/src/app/services/alerts.py +++ b/src/app/services/alerts.py @@ -17,7 +17,7 @@ __all__ = ["refresh_alert_state"] -async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: AlertCRUD) -> None: +async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: AlertCRUD, *, commit: bool = True) -> None: """Recompute an alert's bounds and location from its remaining sequences, or delete it if empty.""" remaining_stmt: Any = ( select(Sequence, Camera) @@ -28,7 +28,7 @@ async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: Aler remaining_res = await session.exec(remaining_stmt) rows = remaining_res.all() if not rows: - await alerts.delete(alert_id) + await alerts.delete(alert_id, commit=commit) return seqs = [row[0] for row in rows] @@ -62,4 +62,5 @@ async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: Aler lat=loc[0] if loc else None, lon=loc[1] if loc else None, ), + commit=commit, ) diff --git a/src/tests/endpoints/test_alerts.py b/src/tests/endpoints/test_alerts.py index 170c4e7c..47c65434 100644 --- a/src/tests/endpoints/test_alerts.py +++ b/src/tests/endpoints/test_alerts.py @@ -7,6 +7,7 @@ import io from datetime import datetime, timedelta from typing import Any, Dict, List, Tuple, cast +from unittest.mock import AsyncMock import pytest # type: ignore from httpx import AsyncClient @@ -16,6 +17,7 @@ from app.api.api_v1.endpoints.alerts import _ALERT_EXPORT_COLUMNS, _iter_alerts_csv from app.core.config import settings from app.core.time import utcnow +from app.db import session_factory from app.models import Alert, AlertSequence, AnnotationType, Camera, Detection, Organization, Pose, Sequence from app.services.overlap import compute_overlap @@ -994,3 +996,34 @@ async def test_alerts_export_invalid_range( async def test_alerts_export_unauthenticated(async_client: AsyncClient, detection_session: AsyncSession): resp = await async_client.get("/alerts/export?from_date=2026-04-10&to_date=2026-04-12") assert resp.status_code == 401, resp.text + + +@pytest.mark.parametrize( + ("method", "path", "failure"), + [ + ("PATCH", "/sequences/{sid}/label", "app.crud.AlertCRUD.create"), + ("POST", "/alerts/{aid}/sequences/{sid}/unmatch", "app.crud.AlertCRUD.create"), + ("DELETE", "/sequences/{sid}", "app.api.api_v1.endpoints.sequences.refresh_alert_state"), + ("DELETE", "/alerts/{aid}", "app.crud.AlertCRUD.delete"), + ], +) +@pytest.mark.asyncio +async def test_alert_changes_roll_back_on_late_failure( + async_client, detection_session, monkeypatch, method, path, failure +): + alert, sequence_ids, counts = await _create_alert_with_sequences(detection_session, 1, 1, 48, 2) + sid = sequence_ids[0] + before = (await detection_session.get(Sequence, sid)).model_dump() + monkeypatch.setattr(failure, AsyncMock(side_effect=RuntimeError("late failure"))) + kwargs = {"json": {"is_wildfire": "other_smoke"}} if method == "PATCH" else {} + with pytest.raises(RuntimeError, match="late failure"): + await async_client.request( + method, path.format(aid=alert.id, sid=sid), headers=pytest.get_token(1, ["admin"], 1), **kwargs + ) + async with session_factory() as observer: + assert await observer.get(Alert, alert.id) is not None + assert (await observer.get(Sequence, sid)).model_dump() == before + links = (await observer.exec(select(AlertSequence.sequence_id).where(AlertSequence.alert_id == alert.id))).all() + assert set(links) == set(sequence_ids) + detections = (await observer.exec(select(Detection).where(Detection.sequence_id == sid))).all() + assert len(detections) == counts[0] diff --git a/src/tests/endpoints/test_detections.py b/src/tests/endpoints/test_detections.py index dc413c30..e2b4416a 100644 --- a/src/tests/endpoints/test_detections.py +++ b/src/tests/endpoints/test_detections.py @@ -10,7 +10,7 @@ import pytest # type: ignore from fastapi import HTTPException, UploadFile from httpx import AsyncClient -from sqlmodel import select +from sqlmodel import select, text from sqlmodel.ext.asyncio.session import AsyncSession from app.api.api_v1.endpoints import detections as detections_api @@ -34,6 +34,7 @@ from app.core.config import settings from app.core.time import utcnow from app.crud import AlertCRUD, CameraCRUD, DetectionCRUD, OrganizationCRUD, PoseCRUD, SequenceCRUD +from app.db import session_factory from app.models import ( Alert, AlertSequence, @@ -2243,3 +2244,125 @@ async def fake_upload_file(file: UploadFile, organization_id: int, camera_id: in assert response.status_code == 403, response.text assert response.json()["detail"] == "Access forbidden." assert upload_calls == [] + + +@pytest.mark.asyncio +async def test_ingestion_rolls_back_on_late_queue_failure(async_client, detection_session, monkeypatch): + sequence = await detection_session.get(Sequence, 1) + sequence.last_seen_at, sequence.max_conf, sequence.is_validated = utcnow(), 0.1, False + await detection_session.commit() + before = sequence.model_dump() + original_ids = (await detection_session.exec(select(Detection.id))).all() + enqueue = SequenceCRUD.enqueue_validation + + async def fail_enqueue(self, sequence_id, **kwargs): + await enqueue(self, sequence_id, **kwargs) + raise RuntimeError("late queue failure") + + monkeypatch.setattr(detections_api, "upload_file", AsyncMock(return_value="new-frame")) + monkeypatch.setattr(SequenceCRUD, "enqueue_validation", fail_enqueue) + with pytest.raises(RuntimeError, match="late queue failure"): + await async_client.post( + "/detections", + headers=pytest.get_token(1, ["camera"], 1), + data={"pose_id": 1, "bboxes": "[(0.1,0.1,0.2,0.2,0.9)]"}, + files={"file": ("frame.jpg", b"frame", "image/jpeg")}, + ) + async with session_factory() as observer: + assert (await observer.exec(select(Detection.id))).all() == original_ids + assert (await observer.get(Sequence, 1)).model_dump() == before + await observer.exec(text("SET LOCAL lock_timeout = '1s'")) + await SequenceCRUD(observer).lock_pose(1) # Rollback released the transaction lock. + + +@pytest.mark.parametrize( + "second_boxes", + [ + "[(0.8,0.8,0.9,0.9,0.8),(0.1,0.1,0.2,0.2,0.9)]", + "[(0.8,0.8,0.9,0.9,0.8)]", + "[]", + ], + ids=["opposite-order", "matched-and-continuity", "empty-continuity"], +) +@pytest.mark.asyncio +async def test_concurrent_ingestion_serializes_only_the_same_pose( + async_client, detection_session, monkeypatch, second_boxes +): + sequence = await detection_session.get(Sequence, 1) + sequence.last_seen_at, sequence.is_validated = utcnow(), False + other = Sequence(**(sequence.model_dump() | {"id": None})) + detection_session.add(other) + await detection_session.flush() + detection_session.add( + Detection( + camera_id=1, + pose_id=1, + sequence_id=other.id, + bucket_key="other-seed", + bbox="[(0.8,0.8,0.9,0.9,0.8)]", + ) + ) + await detection_session.commit() + monkeypatch.setattr(settings, "SEQUENCE_BBOX_TOLERANCE", 0.0) + monkeypatch.setattr( + detections_api, + "upload_file", + AsyncMock(side_effect=["first-frame", "second-frame", "other-pose"]), + ) + paused, resume = asyncio.Event(), asyncio.Event() + update_sequence = SequenceCRUD.update + holder_pid = 0 + + async def pause_after_first_update(self, sequence_id, payload, **kwargs): + nonlocal holder_pid + result = await update_sequence(self, sequence_id, payload, **kwargs) + if sequence_id == 1 and not paused.is_set(): + holder_pid = (await self.session.exec(text("SELECT pg_backend_pid()"))).scalar_one() + paused.set() + await resume.wait() + return result + + monkeypatch.setattr(SequenceCRUD, "update", pause_after_first_update) + + async def post(boxes, pose_id=1): + return await async_client.post( + "/detections", + headers=pytest.get_token(1, ["camera"], 1), + data={"pose_id": pose_id, "bboxes": boxes}, + files={"file": ("frame.jpg", b"frame", "image/jpeg")}, + ) + + tasks = [] + try: + async with asyncio.timeout(10): + tasks.append(asyncio.create_task(post("[(0.1,0.1,0.2,0.2,0.9),(0.8,0.8,0.9,0.9,0.8)]"))) + await paused.wait() # First request holds sequence A while the second starts with B. + tasks.append(asyncio.create_task(post(second_boxes))) + async with session_factory() as observer: + # Wait for a real database lock wait, not a scheduler-dependent sleep. + while not ( + await observer.exec( + text( + "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE locktype = 'advisory' " + "AND NOT granted AND :holder = ANY(pg_blocking_pids(pid)))" + ), + params={"holder": holder_pid}, + ) + ).scalar_one(): + pass + assert not tasks[1].done() + assert (await post("[(0.1,0.1,0.2,0.2,0.9)]", pose_id=2)).status_code == 201 + resume.set() + assert [r.status_code for r in await asyncio.gather(*tasks)] == [201, 201] + finally: + resume.set() + for task in tasks: + if not task.done(): + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + async with session_factory() as observer: + for key in ("first-frame", "second-frame"): + rows = (await observer.exec(select(Detection).where(Detection.bucket_key == key))).all() + assert len(rows) == 2 + assert {row.sequence_id for row in rows} == {1, other.id} diff --git a/src/tests/endpoints/test_sequences.py b/src/tests/endpoints/test_sequences.py index b4bd63ee..d74bfb7c 100644 --- a/src/tests/endpoints/test_sequences.py +++ b/src/tests/endpoints/test_sequences.py @@ -753,11 +753,11 @@ async def test_unit_label_sequence_as_other_smoke_refreshes_alert( # 3. Assertions mock_sequences_crud.get.assert_called_once_with(1, strict=True) - mock_sequences_crud.update.assert_called_once_with(1, payload) + mock_sequences_crud.update.assert_called_once_with(1, payload, commit=False) # Three session.exec calls: alert_ids lookup, siblings probe, delete links assert mock_session.exec.call_count == 3 - mock_refresh_alert_state.assert_called_once_with(101, mock_session, mock_alerts_crud) + mock_refresh_alert_state.assert_called_once_with(101, mock_session, mock_alerts_crud, commit=False) # Verify a new alert was created for this sequence mock_alerts_crud.create.assert_called_once() @@ -813,6 +813,7 @@ async def test_unit_relabel_sequence_to_wildfire_smoke_reattaches( mock_sequences_crud, mock_alerts_crud, anchor_on_sequence=True, + commit=False, ) mock_session.exec.assert_not_called() mock_alerts_crud.create.assert_not_called() @@ -899,7 +900,7 @@ async def test_unit_label_sequence_as_wildfire_smoke_does_not_refresh(): ) # 3. Assertions - mock_sequences_crud.update.assert_called_once_with(1, payload) + mock_sequences_crud.update.assert_called_once_with(1, payload, commit=False) mock_session.exec.assert_not_called() mock_alerts_crud.create.assert_not_called() diff --git a/src/tests/endpoints/test_temporal_validation.py b/src/tests/endpoints/test_temporal_validation.py index 0cd99579..a1ed2250 100644 --- a/src/tests/endpoints/test_temporal_validation.py +++ b/src/tests/endpoints/test_temporal_validation.py @@ -78,6 +78,7 @@ async def _seed_sequence( async def _enqueue(session: AsyncSession, sequence_id: int) -> None: await SequenceCRUD(session).enqueue_validation(sequence_id) + await session.commit() async def _backdate_due(seconds: float = 60.0) -> None: @@ -159,12 +160,14 @@ async def test_enqueue_is_idempotent_and_fifo_preserving(detection_session: Asyn seq = await _seed_sequence(detection_session, 5) crud = SequenceCRUD(detection_session) await crud.enqueue_validation(cast(int, seq.id)) + await detection_session.commit() await detection_session.refresh(seq) first_due = seq.validation_due_at assert first_due is not None await asyncio.sleep(0.01) await crud.enqueue_validation(cast(int, seq.id)) # second detection while queued + await detection_session.commit() await detection_session.refresh(seq) assert seq.validation_due_at == first_due # one entry per sequence, oldest due kept @@ -175,12 +178,14 @@ async def test_enqueue_skips_validated_and_window_exhausted(detection_session: A crud = SequenceCRUD(detection_session) await crud.claim_validation(cast(int, seq.id)) await crud.enqueue_validation(cast(int, seq.id)) + await detection_session.commit() await detection_session.refresh(seq) assert seq.validation_due_at is None # validated: nothing to do seq2 = await _seed_sequence(detection_session, 5) await crud.finish_validation_job(cast(int, seq2.id), validation_status=WINDOW_EXHAUSTED) await crud.enqueue_validation(cast(int, seq2.id)) + await detection_session.commit() await detection_session.refresh(seq2) assert seq2.validation_due_at is None # terminal: never resurrected @@ -377,6 +382,7 @@ async def test_process_window_exhausted_when_scored_and_past_max_frames(detectio # Terminal: a later fail-open or enqueue can't resurrect it. await crud.enqueue_validation(cast(int, seq.id)) + await detection_session.commit() await detection_session.refresh(seq) assert seq.validation_due_at is None @@ -585,6 +591,7 @@ async def test_poison_job_dead_letters_after_max_attempts(detection_session: Asy # Terminal: new detections can't resurrect it, the worker sees nothing due. await crud.enqueue_validation(cast(int, seq.id)) + await detection_session.commit() await detection_session.refresh(seq) assert seq.validation_due_at is None assert await _process_one_due() is False @@ -897,6 +904,7 @@ async def test_first_score_past_window_below_threshold_is_terminal(detection_ses # Terminal: a later enqueue can't resurrect it. await SequenceCRUD(detection_session).enqueue_validation(cast(int, seq.id)) + await detection_session.commit() await detection_session.refresh(seq) assert seq.validation_due_at is None