From 8b2212e40ac0d12bf08e4f5d4c9703c5b67dfb89 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Wed, 25 Feb 2026 18:08:49 +0700 Subject: [PATCH] fix: distribution stability --- crates/dashboard/src/lib.rs | 8 +- crates/distribution/src/iroh_driver.rs | 247 ++++++++++++++++-- crates/distribution/src/node.rs | 5 + crates/distribution/src/snapshot.rs | 19 ++ crates/distribution/src/swim/dissemination.rs | 9 + crates/distribution/src/swim/member_list.rs | 5 + crates/distribution/src/swim/node.rs | 15 ++ crates/node/src/main.rs | 239 ++++++++++++++++- crates/node/src/plugins/distribution.rs | 34 ++- .../node/src/plugins/distribution_page.html | 195 +++++++++++++- crates/node/src/plugins/peers.rs | 62 ++++- crates/simulation/src/distribution/sim.rs | 6 + crates/simulation/tests/cluster_scenarios.rs | 90 +++++++ .../tests/distribution_lifecycle.rs | 5 + .../tests/distribution_properties.rs | 2 + .../simulation/tests/distribution_registry.rs | 3 + .../simulation/tests/topology_adversarial.rs | 8 + tests/integration/Cargo.toml | 2 +- tests/integration/tests/stream_integration.rs | 2 + 19 files changed, 899 insertions(+), 57 deletions(-) diff --git a/crates/dashboard/src/lib.rs b/crates/dashboard/src/lib.rs index 4deb18e..948bd81 100644 --- a/crates/dashboard/src/lib.rs +++ b/crates/dashboard/src/lib.rs @@ -34,8 +34,12 @@ use crate::layer::{now_ms, DashboardLayer, EventStore}; use crate::plugin::PluginRegistry; use crate::trace::{RuntimeTrace, TimestampedStats}; -/// Peer info sent through the join channel: (public_key, optional_relay_url). -pub type JoinPeerInfo = ([u8; 32], Option); +/// Peer info sent through the join channel. +pub struct JoinPeerInfo { + pub node_id: [u8; 32], + pub relay_url: Option, + pub direct_addrs: Vec, +} /// Configuration for the runtime dashboard. #[derive(Debug, Clone)] diff --git a/crates/distribution/src/iroh_driver.rs b/crates/distribution/src/iroh_driver.rs index 13eb41d..8a9c6fc 100644 --- a/crates/distribution/src/iroh_driver.rs +++ b/crates/distribution/src/iroh_driver.rs @@ -8,8 +8,9 @@ //! (`tick()`, `recv()`, `join()`) to match the existing main loop pattern. use std::collections::HashMap; +use std::net::{IpAddr, SocketAddr}; use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::time::{Duration, Instant}; use iroh::endpoint::Connection; use iroh::{Endpoint, EndpointAddr, PublicKey, RelayMode, SecretKey}; @@ -62,6 +63,94 @@ struct JoinResult { conn: Connection, } +// ─── LAN IP Discovery ────────────────────────────────────────────────────── + +/// Discover all non-loopback LAN IP addresses on this host. +/// +/// Uses UDP socket tricks to multiple broadcast destinations to find +/// addresses across different subnets. Also parses `/proc/net/if_inet6` +/// for IPv6 addresses on Linux. +pub fn discover_lan_ips() -> Vec { + let mut ips = Vec::new(); + let mut seen = std::collections::HashSet::new(); + + // UDP socket trick: connect to a broadcast-ish address, read local_addr + let targets: &[&str] = &[ + "10.255.255.255:1", + "192.168.255.255:1", + "172.31.255.255:1", + ]; + for target in targets { + if let Ok(sock) = std::net::UdpSocket::bind("0.0.0.0:0") { + if sock.connect(target).is_ok() { + if let Ok(local) = sock.local_addr() { + let ip = local.ip(); + if !ip.is_loopback() && !ip.is_unspecified() && seen.insert(ip) { + ips.push(ip); + } + } + } + } + } + + // Parse /proc/net/if_inet6 for IPv6 addresses (Linux only) + if let Ok(contents) = std::fs::read_to_string("/proc/net/if_inet6") { + for line in contents.lines() { + let parts: Vec<&str> = line.split_whitespace().collect(); + if parts.len() >= 6 { + let hex = parts[0]; + if hex.len() == 32 { + let mut bytes = [0u8; 16]; + let mut valid = true; + for i in 0..16 { + match u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16) { + Ok(b) => bytes[i] = b, + Err(_) => { valid = false; break; } + } + } + if valid { + let ip = IpAddr::V6(std::net::Ipv6Addr::from(bytes)); + if !ip.is_loopback() && !ip.is_unspecified() { + // Skip link-local (fe80::) + if let IpAddr::V6(v6) = ip { + if (v6.segments()[0] & 0xffc0) == 0xfe80 { + continue; + } + } + if seen.insert(ip) { + ips.push(ip); + } + } + } + } + } + } + } + + ips +} + +// ─── Join Status ─────────────────────────────────────────────────────────── + +/// Phase of a join attempt. +#[derive(Debug, Clone)] +pub enum JoinPhase { + Connecting { attempt: u32, max_attempts: u32 }, + Sending { attempt: u32, max_attempts: u32 }, + Sent, + Failed { error: String }, +} + +/// Real-time status of a join attempt to a specific peer. +#[derive(Debug, Clone)] +pub struct JoinStatus { + pub phase: JoinPhase, + pub has_relay: bool, + pub has_direct: bool, + pub direct_addr_count: usize, + pub updated_at: Instant, +} + // ─── Driver ───────────────────────────────────────────────────────────────── /// iroh P2P network driver. @@ -82,6 +171,8 @@ pub struct IrohDriver { other_accepted_conns: Arc>>, /// Relay URLs learned from join seeds, used for reconnection. peer_relay_urls: HashMap, + /// Real-time join status for each peer being joined. + join_statuses: Arc>>, /// Embedded relay server (if started). #[cfg(feature = "relay")] relay_server: Option, @@ -211,6 +302,7 @@ impl IrohDriver { accepted_conns, other_accepted_conns, peer_relay_urls: HashMap::new(), + join_statuses: Arc::new(Mutex::new(HashMap::new())), #[cfg(feature = "relay")] relay_server, relay_url, @@ -240,29 +332,55 @@ impl IrohDriver { /// The endpoint's full address (public key + direct socket addresses). /// /// Constructs the address from the endpoint's public key and bound - /// sockets. Unspecified addresses (`0.0.0.0` / `[::]`) are mapped to - /// their loopback equivalents so peers on the same host can connect. + /// sockets. For sockets bound to `0.0.0.0`, emits one address per + /// discovered LAN IP so that peers on the same network can connect + /// directly. IPv6 unspecified is mapped to localhost. pub fn endpoint_addr(&self) -> EndpointAddr { - use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; - let key = PublicKey::from_bytes(&self.node.node_id().0) .expect("node_id is a valid public key"); let mut addr = EndpointAddr::new(key); - for sock in self.endpoint.bound_sockets() { - let resolved = match sock.ip() { - IpAddr::V4(ip) if ip.is_unspecified() => { - SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), sock.port()) - } - IpAddr::V6(ip) if ip.is_unspecified() => { - SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), sock.port()) - } - _ => sock, - }; - addr = addr.with_ip_addr(resolved); + for sa in self.direct_addresses() { + addr = addr.with_ip_addr(sa); } addr } + /// Compute direct socket addresses from bound sockets + LAN discovery. + /// + /// For sockets bound to `0.0.0.0`, emits one `SocketAddr` per discovered + /// LAN IP using the bound port. Specific-IP binds are kept as-is. + pub fn direct_addresses(&self) -> Vec { + let lan_ips = discover_lan_ips(); + let mut addrs = Vec::new(); + for sock in self.endpoint.bound_sockets() { + match sock.ip() { + IpAddr::V4(ip) if ip.is_unspecified() => { + // Emit one address per discovered LAN IP + for lip in &lan_ips { + if lip.is_ipv4() { + addrs.push(SocketAddr::new(*lip, sock.port())); + } + } + // Also include localhost for same-host connectivity + addrs.push(SocketAddr::new( + IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), + sock.port(), + )); + } + IpAddr::V6(ip) if ip.is_unspecified() => { + addrs.push(SocketAddr::new( + IpAddr::V6(std::net::Ipv6Addr::LOCALHOST), + sock.port(), + )); + } + _ => { + addrs.push(sock); + } + } + } + addrs + } + /// Access the underlying node (read-only). pub fn node(&self) -> &DistributedNode { &self.node @@ -284,6 +402,24 @@ impl IrohDriver { snap } + /// Get a snapshot of all join statuses. + pub fn join_statuses(&self) -> HashMap { + self.join_statuses.lock().unwrap().clone() + } + + /// Clear join statuses for the given node IDs (e.g. peers that are now alive). + pub fn clear_join_statuses(&self, node_ids: &[NodeId]) { + let mut map = self.join_statuses.lock().unwrap(); + for id in node_ids { + map.remove(id); + } + } + + /// Clear a single join status entry. + pub fn clear_join_status(&self, node_id: &NodeId) { + self.join_statuses.lock().unwrap().remove(node_id); + } + /// Join a cluster by connecting to seed nodes via iroh. /// /// Each seed is identified by its `EndpointAddr` (public key + optional @@ -297,7 +433,31 @@ impl IrohDriver { if let Some(relay) = seed_addr.relay_urls().next() { self.peer_relay_urls.insert(seed_node_id, relay.clone()); } - self.spawn_join_request(seed_addr.clone()); + // Clear any Dead entry so the JoinResponse can re-establish it. + // Without this, SWIM merge semantics reject Alive at the same + // incarnation when the local entry is Dead (Dead > Alive). + self.node.clear_dead_member(seed_node_id); + // Drop stale cached connection so iroh establishes a fresh one + self.connections.remove(&seed_node_id); + // Enrich the seed addr with a cached relay URL if it doesn't + // have one. The re-peer flow sends only a bare public key + // because metadata (including relay URL) is stripped when a + // node is declared dead. Without a relay URL iroh cannot + // reach the peer through NAT. + let enriched = if seed_addr.relay_urls().next().is_none() { + if let Some(relay) = self.peer_relay_urls.get(&seed_node_id).cloned() + .or_else(|| self.node.relay_url(&seed_node_id) + .and_then(|s| s.parse::().ok())) + .or_else(|| self.endpoint.addr().relay_urls().next().cloned()) + { + seed_addr.clone().with_relay_url(relay) + } else { + seed_addr.clone() + } + } else { + seed_addr.clone() + }; + self.spawn_join_request(enriched); } } @@ -310,11 +470,16 @@ impl IrohDriver { let endpoint = self.endpoint.clone(); let seed_node_id = NodeId(*seed_addr.id.as_bytes()); let pending = Arc::clone(&self.pending_joins); + let statuses = Arc::clone(&self.join_statuses); + + let has_relay = seed_addr.relay_urls().next().is_some(); + let direct_addr_count = seed_addr.ip_addrs().count(); + let has_direct = direct_addr_count > 0; self.rt.spawn(async move { let mut delay = Duration::from_secs(2); let max_delay = Duration::from_secs(30); - let max_attempts = 5; + let max_attempts: u32 = 5; for attempt in 1..=max_attempts { if attempt > 1 { @@ -322,6 +487,18 @@ impl IrohDriver { delay = (delay * 2).min(max_delay); } + // Update status: Connecting + { + let mut map = statuses.lock().unwrap(); + map.insert(seed_node_id, JoinStatus { + phase: JoinPhase::Connecting { attempt, max_attempts }, + has_relay, + has_direct, + direct_addr_count, + updated_at: Instant::now(), + }); + } + eprintln!("iroh driver: join attempt {attempt}/{max_attempts} connecting to {}...", seed_addr.id); let connect_result = tokio::time::timeout( Duration::from_secs(10), @@ -330,6 +507,18 @@ impl IrohDriver { match connect_result { Ok(Ok(conn)) => { + // Update status: Sending + { + let mut map = statuses.lock().unwrap(); + map.insert(seed_node_id, JoinStatus { + phase: JoinPhase::Sending { attempt, max_attempts }, + has_relay, + has_direct, + direct_addr_count, + updated_at: Instant::now(), + }); + } + eprintln!("iroh driver: join attempt {attempt}/{max_attempts} connected to {}, sending...", seed_addr.id); let send_result: Result<(), String> = async { let mut send = conn.open_uni().await.map_err(|e| e.to_string())?; @@ -345,6 +534,17 @@ impl IrohDriver { match send_result { Ok(()) => { eprintln!("iroh driver: join attempt {attempt}/{max_attempts} sent to {}", seed_addr.id); + // Update status: Sent + { + let mut map = statuses.lock().unwrap(); + map.insert(seed_node_id, JoinStatus { + phase: JoinPhase::Sent, + has_relay, + has_direct, + direct_addr_count, + updated_at: Instant::now(), + }); + } pending.lock().unwrap().push(JoinResult { node_id: seed_node_id, conn, @@ -376,6 +576,17 @@ impl IrohDriver { } } } + // Update status: Failed + { + let mut map = statuses.lock().unwrap(); + map.insert(seed_node_id, JoinStatus { + phase: JoinPhase::Failed { error: "all attempts exhausted".into() }, + has_relay, + has_direct, + direct_addr_count, + updated_at: Instant::now(), + }); + } eprintln!("iroh driver: join failed after {max_attempts} attempts to {}", seed_addr.id); }); } diff --git a/crates/distribution/src/node.rs b/crates/distribution/src/node.rs index 58758db..b6b2caa 100644 --- a/crates/distribution/src/node.rs +++ b/crates/distribution/src/node.rs @@ -198,6 +198,11 @@ impl DistributedNode { self.inject_piggyback(actions) } + /// Clear a Dead member so a subsequent JoinResponse can re-establish it. + pub fn clear_dead_member(&mut self, node_id: NodeId) { + self.swim.clear_dead_member(node_id); + } + pub fn handle_join_request(&mut self, from: NodeId) -> Vec { let actions = self.swim.handle_join_request(from); self.maybe_update_routing_table(from); diff --git a/crates/distribution/src/snapshot.rs b/crates/distribution/src/snapshot.rs index 9ce985a..71022e1 100644 --- a/crates/distribution/src/snapshot.rs +++ b/crates/distribution/src/snapshot.rs @@ -48,6 +48,20 @@ pub struct RegistryEntryInfo { pub tombstone: bool, } +/// Snapshot of a join attempt's real-time status. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JoinStatusInfo { + pub node_id: String, + /// "connecting", "sending", "sent", "failed" + pub phase: String, + /// E.g. "2/5" for attempt progress, or error message for failed + #[serde(skip_serializing_if = "Option::is_none")] + pub detail: Option, + pub has_relay: bool, + pub has_direct: bool, + pub direct_addr_count: usize, +} + /// Complete snapshot of a `DistributedNode`'s observable state. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct DistributionNodeSnapshot { @@ -121,6 +135,10 @@ pub struct DistributionNodeSnapshot { /// Build version string (e.g. "branch @ hash"). #[serde(skip_serializing_if = "Option::is_none", default)] pub version: Option, + + /// Real-time join statuses for peers being connected to. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub join_statuses: Vec, } fn node_id_hex(id: &NodeId) -> String { @@ -221,6 +239,7 @@ impl DistributedNode { invite_code: None, relay_url: self.metadata().relay_url(&self.node_id()).map(String::from), version: None, + join_statuses: Vec::new(), } } } diff --git a/crates/distribution/src/swim/dissemination.rs b/crates/distribution/src/swim/dissemination.rs index d4aaf5e..c07c0af 100644 --- a/crates/distribution/src/swim/dissemination.rs +++ b/crates/distribution/src/swim/dissemination.rs @@ -108,6 +108,15 @@ impl DisseminationQueue { self.entries.is_empty() } + /// Remove all pending updates for a given node. + /// + /// Called when clearing a Dead member before re-peering, so stale + /// `(node_id, Dead, incarnation)` gossip doesn't leak out and re-infect + /// the cluster. + pub fn purge_node(&mut self, node_id: &NodeId) { + self.entries.retain(|e| e.update.node_id != *node_id); + } + /// Compute the transmit budget: `Λ * ceil(log2(max(n, 2)))`. fn transmit_budget(&self, cluster_size: usize) -> usize { let n = cluster_size.max(2) as f64; diff --git a/crates/distribution/src/swim/member_list.rs b/crates/distribution/src/swim/member_list.rs index b7248a1..69a2315 100644 --- a/crates/distribution/src/swim/member_list.rs +++ b/crates/distribution/src/swim/member_list.rs @@ -94,6 +94,11 @@ impl MemberList { .count() } + /// Remove a member entry entirely. + pub fn remove(&mut self, node_id: &NodeId) -> bool { + self.members.remove(node_id).is_some() + } + /// Total members including dead. pub fn len(&self) -> usize { self.members.len() diff --git a/crates/distribution/src/swim/node.rs b/crates/distribution/src/swim/node.rs index d1d7830..f3b5217 100644 --- a/crates/distribution/src/swim/node.rs +++ b/crates/distribution/src/swim/node.rs @@ -67,6 +67,21 @@ impl SwimNode { &self.members } + /// Clear a Dead member entry so a subsequent JoinResponse can re-establish it. + /// + /// Used by the re-peer flow: a JoinResponse carries the remote node's + /// self-report as `(Alive, incarnation)`, but SWIM merge semantics reject + /// Alive at the same incarnation when the local entry is Dead. Removing + /// the stale Dead entry lets the fresh Alive record take effect. + pub fn clear_dead_member(&mut self, node_id: NodeId) { + if let Some(entry) = self.members.get(&node_id) { + if entry.state == MemberState::Dead { + self.members.remove(&node_id); + self.dissemination.purge_node(&node_id); + } + } + } + /// Recent probe targets from the SWIM probe cycle. pub fn recent_probe_targets(&self) -> &std::collections::VecDeque { self.probe.recent_probe_targets() diff --git a/crates/node/src/main.rs b/crates/node/src/main.rs index 602db31..d9afd78 100644 --- a/crates/node/src/main.rs +++ b/crates/node/src/main.rs @@ -176,6 +176,9 @@ fn generate_default_config(config_dir: &std::path::Path) -> std::path::PathBuf { }); } + // Auto-detect public IP for relay_hosts + let relay_hosts_line = detect_public_ip_for_config(); + // Write default config with absolute paths let contents = format!( r#"transport = "iroh" @@ -187,8 +190,9 @@ auth = true auth_dir = "{dir}/auth" relay = true relay_port = 3340 -"#, +{relay_hosts}"#, dir = config_dir.display(), + relay_hosts = relay_hosts_line, ); std::fs::write(&config_path, &contents).unwrap_or_else(|e| { eprintln!("Failed to write {}: {e}", config_path.display()); @@ -208,6 +212,41 @@ relay_port = 3340 config_path } +/// Detect outbound IP; if public, return a `relay_hosts = [""]` TOML line. +fn detect_public_ip_for_config() -> String { + let public_ip = (|| -> Option { + let sock = std::net::UdpSocket::bind("0.0.0.0:0").ok()?; + sock.connect("192.0.2.1:80").ok()?; // RFC 5737 TEST-NET-1 + let ip = sock.local_addr().ok()?.ip(); + match ip { + std::net::IpAddr::V4(v4) => { + if !v4.is_loopback() && !v4.is_private() + && !v4.is_link_local() && !v4.is_unspecified() + { + Some(ip) + } else { + None + } + } + std::net::IpAddr::V6(v6) => { + if !v6.is_loopback() && !v6.is_unspecified() { + Some(ip) + } else { + None + } + } + } + })(); + + match public_ip { + Some(ip) => { + eprintln!("Detected public IP {ip} — adding to relay_hosts"); + format!("relay_hosts = [\"{ip}\"]") + } + None => String::new(), + } +} + fn main() { let args = Args::parse(); @@ -360,11 +399,18 @@ fn main() { return; } Some(Subcmd::Invite) => { - println!("{invite_code}"); + let rich = if let Some(host) = relay_hosts.first() { + format!("{invite_code}@http://{host}:{relay_port}/") + } else { + invite_code.clone() + }; + println!("{rich}"); return; } Some(Subcmd::Join { code }) => { - let peer_bytes = base58_decode(code).unwrap_or_else(|| { + // Parse rich invite code: #@ + let (node_id_str, _direct_addrs_str, relay_str) = parse_rich_invite(code); + let peer_bytes = base58_decode(&node_id_str).unwrap_or_else(|| { eprintln!("Invalid invite code (expected base58-encoded 32-byte key)"); std::process::exit(1); }); @@ -389,9 +435,17 @@ fn main() { std::process::exit(1); }); - // Persist seed_node_id into config so the next startup auto-joins + // Persist seed_node_id and relay host into config if let Some(config_path) = &resolved_config_path { persist_config_key(config_path, "seed_node_id", &peer_hex); + + // Extract relay host from invite URL and save to config + if let Some(ref relay_url) = relay_str { + if let Some(host) = extract_relay_host(relay_url) { + persist_config_array_key(config_path, "relay_hosts", &[&host]); + eprintln!("Relay host saved: {host}"); + } + } } eprintln!("Peer added: {} ({})", code, &peer_hex[..8]); @@ -643,7 +697,12 @@ fn run_iroh( }) .collect(); if urls.is_empty() { - RelayMode::Disabled + if relay_enabled { + eprintln!("Relay: no hosts configured, using iroh default relays"); + RelayMode::Default + } else { + RelayMode::Disabled + } } else { eprintln!("Relay: using {} known relay(s)", urls.len()); RelayMode::Custom(urls.into_iter().collect::()) @@ -725,6 +784,7 @@ fn run_iroh( Arc::clone(&cached_snapshot), Some(join_tx_dist), ); + let dismissed_statuses = dist_plugin.dismissed_statuses(); dash.register_plugin(Arc::new(dist_plugin)); // Start dashboard HTTP on IrohDriver's tokio runtime @@ -760,7 +820,7 @@ fn run_iroh( // Drain discovered peers (dashboard "Add Peer") and auto-join them { - let mut new_peers = Vec::new(); + let mut new_peers: Vec = Vec::new(); while let Ok(info) = join_rx.try_recv() { new_peers.push(info); } @@ -768,22 +828,25 @@ fn run_iroh( let own_id = driver.node_id().0; let addrs: Vec = new_peers .iter() - .filter(|(bytes, _)| *bytes != own_id) - .filter_map(|(bytes, relay_url)| { - iroh::PublicKey::from_bytes(bytes).ok().map(|k| { + .filter(|info| info.node_id != own_id) + .filter_map(|info| { + iroh::PublicKey::from_bytes(&info.node_id).ok().map(|k| { let mut addr = iroh::EndpointAddr::from(k); - if let Some(url_str) = relay_url { + if let Some(url_str) = &info.relay_url { match url_str.parse::() { Ok(url) => { - eprintln!("Auto-joining peer {} via relay {}", base58_encode(bytes), url); + eprintln!("Auto-joining peer {} via relay {}", base58_encode(&info.node_id), url); addr = addr.with_relay_url(url); } Err(e) => { - eprintln!("Auto-joining peer {} (bad relay URL {}: {e})", base58_encode(bytes), url_str); + eprintln!("Auto-joining peer {} (bad relay URL {}: {e})", base58_encode(&info.node_id), url_str); } } } else { - eprintln!("Auto-joining peer {} (no relay URL)", base58_encode(bytes)); + eprintln!("Auto-joining peer {} (no relay URL)", base58_encode(&info.node_id)); + } + for sa in &info.direct_addrs { + addr = addr.with_ip_addr(*sa); } addr }) @@ -809,7 +872,82 @@ fn run_iroh( let mut snap = driver.snapshot(); snap.node_name = Some(node_name.clone()); - snap.invite_code = Some(invite_code.clone()); + + // Build rich invite code: #,@ + { + let direct_addrs = driver.direct_addresses(); + let addrs_part = if direct_addrs.is_empty() { + String::new() + } else { + let addrs_str: Vec = direct_addrs.iter().map(|a| a.to_string()).collect(); + format!("#{}", addrs_str.join(",")) + }; + let relay_part = match &snap.relay_url { + Some(relay) => format!("@{}", relay), + None => String::new(), + }; + snap.invite_code = Some(format!("{}{}{}", invite_code, addrs_part, relay_part)); + } + + // Drain dismissed join statuses from the dashboard + { + let mut dismissed = dismissed_statuses.lock().unwrap(); + for bytes in dismissed.drain(..) { + driver.clear_join_status(&swactor::transport::NodeId(bytes)); + } + } + + // Populate join statuses, auto-clearing alive peers + { + use distribution::iroh_driver::JoinPhase; + use distribution::snapshot::JoinStatusInfo; + + let statuses = driver.join_statuses(); + let alive_node_ids: Vec = snap.members.iter() + .filter(|m| m.state == "alive") + .filter_map(|m| { + let mut bytes = [0u8; 32]; + if m.node_id.len() == 64 { + for i in 0..32 { + bytes[i] = u8::from_str_radix(&m.node_id[i*2..i*2+2], 16).unwrap_or(0); + } + Some(swactor::transport::NodeId(bytes)) + } else { + None + } + }) + .collect(); + + // Clear statuses for alive peers + if !alive_node_ids.is_empty() { + driver.clear_join_statuses(&alive_node_ids); + } + + // Convert remaining statuses to snapshot format + snap.join_statuses = statuses.iter() + .filter(|(nid, _)| !alive_node_ids.contains(nid)) + .map(|(nid, status)| { + let node_id_hex: String = nid.0.iter().map(|b| format!("{:02x}", b)).collect(); + let (phase_str, detail) = match &status.phase { + JoinPhase::Connecting { attempt, max_attempts } => + ("connecting".into(), Some(format!("{}/{}", attempt, max_attempts))), + JoinPhase::Sending { attempt, max_attempts } => + ("sending".into(), Some(format!("{}/{}", attempt, max_attempts))), + JoinPhase::Sent => ("sent".into(), None), + JoinPhase::Failed { error } => ("failed".into(), Some(error.clone())), + }; + JoinStatusInfo { + node_id: node_id_hex, + phase: phase_str, + detail, + has_relay: status.has_relay, + has_direct: status.has_direct, + direct_addr_count: status.direct_addr_count, + } + }) + .collect(); + } + snap.version = Some(VERSION.to_string()); *cached_snapshot.lock().unwrap() = Some(snap); @@ -923,6 +1061,79 @@ fn persist_config_key(path: &std::path::Path, key: &str, value: &str) { } } +/// Extract hostname from a relay URL like `http://167.71.x.x:3340/`. +fn extract_relay_host(url: &str) -> Option { + let stripped = url.strip_prefix("http://") + .or_else(|| url.strip_prefix("https://"))?; + let host_port = stripped.trim_end_matches('/'); + // Handle bracket-enclosed IPv6: [::1]:3340 + if host_port.starts_with('[') { + let end = host_port.find(']')?; + Some(host_port[1..end].to_string()) + } else { + let host = match host_port.rfind(':') { + Some(idx) => &host_port[..idx], + None => host_port, + }; + if host.is_empty() { None } else { Some(host.to_string()) } + } +} + +/// Persist a TOML array key into an existing config file. +fn persist_config_array_key(path: &std::path::Path, key: &str, values: &[&str]) { + let contents = std::fs::read_to_string(path).unwrap_or_default(); + let array_str = values + .iter() + .map(|v| format!("\"{v}\"")) + .collect::>() + .join(", "); + let new_line = format!("{key} = [{array_str}]"); + + let updated = if contents.contains(key) { + contents + .lines() + .map(|line| { + if line.trim_start().starts_with(key) { + new_line.as_str() + } else { + line + } + }) + .collect::>() + .join("\n") + + "\n" + } else { + let mut s = contents; + if !s.ends_with('\n') && !s.is_empty() { + s.push('\n'); + } + s.push_str(&new_line); + s.push('\n'); + s + }; + + if let Err(e) = std::fs::write(path, &updated) { + eprintln!("Warning: could not persist {key} to {}: {e}", path.display()); + } +} + +/// Parse a rich invite code: `#,@` +/// +/// Returns (node_id_str, optional_direct_addrs_csv, optional_relay_url). +fn parse_rich_invite(raw: &str) -> (String, Option, Option) { + // Split on last '@' for relay + let (left, relay) = match raw.rfind('@') { + Some(idx) => (&raw[..idx], Some(raw[idx + 1..].to_string())), + None => (raw, None), + }; + // Split on '#' for direct addrs + let (node_id, addrs) = match left.find('#') { + Some(idx) => (&left[..idx], Some(left[idx + 1..].to_string())), + None => (left, None), + }; + (node_id.to_string(), addrs, relay) +} + /// Parse a node ID from either hex (64 chars) or base58 (~44 chars). fn parse_node_id_str(s: &str) -> Option<[u8; 32]> { if s.len() == 64 { diff --git a/crates/node/src/plugins/distribution.rs b/crates/node/src/plugins/distribution.rs index 7adfea7..184f4c6 100644 --- a/crates/node/src/plugins/distribution.rs +++ b/crates/node/src/plugins/distribution.rs @@ -17,6 +17,9 @@ const DISTRIBUTION_HTML: &str = include_str!("distribution_page.html"); pub struct DistributionPlugin { cached: Arc>>, join_sender: Option>, + /// Node IDs whose join status has been dismissed by the user. + /// The main loop drains these and calls `clear_join_status` on the driver. + dismissed_statuses: Arc>>, } impl DistributionPlugin { @@ -24,7 +27,16 @@ impl DistributionPlugin { cached: Arc>>, join_sender: Option>, ) -> Self { - Self { cached, join_sender } + Self { + cached, + join_sender, + dismissed_statuses: Arc::new(Mutex::new(Vec::new())), + } + } + + /// Get a clone of the dismissed-statuses Arc for the main loop to drain. + pub fn dismissed_statuses(&self) -> Arc>> { + Arc::clone(&self.dismissed_statuses) } } @@ -58,6 +70,7 @@ impl DashboardPlugin for DistributionPlugin { } } ("POST", "rejoin") => self.handle_rejoin(body), + ("POST", "clear_status") => self.handle_clear_status(body), _ => PluginResponse::not_found(), } } @@ -100,11 +113,28 @@ impl DistributionPlugin { }) }; - match tx.send((bytes, relay_url)) { + match tx.send(JoinPeerInfo { node_id: bytes, relay_url, direct_addrs: vec![] }) { Ok(()) => PluginResponse::json(r#"{"ok":true}"#.into()), Err(_) => PluginResponse::json(r#"{"error":"channel closed"}"#.into()), } } + + fn handle_clear_status(&self, body: &[u8]) -> PluginResponse { + let parsed: serde_json::Value = match serde_json::from_slice(body) { + Ok(v) => v, + Err(_) => return PluginResponse::json(r#"{"error":"invalid json"}"#.into()), + }; + let node_id_hex = match parsed.get("node_id").and_then(|v| v.as_str()) { + Some(s) => s, + None => return PluginResponse::json(r#"{"error":"missing node_id"}"#.into()), + }; + let bytes = match parse_hex_node_id(node_id_hex) { + Some(b) => b, + None => return PluginResponse::json(r#"{"error":"invalid node_id hex"}"#.into()), + }; + self.dismissed_statuses.lock().unwrap().push(bytes); + PluginResponse::json(r#"{"ok":true}"#.into()) + } } fn parse_hex_node_id(hex: &str) -> Option<[u8; 32]> { diff --git a/crates/node/src/plugins/distribution_page.html b/crates/node/src/plugins/distribution_page.html index 1e6ba0d..843c843 100644 --- a/crates/node/src/plugins/distribution_page.html +++ b/crates/node/src/plugins/distribution_page.html @@ -263,10 +263,11 @@
- +
+
@@ -894,37 +895,177 @@ memberMap[data.members[i].node_id] = data.members[i].state; } } + // Build join status map from SSE data + var joinStatusMap = {}; + if (data && data.join_statuses) { + for (var i = 0; i < data.join_statuses.length; i++) { + var js = data.join_statuses[i]; + joinStatusMap[js.node_id] = js; + } + } for (var i = 0; i < d.peers.length; i++) { var p = d.peers[i]; var state = memberMap[p.node_id] || 'offline'; - var color = state === 'alive' ? '#4caf50' : state === 'suspect' ? '#ff9800' : '#555'; + var js = joinStatusMap[p.node_id]; + var color, statusText; + + if (js && state !== 'alive') { + // Show join status inline + var methodParts = []; + if (js.has_relay) methodParts.push('relay'); + if (js.has_direct) methodParts.push(js.direct_addr_count + ' direct'); + var methodStr = methodParts.length > 0 ? ' via ' + methodParts.join(' + ') : ''; + + if (js.phase === 'connecting') { + color = '#6366f1'; + statusText = 'connecting' + methodStr + ' (' + (js.detail || '') + ')…'; + } else if (js.phase === 'sending') { + color = '#6366f1'; + statusText = 'sending' + methodStr + ' (' + (js.detail || '') + ')…'; + } else if (js.phase === 'sent') { + color = '#ff9800'; + statusText = 'join sent, waiting…'; + } else if (js.phase === 'failed') { + color = '#f44336'; + statusText = 'failed: ' + (js.detail || 'unknown error'); + } else { + color = '#555'; + statusText = js.phase; + } + } else { + color = state === 'alive' ? '#4caf50' : state === 'suspect' ? '#ff9800' : '#555'; + statusText = state; + } + var div = document.createElement('div'); - div.style.cssText = 'display:flex;align-items:center;gap:6px;padding:2px 0;font-size:10px;'; + div.style.cssText = 'display:flex;align-items:center;gap:6px;padding:2px 0;font-size:10px;flex-wrap:wrap;'; + + var statusSpan = '' + statusText + ''; + + var actionBtns = ''; + if (js && js.phase === 'failed') { + actionBtns = + '' + + ''; + } + div.innerHTML = - '' + + '' + '' + p.node_id.substring(0, 16) + '\u2026' + '' + (p.label || '') + '' + - '' + state + '' + + statusSpan + actionBtns + ''; list.appendChild(div); } } window.addPeer = function() { - var nid = document.getElementById('peerNodeId').value.trim(); + var raw = document.getElementById('peerNodeId').value.trim(); var label = document.getElementById('peerLabel').value.trim(); - if (!nid) return; + if (!raw) return; + var status = document.getElementById('peerAddStatus'); + + // Parse rich invite code: #,@ + var atIdx = raw.lastIndexOf('@'); + var relay = atIdx >= 0 ? raw.substring(atIdx + 1) : null; + var left = atIdx >= 0 ? raw.substring(0, atIdx) : raw; + var hashIdx = left.indexOf('#'); + var nid = hashIdx >= 0 ? left.substring(0, hashIdx) : left; + var addrs = hashIdx >= 0 ? left.substring(hashIdx + 1) : null; + + var payload = { node_id: nid, label: label }; + if (relay) payload.relay_url = relay; + if (addrs) payload.direct_addrs = addrs; + + var methods = []; + if (relay) methods.push('relay'); + if (addrs) methods.push(addrs.split(',').length + ' direct'); + var methodStr = methods.length > 0 ? methods.join(' + ') : 'no relay or direct addrs'; + status.textContent = 'Connecting via ' + methodStr + '\u2026'; + status.style.color = '#6366f1'; + fetch('/api/plugin/peers/add', { method: 'POST', headers: { 'Content-Type': 'application/json', 'Authorization': 'Bearer ' + dashToken }, - body: JSON.stringify({ node_id: nid, label: label }) - }).then(function() { + body: JSON.stringify(payload) + }).then(function(r) { return r.json(); }).then(function(d) { document.getElementById('peerNodeId').value = ''; document.getElementById('peerLabel').value = ''; + var parts = []; + if (d.has_relay) parts.push('relay'); + if (d.has_direct) parts.push(d.direct_count + ' direct addr' + (d.direct_count > 1 ? 's' : '')); + if (parts.length > 0) { + status.textContent = 'Added \u2014 connecting via ' + parts.join(' + '); + status.style.color = '#4caf50'; + } else { + status.textContent = 'Added \u2014 no relay or direct addrs (may not connect)'; + status.style.color = '#ff9800'; + } + setTimeout(function() { status.textContent = ''; }, 8000); fetchPeers(); + }).catch(function() { + status.textContent = 'Failed to add peer'; + status.style.color = '#f44336'; + setTimeout(function() { status.textContent = ''; }, 8000); }); }; + window.showJoinDetail = function(nid) { + // Find join status from data + if (!data || !data.join_statuses) return; + var js = null; + for (var i = 0; i < data.join_statuses.length; i++) { + if (data.join_statuses[i].node_id === nid) { js = data.join_statuses[i]; break; } + } + if (!js) return; + + // Remove any existing popup + var existing = document.getElementById('joinDetailPopup'); + if (existing) existing.remove(); + + var popup = document.createElement('div'); + popup.id = 'joinDetailPopup'; + popup.style.cssText = 'position:fixed;top:50%;left:50%;transform:translate(-50%,-50%);background:#1c1f2e;border:1px solid #2a2d3e;border-radius:6px;padding:16px;z-index:2000;min-width:280px;max-width:90vw;font-size:11px;color:#e0e0e0;box-shadow:0 4px 20px rgba(0,0,0,0.5);'; + + var methodLines = []; + methodLines.push('Relay: ' + (js.has_relay ? 'yes' : 'no')); + methodLines.push('Direct: ' + (js.has_direct ? js.direct_addr_count + ' addr(s)' : 'none')); + methodLines.push('Phase: ' + js.phase + (js.detail ? ' (' + js.detail + ')' : '')); + + popup.innerHTML = + '
' + + 'Join Status Detail' + + '' + + '
' + + '
Node: ' + nid.substring(0, 16) + '\u2026
' + + '
' + + methodLines.join('
') + + '
'; + + document.body.appendChild(popup); + + // Click outside to dismiss + function dismissPopup(e) { + if (!popup.contains(e.target)) { + popup.remove(); + document.removeEventListener('mousedown', dismissPopup); + } + } + setTimeout(function() { document.addEventListener('mousedown', dismissPopup); }, 10); + }; + + window.dismissJoinStatus = function(nid) { + fetch('/api/plugin/distribution/clear_status', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ node_id: nid }) + }).then(function(r) { return r.json(); }).then(function(d) { + if (d.ok) fetchPeers(); + }).catch(function() {}); + }; + window.removePeer = function(nid) { fetch('/api/plugin/peers/remove', { method: 'POST', @@ -972,10 +1113,42 @@ }); window.copyInvite = function() { - navigator.clipboard.writeText(currentInvite).then(function() { + function onSuccess() { document.getElementById('copyBtn').textContent = 'Copied!'; setTimeout(function() { document.getElementById('copyBtn').textContent = 'Copy'; }, 2000); - }); + } + function fallbackSelect() { + var inp = document.getElementById('inviteInput'); + inp.select(); + inp.setSelectionRange(0, inp.value.length); + } + if (navigator.clipboard && navigator.clipboard.writeText) { + navigator.clipboard.writeText(currentInvite).then(onSuccess).catch(function() { + try { + var ta = document.createElement('textarea'); + ta.value = currentInvite; + ta.style.position = 'fixed'; + ta.style.opacity = '0'; + document.body.appendChild(ta); + ta.select(); + var ok = document.execCommand('copy'); + document.body.removeChild(ta); + if (ok) { onSuccess(); } else { fallbackSelect(); } + } catch(e) { fallbackSelect(); } + }); + } else { + try { + var ta = document.createElement('textarea'); + ta.value = currentInvite; + ta.style.position = 'fixed'; + ta.style.opacity = '0'; + document.body.appendChild(ta); + ta.select(); + var ok = document.execCommand('copy'); + document.body.removeChild(ta); + if (ok) { onSuccess(); } else { fallbackSelect(); } + } catch(e) { fallbackSelect(); } + } }; // ── Minimal QR code renderer (Mode 2 alphanumeric, version auto) ── diff --git a/crates/node/src/plugins/peers.rs b/crates/node/src/plugins/peers.rs index b363529..aeb41af 100644 --- a/crates/node/src/plugins/peers.rs +++ b/crates/node/src/plugins/peers.rs @@ -108,7 +108,7 @@ impl PeersPlugin { Err(e) => return PluginResponse::error(400, format!("invalid JSON: {e}")), }; - let node_id_str = match parsed.get("node_id").and_then(|v| v.as_str()) { + let raw_node_id = match parsed.get("node_id").and_then(|v| v.as_str()) { Some(s) => s, None => return PluginResponse::error(400, "missing node_id field"), }; @@ -118,6 +118,30 @@ impl PeersPlugin { .unwrap_or("") .to_string(); + // Parse rich invite code: #,@ + // Split on last '@' for relay, then '#' for direct addrs + let (left, invite_relay_url) = match raw_node_id.rfind('@') { + Some(idx) => (&raw_node_id[..idx], Some(raw_node_id[idx + 1..].to_string())), + None => (raw_node_id, None), + }; + let (node_id_str, invite_addrs_str) = match left.find('#') { + Some(idx) => (&left[..idx], Some(&left[idx + 1..])), + None => (left, None), + }; + + // Parse direct addrs from invite code or explicit body field + let direct_addrs_str = parsed + .get("direct_addrs") + .and_then(|v| v.as_str()) + .or(invite_addrs_str); + let direct_addrs: Vec = direct_addrs_str + .map(|s| { + s.split(',') + .filter_map(|a| a.trim().parse::().ok()) + .collect() + }) + .unwrap_or_default(); + let bytes: [u8; 32] = if let Some(b) = hex_decode(node_id_str) { match b.try_into() { Ok(arr) => arr, @@ -131,10 +155,12 @@ impl PeersPlugin { return PluginResponse::error(400, "invalid node_id (expected 64-char hex or base58)"); }; + // Explicit relay_url field takes precedence, then invite code's @relay let relay_url = parsed .get("relay_url") .and_then(|v| v.as_str()) - .map(|s| s.to_string()); + .map(|s| s.to_string()) + .or(invite_relay_url); let node_id = NodeId(bytes); let mut list = self.peer_auth.lock().unwrap(); @@ -145,11 +171,21 @@ impl PeersPlugin { drop(list); // Trigger a SWIM join for the newly added peer + let has_direct = !direct_addrs.is_empty(); + let direct_count = direct_addrs.len(); if let Some(tx) = &self.join_sender { - let _ = tx.send((bytes, relay_url)); + let _ = tx.send(JoinPeerInfo { + node_id: bytes, + relay_url: relay_url.clone(), + direct_addrs, + }); } - PluginResponse::json(r#"{"ok":true}"#.to_string()) + let has_relay = relay_url.is_some(); + let node_id_hex: String = bytes.iter().map(|b| format!("{:02x}", b)).collect(); + PluginResponse::json(format!( + r#"{{"ok":true,"has_relay":{has_relay},"has_direct":{has_direct},"direct_count":{direct_count},"node_id":"{node_id_hex}"}}"# + )) } fn handle_sync(&self, body: &[u8]) -> PluginResponse { @@ -236,7 +272,11 @@ impl PeersPlugin { }); if let Some(tx) = &self.join_sender { - let _ = tx.send((bytes, relay_url)); + let _ = tx.send(JoinPeerInfo { + node_id: bytes, + relay_url, + direct_addrs: vec![], + }); } } } @@ -261,11 +301,15 @@ impl PeersPlugin { None => return PluginResponse::error(400, "missing node_id field"), }; - let bytes = match hex_decode(node_id_hex) { - Some(b) if b.len() == 32 => b, - _ => { - return PluginResponse::error(400, "invalid node_id hex (must be 64 hex chars)"); + let bytes: Vec = if let Some(b) = hex_decode(node_id_hex) { + if b.len() != 32 { + return PluginResponse::error(400, "invalid node_id (hex decoded to wrong length)"); } + b + } else if let Some(arr) = base58_decode(node_id_hex) { + arr.to_vec() + } else { + return PluginResponse::error(400, "invalid node_id (expected 64-char hex or base58)"); }; let node_id = NodeId(bytes.try_into().unwrap()); diff --git a/crates/simulation/src/distribution/sim.rs b/crates/simulation/src/distribution/sim.rs index 7184aab..b0d5db3 100644 --- a/crates/simulation/src/distribution/sim.rs +++ b/crates/simulation/src/distribution/sim.rs @@ -69,6 +69,7 @@ impl Default for DistributionSimConfig { indirect_probes: 1, suspicion_timeout: 5, dead_reprobe_interval: 10, + probe_mode: distribution::swim::probe::ProbeMode::Periodic, }, actors_per_node: 2, kill_schedule: Vec::new(), @@ -365,6 +366,11 @@ fn run_simulation_inner(config: DistributionSimConfig) -> (DistTrace, Vec