@@ -96,9 +96,14 @@
 _END_STREAM_FLAG, _ACK_FLAG, _END_HEADERS_FLAG = 0x1, 0x1, 0x4
 _MAX_CONCURRENT_STREAMS = 0x3
 _PUSH_PROMISE = 0x5
-_PROTOCOL_ERROR, _INTERNAL_ERROR, _CANCEL, _ENHANCE_YOUR_CALM = 0x1, 0x2, 0x8, 0xB
-# The largest frame a peer may send before it has the other's SETTINGS (RFC 9113 4.2).
+_PROTOCOL_ERROR, _INTERNAL_ERROR, _FRAME_SIZE_ERROR = 0x1, 0x2, 0x6
+_CANCEL, _ENHANCE_YOUR_CALM = 0x8, 0xB
+# The largest frame a client may send before it has the server's SETTINGS (RFC 9113
+# 4.2); the face holds the server's first frame to it too, though the client's SETTINGS
+# could allow more (a SETTINGS frame that size would carry over 2,700 entries).
 _INITIAL_MAX_FRAME = 16384
+# gRPC's channel arguments are C ints.
+_GRPC_MAX_INT = 2**31 - 1
 # 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). The face advertises it to
@@ -294,11 +299,12 @@ class _AnswerBounds:
     them as they are, ``json`` for one that converts them, checked before converting.
     ``received`` is the channel's own receive bound: one byte past both, so the face
     checks every bound itself, and the number grpcio names when it refuses past it is
-    one no service's client has by chance with the same text."""
+    one no service's client has by chance with the same text. A bound at gRPC's largest
+    (2**31 - 1, its "unlimited") keeps the channel there."""
 
     def __init__(self, proto: int, json: int):
         self.proto, self.json = proto, json
-        self.received = max(proto, json) + 1
+        self.received = min(max(proto, json) + 1, _GRPC_MAX_INT)
 
     def of(self, codec: _Codec) -> tuple[int, str]:
         if codec.name == "proto":
@@ -344,13 +350,15 @@ class _Frames:
     where frames end in the data, each with the bytes ``on_frame`` returned, to be written
     right after that frame in this direction. The payload passed is, for a SETTINGS
     frame, its last SETTINGS_MAX_CONCURRENT_STREAMS entry if it has one (the frame is
-    scanned whole as it streams, holding one entry), for a GOAWAY its first
+    scanned whole as it streams, holding one entry, when ``scan_settings``: the server's
+    direction only, since only the server's limit matters), for a GOAWAY its first
     ``_CONTROL_PAYLOAD`` bytes, and empty otherwise. A
     ``_ProtocolError`` from ``on_frame`` stops the feed; it is kept in ``error`` and the
     ends before the offending frame are returned."""
 
-    def __init__(self, on_frame):
+    def __init__(self, on_frame, *, scan_settings: bool = False):
         self._on_frame = on_frame
+        self._scan_settings_frames = scan_settings
         self._header = bytearray()
         self._frame = None  # (type, flags, stream) while its payload is read
         self._after = b""  # what on_frame returned for it
@@ -429,7 +437,7 @@ def _start(self, high, low, kind, flags, stream):
         if kind in (_HEADERS, _PUSH_PROMISE, _CONTINUATION):
             self._in_block = not flags & _END_HEADERS_FLAG
         self._keep = min(length, _CONTROL_PAYLOAD) if kind == _GOAWAY else 0
-        self._scan = length if kind == _SETTINGS else 0
+        self._scan = length if kind == _SETTINGS and self._scan_settings_frames else 0
         self._skip = length - self._keep - self._scan
         if not (self._keep or self._scan):
             self._emit()
@@ -450,17 +458,43 @@ def _emit(self):
         self._after = self._on_frame(*self._frame, payload) or b""
 
 
+class _FrameSizeError(_ProtocolError):
+    """A first frame past ``_INITIAL_MAX_FRAME``."""
+
+
 async def _first_frame(reader) -> bytes:
-    """A peer's first frame, whole: its SETTINGS (RFC 9113 section 3.4), within the frame
-    size any peer may send before it has the other's."""
+    """A peer's first frame, whole, which RFC 9113 section 3.4 makes its SETTINGS: a
+    SETTINGS frame on stream 0, not an acknowledgement, within ``_INITIAL_MAX_FRAME``."""
     header = await reader.readexactly(_FRAME.size)
-    high, low, *_ = _FRAME.unpack(header)
+    high, low, kind, flags, stream = _FRAME.unpack(header)
     length = high << 16 | low
     if length > _INITIAL_MAX_FRAME:
-        raise _ProtocolError(f"a first frame of {length} bytes")
+        raise _FrameSizeError(
+            f"a first frame of {length} bytes, past the {_INITIAL_MAX_FRAME} "
+            "allowed before SETTINGS"
+        )
+    if kind != _SETTINGS or flags & _ACK_FLAG or stream:
+        raise _ProtocolError(
+            f"a first frame of type {kind} (flags {flags}, stream {stream}), not SETTINGS"
+        )
     return header + await reader.readexactly(length)
 
 
+def _opening_refusal(exc: BaseException, peer: str, timeout: float) -> str:
+    """Why a peer's opening failed, in words an operator can act on."""
+    if isinstance(exc, TimeoutError):
+        return f"the {peer} sent no SETTINGS within {timeout:g} s"
+    if isinstance(exc, _ProtocolError):
+        return f"the {peer} sent {exc}"
+    return f"the {peer} closed the connection before its SETTINGS"
+
+
+def _refuse_opening(writer, code: int, reason: str) -> None:
+    """Tell the client why its connection ends, as a server's first frames must: SETTINGS
+    first (RFC 9113 section 3.4: a client discards anything else), then the GOAWAY."""
+    writer.write(_h2_frame(_SETTINGS, 0, 0) + _goaway(0, code, reason))
+
+
 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
@@ -560,7 +594,7 @@ def from_server(kind, flags, stream, payload):
                 calls.difference_update({call for call in calls if call > last_stream})
             return None
 
-        client, server = _Frames(from_client), _Frames(from_server)
+        client, server = _Frames(from_client), _Frames(from_server, scan_settings=True)
 
         def tell_client(frame):
             nonlocal told
@@ -648,7 +682,11 @@ async def idle():
             async with asyncio.timeout(self.read_timeout):
                 opening = await _first_frame(reader)
         except opening_errors as exc:
-            _log.info("connect face: the client sent no SETTINGS: %r", exc)
+            reason = _opening_refusal(exc, "client", self.read_timeout)
+            _log.info("connect face: %s", reason)
+            if isinstance(exc, (TimeoutError, _ProtocolError)):
+                code = _FRAME_SIZE_ERROR if isinstance(exc, _FrameSizeError) else _PROTOCOL_ERROR
+                _refuse_opening(writer, code, reason)
             upstream_writer.close()
             return
         client.feed(opening)
@@ -660,9 +698,9 @@ async def idle():
             async with asyncio.timeout(self.read_timeout):
                 preface = await _first_frame(upstream_reader)
         except opening_errors as exc:
-            reason = f"the gRPC server sent no SETTINGS within {self.read_timeout:g} s"
-            _log.warning("connect face: %s: %r", reason, exc)
-            writer.write(_goaway(0, _INTERNAL_ERROR, reason))
+            reason = _opening_refusal(exc, "gRPC server", self.read_timeout)
+            _log.warning("connect face: %s", reason)
+            _refuse_opening(writer, _INTERNAL_ERROR, reason)
             upstream_writer.close()
             return
         server.feed(preface)
 
@@ -4,6 +4,8 @@
 import asyncio
 import base64
 import collections
+import os
+import sys
 import json
 import struct
 
@@ -595,8 +597,9 @@ async def test_connect_json_ignores_unknown_fields(face):
     assert r.json() == {"text": "hi|maya", "count": 1}
 
 
-async def test_a_silent_grpc_connection_is_closed_after_the_idle_bound():
-    # An upstream that accepts and never answers: only the face's own bound ends it.
+async def test_a_client_that_sends_no_settings_is_told_why():
+    # A client that sends the preface's magic and then nothing: the face's read bound
+    # ends it, and the client is told why, after the SETTINGS a server's first frame is.
     upstream = await asyncio.start_server(lambda r, w: None, "127.0.0.1", 0)
     listener = await connect_gen.serve(
         0,
@@ -609,7 +612,10 @@ async def test_a_silent_grpc_connection_is_closed_after_the_idle_bound():
         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""
+        sent = read_frames(await asyncio.wait_for(reader.read(), 3))
+        assert [f[:3] for f in sent] == [(0x4, 0, 0), (0x7, 0, 0)]
+        reason = b"the client sent no SETTINGS within 0.5 s"
+        assert sent[1][3] == struct.pack(">II", 0, 0x1) + reason
         writer.close()
     finally:
         await listener.stop()
@@ -938,15 +944,134 @@ async def accept(reader, writer):
         upstream.close()
 
 
-async def test_a_server_that_sends_no_settings_gets_the_client_a_goaway(silent_upstream):
+def _refusal(got):
+    """The face's refusal of a connection: (GOAWAY code, reason), after an empty SETTINGS."""
+    frames = read_frames(got)
+    assert [f[:3] for f in frames] == [(0x4, 0, 0), (0x7, 0, 0)], frames
+    assert frames[0][3] == b""
+    last, code = struct.unpack_from(">II", frames[1][3])
+    assert last == 0
+    return code, frames[1][3][8:].decode()
+
+
+@pytest.mark.parametrize(
+    "preface, reason",
+    [
+        (None, "the gRPC server sent no SETTINGS within 0.5 s"),
+        (b"", "the gRPC server closed the connection before its SETTINGS"),
+        (
+            h2_frame(0x4, 0, 0, b"\0" * 16386),
+            "the gRPC server sent a first frame of 16386 bytes, past the 16384 allowed "
+            "before SETTINGS",
+        ),
+        (
+            h2_frame(0x6, 0, 0, b"\0" * 8),
+            "the gRPC server sent a first frame of type 6 (flags 0, stream 0), not SETTINGS",
+        ),
+    ],
+    ids=["silent", "closed", "oversized", "not-settings"],
+)
+async def test_a_server_that_sends_no_settings_gets_the_client_a_goaway(preface, reason):
+    # Whatever went wrong, the client hears why, after an empty SETTINGS (RFC 9113 3.4:
+    # a server's first frame; a client discards anything else).
+    async def accept(reader, writer):
+        if preface is not None:
+            writer.write(preface)
+            await writer.drain()
+            if preface == b"":
+                writer.close()
+                return
+        while await reader.read(65536):
+            pass
+
+    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", read_timeout=0.5
+    )
+    try:
+        closed, got = await exchange(listener, PREFACE + SETTINGS, 2)
+        assert closed
+        assert _refusal(got) == (0x2, reason)
+    finally:
+        await listener.stop()
+        upstream.close()
+
+
+_GRPC_CLIENT = """
+import sys, grpc
+channel = grpc.insecure_channel(sys.argv[1])
+try:
+    channel.unary_unary("/echo.v1.Echo/Say")(b"", timeout=2)
+except grpc.RpcError as error:
+    print(error.code())
+"""
+
+
+async def test_a_grpc_client_hears_why_the_server_never_answered():
+    # A real gRPC client, in its own process so its transport log can be read: it parses
+    # the face's refusal (a GOAWAY first would be "Expected SETTINGS frame") and logs
+    # its reason.
+    async def accept(reader, writer):
+        while await reader.read(65536):
+            pass
+
+    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", read_timeout=0.5
+    )
+    try:
+        client = await asyncio.create_subprocess_exec(
+            sys.executable,
+            "-c",
+            _GRPC_CLIENT,
+            f"127.0.0.1:{listener.port}",
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            env={**os.environ, "GRPC_VERBOSITY": "info"},
+        )
+        out, err = await asyncio.wait_for(client.communicate(), 30)
+        log = err.decode()
+        assert "Expected SETTINGS frame" not in log
+        assert "Debug Text: the gRPC server sent no SETTINGS within 0.5 s" in log, log[-2000:]
+    finally:
+        await listener.stop()
+        upstream.close()
+
+
+@pytest.mark.parametrize(
+    "opening, code, reason",
+    [
+        (
+            h2_frame(0x4, 0, 0, b"\0" * 16386),
+            0x6,
+            "the client sent a first frame of 16386 bytes, past the 16384 allowed before SETTINGS",
+        ),
+        (
+            h2_frame(0x1, 0x4, 1, b"\x82"),
+            0x1,
+            "the client sent a first frame of type 1 (flags 4, stream 1), not SETTINGS",
+        ),
+        (
+            h2_frame(0x4, 0x1, 0),
+            0x1,
+            "the client sent a first frame of type 4 (flags 1, stream 0), not SETTINGS",
+        ),
+    ],
+    ids=["oversized", "headers", "settings-ack"],
+)
+async def test_a_client_whose_first_frame_is_not_its_settings_is_refused(
+    silent_upstream, opening, code, reason
+):
     face, state = silent_upstream
-    state["preface"] = b""  # a server that never sends its SETTINGS
-    closed, got = await exchange(await face(read_timeout=0.5), PREFACE + SETTINGS, 2)
+    closed, got = await exchange(await face(), PREFACE + opening, 2)
     assert closed
-    ((kind, _, _, payload),) = read_frames(got)
-    last, code = struct.unpack_from(">II", payload)
-    assert (kind, last, code) == (0x7, 0, 0x2)
-    assert payload[8:] == b"the gRPC server sent no SETTINGS within 0.5 s"
+    assert _refusal(got) == (code, reason)
+    assert bytes(state["received"]) == PREFACE  # the frame never reached the server
+
+
+async def test_a_bound_at_grpcs_largest_starts():
+    listener = await connect_gen.serve(0, 1, host="127.0.0.1", max_response_bytes=2**31 - 1)
+    await listener.stop()
 
 
 async def test_the_servers_limit_is_found_anywhere_in_its_settings():
@@ -958,11 +1083,19 @@ async def test_the_servers_limit_is_found_anywhere_in_its_settings():
     assert (0x3, connect_gen.DEFAULT_MAX_OPEN_CALLS) not in _settings(to_client[0])
 
     seen = []
-    frames = connect_gen._Frames(lambda kind, flags, stream, payload: seen.append(payload))
+    frames = connect_gen._Frames(
+        lambda kind, flags, stream, payload: seen.append(payload), scan_settings=True
+    )
     for at in range(len(server_settings)):
         frames.feed(server_settings[at : at + 1])
     assert seen == [struct.pack(">HI", 0x3, 7)] and frames.at_boundary
 
+    # The client's direction only skips them: its SETTINGS cost nothing to read.
+    seen.clear()
+    frames = connect_gen._Frames(lambda kind, flags, stream, payload: seen.append(payload))
+    frames.feed(server_settings)
+    assert seen == [b""] and frames.at_boundary
+
 
 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