diff --git a/src/adn_server/application/bridge_use_cases.py b/src/adn_server/application/bridge_use_cases.py index a963bfb..4dbbe48 100644 --- a/src/adn_server/application/bridge_use_cases.py +++ b/src/adn_server/application/bridge_use_cases.py @@ -2441,17 +2441,6 @@ class BridgeUseCases: # a table that contains a matching ACTIVE source row (not a flat merge of TG + #TG only). dst_id_b = dst_id if isinstance(dst_id, bytes) and len(dst_id) >= 3 else bytes_3(dst_int) - def _row_is_active_source(row: dict[str, Any]) -> bool: - return bool( - row.get("SYSTEM") == system_name - and row.get("TS") == bridge_match_slot - and row.get("ACTIVE") - and ( - row.get("TGID") == dst_id_b - or int_id(row.get("TGID") or b"\x00\x00\x00") == dst_int - ) - ) - if source_is_obp: self._ensure_obp_source_for_tg(system_name, bridge_key, dst_id_b, dst_int) bridges = self._router.get_bridges() @@ -2471,11 +2460,15 @@ class BridgeUseCases: obp_hops if obp_use_parsed else b"", ): return - has_source = any(_row_is_active_source(e) for elist in bridges.values() for e in elist) + has_source = bool( + self._router.bridge_tables_with_active_source(system_name, bridge_match_slot, dst_int) + ) if not has_source and systems_cfg.get(system_name, {}).get("MODE") == "MASTER": self.options_config_for_system(system_name) bridges = self._router.get_bridges() - has_source = any(_row_is_active_source(e) for elist in bridges.values() for e in elist) + has_source = bool( + self._router.bridge_tables_with_active_source(system_name, bridge_match_slot, dst_int) + ) # Do not call make_single_bridge here for 9990–9999 when BRIDGES["9990"] already exists: # make_single_bridge replaces the whole table and only sets the source MASTER ACTIVE; every # other system (including ECHO with TO_TYPE NONE) becomes ACTIVE False — to_target then has @@ -2579,12 +2572,13 @@ class BridgeUseCases: # pass; dedupe (SYSTEM, TS) for OpenBridge targets so the same leg is not sent twice per packet. sys_ignore_obp: set[tuple[str, int]] = set() forwarded = [] - for _bridge_table_name, _bridge_rows in list(bridges.items()): + for _bridge_table_name in self._router.bridge_tables_with_active_source( + system_name, bridge_match_slot, dst_int + ): # Legacy: routerOBP/routerHBP to_target — if BRIDGES[_bridge] was removed mid-routing, skip if _bridge_table_name not in bridges: continue - if not any(_row_is_active_source(r) for r in _bridge_rows): - continue + _bridge_rows = bridges[_bridge_table_name] for entry in _bridge_rows: if entry.get("SYSTEM") == system_name: continue diff --git a/src/adn_server/application/ports.py b/src/adn_server/application/ports.py index 743536b..946d731 100644 --- a/src/adn_server/application/ports.py +++ b/src/adn_server/application/ports.py @@ -120,6 +120,16 @@ class BridgeRouter(ABC): """Replace BRIDGES (e.g. after rule_timer or make_single_bridge).""" ... + @abstractmethod + def rebuild_source_index(self) -> None: + """Rebuild index of ACTIVE source rows by (system, TS, dst_tgid).""" + ... + + @abstractmethod + def bridge_tables_with_active_source(self, system: str, ts: int, dst_tgid: int) -> list[str]: + """Return bridge table names with matching ACTIVE source (legacy full-scan parity).""" + ... + @abstractmethod def acl_check(self, id_bytes_or_int: bytes | int, acl: tuple[bool, list[tuple[int, int]]]) -> bool: """Check ID against ACL; return True if permitted.""" diff --git a/src/adn_server/infrastructure/bridge_router_impl.py b/src/adn_server/infrastructure/bridge_router_impl.py index 98e7d0c..a4b3b51 100644 --- a/src/adn_server/infrastructure/bridge_router_impl.py +++ b/src/adn_server/infrastructure/bridge_router_impl.py @@ -28,6 +28,7 @@ from __future__ import annotations from typing import Any from ..application.ports import BridgeRouter +from ..domain import int_id def _int_id(val: bytes | int) -> int: @@ -45,12 +46,39 @@ class InMemoryBridgeRouter(BridgeRouter): def __init__(self) -> None: self._bridges: dict[str, list[dict[str, Any]]] = {} + self._source_index: dict[tuple[str, int, int], list[str]] = {} def get_bridges(self) -> dict[str, list[dict[str, Any]]]: return self._bridges def set_bridges(self, bridges: dict[str, list[dict[str, Any]]]) -> None: self._bridges = bridges + self._source_index = {} + + def rebuild_source_index(self) -> None: + """Rebuild O(1) lookup: (system, TS, dst_tgid) -> bridge table names with ACTIVE source row.""" + index: dict[tuple[str, int, int], list[str]] = {} + for bridge_name, rows in self._bridges.items(): + for row in rows: + if not row.get("ACTIVE"): + continue + system = row.get("SYSTEM") + if not system: + continue + ts = row.get("TS", 1) + if ts is None: + ts = 1 + tgid = int_id(row.get("TGID") or b"\x00\x00\x00") + key = (system, int(ts), tgid) + names = index.setdefault(key, []) + if bridge_name not in names: + names.append(bridge_name) + self._source_index = index + + def bridge_tables_with_active_source(self, system: str, ts: int, dst_tgid: int) -> list[str]: + """Bridge tables containing an ACTIVE source on (system, ts) for dst_tgid (legacy scan parity).""" + self.rebuild_source_index() + return list(self._source_index.get((system, int(ts), int(dst_tgid)), [])) def acl_check(self, id_bytes_or_int: bytes | int, acl: tuple[bool, list[tuple[int, int]]]) -> bool: """Legacy acl_check: (action, ranges). If id in any range return action else not action.""" diff --git a/tests/README.md b/tests/README.md index df9d51f..aa54bcf 100644 --- a/tests/README.md +++ b/tests/README.md @@ -7,7 +7,7 @@ python3 -m pip install -e ".[dev]" python3 -m pytest tests//test_.py -q # single file python3 -m pytest tests//test_.py::test_foo -q # single test python3 -m pytest tests/bridge/ -q # whole domain -python3 -m pytest tests/ -q # full suite (142) +python3 -m pytest tests/ -q # full suite (147) ``` Use the project interpreter, e.g. `/opt/.pyenv/versions/3.11.8/bin/python3`. @@ -103,6 +103,7 @@ Use the project interpreter, e.g. `/opt/.pyenv/versions/3.11.8/bin/python3`. | `smoke/test_bridge_routing.py` | 1 | Static TG forward smoke | | `smoke/test_packet_builder.py` | 1 | PacketSpec builder | | `infrastructure/test_logging_reload.py` | 2 | Log level reload | +| `infrastructure/test_bridge_router_index.py` | 5 | BRIDGES O(1) index vs legacy scan | | `scripts/test_freedmr_cfg_to_yaml.py` | 5 | Legacy cfg → YAML | ## Examples (copy-paste) diff --git a/tests/infrastructure/test_bridge_router_index.py b/tests/infrastructure/test_bridge_router_index.py new file mode 100644 index 0000000..b5bf6b3 --- /dev/null +++ b/tests/infrastructure/test_bridge_router_index.py @@ -0,0 +1,115 @@ +"""BRIDGES source index parity with legacy full scan.""" + +from __future__ import annotations + +from adn_server.domain import bytes_3, int_id +from adn_server.infrastructure.bridge_router_impl import InMemoryBridgeRouter + + +def _legacy_scan_tables( + bridges: dict, + system_name: str, + bridge_match_slot: int, + dst_id_b: bytes, + dst_int: int, +) -> list[str]: + result: list[str] = [] + + def row_is_active_source(row: dict) -> bool: + return bool( + row.get("SYSTEM") == system_name + and row.get("TS") == bridge_match_slot + and row.get("ACTIVE") + and ( + row.get("TGID") == dst_id_b + or int_id(row.get("TGID") or b"\x00\x00\x00") == dst_int + ) + ) + + for bridge_name, rows in bridges.items(): + if any(row_is_active_source(r) for r in rows): + result.append(bridge_name) + return result + + +def test_index_matches_legacy_scan_static_tg() -> None: + router = InMemoryBridgeRouter() + tgid = bytes_3(91) + router.set_bridges( + { + "91": [ + {"SYSTEM": "MASTER-A", "TS": 2, "TGID": tgid, "ACTIVE": True}, + {"SYSTEM": "MASTER-B", "TS": 2, "TGID": tgid, "ACTIVE": True}, + ] + } + ) + dst_b = tgid + dst_int = 91 + legacy = _legacy_scan_tables(router.get_bridges(), "MASTER-A", 2, dst_b, dst_int) + indexed = router.bridge_tables_with_active_source("MASTER-A", 2, dst_int) + assert indexed == legacy == ["91"] + + +def test_index_matches_legacy_scan_multiple_tables() -> None: + router = InMemoryBridgeRouter() + tgid = bytes_3(310) + router.set_bridges( + { + "91": [ + {"SYSTEM": "MASTER-A", "TS": 2, "TGID": bytes_3(91), "ACTIVE": True}, + {"SYSTEM": "MASTER-B", "TS": 2, "TGID": bytes_3(91), "ACTIVE": True}, + ], + "#310": [ + {"SYSTEM": "MASTER-A", "TS": 2, "TGID": tgid, "ACTIVE": True}, + {"SYSTEM": "OBP-1", "TS": 1, "TGID": tgid, "ACTIVE": True}, + ], + } + ) + legacy = _legacy_scan_tables(router.get_bridges(), "MASTER-A", 2, tgid, 310) + indexed = router.bridge_tables_with_active_source("MASTER-A", 2, 310) + assert indexed == legacy == ["#310"] + + +def test_index_empty_when_no_active_source() -> None: + router = InMemoryBridgeRouter() + router.set_bridges( + { + "91": [ + {"SYSTEM": "MASTER-A", "TS": 2, "TGID": bytes_3(91), "ACTIVE": False}, + {"SYSTEM": "MASTER-B", "TS": 2, "TGID": bytes_3(91), "ACTIVE": True}, + ] + } + ) + assert router.bridge_tables_with_active_source("MASTER-A", 2, 91) == [] + + +def test_index_updates_after_in_place_mutation() -> None: + router = InMemoryBridgeRouter() + bridges = router.get_bridges() + bridges["91"] = [ + {"SYSTEM": "MASTER-A", "TS": 2, "TGID": bytes_3(91), "ACTIVE": False}, + {"SYSTEM": "MASTER-B", "TS": 2, "TGID": bytes_3(91), "ACTIVE": True}, + ] + assert router.bridge_tables_with_active_source("MASTER-A", 2, 91) == [] + bridges["91"][0]["ACTIVE"] = True + legacy = _legacy_scan_tables(bridges, "MASTER-A", 2, bytes_3(91), 91) + assert router.bridge_tables_with_active_source("MASTER-A", 2, 91) == legacy == ["91"] + + +def test_index_many_bridge_tables() -> None: + """Regression: index order and membership match legacy scan with many tables.""" + router = InMemoryBridgeRouter() + tgid = bytes_3(500) + bridges: dict[str, list] = {} + for i in range(120): + bridges[str(1000 + i)] = [ + {"SYSTEM": "MASTER-X", "TS": 1, "TGID": bytes_3(1000 + i), "ACTIVE": True}, + ] + bridges["500"] = [ + {"SYSTEM": "MASTER-A", "TS": 2, "TGID": tgid, "ACTIVE": True}, + {"SYSTEM": "MASTER-B", "TS": 2, "TGID": tgid, "ACTIVE": True}, + ] + router.set_bridges(bridges) + legacy = _legacy_scan_tables(router.get_bridges(), "MASTER-A", 2, tgid, 500) + indexed = router.bridge_tables_with_active_source("MASTER-A", 2, 500) + assert indexed == legacy == ["500"]