Allow sending 0-byte DATA frames when flow-control window is negative

Author: vladyslav

CHANGELOG.rst

@@ -23,7 +23,7 @@ dev
 
 **Bugfixes**
 
--
+- Fix to allow sending 0 bytes on a stream even if the flow control window is negative.
 
 4.3.0 (2025-08-23)
 ------------------

src/h2/connection.py

@@ -888,7 +888,7 @@ def send_data(self,
             "Frame size on stream ID %d is %d", stream_id, frame_size,
         )
 
-        if frame_size > self.local_flow_control_window(stream_id):
+        if frame_size > 0 and frame_size > self.local_flow_control_window(stream_id):
             msg = f"Cannot send {frame_size} bytes, flow control window is {self.local_flow_control_window(stream_id)}"
             raise FlowControlError(msg)
         if frame_size > self.max_outbound_frame_size:
@@ -907,7 +907,7 @@ def send_data(self,
             "Outbound flow control window size is %d",
             self.outbound_flow_control_window,
         )
-        assert self.outbound_flow_control_window >= 0
+        assert self.outbound_flow_control_window >= 0 or frame_size == 0
 
     def end_stream(self, stream_id: int) -> None:
         """

src/h2/stream.py

@@ -981,7 +981,7 @@ def send_data(self,
 
         # Subtract flow_controlled_length to account for possible padding
         self.outbound_flow_control_window -= df.flow_controlled_length
-        assert self.outbound_flow_control_window >= 0
+        assert self.outbound_flow_control_window >= 0 or df.flow_controlled_length == 0
 
         return [df]
 

tests/test_flow_control_window.py

@@ -103,6 +103,17 @@ def test_flow_control_decreases_with_padded_data(self, frame_factory) -> None:
             self.DEFAULT_FLOW_WINDOW - len(b"some data") - 10 - 1
         )
         assert (c.remote_flow_control_window(1) == remaining_length)
+        
+    def test_can_send_zero_bytes_when_window_negative(self) -> None:
+        """
+        Sending 0 bytes when the flow control window is negative should not
+        raise a FlowControlError, allowing empty END_STREAM frames to be sent.
+        """
+        c = h2.connection.H2Connection()
+        c.send_headers(1, self.example_request_headers)
+        c.outbound_flow_control_window = -100
+        c._get_stream_by_id(1).outbound_flow_control_window = -100
+        c.send_data(1, b"", end_stream=True)
 
     def test_flow_control_is_limited_by_connection(self) -> None:
         """