laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-07-22 07:48:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

BLE continuous advertising, probe cooldown replaces burst beacon

Browse Source 
Replace burst beacon pattern (1s on / 30s off) with continuous
advertising. The burst pattern caused L2CAP connect timeouts because
the remote side was no longer connectable when the probe fired after
jitter delay. BLE advertising overhead is negligible (~0.15% duty
cycle on advertising channels).

Replace the seen HashSet + jitter delay queue with a simple cooldown
map. After probing an address (success or failure), suppress re-probe
for 30s (configurable via probe_cooldown_secs). Connected peers are
filtered by pool membership check. This eliminates the bug where
failed probes permanently blacklisted addresses for the session
lifetime.

Remove config fields: scan_interval_secs, beacon_interval_secs,
beacon_duration_secs. Add: probe_cooldown_secs.
This commit is contained in:
Johnathan Corgan
2026-03-25 15:22:43 +00:00
parent 88fcf57067
commit d801fd0052
4 changed files with 125 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
CHANGELOG.md
View File

@@ -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

24
docs/design/fips-configuration.md
View File

@@ -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..."

44
src/config/transport.rs
View File

@@ -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<bool>,
/// 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<u64>,
/// Beacon interval in seconds between advertising bursts. Default: 10.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub beacon_interval_secs: Option<u64>,
/// Beacon duration in seconds per advertising burst. Default: 3.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub beacon_duration_secs: Option<u64>,
pub probe_cooldown_secs: Option<u64>,
}
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)
}
}

286
src/transport/ble/mod.rs
View File

@@ -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<I: BleIo> {
accept_task: Option<JoinHandle<()>>,
/// Combined scan + probe loop task handle.
scan_probe_task: Option<JoinHandle<()>>,
/// Advertising task handle.
advertise_task: Option<JoinHandle<()>>,
/// Discovery buffer for discovered peers.
discovery_buffer: Arc<DiscoveryBuffer>,
/// Transport statistics.
@@ -131,7 +128,6 @@ impl<I: BleIo> BleTransport<I> {
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<I: BleIo> BleTransport<I> {
}
}
// 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<I: BleIo> BleTransport<I> {
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<I: BleIo> BleTransport<I> {
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<S: BleStream>(
.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<I: io::BleIo>(
io: Arc<I>,
interval_secs: u64,
duration_secs: u64,
stats: Arc<BleStats>,
) {
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<S: BleStream>(
/// 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<I: io::BleIo>(
mut scanner: I::Scanner,
@@ -903,129 +851,97 @@ async fn scan_probe_loop<I: io::BleIo>(
local_pubkey: Option<[u8; 32]>,
psm: u16,
connect_timeout_ms: u64,
cooldown_secs: u64,
local_node_addr: Option<NodeAddr>,
) {
use rand::RngExt;
// Time-sorted delay queue: (fire_time, addr). Min-heap on fire_time.
let mut delay_queue: BinaryHeap<Reverse<(tokio::time::Instant, BleAddr)>> = BinaryHeap::new();
// Track queued/probed addresses to deduplicate
let mut seen: HashSet<BleAddr> = HashSet::new();
let max_jitter_ms: u64 = 5000;
// Track last probe time per address for cooldown
let mut last_probed: HashMap<BleAddr, tokio::time::Instant> = 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)
}
}