From ee5daa450e146ef8b720ee4bffa573f3c0119792 Mon Sep 17 00:00:00 2001 From: Stryxion Date: Tue, 2 Jun 2026 08:45:52 +0200 Subject: [PATCH 1/4] fix: host race condition on hci acl data packets --- bumble/host.py | 47 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/bumble/host.py b/bumble/host.py index db7359c1a..6125b2271 100644 --- a/bumble/host.py +++ b/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.100 # 100 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: @@ -1032,6 +1038,16 @@ def on_hci_acl_data_packet(self, packet: hci.HCI_AclDataPacket) -> None: if connection := self.connections.get(packet.connection_handle): 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. @@ -1119,6 +1135,37 @@ 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, acl_timed_out = [], [] + 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.critical( + f"Dropping stale buffered ACL packet for handle {handle}" + f" (age {(now - ts) * self._ACL_STALE_TIMEOUT_S * 10000} ms > " + f"{self._ACL_STALE_TIMEOUT_S * self._ACL_STALE_TIMEOUT_S * 10000} ms timeout)" + ) + if acl_to_keep: + self._pending_acl[handle] = acl_to_keep + else: + del self._pending_acl[handle] + + def _drain_buffered_acl(self, handle: int, *_args: Any, **_kwargs: Any) -> None: + """Once le connection happen emit the buffered packets""" + 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 ): From 4864e7ea9fc2be6498b133c8dd5f4bf4596302a0 Mon Sep 17 00:00:00 2001 From: Stryxion Date: Mon, 8 Jun 2026 13:59:10 +0200 Subject: [PATCH 2/4] fix: rebase + _drain_buffered_acl async --- bumble/host.py | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/bumble/host.py b/bumble/host.py index 6125b2271..391bd5dea 100644 --- a/bumble/host.py +++ b/bumble/host.py @@ -1038,6 +1038,7 @@ def on_hci_acl_data_packet(self, packet: hci.HCI_AclDataPacket) -> None: if connection := self.connections.get(packet.connection_handle): connection.on_hci_acl_data_packet(packet) return + logger.warning( f"ACL data arrived before Connection Complete for handle {packet.connection_handle} — buffering" ) @@ -1141,11 +1142,16 @@ async def _clean_pending_acl_to_old(self) -> None: await asyncio.sleep(self._ACL_STALE_TIMEOUT_S) now = time.monotonic() for handle in list(self._pending_acl): - acl_to_keep, acl_timed_out = [], [] + 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)) + ( + 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.critical( + logger.info( f"Dropping stale buffered ACL packet for handle {handle}" f" (age {(now - ts) * self._ACL_STALE_TIMEOUT_S * 10000} ms > " f"{self._ACL_STALE_TIMEOUT_S * self._ACL_STALE_TIMEOUT_S * 10000} ms timeout)" @@ -1155,8 +1161,15 @@ async def _clean_pending_acl_to_old(self) -> None: else: del self._pending_acl[handle] - def _drain_buffered_acl(self, handle: int, *_args: Any, **_kwargs: Any) -> None: - """Once le connection happen emit the buffered packets""" + 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 From f88e32ebbcdc80bd837d6bfe796c370cb6d6c0dc Mon Sep 17 00:00:00 2001 From: Stryxion Date: Mon, 8 Jun 2026 15:06:42 +0200 Subject: [PATCH 3/4] fix: increase timing --- bumble/host.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/bumble/host.py b/bumble/host.py index 391bd5dea..cc24fcdff 100644 --- a/bumble/host.py +++ b/bumble/host.py @@ -257,7 +257,7 @@ 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.100 # 100 ms + _ACL_STALE_TIMEOUT_S: float = 0.500 # 500 ms def __init__( self, @@ -1153,8 +1153,8 @@ async def _clean_pending_acl_to_old(self) -> None: 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 * 10000} ms > " - f"{self._ACL_STALE_TIMEOUT_S * self._ACL_STALE_TIMEOUT_S * 10000} ms timeout)" + 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 From ffb2a6854117aef78699c04fdbe1a546c288a982 Mon Sep 17 00:00:00 2001 From: Stryxion Date: Tue, 9 Jun 2026 09:52:57 +0200 Subject: [PATCH 4/4] fix: review cleanup task + logging + unnecessary sleep --- bumble/host.py | 44 ++++++++------------------------------------ 1 file changed, 8 insertions(+), 36 deletions(-) diff --git a/bumble/host.py b/bumble/host.py index cc24fcdff..f8f524f04 100644 --- a/bumble/host.py +++ b/bumble/host.py @@ -297,7 +297,7 @@ def __init__( 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._drain_acl_buffer_task: asyncio.Task[None] | None = None self.on("le_connection", self._drain_buffered_acl) # Connect to the source and sink if specified @@ -1045,10 +1045,6 @@ def on_hci_acl_data_packet(self, packet: hci.HCI_AclDataPacket) -> None: 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. @@ -1136,48 +1132,24 @@ 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_acl_buffer_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 + now = time.monotonic() + logger.info(f"Replaying buffered ACL packets for handle: {handle}") 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) + # Sleep to allow other tasks to run + await asyncio.sleep(0) + if (now - ts) <= self._ACL_STALE_TIMEOUT_S: + self.on_hci_acl_data_packet(packet) def on_command_processed( self, event: hci.HCI_Command_Complete_Event | hci.HCI_Command_Status_Event