diff --git a/httpcore/_async/http2.py b/httpcore/_async/http2.py index dbd0beeb..de2ae5b3 100644 --- a/httpcore/_async/http2.py +++ b/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 diff --git a/httpcore/_sync/http2.py b/httpcore/_sync/http2.py index ddcc1890..46629726 100644 --- a/httpcore/_sync/http2.py +++ b/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 diff --git a/tests/_async/test_http2.py b/tests/_async/test_http2.py index b4ec6648..d919383e 100644 --- a/tests/_async/test_http2.py +++ b/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" diff --git a/tests/_sync/test_http2.py b/tests/_sync/test_http2.py index 695359bd..3573450c 100644 --- a/tests/_sync/test_http2.py +++ b/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"