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. Encoding Must Define Ownership 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. The Decoder Is a Per-Subscription State Machine 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 accept(ByteBuf source) { return Flux. 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. Validate Peer Input Before Allocating 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) 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. Test Every Header Split Point 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 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. Cancellation Must Release Undelivered Data 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. Metadata: Ordering, Duplicates, and Binary Values gRPC metadata is not a simple Map . 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 , orgrpc-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 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. Status: 17 Codes and Percent-Encoding 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