Skip to content

Commit 43edffd

Browse files
fix(connect): the minors of #111 and #117's reviews
#111 (review r106): - m1: the channel's own receive bound is one byte past both answer bounds, so a refusal a service passes on from a client whose bound equals the face's is never taken for the face's own (grpcio names the client's bound). - m2/m3: the docstring says a list of enums costs with the length of their names (measured 130x at 22 characters, 373x at 100, 687x at 200), so the JSON default's worst case depends on the service's enum names, and gives the measured process memory of a proto answer (54 to 98 MiB for 15 MiB). #117 (review r107): - m1: the client's SETTINGS, which completes its preface, goes to the server before the face waits for the server's, so a server that waits for the client's first is not stalled. - m2: a server that sends no SETTINGS within read_timeout gets the client a GOAWAY (INTERNAL_ERROR) naming why. - m3: a SETTINGS frame is scanned whole as it streams, holding one entry, so the server's limit is found wherever it is in the frame (not only in its first 256 bytes). A first frame is held to the 16 KiB any peer may send before it has the other's SETTINGS. Tests (each fails on main): test_a_refusal_naming_the_faces_own_bound_is_still_the_services, test_a_server_that_waits_for_the_clients_settings_is_not_stalled, test_a_server_that_sends_no_settings_gets_the_client_a_goaway, test_the_servers_limit_is_found_anywhere_in_its_settings (also fed one byte at a time). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 097a4fc commit 43edffd

2 files changed

Lines changed: 154 additions & 27 deletions

File tree

‎templates/connect/connect_gen.py‎

Lines changed: 79 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -23,16 +23,19 @@
2323
2424
- ``max_response_bytes`` for an answer relayed as proto (Connect and gRPC-Web proto): by
2525
default the larger of ``DEFAULT_MAX_RESPONSE_BYTES`` (16 MiB) and ``max_message_bytes``.
26-
The face holds such an answer about twice over while it writes it, beyond the
27-
service's own copy: measured, 33 MiB for a 16 MiB answer.
26+
The face holds such an answer more than once while it writes it: measured as process
27+
memory, 54 to 98 MiB for a 15 MiB answer, by protocol.
2828
- ``max_json_response_bytes`` for an answer converted to JSON: by default the larger of
2929
``DEFAULT_MAX_JSON_RESPONSE_BYTES`` (4 MiB, the bound every answer had before the two
3030
were apart) and ``max_message_bytes``. It is checked before the conversion, because the
3131
conversion is what costs: measured as process memory (protobuf 6.33, one 4 MiB answer),
32-
a packed list of floats peaks at 18 times the proto, nested messages at 62 times, and a
33-
list of enums at 130 times (its JSON alone is 25 times the proto). So up to about
34-
520 MiB at the default, for one conversion; conversions run on the face's event loop
35-
one at a time, so that peak does not add up across calls.
32+
a packed list of floats peaks at 18 times the proto and nested messages at 62 times. A
33+
list of enums costs with the length of their names, since each one-byte value becomes
34+
its name in JSON: 130 times with 22-character names, 373 times with 100, 687 times with
35+
200. So the worst case at the default depends on the service's enum names: about
36+
520 MiB for one answer with names of about 20 characters, 2.7 GiB with 200.
37+
Conversions run on the face's event loop one at a time, so that peak does not add up
38+
across calls; a service with long enum names in large lists lowers the bound.
3639
3740
Past its bound a call ends ``failed_precondition`` naming the bound: the answer is too
3841
large for this face's configuration, and asking again does not change it. Within it,
@@ -93,7 +96,9 @@
9396
_END_STREAM_FLAG, _ACK_FLAG, _END_HEADERS_FLAG = 0x1, 0x1, 0x4
9497
_MAX_CONCURRENT_STREAMS = 0x3
9598
_PUSH_PROMISE = 0x5
96-
_PROTOCOL_ERROR, _CANCEL, _ENHANCE_YOUR_CALM = 0x1, 0x8, 0xB
99+
_PROTOCOL_ERROR, _INTERNAL_ERROR, _CANCEL, _ENHANCE_YOUR_CALM = 0x1, 0x2, 0x8, 0xB
100+
# The largest frame a peer may send before it has the other's SETTINGS (RFC 9113 4.2).
101+
_INITIAL_MAX_FRAME = 16384
97102
# Calls one relayed connection may hold open when the gRPC server advertises no
98103
# SETTINGS_MAX_CONCURRENT_STREAMS (grpcio advertises none unless the service sets
99104
# grpc.max_concurrent_streams, as the scaffold's server does). The face advertises it to
@@ -113,7 +118,8 @@
113118
# How long a closing relay waits for the server's frame in flight to complete, so the
114119
# client's GOAWAY lands between two frames.
115120
_GOAWAY_GRACE_S = 5.0
116-
# The most of a SETTINGS or GOAWAY payload the relay reads; the rest is skipped.
121+
# The most of a GOAWAY payload the relay reads (the last stream, the code, the start of
122+
# the reason); the rest is skipped. A SETTINGS frame is scanned whole, keeping nothing.
117123
_CONTROL_PAYLOAD = 256
118124
_ENVELOPE = struct.Struct(">BI")
119125
_COMPRESSED, _END_STREAM, _TRAILERS = 0x01, 0x02, 0x80
@@ -285,11 +291,14 @@ def _positive(name: str, value: int | None, default: int, message_bound: int) ->
285291

286292
class _AnswerBounds:
287293
"""The answers the face takes from the service: ``proto`` for a codec that relays
288-
them as they are, ``json`` for one that converts them, checked before converting."""
294+
them as they are, ``json`` for one that converts them, checked before converting.
295+
``received`` is the channel's own receive bound: one byte past both, so the face
296+
checks every bound itself, and the number grpcio names when it refuses past it is
297+
one no service's client has by chance with the same text."""
289298

290299
def __init__(self, proto: int, json: int):
291300
self.proto, self.json = proto, json
292-
self.received = max(proto, json)
301+
self.received = max(proto, json) + 1
293302

294303
def of(self, codec: _Codec) -> tuple[int, str]:
295304
if codec.name == "proto":
@@ -333,8 +342,10 @@ class _Frames:
333342
334343
``feed`` calls ``on_frame(type, flags, stream, payload)`` for each frame and returns
335344
where frames end in the data, each with the bytes ``on_frame`` returned, to be written
336-
right after that frame in this direction. The payload is read for SETTINGS and GOAWAY
337-
only (at most ``_CONTROL_PAYLOAD`` bytes of it) and skipped otherwise. A
345+
right after that frame in this direction. The payload passed is, for a SETTINGS
346+
frame, its last SETTINGS_MAX_CONCURRENT_STREAMS entry if it has one (the frame is
347+
scanned whole as it streams, holding one entry), for a GOAWAY its first
348+
``_CONTROL_PAYLOAD`` bytes, and empty otherwise. A
338349
``_ProtocolError`` from ``on_frame`` stops the feed; it is kept in ``error`` and the
339350
ends before the offending frame are returned."""
340351

@@ -346,14 +357,16 @@ def __init__(self, on_frame):
346357
self._payload = bytearray()
347358
self._keep = 0 # payload bytes still to read
348359
self._skip = 0 # payload bytes still to skip
360+
self._scan = 0 # SETTINGS payload bytes still to scan
361+
self._entry = bytearray() # a SETTINGS entry split across reads
349362
self._in_block = False # inside a header block: HEADERS ... CONTINUATION
350363
self.error: _ProtocolError | None = None
351364

352365
@property
353366
def at_boundary(self) -> bool:
354367
"""Between two frames, and outside a header block: where another frame may go
355368
(RFC 9113 section 6.10: nothing comes between a HEADERS and its CONTINUATION)."""
356-
return not self._header and not self._keep and not self._skip and not self._in_block
369+
return not (self._header or self._keep or self._skip or self._scan or self._in_block)
357370

358371
def feed(self, data: bytes, every_end: bool = False) -> list[tuple[int, bytes]]:
359372
"""Where frames end in ``data``: each end where ``on_frame`` returned bytes to
@@ -379,6 +392,14 @@ def feed(self, data: bytes, every_end: bool = False) -> list[tuple[int, bytes]]:
379392
at += step
380393
if self._skip:
381394
break
395+
elif self._scan:
396+
step = min(self._scan, size - at)
397+
self._scan_settings(data[at : at + step])
398+
at += step
399+
if self._scan:
400+
break
401+
self._entry.clear()
402+
self._emit()
382403
elif self._header or size - at < _FRAME.size:
383404
step = min(_FRAME.size - len(self._header), size - at)
384405
self._header += data[at : at + step]
@@ -390,7 +411,7 @@ def feed(self, data: bytes, every_end: bool = False) -> list[tuple[int, bytes]]:
390411
else: # a whole frame header in this chunk: read it in place
391412
self._start(*_FRAME.unpack_from(data, at))
392413
at += _FRAME.size
393-
if not (self._keep or self._skip or self._in_block):
414+
if not (self._keep or self._skip or self._scan or self._in_block):
394415
clear = at
395416
if every_end or self._after:
396417
ends.append((at, self._after))
@@ -407,22 +428,37 @@ def _start(self, high, low, kind, flags, stream):
407428
self._frame = (kind, flags, stream & 0x7FFFFFFF)
408429
if kind in (_HEADERS, _PUSH_PROMISE, _CONTINUATION):
409430
self._in_block = not flags & _END_HEADERS_FLAG
410-
self._keep = min(length, _CONTROL_PAYLOAD) if kind in (_SETTINGS, _GOAWAY) else 0
411-
self._skip = length - self._keep
412-
if not self._keep:
431+
self._keep = min(length, _CONTROL_PAYLOAD) if kind == _GOAWAY else 0
432+
self._scan = length if kind == _SETTINGS else 0
433+
self._skip = length - self._keep - self._scan
434+
if not (self._keep or self._scan):
413435
self._emit()
414436

437+
def _scan_settings(self, data: bytes) -> None:
438+
"""Read SETTINGS entries as they stream; keep the last MAX_CONCURRENT_STREAMS."""
439+
self._scan -= len(data)
440+
self._entry += data
441+
whole = len(self._entry) - len(self._entry) % 6
442+
for at in range(0, whole, 6):
443+
if self._entry[at : at + 2] == b"\x00\x03":
444+
self._payload[:] = self._entry[at : at + 6]
445+
del self._entry[:whole]
446+
415447
def _emit(self):
416448
payload = bytes(self._payload)
417449
self._payload.clear()
418450
self._after = self._on_frame(*self._frame, payload) or b""
419451

420452

421-
async def _server_preface(reader) -> bytes:
422-
"""The server's first frame, whole (RFC 9113 section 3.4: a SETTINGS frame)."""
453+
async def _first_frame(reader) -> bytes:
454+
"""A peer's first frame, whole: its SETTINGS (RFC 9113 section 3.4), within the frame
455+
size any peer may send before it has the other's."""
423456
header = await reader.readexactly(_FRAME.size)
424457
high, low, *_ = _FRAME.unpack(header)
425-
return header + await reader.readexactly(high << 16 | low)
458+
length = high << 16 | low
459+
if length > _INITIAL_MAX_FRAME:
460+
raise _ProtocolError(f"a first frame of {length} bytes")
461+
return header + await reader.readexactly(length)
426462

427463

428464
def _with_max_streams(frame: bytes, limit: int) -> bytes:
@@ -600,14 +636,33 @@ async def idle():
600636
return
601637
await asyncio.sleep(remaining if remaining > 0 else self.idle_timeout)
602638

603-
# The server's connection preface, a SETTINGS frame, before the client is read: so
639+
# The client's SETTINGS, which completes its preface, goes to the server first: a
640+
# server may wait for it before it sends its own.
641+
opening_errors = (
642+
TimeoutError,
643+
asyncio.IncompleteReadError,
644+
ConnectionError,
645+
_ProtocolError,
646+
)
647+
try:
648+
async with asyncio.timeout(self.read_timeout):
649+
opening = await _first_frame(reader)
650+
except opening_errors as exc:
651+
_log.info("connect face: the client sent no SETTINGS: %r", exc)
652+
upstream_writer.close()
653+
return
654+
client.feed(opening)
655+
upstream_writer.write(opening)
656+
# Then the server's preface, its SETTINGS, before the client is read further: so
604657
# the face knows the server's limit before the client's first call, and a client
605658
# of a server that advertises none learns the face's.
606659
try:
607660
async with asyncio.timeout(self.read_timeout):
608-
preface = await _server_preface(upstream_reader)
609-
except (TimeoutError, asyncio.IncompleteReadError, ConnectionError) as exc:
610-
_log.info("connect face: the gRPC server sent no SETTINGS: %r", exc)
661+
preface = await _first_frame(upstream_reader)
662+
except opening_errors as exc:
663+
reason = f"the gRPC server sent no SETTINGS within {self.read_timeout:g} s"
664+
_log.warning("connect face: %s: %r", reason, exc)
665+
writer.write(_goaway(0, _INTERNAL_ERROR, reason))
611666
upstream_writer.close()
612667
return
613668
server.feed(preface)

‎testdata/connect/test_connect_gen.py‎

Lines changed: 75 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -478,6 +478,30 @@ async def test_a_refusal_the_service_passes_on_keeps_its_code_and_words(face):
478478
assert r.json() == {"code": "resource_exhausted", "message": PASSED_ON}
479479

480480

481+
async def test_a_refusal_naming_the_faces_own_bound_is_still_the_services():
482+
# The face's bounds both 4 MiB, the same bound the service's client names: the
483+
# face's channel refuses only past one byte more, so the number cannot be its own.
484+
server = grpc.aio.server(interceptors=(Proof(),))
485+
server.add_generic_rpc_handlers((_handler(),))
486+
grpc_port = server.add_insecure_port("127.0.0.1:0")
487+
await server.start()
488+
listener = await connect_gen.serve(
489+
0, grpc_port, host="127.0.0.1", max_response_bytes=4194304, max_json_response_bytes=4194304
490+
)
491+
try:
492+
async with httpx.AsyncClient() as client:
493+
r = await client.post(
494+
_url(listener, "Say"),
495+
json={"text": "passing-on"},
496+
headers={"content-type": "application/json", **STAMP},
497+
)
498+
assert r.status_code == 429
499+
assert r.json() == {"code": "resource_exhausted", "message": PASSED_ON}
500+
finally:
501+
await listener.stop()
502+
await server.stop(None)
503+
504+
481505
@pytest.mark.parametrize("option", ["max_response_bytes", "max_json_response_bytes"])
482506
@pytest.mark.parametrize("value", [0, -1])
483507
async def test_an_answer_bound_is_a_positive_size(option, value):
@@ -609,7 +633,7 @@ async def drip(reader, writer):
609633
)
610634
try:
611635
reader, writer = await asyncio.open_connection("127.0.0.1", listener.port)
612-
writer.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
636+
writer.write(PREFACE + SETTINGS)
613637
await writer.drain()
614638
got = read_frames(await asyncio.wait_for(reader.read(), 3))
615639
assert [f[3] for f in got if f[0] == 0x6] == [bytes([n]) * 8 for n in range(6)]
@@ -715,11 +739,11 @@ async def silent_upstream():
715739
``state["mid_frame"]`` (an event) makes it send half a frame first and the rest when
716740
the event is set."""
717741
held, faces = [], []
718-
state = {"close_after": None, "received": bytearray(), "mid_frame": None}
742+
state = {"close_after": None, "received": bytearray(), "mid_frame": None, "preface": SETTINGS}
719743

720744
async def accept(reader, writer):
721745
held.append(writer)
722-
writer.write(SETTINGS) # a server's connection preface
746+
writer.write(state["preface"]) # a server's connection preface
723747
if state["mid_frame"] is not None:
724748
# Half a PING, the rest once the test sets the event: the server is mid-frame.
725749
ping = h2_frame(0x6, 0, 0, b"12345678")
@@ -892,6 +916,54 @@ async def accept(reader, writer):
892916
upstream.close()
893917

894918

919+
async def test_a_server_that_waits_for_the_clients_settings_is_not_stalled():
920+
# A server may send its SETTINGS only once it has the client's: the face forwards
921+
# the client's first, so the two prefaces cross.
922+
async def accept(reader, writer):
923+
await reader.readexactly(len(PREFACE) + len(SETTINGS))
924+
writer.write(h2_frame(0x4, 0, 0, struct.pack(">HI", 0x3, 7)))
925+
await writer.drain()
926+
while await reader.read(65536):
927+
pass
928+
929+
upstream = await asyncio.start_server(accept, "127.0.0.1", 0)
930+
listener = await connect_gen.serve(
931+
0, upstream.sockets[0].getsockname()[1], host="127.0.0.1", read_timeout=5
932+
)
933+
try:
934+
_, got = await exchange(listener, PREFACE + SETTINGS, 1)
935+
assert [_settings(f) for f in read_frames(got) if f[0] == 0x4] == [[(0x3, 7)]]
936+
finally:
937+
await listener.stop()
938+
upstream.close()
939+
940+
941+
async def test_a_server_that_sends_no_settings_gets_the_client_a_goaway(silent_upstream):
942+
face, state = silent_upstream
943+
state["preface"] = b"" # a server that never sends its SETTINGS
944+
closed, got = await exchange(await face(read_timeout=0.5), PREFACE + SETTINGS, 2)
945+
assert closed
946+
((kind, _, _, payload),) = read_frames(got)
947+
last, code = struct.unpack_from(">II", payload)
948+
assert (kind, last, code) == (0x7, 0, 0x2)
949+
assert payload[8:] == b"the gRPC server sent no SETTINGS within 0.5 s"
950+
951+
952+
async def test_the_servers_limit_is_found_anywhere_in_its_settings():
953+
# Past the first 256 bytes, and in a frame split across reads.
954+
entries = b"".join(struct.pack(">HI", 0xF000 + n, n) for n in range(60))
955+
server_settings = h2_frame(0x4, 0, 0, entries + struct.pack(">HI", 0x3, 7))
956+
to_client, _ = await _first_frames(server_settings, PREFACE + SETTINGS)
957+
assert _settings(to_client[0])[-1] == (0x3, 7)
958+
assert (0x3, connect_gen.DEFAULT_MAX_OPEN_CALLS) not in _settings(to_client[0])
959+
960+
seen = []
961+
frames = connect_gen._Frames(lambda kind, flags, stream, payload: seen.append(payload))
962+
for at in range(len(server_settings)):
963+
frames.feed(server_settings[at : at + 1])
964+
assert seen == [struct.pack(">HI", 0x3, 7)] and frames.at_boundary
965+
966+
895967
async def test_a_server_advertising_no_limit_is_given_the_faces():
896968
# grpcio advertises no SETTINGS_MAX_CONCURRENT_STREAMS: the client is told the face's
897969
# DEFAULT_MAX_OPEN_CALLS in the server's own SETTINGS, so it queues within it.

0 commit comments

Comments
 (0)