diff --git a/backend/core/state.py b/backend/core/state.py index 64c0093e..5b21d8d2 100644 --- a/backend/core/state.py +++ b/backend/core/state.py @@ -583,12 +583,15 @@ def _adsb_for_seeding() -> dict[str, dict]: solver_stale_drops: int = 0 # Candidates dequeued and skipped because every single-node track they carry -# was already solved within _SOLVER_RESOLVE_INTERVAL_S at no fewer nodes (see -# solver.py's _claim_resolve_slot). Association is per-node and rate-limited -# per node, so one aircraft arrives as one candidate per node that can see it; -# this counts the copies that were never worth solving. High against -# solver_successes is normal and is the mechanism working — it is -# solver_stale_drops that means work was lost. +# was already PUBLISHED within _SOLVER_RESOLVE_INTERVAL_S at no fewer nodes +# (see solver.py's _resolve_slot_covered). Association is per-node and +# rate-limited per node, so one aircraft arrives as one candidate per node +# that can see it; this counts the copies that were never worth solving. High +# against solver_successes is normal and is the mechanism working — it is +# solver_stale_drops that means work was lost. Read it against +# solver_successes, not against attempts: while the claim was taken on +# ADMISSION rather than on publication, a rejected candidate blacked out every +# later one sharing a track id and this counter ran at ~2.4x attempts. solver_resolve_skips: int = 0 # The dark-lane share of the counter above, split out because the two lanes diff --git a/backend/routes/test.py b/backend/routes/test.py index 5909ff9d..c9431cd2 100644 --- a/backend/routes/test.py +++ b/backend/routes/test.py @@ -210,7 +210,7 @@ def _build_dashboard_data() -> bytes: # not the queue size. "stale_drops": state.solver_stale_drops, # Duplicate candidates for an aircraft already solved this - # window (see solver.py's _claim_resolve_slot). Read it + # window (see solver.py's _resolve_slot_covered). Read it # against stale_drops: skips are work correctly not done, # stale drops are work lost. "resolve_skips": state.solver_resolve_skips, diff --git a/backend/services/tasks/solver.py b/backend/services/tasks/solver.py index a3a80205..2b1a6d9b 100644 --- a/backend/services/tasks/solver.py +++ b/backend/services/tasks/solver.py @@ -1075,6 +1075,34 @@ def _supersession_match( # Sized against the map, not the association cadence: multinode_tracks expire # at 60 s, so refreshing an aircraft every 12 s leaves four solves' worth of # margin. 0 disables the suppression entirely. +# +# The claim is recorded ON PUBLICATION, not on admission, and from the +# POST-TRIM survivors. Claiming on admission made a candidate that never +# reached the map suppress every later candidate sharing any of its track ids +# for the full window — including other aircraft's, since tracker track ids +# are shared across the association candidates of different aircraft (74 of +# 178 ids in a 6 min live window appeared in solves of more than one +# ground-truth aircraft; the same finding that forced _supersession_match's +# spatial guard). A rejected candidate, or a contaminated superset that the +# gates sank, therefore blacked out the clean subsets behind it for 12 s and +# nothing was refreshed at all. Live that cost ~1 537 skips per 646 dark +# attempts per 30 min — more candidates suppressed than solved, by a factor +# of two. The rule this suppression is FOR is "an aircraft already on the map +# at this width does not need re-solving yet", and only a publication puts an +# aircraft on the map. +# +# Two consequences, both accepted deliberately: +# * the check no longer claims under the same lock, so two workers can now +# both solve duplicates of one aircraft that arrived together. The pair +# costs one extra solve and is resolved downstream by keying and +# supersession, which already handle exactly this; the alternative is the +# starvation above. +# * trimmed nodes' track ids are NOT claimed (_filter_s_in_to_nodes rebuilds +# track_ids from the surviving track_ids_by_node, so result's +# source_track_ids are the survivors). A node dropped for a bad residual +# was probably another aircraft's — claiming its track would suppress that +# aircraft's own candidate on the strength of a measurement this solve +# threw away. _SOLVER_RESOLVE_INTERVAL_S = float(os.getenv("SOLVER_RESOLVE_INTERVAL_S", "12")) _RECENT_SOLVES: dict[str, tuple[float, int]] = {} # track_id → (solved_at, n_nodes) _RECENT_SOLVES_LOCK = threading.Lock() @@ -1092,61 +1120,60 @@ def _sweep_recent_solves(now_s: float) -> None: del _RECENT_SOLVES[tid] -def _claim_resolve_slot(s_in, now_s: float) -> bool: - """False when this candidate re-solves tracks another candidate just took. +def _resolve_slot_covered(s_in, now_s: float) -> tuple[bool, list[dict]]: + """Is every track this candidate carries already ON THE MAP at this width? - Records the claim as a side effect, under one lock with the test, so two - workers cannot both admit the same aircraft's duplicates. An input with no - track provenance (detection-level, or an anchored input carrying none) is - always admitted — there is nothing to match it against. + Pure: it reads the claims and mutates nothing, so a candidate that is + admitted here and then rejected by the gate stack leaves no trace. The + claim is made afterwards by _record_resolve_slot, from the publish path + only — see the block comment above for why, and for what the loss of + atomic test-and-claim costs. + + Returns (covered, blocking). ``blocking`` is the claims that covered it, + for the skip record; it is empty whenever ``covered`` is False. An input + with no track provenance (detection-level, or an anchored input carrying + none) is never covered — there is nothing to match it against. """ if _SOLVER_RESOLVE_INTERVAL_S <= 0 or not isinstance(s_in, dict): - return True + return False, [] track_ids = s_in.get("track_ids") if not track_ids: - return True + return False, [] n_nodes = int(s_in.get("n_nodes") or 0) cutoff = now_s - _SOLVER_RESOLVE_INTERVAL_S + blocking: list[dict] = [] with _RECENT_SOLVES_LOCK: - covered = True for tid in track_ids: held = _RECENT_SOLVES.get(tid) if held is None or held[0] <= cutoff or held[1] < n_nodes: - covered = False - break - if covered: - return False - for tid in track_ids: - held = _RECENT_SOLVES.get(tid) - # Keep the widest claim of the window: a narrow candidate admitted - # after a wide one must not lower the bar the next copy is tested - # against. - held_nodes = held[1] if held is not None and held[0] > cutoff else 0 - _RECENT_SOLVES[tid] = (now_s, max(n_nodes, held_nodes)) - _sweep_recent_solves(now_s) - return True + return False, [] + blocking.append({"track_id": tid, "held_ts": round(held[0], 3), "held_n": held[1]}) + return True, blocking -def _resolve_slot_blockers(track_ids, now_s: float) -> list[dict]: - """The live claims covering ``track_ids``, for a skip record. +def _record_resolve_slot(track_ids, n_nodes: int, now_s: float) -> None: + """Record that ``track_ids`` are covered by a PUBLISHED solve at n_nodes. - Read-only, and taken after the refusal rather than during it: the check - itself must stay one atomic test-and-claim, and a skip is rare enough - (relative to the queue drain rate) that a second lock acquisition on that - path costs nothing. Any claim that moves between the two is a claim the - diagnosis would have wanted to name anyway. + Called from the publish path alone, with the post-trim survivors + (``result["source_track_ids"]``). Nothing else may call it: a claim is a + statement that this aircraft is on the map, and a rejected solve puts + nothing there. """ + if _SOLVER_RESOLVE_INTERVAL_S <= 0 or not track_ids: + return + n_nodes = int(n_nodes or 0) cutoff = now_s - _SOLVER_RESOLVE_INTERVAL_S - out: list[dict] = [] with _RECENT_SOLVES_LOCK: for tid in track_ids: held = _RECENT_SOLVES.get(tid) - if held is not None and held[0] > cutoff: - out.append({"track_id": tid, "held_ts": round(held[0], 3), "held_n": held[1]}) - return out + # Keep the widest claim of the window: a narrow publish after a + # wide one must not lower the bar the next copy is tested against. + held_nodes = held[1] if held is not None and held[0] > cutoff else 0 + _RECENT_SOLVES[tid] = (now_s, max(n_nodes, held_nodes)) + _sweep_recent_solves(now_s) -def _record_resolve_skip(s_in, now_s: float, blocking: list[dict] | None = None) -> None: +def _record_resolve_skip(s_in, now_s: float, blocking: list[dict]) -> None: """Count and remember one resolve-slot refusal. The counter alone could not answer the question the suppression rule @@ -1178,7 +1205,7 @@ def _record_resolve_skip(s_in, now_s: float, blocking: list[dict] | None = None) "lane": "dark" if dark else "adsb", "track_ids": track_ids, "n_nodes": int(s.get("n_nodes") or 0), - "blocking": _resolve_slot_blockers(track_ids, now_s) if blocking is None else blocking, + "blocking": blocking, "guess_lat": round(float(ig["lat"]), 6) if ig.get("lat") else None, "guess_lon": round(float(ig["lon"]), 6) if ig.get("lon") else None, } @@ -1948,8 +1975,9 @@ def _process_solver_item( # and a copy that queued before its twin was solved can only be recognised # once it reaches a worker. _now_s = time.time() - if not _claim_resolve_slot(s_in, _now_s): - _record_resolve_skip(s_in, _now_s) + _covered, _blocking = _resolve_slot_covered(s_in, _now_s) + if _covered: + _record_resolve_skip(s_in, _now_s, _blocking) return None n_nodes = s_in.get("n_nodes", 0) if isinstance(s_in, dict) else 0 consensus_meta: dict | None = None @@ -2529,6 +2557,12 @@ def _process_solver_item( archive_record = dict(result) archive_record["solve_ts_ms"] = int(time.time() * 1000) state.track_archive_buffer.append(archive_record) + # The re-solve claim, taken here and nowhere else: this aircraft is now + # on the map at this width, which is the only thing that makes a + # duplicate not worth solving. Survivors only — source_track_ids is + # rebuilt from the post-trim node set. Outside _MN_TRACKS_LOCK on + # purpose, so _RECENT_SOLVES_LOCK is never nested inside it. + _record_resolve_slot(result.get("source_track_ids"), result.get("n_nodes"), time.time()) _record_solve_history( "published", s_in, diff --git a/backend/tests/test_mlat_history.py b/backend/tests/test_mlat_history.py index e8e91bbe..a406d1b4 100644 --- a/backend/tests/test_mlat_history.py +++ b/backend/tests/test_mlat_history.py @@ -1031,9 +1031,10 @@ def _client(self): def _skip(self, track_ids=("a1", "b1"), n_nodes=3, **s_in): now = time.time() s = dict(_CONFIRMED_N2, n_nodes=n_nodes, track_ids=list(track_ids), **s_in) - solver_mod._claim_resolve_slot(s, now) - assert solver_mod._claim_resolve_slot(dict(s), now) is False - solver_mod._record_resolve_skip(dict(s), now) + solver_mod._record_resolve_slot(list(track_ids), n_nodes, now) + covered, blocking = solver_mod._resolve_slot_covered(dict(s), now) + assert covered is True + solver_mod._record_resolve_skip(dict(s), now, blocking) def test_skip_records_the_blocking_claim(self): self._skip() diff --git a/backend/tests/test_solver_trimming.py b/backend/tests/test_solver_trimming.py index 93504a5a..0f529848 100644 --- a/backend/tests/test_solver_trimming.py +++ b/backend/tests/test_solver_trimming.py @@ -827,3 +827,76 @@ def solve_fn(_s_in, _cfgs): assert state.fov_shadow_agree == 1 assert state.fov_shadow_would_pass == 0 assert state.fov_shadow_would_reject == 0 + + +class TestTrimmedTracksAreNotClaimed(_TrimmingTestBase): + """A trimmed node's tracks must not take a re-solve claim. + + The claim says "this aircraft is on the map at this width". A node + dropped for a bad residual contributed nothing to the published position + and its track was probably a different aircraft's — claiming it would + suppress that aircraft's own candidate on the strength of a measurement + this solve threw away. + """ + + _FULL = ["n1", "n2", "n3", "n4", "bad"] + _TRIM = ["n1", "n2", "n3", "n4"] + + def test_the_dropped_nodes_track_is_left_unclaimed(self): + table = { + frozenset(self._FULL): _stub_result( + self._FULL, + rms_delay=8.0, + per_node={"n1": 0.5, "n2": 0.5, "n3": 0.5, "n4": 0.5, "bad": 12.0}, + ), + frozenset(self._TRIM): _stub_result( + self._TRIM, + rms_delay=0.8, + per_node={"n1": 0.3, "n2": 0.3, "n3": 0.3, "n4": 0.3}, + ), + } + s_in = _s_in( + self._FULL, + track_ids=["t1", "t2", "t3", "t4", "tbad"], + track_ids_by_node={ + "n1": ["t1"], + "n2": ["t2"], + "n3": ["t3"], + "n4": ["t4"], + "bad": ["tbad"], + }, + ) + result = self._run(s_in, _stub_solve_fn(table)) + assert result is not None and result["success"] + assert result["source_track_ids"] == ["t1", "t2", "t3", "t4"] + assert set(solver_mod._RECENT_SOLVES) == {"t1", "t2", "t3", "t4"} + + def test_a_candidate_built_on_the_dropped_track_still_runs(self): + """The other half of the same claim: whoever "tbad" really belongs to + keeps its slot.""" + table = { + frozenset(self._FULL): _stub_result( + self._FULL, + rms_delay=8.0, + per_node={"n1": 0.5, "n2": 0.5, "n3": 0.5, "n4": 0.5, "bad": 12.0}, + ), + frozenset(self._TRIM): _stub_result( + self._TRIM, + rms_delay=0.8, + per_node={"n1": 0.3, "n2": 0.3, "n3": 0.3, "n4": 0.3}, + ), + } + s_in = _s_in( + self._FULL, + track_ids=["t1", "t2", "t3", "t4", "tbad"], + track_ids_by_node={ + "n1": ["t1"], + "n2": ["t2"], + "n3": ["t3"], + "n4": ["t4"], + "bad": ["tbad"], + }, + ) + self._run(s_in, _stub_solve_fn(table)) + neighbour = {"n_nodes": 2, "track_ids": ["tbad", "tother"]} + assert solver_mod._resolve_slot_covered(neighbour, time.time())[0] is False diff --git a/backend/tests/test_solver_worker.py b/backend/tests/test_solver_worker.py index a4a6c8e9..6e3a2035 100644 --- a/backend/tests/test_solver_worker.py +++ b/backend/tests/test_solver_worker.py @@ -335,56 +335,153 @@ class TestResolveSuppression: emits its own candidate for it inside one association window. Solving all of them starves aircraft that have no solve at all — the queue ages out behind work whose result is superseded the moment it lands. + + The claim that suppresses a duplicate is taken on PUBLICATION + (_record_resolve_slot), not on admission: the rule is "this aircraft is + already on the map at this width", and only a publish puts it there. + _resolve_slot_covered is the pure test run before the solve. """ def _s_in(self, track_ids, n_nodes=2): return dict(_CONFIRMED_N2, n_nodes=n_nodes, track_ids=list(track_ids)) - def test_a_second_copy_of_the_same_tracks_is_skipped(self): + def _covered(self, track_ids, n_nodes=2, now=None): + return solver_mod._resolve_slot_covered(self._s_in(track_ids, n_nodes), now or time.time())[0] + + def _publish(self, track_ids, n_nodes=2, now=None): + solver_mod._record_resolve_slot(list(track_ids), n_nodes, now or time.time()) + + def test_a_second_copy_of_a_published_candidate_is_skipped(self): + now = time.time() + assert self._covered(["a1", "b1"], now=now) is False + self._publish(["a1", "b1"], now=now) + assert self._covered(["a1", "b1"], now=now) is True + + def test_the_check_alone_claims_nothing(self): + """The whole point of the split: a candidate that is admitted and then + rejected by the gate stack must leave no trace.""" now = time.time() - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"]), now) is True - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"]), now) is False + assert self._covered(["a1", "b1"], now=now) is False + assert self._covered(["a1", "b1"], now=now) is False - def test_a_candidate_carrying_an_unsolved_track_runs(self): + def test_a_candidate_carrying_an_unpublished_track_runs(self): """An aircraft entering coverage must never be suppressed.""" now = time.time() - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"]), now) is True - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b2"]), now) is True + self._publish(["a1", "b1"], now=now) + assert self._covered(["a1", "b2"], now=now) is False def test_a_wider_view_of_the_same_tracks_runs(self): now = time.time() - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"], n_nodes=2), now) is True - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"], n_nodes=5), now) is True + self._publish(["a1", "b1"], n_nodes=2, now=now) + assert self._covered(["a1", "b1"], n_nodes=5, now=now) is False def test_a_narrower_view_after_a_wider_one_is_skipped(self): now = time.time() - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"], n_nodes=5), now) is True - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"], n_nodes=2), now) is False + self._publish(["a1", "b1"], n_nodes=5, now=now) + assert self._covered(["a1", "b1"], n_nodes=2, now=now) is True - def test_a_narrow_admission_does_not_lower_the_bar(self): + def test_a_narrow_publish_does_not_lower_the_bar(self): """The window holds the widest claim, not the most recent one.""" now = time.time() - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"], n_nodes=5), now) is True - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b2"], n_nodes=2), now) is True - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"], n_nodes=3), now) is False + self._publish(["a1", "b1"], n_nodes=5, now=now) + self._publish(["a1", "b2"], n_nodes=2, now=now) + assert self._covered(["a1", "b1"], n_nodes=3, now=now) is True def test_claims_expire(self): now = time.time() - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"]), now) is True + self._publish(["a1", "b1"], now=now) later = now + solver_mod._SOLVER_RESOLVE_INTERVAL_S + 1.0 - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"]), later) is True + assert self._covered(["a1", "b1"], now=later) is False def test_an_input_without_track_provenance_always_runs(self): """Detection-level inputs carry no track ids — nothing to match on.""" now = time.time() - assert solver_mod._claim_resolve_slot({"n_nodes": 2}, now) is True - assert solver_mod._claim_resolve_slot({"n_nodes": 2}, now) is True + assert solver_mod._resolve_slot_covered({"n_nodes": 2}, now)[0] is False + solver_mod._record_resolve_slot(None, 2, now) + assert solver_mod._resolve_slot_covered({"n_nodes": 2}, now)[0] is False def test_zero_interval_disables_suppression(self, monkeypatch): monkeypatch.setattr(solver_mod, "_SOLVER_RESOLVE_INTERVAL_S", 0.0) now = time.time() - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"]), now) is True - assert solver_mod._claim_resolve_slot(self._s_in(["a1", "b1"]), now) is True + self._publish(["a1", "b1"], now=now) + assert self._covered(["a1", "b1"], now=now) is False + + def test_the_check_names_every_blocking_claim(self): + now = time.time() + self._publish(["a1", "b1"], n_nodes=4, now=now) + covered, blocking = solver_mod._resolve_slot_covered(self._s_in(["a1", "b1"], n_nodes=3), now) + assert covered is True + assert {b["track_id"]: b["held_n"] for b in blocking} == {"a1": 4, "b1": 4} + + def test_an_admitted_candidate_reports_no_blockers(self): + covered, blocking = solver_mod._resolve_slot_covered(self._s_in(["a1", "b1"]), time.time()) + assert (covered, blocking) == (False, []) + + def _solve_fn(self, calls, rms_delay=0.5, lat=37.5, lon=-122.1): + def fn(s_in, cfgs): + calls.append(s_in) + return { + "success": True, + "lat": lat, + "lon": lon, + "alt_m": 9000.0, + "rms_delay": rms_delay, + "rms_doppler": 5.0, + "timestamp_ms": int(time.time() * 1000), + "contributing_node_ids": ["n1", "n2"], + "n_nodes": s_in.get("n_nodes", 2), + } + + return fn + + def test_a_rejected_candidate_does_not_block_an_identical_twin(self, monkeypatch): + """The bug this split exists to fix. A candidate the gate stack sank + put nothing on the map, so the next copy of the same aircraft is its + first real chance — and used to be blacked out for the full 12 s.""" + _reset_state() + monkeypatch.setattr(state, "node_analytics", _StubAnalytics()) + calls: list = [] + s_in = self._s_in(["a1", "b1"]) + + # rms_delay past the gate: solves, then rejected, publishes nothing. + solver_mod._process_solver_item((dict(s_in), {}, time.time()), self._solve_fn(calls, rms_delay=10.0)) + assert state.solver_fail_rms_delay == 1 + assert not state.multinode_tracks + + solver_mod._process_solver_item((dict(s_in), {}, time.time()), self._solve_fn(calls)) + assert len(calls) == 2, "the twin must not be suppressed by a reject" + assert state.multinode_tracks + assert state.solver_resolve_skips == 0 + + def test_a_subset_for_another_aircraft_survives_a_rejected_superset(self, monkeypatch): + """Tracker track ids are shared across the candidates of DIFFERENT + aircraft, so a contaminated superset that the gates sank used to take + every clean subset behind it down with it — including its neighbour's + only candidate.""" + _reset_state() + monkeypatch.setattr(state, "node_analytics", _StubAnalytics()) + calls: list = [] + superset = self._s_in(["a1", "b1", "c1"], n_nodes=3) + solver_mod._process_solver_item((superset, {}, time.time()), self._solve_fn(calls, rms_delay=10.0)) + assert not state.multinode_tracks + + # The neighbour: fewer nodes, sharing one contaminated track id. + subset = self._s_in(["a1", "b1"], n_nodes=2) + solver_mod._process_solver_item((subset, {}, time.time()), self._solve_fn(calls)) + assert len(calls) == 2 + assert state.multinode_tracks + + def test_only_the_published_width_is_claimed(self, monkeypatch): + """A publish claims at the width it published, so a later narrower + copy is suppressed and a wider one still runs.""" + _reset_state() + monkeypatch.setattr(state, "node_analytics", _StubAnalytics()) + calls: list = [] + solver_mod._process_solver_item((self._s_in(["a1", "b1"], n_nodes=3), {}, time.time()), self._solve_fn(calls)) + assert state.multinode_tracks + now = time.time() + assert self._covered(["a1", "b1"], n_nodes=2, now=now) is True + assert self._covered(["a1", "b1"], n_nodes=4, now=now) is False def test_a_skipped_item_never_reaches_the_solver(self, monkeypatch): _reset_state() @@ -408,6 +505,8 @@ def solve_fn(s_in, cfgs): solver_mod._process_solver_item((s_in, {}, time.time()), solve_fn) assert len(solve_calls) == 1 assert state.solver_successes == 1 + # The first item PUBLISHED, which is what makes the second redundant. + assert state.multinode_tracks assert solver_mod._process_solver_item((dict(s_in), {}, time.time()), solve_fn) is None assert len(solve_calls) == 1, "the duplicate must not be solved" @@ -425,9 +524,10 @@ def test_a_skip_is_recorded_with_the_claim_that_blocked_it(self): state.solver_resolve_skips_recent.clear() now = time.time() s_in = dict(self._s_in(["a1", "b1"], n_nodes=4), initial_guess={"lat": 35.0, "lon": -82.0}) - assert solver_mod._claim_resolve_slot(dict(s_in), now) is True - assert solver_mod._claim_resolve_slot(dict(s_in), now) is False - solver_mod._record_resolve_skip(dict(s_in), now) + solver_mod._record_resolve_slot(["a1", "b1"], 4, now) + covered, blocking = solver_mod._resolve_slot_covered(dict(s_in), now) + assert covered is True + solver_mod._record_resolve_skip(dict(s_in), now, blocking) assert state.solver_resolve_skips == 1 assert state.solver_resolve_skips_dark == 1 @@ -444,7 +544,7 @@ def test_a_tagged_candidate_is_counted_but_not_as_dark(self): state.solver_resolve_skips_recent.clear() now = time.time() s_in = dict(self._s_in(["a1"], n_nodes=3), adsb_hex="abc123") - solver_mod._record_resolve_skip(s_in, now) + solver_mod._record_resolve_skip(s_in, now, []) assert state.solver_resolve_skips == 1 assert state.solver_resolve_skips_dark == 0 assert state.solver_resolve_skips_recent[0]["lane"] == "adsb" @@ -455,7 +555,7 @@ def test_skips_never_enter_the_solve_history(self): _reset_state() state.mlat_solve_history.clear() s_in = self._s_in(["a1", "b1"]) - solver_mod._record_resolve_skip(s_in, time.time()) + solver_mod._record_resolve_skip(s_in, time.time(), []) assert not state.mlat_solve_history assert not state.mlat_solve_history_known diff --git a/docs/solverflow.md b/docs/solverflow.md index 980e577a..db3bdedd 100644 --- a/docs/solverflow.md +++ b/docs/solverflow.md @@ -384,13 +384,30 @@ The centerpiece: every candidate from either lane, once dequeued from any gate stops the chain, bumps a counter, and (from 6.5 onward) writes a named record to solve history. +**6.2 claims on publication, not on admission.** The suppression rule is +"this aircraft is already on the map at this width, at every track it is +built from" — so `_resolve_slot_covered` only *reads* the claims, and +`_record_resolve_slot` takes them from the publish path, with the +**post-trim survivors** (`result["source_track_ids"]`, rebuilt from the +surviving `track_ids_by_node`). Claiming on admission instead meant a +candidate that never reached the map still blacked out every later candidate +sharing any of its track ids for the full 12 s — including *other aircraft's*, +since tracker track ids are shared across the association candidates of +different aircraft (the same finding behind `_supersession_match`'s spatial +guard; see Caveats). Live that ran at ~1 537 skips per 646 dark attempts per +30 min: more candidates suppressed than solved, by a factor of two. The price +of the split is that the check no longer claims under the same lock, so two +workers can now both solve duplicates that arrived together; that costs one +extra solve and is resolved downstream by keying and supersession, which +handle exactly this case already. + ```mermaid flowchart TD deq["Dequeue (s_in, node_cfgs, enqueued_at)"] deq --> g61{"6.1 Staleness
age_s > _SOLVER_MAX_QUEUE_AGE_S 45.0s?"} g61 -->|"yes"| f61["solver_stale_drops
(no history record)"]:::inert - g61 -->|"no"| g62{"6.2 Re-solve suppression
_claim_resolve_slot False?"} - g62 -->|"yes"| f62["solver_resolve_skips"]:::inert + g61 -->|"no"| g62{"6.2 Re-solve suppression
_resolve_slot_covered (pure)?"} + g62 -->|"yes"| f62["solver_resolve_skips (+_dark)
+ skip record with blockers"]:::inert g62 -->|"no"| g63["6.3 Solve dispatch:
no guess -> bare solve_fn;
n>=3 -> consensus? then
_solve_best_altitude (sweep);
n=2 -> _solve_best_altitude_n2
(single altitude)"] g63 -->|"exception"| f63["solver_failures +
solver_fail_exception,
result=None"]:::inert g63 --> g64{"6.4 Trim & resolve (recovery):
guess AND n>=4 AND
rms_delay > 3.0us?"} @@ -519,7 +536,7 @@ mode per environment and read the two lanes' `rms_delay` and `gt_error_km` off | Constant | Value | Defined in | |---|---|---| | `_SOLVER_MAX_QUEUE_AGE_S` (6.1) | 45.0 s | `services/tasks/solver.py` | -| `SOLVER_RESOLVE_INTERVAL_S` (6.2) | 12 s (0 disables) | `services/tasks/solver.py` (`_SOLVER_RESOLVE_INTERVAL_S`) | +| `SOLVER_RESOLVE_INTERVAL_S` (6.2) | 12 s (0 disables) | `services/tasks/solver.py` (`_SOLVER_RESOLVE_INTERVAL_S`, `_resolve_slot_covered`, `_record_resolve_slot`) | | `_TRIM_MAX_ROUNDS` / `_TRIM_RESID_FACTOR` / `_TRIM_MIN_NODES` (6.4) | 4 / 1.5 / 3 | `services/tasks/solver.py` | | `SOLVER_RMS_DELAY_MAX_US` (6.5) | 3.0 us | `services/tasks/solver.py` (`_SOLVER_RMS_DELAY_MAX_US`) | | `_SOLVER_RMS_DOPPLER_MAX_HZ` (6.6) | 200.0 Hz (hardcoded) | `services/tasks/solver.py` | @@ -564,7 +581,8 @@ flowchart TD popped --> store["state.multinode_tracks[key] = result"] blocked --> store store --> archive["track-archive buffer append"] - archive --> histpub["_record_solve_history: published"] + archive --> claimslot["_record_resolve_slot:
claim the POST-TRIM survivors
for _SOLVER_RESOLVE_INTERVAL_S"] + claimslot --> histpub["_record_solve_history: published"] histpub --> feed["build_combined_aircraft_json
(1 Hz flush)"] feed --> gN2{"n=2 display gate:
solve_count < MN_N2_MIN_SOLVES 2?"} @@ -628,7 +646,7 @@ are all the DARK lane; `lane_split` gives the per-lane record counts and | Block | Says | Watch for | |---|---|---| | `contamination` | Of the dark records that matched ground truth, how many carried a node that could not see the aircraft (`foreign_node_ids` on the record; verdict is the associator's own `_point_in_beam`, the same gate known-lane claiming uses) | `pct` is the live version of the offline ~60 % the cluster-splitting work exists to move. Records with no GT match, or no registered geometry for any contributing node, are **out of the denominator** — abstention, not innocence | -| `resolve_skips` | Candidates the re-solve suppression refused in this window, from `state.solver_resolve_skips_recent`, with the claims that blocked each one | `attempts_ratio` is all-lane skips over DARK attempts (live baseline ~2.4). The deque holds 500 entries against ~50 skips/min, so read `window_effective_minutes` before reading `total` as a window count | +| `resolve_skips` | Candidates the re-solve suppression refused in this window, from `state.solver_resolve_skips_recent`, with the claims that blocked each one | `attempts_ratio` is all-lane skips over DARK attempts. It read ~2.4 while 6.2 claimed on admission; with the claim on publication it should sit at or below 0.5. The deque holds 500 entries against a live rate of tens per minute, so read `window_effective_minutes` before reading `total` as a window count | | `counters.resolve_skips_dark` | Dark share of the since-boot skip counter | — | | `counters.node_frames_rate_limited` | Frames `NODE_FRAME_MIN_INTERVAL_S` refused before the tracker saw them (Gate B in §2) | Not the same event as `/api/admin/metrics`' `frames_dropped`, which is `frame_queue` saturation and normally reads zero |