mirror of
https://github.com/meshcore-dev/meshcore_py.git
synced 2026-09-30 01:16:40 +00:00
fix(ble): bound the write, and correct two reply-path edge cases
Follow-up to review of the two preceding commits.
The write lock added in 00135bb prevented concurrent writes from dropping the
link, but bounded nothing. CommandHandler's own timeout does not cover the
write: send() awaits _sender_func() and only afterwards arms
asyncio.wait(futures, timeout=...). So a stalled write -- observed on hardware
running to minutes -- held the lock indefinitely while every other command
queued behind it, with nothing logged, no error raised and no DISCONNECTED
event. Before the lock only the stalled command hung; the others went out and
could trip the error-19 disconnect, which at least recovered. The lock turned
a bounded, self-healing failure into an unbounded silent one, and also blocked
the post-reconnect CMD_APP_START behind the dead connection's holder.
The write (lock acquisition included) is now bounded by
BLEConnection.WRITE_TIMEOUT. On expiry the link is torn down rather than the
lock merely released: the underlying CoreBluetooth write may still be in
flight, and a second write racing it re-creates the overlap the lock exists to
prevent. Tearing down hands over to the reconnect path, which is bounded and
self-healing.
The lock is now a lazily-created property, mirroring _mesh_request_lock in
commands/base.py, so an instance built without __init__ still works. The two
BLE tests previously assigned _write_lock themselves, which meant deleting the
__init__ line left them green; there is now a test that __init__ provides it.
Also in send_anon_req:
- A failed change_contact_path() no longer proceeds. The device still has the
contact as flood, so sendAnonReq() floods the request, and the server gates
REGIONS/OWNER/BASIC behind isRouteDirect() and drops it -- the caller then
waits out a full path-scaled timeout for a reply that cannot arrive. It now
returns ERROR path_reset_failed.
- encode_reply_path() clamps to the server's 64-byte reply_path buffer, which
MyMesh.cpp memcpys into with no length check, and rejects hash mode 3 (the
4-byte hops Packet::isValidPathLen refuses). Not a regression -- the previous
encoder overflowed identically -- but this function is the chokepoint and its
comment claimed to bound the read.
- The hop count saturates at 63 instead of being masked with & 63, which would
wrap a 64-hop path to zero hops, i.e. request a zero-hop reply from a distant
node. Not reachable from a device-sourced contact (the reader caps the field
at 63) but silent if it ever were.
- The suggested_timeout multiplier now scales by the hops actually emitted
rather than the contact's claimed out_path_len, which can differ once the
encoder clamps or truncates.
Correction to 30446ed's message: the claim that mode 0 is "byte-for-byte
identical" is wrong. Differentially, over 30000 randomised contact fields
restricted to what a device can actually emit, mode 0 diverges in 1177 of
10118 cases -- every one of them a path containing a 0x00 byte, and in every
one the old encoder was the wrong one. The accurate claim is "unchanged for
mode-0 paths containing no 0x00 byte".
This commit is contained in:
+47
-11
@@ -28,6 +28,11 @@ UART_RX_CHAR_UUID = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"
|
||||
UART_TX_CHAR_UUID = "6E400003-B5A3-F393-E0A9-E50E24DCCA9E"
|
||||
|
||||
class BLEConnection:
|
||||
# Upper bound on a single write (lock acquisition included). Healthy writes
|
||||
# measured 0.06-0.17s against real hardware; observed stalls ran 20s to
|
||||
# minutes, so this preempts them rather than waiting for CoreBluetooth.
|
||||
WRITE_TIMEOUT = 10.0
|
||||
|
||||
def __init__(self, address=None, device=None, client=None, pin=None):
|
||||
"""
|
||||
Constructor: specify address or an existing BleakClient.
|
||||
@@ -53,13 +58,27 @@ class BLEConnection:
|
||||
self.rx_char = None
|
||||
self._disconnect_callback = None
|
||||
self._background_tasks: set[asyncio.Task] = set()
|
||||
# Serialises write_gatt_char(). Two overlapping writes to the same
|
||||
# characteristic drop the link outright (observed on macOS/CoreBluetooth:
|
||||
# "BLE write failed: 19", connection gone). Nothing above this layer
|
||||
# guarantees callers are sequential -- schedulers, health checks and
|
||||
# user commands all issue commands independently -- so the transport has
|
||||
# to enforce it. Lazily created so it binds to the running loop.
|
||||
self._write_lock: Optional[asyncio.Lock] = None
|
||||
self._write_lock_obj: Optional[asyncio.Lock] = None
|
||||
|
||||
@property
|
||||
def _write_lock(self) -> asyncio.Lock:
|
||||
"""Serialises write_gatt_char().
|
||||
|
||||
Two overlapping writes to the same characteristic drop the link outright
|
||||
(observed on macOS/CoreBluetooth: "BLE write failed: 19", connection
|
||||
gone). Nothing above this layer guarantees callers are sequential --
|
||||
schedulers, health checks and user commands all issue independently --
|
||||
so the transport has to enforce it.
|
||||
|
||||
Lazily created so it binds to the running loop, mirroring the
|
||||
_mesh_request_lock property in commands/base.py. Read through getattr so
|
||||
an instance built without __init__ still works.
|
||||
"""
|
||||
lock = getattr(self, "_write_lock_obj", None)
|
||||
if lock is None:
|
||||
lock = asyncio.Lock()
|
||||
self._write_lock_obj = lock
|
||||
return lock
|
||||
|
||||
def _spawn_background(self, coro) -> asyncio.Task:
|
||||
"""Create a tracked background task (prevents GC of fire-and-forget tasks)."""
|
||||
@@ -198,6 +217,10 @@ class BLEConnection:
|
||||
if self.reader is not None:
|
||||
self._spawn_background(self.reader.handle_rx(data))
|
||||
|
||||
async def _write_locked(self, data):
|
||||
async with self._write_lock:
|
||||
await self.client.write_gatt_char(self.rx_char, bytes(data), response=True)
|
||||
|
||||
async def send(self, data):
|
||||
if not self.client:
|
||||
logger.error("Client is not connected")
|
||||
@@ -207,11 +230,24 @@ class BLEConnection:
|
||||
if not self.rx_char:
|
||||
logger.error("RX characteristic not found")
|
||||
return False
|
||||
if self._write_lock is None:
|
||||
self._write_lock = asyncio.Lock()
|
||||
# Bound the whole acquire-plus-write. A stalled write has been seen to
|
||||
# hang for minutes, and CommandHandler's own timeout does not cover this
|
||||
# -- it starts only after _sender_func returns -- so without a bound the
|
||||
# serialising lock would queue every other command behind the stall
|
||||
# indefinitely, with nothing logged and no disconnect raised. Turning one
|
||||
# hung command into a silent whole-client stall would be worse than the
|
||||
# overlap the lock exists to prevent.
|
||||
try:
|
||||
async with self._write_lock:
|
||||
await self.client.write_gatt_char(self.rx_char, bytes(data), response=True)
|
||||
await asyncio.wait_for(self._write_locked(data), timeout=self.WRITE_TIMEOUT)
|
||||
except asyncio.TimeoutError:
|
||||
# Do not simply release and carry on: the underlying write may still
|
||||
# be in flight, and a second write racing it re-creates the exact
|
||||
# overlap that kills the link. Tear the connection down so the
|
||||
# reconnect path takes over -- bounded and self-healing.
|
||||
logger.warning(f"BLE write timed out after {self.WRITE_TIMEOUT}s")
|
||||
if self._disconnect_callback:
|
||||
await self._disconnect_callback("ble_write_timeout")
|
||||
return False
|
||||
except Exception as exc:
|
||||
logger.warning(f"BLE write failed: {exc}")
|
||||
if self._disconnect_callback:
|
||||
|
||||
@@ -57,6 +57,13 @@ def _validate_destination(dst: DestinationType, prefix_length: int = 6) -> bytes
|
||||
)
|
||||
|
||||
|
||||
# Size of the server-side reply_path buffer (uint8_t reply_path[64] in
|
||||
# simple_repeater/MyMesh.h). It is memcpy'd into without a length check.
|
||||
MAX_REPLY_PATH_BYTES = 64
|
||||
# reply_path_len is the low 6 bits of the header byte.
|
||||
MAX_REPLY_PATH_HOPS = 63
|
||||
|
||||
|
||||
def encode_reply_path(out_path_len: int, out_path_hex: str, out_path_hash_mode: int) -> bytes:
|
||||
"""Encode the reply path a server should use when answering us.
|
||||
|
||||
@@ -71,17 +78,36 @@ def encode_reply_path(out_path_len: int, out_path_hex: str, out_path_hash_mode:
|
||||
|
||||
The path itself is reversed by *hop*, not by byte: a return path visits the
|
||||
same hops in the opposite order, and each hop's multi-byte hash must stay
|
||||
intact. (At hash mode 0 the two are indistinguishable, which is why this
|
||||
went unnoticed - mode 0 is the default.)
|
||||
intact. (For single-byte hops the two are indistinguishable, which is most
|
||||
of why this went unnoticed - mode 0 is the default.)
|
||||
"""
|
||||
hash_mode = max(out_path_hash_mode, 0) # -1 means "flood", i.e. no path
|
||||
if hash_mode > 2:
|
||||
# The server computes hash_size = mode + 1, and Packet::isValidPathLen
|
||||
# rejects 4-byte hops outright, so such a path is unusable on the wire.
|
||||
logger.warning(
|
||||
f"Unsupported out_path_hash_mode {out_path_hash_mode}; "
|
||||
"requesting a zero-hop reply path instead"
|
||||
)
|
||||
return b"\x00"
|
||||
hash_size = hash_mode + 1
|
||||
hops = max(out_path_len, 0) & 63
|
||||
# Saturate rather than mask: `& 63` would silently wrap a 64-hop path to
|
||||
# zero hops, i.e. a zero-hop reply for a distant node.
|
||||
hops = min(max(out_path_len, 0), MAX_REPLY_PATH_HOPS)
|
||||
|
||||
raw = bytes.fromhex(out_path_hex or "")
|
||||
# Never read past what the contact actually carries; a truncated or padded
|
||||
# field would otherwise yield short trailing hops.
|
||||
hops = min(hops, len(raw) // hash_size)
|
||||
# The server memcpys into a fixed 64-byte reply_path with no bounds check
|
||||
# (simple_repeater/MyMesh.cpp), so never describe more than fits.
|
||||
max_hops = min(MAX_REPLY_PATH_HOPS, MAX_REPLY_PATH_BYTES // hash_size)
|
||||
if hops > max_hops:
|
||||
logger.warning(
|
||||
f"Reply path of {hops} hops x {hash_size}B exceeds the "
|
||||
f"{MAX_REPLY_PATH_BYTES}B the server can hold; truncating to {max_hops}"
|
||||
)
|
||||
hops = max_hops
|
||||
|
||||
path = b"".join(
|
||||
raw[i * hash_size:(i + 1) * hash_size] for i in range(hops - 1, -1, -1)
|
||||
@@ -362,7 +388,15 @@ class CommandHandlerBase:
|
||||
if contact["out_path_len"] == -1:
|
||||
logger.info("No path set trying zero hop")
|
||||
zero_hop = True
|
||||
await self.change_contact_path(contact, "")
|
||||
path_res = await self.change_contact_path(contact, "")
|
||||
if path_res is not None and path_res.type == EventType.ERROR:
|
||||
# The device still has this contact as flood, so sendAnonReq
|
||||
# will flood the request -- and the server gates REGIONS,
|
||||
# OWNER and BASIC behind isRouteDirect(), silently dropping
|
||||
# it. Better to fail here than to wait out a full timeout
|
||||
# for a reply that cannot come.
|
||||
logger.error("Could not set zero-hop path, aborting anon request")
|
||||
return Event(EventType.ERROR, {"reason": "path_reset_failed"})
|
||||
# update_contact() normally reflects the change back onto the dict, so
|
||||
# out_path_len reads 0 here. Clamp anyway: if that call failed (e.g. the
|
||||
# device query inside it errored) the dict is still -1, and the unsigned
|
||||
@@ -388,7 +422,8 @@ class CommandHandlerBase:
|
||||
|
||||
exp_tag = result.payload["expected_ack"].hex()
|
||||
# Use provided timeout or fallback to suggested timeout (with 5s default)
|
||||
result.payload["suggested_timeout"] = result.payload.get("suggested_timeout", 4000) * (out_path_len + 1) # update timeout from path_len
|
||||
emitted_hops = reply_path[0] & 63 # what actually went on the wire
|
||||
result.payload["suggested_timeout"] = result.payload.get("suggested_timeout", 4000) * (emitted_hops + 1) # update timeout from path_len
|
||||
actual_timeout = timeout if timeout is not None and timeout > 0 else result.payload.get("suggested_timeout", 4000) / 800.0
|
||||
actual_timeout = min_timeout if actual_timeout < min_timeout else actual_timeout
|
||||
self._reader.register_binary_request(pubkey_prefix.hex(), exp_tag, request_type, actual_timeout, context=context, is_anon=True)
|
||||
|
||||
Reference in New Issue
Block a user