Skip to content

Commit a7eefee

Browse files
fix(connect): the answer bound follows the message bound, counts what is sent, and names itself
Review r45 of #111: - f1 (blocker): max_response_bytes defaulted to a fixed 64 MiB, so a service that raised max_message_bytes above it lost answers it used to get. It now defaults to the larger of DEFAULT_MAX_RESPONSE_BYTES and max_message_bytes. - f2 (major): the default drops to 16 MiB, the bound counts the bytes the face sends (the converted document for JSON, not only the proto), and the docstring states the real cost: per call in flight the answer and its encoding, each within the bound; a JSON conversion's transient peak (about 13x the proto for a float list, measured: 3.8 MiB proto, 51 MiB peak) runs on the face's event loop one at a time; the face bounds neither calls in flight nor what the gRPC server holds. - f3: the refusal of an answer past the bound names max_response_bytes and is failed_precondition (400, grpc-status 9), not resource_exhausted (429): the same call is refused the same way when retried. - f4: the docstring is reflowed. Tests: test_a_raised_message_bound_raises_the_answer_bound (100 MiB bound, 70 MiB answer), test_max_response_bytes_counts_the_json_the_face_sends, test_the_default_answer_bound_is_16_mib, and the refusal's code and message on every media type; each fails on 434f403. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 434f403 commit a7eefee

2 files changed

Lines changed: 168 additions & 31 deletions

File tree

‎templates/connect/connect_gen.py‎

Lines changed: 71 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -19,15 +19,29 @@
1919
2020
Request bodies are bounded by ``max_message_bytes`` (gRPC's 4 MiB unless the service
2121
raises it) before they are read. An answer the face transcodes is bounded apart, by
22-
``max_response_bytes`` (64 MiB unless the service sets it), so a service that keeps its
23-
requests small still answers with what its handler returns; past it the call ends
24-
``resource_exhausted``. Both are keyword options to ``serve`` and ``start``; a relayed gRPC
25-
call keeps the service's and the caller's own bounds. Every read is bounded in time: an
26-
HTTP/1.1 read by
27-
``read_timeout`` (``idle_timeout`` between requests), and a relayed gRPC connection is
28-
closed once it has had no open call for ``idle_timeout``, as Go's HTTP/2 server bounds an
29-
idle connection. A call in flight is never cut by the face, however long its handler
30-
takes to answer: its deadline and the gRPC server's own policy bound it.
22+
``max_response_bytes``: by default the larger of ``DEFAULT_MAX_RESPONSE_BYTES`` (16 MiB)
23+
and ``max_message_bytes``, so a service that keeps its requests small still answers
24+
with what its handler returns, and one that raised its message bound keeps answers that
25+
large. The bound counts the bytes the face sends: the answer as the service encoded it
26+
for proto, the converted document for JSON. Past it the call ends ``failed_precondition``
27+
naming ``max_response_bytes``: the answer is too large for this face's configuration, and
28+
asking again does not change it. Within the bound, Connect and gRPC-Web return what the
29+
gRPC server returns. Both bounds are keyword options to ``serve`` and ``start``; a relayed
30+
gRPC call keeps the service's and the caller's own bounds.
31+
32+
What a transcoded answer costs the face: while a call is in flight it holds the answer as
33+
received and its encoding, each at most ``max_response_bytes``, until the client reads
34+
it. Converting to JSON holds more while it runs, about 13 times the proto size for a list
35+
of floats (protobuf 6.33), so about 210 MiB at the default bound; conversions run on the
36+
face's event loop, one at a time, so that peak does not add up across calls. A proto
37+
answer is not converted. The face does not bound how many calls are in flight, as the
38+
gRPC server does not.
39+
40+
Every read is bounded in time: an HTTP/1.1 read by ``read_timeout`` (``idle_timeout``
41+
between requests), and a relayed gRPC connection is closed once it has had no open call
42+
for ``idle_timeout``, as Go's HTTP/2 server bounds an idle connection. A call in flight
43+
is never cut by the face, however long its handler takes to answer: its deadline and the
44+
gRPC server's own policy bound it.
3145
A JSON request field the service's proto does not know is ignored, as connect-go's codec
3246
ignores it, so a client built from a newer proto calls a Python service and a Go one alike.
3347
@@ -62,7 +76,7 @@
6276
# The port a deployment hands the face (CODEFLY_CONNECT_PORT); ``start`` reads only the variable.
6377
DEFAULT_PORT = 8081
6478
DEFAULT_MAX_MESSAGE_BYTES = 4 * 1024 * 1024
65-
DEFAULT_MAX_RESPONSE_BYTES = 64 * 1024 * 1024
79+
DEFAULT_MAX_RESPONSE_BYTES = 16 * 1024 * 1024
6680
DEFAULT_READ_TIMEOUT_S = 30.0
6781
DEFAULT_IDLE_TIMEOUT_S = 120.0
6882

@@ -102,6 +116,8 @@
102116
)
103117
_TRANSPORT_PREFIXES = ("connect-", "grpc-", "x-grpc-web", ":")
104118
_GRPC_TIMEOUT = re.compile(r"^(\d{1,8})([HMSmun])$")
119+
# grpcio's refusal of an answer past the face's own receive bound (its channel is the client).
120+
_ANSWER_PAST_BOUND = re.compile(r"CLIENT: Received message larger than max \((\d+) vs\. (\d+)\)")
105121
_GRPC_UNITS = {"H": 3600.0, "M": 60.0, "S": 1.0, "m": 1e-3, "u": 1e-6, "n": 1e-9}
106122
_CODES = {
107123
grpc.StatusCode.CANCELLED: ("canceled", 499),
@@ -135,12 +151,18 @@ def __init__(self, code: grpc.StatusCode, message: str, header=(), trailer=()):
135151

136152
@classmethod
137153
def of(cls, error: grpc.aio.AioRpcError) -> CallError:
138-
return cls(
139-
error.code(),
140-
error.details() or "",
141-
error.initial_metadata() or (),
142-
error.trailing_metadata() or (),
143-
)
154+
code, details = error.code(), error.details() or ""
155+
past = _ANSWER_PAST_BOUND.search(details)
156+
if code is grpc.StatusCode.RESOURCE_EXHAUSTED and past:
157+
code, details = _answer_past_bound(int(past.group(1)), int(past.group(2)))
158+
return cls(code, details, error.initial_metadata() or (), error.trailing_metadata() or ())
159+
160+
161+
def _answer_past_bound(size: int, bound: int) -> tuple[grpc.StatusCode, str]:
162+
return (
163+
grpc.StatusCode.FAILED_PRECONDITION,
164+
f"the answer ({size} bytes) exceeds the Connect face's max_response_bytes ({bound})",
165+
)
144166

145167

146168
class _Plain(Exception):
@@ -174,13 +196,15 @@ async def serve(
174196
host: str = "0.0.0.0",
175197
grpc_host: str = "127.0.0.1",
176198
max_message_bytes: int = DEFAULT_MAX_MESSAGE_BYTES,
177-
max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
199+
max_response_bytes: int | None = None,
178200
read_timeout: float = DEFAULT_READ_TIMEOUT_S,
179201
idle_timeout: float = DEFAULT_IDLE_TIMEOUT_S,
180202
) -> ConnectFace:
181203
"""Start the Connect face on ``host``:``port`` (0 for an ephemeral one) over the
182204
grpc.aio server listening on ``grpc_host``:``grpc_port``, and return it started.
183205
Call it on the loop that serves the gRPC server, once that server listens."""
206+
if max_response_bytes is None:
207+
max_response_bytes = max(DEFAULT_MAX_RESPONSE_BYTES, max_message_bytes)
184208
target = f"{grpc_host}:{grpc_port}"
185209
channel = grpc.aio.insecure_channel(
186210
target,
@@ -191,7 +215,12 @@ async def serve(
191215
],
192216
)
193217
connection = _Connection(
194-
channel, (grpc_host, grpc_port), max_message_bytes, read_timeout, idle_timeout
218+
channel,
219+
(grpc_host, grpc_port),
220+
max_message_bytes,
221+
max_response_bytes,
222+
read_timeout,
223+
idle_timeout,
195224
)
196225
face: ConnectFace
197226

@@ -284,8 +313,9 @@ def _emit(self):
284313

285314

286315
class _Connection:
287-
def __init__(self, channel, grpc_address, limit, read_timeout, idle_timeout):
316+
def __init__(self, channel, grpc_address, limit, answer_limit, read_timeout, idle_timeout):
288317
self.channel, self.grpc_address, self.limit = channel, grpc_address, limit
318+
self.answer_limit = answer_limit
289319
self.read_timeout, self.idle_timeout = read_timeout, idle_timeout
290320

291321
async def serve(self, reader, writer):
@@ -466,7 +496,16 @@ async def _dispatch(self, exchange, request, body):
466496
codec = protocol.codec
467497
if codec is None:
468498
raise _Plain(415, f"unsupported codec in {content_type!r}")
469-
call = _Call(self.channel, path, method, codec, request.headers, headers, self.limit)
499+
call = _Call(
500+
self.channel,
501+
path,
502+
method,
503+
codec,
504+
request.headers,
505+
headers,
506+
self.limit,
507+
self.answer_limit,
508+
)
470509
await protocol.answer(exchange, call, body, headers)
471510

472511

@@ -496,11 +535,15 @@ def request(self, cls, data: bytes) -> bytes:
496535
except (json_format.ParseError, UnicodeDecodeError, ValueError) as exc:
497536
raise CallError(grpc.StatusCode.INVALID_ARGUMENT, f"unmarshal JSON: {exc}") from exc
498537

499-
def response(self, cls, data: bytes) -> bytes:
538+
def response(self, cls, data: bytes, limit: int) -> bytes:
539+
"""The answer in this codec; the proto as received is within ``limit`` already."""
500540
if self.name == "proto":
501541
return data
502542
document = json_format.MessageToDict(cls.FromString(data))
503-
return json.dumps(document, separators=(",", ":")).encode()
543+
encoded = json.dumps(document, separators=(",", ":")).encode()
544+
if len(encoded) > limit:
545+
raise CallError(*_answer_past_bound(len(encoded), limit))
546+
return encoded
504547

505548

506549
def _codec(name):
@@ -511,9 +554,9 @@ class _Call:
511554
"""One call dispatched through the gRPC server: request bytes in the protocol's
512555
codec, answers out in it."""
513556

514-
def __init__(self, channel, path, method, codec, raw_headers, headers, limit):
557+
def __init__(self, channel, path, method, codec, raw_headers, headers, limit, answer_limit):
515558
self.channel, self.path, self.method, self.codec = channel, path, method, codec
516-
self.limit = limit
559+
self.limit, self.answer_limit = limit, answer_limit
517560
self.input = message_factory.GetMessageClass(method.input_type)
518561
self.output = message_factory.GetMessageClass(method.output_type)
519562
self.metadata = _metadata(raw_headers)
@@ -555,7 +598,7 @@ async def unary(self, message: bytes):
555598
try:
556599
answer = await call
557600
return (
558-
self.codec.response(self.output, answer),
601+
self.codec.response(self.output, answer, self.answer_limit),
559602
tuple(await call.initial_metadata() or ()),
560603
tuple(await call.trailing_metadata() or ()),
561604
)
@@ -720,11 +763,13 @@ async def _relay(exchange, call, stream):
720763
return stream.trailer
721764
try:
722765
async for answer in stream:
723-
data = call.codec.response(call.output, answer)
766+
data = call.codec.response(call.output, answer, call.answer_limit)
724767
await exchange.send(h11.Data(data=_ENVELOPE.pack(0, len(data)) + data))
725768
return tuple(await stream.trailing_metadata() or ())
726769
except grpc.aio.AioRpcError as exc:
727770
return CallError.of(exc)
771+
except CallError as error:
772+
return error
728773
finally:
729774
stream.cancel()
730775

‎testdata/connect/test_connect_gen.py‎

Lines changed: 97 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -369,18 +369,23 @@ async def test_max_response_bytes_bounds_answers_when_set(media):
369369
)
370370
assert text == "x" * size
371371
elif "grpc-web" in media:
372-
assert "grpc-status: 8" in frames(r.content)[-1][1].decode()
373-
assert "larger than max" in frames(r.content)[-1][1].decode()
372+
trailers = frames(r.content)[-1][1].decode()
373+
assert "grpc-status: 9" in trailers # FAILED_PRECONDITION
374+
assert "max_response_bytes" in trailers
374375
else:
375376
if streaming:
376377
assert r.status_code == 200
377378
assert len(frames(r.content)) == 1
378379
error = json.loads(frames(r.content)[0][1])["error"]
379380
else:
380-
assert r.status_code == 429
381+
# Not 429: the same call is refused the same way when retried.
382+
assert r.status_code == 400
381383
error = r.json()
382-
assert error["code"] == "resource_exhausted"
383-
assert "larger than max" in error["message"], error
384+
assert error["code"] == "failed_precondition"
385+
assert error["message"] == (
386+
"the answer (4099 bytes) exceeds the Connect face's "
387+
"max_response_bytes (2048)"
388+
), error
384389
finally:
385390
if listener is not None:
386391
await listener.stop()
@@ -649,3 +654,90 @@ async def test_the_relay_ends_when_the_grpc_server_hangs_up(silent_upstream):
649654
assert await asyncio.wait_for(reader.read(), 1) == b""
650655
finally:
651656
writer.close()
657+
658+
659+
@pytest.mark.parametrize("media", ["application/json", "application/connect+json"])
660+
async def test_max_response_bytes_counts_the_json_the_face_sends(media):
661+
# 2040 characters: 2043 bytes as proto, within 2048, and 2051 as JSON, past it.
662+
server = grpc.aio.server()
663+
server.add_generic_rpc_handlers((_handler(),))
664+
grpc_port = server.add_insecure_port("127.0.0.1:0")
665+
await server.start()
666+
listener = await connect_gen.serve(
667+
0, grpc_port, host="127.0.0.1", max_message_bytes=1024, max_response_bytes=2048
668+
)
669+
try:
670+
streaming = media.startswith("application/connect+")
671+
body = json.dumps({"text": "big", "count": 2040}).encode()
672+
async with httpx.AsyncClient() as client:
673+
proto = await client.post(
674+
_url(listener, "Say"),
675+
content=Msg(text="big", count=2040).SerializeToString(),
676+
headers={"content-type": "application/proto"},
677+
)
678+
r = await client.post(
679+
_url(listener, "Count" if streaming else "Say"),
680+
content=envelope(body) if streaming else body,
681+
headers={"content-type": media},
682+
)
683+
assert proto.status_code == 200 and len(proto.content) == 2043
684+
error = json.loads(frames(r.content)[-1][1])["error"] if streaming else r.json()
685+
assert r.status_code == (200 if streaming else 400)
686+
assert error == {
687+
"code": "failed_precondition",
688+
"message": "the answer (2051 bytes) exceeds the Connect face's max_response_bytes (2048)",
689+
}
690+
finally:
691+
await listener.stop()
692+
await server.stop(None)
693+
694+
695+
async def test_a_raised_message_bound_raises_the_answer_bound():
696+
# max_response_bytes defaults to the larger of DEFAULT_MAX_RESPONSE_BYTES and
697+
# max_message_bytes: a service that raised its bound to 100 MiB answers 70 MiB.
698+
server = grpc.aio.server(options=[("grpc.max_send_message_length", 100 * 2**20)])
699+
server.add_generic_rpc_handlers((_handler(),))
700+
grpc_port = server.add_insecure_port("127.0.0.1:0")
701+
await server.start()
702+
listener = await connect_gen.serve(
703+
0, grpc_port, host="127.0.0.1", max_message_bytes=100 * 2**20
704+
)
705+
try:
706+
size = 70 * 2**20
707+
async with httpx.AsyncClient(timeout=60) as client:
708+
r = await client.post(
709+
_url(listener, "Say"),
710+
content=Msg(text="big", count=size).SerializeToString(),
711+
headers={"content-type": "application/proto"},
712+
)
713+
assert r.status_code == 200, r.text[:200]
714+
assert len(Msg.FromString(r.content).text) == size
715+
finally:
716+
await listener.stop()
717+
await server.stop(None)
718+
719+
720+
async def test_the_default_answer_bound_is_16_mib():
721+
server = grpc.aio.server()
722+
server.add_generic_rpc_handlers((_handler(),))
723+
grpc_port = server.add_insecure_port("127.0.0.1:0")
724+
await server.start()
725+
listener = await connect_gen.serve(0, grpc_port, host="127.0.0.1")
726+
try:
727+
async with httpx.AsyncClient(timeout=60) as client:
728+
within = await client.post(
729+
_url(listener, "Say"),
730+
content=Msg(text="big", count=16 * 2**20 - 8).SerializeToString(),
731+
headers={"content-type": "application/proto"},
732+
)
733+
past = await client.post(
734+
_url(listener, "Say"),
735+
content=Msg(text="big", count=16 * 2**20).SerializeToString(),
736+
headers={"content-type": "application/proto"},
737+
)
738+
assert within.status_code == 200
739+
assert past.status_code == 400
740+
assert "max_response_bytes (16777216)" in past.json()["message"]
741+
finally:
742+
await listener.stop()
743+
await server.stop(None)

0 commit comments

Comments
 (0)