diff --git a/context/getting-started.md b/context/getting-started.md index 9104567..be907ab 100644 --- a/context/getting-started.md +++ b/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: diff --git a/guides/getting-started/readme.md b/guides/getting-started/readme.md index 9104567..be907ab 100644 --- a/guides/getting-started/readme.md +++ b/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: diff --git a/lib/async/grpc/client.rb b/lib/async/grpc/client.rb index eb831ed..36a37a2 100644 --- a/lib/async/grpc/client.rb +++ b/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? diff --git a/test/async/grpc/maximum_receive_message_size.rb b/test/async/grpc/maximum_receive_message_size.rb new file mode 100644 index 0000000..468f93d --- /dev/null +++ b/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