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

Unit data to an individual ID (>= 1000000) was copied to every OpenBridge
(VER > 1) even when the destination was just heard on a local system and the
SUB_MAP lookup already delivers it there. A D-APRS gateway's ARS/LRRP answers
(one per minute per radio) went out on every configured bridge, multiplying
traffic and causing remote masters that mis-classify those frames to create 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 (900s).
Subscribers learned via an OpenBridge, or stale entries, keep the current
behavior, so a radio that moved to another master stays reachable.

Rescued from #72 (opened and self-closed by pyopower without a merge),
adapted to the current file layout.

src/adn_server/application/routing_use_cases.py

@@ -81,6 +81,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,
@@ -1238,8 +1242,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

@@ -26,6 +26,7 @@ import pytest
 from tests.harness.assertions import assert_forwarded, assert_not_forwarded
 from tests.harness.deterministic import (
     DeterministicScenario,
+    FakeClock,
     PacketSpec,
     add_openbridge_system,
     minimal_config,
@@ -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)."""