fix: signal graceful leave at the transport so a disconnect doesn't show "reconnecting"

Field-testing the reconnect fix surfaced a follow-on bug: clicking Disconnect
(or quitting) showed the peer as yellow "Reconnecting" on the other end instead
of leaving cleanly. Logs showed the remote saw NeighborDown first and the gossip
`Leave` arrived ~15s later (broadcast then router torn down ~450ms after, so
NeighborDown — now meaning "reconnecting" — beats the slow, swarm-routed Leave).

Fix: make graceful leave a prompt, reliable TRANSPORT signal instead of relying
on gossip. On leave/quit, IrohTransport::leave() explicitly closes each live
connection with a distinguished goodbye code (GOODBYE_CODE) and aborts all
supervisors (so none linger redialing the about-to-close endpoint). The remote's
supervisor inspects the close reason: an application close with our goodbye code
=> emit the new ConnEvent::Left (evict now); a timeout/reset/other code =>
transient drop, reconnect as before. core handles ConnEvent::Left exactly like a
PeerLeft (cancel grace timer, disconnect, drop jitter, UI remove). ActiveSession
now holds the transport and calls leave() before shutting the router down.

Tracks the live connection per peer in Shared.live_conns (inserted on link up,
removed on drop) so leave() can close them. The gossip Leave path stays as a
harmless backup.

New test dialer_reports_left_on_graceful_close: a close with the goodbye code
reports Left and does NOT reconnect (mirror of the code-0 transient-drop test).
clippy clean, 4 transport tests pass. NOT yet field-verified.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-01 00:08:33 -04:00
co-authored by Claude Opus 4.8
parent be818467dd
commit ffe5b43d8d
4 changed files with 156 additions and 12 deletions
+23 -6
View File
@@ -75,6 +75,7 @@ struct ActiveSession {
event_task: tokio::task::JoinHandle<()>, event_task: tokio::task::JoinHandle<()>,
conn_event_task: tokio::task::JoinHandle<()>, conn_event_task: tokio::task::JoinHandle<()>,
grace_timers: GraceTimers, grace_timers: GraceTimers,
transport: Arc<IrohTransport>,
} }
impl ActiveSession { impl ActiveSession {
@@ -100,6 +101,10 @@ impl ActiveSession {
crate::log_msg("Leaving room..."); crate::log_msg("Leaving room...");
let _ = self.room_state.leave().await; 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("Room left");
crate::log_msg("Shutting down router..."); crate::log_msg("Shutting down router...");
@@ -529,19 +534,30 @@ async fn run_core_loop(
}; };
let ui_tx_conn = ui_tx.clone(); let ui_tx_conn = ui_tx.clone();
let grace_timers_conn = grace_timers.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 { let conn_event_task = tokio::spawn(async move {
while let Some(event) = conn_events.recv().await { while let Some(event) = conn_events.recv().await {
let msg = match event { match event {
ConnEvent::Connecting(id) => UiEvent::PeerConnecting { id }, ConnEvent::Connecting(id) => {
let _ = ui_tx_conn.send(UiEvent::PeerConnecting { id }).await;
}
ConnEvent::Connected(id) => { ConnEvent::Connected(id) => {
// The audio link came back — the peer recovered // The audio link came back — the peer recovered
// within the grace window, so cancel its eviction. // within the grace window, so cancel its eviction.
cancel_grace_timer(&grace_timers_conn, &id); 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, event_task,
conn_event_task, conn_event_task,
grace_timers, grace_timers,
transport: transport.clone(),
}; };
let self_id = endpoint.id().to_string(); let self_id = endpoint.id().to_string();
+65 -6
View File
@@ -1,6 +1,6 @@
use crate::network::{NetworkTransport, NetError, ConnEvent}; use crate::network::{NetworkTransport, NetError, ConnEvent};
use iroh::{Endpoint, EndpointId}; use iroh::{Endpoint, EndpointId};
use iroh::endpoint::Connection; use iroh::endpoint::{Connection, ConnectionError, VarInt};
use bytes::Bytes; use bytes::Bytes;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use tokio::sync::mpsc::Receiver; use tokio::sync::mpsc::Receiver;
@@ -25,6 +25,12 @@ const INBOUND_QUEUE_DEPTH: usize = 4;
const INITIAL_BACKOFF: Duration = Duration::from_millis(250); const INITIAL_BACKOFF: Duration = Duration::from_millis(250);
const MAX_BACKOFF: Duration = Duration::from_secs(5); 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 /// State shared between the transport, its protocol handler, and every per-peer
/// supervisor task. One supervisor owns a peer's whole connection lifecycle. /// supervisor task. One supervisor owns a peer's whole connection lifecycle.
struct Shared { struct Shared {
@@ -44,6 +50,12 @@ struct Shared {
addrs: StdMutex<HashMap<EndpointId, iroh::EndpointAddr>>, addrs: StdMutex<HashMap<EndpointId, iroh::EndpointAddr>>,
/// One supervisor task per known peer. /// One supervisor task per known peer.
peers: tokio::sync::Mutex<HashMap<EndpointId, PeerHandle>>, peers: tokio::sync::Mutex<HashMap<EndpointId, PeerHandle>>,
/// 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<HashMap<EndpointId, Connection>>,
incoming_tx: mpsc::Sender<(EndpointId, Bytes)>, incoming_tx: mpsc::Sender<(EndpointId, Bytes)>,
/// Best-effort link-state notifications for the UI (connecting / connected). /// Best-effort link-state notifications for the UI (connecting / connected).
conn_events_tx: mpsc::Sender<ConnEvent>, conn_events_tx: mpsc::Sender<ConnEvent>,
@@ -104,14 +116,25 @@ impl Shared {
/// Why a peer's live-link wait woke up. /// Why a peer's live-link wait woke up.
enum Wake { enum Wake {
/// The QUIC connection closed (idle timeout or error) — need to reconnect. /// The QUIC connection closed; the error says how. An *application* close
Closed, /// 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. /// A fresh inbound connection arrived (acceptor side) — switch to it.
Replacement(Connection), Replacement(Connection),
/// The peer was retired; tear down. /// The peer was retired; tear down.
Shutdown, 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 /// 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 — /// 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 /// 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::<Bytes>(SEND_QUEUE_DEPTH); let (send_tx, mut send_rx) = mpsc::channel::<Bytes>(SEND_QUEUE_DEPTH);
shared.senders.lock().unwrap().insert(peer_id, send_tx); 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)); let _ = shared.conn_events_tx.try_send(ConnEvent::Connected(peer_id));
crate::log_msg(&format!("Transport: peer {:?} link up", 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 drop and re-dials before our own `closed()` fires, we switch to
// the new link immediately instead of waiting out the idle timeout. // the new link immediately instead of waiting out the idle timeout.
tokio::select! { tokio::select! {
_ = conn.closed() => Wake::Closed, reason = conn.closed() => Wake::Closed(reason),
inbound = inbound_rx.recv(), if !is_dialer => match inbound { inbound = inbound_rx.recv(), if !is_dialer => match inbound {
Some(new_conn) => Wake::Replacement(new_conn), Some(new_conn) => Wake::Replacement(new_conn),
None => Wake::Shutdown, None => Wake::Shutdown,
@@ -179,8 +205,10 @@ async fn supervise(
}; };
// Dropping the send handle makes `broadcast` skip this peer until the // 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.senders.lock().unwrap().remove(&peer_id);
shared.live_conns.lock().unwrap().remove(&peer_id);
match wake { match wake {
Wake::Shutdown => return, Wake::Shutdown => return,
@@ -189,7 +217,14 @@ async fn supervise(
let _ = shared.conn_events_tx.try_send(ConnEvent::Connecting(peer_id)); let _ = shared.conn_events_tx.try_send(ConnEvent::Connecting(peer_id));
conn = new_conn; 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)); crate::log_msg(&format!("Transport: peer {:?} link dropped; reconnecting", peer_id));
let _ = shared.conn_events_tx.try_send(ConnEvent::Connecting(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 { 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()), senders: StdMutex::new(HashMap::new()),
addrs: StdMutex::new(HashMap::new()), addrs: StdMutex::new(HashMap::new()),
peers: tokio::sync::Mutex::new(HashMap::new()), peers: tokio::sync::Mutex::new(HashMap::new()),
live_conns: StdMutex::new(HashMap::new()),
incoming_tx, incoming_tx,
conn_events_tx, conn_events_tx,
}); });
@@ -310,6 +346,29 @@ impl IrohTransport {
(transport, protocol) (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<Connection> = 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] #[async_trait]
+6
View File
@@ -54,6 +54,12 @@ pub enum ConnEvent {
Connecting(EndpointId), Connecting(EndpointId),
/// A live audio link is established and carrying datagrams. /// A live audio link is established and carrying datagrams.
Connected(EndpointId), 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)] #[derive(Serialize, Deserialize, Clone, Debug)]
+62
View File
@@ -148,6 +148,7 @@ async fn await_reconnect(
match tokio::time::timeout_at(deadline, rx.recv()).await { match tokio::time::timeout_at(deadline, rx.recv()).await {
Ok(Some(ConnEvent::Connecting(_))) => saw_connecting = true, Ok(Some(ConnEvent::Connecting(_))) => saw_connecting = true,
Ok(Some(ConnEvent::Connected(_))) => return saw_connecting, 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"), Ok(None) => panic!("conn events channel closed unexpectedly"),
Err(_) => panic!("timed out waiting for Connected"), 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] #[tokio::test]
async fn dialer_connects_and_reconnects_without_an_address_lookup() { async fn dialer_connects_and_reconnects_without_an_address_lookup() {
// Regression for the redial-loop bug: a transient drop that also triggers a // Regression for the redial-loop bug: a transient drop that also triggers a