From dbf27d20b787478dd58285314e6678917f1524d8 Mon Sep 17 00:00:00 2001 From: antoinetoussaint-byte <312536060+antoinetoussaint-byte@users.noreply.github.com> Date: Sun, 25 Oct 2026 21:41:56 -0400 Subject: [PATCH] fix(connect): the face knows the server's limit before the client's first call, and tells it its own Follow-ups m1 and m2 of review r77 on #110: - m1: the relay read the client at once, so a call opened before the server's first SETTINGS met the face's default (ENHANCE_YOUR_CALM) where the server would have refused it itself, retry-safe. The relay now reads the server's connection preface (its SETTINGS frame, bounded by read_timeout) before it reads the client. - m2: a server that advertises no SETTINGS_MAX_CONCURRENT_STREAMS (grpcio's default) has the face's DEFAULT_MAX_OPEN_CALLS added to its first SETTINGS, so a client that keeps to HTTP/2 queues its calls within what the face enforces. The client acknowledges that one frame as the server's, so the server sees the ACK it expects. 1100 calls on one grpc.aio channel are now all answered (1000 before, the rest reset). Tests: test_a_server_advertising_no_limit_is_given_the_faces, test_the_client_is_read_once_the_servers_limit_is_known and test_a_burst_past_the_limit_queues_and_loses_no_call fail on main; test_a_server_limit_reaches_the_client_untouched. The fake servers in the suite now open with SETTINGS, as a server's preface does. Co-Authored-By: Claude Opus 5.5 --- templates/connect/connect_gen.py | 69 ++++++++++++++++------ testdata/connect/test_connect_gen.py | 88 ++++++++++++++++++++++------ 2 files changed, 123 insertions(+), 34 deletions(-) diff --git a/templates/connect/connect_gen.py b/templates/connect/connect_gen.py index 4428566..534fb3a 100644 --- a/templates/connect/connect_gen.py +++ b/templates/connect/connect_gen.py @@ -70,13 +70,15 @@ _PROTOCOL_ERROR, _CANCEL, _ENHANCE_YOUR_CALM = 0x1, 0x8, 0xB # Calls one relayed connection may hold open when the gRPC server advertises no # SETTINGS_MAX_CONCURRENT_STREAMS (grpcio advertises none unless the service sets -# grpc.max_concurrent_streams, as the scaffold's server does). Past it the face resets -# the call alone: ENHANCE_YOUR_CALM to the client (gRPC's RESOURCE_EXHAUSTED, which a -# client does not retry on its own), and CANCEL to the server. The server still receives -# the call's header block, which keeps the connection's header compression in step, so it -# may have started the call before the cancel: the face never promises it did not. What -# the face tracks per connection is bounded by it. When the server advertises a limit the -# face enforces nothing: the server refuses what is past it (REFUSED_STREAM), truthfully. +# grpc.max_concurrent_streams, as the scaffold's server does). The face advertises it to +# the client in the server's first SETTINGS, so a client that keeps to HTTP/2 queues its +# calls within it. Past it (a client that does not) the face resets the call alone: +# ENHANCE_YOUR_CALM to the client (gRPC's RESOURCE_EXHAUSTED, which a client does not +# retry on its own), and CANCEL to the server. The server still receives the call's +# header block, which keeps the connection's header compression in step, so it may have +# started the call before the cancel: the face never promises it did not. What the face +# tracks per connection is bounded by it. When the server advertises a limit the face +# enforces nothing: the server refuses what is past it (REFUSED_STREAM), truthfully. DEFAULT_MAX_OPEN_CALLS = 1000 # Bytes of frames for the client (refusals, a GOAWAY) the relay holds while the server's # frame in flight completes; past it the relay reads nothing more from the client until @@ -342,6 +344,24 @@ def _emit(self): self._after = self._on_frame(*self._frame, payload) or b"" +async def _server_preface(reader) -> bytes: + """The server's first frame, whole (RFC 9113 section 3.4: a SETTINGS frame).""" + header = await reader.readexactly(_FRAME.size) + high, low, *_ = _FRAME.unpack(header) + return header + await reader.readexactly(high << 16 | low) + + +def _with_max_streams(frame: bytes, limit: int) -> bytes: + """A server SETTINGS frame that advertises no SETTINGS_MAX_CONCURRENT_STREAMS, with + ``limit`` added: the client then holds its calls within what the face enforces. The + client acknowledges it as the server's one frame, so the server sees what it expects.""" + _, _, kind, flags, stream = _FRAME.unpack_from(frame) + if kind != _SETTINGS or flags & _ACK_FLAG: + return frame + payload = frame[_FRAME.size :] + struct.pack(">HI", _MAX_CONCURRENT_STREAMS, limit) + return _h2_frame(_SETTINGS, flags, stream, payload) + + class _Connection: def __init__(self, channel, grpc_address, limit, read_timeout, idle_timeout): self.channel, self.grpc_address, self.limit = channel, grpc_address, limit @@ -364,16 +384,18 @@ async def _relay(self, reader, writer, head): """gRPC over HTTP/2: the bytes go to the gRPC server as they came, until either side closes, or the connection has had no open call for ``idle_timeout``. - The frame headers say which calls are open: a client HEADERS on a new stream (odd, - above every stream before it) opens one, and the server's END_STREAM, either - side's RST_STREAM, or the server's GOAWAY above the stream ends it. A call past - the server's own SETTINGS_MAX_CONCURRENT_STREAMS is relayed as it came, for the - server to refuse. When the server advertises none, a call past - ``DEFAULT_MAX_OPEN_CALLS`` is reset alone (see there). The face's frames for the - client are written only between two of the server's frames and outside a header - block; while more than ``_MAX_QUEUED_FOR_CLIENT`` bytes of them wait, the client - is not read. A client that opens an even stream, which no HTTP/2 client does, gets - a GOAWAY naming why, and the connection is closed.""" + The frame headers say which calls are open: a client HEADERS on a new stream + (odd, above every stream before it) opens one, and the server's END_STREAM, + either side's RST_STREAM, or the server's GOAWAY above the stream ends it. A + call past the server's own SETTINGS_MAX_CONCURRENT_STREAMS is relayed as it + came, for the server to refuse. When the server advertises none, the face + advertises ``DEFAULT_MAX_OPEN_CALLS`` in the server's first SETTINGS, and a call + past it is reset alone (see there). The client is read only once that first + SETTINGS is. The face's frames for the client are written only between two of + the server's frames and outside a header block; while more than + ``_MAX_QUEUED_FOR_CLIENT`` bytes of them wait, the client is not read. A client + that opens an even stream, which no HTTP/2 client does, gets a GOAWAY naming + why, and the connection is closed.""" upstream_reader, upstream_writer = await asyncio.open_connection(*self.grpc_address) upstream_writer.write(head) loop = asyncio.get_running_loop() @@ -503,6 +525,19 @@ async def idle(): return await asyncio.sleep(remaining if remaining > 0 else self.idle_timeout) + # The server's connection preface, a SETTINGS frame, before the client is read: so + # the face knows the server's limit before the client's first call, and a client + # of a server that advertises none learns the face's. + try: + async with asyncio.timeout(self.read_timeout): + preface = await _server_preface(upstream_reader) + except (TimeoutError, asyncio.IncompleteReadError, ConnectionError) as exc: + _log.info("connect face: the gRPC server sent no SETTINGS: %r", exc) + upstream_writer.close() + return + server.feed(preface) + writer.write(preface if advertised else _with_max_streams(preface, max_calls)) + tasks = ( asyncio.ensure_future(pump(reader, upstream_writer, client, bytearray())), asyncio.ensure_future(pump(upstream_reader, writer, server, to_client)), diff --git a/testdata/connect/test_connect_gen.py b/testdata/connect/test_connect_gen.py index 83e9462..c60ad23 100644 --- a/testdata/connect/test_connect_gen.py +++ b/testdata/connect/test_connect_gen.py @@ -325,8 +325,9 @@ async def test_a_grpc_connection_carrying_traffic_outlives_the_idle_bound(): # Bytes in either direction keep the relay open: a server stream whose client sends # nothing is not idle. async def drip(reader, writer): - for _ in range(6): - writer.write(b".") + writer.write(SETTINGS) + for n in range(6): + writer.write(h2_frame(0x6, 0, 0, bytes([n]) * 8)) await writer.drain() await asyncio.sleep(0.2) writer.close() @@ -339,7 +340,8 @@ async def drip(reader, writer): reader, writer = await asyncio.open_connection("127.0.0.1", listener.port) writer.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n") await writer.drain() - assert await asyncio.wait_for(reader.read(), 3) == b"......" + got = read_frames(await asyncio.wait_for(reader.read(), 3)) + assert [f[3] for f in got if f[0] == 0x6] == [bytes([n]) * 8 for n in range(6)] writer.close() finally: await listener.stop() @@ -446,6 +448,7 @@ async def silent_upstream(): async def accept(reader, writer): held.append(writer) + writer.write(SETTINGS) # a server's connection preface if state["mid_frame"] is not None: # Half a PING, the rest once the test sets the event: the server is mid-frame. ping = h2_frame(0x6, 0, 0, b"12345678") @@ -549,7 +552,7 @@ async def test_a_call_past_the_limit_is_refused_alone(silent_upstream): payload = PREFACE + SETTINGS + b"".join(opens(2 * n + 1) for n in range(calls + 1)) closed, got = await exchange(await face(), payload, 1) assert not closed - assert read_frames(got) == [(0x3, 0, past, struct.pack(">I", 0xB))] + assert [f for f in read_frames(got) if f[0] == 0x3] == [(0x3, 0, past, struct.pack(">I", 0xB))] upstream = read_frames(bytes(state["received"])[len(PREFACE) :]) assert upstream[-2:] == [(0x1, 0x4, past, b"\x82"), (0x3, 0, past, struct.pack(">I", 0x8))] assert sum(1 for f in upstream if f[0] == 0x1) == calls + 1 @@ -569,10 +572,9 @@ async def test_a_refused_call_waits_for_the_end_of_its_header_block(silent_upstr assert [f[0] for f in upstream[-3:]] == [0x1, 0x9, 0x3] -async def test_a_burst_past_the_limit_loses_no_call(short_idle): +async def test_a_burst_past_the_limit_queues_and_loses_no_call(short_idle): # 1100 concurrent calls on one channel to a server advertising no limit: the face - # resets only the calls past 1000, each with ENHANCE_YOUR_CALM (never REFUSED_STREAM, - # which a client would retry on its own); every call it let through is answered. + # advertises its own 1000, so the client queues the rest, and every call is answered. async with grpc.aio.insecure_channel(f"127.0.0.1:{short_idle.port}") as channel: say_call = channel.unary_unary( "/echo.v1.Echo/Say", @@ -586,14 +588,65 @@ async def test_a_burst_past_the_limit_loses_no_call(short_idle): ], return_exceptions=True, ) - answered = [a for a in answers if not isinstance(a, Exception)] - refused = [a for a in answers if isinstance(a, Exception)] - assert all(a.text == "slow|maya" for a in answered) - assert len(answered) == connect_gen.DEFAULT_MAX_OPEN_CALLS - assert all( - a.code() == grpc.StatusCode.RESOURCE_EXHAUSTED and "error code 11" in a.details() - for a in refused - ), {a.details() for a in refused} + failed = {str(a)[:200] for a in answers if isinstance(a, Exception)} + assert not failed, failed + assert all(a.text == "slow|maya" for a in answers) + + +def _settings(frame): + """The (id, value) entries of a SETTINGS frame.""" + payload = frame[3] + return [struct.unpack_from(">HI", payload, at) for at in range(0, len(payload), 6)] + + +async def _first_frames(upstream_preface, client_payload, monkeypatch=None): + """What the client receives, and what the server receives, when a fake gRPC server + sends ``upstream_preface`` 0.3 s after the connection and reads everything after.""" + received = bytearray() + + async def accept(reader, writer): + await asyncio.sleep(0.3) + writer.write(upstream_preface) + await writer.drain() + while chunk := await reader.read(65536): + received.extend(chunk) + + upstream = await asyncio.start_server(accept, "127.0.0.1", 0) + listener = await connect_gen.serve(0, upstream.sockets[0].getsockname()[1], host="127.0.0.1") + try: + closed, got = await exchange(listener, client_payload, 1) + return read_frames(got), read_frames(bytes(received)[len(PREFACE) :]) + finally: + await listener.stop() + upstream.close() + + +async def test_a_server_advertising_no_limit_is_given_the_faces(): + # grpcio advertises no SETTINGS_MAX_CONCURRENT_STREAMS: the client is told the face's + # DEFAULT_MAX_OPEN_CALLS in the server's own SETTINGS, so it queues within it. + server_settings = h2_frame(0x4, 0, 0, struct.pack(">HI", 0x4, 65535)) + to_client, _ = await _first_frames(server_settings, PREFACE + SETTINGS) + assert to_client[0][0] == 0x4 + assert _settings(to_client[0]) == [(0x4, 65535), (0x3, connect_gen.DEFAULT_MAX_OPEN_CALLS)] + + +async def test_a_server_limit_reaches_the_client_untouched(): + server_settings = h2_frame(0x4, 0, 0, struct.pack(">HI", 0x3, 7)) + to_client, _ = await _first_frames(server_settings, PREFACE + SETTINGS) + assert _settings(to_client[0]) == [(0x3, 7)] + + +async def test_the_client_is_read_once_the_servers_limit_is_known(monkeypatch): + # The client opens three calls at once; the server's SETTINGS (limit 2) arrives 0.3 s + # later. The face waits for it, so the third call goes to the server, which refuses it + # itself (retry-safe), rather than being reset by the face as past its own default. + monkeypatch.setattr(connect_gen, "DEFAULT_MAX_OPEN_CALLS", 2) + server_settings = h2_frame(0x4, 0, 0, struct.pack(">HI", 0x3, 2)) + to_client, to_server = await _first_frames( + server_settings, PREFACE + SETTINGS + opens(1) + opens(3) + opens(5) + ) + assert [f for f in to_client if f[0] == 0x3] == [] + assert [f[0] for f in to_server] == [0x4, 0x1, 0x1, 0x1] def _literal_headers(headers): @@ -782,7 +835,8 @@ async def test_the_relay_ends_when_the_grpc_server_hangs_up(silent_upstream): break await asyncio.sleep(0.1) assert not listener._connections, "the face still holds the client's connection" - assert await asyncio.wait_for(reader.read(), 1) == b"" + # The server's SETTINGS, then the end. + assert [f[0] for f in read_frames(await asyncio.wait_for(reader.read(), 1))] == [0x4] finally: writer.close() @@ -896,7 +950,7 @@ async def test_the_even_stream_goaway_waits_for_the_servers_frame(silent_upstrea state["mid_frame"].set() await frames.until(lambda: frames.eof, 3) kinds = [f[0] for f in frames.frames] - assert kinds == [0x6, 0x7] # the server's frame whole, then the GOAWAY + assert kinds == [0x4, 0x6, 0x7] # the server's frame whole, then the GOAWAY last, code = struct.unpack_from(">II", frames.frames[-1][3]) assert (last, code) == (1, 0x1) finally: