From 7c8db554a3e21393c8e5cfb84f1c2e7e7d483d68 Mon Sep 17 00:00:00 2001 From: yo Date: Thu, 24 Sep 2026 22:02:01 +0200 Subject: [PATCH] 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 --- src/adn_server/application/ports.py | 5 +- src/adn_server/application/routing/lc_ta.py | 10 ++-- .../application/routing/voice_subscription.py | 47 ++++++++++++------- .../application/routing_use_cases.py | 2 + tests/routing/test_forward_plan_cache.py | 13 ++++- 5 files changed, 50 insertions(+), 27 deletions(-) diff --git a/src/adn_server/application/ports.py b/src/adn_server/application/ports.py index 9a97b58..8c55dca 100644 --- a/src/adn_server/application/ports.py +++ b/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: diff --git a/src/adn_server/application/routing/lc_ta.py b/src/adn_server/application/routing/lc_ta.py index 3ad23d7..6c3ca27 100644 --- a/src/adn_server/application/routing/lc_ta.py +++ b/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 diff --git a/src/adn_server/application/routing/voice_subscription.py b/src/adn_server/application/routing/voice_subscription.py index 7505b98..3778b69 100644 --- a/src/adn_server/application/routing/voice_subscription.py +++ b/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 diff --git a/src/adn_server/application/routing_use_cases.py b/src/adn_server/application/routing_use_cases.py index c9c2265..34935e8 100644 --- a/src/adn_server/application/routing_use_cases.py +++ b/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] diff --git a/tests/routing/test_forward_plan_cache.py b/tests/routing/test_forward_plan_cache.py index 2724284..cd40ddc 100644 --- a/tests/routing/test_forward_plan_cache.py +++ b/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:]