Building a Leak-Safe gRPC Frame Decoder on Reactor Netty

This is the second article in my grpc-reactor series. The first article explains why I chose to build the runtime directly on Reactor Netty and where its compatibility boundary sits. This article moves one layer down into the Stage 1 protocol implementation: the frame decoder that every RPC shape relies on. gRPC protobuf messages are not written directly as raw bytes into HTTP/2 DATA frames. Every message starts with a five-byte envelope: byte 0 bit 0 indicates compression; bits 1-7 must be zero bytes 1-4 unsigned big-endian payload length byte 5..n protobuf message, or its compressed representation Encoding this envelope is straightforward. The difficult part is decoding it without assuming that one input buffer contains one complete frame. HTTP/2, TCP, and Reactor Netty do not promise that buffer boundaries will line up with gRPC message boundaries. This post describes the Stage 1 protocol layer. The project has since progressed beyond it, but the ownership and bounded-decoding rules introduced here remain the foundation for the later transport stages. The contract of GrpcFrameCodec.encode is deliberately explicit: the returned frame and the input message have independent lifetimes. Encoding must not move the input reader index or release the input buffer. The implementation currently copies the readable bytes into a byte array before applying compression: public static ByteBuf encode( ByteBufAllocator allocator, ByteBuf message, GrpcCompression.Codec compression) { boolean compressed = !compression.name().equals("identity"); byte[] payload = new byte[message.readableBytes()]; message.getBytes(message.readerIndex(), payload); if (compressed) { payload = compression.compress(payload); } return allocator.buffer(GrpcFrameCodec.HEADER_SIZE + payload.length) .writeByte(compressed ? 1 : 0) .writeInt(payload.length) .writeBytes(payload); } This is not a zero-copy implementation, and it should not be presented as the fastest possible design. The copy makes the ownership boundary easy to reason about first. If a later optimization uses a slice or a composite buffer, cancellation and exception paths must be re-proven instead of assuming that the old ownership rules still hold. decode uses Flux.defer so every subscription receives an independent decoder instance: return Flux.defer(() -> { var decoder = new Decoder( maxWireMessageSize, maxDecompressedMessageSize, compression, maxBufferedBytesPerStream); return input.concatMap(decoder::accept, 1) .concatWith(Flux.defer(decoder::finish)) .publishOn(Schedulers.immediate(), 1); }); The state is intentionally small: either the five-byte header is incomplete, or the header is complete and the decoder is collecting a payload of a known length. One input ByteBuf may contain one byte of a header, or three complete messages back-to-back. concatMap(..., 1) preserves source order and limits the number of source buffers being processed at once. The decoder still has to respect downstream demand when it emits decoded messages. Every source buffer is released in doFinally, including success, failure, and cancellation: private Flux<ByteBuf> accept(ByteBuf source) { return Flux.<ByteBuf>generate(sink -> { try { while (source.isReadable()) { // read the header, allocate a bounded payload, // and emit a complete message when available } sink.complete(); } catch (Throwable error) { sink.error(error); } }).doFinally(ignored -> source.release()); } The complete implementation also tracks an explicit per-stream buffered-byte limit. That limit covers an incomplete header, a partial payload, and bytes still present in the current source buffer. The length field comes from the peer, so it must be validated before allocating a payload array. The decoder first combines the unsigned big-endian bytes in a long, then checks the wire-size limit: long length = ((long) (header[1] & 0xff) << 24) | ((long) (header[2] & 0xff) << 16) | ((long) (header[3] & 0xff) << 8) | (header[4] & 0xffL); if (length > maxWireMessageSize) { throw new GrpcProtocolException("wire message length exceeds limit"); } Wire size and decompressed size are separate limits. A tiny gzip payload can expand into a huge message, so gzip decompression must enforce a second output limit to defend against decompression bombs. Only the lowest compression-flag bit is valid. Any reserved bit is a protocol error. A compressed frame received while the negotiated codec is still identity is also rejected; the decoder must not guess which algorithm the peer intended. Testing one arbitrary two-buffer split is not enough. A five-byte header has six representative split positions, including before the first byte and after the complete header. The test suite uses a dynamic test for every split from 0 through 5: IntStream.rangeClosed(0, GrpcFrameCodec.HEADER_SIZE) .mapToObj(split -> DynamicTest.dynamicTest( "split after byte " + split, () -> { byte[] wire = wireBytes("hello"); Flux<ByteBuf> chunks = Flux.just( wrapped(wire, 0, split), wrapped(wire, split, wire.length - split)); StepVerifier.create(GrpcFrameCodec.decode(chunks)) .assertNext(message -> assertMessage(message, "hello")) .verifyComplete(); })); The same test class covers arbitrary body fragmentation, multiple messages coalesced into one buffer, empty messages, gzip, reserved flags, truncated frames, wire/decompressed limits, and the fact that encoding does not consume the input buffer. See GrpcFrameCodecTest for the executable cases. Suppose one source ByteBuf contains three messages: one, two, and three. The downstream requests two messages and then cancels. The test must assert not only the values it received, but also that the source buffer was released: StepVerifier.create(GrpcFrameCodec.decode(Flux.just(source)), 0) .thenRequest(1) .assertNext(message -> assertMessage(message, "one")) .thenRequest(1) .assertNext(message -> assertMessage(message, "two")) .thenCancel() .verify(); assertEquals(0, source.refCnt()); That assertion is more important than a happy-path content check. Network code often behaves correctly under normal completion; leaks tend to appear during cancellation, size-limit failures, truncated frames, or competing terminal signals. Cancellation can also arrive before a complete message exists. In that case there is no decoded value for the subscriber to release, so the decoder itself must release the partially accumulated source buffer: @Test void releasesPartialFrameInputWhenCancelled() { ByteBuf partial = Unpooled.buffer(8) .writeByte(0) .writeInt(16) .writeBytes(new byte[]{1, 2, 3}); StepVerifier.create( GrpcFrameCodec.decode( Flux.just(partial).concatWith(Flux.never())), 0) .thenRequest(1) .thenAwait(java.time.Duration.ofMillis(10)) .thenCancel() .verify(); assertEquals(0, partial.refCnt()); } The full executable case is releasesPartialFrameInputWhenCancelled. It covers the lifecycle edge that a normal decode-complete test cannot exercise. Run only the frame codec suite from the repository root: ./gradlew :grpc-reactor-protocol:test \ --tests io.github.qianwj.grpc.reactor.protocol.GrpcFrameCodecTest \ --no-daemon On JDK 25, the Gradle build and generated JUnit report produced: GrpcFrameCodecTest: 15 tests, 0 failures, 0 errors, 0 skipped BUILD SUCCESSFUL in 3s At the Stage 1 boundary, the frame decoder verifies message-level demand but does not yet implement the two-level flow-control problem of the streaming transport. Reactive Streams counts messages, while HTTP/2 flow control counts bytes. They cannot be treated as the same quantity. Stage 3 later adds bounded inbound buffering and demand-aware delivery, and Stage 4 extends those rules to bidirectional streaming. Those transport and stress tests are covered in later posts. gRPC metadata is not a simple Map<String, String>. It must satisfy all of these rules: The same key may occur more than once, and insertion order matters. Keys may contain only lowercase letters, digits, _, ., and -, validated by [0-9a-z_.-]+. Keys ending in -bin carry binary values and use unpadded Base64 on the wire. Applications cannot set reserved fields such as content-type, te, grpc-status, or grpc-timeout. GrpcMetadata stores an immutable entry list so it can be safely shared across asynchronous boundaries: GrpcMetadata metadata = GrpcMetadata.builder() .addAscii("trace-id", "abc123") .addAscii("trace-id", "def456") // duplicates are allowed .addBinary("auth-token-bin", tokenBytes) .build(); // Order is preserved when reading. List<GrpcMetadata.Entry> all = metadata.getAll("trace-id"); // [abc123, def456] When parsing HTTP/2 headers, a binary value may be comma-joined by header handling. The implementation splits it on commas and decodes each Base64 segment independently. The total encoded size is bounded at 8 KiB by default, preventing a peer from exhausting memory with oversized headers. gRPC defines 17 standard status codes, each with a specific meaning for client error handling and future retry policies. GrpcStatus is a record containing a code and a human-readable message: public record GrpcStatus(Code code, String message) { public enum Code { OK(0), CANCELLED(1), UNKNOWN(2), INVALID_ARGUMENT(3), DEADLINE_EXCEEDED(4), NOT_FOUND(5), ALREADY_EXISTS(6), PERMISSION_DENIED(7), RESOURCE_EXHAUSTED(8), FAILED_PRECONDITION(9), ABORTED(10), OUT_OF_RANGE(11), UNIMPLEMENTED(12), INTERNAL(13), UNAVAILABLE(14), DATA_LOSS(15), UNAUTHENTICATED(16); } } Several details matter in practice: DEADLINE_EXCEEDED may be returned even after the operation completed successfully. If the successful response crosses the deadline in transit, the client can still observe a timeout. UNAVAILABLE indicates a transient failure for which a client may later retry safely; INTERNAL generally describes a server-side bug and should not be blindly retried. UNIMPLEMENTED carries the semantic meaning of an unsupported method, commonly surfaced through an HTTP 404 response at the protocol boundary. Text in the grpc-message trailer uses percent-encoding: printable ASCII characters other than % can pass through, while other bytes become %HH. This allows UTF-8 error descriptions to travel through ASCII HTTP/2 headers safely. Unknown numeric status codes are mapped to UNKNOWN instead of causing a parse failure. That preserves forward compatibility when a peer adopts a newer gRPC specification. The gRPC grpc-timeout header carries a relative duration, not an absolute timestamp. By the time the server receives a request, part of the caller's original time budget has already been consumed by transport latency. The wire format is compact: at most eight decimal digits followed by a unit suffix: 100m -> 100 milliseconds 2S -> 2 seconds 99999999H -> roughly 11,415 years (the maximum value) The six units are H (hours), M (minutes), S (seconds), m (milliseconds), u (microseconds), and n (nanoseconds). When formatting a Duration, the implementation rounds upward using ceiling division. The encoded deadline must never be shorter than the caller's requested duration: BigInteger amount = nanos.add(unitNanos.subtract(BigInteger.ONE)) .divide(unitNanos); // ceiling division BigInteger avoids overflow during nanosecond arithmetic. The formatter scans from nanoseconds upward and selects the first unit whose value fits within 99,999,999. GrpcCompression manages codec registration and negotiation: GrpcCompression.Registry registry = GrpcCompression.Registry.builder() .add(GrpcCompression.GZIP) .build(); // Produces grpc-accept-encoding: gzip String advertised = registry.advertisedEncodings(); identity is always implicit and appears first in the registry. The codec interface has only three operations: a name, byte-array compression, and bounded byte-array decompression. Gzip decompression reads in 8 KiB chunks and uses Math.addExact() while accumulating the output size. It can stop immediately after exceeding maxDecompressedSize, and arithmetic overflow cannot silently wrap the counter. A 100-byte gzip payload can expand to gigabytes, so decompression-bomb protection is part of the protocol contract rather than an optional optimization. Each RPC is described by one GrpcMethod record: var method = new GrpcMethod<>( "testing.InteropTestService", // full service name "Unary", // method name GrpcMethod.Cardinality.UNARY, new ProtobufMarshaller<>(TestRequest.parser()), new ProtobufMarshaller<>(TestResponse.parser())); method.path(); // /testing.InteropTestService/Unary method.fullMethodName(); // testing.InteropTestService/Unary The Cardinality enum exposes singleRequest() and singleResponse(). The transport uses those flags to insert single() at the API boundary, turning cardinality violations into explicit errors instead of silently dropping values. ProtobufMarshaller wraps a protobuf Parser<T>: public ByteBuf serialize(ByteBufAllocator allocator, T value) { var result = allocator.buffer(value.getSerializedSize()); return result.writeBytes(value.toByteArray()); } public T deserialize(ByteBuf message) { ByteBuffer bytes = message.nioBuffer(message.readerIndex(), message.readableBytes()); return parser.parseFrom(bytes); } The important contract is that deserialize reads through a NIO ByteBuffer view. It does not move the input reader index and does not release the input. Ownership remains with the caller, allowing the frame decoder's doFinally(release) to manage the buffer lifecycle uniformly whether parsing succeeds or fails. The protocol module does not need Reactor Netty on its classpath. That dependency boundary is itself part of the verification. The Stage 1 exit criteria are: every header and payload split point decodes correctly; cancellation leaves no unreleased ByteBuf (refCnt assertions); metadata preserves order, supports binary values, and enforces its size limit; unknown status codes do not throw; all timeout units parse correctly, formatting rounds upward, and arithmetic is overflow-safe; gzip decompression limits reject bombs; the marshaller does not change the input buffer state. These tests use JUnit 5 @TestFactory and DynamicTest for parameterization, together with Reactor Test's StepVerifier for asynchronous behavior. The protocol layer is the reason the later transport stages can focus on HTTP/2 lifecycle instead of rediscovering framing and ownership rules. The complete implementation is in GrpcFrameCodec.java. The related protocol primitives are GrpcMetadata, GrpcStatus, GrpcTimeout, GrpcCompression, and ProtobufMarshaller.