diff --git a/src/app/mod.rs b/src/app/mod.rs index 3a25827..a933648 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -10,7 +10,7 @@ use iced::{ Color, Background, Border, Element, Subscription, Task, Theme, Event, keyboard, }; use iroh::EndpointId; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::{Arc, OnceLock}; use tokio::sync::Mutex; @@ -80,6 +80,11 @@ pub struct AppState { peers: HashMap, peer_volumes: HashMap, audio_levels: HashMap, + /// Peers whose audio link is currently down (initial connect or reconnect). + connecting: HashSet, + /// Peers we've had a live link to at least once — used to say "Reconnecting" + /// rather than "Connecting" the second time around. + ever_connected: HashSet, controller: Arc, current_screen: Screen, } @@ -121,6 +126,8 @@ impl Default for AppState { peers: HashMap::new(), peer_volumes: HashMap::new(), audio_levels: HashMap::new(), + connecting: HashSet::new(), + ever_connected: HashSet::new(), controller, current_screen: Screen::Home, } @@ -206,6 +213,8 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { state.ticket = "".to_string(); state.peers.clear(); state.audio_levels.clear(); + state.connecting.clear(); + state.ever_connected.clear(); state.status_message = "Ready to connect".to_string(); state.current_screen = Screen::Home; } @@ -215,10 +224,19 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { UiEvent::PeerLeft { id } => { state.peers.remove(&id); state.audio_levels.remove(&id); + state.connecting.remove(&id); + state.ever_connected.remove(&id); } UiEvent::PeerUpdated { id, state: peer_state } => { state.peers.insert(id, peer_state); } + UiEvent::PeerConnecting { id } => { + state.connecting.insert(id); + } + UiEvent::PeerConnected { id } => { + state.connecting.remove(&id); + state.ever_connected.insert(id); + } UiEvent::AudioLevels(levels) => { for (id, val) in levels { state.audio_levels.insert(id, val); @@ -326,6 +344,7 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { let color_red = Color::from_rgb8(243, 139, 168); let color_maroon = Color::from_rgb8(233, 146, 160); let color_green = Color::from_rgb8(166, 227, 161); + let color_yellow = Color::from_rgb8(249, 226, 175); // Style Helpers let c_style = move |bg: Color, b_color: Color, radius: f32| { @@ -585,9 +604,17 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { for (peer_id, peer) in &state.peers { let level = state.audio_levels.get(peer_id).copied().unwrap_or(0.0); - let is_speaking = level > 0.01; + let is_connecting = state.connecting.contains(peer_id); + let is_speaking = !is_connecting && level > 0.01; - let indicator = if peer.is_muted { + let indicator = if is_connecting { + let label = if state.ever_connected.contains(peer_id) { + "[Reconnecting…]" + } else { + "[Connecting…]" + }; + text(label).size(14).color(color_yellow) + } else if peer.is_muted { text("[Muted]").size(14).color(color_red) } else if is_speaking { text("[Speaking]").size(14).color(color_green) @@ -620,7 +647,7 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { let card = container(card_content) .style(c_style( if is_speaking { color_base } else { color_mantle }, - if is_speaking { color_green } else { color_surface }, + if is_connecting { color_yellow } else if is_speaking { color_green } else { color_surface }, 6.0 )) .padding(12); diff --git a/src/core/messages.rs b/src/core/messages.rs index 5d535e0..37933a8 100644 --- a/src/core/messages.rs +++ b/src/core/messages.rs @@ -24,6 +24,10 @@ pub enum UiEvent { PeerJoined { id: EndpointId, state: PeerState }, PeerLeft { id: EndpointId }, PeerUpdated { id: EndpointId, state: PeerState }, + /// Audio link to a peer is being (re)established — show a connecting state. + PeerConnecting { id: EndpointId }, + /// Audio link to a peer is up and carrying audio. + PeerConnected { id: EndpointId }, AudioLevels(Vec<(EndpointId, f32)>), Error(String), } diff --git a/src/core/mod.rs b/src/core/mod.rs index 61c3641..2238ed6 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -5,7 +5,7 @@ use crate::audio::{AudioBackend, pipewire_impl::PipeWireBackend}; use crate::codec::{AudioEncoder, opus_impl::OpusEncoder}; use crate::core::jitter::{JitterBuffer, FRAME_SAMPLES}; use crate::network::{ - NetworkTransport, RoomState, PeerState, RoomEvent, PeerSpeakTicket, + NetworkTransport, RoomState, PeerState, RoomEvent, ConnEvent, PeerSpeakTicket, iroh_impl::IrohTransport, gossip::IrohGossipState, }; @@ -54,6 +54,7 @@ struct ActiveSession { datagram_task: tokio::task::JoinHandle<()>, mixer_task: tokio::task::JoinHandle<()>, event_task: tokio::task::JoinHandle<()>, + conn_event_task: tokio::task::JoinHandle<()>, } impl ActiveSession { @@ -62,6 +63,7 @@ impl ActiveSession { self.datagram_task.abort(); self.mixer_task.abort(); self.event_task.abort(); + self.conn_event_task.abort(); crate::log_msg("Aborted tasks"); let audio_backend_clone = audio_backend.clone(); @@ -413,6 +415,28 @@ async fn run_core_loop( } }); + // 5. Connection-state forwarder: turns transport link state into + // per-peer UI indicators (connecting / reconnecting vs. live). + let mut conn_events = match transport.subscribe_conn_events().await { + Ok(rx) => rx, + Err(e) => { + let _ = ui_tx.send(UiEvent::Error(format!("Failed to subscribe conn events: {}", e))).await; + continue; + } + }; + let ui_tx_conn = ui_tx.clone(); + let conn_event_task = tokio::spawn(async move { + while let Some(event) = conn_events.recv().await { + let msg = match event { + ConnEvent::Connecting(id) => UiEvent::PeerConnecting { id }, + ConnEvent::Connected(id) => UiEvent::PeerConnected { id }, + }; + if ui_tx_conn.send(msg).await.is_err() { + break; + } + } + }); + let session = ActiveSession { endpoint: endpoint.clone(), router, @@ -421,6 +445,7 @@ async fn run_core_loop( datagram_task, mixer_task, event_task, + conn_event_task, }; let self_id = endpoint.id().to_string(); diff --git a/src/network/iroh_impl.rs b/src/network/iroh_impl.rs index 93d05c5..b88ba78 100644 --- a/src/network/iroh_impl.rs +++ b/src/network/iroh_impl.rs @@ -1,4 +1,4 @@ -use crate::network::{NetworkTransport, NetError}; +use crate::network::{NetworkTransport, NetError, ConnEvent}; use iroh::{Endpoint, EndpointId}; use iroh::endpoint::Connection; use bytes::Bytes; @@ -6,6 +6,7 @@ use tokio::sync::mpsc; use tokio::sync::mpsc::Receiver; use std::sync::{Arc, Mutex as StdMutex}; use std::collections::HashMap; +use std::time::Duration; use async_trait::async_trait; const AUDIO_ALPN: &[u8] = b"peerspeak-audio"; @@ -14,77 +15,215 @@ const AUDIO_ALPN: &[u8] = b"peerspeak-audio"; /// useless latency — keep it shallow and drop the oldest frame when full. const SEND_QUEUE_DEPTH: usize = 8; -/// State shared between the transport and its protocol handler so both inbound -/// (accepted) and outbound (dialed) connections register the same way. +/// How many inbound connections we'll buffer for a single peer's supervisor. +/// During a healthy link the supervisor drains this promptly; a small buffer +/// just covers the brief window between detecting a drop and re-awaiting. +const INBOUND_QUEUE_DEPTH: usize = 4; + +/// Reconnect backoff bounds for the dialing side. Starts gentle so a momentary +/// blip recovers almost instantly, and caps so a long outage doesn't hammer. +const INITIAL_BACKOFF: Duration = Duration::from_millis(250); +const MAX_BACKOFF: Duration = Duration::from_secs(5); + +/// State shared between the transport, its protocol handler, and every per-peer +/// supervisor task. One supervisor owns a peer's whole connection lifecycle. struct Shared { + /// Used by dialer-side supervisors to (re)establish connections. + endpoint: Endpoint, + self_id: EndpointId, /// Sync-lockable send handles, so `broadcast` can fan out from the (non-async) - /// capture/encode thread without touching the Tokio runtime. + /// capture/encode thread without touching the Tokio runtime. A supervisor + /// 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>>, - /// Per-peer task handles + a retained connection clone. Keeping the clone - /// alive is what stops iroh from closing an accepted connection once the - /// `accept()` future returns. - peers: tokio::sync::Mutex>, + /// One supervisor task per known peer. + peers: tokio::sync::Mutex>, incoming_tx: mpsc::Sender<(EndpointId, Bytes)>, + /// Best-effort link-state notifications for the UI (connecting / connected). + conn_events_tx: mpsc::Sender, } -struct PeerTasks { - send_task: tokio::task::JoinHandle<()>, - read_task: tokio::task::JoinHandle<()>, - _conn: Connection, +struct PeerHandle { + /// The supervisor owning this peer's connect → run → reconnect loop. + supervisor: tokio::task::JoinHandle<()>, + /// Inbound connections accepted for this peer are routed here so the + /// acceptor-side supervisor can pick them up (initial and reconnect). + inbound_tx: mpsc::Sender, +} + +/// Aborts a spawned task when dropped. The supervisor holds its send/read loops +/// in these so that cancelling the supervisor (via `remove`'s `abort`) tears the +/// loops down too — otherwise they'd keep their `Connection` clones alive and +/// the link would never actually close. +struct AbortOnDrop(tokio::task::JoinHandle<()>); + +impl Drop for AbortOnDrop { + fn drop(&mut self) { + self.0.abort(); + } } impl Shared { - /// Register a live connection: spin up its send loop (datagrams out) and - /// read loop (datagrams in). Idempotent — a second registration for an - /// already-known peer is ignored so we never run duplicate loops. - async fn register(self: &Arc, peer_id: EndpointId, conn: Connection) { - { - let peers = self.peers.lock().await; - if peers.contains_key(&peer_id) { - return; - } + /// Ensure a supervisor task exists for `peer_id`, returning its inbound + /// connection sender. Idempotent: a second call for a known peer just hands + /// back the existing sender. Both the reactive `connect_peer` path and the + /// inbound `accept()` path funnel through here, so whichever fires first + /// spins the peer up and the roles stay deterministic by id. + async fn ensure_supervisor(self: &Arc, peer_id: EndpointId) -> mpsc::Sender { + let mut peers = self.peers.lock().await; + if let Some(handle) = peers.get(&peer_id) { + return handle.inbound_tx.clone(); } - let (send_tx, mut send_rx) = mpsc::channel::(SEND_QUEUE_DEPTH); - - let conn_send = conn.clone(); - let send_task = tokio::spawn(async move { - while let Some(data) = send_rx.recv().await { - if conn_send.send_datagram(data).is_err() { - break; - } - } - }); - - let conn_read = conn.clone(); - let incoming_tx = self.incoming_tx.clone(); - let read_task = tokio::spawn(async move { - while let Ok(bytes) = conn_read.read_datagram().await { - if incoming_tx.send((peer_id, bytes)).await.is_err() { - break; - } - } - }); - - self.senders.lock().unwrap().insert(peer_id, send_tx); - self.peers.lock().await.insert( - peer_id, - PeerTasks { send_task, read_task, _conn: conn }, - ); - crate::log_msg(&format!("Transport: registered peer {:?}", peer_id)); + let (inbound_tx, inbound_rx) = mpsc::channel::(INBOUND_QUEUE_DEPTH); + let shared = self.clone(); + let supervisor = tokio::spawn(supervise(shared, peer_id, inbound_rx)); + let inbound_tx_ret = inbound_tx.clone(); + peers.insert(peer_id, PeerHandle { supervisor, inbound_tx }); + crate::log_msg(&format!("Transport: supervising peer {:?}", peer_id)); + inbound_tx_ret } + /// Permanently retire a peer (it left the room). Stops its supervisor — which + /// 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); - if let Some(tasks) = self.peers.lock().await.remove(&peer_id) { - tasks.send_task.abort(); - tasks.read_task.abort(); - // Dropping `_conn` (the last retained clone) closes the connection. - crate::log_msg(&format!("Transport: removed peer {:?}", 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)); } } } +/// Why a peer's live-link wait woke up. +enum Wake { + /// The QUIC connection closed (idle timeout or error) — need to reconnect. + Closed, + /// A fresh inbound connection arrived (acceptor side) — switch to it. + Replacement(Connection), + /// The peer was retired; tear down. + Shutdown, +} + +/// Owns a single peer's connection lifecycle for as long as the peer is in the +/// room: obtain a link, run the send/read loops, and on loss obtain a new one — +/// with capped backoff on the dialing side. The deterministic-initiator rule +/// (lower id dials, higher id accepts) is the same on first connect and every +/// reconnect, so exactly one shared connection forms each time. +async fn supervise( + shared: Arc, + peer_id: EndpointId, + mut inbound_rx: mpsc::Receiver, +) { + let is_dialer = shared.self_id.to_string() < peer_id.to_string(); + let mut backoff = INITIAL_BACKOFF; + + // Show "connecting" until the first link is actually up. + let _ = shared.conn_events_tx.try_send(ConnEvent::Connecting(peer_id)); + + let mut conn = match obtain_conn(&shared, peer_id, is_dialer, &mut inbound_rx, &mut backoff).await { + Some(conn) => conn, + None => return, // retired before we ever connected + }; + + loop { + // A healthy link resets the dialer's backoff for the next outage. + backoff = INITIAL_BACKOFF; + + let (send_tx, mut send_rx) = mpsc::channel::(SEND_QUEUE_DEPTH); + shared.senders.lock().unwrap().insert(peer_id, send_tx); + let _ = shared.conn_events_tx.try_send(ConnEvent::Connected(peer_id)); + crate::log_msg(&format!("Transport: peer {:?} link up", peer_id)); + + // Run until the link dies, a replacement arrives, or we're retired. The + // send/read loops are held in abort-on-drop guards scoped to this block, + // so however we leave it — normal exit or the supervisor being aborted + // mid-`select` — both loops stop and release their `Connection` clones. + let wake = { + let conn_send = conn.clone(); + let _send_guard = AbortOnDrop(tokio::spawn(async move { + while let Some(data) = send_rx.recv().await { + if conn_send.send_datagram(data).is_err() { + break; + } + } + })); + + let conn_read = conn.clone(); + let incoming_tx = shared.incoming_tx.clone(); + let _read_guard = AbortOnDrop(tokio::spawn(async move { + while let Ok(bytes) = conn_read.read_datagram().await { + if incoming_tx.send((peer_id, bytes)).await.is_err() { + break; + } + } + })); + + // Only the acceptor watches for replacements: if the dialer notices + // the drop and re-dials before our own `closed()` fires, we switch to + // the new link immediately instead of waiting out the idle timeout. + tokio::select! { + _ = conn.closed() => Wake::Closed, + inbound = inbound_rx.recv(), if !is_dialer => match inbound { + Some(new_conn) => Wake::Replacement(new_conn), + None => Wake::Shutdown, + }, + } + }; + + // Dropping the send handle makes `broadcast` skip this peer until the + // link is back, instead of piling frames into a dead connection. + shared.senders.lock().unwrap().remove(&peer_id); + + match wake { + Wake::Shutdown => return, + Wake::Replacement(new_conn) => { + crate::log_msg(&format!("Transport: peer {:?} replaced with new inbound link", peer_id)); + let _ = shared.conn_events_tx.try_send(ConnEvent::Connecting(peer_id)); + conn = new_conn; + } + Wake::Closed => { + crate::log_msg(&format!("Transport: peer {:?} link dropped; reconnecting", peer_id)); + let _ = shared.conn_events_tx.try_send(ConnEvent::Connecting(peer_id)); + conn = match obtain_conn(&shared, peer_id, is_dialer, &mut inbound_rx, &mut backoff).await { + Some(conn) => conn, + None => return, // retired while reconnecting + }; + } + } + } +} + +/// Block until we hold a live connection to the peer, or `None` if the peer was +/// retired first. The dialer redials with capped exponential backoff; the +/// acceptor waits for the dialer to come to it (a closed `inbound_rx` means the +/// supervisor was retired). +async fn obtain_conn( + shared: &Arc, + peer_id: EndpointId, + is_dialer: bool, + inbound_rx: &mut mpsc::Receiver, + backoff: &mut Duration, +) -> Option { + if is_dialer { + loop { + match shared.endpoint.connect(peer_id, AUDIO_ALPN).await { + Ok(conn) => return Some(conn), + Err(e) => { + crate::log_msg(&format!( + "Transport: dial to {:?} failed: {:?}; retrying in {:?}", + peer_id, e, backoff + )); + tokio::time::sleep(*backoff).await; + *backoff = (*backoff * 2).min(MAX_BACKOFF); + } + } + } + } else { + inbound_rx.recv().await + } +} + #[derive(Clone)] pub struct AudioProtocol { shared: Arc, @@ -104,38 +243,51 @@ impl iroh::protocol::ProtocolHandler for AudioProtocol { let peer_id = connection.remote_id(); let shared = self.shared.clone(); async move { - // Register and return: the retained connection clone in `PeerTasks` - // keeps the connection open after this future resolves. - shared.register(peer_id, connection).await; + // Deterministic roles: the lower id dials, the higher id accepts. + // If we're the dialer for this peer we never consume inbound links + // (we make our own), so just let this one drop. + if shared.self_id.to_string() < peer_id.to_string() { + return Ok(()); + } + // Route the connection to this peer's supervisor (creating it if the + // inbound link beat the gossip join event). try_send keeps the + // protocol handler from ever blocking; a full queue only happens if + // links are churning, and the supervisor will get the next one. + let inbound_tx = shared.ensure_supervisor(peer_id).await; + if inbound_tx.try_send(connection).is_err() { + crate::log_msg(&format!("Transport: dropped inbound link from {:?} (queue full)", peer_id)); + } Ok(()) } } } pub struct IrohTransport { - endpoint: Endpoint, - self_id: EndpointId, shared: Arc, incoming_rx: tokio::sync::Mutex>>, + conn_events_rx: tokio::sync::Mutex>>, } impl IrohTransport { pub fn new(endpoint: Endpoint) -> (Self, AudioProtocol) { let (incoming_tx, incoming_rx) = mpsc::channel(1000); + let (conn_events_tx, conn_events_rx) = mpsc::channel(100); let self_id = endpoint.id(); let shared = Arc::new(Shared { + endpoint, + self_id, senders: StdMutex::new(HashMap::new()), peers: tokio::sync::Mutex::new(HashMap::new()), incoming_tx, + conn_events_tx, }); let protocol = AudioProtocol { shared: shared.clone() }; let transport = Self { - endpoint, - self_id, shared, incoming_rx: tokio::sync::Mutex::new(Some(incoming_rx)), + conn_events_rx: tokio::sync::Mutex::new(Some(conn_events_rx)), }; (transport, protocol) @@ -145,20 +297,9 @@ impl IrohTransport { #[async_trait] impl NetworkTransport for IrohTransport { async fn connect_peer(&self, peer_id: EndpointId) { - // Deterministic initiator: only the lexicographically-lower id dials, so - // a full-mesh pair forms exactly one shared connection instead of two - // racing ones. The higher id waits for the inbound `accept()`. - if self.self_id.to_string() >= peer_id.to_string() { - return; - } - if self.shared.peers.lock().await.contains_key(&peer_id) { - return; - } - - match self.endpoint.connect(peer_id, AUDIO_ALPN).await { - Ok(conn) => self.shared.register(peer_id, conn).await, - Err(e) => crate::log_msg(&format!("Transport: dial to {:?} failed: {:?}", peer_id, e)), - } + // 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; } async fn disconnect_peer(&self, peer_id: EndpointId) { @@ -180,4 +321,11 @@ impl NetworkTransport for IrohTransport { .take() .ok_or_else(|| NetError::Other("Datagram receiver already subscribed".to_string())) } + + async fn subscribe_conn_events(&self) -> Result, NetError> { + let mut rx_guard = self.conn_events_rx.lock().await; + rx_guard + .take() + .ok_or_else(|| NetError::Other("Connection events already subscribed".to_string())) + } } diff --git a/src/network/mod.rs b/src/network/mod.rs index 00eea11..0d72631 100644 --- a/src/network/mod.rs +++ b/src/network/mod.rs @@ -36,6 +36,18 @@ pub enum RoomEvent { PeerUpdated(EndpointId, PeerState), } +/// Transport-level link state for a peer, surfaced so the UI can show when a +/// peer's audio connection is being (re)established versus actually carrying +/// audio. Distinct from `RoomEvent`: a peer can be present in the gossip room +/// while its audio link is momentarily down and reconnecting. +#[derive(Debug, Clone)] +pub enum ConnEvent { + /// No live audio link yet — initial connect or reconnecting after a drop. + Connecting(EndpointId), + /// A live audio link is established and carrying datagrams. + Connected(EndpointId), +} + #[derive(Serialize, Deserialize, Clone, Debug)] pub struct PeerSpeakTicket { pub host_addr: iroh::EndpointAddr, @@ -83,6 +95,9 @@ pub trait NetworkTransport: Send + Sync { /// Subscribes to incoming datagrams from any peer. async fn receive_datagrams(&self) -> Result, NetError>; + + /// Subscribes to per-peer link-state changes (connecting / connected). + async fn subscribe_conn_events(&self) -> Result, NetError>; } #[async_trait] diff --git a/tests/transport_loopback.rs b/tests/transport_loopback.rs index 94d2b15..39ed884 100644 --- a/tests/transport_loopback.rs +++ b/tests/transport_loopback.rs @@ -8,6 +8,8 @@ use std::sync::Arc; use std::time::Duration; +use tokio::sync::mpsc; + use bytes::Bytes; use iroh::address_lookup::memory::MemoryLookup; use iroh::endpoint::presets; @@ -15,10 +17,13 @@ use iroh::protocol::Router; use iroh::{Endpoint, RelayMode}; use opus::{Application, Channels}; +use iroh::endpoint::Connection; +use iroh::protocol::{AcceptError, ProtocolHandler}; + use peerspeak::codec::AudioEncoder; use peerspeak::codec::opus_impl::OpusEncoder; use peerspeak::core::jitter::{FRAME_SAMPLES, JitterBuffer}; -use peerspeak::network::NetworkTransport; +use peerspeak::network::{ConnEvent, NetworkTransport}; use peerspeak::network::iroh_impl::IrohTransport; const AUDIO_ALPN: &[u8] = b"peerspeak-audio"; @@ -31,9 +36,72 @@ struct Node { } async fn spawn_node() -> Node { + spawn_node_with_key(iroh::SecretKey::generate()).await +} + +/// A minimal protocol handler for the reconnect test's "peer" side: it hands +/// every accepted audio connection out through a channel so the test can hold +/// it, read datagrams from it, and explicitly `close()` it to drive a drop — +/// all on a single, stable endpoint address that the dialer can re-reach. +#[derive(Clone)] +struct CaptureProtocol { + conns_tx: mpsc::UnboundedSender, +} + +impl std::fmt::Debug for CaptureProtocol { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CaptureProtocol").finish_non_exhaustive() + } +} + +impl ProtocolHandler for CaptureProtocol { + fn accept( + &self, + connection: Connection, + ) -> impl std::future::Future> + Send { + let conns_tx = self.conns_tx.clone(); + async move { + // The channel retains the connection, which keeps it open after this + // future resolves (iroh closes accepted connections with no retained + // handle). + let _ = conns_tx.send(connection); + Ok(()) + } + } +} + +struct CapturePeer { + endpoint: Endpoint, + _router: Router, + lookup: MemoryLookup, + conns_rx: mpsc::UnboundedReceiver, +} + +async fn spawn_capture_peer(secret: iroh::SecretKey) -> CapturePeer { let lookup = MemoryLookup::new(); let endpoint = Endpoint::builder(presets::Minimal) - .secret_key(iroh::SecretKey::generate()) + .secret_key(secret) + .relay_mode(RelayMode::Disabled) + .address_lookup(lookup.clone()) + .bind() + .await + .expect("bind capture peer"); + + let (conns_tx, conns_rx) = mpsc::unbounded_channel(); + let router = Router::builder(endpoint.clone()) + .accept(AUDIO_ALPN, CaptureProtocol { conns_tx }) + .spawn(); + + CapturePeer { endpoint, _router: router, lookup, conns_rx } +} + +/// Spawn a node with a specific secret key. Reusing a key gives the respawned +/// node the same `EndpointId`, which is how we simulate a peer dropping off the +/// network and coming back (new address, same identity) for the reconnect test. +async fn spawn_node_with_key(secret: iroh::SecretKey) -> Node { + let lookup = MemoryLookup::new(); + let endpoint = Endpoint::builder(presets::Minimal) + .secret_key(secret) // Direct-only: two endpoints on the same host reach each other via the // addresses we seed into each other's lookup, with no external relay. .relay_mode(RelayMode::Disabled) @@ -68,6 +136,24 @@ fn packet(enc: &mut OpusEncoder, seq: u32) -> Bytes { Bytes::from(buf) } +/// Drain link-state events until a `Connected` arrives, returning whether a +/// `Connecting` was seen first (the down→up transition). Panics on timeout so a +/// stuck reconnect fails loudly rather than hanging the suite. +async fn await_reconnect( + rx: &mut mpsc::Receiver, + deadline: tokio::time::Instant, +) -> bool { + let mut saw_connecting = false; + loop { + match tokio::time::timeout_at(deadline, rx.recv()).await { + Ok(Some(ConnEvent::Connecting(_))) => saw_connecting = true, + Ok(Some(ConnEvent::Connected(_))) => return saw_connecting, + Ok(None) => panic!("conn events channel closed unexpectedly"), + Err(_) => panic!("timed out waiting for Connected"), + } + } +} + #[tokio::test] async fn loopback_sequenced_audio_reaches_peer_and_decodes() { let a = spawn_node().await; @@ -142,3 +228,94 @@ async fn loopback_sequenced_audio_reaches_peer_and_decodes() { "expected to decode ~{N} frames, got {decoded_frames} (received {received})" ); } + +/// Read datagrams off a raw connection until `target` arrive or the deadline +/// passes, asserting each carries the 4-byte sequence header. +async fn count_audio(conn: &Connection, target: u32, deadline: tokio::time::Instant) -> u32 { + let mut received = 0; + while let Ok(Ok(bytes)) = tokio::time::timeout_at(deadline, conn.read_datagram()).await { + assert!(bytes.len() >= 4, "packet carries a sequence header"); + received += 1; + if received >= target { + break; + } + } + received +} + +#[tokio::test] +async fn dialer_reconnects_after_link_drops() { + // The unit under test is the IrohTransport dialer. The peer is a hand-rolled + // endpoint we fully control, so we can explicitly close the live connection + // (which the dialer observes promptly, unlike a silent handle drop) and then + // accept the dialer's re-dial — all on one stable address it can re-reach. + // + // The dialer must hold the lower id so the transport's deterministic rule + // makes it dial, so order the two identities up front. + 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()); + + // Seed addresses both ways and form the initial link. + dialer.lookup.add_endpoint_info(peer.endpoint.addr()); + peer.lookup.add_endpoint_info(dialer.endpoint.addr()); + + let mut conn_events = dialer + .transport + .subscribe_conn_events() + .await + .expect("subscribe dialer conn events"); + dialer.transport.connect_peer(peer_id).await; + + // The peer accepts the dialer's initial connection. + let conn1 = tokio::time::timeout(Duration::from_secs(10), peer.conns_rx.recv()) + .await + .expect("timed out awaiting initial connection") + .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 by closing it from the peer side. An explicit close + // sends a CONNECTION_CLOSE the dialer sees right away. + conn1.close(0u32.into(), b"transient drop"); + drop(conn1); + + // The dialer's supervisor should re-dial the (unchanged) peer address, and + // the peer accepts the fresh connection. + let conn2 = tokio::time::timeout(Duration::from_secs(15), peer.conns_rx.recv()) + .await + .expect("timed out awaiting reconnect") + .expect("connection channel closed"); + assert!( + await_reconnect(&mut conn_events, tokio::time::Instant::now() + Duration::from_secs(15)).await, + "dropped link should report Connecting (down) then Connected (recovered)" + ); + + // Audio must flow over the rebuilt link. Give the freshly-registered send + // path a beat, then stream frames and confirm they reach the peer on conn2. + 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 reconnect; got {received} frames" + ); +}