diff --git a/bhulan/analytics/insights.py b/bhulan/analytics/insights.py index f95d727..fa1cb31 100644 --- a/bhulan/analytics/insights.py +++ b/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, diff --git a/bhulan/analytics/stops.py b/bhulan/analytics/stops.py index a2883a9..3497c76 100644 --- a/bhulan/analytics/stops.py +++ b/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 ): diff --git a/spec/adrs/0016-stop-scan-work-budget.md b/spec/adrs/0016-stop-scan-work-budget.md new file mode 100644 index 0000000..9c27007 --- /dev/null +++ b/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. diff --git a/tests/adversary/test_detect_stops_scan_work_dos.py b/tests/adversary/test_detect_stops_scan_work_dos.py new file mode 100644 index 0000000..f396b74 --- /dev/null +++ b/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