From 2d5353ad7d011d9ef1f6125e30d46077875ce6e3 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 16 May 2026 17:32:15 -0700 Subject: [PATCH 1/5] fix(companion): improve transmission handling in RepeaterHandler - Updated the __call__ method to return a boolean indicating transmission success. - Adjusted handling of transmission failures to increment dropped_count appropriately. - Enhanced logging for transmission failures in PacketRouter to ensure visibility of issues. - Added tests to verify correct behavior during transmission exceptions and failures. --- repeater/engine.py | 45 +++++++++++++++++++++++++++---------- repeater/main.py | 2 +- repeater/packet_router.py | 36 +++++++++++++++++++++++++++-- tests/test_engine.py | 40 +++++++++++++++++++++++++++++++++ tests/test_packet_router.py | 15 ++++++++++++- tests/test_tx_lock.py | 3 ++- 6 files changed, 124 insertions(+), 17 deletions(-) diff --git a/repeater/engine.py b/repeater/engine.py index 6bf0aaf..f6d1fd2 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -159,7 +159,7 @@ class RepeaterHandler(BaseHandler): async def __call__( self, packet: Packet, metadata: Optional[dict] = None, local_transmission: bool = False - ) -> None: + ) -> bool: if metadata is None: metadata = {} @@ -256,19 +256,23 @@ class RepeaterHandler(BaseHandler): f"Duty-cycle limit: deferring local TX by {wait_time:.1f}s " f"(airtime={airtime_ms:.1f}ms)" ) - self.forwarded_count += 1 - transmitted = True tx_task = await self.schedule_retransmit( fwd_pkt, deferred_delay, airtime_ms, local_transmission=True ) try: - await tx_task + tx_success = await tx_task except Exception as e: - self.forwarded_count -= 1 transmitted = False drop_reason = "TX failed (deferred)" logger.warning(f"Deferred local TX failed: {e}") raise + if not tx_success: + transmitted = False + drop_reason = "TX failed (deferred)" + self.dropped_count += 1 + else: + self.forwarded_count += 1 + transmitted = True tx_metadata = getattr(fwd_pkt, "_tx_metadata", None) if tx_metadata: lbt_attempts = tx_metadata.get("lbt_attempts", 0) @@ -291,19 +295,23 @@ class RepeaterHandler(BaseHandler): self.dropped_count += 1 drop_reason = "Duty cycle limit" else: - self.forwarded_count += 1 - transmitted = True tx_task = await self.schedule_retransmit( fwd_pkt, delay, airtime_ms, local_transmission=local_transmission ) try: - await tx_task + tx_success = await tx_task except Exception as e: - self.forwarded_count -= 1 transmitted = False drop_reason = "TX failed" logger.warning(f"Local TX failed: {e}") raise + if not tx_success: + transmitted = False + drop_reason = "TX failed" + self.dropped_count += 1 + else: + self.forwarded_count += 1 + transmitted = True tx_metadata = getattr(fwd_pkt, "_tx_metadata", None) if tx_metadata: lbt_attempts = tx_metadata.get("lbt_attempts", 0) @@ -414,6 +422,8 @@ class RepeaterHandler(BaseHandler): # Not a duplicate or first occurrence self._append_recent_packet(packet_record) + return transmitted + def log_trace_record(self, packet_record: dict) -> None: """Manually log a packet trace record (used by external callers)""" self._append_recent_packet(packet_record) @@ -1116,10 +1126,20 @@ class RepeaterHandler(BaseHandler): "Packet dropped at TX time: duty-cycle exceeded " "(airtime=%.1fms)", airtime_ms, ) - return + return False try: - await self.dispatcher.send_packet(fwd_pkt, wait_for_ack=False) + sent = await self.dispatcher.send_packet( + fwd_pkt, wait_for_ack=False + ) + if not sent: + logger.warning( + "Retransmit failed (attempt %d): dispatcher returned false", + attempt + 1, + ) + if local_transmission and attempt == 0: + continue + return False self._record_packet_sent(fwd_pkt) if airtime_ms > 0: self.airtime_mgr.record_tx(airtime_ms) @@ -1128,13 +1148,14 @@ class RepeaterHandler(BaseHandler): f"Retransmitted packet ({packet_size} bytes, " f"{airtime_ms:.1f}ms airtime)" ) - return + return True except Exception as e: logger.error(f"Retransmit failed (attempt {attempt + 1}): {e}") if local_transmission and attempt == 0: pass # release lock, outer loop sleeps, then retries else: raise + return False return asyncio.create_task(delayed_send()) diff --git a/repeater/main.py b/repeater/main.py index 7087738..89b41a9 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -1049,7 +1049,7 @@ class RepeaterDaemon: logger.debug("Marked own advert as seen in duplicate cache") logger.info( - "Sent flood advert '%s' at (% .6f, % .6f) source=%s", + "Sent flood advert '%s' at (%.6f, %.6f) source=%s", node_name, latitude, longitude, diff --git a/repeater/packet_router.py b/repeater/packet_router.py index 0b5024e..fd8f831 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -173,9 +173,12 @@ class PacketRouter: # (avoids duty-cycle or dispatcher races where a later packet goes out first) async with self._inject_lock: # Use local_transmission=True to bypass forwarding logic - await self.daemon.repeater_handler( + sent = await self.daemon.repeater_handler( packet, metadata, local_transmission=True ) + if not sent: + logger.warning("Injected packet failed local transmission") + return False # Mark so when this packet is dequeued we don't pass to engine again (avoid double-send / double-count) packet._injected_for_tx = True @@ -183,6 +186,28 @@ class PacketRouter: # Enqueue so router can deliver to companion(s): TXT_MSG -> dest bridge, ACK -> all bridges (sender sees ACK) await self.enqueue(packet) + if wait_for_ack: + ptype = getattr(packet, "get_payload_type", lambda: None)() + if ptype not in { + AckHandler.payload_type(), + AdvertHandler.payload_type(), + }: + 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 + ) + if not ack_ok: + logger.warning( + "Injected packet ACK timeout (crc=%08X)", expected_crc + ) + return False + except Exception as e: + logger.warning("Injected packet ACK wait failed: %s", e) + return False + packet_len = len(packet.payload) if packet.payload else 0 logger.debug( f"Injected packet processed by engine as local transmission ({packet_len} bytes)" @@ -458,4 +483,11 @@ class PacketRouter: "snr": getattr(packet, "snr", 0.0), "timestamp": getattr(packet, "timestamp", 0), } - await self.daemon.repeater_handler(packet, metadata) + sent = await self.daemon.repeater_handler(packet, metadata) + if sent is False: + logger.warning( + "Inbound packet not transmitted by repeater handler " + "(type=%s, header=0x%02x)", + payload_type, + getattr(packet, "header", 0), + ) diff --git a/tests/test_engine.py b/tests/test_engine.py index bb50883..1488a90 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -1044,6 +1044,46 @@ class TestTxMode: await asyncio.sleep(0) handler.dispatcher.send_packet.assert_called_once() + async def test_local_tx_exception_does_not_underflow_forwarded_count(self, handler): + """TX exceptions must not decrement forwarded_count before any increment happened.""" + handler.config["repeater"]["mode"] = "forward" + pkt = _make_flood_packet() + handler.forwarded_count = 0 + + async def _boom(): + raise RuntimeError("simulated tx failure") + + with patch("repeater.engine.asyncio.sleep", new_callable=AsyncMock): + with patch.object( + handler, "schedule_retransmit", new=AsyncMock(return_value=asyncio.create_task(_boom())) + ): + with pytest.raises(RuntimeError, match="simulated tx failure"): + await handler(pkt, {"snr": 0.0, "rssi": -80}, local_transmission=True) + + assert handler.forwarded_count == 0 + + async def test_rx_tx_failure_increments_dropped_count_and_preserves_accounting(self, handler): + """A graceful TX failure should count as dropped and keep counters balanced.""" + handler.config["repeater"]["mode"] = "forward" + pkt = _make_flood_packet() + handler.rx_count = 0 + handler.forwarded_count = 0 + handler.dropped_count = 0 + + async def _tx_false(): + return False + + with patch.object( + handler, "schedule_retransmit", new=AsyncMock(return_value=asyncio.create_task(_tx_false())) + ): + transmitted = await handler(pkt, {"snr": 0.0, "rssi": -80}, local_transmission=False) + + assert transmitted is False + assert handler.rx_count == 1 + assert handler.forwarded_count == 0 + assert handler.dropped_count == 1 + assert handler.rx_count == handler.forwarded_count + handler.dropped_count + # =================================================================== # 16. Airtime calculation correctness diff --git a/tests/test_packet_router.py b/tests/test_packet_router.py index 2b46514..9c91565 100644 --- a/tests/test_packet_router.py +++ b/tests/test_packet_router.py @@ -31,7 +31,7 @@ from repeater.packet_router import PacketRouter def _make_daemon(): """Minimal daemon that satisfies PacketRouter without touching hardware.""" daemon = MagicMock() - daemon.repeater_handler = AsyncMock(return_value=None) + daemon.repeater_handler = AsyncMock(return_value=True) daemon.trace_helper = None daemon.discovery_helper = None daemon.advert_helper = None @@ -180,6 +180,19 @@ class TestInFlightCap(unittest.IsolatedAsyncioTestCase): finally: await router.stop() + async def test_non_injected_handler_false_is_logged(self): + """Inbound packets should log when repeater_handler reports TX failure.""" + daemon = _make_daemon() + daemon.repeater_handler = AsyncMock(return_value=False) + router = PacketRouter(daemon) + pkt = _make_packet(payload_type=0xFF) + + with patch("repeater.packet_router.logger.warning") as mock_warn: + await router._route_packet(pkt) + + daemon.repeater_handler.assert_awaited_once() + mock_warn.assert_called() + # ── 3. Shutdown: in-flight tasks drained ──────────────────────────────── async def test_stop_waits_for_in_flight_tasks(self): diff --git a/tests/test_tx_lock.py b/tests/test_tx_lock.py index 0ca9193..2c96435 100644 --- a/tests/test_tx_lock.py +++ b/tests/test_tx_lock.py @@ -257,7 +257,8 @@ class TestTxLockSerialisation(unittest.IsolatedAsyncioTestCase): task = await h.schedule_retransmit( pkt, delay=0.0, airtime_ms=100.0, local_transmission=True ) - await task # should complete without error (gate returns silently) + result = await task + self.assertFalse(result, "Duty-cycle drop should report TX failure") self.assertEqual(send_calls[0], 1, "send_packet called on retry despite duty-cycle rejection") From 5fcb6255d594ad0c19db64a7ba65840504907067 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 30 May 2026 16:19:52 -0700 Subject: [PATCH 2/5] feat: enhance login handler with anonymous request support and region name formatting - Updated the `LoginHelper` class to wrap the login handler in an `AnonRequestHandler`, allowing for proper handling of anonymous requests. - Introduced methods for formatting region names based on flood policies and added support for retrieving transport keys from SQLite storage. - Enhanced the constructor to accept additional parameters for SQLite handler and configuration, improving flexibility for owner-info and feature-flag replies. - Added tests to validate the new functionality and ensure correct behavior of region name formatting and owner/features callbacks. --- repeater/handler_helpers/login.py | 107 +++++++++++++++++- repeater/main.py | 6 + ...t_handler_helpers_trace_discovery_login.py | 62 +++++++++- 3 files changed, 168 insertions(+), 7 deletions(-) diff --git a/repeater/handler_helpers/login.py b/repeater/handler_helpers/login.py index 6c05a4e..4db7c98 100644 --- a/repeater/handler_helpers/login.py +++ b/repeater/handler_helpers/login.py @@ -6,7 +6,9 @@ This module processes login requests and manages authentication for all identiti import asyncio import logging +import time +from pymc_core.node.handlers.anon_request import AnonRateLimiter, AnonRequestHandler from pymc_core.node.handlers.login_server import LoginServerHandler from pymc_core.protocol.constants import PAYLOAD_TYPE_ANON_REQ @@ -14,15 +16,27 @@ logger = logging.getLogger("LoginHelper") class LoginHelper: - def __init__(self, identity_manager, packet_injector=None, log_fn=None): + def __init__( + self, + identity_manager, + packet_injector=None, + log_fn=None, + sqlite_handler=None, + config=None, + ): self.identity_manager = identity_manager self.packet_injector = packet_injector self.log_fn = log_fn or logger.info + self.sqlite_handler = sqlite_handler + self.config = config or {} self.handlers = {} self.acls = {} # Per-identity ACLs keyed by hash_byte self._pending_tasks = set() + # Shared across all identities so the node's total anon-reply rate is + # bounded (mirrors firmware anon_limiter: ~4 requests / 2 min). + self.anon_limiter = AnonRateLimiter() def _track_task(self, task: asyncio.Task) -> None: self._pending_tasks.add(task) @@ -126,12 +140,88 @@ class LoginHelper: is_room_server=(identity_type == "room_server"), ) - handler.set_send_packet_callback(self._send_packet_with_delay) + # Wrap the login handler in an anon-request dispatcher so anonymous + # regions/owner/basic discovery queries are answered instead of being + # mis-parsed as failed logins (MeshCore 1.16.0 discovery feature). + anon_handler = AnonRequestHandler( + local_identity=identity, + log_fn=self.log_fn, + login_handler=handler, + anon_limiter=self.anon_limiter, + region_names_fn=self._format_region_names, + owner_info_fn=self._make_owner_info_fn(name, config), + features_fn=self._make_features_fn(config), + clock_fn=lambda: int(time.time()), + ) + # Wires the send callback through to both the wrapper and login handler. + anon_handler.set_send_packet_callback(self._send_packet_with_delay) - self.handlers[hash_byte] = handler + self.handlers[hash_byte] = anon_handler logger.info(f"Registered {identity_type} '{name}' login handler: hash=0x{hash_byte:02X}") + def _format_region_names(self) -> str: + """Build the comma-separated region-names string for an anon regions reply. + + Mirrors firmware ``RegionMap::exportNamesTo`` with ``REGION_DENY_FLOOD``: + emit the ``*`` wildcard region first (unless unscoped flood is denied), + then each allow-flood named region with a leading ``#`` stripped, with no + trailing comma. The firmware wildcard is the always-present default flood + scope; pyMC_repeater models that via ``mesh.unscoped_flood_allow`` + (falling back to ``mesh.global_flood_allow``, default allow). + """ + parts = [] + + mesh_cfg = self.config.get("mesh", {}) if isinstance(self.config, dict) else {} + unscoped_allow = mesh_cfg.get( + "unscoped_flood_allow", mesh_cfg.get("global_flood_allow", True) + ) + if unscoped_allow: + parts.append("*") + + if self.sqlite_handler: + try: + keys = self.sqlite_handler.get_transport_keys() + except Exception as e: + logger.warning(f"Failed to read transport keys for regions reply: {e}") + keys = [] + for rec in keys or []: + if rec.get("flood_policy", "deny") != "allow": + continue + name = (rec.get("name") or "").strip() + if not name or name == "*": + continue # wildcard handled above + parts.append(name[1:] if name.startswith("#") else name) + + return ",".join(parts) + + @staticmethod + def _make_owner_info_fn(name: str, config: dict): + """Build an owner-info callback returning ``(node_name, owner_info)``.""" + + def owner_info_fn(): + cfg = config or {} + repeater_cfg = cfg.get("repeater", {}) + node_name = repeater_cfg.get("node_name") or name or "pyMC" + owner = repeater_cfg.get("owner_info", "") or "" + return (node_name, owner) + + return owner_info_fn + + @staticmethod + def _make_features_fn(config: dict): + """Build a feature-flags callback (bit0 = bridge, bit7 = forwarding disabled).""" + + def features_fn(): + cfg = config or {} + mode = cfg.get("repeater", {}).get("mode", "forward") + features = 0 + if mode != "forward": # monitor / no_tx => not forwarding + features |= 0x80 + return features + + return features_fn + async def process_login_packet(self, packet): try: @@ -147,9 +237,16 @@ class LoginHelper: packet.mark_do_not_retransmit() return True else: - # ANON_REQ to other nodes (e.g. owner-info to firmware) is normal; skip log to avoid spam + # ANON_REQ to other nodes (e.g. another repeater's regions/owner + # query overheard on-air) is normal; log at DEBUG so the dest is + # visible when diagnosing "why didn't my repeater answer". ptype = getattr(packet, "get_payload_type", lambda: None)() - if ptype != PAYLOAD_TYPE_ANON_REQ: + if ptype == PAYLOAD_TYPE_ANON_REQ: + logger.debug( + f"ANON_REQ for hash 0x{dest_hash:02X} not addressed to a local " + f"identity ({sorted(f'0x{h:02X}' for h in self.handlers)}); ignoring" + ) + else: logger.debug( f"No login handler registered for hash 0x{dest_hash:02X}, allowing forward" ) diff --git a/repeater/main.py b/repeater/main.py index 8646a1b..8646919 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -268,6 +268,12 @@ class RepeaterDaemon: identity_manager=self.identity_manager, packet_injector=self.router.inject_packet, log_fn=logger.info, + sqlite_handler=( + self.repeater_handler.storage.sqlite_handler + if self.repeater_handler and self.repeater_handler.storage + else None + ), # For anon regions-discovery replies + config=self.config, # For owner-info / feature-flags replies ) # Register default repeater identity diff --git a/tests/test_handler_helpers_trace_discovery_login.py b/tests/test_handler_helpers_trace_discovery_login.py index 593976e..54d21fe 100644 --- a/tests/test_handler_helpers_trace_discovery_login.py +++ b/tests/test_handler_helpers_trace_discovery_login.py @@ -251,12 +251,16 @@ def test_login_register_identity_repeater_creates_acl_and_handler(): identity = FakeIdentity(0x52) acl_obj = MagicMock() handler_obj = MagicMock() + anon_obj = MagicMock() with ( patch("repeater.handler_helpers.acl.ACL", return_value=acl_obj) as acl_cls, patch( "repeater.handler_helpers.login.LoginServerHandler", return_value=handler_obj ) as handler_cls, + patch( + "repeater.handler_helpers.login.AnonRequestHandler", return_value=anon_obj + ) as anon_cls, ): helper.register_identity( name="repeater-main", @@ -271,11 +275,65 @@ def test_login_register_identity_repeater_creates_acl_and_handler(): acl_cls.assert_called_once() handler_cls.assert_called_once() - handler_obj.set_send_packet_callback.assert_called_once() - assert helper.handlers[0x52] is handler_obj + # The login handler is wrapped in an AnonRequestHandler, and that wrapper is + # what gets stored + wired with the send callback. + anon_cls.assert_called_once() + assert anon_cls.call_args.kwargs["login_handler"] is handler_obj + anon_obj.set_send_packet_callback.assert_called_once() + assert helper.handlers[0x52] is anon_obj assert helper.acls[0x52] is acl_obj +class _FakeSqlite: + def __init__(self, keys): + self._keys = keys + + def get_transport_keys(self): + return self._keys + + +def test_format_region_names_filters_and_strips(): + keys = [ + {"name": "#VHF", "flood_policy": "allow"}, + {"name": "USA", "flood_policy": "allow"}, + {"name": "secret", "flood_policy": "deny"}, + {"name": "*", "flood_policy": "allow"}, # duplicate wildcard ignored + {"name": "", "flood_policy": "allow"}, + ] + # Default config => unscoped flood allowed => wildcard '*' present. + helper = LoginHelper(identity_manager=MagicMock(), sqlite_handler=_FakeSqlite(keys)) + # Wildcard first (from policy), '#' stripped, deny + empty + literal '*' excluded. + assert helper._format_region_names() == "*,VHF,USA" + + +def test_format_region_names_wildcard_suppressed_when_unscoped_denied(): + keys = [{"name": "USA", "flood_policy": "allow"}] + helper = LoginHelper( + identity_manager=MagicMock(), + sqlite_handler=_FakeSqlite(keys), + config={"mesh": {"unscoped_flood_allow": False}}, + ) + # No wildcard when unscoped flood is denied (firmware: wildcard deny-flood). + assert helper._format_region_names() == "USA" + + +def test_format_region_names_without_storage_is_just_wildcard(): + # No named regions, but unscoped flood allowed by default => bare wildcard. + helper = LoginHelper(identity_manager=MagicMock(), sqlite_handler=None) + assert helper._format_region_names() == "*" + + +def test_owner_and_features_callbacks_from_config(): + config = {"repeater": {"node_name": "node-x", "owner_info": "me", "mode": "monitor"}} + helper = LoginHelper(identity_manager=MagicMock(), config=config) + + assert helper._make_owner_info_fn("fallback", config)() == ("node-x", "me") + # Non-forward mode sets the forwarding-disabled bit (0x80). + assert helper._make_features_fn(config)() == 0x80 + # Forwarding mode clears it. + assert helper._make_features_fn({"repeater": {"mode": "forward"}})() == 0x00 + + @pytest.mark.asyncio async def test_login_process_packet_routes_to_registered_handler_and_marks_no_retransmit(): helper = LoginHelper(identity_manager=MagicMock(), packet_injector=AsyncMock()) From 778adb69177f78501959a0bb16ca298e14743627 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 30 May 2026 18:07:23 -0700 Subject: [PATCH 3/5] feat: implement randomized response jitter in DiscoveryHelper to prevent packet collisions - Added a default upper bound for randomized pre-send jitter in discovery responses to avoid collisions when multiple repeaters respond simultaneously. - Introduced a new parameter `response_jitter_ms` in the `DiscoveryHelper` constructor to configure the jitter. - Updated the `_send_packet_async` method to apply the jitter before sending responses. - Added tests to verify the correct application of jitter and ensure functionality when jitter is disabled. --- repeater/handler_helpers/discovery.py | 27 +++++++++++++ ...t_handler_helpers_trace_discovery_login.py | 40 ++++++++++++++++++- 2 files changed, 66 insertions(+), 1 deletion(-) diff --git a/repeater/handler_helpers/discovery.py b/repeater/handler_helpers/discovery.py index 8153747..fab9d76 100644 --- a/repeater/handler_helpers/discovery.py +++ b/repeater/handler_helpers/discovery.py @@ -7,11 +7,21 @@ allowing other nodes to discover repeaters on the mesh network. import asyncio import logging +import secrets from pymc_core.node.handlers.control import ControlHandler logger = logging.getLogger("DiscoveryHelper") +# Default upper bound (ms) for the randomized pre-send jitter applied to node +# discovery responses. A node-discover request is a broadcast that every +# in-range repeater answers at once, so without jitter they all transmit at the +# same engine-scheduled instant and collide. Mirrors the firmware, which spreads +# these replies deliberately (MyMesh.cpp:797, sendZeroHop with +# getRetransmitDelay*4). Safe to be generous: the requester's discovery window is +# 60s (firmware pending_discover_until = futureMillis(60000)). +DEFAULT_DISCOVERY_RESPONSE_JITTER_MS = 2000 + class DiscoveryHelper: """Helper class for processing discovery requests in the repeater.""" @@ -23,6 +33,7 @@ class DiscoveryHelper: node_type: int = 2, log_fn=None, debug_log_fn=None, + response_jitter_ms: int = DEFAULT_DISCOVERY_RESPONSE_JITTER_MS, ): """ Initialize the discovery helper. @@ -34,10 +45,14 @@ class DiscoveryHelper: log_fn: Optional logging function for ControlHandler debug_log_fn: Optional logging for verbose ControlHandler messages (e.g. callback presence). Pass logger.debug to avoid INFO noise when forwarding to companions. + response_jitter_ms: Upper bound (ms) for the randomized delay added before + transmitting a discovery response, to avoid multiple repeaters colliding + when answering the same broadcast. Set to 0 to disable (e.g. in tests). """ self.local_identity = local_identity self.packet_injector = packet_injector # Function to inject packets into router self.node_type = node_type + self.response_jitter_ms = max(0, int(response_jitter_ms)) # Create ControlHandler internally as a parsing utility self.control_handler = ControlHandler( @@ -147,6 +162,18 @@ class DiscoveryHelper: tag: The tag for logging purposes """ try: + # Randomized pre-send jitter so multiple repeaters answering the same + # zero-hop discovery broadcast don't transmit at the same engine-scheduled + # instant and collide (the engine's DIRECT delay is fixed, not random). + # Mirrors firmware MyMesh.cpp:797. Uses secrets like the engine's TX jitter. + if self.response_jitter_ms > 0: + jitter_s = secrets.randbelow(self.response_jitter_ms + 1) / 1000.0 + if jitter_s > 0: + logger.debug( + f"Discovery response jitter {jitter_s * 1000:.0f}ms for tag 0x{tag:08X}" + ) + await asyncio.sleep(jitter_s) + success = await self.packet_injector(packet, wait_for_ack=False) if success: logger.info(f"Response sent for tag 0x{tag:08X}") diff --git a/tests/test_handler_helpers_trace_discovery_login.py b/tests/test_handler_helpers_trace_discovery_login.py index 54d21fe..d286d7d 100644 --- a/tests/test_handler_helpers_trace_discovery_login.py +++ b/tests/test_handler_helpers_trace_discovery_login.py @@ -207,7 +207,10 @@ def test_discovery_request_without_identity_does_not_send(): @pytest.mark.asyncio async def test_discovery_send_packet_async_success_failure_and_exception(): injector = AsyncMock(side_effect=[True, False, RuntimeError("send fail")]) - helper = DiscoveryHelper(local_identity=FakeIdentity(0x42), packet_injector=injector) + # jitter disabled so the test doesn't sleep + helper = DiscoveryHelper( + local_identity=FakeIdentity(0x42), packet_injector=injector, response_jitter_ms=0 + ) await helper._send_packet_async(packet=object(), tag=0x11) await helper._send_packet_async(packet=object(), tag=0x12) @@ -216,6 +219,41 @@ async def test_discovery_send_packet_async_success_failure_and_exception(): assert injector.await_count == 3 +@pytest.mark.asyncio +async def test_discovery_response_applies_bounded_jitter_before_send(): + injector = AsyncMock(return_value=True) + helper = DiscoveryHelper( + local_identity=FakeIdentity(0x42), packet_injector=injector, response_jitter_ms=2000 + ) + + slept = [] + + async def fake_sleep(secs): + slept.append(secs) + + with patch("repeater.handler_helpers.discovery.asyncio.sleep", side_effect=fake_sleep): + await helper._send_packet_async(packet=object(), tag=0x55) + + # Jitter applied exactly once, bounded to [0, 2.0]s, before the injection. + assert len(slept) == 1 + assert 0.0 <= slept[0] <= 2.0 + injector.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_discovery_response_jitter_disabled_does_not_sleep(): + injector = AsyncMock(return_value=True) + helper = DiscoveryHelper( + local_identity=FakeIdentity(0x42), packet_injector=injector, response_jitter_ms=0 + ) + + with patch("repeater.handler_helpers.discovery.asyncio.sleep") as sleep_mock: + await helper._send_packet_async(packet=object(), tag=0x56) + + sleep_mock.assert_not_called() + injector.assert_awaited_once() + + def test_discovery_send_response_without_injector_is_safe(): helper = DiscoveryHelper(local_identity=FakeIdentity(0x42), packet_injector=None) From ee92f5b1a99158ada53a0a93dec73318df0ceaa9 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 30 May 2026 21:43:00 -0700 Subject: [PATCH 4/5] test: expect True from deferred local TX mock after companion send fix The merged RepeaterHandler treats a falsy tx_success as failure; update the duty-cycle deferral test to return True instead of None. Co-authored-by: Cursor --- tests/test_engine.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_engine.py b/tests/test_engine.py index 7479a4f..c6623fa 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -1833,7 +1833,7 @@ class TestEngineTransmissionAndBackgroundLifecycle: with patch.object(handler, "_calculate_tx_delay", return_value=0.5): loop = asyncio.get_running_loop() completed = loop.create_future() - completed.set_result(None) + completed.set_result(True) async def _fake_schedule(packet, delay, airtime_ms, local_transmission=False): packet._tx_metadata = { From e24cdca0556f5e715f4d3175d669049bb3ca073f Mon Sep 17 00:00:00 2001 From: agessaman Date: Mon, 1 Jun 2026 17:05:38 -0700 Subject: [PATCH 5/5] feat(companion): echo injected TX to companion clients as raw RX (0x88) Push locally-injected TX packets to connected companion frame server clients as PUSH_CODE_LOG_RX_DATA (0x88) with snr=0/rssi=0, so apps that decrypt locally from raw RX (e.g. RemoteTerm) see companion-originated channel traffic. The originating companion is excluded so a node never hears its own transmission, matching physical firmware behavior. inject_packet now takes an origin_hash (threaded per-companion via the packet_injector partial); _on_raw_rx_for_companions gains exclude_hash to skip that companion's frame server. OTA RX is unaffected. --- repeater/main.py | 20 ++++++++-- repeater/packet_router.py | 28 +++++++++++++- tests/test_main_py_coverage.py | 9 +++++ tests/test_packet_router.py | 67 ++++++++++++++++++++++++++++++++++ 4 files changed, 120 insertions(+), 4 deletions(-) diff --git a/repeater/main.py b/repeater/main.py index 138b82a..de10c00 100644 --- a/repeater/main.py +++ b/repeater/main.py @@ -572,7 +572,12 @@ class RepeaterDaemon: bridge = RepeaterCompanionBridge( identity=identity, - packet_injector=self.router.inject_packet, + # Tag the injector with this companion's hash so inject_packet can + # skip its own frame server when echoing TX as raw RX (a node never + # hears its own transmission). + packet_injector=functools.partial( + self.router.inject_packet, origin_hash=companion_hash_str + ), node_name=node_name, radio_config=radio_config, sqlite_handler=sqlite_handler, @@ -819,12 +824,21 @@ class RepeaterDaemon: f"port={tcp_port}, bind={bind_address}, client_idle_timeout_sec={client_idle_timeout_sec}" ) - async def _on_raw_rx_for_companions(self, data: bytes, rssi: int, snr: float) -> None: - """Raw RX subscriber: push PUSH_CODE_LOG_RX_DATA (0x88) to connected companion clients.""" + async def _on_raw_rx_for_companions( + self, data: bytes, rssi: int, snr: float, exclude_hash: str | None = None + ) -> None: + """Raw RX subscriber: push PUSH_CODE_LOG_RX_DATA (0x88) to connected companion clients. + + ``exclude_hash`` skips the frame server for that companion hash; used when + echoing a companion's own injected TX so it never hears its own transmission. + OTA RX subscribers leave it unset, so received packets reach every companion. + """ servers = getattr(self, "companion_frame_servers", []) if not servers: return for fs in servers: + if exclude_hash is not None and getattr(fs, "companion_hash", None) == exclude_hash: + continue try: fs.push_rx_raw(snr, rssi, data) except Exception as e: diff --git a/repeater/packet_router.py b/repeater/packet_router.py index 832f8d1..3f2e3e7 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -160,7 +160,7 @@ class PacketRouter: pass await self.queue.put(packet) - async def inject_packet(self, packet, wait_for_ack: bool = False): + async def inject_packet(self, packet, wait_for_ack: bool = False, origin_hash=None): try: metadata = { "rssi": getattr(packet, "rssi", 0), @@ -182,6 +182,32 @@ class PacketRouter: # Mark so when this packet is dequeued we don't pass to engine again (avoid double-send / double-count) packet._injected_for_tx = True + # Echo this local TX to companion frame server clients as raw RX + # (PUSH_CODE_LOG_RX_DATA 0x88, snr=0/rssi=0 = local origin) so apps that + # decrypt locally from raw RX (e.g. RemoteTerm) see companion-originated + # traffic, matching what other mesh nodes would hear off the air. The + # originating companion (origin_hash) is excluded so it never hears its own TX. + push_rx = getattr(self.daemon, "_on_raw_rx_for_companions", None) + if push_rx is not None: + try: + raw = packet.write_to() + await push_rx(raw, 0, 0.0, exclude_hash=origin_hash) + servers = getattr(self.daemon, "companion_frame_servers", []) + pushed = sum( + 1 + for fs in servers + if getattr(fs, "companion_hash", None) != origin_hash + ) + logger.debug( + "Echoed injected TX as raw RX (0x88) to %d companion client(s) " + "(%d bytes, origin=%s excluded)", + pushed, + len(raw), + origin_hash, + ) + except Exception as e: + logger.debug("Failed to echo injected TX to companions: %s", e) + # Enqueue so router can deliver to companion(s): TXT_MSG -> dest bridge, ACK -> all bridges (sender sees ACK) await self.enqueue(packet) diff --git a/tests/test_main_py_coverage.py b/tests/test_main_py_coverage.py index ec7330f..cc388d0 100644 --- a/tests/test_main_py_coverage.py +++ b/tests/test_main_py_coverage.py @@ -143,6 +143,15 @@ async def test_raw_rx_and_duplicate_logging_hooks(): await daemon._on_raw_rx_for_companions(b"abc", rssi=-90, snr=2.0) fs_ok.push_rx_raw.assert_called_once() + # exclude_hash skips the matching companion's own frame server (no self-echo) + fs_self = SimpleNamespace(companion_hash="0x1a", push_rx_raw=MagicMock()) + fs_other = SimpleNamespace(companion_hash="0x2b", push_rx_raw=MagicMock()) + daemon.companion_frame_servers = [fs_self, fs_other] + await daemon._on_raw_rx_for_companions(b"xyz", rssi=0, snr=0.0, exclude_hash="0x1a") + fs_self.push_rx_raw.assert_not_called() + fs_other.push_rx_raw.assert_called_once() + daemon.companion_frame_servers = [fs_ok, fs_fail] + engine = SimpleNamespace( is_duplicate=MagicMock(side_effect=[False, True]), record_duplicate=MagicMock(), diff --git a/tests/test_packet_router.py b/tests/test_packet_router.py index 0ca8392..d441090 100644 --- a/tests/test_packet_router.py +++ b/tests/test_packet_router.py @@ -561,3 +561,70 @@ class TestPacketRouterRoutingBranches(unittest.IsolatedAsyncioTestCase): await router._route_packet(pkt) b1.process_received_packet.assert_awaited_once() daemon.repeater_handler.assert_awaited_once() + + +class TestInjectedTxRawEcho(unittest.IsolatedAsyncioTestCase): + """inject_packet echoes local TX to companion clients as raw RX (0x88).""" + + async def test_inject_packet_echoes_raw_tx_to_companions(self): + """Successful local TX is pushed via _on_raw_rx_for_companions with snr=0/rssi=0.""" + daemon = _make_daemon() + daemon._on_raw_rx_for_companions = AsyncMock() + router = PacketRouter(daemon) + pkt = _make_packet() + pkt.write_to.return_value = b"\x10\x20\x30" + + ok = await router.inject_packet(pkt) + + self.assertTrue(ok) + daemon._on_raw_rx_for_companions.assert_awaited_once_with( + b"\x10\x20\x30", 0, 0.0, exclude_hash=None + ) + + async def test_inject_packet_excludes_originating_companion(self): + """A companion's own TX is echoed with its hash excluded (no self-echo).""" + daemon = _make_daemon() + daemon._on_raw_rx_for_companions = AsyncMock() + router = PacketRouter(daemon) + pkt = _make_packet() + pkt.write_to.return_value = b"\xaa\xbb" + + ok = await router.inject_packet(pkt, origin_hash="0x1a") + + self.assertTrue(ok) + daemon._on_raw_rx_for_companions.assert_awaited_once_with( + b"\xaa\xbb", 0, 0.0, exclude_hash="0x1a" + ) + + async def test_inject_packet_no_echo_when_tx_fails(self): + """A failed local transmission must not echo a raw RX frame.""" + daemon = _make_daemon() + daemon.repeater_handler = AsyncMock(return_value=False) + daemon._on_raw_rx_for_companions = AsyncMock() + router = PacketRouter(daemon) + + ok = await router.inject_packet(_make_packet()) + + self.assertFalse(ok) + daemon._on_raw_rx_for_companions.assert_not_awaited() + + async def test_inject_packet_survives_echo_failure(self): + """An error while echoing must not fail the injection.""" + daemon = _make_daemon() + daemon._on_raw_rx_for_companions = AsyncMock(side_effect=RuntimeError("boom")) + router = PacketRouter(daemon) + + ok = await router.inject_packet(_make_packet()) + + self.assertTrue(ok) + daemon._on_raw_rx_for_companions.assert_awaited_once() + + async def test_inject_packet_without_echo_hook(self): + """Injection succeeds even if the daemon has no raw-RX companion hook.""" + daemon = _make_daemon() + daemon._on_raw_rx_for_companions = None + router = PacketRouter(daemon) + + ok = await router.inject_packet(_make_packet()) + + self.assertTrue(ok)