fix(routing): don't fan out unit data to OBPs for a fresh local subscriber

Unit data to an individual ID (>= 1000000) is copied to every OpenBridge
(VER > 1) even when the destination was just heard on a local system and the
SUB_MAP lookup below already delivers it here.

Seen in the field with a D-APRS gateway and a local MOTOTRBO behind a hotspot:
every ARS/LRRP answer of the gateway (one per minute per radio) went out on all
6 bridges of the master (478 copies in 5 minutes for a single radio), and remote
masters that mis-classify those frames ended up creating a dynamic talkgroup
named after the radio id.

Skip the OBP fan-out when the SUB_MAP entry for the destination is on a
non-OPENBRIDGE system and younger than UNIT_DATA_LOCAL_SUB_MAX_AGE (900 s).
Subscribers learned via an OpenBridge, or stale entries, keep the current
behaviour, so a radio that moved to another master is still reachable.

Tests: fresh local sub -> no OBP copy (fails without the fix); stale local sub
and sub learned via OBP -> still fanned out. Full suite: 753 passed (the only
failure, test_udp_rcvbuf, is environment-specific and fails on master too).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Jumj5dWY3YwV9yP6YdFE6n

src/adn_server/application/routing_use_cases.py

@@ -69,6 +69,10 @@ from .talker_alias_use_cases import TalkerAliasUseCases
 
 logger = logging.getLogger(__name__)
 
+# A SUB_MAP entry younger than this (seconds) on a non-OPENBRIDGE system means the unit is
+# served by this server, so unit data to it is not fanned out to OpenBridges.
+UNIT_DATA_LOCAL_SUB_MAX_AGE = 900.0
+
 
 class RoutingUseCases(
     RoutingTimerMixin,
@@ -1144,8 +1148,21 @@ class RoutingUseCases(
                 )
 
         # Fan-out to all OBP systems with VER > 1 and dst_id >= 1000000 (legacy ~2286-2295 / ~3088-3097)
+        # ...unless the destination is a subscriber recently seen on a LOCAL (non-OPENBRIDGE)
+        # system: SUB_MAP below already delivers it here. Flooding every OpenBridge with
+        # e.g. each ARS/LRRP answer of a D-APRS gateway to a local MOTOTRBO multiplies
+        # traffic by the number of bridges, and remote masters that mis-file those frames
+        # create a dynamic talkgroup named after the radio id.
         protocols = self._get_protocols() if self._get_protocols else {}
+        _local_sub = self._config.get("_SUB_MAP", {}).get(dst_id)
+        _dst_is_fresh_local_sub = bool(
+            _local_sub
+            and systems_cfg.get(_local_sub[0], {}).get("MODE") != "OPENBRIDGE"
+            and pkt_time - _local_sub[2] < UNIT_DATA_LOCAL_SUB_MAX_AGE
+        )
         for sys_name, sys_cfg in systems_cfg.items():
+            if _dst_is_fresh_local_sub:
+                break
             if sys_name == system_name:
                 continue
             if sys_name == "DATA-GATEWAY":

tests/routing/test_unit_data_routing.py

@@ -31,6 +31,7 @@ from tests.harness.deterministic import (
     minimal_config,
     parse_dmr_fields,
     patch_routing_wall_time,
+    FakeClock,
 )
 from tests.routing.unit_data_helpers import idle_hbp_slot
 
@@ -39,6 +40,7 @@ from adn_server.domain.hbp_protocol import HBPF_SLT_VHEAD, HBPF_SLT_VTERM
 
 DAPRS_GATEWAY_ID = 900999
 HOTSPOT_SUB_ID = 7300392
+_HARNESS_T0 = FakeClock().time()
 
 
 @pytest.mark.behavior
@@ -167,6 +169,53 @@ def test_unit_data_fanout_to_other_obp_with_ver_gt_1() -> None:
     assert parse_dmr_fields(scenario.capture.for_system("OBP-FAN")[0].packet)["call_type"] == "unit"
 
 
+@pytest.mark.behavior
+def test_unit_data_to_fresh_local_sub_is_not_fanned_out_to_obp() -> None:
+    """Unit data to a subscriber recently seen on a local system goes only via SUB_MAP."""
+    config = minimal_config(("D-APRS", "SYSTEM"))
+    add_openbridge_system(config, "OBP-FAN")
+    config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("SYSTEM", 2, _HARNESS_T0 - 60)}
+    scenario = DeterministicScenario(config=config)
+    scenario.protocols["SYSTEM"].STATUS[2] = idle_hbp_slot()
+    base = PacketSpec(call_type="unit", rf_src=DAPRS_GATEWAY_ID, dst_id=HOTSPOT_SUB_ID, stream_id=0x67676767, slot=2)
+
+    with patch_routing_wall_time(scenario.clock):
+        scenario.inject_unit("D-APRS", DeterministicScenario.unit_data_header_spec(base))
+
+    assert len(scenario.capture.for_system("OBP-FAN")) == 0
+    assert len(scenario.capture.for_system("SYSTEM")) >= 1
+
+
+@pytest.mark.behavior
+def test_unit_data_to_stale_local_sub_is_still_fanned_out_to_obp() -> None:
+    config = minimal_config(("D-APRS", "SYSTEM"))
+    add_openbridge_system(config, "OBP-FAN")
+    config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("SYSTEM", 2, _HARNESS_T0 - 3600)}
+    scenario = DeterministicScenario(config=config)
+    scenario.protocols["SYSTEM"].STATUS[2] = idle_hbp_slot()
+    base = PacketSpec(call_type="unit", rf_src=DAPRS_GATEWAY_ID, dst_id=HOTSPOT_SUB_ID, stream_id=0x68686868, slot=2)
+
+    with patch_routing_wall_time(scenario.clock):
+        scenario.inject_unit("D-APRS", DeterministicScenario.unit_data_header_spec(base))
+
+    assert len(scenario.capture.for_system("OBP-FAN")) == 1
+
+
+@pytest.mark.behavior
+def test_unit_data_to_sub_learned_via_obp_is_still_fanned_out_to_obp() -> None:
+    config = minimal_config(("D-APRS",))
+    add_openbridge_system(config, "OBP-FAN")
+    add_openbridge_system(config, "OBP-OTHER")
+    config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("OBP-OTHER", 1, _HARNESS_T0 - 60)}
+    scenario = DeterministicScenario(config=config)
+    base = PacketSpec(call_type="unit", rf_src=DAPRS_GATEWAY_ID, dst_id=HOTSPOT_SUB_ID, stream_id=0x69696969, slot=2)
+
+    with patch_routing_wall_time(scenario.clock):
+        scenario.inject_unit("D-APRS", DeterministicScenario.unit_data_header_spec(base))
+
+    assert len(scenario.capture.for_system("OBP-FAN")) == 1
+
+
 @pytest.mark.behavior
 def test_unit_data_daprs_peer_id_match_forwards() -> None:
     """Legacy parity: unit data routes when dst matches a PEER on D-APRS (runtime or static)."""