diff --git a/crates/bemo/src/http/h2.rs b/crates/bemo/src/http/h2.rs index 9dc5294..d0ea0fe 100644 --- a/crates/bemo/src/http/h2.rs +++ b/crates/bemo/src/http/h2.rs @@ -534,8 +534,16 @@ impl Connection { /// Input is closed and accepted streams have drained; no peer PONG can arrive now. pub(crate) fn finish_shutdown(&mut self) { - if let State::Connected(connection) = &mut self.state { - connection.abrupt_shutdown(Reason::NO_ERROR); + match &mut self.state { + State::Connected(connection) => connection.abrupt_shutdown(Reason::NO_ERROR), + State::Handshake(_) => { + // The peer closed input before the HTTP/2 client preface was fully received. The + // server preface already sits in `Wire::output` and stays flushable via + // `take_output_limit`; dropping the handshake future retires the connection so the + // driver's `is_closed() && wire.is_empty() && !http.sending` condition can fire. + self.state = State::Closed; + } + State::Closed => {} } } @@ -1133,6 +1141,49 @@ pub(crate) mod tests { assert!(pair.server.poll(&mut cx, 8).unwrap().is_empty()); } + #[test] + fn finish_shutdown_closes_a_pending_handshake_on_peer_close() { + let mut connection = Connection::new(); + let mut cx = Context::from_waker(Waker::noop()); + // A partial client preface (9 of 24 bytes) leaves the h2 handshake pending. + connection.feed(Bytes::from_static(b"PRI * HTT")).unwrap(); + assert!(connection.poll(&mut cx, 8).unwrap().is_empty()); + assert!(!connection.is_closed()); + // The server preface is written eagerly and stays flushable after shutdown. + assert!(connection.take_output().is_some()); + // Mimic the ABI peer-close path: the driver calls `finish_shutdown` when + // `peer_closed && streams.is_empty()`. Without handling `State::Handshake` + // this is a no-op and the connection leaks forever. + connection.finish_shutdown(); + assert!(connection.is_closed()); + assert!(connection.poll(&mut cx, 8).unwrap().is_empty()); + } + + #[test] + fn finish_shutdown_closes_a_handshake_that_received_no_preface() { + let mut connection = Connection::new(); + let mut cx = Context::from_waker(Waker::noop()); + assert!(connection.poll(&mut cx, 8).unwrap().is_empty()); + assert!(!connection.is_closed()); + connection.finish_shutdown(); + assert!(connection.is_closed()); + assert!(connection.poll(&mut cx, 8).unwrap().is_empty()); + } + + #[test] + fn finish_shutdown_is_idempotent_after_close() { + let mut connection = Connection::new(); + let mut cx = Context::from_waker(Waker::noop()); + connection.feed(Bytes::from_static(b"PRI * HTT")).unwrap(); + connection.poll(&mut cx, 8).unwrap(); + connection.finish_shutdown(); + assert!(connection.is_closed()); + // Calling again must not panic or error. + connection.finish_shutdown(); + assert!(connection.is_closed()); + assert!(connection.poll(&mut cx, 8).unwrap().is_empty()); + } + #[test] fn connection_errors_after_the_handshake_close_the_connection() { let mut pair = Pair::new(65535); diff --git a/crates/bemo/tests/http.rs b/crates/bemo/tests/http.rs index 49d9a54..5c029a3 100644 --- a/crates/bemo/tests/http.rs +++ b/crates/bemo/tests/http.rs @@ -3107,6 +3107,90 @@ fn native_h2_free_on_flow_control_stalled_stream_force_resets_and_closes() { assert_eq!(elide_transport_owner_release(h.owner), 0); } +/// A peer that completes TLS with ALPN `h2` but sends only a partial HTTP/2 client preface +/// (fewer than 24 bytes) and then closes cleanly via `close_notify` must not leak the accepted +/// socket forever. The driver's `finish_shutdown` path transitions the pending h2 handshake to +/// `State::Closed` so `is_closed()` becomes true, the TLS lane closes, and `EVENT_CLOSED` fires. +#[test] +#[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] +fn native_h2_tls_partial_preface_and_close_notify_closes_the_socket() { + use rustls::pki_types::ServerName; + use std::sync::Arc; + let (mut h, config) = tls_fixture_with_protocol(&[&rustls::version::TLS13], b"h2"); + let peer = h.peer.try_clone().unwrap(); + let client = std::thread::spawn(move || { + let client = rustls::ClientConnection::new(Arc::new(config), ServerName::try_from("localhost").unwrap()).unwrap(); + let mut stream = rustls::StreamOwned::new(client, peer); + // Send only 9 of the 24-byte HTTP/2 client preface, then half-close cleanly. + stream.write_all(b"PRI * HTT").unwrap(); + stream.conn.send_close_notify(); + stream.flush().unwrap(); + let mut reply = Vec::new(); + let _ = stream.read_to_end(&mut reply); + }); + // The server must observe EVENT_CLOSED within a bounded number of polls — not leak forever. + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if h.pending.iter().any(|event| event.kind == EVENT_CLOSED) { + break; + } + assert!( + Instant::now() < deadline, + "EVENT_CLOSED never fired; partial-preface socket leaked" + ); + h.poll(200_000_000); + } + h.take(EVENT_CLOSED, 1); + // No request should have been dispatched (the handshake never completed). + assert!(!h.pending.iter().any(|event| event.kind == EVENT_REQUEST)); + client.join().unwrap(); + assert_eq!(elide_transport_driver_release(h.driver), 0); + assert_eq!(elide_transport_buffer_release(h.batch), 0); + assert_eq!(elide_transport_owner_used(h.owner), 0); + assert_eq!(elide_transport_owner_release(h.owner), 0); +} + +/// A peer that completes TLS with ALPN `h2` but sends only a single byte of the HTTP/2 client +/// preface (1 of 24) and then closes cleanly via `close_notify` must not leak the accepted socket +/// forever. This exercises the boundary where the h2 handshake has received input but cannot +/// complete, and the TLS-lane deadline is already disarmed. +#[test] +#[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] +fn native_h2_tls_single_byte_preface_and_close_notify_closes_the_socket() { + use rustls::pki_types::ServerName; + use std::sync::Arc; + let (mut h, config) = tls_fixture_with_protocol(&[&rustls::version::TLS13], b"h2"); + let peer = h.peer.try_clone().unwrap(); + let client = std::thread::spawn(move || { + let client = rustls::ClientConnection::new(Arc::new(config), ServerName::try_from("localhost").unwrap()).unwrap(); + let mut stream = rustls::StreamOwned::new(client, peer); + // Send only the first byte of the 24-byte HTTP/2 client preface, then half-close cleanly. + stream.write_all(b"P").unwrap(); + stream.conn.send_close_notify(); + stream.flush().unwrap(); + let mut reply = Vec::new(); + let _ = stream.read_to_end(&mut reply); + }); + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if h.pending.iter().any(|event| event.kind == EVENT_CLOSED) { + break; + } + assert!( + Instant::now() < deadline, + "EVENT_CLOSED never fired; single-byte-preface socket leaked" + ); + h.poll(200_000_000); + } + h.take(EVENT_CLOSED, 1); + assert!(!h.pending.iter().any(|event| event.kind == EVENT_REQUEST)); + client.join().unwrap(); + assert_eq!(elide_transport_driver_release(h.driver), 0); + assert_eq!(elide_transport_buffer_release(h.batch), 0); + assert_eq!(elide_transport_owner_used(h.owner), 0); + assert_eq!(elide_transport_owner_release(h.owner), 0); +} + #[test] #[cfg_attr(miri, ignore = "io_uring and real sockets are unavailable under miri")] fn parsed_header_metadata_is_advertised_and_preserves_first_host() {