diff --git a/tests/common.py b/tests/common.py index 6e298a6..a40ae90 100644 --- a/tests/common.py +++ b/tests/common.py @@ -98,6 +98,7 @@ def __init__(self) -> None: self.route_table: list[p.RouteEntry] = [] self.beacons: list[p.Beacon] = [] self.captured_packets: list[p.CapturedPacket] = [] + self._aps_seq = 0 self.handlers: dict[p.RequestCommand, Handler] = { p.RequestCommand.RESET: self.on_empty_ok, p.RequestCommand.SHUTDOWN: self.on_empty_ok, @@ -287,7 +288,16 @@ async def on_scan_route_table( p.RequestCommand.SCAN_ROUTE_TABLE, request_id, self.route_table ) - async def on_send_unicast(self, request: p.SendUnicast, request_id: int) -> None: + def assign_aps_seq(self, request: p.SendUnicast) -> t.uint8_t: + """The APS counter a unicast is sent with: the forced one, or the next one.""" + if request.force_aps_seq: + return request.aps_seq + self._aps_seq = (self._aps_seq + 1) % 256 + return t.uint8_t(self._aps_seq) + + async def on_send_unicast( + self, request: p.SendUnicast, request_id: int + ) -> p.SendUnicastResult: # The local handoff is terminal for a no-ack unicast; an ack-requested one # is only confirmed by the end-to-end APS ack that follows. await self.send_notification( @@ -297,7 +307,7 @@ async def on_send_unicast(self, request: p.SendUnicast, request_id: int) -> None await self.send_notification( p.ApsAckConfirm(status=p.SendStatus.SUCCESS), request_id ) - return None + return p.SendUnicastResult(aps_seq=self.assign_aps_seq(request)) async def on_send_broadcast( self, request: p.SendBroadcast, request_id: int diff --git a/tests/test_api.py b/tests/test_api.py index 02f6218..9c3d5f4 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -12,6 +12,9 @@ from zigpy_ziggurat.zigbee import api as api_module, protocol as p from zigpy_ziggurat.zigbee.api import ZigguratApi +# The APS counter the fake firmware sends every unicast with +_APS_SEQ = 0x42 + _Bytes = t.LVList[t.uint8_t, t.uint16_t] RequestT = TypeVar("RequestT", bound=p.Request) Handler = Callable[[p.Request, int], Awaitable[None]] @@ -95,6 +98,13 @@ def ok( body = bytes([p.Status.OK]) + (payload.serialize() if payload else b"") self._on_frame(p.encode_reply(p.FrameType.RESPONSE, command, request_id, body)) + def accept_unicast(self, request_id: int) -> None: + self.ok( + p.RequestCommand.SEND_UNICAST, + request_id, + p.SendUnicastResult(aps_seq=t.uint8_t(_APS_SEQ)), + ) + def error( self, command: p.RequestCommand, request_id: int, status: p.Status ) -> None: @@ -158,7 +168,7 @@ async def _hw_address(self, request: p.Request, request_id: int) -> None: self.ok(request.command, request_id, p.HwAddress(ieee=self.hw_ieee)) async def _send_aps(self, request: p.Request, request_id: int) -> None: - self.ok(request.command, request_id) + self.accept_unicast(request_id) self.send_confirm(request_id) if request.aps_ack: # type: ignore[attr-defined] self.aps_ack_confirm(request_id) @@ -260,7 +270,9 @@ async def test_request_confirmed( api: RecordingApi, transport: SyntheticBinaryTransport ) -> None: """An APS-ack send resolves once the end-to-end APS ack arrives.""" - await api.request_confirmed(_send_aps(aps_ack=True)) + accepted, status = await api.request_confirmed(_send_aps(aps_ack=True)) + assert accepted == p.SendUnicastResult(aps_seq=t.uint8_t(_APS_SEQ)) + assert status == p.SendStatus.SUCCESS assert transport.sent(p.SendUnicast)[-1].aps_ack @@ -272,7 +284,7 @@ async def test_cancel_on_abandon( async def accept_only(request: p.Request, request_id: int) -> None: send_ids.append(request_id) - transport.ok(request.command, request_id) # accepted, never confirmed + transport.accept_unicast(request_id) # accepted, never confirmed transport.handlers[p.RequestCommand.SEND_UNICAST] = accept_only @@ -303,7 +315,7 @@ async def test_cancel_on_timeout( async def accept_only(request: p.Request, request_id: int) -> None: send_ids.append(request_id) - transport.ok(request.command, request_id) # accepted, never confirmed + transport.accept_unicast(request_id) # accepted, never confirmed transport.handlers[p.RequestCommand.SEND_UNICAST] = accept_only @@ -326,7 +338,7 @@ async def test_cancel_skipped_while_closing( async def accept_only(request: p.Request, request_id: int) -> None: send_ids.append(request_id) - transport.ok(request.command, request_id) # accepted, never confirmed + transport.accept_unicast(request_id) # accepted, never confirmed transport.handlers[p.RequestCommand.SEND_UNICAST] = accept_only @@ -376,14 +388,16 @@ async def test_request_confirmed_failure( """The frame is handed off but the end-to-end APS ack never arrives.""" async def ack_timeout(request: p.Request, request_id: int) -> None: - transport.ok(request.command, request_id) + transport.accept_unicast(request_id) transport.send_confirm(request_id, status=p.SendStatus.SUCCESS) transport.aps_ack_confirm(request_id, status=p.SendStatus.APS_ACK_TIMEOUT) transport.handlers[p.RequestCommand.SEND_UNICAST] = ack_timeout - with pytest.raises(DeliveryError, match="APS_ACK_TIMEOUT"): - await api.request_confirmed(_send_aps(aps_ack=True)) + # The failure comes back with the accept response, which a retry needs + accepted, status = await api.request_confirmed(_send_aps(aps_ack=True)) + assert accepted == p.SendUnicastResult(aps_seq=t.uint8_t(_APS_SEQ)) + assert status == p.SendStatus.APS_ACK_TIMEOUT async def test_request_stream( @@ -562,13 +576,13 @@ async def test_confirmed_send_delivery_failure( api: RecordingApi, transport: SyntheticBinaryTransport ) -> None: async def failed_confirm(request: p.Request, request_id: int) -> None: - transport.ok(request.command, request_id) + transport.accept_unicast(request_id) transport.send_confirm(request_id, status=p.SendStatus.ROUTE_DISCOVERY_TIMEOUT) transport.handlers[p.RequestCommand.SEND_UNICAST] = failed_confirm - with pytest.raises(DeliveryError, match="ROUTE_DISCOVERY_TIMEOUT"): - await api.request_confirmed(_send_aps(aps_ack=False)) + _, status = await api.request_confirmed(_send_aps(aps_ack=False)) + assert status == p.SendStatus.ROUTE_DISCOVERY_TIMEOUT async def test_connection_lost_fails_pending_confirm( @@ -576,7 +590,7 @@ async def test_connection_lost_fails_pending_confirm( ) -> None: async def accept_only(request: p.Request, request_id: int) -> None: # Accept the send but never confirm, leaving a pending confirmation. - transport.ok(request.command, request_id) + transport.accept_unicast(request_id) transport.handlers[p.RequestCommand.SEND_UNICAST] = accept_only diff --git a/tests/test_application.py b/tests/test_application.py index 00fdb12..935ef5e 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -906,10 +906,11 @@ async def fail(request: p.SendUnicast, request_id: int) -> None: async def test_send_packet_confirm_failure( app: ControllerApplication, server: SyntheticZiggurat ) -> None: - async def no_route(request: p.SendUnicast, request_id: int) -> None: + async def no_route(request: p.SendUnicast, request_id: int) -> p.SendUnicastResult: await server.send_notification( p.SendConfirm(status=p.SendStatus.ROUTE_DISCOVERY_TIMEOUT), request_id ) + return p.SendUnicastResult(aps_seq=server.assign_aps_seq(request)) server.handlers[p.RequestCommand.SEND_UNICAST] = no_route @@ -919,6 +920,100 @@ async def no_route(request: p.SendUnicast, request_id: int) -> None: ) +@pytest.mark.parametrize( + ("retry_request_id", "retry_attempt", "force_aps_seq", "aps_seq"), + [ + # A retry of the undelivered request resends with its APS counter (the first + # one the server assigns) + (7, 1, 1, 1), + # Another request's attempt gets a fresh counter + (8, 1, 0, 0), + # A first attempt is a new request, even with the id of an old one + (7, 0, 0, 0), + ], +) +async def test_send_packet_retry_reuses_aps_seq( + app: ControllerApplication, + server: SyntheticZiggurat, + retry_request_id: int, + retry_attempt: int, + force_aps_seq: int, + aps_seq: int, +) -> None: + """A retry of an undelivered unicast resends it with the same APS counter, so a + device that got it but lost its ack drops the retry as a duplicate.""" + statuses = [p.SendStatus.APS_ACK_TIMEOUT, p.SendStatus.SUCCESS] + + async def confirm(request: p.SendUnicast, request_id: int) -> p.SendUnicastResult: + await server.send_notification( + p.SendConfirm(status=statuses.pop(0)), request_id + ) + return p.SendUnicastResult(aps_seq=server.assign_aps_seq(request)) + + server.handlers[p.RequestCommand.SEND_UNICAST] = confirm + packet = aps_packet( + t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=DEVICE_NWK) + ).replace(request_id=7) + + with pytest.raises(DeliveryError, match="APS_ACK_TIMEOUT"): + await app.send_packet(packet) + + await app.send_packet( + packet.replace(request_id=retry_request_id, attempt=retry_attempt) + ) + + retry = server.sent(p.SendUnicast)[-1] + assert retry.force_aps_seq == force_aps_seq + assert retry.aps_seq == aps_seq + + +async def test_send_packet_retry_after_expiry_gets_fresh_aps_seq( + app: ControllerApplication, + server: SyntheticZiggurat, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An undelivered attempt's APS counter is forgotten once it is too old.""" + monkeypatch.setattr( + application_module, "UNDELIVERED_APS_SEQ_LIFETIME", timedelta(0) + ) + statuses = [p.SendStatus.APS_ACK_TIMEOUT, p.SendStatus.SUCCESS] + + async def confirm(request: p.SendUnicast, request_id: int) -> p.SendUnicastResult: + await server.send_notification( + p.SendConfirm(status=statuses.pop(0)), request_id + ) + return p.SendUnicastResult(aps_seq=server.assign_aps_seq(request)) + + server.handlers[p.RequestCommand.SEND_UNICAST] = confirm + packet = aps_packet( + t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=DEVICE_NWK) + ).replace(request_id=7) + + with pytest.raises(DeliveryError, match="APS_ACK_TIMEOUT"): + await app.send_packet(packet) + + await app.send_packet(packet.replace(attempt=1)) + + retry = server.sent(p.SendUnicast)[-1] + assert not retry.force_aps_seq + + +async def test_send_packet_retry_after_delivery_gets_fresh_aps_seq( + app: ControllerApplication, server: SyntheticZiggurat +) -> None: + """A retry after a delivered attempt (zigpy saw no reply) gets a fresh APS + counter: with the same one the device would drop it as a duplicate.""" + packet = aps_packet( + t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=DEVICE_NWK) + ).replace(request_id=7) + + await app.send_packet(packet) + await app.send_packet(packet.replace(attempt=1)) + + retry = server.sent(p.SendUnicast)[-1] + assert not retry.force_aps_seq + + async def test_add_endpoint( app: ControllerApplication, server: SyntheticZiggurat ) -> None: diff --git a/zigpy_ziggurat/zigbee/api.py b/zigpy_ziggurat/zigbee/api.py index d2c4102..08e3022 100644 --- a/zigpy_ziggurat/zigbee/api.py +++ b/zigpy_ziggurat/zigbee/api.py @@ -7,7 +7,6 @@ from datetime import timedelta import logging -from zigpy.exceptions import DeliveryError import zigpy.types as t from zigpy_ziggurat.zigbee import protocol as p @@ -125,13 +124,13 @@ async def request(self, request: p.Request) -> p.Response | None: async def request_confirmed( self, send: p.SendUnicast | p.SendBroadcast | p.SendGroupcast - ) -> None: - """Send and await the terminal confirmation.""" + ) -> tuple[p.Response | None, p.SendStatus]: + """Send; return the accept response and the terminal confirmation status.""" # The terminal confirmation is the end-to-end APS ack for an ack-requested # unicast, the passive-ack quorum for a broadcast/groupcast, otherwise the local - # handoff. A rejected frame raises `DeliveryError` before any confirm; a failed - # confirmation raises it too. + # handoff. A rejected frame raises `DeliveryError` before any confirm. A failed + # confirmation is returned, not raised: a retry needs the accept response. request_id = self._next_id() pending = _Pending(send, streaming=False) self._pending[request_id] = pending @@ -148,7 +147,7 @@ async def request_confirmed( try: async with asyncio.timeout(CONFIRM_TIMEOUT): await self._transport.send_frame(p.encode_request(send, request_id)) - await pending.response # accepted / rejected + accepted = await pending.response result = await confirm finally: self._pending.pop(request_id, None) @@ -158,8 +157,7 @@ async def request_confirmed( if not confirm.done() or confirm.cancelled(): await self._cancel_send(request_id) - if result.status != p.SendStatus.SUCCESS: - raise DeliveryError(f"Send failed: {result.status.name}") + return accepted, result.status async def request_stream( self, request: p.Request diff --git a/zigpy_ziggurat/zigbee/application.py b/zigpy_ziggurat/zigbee/application.py index 7a5ce36..367fdd5 100644 --- a/zigpy_ziggurat/zigbee/application.py +++ b/zigpy_ziggurat/zigbee/application.py @@ -62,6 +62,11 @@ def _max_concurrent_requests(url: str) -> int: # How long route hints are supplied to the stack, to reduce startup churn. ROUTE_HINT_DURATION = timedelta(minutes=10) +# How long an undelivered unicast's APS counter is kept for its retry. zigpy retries +# within a fraction of a second, and a device forgets the counter after its duplicate +# rejection window anyway. +UNDELIVERED_APS_SEQ_LIFETIME = timedelta(seconds=15) + def logistic(x: float, *, L: float = 1, x_0: float = 0, k: float = 1) -> float: """Logistic function.""" @@ -107,6 +112,9 @@ def __init__(self, config: dict[str, Any]) -> None: super().__init__(config) self._api: ZigguratApi | None = None self._start_time: datetime | None = None + # The APS counter of each zigpy request whose last attempt was accepted but not + # delivered, and when it was stored. The oldest entries come first + self._undelivered_aps_seqs: dict[int, tuple[t.uint8_t, datetime]] = {} async def connect(self) -> None: # The device path is either the WebSocket URL of a ziggurat server or the @@ -922,6 +930,27 @@ async def send_packet(self, packet: t.ZigbeePacket) -> None: route_control = p.RouteControl.HINT_NEXT_HOP next_hop = maybe_next_hop + # Clean up the APS sequence mapping to keep it from growing unbounded + now = datetime.now(UTC) + while self._undelivered_aps_seqs: + oldest = next(iter(self._undelivered_aps_seqs)) + _, stored_at = self._undelivered_aps_seqs[oldest] + + if now - stored_at < UNDELIVERED_APS_SEQ_LIFETIME: + break + + del self._undelivered_aps_seqs[oldest] + + # A retry reuses the APS counter of the undelivered attempt before it, + # so a device that got the frame but lost its ack drops the retry as a + # duplicate + aps_seq = None + if ( + packet.attempt > 0 + and packet.request_id in self._undelivered_aps_seqs + ): + aps_seq, _ = self._undelivered_aps_seqs.pop(packet.request_id) + send = p.SendUnicast.build( destination=destination, destination_eui64=destination_eui64, @@ -940,6 +969,18 @@ async def send_packet(self, packet: t.ZigbeePacket) -> None: next_hop=next_hop, relays=relays, asdu=asdu, + aps_seq=aps_seq, ) - await self._api.request_confirmed(send) + accepted, status = await self._api.request_confirmed(send) + + if status == p.SendStatus.SUCCESS: + return + + if packet.request_id is not None and isinstance(accepted, p.SendUnicastResult): + self._undelivered_aps_seqs[packet.request_id] = ( + accepted.aps_seq, + datetime.now(UTC), + ) + + raise DeliveryError(f"Send failed: {status.name}") diff --git a/zigpy_ziggurat/zigbee/protocol.py b/zigpy_ziggurat/zigbee/protocol.py index f5c6f87..6472184 100644 --- a/zigpy_ziggurat/zigbee/protocol.py +++ b/zigpy_ziggurat/zigbee/protocol.py @@ -96,6 +96,10 @@ class CancelResult(Response, wire.CancelResultPayload): pass +class SendUnicastResult(Response, wire.SendUnicastResultPayload): + pass + + class EnergyResult(Response, wire.EnergyResultPayload): pass @@ -190,6 +194,7 @@ class ScanRouteTable(Request): class SendUnicast(Request, wire.SendUnicastPayload): command = RequestCommand.SEND_UNICAST + response = SendUnicastResult @classmethod def build( @@ -210,13 +215,15 @@ def build( route_control: RouteControl = RouteControl.STACK_DECIDES, next_hop: t.NWK | None = None, relays: list[t.NWK] | None = None, + aps_seq: int | None = None, ) -> SendUnicast: return cls( has_eui64=t.uint1_t(destination_eui64 is not None), aps_ack=t.uint1_t(aps_ack), aps_encryption=t.uint1_t(aps_encryption), sleepy_destination=t.uint1_t(sleepy_destination), - reserved=t.uint4_t(0), + force_aps_seq=t.uint1_t(aps_seq is not None), + reserved=t.uint3_t(0), # 0xFFFE stands in for "no short address"; the firmware resolves the EUI64. destination=destination if destination is not None else t.NWK(0xFFFE), destination_eui64=destination_eui64 or t.EUI64([0] * 8), @@ -224,7 +231,8 @@ def build( cluster_id=t.uint16_t(cluster_id), src_ep=t.uint8_t(src_ep), dst_ep=t.uint8_t(dst_ep), - _aps_seq=t.uint8_t(0), + # `aps_seq` sends with an earlier send's APS counter instead of a fresh one + aps_seq=t.uint8_t(aps_seq if aps_seq is not None else 0), radius=t.uint8_t(radius), priority=t.int8s(priority), route=route_control, diff --git a/zigpy_ziggurat/zigbee/wire.py b/zigpy_ziggurat/zigbee/wire.py index 8efb829..5cb17a4 100644 --- a/zigpy_ziggurat/zigbee/wire.py +++ b/zigpy_ziggurat/zigbee/wire.py @@ -261,14 +261,15 @@ class SendUnicastPayload(t.Struct): aps_ack: t.uint1_t aps_encryption: t.uint1_t sleepy_destination: t.uint1_t - reserved: t.uint4_t + force_aps_seq: t.uint1_t + reserved: t.uint3_t destination: t.NWK destination_eui64: t.EUI64 profile_id: t.uint16_t cluster_id: t.uint16_t src_ep: t.uint8_t dst_ep: t.uint8_t - _aps_seq: t.uint8_t + aps_seq: t.uint8_t radius: t.uint8_t priority: t.int8s route: RouteControl @@ -287,6 +288,10 @@ class SendUnicastPayload(t.Struct): asdu: t.LongOctetString +class SendUnicastResultPayload(t.Struct): + aps_seq: t.uint8_t + + class SendBroadcastPayload(t.Struct): reserved: t.uint8_t destination: t.NWK @@ -430,6 +435,7 @@ class SendStatus(t.enum8): FRAME_BUDGET_EXHAUSTED = 10 CANCELLED = 11 RADIO_ERROR = 12 + HANDOFF_TIMEOUT = 13 class SendConfirmPayload(t.Struct):