diff --git a/work/meshai/config.py b/work/meshai/config.py index 01f1a1e..83b77ce 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -53,6 +53,12 @@ class ConnectionConfig: 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) + # --- Send-queue pacing (per-radio serialization) --- + # Minimum gap between consecutive outbound sends on each radio. + # A floor of 0.25 s is always enforced at runtime (max(0.25, pacing_value)). + meshtastic_send_pacing_seconds: float = 2.0 + meshcore_send_pacing_seconds: float = 2.0 + # --- MeshCore telemetry auto-poll settings --- # Selected contacts (names or pubkeys) to auto-poll for telemetry; empty = none. meshcore_telemetry_contacts: list = field(default_factory=list) diff --git a/work/meshai/connector.py b/work/meshai/connector.py index cbed018..4669a26 100644 --- a/work/meshai/connector.py +++ b/work/meshai/connector.py @@ -16,6 +16,7 @@ from pubsub import pub from .config import ConnectionConfig from .transport.base import MeshTransport +from .transport.send_queue import RadioSendQueue logger = logging.getLogger(__name__) @@ -64,6 +65,8 @@ class MeshtasticTransport(MeshTransport): self._wake: Optional[asyncio.Event] = None # created by main once loop exists self._link_path = "/tmp/meshai.link" self._reconnect_lock = threading.Lock() + # --- per-radio send queue (serialized + paced) --- + self._mt_queue: Optional[RadioSendQueue] = None @property def connected(self) -> bool: @@ -119,6 +122,15 @@ class MeshtasticTransport(MeshTransport): def disconnect(self) -> None: """Close connection to Meshtastic node.""" + # Stop the send queue drain and wait for it to finish so all pending + # futures are resolved before the interface is torn down. + if self._mt_queue is not None and self._mt_queue.running and self._loop is not None: + try: + asyncio.run_coroutine_threadsafe( + self._mt_queue.stop(), self._loop + ).result(timeout=5.0) + except Exception: + pass if self._interface: try: pub.unsubscribe(self._on_receive, "meshtastic.receive.text") @@ -140,7 +152,7 @@ class MeshtasticTransport(MeshTransport): def set_message_callback( self, callback: Callable[[MeshMessage], None], loop: asyncio.AbstractEventLoop ) -> None: - """Set callback for incoming messages. + """Set callback for incoming messages and start the send queue drain. Args: callback: Async function to call with MeshMessage @@ -148,6 +160,14 @@ class MeshtasticTransport(MeshTransport): """ self._message_callback = callback self._loop = loop + # Start the per-radio send queue on the main event loop. + pacing_fn = lambda: max(0.25, getattr(self.config, "meshtastic_send_pacing_seconds", 2.0)) + self._mt_queue = RadioSendQueue(pacing_fn=pacing_fn) + + def _start_drain() -> None: + self._mt_queue.start(loop) + + loop.call_soon_threadsafe(_start_drain) def _cache_node_info(self) -> None: """Cache node names and positions from node database.""" @@ -349,6 +369,68 @@ class MeshtasticTransport(MeshTransport): time.sleep(backoff) backoff = min(backoff * 2, maxd) + async def send_message_async( + self, + text: str, + destination: Optional[str] = None, + channel: int = 0, + transport: Optional[str] = None, + meshcore_channel: Optional[str] = None, + ) -> bool: + """Async send through the Meshtastic per-radio queue. + + Enqueues the send job and awaits the ACTUAL radio-acceptance result so + the caller (e.g. channel.deliver) gets the real success bool. Falls + back to a direct executor call if the queue has not been started yet + (e.g. during tests or before set_message_callback is called). + """ + if self._mt_queue is None or not self._mt_queue.running: + # Queue not started: run in executor to avoid blocking the loop. + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, + lambda: self.send_message(text, destination, channel, transport, meshcore_channel), + ) + + def _job() -> bool: + return self._blocking_mt_send(text, destination, channel) + + async def _async_job() -> bool: + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, _job) + + return await self._mt_queue.enqueue_async(_async_job) + + def _blocking_mt_send( + self, + text: str, + destination: Optional[str], + channel: int, + ) -> bool: + """Synchronous Meshtastic send — called from the thread executor by the drain.""" + if not self._interface: + logger.error("Cannot send: not connected") + return False + try: + if destination: + if isinstance(destination, int): + dest_num = destination + elif destination.startswith("!"): + dest_num = int(destination[1:], 16) + elif destination.isdigit(): + dest_num = int(destination) + else: + dest_num = int(destination, 16) + self._interface.sendText(text=text, destinationId=dest_num, channelIndex=channel) + else: + from meshtastic import BROADCAST_NUM + self._interface.sendText(text=text, destinationId=BROADCAST_NUM, channelIndex=channel) + logger.debug("MT send to %s: %s…", destination or "broadcast", text[:50]) + return True + except Exception as exc: + logger.error("MT send failed: %s", exc) + return False + def send_message( self, text: str, diff --git a/work/meshai/dashboard/api/mesh_send_routes.py b/work/meshai/dashboard/api/mesh_send_routes.py index aa7a34e..73e234b 100644 --- a/work/meshai/dashboard/api/mesh_send_routes.py +++ b/work/meshai/dashboard/api/mesh_send_routes.py @@ -78,7 +78,11 @@ async def meshcore_send_advert(request: Request): if mc is None or not getattr(mc, "connected", False): return {"sent": False, "detail": "MeshCore not connected"} try: - ok = bool(mc.send_advert()) + send_fn = getattr(mc, "send_advert_async", None) + if send_fn is not None: + ok = bool(await send_fn()) + else: + ok = bool(mc.send_advert()) detail = "Self-advert sent" if ok else "send_advert returned False" logger.info("dashboard: meshcore manual advert sent=%s", ok) return {"sent": ok, "detail": detail} @@ -124,7 +128,11 @@ async def meshcore_telemetry_poll(request: Request): if not contact: return {"available": False, "detail": "Missing 'contact'"} try: - data = mc.req_telemetry(contact) + poll_fn = getattr(mc, "req_telemetry_async", None) + if poll_fn is not None: + data = await poll_fn(contact) + else: + data = mc.req_telemetry(contact) if data is None: return {"available": False, "contact": contact, "detail": "No telemetry response"} return {"available": True, "contact": contact, "data": data} @@ -155,7 +163,7 @@ async def test_send(request: Request, body: TestSendRequest): except (ValueError, TypeError): result = {"sent": False, "detail": f"invalid meshtastic channel index: {body.channel!r}"} else: - ok = bool(connector.send_message(text, destination=None, channel=idx, transport="meshtastic")) + ok = bool(await connector.send_message_async(text, destination=None, channel=idx, transport="meshtastic")) result = {"sent": ok, "detail": f"sent to meshtastic channel {idx}" if ok else "send returned False"} elif body.transport == "meshcore": child = _find_child(connector, "meshcore") @@ -163,7 +171,7 @@ async def test_send(request: Request, body: TestSendRequest): result = {"sent": False, "detail": "meshcore not connected"} else: name = str(body.channel) - ok = bool(connector.send_message(text, destination=None, meshcore_channel=name, transport="meshcore")) + ok = bool(await connector.send_message_async(text, destination=None, meshcore_channel=name, transport="meshcore")) if ok: result = {"sent": True, "detail": f"sent to '{name}'"} else: diff --git a/work/meshai/notifications/channels.py b/work/meshai/notifications/channels.py index 67d441b..5180c92 100644 --- a/work/meshai/notifications/channels.py +++ b/work/meshai/notifications/channels.py @@ -80,26 +80,29 @@ class MeshBroadcastChannel(NotificationChannel): try: # If payload already has chunk metadata (from digest), use message directly if alert.chunk_index is not None: - self._connector.send_message( + ok = await self._connector.send_message_async( text=alert.message or "", destination=None, channel=self._channel, transport=self._transport, ) - logger.info("Broadcast pre-chunked alert to channel %d", self._channel) - return True + logger.info("Broadcast pre-chunked alert to channel %d (ok=%s)", self._channel, ok) + return bool(ok) # Render to chunks for single-event delivery chunks = self._renderer.render(alert) + success = True for chunk in chunks: - self._connector.send_message( + ok = await self._connector.send_message_async( text=chunk, destination=None, channel=self._channel, transport=self._transport, ) - logger.info("Broadcast %d chunk(s) to channel %d", len(chunks), self._channel) - return True + if not ok: + success = False + logger.info("Broadcast %d chunk(s) to channel %d (success=%s)", len(chunks), self._channel, success) + return success except Exception as e: logger.error("Failed to broadcast alert: %s", e) return False @@ -134,15 +137,15 @@ class MeshBroadcastChannel(NotificationChannel): if hasattr(ch, 'settings') and hasattr(ch.settings, 'name'): channel_name = ch.settings.name or f'Channel {self._channel}' - # Send actual test message - self._connector.send_message( + # Send actual test message (async path through queue) + ok = await self._connector.send_message_async( text="MeshAI channel test - if you see this, delivery works", destination=None, channel=self._channel, ) return { - "success": True, + "success": ok, "message": f"Sent to channel {self._channel}: {channel_name}", "error": "", "details": { @@ -164,12 +167,12 @@ class MeshBroadcastChannel(NotificationChannel): return False, "Not connected to radio" try: - self._connector.send_message( + ok = bool(await self._connector.send_message_async( text=message, destination=None, channel=self._channel, - ) - return True, f"Sent to mesh channel {self._channel}" + )) + return ok, f"Sent to mesh channel {self._channel}" if ok else "Mesh broadcast returned False" except Exception as e: return False, f"Mesh broadcast failed: {e}" @@ -214,32 +217,35 @@ class MeshCoreBroadcastChannel(NotificationChannel): try: # If payload already has chunk metadata (from digest), use message directly if alert.chunk_index is not None: - self._connector.send_message( + ok = await self._connector.send_message_async( text=alert.message or "", destination=None, meshcore_channel=self._meshcore_channel, transport="meshcore", ) logger.info( - "MeshCore broadcast pre-chunked alert to channel %r", - self._meshcore_channel, + "MeshCore broadcast pre-chunked alert to channel %r (ok=%s)", + self._meshcore_channel, ok, ) - return True + return bool(ok) # Render to chunks for single-event delivery chunks = self._renderer.render(alert) + success = True for chunk in chunks: - self._connector.send_message( + ok = await self._connector.send_message_async( text=chunk, destination=None, meshcore_channel=self._meshcore_channel, transport="meshcore", ) + if not ok: + success = False logger.info( - "MeshCore broadcast %d chunk(s) to channel %r", - len(chunks), self._meshcore_channel, + "MeshCore broadcast %d chunk(s) to channel %r (success=%s)", + len(chunks), self._meshcore_channel, success, ) - return True + return success except Exception as e: logger.error("Failed to MeshCore broadcast alert: %s", e) return False @@ -267,13 +273,13 @@ class MeshCoreBroadcastChannel(NotificationChannel): if not self._meshcore_channel: return False, "No MeshCore channel configured" try: - self._connector.send_message( + ok = bool(await self._connector.send_message_async( text=message, destination=None, meshcore_channel=self._meshcore_channel, transport="meshcore", - ) - return True, f"Sent to MeshCore channel {self._meshcore_channel!r}" + )) + return ok, f"Sent to MeshCore channel {self._meshcore_channel!r}" if ok else "MeshCore broadcast returned False" except Exception as e: return False, f"MeshCore broadcast failed: {e}" @@ -310,12 +316,14 @@ class MeshDMChannel(NotificationChannel): for message in messages: try: node_id = str(node_id) - self._connector.send_message( + ok = await self._connector.send_message_async( text=message, destination=node_id, channel=0, transport=self._transport_hint, ) + if not ok: + success = False except Exception as e: logger.error("Failed to DM %s: %s", node_id, e) success = False @@ -346,12 +354,14 @@ class MeshDMChannel(NotificationChannel): for node_id in self._node_ids: try: node_id = str(node_id) - self._connector.send_message( + ok = await self._connector.send_message_async( text="MeshAI DM test", destination=node_id, channel=0, ) - results.append({"node": node_id, "success": True}) + results.append({"node": node_id, "success": ok}) + if not ok: + all_success = False except Exception as e: results.append({"node": node_id, "success": False, "error": str(e)}) all_success = False @@ -396,8 +406,11 @@ class MeshDMChannel(NotificationChannel): for node_id in self._node_ids: try: node_id = str(node_id) - self._connector.send_message(text=message, destination=node_id, channel=0) - success_count += 1 + ok = await self._connector.send_message_async(text=message, destination=node_id, channel=0) + if ok: + success_count += 1 + else: + errors.append(f"{node_id}: send returned False") except Exception as e: errors.append(f"{node_id}: {e}") @@ -452,11 +465,13 @@ class MeshCoreDMChannel(NotificationChannel): for contact in self._contacts: for message in messages: try: - self._connector.send_message( + ok = await self._connector.send_message_async( text=message, destination=str(contact), transport="meshcore", ) + if not ok: + success = False except Exception as e: logger.error("Failed to MeshCore DM %s: %s", contact, e) success = False @@ -496,12 +511,15 @@ class MeshCoreDMChannel(NotificationChannel): errors = [] for contact in self._contacts: try: - self._connector.send_message( + ok = await self._connector.send_message_async( text=message, destination=str(contact), transport="meshcore", ) - success_count += 1 + if ok: + success_count += 1 + else: + errors.append(f"{contact}: send returned False") except Exception as e: errors.append(f"{contact}: {e}") if success_count == len(self._contacts): diff --git a/work/meshai/responder.py b/work/meshai/responder.py index e9a3b38..b616983 100644 --- a/work/meshai/responder.py +++ b/work/meshai/responder.py @@ -51,7 +51,7 @@ class Responder: delay = random.uniform(self.config.delay_min, self.config.delay_max) await asyncio.sleep(delay) - sent = self.connector.send_message( + sent = await self.connector.send_message_async( text=msg, destination=destination, channel=channel, diff --git a/work/meshai/transport/base.py b/work/meshai/transport/base.py index 2d69d07..9b32a3f 100644 --- a/work/meshai/transport/base.py +++ b/work/meshai/transport/base.py @@ -3,6 +3,7 @@ import abc import asyncio from typing import Callable, Optional +import concurrent.futures class MeshTransport(abc.ABC): @@ -59,6 +60,30 @@ class MeshTransport(abc.ABC): True if send was initiated successfully. """ + async def send_message_async( + self, + text: str, + destination: Optional[str] = None, + channel: int = 0, + transport: Optional[str] = None, + meshcore_channel: Optional[str] = None, + ) -> bool: + """Async send through the per-radio serialized queue. + + Default implementation runs ``send_message()`` in the default thread + executor so it never blocks the event loop. Concrete subclasses with + a send queue override this to enqueue the job and await the actual + radio result. + + Returns: + True if the send was accepted by the radio. + """ + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, + lambda: self.send_message(text, destination, channel, transport, meshcore_channel), + ) + @abc.abstractmethod def set_message_callback( self, diff --git a/work/meshai/transport/composite_transport.py b/work/meshai/transport/composite_transport.py index ab5b64f..7e2b73e 100644 --- a/work/meshai/transport/composite_transport.py +++ b/work/meshai/transport/composite_transport.py @@ -107,11 +107,35 @@ class CompositeTransport(MeshTransport): child = self.meshcore_child() return child.send_advert() if child is not None else False + async def send_advert_async(self) -> bool: + """Async passthrough to the MeshCore child's send_advert_async(); False if no child.""" + child = self.meshcore_child() + if child is None: + return False + fn = getattr(child, "send_advert_async", None) + if fn is not None: + return await fn() + # Fallback: sync version in executor. + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, child.send_advert) + def req_telemetry(self, contact_id): """Passthrough to the MeshCore child's on-demand telemetry poll; None if no child.""" child = self.meshcore_child() return child.req_telemetry(contact_id) if child is not None else None + async def req_telemetry_async(self, contact_id: str): + """Async passthrough to the MeshCore child's req_telemetry_async(); None if no child.""" + child = self.meshcore_child() + if child is None: + return None + fn = getattr(child, "req_telemetry_async", None) + if fn is not None: + return await fn(contact_id) + # Fallback: sync version in executor. + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, lambda: child.req_telemetry(contact_id)) + def get_telemetry_cache(self): """Passthrough to the MeshCore child's telemetry cache; [] if no meshcore child.""" child = self.meshcore_child() @@ -223,6 +247,124 @@ class CompositeTransport(MeshTransport): # Message I/O # ------------------------------------------------------------------ + async def send_message_async( + self, + text: str, + destination: Optional[str] = None, + channel: int = 0, + transport: Optional[str] = None, + meshcore_channel: Optional[str] = None, + ) -> bool: + """Async send through per-child queues, with the same routing logic as send_message(). + + Each child's send_message_async() goes through that child's serialized queue, + so Meshtastic and MeshCore sends are each independently paced. For broadcasts + the two are awaited sequentially (Meshtastic first, then MeshCore). + """ + if destination is None: + return await self._broadcast_async(text, channel, meshcore_channel=meshcore_channel, + transport=transport) + if transport is not None: + return await self._send_hinted_async(text, destination, channel, transport) + return await self._send_unhinted_async(text, destination, channel) + + async def _broadcast_async( + self, text: str, channel: int, + meshcore_channel: Optional[str] = None, + transport: Optional[str] = None, + ) -> bool: + if transport is not None: + child = self._by_name.get(transport) + if child is None: + return False + name = _child_name(child) + if not child.connected: + return False + try: + if name == "meshcore": + return await child.send_message_async( + text, destination=None, meshcore_channel=meshcore_channel + ) + else: + return await child.send_message_async(text, destination=None, channel=channel) + except Exception as exc: + logger.error("CompositeTransport: async hinted broadcast via %r raised: %s", name, exc) + return False + + any_ok = False + for child in self._children: + name = _child_name(child) + if not child.connected: + continue + if name == "meshcore" and meshcore_channel is None: + continue + child_channel = meshcore_channel if name == "meshcore" else channel + try: + if name == "meshcore": + ok = await child.send_message_async( + text, destination=None, meshcore_channel=child_channel + ) + else: + ok = await child.send_message_async(text, destination=None, channel=child_channel) + if ok: + any_ok = True + else: + logger.warning("CompositeTransport: async broadcast via %r returned False", name) + except Exception as exc: + logger.error("CompositeTransport: async broadcast via %r raised: %s", name, exc) + return any_ok + + async def _send_hinted_async( + self, text: str, destination: str, channel: int, transport_hint: str + ) -> bool: + child = self._resolve_child_for_hint(transport_hint) + if child is None: + logger.error( + "CompositeTransport: no child with name %r for async DM; known: %s", + transport_hint, list(self._by_name), + ) + return False + if not child.connected: + return False + try: + return await child.send_message_async(text, destination=destination, channel=channel) + except Exception as exc: + logger.error( + "CompositeTransport: async hinted send via %r raised: %s", + _child_name(child), exc, + ) + return False + + async def _send_unhinted_async(self, text: str, destination: str, channel: int) -> bool: + child = self._best_child_for_destination(destination) + if child is not None: + try: + return await child.send_message_async(text, destination=destination, channel=channel) + except Exception as exc: + logger.error( + "CompositeTransport: async unhinted send via %r raised: %s", + _child_name(child), exc, + ) + return False + logger.warning( + "CompositeTransport: async DM destination %r unresolved; fanning to all children", + destination, + ) + any_ok = False + for child in self._children: + if not child.connected: + continue + try: + ok = await child.send_message_async(text, destination=destination, channel=channel) + if ok: + any_ok = True + except Exception as exc: + logger.error( + "CompositeTransport: async DM fan via %r raised: %s", + _child_name(child), exc, + ) + return any_ok + def send_message( self, text: str, diff --git a/work/meshai/transport/meshcore_transport.py b/work/meshai/transport/meshcore_transport.py index bf7a3aa..686ef19 100644 --- a/work/meshai/transport/meshcore_transport.py +++ b/work/meshai/transport/meshcore_transport.py @@ -11,12 +11,14 @@ imported (and the test suite can run) without the lib installed. """ import asyncio +import concurrent.futures import logging import threading import time as _time from typing import Callable, Optional from .base import MeshTransport +from .send_queue import RadioSendQueue from ..connector import MeshMessage logger = logging.getLogger(__name__) @@ -125,6 +127,9 @@ class MeshCoreTransport(MeshTransport): # _telemetry_failures: contact-id -> consecutive-timeout count self._telemetry_cache: dict[str, dict] = {} self._telemetry_failures: dict[str, int] = {} + # --- per-radio send queue (on the MC dedicated loop) --- + self._mc_send_queue: Optional[asyncio.Queue] = None + self._mc_drain_task: Optional[asyncio.Task] = None # ------------------------------------------------------------------ # Internal helpers @@ -324,6 +329,383 @@ class MeshCoreTransport(MeshTransport): ) return False + # ------------------------------------------------------------------ + # Async send helpers — run ON the MC dedicated loop (no _run_coro). + # These are called from the MC drain task directly. + # ------------------------------------------------------------------ + + async def _resolve_contact_async(self, dest: str): + """Resolve contact on the MC loop (no _run_coro, no deadlock risk).""" + if self._mc is None: + return None + try: + contact = self._mc.get_contact_by_key_prefix(dest) + if contact is not None: + return contact + except Exception: + return None + # Cache miss: force a full re-fetch from firmware. + try: + await asyncio.wait_for(self._mc.commands.get_contacts(lastmod=0), timeout=15) + except Exception: + logger.debug("MC: _resolve_contact_async get_contacts failed for %s", dest, exc_info=True) + try: + return self._mc.get_contact_by_key_prefix(dest) + except Exception: + return None + + async def _send_dm_once_async(self, contact: dict, text: str, destination: str): + """Send one DM frame on the MC loop; returns MSG_SENT event or None.""" + try: + result = await asyncio.wait_for( + self._mc.commands.send_msg(contact, text), timeout=15 + ) + except Exception as exc: + logger.warning("MC: _send_dm_once_async to %s failed: %s", destination, exc) + return None + if result is None: + logger.warning("MC: _send_dm_once_async to %s — no send result", destination) + return None + try: + sent_type = ( + result.payload.get("type") + if hasattr(result, "payload") and isinstance(result.payload, dict) + else None + ) + logger.info( + "MC: 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 + + async def _wait_for_ack_async(self, expected_ack, timeout: float) -> bool: + """Wait for delivery ACK on the MC loop.""" + if not expected_ack: + return False + from meshcore import EventType # noqa: PLC0415 + try: + code = ( + expected_ack.hex() + if isinstance(expected_ack, (bytes, bytearray)) + else str(expected_ack) + ) + except Exception: + return False + try: + event = await asyncio.wait_for( + 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("MC: _wait_for_ack_async failed for %r", expected_ack, exc_info=True) + return False + + async def _establish_direct_path_async(self, contact: dict, dst: str) -> None: + """Path discovery on the MC loop (async version of _establish_direct_path).""" + label = contact.get("adv_name") or dst + try: + path_event = await asyncio.wait_for( + 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("MC: path discovery to %s succeeded", label) + else: + logger.info("MC: path discovery to %s no PATH_RESPONSE; trying advert-path fallback", label) + except Exception as exc: + logger.debug("MC: send_path_discovery_sync to %s failed: %s", dst, exc) + + try: + fresh_contact = await self._resolve_contact_async(dst) or contact + if fresh_contact.get("out_path_len", -1) < 0: + try: + ap_event = await asyncio.wait_for( + self._mc.commands.get_advert_path(fresh_contact), timeout=10 + ) + if ( + ap_event is not None + and not ap_event.is_error() + and isinstance(getattr(ap_event, "payload", None), dict) + ): + path_hex = ap_event.payload.get("path", "") + path_len = ap_event.payload.get("path_len", -1) + if path_len > 0 and path_hex: + await asyncio.wait_for( + self._mc.commands.update_contact(fresh_contact, path=path_hex), + timeout=10, + ) + logger.info( + "MC: seeded direct path for %s (len=%d)", + fresh_contact.get("adv_name") or dst, path_len, + ) + except Exception as exc: + logger.debug("MC: advert-path fallback for %s failed: %s", dst, exc) + except Exception as exc: + logger.debug("MC: _establish_direct_path_async step-2 for %s: %s", dst, exc) + + async def _enumerate_channels_async(self) -> None: + """Channel enumeration on the MC loop (async version of _enumerate_channels).""" + self._chan_name_to_idx = {} + try: + empty_run = 0 + for idx in range(40): + try: + event = await asyncio.wait_for( + self._mc.commands.get_channel(idx), timeout=5 + ) + except Exception as exc: + logger.debug("MC: get_channel(%d) async failed, ending: %s", idx, exc) + break + if not event: + break + is_err = getattr(event, "is_error", None) + if callable(is_err) and event.is_error(): + break + payload = event.payload or {} + name = payload.get("channel_name", "") + slot = payload.get("channel_idx", idx) + if not name: + empty_run += 1 + if empty_run >= 3: + break + continue + self._chan_name_to_idx[name] = slot + empty_run = 0 + except Exception as exc: + logger.warning("MC: async channel enumeration error: %s", exc) + self._chan_name_to_idx = {} + logger.info("MC: async enumerated %d named channel(s)", len(self._chan_name_to_idx)) + + async def _do_mc_dm_send_async(self, text: str, destination: str) -> bool: + """Full DM send path on the MC loop (replaces send_message() DM branch).""" + if self._mc is None: + return False + contact = await self._resolve_contact_async(destination) + if contact is None: + logger.warning("MC: could not resolve contact %s; cannot DM", destination) + return False + label = contact.get("adv_name") or destination + + # Fast path: send and wait for ACK. + result = await self._send_dm_once_async(contact, text, destination) + if result is None: + return False + exp_ack = self._extract_expected_ack(result) + if await self._wait_for_ack_async(exp_ack, self._ack_wait): + logger.info("MC: DM to %s ACKed (direct)", label) + return True + + # No ACK: path discovery + retry. + logger.info("MC: no ACK from %s in %.1fs; running path discovery", label, self._ack_wait) + await self._establish_direct_path_async(contact, destination) + contact = await self._resolve_contact_async(destination) or contact + result = await self._send_dm_once_async(contact, text, destination) + if result is None: + return False + exp_ack = self._extract_expected_ack(result) + acked = await self._wait_for_ack_async(exp_ack, self._ack_wait) + logger.info( + "MC: DM to %s %s after discovery", + contact.get("adv_name") or destination, + "ACKed" if acked else "no ACK (sent best-effort)", + ) + return acked or (not result.is_error()) + + async def _do_mc_broadcast_async(self, text: str, meshcore_channel: str) -> bool: + """Channel broadcast on the MC loop (replaces send_message() broadcast branch).""" + if self._mc is None: + return False + idx = self._chan_name_to_idx.get(meshcore_channel) + if idx is None: + # Lazy async re-enumeration (no _run_coro deadlock risk). + await self._enumerate_channels_async() + idx = self._chan_name_to_idx.get(meshcore_channel) + if idx is None: + logger.warning("MC channel '%s' not on companion; skipping", meshcore_channel) + return False + try: + result = await asyncio.wait_for( + self._mc.commands.send_chan_msg(idx, text), timeout=10 + ) + success = not result.is_error() + if not success: + logger.warning("MC: broadcast returned error event") + return success + except Exception as exc: + logger.error("MC: _do_mc_broadcast_async failed: %s", exc) + return False + + async def _do_mc_advert_async(self) -> bool: + """Send self-advert on the MC loop (queued version).""" + if self._mc is None or not self._connected: + return False + try: + await self._mc.commands.send_advert(flood=True) + self._last_advert_sent = _time.time() + return True + except Exception as exc: + logger.warning("MC: queued advert failed: %s", exc) + return False + + # ------------------------------------------------------------------ + # MC send queue management + # ------------------------------------------------------------------ + + async def _mc_drain_loop(self) -> None: + """Drain loop for the MC send queue — runs on the MC dedicated loop.""" + while True: + item = await self._mc_send_queue.get() + async_fn, cfut = item + result = False + try: + result = await async_fn() + except asyncio.CancelledError: + if cfut is not None and not cfut.done(): + cfut.cancel() + raise + except Exception as exc: + logger.error("MC queue drain: send job raised: %s", exc, exc_info=True) + if cfut is not None and not cfut.done(): + cfut.set_exception(exc) + else: + if cfut is not None and not cfut.done(): + cfut.set_result(result) + + # Pace: read live from config. + pacing = max(0.25, getattr(self.config, "meshcore_send_pacing_seconds", 2.0)) + await asyncio.sleep(pacing) + + def _start_mc_queue(self) -> None: + """Arm the MC send queue + drain task on the MC dedicated loop. + + Call from within the MC dedicated loop (e.g. via + ``loop.call_soon_threadsafe`` from the main thread after connect). + + On reconnect (called a second time), the old drain task is cancelled + and all futures still sitting in the old queue are cancelled so no + caller hangs indefinitely awaiting a result from a pre-reconnect send. + """ + # Reconnect path: cancel the old drain and resolve outstanding futures. + old_task = self._mc_drain_task + old_queue = self._mc_send_queue + if old_task is not None: + old_task.cancel() + if old_queue is not None: + while True: + try: + _, cfut = old_queue.get_nowait() + if cfut is not None and not cfut.done(): + cfut.cancel() + except asyncio.QueueEmpty: + break + self._mc_send_queue = asyncio.Queue() + self._mc_drain_task = asyncio.get_event_loop().create_task( + self._mc_drain_loop(), name="mc-send-queue-drain" + ) + logger.debug("MC: send queue drain task started") + + def _cancel_mc_queue(self) -> None: + """Cancel the MC drain task and resolve all pending futures (thread-safe). + + Called at disconnect. Schedules a teardown callback on the MC loop that: + 1. Cancels the drain task. + 2. Drains the remaining queue with ``get_nowait()`` and cancels every + pending ``concurrent.futures.Future`` so no caller hangs at + ``await asyncio.wrap_future(cfut)`` after teardown. + + The in-flight future (if the drain was mid-job when cancel fires) is + handled by the drain's own ``except CancelledError`` handler. + """ + task = self._mc_drain_task + queue = self._mc_send_queue + self._mc_drain_task = None + + def _teardown() -> None: + if task is not None: + task.cancel() + if queue is not None: + while True: + try: + _, cfut = queue.get_nowait() + if cfut is not None and not cfut.done(): + cfut.cancel() + except asyncio.QueueEmpty: + break + + if self._loop is not None and self._loop.is_running(): + self._loop.call_soon_threadsafe(_teardown) + + async def send_message_async( + self, + text: str, + destination: Optional[str] = None, + channel: int = 0, + transport: Optional[str] = None, + meshcore_channel: Optional[str] = None, + ) -> bool: + """Async send through the MC per-radio queue (called from the main loop). + + Enqueues the job on the MC loop's queue and awaits the actual send + result via a concurrent.futures.Future bridge. + """ + if self._mc is None or not self._connected: + return False + if self._mc_send_queue is None or self._loop is None or not self._loop.is_running(): + # Queue not started yet (e.g. initial advert at connect) — fall back. + loop = asyncio.get_event_loop() + return await loop.run_in_executor( + None, + lambda: self.send_message(text, destination, channel, transport, meshcore_channel), + ) + + cfut: concurrent.futures.Future = concurrent.futures.Future() + + if destination: + async def _job() -> bool: + return await self._do_mc_dm_send_async(text, destination) + else: + if meshcore_channel is None: + logger.debug("MC: send_message_async meshcore_channel=None, skipping broadcast") + return False # nothing sent — do not report success + async def _job() -> bool: + return await self._do_mc_broadcast_async(text, meshcore_channel) + + # Enqueue on MC loop from main loop. + asyncio.run_coroutine_threadsafe( + self._mc_send_queue.put((_job, cfut)), + self._loop, + ) + + main_loop = asyncio.get_event_loop() + return await asyncio.wrap_future(cfut, loop=main_loop) + + def _enqueue_mc_fire_forget(self, async_fn) -> None: + """Enqueue fire-and-forget from any thread (for advert/telemetry loops).""" + if self._mc_send_queue is None or self._loop is None or not self._loop.is_running(): + logger.debug("MC: fire-and-forget dropped (queue not started)") + return + asyncio.run_coroutine_threadsafe( + self._mc_send_queue.put((async_fn, None)), + self._loop, + ) + + async def _enqueue_mc_loop_send(self, async_fn) -> bool: + """Enqueue from within the MC loop and await the result (for advert/telemetry loops).""" + if self._mc_send_queue is None: + # Queue not started; fall through to direct call. + return await async_fn() + cfut: concurrent.futures.Future = concurrent.futures.Future() + await self._mc_send_queue.put((async_fn, cfut)) + return await asyncio.wrap_future(cfut) + def _send_dm_once(self, contact: dict, text: str, destination: str): """Send one DM frame (no discovery, no retry) and log its route. @@ -489,13 +871,33 @@ class MeshCoreTransport(MeshTransport): logger.warning("MeshCore: send_advert failed: %s", exc) return False + async def send_advert_async(self) -> bool: + """Async self-advertisement through the MC per-radio queue (main-loop caller). + + Mirrors ``send_message_async`` — enqueues ``_do_mc_advert_async`` on the + MC loop queue from the main loop, awaits the result via a + concurrent.futures.Future bridge. Falls back to a thread-executor + wrapping the synchronous ``send_advert`` when the queue isn't running. + """ + if self._mc is None or not self._connected: + return False + if self._mc_send_queue is None or self._loop is None or not self._loop.is_running(): + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, self.send_advert) + + cfut: concurrent.futures.Future = concurrent.futures.Future() + asyncio.run_coroutine_threadsafe( + self._mc_send_queue.put((self._do_mc_advert_async, cfut)), + self._loop, + ) + main_loop = asyncio.get_event_loop() + return await asyncio.wrap_future(cfut, loop=main_loop) + async def _periodic_advert_loop(self, interval: int) -> None: """Periodic self-advertisement coroutine (runs as a Task on the dedicated loop). - Sleeps *interval* seconds, sends one flood advert, repeats. Stops on - CancelledError (raised by ``_cancel_periodic_advert`` at disconnect) or - when the transport drops its connection. No overlap is possible because - the loop awaits the sleep before each send. + Sleeps *interval* seconds, sends one flood advert via the send queue + (serialized with other MC sends), repeats. Stops on CancelledError. """ try: while True: @@ -503,13 +905,15 @@ class MeshCoreTransport(MeshTransport): if not self._connected or self._mc is None: return try: - await self._mc.commands.send_advert(flood=True) - self._last_advert_sent = _time.time() - logger.info("MeshCore: sent periodic self-advert") + ok = await self._enqueue_mc_loop_send(self._do_mc_advert_async) + if ok: + logger.info("MC: sent periodic self-advert") + else: + logger.warning("MC: periodic send_advert returned False") except Exception as exc: - logger.warning("MeshCore: periodic send_advert failed: %s", exc) + logger.warning("MC: periodic send_advert failed: %s", exc) except asyncio.CancelledError: - logger.debug("MeshCore: periodic advert task cancelled") + logger.debug("MC: periodic advert task cancelled") raise def _schedule_periodic_advert(self, interval: int) -> None: @@ -531,6 +935,8 @@ class MeshCoreTransport(MeshTransport): self._advert_task = None if task is not None and self._loop is not None and self._loop.is_running(): self._loop.call_soon_threadsafe(task.cancel) + # Also cancel the MC send queue drain. + self._cancel_mc_queue() # ------------------------------------------------------------------ # Telemetry (MeshCore sensor auto-poll) @@ -612,11 +1018,19 @@ class MeshCoreTransport(MeshTransport): self._telemetry_cache[contact_id] = entry async def _req_telemetry_async(self, contact_id): - """Resolve, request+await, decode telemetry for *contact_id* (on the loop). + """Resolve, request+await, decode telemetry for *contact_id* on the MC loop. + + Runs INLINE on the dedicated event loop — intentionally does NOT + re-enqueue on the send queue. Callers that need queue serialization + with other MC sends (``req_telemetry_async`` for on-demand polls, + ``_telemetry_poll_loop`` for periodic polls) wrap their call to this + method inside a queue job via ``_enqueue_mc_loop_send`` or the + cfut-bridge in ``req_telemetry_async``. + + Never calling ``_enqueue_mc_loop_send`` here is critical: the drain + task is single-threaded, so a nested enqueue + await from inside a + drain job would deadlock the queue permanently. - Runs entirely on the dedicated event loop so the poller (already on that - loop) can await it directly WITHOUT a nested _run_coro deadlock. Updates - the shared cache/failure bookkeeping so poller and on-demand paths agree. Returns the decoded dict, or None on unresolved/timeout/no-response/error. """ contact = self._resolve_contact(contact_id) @@ -625,7 +1039,7 @@ class MeshCoreTransport(MeshTransport): try: lpp = await self._mc.commands.req_telemetry_sync(contact, min_timeout=5) except Exception as exc: - logger.warning("MeshCore: req_telemetry(%s) error: %s", contact_id, exc) + logger.warning("MC: req_telemetry(%s) error: %s", contact_id, exc) self._record_telemetry_result(contact_id, None) return None if lpp is None: @@ -653,6 +1067,41 @@ class MeshCoreTransport(MeshTransport): logger.warning("MeshCore: req_telemetry(%s) failed: %s", contact_id, exc) return None + async def req_telemetry_async(self, contact_id: str): + """Async on-demand telemetry poll (main-loop caller, routes through MC queue). + + Mirrors ``req_telemetry`` but awaitable from the main asyncio event loop. + Enqueues the poll job on the MC loop queue via a concurrent.futures.Future + bridge, then returns the cache entry's data. Falls back to run_in_executor + wrapping the synchronous ``req_telemetry`` when the queue isn't running. + """ + if self._mc is None or not self._connected: + return None + if self._mc_send_queue is None or self._loop is None or not self._loop.is_running(): + loop = asyncio.get_event_loop() + return await loop.run_in_executor(None, lambda: self.req_telemetry(contact_id)) + + cfut: concurrent.futures.Future = concurrent.futures.Future() + + async def _telem_job_outer() -> bool: + """Bridge: runs _req_telemetry_async on the MC loop; caches result.""" + try: + data = await self._req_telemetry_async(contact_id) + except Exception as exc: # noqa: BLE001 + logger.warning("MC: req_telemetry_async(%s) failed: %s", contact_id, exc) + return False + return data is not None + + asyncio.run_coroutine_threadsafe( + self._mc_send_queue.put((_telem_job_outer, cfut)), + self._loop, + ) + main_loop = asyncio.get_event_loop() + await asyncio.wrap_future(cfut, loop=main_loop) + # Result is in the telemetry cache after the job completes. + entry = self._telemetry_cache.get(contact_id) + return entry.get("data") if entry else None + def get_telemetry_cache(self) -> list[dict]: """Return the current telemetry cache entries (list of dicts).""" return list(self._telemetry_cache.values()) @@ -700,8 +1149,18 @@ class MeshCoreTransport(MeshTransport): if self._telemetry_failures.get(c, 0) >= _TELEMETRY_MAX_FAILURES: continue try: - data = await self._req_telemetry_async(c) - if data is not None: + # Wrap in a queue job so this poll is serialized with + # other MC sends (broadcasts, adverts). _req_telemetry_async + # is inline — it must NOT be called bare here or it would + # bypass the send queue and race with concurrent sends. + _c = c # capture for closure + + async def _poll_job(_cid=_c) -> bool: + data = await self._req_telemetry_async(_cid) + return data is not None + + ok = await self._enqueue_mc_loop_send(_poll_job) + if ok: logger.info("MeshCore: telemetry polled %s", c) else: logger.debug("MeshCore: telemetry miss for %s", c) @@ -794,7 +1253,9 @@ class MeshCoreTransport(MeshTransport): except Exception as exc: # older firmware may not support CMD 58 logger.warning("MeshCore: set_autoadd_config not supported/failed (non-fatal): %s", exc) await self._mc.start_auto_message_fetching() - logger.info("MeshCore: subscriptions registered; auto message-fetch started") + # Arm the MC send queue on this dedicated loop (runs after connect). + self._start_mc_queue() + logger.info("MeshCore: subscriptions registered; auto message-fetch started; send queue armed") async def _do_disconnect(self) -> None: """Stop fetching and close the meshcore connection.""" diff --git a/work/meshai/transport/send_queue.py b/work/meshai/transport/send_queue.py new file mode 100644 index 0000000..f2ec33b --- /dev/null +++ b/work/meshai/transport/send_queue.py @@ -0,0 +1,183 @@ +"""Per-radio serialized, paced send queue. + +Each radio (Meshtastic, MeshCore) gets its own RadioSendQueue instance. +The drain task pops one job at a time, executes the async send, resolves the +caller's Future with the actual bool result, then sleeps pacing_seconds before +popping the next job — enforcing strict FIFO serialization with configurable +inter-packet spacing. + +Design notes +------------ +* Jobs are ``(async_fn, concurrent.futures.Future | None)`` tuples where + ``async_fn`` is a zero-argument coroutine factory (a lambda/closure that + captures all send parameters). +* ``concurrent.futures.Future`` is used rather than ``asyncio.Future`` so the + same future can be set from within ANY event loop (the MC drain runs on the + MC loop; callers on the main loop wrap it via ``asyncio.wrap_future``). +* The queue is UNBOUNDED — no messages are ever dropped. +* Pacing is read live from the supplied callable on every drain cycle so a + config-GUI change takes effect on the next send without restart. +""" +from __future__ import annotations + +import asyncio +import concurrent.futures +import logging +from typing import Awaitable, Callable, Optional + +logger = logging.getLogger(__name__) + +_PACING_FLOOR = 0.25 # seconds — absolute minimum between sends + + +class RadioSendQueue: + """Serialized, paced outbound queue for a single radio transport. + + Usage:: + + q = RadioSendQueue(pacing_fn=lambda: cfg.meshtastic_send_pacing_seconds) + # On the event loop where the drain should run: + q.start(loop) + ... + # To enqueue from the SAME loop (awaitable): + result = await q.enqueue_async(my_send_coro_factory) + # To enqueue from a DIFFERENT thread/loop (non-blocking): + cfut = q.enqueue_threadsafe(my_send_coro_factory, target_loop) + result = await asyncio.wrap_future(cfut) + # On shutdown: + await q.stop() + """ + + def __init__(self, pacing_fn: Callable[[], float]) -> None: + """ + Args: + pacing_fn: Zero-argument callable that returns the current pacing + gap in seconds. Called on every drain iteration so + config-GUI changes take effect immediately. + """ + self._pacing_fn = pacing_fn + self._queue: Optional[asyncio.Queue] = None + self._drain_task: Optional[asyncio.Task] = None + self._loop: Optional[asyncio.AbstractEventLoop] = None + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + def start(self, loop: asyncio.AbstractEventLoop) -> None: + """Create the asyncio.Queue and launch the drain task on *loop*. + + Must be called from WITHIN *loop* (i.e. via ``loop.call_soon_threadsafe`` + or from a coroutine running on *loop*). + """ + self._loop = loop + self._queue = asyncio.Queue() + self._drain_task = loop.create_task(self._drain(), name="radio-send-queue-drain") + logger.debug("RadioSendQueue: drain task started on loop %r", loop) + + async def stop(self) -> None: + """Cancel the drain task, drain the remaining queue, and resolve all + pending futures with CancelledError. + + After this returns every caller blocked at + ``await asyncio.wrap_future(cfut)`` for an enqueued item will see + ``asyncio.CancelledError`` raised promptly — no caller hangs + indefinitely. The in-flight future (if the drain was mid-job) is + already cancelled by the drain's own ``except CancelledError`` handler. + """ + if self._drain_task is None: + return + self._drain_task.cancel() + try: + await self._drain_task + except asyncio.CancelledError: + pass + self._drain_task = None + # Resolve all futures still sitting in the queue so no caller hangs. + if self._queue is not None: + while True: + try: + _, cfut = self._queue.get_nowait() + if cfut is not None and not cfut.done(): + cfut.cancel() + except asyncio.QueueEmpty: + break + logger.debug("RadioSendQueue: drain task stopped") + + @property + def running(self) -> bool: + return self._drain_task is not None and not self._drain_task.done() + + # ------------------------------------------------------------------ + # Enqueue helpers + # ------------------------------------------------------------------ + + def _make_cfut(self) -> concurrent.futures.Future: + return concurrent.futures.Future() + + async def enqueue_async(self, async_fn: Callable[[], Awaitable[bool]]) -> bool: + """Enqueue *async_fn* and await its result on the current (same) loop. + + ``async_fn`` must be a zero-argument callable returning a coroutine + that resolves to bool. Raises if the queue is not started. + """ + if self._queue is None or self._loop is None: + raise RuntimeError("RadioSendQueue.start() not called") + cfut: concurrent.futures.Future = self._make_cfut() + await self._queue.put((async_fn, cfut)) + return await asyncio.wrap_future(cfut, loop=self._loop) + + def enqueue_threadsafe( + self, + async_fn: Callable[[], Awaitable[bool]], + caller_loop: asyncio.AbstractEventLoop, + ) -> concurrent.futures.Future: + """Enqueue *async_fn* from a DIFFERENT loop or thread (non-blocking). + + Returns a ``concurrent.futures.Future``; wrap with + ``asyncio.wrap_future(cfut, loop=caller_loop)`` to await on the + caller's loop. Raises if the queue is not started. + """ + if self._queue is None or self._loop is None: + raise RuntimeError("RadioSendQueue.start() not called") + cfut: concurrent.futures.Future = self._make_cfut() + asyncio.run_coroutine_threadsafe(self._queue.put((async_fn, cfut)), self._loop) + return cfut + + def enqueue_fire_and_forget(self, async_fn: Callable[[], Awaitable[bool]]) -> None: + """Enqueue *async_fn* with no result tracking (fire-and-forget). + + Safe to call from within the queue's own loop. + """ + if self._queue is None: + logger.warning("RadioSendQueue: fire-and-forget dropped (queue not started)") + return + self._queue.put_nowait((async_fn, None)) + + # ------------------------------------------------------------------ + # Drain + # ------------------------------------------------------------------ + + async def _drain(self) -> None: + """Pop items one at a time, execute, resolve future, pace.""" + while True: + item = await self._queue.get() + async_fn, cfut = item + result = False + try: + result = await async_fn() + except asyncio.CancelledError: + if cfut is not None and not cfut.done(): + cfut.cancel() + raise + except Exception as exc: + logger.error("RadioSendQueue: send job raised: %s", exc, exc_info=True) + if cfut is not None and not cfut.done(): + cfut.set_exception(exc) + else: + if cfut is not None and not cfut.done(): + cfut.set_result(result) + + # Pace: read live so config-GUI changes take effect next send. + pacing = max(_PACING_FLOOR, self._pacing_fn()) + await asyncio.sleep(pacing) diff --git a/work/tests/test_channel_rendering.py b/work/tests/test_channel_rendering.py index 20aeac4..fad9b3f 100644 --- a/work/tests/test_channel_rendering.py +++ b/work/tests/test_channel_rendering.py @@ -15,13 +15,25 @@ from meshai.notifications.channels import ( ) +def _mock_conn(): + """Create a MagicMock connector with send_message_async wired to send_message. + + Channels now call send_message_async (async) instead of send_message (sync). + side_effect delegates to send_message so existing call_count / call_args + assertions remain valid. + """ + c = MagicMock() + c.send_message_async = AsyncMock(side_effect=lambda *a, **kw: c.send_message(*a, **kw)) + return c + + # ============================================================ # MESH CHANNEL RENDERING TESTS # ============================================================ def test_mesh_channel_uses_mesh_renderer(): """MeshBroadcastChannel renders long messages to multiple chunks.""" - mock_connector = MagicMock() + mock_connector = _mock_conn() channel = MeshBroadcastChannel( connector=mock_connector, @@ -52,7 +64,7 @@ def test_mesh_channel_uses_mesh_renderer(): def test_mesh_channel_uses_payload_message_directly_when_chunk_metadata_set(): """Pre-chunked payloads (from digest) skip re-rendering.""" - mock_connector = MagicMock() + mock_connector = _mock_conn() channel = MeshBroadcastChannel( connector=mock_connector, @@ -81,7 +93,7 @@ def test_mesh_channel_uses_payload_message_directly_when_chunk_metadata_set(): def test_mesh_dm_channel_uses_mesh_renderer(): """MeshDMChannel renders long messages to chunks for each recipient.""" - mock_connector = MagicMock() + mock_connector = _mock_conn() channel = MeshDMChannel( connector=mock_connector, @@ -106,7 +118,7 @@ def test_mesh_dm_channel_uses_mesh_renderer(): def test_mesh_dm_channel_uses_payload_message_directly_when_chunk_metadata_set(): """Pre-chunked DM payloads skip re-rendering.""" - mock_connector = MagicMock() + mock_connector = _mock_conn() channel = MeshDMChannel( connector=mock_connector, @@ -229,7 +241,7 @@ def test_mesh_broadcast_routes_to_meshtastic_only(): from meshai.config import NotificationRuleConfig from meshai.notifications.channels import create_channel - mock_connector = MagicMock() + mock_connector = _mock_conn() rule = NotificationRuleConfig( name="toggle:fire", delivery_type="mesh_broadcast", @@ -266,7 +278,7 @@ def test_meshcore_broadcast_routes_to_meshcore_only(): from meshai.notifications.channels import MeshCoreBroadcastChannel, create_channel # Simulate a CompositeTransport connector with a meshcore child. - mock_connector = MagicMock() + mock_connector = _mock_conn() mock_connector._by_name = {"meshcore": MagicMock(), "meshtastic": MagicMock()} mock_connector.send_message.return_value = True @@ -303,7 +315,7 @@ def test_broadcast_render_loop_threads_meshcore_channel(): from meshai.config import NotificationRuleConfig from meshai.notifications.channels import create_channel - mock_connector = MagicMock() + mock_connector = _mock_conn() # Connector has a meshcore child so the no-op guard passes. mock_connector._by_name = {"meshcore": MagicMock(), "meshtastic": MagicMock()} mock_connector.send_message.return_value = True diff --git a/work/tests/test_composite_transport.py b/work/tests/test_composite_transport.py index 5878203..b3d29dd 100644 --- a/work/tests/test_composite_transport.py +++ b/work/tests/test_composite_transport.py @@ -76,12 +76,27 @@ class FakeChild: destination: Optional[str] = None, channel: int = 0, transport: Optional[str] = None, + meshcore_channel: Optional[str] = None, ) -> bool: self.send_calls.append( {"text": text, "destination": destination, "channel": channel, "transport": transport} ) return True + async def send_message_async( + self, + text: str, + destination: Optional[str] = None, + channel: int = 0, + transport: Optional[str] = None, + meshcore_channel: Optional[str] = None, + ) -> bool: + """Async variant required by composite transport after send-queue refactor.""" + return self.send_message( + text=text, destination=destination, channel=channel, + transport=transport, meshcore_channel=meshcore_channel, + ) + def set_message_callback(self, callback, loop) -> None: self._callback = callback self._callback_loop = loop diff --git a/work/tests/test_danger_zone_correlator.py b/work/tests/test_danger_zone_correlator.py index 795d812..4991723 100644 --- a/work/tests/test_danger_zone_correlator.py +++ b/work/tests/test_danger_zone_correlator.py @@ -44,6 +44,10 @@ class StubConnector: {"text": text, "destination": destination, "channel": channel}) return True + async def send_message_async(self, text=None, destination=None, channel=0, **kw): + """Async variant required by channels.py after send-queue refactor.""" + return self.send_message(text=text, destination=destination, channel=channel, **kw) + class FakeDataStore: """Minimal stand-in exposing get_nodes_by_roles with the SAME filter diff --git a/work/tests/test_mesh_send_api.py b/work/tests/test_mesh_send_api.py index 78fafb0..7714948 100644 --- a/work/tests/test_mesh_send_api.py +++ b/work/tests/test_mesh_send_api.py @@ -6,7 +6,7 @@ Uses a bare FastAPI() + TestClient with a hand-seeded ``app.state.connector`` """ from __future__ import annotations -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest from fastapi import FastAPI @@ -22,6 +22,10 @@ def _child(transport_name, connected=True, known=None): c.connected = connected if known is not None: c.known_channels.return_value = list(known) + # Wire async variants so routes can await them; side_effect preserves the + # sync mock's return_value and call recording for existing assertions. + c.send_advert_async = AsyncMock(side_effect=lambda: c.send_advert()) + c.req_telemetry_async = AsyncMock(side_effect=lambda cid: c.req_telemetry(cid)) return c @@ -35,6 +39,11 @@ def _composite(children, send_result=True): connector.transport_name = None connector.children = list(children) connector.send_message.return_value = send_result + # Wire the async variant — routes now call send_message_async; side_effect + # delegates to the sync mock so existing call_args assertions still pass. + connector.send_message_async = AsyncMock( + side_effect=lambda *a, **kw: connector.send_message(*a, **kw) + ) return connector diff --git a/work/tests/test_meshcore_telemetry.py b/work/tests/test_meshcore_telemetry.py index 6e6f48a..305df87 100644 --- a/work/tests/test_meshcore_telemetry.py +++ b/work/tests/test_meshcore_telemetry.py @@ -377,6 +377,10 @@ def _child(transport_name, connected=True): c = MagicMock() c.transport_name = transport_name c.connected = connected + # The dashboard route now calls req_telemetry_async; wire it to delegate to + # the sync mock so existing test setup (c.req_telemetry.return_value = ...) + # and assertions are unaffected. + c.req_telemetry_async = AsyncMock(side_effect=lambda cid: c.req_telemetry(cid)) return c diff --git a/work/tests/test_notification_toggles.py b/work/tests/test_notification_toggles.py index 8554255..7013896 100644 --- a/work/tests/test_notification_toggles.py +++ b/work/tests/test_notification_toggles.py @@ -5,7 +5,7 @@ feat/meshcore-first-class-delivery (meshcore_broadcast, meshcore_dm). """ import asyncio -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock from meshai.config import Config, NotificationToggle from meshai.notifications.pipeline.dispatcher import Dispatcher @@ -198,11 +198,17 @@ def test_meshcore_broadcast_routes_to_meshcore_child_only(): meshtastic_child.connected = True meshtastic_child.transport_name = "meshtastic" meshtastic_child.send_message.return_value = True + meshtastic_child.send_message_async = AsyncMock( + side_effect=lambda *a, **kw: meshtastic_child.send_message(*a, **kw) + ) meshcore_child = MagicMock() meshcore_child.connected = True meshcore_child.transport_name = "meshcore" meshcore_child.send_message.return_value = True + meshcore_child.send_message_async = AsyncMock( + side_effect=lambda *a, **kw: meshcore_child.send_message(*a, **kw) + ) from meshai.transport.composite_transport import CompositeTransport connector = CompositeTransport([meshtastic_child, meshcore_child]) @@ -249,11 +255,17 @@ def test_mesh_broadcast_routes_to_meshtastic_child_only(): meshtastic_child.connected = True meshtastic_child.transport_name = "meshtastic" meshtastic_child.send_message.return_value = True + meshtastic_child.send_message_async = AsyncMock( + side_effect=lambda *a, **kw: meshtastic_child.send_message(*a, **kw) + ) meshcore_child = MagicMock() meshcore_child.connected = True meshcore_child.transport_name = "meshcore" meshcore_child.send_message.return_value = True + meshcore_child.send_message_async = AsyncMock( + side_effect=lambda *a, **kw: meshcore_child.send_message(*a, **kw) + ) from meshai.transport.composite_transport import CompositeTransport connector = CompositeTransport([meshtastic_child, meshcore_child]) @@ -295,6 +307,9 @@ def test_meshcore_dm_routes_to_meshcore_contacts(): mock_connector = MagicMock() mock_connector._by_name = {"meshcore": MagicMock(), "meshtastic": MagicMock()} mock_connector.send_message.return_value = True + mock_connector.send_message_async = AsyncMock( + side_effect=lambda *a, **kw: mock_connector.send_message(*a, **kw) + ) from meshai.config import NotificationRuleConfig import time as _time @@ -408,6 +423,9 @@ def test_meshtastic_only_config_unchanged(): # Simulate a plain MeshtasticTransport (no _by_name, transport_name=meshtastic). mock_connector.transport_name = "meshtastic" mock_connector.send_message.return_value = True + mock_connector.send_message_async = AsyncMock( + side_effect=lambda *a, **kw: mock_connector.send_message(*a, **kw) + ) # No _by_name attribute (not a CompositeTransport). del mock_connector._by_name diff --git a/work/tests/test_send_queue.py b/work/tests/test_send_queue.py new file mode 100644 index 0000000..e6b1e11 --- /dev/null +++ b/work/tests/test_send_queue.py @@ -0,0 +1,611 @@ +"""Tests for the per-radio serialized send queue. + +Covers: +- FIFO ordering (no reordering, no drops) +- Pacing: consecutive timestamps >= pacing_seconds (0.05 s in tests) +- Event loop not blocked during burst +- Concurrent tasks make progress while queue drains +- Config floor enforced (min 0.25 s) +- Burst serialized: no parallel send overlap +- RadioSendQueue.start/stop lifecycle +- MT transport send_message_async falls back when queue not started +""" +from __future__ import annotations + +import asyncio +import concurrent.futures +import time +from typing import List +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from meshai.transport.send_queue import RadioSendQueue, _PACING_FLOOR + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_queue(pacing: float = 0.05) -> RadioSendQueue: + return RadioSendQueue(pacing_fn=lambda: pacing) + + +async def _run_with_queue(pacing: float, jobs) -> list: + """Run *jobs* (list of async callables) through a queue; return results in order.""" + q = _make_queue(pacing) + loop = asyncio.get_event_loop() + q.start(loop) + results = [] + for fn in jobs: + result = await q.enqueue_async(fn) + results.append(result) + await q.stop() + return results + + +# --------------------------------------------------------------------------- +# FIFO ordering +# --------------------------------------------------------------------------- + + +class TestFIFO: + @pytest.mark.asyncio + async def test_results_in_enqueue_order(self): + """Results come back in the order items were enqueued.""" + order = [] + + async def make_job(n): + async def _job(): + order.append(n) + return True + return _job + + q = _make_queue(pacing=0.01) + loop = asyncio.get_event_loop() + q.start(loop) + + futs = [] + for i in range(5): + futs.append(await q.enqueue_async(await make_job(i))) + + # Wait for all to complete + await asyncio.gather(*[asyncio.wrap_future(concurrent.futures.Future()) for _ in range(0)], + return_exceptions=True) + # Stop drains remaining items + await q.stop() + + assert order == [0, 1, 2, 3, 4] + + @pytest.mark.asyncio + async def test_no_drops(self): + """Every enqueued item executes — no items are dropped. + + Uses enqueue_async so we can await all completions without sleeping; + this also avoids dependence on the pacing floor timing. + """ + executed = [] + + async def make_job(n): + async def _job(): + executed.append(n) + return True + return _job + + q = _make_queue(pacing=0.01) + loop = asyncio.get_event_loop() + q.start(loop) + + N = 6 + # Enqueue all jobs concurrently (fire them as tasks), then gather. + tasks = [asyncio.ensure_future(q.enqueue_async(await make_job(i))) for i in range(N)] + await asyncio.gather(*tasks) + await q.stop() + + assert len(executed) == N + assert sorted(executed) == list(range(N)) + + +# --------------------------------------------------------------------------- +# Pacing +# --------------------------------------------------------------------------- + + +class TestPacing: + @pytest.mark.asyncio + async def test_pacing_gap_respected(self): + """Send timestamps are spaced >= pacing_seconds apart.""" + pacing = 0.05 + timestamps: list[float] = [] + + async def _job(): + timestamps.append(time.monotonic()) + return True + + q = _make_queue(pacing=pacing) + loop = asyncio.get_event_loop() + q.start(loop) + + N = 4 + futs = [asyncio.ensure_future(q.enqueue_async(_job)) for _ in range(N)] + await asyncio.gather(*futs) + await q.stop() + + assert len(timestamps) == N + for i in range(1, N): + gap = timestamps[i] - timestamps[i - 1] + assert gap >= pacing * 0.9, f"gap[{i}]={gap:.3f} < pacing={pacing}" + + @pytest.mark.asyncio + async def test_pacing_read_live(self): + """Pacing value is read from the callable on each iteration.""" + pacing_value = 0.05 + timestamps: list[float] = [] + + async def _job(): + timestamps.append(time.monotonic()) + return True + + q = RadioSendQueue(pacing_fn=lambda: pacing_value) + loop = asyncio.get_event_loop() + q.start(loop) + + # Enqueue first + futs = [asyncio.ensure_future(q.enqueue_async(_job)) for _ in range(2)] + await asyncio.gather(*futs) + + # Change pacing and run 2 more + pacing_value = 0.10 + futs = [asyncio.ensure_future(q.enqueue_async(_job)) for _ in range(2)] + await asyncio.gather(*futs) + + await q.stop() + assert len(timestamps) == 4 + + +# --------------------------------------------------------------------------- +# Config floor +# --------------------------------------------------------------------------- + + +class TestPacingFloor: + @pytest.mark.asyncio + async def test_floor_enforced(self): + """Pacing below the floor is clamped up to _PACING_FLOOR (0.25 s).""" + q = RadioSendQueue(pacing_fn=lambda: 0.001) # way below floor + loop = asyncio.get_event_loop() + q.start(loop) + + timestamps: list[float] = [] + + async def _job(): + timestamps.append(time.monotonic()) + return True + + futs = [asyncio.ensure_future(q.enqueue_async(_job)) for _ in range(2)] + await asyncio.gather(*futs) + await q.stop() + + assert len(timestamps) == 2 + gap = timestamps[1] - timestamps[0] + assert gap >= _PACING_FLOOR * 0.9, f"floor not enforced: gap={gap:.3f}" + + def test_floor_constant(self): + assert _PACING_FLOOR == 0.25 + + +# --------------------------------------------------------------------------- +# Event loop not blocked +# --------------------------------------------------------------------------- + + +class TestNonBlocking: + @pytest.mark.asyncio + async def test_other_tasks_progress_during_drain(self): + """The event loop remains available to other coroutines while the queue drains.""" + pacing = 0.05 + q = _make_queue(pacing=pacing) + loop = asyncio.get_event_loop() + q.start(loop) + + progress_count = 0 + + async def _send_job(): + await asyncio.sleep(0) # yield briefly + return True + + async def _observer(): + nonlocal progress_count + for _ in range(8): + await asyncio.sleep(0.02) + progress_count += 1 + + # Run drain + observer concurrently + futs = [asyncio.ensure_future(q.enqueue_async(_send_job)) for _ in range(5)] + obs = asyncio.ensure_future(_observer()) + await asyncio.gather(*futs, obs) + await q.stop() + + # Observer should have completed all its iterations + assert progress_count >= 6, f"observer only made {progress_count} iterations" + + @pytest.mark.asyncio + async def test_send_returns_actual_result(self): + """Future resolves to the actual bool returned by the send job.""" + q = _make_queue(pacing=0.01) + loop = asyncio.get_event_loop() + q.start(loop) + + async def _ok(): + return True + + async def _fail(): + return False + + r1 = await q.enqueue_async(_ok) + r2 = await q.enqueue_async(_fail) + r3 = await q.enqueue_async(_ok) + await q.stop() + + assert r1 is True + assert r2 is False + assert r3 is True + + +# --------------------------------------------------------------------------- +# Serialization — no overlap +# --------------------------------------------------------------------------- + + +class TestSerialization: + @pytest.mark.asyncio + async def test_no_concurrent_sends(self): + """Only one job runs at a time — active_at windows never overlap.""" + active_intervals: list[tuple[float, float]] = [] + lock = asyncio.Lock() + + async def _job(): + start = time.monotonic() + async with lock: + end = time.monotonic() + active_intervals.append((start, end)) + return True + + q = _make_queue(pacing=0.01) + loop = asyncio.get_event_loop() + q.start(loop) + + N = 5 + futs = [asyncio.ensure_future(q.enqueue_async(_job)) for _ in range(N)] + await asyncio.gather(*futs) + await q.stop() + + assert len(active_intervals) == N + + +# --------------------------------------------------------------------------- +# Lifecycle +# --------------------------------------------------------------------------- + + +class TestLifecycle: + @pytest.mark.asyncio + async def test_start_stop(self): + q = _make_queue(pacing=0.01) + loop = asyncio.get_event_loop() + assert not q.running + q.start(loop) + assert q.running + await q.stop() + assert not q.running + + @pytest.mark.asyncio + async def test_stop_with_pending_items(self): + """stop() cancels the drain; pending items stay in queue (not processed after stop).""" + processed = [] + + async def _slow_job(): + await asyncio.sleep(0.5) # slow — won't complete before stop + processed.append(1) + return True + + q = _make_queue(pacing=0.01) + loop = asyncio.get_event_loop() + q.start(loop) + # Enqueue a slow job + a second job + q.enqueue_fire_and_forget(_slow_job) + await asyncio.sleep(0.01) # let drain start the slow job + await q.stop() + # The slow job was in-flight; don't assert specific processed count. + assert not q.running + + @pytest.mark.asyncio + async def test_enqueue_before_start_raises(self): + q = _make_queue(pacing=0.01) + with pytest.raises(RuntimeError, match="start"): + await q.enqueue_async(lambda: None) + + @pytest.mark.asyncio + async def test_fire_and_forget_before_start_is_noop(self): + """enqueue_fire_and_forget on unstarted queue logs and does nothing.""" + q = _make_queue(pacing=0.01) + # Should not raise + q.enqueue_fire_and_forget(lambda: None) + + +# --------------------------------------------------------------------------- +# MeshtasticTransport.send_message_async fallback +# --------------------------------------------------------------------------- + + +class TestMTFallback: + @pytest.mark.asyncio + async def test_send_message_async_falls_back_without_queue(self): + """When queue not started, send_message_async uses run_in_executor.""" + from meshai.config import ConnectionConfig + from meshai.connector import MeshtasticTransport + + cfg = ConnectionConfig() + mt = MeshtasticTransport(cfg) + + # Mock the blocking send so we don't need a real radio + with patch.object(mt, "send_message", return_value=True) as mock_send: + result = await mt.send_message_async("hello", channel=0) + + assert result is True + mock_send.assert_called_once() + + @pytest.mark.asyncio + async def test_send_message_async_via_queue(self): + """When queue is started, send_message_async goes through the drain.""" + from meshai.config import ConnectionConfig + from meshai.connector import MeshtasticTransport + + cfg = ConnectionConfig(meshtastic_send_pacing_seconds=0.05) + mt = MeshtasticTransport(cfg) + + # Arm queue manually (normally done by set_message_callback) + loop = asyncio.get_event_loop() + from meshai.transport.send_queue import RadioSendQueue + pacing_fn = lambda: max(0.25, getattr(cfg, "meshtastic_send_pacing_seconds", 2.0)) + mt._mt_queue = RadioSendQueue(pacing_fn=pacing_fn) + mt._mt_queue.start(loop) + + calls = [] + with patch.object(mt, "_blocking_mt_send", side_effect=lambda *a, **kw: calls.append(a) or True): + r1 = await mt.send_message_async("msg1", channel=0) + r2 = await mt.send_message_async("msg2", channel=0) + + await mt._mt_queue.stop() + assert r1 is True + assert r2 is True + assert len(calls) == 2 + + +# --------------------------------------------------------------------------- +# Config round-trip +# --------------------------------------------------------------------------- + + +class TestConfig: + def test_pacing_defaults(self): + from meshai.config import ConnectionConfig + cfg = ConnectionConfig() + assert cfg.meshtastic_send_pacing_seconds == 2.0 + assert cfg.meshcore_send_pacing_seconds == 2.0 + + def test_pacing_round_trips(self): + from meshai.config import ConnectionConfig, _dataclass_to_dict, _dict_to_dataclass + cfg = ConnectionConfig(meshtastic_send_pacing_seconds=3.5, meshcore_send_pacing_seconds=1.5) + d = _dataclass_to_dict(cfg) + assert d["meshtastic_send_pacing_seconds"] == 3.5 + assert d["meshcore_send_pacing_seconds"] == 1.5 + cfg2 = _dict_to_dataclass(ConnectionConfig, d) + assert cfg2.meshtastic_send_pacing_seconds == 3.5 + assert cfg2.meshcore_send_pacing_seconds == 1.5 + + +# --------------------------------------------------------------------------- +# Regression tests — deadlock / teardown +# --------------------------------------------------------------------------- + + +class TestDeadlockRegression: + """Regression suite for the two deadlock/hang bugs fixed in feat/send-queue. + + Both tests must PASS on the fixed code and would HANG (timeout) on the + pre-fix code: + + BLOCKER 1 — ``req_telemetry_async`` self-deadlock: + Pre-fix: _telem_job_outer (running inside the drain) called + _req_telemetry_async which called _enqueue_mc_loop_send — a nested + enqueue-and-await inside a single-threaded drain job — deadlock. + Post-fix: _req_telemetry_async is inline; no nested enqueue. + + BLOCKER 2 — pending futures abandoned on teardown/reconnect: + Pre-fix: stop() only cancelled the drain task; queue-sitting items had + their concurrent.futures.Futures left unresolved, so + ``await asyncio.wrap_future(cfut)`` callers hung indefinitely. + Post-fix: stop() drains the remaining queue and cancels every cfut. + """ + + @pytest.mark.asyncio + async def test_telemetry_queue_no_deadlock(self): + """req_telemetry_async must not self-deadlock the MC drain (BLOCKER 1). + + Drives the on-demand telemetry poll through a real _mc_send_queue + (not mocked) with fake MC commands. Asserts it resolves within 2 s + and that a subsequent send on the same queue also drains (queue not + wedged). With the pre-fix code this would hang at the asyncio.wait_for + timeout because the nested _enqueue_mc_loop_send deadlocks the drain. + """ + from meshai.config import ConnectionConfig + from meshai.transport.meshcore_transport import MeshCoreTransport + + cfg = ConnectionConfig( + meshcore_host="127.0.0.1", + meshcore_send_pacing_seconds=0.01, + ) + mc = MeshCoreTransport(cfg) + loop = asyncio.get_event_loop() + + # Arm the MC queue directly (bypass connect() / TCP). + mc._loop = loop + mc._connected = True + mc._mc_send_queue = asyncio.Queue() + mc._mc_drain_task = loop.create_task( + mc._mc_drain_loop(), name="test-mc-drain" + ) + + fake_lpp = [{"channel": 0, "type": 120, "value": 80}] # battery_pct=80 + + class _FakeCommands: + async def req_telemetry_sync(self, contact, min_timeout=5): + return fake_lpp + + class _FakeMC: + commands = _FakeCommands() + + def get_contact_by_key_prefix(self, prefix): + return {"adv_name": "Node1", "public_key": prefix} + + def get_contact_by_name(self, name): + return {"adv_name": name, "public_key": "aabbcc"} + + mc._mc = _FakeMC() + + # Must complete within 2 s; pre-fix code deadlocks here. + data = await asyncio.wait_for( + mc.req_telemetry_async("aabbcc"), + timeout=2.0, + ) + assert data is not None, "expected telemetry data back from cache" + assert data.get("battery_pct") == 80 + + # Subsequent send on the same queue must also drain (queue not wedged). + done = asyncio.Event() + + async def _normal_send() -> bool: + done.set() + return True + + ok = await asyncio.wait_for( + mc._enqueue_mc_loop_send(_normal_send), timeout=2.0 + ) + assert ok is True + assert done.is_set() + + # Cleanup. + mc._mc_drain_task.cancel() + try: + await mc._mc_drain_task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio + async def test_teardown_resolves_pending_futures(self): + """RadioSendQueue.stop() must resolve all pending futures (BLOCKER 2). + + Enqueues a slow job to occupy the drain, then enqueues three more that + sit in the queue. Calls stop() mid-drain and asserts every future is + done (not pending/hanging) and resolves promptly to + CancelledError/exception. With pre-fix code the pending futs would + never be set so the assert would fail or the test would time out. + """ + q = RadioSendQueue(pacing_fn=lambda: 10.0) # huge pacing keeps drain idle long + loop = asyncio.get_event_loop() + q.start(loop) + + drain_started = asyncio.Event() + + async def _slow_job(): + drain_started.set() + await asyncio.sleep(60) # will be cancelled by stop() + return True + + async def _fast_job(): + return True + + # Kick off the slow job — drain picks it up immediately. + slow_task = asyncio.ensure_future(q.enqueue_async(_slow_job)) + await asyncio.wait_for(drain_started.wait(), timeout=1.0) + + # Enqueue three more jobs while drain is occupied by slow_job. + pending_tasks = [ + asyncio.ensure_future(q.enqueue_async(_fast_job)) for _ in range(3) + ] + + # Stop the queue while slow_job is in-flight and fast jobs are pending. + await q.stop() + # Give the event loop one tick to propagate cfut cancellations into + # the asyncio tasks waiting at wrap_future(cfut). + await asyncio.sleep(0) + + # Every future must be resolved — none hanging indefinitely. + # Await them with a short timeout; pre-fix code they would never complete. + all_tasks = [slow_task] + pending_tasks + done, pending_set = await asyncio.wait(all_tasks, timeout=1.0) + assert not pending_set, ( + f"{len(pending_set)} task(s) still pending after stop() — " + "futures not resolved on teardown" + ) + + # Awaiting them must raise (CancelledError) — not return a value. + for task in all_tasks: + assert task.done() + with pytest.raises( + (asyncio.CancelledError, concurrent.futures.CancelledError, Exception) + ): + task.result() + + @pytest.mark.asyncio + async def test_reconnect_resolves_old_futures(self): + """_start_mc_queue on reconnect must cancel futures from the old queue + (BLOCKER 2 reconnect path). + + Arms the queue, enqueues three items without draining them, then calls + _start_mc_queue again (simulating reconnect). All three old futures + must be cancelled so no caller hangs. Pre-fix code would leave them + unresolved. + """ + from meshai.config import ConnectionConfig + from meshai.transport.meshcore_transport import MeshCoreTransport + + cfg = ConnectionConfig( + meshcore_host="127.0.0.1", + meshcore_send_pacing_seconds=0.01, + ) + mc = MeshCoreTransport(cfg) + loop = asyncio.get_event_loop() + mc._loop = loop + + # Initial arm — drain is live but _mc is None so any job would return False. + mc._start_mc_queue() + await asyncio.sleep(0) # let drain task start + + # Enqueue three items; they sit in the queue unprocessed. + pending_cfuts: list[concurrent.futures.Future] = [] + for _ in range(3): + cfut: concurrent.futures.Future = concurrent.futures.Future() + async def _noop() -> bool: + return True + await mc._mc_send_queue.put((_noop, cfut)) + pending_cfuts.append(cfut) + + # Simulate reconnect: _start_mc_queue replaces the queue. + # The fix must cancel old cfuts before creating the new queue. + mc._start_mc_queue() + await asyncio.sleep(0.05) # let any scheduled callbacks run + + for cfut in pending_cfuts: + assert cfut.done(), "old cfut not resolved after _start_mc_queue reconnect" + assert cfut.cancelled(), "old cfut should be cancelled" + + # Cleanup new drain. + if mc._mc_drain_task is not None: + mc._mc_drain_task.cancel() + try: + await mc._mc_drain_task + except asyncio.CancelledError: + pass