Repository: seanmonstar/reqwest (language: rust)

Issue:
bug: connection pooling on http/2 breaks with high concurrency
I have an application that uses reqwest to make a large amount of outbound HTTP requests in a short timespan. It uses a single process wide `reqwest::Client`. It uses `rustls` for tls. and runs in a single threaded tokio event loop (the application is Deno).

When sending a large amount of requests concurrently, reqwest will reliably fail one of the requests with this error:
```
error sending request for url (https://deno-website2.now.sh/): http2 error: protocol error: not a result of an error
```
<details>
<summary>Full error</summary>

```
reqwest::Error { kind: Request, url: Url { scheme: "https", cannot_be_a_base: false, username: "", password: None, host: Some(Domain("deno-website2.now.sh")), port: None, path: "/", query: None, fragment: None }, source: hyper::Error(Http2, Error { kind: Proto(NO_ERROR) }) }
```
</details>

A reliable reproduction of this is to start 100 futures, each doing 20 requests to https://deno-website2.now.sh/ in series, and then polling them concurrently using a `FuturesUnordered`.

This is not an issue with the upstream server, as the exact same scenario (100 concurrent tasks, with 20 serial requests each) works fine with the Go std library with `GOMAXPROCS=1`.

You can view / try the reproduction here: https://github.com/lucacasonato/reqwest_connection_pooling_bug. Instructions for running are in the README.

It seems that this issue is related to connection pooling as using a single `reqwest::Client` per concurrent future does not exhibit the same behaviour. You can try this configuration by running the reproduction with `--features no-pool`.




Discussion / hints:
Confirmed with @piscisaureus that this is very likely the same root cause as hyperium/hyper#2500

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index 83ba1788e..d5f82e885 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -96,6 +96,7 @@ futures-core = { version = "0.3.0", default-features = false }
 futures-util = { version = "0.3.0", default-features = false }
 http-body = "0.4.0"
 hyper = { version = "0.14", default-features = false, features = ["tcp", "http1", "http2", "client", "runtime"] }
+h2 = "0.3.10"
 lazy_static = "1.4"
 log = "0.4"
 mime = "0.3.16"
diff --git a/src/async_impl/client.rs b/src/async_impl/client.rs
index d1526d9dd..43772ae42 100644
--- a/src/async_impl/client.rs
+++ b/src/async_impl/client.rs
@@ -22,7 +22,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use tokio::time::Sleep;
 
-use log::debug;
+use log::{debug, trace};
 
 use super::decoder::Accepts;
 use super::request::{Request, RequestBuilder};
@@ -1418,6 +1418,8 @@ impl Client {
 
                 urls: Vec::new(),
 
+                retry_count: 0,
+
                 client: self.inner.clone(),
 
                 in_flight,
@@ -1628,6 +1630,8 @@ pin_project! {
 
         urls: Vec<Url>,
 
+        retry_count: usize,
+
         client: Arc<ClientRef>,
 
         #[pin]
@@ -1653,6 +1657,54 @@ impl PendingRequest {
     fn headers(self: Pin<&mut Self>) -> &mut HeaderMap {
         self.project().headers
     }
+
+    fn retry_error(mut self: Pin<&mut Self>, err: &(dyn std::error::Error + 'static)) -> bool {
+        if !is_retryable_error(err) {
+            return false;
+        }
+
+        trace!("can retry {:?}", err);
+
+        let body = match self.body {
+            Some(Some(ref body)) => Body::reusable(body.clone()),
+            Some(None) => {
+                debug!("error was retryable, but body not reusable");
+                return false;
+            }
+            None => Body::empty(),
+        };
+
+        if self.retry_count >= 2 {
+            trace!("retry count too high");
+            return false;
+        }
+        self.retry_count += 1;
+
+        let uri = expect_uri(&self.url);
+        let mut req = hyper::Request::builder()
+            .method(self.method.clone())
+            .uri(uri)
+            .body(body.into_stream())
+            .expect("valid request parts");
+
+        *req.headers_mut() = self.headers.clone();
+
+        *self.as_mut().in_flight().get_mut() = self.client.hyper.request(req);
+
+        true
+    }
+}
+
+fn is_retryable_error(err: &(dyn std::error::Error + 'static)) -> bool {
+    if let Some(cause) = err.source() {
+        if let Some(err) = cause.downcast_ref::<h2::Error>() {
+            // They sent us a graceful shutdown, try with a new connection!
+            return err.is_go_away()
+                && err.is_remote()
+                && err.reason() == Some(h2::Reason::NO_ERROR);
+        }
+    }
+    false
 }
 
 impl Pending {
@@ -1696,6 +1748,9 @@ impl Future for PendingRequest {
         loop {
             let res = match self.as_mut().in_flight().as_mut().poll(cx) {
                 Poll::Ready(Err(e)) => {
+                    if self.as_mut().retry_error(&e) {
+                        continue;
+                    }
                     return Poll::Ready(Err(crate::error::request(e).with_url(self.url.clone())));
                 }
                 Poll::Ready(Ok(res)) => res,


Test patch: