laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-09 08:14:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

node: refresh active peer paths without dropping links

Browse Source 
Add Node::update_peers for runtime peer-list refresh. It re-derives the
active peer connections from a new peer configuration, adding newly
configured peers and removing those no longer present, while keeping
links to peers that remain in the set rather than tearing every
connection down. The call returns an UpdatePeersOutcome summarizing the
added, removed, and retained peers.

PeerAddress gains a seen_at_ms recency field (with_seen_at_ms). Active
path selection now sorts address candidates by recency so the most
recently observed address wins when concurrent path probes race.

complete_rekey_msg2 now returns the remote peer's startup epoch
alongside the new Noise session, letting the rekey path detect a peer
restart and clear stale session state. A stale FSP session is cleared
when a peer restart is detected during FMP rekey or cross-connection
promotion, so the session-layer map no longer lingers out of sync with
the freshly promoted peer.

Per-tick work budgets bound the connection churn in a single node tick
(MAX_DISCOVERY_CONNECTS_PER_TICK, MAX_RETRY_CONNECTIONS_PER_TICK,
MAX_PARALLEL_PATH_CANDIDATES_PER_PEER); work beyond a tick's budget is
deferred to the next tick rather than discarded.

Co-authored-by: Johnathan Corgan <johnathan@corganlabs.com>
This commit is contained in:
Martti Malmi
2026-05-30 00:46:38 +00:00
committed by Johnathan Corgan
co-authored by Johnathan Corgan
parent 8d94c0f29c
commit da0d9d39a0
10 changed files with 872 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/lifecycle.rs
+506 -56
View File
@@ -14,14 +14,177 @@ use crate::protocol::{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::HashSet;
use std::collections::{HashMap, HashSet};
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;
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);
}
let current_by_addr: HashMap<NodeAddr, PeerConfig> = self
.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(&current_addrs).copied().collect();
let kept: Vec<_> = new_addrs.intersection(&current_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 = &current_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();
self.config.peers = new_by_addr.into_values().collect();
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(()) => {
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(
self.config.node.rate_limit.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
@@ -142,6 +305,25 @@ impl Node {
})
}
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
})
}
/// Initiate a connection to a peer on a specific transport and address.
///
/// For connectionless transports (UDP, Ethernet): allocates a link, starts
@@ -340,6 +522,9 @@ impl Node {
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() {
@@ -366,29 +551,80 @@ impl Node {
if node_addr == *self.identity.node_addr() {
continue;
}
// Skip if already connected
let candidate_transport_id = *transport_id;
let remote_addr = peer.addr;
if self.peers.contains_key(&node_addr) {
continue;
}
// Skip if connection already in progress
let connecting = self.connections.values().any(|c| {
c.expected_identity()
.map(|id| id.node_addr() == &node_addr)
.unwrap_or(false)
});
if connecting {
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;
}
to_connect.push((*transport_id, peer.addr, identity));
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);
}
}
for (transport_id, remote_addr, identity) in to_connect {
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
@@ -1228,8 +1464,10 @@ impl Node {
.max()
.unwrap_or(100)
.saturating_add(1);
let seen_at_ms = Self::now_ms();
for endpoint in endpoints {
let Some(candidate) = Self::overlay_endpoint_to_peer_address(&endpoint, next_priority)
let Some(candidate) =
Self::overlay_endpoint_to_peer_address(&endpoint, next_priority, seen_at_ms)
else {
continue;
};
@@ -1251,17 +1489,27 @@ impl Node {
fn overlay_endpoint_to_peer_address(
endpoint: &OverlayEndpointAdvert,
priority: u8,
seen_at_ms: u64,
) -> Option<PeerAddress> {
let transport = match endpoint.transport {
OverlayTransportKind::Udp => "udp",
OverlayTransportKind::Tcp => "tcp",
OverlayTransportKind::Tor => "tor",
};
Some(PeerAddress::with_priority(
transport,
endpoint.addr.clone(),
priority,
))
Some(
PeerAddress::with_priority(transport, endpoint.addr.clone(), priority)
.with_seen_at_ms(seen_at_ms),
)
}
async fn request_nostr_bootstrap(&self, peer_config: &PeerConfig) -> bool {
let Some(bootstrap) = self.nostr_discovery.clone() else {
debug!(npub = %peer_config.npub, "No Nostr overlay runtime for udp:nat address");
return false;
};
bootstrap.request_connect(peer_config.clone()).await;
info!(npub = %peer_config.npub, "Started Nostr UDP NAT traversal attempt");
true
}
async fn attempt_peer_address_list(
@@ -1271,18 +1519,28 @@ impl Node {
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;
}
let Some(bootstrap) = self.nostr_discovery.clone() else {
debug!(npub = %peer_config.npub, "No Nostr overlay runtime for udp:nat address");
continue;
};
bootstrap.request_connect(peer_config.clone()).await;
info!(npub = %peer_config.npub, "Started Nostr UDP NAT traversal attempt");
return Ok(());
if self.request_nostr_bootstrap(peer_config).await {
attempted = attempted.saturating_add(1);
}
continue;
}
let (transport_id, remote_addr) = if addr.transport == "ethernet" {
@@ -1334,11 +1592,15 @@ impl Node {
(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(()) => return Ok(()),
Ok(()) => attempted = attempted.saturating_add(1),
Err(e @ NodeError::AccessDenied(_)) => return Err(e),
Err(e) => {
debug!(
@@ -1351,6 +1613,10 @@ impl Node {
}
}
if attempted > 0 {
return Ok(());
}
Err(NodeError::NoTransportForType(format!(
"no operational transport for any of {}'s addresses",
peer_config.npub
@@ -1427,6 +1693,21 @@ impl Node {
}
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;
}
@@ -1467,8 +1748,10 @@ impl Node {
let mut addresses = Vec::new();
let mut priority = 120u8;
let seen_at_ms = Self::now_ms();
for endpoint in endpoints {
let Some(candidate) = Self::overlay_endpoint_to_peer_address(&endpoint, priority)
let Some(candidate) =
Self::overlay_endpoint_to_peer_address(&endpoint, priority, seen_at_ms)
else {
continue;
};
@@ -1607,6 +1890,61 @@ impl Node {
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
@@ -1841,47 +2179,159 @@ impl Node {
return Ok(());
}
// Static-first dialing: avoid delaying configured address attempts on
// advert fetch/network latency.
let static_addresses = self.static_peer_addresses(peer_config);
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,
&static_addresses,
)
.attempt_peer_address_list(peer_config, peer_identity, allow_bootstrap_nat, &candidates)
.await
.is_ok()
{
return Ok(());
}
{
let fallback = self
.nostr_peer_fallback_addresses(peer_config, &static_addresses)
.await;
if !fallback.is_empty()
&& self
.attempt_peer_address_list(
peer_config,
peer_identity,
allow_bootstrap_nat,
&fallback,
)
.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
.nostr_peer_fallback_addresses(peer_config, &static_addresses)
.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.bootstrap_transports.contains(&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.