Serialize h2 state machine access for thread-safe HTTP/2 connections - #1153
Serialize h2 state machine access for thread-safe HTTP/2 connections#1153Kludex wants to merge 6 commits into
Conversation
|
Docs preview: https://7ad24a5a-httpx2-docs.pydantic.workers.dev |
Merging this PR will not alter performance
Comparing 
Footnotes
|
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
![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: 1f4bda20e8
ℹ️ 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._request_count -= 1 | ||
| await self._max_streams_semaphore.release() | ||
| raise ConnectionNotAvailable() | ||
| async with self._write_lock: |
There was a problem hiding this comment.
When remote settings permit multiple streams and one task holds _write_lock during a slow network write, another task can acquire _max_streams_semaphore and then be cancelled while awaiting this new async with. Because the cleanup try begins only after lock acquisition, cancellation escapes without releasing the semaphore permit; repeated cancellations can permanently exhaust the connection's stream capacity and make later requests hang. Move lock acquisition inside the cleanup scope or release the permit if acquisition is interrupted.
Useful? React with 👍 / 👎.
|
|
||
| ### Fixed | ||
|
|
||
| * Serialize access to the `h2` state machine so that a single HTTP/2 connection can be safely shared across threads. ([#1153](https://github.com/pydantic/httpx2/pull/1153)) |
There was a problem hiding this comment.
The reviewed commit records Codex <codex@openai.com> as both its Author and Committer, whereas repository policy requires authorship to belong only to the engineer. Recreate the commit with the engineer's author identity before merging so the permanent history does not attribute the coding agent.
AGENTS.md reference: AGENTS.md:L3-L3
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.
All reported issues were addressed
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
There was a problem hiding this comment.
1 issue found across 4 files (changes from recent commits).
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/httpcore2/httpcore2/_sync/http2.py">
<violation number="1" location="src/httpcore2/httpcore2/_sync/http2.py:122">
P3: The `http2.send_request_headers.started` trace no longer includes `stream_id`. `kwargs` is initialized as `{"request": request}` and the Trace captures it before `__enter__` fires; `kwargs["stream_id"] = stream_id` is only assigned later, inside the write lock, so the started event's info dict is emitted without the stream ID. Before this delta, `kwargs` was built as `{"request", "stream_id"}` before entering the Trace, so the started event carried the stream ID. Trace-extension hooks that key `http2.send_request_headers.started` off the stream ID (and which still receive it for `send_request_body`/`receive_response_headers`) now lose it. If the stream ID is needed in the started event, allocate it before tracing is not possible without entering the lock, so either accept the loss or document it; the mismatch with the other h2 traces is what makes this worth confirming.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| stream_id = self._h2_state.get_next_available_stream_id() | ||
| self._events[stream_id] = [] | ||
| kwargs: dict[str, typing.Any] = {"request": request} | ||
| with Trace("send_request_headers", logger, request, kwargs): |
There was a problem hiding this comment.
P3: The http2.send_request_headers.started trace no longer includes stream_id. kwargs is initialized as {"request": request} and the Trace captures it before __enter__ fires; kwargs["stream_id"] = stream_id is only assigned later, inside the write lock, so the started event's info dict is emitted without the stream ID. Before this delta, kwargs was built as {"request", "stream_id"} before entering the Trace, so the started event carried the stream ID. Trace-extension hooks that key http2.send_request_headers.started off the stream ID (and which still receive it for send_request_body/receive_response_headers) now lose it. If the stream ID is needed in the started event, allocate it before tracing is not possible without entering the lock, so either accept the loss or document it; the mismatch with the other h2 traces is what makes this worth confirming.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpcore2/httpcore2/_sync/http2.py, line 122:
<comment>The `http2.send_request_headers.started` trace no longer includes `stream_id`. `kwargs` is initialized as `{"request": request}` and the Trace captures it before `__enter__` fires; `kwargs["stream_id"] = stream_id` is only assigned later, inside the write lock, so the started event's info dict is emitted without the stream ID. Before this delta, `kwargs` was built as `{"request", "stream_id"}` before entering the Trace, so the started event carried the stream ID. Trace-extension hooks that key `http2.send_request_headers.started` off the stream ID (and which still receive it for `send_request_body`/`receive_response_headers`) now lose it. If the stream ID is needed in the started event, allocate it before tracing is not possible without entering the lock, so either accept the loss or document it; the mismatch with the other h2 traces is what makes this worth confirming.</comment>
<file context>
@@ -116,30 +116,34 @@ def handle_request(self, request: Request) -> Response:
+ stream_id: int | None = None
+ try:
+ kwargs: dict[str, typing.Any] = {"request": request}
+ with Trace("send_request_headers", logger, request, kwargs):
+ with self._write_lock:
+ # Stream ID allocation, HPACK encoding of the outgoing headers, and
</file context>
There was a problem hiding this comment.
Accepting the loss, intentionally. The stream ID cannot exist before entering the lock, and allocating it earlier reintroduces the ordering race this PR fixes. The .complete event and the send_request_body/receive_response_headers traces still carry stream_id; only send_request_headers.started loses it.
There was a problem hiding this comment.
3 issues found and verified against the latest diff
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="tests/httpcore2/_async/test_http2.py">
<violation number="1" location="tests/httpcore2/_async/test_http2.py:113">
P2: The process-global switch interval is lowered to 1e-6 but restored only on the happy path at the end of the test. If any assert or request raises (exactly the race failure this test targets), the restore line is skipped and the low interval leaks into every subsequent test in the session, slowing them down. Wrap the body in try/finally so `sys.setswitchinterval(switch_interval)` always runs.</violation>
<violation number="2" location="tests/httpcore2/_async/test_http2.py:113">
P3: This trio test sets `sys.setswitchinterval(1e-6)`, but trio (and the async path here) schedules cooperative tasks on a single thread and does not use GIL-based thread switching, so this call changes nothing for the async concurrency under test. It only adds a process-global side effect that, if the test body raises before the trailing restore, leaks into the rest of the test session. Drop the switch-interval manipulation here (or place it in try/finally).</violation>
</file>
<file name="tests/httpcore2/_sync/test_http2.py">
<violation number="1" location="tests/httpcore2/_sync/test_http2.py:113">
P2: The process-wide thread-switch interval is lowered to 1e-6 and only restored at the final line of the test, not in a finally. Because this test intentionally fails on main/regressions, a failure leaves the interpreter switching threads at 1e-6 for the rest of the session, which can make unrelated threaded tests flaky and obscure the real failure. Wrap the test body so the restore runs in a finally (or use a fixture).</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| sharing one connection. See https://github.com/pydantic/httpx2/discussions/1152 | ||
| """ | ||
| switch_interval = sys.getswitchinterval() | ||
| sys.setswitchinterval(1e-6) |
There was a problem hiding this comment.
P2: The process-global switch interval is lowered to 1e-6 but restored only on the happy path at the end of the test. If any assert or request raises (exactly the race failure this test targets), the restore line is skipped and the low interval leaks into every subsequent test in the session, slowing them down. Wrap the body in try/finally so sys.setswitchinterval(switch_interval) always runs.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/httpcore2/_async/test_http2.py, line 113:
<comment>The process-global switch interval is lowered to 1e-6 but restored only on the happy path at the end of the test. If any assert or request raises (exactly the race failure this test targets), the restore line is skipped and the low interval leaks into every subsequent test in the session, slowing them down. Wrap the body in try/finally so `sys.setswitchinterval(switch_interval)` always runs.</comment>
<file context>
@@ -99,6 +102,54 @@ async def test_http2_response_closed_twice() -> None:
+ sharing one connection. See https://github.com/pydantic/httpx2/discussions/1152
+ """
+ switch_interval = sys.getswitchinterval()
+ sys.setswitchinterval(1e-6)
+ requests_count = 10
+ origin = httpcore2.Origin(b"https", b"example.com", 443)
</file context>
There was a problem hiding this comment.
Stale - this was addressed in 0d3a916; the test body is already wrapped in try/finally.
| sharing one connection. See https://github.com/pydantic/httpx2/discussions/1152 | ||
| """ | ||
| switch_interval = sys.getswitchinterval() | ||
| sys.setswitchinterval(1e-6) |
There was a problem hiding this comment.
P2: The process-wide thread-switch interval is lowered to 1e-6 and only restored at the final line of the test, not in a finally. Because this test intentionally fails on main/regressions, a failure leaves the interpreter switching threads at 1e-6 for the rest of the session, which can make unrelated threaded tests flaky and obscure the real failure. Wrap the test body so the restore runs in a finally (or use a fixture).
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/httpcore2/_sync/test_http2.py, line 113:
<comment>The process-wide thread-switch interval is lowered to 1e-6 and only restored at the final line of the test, not in a finally. Because this test intentionally fails on main/regressions, a failure leaves the interpreter switching threads at 1e-6 for the rest of the session, which can make unrelated threaded tests flaky and obscure the real failure. Wrap the test body so the restore runs in a finally (or use a fixture).</comment>
<file context>
@@ -100,6 +103,54 @@ def test_http2_response_closed_twice() -> None:
+ sharing one connection. See https://github.com/pydantic/httpx2/discussions/1152
+ """
+ switch_interval = sys.getswitchinterval()
+ sys.setswitchinterval(1e-6)
+ requests_count = 10
+ origin = httpcore2.Origin(b"https", b"example.com", 443)
</file context>
There was a problem hiding this comment.
Stale - already wrapped in try/finally since 0d3a916.
| sharing one connection. See https://github.com/pydantic/httpx2/discussions/1152 | ||
| """ | ||
| switch_interval = sys.getswitchinterval() | ||
| sys.setswitchinterval(1e-6) |
There was a problem hiding this comment.
P3: This trio test sets sys.setswitchinterval(1e-6), but trio (and the async path here) schedules cooperative tasks on a single thread and does not use GIL-based thread switching, so this call changes nothing for the async concurrency under test. It only adds a process-global side effect that, if the test body raises before the trailing restore, leaks into the rest of the test session. Drop the switch-interval manipulation here (or place it in try/finally).
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/httpcore2/_async/test_http2.py, line 113:
<comment>This trio test sets `sys.setswitchinterval(1e-6)`, but trio (and the async path here) schedules cooperative tasks on a single thread and does not use GIL-based thread switching, so this call changes nothing for the async concurrency under test. It only adds a process-global side effect that, if the test body raises before the trailing restore, leaks into the rest of the test session. Drop the switch-interval manipulation here (or place it in try/finally).</comment>
<file context>
@@ -99,6 +102,54 @@ async def test_http2_response_closed_twice() -> None:
+ sharing one connection. See https://github.com/pydantic/httpx2/discussions/1152
+ """
+ switch_interval = sys.getswitchinterval()
+ sys.setswitchinterval(1e-6)
+ requests_count = 10
+ origin = httpcore2.Origin(b"https", b"example.com", 443)
</file context>
There was a problem hiding this comment.
The switch interval tweak is only meaningful for the sync test, but both are generated from one async source via unasync, so it stays in the async version to keep them in lockstep. It is harmless under trio and already guarded by try/finally.
| with Trace("send_request_headers", logger, request, kwargs): | ||
| with self._write_lock: | ||
| # Stream ID allocation, HPACK encoding of the outgoing headers, and | ||
| # writing the HEADERS frame to the network must be a single atomic | ||
| # section with respect to other streams on this connection. | ||
| stream_id = self._h2_state.get_next_available_stream_id() | ||
| self._events[stream_id] = [] | ||
| kwargs["stream_id"] = stream_id | ||
| self._send_request_headers(request=request, stream_id=stream_id) |
There was a problem hiding this comment.
🟡 Medium _sync/http2.py:122
http2.send_request_headers.started callbacks now receive only request, so trace consumers cannot correlate the event with its HTTP/2 stream. kwargs["stream_id"] is assigned after entering Trace; allocate and assign the stream ID before emitting the started event.
- with Trace("send_request_headers", logger, request, kwargs):
- with self._write_lock:
- # Stream ID allocation, HPACK encoding of the outgoing headers, and
- # writing the HEADERS frame to the network must be a single atomic
- # section with respect to other streams on this connection.
- stream_id = self._h2_state.get_next_available_stream_id()
- self._events[stream_id] = []
- kwargs["stream_id"] = stream_id
- self._send_request_headers(request=request, stream_id=stream_id)
+ with self._write_lock:
+ # Stream ID allocation, HPACK encoding of the outgoing headers, and
+ # writing the HEADERS frame to the network must be a single atomic
+ # section with respect to other streams on this connection.
+ stream_id = self._h2_state.get_next_available_stream_id()
+ self._events[stream_id] = []
+ kwargs["stream_id"] = stream_id
+ with Trace("send_request_headers", logger, request, kwargs):
+ self._send_request_headers(request=request, stream_id=stream_id)Also found in 1 other location(s)
src/httpcore2/httpcore2/_async/http2.py:122
kwargs["stream_id"]is populated only after enteringTrace, so thehttp2.send_request_headers.startedtrace callback now receives onlyrequest, whereas it previously received bothrequestandstream_id. This silently breaks trace consumers that correlate the header-send start event by HTTP/2 stream ID. Allocate the ID under the lock before emitting the started event, or otherwise preserve the callback payload.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @src/httpcore2/httpcore2/_sync/http2.py around lines 122-130:
`http2.send_request_headers.started` callbacks now receive only `request`, so trace consumers cannot correlate the event with its HTTP/2 stream. `kwargs["stream_id"]` is assigned after entering `Trace`; allocate and assign the stream ID before emitting the started event.
Also found in 1 other location(s):
- src/httpcore2/httpcore2/_async/http2.py:122 -- `kwargs["stream_id"]` is populated only after entering `Trace`, so the `http2.send_request_headers.started` trace callback now receives only `request`, whereas it previously received both `request` and `stream_id`. This silently breaks trace consumers that correlate the header-send start event by HTTP/2 stream ID. Allocate the ID under the lock before emitting the started event, or otherwise preserve the callback payload.
There was a problem hiding this comment.
Declining this one: moving the Trace inside _write_lock runs user trace callbacks while holding the lock, which reintroduces the deadlock flagged earlier (a callback awaiting another request on this connection would block forever). The stream ID genuinely does not exist before the atomic section starts. The .complete event and the send_request_body/receive_response_headers traces still carry stream_id, so correlation remains possible - only the started payload loses it.
There was a problem hiding this comment.
All reported issues were addressed across 2 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
johnflavin
left a comment
johnflavin
left a comment
There was a problem hiding this comment.
I had been looking at this with Claude before I saw you had already fixed it. But my Claude made a slightly different choice about the lock than you did here—reusing _write_lock vs making a dedicated AsyncThreadLock()—which it seems to think could cause some bugs.
Why the lock choice matters
There are two different kinds of critical section in this file, and they have opposite requirements:
- Protecting the h2 state machine. Pure CPU work, microseconds long, and it must be indivisible — the invariant is "bytes that left the socket have been handed to h2", "the stream ID we were given is the one we send headers on".
- Serializing the socket. Held across
await network_stream.write(...). Long, blocking, and — this is the part that bites — suspendable.
_write_lockis the second kind. Reusing it as the h2 mutex makes every h2 critical section inherit the properties of an I/O lock: in async, acquiring a contended lock is anawait, and anawaitis a cancellation point. So each place the PR wraps h2 state in_write_lockbecomes a place a task can be suspended and torn out mid-invariant. I think that's also visible in the commit history here — several of the follow-up commits (restoring idle state, cleaning up pre-allocation state, narrowing theNoAvailableStreamIDErrorhandling) are repairs for states that only became reachable oncehandle_requestgrew a suspension point between acquiring the semaphore and allocating the stream.The second half of the argument: the async path never needed this lock. The bug is a thread-safety bug, not a task-safety bug. Within one event loop these regions contain no
await, so tasks cannot interleave inside them — the PR description says as much ("async locks remain no-op under single event loop"), but_write_lockis a real lock in async, so async pays the cost and takes on the new failure mode without getting anything back.
httpcorealready has a primitive for precisely this case, andconnection_pool.pyalready uses it:class AsyncThreadLock: """ This is a threading-only lock for no-I/O contexts. In the sync case `ThreadLock` provides thread locking. In the async case `AsyncThreadLock` is a no-op. """Choosing it isn't a micro-optimization — it's a statement that the section contains no I/O, and it makes the cancellation hazard structurally impossible rather than something reviewers have to keep re-checking.
The concrete failure
_receive_eventsnow reads from the network and then awaits the lock before handing the bytes to h2:data = await self._read_incoming_data(request) # bytes are off the socket async with self._write_lock: # <-- await == cancellation point events: list[h2.events.Event] = self._h2_state.receive_data(data)If the task is cancelled while waiting for that lock, the bytes it already consumed are gone. h2 never sees them, so framing is corrupted for every stream on that connection, not just the cancelled one. On
mainthere is noawaitbetween the read andreceive_data, so the pair is atomic with respect to cancellation. This is reachable from ordinary code — a per-request timeout, or a task group cancelling siblings, on an async client multiplexing over one connection, i.e. exactly when the write lock is contended.Test below (drop it in
tests/httpcore2/); it opens a stream, lets another task hold the write lock, and cancels a_receive_eventscall while it waits:
asyncio trio mainframe recorded — PASS PASS this PR frame read and discarded — FAIL FAIL The sync path is unaffected, since
threading.Lock.acquire()isn't a cancellation point. That asymmetry is itself the tell: the regression exists only in the half of the codebase that didn't have the bug.Options
- Use
AsyncThreadLock/ThreadLockfor the h2-state sections and leave_write_lockowning the socket. Async goes back to a no-op, sync gets the serialization it needs, and read/receive_datastays atomic. Ordering is still safe:data_to_send()drains FIFO under the write lock, so wire order still matches encode order.- Keep
_write_lock, but re-join the read andreceive_data— either both under the lock, or by stashing unconsumed bytes on the connection so a later call feeds them to h2 instead of dropping them.
tests/httpcore2/test_cancel_loss.py
"""
Does a cancellation between "read bytes off the socket" and "feed them to h2"
lose those bytes?
On main (and with the ThreadLock approach) `_read_incoming_data` reads and then
calls `h2_state.receive_data(...)` with no await in between, so the pair is
atomic with respect to cancellation. PR #1153 moves `receive_data` behind
`async with self._write_lock`, which is an await point.
"""
import anyio
import hpack
import hyperframe.frame
import pytest
import httpcore2
class ReplayStream(httpcore2.AsyncNetworkStream):
def __init__(self, buffer):
self._buffer = list(buffer)
self.reads = 0
async def read(self, max_bytes: int, timeout: float | None = None) -> bytes:
if not self._buffer:
await anyio.sleep(999)
self.reads += 1
return self._buffer.pop(0)
async def write(self, buffer: bytes, timeout: float | None = None) -> None:
pass
async def aclose(self) -> None:
pass
@pytest.mark.anyio
async def test_cancellation_between_read_and_receive_data() -> None:
encoder = hpack.Encoder()
buffer = [
hyperframe.frame.SettingsFrame().serialize(),
hyperframe.frame.HeadersFrame(
stream_id=1,
data=encoder.encode([(b":status", b"200"), (b"content-type", b"plain/text")]),
flags=["END_HEADERS"],
).serialize(),
hyperframe.frame.DataFrame(stream_id=1, data=b"Hello, world!", flags=["END_STREAM"]).serialize(),
]
origin = httpcore2.Origin(b"https", b"example.com", 443)
stream = ReplayStream(buffer)
conn = httpcore2.AsyncHTTP2Connection(origin=origin, stream=stream)
request = httpcore2.Request("GET", "https://example.com/", headers={"Host": "example.com"})
await conn._send_connection_init(request)
# Open stream 1 for real, so that the HEADERS frame the server sends back
# for it is legitimate. (Done via h2 directly so that the test runs
# unchanged against every variant of `_send_request_headers`.)
async with conn._write_lock:
stream_id = conn._h2_state.get_next_available_stream_id()
assert stream_id == 1
conn._events[stream_id] = []
conn._h2_state.send_headers(
stream_id,
[(b":method", b"GET"), (b":authority", b"example.com"), (b":scheme", b"https"), (b":path", b"/")],
end_stream=True,
)
conn._h2_state.data_to_send()
# Consume the server's SETTINGS frame first, so that the read we're
# interested in is the one that returns the response HEADERS frame.
await conn._receive_events(request, stream_id=1)
assert conn._events[1] == []
holding = anyio.Event()
release = anyio.Event()
async def hold_write_lock() -> None:
# Stand in for another task that is stuck writing to the network.
async with conn._write_lock:
holding.set()
await release.wait()
async with anyio.create_task_group() as tg:
tg.start_soon(hold_write_lock)
await holding.wait()
# Our task reads a frame off the network, then gets cancelled while
# waiting for that same lock.
with anyio.move_on_after(0.05):
await conn._receive_events(request, stream_id=1)
release.set()
frames_read = stream.reads - 1 # not counting the SETTINGS frame
events_recorded = len(conn._events[1])
print(f"\nframes read after cancellation point: {frames_read}, events recorded: {events_recorded}")
assert frames_read == 1, "the test did not get as far as reading the HEADERS frame"
# If the frame that was read never reached h2, it is gone for good: the
# response it carried can never be delivered.
assert events_recorded == 1, "the HEADERS frame read from the network was discarded by the cancellation"Generated by Claude Code
2 participants
Closes #908, addresses discussion #1152.
A sync
Client(http2=True)shared across threads corrupts the connection'sh2state machine: stream ID allocation, HPACK encoding, and frame writes were not atomic with respect to each other, producingStreamIDTooLowError,LocalProtocolError,KeyError, and interleaved HEADERS frames on the wire.Changes
_eventsregistration, and sending the HEADERS frame now happen as a single atomic section under the write lock. Two streams can no longer send HEADERS out of stream ID order or interleave HPACK encoder state.h2state mutation (send_data,end_stream,acknowledge_received_data,receive_data) is now paired with itsdata_to_send()flush under the write lock, so frames from concurrent streams cannot interleave mid-sequence._wait_for_outgoing_flowis folded into_send_stream_data: the flow control window is checked and consumed under the same lock as thesend_datathat spends it, closing the window where two streams could both observe the same credit.h2.ProtocolError->RemoteProtocolError/LocalProtocolErrormapping is extracted into_map_exceptionsince the request flow now has two exception paths.The async connection gets the same structure via unasync. Async locks are no-op-cheap under a single event loop, so behaviour there is unchanged.
Testing
New
test_http2_connection_concurrent_requestsruns 10 concurrent requests over one connection - threads in the sync case, trio tasks in the async case. With a reducedsys.setswitchintervalthe sync test fails deterministically onmain(StreamIDTooLowError/LocalProtocolError) and passes with this change.AI Disclaimer
This PR was developed with the assistance of either Claude or Codex. I've reviewed and verified the changes.