From 180950badf536d882eec9f380feac00817d470c8 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 6 Jun 2026 18:33:40 +0000 Subject: [PATCH 1/2] node: estimate mesh size by OR-union of filters instead of summing cardinalities The mesh-size estimator summed the per-filter cardinality of the parent filter and each child filter, which assumes those filters are perfectly disjoint. When they overlap -- a stale or oversized parent filter, or a routing loop -- the sum over-counts and inflates the reported mesh size to as much as several times the true size. Estimate the cardinality of the OR-union of the contributing filters (self + parent + children) once instead. OR is idempotent, so any overlap is deduplicated: the result equals the old sum in the disjoint case and stays correct under overlap. The union is seeded from a clone of a contributing filter so it keeps that filter's size class, and a filter whose size class does not match is skipped rather than panicking. The refuse-to-estimate behavior on a saturated or above-cap filter is preserved. Add a regression test with overlapping parent and child filters where the naive sum over-counts and the union estimate tracks the distinct member count. --- src/node/mod.rs | 75 ++++++++++++++++------------- src/node/tests/bloom.rs | 104 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 33 deletions(-) diff --git a/src/node/mod.rs b/src/node/mod.rs index 4b72da8..a2e2171 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -35,7 +35,7 @@ use self::wire::{ FLAG_CE, FLAG_KEY_EPOCH, FLAG_SP, 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}; @@ -1146,32 +1146,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; @@ -1181,27 +1185,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(); @@ -1214,7 +1223,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() { From 974e146bb96945218def87b70b43a448623f60e1 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 6 Jun 2026 19:11:45 +0000 Subject: [PATCH 2/2] node: demote routine per-peer and capacity-cap events from info/warn to debug On a saturated public-mesh node the connection-lifecycle and capacity-cap events fire continuously and drown out the genuinely notable INFO/WARN lines. Demote them to debug and drop a redundant duplicate: - FMP K-bit cutover promotion (encrypted): info -> debug - "Connection promoted to active peer" (handshake): info -> debug, and remove the duplicate "Inbound peer promoted to active" line that shadowed it on the inbound path - "Peer restart detected" (handshake): info -> debug - "Peer removed and state cleaned up" (dispatch): info -> debug - "Rejecting inbound TCP connection (max_inbound_connections reached)" (tcp): warn -> debug - "Congestion detected, CE flag set on forwarded packet" (forwarding): warn -> debug - "Removing peer: link dead timeout" (mmp): warn -> debug These are expected, high-frequency conditions on a busy public node (new and reconnecting peers, ECN CE marking, the inbound connection cap, and link-dead churn), not operator-actionable signals. --- src/node/handlers/dispatch.rs | 2 +- src/node/handlers/encrypted.rs | 4 ++-- src/node/handlers/forwarding.rs | 2 +- src/node/handlers/handshake.rs | 13 +++++-------- src/node/handlers/mmp.rs | 2 +- src/transport/tcp/mod.rs | 2 +- 6 files changed, 11 insertions(+), 14 deletions(-) diff --git a/src/node/handlers/dispatch.rs b/src/node/handlers/dispatch.rs index faaccec..cc491df 100644 --- a/src/node/handlers/dispatch.rs +++ b/src/node/handlers/dispatch.rs @@ -194,7 +194,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 211c4f6..6b3675c 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, FLAG_SP, strip 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; @@ -94,7 +94,7 @@ impl Node { }); if let Some(plaintext) = pending_plaintext { - info!( + debug!( peer = %display_name, "Peer new-epoch frame authenticated, K-bit flip promoting new session" ); diff --git a/src/node/handlers/forwarding.rs b/src/node/handlers/forwarding.rs index 245004b..48ef9e9 100644 --- a/src/node/handlers/forwarding.rs +++ b/src/node/handlers/forwarding.rs @@ -106,7 +106,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 0f4dc26..bad38de 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -248,7 +248,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" ); @@ -510,12 +510,9 @@ impl Node { if let Some(peer) = self.peers.get_mut(&node_addr) { peer.set_handshake_msg2(wire_msg2.clone()); } - 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"); @@ -1166,7 +1163,7 @@ impl Node { self.retry_pending.remove(&peer_node_addr); self.register_identity(peer_node_addr, verified_identity.pubkey_full()); - 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 f9941ae..64cf2a2 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -573,7 +573,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/transport/tcp/mod.rs b/src/transport/tcp/mod.rs index 1e8f71b..f3c20fe 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,