From aceaa79944cb0373a79454462aa372bf63bf400b Mon Sep 17 00:00:00 2001 From: emicuencac Date: Wed, 9 Sep 2026 15:04:09 -0300 Subject: [PATCH 1/7] Bugfix bound the shutdown drain by graceful_timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Server.wait_closed()` returned as soon as the listening socket was closed until Python 3.12.1, where it changed to wait until every active connection has finished (python/cpython#79033, python/cpython#104344). The asyncio worker awaits it before the `wait_for(..., graceful_timeout)` that is meant to bound the drain, so on 3.12.1+ a single connection that never finishes holds the worker in `worker_serve` forever and `graceful_timeout` is never reached. `Server.close()` already stops serving on the listening sockets, and every connection task is tracked in `server_tasks`, which the following `wait_for` gathers and cancels. Dropping the `wait_closed()` call restores the documented behaviour without losing anything. This also unwedges `max_requests` recycling: a worker that trips its limit while a request is still in flight never exits, and `run.py` only spawns the replacement once the old process has exited, so the arbiter is left holding the listening socket with no worker accepting on it. The trio worker is unaffected — it bounds the same drain with `server_nursery.cancel_scope.deadline`. --- src/hypercorn/asyncio/run.py | 1 - tests/asyncio/test_run.py | 54 ++++++++++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) create mode 100644 tests/asyncio/test_run.py diff --git a/src/hypercorn/asyncio/run.py b/src/hypercorn/asyncio/run.py index 09fec25..82ddb9e 100644 --- a/src/hypercorn/asyncio/run.py +++ b/src/hypercorn/asyncio/run.py @@ -163,7 +163,6 @@ async def _server_callback(reader: asyncio.StreamReader, writer: asyncio.StreamW for server in servers: server.close() - await server.wait_closed() try: gathered_server_tasks = asyncio.gather(*server_tasks) diff --git a/tests/asyncio/test_run.py b/tests/asyncio/test_run.py new file mode 100644 index 0000000..7be6d0a --- /dev/null +++ b/tests/asyncio/test_run.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from hypercorn.app_wrappers import ASGIWrapper +from hypercorn.asyncio.run import worker_serve +from hypercorn.config import Config +from hypercorn.typing import ASGIReceiveCallable, ASGISendCallable, Scope + + +async def slow_framework( + scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable +) -> None: + if scope["type"] == "lifespan": + while True: + message = await receive() + if message["type"] == "lifespan.startup": + await send({"type": "lifespan.startup.complete"}) + elif message["type"] == "lifespan.shutdown": + await send({"type": "lifespan.shutdown.complete"}) + return + await asyncio.sleep(100) + + +@pytest.mark.asyncio +async def test_graceful_timeout_bounds_active_connections() -> None: + config = Config() + config.bind = ["127.0.0.1:0"] + config.graceful_timeout = 0.5 + sockets = config.create_sockets() + port = sockets.insecure_sockets[0].getsockname()[1] + + shutdown = asyncio.Event() + serving = asyncio.ensure_future( + worker_serve( + ASGIWrapper(slow_framework), config, sockets=sockets, shutdown_trigger=shutdown.wait + ) + ) + await asyncio.sleep(0.1) + + reader, writer = await asyncio.open_connection("127.0.0.1", port) + writer.write(b"GET / HTTP/1.1\r\nHost: hypercorn\r\n\r\n") + await writer.drain() + await asyncio.sleep(0.1) + + shutdown.set() + try: + # The request never completes, so the drain must be bounded by + # graceful_timeout rather than waiting for the connection to close. + await asyncio.wait_for(serving, timeout=5) + finally: + writer.close() From c1f21b42d0cd2d2a0d28370b64d41a38f6b03efe Mon Sep 17 00:00:00 2001 From: emicuencac Date: Wed, 9 Sep 2026 20:00:31 -0300 Subject: [PATCH 2/7] Bugfix close terminated HTTP/2 connections instead of restarting the idle task When the last stream on an HTTP/2 connection closes after the worker has been told to terminate, `H2Protocol` sends GOAWAY and then emits `Updated(idle=True)`. The asyncio `TCPServer` reacts to that by restarting the idle task inside the connection's `TaskGroup`. If the stream was closed because `graceful_timeout` expired and the connection task is being cancelled, that `TaskGroup` is already shutting down and `create_task` raises `RuntimeError`. The error replaces the `CancelledError`, escapes `_server_callback` as an `ExceptionGroup`, is not matched by the `except asyncio.TimeoutError` in `worker_serve`, and the worker exits non-zero. `run.py` treats any non-zero worker exit as fatal and shuts down every other worker, so a single long-lived stream (e.g. a Restate invocation) still open at `graceful_timeout` could take the whole server down on a `max_requests` recycle. `H11Protocol._maybe_recycle` already sends `Closed()` rather than `Updated(idle=True)` once terminated; do the same in `H2Protocol`. The bytes on the wire are unchanged (GOAWAY, then FIN) and, since an idle connection has no live streams, nothing is skipped by closing directly rather than via the idle task. This has been reachable since ab98383 moved the idle task into the per-connection `TaskGroup`, but on Python 3.12.1+ was masked by the unbounded `Server.wait_closed()` removed in aceaa799. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/hypercorn/protocol/h2.py | 2 + tests/asyncio/test_run.py | 91 +++++++++++++++++++++++++++++++++--- tests/protocol/test_h2.py | 50 +++++++++++++++++++- 3 files changed, 135 insertions(+), 8 deletions(-) diff --git a/src/hypercorn/protocol/h2.py b/src/hypercorn/protocol/h2.py index d50e0f8..10aaec7 100644 --- a/src/hypercorn/protocol/h2.py +++ b/src/hypercorn/protocol/h2.py @@ -228,6 +228,8 @@ async def stream_send(self, event: StreamEvent) -> None: if idle and self.context.terminated.is_set(): self.connection.close_connection() await self._flush() + await self.send(Closed()) + return await self.send(Updated(idle=idle)) elif isinstance(event, Request): await self._create_server_push(event.stream_id, event.raw_path, event.headers) diff --git a/tests/asyncio/test_run.py b/tests/asyncio/test_run.py index 7be6d0a..8729436 100644 --- a/tests/asyncio/test_run.py +++ b/tests/asyncio/test_run.py @@ -2,6 +2,10 @@ import asyncio +import h2.config +import h2.connection +import h2.errors +import h2.events import pytest from hypercorn.app_wrappers import ASGIWrapper @@ -10,20 +14,40 @@ from hypercorn.typing import ASGIReceiveCallable, ASGISendCallable, Scope +async def _lifespan(receive: ASGIReceiveCallable, send: ASGISendCallable) -> None: + while True: + message = await receive() + if message["type"] == "lifespan.startup": + await send({"type": "lifespan.startup.complete"}) + elif message["type"] == "lifespan.shutdown": + await send({"type": "lifespan.shutdown.complete"}) + return + + async def slow_framework( scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable ) -> None: if scope["type"] == "lifespan": - while True: - message = await receive() - if message["type"] == "lifespan.startup": - await send({"type": "lifespan.startup.complete"}) - elif message["type"] == "lifespan.shutdown": - await send({"type": "lifespan.shutdown.complete"}) - return + await _lifespan(receive, send) + return await asyncio.sleep(100) +async def streaming_framework( + scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable +) -> None: + if scope["type"] == "lifespan": + await _lifespan(receive, send) + return + # Bidirectional stream: respond immediately, then keep both directions open. + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"start", "more_body": True}) + while True: + message = await receive() + if message["type"] == "http.disconnect": + return + + @pytest.mark.asyncio async def test_graceful_timeout_bounds_active_connections() -> None: config = Config() @@ -52,3 +76,56 @@ async def test_graceful_timeout_bounds_active_connections() -> None: await asyncio.wait_for(serving, timeout=5) finally: writer.close() + + +@pytest.mark.asyncio +async def test_graceful_timeout_cancels_active_h2_stream( + caplog: pytest.LogCaptureFixture, +) -> None: + config = Config() + config.bind = ["127.0.0.1:0"] + config.graceful_timeout = 0.5 + sockets = config.create_sockets() + port = sockets.insecure_sockets[0].getsockname()[1] + + shutdown = asyncio.Event() + serving = asyncio.ensure_future( + worker_serve( + ASGIWrapper(streaming_framework), + config, + sockets=sockets, + shutdown_trigger=shutdown.wait, + ) + ) + await asyncio.sleep(0.1) + + reader, writer = await asyncio.open_connection("127.0.0.1", port) + client = h2.connection.H2Connection(h2.config.H2Configuration(client_side=True)) + client.initiate_connection() + client.send_headers( + 1, + [(":method", "POST"), (":path", "/"), (":scheme", "http"), (":authority", "hypercorn")], + ) + client.send_data(1, b"start") # No end_stream, the request stays open + writer.write(client.data_to_send()) + await writer.drain() + await asyncio.sleep(0.1) + client.receive_data(await reader.read(65536)) + writer.write(client.data_to_send()) + await writer.drain() + + shutdown.set() + try: + # The stream never completes, so it is cancelled when graceful_timeout + # expires. This must not raise out of the worker. + await asyncio.wait_for(serving, timeout=5) + + data = b"" + while chunk := await asyncio.wait_for(reader.read(65536), timeout=1): + data += chunk + events = client.receive_data(data) + assert isinstance(events[-1], h2.events.ConnectionTerminated) + assert events[-1].error_code == h2.errors.ErrorCodes.NO_ERROR + assert "Unhandled exception" not in caplog.text + finally: + writer.close() diff --git a/tests/protocol/test_h2.py b/tests/protocol/test_h2.py index b549496..74f2466 100644 --- a/tests/protocol/test_h2.py +++ b/tests/protocol/test_h2.py @@ -9,7 +9,8 @@ from hypercorn.asyncio.worker_context import EventWrapper, WorkerContext from hypercorn.config import Config -from hypercorn.events import Closed, RawData +from hypercorn.events import Closed, RawData, Updated +from hypercorn.protocol.events import StreamClosed from hypercorn.protocol.h2 import BUFFER_HIGH_WATER, BufferCompleteError, H2Protocol, StreamBuffer from hypercorn.typing import ConnectionState @@ -116,3 +117,50 @@ async def test_protocol_keep_alive_max_requests() -> None: protocol.send.assert_awaited() # type: ignore events = client.receive_data(protocol.send.call_args_list[1].args[0].data) # type: ignore assert isinstance(events[-1], ConnectionTerminated) + + +async def _terminated_protocol_with_streams(*stream_ids: int) -> tuple[H2Protocol, H2Connection]: + protocol = H2Protocol( + Mock(), + Config(), + WorkerContext(None), + AsyncMock(), + ConnectionState({}), + False, + None, + None, + AsyncMock(), + ) + client = H2Connection() + client.initiate_connection() + headers = [ + (":method", "POST"), + (":path", "/"), + (":authority", "hypercorn"), + (":scheme", "https"), + ] + for stream_id in stream_ids: + client.send_headers(stream_id, headers) + await protocol.handle(RawData(data=client.data_to_send())) + await protocol.context.terminated.set() + protocol.send.reset_mock() # type: ignore + return protocol, client + + +@pytest.mark.asyncio +async def test_protocol_terminated_last_stream_closed() -> None: + protocol, client = await _terminated_protocol_with_streams(1) + await protocol.stream_send(StreamClosed(stream_id=1)) + # GOAWAY is sent and then the connection is closed, rather than + # restarting the idle task on a connection that is being torn down. + assert len(protocol.send.call_args_list) == 2 # type: ignore + events = client.receive_data(protocol.send.call_args_list[0].args[0].data) # type: ignore + assert isinstance(events[-1], ConnectionTerminated) + assert protocol.send.call_args_list[1] == call(Closed()) # type: ignore + + +@pytest.mark.asyncio +async def test_protocol_terminated_stream_closed_with_active_streams() -> None: + protocol, client = await _terminated_protocol_with_streams(1, 3) + await protocol.stream_send(StreamClosed(stream_id=1)) + assert protocol.send.call_args_list == [call(Updated(idle=False))] # type: ignore From 18a3b2ea32e871f01c0ddd0905369e9a18f77cd9 Mon Sep 17 00:00:00 2001 From: Gaston Avila Date: Thu, 10 Sep 2026 12:50:55 -0300 Subject: [PATCH 3/7] Add failing tests for the terminated HTTP/2 drain path Three tests that fail on fix-308-graceful-timeout-drain: - test_protocol_terminated_data_for_refused_stream: DataReceived for a stream that was refused after termination raises KeyError in H2Protocol._handle_events. - test_request_during_drain_does_not_crash_worker: that KeyError now propagates through wait_for(gather(...)) in worker_serve, so the worker exits non-zero and the master shuts every worker down. - test_graceful_timeout_bounds_app_that_swallows_cancellation: an app that swallows CancelledError keeps worker_serve stuck after graceful_timeout, because wait_for awaits the cancellation without a bound. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- tests/asyncio/test_run.py | 117 ++++++++++++++++++++++++++++++++++++++ tests/protocol/test_h2.py | 22 ++++++- 2 files changed, 138 insertions(+), 1 deletion(-) diff --git a/tests/asyncio/test_run.py b/tests/asyncio/test_run.py index 8729436..3ac3c4f 100644 --- a/tests/asyncio/test_run.py +++ b/tests/asyncio/test_run.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +from typing import Any import h2.config import h2.connection @@ -129,3 +130,119 @@ async def test_graceful_timeout_cancels_active_h2_stream( assert "Unhandled exception" not in caplog.text finally: writer.close() + + +class CancellationSwallowingFramework: + # Some frameworks catch the cancellation and carry on, e.g. restate_sdk's + # ServerInvocationContext.enter() does `except asyncio.CancelledError: pass` + # and then keeps awaiting the request. The server must still exit. + # `release` lets the test unwind the stuck request afterwards; otherwise a + # failing run also hangs the event loop teardown. + + def __init__(self) -> None: + self.release = asyncio.Event() + + async def __call__( + self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable + ) -> None: + if scope["type"] == "lifespan": + await _lifespan(receive, send) + return + try: + await asyncio.sleep(100) + except asyncio.CancelledError: + pass + await self.release.wait() + + +H2_REQUEST_HEADERS = [ + (":method", "POST"), + (":path", "/"), + (":scheme", "http"), + (":authority", "hypercorn"), +] + + +async def _serve( + framework: Any, graceful_timeout: float +) -> tuple[asyncio.Future, asyncio.Event, int]: + config = Config() + config.bind = ["127.0.0.1:0"] + config.graceful_timeout = graceful_timeout + sockets = config.create_sockets() + port = sockets.insecure_sockets[0].getsockname()[1] + shutdown = asyncio.Event() + serving = asyncio.ensure_future( + worker_serve( + ASGIWrapper(framework), config, sockets=sockets, shutdown_trigger=shutdown.wait + ) + ) + await asyncio.sleep(0.1) + return serving, shutdown, port + + +async def _open_h2_stream( + port: int, +) -> tuple[asyncio.StreamReader, asyncio.StreamWriter, h2.connection.H2Connection]: + reader, writer = await asyncio.open_connection("127.0.0.1", port) + client = h2.connection.H2Connection(h2.config.H2Configuration(client_side=True)) + client.initiate_connection() + client.send_headers(1, H2_REQUEST_HEADERS) + client.send_data(1, b"start") # No end_stream, the request stays open + writer.write(client.data_to_send()) + await writer.drain() + await asyncio.sleep(0.1) + client.receive_data(await reader.read(65536)) + writer.write(client.data_to_send()) + await writer.drain() + return reader, writer, client + + +@pytest.mark.asyncio +async def test_graceful_timeout_bounds_app_that_swallows_cancellation() -> None: + framework = CancellationSwallowingFramework() + serving, shutdown, port = await _serve(framework, 0.5) + reader, writer, client = await _open_h2_stream(port) + + shutdown.set() + try: + # Cancelling the connection tasks is not enough on its own: the app + # ignores the first cancellation, so waiting for the cancellation to + # complete hangs. The drain must be bounded regardless of the app. + # asyncio.wait rather than wait_for: wait_for would cancel `serving` + # and then hang on that cancellation for the same reason. + done, _ = await asyncio.wait({serving}, timeout=5) + assert serving in done, "worker_serve did not exit after graceful_timeout" + serving.result() + finally: + framework.release.set() + writer.close() + await asyncio.wait({serving}, timeout=5) + + +@pytest.mark.asyncio +async def test_request_during_drain_does_not_crash_worker( + caplog: pytest.LogCaptureFixture, +) -> None: + serving, shutdown, port = await _serve(streaming_framework, 2) + reader, writer, client = await _open_h2_stream(port) + + shutdown.set() + await asyncio.sleep(0.1) + # A client that has not seen a GOAWAY yet starts a new request. HEADERS + # and DATA arrive in one read. The terminated worker refuses the stream + # with RST_STREAM and must ignore the DATA that follows it. It must not + # let the connection's exception escape worker_serve: a non-zero worker + # exit makes the master shut down every other worker. + client.send_headers(3, H2_REQUEST_HEADERS) + client.send_data(3, b"body") + writer.write(client.data_to_send()) + await writer.drain() + + try: + done, _ = await asyncio.wait({serving}, timeout=5) + assert serving in done, "worker_serve did not exit after graceful_timeout" + serving.result() + assert "Unhandled exception" not in caplog.text + finally: + writer.close() diff --git a/tests/protocol/test_h2.py b/tests/protocol/test_h2.py index 74f2466..ec19534 100644 --- a/tests/protocol/test_h2.py +++ b/tests/protocol/test_h2.py @@ -5,7 +5,7 @@ import pytest from h2.connection import H2Connection -from h2.events import ConnectionTerminated +from h2.events import ConnectionTerminated, StreamReset from hypercorn.asyncio.worker_context import EventWrapper, WorkerContext from hypercorn.config import Config @@ -164,3 +164,23 @@ async def test_protocol_terminated_stream_closed_with_active_streams() -> None: protocol, client = await _terminated_protocol_with_streams(1, 3) await protocol.stream_send(StreamClosed(stream_id=1)) assert protocol.send.call_args_list == [call(Updated(idle=False))] # type: ignore + + +@pytest.mark.asyncio +async def test_protocol_terminated_data_for_refused_stream() -> None: + protocol, client = await _terminated_protocol_with_streams(1) + # The client sent HEADERS and DATA back to back, before it could see the + # RST_STREAM that a terminated worker answers new requests with. Both + # frames arrive in one read, so the DataReceived event refers to a stream + # that was refused and never entered protocol.streams. + client.send_headers( + 3, + [(":method", "POST"), (":path", "/"), (":authority", "hypercorn"), (":scheme", "https")], + ) + client.send_data(3, b"body") + await protocol.handle(RawData(data=client.data_to_send())) + events = client.receive_data(protocol.send.call_args_list[0].args[0].data) # type: ignore + assert isinstance(events[0], StreamReset) + assert events[0].stream_id == 3 + # The pre-existing stream is unaffected + assert 1 in protocol.streams From 22289ebf378ea1b832b7fa6e56d6bfcdc43a75fd Mon Sep 17 00:00:00 2001 From: Gaston Avila Date: Thu, 10 Sep 2026 12:50:55 -0300 Subject: [PATCH 4/7] Bound the drain and ignore DataReceived for refused streams Two changes that make the new tests pass: - H2Protocol._handle_events: look the stream up with .get() for DataReceived. The stream is gone when the request was refused as it arrived (connection terminated) or the response was sent before the full request was received. The window credit is still returned. - worker_serve: replace wait_for(gather(*server_tasks)) with asyncio.wait with a timeout, cancel what is still pending, then wait for the cancellation with a second timeout. gather() re-raised the first connection error into worker_serve, and wait_for() awaited the cancellation forever when an app swallowed CancelledError. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/hypercorn/asyncio/run.py | 16 ++++++++-------- src/hypercorn/protocol/h2.py | 9 ++++++--- 2 files changed, 14 insertions(+), 11 deletions(-) diff --git a/src/hypercorn/asyncio/run.py b/src/hypercorn/asyncio/run.py index 82ddb9e..1fe6fe7 100644 --- a/src/hypercorn/asyncio/run.py +++ b/src/hypercorn/asyncio/run.py @@ -165,15 +165,15 @@ async def _server_callback(reader: asyncio.StreamReader, writer: asyncio.StreamW server.close() try: - gathered_server_tasks = asyncio.gather(*server_tasks) - await asyncio.wait_for(gathered_server_tasks, config.graceful_timeout) - except asyncio.TimeoutError: - pass + if server_tasks: + _, pending = await asyncio.wait(server_tasks, timeout=config.graceful_timeout) + for task in pending: + task.cancel() + if pending: + # Bounded, so an app that swallows CancelledError + # cannot keep the worker alive. + await asyncio.wait(pending, timeout=config.graceful_timeout) finally: - # Retrieve the Gathered Tasks Cancelled Exception, to - # prevent a warning that this hasn't been done. - gathered_server_tasks.exception() - await lifespan.wait_for_shutdown() lifespan_task.cancel() await lifespan_task diff --git a/src/hypercorn/protocol/h2.py b/src/hypercorn/protocol/h2.py index 10aaec7..dbe2c0e 100644 --- a/src/hypercorn/protocol/h2.py +++ b/src/hypercorn/protocol/h2.py @@ -258,9 +258,12 @@ async def _handle_events(self, events: list[h2.events.Event]) -> None: if self.keep_alive_requests > self.config.keep_alive_max_requests: self.connection.close_connection() elif isinstance(event, h2.events.DataReceived): - await self.streams[event.stream_id].handle( - Body(stream_id=event.stream_id, data=event.data) - ) + stream = self.streams.get(event.stream_id) + if stream is not None: + # The stream is gone when the request was refused + # as it arrived (terminated) or the response was + # sent before the full request was received. + await stream.handle(Body(stream_id=event.stream_id, data=event.data)) self.connection.acknowledge_received_data( event.flow_controlled_length, event.stream_id ) From f3803056d7e2819d31ada80c991e8bb6b426a2ca Mon Sep 17 00:00:00 2001 From: emicuencac Date: Thu, 10 Sep 2026 14:15:01 -0300 Subject: [PATCH 5/7] Bugfix treat any OSError as a closed connection when writing or closing StreamWriter.drain() and wait_closed() re-raise whatever exception connection_lost() received. Only ConnectionError subclasses were handled, so an ssl.SSLError (application data after close notify), a TimeoutError (SSL shutdown timed out) or a plain OSError such as EHOSTUNREACH escaped protocol_send() and _close(). From an app task that becomes an ExceptionGroup, which TCPServer.run() does not catch, and the whole connection fails with "Unhandled exception in client_connected_cb". These cluster during the graceful_timeout drain, when many connections close at once against peers that may be gone. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/hypercorn/asyncio/tcp_server.py | 10 ++----- tests/asyncio/test_tcp_server.py | 41 +++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 8 deletions(-) diff --git a/src/hypercorn/asyncio/tcp_server.py b/src/hypercorn/asyncio/tcp_server.py index 5612214..453b5dd 100644 --- a/src/hypercorn/asyncio/tcp_server.py +++ b/src/hypercorn/asyncio/tcp_server.py @@ -82,7 +82,7 @@ async def protocol_send(self, event: Event) -> None: try: self.writer.write(event.data) await self.writer.drain() - except (ConnectionError, RuntimeError): + except (OSError, RuntimeError): await self.protocol.handle(Closed()) elif isinstance(event, Closed): await self._close() @@ -118,13 +118,7 @@ async def _close(self) -> None: try: self.writer.close() await self.writer.wait_closed() - except ( - BrokenPipeError, - ConnectionAbortedError, - ConnectionResetError, - RuntimeError, - asyncio.CancelledError, - ): + except (OSError, RuntimeError, asyncio.CancelledError): pass # Already closed finally: await self.idle_task.stop() diff --git a/tests/asyncio/test_tcp_server.py b/tests/asyncio/test_tcp_server.py index 1aa2898..719c4fd 100644 --- a/tests/asyncio/test_tcp_server.py +++ b/tests/asyncio/test_tcp_server.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +from unittest.mock import AsyncMock import pytest @@ -8,10 +9,50 @@ from hypercorn.asyncio.tcp_server import TCPServer from hypercorn.asyncio.worker_context import WorkerContext from hypercorn.config import Config +from hypercorn.events import Closed, RawData from .helpers import MemoryReader, MemoryWriter from ..helpers import echo_framework +class OSErrorOnDrainWriter(MemoryWriter): + async def drain(self) -> None: + raise OSError("connection_lost carried a non ConnectionError") + + +class OSErrorOnWaitClosedWriter(MemoryWriter): + async def wait_closed(self) -> None: + raise OSError("SSL shutdown timed out") + + +def _server(writer: MemoryWriter) -> TCPServer: + return TCPServer( + ASGIWrapper(echo_framework), + asyncio.get_running_loop(), + Config(), + WorkerContext(None), + {}, + MemoryReader(), # type: ignore + writer, # type: ignore + ) + + +@pytest.mark.asyncio +async def test_protocol_send_closes_on_oserror() -> None: + server = _server(OSErrorOnDrainWriter()) + server.protocol = AsyncMock() + + await server.protocol_send(RawData(data=b"data")) + + server.protocol.handle.assert_awaited_once_with(Closed()) + + +@pytest.mark.asyncio +async def test_close_ignores_oserror() -> None: + server = _server(OSErrorOnWaitClosedWriter()) + + await server._close() + + @pytest.mark.asyncio async def test_completes_on_closed() -> None: event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop() From 821ae470efe3866f6f6b0882868dab4752cfe9c6 Mon Sep 17 00:00:00 2001 From: emicuencac Date: Thu, 10 Sep 2026 14:15:59 -0300 Subject: [PATCH 6/7] Bugfix do not swallow the caller's cancellation in AsyncioSingleTask restart() and stop() cancel the previous idle task and await it, ignoring the CancelledError that the await raises. That await also raises CancelledError when the calling task is itself being cancelled, for example when the connection TaskGroup is torn down at graceful_timeout. Swallowing it let restart() continue and call create_task() on a TaskGroup that is already shutting down, which raises RuntimeError and fails the connection. Wait for the cancelled task with asyncio.wait(), which does not raise its CancelledError but still lets the caller's own cancellation through. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/hypercorn/asyncio/worker_context.py | 10 ++----- tests/asyncio/test_worker_context.py | 39 +++++++++++++++++++++++++ 2 files changed, 41 insertions(+), 8 deletions(-) create mode 100644 tests/asyncio/test_worker_context.py diff --git a/src/hypercorn/asyncio/worker_context.py b/src/hypercorn/asyncio/worker_context.py index 551ed40..27526fa 100644 --- a/src/hypercorn/asyncio/worker_context.py +++ b/src/hypercorn/asyncio/worker_context.py @@ -15,10 +15,7 @@ async def restart(self, task_group: TaskGroup, action: Callable) -> None: async with self._lock: if self._handle is not None: self._handle.cancel() - try: - await self._handle - except asyncio.CancelledError: - pass + await asyncio.wait({self._handle}) self._handle = task_group._task_group.create_task(action()) # type: ignore @@ -26,10 +23,7 @@ async def stop(self) -> None: async with self._lock: if self._handle is not None: self._handle.cancel() - try: - await self._handle - except asyncio.CancelledError: - pass + await asyncio.wait({self._handle}) self._handle = None diff --git a/tests/asyncio/test_worker_context.py b/tests/asyncio/test_worker_context.py new file mode 100644 index 0000000..36c8b52 --- /dev/null +++ b/tests/asyncio/test_worker_context.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from hypercorn.asyncio.task_group import TaskGroup +from hypercorn.asyncio.worker_context import AsyncioSingleTask + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method", ["restart", "stop"]) +async def test_single_task_propagates_caller_cancellation(method: str) -> None: + event_loop = asyncio.get_running_loop() + single_task = AsyncioSingleTask() + cancelled = asyncio.Event() + release = asyncio.Event() + + async def action() -> None: + try: + await asyncio.sleep(100) + finally: + cancelled.set() + await release.wait() + + async with TaskGroup(event_loop) as task_group: + await single_task.restart(task_group, action) + if method == "restart": + caller = event_loop.create_task(single_task.restart(task_group, action)) + else: + caller = event_loop.create_task(single_task.stop()) + await cancelled.wait() + caller.cancel() + release.set() + + with pytest.raises(asyncio.CancelledError): + await caller + + await single_task.stop() From 05c1567c53fdccb39ed7dbe98447cfbf678a9d06 Mon Sep 17 00:00:00 2001 From: emicuencac Date: Thu, 10 Sep 2026 14:17:22 -0300 Subject: [PATCH 7/7] Bugfix make HTTP/2 send cleanup tolerate an already removed stream When _send_data() hits a closed stream it closes and removes the stream's buffer and priority entry. If either is already gone, the cleanup itself raises a second KeyError or MissingStreamError, which kills the connection's send task and with it every stream on the connection. Resets are frequent while a terminated connection drains, so make the cleanup idempotent. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/hypercorn/protocol/h2.py | 9 ++++++--- tests/protocol/test_h2.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/src/hypercorn/protocol/h2.py b/src/hypercorn/protocol/h2.py index dbe2c0e..7b379fa 100644 --- a/src/hypercorn/protocol/h2.py +++ b/src/hypercorn/protocol/h2.py @@ -1,6 +1,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable +from contextlib import suppress import h2 import h2.connection @@ -175,9 +176,11 @@ async def _send_data(self, stream_id: int) -> None: except (h2.exceptions.StreamClosedError, KeyError, h2.exceptions.ProtocolError): # Stream or connection has closed whilst waiting to send # data, not a problem - just force close it. - await self.stream_buffers[stream_id].close() - del self.stream_buffers[stream_id] - self.priority.remove_stream(stream_id) + buffer = self.stream_buffers.pop(stream_id, None) + if buffer is not None: + await buffer.close() + with suppress(priority.MissingStreamError): + self.priority.remove_stream(stream_id) async def handle(self, event: Event) -> None: if isinstance(event, RawData): diff --git a/tests/protocol/test_h2.py b/tests/protocol/test_h2.py index ec19534..d67ca94 100644 --- a/tests/protocol/test_h2.py +++ b/tests/protocol/test_h2.py @@ -184,3 +184,33 @@ async def test_protocol_terminated_data_for_refused_stream() -> None: assert events[0].stream_id == 3 # The pre-existing stream is unaffected assert 1 in protocol.streams + + +@pytest.mark.asyncio +async def test_send_data_cleanup_without_stream_buffer() -> None: + protocol, _ = await _terminated_protocol_with_streams(1) + del protocol.stream_buffers[1] + + await protocol._send_data(1) + + assert 1 not in protocol.stream_buffers + + +@pytest.mark.asyncio +async def test_send_data_cleanup_without_priority_stream() -> None: + protocol = H2Protocol( + Mock(), + Config(), + WorkerContext(None), + AsyncMock(), + ConnectionState({}), + False, + None, + None, + AsyncMock(), + ) + protocol.stream_buffers[1] = StreamBuffer(EventWrapper) + + await protocol._send_data(1) + + assert 1 not in protocol.stream_buffers