test graceful shutdown with idle connections

.changesets/fix_geal_idle_connections.md

@@ -0,0 +1,9 @@
+### Fix router hang when opening the explorer, prometheus or health check page ([Issue #3941](https://github.com/apollographql/router/issues/3941))
+
+The Router did not gracefully shutdown when an idle connections are made by a client, and would instead hang. In particular, web browsers make such connection in anticipation of future traffic.
+
+This is now fixed, and the Router will now gracefully shut down in a timely fashion.
+
+---
+
+By [@Geal](https://github.com/Geal) in https://github.com/apollographql/router/pull/3969

apollo-router/src/axum_factory/listeners.rs

@@ -2,6 +2,8 @@
 
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
@@ -16,6 +18,7 @@ use multimap::MultiMap;
 use tokio::net::UnixListener;
 use tokio::sync::mpsc;
 use tokio::sync::Notify;
+use tower_service::Service;
 
 use crate::configuration::Configuration;
 use crate::http_server_factory::Listener;
@@ -229,6 +232,9 @@ pub(super) fn serve_router_on_listen_addr(
 
                                 match res {
                                     NetworkStream::Tcp(stream) => {
+                                        let received_first_request = Arc::new(AtomicBool::new(false));
+                                        let app = IdleConnectionChecker::new(received_first_request.clone(), app);
+
                                         stream
                                             .set_nodelay(true)
                                             .expect(
@@ -251,7 +257,12 @@ pub(super) fn serve_router_on_listen_addr(
                                                 let c = connection.as_mut();
                                                 c.graceful_shutdown();
 
-                                                let _= connection.await;
+                                                // if the connection was idle and we never received the first request,
+                                                // hyper's graceful shutdown would wait indefinitely, so instead we
+                                                // close the connection right away
+                                                if received_first_request.load(Ordering::Relaxed) {
+                                                    let _= connection.await;
+                                                }
                                             }
                                         }
                                     }
@@ -278,6 +289,9 @@ pub(super) fn serve_router_on_listen_addr(
                                         }
                                     },
                                     NetworkStream::Tls(stream) => {
+                                        let received_first_request = Arc::new(AtomicBool::new(false));
+                                        let app = IdleConnectionChecker::new(received_first_request.clone(), app);
+
                                         stream.get_ref().0
                                             .set_nodelay(true)
                                             .expect(
@@ -305,7 +319,12 @@ pub(super) fn serve_router_on_listen_addr(
                                                 let c = connection.as_mut();
                                                 c.graceful_shutdown();
 
-                                                let _= connection.await;
+                                                // if the connection was idle and we never received the first request,
+                                                // hyper's graceful shutdown would wait indefinitely, so instead we
+                                                // close the connection right away
+                                                if received_first_request.load(Ordering::Relaxed) {
+                                                    let _= connection.await;
+                                                }
                                             }
                                         }
                                     }
@@ -405,6 +424,42 @@ pub(super) fn serve_router_on_listen_addr(
     (server, shutdown_sender)
 }
 
+struct IdleConnectionChecker<S> {
+    received_request: Arc<AtomicBool>,
+    inner: S,
+}
+
+impl<S> IdleConnectionChecker<S> {
+    fn new(b: Arc<AtomicBool>, service: S) -> Self {
+        IdleConnectionChecker {
+            received_request: b,
+            inner: service,
+        }
+    }
+}
+impl<S, B> Service<http::Request<B>> for IdleConnectionChecker<S>
+where
+    S: Service<http::Request<B>>,
+{
+    type Response = <S as Service<http::Request<B>>>::Response;
+
+    type Error = <S as Service<http::Request<B>>>::Error;
+
+    type Future = <S as Service<http::Request<B>>>::Future;
+
+    fn poll_ready(
+        &mut self,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::result::Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: http::Request<B>) -> Self::Future {
+        self.received_request.store(true, Ordering::Relaxed);
+        self.inner.call(req)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::net::SocketAddr;

apollo-router/tests/lifecycle_tests.rs

@@ -171,6 +171,22 @@ async fn test_reload_via_sighup() -> Result<(), BoxError> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_shutdown_with_idle_connection() -> Result<(), BoxError> {
+    let mut router = IntegrationTest::builder()
+        .config(HAPPY_CONFIG)
+        .build()
+        .await;
+    router.start().await;
+    router.assert_started().await;
+    let _conn = std::net::TcpStream::connect("127.0.0.1:4000").unwrap();
+    router.execute_default_query().await;
+    tokio::time::timeout(Duration::from_millis(100), router.graceful_shutdown())
+        .await
+        .unwrap();
+    Ok(())
+}
+
 async fn command_output(command: &mut Command) -> String {
     let output = command.output().await.unwrap();
     let success = output.status.success();