From d886af0991a5be8a8cf20e83a3d3ba2a5182a6ac Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 16 Jul 2026 10:48:52 -0700 Subject: [PATCH] fix(engine): relay routed ACKs through their MeshCore branch An ACK at an intermediate direct hop is now regenerated as a plain DIRECT ACK (transport codes dropped, version bits cleared) and relayed with zero retransmit delay, matching routeDirectRecvAcks, instead of taking the generic delayed direct forward. The repeater.multi_acks preference (already exposed by the mesh CLI) is now consumed by the engine: when enabled, both ACK relay sites precede the plain ACK with a MULTIPART-wrapped redundancy copy spaced by the direct retransmit delay + 300 ms. The multipart relay seen-key now keeps the MULTIPART payload type over the unwrapped payload, matching firmware hasSeen so a redundancy pair survives each hop. --- config.yaml.example | 5 + repeater/engine.py | 197 ++++++++++++++++++++++++--- tests/test_engine.py | 312 +++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 481 insertions(+), 33 deletions(-) diff --git a/config.yaml.example b/config.yaml.example index d8e17f2..f404dd7 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -35,6 +35,11 @@ repeater: # Enable quality-based packet filtering and adaptive delays use_score_for_tx: false + # Multi-ack redundancy (0 = off, 1 = on; matches MeshCore's multi.acks pref). + # When on, a relayed routed ACK is preceded by a MULTIPART-wrapped copy so + # ACKs survive lossy links. Also settable via the mesh CLI: set multi.acks + multi_acks: 0 + # Score threshold for quality monitoring (future use) # Currently reserved for potential future features like dashboard alerts, # proactive statistics collection, or advanced filtering strategies. diff --git a/repeater/engine.py b/repeater/engine.py index cbcd04f..d0324fc 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -23,7 +23,12 @@ from openhop_core.protocol.constants import ( ROUTE_TYPE_TRANSPORT_DIRECT, ROUTE_TYPE_TRANSPORT_FLOOD, ) -from openhop_core.protocol.packet_utils import PacketHeaderUtils, PathUtils, packet_score +from openhop_core.protocol.packet_utils import ( + PacketHashingUtils, + PacketHeaderUtils, + PathUtils, + packet_score, +) from repeater.airtime import AirtimeManager from repeater.data_acquisition import StorageCollector @@ -83,6 +88,22 @@ class DropReason(str, Enum): return self.value +class ForwardResult(tuple): + """A forwarding decision that unpacks as a plain ``(packet, delay_seconds)`` pair. + + ``extras`` carries additional ``(packet, delay_seconds)`` transmissions that + accompany the primary packet: MeshCore's routeDirectRecvAcks queues its + optional multi-ack redundancy copies ahead of the plain ACK at the same + scheduled time, so the caller must create the extra TX tasks before the + primary one. + """ + + def __new__(cls, packet: Packet, delay_s: float, extras=()): + result = super().__new__(cls, (packet, delay_s)) + result.extras = tuple(extras) + return result + + class RepeaterHandler(BaseHandler): @staticmethod def payload_type() -> int: @@ -124,6 +145,7 @@ class RepeaterHandler(BaseHandler): ) self.use_score_for_tx = config.get("repeater", {}).get("use_score_for_tx", False) self.score_threshold = config.get("repeater", {}).get("score_threshold", 0.3) + self.multi_acks = self._normalize_multi_acks(config) self.max_flood_hops = config.get("repeater", {}).get("max_flood_hops", 64) self.send_advert_interval_hours = config.get("repeater", {}).get( "send_advert_interval_hours", 10 @@ -341,6 +363,16 @@ class RepeaterHandler(BaseHandler): # Capture the forwarded path (after modification) forwarded_path_hashes = fwd_pkt.get_path_hashes_hex() + # MeshCore queues multi-ack redundancy copies ahead of the primary + # ACK at the same scheduled time, so create their TX tasks first. + # Each task re-checks the duty-cycle gate before transmitting. + extra_tx_tasks = [] + for extra_pkt, extra_delay in getattr(result, "extras", ()): + extra_airtime_ms = self.airtime_mgr.calculate_airtime(extra_pkt.get_raw_length()) + extra_tx_tasks.append( + await self.schedule_retransmit(extra_pkt, extra_delay, extra_airtime_ms) + ) + # Check duty-cycle before scheduling TX airtime_ms = self.airtime_mgr.calculate_airtime(fwd_pkt.get_raw_length()) @@ -426,6 +458,15 @@ class RepeaterHandler(BaseHandler): f"LBT: {lbt_attempts} attempts, {total_lbt_delay:.0f}ms delay, " f"backoffs={lbt_backoff_delays_ms}" ) + + # Redundancy copies ride alongside the primary result: collect them + # without letting a failed extra change the primary outcome. + for extra_task in extra_tx_tasks: + try: + if await extra_task: + self.forwarded_count += 1 + except Exception as e: + logger.warning(f"Multi-ack redundancy TX failed: {e}") else: self.dropped_count += 1 # Determine drop reason @@ -885,6 +926,16 @@ class RepeaterHandler(BaseHandler): return True, "" + @staticmethod + def _normalize_multi_acks(config: dict) -> int: + """Read ``repeater.multi_acks``, constrained to 0..1 like MeshCore's + CommonCLI prefs load.""" + try: + value = int(config.get("repeater", {}).get("multi_acks", 0)) + except (TypeError, ValueError): + return 0 + return max(0, min(1, value)) + def _normalize_loop_detect_mode(self, mode) -> str: if isinstance(mode, str): normalized = mode.strip().lower() @@ -1138,8 +1189,8 @@ class RepeaterHandler(BaseHandler): MULTIPART_ACK_SPACING_MS = 300 def forward_multipart_direct( - self, packet: Packet, packet_hash: Optional[str] = None - ) -> Optional[Tuple[Packet, float]]: + self, packet: Packet, snr: float = 0.0, packet_hash: Optional[str] = None + ) -> Optional["ForwardResult"]: """Forward a MULTIPART packet the way MeshCore's forwardMultipartDirect does. MeshCore never relays the multipart wrapper as an ordinary directed @@ -1147,7 +1198,8 @@ class RepeaterHandler(BaseHandler): and it is relayed by regenerating the embedded ACK as a plain DIRECT ACK (the wrapper byte stripped) rather than repeating the wrapper. Every other multipart case — flood routing, a final hop, or a non-ACK embedded - type — is dropped. Returns ``(ack_packet, delay_seconds)`` when a + type — is dropped. Returns a ``ForwardResult`` — ``(ack_packet, + delay_seconds)`` plus any multi-ack redundancy ``extras`` — when a regenerated ACK should be sent, otherwise ``None``. INVARIANT: purely synchronous, mirroring flood_forward / direct_forward — @@ -1189,27 +1241,121 @@ class RepeaterHandler(BaseHandler): packet.drop_reason = DropReason.DIRECT_NOT_FOR_US return None + # Dedupe before regenerating, on MeshCore's exact seen key: hasSeen() + # runs on a copy that keeps the MULTIPART payload type but carries the + # unwrapped ACK payload (the `remaining` count is stripped). Keeping + # that key distinct from the plain ACK's is what lets a multi-ack + # redundancy pair — MULTIPART copy plus plain ACK — survive every hop + # instead of the second half being swallowed as a duplicate. The + # pre-computed hash belongs to the wrapper as received, so recompute. + seen_key = ( + PacketHashingUtils.calculate_packet_hash( + PAYLOAD_TYPE_MULTIPART, packet.path_len, bytes(payload[1:]) + ) + .hex() + .upper() + ) + if self.is_duplicate(packet, packet_hash=seen_key): + packet.drop_reason = DropReason.DUPLICATE + return None + self.mark_seen(packet, packet_hash=seen_key) + # Rebuild the received wrapper as the plain DIRECT ACK MeshCore would emit: - # drop the multipart header byte and force the ACK payload/route type. The - # path is left intact for the seen-check, then this node is removed from it, - # matching forwardMultipartDirect's hasSeen()/removeSelfFromPath() order. + # drop the multipart header byte and force the ACK payload/route type, then + # remove this node from the path (hasSeen()/removeSelfFromPath() order). packet.payload = bytearray(payload[1:]) packet.payload_len = len(packet.payload) packet.header = (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT - - # Dedupe on the regenerated ACK form (full path) before removing ourselves, - # so a repeated multipart ACK — or an equivalent plain ACK — is not relayed - # twice. The pre-computed hash belongs to the original wrapper, so recompute. - if self.is_duplicate(packet): - packet.drop_reason = DropReason.DUPLICATE - return None - self.mark_seen(packet) - + packet.transport_codes = [0, 0] packet.path = bytearray(packet.path[hash_size:]) packet.path_len = PathUtils.encode_path_len(hash_size, hop_count - 1) - delay_s = ((remaining + 1) * self.MULTIPART_ACK_SPACING_MS) / 1000.0 - return packet, delay_s + extras, delay_ms = self._multi_ack_extras( + packet, float((remaining + 1) * self.MULTIPART_ACK_SPACING_MS), snr + ) + return ForwardResult(packet, delay_ms / 1000.0, extras) + + def _multi_ack_extras( + self, ack_packet: Packet, base_delay_ms: float, snr: float + ) -> Tuple[list, float]: + """Build MeshCore's optional multi-ack redundancy copies for a relayed ACK. + + With ``multi_acks`` enabled, routeDirectRecvAcks precedes the plain ACK + with a MULTIPART-wrapped copy (``remaining<<4 | ACK`` prefix byte) on the + already self-removed path, pushing the accumulated delay out by a direct + retransmit delay + 300 ms per copy; the plain ACK itself slides to the + final accumulated delay. Returns ``(extras, plain_ack_delay_ms)``. + """ + extra = self.multi_acks + extras = [] + delay_ms = base_delay_ms + while extra > 0: + delay_ms += ( + self._calculate_tx_delay(ack_packet, snr) * 1000.0 + self.MULTIPART_ACK_SPACING_MS + ) + wrapped = Packet() + wrapped.header = (PAYLOAD_TYPE_MULTIPART << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT + wrapped.payload = bytearray( + bytes([(extra << 4) | PAYLOAD_TYPE_ACK]) + bytes(ack_packet.payload) + ) + wrapped.payload_len = len(wrapped.payload) + wrapped.path = bytearray(ack_packet.path) + wrapped.path_len = ack_packet.path_len + extras.append((wrapped, delay_ms / 1000.0)) + extra -= 1 + return extras, delay_ms + + def forward_routed_ack( + self, packet: Packet, snr: float = 0.0, packet_hash: Optional[str] = None + ) -> Optional["ForwardResult"]: + """Relay a routed ACK the way MeshCore's routeDirectRecvAcks does. + + An ACK at an intermediate direct hop is not repeated verbatim: MeshCore + regenerates it (createAck) as a plain DIRECT ACK — transport codes + dropped, every other header bit cleared — removes this node from the + path, and transmits it with zero retransmit delay rather than the + generic direct forwarding delay. With ``multi_acks`` enabled, the plain + ACK is preceded by a MULTIPART-wrapped redundancy copy and both slide + out by a direct retransmit delay + 300 ms. + + INVARIANT: purely synchronous, mirroring direct_forward — the + is_duplicate + mark_seen pair must stay atomic within the event loop. + """ + valid, reason = self.validate_packet(packet) + if not valid: + packet.drop_reason = reason + return None + + if packet.is_marked_do_not_retransmit(): + packet.drop_reason = packet.drop_reason or DropReason.MARKED_DO_NOT_RETRANSMIT + return None + + hash_size = packet.get_path_hash_size() + hop_count = packet.get_path_hash_count() + + # A final-hop ACK (no remaining path) is consumed locally, never relayed. + if not packet.path or len(packet.path) < hash_size: + packet.drop_reason = DropReason.DIRECT_NO_PATH + return None + if bytes(packet.path[:hash_size]) != self.local_hash_bytes[:hash_size]: + packet.drop_reason = DropReason.DIRECT_NOT_FOR_US + return None + + # The packet hash covers only the payload type and payload, so the + # received form and the regenerated plain ACK share one key — matching + # MeshCore's hasSeen() on the incoming ACK before removeSelfFromPath(). + if self.is_duplicate(packet, packet_hash=packet_hash): + packet.drop_reason = DropReason.DUPLICATE + return None + self.mark_seen(packet, packet_hash=packet_hash) + + packet.header = (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT + packet.transport_codes = [0, 0] + packet.path = bytearray(packet.path[hash_size:]) + packet.path_len = PathUtils.encode_path_len(hash_size, hop_count - 1) + + extras, delay_ms = self._multi_ack_extras(packet, 0.0, snr) + return ForwardResult(packet, delay_ms / 1000.0, extras) @staticmethod def calculate_packet_score(snr: float, packet_len: int, spreading_factor: int = 8) -> float: @@ -1279,8 +1425,18 @@ class RepeaterHandler(BaseHandler): # generic flood/direct path — it regenerates embedded ACKs and drops # everything else. Its delay is fixed by the fragment count, so return # the (packet, delay) pair directly instead of _calculate_tx_delay. - if packet.get_payload_type() == PAYLOAD_TYPE_MULTIPART: - return self.forward_multipart_direct(packet, packet_hash=packet_hash) + payload_type = packet.get_payload_type() + if payload_type == PAYLOAD_TYPE_MULTIPART: + return self.forward_multipart_direct(packet, snr, packet_hash=packet_hash) + + # Routed ACKs also get their own MeshCore branch: an intermediate direct + # hop re-emits the ACK immediately (routeDirectRecvAcks) instead of the + # delayed generic direct forward. Flood ACKs stay on the generic path. + if payload_type == PAYLOAD_TYPE_ACK and route_type in ( + ROUTE_TYPE_DIRECT, + ROUTE_TYPE_TRANSPORT_DIRECT, + ): + return self.forward_routed_ack(packet, snr, packet_hash=packet_hash) if route_type == ROUTE_TYPE_FLOOD or route_type == ROUTE_TYPE_TRANSPORT_FLOOD: fwd_pkt = self.flood_forward(packet, packet_hash=packet_hash) @@ -1622,6 +1778,7 @@ class RepeaterHandler(BaseHandler): repeater_config = self.config.get("repeater", {}) self.use_score_for_tx = repeater_config.get("use_score_for_tx", False) self.score_threshold = repeater_config.get("score_threshold", 0.3) + self.multi_acks = self._normalize_multi_acks(self.config) self.send_advert_interval_hours = repeater_config.get("send_advert_interval_hours", 10) self.cache_ttl = repeater_config.get("cache_ttl", 60) self.max_flood_hops = repeater_config.get("max_flood_hops", 64) diff --git a/tests/test_engine.py b/tests/test_engine.py index 8829177..12df78e 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -38,6 +38,13 @@ LOCAL_HASH = 0xAB # repeater's own 1-byte path hash # (forward_multipart_direct), covered by TestForwardMultipartDirect. _GENERIC_FORWARD_PAYLOAD_TYPES = [pt for pt in range(16) if pt != PAYLOAD_TYPE_MULTIPART] +# On direct routes, ACK also leaves the generic path — routed ACKs are +# regenerated by forward_routed_ack (TestForwardRoutedAck); flood ACKs stay +# generic. +_GENERIC_DIRECT_FORWARD_PAYLOAD_TYPES = [ + pt for pt in _GENERIC_FORWARD_PAYLOAD_TYPES if pt != PAYLOAD_TYPE_ACK +] + def _make_config(**overrides) -> dict: """Return a minimal valid config dict for RepeaterHandler.""" @@ -477,21 +484,47 @@ class TestForwardMultipartDirect: assert result is None assert pkt2.drop_reason == "Duplicate" - def test_dedup_key_is_regenerated_ack_before_pop(self, handler): - """The seen key is the plain-ACK form with the full path, matching - MeshCore's hasSeen()-before-removeSelfFromPath() order.""" - pkt = _make_multipart_ack_packet(path=bytes([LOCAL_HASH, 0xCC])) - expected = Packet() - expected.header = ROUTE_TYPE_DIRECT | (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) - expected.payload = bytearray(b"\x4d\xab\xaf\x95") - expected.payload_len = 4 - expected.path = bytearray([LOCAL_HASH, 0xCC]) - expected.path_len = 2 - key = expected.calculate_packet_hash().hex().upper() + def test_dedup_key_keeps_multipart_type_with_unwrapped_payload(self, handler): + """The seen key matches MeshCore's hasSeen() copy: MULTIPART payload + type over the unwrapped ACK payload, checked before removeSelfFromPath. + The `remaining` count is stripped, so re-wraps of the same ACK dedupe.""" + from openhop_core.protocol.packet_utils import PacketHashingUtils + + pkt = _make_multipart_ack_packet(remaining=2, path=bytes([LOCAL_HASH, 0xCC])) + key = ( + PacketHashingUtils.calculate_packet_hash( + PAYLOAD_TYPE_MULTIPART, pkt.path_len, b"\x4d\xab\xaf\x95" + ) + .hex() + .upper() + ) handler.forward_multipart_direct(pkt) assert key in handler.seen_packets + # A re-wrap of the same ACK with a different remaining count is a duplicate. + rewrap = _make_multipart_ack_packet(remaining=1, path=bytes([LOCAL_HASH, 0xCC])) + assert handler.forward_multipart_direct(rewrap) is None + assert rewrap.drop_reason == "Duplicate" + + def test_multipart_relay_does_not_block_plain_ack_relay(self, handler): + """A multi-ack redundancy pair (MULTIPART copy + plain ACK) must survive + the hop: MeshCore's seen keys differ by payload type, so relaying the + MULTIPART copy first must not swallow the plain ACK as a duplicate.""" + wrapper = _make_multipart_ack_packet(remaining=1, path=bytes([LOCAL_HASH, 0xCC])) + assert handler.forward_multipart_direct(wrapper) is not None + + plain = _make_direct_packet( + payload=b"\x4d\xab\xaf\x95", + path=bytes([LOCAL_HASH, 0xCC]), + payload_type=PAYLOAD_TYPE_ACK, + ) + result = handler.forward_routed_ack(plain) + assert result is not None + ack, delay = result + assert delay == pytest.approx(0.0) + assert bytes(ack.payload) == b"\x4d\xab\xaf\x95" + def test_marked_do_not_retransmit_multipart_dropped(self, handler): pkt = _make_multipart_ack_packet() pkt.mark_do_not_retransmit() @@ -554,6 +587,259 @@ def test_forward_multipart_matches_firmware_wire_vectors(local_hash_bytes, wire_ assert delay == pytest.approx(0.9) +# =================================================================== +# 2c. forward_routed_ack — MeshCore routeDirectRecvAcks parity +# =================================================================== + + +def _make_direct_ack_packet( + crc: bytes = b"\x4d\xab\xaf\x95", + path: bytes = None, + route: int = ROUTE_TYPE_DIRECT, + transport_codes=None, +) -> Packet: + """Build a routed ACK: 4-byte CRC payload on a direct route.""" + if path is None: + path = bytes([LOCAL_HASH, 0xCC]) + pkt = Packet() + pkt.header = route | (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) + pkt.payload = bytearray(crc) + pkt.payload_len = len(pkt.payload) + pkt.path = bytearray(path) + pkt.path_len = len(path) + if transport_codes is not None: + pkt.transport_codes = list(transport_codes) + return pkt + + +class TestForwardRoutedAck: + """forward_routed_ack mirrors MeshCore: regenerate the ACK as a plain DIRECT + packet and relay it immediately, never through the generic delayed forward.""" + + def test_intermediate_ack_relays_with_zero_delay(self, handler): + pkt = _make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC])) + result = handler.forward_routed_ack(pkt) + assert result is not None + ack, delay = result + assert delay == pytest.approx(0.0) + assert ack.get_payload_type() == PAYLOAD_TYPE_ACK + assert ack.get_route_type() == ROUTE_TYPE_DIRECT + assert bytes(ack.payload) == b"\x4d\xab\xaf\x95" + # This node removed from the path. + assert list(ack.path) == [0xCC] + assert ack.path_len == PathUtils.encode_path_len(1, 1) + + def test_transport_direct_ack_regenerated_as_plain_direct(self, handler): + pkt = _make_direct_ack_packet( + route=ROUTE_TYPE_TRANSPORT_DIRECT, transport_codes=(0x1234, 0x5678) + ) + result = handler.forward_routed_ack(pkt) + assert result is not None + ack, delay = result + assert delay == pytest.approx(0.0) + # createAck rebuilds the header: plain DIRECT, transport codes dropped. + assert ack.get_route_type() == ROUTE_TYPE_DIRECT + assert not ack.has_transport_codes() + assert ack.transport_codes == [0, 0] + + def test_final_hop_ack_not_relayed(self, handler): + pkt = _make_direct_ack_packet(path=b"") + pkt.path_len = 0 + result = handler.forward_routed_ack(pkt) + assert result is None + assert "no path" in pkt.drop_reason + + def test_wrong_next_hop_ack_dropped(self, handler): + pkt = _make_direct_ack_packet(path=bytes([0xFE, 0xCC])) + result = handler.forward_routed_ack(pkt) + assert result is None + assert "not for us" in pkt.drop_reason + + def test_duplicate_ack_dropped(self, handler): + handler.forward_routed_ack(_make_direct_ack_packet()) + pkt2 = _make_direct_ack_packet() + result = handler.forward_routed_ack(pkt2) + assert result is None + assert pkt2.drop_reason == "Duplicate" + + def test_marked_do_not_retransmit_ack_dropped(self, handler): + pkt = _make_direct_ack_packet() + pkt.mark_do_not_retransmit() + result = handler.forward_routed_ack(pkt) + assert result is None + + def test_process_packet_dispatches_direct_ack(self, handler): + pkt = _make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC])) + result = handler.process_packet(pkt, snr=5.0) + assert result is not None + ack, delay = result + # The routed-ACK branch relays immediately; the generic direct forward + # would have applied a randomized direct TX delay instead. + assert delay == pytest.approx(0.0) + assert ack.get_route_type() == ROUTE_TYPE_DIRECT + + def test_process_packet_keeps_flood_ack_on_generic_path(self, handler): + pkt = _make_flood_packet( + payload=b"\x4d\xab\xaf\x95", path=b"\x11", payload_type=PAYLOAD_TYPE_ACK + ) + result = handler.process_packet(pkt, snr=5.0) + assert result is not None + fwd, _ = result + # Generic flood forwarding appends this node's hash. + assert fwd.get_route_type() == ROUTE_TYPE_FLOOD + assert list(fwd.path) == [0x11, LOCAL_HASH] + + +# Literal MeshCore-format routed-ACK frames (Packet::writeTo layout). MeshCore's +# routeDirectRecvAcks re-emits through createAck: the header is rebuilt from +# scratch (payload type ACK, route DIRECT, version bits cleared, transport codes +# dropped) and this node is removed from the path. Independent wire fixtures — +# do NOT rebuild them through Packet.write_to()/PathUtils. +FIRMWARE_ROUTED_ACK_VECTORS = ( + # header 0x0E = ACK<<2 | ROUTE_DIRECT; path_len 0x02 = 1-byte/2-hop; + # path AB CC (AB = us); CRC 95AFAB4D. out: path popped to CC. + pytest.param( + bytes([0xAB]), + bytes.fromhex("0E02ABCC4DABAF95"), + bytes.fromhex("0E01CC4DABAF95"), + id="direct-1-byte-2-hop", + ), + # 2-byte path width: path_len 0x42 = 2-byte/2-hop; path ABCD (us) + 1122. + pytest.param( + bytes([0xAB, 0xCD, 0xEF]), + bytes.fromhex("0E42ABCD11224DABAF95"), + bytes.fromhex("0E4111224DABAF95"), + id="direct-2-byte-2-hop", + ), + # header 0x0F = ACK<<2 | ROUTE_TRANSPORT_DIRECT; transport codes 34127856; + # the regenerated ACK is forced to plain DIRECT (codes dropped). + pytest.param( + bytes([0xAB]), + bytes.fromhex("0F3412785602ABCC4DABAF95"), + bytes.fromhex("0E01CC4DABAF95"), + id="transport-direct-1-byte-2-hop", + ), + # header 0x4E = version bit set on a direct ACK; createAck clears it. + pytest.param( + bytes([0xAB]), + bytes.fromhex("4E02ABCC4DABAF95"), + bytes.fromhex("0E01CC4DABAF95"), + id="direct-version-bits-cleared", + ), +) + + +@pytest.mark.parametrize("local_hash_bytes, wire_in, wire_out", FIRMWARE_ROUTED_ACK_VECTORS) +def test_forward_routed_ack_matches_firmware_wire_vectors(local_hash_bytes, wire_in, wire_out): + """Parse a firmware-format routed ACK and assert the exact relayed bytes.""" + handler = _make_handler_with_hash(local_hash_bytes) + pkt = Packet() + pkt.read_from(wire_in) + result = handler.forward_routed_ack(pkt) + assert result is not None + ack, delay = result + assert ack.write_to() == wire_out + assert delay == pytest.approx(0.0) + + +class TestMultiAckRedundancy: + """multi_acks=1 adds MeshCore's MULTIPART-wrapped redundancy copy ahead of + the plain ACK; both slide out by the direct retransmit delay + 300 ms.""" + + def test_default_off_produces_no_extras(self, handler): + result = handler.forward_routed_ack(_make_direct_ack_packet()) + assert result.extras == () + + def test_extra_precedes_plain_ack_at_same_delay(self, handler): + handler.multi_acks = 1 + pkt = _make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC])) + with patch.object(handler, "_calculate_tx_delay", return_value=0.25): + result = handler.forward_routed_ack(pkt) + assert result is not None + ack, delay = result + # delay = direct retransmit delay (0.25s) + 300ms, shared by both copies. + assert delay == pytest.approx(0.55) + assert len(result.extras) == 1 + wrapped, wrapped_delay = result.extras[0] + assert wrapped_delay == pytest.approx(0.55) + # MULTIPART wrapper: (remaining=1)<<4 | ACK prefix byte + the CRC, on the + # already self-removed path, plain DIRECT route. + assert wrapped.get_payload_type() == PAYLOAD_TYPE_MULTIPART + assert wrapped.get_route_type() == ROUTE_TYPE_DIRECT + assert bytes(wrapped.payload) == b"\x13\x4d\xab\xaf\x95" + assert bytes(wrapped.path) == bytes(ack.path) + assert wrapped.path_len == ack.path_len + # Exact wire bytes: header 0x2A = MULTIPART<<2 | DIRECT. + assert wrapped.write_to() == bytes.fromhex("2A01CC134DABAF95") + + def test_multipart_relay_also_gets_extra(self, handler): + handler.multi_acks = 1 + pkt = _make_multipart_ack_packet(remaining=2, path=bytes([LOCAL_HASH, 0xCC])) + with patch.object(handler, "_calculate_tx_delay", return_value=0.25): + result = handler.forward_multipart_direct(pkt) + assert result is not None + ack, delay = result + # base (remaining+1)*300ms = 0.9s, plus 0.25s + 300ms for the extra. + assert delay == pytest.approx(1.45) + assert len(result.extras) == 1 + _, wrapped_delay = result.extras[0] + assert wrapped_delay == pytest.approx(1.45) + + def test_multi_acks_clamped_like_firmware(self): + # MeshCore CommonCLI constrains multi_acks to 0..1 on prefs load. + handler = _make_handler_with_hash(bytes([LOCAL_HASH])) + handler.config.setdefault("repeater", {})["multi_acks"] = 5 + handler.reload_runtime_config() + assert handler.multi_acks == 1 + + handler.config["repeater"]["multi_acks"] = "bogus" + handler.reload_runtime_config() + assert handler.multi_acks == 0 + + handler.config["repeater"]["multi_acks"] = -3 + handler.reload_runtime_config() + assert handler.multi_acks == 0 + + def test_init_reads_repeater_multi_acks(self): + with ( + patch("repeater.engine.StorageCollector"), + patch("repeater.engine.RepeaterHandler._start_background_tasks"), + ): + from repeater.engine import RepeaterHandler + + handler = RepeaterHandler( + _make_config(repeater={"multi_acks": 1}), + _make_dispatcher(), + LOCAL_HASH, + local_hash_bytes=bytes([LOCAL_HASH]), + ) + assert handler.multi_acks == 1 + + @pytest.mark.asyncio + async def test_call_transmits_wrapped_copy_before_plain_ack(self, handler): + handler.multi_acks = 1 + handler.airtime_mgr.calculate_airtime = MagicMock(return_value=20.0) + handler.airtime_mgr.can_transmit = MagicMock(return_value=(True, 0.0)) + handler.airtime_mgr.record_tx = MagicMock() + handler.airtime_mgr.record_rx = MagicMock() + handler.dispatcher.send_packet = AsyncMock(return_value=True) + + pkt = _inject_from_wire(_make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC]))) + with ( + patch.object(handler, "_calculate_tx_delay", return_value=0.0), + patch("repeater.engine.asyncio.sleep", new_callable=AsyncMock), + ): + await handler(pkt, {"snr": 3.0, "rssi": -80}, local_transmission=False) + + assert handler.dispatcher.send_packet.call_count == 2 + first = handler.dispatcher.send_packet.call_args_list[0].args[0] + second = handler.dispatcher.send_packet.call_args_list[1].args[0] + # MeshCore queues the MULTIPART redundancy copy ahead of the plain ACK. + assert first.write_to() == bytes.fromhex("2A01CC134DABAF95") + assert second.write_to() == bytes.fromhex("0E01CC4DABAF95") + assert handler.forwarded_count == 2 + + # =================================================================== # 3. process_packet — route dispatch # =================================================================== @@ -2159,7 +2445,7 @@ class TestPacketInjectionRouting: assert sent_pkt.get_payload_type() == payload_type assert sent_pkt.path[-1] == LOCAL_HASH - @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES) + @pytest.mark.parametrize("payload_type", _GENERIC_DIRECT_FORWARD_PAYLOAD_TYPES) async def test_all_payload_types_direct_injection_forwards(self, handler, payload_type): self._prepare_fast_tx(handler) pkt = _inject_from_wire( @@ -2207,7 +2493,7 @@ class TestPacketInjectionRouting: assert sent_pkt.get_payload_type() == payload_type assert sent_pkt.transport_codes == [0x1111, 0x2222] - @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES) + @pytest.mark.parametrize("payload_type", _GENERIC_DIRECT_FORWARD_PAYLOAD_TYPES) async def test_all_payload_types_transport_direct_injection_forwards( self, handler, payload_type ):