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

Consolidate the nostr rendezvous driver into src/nostr

Browse Source 
Pull the rendezvous driver state and the movable driver logic out of the
Node struct into the src/nostr home. A new RendezvousDriver owns the
engine handle and the four bookkeeping fields (traversal start time,
startup-sweep latch, adopted bootstrap-transport set and their npubs)
that previously sat loose on Node; Node holds a single driver field and
reaches the same data through thin accessors, including the rx-loop
hot-path protocol-mismatch hook.

The advert build/refresh, the via_nostr fallback-address resolve, the
overlay-endpoint-to-PeerAddress mapping, and the bootstrap request move
onto the driver, taking their Node inputs explicitly (a transport-
endpoint snapshot for advert building) rather than reading Node fields
directly. Transport/connection-table-bound work (traversal adoption,
bootstrap-transport cleanup, the open-discovery sweep, and the outbound
budget calculators) stays on Node as thin glue that calls into the
driver.

Behavior-neutral relocation: statements moved verbatim, no logic, wire,
config-key, or metric changes. lifecycle.rs shrinks ~230 lines. cargo
fmt/build/clippy clean; lib suite 1547 passing (baseline unchanged).
This commit is contained in:
Johnathan Corgan
2026-07-09 22:13:56 +00:00
parent 3f80530cc5
commit 1208f6a5c2
7 changed files with 486 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/handlers/rx_loop.rs
+5 -3
View File
@@ -319,10 +319,12 @@ impl Node {
// though no msg1/msg2 exchange can ever succeed. Bump the
// discovery-layer cooldown to the long protocol-mismatch
// window and emit a single WARN per fresh observation.
if self.bootstrap_transports.contains(&packet.transport_id)
if self
.nostr_rendezvous
.is_bootstrap_transport(&packet.transport_id)
&& let Some(npub) = self
.bootstrap_transport_npubs
.get(&packet.transport_id)
.nostr_rendezvous
.bootstrap_transport_npub(&packet.transport_id)
.cloned()
&& let Some(handle) = self.nostr_rendezvous_handle()
{
src/node/lifecycle.rs
+63 -235
View File
@@ -4,10 +4,7 @@ use super::{Node, NodeError, NodeState};
use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
use crate::node::acl::PeerAclContext;
use crate::node::wire::build_msg1;
use crate::nostr::{
ADVERT_IDENTIFIER, ADVERT_VERSION, BootstrapEvent, NostrRendezvous, OverlayAdvert,
OverlayEndpointAdvert, OverlayTransportKind,
};
use crate::nostr::{BootstrapEvent, NostrRendezvous};
use crate::nostr::{BootstrapHandoffResult, EstablishedTraversal};
use crate::peer::PeerConnection;
use crate::proto::fmp::{Disconnect, DisconnectReason};
@@ -266,7 +263,7 @@ impl Node {
// 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.nostr_rendezvous.clone()
&& let Some(bootstrap) = self.nostr_rendezvous.engine_arc()
{
let npub = peer_config.npub.clone();
tokio::spawn(async move {
@@ -357,7 +354,7 @@ impl Node {
.filter(|(id, handle)| {
handle.transport_type().name == "udp"
&& handle.is_operational()
&& !self.bootstrap_transports.contains(id)
&& !self.nostr_rendezvous.is_bootstrap_transport(id)
})
.filter_map(|(id, handle)| {
let local_addr = handle.local_addr()?;
@@ -680,7 +677,7 @@ impl Node {
}
pub(super) async fn poll_nostr_rendezvous(&mut self) {
let Some(bootstrap) = self.nostr_rendezvous.clone() else {
let Some(bootstrap) = self.nostr_rendezvous.engine_arc() else {
return;
};
@@ -1149,8 +1146,8 @@ impl Node {
if let Err(err) = self.refresh_overlay_advert(&runtime).await {
warn!(error = %err, "Failed to publish initial Nostr overlay advert");
}
self.nostr_rendezvous = Some(runtime);
self.nostr_rendezvous_started_at_ms = Some(Self::now_ms());
self.nostr_rendezvous.set_engine(runtime);
self.nostr_rendezvous.set_started_at_ms(Self::now_ms());
info!("Nostr overlay discovery enabled");
}
Err(err) => {
@@ -1175,7 +1172,7 @@ impl Node {
.filter(|(id, h)| {
h.transport_type().name == "udp"
&& h.is_operational()
&& !self.bootstrap_transports.contains(id)
&& !self.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())
@@ -1479,7 +1476,7 @@ impl Node {
.await;
// Stop Nostr overlay discovery background work and withdraw any advert.
if let Some(bootstrap) = self.nostr_rendezvous.take()
if let Some(bootstrap) = self.nostr_rendezvous.take_engine()
&& let Err(e) = bootstrap.shutdown().await
{
warn!(error = %e, "Failed to shutdown Nostr overlay discovery");
@@ -1621,89 +1618,6 @@ impl Node {
.collect()
}
async fn nostr_peer_fallback_addresses(
&self,
peer_config: &PeerConfig,
existing: &[PeerAddress],
) -> Vec<PeerAddress> {
if !self.config().node.rendezvous.nostr.enabled
|| !peer_config.via_nostr
|| self.config().node.rendezvous.nostr.policy
== crate::config::NostrRendezvousPolicy::Disabled
{
return Vec::new();
}
let Some(bootstrap) = self.nostr_rendezvous.clone() else {
return Vec::new();
};
let endpoints = match bootstrap.advert_endpoints_for_peer(&peer_config.npub).await {
Ok(endpoints) => endpoints,
Err(err) => {
debug!(
npub = %peer_config.npub,
error = %err,
"Failed to resolve Nostr advert endpoints for configured peer"
);
return Vec::new();
}
};
let mut fallback = Vec::new();
let mut next_priority = existing
.iter()
.map(|addr| addr.priority)
.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, seen_at_ms)
else {
continue;
};
if existing
.iter()
.any(|addr| addr.transport == candidate.transport && addr.addr == candidate.addr)
|| fallback.iter().any(|addr: &PeerAddress| {
addr.transport == candidate.transport && addr.addr == candidate.addr
})
{
continue;
}
fallback.push(candidate);
next_priority = next_priority.saturating_add(1);
}
fallback
}
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)
.with_seen_at_ms(seen_at_ms),
)
}
async fn request_nostr_bootstrap(&self, peer_config: &PeerConfig) -> bool {
let Some(bootstrap) = self.nostr_rendezvous.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(
&mut self,
peer_config: &PeerConfig,
@@ -1729,7 +1643,11 @@ impl Node {
if !allow_bootstrap_nat {
continue;
}
if self.request_nostr_bootstrap(peer_config).await {
if self
.nostr_rendezvous
.request_nostr_bootstrap(peer_config)
.await
{
attempted = attempted.saturating_add(1);
}
continue;
@@ -1960,7 +1878,9 @@ impl Node {
let seen_at_ms = Self::now_ms();
for endpoint in endpoints {
let Some(candidate) =
Self::overlay_endpoint_to_peer_address(&endpoint, priority, seen_at_ms)
crate::nostr::RendezvousDriver::overlay_endpoint_to_peer_address(
&endpoint, priority, seen_at_ms,
)
else {
continue;
};
@@ -2048,7 +1968,7 @@ impl Node {
&mut self,
bootstrap: &std::sync::Arc<NostrRendezvous>,
) {
if self.startup_open_discovery_sweep_done {
if self.nostr_rendezvous.startup_sweep_done() {
return;
}
if !self.config().node.rendezvous.nostr.enabled
@@ -2056,10 +1976,10 @@ impl Node {
!= crate::config::NostrRendezvousPolicy::Open
{
// Mark done so we don't keep re-checking on every tick.
self.startup_open_discovery_sweep_done = true;
self.nostr_rendezvous.set_startup_sweep_done();
return;
}
let Some(started_at_ms) = self.nostr_rendezvous_started_at_ms else {
let Some(started_at_ms) = self.nostr_rendezvous.started_at_ms() else {
return;
};
let now_ms = Self::now_ms();
@@ -2082,7 +2002,7 @@ impl Node {
.startup_sweep_max_age_secs;
self.run_open_discovery_sweep(bootstrap, Some(max_age_secs), "startup")
.await;
self.startup_open_discovery_sweep_done = true;
self.nostr_rendezvous.set_startup_sweep_done();
}
fn available_outbound_slots(&self) -> usize {
@@ -2190,163 +2110,66 @@ impl Node {
)
}
async fn build_overlay_advert(
&self,
bootstrap: &std::sync::Arc<NostrRendezvous>,
) -> Option<OverlayAdvert> {
if !self.config().node.rendezvous.nostr.enabled {
return None;
}
let mut endpoints = Vec::new();
let mut has_udp_nat = false;
/// 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;
};
if !cfg.advertise_on_nostr() {
continue;
}
if cfg.is_public() {
// Precedence:
// 1. operator-supplied `external_addr` (skips STUN)
// 2. non-wildcard `local_addr` (operator bound to
// a specific public IP directly)
// 3. STUN auto-discovery against ephemeral socket
// 4. loud warn + omit endpoint
if let Some(explicit) = cfg.external_advert_addr() {
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Udp,
addr: explicit.to_string(),
});
} else {
match handle.local_addr() {
Some(addr) if !addr.ip().is_unspecified() => {
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Udp,
addr: addr.to_string(),
});
}
Some(addr) => {
let key = handle.transport_id().as_u32();
let port = addr.port();
if let Some(public) =
bootstrap.learn_public_udp_addr(key, port).await
{
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Udp,
addr: public.to_string(),
});
} else {
warn!(
transport_id = key,
bind_addr = %addr,
"advert: udp public=true bound to wildcard but \
STUN observation failed; advertising no UDP \
endpoint. Either set transports.udp.external_addr, \
bind to a specific public IP, or ensure \
node.rendezvous.nostr.stun_servers is reachable"
);
}
}
None => {}
}
}
} else {
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Udp,
addr: "nat".to_string(),
});
has_udp_nat = true;
}
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;
};
if !cfg.advertise_on_nostr() {
continue;
}
// Precedence:
// 1. operator-supplied `external_addr` (only path that
// works on cloud-NAT setups where the public IP is
// not on a host interface).
// 2. non-wildcard `local_addr` (operator bound to a
// specific public IP directly).
// 3. loud warn + omit endpoint (no TCP STUN equivalent).
if let Some(explicit) = cfg.external_advert_addr() {
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Tcp,
addr: explicit.to_string(),
});
} else {
match handle.local_addr() {
Some(addr) if !addr.ip().is_unspecified() => {
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Tcp,
addr: addr.to_string(),
});
}
Some(addr) => {
warn!(
bind_addr = %addr,
"advert: tcp advertise_on_nostr=true bound to wildcard \
and no transports.tcp.external_addr set; advertising no \
TCP endpoint. Either set external_addr to the public \
IP (recommended for cloud 1:1-NAT setups) or bind \
explicitly to the public IP"
);
}
None => {}
}
}
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;
};
if !cfg.advertise_on_nostr() {
continue;
}
if let Some(addr) = handle.onion_address() {
endpoints.push(OverlayEndpointAdvert {
transport: OverlayTransportKind::Tor,
addr: format!("{}:{}", addr, cfg.advertised_port()),
});
}
snapshot.push(AdvertTransportSnapshot::Tor {
advertise: cfg.advertise_on_nostr(),
onion_addr: handle.onion_address().map(|s| s.to_string()),
advertised_port: cfg.advertised_port(),
});
}
_ => {}
}
}
if endpoints.is_empty() {
return None;
}
Some(OverlayAdvert {
identifier: ADVERT_IDENTIFIER.to_string(),
version: ADVERT_VERSION,
endpoints,
signal_relays: has_udp_nat
.then(|| self.config().node.rendezvous.nostr.dm_relays.clone()),
stun_servers: has_udp_nat
.then(|| self.config().node.rendezvous.nostr.stun_servers.clone()),
})
snapshot
}
async fn refresh_overlay_advert(
&self,
bootstrap: &std::sync::Arc<NostrRendezvous>,
) -> Result<(), crate::nostr::BootstrapError> {
let advert = self.build_overlay_advert(bootstrap).await;
bootstrap.update_local_advert(advert).await
let snapshot = self.advert_transport_snapshot();
self.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> {
@@ -2463,7 +2286,13 @@ impl Node {
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)
.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());
@@ -2531,7 +2360,7 @@ impl Node {
};
if peer
.transport_id()
.map(|id| self.bootstrap_transports.contains(&id))
.map(|id| self.nostr_rendezvous.is_bootstrap_transport(&id))
.unwrap_or(false)
{
return false;
@@ -2727,17 +2556,16 @@ impl Node {
transport_id,
crate::transport::TransportHandle::Udp(transport),
);
self.bootstrap_transports.insert(transport_id);
self.bootstrap_transport_npubs
.insert(transport_id, traversal.peer_npub.clone());
self.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.bootstrap_transports.remove(&transport_id);
self.bootstrap_transport_npubs.remove(&transport_id);
self.nostr_rendezvous
.remove_bootstrap_transport(&transport_id);
if let Some(mut handle) = self.transports.remove(&transport_id) {
let _ = handle.stop().await;
}
src/node/mod.rs
+12 -35
View File
@@ -65,7 +65,7 @@ use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx};
use crate::utils::index::IndexAllocator;
use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity, TreeCoordinate};
use rand::Rng;
use std::collections::{BTreeSet, HashMap, HashSet, VecDeque};
use std::collections::{BTreeSet, HashMap, VecDeque};
use std::fmt;
use std::sync::Arc;
use std::thread::JoinHandle;
@@ -444,31 +444,16 @@ pub struct Node {
/// are exhausted.
retry_pending: HashMap<NodeAddr, retry::RetryState>,
/// Optional Nostr/STUN overlay discovery coordinator for `udp:nat` peers.
nostr_rendezvous: Option<Arc<crate::nostr::NostrRendezvous>>,
/// Node-side driver state for the Nostr overlay peer-rendezvous
/// subsystem: the engine handle, its startup timestamp, the one-shot
/// startup-sweep latch, and the per-peer bootstrap-transport bookkeeping
/// adopted from NAT-traversal handoffs.
nostr_rendezvous: crate::nostr::RendezvousDriver,
/// mDNS / DNS-SD responder + browser for local-link peer discovery.
/// Identity is unverified at this layer — the Noise XX handshake
/// initiated against an mDNS-observed endpoint is what proves the
/// peer holds the matching private key.
lan_rendezvous: Option<Arc<crate::mdns::LanRendezvous>>,
/// Wall-clock ms when Nostr discovery successfully started, used to
/// schedule the one-shot startup advert sweep after a settle delay.
/// `None` until discovery comes up; remains `None` if discovery is
/// disabled or failed to start.
nostr_rendezvous_started_at_ms: Option<u64>,
/// Whether the one-shot startup advert sweep has run. Set to true
/// after the first sweep fires (under `policy: open`); thereafter
/// only the per-tick `queue_open_discovery_retries` continues.
startup_open_discovery_sweep_done: bool,
/// Per-peer UDP transports adopted from NAT traversal handoff.
bootstrap_transports: HashSet<TransportId>,
/// Originating peer npub (bech32) for each adopted bootstrap
/// transport, captured at `adopt_established_traversal` time.
/// Populated alongside `bootstrap_transports`; cleared in
/// `cleanup_bootstrap_transport_if_unused`. Used by the rx loop to
/// route fatal-protocol-mismatch observations back to the
/// Nostr-discovery `failure_state` for long cooldown application.
bootstrap_transport_npubs: HashMap<TransportId, String>,
// === Periodic Parent Re-evaluation ===
/// Timestamp of last periodic parent re-evaluation (for pacing).
@@ -684,12 +669,8 @@ impl Node {
),
pending_connects: Vec::new(),
retry_pending: HashMap::new(),
nostr_rendezvous: None,
nostr_rendezvous_started_at_ms: None,
nostr_rendezvous: crate::nostr::RendezvousDriver::default(),
lan_rendezvous: None,
startup_open_discovery_sweep_done: false,
bootstrap_transports: HashSet::new(),
bootstrap_transport_npubs: HashMap::new(),
last_parent_reeval: None,
last_congestion_log: None,
estimated_mesh_size: None,
@@ -846,12 +827,8 @@ impl Node {
lookup: Lookup::new(LookupBackoff::new(), LookupForwardRateLimiter::new()),
pending_connects: Vec::new(),
retry_pending: HashMap::new(),
nostr_rendezvous: None,
nostr_rendezvous_started_at_ms: None,
nostr_rendezvous: crate::nostr::RendezvousDriver::default(),
lan_rendezvous: None,
startup_open_discovery_sweep_done: false,
bootstrap_transports: HashSet::new(),
bootstrap_transport_npubs: HashMap::new(),
last_parent_reeval: None,
last_congestion_log: None,
estimated_mesh_size: None,
@@ -2292,7 +2269,7 @@ impl Node {
}
pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) {
if !self.bootstrap_transports.contains(&transport_id) {
if !self.nostr_rendezvous.is_bootstrap_transport(&transport_id) {
return;
}
@@ -2322,8 +2299,8 @@ impl Node {
"bootstrap transport has no remaining references; dropping"
);
self.bootstrap_transports.remove(&transport_id);
self.bootstrap_transport_npubs.remove(&transport_id);
self.nostr_rendezvous
.remove_bootstrap_transport(&transport_id);
self.transport_drops.remove(&transport_id);
self.transports.remove(&transport_id);
}
@@ -2399,7 +2376,7 @@ impl Node {
/// Used by control queries (`show_peers` per-peer Nostr-traversal
/// state) to read failure-state without taking shared ownership.
pub fn nostr_rendezvous_handle(&self) -> Option<&crate::nostr::NostrRendezvous> {
self.nostr_rendezvous.as_deref()
self.nostr_rendezvous.engine()
}
/// Iterate over all peer node IDs.
src/node/retry.rs
+2 -2
View File
@@ -282,7 +282,7 @@ impl Node {
// evicts if the relay has nothing, otherwise leaves it. Cheap
// (one Filter fetch with 2s timeout) and bounded by the retry
// backoff cadence.
if let Some(bootstrap) = self.nostr_rendezvous.clone() {
if let Some(bootstrap) = self.nostr_rendezvous.engine_arc() {
let _ = bootstrap
.refetch_advert_for_stale_check(&peer_config.npub)
.await;
@@ -318,7 +318,7 @@ impl Node {
// entry expires. Force a re-fetch so the next retry tick
// picks up fresh endpoints.
if matches!(e, NodeError::NoTransportForType(_))
&& let Some(bootstrap) = self.nostr_rendezvous.clone()
&& let Some(bootstrap) = self.nostr_rendezvous.engine_arc()
{
let npub = peer_config.npub.clone();
tokio::spawn(async move {
src/node/tests/unit.rs
+3 -3
View File
@@ -1130,7 +1130,7 @@ async fn test_nostr_traversal_failure_skips_connected_peer() {
peer_config: crate::config::PeerConfig::new(peer_identity.npub(), "udp", "127.0.0.1:9"),
reason: "stale traversal failure".to_string(),
});
node.nostr_rendezvous = Some(bootstrap.clone());
node.nostr_rendezvous.set_engine(bootstrap.clone());
node.poll_nostr_rendezvous().await;
@@ -1170,7 +1170,7 @@ async fn test_nostr_traversal_established_skips_connected_peer() {
socket,
),
});
node.nostr_rendezvous = Some(bootstrap.clone());
node.nostr_rendezvous.set_engine(bootstrap.clone());
node.poll_nostr_rendezvous().await;
@@ -1727,7 +1727,7 @@ async fn poll_nostr_rendezvous_established_gated_at_capacity() {
socket,
),
});
node.nostr_rendezvous = Some(bootstrap.clone());
node.nostr_rendezvous.set_engine(bootstrap.clone());
let before_peers = node.peer_count();
let before_links = node.link_count();