From fe6a144754bb0619dce96103d8a12baf115ee386 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EC=9C=A0=EC=B0=BD=EC=9A=B1?= Date: Sat, 25 Jul 2026 23:18:19 +0900 Subject: [PATCH] fix(edge): stop hiding a failed camera; fix ghost dedup leak + silent zone drop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Heartbeat reports the CONFIGURED camera count, not just the started ones: a camera that failed to construct at boot (wrong index / unseated ribbon) now shows as total/erroring instead of vanishing, so the server can see it is dark. - GHOST emits without an engine cooldown key (ghost.update already fires once per never-reused track id) — removes a permanent _last_emit key that forget() could never reclaim (a slow 24/7 leak). - Malformed zone rows are logged loudly instead of silently swallowed, matching the pixel-space branch's "don't fail dark" — a dead zone is no longer discovered only after a worker enters it. --- edge/daemon.py | 18 +++++++++++++----- edge/engine.py | 15 +++++++++------ edge/zones.py | 9 ++++++++- tests/test_daemon.py | 14 ++++++++++++++ tests/test_engine.py | 15 +++++++++++++++ tests/test_zones.py | 9 +++++++-- 6 files changed, 66 insertions(+), 14 deletions(-) diff --git a/edge/daemon.py b/edge/daemon.py index 5b928cb..4463a5a 100644 --- a/edge/daemon.py +++ b/edge/daemon.py @@ -18,7 +18,7 @@ def _default_log(msg, exc=None): # pragma: no cover (thin stderr wrapper) def run_loop(runtime, client, queue, *, clock, sleep, iterations=None, sync_every=150, heartbeat_every=900, flush_every=300, frame_interval=0.0, zones_for=None, log=None, - heartbeat_interval_s=None, health_fn=None): + heartbeat_interval_s=None, health_fn=None, configured_cameras=None): # `runtime` may be a single PipelineRuntime or a list of them (multi-camera). All # cameras share this loop, one shared clock read per tick, and the periodic # sync/heartbeat/flush. A per-camera step exception is counted and rate-limit-logged, @@ -102,9 +102,16 @@ def run_loop(runtime, client, queue, *, clock, sleep, iterations=None, t0, f0 = beat_anchor elapsed = now - t0 fps = (counters["frames"] - f0) / elapsed if elapsed > 0 else 0.0 - erroring = sum(1 for rt in runtimes if id(rt) in streak) + # cameras_total is the CONFIGURED count, not just the started one: a camera that + # failed to construct at boot (wrong /dev index, unseated ribbon) is skipped by + # build_runtimes and would otherwise vanish from the heartbeat entirely — the + # server could never see that a configured camera is missing. Count the not-started + # cameras as erroring so a 4-cam device that lost one reports 4 total / 1 erroring. + total = configured_cameras if configured_cameras is not None else len(runtimes) + not_started = max(0, total - len(runtimes)) + erroring = not_started + sum(1 for rt in runtimes if id(rt) in streak) metrics = {"fps": round(fps, 2), "frames": counters["frames"], - "cameras_total": len(runtimes), "cameras_erroring": erroring, + "cameras_total": total, "cameras_erroring": erroring, "errors": counters["errors"], "queue_pending": queue.pending()} # Pi thermal telemetry (None off-Pi or on failure → field omitted, never a fake 0). health = health_fn() @@ -222,10 +229,11 @@ def main() -> None: # pragma: no cover (integration entrypoint, exercised on de return PipelineRuntime(cam_cfg, source, pose, queue, wall_clock=wall) # A missing/unseated camera is skipped (logged), not fatal — the others still run. - runtimes = build_runtimes(parse_cameras(dict(os.environ), cfg.camera_id), - build_source, make_runtime) + cameras = parse_cameras(dict(os.environ), cfg.camera_id) + runtimes = build_runtimes(cameras, build_source, make_runtime) run_loop(runtimes, client, queue, clock=mono, sleep=time.sleep, iterations=None, frame_interval=frame_interval_from_env(dict(os.environ)), heartbeat_interval_s=30, # predictable cadence; server marks offline after 120s (4x) + configured_cameras=len(cameras), # so a camera that failed to start stays visible zones_for=lambda c, rt: zones_from_config(c, rt.camera_id)) diff --git a/edge/engine.py b/edge/engine.py index 62109bc..bb2ccd6 100644 --- a/edge/engine.py +++ b/edge/engine.py @@ -87,12 +87,15 @@ class EventEngine: events.append(self._mk(etype, occurred_at, tr.track_id, zone_id=hit["zone_id"])) for g in self.ghost.update(active_tracks, now): - key = (EventType.GHOST, g["track_id"]) - if self._allow(key, now): - events.append(self._mk(EventType.GHOST, occurred_at, track_id=g["track_id"], - spatial=g["spatial"], - metrics={"last_state": g["last_state"], - "waited_s": g["waited_s"]})) + # No _allow cooldown for GHOST: ghost.update pops each incident from its candidates as it + # emits, and track ids are never reused, so a given track fires GHOST at most once ever. + # Routing it through _allow would only insert a permanent (GHOST, track_id) dedup key for + # a track that was already forgotten reappear_s ago — a slow 24/7 memory leak that + # forget() can never reclaim (the track never appears in lost_tracks again). + events.append(self._mk(EventType.GHOST, occurred_at, track_id=g["track_id"], + spatial=g["spatial"], + metrics={"last_state": g["last_state"], + "waited_s": g["waited_s"]})) # ghost.on_lost already consumed lost tracks' state above, so forgetting here is safe. for tr in lost_tracks: diff --git a/edge/zones.py b/edge/zones.py index 9b2ea7c..15bdc6c 100644 --- a/edge/zones.py +++ b/edge/zones.py @@ -41,7 +41,14 @@ def zones_from_config(config: dict, camera_id: str | None = None) -> list[Zone]: row.get("zone_id"), row.get("camera_id"), polygon[:2]) continue out.append(Zone(str(row["zone_id"]), str(row["level"]), polygon)) - except (KeyError, ValueError, TypeError, IndexError, AttributeError, json.JSONDecodeError): + except (KeyError, ValueError, TypeError, IndexError, AttributeError, json.JSONDecodeError) as e: + # Same principle as the pixel-space branch above: never fail dark. A row missing a key + # (level/zone_id), with unparseable polygon_json, or degenerate points would otherwise be + # dropped silently — intrusion detection for that zone dead with no log, no dashboard + # signal. Flag it so it can be fixed instead of discovered after a worker enters the zone. + zid = row.get("zone_id") if isinstance(row, dict) else None + log.warning("skipping malformed zone row %r (%s: %s) — its intrusion detection is " + "inactive until the row is fixed in the dashboard.", zid, type(e).__name__, e) continue return out diff --git a/tests/test_daemon.py b/tests/test_daemon.py index 2a18f57..5de35f6 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -122,6 +122,20 @@ def test_no_frame_counts_as_erroring_not_a_healthy_frame(): assert m["cameras_erroring"] == 1 # the dead camera is visible to the server +def test_heartbeat_reports_configured_camera_count_when_one_failed_to_start(): + # build_runtimes skips a camera that fails to construct at boot, so `runtime` holds only the + # started cameras. The heartbeat must still report the CONFIGURED total and count the missing + # one as erroring — otherwise the server can never see that a configured camera is dark. + rt, cl, q = _Runtime(), _Client(), _Queue() # 1 runtime started... + ticks = iter(float(t) for t in range(1000)) + run_loop(rt, cl, q, clock=lambda: next(ticks), sleep=lambda s: None, + iterations=10, sync_every=99, heartbeat_every=10, flush_every=99, + configured_cameras=2) # ...but 2 were configured + m = cl.metrics[0] + assert m["cameras_total"] == 2 # the missing camera is not hidden + assert m["cameras_erroring"] == 1 # and shows as erroring + + def test_heartbeat_reports_queue_depth_and_echoes_config_version(): class _VersionClient(_Client): def fetch_config(self): diff --git a/tests/test_engine.py b/tests/test_engine.py index 6e0593a..913111d 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -117,6 +117,21 @@ def test_pinch_zone_emits_pinch_event_not_warning(): assert not any(e.event_type == EventType.ZONE_WARNING for e in events) +def test_ghost_emit_does_not_leak_a_permanent_dedup_key(): + # A GHOST fires reappear_s AFTER the track was lost (and already forgotten). Routing it through + # the engine cooldown used to insert a (GHOST, track_id) key that forget() could never reclaim + # (the track never appears in lost_tracks again) — a slow 24/7 memory leak. Emitting directly + # must leave no such key behind. + eng = _engine() + tr = Track(1, deque(maxlen=90)) + _lying_track(tr, 0.0); eng.process([tr], [], now=0.0) + _lying_track(tr, 0.6); eng.process([tr], [], now=0.6) # confirmed fallen -> dangerous state + eng.process([], [tr], now=1.0) # lost -> ghost candidate registered + events = eng.process([], [], now=12.0) # > reappear_s (10s), no reappearance + assert any(e.event_type == EventType.GHOST for e in events) + assert not any(k[0] == EventType.GHOST for k in eng._last_emit) # no leaked ghost dedup key + + def test_two_simultaneous_tracks_have_independent_fall_state(): # Per-track detector state must not leak between two people monitored at once: a fall on track A # must arm and emit for A only, never for a standing track B. All other engine/detector tests diff --git a/tests/test_zones.py b/tests/test_zones.py index a8f936c..f9578db 100644 --- a/tests/test_zones.py +++ b/tests/test_zones.py @@ -16,9 +16,14 @@ def test_parses_zone_rows_json_string_polygon(): assert zones[0].polygon == [(0.0, 0.0), (0.9, 0.0), (0.9, 0.9)] assert thresholds_from_config(cfg)["fall_prone_s"] == 3.0 -def test_skips_malformed_row(): +def test_skips_malformed_row_and_warns(caplog): + # A malformed row (missing polygon_json/level here) must be DROPPED and LOGGED — never silently + # swallowed, matching the pixel-space branch's "don't fail dark" principle: a dead zone with no + # signal is only discovered after a worker enters it. cfg = {"zones": [{"zone_id": "bad"}], "thresholds": {}} - assert zones_from_config(cfg) == [] + with caplog.at_level(logging.WARNING): + assert zones_from_config(cfg) == [] + assert any("bad" in r.message and "malformed" in r.message for r in caplog.records) def test_empty_config(): assert zones_from_config({}) == []