fix(ble): recognise a peer by node identity, not by its rotating link address

F-05 in Myco's Phase 1 field findings recorded a peer that dialled in 28 times
from 28 distinct BLE addresses in twenty minutes. It was one node the whole time:
peers using resolvable private addresses rotate them continually, and every
rotation presents as a brand-new device.

Every identity check in this transport keyed on the link address, so none of them
could tell. ConnectionPool is HashMap<TransportAddr, _>, and all three
already-connected guards — accept_loop and both in scan_probe_loop — ask
pool.contains(addr.to_transport_addr()). For a rotated address the answer is
always "not connected", so the caller opens another link, and another.

Those 28 rotations were harmless only because the cross-probe tiebreaker happened
to reject every one of them. Which side the tiebreaker protects is decided by a
byte comparison of two node addresses; had they sorted the other way, the same 28
inbound dials would have been admitted into a pool that holds 7, evicting genuine
peers roughly four times over. The tiebreaker is not the problem and is unchanged
here — the problem is that link identity was being used as node identity.

BleConnection now carries the peer's NodeAddr once the pubkey exchange learns it,
and ConnectionPool::find_by_node looks a peer up by the identity that does not
rotate. Both admission points consult it after the exchange and decline a
duplicate, keeping the incumbent link: it is known-good, and a genuinely dead one
is already reaped by the send-error and receive-loop paths.

Declines are recorded as a new BleAttemptOutcome::DuplicateNode ("duplicate-node")
so the absorption is visible in the same per-peer log that exposed the problem,
rather than becoming silent.

Full suite 1406 passed, 0 failed; 4 new pool tests including one that pins the
regression — ten rotations of one peer leave the pool holding exactly one link.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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",
         }
     }
 }

src/transport/ble/mod.rs

@@ -406,10 +406,12 @@ impl<I: BleIo> BleTransport<I> {
         let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
 
         // Pre-handshake pubkey exchange (temporary, pre-XX)
+        let mut peer_node: Option<NodeAddr> = 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<I: BleIo> BleTransport<I> {
             }
         }
 
-        self.promote_connection(addr, &ble_addr, stream, reader)
+        self.promote_connection(addr, &ble_addr, stream, reader, peer_node)
             .await
     }
 
@@ -434,6 +436,7 @@ impl<I: BleIo> BleTransport<I> {
         ble_addr: &BleAddr,
         stream: Arc<I::Stream>,
         reader: BleStreamRead<I::Stream>,
+        node_addr: Option<NodeAddr>,
     ) -> Result<(), TransportError> {
         let send_mtu = stream.send_mtu();
         let recv_mtu = stream.recv_mtu();
@@ -456,6 +459,7 @@ impl<I: BleIo> BleTransport<I> {
             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<I: BleIo> BleTransport<I> {
                     let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
 
                     // Pre-handshake pubkey exchange (temporary, pre-XX)
+                    let mut peer_node: Option<NodeAddr> = 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<I: BleIo> BleTransport<I> {
                         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<A>(
                 // 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<NodeAddr> = None;
 
                 // Pre-handshake pubkey exchange (temporary, pre-XX)
                 if let Some(ref our_pubkey) = local_pubkey {
@@ -824,7 +832,38 @@ async fn accept_loop<A>(
                         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<A>(
                     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<I: io::BleIo>(
                     }
                 }
 
+                // 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<I: io::BleIo>(
                     established_at: tokio::time::Instant::now(),
                     is_static: false,
                     addr: addr.clone(),
+                    node_addr: Some(peer_node),
                 };
 
                 let mut pool_guard = pool.lock().await;

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<S> {
     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<NodeAddr>,
 }
 
 impl<S> BleConnection<S> {
@@ -95,6 +105,24 @@ impl<S> ConnectionPool<S> {
         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<TransportAddr> {
+        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);
+    }
 }