fix(connect): refusals the client has not read pause the client, never close it

Review r41 of #110:

- B1: a refusal waits for the end of the server's frame in flight, so how
  many wait depends on how the server's reads split, not on the client. The
  1024-refusal cap closed clients that were reading (after ~55k-63k
  refusals: 1000 held calls lost, no GOAWAY). Now the refusals are queued as
  bytes, and past _MAX_QUEUED_FOR_CLIENT (16 KiB) the client pump waits for
  the server pump to write them before it reads the client again: a client
  that reads keeps its calls, and one that does not stalls itself while the
  face holds at most the cap plus one read's refusals.
- m1: the even-stream GOAWAY is queued behind the server's frame like the
  refusals, and the relay waits (at most _GOAWAY_GRACE_S) for it to be
  written before closing.
- m2: the scaffold's comment says most of a burst queues and the rest come
  back REFUSED_STREAM, safe to retry.
- m3: test_a_reading_client_past_the_limit_keeps_its_calls (3000 refusals
  behind a server frame in flight, all delivered, connection kept),
  test_a_client_that_reads_nothing_is_not_read (the face stops reading; the
  client's writes back up) and
  test_the_even_stream_goaway_waits_for_the_servers_frame each fail on
  0aca1b9; test_at_the_scaffolds_limit_a_burst_fails_only_retry_safely
  (limit 1000, 1200 calls, every failure REFUSED_STREAM).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: antoinetoussaint-byte

templates/connect/connect_gen.py

@@ -73,9 +73,13 @@
 # refused alone, with REFUSED_STREAM, which a client retries safely: the server never
 # processed it. What the face tracks per connection is bounded by it.
 DEFAULT_MAX_OPEN_CALLS = 1000
-# Frames for the client (refusals) the relay holds while the server's frame in flight
-# completes; a client that makes it hold more is closed.
-_MAX_QUEUED_FOR_CLIENT = 1024
+# 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
+# they are written, so a client that opens calls faster than it reads them stalls itself.
+_MAX_QUEUED_FOR_CLIENT = 16 * 1024
+# How long a closing relay waits for the server's frame in flight to complete, so the
+# client's GOAWAY lands between two frames.
+_GOAWAY_GRACE_S = 5.0
 # The most of a SETTINGS or GOAWAY payload the relay reads; the rest is skipped.
 _CONTROL_PAYLOAD = 256
 _ENVELOPE = struct.Struct(">BI")
@@ -338,9 +342,11 @@ async def _relay(self, reader, writer, head):
         what the server allows at once (SETTINGS_MAX_CONCURRENT_STREAMS,
         ``DEFAULT_MAX_OPEN_CALLS`` when it advertises none) is refused alone: the client
         gets RST_STREAM REFUSED_STREAM, and the server its header block, which keeps the
-        connection's header compression in step, then the same reset. A client that opens
-        an even stream, which no HTTP/2 client does, gets a GOAWAY naming why, and the
-        connection is closed."""
+        connection's header compression in step, then the same reset. A refusal is
+        written between two of the server's frames; 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()
@@ -349,7 +355,8 @@ async def _relay(self, reader, writer, head):
         highest = accepted = 0
         refusing = 0  # the refused stream whose header block is still arriving
         max_calls = DEFAULT_MAX_OPEN_CALLS
-        to_client: list[bytes] = []  # frames for the client, written at a frame boundary
+        to_client = bytearray()  # frames for the client, written at a frame boundary
+        flushed = asyncio.Event()  # set when the server pump wrote them
         told = False  # the face wrote to the client since the client pump last drained
 
         def from_client(kind, flags, stream, payload):
@@ -395,24 +402,29 @@ def from_server(kind, flags, stream, payload):
 
         def tell_client(frame):
             nonlocal told
-            # Only between two of the server's frames: queued while one is half relayed.
-            # A client that keeps opening calls while it reads nothing is not served.
+            # Only between two of the server's frames: queued while one is half relayed,
+            # and written by the server pump at that frame's end.
             if server.at_boundary and not to_client:
                 writer.write(frame)
                 told = True
-            elif len(to_client) >= _MAX_QUEUED_FOR_CLIENT:
-                raise _ProtocolError("the client opens calls past the limit and reads nothing")
             else:
-                to_client.append(frame)
+                to_client.extend(frame)
+
+        async def written():
+            """Wait until the server pump has written what is queued for the client."""
+            while to_client:
+                flushed.clear()
+                await flushed.wait()
 
         def relay(chunk, sink, frames, ends, queued):
             at = 0
             for end, after in ends:
                 sink.write(chunk[at:end] + after)
                 at = end
                 if queued:
-                    sink.write(b"".join(queued))
+                    sink.write(bytes(queued))
                     queued.clear()
+                    flushed.set()
             if frames.error is None:
                 sink.write(chunk[at:])
 
@@ -424,14 +436,20 @@ async def pump(source, sink, frames, queued):
                     relay(chunk, sink, frames, frames.feed(chunk), queued)
                     if frames.error is not None:
                         _log.info("connect face: closing a gRPC connection: %s", frames.error)
-                        if server.at_boundary:
-                            reason = str(frames.error)
-                            writer.write(_goaway(accepted, _PROTOCOL_ERROR, reason))
+                        tell_client(_goaway(accepted, _PROTOCOL_ERROR, str(frames.error)))
+                        try:
+                            await asyncio.wait_for(written(), _GOAWAY_GRACE_S)
+                        except TimeoutError:
+                            pass
                         return
                     await sink.drain()
-                    if sink is not writer and told:
-                        told = False
-                        await writer.drain()  # the refusals: a client must read them
+                    if sink is not writer:
+                        # The refusals: a client must read them before it is read again.
+                        if len(to_client) >= _MAX_QUEUED_FOR_CLIENT:
+                            await written()
+                        if told:
+                            told = False
+                            await writer.drain()
             except (ConnectionError, OSError):
                 pass
 
@@ -445,7 +463,7 @@ async def idle():
                 await asyncio.sleep(remaining if remaining > 0 else self.idle_timeout)
 
         tasks = (
-            asyncio.ensure_future(pump(reader, upstream_writer, client, [])),
+            asyncio.ensure_future(pump(reader, upstream_writer, client, bytearray())),
             asyncio.ensure_future(pump(upstream_reader, writer, server, to_client)),
             asyncio.ensure_future(idle()),
         )

templates/grpc/code/src/rpc/server.py.tmpl

@@ -31,8 +31,10 @@ def register(server: grpc.aio.Server) -> None:
 
 
 # How many calls one client connection may have open at once, advertised to clients
-# (SETTINGS_MAX_CONCURRENT_STREAMS) so a burst queues in the client instead of being
-# refused. The Connect face reads the same setting and enforces it.
+# (SETTINGS_MAX_CONCURRENT_STREAMS) so most of a burst queues in the client. The rest
+# come back REFUSED_STREAM (grpcio counts a stream a little longer than the client
+# does), which the server never processed and a client may safely retry. The Connect
+# face reads the same setting and enforces it.
 MAX_CONCURRENT_STREAMS = 1000
 
 

testdata/connect/test_connect_gen.py

@@ -3,6 +3,7 @@
 
 import asyncio
 import base64
+import collections
 import json
 import struct
 
@@ -437,12 +438,25 @@ def opens(stream):
 async def silent_upstream():
     """A gRPC address that accepts, reads everything and never answers, so only the face's
     own rules end a connection relayed to it, and ``face(**options)`` serving over it.
-    ``state["close_after"]`` makes it hang up after that many seconds instead."""
+    ``state["close_after"]`` makes it hang up after that many seconds instead, and
+    ``state["mid_frame"]`` (an event) makes it send half a frame first and the rest when
+    the event is set."""
     held, faces = [], []
-    state = {"close_after": None, "received": bytearray()}
+    state = {"close_after": None, "received": bytearray(), "mid_frame": None}
 
     async def accept(reader, writer):
         held.append(writer)
+        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")
+            writer.write(ping[:13])
+            await writer.drain()
+
+            async def finish():
+                await state["mid_frame"].wait()
+                writer.write(ping[13:])
+
+            asyncio.ensure_future(finish())
         if state["close_after"] is not None:
             await asyncio.sleep(state["close_after"])
             writer.close()
@@ -635,3 +649,141 @@ async def test_the_relay_ends_when_the_grpc_server_hangs_up(silent_upstream):
         assert await asyncio.wait_for(reader.read(), 1) == b""
     finally:
         writer.close()
+
+
+class FrameReader:
+    """Reads a connection's frames as they come, counting each type."""
+
+    def __init__(self, reader):
+        self.kinds = collections.Counter()
+        self.frames = []
+        self.eof = False
+        self._task = asyncio.ensure_future(self._read(reader))
+
+    async def _read(self, reader):
+        buffer = bytearray()
+        while chunk := await reader.read(65536):
+            buffer += chunk
+            while len(buffer) >= 9:
+                high, low, kind, flags, stream = struct.unpack_from(">BHBBI", buffer)
+                size = high << 16 | low
+                if len(buffer) < 9 + size:
+                    break
+                self.kinds[kind] += 1
+                if kind != 0x3:
+                    self.frames.append((kind, flags, stream, bytes(buffer[9 : 9 + size])))
+                del buffer[: 9 + size]
+        self.eof = True
+
+    async def until(self, condition, seconds):
+        async with asyncio.timeout(seconds):
+            while not condition():
+                await asyncio.sleep(0.02)
+
+    def close(self):
+        self._task.cancel()
+
+
+async def test_a_reading_client_past_the_limit_keeps_its_calls(silent_upstream):
+    # The server is mid-frame, so every refusal waits for that frame's end: far more of
+    # them than the face holds. A client that reads keeps its connection and its calls,
+    # and gets every refusal once the frame completes.
+    face, state = silent_upstream
+    state["mid_frame"] = asyncio.Event()
+    listener = await face()
+    calls, past = connect_gen.DEFAULT_MAX_OPEN_CALLS, 3000
+    reader, writer = await asyncio.open_connection("127.0.0.1", listener.port)
+    frames = FrameReader(reader)
+    try:
+        writer.write(PREFACE + SETTINGS + b"".join(opens(2 * n + 1) for n in range(calls)))
+        await writer.drain()
+        await asyncio.sleep(0.2)
+        writer.write(b"".join(opens(2 * n + 1) for n in range(calls, calls + past)))
+        await writer.drain()
+        await asyncio.sleep(0.3)
+        assert not frames.eof and listener._connections
+        state["mid_frame"].set()
+        await frames.until(lambda: frames.kinds[0x3] == past, 5)
+        assert not frames.eof and listener._connections
+        upstream = read_frames(bytes(state["received"])[len(PREFACE) :])
+        assert sum(1 for f in upstream if f[0] == 0x1) == calls + past
+    finally:
+        frames.close()
+        writer.close()
+
+
+async def test_a_client_that_reads_nothing_is_not_read(silent_upstream):
+    # The refusals cannot be written (the server stays mid-frame): past what the face
+    # holds it stops reading the client, so the client's own writes back up and the
+    # face's memory does not grow with them.
+    face, state = silent_upstream
+    state["mid_frame"] = asyncio.Event()
+    listener = await face()
+    calls = connect_gen.DEFAULT_MAX_OPEN_CALLS
+    reader, writer = await asyncio.open_connection("127.0.0.1", listener.port)
+    try:
+        writer.write(PREFACE + SETTINGS + b"".join(opens(2 * n + 1) for n in range(calls)))
+        await writer.drain()
+        await asyncio.sleep(0.2)
+        flood = b"".join(opens(2 * n + 1) for n in range(calls, calls + 1_000_000))
+        writer.write(flood)
+        with pytest.raises(TimeoutError):
+            await asyncio.wait_for(writer.drain(), 2)
+        forwarded = read_frames(bytes(state["received"])[len(PREFACE) :])
+        assert sum(1 for f in forwarded if f[0] == 0x1) < calls + 20_000
+        assert listener._connections
+    finally:
+        writer.transport.abort()
+
+
+async def test_the_even_stream_goaway_waits_for_the_servers_frame(silent_upstream):
+    face, state = silent_upstream
+    state["mid_frame"] = asyncio.Event()
+    listener = await face()
+    reader, writer = await asyncio.open_connection("127.0.0.1", listener.port)
+    frames = FrameReader(reader)
+    try:
+        writer.write(PREFACE + SETTINGS + opens(1))
+        await writer.drain()
+        await asyncio.sleep(0.2)
+        writer.write(opens(2))
+        await writer.drain()
+        await asyncio.sleep(0.2)
+        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
+        last, code = struct.unpack_from(">II", frames.frames[-1][3])
+        assert (last, code) == (1, 0x1)
+    finally:
+        frames.close()
+        writer.close()
+
+
+async def test_at_the_scaffolds_limit_a_burst_fails_only_retry_safely():
+    # The scaffold advertises 1000: most of a 1200-call burst queues in the client, and
+    # any call grpcio or the face refuses comes back REFUSED_STREAM, never processed.
+    server = grpc.aio.server(options=[("grpc.max_concurrent_streams", 1000)])
+    server.add_generic_rpc_handlers((_handler(),))
+    grpc_port = server.add_insecure_port("127.0.0.1:0")
+    await server.start()
+    listener = await connect_gen.serve(0, grpc_port, host="127.0.0.1")
+    try:
+        async with grpc.aio.insecure_channel(f"127.0.0.1:{listener.port}") as channel:
+            say_call = channel.unary_unary(
+                "/echo.v1.Echo/Say",
+                request_serializer=Msg.SerializeToString,
+                response_deserializer=Msg.FromString,
+            )
+            answers = await asyncio.gather(
+                *[say_call(Msg(text=str(n)), timeout=30) for n in range(1200)],
+                return_exceptions=True,
+            )
+        refused = [a for a in answers if isinstance(a, Exception)]
+        assert all("error code 7" in a.details() for a in refused), {
+            a.details() for a in refused
+        }
+        assert len(refused) < 600
+    finally:
+        await listener.stop()
+        await server.stop(None)