Reuse the APS sequence number when retrying sends

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

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,21 +576,21 @@ 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(
     api: RecordingApi, transport: SyntheticBinaryTransport
 ) -> 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
 

tests/test_application.py

@@ -664,7 +664,7 @@
 ) -> None:
     app.add_device(DEVICE_IEEE, DEVICE_NWK)
 
     await app.send_packet(
         aps_packet(
             t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=DEVICE_NWK),
             tx_options=t.TransmitOptions.ACK,
@@ -688,7 +688,7 @@
 async def test_send_packet_unicast_by_ieee(
     app: ControllerApplication, server: SyntheticZiggurat
 ) -> None:
     await app.send_packet(
         aps_packet(t.AddrModeAddress(addr_mode=t.AddrMode.IEEE, address=DEVICE_IEEE))
     )
 
@@ -703,7 +703,7 @@
     app: ControllerApplication, server: SyntheticZiggurat
 ) -> None:
     # An address with no device behind it carries no EUI64 at all
     await app.send_packet(
         aps_packet(t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=t.NWK(0x9999)))
     )
 
@@ -717,7 +717,7 @@
 ) -> None:
     app.add_device(DEVICE_IEEE, DEVICE_NWK)
 
     await app.send_packet(
         aps_packet(
             t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=DEVICE_NWK),
             tx_options=t.TransmitOptions.ACK | t.TransmitOptions.APS_Encryption,
@@ -782,7 +782,7 @@
 async def test_send_packet_sleepy_destination(
     app: ControllerApplication, server: SyntheticZiggurat
 ) -> None:
     await app.send_packet(
         t.ZigbeePacket(
             src=t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=t.NWK(0x0000)),
             src_ep=t.uint8_t(1),
@@ -853,7 +853,7 @@
     device = app.add_device(DEVICE_IEEE, DEVICE_NWK)
     device.relays = t.Relays(relays) if relays is not None else None
 
     await app.send_packet(
         aps_packet(t.AddrModeAddress(addr_mode=t.AddrMode.NWK, address=DEVICE_NWK))
     )
 
@@ -906,10 +906,11 @@
 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 @@
         )
 
 
+@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:

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

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}")