fix: auto-release padding from DATA frames

src/frame/data.rs

@@ -139,6 +139,20 @@ impl Data<Bytes> {
             pad_len,
         })
     }
+
+    pub(crate) fn flow_controlled_len(&self) -> usize {
+        if let Some(pad_len) = self.pad_len {
+            // if PADDED, pad length field counts too (the + 1)
+            self.data.len() + usize::from(pad_len) + 1
+        } else {
+            self.data.len()
+        }
+    }
+
+    /// If this frame is PADDED, it returns the pad len + 1 (length field).
+    pub(crate) fn padded_len(&self) -> Option<u8> {
+        self.pad_len.map(|n| n + 1)
+    }
 }
 
 impl<T: Buf> Data<T> {

src/proto/streams/recv.rs

@@ -577,7 +577,8 @@ impl Recv {
     }
 
     pub fn recv_data(&mut self, frame: frame::Data, stream: &mut store::Ptr) -> Result<(), Error> {
-        let sz = frame.payload().len();
+        // could include padding
+        let sz = frame.flow_controlled_len();
 
         // This should have been enforced at the codec::FramedRead layer, so
         // this is just a sanity check.
@@ -628,6 +629,7 @@ impl Recv {
             return Err(Error::library_reset(stream.id, Reason::FLOW_CONTROL_ERROR));
         }
 
+        // use payload len, padding doesn't count for content-length
         if stream.dec_content_length(frame.payload().len()).is_err() {
             proto_err!(stream:
                 "recv_data: content-length overflow; stream={:?}; len={:?}",
@@ -672,6 +674,18 @@ impl Recv {
         // Track the data as in-flight
         stream.in_flight_recv_data += sz;
 
+        // We auto-release the padded length, since the user cannot.
+        if let Some(padded_len) = frame.padded_len() {
+            tracing::trace!(
+                "recv_data; auto-releasing padded length of {:?} for {:?}",
+                padded_len,
+                stream.id,
+            );
+            let _res = self.release_capacity(padded_len.into(), stream, &mut None);
+            // cannot fail, we JUST added more in_flight data above.
+            debug_assert!(_res.is_ok());
+        }
+
         let event = Event::Data(frame.into_payload());
 
         // Push the frame onto the recv buffer

tests/h2-tests/tests/flow_control.rs

@@ -117,6 +117,84 @@ async fn release_capacity_sends_window_update() {
     join(mock, h2).await;
 }
 
+#[tokio::test]
+async fn window_updates_include_padded_length() {
+    h2_support::trace_init!();
+
+    // Our manual way of sending padding frames, not supported publicly
+    const PAYLOAD_LEN: usize = 16_378; // 16_384; does padding + payload count for max frame size?
+    let mut payload = Vec::with_capacity(PAYLOAD_LEN + 6);
+    payload.push(5);
+    payload.extend_from_slice(&[b'z'; PAYLOAD_LEN][..]);
+    payload.extend_from_slice(&[b'0'; 5][..]);
+
+    let (io, mut srv) = mock::new();
+
+    let mock = async move {
+        let settings = srv.assert_client_handshake().await;
+        assert_default_settings!(settings);
+        srv.recv_frame(
+            frames::headers(1)
+                .request("GET", "https://http2.akamai.com/")
+                .eos(),
+        )
+        .await;
+        srv.send_frame(frames::headers(1).response(200)).await;
+        srv.send_frame(frames::data(1, &payload[..]).padded()).await;
+        srv.send_frame(frames::data(1, &payload[..]).padded()).await;
+        srv.send_frame(frames::data(1, &payload[..]).padded()).await;
+        // the other 6 was auto-released earlier
+        srv.recv_frame(frames::window_update(0, 32_774)).await;
+        srv.recv_frame(frames::window_update(1, 32_774)).await;
+        srv.send_frame(frames::data(1, &payload[..]).padded().eos())
+            .await;
+        // but not double released here
+        srv.recv_frame(frames::window_update(0, 32_762)).await;
+        // and no one cares about closed stream window
+    };
+
+    let h2 = async move {
+        let (mut client, h2) = client::handshake(io).await.unwrap();
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("https://http2.akamai.com/")
+            .body(())
+            .unwrap();
+
+        let req = async move {
+            let resp = client.send_request(request, true).unwrap().0.await.unwrap();
+            // Get the response
+            assert_eq!(resp.status(), StatusCode::OK);
+            let mut body = resp.into_parts().1;
+
+            // read some body to use up window size to below half
+            let buf = body.data().await.unwrap().unwrap();
+            assert_eq!(buf.len(), PAYLOAD_LEN);
+
+            let buf = body.data().await.unwrap().unwrap();
+            assert_eq!(buf.len(), PAYLOAD_LEN);
+
+            let buf = body.data().await.unwrap().unwrap();
+            assert_eq!(buf.len(), PAYLOAD_LEN);
+            body.flow_control().release_capacity(buf.len() * 2).unwrap();
+
+            let buf = body.data().await.unwrap().unwrap();
+            assert_eq!(buf.len(), PAYLOAD_LEN);
+            drop(body);
+            idle_ms(20).await;
+        };
+
+        join(
+            async move {
+                h2.await.unwrap();
+            },
+            req,
+        )
+        .await
+    };
+    join(mock, h2).await;
+}
+
 #[tokio::test]
 async fn release_capacity_of_small_amount_does_not_send_window_update() {
     h2_support::trace_init!();