Spike targeted gossip rebootstrap

src/network/gossip.rs

@@ -516,6 +516,37 @@ impl RoomState for IrohGossipState {
         Ok(())
     }
 
+    async fn rebootstrap_peers(&self, peers: Vec<EndpointAddr>) -> 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();

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

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<IrohGossipState>,
+    _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<RoomEvent>, 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<RoomEvent>,
+) {
+    // 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");
+}

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,