usb: Preserve HCI event and ACL packet ordering in UsbPacketSource

bumble/transport/usb.py

@@ -22,6 +22,7 @@
 import logging
 import platform
 import threading
+from collections import deque
 from collections.abc import Callable
 from typing import Any
 
@@ -492,8 +493,22 @@ def __init__(self, emit: Callable[[bytes], Any]) -> None:
         super().__init__(length_offset=2, length_size=2, emit=emit)
 
 
+# Default USB interrupt polling sync window (4 ms, matching 1 kernel jiffy at HZ=250
+# in Linux btusb.c force_poll_sync).
+DEFAULT_INTR_INTERVAL = 0.004
+
+
 class UsbPacketSource(asyncio.Protocol, BaseSource):
-    def __init__(self, device, metadata, interrupt_in, bulk_in, isochronous_in):
+
+    def __init__(
+        self,
+        device,
+        metadata,
+        interrupt_in,
+        bulk_in,
+        isochronous_in,
+        intr_interval: float | None = None,
+    ):
         super().__init__()
         self.device = device
         self.metadata = metadata
@@ -502,11 +517,23 @@ def __init__(self, device, metadata, interrupt_in, bulk_in, isochronous_in):
         self.bulk_in = bulk_in
         self.bulk_in_transfer = None
         self.isochronous_in = isochronous_in
-        self.isochronous_in_transfers = []
+        self.isochronous_in_transfers: list[Any] = []
         self.loop = asyncio.get_running_loop()
-        self.queue = asyncio.Queue()
+        self.queue: asyncio.Queue[bytes] = asyncio.Queue()
+        self.acl_queue: deque[tuple[float, bytes]] = deque()
+        self.acl_flush_handle: asyncio.TimerHandle | None = None
+        if intr_interval is not None:
+            self.intr_interval = float(intr_interval)
+        else:
+            raw_interval = getattr(interrupt_in, 'getInterval', lambda: None)()
+            if isinstance(raw_interval, (int, float)) and raw_interval > 0:
+                self.intr_interval = max(
+                    float(raw_interval) / 1000.0, DEFAULT_INTR_INTERVAL
+                )
+            else:
+                self.intr_interval = DEFAULT_INTR_INTERVAL
         self.dequeue_task = None
-        self.done = {}
+        self.done: dict[Any, asyncio.Event] = {}
         self.splitters = {
             hci.HCI_EVENT_PACKET: EventPacketSplitter(
                 lambda packet: self.queue_packet(hci.HCI_EVENT_PACKET, packet)
@@ -560,10 +587,73 @@ def start(self):
 
         self.dequeue_task = self.loop.create_task(self.dequeue())
 
+    def _flush_acl_queue(self) -> None:
+        if self.acl_flush_handle is not None:
+            self.acl_flush_handle.cancel()
+            self.acl_flush_handle = None
+        while self.acl_queue:
+            _, packet = self.acl_queue.popleft()
+            self.queue.put_nowait(packet)
+
+    def _on_acl_timer(self) -> None:
+        self.acl_flush_handle = None
+        now = self.loop.time()
+        # Pop the head packet unconditionally because asyncio.call_at may fire
+        # slightly before the nominal deadline due to OS timer resolution.
+        if self.acl_queue:
+            _, packet = self.acl_queue.popleft()
+            self.queue.put_nowait(packet)
+        while self.acl_queue and self.acl_queue[0][0] <= now:
+            _, packet = self.acl_queue.popleft()
+            self.queue.put_nowait(packet)
+        if self.acl_queue:
+            self.acl_flush_handle = self.loop.call_at(
+                self.acl_queue[0][0], self._on_acl_timer
+            )
+
+    def _enqueue_packet(self, packet_type: int, packet_data: bytes) -> None:
+        if self.closed:
+            return
+
+        packet = bytes([packet_type]) + packet_data
+        if packet_type == hci.HCI_ACL_DATA_PACKET and self.intr_interval > 0:
+            # Defer each ACL packet by its own intr_interval deadline so that
+            # any HCI event generated by the controller at the same time (which
+            # waits for the periodic Interrupt IN poll) is delivered first,
+            # without allowing an earlier ACL packet's timer expiry to
+            # prematurely flush a newer ACL packet. Mirrors Linux btusb.c
+            # force_poll_sync / rx_work behavior.
+            deadline = self.loop.time() + self.intr_interval
+            self.acl_queue.append((deadline, packet))
+            if self.acl_flush_handle is None:
+                self.acl_flush_handle = self.loop.call_at(deadline, self._on_acl_timer)
+            return
+
+        if (
+            packet_type == hci.HCI_EVENT_PACKET
+            and self.acl_queue
+            and len(packet_data) > 0
+            and packet_data[0] == hci.HCI_DISCONNECTION_COMPLETE_EVENT
+        ):
+            # A connection cannot receive valid ACL data after Disconnection
+            # Complete. Flush all pending ACL packets (including any on other
+            # active handles, since flushing early is a safe no-op for ordering
+            # once an event arrives) BEFORE delivering Disconnection Complete
+            # so the Host does not tear down the connection handle before
+            # processing the final ACL packet.
+            self._flush_acl_queue()
+            self.queue.put_nowait(packet)
+            return
+
+        # Enqueue HCI Event (or SCO/unbuffered packet) BEFORE flushing deferred
+        # ACL packets, matching upstream btusb patch:
+        # "Bluetooth: btusb: Preserve event and ACL packet ordering".
+        self.queue.put_nowait(packet)
+        if packet_type == hci.HCI_EVENT_PACKET and self.acl_queue:
+            self._flush_acl_queue()
+
     def queue_packet(self, packet_type: int, packet_data: bytes) -> None:
-        _safe_call_soon(
-            self.loop, self.queue.put_nowait, bytes([packet_type]) + packet_data
-        )
+        _safe_call_soon(self.loop, self._enqueue_packet, packet_type, packet_data)
 
     def transfer_callback(self, transfer):
         packet_type = transfer.getUserData()
@@ -631,9 +721,20 @@ async def dequeue(self):
                 except Exception:
                     logger.exception(color('!!! Exception in sink.on_packet', 'red'))
 
+    def on_transport_lost(self):
+        if self.acl_flush_handle is not None:
+            self.acl_flush_handle.cancel()
+            self.acl_flush_handle = None
+        self.acl_queue.clear()
+        super().on_transport_lost()
+
     def close(self):
         with self.lock:
             self.closed = True
+        if self.acl_flush_handle is not None:
+            self.acl_flush_handle.cancel()
+            self.acl_flush_handle = None
+        self.acl_queue.clear()
 
     async def terminate(self):
         self.close()

tests/transport/usb_test.py

@@ -205,3 +205,40 @@ async def test_open_usb_transport_by_index_checks_interfaces(device_class):
         pytest.raises(usb.TransportInitError, match='no compatible interface'),
     ):
         await usb.open_usb_transport('0')
+
+
+@pytest.mark.asyncio
+async def test_usb_packet_source_acl_ordering():
+    source = usb.UsbPacketSource(
+        mock.Mock(),
+        {},
+        mock.Mock(),
+        mock.Mock(),
+        None,
+        intr_interval=0.01,
+    )
+    acl = b'\x00\x28\x02\x00\xaa\xbb'
+    enc_event = b'\x08\x04\x00\x00\x08\x01'
+    disconn_event = bytes(
+        [hci.HCI_DISCONNECTION_COMPLETE_EVENT, 0x04, 0x00, 0x00, 0x08, 0x13]
+    )
+
+    # 1. Event arrives while ACL is deferred -> Event is queued before ACL.
+    source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl)
+    source.splitters[hci.HCI_EVENT_PACKET].feed(enc_event)
+    assert await source.queue.get() == bytes([hci.HCI_EVENT_PACKET]) + enc_event
+    assert await source.queue.get() == bytes([hci.HCI_ACL_DATA_PACKET]) + acl
+
+    # 2. Disconnection Complete arrives while ACL is deferred -> ACL is flushed first.
+    source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl)
+    source.splitters[hci.HCI_EVENT_PACKET].feed(disconn_event)
+    assert await source.queue.get() == bytes([hci.HCI_ACL_DATA_PACKET]) + acl
+    assert await source.queue.get() == bytes([hci.HCI_EVENT_PACKET]) + disconn_event
+
+    # 3. Standalone ACL packet is flushed after intr_interval expires.
+    source.splitters[hci.HCI_ACL_DATA_PACKET].feed(acl)
+    assert source.queue.empty()
+    assert await asyncio.wait_for(source.queue.get(), timeout=1.0) == (
+        bytes([hci.HCI_ACL_DATA_PACKET]) + acl
+    )
+    source.close()