fix(h2): close pending handshake on peer close_notify to prevent socket leak

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);

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() {