fix: host race condition on hci acl data packets

bumble/host.py

@@ -17,6 +17,7 @@
 # -----------------------------------------------------------------------------
 from __future__ import annotations
 
+import time
 import asyncio
 import collections
 import dataclasses
@@ -256,6 +257,8 @@ class Host(utils.EventEmitter):
     long_term_key_provider: Callable[[int, bytes, int], Awaitable[bytes | None]] | None
     link_key_provider: Callable[[hci.Address], Awaitable[bytes | None]] | None
 
+    _ACL_STALE_TIMEOUT_S: float = 0.500  # 500 ms
+
     def __init__(
         self,
         controller_source: TransportSource | None = None,
@@ -293,6 +296,9 @@ def __init__(
         self.link_key_provider = None
         self.pairing_io_capability_provider = None  # Classic only
         self.snooper: Snooper | None = None
+        self._pending_acl: dict[int, list[tuple[float, hci.HCI_AclDataPacket]]] = {}
+        self._cleanup_task: asyncio.Task[None] | None = None
+        self.on("le_connection", self._drain_buffered_acl)
 
         # Connect to the source and sink if specified
         if controller_source:
@@ -1033,6 +1039,17 @@ def on_hci_acl_data_packet(self, packet: hci.HCI_AclDataPacket) -> None:
             connection.on_hci_acl_data_packet(packet)
             return
 
+        logger.warning(
+            f"ACL data arrived before Connection Complete for handle {packet.connection_handle} — buffering"
+        )
+        self._pending_acl.setdefault(packet.connection_handle, []).append(
+            (time.monotonic(), packet)
+        )
+        if self._cleanup_task is None or self._cleanup_task.done():
+            self._cleanup_task = asyncio.get_event_loop().create_task(
+                self._clean_pending_acl_to_old()
+            )
+
         # WORKAROUND: Some controllers (e.g. Intel BE200) send ISO data wrapped in ACL packets
         # using the CIS handle.
         is_cis = packet.connection_handle in self.cis_links
@@ -1119,6 +1136,49 @@ def on_hci_iso_data_packet(self, packet: hci.HCI_IsoDataPacket) -> None:
     def on_l2cap_pdu(self, connection: Connection, cid: int, pdu: bytes) -> None:
         self.emit('l2cap_pdu', connection.handle, cid, pdu)
 
+    async def _clean_pending_acl_to_old(self) -> None:
+        """Periodically discard buffered ACL packets that have exceeded the timeout."""
+        while self._pending_acl:
+            await asyncio.sleep(self._ACL_STALE_TIMEOUT_S)
+            now = time.monotonic()
+            for handle in list(self._pending_acl):
+                acl_to_keep: list[tuple[float, hci.HCI_AclDataPacket]] = []
+                acl_timed_out: list[tuple[float, hci.HCI_AclDataPacket]] = []
+                for ts, packet in self._pending_acl[handle]:
+                    (
+                        acl_to_keep
+                        if (now - ts) <= self._ACL_STALE_TIMEOUT_S
+                        else acl_timed_out
+                    ).append((ts, packet))
+                for ts, _ in acl_timed_out:
+                    logger.info(
+                        f"Dropping stale buffered ACL packet for handle {handle}"
+                        f" (age {(now - ts) * self._ACL_STALE_TIMEOUT_S * 1000} ms > "
+                        f"{self._ACL_STALE_TIMEOUT_S * self._ACL_STALE_TIMEOUT_S * 1000} ms timeout)"
+                    )
+                if acl_to_keep:
+                    self._pending_acl[handle] = acl_to_keep
+                else:
+                    del self._pending_acl[handle]
+
+    def _drain_buffered_acl_sync(self, handle: int, *args: Any, **kwargs: Any) -> None:
+        """Create an async task to replay buffered ACL packet"""
+        self._cleanup_task = asyncio.get_event_loop().create_task(
+            self._drain_buffered_acl(handle)
+        )
+
+    async def _drain_buffered_acl(self, handle: int) -> None:
+        """Replay all buffered ACL packet"""
+        await asyncio.sleep(0.1)
+        queued = self._pending_acl.pop(handle, [])
+        if not queued:
+            return
+        for ts, packet in queued:
+            logger.info(
+                f"Replaying buffered ACL packet for handle handle (age {(time.monotonic() - ts) * 1000} ms)"
+            )
+            self.on_hci_acl_data_packet(packet)
+
     def on_command_processed(
         self, event: hci.HCI_Command_Complete_Event | hci.HCI_Command_Status_Event
     ):