Add httpunk-backed async transport - #1089
Conversation
|
Docs preview: https://42fb959d-httpx2-docs.pydantic.workers.dev |
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
Here are some automated review suggestions for this pull request.
Reviewed commit: 6c16227c35
ℹ️ 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".
| @@ -0,0 +1,190 @@ | |||
| from __future__ import annotations | |||
There was a problem hiding this comment.
The reviewed commit records Codex <codex@openai.com> as its Author rather than the engineer, contrary to the repository requirement that coding-agent work retain engineer-only authorship; recreate the commit with the engineer as its sole author.
AGENTS.md reference: AGENTS.md:L3-L3
Useful? React with 👍 / 👎.
|
|
||
| import httpx2 | ||
|
|
||
| pytest.importorskip("httpunk") |
There was a problem hiding this comment.
Checked .github/workflows/main.yml: each test job runs scripts/install, which executes uv sync --frozen, but the root dev group requests the httpx2 extras without httpunk. Consequently this module-level skip fires on every CI Python version, none of these tests run, and the imported httpunk.py remains largely uncovered before scripts/coverage enforces 100% coverage. Include the extra in the development/CI environment rather than silently skipping the new transport tests.
Useful? React with 👍 / 👎.
| pool.count += 1 | ||
| connection = await self._connect(origin) |
There was a problem hiding this comment.
With max_connections=1, or once every slot has encountered a transient DNS, TCP, or TLS failure, _connect raises after pool.count is incremented and the count is never restored. The next request then sees the pool at capacity and waits forever on an empty idle queue. Roll back the reserved slot on every setup failure or cancellation.
Useful? React with 👍 / 👎.
| return | ||
| with contextlib.suppress(ValueError): | ||
| pool.all_connections.remove(connection) | ||
| pool.count -= 1 |
There was a problem hiding this comment.
When the pool is saturated and another request is already blocked in pool.idle.get(), an error on the active request calls _discard and only decrements the count. The queued request is never notified that capacity is now available, so it can hang indefinitely even though pool.count < pool.max_connections; use a condition/semaphore or otherwise wake the waiter so it can create a replacement connection.
Useful? React with 👍 / 👎.
| _transport, protocol = await loop.create_connection( | ||
| lambda: httpunk.asyncio.H1ClientProtocol(authority=authority), | ||
| host, | ||
| port, | ||
| ssl=ssl_context, |
There was a problem hiding this comment.
When DNS resolution, TCP connection, or TLS negotiation fails, loop.create_connection raises raw exceptions such as OSError or ssl.SSLError. Because this path performs no exception translation, callers using the documented except httpx2.RequestError hierarchy will not catch these transport failures. Convert setup failures to httpx2.ConnectError or the appropriate transport exception.
Useful? React with 👍 / 👎.
| connection = await self._acquire(origin) | ||
|
|
||
| target = request.url.raw_path.decode("ascii") or "/" | ||
| body = await request.aread() |
There was a problem hiding this comment.
When an asynchronous upload stream raises or the task is cancelled during request.aread(), the connection has already been acquired but this await occurs outside the cleanup block. That connection remains counted and is never returned to the idle queue, so a small pool can block all subsequent requests. Read the body before acquiring a connection or include this operation in cleanup that releases the slot.
Useful? React with 👍 / 👎.
|
I think we may want httpx2-httpunk instead of being in the transports path. |
![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 6 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/_transports/httpunk.py">
<violation number="1" location="src/httpx2/httpx2/_transports/httpunk.py:94">
P1: `request.aread()` is called after acquiring the connection but before the `try/except` block that calls `_discard`. If the body read raises (e.g. stream error or cancellation), the acquired connection is never released or discarded, permanently leaking a pool slot. Move `aread()` before `_acquire`, or include it inside the try block.</violation>
<violation number="2" location="src/httpx2/httpx2/_transports/httpunk.py:127">
P1: If `_connect` raises (e.g. DNS resolution failure, TCP timeout, TLS error), `pool.count` has already been incremented but is never decremented. With `max_connections=1` this permanently exhausts the pool, causing all subsequent requests to hang indefinitely on `pool.idle.get()`. Wrap the connect call in a try/except that decrements `pool.count` on failure.</violation>
<violation number="3" location="src/httpx2/httpx2/_transports/httpunk.py:131">
P2: When the connection pool is exhausted (all `max_connections` in use), `pool.idle.get()` blocks indefinitely without any timeout. Unlike the default transport which supports pool timeouts via the `"timeout"` extension, this transport offers no back-pressure or timeout mechanism, so callers can hang forever waiting for a connection to become available. Consider adding an `asyncio.wait_for` around the `get()` call, or reading a timeout from `request.extensions`.</violation>
<violation number="4" location="src/httpx2/httpx2/_transports/httpunk.py:145">
P1: `loop.create_connection` raises raw `OSError` or `ssl.SSLError` on DNS/TCP/TLS failures. These are not mapped to `httpx2.ConnectError`, so callers using `except httpx2.RequestError` (or `except httpx2.ConnectError`) won't catch transport setup failures. Wrap this call and translate connection-related exceptions to `ConnectError`.</violation>
<violation number="5" location="src/httpx2/httpx2/_transports/httpunk.py:182">
P1: After decrementing `pool.count`, tasks blocked on `await pool.idle.get()` in `_acquire` are never woken. When the pool is saturated and a connection error triggers `_discard`, waiters hang indefinitely despite capacity becoming available. A sentinel value should be placed on the queue, or a semaphore/condition should be used to notify waiters that they can create a new connection.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| connection = await self._connect(origin) | ||
| pool.all_connections.append(connection) | ||
| return connection |
There was a problem hiding this comment.
P1: If _connect raises (e.g. DNS resolution failure, TCP timeout, TLS error), pool.count has already been incremented but is never decremented. With max_connections=1 this permanently exhausts the pool, causing all subsequent requests to hang indefinitely on pool.idle.get(). Wrap the connect call in a try/except that decrements pool.count on failure.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/_transports/httpunk.py, line 127:
<comment>If `_connect` raises (e.g. DNS resolution failure, TCP timeout, TLS error), `pool.count` has already been incremented but is never decremented. With `max_connections=1` this permanently exhausts the pool, causing all subsequent requests to hang indefinitely on `pool.idle.get()`. Wrap the connect call in a try/except that decrements `pool.count` on failure.</comment>
<file context>
@@ -0,0 +1,190 @@
+ async with pool.lock:
+ if pool.count < pool.max_connections:
+ pool.count += 1
+ connection = await self._connect(origin)
+ pool.all_connections.append(connection)
+ return connection
</file context>
| connection = await self._connect(origin) | |
| pool.all_connections.append(connection) | |
| return connection | |
| try: | |
| connection = await self._connect(origin) | |
| except BaseException: | |
| pool.count -= 1 | |
| raise | |
| pool.all_connections.append(connection) | |
| return connection |
| connection = await self._acquire(origin) | ||
|
|
||
| target = request.url.raw_path.decode("ascii") or "/" | ||
| body = await request.aread() |
There was a problem hiding this comment.
P1: request.aread() is called after acquiring the connection but before the try/except block that calls _discard. If the body read raises (e.g. stream error or cancellation), the acquired connection is never released or discarded, permanently leaking a pool slot. Move aread() before _acquire, or include it inside the try block.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/_transports/httpunk.py, line 94:
<comment>`request.aread()` is called after acquiring the connection but before the `try/except` block that calls `_discard`. If the body read raises (e.g. stream error or cancellation), the acquired connection is never released or discarded, permanently leaking a pool slot. Move `aread()` before `_acquire`, or include it inside the try block.</comment>
<file context>
@@ -0,0 +1,190 @@
+ connection = await self._acquire(origin)
+
+ target = request.url.raw_path.decode("ascii") or "/"
+ body = await request.aread()
+
+ try:
</file context>
| _transport, protocol = await loop.create_connection( | ||
| lambda: httpunk.asyncio.H1ClientProtocol(authority=authority), | ||
| host, | ||
| port, | ||
| ssl=ssl_context, | ||
| ) |
There was a problem hiding this comment.
P1: loop.create_connection raises raw OSError or ssl.SSLError on DNS/TCP/TLS failures. These are not mapped to httpx2.ConnectError, so callers using except httpx2.RequestError (or except httpx2.ConnectError) won't catch transport setup failures. Wrap this call and translate connection-related exceptions to ConnectError.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/_transports/httpunk.py, line 145:
<comment>`loop.create_connection` raises raw `OSError` or `ssl.SSLError` on DNS/TCP/TLS failures. These are not mapped to `httpx2.ConnectError`, so callers using `except httpx2.RequestError` (or `except httpx2.ConnectError`) won't catch transport setup failures. Wrap this call and translate connection-related exceptions to `ConnectError`.</comment>
<file context>
@@ -0,0 +1,190 @@
+ authority = f"{host}:{port}"
+
+ loop = asyncio.get_running_loop()
+ _transport, protocol = await loop.create_connection(
+ lambda: httpunk.asyncio.H1ClientProtocol(authority=authority),
+ host,
</file context>
| _transport, protocol = await loop.create_connection( | |
| lambda: httpunk.asyncio.H1ClientProtocol(authority=authority), | |
| host, | |
| port, | |
| ssl=ssl_context, | |
| ) | |
| try: | |
| _transport, protocol = await loop.create_connection( | |
| lambda: httpunk.asyncio.H1ClientProtocol(authority=authority), | |
| host, | |
| port, | |
| ssl=ssl_context, | |
| ) | |
| except (OSError, ssl.SSLError) as exc: | |
| from .._exceptions import ConnectError | |
| msg = f"Failed to connect to {host}:{port}" | |
| raise ConnectError(msg) from exc |
| return | ||
| with contextlib.suppress(ValueError): | ||
| pool.all_connections.remove(connection) | ||
| pool.count -= 1 |
There was a problem hiding this comment.
P1: After decrementing pool.count, tasks blocked on await pool.idle.get() in _acquire are never woken. When the pool is saturated and a connection error triggers _discard, waiters hang indefinitely despite capacity becoming available. A sentinel value should be placed on the queue, or a semaphore/condition should be used to notify waiters that they can create a new connection.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/_transports/httpunk.py, line 182:
<comment>After decrementing `pool.count`, tasks blocked on `await pool.idle.get()` in `_acquire` are never woken. When the pool is saturated and a connection error triggers `_discard`, waiters hang indefinitely despite capacity becoming available. A sentinel value should be placed on the queue, or a semaphore/condition should be used to notify waiters that they can create a new connection.</comment>
<file context>
@@ -0,0 +1,190 @@
+ return
+ with contextlib.suppress(ValueError):
+ pool.all_connections.remove(connection)
+ pool.count -= 1
+
+ async def aclose(self) -> None:
</file context>
| pool.all_connections.append(connection) | ||
| return connection | ||
|
|
||
| return await pool.idle.get() |
There was a problem hiding this comment.
P2: When the connection pool is exhausted (all max_connections in use), pool.idle.get() blocks indefinitely without any timeout. Unlike the default transport which supports pool timeouts via the "timeout" extension, this transport offers no back-pressure or timeout mechanism, so callers can hang forever waiting for a connection to become available. Consider adding an asyncio.wait_for around the get() call, or reading a timeout from request.extensions.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpx2/httpx2/_transports/httpunk.py, line 131:
<comment>When the connection pool is exhausted (all `max_connections` in use), `pool.idle.get()` blocks indefinitely without any timeout. Unlike the default transport which supports pool timeouts via the `"timeout"` extension, this transport offers no back-pressure or timeout mechanism, so callers can hang forever waiting for a connection to become available. Consider adding an `asyncio.wait_for` around the `get()` call, or reading a timeout from `request.extensions`.</comment>
<file context>
@@ -0,0 +1,190 @@
+ pool.all_connections.append(connection)
+ return connection
+
+ return await pool.idle.get()
+
+ async def _connect(self, origin: tuple[str, str, int]) -> _HTTPunkConnection:
</file context>
1 participant
Summary
Adds an optional async HTTP/1.1 transport backed by httpunk:
httpx2.AsyncHTTPunkTransporthttpx2[httpunk]extraThis is intentionally small and experimental: it supports async HTTP/1.1 with a simple per-origin connection pool.
Test plan
uv run --with httpunk ruff check src/httpx2/httpx2/_transports/httpunk.py tests/httpx2/test_httpunk_transport.pyuv run --with httpunk pytest tests/httpx2/test_httpunk_transport.py -q