Switch pseudo header order to mimic chrome

Http2 Server are allowed to early respond without fully
  consuming client input stream, but must respond with an
  error code of NO_ERROR when sending RST_STREAM.
  Nginx treat any other error code as fatal if not done so

  Commit change error code from CANCEL to NO_ERROR, when the
  server is early responding to the client

  https://github.com/hyperium/h2/issues/633
  https://trac.nginx.org/nginx/ticket/2376

.github/workflows/CI.yml

@@ -90,7 +90,7 @@ jobs:
     strategy:
       matrix:
         rust:
-          - 1.49 # never go past Hyper's own MSRV
+          - 1.56 # never go past Hyper's own MSRV
 
         os:
           - ubuntu-latest

CHANGELOG.md

@@ -1,3 +1,9 @@
+# 0.3.14 (August 16, 2022)
+
+* Add `Error::is_reset` function.
+* Bump MSRV to Rust 1.56.
+* Return `RST_STREAM(NO_ERROR)` when the server early responds.
+
 # 0.3.13 (March 31, 2022)
 
 * Update private internal `tokio-util` dependency.

Cargo.toml

@@ -5,7 +5,7 @@ name = "h2"
 #   - html_root_url.
 # - Update CHANGELOG.md.
 # - Create git tag
-version = "0.3.13"
+version = "0.3.14"
 license = "MIT"
 authors = [
   "Carl Lerche <me@carllerche.com>",

src/client.rs

@@ -526,7 +526,7 @@ where
     ///
     /// This setting is configured by the server peer by sending the
     /// [`SETTINGS_ENABLE_CONNECT_PROTOCOL` parameter][2] in a `SETTINGS` frame.
-    /// This method returns the currently acknowledged value recieved from the
+    /// This method returns the currently acknowledged value received from the
     /// remote.
     ///
     /// [1]: https://datatracker.ietf.org/doc/html/rfc8441#section-4
@@ -1022,6 +1022,39 @@ impl Builder {
         self
     }
 
+    /// Sets the header table size.
+    ///
+    /// This setting informs the peer of the maximum size of the header compression
+    /// table used to encode header blocks, in octets. The encoder may select any value
+    /// equal to or less than the header table size specified by the sender.
+    ///
+    /// The default value is 4,096.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # use tokio::io::{AsyncRead, AsyncWrite};
+    /// # use h2::client::*;
+    /// # use bytes::Bytes;
+    /// #
+    /// # async fn doc<T: AsyncRead + AsyncWrite + Unpin>(my_io: T)
+    /// # -> Result<((SendRequest<Bytes>, Connection<T, Bytes>)), h2::Error>
+    /// # {
+    /// // `client_fut` is a future representing the completion of the HTTP/2
+    /// // handshake.
+    /// let client_fut = Builder::new()
+    ///     .header_table_size(1_000_000)
+    ///     .handshake(my_io);
+    /// # client_fut.await
+    /// # }
+    /// #
+    /// # pub fn main() {}
+    /// ```
+    pub fn header_table_size(&mut self, size: u32) -> &mut Self {
+        self.settings.set_header_table_size(Some(size));
+        self
+    }
+
     /// Sets the first stream ID to something other than 1.
     #[cfg(feature = "unstable")]
     pub fn initial_stream_id(&mut self, stream_id: u32) -> &mut Self {
@@ -1280,7 +1313,7 @@ where
     ///
     /// This limit is configured by the server peer by sending the
     /// [`SETTINGS_MAX_CONCURRENT_STREAMS` parameter][1] in a `SETTINGS` frame.
-    /// This method returns the currently acknowledged value recieved from the
+    /// This method returns the currently acknowledged value received from the
     /// remote.
     ///
     /// [1]: https://tools.ietf.org/html/rfc7540#section-5.1.2

src/codec/framed_read.rs

@@ -88,6 +88,12 @@ impl<T> FramedRead<T> {
     pub fn set_max_header_list_size(&mut self, val: usize) {
         self.max_header_list_size = val;
     }
+
+    /// Update the header table size setting.
+    #[inline]
+    pub fn set_header_table_size(&mut self, val: usize) {
+        self.hpack.queue_size_update(val);
+    }
 }
 
 /// Decodes a frame.

src/codec/mod.rs

@@ -95,6 +95,11 @@ impl<T, B> Codec<T, B> {
         self.framed_write().set_header_table_size(val)
     }
 
+    /// Set the decoder header table size size.
+    pub fn set_recv_header_table_size(&mut self, val: usize) {
+        self.inner.set_header_table_size(val)
+    }
+
     /// Set the max header list size that can be received.
     pub fn set_max_recv_header_list_size(&mut self, val: usize) {
         self.inner.set_max_header_list_size(val);

src/error.rs

@@ -59,10 +59,7 @@ impl Error {
 
     /// Returns true if the error is an io::Error
     pub fn is_io(&self) -> bool {
-        match self.kind {
-            Kind::Io(_) => true,
-            _ => false,
-        }
+        matches!(self.kind, Kind::Io(..))
     }
 
     /// Returns the error if the error is an io::Error
@@ -92,6 +89,11 @@ impl Error {
         matches!(self.kind, Kind::GoAway(..))
     }
 
+    /// Returns true if the error is from a `RST_STREAM`.
+    pub fn is_reset(&self) -> bool {
+        matches!(self.kind, Kind::Reset(..))
+    }
+
     /// Returns true if the error was received in a frame from the remote.
     ///
     /// Such as from a received `RST_STREAM` or `GOAWAY` frame.

src/frame/headers.rs

@@ -686,13 +686,13 @@ impl Iterator for Iter {
                 return Some(Method(method));
             }
 
-            if let Some(scheme) = pseudo.scheme.take() {
-                return Some(Scheme(scheme));
-            }
-
             if let Some(authority) = pseudo.authority.take() {
                 return Some(Authority(authority));
             }
+            
+            if let Some(scheme) = pseudo.scheme.take() {
+                return Some(Scheme(scheme));
+            }
 
             if let Some(path) = pseudo.path.take() {
                 return Some(Path(path));

src/frame/settings.rs

@@ -121,11 +121,9 @@ impl Settings {
         self.header_table_size
     }
 
-    /*
     pub fn set_header_table_size(&mut self, size: Option<u32>) {
         self.header_table_size = size;
     }
-    */
 
     pub fn load(head: Head, payload: &[u8]) -> Result<Settings, Error> {
         use self::Setting::*;

src/lib.rs

@@ -78,7 +78,7 @@
 //! [`server::handshake`]: server/fn.handshake.html
 //! [`client::handshake`]: client/fn.handshake.html
 
-#![doc(html_root_url = "https://docs.rs/h2/0.3.13")]
+#![doc(html_root_url = "https://docs.rs/h2/0.3.14")]
 #![deny(missing_debug_implementations, missing_docs)]
 #![cfg_attr(test, deny(warnings))]
 
@@ -133,44 +133,3 @@ pub use crate::share::{FlowControl, Ping, PingPong, Pong, RecvStream, SendStream
 
 #[cfg(feature = "unstable")]
 pub use codec::{Codec, SendError, UserError};
-
-use std::task::Poll;
-
-// TODO: Get rid of this trait once https://github.com/rust-lang/rust/pull/63512
-// is stabilized.
-trait PollExt<T, E> {
-    /// Changes the success value of this `Poll` with the closure provided.
-    fn map_ok_<U, F>(self, f: F) -> Poll<Option<Result<U, E>>>
-    where
-        F: FnOnce(T) -> U;
-    /// Changes the error value of this `Poll` with the closure provided.
-    fn map_err_<U, F>(self, f: F) -> Poll<Option<Result<T, U>>>
-    where
-        F: FnOnce(E) -> U;
-}
-
-impl<T, E> PollExt<T, E> for Poll<Option<Result<T, E>>> {
-    fn map_ok_<U, F>(self, f: F) -> Poll<Option<Result<U, E>>>
-    where
-        F: FnOnce(T) -> U,
-    {
-        match self {
-            Poll::Ready(Some(Ok(t))) => Poll::Ready(Some(Ok(f(t)))),
-            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))),
-            Poll::Ready(None) => Poll::Ready(None),
-            Poll::Pending => Poll::Pending,
-        }
-    }
-
-    fn map_err_<U, F>(self, f: F) -> Poll<Option<Result<T, U>>>
-    where
-        F: FnOnce(E) -> U,
-    {
-        match self {
-            Poll::Ready(Some(Ok(t))) => Poll::Ready(Some(Ok(t))),
-            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(f(e)))),
-            Poll::Ready(None) => Poll::Ready(None),
-            Poll::Pending => Poll::Pending,
-        }
-    }
-}

src/proto/settings.rs

@@ -60,6 +60,10 @@ impl Settings {
                         codec.set_max_recv_header_list_size(max as usize);
                     }
 
+                    if let Some(val) = local.header_table_size() {
+                        codec.set_recv_header_table_size(val as usize);
+                    }
+
                     streams.apply_local_settings(local)?;
                     self.local = Local::Synced;
                     Ok(())

src/proto/streams/streams.rs

@@ -12,7 +12,6 @@ use http::{HeaderMap, Request, Response};
 use std::task::{Context, Poll, Waker};
 use tokio::io::AsyncWrite;
 
-use crate::PollExt;
 use std::sync::{Arc, Mutex};
 use std::{fmt, io};
 
@@ -1282,7 +1281,7 @@ impl OpaqueStreamRef {
         me.actions
             .recv
             .poll_pushed(cx, &mut stream)
-            .map_ok_(|(h, key)| {
+            .map_ok(|(h, key)| {
                 me.refs += 1;
                 let opaque_ref =
                     OpaqueStreamRef::new(self.inner.clone(), &mut me.store.resolve(key));
@@ -1462,9 +1461,21 @@ fn drop_stream_ref(inner: &Mutex<Inner>, key: store::Key) {
 
 fn maybe_cancel(stream: &mut store::Ptr, actions: &mut Actions, counts: &mut Counts) {
     if stream.is_canceled_interest() {
+        // Server is allowed to early respond without fully consuming the client input stream
+        // But per the RFC, must send a RST_STREAM(NO_ERROR) in such cases. https://www.rfc-editor.org/rfc/rfc7540#section-8.1
+        // Some other http2 implementation may interpret other error code as fatal if not respected (i.e: nginx https://trac.nginx.org/nginx/ticket/2376)
+        let reason = if counts.peer().is_server()
+            && stream.state.is_send_closed()
+            && stream.state.is_recv_streaming()
+        {
+            Reason::NO_ERROR
+        } else {
+            Reason::CANCEL
+        };
+
         actions
             .send
-            .schedule_implicit_reset(stream, Reason::CANCEL, counts, &mut actions.task);
+            .schedule_implicit_reset(stream, reason, counts, &mut actions.task);
         actions.recv.enqueue_reset_expiration(stream, counts);
     }
 }

src/server.rs

@@ -554,7 +554,7 @@ where
     ///
     /// This limit is configured by the client peer by sending the
     /// [`SETTINGS_MAX_CONCURRENT_STREAMS` parameter][1] in a `SETTINGS` frame.
-    /// This method returns the currently acknowledged value recieved from the
+    /// This method returns the currently acknowledged value received from the
     /// remote.
     ///
     /// [1]: https://tools.ietf.org/html/rfc7540#section-5.1.2

src/share.rs

@@ -5,7 +5,6 @@ use crate::proto::{self, WindowSize};
 use bytes::{Buf, Bytes};
 use http::HeaderMap;
 
-use crate::PollExt;
 use std::fmt;
 #[cfg(feature = "stream")]
 use std::pin::Pin;
@@ -307,8 +306,8 @@ impl<B: Buf> SendStream<B> {
     pub fn poll_capacity(&mut self, cx: &mut Context) -> Poll<Option<Result<usize, crate::Error>>> {
         self.inner
             .poll_capacity(cx)
-            .map_ok_(|w| w as usize)
-            .map_err_(Into::into)
+            .map_ok(|w| w as usize)
+            .map_err(Into::into)
     }
 
     /// Sends a single data frame to the remote peer.
@@ -403,7 +402,7 @@ impl RecvStream {
 
     /// Poll for the next data frame.
     pub fn poll_data(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<Bytes, crate::Error>>> {
-        self.inner.inner.poll_data(cx).map_err_(Into::into)
+        self.inner.inner.poll_data(cx).map_err(Into::into)
     }
 
     #[doc(hidden)]

tests/h2-support/src/frames.rs

@@ -372,6 +372,11 @@ impl Mock<frame::Settings> {
         self.0.set_enable_connect_protocol(Some(val));
         self
     }
+
+    pub fn header_table_size(mut self, val: u32) -> Self {
+        self.0.set_header_table_size(Some(val));
+        self
+    }
 }
 
 impl From<Mock<frame::Settings>> for frame::Settings {

tests/h2-tests/tests/client_request.rs

@@ -1452,6 +1452,40 @@ async fn extended_connect_request() {
     join(srv, h2).await;
 }
 
+#[tokio::test]
+async fn client_builder_header_table_size() {
+    h2_support::trace_init!();
+    let (io, mut srv) = mock::new();
+    let mut settings = frame::Settings::default();
+
+    settings.set_header_table_size(Some(10000));
+
+    let srv = async move {
+        let recv_settings = srv.assert_client_handshake().await;
+        assert_frame_eq(recv_settings, settings);
+
+        srv.recv_frame(
+            frames::headers(1)
+                .request("GET", "https://example.com/")
+                .eos(),
+        )
+        .await;
+        srv.send_frame(frames::headers(1).response(200).eos()).await;
+    };
+
+    let mut builder = client::Builder::new();
+    builder.header_table_size(10000);
+
+    let h2 = async move {
+        let (mut client, mut h2) = builder.handshake::<_, Bytes>(io).await.unwrap();
+        let request = Request::get("https://example.com/").body(()).unwrap();
+        let (response, _) = client.send_request(request, true).unwrap();
+        h2.drive(response).await.unwrap();
+    };
+
+    join(srv, h2).await;
+}
+
 const SETTINGS: &'static [u8] = &[0, 0, 0, 4, 0, 0, 0, 0, 0];
 const SETTINGS_ACK: &'static [u8] = &[0, 0, 0, 4, 1, 0, 0, 0, 0];
 

tests/h2-tests/tests/server.rs

@@ -566,7 +566,9 @@ async fn sends_reset_cancel_when_req_body_is_dropped() {
         client
             .recv_frame(frames::headers(1).response(200).eos())
             .await;
-        client.recv_frame(frames::reset(1).cancel()).await;
+        client
+            .recv_frame(frames::reset(1).reason(Reason::NO_ERROR))
+            .await;
     };
 
     let srv = async move {