zacheryasc/swactor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: reduce idle cpu, gossip noise, stability #51

Merged
zacheryasc merged 2 commits from idle-cpu into master 2026-02-25 11:11:05 +00:00
Conversation 0 Commits 2 Files changed 44 +1340 -993
44 changed files with 1340 additions and 993 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -1334,7 +1334,6 @@ dependencies = [
"serde",
"serde_json",
"swactor",
"swactor-transport",
"tokio",
]

8
benches/mt_benchmarks.rs
View file

@ -3,7 +3,7 @@ use std::time::{Duration, Instant};
use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
use swactor::{
actor::{ActorAddress, ActorInterface},
config::{BackoffPolicy, RuntimeConfig},
config::RuntimeConfig,
runtime::{Ctx, Runtime},
};
@ -16,12 +16,6 @@ fn mt_config(threads: usize, max_actors: usize, max_messages: usize) -> RuntimeC
num_threads: threads,
max_actors,
channel_buffer_size: max_messages,
backoff_policy: BackoffPolicy {
spin_threshold: 32,
yield_threshold: 64,
sleep_increment_us: 10,
sleep_max_us: 100,
},
..Default::default()
}
}

28
crates/bindings/python/src/lib.rs
View file

@ -5,7 +5,7 @@ use pyo3::prelude::*;
use pyo3::types::PyModule;
use ::swactor::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Ctx, Environment, SpawnRequest};
use ::swactor::config::{BackoffPolicy, RuntimeConfig};
use ::swactor::config::RuntimeConfig;
use ::swactor::runtime::{Inbox, Runtime, RuntimeHandle};
// ─── PyMsg newtype ───────────────────────────────────────────────────────────
@ -231,14 +231,6 @@ pub struct PyRuntimeConfig {
max_actors: usize,
#[pyo3(get, set)]
channel_buffer_size: usize,
#[pyo3(get, set)]
spin_threshold: u32,
#[pyo3(get, set)]
yield_threshold: u32,
#[pyo3(get, set)]
sleep_increment_us: u64,
#[pyo3(get, set)]
sleep_max_us: u64,
}
#[pymethods]
@ -249,28 +241,16 @@ impl PyRuntimeConfig {
num_threads = 1,
max_actors = 1_000,
channel_buffer_size = 1_000,
spin_threshold = 64,
yield_threshold = 256,
sleep_increment_us = 50,
sleep_max_us = 1_000,
))]
fn new(
num_threads: usize,
max_actors: usize,
channel_buffer_size: usize,
spin_threshold: u32,
yield_threshold: u32,
sleep_increment_us: u64,
sleep_max_us: u64,
) -> Self {
Self {
num_threads,
max_actors,
channel_buffer_size,
spin_threshold,
yield_threshold,
sleep_increment_us,
sleep_max_us,
}
}
}
@ -281,12 +261,6 @@ impl From<PyRuntimeConfig> for RuntimeConfig {
num_threads: py.num_threads,
max_actors: py.max_actors,
channel_buffer_size: py.channel_buffer_size,
backoff_policy: BackoffPolicy {
spin_threshold: py.spin_threshold,
yield_threshold: py.yield_threshold,
sleep_increment_us: py.sleep_increment_us,
sleep_max_us: py.sleep_max_us,
},
..Default::default()
}
}

1
crates/dashboard/src/actor_detail_html.rs
View file

@ -392,6 +392,7 @@ pub const ACTOR_DETAIL_HTML: &str = r##"<!DOCTYPE html>
// ─── SSE ────────────────────────────────────────────────────
var es = new EventSource('/events');
window.addEventListener('beforeunload', function() { es.close(); });
es.addEventListener('stats', function(e) {
try { updateDetail(JSON.parse(e.data)); } catch(err) { console.error(err); }

1
crates/dashboard/src/actors_html.rs
View file

@ -721,6 +721,7 @@ pub const ACTORS_HTML: &str = r##"<!DOCTYPE html>
// ── SSE connection ─────────────────────────────────────
var es = new EventSource('/events');
window.addEventListener('beforeunload', function() { es.close(); });
es.addEventListener('stats', function(e) {
try {

1
crates/dashboard/src/dashboard_html.rs
View file

@ -578,6 +578,7 @@ pub const DASHBOARD_HTML: &str = r##"<!DOCTYPE html>
}
var es = new EventSource('/events');
window.addEventListener('beforeunload', function() { es.close(); });
es.addEventListener('stats', function(e) {
try {

8
crates/dashboard/src/lib.rs
View file

@ -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<String>);
/// Peer info sent through the join channel.
pub struct JoinPeerInfo {
pub node_id: [u8; 32],
pub relay_url: Option<String>,
pub direct_addrs: Vec<std::net::SocketAddr>,
}
/// Configuration for the runtime dashboard.
#[derive(Debug, Clone)]

1
crates/dashboard/src/topology_html.rs
View file

@ -263,6 +263,7 @@ pub const TOPOLOGY_HTML: &str = r##"<!DOCTYPE html>
tick();
var es = new EventSource('/events');
window.addEventListener('beforeunload', function() { es.close(); });
es.addEventListener('topology', function(e) {
try { updateTopology(JSON.parse(e.data)); } catch(err) { console.error(err); }

4
crates/distribution/Cargo.toml
View file

@ -4,14 +4,12 @@ version = "0.1.0"
edition = "2024"
[features]
default = ["tcp"]
tcp = ["dep:swactor-transport", "swactor-transport/tcp"]
default = []
iroh = ["dep:iroh", "dep:tokio"]
relay = ["iroh", "dep:iroh-relay"]
[dependencies]
swactor = { path = "../..", features = ["serde", "transport"] }
swactor-transport = { path = "../transport", optional = true }
ed25519-dalek = { version = "2", features = ["rand_core"] }
rand_core = { version = "0.6", features = ["getrandom"] }
serde = { version = "1", features = ["derive"] }

496
crates/distribution/src/driver.rs
View file

@ -1,496 +0,0 @@
//! Network driver — bridges `DistributedNode` logic with TCP I/O.
//!
//! Translates outgoing `NodeAction`s into wire messages sent via `TcpTransport`,
//! and dispatches incoming wire messages to the appropriate `DistributedNode`
//! handler methods.
//!
//! The driver maintains a `PeerAddressBook` mapping `NodeId → SocketAddr`.
//! Address hints travel in the TCP wire frame (not in protocol messages),
//! keeping the protocol layer transport-agnostic.
use std::collections::HashMap;
use std::net::{SocketAddr, TcpStream};
use serde::{Deserialize, Serialize};
use swactor::actor::ActorAddress;
use swactor::transport::{NetworkMessage, WireEnvelope};
use crate::messages::*;
use crate::node::{DistributedNode, DistributedNodeConfig};
use crate::snapshot::DistributionNodeSnapshot;
use crate::swim::node::NodeAction;
use swactor_transport::tcp::{TcpAcceptor, TcpTransport, encode_wire_envelope_with_hints};
use crate::types::NodeId;
/// Dummy destination address used in wire envelopes for SWIM protocol messages.
/// SWIM messages are routed by `SocketAddr`, not by `ActorAddress`, so this
/// field is unused but required by the wire format.
const SWIM_DEST: ActorAddress = ActorAddress([0u8; 32]);
// ─── Address Hints ──────────────────────────────────────────────────────────
/// An address hint bundled in the TCP wire frame.
///
/// Each outgoing TCP message includes the sender's own (NodeId, SocketAddr)
/// as a hint. JoinResponse messages include all known member addresses.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AddressHint {
pub node_id: NodeId,
pub addr: SocketAddr,
}
// ─── Peer Address Book ──────────────────────────────────────────────────────
/// Maps NodeId → SocketAddr. Maintained by the TCP driver layer.
pub struct PeerAddressBook(HashMap<NodeId, SocketAddr>);
impl Default for PeerAddressBook {
fn default() -> Self {
Self::new()
}
}
impl PeerAddressBook {
pub fn new() -> Self {
Self(HashMap::new())
}
/// Learn a node's address from a hint.
pub fn learn(&mut self, node_id: NodeId, addr: SocketAddr) {
self.0.insert(node_id, addr);
}
/// Resolve a node's address.
pub fn resolve(&self, node_id: &NodeId) -> Option<SocketAddr> {
self.0.get(node_id).copied()
}
/// Learn multiple hints at once.
pub fn bulk_learn(&mut self, hints: &[AddressHint]) {
for hint in hints {
self.learn(hint.node_id, hint.addr);
}
}
}
// ─── NodeDriver ─────────────────────────────────────────────────────────────
/// Network driver that owns a `DistributedNode` and performs real TCP I/O.
pub struct NodeDriver {
node: DistributedNode,
transport: TcpTransport,
acceptor: TcpAcceptor,
streams: Vec<TcpStream>,
address_book: PeerAddressBook,
listen_addr: SocketAddr,
}
impl NodeDriver {
/// Create a new driver. Binds a TCP listener on `listen_addr`.
pub fn new(listen_addr: SocketAddr, config: DistributedNodeConfig) -> Result<Self, swactor::Error> {
let acceptor = TcpAcceptor::bind(listen_addr)?;
let actual_addr = acceptor.local_addr();
let node = DistributedNode::new(config);
Ok(Self {
node,
transport: TcpTransport::pool(),
acceptor,
streams: Vec::new(),
address_book: PeerAddressBook::new(),
listen_addr: actual_addr,
})
}
/// Create a driver with a specific keypair for persistent identity.
pub fn with_keypair(
listen_addr: SocketAddr,
keypair: crate::crypto::Keypair,
config: DistributedNodeConfig,
) -> Result<Self, swactor::Error> {
let acceptor = TcpAcceptor::bind(listen_addr)?;
let actual_addr = acceptor.local_addr();
let node = DistributedNode::with_keypair(keypair, config);
Ok(Self {
node,
transport: TcpTransport::pool(),
acceptor,
streams: Vec::new(),
address_book: PeerAddressBook::new(),
listen_addr: actual_addr,
})
}
/// The node's identity.
pub fn node_id(&self) -> NodeId {
self.node.node_id()
}
/// The address this driver is listening on.
pub fn listen_addr(&self) -> SocketAddr {
self.listen_addr
}
/// Access the underlying node (read-only).
pub fn node(&self) -> &DistributedNode {
&self.node
}
/// Access the underlying node (mutable).
pub fn node_mut(&mut self) -> &mut DistributedNode {
&mut self.node
}
/// Capture a snapshot of the node's state, enriched with addresses
/// from the driver's address book.
pub fn snapshot(&self) -> DistributionNodeSnapshot {
let mut snap = self.node.snapshot();
snap.listen_addr = Some(self.listen_addr.to_string());
// Enrich member addresses from address book
for member in &mut snap.members {
if let Some(node_id) = parse_node_id_hex(&member.node_id)
&& let Some(addr) = self.address_book.resolve(&node_id) {
member.addr = Some(addr.to_string());
}
}
// Enrich routing neighbor addresses from address book
for neighbor in &mut snap.routing_neighbors {
if let Some(node_id) = parse_node_id_hex(&neighbor.node_id)
&& let Some(addr) = self.address_book.resolve(&node_id) {
neighbor.addr = Some(addr.to_string());
}
}
snap
}
/// Join a cluster by contacting seed nodes.
///
/// Sends `JoinRequest` messages directly to each seed over TCP.
/// On receiving a `JoinResponse`, extracts address hints and passes
/// the member records to the protocol layer.
pub fn join(&mut self, seeds: &[SocketAddr]) {
for seed_addr in seeds {
if let Err(e) = self.send_join_request(*seed_addr) {
eprintln!("driver: join send error to {seed_addr}: {e}");
}
}
}
fn send_join_request(&mut self, seed_addr: SocketAddr) -> Result<(), swactor::Error> {
let msg = JoinRequest {
from: self.node.node_id(),
};
let hints = vec![AddressHint {
node_id: self.node.node_id(),
addr: self.listen_addr,
}];
self.send_wire_with_hints::<JoinRequest>(&msg, seed_addr, &hints)
}
/// Advance the node by one tick.
///
/// Drives the SWIM probe cycle, sends outgoing protocol messages,
/// and handles periodic republishing.
pub fn tick(&mut self) {
let actions = self.node.tick();
self.send_actions(&actions);
}
/// Process incoming TCP messages.
///
/// Reads all available wire envelopes from the acceptor, dispatches
/// each to the appropriate handler, and sends any response actions.
pub fn recv(&mut self) {
let envelopes = self.acceptor.try_recv(&mut self.streams);
for (envelope, _peer_addr, hints_bytes) in envelopes {
if !hints_bytes.is_empty()
&& let Ok(hints) = serde_json::from_slice::<Vec<AddressHint>>(&hints_bytes) {
self.learn_hints(&hints);
}
let response_actions = self.dispatch_incoming(envelope);
self.send_actions(&response_actions);
}
}
/// Process incoming TCP messages with peer auth filtering.
///
/// Same as `recv()`, but checks the sender's NodeId against the
/// peer allow-list before dispatching. Unauthorized messages are dropped.
pub fn recv_with_auth(
&mut self,
peer_auth: &std::sync::Arc<std::sync::Mutex<crate::peer_auth::PeerAllowList>>,
) {
let envelopes = self.acceptor.try_recv(&mut self.streams);
for (envelope, _peer_addr, hints_bytes) in envelopes {
// Extract sender NodeId from hints
let mut sender_node_id = None;
if !hints_bytes.is_empty()
&& let Ok(hints) = serde_json::from_slice::<Vec<AddressHint>>(&hints_bytes) {
if let Some(first) = hints.first() {
sender_node_id = Some(first.node_id);
}
self.learn_hints(&hints);
}
// Check peer auth if we know the sender
if let Some(node_id) = sender_node_id {
let allowed = peer_auth.lock().unwrap().is_allowed(&node_id);
if !allowed {
let hex: String = node_id.0[..4].iter().map(|b| format!("{b:02x}")).collect();
eprintln!("driver: rejected message from unauthorized peer {hex}");
continue;
}
}
let response_actions = self.dispatch_incoming(envelope);
self.send_actions(&response_actions);
}
}
// ─── Outgoing: NodeAction → TCP ─────────────────────────────────────
fn send_actions(&mut self, actions: &[NodeAction]) {
for action in actions {
if let Err(e) = self.send_action(action) {
eprintln!("driver: send error: {e}");
}
}
}
fn send_action(&mut self, action: &NodeAction) -> Result<(), swactor::Error> {
// Standard sender hint
let sender_hint = AddressHint {
node_id: self.node.node_id(),
addr: self.listen_addr,
};
match action {
NodeAction::SendPing {
to,
sequence,
piggyback,
} => {
let dest = self.resolve_addr(to)?;
let msg = Ping {
from: self.node.node_id(),
sequence: *sequence,
piggyback: piggyback.clone(),
};
self.send_wire_with_hints::<Ping>(&msg, dest, &[sender_hint])
}
NodeAction::SendAck {
to,
sequence,
piggyback,
} => {
let dest = self.resolve_addr(to)?;
let msg = Ack {
from: self.node.node_id(),
sequence: *sequence,
piggyback: piggyback.clone(),
};
self.send_wire_with_hints::<Ack>(&msg, dest, &[sender_hint])
}
NodeAction::SendPingReq {
relay,
target,
sequence,
piggyback,
} => {
let dest = self.resolve_addr(relay)?;
let msg = PingReq {
from: self.node.node_id(),
target: *target,
sequence: *sequence,
piggyback: piggyback.clone(),
};
// Include target hint so the relay can forward
let mut hints = vec![sender_hint];
if let Some(target_addr) = self.address_book.resolve(target) {
hints.push(AddressHint {
node_id: *target,
addr: target_addr,
});
}
self.send_wire_with_hints::<PingReq>(&msg, dest, &hints)
}
NodeAction::SendJoinResponse {
to, members,
} => {
let dest = self.resolve_addr(to)?;
let msg = JoinResponse {
members: members.clone(),
};
// Include all known member addresses as hints
let mut hints = vec![sender_hint];
for record in members {
if let Some(addr) = self.address_book.resolve(&record.node_id) {
hints.push(AddressHint {
node_id: record.node_id,
addr,
});
}
}
self.send_wire_with_hints::<JoinResponse>(&msg, dest, &hints)
}
NodeAction::ForwardAck {
to,
target,
sequence,
piggyback,
} => {
let dest = self.resolve_addr(to)?;
let msg = IndirectAck {
target: *target,
sequence: *sequence,
piggyback: piggyback.clone(),
};
self.send_wire_with_hints::<IndirectAck>(&msg, dest, &[sender_hint])
}
NodeAction::MembershipChanged { .. } => {
// Internal notification — no network I/O.
Ok(())
}
}
}
fn resolve_addr(&self, node_id: &NodeId) -> Result<SocketAddr, swactor::Error> {
self.address_book
.resolve(node_id)
.ok_or_else(|| swactor::Error::from(format!(
"no address known for node {:?}",
node_id
)))
}
fn send_wire_with_hints<M: NetworkMessage + serde::Serialize>(
&mut self,
msg: &M,
dest_addr: SocketAddr,
hints: &[AddressHint],
) -> Result<(), swactor::Error> {
let payload = serde_json::to_vec(msg)
.map_err(|e| swactor::Error::from(format!("encode {}: {e}", M::type_tag())))?;
let hints_bytes = serde_json::to_vec(hints).unwrap_or_default();
let envelope = WireEnvelope {
dest: SWIM_DEST,
type_tag: M::type_tag().to_string(),
payload,
};
let buf = encode_wire_envelope_with_hints(&envelope, &hints_bytes);
let mut stream = self.transport_get_or_connect(dest_addr)?;
use std::io::Write;
match stream.write_all(&buf) {
Ok(()) => Ok(()),
Err(_) => {
self.transport.evict(dest_addr);
let mut stream = self.transport_get_or_connect(dest_addr)?;
stream
.write_all(&buf)
.map_err(|e| swactor::Error::from(format!("TCP send to {dest_addr}: {e}")))
}
}
}
fn transport_get_or_connect(&self, addr: SocketAddr) -> Result<TcpStream, swactor::Error> {
self.transport.get_or_connect(addr)
}
// ─── Incoming: TCP → handler ────────────────────────────────────────
fn dispatch_incoming(&mut self, envelope: WireEnvelope) -> Vec<NodeAction> {
match envelope.type_tag.as_str() {
"swactor_dist::Ping" => match decode::<Ping>(&envelope.payload) {
Ok(msg) => {
self.node.handle_ping(
msg.from,
msg.sequence,
&msg.piggyback,
)
}
Err(e) => {
eprintln!("driver: decode Ping: {e}");
Vec::new()
}
},
"swactor_dist::Ack" => match decode::<Ack>(&envelope.payload) {
Ok(msg) => self.node.handle_ack(msg.from, msg.sequence, &msg.piggyback),
Err(e) => {
eprintln!("driver: decode Ack: {e}");
Vec::new()
}
},
"swactor_dist::PingReq" => match decode::<PingReq>(&envelope.payload) {
Ok(msg) => self.node.handle_ping_req(
msg.from,
msg.target,
msg.sequence,
&msg.piggyback,
),
Err(e) => {
eprintln!("driver: decode PingReq: {e}");
Vec::new()
}
},
"swactor_dist::JoinRequest" => match decode::<JoinRequest>(&envelope.payload) {
Ok(msg) => self.node.handle_join_request(msg.from),
Err(e) => {
eprintln!("driver: decode JoinRequest: {e}");
Vec::new()
}
},
"swactor_dist::JoinResponse" => match decode::<JoinResponse>(&envelope.payload) {
Ok(msg) => self.node.handle_join_response(msg.members),
Err(e) => {
eprintln!("driver: decode JoinResponse: {e}");
Vec::new()
}
},
"swactor_dist::IndirectAck" => match decode::<IndirectAck>(&envelope.payload) {
Ok(msg) => self.node.handle_indirect_ack(msg.target, msg.sequence, &msg.piggyback),
Err(e) => {
eprintln!("driver: decode IndirectAck: {e}");
Vec::new()
}
},
other => {
eprintln!("driver: unknown message type: {other}");
Vec::new()
}
}
}
/// Process address hints extracted from incoming frames.
pub fn learn_hints(&mut self, hints: &[AddressHint]) {
self.address_book.bulk_learn(hints);
}
}
fn decode<M: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<M, String> {
serde_json::from_slice(bytes).map_err(|e| e.to_string())
}
fn parse_node_id_hex(hex: &str) -> Option<NodeId> {
if hex.len() != 64 {
return None;
}
let mut bytes = [0u8; 32];
for i in 0..32 {
bytes[i] = u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16).ok()?;
}
Some(NodeId(bytes))
}

276
crates/distribution/src/iroh_driver.rs
View file

@ -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<IpAddr> {
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<Mutex<Vec<(NodeId, Connection)>>>,
/// Relay URLs learned from join seeds, used for reconnection.
peer_relay_urls: HashMap<NodeId, iroh::RelayUrl>,
/// Real-time join status for each peer being joined.
join_statuses: Arc<Mutex<HashMap<NodeId, JoinStatus>>>,
/// Embedded relay server (if started).
#[cfg(feature = "relay")]
relay_server: Option<iroh_relay::server::Server>,
@ -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<SocketAddr> {
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<NodeId, JoinStatus> {
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::<iroh::RelayUrl>().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);
});
}
@ -416,9 +627,24 @@ impl IrohDriver {
// ─── Outgoing: NodeAction → iroh ─────────────────────────────────
fn send_actions(&mut self, actions: &[NodeAction]) {
let mut failure_targets: Vec<NodeId> = Vec::new();
for action in actions {
if let Err(e) = self.send_action(action) {
eprintln!("iroh driver: send error: {e}");
if let Some(target) = action_target(action) {
if !failure_targets.contains(&target) {
failure_targets.push(target);
}
}
}
}
for target in failure_targets {
let probe_actions = self.node.report_send_failure(target);
// Best-effort send of probe actions — no recursion on failure
for action in &probe_actions {
if let Err(e) = self.send_action(action) {
eprintln!("iroh driver: probe send error: {e}");
}
}
}
}
@ -765,6 +991,20 @@ async fn start_embedded_relay(
Ok((server, url))
}
// ─── Helpers ─────────────────────────────────────────────────────────────────
/// Extract the send target from a node action (if it has one).
fn action_target(action: &NodeAction) -> Option<NodeId> {
match action {
NodeAction::SendPing { to, .. } => Some(*to),
NodeAction::SendAck { to, .. } => Some(*to),
NodeAction::SendPingReq { relay, .. } => Some(*relay),
NodeAction::SendJoinResponse { to, .. } => Some(*to),
NodeAction::ForwardAck { to, .. } => Some(*to),
NodeAction::MembershipChanged { .. } => None,
}
}
// ─── Wire Framing Over QUIC Streams ─────────────────────────────────────────
/// Write a tagged message to a QUIC send stream.

2
crates/distribution/src/lib.rs
View file

@ -10,7 +10,5 @@ pub mod node;
pub mod registry;
pub mod node_metadata;
pub mod snapshot;
#[cfg(feature = "tcp")]
pub mod driver;
#[cfg(feature = "iroh")]
pub mod iroh_driver;

27
crates/distribution/src/node.rs
View file

@ -191,6 +191,18 @@ impl DistributedNode {
self.inject_piggyback(actions)
}
/// Report that a send to `target` failed, triggering a reactive probe.
pub fn report_send_failure(&mut self, target: NodeId) -> Vec<NodeAction> {
let actions = self.swim.report_send_failure(target);
self.process_membership_changes(&actions);
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<NodeAction> {
let actions = self.swim.handle_join_request(from);
self.maybe_update_routing_table(from);
@ -286,8 +298,16 @@ impl DistributedNode {
/// Set this node's relay URL and begin gossiping it to the cluster.
pub fn set_relay_url(&mut self, url: Option<String>) {
let name = self.metadata.node_name(&self.node_id()).map(String::from);
self.metadata
.set_local(self.node_id(), url, self.cluster_size());
.set_local(self.node_id(), url, name, self.cluster_size());
}
/// Set this node's human-readable name and begin gossiping it to the cluster.
pub fn set_node_name(&mut self, name: String) {
let relay_url = self.metadata.relay_url(&self.node_id()).map(String::from);
self.metadata
.set_local(self.node_id(), relay_url, Some(name), self.cluster_size());
}
/// Look up a node's relay URL.
@ -295,6 +315,11 @@ impl DistributedNode {
self.metadata.relay_url(node_id)
}
/// Look up a node's human-readable name.
pub fn node_name(&self, node_id: &NodeId) -> Option<&str> {
self.metadata.node_name(node_id)
}
/// Read-only access to the metadata disseminator.
pub fn metadata(&self) -> &NodeMetadataDisseminator {
&self.metadata

20
crates/distribution/src/node_metadata.rs
View file

@ -15,6 +15,8 @@ use crate::types::NodeId;
pub struct NodeMetadataEntry {
pub node_id: NodeId,
pub relay_url: Option<String>,
#[serde(default)]
pub node_name: Option<String>,
pub generation: u64,
}
@ -43,12 +45,19 @@ impl NodeMetadataDisseminator {
}
}
/// Set this node's relay URL and enqueue for dissemination.
pub fn set_local(&mut self, node_id: NodeId, relay_url: Option<String>, cluster_size: usize) {
/// Set this node's metadata and enqueue for dissemination.
pub fn set_local(
&mut self,
node_id: NodeId,
relay_url: Option<String>,
node_name: Option<String>,
cluster_size: usize,
) {
self.local_generation += 1;
let entry = NodeMetadataEntry {
node_id,
relay_url,
node_name,
generation: self.local_generation,
};
self.store.insert(node_id, entry.clone());
@ -91,6 +100,13 @@ impl NodeMetadataDisseminator {
.and_then(|e| e.relay_url.as_deref())
}
/// Look up a node's human-readable name.
pub fn node_name(&self, node_id: &NodeId) -> Option<&str> {
self.store
.get(node_id)
.and_then(|e| e.node_name.as_deref())
}
/// Remove metadata for a dead node.
pub fn remove_node(&mut self, node_id: &NodeId) {
self.store.remove(node_id);

29
crates/distribution/src/snapshot.rs
View file

@ -21,6 +21,8 @@ pub struct MemberInfo {
pub label: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub relay_url: Option<String>,
#[serde(skip_serializing_if = "Option::is_none", default)]
pub node_name: Option<String>,
}
/// Snapshot of a node in the Kademlia routing table.
@ -46,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<String>,
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 {
@ -115,6 +131,14 @@ pub struct DistributionNodeSnapshot {
/// This node's relay URL, if running an embedded relay server.
#[serde(skip_serializing_if = "Option::is_none", default)]
pub relay_url: Option<String>,
/// Build version string (e.g. "branch @ hash").
#[serde(skip_serializing_if = "Option::is_none", default)]
pub version: Option<String>,
/// Real-time join statuses for peers being connected to.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub join_statuses: Vec<JoinStatusInfo>,
}
fn node_id_hex(id: &NodeId) -> String {
@ -146,6 +170,7 @@ impl DistributedNode {
is_authorized: None,
label: None,
relay_url: self.metadata().relay_url(&m.node_id).map(String::from),
node_name: self.metadata().node_name(&m.node_id).map(String::from),
})
.collect();
@ -210,9 +235,11 @@ impl DistributedNode {
recent_probe_targets: recent_targets,
peer_auth_mode: "open".into(),
authorized_peer_count: None,
node_name: None,
node_name: self.metadata().node_name(&self.node_id()).map(String::from),
invite_code: None,
relay_url: self.metadata().relay_url(&self.node_id()).map(String::from),
version: None,
join_statuses: Vec::new(),
}
}
}

9
crates/distribution/src/swim/dissemination.rs
View file

@ -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;

5
crates/distribution/src/swim/member_list.rs
View file

@ -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()

32
crates/distribution/src/swim/node.rs
View file

@ -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<NodeId> {
self.probe.recent_probe_targets()
@ -142,6 +157,15 @@ impl SwimNode {
actions
}
/// Report that a send to `target` failed, triggering a reactive probe.
pub fn report_send_failure(&mut self, target: NodeId) -> Vec<NodeAction> {
let probe_actions = self.probe.step(
SwimEvent::SendFailed { to: target },
&mut self.members,
);
self.translate_probe_actions(probe_actions)
}
/// Handle a received indirect ack (forwarded by a relay node).
pub fn handle_indirect_ack(&mut self, target: NodeId, sequence: u64, piggyback: &[u8]) -> Vec<NodeAction> {
let mut actions = self.apply_piggyback(piggyback);
@ -202,6 +226,11 @@ impl SwimNode {
state: record.state,
incarnation: record.incarnation,
});
// In reactive mode, probe newly discovered alive peers so they
// don't decay to dead before we ever exchange a ping/ack.
if record.state == MemberState::Alive && record.node_id != self.members.self_id() {
self.probe.enqueue_demand_probe(record.node_id);
}
}
}
actions
@ -259,6 +288,9 @@ impl SwimNode {
if changed {
if update.state == MemberState::Alive {
eprintln!("SWIM: alive {}", &hex_encode(&update.node_id.0)[..8]);
// In reactive mode, probe newly discovered alive peers so they
// don't decay to dead before we ever exchange a ping/ack.
self.probe.enqueue_demand_probe(update.node_id);
}
// Re-disseminate the update
self.dissemination.enqueue(

150
crates/distribution/src/swim/probe.rs
View file

@ -14,10 +14,20 @@ const PROBE_HISTORY_SIZE: usize = 16;
// ─── Configuration ──────────────────────────────────────────────────────────
/// Controls how the probe cycle triggers probes.
#[derive(Debug, Clone, PartialEq)]
pub enum ProbeMode {
/// Classic SWIM: probe one random member every `probe_interval` ticks.
Periodic,
/// No periodic probing. Probes triggered externally via `SendFailed`.
/// Safety sweep probes one random member every `safety_sweep_interval` ticks.
Reactive { safety_sweep_interval: u64 },
}
/// SWIM protocol configuration.
#[derive(Debug, Clone)]
pub struct SwimConfig {
/// Ticks between probe cycles.
/// Ticks between probe cycles (used in Periodic mode).
pub probe_interval: u64,
/// Ticks to wait for a direct ack before sending indirect probes.
pub probe_timeout: u64,
@ -28,6 +38,8 @@ pub struct SwimConfig {
/// Ticks between dead-node reprobe attempts. 0 = disabled.
/// When enabled, periodically pings dead nodes to detect partition heals.
pub dead_reprobe_interval: u64,
/// Probe mode: Periodic (default) or Reactive (probe-on-failure).
pub probe_mode: ProbeMode,
}
impl Default for SwimConfig {
@ -38,6 +50,7 @@ impl Default for SwimConfig {
indirect_probes: 3,
suspicion_timeout: 30,
dead_reprobe_interval: 50,
probe_mode: ProbeMode::Periodic,
}
}
}
@ -53,6 +66,8 @@ pub enum SwimEvent {
AckReceived { from: NodeId, sequence: u64 },
/// Received an indirect ack (forwarded through a relay).
IndirectAckReceived { target: NodeId, sequence: u64 },
/// A send to the given peer failed (reactive probe trigger).
SendFailed { to: NodeId },
}
// ─── Actions (outputs) ──────────────────────────────────────────────────────
@ -103,6 +118,9 @@ struct SuspicionTimer {
started_at: u64,
}
/// Maximum demand queue size to prevent unbounded growth.
const MAX_DEMAND_QUEUE: usize = 32;
/// The SWIM probe state machine.
pub struct SwimProbe {
config: SwimConfig,
@ -122,6 +140,10 @@ pub struct SwimProbe {
next_reprobe_tick: u64,
/// Round-robin index into the dead member list for reprobe target selection.
reprobe_index: usize,
/// Peers needing probes due to send failures (reactive mode).
demand_queue: VecDeque<NodeId>,
/// Tick at which the next safety sweep fires (reactive mode).
next_sweep_tick: u64,
}
impl SwimProbe {
@ -131,6 +153,10 @@ impl SwimProbe {
} else {
u64::MAX
};
let next_sweep = match &config.probe_mode {
ProbeMode::Reactive { safety_sweep_interval } => *safety_sweep_interval,
ProbeMode::Periodic => u64::MAX,
};
Self {
next_probe_tick: config.probe_interval,
next_reprobe_tick: next_reprobe,
@ -143,6 +169,8 @@ impl SwimProbe {
probe_order: Vec::new(),
suspicion_timers: Vec::new(),
recent_targets: VecDeque::with_capacity(PROBE_HISTORY_SIZE),
demand_queue: VecDeque::new(),
next_sweep_tick: next_sweep,
}
}
@ -155,7 +183,15 @@ impl SwimProbe {
self.tick += 1;
self.check_probe_timeout(members, &mut actions);
self.check_suspicion_timeouts(members, &mut actions);
match &self.config.probe_mode {
ProbeMode::Periodic => {
self.maybe_start_probe(members, &mut actions);
}
ProbeMode::Reactive { .. } => {
self.maybe_start_demand_probe(members, &mut actions);
self.maybe_safety_sweep(members, &mut actions);
}
}
self.maybe_reprobe_dead(members, &mut actions);
}
SwimEvent::AckReceived { from, sequence } => {
@ -164,6 +200,9 @@ impl SwimProbe {
SwimEvent::IndirectAckReceived { target, sequence } => {
self.handle_indirect_ack(target, sequence, members, &mut actions);
}
SwimEvent::SendFailed { to } => {
self.handle_send_failed(to, members, &mut actions);
}
}
actions
@ -390,6 +429,115 @@ impl SwimProbe {
sequence: seq,
});
}
// ─── Reactive mode ─────────────────────────────────────────────────
/// Enqueue a demand probe for a newly discovered peer (reactive mode only).
///
/// Called when gossip or a join response introduces a new Alive member.
/// In Periodic mode this is a no-op (periodic probing covers it).
pub fn enqueue_demand_probe(&mut self, target: NodeId) {
if matches!(self.config.probe_mode, ProbeMode::Periodic) {
return;
}
if self.is_currently_probing(target) || self.demand_queue.contains(&target) {
return;
}
if self.demand_queue.len() < MAX_DEMAND_QUEUE {
self.demand_queue.push_back(target);
}
}
/// Handle a send failure: start a probe immediately or queue it.
fn handle_send_failed(&mut self, target: NodeId, members: &MemberList, actions: &mut Vec<SwimAction>) {
// Ignore failures for dead peers, self, or already-queued targets
if let Some(entry) = members.get(&target) {
if entry.state == MemberState::Dead {
return;
}
} else {
// Unknown peer — nothing to probe
return;
}
// Ignore if we're already probing this target
if self.is_currently_probing(target) {
return;
}
// Ignore if already in demand queue
if self.demand_queue.contains(&target) {
return;
}
if matches!(self.phase, ProbePhase::Idle) {
// Start probe immediately
self.start_probe_for(target, actions);
} else {
// Queue it (capped)
if self.demand_queue.len() < MAX_DEMAND_QUEUE {
self.demand_queue.push_back(target);
}
}
}
/// Start a directed probe to a specific target.
fn start_probe_for(&mut self, target: NodeId, actions: &mut Vec<SwimAction>) {
if self.recent_targets.len() >= PROBE_HISTORY_SIZE {
self.recent_targets.pop_front();
}
self.recent_targets.push_back(target);
let seq = self.next_sequence();
actions.push(SwimAction::SendPing {
to: target,
sequence: seq,
});
self.phase = ProbePhase::WaitingDirectAck {
target,
sequence: seq,
sent_at: self.tick,
};
}
/// On each tick in reactive mode, if idle and queue non-empty, pop and probe.
fn maybe_start_demand_probe(&mut self, _members: &MemberList, actions: &mut Vec<SwimAction>) {
if !matches!(self.phase, ProbePhase::Idle) {
return;
}
if let Some(target) = self.demand_queue.pop_front() {
self.start_probe_for(target, actions);
}
}
/// At safety_sweep_interval, probe one random alive member.
fn maybe_safety_sweep(&mut self, members: &MemberList, actions: &mut Vec<SwimAction>) {
if self.tick < self.next_sweep_tick {
return;
}
let interval = match &self.config.probe_mode {
ProbeMode::Reactive { safety_sweep_interval } => *safety_sweep_interval,
ProbeMode::Periodic => return,
};
self.next_sweep_tick = self.tick + interval;
if !matches!(self.phase, ProbePhase::Idle) {
return;
}
if let Some(target) = self.pick_probe_target(&MemberList::clone_shallow(members)) {
self.start_probe_for(target, actions);
}
}
/// Check if we are currently probing a specific target.
fn is_currently_probing(&self, target: NodeId) -> bool {
match &self.phase {
ProbePhase::WaitingDirectAck { target: t, .. }
| ProbePhase::WaitingIndirectAck { target: t, .. } => *t == target,
ProbePhase::Idle => false,
}
}
}
// Helper: we need a read-only borrow of members in pick_probe_target

1
crates/distribution/tests/common/mod.rs
View file

@ -21,6 +21,7 @@ pub fn test_config() -> DistributedNodeConfig {
indirect_probes: 1,
suspicion_timeout: 5,
dead_reprobe_interval: 0,
..SwimConfig::default()
},
cache_capacity: 100,
republish_interval: 50,

1
crates/distribution/tests/swim_node.rs
View file

@ -13,6 +13,7 @@ fn fast_config() -> SwimConfig {
indirect_probes: 2,
suspicion_timeout: 10,
dead_reprobe_interval: 0,
..SwimConfig::default()
}
}

9
crates/distribution/tests/swim_probe.rs
View file

@ -95,6 +95,7 @@ fn probe_sends_ping_after_interval() {
indirect_probes: 2,
suspicion_timeout: 20,
dead_reprobe_interval: 0,
..SwimConfig::default()
};
let mut probe = SwimProbe::new(config);
let mut members = MemberList::new(node(0));
@ -118,6 +119,7 @@ fn probe_ack_completes_cycle() {
indirect_probes: 2,
suspicion_timeout: 20,
dead_reprobe_interval: 0,
..SwimConfig::default()
};
let mut probe = SwimProbe::new(config);
let mut members = MemberList::new(node(0));
@ -148,6 +150,7 @@ fn probe_timeout_triggers_indirect_probes() {
indirect_probes: 2,
suspicion_timeout: 20,
dead_reprobe_interval: 0,
..SwimConfig::default()
};
let mut probe = SwimProbe::new(config);
let mut members = MemberList::new(node(0));
@ -175,6 +178,7 @@ fn no_ack_at_all_causes_suspicion() {
indirect_probes: 2,
suspicion_timeout: 20,
dead_reprobe_interval: 0,
..SwimConfig::default()
};
let mut probe = SwimProbe::new(config);
let mut members = MemberList::new(node(0));
@ -204,6 +208,7 @@ fn suspicion_timeout_causes_death_declaration() {
indirect_probes: 0,
suspicion_timeout: 10,
dead_reprobe_interval: 0,
..SwimConfig::default()
};
let mut probe = SwimProbe::new(config);
let mut members = MemberList::new(node(0));
@ -238,6 +243,7 @@ fn indirect_ack_rescues_suspected_node() {
indirect_probes: 2,
suspicion_timeout: 20,
dead_reprobe_interval: 0,
..SwimConfig::default()
};
let mut probe = SwimProbe::new(config);
let mut members = MemberList::new(node(0));
@ -289,6 +295,7 @@ fn reprobe_sends_ping_to_dead_node() {
indirect_probes: 0,
suspicion_timeout: 10,
dead_reprobe_interval: 20,
..SwimConfig::default()
};
let mut probe = SwimProbe::new(config);
let mut members = MemberList::new(node(0));
@ -314,6 +321,7 @@ fn reprobe_disabled_when_interval_is_zero() {
indirect_probes: 0,
suspicion_timeout: 10,
dead_reprobe_interval: 0,
..SwimConfig::default()
};
let mut probe = SwimProbe::new(config);
let mut members = MemberList::new(node(0));
@ -336,6 +344,7 @@ fn reprobe_does_nothing_when_no_dead_members() {
indirect_probes: 0,
suspicion_timeout: 10,
dead_reprobe_interval: 20,
..SwimConfig::default()
};
let mut probe = SwimProbe::new(config);
let mut members = MemberList::new(node(0));

172
crates/distribution/tests/transport_and_codec.rs
View file

@ -74,175 +74,3 @@ fn all_message_types_registered_in_codec_registry() {
}
}
// ─── TCP transport tests (require "tcp" feature) ────────────────────────────
#[cfg(feature = "tcp")]
mod tcp_transport {
use swactor::actor::ActorAddress;
use swactor::transport::WireEnvelope;
use distribution::codec::distribution_codec_registry;
use distribution::messages::*;
use swactor_transport::tcp::{TcpAcceptor, TcpTransport};
use distribution::types::NodeId;
#[test]
fn wire_envelope_roundtrips_through_tcp() {
let acceptor = TcpAcceptor::bind("127.0.0.1:0".parse().unwrap()).unwrap();
let addr = acceptor.local_addr();
let original = WireEnvelope {
dest: ActorAddress::new_random(),
type_tag: "test::Msg".to_string(),
payload: vec![1, 2, 3, 4, 5],
};
let original_clone = original.clone();
let sender = std::thread::spawn(move || {
let transport = TcpTransport::new(addr);
transport.send_to(addr, original_clone).unwrap();
});
std::thread::sleep(std::time::Duration::from_millis(50));
let mut streams = Vec::new();
let envelopes = loop {
let envs = acceptor.try_recv(&mut streams);
if !envs.is_empty() {
break envs;
}
std::thread::sleep(std::time::Duration::from_millis(10));
};
sender.join().unwrap();
assert_eq!(envelopes.len(), 1);
let (received, _peer, hints) = &envelopes[0];
assert_eq!(received.dest, original.dest);
assert_eq!(received.type_tag, original.type_tag);
assert_eq!(received.payload, original.payload);
assert!(hints.is_empty(), "no hints expected from plain send_to");
}
#[test]
fn wire_envelope_minimal_roundtrips() {
let acceptor = TcpAcceptor::bind("127.0.0.1:0".parse().unwrap()).unwrap();
let addr = acceptor.local_addr();
let original = WireEnvelope {
dest: ActorAddress::new_random(),
type_tag: "test::Minimal".to_string(),
payload: vec![42],
};
let original_clone = original.clone();
let sender = std::thread::spawn(move || {
let transport = TcpTransport::new(addr);
transport.send_to(addr, original_clone).unwrap();
});
std::thread::sleep(std::time::Duration::from_millis(50));
let mut streams = Vec::new();
let envelopes = loop {
let envs = acceptor.try_recv(&mut streams);
if !envs.is_empty() {
break envs;
}
std::thread::sleep(std::time::Duration::from_millis(10));
};
sender.join().unwrap();
let (received, _, _hints) = &envelopes[0];
assert_eq!(received.payload, vec![42]);
}
#[test]
fn two_drivers_complete_join_handshake() {
use distribution::driver::NodeDriver;
use distribution::node::DistributedNodeConfig;
let mut driver_a = NodeDriver::new(
"127.0.0.1:0".parse().unwrap(),
DistributedNodeConfig::default(),
)
.unwrap();
let mut driver_b = NodeDriver::new(
"127.0.0.1:0".parse().unwrap(),
DistributedNodeConfig::default(),
)
.unwrap();
let b_addr = driver_b.listen_addr();
// A sends JoinRequest to B
driver_a.join(&[b_addr]);
// B receives the JoinRequest, learns A's address from hints, sends JoinResponse
std::thread::sleep(std::time::Duration::from_millis(50));
driver_b.recv();
// A receives the JoinResponse with member list
std::thread::sleep(std::time::Duration::from_millis(50));
driver_a.recv();
// A should now have members — proving the full roundtrip worked.
// If hints were broken, B couldn't resolve A's address to send the
// JoinResponse, so A's member list would stay empty.
let members = driver_a.node().members();
assert!(
!members.is_empty(),
"join handshake should complete: A needs members from JoinResponse"
);
}
#[test]
fn ping_message_survives_codec_and_tcp_roundtrip() {
let codecs = distribution_codec_registry();
let acceptor = TcpAcceptor::bind("127.0.0.1:0".parse().unwrap()).unwrap();
let server_addr = acceptor.local_addr();
let dest = ActorAddress::new_random();
let ping = Ping {
from: NodeId([0xBB; 32]),
sequence: 99,
piggyback: vec![],
};
let type_id = std::any::TypeId::of::<Ping>();
let (tag, payload) = codecs.encode(type_id, Box::new(ping.clone())).unwrap();
let envelope = WireEnvelope {
dest,
type_tag: tag,
payload,
};
let envelope_clone = envelope.clone();
let sender = std::thread::spawn(move || {
let transport = TcpTransport::new(server_addr);
transport.send_to(server_addr, envelope_clone).unwrap();
});
std::thread::sleep(std::time::Duration::from_millis(50));
let mut streams = Vec::new();
let envelopes = loop {
let envs = acceptor.try_recv(&mut streams);
if !envs.is_empty() {
break envs;
}
std::thread::sleep(std::time::Duration::from_millis(10));
};
sender.join().unwrap();
let (received, _, _hints) = &envelopes[0];
let (addr, msg_any) = codecs.receive(received.clone()).unwrap();
assert_eq!(addr, dest);
let decoded: &Ping = msg_any.downcast_ref().unwrap();
assert_eq!(decoded.from, NodeId([0xBB; 32]));
assert_eq!(decoded.sequence, 99);
}
}

1
crates/node/Cargo.toml
View file

@ -23,7 +23,6 @@ path = "src/lib.rs"
[features]
default = ["iroh", "relay"]
tcp = ["distribution/tcp"]
iroh = ["distribution/iroh", "dep:iroh"]
relay = ["iroh", "distribution/relay"]

404
crates/node/src/main.rs
View file

@ -73,19 +73,7 @@ struct Args {
#[arg(long)]
config: Option<std::path::PathBuf>,
/// Transport to use: iroh or tcp
#[arg(long, default_value = "iroh")]
transport: String,
/// Address to listen on for TCP transport (e.g. 10.0.1.10:7000)
#[arg(long)]
listen: Option<std::net::SocketAddr>,
/// Seed node address to join (TCP mode: host:port)
#[arg(long)]
seed: Option<String>,
/// Seed node's iroh public key (iroh mode: hex-encoded 32-byte key)
/// Seed node's iroh public key (hex-encoded 32-byte key)
#[arg(long)]
seed_node_id: Option<String>,
@ -188,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"
@ -199,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());
@ -220,6 +212,41 @@ relay_port = 3340
config_path
}
/// Detect outbound IP; if public, return a `relay_hosts = ["<ip>"]` TOML line.
fn detect_public_ip_for_config() -> String {
let public_ip = (|| -> Option<std::net::IpAddr> {
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();
@ -272,11 +299,6 @@ fn main() {
// Layer: CLI > config > defaults
// For args with default values, we check if the user explicitly provided
// the CLI flag; if not, we fall back to config, then to the default.
let transport = if args.transport != "iroh" {
args.transport.clone()
} else {
cfg.transport.unwrap_or_else(|| args.transport.clone())
};
let dashboard_port = if args.dashboard_port != 9090 {
args.dashboard_port
} else {
@ -295,8 +317,6 @@ fn main() {
let storage_path = args.storage_path.clone().or(cfg.storage_path);
let peers_file = args.peers_file.clone().or(cfg.peers_file);
let seed_node_id = args.seed_node_id.clone().or(cfg.seed_node_id);
#[cfg(feature = "tcp")]
let seed = args.seed.clone().or(cfg.seed);
let auth_enabled = args.auth || cfg.auth.unwrap_or(false);
let actors = if args.actors != 0 {
args.actors
@ -331,11 +351,6 @@ fn main() {
cfg.relay_bind.unwrap_or_else(|| args.relay_bind.clone())
};
let relay_hosts = cfg.relay_hosts.unwrap_or_default();
#[cfg(feature = "tcp")]
let listen = args.listen.or_else(|| {
cfg.listen.as_ref().and_then(|s| s.parse().ok())
});
// Signal handler — second Ctrl+C forces immediate exit
{
let stop = Arc::clone(&stop);
@ -384,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: <base58>#<addrs>@<relay>
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);
});
@ -413,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]);
@ -529,13 +559,16 @@ fn main() {
};
// Distribution config
eprintln!("Distribution: SWIM (transport: {transport})");
eprintln!("Distribution: SWIM (transport: iroh, mode: reactive)");
let swim_config = SwimConfig {
probe_interval: 5,
probe_timeout: 6, // 600ms — allows relay round-trip
probe_interval: 10, // unused in Reactive mode, kept for compat
probe_timeout: 15, // 1.5s — generous for relay roundtrips
indirect_probes: 2,
suspicion_timeout: 40, // 4s — gives refutation time to piggyback
dead_reprobe_interval: 50,
suspicion_timeout: 80, // 8s — gives refutation time to gossip back
dead_reprobe_interval: 100,
probe_mode: distribution::swim::probe::ProbeMode::Reactive {
safety_sweep_interval: 3000, // 5 minutes at 100ms/tick
},
};
let node_config = DistributedNodeConfig {
swim: swim_config,
@ -544,8 +577,9 @@ fn main() {
..Default::default()
};
// Channel for triggering SWIM joins (fed by peers plugin "Add Peer")
// Channel for triggering SWIM joins (fed by peers plugin "Add Peer" and distribution "Re-peer")
let (join_tx, join_rx) = std::sync::mpsc::channel::<dashboard::JoinPeerInfo>();
let join_tx_dist = join_tx.clone();
// Register peers plugin
let peers_plugin = plugins::peers::PeersPlugin::new(
@ -554,9 +588,8 @@ fn main() {
);
dash.register_plugin(Arc::new(peers_plugin));
match transport.as_str() {
#[cfg(feature = "iroh")]
"iroh" => run_iroh(
run_iroh(
seed_node_id,
dashboard_port,
actors,
@ -570,127 +603,24 @@ fn main() {
node_name,
invite_code,
join_rx,
join_tx_dist,
relay_enabled,
&relay_bind,
relay_port,
relay_hosts,
),
#[cfg(feature = "tcp")]
"tcp" => run_tcp(
listen,
seed,
dashboard_port,
actors,
node_config,
keypair,
Arc::clone(&peer_auth),
&handle,
&dash,
&stop,
&ds_group,
node_name,
invite_code,
join_rx,
),
other => {
eprintln!("Unknown or unavailable transport: {other}");
eprintln!("Available transports:");
#[cfg(feature = "iroh")]
eprintln!(" iroh");
#[cfg(feature = "tcp")]
eprintln!(" tcp");
);
#[cfg(not(feature = "iroh"))]
{
eprintln!("iroh feature is required but not enabled");
std::process::exit(1);
}
}
handle.shutdown();
dash.shutdown();
handle.join();
}
// ── TCP transport ────────────────────────────────────────────────────────
#[cfg(feature = "tcp")]
fn run_tcp(
listen: Option<std::net::SocketAddr>,
seed: Option<String>,
dashboard_port: u16,
actors: usize,
node_config: DistributedNodeConfig,
keypair: Keypair,
peer_auth: Arc<Mutex<PeerAllowList>>,
handle: &swactor::runtime::RuntimeHandle,
dash: &dashboard::DashboardHandle,
stop: &Arc<AtomicBool>,
ds_group: &Option<DatastoreGroup>,
node_name: String,
invite_code: String,
_join_rx: std::sync::mpsc::Receiver<dashboard::JoinPeerInfo>,
) {
use distribution::driver::NodeDriver;
let listen_addr = listen.expect("--listen is required for TCP mode");
let dist_keypair = distribution::crypto::Keypair::from_bytes(&keypair.secret_bytes());
let mut driver = NodeDriver::with_keypair(listen_addr, dist_keypair, node_config)
.expect("failed to create node driver");
eprintln!(
"Node {} listening on {} (TCP)",
hex(&driver.node_id().0[..4]),
driver.listen_addr(),
);
// Join seed if provided
if let Some(seed) = seed {
let seed_addr: std::net::SocketAddr = seed.parse().expect("invalid seed address");
eprintln!("Joining cluster via seed {seed_addr}");
driver.join(&[seed_addr]);
}
// Spawn and register actors
let actor_addrs = spawn_actors(actors, handle, driver.node_mut());
// Wire distribution snapshot to dashboard via plugin
let mut snap = driver.snapshot();
snap.node_name = Some(node_name.clone());
snap.invite_code = Some(invite_code.clone());
let cached_snapshot: Arc<Mutex<Option<DistributionNodeSnapshot>>> =
Arc::new(Mutex::new(Some(snap)));
let dist_plugin = plugins::distribution::DistributionPlugin::new(
Arc::clone(&cached_snapshot),
);
dash.register_plugin(Arc::new(dist_plugin));
// Start dashboard HTTP on a standalone tokio runtime (no iroh runtime in TCP mode)
dash.start_http_standalone();
eprintln!("Dashboard at http://0.0.0.0:{dashboard_port}");
// Main loop
let mut round: u64 = 0;
while !stop.load(Ordering::Relaxed) {
round += 1;
driver.recv_with_auth(&peer_auth);
driver.tick();
for addr in &actor_addrs {
let _ = handle.runtime.send_to(*addr, Heartbeat);
}
let mut snap = driver.snapshot();
snap.node_name = Some(node_name.clone());
snap.invite_code = Some(invite_code.clone());
*cached_snapshot.lock().unwrap() = Some(snap);
// Datastore ticks
if let Some(group) = ds_group {
group.tick(round);
}
thread::sleep(Duration::from_millis(100));
}
}
// ── iroh transport ───────────────────────────────────────────────────────
#[cfg(feature = "iroh")]
@ -708,6 +638,7 @@ fn run_iroh(
node_name: String,
invite_code: String,
join_rx: std::sync::mpsc::Receiver<dashboard::JoinPeerInfo>,
join_tx_dist: std::sync::mpsc::Sender<dashboard::JoinPeerInfo>,
relay_enabled: bool,
relay_bind: &str,
relay_port: u16,
@ -766,7 +697,12 @@ fn run_iroh(
})
.collect();
if urls.is_empty() {
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::<iroh::RelayMap>())
@ -828,7 +764,8 @@ fn run_iroh(
// Spawn and register actors
let actor_addrs = spawn_actors(actors, handle, driver.node_mut());
// Announce relay URL to cluster gossip
// Announce node name and relay URL to cluster gossip
driver.node_mut().set_node_name(node_name.clone());
let mut home_relay_set = if let Some(url) = driver.relay_url().map(|u| u.to_string()) {
driver.node_mut().set_relay_url(Some(url));
true
@ -840,11 +777,14 @@ fn run_iroh(
let mut snap = driver.snapshot();
snap.node_name = Some(node_name.clone());
snap.invite_code = Some(invite_code.clone());
snap.version = Some(VERSION.to_string());
let cached_snapshot: Arc<Mutex<Option<DistributionNodeSnapshot>>> =
Arc::new(Mutex::new(Some(snap)));
let dist_plugin = plugins::distribution::DistributionPlugin::new(
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
@ -880,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<dashboard::JoinPeerInfo> = Vec::new();
while let Ok(info) = join_rx.try_recv() {
new_peers.push(info);
}
@ -888,22 +828,25 @@ fn run_iroh(
let own_id = driver.node_id().0;
let addrs: Vec<iroh::EndpointAddr> = 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::<iroh::RelayUrl>() {
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
})
@ -929,7 +872,83 @@ 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: <base58>#<addr1>,<addr2>@<relay_url>
{
let direct_addrs = driver.direct_addresses();
let addrs_part = if direct_addrs.is_empty() {
String::new()
} else {
let addrs_str: Vec<String> = 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<swactor::transport::NodeId> = 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);
// Datastore ticks
@ -1042,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<String> {
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::<Vec<_>>()
.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::<Vec<_>>()
.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: `<base58>#<addr1>,<addr2>@<relay_url>`
///
/// Returns (node_id_str, optional_direct_addrs_csv, optional_relay_url).
fn parse_rich_invite(raw: &str) -> (String, Option<String>, Option<String>) {
// 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 {

1
crates/node/src/plugins/datastore_page.html
View file

@ -543,6 +543,7 @@
// ── SSE connection ───────────────────────────────────────────────────
var es = new EventSource('/events');
window.addEventListener('beforeunload', function() { es.close(); });
es.addEventListener('datastore', function(e) {
try {

93
crates/node/src/plugins/distribution.rs
View file

@ -7,6 +7,7 @@ use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use dashboard::plugin::{DashboardPlugin, PluginResponse};
use dashboard::JoinPeerInfo;
use distribution::snapshot::DistributionNodeSnapshot;
/// HTML page for the distribution plugin.
@ -15,11 +16,27 @@ const DISTRIBUTION_HTML: &str = include_str!("distribution_page.html");
/// Dashboard plugin that exposes distribution node snapshots.
pub struct DistributionPlugin {
cached: Arc<Mutex<Option<DistributionNodeSnapshot>>>,
join_sender: Option<std::sync::mpsc::Sender<JoinPeerInfo>>,
/// 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<Mutex<Vec<[u8; 32]>>>,
}
impl DistributionPlugin {
pub fn new(cached: Arc<Mutex<Option<DistributionNodeSnapshot>>>) -> Self {
Self { cached }
pub fn new(
cached: Arc<Mutex<Option<DistributionNodeSnapshot>>>,
join_sender: Option<std::sync::mpsc::Sender<JoinPeerInfo>>,
) -> Self {
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<Mutex<Vec<[u8; 32]>>> {
Arc::clone(&self.dismissed_statuses)
}
}
@ -39,7 +56,7 @@ impl DashboardPlugin for DistributionPlugin {
method: &str,
path: &str,
_query: &HashMap<String, String>,
_body: &[u8],
body: &[u8],
) -> PluginResponse {
match (method, path) {
("GET", "") => {
@ -52,6 +69,8 @@ impl DashboardPlugin for DistributionPlugin {
None => PluginResponse::json("{}".into()),
}
}
("POST", "rejoin") => self.handle_rejoin(body),
("POST", "clear_status") => self.handle_clear_status(body),
_ => PluginResponse::not_found(),
}
}
@ -60,3 +79,71 @@ impl DashboardPlugin for DistributionPlugin {
Some(DISTRIBUTION_HTML)
}
}
impl DistributionPlugin {
fn handle_rejoin(&self, body: &[u8]) -> PluginResponse {
let tx = match &self.join_sender {
Some(tx) => tx,
None => return PluginResponse::json(r#"{"error":"rejoin not available"}"#.into()),
};
// Parse { "node_id": "<hex>" } from body
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()),
};
// Parse hex node_id into [u8; 32]
let bytes = match parse_hex_node_id(node_id_hex) {
Some(b) => b,
None => return PluginResponse::json(r#"{"error":"invalid node_id hex"}"#.into()),
};
// Look up relay_url from cached snapshot
let relay_url = {
let guard = self.cached.lock().unwrap();
guard.as_ref().and_then(|snap| {
snap.members.iter()
.find(|m| m.node_id == node_id_hex)
.and_then(|m| m.relay_url.clone())
})
};
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]> {
if hex.len() != 64 {
return None;
}
let mut bytes = [0u8; 32];
for i in 0..32 {
bytes[i] = u8::from_str_radix(&hex[i * 2..i * 2 + 2], 16).ok()?;
}
Some(bytes)
}

237
crates/node/src/plugins/distribution_page.html
View file

@ -80,6 +80,13 @@
.state-suspect { color: #ff9800; }
.state-dead { color: #f44336; }
.repeer-btn {
background: none; border: 1px solid #6366f1; color: #6366f1;
padding: 1px 6px; font-size: 9px; font-family: inherit;
border-radius: 3px; cursor: pointer; margin-left: 4px;
}
.repeer-btn:hover { background: #6366f1; color: #fff; }
.bottom-panel {
grid-column: 1 / -1; border-top: 1px solid #2a2d3e; background: #161822;
display: flex; gap: 12px; padding: 12px 16px; height: 200px;
@ -171,6 +178,7 @@
</nav>
</div>
<div class="header-right">
<span id="versionTag" style="color:#555;font-size:10px;"></span>
<span id="nodeLabel" style="color:#888;font-size:12px;">Waiting for data...</span>
<button class="share-btn" id="shareBtn" onclick="openShareModal()">Share</button>
</div>
@ -215,7 +223,7 @@
<h2>Members <span id="memberCount" style="color:#555;font-weight:400;"></span></h2>
<div class="table-scroll">
<table>
<thead><tr><th>State</th><th>Node ID</th><th>Address</th><th>Inc</th></tr></thead>
<thead><tr><th>State</th><th>Node ID</th><th>Address</th><th>Inc</th><th></th></tr></thead>
<tbody id="membersBody"></tbody>
</table>
</div>
@ -255,10 +263,11 @@
<div class="scroll-wrap" id="peersPanel">
<div id="peersList"></div>
<div style="margin-top:8px;display:flex;gap:4px;">
<input id="peerNodeId" placeholder="Node ID (hex)" style="flex:1;background:#1c1f2e;border:1px solid #2a2d3e;color:#e0e0e0;padding:3px 6px;font-size:10px;font-family:monospace;border-radius:3px;" />
<input id="peerNodeId" placeholder="Invite code or Node ID" style="flex:1;background:#1c1f2e;border:1px solid #2a2d3e;color:#e0e0e0;padding:3px 6px;font-size:10px;font-family:monospace;border-radius:3px;" />
<input id="peerLabel" placeholder="Label" style="width:80px;background:#1c1f2e;border:1px solid #2a2d3e;color:#e0e0e0;padding:3px 6px;font-size:10px;font-family:monospace;border-radius:3px;" />
<button onclick="addPeer()" style="background:#6366f1;color:#fff;border:none;padding:3px 8px;font-size:10px;border-radius:3px;cursor:pointer;">Add</button>
</div>
<div id="peerAddStatus" style="font-size:9px;margin-top:4px;min-height:12px;"></div>
</div>
</div>
</div>
@ -632,7 +641,8 @@
ctx.globalAlpha = opacity;
ctx.font = Math.round(9 / vs) + 'px monospace';
ctx.textAlign = 'center';
var label = (i === 0 && data && data.node_name) ? data.node_name : (nodeIds[i] ? nodeIds[i].substring(0, 8) : '');
var memberName = (i > 0 && data && data.members[i - 1]) ? data.members[i - 1].node_name : null;
var label = (i === 0 && data && data.node_name) ? data.node_name : (memberName || (nodeIds[i] ? nodeIds[i].substring(0, 8) : ''));
ctx.fillText(label, posX[i], posY[i] - nr - 3/vs);
ctx.globalAlpha = 1;
}
@ -696,6 +706,7 @@
// Self label
document.getElementById('selfLabel').textContent = 'Node: ' + (d.node_name || d.node_id.substring(0, 16) + '\u2026');
document.getElementById('nodeLabel').textContent = d.node_name || d.listen_addr;
document.getElementById('versionTag').textContent = d.version || '';
// Stats cards
document.getElementById('statMembers').textContent = d.members.length;
@ -721,16 +732,22 @@
var m = d.members[i];
var cls = 'state-' + m.state;
var tr = document.createElement('tr');
var idCell = m.node_id.substring(0, 16) + '\u2026';
if (m.label) idCell = m.label + ' <span style="color:#555;">(' + m.node_id.substring(0, 8) + ')</span>';
var displayName = m.label || m.node_name;
var idCell = displayName
? displayName + ' <span style="color:#555;">(' + m.node_id.substring(0, 8) + ')</span>'
: m.node_id.substring(0, 16) + '\u2026';
var relayBadge = m.relay_url
? ' <span class="relay-badge" title="' + m.relay_url + '">RELAY</span>'
: '';
var actionCell = m.state === 'dead'
? '<button class="repeer-btn" onclick="repeerNode(\'' + m.node_id + '\')">Re-peer</button>'
: '';
tr.innerHTML =
'<td class="' + cls + '">' + m.state + '</td>' +
'<td style="color:#aaa;font-size:10px;">' + idCell + relayBadge + '</td>' +
'<td>' + m.addr + '</td>' +
'<td>' + m.incarnation + '</td>';
'<td>' + m.incarnation + '</td>' +
'<td>' + actionCell + '</td>';
body.appendChild(tr);
}
@ -763,9 +780,13 @@
var addr = mem ? mem.addr : '\u2014';
var cls = 'state-' + state;
var tr = document.createElement('tr');
var probeName = mem && (mem.label || mem.node_name);
var probeLabel = probeName
? probeName + ' <span style="color:#555;">(' + pid.substring(0, 8) + ')</span>'
: pid.substring(0, 12) + '\u2026';
tr.innerHTML =
'<td class="' + cls + '" style="font-size:10px;">' + state + '</td>' +
'<td style="color:#aaa;font-size:10px;">' + pid.substring(0, 12) + '\u2026</td>' +
'<td style="color:#aaa;font-size:10px;">' + probeLabel + '</td>' +
'<td style="font-size:10px;">' + addr + '</td>';
probesBody.appendChild(tr);
}
@ -824,6 +845,7 @@
// ── SSE connection ─────────────────────────────────────────────
var es = new EventSource('/events');
window.addEventListener('beforeunload', function() { es.close(); });
es.addEventListener('distribution', function(e) {
try {
@ -873,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 = '<span class="join-status-text" style="color:' + color + ';margin-left:auto;cursor:' + (js ? 'pointer' : 'default') + ';"'
+ (js ? ' onclick="showJoinDetail(\'' + p.node_id + '\')"' : '')
+ '>' + statusText + '</span>';
var actionBtns = '';
if (js && js.phase === 'failed') {
actionBtns =
'<button onclick="repeerNode(\'' + p.node_id + '\')" style="background:none;border:1px solid #6366f1;color:#6366f1;padding:1px 4px;font-size:9px;border-radius:2px;cursor:pointer;margin-left:2px;">Retry</button>' +
'<button onclick="dismissJoinStatus(\'' + p.node_id + '\')" style="background:none;border:1px solid #555;color:#555;padding:1px 4px;font-size:9px;border-radius:2px;cursor:pointer;">\u00d7</button>';
}
div.innerHTML =
'<span style="width:6px;height:6px;border-radius:50%;background:' + color + ';display:inline-block;"></span>' +
'<span style="width:6px;height:6px;border-radius:50%;background:' + color + ';display:inline-block;flex-shrink:0;"></span>' +
'<span style="color:#aaa;">' + p.node_id.substring(0, 16) + '\u2026</span>' +
'<span style="color:#666;">' + (p.label || '') + '</span>' +
'<span style="color:' + color + ';margin-left:auto;">' + state + '</span>' +
statusSpan + actionBtns +
'<button onclick="removePeer(\'' + p.node_id + '\')" style="background:none;border:1px solid #f44336;color:#f44336;padding:1px 4px;font-size:9px;border-radius:2px;cursor:pointer;">x</button>';
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: <base58>#<addr1>,<addr2>@<relay_url>
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 =
'<div style="display:flex;justify-content:space-between;align-items:center;margin-bottom:8px;">' +
'<span style="color:#888;font-weight:600;">Join Status Detail</span>' +
'<button onclick="document.getElementById(\'joinDetailPopup\').remove()" style="background:none;border:none;color:#888;cursor:pointer;font-size:14px;">\u00d7</button>' +
'</div>' +
'<div style="color:#aaa;margin-bottom:4px;">Node: ' + nid.substring(0, 16) + '\u2026</div>' +
'<div style="background:#0f1117;border-radius:3px;padding:8px;font-family:monospace;font-size:10px;color:#ccc;">' +
methodLines.join('<br>') +
'</div>';
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',
@ -912,6 +1074,17 @@
}).then(function() { fetchPeers(); });
};
window.repeerNode = function(nid) {
fetch('/api/plugin/distribution/rejoin', {
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) { console.log('Re-peer triggered for ' + nid.substring(0, 8)); }
else { console.error('Re-peer failed:', d.error); }
}).catch(function(e) { console.error('Re-peer error:', e); });
};
// Poll peers every 5 seconds
fetchPeers();
setInterval(fetchPeers, 5000);
@ -940,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) ──

62
crates/node/src/plugins/peers.rs
View file

@ -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: <base58>#<addr1>,<addr2>@<relay_url>
// 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<std::net::SocketAddr> = direct_addrs_str
.map(|s| {
s.split(',')
.filter_map(|a| a.trim().parse::<std::net::SocketAddr>().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<u8> = 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());

6
crates/simulation/src/distribution/sim.rs
View file

@ -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<Option
}
SimAction::Join { node_idx, seed_idx } => {
if *node_idx < n && *seed_idx < n {
// Mirror IrohDriver::join(): clear dead state before re-peering
// so stale Dead gossip doesn't leak from the dissemination queue.
if let Some(ref mut joining_node) = nodes[*node_idx] {
joining_node.clear_dead_member(node_ids[*seed_idx]);
}
if let Some(ref mut seed_node) = nodes[*seed_idx] {
let join_actions = seed_node.handle_join_request(node_ids[*node_idx]);
let tagged_responses = deliver_actions_tagged_with_net(

90
crates/simulation/tests/cluster_scenarios.rs
View file

@ -142,6 +142,7 @@ fn cluster_converges_under_10_percent_message_loss() {
indirect_probes: 3,
suspicion_timeout: 60,
dead_reprobe_interval: 15,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
network_faults: vec![NetworkFault::SetDropRate {
round: 1,
@ -182,6 +183,7 @@ fn heavy_message_loss_causes_membership_instability() {
indirect_probes: 2,
suspicion_timeout: 15,
dead_reprobe_interval: 0,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
network_faults: vec![NetworkFault::SetDropRate {
round: 1,
@ -329,6 +331,7 @@ fn cluster_of_fifty_converges() {
indirect_probes: 2,
suspicion_timeout: 10,
dead_reprobe_interval: 0,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..default_config()
};
@ -415,6 +418,7 @@ fn graceful_leave_detected_faster_than_crash() {
indirect_probes: 1,
suspicion_timeout: 5,
dead_reprobe_interval: 0,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..default_config()
};
@ -559,6 +563,7 @@ fn membership_changes_disseminate_to_all_nodes() {
indirect_probes: 2,
suspicion_timeout: 8,
dead_reprobe_interval: 0,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..default_config()
};
@ -646,6 +651,7 @@ fn cluster_survives_brief_message_loss() {
indirect_probes: 3,
suspicion_timeout: 60,
dead_reprobe_interval: 15,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
network_faults: vec![
NetworkFault::SetDropRate {
@ -698,6 +704,7 @@ fn partition_heals_via_dead_reprobe() {
indirect_probes: 1,
suspicion_timeout: 5,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
network_faults: vec![
NetworkFault::Partition {
@ -771,6 +778,7 @@ fn accuracy_no_false_permanent_deaths() {
indirect_probes: 1,
suspicion_timeout: 5,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
network_faults: vec![
NetworkFault::Partition {
@ -812,6 +820,7 @@ fn convergence_after_partition_heal() {
indirect_probes: 1,
suspicion_timeout: 5,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
network_faults: vec![
NetworkFault::Partition {
@ -837,6 +846,86 @@ fn convergence_after_partition_heal() {
);
}
// ────────────────────────────────────────────────────────────────────────────
// 16. Re-peer after partition death — stale Dead gossip must not re-kill
// ────────────────────────────────────────────────────────────────────────────
#[test]
fn repeer_after_partition_death_stays_alive() {
// Scenario: 3 nodes. Partition isolates node 2. Node 2 gets declared Dead
// by nodes 0 and 1 (and vice versa). Partition heals. SimAction::Join
// re-peers all pairs. All nodes must converge to full alive membership.
//
// The purge_node fix ensures that clear_dead_member removes stale
// (node_id, Dead, incarnation) updates from the DisseminationQueue,
// preventing stale Dead gossip from leaking out on subsequent messages
// and re-infecting the cluster during the recovery window.
use simulation::distribution::sim::SimAction;
let config = DistributionSimConfig {
name: "repeer-after-partition-death".into(),
num_nodes: 3,
num_rounds: 100,
ticks_per_round: 3,
actors_per_node: 0,
swim: distribution::swim::probe::SwimConfig {
probe_interval: 1,
probe_timeout: 3,
indirect_probes: 1,
suspicion_timeout: 5,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
network_faults: vec![
// Isolate node 2 from nodes 0 and 1
NetworkFault::Partition {
round: 5,
partition: Partition {
side_a: vec![0, 1],
side_b: vec![2],
asymmetric: false,
},
},
// Heal the partition before the re-peer
NetworkFault::Heal { round: 20 },
],
// Re-peer all cross-partition pairs after heal
action_schedule: vec![
(25, SimAction::Join { node_idx: 0, seed_idx: 2 }),
(25, SimAction::Join { node_idx: 2, seed_idx: 0 }),
(25, SimAction::Join { node_idx: 1, seed_idx: 2 }),
(25, SimAction::Join { node_idx: 2, seed_idx: 1 }),
],
..default_config()
};
let trace = run_simulation(config);
// After re-peer + refutation cycles, all 3 nodes should converge.
// Check the last 10 rounds: every node must be alive and see ≥2 members.
for round_idx in (trace.num_rounds - 10)..trace.num_rounds {
let round_snaps = &trace.snapshots_per_round[round_idx];
for (name, snap) in round_snaps {
assert!(
snap.is_alive,
"round {}: {name} should be alive",
round_idx + 1
);
}
let min_members = round_snaps
.iter()
.filter(|(_, s)| s.is_alive)
.map(|(_, s)| s.member_count)
.min()
.unwrap_or(0);
assert!(
min_members >= 2,
"round {}: all nodes should see full membership (min member_count = {min_members})",
round_idx + 1
);
}
}
// ────────────────────────────────────────────────────────────────────────────
// Postmortem 3f — suspect→refute race: no false death after refutation
// ────────────────────────────────────────────────────────────────────────────
@ -866,6 +955,7 @@ fn suspect_refuted_before_timeout_no_false_death() {
indirect_probes: 1,
suspicion_timeout: 50,
dead_reprobe_interval: 0, // disabled — refutation must happen before death
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
network_faults: vec![
// Block 0→1 (but 1→0 still works)

5
crates/simulation/tests/distribution_lifecycle.rs
View file

@ -196,6 +196,7 @@ fn routing_table_recovers_after_partition_heals() {
indirect_probes: 1,
suspicion_timeout: 5,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
network_faults: vec![
NetworkFault::Partition {
@ -296,6 +297,7 @@ fn partition_then_death_during_partition_then_heal() {
// after kill, well after partition heals.
suspicion_timeout: 100,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..default_config()
};
@ -453,6 +455,7 @@ fn asymmetric_one_way_block_does_not_kill_node() {
// Gossip through intermediate nodes refutes suspicion each cycle.
suspicion_timeout: 500,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
action_schedule: vec![
(5, SimAction::RegisterName { node_idx: 5, name: "target-svc".into() }),
@ -518,6 +521,7 @@ fn names_registered_during_partition_propagate_after_heal() {
// High timeout: cross-partition nodes stay Suspect during 40-round partition
suspicion_timeout: 200,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..default_config()
};
@ -581,6 +585,7 @@ fn bidirectional_suspicion_both_nodes_recover() {
// Must exceed partition duration (20 rounds × 3 ticks = 60 ticks)
suspicion_timeout: 100,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..default_config()
};

2
crates/simulation/tests/distribution_properties.rs
View file

@ -314,6 +314,7 @@ fn asymmetric_partition_registry_converges_after_heal() {
indirect_probes: 1,
suspicion_timeout: 200,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..DistributionSimConfig::default()
};
@ -556,6 +557,7 @@ fn three_way_partition_heals_and_converges() {
// Must exceed partition duration (50 rounds × 3 ticks = 150 ticks)
suspicion_timeout: 200,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..DistributionSimConfig::default()
};

3
crates/simulation/tests/distribution_registry.rs
View file

@ -120,6 +120,7 @@ fn split_brain_naming_converges_after_partition_heals() {
// re_disseminate_all which propagates both sides' registry entries.
suspicion_timeout: 200,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..default_config()
};
@ -564,6 +565,7 @@ fn registry_converges_despite_message_loss() {
// High timeout prevents false deaths during the loss window.
suspicion_timeout: 200,
dead_reprobe_interval: 15,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..default_config()
};
@ -668,6 +670,7 @@ fn suspected_name_owner_recovers_and_registry_survives() {
// Node stays Suspect during the 10-round partition (30 ticks < 200)
suspicion_timeout: 200,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
..default_config()
};

8
crates/simulation/tests/topology_adversarial.rs
View file

@ -43,6 +43,7 @@ fn per_link_degradation_causes_asymmetric_views() {
indirect_probes: 2,
suspicion_timeout: 8,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
topology: Some(NetworkTopology {
locations: vec![
@ -133,6 +134,7 @@ fn relay_penalty_causes_false_suspicions() {
indirect_probes: 1,
suspicion_timeout: 5,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
topology: Some(NetworkTopology {
locations: vec![
@ -195,6 +197,7 @@ fn asymmetric_relay_links_create_view_divergence() {
indirect_probes: 2,
suspicion_timeout: 8,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
topology: Some(NetworkTopology {
locations: vec![
@ -265,6 +268,7 @@ fn relay_flapping_causes_membership_oscillation() {
indirect_probes: 1,
suspicion_timeout: 5,
dead_reprobe_interval: 8,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
topology: Some(NetworkTopology {
locations: vec![
@ -335,6 +339,7 @@ fn hub_saturation_degrades_spoke_connectivity() {
indirect_probes: 2,
suspicion_timeout: 8,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
topology: Some(NetworkTopology {
locations: vec![
@ -405,6 +410,7 @@ fn correlated_nat_gateway_failure() {
indirect_probes: 2,
suspicion_timeout: 8,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
topology: Some(NetworkTopology {
locations: vec![
@ -483,6 +489,7 @@ fn split_brain_with_dual_relays() {
indirect_probes: 2,
suspicion_timeout: 8,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
topology: Some(NetworkTopology {
locations: vec![
@ -578,6 +585,7 @@ fn relay_is_target_causes_isolation_on_death() {
indirect_probes: 2,
suspicion_timeout: 8,
dead_reprobe_interval: 10,
probe_mode: distribution::swim::probe::ProbeMode::Periodic,
},
topology: Some(NetworkTopology {
locations: vec![

8
src/channel.rs
View file

@ -28,6 +28,10 @@ impl<T> HybridChannel<T> {
pub fn pop(&self) -> Option<T> {
self.ring.pop().or_else(|| self.overflow.pop())
}
pub fn is_empty(&self) -> bool {
self.ring.is_empty() && self.overflow.is_empty()
}
}
pub(crate) struct Receiver<T> {
@ -44,6 +48,10 @@ impl<T> Receiver<T> {
self.queue.pop()
}
pub fn is_empty(&self) -> bool {
self.queue.is_empty()
}
pub fn new_sender(&self) -> Sender<T> {
Sender {
queue: self.queue.clone(),

27
src/config.rs
View file

@ -1,28 +1,3 @@
/// Backoff policy for worker threads when idle.
///
/// Workers spin → yield → sleep with increasing delay when no work is available.
pub struct BackoffPolicy {
/// Number of idle ticks before switching from spin to yield.
pub spin_threshold: u32,
/// Number of idle ticks before switching from yield to sleep.
pub yield_threshold: u32,
/// Microseconds added per tick beyond the yield threshold.
pub sleep_increment_us: u64,
/// Maximum sleep duration in microseconds.
pub sleep_max_us: u64,
}
impl Default for BackoffPolicy {
fn default() -> Self {
Self {
spin_threshold: 64,
yield_threshold: 256,
sleep_increment_us: 50,
sleep_max_us: 1000,
}
}
}
/// What to do when a bounded mailbox is full.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MailboxOverflow {
@ -37,7 +12,6 @@ pub struct RuntimeConfig {
pub max_actors: usize,
pub channel_buffer_size: usize,
pub num_threads: usize,
pub backoff_policy: BackoffPolicy,
/// Maximum messages processed per actor per tick.
/// Prevents a single actor with a large mailbox from starving others.
/// `0` means unlimited (drain entire mailbox).
@ -67,7 +41,6 @@ impl Default for RuntimeConfig {
max_actors: DEFAULT_MAX_ACTORS,
channel_buffer_size: DEFAULT_CHANNEL_BUFFER_SIZE,
num_threads: 1,
backoff_policy: BackoffPolicy::default(),
actor_message_budget: DEFAULT_ACTOR_MESSAGE_BUDGET,
default_mailbox_capacity: 0,
mailbox_overflow: MailboxOverflow::DropNewest,

4
src/extension.rs
View file

@ -53,6 +53,10 @@ pub trait RuntimeExtension: Send + Sync {
/// - `handle_request`: phase 5.5 — processes deferred requests from handlers
/// - `gc_dead`: after cleanup_dead — removes state for dead actors
pub trait WorkerExtension: Send {
/// Returns `true` if this extension has pending work (e.g., active timers).
/// Used by the fast idle path to avoid unnecessary ticks.
fn has_pending_work(&self) -> bool { false }
/// Called each tick before tick_all. Returns messages to deliver.
fn on_tick(&mut self) -> Vec<(ActorAddress, Box<dyn Any + Send>)>;

2
src/runtime.rs
View file

@ -10,7 +10,7 @@ use crate::Instant;
use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Environment, ExitValue, Message, ResumeSignal, SpawnRequest, StopSignal, StopWithSignal, SystemInfo};
use crate::channel::{Receiver, Sender};
// Re-export config types so existing code using `runtime::RuntimeConfig` still works
pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig};
pub use crate::config::{MailboxOverflow, RuntimeConfig};
use crate::delivery::{AddressMap, Envelope, InboxRegistry, Placement, TickContext, WorkerId};
use crate::extension::RuntimeExtension;
use crate::stats::{StatsHook, WorkerStats};

7
src/std/timer_wheel.rs
View file

@ -127,7 +127,14 @@ impl TimerWheel {
}
impl WorkerExtension for TimerWheel {
fn has_pending_work(&self) -> bool {
!self.once_timers.is_empty() || !self.interval_timers.is_empty()
}
fn on_tick(&mut self) -> Vec<(ActorAddress, Box<dyn Any + Send>)> {
if self.once_timers.is_empty() && self.interval_timers.is_empty() {
return Vec::new();
}
self.fire()
}

39
src/worker.rs
View file

@ -51,6 +51,9 @@ pub(crate) struct Worker {
snapshot_buf: Vec<ActorSnapshot>,
/// Per-worker extension (e.g., timer wheel). Created by RuntimeExtension factory.
pub(crate) worker_ext: Option<Box<dyn WorkerExtension>>,
/// True if the previous tick did work — ensures one full tick follows a productive
/// tick so pending_local messages delivered to mailboxes get drained.
has_backlog: bool,
}
impl Worker {
@ -70,6 +73,7 @@ impl Worker {
stats,
snapshot_buf: Vec::new(),
worker_ext: None,
has_backlog: false,
}
}
@ -154,6 +158,16 @@ impl Worker {
#[cfg(feature = "tracing")]
let _span = tracing::trace_span!("worker.tick", worker_id = self.id.0).entered();
// Fast idle path: skip the entire tick when nothing could have changed.
// Cost: ~3 atomic loads, zero syscalls, zero actor iteration.
if !self.has_backlog
&& self.spawn_rx.is_empty()
&& self.transfer_rx.is_empty()
&& !self.worker_ext.as_ref().map_or(false, |e| e.has_pending_work())
{
return false;
}
let mut did_work = false;
let t0 = Instant::now();
@ -285,6 +299,7 @@ impl Worker {
// 7. Clean up poisoned and stopping actors
did_work |= self.cleanup_dead_actors(tc);
self.has_backlog = did_work;
did_work
}
@ -292,27 +307,11 @@ impl Worker {
#[cfg(feature = "tracing")]
let _span = tracing::info_span!("worker.run", worker_id = self.id.0).entered();
let backoff = &tc.config.backoff_policy;
let mut idle_count: u32 = 0;
while is_running.load(Ordering::Acquire) {
let did_work = self.tick_once(tc);
if did_work {
idle_count = 0;
} else {
idle_count = idle_count.saturating_add(1);
if idle_count < backoff.spin_threshold {
// Hot spin
} else if idle_count < backoff.yield_threshold {
thread::yield_now();
} else {
let micros = std::cmp::min(
(idle_count - backoff.yield_threshold) as u64 * backoff.sleep_increment_us,
backoff.sleep_max_us,
);
// park_timeout allows instant wakeup via Thread::unpark()
// when new work arrives (send_to/spawn notify the target worker)
thread::park_timeout(std::time::Duration::from_micros(micros));
}
if !self.tick_once(tc) {
// Park indefinitely — woken by unpark() from send_to/spawn/stop/shutdown.
// Spurious wakes hit the fast idle path (~3 atomic loads) and park again.
thread::park();
}
}
}

2
tests/integration/Cargo.toml
View file

@ -6,7 +6,7 @@ publish = false
[dev-dependencies]
swactor-datastore = { path = "../../crates/datastore", features = ["node"] }
distribution = { path = "../../crates/distribution", features = ["iroh"] }
distribution = { path = "../../crates/distribution", features = ["iroh", "relay"] }
dashboard = { path = "../../crates/dashboard" }
swactor = { path = "../..", features = ["serde", "transport"] }
iroh = "0.96"

2
tests/integration/tests/stream_integration.rs
View file

@ -20,6 +20,8 @@ fn make_driver_with_streams() -> IrohDriver {
node: DistributedNodeConfig::default(),
peer_auth: None,
additional_alpns: vec![swactor_datastore::streams::ALPN.to_vec()],
embedded_relay_bind: None,
relay_public_ip: None,
})
.expect("create iroh driver")
}

4
tests/test_python.py
View file

@ -38,10 +38,6 @@ class TestRuntimeConfig(unittest.TestCase):
self.assertEqual(cfg.num_threads, 1)
self.assertEqual(cfg.max_actors, 1000)
self.assertEqual(cfg.channel_buffer_size, 1000)
self.assertEqual(cfg.spin_threshold, 64)
self.assertEqual(cfg.yield_threshold, 256)
self.assertEqual(cfg.sleep_increment_us, 50)
self.assertEqual(cfg.sleep_max_us, 1000)
def test_custom(self):
cfg = RuntimeConfig(num_threads=4, max_actors=500)