Skip to content

Commit 434f403

Browse files
fix(connect): bound answers apart from requests
The face's channel to the service took max_message_bytes as its receive length, so an answer was capped at the request bound: a service that bounds requests at a few hundred KB had a 393 KB embedding answer refused resource_exhausted on Connect while gRPC, relayed as it came, returned it. serve and start take max_response_bytes (64 MiB by default) for the answers the face transcodes; max_message_bytes still bounds requests before they are read and what the face sends the service. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 2c266fd commit 434f403

2 files changed

Lines changed: 128 additions & 2 deletions

File tree

‎templates/connect/connect_gen.py‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,12 @@
1818
end to end, so they are served on the gRPC protocol only.
1919
2020
Request bodies are bounded by ``max_message_bytes`` (gRPC's 4 MiB unless the service
21-
raises it) before they are read, and every read is bounded in time: an HTTP/1.1 read by
21+
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
2227
``read_timeout`` (``idle_timeout`` between requests), and a relayed gRPC connection is
2328
closed once it has had no open call for ``idle_timeout``, as Go's HTTP/2 server bounds an
2429
idle connection. A call in flight is never cut by the face, however long its handler
@@ -57,6 +62,7 @@
5762
# The port a deployment hands the face (CODEFLY_CONNECT_PORT); ``start`` reads only the variable.
5863
DEFAULT_PORT = 8081
5964
DEFAULT_MAX_MESSAGE_BYTES = 4 * 1024 * 1024
65+
DEFAULT_MAX_RESPONSE_BYTES = 64 * 1024 * 1024
6066
DEFAULT_READ_TIMEOUT_S = 30.0
6167
DEFAULT_IDLE_TIMEOUT_S = 120.0
6268

@@ -168,6 +174,7 @@ async def serve(
168174
host: str = "0.0.0.0",
169175
grpc_host: str = "127.0.0.1",
170176
max_message_bytes: int = DEFAULT_MAX_MESSAGE_BYTES,
177+
max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES,
171178
read_timeout: float = DEFAULT_READ_TIMEOUT_S,
172179
idle_timeout: float = DEFAULT_IDLE_TIMEOUT_S,
173180
) -> ConnectFace:
@@ -178,7 +185,8 @@ async def serve(
178185
channel = grpc.aio.insecure_channel(
179186
target,
180187
options=[
181-
("grpc.max_receive_message_length", max_message_bytes),
188+
# The face receives the service's answers and sends it the requests.
189+
("grpc.max_receive_message_length", max_response_bytes),
182190
("grpc.max_send_message_length", max_message_bytes),
183191
],
184192
)

‎testdata/connect/test_connect_gen.py‎

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,12 +63,18 @@ async def say(request, context):
6363
await context.abort(grpc.StatusCode.PERMISSION_DENIED, "denied: not yours")
6464
if request.text == "slow":
6565
await asyncio.sleep(2)
66+
if request.text == "big":
67+
# An answer of ``count`` bytes from a request of a few.
68+
return Msg(text="x" * request.count)
6669
await context.send_initial_metadata((("x-answer", "header"),))
6770
context.set_trailing_metadata((("x-trail", "trailer"), ("x-raw-bin", b"\x00\xff")))
6871
return Msg(text=f"{request.text}|{metadata.get('x-caller', '')}", count=request.count)
6972

7073

7174
async def count(request, context):
75+
if request.text == "big":
76+
yield Msg(text="x" * request.count)
77+
return
7278
for n in range(request.count):
7379
yield Msg(text=request.text, count=n)
7480
if request.text == "fail":
@@ -267,6 +273,118 @@ async def test_http_refusals(face):
267273
)
268274
assert (unknown.status_code, get.status_code, media.status_code) == (404, 405, 415)
269275
assert big.status_code == 413
276+
assert big.text == "content length (4096) exceeds the message bound"
277+
278+
279+
async def test_an_oversized_request_never_reaches_the_service(face):
280+
# A separate response bound leaves requests past 1024 bytes refused.
281+
async with httpx.AsyncClient() as client:
282+
r = await client.post(
283+
_url(face, "Say"),
284+
content=(request := Msg(text="x" * 2048).SerializeToString()),
285+
headers={"content-type": "application/proto", **STAMP},
286+
)
287+
assert r.status_code == 413
288+
assert r.text == f"content length ({len(request)}) exceeds the message bound"
289+
290+
291+
@pytest.mark.parametrize("size", [393 * 1024, 5 * 1024 * 1024])
292+
async def test_an_answer_larger_than_the_request_bound_is_answered(face, size):
293+
# The face bounds requests at 1024 bytes; embedding-sized answers and answers
294+
# above grpcio's default 4 MiB receive bound still come back on every protocol.
295+
big = Msg(text="big", count=size)
296+
async with httpx.AsyncClient() as client:
297+
unary = await client.post(
298+
_url(face, "Say"),
299+
content=big.SerializeToString(),
300+
headers={"content-type": "application/proto", **STAMP},
301+
)
302+
as_json = await client.post(
303+
_url(face, "Say"),
304+
json={"text": "big", "count": size},
305+
headers={"content-type": "application/json", **STAMP},
306+
)
307+
web = await client.post(
308+
_url(face, "Say"),
309+
content=envelope(big.SerializeToString()),
310+
headers={"content-type": "application/grpc-web+proto", **STAMP},
311+
)
312+
stream = await client.post(
313+
_url(face, "Count"),
314+
content=envelope(big.SerializeToString()),
315+
headers={"content-type": "application/connect+proto", **STAMP},
316+
)
317+
assert unary.status_code == 200, unary.text
318+
assert Msg.FromString(unary.content).text == "x" * size
319+
assert as_json.status_code == 200, as_json.text
320+
assert as_json.json() == {"text": "x" * size}
321+
out = frames(web.content)
322+
assert Msg.FromString(out[0][1]).text == "x" * size
323+
assert "grpc-status: 0" in out[-1][1].decode()
324+
out = frames(stream.content)
325+
assert Msg.FromString(out[0][1]).text == "x" * size
326+
assert out[-1] == (0x02, b"{}")
327+
328+
329+
@pytest.mark.parametrize(
330+
"media",
331+
[
332+
"application/json",
333+
"application/proto",
334+
"application/connect+proto",
335+
"application/connect+json",
336+
"application/grpc-web+proto",
337+
],
338+
)
339+
async def test_max_response_bytes_bounds_answers_when_set(media):
340+
server = grpc.aio.server()
341+
server.add_generic_rpc_handlers((_handler(),))
342+
grpc_port = server.add_insecure_port("127.0.0.1:0")
343+
await server.start()
344+
listener = None
345+
try:
346+
listener = await connect_gen.serve(
347+
0, grpc_port, host="127.0.0.1", max_message_bytes=1024, max_response_bytes=2048
348+
)
349+
streaming = media.startswith("application/connect+")
350+
enveloped = streaming or "grpc-web" in media
351+
async with httpx.AsyncClient() as client:
352+
for size in (1024, 4096):
353+
message = Msg(text="big", count=size)
354+
body = (
355+
json.dumps({"text": "big", "count": size}).encode()
356+
if "json" in media
357+
else message.SerializeToString()
358+
)
359+
r = await client.post(
360+
_url(listener, "Count" if streaming else "Say"),
361+
content=envelope(body) if enveloped else body,
362+
headers={"content-type": media},
363+
)
364+
if size == 1024:
365+
assert r.status_code == 200, r.text
366+
data = frames(r.content)[0][1] if enveloped else r.content
367+
text = (
368+
json.loads(data)["text"] if "json" in media else Msg.FromString(data).text
369+
)
370+
assert text == "x" * size
371+
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()
374+
else:
375+
if streaming:
376+
assert r.status_code == 200
377+
assert len(frames(r.content)) == 1
378+
error = json.loads(frames(r.content)[0][1])["error"]
379+
else:
380+
assert r.status_code == 429
381+
error = r.json()
382+
assert error["code"] == "resource_exhausted"
383+
assert "larger than max" in error["message"], error
384+
finally:
385+
if listener is not None:
386+
await listener.stop()
387+
await server.stop(None)
270388

271389

272390
async def test_one_connection_serves_several_calls(face):

0 commit comments

Comments
 (0)