KB8PMY
/
ADN-DMR-Peer-Server
mirror of https://github.com/Amateur-Digital-Network/ADN-DMR-Peer-Server.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: per-stream OBP bridge TX legs for concurrent MASTER downlink

Browse Source 
When two OBP voice streams share the same MASTER timeslot, stop
flip-flopping the flat TX row so per-peer downlink gates stay stable.
pull/36/head
Rodrigo Pérez 3 months ago
parent d1d7703d05
commit 2f31217b09
4 changed files with 975 additions and 53 deletions
Show Stats Download Patch File Download Diff File

106
src/adn_server/application/routing/helpers.py
Unescape View File

@ -461,6 +461,112 @@ def inject_only_defer_obp_hbp_slot_contention(
return source_is_hbp and connected_count > 1 return source_is_hbp and connected_count > 1
_OBP_FLAT_TX_KEYS = (
"TX_START",
"TX_TGID",
"TX_STREAM_ID",
"TX_RFS",
"TX_PEER",
"TX_H_LC",
"TX_T_LC",
"TX_EMB_LC",
"TX_TIME",
"TX_TYPE",
)
def obp_deferred_bridge_tx_leg(slot_st: dict[str, Any], stream_id: bytes) -> dict[str, Any]:
"""Per-stream bridge TX stamp for inject-only OBP→MASTER legs on a shared slot row."""
legs = slot_st.setdefault("TX_STREAMS", {})
if not isinstance(legs, dict):
legs = {}
slot_st["TX_STREAMS"] = legs
return legs.setdefault(stream_id, {})
def obp_flat_bridge_tx_idle(slot_st: dict[str, Any], pkt_time: float) -> bool:
"""True when flat ``TX_*`` on the slot is not carrying an active bridge leg."""
tx_type = slot_st.get("TX_TYPE")
if tx_type is None or tx_type == HBPF_SLT_VTERM:
return True
tx_time = float(slot_st.get("TX_TIME", 0) or 0)
return pkt_time >= tx_time + STREAM_TO
def obp_bridge_tx_leg_active(leg: dict[str, Any], pkt_time: float) -> bool:
tx_type = leg.get("TX_TYPE")
if tx_type is None or tx_type == HBPF_SLT_VTERM:
return False
tx_time = float(leg.get("TX_TIME", 0) or 0)
return pkt_time < tx_time + STREAM_TO
def obp_publish_flat_bridge_tx(slot_st: dict[str, Any], leg: dict[str, Any]) -> None:
"""Mirror one per-stream bridge leg onto the legacy flat ``TX_*`` slot row."""
for key in _OBP_FLAT_TX_KEYS:
if key in leg:
slot_st[key] = leg[key]
def obp_sync_flat_bridge_tx_times(
slot_st: dict[str, Any],
leg: dict[str, Any],
stream_id: bytes,
pkt_time: float,
dtype_vseq: int,
) -> None:
"""Refresh per-stream leg activity; mirror times to flat row only for the owner stream."""
leg["TX_TIME"] = pkt_time
leg["TX_TYPE"] = dtype_vseq
if slot_st.get("TX_STREAM_ID") == stream_id:
slot_st["TX_TIME"] = pkt_time
slot_st["TX_TYPE"] = dtype_vseq
def obp_pick_active_bridge_tx_leg(
slot_st: dict[str, Any],
pkt_time: float,
) -> tuple[bytes | None, dict[str, Any] | None]:
legs = slot_st.get("TX_STREAMS")
if not isinstance(legs, dict):
return None, None
best_sid: bytes | None = None
best_leg: dict[str, Any] | None = None
best_time = -1.0
for sid, leg in legs.items():
if not isinstance(leg, dict) or not obp_bridge_tx_leg_active(leg, pkt_time):
continue
t = float(leg.get("TX_TIME", 0) or 0)
if t > best_time:
best_time = t
best_sid = sid
best_leg = leg
return best_sid, best_leg
def obp_clear_flat_bridge_tx(slot_st: dict[str, Any]) -> None:
slot_st["TX_TYPE"] = HBPF_SLT_VTERM
slot_st["TX_STREAM_ID"] = b"\x00"
slot_st["TX_TIME"] = 0.0
def obp_clear_deferred_bridge_tx_leg(
slot_st: dict[str, Any],
stream_id: bytes,
pkt_time: float,
) -> None:
"""End one per-stream OBP bridge leg; republish flat ``TX_*`` for another active leg if any."""
legs = slot_st.get("TX_STREAMS")
if isinstance(legs, dict):
legs.pop(stream_id, None)
if slot_st.get("TX_STREAM_ID") == stream_id:
repl_sid, repl_leg = obp_pick_active_bridge_tx_leg(slot_st, pkt_time)
if repl_leg is not None and repl_sid is not None:
obp_publish_flat_bridge_tx(slot_st, repl_leg)
else:
obp_clear_flat_bridge_tx(slot_st)
def hbp_slot_blocks_group_voice_for_peer( def hbp_slot_blocks_group_voice_for_peer(
slot_st: dict[str, Any], slot_st: dict[str, Any],
peer_id: bytes, peer_id: bytes,

21
src/adn_server/application/routing/timers.py
Unescape View File

@ -48,6 +48,7 @@ import time
from typing import Any from typing import Any
from ...domain import HBPF_SLT_VTERM, int_id from ...domain import HBPF_SLT_VTERM, int_id
from .helpers import obp_clear_deferred_bridge_tx_leg
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -313,6 +314,26 @@ class RoutingTimerMixin:
) )
if _slot.get("RX_TIME", 0) < now - 60: if _slot.get("RX_TIME", 0) < now - 60:
_slot["RX_STREAM_ID"] = b"\x00" _slot["RX_STREAM_ID"] = b"\x00"
tx_streams = _slot.get("TX_STREAMS")
if isinstance(tx_streams, dict):
for sid, leg in list(tx_streams.items()):
if not isinstance(leg, dict):
continue
if leg.get("TX_TYPE") != HBPF_SLT_VTERM and leg.get("TX_TIME", 0) < now - 5:
logger.debug(
"(%s) *TIME OUT* TX STREAM ID: %s SUB: %s TGID %s, TS %s, Duration: %.2f",
system_name, int_id(sid), int_id(leg.get("TX_RFS", b"")),
int_id(leg.get("TX_TGID", b"")), slot,
leg.get("TX_TIME", 0) - leg.get("TX_START", 0),
)
self._send_routing_event(
"GROUP VOICE,END,TX,{},{},{},{},{},{},{:.2f}".format(
system_name, int_id(sid), int_id(leg.get("TX_PEER", b"")),
int_id(leg.get("TX_RFS", b"")), slot, int_id(leg.get("TX_TGID", b"")),
leg.get("TX_TIME", 0) - leg.get("TX_START", 0),
)
)
obp_clear_deferred_bridge_tx_leg(_slot, sid, now)
if _slot.get("TX_TYPE") != HBPF_SLT_VTERM and _slot.get("TX_TIME", 0) < now - 5: if _slot.get("TX_TYPE") != HBPF_SLT_VTERM and _slot.get("TX_TIME", 0) < now - 5:
_slot["TX_TYPE"] = HBPF_SLT_VTERM _slot["TX_TYPE"] = HBPF_SLT_VTERM
logger.debug( logger.debug(

167
src/adn_server/application/routing_use_cases.py
Unescape View File

@ -46,6 +46,11 @@ from .routing.helpers import (
inject_only_defer_obp_hbp_slot_contention, inject_only_defer_obp_hbp_slot_contention,
is_private_subscriber_dst, is_private_subscriber_dst,
is_unit_data_ingress, is_unit_data_ingress,
obp_clear_deferred_bridge_tx_leg,
obp_deferred_bridge_tx_leg,
obp_flat_bridge_tx_idle,
obp_publish_flat_bridge_tx,
obp_sync_flat_bridge_tx_times,
obp_target_bcsq_quenches_stream, obp_target_bcsq_quenches_stream,
resolve_voice_peer_id, resolve_voice_peer_id,
slot_has_active_voice, slot_has_active_voice,
@ -611,15 +616,48 @@ class RoutingUseCases(
_src_stream_st = getattr(src_proto, "STATUS", {}).get(stream_id, {}) if src_proto else {} _src_stream_st = getattr(src_proto, "STATUS", {}).get(stream_id, {}) if src_proto else {}
else: else:
_src_stream_st = getattr(src_proto, "STATUS", {}).get(slot, {}) if src_proto else {} _src_stream_st = getattr(src_proto, "STATUS", {}).get(slot, {}) if src_proto else {}
_closing_bridge_leg = ( # Slot contention: active QSO blocks any other stream; post-VTERM uses GROUP_HANGTIME.
frame_type == HBPF_DATA_SYNC _group_hangtime = float(_target_system.get("GROUP_HANGTIME", 0) or 0)
and dtype_vseq == HBPF_SLT_VTERM _tgt_peers = getattr(tgt_proto, "_peers", None) if tgt_proto else None
and _ts_st.get("TX_STREAM_ID") == stream_id if _tgt_peers is None:
and _ts_st.get("TX_TYPE") != HBPF_SLT_VTERM _tgt_peers = _target_system.get("PEERS", {})
_target_connected = (
count_connected_peers(_tgt_peers) if isinstance(_tgt_peers, dict) else 0
)
_defer_slot_contention = inject_only_defer_obp_hbp_slot_contention(
self._config,
entry["SYSTEM"],
_target_system,
source_is_obp=source_is_obp,
source_is_hbp=not source_is_obp,
connected_count=_target_connected,
) )
_obp_deferred_bridge = _defer_slot_contention and source_is_obp
_bridge_tx_leg: dict[str, Any] | None = None
if _obp_deferred_bridge:
_bridge_tx_leg = obp_deferred_bridge_tx_leg(_ts_st, stream_id)
_lc_row = _bridge_tx_leg
_closing_bridge_leg = (
frame_type == HBPF_DATA_SYNC
and dtype_vseq == HBPF_SLT_VTERM
and _bridge_tx_leg.get("TX_TYPE") is not None
and _bridge_tx_leg.get("TX_TYPE") != HBPF_SLT_VTERM
)
else:
_lc_row = _ts_st
_closing_bridge_leg = (
frame_type == HBPF_DATA_SYNC
and dtype_vseq == HBPF_SLT_VTERM
and _ts_st.get("TX_STREAM_ID") == stream_id
and _ts_st.get("TX_TYPE") != HBPF_SLT_VTERM
)
if _closing_bridge_leg: if _closing_bridge_leg:
call_duration = pkt_time - _ts_st.get("TX_START", pkt_time) if _obp_deferred_bridge and _bridge_tx_leg is not None:
_end_peer = _ts_st.get("TX_PEER", peer_id) call_duration = pkt_time - _bridge_tx_leg.get("TX_START", pkt_time)
_end_peer = _bridge_tx_leg.get("TX_PEER", peer_id)
else:
call_duration = pkt_time - _ts_st.get("TX_START", pkt_time)
_end_peer = _ts_st.get("TX_PEER", peer_id)
_end_report_peer = int_id(_end_peer) _end_report_peer = int_id(_end_peer)
if not source_is_obp: if not source_is_obp:
_end_report_peer = int_id( _end_report_peer = int_id(
@ -638,23 +676,8 @@ class RoutingUseCases(
call_duration, call_duration,
) )
) )
_ts_st["TX_TYPE"] = HBPF_SLT_VTERM if not _obp_deferred_bridge:
# Slot contention: active QSO blocks any other stream; post-VTERM uses GROUP_HANGTIME. _ts_st["TX_TYPE"] = HBPF_SLT_VTERM
_group_hangtime = float(_target_system.get("GROUP_HANGTIME", 0) or 0)
_tgt_peers = getattr(tgt_proto, "_peers", None) if tgt_proto else None
if _tgt_peers is None:
_tgt_peers = _target_system.get("PEERS", {})
_target_connected = (
count_connected_peers(_tgt_peers) if isinstance(_tgt_peers, dict) else 0
)
_defer_slot_contention = inject_only_defer_obp_hbp_slot_contention(
self._config,
entry["SYSTEM"],
_target_system,
source_is_obp=source_is_obp,
source_is_hbp=not source_is_obp,
connected_count=_target_connected,
)
if ( if (
not _defer_slot_contention not _defer_slot_contention
and hbp_slot_blocks_group_voice( and hbp_slot_blocks_group_voice(
@ -690,32 +713,56 @@ class RoutingUseCases(
int_id(_ts_st.get("RX_TGID", b"") or _ts_st.get("TX_TGID", b"")), int_id(_ts_st.get("RX_TGID", b"") or _ts_st.get("TX_TGID", b"")),
) )
continue continue
# New stream detection — legacy OBP uses _target_status[TS]['TX_STREAM_ID'], legacy HBP uses self.STATUS[_slot]['RX_STREAM_ID'] # New stream detection — legacy OBP uses flat TX_STREAM_ID; deferred OBP uses per-stream legs.
if source_is_obp: if _obp_deferred_bridge and _bridge_tx_leg is not None:
_is_new_stream = "TX_STREAM_ID" not in _bridge_tx_leg
elif source_is_obp:
_is_new_stream = (_ts_st.get("TX_STREAM_ID") != stream_id) _is_new_stream = (_ts_st.get("TX_STREAM_ID") != stream_id)
else: else:
src_status = getattr(src_proto, "STATUS", None) if src_proto else None src_status = getattr(src_proto, "STATUS", None) if src_proto else None
_is_new_stream = (src_status.get(slot, {}).get("RX_STREAM_ID") if src_status else b"") != stream_id _is_new_stream = (src_status.get(slot, {}).get("RX_STREAM_ID") if src_status else b"") != stream_id
if _is_new_stream: if _is_new_stream:
_ts_st["TX_START"] = pkt_time
_ts_st["TX_TGID"] = entry_tgid_b
_ts_st["TX_STREAM_ID"] = stream_id
_ts_st["TX_RFS"] = rf_src
_ts_st["TX_PEER"] = peer_id
dst_lc = source_lc[0:3] + entry_tgid_b + rf_src dst_lc = source_lc[0:3] + entry_tgid_b + rf_src
_ts_st["TX_H_LC"] = bptc.encode_header_lc(dst_lc) if _obp_deferred_bridge and _bridge_tx_leg is not None:
_ts_st["TX_T_LC"] = bptc.encode_terminator_lc(dst_lc) _bridge_tx_leg["TX_START"] = pkt_time
_ts_st["TX_EMB_LC"] = self._encode_emblc(dst_lc) _bridge_tx_leg["TX_TGID"] = entry_tgid_b
self._dispatch_talker_alias_on_bridge_open( _bridge_tx_leg["TX_STREAM_ID"] = stream_id
_ts_st, _bridge_tx_leg["TX_RFS"] = rf_src
system_name, _bridge_tx_leg["TX_PEER"] = peer_id
entry["SYSTEM"], _bridge_tx_leg["TX_H_LC"] = bptc.encode_header_lc(dst_lc)
rf_src, _bridge_tx_leg["TX_T_LC"] = bptc.encode_terminator_lc(dst_lc)
stream_id, _bridge_tx_leg["TX_EMB_LC"] = self._encode_emblc(dst_lc)
peer_id, if obp_flat_bridge_tx_idle(_ts_st, pkt_time) or _ts_st.get("TX_STREAM_ID") == stream_id:
int(entry_ts), obp_publish_flat_bridge_tx(_ts_st, _bridge_tx_leg)
int_id(entry_tgid_b), self._dispatch_talker_alias_on_bridge_open(
) _bridge_tx_leg,
system_name,
entry["SYSTEM"],
rf_src,
stream_id,
peer_id,
int(entry_ts),
int_id(entry_tgid_b),
)
else:
_ts_st["TX_START"] = pkt_time
_ts_st["TX_TGID"] = entry_tgid_b
_ts_st["TX_STREAM_ID"] = stream_id
_ts_st["TX_RFS"] = rf_src
_ts_st["TX_PEER"] = peer_id
_ts_st["TX_H_LC"] = bptc.encode_header_lc(dst_lc)
_ts_st["TX_T_LC"] = bptc.encode_terminator_lc(dst_lc)
_ts_st["TX_EMB_LC"] = self._encode_emblc(dst_lc)
self._dispatch_talker_alias_on_bridge_open(
_ts_st,
system_name,
entry["SYSTEM"],
rf_src,
stream_id,
peer_id,
int(entry_ts),
int_id(entry_tgid_b),
)
logger.info( logger.info(
"(%s) Conference Bridge: %s, Call Bridged to HBP System: %s TS: %s, TGID: %s", "(%s) Conference Bridge: %s, Call Bridged to HBP System: %s TS: %s, TGID: %s",
system_name, _relay_table_key, entry["SYSTEM"], entry_ts, int_id(entry_tgid_b), system_name, _relay_table_key, entry["SYSTEM"], entry_ts, int_id(entry_tgid_b),
@ -730,8 +777,13 @@ class RoutingUseCases(
int_id(entry_tgid_b), int_id(entry_tgid_b),
) )
) )
_ts_st["TX_TIME"] = pkt_time if _obp_deferred_bridge and _bridge_tx_leg is not None:
_ts_st["TX_TYPE"] = dtype_vseq obp_sync_flat_bridge_tx_times(
_ts_st, _bridge_tx_leg, stream_id, pkt_time, dtype_vseq,
)
else:
_ts_st["TX_TIME"] = pkt_time
_ts_st["TX_TYPE"] = dtype_vseq
# Slot bit rewrite (legacy bridge.py 457-460 / 770-773) # Slot bit rewrite (legacy bridge.py 457-460 / 770-773)
_src_entry_ts = slot _src_entry_ts = slot
if _src_entry_ts != entry_ts: if _src_entry_ts != entry_ts:
@ -743,12 +795,16 @@ class RoutingUseCases(
dmrbits = bitarray(endian="big") dmrbits = bitarray(endian="big")
dmrbits.frombytes(dmrpkt) dmrbits.frombytes(dmrpkt)
if frame_type == HBPF_DATA_SYNC and dtype_vseq == HBPF_SLT_VHEAD: if frame_type == HBPF_DATA_SYNC and dtype_vseq == HBPF_SLT_VHEAD:
dmrbits = _ts_st["TX_H_LC"][0:98] + dmrbits[98:166] + _ts_st["TX_H_LC"][98:197] dmrbits = _lc_row["TX_H_LC"][0:98] + dmrbits[98:166] + _lc_row["TX_H_LC"][98:197]
elif frame_type == HBPF_DATA_SYNC and dtype_vseq == HBPF_SLT_VTERM: elif frame_type == HBPF_DATA_SYNC and dtype_vseq == HBPF_SLT_VTERM:
dmrbits = _ts_st["TX_T_LC"][0:98] + dmrbits[98:166] + _ts_st["TX_T_LC"][98:197] dmrbits = _lc_row["TX_T_LC"][0:98] + dmrbits[98:166] + _lc_row["TX_T_LC"][98:197]
if not _closing_bridge_leg: if not _closing_bridge_leg:
call_duration = pkt_time - _ts_st.get("TX_START", pkt_time) if _obp_deferred_bridge and _bridge_tx_leg is not None:
_end_peer = _ts_st.get("TX_PEER", peer_id) call_duration = pkt_time - _bridge_tx_leg.get("TX_START", pkt_time)
_end_peer = _bridge_tx_leg.get("TX_PEER", peer_id)
else:
call_duration = pkt_time - _ts_st.get("TX_START", pkt_time)
_end_peer = _ts_st.get("TX_PEER", peer_id)
_end_report_peer = int_id(_end_peer) _end_report_peer = int_id(_end_peer)
if not source_is_obp: if not source_is_obp:
_end_report_peer = int_id( _end_report_peer = int_id(
@ -767,10 +823,11 @@ class RoutingUseCases(
call_duration, call_duration,
) )
) )
_ts_st["TX_TYPE"] = HBPF_SLT_VTERM if not _obp_deferred_bridge:
_ts_st["TX_TYPE"] = HBPF_SLT_VTERM
elif dtype_vseq in (1, 2, 3, 4): elif dtype_vseq in (1, 2, 3, 4):
self._rewrite_embed_lc( self._rewrite_embed_lc(
dmrbits, _ts_st, dtype_vseq, "TX_EMB_LC", dmrbits, _lc_row, dtype_vseq, "TX_EMB_LC",
) )
dmrpkt_out = dmrbits.tobytes() dmrpkt_out = dmrbits.tobytes()
# bridge_master.routerOBP.to_target HBP branch: _tmp_data + dmrpkt only (~2041-2042); # bridge_master.routerOBP.to_target HBP branch: _tmp_data + dmrpkt only (~2041-2042);
@ -789,7 +846,11 @@ class RoutingUseCases(
except Exception as e: except Exception as e:
logger.warning("(ROUTER) send_to_system %s failed: %s", entry.get("SYSTEM"), e) logger.warning("(ROUTER) send_to_system %s failed: %s", entry.get("SYSTEM"), e)
if frame_type == HBPF_DATA_SYNC and dtype_vseq == HBPF_SLT_VTERM: if frame_type == HBPF_DATA_SYNC and dtype_vseq == HBPF_SLT_VTERM:
self._clear_talker_alias_embed(_ts_st) if _obp_deferred_bridge and _bridge_tx_leg is not None:
self._clear_talker_alias_embed(_bridge_tx_leg)
obp_clear_deferred_bridge_tx_leg(_ts_st, stream_id, pkt_time)
else:
self._clear_talker_alias_embed(_ts_st)
self._talker_alias.clear_stream(entry["SYSTEM"], stream_id) self._talker_alias.clear_stream(entry["SYSTEM"], stream_id)
# Legacy bridge_master routerOBP ~2420-2434: after to_target, VTERM — CALL END log, END RX report, _fin, lastSeq # Legacy bridge_master routerOBP ~2420-2434: after to_target, VTERM — CALL END log, END RX report, _fin, lastSeq
if ( if (

734
tests/routing/test_obp_concurrent_streams_downlink.py
Unescape View File

@ -0,0 +1,734 @@
# ADN DMR Peer Server - concurrent OBP streams downlink reproduction
#
# Copyright (C) 2026 Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
"""Reproduce production bug: two concurrent OBP streams on one MASTER TS2 with a
dual-TG hotspot (SINGLE=0) intermittently drops the active listen stream.
Diagnosis-only: no production code changes. Uses DeterministicScenario for OBP
ingress + real HBPProtocol send_peer downlink gate.
"""
from __future__ import annotations
import copy
import logging
from contextlib import contextmanager
from typing import Any
import pytest
from tests.harness.deterministic import (
DeterministicScenario,
FakeClock,
PacketSpec,
add_openbridge_system,
patch_routing_wall_time,
)
from tests.support.hbp_repeat_stack import RecordingTransport
from adn_server.application.routing.downlink import (
DownlinkContext,
peer_slot_blocks_downlink,
)
from adn_server.application.routing.helpers import (
_peer_status_rx_hangtime_blocks,
_peer_transmit_hangtime_blocks,
hbp_slot_blocks_group_voice_for_peer,
peer_hotspot_voice_slot_busy,
peer_single_blocks_foreign_same_tg_downlink,
peer_single_same_tg_foreign_tx_blocks,
slot_has_active_voice,
)
from adn_server.domain import HBPF_DATA_SYNC, HBPF_SLT_VHEAD, HBPF_SLT_VTERM, bytes_3, bytes_4, int_id
from adn_server.domain.hbp_protocol import HBPF_VOICE, STREAM_TO
from adn_server.infrastructure.acl_router import InMemoryAclRouter
from adn_server.infrastructure.config_normalizer import ensure_system_runtime_config
from adn_server.infrastructure.twisted_adapters.udp_hbp import HBPProtocol
_TG_A = 3109050
_TG_B = 52090
_STREAM_A = 0xAABBCCDD
_STREAM_B = 0x11223344
_HS_PEER = 730039101
_OBP_PEER = 73010
_RF_A = 3340001
_RF_B = 3340002
_INTERVAL_S = 0.06
_VOICE_BURSTS = 8
def _bridge_row(*, system: str, ts: int, tgid: int) -> dict[str, Any]:
tg_b = bytes_3(tgid)
return {
"SYSTEM": system,
"TS": ts,
"TGID": tg_b,
"ACTIVE": True,
"TIMEOUT": 3600.0,
"TO_TYPE": "ON",
"ON": [tg_b],
"OFF": [],
"RESET": [],
"TIMER": 0.0,
}
def _dual_tg_routing_table() -> dict[str, list[dict[str, Any]]]:
return {
str(_TG_A): [
_bridge_row(system="OBP-CL", ts=1, tgid=_TG_A),
_bridge_row(system="MASTER-A", ts=2, tgid=_TG_A),
],
str(_TG_B): [
_bridge_row(system="OBP-CL", ts=1, tgid=_TG_B),
_bridge_row(system="MASTER-A", ts=2, tgid=_TG_B),
],
}
@contextmanager
def _patch_harness_time(clock: FakeClock):
"""Patch wall clock on routing ingress and downlink gate paths."""
import adn_server.application.routing.downlink as downlink_mod
import adn_server.application.routing_use_cases as routing_mod
orig_routing = routing_mod.time.time
orig_downlink = downlink_mod.time.time
routing_mod.time.time = clock.time
downlink_mod.time.time = clock.time
try:
yield
finally:
routing_mod.time.time = orig_routing
downlink_mod.time.time = orig_downlink
def _build_stack() -> tuple[DeterministicScenario, HBPProtocol, RecordingTransport]:
routing_table = _dual_tg_routing_table()
scenario = DeterministicScenario(
routing_table=routing_table,
enable_reporting=True,
)
config = scenario.config
add_openbridge_system(config, "OBP-CL")
master = config["SYSTEMS"]["MASTER-A"]
master.update(
{
"MAX_PEERS": 8,
"GROUP_HANGTIME": 5.0,
"SINGLE_MODE": False,
"TS2_STATIC": f"{_TG_A},{_TG_B}",
}
)
ensure_system_runtime_config(config)
transport = RecordingTransport()
hbp = HBPProtocol("MASTER-A", config, router=InMemoryAclRouter())
hbp.transport = transport # type: ignore[assignment]
scenario.protocols["MASTER-A"] = hbp
def _send_to_system(target: str, packet: bytes, **kwargs: Any) -> None:
scenario.capture.recorder(target)(
packet,
hops=kwargs.get("_hops", kwargs.get("hops", b"")),
ber=kwargs.get("_ber", kwargs.get("ber", b"\x00")),
rssi=kwargs.get("_rssi", kwargs.get("rssi", b"\x00")),
source_server=kwargs.get(
"_source_server", kwargs.get("source_server", b"\x00\x00\x00\x00"),
),
source_rptr=kwargs.get(
"_source_rptr", kwargs.get("source_rptr", b"\x00\x00\x00\x00"),
),
)
proto = scenario.protocols.get(target)
if proto is not None and hasattr(proto, "send_system"):
proto.send_system(
packet,
_hops=kwargs.get("_hops", b""),
_ber=kwargs.get("_ber", b"\x00"),
_rssi=kwargs.get("_rssi", b"\x00"),
_source_server=kwargs.get("_source_server", b"\x00\x00\x00\x00"),
_source_rptr=kwargs.get("_source_rptr", b"\x00\x00\x00\x00"),
)
scenario.routing._send_to_system = _send_to_system
hs = bytes_4(_HS_PEER)
hbp._peers[hs] = {
"CONNECTION": "YES",
"CONNECTED": scenario.clock.time(),
"LAST_PING": scenario.clock.time(),
"SOCKADDR": ("127.0.0.1", 62031),
"OPTIONS": f"TS2={_TG_A},{_TG_B};SINGLE=0;".encode(),
}
master.setdefault("PEERS", {})[hs] = hbp._peers[hs]
hbp._refresh_connected_peer_count()
hbp._mark_downlink_index_dirty()
return scenario, hbp, transport
def _base_spec(*, tgid: int, stream_id: int, rf_src: int) -> PacketSpec:
return PacketSpec(
peer_id=_OBP_PEER,
rf_src=rf_src,
dst_id=tgid,
slot=1,
stream_id=stream_id,
)
def _dmrd_packet(
*,
tgid: int,
stream_id: int,
rf_src: int,
frame_type: int,
dtype_vseq: int,
seq: int = 0,
) -> bytes:
base = _base_spec(tgid=tgid, stream_id=stream_id, rf_src=rf_src)
spec = PacketSpec(
peer_id=base.peer_id,
rf_src=base.rf_src,
dst_id=base.dst_id,
slot=base.slot,
stream_id=base.stream_id,
seq=seq,
frame_type=frame_type,
dtype_vseq=dtype_vseq,
payload=base.payload,
)
return spec.data()
def _is_voice_burst(packet: bytes) -> bool:
return ((packet[15] & 0x30) >> 4) == HBPF_VOICE
def _parse_dmrd_args(packet: bytes) -> dict[str, Any]:
bits = packet[15]
return {
"peer_id": packet[11:15],
"rf_src": packet[5:8],
"dst_id": packet[8:11],
"seq": packet[4],
"slot": 2 if bits & 0x80 else 1,
"call_type": "group",
"frame_type": (bits & 0x30) >> 4,
"dtype_vseq": bits & 0xF,
"stream_id": packet[16:20],
"data": packet,
}
def _wrap_send_peer_trace(hbp: HBPProtocol) -> dict[str, Any]:
"""Record per-packet send_peer accept/reject without modifying production code."""
trace: dict[str, Any] = {"accepted": [], "rejected": []}
orig = hbp.send_peer
def _traced(peer_id: bytes, packet: bytes, *, _skip_dual_expand: bool = False) -> None:
route_pkt = packet
peer = hbp._peers.get(peer_id)
if peer is not None and packet[:4] == b"DMRD":
from adn_server.application.routing.downlink import remap_dmrd_for_peer
route_pkt = remap_dmrd_for_peer(packet, peer, hbp._config, peer_id=peer_id)
opts_ok = hbp._peer_should_receive_dmrd(peer_id, packet)
gate_ok = hbp._peer_would_accept_group_dmrd(
peer_id, packet if not _skip_dual_expand else route_pkt, routed=_skip_dual_expand,
)
row = {
"tgid": int_id(route_pkt[8:11]),
"stream": route_pkt[16:20],
"opts_ok": opts_ok,
"gate_ok": gate_ok,
"blocked": not (opts_ok and (gate_ok if packet[:4] == b"DMRD" else True)),
}
if row["blocked"]:
trace["rejected"].append(row)
else:
trace["accepted"].append(row)
orig(peer_id, packet, _skip_dual_expand=_skip_dual_expand)
hbp.send_peer = _traced # type: ignore[method-assign]
return trace
bits = packet[15]
return {
"peer_id": packet[11:15],
"rf_src": packet[5:8],
"dst_id": packet[8:11],
"seq": packet[4],
"slot": 2 if bits & 0x80 else 1,
"call_type": "group",
"frame_type": (bits & 0x30) >> 4,
"dtype_vseq": bits & 0xF,
"stream_id": packet[16:20],
"data": packet,
}
def _inject_obp_at(
scenario: DeterministicScenario,
packet: bytes,
*,
pkt_time: float,
) -> None:
scenario.clock.now = pkt_time
with _patch_harness_time(scenario.clock), patch_routing_wall_time(scenario.clock):
scenario.routing.dmrd_received(
"OBP-CL",
ingress_pkt_time=pkt_time,
obp_use_parsed=True,
obp_hops=b"\x00\x00\x00\x00",
obp_source_server=bytes_4(9990),
**_parse_dmrd_args(packet),
)
def _run_interleaved_qsos(
scenario: DeterministicScenario,
*,
voice_bursts: int = _VOICE_BURSTS,
interval_s: float = _INTERVAL_S,
) -> dict[str, Any]:
t0 = scenario.clock.time()
step = 0
tx_stamp_changes = 0
start_tx_events = 0
prev_tx_stream: bytes | None = None
hs_addr = ("127.0.0.1", 62031)
delivered_a: list[bytes] = []
delivered_b: list[bytes] = []
blocked_a_checks: list[dict[str, Any]] = []
def _slot_st() -> dict[str, Any]:
proto = scenario.protocols["MASTER-A"]
assert isinstance(proto, HBPProtocol)
return proto.STATUS.setdefault(2, {})
def _record_delivery(pkt: bytes) -> None:
tgid = int_id(pkt[8:11])
sid = pkt[16:20]
if tgid == _TG_A and sid == bytes_4(_STREAM_A):
delivered_a.append(pkt)
elif tgid == _TG_B and sid == bytes_4(_STREAM_B):
delivered_b.append(pkt)
packets_plan: list[tuple[str, bytes]] = []
for label, tgid, stream, rf in (
("A", _TG_A, _STREAM_A, _RF_A),
("B", _TG_B, _STREAM_B, _RF_B),
):
packets_plan.append(
(
label,
_dmrd_packet(
tgid=tgid,
stream_id=stream,
rf_src=rf,
frame_type=HBPF_DATA_SYNC,
dtype_vseq=HBPF_SLT_VHEAD,
),
)
)
for seq in range(1, voice_bursts + 1):
for label, tgid, stream, rf in (
("A", _TG_A, _STREAM_A, _RF_A),
("B", _TG_B, _STREAM_B, _RF_B),
):
packets_plan.append(
(
label,
_dmrd_packet(
tgid=tgid,
stream_id=stream,
rf_src=rf,
frame_type=HBPF_VOICE,
dtype_vseq=(seq % 4) or 1,
seq=seq,
),
)
)
for label, tgid, stream, rf in (
("A", _TG_A, _STREAM_A, _RF_A),
("B", _TG_B, _STREAM_B, _RF_B),
):
packets_plan.append(
(
label,
_dmrd_packet(
tgid=tgid,
stream_id=stream,
rf_src=rf,
frame_type=HBPF_DATA_SYNC,
dtype_vseq=HBPF_SLT_VTERM,
seq=99,
),
)
)
scenario.capture.packets.clear()
transport = scenario.protocols["MASTER-A"]
assert isinstance(transport, HBPProtocol)
rec = transport.transport
assert isinstance(rec, RecordingTransport)
rec.clear()
hbp = transport
hs = bytes_4(_HS_PEER)
peer = hbp._peers[hs]
for label, pkt in packets_plan:
pkt_time = t0 + step * interval_s
step += 1
_inject_obp_at(scenario, pkt, pkt_time=pkt_time)
cur = _slot_st().get("TX_STREAM_ID")
if cur != prev_tx_stream:
tx_stamp_changes += 1
prev_tx_stream = cur
bits = pkt[15] | 0x80
route_pkt = pkt[:15] + bytes([bits]) + pkt[16:]
ctx = hbp._downlink_ctx()
if (
label == "A"
and int_id(pkt[8:11]) == _TG_A
and pkt[16:20] == bytes_4(_STREAM_A)
and (pkt[15] & 0xF) != HBPF_SLT_VTERM
):
blocked = peer_slot_blocks_downlink(ctx, hs, peer, route_pkt, pkt_time=pkt_time)
if blocked:
blocked_a_checks.append(
{
"pkt_time": pkt_time,
"step": step - 1,
"slot_st": copy.deepcopy(_slot_st()),
"peer_slots": copy.deepcopy(ctx.peer_voice_slots.get(hs, {})),
"reason": diagnose_downlink_block(ctx, hs, peer, route_pkt, pkt_time),
}
)
for sent_pkt, addr in rec.sent:
if addr == hs_addr:
_record_delivery(sent_pkt)
rec.clear()
return {
"tx_stamp_changes": tx_stamp_changes,
"start_tx_events": len(
[ev for ev in (scenario.report_factory.events if scenario.report_factory else []) if ",START,TX," in ev]
),
"delivered_a": delivered_a,
"delivered_b": delivered_b,
"blocked_a_checks": blocked_a_checks,
"final_slot_st": copy.deepcopy(_slot_st()),
}
def diagnose_downlink_block(
ctx: DownlinkContext,
peer_id: bytes,
peer: dict[str, Any],
route_pkt: bytes,
pkt_time: float,
) -> str:
"""Pin which gate branch would block this downlink (mirrors helpers/downlink)."""
if route_pkt[:4] != b"DMRD":
return "not_dmrd"
stream_id = route_pkt[16:20]
incoming_tgid_b = route_pkt[8:11]
hang = float(ctx.sys_cfg.get("GROUP_HANGTIME", 0) or 0)
peer_slots = ctx.peer_voice_slots.get(bytes_4(int_id(peer_id)))
pk = bytes_4(int_id(peer_id))
for voice_slot in (2,):
hang_row = ctx.peer_voice_hangtime.get(pk, {}).get(voice_slot)
slot_st = ctx.status.get(voice_slot, {})
if _peer_transmit_hangtime_blocks(hang_row, incoming_tgid_b, pkt_time, hang):
return f"helpers.py:_peer_transmit_hangtime_blocks voice_slot={voice_slot}"
active = (peer_slots or {}).get(voice_slot)
if isinstance(active, dict):
incoming_tgid = int_id(incoming_tgid_b)
active_tgid = int(active.get("tgid", 0) or 0)
active_stream = active.get("stream_id")
active_time = float(active.get("time", 0) or 0)
age = pkt_time - active_time
if active.get("ingress"):
return f"helpers.py:peer_hotspot_voice_slot_busy ingress voice_slot={voice_slot}"
if active.get("bridge_hold") and active_tgid and incoming_tgid != active_tgid:
if age <= hang:
return (
f"helpers.py:peer_hotspot_voice_slot_busy bridge_hold "
f"active_tg={active_tgid} incoming={incoming_tgid}"
)
if active_stream and stream_id:
if active_stream == stream_id:
pass
elif active_tgid and active_tgid == incoming_tgid:
if age < STREAM_TO:
return (
f"helpers.py:peer_hotspot_voice_slot_busy "
f"same_tg_different_stream active={active_stream!r} incoming={stream_id!r}"
)
elif active_tgid and incoming_tgid != active_tgid:
return (
f"helpers.py:peer_hotspot_voice_slot_busy "
f"elif active_tgid={active_tgid} incoming_tgid={incoming_tgid}"
)
else:
return "helpers.py:peer_hotspot_voice_slot_busy else branch (stream/tg mismatch)"
elif isinstance(active, dict):
if not (active_tgid and active_tgid == incoming_tgid):
return (
f"helpers.py:peer_hotspot_voice_slot_busy "
f"no_active_stream active_tg={active_tgid} incoming={incoming_tgid}"
)
if peer_single_blocks_foreign_same_tg_downlink(
peer, pk, voice_slot, incoming_tgid_b, peer_slots, ctx.sys_cfg, now=pkt_time,
):
return f"helpers.py:peer_single_blocks_foreign_same_tg_downlink voice_slot={voice_slot}"
if peer_single_same_tg_foreign_tx_blocks(
peer, pk, incoming_tgid_b, stream_id, slot_st, ctx.sys_cfg, pkt_time=pkt_time,
):
return (
f"helpers.py:peer_single_same_tg_foreign_tx_blocks "
f"TX_STREAM_ID={slot_st.get('TX_STREAM_ID')!r} slot_active={slot_has_active_voice(slot_st, pkt_time)}"
)
if bytes_4(int_id(slot_st.get("RX_PEER", b""))) == pk:
rx_active = (
slot_st.get("RX_TYPE") is not None
and slot_st.get("RX_TYPE") != HBPF_SLT_VTERM
and (pkt_time - float(slot_st.get("RX_TIME", 0))) < STREAM_TO
)
if rx_active and stream_id != slot_st.get("RX_STREAM_ID"):
return (
f"helpers.py:peer_hotspot_voice_slot_busy "
f"rx_active stream_mismatch RX_STREAM_ID={slot_st.get('RX_STREAM_ID')!r}"
)
if _peer_status_rx_hangtime_blocks(
peer_id, slot_st, incoming_tgid_b, pkt_time, hang,
):
return f"helpers.py:_peer_status_rx_hangtime_blocks voice_slot={voice_slot}"
if hbp_slot_blocks_group_voice_for_peer(
slot_st,
peer_id,
incoming_tgid_b,
stream_id,
pkt_time,
hang,
per_peer=True,
peers=ctx.peers,
peer_slots=peer_slots,
peer_hang_row=hang_row,
voice_slot=voice_slot,
sys_cfg=ctx.sys_cfg,
):
return f"helpers.py:hbp_slot_blocks_group_voice_for_peer voice_slot={voice_slot}"
if peer_slot_blocks_downlink(ctx, peer_id, peer, route_pkt, pkt_time=pkt_time):
return "downlink.py:peer_slot_blocks_downlink (composite)"
return "not_blocked"
def test_flip_flop_slot_st_blocks_active_listen_on_cross_tg() -> None:
"""Pure unit: peer listening on TG A must not RX TG B while session is open."""
now = 1_000_000.0
hs = bytes_4(_HS_PEER)
peer = {"OPTIONS": f"TS2={_TG_A},{_TG_B};SINGLE=0;".encode()}
stream_a = bytes_4(_STREAM_A)
stream_b = bytes_4(_STREAM_B)
peer_slots = {
2: {"stream_id": stream_a, "tgid": _TG_A, "time": now, "ingress": False},
}
slot_st = {
"TX_PEER": bytes_4(_OBP_PEER),
"TX_STREAM_ID": stream_b,
"TX_TYPE": HBPF_SLT_VHEAD,
"TX_TIME": now,
"TX_TGID": bytes_3(_TG_B),
"RX_TYPE": HBPF_SLT_VTERM,
}
assert peer_hotspot_voice_slot_busy(
hs, 2, stream_b, bytes_3(_TG_B), slot_st, peer_slots, None, now + 0.06, 5.0,
peer=peer, sys_cfg={"GROUP_HANGTIME": 5.0, "SINGLE_MODE": False},
)
route_b = _dmrd_packet(
tgid=_TG_B,
stream_id=_STREAM_B,
rf_src=_RF_B,
frame_type=HBPF_VOICE,
dtype_vseq=1,
seq=1,
)
route_b = route_b[:15] + bytes([route_b[15] | 0x80]) + route_b[16:]
ctx = DownlinkContext(
config={"SYSTEMS": {"MASTER-A": {"MODE": "MASTER"}}},
system_name="MASTER-A",
sys_cfg={"GROUP_HANGTIME": 5.0, "SINGLE_MODE": False, "MODE": "MASTER"},
peers={hs: peer},
status={2: slot_st},
peer_voice_slots={hs: copy.deepcopy(peer_slots)},
connected_count=2,
)
reason = diagnose_downlink_block(ctx, hs, peer, route_b, now + 0.06)
assert "active_tgid=3109050 incoming_tgid=52090" in reason
def test_flip_flop_slot_st_same_stream_not_blocked_despite_foreign_tx_stamp() -> None:
"""Same stream id on active listen must pass even when flat TX row shows other stream."""
now = 1_000_000.0
hs = bytes_4(_HS_PEER)
peer = {"OPTIONS": f"TS2={_TG_A},{_TG_B};SINGLE=0;".encode()}
stream_a = bytes_4(_STREAM_A)
stream_b = bytes_4(_STREAM_B)
peer_slots = {
2: {"stream_id": stream_a, "tgid": _TG_A, "time": now, "ingress": False},
}
slot_st = {
"TX_PEER": bytes_4(_OBP_PEER),
"TX_STREAM_ID": stream_b,
"TX_TYPE": HBPF_SLT_VHEAD,
"TX_TIME": now,
"TX_TGID": bytes_3(_TG_B),
"RX_TYPE": HBPF_SLT_VTERM,
}
assert not peer_hotspot_voice_slot_busy(
hs, 2, stream_a, bytes_3(_TG_A), slot_st, peer_slots, None, now + 0.06, 5.0,
peer=peer, sys_cfg={"GROUP_HANGTIME": 5.0, "SINGLE_MODE": False},
)
def test_rx_active_own_peer_foreign_stream_blocks_active_downlink() -> None:
"""When STATUS RX shows the hotspot mid-TX on another stream, stream-A downlink blocks.
helpers.py peer_hotspot_voice_slot_busy lines 411-418:
``rx_active and stream_id != slot_st.get('RX_STREAM_ID')`` → True.
"""
now = 1_000_000.0
hs = bytes_4(_HS_PEER)
peer = {"OPTIONS": f"TS2={_TG_A},{_TG_B};SINGLE=0;".encode()}
stream_a = bytes_4(_STREAM_A)
stream_b = bytes_4(_STREAM_B)
peer_slots = {
2: {"stream_id": stream_a, "tgid": _TG_A, "time": now, "ingress": False},
}
slot_st = {
"RX_PEER": hs,
"RX_STREAM_ID": stream_b,
"RX_TYPE": HBPF_SLT_VHEAD,
"RX_TIME": now,
"RX_TGID": bytes_3(_TG_B),
"TX_PEER": bytes_4(_OBP_PEER),
"TX_STREAM_ID": stream_a,
"TX_TYPE": HBPF_SLT_VHEAD,
"TX_TIME": now,
"TX_TGID": bytes_3(_TG_A),
}
route_a = _dmrd_packet(
tgid=_TG_A,
stream_id=_STREAM_A,
rf_src=_RF_A,
frame_type=HBPF_VOICE,
dtype_vseq=1,
seq=1,
)
route_a = route_a[:15] + bytes([route_a[15] | 0x80]) + route_a[16:]
ctx = DownlinkContext(
config={"SYSTEMS": {"MASTER-A": {"MODE": "MASTER"}}},
system_name="MASTER-A",
sys_cfg={"GROUP_HANGTIME": 5.0, "SINGLE_MODE": False, "MODE": "MASTER"},
peers={hs: peer},
status={2: slot_st},
peer_voice_slots={hs: copy.deepcopy(peer_slots)},
connected_count=2,
)
reason = diagnose_downlink_block(ctx, hs, peer, route_a, now + 0.06)
assert "rx_active stream_mismatch" in reason
assert peer_slot_blocks_downlink(ctx, hs, peer, route_a, pkt_time=now + 0.06)
def test_concurrent_obp_tx_row_flip_flop(caplog: pytest.LogCaptureFixture) -> None:
"""BUG: shared STATUS[2] TX_STREAM_ID must change at most twice per stream (VHEAD), not per packet."""
caplog.set_level(logging.INFO)
scenario, _hbp, _transport = _build_stack()
result = _run_interleaved_qsos(scenario, voice_bursts=6)
# Expect: 2 streams × 1 VHEAD stamp each (+ maybe VTERM) — not ~2×packets.
max_expected_stamps = 6
assert result["tx_stamp_changes"] <= max_expected_stamps, (
f"TX_STREAM_ID flip-flop: {result['tx_stamp_changes']} changes "
f"(expected <= {max_expected_stamps}); final STATUS[2]={result['final_slot_st']}"
)
assert result["start_tx_events"] <= max_expected_stamps, (
f"START,TX report spam: {result['start_tx_events']} events"
)
def test_concurrent_obp_active_stream_delivered_without_gaps() -> None:
"""After VHEAD, stream-A voice should not be dropped mid-QSO (minimal harness)."""
scenario, hbp, transport = _build_stack()
send_trace = _wrap_send_peer_trace(hbp)
result = _run_interleaved_qsos(scenario, voice_bursts=_VOICE_BURSTS)
voice_a = [p for p in result["delivered_a"] if _is_voice_burst(p)]
rejected_a = [r for r in send_trace["rejected"] if r["tgid"] == _TG_A]
expected_voice = _VOICE_BURSTS
# Minimal OBP-only harness delivers all A packets; production drops use RX/status
# corruption paths (see test_rx_active_own_peer_foreign_stream_blocks_active_downlink).
assert not rejected_a, f"unexpected stream-A send_peer rejects: {rejected_a}"
assert len(voice_a) == expected_voice, (
f"stream A voice delivered {len(voice_a)}/{expected_voice}"
)
# Stream B may be dropped for the listening hotspot — not asserted.
def test_concurrent_obp_stream_b_blocked_on_listen_session() -> None:
"""Stream B must be dropped while hotspot listens to stream A (one QSO per slot)."""
scenario, hbp, transport = _build_stack()
send_trace = _wrap_send_peer_trace(hbp)
_run_interleaved_qsos(scenario, voice_bursts=4)
rejected_b = [r for r in send_trace["rejected"] if r["tgid"] == _TG_B]
assert rejected_b, "stream B should be blocked for listening hotspot"
ctx = hbp._downlink_ctx()
hs = bytes_4(_HS_PEER)
peer = hbp._peers[hs]
sample = _dmrd_packet(
tgid=_TG_B, stream_id=_STREAM_B, rf_src=_RF_B,
frame_type=HBPF_VOICE, dtype_vseq=1, seq=1,
)
sample = sample[:15] + bytes([sample[15] | 0x80]) + sample[16:]
reason = diagnose_downlink_block(ctx, hs, peer, sample, scenario.clock.time())
assert (
"active_tgid=3109050 incoming_tgid=52090" in reason
or "_peer_transmit_hangtime_blocks" in reason
or "peer_hotspot" in reason
or "peer_slot" in reason
), f"stream B blocked but unexpected gate: {reason}"
def test_concurrent_obp_diagnose_active_stream_block_branch() -> None:
"""Identify the exact branch that drops active stream-A packets mid-QSO."""
scenario, hbp, transport = _build_stack()
send_trace = _wrap_send_peer_trace(hbp)
result = _run_interleaved_qsos(scenario, voice_bursts=_VOICE_BURSTS)
rejected_a = [
r for r in send_trace["rejected"]
if r["tgid"] == _TG_A and r["stream"] == bytes_4(_STREAM_A)
]
if not rejected_a and not result["blocked_a_checks"]:
# After per-leg OBP bridge TX wiring, flat TX_STREAM_ID no longer flip-flops per packet.
if result["tx_stamp_changes"] <= 4:
pytest.skip("no stream-A rejects; per-leg bridge TX fix removed flat-row flip-flop")
assert result["tx_stamp_changes"] > 4, "expected TX row flip-flop in concurrent OBP harness"
pytest.skip("no stream-A send_peer rejects — inspect flip-flop collateral (START,TX spam)")
reasons: dict[str, int] = {}
for row in result["blocked_a_checks"]:
reasons[row["reason"]] = reasons.get(row["reason"], 0) + 1
for rej in rejected_a:
if not rej["opts_ok"]:
key = "udp_hbp.py:_peer_should_receive_dmrd OPTIONS/eligibility"
elif not rej["gate_ok"]:
key = "udp_hbp.py:_peer_would_accept_group_dmrd -> peer_slot_blocks_downlink"
else:
key = "unknown"
reasons[key] = reasons.get(key, 0) + 1
top = max(reasons, key=reasons.get)
assert "peer_hotspot" in top or "peer_slot" in top or "send_peer" in top or "udp_hbp" in top, (
f"unexpected block reasons: {reasons}"
)
Write Preview
Loading…
Cancel
Save

Powered by TurnKey Linux.