Run each inbound BLE handshake off the accept loop, bounded

The accept loop ran the pre-handshake pubkey exchange inline, so a peer that
connected an L2CAP channel and then said nothing held the loop for the full
5-second exchange deadline and no other inbound connection was accepted in
that window. The max-connections argument the loop was given was never used,
so the effective inbound concurrency was one.

Move the per-connection work into admit_inbound and spawn it, with eight
handshakes allowed in flight. At the bound, abort the oldest pending
handshake rather than waiting for a slot: waiting would leave the same denial
standing at eight connections instead of one, and a healthy exchange is a
single round trip, so anything still pending under a flood is overwhelmingly
the flooder's. Count the aborts in the transport stats.

Keep the tasks in a JoinSet owned by the accept loop rather than behind a
semaphore. Stopping the transport aborts that task and nothing else, so the
set is what stops the handshakes; bare spawns would survive stop and could
insert into a pool it had just drained.

The budget is a module constant and deliberately not the pool capacity: an
operator who sets max_connections low would otherwise get a serial accept
loop back, which is the defect itself.

Give the send half of the pubkey exchange the same deadline the receive half
already had. A peer that stops draining its channel could park that write
indefinitely, which also reached the outbound connect and scan-probe paths.

CHANGELOG.md

@@ -1570,6 +1570,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   doing so, which is a denial of new sessions rather than the unbounded memory
   growth it replaces.
 
+- One inbound BLE connector can no longer stall every other inbound
+  connection. The accept loop ran the pre-handshake pubkey exchange inline, so
+  a peer that connected an L2CAP channel and then said nothing held the loop
+  for the full 5-second exchange deadline and no other inbound connection was
+  accepted in that window; the maximum-connections argument the loop was given
+  was never used, so the effective concurrency was one. Each inbound connection
+  now runs its handshake in its own task, up to eight in flight, and at that
+  bound the oldest pending handshake is aborted to make room rather than the
+  loop waiting for a slot: a healthy exchange is one round trip, so anything
+  still pending under a flood is overwhelmingly the flooder's, and a genuinely
+  slow peer that is aborted reconnects, which is better than never being
+  accepted at all. Aborted handshakes are counted in the transport's stats as
+  `handshakes_aborted`. The tasks live in a set owned by the accept loop, so
+  stopping the transport stops them too and none can insert into a pool that
+  stop has just drained. Separately, the send half of the pubkey exchange had
+  no deadline at all while the receive half had one, so a peer that stopped
+  draining its channel could park the write forever; it now shares the same
+  5-second deadline, which also covers the outbound connect and scan-probe
+  paths. **What this does not close**: eight simultaneous silent connectors
+  still occupy the whole in-flight budget, and no BlueZ hardware was exercised,
+  so the controller's own concurrent-link limit and accept backlog depth stay
+  unmeasured.
+
 - An accepted inbound TCP connection no longer holds a slot indefinitely
   without sending anything. The cap was tested at accept and the pool insert
   and counter bump followed with no read in between, while the frame reader's

src/transport/ble/mod.rs

@@ -846,11 +846,17 @@ async fn pubkey_exchange<S: BleStream + 'static>(
     reader: &mut BleStreamRead<S>,
     local_pubkey: &[u8; 32],
 ) -> Result<XOnlyPublicKey, TransportError> {
-    // Send our pubkey
+    let timeout = std::time::Duration::from_secs(PUBKEY_EXCHANGE_TIMEOUT_SECS);
+
+    // Send our pubkey. The deadline matters as much as the one below it: a
+    // peer that never drains the L2CAP channel stalls the write instead.
     let mut msg = [0u8; PUBKEY_EXCHANGE_SIZE];
     msg[0] = PUBKEY_EXCHANGE_PREFIX;
     msg[1..].copy_from_slice(local_pubkey);
-    stream.send(&msg).await?;
+    match tokio::time::timeout(timeout, stream.send(&msg)).await {
+        Ok(result) => result?,
+        Err(_) => return Err(TransportError::Timeout),
+    }
 
     // Receive peer's pubkey (with timeout to prevent indefinite blocking)
     let mut buf = [0u8; PUBKEY_EXCHANGE_SIZE];
@@ -880,8 +886,30 @@ async fn pubkey_exchange<S: BleStream + 'static>(
 // 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,
+/// Inbound handshakes allowed to be in flight at once.
-/// and adds to pool.
+///
+/// Deliberately independent of the pool capacity: that is the budget for
+/// established links, and tying the two together would mean an operator who
+/// sets `max_connections = 1` also gets a serial accept loop, which is the
+/// defect this bound exists to close. A healthy exchange is one round trip
+/// and completes in milliseconds, so this is never reached honestly. Raising
+/// it lets a flood hold more concurrent handshakes; lowering it makes a
+/// legitimate slow peer likelier to be aborted under flood.
+const INBOUND_HANDSHAKE_INFLIGHT: usize = 8;
+
+/// Accept loop: accepts inbound L2CAP connections and hands each to its own
+/// task for the pubkey exchange and pool insert.
+///
+/// One iteration is bounded by `accept()` alone. Nothing a connecting peer
+/// chooses to do can delay the next accept: the handshake runs off the loop,
+/// and when the in-flight budget is full the oldest pending handshake is
+/// aborted to make room rather than the loop waiting for one to finish.
+///
+/// The in-flight tasks live in a `JoinSet` and not behind a `Semaphore` for a
+/// reason that is easy to lose: `stop_async` aborts this task and nothing
+/// else, so dropping the `JoinSet` with it is what stops the handshakes. Bare
+/// `tokio::spawn` would leave them running past stop, able to insert into a
+/// pool that stop has just drained.
 #[allow(clippy::too_many_arguments)]
 async fn accept_loop<A>(
     mut acceptor: A,
@@ -897,13 +925,25 @@ async fn accept_loop<A>(
     A: io::BleAcceptor,
     A::Stream: 'static,
 {
+    let mut inflight: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
+    // Spawn order, so the oldest handshake is the one evicted at the budget.
+    let mut pending: std::collections::VecDeque<tokio::task::AbortHandle> =
+        std::collections::VecDeque::new();
+
     loop {
+        // Reap anything finished. Neither call waits. `retain` rather than
+        // popping the front run, so a handshake that completed out of order
+        // still frees its slot instead of being aborted as the oldest later.
+        while inflight.try_join_next().is_some() {}
+        pending.retain(|handle| !handle.is_finished());
+
         match acceptor.accept().await {
             Ok(stream) => {
                 let addr = stream.remote_addr().clone();
                 let ta = addr.to_transport_addr();
 
-                // Skip if already connected (outbound won the race)
+                // Skip if already connected (outbound won the race). This
+                // awaits only our own mutex, never the peer.
                 {
                     let pool_guard = pool.lock().await;
                     if pool_guard.contains(&ta) {
@@ -912,118 +952,29 @@ async fn accept_loop<A>(
                     }
                 }
 
-                let send_mtu = stream.send_mtu();
+                if pending.len() >= INBOUND_HANDSHAKE_INFLIGHT
-                let recv_mtu = stream.recv_mtu();
+                    && let Some(oldest) = pending.pop_front()
-                let stream = Arc::new(stream);
+                {
-                // One reader across both phases — see `pubkey_exchange`.
+                    oldest.abort();
-                let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
+                    stats.record_handshake_aborted();
-
+                    debug!(
-                // Pre-handshake pubkey exchange (temporary, pre-XX)
+                        addr = %ta,
-                let mut peer_node_addr: Option<NodeAddr> = None;
+                        budget = INBOUND_HANDSHAKE_INFLIGHT,
-                if let Some(ref our_pubkey) = local_pubkey {
+                        "BLE inbound handshake budget full, aborting the oldest"
-                    match pubkey_exchange(stream.as_ref(), &mut reader, our_pubkey).await {
+                    );
-                        Ok(peer_pubkey) => {
-                            debug!(addr = %ta, "BLE inbound pubkey exchange complete");
-                            let peer_node = NodeAddr::from_pubkey(&peer_pubkey);
-                            peer_node_addr = Some(peer_node);
-                            let announced = announced_addr(&pool, &peer_node, &addr).await;
-                            neighbor_buffer.add_peer_with_pubkey(&announced, peer_pubkey);
-
-                            // 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
-                                && existing != ta
-                            {
-                                debug!(
-                                    addr = %ta,
-                                    role = "peripheral",
-                                    outcome = "duplicate-node-decline",
-                                    existing = %existing,
-                                    "BLE inbound: peer already connected on another address, dropping duplicate"
-                                );
-                                stats.record_duplicate_node_decline();
-                                continue;
-                            }
-
-                            // Cross-probe tie-breaker: smaller NodeAddr's
-                            // outbound wins. If we're smaller, our outbound
-                            // should win — drop this inbound.
-                            if let Some(ref our_addr) = local_node_addr
-                                && our_addr < &peer_node
-                            {
-                                stats.record_tiebreaker_drop();
-                                debug!(
-                                    addr = %ta,
-                                    role = "peripheral",
-                                    outcome = "tiebreaker-drop",
-                                    "BLE inbound tie-breaker: dropping (our addr < peer, outbound wins)"
-                                );
-                                continue;
-                            }
-                        }
-                        Err(e) => {
-                            stats.record_pubkey_exchange_failure();
-                            debug!(
-                                addr = %ta, role = "peripheral",
-                                outcome = "pubkey-exchange-failed", error = %e,
-                                "BLE inbound pubkey exchange failed"
-                            );
-                            continue;
-                        }
-                    }
                 }
 
-                // Spawn receive loop
+                let handle = inflight.spawn(admit_inbound(
-                let recv_task = tokio::spawn(receive_loop(
+                    stream,
-                    reader,
-                    ta.clone(),
                     Arc::clone(&pool),
                     packet_tx.clone(),
                     transport_id,
                     Arc::clone(&stats),
-                    recv_mtu,
+                    local_pubkey,
+                    Arc::clone(&neighbor_buffer),
+                    local_node_addr,
                 ));
-
+                pending.push_back(handle);
-                let conn = BleConnection {
-                    stream,
-                    recv_task: Some(recv_task),
-                    send_mtu,
-                    recv_mtu,
-                    established_at: tokio::time::Instant::now(),
-                    is_static: false,
-                    addr,
-                    node_addr: peer_node_addr,
-                };
-
-                let mut pool_guard = pool.lock().await;
-                match pool_guard.insert(ta.clone(), conn) {
-                    Ok(Some(evicted)) => {
-                        stats.record_pool_eviction();
-                        info!(addr = %ta, evicted = %evicted, "BLE inbound accepted (evicted peer)");
-                    }
-                    Ok(None) => {
-                        info!(addr = %ta, send_mtu, recv_mtu, "BLE inbound connection accepted");
-                    }
-                    Err(e) => {
-                        stats.record_connection_rejected();
-                        warn!(
-                            addr = %ta, role = "peripheral", outcome = "pool-rejected",
-                            error = %e, "BLE pool full, inbound connection rejected"
-                        );
-                        continue;
-                    }
-                }
-                stats.record_connection_accepted();
             }
             Err(e) => {
                 warn!(error = %e, "BLE accept error");
@@ -1033,6 +984,142 @@ async fn accept_loop<A>(
     }
 }
 
+/// Run one inbound connection's pubkey exchange and admit it to the pool.
+///
+/// Runs off the accept loop so a peer that never answers delays nobody else.
+/// Everything the accept loop used to do inline lives here, including the
+/// duplicate-node decline and the cross-probe tie-break that `fix/platform-ble`
+/// added: moving the work off the loop must not drop the checks that guard it.
+/// The loop's `continue` becomes `return` — this task admits one connection.
+#[allow(clippy::too_many_arguments)]
+async fn admit_inbound<S>(
+    stream: S,
+    pool: Arc<Mutex<ConnectionPool<Arc<S>>>>,
+    packet_tx: PacketTx,
+    transport_id: TransportId,
+    stats: Arc<BleStats>,
+    local_pubkey: Option<[u8; 32]>,
+    neighbor_buffer: Arc<NeighborBuffer>,
+    local_node_addr: Option<NodeAddr>,
+) where
+    S: BleStream + 'static,
+{
+    let addr = stream.remote_addr().clone();
+    let ta = addr.to_transport_addr();
+    let send_mtu = stream.send_mtu();
+    let recv_mtu = stream.recv_mtu();
+    let stream = Arc::new(stream);
+    // One reader across both phases — see `pubkey_exchange`.
+    let mut reader = BleStreamRead::new(Arc::clone(&stream), recv_mtu);
+
+    // Pre-handshake pubkey exchange (temporary, pre-XX)
+    let mut peer_node_addr: Option<NodeAddr> = None;
+    if let Some(ref our_pubkey) = local_pubkey {
+        match pubkey_exchange(stream.as_ref(), &mut reader, our_pubkey).await {
+            Ok(peer_pubkey) => {
+                debug!(addr = %ta, "BLE inbound pubkey exchange complete");
+                let peer_node = NodeAddr::from_pubkey(&peer_pubkey);
+                peer_node_addr = Some(peer_node);
+                let announced = announced_addr(&pool, &peer_node, &addr).await;
+                neighbor_buffer.add_peer_with_pubkey(&announced, peer_pubkey);
+
+                // 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
+                    && existing != ta
+                {
+                    debug!(
+                        addr = %ta,
+                        role = "peripheral",
+                        outcome = "duplicate-node-decline",
+                        existing = %existing,
+                        "BLE inbound: peer already connected on another address, dropping duplicate"
+                    );
+                    stats.record_duplicate_node_decline();
+                    return;
+                }
+
+                // Cross-probe tie-breaker: smaller NodeAddr's
+                // outbound wins. If we're smaller, our outbound
+                // should win — drop this inbound.
+                if let Some(ref our_addr) = local_node_addr
+                    && our_addr < &peer_node
+                {
+                    stats.record_tiebreaker_drop();
+                    debug!(
+                        addr = %ta,
+                        role = "peripheral",
+                        outcome = "tiebreaker-drop",
+                        "BLE inbound tie-breaker: dropping (our addr < peer, outbound wins)"
+                    );
+                    return;
+                }
+            }
+            Err(e) => {
+                stats.record_pubkey_exchange_failure();
+                debug!(
+                    addr = %ta, role = "peripheral",
+                    outcome = "pubkey-exchange-failed", error = %e,
+                    "BLE inbound pubkey exchange failed"
+                );
+                return;
+            }
+        }
+    }
+
+    // Spawn receive loop
+    let recv_task = tokio::spawn(receive_loop(
+        reader,
+        ta.clone(),
+        Arc::clone(&pool),
+        packet_tx.clone(),
+        transport_id,
+        Arc::clone(&stats),
+        recv_mtu,
+    ));
+
+    let conn = BleConnection {
+        stream,
+        recv_task: Some(recv_task),
+        send_mtu,
+        recv_mtu,
+        established_at: tokio::time::Instant::now(),
+        is_static: false,
+        addr,
+        node_addr: peer_node_addr,
+    };
+
+    let mut pool_guard = pool.lock().await;
+    match pool_guard.insert(ta.clone(), conn) {
+        Ok(Some(evicted)) => {
+            stats.record_pool_eviction();
+            info!(addr = %ta, evicted = %evicted, "BLE inbound accepted (evicted peer)");
+        }
+        Ok(None) => {
+            info!(addr = %ta, send_mtu, recv_mtu, "BLE inbound connection accepted");
+        }
+        Err(e) => {
+            stats.record_connection_rejected();
+            warn!(
+                addr = %ta, role = "peripheral", outcome = "pool-rejected",
+                error = %e, "BLE pool full, inbound connection rejected"
+            );
+            return;
+        }
+    }
+    stats.record_connection_accepted();
+}
+
 /// Receive loop: reads packets from a BLE stream and delivers to node.
 ///
 /// Takes the connection's `BleStreamRead` — already positioned past the
@@ -2650,10 +2737,183 @@ mod tests {
             "advertisements_sent",
             "scan_results",
             "duplicate_node_declines",
+            "handshakes_aborted",
         ];
         for key in expected {
             assert!(object.contains_key(key), "snapshot lost `{key}`");
         }
         assert_eq!(object.len(), expected.len(), "snapshot gained a key");
     }
+
+    /// A secret/public key pair from a fixed seed.
+    ///
+    /// The exchange parses the peer's 32 bytes with `XOnlyPublicKey::from_slice`,
+    /// so arbitrary bytes will not do.
+    fn test_keypair(seed: u8) -> ([u8; 32], XOnlyPublicKey) {
+        let secp = secp256k1::Secp256k1::new();
+        let sk = secp256k1::SecretKey::from_slice(&[seed; 32]).unwrap();
+        let (xonly, _) = sk.public_key(&secp).x_only_public_key();
+        (xonly.serialize(), xonly)
+    }
+
+    /// Inject an inbound connection and return the peer end of the link.
+    ///
+    /// The peer end must be kept alive: dropping it closes the channel, which
+    /// the mock reports as a zero-length read rather than as silence.
+    async fn connect_inbound(
+        transport: &BleTransport<MockBleIo>,
+        peer: &BleAddr,
+    ) -> io::MockBleStream {
+        let (inbound, peer_end) = io::MockBleStream::pair(test_addr(1), peer.clone(), 512);
+        transport.io.inject_inbound(inbound).await;
+        peer_end
+    }
+
+    /// Complete the peer half of the pubkey exchange.
+    async fn send_pubkey(stream: &io::MockBleStream, pubkey: &XOnlyPublicKey) {
+        let mut msg = [0u8; PUBKEY_EXCHANGE_SIZE];
+        msg[0] = PUBKEY_EXCHANGE_PREFIX;
+        msg[1..].copy_from_slice(&pubkey.serialize());
+        stream.send(&msg).await.unwrap();
+    }
+
+    /// Poll the discovery buffer until every wanted address has appeared.
+    ///
+    /// Asserts on the discovery buffer rather than the pool because the buffer
+    /// is populated before the cross-probe tie-breaker, which drops an inbound
+    /// whose NodeAddr sorts above ours and would make the result depend on the
+    /// keys the test happened to pick. Sleeps rather than yields, so a paused
+    /// clock advances instead of the runtime staying busy forever.
+    async fn wait_for_discovered(buffer: &NeighborBuffer, wanted: &[BleAddr]) -> bool {
+        let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
+        for _ in 0..20_000 {
+            for peer in buffer.take() {
+                if let Some(addr) = peer.addr.as_str() {
+                    seen.insert(addr.to_string());
+                }
+            }
+            if wanted.iter().all(|a| seen.contains(&a.to_string_repr())) {
+                return true;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+        }
+        false
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_silent_inbound_peer_does_not_delay_the_next_accept() {
+        let io = MockBleIo::new("hci0", test_addr(1));
+        let (mut transport, _rx) = make_transport(io);
+        let (our_pubkey, _) = test_keypair(1);
+        transport.set_local_pubkey(our_pubkey);
+        transport.start_async().await.unwrap();
+
+        // A connects and never says anything.
+        let _silent = connect_inbound(&transport, &test_addr(2)).await;
+
+        // B connects behind it and completes the exchange at once.
+        let good = connect_inbound(&transport, &test_addr(3)).await;
+        let (_, peer_pubkey) = test_keypair(7);
+        send_pubkey(&good, &peer_pubkey).await;
+
+        let start = tokio::time::Instant::now();
+        assert!(
+            wait_for_discovered(&transport.neighbor_buffer, &[test_addr(3)]).await,
+            "the well-behaved peer was never admitted"
+        );
+        let elapsed = start.elapsed();
+        assert!(
+            elapsed < std::time::Duration::from_secs(1),
+            "the well-behaved peer waited {:?} on the silent one's handshake deadline",
+            elapsed
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_flood_of_silent_connectors_does_not_delay_a_well_behaved_one() {
+        // The test that breaks what the guard guards: it fails against a bound
+        // that waits for a slot rather than reclaiming one.
+        let io = MockBleIo::new("hci0", test_addr(1));
+        let (mut transport, _rx) = make_transport(io);
+        let (our_pubkey, _) = test_keypair(1);
+        transport.set_local_pubkey(our_pubkey);
+        transport.start_async().await.unwrap();
+
+        let mut silent = Vec::new();
+        for n in 0..INBOUND_HANDSHAKE_INFLIGHT + 1 {
+            silent.push(connect_inbound(&transport, &test_addr(20 + n as u8)).await);
+        }
+
+        let good = connect_inbound(&transport, &test_addr(3)).await;
+        let (_, peer_pubkey) = test_keypair(7);
+        send_pubkey(&good, &peer_pubkey).await;
+
+        let start = tokio::time::Instant::now();
+        assert!(
+            wait_for_discovered(&transport.neighbor_buffer, &[test_addr(3)]).await,
+            "the well-behaved peer was never admitted behind the flood"
+        );
+        let elapsed = start.elapsed();
+        assert!(
+            elapsed < std::time::Duration::from_secs(1),
+            "the well-behaved peer waited {:?} behind {} silent connectors",
+            elapsed,
+            silent.len()
+        );
+        assert!(transport.stats.snapshot().handshakes_aborted > 0);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn inbound_peers_within_the_handshake_budget_are_all_admitted() {
+        // The guard must not red a legitimately clean run.
+        let io = MockBleIo::new("hci0", test_addr(1));
+        let (mut transport, _rx) = make_transport(io);
+        let (our_pubkey, _) = test_keypair(1);
+        transport.set_local_pubkey(our_pubkey);
+        transport.start_async().await.unwrap();
+
+        let mut peers = Vec::new();
+        let mut wanted = Vec::new();
+        for n in 0..INBOUND_HANDSHAKE_INFLIGHT {
+            let addr = test_addr(40 + n as u8);
+            let stream = connect_inbound(&transport, &addr).await;
+            let (_, peer_pubkey) = test_keypair(10 + n as u8);
+            send_pubkey(&stream, &peer_pubkey).await;
+            peers.push(stream);
+            wanted.push(addr);
+        }
+
+        assert!(
+            wait_for_discovered(&transport.neighbor_buffer, &wanted).await,
+            "a well-behaved peer inside the budget was not admitted"
+        );
+        assert_eq!(transport.stats.snapshot().handshakes_aborted, 0);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn the_pubkey_exchange_send_half_has_a_deadline() {
+        // The mock's link is a 64-slot channel, so a peer that never reads
+        // parks our write once it is full — which is what an L2CAP peer that
+        // stops draining does.
+        let (ours, _peer) = io::MockBleStream::pair(test_addr(1), test_addr(2), 512);
+        for _ in 0..64 {
+            ours.send(&[0u8; 1]).await.unwrap();
+        }
+        let (our_pubkey, _) = test_keypair(1);
+        // `fix/platform-ble` split the read half out; the exchange takes it as
+        // its own argument now. The deadline under test is on the send half,
+        // which never reaches a read, so the reader is only here to type-check.
+        let recv_mtu = ours.recv_mtu();
+        let ours = Arc::new(ours);
+        let mut reader = BleStreamRead::new(Arc::clone(&ours), recv_mtu);
+
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(60),
+            pubkey_exchange(ours.as_ref(), &mut reader, &our_pubkey),
+        )
+        .await
+        .expect("the pubkey exchange send half parked with no deadline of its own");
+
+        assert!(matches!(result, Err(TransportError::Timeout)));
+    }
 }

src/transport/ble/stats.rs

@@ -27,6 +27,7 @@ pub struct BleStats {
     pub connections_established: AtomicU64,
     pub connections_accepted: AtomicU64,
     pub connections_rejected: AtomicU64,
+    pub handshakes_aborted: AtomicU64,
     pub connect_timeouts: AtomicU64,
     /// Outbound connects that failed with an error rather than timing out.
     pub connect_errors: AtomicU64,
@@ -58,6 +59,7 @@ impl BleStats {
             connections_established: AtomicU64::new(0),
             connections_accepted: AtomicU64::new(0),
             connections_rejected: AtomicU64::new(0),
+            handshakes_aborted: AtomicU64::new(0),
             connect_timeouts: AtomicU64::new(0),
             connect_errors: AtomicU64::new(0),
             pubkey_exchange_failures: AtomicU64::new(0),
@@ -176,6 +178,11 @@ impl BleStats {
         self.duplicate_node_declines.fetch_add(1, Ordering::Relaxed);
     }
 
+    /// Record an inbound handshake aborted to free an in-flight slot.
+    pub fn record_handshake_aborted(&self) {
+        self.handshakes_aborted.fetch_add(1, Ordering::Relaxed);
+    }
+
     /// Take a snapshot of all counters.
     pub fn snapshot(&self) -> BleStatsSnapshot {
         BleStatsSnapshot {
@@ -189,6 +196,7 @@ impl BleStats {
             connections_established: self.connections_established.load(Ordering::Relaxed),
             connections_accepted: self.connections_accepted.load(Ordering::Relaxed),
             connections_rejected: self.connections_rejected.load(Ordering::Relaxed),
+            handshakes_aborted: self.handshakes_aborted.load(Ordering::Relaxed),
             connect_timeouts: self.connect_timeouts.load(Ordering::Relaxed),
             connect_errors: self.connect_errors.load(Ordering::Relaxed),
             pubkey_exchange_failures: self.pubkey_exchange_failures.load(Ordering::Relaxed),
@@ -221,6 +229,7 @@ pub struct BleStatsSnapshot {
     pub connections_established: u64,
     pub connections_accepted: u64,
     pub connections_rejected: u64,
+    pub handshakes_aborted: u64,
     pub connect_timeouts: u64,
     pub connect_errors: u64,
     pub pubkey_exchange_failures: u64,