diff --git a/src/node/handlers/connected_udp.rs b/src/node/handlers/connected_udp.rs index 75c0da2..e125b1b 100644 --- a/src/node/handlers/connected_udp.rs +++ b/src/node/handlers/connected_udp.rs @@ -36,7 +36,7 @@ use crate::transport::TransportHandle; #[cfg(any(target_os = "linux", target_os = "macos"))] use std::sync::atomic::{AtomicU64, Ordering::Relaxed}; #[cfg(any(target_os = "linux", target_os = "macos"))] -use tracing::{debug, info, warn}; +use tracing::{debug, warn}; impl Node { /// Tick-driven activation of per-peer connected UDP sockets. @@ -174,7 +174,7 @@ impl Node { } peer.set_connected_udp(socket, drain); crate::perf_profile::record_event(crate::perf_profile::Event::ConnectedUdpInstalled); - info!( + debug!( peer = %self.peer_display_name(node_addr), peer_addr = %peer_socket_addr, "connected UDP socket installed" diff --git a/src/node/handlers/dispatch.rs b/src/node/handlers/dispatch.rs index a6e9d25..8ff72bb 100644 --- a/src/node/handlers/dispatch.rs +++ b/src/node/handlers/dispatch.rs @@ -211,7 +211,7 @@ impl Node { let remaining_peers: Vec = self.peers.keys().copied().collect(); self.bloom_state.mark_all_updates_needed(remaining_peers); - info!( + debug!( peer = %self.peer_display_name(node_addr), link_id = %link_id, tree_changed = tree_changed, diff --git a/src/node/handlers/encrypted.rs b/src/node/handlers/encrypted.rs index 2e3a12f..47b4a5e 100644 --- a/src/node/handlers/encrypted.rs +++ b/src/node/handlers/encrypted.rs @@ -5,7 +5,7 @@ use crate::node::wire::{EncryptedHeader, FLAG_CE, FLAG_KEY_EPOCH, strip_inner_he use crate::noise::NoiseError; use crate::transport::ReceivedPacket; use std::time::Instant; -use tracing::{debug, info, trace, warn}; +use tracing::{debug, trace, warn}; /// Force-remove a peer after this many consecutive decryption failures. const DECRYPT_FAILURE_THRESHOLD: u32 = 20; @@ -91,7 +91,7 @@ impl Node { if let Some(plaintext) = pending_plaintext { let pending_our = peer.pending_our_index(); let pending_their = peer.pending_their_index(); - info!( + debug!( peer = %display_name, our_addr = %our_addr, their_addr = %node_addr, diff --git a/src/node/handlers/forwarding.rs b/src/node/handlers/forwarding.rs index 0bb2a62..3e01778 100644 --- a/src/node/handlers/forwarding.rs +++ b/src/node/handlers/forwarding.rs @@ -120,7 +120,7 @@ impl Node { .unwrap_or(true); if should_log { self.last_congestion_log = Some(now); - warn!(next_hop = %next_hop_addr, "Congestion detected, CE flag set on forwarded packet"); + debug!(next_hop = %next_hop_addr, "Congestion detected, CE flag set on forwarded packet"); } } diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 3c89d7f..dc81d47 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -450,7 +450,7 @@ impl Node { "Cleared stale FSP session after peer restart during FMP rekey" ); } - info!( + debug!( peer = %display_name, "Peer restart detected during FMP rekey, replacing stale endpoint session" ); @@ -1050,7 +1050,7 @@ impl Node { match (existing_epoch, new_epoch) { (Some(existing), Some(new)) if existing != new => { // Epoch mismatch — peer restarted. Tear down stale session. - info!( + debug!( peer = %self.peer_display_name(&peer_node_addr), "Peer restart detected (epoch mismatch), removing stale session" ); @@ -1369,12 +1369,9 @@ impl Node { { peer.set_handshake_msg2(msg2); } - debug!( - peer = %self.peer_display_name(&node_addr), - link_id = %link_id, - our_index = %our_index, - "Inbound peer promoted to active" - ); + // Promotion is logged once by `promote_connection` + // ("Connection promoted to active peer"); no separate + // inbound-path line. // Send initial tree announce to new peer if let Err(e) = self.send_tree_announce_to_peer(&node_addr).await { debug!(peer = %self.peer_display_name(&node_addr), error = %e, "Failed to send initial TreeAnnounce"); @@ -1665,7 +1662,7 @@ impl Node { "Cleared stale FSP session after peer restart during promotion" ); } - info!( + debug!( peer = %self.peer_display_name(&peer_node_addr), winner_link = %link_id, loser_link = %loser_link_id, @@ -1821,7 +1818,7 @@ impl Node { self.bloom_state.add_leaf_dependent(peer_node_addr); } - info!( + debug!( peer = %self.peer_display_name(&peer_node_addr), link_id = %link_id, our_index = %our_index, diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index 0cff5e1..9cd15c0 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -601,7 +601,7 @@ impl Node { .unwrap_or(0); for addr in &dead_peers { - warn!( + debug!( peer = %self.peer_display_name(addr), timeout_secs = self.config().node.link_dead_timeout_secs, "Removing peer: link dead timeout" diff --git a/src/node/mod.rs b/src/node/mod.rs index 180745c..5400812 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -57,7 +57,7 @@ use self::wire::{ ESTABLISHED_HEADER_SIZE, FLAG_CE, FLAG_KEY_EPOCH, build_encrypted, build_established_header, prepend_inner_header, }; -use crate::bloom::BloomState; +use crate::bloom::{BloomFilter, BloomState}; use crate::cache::CoordCache; use crate::node::session::SessionEntry; use crate::peer::{ActivePeer, PeerConnection}; @@ -1295,32 +1295,36 @@ impl Node { let is_root = self.tree_state.is_root(); let max_fpr = self.config().node.bloom.max_inbound_fpr; - let mut total: f64 = 1.0; // count self let mut child_count: u32 = 0; - let mut has_data = false; + + // OR-union of the contributing filters. Summing per-filter + // cardinalities over-counts whenever the filters overlap (a stale + // or oversized parent filter, a topology loop); OR is idempotent, + // so unioning and estimating once deduplicates the overlap. + // Membership is exactly: self + parent + each child inbound_filter. + let mut union: Option = None; + + // Helper: fold a contributing filter into the union, starting it + // from a clone of the first filter (already the right size class). + // BloomFilter::new() uses default size params that may not match + // the stored peer filters, so we must not seed from a fresh filter. + let add_to_union = |union: &mut Option, filter: &BloomFilter| match union { + None => *union = Some(filter.clone()), + Some(existing) => { + // Size-class mismatch is skipped rather than fatal. + let _ = existing.merge(filter); + } + }; // Parent's filter: nodes reachable upward through the tree. - // If any contributing filter is above the FPR cap, we refuse to - // estimate rather than substitute a partial/biased aggregate — - // Node.estimated_mesh_size is already Option and consumers - // (control socket, fipstop, periodic debug log) handle None. if !is_root && let Some(parent) = self.peers.get(&parent_id) && let Some(filter) = parent.inbound_filter() { - match filter.estimated_count(max_fpr) { - Some(n) => { - total += n; - has_data = true; - } - None => { - self.estimated_mesh_size = None; - return; - } - } + add_to_union(&mut union, filter); } - // Children's filters: each child's subtree is disjoint + // Children's filters: each child's subtree is (ideally) disjoint. for (peer_addr, peer) in &self.peers { if peer_addr == &parent_id { continue; @@ -1330,27 +1334,32 @@ impl Node { { child_count += 1; if let Some(filter) = peer.inbound_filter() { - match filter.estimated_count(max_fpr) { - Some(n) => { - total += n; - has_data = true; - } - None => { - self.estimated_mesh_size = None; - return; - } - } + add_to_union(&mut union, filter); } } } - if !has_data { + // No contributing filter at all -> refuse to estimate (matches + // the prior `!has_data` early return). + let Some(mut union) = union else { self.estimated_mesh_size = None; return; - } + }; - let size = total.round() as u64; - self.estimated_mesh_size = Some(size); + // Count self in the union (idempotent). + union.insert(&my_addr); + + // Estimate once. If the union is saturated or above the FPR cap, + // refuse to estimate (matches the prior per-filter None behavior). + // Node.estimated_mesh_size is already Option and consumers + // (control socket, fipstop, periodic debug log) handle None. + let Some(union_estimate) = union.estimated_count(max_fpr) else { + self.estimated_mesh_size = None; + return; + }; + + let union_size = union_estimate.round() as u64; + self.estimated_mesh_size = Some(union_size); // Periodic logging (reuse MMP default interval: 30s) let now = std::time::Instant::now(); @@ -1363,7 +1372,7 @@ impl Node { }; if should_log { tracing::debug!( - estimated_mesh_size = size, + estimated_mesh_size = union_size, peers = self.peers.len(), children = child_count, "Mesh size estimate" diff --git a/src/node/tests/bloom.rs b/src/node/tests/bloom.rs index ccac45b..8f53ab6 100644 --- a/src/node/tests/bloom.rs +++ b/src/node/tests/bloom.rs @@ -533,6 +533,110 @@ fn compute_mesh_size_skips_parent_under_stale_peer_declaration() { ); } +/// Overlapping parent and child inbound filters must be OR-unioned, not +/// summed. The parent and the child here share several NodeAddrs (plus a +/// few distinct ones each). The naive sum of per-filter cardinalities +/// would over-count the shared entries; the union estimate must instead +/// approximate the number of *distinct* addresses across both filters +/// (plus self). This asserts the over-count fingerprint of the old +/// summing code is gone, and the assertion holds on the union result +/// (which is what estimated_mesh_size carries), so it survives removal +/// of the temporary dual-estimator instrumentation. +#[test] +fn compute_mesh_size_unions_overlapping_filters() { + 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(); + + // Build the set of addresses. SHARED appear in both filters; the + // distinct sets appear in only one each. + let mk = |hi: u8, lo: u8| { + let mut bytes = [0u8; 16]; + bytes[0] = hi; + bytes[1] = lo; + NodeAddr::from_bytes(bytes) + }; + let shared: Vec = (0..6u8).map(|i| mk(0x10, i)).collect(); + let parent_only: Vec = (0..3u8).map(|i| mk(0x20, i)).collect(); + let child_only: Vec = (0..3u8).map(|i| mk(0x30, i)).collect(); + + // Distinct addresses across the union: shared + parent_only + + // child_only + self = 6 + 3 + 3 + 1 = 13. The naive sum of the two + // filters' cardinalities would be (6+3) + (6+3) + 1 = 19. + let distinct = shared.len() + parent_only.len() + child_only.len() + 1; // 13 + let naive_sum = (shared.len() + parent_only.len()) + (shared.len() + child_only.len()) + 1; // 19 + + // Generate a parent identity strictly less than my_addr so the + // tree_state defensive check accepts the extension. + 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 addr in shared.iter().chain(parent_only.iter()) { + parent_filter.insert(addr); + } + parent_peer.update_filter(parent_filter, 1, 0); + node.peers.insert(parent_addr, parent_peer); + + // Child Q with a filter that overlaps the parent's on `shared`. + 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 addr in shared.iter().chain(child_only.iter()) { + child_filter.insert(addr); + } + child_peer.update_filter(child_filter, 1, 0); + node.peers.insert(child_addr, child_peer); + + // Wire up the tree so the child names us as parent and our parent is P. + 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); + let parent_decl = 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, parent_ancestry); + node.tree_state_mut() + .update_peer(child_decl, child_ancestry); + 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") as i64; + + // The union estimate should approximate the distinct count (13), not + // the naive sum (19). Bloom cardinality estimation rounds, so allow a + // small absolute tolerance, and require we are clearly below the sum. + let diff = (estimate - distinct as i64).abs(); + assert!( + diff <= 2, + "expected union mesh-size estimate ~{} (distinct addrs), got {}", + distinct, + estimate + ); + assert!( + estimate < naive_sum as i64, + "estimate {} must be below the naive sum {} (overlap should be deduplicated)", + estimate, + naive_sum + ); +} + /// 100-node random graph: bloom filter exchange at scale. #[tokio::test] async fn test_bloom_filter_convergence_100_nodes() { diff --git a/src/transport/tcp/mod.rs b/src/transport/tcp/mod.rs index 6852929..1d8a6a7 100644 --- a/src/transport/tcp/mod.rs +++ b/src/transport/tcp/mod.rs @@ -801,7 +801,7 @@ async fn accept_loop( // operator-facing inbound cap. if stats.pool_inbound_count() >= max_inbound as u64 { stats.record_connection_rejected(); - warn!( + debug!( transport_id = %transport_id, peer_addr = %peer_addr, max = max_inbound,