From 081855e8e2b29b5358f89b3c5380c90479d367cc Mon Sep 17 00:00:00 2001 From: Arjen <18398758+Origami74@users.noreply.github.com> Date: Tue, 30 Jun 2026 18:21:53 +0200 Subject: [PATCH] fix(ble): reframe inbound L2CAP stream so packets survive non-SeqPacket backends MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BLE delivery was unreliable on every backend except BlueZ. The receive path assumed one `recv()` returned exactly one whole FIPS packet, which only holds for BlueZ's SOCK_SEQPACKET (boundary-preserving). Android's BluetoothSocket input stream and macOS CoreBluetooth are byte-stream oriented: a read can return a fragment of a packet or several packets coalesced. Under the old loop a fragment was shipped up as a runt packet (rejected by FMP/Noise) and a coalesced tail was silently truncated and dropped — so packets were lost and the transport thrashed. Recover packet boundaries from the byte stream instead of trusting the OS to preserve them. FIPS packets are self-delimiting via the 4-byte FMP common prefix, so this reuses the exact length-prefixed framer TCP already uses (`tcp::stream::read_fmp_packet`) — kept transport-agnostic for this reason. A new `BleStreamRead` adapter turns the datagram-shaped `BleStream` into the `AsyncRead` that framer expects, buffering leftover bytes across reads. The recv future owns its scratch buffer and returns an owned Vec so it is `'static` and storable across `poll_read` calls. One reader is threaded through the pubkey exchange and the receive loop per connection, so bytes a peer coalesces after the 33-byte pubkey stay buffered rather than being lost at the handoff. The pubkey exchange now reads via `read_exact` (reassembles a fragmented pubkey) instead of a brittle single `recv()` with an exact-length check. On BlueZ this is a transparent pass-through (one recv already equals one packet); on stream backends it reassembles. Either way the layer above sees one whole packet per read, identically on every platform. Also bump the Android outbound SEND_QUEUE_CAP from 8 to 32. Swept against the peer speedtest: 8 starved the radio's connection events (~half throughput), 64 bufferbloated TCP, 32 was best (~200/500 kbps up/down). The sweep was noisy and non-monotonic — run-to-run BLE variance (RF, 2M PHY / connection-priority grants) rivals the knob's effect — so 32 is the best-observed value pending re-validation with PHY/interval instrumentation, not a proven optimum. --- src/transport/ble/android_io.rs | 19 ++-- src/transport/ble/mod.rs | 130 ++++++++++++++--------- src/transport/ble/stream_read.rs | 175 +++++++++++++++++++++++++++++++ 3 files changed, 269 insertions(+), 55 deletions(-) create mode 100644 src/transport/ble/stream_read.rs diff --git a/src/transport/ble/android_io.rs b/src/transport/ble/android_io.rs index af21ff1..8f1186b 100644 --- a/src/transport/ble/android_io.rs +++ b/src/transport/ble/android_io.rs @@ -59,12 +59,19 @@ const ANDROID_ADAPTER: &str = "ble0"; /// tolerable since FMP/Noise above retransmits. const CHANNEL_CAP: usize = 256; -/// Bound on the **outbound byte** queue (Rust → Kotlin writer). Kept shallow on -/// purpose: a BLE link's bandwidth-delay product is ~1 packet, so a deep queue -/// only bufferbloats it — RTT balloons to seconds and TCP above can't ramp. A -/// small bound makes `BleStream::send` backpressure (the `SyncSender` blocks), -/// which propagates up through FSP/MMP to the TUN and TCP's own flow control. -const SEND_QUEUE_CAP: usize = 8; +/// Bound on the **outbound byte** queue (Rust → Kotlin writer), fixed at channel +/// creation. A bounded queue makes `BleStream::send` backpressure (the `SyncSender` +/// blocks), propagating flow control up through FSP/MMP to the TUN and TCP rather +/// than letting an unbounded queue bufferbloat the link. +/// +/// 32 is empirical, swept against the peer speedtest: 8 starved the radio's +/// connection events (~half throughput), 64 bufferbloated TCP (regressed), and 32 +/// was best (~200/500 kbps up/down). The sweep was noisy and non-monotonic across +/// single runs, though — run-to-run BLE variance (RF, and whether the OS grants 2M +/// PHY / high connection priority that session) rivals the effect of this knob — so +/// 32 is the best-observed working value, to be re-validated with repeated runs + +/// PHY/interval instrumentation, not a proven optimum. +const SEND_QUEUE_CAP: usize = 32; /// Transport default MTU, used when the OS reports an unknown (0) channel MTU. /// Matches `DEFAULT_BLE_MTU` in `config/transport.rs`. diff --git a/src/transport/ble/mod.rs b/src/transport/ble/mod.rs index 59a8a7a..9925f6a 100644 --- a/src/transport/ble/mod.rs +++ b/src/transport/ble/mod.rs @@ -1,9 +1,11 @@ //! BLE L2CAP Transport Implementation //! //! Provides BLE-based transport for FIPS peer communication using L2CAP -//! Connection-Oriented Channels (CoC) in SeqPacket mode. L2CAP CoC -//! preserves message boundaries (unlike TCP byte streams), so no FMP -//! framing is needed — each send/recv is one FIPS packet. +//! Connection-Oriented Channels (CoC). BlueZ (SeqPacket) preserves message +//! boundaries, but stream-oriented backends (Android `BluetoothSocket`, +//! CoreBluetooth) do not, so the receive path recovers FIPS packet boundaries +//! with the shared FMP framer (`tcp::stream::read_fmp_packet`) over a +//! [`stream_read::BleStreamRead`] adapter — reliable on every platform. //! //! ## Architecture //! @@ -24,6 +26,7 @@ pub mod io; pub mod pool; pub mod psm; pub mod stats; +pub mod stream_read; // The Android backend (radio in Kotlin, bytes over a bridge). Compiled on // Android, and under `cfg(test)` on any host so its channel logic is unit-tested @@ -42,6 +45,9 @@ use discovery::DiscoveryBuffer; use io::{BleIo, BleScanner, BleStream}; use pool::{BleConnection, ConnectionPool}; use stats::BleStats; +use stream_read::BleStreamRead; + +use crate::transport::tcp::stream::{StreamError, read_fmp_packet}; use secp256k1::XOnlyPublicKey; use std::collections::HashMap; @@ -376,9 +382,15 @@ impl BleTransport { } }; + // One buffering reader per connection, shared by the pubkey exchange + // and the receive loop so coalesced bytes are never dropped. + let recv_mtu = stream.recv_mtu(); + let stream = Arc::new(stream); + let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu); + // Pre-handshake pubkey exchange (temporary, pre-XX) if let Some(ref our_pubkey) = self.local_pubkey { - match pubkey_exchange(&stream, our_pubkey).await { + match pubkey_exchange(&mut reader, our_pubkey).await { Ok(peer_pubkey) => { debug!(addr = %addr, "BLE outbound pubkey exchange complete"); self.discovery_buffer @@ -391,24 +403,26 @@ impl BleTransport { } } - self.promote_connection(addr, &ble_addr, stream).await + self.promote_connection(addr, &ble_addr, stream, reader) + .await } /// Promote a newly established stream into the connection pool. /// - /// Spawns the receive loop and inserts into the pool with eviction. + /// Spawns the receive loop (driven by `reader`) and inserts into the pool + /// with eviction. async fn promote_connection( &self, addr: &TransportAddr, ble_addr: &BleAddr, - stream: I::Stream, + stream: Arc, + reader: BleStreamRead, ) -> Result<(), TransportError> { let send_mtu = stream.send_mtu(); let recv_mtu = stream.recv_mtu(); - let stream = Arc::new(stream); let recv_task = tokio::spawn(receive_loop( - Arc::clone(&stream), + reader, addr.clone(), Arc::clone(&self.pool), self.packet_tx.clone(), @@ -496,9 +510,16 @@ impl BleTransport { match result { Ok(Ok(stream)) => { + let send_mtu = stream.send_mtu(); + let recv_mtu = stream.recv_mtu(); + let stream = Arc::new(stream); + // Shared reader: pubkey exchange + receive loop read the + // same buffer so coalesced bytes survive the handoff. + let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu); + // Pre-handshake pubkey exchange (temporary, pre-XX) if let Some(ref our_pubkey) = local_pubkey { - match pubkey_exchange(&stream, our_pubkey).await { + match pubkey_exchange(&mut reader, our_pubkey).await { Ok(peer_pubkey) => { debug!(addr = %addr_clone, "BLE outbound pubkey exchange complete"); discovery_buffer.add_peer_with_pubkey(&ble_addr, peer_pubkey); @@ -513,12 +534,8 @@ impl BleTransport { } } - let send_mtu = stream.send_mtu(); - let recv_mtu = stream.recv_mtu(); - let stream = Arc::new(stream); - let recv_task = tokio::spawn(receive_loop( - Arc::clone(&stream), + reader, addr_clone.clone(), Arc::clone(&pool), packet_tx, @@ -689,31 +706,38 @@ const PUBKEY_EXCHANGE_TIMEOUT_SECS: u64 = 5; /// Exchange public keys over a newly established L2CAP connection. /// -/// Both sides send `[0x00][our_pubkey:32]` and receive the peer's. +/// Both sides send `[0x00][our_pubkey:32]` and receive the peer's. Reads go +/// through the connection's [`BleStreamRead`] buffer so a peer that fragments +/// the 33-byte message (stream-oriented backends) is read correctly, and any +/// bytes it coalesced after the pubkey stay buffered for the receive loop. /// Returns the peer's XOnlyPublicKey on success. async fn pubkey_exchange( - stream: &S, + reader: &mut BleStreamRead, local_pubkey: &[u8; 32], ) -> Result { + use tokio::io::AsyncReadExt; + // Send our pubkey let mut msg = [0u8; PUBKEY_EXCHANGE_SIZE]; msg[0] = PUBKEY_EXCHANGE_PREFIX; msg[1..].copy_from_slice(local_pubkey); - stream.send(&msg).await?; + reader.stream().send(&msg).await?; - // Receive peer's pubkey (with timeout to prevent indefinite blocking) + // Receive peer's pubkey (with timeout to prevent indefinite blocking). + // read_exact reassembles across fragmented recvs and leaves any trailing + // bytes in the reader's buffer. let mut buf = [0u8; PUBKEY_EXCHANGE_SIZE]; let timeout = std::time::Duration::from_secs(PUBKEY_EXCHANGE_TIMEOUT_SECS); - let n = match tokio::time::timeout(timeout, stream.recv(&mut buf)).await { - Ok(result) => result?, + match tokio::time::timeout(timeout, reader.read_exact(&mut buf)).await { + Ok(Ok(_)) => {} + Ok(Err(e)) => { + return Err(TransportError::RecvFailed(format!( + "pubkey exchange: {}", + e + ))); + } Err(_) => return Err(TransportError::Timeout), }; - if n != PUBKEY_EXCHANGE_SIZE { - return Err(TransportError::RecvFailed(format!( - "pubkey exchange: expected {} bytes, got {}", - PUBKEY_EXCHANGE_SIZE, n - ))); - } if buf[0] != PUBKEY_EXCHANGE_PREFIX { return Err(TransportError::RecvFailed(format!( "pubkey exchange: bad prefix 0x{:02X}", @@ -763,10 +787,13 @@ async fn accept_loop( let send_mtu = stream.send_mtu(); let recv_mtu = stream.recv_mtu(); + let stream = Arc::new(stream); + // Shared reader for pubkey exchange + receive loop. + let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu); // Pre-handshake pubkey exchange (temporary, pre-XX) if let Some(ref our_pubkey) = local_pubkey { - match pubkey_exchange(&stream, our_pubkey).await { + match pubkey_exchange(&mut reader, our_pubkey).await { Ok(peer_pubkey) => { debug!(addr = %ta, "BLE inbound pubkey exchange complete"); discovery_buffer.add_peer_with_pubkey(&addr, peer_pubkey); @@ -792,11 +819,9 @@ async fn accept_loop( } } - let stream = Arc::new(stream); - // Spawn receive loop let recv_task = tokio::spawn(receive_loop( - Arc::clone(&stream), + reader, ta.clone(), Arc::clone(&pool), packet_tx.clone(), @@ -841,8 +866,14 @@ async fn accept_loop( } /// Receive loop: reads packets from a BLE stream and delivers to node. -async fn receive_loop( - stream: Arc, +/// +/// Recovers FIPS packet boundaries from the byte stream via the shared FMP +/// framer ([`read_fmp_packet`]) rather than assuming one `recv` is one packet. +/// L2CAP only preserves SDU boundaries on BlueZ (SeqPacket); stream-oriented +/// backends (Android, CoreBluetooth) fragment and coalesce, so the framer's +/// length-prefixed reads are what make delivery reliable across platforms. +async fn receive_loop( + mut reader: BleStreamRead, addr: TransportAddr, pool: Arc>>>, packet_tx: PacketTx, @@ -850,21 +881,21 @@ async fn receive_loop( stats: Arc, recv_mtu: u16, ) { - let mut buf = vec![0u8; recv_mtu as usize]; loop { - match stream.recv(&mut buf).await { - Ok(0) => { - debug!(addr = %addr, "BLE connection closed by peer"); - break; - } - Ok(n) => { - stats.record_recv(n); - let packet = ReceivedPacket::new(transport_id, addr.clone(), buf[..n].to_vec()); - if packet_tx.send(packet).await.is_err() { + match read_fmp_packet(&mut reader, recv_mtu).await { + Ok(packet) => { + stats.record_recv(packet.len()); + let received = ReceivedPacket::new(transport_id, addr.clone(), packet); + if packet_tx.send(received).await.is_err() { trace!("BLE packet_tx closed, stopping receive loop"); break; } } + // Clean peer close (EOF at a packet boundary) is expected, not an error. + Err(StreamError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => { + debug!(addr = %addr, "BLE connection closed by peer"); + break; + } Err(e) => { debug!(addr = %addr, error = %e, "BLE receive error"); stats.record_recv_error(); @@ -998,7 +1029,12 @@ async fn scan_probe_loop( // Pubkey exchange, then promote connection to pool let ta = addr.to_transport_addr(); - match pubkey_exchange(&stream, &our_pubkey).await { + let send_mtu = stream.send_mtu(); + let recv_mtu = stream.recv_mtu(); + let stream = Arc::new(stream); + // Shared reader for pubkey exchange + receive loop. + let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu); + match pubkey_exchange(&mut reader, &our_pubkey).await { Ok(peer_pubkey) => { debug!(addr = %addr, "BLE probe complete"); @@ -1017,12 +1053,8 @@ async fn scan_probe_loop( } // Promote connection to pool — no second L2CAP connect needed - let send_mtu = stream.send_mtu(); - let recv_mtu = stream.recv_mtu(); - let stream = Arc::new(stream); - let recv_task = tokio::spawn(receive_loop( - Arc::clone(&stream), + reader, ta.clone(), Arc::clone(&pool), packet_tx.clone(), diff --git a/src/transport/ble/stream_read.rs b/src/transport/ble/stream_read.rs new file mode 100644 index 0000000..8490b44 --- /dev/null +++ b/src/transport/ble/stream_read.rs @@ -0,0 +1,175 @@ +//! `AsyncRead` adapter over a datagram-shaped [`BleStream`]. +//! +//! L2CAP delivers different boundary guarantees per platform: +//! +//! - **BlueZ** (`SOCK_SEQPACKET`) preserves SDU boundaries — one `recv` is +//! exactly one FIPS packet. +//! - **Android** (`BluetoothSocket` input stream) and **CoreBluetooth** are +//! byte-stream oriented — a `recv` may return a fragment of a packet, or +//! several packets coalesced. +//! +//! FIPS packets are self-delimiting via the 4-byte FMP common prefix, so the +//! shared framer [`crate::transport::tcp::stream::read_fmp_packet`] can recover +//! boundaries from any byte stream. This adapter turns a [`BleStream`] (whose +//! `recv` fills a `&mut [u8]`) into the [`AsyncRead`] that framer expects. On a +//! SeqPacket backend it's a no-op pass-through; on a stream backend it +//! reassembles. Either way the layer above sees one whole packet per read. + +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use tokio::io::{AsyncRead, ReadBuf}; + +use super::io::BleStream; + +/// An in-flight `recv`: owns its scratch buffer and yields an owned `Vec`, so +/// the future is `'static` and can live across `poll_read` calls. +type PendingRecv = Pin>> + Send>>; + +/// Buffers a [`BleStream`] into an [`AsyncRead`] byte stream. +pub struct BleStreamRead { + stream: Arc, + /// Per-`recv` scratch size; also the framer's MTU bound. + mtu: u16, + /// Bytes from the last `recv` not yet consumed by the framer. + leftover: Vec, + /// Read cursor into `leftover`. + pos: usize, + /// In-flight `recv`, if one is underway. + pending: Option, +} + +impl BleStreamRead { + /// Wrap a stream. `mtu` is the per-`recv` scratch size (use the channel's + /// recv MTU). + pub fn new(stream: Arc, mtu: u16) -> Self { + Self { + stream, + mtu: mtu.max(1), + leftover: Vec::new(), + pos: 0, + pending: None, + } + } + + /// The wrapped stream, for sending on the same channel (e.g. the pubkey + /// exchange writes here while reads come through the buffer). + pub fn stream(&self) -> &S { + &self.stream + } +} + +impl AsyncRead for BleStreamRead { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + dst: &mut ReadBuf<'_>, + ) -> Poll> { + // BleStreamRead is Unpin (all fields are), so get_mut is sound. + let this = self.get_mut(); + loop { + // Serve buffered bytes first. + if this.pos < this.leftover.len() { + let avail = &this.leftover[this.pos..]; + let n = avail.len().min(dst.remaining()); + dst.put_slice(&avail[..n]); + this.pos += n; + return Poll::Ready(Ok(())); + } + + // Buffer drained: pull the next datagram. The future owns its + // scratch and returns it truncated, so it captures only `Arc`. + if this.pending.is_none() { + let stream = Arc::clone(&this.stream); + let mtu = this.mtu as usize; + this.pending = Some(Box::pin(async move { + let mut scratch = vec![0u8; mtu]; + let n = stream + .recv(&mut scratch) + .await + .map_err(|e| std::io::Error::other(e.to_string()))?; + scratch.truncate(n); + Ok(scratch) + })); + } + + match this.pending.as_mut().unwrap().as_mut().poll(cx) { + Poll::Ready(Ok(buf)) => { + this.pending = None; + // A zero-length recv is the BleStream peer-closed signal; + // leaving `dst` unfilled surfaces as EOF to the framer. + if buf.is_empty() { + return Poll::Ready(Ok(())); + } + this.leftover = buf; + this.pos = 0; + // loop to copy out + } + Poll::Ready(Err(e)) => { + this.pending = None; + return Poll::Ready(Err(e)); + } + Poll::Pending => return Poll::Pending, + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::ble::addr::BleAddr; + use crate::transport::ble::io::MockBleStream; + use tokio::io::AsyncReadExt; + + fn addr(n: u8) -> BleAddr { + BleAddr { + adapter: "hci0".to_string(), + device: [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, n], + } + } + + /// Several small recvs reassemble into one read_exact. + #[tokio::test] + async fn reassembles_fragmented_recv() { + let (a, b) = MockBleStream::pair(addr(1), addr(2), 2048); + // Peer sends three fragments that together form one 10-byte message. + a.send(b"abc").await.unwrap(); + a.send(b"defg").await.unwrap(); + a.send(b"hij").await.unwrap(); + + let mut reader = BleStreamRead::new(Arc::new(b), 2048); + let mut out = [0u8; 10]; + reader.read_exact(&mut out).await.unwrap(); + assert_eq!(&out, b"abcdefghij"); + } + + /// A recv carrying several packets is served across multiple reads without + /// dropping the tail (the coalescing case). + #[tokio::test] + async fn serves_coalesced_recv_in_pieces() { + let (a, b) = MockBleStream::pair(addr(1), addr(2), 2048); + a.send(b"0123456789").await.unwrap(); + + let mut reader = BleStreamRead::new(Arc::new(b), 2048); + let mut first = [0u8; 4]; + reader.read_exact(&mut first).await.unwrap(); + assert_eq!(&first, b"0123"); + let mut rest = [0u8; 6]; + reader.read_exact(&mut rest).await.unwrap(); + assert_eq!(&rest, b"456789"); + } + + /// A closed stream surfaces as EOF (read_exact errors with UnexpectedEof). + #[tokio::test] + async fn closed_stream_is_eof() { + let (a, b) = MockBleStream::pair(addr(1), addr(2), 2048); + drop(a); // peer closes + let mut reader = BleStreamRead::new(Arc::new(b), 2048); + let mut out = [0u8; 4]; + let err = reader.read_exact(&mut out).await.unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof); + } +}