From f132aef3d721526c9b478e2d0c8c73dcbd881319 Mon Sep 17 00:00:00 2001
From: Manish Mishra <mmishra@x.ai>
Date: Thu, 26 Mar 2026 14:25:24 +0000
Subject: [PATCH] fix: send RST_STREAM immediately for CANCEL resets (#880)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

When an implicit CANCEL reset is scheduled (both ResponseFuture and
SendStream dropped), pop_frame() tried to drain buffered DATA through
flow control before producing the RST_STREAM. If the send window was
exhausted the DATA could never drain, so the RST_STREAM was never sent.

Now pop_frame() discards buffered DATA and sends RST_STREAM immediately
for CANCEL resets. NO_ERROR resets still drain their data first, since
they follow a complete response (RFC 9113 §8.1).
---
 src/proto/streams/prioritize.rs      | 21 ++++++++++++
 tests/h2-tests/tests/flow_control.rs | 51 ++++++++++++++++++++++++++++
 tests/h2-tests/tests/server.rs       |  3 +-
 3 files changed, 74 insertions(+), 1 deletion(-)

diff --git a/src/proto/streams/prioritize.rs b/src/proto/streams/prioritize.rs
index 75358d47..0bef6283 100644
--- a/src/proto/streams/prioritize.rs
+++ b/src/proto/streams/prioritize.rs
@@ -731,6 +731,27 @@ impl Prioritize {
 
                     let frame = match stream.pending_send.pop_front(buffer) {
                         Some(Frame::Data(mut frame)) => {
+                            // If a CANCEL reset is scheduled, the stream is
+                            // "no longer needed" (RFC 9113 §7) and we can
+                            // discard buffered DATA and send RST_STREAM
+                            // immediately. Contrast with NO_ERROR, which is
+                            // sent after a complete response (RFC 9113 §8.1)
+                            // and requires all queued data to be sent first.
+                            //
+                            // See https://github.com/hyperium/h2/issues/880.
+                            if stream.state.get_scheduled_reset() == Some(Reason::CANCEL) {
+                                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::CANCEL, Initiator::Library);
+
+                                let frame = frame::Reset::new(stream.id, Reason::CANCEL);
+
+                                counts.transition_after(stream, is_pending_reset);
+                                return Some(Frame::Reset(frame));
+                            }
+
                             // Get the amount of capacity remaining for stream's
                             // window.
                             let stream_capacity = stream.send_flow.available();
diff --git a/tests/h2-tests/tests/flow_control.rs b/tests/h2-tests/tests/flow_control.rs
index 57453e78..80bc0685 100644
--- a/tests/h2-tests/tests/flow_control.rs
+++ b/tests/h2-tests/tests/flow_control.rs
@@ -2237,3 +2237,54 @@ async fn too_many_window_update_resets_causes_go_away() {
 
     join(srv, client).await;
 }
+
+/// Regression test for https://github.com/hyperium/h2/issues/880
+#[tokio::test]
+async fn implicit_reset_with_cancel_discards_buffered_data() {
+    h2_support::trace_init!();
+
+    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;
+
+        tokio::time::timeout(
+            Duration::from_secs(5),
+            srv.recv_frame(frames::reset(1).cancel()),
+        )
+        .await
+        .expect("RST_STREAM not received within 5s");
+    };
+
+    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);
+
+        send_stream.send_data(vec![0u8; 10].into(), false).unwrap();
+
+        drop(response);
+        drop(send_stream);
+
+        h2.await.unwrap();
+    };
+
+    join(srv, h2).await;
+}
diff --git a/tests/h2-tests/tests/server.rs b/tests/h2-tests/tests/server.rs
index 72bdca7f..bf457b4d 100644
--- a/tests/h2-tests/tests/server.rs
+++ b/tests/h2-tests/tests/server.rs
@@ -782,7 +782,8 @@ async fn sends_reset_cancel_when_res_body_is_dropped() {
             )
             .await;
         client.recv_frame(frames::headers(3).response(200)).await;
-        client.recv_frame(frames::data(3, vec![0; 10])).await;
+        // CANCEL means "stream is no longer needed" (RFC 9113 §7).
+        // Buffered DATA is discarded and RST_STREAM is sent immediately.
         client.recv_frame(frames::reset(3).cancel()).await;
     };