refactor(runtime-host): fence peer connection attempts

native/runtime-host-peer/Cargo.toml

@@ -46,6 +46,7 @@ libp2p = { version = "0.56", features = [
 napi = { version = "3.12", features = ["napi8", "tokio_rt"] }
 napi-derive = "3.5"
 tokio = { version = "1.53", features = ["fs", "io-util", "rt-multi-thread", "sync", "time"] }
+tokio-util = { version = "0.7", features = ["rt"] }
 
 [build-dependencies]
 napi-build = "2.4"

native/runtime-host-peer/src/engine.rs

@@ -39,6 +39,7 @@ use libp2p::{
     tcp, yamux,
 };
 use tokio::sync::{mpsc, oneshot};
+use tokio_util::sync::CancellationToken;
 
 mod address;
 mod application_stream;
@@ -199,6 +200,7 @@ struct Behaviour {
 }
 
 struct PendingConnect {
+    attempt_id: ConnectAttemptId,
     peer_id: PeerId,
     result: oneshot::Sender<Result<PeerStream, PeerError>>,
     stream_kind: StreamKind,
@@ -212,8 +214,12 @@ struct PendingConnect {
     transit_after: Instant,
     next_route_attempt: Instant,
     retry_coordination: bool,
+    cancellation: CancellationToken,
 }
 
+#[derive(Clone, Copy, PartialEq, Eq)]
+struct ConnectAttemptId(u64);
+
 #[derive(Clone, Copy, PartialEq, Eq)]
 pub enum StreamKind {
     Application,
@@ -262,11 +268,47 @@ impl relay::RateLimiter for AllowedPeerLimiter {
 
 #[derive(Default)]
 struct DirectConnectState {
+    next_attempt_id: u64,
     pending: HashMap<u32, PendingConnect>,
     active: HashMap<ConnectionId, usize>,
     retiring_connections: HashSet<ConnectionId>,
 }
 
+enum PendingAttemptAdmission {
+    Active(PendingConnect),
+    Expired(PendingConnect),
+}
+
+impl DirectConnectState {
+    fn allocate_attempt_id(&mut self) -> ConnectAttemptId {
+        let attempt_id = ConnectAttemptId(self.next_attempt_id);
+        self.next_attempt_id += 1;
+        attempt_id
+    }
+
+    fn take_pending_attempt(
+        &mut self,
+        request_id: u32,
+        attempt_id: ConnectAttemptId,
+        now: Instant,
+    ) -> Option<PendingAttemptAdmission> {
+        if self
+            .pending
+            .get(&request_id)
+            .is_none_or(|pending| pending.attempt_id != attempt_id)
+        {
+            return None;
+        }
+        self.pending.remove(&request_id).map(|pending| {
+            if pending.deadline <= now {
+                PendingAttemptAdmission::Expired(pending)
+            } else {
+                PendingAttemptAdmission::Active(pending)
+            }
+        })
+    }
+}
+
 struct CoordinationRelay {
     addresses: Vec<Multiaddr>,
     automatic_addresses: Vec<Multiaddr>,
@@ -368,6 +410,7 @@ impl CoordinationRelay {
 
 struct OpenedStream {
     request_id: u32,
+    attempt_id: ConnectAttemptId,
     result: Result<application_stream::OpenedStream, String>,
 }
 
@@ -656,7 +699,9 @@ async fn run_endpoint_async(
                     };
                     let retry_coordination = stream_kind == StreamKind::Application
                         && stream_control.has_relayed_connection(options.peer_id);
+                    let attempt_id = direct.allocate_attempt_id();
                     direct.pending.insert(request_id, PendingConnect {
+                        attempt_id,
                         peer_id: options.peer_id,
                         result,
                         stream_kind,
@@ -670,6 +715,7 @@ async fn run_endpoint_async(
                         transit_after,
                         next_route_attempt: Instant::now(),
                         retry_coordination,
+                        cancellation: CancellationToken::new(),
                     });
                     retry_connect_routes(
                         &mut swarm,
@@ -847,7 +893,27 @@ async fn run_endpoint_async(
             }
             Some(opened) = opened_rx.recv() => {
                 let request_id = opened.request_id;
-                if let Some(mut waiter) = direct.pending.remove(&opened.request_id) {
+                let Some(admission) = direct.take_pending_attempt(
+                    opened.request_id,
+                    opened.attempt_id,
+                    Instant::now(),
+                ) else {
+                    continue;
+                };
+                let mut waiter = match admission {
+                    PendingAttemptAdmission::Active(waiter) => waiter,
+                    PendingAttemptAdmission::Expired(waiter) => {
+                        let error = pending_connect_deadline_error(&waiter);
+                        fail_pending_connect(
+                            &mut swarm,
+                            &mut direct,
+                            &mut coordination_relays,
+                            waiter,
+                            error,
+                        );
+                        continue;
+                    }
+                };
                     match opened.result {
                         Ok(opened) => {
                             if waiter.stream_kind == StreamKind::Application
@@ -880,6 +946,7 @@ async fn run_endpoint_async(
                                 );
                                 continue;
                             }
+                            waiter.cancellation.cancel();
                             let result = match waiter.stream_kind {
                             StreamKind::Application => {
                                 let connection_id = opened.connection_id;
@@ -949,7 +1016,6 @@ async fn run_endpoint_async(
                             );
                         }
                     }
-                }
             }
             event = swarm.select_next_some() => {
                 handle_swarm_event(
@@ -1024,27 +1090,13 @@ async fn run_endpoint_async(
                     .collect::<Vec<_>>();
                 for request_id in expired {
                     if let Some(waiter) = direct.pending.remove(&request_id) {
-                        let (code, message) = match waiter.stream_kind {
-                            StreamKind::Application
-                                if !waiter.transit_relay_peers.is_empty() => (
-                                "transit_unavailable",
-                                "no direct or approved transit path was established before the deadline",
-                            ),
-                            StreamKind::Application => (
-                                "direct_path_unavailable",
-                                "no direct path was established before the deadline",
-                            ),
-                            StreamKind::MeshControl => (
-                                "mesh_control_unavailable",
-                                "no Mesh control path was established before the deadline",
-                            ),
-                        };
+                        let error = pending_connect_deadline_error(&waiter);
                         fail_pending_connect(
                             &mut swarm,
                             &mut direct,
                             &mut coordination_relays,
                             waiter,
-                            PeerError::new(code, message),
+                            error,
                         );
                     }
                 }
@@ -1053,6 +1105,24 @@ async fn run_endpoint_async(
     }
 }
 
+fn pending_connect_deadline_error(waiter: &PendingConnect) -> PeerError {
+    let (code, message) = match waiter.stream_kind {
+        StreamKind::Application if !waiter.transit_relay_peers.is_empty() => (
+            "transit_unavailable",
+            "no direct or approved transit path was established before the deadline",
+        ),
+        StreamKind::Application => (
+            "direct_path_unavailable",
+            "no direct path was established before the deadline",
+        ),
+        StreamKind::MeshControl => (
+            "mesh_control_unavailable",
+            "no Mesh control path was established before the deadline",
+        ),
+    };
+    PeerError::new(code, message)
+}
+
 type BuiltSwarm = (
     Swarm<Behaviour>,
     application_stream::Control,
@@ -1293,17 +1363,27 @@ fn maybe_open_peer_stream(
         return;
     }
     let stream_kind = waiter.stream_kind;
+    let attempt_id = waiter.attempt_id;
+    let cancellation = waiter.cancellation.clone();
     let retiring_connections = retiring_connections.clone();
     waiter.opening = Some(tokio::spawn(async move {
         let control = match stream_kind {
             StreamKind::Application => &mut application_control,
             StreamKind::MeshControl => &mut mesh_control,
         };
-        let result = control
-            .open_stream(peer_id, &retiring_connections, &eligible_relay_peers)
-            .await
-            .map_err(|error| error.to_string());
-        let _ = opened_tx.send(OpenedStream { request_id, result }).await;
+        let result = tokio::select! {
+            _ = cancellation.cancelled() => return,
+            result = control.open_stream(peer_id, &retiring_connections, &eligible_relay_peers) => {
+                result.map_err(|error| error.to_string())
+            }
+        };
+        let _ = opened_tx
+            .send(OpenedStream {
+                request_id,
+                attempt_id,
+                result,
+            })
+            .await;
     }));
 }
 
@@ -1781,12 +1861,10 @@ fn fail_pending_connect(
     swarm: &mut Swarm<Behaviour>,
     direct: &mut DirectConnectState,
     coordination_relays: &mut HashMap<PeerId, CoordinationRelay>,
-    mut waiter: PendingConnect,
+    waiter: PendingConnect,
     error: PeerError,
 ) {
-    if let Some(opening) = waiter.opening.take() {
-        opening.abort();
-    }
+    waiter.cancellation.cancel();
     retire_direct_dials(swarm, &mut direct.retiring_connections, waiter.dials, None);
     release_coordination_relays(
         swarm,
@@ -2591,6 +2669,40 @@ fn native_error(error: impl std::fmt::Display) -> PeerError {
 mod tests {
     use super::*;
 
+    #[test]
+    fn completion_at_the_immutable_deadline_cannot_commit() {
+        let now = Instant::now();
+        let mut direct = DirectConnectState::default();
+        let attempt_id = direct.allocate_attempt_id();
+        let (result, _response) = oneshot::channel();
+        direct.pending.insert(
+            7,
+            PendingConnect {
+                attempt_id,
+                peer_id: PeerId::random(),
+                result,
+                stream_kind: StreamKind::Application,
+                deadline: now,
+                opening: None,
+                dials: HashMap::new(),
+                direct_routes: Vec::new(),
+                coordination_relays: Vec::new(),
+                coordination_relay_peers: Vec::new(),
+                transit_relay_peers: HashSet::new(),
+                transit_after: now,
+                next_route_attempt: now,
+                retry_coordination: false,
+                cancellation: CancellationToken::new(),
+            },
+        );
+
+        assert!(matches!(
+            direct.take_pending_attempt(7, attempt_id, now),
+            Some(PendingAttemptAdmission::Expired(_))
+        ));
+        assert!(direct.pending.is_empty());
+    }
+
     #[tokio::test]
     async fn identity_signature_is_bound_to_peer_and_payload() {
         let root = std::env::temp_dir().join(format!("maka-peer-signature-{}", PeerId::random()));
@@ -2946,6 +3058,35 @@ mod tests {
         };
         assert_eq!(error.code, "peer_connect_cancelled");
 
+        let source_stream = connect_test_stream(
+            &source,
+            target.peer_id,
+            target
+                .listen_addresses
+                .first()
+                .expect("target retry route")
+                .clone(),
+            4,
+            StreamKind::Application,
+        )
+        .await;
+        let mut target_stream =
+            tokio::time::timeout(Duration::from_secs(5), target.incoming.recv())
+                .await
+                .expect("retry inbound timeout")
+                .expect("retry inbound stream");
+        write_test_stream(&source_stream, b"cancelled-request-id-reused").await;
+        assert_eq!(
+            tokio::time::timeout(Duration::from_secs(5), target_stream.incoming.recv())
+                .await
+                .expect("retry read timeout")
+                .expect("retry stream ended")
+                .expect("retry read failed"),
+            b"cancelled-request-id-reused",
+        );
+        close_test_stream(source_stream).await;
+        close_test_stream(target_stream).await;
+
         stop_test_endpoint(source).await;
         stop_test_endpoint(target).await;
         stop_test_endpoint(relay).await;

packages/cli/RUNTIME_HOST_PEER_DEPENDENCIES.rust.tsv

@@ -221,6 +221,7 @@ tinyvec@1.12.0	X				X			X
 tinyvec_macros@0.1.1	X				X			X
 tokio@1.53.1					X			
 tokio-macros@2.7.2					X			
+tokio-util@0.7.19					X			
 tracing@0.1.44					X			
 tracing-attributes@0.1.31					X			
 tracing-core@0.1.36					X			

packages/cli/RUNTIME_HOST_PEER_THIRD_PARTY_NOTICES.txt

@@ -5,8 +5,8 @@ Generated by scripts/generate-runtime-host-peer-notices.mjs from the exact
 four-target production dependency inventory. Do not edit this file by hand.
 
 Manifest: native/runtime-host-peer/Cargo.toml
-Cargo.lock SHA-256: f475d62519bef1d83c2e33b17f1b1eeffe7a03370b041dabea887b7f57aa43ff
-Inventory SHA-256: 4078d1a287187fbfe21e4937305dabf6e81afac6f2f45388b47a04f311cd8223
+Cargo.lock SHA-256: fdfaf86fa6b2d4ad405959e6cbb4a8d4b3f77a26bc24215aa3f840b61aaae0e3
+Inventory SHA-256: b33f38e600eaf377fb15ddcfeb787a11e1d30e95a1b904597b306419a7614f28
 
 Packages
 --------
@@ -1482,6 +1482,12 @@ SPDX license: MIT
 Source: https://github.com/tokio-rs/tokio
 License text: c0fdcda1a4ffc5fd63c452e40ce25e54f2098669e3415033b228773eea24a871 (LICENSE)
 
+tokio-util@0.7.19
+-----------------
+SPDX license: MIT
+Source: https://github.com/tokio-rs/tokio
+License text: 3e1bef82aa0dfee4504ad211f7e4895153e3191db6ab179f94a7e20f990a0b67 (LICENSE)
+
 tracing@0.1.44
 --------------
 SPDX license: MIT
@@ -3799,6 +3805,7 @@ DEALINGS IN THE SOFTWARE.
 3e1bef82aa0dfee4504ad211f7e4895153e3191db6ab179f94a7e20f990a0b67
 ----------------------------------------------------------------
 Used by:
+- tokio-util@0.7.19 (LICENSE)
 - tokio@1.53.1 (LICENSE)
 
 MIT License