Add optional receive-message size bounds

context/getting-started.md

@@ -108,6 +108,28 @@ end
 Protocol::GRPC::Status::NAMES[status] # => "OK"
 ```
 
+### Limiting Message Size
+
+To bound the size of messages received from a peer, pass `maximum_size` to the readable body:
+
+``` ruby
+readable_body = Protocol::GRPC::Body::Readable.new(
+	response.body,
+	message_class: Hello::HelloReply,
+	maximum_size: 4 * 1024 * 1024
+)
+
+begin
+	message = readable_body.read
+ensure
+	readable_body.close
+end
+```
+
+The limit applies to each message's payload, excluding the five-byte frame prefix. Exactly the configured number of bytes is accepted. The declared frame length is checked before reading its payload, and gzip/deflate output is checked incrementally before it is accumulated. Exceeding either limit raises {ruby Protocol::GRPC::Error} with `RESOURCE_EXHAUSTED` status.
+
+The underlying HTTP body may already have buffered a chunk containing multiple frames; this option bounds gRPC message processing, not transport buffering. The default is `nil` (unlimited). A limit of zero allows only empty, uncompressed payloads.
+
 ### Server Middleware
 
 Create a server middleware by subclassing `Protocol::GRPC::Middleware`:

guides/getting-started/readme.md

@@ -108,6 +108,28 @@ end
 Protocol::GRPC::Status::NAMES[status] # => "OK"
 ```
 
+### Limiting Message Size
+
+To bound the size of messages received from a peer, pass `maximum_size` to the readable body:
+
+``` ruby
+readable_body = Protocol::GRPC::Body::Readable.new(
+	response.body,
+	message_class: Hello::HelloReply,
+	maximum_size: 4 * 1024 * 1024
+)
+
+begin
+	message = readable_body.read
+ensure
+	readable_body.close
+end
+```
+
+The limit applies to each message's payload, excluding the five-byte frame prefix. Exactly the configured number of bytes is accepted. The declared frame length is checked before reading its payload, and gzip/deflate output is checked incrementally before it is accumulated. Exceeding either limit raises {ruby Protocol::GRPC::Error} with `RESOURCE_EXHAUSTED` status.
+
+The underlying HTTP body may already have buffered a chunk containing multiple frames; this option bounds gRPC message processing, not transport buffering. The default is `nil` (unlimited). A limit of zero allows only empty, uncompressed payloads.
+
 ### Server Middleware
 
 Create a server middleware by subclassing `Protocol::GRPC::Middleware`:

lib/protocol/grpc/body/readable.rb

@@ -36,8 +36,15 @@ def self.wrap(message, **options)
 				# @parameter message_class [Class | Nil] Protobuf message class with .decode method.
 				#   If `nil`, returns raw binary data (useful for channel adapters)
 				# @parameter encoding [String | Nil] Compression encoding (from grpc-encoding header)
-				def initialize(body, message_class: nil, encoding: nil)
+				# @parameter maximum_size [Integer | Nil] Maximum wire and decompressed payload size in bytes, excluding the frame prefix. `nil` disables the limit.
+				# @raises [ArgumentError] If the maximum size is not a non-negative integer or `nil`.
+				def initialize(body, message_class: nil, encoding: nil, maximum_size: nil)
+					unless maximum_size.nil? || (maximum_size.is_a?(Integer) && maximum_size >= 0)
+						raise ArgumentError, "Maximum size must be a non-negative integer or nil!"
+					end
+
 					super(body)
+					@maximum_size = maximum_size
 					@message_class = message_class
 					@encoding = encoding
 					@buffer = String.new.force_encoding(Encoding::BINARY)
@@ -46,17 +53,22 @@ def initialize(body, message_class: nil, encoding: nil)
 				# @attribute [String | Nil] The compression encoding.
 				attr_reader :encoding
 
+				# @attribute [Integer | Nil] The maximum wire and decompressed payload size in bytes.
+				attr_reader :maximum_size
+
 				# Read the next gRPC message.
 				# Overrides Wrapper#read to transform raw HTTP body chunks into decoded gRPC messages.
 				# Errors raised by the underlying body propagate unchanged.
 				# @returns [Object | String | Nil] Decoded message, raw binary, or `Nil` if stream ended
+				# @raises [Error] With `RESOURCE_EXHAUSTED` if a message exceeds the maximum size.
 				def read
 					# Read 5-byte prefix: 1 byte compression flag + 4 bytes length
 					prefix = read_exactly(5)
 					return nil unless prefix
 
 					compressed = prefix[0].unpack1("C") == 1
 					length = prefix[1..4].unpack1("N")
+					check_size!(length)
 
 					# Read the message body:
 					data = read_exactly(length)
@@ -106,26 +118,62 @@ def read_exactly(n)
 				end
 
 				# Decompress data using the configured encoding.
-				# @parameter data [String] The compressed data
-				# @returns [String] The decompressed data
-				# @raises [Error] If decompression fails
+				# @parameter data [String] The compressed data.
+				# @returns [String] The decompressed data.
+				# @raises [Error] If decompression fails or the maximum size is exceeded.
 				def decompress(data)
 					case @encoding
 					when "gzip"
-						begin
+						if @maximum_size
+							inflate(data, Zlib::MAX_WBITS + 16)
+						else
 							Zlib.gunzip(data)
-						rescue => error
-							raise Error.new(Status::INTERNAL, "Failed to decompress message: #{error.message}")
 						end
 					when "deflate"
-						begin
+						if @maximum_size
+							inflate(data, Zlib::MAX_WBITS)
+						else
 							Zlib::Inflate.inflate(data)
-						rescue => error
-							raise Error.new(Status::INTERNAL, "Failed to decompress message: #{error.message}")
 						end
 					else
 						raise Error.new(Status::UNIMPLEMENTED, "Unsupported compression encoding: #{@encoding.inspect}")
 					end
+				rescue Zlib::Error => error
+					raise Error.new(Status::INTERNAL, "Failed to decompress message: #{error.message}")
+				end
+
+				# Inflate a message without accumulating output beyond the maximum size.
+				# @parameter data [String] The compressed data.
+				# @parameter window_bits [Integer] The compression format.
+				# @returns [String] The decompressed data.
+				# @raises [Error] If the message is truncated or the maximum size is exceeded.
+				# @raises [Zlib::Error] If decompression fails.
+				def inflate(data, window_bits)
+					inflater = Zlib::Inflate.new(window_bits)
+					result = String.new.force_encoding(Encoding::BINARY)
+
+					# Check each output chunk before retaining it:
+					inflater.inflate(data) do |chunk|
+						check_size!(result.bytesize + chunk.bytesize)
+						result << chunk
+					end
+
+					unless inflater.finished?
+						raise Error.new(Status::INTERNAL, "Truncated compressed gRPC message!")
+					end
+
+					result
+				ensure
+					inflater&.close
+				end
+
+				# Check a message size against the configured maximum.
+				# @parameter size [Integer] The payload size in bytes.
+				# @raises [Error] With `RESOURCE_EXHAUSTED` if the maximum size is exceeded.
+				def check_size!(size)
+					if @maximum_size && size > @maximum_size
+						raise Error.new(Status::RESOURCE_EXHAUSTED, "Received message larger than maximum size: #{size} > #{@maximum_size}")
+					end
 				end
 			end
 		end

test/protocol/grpc/body/maximum_size.rb

@@ -0,0 +1,214 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Ateş Göral.
+
+require "protocol/grpc/body/readable"
+require "protocol/http/body/buffered"
+require "protocol/http/body/writable"
+require_relative "../../../../fixtures/protocol/grpc/test_message"
+
+require "zlib"
+
+describe Protocol::GRPC::Body::Readable do
+	with "maximum size" do
+		let(:source_body) {Protocol::HTTP::Body::Buffered.new}
+		let(:body) {subject.new(source_body, maximum_size: 1024)}
+
+		def frame(data, compressed: false)
+			[compressed ? 1 : 0, data.bytesize].pack("CN") + data
+		end
+
+		def expect_size_error
+			expect{yield}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::RESOURCE_EXHAUSTED
+				expect(error.message).to be =~ /Received message larger than maximum size: \d+ > 1024/
+			end
+		end
+
+		it "defaults to unlimited messages" do
+			data = "a" * 2048
+			source_body.write(frame(data))
+			reader = subject.new(source_body)
+
+			expect(reader.maximum_size).to be_nil
+			expect(reader.read).to be == data
+		end
+
+		[-1, 1.5, "1024", false].each do |maximum_size|
+			it "rejects an invalid maximum size #{maximum_size.inspect}" do
+				expect{subject.new(source_body, maximum_size: maximum_size)}.to raise_exception(ArgumentError)
+			end
+		end
+
+		it "accepts exactly the maximum payload size, excluding the prefix" do
+			data = "a" * 1024
+			source_body.write(frame(data))
+
+			expect(body.maximum_size).to be == 1024
+			expect(body.read).to be == data
+		end
+
+		it "accepts empty messages with a zero limit" do
+			source_body.write(frame(""))
+			reader = subject.new(source_body, maximum_size: 0)
+
+			expect(reader.read).to be == ""
+			expect(reader.read).to be_nil
+		end
+
+		it "rejects non-empty messages with a zero limit" do
+			source_body.write(frame("a"))
+			reader = subject.new(source_body, maximum_size: 0)
+
+			expect{reader.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::RESOURCE_EXHAUSTED
+			end
+		end
+
+		it "rejects an oversized declared payload without reading it" do
+			source_body.write([0, 1025].pack("CN"))
+			reads = 0
+			mock(source_body) do |wrapper|
+				wrapper.wrap(:read) do |original|
+					reads += 1
+					raise "Payload must not be read!" if reads > 1
+					original.call
+				end
+			end
+
+			expect_size_error {body.read}
+			expect(reads).to be == 1
+		end
+
+		it "reads fragmented prefixes and payloads" do
+			data = "a" * 1024
+			frame(data).bytes.each_slice(3) do |bytes|
+				source_body.write(bytes.pack("C*"))
+			end
+
+			expect(body.read).to be == data
+			expect(body.read).to be_nil
+		end
+
+		it "limits each coalesced message independently" do
+			data = "a" * 1024
+			source_body.write(frame(data) + frame(data))
+
+			expect(body.read).to be == data
+			expect(source_body).to be(:empty?)
+			expect(body.read).to be == data
+			expect(body.read).to be_nil
+		end
+
+		it "rejects an oversized message after a complete coalesced message" do
+			source_body.write(frame("a") + frame("b" * 1025))
+
+			expect(body.read).to be == "a"
+			expect_size_error {body.read}
+		end
+
+		it "preserves protobuf decoding within the limit" do
+			message_class = Protocol::GRPC::Fixtures::TestMessage
+			message = message_class.new(value: "Hello")
+			source_body.write(frame(message.to_proto))
+			reader = subject.new(source_body, message_class: message_class, maximum_size: message.to_proto.bytesize)
+
+			expect(reader.read).to be == message
+		end
+
+		["\x00\x00".b, [0, 2].pack("CN"), [0, 2].pack("CN") + "a"].each do |partial_frame|
+			it "rejects a truncated #{partial_frame.bytesize}-byte frame" do
+				source_body.write(partial_frame)
+
+				expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+					expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+				end
+			end
+		end
+
+		it "rejects an incomplete next frame after a complete message" do
+			source_body.write(frame("a") + "\x00\x00".b)
+
+			expect(body.read).to be == "a"
+			expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+			end
+		end
+
+		it "preserves a pending underlying read error" do
+			source = Protocol::HTTP::Body::Writable.new
+			error = IOError.new("Stream failed!")
+			source.close_write(error)
+			reader = subject.new(source, maximum_size: 1024)
+
+			expect{reader.read}.to raise_exception(IOError).and(be_equal(error))
+		end
+
+		{"gzip" => Zlib.method(:gzip), "deflate" => Zlib::Deflate.method(:deflate)}.each do |encoding, compress|
+			with encoding do
+				let(:body) {subject.new(source_body, encoding: encoding, maximum_size: 1024)}
+
+				it "accepts exactly the maximum decompressed size" do
+					data = "a" * 1024
+					source_body.write(frame(compress.call(data), compressed: true))
+
+					expect(body.read).to be == data
+				end
+
+				it "rejects expansion beyond the limit before accumulating the full output" do
+					data = "a" * 1_000_000
+					compressed = compress.call(data)
+					expect(compressed.bytesize).to be < 1024
+					source_body.write(frame(compressed, compressed: true))
+					sizes = []
+					mock(body) do |wrapper|
+						wrapper.wrap(:check_size!) do |original, size|
+							sizes << size
+							original.call(size)
+						end
+					end
+
+					expect_size_error {body.read}
+					expect(sizes.last).to be > 1024
+					expect(sizes.last).to be < data.bytesize
+				end
+
+				it "resets the expanded-size limit for each message" do
+					data = "a" * 1024
+					message = frame(compress.call(data), compressed: true)
+					source_body.write(message + message)
+
+					expect(body.read).to be == data
+					expect(body.read).to be == data
+				end
+
+				it "rejects a truncated compressed stream" do
+					compressed = compress.call("a" * 512).byteslice(0...-1)
+					source_body.write(frame(compressed, compressed: true))
+
+					expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+						expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+					end
+				end
+
+				it "rejects invalid compressed data" do
+					source_body.write(frame("invalid", compressed: true))
+
+					expect{body.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+						expect(error.status_code).to be == Protocol::GRPC::Status::INTERNAL
+					end
+				end
+			end
+		end
+
+		it "preserves unsupported compression errors" do
+			source_body.write(frame("data", compressed: true))
+			reader = subject.new(source_body, encoding: "custom", maximum_size: 1024)
+
+			expect{reader.read}.to raise_exception(Protocol::GRPC::Error) do |error|
+				expect(error.status_code).to be == Protocol::GRPC::Status::UNIMPLEMENTED
+			end
+		end
+	end
+end