diff --git a/src/core/mod.rs b/src/core/mod.rs index 03f6252..b4a68c2 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -75,6 +75,7 @@ struct ActiveSession { event_task: tokio::task::JoinHandle<()>, conn_event_task: tokio::task::JoinHandle<()>, grace_timers: GraceTimers, + transport: Arc, } impl ActiveSession { @@ -100,6 +101,10 @@ impl ActiveSession { crate::log_msg("Leaving room..."); let _ = self.room_state.leave().await; + // Close peer links with the graceful goodbye code so remotes evict us + // promptly (not after the reconnect grace / slow gossip Leave), and stop + // our supervisors so none redial the about-to-close endpoint. + self.transport.leave().await; crate::log_msg("Room left"); crate::log_msg("Shutting down router..."); @@ -529,19 +534,30 @@ async fn run_core_loop( }; let ui_tx_conn = ui_tx.clone(); let grace_timers_conn = grace_timers.clone(); + let transport_conn = transport.clone(); + let jitter_conn = jitter.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 }, + match event { + ConnEvent::Connecting(id) => { + let _ = ui_tx_conn.send(UiEvent::PeerConnecting { id }).await; + } ConnEvent::Connected(id) => { // The audio link came back — the peer recovered // within the grace window, so cancel its eviction. cancel_grace_timer(&grace_timers_conn, &id); - UiEvent::PeerConnected { id } + let _ = ui_tx_conn.send(UiEvent::PeerConnected { id }).await; + } + ConnEvent::Left(id) => { + // The peer closed its link gracefully (intentional + // leave) — evict immediately, like a PeerLeft, + // instead of leaving it "reconnecting" until the + // grace timer or the slow gossip Leave. + cancel_grace_timer(&grace_timers_conn, &id); + transport_conn.disconnect_peer(id).await; + jitter_conn.lock().await.remove(&id); + let _ = ui_tx_conn.send(UiEvent::PeerLeft { id }).await; } - }; - if ui_tx_conn.send(msg).await.is_err() { - break; } } }); @@ -556,6 +572,7 @@ async fn run_core_loop( event_task, conn_event_task, grace_timers, + transport: transport.clone(), }; let self_id = endpoint.id().to_string(); diff --git a/src/network/iroh_impl.rs b/src/network/iroh_impl.rs index 35e8572..470d120 100644 --- a/src/network/iroh_impl.rs +++ b/src/network/iroh_impl.rs @@ -1,6 +1,6 @@ use crate::network::{NetworkTransport, NetError, ConnEvent}; use iroh::{Endpoint, EndpointId}; -use iroh::endpoint::Connection; +use iroh::endpoint::{Connection, ConnectionError, VarInt}; use bytes::Bytes; use tokio::sync::mpsc; use tokio::sync::mpsc::Receiver; @@ -25,6 +25,12 @@ const INBOUND_QUEUE_DEPTH: usize = 4; const INITIAL_BACKOFF: Duration = Duration::from_millis(250); const MAX_BACKOFF: Duration = Duration::from_secs(5); +/// QUIC application error code sent when we leave/quit intentionally, so the +/// remote supervisor can tell a graceful departure from a transient drop. The +/// specific value is informational; any *application* close (vs a timeout/reset) +/// already means "the peer closed this on purpose." +const GOODBYE_CODE: u32 = 1; + /// 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 { @@ -44,6 +50,12 @@ struct Shared { addrs: StdMutex>, /// One supervisor task per known peer. peers: tokio::sync::Mutex>, + /// The currently-live connection per peer, tracked so an intentional leave + /// can explicitly close each one with [`GOODBYE_CODE`] (a prompt, reliable + /// "I'm leaving" signal the remote can distinguish from a network drop). A + /// supervisor inserts its connection when the link comes up and removes it + /// when the link dies. + live_conns: StdMutex>, incoming_tx: mpsc::Sender<(EndpointId, Bytes)>, /// Best-effort link-state notifications for the UI (connecting / connected). conn_events_tx: mpsc::Sender, @@ -104,14 +116,25 @@ impl Shared { /// Why a peer's live-link wait woke up. enum Wake { - /// The QUIC connection closed (idle timeout or error) — need to reconnect. - Closed, + /// The QUIC connection closed; the error says how. An *application* close + /// means the peer left on purpose (reconnect would be wrong); a timeout/reset + /// is a transient drop we should reconnect through. + Closed(ConnectionError), /// A fresh inbound connection arrived (acceptor side) — switch to it. Replacement(Connection), /// The peer was retired; tear down. Shutdown, } +/// True when the peer closed the connection with our [`GOODBYE_CODE`] — i.e. an +/// intentional leave/quit via [`IrohTransport::leave`]. Any other close (idle +/// timeout, reset, or an application close with a different code) is treated as a +/// transient drop to reconnect through, so this never mistakes a blip for a +/// departure. +fn is_graceful_leave(err: &ConnectionError) -> bool { + matches!(err, ConnectionError::ApplicationClosed(frame) if frame.error_code == VarInt::from_u32(GOODBYE_CODE)) +} + /// 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 @@ -139,6 +162,9 @@ async fn supervise( let (send_tx, mut send_rx) = mpsc::channel::(SEND_QUEUE_DEPTH); shared.senders.lock().unwrap().insert(peer_id, send_tx); + // Publish the live connection so an intentional leave can close it with + // the goodbye code. + shared.live_conns.lock().unwrap().insert(peer_id, conn.clone()); let _ = shared.conn_events_tx.try_send(ConnEvent::Connected(peer_id)); crate::log_msg(&format!("Transport: peer {:?} link up", peer_id)); @@ -170,7 +196,7 @@ async fn supervise( // 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, + reason = conn.closed() => Wake::Closed(reason), inbound = inbound_rx.recv(), if !is_dialer => match inbound { Some(new_conn) => Wake::Replacement(new_conn), None => Wake::Shutdown, @@ -179,8 +205,10 @@ async fn supervise( }; // Dropping the send handle makes `broadcast` skip this peer until the - // link is back, instead of piling frames into a dead connection. + // link is back, instead of piling frames into a dead connection. The + // live connection is gone too until a new one is established. shared.senders.lock().unwrap().remove(&peer_id); + shared.live_conns.lock().unwrap().remove(&peer_id); match wake { Wake::Shutdown => return, @@ -189,7 +217,14 @@ async fn supervise( let _ = shared.conn_events_tx.try_send(ConnEvent::Connecting(peer_id)); conn = new_conn; } - Wake::Closed => { + Wake::Closed(reason) => { + // A graceful application close means the peer left on purpose — + // don't reconnect; tell the core to evict it now. + if is_graceful_leave(&reason) { + crate::log_msg(&format!("Transport: peer {:?} left gracefully ({:?})", peer_id, reason)); + let _ = shared.conn_events_tx.try_send(ConnEvent::Left(peer_id)); + return; + } 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 { @@ -297,6 +332,7 @@ impl IrohTransport { senders: StdMutex::new(HashMap::new()), addrs: StdMutex::new(HashMap::new()), peers: tokio::sync::Mutex::new(HashMap::new()), + live_conns: StdMutex::new(HashMap::new()), incoming_tx, conn_events_tx, }); @@ -310,6 +346,29 @@ impl IrohTransport { (transport, protocol) } + + /// Tear down for an intentional leave/quit: close every live connection with + /// the graceful goodbye code so each remote peer evicts us promptly (its + /// supervisor sees an application close, not a reconnectable drop), then stop + /// all supervisors so none linger redialing the about-to-close endpoint. + /// Call this before shutting the router down. + pub async fn leave(&self) { + let conns: Vec = self.shared.live_conns.lock().unwrap().drain().map(|(_, c)| c).collect(); + for conn in &conns { + conn.close(VarInt::from_u32(GOODBYE_CODE), b"leave"); + } + // Stop every supervisor so it can't re-dial during/after teardown. + let mut peers = self.shared.peers.lock().await; + for (_, handle) in peers.drain() { + handle.supervisor.abort(); + } + self.shared.senders.lock().unwrap().clear(); + self.shared.addrs.lock().unwrap().clear(); + // Give the CONNECTION_CLOSE frames a moment to flush before the caller + // shuts the endpoint/router down (the `conns` clones are still alive + // here, so the endpoint can still transmit them). + tokio::time::sleep(Duration::from_millis(150)).await; + } } #[async_trait] diff --git a/src/network/mod.rs b/src/network/mod.rs index fd7afe9..80ee74a 100644 --- a/src/network/mod.rs +++ b/src/network/mod.rs @@ -54,6 +54,12 @@ pub enum ConnEvent { Connecting(EndpointId), /// A live audio link is established and carrying datagrams. Connected(EndpointId), + /// The peer closed its link *gracefully* (an explicit QUIC application close, + /// which only happens on an intentional leave/quit — a network drop yields a + /// timeout, not this). Distinct from a transient drop so the core can evict + /// the peer immediately instead of waiting out the reconnect grace window. + /// This is the prompt, reliable leave signal; the gossip `Leave` is too slow. + Left(EndpointId), } #[derive(Serialize, Deserialize, Clone, Debug)] diff --git a/tests/transport_loopback.rs b/tests/transport_loopback.rs index 251de9f..a3a9036 100644 --- a/tests/transport_loopback.rs +++ b/tests/transport_loopback.rs @@ -148,6 +148,7 @@ async fn await_reconnect( match tokio::time::timeout_at(deadline, rx.recv()).await { Ok(Some(ConnEvent::Connecting(_))) => saw_connecting = true, Ok(Some(ConnEvent::Connected(_))) => return saw_connecting, + Ok(Some(ConnEvent::Left(_))) => panic!("unexpected graceful Left during reconnect"), Ok(None) => panic!("conn events channel closed unexpectedly"), Err(_) => panic!("timed out waiting for Connected"), } @@ -319,6 +320,67 @@ async fn dialer_reconnects_after_link_drops() { ); } +/// A graceful close (our goodbye code `1`) must report `Left` and NOT reconnect — +/// the opposite of `dialer_reconnects_after_link_drops`, which closes with code 0 +/// (a transient drop). This is what lets a clicked "disconnect" evict the peer +/// promptly instead of leaving it stuck "reconnecting". Mirrors the reconnect +/// test's harness but asserts the supervisor stops rather than re-dials. +#[tokio::test] +async fn dialer_reports_left_on_graceful_close() { + 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; + + 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.endpoint.addr()).await; + + 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" + ); + + // Close with the goodbye code (1) — an intentional leave, not a blip. + conn1.close(1u32.into(), b"leave"); + drop(conn1); + + // The supervisor must surface `Left` (and must NOT try to reconnect: any + // Connecting/Connected here would be the bug). The peer also must not see a + // re-dial. + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + match tokio::time::timeout_at(deadline, conn_events.recv()).await { + Ok(Some(ConnEvent::Left(id))) => { + assert_eq!(id, peer.endpoint.id(), "Left should name the peer"); + } + Ok(Some(ConnEvent::Connecting(_))) | Ok(Some(ConnEvent::Connected(_))) => { + panic!("a graceful close must not trigger a reconnect"); + } + Ok(None) => panic!("conn events channel closed unexpectedly"), + Err(_) => panic!("timed out waiting for Left after graceful close"), + } + + // And no re-dial reaches the peer within a short window. + let redial = tokio::time::timeout(Duration::from_secs(2), peer.conns_rx.recv()).await; + assert!(redial.is_err(), "supervisor must not re-dial after a graceful leave"); +} + #[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