diff --git a/.github/workflows/cli-package-validation.yml b/.github/workflows/cli-package-validation.yml index 54130d0938..d7c08009fc 100644 --- a/.github/workflows/cli-package-validation.yml +++ b/.github/workflows/cli-package-validation.yml @@ -30,8 +30,14 @@ on: - 'packages/cli/src/runtime-host-cli.ts' - 'packages/cli/src/runtime-host-peer-*' - 'packages/cli/src/runtime-host-service-*' + - 'packages/runtime-host/package.json' + - 'packages/runtime-host/src/client/peer-client.ts' + - 'packages/runtime-host/src/peer-mesh/**' - 'packages/runtime-host/src/server/peer-listener.ts' - 'packages/runtime-host/src/transport/peer-native.ts' + - 'packages/storage/package.json' + - 'packages/storage/src/file-lifetime-owner.ts' + - 'packages/storage/src/native-file-lock.ts' - 'scripts/generate-runtime-host-peer-*' - 'scripts/release-cli-package.mjs' - 'scripts/smoke-release-cli-package.mjs' diff --git a/native/runtime-host-peer/src/bindings.rs b/native/runtime-host-peer/src/bindings.rs index 2d76035670..a67ab4aaba 100644 --- a/native/runtime-host-peer/src/bindings.rs +++ b/native/runtime-host-peer/src/bindings.rs @@ -26,7 +26,7 @@ use std::{ use libp2p::{Multiaddr, PeerId}; use napi::bindgen_prelude::{Buffer, Error, Result, Status}; use napi_derive::napi; -use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot}; +use tokio::sync::{Mutex as AsyncMutex, mpsc, oneshot, watch}; use crate::engine::{self, EngineCommand, PeerError, StreamCommand}; @@ -55,6 +55,7 @@ pub struct PeerEndpoint { listen_addresses: Vec, commands: mpsc::Sender, incoming: Arc>>, + mesh_incoming: Arc>>, terminal: Arc>>, thread: Arc>>>, } @@ -73,43 +74,23 @@ impl PeerEndpoint { #[napi] pub async fn connect(&self, options: ConnectPeerOptions) -> Result { - let peer_id = parse_peer_id(&options.peer_id)?; - let route_hints = parse_addresses(options.route_hints, "route hint")?; - let coordination_relays = parse_addresses( - options.coordination_relays.unwrap_or_default(), - "coordination relay", - )?; - if !(1..=120_000).contains(&options.direct_deadline_ms) { - return Err(Error::new( - Status::InvalidArg, - "direct deadline must be between 1 and 120000 milliseconds", - )); - } - let (result_tx, result_rx) = oneshot::channel(); - self.commands - .send(EngineCommand::Connect { - options: engine::ConnectOptions { - request_id: options.request_id, - peer_id, - route_hints, - coordination_relays, - deadline: Duration::from_millis(u64::from(options.direct_deadline_ms)), - }, - result: result_tx, - }) + connect_peer(self, options, engine::StreamKind::Application).await + } + + #[napi] + pub async fn connect_mesh_control(&self, options: ConnectPeerOptions) -> Result { + connect_peer(self, options, engine::StreamKind::MeshControl).await + } + + #[napi] + pub async fn accept_mesh_control(&self) -> Result> { + self.mesh_incoming + .lock() .await - .map_err(|_| { - peer_error(PeerError { - code: "peer_native_failed", - message: "peer endpoint is closed".to_owned(), - }) - })?; - wrap_stream( - result_rx - .await - .map_err(|_| native_closed_error())? - .map_err(peer_error)?, - ) + .recv() + .await + .map(wrap_stream) + .transpose() } #[napi] @@ -164,6 +145,52 @@ impl PeerEndpoint { } } +async fn connect_peer( + endpoint: &PeerEndpoint, + options: ConnectPeerOptions, + stream_kind: engine::StreamKind, +) -> Result { + let peer_id = parse_peer_id(&options.peer_id)?; + let route_hints = parse_addresses(options.route_hints, "route hint")?; + let coordination_relays = parse_addresses( + options.coordination_relays.unwrap_or_default(), + "coordination relay", + )?; + if !(1..=120_000).contains(&options.direct_deadline_ms) { + return Err(Error::new( + Status::InvalidArg, + "direct deadline must be between 1 and 120000 milliseconds", + )); + } + let (result_tx, result_rx) = oneshot::channel(); + endpoint + .commands + .send(EngineCommand::Connect { + options: engine::ConnectOptions { + request_id: options.request_id, + peer_id, + route_hints, + coordination_relays, + deadline: Duration::from_millis(u64::from(options.direct_deadline_ms)), + }, + stream_kind, + result: result_tx, + }) + .await + .map_err(|_| { + peer_error(PeerError { + code: "peer_native_failed", + message: "peer endpoint is closed".to_owned(), + }) + })?; + wrap_stream( + result_rx + .await + .map_err(|_| native_closed_error())? + .map_err(peer_error)?, + ) +} + impl Drop for PeerEndpoint { fn drop(&mut self) { if Arc::strong_count(&self.thread) == 1 { @@ -175,12 +202,19 @@ impl Drop for PeerEndpoint { #[napi] pub struct PeerStream { + peer_id: String, incoming: Arc>, commands: mpsc::Sender, + abort: watch::Sender, } #[napi] impl PeerStream { + #[napi(getter)] + pub fn peer_id(&self) -> String { + self.peer_id.clone() + } + #[napi] pub async fn read(&self) -> Result> { match self.incoming.lock().await.recv().await { @@ -225,7 +259,7 @@ impl PeerStream { #[napi] pub fn abort(&self) { - let _ = self.commands.try_send(StreamCommand::Abort); + self.abort.send_replace(true); } } @@ -253,6 +287,7 @@ pub fn start_peer_endpoint(options: StartPeerEndpointOptions) -> Result Result { fn wrap_stream(stream: engine::PeerStream) -> Result { Ok(PeerStream { + peer_id: stream.peer_id.to_string(), incoming: Arc::new(AsyncMutex::new(stream.incoming)), commands: stream.commands, + abort: stream.abort, }) } diff --git a/native/runtime-host-peer/src/engine.rs b/native/runtime-host-peer/src/engine.rs index b39ac5178d..81a7dcbbaa 100644 --- a/native/runtime-host-peer/src/engine.rs +++ b/native/runtime-host-peer/src/engine.rs @@ -52,9 +52,11 @@ use peer_stream::spawn_stream; pub use peer_stream::{PeerStream, StreamCommand}; const APPLICATION_PROTOCOL: &str = "/maka/runtime-host/peer/1"; +const MESH_CONTROL_PROTOCOL: &str = "/maka/runtime-host/mesh-control/1"; const IDENTIFY_PROTOCOL: &str = "/maka/runtime-host/peer-identify/1"; const COMMAND_CAPACITY: usize = 32; const INCOMING_STREAM_CAPACITY: usize = 16; +const MESH_INCOMING_STREAM_CAPACITY: usize = 32; const MAX_PENDING_INCOMING_CONNECTIONS: u32 = 32; const MAX_PENDING_OUTGOING_CONNECTIONS: u32 = 1024; const MAX_ESTABLISHED_INCOMING_CONNECTIONS: u32 = 32; @@ -62,6 +64,7 @@ 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); +const IDLE_CONNECTION_TIMEOUT: Duration = Duration::from_secs(10); #[derive(Clone)] pub struct StartOptions { @@ -76,6 +79,7 @@ pub struct StartedEndpoint { pub listen_addresses: Vec, pub commands: mpsc::Sender, pub incoming: mpsc::Receiver, + pub mesh_incoming: mpsc::Receiver, pub terminal: mpsc::Receiver, pub thread: thread::JoinHandle<()>, } @@ -91,6 +95,7 @@ pub struct ConnectOptions { pub enum EngineCommand { Connect { options: ConnectOptions, + stream_kind: StreamKind, result: oneshot::Sender>, }, CancelConnect { @@ -125,11 +130,13 @@ struct Behaviour { identify: identify::Behaviour, ping: ping::Behaviour, application_stream: application_stream::Behaviour, + mesh_control: application_stream::Behaviour, } struct PendingConnect { peer_id: PeerId, result: oneshot::Sender>, + stream_kind: StreamKind, deadline: Instant, opening: Option>, dials: HashMap, @@ -137,13 +144,19 @@ struct PendingConnect { coordination_relays: Vec, coordination_relay_peers: Vec, next_route_attempt: Instant, + retry_coordination: bool, } #[derive(Clone, Copy, PartialEq, Eq)] -enum DialOrigin { +pub enum StreamKind { + Application, + MeshControl, +} + +#[derive(Clone, Copy, PartialEq, Eq)] +pub(super) enum DialOrigin { DirectRoute, CoordinationRoute, - DirectConnection, } struct StartedConnect { @@ -156,7 +169,6 @@ struct DirectConnectState { pending: HashMap, active: HashMap, retiring_connections: HashSet, - outbound_hole_punch_peers: HashSet, } struct CoordinationRelay { @@ -215,7 +227,19 @@ impl CoordinationRelay { struct OpenedStream { request_id: u32, - result: Result, + result: Result, +} + +pub(super) enum StreamCompletion { + Application(ConnectionId), + MeshControl { + coordination_relay_peers: Vec, + }, +} + +pub(super) struct CompletedStream { + kind: StreamCompletion, + acknowledged: oneshot::Sender<()>, } pub async fn ensure_identity(key_path: PathBuf) -> Result { @@ -229,11 +253,18 @@ pub fn start(options: StartOptions) -> Result { let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel(1); let (command_tx, command_rx) = mpsc::channel(COMMAND_CAPACITY); let (incoming_tx, incoming_rx) = mpsc::channel(INCOMING_STREAM_CAPACITY); + let (mesh_incoming_tx, mesh_incoming_rx) = mpsc::channel(MESH_INCOMING_STREAM_CAPACITY); let (terminal_tx, terminal_rx) = mpsc::channel(1); let thread = thread::Builder::new() .name("maka-runtime-host-peer".to_owned()) .spawn(move || { - let result = run_endpoint(options, command_rx, incoming_tx, ready_tx.clone()); + let result = run_endpoint( + options, + command_rx, + incoming_tx, + mesh_incoming_tx, + ready_tx.clone(), + ); if let Err(error) = result { let _ = ready_tx.send(Err(error.clone())); let _ = terminal_tx.blocking_send(error); @@ -248,6 +279,7 @@ pub fn start(options: StartOptions) -> Result { listen_addresses: ready.1, commands: command_tx, incoming: incoming_rx, + mesh_incoming: mesh_incoming_rx, terminal: terminal_rx, thread, }) @@ -257,6 +289,7 @@ fn run_endpoint( options: StartOptions, commands: mpsc::Receiver, incoming_tx: mpsc::Sender, + mesh_incoming_tx: mpsc::Sender, ready_tx: std::sync::mpsc::SyncSender), PeerError>>, ) -> Result<(), PeerError> { let runtime = tokio::runtime::Builder::new_multi_thread() @@ -264,13 +297,20 @@ fn run_endpoint( .thread_name("maka-peer-io") .build() .map_err(|error| PeerError::new("peer_native_failed", error.to_string()))?; - runtime.block_on(run_endpoint_async(options, commands, incoming_tx, ready_tx)) + runtime.block_on(run_endpoint_async( + options, + commands, + incoming_tx, + mesh_incoming_tx, + ready_tx, + )) } async fn run_endpoint_async( options: StartOptions, mut commands: mpsc::Receiver, incoming_tx: mpsc::Sender, + mesh_incoming_tx: mpsc::Sender, ready_tx: std::sync::mpsc::SyncSender), PeerError>>, ) -> Result<(), PeerError> { let key = match options.expected_peer_id { @@ -287,7 +327,8 @@ async fn run_endpoint_async( None => load_or_create_key(&options.key_path).await?, }; let local_peer_id = PeerId::from(key.public()); - let (mut swarm, stream_control, mut incoming_streams) = build_swarm(key)?; + let (mut swarm, stream_control, mut incoming_streams, mesh_control, mut mesh_incoming) = + build_swarm(key)?; let listen_addresses = if options.listen_addresses.is_empty() { vec![ @@ -355,8 +396,8 @@ async fn run_endpoint_async( let _ = ready_tx.send(Ok((local_peer_id, bound_addresses))); 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 (stream_completed_tx, mut stream_completed_rx) = + mpsc::channel::(MAX_ESTABLISHED_CONNECTIONS as usize); let mut direct = DirectConnectState::default(); let mut relayed = HashMap::>::new(); let mut external_candidate_ready = startup_external_candidate_ready; @@ -366,10 +407,11 @@ async fn run_endpoint_async( loop { tokio::select! { command = commands.recv() => match command { - Some(EngineCommand::Connect { options, result }) => { + Some(EngineCommand::Connect { options, stream_kind, result }) => { 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) + || (stream_kind == StreamKind::Application + && direct.active.contains_key(&options.peer_id)) { let _ = result.send(Err(PeerError::new( "peer_connect_in_progress", @@ -382,6 +424,7 @@ async fn run_endpoint_async( &mut coordination_relays, &options, local_peer_id, + stream_kind, ) { Ok(peers) => peers, Err(error) => { @@ -390,14 +433,14 @@ async fn run_endpoint_async( } }; 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); - } + let retry_coordination = stream_kind == StreamKind::Application + && relayed + .get(&options.peer_id) + .is_some_and(|connections| !connections.is_empty()); direct.pending.insert(request_id, PendingConnect { peer_id: options.peer_id, result, + stream_kind, deadline: Instant::now() + options.deadline, opening: None, dials: HashMap::new(), @@ -405,6 +448,7 @@ async fn run_endpoint_async( coordination_relays: options.coordination_relays, coordination_relay_peers: started.coordination_relay_peers, next_route_attempt: Instant::now(), + retry_coordination, }); retry_connect_routes( &mut swarm, @@ -415,11 +459,12 @@ async fn run_endpoint_async( external_candidate_ready, Instant::now(), ); - maybe_open_direct_stream( + maybe_open_peer_stream( request_id, &mut direct.pending, &direct.retiring_connections, stream_control.clone(), + mesh_control.clone(), opened_tx.clone(), ); } @@ -457,38 +502,81 @@ async fn run_endpoint_async( None => return Ok(()), }, Some(stream) = incoming_streams.recv() => { - let peer_stream = spawn_stream( - stream.stream, - Some((stream.connection_id, close_connection_tx.clone())), - ); + let peer_stream = spawn_stream(stream.peer_id, stream.stream, None); if incoming_tx.try_send(peer_stream).is_err() { // Dropping the stream closes it. A slow Host cannot create an unbounded queue. } } - Some(connection_id) = close_connection_rx.recv() => { - direct.active.retain(|_, active| *active != connection_id); - retire_established_connection( - &mut swarm, - &mut direct.retiring_connections, - connection_id, - ); + Some(stream) = mesh_incoming.recv() => { + let peer_stream = spawn_stream(stream.peer_id, stream.stream, None); + if mesh_incoming_tx.try_send(peer_stream).is_err() { + // Dropping the stream applies bounded backpressure to Mesh control callers. + } + } + Some(completed) = stream_completed_rx.recv() => { + match completed.kind { + StreamCompletion::Application(connection_id) => { + direct.active.retain(|_, active| *active != connection_id); + } + StreamCompletion::MeshControl { coordination_relay_peers } => { + release_coordination_relays( + &mut swarm, + &mut coordination_relays, + &coordination_relay_peers, + &direct.active, + ); + } + } + let _ = completed.acknowledged.send(()); } Some(opened) = opened_rx.recv() => { 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())), - )) - } + Ok(opened) => match waiter.stream_kind { + StreamKind::Application => { + let connection_id = opened.connection_id; + retire_direct_dials( + &mut swarm, + &mut direct.retiring_connections, + waiter.dials, + Some(connection_id), + ); + direct.active.insert(waiter.peer_id, connection_id); + release_coordination_relays( + &mut swarm, + &mut coordination_relays, + &waiter.coordination_relay_peers, + &direct.active, + ); + Ok(spawn_stream( + waiter.peer_id, + opened.stream, + Some(( + StreamCompletion::Application(connection_id), + stream_completed_tx.clone(), + )), + )) + } + StreamKind::MeshControl => { + retire_direct_dials( + &mut swarm, + &mut direct.retiring_connections, + waiter.dials, + None, + ); + Ok(spawn_stream( + waiter.peer_id, + opened.stream, + Some(( + StreamCompletion::MeshControl { + coordination_relay_peers: waiter + .coordination_relay_peers, + }, + stream_completed_tx.clone(), + )), + )) + } + }, Err(message) => { retire_direct_dials( &mut swarm, @@ -496,22 +584,20 @@ async fn run_endpoint_async( 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 code = match waiter.stream_kind { + StreamKind::Application => "direct_path_unavailable", + StreamKind::MeshControl => "mesh_control_unavailable", + }; + Err(PeerError::new(code, 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() => { @@ -531,11 +617,12 @@ async fn run_endpoint_async( ); let requests = direct.pending.keys().copied().collect::>(); for request_id in requests { - maybe_open_direct_stream( + maybe_open_peer_stream( request_id, &mut direct.pending, &direct.retiring_connections, stream_control.clone(), + mesh_control.clone(), opened_tx.clone(), ); } @@ -577,10 +664,17 @@ async fn run_endpoint_async( &waiter.coordination_relay_peers, &direct.active, ); - let _ = waiter.result.send(Err(PeerError::new( - "direct_path_unavailable", - "no direct path was established before the deadline", - ))); + let (code, message) = match waiter.stream_kind { + StreamKind::Application => ( + "direct_path_unavailable", + "no direct path was established before the deadline", + ), + StreamKind::MeshControl => ( + "mesh_control_unavailable", + "no Mesh control path was established before the deadline", + ), + }; + let _ = waiter.result.send(Err(PeerError::new(code, message))); } } } @@ -592,12 +686,20 @@ type BuiltSwarm = ( Swarm, application_stream::Control, mpsc::Receiver, + application_stream::Control, + mpsc::Receiver, ); fn build_swarm(key: identity::Keypair) -> Result { let (application_stream, control, incoming) = application_stream::Behaviour::new( StreamProtocol::new(APPLICATION_PROTOCOL), INCOMING_STREAM_CAPACITY, + true, + ); + let (mesh_stream, mesh_control, mesh_incoming) = application_stream::Behaviour::new( + StreamProtocol::new(MESH_CONTROL_PROTOCOL), + MESH_INCOMING_STREAM_CAPACITY, + false, ); let swarm = SwarmBuilder::with_existing_identity(key) .with_tokio() @@ -630,10 +732,16 @@ fn build_swarm(key: identity::Keypair) -> Result { )), ping: ping::Behaviour::new(ping::Config::new()), application_stream, + mesh_control: mesh_stream, }) .map_err(native_error) + .map(|builder| { + builder.with_swarm_config(|config| { + config.with_idle_connection_timeout(IDLE_CONNECTION_TIMEOUT) + }) + }) .map(|builder| builder.build())?; - Ok((swarm, control, incoming)) + Ok((swarm, control, incoming, mesh_control, mesh_incoming)) } fn start_connect( @@ -641,10 +749,15 @@ fn start_connect( coordination_relays: &mut HashMap, options: &ConnectOptions, local_peer_id: PeerId, + stream_kind: StreamKind, ) -> Result { if options.route_hints.is_empty() && options.coordination_relays.is_empty() { + let code = match stream_kind { + StreamKind::Application => "direct_path_unavailable", + StreamKind::MeshControl => "mesh_control_unavailable", + }; return Err(PeerError::new( - "direct_path_unavailable", + code, "the peer profile has no route hints or coordination relays", )); } @@ -691,22 +804,34 @@ fn start_connect( }) } -fn maybe_open_direct_stream( +fn maybe_open_peer_stream( request_id: u32, pending: &mut HashMap, retiring_connections: &HashSet, - mut control: application_stream::Control, + mut application_control: application_stream::Control, + mut mesh_control: application_stream::Control, opened_tx: mpsc::Sender, ) { let Some(waiter) = pending.get_mut(&request_id) else { return; }; let peer_id = waiter.peer_id; - if waiter.opening.is_some() || !control.has_connection(peer_id, retiring_connections) { + let available = match waiter.stream_kind { + StreamKind::Application => { + application_control.has_connection(peer_id, retiring_connections) + } + StreamKind::MeshControl => mesh_control.has_connection(peer_id, retiring_connections), + }; + if waiter.opening.is_some() || !available { return; } + let stream_kind = waiter.stream_kind; let retiring_connections = retiring_connections.clone(); waiter.opening = Some(tokio::spawn(async move { + let control = match stream_kind { + StreamKind::Application => &mut application_control, + StreamKind::MeshControl => &mut mesh_control, + }; let result = control .open_stream(peer_id, &retiring_connections) .await @@ -744,24 +869,11 @@ fn handle_swarm_event( let _ = swarm.close_connection(connection_id); } } + for connect in direct.pending.values_mut() { + connect.dials.remove(&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); - } - } } } SwarmEvent::ConnectionClosed { @@ -798,45 +910,18 @@ fn handle_swarm_event( } } } - 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); - } + for connect in direct.pending.values_mut().filter(|connect| { + connect.peer_id == remote_peer_id && connect.stream_kind == StreamKind::Application + }) { + connect.retry_coordination = true; + connect.next_route_attempt = Instant::now(); } } + SwarmEvent::Behaviour(BehaviourEvent::Dcutr(_)) => {} SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Received { peer_id, .. @@ -1058,7 +1143,8 @@ fn retry_connect_routes( .dials .values() .any(|origin| *origin == DialOrigin::CoordinationRoute) - || relayed.get(&peer_id).is_some_and(|ids| !ids.is_empty()) + || (!connect.retry_coordination + && relayed.get(&peer_id).is_some_and(|ids| !ids.is_empty())) { continue; } @@ -1084,6 +1170,7 @@ fn retry_connect_routes( connect .dials .insert(connection_id, DialOrigin::CoordinationRoute); + connect.retry_coordination = false; } } } @@ -1119,16 +1206,6 @@ fn retire_direct_dials( } } -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, @@ -1151,6 +1228,151 @@ fn native_error(error: impl std::fmt::Display) -> PeerError { mod tests { use super::*; + #[tokio::test(flavor = "multi_thread")] + async fn mesh_control_survives_repeated_application_streams_on_one_endpoint() { + let root = std::env::temp_dir().join(format!("maka-peer-test-{}", PeerId::random())); + std::fs::create_dir_all(&root).expect("create test root"); + let left = start(test_endpoint_options(root.join("left.key"))).expect("start left"); + let mut right = start(test_endpoint_options(root.join("right.key"))).expect("start right"); + let route = right + .listen_addresses + .first() + .expect("right listen address") + .clone(); + + let mesh_left = connect_test_stream( + &left, + right.peer_id, + route.clone(), + 1, + StreamKind::MeshControl, + ) + .await; + let mut mesh_right = + tokio::time::timeout(Duration::from_secs(5), right.mesh_incoming.recv()) + .await + .expect("Mesh inbound timeout") + .expect("Mesh inbound stream"); + + for request_id in 2..=3 { + let application_left = connect_test_stream( + &left, + right.peer_id, + route.clone(), + request_id, + StreamKind::Application, + ) + .await; + let application_right = + tokio::time::timeout(Duration::from_secs(5), right.incoming.recv()) + .await + .expect("application inbound timeout") + .expect("application inbound stream"); + close_test_stream(application_left).await; + close_test_stream(application_right).await; + } + + write_test_stream(&mesh_left, b"still-open").await; + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), mesh_right.incoming.recv()) + .await + .expect("Mesh read timeout") + .expect("Mesh stream ended") + .expect("Mesh read failed"), + b"still-open", + ); + close_test_stream(mesh_left).await; + close_test_stream(mesh_right).await; + stop_test_endpoint(left).await; + stop_test_endpoint(right).await; + std::fs::remove_dir_all(root).expect("remove test root"); + } + + fn test_endpoint_options(key_path: PathBuf) -> StartOptions { + StartOptions { + key_path, + expected_peer_id: None, + listen_addresses: vec![ + "/ip4/127.0.0.1/udp/0/quic-v1" + .parse() + .expect("test listen address"), + ], + coordination_relays: Vec::new(), + } + } + + async fn connect_test_stream( + endpoint: &StartedEndpoint, + peer_id: PeerId, + route: Multiaddr, + request_id: u32, + stream_kind: StreamKind, + ) -> PeerStream { + let (result, response) = oneshot::channel(); + endpoint + .commands + .send(EngineCommand::Connect { + options: ConnectOptions { + request_id, + peer_id, + route_hints: vec![route], + coordination_relays: Vec::new(), + deadline: Duration::from_secs(5), + }, + stream_kind, + result, + }) + .await + .expect("send connect"); + tokio::time::timeout(Duration::from_secs(5), response) + .await + .expect("connect timeout") + .expect("connect response") + .expect("connect failed") + } + + async fn write_test_stream(stream: &PeerStream, bytes: &[u8]) { + let (result, response) = oneshot::channel(); + stream + .commands + .send(StreamCommand::Write { + bytes: bytes.to_vec(), + result, + }) + .await + .expect("send write"); + response + .await + .expect("write response") + .expect("write failed"); + } + + async fn close_test_stream(stream: PeerStream) { + let (result, response) = oneshot::channel(); + if stream + .commands + .send(StreamCommand::Close { result }) + .await + .is_err() + { + return; + } + if let Ok(outcome) = response.await { + outcome.expect("close failed"); + } + } + + async fn stop_test_endpoint(endpoint: StartedEndpoint) { + let (result, response) = oneshot::channel(); + endpoint + .commands + .send(EngineCommand::Stop { result }) + .await + .expect("send stop"); + response.await.expect("stop response"); + endpoint.thread.join().expect("join endpoint thread"); + } + #[test] fn coordination_reservation_can_be_recreated_after_its_lifecycle_ends() { let now = Instant::now(); diff --git a/native/runtime-host-peer/src/engine/application_stream.rs b/native/runtime-host-peer/src/engine/application_stream.rs index 7ed44f0ffc..cc9eb0b7e6 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, HashSet, VecDeque}, + collections::{HashMap, HashSet}, convert::Infallible, future::{Ready, ready}, io, @@ -34,9 +34,8 @@ use libp2p::{ upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo}, }, swarm::{ - CloseConnection, ConnectionDenied, ConnectionHandler, ConnectionId, FromSwarm, - NetworkBehaviour, Stream, StreamProtocol, THandler, THandlerInEvent, THandlerOutEvent, - ToSwarm, + ConnectionDenied, ConnectionHandler, ConnectionId, FromSwarm, NetworkBehaviour, Stream, + StreamProtocol, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, behaviour::ConnectionClosed, handler::{ ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound, @@ -51,13 +50,13 @@ const OUTBOUND_COMMAND_CAPACITY: usize = 1; pub(super) struct Behaviour { protocol: StreamProtocol, + direct_only: bool, incoming: mpsc::Sender, shared: Arc>, - closing: VecDeque<(PeerId, ConnectionId)>, } pub(super) struct InboundStream { - pub(super) connection_id: ConnectionId, + pub(super) peer_id: PeerId, pub(super) stream: Stream, } @@ -65,15 +64,16 @@ impl Behaviour { pub(super) fn new( protocol: StreamProtocol, incoming_capacity: usize, + direct_only: bool, ) -> (Self, Control, mpsc::Receiver) { let (incoming, receiver) = mpsc::channel(incoming_capacity); let shared = Arc::new(Mutex::new(DirectConnections::default())); ( Self { protocol, + direct_only, incoming, shared: shared.clone(), - closing: VecDeque::new(), }, Control { shared }, receiver, @@ -81,13 +81,13 @@ impl Behaviour { } fn handler(&mut self, connection_id: ConnectionId, peer_id: PeerId, relayed: bool) -> Handler { - if relayed { + if self.direct_only && relayed { return Handler::relayed(); } let (sender, receiver) = mpsc::channel(OUTBOUND_COMMAND_CAPACITY); lock(&self.shared).insert(connection_id, peer_id, sender); Handler::direct( - connection_id, + peer_id, self.protocol.clone(), self.incoming.clone(), receiver, @@ -132,21 +132,15 @@ impl NetworkBehaviour for Behaviour { fn on_connection_handler_event( &mut self, - peer_id: PeerId, - connection_id: ConnectionId, - _: THandlerOutEvent, + _: PeerId, + _: ConnectionId, + event: THandlerOutEvent, ) { - self.closing.push_back((peer_id, connection_id)); + libp2p::core::util::unreachable(event); } fn poll(&mut self, _: &mut Context<'_>) -> Poll>> { - let Some((peer_id, connection_id)) = self.closing.pop_front() else { - return Poll::Pending; - }; - Poll::Ready(ToSwarm::CloseConnection { - peer_id, - connection: CloseConnection::One(connection_id), - }) + Poll::Pending } } @@ -164,7 +158,7 @@ impl Control { &mut self, peer_id: PeerId, excluded: &HashSet, - ) -> Result { + ) -> Result { let (connection_id, sender) = lock(&self.shared) .connection(peer_id, excluded) .ok_or(OpenStreamError::NoDirectConnection)?; @@ -176,14 +170,14 @@ impl Control { let stream = receiver .await .map_err(|_| OpenStreamError::ConnectionClosed)??; - Ok(OpenedApplicationStream { + Ok(OpenedStream { connection_id, stream, }) } } -pub(super) struct OpenedApplicationStream { +pub(super) struct OpenedStream { pub(super) connection_id: ConnectionId, pub(super) stream: Stream, } @@ -257,53 +251,50 @@ fn lock(shared: &Arc>) -> MutexGuard<'_, DirectConnecti } pub(super) struct Handler { - connection_id: Option, + peer_id: Option, protocol: Option, incoming: Option>, commands: Option>, - pending: Option>>, - accepted_inbound: bool, - close: bool, + pending: HashMap>>, + next_request_id: u64, } impl Handler { fn direct( - connection_id: ConnectionId, + peer_id: PeerId, protocol: StreamProtocol, incoming: mpsc::Sender, commands: mpsc::Receiver, ) -> Self { Self { - connection_id: Some(connection_id), + peer_id: Some(peer_id), protocol: Some(protocol), incoming: Some(incoming), commands: Some(commands), - pending: None, - accepted_inbound: false, - close: false, + pending: HashMap::new(), + next_request_id: 0, } } fn relayed() -> Self { Self { - connection_id: None, + peer_id: None, protocol: None, incoming: None, commands: None, - pending: None, - accepted_inbound: false, - close: false, + pending: HashMap::new(), + next_request_id: 0, } } } impl ConnectionHandler for Handler { type FromBehaviour = Infallible; - type ToBehaviour = (); + type ToBehaviour = Infallible; type InboundProtocol = ProtocolUpgrade; type OutboundProtocol = ProtocolUpgrade; type InboundOpenInfo = (); - type OutboundOpenInfo = (); + type OutboundOpenInfo = u64; fn listen_protocol( &self, @@ -321,79 +312,68 @@ impl ConnectionHandler for Handler { fn poll( &mut self, context: &mut Context<'_>, - ) -> Poll> + ) -> Poll> { - if std::mem::take(&mut self.close) { - return Poll::Ready(libp2p::swarm::ConnectionHandlerEvent::NotifyBehaviour(())); - } - 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 { return Poll::Pending; }; - match commands.poll_recv(context) { - Poll::Ready(Some(command)) => { - let protocol = self - .protocol - .clone() - .expect("only direct handlers receive stream commands"); - self.pending = Some(command.result); - Poll::Ready( - libp2p::swarm::ConnectionHandlerEvent::OutboundSubstreamRequest { - protocol: libp2p::swarm::SubstreamProtocol::new( - ProtocolUpgrade(vec![protocol]), - (), - ), - }, - ) + loop { + match commands.poll_recv(context) { + Poll::Ready(Some(command)) if command.result.is_closed() => continue, + Poll::Ready(Some(command)) => { + let protocol = self + .protocol + .clone() + .expect("only direct handlers receive stream commands"); + let request_id = self.next_request_id; + self.next_request_id = self.next_request_id.wrapping_add(1); + self.pending.insert(request_id, command.result); + return Poll::Ready( + libp2p::swarm::ConnectionHandlerEvent::OutboundSubstreamRequest { + protocol: libp2p::swarm::SubstreamProtocol::new( + ProtocolUpgrade(vec![protocol]), + request_id, + ), + }, + ); + } + Poll::Ready(None) | Poll::Pending => return Poll::Pending, } - Poll::Ready(None) | Poll::Pending => Poll::Pending, } } fn on_connection_event( &mut self, - event: ConnectionEvent, + event: ConnectionEvent< + Self::InboundProtocol, + Self::OutboundProtocol, + Self::InboundOpenInfo, + Self::OutboundOpenInfo, + >, ) { match event { ConnectionEvent::FullyNegotiatedInbound(FullyNegotiatedInbound { protocol: (stream, _), .. }) => { - if self.accepted_inbound { - self.close = true; - return; - } - self.accepted_inbound = true; - if self.incoming.as_ref().is_none_or(|incoming| { - incoming - .try_send(InboundStream { - connection_id: self - .connection_id - .expect("direct handlers have a connection id"), - stream, - }) - .is_err() - }) { - self.close = true; + if let Some(incoming) = self.incoming.as_ref() { + let _ = incoming.try_send(InboundStream { + peer_id: self.peer_id.expect("direct handlers have a peer id"), + stream, + }); } } ConnectionEvent::FullyNegotiatedOutbound(FullyNegotiatedOutbound { protocol: (stream, _), - .. + info, }) => { - let Some(result) = self.pending.take() else { + let Some(result) = self.pending.remove(&info) else { return; }; let _ = result.send(Ok(stream)); } - ConnectionEvent::DialUpgradeError(DialUpgradeError { error, .. }) => { - let Some(result) = self.pending.take() else { + ConnectionEvent::DialUpgradeError(DialUpgradeError { info, error }) => { + let Some(result) = self.pending.remove(&info) else { return; }; let error = match error { @@ -459,7 +439,7 @@ mod tests { fn application_protocol_is_registered_only_on_direct_connections() { let protocol = StreamProtocol::new("/maka/test/1"); let peer_id = PeerId::random(); - let (mut behaviour, control, _) = Behaviour::new(protocol.clone(), 1); + let (mut behaviour, control, _) = Behaviour::new(protocol.clone(), 1, true); let relayed = behaviour .handle_established_outbound_connection( diff --git a/native/runtime-host-peer/src/engine/peer_stream.rs b/native/runtime-host-peer/src/engine/peer_stream.rs index 6cd0556487..83ecd6e1e3 100644 --- a/native/runtime-host-peer/src/engine/peer_stream.rs +++ b/native/runtime-host-peer/src/engine/peer_stream.rs @@ -18,17 +18,19 @@ */ use futures::{AsyncReadExt as _, AsyncWriteExt as _}; -use libp2p::swarm::ConnectionId; -use tokio::sync::{mpsc, oneshot}; +use libp2p::PeerId; +use tokio::sync::{mpsc, oneshot, watch}; -use super::PeerError; +use super::{CompletedStream, PeerError, StreamCompletion}; const QUEUE_CAPACITY: usize = 64; const CHUNK_BYTES: usize = 64 * 1024; pub struct PeerStream { + pub peer_id: PeerId, pub incoming: mpsc::Receiver, PeerError>>, pub commands: mpsc::Sender, + pub abort: watch::Sender, } pub enum StreamCommand { @@ -39,59 +41,104 @@ pub enum StreamCommand { Close { result: oneshot::Sender>, }, - Abort, } pub(super) fn spawn_stream( + peer_id: PeerId, stream: libp2p::swarm::Stream, - close_connection: Option<(ConnectionId, mpsc::Sender)>, + completion: Option<(StreamCompletion, mpsc::Sender)>, ) -> PeerStream { let (incoming_tx, incoming_rx) = mpsc::channel(QUEUE_CAPACITY); let (command_tx, mut command_rx) = mpsc::channel(QUEUE_CAPACITY); + let (abort_tx, mut abort_rx) = watch::channel(false); + let abort_guard = abort_tx.clone(); tokio::spawn(async move { + let _abort_guard = abort_guard; let (mut reader, mut writer) = stream.split(); let mut buffer = vec![0_u8; CHUNK_BYTES]; + let mut pending_read: Option, PeerError>> = None; + let mut finish_after_delivery = false; + let mut close_result = None; loop { tokio::select! { - read = reader.read(&mut buffer) => match read { - Ok(0) => break, - Ok(size) => { - if incoming_tx.send(Ok(buffer[..size].to_vec())).await.is_err() { - break; - } - } - Err(error) => { - let _ = incoming_tx.send(Err(PeerError::new( - "peer_native_failed", - error.to_string(), - ))).await; - break; - } - }, + biased; + changed = abort_rx.changed() => { + let _ = changed; + break; + } command = command_rx.recv() => match command { Some(StreamCommand::Write { bytes, result }) => { - let outcome = writer.write_all(&bytes).await - .map_err(|error| PeerError::new("peer_native_failed", error.to_string())); + let outcome = tokio::select! { + biased; + _ = abort_rx.changed() => Err(PeerError::new( + "peer_stream_aborted", + "peer stream was aborted", + )), + outcome = writer.write_all(&bytes) => outcome.map_err(|error| { + PeerError::new("peer_native_failed", error.to_string()) + }), + }; let failed = outcome.is_err(); let _ = result.send(outcome); if failed { break; } } Some(StreamCommand::Close { result }) => { - let outcome = writer.close().await - .map_err(|error| PeerError::new("peer_native_failed", error.to_string())); - let _ = result.send(outcome); + let outcome = tokio::select! { + biased; + _ = abort_rx.changed() => Err(PeerError::new( + "peer_stream_aborted", + "peer stream was aborted", + )), + outcome = writer.close() => outcome.map_err(|error| { + PeerError::new("peer_native_failed", error.to_string()) + }), + }; + close_result = Some((result, outcome)); break; } - Some(StreamCommand::Abort) | None => break, - } + None => break, + }, + permit = incoming_tx.reserve(), if pending_read.is_some() => match permit { + Ok(permit) => { + permit.send(pending_read.take().expect("read is pending")); + if finish_after_delivery { break; } + } + Err(_) => break, + }, + read = reader.read(&mut buffer), if pending_read.is_none() => match read { + Ok(0) => break, + Ok(size) => pending_read = Some(Ok(buffer[..size].to_vec())), + Err(error) => { + pending_read = Some(Err(PeerError::new( + "peer_native_failed", + error.to_string(), + ))); + finish_after_delivery = true; + } + }, + } + } + if let Some((completion, completed)) = completion { + let (acknowledged, acknowledgment) = oneshot::channel(); + if completed + .send(CompletedStream { + kind: completion, + acknowledged, + }) + .await + .is_ok() + { + let _ = acknowledgment.await; } } - if let Some((connection_id, close)) = close_connection { - let _ = close.send(connection_id).await; + if let Some((result, outcome)) = close_result { + let _ = result.send(outcome); } }); PeerStream { + peer_id, incoming: incoming_rx, commands: command_tx, + abort: abort_tx, } } diff --git a/packages/runtime-host/package.json b/packages/runtime-host/package.json index d7591a76c4..32d639dee7 100644 --- a/packages/runtime-host/package.json +++ b/packages/runtime-host/package.json @@ -9,6 +9,7 @@ "./adapter": "./dist/adapter/index.js", "./protocol": "./dist/protocol/index.js", "./client": "./dist/client/index.js", + "./peer-mesh": "./dist/peer-mesh/index.js", "./operator": "./dist/operator/index.js", "./execution-candidate-main": "./dist/execution-candidate-main.js", "./server": "./dist/server/index.js", diff --git a/packages/runtime-host/src/__tests__/peer-listener.test.ts b/packages/runtime-host/src/__tests__/peer-listener.test.ts index 72db115299..f1a3a2d62a 100644 --- a/packages/runtime-host/src/__tests__/peer-listener.test.ts +++ b/packages/runtime-host/src/__tests__/peer-listener.test.ts @@ -27,7 +27,7 @@ import type { } from '../transport/peer-native.js'; test('bounds and aborts pending peer authentication', async () => { - const streams = Array.from({ length: 17 }, () => pendingStream()); + const streams = Array.from({ length: 17 }, (_, index) => pendingStream(`remote-peer-${index}`)); const listener = createRuntimeHostPeerListener(endpointWith([...streams]), {} as never, () => {}); await waitForImmediate(); @@ -75,6 +75,7 @@ test('rechecks peer authority at admission after the authentication response is let aborted = false; let reads = 0; const stream: RuntimeHostPeerNativeStream = { + peerId: 'remote-peer', read: async () => (reads++ === 0 ? Buffer.from('{"v":1,"credential":"revoked"}\n') : null), write: async () => writeReleased, close: async () => undefined, @@ -105,6 +106,24 @@ test('rechecks peer authority at admission after the authentication response is await listener.cleanup(); }); +test('bounds active application streams from one authenticated peer', async () => { + const streams = Array.from({ length: 5 }, () => authenticatedPendingStream('remote-peer')); + let accepted = 0; + const listener = createRuntimeHostPeerListener( + endpointWith([...streams]), + { authenticate: () => ({ operationGrants: 'all' }) } as never, + () => { + accepted += 1; + }, + ); + await waitForImmediate(); + await waitForImmediate(); + + assert.equal(accepted, 4); + assert.equal(streams[4]?.aborted, true); + await listener.cleanup(); +}); + function endpointWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerNativeEndpoint { return { peerId: 'peer', @@ -112,19 +131,26 @@ function endpointWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerNa connect: async () => { throw new Error('not used'); }, + connectMeshControl: async () => { + throw new Error('not used'); + }, cancelConnect: async () => false, accept: async () => streams.shift() ?? null, + acceptMeshControl: async () => null, close: async () => undefined, }; } -function pendingStream(): RuntimeHostPeerNativeStream & { readonly aborted: boolean } { +function pendingStream( + peerId = 'remote-peer', +): RuntimeHostPeerNativeStream & { readonly aborted: boolean } { let finish!: (value: null) => void; const read = new Promise((resolve) => { finish = resolve; }); let aborted = false; return { + peerId, get aborted() { return aborted; }, @@ -138,6 +164,36 @@ function pendingStream(): RuntimeHostPeerNativeStream & { readonly aborted: bool }; } +function authenticatedPendingStream( + peerId: string, +): RuntimeHostPeerNativeStream & { readonly aborted: boolean } { + let finish!: (value: null) => void; + const pending = new Promise((resolve) => { + finish = resolve; + }); + let first = true; + let aborted = false; + return { + peerId, + get aborted() { + return aborted; + }, + read: async () => { + if (first) { + first = false; + return Buffer.from('{"v":1,"credential":"accepted"}\n'); + } + return pending; + }, + write: async () => undefined, + close: async () => finish(null), + abort: () => { + aborted = true; + finish(null); + }, + }; +} + function recordingStream(initial: Buffer): RuntimeHostPeerNativeStream & { readonly writes: readonly Buffer[]; readonly closed: boolean; @@ -146,6 +202,7 @@ function recordingStream(initial: Buffer): RuntimeHostPeerNativeStream & { let closed = false; const writes: Buffer[] = []; return { + peerId: 'remote-peer', get writes() { return writes; }, diff --git a/packages/runtime-host/src/__tests__/peer-mesh.test.ts b/packages/runtime-host/src/__tests__/peer-mesh.test.ts new file mode 100644 index 0000000000..81d88fc30b --- /dev/null +++ b/packages/runtime-host/src/__tests__/peer-mesh.test.ts @@ -0,0 +1,346 @@ +/* + * 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 assert from 'node:assert/strict'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { setImmediate as waitForImmediate } from 'node:timers/promises'; +import { test } from 'node:test'; +import type { RuntimeHostPeerNativeStream } from '../transport/peer-native.js'; +import { + decodeSignedPeerMeshRoster, + generatePeerMeshAuthorityKeyPair, + peerMeshId, + signPeerMeshRoster, +} from '../peer-mesh/model.js'; +import { openPeerMeshNode, type PeerMeshNode, type PeerMeshTransport } from '../peer-mesh/node.js'; + +test('authenticates three peers, consumes invitations once, and keeps authority state private', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-')); + const network = new MemoryPeerNetwork(); + const peers = ['peer-a', 'peer-b', 'peer-c'].map((peerId) => network.create(peerId)); + const nodes: PeerMeshNode[] = []; + try { + for (const [index, peer] of peers.entries()) { + nodes.push(await openPeerMeshNode({ dataRoot: join(root, String(index)), peer })); + } + const [authority, memberB, memberC] = nodes as [PeerMeshNode, PeerMeshNode, PeerMeshNode]; + const mesh = await authority.create(); + assert.deepEqual(mesh.authority.coordinationRelays, ['/memory/relay/peer-a']); + const serving = authority.serve(); + + const contested = await authority.invite(mesh.roster.roster.meshId); + assert.deepEqual(contested.coordinationRelays, mesh.authority.coordinationRelays); + const attempts = await Promise.allSettled([memberB.join(contested), memberC.join(contested)]); + assert.equal(attempts.filter(({ status }) => status === 'fulfilled').length, 1); + assert.equal(attempts.filter(({ status }) => status === 'rejected').length, 1); + + const loser = attempts[0]?.status === 'rejected' ? memberB : memberC; + await loser.join(await authority.invite(mesh.roster.roster.meshId)); + const current = authority.status()[0]; + assert.deepEqual(current?.roster.roster.members, ['peer-a', 'peer-b', 'peer-c']); + assert.equal('authorityPrivateKey' in (current ?? {}), false); + + await authority.remove(mesh.roster.roster.meshId, 'peer-b'); + assert.deepEqual(authority.status()[0]?.roster.roster.members, ['peer-a', 'peer-c']); + + const closing = authority.close(); + await assert.rejects(authority.invite(mesh.roster.roster.meshId), /closed/u); + await closing; + await peers[0]!.close(); + await serving; + } finally { + await Promise.allSettled(nodes.map((node) => node.close())); + await Promise.allSettled(peers.map((peer) => peer.close())); + await rm(root, { recursive: true, force: true }); + } +}); + +test('rejects a modified authority-signed roster', () => { + const keys = generatePeerMeshAuthorityKeyPair(); + const signed = signPeerMeshRoster( + { + version: 1, + meshId: peerMeshId(keys.publicKey), + revision: 1, + members: ['peer-a'], + closed: false, + }, + keys, + ); + assert.throws(() => + decodeSignedPeerMeshRoster({ + ...signed, + roster: { ...signed.roster, members: ['peer-b'] }, + }), + ); +}); + +test('closed Mesh records do not permanently consume membership capacity', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-capacity-')); + const peer = new MemoryPeerNetwork().create('peer-a'); + const node = await openPeerMeshNode({ dataRoot: root, peer }); + try { + for (let index = 0; index < 16; index += 1) { + const mesh = await node.create(); + await node.closeMesh(mesh.roster.roster.meshId); + } + assert.equal((await node.create()).roster.roster.closed, false); + assert.equal(node.status().length, 16); + } finally { + await node.close(); + await peer.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test('retries a committed invitation redemption for the same authenticated peer', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-retry-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const memberPeer = network.create('peer-b'); + const authorityRoot = join(root, 'authority'); + let now = Date.now(); + let authority = await openPeerMeshNode({ + dataRoot: authorityRoot, + peer: authorityPeer, + now: () => now, + }); + const member = await openPeerMeshNode({ dataRoot: join(root, 'member'), peer: memberPeer }); + let serving = authority.serve(); + try { + const mesh = await authority.create(); + const invitation = await authority.invite(mesh.roster.roster.meshId, { ttlMs: 1_000 }); + authorityPeer.failNextResponse(); + + await assert.rejects(member.join(invitation)); + await authority.closeMesh(mesh.roster.roster.meshId); + await authority.close(); + await serving; + + now += 2_000; + await assert.rejects( + openPeerMeshNode({ dataRoot: authorityRoot, peer: memberPeer }), + /different peer identity/u, + ); + authority = await openPeerMeshNode({ + dataRoot: authorityRoot, + peer: authorityPeer, + now: () => now, + }); + serving = authority.serve(); + const joined = await member.join(invitation); + assert.deepEqual(joined.roster.roster.members, ['peer-a', 'peer-b']); + assert.equal(joined.roster.roster.closed, true); + assert.equal(authority.status()[0]?.roster.roster.revision, 3); + + await authority.close(); + await authorityPeer.close(); + await serving; + } finally { + await Promise.allSettled([authority.close(), member.close()]); + await Promise.allSettled([authorityPeer.close(), memberPeer.close()]); + await Promise.allSettled([serving]); + await rm(root, { recursive: true, force: true }); + } +}); + +test('cancels a redemption stalled after the control connection opens', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-abort-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const memberPeer = network.create('peer-b'); + const authority = await openPeerMeshNode({ + dataRoot: join(root, 'authority'), + peer: authorityPeer, + }); + const member = await openPeerMeshNode({ dataRoot: join(root, 'member'), peer: memberPeer }); + const serving = authority.serve(); + try { + const mesh = await authority.create(); + const invitation = await authority.invite(mesh.roster.roster.meshId); + authorityPeer.stallNextControl(); + const abort = new AbortController(); + const joining = member.join(invitation, abort.signal); + await waitForImmediate(); + abort.abort(); + await assert.rejects( + joining, + (error: unknown) => error instanceof Error && error.name === 'AbortError', + ); + assert.deepEqual(authority.status()[0]?.roster.roster.members, ['peer-a']); + } finally { + await Promise.allSettled([authority.close(), member.close()]); + await Promise.allSettled([authorityPeer.close(), memberPeer.close(), serving]); + await rm(root, { recursive: true, force: true }); + } +}); + +class MemoryPeerNetwork { + readonly #peers = new Map(); + + create(peerId: string): MemoryPeerClient { + const peer = new MemoryPeerClient(peerId, this.#peers); + this.#peers.set(peerId, peer); + return peer; + } +} + +class MemoryPeerClient implements PeerMeshTransport { + #meshServer: + | { + readonly onStream: (stream: RuntimeHostPeerNativeStream) => void; + readonly stop: () => void; + } + | undefined; + #closed = false; + #failNextResponse = false; + #stallNextControl = false; + + constructor( + private readonly peerId: string, + private readonly peers: ReadonlyMap, + ) {} + + identity() { + return { + peerId: this.peerId, + listenAddresses: [`/memory/${this.peerId}`], + coordinationRelays: [`/memory/relay/${this.peerId}`], + } as const; + } + + async connectMeshControl(input: { + readonly peerId: string; + }): Promise { + const remote = this.peers.get(input.peerId); + if (!remote) throw new Error('Peer is unavailable'); + const [localStream, remoteStream] = memoryStreamPair(this.peerId, input.peerId); + if (remote.#failNextResponse) { + remote.#failNextResponse = false; + remoteStream.failNextWrite(); + } + remote.accept(remoteStream); + return localStream; + } + + failNextResponse(): void { + this.#failNextResponse = true; + } + + stallNextControl(): void { + this.#stallNextControl = true; + } + + serveMeshControl( + onStream: (stream: RuntimeHostPeerNativeStream) => void, + signal: AbortSignal, + ): Promise { + if (this.#meshServer) return Promise.reject(new Error('Mesh control is already served')); + signal.throwIfAborted(); + return new Promise((resolve) => { + const stop = () => { + if (this.#meshServer?.stop === stop) this.#meshServer = undefined; + signal.removeEventListener('abort', stop); + resolve(); + }; + this.#meshServer = { onStream, stop }; + signal.addEventListener('abort', stop, { once: true }); + if (signal.aborted) stop(); + }); + } + + close(): Promise { + if (this.#closed) return Promise.resolve(); + this.#closed = true; + this.#meshServer?.stop(); + return Promise.resolve(); + } + + accept(stream: RuntimeHostPeerNativeStream): void { + if (this.#stallNextControl) { + this.#stallNextControl = false; + return; + } + const server = this.#meshServer; + if (server) server.onStream(stream); + else stream.abort(); + } +} + +function memoryStreamPair(localPeerId: string, remotePeerId: string): [MemoryStream, MemoryStream] { + const local = new MemoryStream(remotePeerId); + const remote = new MemoryStream(localPeerId); + local.connect(remote); + remote.connect(local); + return [local, remote]; +} + +class MemoryStream implements RuntimeHostPeerNativeStream { + readonly #incoming: Array = []; + readonly #waiters: Array<(chunk: Buffer | null) => void> = []; + #remote: MemoryStream | undefined; + #closed = false; + #failNextWrite = false; + + constructor(readonly peerId: string) {} + + connect(remote: MemoryStream): void { + this.#remote = remote; + } + + read(): Promise { + const chunk = this.#incoming.shift(); + if (chunk !== undefined) return Promise.resolve(chunk); + return new Promise((resolve) => this.#waiters.push(resolve)); + } + + async write(bytes: Buffer): Promise { + if (this.#closed || !this.#remote) throw new Error('Stream is closed'); + if (this.#failNextWrite) { + this.#failNextWrite = false; + throw new Error('Simulated response loss'); + } + this.#remote.push(Buffer.from(bytes)); + } + + failNextWrite(): void { + this.#failNextWrite = true; + } + + async close(): Promise { + if (this.#closed) return; + this.#closed = true; + this.#remote?.push(null); + } + + abort(): void { + if (this.#closed) return; + this.#closed = true; + this.push(null); + this.#remote?.push(null); + } + + push(chunk: Buffer | null): void { + const waiter = this.#waiters.shift(); + if (waiter) waiter(chunk); + else this.#incoming.push(chunk); + } +} diff --git a/packages/runtime-host/src/__tests__/peer-native.test.ts b/packages/runtime-host/src/__tests__/peer-native.test.ts index 423567335a..97f079daea 100644 --- a/packages/runtime-host/src/__tests__/peer-native.test.ts +++ b/packages/runtime-host/src/__tests__/peer-native.test.ts @@ -40,13 +40,14 @@ test('shares one peer endpoint while cancelling connection attempts independentl await writeFile( nativePath, `let finishAccept; +let finishMeshAccept; 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), + failEndpoint: () => { finishAccept?.(null); finishMeshAccept?.(null); }, ensurePeerIdentity: async () => 'client', startPeerEndpoint: () => { stats.starts += 1; @@ -58,6 +59,11 @@ module.exports = { if (peerId === 'ready') return Promise.resolve(stream); return new Promise((_resolve, reject) => pending.set(requestId, reject)); }, + connectMeshControl: ({ 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) { @@ -69,7 +75,8 @@ module.exports = { return true; }, accept: () => new Promise((resolve) => { finishAccept = resolve; }), - close: async () => { stats.closes += 1; finishAccept?.(null); }, + acceptMeshControl: () => new Promise((resolve) => { finishMeshAccept = resolve; }), + close: async () => { stats.closes += 1; finishAccept?.(null); finishMeshAccept?.(null); }, }; }, }; @@ -108,13 +115,42 @@ module.exports = { } }); -test('loads a relative native module path from the process working directory', async () => { +test('rejects an incomplete endpoint API and loads a compatible relative native module', async () => { const directory = await mkdtemp(join(tmpdir(), 'maka-peer-native-')); try { + const incompletePath = join(directory, 'incomplete.cjs'); + await writeFile( + incompletePath, + 'module.exports = { ensurePeerIdentity: async () => "peer", startPeerEndpoint: () => ({ peerId: "peer", listenAddresses: [] }) };\n', + ); + assert.throws( + () => + startRuntimeHostPeerEndpoint({ + nativePath: relative(process.cwd(), incompletePath), + keyPath: 'unused', + }), + (error: unknown) => + error instanceof RuntimeHostPeerError && error.code === 'peer_native_unavailable', + ); + const modulePath = join(directory, 'peer.cjs'); await writeFile( modulePath, - 'module.exports = { ensurePeerIdentity: async () => "peer", startPeerEndpoint: () => ({ peerId: "peer", listenAddresses: [] }) };\n', + `const stream = { read: async () => null, write: async () => {}, close: async () => {}, abort: () => {} }; +module.exports = { + ensurePeerIdentity: async () => 'peer', + startPeerEndpoint: () => ({ + peerId: 'peer', + listenAddresses: [], + connect: async () => stream, + connectMeshControl: async () => stream, + cancelConnect: async () => true, + accept: async () => null, + acceptMeshControl: async () => null, + close: async () => {}, + }), +}; +`, ); const endpoint = startRuntimeHostPeerEndpoint({ nativePath: relative(process.cwd(), modulePath), @@ -156,6 +192,7 @@ test('bounds and separates the peer credential preface from Runtime Host frames' function streamWith(chunk: Buffer): RuntimeHostPeerNativeStream { let pending: Buffer | null = chunk; return { + peerId: 'remote-peer', read: async () => { const value = pending; pending = null; diff --git a/packages/runtime-host/src/client/peer-client.ts b/packages/runtime-host/src/client/peer-client.ts index df6707e027..d7ca0c6f1a 100644 --- a/packages/runtime-host/src/client/peer-client.ts +++ b/packages/runtime-host/src/client/peer-client.ts @@ -33,15 +33,32 @@ export interface RuntimeHostPeerConnectInput { } export interface RuntimeHostPeerClient { + identity(): Readonly<{ + peerId: string; + listenAddresses: readonly string[]; + coordinationRelays: readonly string[]; + }>; connect( input: RuntimeHostPeerConnectInput, signal?: AbortSignal, ): Promise; + connectMeshControl( + input: RuntimeHostPeerConnectInput, + signal?: AbortSignal, + ): Promise; + serveMeshControl( + onStream: (stream: RuntimeHostPeerNativeStream) => void, + signal: AbortSignal, + ): Promise; close(): Promise; } export function createRuntimeHostPeerClientFromEnvironment( environment: NodeJS.ProcessEnv = process.env, + options: { + readonly listenAddresses?: readonly string[]; + readonly coordinationRelays?: readonly string[]; + } = {}, ): RuntimeHostPeerClient { const nativePath = environment.MAKA_RUNTIME_HOST_PEER_NATIVE_PATH; const keyPath = environment.MAKA_RUNTIME_HOST_PEER_KEY_PATH; @@ -51,12 +68,14 @@ export function createRuntimeHostPeerClientFromEnvironment( 'Experimental direct peer requires MAKA_RUNTIME_HOST_PEER_NATIVE_PATH and MAKA_RUNTIME_HOST_PEER_KEY_PATH', ); } - return createRuntimeHostPeerClient({ nativePath, keyPath }); + return createRuntimeHostPeerClient({ nativePath, keyPath, ...options }); } export function createRuntimeHostPeerClient(input: { readonly nativePath: string; readonly keyPath: string; + readonly listenAddresses?: readonly string[]; + readonly coordinationRelays?: readonly string[]; }): RuntimeHostPeerClient { return new RuntimeHostPeerClientImpl(input); } @@ -64,26 +83,104 @@ export function createRuntimeHostPeerClient(input: { class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { readonly #nativePath: string; readonly #keyPath: string; + readonly #listenAddresses: readonly string[] | undefined; + readonly #coordinationRelays: readonly string[] | undefined; #endpoint: RuntimeHostPeerNativeEndpoint | undefined; #draining: Promise | undefined; + #meshDraining: Promise | undefined; + #meshConsumer: + | { + readonly onStream: (stream: RuntimeHostPeerNativeStream) => void; + readonly resolve: () => void; + readonly reject: (error: Error) => void; + } + | undefined; #terminalError: Error | undefined; #nextRequestId = 1; #closed = false; #closeTask: Promise | undefined; - constructor(input: { readonly nativePath: string; readonly keyPath: string }) { + constructor(input: { + readonly nativePath: string; + readonly keyPath: string; + readonly listenAddresses?: readonly string[]; + readonly coordinationRelays?: readonly string[]; + }) { this.#nativePath = input.nativePath; this.#keyPath = input.keyPath; + this.#listenAddresses = input.listenAddresses; + this.#coordinationRelays = input.coordinationRelays; + } + + identity(): Readonly<{ + peerId: string; + listenAddresses: readonly string[]; + coordinationRelays: readonly string[]; + }> { + const endpoint = this.#requireEndpoint(); + return Object.freeze({ + peerId: endpoint.peerId, + listenAddresses: Object.freeze([...endpoint.listenAddresses]), + coordinationRelays: Object.freeze([...(this.#coordinationRelays ?? [])]), + }); } async connect( input: RuntimeHostPeerConnectInput, signal?: AbortSignal, + ): Promise { + return this.#connect(input, signal, 'application'); + } + + async connectMeshControl( + input: RuntimeHostPeerConnectInput, + signal?: AbortSignal, + ): Promise { + return this.#connect(input, signal, 'mesh-control'); + } + + serveMeshControl( + onStream: (stream: RuntimeHostPeerNativeStream) => void, + signal: AbortSignal, + ): Promise { + signal.throwIfAborted(); + if (this.#meshConsumer) { + return Promise.reject(new Error('Runtime Host peer Mesh control is already being served')); + } + this.#requireEndpoint(); + let resolve!: () => void; + let reject!: (error: Error) => void; + const serving = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + const consumer = { onStream, resolve, reject }; + this.#meshConsumer = consumer; + const stop = () => { + if (this.#meshConsumer !== consumer) return; + this.#meshConsumer = undefined; + resolve(); + }; + signal.addEventListener('abort', stop, { once: true }); + if (signal.aborted) stop(); + return serving.finally(() => { + signal.removeEventListener('abort', stop); + if (this.#meshConsumer === consumer) this.#meshConsumer = undefined; + }); + } + + async #connect( + input: RuntimeHostPeerConnectInput, + signal: AbortSignal | undefined, + kind: 'application' | 'mesh-control', ): Promise { signal?.throwIfAborted(); const endpoint = this.#requireEndpoint(); const requestId = this.#allocateRequestId(); - const connection = endpoint.connect({ ...input, requestId }); + const connection = endpoint[kind === 'application' ? 'connect' : 'connectMeshControl']({ + ...input, + requestId, + }); let settled = false; const cancel = () => { void cancelPeerConnect(endpoint, requestId, () => settled); @@ -125,9 +222,12 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { const endpoint = startRuntimeHostPeerEndpoint({ nativePath: this.#nativePath, keyPath: this.#keyPath, + ...(this.#listenAddresses ? { listenAddresses: this.#listenAddresses } : {}), + ...(this.#coordinationRelays ? { coordinationRelays: this.#coordinationRelays } : {}), }); this.#endpoint = endpoint; this.#draining = this.#drainInbound(endpoint); + this.#meshDraining = this.#drainMeshInbound(endpoint); return endpoint; } @@ -150,6 +250,35 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { } } + async #drainMeshInbound(endpoint: RuntimeHostPeerNativeEndpoint): Promise { + try { + while (true) { + const stream = await endpoint.acceptMeshControl(); + if (!stream) { + const error = new Error('Runtime Host peer networking stopped unexpectedly'); + if (!this.#closed) this.#terminalError = error; + this.#finishMeshConsumer(this.#closed ? undefined : error); + return; + } + const consumer = this.#meshConsumer; + if (consumer) consumer.onStream(stream); + else stream.abort(); + } + } catch (error) { + const failure = error instanceof Error ? error : new Error(String(error)); + if (!this.#closed) this.#terminalError = failure; + this.#finishMeshConsumer(this.#closed ? undefined : failure); + } + } + + #finishMeshConsumer(error?: Error): void { + const consumer = this.#meshConsumer; + if (!consumer) return; + this.#meshConsumer = undefined; + if (error) consumer.reject(error); + else consumer.resolve(); + } + async #close(): Promise { this.#closed = true; const endpoint = this.#endpoint; @@ -163,7 +292,7 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { closeFailed = true; closeError = error; } - await this.#draining; + await Promise.all([this.#draining, this.#meshDraining]); if (closeFailed) throw closeError; } diff --git a/packages/runtime-host/src/peer-mesh/index.ts b/packages/runtime-host/src/peer-mesh/index.ts new file mode 100644 index 0000000000..85f3fa825b --- /dev/null +++ b/packages/runtime-host/src/peer-mesh/index.ts @@ -0,0 +1,32 @@ +/* + * 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. + */ + +export { + decodePeerMeshInvitation, + type PeerMeshAuthorityTarget, + type PeerMeshInvitationV1, + type PeerMeshRosterV1, + type SignedPeerMeshRosterV1, +} from './model.js'; +export { + openPeerMeshNode, + type PeerMeshNode, + type PeerMeshStatus, + type PeerMeshTransport, +} from './node.js'; diff --git a/packages/runtime-host/src/peer-mesh/model.ts b/packages/runtime-host/src/peer-mesh/model.ts new file mode 100644 index 0000000000..bb618322f5 --- /dev/null +++ b/packages/runtime-host/src/peer-mesh/model.ts @@ -0,0 +1,344 @@ +/* + * 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 { + createHash, + createPrivateKey, + createPublicKey, + generateKeyPairSync, + randomBytes, + sign, + timingSafeEqual, + verify, +} from 'node:crypto'; + +export const PEER_MESH_MAX_MEMBERS = 64; +export const PEER_MESH_MAX_MESHES = 16; +export const PEER_MESH_MAX_PENDING_INVITATIONS = 32; +export const PEER_MESH_MAX_INVITATION_RECORDS = PEER_MESH_MAX_PENDING_INVITATIONS * 3; +export const PEER_MESH_MAX_ROUTE_HINTS = 16; + +export interface PeerMeshRosterV1 { + readonly version: 1; + readonly meshId: string; + readonly revision: number; + readonly members: readonly string[]; + readonly closed: boolean; +} + +export interface SignedPeerMeshRosterV1 { + readonly roster: PeerMeshRosterV1; + readonly authorityPublicKey: string; + readonly signature: string; +} + +export interface PeerMeshAuthorityTarget { + readonly peerId: string; + readonly routeHints: readonly string[]; + readonly coordinationRelays: readonly string[]; +} + +export interface PeerMeshInvitationV1 extends PeerMeshAuthorityTarget { + readonly version: 1; + readonly meshId: string; + readonly authorityPublicKey: string; + readonly secret: string; +} + +export interface PeerMeshAuthorityKeyPair { + readonly publicKey: string; + readonly privateKey: string; +} + +export function generatePeerMeshAuthorityKeyPair(): PeerMeshAuthorityKeyPair { + const { publicKey, privateKey } = generateKeyPairSync('ed25519'); + return Object.freeze({ + publicKey: publicKey.export({ format: 'der', type: 'spki' }).toString('base64url'), + privateKey: privateKey.export({ format: 'der', type: 'pkcs8' }).toString('base64url'), + }); +} + +export function peerMeshId(authorityPublicKey: string): string { + decodePublicKey(authorityPublicKey); + return `mesh_${createHash('sha256').update(authorityPublicKey).digest('base64url')}`; +} + +export function signPeerMeshRoster( + roster: PeerMeshRosterV1, + keys: PeerMeshAuthorityKeyPair, +): SignedPeerMeshRosterV1 { + const canonical = canonicalPeerMeshRoster(roster); + if (canonical.meshId !== peerMeshId(keys.publicKey)) { + throw new Error('Peer Mesh roster does not belong to the authority key'); + } + const privateKey = createPrivateKey({ + key: Buffer.from(keys.privateKey, 'base64url'), + format: 'der', + type: 'pkcs8', + }); + if (privateKey.asymmetricKeyType !== 'ed25519') { + throw new Error('Peer Mesh authority key must be Ed25519'); + } + return Object.freeze({ + roster: canonical, + authorityPublicKey: keys.publicKey, + signature: sign(null, encodeRoster(canonical), privateKey).toString('base64url'), + }); +} + +export function validatePeerMeshAuthorityKeyPair(keys: PeerMeshAuthorityKeyPair): void { + let privateKey; + try { + privateKey = createPrivateKey({ + key: decodeCanonicalBase64Url(keys.privateKey, 'authority private key'), + format: 'der', + type: 'pkcs8', + }); + } catch (error) { + throw new Error('Invalid Peer Mesh authority private key', { cause: error }); + } + if (privateKey.asymmetricKeyType !== 'ed25519') { + throw new Error('Peer Mesh authority key must be Ed25519'); + } + const derived = createPublicKey(privateKey).export({ format: 'der', type: 'spki' }); + if (derived.toString('base64url') !== keys.publicKey) { + throw new Error('Peer Mesh authority private key does not match its public key'); + } +} + +export function decodeSignedPeerMeshRoster(value: unknown): SignedPeerMeshRosterV1 { + const record = exactObject(value, 'signed Peer Mesh roster', [ + 'roster', + 'authorityPublicKey', + 'signature', + ]); + const authorityPublicKey = string(record.authorityPublicKey, 'authorityPublicKey', 256); + const roster = canonicalPeerMeshRoster(record.roster); + if (roster.meshId !== peerMeshId(authorityPublicKey)) { + throw new Error('Peer Mesh roster has the wrong authority'); + } + const signature = string(record.signature, 'signature', 128); + const signatureBytes = decodeCanonicalBase64Url(signature, 'roster signature'); + if (signatureBytes.length !== 64) throw new Error('Invalid Peer Mesh roster signature'); + const verified = verify( + null, + encodeRoster(roster), + decodePublicKey(authorityPublicKey), + signatureBytes, + ); + if (!verified) throw new Error('Peer Mesh roster signature is invalid'); + return Object.freeze({ roster, authorityPublicKey, signature }); +} + +export function createPeerMeshInvitationSecret(): string { + return randomBytes(32).toString('base64url'); +} + +export function peerMeshInvitationSecretDigest(secret: string): string { + return createHash('sha256').update(decodeSecret(secret)).digest('base64url'); +} + +export function matchesPeerMeshInvitationSecret(secret: string, expectedDigest: string): boolean { + const actual = Buffer.from(peerMeshInvitationSecretDigest(secret), 'base64url'); + const expected = Buffer.from(expectedDigest, 'base64url'); + return actual.length === expected.length && timingSafeEqual(actual, expected); +} + +export function decodePeerMeshInvitation(value: unknown): PeerMeshInvitationV1 { + const record = exactObject(value, 'Peer Mesh invitation', [ + 'version', + 'meshId', + 'authorityPublicKey', + 'secret', + 'peerId', + 'routeHints', + 'coordinationRelays', + ]); + if (record.version !== 1) throw new Error('Unsupported Peer Mesh invitation version'); + const authorityPublicKey = string(record.authorityPublicKey, 'authorityPublicKey', 256); + const meshId = string(record.meshId, 'meshId', 128); + if (meshId !== peerMeshId(authorityPublicKey)) { + throw new Error('Peer Mesh invitation has the wrong authority'); + } + return Object.freeze({ + version: 1, + meshId, + authorityPublicKey, + secret: validateSecret(record.secret), + ...decodeAuthorityTarget({ + peerId: record.peerId, + routeHints: record.routeHints, + coordinationRelays: record.coordinationRelays, + }), + }); +} + +export function canonicalPeerMeshRoster(value: unknown): PeerMeshRosterV1 { + const record = exactObject(value, 'Peer Mesh roster', [ + 'version', + 'meshId', + 'revision', + 'members', + 'closed', + ]); + if (record.version !== 1) throw new Error('Unsupported Peer Mesh roster version'); + const members = stringArray(record.members, 'members', PEER_MESH_MAX_MEMBERS, 256) + .map((member) => token(member, 'member', 256)) + .sort(); + if (members.length === 0 || new Set(members).size !== members.length) { + throw new Error('Peer Mesh roster members must be unique and non-empty'); + } + if (typeof record.closed !== 'boolean') throw new Error('Invalid Peer Mesh roster closed'); + return Object.freeze({ + version: 1, + meshId: string(record.meshId, 'meshId', 128), + revision: integer(record.revision, 'revision', 1), + members: Object.freeze(members), + closed: record.closed, + }); +} + +export function decodeAuthorityTarget(value: unknown): PeerMeshAuthorityTarget { + const record = exactObject(value, 'Peer Mesh authority target', [ + 'peerId', + 'routeHints', + 'coordinationRelays', + ]); + return Object.freeze({ + peerId: token(record.peerId, 'peerId', 256), + routeHints: Object.freeze(addressArray(record.routeHints, 'routeHints')), + coordinationRelays: Object.freeze( + addressArray(record.coordinationRelays, 'coordinationRelays'), + ), + }); +} + +function encodeRoster(roster: PeerMeshRosterV1): Buffer { + return Buffer.from( + `maka.peer-mesh.roster.v1\n${JSON.stringify({ + closed: roster.closed, + members: roster.members, + meshId: roster.meshId, + revision: roster.revision, + version: roster.version, + })}`, + ); +} + +function decodePublicKey(encoded: string) { + try { + const key = createPublicKey({ + key: decodeCanonicalBase64Url(encoded, 'authority public key'), + format: 'der', + type: 'spki', + }); + if (key.asymmetricKeyType !== 'ed25519') { + throw new Error('Peer Mesh authority key must be Ed25519'); + } + return key; + } catch (error) { + throw new Error('Invalid Peer Mesh authority public key', { cause: error }); + } +} + +function validateSecret(value: unknown): string { + const secret = string(value, 'secret', 64); + decodeSecret(secret); + return secret; +} + +function decodeSecret(secret: string): Buffer { + const bytes = decodeCanonicalBase64Url(secret, 'invitation secret'); + if (bytes.length !== 32) { + throw new Error('Invalid Peer Mesh invitation secret'); + } + return bytes; +} + +function decodeCanonicalBase64Url(value: string, label: string): Buffer { + const bytes = Buffer.from(value, 'base64url'); + if (bytes.toString('base64url') !== value) { + throw new Error(`Invalid Peer Mesh ${label}`); + } + return bytes; +} + +function object(value: unknown, label: string): Record { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new Error(`Invalid ${label}`); + } + return value as Record; +} + +function exactObject( + value: unknown, + label: string, + keys: readonly string[], +): Record { + const record = object(value, label); + if ( + Object.keys(record).length !== keys.length || + keys.some((key) => !Object.hasOwn(record, key)) + ) { + throw new Error(`Invalid ${label}`); + } + return record; +} + +function string(value: unknown, label: string, max: number): string { + if (typeof value !== 'string' || value.length === 0 || value.length > max) { + throw new Error(`Invalid Peer Mesh ${label}`); + } + return value; +} + +function token(value: unknown, label: string, max: number): string { + const result = string(value, label, max); + if (/\s|[\u0000-\u001f\u007f]/u.test(result)) { + throw new Error(`Invalid Peer Mesh ${label}`); + } + return result; +} + +function integer(value: unknown, label: string, minimum: number): number { + if (!Number.isSafeInteger(value) || (value as number) < minimum) { + throw new Error(`Invalid Peer Mesh ${label}`); + } + return value as number; +} + +function stringArray(value: unknown, label: string, maxItems: number, maxLength: number): string[] { + if (!Array.isArray(value) || value.length > maxItems) { + throw new Error(`Invalid Peer Mesh ${label}`); + } + return value.map((item) => string(item, label, maxLength)); +} + +function addressArray(value: unknown, label: string): string[] { + const addresses = stringArray(value, label, PEER_MESH_MAX_ROUTE_HINTS, 1024); + if ( + addresses.some( + (address) => !address.startsWith('/') || /\s|[\u0000-\u001f\u007f]/u.test(address), + ) || + new Set(addresses).size !== addresses.length + ) { + throw new Error(`Invalid Peer Mesh ${label}`); + } + return addresses; +} diff --git a/packages/runtime-host/src/peer-mesh/node.ts b/packages/runtime-host/src/peer-mesh/node.ts new file mode 100644 index 0000000000..76cac380cb --- /dev/null +++ b/packages/runtime-host/src/peer-mesh/node.ts @@ -0,0 +1,734 @@ +/* + * 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 type { RuntimeHostPeerNativeStream } from '../transport/peer-native.js'; +import { + canonicalPeerMeshRoster, + createPeerMeshInvitationSecret, + decodePeerMeshInvitation, + decodeSignedPeerMeshRoster, + generatePeerMeshAuthorityKeyPair, + matchesPeerMeshInvitationSecret, + PEER_MESH_MAX_MEMBERS, + PEER_MESH_MAX_MESHES, + PEER_MESH_MAX_INVITATION_RECORDS, + PEER_MESH_MAX_PENDING_INVITATIONS, + peerMeshId, + peerMeshInvitationSecretDigest, + signPeerMeshRoster, + type PeerMeshAuthorityTarget, + type PeerMeshInvitationV1, + type SignedPeerMeshRosterV1, +} from './model.js'; +import { + authorityKeys, + openPeerMeshStateStore, + type PeerMeshAuthorityStateV1, + type PeerMeshStateStore, + type PeerMeshStateV1, +} from './store.js'; + +const CONTROL_FRAME_MAX_BYTES = 64 * 1024; +const DEFAULT_INVITATION_TTL_MS = 15 * 60 * 1_000; +const CONNECT_DEADLINE_MS = 30_000; +const CONTROL_REQUEST_DEADLINE_MS = 10_000; +const MAX_ACTIVE_CONTROL_STREAMS = 32; +const MAX_ACTIVE_CONTROL_STREAMS_PER_PEER = 2; + +interface RedeemInvitationRequest { + readonly kind: 'redeem-invitation'; + readonly meshId: string; + readonly secret: string; +} + +type RedeemInvitationResponse = + | { + readonly kind: 'invitation-redeemed'; + readonly roster: SignedPeerMeshRosterV1; + } + | { + readonly kind: 'invitation-rejected'; + readonly reason: RedeemInvitationRejectionReason; + }; + +type RedeemInvitationRejectionReason = 'invalid' | 'expired' | 'closed' | 'full'; + +export interface PeerMeshNode { + status(): readonly PeerMeshStatus[]; + create(): Promise; + invite(meshId: string, input?: { readonly ttlMs?: number }): Promise; + join(invitation: PeerMeshInvitationV1, signal?: AbortSignal): Promise; + remove(meshId: string, peerId: string): Promise; + closeMesh(meshId: string): Promise; + serve(): Promise; + close(): Promise; +} + +export interface PeerMeshStatus { + readonly role: 'authority' | 'member'; + readonly localPeerId: string; + readonly authority: PeerMeshAuthorityTarget; + readonly roster: SignedPeerMeshRosterV1; + readonly pendingInvitationCount: number; +} + +export interface PeerMeshTransport { + identity(): Readonly<{ + peerId: string; + listenAddresses: readonly string[]; + coordinationRelays: readonly string[]; + }>; + connectMeshControl( + input: { + readonly peerId: string; + readonly routeHints: readonly string[]; + readonly coordinationRelays?: readonly string[]; + readonly directDeadlineMs: number; + }, + signal?: AbortSignal, + ): Promise; + serveMeshControl( + onStream: (stream: RuntimeHostPeerNativeStream) => void, + signal: AbortSignal, + ): Promise; +} + +export async function openPeerMeshNode(input: { + readonly dataRoot: string; + readonly peer: PeerMeshTransport; + readonly now?: () => number; +}): Promise { + const store = await openPeerMeshStateStore(input.dataRoot, input.peer.identity().peerId); + return new PeerMeshNodeImpl({ ...input, store }); +} + +class PeerMeshNodeImpl implements PeerMeshNode { + readonly #store: PeerMeshStateStore; + readonly #peer: PeerMeshTransport; + readonly #now: () => number; + readonly #activeControlStreams = new Set(); + readonly #lifetime = new AbortController(); + #admissionTail = Promise.resolve(); + #serveTask: Promise | undefined; + #closeTask: Promise | undefined; + + constructor(input: { + readonly store: PeerMeshStateStore; + readonly peer: PeerMeshTransport; + readonly now?: () => number; + }) { + this.#store = input.store; + this.#peer = input.peer; + this.#now = input.now ?? Date.now; + } + + status(): readonly PeerMeshStatus[] { + this.#assertOpen(); + const identity = this.#peer.identity(); + return Object.freeze( + this.#store + .read() + .filter( + (state) => + state.role === 'authority' || state.roster.roster.members.includes(identity.peerId), + ) + .map((state) => peerMeshStatus(state, identity)), + ); + } + + create(): Promise { + return this.#admitMesh(async () => { + const identity = this.#peer.identity(); + const state = await this.#store.mutate((current) => { + assertMeshCapacity(current, identity.peerId); + const keys = generatePeerMeshAuthorityKeyPair(); + const roster = signPeerMeshRoster( + canonicalPeerMeshRoster({ + version: 1, + meshId: peerMeshId(keys.publicKey), + revision: 1, + members: [identity.peerId], + closed: false, + }), + keys, + ); + const state: PeerMeshStateV1 = { + role: 'authority', + roster, + authorityPrivateKey: keys.privateKey, + invitations: [], + }; + return { state: appendMesh(current, state, identity.peerId), result: state }; + }); + return peerMeshStatus(state, identity); + }); + } + + invite(meshId: string, input: { readonly ttlMs?: number } = {}): Promise { + if (this.#lifetime.signal.aborted) return Promise.reject(new Error('Peer Mesh node is closed')); + const now = this.#now(); + const identity = this.#peer.identity(); + const ttlMs = input.ttlMs ?? DEFAULT_INVITATION_TTL_MS; + if (!Number.isSafeInteger(ttlMs) || ttlMs < 1_000 || ttlMs > 24 * 60 * 60 * 1_000) { + return Promise.reject( + new Error('Peer Mesh invitation TTL must be between 1 second and 1 day'), + ); + } + return this.#store.mutate((current) => { + const state = requireAuthority(current, meshId); + if (state.roster.roster.closed) throw new Error('Peer Mesh is closed'); + const invitations = state.invitations.filter( + (invitation) => invitation.status === 'redeemed' || invitation.expiresAt > now, + ); + if ( + invitations.filter(({ status }) => status === 'pending').length >= + PEER_MESH_MAX_PENDING_INVITATIONS + ) + throw new Error('Peer Mesh has too many pending invitations'); + if (invitations.length >= PEER_MESH_MAX_INVITATION_RECORDS) + throw new Error('Peer Mesh has too many recent invitations'); + const secret = createPeerMeshInvitationSecret(); + const expiresAt = now + ttlMs; + const target = authorityTarget(identity); + const invitation: PeerMeshInvitationV1 = { + version: 1, + meshId: state.roster.roster.meshId, + authorityPublicKey: state.roster.authorityPublicKey, + secret, + ...target, + }; + return { + state: replaceMesh(current, { + ...state, + invitations: [ + ...invitations, + { + status: 'pending', + secretDigest: peerMeshInvitationSecretDigest(secret), + expiresAt, + }, + ], + }), + result: Object.freeze(invitation), + }; + }); + } + + join(invitationValue: PeerMeshInvitationV1, signal?: AbortSignal): Promise { + return this.#admitMesh(async () => { + const invitation = decodePeerMeshInvitation(invitationValue); + const current = this.#store.read(); + const existing = findMesh(current, invitation.meshId); + const localPeerId = this.#peer.identity().peerId; + if (existing?.role === 'authority') { + throw new Error('This peer already belongs to that Peer Mesh'); + } + if (!existing) assertMeshCapacity(current, localPeerId); + const operationSignal = signal + ? AbortSignal.any([signal, this.#lifetime.signal]) + : this.#lifetime.signal; + const stream = await this.#peer.connectMeshControl( + { + peerId: invitation.peerId, + routeHints: invitation.routeHints, + coordinationRelays: invitation.coordinationRelays, + directDeadlineMs: CONNECT_DEADLINE_MS, + }, + operationSignal, + ); + try { + const request: RedeemInvitationRequest = { + kind: 'redeem-invitation', + meshId: invitation.meshId, + secret: invitation.secret, + }; + const response = await exchangeControl(stream, request, operationSignal); + if (response.kind === 'invitation-rejected') { + throw new Error(`Peer Mesh invitation was rejected: ${response.reason}`); + } + const roster = decodeSignedPeerMeshRoster(response.roster); + const identity = this.#peer.identity(); + if ( + roster.roster.meshId !== invitation.meshId || + roster.authorityPublicKey !== invitation.authorityPublicKey || + !roster.roster.members.includes(identity.peerId) + ) { + throw new Error('Peer Mesh authority returned an unrelated roster'); + } + const state: PeerMeshStateV1 = { + role: 'replica', + authority: { + peerId: invitation.peerId, + routeHints: invitation.routeHints, + coordinationRelays: invitation.coordinationRelays, + }, + roster, + }; + const joined = await this.#store.mutate((current) => { + const existing = findMesh(current, invitation.meshId); + if (existing?.role === 'authority') { + throw new Error('This peer already belongs to that Peer Mesh'); + } + if ( + existing && + (existing.roster.authorityPublicKey !== roster.authorityPublicKey || + roster.roster.revision <= existing.roster.roster.revision) + ) { + throw new Error('Peer Mesh invitation did not advance the existing membership'); + } + if (!existing) assertMeshCapacity(current, identity.peerId); + return { + state: existing + ? replaceMesh(current, state) + : appendMesh(current, state, identity.peerId), + result: state, + }; + }); + return peerMeshStatus(joined, identity); + } finally { + await stream.close().catch(() => undefined); + } + }); + } + + remove(meshId: string, peerId: string): Promise { + if (this.#lifetime.signal.aborted) return Promise.reject(new Error('Peer Mesh node is closed')); + return this.#updateAuthorityRoster(meshId, false, (state) => { + if (peerId === this.#peer.identity().peerId) { + throw new Error('Peer Mesh authority cannot remove itself'); + } + const members = state.roster.roster.members.filter((member) => member !== peerId); + if (members.length === state.roster.roster.members.length) { + throw new Error('Peer is not a member of this Peer Mesh'); + } + return { members, closed: false }; + }); + } + + closeMesh(meshId: string): Promise { + if (this.#lifetime.signal.aborted) return Promise.reject(new Error('Peer Mesh node is closed')); + return this.#updateAuthorityRoster(meshId, true, (state) => ({ + members: state.roster.roster.members, + closed: true, + })); + } + + async serve(): Promise { + if (this.#lifetime.signal.aborted) throw new Error('Peer Mesh node is closed'); + if (this.#serveTask) throw new Error('Peer Mesh node is already serving'); + const serving = this.#peer.serveMeshControl( + (stream) => this.#acceptIncoming(stream), + this.#lifetime.signal, + ); + this.#serveTask = serving; + try { + await serving; + if (!this.#lifetime.signal.aborted) + throw new Error('Peer Mesh control transport stopped unexpectedly'); + } finally { + if (this.#serveTask === serving) { + this.#serveTask = undefined; + } + } + } + + close(): Promise { + this.#closeTask ??= this.#close(); + return this.#closeTask; + } + + async #close(): Promise { + this.#lifetime.abort(); + await this.#serveTask?.catch(() => undefined); + for (const stream of this.#activeControlStreams) stream.abort(); + this.#activeControlStreams.clear(); + await this.#admissionTail; + return this.#store.close(); + } + + #assertOpen(): void { + if (this.#lifetime.signal.aborted) throw new Error('Peer Mesh node is closed'); + } + + #updateAuthorityRoster( + meshId: string, + closedIsSuccess: boolean, + update: (state: PeerMeshAuthorityStateV1) => { + readonly members: readonly string[]; + readonly closed: boolean; + }, + ): Promise { + return this.#store.mutate((current) => { + const state = requireAuthority(current, meshId); + if (state.roster.roster.closed) { + if (closedIsSuccess) { + return { state: current, result: peerMeshStatus(state, this.#peer.identity()) }; + } + throw new Error('Peer Mesh is closed'); + } + const next = update(state); + const roster = signPeerMeshRoster( + { + version: 1, + meshId: state.roster.roster.meshId, + revision: state.roster.roster.revision + 1, + members: next.members, + closed: next.closed, + }, + authorityKeys(state), + ); + const updated = { + ...state, + roster, + invitations: next.closed + ? state.invitations.filter(({ status }) => status === 'redeemed') + : state.invitations.filter( + (invitation) => + invitation.status === 'pending' || next.members.includes(invitation.peerId), + ), + }; + return { + state: replaceMesh(current, updated), + result: peerMeshStatus(updated, this.#peer.identity()), + }; + }); + } + + #acceptIncoming(stream: RuntimeHostPeerNativeStream): void { + let peerStreams = 0; + for (const active of this.#activeControlStreams) { + if (active.peerId === stream.peerId) peerStreams += 1; + } + if ( + this.#lifetime.signal.aborted || + this.#activeControlStreams.size >= MAX_ACTIVE_CONTROL_STREAMS || + peerStreams >= MAX_ACTIVE_CONTROL_STREAMS_PER_PEER + ) { + stream.abort(); + return; + } + this.#activeControlStreams.add(stream); + void this.#handleIncoming(stream).finally(() => { + this.#activeControlStreams.delete(stream); + }); + } + + #admitMesh(operation: () => Promise): Promise { + if (this.#lifetime.signal.aborted) return Promise.reject(new Error('Peer Mesh node is closed')); + const task = this.#admissionTail.then(() => { + if (this.#lifetime.signal.aborted) throw new Error('Peer Mesh node is closed'); + return operation(); + }); + this.#admissionTail = task.then( + () => undefined, + () => undefined, + ); + return task; + } + + async #handleIncoming(stream: RuntimeHostPeerNativeStream): Promise { + const deadline = setTimeout(() => stream.abort(), CONTROL_REQUEST_DEADLINE_MS); + try { + const request = decodeRedeemRequest(await readFrame(stream)); + const response = await this.#redeem(request, stream.peerId); + await writeFrame(stream, response); + await stream.close(); + } catch { + stream.abort(); + } finally { + clearTimeout(deadline); + } + } + + #redeem( + request: RedeemInvitationRequest, + remotePeerId: string, + ): Promise { + const now = this.#now(); + return this.#store.mutate((current) => { + const state = findMesh(current, request.meshId); + if (!state || state.role !== 'authority') + return { state: current, result: rejected('invalid') }; + const invitation = state.invitations.find(({ secretDigest }) => + matchesPeerMeshInvitationSecret(request.secret, secretDigest), + ); + if (request.meshId !== state.roster.roster.meshId || !invitation) { + return { state: current, result: rejected('invalid') }; + } + if (invitation.status === 'redeemed') { + if ( + invitation.peerId !== remotePeerId || + !state.roster.roster.members.includes(remotePeerId) + ) { + return { state: current, result: rejected('invalid') }; + } + return { + state: current, + result: { kind: 'invitation-redeemed', roster: state.roster }, + }; + } + const remaining = state.invitations.filter( + (record) => + record !== invitation && (record.status === 'redeemed' || record.expiresAt > now), + ); + if (invitation.expiresAt <= now) { + return { + state: replaceMesh(current, { ...state, invitations: remaining }), + result: rejected('expired'), + }; + } + if (state.roster.roster.closed) { + return { + state: replaceMesh(current, { ...state, invitations: remaining }), + result: rejected('closed'), + }; + } + if ( + !state.roster.roster.members.includes(remotePeerId) && + state.roster.roster.members.length >= PEER_MESH_MAX_MEMBERS + ) { + return { + state: replaceMesh(current, { ...state, invitations: remaining }), + result: rejected('full'), + }; + } + if (state.roster.roster.members.includes(remotePeerId)) { + return { + state: replaceMesh(current, { + ...state, + invitations: [ + ...remaining.filter( + (record) => record.status === 'pending' || record.peerId !== remotePeerId, + ), + redeemedInvitation(invitation, remotePeerId), + ], + }), + result: { kind: 'invitation-redeemed', roster: state.roster }, + }; + } + const members = [...state.roster.roster.members, remotePeerId].sort(); + const roster = signPeerMeshRoster( + { + ...state.roster.roster, + revision: state.roster.roster.revision + 1, + members, + }, + authorityKeys(state), + ); + return { + state: replaceMesh(current, { + ...state, + roster, + invitations: [ + ...remaining.filter( + (record) => record.status === 'pending' || record.peerId !== remotePeerId, + ), + redeemedInvitation(invitation, remotePeerId), + ], + }), + result: { kind: 'invitation-redeemed', roster }, + }; + }); + } +} + +function peerMeshStatus( + state: PeerMeshStateV1, + identity: ReturnType, +): PeerMeshStatus { + return Object.freeze({ + role: state.role === 'authority' ? 'authority' : 'member', + localPeerId: identity.peerId, + authority: state.role === 'authority' ? authorityTarget(identity) : state.authority, + roster: state.roster, + pendingInvitationCount: + state.role === 'authority' + ? state.invitations.filter(({ status }) => status === 'pending').length + : 0, + }); +} + +function requireAuthority( + states: readonly PeerMeshStateV1[], + meshId: string, +): PeerMeshAuthorityStateV1 { + const state = findMesh(states, meshId); + if (!state || state.role !== 'authority') + throw new Error('Peer Mesh operation requires authority'); + return state; +} + +function findMesh(states: readonly PeerMeshStateV1[], meshId: string): PeerMeshStateV1 | undefined { + return states.find(({ roster }) => roster.roster.meshId === meshId); +} + +function replaceMesh( + states: readonly PeerMeshStateV1[], + next: PeerMeshStateV1, +): readonly PeerMeshStateV1[] { + return states.map((state) => + state.roster.roster.meshId === next.roster.roster.meshId ? next : state, + ); +} + +function rejected(reason: RedeemInvitationRejectionReason) { + return { kind: 'invitation-rejected', reason } as const; +} + +function redeemedInvitation(invitation: { readonly secretDigest: string }, peerId: string) { + return { + status: 'redeemed' as const, + secretDigest: invitation.secretDigest, + peerId, + }; +} + +function authorityTarget( + identity: ReturnType, +): PeerMeshAuthorityTarget { + return Object.freeze({ + peerId: identity.peerId, + routeHints: identity.listenAddresses, + coordinationRelays: identity.coordinationRelays, + }); +} + +function assertMeshCapacity(states: readonly PeerMeshStateV1[], localPeerId: string): void { + if ( + states.filter((state) => isActiveMembership(state, localPeerId)).length >= PEER_MESH_MAX_MESHES + ) { + throw new Error('This peer belongs to too many Peer Meshes'); + } +} + +function appendMesh( + states: readonly PeerMeshStateV1[], + state: PeerMeshStateV1, + localPeerId: string, +): readonly PeerMeshStateV1[] { + if (states.length < PEER_MESH_MAX_MESHES) return [...states, state]; + const retired = states.findIndex((candidate) => !isActiveMembership(candidate, localPeerId)); + if (retired < 0) throw new Error('This peer belongs to too many Peer Meshes'); + return [...states.slice(0, retired), ...states.slice(retired + 1), state]; +} + +function isActiveMembership(state: PeerMeshStateV1, localPeerId: string): boolean { + return !state.roster.roster.closed && state.roster.roster.members.includes(localPeerId); +} + +async function exchangeControl( + stream: RuntimeHostPeerNativeStream, + request: RedeemInvitationRequest, + signal?: AbortSignal, +): Promise { + const timeout = AbortSignal.timeout(CONTROL_REQUEST_DEADLINE_MS); + const operationSignal = signal ? AbortSignal.any([signal, timeout]) : timeout; + const abort = () => stream.abort(); + operationSignal.addEventListener('abort', abort, { once: true }); + if (operationSignal.aborted) abort(); + try { + operationSignal.throwIfAborted(); + await writeFrame(stream, request); + const response = decodeRedeemResponse(await readFrame(stream)); + operationSignal.throwIfAborted(); + return response; + } catch (error) { + operationSignal.throwIfAborted(); + throw error; + } finally { + operationSignal.removeEventListener('abort', abort); + } +} + +function decodeRedeemRequest(value: unknown): RedeemInvitationRequest { + const record = recordValue(value); + if (record.kind !== 'redeem-invitation' || !hasExactKeys(record, ['kind', 'meshId', 'secret'])) { + throw new Error('Unsupported Peer Mesh control request'); + } + return { + kind: 'redeem-invitation', + meshId: requiredString(record.meshId, 128), + secret: requiredString(record.secret, 64), + }; +} + +function decodeRedeemResponse(value: unknown): RedeemInvitationResponse { + const record = recordValue(value); + if (record.kind === 'invitation-redeemed' && hasExactKeys(record, ['kind', 'roster'])) { + return { + kind: 'invitation-redeemed', + roster: decodeSignedPeerMeshRoster(record.roster), + }; + } + if ( + record.kind === 'invitation-rejected' && + hasExactKeys(record, ['kind', 'reason']) && + (record.reason === 'invalid' || + record.reason === 'expired' || + record.reason === 'closed' || + record.reason === 'full') + ) { + return { kind: 'invitation-rejected', reason: record.reason }; + } + throw new Error('Invalid Peer Mesh control response'); +} + +async function writeFrame(stream: RuntimeHostPeerNativeStream, value: unknown): Promise { + const bytes = Buffer.from(`${JSON.stringify(value)}\n`); + if (bytes.length > CONTROL_FRAME_MAX_BYTES) + throw new Error('Peer Mesh control frame is too large'); + await stream.write(bytes); +} + +async function readFrame(stream: RuntimeHostPeerNativeStream): Promise { + let buffered = Buffer.alloc(0); + for (;;) { + const chunk = await stream.read(); + if (!chunk) throw new Error('Peer Mesh control stream ended before a frame arrived'); + buffered = Buffer.concat([buffered, chunk]); + if (buffered.length > CONTROL_FRAME_MAX_BYTES) + throw new Error('Peer Mesh control frame is too large'); + const newline = buffered.indexOf(0x0a); + if (newline < 0) continue; + if (buffered.subarray(newline + 1).some((byte) => byte > 0x20)) { + throw new Error('Peer Mesh control stream contained multiple frames'); + } + return JSON.parse(buffered.subarray(0, newline).toString('utf8')) as unknown; + } +} + +function recordValue(value: unknown): Record { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new Error('Invalid Peer Mesh control frame'); + } + return value as Record; +} + +function hasExactKeys(record: Record, keys: readonly string[]): boolean { + return ( + Object.keys(record).length === keys.length && keys.every((key) => Object.hasOwn(record, key)) + ); +} + +function requiredString(value: unknown, maxLength: number): string { + if (typeof value !== 'string' || value.length === 0 || value.length > maxLength) { + throw new Error('Invalid Peer Mesh control value'); + } + return value; +} diff --git a/packages/runtime-host/src/peer-mesh/store.ts b/packages/runtime-host/src/peer-mesh/store.ts new file mode 100644 index 0000000000..4e01171259 --- /dev/null +++ b/packages/runtime-host/src/peer-mesh/store.ts @@ -0,0 +1,425 @@ +/* + * 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 { chmod, lstat, mkdir, open, readFile, rename, unlink } from 'node:fs/promises'; +import { dirname, join } from 'node:path'; +import { + acquireFileLifetimeOwner, + type FileLifetimeOwner, +} from '@maka/storage/file-lifetime-owner'; +import { + decodeAuthorityTarget, + decodeSignedPeerMeshRoster, + PEER_MESH_MAX_INVITATION_RECORDS, + PEER_MESH_MAX_MESHES, + PEER_MESH_MAX_PENDING_INVITATIONS, + type PeerMeshAuthorityKeyPair, + type PeerMeshAuthorityTarget, + type SignedPeerMeshRosterV1, + validatePeerMeshAuthorityKeyPair, +} from './model.js'; + +const STATE_FILE = 'peer-mesh.json'; +const LOCK_FILE = 'peer-mesh.owner'; +const MAX_STATE_BYTES = 1024 * 1024; + +export type PeerMeshInvitationRecord = PendingPeerMeshInvitation | RedeemedPeerMeshInvitation; + +interface PendingPeerMeshInvitation { + readonly status: 'pending'; + readonly secretDigest: string; + readonly expiresAt: number; +} + +interface RedeemedPeerMeshInvitation { + readonly status: 'redeemed'; + readonly secretDigest: string; + readonly peerId: string; +} + +interface PeerMeshStateBase { + readonly roster: SignedPeerMeshRosterV1; +} + +export interface PeerMeshAuthorityStateV1 extends PeerMeshStateBase { + readonly role: 'authority'; + readonly authorityPrivateKey: string; + readonly invitations: readonly PeerMeshInvitationRecord[]; +} + +export interface PeerMeshReplicaStateV1 extends PeerMeshStateBase { + readonly role: 'replica'; + readonly authority: PeerMeshAuthorityTarget; +} + +export type PeerMeshStateV1 = PeerMeshAuthorityStateV1 | PeerMeshReplicaStateV1; + +export interface PeerMeshStateStore { + read(): readonly PeerMeshStateV1[]; + mutate( + operation: (state: readonly PeerMeshStateV1[]) => { + readonly state: readonly PeerMeshStateV1[]; + readonly result: T; + }, + ): Promise; + close(): Promise; +} + +export async function openPeerMeshStateStore( + dataRoot: string, + localPeerId: string, +): Promise { + await mkdir(dataRoot, { recursive: true, mode: 0o700 }); + if (process.platform !== 'win32') await chmod(dataRoot, 0o700); + const owner = await acquireFileLifetimeOwner(join(dataRoot, LOCK_FILE)); + try { + const state = await readState(join(dataRoot, STATE_FILE), localPeerId); + return new PeerMeshStateStoreImpl(dataRoot, localPeerId, owner, state); + } catch (error) { + await owner.close(); + throw error; + } +} + +class PeerMeshStateStoreImpl implements PeerMeshStateStore { + readonly #path: string; + #state: readonly PeerMeshStateV1[]; + #tail = Promise.resolve(); + #failure: Error | undefined; + #closeTask: Promise | undefined; + #closed = false; + + constructor( + dataRoot: string, + private readonly localPeerId: string, + private readonly owner: FileLifetimeOwner, + state: readonly PeerMeshStateV1[], + ) { + this.#path = join(dataRoot, STATE_FILE); + this.#state = state; + } + + read(): readonly PeerMeshStateV1[] { + this.#assertOpen(); + return this.#state; + } + + mutate( + operation: (state: readonly PeerMeshStateV1[]) => { + readonly state: readonly PeerMeshStateV1[]; + readonly result: T; + }, + ): Promise { + this.#assertOpen(); + const task = this.#tail.then(async () => { + if (this.#failure) throw this.#failure; + const updated = operation(this.#state); + if (updated.state === this.#state) return updated.result; + const canonical = decodePeerMeshStates(updated.state, this.localPeerId); + assertStateAdvance(this.#state, canonical, this.localPeerId); + try { + await writeState(this.#path, this.localPeerId, canonical); + this.#state = canonical; + } catch (error) { + if (error instanceof PeerMeshPostCommitError) { + this.#state = canonical; + this.#failure = error; + } + throw error; + } + return updated.result; + }); + this.#tail = task.then( + () => undefined, + () => undefined, + ); + return task; + } + + close(): Promise { + this.#closeTask ??= this.#close(); + return this.#closeTask; + } + + async #close(): Promise { + this.#closed = true; + await this.#tail; + await this.owner.close(); + } + + #assertOpen(): void { + if (this.#closed) throw new Error('Peer Mesh state store is closed'); + if (this.#failure) throw this.#failure; + } +} + +export function decodePeerMeshState(value: unknown, localPeerId: string): PeerMeshStateV1 { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new Error('Invalid Peer Mesh state'); + } + const record = value as Record; + const expectedKeys = + record.role === 'authority' + ? ['role', 'roster', 'authorityPrivateKey', 'invitations'] + : ['role', 'roster', 'authority']; + if ( + Object.keys(record).length !== expectedKeys.length || + expectedKeys.some((key) => !Object.hasOwn(record, key)) + ) { + throw new Error('Invalid Peer Mesh state'); + } + if (record.role !== 'authority' && record.role !== 'replica') { + throw new Error('Unsupported Peer Mesh state'); + } + const roster = decodeSignedPeerMeshRoster(record.roster); + if (record.role === 'authority') { + if (!roster.roster.members.includes(localPeerId)) { + throw new Error('Peer Mesh authority is not present in its roster'); + } + const privateKey = boundedString(record.authorityPrivateKey, 'authorityPrivateKey', 256); + validatePeerMeshAuthorityKeyPair({ + publicKey: roster.authorityPublicKey, + privateKey, + }); + return Object.freeze({ + role: 'authority', + roster, + authorityPrivateKey: privateKey, + invitations: Object.freeze(decodeInvitations(record.invitations)), + }); + } + const authority = decodeAuthorityTarget(record.authority); + if (!roster.roster.members.includes(authority.peerId)) { + throw new Error('Peer Mesh authority is not present in its roster'); + } + return Object.freeze({ + role: 'replica', + authority, + roster, + }); +} + +function decodePeerMeshStates(value: unknown, localPeerId: string): readonly PeerMeshStateV1[] { + if (!Array.isArray(value) || value.length > PEER_MESH_MAX_MESHES) { + throw new Error('Invalid Peer Mesh state collection'); + } + const states = value.map((state) => decodePeerMeshState(state, localPeerId)); + const meshIds = states.map(({ roster }) => roster.roster.meshId); + if (new Set(meshIds).size !== meshIds.length) { + throw new Error('Duplicate Peer Mesh state'); + } + return Object.freeze(states); +} + +function assertStateAdvance( + current: readonly PeerMeshStateV1[], + next: readonly PeerMeshStateV1[], + localPeerId: string, +): void { + for (const previous of current) { + const updated = next.find( + ({ roster }) => roster.roster.meshId === previous.roster.roster.meshId, + ); + if (!updated) { + if (isRetired(previous, localPeerId)) continue; + throw new Error('Active Peer Mesh state cannot be removed implicitly'); + } + if (updated.role !== previous.role) { + throw new Error('Peer Mesh state identity cannot change'); + } + if (updated.roster.roster.revision < previous.roster.roster.revision) { + throw new Error('Peer Mesh roster revision cannot roll back'); + } + if ( + updated.roster.roster.revision === previous.roster.roster.revision && + JSON.stringify(updated.roster) !== JSON.stringify(previous.roster) + ) { + throw new Error('Peer Mesh roster revision cannot identify different facts'); + } + if (previous.roster.roster.closed && !updated.roster.roster.closed) { + throw new Error('Closed Peer Mesh state is terminal'); + } + } +} + +function isRetired(state: PeerMeshStateV1, localPeerId: string): boolean { + return ( + state.roster.roster.closed || + (state.role === 'replica' && !state.roster.roster.members.includes(localPeerId)) + ); +} + +export function authorityKeys(state: PeerMeshStateV1): PeerMeshAuthorityKeyPair { + if (state.role !== 'authority') { + throw new Error('Peer Mesh operation requires the authority'); + } + return Object.freeze({ + publicKey: state.roster.authorityPublicKey, + privateKey: state.authorityPrivateKey, + }); +} + +async function readState( + path: string, + expectedLocalPeerId: string, +): Promise { + try { + const info = await lstat(path); + if (!info.isFile() || info.size > MAX_STATE_BYTES) + throw new Error('Invalid Peer Mesh state file'); + const document = JSON.parse(await readFile(path, 'utf8')) as unknown; + if (!document || typeof document !== 'object' || Array.isArray(document)) { + throw new Error('Invalid Peer Mesh state document'); + } + const record = document as Record; + if ( + record.version !== 1 || + Object.keys(record).length !== 3 || + !Object.hasOwn(record, 'localPeerId') || + !Object.hasOwn(record, 'meshes') + ) { + throw new Error('Unsupported Peer Mesh state document'); + } + if (boundedString(record.localPeerId, 'localPeerId', 256) !== expectedLocalPeerId) { + throw new Error('Peer Mesh state belongs to a different peer identity'); + } + return decodePeerMeshStates(record.meshes, expectedLocalPeerId); + } catch (error) { + if (isNodeError(error, 'ENOENT')) return Object.freeze([]); + throw error; + } +} + +async function writeState( + path: string, + localPeerId: string, + state: readonly PeerMeshStateV1[], +): Promise { + const document = `${JSON.stringify({ version: 1, localPeerId, meshes: state }, null, 2)}\n`; + if (Buffer.byteLength(document) > MAX_STATE_BYTES) + throw new Error('Peer Mesh state is too large'); + const temporary = `${path}.tmp`; + let replaced = false; + try { + await unlink(temporary).catch((error: unknown) => { + if (!isNodeError(error, 'ENOENT')) throw error; + }); + const handle = await open(temporary, 'wx', 0o600); + try { + await handle.writeFile(document, 'utf8'); + await handle.sync(); + } finally { + await handle.close(); + } + if (process.platform !== 'win32') await chmod(temporary, 0o600); + await rename(temporary, path); + replaced = true; + try { + await syncDirectory(dirname(path)); + } catch (error) { + throw new PeerMeshPostCommitError(error); + } + } finally { + if (!replaced) await unlink(temporary).catch(() => undefined); + } +} + +class PeerMeshPostCommitError extends Error { + constructor(cause: unknown) { + super('Peer Mesh state was replaced but its durability could not be confirmed; reopen it', { + cause, + }); + } +} + +function decodeInvitations(value: unknown): PeerMeshInvitationRecord[] { + if (!Array.isArray(value) || value.length > PEER_MESH_MAX_INVITATION_RECORDS) { + throw new Error('Invalid Peer Mesh invitations'); + } + const invitations = value.map((entry) => { + if (!entry || typeof entry !== 'object' || Array.isArray(entry)) { + throw new Error('Invalid Peer Mesh invitation'); + } + const record = entry as Record; + const base = { secretDigest: decodeSecretDigest(record.secretDigest) }; + if ( + record.status === 'pending' && + Object.keys(record).length === 3 && + Object.hasOwn(record, 'expiresAt') + ) { + const expiresAt = record.expiresAt; + if (!Number.isSafeInteger(expiresAt) || (expiresAt as number) < 1) { + throw new Error('Invalid Peer Mesh invitation expiry'); + } + return Object.freeze({ status: 'pending' as const, ...base, expiresAt: expiresAt as number }); + } + if ( + record.status === 'redeemed' && + Object.keys(record).length === 3 && + Object.hasOwn(record, 'peerId') + ) { + return Object.freeze({ + status: 'redeemed' as const, + ...base, + peerId: boundedString(record.peerId, 'peerId', 256), + }); + } + throw new Error('Invalid Peer Mesh invitation'); + }); + if (new Set(invitations.map(({ secretDigest }) => secretDigest)).size !== invitations.length) { + throw new Error('Duplicate Peer Mesh invitation'); + } + if ( + invitations.filter(({ status }) => status === 'pending').length > + PEER_MESH_MAX_PENDING_INVITATIONS + ) { + throw new Error('Too many pending Peer Mesh invitations'); + } + return invitations; +} + +function boundedString(value: unknown, label: string, max: number): string { + if (typeof value !== 'string' || value.length === 0 || value.length > max) { + throw new Error(`Invalid Peer Mesh ${label}`); + } + return value; +} + +function decodeSecretDigest(value: unknown): string { + const digest = boundedString(value, 'secretDigest', 64); + const bytes = Buffer.from(digest, 'base64url'); + if (bytes.length !== 32 || bytes.toString('base64url') !== digest) { + throw new Error('Invalid Peer Mesh secretDigest'); + } + return digest; +} + +async function syncDirectory(path: string): Promise { + if (process.platform === 'win32') return; + const handle = await open(path, 'r'); + try { + await handle.sync(); + } finally { + await handle.close(); + } +} + +function isNodeError(error: unknown, code: string): error is NodeJS.ErrnoException { + return error instanceof Error && 'code' in error && error.code === code; +} diff --git a/packages/runtime-host/src/server/peer-listener.ts b/packages/runtime-host/src/server/peer-listener.ts index 419d32db54..c509157bd8 100644 --- a/packages/runtime-host/src/server/peer-listener.ts +++ b/packages/runtime-host/src/server/peer-listener.ts @@ -34,6 +34,8 @@ import type { } from './listener-set.js'; const MAX_PENDING_AUTHENTICATIONS = 16; +const MAX_ACTIVE_STREAMS = 64; +const MAX_ACTIVE_STREAMS_PER_PEER = 4; export interface StartRuntimeHostPeerListenerOptions { readonly nativePath: string; @@ -69,8 +71,9 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { readonly #accessAuthority: RuntimeHostAccessAuthority; readonly #accept: (connection: RuntimeHostListenerConnection) => void; readonly #transports = new Set(); + readonly #streams = new Set(); readonly #authentications = new Map>(); - readonly #acceptTask: Promise; + readonly #acceptTasks: readonly Promise[]; #acceptFailure: unknown; #admitting = true; #closeAdmissionTask: Promise | undefined; @@ -87,9 +90,13 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { this.#endpoint = endpoint; this.#accessAuthority = accessAuthority; this.#accept = accept; - this.#acceptTask = this.#acceptStreams().catch((error: unknown) => { - this.#acceptFailure = error; - }); + const captureFailure = (error: unknown) => { + this.#acceptFailure ??= error; + }; + this.#acceptTasks = [ + this.#acceptStreams().catch(captureFailure), + this.#discardMeshStreams().catch(captureFailure), + ]; } closeAdmission(): Promise { @@ -106,7 +113,7 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { await this.closeAdmission(); for (const transport of this.#transports) transport.abort(); await this.#endpoint.close(); - await this.#acceptTask; + await Promise.all(this.#acceptTasks); if (this.#acceptFailure) throw this.#acceptFailure; })(); return this.#cleanupTask; @@ -130,6 +137,15 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { stream.abort(); continue; } + let peerStreams = 0; + for (const admitted of this.#streams) { + if (admitted.peerId === stream.peerId) peerStreams += 1; + } + if (this.#streams.size >= MAX_ACTIVE_STREAMS || peerStreams >= MAX_ACTIVE_STREAMS_PER_PEER) { + stream.abort(); + continue; + } + this.#streams.add(stream); const task = this.#authenticateAndAccept(stream).finally(() => { this.#authentications.delete(stream); }); @@ -138,7 +154,21 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { } } + async #discardMeshStreams(): Promise { + while (true) { + try { + const stream = await this.#endpoint.acceptMeshControl(); + if (!stream) return; + stream.abort(); + } catch (error) { + if (this.#cleanupTask) return; + throw error; + } + } + } + async #authenticateAndAccept(stream: RuntimeHostPeerNativeStream): Promise { + let transportOwnsStream = false; try { const authenticated = await withDeadline( readRuntimeHostPeerAuthentication(stream), @@ -169,7 +199,11 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { new RuntimeHostPeerByteStream(stream, authenticated.remainder), ); this.#transports.add(transport); - void transport.closed.then(() => this.#transports.delete(transport)); + transportOwnsStream = true; + void transport.closed.then(() => { + this.#transports.delete(transport); + this.#streams.delete(stream); + }); try { this.#accept({ transport, authority: admittedAuthority }); } catch (error) { @@ -177,6 +211,8 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { } } catch { stream.abort(); + } finally { + if (!transportOwnsStream) this.#streams.delete(stream); } } } diff --git a/packages/runtime-host/src/transport/peer-native.ts b/packages/runtime-host/src/transport/peer-native.ts index a517b41ee3..aab2e1abd9 100644 --- a/packages/runtime-host/src/transport/peer-native.ts +++ b/packages/runtime-host/src/transport/peer-native.ts @@ -29,6 +29,7 @@ const require = createRequire(import.meta.url); export type RuntimeHostPeerErrorCode = | 'peer_identity_mismatch' | 'direct_path_unavailable' + | 'mesh_control_unavailable' | 'coordination_unavailable' | 'peer_native_unavailable' | 'peer_native_failed' @@ -46,6 +47,7 @@ export class RuntimeHostPeerError extends Error { } export interface RuntimeHostPeerNativeStream { + readonly peerId: string; read(): Promise; write(bytes: Buffer): Promise; close(): Promise; @@ -62,8 +64,16 @@ export interface RuntimeHostPeerNativeEndpoint { readonly coordinationRelays?: readonly string[]; readonly directDeadlineMs: number; }): Promise; + connectMeshControl(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; + acceptMeshControl(): Promise; close(): Promise; } @@ -74,7 +84,7 @@ interface RuntimeHostPeerNativeModule { readonly expectedPeerId?: string; readonly listenAddresses?: readonly string[]; readonly coordinationRelays?: readonly string[]; - }): RuntimeHostPeerNativeEndpoint; + }): unknown; } export async function ensureRuntimeHostPeerIdentity(input: { @@ -100,12 +110,19 @@ export function startRuntimeHostPeerEndpoint(input: { readonly coordinationRelays?: readonly string[]; }): RuntimeHostPeerNativeEndpoint { try { - return loadNativeModule(input.nativePath).startPeerEndpoint({ + const endpoint = loadNativeModule(input.nativePath).startPeerEndpoint({ keyPath: input.keyPath, ...(input.expectedPeerId ? { expectedPeerId: input.expectedPeerId } : {}), ...(input.listenAddresses ? { listenAddresses: input.listenAddresses } : {}), ...(input.coordinationRelays ? { coordinationRelays: input.coordinationRelays } : {}), }); + if (!isPeerNativeEndpoint(endpoint)) { + throw new RuntimeHostPeerError( + 'peer_native_unavailable', + 'Runtime Host peer native endpoint has an incompatible API', + ); + } + return endpoint; } catch (error) { throw normalizePeerError(error); } @@ -314,9 +331,10 @@ export class RuntimeHostPeerByteStream implements RuntimeHostByteStream { export function normalizePeerError(error: unknown): RuntimeHostPeerError { if (error instanceof RuntimeHostPeerError) return error; const cause = asError(error); - const match = /^(peer_[a-z_]+|direct_path_unavailable|coordination_unavailable):\s*(.*)$/su.exec( - cause.message, - ); + const match = + /^(peer_[a-z_]+|direct_path_unavailable|mesh_control_unavailable|coordination_unavailable):\s*(.*)$/su.exec( + cause.message, + ); if (match && isPeerErrorCode(match[1])) { return new RuntimeHostPeerError(match[1], match[2] || match[1], { cause }); } @@ -334,6 +352,30 @@ function isPeerNativeModule(value: unknown): value is RuntimeHostPeerNativeModul ); } +function isPeerNativeEndpoint(value: unknown): value is RuntimeHostPeerNativeEndpoint { + return ( + typeof value === 'object' && + value !== null && + 'peerId' in value && + isPeerId(value.peerId) && + 'listenAddresses' in value && + Array.isArray(value.listenAddresses) && + value.listenAddresses.every((address) => typeof address === 'string') && + 'connect' in value && + typeof value.connect === 'function' && + 'connectMeshControl' in value && + typeof value.connectMeshControl === 'function' && + 'cancelConnect' in value && + typeof value.cancelConnect === 'function' && + 'accept' in value && + typeof value.accept === 'function' && + 'acceptMeshControl' in value && + typeof value.acceptMeshControl === 'function' && + 'close' in value && + typeof value.close === 'function' + ); +} + function isAuthenticationPreface(value: unknown): value is { v: 1; credential: string } { return ( typeof value === 'object' && @@ -388,6 +430,7 @@ function isPeerErrorCode(value: string | undefined): value is RuntimeHostPeerErr return ( value === 'peer_identity_mismatch' || value === 'direct_path_unavailable' || + value === 'mesh_control_unavailable' || value === 'coordination_unavailable' || value === 'peer_native_unavailable' || value === 'peer_native_failed' || diff --git a/packages/storage/package.json b/packages/storage/package.json index 69f0eddf97..47ea5c4206 100644 --- a/packages/storage/package.json +++ b/packages/storage/package.json @@ -17,6 +17,7 @@ "./encrypted-file-managed-secret-store": "./dist/encrypted-file-managed-secret-store.js", "./execution-stores": "./dist/execution-stores.js", "./external-sessions": "./dist/external-sessions.js", + "./file-lifetime-owner": "./dist/file-lifetime-owner.js", "./file-update-lock": "./dist/file-update-lock.js", "./foreign-session-store": "./dist/foreign-session-store.js", "./git-worktree-child-executor": "./dist/git-worktree-child-executor.js", diff --git a/packages/storage/src/file-lifetime-owner.ts b/packages/storage/src/file-lifetime-owner.ts new file mode 100644 index 0000000000..843ee62af8 --- /dev/null +++ b/packages/storage/src/file-lifetime-owner.ts @@ -0,0 +1,54 @@ +/* + * 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 type { FileHandle } from 'node:fs/promises'; +import { + openStableNativeLockFile, + releaseNativeFileLock, + tryAcquireNativeFileLock, +} from './native-file-lock.js'; + +export interface FileLifetimeOwner { + close(): Promise; +} + +export async function acquireFileLifetimeOwner(path: string): Promise { + const handle = await openStableNativeLockFile(path); + if (!tryAcquireNativeFileLock(handle)) { + await handle.close(); + throw new Error(`Another process owns ${path}`); + } + return new FileLifetimeOwnerImpl(handle); +} + +class FileLifetimeOwnerImpl implements FileLifetimeOwner { + #closeTask: Promise | undefined; + + constructor(private readonly handle: FileHandle) {} + + close(): Promise { + this.#closeTask ??= this.#close(); + return this.#closeTask; + } + + async #close(): Promise { + releaseNativeFileLock(this.handle); + await this.handle.close(); + } +} diff --git a/scripts/smoke-release-cli-package.mjs b/scripts/smoke-release-cli-package.mjs index c832a5a395..d5801ff8d5 100644 --- a/scripts/smoke-release-cli-package.mjs +++ b/scripts/smoke-release-cli-package.mjs @@ -213,6 +213,10 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } packageRoot, 'node_modules/@maka/runtime-host/dist/client/index.js', ); + const mesh = await importInstalled( + packageRoot, + 'node_modules/@maka/runtime-host/dist/peer-mesh/index.js', + ); const access = await importInstalled(packageRoot, 'dist/runtime-host-access-command.js'); const clientDataRoot = join(root, 'peer-client'); const hostRoot = join(root, 'peer-host'); @@ -225,6 +229,11 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } let host; let connection; let peerClient; + let meshAuthority; + let meshMember; + let meshAuthorityPeer; + let meshMemberPeer; + let meshServing; try { delete process.env.MAKA_RUNTIME_HOST_PEER_NATIVE_PATH; delete process.env.MAKA_RUNTIME_HOST_PEER_KEY_PATH; @@ -293,7 +302,70 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } if (status.state !== 'ready') { throw new Error(`Installed Runtime Host direct-peer status is ${status.state}`); } + + const meshMemberKeyPath = join(root, 'mesh-member.key'); + const meshMemberDataRoot = join(root, 'mesh-member'); + meshAuthorityPeer = client.createRuntimeHostPeerClient({ + nativePath, + keyPath: join(root, 'mesh-authority.key'), + listenAddresses: ['/ip4/127.0.0.1/udp/0/quic-v1'], + }); + meshMemberPeer = client.createRuntimeHostPeerClient({ + nativePath, + keyPath: meshMemberKeyPath, + listenAddresses: ['/ip4/127.0.0.1/udp/0/quic-v1'], + }); + meshAuthority = await mesh.openPeerMeshNode({ + dataRoot: join(root, 'mesh-authority'), + peer: meshAuthorityPeer, + }); + meshMember = await mesh.openPeerMeshNode({ + dataRoot: meshMemberDataRoot, + peer: meshMemberPeer, + }); + meshServing = meshAuthority.serve(); + const created = await meshAuthority.create(); + const joined = await meshMember.join(await meshAuthority.invite(created.roster.roster.meshId)); + if (joined.roster.roster.members.length !== 2) { + throw new Error('Installed Runtime Host peer Mesh did not admit the invited peer'); + } + const removed = await meshAuthority.remove( + created.roster.roster.meshId, + meshMemberPeer.identity().peerId, + ); + if (removed.roster.roster.members.length !== 1) { + throw new Error('Installed Runtime Host peer Mesh did not remove the invited peer'); + } + await meshMember.close(); + await meshMemberPeer.close(); + meshMemberPeer = client.createRuntimeHostPeerClient({ + nativePath, + keyPath: meshMemberKeyPath, + listenAddresses: ['/ip4/127.0.0.1/udp/0/quic-v1'], + }); + meshMember = await mesh.openPeerMeshNode({ + dataRoot: meshMemberDataRoot, + peer: meshMemberPeer, + }); + const stale = meshMember.status()[0]; + if (stale?.roster.roster.revision !== joined.roster.roster.revision) { + throw new Error('Installed Runtime Host peer Mesh did not recover the last-known roster'); + } + const rejoined = await meshMember.join( + await meshAuthority.invite(created.roster.roster.meshId), + ); + if ( + rejoined.roster.roster.members.length !== 2 || + rejoined.roster.roster.revision <= stale.roster.roster.revision + ) { + throw new Error('Installed Runtime Host peer Mesh did not re-admit the removed peer'); + } } finally { + await meshAuthority?.close().catch(() => undefined); + await meshMember?.close().catch(() => undefined); + await meshServing?.catch(() => undefined); + await meshAuthorityPeer?.close().catch(() => undefined); + await meshMemberPeer?.close().catch(() => undefined); await connection?.close().catch(() => undefined); await peerClient?.close().catch(() => undefined); await host?.close().catch(() => undefined);