diff --git a/crates/lanspread-peer/ARCHITECTURE.md b/crates/lanspread-peer/ARCHITECTURE.md index 2da5a80..6164b4d 100644 --- a/crates/lanspread-peer/ARCHITECTURE.md +++ b/crates/lanspread-peer/ARCHITECTURE.md @@ -166,6 +166,9 @@ encoded author snapshot at 4 MiB, with reserved capacity for terminal actions. - Cancellation stops admission, drains all lexically owned connection, stream, state-sync, operation, and mDNS children, closes the shared endpoints, and joins the supervisor. +- After application children drain, endpoint tasks stop and join immediately; + exit does not wait for cancelled handshakes or closed connections to exhaust + their transport timers. - There is no `Goodbye` control message. Pinned liveness failure and stale generation-conditional removal are authoritative for departure. diff --git a/crates/lanspread-peer/src/quic_runtime.rs b/crates/lanspread-peer/src/quic_runtime.rs index 716c2e4..6594c60 100644 --- a/crates/lanspread-peer/src/quic_runtime.rs +++ b/crates/lanspread-peer/src/quic_runtime.rs @@ -449,42 +449,13 @@ pub(crate) struct QuicClientRuntime { } impl QuicClientRuntime { - /// Waits for outstanding connections to settle, then always stops and - /// joins the endpoint. A stuck transport can consume the grace period, but - /// cannot make peer shutdown unbounded. - pub(crate) async fn shutdown(mut self) -> eyre::Result<()> { - let idle_result = - match tokio::time::timeout(QUIC_ENDPOINT_SHUTDOWN_GRACE, self.client.wait_idle()).await - { - Ok(result) => result.map_err(eyre::Report::from), - Err(_) => Err(eyre::eyre!( - "QUIC client did not become idle within {QUIC_ENDPOINT_SHUTDOWN_GRACE:?}" - )), - }; - - // No application connector remains after the runtime's child scopes - // have drained. Drop this final handle before forcing endpoint stop. - drop(self.client); - let endpoint_result = self.endpoint.shutdown_and_join().await; - - match (idle_result, endpoint_result) { - (Ok(()), Ok(())) => Ok(()), - (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error), - (Err(idle_error), Err(endpoint_error)) => Err(eyre::eyre!( - "QUIC client shutdown failed: {idle_error:#}; endpoint shutdown also failed: {endpoint_error:#}" - )), - } - } - - /// Settles the test-only hostile-handshake fixture after the verifier has - /// already proved that connection establishment was rejected. + /// Stops and joins the endpoint after its application child scopes drain. /// - /// A failed TLS `CertificateVerify` can remain in s2n-quic's endpoint-owned - /// closing state beyond the normal wait-idle grace. This seam is not a - /// general timeout bypass: it drops the final client handle, then still - /// explicitly stops and joins the endpoint task. - #[cfg(test)] - pub(crate) async fn shutdown_rejected_handshake_fixture(self) -> eyre::Result<()> { + /// Cancelled handshakes and closed connections can remain in transport + /// timers for seconds after their application owners return. Waiting for + /// client idleness here would delay application exit and sharing disable + /// without protecting any unfinished application work. + pub(crate) async fn shutdown(self) -> eyre::Result<()> { drop(self.client); self.endpoint.shutdown_and_join().await } diff --git a/crates/lanspread-peer/src/startup.rs b/crates/lanspread-peer/src/startup.rs index 57f3acd..6fa1c7b 100644 --- a/crates/lanspread-peer/src/startup.rs +++ b/crates/lanspread-peer/src/startup.rs @@ -745,6 +745,64 @@ mod tests { ); } + #[tokio::test] + async fn runtime_shutdown_joins_without_waiting_for_cancelled_handshake_timeout() { + let remote = tokio::net::UdpSocket::bind("127.0.0.1:0") + .await + .expect("silent remote endpoint should bind"); + let identity = PeerIdentity::generate().expect("test responder identity should generate"); + let endpoint = lanspread_proto::PeerEndpoint { + peer_id: identity.peer_id(), + addr: remote + .local_addr() + .expect("remote endpoint should have an address"), + }; + let shutdown = CancellationToken::new(); + let runtime_shutdown = shutdown.clone(); + let (cleanup_tx, cleanup_rx) = tokio::sync::oneshot::channel(); + let mut handle = test_runtime_handle(shutdown, move || { + let (client, connector) = crate::quic_runtime::start_quic_client()?; + Ok(async move { + tokio::select! { + () = runtime_shutdown.cancelled() => {}, + _ = connector.connect(&endpoint) => panic!("silent remote unexpectedly completed dial"), + } + drop(connector); + cleanup_tx + .send(client.shutdown().await.map_err(|error| error.to_string())) + .expect("test should observe client cleanup"); + }) + }); + + // Wait until the real endpoint owns an in-flight handshake before + // requesting shutdown; the remote deliberately never responds. + let mut datagram = [0; 1500]; + let (_, local_addr) = + tokio::time::timeout(Duration::from_secs(1), remote.recv_from(&mut datagram)) + .await + .expect("client should send its Initial promptly") + .expect("silent remote should receive the Initial"); + let started = tokio::time::Instant::now(); + handle.shutdown(); + handle.wait_stopped().await; + assert!( + started.elapsed() < Duration::from_secs(1), + "runtime shutdown should join promptly after cancelling the dial; took {:?}", + started.elapsed(), + ); + cleanup_rx + .await + .expect("client cleanup should complete before its supervisor joins") + .expect("client endpoint should join cleanly after cancelling the dial"); + assert!(handle.supervisor.is_joined()); + + // s2n's adapter RX/TX tasks own socket clones outside its endpoint + // task. The supervisor joins its entire isolated Tokio runtime, so + // these sockets must be gone at the public wait_stopped boundary. + let _rebound = std::net::UdpSocket::bind(local_addr) + .expect("joined peer runtime should release its UDP socket"); + } + #[tokio::test] async fn runtime_handle_retains_exact_identity_arc_and_durability() { let shutdown = CancellationToken::new(); diff --git a/crates/lanspread-peer/src/tls_identity_spike.rs b/crates/lanspread-peer/src/tls_identity_spike.rs index 4b91100..76a09da 100644 --- a/crates/lanspread-peer/src/tls_identity_spike.rs +++ b/crates/lanspread-peer/src/tls_identity_spike.rs @@ -141,15 +141,6 @@ impl TestClient { .shutdown() .await } - - async fn shutdown_rejected_handshake_fixture(mut self) -> eyre::Result<()> { - drop(self.connector.take()); - self.runtime - .take() - .ok_or_else(|| eyre::eyre!("test client runtime was already joined"))? - .shutdown_rejected_handshake_fixture() - .await - } } struct TestPair { @@ -174,12 +165,6 @@ impl TestPair { tokio::join!(self.server.shutdown(), self.client.shutdown()); merge_results(server_result, client_result) } - - async fn shutdown_rejected_handshake_fixture(self) -> eyre::Result<()> { - let server_result = self.server.shutdown().await; - let client_result = self.client.shutdown_rejected_handshake_fixture().await; - merge_results(server_result, client_result) - } } fn merge_results(primary: eyre::Result, cleanup: eyre::Result<()>) -> eyre::Result { @@ -432,7 +417,7 @@ async fn execute_rejected_exchange( let operation = expect_rejected_connection(&mut pair.server, &pair.client, endpoint) .await .and_then(|()| verify_rejection()); - let cleanup = pair.shutdown_rejected_handshake_fixture().await; + let cleanup = pair.shutdown().await; merge_results(operation, cleanup) }