From 73aed0edf1ceb158fe5bb144e7add636043e37e1 Mon Sep 17 00:00:00 2001 From: "appunni.m" <31534711+appunni-m@users.noreply.github.com> Date: Sat, 3 Oct 2026 22:26:11 +0530 Subject: [PATCH 1/3] Fix HTTP/3 ASGI disconnect delivery --- docs/discussion/closing.rst | 5 +++ src/hypercorn/protocol/h3.py | 15 ++++++- tests/protocol/test_h3.py | 77 ++++++++++++++++++++++++++++++++++++ tox.ini | 1 + 4 files changed, 97 insertions(+), 1 deletion(-) create mode 100644 tests/protocol/test_h3.py diff --git a/docs/discussion/closing.rst b/docs/discussion/closing.rst index 23788415..6d35b2b3 100644 --- a/docs/discussion/closing.rst +++ b/docs/discussion/closing.rst @@ -23,6 +23,11 @@ raise an exception. This exception is caught and a Closed event is sent to the protocol. The protocol should then send each stream a StreamClosed event and delete the stream. +For HTTP/3, a QUIC stream reset closes the affected request stream. A +QUIC connection termination closes each active request stream. Both +events are passed to the stream so the ASGI application receives its +``http.disconnect`` message. + Server disconnection -------------------- diff --git a/src/hypercorn/protocol/h3.py b/src/hypercorn/protocol/h3.py index 1bcf584e..b91c0325 100644 --- a/src/hypercorn/protocol/h3.py +++ b/src/hypercorn/protocol/h3.py @@ -6,7 +6,7 @@ from aioquic.h3.events import DataReceived, HeadersReceived from aioquic.h3.exceptions import NoAvailablePushIDError from aioquic.quic.connection import QuicConnection -from aioquic.quic.events import QuicEvent +from aioquic.quic.events import ConnectionTerminated, QuicEvent, StreamReset from .events import ( Body, @@ -52,6 +52,14 @@ def __init__( self.state = state async def handle(self, quic_event: QuicEvent) -> None: + if isinstance(quic_event, ConnectionTerminated): + for stream_id in list(self.streams): + await self._close_stream(stream_id) + return + elif isinstance(quic_event, StreamReset): + await self._close_stream(quic_event.stream_id) + return + for event in self.connection.handle_event(quic_event): if isinstance(event, HeadersReceived): if not self.context.terminated.is_set(): @@ -67,6 +75,11 @@ async def handle(self, quic_event: QuicEvent) -> None: if event.stream_ended: await self.streams[event.stream_id].handle(EndBody(stream_id=event.stream_id)) + async def _close_stream(self, stream_id: int) -> None: + stream = self.streams.pop(stream_id, None) + if stream is not None: + await stream.handle(StreamClosed(stream_id=stream_id)) + async def stream_send(self, event: StreamEvent) -> None: if isinstance(event, (InformationalResponse, Response)): self.connection.send_headers( diff --git a/tests/protocol/test_h3.py b/tests/protocol/test_h3.py new file mode 100644 index 00000000..222f057b --- /dev/null +++ b/tests/protocol/test_h3.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import importlib +from typing import Any +from unittest.mock import AsyncMock, Mock + +import pytest + +from hypercorn.asyncio.worker_context import WorkerContext +from hypercorn.config import Config +from hypercorn.typing import ConnectionState + + +async def _protocol_with_requests( + stream_ids: list[int], +) -> tuple[Any, dict[int, AsyncMock]]: + pytest.importorskip("aioquic") + h3_events = importlib.import_module("aioquic.h3.events") + h3_protocol = importlib.import_module("hypercorn.protocol.h3") + config = Config() + config._log = AsyncMock() + app_receivers = {stream_id: AsyncMock() for stream_id in stream_ids} + task_group = Mock() + task_group.spawn_app = AsyncMock(side_effect=list(app_receivers.values())) + protocol = h3_protocol.H3Protocol( + Mock(), + config, + WorkerContext(None), + task_group, + ConnectionState({}), + None, + None, + Mock(), + AsyncMock(), + ) + for stream_id in stream_ids: + await protocol._create_stream( + h3_events.HeadersReceived( + stream_id=stream_id, + stream_ended=False, + headers=[ + (b":method", b"POST"), + (b":path", f"/upload/{stream_id}".encode()), + (b":scheme", b"https"), + (b":authority", b"hypercorn"), + ], + ) + ) + + return protocol, app_receivers + + +@pytest.mark.asyncio +async def test_stream_reset_reaches_asgi_as_http_disconnect() -> None: + protocol, app_receivers = await _protocol_with_requests([0]) + quic_events = importlib.import_module("aioquic.quic.events") + + await protocol.handle(quic_events.StreamReset(error_code=0, stream_id=0)) + + app_receivers[0].assert_awaited_once_with({"type": "http.disconnect"}) + assert protocol.streams == {} + + +@pytest.mark.asyncio +async def test_connection_termination_disconnects_all_active_requests() -> None: + protocol, app_receivers = await _protocol_with_requests([0, 4]) + quic_events = importlib.import_module("aioquic.quic.events") + + await protocol.handle( + quic_events.ConnectionTerminated( + error_code=0, frame_type=None, reason_phrase="client closed" + ) + ) + + for receiver in app_receivers.values(): + receiver.assert_awaited_once_with({"type": "http.disconnect"}) + assert protocol.streams == {} diff --git a/tox.ini b/tox.ini index d4807aba..d86e1034 100644 --- a/tox.ini +++ b/tox.ini @@ -11,6 +11,7 @@ deps = pytest-cov pytest-sugar pytest-trio + py312: aioquic commands = pytest --cov=hypercorn {posargs} [testenv:docs] From 555ed38cf5211e0c62b3c27336795f1ef81ef402 Mon Sep 17 00:00:00 2001 From: "appunni.m" <31534711+appunni-m@users.noreply.github.com> Date: Sun, 4 Oct 2026 10:58:35 +0530 Subject: [PATCH 2/3] Handle HTTP/3 response cancellation disconnects --- src/hypercorn/protocol/h3.py | 6 +- tests/protocol/test_h3.py | 244 +++++++++++++++++++++++++++++++++-- 2 files changed, 239 insertions(+), 11 deletions(-) diff --git a/src/hypercorn/protocol/h3.py b/src/hypercorn/protocol/h3.py index b91c0325..8f0c654b 100644 --- a/src/hypercorn/protocol/h3.py +++ b/src/hypercorn/protocol/h3.py @@ -6,7 +6,7 @@ from aioquic.h3.events import DataReceived, HeadersReceived from aioquic.h3.exceptions import NoAvailablePushIDError from aioquic.quic.connection import QuicConnection -from aioquic.quic.events import ConnectionTerminated, QuicEvent, StreamReset +from aioquic.quic.events import ConnectionTerminated, QuicEvent, StopSendingReceived, StreamReset from .events import ( Body, @@ -56,7 +56,7 @@ async def handle(self, quic_event: QuicEvent) -> None: for stream_id in list(self.streams): await self._close_stream(stream_id) return - elif isinstance(quic_event, StreamReset): + elif isinstance(quic_event, (StreamReset, StopSendingReceived)): await self._close_stream(quic_event.stream_id) return @@ -99,7 +99,7 @@ async def stream_send(self, event: StreamEvent) -> None: self.connection.send_headers(event.stream_id, event.headers) await self.send() elif isinstance(event, StreamClosed): - self.streams.pop(event.stream_id, None) + await self._close_stream(event.stream_id) elif isinstance(event, Request): await self._create_server_push(event.stream_id, event.raw_path, event.headers) diff --git a/tests/protocol/test_h3.py b/tests/protocol/test_h3.py index 222f057b..0d909834 100644 --- a/tests/protocol/test_h3.py +++ b/tests/protocol/test_h3.py @@ -1,18 +1,23 @@ from __future__ import annotations +import asyncio import importlib +import ssl +from pathlib import Path from typing import Any -from unittest.mock import AsyncMock, Mock +from unittest.mock import AsyncMock, call, Mock import pytest +from hypercorn.asyncio.task_group import TaskGroup from hypercorn.asyncio.worker_context import WorkerContext from hypercorn.config import Config +from hypercorn.protocol.events import StreamClosed from hypercorn.typing import ConnectionState async def _protocol_with_requests( - stream_ids: list[int], + stream_ids: list[int], *, websocket: bool = False ) -> tuple[Any, dict[int, AsyncMock]]: pytest.importorskip("aioquic") h3_events = importlib.import_module("aioquic.h3.events") @@ -34,16 +39,25 @@ async def _protocol_with_requests( AsyncMock(), ) for stream_id in stream_ids: + method = b"CONNECT" if websocket else b"POST" + headers = [ + (b":method", method), + (b":path", f"/upload/{stream_id}".encode()), + (b":scheme", b"https"), + (b":authority", b"hypercorn"), + ] + if websocket: + headers.extend( + [ + (b":protocol", b"websocket"), + (b"sec-websocket-version", b"13"), + ] + ) await protocol._create_stream( h3_events.HeadersReceived( stream_id=stream_id, stream_ended=False, - headers=[ - (b":method", b"POST"), - (b":path", f"/upload/{stream_id}".encode()), - (b":scheme", b"https"), - (b":authority", b"hypercorn"), - ], + headers=headers, ) ) @@ -75,3 +89,217 @@ async def test_connection_termination_disconnects_all_active_requests() -> None: for receiver in app_receivers.values(): receiver.assert_awaited_once_with({"type": "http.disconnect"}) assert protocol.streams == {} + + +@pytest.mark.asyncio +async def test_stream_reset_reaches_asgi_as_websocket_disconnect() -> None: + protocol, app_receivers = await _protocol_with_requests([0], websocket=True) + quic_events = importlib.import_module("aioquic.quic.events") + + await protocol.handle(quic_events.StreamReset(error_code=0, stream_id=0)) + + assert app_receivers[0].await_args_list == [ + call({"type": "websocket.connect"}), + call({"type": "websocket.disconnect", "code": 1006}), + ] + assert protocol.streams == {} + + +@pytest.mark.asyncio +async def test_stream_send_closed_reaches_asgi_as_websocket_disconnect() -> None: + protocol, app_receivers = await _protocol_with_requests([0], websocket=True) + + await protocol.stream_send(StreamClosed(stream_id=0)) + + assert app_receivers[0].await_args_list == [ + call({"type": "websocket.connect"}), + call({"type": "websocket.disconnect", "code": 1006}), + ] + assert protocol.streams == {} + + +@pytest.mark.asyncio +async def test_stream_reset_then_connection_termination_disconnects_once() -> None: + protocol, app_receivers = await _protocol_with_requests([0, 4]) + quic_events = importlib.import_module("aioquic.quic.events") + + await protocol.handle(quic_events.StreamReset(error_code=0, stream_id=0)) + await protocol.handle( + quic_events.ConnectionTerminated( + error_code=0, frame_type=None, reason_phrase="client closed" + ) + ) + + for receiver in app_receivers.values(): + receiver.assert_awaited_once_with({"type": "http.disconnect"}) + assert protocol.streams == {} + + +@pytest.mark.parametrize( + "stop_sending", [False, True], ids=["after-final-response-body", "h3-request-cancelled"] +) +@pytest.mark.asyncio +async def test_http3_disconnect_unblocks_application_waiting_for_receive( + stop_sending: bool, +) -> None: + pytest.importorskip("aioquic") + aioquic_buffer = importlib.import_module("aioquic.buffer") + h3_connection = importlib.import_module("aioquic.h3.connection") + h3_events = importlib.import_module("aioquic.h3.events") + quic_configuration = importlib.import_module("aioquic.quic.configuration") + quic_connection = importlib.import_module("aioquic.quic.connection") + quic_events = importlib.import_module("aioquic.quic.events") + quic_packet = importlib.import_module("aioquic.quic.packet") + h3_protocol = importlib.import_module("hypercorn.protocol.h3") + + response_started = asyncio.Event() + app_finished = {path: asyncio.Event() for path in ("/events", "/health")} + app_disconnects: dict[str, dict[str, Any]] = {} + + async def app(scope: Any, receive: Any, send: Any, sync_spawn: Any, call_soon: Any) -> None: + assert await receive() == { + "type": "http.request", + "body": b"", + "more_body": False, + } + await send({"type": "http.response.start", "status": 200, "headers": []}) + is_streaming_cancelled_response = stop_sending and scope["path"] == "/events" + await send( + { + "type": "http.response.body", + "body": b"first", + "more_body": is_streaming_cancelled_response, + } + ) + if is_streaming_cancelled_response: + response_started.set() + app_disconnects[scope["path"]] = await receive() + app_finished[scope["path"]].set() + + config = Config() + config.certfile = str(Path(__file__).parents[1] / "assets" / "cert.pem") + config.keyfile = str(Path(__file__).parents[1] / "assets" / "key.pem") + config._log = AsyncMock() + server_config = quic_configuration.QuicConfiguration( + is_client=False, alpn_protocols=h3_connection.H3_ALPN + ) + server_config.load_cert_chain(config.certfile, config.keyfile) + client_config = quic_configuration.QuicConfiguration( + is_client=True, alpn_protocols=h3_connection.H3_ALPN + ) + client_config.verify_mode = ssl.CERT_NONE + + loop = asyncio.get_running_loop() + server_address = ("127.0.0.1", 4433) + client_address = ("127.0.0.1", 45678) + client_quic = quic_connection.QuicConnection(configuration=client_config) + client_quic.connect(server_address, now=loop.time()) + client_h3 = None + client_h3_events: list[Any] = [] + server_quic = None + server_h3 = None + task_group = TaskGroup(loop) + + async def send_server() -> None: + return None + + async def pump_packets() -> None: + nonlocal client_h3, server_h3, server_quic + idle_rounds = 0 + for _ in range(100): + progressed = False + for data, _ in client_quic.datagrams_to_send(now=loop.time()): + progressed = True + if server_quic is None: + header = quic_packet.pull_quic_header( + aioquic_buffer.Buffer(data=data), host_cid_length=8 + ) + server_quic = quic_connection.QuicConnection( + configuration=server_config, + original_destination_connection_id=header.destination_cid, + ) + server_quic.receive_datagram(data, client_address, now=loop.time()) + while (event := server_quic.next_event()) is not None: + if isinstance(event, quic_events.ProtocolNegotiated) and server_h3 is None: + server_h3 = h3_protocol.H3Protocol( + app, + config, + WorkerContext(None), + task_group, + ConnectionState({}), + None, + None, + server_quic, + send_server, + ) + if server_h3 is not None: + await server_h3.handle(event) + + if server_quic is not None: + for data, _ in server_quic.datagrams_to_send(now=loop.time()): + progressed = True + client_quic.receive_datagram(data, server_address, now=loop.time()) + while (event := client_quic.next_event()) is not None: + if isinstance(event, quic_events.ProtocolNegotiated) and client_h3 is None: + client_h3 = h3_connection.H3Connection(client_quic) + if client_h3 is not None: + client_h3_events.extend(client_h3.handle_event(event)) + + await asyncio.sleep(0) + if progressed: + idle_rounds = 0 + else: + idle_rounds += 1 + if idle_rounds == 2: + return + + raise AssertionError("HTTP/3 packet exchange did not become idle") + + async with task_group: + await pump_packets() + assert client_h3 is not None + + for stream_id, path in ((0, "/events"), (4, "/health")): + client_h3.send_headers( + stream_id, + [ + (b":method", b"GET"), + (b":scheme", b"https"), + (b":authority", b"hypercorn"), + (b":path", path.encode("ascii")), + ], + end_stream=True, + ) + await pump_packets() + if stream_id == 0 and stop_sending: + await asyncio.wait_for(response_started.wait(), timeout=1) + client_quic.stop_stream( + stream_id, int(h3_connection.ErrorCode.H3_REQUEST_CANCELLED) + ) + await pump_packets() + await asyncio.wait_for(app_finished[path].wait(), timeout=1) + + assert app_disconnects == { + "/events": {"type": "http.disconnect"}, + "/health": {"type": "http.disconnect"}, + } + assert server_h3 is not None + assert server_h3.streams == {} + + response_headers = [ + event for event in client_h3_events if isinstance(event, h3_events.HeadersReceived) + ] + response_bodies = [ + event for event in client_h3_events if isinstance(event, h3_events.DataReceived) + ] + for stream_id in (0, 4): + assert any( + event.stream_id == stream_id and (b":status", b"200") in event.headers + for event in response_headers + ) + assert any( + event.stream_id == stream_id and event.data == b"first" for event in response_bodies + ) + + client_quic.close() + await pump_packets() From 05716d4d627947dad73e5cfae5d2924936e3fea9 Mon Sep 17 00:00:00 2001 From: "appunni.m" <31534711+appunni-m@users.noreply.github.com> Date: Sun, 4 Oct 2026 11:14:10 +0530 Subject: [PATCH 3/3] test: run HTTP/3 tests across Python matrix --- tests/asyncio/test_sanity.py | 5 +++-- tests/trio/test_sanity.py | 5 +++-- tox.ini | 2 +- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/tests/asyncio/test_sanity.py b/tests/asyncio/test_sanity.py index 0d1d0f90..2cd5d342 100644 --- a/tests/asyncio/test_sanity.py +++ b/tests/asyncio/test_sanity.py @@ -203,9 +203,10 @@ async def test_http2_websocket() -> None: stream_id, [ (":method", "CONNECT"), - (":path", "/"), - (":authority", "hypercorn"), + (":protocol", "websocket"), (":scheme", "https"), + (":authority", "hypercorn"), + (":path", "/"), ("sec-websocket-version", "13"), ], ) diff --git a/tests/trio/test_sanity.py b/tests/trio/test_sanity.py index 98387a24..915393e6 100644 --- a/tests/trio/test_sanity.py +++ b/tests/trio/test_sanity.py @@ -175,9 +175,10 @@ async def test_http2_websocket(nursery: trio._core._run.Nursery) -> None: stream_id, [ (":method", "CONNECT"), - (":path", "/"), - (":authority", "hypercorn"), + (":protocol", "websocket"), (":scheme", "https"), + (":authority", "hypercorn"), + (":path", "/"), ("sec-websocket-version", "13"), ], ) diff --git a/tox.ini b/tox.ini index d86e1034..e794a871 100644 --- a/tox.ini +++ b/tox.ini @@ -11,7 +11,7 @@ deps = pytest-cov pytest-sugar pytest-trio - py312: aioquic + aioquic commands = pytest --cov=hypercorn {posargs} [testenv:docs]