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..1bb8081 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, @@ -520,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(()); @@ -559,10 +586,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 +1216,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)]