From 5564af02f92f914001e7abf2ee6ac74b32c4c6fe Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sat, 20 Jun 2026 04:28:53 -0400 Subject: [PATCH] Spike targeted gossip rebootstrap --- src/network/gossip.rs | 31 ++++++ src/network/mod.rs | 7 +- tests/gossip_rebootstrap.rs | 217 ++++++++++++++++++++++++++++++++++++ tests/transport_loopback.rs | 3 +- 4 files changed, 255 insertions(+), 3 deletions(-) create mode 100644 tests/gossip_rebootstrap.rs diff --git a/src/network/gossip.rs b/src/network/gossip.rs index 23f0e8a..355b055 100644 --- a/src/network/gossip.rs +++ b/src/network/gossip.rs @@ -516,6 +516,37 @@ impl RoomState for IrohGossipState { Ok(()) } + async fn rebootstrap_peers(&self, peers: Vec) -> Result<(), NetError> { + let self_id = self._endpoint.id(); + let mut peer_ids = Vec::new(); + for addr in peers { + if addr.id == self_id || peer_ids.contains(&addr.id) { + continue; + } + self.address_lookup.add_endpoint_info(addr.clone()); + peer_ids.push(addr.id); + } + + if peer_ids.is_empty() { + return Ok(()); + } + + // Clone the sender before awaiting: active_sender is a standard mutex and + // must never be held across an async gossip operation. + let sender = self + .active_sender + .lock() + .unwrap() + .clone() + .ok_or_else(|| NetError::Other("Not in a room".to_string()))?; + + crate::log_msg(&format!("Rebootstrapping gossip peers: {:?}", peer_ids)); + sender + .join_peers(peer_ids) + .await + .map_err(|e| NetError::Gossip(e.to_string())) + } + async fn send_chat(&self, text: String) -> Result<(), NetError> { let name = { let guard = self.self_state.lock().unwrap(); diff --git a/src/network/mod.rs b/src/network/mod.rs index 7b0665a..37d41be 100644 --- a/src/network/mod.rs +++ b/src/network/mod.rs @@ -194,6 +194,12 @@ pub trait RoomState: Send + Sync { /// Updates our local state (e.g. when user mutes/unmutes) and broadcasts it. async fn update_self_state(&self, self_state: PeerState) -> Result<(), NetError>; + /// Ask the active gossip topic to connect to retained peer addresses without + /// leaving or replacing the subscription. This is a recovery primitive only: + /// it does not add peers to the authenticated room roster. A peer becomes + /// active only after its normal signed `Announce` is received and verified. + async fn rebootstrap_peers(&self, peers: Vec) -> Result<(), NetError>; + /// Broadcasts a room text-chat message authored by us (our display name is /// taken from the current self-state). async fn send_chat(&self, text: String) -> Result<(), NetError>; @@ -342,4 +348,3 @@ mod tests { assert_eq!(original, deserialized); } } - diff --git a/tests/gossip_rebootstrap.rs b/tests/gossip_rebootstrap.rs new file mode 100644 index 0000000..f0c5e71 --- /dev/null +++ b/tests/gossip_rebootstrap.rs @@ -0,0 +1,217 @@ +//! Phase-0 spike for post-grace gossip recovery. +//! +//! These tests prove that `GossipSender::join_peers` can restore an existing +//! topic subscription after the other peer drops and rejoins without its own +//! bootstrap target. The second case disables relays, clears the surviving +//! node's lookup, and moves the peer to a fresh endpoint address so only the +//! retained full address passed to `rebootstrap_peers` can drive recovery. + +use std::sync::Arc; +use std::time::Duration; + +use iroh::address_lookup::memory::MemoryLookup; +use iroh::endpoint::presets; +use iroh::protocol::Router; +use iroh::{Endpoint, EndpointAddr, EndpointId, RelayMode, SecretKey}; +use iroh_gossip::net::Gossip; +use tokio::sync::mpsc; + +use peerspeak::network::gossip::IrohGossipState; +use peerspeak::network::{PeerSpeakTicket, PeerState, RoomEvent, RoomState}; + +const EVENT_TIMEOUT: Duration = Duration::from_secs(10); + +struct GossipNode { + endpoint: Endpoint, + lookup: MemoryLookup, + room: Arc, + _router: Router, +} + +async fn spawn_node(secret: SecretKey) -> GossipNode { + let lookup = MemoryLookup::new(); + let endpoint = Endpoint::builder(presets::Minimal) + .secret_key(secret.clone()) + .relay_mode(RelayMode::Disabled) + .address_lookup(lookup.clone()) + .bind() + .await + .expect("bind gossip endpoint"); + let gossip = Gossip::builder().spawn(endpoint.clone()); + let router = Router::builder(endpoint.clone()) + .accept(iroh_gossip::net::GOSSIP_ALPN, gossip.clone()) + .spawn(); + let room = Arc::new(IrohGossipState::new( + endpoint.clone(), + gossip, + lookup.clone(), + secret, + )); + + GossipNode { + endpoint, + lookup, + room, + _router: router, + } +} + +fn state(name: &str, addr: EndpointAddr) -> PeerState { + PeerState { + name: name.to_string(), + is_muted: false, + addr, + sharing: None, + avatar: Default::default(), + } +} + +async fn await_joined(rx: &mut mpsc::Receiver, peer_id: EndpointId) -> PeerState { + let deadline = tokio::time::Instant::now() + EVENT_TIMEOUT; + loop { + match tokio::time::timeout_at(deadline, rx.recv()).await { + Ok(Some(RoomEvent::PeerJoined(id, peer_state))) if id == peer_id => return peer_state, + Ok(Some(_)) => continue, + Ok(None) => panic!("room event channel closed while waiting for PeerJoined"), + Err(_) => panic!("timed out waiting for PeerJoined({peer_id:?})"), + } + } +} + +async fn await_absent(room: &IrohGossipState, peer_id: EndpointId) { + let deadline = tokio::time::Instant::now() + EVENT_TIMEOUT; + loop { + if !room.active_peers().iter().any(|(id, _)| *id == peer_id) { + return; + } + assert!( + tokio::time::Instant::now() < deadline, + "timed out waiting for peer to leave the roster" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +fn ticket(host_addr: EndpointAddr) -> String { + PeerSpeakTicket { + host_addr, + topic_id: rand::random(), + name: "rebootstrap-spike".to_string(), + } + .to_string() +} + +async fn establish_room( + a: &GossipNode, + b: &GossipNode, + ticket: &str, + events_a: &mut mpsc::Receiver, +) { + // B is the ticket host. Its own bootstrap set is empty; A is the only side + // that initially dials, which is also how the recovery setup is controlled. + b.room + .join(ticket, state("Bob", b.endpoint.addr()), vec![]) + .await + .expect("host joins topic"); + a.room + .join(ticket, state("Alice", a.endpoint.addr()), vec![]) + .await + .expect("client joins topic"); + let joined = await_joined(events_a, b.endpoint.id()).await; + assert_eq!(joined.name, "Bob"); +} + +#[tokio::test] +async fn rebootstrap_restores_roster_on_existing_subscription() { + let a = spawn_node(SecretKey::generate()).await; + let b = spawn_node(SecretKey::generate()).await; + let ticket = ticket(b.endpoint.addr()); + let mut events_a = a.room.subscribe_events().await.expect("subscribe A events"); + + establish_room(&a, &b, &ticket, &mut events_a).await; + + // Drop only B's topic subscription. A stays subscribed. B then rejoins as + // the ticket host, so compute_bootstrap removes self and B has nobody to dial. + b.room.leave().await.expect("B leaves topic"); + await_absent(&a.room, b.endpoint.id()).await; + b.room + .join(&ticket, state("Bob recovered", b.endpoint.addr()), vec![]) + .await + .expect("B rejoins without bootstrap peers"); + + tokio::time::sleep(Duration::from_millis(250)).await; + assert!( + !a.room + .active_peers() + .iter() + .any(|(id, _)| *id == b.endpoint.id()), + "B must not recover before A explicitly re-bootstraps it" + ); + + a.room + .rebootstrap_peers(vec![b.endpoint.addr()]) + .await + .expect("targeted gossip re-bootstrap"); + + let recovered = await_joined(&mut events_a, b.endpoint.id()).await; + assert_eq!(recovered.name, "Bob recovered"); +} + +#[tokio::test] +async fn rebootstrap_uses_retained_full_address_with_empty_lookup() { + let a = spawn_node(SecretKey::generate()).await; + let b_secret = SecretKey::generate(); + let b = spawn_node(b_secret.clone()).await; + let b_id = b.endpoint.id(); + let old_b_addr = b.endpoint.addr(); + let ticket = ticket(old_b_addr.clone()); + let mut events_a = a.room.subscribe_events().await.expect("subscribe A events"); + + establish_room(&a, &b, &ticket, &mut events_a).await; + b.room.leave().await.expect("old B leaves topic"); + await_absent(&a.room, b_id).await; + + // Move the same authenticated identity to a newly-bound direct-only endpoint. + // The old cached path is now dead; the new full address is the only valid one. + b.endpoint.close().await; + drop(b); + tokio::time::sleep(Duration::from_millis(250)).await; + let b_rebound = spawn_node(b_secret).await; + let new_b_addr = b_rebound.endpoint.addr(); + assert_eq!(new_b_addr.id, b_id, "identity must survive the rebind"); + assert_ne!( + new_b_addr, old_b_addr, + "rebound peer must have a fresh address" + ); + + b_rebound + .room + .join(&ticket, state("Bob rebound", new_b_addr.clone()), vec![]) + .await + .expect("rebound host joins without bootstrap peers"); + + // Remove the stale lookup entry. `rebootstrap_peers` must seed the retained + // new full address before asking gossip to join the peer by id. + a.lookup.remove_endpoint_info(b_id); + assert!( + a.lookup.get_endpoint_info(b_id).is_none(), + "A lookup starts empty for B" + ); + tokio::time::sleep(Duration::from_millis(250)).await; + assert!( + !a.room.active_peers().iter().any(|(id, _)| *id == b_id), + "rebound B must not be rediscovered without the retained address" + ); + + a.room + .rebootstrap_peers(vec![new_b_addr]) + .await + .expect("retained-address gossip re-bootstrap"); + assert!( + a.lookup.get_endpoint_info(b_id).is_some(), + "re-bootstrap must restore B's address to the lookup" + ); + + let recovered = await_joined(&mut events_a, b_id).await; + assert_eq!(recovered.name, "Bob rebound"); +} diff --git a/tests/transport_loopback.rs b/tests/transport_loopback.rs index 7638375..3efc08a 100644 --- a/tests/transport_loopback.rs +++ b/tests/transport_loopback.rs @@ -25,8 +25,7 @@ use peerspeak::codec::opus_impl::OpusEncoder; use peerspeak::core::jitter::{FRAME_SAMPLES, JitterBuffer}; use peerspeak::network::{ConnEvent, NetworkTransport}; use peerspeak::network::iroh_impl::{AudioRouter, IrohTransport}; - -const AUDIO_ALPN: &[u8] = b"peerspeak-audio"; +use peerspeak::protocol::AUDIO_ALPN; struct Node { endpoint: Endpoint,