diff --git a/src/node/bloom.rs b/src/node/bloom.rs index a86f729..5c6586a 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, diff --git a/src/node/mod.rs b/src/node/mod.rs index 20d30c3..f1f0158 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -1176,7 +1176,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/peering/driver.rs b/src/node/peering/driver.rs index 7bd5f71..d2bb529 100644 --- a/src/node/peering/driver.rs +++ b/src/node/peering/driver.rs @@ -104,16 +104,24 @@ impl Node { continue; }; - // 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, so a retry without a refetch dials the same // cached endpoint — and the most common reason a peer landed in the // retry schedule is that endpoint just stopped working (NAT rebind, - // port change, peer restart). Cheap (one Filter fetch, bounded by - // the retry backoff cadence). + // port change, peer restart). + // + // 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. 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.supervisor.nostr_rendezvous.engine_arc() { - 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 0d675c7..28ae440 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -1997,6 +1997,101 @@ 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 = crate::nostr::NostrRendezvous::new_for_test(); + bootstrap + .set_advert_relays_for_test(vec![spawn_blackhole_relay()]) + .await; + node.supervisor + .nostr_rendezvous + .set_engine(Arc::new(bootstrap)); + // Running gate, so the admission short-circuit is not what suppresses the + // dial — same fingerprint the capacity-gate test relies on. + node.supervisor.state = crate::node::NodeState::Running; + + 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::peering::retry::RetryState::new( + crate::config::PeerConfig::new(peer_npub, "udp", "127.0.0.1:9"), + ); + state.retry_after_ms = 0; + state.reconnect = true; + node.peering + .reconciler + .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 + .peering + .reconciler + .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_rendezvous_established_gated_at_capacity() { use crate::nostr::EstablishedTraversal; @@ -2776,3 +2871,49 @@ async fn test_machine_carries_link_direction_and_address_on_dial_and_inbound() { ); assert_eq!(machine.link_id(), *link); } + +#[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/nostr/runtime.rs b/src/nostr/runtime.rs index 11b1800..82cd091 100644 --- a/src/nostr/runtime.rs +++ b/src/nostr/runtime.rs @@ -176,21 +176,51 @@ pub struct NostrRendezvous { } impl NostrRendezvous { - /// Whether the primary Nostr connection task has exited (runtime liveness). + /// Whether the Nostr subsystem has stopped being able to do its job + /// (runtime liveness). /// - /// "Nostr exited" is defined as the primary `connect_task` having finished. - /// It is `Some` for the engine's whole running life (installed in `start`); - /// `shutdown` takes it, leaving `None` — a taken handle means the engine has - /// been shut down, which counts as finished, so a `None` inner maps to - /// `true` (this lets the liveness poll monitor terminate after a stop rather - /// than spinning forever). `connect_task` is a `tokio::sync::Mutex`, so this - /// sync accessor uses the non-blocking `try_lock`: a momentarily-contended - /// lock (only start/stop hold it, briefly) reports "not finished", the safe - /// direction — the 2s liveness poll re-checks next tick and never spuriously - /// degrades a healthy node. + /// "Nostr exited" is defined as *any* of the three service loops that never + /// return by design having finished: + /// + /// - `notify_task` — the inbound receive loop. Without it no advert and no + /// traversal signal is ever observed again. + /// - `publish_task` — the advert publisher. Without it this node stops being + /// discoverable. + /// - `advertise_task` — the refresh ticker that drives the publisher. + /// + /// Each of these is an unconditional `loop` (the notify loop's only `break` + /// is the relay-pool broadcast channel closing, i.e. the pool itself is + /// gone), so a finished handle means the task panicked or was aborted: + /// unrecoverable, which matches the one-way `ChildExited` → `Degraded` + /// latch in the supervisor FSM. + /// + /// Deliberately *not* watched: `connect_task` and `relay_startup_task`. + /// `Client::connect()` only spawns a per-relay background connection task + /// and returns, so `connect_task` finishes moments after start on a + /// perfectly healthy node; `relay_startup_task` breaks out of its retry loop + /// on the first successful subscribe. Watching either reports Degraded on + /// every node forever. + /// + /// Each handle is `Some` for the engine's whole running life (installed in + /// `start`); `shutdown` takes them all, leaving `None` — a taken handle + /// means the engine has been shut down, which counts as finished, so a + /// `None` inner maps to `true` (this lets the liveness poll monitor + /// terminate after a stop rather than spinning forever). The slots are + /// `tokio::sync::Mutex`es, so this sync accessor uses the non-blocking + /// `try_lock`: a momentarily-contended lock (only start/stop hold it, + /// briefly) reports "not finished", the safe direction — the 2s liveness + /// poll re-checks next tick and never spuriously degrades a healthy node. pub fn is_finished(&self) -> bool { - self.connect_task - .try_lock() + Self::task_finished(&self.notify_task) + || Self::task_finished(&self.publish_task) + || Self::task_finished(&self.advertise_task) + } + + /// Liveness of one task slot: finished if the handle is gone (shut down) or + /// the task has completed; "not finished" when the slot is momentarily + /// locked by start/stop. + fn task_finished(slot: &Mutex>>) -> bool { + slot.try_lock() .map(|g| g.as_ref().is_none_or(|h| h.is_finished())) .unwrap_or(false) } @@ -1714,6 +1744,24 @@ impl NostrRendezvous { } } + /// Install the five background-task handles that `start` would install, so + /// liveness tests can drive `is_finished()` without live relays. Each + /// argument is the handle to place in the matching slot. + pub(crate) async fn install_tasks_for_test( + &self, + connect: JoinHandle<()>, + relay_startup: JoinHandle<()>, + notify: JoinHandle<()>, + publish: JoinHandle<()>, + advertise: JoinHandle<()>, + ) { + *self.connect_task.lock().await = Some(connect); + *self.relay_startup_task.lock().await = Some(relay_startup); + *self.notify_task.lock().await = Some(notify); + *self.publish_task.lock().await = Some(publish); + *self.advertise_task.lock().await = Some(advertise); + } + /// Build a `CachedOverlayAdvert` for tests with a single endpoint and /// a generous validity window (one hour from `now_ms()`). pub(crate) fn cached_advert_for_test( @@ -1735,6 +1783,18 @@ impl NostrRendezvous { } } + /// 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/nostr/tests.rs b/src/nostr/tests.rs index 34da425..899f216 100644 --- a/src/nostr/tests.rs +++ b/src/nostr/tests.rs @@ -857,3 +857,106 @@ fn signal_relays_without_an_advert_still_resolves() { "the responder path passes no advert and must still produce a target set" ); } + +/// A `JoinHandle` for a task that has definitely completed. Yields until the +/// runtime has polled the no-op task to completion, so `is_finished()` is +/// deterministically `true` on return. +async fn finished_handle() -> tokio::task::JoinHandle<()> { + let handle = tokio::spawn(async {}); + while !handle.is_finished() { + tokio::task::yield_now().await; + } + handle +} + +/// A `JoinHandle` for a task that never completes. +fn live_handle() -> tokio::task::JoinHandle<()> { + tokio::spawn(std::future::pending::<()>()) +} + +/// The regression case: `connect_task` and `relay_startup_task` both return by +/// design (`Client::connect()` only kicks off per-relay connection tasks; +/// the startup loop breaks on the first successful subscribe), so a healthy +/// node has two finished handles and must still report live. +#[tokio::test] +async fn nostr_liveness_ignores_the_tasks_that_return_by_design() { + let runtime = NostrRendezvous::new_for_test(); + runtime + .install_tasks_for_test( + finished_handle().await, + finished_handle().await, + live_handle(), + live_handle(), + live_handle(), + ) + .await; + assert!( + !runtime.is_finished(), + "a node whose connect/relay-startup tasks have returned normally is healthy" + ); +} + +#[tokio::test] +async fn nostr_liveness_fires_when_the_notify_loop_dies() { + let runtime = NostrRendezvous::new_for_test(); + runtime + .install_tasks_for_test( + live_handle(), + live_handle(), + finished_handle().await, + live_handle(), + live_handle(), + ) + .await; + assert!( + runtime.is_finished(), + "a dead inbound notify loop means no advert or signal is ever received again" + ); +} + +#[tokio::test] +async fn nostr_liveness_fires_when_the_publish_loop_dies() { + let runtime = NostrRendezvous::new_for_test(); + runtime + .install_tasks_for_test( + live_handle(), + live_handle(), + live_handle(), + finished_handle().await, + live_handle(), + ) + .await; + assert!( + runtime.is_finished(), + "a dead publish loop means this node stops being discoverable" + ); +} + +#[tokio::test] +async fn nostr_liveness_fires_when_the_advertise_loop_dies() { + let runtime = NostrRendezvous::new_for_test(); + runtime + .install_tasks_for_test( + live_handle(), + live_handle(), + live_handle(), + live_handle(), + finished_handle().await, + ) + .await; + assert!( + runtime.is_finished(), + "a dead advertise ticker means the advert is never refreshed" + ); +} + +/// `shutdown` takes every handle, leaving `None`. That must read as finished so +/// the 2s liveness poll monitor terminates instead of spinning after a stop. +#[tokio::test] +async fn nostr_liveness_reports_finished_once_the_handles_are_taken() { + let runtime = NostrRendezvous::new_for_test(); + assert!( + runtime.is_finished(), + "no installed handles (post-shutdown) reads as finished" + ); +} diff --git a/src/peer/active.rs b/src/peer/active.rs index 9218ce3..384cb66 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -214,6 +214,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 === /// Current connectivity state. @@ -325,6 +335,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, connectivity: ConnectivityState::Connected, declaration: None, @@ -417,6 +429,8 @@ impl ActivePeer { mmp_config.owd_window_size, )); Self { + npub: identity.npub(), + short_npub: identity.short_npub(), identity, connectivity: ConnectivityState::Connected, declaration: None, @@ -516,8 +530,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 === @@ -1534,12 +1561,37 @@ impl ActivePeer { mod tests { use super::*; use crate::Identity; + use crate::noise::HandshakeState; fn make_peer_identity() -> PeerIdentity { let identity = Identity::generate(); PeerIdentity::from_pubkey(identity.pubkey()) } + /// A completed XX handshake, returning the initiator's session. Only the + /// session's existence matters here: this test is about `with_session` + /// populating the npub cache from the identity it is handed. + fn xx_session_pair() -> (NoiseSession, NoiseSession) { + let initiator_id = Identity::generate(); + let responder_id = Identity::generate(); + let mut initiator = HandshakeState::new_initiator(initiator_id.keypair()); + initiator.set_local_epoch([0xA1, 0xB2, 0xC3, 0xD4, 0x11, 0x22, 0x33, 0x44]); + let mut responder = HandshakeState::new_responder(responder_id.keypair()); + responder.set_local_epoch([0xD4, 0xC3, 0xB2, 0xA1, 0x44, 0x33, 0x22, 0x11]); + + let msg1 = initiator.write_message_1().unwrap(); + responder.read_message_1(&msg1).unwrap(); + let msg2 = responder.write_message_2().unwrap(); + initiator.read_message_2(&msg2).unwrap(); + let msg3 = initiator.write_message_3().unwrap(); + responder.read_message_3(&msg3).unwrap(); + + ( + initiator.into_session().unwrap(), + responder.into_session().unwrap(), + ) + } + fn make_node_addr(val: u8) -> NodeAddr { let mut bytes = [0u8; 16]; bytes[0] = val; @@ -1577,6 +1629,62 @@ 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) = xx_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, + NodeProfile::Full, + NodeProfile::Full, + ); + + 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(); diff --git a/src/proto/bloom/state.rs b/src/proto/bloom/state.rs index abbd451..60d1d56 100644 --- a/src/proto/bloom/state.rs +++ b/src/proto/bloom/state.rs @@ -172,21 +172,79 @@ impl BloomState { peer_addrs: &[NodeAddr], peer_filters: &BTreeMap, ) { - 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: &BTreeMap, + ) -> BTreeMap { + 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: BTreeMap = BTreeMap::new(); + 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/proto/bloom/tests/core.rs b/src/proto/bloom/tests/core.rs index 48f7e02..9284ec2 100644 --- a/src/proto/bloom/tests/core.rs +++ b/src/proto/bloom/tests/core.rs @@ -1,5 +1,9 @@ //! Tests for the `BloomFilter` data structure. +use alloc::collections::BTreeMap; + +use crate::NodeAddr; +use crate::proto::bloom::state::BloomState; use crate::proto::bloom::{BloomError, BloomFilter, DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT}; use crate::testutil::make_node_addr; @@ -365,3 +369,70 @@ fn test_bloom_filter_bit_indices_match_double_hashing_formula() { let absent = make_node_addr(200); assert!(!filter.contains(&absent)); } + +#[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 = BTreeMap::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 = BTreeMap::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/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