detect_stops: bound scan work — fix the 100k-point DoS

bhulan/analytics/insights.py

@@ -26,6 +26,8 @@
     DEFAULT_MIN_DURATION_S,
     DEFAULT_RADIUS_M,
     Stop,
+    StopScanBudgetExceeded,
+    _MAX_STOP_SCAN_WORK,
     detect_stops,
     merge_nearby_stops,
 )
@@ -438,14 +440,29 @@ def compute_insights(request: InsightsRequest) -> InsightsReport:
     start_ts, end_ts = mobility.time_range(prepared)
     box = mobility.bbox(prepared)
 
-    raw_stops = detect_stops(
-        prepared,
-        radius_m=opts.stop_radius_m,
-        min_duration_s=opts.min_stop_minutes * 60.0,
-        # Reuse the trip gap setting so stops split on the same real-world
-        # absence trips do — one knob, no divergent gap mechanism.
-        split_gap_s=opts.trip_split_gap_minutes * 60.0,
-    )
+    try:
+        raw_stops = detect_stops(
+            prepared,
+            radius_m=opts.stop_radius_m,
+            min_duration_s=opts.min_stop_minutes * 60.0,
+            # Reuse the trip gap setting so stops split on the same real-world
+            # absence trips do — one knob, no divergent gap mechanism.
+            split_gap_s=opts.trip_split_gap_minutes * 60.0,
+            # Bound the scan so a pathological single giant cluster (a very slow
+            # drift, or a same-timestamp mass) can't tie up a worker — see ADR
+            # 0016. A real track does far less work than this absolute cap.
+            max_scan_work=_MAX_STOP_SCAN_WORK,
+        )
+    except StopScanBudgetExceeded:
+        # The input is too dense/degenerate for stop detection within budget.
+        # Skip stops (and therefore trips) but still return every other insight
+        # — distance, speed, bbox, hotspots — rather than hang or 500.
+        raw_stops = []
+        quality.issues.append(
+            "Stop detection skipped: the track is too dense or degenerate "
+            "(e.g. a very slow drift or many samples at one timestamp) to "
+            "analyze within the allotted budget."
+        )
     stops = merge_nearby_stops(
         raw_stops,
         merge_radius_m=opts.merge_stops_within_m,

bhulan/analytics/stops.py

@@ -19,7 +19,7 @@
 import math
 from dataclasses import dataclass
 from datetime import datetime
-from typing import List, Optional, Sequence
+from typing import List, Optional, Sequence, Tuple
 
 import numpy as np
 
@@ -35,6 +35,35 @@
 # here to avoid a circular import — trips imports :class:`Stop`).
 DEFAULT_SPLIT_GAP_S = 60 * 60.0  # 60 minutes
 
+# ``detect_stops`` re-grows a spatial cluster from each starting sample whenever
+# the previous cluster was *rejected* (too short in time, or progressive
+# movement). For a real track — where dwells are accepted and the scan jumps
+# past them — that is ~O(n). But a crafted single giant cluster (a very slow
+# drift, or thousands of samples sharing one timestamp) is never accepted, so
+# the scan re-grows an O(n)-sized cluster from every sample: O(n·cluster_size),
+# which at the public 100k-point cap ties up a worker for ~70s (an
+# unauthenticated availability/DoS hole). This budget caps the total scan work
+# at a multiple of the sample count; a real track (dwells accepted, the scan
+# jumps past them → ~1× n work) stays far under it, while a pathological one is
+# bounded to a few seconds. This is an ABSOLUTE cap (not per-point): a realistic
+# track does well under a million work units regardless of size, so it is never
+# touched, while any pathological input — the slow drift, or a dense mass of
+# just-below-``min_duration`` dwells that are rejected and re-grown — is bounded
+# to ~5s of stop scanning before it degrades gracefully (no stops + a quality
+# note), rather than tying up a worker for ~70s. ``work`` counts grown samples
+# plus the window size of every exact-radius recompute (the real time driver, at
+# ~2-3M units/sec). The same-timestamp variant is additionally handled exactly
+# (a zero-duration cluster can never be a stop, so it is skipped wholesale with
+# no re-growth). See ADR 0016.
+_MAX_STOP_SCAN_WORK = 12_000_000
+
+
+class StopScanBudgetExceeded(Exception):
+    """detect_stops' scan work exceeded its budget — the input is a pathological
+    single giant cluster (a slow drift or same-timestamp mass), not a real
+    track. Callers degrade gracefully (report no stops + a quality note) rather
+    than hang."""
+
 
 @dataclass(frozen=True)
 class Stop:
@@ -66,13 +95,16 @@ def _cluster_end(
     n: int,
     radius_m: float,
     split_gap_s: float,
-) -> int:
+) -> Tuple[int, int]:
     """
-    Largest ``end >= i`` such that the window ``[i..end]`` has centroid-spread
-    ``<= radius_m`` *and* no gap between consecutive samples reaches
-    ``split_gap_s``, growing one sample at a time and stopping at the first
-    sample that would push the spread over ``radius_m`` or that sits across a
-    real-world absence.
+    Return ``(end, work)`` where ``end`` is the largest ``end >= i`` such that
+    the window ``[i..end]`` has centroid-spread ``<= radius_m`` *and* no gap
+    between consecutive samples reaches ``split_gap_s`` (growing one sample at a
+    time), and ``work`` is a cost estimate for this grow — one unit per grown
+    sample plus the window size of every exact-spread recompute, which is what
+    actually dominates the running time (the O(window) ``_cluster_radius_m``
+    numpy pass). Callers use ``work`` to enforce a scan budget so a pathological
+    single giant cluster can't run away.
 
     Equivalent to calling :func:`_cluster_radius_m` on every prefix window, but
     avoids the O(k^2) blow-up on a long single cluster: it maintains a running
@@ -91,12 +123,14 @@ def _cluster_end(
     sy = float(ys[i])
     m = 1
     r_est = 0.0
+    work = 0
     j = i
     while j + 1 < n:
+        work += 1
         a = ts[j]
         b = ts[j + 1]
         if a is not None and b is not None and (b - a).total_seconds() >= split_gap_s:
-            return j  # a real-world absence ends the cluster at j
+            return j, work  # a real-world absence ends the cluster at j
         x = float(xs[j + 1])
         y = float(ys[j + 1])
         cox, coy = sx / m, sy / m
@@ -110,12 +144,15 @@ def _cluster_end(
         d_new = math.hypot(x - cnx, y - cny)
         r_est = max(r_est + shift, d_new)
         if r_est > radius_m:
+            # Exact recompute is an O(window) numpy pass — the real cost driver;
+            # charge its size so the scan budget tracks wall-clock work.
+            work += j + 2 - i
             r_exact = _cluster_radius_m(xs[i : j + 2], ys[i : j + 2])
             if r_exact > radius_m:
-                return j  # adding j+1 breaks the cluster; window ends at j
+                return j, work  # adding j+1 breaks the cluster; window ends at j
             r_est = r_exact  # bound was loose — reset it tight and keep going
         j += 1
-    return j
+    return j, work
 
 
 # A cluster is a *stop* only if its points are clustered around a common centre,
@@ -158,6 +195,7 @@ def detect_stops(
     radius_m: float = DEFAULT_RADIUS_M,
     min_duration_s: float = DEFAULT_MIN_DURATION_S,
     split_gap_s: float = DEFAULT_SPLIT_GAP_S,
+    max_scan_work: Optional[int] = None,
 ) -> List[Stop]:
     """
     Return chronologically ordered stops found in the track.
@@ -175,6 +213,12 @@ def detect_stops(
             separated by a real-world absence are reported as two stops rather
             than one stop spanning the calendar gap. Reuses the split-on-gap
             approach of :func:`bhulan.analytics.trips._trip_bounds`.
+        max_scan_work: Optional cap on the total cluster-scan work (summed
+            grown-cluster sizes). When exceeded, :class:`StopScanBudgetExceeded`
+            is raised — the input is a pathological single giant cluster, not a
+            real track. ``None`` (the default) leaves the scan uncapped; the
+            public API passes ``_MAX_STOP_SCAN_WORK_PER_POINT * n`` so a crafted
+            input can't tie up a worker.
     """
     ts_points: List[TrackSample] = [p for p in points if p.ts_utc is not None]
     n = len(ts_points)
@@ -187,13 +231,31 @@ def detect_stops(
     ts = [p.ts_utc for p in ts_points]
 
     stops: List[Stop] = []
+    scan_work = 0
     i = 0
     while i < n:
-        end = _cluster_end(xs, ys, ts, i, n, radius_m, split_gap_s)
+        end, work = _cluster_end(xs, ys, ts, i, n, radius_m, split_gap_s)
+        # Charge this grow's cost (samples scanned + exact-recompute window
+        # sizes) and bail if the cumulative work blows past the budget — a real
+        # track stays far under, a pathological single giant cluster (re-grown
+        # from every sample) would otherwise run O(n·cluster_size).
+        if max_scan_work is not None:
+            scan_work += work
+            if scan_work > max_scan_work:
+                raise StopScanBudgetExceeded(
+                    f"stop scan exceeded {max_scan_work} work units at sample {i}"
+                )
         if end > i:
             duration = (
                 ts_points[end].ts_utc - ts_points[i].ts_utc  # type: ignore[union-attr, operator]
             ).total_seconds()
+            if duration <= 0.0:
+                # Every sample in [i..end] shares one timestamp, so no sub-window
+                # can meet ``min_duration_s`` — the whole cluster is skipped
+                # rather than re-grown from i+1. This makes a same-timestamp mass
+                # O(n) instead of O(n·cluster_size); real tracks never hit it.
+                i = end + 1
+                continue
             if duration >= min_duration_s and not _is_progressive_translation(
                 xs, ys, i, end, radius_m
             ):

spec/adrs/0016-stop-scan-work-budget.md

@@ -0,0 +1,58 @@
+# ADR 0016 — Bound `detect_stops` scan work so a large body can't tie up a worker
+
+**Status:** accepted (cockpit decision, owner-requested)
+**Date:** 2026-07-21
+**Related:** [[0013]] (merge cap), [[0014]] (progressive-movement rejection),
+[[0015]] (request-body depth guard)
+
+## Context
+
+`detect_stops` grows a spatial cluster from a start sample, and when that cluster
+is *rejected* — too short in time, or progressive movement ([[0014]]) — advances
+the start by one and re-grows. For a real track (dwells are accepted and the scan
+jumps past them) that is ~O(n). But a crafted **single giant cluster** is never
+accepted, so the scan re-grows an O(n)-sized cluster from every sample:
+O(n·cluster_size). At the public `MAX_PUBLIC_POINTS` cap of 100 000 a very slow
+drift or a same-timestamp mass takes **~70 s of CPU on one worker**, from a single
+unauthenticated `POST /v1/insights` — an availability/DoS hole the per-IP rate
+limit (30/min) doesn't close. (Progressive-movement rejection [[0014]] widened
+this: slow drifts that used to become stops are now rejected and re-grown.)
+
+A provably-O(n) rewrite needs an incrementally-maintained centroid spread under
+both add and remove (a dynamic farthest-point problem) or a different, results-
+changing spread metric — out of proportion to a demo-API DoS mitigation.
+
+## Decision
+
+Two bounded, real-data-safe guards:
+
+1. **Zero-duration skip (exact).** If a grown cluster's samples all share one
+   timestamp, no sub-window can meet `min_duration_s`, so the whole cluster is
+   skipped (`i = end + 1`) instead of re-grown. This makes a same-timestamp mass
+   O(n) — a 100k same-timestamp body drops from ~70 s to **~0.8 s** — and never
+   affects a real track.
+
+2. **Absolute scan-work budget.** `detect_stops` accepts `max_scan_work` and
+   raises `StopScanBudgetExceeded` once the cumulative scan work passes it, where
+   *work* counts grown samples **plus the window size of every exact-radius
+   recompute** — the real time driver (~2–3 M units/s). The public pipeline
+   passes an **absolute** cap (`_MAX_STOP_SCAN_WORK = 12_000_000`, ≈ 5 s), not a
+   per-point one: a realistic track does well under a million units regardless of
+   size, while any pathological input is bounded to ~5 s. `compute_insights`
+   catches the exception and **degrades gracefully** — reports no stops (and so no
+   trips) plus a `quality` note, but still returns distance, speed, bbox, and
+   hotspots. `max_scan_work=None` (the default) leaves direct callers uncapped.
+
+## Consequences
+
+- Worst-case stop scanning drops from **~70 s to ~5 s**; the same-timestamp
+  variant to **~0.8 s**. Realistic and small-dense tracks are untouched (verified:
+  a 41k drive+stop+drive and a 6k dense track both run < 1 s, no degradation).
+- The cap is **absolute**, so it never false-positives a small dense track whose
+  absolute time is fine — only genuinely large/pathological inputs (a 100k slow
+  drift, or a misconfigured 100k dense mass of just-below-`min_duration` dwells)
+  degrade, and they degrade *gracefully* with an actionable note ("too dense or
+  degenerate … reduce points or adjust `min_stop_minutes`").
+- **Residual / follow-up:** ~5 s is a bound, not a real O(n) fix. Two options for
+  later — a true O(n) sliding-window `detect_stops` (hard), or lowering
+  `MAX_PUBLIC_POINTS` (a product call) to shrink the absolute worst case further.

tests/adversary/test_detect_stops_scan_work_dos.py

@@ -0,0 +1,84 @@
+"""
+A single unauthenticated request must not tie up a worker. ``detect_stops``
+re-grows a spatial cluster from each start sample when the previous cluster is
+rejected; a crafted single giant cluster (a very slow drift, or a same-timestamp
+mass) is never accepted, so the scan runs O(n·cluster_size) — ~70s of CPU at the
+100k-point cap. ADR 0016 bounds this: a zero-duration cluster is skipped
+wholesale (exact), and a scan-work budget caps the rest, degrading gracefully.
+"""
+
+import time
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi.testclient import TestClient
+
+import bhulan.storage.mongo_repo as mongo_repo  # noqa: F401  (fixture parity)
+from bhulan.analytics.mobility import TrackSample
+from bhulan.analytics.stops import StopScanBudgetExceeded, detect_stops
+
+_T0 = datetime(2024, 1, 1, tzinfo=timezone.utc)
+
+
+def _drift(n: int, step_deg: float, same_ts: bool):
+    return [
+        TrackSample(
+            lat=40.0 + i * step_deg,
+            lon=-100.0,
+            ts_utc=_T0 if same_ts else _T0 + timedelta(seconds=i),
+        )
+        for i in range(n)
+    ]
+
+
+def test_same_timestamp_mass_is_on_not_quadratic():
+    # 30k samples drifting slowly, ALL sharing one timestamp: zero duration, so
+    # every cluster is skipped wholesale rather than re-grown. Must be quick and
+    # report no stops. (Without the skip this is tens of seconds.)
+    pts = _drift(30_000, 1e-6, same_ts=True)
+    start = time.monotonic()
+    stops = detect_stops(pts, radius_m=50.0, min_duration_s=300.0)
+    elapsed = time.monotonic() - start
+    assert stops == []
+    assert elapsed < 3.0, f"same-timestamp mass took {elapsed:.1f}s — the zero-duration skip is not working"
+
+
+def test_scan_work_budget_raises_on_a_giant_rejected_cluster():
+    # A slow drift with real timestamps forms one big progressively-moving
+    # cluster that is rejected and re-grown from every sample. With a modest
+    # explicit budget the scan must bail rather than grind.
+    pts = _drift(20_000, 1e-6, same_ts=False)
+    with pytest.raises(StopScanBudgetExceeded):
+        detect_stops(pts, radius_m=50.0, min_duration_s=300.0, max_scan_work=200_000)
+
+
+def test_uncapped_default_is_unchanged_for_a_normal_track():
+    # A real drive → dwell → drive with the default (no budget) still finds the
+    # one stop; the budget only applies when a caller passes max_scan_work.
+    pts = []
+    for i in range(2000):
+        pts.append(TrackSample(lat=40.0 + i * 3e-4, lon=-100.0, ts_utc=_T0 + timedelta(seconds=i)))
+    for i in range(700):
+        pts.append(TrackSample(lat=46.0 + (i % 3) * 1e-5, lon=-100.0, ts_utc=_T0 + timedelta(seconds=2000 + i)))
+    stops = detect_stops(pts, radius_m=50.0, min_duration_s=300.0)
+    assert len(stops) == 1
+
+
+def test_insights_endpoint_bounds_a_pathological_body(client: TestClient):
+    # The real DoS: a 100k-point slow drift through /v1/insights must not hang.
+    # It returns 200 with the other insights intact and a quality note that stop
+    # detection was skipped — bounded to a few seconds, not ~70s.
+    points = [
+        {"lat": 40.0 + i * 1e-6, "lon": -100.0, "ts_utc": (_T0 + timedelta(seconds=i)).isoformat()}
+        for i in range(100_000)
+    ]
+    start = time.monotonic()
+    r = client.post("/v1/insights", json={"points": points})
+    elapsed = time.monotonic() - start
+    assert r.status_code == 200
+    assert elapsed < 20.0, f"a 100k-point drift took {elapsed:.1f}s — the scan-work budget is not bounding it"
+    body = r.json()
+    assert body["stops"] == []
+    assert any("too dense or degenerate" in issue for issue in body["quality"]["issues"])
+    # the rest of the pipeline still ran
+    assert body["summary"]["total_distance_km"] > 0