diff --git a/src/defib/agent/client.py b/src/defib/agent/client.py index 1f945cb..c3e923e 100644 --- a/src/defib/agent/client.py +++ b/src/defib/agent/client.py @@ -34,9 +34,7 @@ RSP_DATA, RSP_INFO, RSP_MEMBW, - RSP_READY, RSP_SCAN, - recv_packet, recv_response, send_packet, wait_for_ready, @@ -414,10 +412,8 @@ async def read_memory( # `agent-read-silent-packet-loss-on-long-uart-streams-2026-05-15`. expected_seq = 0 while True: - cmd, data = await recv_packet(self._transport, timeout=60.0) - if cmd == RSP_READY: - continue - elif cmd == RSP_DATA and len(data) > 2: + cmd, data = await recv_response(self._transport, timeout=60.0) + if cmd == RSP_DATA and len(data) > 2: seq = data[0] | (data[1] << 8) if seq != expected_seq: raise RuntimeError( @@ -622,9 +618,7 @@ async def erase_flash( # Receive progress packets (RSP_DATA with sectors_done count) for _ in range(num_sectors): - cmd, data = await recv_packet(self._transport, timeout=30.0) - if cmd == RSP_READY: - continue + cmd, data = await recv_response(self._transport, timeout=30.0) if cmd == RSP_DATA and len(data) >= 2: done = data[0] | (data[1] << 8) if on_progress: @@ -714,10 +708,8 @@ async def write_flash( # Wait for progress — agent sends RSP_DATA for both data and # skipped sectors (immediately for skipped, after receive for data). while True: - cmd, data_pkt = await recv_packet(self._transport, timeout=120.0) - if cmd == RSP_READY: - continue - elif cmd == RSP_DATA: + cmd, data_pkt = await recv_response(self._transport, timeout=120.0) + if cmd == RSP_DATA: done = data_pkt[0] | (data_pkt[1] << 8) total_steps = data_pkt[2] | (data_pkt[3] << 8) if on_progress and total_steps > 0: @@ -733,10 +725,8 @@ async def write_flash( # Wait for final ACK (agent finishes programming last sector) while True: - cmd, data_pkt = await recv_packet(self._transport, timeout=120.0) - if cmd == RSP_READY: - continue - elif cmd == RSP_ACK: + cmd, data_pkt = await recv_response(self._transport, timeout=120.0) + if cmd == RSP_ACK: return data_pkt[0] == ACK_OK elif cmd == RSP_DATA: continue # Late progress packet @@ -873,9 +863,7 @@ async def scan_flash( sectors: list[SectorResult] = [] while len(sectors) < num_sectors: - cmd, data = await recv_packet(self._transport, timeout=30.0) - if cmd == RSP_READY: - continue + cmd, data = await recv_response(self._transport, timeout=30.0) if cmd == RSP_ACK: break if cmd != RSP_SCAN: diff --git a/src/defib/agent/protocol.py b/src/defib/agent/protocol.py index 6dacee6..e64fb65 100644 --- a/src/defib/agent/protocol.py +++ b/src/defib/agent/protocol.py @@ -267,21 +267,22 @@ def _consume(data: bytes) -> tuple[int, bytes] | None: async def recv_response(transport: Transport, timeout: float = 5.0) -> tuple[int, bytes]: """Receive a response packet, skipping stale READY packets. - Timeout resets after each READY skip — READY packets are expected - background traffic and shouldn't consume the timeout budget. + READY is skipped, but it never extends the wait. The agent sends READY + only from its idle loop (agent/main.c, every ~2 s with no command in + hand), so one that arrives after a command was sent means the agent is + idle: it never got the command. Resetting the deadline on each READY + turned a lost command into a wait that could not end (an `agent flash` + sat 12 minutes on hi3536dv100 with nothing written). """ import time deadline = time.monotonic() + timeout - while time.monotonic() < deadline: + while True: remaining = deadline - time.monotonic() if remaining <= 0: break cmd, data = await recv_packet(transport, remaining) if cmd != RSP_READY: return cmd, data - # Reset deadline after READY — don't let background READYs - # consume the timeout meant for the actual response - deadline = time.monotonic() + timeout raise TransportTimeout(f"No response within {timeout}s") diff --git a/tests/test_flash_stream_regression.py b/tests/test_flash_stream_regression.py index 861441e..6f9d365 100644 --- a/tests/test_flash_stream_regression.py +++ b/tests/test_flash_stream_regression.py @@ -4,7 +4,8 @@ 1. COBS trailing zero strip: ~1/256 packets lost trailing 0x00 from CRC32 2. CRC32 high-byte UB: uint8_t << 24 is undefined when value >= 128 3. _recv_packet_sync partial frame stashing: corrupted next call -4. recv_response READY timeout: READY packets consumed timeout budget +4. recv_response READY timeout: READY must neither crowd out a response nor + extend the wait (an idle agent's READYs once made it endless) 5. FMC memory window 1MB wrap: reads repeated every 1MB (masking write bugs) 6. Sector bitmap: host and agent must agree on bit ordering """ @@ -191,45 +192,52 @@ def test_interleaved_ready_and_data(self): # --------------------------------------------------------------------------- -# Bug 3: recv_response timeout consumed by READY packets -# Multiple READY packets consumed the timeout budget meant for the actual -# response. Fix: reset deadline after each READY skip. +# Bug 3: recv_response and READY packets +# READYs queued before a response must not crowd it out (the original bug), +# and READYs must never extend the wait either: the agent sends READY only +# from its idle loop, so a READY after a command means the command was lost. +# Resetting the deadline on each READY turned that into a wait with no end: +# `defib agent flash` sat 12 minutes on hi3536dv100 with nothing written. # --------------------------------------------------------------------------- -class TestRecvResponseReadyTimeout: - def test_ready_does_not_consume_timeout(self): - """recv_response with many READYs followed by ACK must succeed. +class _BeaconTransport: + """An agent that lost the command: idle, a READY every 20 ms, forever.""" + + def __init__(self, then: bytes = b"") -> None: + self._then = bytearray(then) + + async def read(self, size: int, timeout: float | None = None) -> bytes: + import asyncio + if self._then: + out = bytes(self._then) + self._then.clear() + return out + await asyncio.sleep(0.02) + return make_device_packet(RSP_READY, b"DEFIB") + + async def write(self, data: bytes) -> None: + pass - Bug: READY packets consumed the timeout budget. After receiving - several READYs, there was no time left for the actual response. - Fix: reset deadline after each READY skip. - """ - port = FakePort() - # 5 READY packets then the real response - for _ in range(5): - port.feed(make_device_packet(RSP_READY, b"DEFIB")) - port.feed(make_device_packet(RSP_ACK, bytes([ACK_OK]))) - # recv_response uses _recv_packet_sync when transport has _port - # so we test via the sync path directly +class TestRecvResponseReadyTimeout: + @pytest.mark.asyncio + async def test_queued_readys_do_not_crowd_out_the_response(self): + from defib.agent.protocol import recv_response + stale = b"".join(make_device_packet(RSP_READY, b"DEFIB") for _ in range(5)) + t = _BeaconTransport(then=stale + make_device_packet(RSP_ACK, bytes([ACK_OK]))) + cmd, data = await recv_response(t, timeout=1.0) + assert cmd == RSP_ACK and data[0] == ACK_OK + + @pytest.mark.asyncio + async def test_endless_readys_time_out_instead_of_waiting_forever(self): + import asyncio import time - from defib.agent.protocol import RSP_READY as _READY - - deadline = time.monotonic() + 2.0 - result_cmd = None - result_data = None - while time.monotonic() < deadline: - remaining = deadline - time.monotonic() - cmd, data = _recv_packet_sync(port, timeout=remaining) - if cmd != _READY: - result_cmd = cmd - result_data = data - break - # Reset deadline after READY (the fix) - deadline = time.monotonic() + 2.0 - - assert result_cmd == RSP_ACK - assert result_data[0] == ACK_OK + from defib.agent.protocol import recv_response + from defib.transport.base import TransportTimeout + t0 = time.monotonic() + with pytest.raises(TransportTimeout): + await asyncio.wait_for(recv_response(_BeaconTransport(), timeout=0.3), timeout=3.0) + assert time.monotonic() - t0 < 1.5 # ---------------------------------------------------------------------------