Fix HTTP/2 send when flow-control window is negative

`_wait_for_outgoing_flow` previously waited only while the available
flow was exactly zero. Per RFC 7540 §6.9.2, when a peer sends a SETTINGS
frame that reduces INITIAL_WINDOW_SIZE, every existing stream send
window is adjusted by the delta and may become negative. With the
`while flow == 0` guard, a negative window passed straight through and
h2 raised `LocalProtocolError("Cannot send N bytes, flow control window
is -M")` on the next send_data call.

Switching the guard to `while flow <= 0` keeps the stream parked until
WINDOW_UPDATE frames restore positive credit.

Refs: #1082
Refs: encode/httpx#3601

Author: mbeijen

httpcore/_async/http2.py

@@ -496,7 +496,7 @@ async def _wait_for_outgoing_flow(self, request: Request, stream_id: int) -> int
         local_flow: int = self._h2_state.local_flow_control_window(stream_id)
         max_frame_size: int = self._h2_state.max_outbound_frame_size
         flow = min(local_flow, max_frame_size)
-        while flow == 0:
+        while flow <= 0:
             await self._receive_events(request)
             local_flow = self._h2_state.local_flow_control_window(stream_id)
             max_frame_size = self._h2_state.max_outbound_frame_size

httpcore/_sync/http2.py

@@ -496,7 +496,7 @@ def _wait_for_outgoing_flow(self, request: Request, stream_id: int) -> int:
         local_flow: int = self._h2_state.local_flow_control_window(stream_id)
         max_frame_size: int = self._h2_state.max_outbound_frame_size
         flow = min(local_flow, max_frame_size)
-        while flow == 0:
+        while flow <= 0:
             self._receive_events(request)
             local_flow = self._h2_state.local_flow_control_window(stream_id)
             max_frame_size = self._h2_state.max_outbound_frame_size

tests/_async/test_http2.py

@@ -223,6 +223,66 @@ async def test_http2_connection_with_goaway():
             await conn.request("GET", "https://example.com/")
 
 
+@pytest.mark.anyio
+async def test_http2_connection_with_negative_flow_control_window():
+    """
+    When a server sends a SETTINGS frame that reduces INITIAL_WINDOW_SIZE after a
+    stream's window has been fully consumed, the stream's flow control window can go
+    negative. httpcore must wait for WINDOW_UPDATE frames to bring the window back
+    to a positive value before sending more data, rather than proceeding with the
+    negative window and letting h2 raise a LocalProtocolError.
+    See: https://github.com/encode/httpcore/issues/1082
+    """
+    origin = httpcore.Origin(b"https", b"example.com", 443)
+    # The default HTTP/2 stream flow control window is 65535.
+    # We will send 100,000 bytes, exhausting that window after 65535 bytes.
+    # At that point the server reduces INITIAL_WINDOW_SIZE from 65535 to 32768,
+    # which adjusts the stream window from 0 to -32767.
+    # Without the fix (_wait_for_outgoing_flow uses `while flow == 0` instead of
+    # `while flow <= 0`), the negative window is returned directly and h2 raises
+    # LocalProtocolError when send_data is called.
+    reduce_settings = hyperframe.frame.SettingsFrame(stream_id=0)
+    reduce_settings.settings = {
+        hyperframe.frame.SettingsFrame.INITIAL_WINDOW_SIZE: 32768
+    }
+    stream = httpcore.AsyncMockStream(
+        [
+            hyperframe.frame.SettingsFrame(stream_id=0).serialize(),
+            # After window exhaustion, server reduces INITIAL_WINDOW_SIZE:
+            # stream window goes from 0 to 0 + (32768 - 65535) = -32767
+            reduce_settings.serialize(),
+            # Server then provides enough window credit to finish the upload
+            hyperframe.frame.WindowUpdateFrame(
+                stream_id=0, window_increment=100_000
+            ).serialize(),
+            hyperframe.frame.WindowUpdateFrame(
+                stream_id=1, window_increment=100_000
+            ).serialize(),
+            hyperframe.frame.HeadersFrame(
+                stream_id=1,
+                data=hpack.Encoder().encode(
+                    [
+                        (b":status", b"200"),
+                        (b"content-type", b"plain/text"),
+                    ]
+                ),
+                flags=["END_HEADERS"],
+            ).serialize(),
+            hyperframe.frame.DataFrame(
+                stream_id=1, data=b"response", flags=["END_STREAM"]
+            ).serialize(),
+        ]
+    )
+    async with httpcore.AsyncHTTP2Connection(origin=origin, stream=stream) as conn:
+        response = await conn.request(
+            "POST",
+            "https://example.com/",
+            content=b"x" * 100_000,
+        )
+        assert response.status == 200
+        assert response.content == b"response"
+
+
 @pytest.mark.anyio
 async def test_http2_connection_with_flow_control():
     origin = httpcore.Origin(b"https", b"example.com", 443)

tests/_sync/test_http2.py

@@ -224,6 +224,66 @@ def test_http2_connection_with_goaway():
 
 
 
+def test_http2_connection_with_negative_flow_control_window():
+    """
+    When a server sends a SETTINGS frame that reduces INITIAL_WINDOW_SIZE after a
+    stream's window has been fully consumed, the stream's flow control window can go
+    negative. httpcore must wait for WINDOW_UPDATE frames to bring the window back
+    to a positive value before sending more data, rather than proceeding with the
+    negative window and letting h2 raise a LocalProtocolError.
+    See: https://github.com/encode/httpcore/issues/1082
+    """
+    origin = httpcore.Origin(b"https", b"example.com", 443)
+    # The default HTTP/2 stream flow control window is 65535.
+    # We will send 100,000 bytes, exhausting that window after 65535 bytes.
+    # At that point the server reduces INITIAL_WINDOW_SIZE from 65535 to 32768,
+    # which adjusts the stream window from 0 to -32767.
+    # Without the fix (_wait_for_outgoing_flow uses `while flow == 0` instead of
+    # `while flow <= 0`), the negative window is returned directly and h2 raises
+    # LocalProtocolError when send_data is called.
+    reduce_settings = hyperframe.frame.SettingsFrame(stream_id=0)
+    reduce_settings.settings = {
+        hyperframe.frame.SettingsFrame.INITIAL_WINDOW_SIZE: 32768
+    }
+    stream = httpcore.MockStream(
+        [
+            hyperframe.frame.SettingsFrame(stream_id=0).serialize(),
+            # After window exhaustion, server reduces INITIAL_WINDOW_SIZE:
+            # stream window goes from 0 to 0 + (32768 - 65535) = -32767
+            reduce_settings.serialize(),
+            # Server then provides enough window credit to finish the upload
+            hyperframe.frame.WindowUpdateFrame(
+                stream_id=0, window_increment=100_000
+            ).serialize(),
+            hyperframe.frame.WindowUpdateFrame(
+                stream_id=1, window_increment=100_000
+            ).serialize(),
+            hyperframe.frame.HeadersFrame(
+                stream_id=1,
+                data=hpack.Encoder().encode(
+                    [
+                        (b":status", b"200"),
+                        (b"content-type", b"plain/text"),
+                    ]
+                ),
+                flags=["END_HEADERS"],
+            ).serialize(),
+            hyperframe.frame.DataFrame(
+                stream_id=1, data=b"response", flags=["END_STREAM"]
+            ).serialize(),
+        ]
+    )
+    with httpcore.HTTP2Connection(origin=origin, stream=stream) as conn:
+        response = conn.request(
+            "POST",
+            "https://example.com/",
+            content=b"x" * 100_000,
+        )
+        assert response.status == 200
+        assert response.content == b"response"
+
+
+
 def test_http2_connection_with_flow_control():
     origin = httpcore.Origin(b"https", b"example.com", 443)
     stream = httpcore.MockStream(