From b2e45c2038a831b83ad25843dfcdc44465c5441c Mon Sep 17 00:00:00 2001 From: agessaman Date: Tue, 7 Jul 2026 11:17:54 -0700 Subject: [PATCH] fix: resolve room-server push ACKs through the dispatcher MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Room server pushes waited on dispatcher.wait_for_ack, but nothing in the repeater could ever resolve it, so every push timed out and re-pushed on the backoff schedule (issue #286's duplicate floods — worst for virtual companions on the same instance, whose ACKs never even cross the air): - no core AckHandler is registered (all RX lands in the router fallback), so received ACK CRCs were never fed to dispatcher ACK matching - inject_packet waited on packet.get_crc(), a packet-hash CRC that no ACK sender produces; the crypto CRC only the room server knows was ignored - PATH returns (how a flood-received DM is ACKed) were decrypted by PathHelper, but it read the encoded path_len wire byte as a raw count — an empty 3-byte-hash path (0x80) parsed as a 128-byte truncated payload — and the router's PATH branch never ran PathHelper for room/repeater destinations when any companion bridge existed Fixes: - router ACK branch feeds discrete ACK CRCs (RF and locally injected) to dispatcher._register_ack_received - PATH packets addressed to a local server identity are processed by PathHelper regardless of companion bridges; PathHelper decodes the encoded path_len via PathUtils, keeps the encoded byte in out_path_len, and registers the embedded ACK via ack_received_fn - inject_packet accepts expected_ack_crc/ack_timeout; the room server passes its crypto CRC and a hop-count-based timeout (the encoded byte would have produced a ~4-minute direct-push wait) Verified live on a real mesh: push to a same-instance virtual companion resolves via the PATH-embedded ACK (encoded path_len 0x80) and push to a firmware client resolves via the discrete RF ACK, no re-push loops. Fixes #286 Fixes #341 --- repeater/handler_helpers/path.py | 52 +++++++++++--- repeater/handler_helpers/room_server.py | 29 ++++++-- repeater/main.py | 3 + repeater/packet_router.py | 52 ++++++++++++-- ...test_handler_helpers_path_protocol_text.py | 68 +++++++++++++++++++ tests/test_handler_helpers_room_server.py | 16 ++++- tests/test_packet_router.py | 63 +++++++++++++++++ 7 files changed, 260 insertions(+), 23 deletions(-) diff --git a/repeater/handler_helpers/path.py b/repeater/handler_helpers/path.py index 98e8e1b..e36af51 100644 --- a/repeater/handler_helpers/path.py +++ b/repeater/handler_helpers/path.py @@ -5,10 +5,13 @@ logger = logging.getLogger("PathHelper") class PathHelper: - def __init__(self, acl_dict=None, log_fn=None): + def __init__(self, acl_dict=None, log_fn=None, ack_received_fn=None): self.acl_dict = acl_dict or {} self.log_fn = log_fn or logger.info + # Async callback fed with ACK CRCs found embedded in PATH payloads + # (dispatcher._register_ack_received) so local waiters resolve. + self.ack_received_fn = ack_received_fn async def process_path_packet(self, packet): @@ -60,30 +63,61 @@ class PathHelper: return False # Parse decrypted PATH data - # Format: path_len(1) + path[path_len] + extra_type(1) + extra[...] + # Format: path_len(1) + path[...] + extra_type(1) + extra[...] + # path_len is the ENCODED wire byte (bits 0-5 = hash count, bits + # 6-7 = hash size - 1), matching Packet.path_len — with 3-byte + # hashes an empty path is 0x80, not 0x00. Reading it as a raw byte + # count made every such path return look truncated. if len(decrypted) < 1: logger.debug("Decrypted PATH data too short") return False - path_len = decrypted[0] - if len(decrypted) < 1 + path_len: + from openhop_core.protocol.packet_utils import PathUtils + + path_len_byte = decrypted[0] + if not PathUtils.is_valid_path_len(path_len_byte): + logger.debug(f"Invalid encoded path_len 0x{path_len_byte:02X} in PATH data") + return False + path_byte_len = PathUtils.get_path_byte_len(path_len_byte) + if len(decrypted) < 1 + path_byte_len: logger.debug( - f"PATH data truncated: need {1 + path_len} bytes, got {len(decrypted)}" + f"PATH data truncated: need {1 + path_byte_len} bytes, got {len(decrypted)}" ) return False - path_data = decrypted[1 : 1 + path_len] + path_data = decrypted[1 : 1 + path_byte_len] - # Update client's out_path (same as C++ memcpy) + # Update client's out_path (same as C++ memcpy); out_path_len keeps + # the encoded byte so direct sends put it on the wire as-is. client.out_path = bytearray(path_data) - client.out_path_len = path_len + client.out_path_len = path_len_byte client.last_activity = int(time.time()) logger.info( f"Updated out_path for client 0x{src_hash:02X} -> 0x{dest_hash:02X}: " - f"path_len={path_len}, path={[hex(b) for b in path_data]}" + f"path_len=0x{path_len_byte:02X}, path={[hex(b) for b in path_data]}" ) + # Extra section after the path: extra_type(1) + extra[...]. Firmware + # answers a flood-received DM with a path return that embeds the + # delivery ACK here (createPathReturn, extra_type=PAYLOAD_TYPE_ACK); + # register it so local waiters (e.g. room server pushes) resolve. + from openhop_core.protocol.constants import PAYLOAD_TYPE_ACK + + extra_start = 1 + path_byte_len + if ( + self.ack_received_fn is not None + and len(decrypted) >= extra_start + 5 + and decrypted[extra_start] == PAYLOAD_TYPE_ACK + ): + ack_crc = int.from_bytes( + bytes(decrypted[extra_start + 1 : extra_start + 5]), "little" + ) + await self.ack_received_fn(ack_crc) + logger.info( + f"PATH from 0x{src_hash:02X} carried embedded ACK CRC={ack_crc:08X}" + ) + # Don't mark as do_not_retransmit - let it forward normally return False diff --git a/repeater/handler_helpers/room_server.py b/repeater/handler_helpers/room_server.py index 8140980..d8fbaa7 100644 --- a/repeater/handler_helpers/room_server.py +++ b/repeater/handler_helpers/room_server.py @@ -6,6 +6,7 @@ from typing import Dict from openhop_core.protocol import CryptoUtils, PacketBuilder from openhop_core.protocol.constants import PAYLOAD_TYPE_TXT_MSG +from openhop_core.protocol.packet_utils import PathUtils logger = logging.getLogger("RoomServer") @@ -377,18 +378,25 @@ class RoomServer: route_type=route_type, ) - # Add stored path for direct routing + # Add stored path for direct routing. out_path_len is the encoded + # wire byte (bits 0-5 = hash count, bits 6-7 = hash size - 1); + # out_path already holds exactly the path bytes. if route_type == "direct" and len(client_info.out_path) > 0: - packet.path = bytearray(client_info.out_path[: client_info.out_path_len]) + packet.path = bytearray(client_info.out_path) packet.path_len = client_info.out_path_len - # Calculate ACK timeout + # Calculate ACK timeout from the HOP count, not the encoded byte + # (0x80 = empty 3-byte-hash path would otherwise give a ~4min wait) if route_type == "flood": ack_timeout = PUSH_ACK_TIMEOUT_FLOOD_MS / 1000.0 else: - path_len = client_info.out_path_len if client_info.out_path_len >= 0 else 0 + hops = ( + PathUtils.get_path_hash_count(client_info.out_path_len) + if client_info.out_path_len >= 0 + else 0 + ) ack_timeout = ( - PUSH_TIMEOUT_BASE_MS + PUSH_ACK_TIMEOUT_FACTOR_MS * (path_len + 1) + PUSH_TIMEOUT_BASE_MS + PUSH_ACK_TIMEOUT_FACTOR_MS * (hops + 1) ) / 1000.0 # Update client sync state with pending ACK @@ -399,9 +407,16 @@ class RoomServer: push_post_timestamp=post["post_timestamp"], ack_timeout_time=time.time() + ack_timeout, ) - # Send packet (dispatcher will track ACK automatically) + # Send and wait for the client's delivery ACK. The injector must be + # told the crypto ACK CRC we computed above — its default + # (packet.get_crc()) is a packet-hash CRC no client ever sends. # This blocks for the entire transmission duration (0.5-9 seconds) - success = await self.packet_injector(packet, wait_for_ack=True) + success = await self.packet_injector( + packet, + wait_for_ack=True, + expected_ack_crc=expected_ack_crc, + ack_timeout=ack_timeout, + ) # SAFETY: Release transmission lock AFTER send completes self.global_limiter.release() diff --git a/repeater/main.py b/repeater/main.py index ab0a206..44b9239 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -412,6 +412,9 @@ class RepeaterDaemon: self.path_helper = PathHelper( acl_dict=self.login_helper.get_acl_dict(), # Per-identity ACLs log_fn=logger.info, + # Embedded PATH ACKs must reach the dispatcher's ACK matching or + # wait_for_ack() (e.g. room server pushes) never resolves. + ack_received_fn=self.dispatcher._register_ack_received, ) logger.info("PATH packet processing helper initialized") diff --git a/repeater/packet_router.py b/repeater/packet_router.py index 5cb5c67..862c99e 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -271,7 +271,14 @@ class PacketRouter: pass await self.queue.put(packet) - async def inject_packet(self, packet, wait_for_ack: bool = False, origin_hash=None): + async def inject_packet( + self, + packet, + wait_for_ack: bool = False, + origin_hash=None, + expected_ack_crc=None, + ack_timeout=None, + ): try: metadata = { "rssi": getattr(packet, "rssi", 0), @@ -327,8 +334,16 @@ class PacketRouter: dispatcher = getattr(self.daemon, "dispatcher", None) if dispatcher and hasattr(dispatcher, "wait_for_ack"): try: - expected_crc = packet.get_crc() - ack_ok = await dispatcher.wait_for_ack(expected_crc, timeout=5.0) + # Callers that know the crypto ACK CRC (e.g. room + # server pushes) must pass it: packet.get_crc() is a + # packet-hash CRC that no ACK sender ever produces. + expected_crc = ( + expected_ack_crc + if expected_ack_crc is not None + else packet.get_crc() + ) + timeout = ack_timeout if ack_timeout is not None else 5.0 + ack_ok = await dispatcher.wait_for_ack(expected_crc, timeout=timeout) if not ack_ok: logger.warning( "Injected packet ACK timeout (crc=%08X)", expected_crc @@ -463,6 +478,20 @@ class PacketRouter: self._record_for_ui(packet, metadata) elif payload_type == AckHandler.payload_type(): + # Feed the dispatcher's ACK matching: the repeater registers no core + # AckHandler (all RX lands in this router via the fallback), so + # without this no dispatcher.wait_for_ack() — e.g. a room server + # push — can ever resolve. Covers RF ACKs and locally injected ones + # (companions hosted on this same instance never go over the air). + if packet.payload is not None and len(packet.payload) >= 4: + register_ack = getattr( + getattr(self.daemon, "dispatcher", None), "_register_ack_received", None + ) + if register_ack is not None: + try: + await register_ack(int.from_bytes(bytes(packet.payload[:4]), "little")) + except Exception as e: + logger.debug("ACK registration failed: %s", e) # ACK has no dest in payload (4-byte CRC only); deliver to all bridges so sender sees send_confirmed. # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop. companion_bridges = self._companion_bridges_for_packet(packet, metadata) @@ -488,6 +517,19 @@ class PacketRouter: elif payload_type == PathHandler.payload_type(): dest_hash = packet.payload[0] if packet.payload else None companion_bridges = self._companion_bridges_for_packet(packet, metadata) + # PATH addressed to a local server identity (room server/repeater): + # process for client out_path updates and any embedded ACK before + # bridge delivery. Previously the all-bridges branch below swallowed + # these whenever any companion was configured, so room servers never + # saw path returns — or the delivery ACKs firmware embeds in them. + path_helper = getattr(self.daemon, "path_helper", None) + is_local_identity_dest = ( + path_helper is not None + and dest_hash is not None + and dest_hash in getattr(path_helper, "acl_dict", {}) + ) + if is_local_identity_dest: + await path_helper.process_path_packet(packet) if dest_hash is not None and dest_hash in companion_bridges: if self._should_deliver_path_to_companions(packet): await companion_bridges[dest_hash].process_received_packet(packet) @@ -506,8 +548,8 @@ class PacketRouter: len(companion_bridges), ) # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop. - elif self.daemon.path_helper: - await self.daemon.path_helper.process_path_packet(packet) + elif path_helper and not is_local_identity_dest: + await path_helper.process_path_packet(packet) elif payload_type == LoginResponseHandler.payload_type(): # PAYLOAD_TYPE_RESPONSE (0x01): payload is dest_hash(1)+src_hash(1)+encrypted. diff --git a/tests/test_handler_helpers_path_protocol_text.py b/tests/test_handler_helpers_path_protocol_text.py index b9cf90d..a4fd67a 100644 --- a/tests/test_handler_helpers_path_protocol_text.py +++ b/tests/test_handler_helpers_path_protocol_text.py @@ -67,6 +67,74 @@ async def test_path_helper_updates_client_out_path_on_valid_decrypt(): assert isinstance(client.last_activity, int) +@pytest.mark.asyncio +async def test_path_helper_registers_embedded_ack(): + """Firmware path returns embed the delivery ACK after the path + (extra_type=PAYLOAD_TYPE_ACK + 4-byte CRC); it must reach ack_received_fn + so local waiters (e.g. room server pushes) resolve.""" + client = _FakeClient(pubkey=bytes([0x22]) + b"x" * 31, shared_secret=b"k" * 32) + acl = _FakeACL([client]) + ack_fn = AsyncMock() + helper = PathHelper(acl_dict={0x11: acl}, ack_received_fn=ack_fn) + + packet = _PathPacket(payload=b"\x11\x22\xaa\xbb\xcc") + # path_len(2) + path + extra_type(PAYLOAD_TYPE_ACK=3) + crc(4, LE) + decrypted = b"\x02\x99\x88" + bytes([0x03]) + bytes.fromhex("4dabaf95") + with patch( + "openhop_core.protocol.crypto.CryptoUtils.mac_then_decrypt", + return_value=decrypted, + ): + await helper.process_path_packet(packet) + + ack_fn.assert_awaited_once_with(0x95AFAB4D) + assert client.out_path_len == 2 # path update still applied + + +@pytest.mark.asyncio +async def test_path_helper_handles_encoded_path_len_with_embedded_ack(): + """path_len in a path return is the ENCODED wire byte: with 3-byte hashes an + empty path is 0x80, not 0x00. Reading it as a raw count (128) made the + helper bail as 'truncated' before the embedded ACK was registered, so + room-server pushes to same-instance companions timed out forever. + Bytes below are a decrypted path return captured from a live mesh.""" + client = _FakeClient(pubkey=bytes([0x22]) + b"x" * 31, shared_secret=b"k" * 32) + acl = _FakeACL([client]) + ack_fn = AsyncMock() + helper = PathHelper(acl_dict={0x11: acl}, ack_received_fn=ack_fn) + + packet = _PathPacket(payload=b"\x11\x22\xaa\xbb\xcc") + # path_len 0x80 (3-byte hashes, 0 hops) + extra_type ACK + crc + AES padding + decrypted = bytes.fromhex("80038d48208500000000000000000000") + with patch( + "openhop_core.protocol.crypto.CryptoUtils.mac_then_decrypt", + return_value=decrypted, + ): + await helper.process_path_packet(packet) + + ack_fn.assert_awaited_once_with(0x8520488D) + assert client.out_path_len == 0x80 # encoded byte preserved + assert bytes(client.out_path) == b"" + + +@pytest.mark.asyncio +async def test_path_helper_ignores_non_ack_extra(): + client = _FakeClient(pubkey=bytes([0x22]) + b"x" * 31, shared_secret=b"k" * 32) + acl = _FakeACL([client]) + ack_fn = AsyncMock() + helper = PathHelper(acl_dict={0x11: acl}, ack_received_fn=ack_fn) + + packet = _PathPacket(payload=b"\x11\x22\xaa\xbb\xcc") + # extra_type 0x08 (PATH) instead of ACK: nothing to register + decrypted = b"\x02\x99\x88" + bytes([0x08]) + b"\x01\x02\x03\x04" + with patch( + "openhop_core.protocol.crypto.CryptoUtils.mac_then_decrypt", + return_value=decrypted, + ): + await helper.process_path_packet(packet) + + ack_fn.assert_not_awaited() + + @pytest.mark.asyncio async def test_path_helper_returns_false_for_non_matching_or_invalid_inputs(): client = _FakeClient(pubkey=bytes([0x22]) + b"x" * 31, shared_secret=b"k" * 32) diff --git a/tests/test_handler_helpers_room_server.py b/tests/test_handler_helpers_room_server.py index 5531056..a9fd806 100644 --- a/tests/test_handler_helpers_room_server.py +++ b/tests/test_handler_helpers_room_server.py @@ -146,7 +146,12 @@ async def test_room_server_push_post_to_client_success_direct_route_sets_path_an assert ok is True assert bytes(packet.path) == b"\xaa\xbb" assert packet.path_len == 2 - injector.assert_awaited_once_with(packet, wait_for_ack=True) + injector.assert_awaited_once_with( + packet, + wait_for_ack=True, + expected_ack_crc=int.from_bytes(b"\x01\x02\x03\x04", "little"), + ack_timeout=10.0, # PUSH_TIMEOUT_BASE + FACTOR * (path_len 2 + 1) + ) rs._handle_ack_received.assert_awaited_once_with( client.id.get_public_key(), post["post_timestamp"] ) @@ -174,9 +179,11 @@ async def test_room_server_push_expected_ack_matches_firmware_signed_ack(): db = _FakeDB() sent = [] + injector_kwargs = [] - async def injector(packet, wait_for_ack=False): + async def injector(packet, wait_for_ack=False, **kwargs): sent.append(packet) + injector_kwargs.append(kwargs) return False # no ACK: leaves the pending upsert as the only db write rs = RoomServer( @@ -205,6 +212,11 @@ async def test_room_server_push_expected_ack_matches_firmware_signed_ack(): assert len(upserts) == 1 expected_ack_crc = upserts[0]["pending_ack_crc"] + # The injector must be told the crypto CRC (and the computed timeout) so + # dispatcher.wait_for_ack matches the client's actual ACK. + assert injector_kwargs[0]["expected_ack_crc"] == expected_ack_crc + assert injector_kwargs[0]["ack_timeout"] > 0 + # Decrypt the pushed datagram and verify the signed-plain layout. pkt = sent[0] encrypted = bytes(pkt.payload[2 : pkt.payload_len]) diff --git a/tests/test_packet_router.py b/tests/test_packet_router.py index 5872d9a..671b1fc 100644 --- a/tests/test_packet_router.py +++ b/tests/test_packet_router.py @@ -56,6 +56,9 @@ def _make_daemon(): daemon.text_helper = None daemon.path_helper = None daemon.protocol_request_helper = None + daemon.dispatcher = MagicMock() + daemon.dispatcher._register_ack_received = AsyncMock() + daemon.dispatcher.wait_for_ack = AsyncMock(return_value=True) return daemon @@ -614,6 +617,66 @@ class TestPacketRouterRoutingBranches(unittest.IsolatedAsyncioTestCase): b2.process_received_packet.assert_awaited_once() daemon.repeater_handler.assert_awaited_once() + async def test_route_ack_registers_crc_with_dispatcher(self): + """Discrete ACKs must feed dispatcher ACK matching or wait_for_ack never resolves.""" + daemon = _make_daemon() + router = PacketRouter(daemon) + pkt = _make_packet(AckHandler.payload_type()) + # 6-byte firmware ACK: 4-byte CRC (LE) + ext-attempt + random byte + pkt.payload = bytes.fromhex("4dabaf95") + b"\x00\x7f" + await router._route_packet(pkt) + daemon.dispatcher._register_ack_received.assert_awaited_once_with(0x95AFAB4D) + + async def test_route_ack_short_payload_not_registered(self): + daemon = _make_daemon() + router = PacketRouter(daemon) + pkt = _make_packet(AckHandler.payload_type()) + pkt.payload = b"\x01\x02" + await router._route_packet(pkt) + daemon.dispatcher._register_ack_received.assert_not_awaited() + + async def test_route_locally_injected_ack_registers_crc(self): + """ACKs from same-instance companions never cross the radio: the injected + packet re-entering the router is the only chance to resolve local waiters + (e.g. a room server pushing to a virtual companion on this repeater).""" + daemon = _make_daemon() + router = PacketRouter(daemon) + pkt = _make_packet(AckHandler.payload_type()) + pkt.payload = b"\xaa\xbb\xcc\xdd" + pkt._injected_for_tx = True + await router._route_packet(pkt) + daemon.dispatcher._register_ack_received.assert_awaited_once_with(0xDDCCBBAA) + daemon.repeater_handler.assert_not_awaited() # already transmitted + + async def test_inject_packet_waits_on_expected_ack_crc(self): + """Callers that know the crypto ACK CRC pass it; packet.get_crc() is a + packet-hash CRC no ACK sender ever produces.""" + daemon = _make_daemon() + router = PacketRouter(daemon) + pkt = _make_packet(TextMessageHandler.payload_type()) + ok = await router.inject_packet( + pkt, wait_for_ack=True, expected_ack_crc=0x1234ABCD, ack_timeout=12.0 + ) + self.assertTrue(ok) + daemon.dispatcher.wait_for_ack.assert_awaited_once_with(0x1234ABCD, timeout=12.0) + pkt.get_crc.assert_not_called() + + async def test_route_path_to_local_identity_runs_path_helper_despite_bridges(self): + """PATH addressed to a room server/repeater identity must reach the path + helper (out_path update + embedded ACK) even when companion bridges exist; + previously the all-bridges branch swallowed it.""" + daemon = _make_daemon() + bridge = _make_bridge() + daemon.companion_bridges = {0x01: bridge} + daemon.path_helper = MagicMock() + daemon.path_helper.acl_dict = {0x48: MagicMock()} + daemon.path_helper.process_path_packet = AsyncMock(return_value=False) + router = PacketRouter(daemon) + pkt = _make_packet(PathHandler.payload_type()) + pkt.payload = bytes([0x48, 0x77, 0xAA, 0xBB]) + await router._route_packet(pkt) + daemon.path_helper.process_path_packet.assert_awaited_once_with(pkt) + async def test_route_path_dedupes_companion_delivery(self): daemon = _make_daemon() bridge = _make_bridge()