Merge branch 'master' into header-casing

Author: kotx

CHANGELOG.md

@@ -1,3 +1,45 @@
+### v1.8.1 (2025-11-13)
+
+
+#### Bug Fixes
+
+* **http1:** fix consuming extra CPU from previous change (#3977) ([4492f31e](https://github.com/hyperium/hyper/commit/4492f31e9429c34166da5a069c00b65be20e4a02))
+
+
+## v1.8.0 (2025-11-11)
+
+
+#### Bug Fixes
+
+* **http1:** fix rare missed write wakeup on connections (#3952) ([2377b893](https://github.com/hyperium/hyper/commit/2377b893f6e64ca9878e4f25d1472b96baa7e3ea))
+* **http2:** fix internals of HTTP/2 CONNECT upgrades (#3967) ([58e0e7dc](https://github.com/hyperium/hyper/commit/58e0e7dc70612117ccdc40da395922f791cb273a), closes [#3966](https://github.com/hyperium/hyper/issues/3966))
+
+
+#### Features
+
+* **rt:** add `Timer::now()` method to allow overriding the instant returned (#3965) ([5509ebe6](https://github.com/hyperium/hyper/commit/5509ebe6156e32d4f8986fafa25c2918a30005be))
+
+
+#### Breaking Changes
+
+* The HTTP/2 client connection no longer allows an executor
+  that can not spawn itself.
+
+  This was an oversight originally. The client connection will now include spawning
+  a future that keeps a copy of the executor to spawn other futures. Thus, if it is
+  `!Send`, it needs to spawn `!Send` futures. The likelihood of executors that match
+  the previously allowed behavior should be very remote.
+
+  There is also technically a semver break in here, which is that the
+  `Http2ClientConnExec` trait no longer dyn-compatible, because it now expects to
+  be `Clone`. This should not break usage of the `conn` builder, because it already
+  separately had `E: Clone` bounds. If someone were using `dyn Http2ClientConnExec`,
+  that will break. However, there is no purpose for doing so, and it is not usable
+  otherwise, since the trait only exists to propagate bounds into hyper. Thus, the
+  breakage should not affect anyone.
+ ([58e0e7dc](https://github.com/hyperium/hyper/commit/58e0e7dc70612117ccdc40da395922f791cb273a))
+
+
 ## v1.7.0 (2025-08-18)
 
 

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hyper"
-version = "1.7.0"
+version = "1.8.1"
 description = "A protective and efficient HTTP library for all."
 readme = "README.md"
 homepage = "https://hyper.rs"

benches/support/tokiort.rs

@@ -42,6 +42,10 @@ impl Timer for TokioTimer {
         })
     }
 
+    fn now(&self) -> Instant {
+        tokio::time::Instant::now().into()
+    }
+
     fn reset(&self, sleep: &mut Pin<Box<dyn Sleep>>, new_deadline: Instant) {
         if let Some(sleep) = sleep.as_mut().downcast_mut_pin::<TokioSleep>() {
             sleep.reset(new_deadline)

src/client/conn/http2.rs

@@ -640,44 +640,6 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    #[ignore] // only compilation is checked
-    async fn not_send_not_sync_executor_of_send_futures() {
-        #[derive(Clone)]
-        struct TokioExecutor {
-            // !Send, !Sync
-            _x: std::marker::PhantomData<std::rc::Rc<()>>,
-        }
-
-        impl<F> crate::rt::Executor<F> for TokioExecutor
-        where
-            F: std::future::Future + 'static + Send,
-            F::Output: Send + 'static,
-        {
-            fn execute(&self, fut: F) {
-                tokio::task::spawn(fut);
-            }
-        }
-
-        #[allow(unused)]
-        async fn run(io: impl crate::rt::Read + crate::rt::Write + Send + Unpin + 'static) {
-            let (_sender, conn) =
-                crate::client::conn::http2::handshake::<_, _, http_body_util::Empty<bytes::Bytes>>(
-                    TokioExecutor {
-                        _x: Default::default(),
-                    },
-                    io,
-                )
-                .await
-                .unwrap();
-
-            tokio::task::spawn_local(async move {
-                // can't use spawn here because when executor is !Send
-                conn.await.unwrap();
-            });
-        }
-    }
-
     #[tokio::test]
     #[ignore] // only compilation is checked
     async fn send_not_sync_executor_of_send_futures() {

src/client/dispatch.rs

@@ -325,22 +325,23 @@ impl<T> TrySendError<T> {
 
 #[cfg(feature = "http2")]
 pin_project! {
-    pub struct SendWhen<B>
+    pub struct SendWhen<B, E>
     where
         B: Body,
         B: 'static,
     {
         #[pin]
-        pub(crate) when: ResponseFutMap<B>,
+        pub(crate) when: ResponseFutMap<B, E>,
         #[pin]
         pub(crate) call_back: Option<Callback<Request<B>, Response<Incoming>>>,
     }
 }
 
 #[cfg(feature = "http2")]
-impl<B> Future for SendWhen<B>
+impl<B, E> Future for SendWhen<B, E>
 where
     B: Body + 'static,
+    E: crate::rt::bounds::Http2UpgradedExec<B::Data>,
 {
     type Output = ();
 

src/common/time.rs

@@ -50,6 +50,13 @@ impl Time {
         }
     }
 
+    pub(crate) fn now(&self) -> Instant {
+        match *self {
+            Time::Empty => Instant::now(),
+            Time::Timer(ref t) => t.now(),
+        }
+    }
+
     pub(crate) fn reset(&self, sleep: &mut Pin<Box<dyn Sleep>>, new_deadline: Instant) {
         match *self {
             Time::Empty => {

src/proto/h1/conn.rs

@@ -6,7 +6,7 @@ use std::marker::{PhantomData, Unpin};
 use std::pin::Pin;
 use std::task::{Context, Poll};
 #[cfg(feature = "server")]
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 use crate::rt::{Read, Write};
 use bytes::{Buf, Bytes};
@@ -218,7 +218,7 @@ where
         #[cfg(feature = "server")]
         if !self.state.h1_header_read_timeout_running {
             if let Some(h1_header_read_timeout) = self.state.h1_header_read_timeout {
-                let deadline = Instant::now() + h1_header_read_timeout;
+                let deadline = self.state.timer.now() + h1_header_read_timeout;
                 self.state.h1_header_read_timeout_running = true;
                 match self.state.h1_header_read_timeout_fut {
                     Some(ref mut h1_header_read_timeout_fut) => {

src/proto/h2/client.rs

@@ -18,17 +18,16 @@ use http::{Method, StatusCode};
 use pin_project_lite::pin_project;
 
 use super::ping::{Ponger, Recorder};
-use super::{ping, H2Upgraded, PipeToSendStream, SendBuf};
+use super::{ping, PipeToSendStream, SendBuf};
 use crate::body::{Body, Incoming as IncomingBody};
 use crate::client::dispatch::{Callback, SendWhen, TrySendError};
 use crate::common::either::Either;
 use crate::common::io::Compat;
 use crate::common::time::Time;
 use crate::ext::Protocol;
 use crate::headers;
-use crate::proto::h2::UpgradedSendStream;
 use crate::proto::Dispatched;
-use crate::rt::bounds::Http2ClientConnExec;
+use crate::rt::bounds::{Http2ClientConnExec, Http2UpgradedExec};
 use crate::upgrade::Upgraded;
 use crate::{Request, Response};
 use h2::client::ResponseFuture;
@@ -151,7 +150,7 @@ where
     T: Read + Write + Unpin,
     B: Body + 'static,
     B::Data: Send + 'static,
-    E: Http2ClientConnExec<B, T> + Unpin,
+    E: Http2ClientConnExec<B, T> + Clone + Unpin,
     B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
 {
     let (h2_tx, mut conn) = new_builder(config)
@@ -357,7 +356,7 @@ where
 
 pin_project! {
     #[project = H2ClientFutureProject]
-    pub enum H2ClientFuture<B, T>
+    pub enum H2ClientFuture<B, T, E>
     where
         B: http_body::Body,
         B: 'static,
@@ -372,7 +371,7 @@ pin_project! {
         },
         Send {
             #[pin]
-            send_when: SendWhen<B>,
+            send_when: SendWhen<B, E>,
         },
         Task {
             #[pin]
@@ -381,11 +380,12 @@ pin_project! {
     }
 }
 
-impl<B, T> Future for H2ClientFuture<B, T>
+impl<B, T, E> Future for H2ClientFuture<B, T, E>
 where
     B: http_body::Body + 'static,
     B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
     T: Read + Write + Unpin,
+    E: Http2UpgradedExec<B::Data>,
 {
     type Output = ();
 
@@ -484,7 +484,7 @@ impl<B, E, T> ClientTask<B, E, T>
 where
     B: Body + 'static + Unpin,
     B::Data: Send,
-    E: Http2ClientConnExec<B, T> + Unpin,
+    E: Http2ClientConnExec<B, T> + Clone + Unpin,
     B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
     T: Read + Write + Unpin,
 {
@@ -529,6 +529,7 @@ where
                     fut: f.fut,
                     ping: Some(ping),
                     send_stream: Some(send_stream),
+                    exec: self.executor.clone(),
                 },
                 call_back: Some(f.cb),
             },
@@ -537,28 +538,29 @@ where
 }
 
 pin_project! {
-    pub(crate) struct ResponseFutMap<B>
+    pub(crate) struct ResponseFutMap<B, E>
     where
         B: Body,
         B: 'static,
     {
         #[pin]
         fut: ResponseFuture,
-        #[pin]
         ping: Option<Recorder>,
         #[pin]
         send_stream: Option<Option<SendStream<SendBuf<<B as Body>::Data>>>>,
+        exec: E,
     }
 }
 
-impl<B> Future for ResponseFutMap<B>
+impl<B, E> Future for ResponseFutMap<B, E>
 where
     B: Body + 'static,
+    E: Http2UpgradedExec<B::Data>,
 {
     type Output = Result<Response<crate::body::Incoming>, (crate::Error, Option<Request<B>>)>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let mut this = self.project();
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let mut this = self.as_mut().project();
 
         let result = ready!(this.fut.poll(cx));
 
@@ -585,13 +587,10 @@ where
                     let mut res = Response::from_parts(parts, IncomingBody::empty());
 
                     let (pending, on_upgrade) = crate::upgrade::pending();
-                    let io = H2Upgraded {
-                        ping,
-                        send_stream: unsafe { UpgradedSendStream::new(send_stream) },
-                        recv_stream,
-                        buf: Bytes::new(),
-                    };
-                    let upgraded = Upgraded::new(io, Bytes::new());
+
+                    let (h2_up, up_task) = super::upgrade::pair(send_stream, recv_stream, ping);
+                    self.exec.execute_upgrade(up_task);
+                    let upgraded = Upgraded::new(h2_up, Bytes::new());
 
                     pending.fulfill(upgraded);
                     res.extensions_mut().insert(on_upgrade);
@@ -620,7 +619,7 @@ where
     B: Body + 'static + Unpin,
     B::Data: Send,
     B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
-    E: Http2ClientConnExec<B, T> + Unpin,
+    E: Http2ClientConnExec<B, T> + Clone + Unpin,
     T: Read + Write + Unpin,
 {
     type Output = crate::Result<Dispatched>;