From 46eebdffa42ea48f54c524093703def405cc3c98 Mon Sep 17 00:00:00 2001 From: Wang Date: Wed, 2 Sep 2026 21:40:03 +0800 Subject: [PATCH 1/2] feat(peer): recover accepted relay anchors Generated-by: Codex (gpt-5.6-sol) --- native/runtime-host-peer/src/bindings.rs | 71 +++- native/runtime-host-peer/src/engine.rs | 363 +++++++++++++----- .../src/engine/relay_anchor_store.rs | 264 +++++++++++++ native/runtime-host-peer/src/lib.rs | 5 +- .../src/__tests__/peer-listener.test.ts | 7 + .../src/__tests__/peer-native.test.ts | 8 +- .../runtime-host/src/client/peer-client.ts | 43 ++- .../src/peer-reachability/owner.ts | 21 +- .../runtime-host/src/transport/peer-native.ts | 43 ++- 9 files changed, 689 insertions(+), 136 deletions(-) create mode 100644 native/runtime-host-peer/src/engine/relay_anchor_store.rs diff --git a/native/runtime-host-peer/src/bindings.rs b/native/runtime-host-peer/src/bindings.rs index fed6b83495..8ef78a3fad 100644 --- a/native/runtime-host-peer/src/bindings.rs +++ b/native/runtime-host-peer/src/bindings.rs @@ -43,6 +43,7 @@ const MAX_WEBRTC_STUN_URL_BYTES: usize = 512; #[napi(object)] pub struct StartPeerEndpointOptions { pub key_path: String, + pub relay_anchor_path: Option, pub expected_peer_id: Option, pub listen_addresses: Option>, pub coordination_relays: Option>, @@ -92,11 +93,18 @@ pub struct PeerIdentitySignature { pub signature: Buffer, } +#[napi(object)] +#[derive(Clone)] +pub struct PeerReachabilitySnapshot { + pub generation: u32, + pub listen_addresses: Vec, + pub active_coordination_relays: Vec, +} + #[napi] pub struct PeerEndpoint { peer_id: String, - listen_addresses: Vec, - active_coordination_relays: Arc>>, + reachability: watch::Receiver, transit_snapshot: Arc>, commands: mpsc::Sender, incoming: Arc>>, @@ -113,16 +121,35 @@ impl PeerEndpoint { } #[napi(getter)] - pub fn listen_addresses(&self) -> Vec { - self.listen_addresses.clone() + pub fn reachability_snapshot(&self) -> PeerReachabilitySnapshot { + reachability_snapshot(&self.reachability.borrow()) } - #[napi(getter)] - pub fn active_coordination_relays(&self) -> Vec { - self.active_coordination_relays - .read() - .map(|addresses| addresses.iter().map(ToString::to_string).collect()) - .unwrap_or_default() + #[napi] + pub async fn watch_reachability( + &self, + after_generation: u32, + timeout_ms: u32, + ) -> Result { + if !(1..=300_000).contains(&timeout_ms) { + return Err(Error::new( + Status::InvalidArg, + "reachability watch timeout must be between 1 and 300000 milliseconds", + )); + } + let mut receiver = self.reachability.clone(); + if receiver.borrow().generation == after_generation { + match tokio::time::timeout( + Duration::from_millis(u64::from(timeout_ms)), + receiver.changed(), + ) + .await + { + Ok(Ok(())) | Err(_) => {} + Ok(Err(_)) => return Err(native_closed_error()), + } + } + Ok(reachability_snapshot(&receiver.borrow())) } #[napi(getter)] @@ -390,6 +417,7 @@ impl PeerStream { pub fn start_peer_endpoint(options: StartPeerEndpointOptions) -> Result { let started = engine::start(engine::StartOptions { key_path: PathBuf::from(options.key_path), + relay_anchor_path: options.relay_anchor_path.map(PathBuf::from), expected_peer_id: options .expected_peer_id .map(|value| parse_peer_id(&value)) @@ -408,12 +436,7 @@ pub fn start_peer_endpoint(options: StartPeerEndpointOptions) -> Result Result PeerReachabilitySnapshot { + PeerReachabilitySnapshot { + generation: snapshot.generation, + listen_addresses: snapshot + .listen_addresses + .iter() + .map(ToString::to_string) + .collect(), + active_coordination_relays: snapshot + .active_coordination_relays + .iter() + .map(ToString::to_string) + .collect(), + } +} + #[napi] pub async fn ensure_peer_identity(key_path: String) -> Result { engine::ensure_identity(PathBuf::from(key_path)) diff --git a/native/runtime-host-peer/src/engine.rs b/native/runtime-host-peer/src/engine.rs index 7f650253f6..b8aa00be21 100644 --- a/native/runtime-host-peer/src/engine.rs +++ b/native/runtime-host-peer/src/engine.rs @@ -39,7 +39,7 @@ use libp2p::{ }, tcp, yamux, }; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{mpsc, oneshot, watch}; use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; @@ -52,6 +52,7 @@ mod address; mod application_stream; mod identity_store; mod peer_stream; +mod relay_anchor_store; mod relay_discovery; use address::{address_with_expected_peer, address_with_peer, is_relayed_address}; @@ -59,6 +60,7 @@ pub(crate) use address::{coordination_relay_peer_id, transit_relay_peer_id}; use identity_store::load_or_create_key; use peer_stream::spawn_stream; pub use peer_stream::{DirectTransport, PeerConnectionPath, PeerStream, StreamCommand}; +use relay_anchor_store::RelayAnchorHistory; const APPLICATION_PROTOCOL: &str = "/maka/runtime-host/peer/1"; const MESH_CONTROL_PROTOCOL: &str = "/maka/runtime-host/mesh-control/1"; @@ -81,6 +83,7 @@ const AUTOMATIC_RELAY_COOLDOWN: Duration = Duration::from_secs(30); const IDLE_CONNECTION_TIMEOUT: Duration = Duration::from_secs(10); const TARGET_COORDINATION_RESERVATIONS: usize = 2; const MAX_AUTOMATIC_RELAY_CANDIDATES: usize = 8; +const MAX_REMEMBERED_RELAY_FAILURES: u8 = 3; const MAX_RELAY_ADDRESSES_PER_PEER: usize = 4; const MAX_PUBLISHED_COORDINATION_RELAY_ADDRESSES: usize = 16; const TRANSIT_FALLBACK_DELAY: Duration = Duration::from_secs(3); @@ -93,6 +96,7 @@ const MAX_TRANSIT_CIRCUIT_BYTES: u64 = 256 * 1024 * 1024; #[derive(Clone)] pub struct StartOptions { pub key_path: PathBuf, + pub relay_anchor_path: Option, pub expected_peer_id: Option, pub listen_addresses: Vec, pub coordination_relays: Vec, @@ -102,8 +106,7 @@ pub struct StartOptions { pub struct StartedEndpoint { pub peer_id: PeerId, - pub listen_addresses: Vec, - pub active_coordination_relays: Arc>>, + pub reachability: tokio::sync::watch::Receiver, pub transit_snapshot: Arc>, pub commands: mpsc::Sender, pub incoming: mpsc::Receiver, @@ -112,6 +115,13 @@ pub struct StartedEndpoint { pub thread: thread::JoinHandle<()>, } +#[derive(Clone, Default, PartialEq, Eq)] +pub struct ReachabilitySnapshot { + pub generation: u32, + pub listen_addresses: Vec, + pub active_coordination_relays: Vec, +} + pub struct IdentitySignature { pub public_key: Vec, pub signature: Vec, @@ -272,7 +282,8 @@ struct TransitRuntime { } struct RouteRuntime<'a> { - active_coordination_relays: &'a Arc>>, + reachability: Option<&'a tokio::sync::watch::Sender>, + relay_anchors: Option<&'a mut RelayAnchorHistory>, transit: &'a mut TransitRuntime, } @@ -349,6 +360,8 @@ struct CoordinationRelay { reservation_accepted: bool, client_references: usize, reservation_listener: Option, + remembered: bool, + remembered_failures: u8, next_connection_attempt: Instant, next_reservation_attempt: Instant, replace_relayed_at: Option, @@ -376,6 +389,8 @@ impl Default for CoordinationRelay { reservation_accepted: false, client_references: 0, reservation_listener: None, + remembered: false, + remembered_failures: 0, next_connection_attempt: now, next_reservation_attempt: now, replace_relayed_at: None, @@ -402,7 +417,7 @@ impl CoordinationRelay { self.next_connection_attempt = now; self.replace_relayed_at = None; self.next_reservation_attempt = now - + if self.is_automatic() { + + if self.is_automatic() && !self.remembered { AUTOMATIC_RELAY_COOLDOWN } else { COORDINATION_RETRY_INTERVAL @@ -420,13 +435,23 @@ impl CoordinationRelay { self.reservation_accepted = false; self.reservation_addresses.clear(); self.next_reservation_attempt = now - + if self.is_automatic() { + + if self.is_automatic() && !self.remembered { AUTOMATIC_RELAY_COOLDOWN } else { COORDINATION_RETRY_INTERVAL }; true } + + fn record_remembered_failure(&mut self) { + if !self.remembered { + return; + } + self.remembered_failures = self.remembered_failures.saturating_add(1); + if self.remembered_failures >= MAX_REMEMBERED_RELAY_FAILURES { + self.remembered = false; + } + } } struct OpenedStream { @@ -502,8 +527,7 @@ pub fn start(options: StartOptions) -> Result { 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 active_coordination_relays = Arc::new(RwLock::new(Vec::new())); - let active_coordination_relays_for_thread = Arc::clone(&active_coordination_relays); + let (reachability_tx, reachability_rx) = watch::channel(ReachabilitySnapshot::default()); let transit_snapshot = Arc::new(RwLock::new(TransitSnapshot::default())); let transit_snapshot_for_thread = Arc::clone(&transit_snapshot); let thread = thread::Builder::new() @@ -515,7 +539,7 @@ pub fn start(options: StartOptions) -> Result { incoming_tx, mesh_incoming_tx, ready_tx.clone(), - active_coordination_relays_for_thread, + reachability_tx, transit_snapshot_for_thread, ); if let Err(error) = result { @@ -528,9 +552,8 @@ pub fn start(options: StartOptions) -> Result { .recv_timeout(Duration::from_secs(10)) .map_err(|error| PeerError::new("peer_native_failed", error.to_string()))??; Ok(StartedEndpoint { - peer_id: ready.0, - listen_addresses: ready.1, - active_coordination_relays, + peer_id: ready, + reachability: reachability_rx, transit_snapshot, commands: command_tx, incoming: incoming_rx, @@ -545,8 +568,8 @@ fn run_endpoint( commands: mpsc::Receiver, incoming_tx: mpsc::Sender, mesh_incoming_tx: mpsc::Sender, - ready_tx: std::sync::mpsc::SyncSender), PeerError>>, - active_coordination_relays: Arc>>, + ready_tx: std::sync::mpsc::SyncSender>, + reachability: watch::Sender, transit_snapshot: Arc>, ) -> Result<(), PeerError> { let runtime = tokio::runtime::Builder::new_multi_thread() @@ -560,7 +583,7 @@ fn run_endpoint( incoming_tx, mesh_incoming_tx, ready_tx, - active_coordination_relays, + reachability, transit_snapshot, )) } @@ -570,8 +593,8 @@ async fn run_endpoint_async( mut commands: mpsc::Receiver, incoming_tx: mpsc::Sender, mesh_incoming_tx: mpsc::Sender, - ready_tx: std::sync::mpsc::SyncSender), PeerError>>, - active_coordination_relays: Arc>>, + ready_tx: std::sync::mpsc::SyncSender>, + reachability: watch::Sender, transit_snapshot: Arc>, ) -> Result<(), PeerError> { let key = match options.expected_peer_id { @@ -588,6 +611,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 relay_anchors = + RelayAnchorHistory::open(options.relay_anchor_path.clone(), local_peer_id).await; let webrtc_stun_urls = options.web_rtc_stun_urls.clone(); let allowed_transit_peers = Arc::new(RwLock::new(HashSet::new())); let trusted_transit_relays = Arc::new(RwLock::new(HashSet::new())); @@ -641,6 +666,17 @@ async fn run_endpoint_async( for relay in &options.coordination_relays { register_coordination_relay(&mut coordination_relays, relay, local_peer_id, true, false)?; } + for anchor in relay_anchors.anchors().to_vec() { + register_automatic_relay_candidate( + &mut coordination_relays, + relay_discovery::RelayCandidate { + peer_id: anchor.peer_id, + addresses: anchor.addresses, + }, + local_peer_id, + true, + ); + } maintain_coordination_relays( &mut swarm, &mut coordination_relays, @@ -687,7 +723,13 @@ async fn run_endpoint_async( .listen_on("/webrtc".parse().expect("constant multiaddr")) .map_err(|error| PeerError::new("peer_native_failed", error.to_string()))?; } - let _ = ready_tx.send(Ok((local_peer_id, bound_addresses))); + publish_active_coordination_relays( + &mut coordination_relays, + &reachability, + &mut relay_anchors, + &bound_addresses, + ); + let _ = ready_tx.send(Ok(local_peer_id)); let (opened_tx, mut opened_rx) = mpsc::channel::(COMMAND_CAPACITY); let (stream_completed_tx, mut stream_completed_rx) = @@ -853,6 +895,8 @@ async fn run_endpoint_async( pending.cancellation.cancel(); } outgoing_webrtc_upgrades.abort_all(); + gracefully_disconnect(&mut swarm).await; + relay_anchors.close().await; let _ = result.send(()); return Ok(()); } @@ -863,6 +907,8 @@ async fn run_endpoint_async( pending.cancellation.cancel(); } outgoing_webrtc_upgrades.abort_all(); + gracefully_disconnect(&mut swarm).await; + relay_anchors.close().await; return Ok(()); } }, @@ -979,6 +1025,7 @@ async fn run_endpoint_async( &mut coordination_relays, candidate, local_peer_id, + false, ); rebalance_automatic_relays( &mut swarm, @@ -987,8 +1034,10 @@ async fn run_endpoint_async( Instant::now(), ); publish_active_coordination_relays( - &coordination_relays, - &active_coordination_relays, + &mut coordination_relays, + &reachability, + &mut relay_anchors, + &bound_addresses, ); maintain_coordination_relays( &mut swarm, @@ -1200,7 +1249,8 @@ async fn run_endpoint_async( &mut coordination_relays, &mut direct, RouteRuntime { - active_coordination_relays: &active_coordination_relays, + reachability: Some(&reachability), + relay_anchors: Some(&mut relay_anchors), transit: &mut transit, }, ); @@ -1211,8 +1261,10 @@ async fn run_endpoint_async( Instant::now(), ); publish_active_coordination_relays( - &coordination_relays, - &active_coordination_relays, + &mut coordination_relays, + &reachability, + &mut relay_anchors, + &bound_addresses, ); maintain_coordination_relays( &mut swarm, @@ -1248,8 +1300,10 @@ async fn run_endpoint_async( now, ); publish_active_coordination_relays( - &coordination_relays, - &active_coordination_relays, + &mut coordination_relays, + &reachability, + &mut relay_anchors, + &bound_addresses, ); maintain_coordination_relays( &mut swarm, @@ -1767,12 +1821,25 @@ fn stream_candidate_priority(path: &PeerConnectionPath) -> u8 { } } +async fn gracefully_disconnect(swarm: &mut Swarm) { + let peers = swarm.connected_peers().copied().collect::>(); + for peer_id in peers { + let _ = swarm.disconnect_peer_id(peer_id); + } + let _ = tokio::time::timeout(Duration::from_secs(1), async { + while swarm.connected_peers().next().is_some() { + let _ = swarm.select_next_some().await; + } + }) + .await; +} + fn handle_swarm_event( swarm: &mut Swarm, event: SwarmEvent, coordination_relays: &mut HashMap, direct: &mut DirectConnectState, - route_runtime: RouteRuntime<'_>, + mut route_runtime: RouteRuntime<'_>, ) { match event { SwarmEvent::ConnectionEstablished { @@ -1859,7 +1926,7 @@ fn handle_swarm_event( swarm.remove_listener(listener); } reservation_changed = true; - discard_automatic = relay.is_automatic() && !was_accepted; + discard_automatic = relay.is_automatic() && !relay.remembered && !was_accepted; } if discard_automatic { discard_automatic_relay_candidate( @@ -1870,10 +1937,7 @@ fn handle_swarm_event( ); } if reservation_changed { - publish_active_coordination_relays( - coordination_relays, - route_runtime.active_coordination_relays, - ); + publish_route_runtime(coordination_relays, &mut route_runtime); } } SwarmEvent::OutgoingConnectionError { @@ -1899,7 +1963,8 @@ fn handle_swarm_event( for (peer_id, relay) in coordination_relays.iter_mut() { if relay.pending_connection == Some(connection_id) { relay.pending_connection = None; - if relay.is_automatic() && !relay.reservation_accepted { + relay.record_remembered_failure(); + if relay.is_automatic() && !relay.remembered && !relay.reservation_accepted { failed_automatic = Some(*peer_id); } break; @@ -1946,10 +2011,7 @@ fn handle_swarm_event( if let Some(relay) = coordination_relays.get_mut(&relay_peer_id) { relay.reservation_accepted = true; } - publish_active_coordination_relays( - coordination_relays, - route_runtime.active_coordination_relays, - ); + publish_route_runtime(coordination_relays, &mut route_runtime); } SwarmEvent::Behaviour(BehaviourEvent::RelayServer(event)) => { handle_transit_event(route_runtime.transit, event); @@ -1962,10 +2024,17 @@ fn handle_swarm_event( (relay.reservation_listener == Some(listener_id)).then_some(*peer_id) }); if let Some(relay_peer) = relay_peer { - let automatic = coordination_relays + let (automatic, require_public) = coordination_relays .get(&relay_peer) - .is_some_and(CoordinationRelay::is_automatic); - if let Some(base_address) = reservation_base_address(address, relay_peer, automatic) + .map(|relay| { + ( + relay.is_automatic(), + relay.is_automatic() && !relay.remembered, + ) + }) + .unwrap_or_default(); + if let Some(base_address) = + reservation_base_address(address, relay_peer, require_public) { if let Some(relay) = coordination_relays.get_mut(&relay_peer) { remember_reservation_address(relay, base_address); @@ -1981,10 +2050,7 @@ fn handle_swarm_event( &direct.active, ); } - publish_active_coordination_relays( - coordination_relays, - route_runtime.active_coordination_relays, - ); + publish_route_runtime(coordination_relays, &mut route_runtime); } } SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Received { @@ -2017,7 +2083,10 @@ fn handle_swarm_event( for (peer_id, relay) in coordination_relays.iter_mut() { let was_accepted = relay.reservation_accepted; if relay.listener_closed(listener_id, Instant::now()) { - if relay.is_automatic() && !was_accepted { + if !was_accepted { + relay.record_remembered_failure(); + } + if relay.is_automatic() && !relay.remembered && !was_accepted { rejected_automatic = Some(*peer_id); } changed = true; @@ -2033,10 +2102,7 @@ fn handle_swarm_event( ); } if changed { - publish_active_coordination_relays( - coordination_relays, - route_runtime.active_coordination_relays, - ); + publish_route_runtime(coordination_relays, &mut route_runtime); } } _ => {} @@ -2055,7 +2121,8 @@ fn handle_startup_event( coordination_relays, &mut DirectConnectState::default(), RouteRuntime { - active_coordination_relays: &Arc::new(RwLock::new(Vec::new())), + reachability: None, + relay_anchors: None, transit, }, ); @@ -2451,6 +2518,8 @@ fn register_coordination_relay( let relay = relays.entry(relay_peer).or_default(); if reserve { relay.automatic_addresses.clear(); + relay.remembered = false; + relay.remembered_failures = 0; } relay.reserve |= reserve; if add_client_reference { @@ -2464,6 +2533,7 @@ fn register_automatic_relay_candidate( relays: &mut HashMap, candidate: relay_discovery::RelayCandidate, local_peer_id: PeerId, + remembered: bool, ) { if candidate.peer_id == local_peer_id { return; @@ -2485,6 +2555,10 @@ fn register_automatic_relay_candidate( return; } relay.automatic_addresses = addresses; + if remembered { + relay.remembered = true; + relay.remembered_failures = 0; + } } fn rebalance_automatic_relays( @@ -2506,14 +2580,21 @@ fn rebalance_automatic_relays( let mut automatic = relays .iter() .filter(|(_, relay)| relay.is_automatic()) - .map(|(peer_id, relay)| (*peer_id, relay.reservation_accepted, relay.reserve)) + .map(|(peer_id, relay)| { + ( + *peer_id, + relay.reservation_accepted, + relay.remembered, + relay.reserve, + ) + }) .collect::>(); - automatic.sort_unstable_by_key(|(peer_id, accepted, reserved)| { - (!*accepted, !*reserved, peer_id.to_string()) + automatic.sort_unstable_by_key(|(peer_id, accepted, remembered, reserved)| { + (!*accepted, !*remembered, !*reserved, peer_id.to_string()) }); let mut selected = 0; - for (peer_id, accepted, _) in automatic { + for (peer_id, accepted, _, _) in automatic { let relay = relays .get_mut(&peer_id) .expect("automatic relay was collected from the same map"); @@ -2549,9 +2630,34 @@ fn rebalance_automatic_relays( } fn publish_active_coordination_relays( - relays: &HashMap, - snapshot: &Arc>>, + relays: &mut HashMap, + reachability: &watch::Sender, + relay_anchors: &mut RelayAnchorHistory, + listen_addresses: &[Multiaddr], ) { + for (peer_id, relay) in relays.iter_mut().filter(|(_, relay)| { + relay.reserve && relay.reservation_accepted && !relay.reservation_addresses.is_empty() + }) { + relay_anchors.remember(*peer_id, &relay.reservation_addresses); + relay.remembered = true; + relay.remembered_failures = 0; + } + let addresses = active_coordination_routes(relays); + let current = reachability.borrow().clone(); + if current.listen_addresses == listen_addresses + && current.active_coordination_relays == addresses + { + return; + } + let generation = current.generation.wrapping_add(1).max(1); + reachability.send_replace(ReachabilitySnapshot { + generation, + listen_addresses: listen_addresses.to_vec(), + active_coordination_relays: addresses, + }); +} + +fn active_coordination_routes(relays: &HashMap) -> Vec { let mut relay_routes = relays .iter() .filter(|(_, relay)| relay.reserve && relay.reservation_accepted) @@ -2584,9 +2690,20 @@ fn publish_active_coordination_relays( } } } - if let Ok(mut current) = snapshot.write() { - *current = addresses; - } + addresses +} + +fn publish_route_runtime( + relays: &mut HashMap, + runtime: &mut RouteRuntime<'_>, +) { + let (Some(reachability), Some(relay_anchors)) = + (runtime.reachability, runtime.relay_anchors.as_deref_mut()) + else { + return; + }; + let listen_addresses = reachability.borrow().listen_addresses.clone(); + publish_active_coordination_relays(relays, reachability, relay_anchors, &listen_addresses); } fn bounded_relay_addresses(mut addresses: Vec) -> Vec { @@ -2716,6 +2833,7 @@ fn discard_automatic_relay_candidate( return; } relay.automatic_addresses.clear(); + relay.remembered = false; relay.reserve = false; if !relay.transit_addresses.is_empty() { relay.next_connection_attempt = Instant::now(); @@ -3208,11 +3326,7 @@ mod tests { 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 route = test_listen_address(&right); let mesh_left = connect_test_stream( &left, @@ -3288,11 +3402,7 @@ mod tests { .await .expect("create target identity"); 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 relay_address = test_listen_address(&relay); let mut target = start(test_endpoint_options(target_key)).expect("start target"); let target_relay_stream = connect_test_stream( &target, @@ -3528,11 +3638,7 @@ mod tests { let source_stream = connect_test_stream( &source, target.peer_id, - target - .listen_addresses - .first() - .expect("target retry route") - .clone(), + test_listen_address(&target), 4, StreamKind::Application, ) @@ -3567,11 +3673,7 @@ mod tests { std::fs::create_dir_all(&root).expect("create test root"); let coordination = start(test_endpoint_options(root.join("coordination.key"))) .expect("start coordination"); - let coordination_address = coordination - .listen_addresses - .first() - .expect("coordination listen address") - .clone(); + let coordination_address = test_listen_address(&coordination); let provider_key = root.join("provider.key"); let source_key = root.join("source.key"); @@ -3596,11 +3698,7 @@ mod tests { let mut target = start(test_endpoint_options(root.join("target.key"))).expect("start target"); configure_test_transit(&provider, HashSet::from([source.peer_id, target.peer_id])).await; - let provider_address = provider - .listen_addresses - .first() - .expect("provider listen address") - .clone(); + let provider_address = test_listen_address(&provider); configure_test_transit_with_reservations(&target, HashSet::new(), vec![provider_address]) .await; configure_test_transit_with_candidates( @@ -3658,9 +3756,52 @@ mod tests { std::fs::remove_dir_all(root).expect("remove test root"); } + #[tokio::test(flavor = "multi_thread")] + async fn accepted_relay_anchor_is_reacquired_after_endpoint_restart() { + let root = + std::env::temp_dir().join(format!("maka-peer-anchor-restart-{}", PeerId::random())); + std::fs::create_dir_all(&root).expect("create test root"); + let client_key = root.join("client.key"); + let client_peer_id = ensure_identity(client_key.clone()) + .await + .expect("create client identity"); + let relay = start(test_endpoint_options(root.join("relay.key"))).expect("start relay"); + configure_test_transit(&relay, HashSet::from([client_peer_id])).await; + let relay_address = test_listen_address(&relay); + let anchor_path = root.join("relay-anchors.json"); + + let mut first_options = test_endpoint_options(client_key.clone()); + first_options.relay_anchor_path = Some(anchor_path.clone()); + first_options.coordination_relays = vec![relay_address]; + let first = start(first_options).expect("start first endpoint"); + wait_for_test_coordination_route(&first).await; + assert_eq!(first.peer_id, client_peer_id); + stop_test_endpoint(first).await; + assert!(anchor_path.is_file()); + + let mut restarted_options = test_endpoint_options(client_key); + restarted_options.relay_anchor_path = Some(anchor_path); + let restarted = start(restarted_options).expect("restart endpoint from anchor history"); + wait_for_test_coordination_route(&restarted).await; + assert_eq!(restarted.peer_id, client_peer_id); + assert!( + restarted + .reachability + .borrow() + .active_coordination_relays + .iter() + .any(|address| coordination_relay_peer_id(address).ok() == Some(relay.peer_id)) + ); + + stop_test_endpoint(restarted).await; + stop_test_endpoint(relay).await; + std::fs::remove_dir_all(root).expect("remove test root"); + } + fn test_endpoint_options(key_path: PathBuf) -> StartOptions { StartOptions { key_path, + relay_anchor_path: None, expected_peer_id: None, listen_addresses: vec![ "/ip4/127.0.0.1/udp/0/quic-v1" @@ -3852,11 +3993,11 @@ mod tests { async fn wait_for_test_coordination_route(endpoint: &StartedEndpoint) { tokio::time::timeout(Duration::from_secs(10), async { loop { - if endpoint + if !endpoint + .reachability + .borrow() .active_coordination_relays - .read() - .map(|routes| !routes.is_empty()) - .unwrap_or(false) + .is_empty() { return; } @@ -3867,6 +4008,16 @@ mod tests { .expect("coordination route timeout"); } + fn test_listen_address(endpoint: &StartedEndpoint) -> Multiaddr { + endpoint + .reachability + .borrow() + .listen_addresses + .first() + .expect("test endpoint listen address") + .clone() + } + async fn wait_for_test_snapshot( endpoint: &StartedEndpoint, ready: impl Fn(&TransitSnapshot) -> bool, @@ -3919,6 +4070,26 @@ mod tests { assert_eq!(relay.next_connection_attempt, now); } + #[test] + fn remembered_relay_is_demoted_only_after_bounded_failures() { + let mut relay = CoordinationRelay { + remembered: true, + automatic_addresses: vec![ + "/ip4/192.0.2.1/tcp/4001/p2p/12D3KooWQjzP3hABKwL5qgX6nGkL5u1d4TC7kpBdNJxRYrcx7nVc" + .parse() + .expect("valid automatic relay address"), + ], + ..CoordinationRelay::default() + }; + + for _ in 1..MAX_REMEMBERED_RELAY_FAILURES { + relay.record_remembered_failure(); + assert!(relay.remembered); + } + relay.record_remembered_failure(); + assert!(!relay.remembered); + } + #[test] fn active_coordination_routes_only_publish_accepted_reservations() { let accepted_peer = PeerId::random(); @@ -3947,14 +4118,7 @@ mod tests { }, ), ]); - let snapshot = Arc::new(RwLock::new(Vec::new())); - - publish_active_coordination_relays(&relays, &snapshot); - - assert_eq!( - *snapshot.read().expect("read snapshot"), - vec![accepted_address] - ); + assert_eq!(active_coordination_routes(&relays), vec![accepted_address]); } #[test] @@ -3984,11 +4148,7 @@ mod tests { ) }) .collect::>(); - let snapshot = Arc::new(RwLock::new(Vec::new())); - - publish_active_coordination_relays(&relays, &snapshot); - - let routes = snapshot.read().expect("read snapshot"); + let routes = active_coordination_routes(&relays); assert_eq!(routes.len(), MAX_PUBLISHED_COORDINATION_RELAY_ADDRESSES); assert_eq!( routes @@ -4096,6 +4256,7 @@ mod tests { addresses: vec![automatic_address], }, local, + false, ); register_automatic_relay_candidate( &mut relays, @@ -4104,6 +4265,7 @@ mod tests { addresses: vec![replacement.clone()], }, local, + false, ); let automatic = relays .get_mut(&automatic_peer) @@ -4157,6 +4319,7 @@ mod tests { addresses: vec![client_address], }, local, + false, ); assert!(!bounded[&client_peer].is_automatic()); } diff --git a/native/runtime-host-peer/src/engine/relay_anchor_store.rs b/native/runtime-host-peer/src/engine/relay_anchor_store.rs new file mode 100644 index 0000000000..b6ffe4cec6 --- /dev/null +++ b/native/runtime-host-peer/src/engine/relay_anchor_store.rs @@ -0,0 +1,264 @@ +/* + * 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. + */ + +use std::{collections::HashSet, path::PathBuf}; + +use libp2p::{Multiaddr, PeerId}; +use serde_json::{Map, Value, json}; +use tokio::{io::AsyncWriteExt as _, sync::mpsc, task::JoinHandle}; + +use super::{PeerError, coordination_relay_peer_id, supported_relay_address}; + +const MAX_ANCHOR_PEERS: usize = 8; +const MAX_ADDRESSES_PER_ANCHOR: usize = 4; +const MAX_STATE_BYTES: usize = 64 * 1024; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct RelayAnchor { + pub peer_id: PeerId, + pub addresses: Vec, +} + +pub(super) struct RelayAnchorHistory { + anchors: Vec, + updates: Option>>, + writer: Option>, +} + +impl RelayAnchorHistory { + pub async fn open(path: Option, local_peer_id: PeerId) -> Self { + let anchors = match path.as_ref() { + Some(path) => match read(path, local_peer_id).await { + Ok(anchors) => anchors, + Err(error) => { + eprintln!( + "[peer-relay-anchor] ignored unusable history: {}: {}", + error.code, error.message + ); + Vec::new() + } + }, + None => Vec::new(), + }; + let Some(path) = path else { + return Self { + anchors, + updates: None, + writer: None, + }; + }; + let (updates, mut receiver) = mpsc::unbounded_channel::>(); + let writer = tokio::spawn(async move { + while let Some(anchors) = receiver.recv().await { + if let Err(error) = write(&path, local_peer_id, &anchors).await { + eprintln!( + "[peer-relay-anchor] could not persist history: {}: {}", + error.code, error.message + ); + } + } + }); + Self { + anchors, + updates: Some(updates), + writer: Some(writer), + } + } + + pub fn anchors(&self) -> &[RelayAnchor] { + &self.anchors + } + + pub fn remember(&mut self, peer_id: PeerId, addresses: &[Multiaddr]) { + let mut accepted = Vec::new(); + for address in addresses { + if coordination_relay_peer_id(address).ok() == Some(peer_id) + && supported_relay_address(address, false) + && !accepted.contains(address) + { + accepted.push(address.clone()); + if accepted.len() == MAX_ADDRESSES_PER_ANCHOR { + break; + } + } + } + if accepted.is_empty() { + return; + } + if let Some(anchor) = self + .anchors + .iter_mut() + .find(|anchor| anchor.peer_id == peer_id) + { + if anchor.addresses == accepted { + return; + } + anchor.addresses = accepted; + } else { + self.anchors.insert( + 0, + RelayAnchor { + peer_id, + addresses: accepted, + }, + ); + } + self.anchors.truncate(MAX_ANCHOR_PEERS); + if let Some(updates) = &self.updates { + let _ = updates.send(self.anchors.clone()); + } + } + + pub async fn close(mut self) { + self.updates.take(); + if let Some(writer) = self.writer.take() { + let _ = writer.await; + } + } +} + +async fn read(path: &PathBuf, expected_peer_id: PeerId) -> Result, PeerError> { + let bytes = match tokio::fs::read(path).await { + Ok(bytes) => bytes, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => return Err(invalid_state(error)), + }; + if bytes.len() > MAX_STATE_BYTES { + return Err(invalid_state("relay anchor history is too large")); + } + let document = serde_json::from_slice::(&bytes).map_err(invalid_state)?; + let record = exact_object(&document, &["version", "localPeerId", "anchors"])?; + if record.get("version").and_then(Value::as_u64) != Some(1) + || record.get("localPeerId").and_then(Value::as_str) + != Some(expected_peer_id.to_string().as_str()) + { + return Err(invalid_state( + "relay anchor history belongs to another peer", + )); + } + let entries = record + .get("anchors") + .and_then(Value::as_array) + .filter(|entries| entries.len() <= MAX_ANCHOR_PEERS) + .ok_or_else(|| invalid_state("invalid relay anchor entries"))?; + let mut seen = HashSet::new(); + let mut anchors = Vec::with_capacity(entries.len()); + for entry in entries { + let entry = exact_object(entry, &["peerId", "addresses"])?; + let peer_id = entry + .get("peerId") + .and_then(Value::as_str) + .ok_or_else(|| invalid_state("invalid relay anchor peer"))? + .parse::() + .map_err(invalid_state)?; + if peer_id == expected_peer_id || !seen.insert(peer_id) { + return Err(invalid_state("duplicate or local relay anchor peer")); + } + let addresses = entry + .get("addresses") + .and_then(Value::as_array) + .filter(|addresses| { + !addresses.is_empty() && addresses.len() <= MAX_ADDRESSES_PER_ANCHOR + }) + .ok_or_else(|| invalid_state("invalid relay anchor addresses"))? + .iter() + .map(|address| { + address + .as_str() + .ok_or_else(|| invalid_state("invalid relay anchor address"))? + .parse::() + .map_err(invalid_state) + }) + .collect::, _>>()?; + if addresses.iter().any(|address| { + coordination_relay_peer_id(address).ok() != Some(peer_id) + || !supported_relay_address(address, false) + }) || addresses.iter().collect::>().len() != addresses.len() + { + return Err(invalid_state( + "relay anchor address is not bound to its peer", + )); + } + anchors.push(RelayAnchor { peer_id, addresses }); + } + Ok(anchors) +} + +async fn write( + path: &PathBuf, + local_peer_id: PeerId, + anchors: &[RelayAnchor], +) -> Result<(), PeerError> { + if let Some(parent) = path + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + { + tokio::fs::create_dir_all(parent) + .await + .map_err(invalid_state)?; + } + let document = json!({ + "version": 1, + "localPeerId": local_peer_id.to_string(), + "anchors": anchors.iter().map(|anchor| json!({ + "peerId": anchor.peer_id.to_string(), + "addresses": anchor.addresses.iter().map(ToString::to_string).collect::>(), + })).collect::>(), + }); + let bytes = serde_json::to_vec_pretty(&document).map_err(invalid_state)?; + if bytes.len() > MAX_STATE_BYTES { + return Err(invalid_state("relay anchor history is too large")); + } + let temporary = path.with_extension("tmp"); + let _ = tokio::fs::remove_file(&temporary).await; + let mut options = tokio::fs::OpenOptions::new(); + options.create_new(true).write(true); + #[cfg(unix)] + { + options.mode(0o600); + } + let mut file = options.open(&temporary).await.map_err(invalid_state)?; + file.write_all(&bytes).await.map_err(invalid_state)?; + file.write_all(b"\n").await.map_err(invalid_state)?; + file.sync_all().await.map_err(invalid_state)?; + drop(file); + #[cfg(windows)] + if tokio::fs::try_exists(path).await.unwrap_or(false) { + tokio::fs::remove_file(path).await.map_err(invalid_state)?; + } + if let Err(error) = tokio::fs::rename(&temporary, path).await { + let _ = tokio::fs::remove_file(&temporary).await; + return Err(invalid_state(error)); + } + Ok(()) +} + +fn exact_object<'a>(value: &'a Value, keys: &[&str]) -> Result<&'a Map, PeerError> { + let record = value + .as_object() + .ok_or_else(|| invalid_state("invalid relay anchor document"))?; + if record.len() != keys.len() || keys.iter().any(|key| !record.contains_key(*key)) { + return Err(invalid_state("invalid relay anchor document fields")); + } + Ok(record) +} + +fn invalid_state(error: impl std::fmt::Display) -> PeerError { + PeerError::new("peer_native_failed", error.to_string()) +} diff --git a/native/runtime-host-peer/src/lib.rs b/native/runtime-host-peer/src/lib.rs index e6ea9d0f13..5d12abf38e 100644 --- a/native/runtime-host-peer/src/lib.rs +++ b/native/runtime-host-peer/src/lib.rs @@ -23,6 +23,7 @@ mod webrtc_direct; pub use bindings::{ ConfigurePeerTransitOptions, ConnectPeerOptions, PeerEndpoint, PeerIdentitySignature, - PeerStream, PeerTransitRelayCandidate, PeerTransitSnapshot, StartPeerEndpointOptions, - ensure_peer_identity, sign_peer_identity, start_peer_endpoint, verify_peer_identity, + PeerReachabilitySnapshot, 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 85681e6414..4b3550b795 100644 --- a/packages/runtime-host/src/__tests__/peer-listener.test.ts +++ b/packages/runtime-host/src/__tests__/peer-listener.test.ts @@ -184,7 +184,14 @@ test('projects newly accepted coordination relays from the running peer endpoint }); function peerWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerClient { + const reachability = { + generation: 0, + listenAddresses: [], + activeCoordinationRelays: [], + } as const; return { + reachability: () => reachability, + watchReachability: async () => reachability, identity: () => ({ peerId: 'peer', listenAddresses: [], coordinationRelays: [] }), signIdentity: async () => { throw new Error('not used'); diff --git a/packages/runtime-host/src/__tests__/peer-native.test.ts b/packages/runtime-host/src/__tests__/peer-native.test.ts index c1cccb4200..935c3f332f 100644 --- a/packages/runtime-host/src/__tests__/peer-native.test.ts +++ b/packages/runtime-host/src/__tests__/peer-native.test.ts @@ -67,9 +67,9 @@ module.exports = { stats.starts += 1; return { peerId: 'client', - listenAddresses: [], - activeCoordinationRelays: [], + reachabilitySnapshot: { generation: 0, listenAddresses: [], activeCoordinationRelays: [] }, transitSnapshot: { allowedPeerCount: 0, activeReservationCount: 0, activeCircuitCount: 0, maxReservationCount: 32, maxCircuitCount: 8, maxCircuitsPerPeer: 2, maxCircuitDurationSeconds: 7_200, maxCircuitBytes: 256 * 1024 * 1024 }, + watchReachability: async () => ({ generation: 0, listenAddresses: [], activeCoordinationRelays: [] }), connect: ({ requestId, peerId, routeHints, coordinationRelays, transitRelayPeerIds }) => { stats.requests.push({ requestId, peerId, routeHints, coordinationRelays, transitRelayPeerIds }); if (peerId === 'unreachable') return Promise.reject(Object.assign(new Error('transit_unavailable: no approved route'), { code: 'GenericFailure' })); @@ -303,9 +303,9 @@ module.exports = { starts.push(options); return ({ peerId: 'peer', - listenAddresses: [], - activeCoordinationRelays: [], + reachabilitySnapshot: { generation: 0, listenAddresses: [], activeCoordinationRelays: [] }, transitSnapshot: { allowedPeerCount: 0, activeReservationCount: 0, activeCircuitCount: 0, maxReservationCount: 32, maxCircuitCount: 8, maxCircuitsPerPeer: 2, maxCircuitDurationSeconds: 7_200, maxCircuitBytes: 256 * 1024 * 1024 }, + watchReachability: async () => ({ generation: 0, listenAddresses: [], activeCoordinationRelays: [] }), connect: async () => stream, connectMeshControl: async () => stream, configureTransit: async () => {}, diff --git a/packages/runtime-host/src/client/peer-client.ts b/packages/runtime-host/src/client/peer-client.ts index 998b96dcbc..172e2f7974 100644 --- a/packages/runtime-host/src/client/peer-client.ts +++ b/packages/runtime-host/src/client/peer-client.ts @@ -25,6 +25,7 @@ import { verifyRuntimeHostPeerIdentity, type RuntimeHostPeerIdentityProof, type RuntimeHostPeerNativeEndpoint, + type RuntimeHostPeerNativeReachabilitySnapshot, type RuntimeHostPeerNativeStream, type RuntimeHostPeerTransitRelayCandidate, type RuntimeHostPeerTransitSnapshot, @@ -54,6 +55,11 @@ export interface RuntimeHostPeerRouteResolver { } export interface RuntimeHostPeerClient { + reachability(): RuntimeHostPeerNativeReachabilitySnapshot; + watchReachability( + afterGeneration: number, + timeoutMs: number, + ): Promise; identity(): Readonly<{ peerId: string; listenAddresses: readonly string[]; @@ -96,6 +102,7 @@ export function createRuntimeHostPeerClientFromEnvironment( environment: NodeJS.ProcessEnv = process.env, options: { readonly listenAddresses?: readonly string[]; + readonly relayAnchorPath?: string; readonly coordinationRelays?: readonly string[]; readonly automaticRelayDiscovery?: boolean; readonly webRtcStunUrls?: readonly string[]; @@ -116,6 +123,7 @@ export function createRuntimeHostPeerClientFromEnvironment( export function createRuntimeHostPeerClient(input: { readonly nativePath: string; readonly keyPath: string; + readonly relayAnchorPath?: string; readonly expectedPeerId?: string; readonly listenAddresses?: readonly string[]; readonly coordinationRelays?: readonly string[]; @@ -129,6 +137,7 @@ export function createRuntimeHostPeerClient(input: { class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { readonly #nativePath: string; readonly #keyPath: string; + readonly #relayAnchorPath: string | undefined; readonly #expectedPeerId: string | undefined; readonly #listenAddresses: readonly string[] | undefined; readonly #coordinationRelays: readonly string[] | undefined; @@ -156,6 +165,7 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { constructor(input: { readonly nativePath: string; readonly keyPath: string; + readonly relayAnchorPath?: string; readonly expectedPeerId?: string; readonly listenAddresses?: readonly string[]; readonly coordinationRelays?: readonly string[]; @@ -165,6 +175,7 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { }) { this.#nativePath = input.nativePath; this.#keyPath = input.keyPath; + this.#relayAnchorPath = input.relayAnchorPath; this.#expectedPeerId = input.expectedPeerId; this.#listenAddresses = input.listenAddresses; this.#coordinationRelays = input.coordinationRelays; @@ -179,14 +190,31 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { listenAddresses: readonly string[]; coordinationRelays: readonly string[]; }> { - const endpoint = this.#requireEndpoint(); + const endpoint = this.reachability(); return Object.freeze({ - peerId: endpoint.peerId, + peerId: this.#requireEndpoint().peerId, listenAddresses: Object.freeze([...endpoint.listenAddresses]), coordinationRelays: Object.freeze([...endpoint.activeCoordinationRelays]), }); } + reachability(): RuntimeHostPeerNativeReachabilitySnapshot { + return freezeReachability(this.#requireEndpoint().reachabilitySnapshot); + } + + async watchReachability( + afterGeneration: number, + timeoutMs: number, + ): Promise { + try { + return freezeReachability( + await this.#requireEndpoint().watchReachability(afterGeneration, timeoutMs), + ); + } catch (error) { + throw normalizePeerError(error); + } + } + signIdentity(payload: Buffer): Promise { const peerId = this.#requireEndpoint().peerId; return signRuntimeHostPeerIdentity({ @@ -427,6 +455,7 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { const endpoint = startRuntimeHostPeerEndpoint({ nativePath: this.#nativePath, keyPath: this.#keyPath, + ...(this.#relayAnchorPath ? { relayAnchorPath: this.#relayAnchorPath } : {}), ...(this.#expectedPeerId ? { expectedPeerId: this.#expectedPeerId } : {}), ...(this.#listenAddresses ? { listenAddresses: this.#listenAddresses } : {}), ...(this.#coordinationRelays ? { coordinationRelays: this.#coordinationRelays } : {}), @@ -555,6 +584,16 @@ function mergeAddresses( return mergeValues(primary, secondary, 32); } +function freezeReachability( + snapshot: RuntimeHostPeerNativeReachabilitySnapshot, +): RuntimeHostPeerNativeReachabilitySnapshot { + return Object.freeze({ + generation: snapshot.generation, + listenAddresses: Object.freeze([...snapshot.listenAddresses]), + activeCoordinationRelays: Object.freeze([...snapshot.activeCoordinationRelays]), + }); +} + function mergeValues( primary: readonly string[], secondary: readonly string[] | undefined, diff --git a/packages/runtime-host/src/peer-reachability/owner.ts b/packages/runtime-host/src/peer-reachability/owner.ts index a8136e83af..ed2f112ee1 100644 --- a/packages/runtime-host/src/peer-reachability/owner.ts +++ b/packages/runtime-host/src/peer-reachability/owner.ts @@ -29,13 +29,14 @@ import { type RuntimeHostPeerClient, type RuntimeHostPeerRouteResolver, } from '../client/peer-client.js'; +import { RuntimeHostPermanentReconnectError } from '../client/reconnect-lifecycle.js'; import { openPeerReachabilityPublisher, PeerReachabilityPostCommitError, type PeerReachabilityPublisher, } from './publisher.js'; -const REACHABILITY_OBSERVATION_INTERVAL_MS = 1_000; +const REACHABILITY_WATCH_TIMEOUT_MS = 60_000; const REACHABILITY_RETRY_INTERVAL_MS = 5_000; export interface RuntimeHostPeerEndpointOwner { @@ -66,6 +67,9 @@ export async function openRuntimeHostPeerEndpointOwner(input: { client = createRuntimeHostPeerClient({ nativePath: input.nativePath, keyPath: input.keyPath, + ...(input.automaticRelayDiscovery === true + ? { relayAnchorPath: join(input.dataRoot, 'relay-anchors.json') } + : {}), ...(input.expectedPeerId ? { expectedPeerId: input.expectedPeerId } : {}), ...(input.listenAddresses ? { listenAddresses: input.listenAddresses } : {}), ...(input.coordinationRelays ? { coordinationRelays: input.coordinationRelays } : {}), @@ -91,6 +95,7 @@ export async function openRuntimeHostPeerEndpointOwner(input: { const ownedReachability = reachability; const lifetime = new AbortController(); const maintenance = maintainReachability( + ownedClient, ownedReachability, lifetime.signal, input.onBackgroundReachabilityError, @@ -115,9 +120,9 @@ async function closeEndpointOwner( maintenance: Promise, ): Promise { const errors: unknown[] = []; + await client.close().catch((error: unknown) => errors.push(error)); await maintenance.catch((error: unknown) => errors.push(error)); await reachability.close().catch((error: unknown) => errors.push(error)); - await client.close().catch((error: unknown) => errors.push(error)); await rootOwner.close().catch((error: unknown) => errors.push(error)); if (errors.length === 1) throw errors[0]; if (errors.length > 1) { @@ -126,17 +131,25 @@ async function closeEndpointOwner( } async function maintainReachability( + client: RuntimeHostPeerClient, publisher: PeerReachabilityPublisher, signal: AbortSignal, onError: ((error: unknown) => void) | undefined, ): Promise { + let generation = client.reachability().generation; while (!signal.aborted) { try { - await delay(REACHABILITY_OBSERVATION_INTERVAL_MS, undefined, { signal }); await publisher.refresh(); + const observed = await client.watchReachability(generation, REACHABILITY_WATCH_TIMEOUT_MS); + generation = observed.generation; } catch (error) { if (signal.aborted) return; - if (error instanceof PeerReachabilityPostCommitError) throw error; + if ( + error instanceof PeerReachabilityPostCommitError || + error instanceof RuntimeHostPermanentReconnectError + ) { + throw error; + } try { onError?.(error); } catch { diff --git a/packages/runtime-host/src/transport/peer-native.ts b/packages/runtime-host/src/transport/peer-native.ts index 54d91aed04..29e46a7b9b 100644 --- a/packages/runtime-host/src/transport/peer-native.ts +++ b/packages/runtime-host/src/transport/peer-native.ts @@ -73,9 +73,12 @@ export interface RuntimeHostPeerIdentityProof { export interface RuntimeHostPeerNativeEndpoint { readonly peerId: string; - readonly listenAddresses: readonly string[]; - readonly activeCoordinationRelays: readonly string[]; + readonly reachabilitySnapshot: RuntimeHostPeerNativeReachabilitySnapshot; readonly transitSnapshot: RuntimeHostPeerTransitSnapshot; + watchReachability( + afterGeneration: number, + timeoutMs: number, + ): Promise; connect(options: { readonly requestId: number; readonly peerId: string; @@ -103,6 +106,12 @@ export interface RuntimeHostPeerNativeEndpoint { close(): Promise; } +export interface RuntimeHostPeerNativeReachabilitySnapshot { + readonly generation: number; + readonly listenAddresses: readonly string[]; + readonly activeCoordinationRelays: readonly string[]; +} + export interface RuntimeHostPeerTransitSnapshot { readonly allowedPeerCount: number; readonly activeReservationCount: number; @@ -135,6 +144,7 @@ interface RuntimeHostPeerNativeModule { ): boolean; startPeerEndpoint(options: { readonly keyPath: string; + readonly relayAnchorPath?: string; readonly expectedPeerId?: string; readonly listenAddresses?: readonly string[]; readonly coordinationRelays?: readonly string[]; @@ -207,6 +217,7 @@ export async function ensureRuntimeHostPeerIdentity(input: { export function startRuntimeHostPeerEndpoint(input: { readonly nativePath: string; readonly keyPath: string; + readonly relayAnchorPath?: string; readonly expectedPeerId?: string; readonly listenAddresses?: readonly string[]; readonly coordinationRelays?: readonly string[]; @@ -216,6 +227,7 @@ export function startRuntimeHostPeerEndpoint(input: { try { const endpoint = loadNativeModule(input.nativePath).startPeerEndpoint({ keyPath: input.keyPath, + ...(input.relayAnchorPath ? { relayAnchorPath: input.relayAnchorPath } : {}), ...(input.expectedPeerId ? { expectedPeerId: input.expectedPeerId } : {}), ...(input.listenAddresses ? { listenAddresses: input.listenAddresses } : {}), ...(input.coordinationRelays ? { coordinationRelays: input.coordinationRelays } : {}), @@ -485,12 +497,8 @@ function isPeerNativeEndpoint(value: unknown): value is RuntimeHostPeerNativeEnd value !== null && 'peerId' in value && isPeerId(value.peerId) && - 'listenAddresses' in value && - Array.isArray(value.listenAddresses) && - value.listenAddresses.every((address) => typeof address === 'string') && - 'activeCoordinationRelays' in value && - Array.isArray(value.activeCoordinationRelays) && - value.activeCoordinationRelays.every((address) => typeof address === 'string') && + 'reachabilitySnapshot' in value && + isPeerReachabilitySnapshot(value.reachabilitySnapshot) && 'transitSnapshot' in value && isPeerTransitSnapshot(value.transitSnapshot) && 'connect' in value && @@ -499,6 +507,8 @@ function isPeerNativeEndpoint(value: unknown): value is RuntimeHostPeerNativeEnd typeof value.connectMeshControl === 'function' && 'configureTransit' in value && typeof value.configureTransit === 'function' && + 'watchReachability' in value && + typeof value.watchReachability === 'function' && 'cancelConnect' in value && typeof value.cancelConnect === 'function' && 'accept' in value && @@ -510,6 +520,23 @@ function isPeerNativeEndpoint(value: unknown): value is RuntimeHostPeerNativeEnd ); } +function isPeerReachabilitySnapshot( + value: unknown, +): value is RuntimeHostPeerNativeReachabilitySnapshot { + return ( + typeof value === 'object' && + value !== null && + 'generation' in value && + isCount(value.generation) && + 'listenAddresses' in value && + Array.isArray(value.listenAddresses) && + value.listenAddresses.every((address) => typeof address === 'string') && + 'activeCoordinationRelays' in value && + Array.isArray(value.activeCoordinationRelays) && + value.activeCoordinationRelays.every((address) => typeof address === 'string') + ); +} + function isPeerTransitSnapshot(value: unknown): value is RuntimeHostPeerTransitSnapshot { return ( typeof value === 'object' && From eea3454a35a65ab8c5b6fedd532aefb6d91cf284 Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 01:51:46 +0800 Subject: [PATCH 2/2] fix(peer): keep relay anchor recovery bounded Generated-by: Codex (gpt-5.6-sol) --- .../src/engine/relay_anchor_store.rs | 15 ++++++++++----- .../runtime-host/src/peer-reachability/owner.ts | 4 +--- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/native/runtime-host-peer/src/engine/relay_anchor_store.rs b/native/runtime-host-peer/src/engine/relay_anchor_store.rs index b6ffe4cec6..c8c5b9826b 100644 --- a/native/runtime-host-peer/src/engine/relay_anchor_store.rs +++ b/native/runtime-host-peer/src/engine/relay_anchor_store.rs @@ -21,7 +21,7 @@ use std::{collections::HashSet, path::PathBuf}; use libp2p::{Multiaddr, PeerId}; use serde_json::{Map, Value, json}; -use tokio::{io::AsyncWriteExt as _, sync::mpsc, task::JoinHandle}; +use tokio::{io::AsyncWriteExt as _, sync::watch, task::JoinHandle}; use super::{PeerError, coordination_relay_peer_id, supported_relay_address}; @@ -37,7 +37,7 @@ pub(super) struct RelayAnchor { pub(super) struct RelayAnchorHistory { anchors: Vec, - updates: Option>>, + updates: Option>>>, writer: Option>, } @@ -63,9 +63,14 @@ impl RelayAnchorHistory { writer: None, }; }; - let (updates, mut receiver) = mpsc::unbounded_channel::>(); + // History is a snapshot, not a journal. Coalesce churn into one pending + // latest value so slow storage cannot create an unbounded write queue. + let (updates, mut receiver) = watch::channel::>>(None); let writer = tokio::spawn(async move { - while let Some(anchors) = receiver.recv().await { + while receiver.changed().await.is_ok() { + let Some(anchors) = receiver.borrow_and_update().clone() else { + continue; + }; if let Err(error) = write(&path, local_peer_id, &anchors).await { eprintln!( "[peer-relay-anchor] could not persist history: {}: {}", @@ -121,7 +126,7 @@ impl RelayAnchorHistory { } self.anchors.truncate(MAX_ANCHOR_PEERS); if let Some(updates) = &self.updates { - let _ = updates.send(self.anchors.clone()); + updates.send_replace(Some(self.anchors.clone())); } } diff --git a/packages/runtime-host/src/peer-reachability/owner.ts b/packages/runtime-host/src/peer-reachability/owner.ts index ed2f112ee1..f1ce0beaaa 100644 --- a/packages/runtime-host/src/peer-reachability/owner.ts +++ b/packages/runtime-host/src/peer-reachability/owner.ts @@ -67,9 +67,7 @@ export async function openRuntimeHostPeerEndpointOwner(input: { client = createRuntimeHostPeerClient({ nativePath: input.nativePath, keyPath: input.keyPath, - ...(input.automaticRelayDiscovery === true - ? { relayAnchorPath: join(input.dataRoot, 'relay-anchors.json') } - : {}), + relayAnchorPath: join(input.dataRoot, 'relay-anchors.json'), ...(input.expectedPeerId ? { expectedPeerId: input.expectedPeerId } : {}), ...(input.listenAddresses ? { listenAddresses: input.listenAddresses } : {}), ...(input.coordinationRelays ? { coordinationRelays: input.coordinationRelays } : {}),