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: