try_assign_capacity triggers debug assertion

[phu-cinemo]

Hi!

One of our tests occasionally triggers this debug assertion. Unfortunately, I cannot share a minimal reproduction at the moment.

    thread 'tokio-runtime-worker' panicked at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/h2-0.4.13/src/proto/streams/prioritize.rs:448:9:
    state=Closed(ScheduledLibraryReset(CANCEL))
    stack backtrace:
       0: __rustc::rust_begin_unwind
                 at /rustc/29483883eed69d5fb4db01964cdf2af4d86e9cb2/library/std/src/panicking.rs:697:5
       1: core::panicking::panic_fmt
                 at /rustc/29483883eed69d5fb4db01964cdf2af4d86e9cb2/library/core/src/panicking.rs:75:14
       2: h2::proto::streams::prioritize::Prioritize::try_assign_capacity
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/h2-0.4.13/src/proto/streams/prioritize.rs:448:9
       3: h2::proto::streams::prioritize::Prioritize::poll_complete
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/h2-0.4.13/src/proto/streams/prioritize.rs:538:22
       4: h2::proto::streams::send::Send::poll_complete
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/h2-0.4.13/src/proto/streams/send.rs:339:14
       5: h2::proto::streams::streams::Inner::poll_complete
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/h2-0.4.13/src/proto/streams/streams.rs:905:34
       6: h2::proto::streams::streams::Streams<B,P>::poll_complete
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/h2-0.4.13/src/proto/streams/streams.rs:186:12
       7: h2::proto::connection::Connection<T,P,B>::poll
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/h2-0.4.13/src/proto/connection.rs:287:55
       8: <h2::client::Connection<T,B> as core::future::future::Future>::poll
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/h2-0.4.13/src/client.rs:1441:33
       9: <hyper::common::either::Either<F1,F2> as core::future::future::Future>::poll
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/hyper-1.8.1/src/common/either.rs:43:46
      10: <hyper::proto::h2::client::ConnMapErr<T,B> as core::future::future::Future>::poll
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/hyper-1.8.1/src/proto/h2/client.rs:275:32
      11: <core::pin::Pin<P> as core::future::future::Future>::poll
                 at /home/phueber/.rustup/toolchains/stable-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/future/future.rs:124:9
      12: <hyper::proto::h2::client::ConnTask<T,B> as core::future::future::Future>::poll
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/hyper-1.8.1/src/proto/h2/client.rs:340:67
      13: <hyper::proto::h2::client::H2ClientFuture<B,T,E> as core::future::future::Future>::poll
                 at /home/phueber/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/hyper-1.8.1/src/proto/h2/client.rs:398:58

[phu-cinemo]

this is what copilot claims to be the cause:

The root cause is that pop_pending_open (line 893-911) doesn't check whether the stream has been reset/closed before returning it. Compare this with recv_connection_window_update (line 396) which explicitly skips streams that are no longer in a valid
state when popping from pending_capacity. The poll_complete path at line 536-538 lacks an equivalent guard.

[matze]

This reproduces it every time

//! Repro.
//!
//! Cargo.toml:
//! ```toml
//! [package]
//! name = "h2-repro-878"
//! version = "0.1.0"
//! edition = "2021"
//! publish = false
//!
//! [dependencies]
//! h2 = "=0.4.13"
//! bytes = "1"
//! http = "1"
//! tokio = { version = "1", features = ["rt-multi-thread", "macros", "io-util", "time", "sync"] }
//! ```
use std::sync::Arc;

use bytes::Bytes;
use h2::{client, server};
use http::Request;
use tokio::io::DuplexStream;
use tokio::sync::Notify;

async fn run_server(io: DuplexStream, client_ready: Arc<Notify>) {
    let mut builder = server::Builder::new();
    builder.max_concurrent_streams(1);

    let mut conn: server::Connection<_, Bytes> = builder.handshake(io).await.unwrap();

    // Accept stream 1. Don't respond until the client has set up & dropped stream 2.
    if let Some(Ok((_request, mut respond))) = conn.accept().await {
        // Wait for the client to signal that stream 2 handles are dropped.
        client_ready.notified().await;

        // Respond to stream 1 with end_stream=true → stream 1 closes → slot freed.
        let response = http::Response::builder().status(200).body(()).unwrap();
        let _ = respond.send_response(response, true);
    }

    // Drain remaining (the connection should error/close after the panic).
    while let Some(_) = conn.accept().await {}
}

async fn run_client(io: DuplexStream, client_ready: Arc<Notify>) {
    let (mut send_request, connection) = client::handshake(io).await.unwrap();

    // Drive the connection in a separate task — this is where the panic happens.
    let conn_handle = tokio::spawn(async move {
        let _ = connection.await;
    });

    // --- Stream 1: fill the single MAX_CONCURRENT_STREAMS slot ---
    send_request = send_request.ready().await.unwrap();
    let req1 = Request::builder()
        .uri("https://example.com/1")
        .body(())
        .unwrap();
    // end_stream=true: no body, just headers.
    let _resp1 = send_request.send_request(req1, true);

    // --- Stream 2: will be queued in pending_open ---
    let mut send_request2 = send_request.clone();
    send_request2 = send_request2.ready().await.unwrap();
    let req2 = Request::builder()
        .uri("https://example.com/2")
        .body(())
        .unwrap();
    // end_stream=false: the stream expects body data → is_send_streaming() is true.
    let (_fut, mut send_stream) = send_request2.send_request(req2, false).unwrap();

    // Set requested_send_capacity > 0. Without this, try_assign_capacity returns early at the
    // `additional == 0` check before the debug_assert.
    send_stream.reserve_capacity(1);

    // Drop both handles → schedule_implicit_reset(CANCEL) → state =
    // Closed(ScheduledLibraryReset(CANCEL)) Stream stays in pending_open with
    // requested_send_capacity = 1.
    drop(send_stream);
    drop(_fut);

    // Signal the server to respond to stream 1, which frees the slot. The connection task will
    // then pop stream 2 from pending_open and panic.
    client_ready.notify_one();

    conn_handle.await.unwrap();
}

#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() {
    let (client_io, server_io) = tokio::io::duplex(65536);

    // Synchronization: client signals when stream 2 handles have been dropped, so the server can
    // respond to stream 1 (freeing the slot).
    let client_ready = Arc::new(Notify::new());
    let client_ready2 = client_ready.clone();

    let server_handle = tokio::spawn(async move {
        run_server(server_io, client_ready2).await;
    });

    let client_handle = tokio::spawn(async move {
        run_client(client_io, client_ready).await;
    });

    let _ = tokio::join!(server_handle, client_handle);
}