fix(connect): the face knows the server's limit before the client's first call, and tells it its own

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)),

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: