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);