fix(http2): send a GOAWAY when the user's Service::poll_ready errors

The `Error::source()` is searched for an `h2::Error` to allow sending
different error codes in the GOAWAY. If none is found, it defaults to
`INTERNAL_ERROR`.

src/proto/h2/server.rs

@@ -38,6 +38,7 @@ where
     B: Payload,
 {
     conn: Connection<T, SendBuf<B::Data>>,
+    closing: Option<::Error>,
 }
 
 
@@ -46,7 +47,6 @@ where
     T: AsyncRead + AsyncWrite,
     S: Service<ReqBody=Body, ResBody=B>,
     S::Error: Into<Box<::std::error::Error + Send + Sync>>,
-    //S::Future: Send + 'static,
     B: Payload,
     E: H2Exec<S::Future, B>,
 {
@@ -66,7 +66,9 @@ where
                 // fall-through, to replace state with Closed
             },
             State::Serving(ref mut srv) => {
-                srv.conn.graceful_shutdown();
+                if srv.closing.is_none() {
+                    srv.conn.graceful_shutdown();
+                }
                 return;
             },
             State::Closed => {
@@ -82,7 +84,6 @@ where
     T: AsyncRead + AsyncWrite,
     S: Service<ReqBody=Body, ResBody=B>,
     S::Error: Into<Box<::std::error::Error + Send + Sync>>,
-    //S::Future: Send + 'static,
     B: Payload,
     E: H2Exec<S::Future, B>,
 {
@@ -95,7 +96,8 @@ where
                 State::Handshaking(ref mut h) => {
                     let conn = try_ready!(h.poll().map_err(::Error::new_h2));
                     State::Serving(Serving {
-                        conn: conn,
+                        conn,
+                        closing: None,
                     })
                 },
                 State::Serving(ref mut srv) => {
@@ -127,37 +129,57 @@ where
         S::Error: Into<Box<::std::error::Error + Send + Sync>>,
         E: H2Exec<S::Future, B>,
     {
-        loop {
+        if self.closing.is_none() {
-            // At first, polls the readiness of supplied service.
+            loop {
-            match service.poll_ready() {
+                // At first, polls the readiness of supplied service.
-                Ok(Async::Ready(())) => (),
+                match service.poll_ready() {
-                Ok(Async::NotReady) => {
+                    Ok(Async::Ready(())) => (),
-                    // use `poll_close` instead of `poll`, in order to avoid accepting a request.
+                    Ok(Async::NotReady) => {
-                    try_ready!(self.conn.poll_close().map_err(::Error::new_h2));
+                        // use `poll_close` instead of `poll`, in order to avoid accepting a request.
-                    trace!("incoming connection complete");
+                        try_ready!(self.conn.poll_close().map_err(::Error::new_h2));
-                    return Ok(Async::Ready(()));
+                        trace!("incoming connection complete");
-                }
+                        return Ok(Async::Ready(()));
-                Err(err) => {
+                    }
-                    trace!("service closed");
+                    Err(err) => {
-                    return Err(::Error::new_user_service(err));
+                        let err = ::Error::new_user_service(err);
-                }
+                        debug!("service closed: {}", err);
-            }
 
-            // When the service is ready, accepts an incoming request.
+                        let reason = err.h2_reason();
-            if let Some((req, respond)) = try_ready!(self.conn.poll().map_err(::Error::new_h2)) {
+                        if reason == h2::Reason::NO_ERROR {
-                trace!("incoming request");
+                            // NO_ERROR is only used for graceful shutdowns...
-                let content_length = content_length_parse_all(req.headers());
+                            trace!("interpretting NO_ERROR user error as graceful_shutdown");
-                let req = req.map(|stream| {
+                            self.conn.graceful_shutdown();
-                    ::Body::h2(stream, content_length)
+                        } else {
-                });
+                            trace!("abruptly shutting down with {:?}", reason);
-                let fut = H2Stream::new(service.call(req), respond);
+                            self.conn.abrupt_shutdown(reason);
-                exec.execute_h2stream(fut)?;
+                        }
-            } else {
+                        self.closing = Some(err);
-                // no more incoming streams...
+                        break;
-                trace!("incoming connection complete");
+                    }
-                return Ok(Async::Ready(()))
+                }
+
+                // When the service is ready, accepts an incoming request.
+                if let Some((req, respond)) = try_ready!(self.conn.poll().map_err(::Error::new_h2)) {
+                    trace!("incoming request");
+                    let content_length = content_length_parse_all(req.headers());
+                    let req = req.map(|stream| {
+                        ::Body::h2(stream, content_length)
+                    });
+                    let fut = H2Stream::new(service.call(req), respond);
+                    exec.execute_h2stream(fut)?;
+                } else {
+                    // no more incoming streams...
+                    trace!("incoming connection complete");
+                    return Ok(Async::Ready(()))
+                }
             }
         }
+
+        debug_assert!(self.closing.is_some(), "poll_server broke loop without closing");
+
+        try_ready!(self.conn.poll_close().map_err(::Error::new_h2));
+
+        Err(self.closing.take().expect("polled after error"))
     }
 }
 

src/server/conn.rs

@@ -555,7 +555,6 @@ where
             match polled {
                 Ok(x) => return Ok(x),
                 Err(e) => {
-                    debug!("error polling connection protocol without shutdown: {}", e);
                     match *e.kind() {
                         Kind::Parse(Parse::VersionH2) if self.fallback.to_h2() => {
                             self.upgrade_h2();
@@ -644,7 +643,6 @@ where
                     }
                 })),
                 Err(e) => {
-                    debug!("error polling connection protocol: {}", e);
                     match *e.kind() {
                         Kind::Parse(Parse::VersionH2) if self.fallback.to_h2() => {
                             self.upgrade_h2();
@@ -962,7 +960,6 @@ mod upgrades {
                         return Ok(Async::Ready(()));
                     },
                     Err(e) => {
-                        debug!("error polling connection protocol: {}", e);
                         match *e.kind() {
                             Kind::Parse(Parse::VersionH2) if self.inner.fallback.to_h2() => {
                                 self.inner.upgrade_h2();

tests/server.rs

@@ -1630,6 +1630,63 @@ fn http2_body_user_error_sends_reset_reason() {
     assert_eq!(h2_err.reason(), Some(h2::Reason::INADEQUATE_SECURITY));
 }
 
+#[test]
+fn http2_service_poll_ready_error_sends_goaway() {
+    use std::error::Error;
+
+    struct Svc;
+
+    impl hyper::service::Service for Svc {
+        type ReqBody = hyper::Body;
+        type ResBody = hyper::Body;
+        type Error = h2::Error;
+        type Future = Box<dyn Future<
+            Item = hyper::Response<Self::ResBody>,
+            Error = Self::Error
+        > + Send + Sync>;
+
+        fn poll_ready(&mut self) -> futures::Poll<(), Self::Error> {
+            Err(h2::Error::from(h2::Reason::INADEQUATE_SECURITY))
+        }
+
+        fn call(&mut self, _: hyper::Request<Self::ResBody>) -> Self::Future {
+            unreachable!("poll_ready error should have shutdown conn");
+        }
+    }
+
+    let _ = pretty_env_logger::try_init();
+
+    let server = hyper::Server::bind(&([127, 0, 0, 1], 0).into())
+        .http2_only(true)
+        .serve(|| Ok::<_, BoxError>(Svc));
+
+    let addr_str = format!("http://{}", server.local_addr());
+
+
+    let mut rt = Runtime::new().expect("runtime new");
+
+    rt.spawn(server.map_err(|e| unreachable!("server shouldn't error: {:?}", e)));
+
+    let err = rt.block_on(hyper::rt::lazy(move || {
+        let client = Client::builder()
+            .http2_only(true)
+            .build_http::<hyper::Body>();
+        let uri = addr_str.parse().expect("server addr should parse");
+
+        client
+            .get(uri)
+    })).unwrap_err();
+
+    // client request should have gotten the specific GOAWAY error...
+    let h2_err = err
+        .source()
+        .expect("source")
+        .downcast_ref::<h2::Error>()
+        .expect("downcast");
+
+    assert_eq!(h2_err.reason(), Some(h2::Reason::INADEQUATE_SECURITY));
+}
+
 // -------------------------------------------------
 // the Server that is used to run all the tests with
 // -------------------------------------------------