diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ae94ed..c06afe1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -236,6 +236,26 @@ with v0.3.x peers. ### Fixed +- Outbound connection initiation now honors the `node.limits.max_peers` + cap that was previously only checked on inbound msg1 admission. Four + paths gated: auto-reconnect retries (`process_pending_retries`), + Nostr-mediated discovery's `BootstrapEvent::Established` adoption, and + both sides of the Nostr-mediated NAT-traversal punch (offer initiation + in the runtime's outgoing path, offer acceptance in the responder's + incoming-offer handler). At saturation, a node now performs zero + outbound work on these paths; only existing peer maintenance and + overlay-advert refresh continue. The inbound gate at + `handshake.rs:1114` is unchanged. Introduces a shared + `Node::outbound_admission_check()` helper so the invariant is + grep-able and unit-testable. +- Mesh-size estimator (`compute_mesh_size`) no longer double-counts the + parent's bloom cardinality during the transient cache window after a + local parent-switch. Symptom: `fipsctl show status` / fipstop displayed + mesh size nearly-but-not-exactly doubling during tree rebalancing. + Fix: explicit parent-skip at the head of the children loop, making the + disjoint-subtree invariant structural rather than dependent on + `peer_declaration` cache freshness. Per-peer 500 ms rate-limiter and + overall recompute cadence are unchanged. - Spanning-tree state distribution is now eventually-consistent. Previously every `send_tree_announce_to_all` call site fired only on a local state-change event (parent switch, self-root promotion, diff --git a/src/discovery/nostr/runtime.rs b/src/discovery/nostr/runtime.rs index aceb7e2..1633bc5 100644 --- a/src/discovery/nostr/runtime.rs +++ b/src/discovery/nostr/runtime.rs @@ -1,6 +1,7 @@ use std::collections::{HashMap, HashSet}; use std::net::SocketAddr; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; use nostr::nips::nip17; @@ -165,6 +166,13 @@ pub struct NostrDiscovery { /// (keyed by `TransportId.as_u32()`). Populated on demand by /// `learn_public_udp_addr()` and refreshed by TTL. public_udp_addr_cache: RwLock>, + /// Outbound-admission flag refreshed once per Node tick from + /// `Node::outbound_admission_check()`. Used to suppress NAT-traversal + /// punch initiation (initiator path) and offer acceptance (responder + /// path) when the Node is at `max_peers`. Loose granularity by + /// design: the inbound msg1 gate in `handshake.rs` remains the + /// authoritative cap. + outbound_admission: AtomicBool, } impl NostrDiscovery { @@ -227,6 +235,7 @@ impl NostrDiscovery { advertise_task: Mutex::new(None), failure_state, public_udp_addr_cache: RwLock::new(HashMap::new()), + outbound_admission: AtomicBool::new(true), }); // Subscribe to the relay-pool broadcast channel BEFORE issuing the @@ -248,6 +257,19 @@ impl NostrDiscovery { Ok(runtime) } + /// Update the cached outbound-admission flag. Called once per Node + /// tick with the current value of `Node::outbound_admission_check()`. + /// Cheap atomic store; safe to call unconditionally. + pub fn set_outbound_admission(&self, allow: bool) { + self.outbound_admission.store(allow, Ordering::Relaxed); + } + + /// Read the cached outbound-admission flag. Returns `true` when the + /// Node is below `max_peers` (or `max_peers == 0`), `false` otherwise. + pub(crate) fn outbound_admission_allowed(&self) -> bool { + self.outbound_admission.load(Ordering::Relaxed) + } + pub async fn request_connect(self: &Arc, peer_config: PeerConfig) { let peer_npub = peer_config.npub.clone(); { @@ -981,6 +1003,13 @@ impl NostrDiscovery { peer_config: PeerConfig, ) -> Result { let peer_short = short_npub(&peer_config.npub); + if !self.outbound_admission_allowed() { + debug!( + peer = %peer_short, + "traversal: initiator suppressed, Node at capacity" + ); + return Err(BootstrapError::Disabled); + } debug!(peer = %peer_short, "traversal: initiator starting"); let target_pubkey = PublicKey::parse(&peer_config.npub).map_err(|e| BootstrapError::InvalidPeerNpub { @@ -1163,6 +1192,14 @@ impl NostrDiscovery { sender_npub: String, ) -> Result<(), BootstrapError> { let peer_short = short_npub(&sender_npub); + if !self.outbound_admission_allowed() { + debug!( + peer = %peer_short, + session = %short_id(&offer.session_id), + "traversal: incoming offer dropped, Node at capacity" + ); + return Ok(()); + } let offer_received_at = now_ms(); debug!( peer = %peer_short, @@ -1665,6 +1702,7 @@ impl NostrDiscovery { advertise_task: Mutex::new(None), failure_state, public_udp_addr_cache: RwLock::new(HashMap::new()), + outbound_admission: AtomicBool::new(true), } } diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index f285d75..526d57c 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -446,6 +446,13 @@ impl Node { 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"); } @@ -453,6 +460,15 @@ impl Node { 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(); diff --git a/src/node/mod.rs b/src/node/mod.rs index 1d01920..949b4dc 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -1276,6 +1276,9 @@ impl Node { // Children's filters: each child's subtree is disjoint for (peer_addr, peer) in &self.peers { + if peer_addr == &parent_id { + continue; + } if let Some(decl) = self.tree_state.peer_declaration(peer_addr) && *decl.parent_id() == my_addr { @@ -1435,6 +1438,18 @@ impl Node { self.max_peers = max; } + /// Returns false when we are at or above the configured `max_peers` + /// cap, suppressing outbound connection-initiation. `max_peers == 0` + /// is the "no cap" sentinel and always returns true. The inbound + /// msg1 gate in `handshake.rs` is the authoritative cap; this helper + /// keeps the four outbound initiation paths (auto-reconnect retries, + /// Nostr-discovery `Established` adoption, and both sides of the + /// Nostr-mediated NAT-traversal punch) from doing pointless work + /// when saturated. + pub(crate) fn outbound_admission_check(&self) -> bool { + self.max_peers == 0 || self.peers.len() < self.max_peers + } + /// Set the maximum number of links. pub fn set_max_links(&mut self, max: usize) { self.max_links = max; diff --git a/src/node/retry.rs b/src/node/retry.rs index 8ddf6dc..34f7e0e 100644 --- a/src/node/retry.rs +++ b/src/node/retry.rs @@ -231,6 +231,16 @@ impl Node { return; } + if !self.outbound_admission_check() { + debug!( + peers = self.peers.len(), + max_peers = self.max_peers, + retry_pending = self.retry_pending.len(), + "Suppressing auto-reconnect retries: at capacity" + ); + return; + } + // Collect retries that are due let due: Vec = self .retry_pending diff --git a/src/node/tests/bloom.rs b/src/node/tests/bloom.rs index c8b9264..ccac45b 100644 --- a/src/node/tests/bloom.rs +++ b/src/node/tests/bloom.rs @@ -443,6 +443,96 @@ async fn test_bloom_filter_split_horizon() { cleanup_nodes(&mut nodes).await; } +/// Regression for `compute_mesh_size` parent double-count. +/// +/// Reproduces the stale-peer-declaration window that follows a local +/// parent-switch: our `my_declaration().parent_id()` already names P, but +/// the cached `peer_declaration(P)` is still the pre-switch advert in +/// which P names US as its parent. Without the explicit parent-skip in +/// the children loop, P would be iterated as a child and its bloom +/// cardinality added a second time on top of the parent contribution. +#[test] +fn compute_mesh_size_skips_parent_under_stale_peer_declaration() { + use crate::bloom::BloomFilter; + use crate::peer::ActivePeer; + use crate::tree::ParentDeclaration; + + let mut node = make_node(); + let my_addr = *node.tree_state().my_node_addr(); + + // Generate a parent identity strictly less than my_addr so the + // tree_state defensive check (my_node_addr > parent_root) accepts + // the extension; otherwise recompute_coords would demote us back + // to self-root and is_root() would stay true. + let (parent_identity, parent_addr) = loop { + let candidate = make_peer_identity(); + let addr = *candidate.node_addr(); + if addr < my_addr { + break (candidate, addr); + } + }; + let mut parent_peer = ActivePeer::new(parent_identity, LinkId::new(1), 0); + let mut parent_filter = BloomFilter::new(); + for i in 0..5u8 { + let mut bytes = [0u8; 16]; + bytes[0] = 0x80 | i; // distinct namespace + parent_filter.insert(&NodeAddr::from_bytes(bytes)); + } + parent_peer.update_filter(parent_filter, 1, 0); + node.peers.insert(parent_addr, parent_peer); + + // Inject legitimate child Q with a 3-entry inbound filter. + let child_identity = make_peer_identity(); + let child_addr = *child_identity.node_addr(); + let mut child_peer = ActivePeer::new(child_identity, LinkId::new(2), 0); + let mut child_filter = BloomFilter::new(); + for i in 0..3u8 { + let mut bytes = [0u8; 16]; + bytes[0] = 0xC0 | i; + child_filter.insert(&NodeAddr::from_bytes(bytes)); + } + child_peer.update_filter(child_filter, 1, 0); + node.peers.insert(child_addr, child_peer); + + // Seed parent ancestry first so recompute_coords can extend it and + // flip is_root() to false; child ancestry is for completeness. + let parent_ancestry = crate::tree::TreeCoordinate::root_with_meta(parent_addr, 1, 1); + let child_ancestry = crate::tree::TreeCoordinate::root_with_meta(child_addr, 1, 1); + // Inject the stale-cache scenario: peer_declaration(P) still names + // US (M) as P's parent (the pre-switch advert that the cache hasn't + // refreshed yet). Q is a legitimate child also naming M as parent. + let parent_decl_stale = ParentDeclaration::new(parent_addr, my_addr, 1, 1); + let child_decl = ParentDeclaration::new(child_addr, my_addr, 1, 1); + node.tree_state_mut() + .update_peer(parent_decl_stale, parent_ancestry); + node.tree_state_mut() + .update_peer(child_decl, child_ancestry); + + // Switch our parent to P and recompute coords so root flips off self. + node.tree_state_mut().set_parent(parent_addr, 2, 1); + node.tree_state_mut().recompute_coords(); + assert!( + !node.tree_state().is_root(), + "test setup broken: node should not be its own root after parent switch" + ); + + node.compute_mesh_size(); + + let estimate = node + .estimated_mesh_size() + .expect("estimator should produce a value with filter data present"); + + // Expected (parent counted once + child counted once): 1 + 5 + 3 = 9. + // Unfixed (parent double-counted via children loop): 1 + 2*5 + 3 = 14. + // The estimator's log-based math rounds, so allow +/-1 tolerance. + let diff = (estimate as i64 - 9).abs(); + assert!( + diff <= 1, + "expected mesh-size estimate ~9 (1+5+3), got {} (double-count fingerprint is ~14)", + estimate + ); +} + /// 100-node random graph: bloom filter exchange at scale. #[tokio::test] async fn test_bloom_filter_convergence_100_nodes() { diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index fa77957..03a31d8 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -1360,3 +1360,172 @@ async fn test_seed_path_mtu_noop_for_unknown_transport() { "Seed must be a no-op when transport_id is not registered" ); } + +// === Outbound admission gate tests === + +/// Inject `count` synthetic active peers into `node.peers` so peer_count() +/// reflects a desired saturation level for admission-gate tests. +fn inject_dummy_peers(node: &mut Node, count: usize) { + use crate::peer::ActivePeer; + for i in 0..count { + let identity = make_peer_identity(); + let addr = *identity.node_addr(); + let peer = ActivePeer::new(identity, LinkId::new((i + 1) as u64), 0); + node.peers.insert(addr, peer); + } +} + +#[test] +fn outbound_admission_check_direct() { + // max_peers cap honored: above-cap returns false, below-cap returns true. + let mut node = make_node(); + node.set_max_peers(3); + + assert!(node.outbound_admission_check(), "0/3 should be admissible"); + inject_dummy_peers(&mut node, 2); + assert!(node.outbound_admission_check(), "2/3 should be admissible"); + inject_dummy_peers(&mut node, 1); + assert!( + !node.outbound_admission_check(), + "3/3 (at cap) should suppress" + ); + inject_dummy_peers(&mut node, 1); + assert!( + !node.outbound_admission_check(), + "4/3 (above cap) should suppress" + ); + + // No-cap sentinel: max_peers == 0 admits unconditionally. + let mut uncapped = make_node(); + uncapped.set_max_peers(0); + assert!(uncapped.outbound_admission_check()); + inject_dummy_peers(&mut uncapped, 50); + assert!( + uncapped.outbound_admission_check(), + "max_peers=0 (no cap) must always admit" + ); +} + +#[tokio::test] +async fn process_pending_retries_gated_at_capacity() { + let mut node = make_node(); + node.set_max_peers(2); + inject_dummy_peers(&mut node, 2); + + // Queue a retry that would otherwise be due. + let peer_identity = Identity::generate(); + let peer_npub = peer_identity.npub(); + let peer_node_addr = *PeerIdentity::from_npub(&peer_npub).unwrap().node_addr(); + let mut state = super::super::retry::RetryState::new(crate::config::PeerConfig::new( + peer_npub, + "udp", + "127.0.0.1:9", + )); + state.retry_after_ms = 0; + state.reconnect = true; + node.retry_pending.insert(peer_node_addr, state); + + let before_peers = node.peer_count(); + let before_connections = node.connection_count(); + + node.process_pending_retries(1_000).await; + + // At capacity: gate short-circuits before due-list collection. The + // retry entry must still be present (untouched) and no connection + // attempt may have been started. Without the gate, the due-list + // collector would pick the entry up, fire `initiate_peer_connection` + // (which fails without a registered transport), and the failure + // handler would call `schedule_retry`, bumping `retry_count` to 1. + let state = node + .retry_pending + .get(&peer_node_addr) + .expect("retry entry must be preserved when suppressed at capacity"); + assert_eq!( + state.retry_count, 0, + "gate must short-circuit before initiate_peer_connection; \ + a bumped retry_count is the fingerprint of the ungated path" + ); + assert_eq!( + state.retry_after_ms, 0, + "gate must short-circuit before initiate_peer_connection; \ + retry_after_ms still zero means no attempt fired" + ); + assert_eq!( + node.peer_count(), + before_peers, + "no peer adoption while suppressed" + ); + assert_eq!( + node.connection_count(), + before_connections, + "no connection initiated while suppressed" + ); +} + +#[tokio::test] +async fn poll_nostr_discovery_established_gated_at_capacity() { + use crate::discovery::EstablishedTraversal; + use std::net::UdpSocket; + + let mut node = make_node(); + node.set_max_peers(2); + inject_dummy_peers(&mut node, 2); + + let bootstrap = Arc::new(NostrDiscovery::new_for_test()); + let socket = UdpSocket::bind("127.0.0.1:0").expect("bind local UDP socket"); + let remote_addr = "127.0.0.1:9999".parse().expect("parse remote addr"); + let peer_identity = Identity::generate(); + bootstrap.push_event_for_test(BootstrapEvent::Established { + traversal: EstablishedTraversal::new( + "cap-test-session", + peer_identity.npub(), + remote_addr, + socket, + ), + }); + node.nostr_discovery = Some(bootstrap.clone()); + + let before_peers = node.peer_count(); + let before_links = node.link_count(); + let before_connections = node.connection_count(); + + node.poll_nostr_discovery().await; + + assert_eq!( + node.peer_count(), + before_peers, + "Established event must not add a peer while at capacity" + ); + assert_eq!( + node.link_count(), + before_links, + "Established event must not allocate a link while at capacity" + ); + assert_eq!( + node.connection_count(), + before_connections, + "Established event must not start a handshake while at capacity" + ); +} + +#[test] +fn nostr_discovery_outbound_admission_atomic_roundtrip() { + // Verifies the runtime-side plumbing for the two NAT-traversal gate + // points: the setter mutates the atomic and the (super-visible) + // reader observes the value the Node-side wiring would publish. + let bootstrap = NostrDiscovery::new_for_test(); + assert!( + bootstrap.outbound_admission_allowed(), + "default must allow (start unsaturated)" + ); + bootstrap.set_outbound_admission(false); + assert!( + !bootstrap.outbound_admission_allowed(), + "after suppression store: traversal initiator/responder must see false" + ); + bootstrap.set_outbound_admission(true); + assert!( + bootstrap.outbound_admission_allowed(), + "after recovery store: traversal initiator/responder must see true" + ); +}