Implicit RST_STREAM blocked until buffered DATA is fully drained

[mmishra100]

When both ResponseFuture and SendStream handles are dropped for a stream that has buffered DATA frames, maybe_cancel() schedules an implicit reset via schedule_implicit_reset(CANCEL). However, the resulting RST_STREAM is not sent until all buffered DATA frames have been drained through flow control first. If the stream's send window is exhausted and the server never sends a WINDOW_UPDATE, the DATA never drains and the RST_STREAM is indefinitely deferred.

How this happens

schedule_implicit_reset() cannot call clear_queue() because it is invoked from type-erased OpaqueStreamRef::drop() which has no access to Buffer<Frame<B>>. It sets ScheduledLibraryReset on the stream state and defers to pop_frame().

In pop_frame(), the ScheduledLibraryReset state is only checked after the stream's pending_send queue is fully drained (the None arm). Before that, pop_frame() attempts to send each buffered DATA frame through flow control like any normal frame. The implicit RST_STREAM effectively sits behind all buffered DATA in priority, rather than preempting it.

This contrasts with explicit send_reset() which works correctly — it calls clear_queue() to discard buffered DATA, queues RST_STREAM directly, and reclaims capacity immediately.

Reproduction

Steps:

1. Set up a server that advertises initial_window_size = 0 (or any small value that will be exhausted)
2. Client sends a POST request with end_of_stream = false
3. Client buffers DATA on the stream via send_data()
4. Client drops both ResponseFuture and SendStream
5. Observe that RST_STREAM is never sent — it is deferred behind the buffered DATA which cannot drain through the exhausted window

Test (for tests/h2-tests/tests/flow_control.rs):

/// When both handles of a stream are dropped while DATA frames are
/// buffered and the send window is exhausted, the implicit RST_STREAM
/// should be sent promptly.
///
/// Currently it is deferred until the buffered DATA is fully drained
/// through flow control, which never completes when the window is zero.
#[tokio::test]
async fn implicit_rst_stream_with_buffered_data_and_zero_window() {
    h2_support::trace_init!();

    let test = async {
        let (io, mut srv) = mock::new();

        let srv = async move {
            let settings = srv
                .assert_client_handshake_with_settings(
                    frames::settings().initial_window_size(0),
                )
                .await;
            assert_default_settings!(settings);

            srv.recv_frame(
                frames::headers(1)
                    .request("POST", "https://example.com/"),
            )
            .await;

            srv.send_frame(frames::headers(1).response(200)).await;

            srv.recv_frame(frames::reset(1).cancel()).await;
        };

        let h2 = async move {
            let (mut client, mut h2) = client::handshake(io).await.unwrap();

            let request = Request::builder()
                .method(Method::POST)
                .uri("https://example.com/")
                .body(())
                .unwrap();

            let (response, mut send_stream) =
                client.send_request(request, false).unwrap();

            let response = h2.drive(response).await.unwrap();
            assert_eq!(response.status(), StatusCode::OK);

            // Buffer DATA that can never be sent (window is 0).
            send_stream
                .send_data(vec![0u8; 10].into(), false)
                .unwrap();

            // Drop both handles — triggers schedule_implicit_reset(CANCEL).
            drop(response);
            drop(send_stream);

            h2.await.unwrap();
        };

        join(srv, h2).await;
    };

    let result = tokio::time::timeout(
        Duration::from_secs(60), test
    ).await;
    assert!(
        result.is_ok(),
        "Timed out: implicit RST_STREAM was not sent \
         (deferred behind buffered DATA that could not drain)",
    );
}

Suggested fix

In pop_frame() in prioritize.rs, when a DATA frame is popped for a stream with get_scheduled_reset(), clear the queue and produce RST_STREAM immediately instead of trying to drain the DATA first:

Some(Frame::Data(frame)) => {
    if let Some(reason) = stream.state.get_scheduled_reset() {
        stream.pending_send.push_front(buffer, frame.into());
        self.clear_queue(buffer, &mut stream);
        self.reclaim_all_capacity(&mut stream, counts);
        stream.set_reset(reason, Initiator::Library);
        let frame = frame::Reset::new(stream.id, reason);
        Frame::Reset(frame)
    } else {
        // existing DATA flow control path
    }
}

This mirrors what explicit send_reset() already does, but triggered from within pop_frame() where Buffer<Frame<B>> is available.

[mmishra100]

created PR for this : #881