commit 049e0af18a54a77826e59c2066e5738eb9e123d7 parent ba88c5fff1a25f0b731a2bcc92b89931455f26ce Author: Stéphan Kochen <git@stephank.nl> Date: Thu, 03 Sep 2026 20:10:09 +0200 Drain command response streams correctly
diff --git a/custom_components/inlite/lib/inlite_ble/hub.py b/custom_components/inlite/lib/inlite_ble/hub.py index 5edbbc3aaa0c379aee19c68f3ade25d16bf7d484..9a6ac5d0676ab6c9827f3b34e614b56d3795c177 100644 --- a/custom_components/inlite/lib/inlite_ble/hub.py +++ b/custom_components/inlite/lib/inlite_ble/hub.py @@ -113,9 +113,10 @@ self._crypto = CsrMeshCrypto(passphrase) self._ble_address = ble_address self._client: BleakClient | None = None self._loop: asyncio.AbstractEventLoop | None = None - self._ack_event = asyncio.Event() + # Control packets are queued so a response FLUSH arriving immediately + # after a request-completion ACK cannot be lost by clearing an Event. + self._ack_queue: asyncio.Queue[bytes] = asyncio.Queue() self._stream_event = asyncio.Event() - self._last_ack_data = b"" self._last_stream_data = b"" self._stream_buffer = bytearray() self._stream_invalid = False @@ -244,8 +245,14 @@ if loop is None or loop.is_closed(): return if pkt_type == PKT_BLOCK_ACK: - self._last_ack_data = payload - loop.call_soon_threadsafe(self._ack_event.set) + # ACK/FLUSH control packets for a command are unicast. Filter + # before queuing so traffic from another mesh node cannot satisfy + # this hub's active command. + if ( + decrypted["src_id"] == self._device_id + and decrypted["dest_id"] == self._crypto.controller_address + ): + loop.call_soon_threadsafe(self._ack_queue.put_nowait, payload) elif pkt_type == PKT_BLOCK_STREAM: if self._accept_stream_segment(payload): loop.call_soon_threadsafe(self._stream_event.set) @@ -350,16 +357,12 @@ packet = self._crypto.encrypt_packet(dest_id, pkt_type, data) await self._write(packet) async def _wait_ack(self, timeout: float = ACK_TIMEOUT) -> bytes: - """Wait for an ACK notification from the hub. - - The caller must clear _ack_event BEFORE sending the write. - """ + """Return the next queued control packet from this hub.""" try: - await asyncio.wait_for(self._ack_event.wait(), timeout) + return await asyncio.wait_for(self._ack_queue.get(), timeout) except asyncio.TimeoutError: _LOGGER.warning("ACK timeout after %.1fs", timeout) return b"" - return self._last_ack_data async def _send_command( self, opcode: int, command_data: bytes, acknowledged: bool = True @@ -372,16 +375,28 @@ 2. BLK_DATA(offset, opcode, data) → wait ACK with byte count 3. BLK_FLUSH(byte_count) → wait ACK with 'ef' suffix (done) 4. ACK the hub's response """ - return await self._send_raw_stream(build_block_data_payload(opcode, command_data)) + if not await self._send_raw_stream(build_block_data_payload(opcode, command_data)): + return False + # Ordinary commands have a response stream too. Drain it even when + # the contents are only a status acknowledgement. + return await self._receive_response_stream() is not None async def _send_raw_stream(self, block_data: bytes) -> bool: - """Send a raw in-lite block stream and wait for its completion ACK.""" + """Send a request block stream and wait for its completion ACK. + + Association is the observed exception that changes encryption key and + has no response stream; callers of normal commands must subsequently + invoke ``_receive_response_stream``. + """ dest = self._device_id - self._ack_event.clear() + self._stream_event.clear() + self._stream_buffer.clear() + self._stream_invalid = False + while not self._ack_queue.empty(): + self._ack_queue.get_nowait() await self._write_mesh(dest, PKT_BLOCK_FLUSH, build_flush_payload(0)) await self._wait_ack() - self._ack_event.clear() await self._write_mesh(dest, PKT_BLOCK_DATA, block_data) ack = await self._wait_ack() @@ -392,7 +407,6 @@ else: acked_bytes = len(block_data) # Step 3: Flush (end) - self._ack_event.clear() await self._write_mesh(dest, PKT_BLOCK_FLUSH, build_flush_payload(acked_bytes)) ack = await self._wait_ack() @@ -403,11 +417,39 @@ _LOGGER.info("Block stream sent successfully") else: _LOGGER.warning("Block stream: no completion ACK") - # ACK the hub's response flush + return success + + async def _receive_response_stream(self) -> bytes | None: + """Drain the hub's FLUSH → STREAM → FLUSH response exchange.""" + dest = self._device_id + + # The next queued control packet is the hub's response FLUSH(0). + await self._wait_ack(timeout=STREAM_TIMEOUT) await asyncio.sleep(WRITE_DELAY) await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(0)) - return success + try: + await asyncio.wait_for(self._stream_event.wait(), STREAM_TIMEOUT) + except asyncio.TimeoutError: + _LOGGER.warning("STREAM timeout waiting for hub response") + return None + # More contiguous ATT notifications may already be in flight. + await asyncio.sleep(WRITE_DELAY) + if self._stream_invalid: + _LOGGER.warning("Discarding malformed STREAM response") + return None + stream_data = bytes(self._last_stream_data) + await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(len(stream_data))) + + flush = await self._wait_ack(timeout=STREAM_TIMEOUT) + flush_count = ( + flush[0] | (flush[1] << 8) if len(flush) >= 2 else len(stream_data) + ) + await asyncio.sleep(WRITE_DELAY) + # The 0xef terminator is a third byte, not an arithmetic adjustment + # of the 16-bit byte count. + await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(flush_count, end=True)) + return stream_data async def _send_acknowledged_command( self, opcode: int, command_data: bytes @@ -422,76 +464,13 @@ Returns: The STREAM payload bytes, or None on failure. """ - dest = self._device_id - - # Step 1: Send command via block streaming - self._ack_event.clear() - await self._write_mesh(dest, PKT_BLOCK_FLUSH, build_flush_payload(0)) - await self._wait_ack() - block_data = build_block_data_payload(opcode, command_data) - self._ack_event.clear() - await self._write_mesh(dest, PKT_BLOCK_DATA, block_data) - ack = await self._wait_ack() - - if len(ack) >= 2: - acked_bytes = ack[0] | (ack[1] << 8) - else: - acked_bytes = len(block_data) - - self._ack_event.clear() - await self._write_mesh(dest, PKT_BLOCK_FLUSH, build_flush_payload(acked_bytes)) - ack = await self._wait_ack() - - success = len(ack) >= 3 and ack[-1] == 0xEF - if not success: + if not await self._send_raw_stream(block_data): _LOGGER.warning("Acknowledged command 0x%04X: no completion ACK", opcode) return None - - # Step 2: Hub sends FLUSH(0) — ACK it - self._ack_event.clear() - ack = await self._wait_ack(timeout=STREAM_TIMEOUT) - # ACK the hub's flush with 0 - await asyncio.sleep(WRITE_DELAY) - await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(0)) - - # Step 3: Hub sends STREAM(data) — collect and ACK - self._stream_event.clear() - self._stream_buffer.clear() - self._stream_invalid = False - try: - await asyncio.wait_for(self._stream_event.wait(), STREAM_TIMEOUT) - except asyncio.TimeoutError: - _LOGGER.warning("STREAM timeout for command 0x%04X", opcode) - return None - - # Consecutive stream segments can already be in flight from the BLE - # bridge. Collect them briefly; offsets are validated on arrival. - await asyncio.sleep(WRITE_DELAY) - stream_data = self._last_stream_data - if self._stream_invalid: - _LOGGER.warning("Discarding malformed STREAM response") + stream_data = await self._receive_response_stream() + if stream_data is None: return None - - stream_bytes = len(stream_data) - - # ACK the stream data - await asyncio.sleep(WRITE_DELAY) - await self._write_mesh(dest, PKT_BLOCK_ACK, build_ack_payload(stream_bytes)) - - # Step 4: Hub sends FLUSH(n) — ACK with completion marker - self._ack_event.clear() - ack = await self._wait_ack(timeout=STREAM_TIMEOUT) - if len(ack) >= 2: - flush_count = ack[0] | (ack[1] << 8) - else: - flush_count = stream_bytes - await asyncio.sleep(WRITE_DELAY) - await self._write_mesh( - dest, PKT_BLOCK_ACK, - build_ack_payload(flush_count + 0xEF, end=False) - ) - _LOGGER.info("Acknowledged command 0x%04X: got %d bytes", opcode, len(stream_data)) # Restore the stream's offset prefix expected by the response parser. return b"\x00\x00" + bytes(stream_data) diff --git a/tests/test_hub.py b/tests/test_hub.py index 00193d4cd5bc4018541573208e821f9b76bcf5ac..c44839b2c1a1a1f729d305f7e6dc80c82733cb48 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -70,6 +70,28 @@ assert hub._accept_stream_segment(b"\x00\x00abc") assert not hub._accept_stream_segment(b"\x04\x00de") assert hub._stream_invalid is True + def test_response_stream_uses_literal_completion_marker(self) -> None: + """The final ACK is count LE followed by ef, as captured from the app.""" + async def run() -> list[tuple[int, bytes]]: + hub = InliteHub(device_id=1, passphrase="test") + writes: list[tuple[int, bytes]] = [] + + async def write_mesh(_dest: int, packet_type: int, data: bytes) -> None: + writes.append((packet_type, data)) + + hub._write_mesh = write_mesh # type: ignore[method-assign] + hub._ack_queue.put_nowait(b"\x00\x00") + hub._ack_queue.put_nowait(b"\x03\x00") + hub._stream_buffer.extend(b"abc") + hub._last_stream_data = b"abc" + hub._stream_event.set() + + assert await hub._receive_response_stream() == b"abc" + return writes + + writes = asyncio.run(run()) + assert writes == [(0x72, b"\x00\x00"), (0x72, b"\x03\x00"), (0x72, b"\x03\x00\xef")] + def test_teach_in_tlv_is_forwarded(self) -> None: received: list[tuple[int, int]] = [] hub = InliteHub(