mirror of
https://github.com/jmcorgan/fips.git
synced 2026-07-30 19:46:15 +00:00
Johnathan Corgan
6c5fd3f4b0
Introduce a sans-IO lifecycle supervisor: a synchronous step(event) ->
[action] state machine (SupervisorFsm) that authors the substrate-child
spawn and teardown order, plus an owner struct (Supervisor) holding the
substrate runtime fields and embedding the FSM. The substrate-lifecycle
fields leave Node's flat list into the owner - state, packet_tx, the TUN
reader/writer handles + shutdown fd + channels, the DNS task + identity
channel, the Nostr/LAN rendezvous drivers, and the encrypt/decrypt worker
pools; the dataplane keeps packet_rx.
start()/stop() become the driver executing the FSM's SpawnChild/StopChild
actions: same children, same order, same warn/debug-and-continue on
optional failures, same logs, same NodeState transitions. Behavior- and
wire-neutral. The only determinism change is that transports now tear
down in ascending-id order, previously nondeterministic HashMap iteration.
The bounded Draining phase and the Running{Full|Degraded} health split
land as separate follow-on commits.
2779 lines
114 KiB
Rust
2779 lines
114 KiB
Rust
//! Node lifecycle management: start, stop, and peer connection initiation.
|
|
|
|
pub(crate) mod supervisor;
|
|
|
|
use super::{Node, NodeError, NodeState};
|
|
use supervisor::{Action, Child, Event, SupervisorFsm};
|
|
|
|
use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
|
|
use crate::node::acl::PeerAclContext;
|
|
use crate::nostr::{BootstrapEvent, NostrRendezvous};
|
|
use crate::nostr::{BootstrapHandoffResult, EstablishedTraversal};
|
|
use crate::peer::PeerConnection;
|
|
use crate::proto::fmp::wire::build_msg1;
|
|
use crate::proto::fmp::{Disconnect, DisconnectReason};
|
|
use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, packet_channel};
|
|
use crate::upper::tun::{TunDevice, TunState, run_tun_reader, shutdown_tun_interface};
|
|
use crate::{NodeAddr, PeerIdentity};
|
|
use std::collections::{HashMap, HashSet};
|
|
use std::net::SocketAddr;
|
|
use std::thread;
|
|
use std::time::Duration;
|
|
use tracing::{debug, info, warn};
|
|
|
|
const OPEN_DISCOVERY_RETRY_LIFETIME_MULTIPLIER: u64 = 2;
|
|
const MAX_PARALLEL_PATH_CANDIDATES_PER_PEER: usize = 4;
|
|
const MAX_DISCOVERY_CONNECTS_PER_TICK: usize = 16;
|
|
|
|
fn socket_addr_families_compatible(local: SocketAddr, remote: SocketAddr) -> bool {
|
|
matches!(
|
|
(local, remote),
|
|
(SocketAddr::V4(_), SocketAddr::V4(_)) | (SocketAddr::V6(_), SocketAddr::V6(_))
|
|
)
|
|
}
|
|
|
|
impl Node {
|
|
/// Replace the runtime peer list.
|
|
///
|
|
/// Newly added auto-connect peers are dialed immediately, removed peers
|
|
/// are dropped from retry bookkeeping, and existing peers get fresh
|
|
/// address hints without tearing down an active link. If an existing peer
|
|
/// is already connected and a new concrete candidate appears, FIPS starts
|
|
/// an alternate handshake in parallel; promotion switches only after that
|
|
/// handshake authenticates.
|
|
pub async fn update_peers(
|
|
&mut self,
|
|
new_peers: Vec<PeerConfig>,
|
|
) -> Result<crate::node::UpdatePeersOutcome, NodeError> {
|
|
let mut new_by_addr: HashMap<NodeAddr, PeerConfig> =
|
|
HashMap::with_capacity(new_peers.len());
|
|
for peer in new_peers {
|
|
let identity =
|
|
PeerIdentity::from_npub(&peer.npub).map_err(|e| NodeError::InvalidPeerNpub {
|
|
npub: peer.npub.clone(),
|
|
reason: e.to_string(),
|
|
})?;
|
|
new_by_addr.insert(*identity.node_addr(), peer);
|
|
}
|
|
|
|
// Read the current peer set from the context *before* the swap below:
|
|
// update_peers is the config-source mutation owner. It reads the current
|
|
// (pre-update) peer set here, builds a fresh Config + context, then swaps
|
|
// the whole Arc. Reading the live context Arc before the swap yields the
|
|
// pre-update set the diff needs (the `update_peers_races_*` canary depends
|
|
// on this ordering).
|
|
let current_by_addr: HashMap<NodeAddr, PeerConfig> = self
|
|
.context
|
|
.config
|
|
.peers()
|
|
.iter()
|
|
.filter_map(|peer| {
|
|
PeerIdentity::from_npub(&peer.npub)
|
|
.ok()
|
|
.map(|identity| (*identity.node_addr(), peer.clone()))
|
|
})
|
|
.collect();
|
|
|
|
let new_addrs: HashSet<_> = new_by_addr.keys().copied().collect();
|
|
let current_addrs: HashSet<_> = current_by_addr.keys().copied().collect();
|
|
|
|
let removed: Vec<_> = current_addrs.difference(&new_addrs).copied().collect();
|
|
let added: Vec<_> = new_addrs.difference(¤t_addrs).copied().collect();
|
|
let kept: Vec<_> = new_addrs.intersection(¤t_addrs).copied().collect();
|
|
|
|
let mut outcome = crate::node::UpdatePeersOutcome::default();
|
|
let mut refresh_configs = Vec::new();
|
|
|
|
for node_addr in &removed {
|
|
if self.retry_pending.remove(node_addr).is_some() {
|
|
debug!(
|
|
peer = %self.peer_display_name(node_addr),
|
|
"Dropping retry entry for peer removed from runtime peer list"
|
|
);
|
|
}
|
|
self.peer_aliases.remove(node_addr);
|
|
outcome.removed += 1;
|
|
}
|
|
|
|
for node_addr in &kept {
|
|
let new_peer = &new_by_addr[node_addr];
|
|
let current_peer = ¤t_by_addr[node_addr];
|
|
let changed = new_peer.addresses != current_peer.addresses
|
|
|| new_peer.alias != current_peer.alias
|
|
|| new_peer.connect_policy != current_peer.connect_policy
|
|
|| new_peer.auto_reconnect != current_peer.auto_reconnect
|
|
|| new_peer.via_nostr != current_peer.via_nostr;
|
|
|
|
if changed {
|
|
outcome.updated += 1;
|
|
if let Some(state) = self.retry_pending.get_mut(node_addr) {
|
|
state.peer_config = new_peer.clone();
|
|
state.retry_after_ms = Self::now_ms();
|
|
}
|
|
if let Some(alias) = new_peer.alias.clone() {
|
|
self.peer_aliases.insert(*node_addr, alias);
|
|
}
|
|
} else {
|
|
outcome.unchanged += 1;
|
|
}
|
|
|
|
if new_peer.is_auto_connect() && (!new_peer.addresses.is_empty() || new_peer.via_nostr)
|
|
{
|
|
refresh_configs.push(new_peer.clone());
|
|
}
|
|
}
|
|
|
|
let added_configs: Vec<_> = added
|
|
.iter()
|
|
.map(|node_addr| new_by_addr[node_addr].clone())
|
|
.collect();
|
|
|
|
let mut new_config = (*self.context.config).clone();
|
|
new_config.peers = new_by_addr.into_values().collect();
|
|
self.replace_context(|ctx| ctx.config = std::sync::Arc::new(new_config));
|
|
|
|
for peer_config in added_configs {
|
|
outcome.added += 1;
|
|
let Ok(identity) = PeerIdentity::from_npub(&peer_config.npub) else {
|
|
continue;
|
|
};
|
|
let name = peer_config
|
|
.alias
|
|
.clone()
|
|
.unwrap_or_else(|| identity.short_npub());
|
|
self.peer_aliases.insert(*identity.node_addr(), name);
|
|
self.register_identity(*identity.node_addr(), identity.pubkey_full());
|
|
|
|
if peer_config.is_auto_connect()
|
|
&& let Err(err) = self.initiate_peer_connection(&peer_config).await
|
|
{
|
|
debug!(
|
|
npub = %peer_config.npub,
|
|
error = %err,
|
|
"Failed to initiate connection for newly added runtime peer"
|
|
);
|
|
self.schedule_retry(*identity.node_addr(), Self::now_ms());
|
|
}
|
|
}
|
|
|
|
for peer_config in refresh_configs {
|
|
let Ok(identity) = PeerIdentity::from_npub(&peer_config.npub) else {
|
|
continue;
|
|
};
|
|
let node_addr = *identity.node_addr();
|
|
|
|
if self.peers.contains_key(&node_addr) {
|
|
match self
|
|
.try_active_peer_alternative_addresses(&peer_config, identity)
|
|
.await
|
|
{
|
|
Ok(true) => debug!(
|
|
peer = %self.peer_display_name(&node_addr),
|
|
"Started alternate-path handshake for active peer"
|
|
),
|
|
Ok(false) => {}
|
|
Err(err) => debug!(
|
|
npub = %peer_config.npub,
|
|
error = %err,
|
|
"Active peer alternate-path refresh did not start"
|
|
),
|
|
}
|
|
} else {
|
|
match self.initiate_peer_connection(&peer_config).await {
|
|
Ok(()) => {
|
|
let handshake_timeout_secs =
|
|
self.config().node.rate_limit.handshake_timeout_secs;
|
|
if let Some(state) = self.retry_pending.get_mut(&node_addr) {
|
|
state.peer_config = peer_config;
|
|
state.retry_after_ms =
|
|
Self::now_ms().saturating_add(handshake_timeout_secs * 1000);
|
|
}
|
|
}
|
|
Err(err) => {
|
|
debug!(
|
|
npub = %peer_config.npub,
|
|
error = %err,
|
|
"Refreshed peer addresses did not initiate a direct connection"
|
|
);
|
|
self.schedule_retry(node_addr, Self::now_ms());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(outcome)
|
|
}
|
|
|
|
/// Initiate connections to configured static peers.
|
|
///
|
|
/// For each peer configured with AutoConnect policy, creates a link and
|
|
/// peer entry, then starts the Noise handshake by sending the first message.
|
|
pub(super) async fn initiate_peer_connections(&mut self) {
|
|
// Build display name map from all configured peers (alias or short npub),
|
|
// and pre-seed the identity cache from each peer's npub so that TUN packets
|
|
// addressed to a configured peer can be dispatched (and trigger session
|
|
// initiation) immediately on startup — without waiting for the link-layer
|
|
// handshake to complete first.
|
|
let peer_identities: Vec<(PeerIdentity, Option<String>)> = self
|
|
.config()
|
|
.peers()
|
|
.iter()
|
|
.filter_map(|pc| {
|
|
PeerIdentity::from_npub(&pc.npub)
|
|
.ok()
|
|
.map(|id| (id, pc.alias.clone()))
|
|
})
|
|
.collect();
|
|
|
|
for (identity, alias) in peer_identities {
|
|
let name = alias.unwrap_or_else(|| identity.short_npub());
|
|
self.peer_aliases.insert(*identity.node_addr(), name);
|
|
// Pre-seed identity cache. The parity may be wrong (npub is x-only)
|
|
// but will be corrected to the real value when the peer is promoted
|
|
// after a successful Noise handshake.
|
|
self.register_identity(*identity.node_addr(), identity.pubkey_full());
|
|
}
|
|
|
|
// Collect peer configs to avoid borrow conflicts
|
|
let peer_configs: Vec<_> = self.config().auto_connect_peers().cloned().collect();
|
|
|
|
if peer_configs.is_empty() {
|
|
debug!("No static peers configured");
|
|
return;
|
|
}
|
|
|
|
debug!(
|
|
count = peer_configs.len(),
|
|
"Initiating static peer connections"
|
|
);
|
|
|
|
for peer_config in peer_configs {
|
|
if let Err(e) = self.initiate_peer_connection(&peer_config).await {
|
|
warn!(
|
|
npub = %peer_config.npub,
|
|
alias = ?peer_config.alias,
|
|
error = %e,
|
|
"Failed to initiate peer connection"
|
|
);
|
|
// Schedule a retry so transient address-resolution failures
|
|
// (e.g. cached endpoints stale, NAT rebinds, all addresses
|
|
// currently unreachable) recover without a daemon restart.
|
|
if let Ok(peer_identity) = PeerIdentity::from_npub(&peer_config.npub) {
|
|
self.schedule_retry(*peer_identity.node_addr(), Self::now_ms());
|
|
}
|
|
// No-transport failures most often mean the cached overlay
|
|
// advert is pointing at a dead post-NAT-rebind address. The
|
|
// advert cache is read-only inside fetch_advert, so retries
|
|
// would loop on the same dead address until expiry. Force a
|
|
// re-fetch so the next retry tick picks up fresh endpoints.
|
|
if matches!(e, crate::node::NodeError::NoTransportForType(_))
|
|
&& let Some(bootstrap) = self.supervisor.nostr_rendezvous.engine_arc()
|
|
{
|
|
let npub = peer_config.npub.clone();
|
|
tokio::spawn(async move {
|
|
let _ = bootstrap.refetch_advert_for_stale_check(&npub).await;
|
|
});
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Initiate a connection to a single peer.
|
|
///
|
|
/// Creates a link, starts the Noise handshake, and sends the first message.
|
|
pub(super) async fn initiate_peer_connection(
|
|
&mut self,
|
|
peer_config: &crate::config::PeerConfig,
|
|
) -> Result<(), NodeError> {
|
|
// Parse the peer's npub to get their identity
|
|
let peer_identity =
|
|
PeerIdentity::from_npub(&peer_config.npub).map_err(|e| NodeError::InvalidPeerNpub {
|
|
npub: peer_config.npub.clone(),
|
|
reason: e.to_string(),
|
|
})?;
|
|
|
|
let peer_node_addr = *peer_identity.node_addr();
|
|
|
|
// Check if peer already exists (fully authenticated)
|
|
if self.peers.contains_key(&peer_node_addr) {
|
|
debug!(
|
|
npub = %peer_config.npub,
|
|
"Peer already exists, skipping"
|
|
);
|
|
return Ok(());
|
|
}
|
|
|
|
// Check if connection already in progress to this peer
|
|
if self.is_connecting_to_peer(&peer_node_addr) {
|
|
debug!(
|
|
npub = %peer_config.npub,
|
|
"Connection already in progress, skipping"
|
|
);
|
|
return Ok(());
|
|
}
|
|
|
|
self.try_peer_addresses(peer_config, peer_identity, true)
|
|
.await
|
|
}
|
|
|
|
fn is_connecting_to_peer(&self, peer_node_addr: &NodeAddr) -> bool {
|
|
self.connections.values().any(|conn| {
|
|
conn.expected_identity()
|
|
.map(|id| id.node_addr() == peer_node_addr)
|
|
.unwrap_or(false)
|
|
})
|
|
}
|
|
|
|
fn is_connecting_to_peer_on_path(
|
|
&self,
|
|
peer_node_addr: &NodeAddr,
|
|
transport_id: TransportId,
|
|
remote_addr: &TransportAddr,
|
|
) -> bool {
|
|
self.connections.values().any(|conn| {
|
|
conn.expected_identity()
|
|
.map(|id| id.node_addr() == peer_node_addr)
|
|
.unwrap_or(false)
|
|
&& conn.transport_id() == Some(transport_id)
|
|
&& conn.source_addr() == Some(remote_addr)
|
|
}) || self.pending_connects.iter().any(|pending| {
|
|
pending.peer_identity.node_addr() == peer_node_addr
|
|
&& pending.transport_id == transport_id
|
|
&& pending.remote_addr == *remote_addr
|
|
})
|
|
}
|
|
|
|
/// Find a UDP transport whose bound socket can send to `remote_addr`.
|
|
///
|
|
/// LAN discovery can surface both IPv4 and IPv6 addresses for the same
|
|
/// service. A wildcard IPv4 socket cannot send to an IPv6 link-local
|
|
/// target, and vice versa, so callers must choose by socket family rather
|
|
/// than by transport type alone.
|
|
fn find_udp_transport_for_remote_addr(
|
|
&self,
|
|
remote_addr: SocketAddr,
|
|
) -> Option<(TransportId, SocketAddr)> {
|
|
self.transports
|
|
.iter()
|
|
.filter(|(id, handle)| {
|
|
handle.transport_type().name == "udp"
|
|
&& handle.is_operational()
|
|
&& !self.supervisor.nostr_rendezvous.is_bootstrap_transport(id)
|
|
})
|
|
.filter_map(|(id, handle)| {
|
|
let local_addr = handle.local_addr()?;
|
|
socket_addr_families_compatible(local_addr, remote_addr)
|
|
.then_some((*id, local_addr))
|
|
})
|
|
.min_by_key(|(id, _)| id.as_u32())
|
|
}
|
|
|
|
/// Initiate a connection to a peer on a specific transport and address.
|
|
///
|
|
/// For connectionless transports (UDP, Ethernet): allocates a link, starts
|
|
/// the Noise IK handshake, sends msg1, and registers the connection for
|
|
/// msg2 dispatch.
|
|
///
|
|
/// For connection-oriented transports (TCP, Tor): allocates a link and
|
|
/// starts a non-blocking transport connect. The handshake is deferred
|
|
/// until the transport connection is established — the tick handler
|
|
/// polls `connection_state()` and initiates the handshake when ready.
|
|
pub(super) async fn initiate_connection(
|
|
&mut self,
|
|
transport_id: TransportId,
|
|
remote_addr: TransportAddr,
|
|
peer_identity: PeerIdentity,
|
|
) -> Result<(), NodeError> {
|
|
let peer_node_addr = *peer_identity.node_addr();
|
|
|
|
self.authorize_peer(
|
|
&peer_identity,
|
|
PeerAclContext::OutboundConnect,
|
|
transport_id,
|
|
&remote_addr,
|
|
)?;
|
|
|
|
let is_connection_oriented = self
|
|
.transports
|
|
.get(&transport_id)
|
|
.map(|t| t.transport_type().connection_oriented)
|
|
.unwrap_or(false);
|
|
|
|
// Allocate link ID and create link
|
|
let link_id = self.allocate_link_id();
|
|
|
|
let link = if is_connection_oriented {
|
|
Link::new(
|
|
link_id,
|
|
transport_id,
|
|
remote_addr.clone(),
|
|
LinkDirection::Outbound,
|
|
Duration::from_millis(self.config().node.base_rtt_ms),
|
|
)
|
|
} else {
|
|
Link::connectionless(
|
|
link_id,
|
|
transport_id,
|
|
remote_addr.clone(),
|
|
LinkDirection::Outbound,
|
|
Duration::from_millis(self.config().node.base_rtt_ms),
|
|
)
|
|
};
|
|
|
|
self.links.insert(link_id, link);
|
|
|
|
// Add reverse lookup for packet dispatch
|
|
self.addr_to_link
|
|
.insert((transport_id, remote_addr.clone()), link_id);
|
|
|
|
if is_connection_oriented {
|
|
// Connection-oriented: start non-blocking connect, defer handshake
|
|
if let Some(transport) = self.transports.get(&transport_id) {
|
|
match transport.connect(&remote_addr).await {
|
|
Ok(()) => {
|
|
debug!(
|
|
peer = %self.peer_display_name(&peer_node_addr),
|
|
transport_id = %transport_id,
|
|
remote_addr = %remote_addr,
|
|
link_id = %link_id,
|
|
"Transport connect initiated (non-blocking)"
|
|
);
|
|
self.pending_connects.push(super::PendingConnect {
|
|
link_id,
|
|
transport_id,
|
|
remote_addr,
|
|
peer_identity,
|
|
});
|
|
}
|
|
Err(e) => {
|
|
// Clean up link
|
|
self.links.remove(&link_id);
|
|
self.addr_to_link.remove(&(transport_id, remote_addr));
|
|
return Err(NodeError::TransportError(e.to_string()));
|
|
}
|
|
}
|
|
}
|
|
Ok(())
|
|
} else {
|
|
// Connectionless: proceed with immediate handshake
|
|
self.start_handshake(link_id, transport_id, remote_addr, peer_identity)
|
|
.await
|
|
}
|
|
}
|
|
|
|
/// Start the Noise handshake on a link and send msg1.
|
|
///
|
|
/// Called immediately for connectionless transports, or after the
|
|
/// transport connection is established for connection-oriented transports.
|
|
pub(super) async fn start_handshake(
|
|
&mut self,
|
|
link_id: LinkId,
|
|
transport_id: TransportId,
|
|
remote_addr: TransportAddr,
|
|
peer_identity: PeerIdentity,
|
|
) -> Result<(), NodeError> {
|
|
let peer_node_addr = *peer_identity.node_addr();
|
|
|
|
// Create connection in handshake phase (outbound knows expected identity)
|
|
let current_time_ms = Self::now_ms();
|
|
let mut connection = PeerConnection::outbound(link_id, peer_identity, current_time_ms);
|
|
|
|
// Allocate a session index for this handshake
|
|
let our_index = match self.index_allocator.allocate() {
|
|
Ok(idx) => idx,
|
|
Err(e) => {
|
|
// Clean up the link we just created
|
|
self.links.remove(&link_id);
|
|
self.addr_to_link.remove(&(transport_id, remote_addr));
|
|
return Err(NodeError::IndexAllocationFailed(e.to_string()));
|
|
}
|
|
};
|
|
|
|
// Start the Noise handshake and get message 1
|
|
let our_keypair = self.identity().keypair();
|
|
let noise_msg1 =
|
|
match connection.start_handshake(our_keypair, self.startup_epoch(), current_time_ms) {
|
|
Ok(msg) => msg,
|
|
Err(e) => {
|
|
// Clean up the index and link
|
|
let _ = self.index_allocator.free(our_index);
|
|
self.links.remove(&link_id);
|
|
self.addr_to_link.remove(&(transport_id, remote_addr));
|
|
return Err(NodeError::HandshakeFailed(e.to_string()));
|
|
}
|
|
};
|
|
|
|
// Set index and transport info on the connection
|
|
connection.set_our_index(our_index);
|
|
connection.set_transport_id(transport_id);
|
|
connection.set_source_addr(remote_addr.clone());
|
|
|
|
// Build wire format msg1: [0x01][sender_idx:4 LE][noise_msg1:82]
|
|
let wire_msg1 = build_msg1(our_index, &noise_msg1);
|
|
|
|
debug!(
|
|
peer = %self.peer_display_name(&peer_node_addr),
|
|
transport_id = %transport_id,
|
|
remote_addr = %remote_addr,
|
|
link_id = %link_id,
|
|
our_index = %our_index,
|
|
"Connection initiated"
|
|
);
|
|
|
|
// Store msg1 for resend and schedule first resend
|
|
let resend_interval = self.config().node.rate_limit.handshake_resend_interval_ms;
|
|
connection.set_handshake_msg1(wire_msg1.clone(), current_time_ms + resend_interval);
|
|
|
|
// Track in pending_outbound for msg2 dispatch
|
|
self.pending_outbound
|
|
.insert((transport_id, our_index.as_u32()), link_id);
|
|
self.connections.insert(link_id, connection);
|
|
|
|
// Send the wire format handshake message
|
|
if let Some(transport) = self.transports.get(&transport_id) {
|
|
match transport.send(&remote_addr, &wire_msg1).await {
|
|
Ok(bytes) => {
|
|
debug!(
|
|
link_id = %link_id,
|
|
our_index = %our_index,
|
|
bytes,
|
|
"Sent Noise handshake message 1 (wire format)"
|
|
);
|
|
}
|
|
Err(e) => {
|
|
warn!(
|
|
link_id = %link_id,
|
|
error = %e,
|
|
"Failed to send handshake message"
|
|
);
|
|
// Mark connection as failed but don't remove it yet
|
|
// The event loop can handle retry logic
|
|
if let Some(conn) = self.connections.get_mut(&link_id) {
|
|
conn.mark_failed();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Poll all transports for discovered peers and auto-connect.
|
|
///
|
|
/// Called from the tick handler. Iterates operational transports,
|
|
/// drains their discovery buffers, and initiates connections to
|
|
/// newly discovered peers (if auto_connect is enabled).
|
|
pub(super) async fn poll_transport_discovery(&mut self) {
|
|
// Collect discoveries first to avoid borrow conflict with self
|
|
let mut to_connect = Vec::new();
|
|
let mut queued_per_peer: HashMap<NodeAddr, usize> = HashMap::new();
|
|
let mut connect_budget = self.discovery_connect_budget();
|
|
let mut skipped_budget = 0usize;
|
|
|
|
for (transport_id, transport) in &self.transports {
|
|
if !transport.is_operational() {
|
|
continue;
|
|
}
|
|
if !transport.auto_connect() {
|
|
// Still drain the buffer so it doesn't grow unbounded
|
|
let _ = transport.discover();
|
|
continue;
|
|
}
|
|
let discovered = match transport.discover() {
|
|
Ok(peers) => peers,
|
|
Err(_) => continue,
|
|
};
|
|
for peer in discovered {
|
|
let pubkey = match peer.pubkey_hint {
|
|
Some(pk) => pk,
|
|
None => continue,
|
|
};
|
|
let identity = PeerIdentity::from_pubkey(pubkey);
|
|
let node_addr = *identity.node_addr();
|
|
|
|
// Skip self
|
|
if node_addr == *self.identity().node_addr() {
|
|
continue;
|
|
}
|
|
|
|
let candidate_transport_id = *transport_id;
|
|
let remote_addr = peer.addr;
|
|
|
|
if self.peers.contains_key(&node_addr) {
|
|
let transport_name = transport.transport_type().name;
|
|
let candidate = PeerAddress::new(transport_name, remote_addr.to_string());
|
|
if self.active_peer_candidate_is_fresh_enough_to_skip(
|
|
&node_addr,
|
|
std::slice::from_ref(&candidate),
|
|
) {
|
|
continue;
|
|
}
|
|
if self.is_connecting_to_peer_on_path(
|
|
&node_addr,
|
|
candidate_transport_id,
|
|
&remote_addr,
|
|
) {
|
|
continue;
|
|
}
|
|
let queued_for_peer = queued_per_peer.get(&node_addr).copied().unwrap_or(0);
|
|
if connect_budget == 0
|
|
|| self
|
|
.path_candidate_attempt_budget(&node_addr)
|
|
.saturating_sub(queued_for_peer)
|
|
== 0
|
|
{
|
|
skipped_budget = skipped_budget.saturating_add(1);
|
|
continue;
|
|
}
|
|
to_connect.push((candidate_transport_id, remote_addr, identity, true));
|
|
*queued_per_peer.entry(node_addr).or_default() += 1;
|
|
connect_budget = connect_budget.saturating_sub(1);
|
|
continue;
|
|
}
|
|
|
|
if self.is_connecting_to_peer_on_path(
|
|
&node_addr,
|
|
candidate_transport_id,
|
|
&remote_addr,
|
|
) {
|
|
continue;
|
|
}
|
|
let queued_for_peer = queued_per_peer.get(&node_addr).copied().unwrap_or(0);
|
|
if connect_budget == 0
|
|
|| self
|
|
.path_candidate_attempt_budget(&node_addr)
|
|
.saturating_sub(queued_for_peer)
|
|
== 0
|
|
{
|
|
skipped_budget = skipped_budget.saturating_add(1);
|
|
continue;
|
|
}
|
|
|
|
to_connect.push((candidate_transport_id, remote_addr, identity, false));
|
|
*queued_per_peer.entry(node_addr).or_default() += 1;
|
|
connect_budget = connect_budget.saturating_sub(1);
|
|
}
|
|
}
|
|
|
|
if skipped_budget > 0 {
|
|
debug!(
|
|
skipped = skipped_budget,
|
|
queued = to_connect.len(),
|
|
"Transport discovery connect budget exhausted"
|
|
);
|
|
}
|
|
|
|
for (transport_id, remote_addr, identity, active_refresh) in to_connect {
|
|
info!(
|
|
peer = %self.peer_display_name(identity.node_addr()),
|
|
transport_id = %transport_id,
|
|
remote_addr = %remote_addr,
|
|
active_refresh,
|
|
"Auto-connecting to discovered peer"
|
|
);
|
|
if let Err(e) = self
|
|
.initiate_connection(transport_id, remote_addr, identity)
|
|
.await
|
|
{
|
|
warn!(error = %e, "Failed to auto-connect to discovered peer");
|
|
}
|
|
}
|
|
}
|
|
|
|
pub(super) async fn poll_nostr_rendezvous(&mut self) {
|
|
let Some(bootstrap) = self.supervisor.nostr_rendezvous.engine_arc() else {
|
|
return;
|
|
};
|
|
|
|
// Refresh the runtime's outbound-admission view once per tick.
|
|
// The runtime task lives in a separate tokio context with no Node
|
|
// reference, so we publish current capacity state through a
|
|
// cheap atomic store. One-tick lag is acceptable: the inbound
|
|
// msg1 gate in handshake.rs remains the authoritative cap.
|
|
bootstrap.set_outbound_admission(self.outbound_admission_check());
|
|
|
|
if let Err(err) = self.refresh_overlay_advert(&bootstrap).await {
|
|
debug!(error = %err, "Failed to refresh local Nostr overlay advert");
|
|
}
|
|
|
|
for event in bootstrap.drain_events().await {
|
|
match event {
|
|
BootstrapEvent::Established { traversal } => {
|
|
if !self.outbound_admission_check() {
|
|
debug!(
|
|
peer_npub = %traversal.peer_npub,
|
|
peers = self.peers.len(),
|
|
max_peers = self.max_peers(),
|
|
"Dropping established NAT traversal: at capacity"
|
|
);
|
|
continue;
|
|
}
|
|
let peer_npub = traversal.peer_npub.clone();
|
|
if let Ok(peer_identity) = PeerIdentity::from_npub(&peer_npub) {
|
|
let peer_addr = *peer_identity.node_addr();
|
|
if self.peers.contains_key(&peer_addr) {
|
|
debug!(
|
|
peer_npub = %peer_npub,
|
|
"Ignoring established NAT traversal for already-connected peer"
|
|
);
|
|
continue;
|
|
}
|
|
if self.is_connecting_to_peer(&peer_addr) {
|
|
// Dual cross-init: both nodes' Nostr-mediated punches
|
|
// completed simultaneously, and each side already
|
|
// holds in-flight handshake state for the other.
|
|
// Apply the deterministic NodeAddr tie-breaker —
|
|
// smaller NodeAddr wins as adopter (same convention
|
|
// as cross_connection_winner and the rekey dual-
|
|
// init resolution at handshake.rs:269). The winner
|
|
// tears down its in-flight state and adopts the
|
|
// fresh traversal socket; the loser keeps continue
|
|
// semantics, and its existing cross-connection
|
|
// logic in handle_msg1 reconciles when the winner's
|
|
// fresh msg1 arrives over the adopted socket.
|
|
let our_addr = self.identity().node_addr();
|
|
if our_addr >= &peer_addr {
|
|
debug!(
|
|
peer_npub = %peer_npub,
|
|
"Dual cross-init NAT traversal: we lose (larger addr), keeping in-flight handshake"
|
|
);
|
|
continue;
|
|
}
|
|
debug!(
|
|
peer_npub = %peer_npub,
|
|
"Dual cross-init NAT traversal: we win (smaller addr), tearing down in-flight handshake to adopt fresh socket"
|
|
);
|
|
let now_ms = Self::now_ms();
|
|
let stale: Vec<LinkId> = self
|
|
.connections
|
|
.iter()
|
|
.filter(|(_, conn)| {
|
|
conn.expected_identity()
|
|
.map(|id| id.node_addr() == &peer_addr)
|
|
.unwrap_or(false)
|
|
})
|
|
.map(|(link_id, _)| *link_id)
|
|
.collect();
|
|
for link_id in stale {
|
|
self.cleanup_stale_connection(link_id, now_ms);
|
|
}
|
|
}
|
|
}
|
|
match self.adopt_established_traversal(traversal).await {
|
|
Ok(_) => {
|
|
info!(peer_npub = %peer_npub, "Adopted NAT traversal socket");
|
|
}
|
|
Err(err) => {
|
|
warn!(peer_npub = %peer_npub, error = %err, "Failed to adopt NAT traversal");
|
|
if let Ok(peer_identity) = PeerIdentity::from_npub(&peer_npub) {
|
|
self.schedule_retry(*peer_identity.node_addr(), Self::now_ms());
|
|
}
|
|
}
|
|
}
|
|
}
|
|
BootstrapEvent::Failed {
|
|
peer_config,
|
|
reason,
|
|
} => {
|
|
let peer_identity = match PeerIdentity::from_npub(&peer_config.npub) {
|
|
Ok(identity) => identity,
|
|
Err(_) => continue,
|
|
};
|
|
let node_addr = *peer_identity.node_addr();
|
|
if self.peers.contains_key(&node_addr) {
|
|
debug!(
|
|
npub = %peer_config.npub,
|
|
error = %reason,
|
|
"Ignoring failed NAT traversal for already-connected peer"
|
|
);
|
|
continue;
|
|
}
|
|
if self.is_connecting_to_peer(&node_addr) {
|
|
debug!(
|
|
npub = %peer_config.npub,
|
|
error = %reason,
|
|
"Ignoring failed NAT traversal while peer handshake is already in progress"
|
|
);
|
|
continue;
|
|
}
|
|
|
|
let now_ms = Self::now_ms();
|
|
let decision = bootstrap.record_traversal_failure(&peer_config.npub, now_ms);
|
|
if decision.should_warn {
|
|
warn!(
|
|
npub = %peer_config.npub,
|
|
error = %reason,
|
|
consecutive_failures = decision.consecutive_failures,
|
|
cooldown_secs = decision
|
|
.cooldown_until_ms
|
|
.map(|t| t.saturating_sub(now_ms) / 1000),
|
|
"NAT traversal failed"
|
|
);
|
|
} else {
|
|
debug!(
|
|
npub = %peer_config.npub,
|
|
error = %reason,
|
|
consecutive_failures = decision.consecutive_failures,
|
|
"NAT traversal failed (suppressed by warn-rate-limit)"
|
|
);
|
|
}
|
|
|
|
// B6: stale-advert eviction on the streak-threshold
|
|
// crossing. Fire-and-forget; the outcome is logged so
|
|
// operators can see when peers get cleaned up.
|
|
if decision.crossed_threshold {
|
|
let bootstrap = bootstrap.clone();
|
|
let npub = peer_config.npub.clone();
|
|
tokio::spawn(async move {
|
|
let outcome = bootstrap.refetch_advert_for_stale_check(&npub).await;
|
|
match outcome {
|
|
crate::nostr::NostrRefetchOutcome::Evicted => info!(
|
|
npub = %npub,
|
|
"stale-advert sweep: peer evicted from advert cache"
|
|
),
|
|
crate::nostr::NostrRefetchOutcome::Refreshed => info!(
|
|
npub = %npub,
|
|
"stale-advert sweep: peer republished, cache refreshed and streak reset"
|
|
),
|
|
crate::nostr::NostrRefetchOutcome::SameAdvert => debug!(
|
|
npub = %npub,
|
|
"stale-advert sweep: advert unchanged, cooldown stands"
|
|
),
|
|
crate::nostr::NostrRefetchOutcome::Skipped => debug!(
|
|
npub = %npub,
|
|
"stale-advert sweep: skipped (relay error or no advert_relays)"
|
|
),
|
|
}
|
|
});
|
|
}
|
|
|
|
if self
|
|
.try_peer_addresses(&peer_config, peer_identity, false)
|
|
.await
|
|
.is_ok()
|
|
{
|
|
continue;
|
|
}
|
|
|
|
self.schedule_retry(node_addr, now_ms);
|
|
if let Some(cooldown_until_ms) = decision.cooldown_until_ms
|
|
&& let Some(state) = self.retry_pending.get_mut(&node_addr)
|
|
{
|
|
// Push the next retry past the cooldown so the
|
|
// open-discovery sweep doesn't re-enqueue and the
|
|
// per-attempt backoff doesn't fire sooner.
|
|
state.retry_after_ms = state.retry_after_ms.max(cooldown_until_ms);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
self.maybe_run_startup_open_discovery_sweep(&bootstrap)
|
|
.await;
|
|
self.queue_open_discovery_retries(&bootstrap).await;
|
|
}
|
|
|
|
/// Resolve the LAN-only discovery scope. Applications with explicit
|
|
/// connectivity config can set `node.rendezvous.lan.scope` without
|
|
/// changing the public Nostr discovery `app` tag. The older fallback
|
|
/// extracts a scope from the Nostr app tag used by default scoped
|
|
/// discovery.
|
|
pub(super) fn lan_rendezvous_scope(&self) -> Option<String> {
|
|
if let Some(scope) = self.config().node.rendezvous.lan.scope.as_deref() {
|
|
let scope = scope.trim();
|
|
if !scope.is_empty() {
|
|
return Some(scope.to_string());
|
|
}
|
|
}
|
|
|
|
let app = self.config().node.rendezvous.nostr.app.trim();
|
|
if app.is_empty() {
|
|
return None;
|
|
}
|
|
if let Some(rest) = app.strip_prefix("fips-overlay-v1:") {
|
|
let scope = rest.trim();
|
|
if scope.is_empty() {
|
|
None
|
|
} else {
|
|
Some(scope.to_string())
|
|
}
|
|
} else {
|
|
Some(app.to_string())
|
|
}
|
|
}
|
|
|
|
/// Drain mDNS-discovered peers and initiate Noise IK handshakes.
|
|
/// The handshake itself is the authentication — a spoofed mDNS advert
|
|
/// with someone else's npub fails the IK exchange and is dropped.
|
|
pub(super) async fn poll_lan_rendezvous(&mut self) {
|
|
let Some(runtime) = self.supervisor.lan_rendezvous.clone() else {
|
|
return;
|
|
};
|
|
let events = runtime.drain_events().await;
|
|
if events.is_empty() {
|
|
return;
|
|
}
|
|
for event in events {
|
|
let crate::mdns::LanEvent::Discovered(peer) = event;
|
|
let Some((transport_id, local_addr)) =
|
|
self.find_udp_transport_for_remote_addr(peer.addr)
|
|
else {
|
|
debug!(
|
|
addr = %peer.addr,
|
|
"lan: skip discovered peer with no compatible UDP transport"
|
|
);
|
|
continue;
|
|
};
|
|
let identity = match crate::PeerIdentity::from_npub(&peer.npub) {
|
|
Ok(id) => id,
|
|
Err(err) => {
|
|
debug!(npub = %peer.npub, error = %err, "lan: skip bad npub");
|
|
continue;
|
|
}
|
|
};
|
|
let peer_node_addr = *identity.node_addr();
|
|
let remote_addr = crate::transport::TransportAddr::from_string(&peer.addr.to_string());
|
|
if self.peers.contains_key(&peer_node_addr)
|
|
|| self.is_connecting_to_peer(&peer_node_addr)
|
|
{
|
|
continue;
|
|
}
|
|
info!(
|
|
npub = %identity.short_npub(),
|
|
addr = %peer.addr,
|
|
local_addr = %local_addr,
|
|
"lan: initiating handshake to discovered peer"
|
|
);
|
|
if let Err(err) = self
|
|
.initiate_connection(transport_id, remote_addr, identity)
|
|
.await
|
|
{
|
|
debug!(
|
|
npub = %peer.npub,
|
|
error = %err,
|
|
"lan: failed to initiate connection to discovered peer"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Poll pending transport connects and initiate handshakes for ready ones.
|
|
///
|
|
/// Called from the tick handler. For each pending connect, queries the
|
|
/// transport's connection state. When a connection is established,
|
|
/// marks the link as Connected and starts the Noise handshake.
|
|
/// Failed connections are cleaned up and scheduled for retry.
|
|
pub(super) async fn poll_pending_connects(&mut self) {
|
|
if self.pending_connects.is_empty() {
|
|
return;
|
|
}
|
|
|
|
let mut completed = Vec::new();
|
|
|
|
for (i, pending) in self.pending_connects.iter().enumerate() {
|
|
let state = if let Some(transport) = self.transports.get(&pending.transport_id) {
|
|
transport.connection_state(&pending.remote_addr)
|
|
} else {
|
|
crate::transport::ConnectionState::Failed("transport removed".into())
|
|
};
|
|
|
|
match state {
|
|
crate::transport::ConnectionState::Connected => {
|
|
completed.push((i, true, None));
|
|
}
|
|
crate::transport::ConnectionState::Failed(reason) => {
|
|
completed.push((i, false, Some(reason)));
|
|
}
|
|
crate::transport::ConnectionState::Connecting => {
|
|
// Still in progress, check on next tick
|
|
}
|
|
crate::transport::ConnectionState::None => {
|
|
// Shouldn't happen — treat as failure
|
|
completed.push((i, false, Some("no connection attempt found".into())));
|
|
}
|
|
}
|
|
}
|
|
|
|
// Process completions in reverse order to preserve indices
|
|
for (i, success, reason) in completed.into_iter().rev() {
|
|
let pending = self.pending_connects.remove(i);
|
|
|
|
if success {
|
|
// Mark link as Connected
|
|
if let Some(link) = self.links.get_mut(&pending.link_id) {
|
|
link.set_connected();
|
|
}
|
|
|
|
debug!(
|
|
peer = %self.peer_display_name(pending.peer_identity.node_addr()),
|
|
transport_id = %pending.transport_id,
|
|
remote_addr = %pending.remote_addr,
|
|
link_id = %pending.link_id,
|
|
"Transport connected, starting handshake"
|
|
);
|
|
|
|
// Start the handshake now that the transport is connected
|
|
if let Err(e) = self
|
|
.start_handshake(
|
|
pending.link_id,
|
|
pending.transport_id,
|
|
pending.remote_addr.clone(),
|
|
pending.peer_identity,
|
|
)
|
|
.await
|
|
{
|
|
warn!(
|
|
link_id = %pending.link_id,
|
|
error = %e,
|
|
"Failed to start handshake after transport connect"
|
|
);
|
|
// Clean up link on handshake failure
|
|
self.remove_link(&pending.link_id);
|
|
}
|
|
} else {
|
|
let reason = reason.unwrap_or_default();
|
|
warn!(
|
|
peer = %self.peer_display_name(pending.peer_identity.node_addr()),
|
|
transport_id = %pending.transport_id,
|
|
remote_addr = %pending.remote_addr,
|
|
link_id = %pending.link_id,
|
|
reason = %reason,
|
|
"Transport connect failed"
|
|
);
|
|
|
|
// Clean up link and schedule retry
|
|
self.remove_link(&pending.link_id);
|
|
self.links.remove(&pending.link_id);
|
|
self.schedule_retry(*pending.peer_identity.node_addr(), Self::now_ms());
|
|
}
|
|
}
|
|
}
|
|
|
|
// === State Transitions ===
|
|
|
|
/// Start the node.
|
|
///
|
|
/// Initializes the TUN interface (if configured), spawns I/O threads,
|
|
/// and transitions to the Running state.
|
|
pub async fn start(&mut self) -> Result<(), NodeError> {
|
|
if !self.supervisor.state.can_start() {
|
|
return Err(NodeError::AlreadyStarted);
|
|
}
|
|
self.supervisor.state = NodeState::Starting;
|
|
|
|
// Create packet channel for transport -> Node communication
|
|
let packet_buffer_size = self.config().node.buffers.packet_channel;
|
|
let (packet_tx, packet_rx) = packet_channel(packet_buffer_size);
|
|
self.supervisor.packet_tx = Some(packet_tx.clone());
|
|
self.packet_rx = Some(packet_rx);
|
|
|
|
// Initialize transports first (before TUN, before Nostr discovery).
|
|
// Creation allocates each transport's id; the supervisor FSM authors
|
|
// the start order over those ids.
|
|
let transport_handles = self.create_transports(&packet_tx).await;
|
|
let transport_ids: Vec<TransportId> =
|
|
transport_handles.iter().map(|h| h.transport_id()).collect();
|
|
let mut pending_handles: HashMap<_, _> = transport_handles
|
|
.into_iter()
|
|
.map(|h| (h.transport_id(), h))
|
|
.collect();
|
|
|
|
// Singleton child booleans, with today's exact enable conditions.
|
|
let nostr = self.config().node.rendezvous.nostr.enabled;
|
|
let mdns = self.config().node.rendezvous.lan.enabled;
|
|
let tun = self.config().tun.enabled;
|
|
let dns = self.config().dns.enabled;
|
|
|
|
// Worker-pool booleans + counts. Unix only — the workers issue
|
|
// sendmmsg(2) / sendmsg+UDP_GSO on raw fds via `AsRawFd`. Encrypt
|
|
// always spawns on unix; decrypt spawns iff FIPS_DECRYPT_WORKERS != 0.
|
|
// Counts are parsed up-front so the FSM can decide whether the decrypt
|
|
// child exists; the actual spawns run when the SpawnChild actions do.
|
|
#[cfg(unix)]
|
|
let (encrypt_workers, decrypt_workers, encrypt_worker_count, decrypt_worker_count) = {
|
|
let cpu_default = std::thread::available_parallelism()
|
|
.map(|n| n.get())
|
|
.unwrap_or(1)
|
|
.max(1);
|
|
let encrypt_worker_count: usize = std::env::var("FIPS_ENCRYPT_WORKERS")
|
|
.ok()
|
|
.and_then(|s| s.parse().ok())
|
|
.unwrap_or(cpu_default)
|
|
.max(1);
|
|
let decrypt_worker_count: usize = std::env::var("FIPS_DECRYPT_WORKERS")
|
|
.ok()
|
|
.and_then(|s| s.parse().ok())
|
|
.unwrap_or(cpu_default);
|
|
(
|
|
true,
|
|
decrypt_worker_count != 0,
|
|
encrypt_worker_count,
|
|
decrypt_worker_count,
|
|
)
|
|
};
|
|
#[cfg(not(unix))]
|
|
let (encrypt_workers, decrypt_workers) = (false, false);
|
|
|
|
// Ask the supervisor FSM for the canonical spawn order.
|
|
let actions = self.supervisor.fsm.step(Event::Start {
|
|
transports: transport_ids,
|
|
encrypt_workers,
|
|
decrypt_workers,
|
|
nostr,
|
|
mdns,
|
|
tun,
|
|
dns,
|
|
});
|
|
|
|
// Execute each SpawnChild in order, reporting the outcome back so the
|
|
// FSM's up-set tracks what actually came up. Optional failures are
|
|
// warn/debug-and-continue (today's behavior); start still reaches
|
|
// Running. Two driver seams are woven in at their current positions:
|
|
// the post-transport-loop "Transports initialized" info!, and the
|
|
// peer-connect that today sits after mDNS and before TUN.
|
|
let mut transports_info_emitted = false;
|
|
let mut peer_connect_done = false;
|
|
for action in actions {
|
|
let Action::SpawnChild(child) = action else {
|
|
continue;
|
|
};
|
|
|
|
// Post-transport-loop seam: once, after all transport spawns and
|
|
// before the first non-transport child.
|
|
if !transports_info_emitted && !matches!(child, Child::Transport(_)) {
|
|
if !self.transports.is_empty() {
|
|
info!(count = self.transports.len(), "Transports initialized");
|
|
}
|
|
transports_info_emitted = true;
|
|
}
|
|
|
|
// Peer-connect seam: once, immediately before the first Tun-or-Dns
|
|
// child. Connect to static peers before TUN is active so handshake
|
|
// messages can be sent before we start accepting packets.
|
|
if !peer_connect_done && matches!(child, Child::Tun | Child::Dns) {
|
|
self.initiate_peer_connections().await;
|
|
peer_connect_done = true;
|
|
}
|
|
|
|
let feedback = match child {
|
|
Child::Transport(id) => {
|
|
let mut handle = pending_handles
|
|
.remove(&id)
|
|
.expect("supervisor emitted SpawnChild for a created transport");
|
|
let transport_type = handle.transport_type().name;
|
|
let name = handle.name().map(|s| s.to_string());
|
|
|
|
match handle.start().await {
|
|
Ok(()) => {
|
|
self.transports.insert(id, handle);
|
|
Event::SubstrateUp { child }
|
|
}
|
|
Err(e) => {
|
|
if let Some(ref n) = name {
|
|
warn!(transport_type, name = %n, error = %e, "Transport failed to start");
|
|
} else {
|
|
warn!(transport_type, error = %e, "Transport failed to start");
|
|
}
|
|
Event::SubstrateFailed { child }
|
|
}
|
|
}
|
|
}
|
|
Child::EncryptWorkers => {
|
|
// Hash-by-destination pins a TCP flow to one worker
|
|
// (preserves wire ordering); additional workers light up
|
|
// under multi-flow load. Infallible → always up.
|
|
#[cfg(unix)]
|
|
{
|
|
self.supervisor.encrypt_workers = Some(
|
|
super::encrypt_worker::EncryptWorkerPool::spawn(encrypt_worker_count),
|
|
);
|
|
info!(
|
|
workers = encrypt_worker_count,
|
|
"Spawned FMP-encrypt worker pool"
|
|
);
|
|
|
|
// `FIPS_DECRYPT_WORKERS=0` disables the pool entirely
|
|
// and forces the in-line rx_loop decrypt path. When 0
|
|
// no DecryptWorkers child is emitted, so this info!
|
|
// sits here — exactly where the decrypt spawn would be
|
|
// in today's sequence (after the encrypt spawn+info,
|
|
// before nostr).
|
|
if decrypt_worker_count == 0 {
|
|
info!("FIPS_DECRYPT_WORKERS=0 → in-line decrypt in rx_loop");
|
|
}
|
|
}
|
|
Event::SubstrateUp { child }
|
|
}
|
|
Child::DecryptWorkers => {
|
|
// Shard-owned decrypt pool. Infallible → always up.
|
|
#[cfg(unix)]
|
|
{
|
|
self.supervisor.decrypt_workers = Some(
|
|
super::decrypt_worker::DecryptWorkerPool::spawn(decrypt_worker_count),
|
|
);
|
|
info!(
|
|
workers = decrypt_worker_count,
|
|
"Spawned FMP-decrypt worker pool"
|
|
);
|
|
}
|
|
Event::SubstrateUp { child }
|
|
}
|
|
Child::Nostr => {
|
|
match NostrRendezvous::start(
|
|
self.identity(),
|
|
self.config().node.rendezvous.nostr.clone(),
|
|
)
|
|
.await
|
|
{
|
|
Ok(runtime) => {
|
|
if let Err(err) = self.refresh_overlay_advert(&runtime).await {
|
|
warn!(error = %err, "Failed to publish initial Nostr overlay advert");
|
|
}
|
|
self.supervisor.nostr_rendezvous.set_engine(runtime);
|
|
self.supervisor
|
|
.nostr_rendezvous
|
|
.set_started_at_ms(Self::now_ms());
|
|
info!("Nostr overlay discovery enabled");
|
|
Event::SubstrateUp { child }
|
|
}
|
|
Err(err) => {
|
|
warn!(error = %err, "Failed to start Nostr overlay discovery");
|
|
Event::SubstrateFailed { child }
|
|
}
|
|
}
|
|
}
|
|
Child::Mdns => {
|
|
// Advertise the port of a non-bootstrap operational UDP
|
|
// transport. Bootstrap transports must be excluded (they
|
|
// are not the node's listening data-plane socket), and a
|
|
// stable selector (lowest TransportId) is used so the
|
|
// advertised port is deterministic across restarts rather
|
|
// than dependent on HashMap iteration order. This mirrors
|
|
// find_udp_transport_for_remote_addr.
|
|
let advertised_udp_port = self
|
|
.transports
|
|
.iter()
|
|
.filter(|(id, h)| {
|
|
h.transport_type().name == "udp"
|
|
&& h.is_operational()
|
|
&& !self.supervisor.nostr_rendezvous.is_bootstrap_transport(id)
|
|
})
|
|
.filter_map(|(id, h)| h.local_addr().map(|addr| (*id, addr.port())))
|
|
.min_by_key(|(id, _)| id.as_u32())
|
|
.map(|(_, port)| port)
|
|
.unwrap_or(0);
|
|
let scope = self.lan_rendezvous_scope();
|
|
match crate::mdns::LanRendezvous::start(
|
|
self.identity(),
|
|
scope,
|
|
advertised_udp_port,
|
|
self.config().node.rendezvous.lan.clone(),
|
|
)
|
|
.await
|
|
{
|
|
Ok(runtime) => {
|
|
self.supervisor.lan_rendezvous = Some(runtime);
|
|
info!("LAN mDNS discovery enabled");
|
|
Event::SubstrateUp { child }
|
|
}
|
|
Err(err) => {
|
|
debug!(error = %err, "LAN mDNS discovery not started");
|
|
Event::SubstrateFailed { child }
|
|
}
|
|
}
|
|
}
|
|
Child::Tun => {
|
|
// Initialize TUN interface after transports and peers are
|
|
// ready.
|
|
let address = *self.identity().address();
|
|
match TunDevice::create(&self.config().tun, address).await {
|
|
Ok(device) => {
|
|
let mtu = device.mtu();
|
|
let name = device.name().to_string();
|
|
let our_addr = *device.address();
|
|
|
|
info!("TUN device active:");
|
|
info!(" name: {}", name);
|
|
info!(" address: {}", device.address());
|
|
info!(" mtu: {}", mtu);
|
|
|
|
// Calculate max MSS for TCP clamping
|
|
let effective_mtu = self.effective_ipv6_mtu();
|
|
let max_mss = effective_mtu.saturating_sub(40).saturating_sub(20); // IPv6 + TCP headers
|
|
|
|
info!("effective MTU: {} bytes", effective_mtu);
|
|
debug!(" max TCP MSS: {} bytes", max_mss);
|
|
|
|
// On macOS, create a shutdown pipe. Writing to it unblocks the
|
|
// reader thread's select() loop without closing the TUN fd
|
|
// (which would cause a double-close when TunDevice drops).
|
|
#[cfg(target_os = "macos")]
|
|
let (shutdown_read_fd, shutdown_write_fd) = {
|
|
let mut fds = [0i32; 2];
|
|
if unsafe { libc::pipe(fds.as_mut_ptr()) } < 0 {
|
|
return Err(NodeError::Tun(
|
|
crate::upper::tun::TunError::Configure(
|
|
"failed to create shutdown pipe".into(),
|
|
),
|
|
));
|
|
}
|
|
(fds[0], fds[1])
|
|
};
|
|
|
|
// Create writer (dups the fd for independent write access).
|
|
// Pass path_mtu_lookup so inbound SYN-ACK clamp can read
|
|
// per-destination path MTU learned via discovery.
|
|
let (writer, tun_tx) =
|
|
device.create_writer(max_mss, self.path_mtu_lookup.clone())?;
|
|
|
|
// Spawn writer thread
|
|
let writer_handle = thread::spawn(move || {
|
|
writer.run();
|
|
});
|
|
|
|
// Clone tun_tx for the reader
|
|
let reader_tun_tx = tun_tx.clone();
|
|
|
|
// Create outbound channel for TUN reader → Node
|
|
let tun_channel_size = self.config().node.buffers.tun_channel;
|
|
let (outbound_tx, outbound_rx) =
|
|
tokio::sync::mpsc::channel(tun_channel_size);
|
|
|
|
// Spawn reader thread
|
|
let transport_mtu = self.transport_mtu();
|
|
let path_mtu_lookup = self.path_mtu_lookup.clone();
|
|
#[cfg(target_os = "macos")]
|
|
let reader_handle = thread::spawn(move || {
|
|
run_tun_reader(
|
|
device,
|
|
mtu,
|
|
our_addr,
|
|
reader_tun_tx,
|
|
outbound_tx,
|
|
transport_mtu,
|
|
path_mtu_lookup,
|
|
shutdown_read_fd,
|
|
);
|
|
});
|
|
#[cfg(not(target_os = "macos"))]
|
|
let reader_handle = thread::spawn(move || {
|
|
run_tun_reader(
|
|
device,
|
|
mtu,
|
|
our_addr,
|
|
reader_tun_tx,
|
|
outbound_tx,
|
|
transport_mtu,
|
|
path_mtu_lookup,
|
|
);
|
|
});
|
|
|
|
self.tun_state = TunState::Active;
|
|
self.tun_name = Some(name);
|
|
self.supervisor.tun_tx = Some(tun_tx);
|
|
self.supervisor.tun_outbound_rx = Some(outbound_rx);
|
|
self.supervisor.tun_reader_handle = Some(reader_handle);
|
|
self.supervisor.tun_writer_handle = Some(writer_handle);
|
|
#[cfg(target_os = "macos")]
|
|
{
|
|
self.supervisor.tun_shutdown_fd = Some(shutdown_write_fd);
|
|
}
|
|
Event::SubstrateUp { child }
|
|
}
|
|
Err(e) => {
|
|
self.tun_state = TunState::Failed;
|
|
warn!(error = %e, "Failed to initialize TUN, continuing without it");
|
|
Event::SubstrateFailed { child }
|
|
}
|
|
}
|
|
}
|
|
Child::Dns => {
|
|
// Initialize DNS responder (independent of TUN).
|
|
//
|
|
// Default bind_addr is "::1" (IPv6 loopback). The shipped
|
|
// fips-dns-setup configures systemd-resolved via a global
|
|
// /etc/systemd/resolved.conf.d/fips.conf drop-in pointing at
|
|
// [::1]:5354, which sidesteps a Linux IPV6_PKTINFO behaviour
|
|
// where self-destined traffic to fips0's address is attributed
|
|
// to fips0 in PKTINFO and gets silently dropped by the
|
|
// mesh-interface filter in src/upper/dns.rs.
|
|
//
|
|
// For mesh-reachable resolution (rare), set bind_addr: "::"
|
|
// in fips.yaml. The mesh-interface filter remains active to
|
|
// prevent hosts-file alias enumeration in that mode.
|
|
// `IPV6_V6ONLY=0` is set explicitly so IPv4 clients on
|
|
// 127.0.0.1 still reach us regardless of kernel sysctl
|
|
// defaults — but only when bind is on a wildcard / IPv6 path.
|
|
let addr_str = self.config().dns.bind_addr();
|
|
match addr_str.parse::<std::net::IpAddr>() {
|
|
Ok(ip) => {
|
|
let bind = std::net::SocketAddr::new(ip, self.config().dns.port());
|
|
match Self::bind_dns_socket(bind) {
|
|
Ok(socket) => {
|
|
let dns_channel_size = self.config().node.buffers.dns_channel;
|
|
let (identity_tx, identity_rx) =
|
|
tokio::sync::mpsc::channel(dns_channel_size);
|
|
let dns_ttl = self.config().dns.ttl();
|
|
let base_hosts =
|
|
crate::upper::hosts::HostMap::from_peer_configs(
|
|
self.config().peers(),
|
|
);
|
|
let hosts_path = std::path::PathBuf::from(
|
|
crate::upper::hosts::DEFAULT_HOSTS_PATH,
|
|
);
|
|
let reloader = crate::upper::hosts::HostMapReloader::new(
|
|
base_hosts, hosts_path,
|
|
);
|
|
// Resolve the TUN ifindex so the responder can
|
|
// drop queries arriving on the mesh interface
|
|
// (fips0). Without this, the `::` bind exposes
|
|
// /etc/fips/hosts alias probing to any mesh peer.
|
|
// When TUN isn't enabled or the name can't be
|
|
// resolved, `None` disables the filter (there
|
|
// is no mesh surface to defend anyway).
|
|
let mesh_ifindex =
|
|
Self::lookup_mesh_ifindex(self.config().tun.name());
|
|
info!(
|
|
bind = %bind,
|
|
hosts = reloader.hosts().len(),
|
|
mesh_ifindex = ?mesh_ifindex,
|
|
"DNS responder started for .fips domain (auto-reload enabled)"
|
|
);
|
|
let handle =
|
|
tokio::spawn(crate::upper::dns::run_dns_responder(
|
|
socket,
|
|
identity_tx,
|
|
dns_ttl,
|
|
reloader,
|
|
mesh_ifindex,
|
|
));
|
|
self.supervisor.dns_identity_rx = Some(identity_rx);
|
|
self.supervisor.dns_task = Some(handle);
|
|
Event::SubstrateUp { child }
|
|
}
|
|
Err(e) => {
|
|
warn!(bind = %bind, error = %e, "Failed to start DNS responder");
|
|
Event::SubstrateFailed { child }
|
|
}
|
|
}
|
|
}
|
|
Err(e) => {
|
|
warn!(addr = %addr_str, error = %e, "Invalid dns.bind_addr; DNS responder not started");
|
|
Event::SubstrateFailed { child }
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
self.supervisor.fsm.step(feedback);
|
|
}
|
|
|
|
// Seams that never triggered inside the loop: the "Transports
|
|
// initialized" info! when there was no non-transport child, and the
|
|
// peer-connect when there was no Tun/Dns child (today it still runs,
|
|
// after mDNS).
|
|
if !transports_info_emitted && !self.transports.is_empty() {
|
|
info!(count = self.transports.len(), "Transports initialized");
|
|
}
|
|
if !peer_connect_done {
|
|
self.initiate_peer_connections().await;
|
|
}
|
|
|
|
self.supervisor.state = NodeState::Running;
|
|
info!("Node started:");
|
|
info!(" state: {}", self.supervisor.state);
|
|
info!(" transports: {}", self.transports.len());
|
|
info!(" connections: {}", self.connections.len());
|
|
Ok(())
|
|
}
|
|
|
|
/// Bind a UDP socket for the DNS responder.
|
|
///
|
|
/// For IPv6 binds (including `::`), sets `IPV6_V6ONLY=0` so the socket
|
|
/// also accepts IPv4-mapped addresses. This guarantees dual-stack
|
|
/// delivery regardless of `net.ipv6.bindv6only` sysctl on the host —
|
|
/// v4 clients on 127.0.0.1 and v6 clients on the fips0 address both
|
|
/// land on the same socket.
|
|
///
|
|
/// Also enables `IPV6_RECVPKTINFO` on IPv6 sockets so the responder
|
|
/// can learn the arrival interface per packet. The responder uses that
|
|
/// to drop queries arriving on the mesh TUN, closing the hosts-file
|
|
/// probing side-channel created by the `::` bind.
|
|
fn bind_dns_socket(
|
|
addr: std::net::SocketAddr,
|
|
) -> Result<tokio::net::UdpSocket, std::io::Error> {
|
|
use socket2::{Domain, Protocol, Socket, Type};
|
|
let domain = if addr.is_ipv4() {
|
|
Domain::IPV4
|
|
} else {
|
|
Domain::IPV6
|
|
};
|
|
let sock = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
|
|
if addr.is_ipv6() {
|
|
sock.set_only_v6(false)?;
|
|
#[cfg(unix)]
|
|
Self::set_recv_pktinfo_v6(&sock)?;
|
|
}
|
|
sock.set_nonblocking(true)?;
|
|
sock.bind(&addr.into())?;
|
|
tokio::net::UdpSocket::from_std(sock.into())
|
|
}
|
|
|
|
/// Enable `IPV6_RECVPKTINFO` on an IPv6 UDP socket.
|
|
///
|
|
/// After this setsockopt, each `recvmsg()` call on the socket receives
|
|
/// an `IPV6_PKTINFO` control message containing the arrival interface
|
|
/// index, which the DNS responder uses for its mesh-interface filter.
|
|
#[cfg(unix)]
|
|
fn set_recv_pktinfo_v6(sock: &socket2::Socket) -> Result<(), std::io::Error> {
|
|
use std::os::fd::AsRawFd;
|
|
let enable: libc::c_int = 1;
|
|
let ret = unsafe {
|
|
libc::setsockopt(
|
|
sock.as_raw_fd(),
|
|
libc::IPPROTO_IPV6,
|
|
libc::IPV6_RECVPKTINFO,
|
|
&enable as *const _ as *const libc::c_void,
|
|
std::mem::size_of::<libc::c_int>() as libc::socklen_t,
|
|
)
|
|
};
|
|
if ret < 0 {
|
|
return Err(std::io::Error::last_os_error());
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Resolve the mesh TUN interface index by name.
|
|
///
|
|
/// Returns `None` if the interface does not exist (e.g. TUN disabled
|
|
/// or not yet created). A `None` result disables the DNS responder's
|
|
/// mesh-interface filter — safe, because if there is no fips0 there
|
|
/// is no mesh exposure to defend against.
|
|
fn lookup_mesh_ifindex(name: &str) -> Option<u32> {
|
|
#[cfg(unix)]
|
|
{
|
|
let c_name = std::ffi::CString::new(name).ok()?;
|
|
let idx = unsafe { libc::if_nametoindex(c_name.as_ptr()) };
|
|
if idx == 0 { None } else { Some(idx) }
|
|
}
|
|
#[cfg(not(unix))]
|
|
{
|
|
let _ = name;
|
|
None
|
|
}
|
|
}
|
|
|
|
/// Stop the node.
|
|
///
|
|
/// Shuts down TUN interface, stops I/O threads, and transitions to
|
|
/// the Stopped state.
|
|
pub async fn stop(&mut self) -> Result<(), NodeError> {
|
|
if !self.supervisor.state.can_stop() {
|
|
return Err(NodeError::NotStarted);
|
|
}
|
|
self.supervisor.state = NodeState::Stopping;
|
|
info!(state = %self.supervisor.state, "Node stopping");
|
|
|
|
// Reconstruct the supervised up-set from observed runtime presence and
|
|
// let the FSM author the teardown order (dns → nostr → mdns →
|
|
// transports (ascending id) → tun). Worker pools are deliberately
|
|
// absent from the up-set: today's stop() never tears them down.
|
|
let mut up: Vec<Child> = Vec::new();
|
|
if self.supervisor.dns_task.is_some() {
|
|
up.push(Child::Dns);
|
|
}
|
|
if self.supervisor.nostr_rendezvous.engine().is_some() {
|
|
up.push(Child::Nostr);
|
|
}
|
|
if self.supervisor.lan_rendezvous.is_some() {
|
|
up.push(Child::Mdns);
|
|
}
|
|
for id in self.transports.keys() {
|
|
up.push(Child::Transport(*id));
|
|
}
|
|
if self.tun_name.is_some() {
|
|
up.push(Child::Tun);
|
|
}
|
|
self.supervisor.fsm = SupervisorFsm::running_with(up);
|
|
let actions = self.supervisor.fsm.step(Event::Stop);
|
|
|
|
// Execute each StopChild as today's verbatim block. Two driver seams
|
|
// are woven in at their current positions: the shutdown-disconnect
|
|
// fan-out (after Dns, before Nostr), and dropping the packet channels
|
|
// (after all transports, before TUN).
|
|
let mut disconnect_done = false;
|
|
let mut packet_taken = false;
|
|
for action in actions {
|
|
let Action::StopChild(child) = action else {
|
|
continue;
|
|
};
|
|
|
|
// Seam (a): send disconnect notifications to all active peers
|
|
// before closing transports — after any Dns teardown, before
|
|
// everything else.
|
|
if !disconnect_done && !matches!(child, Child::Dns) {
|
|
self.send_disconnect_to_all_peers(DisconnectReason::Shutdown)
|
|
.await;
|
|
disconnect_done = true;
|
|
}
|
|
|
|
// Seam (b): drop the packet channels after all transports have
|
|
// stopped and before the TUN teardown.
|
|
if !packet_taken && matches!(child, Child::Tun) {
|
|
self.supervisor.packet_tx.take();
|
|
self.packet_rx.take();
|
|
packet_taken = true;
|
|
}
|
|
|
|
match child {
|
|
Child::Dns => {
|
|
// Stop DNS responder
|
|
if let Some(handle) = self.supervisor.dns_task.take() {
|
|
handle.abort();
|
|
debug!("DNS responder stopped");
|
|
}
|
|
}
|
|
Child::Nostr => {
|
|
// Stop Nostr overlay discovery background work and withdraw
|
|
// any advert.
|
|
if let Some(bootstrap) = self.supervisor.nostr_rendezvous.take_engine()
|
|
&& let Err(e) = bootstrap.shutdown().await
|
|
{
|
|
warn!(error = %e, "Failed to shutdown Nostr overlay discovery");
|
|
}
|
|
}
|
|
Child::Mdns => {
|
|
// Tear down LAN mDNS responder + browser. Best-effort: the
|
|
// OS will eventually time the advert out via its TTL even if
|
|
// we don't get a clean unregister out before the daemon exits.
|
|
if let Some(lan) = self.supervisor.lan_rendezvous.take() {
|
|
lan.shutdown().await;
|
|
}
|
|
}
|
|
Child::Transport(id) => {
|
|
// Shutdown transport (they're packet producers)
|
|
if let Some(mut handle) = self.transports.remove(&id) {
|
|
let transport_type = handle.transport_type().name;
|
|
match handle.stop().await {
|
|
Ok(()) => {
|
|
info!(transport_id = %id, transport_type, "Transport stopped");
|
|
}
|
|
Err(e) => {
|
|
warn!(
|
|
transport_id = %id,
|
|
transport_type,
|
|
error = %e,
|
|
"Transport stop failed"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
Child::Tun => {
|
|
// Shutdown TUN interface
|
|
if let Some(name) = self.tun_name.take() {
|
|
info!(name = %name, "Shutting down TUN interface");
|
|
|
|
// Drop the tun_tx to signal the writer to stop
|
|
self.supervisor.tun_tx.take();
|
|
|
|
// Delete the interface (on Linux, causes reader to get EFAULT)
|
|
if let Err(e) = shutdown_tun_interface(&name).await {
|
|
warn!(name = %name, error = %e, "Failed to shutdown TUN interface");
|
|
}
|
|
|
|
// On macOS, signal the reader thread to exit by writing to the
|
|
// shutdown pipe. The reader's select() will wake up and break.
|
|
#[cfg(target_os = "macos")]
|
|
if let Some(fd) = self.supervisor.tun_shutdown_fd.take() {
|
|
unsafe {
|
|
libc::write(fd, b"x".as_ptr() as *const libc::c_void, 1);
|
|
libc::close(fd);
|
|
}
|
|
}
|
|
|
|
// Wait for threads to finish
|
|
if let Some(handle) = self.supervisor.tun_reader_handle.take() {
|
|
let _ = handle.join();
|
|
}
|
|
if let Some(handle) = self.supervisor.tun_writer_handle.take() {
|
|
let _ = handle.join();
|
|
}
|
|
|
|
self.tun_state = TunState::Disabled;
|
|
}
|
|
}
|
|
Child::EncryptWorkers | Child::DecryptWorkers => {
|
|
// Worker pools are never torn down in stop() (matches
|
|
// today); the FSM never emits StopChild for them, so this
|
|
// is unreachable.
|
|
}
|
|
}
|
|
|
|
self.supervisor.fsm.step(Event::ChildStopped { child });
|
|
}
|
|
|
|
// Seams that never triggered inside the loop (no non-Dns child for the
|
|
// disconnect fan-out, no Tun child for dropping the packet channels).
|
|
if !disconnect_done {
|
|
self.send_disconnect_to_all_peers(DisconnectReason::Shutdown)
|
|
.await;
|
|
}
|
|
if !packet_taken {
|
|
self.supervisor.packet_tx.take();
|
|
self.packet_rx.take();
|
|
}
|
|
|
|
self.supervisor.state = NodeState::Stopped;
|
|
info!(state = %self.supervisor.state, "Node stopped");
|
|
Ok(())
|
|
}
|
|
|
|
/// Send disconnect notifications to all active peers.
|
|
///
|
|
/// Best-effort: send failures are logged and ignored since the transport
|
|
/// may already be degraded. This runs before transports are shut down.
|
|
async fn send_disconnect_to_all_peers(&mut self, reason: DisconnectReason) {
|
|
// Collect node_addrs to avoid borrow conflict with send helper
|
|
let peer_addrs: Vec<NodeAddr> = self
|
|
.peers
|
|
.iter()
|
|
.filter(|(_, peer)| peer.can_send() && peer.has_session())
|
|
.map(|(addr, _)| *addr)
|
|
.collect();
|
|
|
|
if peer_addrs.is_empty() {
|
|
debug!(
|
|
total_peers = self.peers.len(),
|
|
"No sendable peers for disconnect notification"
|
|
);
|
|
return;
|
|
}
|
|
|
|
let mut sent = 0usize;
|
|
for node_addr in &peer_addrs {
|
|
if self.send_disconnect_to_peer(node_addr, reason).await {
|
|
sent += 1;
|
|
}
|
|
}
|
|
|
|
info!(sent, total = peer_addrs.len(), reason = %reason, "Sent disconnect notifications");
|
|
}
|
|
|
|
/// Send a Disconnect notification to a single peer.
|
|
///
|
|
/// Best-effort: a send failure (peer already gone, transport down) is
|
|
/// logged and swallowed so callers can proceed with teardown regardless.
|
|
/// Returns `true` if the message was sent successfully.
|
|
async fn send_disconnect_to_peer(
|
|
&mut self,
|
|
node_addr: &NodeAddr,
|
|
reason: DisconnectReason,
|
|
) -> bool {
|
|
let plaintext = Disconnect::new(reason).encode();
|
|
match self
|
|
.send_encrypted_link_message(node_addr, &plaintext)
|
|
.await
|
|
{
|
|
Ok(()) => true,
|
|
Err(e) => {
|
|
debug!(
|
|
peer = %self.peer_display_name(node_addr),
|
|
error = %e,
|
|
"Failed to send disconnect (transport may be down)"
|
|
);
|
|
false
|
|
}
|
|
}
|
|
}
|
|
|
|
fn static_peer_addresses(&self, peer_config: &PeerConfig) -> Vec<PeerAddress> {
|
|
peer_config
|
|
.addresses_by_priority()
|
|
.into_iter()
|
|
.cloned()
|
|
.collect()
|
|
}
|
|
|
|
async fn attempt_peer_address_list(
|
|
&mut self,
|
|
peer_config: &PeerConfig,
|
|
peer_identity: PeerIdentity,
|
|
allow_bootstrap_nat: bool,
|
|
addresses: &[PeerAddress],
|
|
) -> Result<(), NodeError> {
|
|
let peer_node_addr = *peer_identity.node_addr();
|
|
let mut attempted = 0usize;
|
|
let max_attempts = self.path_candidate_attempt_budget(&peer_node_addr);
|
|
if max_attempts == 0 {
|
|
return Err(NodeError::NoTransportForType(format!(
|
|
"no outbound slots available for {}",
|
|
peer_config.npub
|
|
)));
|
|
}
|
|
|
|
for addr in addresses {
|
|
if attempted >= max_attempts {
|
|
break;
|
|
}
|
|
if addr.transport == "udp" && addr.addr.eq_ignore_ascii_case("nat") {
|
|
if !allow_bootstrap_nat {
|
|
continue;
|
|
}
|
|
if self
|
|
.supervisor
|
|
.nostr_rendezvous
|
|
.request_nostr_bootstrap(peer_config)
|
|
.await
|
|
{
|
|
attempted = attempted.saturating_add(1);
|
|
}
|
|
continue;
|
|
}
|
|
|
|
let (transport_id, remote_addr) = if addr.transport == "ethernet" {
|
|
match self.resolve_ethernet_addr(&addr.addr) {
|
|
Ok(result) => result,
|
|
Err(e) => {
|
|
debug!(
|
|
transport = %addr.transport,
|
|
addr = %addr.addr,
|
|
error = %e,
|
|
"Failed to resolve Ethernet address"
|
|
);
|
|
continue;
|
|
}
|
|
}
|
|
} else if addr.transport == "ble" {
|
|
#[cfg(bluer_available)]
|
|
{
|
|
match self.resolve_ble_addr(&addr.addr) {
|
|
Ok(result) => result,
|
|
Err(e) => {
|
|
debug!(
|
|
transport = %addr.transport,
|
|
addr = %addr.addr,
|
|
error = %e,
|
|
"Failed to resolve BLE address"
|
|
);
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
#[cfg(not(bluer_available))]
|
|
{
|
|
debug!(transport = %addr.transport, "BLE transport not available on this build");
|
|
continue;
|
|
}
|
|
} else {
|
|
let tid = if addr.transport == "udp"
|
|
&& let Ok(remote_socket_addr) = addr.addr.parse::<SocketAddr>()
|
|
{
|
|
match self.find_udp_transport_for_remote_addr(remote_socket_addr) {
|
|
Some((id, _)) => id,
|
|
None => {
|
|
debug!(
|
|
transport = %addr.transport,
|
|
addr = %addr.addr,
|
|
"No compatible operational UDP transport for address"
|
|
);
|
|
continue;
|
|
}
|
|
}
|
|
} else {
|
|
match self.find_transport_for_type(&addr.transport) {
|
|
Some(id) => id,
|
|
None => {
|
|
debug!(
|
|
transport = %addr.transport,
|
|
addr = %addr.addr,
|
|
"No operational transport for address type"
|
|
);
|
|
continue;
|
|
}
|
|
}
|
|
};
|
|
(tid, TransportAddr::from_string(&addr.addr))
|
|
};
|
|
|
|
if self.is_connecting_to_peer_on_path(&peer_node_addr, transport_id, &remote_addr) {
|
|
continue;
|
|
}
|
|
|
|
match self
|
|
.initiate_connection(transport_id, remote_addr, peer_identity)
|
|
.await
|
|
{
|
|
Ok(()) => attempted = attempted.saturating_add(1),
|
|
Err(e @ NodeError::AccessDenied(_)) => return Err(e),
|
|
Err(e) => {
|
|
debug!(
|
|
npub = %peer_config.npub,
|
|
transport_id = %transport_id,
|
|
error = %e,
|
|
"Connection attempt failed, trying next address"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
if attempted > 0 {
|
|
return Ok(());
|
|
}
|
|
|
|
Err(NodeError::NoTransportForType(format!(
|
|
"no operational transport for any of {}'s addresses",
|
|
peer_config.npub
|
|
)))
|
|
}
|
|
|
|
async fn queue_open_discovery_retries(&mut self, bootstrap: &std::sync::Arc<NostrRendezvous>) {
|
|
self.run_open_discovery_sweep(bootstrap, None, "per-tick")
|
|
.await;
|
|
}
|
|
|
|
/// Open-discovery cache sweep. Iterates the cached overlay adverts and
|
|
/// queues retries for non-configured, not-yet-connected peers.
|
|
///
|
|
/// `max_age_secs`, if set, filters out adverts whose `created_at` is
|
|
/// older than `now - max_age_secs`. The per-tick sweep passes `None`
|
|
/// (relies on the cache's own `valid_until_ms` filter); the one-shot
|
|
/// startup sweep passes `Some(startup_sweep_max_age_secs)`.
|
|
///
|
|
/// `caller` is a short label included in log lines so per-tick and
|
|
/// startup sweeps are distinguishable in operator-facing logs.
|
|
pub(in crate::node) async fn run_open_discovery_sweep(
|
|
&mut self,
|
|
bootstrap: &std::sync::Arc<NostrRendezvous>,
|
|
max_age_secs: Option<u64>,
|
|
caller: &'static str,
|
|
) {
|
|
if !self.config().node.rendezvous.nostr.enabled
|
|
|| self.config().node.rendezvous.nostr.policy
|
|
!= crate::config::NostrRendezvousPolicy::Open
|
|
{
|
|
return;
|
|
}
|
|
|
|
let configured_npubs = self
|
|
.config()
|
|
.peers()
|
|
.iter()
|
|
.map(|peer| peer.npub.clone())
|
|
.collect::<HashSet<_>>();
|
|
let now_ms = Self::now_ms();
|
|
let now_secs = now_ms / 1000;
|
|
let mut enqueue_budget = self.open_discovery_enqueue_budget(&configured_npubs);
|
|
if enqueue_budget == 0 {
|
|
debug!(
|
|
caller = %caller,
|
|
"open-discovery sweep: enqueue budget is 0, skipping"
|
|
);
|
|
return;
|
|
}
|
|
|
|
let candidates = bootstrap.cached_open_discovery_candidates(64).await;
|
|
let cached_count = candidates.len();
|
|
let mut enqueued = 0usize;
|
|
let mut skipped_age = 0usize;
|
|
let mut skipped_configured = 0usize;
|
|
let mut skipped_self = 0usize;
|
|
let mut skipped_connected = 0usize;
|
|
let mut skipped_retry_pending = 0usize;
|
|
let mut skipped_connecting = 0usize;
|
|
let mut skipped_no_endpoints = 0usize;
|
|
let mut skipped_invalid_npub = 0usize;
|
|
let mut skipped_cooldown = 0usize;
|
|
|
|
for (npub, endpoints, created_at_secs) in candidates {
|
|
if enqueue_budget == 0 {
|
|
break;
|
|
}
|
|
|
|
if let Some(max_age) = max_age_secs
|
|
&& now_secs.saturating_sub(created_at_secs) > max_age
|
|
{
|
|
skipped_age = skipped_age.saturating_add(1);
|
|
continue;
|
|
}
|
|
|
|
if configured_npubs.contains(&npub) {
|
|
if let Ok(peer_identity) = PeerIdentity::from_npub(&npub) {
|
|
let node_addr = *peer_identity.node_addr();
|
|
if !self.peers.contains_key(&node_addr)
|
|
&& !self.is_connecting_to_peer(&node_addr)
|
|
&& let Some(state) = self.retry_pending.get_mut(&node_addr)
|
|
&& state.retry_after_ms > now_ms
|
|
{
|
|
state.retry_after_ms = now_ms;
|
|
debug!(
|
|
peer = %peer_identity.short_npub(),
|
|
caller = %caller,
|
|
"open-discovery sweep: fresh configured-peer advert expedited retry"
|
|
);
|
|
}
|
|
}
|
|
skipped_configured = skipped_configured.saturating_add(1);
|
|
continue;
|
|
}
|
|
|
|
let peer_identity = match PeerIdentity::from_npub(&npub) {
|
|
Ok(identity) => identity,
|
|
Err(_) => {
|
|
skipped_invalid_npub = skipped_invalid_npub.saturating_add(1);
|
|
continue;
|
|
}
|
|
};
|
|
let node_addr = *peer_identity.node_addr();
|
|
if node_addr == *self.identity().node_addr() {
|
|
skipped_self = skipped_self.saturating_add(1);
|
|
continue;
|
|
}
|
|
if self.peers.contains_key(&node_addr) {
|
|
skipped_connected = skipped_connected.saturating_add(1);
|
|
continue;
|
|
}
|
|
if self.retry_pending.contains_key(&node_addr) {
|
|
skipped_retry_pending = skipped_retry_pending.saturating_add(1);
|
|
continue;
|
|
}
|
|
if bootstrap.cooldown_until(&npub, now_ms).is_some() {
|
|
skipped_cooldown = skipped_cooldown.saturating_add(1);
|
|
continue;
|
|
}
|
|
let connecting = self.connections.values().any(|conn| {
|
|
conn.expected_identity()
|
|
.map(|id| id.node_addr() == &node_addr)
|
|
.unwrap_or(false)
|
|
});
|
|
if connecting {
|
|
skipped_connecting = skipped_connecting.saturating_add(1);
|
|
continue;
|
|
}
|
|
|
|
let mut addresses = Vec::new();
|
|
let mut priority = 120u8;
|
|
let seen_at_ms = Self::now_ms();
|
|
for endpoint in endpoints {
|
|
let Some(candidate) =
|
|
crate::nostr::RendezvousDriver::overlay_endpoint_to_peer_address(
|
|
&endpoint, priority, seen_at_ms,
|
|
)
|
|
else {
|
|
continue;
|
|
};
|
|
if addresses.iter().any(|existing: &PeerAddress| {
|
|
existing.transport == candidate.transport && existing.addr == candidate.addr
|
|
}) {
|
|
continue;
|
|
}
|
|
addresses.push(candidate);
|
|
priority = priority.saturating_add(1);
|
|
}
|
|
if addresses.is_empty() {
|
|
skipped_no_endpoints = skipped_no_endpoints.saturating_add(1);
|
|
continue;
|
|
}
|
|
|
|
self.peer_aliases
|
|
.entry(node_addr)
|
|
.or_insert_with(|| peer_identity.short_npub());
|
|
self.register_identity(node_addr, peer_identity.pubkey_full());
|
|
|
|
let mut state = super::retry::RetryState::new(PeerConfig {
|
|
npub: npub.clone(),
|
|
alias: None,
|
|
addresses,
|
|
connect_policy: ConnectPolicy::AutoConnect,
|
|
auto_reconnect: true,
|
|
via_nostr: false,
|
|
});
|
|
state.reconnect = false;
|
|
state.retry_after_ms = now_ms;
|
|
state.expires_at_ms = Some(self.open_discovery_retry_expires_at_ms(now_ms));
|
|
self.retry_pending.insert(node_addr, state);
|
|
info!(
|
|
caller = %caller,
|
|
peer = %peer_identity.short_npub(),
|
|
advert_age_secs = now_secs.saturating_sub(created_at_secs),
|
|
"open-discovery sweep: queued retry for cached advert"
|
|
);
|
|
enqueue_budget = enqueue_budget.saturating_sub(1);
|
|
enqueued = enqueued.saturating_add(1);
|
|
}
|
|
|
|
// Always log a one-line summary on the startup sweep so operators
|
|
// can verify it ran. Per-tick sweeps are noisier; only summarize
|
|
// when something happened.
|
|
let total_skipped = skipped_age
|
|
+ skipped_configured
|
|
+ skipped_self
|
|
+ skipped_connected
|
|
+ skipped_retry_pending
|
|
+ skipped_connecting
|
|
+ skipped_no_endpoints
|
|
+ skipped_invalid_npub
|
|
+ skipped_cooldown;
|
|
let should_summarize = caller == "startup" || enqueued > 0;
|
|
if should_summarize {
|
|
info!(
|
|
caller = %caller,
|
|
cached = cached_count,
|
|
queued = enqueued,
|
|
skipped_age = skipped_age,
|
|
skipped_configured = skipped_configured,
|
|
skipped_self = skipped_self,
|
|
skipped_connected = skipped_connected,
|
|
skipped_retry_pending = skipped_retry_pending,
|
|
skipped_connecting = skipped_connecting,
|
|
skipped_no_endpoints = skipped_no_endpoints,
|
|
skipped_invalid_npub = skipped_invalid_npub,
|
|
skipped_cooldown = skipped_cooldown,
|
|
skipped_total = total_skipped,
|
|
"open-discovery sweep complete"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// One-shot startup sweep: runs once after the configured settle
|
|
/// delay, iterating the cached overlay adverts and queueing retries
|
|
/// for any peer with a recent enough advert that we haven't already
|
|
/// configured statically or established a link to.
|
|
///
|
|
/// Gated identically to [`run_open_discovery_sweep`]: requires
|
|
/// `node.rendezvous.nostr.enabled` and `policy == open`.
|
|
async fn maybe_run_startup_open_discovery_sweep(
|
|
&mut self,
|
|
bootstrap: &std::sync::Arc<NostrRendezvous>,
|
|
) {
|
|
if self.supervisor.nostr_rendezvous.startup_sweep_done() {
|
|
return;
|
|
}
|
|
if !self.config().node.rendezvous.nostr.enabled
|
|
|| self.config().node.rendezvous.nostr.policy
|
|
!= crate::config::NostrRendezvousPolicy::Open
|
|
{
|
|
// Mark done so we don't keep re-checking on every tick.
|
|
self.supervisor.nostr_rendezvous.set_startup_sweep_done();
|
|
return;
|
|
}
|
|
let Some(started_at_ms) = self.supervisor.nostr_rendezvous.started_at_ms() else {
|
|
return;
|
|
};
|
|
let now_ms = Self::now_ms();
|
|
let delay_ms = self
|
|
.config()
|
|
.node
|
|
.rendezvous
|
|
.nostr
|
|
.startup_sweep_delay_secs
|
|
.saturating_mul(1000);
|
|
if now_ms < started_at_ms.saturating_add(delay_ms) {
|
|
return;
|
|
}
|
|
|
|
let max_age_secs = self
|
|
.config()
|
|
.node
|
|
.rendezvous
|
|
.nostr
|
|
.startup_sweep_max_age_secs;
|
|
self.run_open_discovery_sweep(bootstrap, Some(max_age_secs), "startup")
|
|
.await;
|
|
self.supervisor.nostr_rendezvous.set_startup_sweep_done();
|
|
}
|
|
|
|
fn available_outbound_slots(&self) -> usize {
|
|
let connection_used = self
|
|
.connections
|
|
.len()
|
|
.saturating_add(self.pending_connects.len());
|
|
let connection_slots = if self.max_connections() == 0 {
|
|
usize::MAX
|
|
} else {
|
|
self.max_connections().saturating_sub(connection_used)
|
|
};
|
|
|
|
let peer_slots = if self.max_peers() == 0 {
|
|
usize::MAX
|
|
} else {
|
|
self.max_peers().saturating_sub(self.peers.len())
|
|
};
|
|
|
|
connection_slots.min(peer_slots)
|
|
}
|
|
|
|
fn outbound_handshake_slots(&self) -> usize {
|
|
let used = self
|
|
.connections
|
|
.len()
|
|
.saturating_add(self.pending_connects.len());
|
|
if self.max_connections() == 0 {
|
|
usize::MAX
|
|
} else {
|
|
self.max_connections().saturating_sub(used)
|
|
}
|
|
}
|
|
|
|
fn outbound_link_slots(&self) -> usize {
|
|
if self.max_links() == 0 {
|
|
usize::MAX
|
|
} else {
|
|
self.max_links().saturating_sub(self.links.len())
|
|
}
|
|
}
|
|
|
|
fn path_candidate_attempt_budget(&self, peer_node_addr: &NodeAddr) -> usize {
|
|
if !self.peers.contains_key(peer_node_addr)
|
|
&& self.max_peers() > 0
|
|
&& self.peers.len() >= self.max_peers()
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
let in_flight_for_peer = self
|
|
.connections
|
|
.values()
|
|
.filter(|conn| {
|
|
conn.expected_identity()
|
|
.map(|identity| identity.node_addr() == peer_node_addr)
|
|
.unwrap_or(false)
|
|
})
|
|
.count()
|
|
.saturating_add(
|
|
self.pending_connects
|
|
.iter()
|
|
.filter(|pending| pending.peer_identity.node_addr() == peer_node_addr)
|
|
.count(),
|
|
);
|
|
|
|
self.outbound_handshake_slots()
|
|
.min(self.outbound_link_slots())
|
|
.min(MAX_PARALLEL_PATH_CANDIDATES_PER_PEER.saturating_sub(in_flight_for_peer))
|
|
}
|
|
|
|
fn discovery_connect_budget(&self) -> usize {
|
|
self.outbound_handshake_slots()
|
|
.min(self.outbound_link_slots())
|
|
.min(MAX_DISCOVERY_CONNECTS_PER_TICK)
|
|
}
|
|
|
|
fn open_discovery_enqueue_budget(&self, configured_npubs: &HashSet<String>) -> usize {
|
|
let current_open_discovery_pending = self
|
|
.retry_pending
|
|
.values()
|
|
.filter(|state| !configured_npubs.contains(&state.peer_config.npub))
|
|
.count();
|
|
|
|
let cap_remaining = self
|
|
.config()
|
|
.node
|
|
.rendezvous
|
|
.nostr
|
|
.open_discovery_max_pending
|
|
.saturating_sub(current_open_discovery_pending);
|
|
|
|
cap_remaining.min(self.available_outbound_slots())
|
|
}
|
|
|
|
fn open_discovery_retry_expires_at_ms(&self, now_ms: u64) -> u64 {
|
|
now_ms.saturating_add(
|
|
self.config()
|
|
.node
|
|
.rendezvous
|
|
.nostr
|
|
.advert_ttl_secs
|
|
.saturating_mul(1000)
|
|
.saturating_mul(OPEN_DISCOVERY_RETRY_LIFETIME_MULTIPLIER),
|
|
)
|
|
}
|
|
|
|
/// Capture the advertisable-endpoint inputs of every operational
|
|
/// transport into a snapshot the rendezvous driver can turn into an
|
|
/// `OverlayAdvert` without borrowing the transport table across the
|
|
/// STUN await. Iteration order matches `self.transports.values()`, and
|
|
/// only transports whose type matched a configured listener are
|
|
/// included, mirroring the original per-transport branch structure.
|
|
fn advert_transport_snapshot(&self) -> Vec<crate::nostr::AdvertTransportSnapshot> {
|
|
use crate::nostr::AdvertTransportSnapshot;
|
|
let mut snapshot = Vec::new();
|
|
for handle in self.transports.values() {
|
|
if !handle.is_operational() {
|
|
continue;
|
|
}
|
|
match handle.transport_type().name {
|
|
"udp" => {
|
|
let Some(cfg) = self.lookup_udp_config(handle.name()) else {
|
|
continue;
|
|
};
|
|
snapshot.push(AdvertTransportSnapshot::Udp {
|
|
advertise: cfg.advertise_on_nostr(),
|
|
is_public: cfg.is_public(),
|
|
external_addr: cfg.external_advert_addr(),
|
|
local_addr: handle.local_addr(),
|
|
transport_key: handle.transport_id().as_u32(),
|
|
});
|
|
}
|
|
"tcp" => {
|
|
let Some(cfg) = self.lookup_tcp_config(handle.name()) else {
|
|
continue;
|
|
};
|
|
snapshot.push(AdvertTransportSnapshot::Tcp {
|
|
advertise: cfg.advertise_on_nostr(),
|
|
external_addr: cfg.external_advert_addr(),
|
|
local_addr: handle.local_addr(),
|
|
});
|
|
}
|
|
"tor" => {
|
|
let Some(cfg) = self.lookup_tor_config(handle.name()) else {
|
|
continue;
|
|
};
|
|
snapshot.push(AdvertTransportSnapshot::Tor {
|
|
advertise: cfg.advertise_on_nostr(),
|
|
onion_addr: handle.onion_address().map(|s| s.to_string()),
|
|
advertised_port: cfg.advertised_port(),
|
|
});
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
snapshot
|
|
}
|
|
|
|
async fn refresh_overlay_advert(
|
|
&self,
|
|
bootstrap: &std::sync::Arc<NostrRendezvous>,
|
|
) -> Result<(), crate::nostr::BootstrapError> {
|
|
let snapshot = self.advert_transport_snapshot();
|
|
self.supervisor
|
|
.nostr_rendezvous
|
|
.refresh_overlay_advert(bootstrap, snapshot, &self.config().node.rendezvous.nostr)
|
|
.await
|
|
}
|
|
|
|
fn lookup_udp_config(&self, transport_name: Option<&str>) -> Option<&crate::config::UdpConfig> {
|
|
match (&self.config().transports.udp, transport_name) {
|
|
(crate::config::TransportInstances::Single(cfg), None) => Some(cfg),
|
|
(crate::config::TransportInstances::Named(configs), Some(name)) => configs.get(name),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
fn lookup_tcp_config(&self, transport_name: Option<&str>) -> Option<&crate::config::TcpConfig> {
|
|
match (&self.config().transports.tcp, transport_name) {
|
|
(crate::config::TransportInstances::Single(cfg), None) => Some(cfg),
|
|
(crate::config::TransportInstances::Named(configs), Some(name)) => configs.get(name),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
fn lookup_tor_config(&self, transport_name: Option<&str>) -> Option<&crate::config::TorConfig> {
|
|
match (&self.config().transports.tor, transport_name) {
|
|
(crate::config::TransportInstances::Single(cfg), None) => Some(cfg),
|
|
(crate::config::TransportInstances::Named(configs), Some(name)) => configs.get(name),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
pub(in crate::node) async fn try_peer_addresses(
|
|
&mut self,
|
|
peer_config: &PeerConfig,
|
|
peer_identity: PeerIdentity,
|
|
allow_bootstrap_nat: bool,
|
|
) -> Result<(), NodeError> {
|
|
let peer_node_addr = *peer_identity.node_addr();
|
|
if self.peers.contains_key(&peer_node_addr) {
|
|
debug!(
|
|
npub = %peer_config.npub,
|
|
"Peer already exists, skipping address attempts"
|
|
);
|
|
return Ok(());
|
|
}
|
|
if self.is_connecting_to_peer(&peer_node_addr) {
|
|
debug!(
|
|
npub = %peer_config.npub,
|
|
"Connection already in progress, skipping address attempts"
|
|
);
|
|
return Ok(());
|
|
}
|
|
|
|
let candidates = self.peer_address_candidates(peer_config).await;
|
|
|
|
if candidates.is_empty() {
|
|
return Err(NodeError::NoTransportForType(format!(
|
|
"no addresses known for {}",
|
|
peer_config.npub
|
|
)));
|
|
}
|
|
|
|
if self
|
|
.attempt_peer_address_list(peer_config, peer_identity, allow_bootstrap_nat, &candidates)
|
|
.await
|
|
.is_ok()
|
|
{
|
|
return Ok(());
|
|
}
|
|
|
|
Err(NodeError::NoTransportForType(format!(
|
|
"no operational transport for any of {}'s addresses",
|
|
peer_config.npub
|
|
)))
|
|
}
|
|
|
|
async fn try_active_peer_alternative_addresses(
|
|
&mut self,
|
|
peer_config: &PeerConfig,
|
|
peer_identity: PeerIdentity,
|
|
) -> Result<bool, NodeError> {
|
|
let peer_node_addr = *peer_identity.node_addr();
|
|
let candidates = self.peer_address_candidates(peer_config).await;
|
|
|
|
if candidates.is_empty() {
|
|
return Err(NodeError::NoTransportForType(format!(
|
|
"no addresses known for {}",
|
|
peer_config.npub
|
|
)));
|
|
}
|
|
|
|
let concrete: Vec<_> = candidates
|
|
.into_iter()
|
|
.filter(|addr| !(addr.transport == "udp" && addr.addr.eq_ignore_ascii_case("nat")))
|
|
.collect();
|
|
let has_alternative = concrete
|
|
.iter()
|
|
.any(|addr| !self.active_peer_matches_candidate(&peer_node_addr, addr));
|
|
let attempt_candidates: Vec<_> = if has_alternative {
|
|
concrete
|
|
.into_iter()
|
|
.filter(|addr| !self.active_peer_matches_candidate(&peer_node_addr, addr))
|
|
.collect()
|
|
} else if self.active_peer_needs_same_path_refresh(&peer_node_addr) {
|
|
concrete
|
|
} else {
|
|
Vec::new()
|
|
};
|
|
|
|
if attempt_candidates.is_empty() {
|
|
return Ok(false);
|
|
}
|
|
|
|
self.attempt_peer_address_list(peer_config, peer_identity, false, &attempt_candidates)
|
|
.await?;
|
|
Ok(true)
|
|
}
|
|
|
|
async fn peer_address_candidates(&self, peer_config: &PeerConfig) -> Vec<PeerAddress> {
|
|
let static_addresses = self.static_peer_addresses(peer_config);
|
|
let overlay_addresses = self
|
|
.supervisor
|
|
.nostr_rendezvous
|
|
.nostr_peer_fallback_addresses(
|
|
peer_config,
|
|
&static_addresses,
|
|
&self.config().node.rendezvous.nostr,
|
|
Self::now_ms(),
|
|
)
|
|
.await;
|
|
|
|
let mut candidates = Vec::with_capacity(overlay_addresses.len() + static_addresses.len());
|
|
for addr in overlay_addresses.into_iter().chain(static_addresses) {
|
|
if !candidates.iter().any(|existing: &PeerAddress| {
|
|
existing.transport == addr.transport && existing.addr == addr.addr
|
|
}) {
|
|
candidates.push(addr);
|
|
}
|
|
}
|
|
|
|
candidates.sort_by(|a, b| match (a.seen_at_ms, b.seen_at_ms) {
|
|
(Some(a_ts), Some(b_ts)) => b_ts.cmp(&a_ts),
|
|
(Some(_), None) => std::cmp::Ordering::Less,
|
|
(None, Some(_)) => std::cmp::Ordering::Greater,
|
|
(None, None) => std::cmp::Ordering::Equal,
|
|
});
|
|
candidates
|
|
}
|
|
|
|
pub(in crate::node) fn active_peer_candidate_is_fresh_enough_to_skip(
|
|
&self,
|
|
peer_node_addr: &NodeAddr,
|
|
candidates: &[PeerAddress],
|
|
) -> bool {
|
|
if !self.active_peer_matches_any_candidate(peer_node_addr, candidates) {
|
|
return false;
|
|
}
|
|
!self.active_peer_needs_same_path_refresh(peer_node_addr)
|
|
}
|
|
|
|
fn active_peer_needs_same_path_refresh(&self, peer_node_addr: &NodeAddr) -> bool {
|
|
let Some(peer) = self.peers.get(peer_node_addr) else {
|
|
return false;
|
|
};
|
|
let stale_after_ms = self
|
|
.config()
|
|
.node
|
|
.heartbeat_interval_secs
|
|
.saturating_mul(1000)
|
|
.max(1000);
|
|
peer.idle_time(Self::now_ms()) > stale_after_ms
|
|
}
|
|
|
|
fn active_peer_matches_any_candidate(
|
|
&self,
|
|
peer_node_addr: &NodeAddr,
|
|
candidates: &[PeerAddress],
|
|
) -> bool {
|
|
candidates
|
|
.iter()
|
|
.any(|candidate| self.active_peer_matches_candidate(peer_node_addr, candidate))
|
|
}
|
|
|
|
fn active_peer_matches_candidate(
|
|
&self,
|
|
peer_node_addr: &NodeAddr,
|
|
candidate: &PeerAddress,
|
|
) -> bool {
|
|
let Some(peer) = self.peers.get(peer_node_addr) else {
|
|
return false;
|
|
};
|
|
let Some(current_addr) = peer.current_addr() else {
|
|
return false;
|
|
};
|
|
if peer
|
|
.transport_id()
|
|
.map(|id| self.supervisor.nostr_rendezvous.is_bootstrap_transport(&id))
|
|
.unwrap_or(false)
|
|
{
|
|
return false;
|
|
}
|
|
let current_addr = current_addr.to_string();
|
|
let current_transport = peer
|
|
.transport_id()
|
|
.and_then(|id| self.transports.get(&id))
|
|
.map(|transport| transport.transport_type().name);
|
|
|
|
candidate.addr == current_addr
|
|
&& current_transport
|
|
.map(|transport| transport == candidate.transport)
|
|
.unwrap_or(true)
|
|
}
|
|
|
|
// === Control API methods ===
|
|
|
|
/// Connect to a peer via the control API.
|
|
///
|
|
/// Creates an ephemeral peer connection (not persisted to config, no
|
|
/// auto-reconnect). Reuses the same connection path as auto-connect
|
|
/// peers. Returns JSON data on success or an error message.
|
|
pub(crate) async fn api_connect(
|
|
&mut self,
|
|
npub: &str,
|
|
address: &str,
|
|
transport: &str,
|
|
) -> Result<serde_json::Value, String> {
|
|
let peer_config = PeerConfig {
|
|
npub: npub.to_string(),
|
|
alias: None,
|
|
addresses: vec![PeerAddress::new(transport, address)],
|
|
connect_policy: ConnectPolicy::Manual,
|
|
auto_reconnect: false,
|
|
via_nostr: false,
|
|
};
|
|
|
|
// Pre-seed identity cache (same as initiate_peer_connections does)
|
|
if let Ok(identity) = PeerIdentity::from_npub(npub) {
|
|
self.peer_aliases
|
|
.insert(*identity.node_addr(), identity.short_npub());
|
|
self.register_identity(*identity.node_addr(), identity.pubkey_full());
|
|
}
|
|
|
|
self.initiate_peer_connection(&peer_config)
|
|
.await
|
|
.map(|()| {
|
|
info!(
|
|
npub = %npub,
|
|
address = %address,
|
|
transport = %transport,
|
|
"API connect initiated"
|
|
);
|
|
serde_json::json!({
|
|
"npub": npub,
|
|
"address": address,
|
|
"transport": transport,
|
|
})
|
|
})
|
|
.map_err(|e| e.to_string())
|
|
}
|
|
|
|
/// Disconnect a peer via the control API.
|
|
///
|
|
/// Notifies the peer, removes it locally, and suppresses auto-reconnect.
|
|
pub(crate) async fn api_disconnect(&mut self, npub: &str) -> Result<serde_json::Value, String> {
|
|
let peer_identity =
|
|
PeerIdentity::from_npub(npub).map_err(|e| format!("invalid npub '{npub}': {e}"))?;
|
|
let node_addr = *peer_identity.node_addr();
|
|
|
|
if !self.peers.contains_key(&node_addr) {
|
|
return Err(format!("peer not found: {npub}"));
|
|
}
|
|
|
|
// Notify the peer before we tear down the link, so it drops its own
|
|
// session and re-handshakes symmetrically rather than holding a stale
|
|
// session that never re-emits its tree/filter announcements. The link
|
|
// must still exist for the send, so this runs before removal.
|
|
// Best-effort: a send failure must not block the local teardown.
|
|
self.send_disconnect_to_peer(&node_addr, DisconnectReason::ConfigurationChange)
|
|
.await;
|
|
|
|
// Remove the peer (full cleanup: sessions, indices, links, tree, bloom)
|
|
self.remove_active_peer(&node_addr);
|
|
|
|
// Suppress any pending auto-reconnect
|
|
self.retry_pending.remove(&node_addr);
|
|
|
|
info!(npub = %npub, "API disconnect completed");
|
|
|
|
Ok(serde_json::json!({
|
|
"npub": npub,
|
|
"disconnected": true,
|
|
}))
|
|
}
|
|
|
|
/// Adopt an already-established UDP traversal and start the normal FIPS
|
|
/// Noise handshake over it.
|
|
///
|
|
/// This is intended for integration with an external rendezvous runtime
|
|
/// that has already completed relay signaling, STUN observation, and UDP
|
|
/// hole punching. After handoff, the adopted socket is owned by FIPS.
|
|
pub async fn adopt_established_traversal(
|
|
&mut self,
|
|
traversal: EstablishedTraversal,
|
|
) -> Result<BootstrapHandoffResult, NodeError> {
|
|
debug!(
|
|
peer_npub = %traversal.peer_npub,
|
|
session_id = %traversal.session_id,
|
|
remote_addr = %traversal.remote_addr,
|
|
"adopting established traversal socket"
|
|
);
|
|
|
|
if !self.supervisor.state.is_operational() {
|
|
return Err(NodeError::NotStarted);
|
|
}
|
|
|
|
let packet_tx = self
|
|
.supervisor
|
|
.packet_tx
|
|
.clone()
|
|
.ok_or(NodeError::NotStarted)?;
|
|
let peer_identity = PeerIdentity::from_npub(&traversal.peer_npub).map_err(|e| {
|
|
NodeError::InvalidPeerNpub {
|
|
npub: traversal.peer_npub.clone(),
|
|
reason: e.to_string(),
|
|
}
|
|
})?;
|
|
let peer_node_addr = *peer_identity.node_addr();
|
|
if self.peers.contains_key(&peer_node_addr) {
|
|
debug!(
|
|
peer_npub = %traversal.peer_npub,
|
|
"Ignoring NAT traversal handoff for already-connected peer"
|
|
);
|
|
return Err(NodeError::PeerAlreadyExists(peer_node_addr));
|
|
}
|
|
if self.is_connecting_to_peer(&peer_node_addr) {
|
|
debug!(
|
|
peer_npub = %traversal.peer_npub,
|
|
"Ignoring NAT traversal handoff while peer handshake is already in progress"
|
|
);
|
|
return Err(NodeError::PeerAlreadyExists(peer_node_addr));
|
|
}
|
|
|
|
self.peer_aliases
|
|
.insert(peer_node_addr, peer_identity.short_npub());
|
|
self.register_identity(peer_node_addr, peer_identity.pubkey_full());
|
|
|
|
let transport_id = self.allocate_transport_id();
|
|
// Adopted ephemeral UDP transports inherit MTU + socket-buffer sizing
|
|
// (and accept_connections / advertise flags) from the operator's
|
|
// configured [transports.udp] when the bootstrap runtime doesn't
|
|
// pass an explicit override. Lookup tries `transport_name` first
|
|
// (covers the `Named` multi-listener variant) and falls back to the
|
|
// unnamed `Single` listener, so single- and named-listener configs
|
|
// both inherit cleanly.
|
|
//
|
|
// Tradeoff: `UdpConfig::default()` sets MTU 1280 (IPv6 minimum), the
|
|
// only value guaranteed to survive arbitrary middlebox paths.
|
|
// Inheriting a higher operator-chosen MTU means NAT-traversed flows
|
|
// initially attempt that MTU and may black-hole on tighter paths
|
|
// until reactive `MtuExceeded` recovery kicks in. Operators who
|
|
// raise the primary MTU based on known-clean topology accept that
|
|
// tradeoff; the silent drop on a too-low default was strictly
|
|
// worse for the common case where the primary MTU is reachable.
|
|
//
|
|
// Bind / external address fields are cleared since the socket is
|
|
// already bound.
|
|
let inherited_config = traversal.transport_config.clone().unwrap_or_else(|| {
|
|
let mut cfg = self
|
|
.lookup_udp_config(traversal.transport_name.as_deref())
|
|
.or_else(|| self.lookup_udp_config(None))
|
|
.cloned()
|
|
.unwrap_or_default();
|
|
cfg.bind_addr = None;
|
|
cfg.external_addr = None;
|
|
cfg
|
|
});
|
|
let mut transport = crate::transport::udp::UdpTransport::new(
|
|
transport_id,
|
|
traversal.transport_name.clone(),
|
|
inherited_config,
|
|
packet_tx,
|
|
);
|
|
|
|
transport
|
|
.adopt_socket_async(traversal.socket)
|
|
.await
|
|
.map_err(|e| NodeError::BootstrapHandoff(e.to_string()))?;
|
|
|
|
let local_addr = transport.local_addr().ok_or_else(|| {
|
|
NodeError::BootstrapHandoff("adopted UDP transport has no local address".into())
|
|
})?;
|
|
|
|
self.transports.insert(
|
|
transport_id,
|
|
crate::transport::TransportHandle::Udp(transport),
|
|
);
|
|
self.supervisor
|
|
.nostr_rendezvous
|
|
.insert_bootstrap_transport(transport_id, traversal.peer_npub.clone());
|
|
|
|
let remote_addr = TransportAddr::from_string(&traversal.remote_addr.to_string());
|
|
if let Err(err) = self
|
|
.initiate_connection(transport_id, remote_addr.clone(), peer_identity)
|
|
.await
|
|
{
|
|
self.supervisor
|
|
.nostr_rendezvous
|
|
.remove_bootstrap_transport(&transport_id);
|
|
if let Some(mut handle) = self.transports.remove(&transport_id) {
|
|
let _ = handle.stop().await;
|
|
}
|
|
return Err(err);
|
|
}
|
|
|
|
info!(
|
|
peer = %self.peer_display_name(&peer_node_addr),
|
|
transport_id = %transport_id,
|
|
local_addr = %local_addr,
|
|
remote_addr = %traversal.remote_addr,
|
|
session_id = %traversal.session_id,
|
|
"adopted NAT traversal socket; handshake initiated"
|
|
);
|
|
|
|
Ok(BootstrapHandoffResult {
|
|
transport_id,
|
|
local_addr,
|
|
remote_addr: traversal.remote_addr,
|
|
peer_node_addr,
|
|
session_id: traversal.session_id,
|
|
})
|
|
}
|
|
}
|