H2 response-header timeout can drop the only ResponseFuture and make the session panic on retry

[YZL0v3ZZ]

Describe the bug

Pingora's HTTP/2 client response-header path can lose the only h2 ResponseFuture for an in-flight stream when Http2Session::read_response_header() is cancelled or times out before the upstream response header is ready.

The vulnerable state transition is:

write_request_header() stores h2::client::ResponseFuture in self.resp_fut
-> read_response_header() removes it with self.resp_fut.take()
-> the method awaits the taken future, optionally under read_timeout
-> timeout/select/task cancellation drops read_response_header() while the ResponseFuture is still pending
-> the taken ResponseFuture is dropped
-> Http2Session survives with resp_fut = None and response_header = None
-> retrying read_response_header() panics instead of resuming or returning a structured stream error

This is a cancellation state-machine issue in the lower-level H2 client session API. The response header has not been installed, but the only continuation capable of reading it has been moved out of the surviving session object and can be destroyed by ordinary Rust async cancellation.

The bug is not that every read timeout must keep using the stream. A product may choose to abort an H2 stream after a response-header timeout. The issue is that the current API can return a timeout while leaving a live Http2Session in an internally unrecoverable state: the request was sent, no response header exists, no response body reader exists, and retrying the same header read hits a panic path.

Root cause

The root cause is visible in pingora-core/src/protocols/http/v2/client.rs. Source links below are pinned to commit e6e677fe9b58555140ab7bd14feff035392b3530.

1. Sending an H2 request creates the response future and stores it as the only resumable header-read continuation:

pingora-core/src/protocols/http/v2/client.rs#L121-L129

let (resp_fut, send_body) = self
    .send_req
    .send_request(request, end)
    .or_err(H2Error, "while sending request")
    .map_err(|e| self.handle_err(e))?;
self.req_sent = Some(req);
self.send_body = Some(send_body);
self.resp_fut = Some(resp_fut);
self.ended = self.ended || end;

2. read_response_header() then destructively removes that future from the session before it is known to be ready:

pingora-core/src/protocols/http/v2/client.rs#L174-L184

pub async fn read_response_header(&mut self) -> Result<()> {
    // TODO: how to read 1xx headers?
    // https://github.com/hyperium/h2/issues/167

    if self.response_header.is_some() {
        panic!("H2 response header is already read")
    }

    let Some(resp_fut) = self.resp_fut.take() else {
        panic!("Try to take response header, but it is already taken")
    };

3. The cancellation point is the subsequent await. With read_timeout set, Pingora wraps the taken future in a timeout; when that timeout elapses, the inner resp_fut is dropped:

pingora-core/src/protocols/http/v2/client.rs#L186-L192

let res = match self.read_timeout {
    Some(t) => timeout(t, resp_fut)
        .await
        .map_err(|_| Error::explain(ReadTimedout, "while reading h2 response header"))
        .map_err(|e| self.handle_err(e))?,
    None => resp_fut.await,
};

External cancellation has the same effect. For example, if a caller uses tokio::select!, an external timeout, task abort, parent future drop, or runtime shutdown while this async method is suspended, the generated read_response_header() future is dropped together with the taken ResponseFuture.

4. The response state is installed only after the awaited future completes successfully:

pingora-core/src/protocols/http/v2/client.rs#L193-L195

let (resp, body_reader) = res.map_err(handle_read_header_error)?.into_parts();
self.response_header = Some(resp.into());
self.response_body_reader = Some(body_reader);

So cancellation can leave exactly this state:

req_sent = Some(...)
send_body = Some(...)
resp_fut = None
response_header = None
response_body_reader = None

That state violates the expected session invariant: if the response header has not been read yet, the session must still retain the future needed to read it, or it must be explicitly marked failed/aborted.

5. The same file contains a safer local contrast. poll_read_response_header() also takes resp_fut, but if the future is not ready it restores it into the session before returning Poll::Pending:

pingora-core/src/protocols/http/v2/client.rs#L201-L227

pub fn poll_read_response_header(
    &mut self,
    cx: &mut Context<'_>,
) -> Poll<Result<(), h2::Error>> {
    if self.response_header.is_some() {
        panic!("H2 response header is already read")
    }

    let Some(mut resp_fut) = self.resp_fut.take() else {
        panic!("Try to take response header, but it is already taken")
    };

    let res = match resp_fut.poll_unpin(cx) {
        Poll::Ready(Ok(res)) => res,
        Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
        Poll::Pending => {
            self.resp_fut = Some(resp_fut);
            return Poll::Pending;
        }
    };

    let (resp, body_reader) = res.into_parts();
    self.response_header = Some(resp.into());
    self.response_body_reader = Some(body_reader);

    Poll::Ready(Ok(()))
}

This contrast is the key state-machine evidence: H2 response-header progress can be kept resumable by restoring the response future on pending. The async wrapper currently stores that resumable progress only inside a cancellable method future.

6. The generic client API exposes this path through HttpSession::H2:

pingora-core/src/protocols/http/client.rs#L103-L135

pub fn set_read_timeout(&mut self, timeout: Option<Duration>) {
    match self {
        HttpSession::H1(h1) => h1.read_timeout = timeout,
        HttpSession::H2(h2) => h2.read_timeout = timeout,
        HttpSession::Custom(c) => c.set_read_timeout(timeout),
    }
}

pub async fn read_response_header(&mut self) -> Result<()> {
    match self {
        HttpSession::H1(h1) => {
            h1.read_response().await?;
            Ok(())
        }
        HttpSession::H2(h2) => h2.read_response_header().await,
        HttpSession::Custom(c) => c.read_response_header().await,
    }
}

One in-tree user, HTTP health checks, sets a read timeout and then awaits the generic response-header API:

pingora-load-balancing/src/health_check.rs#L301-L303

session.set_read_timeout(peer.options.read_timeout);

session.read_response_header().await?;

The H2 proxy path also sets the H2 client read timeout from peer options:

pingora-proxy/src/proxy_h2.rs#L172

client_session.read_timeout = peer.options.read_timeout;

There is a proxy-specific mitigation for one application-level upstream ReadTimedout path: Pingora sends RST_STREAM CANCEL on the request body stream and marks the H2 connection for shutdown.

pingora-proxy/src/proxy_h2.rs#L220-L229

if e.esource == ErrorSource::Upstream && matches!(e.etype, ReadTimedout) {
    client_body.send_reset(h2::Reason::CANCEL);
    // Mark the underlying H2 connection for shutdown so it's not used
    // for new streams in case it is hung.
    client_session.conn.mark_shutdown();
}

That lowers the risk for this specific proxy timeout path, but it does not repair Http2Session::read_response_header() itself. Direct H2 users, generic HttpSession users, caller-side select! or timeout cancellation, task abort, and shutdown cancellation can still leave the lower-level session in the panic-on-retry state.

Pingora info

Please include the following information about your environment:

Pingora version: audited and reproduced against commit e6e677fe9b58555140ab7bd14feff035392b3530; the targeted unit-test log reports pingora-core v0.8.0
Rust version: not captured in the shared reproduction log
Operating system version: reproduced as a Rust unit test on Linux x86 (server36)

Steps to reproduce

The following is a focused whitebox reproduction. It intentionally asserts the current bad behavior, so the test passing means the cancellation window is reachable in the current implementation. After a fix, the assertion should be inverted into a regression property such as: a timed-out or cancelled response-header read must not leave resp_fut = None while response_header = None; retrying should not panic.

Steps:

1. Apply the test-only code below inside the existing #[cfg(test)] mod tests_h2 in pingora-core/src/protocols/http/v2/client.rs.
2. Run the targeted test command shown below.
3. Observe that the first header read returns ReadTimedout, and the retry panics because the only ResponseFuture was lost.

Full whitebox test code for pingora-core/src/protocols/http/v2/client.rs

#[cfg(test)]
mod tests_h2 {
    use super::*;
    use bytes::Bytes;
    use http::{Response, StatusCode};
    use std::panic::AssertUnwindSafe;
    use tokio::io::duplex;
    use tokio::sync::oneshot;

    #[tokio::test(start_paused = true)]
    async fn cancelled_response_header_read_loses_response_future() {
        let (client_io, server_io) = duplex(65536);
        let (request_accepted_tx, request_accepted_rx) = oneshot::channel();
        let (release_response_tx, release_response_rx) = oneshot::channel();

        let server_task = tokio::spawn(async move {
            let mut conn = h2::server::handshake(server_io).await.unwrap();
            if let Some(result) = conn.accept().await {
                let (req, mut send_resp) = result.unwrap();
                assert_eq!(req.method(), http::Method::GET);
                let _ = request_accepted_tx.send(());
                let _ = release_response_rx.await;

                let resp = Response::builder().status(StatusCode::OK).body(()).unwrap();
                let _ = send_resp.send_response(resp, true);
                conn.graceful_shutdown();
            }
        });

        let (send_req, connection) = h2::client::handshake(client_io).await.unwrap();
        let (closed_tx, closed_rx) = tokio::sync::watch::channel(false);
        let ping_timeout = Arc::new(AtomicBool::new(false));
        let connection_task = tokio::spawn(async move {
            let _ = connection.await;
            let _ = closed_tx.send(true);
        });

        let digest = Digest::default();
        let conn_ref = crate::connectors::http::v2::ConnectionRef::new(
            send_req.clone(),
            closed_rx,
            ping_timeout,
            0,
            1,
            digest,
        );
        let mut h2s = Http2Session::new(send_req, conn_ref);

        let mut req = RequestHeader::build("GET", b"/", None).unwrap();
        req.insert_header(http::header::HOST, "example.com")
            .unwrap();
        h2s.write_request_header(Box::new(req), true).unwrap();

        request_accepted_rx
            .await
            .expect("server should accept the request before the read timeout");
        h2s.read_timeout = Some(Duration::from_millis(1));

        let err = h2s
            .read_response_header()
            .await
            .expect_err("delayed response header should hit the read timeout");
        assert!(
            matches!(err.etype, ReadTimedout),
            "unexpected first read error: {err:?}"
        );
        assert!(h2s.response_header().is_none());

        let retry = AssertUnwindSafe(h2s.read_response_header())
            .catch_unwind()
            .await;
        let panic_payload =
            retry.expect_err("retry should panic because the response future was lost");
        let panic_message = if let Some(message) = panic_payload.downcast_ref::<&str>() {
            *message
        } else if let Some(message) = panic_payload.downcast_ref::<String>() {
            message.as_str()
        } else {
            ""
        };
        assert!(
            panic_message.contains("Try to take response header, but it is already taken"),
            "unexpected panic message: {panic_message:?}"
        );

        let _ = release_response_tx.send(());
        drop(h2s);
        server_task.abort();
        connection_task.abort();
    }
}

If the existing tests_h2 module is already present, only the additional imports and the cancelled_response_header_read_loses_response_future test function are needed. The Bytes, Response, StatusCode, duplex, and oneshot imports are already used by the neighboring H2 tests in the current file; the reproduction additionally needs std::panic::AssertUnwindSafe for the caught retry panic.

Targeted command:

cargo test -p pingora-core --lib cancelled_response_header_read_loses_response_future -- --nocapture

Observed result:

Compiling pingora-core v0.8.0 (/home/jcj/whitebox/pingora/pingora-core)
Finished `test` profile [unoptimized + debuginfo] target(s) in 6.40s
     Running unittests src/lib.rs (target/debug/deps/pingora_core-4ef134c88541a33d)

running 1 test

thread 'protocols::http::v2::client::tests_h2::cancelled_response_header_read_loses_response_future' (2954060) panicked at pingora-core/src/protocols/http/v2/client.rs:183:13:
Try to take response header, but it is already taken
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
test protocols::http::v2::client::tests_h2::cancelled_response_header_read_loses_response_future ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 485 filtered out; finished in 0.01s

Why this reproduces the issue:

• The test creates a real in-memory h2 client/server pair over tokio::io::duplex.
• The server accepts the request but waits on a gate before sending the response header, ensuring the client's ResponseFuture remains pending.
• The client builds a real Http2Session, sends a GET request, and sets h2s.read_timeout = Some(Duration::from_millis(1)).
• The first read_response_header().await reaches the timeout path and returns a structured ReadTimedout error.
• At that timeout boundary, the taken ResponseFuture has been dropped by the timeout wrapper, while h2s.response_header() is still None.
• The test then retries h2s.read_response_header() on the same live session and catches the resulting panic.
• The panic message is the self.resp_fut.take() None branch: Try to take response header, but it is already taken.

This is a bug-existence validation test. It passes on the current implementation because it reaches and catches the bad retry panic. A post-fix regression test should instead assert that retrying does not panic and that the session either resumes the header read, returns a structured timeout/error, or has been explicitly transitioned to an aborted/failed state before the first timeout returns.

Expected results

A cancelled or timed-out H2 response-header read should not destroy the only continuation needed to read the response header while leaving the Http2Session object live and apparently usable.

Acceptable outcomes include:

• preserve the same ResponseFuture in self.resp_fut whenever the header-read future yields or is cancelled before completion;
• return ReadTimedout while keeping the session in a resumable ResponsePending state;
• or, if timeout is intended to consume/abort the stream, explicitly transition the stream/session to a failed or aborted state and make later calls return a structured error rather than panicking.

The expected invariant is:

if response_header == None:
    either resp_fut == Some(ResponseFuture)
    or the stream/session is explicitly Failed/Aborted/Closed

Retrying after a timeout should never observe:

resp_fut == None
response_header == None
retry behavior == panic

Observed results

The current implementation can produce this state after read_response_header() times out or is cancelled while the upstream H2 response header is still pending:

request sent: yes
response future retained: no
response header installed: no
response body reader installed: no
stream/session explicitly failed: no
retry behavior: panic at "Try to take response header, but it is already taken"

The immediate observable failure is a panic on retry. The broader state-machine problem is that the response-header continuation has been dropped without either completing the response state or recording an explicit terminal stream state.

This can affect direct users of Http2Session and generic HttpSession::H2 users that treat a timeout or cancellation as a recoverable event, or that rely on caller-side cancellation such as select!, external timeout, task abort, parent future drop, or runtime shutdown. The proxy H2 path has a partial application-level mitigation for one upstream ReadTimedout branch, but that caller-specific cleanup does not make the lower-level H2 response-header API cancellation-safe.

Additional context

The cancellation-safety invariant I expected is:

An H2 client session must not remove its only pending response-header future
from durable session state across a cancellable await unless it also commits
the response state or records an explicit unrecoverable stream state.

A robust fix would likely mirror the existing poll_read_response_header() restore-on-pending pattern in the async API:

• take resp_fut only for the duration of one poll;
• if polling returns Poll::Pending, restore self.resp_fut = Some(resp_fut) before yielding;
• if polling returns Poll::Ready(Ok(_)), install response_header and response_body_reader;
• if polling returns Poll::Ready(Err(_)), return the structured error or transition the session to an explicit failed state;
• for timeout handling, avoid wrapping an owned ResponseFuture that has already been removed from the session, or explicitly abort/mark the stream before returning ReadTimedout.

One possible state-machine shape is:

ResponsePending(resp_fut)
-> poll resp_fut
-> Pending: store ResponsePending(resp_fut), yield Pending
-> Ready(Ok(resp)): store ResponseReady(header, body_reader)
-> timeout/cancel before ready: keep ResponsePending(resp_fut) or store Failed/Aborted

The important repair property is that the response future and the response state should have a single durable owner at every suspension boundary. A cancellable method future should not be the only owner of protocol progress after the public session object has been left alive.