fix: gate OBP source ensure when bridge source is already active

Skip ensure_obp_source_for_tg_store on OBP voice bursts when indexed
store lookup shows an ACTIVE TS1 source for the TG table, avoiding two
full subscription snapshots on every packet under concurrent OBP load.

src/adn_server/application/routing/obp_forward.py

@@ -81,9 +81,16 @@ class ObpForwardMixin:
             return
         if not (79 <= dst_int < 9990 or dst_int > 9999):
             return
-        from ..subscription.obp_source_ops import ensure_obp_source_for_tg_store
+        from ..subscription.obp_source_ops import (
+            ensure_obp_source_for_tg_store,
+            obp_source_needs_ensure,
+        )
+
+        store = self._subscription_store
+        if not obp_source_needs_ensure(store, system_name, relay_table_key, dst_int):
+            return
         ensure_obp_source_for_tg_store(
-            self._subscription_store,
+            store,
             system_name,
             relay_table_key,
             dst_id_b,

src/adn_server/application/subscription/obp_source_ops.py

@@ -48,6 +48,23 @@ def _tgid_match(entry_tgid: Any, dst_id_b: bytes, dst_int: int) -> bool:
         return False
 
 
+def obp_source_needs_ensure(
+    store: SubscriptionStore,
+    system_name: str,
+    relay_table_key: str,
+    dst_int: int,
+) -> bool:
+    """True when a bridge table exists but OBP lacks an ACTIVE TS1 source for ``dst_int``."""
+    active_tables = set(store.relay_tables_with_active_source(system_name, 1, dst_int))
+    pending_keys: list[str] = []
+    for key in (relay_table_key, "#" + relay_table_key):
+        if store.legs_in_table(key):
+            pending_keys.append(key)
+    if not pending_keys:
+        return False
+    return any(key not in active_tables for key in pending_keys)
+
+
 def ensure_obp_source_for_tg_store(
     store: SubscriptionStore,
     system_name: str,

tests/application/test_subscription_table_store_ops.py

@@ -24,7 +24,10 @@ from __future__ import annotations
 
 from tests.harness.deterministic import active_routing_table, minimal_config
 
-from adn_server.application.subscription.obp_source_ops import ensure_obp_source_for_tg_store
+from adn_server.application.subscription.obp_source_ops import (
+    ensure_obp_source_for_tg_store,
+    obp_source_needs_ensure,
+)
 from adn_server.application.subscription.store_sync import replace_store_from_routing_table
 from adn_server.application.subscription.subscription_table_ops import (
     ensure_dynamic_relay_store,
@@ -64,9 +67,22 @@ def test_ensure_obp_source_activates_ts1_leg() -> None:
         if entry["SYSTEM"] == "OBP-CL":
             entry["ACTIVE"] = False
     store = _store(bridges)
+    assert obp_source_needs_ensure(store, "OBP-CL", "7305", 7305) is True
     ensure_obp_source_for_tg_store(store, "OBP-CL", "7305", bytes_3(7305), 7305, now=500.0)
     obp = next(s for s in store.snapshot() if s.system.value == "OBP-CL")
     assert obp.is_active()
+    assert obp_source_needs_ensure(store, "OBP-CL", "7305", 7305) is False
+
+
+def test_obp_source_needs_ensure_false_when_no_bridge_table() -> None:
+    store = _store()
+    assert obp_source_needs_ensure(store, "OBP-CL", "7305", 7305) is False
+
+
+def test_obp_source_needs_ensure_false_when_obp_source_already_active() -> None:
+    bridges = active_routing_table(7305, (("OBP-CL", 1), ("MASTER-A", 2)))
+    store = _store(bridges)
+    assert obp_source_needs_ensure(store, "OBP-CL", "7305", 7305) is False
 
 
 def test_ensure_dynamic_relay_store_obp_leg() -> None: