feat(http2): tell a reset upgraded stream from a half-close

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

src/proto/h2/client.rs

@@ -76,6 +76,7 @@ pub(crate) struct Config {
     pub(crate) max_pending_accept_reset_streams: Option<usize>,
     pub(crate) max_local_error_reset_streams: Option<usize>,
     pub(crate) header_table_size: Option<u32>,
+    pub(crate) upgrade_reset_is_error: bool,
     pub(crate) max_concurrent_streams: Option<u32>,
     pub(crate) reset_stream_duration: Option<Duration>,
 }
@@ -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<SendBuf<B::Data>>,
     req_rx: ClientRx<B>,
     fut_ctx: Option<FutCtx<B>>,
+    /// 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<T>,
 }
 
@@ -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<Option<SendStream<SendBuf<<B as Body>::Data>>>>,
         exec: E,
         cancel_tx: Option<oneshot::Sender<()>>,
+        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());
 

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<u32>,
     pub(crate) max_pending_accept_reset_streams: Option<usize>,
     pub(crate) max_local_error_reset_streams: Option<usize>,
@@ -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<T, B>,
         date_header: bool,
+        upgrade_reset_is_error: bool,
         close_pending: bool
     }
 }
@@ -115,6 +118,9 @@ where
     conn: Connection<Compat<T>, SendBuf<B::Data>>,
     closing: Option<crate::Error>,
     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<T, S, B, E> Server<T, S, B, E>
@@ -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<F, B, E> H2Stream<F, B, E>
@@ -502,6 +512,7 @@ where
                                 send_stream,
                                 connect_parts.recv_stream,
                                 connect_parts.ping,
+                                connect_parts.upgrade_reset_is_error,
                             );
                             connect_parts
                                 .pending

src/proto/h2/upgrade.rs

@@ -17,6 +17,7 @@ pub(super) fn pair<B>(
     send_stream: SendStream<SendBuf<B>>,
     recv_stream: RecvStream,
     ping: Recorder,
+    upgrade_reset_is_error: bool,
 ) -> (H2Upgraded, UpgradedSendStreamTask<B>) {
     let (tx, rx) = mpsc::channel(1);
     let (error_tx, error_rx) = oneshot::channel();
@@ -27,6 +28,7 @@ pub(super) fn pair<B>(
             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<Result<(), std::io::Error>> {
+        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))
                             }

src/server/conn/http2.rs

@@ -260,6 +260,22 @@ impl<E> Builder<E> {
         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

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<usize> {
+    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<IncomingBody>| {
+        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::<Bytes>::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();