From 44e60c40549474bbce5af77237bb5f66fba73cf3 Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Mon, 5 Oct 2026 03:00:48 -0400 Subject: [PATCH] Acks get the wait a flood needs, and a late one still counts Only 24 of 144 direct messages and posts in the history were ever marked delivered. Three faults, each fixed: - The wait was the radio's own estimate times 1.2: about 5.7 s for a flood, while a flooded round trip from a few hops out takes ten (a room's answer from five hops landed at 10.1 s on hardware). It is now the login's patience, the estimate stretched a quarter plus a round trip sized to the route, about 15 s for a flood. - An ack that landed after the wait was dropped. The radio pushes one whenever it arrives, for any of its last eight sends, but nothing was listening, so a delivered message stayed failed for good. The history shows ACK and PATH packets heard seconds after messages recorded as failed. The send now returns a Delivery carrying the ack code, and the chat service keeps the last sixteen messages still waiting. A late ack flips one to delivered, in history and in the message the chat screen holds, and every code a retried message went out under counts. - The Courier trusted the failure and sent the message again: a duplicate for the recipient whenever the ack was only late. A late ack now marks the entry delivered. The ack listener is also armed before the send rather than after it, so a neighbour's ack quicker than the send's own return is not missed. The debug log names how a message went out and the radio's own round-trip time. Co-Authored-By: Claude Opus 5.5 (1M context) --- meshterm/core/connection.py | 138 +++++++++++++++++++----- meshterm/core/models.py | 30 ++++++ meshterm/services/chat_service.py | 130 +++++++++++++++++++---- meshterm/services/courier.py | 26 ++++- tests/test_chat.py | 169 +++++++++++++++++++++++++++++- tests/test_courier.py | 28 ++++- tests/test_rooms.py | 6 +- 7 files changed, 468 insertions(+), 59 deletions(-) diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index 0966732..db388fc 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -39,6 +39,7 @@ from .models import ( NODE_TYPE_ROOM, Ack, Contact, + Delivery, Hop, LoginResult, Message, @@ -1386,16 +1387,17 @@ class Device(ABC): # -- messaging --------------------------------------------------------------- @abstractmethod - async def send_direct_message(self, contact: Contact, text: str) -> Ack | None: - """Send a direct text message to a contact. + async def send_direct_message(self, contact: Contact, text: str) -> Delivery: + """Send a direct text message to a contact, and wait a while for its ack. Args: contact: The recipient; its ``public_key`` addresses the message. text: The message body. Returns: - The delivery :class:`Ack` if one arrived before the send timed out, else - ``None`` (the message was handed to the radio but not yet acknowledged). + The :class:`~meshterm.core.models.Delivery`: the ack code the radio expects + back, and the ack itself if it arrived within the wait. One that comes later is + still pushed by the radio, as an ``ACK`` event carrying the same code. Raises: DeviceCommandError: If the companion rejected the send outright. @@ -3807,35 +3809,86 @@ class MeshCoreDevice(Device): async def send_direct_message( # noqa: D102 - inherited docstring self, contact: Contact, text: str - ) -> Ack | None: + ) -> Delivery: transmit_gate.mark() from meshcore import EventType mc = self._require() pub = self._node_pubkey(contact) - # Only the hand-over holds the transmit lock; the ack wait below is not a send. - async with self.transmitting(): - result = await mc.commands.send_msg(pub, text) - if result is None or getattr(result, "is_error", lambda: False)(): - # A recipient the firmware has no entry for is the one rejection with an obvious - # fix, so it gets its own class for the chat screen to offer that fix on; see - # :class:`ContactNotOnDeviceError` for how a listed contact can be missing here. - if error_code(result) == _ERR_NOT_FOUND: - raise ContactNotOnDeviceError(contact) - raise DeviceCommandError(f"couldn't send to {contact.name}: {reject_reason(result)}") - # The companion acknowledges the send immediately with an ``expected_ack`` code and - # a suggested wait; the recipient's delivery ACK arrives later carrying that code. - payload = getattr(result, "payload", {}) or {} - expected = payload.get("expected_ack") - expected_hex = expected.hex() if isinstance(expected, (bytes, bytearray)) else expected - if not expected_hex: - return None - suggested = payload.get("suggested_timeout") - timeout = suggested / 1000 * 1.2 if suggested else 8.0 - ack = await mc.wait_for_event( - EventType.ACK, attribute_filters={"code": expected_hex}, timeout=timeout + loop = asyncio.get_running_loop() + # Listen before sending, and to every ack: the code to match is only known once the + # send returns, and an ack quicker than that return (a neighbour's) used to land + # before anything was listening for it. Kept by code until the code is known. + heard: dict[str, object] = {} + wanted: asyncio.Future = loop.create_future() + code: list[str] = [] + flood = False + budget = 0.0 + + def on_ack(event) -> None: # noqa: ANN001 - meshcore Event + got = str((getattr(event, "payload", None) or {}).get("code") or "").lower() + heard[got] = event + if code and got == code[0] and not wanted.done(): + wanted.set_result(event) + + subscription = mc.subscribe(EventType.ACK, on_ack) + started = loop.time() + try: + # Only the hand-over holds the transmit lock; the ack wait below is not a send. + async with self.transmitting(): + result = await mc.commands.send_msg(pub, text) + if result is None or getattr(result, "is_error", lambda: False)(): + # A recipient the firmware has no entry for is the one rejection with an + # obvious fix, so it gets its own class for the chat screen to offer that + # fix on; see :class:`ContactNotOnDeviceError` for how a listed contact can + # be missing here. + if error_code(result) == _ERR_NOT_FOUND: + raise ContactNotOnDeviceError(contact) + raise DeviceCommandError( + f"couldn't send to {contact.name}: {reject_reason(result)}" + ) + # The companion accepts the send at once with the ``expected_ack`` code and how + # it sent it; the recipient's ack arrives later, carrying the code. + payload = getattr(result, "payload", {}) or {} + expected = payload.get("expected_ack") + expected_hex = expected.hex() if isinstance(expected, (bytes, bytearray)) else expected + if not expected_hex: + return Delivery(None) + code.append(str(expected_hex).lower()) + if code[0] in heard: + wanted.set_result(heard[code[0]]) + flood = payload.get("type") == 1 + # As patient as a login, whose answer travels exactly as an ack does: the + # radio's own estimate (``suggested_timeout``, which the library stretches by a + # quarter), then a round trip sized to the route. The estimate alone (what this + # waited until 2026-10-05) is ``500 ms + 16 x airtime`` for a flood, about five + # seconds, where a flood's answer from five hops out took ten on hardware. So + # most acks from beyond a neighbour landed after the wait had given up, and four + # messages in five read as failed that had been delivered. + hops = 0 if flood else 2 * len(contact.route_hops or ()) + suggested = float(payload.get("suggested_timeout") or 0) / 1000.0 + budget = suggested * 1.25 + max(trace_timeout(hops), trace_timeout(0)) + if not wanted.done(): + try: + await asyncio.wait_for(asyncio.shield(wanted), budget) + except asyncio.TimeoutError: + pass + finally: + subscription.unsubscribe() + event = wanted.result() if wanted.done() else None + ack = ack_from_event(event) if event is not None else None + _log.debug( + "direct message to %s: sent by %s, code %s, budget=%.1fs -> %s", + contact.name, + "flood" if flood else "its route", + code[0], + budget, + f"acked in {(loop.time() - started) * 1000.0:.0f}ms " + f"(the radio's round trip: {(ack.raw or {}).get('trip_time')}ms)" + if ack is not None + else "no ack yet", ) - return ack_from_event(ack) if ack is not None else None + return Delivery(code[0], ack, flood) async def send_channel_message( # noqa: D102 - inherited docstring self, index: int, text: str @@ -4313,6 +4366,12 @@ class MockDevice(Device): #: Whoever is subscribed to the event stream right now, so a room can push its posts #: to them after a login the way a real room's catch-up arrives — unsolicited. self._listeners: list[EventCallback] = [] + #: Contacts whose acks arrive only after the send has stopped waiting, the common + #: case beyond a neighbour on a real mesh. Empty by default; name one to walk the + #: late-ack path. + self._late_acks: set[str] = set() + #: Direct messages handed over so far, which numbers each one's ack code. + self._sent_codes = 0 #: Nodes whose learned route has gone stale: a message sent along it meets silence, #: and a flood gets through and learns a fresh one. Empty by default; name a contact #: here to walk the "try by flood" path a stale route leaves a reader on. @@ -4489,17 +4548,38 @@ class MockDevice(Device): # chat screen's "add it back and send" offer walkable on the simulator. if self._mock_key(contact) not in {self._mock_key(c) for c in self._contacts}: raise ContactNotOnDeviceError(contact) + self._sent_codes += 1 + code = f"{self._sent_codes:08x}" + held = next(c for c in self._contacts if self._mock_key(c) == self._mock_key(contact)) + flood = held.route_hops is None if contact.is_room: # A post. The room keeps it — and acknowledges — only from a member it knows # who may post; anyone else's is dropped with no reply, which is all a read-only # member ever learns about why a post went nowhere. access = self._room_access.get(self._mock_key(contact)) if access is None or not access.can_post: - return None + return Delivery(code, None, flood) self._room_board.append((self._self_prefix(), utcnow(), text)) + if contact.name in self._late_acks: + # Delivered, but the ack takes the long way home: it arrives after the wait, + # pushed the way the radio pushes one whenever it lands. + self._push_ack_later(code) + return Delivery(code, None, flood) # Otherwise the simulator "delivers" instantly and always acknowledges, so outbound # direct messages show as acked without a radio. - return Ack(code="mock") + return Delivery(code, Ack(code=code), flood) + + def _push_ack_later(self, code: str) -> None: + """Push an ack to whoever listens, a beat after the send returned without one.""" + + async def push() -> None: + await asyncio.sleep(_MOCK_MONITOR_INTERVAL_S) + for listener in list(self._listeners): + listener(MeshEvent.ack_event(Ack(code=code))) + + task = asyncio.create_task(push()) + self._bg_tasks.add(task) + task.add_done_callback(self._bg_tasks.discard) async def send_channel_message( # noqa: D102 - inherited docstring self, index: int, text: str diff --git a/meshterm/core/models.py b/meshterm/core/models.py index 3ea636b..39c5dd9 100644 --- a/meshterm/core/models.py +++ b/meshterm/core/models.py @@ -472,6 +472,36 @@ class Ack: raw: dict | None = None +@dataclass(slots=True) +class Delivery: + """A direct message the radio took: the ack it is waiting for, and whether it came in time. + + The ack is the recipient's proof of receipt (for a room, of *storing* the post). It is + keyed by a four-byte code the radio derives from the message and hands back the moment + it accepts the send; the ack, whenever it arrives, carries the same code. Keeping the + code is what lets an ack that lands after the wait still count — the radio pushes it + regardless, and a flood's ack, riding back on the path the message found, often takes + longer than any wait a person would sit through. + + Attributes: + code: The ack code the radio expects back, lowercase hex; ``None`` when the radio + gave none (nothing to wait for, and nothing a late ack could match). + ack: The acknowledgement, when it arrived within the wait; ``None`` otherwise — + which is *not* "not delivered", only "not confirmed yet". + flood: How the radio sent it: ``True`` to the whole mesh, ``False`` along the route + it learned; ``None`` when it didn't say. + """ + + code: str | None + ack: Ack | None = None + flood: bool | None = None + + @property + def acked(self) -> bool: + """Whether the recipient's acknowledgement arrived within the wait.""" + return self.ack is not None + + def conversation_key(is_channel: bool, channel_id: str | None, peer: str | None) -> str: """Return a stable key identifying a chat conversation. diff --git a/meshterm/services/chat_service.py b/meshterm/services/chat_service.py index a4d42e6..a687a0c 100644 --- a/meshterm/services/chat_service.py +++ b/meshterm/services/chat_service.py @@ -18,6 +18,8 @@ communications, not overheard noise, so they are always recorded while a device from __future__ import annotations import asyncio +from collections import OrderedDict +from collections.abc import Callable from typing import TYPE_CHECKING from ..core.channels import channel_identity @@ -26,6 +28,7 @@ from ..core.events import EventKind, MeshEvent from ..core.models import ( ChatMessage, Contact, + Delivery, Message, is_direct_messageable, utcnow, @@ -36,6 +39,15 @@ if TYPE_CHECKING: from ..context import AppContext +#: How many unacknowledged messages keep waiting for a late ack. The radio itself remembers +#: the codes of only its last eight sends (MeshCore's ``EXPECTED_ACK_TABLE_SIZE``) and +#: pushes an ack for none older, so a few more than that covers every ack still possible. +_AWAITING_CAP = 16 + +#: Called with a message once its ack arrives after the send stopped waiting for it. +LateAck = Callable[[ChatMessage], None] + + def _fallback_channel_id(idx: int | None) -> str: """Identity for a channel we can't read (an unconfigured/unknown slot). @@ -80,6 +92,11 @@ class ChatService: # slot fresh every time (see :meth:`channel_id_for`), so this is not the source of # truth — only a warm map for priming/backfill and a fallback when a read fails. self._channel_ids: dict[int, str] = {} + # Messages sent whose ack had not arrived when the send stopped waiting, by the ack + # code a late one would carry (every attempt's, for a retried message), oldest + # first, each with whoever wants to hear when it lands (see :meth:`_settle_ack`). + self._awaiting: OrderedDict[str, tuple[ChatMessage, LateAck | None]] = OrderedDict() + self._ack_unsubscribe: Unsubscribe | None = None @property def active(self) -> bool: @@ -234,6 +251,7 @@ class ChatService: queue.put_nowait(message) self._unsubscribe = self._ctx.events.subscribe(on_event, EventKind.MESSAGE) + self._ack_unsubscribe = self._ctx.events.subscribe(self._on_ack, EventKind.ACK) self._run_id = run_id self._ctx.log.info("chat recording (run %s)", run_id) await self._prime_channels() @@ -264,8 +282,11 @@ class ChatService: try: assert self._unsubscribe is not None self._unsubscribe() + if self._ack_unsubscribe is not None: + self._ack_unsubscribe() finally: self._unsubscribe = None + self._ack_unsubscribe = None if self._worker is not None: self._worker.cancel() try: @@ -434,7 +455,7 @@ class ChatService: return is_direct_messageable(contact.node_type) return False - async def _deliver_direct(self, contact: Contact, text: str): + async def _deliver_direct(self, contact: Contact, text: str) -> tuple[bool, list[str]]: """Transmit a direct message, softly retrying until it is acknowledged. A single logical send makes one initial transmission plus up to @@ -454,8 +475,8 @@ class ChatService: text: The message body. Returns: - The delivery :class:`~meshterm.core.models.Ack` from the first attempt that landed, - or ``None`` if every attempt went unacknowledged. + Whether an attempt was acknowledged within its wait, and the ack code of every + attempt made — any of which a late ack may still carry (the radio keeps them). Raises: Exception: Propagates a hard send failure (the companion rejecting the send); such @@ -463,11 +484,13 @@ class ChatService: """ device = await self._ctx.device() attempts = max(0, self._ctx.preferences.direct_message_soft_retries) + 1 - ack = None + codes: list[str] = [] for attempt in range(1, attempts + 1): - ack = await device.send_direct_message(contact, text) - if ack is not None: - return ack + delivery: Delivery = await device.send_direct_message(contact, text) + if delivery.code: + codes.append(delivery.code) + if delivery.acked: + return True, codes if attempt < attempts: self._ctx.log.debug( "chat: DM to %s unacked on try %d/%d; soft-retrying", @@ -475,34 +498,41 @@ class ChatService: attempt, attempts, ) - return ack + return False, codes - async def send_direct(self, contact: Contact, text: str) -> ChatMessage: + async def send_direct( + self, contact: Contact, text: str, *, on_late_ack: LateAck | None = None + ) -> ChatMessage: """Send a direct message to a contact and record it in history. Delivery is attempted with up to ``preferences.direct_message_soft_retries`` automatic soft retries (see :meth:`_deliver_direct`) before the message is recorded as - unacknowledged. + unacknowledged — which means *not confirmed yet*: an ack that lands later still + settles it (see :meth:`_settle_ack`), in history and on the screen showing it. Args: contact: The recipient. text: The message body. + on_late_ack: Called with the recorded message if its ack arrives after the + send stopped waiting (the Courier, which would otherwise send it again). Returns: The recorded outbound :class:`ChatMessage` (its ``acked`` reflects whether a - delivery acknowledgement arrived within the soft-retry budget). + delivery acknowledgement arrived within the wait, so far). """ - ack = await self._deliver_direct(contact, text) + acked, codes = await self._deliver_direct(contact, text) chat = ChatMessage( text=text, outbound=True, is_channel=False, peer=contact.key_prefix or contact.public_key[:12] or None, peer_name=contact.name, - acked=ack is not None, + acked=acked, created_at=utcnow(), ) chat.row_id = self._ctx.repo.record_chat_message(chat, run_id=self._run_id) + if not acked: + self._await_ack(chat, codes, on_late_ack) return chat async def resend_direct(self, contact: Contact, message: ChatMessage) -> ChatMessage: @@ -523,10 +553,14 @@ class ChatService: Returns: The same ``message``, with :attr:`~ChatMessage.acked` refreshed. """ - ack = await self._deliver_direct(contact, message.text) - message.acked = ack is not None + acked, codes = await self._deliver_direct(contact, message.text) + if message.acked: + return message # a late ack for an earlier attempt settled it meanwhile + message.acked = acked if message.row_id is not None: self._ctx.repo.update_chat_ack(message.row_id, message.acked) + if not acked: + self._await_ack(message, codes, None) return message async def send_post(self, room: Contact, text: str) -> ChatMessage: @@ -553,17 +587,19 @@ class ChatService: stored it. """ device = await self._ctx.device() - ack = await device.send_direct_message(room, text) + delivery = await device.send_direct_message(room, text) chat = ChatMessage( text=text, outbound=True, is_channel=False, peer=room.key_prefix or room.public_key[:12] or None, peer_name=room.name, - acked=ack is not None, + acked=delivery.acked, created_at=utcnow(), ) chat.row_id = self._ctx.repo.record_chat_message(chat, run_id=self._run_id) + if not delivery.acked: + self._await_ack(chat, [delivery.code] if delivery.code else [], None) return chat async def resend_post(self, room: Contact, message: ChatMessage) -> ChatMessage: @@ -577,12 +613,68 @@ class ChatService: The same ``message``, with :attr:`~ChatMessage.acked` refreshed. """ device = await self._ctx.device() - ack = await device.send_direct_message(room, message.text) - message.acked = ack is not None + delivery = await device.send_direct_message(room, message.text) + if message.acked: + return message # a late ack for the first post settled it meanwhile + message.acked = delivery.acked if message.row_id is not None: self._ctx.repo.update_chat_ack(message.row_id, message.acked) + if not delivery.acked: + self._await_ack(message, [delivery.code] if delivery.code else [], None) return message + # --- late acks ------------------------------------------------------------------------ + + def _await_ack(self, chat: ChatMessage, codes: list[str], on_late_ack: LateAck | None) -> None: + """Keep listening for a sent message's ack after the send has stopped waiting. + + The radio pushes an ack whenever it lands, however late, for any of its last eight + sends; a flood's ack, riding home on the path the message found, routinely lands + after the wait. So the message is filed under every code it was sent with, the + oldest entries giving way past :data:`_AWAITING_CAP`. + """ + for code in codes: + self._awaiting[code.lower()] = (chat, on_late_ack) + self._awaiting.move_to_end(code.lower()) + while len(self._awaiting) > _AWAITING_CAP: + self._awaiting.popitem(last=False) + + def _on_ack(self, event: MeshEvent) -> None: + """Hub callback: an ack arrived; settle the message it belongs to, if one waits.""" + ack = event.ack + if ack is None or not ack.code: + return + self._settle_ack(str(ack.code).lower()) + + def _settle_ack(self, code: str) -> None: + """Mark the message waiting on ``code`` delivered, everywhere it is shown or kept. + + The history row is updated, and the very :class:`ChatMessage` the chat screen + holds flips to acknowledged, so its ✗ turns to ✓ on the next paint. Whoever asked + to hear of it (the Courier) is told. Every other code the message was sent under + (an earlier attempt's) is let go: it was delivered once, and that is the news. + """ + entry = self._awaiting.pop(code, None) + if entry is None: + return + chat, on_late_ack = entry + for other in [c for c, (held, _) in self._awaiting.items() if held is chat]: + del self._awaiting[other] + if chat.acked: + return + chat.acked = True + if chat.row_id is not None: + try: + self._ctx.repo.update_chat_ack(chat.row_id, True) + except Exception as exc: # noqa: BLE001 - the screen still shows it delivered + self._ctx.log.debug("chat: failed to record a late ack: %s", exc) + self._ctx.log.info("chat: late ack from %s", chat.peer_name or chat.peer) + if on_late_ack is not None: + try: + on_late_ack(chat) + except Exception as exc: # noqa: BLE001 - one listener must not break the hub + self._ctx.log.debug("chat: late-ack listener failed: %s", exc) + def channel_scope(self, channel_id: str | None) -> str | None: """The region a channel's messages are sent under, or ``None`` for the device default. diff --git a/meshterm/services/courier.py b/meshterm/services/courier.py index a474489..f3cd338 100644 --- a/meshterm/services/courier.py +++ b/meshterm/services/courier.py @@ -30,7 +30,7 @@ from datetime import datetime from typing import TYPE_CHECKING from ..core.connection import Unsubscribe -from ..core.courier_store import QUEUED, QueuedMessage +from ..core.courier_store import DELIVERED, QUEUED, QueuedMessage from ..core.events import EventKind, MeshEvent from ..core.models import Contact, utcnow @@ -218,7 +218,11 @@ class CourierService: self._ctx.log.debug("courier: no contact for %r; leaving queued", message.node_name) return "unknown contact" store.note_attempt(message.ident) - chat = await self._ctx.chat.send_direct(contact, message.text) + chat = await self._ctx.chat.send_direct( + contact, + message.text, + on_late_ack=lambda _chat, ident=message.ident: self._delivered_late(ident), + ) if chat.acked: store.mark_delivered(message.ident) self._ctx.watch_store.add_alert( @@ -246,6 +250,24 @@ class CourierService: finally: self._sending = False + def _delivered_late(self, ident: int) -> None: + """An attempt's ack arrived after the attempt stopped waiting: it was delivered. + + Without this the entry stayed queued (or gave up) and the next attempt sent the + recipient the same message again — every late ack a duplicate on their screen. + """ + store = self._ctx.courier_store + message = store.get(ident) + if message is None or message.status == DELIVERED: + return + store.mark_delivered(ident) + self._ctx.watch_store.add_alert( + "courier", + message.node_name, + f"delivered “{_preview(message.text)}” (attempt {message.attempts}, ack came late)", + ) + self._ctx.log.info("courier: delivered to %s (late ack)", message.node_name) + async def _resolve(self, message: QueuedMessage) -> Contact | None: """Find the recipient in the device's contact list, by key then by name.""" device = await self._ctx.device() diff --git a/tests/test_chat.py b/tests/test_chat.py index 5c079e6..1df467a 100644 --- a/tests/test_chat.py +++ b/tests/test_chat.py @@ -113,8 +113,8 @@ async def test_mock_send_direct_returns_ack() -> None: device = MockDevice() await device.connect() contact = (await device.get_contacts())[0] - ack = await device.send_direct_message(contact, "hello") - assert ack is not None + delivery = await device.send_direct_message(contact, "hello") + assert delivery.acked and delivery.code await device.send_channel_message(0, "hi channel") # no return, must not raise await device.disconnect() @@ -146,7 +146,7 @@ async def test_adding_a_forgotten_contact_back_makes_it_messageable() -> None: stranger = Contact(name="bob_caribou", public_key="a3" * 32, key_prefix="a3a3a3a3") await device.add_contact(stranger) assert "bob_caribou" in {c.name for c in await device.get_contacts()} - assert await device.send_direct_message(stranger, "hello") is not None + assert (await device.send_direct_message(stranger, "hello")).acked await device.disconnect() @@ -1603,6 +1603,7 @@ class _ScriptedDevice: stand-in (any non-``None`` object counts as delivered) or ``None`` for an unacknowledged transmission — and appends the sent text to :attr:`sent`, so a test can assert exactly how many soft retries fired. Once the script is exhausted every further send goes unacked. + Each send gets its own ack code, as the radio's do. """ def __init__(self, acks: list) -> None: @@ -1613,8 +1614,11 @@ class _ScriptedDevice: return None async def send_direct_message(self, contact: Contact, text: str): + from meshterm.core.models import Delivery + self.sent.append(text) - return self._acks.pop(0) if self._acks else None + ack = self._acks.pop(0) if self._acks else None + return Delivery(f"{len(self.sent):08x}", ack) def _contact() -> Contact: @@ -2568,3 +2572,160 @@ def test_a_link_written_without_a_scheme_gets_its_code_too() -> None: assert isinstance(opened[0], QrScreen) and opened[0].urls == ("https://meshterm.net/map",) asyncio.run(scenario()) + + +# -- acks: listening early, waiting long enough, and catching the late ones ---- + + +class _AckingMeshCore: + """A meshcore stand-in whose send accepts the message, then acks it when scripted. + + ``ack_at`` is seconds after the send returns (``None``: never); ``during_send`` pushes + the ack before the send has even returned, as a neighbour's can. + """ + + def __init__(self, *, ack_at: float | None, during_send: bool = False, flood: bool = True): + self.subs: list = [] + self.commands = self + self._ack_at = ack_at + self._during = during_send + self._flood = flood + + def subscribe(self, event_type, callback, attribute_filters=None): # noqa: ANN001, ANN201 + from types import SimpleNamespace + + sub = SimpleNamespace(event_type=event_type, callback=callback) + sub.unsubscribe = lambda: self.subs.remove(sub) if sub in self.subs else None + self.subs.append(sub) + return sub + + def _push_ack(self) -> None: + from types import SimpleNamespace + + from meshcore import EventType + + event = SimpleNamespace(type=EventType.ACK, payload={"code": "c0ffee01", "trip_time": 900}) + for sub in list(self.subs): + if sub.event_type is EventType.ACK: + sub.callback(event) + + async def send_msg(self, pub, text): # noqa: ANN001, ANN201 + from types import SimpleNamespace + + if self._during: + self._push_ack() + elif self._ack_at is not None: + asyncio.get_running_loop().call_later(self._ack_at, self._push_ack) + return SimpleNamespace( + type="MSG_SENT", + is_error=lambda: False, + payload={ + "type": 1 if self._flood else 0, + "expected_ack": bytes.fromhex("c0ffee01"), + "suggested_timeout": 100, # 0.1 s: the radio's (too hopeful) estimate + }, + ) + + +def _acking_device(mc) -> object: # noqa: ANN001 + from meshterm.core.connection import MeshCoreDevice + + device = MeshCoreDevice(port="mock") + device._mc = mc + return device + + +async def test_an_ack_quicker_than_the_send_is_still_caught() -> None: + """The listener is armed before the send, so a neighbour's instant ack isn't missed.""" + device = _acking_device(_AckingMeshCore(ack_at=None, during_send=True)) + delivery = await device.send_direct_message(_contact(), "hi") + assert delivery.acked and delivery.code == "c0ffee01" and delivery.flood is True + + +async def test_an_ack_after_the_radios_estimate_is_still_waited_for(monkeypatch) -> None: # noqa: ANN001 + """THE bug: the wait was the radio's estimate × 1.2 — 0.12 s here, five on hardware. + + A flood's ack rides home on the path the message found, and on hardware took twice the + estimate. The wait is now the estimate plus a round trip sized like a login's. + """ + import meshterm.core.connection as connection + + monkeypatch.setattr(connection, "trace_timeout", lambda hops: 0.5) + device = _acking_device(_AckingMeshCore(ack_at=0.3)) + delivery = await device.send_direct_message(_contact(), "hi") + assert delivery.acked, "an ack at 0.3 s, after the old 0.12 s wait, counts" + + +async def test_no_ack_in_the_wait_still_says_which_code_to_listen_for(monkeypatch) -> None: # noqa: ANN001 + """Not confirmed yet is not failed: the code goes back, for a late ack to settle.""" + import meshterm.core.connection as connection + + monkeypatch.setattr(connection, "trace_timeout", lambda hops: 0.05) + device = _acking_device(_AckingMeshCore(ack_at=None, flood=False)) + delivery = await device.send_direct_message(_contact(), "hi") + assert not delivery.acked and delivery.code == "c0ffee01" and delivery.flood is False + + +async def _late_ack_chat(tmp_path: Path, repo: Repository): # noqa: ANN202 + """A started chat service over a simulator whose acks for Alice arrive late.""" + device = MockDevice() + device._late_acks.add("Alice") + ctx = _StubContext(device, repo) + chat = ChatService(ctx) + await chat.start() + alice = next(c for c in await device.get_contacts() if c.name == "Alice") + return device, chat, alice + + +async def test_a_late_ack_turns_the_message_delivered(tmp_path: Path, repo: Repository) -> None: + """In history and in the very message the chat screen holds — its ✗ becomes ✓.""" + device, chat, alice = await _late_ack_chat(tmp_path, repo) + heard: list[ChatMessage] = [] + try: + sent = await chat.send_direct(alice, "you there?", on_late_ack=heard.append) + assert sent.acked is False + for _ in range(100): + if sent.acked: + break + await asyncio.sleep(0.01) + assert sent.acked is True + stored = repo.recent_chat_messages(is_channel=False, peer=alice.key_prefix) + assert stored[-1].acked is True + assert heard == [sent] + finally: + await chat.stop() + await device.disconnect() + + +async def test_a_late_ack_for_an_earlier_attempt_settles_a_retried_message( + tmp_path: Path, repo: Repository +) -> None: + """^R sends it again under a new code; the first attempt's ack still counts.""" + device, chat, alice = await _late_ack_chat(tmp_path, repo) + try: + await chat.stop() # hold the late acks back until both attempts are out + sent = await chat.send_direct(alice, "anyone?") + first = next(iter(chat._awaiting)) + await chat.resend_direct(alice, sent) + assert len(chat._awaiting) == 2 and sent.acked is False + chat._settle_ack(first) + assert sent.acked is True and not chat._awaiting + stored = repo.recent_chat_messages(is_channel=False, peer=alice.key_prefix) + assert len(stored) == 1 and stored[0].acked is True + finally: + await device.disconnect() + + +def test_only_the_acks_the_radio_could_still_push_are_awaited(repo: Repository) -> None: + """The radio keeps its last eight codes; a few more than that are kept here, no more.""" + from meshterm.services.chat_service import _AWAITING_CAP + + chat = ChatService(_StubContext(MockDevice(), repo)) + messages = [ChatMessage(text=str(n), outbound=True, acked=False) for n in range(40)] + for n, message in enumerate(messages): + chat._await_ack(message, [f"{n:08x}"], None) + assert len(chat._awaiting) == _AWAITING_CAP + chat._settle_ack("00000000") # long gone: nothing to settle + assert messages[0].acked is False + chat._settle_ack(f"{39:08x}") + assert messages[39].acked is True diff --git a/tests/test_courier.py b/tests/test_courier.py index 7eea41f..f87a2c0 100644 --- a/tests/test_courier.py +++ b/tests/test_courier.py @@ -40,11 +40,15 @@ class _StubChat: def __init__(self, outcomes: list[bool]) -> None: self.outcomes = list(outcomes) self.sent: list[tuple[str, str]] = [] + self.late: list = [] # the late-ack listener each unacked send handed over - async def send_direct(self, contact: Contact, text: str) -> ChatMessage: + async def send_direct(self, contact: Contact, text: str, *, on_late_ack=None) -> ChatMessage: # noqa: ANN001 self.sent.append((contact.name, text)) acked = self.outcomes.pop(0) if self.outcomes else False - return ChatMessage(text=text, outbound=True, acked=acked) + chat = ChatMessage(text=text, outbound=True, acked=acked) + if not acked: + self.late.append((chat, on_late_ack)) + return chat class _StubDevice: @@ -187,6 +191,26 @@ async def test_attempt_gives_up_after_the_budget(tmp_path: Path) -> None: assert len(gave) == 1 +async def test_a_late_ack_is_a_delivery_not_a_reason_to_send_again(tmp_path: Path) -> None: + """The ack outran the attempt's wait; when it lands, the entry is delivered. + + It used to stay queued, and the next attempt sent the recipient the same message again — + on a mesh where most acks from beyond a neighbour arrive late, a duplicate per message. + """ + service = _service(tmp_path, [False]) + ctx = service._ctx + entry = ctx.courier_store.queue(NODE, "Hub", "hello") + await service._attempt(ctx.courier_store.get(entry.ident)) + assert ctx.courier_store.get(entry.ident).status == QUEUED + chat, on_late_ack = ctx.chat.late[0] + on_late_ack(chat) + settled = ctx.courier_store.get(entry.ident) + assert settled.status == "delivered" and settled.attempts == 1 + assert any("ack came late" in a.message for a in ctx.watch_store.alerts()) + on_late_ack(chat) # the radio can push the same ack twice + assert len([a for a in ctx.watch_store.alerts() if "delivered" in a.message]) == 1 + + async def test_pass_attempts_at_most_one_entry(tmp_path: Path) -> None: """A backlog drains one message per pass — the courier never bursts.""" service = _service(tmp_path, [True, True]) diff --git a/tests/test_rooms.py b/tests/test_rooms.py index d39c616..e1c8512 100644 --- a/tests/test_rooms.py +++ b/tests/test_rooms.py @@ -468,7 +468,7 @@ async def test_the_simulators_room_sends_its_board_after_a_login(tmp_path: Path) ) try: heard.clear() # the stream's opening burst carries a synthetic DM of its own - assert await device.send_direct_message(ROOM, "before joining") is None + assert not (await device.send_direct_message(ROOM, "before joining")).acked login = await device.room_login(ROOM, "hello") assert login.access is RoomAccess.MEMBER while len([m for m in heard if m.is_post]) < 5: @@ -478,7 +478,7 @@ async def test_the_simulators_room_sends_its_board_after_a_login(tmp_path: Path) assert [m.sender_timestamp for m in posts] == sorted(m.sender_timestamp for m in posts) assert {m.author for m in posts} == {ALICE_KEY, "c3d4e5f6", STRANGER_KEY, "f6a7b8c9"} - assert await device.send_direct_message(ROOM, "mine") is not None # kept, acked + assert (await device.send_direct_message(ROOM, "mine")).acked # kept, acked heard.clear() await device.room_login(ROOM, "") # the room knows us; nothing new but our own await asyncio.sleep(0.2) @@ -494,7 +494,7 @@ async def test_a_read_only_member_is_heard_but_not_kept(tmp_path: Path) -> None: device._remote_config(ROOM)["allow.read.only"] = "on" login = await device.room_login(ROOM, "anything") assert login.access is RoomAccess.READ_ONLY - assert await device.send_direct_message(ROOM, "lost") is None + assert not (await device.send_direct_message(ROOM, "lost")).acked await device.disconnect()