Populate SubscriptionStore from BRIDGES on OPTIONS/startup paths (P2-008). BRIDGES remains voice routing authority; no dmrd_received changes. Phases 0-4 complete. Next: P2-009 subscription runtime cutover.pull/1/head
Rodrigo Pérez
4 months ago
parent
cd836d1f5a
commit
67710558e8
@ -0,0 +1,75 @@
|
||||
"""Import legacy ``BRIDGES`` rows into domain subscriptions (mirror of ``bridges_export``)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from adn_server.domain import int_id
|
||||
from adn_server.domain.subscription import (
|
||||
ActivationPolicy,
|
||||
AudioChannel,
|
||||
Subscription,
|
||||
SubscriptionPhase,
|
||||
SubscriptionRole,
|
||||
SubscriptionState,
|
||||
SystemId,
|
||||
TgId,
|
||||
)
|
||||
|
||||
|
||||
def subscriptions_from_bridges(bridges: dict[str, list[dict[str, Any]]]) -> list[Subscription]:
|
||||
"""Build subscriptions from a legacy ``BRIDGES`` snapshot (OPTIONS/static TG / ECHO)."""
|
||||
subs: list[Subscription] = []
|
||||
for table_key, rows in bridges.items():
|
||||
for row in rows:
|
||||
if not isinstance(row, dict):
|
||||
continue
|
||||
subs.append(_subscription_from_row(table_key, row))
|
||||
return subs
|
||||
|
||||
|
||||
def _subscription_from_row(table_key: str, row: dict[str, Any]) -> Subscription:
|
||||
ts = int(row.get("TS") or 1)
|
||||
if table_key.startswith("#"):
|
||||
channel_tgid = int_id(row.get("TGID") or b"\x00\x00\x00")
|
||||
bridge_key = table_key
|
||||
else:
|
||||
try:
|
||||
channel_tgid = int(table_key)
|
||||
except ValueError:
|
||||
channel_tgid = int_id(row.get("TGID") or b"\x00\x00\x00")
|
||||
bridge_key = None
|
||||
to_type = str(row.get("TO_TYPE", "ON"))
|
||||
timer = row.get("TIMER")
|
||||
timer_at = float(timer) if isinstance(timer, (int, float)) else None
|
||||
timeout = row.get("TIMEOUT")
|
||||
timeout_sec = float(timeout) if isinstance(timeout, (int, float)) else None
|
||||
return Subscription(
|
||||
channel=AudioChannel(tgid=TgId(channel_tgid), slot=ts), # type: ignore[arg-type]
|
||||
system=SystemId(str(row.get("SYSTEM", ""))),
|
||||
target_tgid=TgId(int_id(row.get("TGID") or b"\x00\x00\x00")),
|
||||
role=_role_from_to_type(to_type),
|
||||
policy=_policy_from_to_type(to_type),
|
||||
state=SubscriptionState(
|
||||
phase=SubscriptionPhase.ACTIVE if row.get("ACTIVE") else SubscriptionPhase.IDLE,
|
||||
timer_expires_at=timer_at,
|
||||
),
|
||||
bridge_key=bridge_key,
|
||||
timeout_seconds=timeout_sec,
|
||||
)
|
||||
|
||||
|
||||
def _role_from_to_type(to_type: str) -> SubscriptionRole:
|
||||
if to_type == "NONE":
|
||||
return SubscriptionRole.ECHO
|
||||
if to_type == "STAT":
|
||||
return SubscriptionRole.PASSIVE_STAT
|
||||
return SubscriptionRole.SINK
|
||||
|
||||
|
||||
def _policy_from_to_type(to_type: str) -> ActivationPolicy:
|
||||
if to_type == "STAT":
|
||||
return ActivationPolicy.OPENBRIDGE_STAT
|
||||
if to_type == "OFF":
|
||||
return ActivationPolicy.STATIC
|
||||
return ActivationPolicy.INBAND
|
||||
@ -0,0 +1,16 @@
|
||||
"""Keep ``SubscriptionStore`` aligned with legacy ``BRIDGES`` (read-only mirror; not routing authority yet)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from adn_server.application.ports import SubscriptionStore
|
||||
from adn_server.application.subscription.bridges_import import subscriptions_from_bridges
|
||||
|
||||
|
||||
def replace_store_from_bridges(
|
||||
store: SubscriptionStore,
|
||||
bridges: dict[str, list[dict[str, Any]]],
|
||||
) -> None:
|
||||
"""Replace store contents from a full ``BRIDGES`` snapshot."""
|
||||
store.replace_all(subscriptions_from_bridges(bridges))
|
||||
@ -0,0 +1,90 @@
|
||||
"""BRIDGES → Subscription import (policy / role parity)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from adn_server.application.subscription.bridges_import import subscriptions_from_bridges
|
||||
from adn_server.domain import bytes_3
|
||||
from adn_server.domain.subscription import (
|
||||
ActivationPolicy,
|
||||
SubscriptionPhase,
|
||||
SubscriptionRole,
|
||||
)
|
||||
|
||||
|
||||
def test_import_echo_and_inband_rows():
|
||||
tgid_b = bytes_3(9990)
|
||||
bridges = {
|
||||
"9990": [
|
||||
{
|
||||
"SYSTEM": "ECHO",
|
||||
"TS": 2,
|
||||
"TGID": tgid_b,
|
||||
"ACTIVE": True,
|
||||
"TIMEOUT": 120.0,
|
||||
"TO_TYPE": "NONE",
|
||||
},
|
||||
{
|
||||
"SYSTEM": "MASTER-A",
|
||||
"TS": 1,
|
||||
"TGID": tgid_b,
|
||||
"ACTIVE": False,
|
||||
"TIMEOUT": 600.0,
|
||||
"TO_TYPE": "ON",
|
||||
},
|
||||
]
|
||||
}
|
||||
subs = subscriptions_from_bridges(bridges)
|
||||
assert len(subs) == 2
|
||||
echo, master = subs
|
||||
assert echo.role == SubscriptionRole.ECHO
|
||||
assert echo.policy == ActivationPolicy.INBAND
|
||||
assert echo.state.phase == SubscriptionPhase.ACTIVE
|
||||
assert master.role == SubscriptionRole.SINK
|
||||
assert master.policy == ActivationPolicy.INBAND
|
||||
assert master.state.phase == SubscriptionPhase.IDLE
|
||||
|
||||
|
||||
def test_import_static_and_stat_rows():
|
||||
bridges = {
|
||||
"12345": [
|
||||
{
|
||||
"SYSTEM": "MASTER-A",
|
||||
"TS": 1,
|
||||
"TGID": bytes_3(12345),
|
||||
"ACTIVE": True,
|
||||
"TIMEOUT": 600.0,
|
||||
"TO_TYPE": "OFF",
|
||||
},
|
||||
{
|
||||
"SYSTEM": "OBP-CL",
|
||||
"TS": 1,
|
||||
"TGID": bytes_3(12345),
|
||||
"ACTIVE": True,
|
||||
"TIMEOUT": "",
|
||||
"TO_TYPE": "STAT",
|
||||
},
|
||||
]
|
||||
}
|
||||
subs = subscriptions_from_bridges(bridges)
|
||||
static, stat = subs
|
||||
assert static.policy == ActivationPolicy.STATIC
|
||||
assert static.role == SubscriptionRole.SINK
|
||||
assert stat.policy == ActivationPolicy.OPENBRIDGE_STAT
|
||||
assert stat.role == SubscriptionRole.PASSIVE_STAT
|
||||
|
||||
|
||||
def test_import_hash_table_key_sets_bridge_key():
|
||||
bridges = {
|
||||
"#730444": [
|
||||
{
|
||||
"SYSTEM": "OBP-CL",
|
||||
"TS": 1,
|
||||
"TGID": bytes_3(730444),
|
||||
"ACTIVE": True,
|
||||
"TIMEOUT": "",
|
||||
"TO_TYPE": "STAT",
|
||||
}
|
||||
]
|
||||
}
|
||||
(sub,) = subscriptions_from_bridges(bridges)
|
||||
assert sub.bridge_key == "#730444"
|
||||
@ -0,0 +1,84 @@
|
||||
"""SubscriptionStore sync from BRIDGES (V2-P2-008)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from adn_server.application.subscription.bridges_export import export_bridges
|
||||
from adn_server.application.subscription.store_sync import replace_store_from_bridges
|
||||
from adn_server.domain import bytes_3, int_id
|
||||
from adn_server.infrastructure.bootstrap.peer_server import _make_echo_bridges
|
||||
from adn_server.infrastructure.subscription_store import InMemorySubscriptionStore
|
||||
|
||||
|
||||
def _row_fingerprint(row: dict[str, Any]) -> tuple[Any, ...]:
|
||||
return (
|
||||
row.get("SYSTEM"),
|
||||
int(row.get("TS") or 1),
|
||||
int_id(row.get("TGID") or b"\x00\x00\x00"),
|
||||
bool(row.get("ACTIVE")),
|
||||
str(row.get("TO_TYPE", "ON")),
|
||||
row.get("TIMEOUT"),
|
||||
)
|
||||
|
||||
|
||||
def _bridges_fingerprints(bridges: dict[str, list[dict[str, Any]]]) -> set[tuple[Any, ...]]:
|
||||
fps: set[tuple[Any, ...]] = set()
|
||||
for rows in bridges.values():
|
||||
for row in rows:
|
||||
if isinstance(row, dict):
|
||||
fps.add(_row_fingerprint(row))
|
||||
return fps
|
||||
|
||||
|
||||
def test_replace_store_from_echo_bridges_round_trip():
|
||||
config = {
|
||||
"SYSTEMS": {
|
||||
"ECHO": {"MODE": "PEER"},
|
||||
"MASTER-A": {"MODE": "MASTER", "DEFAULT_UA_TIMER": 10},
|
||||
"MASTER-B": {"MODE": "MASTER", "DEFAULT_UA_TIMER": 15},
|
||||
}
|
||||
}
|
||||
bridges = _make_echo_bridges(config)
|
||||
store = InMemorySubscriptionStore()
|
||||
replace_store_from_bridges(store, bridges)
|
||||
|
||||
assert len(store.snapshot()) == len(_bridges_fingerprints(bridges))
|
||||
|
||||
now = 1_700_000_000.0
|
||||
exported = export_bridges(store, now=now)
|
||||
assert _bridges_fingerprints(bridges) == _bridges_fingerprints(exported)
|
||||
|
||||
|
||||
def test_replace_store_clears_stale_entries():
|
||||
store = InMemorySubscriptionStore()
|
||||
bridges_a = {
|
||||
"100": [
|
||||
{
|
||||
"SYSTEM": "SYS-A",
|
||||
"TS": 1,
|
||||
"TGID": bytes_3(100),
|
||||
"ACTIVE": False,
|
||||
"TIMEOUT": 600.0,
|
||||
"TO_TYPE": "ON",
|
||||
}
|
||||
]
|
||||
}
|
||||
bridges_b = {
|
||||
"200": [
|
||||
{
|
||||
"SYSTEM": "SYS-B",
|
||||
"TS": 2,
|
||||
"TGID": bytes_3(200),
|
||||
"ACTIVE": True,
|
||||
"TIMEOUT": 600.0,
|
||||
"TO_TYPE": "ON",
|
||||
}
|
||||
]
|
||||
}
|
||||
replace_store_from_bridges(store, bridges_a)
|
||||
assert len(store.snapshot()) == 1
|
||||
replace_store_from_bridges(store, bridges_b)
|
||||
assert len(store.snapshot()) == 1
|
||||
(sub,) = store.snapshot()
|
||||
assert int(sub.channel.tgid) == 200
|
||||
Loading…
Reference in new issue