From 542784d6e6de59060e538adbffb5fca2227b4acc Mon Sep 17 00:00:00 2001 From: Wang Date: Fri, 28 Aug 2026 00:40:29 +0800 Subject: [PATCH] refactor(runtime-host): share the client peer endpoint Keep one lazy peer endpoint for each Desktop or CLI owner so Direct profiles and reconnects reuse one Swarm. Cancel individual connection attempts without tearing down unrelated streams, and close the endpoint with its owner lifecycle. Generated-by: Codex --- apps/desktop/src/main/runtime-host-boot.ts | 5 + .../main/runtime-host-desktop-candidate.ts | 3 + .../src/main/runtime-host-desktop-manager.ts | 5 +- native/runtime-host-peer/src/bindings.rs | 15 + native/runtime-host-peer/src/engine.rs | 557 +++++++++++++++--- .../runtime-host-peer/src/engine/address.rs | 29 + .../src/engine/application_stream.rs | 70 ++- packages/cli/src/runtime-host-cli-context.ts | 42 +- .../src/__tests__/peer-listener.test.ts | 1 + .../src/__tests__/peer-native.test.ts | 100 ++-- .../runtime-host/src/client/host-profile.ts | 54 +- packages/runtime-host/src/client/index.ts | 6 + .../runtime-host/src/client/peer-client.ts | 190 ++++++ .../runtime-host/src/transport/peer-native.ts | 2 + scripts/smoke-release-cli-package.mjs | 4 + 15 files changed, 893 insertions(+), 190 deletions(-) create mode 100644 packages/runtime-host/src/client/peer-client.ts diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index e3aed9b74f..21f1ba6707 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -48,6 +48,7 @@ import { createClientRuntimeHostCredentialStore, createClientRuntimeHostProfileCatalog, createRuntimeHostCandidateLaunchBarrier, + createRuntimeHostPeerClientFromEnvironment, LOCAL_RUNTIME_HOST_PROFILE, loadOrCreateRuntimeHostClientInstanceId, } from "@maka/runtime-host/client"; @@ -220,6 +221,9 @@ const runtimeHostDirectPeerAvailable = await configureDesktopRuntimeHostPeerClie resourcesPath: process.resourcesPath, clientDataRoot: userDataDir, }); +const runtimeHostPeerClient = runtimeHostDirectPeerAvailable + ? createRuntimeHostPeerClientFromEnvironment() + : undefined; const runtimeHostClientInstanceId = await loadOrCreateRuntimeHostClientInstanceId( join(userDataDir, "runtime-host-client.json"), ); @@ -724,6 +728,7 @@ runtimeHostManager = await startRuntimeHostDesktopManager( clientInstanceId: runtimeHostClientInstanceId, generation: runtimeHostGeneration, candidateLaunchBarrier: runtimeHostCandidateLaunchBarrier, + ...(runtimeHostPeerClient ? { peerClient: runtimeHostPeerClient } : {}), // The Desktop E2E composition lives behind its own entry module, which // release packaging drops: picking it here is what keeps FakeBackend and // the E2E bootstrap out of the shipped Runtime Host. diff --git a/apps/desktop/src/main/runtime-host-desktop-candidate.ts b/apps/desktop/src/main/runtime-host-desktop-candidate.ts index 4b0b78e23a..bb06bb2bef 100644 --- a/apps/desktop/src/main/runtime-host-desktop-candidate.ts +++ b/apps/desktop/src/main/runtime-host-desktop-candidate.ts @@ -28,6 +28,7 @@ import { type RuntimeHostSshOperatorActivationInput, connectOrSpawnRuntimeHost, connectRemoteRuntimeHostProfile, + type RuntimeHostPeerClient, type RuntimeHostSshInteraction, type RuntimeHostSshTunnel, type RuntimeHostSshTunnelInput, @@ -176,6 +177,7 @@ export interface DesktopRuntimeHostCandidateStartInput /** Candidate-exit sink forwarded to the launcher; the Desktop owns the sink. */ readonly onExit?: (details: CandidateExitDetails) => void; readonly candidateLaunchBarrier?: RuntimeHostCandidateLaunchBarrier; + readonly peerClient?: RuntimeHostPeerClient; readonly remote?: { readonly profile: RemoteRuntimeHostProfile; readonly credential: string; @@ -388,6 +390,7 @@ async function startRemoteDesktopRuntimeHostCandidate( ? {} : { handshakeTimeoutMs: input.handshakeTimeoutMs }), readyTimeoutMs: input.electionDeadlineMs ?? 45_000, + ...(input.peerClient === undefined ? {} : { peerClient: input.peerClient }), ...(remote.sshInteraction === undefined ? {} : { sshInteraction: remote.sshInteraction }), diff --git a/apps/desktop/src/main/runtime-host-desktop-manager.ts b/apps/desktop/src/main/runtime-host-desktop-manager.ts index e5baea49cb..de733b6044 100644 --- a/apps/desktop/src/main/runtime-host-desktop-manager.ts +++ b/apps/desktop/src/main/runtime-host-desktop-manager.ts @@ -662,9 +662,12 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { const results = await Promise.allSettled( [...this.#targets.values()].map((target) => this.#removeTarget(target)), ); + const peerResults = await Promise.allSettled( + this.#baseInput.peerClient ? [this.#baseInput.peerClient.close()] : [], + ); this.#baseInput.candidateLaunchBarrier?.release(); this.#ipcMain.close(); - const failures = results.filter( + const failures = [...results, ...peerResults].filter( (result): result is PromiseRejectedResult => result.status === 'rejected', ); if (failures.length > 0) { diff --git a/native/runtime-host-peer/src/bindings.rs b/native/runtime-host-peer/src/bindings.rs index ef4d20d790..2d76035670 100644 --- a/native/runtime-host-peer/src/bindings.rs +++ b/native/runtime-host-peer/src/bindings.rs @@ -42,6 +42,7 @@ pub struct StartPeerEndpointOptions { #[napi(object)] pub struct ConnectPeerOptions { + pub request_id: u32, pub peer_id: String, pub route_hints: Vec, pub coordination_relays: Option>, @@ -88,6 +89,7 @@ impl PeerEndpoint { self.commands .send(EngineCommand::Connect { options: engine::ConnectOptions { + request_id: options.request_id, peer_id, route_hints, coordination_relays, @@ -110,6 +112,19 @@ impl PeerEndpoint { ) } + #[napi] + pub async fn cancel_connect(&self, request_id: u32) -> Result { + let (result_tx, result_rx) = oneshot::channel(); + self.commands + .send(EngineCommand::CancelConnect { + request_id, + result: result_tx, + }) + .await + .map_err(|_| native_closed_error())?; + result_rx.await.map_err(|_| native_closed_error()) + } + #[napi] pub async fn accept(&self) -> Result> { let mut incoming = self.incoming.lock().await; diff --git a/native/runtime-host-peer/src/engine.rs b/native/runtime-host-peer/src/engine.rs index ee59f73466..b39ac5178d 100644 --- a/native/runtime-host-peer/src/engine.rs +++ b/native/runtime-host-peer/src/engine.rs @@ -31,7 +31,10 @@ use libp2p::{ dcutr, identify, identity, multiaddr::Protocol, noise, ping, relay, - swarm::{ConnectionId, NetworkBehaviour, SwarmEvent}, + swarm::{ + ConnectionId, NetworkBehaviour, SwarmEvent, + dial_opts::{DialOpts, PeerCondition}, + }, tcp, yamux, }; use tokio::sync::{mpsc, oneshot}; @@ -42,7 +45,7 @@ mod identity_store; mod peer_stream; use address::{ - address_with_expected_peer, address_with_peer, is_relayed_address, peer_id_from_address, + address_with_expected_peer, address_with_peer, coordination_relay_peer_id, is_relayed_address, }; use identity_store::load_or_create_key; use peer_stream::spawn_stream; @@ -52,8 +55,10 @@ const APPLICATION_PROTOCOL: &str = "/maka/runtime-host/peer/1"; const IDENTIFY_PROTOCOL: &str = "/maka/runtime-host/peer-identify/1"; const COMMAND_CAPACITY: usize = 32; const INCOMING_STREAM_CAPACITY: usize = 16; -const MAX_PENDING_CONNECTIONS: u32 = 32; -const MAX_ESTABLISHED_CONNECTIONS: u32 = 64; +const MAX_PENDING_INCOMING_CONNECTIONS: u32 = 32; +const MAX_PENDING_OUTGOING_CONNECTIONS: u32 = 1024; +const MAX_ESTABLISHED_INCOMING_CONNECTIONS: u32 = 32; +const MAX_ESTABLISHED_CONNECTIONS: u32 = 1024; const MAX_CONNECTIONS_PER_PEER: u32 = 4; const LISTENER_ADDRESS_QUIET_PERIOD: Duration = Duration::from_millis(250); const COORDINATION_RETRY_INTERVAL: Duration = Duration::from_secs(1); @@ -76,6 +81,7 @@ pub struct StartedEndpoint { } pub struct ConnectOptions { + pub request_id: u32, pub peer_id: PeerId, pub route_hints: Vec, pub coordination_relays: Vec, @@ -87,6 +93,10 @@ pub enum EngineCommand { options: ConnectOptions, result: oneshot::Sender>, }, + CancelConnect { + request_id: u32, + result: oneshot::Sender, + }, Stop { result: oneshot::Sender<()>, }, @@ -118,18 +128,45 @@ struct Behaviour { } struct PendingConnect { + peer_id: PeerId, result: oneshot::Sender>, deadline: Instant, - opening: bool, + opening: Option>, + dials: HashMap, + direct_routes: Vec, coordination_relays: Vec, - next_coordination_attempt: Instant, + coordination_relay_peers: Vec, + next_route_attempt: Instant, +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum DialOrigin { + DirectRoute, + CoordinationRoute, + DirectConnection, +} + +struct StartedConnect { + direct_routes: Vec, + coordination_relay_peers: Vec, +} + +#[derive(Default)] +struct DirectConnectState { + pending: HashMap, + active: HashMap, + retiring_connections: HashSet, + outbound_hole_punch_peers: HashSet, } struct CoordinationRelay { addresses: Vec, + connections: HashSet, + pending_connection: Option, identify_received: bool, identify_sent: bool, reserve: bool, + client_references: usize, reservation_listener: Option, next_connection_attempt: Instant, next_reservation_attempt: Instant, @@ -140,9 +177,12 @@ impl Default for CoordinationRelay { let now = Instant::now(); Self { addresses: Vec::new(), + connections: HashSet::new(), + pending_connection: None, identify_received: false, identify_sent: false, reserve: false, + client_references: 0, reservation_listener: None, next_connection_attempt: now, next_reservation_attempt: now, @@ -151,6 +191,10 @@ impl Default for CoordinationRelay { } impl CoordinationRelay { + fn is_active(&self) -> bool { + self.reserve || self.client_references > 0 + } + fn connection_lost(&mut self, now: Instant) -> Option { self.identify_received = false; self.identify_sent = false; @@ -170,8 +214,8 @@ impl CoordinationRelay { } struct OpenedStream { - peer_id: PeerId, - result: Result, + request_id: u32, + result: Result, } pub async fn ensure_identity(key_path: PathBuf) -> Result { @@ -264,14 +308,12 @@ async fn run_endpoint_async( } let mut coordination_relays = HashMap::new(); for relay in &options.coordination_relays { - register_coordination_relay( - &mut swarm, - &mut coordination_relays, - relay, - local_peer_id, - true, - )?; + coordination_relay_peer_id(relay)?; } + for relay in &options.coordination_relays { + register_coordination_relay(&mut coordination_relays, relay, local_peer_id, true, false)?; + } + maintain_coordination_relays(&mut swarm, &mut coordination_relays, false, Instant::now()); let startup_deadline = Instant::now() + Duration::from_secs(10); let mut address_quiet_deadline = None; @@ -315,7 +357,7 @@ async fn run_endpoint_async( let (opened_tx, mut opened_rx) = mpsc::channel::(COMMAND_CAPACITY); let (close_connection_tx, mut close_connection_rx) = mpsc::channel::(MAX_ESTABLISHED_CONNECTIONS as usize); - let mut pending = HashMap::::new(); + let mut direct = DirectConnectState::default(); let mut relayed = HashMap::>::new(); let mut external_candidate_ready = startup_external_candidate_ready; let mut deadline_tick = tokio::time::interval(Duration::from_millis(100)); @@ -325,36 +367,89 @@ async fn run_endpoint_async( tokio::select! { command = commands.recv() => match command { Some(EngineCommand::Connect { options, result }) => { - if pending.contains_key(&options.peer_id) { + if direct.pending.contains_key(&options.request_id) + || direct.pending.values().any(|connect| connect.peer_id == options.peer_id) + || direct.active.contains_key(&options.peer_id) + { let _ = result.send(Err(PeerError::new( "peer_connect_in_progress", - "a connection to this peer is already in progress", + "a connection request with this identity is already in progress", ))); continue; } - if let Err(error) = start_connect( + let started = match start_connect( &mut swarm, &mut coordination_relays, &options, local_peer_id, ) { - let _ = result.send(Err(error)); - continue; + Ok(peers) => peers, + Err(error) => { + let _ = result.send(Err(error)); + continue; + } + }; + let request_id = options.request_id; + let can_hole_punch = !options.coordination_relays.is_empty() + || options.route_hints.iter().any(is_relayed_address); + if can_hole_punch { + direct.outbound_hole_punch_peers.insert(options.peer_id); } - pending.insert(options.peer_id, PendingConnect { + direct.pending.insert(request_id, PendingConnect { + peer_id: options.peer_id, result, deadline: Instant::now() + options.deadline, - opening: false, + opening: None, + dials: HashMap::new(), + direct_routes: started.direct_routes, coordination_relays: options.coordination_relays, - next_coordination_attempt: Instant::now(), + coordination_relay_peers: started.coordination_relay_peers, + next_route_attempt: Instant::now(), }); + retry_connect_routes( + &mut swarm, + &mut direct, + &coordination_relays, + &stream_control, + &relayed, + external_candidate_ready, + Instant::now(), + ); maybe_open_direct_stream( - options.peer_id, - &mut pending, + request_id, + &mut direct.pending, + &direct.retiring_connections, stream_control.clone(), opened_tx.clone(), ); } + Some(EngineCommand::CancelConnect { request_id, result }) => { + let cancelled = if let Some(mut waiter) = direct.pending.remove(&request_id) { + if let Some(opening) = waiter.opening.take() { + opening.abort(); + } + retire_direct_dials( + &mut swarm, + &mut direct.retiring_connections, + waiter.dials, + None, + ); + release_coordination_relays( + &mut swarm, + &mut coordination_relays, + &waiter.coordination_relay_peers, + &direct.active, + ); + let _ = waiter.result.send(Err(PeerError::new( + "peer_connect_cancelled", + "the peer connection request was cancelled", + ))); + true + } else { + false + }; + let _ = result.send(cancelled); + } Some(EngineCommand::Stop { result }) => { let _ = result.send(()); return Ok(()); @@ -371,14 +466,52 @@ async fn run_endpoint_async( } } Some(connection_id) = close_connection_rx.recv() => { - let _ = swarm.close_connection(connection_id); + direct.active.retain(|_, active| *active != connection_id); + retire_established_connection( + &mut swarm, + &mut direct.retiring_connections, + connection_id, + ); } Some(opened) = opened_rx.recv() => { - if let Some(waiter) = pending.remove(&opened.peer_id) { - let result = opened.result - .map(|stream| spawn_stream(stream, None)) - .map_err(|message| PeerError::new("direct_path_unavailable", message)); + if let Some(waiter) = direct.pending.remove(&opened.request_id) { + let result = match opened.result { + Ok(opened) => { + retire_direct_dials( + &mut swarm, + &mut direct.retiring_connections, + waiter.dials, + Some(opened.connection_id), + ); + direct.active.insert(waiter.peer_id, opened.connection_id); + Ok(spawn_stream( + opened.stream, + Some((opened.connection_id, close_connection_tx.clone())), + )) + } + Err(message) => { + retire_direct_dials( + &mut swarm, + &mut direct.retiring_connections, + waiter.dials, + None, + ); + Err(PeerError::new("direct_path_unavailable", message)) + } + }; + release_coordination_relays( + &mut swarm, + &mut coordination_relays, + &waiter.coordination_relay_peers, + &direct.active, + ); let _ = waiter.result.send(result); + } else if let Ok(opened) = opened.result { + retire_established_connection( + &mut swarm, + &mut direct.retiring_connections, + opened.connection_id, + ); } } event = swarm.select_next_some() => { @@ -387,6 +520,7 @@ async fn run_endpoint_async( event, &mut relayed, &mut coordination_relays, + &mut direct, &mut external_candidate_ready, ); maintain_coordination_relays( @@ -395,11 +529,12 @@ async fn run_endpoint_async( external_candidate_ready, Instant::now(), ); - let peers = pending.keys().copied().collect::>(); - for peer_id in peers { + let requests = direct.pending.keys().copied().collect::>(); + for request_id in requests { maybe_open_direct_stream( - peer_id, - &mut pending, + request_id, + &mut direct.pending, + &direct.retiring_connections, stream_control.clone(), opened_tx.clone(), ); @@ -413,20 +548,35 @@ async fn run_endpoint_async( external_candidate_ready, now, ); - retry_coordination_routes( + retry_connect_routes( &mut swarm, - &mut pending, + &mut direct, &coordination_relays, &stream_control, &relayed, external_candidate_ready, now, ); - let expired = pending.iter() - .filter_map(|(peer_id, item)| (item.deadline <= now).then_some(*peer_id)) + let expired = direct.pending.iter() + .filter_map(|(request_id, item)| (item.deadline <= now).then_some(*request_id)) .collect::>(); - for peer_id in expired { - if let Some(waiter) = pending.remove(&peer_id) { + for request_id in expired { + if let Some(mut waiter) = direct.pending.remove(&request_id) { + if let Some(opening) = waiter.opening.take() { + opening.abort(); + } + retire_direct_dials( + &mut swarm, + &mut direct.retiring_connections, + waiter.dials, + None, + ); + release_coordination_relays( + &mut swarm, + &mut coordination_relays, + &waiter.coordination_relay_peers, + &direct.active, + ); let _ = waiter.result.send(Err(PeerError::new( "direct_path_unavailable", "no direct path was established before the deadline", @@ -465,9 +615,9 @@ fn build_swarm(key: identity::Keypair) -> Result { .with_behaviour(move |key, relay_client| Behaviour { connection_limits: connection_limits::Behaviour::new( connection_limits::ConnectionLimits::default() - .with_max_pending_incoming(Some(MAX_PENDING_CONNECTIONS)) - .with_max_pending_outgoing(Some(MAX_PENDING_CONNECTIONS)) - .with_max_established_incoming(Some(MAX_ESTABLISHED_CONNECTIONS)) + .with_max_pending_incoming(Some(MAX_PENDING_INCOMING_CONNECTIONS)) + .with_max_pending_outgoing(Some(MAX_PENDING_OUTGOING_CONNECTIONS)) + .with_max_established_incoming(Some(MAX_ESTABLISHED_INCOMING_CONNECTIONS)) .with_max_established_outgoing(Some(MAX_ESTABLISHED_CONNECTIONS)) .with_max_established(Some(MAX_ESTABLISHED_CONNECTIONS)) .with_max_established_per_peer(Some(MAX_CONNECTIONS_PER_PEER)), @@ -491,61 +641,78 @@ fn start_connect( coordination_relays: &mut HashMap, options: &ConnectOptions, local_peer_id: PeerId, -) -> Result<(), PeerError> { +) -> Result { if options.route_hints.is_empty() && options.coordination_relays.is_empty() { return Err(PeerError::new( "direct_path_unavailable", "the peer profile has no route hints or coordination relays", )); } - for address in &options.route_hints { - let target = address_with_expected_peer(address, options.peer_id)?; - let _ = swarm.dial(target); - } + let mut relay_peers = Vec::new(); for relay_address in &options.coordination_relays { - let relay_peer = peer_id_from_address(relay_address).ok_or_else(|| { - PeerError::new( - "coordination_unavailable", - "coordination relay address has no peer identity", - ) - })?; + let relay_peer = coordination_relay_peer_id(relay_address)?; if relay_peer == options.peer_id { return Err(PeerError::new( "coordination_unavailable", "coordination relay cannot be the target peer", )); } + if relay_peer == local_peer_id { + return Err(PeerError::new( + "coordination_unavailable", + "peer endpoint cannot use itself as a coordination relay", + )); + } + if !relay_peers.contains(&relay_peer) { + relay_peers.push(relay_peer); + } + } + let direct_targets = options + .route_hints + .iter() + .map(|address| address_with_expected_peer(address, options.peer_id)) + .collect::, _>>()?; + let mut referenced = HashSet::new(); + for relay_address in &options.coordination_relays { + let relay_peer = coordination_relay_peer_id(relay_address) + .expect("coordination relay was validated before registration"); register_coordination_relay( - swarm, coordination_relays, relay_address, local_peer_id, false, + referenced.insert(relay_peer), )?; } - Ok(()) + maintain_coordination_relays(swarm, coordination_relays, false, Instant::now()); + Ok(StartedConnect { + direct_routes: direct_targets, + coordination_relay_peers: relay_peers, + }) } fn maybe_open_direct_stream( - peer_id: PeerId, - pending: &mut HashMap, + request_id: u32, + pending: &mut HashMap, + retiring_connections: &HashSet, mut control: application_stream::Control, opened_tx: mpsc::Sender, ) { - let Some(waiter) = pending.get_mut(&peer_id) else { + let Some(waiter) = pending.get_mut(&request_id) else { return; }; - if waiter.opening || !control.has_connection(peer_id) { + let peer_id = waiter.peer_id; + if waiter.opening.is_some() || !control.has_connection(peer_id, retiring_connections) { return; } - waiter.opening = true; - tokio::spawn(async move { + let retiring_connections = retiring_connections.clone(); + waiter.opening = Some(tokio::spawn(async move { let result = control - .open_stream(peer_id) + .open_stream(peer_id, &retiring_connections) .await .map_err(|error| error.to_string()); - let _ = opened_tx.send(OpenedStream { peer_id, result }).await; - }); + let _ = opened_tx.send(OpenedStream { request_id, result }).await; + })); } fn handle_swarm_event( @@ -553,6 +720,7 @@ fn handle_swarm_event( event: SwarmEvent, relayed: &mut HashMap>, coordination_relays: &mut HashMap, + direct: &mut DirectConnectState, external_candidate_ready: &mut bool, ) { match event { @@ -562,9 +730,33 @@ fn handle_swarm_event( endpoint, .. } => { + if direct.retiring_connections.contains(&connection_id) { + let _ = swarm.close_connection(connection_id); + return; + } + if let Some(relay) = coordination_relays.get_mut(&peer_id) { + if relay.pending_connection == Some(connection_id) { + relay.pending_connection = None; + } + if relay.is_active() { + relay.connections.insert(connection_id); + } else { + let _ = swarm.close_connection(connection_id); + } + } if endpoint.is_relayed() { relayed.entry(peer_id).or_default().insert(connection_id); } else { + if let Some(connect) = direct + .pending + .values_mut() + .find(|connect| connect.peer_id == peer_id) + { + connect + .dials + .entry(connection_id) + .or_insert(DialOrigin::DirectConnection); + } if let Some(ids) = relayed.get(&peer_id) { for id in ids.iter().copied().collect::>() { swarm.close_connection(id); @@ -578,13 +770,73 @@ fn handle_swarm_event( .. } => { remove_connection(relayed, peer_id, connection_id); + direct.retiring_connections.remove(&connection_id); + for connect in direct.pending.values_mut() { + connect.dials.remove(&connection_id); + } + direct.active.retain(|_, active| *active != connection_id); + if let Some(relay) = coordination_relays.get_mut(&peer_id) { + relay.connections.remove(&connection_id); + } if !swarm.is_connected(&peer_id) && let Some(relay) = coordination_relays.get_mut(&peer_id) + && relay.is_active() && let Some(listener) = relay.connection_lost(Instant::now()) { swarm.remove_listener(listener); } } + SwarmEvent::OutgoingConnectionError { connection_id, .. } => { + direct.retiring_connections.remove(&connection_id); + for connect in direct.pending.values_mut() { + connect.dials.remove(&connection_id); + } + for relay in coordination_relays.values_mut() { + if relay.pending_connection == Some(connection_id) { + relay.pending_connection = None; + break; + } + } + } + SwarmEvent::Behaviour(BehaviourEvent::Dcutr(dcutr::Event { + remote_peer_id, + result: Ok(connection_id), + })) => { + if let Some(connect) = direct + .pending + .values_mut() + .find(|connect| connect.peer_id == remote_peer_id) + { + connect + .dials + .entry(connection_id) + .or_insert(DialOrigin::DirectConnection); + } else if direct.active.get(&remote_peer_id) != Some(&connection_id) + && (direct.active.contains_key(&remote_peer_id) + || direct.outbound_hole_punch_peers.contains(&remote_peer_id)) + { + retire_established_connection( + swarm, + &mut direct.retiring_connections, + connection_id, + ); + } + } + SwarmEvent::Behaviour(BehaviourEvent::Dcutr(dcutr::Event { + remote_peer_id, + result: Err(_), + })) => { + if direct + .pending + .values() + .any(|connect| connect.peer_id == remote_peer_id) + && let Some(connection_ids) = relayed.remove(&remote_peer_id) + { + for connection_id in connection_ids { + let _ = swarm.close_connection(connection_id); + } + } + } SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Received { peer_id, .. @@ -648,52 +900,82 @@ fn handle_startup_event( event, &mut HashMap::new(), coordination_relays, + &mut DirectConnectState::default(), external_candidate_ready, ); } fn register_coordination_relay( - swarm: &mut Swarm, relays: &mut HashMap, address: &Multiaddr, local_peer_id: PeerId, reserve: bool, + add_client_reference: bool, ) -> Result<(), PeerError> { - let relay_peer = peer_id_from_address(address).ok_or_else(|| { - PeerError::new( - "coordination_unavailable", - "coordination relay address has no peer identity", - ) - })?; + let relay_peer = coordination_relay_peer_id(address)?; if relay_peer == local_peer_id { return Err(PeerError::new( "coordination_unavailable", "peer endpoint cannot use itself as a coordination relay", )); } - let connected = swarm.is_connected(&relay_peer); let relay = relays.entry(relay_peer).or_default(); relay.reserve |= reserve; + if add_client_reference { + relay.client_references += 1; + } if !relay.addresses.contains(address) { relay.addresses.push(address.clone()); } - if !connected { - dial_coordination_relay(swarm, relay, Instant::now()); - } Ok(()) } +fn release_coordination_relays( + swarm: &mut Swarm, + relays: &mut HashMap, + peers: &[PeerId], + active_outbound: &HashMap, +) { + for peer_id in peers { + let Some(relay) = relays.get_mut(peer_id) else { + continue; + }; + debug_assert!(relay.client_references > 0); + relay.client_references -= 1; + if relay.is_active() { + continue; + } + relay.addresses.clear(); + if let Some(listener) = relay.reservation_listener.take() { + swarm.remove_listener(listener); + } + for connection_id in relay.connections.drain() { + if !active_outbound + .values() + .any(|active| *active == connection_id) + { + let _ = swarm.close_connection(connection_id); + } + } + } +} + fn dial_coordination_relay( swarm: &mut Swarm, + peer_id: PeerId, relay: &mut CoordinationRelay, now: Instant, ) { - if relay.next_connection_attempt > now { + if relay.pending_connection.is_some() || relay.next_connection_attempt > now { return; } relay.next_connection_attempt = now + COORDINATION_RETRY_INTERVAL; - for address in &relay.addresses { - let _ = swarm.dial(address.clone()); + let options = DialOpts::peer_id(peer_id) + .addresses(relay.addresses.clone()) + .build(); + let connection_id = options.connection_id(); + if swarm.dial(options).is_ok() { + relay.pending_connection = Some(connection_id); } } @@ -704,9 +986,12 @@ fn maintain_coordination_relays( now: Instant, ) { for peer_id in relays.keys().copied().collect::>() { + if relays.get(&peer_id).is_none_or(|relay| !relay.is_active()) { + continue; + } if !swarm.is_connected(&peer_id) { if let Some(relay) = relays.get_mut(&peer_id) { - dial_coordination_relay(swarm, relay, now); + dial_coordination_relay(swarm, peer_id, relay, now); } continue; } @@ -742,30 +1027,45 @@ fn request_coordination_reservation( } } -fn retry_coordination_routes( +fn retry_connect_routes( swarm: &mut Swarm, - pending: &mut HashMap, + direct: &mut DirectConnectState, coordination_relays: &HashMap, stream_control: &application_stream::Control, relayed: &HashMap>, external_candidate_ready: bool, now: Instant, ) { - for (peer_id, connect) in pending { - if connect.next_coordination_attempt > now { + for connect in direct.pending.values_mut() { + let peer_id = connect.peer_id; + if connect.next_route_attempt > now { continue; } - if stream_control.has_connection(*peer_id) { + if stream_control.has_connection(peer_id, &direct.retiring_connections) { continue; } - if relayed.get(peer_id).is_some_and(|ids| !ids.is_empty()) { + connect.next_route_attempt = now + COORDINATION_RETRY_INTERVAL; + if !connect + .dials + .values() + .any(|origin| *origin == DialOrigin::DirectRoute) + && let Some(connection_id) = + dial_direct_targets(swarm, peer_id, connect.direct_routes.clone()) + { + connect.dials.insert(connection_id, DialOrigin::DirectRoute); + } + if connect + .dials + .values() + .any(|origin| *origin == DialOrigin::CoordinationRoute) + || relayed.get(&peer_id).is_some_and(|ids| !ids.is_empty()) + { continue; } - connect.next_coordination_attempt = now + COORDINATION_RETRY_INTERVAL; + let mut targets = Vec::new(); for relay in &connect.coordination_relays { - let Some(relay_peer) = peer_id_from_address(relay) else { - continue; - }; + let relay_peer = coordination_relay_peer_id(relay) + .expect("coordination relay was validated before connecting"); if !external_candidate_ready || coordination_relays .get(&relay_peer) @@ -773,15 +1073,62 @@ fn retry_coordination_routes( { continue; } - let target = relay - .clone() - .with(Protocol::P2pCircuit) - .with(Protocol::P2p(*peer_id)); - let _ = swarm.dial(target); + targets.push( + relay + .clone() + .with(Protocol::P2pCircuit) + .with(Protocol::P2p(peer_id)), + ); + } + if let Some(connection_id) = dial_direct_targets(swarm, peer_id, targets) { + connect + .dials + .insert(connection_id, DialOrigin::CoordinationRoute); } } } +fn dial_direct_targets( + swarm: &mut Swarm, + peer_id: PeerId, + addresses: Vec, +) -> Option { + if addresses.is_empty() { + return None; + } + let options = DialOpts::peer_id(peer_id) + .condition(PeerCondition::Always) + .addresses(addresses) + .build(); + let connection_id = options.connection_id(); + swarm.dial(options).is_ok().then_some(connection_id) +} + +fn retire_direct_dials( + swarm: &mut Swarm, + retiring: &mut HashSet, + dials: HashMap, + retained: Option, +) { + for connection_id in dials.into_keys() { + if retained == Some(connection_id) { + continue; + } + retiring.insert(connection_id); + let _ = swarm.close_connection(connection_id); + } +} + +fn retire_established_connection( + swarm: &mut Swarm, + retiring: &mut HashSet, + connection_id: ConnectionId, +) { + if swarm.close_connection(connection_id) { + retiring.insert(connection_id); + } +} + fn remove_connection( connections: &mut HashMap>, peer_id: PeerId, @@ -826,4 +1173,22 @@ mod tests { assert!(!relay.identify_sent); assert_eq!(relay.next_connection_attempt, now); } + + #[test] + fn coordination_relay_requires_one_terminal_peer_identity() { + let relay = PeerId::random(); + let target = PeerId::random(); + let address: Multiaddr = format!("/ip4/127.0.0.1/udp/4001/quic-v1/p2p/{relay}") + .parse() + .expect("valid relay address"); + assert_eq!( + coordination_relay_peer_id(&address).expect("base relay address is accepted"), + relay, + ); + + let tunneled: Multiaddr = format!("{address}/p2p-circuit/p2p/{target}") + .parse() + .expect("valid relayed address"); + assert!(coordination_relay_peer_id(&tunneled).is_err()); + } } diff --git a/native/runtime-host-peer/src/engine/address.rs b/native/runtime-host-peer/src/engine/address.rs index 4d4f1a29be..2feb7f64ab 100644 --- a/native/runtime-host-peer/src/engine/address.rs +++ b/native/runtime-host-peer/src/engine/address.rs @@ -50,6 +50,35 @@ pub(super) fn peer_id_from_address(address: &Multiaddr) -> Option { }) } +pub(super) fn coordination_relay_peer_id(address: &Multiaddr) -> Result { + let mut peer_id = None; + for protocol in address.iter() { + match protocol { + Protocol::P2p(_) if peer_id.is_some() => { + return Err(PeerError::new( + "coordination_unavailable", + "coordination relay address must name exactly one peer", + )); + } + Protocol::P2p(value) => peer_id = Some(value), + Protocol::P2pCircuit => { + return Err(PeerError::new( + "coordination_unavailable", + "coordination relay address must be a base relay address", + )); + } + _ => {} + } + } + match (peer_id, address.iter().last()) { + (Some(peer_id), Some(Protocol::P2p(terminal))) if peer_id == terminal => Ok(peer_id), + _ => Err(PeerError::new( + "coordination_unavailable", + "coordination relay address must end with its peer identity", + )), + } +} + pub(super) fn is_relayed_address(address: &Multiaddr) -> bool { address .iter() diff --git a/native/runtime-host-peer/src/engine/application_stream.rs b/native/runtime-host-peer/src/engine/application_stream.rs index 33df9a4113..7ed44f0ffc 100644 --- a/native/runtime-host-peer/src/engine/application_stream.rs +++ b/native/runtime-host-peer/src/engine/application_stream.rs @@ -18,7 +18,7 @@ */ use std::{ - collections::{HashMap, VecDeque}, + collections::{HashMap, HashSet, VecDeque}, convert::Infallible, future::{Ready, ready}, io, @@ -156,25 +156,38 @@ pub(super) struct Control { } impl Control { - pub(super) fn has_connection(&self, peer_id: PeerId) -> bool { - lock(&self.shared).sender(peer_id).is_some() + pub(super) fn has_connection(&self, peer_id: PeerId, excluded: &HashSet) -> bool { + lock(&self.shared).connection(peer_id, excluded).is_some() } - pub(super) async fn open_stream(&mut self, peer_id: PeerId) -> Result { - let sender = lock(&self.shared) - .sender(peer_id) + pub(super) async fn open_stream( + &mut self, + peer_id: PeerId, + excluded: &HashSet, + ) -> Result { + let (connection_id, sender) = lock(&self.shared) + .connection(peer_id, excluded) .ok_or(OpenStreamError::NoDirectConnection)?; let (result, receiver) = oneshot::channel(); sender .send(NewStream { result }) .await .map_err(|_| OpenStreamError::ConnectionClosed)?; - receiver + let stream = receiver .await - .map_err(|_| OpenStreamError::ConnectionClosed)? + .map_err(|_| OpenStreamError::ConnectionClosed)??; + Ok(OpenedApplicationStream { + connection_id, + stream, + }) } } +pub(super) struct OpenedApplicationStream { + pub(super) connection_id: ConnectionId, + pub(super) stream: Stream, +} + #[derive(Debug)] pub(super) enum OpenStreamError { NoDirectConnection, @@ -221,11 +234,19 @@ impl DirectConnections { self.connections.remove(&connection_id); } - fn sender(&self, peer_id: PeerId) -> Option> { + fn connection( + &self, + peer_id: PeerId, + excluded: &HashSet, + ) -> Option<(ConnectionId, mpsc::Sender)> { self.connections - .values() - .find(|connection| connection.peer_id == peer_id && !connection.sender.is_closed()) - .map(|connection| connection.sender.clone()) + .iter() + .find(|(connection_id, connection)| { + connection.peer_id == peer_id + && !excluded.contains(connection_id) + && !connection.sender.is_closed() + }) + .map(|(connection_id, connection)| (*connection_id, connection.sender.clone())) } } @@ -305,7 +326,11 @@ impl ConnectionHandler for Handler { if std::mem::take(&mut self.close) { return Poll::Ready(libp2p::swarm::ConnectionHandlerEvent::NotifyBehaviour(())); } - if self.pending.is_some() { + if let Some(result) = self.pending.as_mut() { + if result.poll_closed(context).is_ready() { + self.pending = None; + return Poll::Ready(libp2p::swarm::ConnectionHandlerEvent::NotifyBehaviour(())); + } return Poll::Pending; } let Some(commands) = self.commands.as_mut() else { @@ -449,8 +474,12 @@ mod tests { relayed.listen_protocol().upgrade().protocol_info().count(), 0 ); - assert!(lock(&control.shared).sender(peer_id).is_none()); - assert!(!control.has_connection(peer_id)); + assert!( + lock(&control.shared) + .connection(peer_id, &HashSet::new()) + .is_none() + ); + assert!(!control.has_connection(peer_id, &HashSet::new())); let direct = behaviour .handle_established_outbound_connection( @@ -471,7 +500,14 @@ mod tests { .collect::>(), vec![protocol] ); - assert!(lock(&control.shared).sender(peer_id).is_some()); - assert!(control.has_connection(peer_id)); + assert!( + lock(&control.shared) + .connection(peer_id, &HashSet::new()) + .is_some() + ); + assert!(control.has_connection(peer_id, &HashSet::new())); + assert!( + !control.has_connection(peer_id, &HashSet::from([ConnectionId::new_unchecked(2)]),) + ); } } diff --git a/packages/cli/src/runtime-host-cli-context.ts b/packages/cli/src/runtime-host-cli-context.ts index 5bc5ba6739..303abf282e 100644 --- a/packages/cli/src/runtime-host-cli-context.ts +++ b/packages/cli/src/runtime-host-cli-context.ts @@ -26,6 +26,7 @@ import { connectOrSpawnRuntimeHost, connectRemoteRuntimeHostProfile, createClientRuntimeHostProfileCatalog, + createRuntimeHostPeerClientFromEnvironment, createRuntimeHostReconnectingConnection, loadOrCreateRuntimeHostClientInstanceId, LOCAL_RUNTIME_HOST_PROFILE, @@ -36,6 +37,7 @@ import { type RuntimeHostProfile, type ResolvedRuntimeHostProfile, type RuntimeHostProfileCatalog, + type RuntimeHostPeerClient, } from '@maka/runtime-host/client'; import { INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID, @@ -100,6 +102,7 @@ interface RuntimeHostCliContextDeps { readonly loadClientInstanceId: typeof loadOrCreateRuntimeHostClientInstanceId; readonly executionCandidateEntrypoint: URL; readonly readDeploymentRecord: typeof readLocalHostDeploymentRecord; + readonly createPeerClient: typeof createRuntimeHostPeerClientFromEnvironment; readonly profileCatalog?: RuntimeHostProfileCatalog; } @@ -121,6 +124,7 @@ export async function connectRuntimeHostCli( import.meta.resolve('@maka/runtime-host/execution-candidate-main'), ), readDeploymentRecord: readLocalHostDeploymentRecord, + createPeerClient: createRuntimeHostPeerClientFromEnvironment, ...overrides, }; const resolvedProfile = await resolveHostProfile(input, deps); @@ -131,6 +135,10 @@ export async function connectRuntimeHostCli( : await deps.loadClientInstanceId( join(input.clientDataRoot ?? resolveMakaClientDataRoot(), 'runtime-host-client.json'), ); + const peerClient: RuntimeHostPeerClient | undefined = + profile.kind === 'remote' && profile.transport.kind === 'libp2p-direct' + ? deps.createPeerClient() + : undefined; const connectInput = { rootPath: input.rootPath, protocol: { min: RUNTIME_HOST_PROTOCOL_VERSION, max: RUNTIME_HOST_PROTOCOL_VERSION }, @@ -148,6 +156,7 @@ export async function connectRuntimeHostCli( credential: resolvedProfile.credential!, clientInstanceId, sshInteraction, + ...(peerClient ? { peerClient } : {}), ...(signal ? { signal } : {}), }); } @@ -177,23 +186,32 @@ export async function connectRuntimeHostCli( } return connected.connection; }; - const initialConnection = await connect( - undefined, - input.interactiveSsh && process.stdin.isTTY && process.stdout.isTTY ? 'inherit' : 'batch', - ); - const connection = await createRuntimeHostReconnectingConnection({ - initialConnection, - connect: (signal) => connect(signal, 'batch'), - }); + let connection: Awaited> | undefined; try { + const initialConnection = await connect( + undefined, + input.interactiveSsh && process.stdin.isTTY && process.stdout.isTTY ? 'inherit' : 'batch', + ); + connection = await createRuntimeHostReconnectingConnection({ + initialConnection, + connect: (signal) => connect(signal, 'batch'), + }); + const liveConnection = connection; return { - connection, - catalog: await deps.readConnectionCatalog(connection), + connection: liveConnection, + catalog: await deps.readConnectionCatalog(liveConnection), profile, - close: () => connection.close(), + close: async () => { + try { + await liveConnection.close(); + } finally { + await peerClient?.close(); + } + }, }; } catch (error) { - await connection.close().catch(() => undefined); + await connection?.close().catch(() => undefined); + await peerClient?.close().catch(() => undefined); throw error; } } diff --git a/packages/runtime-host/src/__tests__/peer-listener.test.ts b/packages/runtime-host/src/__tests__/peer-listener.test.ts index 61aeac4b2f..72db115299 100644 --- a/packages/runtime-host/src/__tests__/peer-listener.test.ts +++ b/packages/runtime-host/src/__tests__/peer-listener.test.ts @@ -112,6 +112,7 @@ function endpointWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerNa connect: async () => { throw new Error('not used'); }, + cancelConnect: async () => false, accept: async () => streams.shift() ?? null, close: async () => undefined, }; diff --git a/packages/runtime-host/src/__tests__/peer-native.test.ts b/packages/runtime-host/src/__tests__/peer-native.test.ts index 5946f7ed40..423567335a 100644 --- a/packages/runtime-host/src/__tests__/peer-native.test.ts +++ b/packages/runtime-host/src/__tests__/peer-native.test.ts @@ -23,7 +23,7 @@ import { tmpdir } from 'node:os'; import { join, relative } from 'node:path'; import { setImmediate as waitForImmediate } from 'node:timers/promises'; import { test } from 'node:test'; -import { connectPeerRuntimeHost } from '../client/host-profile.js'; +import { createRuntimeHostPeerClient } from '../client/peer-client.js'; import { ensureRuntimeHostPeerIdentity, readRuntimeHostPeerAuthentication, @@ -33,51 +33,77 @@ import { type RuntimeHostPeerNativeStream, } from '../transport/peer-native.js'; -test('closes an in-flight peer endpoint when connection is cancelled', { - timeout: 2_000, -}, async () => { +test('shares one peer endpoint while cancelling connection attempts independently', async () => { const directory = await mkdtemp(join(tmpdir(), 'maka-peer-abort-')); const nativePath = join(directory, 'peer.cjs'); - const previousNativePath = process.env.MAKA_RUNTIME_HOST_PEER_NATIVE_PATH; - const previousKeyPath = process.env.MAKA_RUNTIME_HOST_PEER_KEY_PATH; try { await writeFile( nativePath, - `let rejectConnect; + `let finishAccept; +const pending = new Map(); +const stats = { starts: 0, closes: 0, requests: [], cancellations: [] }; +let missFirstCancellation = true; +const stream = { read: async () => null, write: async () => {}, close: async () => {}, abort: () => {} }; module.exports = { + stats, + failEndpoint: () => finishAccept?.(null), ensurePeerIdentity: async () => 'client', - startPeerEndpoint: () => ({ - peerId: 'client', - listenAddresses: [], - connect: () => new Promise((_resolve, reject) => { rejectConnect = reject; }), - close: async () => rejectConnect?.(new Error('closed by abort')), - }), + startPeerEndpoint: () => { + stats.starts += 1; + return { + peerId: 'client', + listenAddresses: [], + connect: ({ requestId, peerId }) => { + stats.requests.push(requestId); + if (peerId === 'ready') return Promise.resolve(stream); + return new Promise((_resolve, reject) => pending.set(requestId, reject)); + }, + cancelConnect: async (requestId) => { + stats.cancellations.push(requestId); + if (missFirstCancellation) { + missFirstCancellation = false; + return false; + } + pending.get(requestId)?.(new Error('peer_connect_cancelled: cancelled')); + pending.delete(requestId); + return true; + }, + accept: () => new Promise((resolve) => { finishAccept = resolve; }), + close: async () => { stats.closes += 1; finishAccept?.(null); }, + }; + }, }; `, ); - process.env.MAKA_RUNTIME_HOST_PEER_NATIVE_PATH = nativePath; - process.env.MAKA_RUNTIME_HOST_PEER_KEY_PATH = join(directory, 'peer.key'); + const client = createRuntimeHostPeerClient({ + nativePath, + keyPath: join(directory, 'peer.key'), + }); const abort = new AbortController(); - const connection = connectPeerRuntimeHost({ - profileId: 'peer-test', - transport: { - kind: 'libp2p-direct', - peerId: 'target', - routeHints: ['/memory/1'], - coordinationRelays: [], - }, - credential: 'credential', - expectedRootId: '00000000-0000-4000-8000-000000000001', - clientInstanceId: 'client-test', - signal: abort.signal, - connectTimeoutMs: 120_000, + const pending = client.connect(peerConnectInput('pending'), abort.signal); + abort.abort(); + await assert.rejects(pending, /aborted/u); + + await client.connect(peerConnectInput('ready')); + const native = await import(nativePath); + assert.deepEqual(native.default.stats, { + starts: 1, + closes: 0, + requests: [1, 2], + cancellations: [1, 1], }); + + native.default.failEndpoint(); await waitForImmediate(); - abort.abort(); - await assert.rejects(connection, /closed by abort/u); + await assert.rejects( + client.connect(peerConnectInput('ready')), + /cannot recover until this Client restarts/u, + ); + assert.equal(native.default.stats.starts, 1); + + await client.close(); + assert.equal(native.default.stats.closes, 1); } finally { - restoreEnvironment('MAKA_RUNTIME_HOST_PEER_NATIVE_PATH', previousNativePath); - restoreEnvironment('MAKA_RUNTIME_HOST_PEER_KEY_PATH', previousKeyPath); await rm(directory, { recursive: true, force: true }); } }); @@ -141,7 +167,11 @@ function streamWith(chunk: Buffer): RuntimeHostPeerNativeStream { }; } -function restoreEnvironment(name: string, value: string | undefined): void { - if (value === undefined) delete process.env[name]; - else process.env[name] = value; +function peerConnectInput(peerId: string) { + return { + peerId, + routeHints: ['/memory/1'], + coordinationRelays: [], + directDeadlineMs: 1_000, + } as const; } diff --git a/packages/runtime-host/src/client/host-profile.ts b/packages/runtime-host/src/client/host-profile.ts index 6068b84a7f..afcfffacfa 100644 --- a/packages/runtime-host/src/client/host-profile.ts +++ b/packages/runtime-host/src/client/host-profile.ts @@ -40,9 +40,9 @@ import { RuntimeHostPeerByteStream, RuntimeHostPeerError, readRuntimeHostPeerAuthenticationResult, - startRuntimeHostPeerEndpoint, writeRuntimeHostPeerAuthentication, } from '../transport/peer-native.js'; +import type { RuntimeHostPeerClient } from './peer-client.js'; import { RuntimeHostPermanentReconnectError } from './reconnect-lifecycle.js'; import { RuntimeHostRemoteCompatibilityError } from './remote-compatibility-error.js'; import { @@ -232,6 +232,7 @@ export async function connectRemoteRuntimeHostProfile( readonly handshakeTimeoutMs?: number; readonly readyTimeoutMs?: number; readonly sshInteraction?: RuntimeHostSshInteraction; + readonly peerClient?: RuntimeHostPeerClient; }, overrides: { connect?: typeof connectRemoteRuntimeHost; @@ -251,6 +252,7 @@ export async function connectRemoteRuntimeHostProfile( credential: input.credential, expectedRootId: input.profile.rootId, clientInstanceId: input.clientInstanceId, + peerClient: requireRuntimeHostPeerClient(input.peerClient), ...(input.signal === undefined ? {} : { signal: input.signal }), ...(input.connectTimeoutMs === undefined ? {} : { connectTimeoutMs: input.connectTimeoutMs }), ...(input.handshakeTimeoutMs === undefined @@ -355,42 +357,26 @@ export async function connectPeerRuntimeHost(input: { readonly credential: string; readonly expectedRootId: string; readonly clientInstanceId: string; + readonly peerClient: RuntimeHostPeerClient; readonly signal?: AbortSignal; readonly connectTimeoutMs?: number; readonly handshakeTimeoutMs?: number; }): Promise { input.signal?.throwIfAborted(); - const nativePath = process.env.MAKA_RUNTIME_HOST_PEER_NATIVE_PATH; - const keyPath = process.env.MAKA_RUNTIME_HOST_PEER_KEY_PATH; - if (!nativePath || !keyPath) { - throw new RuntimeHostPeerError( - 'peer_native_unavailable', - 'Experimental direct peer requires MAKA_RUNTIME_HOST_PEER_NATIVE_PATH and MAKA_RUNTIME_HOST_PEER_KEY_PATH', - ); - } - const endpoint = startRuntimeHostPeerEndpoint({ nativePath, keyPath }); - let resolveResourceClosed!: () => void; - let closeTask: Promise | undefined; - const resource = { - closed: new Promise((resolve) => { - resolveResourceClosed = resolve; - }), - close: () => { - closeTask ??= endpoint.close().finally(resolveResourceClosed); - return closeTask; + const stream = await input.peerClient.connect( + { + peerId: input.transport.peerId, + routeHints: input.transport.routeHints, + coordinationRelays: input.transport.coordinationRelays, + directDeadlineMs: Math.min(input.connectTimeoutMs ?? 40_000, 120_000), }, - }; - const abort = () => void resource.close().catch(() => undefined); + input.signal, + ); + const abort = () => stream.abort(); input.signal?.addEventListener('abort', abort, { once: true }); if (input.signal?.aborted) abort(); let transferred = false; try { - const stream = await endpoint.connect({ - peerId: input.transport.peerId, - routeHints: input.transport.routeHints, - coordinationRelays: input.transport.coordinationRelays, - directDeadlineMs: Math.min(input.connectTimeoutMs ?? 40_000, 120_000), - }); input.signal?.throwIfAborted(); await writeRuntimeHostPeerAuthentication(stream, input.credential); const authentication = await readRuntimeHostPeerAuthenticationResult( @@ -416,8 +402,8 @@ export async function connectPeerRuntimeHost(input: { ...(input.handshakeTimeoutMs === undefined ? {} : { handshakeTimeoutMs: input.handshakeTimeoutMs }), - connectionResource: resource, }); + input.signal?.throwIfAborted(); if (result.kind === 'incompatible') { throw new RuntimeHostRemoteCompatibilityError(input.profileId, result.handshake); } @@ -429,10 +415,20 @@ export async function connectPeerRuntimeHost(input: { return result.connection; } finally { input.signal?.removeEventListener('abort', abort); - if (!transferred) await resource.close().catch(() => undefined); + if (!transferred) stream.abort(); } } +function requireRuntimeHostPeerClient( + peerClient: RuntimeHostPeerClient | undefined, +): RuntimeHostPeerClient { + if (peerClient) return peerClient; + throw new RuntimeHostPeerError( + 'peer_native_unavailable', + 'Experimental direct peer requires a Client peer endpoint owner', + ); +} + export function remoteRuntimeHostUnavailableError( subject: string, reason: Extract['reason'], diff --git a/packages/runtime-host/src/client/index.ts b/packages/runtime-host/src/client/index.ts index 167e39df6b..fca93580da 100644 --- a/packages/runtime-host/src/client/index.ts +++ b/packages/runtime-host/src/client/index.ts @@ -137,6 +137,12 @@ export { type RuntimeHostOwnerConnectionCode, } from './owner-connection-code.js'; export { ensureRuntimeHostPeerIdentity } from '../transport/peer-native.js'; +export { + createRuntimeHostPeerClient, + createRuntimeHostPeerClientFromEnvironment, + type RuntimeHostPeerClient, + type RuntimeHostPeerConnectInput, +} from './peer-client.js'; export { createOAuthPresentationClientProvider, type OAuthPresentationBackend, diff --git a/packages/runtime-host/src/client/peer-client.ts b/packages/runtime-host/src/client/peer-client.ts new file mode 100644 index 0000000000..df6707e027 --- /dev/null +++ b/packages/runtime-host/src/client/peer-client.ts @@ -0,0 +1,190 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { + RuntimeHostPeerError, + startRuntimeHostPeerEndpoint, + type RuntimeHostPeerNativeEndpoint, + type RuntimeHostPeerNativeStream, +} from '../transport/peer-native.js'; +import { RuntimeHostPermanentReconnectError } from './reconnect-lifecycle.js'; + +export interface RuntimeHostPeerConnectInput { + readonly peerId: string; + readonly routeHints: readonly string[]; + readonly coordinationRelays?: readonly string[]; + readonly directDeadlineMs: number; +} + +export interface RuntimeHostPeerClient { + connect( + input: RuntimeHostPeerConnectInput, + signal?: AbortSignal, + ): Promise; + close(): Promise; +} + +export function createRuntimeHostPeerClientFromEnvironment( + environment: NodeJS.ProcessEnv = process.env, +): RuntimeHostPeerClient { + const nativePath = environment.MAKA_RUNTIME_HOST_PEER_NATIVE_PATH; + const keyPath = environment.MAKA_RUNTIME_HOST_PEER_KEY_PATH; + if (!nativePath || !keyPath) { + throw new RuntimeHostPeerError( + 'peer_native_unavailable', + 'Experimental direct peer requires MAKA_RUNTIME_HOST_PEER_NATIVE_PATH and MAKA_RUNTIME_HOST_PEER_KEY_PATH', + ); + } + return createRuntimeHostPeerClient({ nativePath, keyPath }); +} + +export function createRuntimeHostPeerClient(input: { + readonly nativePath: string; + readonly keyPath: string; +}): RuntimeHostPeerClient { + return new RuntimeHostPeerClientImpl(input); +} + +class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { + readonly #nativePath: string; + readonly #keyPath: string; + #endpoint: RuntimeHostPeerNativeEndpoint | undefined; + #draining: Promise | undefined; + #terminalError: Error | undefined; + #nextRequestId = 1; + #closed = false; + #closeTask: Promise | undefined; + + constructor(input: { readonly nativePath: string; readonly keyPath: string }) { + this.#nativePath = input.nativePath; + this.#keyPath = input.keyPath; + } + + async connect( + input: RuntimeHostPeerConnectInput, + signal?: AbortSignal, + ): Promise { + signal?.throwIfAborted(); + const endpoint = this.#requireEndpoint(); + const requestId = this.#allocateRequestId(); + const connection = endpoint.connect({ ...input, requestId }); + let settled = false; + const cancel = () => { + void cancelPeerConnect(endpoint, requestId, () => settled); + }; + signal?.addEventListener('abort', cancel, { once: true }); + if (signal?.aborted) cancel(); + try { + const stream = await connection; + if (signal?.aborted) { + stream.abort(); + signal.throwIfAborted(); + } + return stream; + } catch (error) { + signal?.throwIfAborted(); + throw error; + } finally { + settled = true; + signal?.removeEventListener('abort', cancel); + } + } + + close(): Promise { + this.#closeTask ??= this.#close(); + return this.#closeTask; + } + + #requireEndpoint(): RuntimeHostPeerNativeEndpoint { + if (this.#closed) { + throw new RuntimeHostPeerError('peer_native_failed', 'Runtime Host peer client is closed'); + } + if (this.#terminalError) { + throw new RuntimeHostPermanentReconnectError( + 'Runtime Host peer networking stopped and cannot recover until this Client restarts', + { cause: this.#terminalError }, + ); + } + if (this.#endpoint) return this.#endpoint; + const endpoint = startRuntimeHostPeerEndpoint({ + nativePath: this.#nativePath, + keyPath: this.#keyPath, + }); + this.#endpoint = endpoint; + this.#draining = this.#drainInbound(endpoint); + return endpoint; + } + + async #drainInbound(endpoint: RuntimeHostPeerNativeEndpoint): Promise { + try { + while (true) { + const stream = await endpoint.accept(); + if (!stream) { + if (!this.#closed) { + this.#terminalError = new Error('Runtime Host peer networking stopped unexpectedly'); + } + return; + } + stream.abort(); + } + } catch (error) { + // Connection attempts and streams expose a terminal native failure to + // their existing reconnect owners. This owner never replaces its Swarm. + this.#terminalError = error instanceof Error ? error : new Error(String(error)); + } + } + + async #close(): Promise { + this.#closed = true; + const endpoint = this.#endpoint; + this.#endpoint = undefined; + if (!endpoint) return; + let closeError: unknown; + let closeFailed = false; + try { + await endpoint.close(); + } catch (error) { + closeFailed = true; + closeError = error; + } + await this.#draining; + if (closeFailed) throw closeError; + } + + #allocateRequestId(): number { + const requestId = this.#nextRequestId; + this.#nextRequestId = requestId === 0xffff_ffff ? 1 : requestId + 1; + return requestId; + } +} + +async function cancelPeerConnect( + endpoint: RuntimeHostPeerNativeEndpoint, + requestId: number, + isSettled: () => boolean, +): Promise { + try { + while (!isSettled() && !(await endpoint.cancelConnect(requestId))) { + // N-API schedules connect and cancel independently. Retry until the + // engine has observed the request or the connect promise settles. + } + } catch { + // The endpoint closing also settles the connect promise. + } +} diff --git a/packages/runtime-host/src/transport/peer-native.ts b/packages/runtime-host/src/transport/peer-native.ts index 69c46a56f7..a517b41ee3 100644 --- a/packages/runtime-host/src/transport/peer-native.ts +++ b/packages/runtime-host/src/transport/peer-native.ts @@ -56,11 +56,13 @@ export interface RuntimeHostPeerNativeEndpoint { readonly peerId: string; readonly listenAddresses: readonly string[]; connect(options: { + readonly requestId: number; readonly peerId: string; readonly routeHints: readonly string[]; readonly coordinationRelays?: readonly string[]; readonly directDeadlineMs: number; }): Promise; + cancelConnect(requestId: number): Promise; accept(): Promise; close(): Promise; } diff --git a/scripts/smoke-release-cli-package.mjs b/scripts/smoke-release-cli-package.mjs index 2ac6661e9b..c832a5a395 100644 --- a/scripts/smoke-release-cli-package.mjs +++ b/scripts/smoke-release-cli-package.mjs @@ -224,6 +224,7 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } const previousKeyPath = process.env.MAKA_RUNTIME_HOST_PEER_KEY_PATH; let host; let connection; + let peerClient; try { delete process.env.MAKA_RUNTIME_HOST_PEER_NATIVE_PATH; delete process.env.MAKA_RUNTIME_HOST_PEER_KEY_PATH; @@ -267,6 +268,7 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } canUseHostPaths: false, preset: 'terminal-client', }); + peerClient = client.createRuntimeHostPeerClientFromEnvironment(process.env); connection = await client.connectRemoteRuntimeHostProfile({ profile: { id: 'release-smoke-peer', @@ -282,6 +284,7 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } }, credential: issued.credential, clientInstanceId: 'release-smoke-peer-client', + peerClient, connectTimeoutMs: 10_000, handshakeTimeoutMs: 10_000, readyTimeoutMs: 10_000, @@ -292,6 +295,7 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } } } finally { await connection?.close().catch(() => undefined); + await peerClient?.close().catch(() => undefined); await host?.close().catch(() => undefined); restoreEnvironment('MAKA_RUNTIME_HOST_PEER_NATIVE_PATH', previousNativePath); restoreEnvironment('MAKA_RUNTIME_HOST_PEER_KEY_PATH', previousKeyPath);