fix: send RST_STREAM immediately for CANCEL resets (#880)

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();

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;
+}

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;
     };