Skip to content

Commit bfee337

Browse files
Merge pull request #118 from codefly-dev/fix/connect-minors
fix(connect): the minors of #111 and #117's reviews
2 parents 097a4fc + 4ecf543 commit bfee337

2 files changed

Lines changed: 331 additions & 33 deletions

File tree

‎templates/connect/connect_gen.py‎

Lines changed: 120 additions & 27 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,14 @@
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, _FRAME_SIZE_ERROR = 0x1, 0x2, 0x6
100+
_CANCEL, _ENHANCE_YOUR_CALM = 0x8, 0xB
101+
# The largest frame a client may send before it has the server's SETTINGS (RFC 9113
102+
# 4.2); the face holds the server's first frame to it too, though the client's SETTINGS
103+
# could allow more (a SETTINGS frame that size would carry over 2,700 entries).
104+
_INITIAL_MAX_FRAME = 16384
105+
# gRPC's channel arguments are C ints.
106+
_GRPC_MAX_INT = 2**31 - 1
97107
# Calls one relayed connection may hold open when the gRPC server advertises no
98108
# SETTINGS_MAX_CONCURRENT_STREAMS (grpcio advertises none unless the service sets
99109
# grpc.max_concurrent_streams, as the scaffold's server does). The face advertises it to
@@ -113,7 +123,8 @@
113123
# How long a closing relay waits for the server's frame in flight to complete, so the
114124
# client's GOAWAY lands between two frames.
115125
_GOAWAY_GRACE_S = 5.0
116-
# The most of a SETTINGS or GOAWAY payload the relay reads; the rest is skipped.
126+
# The most of a GOAWAY payload the relay reads (the last stream, the code, the start of
127+
# the reason); the rest is skipped. A SETTINGS frame is scanned whole, keeping nothing.
117128
_CONTROL_PAYLOAD = 256
118129
_ENVELOPE = struct.Struct(">BI")
119130
_COMPRESSED, _END_STREAM, _TRAILERS = 0x01, 0x02, 0x80
@@ -285,11 +296,15 @@ def _positive(name: str, value: int | None, default: int, message_bound: int) ->
285296

286297
class _AnswerBounds:
287298
"""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."""
299+
them as they are, ``json`` for one that converts them, checked before converting.
300+
``received`` is the channel's own receive bound: one byte past both, so the face
301+
checks every bound itself, and the number grpcio names when it refuses past it is
302+
one no service's client has by chance with the same text. A bound at gRPC's largest
303+
(2**31 - 1, its "unlimited") keeps the channel there."""
289304

290305
def __init__(self, proto: int, json: int):
291306
self.proto, self.json = proto, json
292-
self.received = max(proto, json)
307+
self.received = min(max(proto, json) + 1, _GRPC_MAX_INT)
293308

294309
def of(self, codec: _Codec) -> tuple[int, str]:
295310
if codec.name == "proto":
@@ -333,27 +348,33 @@ class _Frames:
333348
334349
``feed`` calls ``on_frame(type, flags, stream, payload)`` for each frame and returns
335350
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
351+
right after that frame in this direction. The payload passed is, for a SETTINGS
352+
frame, its last SETTINGS_MAX_CONCURRENT_STREAMS entry if it has one (the frame is
353+
scanned whole as it streams, holding one entry, when ``scan_settings``: the server's
354+
direction only, since only the server's limit matters), for a GOAWAY its first
355+
``_CONTROL_PAYLOAD`` bytes, and empty otherwise. A
338356
``_ProtocolError`` from ``on_frame`` stops the feed; it is kept in ``error`` and the
339357
ends before the offending frame are returned."""
340358

341-
def __init__(self, on_frame):
359+
def __init__(self, on_frame, *, scan_settings: bool = False):
342360
self._on_frame = on_frame
361+
self._scan_settings_frames = scan_settings
343362
self._header = bytearray()
344363
self._frame = None # (type, flags, stream) while its payload is read
345364
self._after = b"" # what on_frame returned for it
346365
self._payload = bytearray()
347366
self._keep = 0 # payload bytes still to read
348367
self._skip = 0 # payload bytes still to skip
368+
self._scan = 0 # SETTINGS payload bytes still to scan
369+
self._entry = bytearray() # a SETTINGS entry split across reads
349370
self._in_block = False # inside a header block: HEADERS ... CONTINUATION
350371
self.error: _ProtocolError | None = None
351372

352373
@property
353374
def at_boundary(self) -> bool:
354375
"""Between two frames, and outside a header block: where another frame may go
355376
(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
377+
return not (self._header or self._keep or self._skip or self._scan or self._in_block)
357378

358379
def feed(self, data: bytes, every_end: bool = False) -> list[tuple[int, bytes]]:
359380
"""Where frames end in ``data``: each end where ``on_frame`` returned bytes to
@@ -379,6 +400,14 @@ def feed(self, data: bytes, every_end: bool = False) -> list[tuple[int, bytes]]:
379400
at += step
380401
if self._skip:
381402
break
403+
elif self._scan:
404+
step = min(self._scan, size - at)
405+
self._scan_settings(data[at : at + step])
406+
at += step
407+
if self._scan:
408+
break
409+
self._entry.clear()
410+
self._emit()
382411
elif self._header or size - at < _FRAME.size:
383412
step = min(_FRAME.size - len(self._header), size - at)
384413
self._header += data[at : at + step]
@@ -390,7 +419,7 @@ def feed(self, data: bytes, every_end: bool = False) -> list[tuple[int, bytes]]:
390419
else: # a whole frame header in this chunk: read it in place
391420
self._start(*_FRAME.unpack_from(data, at))
392421
at += _FRAME.size
393-
if not (self._keep or self._skip or self._in_block):
422+
if not (self._keep or self._skip or self._scan or self._in_block):
394423
clear = at
395424
if every_end or self._after:
396425
ends.append((at, self._after))
@@ -407,22 +436,63 @@ def _start(self, high, low, kind, flags, stream):
407436
self._frame = (kind, flags, stream & 0x7FFFFFFF)
408437
if kind in (_HEADERS, _PUSH_PROMISE, _CONTINUATION):
409438
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:
439+
self._keep = min(length, _CONTROL_PAYLOAD) if kind == _GOAWAY else 0
440+
self._scan = length if kind == _SETTINGS and self._scan_settings_frames else 0
441+
self._skip = length - self._keep - self._scan
442+
if not (self._keep or self._scan):
413443
self._emit()
414444

445+
def _scan_settings(self, data: bytes) -> None:
446+
"""Read SETTINGS entries as they stream; keep the last MAX_CONCURRENT_STREAMS."""
447+
self._scan -= len(data)
448+
self._entry += data
449+
whole = len(self._entry) - len(self._entry) % 6
450+
for at in range(0, whole, 6):
451+
if self._entry[at : at + 2] == b"\x00\x03":
452+
self._payload[:] = self._entry[at : at + 6]
453+
del self._entry[:whole]
454+
415455
def _emit(self):
416456
payload = bytes(self._payload)
417457
self._payload.clear()
418458
self._after = self._on_frame(*self._frame, payload) or b""
419459

420460

421-
async def _server_preface(reader) -> bytes:
422-
"""The server's first frame, whole (RFC 9113 section 3.4: a SETTINGS frame)."""
461+
class _FrameSizeError(_ProtocolError):
462+
"""A first frame past ``_INITIAL_MAX_FRAME``."""
463+
464+
465+
async def _first_frame(reader) -> bytes:
466+
"""A peer's first frame, whole, which RFC 9113 section 3.4 makes its SETTINGS: a
467+
SETTINGS frame on stream 0, not an acknowledgement, within ``_INITIAL_MAX_FRAME``."""
423468
header = await reader.readexactly(_FRAME.size)
424-
high, low, *_ = _FRAME.unpack(header)
425-
return header + await reader.readexactly(high << 16 | low)
469+
high, low, kind, flags, stream = _FRAME.unpack(header)
470+
length = high << 16 | low
471+
if length > _INITIAL_MAX_FRAME:
472+
raise _FrameSizeError(
473+
f"a first frame of {length} bytes, past the {_INITIAL_MAX_FRAME} "
474+
"allowed before SETTINGS"
475+
)
476+
if kind != _SETTINGS or flags & _ACK_FLAG or stream:
477+
raise _ProtocolError(
478+
f"a first frame of type {kind} (flags {flags}, stream {stream}), not SETTINGS"
479+
)
480+
return header + await reader.readexactly(length)
481+
482+
483+
def _opening_refusal(exc: BaseException, peer: str, timeout: float) -> str:
484+
"""Why a peer's opening failed, in words an operator can act on."""
485+
if isinstance(exc, TimeoutError):
486+
return f"the {peer} sent no SETTINGS within {timeout:g} s"
487+
if isinstance(exc, _ProtocolError):
488+
return f"the {peer} sent {exc}"
489+
return f"the {peer} closed the connection before its SETTINGS"
490+
491+
492+
def _refuse_opening(writer, code: int, reason: str) -> None:
493+
"""Tell the client why its connection ends, as a server's first frames must: SETTINGS
494+
first (RFC 9113 section 3.4: a client discards anything else), then the GOAWAY."""
495+
writer.write(_h2_frame(_SETTINGS, 0, 0) + _goaway(0, code, reason))
426496

427497

428498
def _with_max_streams(frame: bytes, limit: int) -> bytes:
@@ -524,7 +594,7 @@ def from_server(kind, flags, stream, payload):
524594
calls.difference_update({call for call in calls if call > last_stream})
525595
return None
526596

527-
client, server = _Frames(from_client), _Frames(from_server)
597+
client, server = _Frames(from_client), _Frames(from_server, scan_settings=True)
528598

529599
def tell_client(frame):
530600
nonlocal told
@@ -600,14 +670,37 @@ async def idle():
600670
return
601671
await asyncio.sleep(remaining if remaining > 0 else self.idle_timeout)
602672

603-
# The server's connection preface, a SETTINGS frame, before the client is read: so
673+
# The client's SETTINGS, which completes its preface, goes to the server first: a
674+
# server may wait for it before it sends its own.
675+
opening_errors = (
676+
TimeoutError,
677+
asyncio.IncompleteReadError,
678+
ConnectionError,
679+
_ProtocolError,
680+
)
681+
try:
682+
async with asyncio.timeout(self.read_timeout):
683+
opening = await _first_frame(reader)
684+
except opening_errors as exc:
685+
reason = _opening_refusal(exc, "client", self.read_timeout)
686+
_log.info("connect face: %s", reason)
687+
if isinstance(exc, (TimeoutError, _ProtocolError)):
688+
code = _FRAME_SIZE_ERROR if isinstance(exc, _FrameSizeError) else _PROTOCOL_ERROR
689+
_refuse_opening(writer, code, reason)
690+
upstream_writer.close()
691+
return
692+
client.feed(opening)
693+
upstream_writer.write(opening)
694+
# Then the server's preface, its SETTINGS, before the client is read further: so
604695
# the face knows the server's limit before the client's first call, and a client
605696
# of a server that advertises none learns the face's.
606697
try:
607698
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)
699+
preface = await _first_frame(upstream_reader)
700+
except opening_errors as exc:
701+
reason = _opening_refusal(exc, "gRPC server", self.read_timeout)
702+
_log.warning("connect face: %s", reason)
703+
_refuse_opening(writer, _INTERNAL_ERROR, reason)
611704
upstream_writer.close()
612705
return
613706
server.feed(preface)

0 commit comments

Comments
 (0)