diff --git a/src/daemon.rs b/src/daemon.rs index 5122011..3b25c03 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -5186,6 +5186,68 @@ mod tests { Ok(()) } + /// DIAGNOSTICS BRANCH ONLY. Control for the instrumentation itself. + /// + /// The trace in the test above polls the client gauge and the server session + /// stats in a loop. If that polling were itself ending the session — by + /// touching a lock, or by keeping something alive, or by observing state into + /// existence — the trace would be measuring its own side effects. This holds a + /// live outbound session, runs the same polling for a bounded window, and + /// proves the gauge does NOT move. + #[tokio::test] + async fn diag_instrumentation_does_not_itself_release_the_gauge() -> 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 server_state = server.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' diag control\n").await?; + read_shell_marker(&mut shell_stream, b"diag-control").await?; + assert_eq!(state.client_attaches.attached(), 1); + + // Same polling shape as the instrumented test, same duration order of + // magnitude, with the session deliberately left alive. + let started = Instant::now(); + for _ in 0..12 { + let srv = server_state.tunnel_sessions.stats().await; + eprintln!( + "DIAG control t_ms={} client_gauge={} srv_active_attaches={}", + started.elapsed().as_millis(), + state.client_attaches.attached(), + srv.active_attaches, + ); + assert_eq!( + state.client_attaches.attached(), + 1, + "the instrumentation must not release the gauge on its own" + ); + tokio::time::sleep(Duration::from_millis(250)).await; + } + + // Still the same session afterwards. + shell::write_client_stdin(&mut shell_stream, b"printf '%s-%s\n' diag intact\n").await?; + read_shell_marker(&mut shell_stream, b"diag-intact").await?; + + client.shutdown().await?; + server.shutdown().await?; + Ok(()) + } + /// One peer's success must not clear another peer's failure record on the /// same ALPN. /// @@ -5697,16 +5759,69 @@ mod tests { // NEGATIVE CONTROL. Drop the local end so the outbound session goes away, // then prove the endpoint is recyclable again. + // + // DIAGNOSTICS BRANCH ONLY. This releases in ~53ms on macOS and did not + // release within 20s on Linux CI, so the series below locates which link + // of the teardown chain stalls: local EOF, client send-close, remote PTY + // child exit, server close, client observing remote-close, attach return, + // gauge decrement. The server-side session stats are the observable for + // the middle of that chain; the client gauge is the observable for its end. + let server_state = server.state(); + async fn diag( + label: &str, + at: Duration, + client: &Arc, + server: &Arc, + ) { + let srv = server.tunnel_sessions.stats().await; + eprintln!( + "DIAG {label} t_ms={} client_gauge={} srv_total={} srv_active_sessions={} srv_active_attaches={} srv_detached={} srv_complete={} srv_done={} srv_buffered={}", + at.as_millis(), + client.client_attaches.attached(), + srv.total_sessions, + srv.active_sessions, + srv.active_attaches, + srv.detached_sessions, + srv.complete_sessions, + srv.done_sessions, + srv.buffered_bytes, + ); + } + diag("before_drop", Duration::ZERO, &state, &server_state).await; drop(shell_stream); - let released = tokio::time::timeout(Duration::from_secs(20), async { - while state.client_attaches.attached() > 0 { - tokio::time::sleep(Duration::from_millis(50)).await; + let release_started = Instant::now(); + // Observe for far longer than the 20s that failed, so a slow-but-working + // release is distinguishable from a permanent stall. + let released = tokio::time::timeout(Duration::from_secs(40), async { + loop { + if state.client_attaches.attached() == 0 { + return; + } + diag("waiting", release_started.elapsed(), &state, &server_state).await; + tokio::time::sleep(Duration::from_millis(250)).await; } }) .await; - assert!( + diag( + "after_wait", + release_started.elapsed(), + &state, + &server_state, + ) + .await; + let srv_after = server_state.tunnel_sessions.stats().await; + // Deliberate failure so the numbers reach the CI log: cargo only shows a + // test's captured output when it fails, and the previous Linux run passed, + // which hid the very timing this branch exists to measure. + panic!( + "DIAG-FORCED released={} elapsed_ms={} client_gauge={} srv_active_attaches={} srv_detached={} srv_complete={} srv_done={}", released.is_ok(), - "the gauge must release when an outbound session ends, or a recycle can never run again" + release_started.elapsed().as_millis(), + state.client_attaches.attached(), + srv_after.active_attaches, + srv_after.detached_sessions, + srv_after.complete_sessions, + srv_after.done_sessions, ); let outcome = state .recycle_endpoint_if_generation(generation, "audit: no outbound session")