Repository: hyperium/hyper (rust)

Issue:
hyper client happens cacel error which should retry
**Version**
List the version(s) of `hyper`, and any relevant hyper dependency (such as `h2` if this is related to HTTP/2).
hyper = { version = "1.1.0", features = ["full"] }
hyper-util = { version = "0.1.3", features = ["full"] }

**Platform**
The output of `uname -a` (UNIX), or version and 32 or 64-bit (Windows)
linux 64

**Description**
hyper client happens cacel error which should retry

[short summary of the bug]
```rust
  let mut connector = HttpConnector::new();
    connector.set_nodelay(true);
    //禁止使用该功能，其会在拿到IPV6和IPV4的时候优先尝试链接IPV6
    connector.set_happy_eyeballs_timeout(None);
    connector.set_connect_timeout(Some(Duration::from_millis(constant::constant_config::CONNECT_TIME_OUT)));
    // connector.set_send_buffer_size(Some(constant::constant_config::BUFFER_SIZE));
    // connector.set_recv_buffer_size(Some(constant::constant_config::BUFFER_SIZE));
    connector.set_reuse_address(true);

    //hyper_util::rt::TokioExecutor::new() 是否会导致上下文切换更多？
    let http_client = Client::builder(TokioExecutor::new())
        .http1_preserve_header_case(true)
        .http1_title_case_headers(false)
        .pool_idle_timeout(Duration::from_millis(constant::constant_config::CLIENT_IDLE_TIME_OUT)) //空闲链接
        // .pool_max_idle_per_host(constant::constant_config::MAX_IDLE_CONNECTIONS_PER_HOST_HTTP_ALL)
        // .executor()
        // .http1_max_buf_size(constant::constant_config::BUFFER_SIZE)
        .http1_ignore_invalid_headers_in_responses(false)
        .http1_allow_spaces_after_header_name_in_responses(false)
        .http1_allow_obsolete_multiline_headers_in_responses(false)
        .timer(TokioTimer::default())
        .pool_timer(TokioTimer::default())
        .build(connector);

    http_client
```

when  use http_client send request,i got some error message as follows:
```
request::error("message", "Error { kind: SendRequest, source: Some(hyper::Error(Canceled, \"connection closed\")) }"
equest::error("message", "Error { kind: SendRequest, source: Some(hyper::Error(Canceled, \"connection was not ready\")) }")
```

 retry_canceled_requests default is true, i think it should not happen these errors



Pull request:
feat(client): add `SendRequest::try_send_request()` method
This change adds a `try_send_request()` method to `SendRequest`. This method returns a `TrySendError` type, which allows for returning the request back to the caller if an error occured between queuing and trying to write the request.

This method is added for both `http1` and `http2`.

Closes #3673 
Closes https://github.com/seanmonstar/reqwest/issues/2325

(Well, after hyper-util makes use of it.)

---

## Problem

If you try to `send_request(req)` at the same time the server is closing the connection, it's possible hyper will notice the HUP before the request gets serialized. If so, theoretically the request is safe to try on another connection.

But there isn't a mechanism for hyper to give the request _back_. And since the pool was moved out of hyper, it doesn't know how to just grab another connection.

**Need**: a mechanism to return the request on specific kinds of errors.

## Options

- Something like `try_send_request(req)`, which returns a `TrySendError`.
- Embed the request into the `hyper::Error`.
- Put some channel-like thing in `req.extensions()`.

### TrySendError

```rust
let mut req = Request::new(body);
loop {
    match tx.try_send_request(req).await {
        Ok(_) => yay,
        Err(TrySendError::Queued { error, request }) => {
            req = req_back;
            continue;
        },
        Err(TrySendError::Sent { error }) => {
            return Err(e);
        }
    }
}
```

Pros:

- From a Rust perspective, it's simple. There's [`std::sync::mpsc::TrySendError`](https://doc.rust-lang.org/std/sync/mpsc/enum.TrySendError.html) as prior art.
- Cheap, no allocations needed, even on the happy path.

Cons:

- Would be needed for each `http1::SendRequest` and `http2::SendRequest`.
- Intangibly, it feels "clumsy". Why wouldn't `send_request()` have handled this?
    - If after trying this out and finding it should be the way, we could create `fn send()` and hide `try_send_request()` and `send_request()`.
- Structural matching and enums must be designed with future-compatility changes.
    - We don't have to expose an enum. It can just have methods.

### Embed in `hyper::Error`

After writing all this out, this is probably not possible.

Pros:

- Feels more elegant.

Cons:

- How do you access the request after?
    - Adding `Error::request()` set of APIs seems to pollute the type.
        - Perhaps an `ext` API can access it...
            - But that would require allocation even on success.
- What if you keep the `Error` around but didn't expect the `Request<B>` to live with it?
    - Is this really a problem? Unknown... But if it is...
    - Could an `ext` be set to signal to hyper that it's OK? Then it becomes opt-in.
- Cannot embed `Request<B>` if the body is `!Send`
    - Can we even do this generically at all?
    - Maybe if the `ext` passed converts into an `Any`...

### Pass `ext::CaptureTrySend`?

```rust
let mut req = Request::new(body);
loop {
    // inserts a capture channel extension, returns the receiver
    let try_rx = hyper::ext::capture_try_send(&mut req);

    match tx.send_request(req).await {
        Ok(_) => yay,
        Err(e) => {
            // take it back
            if let Some(req_back) = try_rx.req().await {
                // try again...
                req = req_back;
                continue;
            } else {
                return Err(e);
            }
        }
    }
}
```

Pros:

- Works with any `B`
- Doesn't keep request stuck in the `Error` for longer than expected


Cons:

- Requires allocating `req.extensions()` for basically every request...



Patch:
diff --git a/src/client/conn/http1.rs b/src/client/conn/http1.rs
index de727582..647171e7 100644
--- a/src/client/conn/http1.rs
+++ b/src/client/conn/http1.rs
@@ -12,7 +12,7 @@ use futures_util::ready;
 use http::{Request, Response};
 use httparse::ParserConfig;
 
-use super::super::dispatch;
+use super::super::dispatch::{self, TrySendError};
 use crate::body::{Body, Incoming as IncomingBody};
 use crate::proto;
 
@@ -51,7 +51,7 @@ pub struct Parts<T> {
 #[must_use = "futures do nothing unless polled"]
 pub struct Connection<T, B>
 where
-    T: Read + Write + 'static,
+    T: Read + Write,
     B: Body + 'static,
 {
     inner: Dispatcher<T, B>,
@@ -59,7 +59,7 @@ where
 
 impl<T, B> Connection<T, B>
 where
-    T: Read + Write + Unpin + 'static,
+    T: Read + Write + Unpin,
     B: Body + 'static,
     B::Error: Into<Box<dyn StdError + Send + Sync>>,
 {
@@ -124,7 +124,7 @@ pub struct Builder {
 /// See [`client::conn`](crate::client::conn) for more.
 pub async fn handshake<T, B>(io: T) -> crate::Result<(SendRequest<B>, Connection<T, B>)>
 where
-    T: Read + Write + Unpin + 'static,
+    T: Read + Write + Unpin,
     B: Body + 'static,
     B::Data: Send,
     B::Error: Into<Box<dyn StdError + Send + Sync>>,
@@ -176,8 +176,16 @@ where
     ///
     /// `req` must have a `Host` header.
     ///
-    /// Absolute-form `Uri`s are not required. If received, they will be serialized
-    /// as-is.
+    /// # Uri
+    ///
+    /// The `Uri` of the request is serialized as-is.
+    ///
+    /// - Usually you want origin-form (`/path?query`).
+    /// - For sending to an HTTP proxy, you want to send in absolute-form
+    ///   (`https://hyper.rs/guides`).
+    ///
+    /// This is however not enforced or validated and it is up to the user
+    /// of this method to ensure the `Uri` is correct for their intended purpose.
     pub fn send_request(
         &mut self,
         req: Request<B>,
@@ -200,33 +208,38 @@ where
         }
     }
 
-    /*
-    pub(super) fn send_request_retryable(
+    /// Sends a `Request` on the associated connection.
+    ///
+    /// Returns a future that if successful, yields the `Response`.
+    ///
+    /// # Error
+    ///
+    /// If there was an error before trying to serialize the request to the
+    /// connection, the message will be returned as part of this error.
+    pub fn try_send_request(
         &mut self,
         req: Request<B>,
-    ) -> impl Future<Output = Result<Response<Body>, (crate::Error, Option<Request<B>>)>> + Unpin
-    where
-        B: Send,
-    {
-        match self.dispatch.try_send(req) {
-            Ok(rx) => {
-                Either::Left(rx.then(move |res| {
-                    match res {
-                        Ok(Ok(res)) => future::ok(res),
-                        Ok(Err(err)) => future::err(err),
-                        // this is definite bug if it happens, but it shouldn't happen!
-                        Err(_) => panic!("dispatch dropped without returning error"),
-                    }
-                }))
-            }
-            Err(req) => {
-                debug!("connection was not ready");
-                let err = crate::Error::new_canceled().with("connection was not ready");
-                Either::Right(future::err((err, Some(req))))
+    ) -> impl Future<Output = Result<Response<IncomingBody>, TrySendError<Request<B>>>> {
+        let sent = self.dispatch.try_send(req);
+        async move {
+            match sent {
+                Ok(rx) => match rx.await {
+                    Ok(Ok(res)) => Ok(res),
+                    Ok(Err(err)) => Err(err),
+                    // this is definite bug if it happens, but it shouldn't happen!
+                    Err(_) => panic!("dispatch dropped without returning error"),
+                },
+                Err(req) => {
+                    debug!("connection was not ready");
+                    let error = crate::Error::new_canceled().with("connection was not ready");
+                    Err(TrySendError {
+                        error,
+                        message: Some(req),
+                    })
+                }
             }
         }
     }
-    */
 }
 
 impl<B> fmt::Debug for SendRequest<B> {
@@ -239,7 +252,7 @@ impl<B> fmt::Debug for SendRequest<B> {
 
 impl<T, B> Connection<T, B>
 where
-    T: Read + Write + Unpin + Send + 'static,
+    T: Read + Write + Unpin + Send,
     B: Body + 'static,
     B::Error: Into<Box<dyn StdError + Send + Sync>>,
 {
@@ -253,7 +266,7 @@ where
 
 impl<T, B> fmt::Debug for Connection<T, B>
 where
-    T: Read + Write + fmt::Debug + 'static,
+    T: Read + Write + fmt::Debug,
     B: Body + 'static,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
@@ -263,7 +276,7 @@ where
 
 impl<T, B> Future for Connection<T, B>
 where
-    T: Read + Write + Unpin + 'static,
+    T: Read + Write + Unpin,
     B: Body + 'static,
     B::Data: Send,
     B::Error: Into<Box<dyn StdError + Send + Sync>>,
@@ -501,7 +514,7 @@ impl Builder {
         io: T,
     ) -> impl Future<Output = crate::Result<(SendRequest<B>, Connection<T, B>)>>
     where
-        T: Read + Write + Unpin + 'static,
+        T: Read + Write + Unpin,
         B: Body + 'static,
         B::Data: Send,
         B::Error: Into<Box<dyn StdError + Send + Sync>>,
diff --git a/src/client/conn/http2.rs b/src/client/conn/http2.rs
index 6d967713..aee135f6 100644
--- a/src/client/conn/http2.rs
+++ b/src/client/conn/http2.rs
@@ -13,7 +13,7 @@ use crate::rt::{Read, Write};
 use futures_util::ready;
 use http::{Request, Response};
 
-use super::super::dispatch;
+use super::super::dispatch::{self, TrySendError};
 use crate::body::{Body, Incoming as IncomingBody};
 use crate::common::time::Time;
 use crate::proto;
@@ -40,7 +40,7 @@ impl<B> Clone for SendRequest<B> {
 #[must_use = "futures do nothing unless polled"]
 pub struct Connection<T, B, E>
 where
-    T: Read + Write + 'static + Unpin,
+    T: Read + Write + Unpin,
     B: Body + 'static,
     E: Http2ClientConnExec<B, T> + Unpin,
     B::Error: Into<Box<dyn Error + Send + Sync>>,
@@ -63,14 +63,14 @@ pub struct Builder<Ex> {
 
 /// Returns a handshake future over some IO.
 ///
-/// This is a shortcut for `Builder::new().handshake(io)`.
+/// This is a shortcut for `Builder::new(exec).handshake(io)`.
 /// See [`client::conn`](crate::client::conn) for more.
 pub async fn handshake<E, T, B>(
     exec: E,
     io: T,
 ) -> crate::Result<(SendRequest<B>, Connection<T, B, E>)>
 where
-    T: Read + Write + Unpin + 'static,
+    T: Read + Write + Unpin,
     B: Body + 'static,
     B::Data: Send,
     B::Error: Into<Box<dyn Error + Send + Sync>>,
@@ -152,33 +152,38 @@ where
         }
     }
 
-    /*
-    pub(super) fn send_request_retryable(
+    /// Sends a `Request` on the associated connection.
+    ///
+    /// Returns a future that if successful, yields the `Response`.
+    ///
+    /// # Error
+    ///
+    /// If there was an error before trying to serialize the request to the
+    /// connection, the message will be returned as part of this error.
+    pub fn try_send_request(
         &mut self,
         req: Request<B>,
-    ) -> impl Future<Output = Result<Response<Body>, (crate::Error, Option<Request<B>>)>> + Unpin
-    where
-        B: Send,
-    {
-        match self.dispatch.try_send(req) {
-            Ok(rx) => {
-                Either::Left(rx.then(move |res| {
-                    match res {
-                        Ok(Ok(res)) => future::ok(res),
-                        Ok(Err(err)) => future::err(err),
-                        // this is definite bug if it happens, but it shouldn't happen!
-                        Err(_) => panic!("dispatch dropped without returning error"),
-                    }
-                }))
-            }
-            Err(req) => {
-                debug!("connection was not ready");
-                let err = crate::Error::new_canceled().with("connection was not ready");
-                Either::Right(future::err((err, Some(req))))
+    ) -> impl Future<Output = Result<Response<IncomingBody>, TrySendError<Request<B>>>> {
+        let sent = self.dispatch.try_send(req);
+        async move {
+            match sent {
+                Ok(rx) => match rx.await {
+                    Ok(Ok(res)) => Ok(res),
+                    Ok(Err(err)) => Err(err),
+                    // this is definite bug if it happens, but it shouldn't happen!
+                    Err(_) => panic!("dispatch dropped without returning error"),
+                },
+                Err(req) => {
+                    debug!("connection was not ready");
+                    let error = crate::Error::new_canceled().with("connection was not ready");
+                    Err(TrySendError {
+                        error,
+                        message: Some(req),
+                    })
+                }
             }
         }
     }
-    */
 }
 
 impl<B> fmt::Debug for SendRequest<B> {
@@ -230,7 +235,7 @@ where
     B::Data: Send,
     E: Unpin,
     B::Error: Into<Box<dyn Error + Send + Sync>>,
-    E: Http2ClientConnExec<B, T> + 'static + Send + Sync + Unpin,
+    E: Http2ClientConnExec<B, T> + Unpin,
 {
     type Output = crate::Result<()>;
 
@@ -432,7 +437,7 @@ where
         io: T,
     ) -> impl Future<Output = crate::Result<(SendRequest<B>, Connection<T, B, Ex>)>>
     where
-        T: Read + Write + Unpin + 'static,
+        T: Read + Write + Unpin,
         B: Body + 'static,
         B::Data: Send,
         B::Error: Into<Box<dyn Error + Send + Sync>>,
@@ -457,3 +462,219 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+
+    #[tokio::test]
+    #[ignore] // only compilation is checked
+    async fn send_sync_executor_of_non_send_futures() {
+        #[derive(Clone)]
+        struct LocalTokioExecutor;
+
+        impl<F> crate::rt::Executor<F> for LocalTokioExecutor
+        where
+            F: std::future::Future + 'static, // not requiring `Send`
+        {
+            fn execute(&self, fut: F) {
+                // This will spawn into the currently running `LocalSet`.
+                tokio::task::spawn_local(fut);
+            }
+        }
+
+        #[allow(unused)]
+        async fn run(io: impl crate::rt::Read + crate::rt::Write + Unpin + 'static) {
+            let (_sender, conn) = crate::client::conn::http2::handshake::<
+                _,
+                _,
+                http_body_util::Empty<bytes::Bytes>,
+            >(LocalTokioExecutor, io)
+            .await
+            .unwrap();
+
+            tokio::task::spawn_local(async move {
+                conn.await.unwrap();
+            });
+        }
+    }
+
+    #[tokio::test]
+    #[ignore] // only compilation is checked
+    async fn not_send_not_sync_executor_of_not_send_futures() {
+        #[derive(Clone)]
+        struct LocalTokioExecutor {
+            _x: std::marker::PhantomData<std::rc::Rc<()>>,
+        }
+
+        impl<F> crate::rt::Executor<F> for LocalTokioExecutor
+        where
+            F: std::future::Future + 'static, // not requiring `Send`
+        {
+            fn execute(&self, fut: F) {
+                // This will spawn into the currently running `LocalSet`.
+                tokio::task::spawn_local(fut);
+            }
+        }
+
+        #[allow(unused)]
+        async fn run(io: impl crate::rt::Read + crate::rt::Write + Unpin + 'static) {
+            let (_sender, conn) =
+                crate::client::conn::http2::handshake::<_, _, http_body_util::Empty<bytes::Bytes>>(
+                    LocalTokioExecutor {
+                        _x: Default::default(),
+                    },
+                    io,
+                )
+                .await
+                .unwrap();
+
+            tokio::task::spawn_local(async move {
+                conn.await.unwrap();
+            });
+        }
+    }
+
+    #[tokio::test]
+    #[ignore] // only compilation is checked
+    async fn send_not_sync_executor_of_not_send_futures() {
+        #[derive(Clone)]
+        struct LocalTokioExecutor {
+            _x: std::marker::PhantomData<std::cell::Cell<()>>,
+        }
+
+        impl<F> crate::rt::Executor<F> for LocalTokioExecutor
+        where
+            F: std::future::Future + 'static, // not requiring `Send`
+        {
+            fn execute(&self, fut: F) {
+                // This will spawn into the currently running `LocalSet`.
+                tokio::task::spawn_local(fut);
+            }
+        }
+
+        #[allow(unused)]
+        async fn run(io: impl crate::rt::Read + crate::rt::Write + Unpin + 'static) {
+            let (_sender, conn) =
+                crate::client::conn::http2::handshake::<_, _, http_body_util::Empty<bytes::Bytes>>(
+                    LocalTokioExecutor {
+                        _x: Default::default(),
+                    },
+                    io,
+                )
+                .await
+                .unwrap();
+
+            tokio::task::spawn_local(async move {
+                conn.await.unwrap();
+            });
+        }
+    }
+
+    #[tokio::test]
+    #[ignore] // only compilation is checked
+    async fn send_sync_executor_of_send_futures() {
+        #[derive(Clone)]
+        struct TokioExecutor;
+
+        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, io)
+            .await
+            .unwrap();
+
+            tokio::task::spawn(async move {
+                conn.await.unwrap();
+            });
+        }
+    }
+
+    #[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() {
+        #[derive(Clone)]
+        struct TokioExecutor {
+            // !Sync
+            _x: std::marker::PhantomData<std::cell::Cell<()>>,
+        }
+
+        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();
+            });
+        }
+    }
+}
diff --git a/src/client/conn/mod.rs b/src/client/conn/mod.rs
index 316aac9d..f982ae6d 100644
--- a/src/client/conn/mod.rs
+++ b/src/client/conn/mod.rs
@@ -18,3 +18,5 @@
 pub mod http1;
 #[cfg(feature = "http2")]
 pub mod http2;
+
+pub use super::dispatch::TrySendError;
diff --git a/src/client/dispatch.rs b/src/client/dispatch.rs
index 7a291c71..b52da60e 100644
--- a/src/client/dispatch.rs
+++ b/src/client/dispatch.rs
@@ -13,10 +13,21 @@ use tokio::sync::{mpsc, oneshot};
 #[cfg(feature = "http2")]
 use crate::{body::Incoming, proto::h2::client::ResponseFutMap};
 
-#[cfg(test)]
-pub(crate) type RetryPromise<T, U> = oneshot::Receiver<Result<U, (crate::Error, Option<T>)>>;
+pub(crate) type RetryPromise<T, U> = oneshot::Receiver<Result<U, TrySendError<T>>>;
 pub(crate) type Promise<T> = oneshot::Receiver<Result<T, crate::Error>>;
 
+/// An error when calling `try_send_request`.
+///
+/// There is a possibility of an error occuring on a connection in-between the
+/// time that a request is queued and when it is actually written to the IO
+/// transport. If that happens, it is safe to return the request back to the
+/// caller, as it was never fully sent.
+#[derive(Debug)]
+pub struct TrySendError<T> {
+    pub(crate) error: crate::Error,
+    pub(crate) message: Option<T>,
+}
+
 pub(crate) fn channel<T, U>() -> (Sender<T, U>, Receiver<T, U>) {
     let (tx, rx) = mpsc::unbounded_channel();
     let (giver, taker) = want::new();
@@ -92,7 +103,7 @@ impl<T, U> Sender<T, U> {
         }
     }
 
-    #[cfg(test)]
+    #[cfg(feature = "http1")]
     pub(crate) fn try_send(&mut self, val: T) -> Result<RetryPromise<T, U>, T> {
         if !self.can_send() {
             return Err(val);
@@ -135,7 +146,6 @@ impl<T, U> UnboundedSender<T, U> {
         self.giver.is_canceled()
     }
 
-    #[cfg(test)]
     pub(crate) fn try_send(&mut self, val: T) -> Result<RetryPromise<T, U>, T> {
         let (tx, rx) = oneshot::channel();
         self.inner
@@ -210,17 +220,17 @@ struct Envelope<T, U>(Option<(T, Callback<T, U>)>);
 impl<T, U> Drop for Envelope<T, U> {
     fn drop(&mut self) {
         if let Some((val, cb)) = self.0.take() {
-            cb.send(Err((
-                crate::Error::new_canceled().with("connection closed"),
-                Some(val),
-            )));
+            cb.send(Err(TrySendError {
+                error: crate::Error::new_canceled().with("connection closed"),
+                message: Some(val),
+            }));
         }
     }
 }
 
 pub(crate) enum Callback<T, U> {
     #[allow(unused)]
-    Retry(Option<oneshot::Sender<Result<U, (crate::Error, Option<T>)>>>),
+    Retry(Option<oneshot::Sender<Result<U, TrySendError<T>>>>),
     NoRetry(Option<oneshot::Sender<Result<U, crate::Error>>>),
 }
 
@@ -229,7 +239,10 @@ impl<T, U> Drop for Callback<T, U> {
         match self {
             Callback::Retry(tx) => {
                 if let Some(tx) = tx.take() {
-                    let _ = tx.send(Err((dispatch_gone(), None)));
+                    let _ = tx.send(Err(TrySendError {
+                        error: dispatch_gone(),
+                        message: None,
+                    }));
                 }
             }
             Callback::NoRetry(tx) => {
@@ -269,18 +282,34 @@ impl<T, U> Callback<T, U> {
         }
     }
 
-    pub(crate) fn send(mut self, val: Result<U, (crate::Error, Option<T>)>) {
+    pub(crate) fn send(mut self, val: Result<U, TrySendError<T>>) {
         match self {
             Callback::Retry(ref mut tx) => {
                 let _ = tx.take().unwrap().send(val);
             }
             Callback::NoRetry(ref mut tx) => {
-                let _ = tx.take().unwrap().send(val.map_err(|e| e.0));
+                let _ = tx.take().unwrap().send(val.map_err(|e| e.error));
             }
         }
     }
 }
 
+impl<T> TrySendError<T> {
+    /// Take the message from this error.
+    ///
+    /// The message will not always have been recovered. If an error occurs
+    /// after the message has been serialized onto the connection, it will not
+    /// be available here.
+    pub fn take_message(&mut self) -> Option<T> {
+        self.message.take()
+    }
+
+    /// Consumes this to return the inner error.
+    pub fn into_error(self) -> crate::Error {
+        self.error
+    }
+}
+
 #[cfg(feature = "http2")]
 pin_project! {
     pub struct SendWhen<B>
@@ -325,8 +354,8 @@ where
                 trace!("send_when canceled");
                 Poll::Ready(())
             }
-            Poll::Ready(Err(err)) => {
-                call_back.send(Err(err));
+            Poll::Ready(Err((error, message))) => {
+                call_back.send(Err(TrySendError { error, message }));
                 Poll::Ready(())
             }
         }
@@ -389,8 +418,8 @@ mod tests {
         let err = fulfilled
             .expect("fulfilled")
             .expect_err("promise should error");
-        match (err.0.kind(), err.1) {
-            (&crate::error::Kind::Canceled, Some(_)) => (),
+        match (err.error.is_canceled(), err.message) {
+            (true, Some(_)) => (),
             e => panic!("expected Error::Cancel(_), found {:?}", e),
         }
     }
diff --git a/src/proto/h1/dispatch.rs b/src/proto/h1/dispatch.rs
index be0a88c1..79ea48be 100644
--- a/src/proto/h1/dispatch.rs
+++ b/src/proto/h1/dispatch.rs
@@ -13,6 +13,8 @@ use http::Request;
 
 use super::{Http1Transaction, Wants};
 use crate::body::{Body, DecodedLength, Incoming as IncomingBody};
+#[cfg(feature = "client")]
+use crate::client::dispatch::TrySendError;
 use crate::common::task;
 use crate::proto::{BodyLength, Conn, Dispatched, MessageHead, RequestHead};
 use crate::upgrade::OnUpgrade;
@@ -107,10 +109,10 @@ where
     ///
     /// This is useful for old-style HTTP upgrades, but ignores
     /// newer-style upgrade API.
-    pub(crate) fn poll_without_shutdown(&mut self, cx: &mut Context<'_>) -> Poll<crate::Result<()>>
-    where
-        Self: Unpin,
-    {
+    pub(crate) fn poll_without_shutdown(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<crate::Result<()>> {
         Pin::new(self).poll_catch(cx, false).map_ok(|ds| {
             if let Dispatched::Upgrade(pending) = ds {
                 pending.manual();
@@ -655,7 +657,10 @@ cfg_client! {
                 }
                 Err(err) => {
                     if let Some(cb) = self.callback.take() {
-                        cb.send(Err((err, None)));
+                        cb.send(Err(TrySendError {
+                            error: err,
+                            message: None,
+                        }));
                         Ok(())
                     } else if !self.rx_closed {
                         self.rx.close();
@@ -663,7 +668,10 @@ cfg_client! {
                             trace!("canceling queued request with connection error: {}", err);
                             // in this case, the message was never even started, so it's safe to tell
                             // the user that the request was completely canceled
-                            cb.send(Err((crate::Error::new_canceled().with(err), Some(req))));
+                            cb.send(Err(TrySendError {
+                                error: crate::Error::new_canceled().with(err),
+                                message: Some(req),
+                            }));
                             Ok(())
                         } else {
                             Err(err)
@@ -729,9 +737,9 @@ mod tests {
             let err = tokio_test::assert_ready_ok!(Pin::new(&mut res_rx).poll(cx))
                 .expect_err("callback should send error");
 
-            match (err.0.kind(), err.1) {
-                (&crate::error::Kind::Canceled, Some(_)) => (),
-                other => panic!("expected Canceled, got {:?}", other),
+            match (err.error.is_canceled(), err.message.as_ref()) {
+                (true, Some(_)) => (),
+                _ => panic!("expected Canceled, got {:?}", err),
             }
         });
     }
diff --git a/src/proto/h2/client.rs b/src/proto/h2/client.rs
index 7cb6c6ed..679b9dfa 100644
--- a/src/proto/h2/client.rs
+++ b/src/proto/h2/client.rs
@@ -22,7 +22,7 @@ use pin_project_lite::pin_project;
 use super::ping::{Ponger, Recorder};
 use super::{ping, H2Upgraded, PipeToSendStream, SendBuf};
 use crate::body::{Body, Incoming as IncomingBody};
-use crate::client::dispatch::{Callback, SendWhen};
+use crate::client::dispatch::{Callback, SendWhen, TrySendError};
 use crate::common::io::Compat;
 use crate::common::time::Time;
 use crate::ext::Protocol;
@@ -137,7 +137,7 @@ pub(crate) async fn handshake<T, B, E>(
     timer: Time,
 ) -> crate::Result<ClientTask<B, E, T>>
 where
-    T: Read + Write + Unpin + 'static,
+    T: Read + Write + Unpin,
     B: Body + 'static,
     B::Data: Send + 'static,
     E: Http2ClientConnExec<B, T> + Unpin,
@@ -611,7 +611,7 @@ where
     B: Body + 'static + Unpin,
     B::Data: Send,
     B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
-    E: Http2ClientConnExec<B, T> + 'static + Send + Sync + Unpin,
+    E: Http2ClientConnExec<B, T> + Unpin,
     T: Read + Write + Unpin,
 {
     type Output = crate::Result<Dispatched>;
@@ -662,10 +662,10 @@ where
                             .map_or(false, |len| len != 0)
                     {
                         warn!("h2 connect request with non-zero body not supported");
-                        cb.send(Err((
-                            crate::Error::new_h2(h2::Reason::INTERNAL_ERROR.into()),
-                            None,
-                        )));
+                        cb.send(Err(TrySendError {
+                            error: crate::Error::new_h2(h2::Reason::INTERNAL_ERROR.into()),
+                            message: None,
+                        }));
                         continue;
                     }
 
@@ -677,7 +677,10 @@ where
                         Ok(ok) => ok,
                         Err(err) => {
                             debug!("client send request error: {}", err);
-                            cb.send(Err((crate::Error::new_h2(err), None)));
+                            cb.send(Err(TrySendError {
+                                error: crate::Error::new_h2(err),
+                                message: None,
+                            }));
                             continue;
                         }
                     };
@@ -702,7 +705,10 @@ where
                         }
                         Poll::Ready(Ok(())) => (),
                         Poll::Ready(Err(err)) => {
-                            f.cb.send(Err((crate::Error::new_h2(err), None)));
+                            f.cb.send(Err(TrySendError {
+                                error: crate::Error::new_h2(err),
+                                message: None,
+                            }));
                             continue;
                         }
                     }
diff --git a/src/server/conn/http1.rs b/src/server/conn/http1.rs
index cf223162..a1e33257 100644
--- a/src/server/conn/http1.rs
+++ b/src/server/conn/http1.rs
@@ -166,7 +166,6 @@ where
     where
         S: Unpin,
         S::Future: Unpin,
-        B: Unpin,
     {
         self.conn.poll_without_shutdown(cx)
     }
@@ -177,12 +176,7 @@ where
     /// # Error
     ///
     /// This errors if the underlying connection protocol is not HTTP/1.
-    pub fn without_shutdown(self) -> impl Future<Output = crate::Result<Parts<I, S>>>
-    where
-        S: Unpin,
-        S::Future: Unpin,
-        B: Unpin,
-    {
+    pub fn without_shutdown(self) -> impl Future<Output = crate::Result<Parts<I, S>>> {
         let mut zelf = Some(self);
         futures_util::future::poll_fn(move |cx| {
             ready!(zelf.as_mut().unwrap().conn.poll_without_shutdown(cx))?;


Test patch:
diff --git a/tests/client.rs b/tests/client.rs
index 89b90796..6b6a6df2 100644
--- a/tests/client.rs
+++ b/tests/client.rs
@@ -2041,6 +2041,91 @@ mod conn {
         assert_eq!(vec, b"bar=foo");
     }
 
+    #[tokio::test]
+    async fn test_try_send_request() {
+        use std::future::Future;
+        let (listener, addr) = setup_tk_test_server().await;
+        let (done_tx, done_rx) = tokio::sync::oneshot::channel::<()>();
+
+        tokio::spawn(async move {
+            let mut sock = listener.accept().await.unwrap().0;
+            let mut buf = [0u8; 8192];
+            sock.read(&mut buf).await.expect("read 1");
+            sock.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n")
+                .await
+                .expect("write 1");
+            let _ = done_rx.await;
+        });
+
+        // make polling fair by putting both in spawns
+        tokio::spawn(async move {
+            let io = tcp_connect(&addr).await.expect("tcp connect");
+            let (mut client, mut conn) = conn::http1::Builder::new()
+                .handshake::<_, Empty<Bytes>>(io)
+                .await
+                .expect("http handshake");
+
+            // get the conn ready
+            assert!(
+                future::poll_fn(|cx| Poll::Ready(Pin::new(&mut conn).poll(cx)))
+                    .await
+                    .is_pending()
+            );
+            assert!(client.is_ready());
+
+            // use the connection once
+            let mut fut1 = std::pin::pin!(client.send_request(http::Request::new(Empty::new())));
+            let _res1 = future::poll_fn(|cx| loop {
+                if let Poll::Ready(res) = fut1.as_mut().poll(cx) {
+                    return Poll::Ready(res);
+                }
+                return match Pin::new(&mut conn).poll(cx) {
+                    Poll::Ready(_) => panic!("ruh roh"),
+                    Poll::Pending => Poll::Pending,
+                };
+            })
+            .await
+            .expect("resp 1");
+
+            assert!(client.is_ready());
+
+            // simulate the server dropping the conn
+            let _ = done_tx.send(());
+            // let the server task die
+            tokio::task::yield_now().await;
+
+            let mut fut2 =
+                std::pin::pin!(client.try_send_request(http::Request::new(Empty::new())));
+            let poll1 = future::poll_fn(|cx| Poll::Ready(fut2.as_mut().poll(cx))).await;
+            assert!(poll1.is_pending(), "not already known to error");
+
+            let mut conn_opt = Some(conn);
+            // wasn't a known error, req is in queue, and now the next poll, the
+            // conn will be noticed as errored
+            let mut err = future::poll_fn(|cx| {
+                loop {
+                    if let Poll::Ready(res) = fut2.as_mut().poll(cx) {
+                        return Poll::Ready(res);
+                    }
+                    if let Some(ref mut conn) = conn_opt {
+                        match Pin::new(conn).poll(cx) {
+                            Poll::Ready(_) => {
+                                conn_opt = None;
+                            } // ok
+                            Poll::Pending => return Poll::Pending,
+                        };
+                    }
+                }
+            })
+            .await
+            .expect_err("resp 2");
+
+            assert!(err.take_message().is_some(), "request was returned");
+        })
+        .await
+        .unwrap();
+    }
+
     #[tokio::test]
     async fn http2_detect_conn_eof() {
         use futures_util::future;