refactor(routing): address review on the forward plan cache

- State the invariant the plan cache relies on: from a SYSTEMS block the
  plan reads only MODE, which only a reload changes, so block identity is
  enough; anything else it reads must join the cache key.
- Build the plan's ingress with only what resolve() routes on, instead of
  passing empty peer, radio and stream ids through _voice_forward_plan.
- SubscriptionStore.has_table is abstract; the scan default is gone.
- Both caches are created in RoutingUseCases.__init__, and a full cache
  drops its oldest entry instead of all of them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/adn_server/application/ports.py

@@ -298,9 +298,10 @@ class SubscriptionStore(ABC):
     def legs_in_table(self, table_key: str) -> tuple["Subscription", ...]:
         ...
 
+    @abstractmethod
     def has_table(self, table_key: str) -> bool:
-        """True when at least one leg belongs to ``table_key``. Stores with an index override this."""
-        return any(sub.table_key() == table_key for sub in self.snapshot())
+        """True when at least one leg belongs to ``table_key``; runs per datagram."""
+        ...
 
     @abstractmethod
     def has_active_target_leg(self, system: str, slot: int, tgid: int) -> bool:

src/adn_server/application/routing/lc_ta.py

@@ -65,6 +65,8 @@ _LC_SET_CACHE_MAX = 512
 class LcTaMixin:
     """Talker Alias DMRA relay and embedded LC overlay on forward legs."""
 
+    _lc_set_cache: dict[bytes, tuple[bitarray, bitarray, dict[int, Any]]]
+
     def _encode_lc_set(self, dst_lc: bytes) -> tuple[bitarray, bitarray, dict[int, Any]]:
         """Header LC, terminator LC and embedded LC for one destination LC.
 
@@ -72,11 +74,7 @@ class LcTaMixin:
         leg: every leg opened for the same TG and radio gets the same encoding.
         The results are shared between legs, and every reader only slices them.
         """
-        cache: dict[bytes, tuple[bitarray, bitarray, dict[int, Any]]] | None = getattr(
-            self, "_lc_set_cache", None
-        )
-        if cache is None:
-            cache = self._lc_set_cache = {}
+        cache = self._lc_set_cache
         if not isinstance(dst_lc, bytes):  # unhashable (bytearray): encode, don't keep
             return (
                 bptc.encode_header_lc(dst_lc),
@@ -91,7 +89,7 @@ class LcTaMixin:
                 self._encode_emblc(dst_lc),
             )
             if len(cache) >= _LC_SET_CACHE_MAX:
-                cache.clear()
+                del cache[next(iter(cache))]  # oldest first, not the whole cache
             cache[dst_lc] = codes
         return codes
 

src/adn_server/application/routing/voice_subscription.py

@@ -46,7 +46,7 @@ from __future__ import annotations
 import logging
 from typing import Any, NamedTuple
 
-from ...domain.value_objects import bytes_3
+from ...domain.value_objects import TgId, bytes_3
 from ...domain.voice_routing import ForwardLeg, VoiceIngress
 from ..ports import SubscriptionStore
 from ..subscription.ingress import build_voice_ingress
@@ -71,16 +71,36 @@ class ForwardPlan(NamedTuple):
 class _CachedPlan(NamedTuple):
     revision: int
     # Every SYSTEMS block the plan read, as the object it read: a reload
-    # replaces the block, which is what makes the plan stale.
+    # replaces the block, which is what makes the plan stale. Identity only
+    # catches reloads, though: blocks are also changed in place at runtime (the
+    # static TGs, for one). That is safe because the plan reads nothing from a
+    # block but MODE, which only a reload changes. A plan that reads any other
+    # field must also be keyed on it, or it will be silently stale.
     blocks: tuple[tuple[str, Any], ...]
     plan: ForwardPlan
 
 
+def _route_ingress(system_name: str, slot: int, dst_int: int, call_type: str, mode: str) -> VoiceIngress:
+    """An ingress carrying only what ``resolve()`` routes on: no peer, radio or stream.
+
+    Leaving those out is what lets one plan serve every frame of a call; a router
+    that starts reading them has to be added to the plan cache key first.
+    """
+    return VoiceIngress(
+        source_system=system_name,
+        slot=1 if int(slot) == 1 else 2,  # type: ignore[arg-type]
+        dst_tgid=TgId(dst_int),
+        source_is_obp=mode == "OPENBRIDGE",
+        call_type=call_type,
+    )
+
+
 class VoiceSubscriptionMixin:
     """Wire ``SubscriptionRouter`` into ``dmrd_received``."""
 
     _subscription_store: SubscriptionStore
     _subscription_router: SubscriptionRouter | None
+    _forward_plan_cache: dict[tuple, _CachedPlan]
 
     def _subscription_router_instance(self) -> SubscriptionRouter:
         router = getattr(self, "_subscription_router", None)
@@ -129,6 +149,8 @@ class VoiceSubscriptionMixin:
         and on nothing in the frame beyond the key, so each frame of a call reuses
         the plan of the first until the store is written or a block is reloaded.
         Per-frame checks (ENABLED, quench, keepalive, contention) stay in the loop.
+
+        Invariant: from the SYSTEMS blocks it reads only MODE (see ``_CachedPlan``).
         """
         systems = self._config.get("SYSTEMS", {})
         revision = getattr(self._subscription_store, "revision", None)
@@ -138,9 +160,7 @@ class VoiceSubscriptionMixin:
         # VoiceIngress.bridge_match_slot, which is what resolve() matches on.
         match_slot = 1 if mode == "OPENBRIDGE" else (1 if int(slot) == 1 else 2)
         key = (system_name, bridge_match_slot, match_slot, dst_int, mode == "OPENBRIDGE", routable)
-        cache: dict[tuple, _CachedPlan] | None = getattr(self, "_forward_plan_cache", None)
-        if cache is None:
-            cache = self._forward_plan_cache = {}
+        cache = self._forward_plan_cache
         hit = cache.get(key) if revision is not None else None
         if (
             hit is not None
@@ -149,18 +169,9 @@ class VoiceSubscriptionMixin:
         ):
             return hit.plan
 
-        tables, legs = self._voice_forward_plan(
-            system_name=system_name,
-            peer_id=b"",
-            rf_src=b"",
-            dst_id=dst_int.to_bytes(3, "big"),
-            slot=slot,
-            call_type=call_type,
-            stream_id=b"",
-            source_is_obp=source_is_obp,
-            bridge_match_slot=bridge_match_slot,
-            dst_int=dst_int,
-        )
+        router = self._subscription_router_instance()
+        tables = router.relay_tables_with_active_source(system_name, bridge_match_slot, dst_int)
+        legs = router.resolve(_route_ingress(system_name, slot, dst_int, call_type, mode)) if routable else ()
         # One leg per (target, translated TGID) on MASTER/PEER targets: their
         # send_peers() picks each peer's slot itself, so a second leg is the same
         # audio twice. OpenBridge targets keep per-slot legs (separate links).
@@ -190,7 +201,7 @@ class VoiceSubscriptionMixin:
         if revision is not None:
             names = {system_name, *(leg.target_system for leg in legs)}
             if len(cache) >= _FORWARD_PLAN_CACHE_MAX:
-                cache.clear()
+                del cache[next(iter(cache))]  # oldest first, not the whole cache
             cache[key] = _CachedPlan(revision, tuple((n, systems.get(n)) for n in names), plan)
         return plan
 

src/adn_server/application/routing_use_cases.py

@@ -124,6 +124,8 @@ class RoutingUseCases(
         self._config = config
         self._subscription_store = subscription_store
         self._subscription_router = None
+        self._forward_plan_cache = {}  # see VoiceSubscriptionMixin._group_voice_forward_plan
+        self._lc_set_cache = {}  # see LcTaMixin._encode_lc_set
         self._routing_table_legacy_view = None
         self._send_to_system = send_to_system  # (system_name, packet, **kwargs) -> None
         self._get_protocols = get_protocols  # () -> dict[str, protocol]

tests/routing/test_forward_plan_cache.py

@@ -207,4 +207,15 @@ def test_lc_set_of_a_bytearray_is_encoded_without_caching() -> None:
     assert header == bptc.encode_header_lc(bytes(lc))
     assert terminator == bptc.encode_terminator_lc(bytes(lc))
     assert emb == bptc.encode_emblc(bytes(lc))
-    assert not getattr(sc.routing, "_lc_set_cache", {})
+    assert not sc.routing._lc_set_cache
+
+
+def test_full_lc_set_cache_drops_only_its_oldest_entry(monkeypatch) -> None:
+    from adn_server.application.routing import lc_ta
+
+    monkeypatch.setattr(lc_ta, "_LC_SET_CACHE_MAX", 3)
+    sc = _scenario()
+    lcs = [b"\x00\x00\x20" + TG.to_bytes(3, "big") + (2130000 + i).to_bytes(3, "big") for i in range(4)]
+    for lc in lcs:
+        sc.routing._encode_lc_set(lc)
+    assert list(sc.routing._lc_set_cache) == lcs[1:]