diff --git a/src/core/mod.rs b/src/core/mod.rs index 13e7d74..5e3c808 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -415,7 +415,9 @@ async fn run_core_loop( RoomEvent::PeerJoined(peer_id, state) => { // Establish the audio connection as soon as the peer // is known (the transport dedupes the full-mesh race). - transport_events.connect_peer(peer_id).await; + // Hand over the full address so reconnects can dial + // it directly rather than via the gossip lookup. + transport_events.connect_peer(state.addr.clone()).await; let _ = ui_tx_events.send(UiEvent::PeerJoined { id: peer_id, state }).await; } RoomEvent::PeerLeft(peer_id) => { @@ -424,6 +426,12 @@ async fn run_core_loop( let _ = ui_tx_events.send(UiEvent::PeerLeft { id: peer_id }).await; } RoomEvent::PeerUpdated(peer_id, state) => { + // A re-announce may carry a fresh address (peer back + // on a new network). Refresh the retained dial target + // so a later reconnect re-reaches it. Idempotent: an + // ordinary mute/unmute update just re-records the same + // address. + transport_events.connect_peer(state.addr.clone()).await; let _ = ui_tx_events.send(UiEvent::PeerUpdated { id: peer_id, state }).await; } } diff --git a/src/network/iroh_impl.rs b/src/network/iroh_impl.rs index b88ba78..35e8572 100644 --- a/src/network/iroh_impl.rs +++ b/src/network/iroh_impl.rs @@ -36,6 +36,12 @@ struct Shared { /// inserts its handle when a link comes up and removes it when the link dies, /// so `broadcast` only ever targets peers that are currently reachable. senders: StdMutex>>, + /// Last-known full address (relay + direct addrs) per peer, retained for the + /// supervisor's lifetime and refreshed whenever the peer (re)announces. The + /// dialer dials this directly so a reconnect never depends on the address + /// lookup, which a transient gossip `Leave`/`NeighborDown` can purge — the + /// bug that left the dialer redial-looping with "no address" forever. + addrs: StdMutex>, /// One supervisor task per known peer. peers: tokio::sync::Mutex>, incoming_tx: mpsc::Sender<(EndpointId, Bytes)>, @@ -88,6 +94,7 @@ impl Shared { /// cancels any in-flight dial or live link — and drops its send handle. async fn remove(&self, peer_id: EndpointId) { self.senders.lock().unwrap().remove(&peer_id); + self.addrs.lock().unwrap().remove(&peer_id); if let Some(handle) = self.peers.lock().await.remove(&peer_id) { handle.supervisor.abort(); crate::log_msg(&format!("Transport: stopped supervising peer {:?}", peer_id)); @@ -207,7 +214,17 @@ async fn obtain_conn( ) -> Option { if is_dialer { loop { - match shared.endpoint.connect(peer_id, AUDIO_ALPN).await { + // Dial the retained full address (relay + direct addrs) so reconnects + // don't depend on the address lookup; it's re-read each attempt so a + // refreshed address (peer back on a new one) is picked up next retry. + // Fall back to the bare id (lookup/relay resolution) only if we somehow + // never recorded an address — the dialer path always carries one. + let target = shared.addrs.lock().unwrap().get(&peer_id).cloned(); + let result = match target { + Some(addr) => shared.endpoint.connect(addr, AUDIO_ALPN).await, + None => shared.endpoint.connect(peer_id, AUDIO_ALPN).await, + }; + match result { Ok(conn) => return Some(conn), Err(e) => { crate::log_msg(&format!( @@ -278,6 +295,7 @@ impl IrohTransport { endpoint, self_id, senders: StdMutex::new(HashMap::new()), + addrs: StdMutex::new(HashMap::new()), peers: tokio::sync::Mutex::new(HashMap::new()), incoming_tx, conn_events_tx, @@ -296,7 +314,11 @@ impl IrohTransport { #[async_trait] impl NetworkTransport for IrohTransport { - async fn connect_peer(&self, peer_id: EndpointId) { + async fn connect_peer(&self, addr: iroh::EndpointAddr) { + let peer_id = addr.id; + // Retain (or refresh) the peer's full address so the dialer can re-reach + // it across reconnects without leaning on the address lookup. + self.shared.addrs.lock().unwrap().insert(peer_id, addr); // Start (or confirm) the peer's supervisor. It owns dialing/accepting and // reconnection from here on; this call is just the reactive trigger. self.shared.ensure_supervisor(peer_id).await; diff --git a/src/network/mod.rs b/src/network/mod.rs index 0d72631..4903a69 100644 --- a/src/network/mod.rs +++ b/src/network/mod.rs @@ -82,8 +82,13 @@ impl FromStr for PeerSpeakTicket { #[async_trait] pub trait NetworkTransport: Send + Sync { /// Establish (or ensure) a connection to a peer and set up its send path. - /// Idempotent; safe to call again for an already-connected peer. - async fn connect_peer(&self, peer_id: EndpointId); + /// Idempotent; safe to call again for an already-connected peer — a repeat + /// call refreshes the retained dial address, so calling it again when a peer + /// re-announces (e.g. on a new address) keeps the dialer able to re-reach it. + /// The full `EndpointAddr` (relay + direct addrs) is retained so reconnects + /// dial it directly instead of depending on a lookup that a transient gossip + /// `Leave`/`NeighborDown` may have purged. + async fn connect_peer(&self, addr: iroh::EndpointAddr); /// Tear down the connection and send path for a peer that has left. async fn disconnect_peer(&self, peer_id: EndpointId); diff --git a/tests/transport_loopback.rs b/tests/transport_loopback.rs index 39ed884..251de9f 100644 --- a/tests/transport_loopback.rs +++ b/tests/transport_loopback.rs @@ -164,7 +164,6 @@ async fn loopback_sequenced_audio_reaches_peer_and_decodes() { b.lookup.add_endpoint_info(a.endpoint.addr()); let a_id = a.endpoint.id(); - let b_id = b.endpoint.id(); // Subscribe to incoming datagrams on B before any are sent. let mut b_rx = b.transport.receive_datagrams().await.expect("subscribe B"); @@ -172,8 +171,8 @@ async fn loopback_sequenced_audio_reaches_peer_and_decodes() { // Reactive connection setup, exactly as core does on peer-join. Calling on // both sides is fine: the lower EndpointId dials, the higher accepts, and a // single shared connection forms. - a.transport.connect_peer(b_id).await; - b.transport.connect_peer(a_id).await; + a.transport.connect_peer(b.endpoint.addr()).await; + b.transport.connect_peer(a.endpoint.addr()).await; // Let the dial + accept registration settle. tokio::time::sleep(Duration::from_millis(500)).await; @@ -276,7 +275,7 @@ async fn dialer_reconnects_after_link_drops() { .subscribe_conn_events() .await .expect("subscribe dialer conn events"); - dialer.transport.connect_peer(peer_id).await; + dialer.transport.connect_peer(peer.endpoint.addr()).await; // The peer accepts the dialer's initial connection. let conn1 = tokio::time::timeout(Duration::from_secs(10), peer.conns_rx.recv()) @@ -319,3 +318,78 @@ async fn dialer_reconnects_after_link_drops() { "audio should resume after reconnect; got {received} frames" ); } + +#[tokio::test] +async fn dialer_connects_and_reconnects_without_an_address_lookup() { + // Regression for the redial-loop bug: a transient drop that also triggers a + // gossip Leave/NeighborDown purges the peer from the address lookup, so a + // dialer that resolves by bare id (the old behavior) has nothing to dial and + // loops forever. The fix retains the peer's full address in the transport and + // dials it directly. To prove the retained address — not the lookup — is what + // carries the dial, we never seed the dialer's lookup with the peer at all + // (and disable the relay), so a bare-id dial could not resolve an address for + // either the initial connect or the reconnect; only `connect_peer`'s retained + // EndpointAddr makes them possible. + let s1 = iroh::SecretKey::generate(); + let s2 = iroh::SecretKey::generate(); + let (dialer_secret, peer_secret) = if s1.public().to_string() < s2.public().to_string() { + (s1, s2) + } else { + (s2, s1) + }; + + let dialer = spawn_node_with_key(dialer_secret).await; + let mut peer = spawn_capture_peer(peer_secret).await; + + let dialer_id = dialer.endpoint.id(); + let peer_id = peer.endpoint.id(); + assert!(dialer_id.to_string() < peer_id.to_string()); + + // Deliberately leave the dialer's lookup empty for the peer. (The peer side + // only ever accepts, so it needs no addressing for the dialer.) + let mut conn_events = dialer + .transport + .subscribe_conn_events() + .await + .expect("subscribe dialer conn events"); + + // The full address handed here is the dialer's only way to reach the peer. + dialer.transport.connect_peer(peer.endpoint.addr()).await; + + let conn1 = tokio::time::timeout(Duration::from_secs(10), peer.conns_rx.recv()) + .await + .expect("timed out awaiting initial connection (retained address path)") + .expect("connection channel closed"); + assert!( + await_reconnect(&mut conn_events, tokio::time::Instant::now() + Duration::from_secs(10)).await, + "initial link should report Connecting then Connected" + ); + + // Drop the live link from the peer side so the dialer must reconnect — again + // with no lookup to fall back on. + conn1.close(0u32.into(), b"transient drop"); + drop(conn1); + + let conn2 = tokio::time::timeout(Duration::from_secs(15), peer.conns_rx.recv()) + .await + .expect("timed out awaiting reconnect (retained address path)") + .expect("connection channel closed"); + assert!( + await_reconnect(&mut conn_events, tokio::time::Instant::now() + Duration::from_secs(15)).await, + "reconnect should report Connecting then Connected with no lookup at all" + ); + + // Confirm the rebuilt link actually carries audio. + tokio::time::sleep(Duration::from_millis(200)).await; + let mut enc = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap(); + for seq in 0..30u32 { + dialer.transport.broadcast(packet(&mut enc, seq)); + tokio::time::sleep(Duration::from_millis(5)).await; + } + + let received = count_audio(&conn2, 25, tokio::time::Instant::now() + Duration::from_secs(3)).await; + assert!( + received >= 20, + "audio should resume after reconnecting via the retained address; got {received} frames" + ); +}