diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d27033..af9b74c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,9 +21,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Bluetooth Low Energy (BLE) L2CAP Connection-Oriented Channel transport with per-link MTU negotiation, behind the `ble` Cargo feature flag (default-on, Linux only, requires BlueZ) -- BLE peer discovery via scan/probe with per-entry random jitter to - prevent herd effects when multiple nodes see the same beacon -- Periodic BLE beacon advertising with configurable interval and duration +- BLE peer discovery via continuous scan/probe with cooldown-based + deduplication (`probe_cooldown_secs`, default 30s) +- Continuous BLE advertising for reliable L2CAP connectivity - Cross-probe tie-breaker using deterministic NodeAddr comparison - Connection pool with configurable capacity and eviction diff --git a/docs/design/fips-configuration.md b/docs/design/fips-configuration.md index 32f43ad..dc87248 100644 --- a/docs/design/fips-configuration.md +++ b/docs/design/fips-configuration.md @@ -503,20 +503,20 @@ using the `bluer` crate. | `transports.ble.scan` | bool | `true` | Listen for BLE beacon advertisements from other nodes | | `transports.ble.auto_connect` | bool | `false` | Automatically connect to discovered peers | | `transports.ble.accept_connections` | bool | `true` | Accept incoming L2CAP connections | -| `transports.ble.scan_interval_secs` | u64 | `10` | Interval between BLE scan cycles | -| `transports.ble.beacon_interval_secs` | u64 | `10` | Interval between beacon advertising bursts | -| `transports.ble.beacon_duration_secs` | u64 | `3` | Duration of each beacon advertising burst | +| `transports.ble.probe_cooldown_secs` | u64 | `30` | Cooldown before re-probing the same BLE address | **Address format.** BLE peer addresses use the form `"adapter/device_address"` — for example, `"hci0/AA:BB:CC:DD:EE:FF"`. -**Scan/probe and tie-breaking.** When `scan` is enabled, the transport -periodically scans for BLE beacons from other FIPS nodes. Discovered -peers are probed with per-entry random jitter to prevent herd effects -when multiple nodes see the same beacon simultaneously. If two nodes -probe each other at the same time (cross-probe), a deterministic -tie-breaker based on NodeAddr comparison ensures only one connection -is established. +**Advertising and scanning.** When `advertise` is enabled, the transport +advertises the FIPS service UUID continuously so that nearby nodes can +discover and connect via L2CAP. When `scan` is enabled, the transport +continuously scans for other FIPS nodes' advertisements. Discovered +peers are probed immediately (L2CAP connect + pubkey exchange) with a +cooldown (`probe_cooldown_secs`) to prevent rapid re-probing of the same +address. If two nodes probe each other at the same time (cross-probe), +a deterministic tie-breaker based on NodeAddr comparison ensures only +one connection is established. **Connection pool.** The `max_connections` parameter limits the number of concurrent BLE connections. When the pool is full, the least-recently-used @@ -802,9 +802,7 @@ transports: # scan: true # listen for BLE beacons # auto_connect: false # connect to discovered peers # accept_connections: true # accept incoming L2CAP connections - # scan_interval_secs: 10 # interval between scan cycles - # beacon_interval_secs: 10 # interval between beacon bursts - # beacon_duration_secs: 3 # duration of each beacon burst + # probe_cooldown_secs: 30 # cooldown before re-probing same address peers: # static peer list # - npub: "npub1..." diff --git a/src/config/transport.rs b/src/config/transport.rs index 9e21edf..a2e7bd3 100644 --- a/src/config/transport.rs +++ b/src/config/transport.rs @@ -514,14 +514,9 @@ const DEFAULT_BLE_MAX_CONNECTIONS: usize = 7; /// Default BLE connect timeout in milliseconds. const DEFAULT_BLE_CONNECT_TIMEOUT_MS: u64 = 10_000; -/// Default BLE scan interval in seconds. -const DEFAULT_BLE_SCAN_INTERVAL_SECS: u64 = 10; - -/// Default BLE beacon interval in seconds. -const DEFAULT_BLE_BEACON_INTERVAL_SECS: u64 = 30; - -/// Default BLE beacon duration in seconds (how long each burst lasts). -const DEFAULT_BLE_BEACON_DURATION_SECS: u64 = 1; +/// Default BLE probe cooldown in seconds. After probing an address +/// (success or failure), wait this long before probing it again. +const DEFAULT_BLE_PROBE_COOLDOWN_SECS: u64 = 30; /// BLE transport instance configuration. /// @@ -566,17 +561,10 @@ pub struct BleConfig { #[serde(default, skip_serializing_if = "Option::is_none")] pub accept_connections: Option, - /// Scan interval in seconds. Default: 10. + /// Probe cooldown in seconds. After probing a BLE address, wait + /// this long before probing the same address again. Default: 30. #[serde(default, skip_serializing_if = "Option::is_none")] - pub scan_interval_secs: Option, - - /// Beacon interval in seconds between advertising bursts. Default: 10. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub beacon_interval_secs: Option, - - /// Beacon duration in seconds per advertising burst. Default: 3. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub beacon_duration_secs: Option, + pub probe_cooldown_secs: Option, } impl BleConfig { @@ -626,22 +614,10 @@ impl BleConfig { self.accept_connections.unwrap_or(true) } - /// Get the scan interval in seconds. Default: 10. - pub fn scan_interval_secs(&self) -> u64 { - self.scan_interval_secs - .unwrap_or(DEFAULT_BLE_SCAN_INTERVAL_SECS) - } - - /// Get the beacon interval in seconds. Default: 10. - pub fn beacon_interval_secs(&self) -> u64 { - self.beacon_interval_secs - .unwrap_or(DEFAULT_BLE_BEACON_INTERVAL_SECS) - } - - /// Get the beacon duration in seconds. Default: 3. - pub fn beacon_duration_secs(&self) -> u64 { - self.beacon_duration_secs - .unwrap_or(DEFAULT_BLE_BEACON_DURATION_SECS) + /// Get the probe cooldown in seconds. Default: 30. + pub fn probe_cooldown_secs(&self) -> u64 { + self.probe_cooldown_secs + .unwrap_or(DEFAULT_BLE_PROBE_COOLDOWN_SECS) } } diff --git a/src/transport/ble/mod.rs b/src/transport/ble/mod.rs index 3c2586e..5311c9e 100644 --- a/src/transport/ble/mod.rs +++ b/src/transport/ble/mod.rs @@ -36,8 +36,7 @@ use io::{BleIo, BleScanner, BleStream}; use pool::{BleConnection, ConnectionPool}; use stats::BleStats; -use std::cmp::Reverse; -use std::collections::{BinaryHeap, HashMap, HashSet}; +use std::collections::HashMap; use std::sync::Arc; use tokio::sync::Mutex; use tokio::task::JoinHandle; @@ -90,8 +89,6 @@ pub struct BleTransport { accept_task: Option>, /// Combined scan + probe loop task handle. scan_probe_task: Option>, - /// Advertising task handle. - advertise_task: Option>, /// Discovery buffer for discovered peers. discovery_buffer: Arc, /// Transport statistics. @@ -131,7 +128,6 @@ impl BleTransport { packet_tx, accept_task: None, scan_probe_task: None, - advertise_task: None, discovery_buffer: Arc::new(DiscoveryBuffer::new(transport_id)), stats: Arc::new(BleStats::new()), local_pubkey: None, @@ -210,24 +206,14 @@ impl BleTransport { } } - // Start periodic beacon (advertise in bursts) + // Start continuous advertising if self.config.advertise() { - let io = Arc::clone(&self.io); - let beacon_interval = self.config.beacon_interval_secs(); - let beacon_duration = self.config.beacon_duration_secs(); - let stats = Arc::clone(&self.stats); - self.advertise_task = Some(tokio::spawn(beacon_loop( - io, - beacon_interval, - beacon_duration, - stats, - ))); - debug!( - adapter = %adapter, - interval_secs = beacon_interval, - duration_secs = beacon_duration, - "BLE beacon loop started" - ); + if let Err(e) = self.io.start_advertising().await { + warn!(adapter = %adapter, error = %e, "failed to start BLE advertising"); + } else { + self.stats.record_advertisement(); + debug!(adapter = %adapter, "BLE advertising started (continuous)"); + } } // Start combined scan + probe loop @@ -243,6 +229,7 @@ impl BleTransport { self.local_pubkey, self.config.psm(), self.config.connect_timeout_ms(), + self.config.probe_cooldown_secs(), local_node_addr, ))); debug!(adapter = %adapter, "BLE scan+probe loop started"); @@ -273,11 +260,6 @@ impl BleTransport { task.abort(); } - // Abort advertising task - if let Some(task) = self.advertise_task.take() { - task.abort(); - } - // Drain connecting pool { let mut connecting = self.connecting.lock().await; @@ -698,42 +680,9 @@ async fn pubkey_exchange( .map_err(|e| TransportError::RecvFailed(format!("pubkey exchange: invalid key: {}", e))) } -/// Beacon loop: periodically advertises the FIPS service UUID. -/// -/// Advertises immediately on startup, then repeats in bursts: -/// advertise for `duration_secs`, stop for `interval_secs`, repeat. -/// This reduces BLE radio usage compared to continuous advertising. -async fn beacon_loop( - io: Arc, - interval_secs: u64, - duration_secs: u64, - stats: Arc, -) { - let interval = std::time::Duration::from_secs(interval_secs); - let duration = std::time::Duration::from_secs(duration_secs); - - loop { - // Start advertising burst - if let Err(e) = io.start_advertising().await { - warn!(error = %e, "BLE beacon: failed to start advertising"); - tokio::time::sleep(interval).await; - continue; - } - stats.record_advertisement(); - trace!(duration_secs, "BLE beacon: advertising"); - - // Hold advertisement for the burst duration - tokio::time::sleep(duration).await; - - // Stop advertising until next burst - if let Err(e) = io.stop_advertising().await { - warn!(error = %e, "BLE beacon: failed to stop advertising"); - } - - // Wait for next beacon interval - tokio::time::sleep(interval).await; - } -} +// Beacon loop removed — advertising is now continuous (started once +// in start_async, stopped in stop_async). BLE advertising overhead +// is negligible (~0.15% duty cycle on advertising channels). /// Accept loop: accepts inbound L2CAP connections, exchanges pubkeys, /// and adds to pool. @@ -884,15 +833,14 @@ async fn receive_loop( /// Combined scan + probe loop. /// -/// Scanner events arrive and get inserted into a delay queue with random -/// jitter. When a delayed entry expires, it's probed (L2CAP connect + -/// pubkey exchange) and the result goes into the DiscoveryBuffer. The -/// node layer picks up discovered peers via `discover()` and connects -/// through the normal auto-connect → send_async → connect_inline path. +/// Scanner events arrive continuously (both sides advertise continuously). +/// Each scan result is probed immediately unless the address is in cooldown +/// (recently probed) or already connected. On successful probe, the peer +/// is reported to the discovery buffer for the node layer to auto-connect. /// -/// The delay queue ensures each beacon response gets an independent -/// random delay, preventing herd effects when multiple nodes see the -/// same advertisement simultaneously. +/// Cooldown prevents rapid re-probing of the same address: after any probe +/// attempt (success or failure), the address is suppressed for +/// `cooldown_secs`. Connected peers are filtered by pool membership. #[allow(clippy::too_many_arguments)] async fn scan_probe_loop( mut scanner: I::Scanner, @@ -903,129 +851,97 @@ async fn scan_probe_loop( local_pubkey: Option<[u8; 32]>, psm: u16, connect_timeout_ms: u64, + cooldown_secs: u64, local_node_addr: Option, ) { - use rand::RngExt; - - // Time-sorted delay queue: (fire_time, addr). Min-heap on fire_time. - let mut delay_queue: BinaryHeap> = BinaryHeap::new(); - // Track queued/probed addresses to deduplicate - let mut seen: HashSet = HashSet::new(); - - let max_jitter_ms: u64 = 5000; + // Track last probe time per address for cooldown + let mut last_probed: HashMap = HashMap::new(); + let cooldown = std::time::Duration::from_secs(cooldown_secs); loop { - // Compute sleep target: next delay queue entry, or far future - let next_fire = delay_queue - .peek() - .map(|Reverse((t, _))| *t) - .unwrap_or_else(|| tokio::time::Instant::now() + std::time::Duration::from_secs(3600)); - - tokio::select! { - // New scan result from adapter - result = scanner.next() => { - let addr = match result { - Some(a) => a, - None => { - debug!("BLE scanner ended"); - break; - } - }; - - trace!(addr = %addr, "BLE scan result"); - stats.record_scan_result(); - - // Dedup: skip if already queued, probed, or connected - if seen.contains(&addr) { - continue; - } - { - let pool_guard = pool.lock().await; - if pool_guard.contains(&addr.to_transport_addr()) { - continue; - } - } - - // Schedule with random jitter - let jitter = std::time::Duration::from_millis( - rand::rng().random_range(0..max_jitter_ms), - ); - let fire_at = tokio::time::Instant::now() + jitter; - delay_queue.push(Reverse((fire_at, addr.clone()))); - seen.insert(addr); + let addr = match scanner.next().await { + Some(a) => a, + None => { + debug!("BLE scanner ended"); + break; } + }; - // Next delayed probe is ready - _ = tokio::time::sleep_until(next_fire) => { - let Reverse((_, addr)) = match delay_queue.pop() { - Some(entry) => entry, - None => continue, - }; + trace!(addr = %addr, "BLE scan result"); + stats.record_scan_result(); - // Skip if connected while waiting - { - let pool_guard = pool.lock().await; - if pool_guard.contains(&addr.to_transport_addr()) { - continue; - } - } - - // Need pubkey for probe - let our_pubkey = match local_pubkey { - Some(pk) => pk, - None => { - // No pubkey — just report MAC without identity - buffer.add_peer(&addr); - continue; - } - }; - - // L2CAP connect - let stream = match tokio::time::timeout( - std::time::Duration::from_millis(connect_timeout_ms), - io.connect(&addr, psm), - ) - .await - { - Ok(Ok(s)) => s, - Ok(Err(e)) => { - debug!(addr = %addr, error = %e, "BLE probe connect failed"); - continue; - } - Err(_) => { - debug!(addr = %addr, "BLE probe connect timeout"); - stats.record_connect_timeout(); - continue; - } - }; - - // Pubkey exchange, then close the L2CAP connection - match pubkey_exchange(&stream, &our_pubkey).await { - Ok(peer_pubkey) => { - debug!(addr = %addr, "BLE probe complete"); - - // Cross-probe tie-breaker: smaller NodeAddr's outbound wins - if let Some(ref our_addr) = local_node_addr { - let peer_addr = NodeAddr::from_pubkey(&peer_pubkey); - if our_addr >= &peer_addr { - debug!( - addr = %addr, - "BLE probe tie-breaker: yielding to peer's outbound" - ); - } - } - - // Report to node layer — auto-connect will establish - // a persistent connection via send_async/connect_inline - buffer.add_peer_with_pubkey(&addr, peer_pubkey); - } - Err(e) => { - debug!(addr = %addr, error = %e, "BLE probe pubkey exchange failed"); - } - } - // L2CAP connection dropped here (stream goes out of scope) + // Skip if already connected + { + let pool_guard = pool.lock().await; + if pool_guard.contains(&addr.to_transport_addr()) { + continue; } } + + // Skip if in cooldown + if last_probed + .get(&addr) + .is_some_and(|last| last.elapsed() < cooldown) + { + continue; + } + + // Record probe time (before attempt, so cooldown applies on failure too) + last_probed.insert(addr.clone(), tokio::time::Instant::now()); + + // Need pubkey for probe + let our_pubkey = match local_pubkey { + Some(pk) => pk, + None => { + buffer.add_peer(&addr); + continue; + } + }; + + // L2CAP connect + let stream = match tokio::time::timeout( + std::time::Duration::from_millis(connect_timeout_ms), + io.connect(&addr, psm), + ) + .await + { + Ok(Ok(s)) => s, + Ok(Err(e)) => { + debug!(addr = %addr, error = %e, "BLE probe connect failed"); + continue; + } + Err(_) => { + debug!(addr = %addr, "BLE probe connect timeout"); + stats.record_connect_timeout(); + continue; + } + }; + + // Pubkey exchange, then close the L2CAP connection + match pubkey_exchange(&stream, &our_pubkey).await { + Ok(peer_pubkey) => { + debug!(addr = %addr, "BLE probe complete"); + + // Cross-probe tie-breaker: smaller NodeAddr's outbound wins + if let Some(ref our_addr) = local_node_addr { + let peer_addr = NodeAddr::from_pubkey(&peer_pubkey); + if our_addr >= &peer_addr { + debug!( + addr = %addr, + "BLE probe tie-breaker: yielding to peer's outbound" + ); + } + } + + // Report to node layer — auto-connect will establish + // a persistent connection via send_async/connect_inline + buffer.add_peer_with_pubkey(&addr, peer_pubkey); + } + Err(e) => { + debug!(addr = %addr, error = %e, "BLE probe pubkey exchange failed"); + } + } + // L2CAP connection dropped here (stream goes out of scope) } }