Merge master into next (mesh-size OR-union estimate; log-level hygiene)

Brings the OR-union mesh-size estimator fix and the per-peer / capacity-cap
log-level demotions forward to next.

Conflict resolution: kept next's XX handshake structure and re-applied the
intent at next's own sites. compute_mesh_size auto-merged (next already
carries the config() accessor). The K-bit cutover promotion log demotion
applied to next's richer index-bearing form in encrypted.rs.

master's log changes lived in handle_msg1's IK-only identity/promotion
blocks, which have no XX equivalent there (next defers identity and
promotion to handle_msg3). next's equivalents were demoted in place:
"Connection promoted to active peer" and "Peer restart detected (epoch
mismatch)" info -> debug, and the duplicate "Inbound peer promoted to
active" removed. The two XX-only restart variants ("during FMP rekey" and
"during promotion") were also demoted info -> debug so the restart-detected
class is uniformly at debug on next.

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"

src/node/handlers/dispatch.rs

@@ -211,7 +211,7 @@ impl Node {
         let remaining_peers: Vec<NodeAddr> = 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,

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,

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");
             }
         }
 

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,

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"

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<BloomFilter> = 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<BloomFilter>, 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<u64> 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<u64> 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"

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<NodeAddr> = (0..6u8).map(|i| mk(0x10, i)).collect();
+    let parent_only: Vec<NodeAddr> = (0..3u8).map(|i| mk(0x20, i)).collect();
+    let child_only: Vec<NodeAddr> = (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() {

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,