You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/09/02 13:44:36 UTC
(maka) 01/01: feat(peer): recover accepted relay anchors
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch feat/peer-relay-anchor-recovery
in repository https://gitbox.apache.org/repos/asf/maka.git
commit f98e1df456ec93c8d2190cdb947d2908b9789b37
Author: Wang <m4...@gmail.com>
AuthorDate: Wed Sep 2 21:40:03 2026 +0800
feat(peer): recover accepted relay anchors
---
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 +
.../runtime-host/src/__tests__/peer-native.test.ts | 8 +-
packages/runtime-host/src/client/peer-client.ts | 43 ++-
.../runtime-host/src/peer-reachability/owner.ts | 21 +-
packages/runtime-host/src/transport/peer-native.ts | 43 ++-
9 files changed, 689 insertions(+), 136 deletions(-)
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<String>,
pub expected_peer_id: Option<String>,
pub listen_addresses: Option<Vec<String>>,
pub coordination_relays: Option<Vec<String>>,
@@ -92,11 +93,18 @@ pub struct PeerIdentitySignature {
pub signature: Buffer,
}
+#[napi(object)]
+#[derive(Clone)]
+pub struct PeerReachabilitySnapshot {
+ pub generation: u32,
+ pub listen_addresses: Vec<String>,
+ pub active_coordination_relays: Vec<String>,
+}
+
#[napi]
pub struct PeerEndpoint {
peer_id: String,
- listen_addresses: Vec<String>,
- active_coordination_relays: Arc<RwLock<Vec<Multiaddr>>>,
+ reachability: watch::Receiver<engine::ReachabilitySnapshot>,
transit_snapshot: Arc<RwLock<engine::TransitSnapshot>>,
commands: mpsc::Sender<EngineCommand>,
incoming: Arc<AsyncMutex<mpsc::Receiver<engine::PeerStream>>>,
@@ -113,16 +121,35 @@ impl PeerEndpoint {
}
#[napi(getter)]
- pub fn listen_addresses(&self) -> Vec<String> {
- self.listen_addresses.clone()
+ pub fn reachability_snapshot(&self) -> PeerReachabilitySnapshot {
+ reachability_snapshot(&self.reachability.borrow())
}
- #[napi(getter)]
- pub fn active_coordination_relays(&self) -> Vec<String> {
- 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<PeerReachabilitySnapshot> {
+ 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<PeerEndpoint> {
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<PeerEndp
.map_err(peer_error)?;
Ok(PeerEndpoint {
peer_id: started.peer_id.to_string(),
- listen_addresses: started
- .listen_addresses
- .into_iter()
- .map(|address| address.to_string())
- .collect(),
- active_coordination_relays: started.active_coordination_relays,
+ reachability: started.reachability,
transit_snapshot: started.transit_snapshot,
commands: started.commands,
incoming: Arc::new(AsyncMutex::new(started.incoming)),
@@ -423,6 +446,22 @@ pub fn start_peer_endpoint(options: StartPeerEndpointOptions) -> Result<PeerEndp
})
}
+fn reachability_snapshot(snapshot: &engine::ReachabilitySnapshot) -> 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<String> {
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<PathBuf>,
pub expected_peer_id: Option<PeerId>,
pub listen_addresses: Vec<Multiaddr>,
pub coordination_relays: Vec<Multiaddr>,
@@ -102,8 +106,7 @@ pub struct StartOptions {
pub struct StartedEndpoint {
pub peer_id: PeerId,
- pub listen_addresses: Vec<Multiaddr>,
- pub active_coordination_relays: Arc<RwLock<Vec<Multiaddr>>>,
+ pub reachability: tokio::sync::watch::Receiver<ReachabilitySnapshot>,
pub transit_snapshot: Arc<RwLock<TransitSnapshot>>,
pub commands: mpsc::Sender<EngineCommand>,
pub incoming: mpsc::Receiver<PeerStream>,
@@ -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<Multiaddr>,
+ pub active_coordination_relays: Vec<Multiaddr>,
+}
+
pub struct IdentitySignature {
pub public_key: Vec<u8>,
pub signature: Vec<u8>,
@@ -272,7 +282,8 @@ struct TransitRuntime {
}
struct RouteRuntime<'a> {
- active_coordination_relays: &'a Arc<RwLock<Vec<Multiaddr>>>,
+ reachability: Option<&'a tokio::sync::watch::Sender<ReachabilitySnapshot>>,
+ 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<ListenerId>,
+ remembered: bool,
+ remembered_failures: u8,
next_connection_attempt: Instant,
next_reservation_attempt: Instant,
replace_relayed_at: Option<Instant>,
@@ -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<StartedEndpoint, PeerError> {
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<StartedEndpoint, PeerError> {
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<StartedEndpoint, PeerError> {
.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<EngineCommand>,
incoming_tx: mpsc::Sender<PeerStream>,
mesh_incoming_tx: mpsc::Sender<PeerStream>,
- ready_tx: std::sync::mpsc::SyncSender<Result<(PeerId, Vec<Multiaddr>), PeerError>>,
- active_coordination_relays: Arc<RwLock<Vec<Multiaddr>>>,
+ ready_tx: std::sync::mpsc::SyncSender<Result<PeerId, PeerError>>,
+ reachability: watch::Sender<ReachabilitySnapshot>,
transit_snapshot: Arc<RwLock<TransitSnapshot>>,
) -> 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<EngineCommand>,
incoming_tx: mpsc::Sender<PeerStream>,
mesh_incoming_tx: mpsc::Sender<PeerStream>,
- ready_tx: std::sync::mpsc::SyncSender<Result<(PeerId, Vec<Multiaddr>), PeerError>>,
- active_coordination_relays: Arc<RwLock<Vec<Multiaddr>>>,
+ ready_tx: std::sync::mpsc::SyncSender<Result<PeerId, PeerError>>,
+ reachability: watch::Sender<ReachabilitySnapshot>,
transit_snapshot: Arc<RwLock<TransitSnapshot>>,
) -> 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::<OpenedStream>(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<Behaviour>) {
+ let peers = swarm.connected_peers().copied().collect::<Vec<_>>();
+ 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<Behaviour>,
event: SwarmEvent<BehaviourEvent>,
coordination_relays: &mut HashMap<PeerId, CoordinationRelay>,
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<PeerId, CoordinationRelay>,
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::<Vec<_>>();
- 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<PeerId, CoordinationRelay>,
- snapshot: &Arc<RwLock<Vec<Multiaddr>>>,
+ relays: &mut HashMap<PeerId, CoordinationRelay>,
+ reachability: &watch::Sender<ReachabilitySnapshot>,
+ 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<PeerId, CoordinationRelay>) -> Vec<Multiaddr> {
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<PeerId, CoordinationRelay>,
+ 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<Multiaddr>) -> Vec<Multiaddr> {
@@ -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::<HashMap<_, _>>();
- 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<Multiaddr>,
+}
+
+pub(super) struct RelayAnchorHistory {
+ anchors: Vec<RelayAnchor>,
+ updates: Option<mpsc::UnboundedSender<Vec<RelayAnchor>>>,
+ writer: Option<JoinHandle<()>>,
+}
+
+impl RelayAnchorHistory {
+ pub async fn open(path: Option<PathBuf>, 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::<Vec<RelayAnchor>>();
+ 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<Vec<RelayAnchor>, 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::<Value>(&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::<PeerId>()
+ .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::<Multiaddr>()
+ .map_err(invalid_state)
+ })
+ .collect::<Result<Vec<_>, _>>()?;
+ if addresses.iter().any(|address| {
+ coordination_relay_peer_id(address).ok() != Some(peer_id)
+ || !supported_relay_address(address, false)
+ }) || addresses.iter().collect::<HashSet<_>>().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::<Vec<_>>(),
+ })).collect::<Vec<_>>(),
+ });
+ 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<String, Value>, 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<RuntimeHostPeerNativeReachabilitySnapshot>;
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<RuntimeHostPeerNativeReachabilitySnapshot> {
+ try {
+ return freezeReachability(
+ await this.#requireEndpoint().watchReachability(afterGeneration, timeoutMs),
+ );
+ } catch (error) {
+ throw normalizePeerError(error);
+ }
+ }
+
signIdentity(payload: Buffer): Promise<RuntimeHostPeerIdentityProof> {
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<void>,
): Promise<void> {
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<void> {
+ 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<RuntimeHostPeerNativeReachabilitySnapshot>;
connect(options: {
readonly requestId: number;
readonly peerId: string;
@@ -103,6 +106,12 @@ export interface RuntimeHostPeerNativeEndpoint {
close(): Promise<void>;
}
+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' &&
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org