Update vendored httpx-ws to upstream v0.9.0#1067
Conversation
The initial vendoring in #1042 accidentally used frankie567/httpx-ws@6bc568b (v0.6.2 era, October 2024), missing 80 upstream commits. This ports every upstream change from that point to the current tip, frankie567/httpx-ws@0341891 (v0.9.0 + 6 commits). See discussion #1066. Highlights: - Serialize stream writes with a write lock (httpx-ws#29) - Rebuild the async session on anyio.AsyncContextManagerMixin, fixing the asyncio hang on exit after cancellation and cancel scope corruption (httpx-ws#108); requires anyio>=4.10 - Rewrite ASGIWebSocketTransport on a task group and anyio streams instead of a blocking portal and queue.Queue - Support the WebSocket Denial Response ASGI extension - Add a configurable initial_receive_timeout with a clear error when the server never accepts - Raise TimeoutError instead of queue.Empty on sync receive timeout - Handle end of stream and closing-state keepalive pings properly - Add class-based WebSocketClient/AsyncWebSocketClient with session_class support (kept off connect_ws/aconnect_ws, where mypy cannot infer the TypeVar default through the contextmanager decorator) Two fixes not in upstream: register aclose before the initial receive so a never-accepting app does not leak memory object streams, and cover the double-__aenter__ guard with a test.
|
Docs preview: |
Merging this PR will not alter performance
Comparing 
Footnotes
|
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Advertise the denial-response extension
The transport now consumes websocket.http.response.* messages, but the ASGI scope built here never includes extensions={'websocket.http.response': {}}. ASGI apps or frameworks that check the scope before using the denial-response extension will conclude it is unsupported and won't send the response, so this only works for apps that emit raw extension messages unconditionally.
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| self.scope = scope | ||
| self.exit_stack = contextlib.AsyncExitStack() | ||
| stream, accept_response = await self.exit_stack.enter_async_context( | ||
| ASGIWebSocketAsyncNetworkStream(self.app, self.scope) # type: ignore[arg-type] | ||
| ) | ||
|
|
||
| stream, accept_response = await self._task_group.start(self._create_asgi_websocket_async_network_stream) |
There was a problem hiding this comment.
When two websocket requests share one ASGIWebSocketTransport, this stores per-request state on the transport and the task started on the next line reads self.scope later. A second request can overwrite it before the first _create_asgi_websocket_async_network_stream constructs its stream, so the first connection can be delivered to the wrong path, headers, or subprotocols. Pass scope into the started helper instead of using mutable transport state.
Useful? React with 👍 / 👎.
| ASGIWebSocketAsyncNetworkStream(self.app, self.scope) # type: ignore[arg-type] | ||
| ) | ||
|
|
||
| stream, accept_response = await self._task_group.start(self._create_asgi_websocket_async_network_stream) |
There was a problem hiding this comment.
This path now assumes __aenter__ has run and created self._task_group. If a caller uses the normal AsyncClient lifecycle without async with (create client, call aconnect_ws, then aclose()), the websocket request reaches this line with no _task_group attribute and fails before the app is called; the previous transport created its per-request stream state here. Either lazily create and close the task group or keep an aclose path for non-context-managed clients.
Useful? React with 👍 / 👎.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
5 issues found across 9 files
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="src/httpx2/httpx2/websockets/_transport.py">
<violation number="1" location="src/httpx2/httpx2/websockets/_transport.py:88">
P2: Denial response body collection loop lacks a receive timeout, so a server that starts a denial response but never completes the body will cause the client to hang indefinitely. Consider applying a timeout to the inner `receive()` calls (reusing `initial_receive_timeout` or a separate configurable value) so that a stalled denial-response body surfaces a `TimeoutError` / `RuntimeError` rather than hanging forever.</violation>
<violation number="2" location="src/httpx2/httpx2/websockets/_transport.py:91">
P2: Valid denial responses that omit `body` (for example, a header-only response) now fail with `KeyError` instead of producing a `WebSocketUpgradeError`. Reading the field with an empty-bytes default preserves the ASGI behavior.</violation>
<violation number="3" location="src/httpx2/httpx2/websockets/_transport.py:221">
P2: Calling `await client.aclose()` leaves the transport task group and all active WebSocket stream queues open, so applications that close clients explicitly leak connection resources. A shared, idempotent `aclose()` implementation should close `_exit_stack`, with `__aexit__` preserving exception-aware exit handling.</violation>
<violation number="4" location="src/httpx2/httpx2/websockets/_transport.py:273">
P2: Using `ASGIWebSocketTransport` with the supported `client = AsyncClient(...); await client.get(...)` pattern now raises `AttributeError` because `_task_group` has not been initialized. The transport needs a lazy/explicit lifecycle path that works for requests made without first entering the transport context.</violation>
</file>
<file name="tests/httpx2/websockets/test_api.py">
<violation number="1" location="tests/httpx2/websockets/test_api.py:855">
P3: The sync portion of `test_concurrency_write` submits `send_text` calls to a `ThreadPoolExecutor` but never checks the returned futures for exceptions. If any concurrent write fails, the exception is captured in the Future but silently discarded when the executor shuts down, so the test would pass despite a failure. The async half uses `anyio.create_task_group()` which correctly propagates exceptions; the sync half should either collect and `.result()` on each future, or use a mechanism that surfaces failures.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| exc_val: BaseException | None = None, | ||
| exc_tb: TracebackType | None = None, | ||
| ) -> None: | ||
| await super().__aexit__(exc_type, exc_val, exc_tb) |
There was a problem hiding this comment.
P2: Calling await client.aclose() leaves the transport task group and all active WebSocket stream queues open, so applications that close clients explicitly leak connection resources. A shared, idempotent aclose() implementation should close _exit_stack, with __aexit__ preserving exception-aware exit handling.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/websockets/_transport.py, line 221:
<comment>Calling `await client.aclose()` leaves the transport task group and all active WebSocket stream queues open, so applications that close clients explicitly leak connection resources. A shared, idempotent `aclose()` implementation should close `_exit_stack`, with `__aexit__` preserving exception-aware exit handling.</comment>
<file context>
@@ -159,9 +199,28 @@ def __init__(
+ exc_val: BaseException | None = None,
+ exc_tb: TracebackType | None = None,
+ ) -> None:
+ await super().__aexit__(exc_type, exc_val, exc_tb)
+ assert self._exit_stack is not None
+ await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)
</file context>
| while True: | ||
| message = await self.receive() | ||
| assert message["type"] == "websocket.http.response.body" | ||
| body.append(message["body"]) |
There was a problem hiding this comment.
P2: Valid denial responses that omit body (for example, a header-only response) now fail with KeyError instead of producing a WebSocketUpgradeError. Reading the field with an empty-bytes default preserves the ASGI behavior.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/websockets/_transport.py, line 91:
<comment>Valid denial responses that omit `body` (for example, a header-only response) now fail with `KeyError` instead of producing a `WebSocketUpgradeError`. Reading the field with an empty-bytes default preserves the ASGI behavior.</comment>
<file context>
@@ -36,33 +36,77 @@ def __init__(self, event: wsproto.events.Event) -> None:
+ while True:
+ message = await self.receive()
+ assert message["type"] == "websocket.http.response.body"
+ body.append(message["body"])
+ if not message.get("more_body", False):
+ break
</file context>
| body.append(message["body"]) | |
| body.append(message.get("body", b"")) |
| ASGIWebSocketAsyncNetworkStream(self.app, self.scope) # type: ignore[arg-type] | ||
| ) | ||
|
|
||
| stream, accept_response = await self._task_group.start(self._create_asgi_websocket_async_network_stream) |
There was a problem hiding this comment.
P2: Using ASGIWebSocketTransport with the supported client = AsyncClient(...); await client.get(...) pattern now raises AttributeError because _task_group has not been initialized. The transport needs a lazy/explicit lifecycle path that works for requests made without first entering the transport context.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/websockets/_transport.py, line 273:
<comment>Using `ASGIWebSocketTransport` with the supported `client = AsyncClient(...); await client.get(...)` pattern now raises `AttributeError` because `_task_group` has not been initialized. The transport needs a lazy/explicit lifecycle path that works for requests made without first entering the transport context.</comment>
<file context>
@@ -196,11 +270,7 @@ async def _handle_ws_request(
- ASGIWebSocketAsyncNetworkStream(self.app, self.scope) # type: ignore[arg-type]
- )
-
+ stream, accept_response = await self._task_group.start(self._create_asgi_websocket_async_network_stream)
accept_response_lines = accept_response.decode("utf-8").splitlines()
headers = [
</file context>
| status_code: int = message["status"] | ||
| headers: list[tuple[bytes, bytes]] = message["headers"] | ||
| body: list[bytes] = [] | ||
| while True: |
There was a problem hiding this comment.
P2: Denial response body collection loop lacks a receive timeout, so a server that starts a denial response but never completes the body will cause the client to hang indefinitely. Consider applying a timeout to the inner receive() calls (reusing initial_receive_timeout or a separate configurable value) so that a stalled denial-response body surfaces a TimeoutError / RuntimeError rather than hanging forever.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/websockets/_transport.py, line 88:
<comment>Denial response body collection loop lacks a receive timeout, so a server that starts a denial response but never completes the body will cause the client to hang indefinitely. Consider applying a timeout to the inner `receive()` calls (reusing `initial_receive_timeout` or a separate configurable value) so that a stalled denial-response body surfaces a `TimeoutError` / `RuntimeError` rather than hanging forever.</comment>
<file context>
@@ -36,33 +36,77 @@ def __init__(self, event: wsproto.events.Event) -> None:
+ status_code: int = message["status"]
+ headers: list[tuple[bytes, bytes]] = message["headers"]
+ body: list[bytes] = []
+ while True:
+ message = await self.receive()
+ assert message["type"] == "websocket.http.response.body"
</file context>
| # Added for completeness, but were not able to reproduce the issue with the sync client | ||
| with httpx.Client(transport=httpx.HTTPTransport(uds=socket)) as client: | ||
| with connect_ws("http://socket/ws", client) as ws: | ||
| with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: |
There was a problem hiding this comment.
P3: The sync portion of test_concurrency_write submits send_text calls to a ThreadPoolExecutor but never checks the returned futures for exceptions. If any concurrent write fails, the exception is captured in the Future but silently discarded when the executor shuts down, so the test would pass despite a failure. The async half uses anyio.create_task_group() which correctly propagates exceptions; the sync half should either collect and .result() on each future, or use a mechanism that surfaces failures.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/httpx2/websockets/test_api.py, line 855:
<comment>The sync portion of `test_concurrency_write` submits `send_text` calls to a `ThreadPoolExecutor` but never checks the returned futures for exceptions. If any concurrent write fails, the exception is captured in the Future but silently discarded when the executor shuts down, so the test would pass despite a failure. The async half uses `anyio.create_task_group()` which correctly propagates exceptions; the sync half should either collect and `.result()` on each future, or use a mechanism that surfaces failures.</comment>
<file context>
@@ -838,3 +830,55 @@ def wait_for_session_threads(expected: int) -> None:
+ # Added for completeness, but were not able to reproduce the issue with the sync client
+ with httpx.Client(transport=httpx.HTTPTransport(uds=socket)) as client:
+ with connect_ws("http://socket/ws", client) as ws:
+ with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
+ for _ in range(10):
+ executor.submit(ws.send_text, "CLIENT_MESSAGE")
</file context>
1 participant
Closes the gap reported in #1066: the vendoring in #1042 accidentally used frankie567/httpx-ws@6bc568b (v0.6.2 era, October 2024). This ports every upstream change from there to the current tip, frankie567/httpx-ws@0341891 (v0.9.0 + 6 commits) - 80 commits, of which 38 touch library or test code; the rest are upstream CI/docs/tooling.
What comes in
anyio.AsyncContextManagerMixin, fixing the asyncio hang on exit after cancellation and cancel scope corruption (httpx-ws#108);anyio>=4.10is now requiredASGIWebSocketTransportrewritten on a task group and anyio streams instead of a blocking portal andqueue.Queueinitial_receive_timeoutwith a clear error when the server never callsaccept()TimeoutErrorinstead ofqueue.Emptyon sync receive timeout (docs updated)WebSocketClient/AsyncWebSocketClientwithsession_classsupport, exported fromhttpx2.websocketsDeliberate deviations from upstream
httpcore2imports, explicit handshake parameters instead of**kwargs,Client.websocket()as the public entry point (connect_ws/aconnect_wsstay unexported)session_classlives only on the client classes, not onconnect_ws/aconnect_ws: mypy cannot infer the PEP 696 TypeVar default through thecontextmanagerdecoratoracloseis registered before the initial receive so a never-accepting app does not leak memory object streams, and the double-__aenter__guard is covered by a testVerification
websockets/files; ruff format, ruff check, mypy strict, unasync all cleanAI Disclaimer
This PR was developed with the assistance of either Claude or Codex. I've reviewed and verified the changes.