fix(core): release the radio lock while waiting for ACKs

Radio commands were serialized per method call, so a composite command held the lock for everything it awaited. A DM's send_msg_with_retry kept every other command waiting through all of its ACK timeouts (up to ~36 s with three attempts), stalling channel replies, other DMs, scheduled sends, and automatic message fetching. req_regions_sync did the same for each neighbor scope reply.

Serialize at the frame instead: wrap CommandHandler.send() on the handler instance, which is where every meshcore command writes its frame and waits for the radio's immediate reply. There is still at most one in-flight companion frame, paced as before, and library methods that call self.send are covered without a proxy. Waits for ACKs and remote responses now run outside the lock.

Add MeshCoreBot.radio_session() for short sequences that must not interleave, and use it for region-scoped channel sends so set scope, send, and restore stay together. The per-call lock never guaranteed that: a DM retry loop queued between set_flood_scope and send_chan_msg sent its flood attempts under the channel's scope.
This commit is contained in:
agessaman committed 2026-09-18 11:25:52 -07:00
1 parent 1a8f108ac0
commit 563b830bc5
7 files changed
+272 -147

No files matched your search

+8
View File
@@ -18,6 +18,14 @@ semantic versioning.
### Fixed
- A DM waiting for its ACK no longer holds the radio. Radio commands were
serialized per call, so a DM's retry loop kept every other command waiting
through all of its ACK timeouts (up to ~36 s with the default three attempts),
stalling channel replies, other DMs, and scheduled sends. The lock now covers
one frame and the radio's immediate reply, and ACK and remote-response waits
run outside it, so neighbor scope requests no longer stall replies either. A
region-scoped channel message now holds the radio from setting its flood scope
through restoring it, so no other send goes out under that scope.
- A radio connection is no longer accepted when every channel read times out or
returns no usable channel data (#266). Startup and reconnect now retry the
channel scan three times, keep an empty result out of the valid cache and
+13 -17
View File
@@ -481,24 +481,20 @@ more minute" would be followed by fourteen more minutes of personal cooldown.
### Region scopes are opt-in, and why
`neighbors_collect_scopes` additionally asks each neighbour for its region scopes.
It defaults to **false** for two reasons specific to running inside the bot:
It defaults to **false** for a reason specific to running inside the bot:
1. **It stalls bot replies.** Every bot radio command is serialised through one lock
(`modules/core.py` `_SerializedCommands`), and `req_regions_sync` waits for its
reply *inside* the call — so one request holds the radio for up to ~25 s. With 32
neighbours the bot's own messages stall in bursts for minutes.
2. **It mutates device contact state.** The zero-hop probe relies on the neighbour
*not* being a known contact. The bot does track contacts, and for a repeater with
no stored path the meshcore library reaches zero-hop by calling
`change_contact_path()` and then `reset_path()` — temporarily rewriting that
contact's path on the device. Those two calls are not paired by a
`try`/`finally` upstream, and one error path returns between them, so a request
cut short — or one whose path change was applied but not acknowledged — would
leave the contact pinned to zero-hop and every later message to it sent
direct-only. `modules/neighbors_discovery.py` restores the path itself in each
of those cases, and warns if the device rejects the restore (which it reports
as an error event rather than an exception), since that contact's routing is
then wrong until something else fixes it.
- **It mutates device contact state.** The zero-hop probe relies on the neighbour
*not* being a known contact. The bot does track contacts, and for a repeater with
no stored path the meshcore library reaches zero-hop by calling
`change_contact_path()` and then `reset_path()` — temporarily rewriting that
contact's path on the device. Those two calls are not paired by a
`try`/`finally` upstream, and one error path returns between them, so a request
cut short — or one whose path change was applied but not acknowledged — would
leave the contact pinned to zero-hop and every later message to it sent
direct-only. `modules/neighbors_discovery.py` restores the path itself in each
of those cases, and warns if the device rejects the restore (which it reports
as an error event rather than an exception), since that contact's routing is
then wrong until something else fixes it.
With it off, the snapshot reports every neighbour it heard with empty `scopes` and
`status: responded`. Enable it on a bench radio first.
+32 -29
View File
@@ -5,6 +5,7 @@ Handles all bot commands, keyword matching, and response generation
"""
import asyncio
import contextlib
import random
import time
from dataclasses import dataclass
@@ -1341,37 +1342,47 @@ class CommandManager:
scope_to_use,
scope_source,
)
if not scope_is_global and not hasattr(self.bot.meshcore.commands, "set_flood_scope"):
scoped = not scope_is_global and hasattr(self.bot.meshcore.commands, "set_flood_scope")
if not scope_is_global and not scoped:
self.logger.warning(
"Regional flood scope %r requested but meshcore.commands.set_flood_scope "
"is unavailable; channel message will use device default (often global flood)",
scope_to_use,
)
elif not scope_is_global:
_scope_result = await self.bot.meshcore.commands.set_flood_scope(scope_to_use)
if _scope_result is None or getattr(_scope_result, "type", None) == "ERROR":
self.logger.warning(
"set_flood_scope(%s) failed (result=%s); "
"message will be sent with current firmware scope",
scope_to_use, _scope_result,
)
target = f"{channel} (channel {channel_num})"
# Retry on no_event_received: max 2 extra attempts, 2s apart
_max_retries = 2
for _attempt in range(_max_retries + 1):
try:
result = await self.bot.meshcore.commands.send_chan_msg(
channel_num, content,
timestamp=int(timestamp.timestamp()) if timestamp else None,
)
finally:
if not scope_is_global and hasattr(self.bot.meshcore.commands, "set_flood_scope"):
_restore_result = await self.bot.meshcore.commands.set_flood_scope("*")
if _restore_result is None or getattr(_restore_result, "type", None) == "ERROR":
self.logger.warning(
"set_flood_scope('*') restore failed (result=%s)", _restore_result
)
# Hold the radio from set to restore so no other send goes out
# under this message's scope.
async with self.bot.radio_session() if scoped else contextlib.nullcontext():
if scoped:
_scope_result = await self.bot.meshcore.commands.set_flood_scope(scope_to_use)
if _scope_result is None or getattr(_scope_result, "type", None) == "ERROR":
if _attempt == 0:
self.logger.warning(
"set_flood_scope(%s) failed (result=%s); "
"message will be sent with current firmware scope",
scope_to_use, _scope_result,
)
else:
self.logger.warning(
"set_flood_scope(%s) failed on retry re-apply (result=%s)",
scope_to_use, _scope_result,
)
try:
result = await self.bot.meshcore.commands.send_chan_msg(
channel_num, content,
timestamp=int(timestamp.timestamp()) if timestamp else None,
)
finally:
if scoped:
_restore_result = await self.bot.meshcore.commands.set_flood_scope("*")
if _restore_result is None or getattr(_restore_result, "type", None) == "ERROR":
self.logger.warning(
"set_flood_scope('*') restore failed (result=%s)", _restore_result
)
if self._is_no_event_received(result) and _attempt < _max_retries:
self.logger.warning(
@@ -1379,14 +1390,6 @@ class CommandManager:
f"(attempt {_attempt + 1}/{_max_retries + 1}), retrying in 2s"
)
await asyncio.sleep(2)
# Re-apply scope for next attempt
if not scope_is_global and hasattr(self.bot.meshcore.commands, "set_flood_scope"):
_scope_result = await self.bot.meshcore.commands.set_flood_scope(scope_to_use)
if _scope_result is None or getattr(_scope_result, "type", None) == "ERROR":
self.logger.warning(
"set_flood_scope(%s) failed on retry re-apply (result=%s)",
scope_to_use, _scope_result,
)
continue
break
+68 -40
View File
@@ -7,8 +7,9 @@ Contains the main bot class and message processing logic
import asyncio
import atexit
import configparser
import contextlib
import contextvars
import functools
import inspect
import json
import logging
import signal
@@ -119,8 +120,15 @@ class _BotAdminServer(threading.Thread):
self._bot.logger.error("BotAdminServer failed to start: %s", exc)
class _SerializedCommands:
"""Serializing proxy around ``meshcore.commands``.
# True while the current task holds the radio through MeshCoreBot.radio_session(),
# so the frames it sends don't try to take the (non-reentrant) lock again.
_radio_session_held: contextvars.ContextVar[bool] = contextvars.ContextVar(
"radio_session_held", default=False
)
def _serialize_command_frames(bot: "MeshCoreBot", commands: Any) -> bool:
"""Route every host->radio frame through the bot's radio command lock.
The companion firmware processes one host serial frame per main-loop
iteration and has no mid-frame resync: a burst of concurrent commands can
@@ -130,34 +138,33 @@ class _SerializedCommands:
(sends, channel/contact ops, scheduler ops, health probes, auto message
fetch) with no shared serialization.
This proxy routes every coroutine command through a single per-bot lock and
enforces a minimum inter-command interval, guaranteeing at most one
in-flight companion frame at a time. Non-coroutine attributes are passed
through untouched, so library internals that read ``_sender_func``,
``default_timeout`` etc. are unaffected. ``meshcore_cli.next_cmd`` calls are
serialized too, since they dispatch through this same ``commands`` object.
Every meshcore command writes its frame through ``CommandHandler.send()``,
which waits for the radio's immediate reply (OK, ERROR, MSG_SENT, ...).
Wrapping ``send`` on the handler instance serializes exactly that exchange
and paces frames by a minimum interval, so there is at most one in-flight
companion frame at a time. Library methods call ``self.send``, so composite
commands (``send_msg_with_retry``, ``req_*_sync``, ``send_login_sync``) and
``meshcore_cli.next_cmd`` are covered too, while their waits for ACKs and
remote responses happen outside the lock and don't block other senders.
Returns False when ``commands`` is already serialized.
"""
send = commands.send
if getattr(send, "_radio_serialized", False):
return False
__slots__ = ("_bot", "_commands")
@functools.wraps(send)
async def _serialized_send(*args: Any, **kwargs: Any) -> Any:
if _radio_session_held.get():
await bot._pace_radio_command()
return await send(*args, **kwargs)
async with bot._get_radio_cmd_lock():
await bot._pace_radio_command()
return await send(*args, **kwargs)
def __init__(self, bot: "MeshCoreBot", commands: Any) -> None:
object.__setattr__(self, "_bot", bot)
object.__setattr__(self, "_commands", commands)
def __getattr__(self, name: str) -> Any:
commands = object.__getattribute__(self, "_commands")
attr = getattr(commands, name)
if not inspect.iscoroutinefunction(attr):
return attr
bot = object.__getattribute__(self, "_bot")
@functools.wraps(attr)
async def _serialized(*args: Any, **kwargs: Any) -> Any:
async with bot._get_radio_cmd_lock():
await bot._pace_radio_command()
return await attr(*args, **kwargs)
return _serialized
_serialized_send._radio_serialized = True # type: ignore[attr-defined]
commands.send = _serialized_send
return True
def __setattr__(self, name: str, value: Any) -> None:
setattr(object.__getattribute__(self, "_commands"), name, value)
@@ -1739,25 +1746,46 @@ long_jokes = false
await asyncio.sleep(wait)
self._radio_cmd_last_ts = time.monotonic()
def _install_command_serializer(self) -> None:
"""Wrap ``meshcore.commands`` so every command is serialized + paced.
@contextlib.asynccontextmanager
async def radio_session(self):
"""Hold the radio for a short sequence of frames that must not interleave.
Idempotent and safe to call after each (re)connect. Wrapping the
``commands`` attribute in place means existing call sites
(``self.meshcore.commands.*`` and ``meshcore_cli.next_cmd``) are
serialized automatically with no per-call changes.
Frames are serialized one at a time, so another task's frame can land
between two of ours. Use this when that matters, e.g. setting the flood
scope, sending, and restoring it, so no other send goes out under the
temporary scope. Keep it short: every other sender waits, so don't wait
for ACKs or remote responses inside it. Re-entering from the same task
is a no-op. Tasks created inside the session inherit it, so don't spawn
work that sends after the session ends.
"""
if _radio_session_held.get():
yield
return
async with self._get_radio_cmd_lock():
token = _radio_session_held.set(True)
try:
yield
finally:
_radio_session_held.reset(token)
def _install_command_serializer(self) -> None:
"""Serialize and pace every frame ``meshcore.commands`` writes.
Idempotent and safe to call after each (re)connect. The handler is
wrapped in place, so existing call sites (``self.meshcore.commands.*``
and ``meshcore_cli.next_cmd``) need no per-call changes.
"""
if not self.meshcore:
return
cmds = getattr(self.meshcore, "commands", None)
if cmds is None or isinstance(cmds, _SerializedCommands):
if cmds is None:
return
try:
self.meshcore.commands = _SerializedCommands(self, cmds)
self.logger.debug(
"Installed serialized command gateway (min interval %.0fms)",
self._radio_cmd_min_interval * 1000,
)
if _serialize_command_frames(self, cmds):
self.logger.debug(
"Installed serialized command gateway (min interval %.0fms)",
self._radio_cmd_min_interval * 1000,
)
except (AttributeError, TypeError) as e:
self.logger.warning(f"Could not install command serializer: {e}")
+19 -21
View File
@@ -18,25 +18,23 @@ link* between two full 32-byte public keys, with a measured SNR. That is
stronger evidence than anything path inference can offer, and it costs one radio
command plus a passive listen window.
Stage 2 (scopes) is deliberately optional here and defaults off, for two reasons
that do not apply to the upstream capture tool:
Stage 2 (scopes) is deliberately optional here and defaults off, for a reason
that does not apply to the upstream capture tool: upstream relies on a freshly
discovered neighbor *not* being a known contact, which is what makes
``send_anon_req`` ask for a zero-hop reply path. This bot populates the
library's contact cache. For a contact with no path (``out_path_len == -1``,
the common case for a flood repeater) the library reaches zero-hop by calling
``change_contact_path()`` and then ``reset_path()`` -- i.e. it *mutates the
device's contact table* per neighbor. Those two calls are not paired by
``try``/``finally`` upstream, so a request cut short between them leaves the
contact pinned to zero-hop; collect_scopes restores it itself (see
``_restore_flood_path``).
* ``req_regions_sync`` awaits its reply inside the call, and every bot command
is serialized through ``modules.core._SerializedCommands``, so one scope
request holds the radio for its whole round trip -- stalling message sends.
* Upstream relies on a freshly discovered neighbor *not* being a known contact,
which is what makes ``send_anon_req`` ask for a zero-hop reply path. This bot
populates the library's contact cache. For a contact with no path
(``out_path_len == -1``, the common case for a flood repeater) the library
reaches zero-hop by calling ``change_contact_path()`` and then
``reset_path()`` -- i.e. it *mutates the device's contact table* per neighbor.
Those two calls are not paired by ``try``/``finally`` upstream, so a request
cut short between them leaves the contact pinned to zero-hop; collect_scopes
restores it itself (see ``_restore_flood_path``).
No device-command lock is passed around: unlike upstream, every coroutine on
``meshcore.commands`` is already serialized and paced by
``modules.core._SerializedCommands``.
No device-command lock is passed around: unlike upstream, every frame
``meshcore.commands`` writes is already serialized and paced by
``modules.core._serialize_command_frames``. The lock covers each frame and its
immediate reply only, so waiting for a neighbor's scope reply doesn't stall the
bot's other sends.
"""
from __future__ import annotations
@@ -484,9 +482,9 @@ async def collect_scopes(
pinned_to_zero_hop = _contact_has_no_path(meshcore, entry.pubkey)
try:
# Bounded: this holds the shared radio command lock for its whole
# round trip, and an unbounded stall here would block every other
# bot command for as long as the write hangs.
# Bounded: a hung write holds the shared radio command lock, and an
# unbounded stall here would block every other bot command for as
# long as the write hangs.
scopes = await asyncio.wait_for(
meshcore.commands.req_regions_sync(
entry.pubkey,
+130 -38
View File
@@ -1,9 +1,11 @@
"""Tests for the host->radio command serializer (_SerializedCommands).
"""Tests for the host->radio command serializer (_serialize_command_frames).
These validate the core mitigation for the firmware corruption / parser-desync
failure mode: every command issued to the radio is serialized to one in-flight
failure mode: every frame written to the radio is serialized to one in-flight
companion frame at a time and paced by a minimum inter-command interval, so the
firmware's single-threaded serial loop cannot be overrun.
firmware's single-threaded serial loop cannot be overrun. The lock covers a
frame and its immediate reply only, so commands that wait for ACKs or remote
responses don't block other senders.
"""
import asyncio
@@ -11,8 +13,10 @@ import time
from pathlib import Path
import pytest
from meshcore.commands import CommandHandler
from meshcore.events import Event, EventDispatcher, EventType
from modules.core import MeshCoreBot, _SerializedCommands
from modules.core import MeshCoreBot, _serialize_command_frames
def _make_bot(tmp_path: Path, min_interval_ms: int = 30) -> MeshCoreBot:
@@ -37,17 +41,15 @@ monitor_channels = #general
class FakeCommands:
"""Stand-in for meshcore.commands with an async command + passthrough attrs."""
not_callable = 42
"""Stand-in for meshcore.commands: every command writes through send()."""
def __init__(self):
self.active = 0
self.max_active = 0
self.calls = 0
self.frames = []
async def do_work(self, delay: float = 0.02):
self.calls += 1
async def send(self, data, expected_events=None, timeout=None, delay=0.01):
self.frames.append(data)
self.active += 1
self.max_active = max(self.max_active, self.active)
try:
@@ -56,8 +58,11 @@ class FakeCommands:
finally:
self.active -= 1
def sync_method(self):
return "sync"
async def send_and_wait_for_reply(self, data, wait: float):
"""Like send_msg_with_retry: write one frame, then wait for a remote reply."""
await self.send(data)
await asyncio.sleep(wait)
return "replied"
class FakeMeshcore:
@@ -75,60 +80,147 @@ def test_min_interval_negative_clamped_to_zero(tmp_path):
assert bot._radio_cmd_min_interval == 0.0
async def test_serializes_concurrent_commands(tmp_path):
"""Only one wrapped command may run at a time."""
async def test_serializes_concurrent_frames(tmp_path):
"""Only one frame may be in flight at a time."""
bot = _make_bot(tmp_path, min_interval_ms=0) # isolate mutex from pacing
fake = FakeCommands()
proxy = _SerializedCommands(bot, fake)
_serialize_command_frames(bot, fake)
await asyncio.gather(*(proxy.do_work(delay=0.01) for _ in range(10)))
await asyncio.gather(*(fake.send(b"\x01") for _ in range(10)))
assert fake.calls == 10
assert len(fake.frames) == 10
assert fake.max_active == 1 # never more than one in-flight frame
async def test_pacing_enforces_minimum_gap(tmp_path):
bot = _make_bot(tmp_path, min_interval_ms=50)
fake = FakeCommands()
proxy = _SerializedCommands(bot, fake)
_serialize_command_frames(bot, fake)
start = time.monotonic()
for _ in range(5):
await proxy.do_work(delay=0.0)
await fake.send(b"\x01", delay=0.0)
elapsed = time.monotonic() - start
# 5 commands => 4 enforced gaps of ~50ms (first call is not delayed).
# 5 frames => 4 enforced gaps of ~50ms (first frame is not delayed).
assert elapsed >= 0.18
async def test_non_coroutine_attributes_pass_through(tmp_path):
bot = _make_bot(tmp_path)
fake = FakeCommands()
proxy = _SerializedCommands(bot, fake)
assert proxy.not_callable == 42
assert proxy.sync_method() == "sync"
async def test_wrapped_command_returns_value(tmp_path):
async def test_wrapped_send_returns_value(tmp_path):
bot = _make_bot(tmp_path, min_interval_ms=0)
fake = FakeCommands()
proxy = _SerializedCommands(bot, fake)
_serialize_command_frames(bot, fake)
assert await proxy.do_work(delay=0.0) == "ok"
assert await fake.send(b"\x01", delay=0.0) == "ok"
def test_install_command_serializer_wraps_and_is_idempotent(tmp_path):
bot = _make_bot(tmp_path)
async def test_reply_wait_does_not_hold_the_lock(tmp_path):
"""A command waiting for a remote reply must not block other frames."""
bot = _make_bot(tmp_path, min_interval_ms=0)
fake = FakeCommands()
_serialize_command_frames(bot, fake)
waiting = asyncio.create_task(fake.send_and_wait_for_reply(b"\x02", wait=1.0))
await asyncio.sleep(0.05) # first frame written, now waiting for the reply
start = time.monotonic()
assert await fake.send(b"\x03") == "ok"
assert time.monotonic() - start < 0.5
assert not waiting.done()
waiting.cancel()
async def test_send_msg_with_retry_ack_wait_does_not_block_other_commands(tmp_path):
"""meshcore's own retry loop sends through the wrapped send(), outside the lock."""
bot = _make_bot(tmp_path, min_interval_ms=0)
dispatcher = EventDispatcher()
await dispatcher.start()
try:
handler = CommandHandler()
handler.set_dispatcher(dispatcher)
contact = {"public_key": "ab" * 32, "out_path_len": 0, "out_path": ""}
handler._get_contact_by_prefix = lambda prefix: contact
frames = []
async def radio(data):
data = bytes(data)
frames.append(data[0])
if data[0] == 0x02: # CMD_SEND_TXT_MSG: accepted, but never ACKed
await dispatcher.dispatch(Event(
EventType.MSG_SENT,
{"type": 0, "expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 1000},
))
else:
await dispatcher.dispatch(Event(EventType.OK, {}))
handler._sender_func = radio
_serialize_command_frames(bot, handler)
dm = asyncio.create_task(handler.send_msg_with_retry(contact, "hi", max_attempts=1))
await asyncio.sleep(0.1) # DM sent, waiting for its ACK
start = time.monotonic()
result = await asyncio.wait_for(handler.send(b"\x14", [EventType.OK]), timeout=0.5)
assert result.type == EventType.OK
assert time.monotonic() - start < 0.5
assert not dm.done()
assert await dm is None # no ACK arrived
assert frames == [0x02, 0x14]
finally:
await dispatcher.stop()
async def test_radio_session_keeps_a_sequence_together(tmp_path):
"""Frames from another task wait until the session ends; frames inside it don't deadlock."""
bot = _make_bot(tmp_path, min_interval_ms=0)
fake = FakeCommands()
_serialize_command_frames(bot, fake)
async def other_sender():
await asyncio.sleep(0.01)
await fake.send(b"other", delay=0.0)
other = asyncio.create_task(other_sender())
async with bot.radio_session():
await fake.send(b"set-scope")
await asyncio.sleep(0.05) # the other task tries to send here
await fake.send(b"send")
await fake.send(b"restore")
await other
assert fake.frames == [b"set-scope", b"send", b"restore", b"other"]
async def test_radio_session_is_reentrant(tmp_path):
bot = _make_bot(tmp_path, min_interval_ms=0)
fake = FakeCommands()
_serialize_command_frames(bot, fake)
async def nested():
async with bot.radio_session():
async with bot.radio_session():
await fake.send(b"\x01")
await asyncio.wait_for(nested(), timeout=1.0)
assert fake.frames == [b"\x01"]
async def test_install_command_serializer_wraps_and_is_idempotent(tmp_path):
bot = _make_bot(tmp_path, min_interval_ms=0)
bot.meshcore = FakeMeshcore()
cmds = bot.meshcore.commands
bot._install_command_serializer()
wrapped = bot.meshcore.commands
assert isinstance(wrapped, _SerializedCommands)
wrapped = cmds.send
assert getattr(wrapped, "_radio_serialized", False)
assert bot.meshcore.commands is cmds # wrapped in place
# Re-installing must not double-wrap.
# Re-installing must not double-wrap (a nested non-reentrant lock would deadlock).
bot._install_command_serializer()
assert bot.meshcore.commands is wrapped
assert cmds.send is wrapped
assert await asyncio.wait_for(cmds.send(b"\x01"), timeout=1.0) == "ok"
def test_install_command_serializer_noop_without_meshcore(tmp_path):
+2 -2
View File
@@ -1,8 +1,8 @@
"""Tests for zero-hop neighbor discovery (modules/neighbors_discovery.py).
Ported from the meshcore-packet-capture project's test_neighbors.py, minus the
device-lock tests (the bot serialises radio commands globally in
modules/core.py _SerializedCommands, so the module takes no lock of its own).
device-lock tests (the bot serialises radio frames globally in
modules/core.py _serialize_command_frames, so the module takes no lock of its own).
"""
from __future__ import annotations