feat(h2): reap a connection that never opens a stream separately

pingora-core/src/apps/mod.rs

@@ -97,8 +97,31 @@ pub struct HttpServerOptions {
     /// If set, close a downstream HTTP/2 connection that has been idle
     /// for this duration.
     ///
+    /// Applies once the connection has carried at least one stream; see
+    /// [`h2_first_stream_timeout`](Self::h2_first_stream_timeout) for the
+    /// period before that.
+    ///
     /// Default: `None`
     pub h2_idle_timeout: Option<Duration>,
+
+    /// If set, close a downstream HTTP/2 connection that completed the
+    /// handshake and has not opened a single stream within this duration.
+    ///
+    /// Separate from [`h2_idle_timeout`](Self::h2_idle_timeout) because the
+    /// two periods deserve different answers: a warm connection between
+    /// requests is worth holding, one that has never been used is not. With
+    /// a single bound, loosening it for the first case loosens it for the
+    /// second, and nothing else reaps a connection that opens no stream.
+    ///
+    /// Choose it with preconnection in mind. A browser acting on a
+    /// `preconnect` hint, a client warming a gRPC channel, and HTTP/2
+    /// connection coalescing all complete a handshake and may legitimately
+    /// open no stream for some seconds. Set too short, this reaps exactly
+    /// those and forces the handshake they existed to avoid; it is a bound
+    /// on how long an *unused* connection may sit, not a liveness check.
+    ///
+    /// Default: `None`, which falls back to `h2_idle_timeout`.
+    pub h2_first_stream_timeout: Option<Duration>,
 }
 
 /// Settings persisted across HTTP/1.x keepalive requests on the same downstream connection.
@@ -340,12 +363,17 @@ where
             // the same code path is exercised by tests in `protocols::http::v2`.
             let app = self.clone();
             let shutdown_for_session = shutdown.clone();
-            let h2_idle_timeout = self.server_options().and_then(|o| o.h2_idle_timeout);
+            let timeouts = server::IdleReaping {
+                idle: self.server_options().and_then(|o| o.h2_idle_timeout),
+                first_stream: self
+                    .server_options()
+                    .and_then(|o| o.h2_first_stream_timeout),
+            };
             server::accept_downstream_sessions(
                 h2_conn,
                 digest,
                 shutdown.clone(),
-                h2_idle_timeout,
+                timeouts,
                 |h2_stream, guard| {
                     let app = app.clone();
                     let shutdown = shutdown_for_session.clone();

pingora-core/src/protocols/http/v2/mod.rs

@@ -355,7 +355,7 @@ mod test {
             connection,
             digest,
             shutdown_rx,
-            None,
+            server::IdleReaping::uniform(None),
             |mut session, _guard| {
                 session_handles.push(tokio::spawn(async move {
                     let req = session.req_header();
@@ -455,7 +455,7 @@ mod test {
             connection,
             digest,
             shutdown_rx,
-            None,
+            server::IdleReaping::uniform(None),
             |mut session, _guard| {
                 session_handles.push(tokio::spawn(async move {
                     let resp = Box::new(ResponseHeader::build(200, None).unwrap());
@@ -520,7 +520,7 @@ mod test {
                 connection,
                 digest,
                 shutdown_rx,
-                None,
+                server::IdleReaping::uniform(None),
                 |_session, _guard| {
                     panic!("did not expect any sessions on an idle connection");
                 },
@@ -566,7 +566,7 @@ mod test {
                 connection,
                 digest,
                 shutdown_rx,
-                Some(Duration::from_millis(100)),
+                server::IdleReaping::uniform(Some(Duration::from_millis(100))),
                 |_session, _guard| panic!("did not expect any sessions on an idle connection"),
             ),
         )
@@ -576,6 +576,205 @@ mod test {
         client_handle.await.unwrap();
     }
 
+    /// A connection that completes the handshake and opens nothing must be
+    /// reaped on the first-stream bound, not the much longer idle bound.
+    /// Without the distinction `idle` is the only reaper and this hangs.
+    #[tokio::test]
+    async fn test_h2_first_stream_timeout_closes_a_connection_that_opens_nothing() {
+        let (mut client, server) = duplex(65536);
+        let (_shutdown_tx, shutdown_rx) = watch::channel(false);
+
+        let client_handle = tokio::spawn(async move {
+            client
+                .write_all(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
+                .await
+                .unwrap();
+            let mut codec: h2::Codec<DuplexStream, Bytes> = h2::Codec::new(client);
+            codec.send(Settings::default().into()).await.unwrap();
+            codec.send(Settings::ack().into()).await.unwrap();
+            while let Some(frame) = codec.next().await {
+                let _ = frame;
+            }
+        });
+
+        let connection = handshake(Box::new(server), None).await.unwrap();
+        let digest = Arc::new(Digest::default());
+
+        // The idle bound is an order of magnitude longer than the test's own
+        // ceiling, so passing proves the *first-stream* bound did the work.
+        let result = pingora_timeout::timeout(
+            Duration::from_secs(2),
+            server::accept_downstream_sessions(
+                connection,
+                digest,
+                shutdown_rx,
+                server::IdleReaping {
+                    idle: Some(Duration::from_secs(60)),
+                    first_stream: Some(Duration::from_millis(100)),
+                },
+                |_session, _guard| panic!("did not expect any sessions"),
+            ),
+        )
+        .await;
+        assert!(
+            result.is_ok(),
+            "a connection that never opened a stream was not reaped"
+        );
+
+        client_handle.await.unwrap();
+    }
+
+    /// The first-stream bound is a deadline from the start of the
+    /// connection, not a duration re-armed on every loop iteration. A
+    /// client that keeps the accept loop busy with streams the codec
+    /// rejects must still be reaped on time: each rejection is a `continue`,
+    /// and `tokio::select!` rebuilds its futures each time round.
+    #[tokio::test]
+    async fn test_h2_first_stream_timeout_is_not_restarted_by_a_rejected_stream() {
+        let (mut client, server) = duplex(65536);
+        let (_shutdown_tx, shutdown_rx) = watch::channel(false);
+        let first_stream = Duration::from_millis(300);
+
+        let client_handle = tokio::spawn(async move {
+            client
+                .write_all(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
+                .await
+                .unwrap();
+            let mut codec: h2::Codec<DuplexStream, Bytes> = h2::Codec::new(client);
+            codec.send(Settings::default().into()).await.unwrap();
+            codec.send(Settings::ack().into()).await.unwrap();
+
+            // A control byte in `:path` is rejected at the stream level, so
+            // the accept loop sees `H2Accept::Rejected` and continues. Stay
+            // under MAX_MALFORMED_STREAMS_PER_CONN so the budget is not what
+            // ends the connection.
+            for id in (1..=21u32).step_by(2) {
+                let pseudo = Pseudo::request(
+                    Method::GET,
+                    Uri::from_static("https://one.one.one.one/"),
+                    None,
+                );
+                // Conflicting Content-Length is an unrecoverable framing
+                // error, reset at the stream level. h2 passes it through, so
+                // unlike a control byte in `:path` it really does reach the
+                // accept loop as `H2Accept::Rejected`.
+                let mut map = HeaderMap::new();
+                map.append("content-length", "10".parse().unwrap());
+                map.append("content-length", "20".parse().unwrap());
+                let mut headers = Headers::new(id.into(), pseudo, map);
+                headers.set_end_headers();
+                if codec.send(headers.into()).await.is_err() {
+                    return;
+                }
+                sleep(first_stream / 4).await;
+            }
+            while let Some(frame) = codec.next().await {
+                let _ = frame;
+            }
+        });
+
+        let connection = handshake(Box::new(server), None).await.unwrap();
+        let digest = Arc::new(Digest::default());
+
+        // 11 rejections at a quarter of the bound each span well over it. If
+        // the deadline restarts on each one the loop runs for ~2.9s and this
+        // times out; held from the start it returns at ~300ms.
+        let result = pingora_timeout::timeout(
+            first_stream * 3,
+            server::accept_downstream_sessions(
+                connection,
+                digest,
+                shutdown_rx,
+                server::IdleReaping {
+                    idle: Some(Duration::from_secs(60)),
+                    first_stream: Some(first_stream),
+                },
+                |_session, _guard| panic!("a rejected stream must not be accepted"),
+            ),
+        )
+        .await;
+        assert!(
+            result.is_ok(),
+            "a rejected stream restarted the first-stream deadline"
+        );
+
+        client_handle.abort();
+    }
+
+    /// Control for the test above: once a stream has been carried, the short
+    /// first-stream bound must no longer apply, or every keepalive connection
+    /// would be closed between requests.
+    #[tokio::test]
+    async fn test_h2_first_stream_timeout_stops_applying_once_a_stream_is_carried() {
+        let (mut client, server) = duplex(65536);
+        let (_shutdown_tx, shutdown_rx) = watch::channel(false);
+
+        let client_handle = tokio::spawn(async move {
+            client
+                .write_all(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
+                .await
+                .unwrap();
+            let mut codec: h2::Codec<DuplexStream, Bytes> = h2::Codec::new(client);
+            codec.send(Settings::default().into()).await.unwrap();
+            codec.send(Settings::ack().into()).await.unwrap();
+
+            let mut headers = Headers::new(
+                1.into(),
+                Pseudo::request(
+                    Method::GET,
+                    Uri::from_static("https://one.one.one.one/"),
+                    None,
+                ),
+                HeaderMap::new(),
+            );
+            headers.set_end_headers();
+            headers.set_end_stream();
+            codec.send(headers.into()).await.unwrap();
+
+            while let Some(frame) = codec.next().await {
+                let _ = frame;
+            }
+        });
+
+        let connection = handshake(Box::new(server), None).await.unwrap();
+        let digest = Arc::new(Digest::default());
+        let (session_tx, session_rx) = oneshot::channel();
+        let mut session_tx = Some(session_tx);
+        let first_stream = Duration::from_millis(100);
+
+        let accept_handle = tokio::spawn(server::accept_downstream_sessions(
+            connection,
+            digest,
+            shutdown_rx,
+            server::IdleReaping {
+                idle: Some(Duration::from_secs(60)),
+                first_stream: Some(first_stream),
+            },
+            move |session, guard| {
+                let sent = session_tx
+                    .take()
+                    .expect("only one session should be accepted")
+                    .send((session, guard));
+                assert!(sent.is_ok(), "test must receive the accepted session");
+            },
+        ));
+
+        let session = pingora_timeout::timeout(Duration::from_secs(1), session_rx)
+            .await
+            .expect("session was not accepted")
+            .expect("accept loop dropped the session");
+        drop(session);
+
+        // Well past the first-stream bound, and far short of the idle one.
+        sleep(first_stream * 5).await;
+        assert!(
+            !accept_handle.is_finished(),
+            "the first-stream bound must not reap a connection that has served a stream"
+        );
+        accept_handle.abort();
+        client_handle.abort();
+    }
+
     #[tokio::test]
     async fn test_h2_idle_timeout_starts_after_active_stream_finishes() {
         let (mut client, server) = duplex(65536);
@@ -618,7 +817,7 @@ mod test {
             connection,
             digest,
             shutdown_rx,
-            Some(idle_timeout),
+            server::IdleReaping::uniform(Some(idle_timeout)),
             move |session, guard| {
                 let sent = guard_tx
                     .take()
@@ -1044,7 +1243,7 @@ mod test {
                 connection,
                 digest,
                 shutdown_rx,
-                None,
+                server::IdleReaping::uniform(None),
                 |mut session, _guard| {
                     session_handles.push(tokio::spawn(async move {
                         let resp = Box::new(ResponseHeader::build(200, None).unwrap());