Skip to content

Commit 940bc54

Browse files
Merge pull request #111 from codefly-dev/fix/connect-response-limit
fix(connect): bound answers apart from requests
2 parents 807e1f8 + 1cadf23 commit 940bc54

2 files changed

Lines changed: 399 additions & 14 deletions

File tree

‎templates/connect/connect_gen.py‎

Lines changed: 128 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,35 @@
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
22-
``read_timeout`` (``idle_timeout`` between requests), and a relayed gRPC connection is
23-
closed once it has had no open call for ``idle_timeout``, as Go's HTTP/2 server bounds an
24-
idle connection. A call in flight is never cut by the face, however long its handler
25-
takes to answer: its deadline and the gRPC server's own policy bound it.
21+
raises it) before they are read. Answers are bounded apart, on the proto the service
22+
returns, by the codec the caller asked for:
23+
24+
- ``max_response_bytes`` for an answer relayed as proto (Connect and gRPC-Web proto): by
25+
default the larger of ``DEFAULT_MAX_RESPONSE_BYTES`` (16 MiB) and ``max_message_bytes``.
26+
The face holds such an answer about twice over while it writes it, beyond the
27+
service's own copy: measured, 33 MiB for a 16 MiB answer.
28+
- ``max_json_response_bytes`` for an answer converted to JSON: by default the larger of
29+
``DEFAULT_MAX_JSON_RESPONSE_BYTES`` (4 MiB, the bound every answer had before the two
30+
were apart) and ``max_message_bytes``. It is checked before the conversion, because the
31+
conversion is what costs: measured as process memory (protobuf 6.33, one 4 MiB answer),
32+
a packed list of floats peaks at 18 times the proto, nested messages at 62 times, and a
33+
list of enums at 130 times (its JSON alone is 25 times the proto). So up to about
34+
520 MiB at the default, for one conversion; conversions run on the face's event loop
35+
one at a time, so that peak does not add up across calls.
36+
37+
Past its bound a call ends ``failed_precondition`` naming the bound: the answer is too
38+
large for this face's configuration, and asking again does not change it. Within it,
39+
Connect and gRPC-Web return what the gRPC server returns, and a refusal the service sends
40+
keeps its code and words. Both bounds are positive numbers of bytes, keyword options to
41+
``serve`` and ``start``; a relayed gRPC call keeps the service's and the caller's own
42+
bounds. The face does not bound how many calls are in flight, as the gRPC server does
43+
not: each holds its answer until its client reads it.
44+
45+
Every read is bounded in time: an HTTP/1.1 read by ``read_timeout`` (``idle_timeout``
46+
between requests), and a relayed gRPC connection is closed once it has had no open call
47+
for ``idle_timeout``, as Go's HTTP/2 server bounds an idle connection. A call in flight
48+
is never cut by the face, however long its handler takes to answer: its deadline and the
49+
gRPC server's own policy bound it.
2650
A JSON request field the service's proto does not know is ignored, as connect-go's codec
2751
ignores it, so a client built from a newer proto calls a Python service and a Go one alike.
2852
@@ -57,6 +81,8 @@
5781
# The port a deployment hands the face (CODEFLY_CONNECT_PORT); ``start`` reads only the variable.
5882
DEFAULT_PORT = 8081
5983
DEFAULT_MAX_MESSAGE_BYTES = 4 * 1024 * 1024
84+
DEFAULT_MAX_RESPONSE_BYTES = 16 * 1024 * 1024
85+
DEFAULT_MAX_JSON_RESPONSE_BYTES = 4 * 1024 * 1024
6086
DEFAULT_READ_TIMEOUT_S = 30.0
6187
DEFAULT_IDLE_TIMEOUT_S = 120.0
6288

@@ -111,6 +137,10 @@
111137
)
112138
_TRANSPORT_PREFIXES = ("connect-", "grpc-", "x-grpc-web", ":")
113139
_GRPC_TIMEOUT = re.compile(r"^(\d{1,8})([HMSmun])$")
140+
# grpcio's refusal of an answer past the face's own receive bound (its channel is the client).
141+
_ANSWER_PAST_BOUND = re.compile(
142+
r"Stream removed \(CLIENT: Received message larger than max \((\d+) vs\. (\d+)\)\)"
143+
)
114144
_GRPC_UNITS = {"H": 3600.0, "M": 60.0, "S": 1.0, "m": 1e-3, "u": 1e-6, "n": 1e-9}
115145
_CODES = {
116146
grpc.StatusCode.CANCELLED: ("canceled", 499),
@@ -183,22 +213,43 @@ async def serve(
183213
host: str = "0.0.0.0",
184214
grpc_host: str = "127.0.0.1",
185215
max_message_bytes: int = DEFAULT_MAX_MESSAGE_BYTES,
216+
max_response_bytes: int | None = None,
217+
max_json_response_bytes: int | None = None,
186218
read_timeout: float = DEFAULT_READ_TIMEOUT_S,
187219
idle_timeout: float = DEFAULT_IDLE_TIMEOUT_S,
188220
) -> ConnectFace:
189221
"""Start the Connect face on ``host``:``port`` (0 for an ephemeral one) over the
190222
grpc.aio server listening on ``grpc_host``:``grpc_port``, and return it started.
191223
Call it on the loop that serves the gRPC server, once that server listens."""
224+
bounds = _AnswerBounds(
225+
_positive(
226+
"max_response_bytes", max_response_bytes, DEFAULT_MAX_RESPONSE_BYTES, max_message_bytes
227+
),
228+
_positive(
229+
"max_json_response_bytes",
230+
max_json_response_bytes,
231+
DEFAULT_MAX_JSON_RESPONSE_BYTES,
232+
max_message_bytes,
233+
),
234+
)
192235
target = f"{grpc_host}:{grpc_port}"
193236
channel = grpc.aio.insecure_channel(
194237
target,
195238
options=[
196-
("grpc.max_receive_message_length", max_message_bytes),
239+
# The face receives the service's answers and sends it the requests. The
240+
# channel stops an answer past both bounds before it is buffered; the face
241+
# checks the bound of the call's codec itself.
242+
("grpc.max_receive_message_length", bounds.received),
197243
("grpc.max_send_message_length", max_message_bytes),
198244
],
199245
)
200246
connection = _Connection(
201-
channel, (grpc_host, grpc_port), max_message_bytes, read_timeout, idle_timeout
247+
channel,
248+
(grpc_host, grpc_port),
249+
max_message_bytes,
250+
bounds,
251+
read_timeout,
252+
idle_timeout,
202253
)
203254
face: ConnectFace
204255

@@ -221,6 +272,29 @@ async def accepted(reader, writer):
221272
return face
222273

223274

275+
def _positive(name: str, value: int | None, default: int, message_bound: int) -> int:
276+
"""A bound as given, or by default the larger of ``default`` and the message bound."""
277+
if value is None:
278+
return max(default, message_bound)
279+
if value <= 0:
280+
raise ValueError(f"{name} must be a positive number of bytes, not {value}")
281+
return value
282+
283+
284+
class _AnswerBounds:
285+
"""The answers the face takes from the service: ``proto`` for a codec that relays
286+
them as they are, ``json`` for one that converts them, checked before converting."""
287+
288+
def __init__(self, proto: int, json: int):
289+
self.proto, self.json = proto, json
290+
self.received = max(proto, json)
291+
292+
def of(self, codec: _Codec) -> tuple[int, str]:
293+
if codec.name == "proto":
294+
return self.proto, "max_response_bytes"
295+
return self.json, "max_json_response_bytes"
296+
297+
224298
async def start(grpc_port: int, **options) -> ConnectFace | None:
225299
"""Start the face the agent's generated ``src/main.py`` starts: on
226300
``CODEFLY_CONNECT_PORT`` over the gRPC server on ``grpc_port``, with ``serve``'s
@@ -343,8 +417,9 @@ def _emit(self):
343417

344418

345419
class _Connection:
346-
def __init__(self, channel, grpc_address, limit, read_timeout, idle_timeout):
420+
def __init__(self, channel, grpc_address, limit, answer_bounds, read_timeout, idle_timeout):
347421
self.channel, self.grpc_address, self.limit = channel, grpc_address, limit
422+
self.answer_bounds = answer_bounds
348423
self.read_timeout, self.idle_timeout = read_timeout, idle_timeout
349424

350425
async def serve(self, reader, writer):
@@ -598,7 +673,16 @@ async def _dispatch(self, exchange, request, body):
598673
codec = protocol.codec
599674
if codec is None:
600675
raise _Plain(415, f"unsupported codec in {content_type!r}")
601-
call = _Call(self.channel, path, method, codec, request.headers, headers, self.limit)
676+
call = _Call(
677+
self.channel,
678+
path,
679+
method,
680+
codec,
681+
request.headers,
682+
headers,
683+
self.limit,
684+
self.answer_bounds,
685+
)
602686
await protocol.answer(exchange, call, body, headers)
603687

604688

@@ -643,9 +727,11 @@ class _Call:
643727
"""One call dispatched through the gRPC server: request bytes in the protocol's
644728
codec, answers out in it."""
645729

646-
def __init__(self, channel, path, method, codec, raw_headers, headers, limit):
730+
def __init__(self, channel, path, method, codec, raw_headers, headers, limit, answer_bounds):
647731
self.channel, self.path, self.method, self.codec = channel, path, method, codec
648732
self.limit = limit
733+
self.answer_bound, self.answer_bound_name = answer_bounds.of(codec)
734+
self.received_bound = answer_bounds.received
649735
self.input = message_factory.GetMessageClass(method.input_type)
650736
self.output = message_factory.GetMessageClass(method.output_type)
651737
self.metadata = _metadata(raw_headers)
@@ -678,6 +764,32 @@ def decompress(self, data: bytes, encoding: str | None) -> bytes:
678764
)
679765
return payload
680766

767+
def answer(self, data: bytes) -> bytes:
768+
"""An answer in the call's codec, within the bound of that codec, checked before it
769+
is converted: the conversion is what costs (see the module's docstring)."""
770+
if len(data) > self.answer_bound:
771+
raise self._too_large(len(data))
772+
return self.codec.response(self.output, data)
773+
774+
def failed(self, error: grpc.aio.AioRpcError) -> CallError:
775+
"""The call's status. An answer past the channel's own receive bound (one past
776+
every bound of the face) is refused by grpcio before it is buffered, with only its
777+
text to say so: it is the face's refusal when that text is grpcio's own and names
778+
exactly the bound the face set. A service passing on a refusal from its own client
779+
names that client's bound, and keeps its code and words."""
780+
past = _ANSWER_PAST_BOUND.fullmatch(error.details() or "")
781+
own = past is not None and int(past.group(2)) == self.received_bound
782+
if error.code() is grpc.StatusCode.RESOURCE_EXHAUSTED and own:
783+
return self._too_large(int(past.group(1)))
784+
return CallError.of(error)
785+
786+
def _too_large(self, size: int) -> CallError:
787+
return CallError(
788+
grpc.StatusCode.FAILED_PRECONDITION,
789+
f"the answer ({size} bytes) exceeds the Connect face's {self.answer_bound_name} "
790+
f"({self.answer_bound})",
791+
)
792+
681793
async def unary(self, message: bytes):
682794
"""The answer and the header and trailer metadata of a unary call."""
683795
request = self.codec.request(self.input, message)
@@ -687,12 +799,12 @@ async def unary(self, message: bytes):
687799
try:
688800
answer = await call
689801
return (
690-
self.codec.response(self.output, answer),
802+
self.answer(answer),
691803
tuple(await call.initial_metadata() or ()),
692804
tuple(await call.trailing_metadata() or ()),
693805
)
694806
except grpc.aio.AioRpcError as exc:
695-
raise CallError.of(exc) from None
807+
raise self.failed(exc) from None
696808

697809
def stream(self, message: bytes):
698810
request = self.codec.request(self.input, message)
@@ -852,11 +964,13 @@ async def _relay(exchange, call, stream):
852964
return stream.trailer
853965
try:
854966
async for answer in stream:
855-
data = call.codec.response(call.output, answer)
967+
data = call.answer(answer)
856968
await exchange.send(h11.Data(data=_ENVELOPE.pack(0, len(data)) + data))
857969
return tuple(await stream.trailing_metadata() or ())
858970
except grpc.aio.AioRpcError as exc:
859-
return CallError.of(exc)
971+
return call.failed(exc)
972+
except CallError as error:
973+
return error
860974
finally:
861975
stream.cancel()
862976

0 commit comments

Comments
 (0)