diff --git a/src/bloom/state.rs b/src/bloom/state.rs index 5a0f5c8..e47422d 100644 --- a/src/bloom/state.rs +++ b/src/bloom/state.rs @@ -172,21 +172,79 @@ impl BloomState { peer_addrs: &[NodeAddr], peer_filters: &HashMap, ) { - for peer_addr in peer_addrs { - if peer_addr == exclude_from { - continue; - } - let new_filter = self.compute_outgoing_filter(peer_addr, peer_filters); - let changed = match self.last_sent_filters.get(peer_addr) { + let targets: Vec = peer_addrs + .iter() + .filter(|addr| *addr != exclude_from) + .copied() + .collect(); + + for (peer_addr, new_filter) in self.compute_outgoing_filters(&targets, peer_filters) { + let changed = match self.last_sent_filters.get(&peer_addr) { Some(last) => *last != new_filter, None => true, // never sent → must send }; if changed { - self.pending_updates.insert(*peer_addr); + self.pending_updates.insert(peer_addr); } } } + /// Compute the outgoing filter for many peers in one pass. + /// + /// Equivalent to calling [`compute_outgoing_filter`](Self::compute_outgoing_filter) + /// once per target, but linear in the number of contributing peer + /// filters instead of quadratic. The per-peer call rebuilds the whole + /// union from scratch, so computing it for every peer costs + /// O(targets × filters) 1 KB merges; announce fan-out on a + /// large node does exactly that, once per tick and again on every + /// inbound announce. + /// + /// The split-horizon exclusion is the only thing that differs between + /// targets, so the union of "everything except peer i" is assembled + /// from a running prefix union and a precomputed suffix union. Merging + /// is a bytewise OR, which is commutative and associative, so the + /// result is bit-identical to the per-peer computation. + pub fn compute_outgoing_filters( + &self, + targets: &[NodeAddr], + peer_filters: &HashMap, + ) -> HashMap { + let base = self.base_filter(); + let keys: Vec = peer_filters.keys().copied().collect(); + let n = keys.len(); + + // suffix[i] = union of peer_filters[keys[i..]]; suffix[n] is empty. + let mut suffix = vec![BloomFilter::new(); n + 1]; + for i in (0..n).rev() { + let mut acc = suffix[i + 1].clone(); + // Size mismatches are skipped, exactly as in the per-peer path. + let _ = acc.merge(&peer_filters[&keys[i]]); + suffix[i] = acc; + } + + // Filter for a target that contributes nothing: everything merged. + let mut all = base.clone(); + let _ = all.merge(&suffix[0]); + + let mut per_key: HashMap = HashMap::with_capacity(n); + let mut prefix = BloomFilter::new(); + for i in 0..n { + let mut outgoing = base.clone(); + let _ = outgoing.merge(&prefix); + let _ = outgoing.merge(&suffix[i + 1]); + per_key.insert(keys[i], outgoing); + let _ = prefix.merge(&peer_filters[&keys[i]]); + } + + targets + .iter() + .map(|target| { + let filter = per_key.get(target).cloned().unwrap_or_else(|| all.clone()); + (*target, filter) + }) + .collect() + } + /// Compute the outgoing filter for a specific peer. /// /// The filter includes: diff --git a/src/bloom/tests.rs b/src/bloom/tests.rs index 37a7e70..4a8ba83 100644 --- a/src/bloom/tests.rs +++ b/src/bloom/tests.rs @@ -684,3 +684,70 @@ fn test_bloom_state_mark_changed_peers_excludes_source() { assert!(!state.needs_update(&peer1)); } + +#[test] +fn test_compute_outgoing_filters_matches_per_peer() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + state.add_leaf_dependent(make_node_addr(200)); + state.add_leaf_dependent(make_node_addr(201)); + + // Six contributing peers with overlapping content, plus one whose + // filter is a different size and must be skipped by both paths. + let mut peer_filters = HashMap::new(); + for i in 1u8..=6 { + let mut filter = BloomFilter::new(); + for j in 0..5u8 { + filter.insert(&make_node_addr(i.wrapping_mul(7).wrapping_add(j))); + } + peer_filters.insert(make_node_addr(i), filter); + } + let odd_peer = make_node_addr(7); + let mut odd = BloomFilter::with_params(4096, DEFAULT_HASH_COUNT).unwrap(); + odd.insert(&make_node_addr(99)); + peer_filters.insert(odd_peer, odd); + + // Targets: every contributing peer, the odd-sized one, and two peers + // that contribute nothing (non-tree peers get announces too). + let mut targets: Vec = (1u8..=7).map(make_node_addr).collect(); + targets.push(make_node_addr(120)); + targets.push(make_node_addr(121)); + + let batch = state.compute_outgoing_filters(&targets, &peer_filters); + + assert_eq!(batch.len(), targets.len()); + for target in &targets { + let expected = state.compute_outgoing_filter(target, &peer_filters); + assert_eq!( + batch.get(target), + Some(&expected), + "batch filter for {:?} differs from per-peer computation", + target + ); + } + + // Split horizon is real, not vacuous: a contributing peer's own + // entries must be absent from its own outgoing filter, and present + // in another peer's. + let peer3 = make_node_addr(3); + let own_entry = make_node_addr(3u8.wrapping_mul(7)); + assert!(!batch[&peer3].contains(&own_entry)); + assert!(batch[&make_node_addr(1)].contains(&own_entry)); +} + +#[test] +fn test_compute_outgoing_filters_empty_inputs() { + let node = make_node_addr(0); + let state = BloomState::new(node); + let peer_filters = HashMap::new(); + + assert!( + state + .compute_outgoing_filters(&[], &peer_filters) + .is_empty() + ); + + let target = make_node_addr(1); + let batch = state.compute_outgoing_filters(&[target], &peer_filters); + assert_eq!(batch[&target], state.base_filter()); +} diff --git a/src/node/bloom.rs b/src/node/bloom.rs index f16c243..cde819f 100644 --- a/src/node/bloom.rs +++ b/src/node/bloom.rs @@ -29,27 +29,19 @@ impl Node { filters } - /// Build a FilterAnnounce for a specific peer. - /// - /// The outgoing filter excludes the destination peer's own filter - /// to prevent routing loops (don't tell a peer about destinations - /// reachable only through them). - fn build_filter_announce(&mut self, exclude_peer: &NodeAddr) -> FilterAnnounce { - let peer_filters = self.peer_inbound_filters(); - let filter = self - .bloom_state - .compute_outgoing_filter(exclude_peer, &peer_filters); - let sequence = self.bloom_state.next_sequence(); - FilterAnnounce::new(filter, sequence) - } - /// Send a FilterAnnounce to a specific peer, respecting debounce. /// + /// `filter` is the outgoing filter for this peer, already computed + /// with the destination peer's own contribution excluded to prevent + /// routing loops (don't tell a peer about destinations reachable + /// only through them). + /// /// If the peer is rate-limited, the update stays pending for /// delivery on the next tick cycle. pub(super) async fn send_filter_announce_to_peer( &mut self, peer_addr: &NodeAddr, + filter: BloomFilter, ) -> Result<(), NodeError> { let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -64,7 +56,7 @@ impl Node { } // Build and encode - let announce = self.build_filter_announce(peer_addr); + let announce = FilterAnnounce::new(filter, self.bloom_state.next_sequence()); let sent_filter = announce.filter.clone(); let encoded = announce.encode().map_err(|e| NodeError::SendFailed { node_addr: *peer_addr, @@ -142,8 +134,24 @@ impl Node { .copied() .collect(); + if ready.is_empty() { + return; + } + + // One snapshot and one union pass for the whole ready set. The + // send path never mutates peer inbound filters or the tree state, + // and the rx loop holds `&mut self` across the awaits, so the + // snapshot cannot go stale mid-loop. + let peer_filters = self.peer_inbound_filters(); + let mut outgoing = self + .bloom_state + .compute_outgoing_filters(&ready, &peer_filters); + for peer_addr in ready { - if let Err(e) = self.send_filter_announce_to_peer(&peer_addr).await { + let Some(filter) = outgoing.remove(&peer_addr) else { + continue; + }; + if let Err(e) = self.send_filter_announce_to_peer(&peer_addr, filter).await { debug!( peer = %self.peer_display_name(&peer_addr), error = %e,