Snapshot track history deques before off-thread iteration

retina_tracker/track.py

@@ -794,8 +794,17 @@ def get_length_bucket(self):
 
     def get_recent_detections(self, n=20):
         """Return the last *n* non-None detections (reverse scan, early exit)."""
+        # Called off the frame worker (analytics, feed builders) while that worker
+        # may be appending, and a deque raises RuntimeError if it is mutated during
+        # Python-level iteration.  list(deque) is a single C call, so under the GIL
+        # it is an atomic snapshot; take both buffers before iterating.  The zip is
+        # unchanged: a reader that lands mid-append can see one more timestamp than
+        # measurement (the mutator appends timestamps first), and zip truncating to
+        # the shorter buffer is the same behaviour these buffers had as lists.
+        meas = list(self.history["measurements"])
+        ts_hist = list(self.history["timestamps"])
         result = []
-        for m, ts in zip(reversed(self.history["measurements"]), reversed(self.history["timestamps"])):
+        for m, ts in zip(reversed(meas), reversed(ts_hist)):
             if m is not None:
                 result.append(
                     {
@@ -812,6 +821,21 @@ def get_recent_detections(self, n=20):
         return result
 
     def to_dict(self):
+        # Served from admin/state-snapshot threads while the frame worker appends.
+        # list(deque) is atomic under the GIL; the comprehensions below are not, so
+        # snapshot every buffer first.  The snapshots can differ in length by one
+        # (the mutator appends timestamps -> frames -> states -> measurements), so
+        # truncate to the common length to keep the emitted arrays aligned.
+        hist_timestamps = list(self.history["timestamps"])
+        hist_states = list(self.history["states"])
+        hist_measurements = list(self.history["measurements"])
+        hist_state_status = list(self.history["state_status"])
+        n_hist = min(len(hist_timestamps), len(hist_states), len(hist_measurements), len(hist_state_status))
+        hist_timestamps = hist_timestamps[:n_hist]
+        hist_states = hist_states[:n_hist]
+        hist_measurements = hist_measurements[:n_hist]
+        hist_state_status = hist_state_status[:n_hist]
+
         duration_sec = (self.death_timestamp - self.birth_timestamp) / 1000.0
         avg_snr = self.total_snr / max(self.n_associated, 1)
         continuity = self.n_associated / max(self.n_frames, 1)
@@ -835,11 +859,11 @@ def to_dict(self):
             "anomaly_types": list(self.anomaly_types),
             "anomaly_detections": self.anomaly_detections,
             "history": {
-                "timestamps": list(self.history["timestamps"]),
-                "states": [s.tolist() for s in self.history["states"]],
-                "delays": [m["delay"] if m else None for m in self.history["measurements"]],
-                "dopplers": [m["doppler"] if m else None for m in self.history["measurements"]],
-                "snrs": [m["snr"] if m else None for m in self.history["measurements"]],
-                "state_status": list(self.history["state_status"]),
+                "timestamps": hist_timestamps,
+                "states": [s.tolist() for s in hist_states],
+                "delays": [m["delay"] if m else None for m in hist_measurements],
+                "dopplers": [m["doppler"] if m else None for m in hist_measurements],
+                "snrs": [m["snr"] if m else None for m in hist_measurements],
+                "state_status": hist_state_status,
             },
         }

tests/test_history_snapshot.py

@@ -0,0 +1,86 @@
+"""Off-thread readers of `Track.history` must not see a mutating deque.
+
+The five history buffers are `deque(maxlen=TRACK_HISTORY_MAX)`.  Only the frame
+worker appends to a given node's tracker, but several readers run on other
+threads (the aircraft-feed flush executor, the analytics executor, the admin
+state-snapshot routes).  A deque raises `RuntimeError: deque mutated during
+iteration` if it is appended to while a Python-level loop walks it, which a
+plain list tolerated -- so `get_recent_detections` and `to_dict` must snapshot
+each buffer with `list()` (one C call, atomic under the GIL) before iterating.
+"""
+
+import threading
+
+from retina_tracker.config import get_config
+from retina_tracker.tracker import Tracker
+
+
+def make_detection(i):
+    return {"delay": 10.0 + i * 0.01, "doppler": 50.0, "snr": 20.0}
+
+
+def make_track():
+    """A live track with a real Kalman filter, via the normal Tracker path."""
+    tracker = Tracker(config=get_config())
+    ts = 0
+    for i in range(20):
+        tracker.process_frame([make_detection(i)], ts)
+        ts += 1000
+    return max(tracker.get_tracks(), key=lambda t: t.n_associated), ts
+
+
+def test_readers_survive_concurrent_appends():
+    track, start_ts = make_track()
+
+    errors = []
+    stop = threading.Event()
+
+    def appender():
+        ts = start_ts
+        i = track.n_frames
+        try:
+            while not stop.is_set():
+                ts += 1000
+                i += 1
+                if i % 5 == 0:
+                    track.mark_missed(ts, frame=i)
+                else:
+                    track.update(make_detection(i), ts, frame=i)
+        except Exception as exc:  # pragma: no cover - failure path
+            errors.append(("appender", exc))
+
+    writer = threading.Thread(target=appender, daemon=True)
+    writer.start()
+    try:
+        deadline = threading.Event()
+        timer = threading.Timer(0.5, deadline.set)
+        timer.start()
+        reads = 0
+        try:
+            while not deadline.is_set():
+                for _ in range(50):
+                    recent = track.get_recent_detections(n=5)
+                    for det in recent:
+                        assert isinstance(det, dict)
+                        assert det["timestamp"] is not None
+                        assert det["delay"] is not None
+                        assert det["doppler"] is not None
+                        assert det["snr"] is not None
+                    dumped = track.to_dict()
+                    history = dumped["history"]
+                    n = len(history["timestamps"])
+                    assert len(history["states"]) == n
+                    assert len(history["delays"]) == n
+                    assert len(history["dopplers"]) == n
+                    assert len(history["snrs"]) == n
+                    assert len(history["state_status"]) == n
+                    reads += 1
+        finally:
+            timer.cancel()
+    finally:
+        stop.set()
+        writer.join(timeout=2.0)
+
+    assert not errors, errors
+    assert reads > 100, f"stress loop did too little work to be meaningful ({reads} reads)"
+    assert track.n_frames > 100, f"appender did too little work to be meaningful ({track.n_frames} frames)"