perf(triangulation): keep API responsive during overlap work

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

@@ -50,7 +50,7 @@
 from app.schemas.login import TokenPayload
 from app.schemas.sequences import SequenceUpdate
 from app.services.cones import resolve_cone
-from app.services.overlap import compute_overlap, haversine_km
+from app.services.overlap import compute_overlap_async, haversine_km
 from app.services.sequence_confidence import max_conf_from_bboxes
 from app.services.storage import s3_service, upload_file
 from app.services.telemetry import telemetry_client
@@ -233,13 +233,14 @@ def _build_overlap_records(
     return records
 
 
-def _resolve_groups_and_locations(
+async def _resolve_groups_and_locations(
     records: List[Dict[str, Any]],
     sequence_id: int,
 ) -> Optional[Tuple[List[Tuple[int, ...]], Dict[Tuple[int, ...], Optional[Tuple[float, float]]]]]:
     if not records:
         return None
-    row = next((row for row in compute_overlap(records) if row["id"] == int(sequence_id)), None)
+    result = await compute_overlap_async(records)
+    row = next((row for row in result if row["id"] == int(sequence_id)), None)
     if row is None:
         return None
     groups = [tuple(g) for g in row["event_groups"]]
@@ -471,7 +472,7 @@ async def _attach_sequence_to_alert(
 
     # Build records for overlap computation
     records = _build_overlap_records(recent_sequences, camera_by_id)
-    resolved = _resolve_groups_and_locations(records, int(sequence_.id))
+    resolved = await _resolve_groups_and_locations(records, int(sequence_.id))
     if resolved is None:
         return None
     groups, group_locations = resolved

src/app/services/alerts.py

@@ -12,7 +12,7 @@
 from app.crud import AlertCRUD
 from app.models import AlertSequence, Camera, Sequence
 from app.schemas.alerts import AlertUpdate
-from app.services.overlap import compute_overlap
+from app.services.overlap import compute_overlap_async
 
 __all__ = ["refresh_alert_state"]
 
@@ -51,7 +51,7 @@ async def refresh_alert_state(alert_id: int, session: AsyncSession, alerts: Aler
                 "started_at": seq.started_at,
                 "last_seen_at": seq.last_seen_at,
             })
-        result = compute_overlap(records)
+        result = await compute_overlap_async(records)
         loc = next((loc for row in result for loc in row["event_smoke_locations"] if loc is not None), None)
 
     await alerts.update(

src/app/services/overlap.py

@@ -15,6 +15,7 @@
 
 import networkx as nx
 import numpy as np
+from anyio import CapacityLimiter, to_thread
 from pyproj import Geod, Transformer
 from shapely import transform as shapely_transform
 from shapely.geometry import Polygon
@@ -28,6 +29,8 @@
 _TO_MERCATOR = Transformer.from_crs("EPSG:4326", "EPSG:3857", always_xy=True)
 _TO_LATLON = Transformer.from_crs("EPSG:3857", "EPSG:4326", always_xy=True)
 _PairCentroids = Dict[Tuple[int, int], Optional[Tuple[float, float]]]
+# Keep geometry jobs serial per process without blocking unrelated API requests.
+_TRIANGULATION_LIMITER = CapacityLimiter(1)
 
 
 def haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
@@ -415,6 +418,11 @@ def _attach_groups_to_records(
         row["event_smoke_locations"] = seq_to_smokes.get(row["id"], [])
 
 
+async def compute_overlap_async(api_sequences: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """Run overlap computation in a bounded worker thread."""
+    return await to_thread.run_sync(compute_overlap, api_sequences, limiter=_TRIANGULATION_LIMITER)
+
+
 def compute_overlap(
     api_sequences: List[Dict[str, Any]],
     r_km: float = 35.0,

src/tests/endpoints/test_detections.py

@@ -604,11 +604,13 @@ async def test_build_overlap_records_skips_missing_cone_data(detection_session:
     assert records == []
 
 
-def test_resolve_groups_and_locations_empty_records_returns_none():
-    assert _resolve_groups_and_locations([], 1) is None
+@pytest.mark.asyncio
+async def test_resolve_groups_and_locations_empty_records_returns_none():
+    assert await _resolve_groups_and_locations([], 1) is None
 
 
-def test_resolve_groups_and_locations_no_match_returns_none():
+@pytest.mark.asyncio
+async def test_resolve_groups_and_locations_no_match_returns_none():
     now = utcnow()
     records = [
         {
@@ -622,7 +624,7 @@ def test_resolve_groups_and_locations_no_match_returns_none():
             "last_seen_at": now,
         }
     ]
-    assert _resolve_groups_and_locations(records, 999) is None
+    assert await _resolve_groups_and_locations(records, 999) is None
 
 
 @pytest.mark.parametrize(

src/tests/services/test_overlap.py

@@ -3,13 +3,16 @@
 # This program is licensed under the Apache License 2.0.
 # See LICENSE or go to <https://www.apache.org/licenses/LICENSE-2.0> for full license details.
 
+import asyncio
 from datetime import datetime, timedelta
+from threading import Event, get_ident
 
 import pytest
 from shapely.geometry import box
 
 from app.core.time import utcnow
-from app.services.overlap import _find_overlapping_pairs, compute_overlap
+from app.services import overlap
+from app.services.overlap import _find_overlapping_pairs, compute_overlap, compute_overlap_async
 
 
 @pytest.mark.parametrize("relaxation", [0, 30])
@@ -173,3 +176,34 @@ def test_compute_overlap_copies_records_and_handles_empty_input(label) -> None:
     assert result[0]["event_groups"] == [(1,)]
     assert result[0]["event_smoke_locations"] == ([None] if label is None else [])
     assert compute_overlap([]) == []
+
+
+@pytest.mark.asyncio
+async def test_compute_overlap_async_keeps_loop_responsive_and_serializes(monkeypatch) -> None:
+    now = utcnow()
+    records = [_make_sequence(1, 48.3792, 2.8208, 276.5, 3.0, now, now)]
+    expected = compute_overlap(records)
+    entered, release = Event(), Event()
+    loop_thread = get_ident()
+    calls = []
+
+    def blocking_compute(data):
+        assert get_ident() != loop_thread
+        calls.append(data)
+        entered.set()
+        assert release.wait(5)
+        return compute_overlap(data)
+
+    monkeypatch.setattr(overlap, "compute_overlap", blocking_compute)
+    tasks = [asyncio.create_task(compute_overlap_async(records))]
+    try:
+        assert await asyncio.to_thread(entered.wait, 1)
+        assert not tasks[0].done()
+        tasks.append(asyncio.create_task(compute_overlap_async(records)))
+        await asyncio.sleep(0.05)
+        assert calls == [records]
+    finally:
+        release.set()
+        results = await asyncio.gather(*tasks)
+    assert results == [expected, expected]
+    assert calls == [records, records]