Merge pull request #47 from ce5rpy/fix/obp-source-ensure-gate

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

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: