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

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.

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:

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