diff --git a/src/transport/ble/attempts.rs b/src/transport/ble/attempts.rs index fd2c944..fd4c4e1 100644 --- a/src/transport/ble/attempts.rs +++ b/src/transport/ble/attempts.rs @@ -70,6 +70,11 @@ pub enum BleAttemptOutcome { LostTiebreaker, /// The connection was usable but the pool had no room for it. PoolRejected, + /// The peer was already connected under a different link address, so this + /// duplicate was dropped. Expected traffic for a peer rotating resolvable + /// private addresses; a high count against one peer is that rotation being + /// absorbed rather than filling the pool. + DuplicateNode, } impl BleAttemptOutcome { @@ -82,6 +87,7 @@ impl BleAttemptOutcome { BleAttemptOutcome::PubkeyExchangeFailed => "pubkey-exchange-failed", BleAttemptOutcome::LostTiebreaker => "lost-tiebreaker", BleAttemptOutcome::PoolRejected => "pool-rejected", + BleAttemptOutcome::DuplicateNode => "duplicate-node", } } } diff --git a/src/transport/ble/mod.rs b/src/transport/ble/mod.rs index 0af22bb..ac8d4ce 100644 --- a/src/transport/ble/mod.rs +++ b/src/transport/ble/mod.rs @@ -406,10 +406,12 @@ impl BleTransport { let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu); // Pre-handshake pubkey exchange (temporary, pre-XX) + let mut peer_node: Option = None; if let Some(ref our_pubkey) = self.local_pubkey { match pubkey_exchange(&mut reader, our_pubkey).await { Ok(peer_pubkey) => { debug!(addr = %addr, "BLE outbound pubkey exchange complete"); + peer_node = Some(NodeAddr::from_pubkey(&peer_pubkey)); self.discovery_buffer .add_peer_with_pubkey(&ble_addr, peer_pubkey); } @@ -420,7 +422,7 @@ impl BleTransport { } } - self.promote_connection(addr, &ble_addr, stream, reader) + self.promote_connection(addr, &ble_addr, stream, reader, peer_node) .await } @@ -434,6 +436,7 @@ impl BleTransport { ble_addr: &BleAddr, stream: Arc, reader: BleStreamRead, + node_addr: Option, ) -> Result<(), TransportError> { let send_mtu = stream.send_mtu(); let recv_mtu = stream.recv_mtu(); @@ -456,6 +459,7 @@ impl BleTransport { established_at: tokio::time::Instant::now(), is_static: false, addr: ble_addr.clone(), + node_addr, }; let mut pool = self.pool.lock().await; @@ -535,10 +539,12 @@ impl BleTransport { let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu); // Pre-handshake pubkey exchange (temporary, pre-XX) + let mut peer_node: Option = None; if let Some(ref our_pubkey) = local_pubkey { match pubkey_exchange(&mut reader, our_pubkey).await { Ok(peer_pubkey) => { debug!(addr = %addr_clone, "BLE outbound pubkey exchange complete"); + peer_node = Some(NodeAddr::from_pubkey(&peer_pubkey)); discovery_buffer.add_peer_with_pubkey(&ble_addr, peer_pubkey); } Err(e) => { @@ -569,6 +575,7 @@ impl BleTransport { established_at: tokio::time::Instant::now(), is_static: false, addr: ble_addr, + node_addr: peer_node, }; let mut pool = pool.lock().await; @@ -817,6 +824,7 @@ async fn accept_loop( // into the connection before the pool outcome is recorded. let addr_s = addr.to_string(); let mut peer_node_hex = String::new(); + let mut peer_node_addr: Option = None; // Pre-handshake pubkey exchange (temporary, pre-XX) if let Some(ref our_pubkey) = local_pubkey { @@ -824,7 +832,38 @@ async fn accept_loop( Ok(peer_pubkey) => { debug!(addr = %ta, "BLE inbound pubkey exchange complete"); discovery_buffer.add_peer_with_pubkey(&addr, peer_pubkey); - peer_node_hex = NodeAddr::from_pubkey(&peer_pubkey).to_string(); + let peer_node = NodeAddr::from_pubkey(&peer_pubkey); + peer_node_hex = peer_node.to_string(); + peer_node_addr = Some(peer_node); + + // Already linked to this peer on another address? + // A peer using resolvable private addresses rotates + // continually, and every rotation dials in looking + // like a new device. Admitting those would put one + // peer in several pool slots and evict real ones. + // The incumbent link is kept: it is known-good, and + // a genuinely dead one is already reaped by the + // send-error and receive-loop paths. + let dup = { + let pool_guard = pool.lock().await; + pool_guard.find_by_node(&peer_node) + }; + if let Some(existing) = dup { + if existing != ta { + debug!( + addr = %ta, + existing = %existing, + "BLE inbound: peer already connected on another address, dropping duplicate" + ); + attempts::ble_attempt_log().record_outcome( + &addr_s, + &peer_node_hex, + attempts::BleRole::Peripheral, + attempts::BleAttemptOutcome::DuplicateNode, + ); + continue; + } + } // Cross-probe tie-breaker: smaller NodeAddr's // outbound wins. If we're smaller, our outbound @@ -882,6 +921,7 @@ async fn accept_loop( established_at: tokio::time::Instant::now(), is_static: false, addr, + node_addr: peer_node_addr, }; let mut pool_guard = pool.lock().await; @@ -1141,6 +1181,36 @@ async fn scan_probe_loop( } } + // Same duplicate guard as the inbound path: a rotated address + // for a peer we already hold a link to must not become a second + // pool entry. Checked after the tiebreaker so the two decisions + // stay independent. + let peer_node = NodeAddr::from_pubkey(&peer_pubkey); + let dup = { + let pool_guard = pool.lock().await; + pool_guard.find_by_node(&peer_node) + }; + if let Some(existing) = dup { + if existing != ta { + debug!( + addr = %ta, + existing = %existing, + "BLE probe: peer already connected on another address, dropping duplicate" + ); + attempts::ble_attempt_log().record_outcome( + &addr.to_string(), + &peer_node.to_string(), + attempts::BleRole::Central, + attempts::BleAttemptOutcome::DuplicateNode, + ); + // Report the peer so the node layer still learns the + // address maps to a peer it already knows. + buffer.add_peer_with_pubkey(&addr, peer_pubkey); + pending_addrs.retain(|a| a != &addr); + continue; + } + } + // Promote connection to pool — no second L2CAP connect needed let recv_task = tokio::spawn(receive_loop( reader, @@ -1160,6 +1230,7 @@ async fn scan_probe_loop( established_at: tokio::time::Instant::now(), is_static: false, addr: addr.clone(), + node_addr: Some(peer_node), }; let mut pool_guard = pool.lock().await; diff --git a/src/transport/ble/pool.rs b/src/transport/ble/pool.rs index ffdf45d..85cc75f 100644 --- a/src/transport/ble/pool.rs +++ b/src/transport/ble/pool.rs @@ -8,6 +8,7 @@ use std::collections::HashMap; use tokio::task::JoinHandle; +use crate::identity::NodeAddr; use crate::transport::{TransportAddr, TransportError}; use super::addr::BleAddr; @@ -28,6 +29,15 @@ pub struct BleConnection { pub is_static: bool, /// Parsed remote address. pub addr: BleAddr, + /// The peer's node address, once the pubkey exchange has learned it. + /// + /// The pool is keyed by *link* address, but a BLE link address is not a + /// stable identity: peers using resolvable private addresses rotate theirs + /// continually, and each rotation looks like a brand-new device. This field + /// carries the identity that does not rotate, so [`ConnectionPool::find_by_node`] + /// can recognise a peer we are already connected to under an address we have + /// not seen before. `None` for a connection whose peer is not yet identified. + pub node_addr: Option, } impl BleConnection { @@ -95,6 +105,24 @@ impl ConnectionPool { self.connections.contains_key(addr) } + /// Find an existing connection to `node`, whatever link address it arrived on. + /// + /// This is the identity check [`Self::contains`] cannot make. A peer using + /// resolvable private addresses presents a different link address every + /// rotation, so an address-keyed lookup reports "not connected" for a peer + /// that is very much connected — and the caller then opens a second link to + /// it, and a third. Callers that know the peer's node address should ask + /// this before admitting a connection. + /// + /// Only connections whose pubkey exchange has completed carry a node + /// address, so an unidentified connection is never matched. + pub fn find_by_node(&self, node: &NodeAddr) -> Option { + self.connections + .iter() + .find(|(_, c)| c.node_addr.as_ref() == Some(node)) + .map(|(addr, _)| addr.clone()) + } + /// Try to insert a connection, evicting if necessary. /// /// Returns `Ok(evicted_addr)` on success (with optional evicted peer), @@ -186,6 +214,13 @@ mod tests { } } + /// A distinct node identity per `n` — the identity that does NOT rotate. + fn test_node(n: u8) -> NodeAddr { + let mut bytes = [0u8; 16]; + bytes[0] = n; + NodeAddr::from_bytes(bytes) + } + fn test_conn(n: u8, is_static: bool) -> BleConnection<()> { BleConnection { stream: (), @@ -195,6 +230,7 @@ mod tests { established_at: tokio::time::Instant::now(), is_static, addr: test_ble_addr(n), + node_addr: None, } } @@ -287,4 +323,71 @@ mod tests { addrs.sort_by(|a, b| a.as_str().cmp(&b.as_str())); assert_eq!(addrs.len(), 2); } + + /// A node address is found regardless of which link address it arrived on — + /// the whole point of the lookup, since the link address rotates. + #[test] + fn find_by_node_matches_across_a_rotated_link_address() { + let mut pool: ConnectionPool<()> = ConnectionPool::new(7); + let node = test_node(1); + let mut conn = test_conn(1, false); + conn.node_addr = Some(node); + pool.insert(test_addr(1), conn).unwrap(); + + // Found under the address it was inserted with... + assert_eq!(pool.find_by_node(&node), Some(test_addr(1))); + // ...and the address-keyed check agrees for that address only. + assert!(pool.contains(&test_addr(1))); + // A rotated address for the same peer is NOT found by `contains` — + // which is exactly the gap `find_by_node` exists to close. + assert!(!pool.contains(&test_addr(99))); + assert_eq!(pool.find_by_node(&node), Some(test_addr(1))); + } + + #[test] + fn find_by_node_ignores_unidentified_connections() { + let mut pool: ConnectionPool<()> = ConnectionPool::new(7); + // No pubkey exchange yet, so no node address. + pool.insert(test_addr(1), test_conn(1, false)).unwrap(); + assert_eq!(pool.find_by_node(&test_node(1)), None); + } + + #[test] + fn find_by_node_returns_none_for_an_unconnected_node() { + let mut pool: ConnectionPool<()> = ConnectionPool::new(7); + let mut conn = test_conn(1, false); + conn.node_addr = Some(test_node(1)); + pool.insert(test_addr(1), conn).unwrap(); + assert_eq!(pool.find_by_node(&test_node(2)), None); + } + + /// The regression this guards: without a node-identity check, N rotated + /// addresses for ONE peer become N pool entries and evict real peers. With + /// it, the caller can see the peer is already present and decline. + #[test] + fn rotated_addresses_would_otherwise_fill_the_pool() { + let mut pool: ConnectionPool<()> = ConnectionPool::new(7); + let node = test_node(1); + + // One genuine connection to the peer. + let mut first = test_conn(1, false); + first.node_addr = Some(node); + pool.insert(test_addr(1), first).unwrap(); + + // Ten rotations arrive. Each is a distinct link address, so `contains` + // says "new" every time — but `find_by_node` recognises all of them. + for n in 2..12u8 { + assert!( + !pool.contains(&test_addr(n)), + "rotation {n} looks new by address" + ); + assert_eq!( + pool.find_by_node(&node), + Some(test_addr(1)), + "rotation {n} is recognised as the peer already connected", + ); + } + // Nothing was admitted, so the pool still holds exactly one link. + assert_eq!(pool.len(), 1); + } }