diff --git a/work/meshai/config.py b/work/meshai/config.py index 8e93466..2004dd8 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -50,6 +50,8 @@ class ConnectionConfig: meshcore_baud: int = 115200 meshcore_ble_address: str = "" # optional; for ble meshcore_auto_add_contacts: bool = True # firmware auto-adds every node it hears an advert from (so AIDA can DM anyone) + meshcore_ack_wait_seconds: float = 6.0 # wait for delivery ACK before falling back to path discovery + meshcore_discovery_wait_seconds: float = 8.0 # path-discovery timeout on the no-ACK fallback (was hardcoded 25s) def __post_init__(self): if self.meshcore_conn_type not in {"tcp", "serial", "ble"}: diff --git a/work/meshai/transport/meshcore_transport.py b/work/meshai/transport/meshcore_transport.py index 1d8e12e..c7176f7 100644 --- a/work/meshai/transport/meshcore_transport.py +++ b/work/meshai/transport/meshcore_transport.py @@ -76,6 +76,9 @@ class MeshCoreTransport(MeshTransport): # or None). None = pass-through (no filtering). Injected at construction # time by the factory; can also be (re)set via set_context_config(). self._mc_context = meshcore_context + # DM reply tuning (config knobs; getattr defaults keep old test configs valid). + self._ack_wait = float(getattr(config, "meshcore_ack_wait_seconds", 6.0)) + self._discovery_wait = float(getattr(config, "meshcore_discovery_wait_seconds", 8.0)) self._mc = None # meshcore.MeshCore instance self._loop: Optional[asyncio.AbstractEventLoop] = None self._loop_thread: Optional[threading.Thread] = None @@ -185,8 +188,8 @@ class MeshCoreTransport(MeshTransport): # response, so a subsequent send_msg will use the direct route. try: path_event = self._run_coro( - self._mc.commands.send_path_discovery_sync(contact, timeout=25), - timeout=30, + self._mc.commands.send_path_discovery_sync(contact, timeout=self._discovery_wait), + timeout=self._discovery_wait + 3, ) if path_event is not None and not path_event.is_error(): logger.info( @@ -236,6 +239,90 @@ class MeshCoreTransport(MeshTransport): except Exception as exc: logger.debug("MeshCore: _establish_direct_path step-2 error for %s: %s", dst, exc) + @staticmethod + def _extract_expected_ack(result): + """Pull the ``expected_ack`` value off a MSG_SENT event, or None. + + The lib's reader puts the raw 4-byte ack code at + ``payload["expected_ack"]`` (see reader.py MSG_SENT branch). Returns it + as-is (bytes on the real lib; may be a hex str in tests) or None if the + event has no dict payload / no ack. + """ + payload = getattr(result, "payload", None) + if isinstance(payload, dict): + return payload.get("expected_ack") + return None + + def _wait_for_ack(self, expected_ack, timeout: float) -> bool: + """Block until the delivery ACK matching *expected_ack* arrives, or timeout. + + Mirrors ``send_msg_with_retry``'s ACK wait (messaging.py): the firmware + emits ``EventType.ACK`` with a hex ``code`` attribute equal to the + MSG_SENT ``expected_ack`` rendered as ``.hex()`` (reader.py ACK branch). + We run the lib's dispatcher wait on the transport loop via ``_run_coro``. + + Returns True only if a matching ACK is dispatched before *timeout*; + False when *expected_ack* is falsy (can't confirm) or on any + timeout/exception. Coexists with the standing ``_on_ack_event`` + subscription — the dispatcher supports concurrent subscribers + waiters. + """ + if not expected_ack: + return False + from meshcore import EventType # noqa: PLC0415 (lazy import, matches module) + try: + code = ( + expected_ack.hex() + if isinstance(expected_ack, (bytes, bytearray)) + else str(expected_ack) + ) + except Exception: + return False + try: + event = self._run_coro( + self._mc.dispatcher.wait_for_event( + EventType.ACK, + attribute_filters={"code": code}, + timeout=timeout, + ), + timeout=timeout + 2, + ) + return event is not None + except Exception: + logger.debug( + "MeshCore: ACK wait failed for code %r", expected_ack, exc_info=True + ) + return False + + def _send_dm_once(self, contact: dict, text: str, destination: str): + """Send one DM frame (no discovery, no retry) and log its route. + + Uses plain ``send_msg`` — NOT ``send_msg_with_retry``, which calls + ``reset_path`` and forces flood, defeating any direct route we hold. + Returns the MSG_SENT event, or None if the radio produced no result. + """ + result = self._run_coro( + self._mc.commands.send_msg(contact, text), + timeout=15, + ) + if result is None: + logger.warning("MeshCoreTransport: DM to %s — no send result", destination) + return None + # Log whether the radio sent DIRECT or FLOOD from the MSG_SENT type field. + try: + sent_type = ( + result.payload.get("type") + if hasattr(result, "payload") and isinstance(result.payload, dict) + else None + ) + logger.info( + "MeshCore: DM to %s sent (route=%s)", + contact.get("adv_name") or destination, + "flood" if sent_type == 1 else ("direct" if sent_type == 0 else "?"), + ) + except Exception: + pass + return result + # ------------------------------------------------------------------ # Channel table enumeration # ------------------------------------------------------------------ @@ -640,38 +727,40 @@ class MeshCoreTransport(MeshTransport): destination, ) return False - # Establish a real route so the reply goes DIRECT — flood DMs are - # silently rejected on this mesh. + label = contact.get("adv_name") or destination + # FAST PATH: send the reply DIRECTLY (no discovery) and confirm + # by the delivery ACK the firmware returns. This is the common + # case (~1-3s) — the firmware already holds a usable route after + # the inbound DM primed it. We do NOT pay the path-discovery cost + # up front (that 25s wait was the ~25s reply-latency bug). + result = self._send_dm_once(contact, text, destination) + if result is None: + return False + exp_ack = self._extract_expected_ack(result) + if self._wait_for_ack(exp_ack, self._ack_wait): + logger.info("MeshCore: DM to %s ACKed (direct)", label) + return True + # No ACK -> route is stale/unknown. NOW run path discovery to + # learn a direct route, resend, and re-confirm by ACK. + logger.info( + "MeshCore: no ACK from %s in %.1fs; running path discovery and retrying", + label, self._ack_wait, + ) self._establish_direct_path(contact, destination) # Re-resolve so we send with the freshly-learned out_path. contact = self._resolve_contact(destination) or contact - # Plain send_msg (NOT send_msg_with_retry — that calls reset_path - # which forces flood, defeating the path we just established). - result = self._run_coro( - self._mc.commands.send_msg(contact, text), - timeout=15, - ) + result = self._send_dm_once(contact, text, destination) if result is None: - logger.warning("MeshCoreTransport: DM to %s — no send result", destination) return False - # Log whether the radio sent DIRECT or FLOOD from RESP_CODE_SENT type field. - try: - sent_type = ( - result.payload.get("type") - if hasattr(result, "payload") and isinstance(result.payload, dict) - else None - ) - logger.info( - "MeshCore: DM to %s sent (route=%s)", - contact.get("adv_name") or destination, - "flood" if sent_type == 1 else ("direct" if sent_type == 0 else "?"), - ) - except Exception: - pass - success = not result.is_error() - if not success: - logger.warning("MeshCoreTransport: DM send returned error event") - return success + exp_ack = self._extract_expected_ack(result) + acked = self._wait_for_ack(exp_ack, self._ack_wait) + logger.info( + "MeshCore: DM to %s %s after discovery", + contact.get("adv_name") or destination, + "ACKed" if acked else "no ACK (sent best-effort)", + ) + # Best-effort success if the frame was at least accepted by the radio. + return acked or (not result.is_error()) else: # Channel broadcast. # Channel-index semantics do NOT cross transports: the passed diff --git a/work/tests/test_meshcore_dm_delivery.py b/work/tests/test_meshcore_dm_delivery.py index 7fb80fd..7dfcb3a 100644 --- a/work/tests/test_meshcore_dm_delivery.py +++ b/work/tests/test_meshcore_dm_delivery.py @@ -1,14 +1,18 @@ -"""Focused tests for the MeshCore DM direct-route delivery fix. +"""Focused tests for the MeshCore DM ACK-confirmed fast-path delivery. Verifies that send_message(..., destination=...) resolves the destination to -the full contact object, calls _establish_direct_path (path discovery via -send_path_discovery_sync) BEFORE send_msg, uses plain send_msg (NOT +the full contact object, sends the reply DIRECTLY first (fast path) and only +runs _establish_direct_path (path discovery via send_path_discovery_sync) on +the no-ACK fallback — never before the first send. Uses plain send_msg (NOT send_msg_with_retry), and correctly maps the return value to True/False: - non-error Event returned → True (sent, route logged) - None returned → False (no send result) - error Event returned → False - contact not in roster → False (logged warning, send never called) +Note: these mocks provide no dispatcher ACK, so _wait_for_ack returns False and +every send exercises the no-ACK fallback (discovery + resend) leg. + The meshcore lib is mocked via sys.modules (same pattern as the existing transport test module). _run_coro is patched to execute the coroutine synchronously so no background event-loop thread is needed. @@ -195,7 +199,8 @@ def _transport_with_mc_mock(contact=_CONTACT_DICT): # --------------------------------------------------------------------------- class TestMeshCoreDMDelivery: - """send_message with destination= must call path discovery then plain send_msg.""" + """send_message with destination= sends directly first (ACK fast path); path + discovery is a no-ACK fallback that runs after the first send, not before.""" def test_successful_dm_returns_true(self): """Non-error send_msg result → True.""" @@ -205,8 +210,14 @@ class TestMeshCoreDMDelivery: assert result is True - def test_path_discovery_called_before_send_msg(self): - """send_path_discovery_sync must be called before send_msg.""" + def test_fast_path_sends_before_discovery(self): + """ACK-confirmed fast path: send_msg goes out FIRST; path discovery only + runs on the no-ACK fallback (i.e. AFTER the first send), never before it. + + (This mock has no dispatcher ACK, so _wait_for_ack returns False and the + no-ACK fallback always fires — which is exactly what exercises the + discovery-then-resend leg here.) + """ t, mc = _transport_with_mc_mock() call_order = [] @@ -223,10 +234,11 @@ class TestMeshCoreDMDelivery: t.send_message("hello", destination="aabbccdd1122") - assert "discovery" in call_order, "send_path_discovery_sync was never called" assert "send_msg" in call_order, "send_msg was never called" - assert call_order.index("discovery") < call_order.index("send_msg"), ( - "send_path_discovery_sync must be called before send_msg" + assert "discovery" in call_order, "send_path_discovery_sync was never called (no-ACK fallback)" + # The reply is sent DIRECTLY first; discovery is a fallback that comes after. + assert call_order.index("send_msg") < call_order.index("discovery"), ( + "fast path must send the reply before running path discovery" ) def test_send_msg_used_not_send_msg_with_retry(self): diff --git a/work/tests/test_meshcore_transport.py b/work/tests/test_meshcore_transport.py index b2ab3a1..9ae5b51 100644 --- a/work/tests/test_meshcore_transport.py +++ b/work/tests/test_meshcore_transport.py @@ -339,43 +339,82 @@ class TestSendMessageChannel: _DM_CONTACT = {"public_key": "a" * 64, "adv_name": "TestContact", "out_path_len": -1} +def _msg_sent(expected_ack=b"\x01\x02\x03\x04", type_=0, is_error=False): + """Build a fake MSG_SENT event carrying an expected_ack (like reader.py).""" + ev = MagicMock() + ev.is_error.return_value = is_error + ev.payload = {"type": type_, "expected_ack": expected_ack} + return ev + + class TestSendMessageDM: - def test_dispatches_send_msg_with_retry(self): - """DM send: path discovery then plain send_msg (not send_msg_with_retry).""" + def test_fast_path_acked_no_discovery(self): + """ACK to the direct send → success WITHOUT path discovery, send_msg once.""" t, mc, _ = _transport_with_mock_mc() try: - # Supply a resolved contact so _resolve_contact succeeds. mc.get_contact_by_key_prefix.return_value = _DM_CONTACT mc.ensure_contacts = AsyncMock(return_value=True) - # Path discovery must be awaitable. - path_ev = MagicMock() - path_ev.is_error.return_value = False - mc.commands.send_path_discovery_sync = AsyncMock(return_value=path_ev) - ok = MagicMock() - ok.is_error.return_value = False - ok.payload = {"type": 0, "expected_ack": "00000000"} - mc.commands.send_msg = AsyncMock(return_value=ok) + mc.commands.send_msg = AsyncMock(return_value=_msg_sent()) + # Delivery ACK arrives → wait_for_event returns a matching ACK event. + ack_ev = MagicMock() + mc.dispatcher.wait_for_event = AsyncMock(return_value=ack_ev) + # Spy: discovery must NOT run on the fast path. + t._establish_direct_path = MagicMock() + result = t.send_message("hi DM", destination="aabbcc") + assert result is True - # Must use send_msg (not send_msg_with_retry) with the CONTACT OBJECT. + t._establish_direct_path.assert_not_called() mc.commands.send_msg.assert_awaited_once_with(_DM_CONTACT, "hi DM") + # ACK was matched on the hex of expected_ack (b"\x01\x02\x03\x04"). + _, kwargs = mc.dispatcher.wait_for_event.await_args + assert kwargs["attribute_filters"] == {"code": "01020304"} finally: _cleanup(t) - def test_send_msg_with_retry_error_returns_false(self): - """Error event from send_msg → False.""" + def test_no_ack_falls_back_to_discovery_then_acks(self): + """No ACK on the direct send → discovery + resend; 2nd ACK → success.""" t, mc, _ = _transport_with_mock_mc() try: mc.get_contact_by_key_prefix.return_value = _DM_CONTACT mc.ensure_contacts = AsyncMock(return_value=True) - path_ev = MagicMock() - path_ev.is_error.return_value = False - mc.commands.send_path_discovery_sync = AsyncMock(return_value=path_ev) - err = MagicMock() - err.is_error.return_value = True - err.payload = {"reason": "test"} - mc.commands.send_msg = AsyncMock(return_value=err) + mc.commands.send_msg = AsyncMock(return_value=_msg_sent()) + ack_ev = MagicMock() + # 1st wait times out (None); 2nd (post-discovery) returns an ACK. + mc.dispatcher.wait_for_event = AsyncMock(side_effect=[None, ack_ev]) + t._establish_direct_path = MagicMock() + + result = t.send_message("hi DM", destination="aabbcc") + + assert result is True + t._establish_direct_path.assert_called_once() + assert mc.commands.send_msg.await_count == 2 + finally: + _cleanup(t) + + def test_wait_for_ack_false_on_none_and_on_exception(self): + """_wait_for_ack: falsy exp_ack → False; wait_for_event raising → False.""" + t, mc, _ = _transport_with_mock_mc() + try: + # Falsy expected_ack short-circuits without touching the dispatcher. + assert t._wait_for_ack(None, 1.0) is False + assert t._wait_for_ack(b"", 1.0) is False + # A raising wait_for_event is swallowed → False. + mc.dispatcher.wait_for_event = AsyncMock(side_effect=RuntimeError("boom")) + assert t._wait_for_ack(b"\x01\x02\x03\x04", 1.0) is False + finally: + _cleanup(t) + + def test_direct_send_no_result_returns_false(self): + """send_msg returning None (no radio result) → False, no discovery.""" + t, mc, _ = _transport_with_mock_mc() + try: + mc.get_contact_by_key_prefix.return_value = _DM_CONTACT + mc.ensure_contacts = AsyncMock(return_value=True) + mc.commands.send_msg = AsyncMock(return_value=None) + t._establish_direct_path = MagicMock() assert t.send_message("hi", destination="deadbeef") is False + t._establish_direct_path.assert_not_called() finally: _cleanup(t)