From d6457fa74f5cf544ff9076f885e1dee9e5e34064 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 29 Jul 2026 01:27:49 +0000 Subject: [PATCH 1/4] Stop awaiting the advert refetch on the retry tick process_pending_retries runs inline on the node's 1-second rx-loop tick. For each due peer it awaited a Nostr relay fetch carrying a 2-second timeout, and discarded the result. With up to sixteen due peers in one tick body, the timeouts stack: field profiling measured single 2.00 s stalls as the common case and a worst tick of 12.4 s against a 1 s period, with every other rx-loop arm delayed behind it by as much as 4.2 s. Spawn the refetch instead of awaiting it, matching the pattern the failure arm of the same loop already uses thirty lines below. The dial now uses whatever advert is cached at that moment and the refreshed one lands for the next retry of that peer. Since retries are backoff-paced, that defers the benefit by one backoff interval rather than losing it, and the result was already being discarded, so nothing downstream read it. The test drives four due peers whose refetches all hang against a local listener that accepts and never speaks, so the fetch burns its full timeout with no network egress. Awaited, the call takes 8.0 s; spawned, it returns in milliseconds. It also asserts every due peer was still attempted and rescheduled, so a version that skipped the dial entirely cannot pass it. --- src/discovery/nostr/runtime.rs | 12 +++++ src/node/retry.rs | 22 ++++++--- src/node/tests/unit.rs | 87 ++++++++++++++++++++++++++++++++++ 3 files changed, 114 insertions(+), 7 deletions(-) diff --git a/src/discovery/nostr/runtime.rs b/src/discovery/nostr/runtime.rs index ad18b7e..ee6ec8d 100644 --- a/src/discovery/nostr/runtime.rs +++ b/src/discovery/nostr/runtime.rs @@ -1861,6 +1861,18 @@ impl NostrDiscovery { } } + /// Point the test instance's advert relays at explicit URLs. Unit tests + /// that exercise `refetch_advert_for_stale_check` use this to replace the + /// default public relay list with a local blackhole, so the refetch runs + /// its full 2s timeout without touching the network. + pub(crate) async fn set_advert_relays_for_test(&mut self, relays: Vec) { + for url in &relays { + let _ = self.client.add_relay(url.as_str()).await; + } + self.client.connect().await; + self.config.advert_relays = relays; + } + /// Insert a cached advert directly into the in-memory cache. Used by /// unit tests to set up consumer-side state without needing live relays. pub(crate) async fn insert_advert_for_test(&self, npub: String, advert: CachedOverlayAdvert) { diff --git a/src/node/retry.rs b/src/node/retry.rs index 720bed5..19d2f0d 100644 --- a/src/node/retry.rs +++ b/src/node/retry.rs @@ -279,7 +279,7 @@ impl Node { let peer_config = state.peer_config.clone(); - // Refresh the peer's overlay advert before retrying. The cache is + // Kick off a refresh of the peer's overlay advert. The cache is // read-only on hit (see fetch_advert), so every retry without a // refetch dials the same cached endpoint — and the most common // reason a peer ended up in retry_pending is that the cached @@ -289,13 +289,21 @@ impl Node { // // refetch_advert_for_stale_check uses the relay's advert as // ground truth: replaces the cache if there's a newer one, - // evicts if the relay has nothing, otherwise leaves it. Cheap - // (one Filter fetch with 2s timeout) and bounded by the retry - // backoff cadence. + // evicts if the relay has nothing, otherwise leaves it. + // + // Fire-and-forget, NOT awaited: this runs inline on the 1s + // rx-loop tick, and the fetch carries a 2s relay timeout that + // would stall the tick — and every other rx-loop arm with it — + // by up to 2s per due peer, MAX_RETRY_CONNECTIONS_PER_TICK times + // over. So the dial below uses whatever advert is cached now and + // the refreshed one lands for the *next* retry of this peer. + // Retries are backoff-paced, so that defers the benefit by one + // backoff interval rather than losing it. if let Some(bootstrap) = self.nostr_discovery.clone() { - let _ = bootstrap - .refetch_advert_for_stale_check(&peer_config.npub) - .await; + let npub = peer_config.npub.clone(); + tokio::spawn(async move { + let _ = bootstrap.refetch_advert_for_stale_check(&npub).await; + }); } match self.initiate_peer_connection(&peer_config).await { diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index 5abca8d..f3c78a6 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -1707,6 +1707,93 @@ async fn process_pending_retries_gated_at_capacity() { ); } +/// A TCP listener that accepts connections and then never speaks. A relay +/// URL pointed at it makes the nostr client's websocket handshake hang, so +/// `refetch_advert_for_stale_check` burns its full 2s fetch timeout without +/// any network egress. +fn spawn_blackhole_relay() -> String { + use std::net::TcpListener; + let listener = TcpListener::bind("127.0.0.1:0").expect("bind blackhole listener"); + let port = listener.local_addr().expect("blackhole local addr").port(); + std::thread::spawn(move || { + let mut held = Vec::new(); + while let Ok((stream, _)) = listener.accept() { + held.push(stream); + } + }); + format!("ws://127.0.0.1:{port}") +} + +/// The per-tick retry loop must not await the pre-dial advert refetch. +/// +/// `process_pending_retries` runs inline on the node's 1s rx-loop tick. Each +/// due peer's refetch carries a 2s relay-fetch timeout, so awaiting it stalls +/// the whole tick by 2s per peer — up to `MAX_RETRY_CONNECTIONS_PER_TICK` +/// times in one tick body. The refresh is fire-and-forget: it exists to make +/// the *next* retry dial a fresh endpoint, and retries are backoff-paced. +/// +/// Discriminator: wall-clock duration of one `process_pending_retries` call +/// with several due peers whose refetches all hang. Awaited, the call takes +/// `2s * peers`; spawned, it returns without waiting on any of them. +#[tokio::test] +async fn process_pending_retries_does_not_await_advert_refetch() { + use std::time::Instant; + + const DUE_PEERS: usize = 4; + // Awaited: >= 8s (4 x 2s). Spawned: milliseconds. A 3s bound sits far + // from both, so neither machine load nor the 2s timeout's own slack can + // flip the verdict. + const MAX_TICK_MS: u128 = 3_000; + + let mut node = make_node_with_max_peers(64); + + let mut bootstrap = NostrDiscovery::new_for_test(); + bootstrap + .set_advert_relays_for_test(vec![spawn_blackhole_relay()]) + .await; + node.nostr_discovery = Some(Arc::new(bootstrap)); + + let mut queued = Vec::new(); + for _ in 0..DUE_PEERS { + let peer_npub = Identity::generate().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); + queued.push(peer_node_addr); + } + + let started = Instant::now(); + node.process_pending_retries(1_000).await; + let elapsed = started.elapsed(); + + assert!( + elapsed.as_millis() < MAX_TICK_MS, + "retry tick must not block on the advert refetch: took {}ms for {} due peers \ + (a per-peer 2s relay-fetch timeout awaited inline is the fingerprint)", + elapsed.as_millis(), + DUE_PEERS + ); + + // The rest of the loop body is unchanged: every due peer was still + // attempted, failed for want of a transport, and was rescheduled. + for addr in &queued { + let state = node + .retry_pending + .get(addr) + .expect("due peer must remain queued after a failed attempt"); + assert_eq!( + state.retry_count, 1, + "each due peer must still have been attempted and rescheduled" + ); + } +} + #[tokio::test] async fn poll_nostr_discovery_established_gated_at_capacity() { use crate::discovery::EstablishedTraversal; From bf2e7a08928121f13b89bbba06006fcb3a1da0bf Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 29 Jul 2026 01:32:48 +0000 Subject: [PATCH 2/4] Derive each peer's npub once instead of once per tick MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-tick stats snapshot ran a bech32 encode for every tracked peer, and for the common mesh peer — one with no hosts-file entry and no configured alias — it ran a second one, because the display-name fallback chain bottoms out in the same encode. At 240 peers that was 14.1 ms per tick, a third of the tick body and its second largest cost, all of it recomputing values that cannot change. Cache the npub and the shortened npub on the peer at construction. An npub is a pure function of the peer's public key, and the identity is never mutated after construction: there is no setter, no identity_mut, and no assignment to the field anywhere in the tree, so the cache cannot go stale. The display name itself is deliberately NOT cached. Two of its inputs do mutate at runtime — the alias map and the host map, the latter reloaded on this same tick — so a resolved name stored on the peer would go stale on an alias change or a hosts reload. Only the immutable component is memoized. Tests cover both constructors, that the cached npub matches the identity, and that the display name still tracks an alias change. The memoization itself is asserted by pointer stability rather than by timing, so it is deterministic under load. Three deliberate breaks were each caught by exactly one test: re-deriving instead of memoizing, populating one constructor's cache from the wrong source, and reordering the display-name fallback so it stops honoring aliases. --- src/node/mod.rs | 2 +- src/node/tests/unit.rs | 46 +++++++++++++++++++++++ src/peer/active.rs | 83 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 129 insertions(+), 2 deletions(-) diff --git a/src/node/mod.rs b/src/node/mod.rs index c3b2a45..aebfa73 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -1223,7 +1223,7 @@ impl Node { return name.clone(); } if let Some(peer) = self.peers.get(addr) { - return peer.identity().short_npub(); + return peer.short_npub().to_string(); } if let Some(entry) = self.sessions.get(addr) { let (xonly, _) = entry.remote_pubkey().x_only_public_key(); diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index f3c78a6..1a82a96 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -2119,3 +2119,49 @@ fn test_transport_drop_state_steady_counter_fires_once() { assert!(!s.observe_drops(7)); assert!(!s.observe_drops(7)); } + +#[test] +fn test_peer_display_name_uses_cached_short_npub() { + // Path 3 of `peer_display_name` (no host entry, no alias) reads the + // per-peer cached short npub; it must still equal the value derived + // from the peer's identity. + let mut node = make_node(); + let peer_identity_full = Identity::generate(); + let peer_addr = *peer_identity_full.node_addr(); + let peer_identity = PeerIdentity::from_pubkey(peer_identity_full.pubkey()); + node.peers + .insert(peer_addr, ActivePeer::new(peer_identity, LinkId::new(1), 0)); + + assert_eq!( + node.peer_display_name(&peer_addr), + peer_identity.short_npub() + ); +} + +#[test] +fn test_peer_display_name_tracks_alias_change() { + // The display name is NOT cached on the peer: `peer_aliases` is a + // runtime-mutable map (`update_peers` inserts and removes entries), so + // a cached name would go stale. Caching only the immutable short npub + // must leave that tracking intact. + let mut node = make_node(); + let peer_identity_full = Identity::generate(); + let peer_addr = *peer_identity_full.node_addr(); + let peer_identity = PeerIdentity::from_pubkey(peer_identity_full.pubkey()); + node.peers + .insert(peer_addr, ActivePeer::new(peer_identity, LinkId::new(1), 0)); + + assert_eq!( + node.peer_display_name(&peer_addr), + peer_identity.short_npub() + ); + + node.peer_aliases.insert(peer_addr, "gateway".to_string()); + assert_eq!(node.peer_display_name(&peer_addr), "gateway"); + + node.peer_aliases.remove(&peer_addr); + assert_eq!( + node.peer_display_name(&peer_addr), + peer_identity.short_npub() + ); +} diff --git a/src/peer/active.rs b/src/peer/active.rs index 1ae58fd..1b437c5 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -81,6 +81,16 @@ pub struct ActivePeer { // === Identity (Verified) === /// Cryptographic identity (verified via handshake). identity: PeerIdentity, + /// Bech32 npub, derived once at construction. + /// + /// The npub is a pure function of `identity`'s public key, and + /// `identity` is never mutated after construction, so this can never + /// go stale. Deriving it costs a bech32 encode, which the per-tick + /// stats snapshot was paying once per peer per tick. + npub: String, + /// Shortened npub for log/UI display, derived once at construction. + /// Immutable for the same reason as [`ActivePeer::npub`]. + short_npub: String, // === Connection === /// Link used to reach this peer. @@ -224,6 +234,8 @@ impl ActivePeer { pub fn new(identity: PeerIdentity, link_id: LinkId, authenticated_at: u64) -> Self { let now = Instant::now(); Self { + npub: identity.npub(), + short_npub: identity.short_npub(), identity, link_id, connectivity: ConnectivityState::Connected, @@ -310,6 +322,8 @@ impl ActivePeer { ) -> Self { let now = Instant::now(); Self { + npub: identity.npub(), + short_npub: identity.short_npub(), identity, link_id, connectivity: ConnectivityState::Connected, @@ -423,8 +437,21 @@ impl ActivePeer { } /// Get the peer's npub string. + /// + /// Returns a clone of the value cached at construction; the bech32 + /// encode is not repeated. pub fn npub(&self) -> String { - self.identity.npub() + self.npub.clone() + } + + /// Borrow the peer's cached npub without allocating. + pub fn npub_str(&self) -> &str { + &self.npub + } + + /// Borrow the peer's cached shortened npub (e.g. `npub1abcd...wxyz`). + pub fn short_npub(&self) -> &str { + &self.short_npub } // === Connection Accessors === @@ -1219,6 +1246,60 @@ mod tests { assert!(peer.needs_filter_update()); // New peers need filter } + #[test] + fn test_npub_cache_matches_identity() { + let identity = make_peer_identity(); + let peer = ActivePeer::new(identity, LinkId::new(1), 1000); + + assert_eq!(peer.npub(), identity.npub()); + assert_eq!(peer.npub_str(), identity.npub()); + assert_eq!(peer.short_npub(), identity.short_npub()); + } + + #[test] + fn test_npub_cache_matches_identity_with_session() { + // `with_session` builds its own struct literal, so it needs its + // own check that the cache is populated from the same identity. + let identity = make_peer_identity(); + let (session, _peer_session) = ik_session_pair(); + + let peer = ActivePeer::with_session( + identity, + LinkId::new(1), + 1000, + session, + SessionIndex::new(1), + SessionIndex::new(2), + TransportId::new(1), + TransportAddr::from_string("127.0.0.1:9000"), + LinkStats::new(), + true, + &MmpConfig::default(), + None, + ); + + assert_eq!(peer.npub(), identity.npub()); + assert_eq!(peer.short_npub(), identity.short_npub()); + } + + #[test] + fn test_npub_is_memoized_not_rederived() { + // The whole point of the fix: the strings are stored on the peer, + // not recomputed per call. A stored string keeps one heap buffer, + // so repeated borrows have a stable address. A per-call bech32 + // encode would hand back a fresh allocation each time. + let identity = make_peer_identity(); + let peer = ActivePeer::new(identity, LinkId::new(1), 1000); + + let first = peer.npub_str().as_ptr(); + let second = peer.npub_str().as_ptr(); + assert_eq!(first, second); + + let short_first = peer.short_npub().as_ptr(); + let short_second = peer.short_npub().as_ptr(); + assert_eq!(short_first, short_second); + } + #[test] fn test_connectivity_transitions() { let identity = make_peer_identity(); From 77ed64bb8800f31d534a113474279b2dd038c37a Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 29 Jul 2026 01:45:12 +0000 Subject: [PATCH 3/4] Compute every peer's outgoing bloom filter in one sweep Each peer must be sent the union of all other peers' inbound filters, excluding its own contribution. Building that per recipient rebuilt the whole peer-filter map and re-ORed it once for every peer, so a tick that announced to R peers did R kilobyte-scale map builds and R by T merges. At 240 peers this was 20.6 ms per tick, roughly half the tick body, and it was steady work rather than a tail: the per-interval maximum had a median of 34.5 ms. Replace it with a prefix and suffix union sweep that produces every target's filter in one pass, T merges instead of R by T. The packet path gets the same treatment, since marking changed peers had the identical shape once per inbound announce. The result is exactly equal, not approximately. Merging is a bytewise OR, so regrouping the unions cannot change the outcome, and a filter whose size does not match is rejected before any byte is touched, at every merge site in both the old and new arrangement. Every accumulator here is default-sized, so an odd-sized peer filter is skipped in the new code exactly where it was skipped in the old. Cadence, the debounce, the sequence rule and the fill-ratio cap are untouched. A sequence number is still drawn after the debounce re-check and before encoding, so a suppressed peer still consumes none and an encode failure still burns one. Known trade-off, measured rather than assumed: the sweep does its full O(T) work regardless of how many peers are ready, so a tick that announces to only one or two peers now costs about twice what it did. Break-even is around three ready peers, and the saving above that grows without bound. The marking path is a pure win, since it always targets every peer. --- src/bloom/state.rs | 72 +++++++++++++++++++++++++++++++++++++++++----- src/bloom/tests.rs | 67 ++++++++++++++++++++++++++++++++++++++++++ src/node/bloom.rs | 40 +++++++++++++++----------- 3 files changed, 156 insertions(+), 23 deletions(-) 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, From 13a98ae7025def95ec594e12897ded6e62819d18 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 29 Jul 2026 02:09:59 +0000 Subject: [PATCH 4/4] Make a half-scoped reap impossible in ci-cleanup.sh The reap runs two selectors, one on the CI label and one on the compose project, and each flag narrows only its own. So --run-id alone leaves the project sweep broad and --project-prefix alone leaves the label sweep broad, and either way the reap still destroys every concurrent run on the host. Both single-flag forms look scoped and are not. The usage text asserted otherwise. It documented --run-id as leaving other runs alone and --project-prefix as scoping the reap to a single run, and both claims were false for the same reason. A caller passing --run-id alone, on the strength of that line, force-removed the containers of three concurrent runs on 2026-07-29, and its own comment recorded the belief that it was scoped. Rather than ask every caller to remember the pair, close the two half-scoped states here. --run-id now derives the matching project prefix when none is given, built from the existing base so the two cannot drift, and an explicit --project-prefix still wins. --project-prefix without --run-id is refused outright, because there is nothing to derive a label scope from and the quiet failure is someone else's run disappearing. Passing neither flag is untouched: that is the deliberate "reap everything" form a manual cleanup wants. Verified against a decoy container labelled as another run: the scoped reap leaves it up, and reproducing the old broad project sweep removes it, so the check discriminates rather than passing because there was nothing to reap. An earlier version of that check ran when no CI containers existed at all and passed without proving anything. --- testing/ci-cleanup.sh | 51 +++++++++++++++++++++++++++++++++---------- 1 file changed, 40 insertions(+), 11 deletions(-) diff --git a/testing/ci-cleanup.sh b/testing/ci-cleanup.sh index 407b11b..010394d 100755 --- a/testing/ci-cleanup.sh +++ b/testing/ci-cleanup.sh @@ -15,9 +15,15 @@ # with that prefix). # # The generic CI label is shared by every run on the host, so an unscoped label -# sweep would tear down a CONCURRENT run's resources. Pass --run-id to narrow -# the label sweep to one run; a run's own teardown always does. Without it the -# label sweep stays broad, which is what a manual "reap everything" wants. +# sweep would tear down a CONCURRENT run's resources. So would an unscoped +# compose-project sweep, and reap_containers runs BOTH. Neither flag alone is +# therefore enough to spare a concurrent run: --run-id narrows only the label +# sweep, --project-prefix only the project sweep, and whichever is left broad +# reaps everything by itself. This cost three concurrent runs on 2026-07-29, +# via a caller that passed --run-id alone and reasonably believed that scoped +# it. --run-id now implies the matching project prefix, and --project-prefix +# without --run-id is refused, so the dangerous half-scoped states are no +# longer reachable. Passing neither is still the broad "reap everything" form. # # Host-namespace veth interfaces are the one non-docker resource reaped here. # The chaos simulation creates each pair in the host namespace and then moves @@ -37,12 +43,17 @@ # # Usage: # ci-cleanup.sh Reap ALL fips-ci resources (any run) +# ci-cleanup.sh --run-id ID Scope the reap to one run: narrows the +# label sweep to ID and, unless +# --project-prefix says otherwise, +# narrows the project sweep to that run +# too. Does NOT scope the host veth +# sweep — see --veth-suffixes # ci-cleanup.sh --project-prefix P Restrict the compose-project sweep to -# names starting with P (scopes the reap -# to a single run). Does NOT scope the -# host veth sweep — see --veth-suffixes -# ci-cleanup.sh --run-id ID Restrict the label sweep to the run -# labelled ID (leaves other runs alone) +# names starting with P. Requires +# --run-id: on its own it leaves the +# label sweep broad, which reaps every +# concurrent run regardless of P # ci-cleanup.sh --label L Override the CI label (default above) # ci-cleanup.sh --images "a b,c" Also `docker rmi -f` these image tags # (space- or comma-separated) @@ -75,11 +86,12 @@ VETH_SUFFIXES="" # empty AND no --run-id: every simulation veth name # only ever run the harness it need not exist at all, and the reap this script # advertises as the remedy for orphaned interfaces would be a permanent no-op. VETH_IMAGE="" +PROJECT_PREFIX_GIVEN=0 while [[ $# -gt 0 ]]; do case "$1" in --label) LABEL="$2"; shift 2 ;; - --project-prefix) PROJECT_PREFIX="$2"; shift 2 ;; + --project-prefix) PROJECT_PREFIX="$2"; PROJECT_PREFIX_GIVEN=1; shift 2 ;; --run-id) RUN_ID="$2"; shift 2 ;; --images) IMAGES="$2"; shift 2 ;; --veth-suffixes) VETH_SUFFIXES="$2"; shift 2 ;; @@ -89,6 +101,22 @@ while [[ $# -gt 0 ]]; do esac done +# A reap scoped on one axis and broad on the other still destroys every +# concurrent run, because both selectors run. Close both half-scoped states +# here rather than trusting each caller to pass the pair. +if [[ -n "$RUN_ID" && "$PROJECT_PREFIX_GIVEN" -eq 0 ]]; then + # Every run's projects are named "_", so the + # run id is all that is needed. Derived from the base rather than a second + # copy of the literal, so the two cannot drift. + PROJECT_PREFIX="${PROJECT_PREFIX}${RUN_ID}" +fi +if [[ -z "$RUN_ID" && "$PROJECT_PREFIX_GIVEN" -eq 1 ]]; then + echo "ci-cleanup.sh: --project-prefix requires --run-id." >&2 + echo " Without it the label sweep stays broad and reaps every concurrent" >&2 + echo " run regardless of the prefix. Pass both, or neither for a full reap." >&2 + exit 2 +fi + # Resolve the image to run ip(8) in: the caller's choice, then the run's own # image, then any surviving fips test image. That last fallback is what keeps # an unscoped `ci-local.sh --reap` working — it execs this script from inside @@ -116,8 +144,9 @@ fi # never wedge a caller (ci-local's signal trap relies on this being bounded). TMO=30 -# Selector for the label sweep. With --run-id it matches only the named run, so -# a concurrent run's resources are left alone; without it, every CI run. +# Selector for the label sweep. With --run-id it matches only the named run; +# without it, every CI run. Note this is only half of what scopes a reap — the +# compose-project sweep below is the other half, and both must be narrow. if [[ -n "$RUN_ID" ]]; then SWEEP_LABEL="${RUN_LABEL_KEY}=${RUN_ID}" else