From 5c8bacd7fb4c785c2c3a426ae7a3705a7ba838b6 Mon Sep 17 00:00:00 2001 From: Robert Ream <154010+robertream@users.noreply.github.com> Date: Sat, 31 Oct 2026 09:31:31 -0700 Subject: [PATCH] feat(http2): tell a reset upgraded stream from a half-close An `Upgraded` stream over HTTP/2 is a CONNECT stream, and `RST_STREAM` is the only way its peer can end one abruptly. Every reset read as an end of stream, so an application could not tell an abandoned session from a peer that had finished. Nothing else on the stream carries the distinction: the write half reports the same error a poll later for a clean half-close as for a reset. `Builder::upgrade_reset_is_error` opts in to reading any reset but `NO_ERROR` as `ConnectionReset`, on both ends. Off by default, because reading `CANCEL` as a close is what RFC 8441 relies on, where a reset is the connection going away under a protocol that signals its own close in-band. A protocol whose session is the CONNECT stream itself needs the opposite. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_013CWmskbkrYMCKcuYaufajw --- src/client/conn/http2.rs | 16 +++++++++ src/proto/h2/client.rs | 10 +++++- src/proto/h2/server.rs | 11 ++++++ src/proto/h2/upgrade.rs | 15 +++++++- src/server/conn/http2.rs | 16 +++++++++ tests/server.rs | 74 ++++++++++++++++++++++++++++++++++++++++ 6 files changed, 140 insertions(+), 2 deletions(-) diff --git a/src/client/conn/http2.rs b/src/client/conn/http2.rs index 4df19a10cc..25a72f7ea4 100644 --- a/src/client/conn/http2.rs +++ b/src/client/conn/http2.rs @@ -401,6 +401,22 @@ where self } + /// Reports a reset of an upgraded HTTP/2 stream as an error rather than as + /// an end of stream. + /// + /// An upgraded stream is a CONNECT stream, and by default a reset carrying + /// `NO_ERROR` or `CANCEL` reads as a clean end of stream, which is what + /// RFC 8441 relies on. With this set, every reset but `NO_ERROR` reads as + /// [`ConnectionReset`](std::io::ErrorKind::ConnectionReset), so a protocol + /// whose session is the CONNECT stream itself can tell an abandoned session + /// from a peer that finished. + /// + /// Default is `false`. + pub fn upgrade_reset_is_error(&mut self, enabled: bool) -> &mut Self { + self.h2_builder.upgrade_reset_is_error = enabled; + self + } + /// Sets the header table size. /// /// This setting informs the peer of the maximum size of the header compression diff --git a/src/proto/h2/client.rs b/src/proto/h2/client.rs index eb768baa76..39fe661513 100644 --- a/src/proto/h2/client.rs +++ b/src/proto/h2/client.rs @@ -76,6 +76,7 @@ pub(crate) struct Config { pub(crate) max_pending_accept_reset_streams: Option, pub(crate) max_local_error_reset_streams: Option, pub(crate) header_table_size: Option, + pub(crate) upgrade_reset_is_error: bool, pub(crate) max_concurrent_streams: Option, pub(crate) reset_stream_duration: Option, } @@ -97,6 +98,7 @@ impl Default for Config { max_pending_accept_reset_streams: None, max_local_error_reset_streams: Some(1024), header_table_size: None, + upgrade_reset_is_error: false, max_concurrent_streams: None, reset_stream_duration: None, } @@ -201,6 +203,7 @@ where h2_tx, req_rx, fut_ctx: None, + upgrade_reset_is_error: config.upgrade_reset_is_error, marker: PhantomData, }) } @@ -434,6 +437,9 @@ where h2_tx: SendRequest>, req_rx: ClientRx, fut_ctx: Option>, + /// Carried as far as the upgraded stream, where it decides whether a + /// reset reads as an error or as an end of stream. + upgrade_reset_is_error: bool, marker: PhantomData, } @@ -571,6 +577,7 @@ where send_stream: Some(send_stream), exec: self.executor.clone(), cancel_tx: Some(cancel_tx), + upgrade_reset_is_error: self.upgrade_reset_is_error, }, call_back: Some(f.cb), }, @@ -591,6 +598,7 @@ pin_project! { send_stream: Option::Data>>>>, exec: E, cancel_tx: Option>, + upgrade_reset_is_error: bool, } } @@ -639,7 +647,7 @@ where let (pending, on_upgrade) = crate::upgrade::pending(); - let (h2_up, up_task) = super::upgrade::pair(send_stream, recv_stream, ping); + let (h2_up, up_task) = super::upgrade::pair(send_stream, recv_stream, ping, *this.upgrade_reset_is_error); self.exec.execute_upgrade(up_task); let upgraded = Upgraded::new(h2_up, Bytes::new()); diff --git a/src/proto/h2/server.rs b/src/proto/h2/server.rs index 2098758d9e..103b4ef4ff 100644 --- a/src/proto/h2/server.rs +++ b/src/proto/h2/server.rs @@ -47,6 +47,7 @@ pub(crate) struct Config { pub(crate) initial_stream_window_size: u32, pub(crate) max_frame_size: u32, pub(crate) enable_connect_protocol: bool, + pub(crate) upgrade_reset_is_error: bool, pub(crate) max_concurrent_streams: Option, pub(crate) max_pending_accept_reset_streams: Option, pub(crate) max_local_error_reset_streams: Option, @@ -66,6 +67,7 @@ impl Default for Config { initial_stream_window_size: DEFAULT_STREAM_WINDOW, max_frame_size: DEFAULT_MAX_FRAME_SIZE, enable_connect_protocol: false, + upgrade_reset_is_error: false, max_concurrent_streams: Some(200), max_pending_accept_reset_streams: None, max_local_error_reset_streams: Some(DEFAULT_MAX_LOCAL_ERROR_RESET_STREAMS), @@ -90,6 +92,7 @@ pin_project! { service: S, state: State, date_header: bool, + upgrade_reset_is_error: bool, close_pending: bool } } @@ -115,6 +118,9 @@ where conn: Connection, SendBuf>, closing: Option, date_header: bool, + /// Carried as far as the upgraded stream, where it decides whether a + /// reset reads as an error or as an end of stream. + upgrade_reset_is_error: bool, } impl Server @@ -178,6 +184,7 @@ where }, service, date_header: config.date_header, + upgrade_reset_is_error: config.upgrade_reset_is_error, close_pending: false, } } @@ -224,6 +231,7 @@ where conn, closing: None, date_header: me.date_header, + upgrade_reset_is_error: me.upgrade_reset_is_error, }) } State::Serving(srv) => { @@ -297,6 +305,7 @@ where pending, ping, recv_stream: stream, + upgrade_reset_is_error: self.upgrade_reset_is_error, }), ) }; @@ -394,6 +403,7 @@ struct ConnectParts { pending: Pending, ping: Recorder, recv_stream: RecvStream, + upgrade_reset_is_error: bool, } impl H2Stream @@ -502,6 +512,7 @@ where send_stream, connect_parts.recv_stream, connect_parts.ping, + connect_parts.upgrade_reset_is_error, ); connect_parts .pending diff --git a/src/proto/h2/upgrade.rs b/src/proto/h2/upgrade.rs index def1fcfc96..972d49c01e 100644 --- a/src/proto/h2/upgrade.rs +++ b/src/proto/h2/upgrade.rs @@ -17,6 +17,7 @@ pub(super) fn pair( send_stream: SendStream>, recv_stream: RecvStream, ping: Recorder, + upgrade_reset_is_error: bool, ) -> (H2Upgraded, UpgradedSendStreamTask) { let (tx, rx) = mpsc::channel(1); let (error_tx, error_rx) = oneshot::channel(); @@ -27,6 +28,7 @@ pub(super) fn pair( recv_stream, ping, buf: Bytes::new(), + upgrade_reset_is_error, }, UpgradedSendStreamTask { h2_tx: send_stream, @@ -42,6 +44,9 @@ pub(super) struct H2Upgraded { send_stream: UpgradedSendStreamBridge, recv_stream: RecvStream, buf: Bytes, + /// Whether a `RST_STREAM` other than `NO_ERROR` reads as an error rather + /// than as an end of stream. + upgrade_reset_is_error: bool, } struct UpgradedSendStreamBridge { @@ -168,6 +173,7 @@ impl Read for H2Upgraded { cx: &mut Context<'_>, mut read_buf: ReadBufCursor<'_>, ) -> Poll> { + let upgrade_reset_is_error = self.upgrade_reset_is_error; if self.buf.is_empty() { self.buf = loop { match ready!(self.recv_stream.poll_data(cx)) { @@ -179,7 +185,14 @@ impl Read for H2Upgraded { } Some(Err(e)) => { return Poll::Ready(match e.reason() { - Some(Reason::NO_ERROR) | Some(Reason::CANCEL) => Ok(()), + // NO_ERROR is the peer saying it finished, so it + // reads as an end of stream either way. + Some(Reason::NO_ERROR) => Ok(()), + Some(_) if upgrade_reset_is_error => Err(std::io::Error::new( + std::io::ErrorKind::ConnectionReset, + e, + )), + Some(Reason::CANCEL) => Ok(()), Some(Reason::STREAM_CLOSED) => { Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, e)) } diff --git a/src/server/conn/http2.rs b/src/server/conn/http2.rs index e4dac78bea..77585669d2 100644 --- a/src/server/conn/http2.rs +++ b/src/server/conn/http2.rs @@ -260,6 +260,22 @@ impl Builder { self } + /// Reports a reset of an upgraded HTTP/2 stream as an error rather than as + /// an end of stream. + /// + /// An upgraded stream is a CONNECT stream, and by default a reset carrying + /// `NO_ERROR` or `CANCEL` reads as a clean end of stream, which is what + /// RFC 8441 relies on. With this set, every reset but `NO_ERROR` reads as + /// [`ConnectionReset`](std::io::ErrorKind::ConnectionReset), so a protocol + /// whose session is the CONNECT stream itself can tell an abandoned session + /// from a peer that finished. + /// + /// Default is `false`. + pub fn upgrade_reset_is_error(&mut self, enabled: bool) -> &mut Self { + self.h2_builder.upgrade_reset_is_error = enabled; + self + } + /// Sets the header table size. /// /// This setting informs the peer of the maximum size of the header compression diff --git a/tests/server.rs b/tests/server.rs index 094558cb61..5961ab2bde 100644 --- a/tests/server.rs +++ b/tests/server.rs @@ -2588,6 +2588,80 @@ async fn h2_connect_reset_during_backpressure() { client_handle.await.unwrap(); } +/// Drive a CONNECT upgrade, have the peer reset it, and report what a read of +/// the upgraded stream saw. +/// +/// `CANCEL` for both cases on purpose: it is not `NO_ERROR`, so the flag is the +/// only thing that differs between the two tests below. +async fn upgraded_read_after_reset(upgrade_reset_is_error: bool) -> io::Result { + let (listener, addr) = setup_tcp_listener(); + let conn = connect_async(addr).await; + + let (h2, connection) = h2::client::Builder::new() + .handshake::<_, Bytes>(conn) + .await + .unwrap(); + tokio::spawn(async move { + let _ = connection.await; + }); + let mut h2 = h2.ready().await.unwrap(); + + let client = tokio::spawn(async move { + let request = Request::connect("localhost").body(()).unwrap(); + let (response, mut send_stream) = h2.send_request(request, false).unwrap(); + assert_eq!(response.await.unwrap().status(), StatusCode::OK); + send_stream.send_reset(h2::Reason::CANCEL); + }); + + let (read_tx, read_rx) = oneshot::channel(); + let read_tx = Arc::new(Mutex::new(Some(read_tx))); + let svc = service_fn(move |req: Request| { + let on_upgrade = hyper::upgrade::on(req); + let read_tx = read_tx.clone(); + + tokio::spawn(async move { + let mut upgraded = TokioIo::new(on_upgrade.await.expect("on_upgrade")); + let mut buf = [0u8; 1]; + let read = upgraded.read(&mut buf).await; + if let Some(tx) = read_tx.lock().unwrap().take() { + let _ = tx.send(read); + } + }); + + future::ok::<_, hyper::Error>(Response::new(Empty::::new())) + }); + + let (socket, _) = listener.accept().await.unwrap(); + let socket = TokioIo::new(socket); + let mut builder = http2::Builder::new(TokioExecutor); + builder.upgrade_reset_is_error(upgrade_reset_is_error); + tokio::spawn(async move { + let _ = builder.serve_connection(socket, svc).await; + }); + + let read = read_rx.await.expect("the upgraded read to report"); + client.await.unwrap(); + read +} + +#[tokio::test] +async fn h2_upgraded_reset_reads_as_end_of_stream_by_default() { + let read = upgraded_read_after_reset(false) + .await + .expect("a reset reads as an end of stream by default"); + + assert_eq!(read, 0, "RFC 8441 relies on a CANCEL reading as a close"); +} + +#[tokio::test] +async fn h2_upgraded_reset_is_an_error_once_enabled() { + let error = upgraded_read_after_reset(true) + .await + .expect_err("a reset is an error once enabled"); + + assert_eq!(error.kind(), io::ErrorKind::ConnectionReset); +} + #[tokio::test] async fn h2_connect_backpressure_bidirectional() { let (listener, addr) = setup_tcp_listener();