Add a configurable maximum receive message size

context/getting-started.md

@@ -38,6 +38,21 @@ http_client = Async::HTTP::Client.new(endpoint)
 client = Async::GRPC::Client.new(http_client)
 ```
 
+### Limiting Response Message Size
+
+To bound response payloads from a peer, configure the client with `maximum_receive_message_size`:
+
+``` ruby
+client = Async::GRPC::Client.new(
+	http_client,
+	maximum_receive_message_size: 4 * 1024 * 1024
+)
+```
+
+The limit applies to each response in all four RPC patterns. It excludes the five-byte frame prefix, accepts exactly the configured size, and bounds both the declared wire payload and incrementally decompressed gzip/deflate output. Oversized messages raise {ruby Protocol::GRPC::Error} with `RESOURCE_EXHAUSTED` status. Requests are not limited by this option.
+
+The default is `nil` (unlimited). {ruby Async::GRPC::Client.open} forwards the option, and {ruby Async::GRPC::Client.with} inherits it unless explicitly overridden. This option requires a version of `protocol-grpc` supporting {ruby Protocol::GRPC::Body::Readable}'s `maximum_size` keyword; it does not bound the underlying HTTP transport's buffering.
+
 ### Using a Stub
 
 Create a stub from an interface definition:

guides/getting-started/readme.md

@@ -38,6 +38,21 @@ http_client = Async::HTTP::Client.new(endpoint)
 client = Async::GRPC::Client.new(http_client)
 ```
 
+### Limiting Response Message Size
+
+To bound response payloads from a peer, configure the client with `maximum_receive_message_size`:
+
+``` ruby
+client = Async::GRPC::Client.new(
+	http_client,
+	maximum_receive_message_size: 4 * 1024 * 1024
+)
+```
+
+The limit applies to each response in all four RPC patterns. It excludes the five-byte frame prefix, accepts exactly the configured size, and bounds both the declared wire payload and incrementally decompressed gzip/deflate output. Oversized messages raise {ruby Protocol::GRPC::Error} with `RESOURCE_EXHAUSTED` status. Requests are not limited by this option.
+
+The default is `nil` (unlimited). {ruby Async::GRPC::Client.open} forwards the option, and {ruby Async::GRPC::Client.with} inherits it unless explicitly overridden. This option requires a version of `protocol-grpc` supporting {ruby Protocol::GRPC::Body::Readable}'s `maximum_size` keyword; it does not bound the underlying HTTP transport's buffering.
+
 ### Using a Stub
 
 Create a stub from an interface definition:

lib/async/grpc/client.rb

@@ -34,6 +34,7 @@ def self.connect(endpoint)
 			# Create a new client for the given endpoint.
 			# @parameter endpoint [Async::HTTP::Endpoint, String] The endpoint to connect to
 			# @parameter headers [Protocol::HTTP::Headers] Default headers to include with requests
+			# @parameter options [Hash] Options forwarded to the initializer.
 			# @yields {|client| ...} Optional block - client will be closed after block execution
 			# @returns [Client] The client instance
 			def self.open(endpoint = self::ENDPOINT, headers: Protocol::HTTP::Headers.new, **options)
@@ -55,25 +56,36 @@ def self.open(endpoint = self::ENDPOINT, headers: Protocol::HTTP::Headers.new, *
 			# Create a new client with merged headers from a parent client.
 			# @parameter parent [Client] The parent client to inherit headers from
 			# @parameter headers [Hash] Additional headers to merge
+			# @parameter maximum_receive_message_size [Integer | Nil] Maximum response payload size in bytes, inherited from the parent by default.
 			# @returns [Client] A new client instance with merged headers
-			def self.with(parent, headers: {})
+			def self.with(parent, headers: {}, maximum_receive_message_size: parent.maximum_receive_message_size)
 				merged_headers = parent.headers.merge(headers)
 
-				new(parent.delegate, headers: merged_headers)
+				new(parent.delegate, headers: merged_headers, maximum_receive_message_size: maximum_receive_message_size)
 			end
 
 			# Initialize a new gRPC client.
 			# @parameter delegate [Async::HTTP::Client] The HTTP client that will handle requests
 			# @parameter headers [Protocol::HTTP::Headers] The default headers that will be supplied with requests
-			def initialize(delegate, headers: Protocol::HTTP::Headers.new)
+			# @parameter maximum_receive_message_size [Integer | Nil] Maximum wire and decompressed response payload size in bytes, excluding the frame prefix. `nil` disables the limit.
+			# @raises [ArgumentError] If the maximum receive message size is not a non-negative integer or `nil`.
+			def initialize(delegate, headers: Protocol::HTTP::Headers.new, maximum_receive_message_size: nil)
+				unless maximum_receive_message_size.nil? || (maximum_receive_message_size.is_a?(Integer) && maximum_receive_message_size >= 0)
+					raise ArgumentError, "Maximum receive message size must be a non-negative integer or nil!"
+				end
+
 				super(delegate)
 
 				@headers = headers
+				@maximum_receive_message_size = maximum_receive_message_size
 			end
 
 			# @attribute [Protocol::HTTP::Headers] The default headers for requests.
 			attr_reader :headers
 
+			# @attribute [Integer | Nil] The maximum wire and decompressed response payload size in bytes.
+			attr_reader :maximum_receive_message_size
+
 			# Get a string representation of the client.
 			# @returns [String] A string representation including headers
 			def inspect
@@ -153,6 +165,13 @@ def invoke(service, method, request = nil, metadata: {}, timeout: nil, encoding:
 
 		protected
 
+			# Options for decoding response messages.
+			# @returns [Hash] The configured receive limit, if enabled.
+			def receive_options
+				# Omit the new keyword when unlimited, for compatibility with earlier protocol-grpc versions:
+				@maximum_receive_message_size ? {maximum_size: @maximum_receive_message_size} : {}
+			end
+
 			# Reject non-gRPC responses before passing their bytes to a frame decoder.
 			# @parameter response [Protocol::HTTP::Response] The HTTP response.
 			# @raises [ResponseError] If the response is not a valid gRPC envelope.
@@ -183,7 +202,7 @@ def unary_call(path, headers, request_message, request_class, response_class, en
 				begin
 					# Read body first - trailers are only available after body is consumed
 					response_encoding = response.headers["grpc-encoding"]
-					response_body = Protocol::GRPC::Body::ReadableBody.wrap(response, message_class: response_class, encoding: response_encoding)
+					response_body = Protocol::GRPC::Body::ReadableBody.wrap(response, message_class: response_class, encoding: response_encoding, **receive_options)
 
 					if response_body
 						response_value = response_body.read
@@ -220,7 +239,7 @@ def server_streaming_call(path, headers, request_message, request_class, respons
 				begin
 					# Read body first - trailers are only available after body is consumed:
 					response_encoding = response.headers["grpc-encoding"]
-					response_body = Protocol::GRPC::Body::ReadableBody.wrap(response, message_class: response_class, encoding: response_encoding)
+					response_body = Protocol::GRPC::Body::ReadableBody.wrap(response, message_class: response_class, encoding: response_encoding, **receive_options)
 
 					if block_given? and response_body
 						response_body.each(&block)
@@ -266,7 +285,8 @@ def client_streaming_call(path, headers, request_class, response_class, encoding
 					readable_body = Protocol::GRPC::Body::ReadableBody.new(
 						response.body,
 						message_class: response_class,
-						encoding: response_encoding
+						encoding: response_encoding,
+						**receive_options
 					)
 
 					message = readable_body.read
@@ -309,7 +329,8 @@ def bidirectional_call(path, headers, request_class, response_class, encoding, i
 					readable_body = Protocol::GRPC::Body::ReadableBody.new(
 						response.body,
 						message_class: response_class,
-						encoding: response_encoding
+						encoding: response_encoding,
+						**receive_options
 					)
 
 					unless block_given?

test/async/grpc/maximum_receive_message_size.rb

@@ -0,0 +1,174 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Ateş Göral.
+
+require "async/grpc"
+require "async/grpc/test_interface"
+require "protocol/http/body/writable"
+require "sus/fixtures/async/http"
+require "zlib"
+
+AReceiveLimit = Sus::Shared("a receive limit") do
+	let(:response_message) {Protocol::GRPC::Fixtures::TestMessage.new(value: "a" * 1024)}
+	let(:response_data) {response_message.to_proto}
+	let(:response_encoding) {nil}
+	let(:response_status) {Protocol::GRPC::Status::OK}
+	let(:status_in_trailer) {false}
+	let(:interface) {Async::GRPC::Fixtures::TestInterface.new("test.Service")}
+	let(:request) {Protocol::GRPC::Fixtures::TestMessage.new(value: "request" * 1024)}
+	let(:app) do
+		Protocol::HTTP::Middleware.for do |request|
+			# Read one request message without waiting for a bidirectional stream to close:
+			Protocol::GRPC::Body::Readable.new(request.body).read
+			request.body.finish unless request.path.end_with?("/BidirectionalCall")
+			headers = Protocol::HTTP::Headers.new([], nil, policy: Protocol::GRPC::HEADER_POLICY)
+			headers["content-type"] = "application/grpc"
+			headers["grpc-encoding"] = response_encoding if response_encoding
+			if status_in_trailer
+				headers["trailer"] = "grpc-status"
+				headers.trailer!
+			end
+			Protocol::GRPC::Metadata.assign_status!(headers, status: response_status)
+			prefix = [response_encoding ? 1 : 0, response_data.bytesize].pack("CN")
+			body = Protocol::HTTP::Body::Writable.new
+			body.write(prefix + response_data)
+			body.close_write
+			Protocol::HTTP::Response[200, headers, body]
+		end
+	end
+
+	def invoke(grpc_client, method)
+		case interface.class.lookup_rpc(method).streaming
+		when :unary
+			grpc_client.invoke(interface, method, request)
+		when :server_streaming
+			value = nil
+			grpc_client.invoke(interface, method, request){|message| value = message}
+			value
+		when :client_streaming
+			grpc_client.invoke(interface, method){|output| output.write(request)}
+		when :bidirectional
+			value = nil
+			grpc_client.invoke(interface, method, initial: [request]) do |output, input|
+				output.close_write
+				value = input.read
+			end
+			value
+		end
+	end
+
+	[:UnaryCall, :ServerStreamingCall, :ClientStreamingCall, :BidirectionalCall].each do |method|
+		with method.to_s do
+			it "accepts exactly the maximum response size without limiting requests" do
+				grpc_client = Async::GRPC::Client.new(client, maximum_receive_message_size: response_data.bytesize)
+
+				expect(invoke(grpc_client, method)).to be == response_message
+			end
+
+			it "rejects an oversized response" do
+				grpc_client = Async::GRPC::Client.new(client, maximum_receive_message_size: response_data.bytesize - 1)
+
+				expect{invoke(grpc_client, method)}.to raise_exception(Protocol::GRPC::Error) do |error|
+					expect(error.status_code).to be == Protocol::GRPC::Status::RESOURCE_EXHAUSTED
+				end
+			end
+		end
+	end
+
+	[:ServerStreamingCall, :BidirectionalCall].each do |method|
+		it "enforces the limit on the returned #{method} stream without a block" do
+			grpc_client = Async::GRPC::Client.new(client, maximum_receive_message_size: response_data.bytesize - 1)
+			input = grpc_client.invoke(interface, method, request, initial: [request])
+
+			expect{input.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::RESOURCE_EXHAUSTED
+			end
+		ensure
+			input&.close
+		end
+	end
+
+	it "leaves responses unlimited by default" do
+		grpc_client = Async::GRPC::Client.new(client)
+
+		expect(grpc_client.maximum_receive_message_size).to be_nil
+		expect(invoke(grpc_client, :UnaryCall)).to be == response_message
+	end
+
+	with "a non-OK final status" do
+		let(:response_status) {Protocol::GRPC::Status::NOT_FOUND}
+		let(:status_in_trailer) {true}
+
+		it "preserves status validation for an in-limit response" do
+			grpc_client = Async::GRPC::Client.new(client, maximum_receive_message_size: response_data.bytesize)
+
+			expect{invoke(grpc_client, :UnaryCall)}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::NOT_FOUND
+			end
+		end
+	end
+
+	{"gzip" => Zlib.method(:gzip), "deflate" => Zlib::Deflate.method(:deflate)}.each do |encoding, compress|
+		with encoding do
+			let(:response_encoding) {encoding}
+			let(:response_message) {Protocol::GRPC::Fixtures::TestMessage.new(value: "a" * 800_000)}
+			let(:response_data) {compress.call(response_message.to_proto)}
+
+			it "rejects decompression expansion below the wire-size limit" do
+				expect(response_data.bytesize).to be < 1024
+				grpc_client = Async::GRPC::Client.new(client, maximum_receive_message_size: 1024)
+
+				expect{invoke(grpc_client, :UnaryCall)}.to raise_exception(Protocol::GRPC::Error) do |error|
+					expect(error.status_code).to be == Protocol::GRPC::Status::RESOURCE_EXHAUSTED
+				end
+			end
+		end
+	end
+end
+
+describe Async::GRPC::Client do
+	with "maximum receive message size" do
+		[-1, 1.5, "1024", false].each do |maximum_size|
+			it "rejects an invalid maximum size #{maximum_size.inspect}" do
+				expect do
+					subject.new(Protocol::HTTP::Middleware::Okay, maximum_receive_message_size: maximum_size)
+				end.to raise_exception(ArgumentError)
+			end
+		end
+
+		it "passes the limit through .open" do
+			client = subject.open("http://localhost:0", maximum_receive_message_size: 1024)
+
+			expect(client.maximum_receive_message_size).to be == 1024
+		ensure
+			client&.close
+		end
+
+		it "inherits the limit through .with" do
+			parent = subject.new(Protocol::HTTP::Middleware::Okay, maximum_receive_message_size: 1024)
+			client = subject.with(parent, headers: {"request-id" => "1234"})
+
+			expect(client.maximum_receive_message_size).to be == 1024
+		end
+
+		it "can override or disable an inherited limit" do
+			parent = subject.new(Protocol::HTTP::Middleware::Okay, maximum_receive_message_size: 1024)
+
+			expect(subject.with(parent, maximum_receive_message_size: 2048).maximum_receive_message_size).to be == 2048
+			expect(subject.with(parent, maximum_receive_message_size: nil).maximum_receive_message_size).to be_nil
+		end
+	end
+
+	include Sus::Fixtures::Async::HTTP::ServerContext
+
+	with "http/1" do
+		let(:protocol) {Async::HTTP::Protocol::HTTP1}
+		it_behaves_like AReceiveLimit
+	end
+
+	with "http/2" do
+		let(:protocol) {Async::HTTP::Protocol::HTTP2}
+		it_behaves_like AReceiveLimit
+	end
+end