From f900c8e2e343523a0ab624534b09d496100b2ca7 Mon Sep 17 00:00:00 2001 From: Wang Date: Mon, 31 Aug 2026 23:53:29 +0800 Subject: [PATCH 1/2] refactor(runtime-host): fence peer connection attempts --- native/runtime-host-peer/Cargo.lock | 2 + native/runtime-host-peer/Cargo.toml | 1 + native/runtime-host-peer/src/engine.rs | 94 +++++++++++++++++++++++--- 3 files changed, 87 insertions(+), 10 deletions(-) diff --git a/native/runtime-host-peer/Cargo.lock b/native/runtime-host-peer/Cargo.lock index ca3fffa36f..1fb8cbdcef 100644 --- a/native/runtime-host-peer/Cargo.lock +++ b/native/runtime-host-peer/Cargo.lock @@ -1753,6 +1753,7 @@ dependencies = [ "napi-build", "napi-derive", "tokio", + "tokio-util", ] [[package]] @@ -2941,6 +2942,7 @@ dependencies = [ "bytes", "futures-core", "futures-sink", + "futures-util", "libc", "pin-project-lite", "tokio", diff --git a/native/runtime-host-peer/Cargo.toml b/native/runtime-host-peer/Cargo.toml index 35c4f35a36..5a6cec0117 100644 --- a/native/runtime-host-peer/Cargo.toml +++ b/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" diff --git a/native/runtime-host-peer/src/engine.rs b/native/runtime-host-peer/src/engine.rs index 9e51cf9b31..163cfd3b51 100644 --- a/native/runtime-host-peer/src/engine.rs +++ b/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>, 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,35 @@ impl relay::RateLimiter for AllowedPeerLimiter { #[derive(Default)] struct DirectConnectState { + next_attempt_id: u64, pending: HashMap, active: HashMap, retiring_connections: HashSet, } +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, + ) -> Option { + if self + .pending + .get(&request_id) + .is_none_or(|pending| pending.attempt_id != attempt_id) + { + return None; + } + self.pending.remove(&request_id) + } +} + struct CoordinationRelay { addresses: Vec, automatic_addresses: Vec, @@ -368,6 +398,7 @@ impl CoordinationRelay { struct OpenedStream { request_id: u32, + attempt_id: ConnectAttemptId, result: Result, } @@ -656,7 +687,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 +703,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 +881,9 @@ 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) { + if let Some(mut waiter) = + direct.take_pending_attempt(opened.request_id, opened.attempt_id) + { match opened.result { Ok(opened) => { if waiter.stream_kind == StreamKind::Application @@ -880,6 +916,7 @@ async fn run_endpoint_async( ); continue; } + waiter.cancellation.cancel(); let result = match waiter.stream_kind { StreamKind::Application => { let connection_id = opened.connection_id; @@ -1293,17 +1330,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 +1828,10 @@ fn fail_pending_connect( swarm: &mut Swarm, direct: &mut DirectConnectState, coordination_relays: &mut HashMap, - 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, @@ -2946,6 +2991,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; From 602cbb89a11f13879b8ddbb167c2816e17d11118 Mon Sep 17 00:00:00 2001 From: Wang Date: Tue, 1 Sep 2026 05:20:30 +0800 Subject: [PATCH 2/2] fix(runtime-host): reject peer completions after deadline --- native/runtime-host-peer/src/engine.rs | 111 ++++++++++++++---- .../RUNTIME_HOST_PEER_DEPENDENCIES.rust.tsv | 1 + .../RUNTIME_HOST_PEER_THIRD_PARTY_NOTICES.txt | 11 +- 3 files changed, 99 insertions(+), 24 deletions(-) diff --git a/native/runtime-host-peer/src/engine.rs b/native/runtime-host-peer/src/engine.rs index 163cfd3b51..6b02f0dc23 100644 --- a/native/runtime-host-peer/src/engine.rs +++ b/native/runtime-host-peer/src/engine.rs @@ -274,6 +274,11 @@ struct DirectConnectState { retiring_connections: HashSet, } +enum PendingAttemptAdmission { + Active(PendingConnect), + Expired(PendingConnect), +} + impl DirectConnectState { fn allocate_attempt_id(&mut self) -> ConnectAttemptId { let attempt_id = ConnectAttemptId(self.next_attempt_id); @@ -285,7 +290,8 @@ impl DirectConnectState { &mut self, request_id: u32, attempt_id: ConnectAttemptId, - ) -> Option { + now: Instant, + ) -> Option { if self .pending .get(&request_id) @@ -293,7 +299,13 @@ impl DirectConnectState { { return None; } - self.pending.remove(&request_id) + self.pending.remove(&request_id).map(|pending| { + if pending.deadline <= now { + PendingAttemptAdmission::Expired(pending) + } else { + PendingAttemptAdmission::Active(pending) + } + }) } } @@ -881,9 +893,27 @@ async fn run_endpoint_async( } Some(opened) = opened_rx.recv() => { let request_id = opened.request_id; - if let Some(mut waiter) = - direct.take_pending_attempt(opened.request_id, opened.attempt_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 @@ -986,7 +1016,6 @@ async fn run_endpoint_async( ); } } - } } event = swarm.select_next_some() => { handle_swarm_event( @@ -1061,27 +1090,13 @@ async fn run_endpoint_async( .collect::>(); 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, ); } } @@ -1090,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, application_stream::Control, @@ -2636,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())); diff --git a/packages/cli/RUNTIME_HOST_PEER_DEPENDENCIES.rust.tsv b/packages/cli/RUNTIME_HOST_PEER_DEPENDENCIES.rust.tsv index 16be61345c..2da2eca52a 100644 --- a/packages/cli/RUNTIME_HOST_PEER_DEPENDENCIES.rust.tsv +++ b/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 diff --git a/packages/cli/RUNTIME_HOST_PEER_THIRD_PARTY_NOTICES.txt b/packages/cli/RUNTIME_HOST_PEER_THIRD_PARTY_NOTICES.txt index 3af1c6add8..b0c88b163f 100644 --- a/packages/cli/RUNTIME_HOST_PEER_THIRD_PARTY_NOTICES.txt +++ b/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