DEV Community: Wang Lee

Unary gRPC on Reactor Netty: Event Loop Serialization, Trailers, and Cancellation

Wang Lee — Sun, 09 Aug 2026 03:18:25 +0000

With protocol values and message framing complete, Stage 2 delivered the first end-to-end call: plaintext h2c unary RPC. This is already on main, and Stage 3 and Stage 4 subsequently completed all four RPC cardinalities on the same transport primitive.

Previous: Building a Leak-Safe gRPC Frame Decoder on Reactor Netty

Method Descriptor Is Where Protocol Meets Types

A method requires a precise service name, method name, cardinality, and request/response marshallers:

var echo = new GrpcMethod<>(
    "testing.EchoService",
    "Echo",
    GrpcMethod.Cardinality.UNARY,
    new ProtobufMarshaller<>(StringValue.parser()),
    new ProtobufMarshaller<>(StringValue.parser()));

The generated path must be:

/testing.EchoService/Echo

The service registry matches by exact full path. An unknown path returns UNIMPLEMENTED; registering the same path twice fails immediately when building the service definition.

Server Validates Protocol Before Subscribing to Business Logic

ReactorGrpcServer uses Reactor Netty h2c:

DisposableServer bound = HttpServer.create()
    .host(host)
    .port(port)
    .protocol(HttpProtocol.H2C)
    .handle(handler::handle)
    .bindNow(Duration.ofSeconds(10));

Incoming requests are validated in order:

HTTP method must be POST;
content-type must be application/grpc or application/grpc+...;
te must declare trailers;
path must exist;
currently only unary cardinality is allowed;
metadata and message size must not exceed limits.

Only after validation passes does it create a GrpcCallContext and subscribe to the request body, preventing invalid requests from entering the business handler.

HTTP 200 Does Not Mean RPC Success

The server writes a compatible content-type first; the final status comes from trailing headers:

response.status(200)
    .header(HttpHeaderNames.CONTENT_TYPE, "application/grpc+proto");

response.trailerHeaders(trailers -> {
  GrpcException error = terminal.get();
  if (error == null) {
    writeStatus(trailers, GrpcStatus.OK, GrpcMetadata.empty());
  } else {
    writeStatus(trailers, error.status(), error.trailers());
  }
});

A GrpcException thrown by the application preserves an explicit status and trailing metadata. Ordinary application exceptions map to UNKNOWN; protocol violations map to INTERNAL. Fatal JVM errors should not be wrapped as ordinary business status.

After receiving a response, the client must read the final grpc-status. HTTP 200 without a status is INTERNAL; a non-gRPC HTTP 404 maps to UNIMPLEMENTED per the standard fallback, and 503 maps to UNAVAILABLE.

Metadata Must Preserve Order and Allow Duplicates

gRPC metadata is not a simple Map<String, String>. The same key can repeat, order must be preserved, and the -bin suffix indicates binary values encoded as no-padding Base64 on the wire.

GrpcMetadata therefore stores an immutable entry list internally:

GrpcMetadata.builder()
    .addAscii("trace-id", "abc")
    .addBinary("span-bin", bytes)
    .build();

User metadata cannot overwrite reserved fields like content-type, te, grpc-status, grpc-timeout, and HTTP/2 pseudo headers. The total encoded size is also limited on read to prevent peers from consuming unbounded memory through headers.

Deadline Wire Values Cannot Be Simply Truncated

grpc-timeout consists of up to eight digits and a unit, such as 100m or 2S. Formatting must use ceiling division to ensure the wire deadline is never shorter than what the caller requested:

BigInteger amount = ceilDivide(nanos, unit.nanos);
if (amount.compareTo(MAX_VALUE) <= 0) {
  return amount + String.valueOf(unit.suffix);
}

In the Stage 2 commit covered by this post, the client only writes the timeout header and sets the Reactor Netty response timeout slightly longer to distinguish gRPC deadlines from underlying connection waits. Merely being able to parse the header does not constitute full support; Stage 5 later adds client and server Reactor deadline timers, handler cancellation, and GrpcCallContext propagation. The current implementation still does not automatically derive shorter deadlines for nested calls on behalf of business code.

Unary Cardinality Is Enforced with single()

The client uses requests.single() for unary requests, and responses likewise require exactly one value. Empty requests, duplicate requests, empty responses, and duplicate responses are not silent defaults — they are explicit errors.

Current transport test coverage:

unary message and ASCII metadata round-trip;
declared error and trailing metadata;
unknown method;
empty/duplicate unary data;
non-gRPC HTTP response and missing status;
client cancellation propagation to server publisher;
connection failure mapping to UNAVAILABLE.

The unary vertical slice tests continue serving as regression baselines for streaming transport. Server, client, and bidirectional streaming have been completed in subsequent commits; TLS, GOAWAY, keepalive, compression, deadline, connection pool, and transport diagnostics were also completed in Stage 5 and merged to main.

Full Lifecycle of a Client Request

A ReactorGrpcClient unary call ultimately maps to a complete HTTP/2 stream interaction. Here is the full path from user call to network response:

client.unary(method, Mono.just(request))
    |
    +-- Flux.defer() -> lazy assembly, independent per subscription
    |
    +-- channel.select() -> pick subchannel (Stage 2: fixed single connection)
    |
    +-- httpClient.post(method.path())
    |   +-- Write HTTP/2 request headers
    |   |   content-type: application/grpc+proto
    |   |   te: trailers
    |   |   grpc-accept-encoding: gzip
    |   |   grpc-timeout: <if set>
    |   |   custom metadata
    |   |
    |   +-- onEventLoop() -> serialize on connection's event loop
    |   |
    |   +-- GrpcOutboundMessageEncoder.encode()
    |   |   +-- marshaller.serialize() -> Protobuf -> ByteBuf
    |   |   +-- validate size <= maxOutboundMessageSize
    |   |   +-- GrpcFrameCodec.encode() -> 5-byte header + payload
    |   |
    |   +-- outbound.send(requestFrames) -> HTTP/2 DATA frame
    |
    +-- .response((response, content) -> decodeResponse)
        +-- Check response headers for grpc-status (trailers-only error)
        +-- Validate HTTP 200 + content-type = application/grpc
        +-- Read grpc-encoding header
        +-- GrpcFrameCodec.decode(content) -> incremental frame parsing
        +-- marshaller.deserialize() -> ByteBuf -> Response protobuf
        +-- Read grpc-status + grpc-message from trailers

Key design: the entire pipeline is lazy — no HTTP/2 connection is initiated until the Mono is subscribed to. This differs from grpc-java's ClientCall.start() which triggers connection immediately.

Why Serialize on the Event Loop

Each Reactor Netty HTTP/2 connection is bound to a single Netty event loop thread. If marshalling happens on the user thread, the serialized ByteBuf must be transferred across threads to the event loop, incurring synchronization overhead.

onEventLoop() defers serialization to the connection's event loop:

private <T> Flux<T> onEventLoop(NettyOutbound outbound, Flux<T> source) {
    return source.publishOn(
        Schedulers.fromExecutor(outbound.alloc().buffer(0).alloc()
            ... // obtain event loop executor
        ));
}

Benefits:

Serialization and network writes happen on the same thread, lock-free;
The ByteBuf allocator is consistent with the connection, reducing cross-arena allocation;
concatMap(..., 1) controls prefetch, ensuring messages are sent in strict order.

Error Mapping: From HTTP/2 Exceptions to gRPC Status

Network-layer errors cannot be exposed directly to users — they need to be mapped to gRPC semantics:

Network Exception	gRPC Status	Meaning
`Http2Exception.StreamException(REFUSED_STREAM)`	UNAVAILABLE	Server refused this stream, safe to retry
GOAWAY frame	UNAVAILABLE	Server is shutting down gracefully
Connection closed / reset	UNAVAILABLE	Network disconnected
HTTP 404 (non-gRPC response)	UNIMPLEMENTED	Path doesn't exist (may have passed through a non-gRPC proxy)
HTTP 503	UNAVAILABLE	Service temporarily unavailable
Missing `grpc-status`	INTERNAL	Protocol violation
`content-type` mismatch	INTERNAL	Response is not gRPC

This mapping follows the HTTP status code to gRPC status conversion table defined in the gRPC protocol spec. UNAVAILABLE means the client can retry; INTERNAL and UNIMPLEMENTED are generally not worth retrying.

How Client Cancellation Propagates to Server

Reactor's dispose() needs to be mapped to HTTP/2's RST_STREAM, letting the server know this call is no longer needed:

client: subscription.dispose()
    -> Reactor Netty observes subscriber cancel
    -> Sends HTTP/2 RST_STREAM frame
    -> Server connection receives RST_STREAM
    -> request body publisher is cancelled
    -> server handler's Mono/Flux receives cancel signal

The server side listens for stream close events through a connection handler and converts them into a peerCancellation signal. The business handler uses takeUntilOther(peerCancellation) to subscribe to this signal, ensuring the server stops unnecessary computation after client cancellation.

This mechanism naturally aligns Reactor's backpressure semantics (cancel = I no longer need data) with gRPC's cancellation semantics (RST_STREAM = call aborted).

Interoperability Verification with grpc-java

Stage 2's exit criteria are not "calling ourselves successfully" — it must interoperate bidirectionally with grpc-java. UnaryInteroperabilityTest verifies both directions:

Reactor client -> grpc-java server:

@Test
void reactorClientCallsGrpcJavaServer() {
    try (var fixture = GrpcJavaFixture.start();
         var client = ReactorGrpcClient.builder().port(fixture.port()).build()) {
        var response = stub.unary(Mono.just(request("hello"))).block();
        assertEquals("hello", response.getValue());

        // Verify error status and trailing metadata propagate correctly
        var error = assertThrows(GrpcException.class,
            () -> stub.unary(Mono.just(request("error"))).block());
        assertEquals(GrpcStatus.Code.INVALID_ARGUMENT, error.status().code());
        assertEquals("rejected", error.status().message());
        assertEquals("fixture", error.trailers().getLast("error-id").orElseThrow().asciiValue());
    }
}

grpc-java client -> Reactor server:

@Test
void grpcJavaClientCallsReactorServer() {
    var service = InteropTestServiceReactorGrpc.bindService(new Service() {
        @Override
        public Mono<TestResponse> unary(GrpcCallContext context, Mono<TestRequest> requests) {
            return requests.map(req -> {
                if (req.getValue().equals("error")) {
                    throw new GrpcException(
                        new GrpcStatus(GrpcStatus.Code.INVALID_ARGUMENT, "rejected"),
                        GrpcMetadata.builder().addAscii("error-id", "reactor").build());
                }
                return TestResponse.newBuilder().setValue(req.getValue()).build();
            });
        }
    });

    try (var server = ReactorGrpcServer.builder().addService(service).start()) {
        var stub = InteropTestServiceGrpc.newBlockingStub(channel);
        assertEquals("hello", stub.unary(request("hello")).getValue());

        var error = assertThrows(StatusRuntimeException.class,
            () -> stub.unary(request("error")));
        assertEquals(Status.Code.INVALID_ARGUMENT, error.getStatus().getCode());
    }
}

Both directions verify: normal responses, error status codes, percent-encoding round-trip of error messages, and custom trailing metadata propagation. Only when grpc-java can correctly parse Reactor server's response can we prove the wire protocol implementation is compatible.

Stage 2 Exit Criteria

./gradlew clean spotlessCheck test --no-daemon passes in full
Reactor client <-> grpc-java server bidirectional unary calls succeed
grpc-java client <-> Reactor server bidirectional unary calls succeed
Error status, message, and trailing metadata propagate correctly
Client cancellation propagates to server via RST_STREAM
Connection failure maps to UNAVAILABLE
Unknown path maps to UNIMPLEMENTED
All Stage 0 and Stage 1 tests remain green

The next post enters Stage 3 and Stage 4: implementing server streaming, client streaming, and bidirectional streaming on the same transport primitive, facing new challenges of multi-message backpressure coordination, bounded inbound buffering, and resource cleanup on premature stream termination.

Building a Leak-Safe gRPC Frame Decoder on Reactor Netty

Wang Lee — Sat, 08 Aug 2026 06:09:36 +0000

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<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.

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) << 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.

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<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.

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<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.

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 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.

Timeout: Eight Digits and a Unit

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.

Compression: Identity by Default, Gzip Built In

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.

GrpcMethod: One Description for Four Cardinalities

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: Serialization Does Not Own the Input

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.

Stage 1 Exit Criteria

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.

Why Build gRPC Directly on Reactor Netty

Wang Lee — Fri, 07 Aug 2026 14:23:22 +0000

I recently started building grpc-reactor: an experimental gRPC implementation built directly on Reactor Netty HTTP/2. The project uses Protobuf but has zero runtime dependency on grpc-java transport, ClientCall, ServerCall, or StreamObserver.

This is not about proving grpc-java is bad. grpc-java is mature, stable, and covers load balancing, NameResolver, retries, and rich observability. This project answers a different question: if your programming model is already Reactor, can Mono and Flux flow all the way from the generated API down to the HTTP/2 stream, without adapting between two async abstractions?

This post is based on JDK 25, Gradle 9.2.1, Reactor 3.8.6, Reactor Netty 1.3.6, and Netty 4.2.15.Final. The main branch has completed Stage 0 through Stage 10: beyond protocol, four RPC cardinalities, production transport, DNS, codegen, standard services, and Stage 9 hardening, Stage 10 adds an optional load-balancing extensions artifact for grpclb, RLS, load reporting, and ORCA.

The API Goal Is Not Wrapping StreamObserver

Protobuf methods have four cardinalities. The target API maps them directly:

Protobuf Method	Reactor Signature
unary	`Mono<Resp> method(Mono<Req>)`
server streaming	`Flux<Resp> method(Mono<Req>)`
client streaming	`Mono<Resp> method(Flux<Req>)`
bidirectional streaming	`Flux<Resp> method(Flux<Req>)`

Why does gRPC define four instead of just unary? It's essentially a 2x2 combination where request and response each independently choose "single value or stream":

	Single Response	Streaming Response
Single Request	unary	server streaming
Streaming Request	client streaming	bidirectional

They solve different scenarios: unary covers classic request-response; server streaming handles server push (event subscriptions, large paginated pulls); client streaming handles bulk uploads (file chunks, batch writes); bidirectional streaming handles real-time two-way communication (chat, collaborative editing). These aren't invented patterns — HTTP/2 streams are inherently full-duplex. A single connection can multiplex hundreds of concurrent streams, with both request and response sending multiple data frames independently. gRPC elevates this transport capability to first-class API semantics: not a variant of chunked transfer encoding, but compiler-level type checking that callers and implementers match.

The low-level transport retains a single unified model:

Flux<Req> -> HTTP/2 stream -> Flux<Resp>

Generated code applies cardinality checks like single() at the boundary. This avoids implementing four separate network logic paths for four RPC types, while explicitly rejecting empty requests, duplicate requests, or duplicate responses in unary calls.

Compatibility Targets the Wire Protocol

The project's compatibility target is the public gRPC over HTTP/2 protocol spec, not grpc-java internal APIs:

generated Reactor API
        |
call dispatcher / service registry
        |
marshaller + message framer/deframer
        |
metadata + status + deadline
        |
Reactor Netty HTTP/2 stream

Each RPC corresponds to one HTTP/2 stream. Requests start with a HEADERS frame containing at minimum:

:method      POST
:path        /<package.Service>/<Method>
content-type application/grpc+proto
te           trailers
grpc-timeout <relative timeout, e.g. 100m for 100ms>

Message bodies use Length-Prefixed-Message format encapsulated in DATA frames — each Protobuf message is preceded by a five-byte envelope (1 byte compression flag + 4 bytes big-endian length). A single DATA frame may contain multiple gRPC messages, and a large message may span multiple DATA frames.

Why does gRPC require HTTP trailers? This is one of the protocol's most counterintuitive designs. HTTP status codes are nearly useless for gRPC — the protocol requires the HTTP layer to always return 200, with the actual call result (grpc-status and grpc-message) placed in trailers. The reason: in streaming scenarios, the server may have already sent thousands of messages, and whether it ultimately succeeded or failed can only be determined after processing the last piece of data. HTTP headers are sent before the body and cannot carry this posterior result. Trailers are the only mechanism in HTTP/2 that can append metadata after the body.

The protocol also defines trailers-only mode: when the server can determine failure before reading the body (e.g., path not found, authentication failed), grpc-status is returned directly in response headers without sending a body, saving one round-trip. Clients must check both headers and trailers to correctly extract the final status.

Modules Split Along Protocol Boundaries

The project currently has nine modules:

grpc-reactor-protocol      → Transport-independent protocol primitives
grpc-reactor-transport     → Reactor Netty HTTP/2 mapping
grpc-reactor-codegen       → Protoc plugin, generates Reactor stubs
grpc-reactor-gradle-plugin → Gradle integration
grpc-reactor-maven-plugin  → Maven generate-sources integration
grpc-reactor-services      → Optional Health, Reflection & Channelz standard services
grpc-reactor-binlog        → Optional, bounded canonical binary logging
grpc-reactor-lb-extensions → Optional grpclb, RLS, load reporting & ORCA
grpc-reactor-interop-test  → grpc-java compatibility tests

protocol handles only transport-independent values and codecs: message framing, metadata, status, timeout, compression, and protobuf marshallers. It depends on Reactor Core and Netty Buffer but not Reactor Netty — meaning the protocol layer can be tested independently without real HTTP/2 connections.

transport maps the protocol onto Reactor Netty HTTP/2, handling client, server, service registry, and per-call context.

codegen consumes Protobuf CodeGeneratorRequest and generates type-safe Reactor client and service binders.

services uses the same codegen to generate canonical gRPC service bindings, implementing Health v1, Reflection v1, and Channelz v1 on top of transport's immutable descriptor/diagnostics snapshots. This module requires explicit registration — adding the dependency alone won't expose management endpoints.

binlog is also an explicitly-enabled standalone module. It captures canonical binary-log v1 events via a transport interceptor, controlling information exposure and memory limits through metadata/message truncation, sensitive key redaction, and fixed-capacity sinks.

interop-test introduces grpc-java in test scope. grpc-java serves as the compatibility oracle here, never entering the project runtime.

Dependency Selection and Version Pinning

The runtime dependency chain is intentionally kept short:

Dependency	Version	Purpose
Reactor Core	3.8.6	Mono/Flux programming model
Reactor Netty	1.3.6	HTTP/2 client/server
Netty	4.2.15	ByteBuf, HTTP/2 codec
Protobuf-java	4.35.1	Message serialization

The project does not introduce Spring, Micrometer, or any DI framework. Tests use JUnit 6.1.2 and Reactor Test's StepVerifier. grpc-java 1.82.2 appears only in interop-test's test classpath with zero runtime intrusion.

Versions are centrally managed via gradle/libs.versions.toml with Gradle dependency locking generating lock files, ensuring fully reproducible builds across machines.

Why the Protocol Layer Must Come First

It's tempting to start with "spin up an HTTP/2 Server" — you quickly get an echo demo, but it pushes the truly difficult problems to later:

DATA may split in the middle of the five-byte gRPC header;
A single DATA buffer may contain multiple messages;
Metadata allows duplicate keys, and binary values require Base64;
Timeout wire values are at most eight digits with a unit suffix;
Final status comes from trailers;
ByteBuf must be released on success, failure, and cancellation paths;
Reactive Streams demand counts messages, HTTP/2 flow control counts bytes.

Therefore the project progresses by Stage: first lock down the build and interop fixtures, then complete the protocol layer, then implement unary transport, streaming, production features, and codegen. Each Stage has executable exit criteria — "classes have been created" is not a completion standard.

Stage 0: Build Baseline and Interop Fixture

Before writing any protocol code, Stage 0 solves "how to prove code is correct":

JDK 25 compilation — The project uses -Xlint:all -parameters -encoding UTF-8 for strict compilation. All warnings are compilation errors; silent suppression is not allowed. JDK 25 was chosen to validate Netty and Protobuf compatibility on the latest JVM early.

Spotless formatting — Unified Eclipse formatter config plus ktlint. spotlessCheck is the first gate in CI. This eliminates all code review discussions about formatting.

interop.proto test fixture — Defines a test service covering all four RPC cardinalities:

service InteropTestService {
  rpc Unary (TestRequest) returns (TestResponse);
  rpc ServerStreaming (TestRequest) returns (stream TestResponse);
  rpc ClientStreaming (stream TestRequest) returns (TestResponse);
  rpc BidirectionalStreaming (stream TestRequest) returns (stream TestResponse);
}

GrpcJavaFixture — In the interop-test module, a test utility class starts both a grpc-java server and client, providing start() / close() lifecycle. Through it, bidirectional verification is possible: Reactor client calls grpc-java server, and grpc-java client calls Reactor server. Both use the same .proto generated code, ensuring wire compatibility.

Utilities:

FreePorts: Allocates independent ports for each test, avoiding parallel test conflicts;
TlsTestCertificates: Pre-generates self-signed certificates for subsequent TLS tests;
LeakDetection: Integrates Netty's ResourceLeakDetector, ensuring ByteBuf leaks are immediately exposed in tests.

Stage 0 exit criteria: ./gradlew clean test passes from fresh checkout, CI is green on Linux + Java 25, protoc generation is deterministically reproducible, grpc-java fixture communicates bidirectionally.

Staged Verification Strategy

The project progresses through 12 Stages, each with clear goals, executable exit criteria, and regression coverage:

Stage	Goal	Key Deliverable
0	Build baseline	CI, formatting, interop fixture
1	Protocol foundation	Frame codec, metadata, status, timeout, compression
2	Unary transport	End-to-end h2c unary call
3	Server streaming	Multi-message response stream
4	Full cardinality	Client streaming + bidirectional
5	Production transport	TLS, gzip, deadline, GOAWAY, connection pool, keepalive
6	Name resolution	DNS, subchannel, pick_first, round_robin
7	Codegen & build integration	Protoc plugin, descriptor registry, Gradle/Maven plugins
8	Standard services	Health v1, Reflection v1, Channelz v1
9	Operations & hardening	Interceptor, observer, binlog, canonical smoke, fuzz/churn
10	Load-balancing extensions	grpclb, RLS, load reporter, ORCA
11	Diagnostics (planned)	Channelz v2 over the bounded diagnostics registry

Each Stage completion requires: ./gradlew clean spotlessCheck test --no-daemon passes in full, and all prior Stage tests continue running as regression. This means when Stage 4 completes, Stage 2's unary tests are still green.

Where Reactor semantics differ from grpc-java

The wire contract is shared, but the application contracts are not. The tests therefore compare the two implementations at the protocol boundary and test each runtime's lifecycle rules separately:

Concern	Reactor contract	grpc-java contract	What the tests must prove
Demand	`Subscription.request(n)` counts decoded messages; transport demand must also respect HTTP/2 byte windows	inbound flow is controlled through `ClientCall.request(n)` / readiness callbacks	no response DATA is delivered before downstream demand, and coalesced frames do not bypass demand
Cancellation	disposing a `Mono`/`Flux` cancels both application publishers and resets an open stream	`ClientCall.cancel` or `Context` cancellation reaches `StreamObserver` callbacks	both sides stop, exactly one terminal signal wins, and late DATA/trailers are ignored
Trailers	success trailers are retained in the call result; non-OK trailers become `GrpcException.trailers()`	trailers are exposed through `ClientCall.Listener#onClose` or `StatusRuntimeException#getTrailers()`	`grpc-status`, `grpc-message`, and custom trailers survive in both directions, including trailers-only errors
Buffer ownership	Netty `ByteBuf` is reference-counted and must be released on success, error, and cancel; decoded protobuf values cross the API boundary	generated protobuf messages hide transport buffers from application code	every rejected, partial, compressed, and cancelled frame reaches `refCnt() == 0`

For example, the Reactor demand test starts with zero demand and asks for one response at a time:

StepVerifier.create(client.bidirectionalStreaming(BIDI, requests), 0)
        .thenRequest(1)
        .expectNext(first)
        .thenRequest(2)
        .expectNext(second, third)
        .expectComplete()
        .verify();

The corresponding grpc-java test uses StreamObserver callbacks and completion signals. The messages and trailers must be wire-compatible, but it would be incorrect to describe the two tests as asserting the same backpressure mechanism. This distinction is why the project keeps both cross-runtime interop and runtime-specific failure tests.

The repository now also has two server-streaming cancellation interop cases in ServerStreamingInteroperabilityTest: a Reactor client receives one response from a grpc-java server and cancels, while a grpc-java client cancels a Reactor server after its first response. Both sides wait for the peer's cancellation callback under paranoid leak detection. The lower-level GrpcFrameCodecTest keeps the direct ownership assertion by checking that cancelled and partial inputs finish with refCnt() == 0.

The essential cancellation paths are deliberately small:

// Reactor client: cancel the HTTP/2 call after the first response.
TestResponse first = reactorStub
        .serverStreaming(Mono.just(request(5)))
        .take(1)
        .single()
        .block(Duration.ofSeconds(5));

// grpc-java client: cancel from the response callback.
@Override
public void onNext(TestResponse response) {
    requestStream.cancel("cancel after first response", null);
}

On the grpc-java server side, the test installs setOnCancelHandler; on the Reactor server side, the response Flux uses doOnCancel. The complete test, including latches, peer status assertions, shutdown, and leak-detection scope, is in ServerStreamingInteroperabilityTest.java.

Current Verification Results

The project uses this unified gate that simultaneously checks formatting, compilation, protocol tests, transport tests, and grpc-java interop tests:

./gradlew clean spotlessCheck test --no-daemon

This command passes as of this writing. The gate now includes the optional Stage 10 load-balancing extension tests in addition to protocol, transport, codegen, services, binary-log, and grpc-java interop tests. The dependency baseline is Protobuf 4.35.1, grpc-java 1.82.2, and JUnit 6.1.2. Under JDK 25, you'll still see Protobuf Unsafe, Netty/Gradle native access, and Gradle deprecated feature warnings — they don't affect test results but need ongoing tracking through future JDK and Gradle upgrades.

Subsequent posts will continue discussing the five-byte gRPC message envelope, ByteBuf ownership, four RPC cardinalities, production transport, DNS, code generation, standard management services, and how to integrate interceptors, observation, and binary logging without breaking Reactive Streams semantics.