diff --git a/native/runtime-host-peer/src/bindings.rs b/native/runtime-host-peer/src/bindings.rs index d52de13197..cf5132e239 100644 --- a/native/runtime-host-peer/src/bindings.rs +++ b/native/runtime-host-peer/src/bindings.rs @@ -33,7 +33,8 @@ use crate::engine::{self, EngineCommand, PeerError, StreamCommand}; type IncomingStreamReceiver = mpsc::Receiver, PeerError>>; const IDENTITY_PAYLOAD_MAX_BYTES: usize = 8 * 1024; -const MAX_TRANSIT_PEERS: usize = 32; +const MAX_TRANSIT_PEERS: usize = 64; +const MAX_TRANSIT_RELAY_ADDRESSES: usize = 256; #[napi(object)] pub struct StartPeerEndpointOptions { @@ -50,14 +51,20 @@ pub struct ConnectPeerOptions { pub peer_id: String, pub route_hints: Vec, pub coordination_relays: Option>, - pub transit_relays: Option>, + pub transit_relay_peer_ids: Option>, pub direct_deadline_ms: u32, } #[napi(object)] pub struct ConfigurePeerTransitOptions { pub allowed_peer_ids: Vec, - pub trusted_relay_peer_ids: Vec, + pub relay_candidates: Vec, +} + +#[napi(object)] +pub struct PeerTransitRelayCandidate { + pub peer_id: String, + pub addresses: Vec, } #[napi(object)] @@ -123,7 +130,11 @@ impl PeerEndpoint { #[napi] pub async fn configure_transit(&self, options: ConfigurePeerTransitOptions) -> Result<()> { let allowed_peers = parse_peer_ids(options.allowed_peer_ids)?; - let trusted_relays = parse_peer_ids(options.trusted_relay_peer_ids)?; + let relays = parse_transit_relay_candidates(options.relay_candidates)?; + let trusted_relays = relays + .iter() + .filter_map(|address| engine::transit_relay_peer_id(address).ok()) + .collect::>(); let local_peer_id = parse_peer_id(&self.peer_id)?; if allowed_peers.contains(&local_peer_id) || trusted_relays.contains(&local_peer_id) { return Err(Error::new( @@ -134,8 +145,10 @@ impl PeerEndpoint { let (result_tx, result_rx) = oneshot::channel(); self.commands .send(EngineCommand::ConfigureTransit { - allowed_peers, - trusted_relays, + policy: engine::TransitPolicy { + allowed_peers, + relays, + }, result: result_tx, }) .await @@ -227,8 +240,8 @@ async fn connect_peer( options.coordination_relays.unwrap_or_default(), "coordination relay", )?; - let transit_relays = - parse_addresses(options.transit_relays.unwrap_or_default(), "transit relay")?; + let transit_relay_peers = + parse_peer_id_list(options.transit_relay_peer_ids.unwrap_or_default())?; if !(1..=120_000).contains(&options.direct_deadline_ms) { return Err(Error::new( Status::InvalidArg, @@ -244,7 +257,7 @@ async fn connect_peer( peer_id, route_hints, coordination_relays, - transit_relays, + transit_relay_peers, deadline: Duration::from_millis(u64::from(options.direct_deadline_ms)), }, stream_kind, @@ -436,17 +449,56 @@ fn parse_addresses(values: Vec, label: &str) -> Result> { .collect() } +fn parse_transit_relay_candidates( + candidates: Vec, +) -> Result> { + let address_count = candidates.iter().try_fold(0usize, |count, candidate| { + count.checked_add(candidate.addresses.len()) + }); + if address_count.is_none_or(|count| count > MAX_TRANSIT_RELAY_ADDRESSES) { + return Err(Error::new( + Status::InvalidArg, + "transit policy cannot contain more than 256 relay addresses", + )); + } + let mut relays = Vec::new(); + for candidate in candidates { + let Ok(expected_peer) = candidate.peer_id.parse::() else { + continue; + }; + for value in candidate.addresses { + let Ok(address) = value.parse::() else { + continue; + }; + if engine::transit_relay_peer_id(&address).ok() == Some(expected_peer) { + relays.push(address); + } + } + } + relays.sort_unstable_by_key(ToString::to_string); + relays.dedup(); + Ok(relays) +} + fn parse_peer_ids(values: Vec) -> Result> { + Ok(parse_peer_id_list(values)?.into_iter().collect()) +} + +fn parse_peer_id_list(values: Vec) -> Result> { if values.len() > MAX_TRANSIT_PEERS { return Err(Error::new( Status::InvalidArg, - "transit policy cannot contain more than 32 peers", + "transit policy cannot contain more than 64 peers", )); } - values - .into_iter() - .map(|value| parse_peer_id(&value)) - .collect() + let mut peers = Vec::new(); + for value in values { + let peer = parse_peer_id(&value)?; + if !peers.contains(&peer) { + peers.push(peer); + } + } + Ok(peers) } fn validate_identity_payload(payload: &[u8]) -> Result<()> { @@ -472,3 +524,26 @@ fn native_closed_error() -> Error { message: "peer stream is closed".to_owned(), }) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn transit_relay_addresses_are_bound_to_the_declared_peer() { + let expected = PeerId::random(); + let other = PeerId::random(); + let accepted = format!("/ip4/192.0.2.1/tcp/4001/p2p/{expected}"); + let relays = parse_transit_relay_candidates(vec![PeerTransitRelayCandidate { + peer_id: expected.to_string(), + addresses: vec![ + accepted.clone(), + format!("/ip4/192.0.2.2/tcp/4001/p2p/{other}"), + "not-a-multiaddr".to_owned(), + ], + }]) + .expect("candidate policy"); + + assert_eq!(relays, vec![accepted.parse().expect("accepted multiaddr")]); + } +} diff --git a/native/runtime-host-peer/src/engine.rs b/native/runtime-host-peer/src/engine.rs index 7a433dba71..619836e94b 100644 --- a/native/runtime-host-peer/src/engine.rs +++ b/native/runtime-host-peer/src/engine.rs @@ -46,9 +46,9 @@ mod identity_store; mod peer_stream; mod relay_discovery; +pub(crate) use address::transit_relay_peer_id; use address::{ address_with_expected_peer, address_with_peer, coordination_relay_peer_id, is_relayed_address, - transit_relay_peer_id, }; use identity_store::load_or_create_key; use peer_stream::spawn_stream; @@ -110,7 +110,7 @@ pub struct ConnectOptions { pub peer_id: PeerId, pub route_hints: Vec, pub coordination_relays: Vec, - pub transit_relays: Vec, + pub transit_relay_peers: Vec, pub deadline: Duration, } @@ -125,8 +125,7 @@ pub enum EngineCommand { result: oneshot::Sender, }, ConfigureTransit { - allowed_peers: HashSet, - trusted_relays: HashSet, + policy: TransitPolicy, result: oneshot::Sender<()>, }, Stop { @@ -134,6 +133,11 @@ pub enum EngineCommand { }, } +pub struct TransitPolicy { + pub allowed_peers: HashSet, + pub relays: Vec, +} + #[derive(Clone, Default)] pub struct TransitSnapshot { pub allowed_peer_count: usize, @@ -178,7 +182,6 @@ struct PendingConnect { direct_routes: Vec, coordination_relays: Vec, coordination_relay_peers: Vec, - transit_relays: Vec, transit_relay_peers: HashSet, transit_after: Instant, next_route_attempt: Instant, @@ -201,7 +204,6 @@ pub(super) enum DialOrigin { struct StartedConnect { direct_routes: Vec, coordination_relay_peers: Vec, - transit_relays: Vec, transit_relay_peers: HashSet, } @@ -241,6 +243,7 @@ struct DirectConnectState { struct CoordinationRelay { addresses: Vec, automatic_addresses: Vec, + transit_addresses: Vec, reservation_addresses: Vec, connections: HashSet, pending_connection: Option, @@ -260,6 +263,7 @@ impl Default for CoordinationRelay { Self { addresses: Vec::new(), automatic_addresses: Vec::new(), + transit_addresses: Vec::new(), reservation_addresses: Vec::new(), connections: HashSet::new(), pending_connection: None, @@ -281,7 +285,7 @@ impl CoordinationRelay { } fn is_active(&self) -> bool { - self.reserve || self.client_references > 0 + self.reserve || !self.transit_addresses.is_empty() || self.client_references > 0 } fn connection_lost(&mut self, now: Instant) -> Option { @@ -324,7 +328,6 @@ struct OpenedStream { pub(super) enum StreamCompletion { Application { connection_id: ConnectionId, - transit_relay_peer: Option, }, MeshControl { coordination_relay_peers: Vec, @@ -610,7 +613,6 @@ async fn run_endpoint_async( direct_routes: started.direct_routes, coordination_relays: options.coordination_relays, coordination_relay_peers: started.coordination_relay_peers, - transit_relays: started.transit_relays, transit_relay_peers: started.transit_relay_peers, transit_after, next_route_attempt: Instant::now(), @@ -634,26 +636,17 @@ async fn run_endpoint_async( ); } Some(EngineCommand::CancelConnect { request_id, result }) => { - let cancelled = if let Some(mut waiter) = direct.pending.remove(&request_id) { - if let Some(opening) = waiter.opening.take() { - opening.abort(); - } - retire_direct_dials( - &mut swarm, - &mut direct.retiring_connections, - waiter.dials, - None, - ); - release_coordination_relays( + let cancelled = if let Some(waiter) = direct.pending.remove(&request_id) { + fail_pending_connect( &mut swarm, + &mut direct, &mut coordination_relays, - &waiter.coordination_relay_peers, - &direct.active, + waiter, + PeerError::new( + "peer_connect_cancelled", + "the peer connection request was cancelled", + ), ); - let _ = waiter.result.send(Err(PeerError::new( - "peer_connect_cancelled", - "the peer connection request was cancelled", - ))); true } else { false @@ -661,17 +654,42 @@ async fn run_endpoint_async( let _ = result.send(cancelled); } Some(EngineCommand::ConfigureTransit { - allowed_peers, - trusted_relays, + policy, result, }) => { - configure_transit( + let revoked_relays = configure_transit( &mut swarm, &stream_control, + &mut coordination_relays, &mut transit, - allowed_peers, - trusted_relays, + policy, + local_peer_id, + ); + reconcile_pending_transit_connects( + &mut swarm, + &mut direct, + &mut coordination_relays, + &revoked_relays, + ); + retry_connect_routes( + &mut swarm, + &mut direct, + &coordination_relays, + &stream_control, + external_candidate_ready, + Instant::now(), ); + let requests = direct.pending.keys().copied().collect::>(); + for request_id in requests { + maybe_open_peer_stream( + request_id, + &mut direct.pending, + &direct.retiring_connections, + stream_control.clone(), + mesh_control.clone(), + opened_tx.clone(), + ); + } let _ = result.send(()); } Some(EngineCommand::Stop { result }) => { @@ -722,19 +740,8 @@ async fn run_endpoint_async( } Some(completed) = stream_completed_rx.recv() => { match completed.kind { - StreamCompletion::Application { - connection_id, - transit_relay_peer, - } => { + StreamCompletion::Application { connection_id } => { direct.active.retain(|_, active| *active != connection_id); - if let Some(peer_id) = transit_relay_peer { - release_coordination_relays( - &mut swarm, - &mut coordination_relays, - &[peer_id], - &direct.active, - ); - } } StreamCompletion::MeshControl { coordination_relay_peers } => { release_coordination_relays( @@ -748,12 +755,43 @@ async fn run_endpoint_async( 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) => match waiter.stream_kind { + let request_id = opened.request_id; + if let Some(mut waiter) = direct.pending.remove(&opened.request_id) { + match opened.result { + Ok(opened) => { + if waiter.stream_kind == StreamKind::Application + && opened.relay_peer_id.is_some_and(|relay_peer| { + !waiter.transit_relay_peers.contains(&relay_peer) + }) + { + let connection_id = opened.connection_id; + direct.retiring_connections.insert(connection_id); + let _ = swarm.close_connection(connection_id); + waiter.opening.take(); + waiter.dials.remove(&connection_id); + waiter.next_route_attempt = Instant::now(); + direct.pending.insert(request_id, waiter); + retry_connect_routes( + &mut swarm, + &mut direct, + &coordination_relays, + &stream_control, + external_candidate_ready, + Instant::now(), + ); + maybe_open_peer_stream( + request_id, + &mut direct.pending, + &direct.retiring_connections, + stream_control.clone(), + mesh_control.clone(), + opened_tx.clone(), + ); + continue; + } + let result = match waiter.stream_kind { StreamKind::Application => { let connection_id = opened.connection_id; - let transit_relay_peer = opened.relay_peer_id; retire_direct_dials( &mut swarm, &mut direct.retiring_connections, @@ -761,26 +799,17 @@ async fn run_endpoint_async( Some(connection_id), ); direct.active.insert(waiter.peer_id, connection_id); - let released_relays = waiter - .coordination_relay_peers - .iter() - .copied() - .filter(|peer_id| Some(*peer_id) != transit_relay_peer) - .collect::>(); release_coordination_relays( &mut swarm, &mut coordination_relays, - &released_relays, + &waiter.coordination_relay_peers, &direct.active, ); Ok(spawn_stream( waiter.peer_id, opened.stream, Some(( - StreamCompletion::Application { - connection_id, - transit_relay_peer, - }, + StreamCompletion::Application { connection_id }, stream_completed_tx.clone(), )), )) @@ -804,31 +833,28 @@ async fn run_endpoint_async( )), )) } - }, + }; + let _ = waiter.result.send(result); + } Err(message) => { - retire_direct_dials( - &mut swarm, - &mut direct.retiring_connections, - waiter.dials, - None, - ); - release_coordination_relays( - &mut swarm, - &mut coordination_relays, - &waiter.coordination_relay_peers, - &direct.active, - ); let code = match waiter.stream_kind { - StreamKind::Application if !waiter.transit_relays.is_empty() => { + StreamKind::Application + if !waiter.transit_relay_peers.is_empty() => + { "transit_unavailable" } StreamKind::Application => "direct_path_unavailable", StreamKind::MeshControl => "mesh_control_unavailable", }; - Err(PeerError::new(code, message)) + fail_pending_connect( + &mut swarm, + &mut direct, + &mut coordination_relays, + waiter, + PeerError::new(code, message), + ); } - }; - let _ = waiter.result.send(result); + } } } event = swarm.select_next_some() => { @@ -895,24 +921,10 @@ async fn run_endpoint_async( .filter_map(|(request_id, item)| (item.deadline <= now).then_some(*request_id)) .collect::>(); for request_id in expired { - if let Some(mut waiter) = direct.pending.remove(&request_id) { - if let Some(opening) = waiter.opening.take() { - opening.abort(); - } - retire_direct_dials( - &mut swarm, - &mut direct.retiring_connections, - waiter.dials, - None, - ); - release_coordination_relays( - &mut swarm, - &mut coordination_relays, - &waiter.coordination_relay_peers, - &direct.active, - ); + if let Some(waiter) = direct.pending.remove(&request_id) { let (code, message) = match waiter.stream_kind { - StreamKind::Application if !waiter.transit_relays.is_empty() => ( + StreamKind::Application + if !waiter.transit_relay_peers.is_empty() => ( "transit_unavailable", "no direct or approved transit path was established before the deadline", ), @@ -925,7 +937,13 @@ async fn run_endpoint_async( "no Mesh control path was established before the deadline", ), }; - let _ = waiter.result.send(Err(PeerError::new(code, message))); + fail_pending_connect( + &mut swarm, + &mut direct, + &mut coordination_relays, + waiter, + PeerError::new(code, message), + ); } } } @@ -1042,7 +1060,7 @@ fn start_connect( ) -> Result { if options.route_hints.is_empty() && options.coordination_relays.is_empty() - && options.transit_relays.is_empty() + && options.transit_relay_peers.is_empty() { let code = match stream_kind { StreamKind::Application => "direct_path_unavailable", @@ -1067,18 +1085,14 @@ fn start_connect( relay_peers.push(relay_peer); } } - for relay_address in &options.transit_relays { - let relay_peer = transit_relay_peer_id(relay_address)?; + for relay_peer in &options.transit_relay_peers { validate_relay_target( - relay_peer, + *relay_peer, options.peer_id, local_peer_id, "transit_unavailable", "transit relay", )?; - if !relay_peers.contains(&relay_peer) { - relay_peers.push(relay_peer); - } } let direct_targets = options .route_hints @@ -1097,30 +1111,11 @@ fn start_connect( referenced.insert(relay_peer), )?; } - for relay_address in &options.transit_relays { - let relay_peer = transit_relay_peer_id(relay_address) - .expect("transit relay was validated before registration"); - register_coordination_relay( - coordination_relays, - relay_address, - local_peer_id, - false, - referenced.insert(relay_peer), - )?; - } maintain_coordination_relays(swarm, coordination_relays, false, Instant::now()); Ok(StartedConnect { direct_routes: direct_targets, coordination_relay_peers: relay_peers, - transit_relays: options.transit_relays.clone(), - transit_relay_peers: options - .transit_relays - .iter() - .map(|address| { - transit_relay_peer_id(address) - .expect("transit relay was validated before collection") - }) - .collect(), + transit_relay_peers: options.transit_relay_peers.iter().copied().collect(), }) } @@ -1441,10 +1436,22 @@ fn handle_startup_event( fn configure_transit( swarm: &mut Swarm, application_stream: &application_stream::Control, + coordination_relays: &mut HashMap, transit: &mut TransitRuntime, - allowed_peers: HashSet, - trusted_relays: HashSet, -) { + policy: TransitPolicy, + local_peer_id: PeerId, +) -> HashSet { + let TransitPolicy { + allowed_peers, + relays, + } = policy; + let trusted_relays = relays + .iter() + .map(|address| { + transit_relay_peer_id(address) + .expect("transit relay address was validated before reconciliation") + }) + .collect::>(); let was_enabled = transit .allowed_peers .read() @@ -1461,14 +1468,20 @@ fn configure_transit( .collect::>() }) .unwrap_or_default(); - let changed_relays = transit + let (changed_relays, revoked_relays) = transit .trusted_relays .read() .map(|current| { - current - .symmetric_difference(&trusted_relays) - .copied() - .collect::>() + ( + current + .symmetric_difference(&trusted_relays) + .copied() + .collect::>(), + current + .difference(&trusted_relays) + .copied() + .collect::>(), + ) }) .unwrap_or_default(); if let Ok(mut current) = transit.allowed_peers.write() { @@ -1499,7 +1512,157 @@ fn configure_transit( for connection_id in application_stream.connections_via(&changed_relays) { let _ = swarm.close_connection(connection_id); } + reconcile_transit_reservations(swarm, coordination_relays, relays, local_peer_id); publish_transit_snapshot(transit); + revoked_relays +} + +fn reconcile_pending_transit_connects( + swarm: &mut Swarm, + direct: &mut DirectConnectState, + coordination_relays: &mut HashMap, + revoked_relays: &HashSet, +) { + if revoked_relays.is_empty() { + return; + } + let now = Instant::now(); + let mut unavailable = Vec::new(); + for (request_id, waiter) in &mut direct.pending { + if waiter.stream_kind != StreamKind::Application + || waiter.transit_relay_peers.is_disjoint(revoked_relays) + { + continue; + } + waiter + .transit_relay_peers + .retain(|peer| !revoked_relays.contains(peer)); + if let Some(opening) = waiter.opening.take() { + opening.abort(); + } + retire_pending_dials_by_origin( + swarm, + &mut direct.retiring_connections, + waiter, + DialOrigin::Transit, + ); + waiter.next_route_attempt = now; + waiter.transit_after = now; + if waiter.direct_routes.is_empty() + && waiter.coordination_relays.is_empty() + && waiter.transit_relay_peers.is_empty() + { + unavailable.push(*request_id); + } + } + for request_id in unavailable { + let Some(waiter) = direct.pending.remove(&request_id) else { + continue; + }; + fail_pending_connect( + swarm, + direct, + coordination_relays, + waiter, + PeerError::new( + "transit_unavailable", + "transit policy changed while the peer connection was pending", + ), + ); + } +} + +fn retire_pending_dials_by_origin( + swarm: &mut Swarm, + retiring: &mut HashSet, + waiter: &mut PendingConnect, + origin: DialOrigin, +) { + let connections = waiter + .dials + .iter() + .filter_map(|(connection_id, current)| (*current == origin).then_some(*connection_id)) + .collect::>(); + for connection_id in connections { + waiter.dials.remove(&connection_id); + retiring.insert(connection_id); + let _ = swarm.close_connection(connection_id); + } +} + +fn fail_pending_connect( + swarm: &mut Swarm, + direct: &mut DirectConnectState, + coordination_relays: &mut HashMap, + mut waiter: PendingConnect, + error: PeerError, +) { + if let Some(opening) = waiter.opening.take() { + opening.abort(); + } + retire_direct_dials(swarm, &mut direct.retiring_connections, waiter.dials, None); + release_coordination_relays( + swarm, + coordination_relays, + &waiter.coordination_relay_peers, + &direct.active, + ); + let _ = waiter.result.send(Err(error)); +} + +fn reconcile_transit_reservations( + swarm: &mut Swarm, + relays: &mut HashMap, + addresses: Vec, + local_peer_id: PeerId, +) { + let mut desired = HashMap::>::new(); + for address in addresses { + let peer_id = transit_relay_peer_id(&address) + .expect("transit relay address was validated before reconciliation"); + if peer_id == local_peer_id { + continue; + } + remember_relay_address(desired.entry(peer_id).or_default(), address); + } + + let removed = relays + .iter() + .filter_map(|(peer_id, relay)| { + (!relay.transit_addresses.is_empty() && !desired.contains_key(peer_id)) + .then_some(*peer_id) + }) + .collect::>(); + for peer_id in removed { + let remove = if let Some(relay) = relays.get_mut(&peer_id) { + relay.transit_addresses.clear(); + if relay.is_active() { + false + } else { + relay.reservation_accepted = false; + relay.reservation_addresses.clear(); + if let Some(listener) = relay.reservation_listener.take() { + swarm.remove_listener(listener); + } + for connection_id in relay.connections.drain() { + let _ = swarm.close_connection(connection_id); + } + true + } + } else { + false + }; + if remove { + relays.remove(&peer_id); + } + } + for (peer_id, addresses) in desired { + let relay = relays.entry(peer_id).or_default(); + relay.transit_addresses = addresses; + relay.next_connection_attempt = Instant::now(); + relay.next_reservation_attempt = Instant::now(); + } + maintain_coordination_relays(swarm, relays, true, Instant::now()); } fn handle_transit_event(transit: &mut TransitRuntime, event: relay::Event) { @@ -1599,7 +1762,7 @@ fn register_automatic_relay_candidate( return; } let relay = relays.entry(candidate.peer_id).or_default(); - if relay.reserve && !relay.is_automatic() { + if (relay.reserve || !relay.transit_addresses.is_empty()) && !relay.is_automatic() { return; } relay.automatic_addresses = addresses; @@ -1614,6 +1777,7 @@ fn rebalance_automatic_relays( .values() .filter(|relay| { !relay.is_automatic() + && relay.reserve && relay.reservation_accepted && !relay.reservation_addresses.is_empty() }) @@ -1643,6 +1807,9 @@ fn rebalance_automatic_relays( continue; } relay.reserve = false; + if !relay.transit_addresses.is_empty() { + continue; + } relay.reservation_accepted = false; relay.reservation_addresses.clear(); if let Some(listener) = relay.reservation_listener.take() { @@ -1665,7 +1832,7 @@ fn publish_active_coordination_relays( ) { let mut addresses = relays .values() - .filter(|relay| relay.reservation_accepted) + .filter(|relay| relay.reserve && relay.reservation_accepted) .flat_map(|relay| relay.reservation_addresses.iter().cloned()) .collect::>(); addresses.sort_unstable_by_key(ToString::to_string); @@ -1802,6 +1969,11 @@ fn discard_automatic_relay_candidate( } relay.automatic_addresses.clear(); relay.reserve = false; + if !relay.transit_addresses.is_empty() { + relay.next_connection_attempt = Instant::now(); + relay.next_reservation_attempt = Instant::now(); + return; + } relay.reservation_addresses.clear(); if let Some(listener) = relay.reservation_listener.take() { swarm.remove_listener(listener); @@ -1896,7 +2068,7 @@ fn request_coordination_reservation( let Some(relay) = relays.get_mut(&peer_id) else { return; }; - if !relay.reserve + if !(relay.reserve || !relay.transit_addresses.is_empty()) || relay.reservation_listener.is_some() || !relay.identify_received || !relay.identify_sent @@ -1925,6 +2097,7 @@ fn request_coordination_reservation( fn relay_dial_addresses(relay: &CoordinationRelay) -> Vec { let mut addresses = relay.addresses.clone(); addresses.extend(relay.automatic_addresses.iter().cloned()); + addresses.extend(relay.transit_addresses.iter().cloned()); addresses.sort_unstable_by_key(ToString::to_string); addresses.dedup(); addresses @@ -1934,7 +2107,11 @@ fn relay_reservation_addresses(relay: &CoordinationRelay) -> Vec { if relay.is_automatic() { relay.automatic_addresses.clone() } else { - relay.addresses.clone() + let mut addresses = relay.addresses.clone(); + addresses.extend(relay.transit_addresses.iter().cloned()); + addresses.sort_unstable_by_key(ToString::to_string); + addresses.dedup(); + addresses } } @@ -2015,10 +2192,12 @@ fn retry_connect_routes( swarm, peer_id, connect - .transit_relays + .transit_relay_peers .iter() - .map(|relay| { - relay + .filter_map(|relay_peer| coordination_relays.get(relay_peer)) + .flat_map(|relay| relay.transit_addresses.iter()) + .map(|address| { + address .clone() .with(Protocol::P2pCircuit) .with(Protocol::P2p(peer_id)) @@ -2172,40 +2351,39 @@ mod tests { let target_peer_id = ensure_identity(target_key.clone()) .await .expect("create target identity"); - configure_test_transit( - &relay, - HashSet::from([source.peer_id, target_peer_id]), - HashSet::new(), - ) - .await; + configure_test_transit(&relay, HashSet::from([source.peer_id, target_peer_id])).await; let relay_address = relay .listen_addresses .first() .expect("relay listen address") .clone(); - let mut target_options = test_endpoint_options(target_key); - target_options.coordination_relays = vec![relay_address.clone()]; - let mut target = start(target_options).expect("start target"); - configure_test_transit(&target, HashSet::new(), HashSet::from([relay.peer_id])).await; - wait_for_test_snapshot(&relay, |snapshot| snapshot.active_reservation_count == 1).await; - - let (result, response) = oneshot::channel(); - source - .commands - .send(EngineCommand::Connect { - options: ConnectOptions { - request_id: 1, - peer_id: target.peer_id, - route_hints: Vec::new(), - coordination_relays: Vec::new(), - transit_relays: vec![relay_address], - deadline: Duration::from_secs(10), - }, - stream_kind: StreamKind::Application, - result, - }) - .await - .expect("send transit connect"); + let mut target = start(test_endpoint_options(target_key)).expect("start target"); + configure_test_transit_with_reservations( + &target, + HashSet::new(), + vec![relay_address.clone()], + ) + .await; + configure_test_transit_with_reservations( + &source, + HashSet::new(), + vec![relay_address.clone()], + ) + .await; + wait_for_test_snapshot(&relay, |snapshot| snapshot.active_reservation_count == 2).await; + + let response = begin_test_connect( + &source, + ConnectOptions { + request_id: 1, + peer_id: target.peer_id, + route_hints: Vec::new(), + coordination_relays: Vec::new(), + transit_relay_peers: vec![relay.peer_id], + deadline: Duration::from_secs(10), + }, + ) + .await; let source_stream = tokio::time::timeout(Duration::from_secs(10), response) .await .expect("transit connect timeout") @@ -2235,7 +2413,7 @@ mod tests { 1, ); - configure_test_transit(&relay, HashSet::from([target.peer_id]), HashSet::new()).await; + configure_test_transit(&relay, HashSet::from([target.peer_id])).await; wait_for_test_snapshot(&relay, |snapshot| snapshot.active_circuit_count == 0).await; let (result, response) = oneshot::channel(); if source_stream @@ -2256,8 +2434,91 @@ mod tests { ); } + let response = begin_test_connect( + &source, + ConnectOptions { + request_id: 2, + peer_id: PeerId::random(), + route_hints: Vec::new(), + coordination_relays: Vec::new(), + transit_relay_peers: vec![relay.peer_id], + deadline: Duration::from_secs(10), + }, + ) + .await; + configure_test_transit(&source, HashSet::new()).await; + let result = tokio::time::timeout(Duration::from_secs(2), response) + .await + .expect("revoked pending connect timeout") + .expect("revoked pending connect response"); + let Err(error) = result else { + panic!("revoked pending transit connect succeeded"); + }; + assert_eq!(error.code, "transit_unavailable"); + + close_test_stream(source_stream).await; + close_test_stream(target_stream).await; + + configure_test_transit(&relay, HashSet::from([source.peer_id, target.peer_id])).await; + let response = begin_test_connect( + &source, + ConnectOptions { + request_id: 3, + peer_id: target.peer_id, + route_hints: Vec::new(), + coordination_relays: Vec::new(), + transit_relay_peers: vec![relay.peer_id], + deadline: Duration::from_secs(10), + }, + ) + .await; + configure_test_transit_with_reservations(&source, HashSet::new(), vec![relay_address]) + .await; + let source_stream = tokio::time::timeout(Duration::from_secs(10), response) + .await + .expect("late transit policy connect timeout") + .expect("late transit policy connect response") + .expect("late transit policy connect failed"); + let target_stream = tokio::time::timeout(Duration::from_secs(5), target.incoming.recv()) + .await + .expect("late transit policy inbound timeout") + .expect("late transit policy inbound stream"); close_test_stream(source_stream).await; close_test_stream(target_stream).await; + + let unreachable_peer = PeerId::random(); + let unreachable_route = format!("/ip4/127.0.0.1/udp/1/quic-v1/p2p/{unreachable_peer}") + .parse() + .expect("unreachable direct route"); + let response = begin_test_connect( + &source, + ConnectOptions { + request_id: 4, + peer_id: unreachable_peer, + route_hints: vec![unreachable_route], + coordination_relays: Vec::new(), + transit_relay_peers: vec![relay.peer_id], + deadline: Duration::from_secs(10), + }, + ) + .await; + configure_test_transit(&source, HashSet::new()).await; + let (result, cancelled) = oneshot::channel(); + source + .commands + .send(EngineCommand::CancelConnect { + request_id: 4, + result, + }) + .await + .expect("cancel multi-path connect"); + assert!(cancelled.await.expect("cancel response")); + let result = response.await.expect("cancelled connect response"); + let Err(error) = result else { + panic!("cancelled connect unexpectedly succeeded"); + }; + assert_eq!(error.code, "peer_connect_cancelled"); + stop_test_endpoint(source).await; stop_test_endpoint(target).await; stop_test_endpoint(relay).await; @@ -2278,6 +2539,23 @@ mod tests { } } + async fn begin_test_connect( + endpoint: &StartedEndpoint, + options: ConnectOptions, + ) -> oneshot::Receiver> { + let (result, response) = oneshot::channel(); + endpoint + .commands + .send(EngineCommand::Connect { + options, + stream_kind: StreamKind::Application, + result, + }) + .await + .expect("send application connect"); + response + } + async fn connect_test_stream( endpoint: &StartedEndpoint, peer_id: PeerId, @@ -2294,7 +2572,7 @@ mod tests { peer_id, route_hints: vec![route], coordination_relays: Vec::new(), - transit_relays: Vec::new(), + transit_relay_peers: Vec::new(), deadline: Duration::from_secs(5), }, stream_kind, @@ -2351,17 +2629,23 @@ mod tests { endpoint.thread.join().expect("join endpoint thread"); } - async fn configure_test_transit( + async fn configure_test_transit(endpoint: &StartedEndpoint, allowed_peers: HashSet) { + configure_test_transit_with_reservations(endpoint, allowed_peers, Vec::new()).await; + } + + async fn configure_test_transit_with_reservations( endpoint: &StartedEndpoint, allowed_peers: HashSet, - trusted_relays: HashSet, + reservation_relays: Vec, ) { let (result, response) = oneshot::channel(); endpoint .commands .send(EngineCommand::ConfigureTransit { - allowed_peers, - trusted_relays, + policy: TransitPolicy { + allowed_peers, + relays: reservation_relays, + }, result, }) .await @@ -2435,6 +2719,7 @@ mod tests { ( accepted_peer, CoordinationRelay { + reserve: true, reservation_accepted: true, reservation_addresses: vec![accepted_address.clone()], ..CoordinationRelay::default() diff --git a/native/runtime-host-peer/src/engine/address.rs b/native/runtime-host-peer/src/engine/address.rs index 008b9beda8..6ff874799f 100644 --- a/native/runtime-host-peer/src/engine/address.rs +++ b/native/runtime-host-peer/src/engine/address.rs @@ -54,7 +54,7 @@ pub(super) fn coordination_relay_peer_id(address: &Multiaddr) -> Result Result { +pub(crate) fn transit_relay_peer_id(address: &Multiaddr) -> Result { base_relay_peer_id(address, "transit_unavailable", "transit relay") } diff --git a/native/runtime-host-peer/src/lib.rs b/native/runtime-host-peer/src/lib.rs index 93e4348737..c243f01b48 100644 --- a/native/runtime-host-peer/src/lib.rs +++ b/native/runtime-host-peer/src/lib.rs @@ -22,6 +22,6 @@ mod engine; pub use bindings::{ ConfigurePeerTransitOptions, ConnectPeerOptions, PeerEndpoint, PeerIdentitySignature, - PeerStream, PeerTransitSnapshot, StartPeerEndpointOptions, ensure_peer_identity, - sign_peer_identity, start_peer_endpoint, verify_peer_identity, + PeerStream, PeerTransitRelayCandidate, PeerTransitSnapshot, StartPeerEndpointOptions, + ensure_peer_identity, sign_peer_identity, start_peer_endpoint, verify_peer_identity, }; diff --git a/packages/runtime-host/src/__tests__/peer-listener.test.ts b/packages/runtime-host/src/__tests__/peer-listener.test.ts index af7e9b7ab7..e5c6d39e93 100644 --- a/packages/runtime-host/src/__tests__/peer-listener.test.ts +++ b/packages/runtime-host/src/__tests__/peer-listener.test.ts @@ -129,6 +129,12 @@ function peerWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerClient throw new Error('not used'); }, verifyIdentity: () => false, + transitSnapshot: () => ({ + allowedPeerCount: 0, + trustedRelayCount: 0, + activeReservationCount: 0, + activeCircuitCount: 0, + }), configureTransit: async () => undefined, connect: async () => { throw new Error('not used'); diff --git a/packages/runtime-host/src/__tests__/peer-mesh.test.ts b/packages/runtime-host/src/__tests__/peer-mesh.test.ts index 8ee98365f2..00944c0c74 100644 --- a/packages/runtime-host/src/__tests__/peer-mesh.test.ts +++ b/packages/runtime-host/src/__tests__/peer-mesh.test.ts @@ -165,23 +165,27 @@ test('reconciles changed routes, propagates removal, and recovers the verified c await memberC.join(await authority.invite(mesh.roster.roster.meshId)); await memberB.reconcile(); - assert.deepEqual(memberB.resolveRoutes('peer-c')?.routeHints, ['/memory/peer-c']); + assert.deepEqual(memberB.resolveRoutes('peer-c')?.routeHints, ['/memory/peer-c/p2p/peer-c']); - memberCPeer.setRouteHints(['/memory/peer-c-moved']); + memberCPeer.setRouteHints(['/memory/peer-c-moved/p2p/peer-c']); await memberC.reconcile(); await memberB.reconcile(); - assert.deepEqual(memberB.resolveRoutes('peer-c')?.routeHints, ['/memory/peer-c-moved']); + assert.deepEqual(memberB.resolveRoutes('peer-c')?.routeHints, [ + '/memory/peer-c-moved/p2p/peer-c', + ]); await memberC.close(); await serving[2]; await rm(join(root, 'member-c'), { recursive: true, force: true }); now += 6 * 60 * 1_000; - authorityPeer.setRouteHints(['/memory/peer-a-moved']); + authorityPeer.setRouteHints(['/memory/peer-a-moved/p2p/peer-a']); await authority.reconcile(); await memberB.reconcile(); - assert.deepEqual(memberB.resolveRoutes('peer-a')?.routeHints, ['/memory/peer-a-moved']); + assert.deepEqual(memberB.resolveRoutes('peer-a')?.routeHints, [ + '/memory/peer-a-moved/p2p/peer-a', + ]); - memberCPeer.setRouteHints(['/memory/peer-c-rejoined']); + memberCPeer.setRouteHints(['/memory/peer-c-rejoined/p2p/peer-c']); memberC = await openPeerMeshNode({ dataRoot: join(root, 'member-c'), peer: memberCPeer, @@ -189,9 +193,13 @@ test('reconciles changed routes, propagates removal, and recovers the verified c }); serving[2] = memberC.serve(); await memberC.join(await authority.invite(mesh.roster.roster.meshId)); - assert.deepEqual(authority.resolveRoutes('peer-c')?.routeHints, ['/memory/peer-c-rejoined']); + assert.deepEqual(authority.resolveRoutes('peer-c')?.routeHints, [ + '/memory/peer-c-rejoined/p2p/peer-c', + ]); await memberB.reconcile(); - assert.deepEqual(memberB.resolveRoutes('peer-c')?.routeHints, ['/memory/peer-c-rejoined']); + assert.deepEqual(memberB.resolveRoutes('peer-c')?.routeHints, [ + '/memory/peer-c-rejoined/p2p/peer-c', + ]); await authority.remove(mesh.roster.roster.meshId, 'peer-b'); authorityPeer.setResponseDelay(25); @@ -203,7 +211,9 @@ test('reconciles changed routes, propagates removal, and recovers the verified c await memberB.reconcile(); assert.equal(memberB.resolveRoutes('peer-c'), undefined); assert.deepEqual(memberC.status()[0]?.roster.roster.members, ['peer-a', 'peer-c']); - assert.deepEqual(memberC.resolveRoutes('peer-a')?.routeHints, ['/memory/peer-a-moved']); + assert.deepEqual(memberC.resolveRoutes('peer-a')?.routeHints, [ + '/memory/peer-a-moved/p2p/peer-a', + ]); await memberC.close(); await serving[2]; @@ -212,7 +222,9 @@ test('reconciles changed routes, propagates removal, and recovers the verified c peer: memberCPeer, now: () => now, }); - assert.deepEqual(memberC.resolveRoutes('peer-a')?.routeHints, ['/memory/peer-a-moved']); + assert.deepEqual(memberC.resolveRoutes('peer-a')?.routeHints, [ + '/memory/peer-a-moved/p2p/peer-a', + ]); } finally { await Promise.allSettled([authority.close(), memberB.close(), memberC.close()]); await Promise.allSettled(serving); @@ -221,6 +233,79 @@ test('reconciles changed routes, propagates removal, and recovers the verified c } }); +test('reconciles one selected Mesh into signed transit routes and native policy', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-transit-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const memberBPeer = network.create('peer-b'); + const memberCPeer = network.create('peer-c'); + const memberDPeer = network.create('peer-d'); + const authority = await openPeerMeshNode({ dataRoot: join(root, 'a'), peer: authorityPeer }); + const memberB = await openPeerMeshNode({ dataRoot: join(root, 'b'), peer: memberBPeer }); + const memberC = await openPeerMeshNode({ dataRoot: join(root, 'c'), peer: memberCPeer }); + const memberD = await openPeerMeshNode({ dataRoot: join(root, 'd'), peer: memberDPeer }); + const serving = [authority.serve(), memberB.serve(), memberC.serve(), memberD.serve()]; + try { + const meshId = (await authority.create()).roster.roster.meshId; + await memberB.join(await authority.invite(meshId)); + await memberC.join(await authority.invite(meshId)); + await memberD.join(await authority.invite(meshId)); + + authorityPeer.failNextTransitConfiguration(); + await authority.setTransitMesh(meshId); + await authority.reconcile(); + await memberB.reconcile(); + assert.equal(authority.status()[0]?.transitEnabled, true); + assert.deepEqual(authorityPeer.transitPolicy.allowedPeerIds, ['peer-b', 'peer-c', 'peer-d']); + assert.deepEqual(memberBPeer.transitPolicy.relayCandidates, [ + { peerId: 'peer-a', addresses: ['/memory/peer-a/p2p/peer-a'] }, + ]); + assert.deepEqual(memberB.resolveRoutes('peer-c')?.transitRelayPeerIds, ['peer-a']); + + await memberD.setTransitMesh(meshId); + await memberD.reconcile(); + memberDPeer.setRouteHints(['/memory/peer-c/p2p/peer-c']); + await memberD.reconcile(); + await memberB.reconcile(); + assert.deepEqual(memberBPeer.transitPolicy.relayCandidates, [ + { peerId: 'peer-a', addresses: ['/memory/peer-a/p2p/peer-a'] }, + ]); + assert.deepEqual(memberB.resolveRoutes('peer-c')?.transitRelayPeerIds, ['peer-a']); + const secondMeshId = (await memberB.create()).roster.roster.meshId; + await memberD.join(await memberB.invite(secondMeshId)); + await authority.remove(meshId, 'peer-d'); + await memberB.reconcile(); + assert.deepEqual(memberB.resolveRoutes('peer-c')?.transitRelayPeerIds, ['peer-a']); + + await authority.remove(meshId, 'peer-b'); + assert.deepEqual(authorityPeer.transitPolicy.allowedPeerIds, ['peer-c']); + await memberB.reconcile(); + assert.deepEqual(memberBPeer.transitPolicy, { + allowedPeerIds: [], + relayCandidates: [], + }); + + await authority.closeMesh(meshId); + assert.deepEqual(authority.status(), []); + assert.deepEqual(authorityPeer.transitPolicy.allowedPeerIds, []); + } finally { + await Promise.allSettled([ + authority.close(), + memberB.close(), + memberC.close(), + memberD.close(), + ]); + await Promise.allSettled([ + ...serving, + authorityPeer.close(), + memberBPeer.close(), + memberCPeer.close(), + memberDPeer.close(), + ]); + await rm(root, { recursive: true, force: true }); + } +}); + 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'); @@ -363,12 +448,20 @@ class MemoryPeerClient implements PeerMeshTransport { #responseDelayMs = 0; #reachable = true; #routeHints: readonly string[]; + transitPolicy = { + allowedPeerIds: [] as readonly string[], + relayCandidates: [] as readonly { + readonly peerId: string; + readonly addresses: readonly string[]; + }[], + }; + #failNextTransitConfiguration = false; constructor( private readonly peerId: string, private readonly peers: ReadonlyMap, ) { - this.#routeHints = [`/memory/${peerId}`]; + this.#routeHints = [`/memory/${peerId}/p2p/${peerId}`]; } identity() { @@ -391,6 +484,10 @@ class MemoryPeerClient implements PeerMeshTransport { this.#responseDelayMs = delayMs; } + failNextTransitConfiguration(): void { + this.#failNextTransitConfiguration = true; + } + signIdentity(payload: Buffer) { return Promise.resolve({ publicKey: Buffer.from(this.peerId), @@ -409,6 +506,35 @@ class MemoryPeerClient implements PeerMeshTransport { ); } + transitSnapshot() { + return { + allowedPeerCount: this.transitPolicy.allowedPeerIds.length, + activeReservationCount: 0, + activeCircuitCount: 0, + }; + } + + configureTransit(input: { + readonly allowedPeerIds: readonly string[]; + readonly relayCandidates: readonly { + readonly peerId: string; + readonly addresses: readonly string[]; + }[]; + }): Promise { + if (this.#failNextTransitConfiguration) { + this.#failNextTransitConfiguration = false; + return Promise.reject(new Error('Injected transit configuration failure')); + } + this.transitPolicy = { + allowedPeerIds: [...input.allowedPeerIds], + relayCandidates: input.relayCandidates.map(({ peerId, addresses }) => ({ + peerId, + addresses: [...addresses], + })), + }; + return Promise.resolve(); + } + async connectMeshControl(input: { readonly peerId: string; }): Promise { diff --git a/packages/runtime-host/src/__tests__/peer-native.test.ts b/packages/runtime-host/src/__tests__/peer-native.test.ts index 751e93f9c4..9b3628c2be 100644 --- a/packages/runtime-host/src/__tests__/peer-native.test.ts +++ b/packages/runtime-host/src/__tests__/peer-native.test.ts @@ -69,13 +69,13 @@ module.exports = { listenAddresses: [], activeCoordinationRelays: [], transitSnapshot: { allowedPeerCount: 0, activeReservationCount: 0, activeCircuitCount: 0 }, - connect: ({ requestId, peerId, routeHints, coordinationRelays, transitRelays }) => { - stats.requests.push({ requestId, peerId, routeHints, coordinationRelays, transitRelays }); + connect: ({ requestId, peerId, routeHints, coordinationRelays, transitRelayPeerIds }) => { + stats.requests.push({ requestId, peerId, routeHints, coordinationRelays, transitRelayPeerIds }); if (peerId === 'ready') return Promise.resolve(stream); return new Promise((resolve, reject) => pending.set(requestId, { resolve, reject })); }, - connectMeshControl: ({ requestId, peerId, routeHints, coordinationRelays, transitRelays }) => { - stats.requests.push({ requestId, peerId, routeHints, coordinationRelays, transitRelays }); + connectMeshControl: ({ requestId, peerId, routeHints, coordinationRelays, transitRelayPeerIds }) => { + stats.requests.push({ requestId, peerId, routeHints, coordinationRelays, transitRelayPeerIds }); if (peerId === 'ready') return Promise.resolve(stream); return new Promise((resolve, reject) => pending.set(requestId, { resolve, reject })); }, @@ -105,7 +105,7 @@ module.exports = { resolveRoutes: () => ({ routeHints: ['/memory/discovered'], coordinationRelays: ['/memory/relay'], - transitRelays: ['/memory/transit'], + transitRelayPeerIds: ['transit-peer'], }), }, }); @@ -142,28 +142,28 @@ module.exports = { peerId: 'pending', routeHints: ['/memory/discovered', '/memory/1'], coordinationRelays: ['/memory/relay'], - transitRelays: ['/memory/transit'], + transitRelayPeerIds: ['transit-peer'], }, { requestId: 2, peerId: 'shared', routeHints: ['/memory/discovered', '/memory/1'], coordinationRelays: ['/memory/relay'], - transitRelays: ['/memory/transit'], + transitRelayPeerIds: ['transit-peer'], }, { requestId: 3, peerId: 'shared', routeHints: ['/memory/1'], coordinationRelays: [], - transitRelays: [], + transitRelayPeerIds: [], }, { requestId: 4, peerId: 'ready', routeHints: ['/memory/discovered', '/memory/1'], coordinationRelays: ['/memory/relay'], - transitRelays: ['/memory/transit'], + transitRelayPeerIds: ['transit-peer'], }, ], cancellations: [1, 1], diff --git a/packages/runtime-host/src/client/peer-client.ts b/packages/runtime-host/src/client/peer-client.ts index 48df03de26..5707319010 100644 --- a/packages/runtime-host/src/client/peer-client.ts +++ b/packages/runtime-host/src/client/peer-client.ts @@ -25,6 +25,8 @@ import { type RuntimeHostPeerIdentityProof, type RuntimeHostPeerNativeEndpoint, type RuntimeHostPeerNativeStream, + type RuntimeHostPeerTransitRelayCandidate, + type RuntimeHostPeerTransitSnapshot, } from '../transport/peer-native.js'; import { RuntimeHostPermanentReconnectError } from './reconnect-lifecycle.js'; @@ -32,7 +34,7 @@ export interface RuntimeHostPeerConnectInput { readonly peerId: string; readonly routeHints: readonly string[]; readonly coordinationRelays?: readonly string[]; - readonly transitRelays?: readonly string[]; + readonly transitRelayPeerIds?: readonly string[]; readonly directDeadlineMs: number; } @@ -41,7 +43,7 @@ export interface RuntimeHostPeerRouteResolver { | { readonly routeHints: readonly string[]; readonly coordinationRelays: readonly string[]; - readonly transitRelays?: readonly string[]; + readonly transitRelayPeerIds?: readonly string[]; } | undefined; } @@ -54,9 +56,10 @@ export interface RuntimeHostPeerClient { }>; signIdentity(payload: Buffer): Promise; verifyIdentity(peerId: string, payload: Buffer, proof: RuntimeHostPeerIdentityProof): boolean; + transitSnapshot(): RuntimeHostPeerTransitSnapshot; configureTransit(input: { readonly allowedPeerIds: readonly string[]; - readonly trustedRelayPeerIds: readonly string[]; + readonly relayCandidates: readonly RuntimeHostPeerTransitRelayCandidate[]; }): Promise; connect( input: RuntimeHostPeerConnectInput, @@ -179,9 +182,13 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { }); } + transitSnapshot(): RuntimeHostPeerTransitSnapshot { + return Object.freeze({ ...this.#requireEndpoint().transitSnapshot }); + } + configureTransit(input: { readonly allowedPeerIds: readonly string[]; - readonly trustedRelayPeerIds: readonly string[]; + readonly relayCandidates: readonly RuntimeHostPeerTransitRelayCandidate[]; }): Promise { return this.#requireEndpoint().configureTransit(input); } @@ -300,7 +307,11 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { discovered?.coordinationRelays ?? [], input.coordinationRelays, ), - transitRelays: mergeAddresses(discovered?.transitRelays ?? [], input.transitRelays), + transitRelayPeerIds: mergeValues( + discovered?.transitRelayPeerIds ?? [], + input.transitRelayPeerIds, + 64, + ), requestId, }); let settled = false; @@ -457,7 +468,15 @@ function mergeAddresses( primary: readonly string[], secondary: readonly string[] | undefined, ): readonly string[] { - return Object.freeze([...new Set([...primary, ...(secondary ?? [])])].slice(0, 32)); + return mergeValues(primary, secondary, 32); +} + +function mergeValues( + primary: readonly string[], + secondary: readonly string[] | undefined, + limit: number, +): readonly string[] { + return Object.freeze([...new Set([...primary, ...(secondary ?? [])])].slice(0, limit)); } async function cancelPeerConnect( diff --git a/packages/runtime-host/src/peer-mesh/limits.ts b/packages/runtime-host/src/peer-mesh/limits.ts index a15f3bc752..95e1c77d5a 100644 --- a/packages/runtime-host/src/peer-mesh/limits.ts +++ b/packages/runtime-host/src/peer-mesh/limits.ts @@ -22,4 +22,6 @@ 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 const PEER_MESH_MAX_TRANSIT_RELAY_ADDRESSES = 256; +export const PEER_MESH_MAX_TRANSIT_ADDRESSES_PER_RELAY = 4; export const PEER_MESH_ROUTE_RECORD_MAX_BYTES = 4 * 1024; diff --git a/packages/runtime-host/src/peer-mesh/model.ts b/packages/runtime-host/src/peer-mesh/model.ts index d16f5bce41..df63cdadc9 100644 --- a/packages/runtime-host/src/peer-mesh/model.ts +++ b/packages/runtime-host/src/peer-mesh/model.ts @@ -43,6 +43,8 @@ export { PEER_MESH_MAX_MESHES, PEER_MESH_MAX_PENDING_INVITATIONS, PEER_MESH_MAX_ROUTE_HINTS, + PEER_MESH_MAX_TRANSIT_ADDRESSES_PER_RELAY, + PEER_MESH_MAX_TRANSIT_RELAY_ADDRESSES, PEER_MESH_ROUTE_RECORD_MAX_BYTES, } from './limits.js'; @@ -75,6 +77,7 @@ export interface PeerMeshRouteRecordV1 extends PeerMeshAuthorityTarget { readonly version: 1; readonly sequence: number; readonly expiresAt: number; + readonly transitMeshId?: string; } export interface SignedPeerMeshRouteRecordV1 { @@ -231,14 +234,24 @@ export function decodeAuthorityTarget(value: unknown): PeerMeshAuthorityTarget { } export function canonicalPeerMeshRouteRecord(value: unknown): PeerMeshRouteRecordV1 { - const record = exactObject(value, 'Peer Mesh route record', [ + const baseKeys = [ 'version', 'peerId', 'sequence', 'expiresAt', 'routeHints', 'coordinationRelays', - ]); + ]; + const record = exactObject( + value, + 'Peer Mesh route record', + value && + typeof value === 'object' && + !Array.isArray(value) && + Object.hasOwn(value, 'transitMeshId') + ? [...baseKeys, 'transitMeshId'] + : baseKeys, + ); if (record.version !== 1) throw new Error('Unsupported Peer Mesh route record version'); const route = Object.freeze({ version: 1 as const, @@ -249,6 +262,9 @@ export function canonicalPeerMeshRouteRecord(value: unknown): PeerMeshRouteRecor coordinationRelays: Object.freeze( addressArray(record.coordinationRelays, 'coordinationRelays'), ), + ...(record.transitMeshId === undefined + ? {} + : { transitMeshId: string(record.transitMeshId, 'transitMeshId', 128) }), }); if (peerMeshRouteRecordSigningBytes(route).byteLength > PEER_MESH_ROUTE_RECORD_MAX_BYTES) { throw new Error('Peer Mesh route record is too large'); @@ -279,6 +295,7 @@ export function peerMeshRouteRecordSigningBytes(route: PeerMeshRouteRecordV1): B peerId: route.peerId, routeHints: route.routeHints, sequence: route.sequence, + ...(route.transitMeshId ? { transitMeshId: route.transitMeshId } : {}), version: route.version, })}`, ); diff --git a/packages/runtime-host/src/peer-mesh/node.ts b/packages/runtime-host/src/peer-mesh/node.ts index 4aa43b614a..3cb92a65b2 100644 --- a/packages/runtime-host/src/peer-mesh/node.ts +++ b/packages/runtime-host/src/peer-mesh/node.ts @@ -20,6 +20,8 @@ import type { RuntimeHostPeerIdentityProof, RuntimeHostPeerNativeStream, + RuntimeHostPeerTransitRelayCandidate, + RuntimeHostPeerTransitSnapshot, } from '../transport/peer-native.js'; import { setTimeout as delay } from 'node:timers/promises'; import { @@ -35,6 +37,8 @@ import { PEER_MESH_MAX_MESHES, PEER_MESH_MAX_INVITATION_RECORDS, PEER_MESH_MAX_PENDING_INVITATIONS, + PEER_MESH_MAX_TRANSIT_ADDRESSES_PER_RELAY, + PEER_MESH_MAX_TRANSIT_RELAY_ADDRESSES, peerMeshRouteRecordSigningBytes, peerMeshId, peerMeshInvitationSecretDigest, @@ -130,10 +134,13 @@ export interface PeerMeshNode { remove(meshId: string, peerId: string): Promise; leave(meshId: string, signal?: AbortSignal): Promise; closeMesh(meshId: string): Promise; + setTransitMesh(meshId: string | null): Promise; + transitSnapshot(): RuntimeHostPeerTransitSnapshot; resolveRoutes(peerId: string): | { readonly routeHints: readonly string[]; readonly coordinationRelays: readonly string[]; + readonly transitRelayPeerIds: readonly string[]; } | undefined; reconcile(signal?: AbortSignal): Promise; @@ -147,6 +154,7 @@ export interface PeerMeshStatus { readonly roster: SignedPeerMeshRosterV1; readonly pendingInvitationCount: number; readonly memberRoutes: readonly PeerMeshMemberRouteStatus[]; + readonly transitEnabled: boolean; } export interface PeerMeshMemberRouteStatus { @@ -163,6 +171,11 @@ export interface PeerMeshTransport { }>; signIdentity(payload: Buffer): Promise; verifyIdentity(peerId: string, payload: Buffer, proof: RuntimeHostPeerIdentityProof): boolean; + transitSnapshot(): RuntimeHostPeerTransitSnapshot; + configureTransit(input: { + readonly allowedPeerIds: readonly string[]; + readonly relayCandidates: readonly RuntimeHostPeerTransitRelayCandidate[]; + }): Promise; connectMeshControl( input: { readonly peerId: string; @@ -220,6 +233,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { async initialize(): Promise { for (const route of this.#store.read().routes) this.#assertRouteSignature(route); + await this.#reconcileTransit(); } localPeerId(): string { @@ -230,11 +244,13 @@ class PeerMeshNodeImpl implements PeerMeshNode { status(): readonly PeerMeshStatus[] { this.#assertOpen(); const identity = this.#peer.identity(); + const stored = this.#store.read(); return Object.freeze( - this.#store - .read() - .meshes.filter((state) => isActiveMembership(state, identity.peerId)) - .map((state) => peerMeshStatus(state, identity, this.#store.read().routes, this.#now())), + stored.meshes + .filter((state) => isActiveMembership(state, identity.peerId)) + .map((state) => + peerMeshStatus(state, identity, stored.routes, stored.transitMeshId, this.#now()), + ), ); } @@ -283,6 +299,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { findMesh(stored.meshes, state.roster.roster.meshId)!, identity, stored.routes, + stored.transitMeshId, now, ); }); @@ -394,40 +411,46 @@ class PeerMeshNodeImpl implements PeerMeshNode { throw new Error('Peer Mesh authority returned an unrelated roster'); } const routes = await this.#validateRoutes(response.routes, roster, this.#now()); - const state: PeerMeshStateV1 = { - role: 'replica', - authority: { - peerId: invitation.peerId, - routeHints: invitation.routeHints, - coordinationRelays: invitation.coordinationRelays, - }, - roster, - }; - const joined = await this.#store.mutate((current) => { + await this.#store.mutate((current) => { const existing = findMesh(current.meshes, 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'); + const selectedRoster = existing ? selectRoster(existing.roster, roster) : roster; + if (!selectedRoster.roster.members.includes(identity.peerId)) { + throw new Error('Peer Mesh invitation did not establish an active membership'); } + const state: PeerMeshStateV1 = { + role: 'replica', + authority: { + peerId: invitation.peerId, + routeHints: invitation.routeHints, + coordinationRelays: invitation.coordinationRelays, + }, + roster: selectedRoster, + }; if (!existing) assertMeshCapacity(current.meshes, identity.peerId); const meshes = existing ? replaceMesh(current.meshes, state) : appendMesh(current.meshes, state, identity.peerId); return { state: { + ...current, meshes, routes: mergeRoutes(current.routes, [...routes, localRoute], this.#now()), }, - result: state, + result: undefined, }; }); - return peerMeshStatus(joined, identity); + await this.#refreshLocalRoute(); + await this.#reconcileTransit(); + const stored = this.#store.read(); + return peerMeshStatus( + findMesh(stored.meshes, invitation.meshId)!, + identity, + stored.routes, + stored.transitMeshId, + ); } finally { await stream.close().catch(() => undefined); } @@ -494,23 +517,77 @@ class PeerMeshNodeImpl implements PeerMeshNode { })); } + setTransitMesh(meshId: string | null): Promise { + return this.#admitMesh(async () => { + const localPeerId = this.#peer.identity().peerId; + await this.#store.mutate((current) => { + if ( + meshId !== null && + !current.meshes.some( + (mesh) => mesh.roster.roster.meshId === meshId && isActiveMembership(mesh, localPeerId), + ) + ) { + throw new Error('Transit requires an active Peer Mesh membership'); + } + return { + state: { ...current, transitMeshId: meshId }, + result: undefined, + }; + }); + try { + await this.#reconcileTransit(); + await this.#refreshLocalRoute(); + } catch { + void this.reconcile().catch(() => undefined); + } + }); + } + + transitSnapshot(): RuntimeHostPeerTransitSnapshot { + this.#assertOpen(); + return this.#peer.transitSnapshot(); + } + resolveRoutes(peerId: string) { this.#assertOpen(); const now = this.#now(); const stored = this.#store.read(); - const visible = stored.meshes.some( - (state) => - isActiveMembership(state, this.#peer.identity().peerId) && - state.roster.roster.members.includes(peerId), - ); + const sharedMeshIds = stored.meshes + .filter( + (state) => + isActiveMembership(state, this.#peer.identity().peerId) && + state.roster.roster.members.includes(peerId), + ) + .map(({ roster }) => roster.roster.meshId); + const visible = sharedMeshIds.length > 0; if (!visible) return undefined; const route = stored.routes .filter(({ route }) => route.peerId === peerId && route.expiresAt > now) .sort((left, right) => right.route.sequence - left.route.sequence)[0]?.route; - if (!route) return undefined; + const localPeerId = this.#peer.identity().peerId; + const transitRelayPeerIds = transitRelayCandidates( + stored.routes + .filter( + ({ route: candidate }) => + candidate.peerId !== localPeerId && + candidate.peerId !== peerId && + candidate.expiresAt > now && + candidate.transitMeshId !== undefined && + sharedMeshIds.includes(candidate.transitMeshId) && + isActiveMeshMember( + stored.meshes, + candidate.transitMeshId, + localPeerId, + candidate.peerId, + ), + ) + .sort((left, right) => left.route.peerId.localeCompare(right.route.peerId)), + ).map(({ peerId: relayPeerId }) => relayPeerId); + if (!route && transitRelayPeerIds.length === 0) return undefined; return Object.freeze({ - routeHints: route.routeHints, - coordinationRelays: route.coordinationRelays, + routeHints: route?.routeHints ?? [], + coordinationRelays: route?.coordinationRelays ?? [], + transitRelayPeerIds: Object.freeze(transitRelayPeerIds), }); } @@ -576,6 +653,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { ? AbortSignal.any([signal, this.#lifetime.signal]) : this.#lifetime.signal; lifetimeSignal.throwIfAborted(); + await this.#reconcileTransit(); await this.#refreshLocalRoute(); const identity = this.#peer.identity(); const stored = this.#store.read(); @@ -633,6 +711,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { Array.from({ length: Math.min(RECONCILE_CONCURRENCY, pending.length) }, worker), ); this.#reconcileCursor = (start + Math.min(next, pending.length)) % pending.length; + await this.#reconcileTransit(); lifetimeSignal.throwIfAborted(); } @@ -703,7 +782,8 @@ class PeerMeshNodeImpl implements PeerMeshNode { existing && existing.route.expiresAt > now + ROUTE_REFRESH_LEAD_MS && sameAddresses(existing.route.routeHints, identity.listenAddresses) && - sameAddresses(existing.route.coordinationRelays, identity.coordinationRelays) + sameAddresses(existing.route.coordinationRelays, identity.coordinationRelays) && + existing.route.transitMeshId === current.transitMeshId ) { return existing; } @@ -717,9 +797,9 @@ class PeerMeshNodeImpl implements PeerMeshNode { async #signLocalRoute(): Promise { const identity = this.#peer.identity(); - const maxSequence = this.#store - .read() - .routes.filter(({ route }) => route.peerId === identity.peerId) + const stored = this.#store.read(); + const maxSequence = stored.routes + .filter(({ route }) => route.peerId === identity.peerId) .reduce((maximum, { route }) => Math.max(maximum, route.sequence), 0); const route = canonicalPeerMeshRouteRecord({ version: 1, @@ -728,6 +808,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { expiresAt: this.#now() + ROUTE_TTL_MS, routeHints: identity.listenAddresses, coordinationRelays: identity.coordinationRelays, + ...(stored.transitMeshId ? { transitMeshId: stored.transitMeshId } : {}), }); const proof = await this.#peer.signIdentity(peerMeshRouteRecordSigningBytes(route)); const signed = decodeSignedPeerMeshRouteRecord({ @@ -809,6 +890,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { }; return { state: { + ...current, meshes: replaceMesh(current.meshes, next), routes: nextRoster.roster.closed || !nextRoster.roster.members.includes(localPeerId) @@ -818,13 +900,15 @@ class PeerMeshNodeImpl implements PeerMeshNode { result: undefined, }; }); + await this.#refreshLocalRoute(); + await this.#reconcileTransit(); } #assertOpen(): void { if (this.#lifetime.signal.aborted) throw new Error('Peer Mesh node is closed'); } - #updateAuthorityRoster( + async #updateAuthorityRoster( meshId: string, closedIsSuccess: boolean, update: (state: PeerMeshAuthorityStateV1) => { @@ -832,13 +916,13 @@ class PeerMeshNodeImpl implements PeerMeshNode { readonly closed: boolean; }, ): Promise { - return this.#store.mutate((current) => { + await this.#store.mutate((current) => { const state = requireAuthority(current.meshes, meshId); if (state.roster.roster.closed) { if (closedIsSuccess) { return { state: current, - result: peerMeshStatus(state, this.#peer.identity()), + result: undefined, }; } throw new Error('Peer Mesh is closed'); @@ -866,9 +950,18 @@ class PeerMeshNodeImpl implements PeerMeshNode { }; return { state: { ...current, meshes: replaceMesh(current.meshes, updated) }, - result: peerMeshStatus(updated, this.#peer.identity()), + result: undefined, }; }); + await this.#refreshLocalRoute(); + await this.#reconcileTransit(); + const stored = this.#store.read(); + return peerMeshStatus( + findMesh(stored.meshes, meshId)!, + this.#peer.identity(), + stored.routes, + stored.transitMeshId, + ); } #acceptIncoming(stream: RuntimeHostPeerNativeStream): void { @@ -920,6 +1013,8 @@ class PeerMeshNodeImpl implements PeerMeshNode { } else { response = await this.#leave(request.meshId, stream.peerId); } + await this.#refreshLocalRoute(); + await this.#reconcileTransit(); await writeFrame(stream, response); await stream.close(); } catch { @@ -957,7 +1052,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { }; const routes = mergeAuthenticatedRoute(current.routes, remoteRoute, now); return { - state: { meshes: replaceMesh(current.meshes, updated), routes }, + state: { ...current, meshes: replaceMesh(current.meshes, updated), routes }, result: { kind: 'invitation-redeemed', roster: updated.roster, @@ -1025,7 +1120,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { }; const routes = mergeAuthenticatedRoute(current.routes, remoteRoute, now); return { - state: { meshes: replaceMesh(current.meshes, updated), routes }, + state: { ...current, meshes: replaceMesh(current.meshes, updated), routes }, result: { kind: 'invitation-redeemed', roster: state.roster, @@ -1059,7 +1154,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { }; const routes = mergeRoutes(current.routes, [remoteRoute], now); return { - state: { meshes: replaceMesh(current.meshes, updated), routes }, + state: { ...current, meshes: replaceMesh(current.meshes, updated), routes }, result: { kind: 'invitation-redeemed', roster, @@ -1142,7 +1237,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { : current.routes; if (!localMember || !remoteMember) { return { - state: { meshes: replaceMesh(current.meshes, updated), routes }, + state: { ...current, meshes: replaceMesh(current.meshes, updated), routes }, result: { kind: 'sync-result', roster, @@ -1153,7 +1248,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { } const page = responseRoutes(updated, routes, request.knownRoutes, this.#now()); return { - state: { meshes: replaceMesh(current.meshes, updated), routes }, + state: { ...current, meshes: replaceMesh(current.meshes, updated), routes }, result: { kind: 'sync-result', roster, @@ -1163,12 +1258,85 @@ class PeerMeshNodeImpl implements PeerMeshNode { }; }); } + + async #reconcileTransit(): Promise { + const stored = this.#store.read(); + const localPeerId = this.#peer.identity().peerId; + const now = this.#now(); + const selected = stored.meshes.find( + (mesh) => + mesh.roster.roster.meshId === stored.transitMeshId && isActiveMembership(mesh, localPeerId), + ); + const eligibleRelays = stored.routes + .filter(({ route }) => { + if ( + route.peerId === localPeerId || + route.expiresAt <= now || + route.transitMeshId === undefined || + route.routeHints.length === 0 + ) { + return false; + } + return isActiveMeshMember(stored.meshes, route.transitMeshId, localPeerId, route.peerId); + }) + .sort((left, right) => left.route.peerId.localeCompare(right.route.peerId)); + const relayCandidates = transitRelayCandidates(eligibleRelays); + await this.#peer.configureTransit({ + allowedPeerIds: selected + ? selected.roster.roster.members.filter((peerId) => peerId !== localPeerId) + : [], + relayCandidates, + }); + } +} + +function transitRelayCandidates( + routes: readonly SignedPeerMeshRouteRecordV1[], +): readonly RuntimeHostPeerTransitRelayCandidate[] { + let remaining = PEER_MESH_MAX_TRANSIT_RELAY_ADDRESSES; + const candidates: RuntimeHostPeerTransitRelayCandidate[] = []; + for (const { route } of routes) { + if (remaining === 0) break; + const addresses = [ + ...new Set(route.routeHints.filter((address) => isBaseRelayFor(address, route.peerId))), + ].slice(0, Math.min(PEER_MESH_MAX_TRANSIT_ADDRESSES_PER_RELAY, remaining)); + if (addresses.length === 0) continue; + candidates.push(Object.freeze({ peerId: route.peerId, addresses: Object.freeze(addresses) })); + remaining -= addresses.length; + } + return Object.freeze(candidates); +} + +function isBaseRelayFor(address: string, peerId: string): boolean { + const segments = address.split('/'); + const peerProtocol = segments.indexOf('p2p'); + return ( + !segments.includes('p2p-circuit') && + peerProtocol === segments.lastIndexOf('p2p') && + peerProtocol === segments.length - 2 && + segments.at(-1) === peerId + ); +} + +function isActiveMeshMember( + meshes: readonly PeerMeshStateV1[], + meshId: string, + localPeerId: string, + peerId: string, +): boolean { + return meshes.some( + (mesh) => + mesh.roster.roster.meshId === meshId && + isActiveMembership(mesh, localPeerId) && + mesh.roster.roster.members.includes(peerId), + ); } function peerMeshStatus( state: PeerMeshStateV1, identity: ReturnType, routes: readonly SignedPeerMeshRouteRecordV1[] = [], + transitMeshId: string | null = null, now = Date.now(), ): PeerMeshStatus { return Object.freeze({ @@ -1181,6 +1349,7 @@ function peerMeshStatus( (invitation) => invitation.status === 'pending' && invitation.expiresAt > now, ).length : 0, + transitEnabled: state.roster.roster.meshId === transitMeshId, memberRoutes: Object.freeze( state.roster.roster.members.map((peerId) => { if (peerId === identity.peerId) return Object.freeze({ peerId, state: 'local' as const }); diff --git a/packages/runtime-host/src/peer-mesh/store.ts b/packages/runtime-host/src/peer-mesh/store.ts index d5accc6c60..b0f25d3549 100644 --- a/packages/runtime-host/src/peer-mesh/store.ts +++ b/packages/runtime-host/src/peer-mesh/store.ts @@ -76,6 +76,7 @@ export type PeerMeshStateV1 = PeerMeshAuthorityStateV1 | PeerMeshReplicaStateV1; export interface PeerMeshStoredStateV1 { readonly meshes: readonly PeerMeshStateV1[]; readonly routes: readonly SignedPeerMeshRouteRecordV1[]; + readonly transitMeshId: string | null; } export interface PeerMeshStateStore { @@ -180,7 +181,7 @@ class PeerMeshStateStoreImpl implements PeerMeshStateStore { if (this.#failure) throw this.#failure; const updated = operation(this.#state); if (updated.state === this.#state) return updated.result; - const candidate = pruneUnreferencedRoutes(updated.state); + const candidate = pruneUnreferencedRoutes(updated.state, this.localPeerId); const canonical = decodePeerMeshStoredState(candidate, this.localPeerId); assertStateAdvance(this.#state.meshes, canonical.meshes, this.localPeerId); try { @@ -317,6 +318,10 @@ function isRetired(state: PeerMeshStateV1, localPeerId: string): boolean { ); } +function isActiveMembership(state: PeerMeshStateV1, localPeerId: string): boolean { + return !isRetired(state, localPeerId) && 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'); @@ -351,14 +356,25 @@ async function readState( Object.hasOwn(record, 'localPeerId') && Object.hasOwn(record, 'meshes') && Object.hasOwn(record, 'routes'); - if (!versionOne && !versionTwo) { + const versionThree = + record.version === 3 && + Object.keys(record).length === 5 && + Object.hasOwn(record, 'localPeerId') && + Object.hasOwn(record, 'meshes') && + Object.hasOwn(record, 'routes') && + Object.hasOwn(record, 'transitMeshId'); + if (!versionOne && !versionTwo && !versionThree) { 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 decodePeerMeshStoredState( - { meshes: record.meshes, routes: versionOne ? [] : record.routes }, + { + meshes: record.meshes, + routes: versionOne ? [] : record.routes, + transitMeshId: versionThree ? record.transitMeshId : null, + }, expectedLocalPeerId, ); } catch (error) { @@ -366,6 +382,7 @@ async function readState( return Object.freeze({ meshes: Object.freeze([]), routes: Object.freeze([]), + transitMeshId: null, }); } throw error; @@ -387,7 +404,7 @@ async function writeState( localPeerId: string, state: PeerMeshStoredStateV1, ): Promise { - const document = `${JSON.stringify({ version: 2, localPeerId, ...state }, null, 2)}\n`; + const document = `${JSON.stringify({ version: 3, localPeerId, ...state }, null, 2)}\n`; if (Buffer.byteLength(document) > MAX_STATE_BYTES) throw new Error('Peer Mesh state is too large'); const temporary = `${path}.tmp`; @@ -509,24 +526,53 @@ function decodePeerMeshStoredState(value: unknown, localPeerId: string): PeerMes } const record = value as Record; if ( - Object.keys(record).length !== 2 || + Object.keys(record).length !== 3 || !Object.hasOwn(record, 'meshes') || - !Object.hasOwn(record, 'routes') + !Object.hasOwn(record, 'routes') || + !Object.hasOwn(record, 'transitMeshId') ) { throw new Error('Invalid Peer Mesh state document'); } const meshes = decodePeerMeshStates(record.meshes, localPeerId); - return Object.freeze({ meshes, routes: decodeRoutes(record.routes, meshes) }); + const transitMeshId = + record.transitMeshId === null + ? null + : boundedString(record.transitMeshId, 'transitMeshId', 128); + if ( + transitMeshId !== null && + !meshes.some( + (mesh) => + mesh.roster.roster.meshId === transitMeshId && isActiveMembership(mesh, localPeerId), + ) + ) { + throw new Error('Peer Mesh transit selection is not an active membership'); + } + return Object.freeze({ + meshes, + routes: decodeRoutes(record.routes, meshes), + transitMeshId, + }); } -function pruneUnreferencedRoutes(state: PeerMeshStoredStateV1): PeerMeshStoredStateV1 { +function pruneUnreferencedRoutes( + state: PeerMeshStoredStateV1, + localPeerId: string, +): PeerMeshStoredStateV1 { const knownPeers = new Set( state.meshes .filter(({ roster }) => !roster.roster.closed) .flatMap(({ roster }) => roster.roster.members), ); const routes = state.routes.filter(({ route }) => knownPeers.has(route.peerId)); - return routes.length === state.routes.length ? state : { ...state, routes }; + const transitMeshId = state.meshes.some( + (mesh) => + mesh.roster.roster.meshId === state.transitMeshId && isActiveMembership(mesh, localPeerId), + ) + ? state.transitMeshId + : null; + return routes.length === state.routes.length && transitMeshId === state.transitMeshId + ? state + : { ...state, routes, transitMeshId }; } function boundedString(value: unknown, label: string, max: number): string { diff --git a/packages/runtime-host/src/transport/peer-native.ts b/packages/runtime-host/src/transport/peer-native.ts index 449ee05331..368f4e34da 100644 --- a/packages/runtime-host/src/transport/peer-native.ts +++ b/packages/runtime-host/src/transport/peer-native.ts @@ -70,7 +70,7 @@ export interface RuntimeHostPeerNativeEndpoint { readonly peerId: string; readonly routeHints: readonly string[]; readonly coordinationRelays?: readonly string[]; - readonly transitRelays?: readonly string[]; + readonly transitRelayPeerIds?: readonly string[]; readonly directDeadlineMs: number; }): Promise; connectMeshControl(options: { @@ -78,12 +78,12 @@ export interface RuntimeHostPeerNativeEndpoint { readonly peerId: string; readonly routeHints: readonly string[]; readonly coordinationRelays?: readonly string[]; - readonly transitRelays?: readonly string[]; + readonly transitRelayPeerIds?: readonly string[]; readonly directDeadlineMs: number; }): Promise; configureTransit(options: { readonly allowedPeerIds: readonly string[]; - readonly trustedRelayPeerIds: readonly string[]; + readonly relayCandidates: readonly RuntimeHostPeerTransitRelayCandidate[]; }): Promise; cancelConnect(requestId: number): Promise; accept(): Promise; @@ -97,6 +97,11 @@ export interface RuntimeHostPeerTransitSnapshot { readonly activeCircuitCount: number; } +export interface RuntimeHostPeerTransitRelayCandidate { + readonly peerId: string; + readonly addresses: readonly string[]; +} + interface RuntimeHostPeerNativeModule { ensurePeerIdentity(keyPath: string): Promise; signPeerIdentity(