fix(http2): wait when outbound flow control window is negative

httpcore/_async/http2.py

@@ -489,14 +489,18 @@ async def _wait_for_outgoing_flow(self, request: Request, stream_id: int) -> int
         """
         Returns the maximum allowable outgoing flow for a given stream.
 
-        If the allowable flow is zero, then waits on the network until
+        If the allowable flow is zero or negative, then waits on the network until
         WindowUpdated frames have increased the flow rate.
         https://tools.ietf.org/html/rfc7540#section-6.9
         """
         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:
+        # h2 may report a *negative* window when prior DATA already overshot the
+        # remote window (issue #1082). Treat any non-positive flow like zero:
+        # block on WINDOW_UPDATE instead of returning a negative max_flow that
+        # later makes `send_data` raise LocalProtocolError.
+        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

@@ -489,14 +489,18 @@ def _wait_for_outgoing_flow(self, request: Request, stream_id: int) -> int:
         """
         Returns the maximum allowable outgoing flow for a given stream.
 
-        If the allowable flow is zero, then waits on the network until
+        If the allowable flow is zero or negative, then waits on the network until
         WindowUpdated frames have increased the flow rate.
         https://tools.ietf.org/html/rfc7540#section-6.9
         """
         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:
+        # h2 may report a *negative* window when prior DATA already overshot the
+        # remote window (issue #1082). Treat any non-positive flow like zero:
+        # block on WINDOW_UPDATE instead of returning a negative max_flow that
+        # later makes `send_data` raise LocalProtocolError.
+        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

@@ -380,3 +380,58 @@ async def test_http2_remote_max_streams_update():
                         conn._h2_state.local_settings.max_concurrent_streams,
                     )
                 i += 1
+
+
+@pytest.mark.anyio
+async def test_http2_wait_for_outgoing_flow_handles_negative_window():
+    """Negative local flow must wait like zero (#1082)."""
+    origin = httpcore.Origin(b"https", b"example.com", 443)
+    stream = httpcore.AsyncMockStream(
+        [
+            hyperframe.frame.SettingsFrame().serialize(),
+            # Read while waiting on non-positive flow.
+            hyperframe.frame.WindowUpdateFrame(
+                stream_id=0, window_increment=50_000
+            ).serialize(),
+            hyperframe.frame.WindowUpdateFrame(
+                stream_id=1, window_increment=50_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"ok", flags=["END_STREAM"]
+            ).serialize(),
+        ]
+    )
+    async with httpcore.AsyncHTTP2Connection(origin=origin, stream=stream) as conn:
+        windows = [-49151, -100, 0, 16_384]
+
+        def fake_local_flow_control_window(stream_id):  # noqa: ARG001
+            return windows.pop(0) if windows else 16_384
+
+        real_send_body = conn._send_request_body
+
+        async def send_body_with_negative_window(request, stream_id):
+            setattr(
+                conn._h2_state,
+                "local_flow_control_window",
+                fake_local_flow_control_window,
+            )
+            return await real_send_body(request, stream_id)
+
+        setattr(conn, "_send_request_body", send_body_with_negative_window)
+        response = await conn.request(
+            "POST",
+            "https://example.com/",
+            content=b"x" * 1_000,
+        )
+        assert response.status == 200
+        assert response.content == b"ok"

tests/_sync/test_http2.py

@@ -380,3 +380,58 @@ def test_http2_remote_max_streams_update():
                         conn._h2_state.local_settings.max_concurrent_streams,
                     )
                 i += 1
+
+
+
+def test_http2_wait_for_outgoing_flow_handles_negative_window():
+    """Negative local flow must wait like zero (#1082)."""
+    origin = httpcore.Origin(b"https", b"example.com", 443)
+    stream = httpcore.MockStream(
+        [
+            hyperframe.frame.SettingsFrame().serialize(),
+            # Read while waiting on non-positive flow.
+            hyperframe.frame.WindowUpdateFrame(
+                stream_id=0, window_increment=50_000
+            ).serialize(),
+            hyperframe.frame.WindowUpdateFrame(
+                stream_id=1, window_increment=50_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"ok", flags=["END_STREAM"]
+            ).serialize(),
+        ]
+    )
+    with httpcore.HTTP2Connection(origin=origin, stream=stream) as conn:
+        windows = [-49151, -100, 0, 16_384]
+
+        def fake_local_flow_control_window(stream_id):  # noqa: ARG001
+            return windows.pop(0) if windows else 16_384
+
+        real_send_body = conn._send_request_body
+
+        def send_body_with_negative_window(request, stream_id):
+            setattr(
+                conn._h2_state,
+                "local_flow_control_window",
+                fake_local_flow_control_window,
+            )
+            return real_send_body(request, stream_id)
+
+        setattr(conn, "_send_request_body", send_body_with_negative_window)
+        response = conn.request(
+            "POST",
+            "https://example.com/",
+            content=b"x" * 1_000,
+        )
+        assert response.status == 200
+        assert response.content == b"ok"