Skip to content

Commit dbf27d2

Browse files
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 <noreply@anthropic.com>
1 parent e34ea7c commit dbf27d2

2 files changed

Lines changed: 123 additions & 34 deletions

File tree

‎templates/connect/connect_gen.py‎

Lines changed: 52 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -70,13 +70,15 @@
7070
_PROTOCOL_ERROR, _CANCEL, _ENHANCE_YOUR_CALM = 0x1, 0x8, 0xB
7171
# Calls one relayed connection may hold open when the gRPC server advertises no
7272
# SETTINGS_MAX_CONCURRENT_STREAMS (grpcio advertises none unless the service sets
73-
# grpc.max_concurrent_streams, as the scaffold's server does). Past it the face resets
74-
# the call alone: ENHANCE_YOUR_CALM to the client (gRPC's RESOURCE_EXHAUSTED, which a
75-
# client does not retry on its own), and CANCEL to the server. The server still receives
76-
# the call's header block, which keeps the connection's header compression in step, so it
77-
# may have started the call before the cancel: the face never promises it did not. What
78-
# the face tracks per connection is bounded by it. When the server advertises a limit the
79-
# face enforces nothing: the server refuses what is past it (REFUSED_STREAM), truthfully.
73+
# grpc.max_concurrent_streams, as the scaffold's server does). The face advertises it to
74+
# the client in the server's first SETTINGS, so a client that keeps to HTTP/2 queues its
75+
# calls within it. Past it (a client that does not) the face resets the call alone:
76+
# ENHANCE_YOUR_CALM to the client (gRPC's RESOURCE_EXHAUSTED, which a client does not
77+
# retry on its own), and CANCEL to the server. The server still receives the call's
78+
# header block, which keeps the connection's header compression in step, so it may have
79+
# started the call before the cancel: the face never promises it did not. What the face
80+
# tracks per connection is bounded by it. When the server advertises a limit the face
81+
# enforces nothing: the server refuses what is past it (REFUSED_STREAM), truthfully.
8082
DEFAULT_MAX_OPEN_CALLS = 1000
8183
# Bytes of frames for the client (refusals, a GOAWAY) the relay holds while the server's
8284
# frame in flight completes; past it the relay reads nothing more from the client until
@@ -342,6 +344,24 @@ def _emit(self):
342344
self._after = self._on_frame(*self._frame, payload) or b""
343345

344346

347+
async def _server_preface(reader) -> bytes:
348+
"""The server's first frame, whole (RFC 9113 section 3.4: a SETTINGS frame)."""
349+
header = await reader.readexactly(_FRAME.size)
350+
high, low, *_ = _FRAME.unpack(header)
351+
return header + await reader.readexactly(high << 16 | low)
352+
353+
354+
def _with_max_streams(frame: bytes, limit: int) -> bytes:
355+
"""A server SETTINGS frame that advertises no SETTINGS_MAX_CONCURRENT_STREAMS, with
356+
``limit`` added: the client then holds its calls within what the face enforces. The
357+
client acknowledges it as the server's one frame, so the server sees what it expects."""
358+
_, _, kind, flags, stream = _FRAME.unpack_from(frame)
359+
if kind != _SETTINGS or flags & _ACK_FLAG:
360+
return frame
361+
payload = frame[_FRAME.size :] + struct.pack(">HI", _MAX_CONCURRENT_STREAMS, limit)
362+
return _h2_frame(_SETTINGS, flags, stream, payload)
363+
364+
345365
class _Connection:
346366
def __init__(self, channel, grpc_address, limit, read_timeout, idle_timeout):
347367
self.channel, self.grpc_address, self.limit = channel, grpc_address, limit
@@ -364,16 +384,18 @@ async def _relay(self, reader, writer, head):
364384
"""gRPC over HTTP/2: the bytes go to the gRPC server as they came, until either
365385
side closes, or the connection has had no open call for ``idle_timeout``.
366386
367-
The frame headers say which calls are open: a client HEADERS on a new stream (odd,
368-
above every stream before it) opens one, and the server's END_STREAM, either
369-
side's RST_STREAM, or the server's GOAWAY above the stream ends it. A call past
370-
the server's own SETTINGS_MAX_CONCURRENT_STREAMS is relayed as it came, for the
371-
server to refuse. When the server advertises none, a call past
372-
``DEFAULT_MAX_OPEN_CALLS`` is reset alone (see there). The face's frames for the
373-
client are written only between two of the server's frames and outside a header
374-
block; while more than ``_MAX_QUEUED_FOR_CLIENT`` bytes of them wait, the client
375-
is not read. A client that opens an even stream, which no HTTP/2 client does, gets
376-
a GOAWAY naming why, and the connection is closed."""
387+
The frame headers say which calls are open: a client HEADERS on a new stream
388+
(odd, above every stream before it) opens one, and the server's END_STREAM,
389+
either side's RST_STREAM, or the server's GOAWAY above the stream ends it. A
390+
call past the server's own SETTINGS_MAX_CONCURRENT_STREAMS is relayed as it
391+
came, for the server to refuse. When the server advertises none, the face
392+
advertises ``DEFAULT_MAX_OPEN_CALLS`` in the server's first SETTINGS, and a call
393+
past it is reset alone (see there). The client is read only once that first
394+
SETTINGS is. The face's frames for the client are written only between two of
395+
the server's frames and outside a header block; while more than
396+
``_MAX_QUEUED_FOR_CLIENT`` bytes of them wait, the client is not read. A client
397+
that opens an even stream, which no HTTP/2 client does, gets a GOAWAY naming
398+
why, and the connection is closed."""
377399
upstream_reader, upstream_writer = await asyncio.open_connection(*self.grpc_address)
378400
upstream_writer.write(head)
379401
loop = asyncio.get_running_loop()
@@ -503,6 +525,19 @@ async def idle():
503525
return
504526
await asyncio.sleep(remaining if remaining > 0 else self.idle_timeout)
505527

528+
# The server's connection preface, a SETTINGS frame, before the client is read: so
529+
# the face knows the server's limit before the client's first call, and a client
530+
# of a server that advertises none learns the face's.
531+
try:
532+
async with asyncio.timeout(self.read_timeout):
533+
preface = await _server_preface(upstream_reader)
534+
except (TimeoutError, asyncio.IncompleteReadError, ConnectionError) as exc:
535+
_log.info("connect face: the gRPC server sent no SETTINGS: %r", exc)
536+
upstream_writer.close()
537+
return
538+
server.feed(preface)
539+
writer.write(preface if advertised else _with_max_streams(preface, max_calls))
540+
506541
tasks = (
507542
asyncio.ensure_future(pump(reader, upstream_writer, client, bytearray())),
508543
asyncio.ensure_future(pump(upstream_reader, writer, server, to_client)),

‎testdata/connect/test_connect_gen.py‎

Lines changed: 71 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -325,8 +325,9 @@ async def test_a_grpc_connection_carrying_traffic_outlives_the_idle_bound():
325325
# Bytes in either direction keep the relay open: a server stream whose client sends
326326
# nothing is not idle.
327327
async def drip(reader, writer):
328-
for _ in range(6):
329-
writer.write(b".")
328+
writer.write(SETTINGS)
329+
for n in range(6):
330+
writer.write(h2_frame(0x6, 0, 0, bytes([n]) * 8))
330331
await writer.drain()
331332
await asyncio.sleep(0.2)
332333
writer.close()
@@ -339,7 +340,8 @@ async def drip(reader, writer):
339340
reader, writer = await asyncio.open_connection("127.0.0.1", listener.port)
340341
writer.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
341342
await writer.drain()
342-
assert await asyncio.wait_for(reader.read(), 3) == b"......"
343+
got = read_frames(await asyncio.wait_for(reader.read(), 3))
344+
assert [f[3] for f in got if f[0] == 0x6] == [bytes([n]) * 8 for n in range(6)]
343345
writer.close()
344346
finally:
345347
await listener.stop()
@@ -446,6 +448,7 @@ async def silent_upstream():
446448

447449
async def accept(reader, writer):
448450
held.append(writer)
451+
writer.write(SETTINGS) # a server's connection preface
449452
if state["mid_frame"] is not None:
450453
# Half a PING, the rest once the test sets the event: the server is mid-frame.
451454
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):
549552
payload = PREFACE + SETTINGS + b"".join(opens(2 * n + 1) for n in range(calls + 1))
550553
closed, got = await exchange(await face(), payload, 1)
551554
assert not closed
552-
assert read_frames(got) == [(0x3, 0, past, struct.pack(">I", 0xB))]
555+
assert [f for f in read_frames(got) if f[0] == 0x3] == [(0x3, 0, past, struct.pack(">I", 0xB))]
553556
upstream = read_frames(bytes(state["received"])[len(PREFACE) :])
554557
assert upstream[-2:] == [(0x1, 0x4, past, b"\x82"), (0x3, 0, past, struct.pack(">I", 0x8))]
555558
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
569572
assert [f[0] for f in upstream[-3:]] == [0x1, 0x9, 0x3]
570573

571574

572-
async def test_a_burst_past_the_limit_loses_no_call(short_idle):
575+
async def test_a_burst_past_the_limit_queues_and_loses_no_call(short_idle):
573576
# 1100 concurrent calls on one channel to a server advertising no limit: the face
574-
# resets only the calls past 1000, each with ENHANCE_YOUR_CALM (never REFUSED_STREAM,
575-
# which a client would retry on its own); every call it let through is answered.
577+
# advertises its own 1000, so the client queues the rest, and every call is answered.
576578
async with grpc.aio.insecure_channel(f"127.0.0.1:{short_idle.port}") as channel:
577579
say_call = channel.unary_unary(
578580
"/echo.v1.Echo/Say",
@@ -586,14 +588,65 @@ async def test_a_burst_past_the_limit_loses_no_call(short_idle):
586588
],
587589
return_exceptions=True,
588590
)
589-
answered = [a for a in answers if not isinstance(a, Exception)]
590-
refused = [a for a in answers if isinstance(a, Exception)]
591-
assert all(a.text == "slow|maya" for a in answered)
592-
assert len(answered) == connect_gen.DEFAULT_MAX_OPEN_CALLS
593-
assert all(
594-
a.code() == grpc.StatusCode.RESOURCE_EXHAUSTED and "error code 11" in a.details()
595-
for a in refused
596-
), {a.details() for a in refused}
591+
failed = {str(a)[:200] for a in answers if isinstance(a, Exception)}
592+
assert not failed, failed
593+
assert all(a.text == "slow|maya" for a in answers)
594+
595+
596+
def _settings(frame):
597+
"""The (id, value) entries of a SETTINGS frame."""
598+
payload = frame[3]
599+
return [struct.unpack_from(">HI", payload, at) for at in range(0, len(payload), 6)]
600+
601+
602+
async def _first_frames(upstream_preface, client_payload, monkeypatch=None):
603+
"""What the client receives, and what the server receives, when a fake gRPC server
604+
sends ``upstream_preface`` 0.3 s after the connection and reads everything after."""
605+
received = bytearray()
606+
607+
async def accept(reader, writer):
608+
await asyncio.sleep(0.3)
609+
writer.write(upstream_preface)
610+
await writer.drain()
611+
while chunk := await reader.read(65536):
612+
received.extend(chunk)
613+
614+
upstream = await asyncio.start_server(accept, "127.0.0.1", 0)
615+
listener = await connect_gen.serve(0, upstream.sockets[0].getsockname()[1], host="127.0.0.1")
616+
try:
617+
closed, got = await exchange(listener, client_payload, 1)
618+
return read_frames(got), read_frames(bytes(received)[len(PREFACE) :])
619+
finally:
620+
await listener.stop()
621+
upstream.close()
622+
623+
624+
async def test_a_server_advertising_no_limit_is_given_the_faces():
625+
# grpcio advertises no SETTINGS_MAX_CONCURRENT_STREAMS: the client is told the face's
626+
# DEFAULT_MAX_OPEN_CALLS in the server's own SETTINGS, so it queues within it.
627+
server_settings = h2_frame(0x4, 0, 0, struct.pack(">HI", 0x4, 65535))
628+
to_client, _ = await _first_frames(server_settings, PREFACE + SETTINGS)
629+
assert to_client[0][0] == 0x4
630+
assert _settings(to_client[0]) == [(0x4, 65535), (0x3, connect_gen.DEFAULT_MAX_OPEN_CALLS)]
631+
632+
633+
async def test_a_server_limit_reaches_the_client_untouched():
634+
server_settings = h2_frame(0x4, 0, 0, struct.pack(">HI", 0x3, 7))
635+
to_client, _ = await _first_frames(server_settings, PREFACE + SETTINGS)
636+
assert _settings(to_client[0]) == [(0x3, 7)]
637+
638+
639+
async def test_the_client_is_read_once_the_servers_limit_is_known(monkeypatch):
640+
# The client opens three calls at once; the server's SETTINGS (limit 2) arrives 0.3 s
641+
# later. The face waits for it, so the third call goes to the server, which refuses it
642+
# itself (retry-safe), rather than being reset by the face as past its own default.
643+
monkeypatch.setattr(connect_gen, "DEFAULT_MAX_OPEN_CALLS", 2)
644+
server_settings = h2_frame(0x4, 0, 0, struct.pack(">HI", 0x3, 2))
645+
to_client, to_server = await _first_frames(
646+
server_settings, PREFACE + SETTINGS + opens(1) + opens(3) + opens(5)
647+
)
648+
assert [f for f in to_client if f[0] == 0x3] == []
649+
assert [f[0] for f in to_server] == [0x4, 0x1, 0x1, 0x1]
597650

598651

599652
def _literal_headers(headers):
@@ -782,7 +835,8 @@ async def test_the_relay_ends_when_the_grpc_server_hangs_up(silent_upstream):
782835
break
783836
await asyncio.sleep(0.1)
784837
assert not listener._connections, "the face still holds the client's connection"
785-
assert await asyncio.wait_for(reader.read(), 1) == b""
838+
# The server's SETTINGS, then the end.
839+
assert [f[0] for f in read_frames(await asyncio.wait_for(reader.read(), 1))] == [0x4]
786840
finally:
787841
writer.close()
788842

@@ -896,7 +950,7 @@ async def test_the_even_stream_goaway_waits_for_the_servers_frame(silent_upstrea
896950
state["mid_frame"].set()
897951
await frames.until(lambda: frames.eof, 3)
898952
kinds = [f[0] for f in frames.frames]
899-
assert kinds == [0x6, 0x7] # the server's frame whole, then the GOAWAY
953+
assert kinds == [0x4, 0x6, 0x7] # the server's frame whole, then the GOAWAY
900954
last, code = struct.unpack_from(">II", frames.frames[-1][3])
901955
assert (last, code) == (1, 0x1)
902956
finally:

0 commit comments

Comments
 (0)