From 5610efac2d6a21990de77a351364e7384b548461 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Mon, 3 Aug 2026 12:14:46 +0200 Subject: [PATCH 1/2] daemon: release the outbound recycle guard on local-input EOF Fixes the Linux defect the diagnostics branch located, and corrects the reasoning I got wrong the first time. The guard was released when the whole attach returned. On Linux that attach can stay up long after the local side is finished: the forced trace measured released=false at 40001ms with the SERVER still reporting srv_active_attaches=1 and srv_detached=0, where macOS detaches the same sequence in under 300ms. So the guard pinned the endpoint for the whole stall and endpoint repair became impossible on exactly the hosts we were about to deploy to. That is worse than the disruption the guard was added to prevent. The guard is now released when the local INPUT side reaches EOF. What I must not claim, and did claim in my first proposal, is that no user remains at that point: a client can half-close its input and go on reading output. So only the recycle guard is released. The session, its buffering and its replay are untouched, and a half-closed reader keeps receiving remote output across the recycle that releasing allowed. The guard now lives on the session rather than in the attach's stack frame, so whichever comes first releases it: local-input EOF, or the attach ending. Nothing is held between attaches, which is what keeps a reconnecting session from blocking the very recycle that might let it reconnect. Proofs, each checked against a control: - A live bidirectional local client still blocks the recycle. - Local-input EOF releases the guard within 500ms. The bound is load-bearing on purpose: the queued remote command sleeps a second before printing, so a release inside that window can only come from local-input EOF. With a generous timeout instead, the test passed even with the fix reverted, because macOS tears down in ~300ms and the two paths were indistinguishable. That is the third time in this work a no-change assertion needed a control to be worth anything. - A half-closed reader receives the delayed output exactly once across the now-allowed recycle: counted, not merely sighted, so a duplicate would fail. - A reconnecting session does not block the recycle; proven by stopping the peer so the session is genuinely between attaches rather than racily so. The underlying Linux teardown latency is NOT addressed here and is being filed separately with the trace. Gates, macOS: shell 12, lib 173, provisioning 10, sync_slice 3, local_slice 21. --- src/daemon.rs | 207 +++++++++++++++++++++++++++++++++++++++++++++++++- src/tunnel.rs | 53 +++++++++++-- 2 files changed, 251 insertions(+), 9 deletions(-) diff --git a/src/daemon.rs b/src/daemon.rs index 5122011..2868e30 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -5098,7 +5098,10 @@ mod tests { } /// Read framed shell output until `marker` appears. - async fn read_shell_marker(stream: &mut UnixStream, marker: &[u8]) -> Result<()> { + async fn read_shell_marker(stream: &mut R, marker: &[u8]) -> Result<()> + where + R: tokio::io::AsyncRead + Unpin, + { let mut seen = Vec::new(); tokio::time::timeout(Duration::from_secs(20), async { loop { @@ -5186,6 +5189,208 @@ mod tests { Ok(()) } + /// A reconnecting session must not block the recycle that could restore it. + /// + /// This is the deadlock direction of the same guard: if a session whose + /// transport is down still pinned the endpoint, then the endpoint could never + /// be rebuilt, and rebuilding it may be exactly what lets that session + /// reconnect. The guard is therefore held per-attach, not per-session. + #[tokio::test] + async fn a_reconnecting_session_does_not_block_the_recycle() -> Result<()> { + let server_dir = tempfile::tempdir()?; + let client_dir = tempfile::tempdir()?; + let server_home = FabricHome::new(server_dir.path()); + let client_home = FabricHome::new(client_dir.path()); + let server = FabricNode::start_with_options(server_home.clone(), true).await?; + let client = FabricNode::start(client_home.clone()).await?; + trust_test_peer(&server_home, &server, client.id(), "client", client.addr()).await?; + trust_test_peer(&client_home, &client, server.id(), "server", server.addr()).await?; + + let state = client.state(); + let socket = state + .dial_alpn( + "server", + shell::SHELL_PROTOCOL, + shell::RESUMABLE_SHELL_ALPN.to_vec(), + false, + ) + .await?; + let mut shell_stream = UnixStream::connect(&socket).await?; + shell::write_client_stdin(&mut shell_stream, b"printf '%s-%s\n' recon up\n").await?; + read_shell_marker(&mut shell_stream, b"recon-up").await?; + assert_eq!(state.client_attaches.attached(), 1); + + // Take the transport away and keep it away, so the session is genuinely + // between attaches rather than momentarily so. Stopping the peer is the + // deterministic way to do that: dropping the tunnel alone lets the client + // reattach within about 100ms and the window would be a race. + server.shutdown().await?; + + // While it is reconnecting, the guard must be down. + let unheld = tokio::time::timeout(Duration::from_secs(20), async { + while state.client_attaches.attached() > 0 { + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await; + assert!( + unheld.is_ok(), + "a session between attaches must not hold the recycle guard" + ); + + // And the recycle it might need is therefore allowed. + let generation = state.endpoint_handle().generation; + let outcome = state + .recycle_endpoint_if_generation(generation, "reconnecting session") + .await?; + assert!( + matches!(outcome, EndpointRecycleOutcome::Recycled), + "a reconnecting session must not block the recycle, got {outcome:?}" + ); + + drop(shell_stream); + client.shutdown().await?; + Ok(()) + } + + /// A half-closed local input releases the recycle guard, and the session + /// still delivers its remaining remote output across the recycle that + /// releasing allowed. + /// + /// This is the correction to a wrong assumption of mine. I first released the + /// guard when the whole attach returned, on the reasoning that a closed local + /// socket meant no user was left to protect. Two things were wrong with that. + /// The attach can stay up for a long time after the local side is done — + /// measured at over 40 seconds on Linux, with the server still reporting the + /// session attached, which pinned the endpoint-recycle guard for that whole + /// window and made endpoint repair impossible. And a client that has finished + /// SENDING may still be READING: a half-close is not a close, so the session + /// must keep working after the guard is released. + #[tokio::test] + async fn half_closed_local_input_releases_guard_and_still_delivers_output() -> Result<()> { + let server_dir = tempfile::tempdir()?; + let client_dir = tempfile::tempdir()?; + let server_home = FabricHome::new(server_dir.path()); + let client_home = FabricHome::new(client_dir.path()); + let server = FabricNode::start_with_options(server_home.clone(), true).await?; + let client = FabricNode::start(client_home.clone()).await?; + trust_test_peer(&server_home, &server, client.id(), "client", client.addr()).await?; + trust_test_peer(&client_home, &client, server.id(), "server", server.addr()).await?; + + let state = client.state(); + let socket = state + .dial_alpn( + "server", + shell::SHELL_PROTOCOL, + shell::RESUMABLE_SHELL_ALPN.to_vec(), + false, + ) + .await?; + let shell_stream = UnixStream::connect(&socket).await?; + let (mut read_half, mut write_half) = shell_stream.into_split(); + + shell::write_client_stdin(&mut write_half, b"printf '%s-%s\n' shell up\n").await?; + read_shell_marker(&mut read_half, b"shell-up").await?; + + // PROOF 1: a live bidirectional local client blocks the recycle. + assert_eq!(state.client_attaches.attached(), 1); + let generation = state.endpoint_handle().generation; + let outcome = state + .recycle_endpoint_if_generation(generation, "half-close: still bidirectional") + .await?; + assert!( + matches!(outcome, EndpointRecycleOutcome::SessionsAttached { .. }), + "a live bidirectional local client must block the recycle, got {outcome:?}" + ); + + // Queue work whose output arrives AFTER the local input is finished, then + // half-close: stop sending, keep reading. + shell::write_client_stdin( + &mut write_half, + b"sleep 1; printf '%s-%s\n' delayed output\n", + ) + .await?; + write_half.shutdown().await?; + drop(write_half); + + // PROOF 2: local input EOF releases the guard, without waiting for the + // remote teardown that on Linux may not have happened at all. + // + // The bound is deliberately far below the remote's completion time. The + // queued command sleeps for a second before it prints, so the session + // cannot possibly have finished tearing down inside this window, and a + // release observed here can only have come from local-input EOF. Without + // that reasoning the check is not load-bearing at all: on macOS the remote + // tears down in about 300ms, so a generous timeout passes either way, which + // is exactly how I first failed to notice this assertion proved nothing. + let release_bound = Duration::from_millis(500); + let release_started = Instant::now(); + let released = tokio::time::timeout(release_bound, async { + while state.client_attaches.attached() > 0 { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await; + assert!( + released.is_ok(), + "local input EOF must release the recycle guard within {release_bound:?} \ + regardless of remote teardown; still held after {:?}", + release_started.elapsed() + ); + + // The recycle that releasing allowed now proceeds. + let outcome = state + .recycle_endpoint_if_generation(generation, "half-close: input finished") + .await?; + assert!( + matches!(outcome, EndpointRecycleOutcome::Recycled), + "with local input finished the recycle must proceed, got {outcome:?}" + ); + + // PROOF 3: the half-closed reader still receives the delayed output, and + // receives it exactly once, across the recycle it just permitted. + let mut seen = Vec::new(); + let read_result = tokio::time::timeout(Duration::from_secs(45), async { + loop { + match shell::read_server_frame(&mut read_half).await { + Ok(Some(shell::ServerFrame::Output(bytes))) => seen.extend_from_slice(&bytes), + Ok(Some(shell::ServerFrame::Exit(_))) | Ok(None) => return Ok(()), + Ok(Some(shell::ServerFrame::Status(_))) => {} + Ok(Some(shell::ServerFrame::Error(error))) => { + return Err(anyhow::anyhow!("shell error: {error}")); + } + Err(error) => return Err(error), + } + if count_occurrences(&seen, b"delayed-output") > 0 { + // Keep draining briefly to catch a duplicate rather than + // returning at the first sighting. + tokio::time::sleep(Duration::from_millis(200)).await; + } + } + }) + .await; + let text = String::from_utf8_lossy(&seen).into_owned(); + let occurrences = count_occurrences(&seen, b"delayed-output"); + assert_eq!( + occurrences, 1, + "delayed output must arrive exactly once across the recycle, saw {occurrences} in {text:?} (read result: {read_result:?})" + ); + + client.shutdown().await?; + server.shutdown().await?; + Ok(()) + } + + fn count_occurrences(haystack: &[u8], needle: &[u8]) -> usize { + if needle.is_empty() || haystack.len() < needle.len() { + return 0; + } + haystack + .windows(needle.len()) + .filter(|w| *w == needle) + .count() + } + /// One peer's success must not clear another peer's failure record on the /// same ALPN. /// diff --git a/src/tunnel.rs b/src/tunnel.rs index f675106..c1ad4da 100644 --- a/src/tunnel.rs +++ b/src/tunnel.rs @@ -108,8 +108,14 @@ impl ClientAttachGauge { } } -/// Holds the gauge up for exactly as long as an attach lives, including when the -/// attach ends by unwinding. +/// Holds the gauge up while an attach is live AND its local input is still open. +/// +/// Dropped by whichever comes first: the attach ending, or the local input +/// reaching EOF. The second case is the one that matters. A remote teardown can +/// stall — measured at over 40 seconds on Linux while the server still reported +/// the session attached — and the endpoint-recycle guard must not be pinned for +/// that whole time by a client that has stopped sending. +#[derive(Debug)] struct ClientAttachGuard(Arc); impl ClientAttachGuard { @@ -283,6 +289,9 @@ struct TunnelState { local_write_closed: bool, pending_remote_close: Option, active_attaches: usize, + /// Live while this session's attach is up and its local input is still open. + /// Released early on local-input EOF; see ClientAttachGuard. + attach_gauge: Option, last_detached: Option, reconnect_attempts: u64, last_error: Option, @@ -399,6 +408,7 @@ impl TunnelSession { local_write_closed: false, pending_remote_close: None, active_attaches: 0, + attach_gauge: None, last_detached: None, reconnect_attempts: 0, last_error: None, @@ -559,10 +569,32 @@ impl TunnelSession { if state.send_closed.is_none() { state.send_closed = Some(state.send_next); } + // The local side will send nothing further, so this session must stop + // holding off an endpoint recycle. It may still be READING queued remote + // output — a half-close is not a close — and that keeps working: only the + // recycle guard is released here, the session and its replay are untouched. + state.attach_gauge = None; drop(state); self.notify.notify_waiters(); } + /// Hold the recycle guard for this attach, unless the local input has already + /// finished, in which case there is nothing left to protect from a recycle. + async fn hold_attach_gauge(&self, gauge: Option>) { + let Some(gauge) = gauge else { + return; + }; + let mut state = self.state.lock().await; + if state.send_closed.is_none() { + state.attach_gauge = Some(ClientAttachGuard::new(gauge)); + } + } + + async fn release_attach_gauge(&self) { + let mut state = self.state.lock().await; + state.attach_gauge = None; + } + async fn apply_peer_ack(&self, recv_next: u64) { let mut state = self.state.lock().await; if recv_next > state.send_acked { @@ -1167,12 +1199,17 @@ async fn attach_connection( notices.emit(&session, ClientConnectionEvent::Resumed).await; } - // Held for the attached period only: the handshake above has succeeded, so - // this session is genuinely on a transport and a recycle would interrupt it. - let _attached = notices - .and_then(|notices| notices.gauge.clone()) - .map(ClientAttachGuard::new); - session.run_attach(send, recv, recv_next).await + // Held for the attached period, and dropped earlier if the local input + // finishes first. Stored on the session so local-input EOF can release it + // without waiting for this attach to return. + session + .hold_attach_gauge(notices.and_then(|notices| notices.gauge.clone())) + .await; + let result = session.clone().run_attach(send, recv, recv_next).await; + // Between attaches nothing is held: a reconnecting session must not block a + // recycle, since a recycle may be exactly what lets it reconnect. + session.release_attach_gauge().await; + result } #[derive(Debug)] From 1edbf6dbaeb154fe9ae3f29c4139a80322c703da Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Mon, 3 Aug 2026 12:23:57 +0200 Subject: [PATCH 2/2] daemon: release the guard on an abrupt local close as well as a clean EOF PR 31's first commit fixed the clean-EOF case and left the Linux CI failure exactly as it was, because the case that fails there is the other one. run_local_reader propagated a read error with ? and returned before reaching mark_send_closed, so an abrupt local close never released the recycle guard. Dropping a local socket yields a clean zero-length read on macOS and an error on Linux, which is why this was invisible locally and why my first fix looked complete while changing nothing about the failing test. Only the guard is released on that path. send_closed and every other teardown semantic stay as they were: an error path is not where I want to be changing session lifecycle, and the underlying Linux teardown latency is filed separately. --- src/tunnel.rs | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/src/tunnel.rs b/src/tunnel.rs index c1ad4da..1bb8081 100644 --- a/src/tunnel.rs +++ b/src/tunnel.rs @@ -530,7 +530,24 @@ impl TunnelSession { let mut buf = [0; LOCAL_READ_BUF]; loop { self.wait_for_buffer_space().await; - let read = read.read(&mut buf).await?; + let read = match read.read(&mut buf).await { + Ok(read) => read, + Err(error) => { + // An abrupt local close ends local input just as surely as a + // clean EOF does: no further bytes can arrive on this session. + // Release the recycle guard so a stalled remote teardown cannot + // pin the endpoint. Deliberately ONLY the guard — send_closed + // and every other teardown semantic stay exactly as they were, + // because this error path is not the place to change them. + // + // This path is why the first version of the fix was incomplete. + // Dropping a local socket yields a clean zero-length read on + // macOS and an error on Linux, so releasing on clean EOF alone + // left the Linux case pinned and the CI failure unchanged. + self.release_attach_gauge().await; + return Err(error.into()); + } + }; if read == 0 { self.mark_send_closed().await; return Ok(());