From e859f8155c7d3a4df5d5cf5883632b5db48a0898 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 5 Nov 2026 09:31:32 -0500 Subject: [PATCH 1/6] Stream gRPC client calls in both directions RestletClientStream sent the whole request and read the whole response, so the transport factory refused every non-unary method. Start the HTTP/2 exchange with the request headers instead, and feed the request body frame by frame from a bounded queue, ending at the half-close; the Netty client already writes a body of unknown size in the background on HTTP/2 and returns at the response head, so messages flow both ways. Report written frames to the SDK, so readiness follows the network, and fail a sender that queues past the budget with RESOURCE_EXHAUSTED. Read the response as it arrives, keeping the bytes the deframer hasn't consumed within the budget, so an application that requests nothing pushes back on the server. Add GrpcClientAdapter.setMaxQueuedBytes, mirroring the server adapter. Only methods of unknown type are refused. Test every streaming type against grpc-java's Netty server and against GrpcServerAdapter, with readiness, demand, large messages, cancellation and deadlines. Give the server's push-back test room on loaded machines. Part of #1561 (R6a). Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 10 +- org.restlet.ext.grpc/pom.xml | 6 + .../restlet/ext/grpc/GrpcClientAdapter.java | 38 +- .../grpc/internal/RestletClientStream.java | 224 +++++-- .../internal/RestletTransportFactory.java | 12 +- .../ext/grpc/GrpcClientAdapterTestCase.java | 6 +- .../ext/grpc/GrpcClientStreamingTestCase.java | 561 ++++++++++++++++++ .../ext/grpc/GrpcServerAdapterTestCase.java | 2 + .../internal/RestletClientStreamTestCase.java | 4 +- readme.md | 4 +- 10 files changed, 808 insertions(+), 59 deletions(-) create mode 100644 org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java diff --git a/changes.md b/changes.md index 28fa95112c..227390c6f4 100644 --- a/changes.md +++ b/changes.md @@ -107,14 +107,18 @@ Changes log leaves the borrowed listener and supplied callback executor running and reports timeout explicitly. TLS configuration and physical gRPC connection controls are not supported by the server adapter. - - Added an experimental SDK-native gRPC unary client on a borrowed Restlet HTTP/2 - connector, with explicit plaintext opt-in, bounded messages and call admission. + - Added an experimental SDK-native gRPC client on a borrowed Restlet HTTP/2 connector, + for unary, server-streaming, client-streaming and bidirectional calls, with explicit + plaintext opt-in, bounded messages and call admission. Each call is one HTTP/2 exchange + whose request body is written while its response is read; request messages are queued + within a per-call budget that the SDK's readiness follows, and response messages are read + as the application requests them, within the same budget. Regression tests cover authority mapping, cancellation before and after headers, disconnect recovery, malformed responses and shared connector ownership. Empty responses retain SDK missing-status diagnostics instead of failing on a null stream. Shutdown arriving before SDK listener installation is replayed once; interrupted shutdown still closes owned transport resources and preserves the interrupt. - TLS, streaming methods and physical gRPC connection controls remain unsupported. + TLS and physical gRPC connection controls remain unsupported. - Added the Smithy extension, with a smithy-java 1.8.0 client transport on Restlet HTTP client connectors. Generated and dynamic SDK clients find it through service loading, before the JDK transport, or are given one borrowing an existing client. Requests are diff --git a/org.restlet.ext.grpc/pom.xml b/org.restlet.ext.grpc/pom.xml index 61a62faf25..909df28fe4 100644 --- a/org.restlet.ext.grpc/pom.xml +++ b/org.restlet.ext.grpc/pom.xml @@ -76,5 +76,11 @@ ${lib-junit-version} test + + org.junit.jupiter + junit-jupiter-params + ${lib-junit-version} + test + diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java index 48950242a0..d23dc1a3dd 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcClientAdapter.java @@ -28,10 +28,14 @@ /** * Single-use native gRPC channel on a borrowed Restlet HTTP/2 client. * - *

This initial profile is unary and plaintext only; applications must explicitly opt in with - * {@link #setPlaintext(boolean)}. TLS/address separation, proxy CONNECT, streaming, retry, - * keepalive and physical-connection controls are not implemented. No SDK network transport is - * selected. The host starts and stops the borrowed connector. + *

Supports unary, server-streaming, client-streaming and bidirectional calls: each call is one + * HTTP/2 exchange whose request body is written while its response is read, so messages flow both + * ways at once. Request messages are queued within {@link #setMaxQueuedBytes}, and the SDK's + * readiness follows their writing; response messages are read as the application requests them, + * within the same budget. This profile is plaintext only; applications must explicitly opt in with + * {@link #setPlaintext(boolean)}. TLS/address separation, proxy CONNECT, retry, keepalive and + * physical-connection controls are not implemented. No SDK network transport is selected. The host + * starts and stops the borrowed connector. */ public final class GrpcClientAdapter implements AutoCloseable { private final Context context; @@ -40,6 +44,7 @@ public final class GrpcClientAdapter implements AutoCloseable { private String target; private boolean plaintext; private int maxMessageBytes = 1024 * 1024; + private int maxQueuedBytes = 256 * 1024; private int maxConcurrentCalls = 128; private Duration maxCallDuration = Duration.ofSeconds(30); private Duration shutdownTimeout = Duration.ofSeconds(5); @@ -100,6 +105,25 @@ public synchronized int getMaxMessageBytes() { return maxMessageBytes; } + /** + * Sets the per-call queue budget, for each direction: request bytes waiting for the network, + * and response bytes read but not yet consumed by the messages the application requested. A + * single message may exceed it. + * + * @param value At least 16384 bytes. + */ + public synchronized void setMaxQueuedBytes(int value) { + requireConfigurable(); + if (value < 16384) { + throw new IllegalArgumentException("Queue budget must be at least 16384 bytes"); + } + maxQueuedBytes = value; + } + + public synchronized int getMaxQueuedBytes() { + return maxQueuedBytes; + } + public synchronized void setMaxConcurrentCalls(int value) { requireConfigurable(); if (value <= 0) { @@ -191,7 +215,11 @@ public synchronized void start() { configured = true; var factory = new RestletTransportFactory( - client, maxMessageBytes, maxConcurrentCalls, shutdownTimeout); + client, + maxMessageBytes, + maxQueuedBytes, + maxConcurrentCalls, + shutdownTimeout); try { var builder = new ManagedChannelImplBuilder(target, () -> factory, () -> 80) diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java index 5b6e6c37bd..56bbbaa08b 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java @@ -24,10 +24,13 @@ import io.grpc.internal.TransportFrameUtil; import io.grpc.internal.TransportTracer; import io.grpc.internal.WritableBuffer; -import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.InputStream; +import java.io.InterruptedIOException; +import java.io.OutputStream; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; import java.util.concurrent.ExecutorService; @@ -40,28 +43,61 @@ import org.restlet.data.MediaType; import org.restlet.data.Method; import org.restlet.data.Reference; -import org.restlet.representation.ByteArrayRepresentation; +import org.restlet.representation.OutputRepresentation; +import org.restlet.representation.Representation; import org.restlet.util.Series; -/** SDK-framed, bounded unary exchange. No gRPC frame/status codec is implemented here. */ +/** + * One gRPC call over one HTTP/2 exchange of a borrowed Restlet client, streaming in both + * directions. The SDK frames and deframes messages; this stream only moves their bytes. + * + *

The exchange starts with the request headers. The request body is fed frame by frame from a + * bounded queue, written by the client's background writer, and ends at the half-close; written + * bytes are reported to the SDK, which derives its readiness from them. The response body is read + * as it arrives, while the request is still written: bytes handed to the SDK but not yet consumed + * by its deframer, which only reads the messages the application requested, stay within the queue + * budget too, so a slow consumer stops the reads, and HTTP/2 flow control stops the server. + */ @SuppressWarnings("null") final class RestletClientStream extends AbstractClientStream { + private static final MediaType GRPC = MediaType.valueOf("application/grpc"); + private final SerializingExecutor serial = new SerializingExecutor(Runnable::run); private final State state; private final Client client; private final String target; private final String authority; private final int wireLimit; + private final int queueBytes; private final ExecutorService io; private final Consumer finished; - private final ByteArrayOutputStream outbound = new ByteArrayOutputStream(); private final StatsTraceContext stats; + + /** Request frames waiting for the network; also guards the outbound fields below. */ + private final ArrayDeque outbound = new ArrayDeque<>(); + + /** Guards {@link #unread}, and wakes the response reader. */ + private final Object inbound = new Object(); + private Metadata headers; private volatile Request request; private volatile Thread worker; private volatile Deadline deadline; private volatile boolean done; - private boolean sent; + private volatile boolean failed; + private boolean started; + + /** Request bytes queued; guarded by {@link #outbound}. */ + private int queued; + + /** Bytes framed for the latest request message, from the frame that started it. */ + private int open; + + /** Whether the SDK half-closed the request; guarded by {@link #outbound}. */ + private boolean halfClosed; + + /** Response bytes handed to the SDK but not consumed yet; guarded by {@link #inbound}. */ + private long unread; RestletClientStream( Client client, @@ -72,6 +108,7 @@ final class RestletClientStream extends AbstractClientStream { CallOptions options, ClientStreamTracer[] tracers, int limit, + int queueBytes, ExecutorService io, Consumer finished) { this( @@ -82,6 +119,7 @@ final class RestletClientStream extends AbstractClientStream { headers, options, limit, + queueBytes, io, finished, StatsTraceContext.newClientContext(tracers, Attributes.EMPTY, headers), @@ -96,6 +134,7 @@ private RestletClientStream( Metadata headers, CallOptions options, int limit, + int queueBytes, ExecutorService io, Consumer finished, StatsTraceContext stats, @@ -115,10 +154,12 @@ private RestletClientStream( } target = "http://" + ip + ":" + address.getPort() + "/" + method.getFullMethodName(); wireLimit = limit + 5; + this.queueBytes = queueBytes; this.io = io; this.finished = finished; this.stats = stats; state = new State(limit, stats, tracer, options); + setOnReadyThreshold(Math.max(1, queueBytes / 2)); } @Override @@ -152,32 +193,51 @@ public void writeHeaders(Metadata metadata, byte[] payload) { headers = metadata; stats.clientOutboundHeaders(); serial.execute(state::allocated); + Status failure = null; + synchronized (outbound) { + if (!done && !started) { + started = true; + try { + io.execute(RestletClientStream.this::exchange); + } catch (RejectedExecutionException e) { + failure = Status.UNAVAILABLE.withCause(e); + } + } + } + if (failure != null) { + fail(failure); + } } @Override public void writeFrame(WritableBuffer frame, boolean end, boolean flush, int messages) { Status failure = null; synchronized (outbound) { - if (!done && frame instanceof Buffer buffer) { - if (buffer.readableBytes() > wireLimit - outbound.size()) { + if (frame instanceof Buffer buffer && !done && !halfClosed) { + int size = buffer.size; + // One message is framed whole, possibly before the writer starts: while + // the queue holds only that message, it may grow up to one framed + // message. A frame counting messages starts a new one. + int message = messages > 0 ? size : open + size; + boolean others = queued + size > message; + if (size > queueBytes - queued && (others || message > wireLimit)) { failure = Status.RESOURCE_EXHAUSTED.withDescription( - "Encoded unary request exceeds byte budget"); - } else { - outbound.write(buffer.data, 0, buffer.size); + "Request queue byte budget exceeded"); + } else if (size > 0) { + open = message; + outbound.addLast(Arrays.copyOf(buffer.data, size)); + queued += size; + onSendingBytes(size); } } if (frame != null) { frame.release(); } - if (failure == null && end && !done && !sent) { - sent = true; - try { - io.execute(RestletClientStream.this::exchange); - } catch (RejectedExecutionException e) { - failure = Status.UNAVAILABLE.withCause(e); - } + if (end) { + halfClosed = true; } + outbound.notifyAll(); } if (failure != null) { fail(failure); @@ -192,6 +252,7 @@ public void cancel(Status status) { } private void fail(Status status) { + failed = true; Deadline current = deadline; Status reported = (status.getCode() == Status.Code.UNAVAILABLE @@ -210,6 +271,7 @@ private void fail(Status status) { state.fail(reported); } }); + wake(); Request active = request; if (active != null) { active.abort(); @@ -220,6 +282,16 @@ private void fail(Status status) { } } + /** Wakes the request writer and the response reader, once the call failed or ended. */ + private void wake() { + synchronized (outbound) { + outbound.notifyAll(); + } + synchronized (inbound) { + inbound.notifyAll(); + } + } + private void exchange() { worker = Thread.currentThread(); Response response = null; @@ -227,17 +299,7 @@ private void exchange() { if (done) { return; } - byte[] body; - synchronized (outbound) { - body = outbound.toByteArray(); - outbound.reset(); - } - Request call = - new Request( - Method.POST, - target, - new ByteArrayRepresentation( - body, MediaType.valueOf("application/grpc"))); + Request call = new Request(Method.POST, target, requestEntity()); // Resource origin selects the resolved peer; base reference selects HTTP authority. call.getResourceRef().setBaseRef(new Reference("http://" + authority)); call.getHeaders().set("te", "trailers"); @@ -254,7 +316,7 @@ private void exchange() { } call.getClientInfo().setAgent("restlet-grpc/1.85.1"); request = call; - if (done || Thread.currentThread().isInterrupted()) { + if (done || failed || Thread.currentThread().isInterrupted()) { call.abort(); return; } @@ -265,6 +327,7 @@ private void exchange() { Status.UNAVAILABLE .withDescription("gRPC peer disconnected") .withCause(cause))); + // Returns at the response head; the request body is still written meanwhile. response = client.handle(call); if (done) { return; @@ -293,15 +356,8 @@ private void exchange() { serial.execute(() -> state.headers(initial)); if (response.getEntity() != null) { try (var input = response.getEntity().getStream()) { - if (input != null) { - byte[] bytes = input.readNBytes(wireLimit + 1); - if (bytes.length > wireLimit) { - fail( - Status.RESOURCE_EXHAUSTED.withDescription( - "Encoded unary response exceeds byte budget")); - return; - } - serial.execute(() -> state.data(bytes)); + if (input != null && !readAll(input)) { + return; } } } @@ -317,6 +373,79 @@ private void exchange() { } } + /** + * Hands the response body to the SDK as it arrives, within the queue budget. + * + * @return False if the call ended before the end of the body. + */ + private boolean readAll(InputStream input) throws IOException { + byte[] chunk = new byte[16384]; + for (; ; ) { + int room; + synchronized (inbound) { + try { + while (unread >= queueBytes && !done && !failed) { + inbound.wait(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted reading the gRPC response"); + } + if (done || failed) { + return false; + } + room = (int) Math.min(chunk.length, queueBytes - unread); + } + int count = input.read(chunk, 0, room); + if (count < 0) { + return true; + } + if (count > 0) { + synchronized (inbound) { + unread += count; + } + byte[] bytes = Arrays.copyOf(chunk, count); + serial.execute(() -> state.data(bytes)); + } + } + } + + /** Returns the request entity, written by the client's background writer from the queue. */ + private Representation requestEntity() { + return new OutputRepresentation(GRPC, Representation.UNKNOWN_SIZE) { + @Override + public void write(OutputStream output) throws IOException { + for (; ; ) { + byte[] data; + synchronized (outbound) { + try { + while (outbound.isEmpty() && !halfClosed && !done && !failed) { + outbound.wait(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException( + "Interrupted writing the gRPC request"); + } + if (failed) { + throw new IOException("gRPC call cancelled"); + } + data = outbound.pollFirst(); + if (data == null) { + // Half-closed, or the call ended first: the request body ends. + return; + } + queued -= data.length; + } + output.write(data); + output.flush(); + int sent = data.length; + serial.execute(() -> state.sent(sent)); + } + } + }; + } + private static Metadata metadata(Series

headers) { var values = new ArrayList(); for (Header header : headers) { @@ -331,7 +460,7 @@ private static Metadata metadata(Series
headers) { } private final class State extends Http2ClientStreamTransportState { - private boolean started; + private boolean allocatedOnce; private Status pendingFailure; private State( @@ -340,7 +469,7 @@ private State( } private void allocated() { - started = true; + allocatedOnce = true; if (!done) { if (pendingFailure != null) { transportReportStatus(pendingFailure, true, new Metadata()); @@ -351,7 +480,7 @@ private void allocated() { } private void fail(Status status) { - if (started) { + if (allocatedOnce) { transportReportStatus(status, true, new Metadata()); } else if (pendingFailure == null) { // Shutdown may arrive after stream creation but before SDK listener installation. @@ -377,6 +506,12 @@ private void trailers(Metadata metadata) { } } + private void sent(int count) { + if (!done) { + onSentBytes(count); + } + } + @Override public void runOnTransportThread(Runnable runnable) { serial.execute(runnable); @@ -384,7 +519,11 @@ public void runOnTransportThread(Runnable runnable) { @Override public void bytesRead(int count) { - // Unary wire input is bounded and materialized before SDK deframing. + // The deframer consumed response bytes: free that much of the inbound budget. + synchronized (inbound) { + unread -= count; + inbound.notifyAll(); + } } @Override @@ -405,6 +544,7 @@ public void deframerClosed(boolean partial) { // and a caller that starts its next call right away must not find the slot still taken. if (!done) { done = true; + wake(); finished.accept(RestletClientStream.this); } super.deframerClosed(partial); diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java index dd28450b53..f60108b2f0 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletTransportFactory.java @@ -43,6 +43,7 @@ public final class RestletTransportFactory implements ClientTransportFactory { private final Client client; private final int maxBytes; + private final int queueBytes; private final Duration shutdownTimeout; private final Semaphore admission; private final Set transports = new HashSet<>(); @@ -52,9 +53,11 @@ public final class RestletTransportFactory implements ClientTransportFactory { 1, Thread.ofPlatform().daemon().name("restlet-grpc-timer").factory()); private boolean closed; - public RestletTransportFactory(Client client, int maxBytes, int maxCalls, Duration timeout) { + public RestletTransportFactory( + Client client, int maxBytes, int queueBytes, int maxCalls, Duration timeout) { this.client = client; this.maxBytes = maxBytes; + this.queueBytes = queueBytes; admission = new Semaphore(maxCalls); shutdownTimeout = timeout; // Cancelled deadline timers of completed calls must not stay queued until they expire. @@ -161,8 +164,10 @@ public synchronized ClientStream newStream( Status.UNIMPLEMENTED.withDescription( "Per-call authority override is not supported"); } - if (rejected == null && method.getType() != MethodDescriptor.MethodType.UNARY) { - rejected = Status.UNIMPLEMENTED.withDescription("R3 supports unary methods only"); + if (rejected == null && method.getType() == MethodDescriptor.MethodType.UNKNOWN) { + rejected = + Status.UNIMPLEMENTED.withDescription( + "Methods of unknown type are not supported"); } if (rejected == null && options.getCompressor() != null @@ -189,6 +194,7 @@ public synchronized ClientStream newStream( options, tracers, maxBytes, + queueBytes, io, this::finished); streams.add(stream); diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java index 5b19cde07b..504ce74be0 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java @@ -522,7 +522,7 @@ void interruptedStopStillClosesOwnedChannelAndPreservesBorrowedResources() throw } @Test - void rejectsStreamingAndOutboundLimitBeforeSending() throws Exception { + void rejectsUnknownMethodTypesAndOutboundLimitBeforeSending() throws Exception { adapter.setMaxMessageBytes(128); start(echo()); assertEquals( @@ -531,8 +531,8 @@ void rejectsStreamingAndOutboundLimitBeforeSending() throws Exception { .getStatus() .getCode()); assertEquals("small", call("small").getValue()); - var streamMethod = - ECHO.toBuilder().setType(MethodDescriptor.MethodType.SERVER_STREAMING).build(); + // Streaming methods are supported; only methods of unknown type are refused. + var streamMethod = ECHO.toBuilder().setType(MethodDescriptor.MethodType.UNKNOWN).build(); assertEquals( Status.Code.UNIMPLEMENTED, assertThrows( diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java new file mode 100644 index 0000000000..eb4d91be66 --- /dev/null +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java @@ -0,0 +1,561 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *

+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.grpc; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.google.protobuf.StringValue; +import io.grpc.CallOptions; +import io.grpc.ClientCall; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.ServerServiceDefinition; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder; +import io.grpc.protobuf.ProtoUtils; +import io.grpc.stub.ClientCalls; +import io.grpc.stub.ServerCallStreamObserver; +import io.grpc.stub.ServerCalls; +import io.grpc.stub.StreamObserver; +import java.time.Duration; +import java.util.ArrayList; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.restlet.Client; +import org.restlet.Context; +import org.restlet.Server; +import org.restlet.data.Protocol; +import org.restlet.routing.Router; + +/** + * Streaming calls of every type through {@link GrpcClientAdapter}, against grpc-java's own Netty + * server and against a {@link GrpcServerAdapter}, so that both ends may be Restlet. + */ +class GrpcClientStreamingTestCase { + private static final MethodDescriptor ECHO = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName("review.Streams/Echo") + .setRequestMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance())) + .setResponseMarshaller(ProtoUtils.marshaller(StringValue.getDefaultInstance())) + .build(); + private static final MethodDescriptor WATCH = + ECHO.toBuilder() + .setType(MethodDescriptor.MethodType.SERVER_STREAMING) + .setFullMethodName("review.Streams/Watch") + .build(); + private static final MethodDescriptor UPLOAD = + ECHO.toBuilder() + .setType(MethodDescriptor.MethodType.CLIENT_STREAMING) + .setFullMethodName("review.Streams/Upload") + .build(); + private static final MethodDescriptor CHAT = + ECHO.toBuilder() + .setType(MethodDescriptor.MethodType.BIDI_STREAMING) + .setFullMethodName("review.Streams/Chat") + .build(); + private static final String BLOCK = "x".repeat(4096); + + private final GrpcClientAdapter adapter = new GrpcClientAdapter(); + private Client client; + private io.grpc.Server nettyPeer; + private GrpcServerAdapter restletPeer; + private Server http; + + /** Starts the peer, Restlet or Netty, and the adapter calling it. */ + private void start(boolean restlet, ServerServiceDefinition service) throws Exception { + int port; + if (restlet) { + restletPeer = new GrpcServerAdapter(); + // Above the 5 s default: the demand tests stall on purpose before moving 4 MiB. + restletPeer.setMaxCallDuration(Duration.ofSeconds(30)); + restletPeer.getServices().add(service); + restletPeer.start(); + Router router = new Router(); + restletPeer.attachTo(router); + Context context = new Context(); + context.getParameters().set("http.transport.protocol", "HTTP2"); + http = new Server(context, Protocol.HTTP, 0, router); + http.start(); + port = http.getActualPort(); + } else { + // A fixed flow control window, so that the peer cannot absorb what the client holds. + nettyPeer = + NettyServerBuilder.forPort(0) + .flowControlWindow(65536) + .addService(service) + .build() + .start(); + port = nettyPeer.getPort(); + } + Context context = new Context(); + context.getParameters().set("httpClientTransportMode", "HTTP2"); + context.getParameters().set("followRedirects", "false"); + client = new Client(context, Protocol.HTTP); + client.start(); + adapter.setClient(client); + adapter.setTarget("dns:///localhost:" + port); + adapter.setPlaintext(true); + adapter.start(); + } + + @AfterEach + void cleanup() throws Exception { + try { + adapter.stop(); + } finally { + try { + if (client != null) { + client.stop(); + } + } finally { + if (nettyPeer != null) { + nettyPeer.shutdownNow(); + assertTrue(nettyPeer.awaitTermination(30, TimeUnit.SECONDS)); + } + if (restletPeer != null) { + restletPeer.stop(); + } + if (http != null) { + http.stop(); + } + } + } + } + + private static CallOptions deadline() { + return CallOptions.DEFAULT.withDeadlineAfter(30, TimeUnit.SECONDS); + } + + /** A service with one method of each streaming type. */ + private static ServerServiceDefinition streams() { + return ServerServiceDefinition.builder("review.Streams") + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, StreamObserver out) -> { + out.onNext(request); + out.onCompleted(); + })) + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, StreamObserver out) -> { + int count = Integer.parseInt(request.getValue()); + for (int i = 0; i < count; i++) { + out.onNext(StringValue.of(i + ":" + BLOCK)); + } + out.onCompleted(); + })) + .addMethod( + UPLOAD, + ServerCalls.asyncClientStreamingCall( + out -> + new StreamObserver() { + private int messages; + private long characters; + + @Override + public void onNext(StringValue value) { + messages++; + characters += value.getValue().length(); + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + out.onNext( + StringValue.of( + messages + ":" + characters)); + out.onCompleted(); + } + })) + .addMethod( + CHAT, + ServerCalls.asyncBidiStreamingCall( + out -> + new StreamObserver() { + @Override + public void onNext(StringValue value) { + out.onNext( + StringValue.of("re:" + value.getValue())); + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + out.onCompleted(); + } + })) + .build(); + } + + /** The producer ignores readiness, so the stream stays within the 256 KiB queue budget. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void serverStreamingPreservesOrder(boolean restlet) throws Exception { + start(restlet, streams()); + var responses = + ClientCalls.blockingServerStreamingCall( + adapter.getChannel(), WATCH, deadline(), StringValue.of("50")); + var values = new ArrayList(); + responses.forEachRemaining(value -> values.add(value.getValue())); + assertEquals(50, values.size()); + for (int i = 0; i < 50; i++) { + assertEquals(i + ":" + BLOCK, values.get(i)); + } + } + + /** The sender ignores readiness, so the stream stays within the 256 KiB queue budget. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void clientStreamingSendsEveryMessage(boolean restlet) throws Exception { + start(restlet, streams()); + var result = new CompletableFuture(); + var requests = + ClientCalls.asyncClientStreamingCall( + adapter.getChannel().newCall(UPLOAD, deadline()), single(result)); + for (int i = 0; i < 50; i++) { + requests.onNext(StringValue.of(BLOCK)); + } + requests.onCompleted(); + assertEquals("50:" + (50 * BLOCK.length()), result.get(30, TimeUnit.SECONDS)); + } + + /** Each reply is received before the next request is sent, over one HTTP/2 stream. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void bidirectionalCallsExchangeMessagesBothWays(boolean restlet) throws Exception { + start(restlet, streams()); + var replies = new LinkedBlockingQueue(); + var done = new CompletableFuture(); + var requests = + ClientCalls.asyncBidiStreamingCall( + adapter.getChannel().newCall(CHAT, deadline()), + new StreamObserver() { + @Override + public void onNext(StringValue value) { + replies.add(value.getValue()); + } + + @Override + public void onError(Throwable cause) { + done.completeExceptionally(cause); + } + + @Override + public void onCompleted() { + done.complete(null); + } + }); + for (int i = 0; i < 5; i++) { + requests.onNext(StringValue.of("ping" + i)); + assertEquals("re:ping" + i, replies.poll(30, TimeUnit.SECONDS)); + } + requests.onCompleted(); + done.get(30, TimeUnit.SECONDS); + assertTrue(replies.isEmpty()); + } + + /** A client respecting readiness stops sending while the server requests nothing. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void requestsFollowTheServerDemand(boolean restlet) throws Exception { + adapter.setMaxQueuedBytes(32768); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger received = new AtomicInteger(); + start( + restlet, + ServerServiceDefinition.builder("review.Streams") + .addMethod( + UPLOAD, + ServerCalls.asyncClientStreamingCall( + out -> { + var observer = + (ServerCallStreamObserver) out; + observer.disableAutoRequest(); + Thread.startVirtualThread( + () -> { + try { + release.await(); + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + return; + } + observer.request(Integer.MAX_VALUE); + }); + return new StreamObserver() { + @Override + public void onNext(StringValue value) { + received.incrementAndGet(); + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + out.onNext( + StringValue.of( + Integer.toString( + received.get()))); + out.onCompleted(); + } + }; + })) + .build()); + int total = 1024; // 4 MiB of 4 KiB messages + AtomicInteger sent = new AtomicInteger(); + var result = new CompletableFuture(); + ClientCall call = + adapter.getChannel().newCall(UPLOAD, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onMessage(StringValue message) { + result.complete(message.getValue()); + } + + @Override + public void onReady() { + while (call.isReady() && sent.get() < total) { + call.sendMessage(StringValue.of(BLOCK)); + if (sent.incrementAndGet() == total) { + call.halfClose(); + } + } + } + + @Override + public void onClose(Status status, Metadata trailers) { + if (!status.isOk()) { + result.completeExceptionally(status.asRuntimeException()); + } + } + }, + new Metadata()); + call.request(1); + // Let the sender fill every buffer on the way. + Thread.sleep(1000); + int stalled = sent.get(); + assertTrue(stalled < total / 2, "sent " + stalled + " of " + total + " messages"); + assertEquals(0, received.get()); + release.countDown(); + assertEquals(Integer.toString(total), result.get(30, TimeUnit.SECONDS)); + } + + /** A server respecting readiness stops sending while the client requests nothing. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void responsesFollowTheClientDemand(boolean restlet) throws Exception { + adapter.setMaxQueuedBytes(32768); + int total = 1024; + AtomicInteger produced = new AtomicInteger(); + start( + restlet, + ServerServiceDefinition.builder("review.Streams") + .addMethod( + WATCH, + ServerCalls.asyncServerStreamingCall( + (StringValue request, StreamObserver out) -> { + var observer = + (ServerCallStreamObserver) out; + observer.setOnReadyHandler( + () -> { + while (observer.isReady() + && produced.get() < total) { + produced.incrementAndGet(); + observer.onNext(StringValue.of(BLOCK)); + } + if (produced.compareAndSet( + total, total + 1)) { + observer.onCompleted(); + } + }); + })) + .build()); + AtomicInteger received = new AtomicInteger(); + var closed = new CompletableFuture(); + ClientCall call = adapter.getChannel().newCall(WATCH, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onMessage(StringValue message) { + received.incrementAndGet(); + } + + @Override + public void onClose(Status status, Metadata trailers) { + closed.complete(status); + } + }, + new Metadata()); + call.sendMessage(StringValue.of("go")); + call.halfClose(); + // Requests nothing yet: the server fills every buffer on the way, then stops. + Thread.sleep(1000); + int stalled = produced.get(); + assertTrue(stalled < total / 2, "produced " + stalled + " of " + total + " messages"); + assertEquals(0, received.get()); + call.request(Integer.MAX_VALUE); + assertEquals(Status.Code.OK, closed.get(30, TimeUnit.SECONDS).getCode()); + assertEquals(total, received.get()); + } + + /** Messages larger than the queue budget still stream through, in both directions. */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void messagesLargerThanTheQueueBudgetStreamThrough(boolean restlet) throws Exception { + adapter.setMaxQueuedBytes(16384); + start(restlet, streams()); + String large = "x".repeat(512 * 1024); + assertEquals( + large, + ClientCalls.blockingUnaryCall( + adapter.getChannel(), ECHO, deadline(), StringValue.of(large)) + .getValue()); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void cancellingAStreamReachesTheServerAndFreesTheSlot(boolean restlet) throws Exception { + adapter.setMaxConcurrentCalls(1); + CountDownLatch receiving = new CountDownLatch(1); + CountDownLatch cancelled = new CountDownLatch(1); + start( + restlet, + ServerServiceDefinition.builder("review.Streams") + .addMethod( + CHAT, + ServerCalls.asyncBidiStreamingCall( + out -> { + ((ServerCallStreamObserver) out) + .setOnCancelHandler(cancelled::countDown); + return new StreamObserver() { + @Override + public void onNext(StringValue value) { + receiving.countDown(); + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + out.onCompleted(); + } + }; + })) + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, StreamObserver out) -> { + out.onNext(request); + out.onCompleted(); + })) + .build()); + var closed = new CompletableFuture(); + ClientCall call = adapter.getChannel().newCall(CHAT, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onClose(Status status, Metadata trailers) { + closed.complete(status); + } + }, + new Metadata()); + call.sendMessage(StringValue.of("open")); + assertTrue(receiving.await(30, TimeUnit.SECONDS)); + call.cancel("client gave up", null); + assertEquals(Status.Code.CANCELLED, closed.get(30, TimeUnit.SECONDS).getCode()); + assertTrue(cancelled.await(30, TimeUnit.SECONDS)); + assertEquals( + "next", + ClientCalls.blockingUnaryCall( + adapter.getChannel(), ECHO, deadline(), StringValue.of("next")) + .getValue()); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void anOpenStreamEndsAtItsDeadline(boolean restlet) throws Exception { + start(restlet, streams()); + var requests = new CompletableFuture>(); + var error = new CompletableFuture(); + requests.complete( + ClientCalls.asyncBidiStreamingCall( + adapter.getChannel() + .newCall( + CHAT, + CallOptions.DEFAULT.withDeadlineAfter( + 500, TimeUnit.MILLISECONDS)), + new StreamObserver() { + @Override + public void onNext(StringValue value) { + // Replies are not checked here. + } + + @Override + public void onError(Throwable cause) { + error.complete(cause); + } + + @Override + public void onCompleted() { + error.complete(null); + } + })); + // Never half-closed: only the deadline ends the call. + requests.get().onNext(StringValue.of("open")); + Throwable cause = error.get(30, TimeUnit.SECONDS); + assertTrue(cause instanceof StatusRuntimeException, String.valueOf(cause)); + assertEquals( + Status.Code.DEADLINE_EXCEEDED, + ((StatusRuntimeException) cause).getStatus().getCode()); + } + + private static StreamObserver single(CompletableFuture result) { + return new StreamObserver<>() { + @Override + public void onNext(StringValue value) { + result.complete(value.getValue()); + } + + @Override + public void onError(Throwable cause) { + result.completeExceptionally(cause); + } + + @Override + public void onCompleted() { + // The single reply already completed the result. + } + }; + } +} diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java index da4d3b61b4..67702809c4 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcServerAdapterTestCase.java @@ -509,6 +509,8 @@ public void onCompleted() { @Test void unrequestedMessagesPushBackOnTheClient() throws Exception { adapter.setMaxQueuedBytes(32768); + // Above the 5 s default: the test stalls on purpose before moving 4 MiB. + adapter.setMaxCallDuration(Duration.ofSeconds(30)); CountDownLatch release = new CountDownLatch(1); AtomicInteger received = new AtomicInteger(); start( diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java index e6b3ba14e4..900778a571 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/internal/RestletClientStreamTestCase.java @@ -36,7 +36,8 @@ class RestletClientStreamTestCase { @Test void cancelledTimersDoNotStayQueued() { - var factory = new RestletTransportFactory(null, 1024, 1, java.time.Duration.ofSeconds(1)); + var factory = + new RestletTransportFactory(null, 1024, 16384, 1, java.time.Duration.ofSeconds(1)); try { var timer = (ScheduledThreadPoolExecutor) factory.getScheduledExecutorService(); for (int i = 0; i < 100; i++) { @@ -79,6 +80,7 @@ private static RestletClientStream stream(ExecutorService io, AtomicInteger rele CallOptions.DEFAULT, new ClientStreamTracer[0], 128, + 16384, io, ignored -> released.incrementAndGet()); } diff --git a/readme.md b/readme.md index 6e0c91d525..71d3caf222 100644 --- a/readme.md +++ b/readme.md @@ -28,11 +28,11 @@ Restlet 3.0 lets the same application and the same connectors serve APIs in seve | Style | Extension | Status | |---|---|---| | [GraphQL](https://graphql.org/) | `org.restlet.ext.graphql` | Experimental: server and client on graphql-java, with subscriptions over SSE and WebSocket | -| [gRPC](https://grpc.io/) | `org.restlet.ext.grpc` | Experimental: services of every streaming type, and unary client, over HTTP/2 | +| [gRPC](https://grpc.io/) | `org.restlet.ext.grpc` | Experimental: services and clients of every streaming type, over HTTP/2 | | [MCP](https://modelcontextprotocol.io/) | `org.restlet.ext.mcp` | Experimental: tools, resources and prompts for AI agents on the official MCP Java SDK, over stateless Streamable HTTP and STDIO | | [Smithy](https://smithy.io/) | `org.restlet.ext.smithy` | smithy-java clients on Restlet HTTP/1.1 and HTTP/2 connectors, discovered or set explicitly; experimental server hosting with the `org.restlet.ext.smithy.server` artifact, on the class path | -Upcoming enhancements include streaming in the gRPC client, and MCP sessions and resumable streams. Feedback is welcome in [Discussions](https://github.com/restlet/restlet-framework-java/discussions). +Upcoming enhancements include MCP sessions and resumable streams. Feedback is welcome in [Discussions](https://github.com/restlet/restlet-framework-java/discussions). ## Learn more From 51a4a7f1f713fdd991d88552082641e00dce45f1 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 5 Nov 2026 09:42:06 -0500 Subject: [PATCH 2/6] Split the gRPC client stream's send and receive paths, and fix a test Move the request queueing out of the stream sink into begin, send and enqueue, and the exchange's request building and response handling into newRequest and receive, to bring their complexity within SonarCloud's limit; behaviour is unchanged. Use unnamed catch variables. Responses now stream as they arrive, so the one message a server sends before waiting may reach the application before a cancellation, as with grpc-netty: cancellationAfterHeadersReleasesAdmissionAndClosesExactlyOnce now allows at most one message instead of none. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../grpc/internal/RestletClientStream.java | 225 ++++++++++-------- .../ext/grpc/GrpcClientAdapterTestCase.java | 4 +- 2 files changed, 129 insertions(+), 100 deletions(-) diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java index 56bbbaa08b..32dd993119 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java @@ -190,58 +190,12 @@ protected Sink abstractClientStreamSink() { return new Sink() { @Override public void writeHeaders(Metadata metadata, byte[] payload) { - headers = metadata; - stats.clientOutboundHeaders(); - serial.execute(state::allocated); - Status failure = null; - synchronized (outbound) { - if (!done && !started) { - started = true; - try { - io.execute(RestletClientStream.this::exchange); - } catch (RejectedExecutionException e) { - failure = Status.UNAVAILABLE.withCause(e); - } - } - } - if (failure != null) { - fail(failure); - } + begin(metadata); } @Override public void writeFrame(WritableBuffer frame, boolean end, boolean flush, int messages) { - Status failure = null; - synchronized (outbound) { - if (frame instanceof Buffer buffer && !done && !halfClosed) { - int size = buffer.size; - // One message is framed whole, possibly before the writer starts: while - // the queue holds only that message, it may grow up to one framed - // message. A frame counting messages starts a new one. - int message = messages > 0 ? size : open + size; - boolean others = queued + size > message; - if (size > queueBytes - queued && (others || message > wireLimit)) { - failure = - Status.RESOURCE_EXHAUSTED.withDescription( - "Request queue byte budget exceeded"); - } else if (size > 0) { - open = message; - outbound.addLast(Arrays.copyOf(buffer.data, size)); - queued += size; - onSendingBytes(size); - } - } - if (frame != null) { - frame.release(); - } - if (end) { - halfClosed = true; - } - outbound.notifyAll(); - } - if (failure != null) { - fail(failure); - } + send(frame, end, messages); } @Override @@ -251,6 +205,71 @@ public void cancel(Status status) { }; } + /** Starts the HTTP/2 exchange with the request headers. */ + private void begin(Metadata metadata) { + headers = metadata; + stats.clientOutboundHeaders(); + serial.execute(state::allocated); + Status failure = null; + synchronized (outbound) { + if (!done && !started) { + started = true; + try { + io.execute(this::exchange); + } catch (RejectedExecutionException e) { + failure = Status.UNAVAILABLE.withCause(e); + } + } + } + if (failure != null) { + fail(failure); + } + } + + /** Queues a request frame for the writer, and records the half-close. */ + private void send(WritableBuffer frame, boolean end, int messages) { + Status failure = null; + synchronized (outbound) { + if (frame instanceof Buffer buffer && !done && !halfClosed) { + failure = enqueue(buffer, messages); + } + if (frame != null) { + frame.release(); + } + if (end) { + halfClosed = true; + } + outbound.notifyAll(); + } + if (failure != null) { + fail(failure); + } + } + + /** + * Copies a frame into the queue, within its budget; must be called holding {@link #outbound}. + * One message is framed whole, possibly before the writer starts: while the queue holds only + * that message, it may grow up to one framed message. A frame counting messages starts a new + * one. + * + * @return The failure if the budget is exceeded, null otherwise. + */ + private Status enqueue(Buffer buffer, int messages) { + int size = buffer.size; + int message = messages > 0 ? size : open + size; + boolean others = queued + size > message; + if (size > queueBytes - queued && (others || message > wireLimit)) { + return Status.RESOURCE_EXHAUSTED.withDescription("Request queue byte budget exceeded"); + } + if (size > 0) { + open = message; + outbound.addLast(Arrays.copyOf(buffer.data, size)); + queued += size; + onSendingBytes(size); + } + return null; + } + private void fail(Status status) { failed = true; Deadline current = deadline; @@ -299,22 +318,7 @@ private void exchange() { if (done) { return; } - Request call = new Request(Method.POST, target, requestEntity()); - // Resource origin selects the resolved peer; base reference selects HTTP authority. - call.getResourceRef().setBaseRef(new Reference("http://" + authority)); - call.getHeaders().set("te", "trailers"); - byte[][] metadata = TransportFrameUtil.toHttp2Headers(headers); - for (int i = 0; i < metadata.length; i += 2) { - String key = new String(metadata[i], StandardCharsets.US_ASCII); - if (!key.startsWith(":") - && !"content-type".equals(key) - && !"user-agent".equals(key) - && !"te".equals(key)) { - call.getHeaders() - .add(key, new String(metadata[i + 1], StandardCharsets.US_ASCII)); - } - } - call.getClientInfo().setAgent("restlet-grpc/1.85.1"); + Request call = newRequest(); request = call; if (done || failed || Thread.currentThread().isInterrupted()) { call.abort(); @@ -329,40 +333,9 @@ private void exchange() { .withCause(cause))); // Returns at the response head; the request body is still written meanwhile. response = client.handle(call); - if (done) { - return; - } - if (response.getProtocol() == null - || !"2.0".equals(response.getProtocol().getVersion())) { - fail( - Status.UNAVAILABLE - .withDescription("gRPC requires an HTTP/2 response") - .withCause(response.getStatus().getThrowable())); - return; - } - Metadata initial = metadata(response.getHeaders()); - initial.put( - InternalMetadata.keyOf(":status", Metadata.ASCII_STRING_MARSHALLER), - Integer.toString(response.getStatus().getCode())); - if (response.getEntity() != null && response.getEntity().getMediaType() != null) { - initial.put( - Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), - response.getEntity().getMediaType().toString()); - } - if (response.getHeaders().getFirstValue("grpc-status", true) != null) { - serial.execute(() -> state.trailers(initial)); - return; - } - serial.execute(() -> state.headers(initial)); - if (response.getEntity() != null) { - try (var input = response.getEntity().getStream()) { - if (input != null && !readAll(input)) { - return; - } - } + if (!done) { + receive(response); } - Metadata trailers = metadata(response.getTrailers()); - serial.execute(() -> state.trailers(trailers)); } catch (IOException | RuntimeException e) { fail(Status.UNAVAILABLE.withDescription("Restlet gRPC exchange failed").withCause(e)); } finally { @@ -373,6 +346,60 @@ private void exchange() { } } + /** Returns the HTTP/2 request of the call, with the gRPC metadata as headers. */ + private Request newRequest() { + Request call = new Request(Method.POST, target, requestEntity()); + // Resource origin selects the resolved peer; base reference selects HTTP authority. + call.getResourceRef().setBaseRef(new Reference("http://" + authority)); + call.getHeaders().set("te", "trailers"); + byte[][] metadata = TransportFrameUtil.toHttp2Headers(headers); + for (int i = 0; i < metadata.length; i += 2) { + String key = new String(metadata[i], StandardCharsets.US_ASCII); + if (!key.startsWith(":") + && !"content-type".equals(key) + && !"user-agent".equals(key) + && !"te".equals(key)) { + call.getHeaders().add(key, new String(metadata[i + 1], StandardCharsets.US_ASCII)); + } + } + call.getClientInfo().setAgent("restlet-grpc/1.85.1"); + return call; + } + + /** Hands the response head, body and trailers to the SDK. */ + private void receive(Response response) throws IOException { + if (response.getProtocol() == null || !"2.0".equals(response.getProtocol().getVersion())) { + fail( + Status.UNAVAILABLE + .withDescription("gRPC requires an HTTP/2 response") + .withCause(response.getStatus().getThrowable())); + return; + } + Metadata initial = metadata(response.getHeaders()); + initial.put( + InternalMetadata.keyOf(":status", Metadata.ASCII_STRING_MARSHALLER), + Integer.toString(response.getStatus().getCode())); + if (response.getEntity() != null && response.getEntity().getMediaType() != null) { + initial.put( + Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), + response.getEntity().getMediaType().toString()); + } + if (response.getHeaders().getFirstValue("grpc-status", true) != null) { + serial.execute(() -> state.trailers(initial)); + return; + } + serial.execute(() -> state.headers(initial)); + if (response.getEntity() != null) { + try (var input = response.getEntity().getStream()) { + if (input != null && !readAll(input)) { + return; + } + } + } + Metadata trailers = metadata(response.getTrailers()); + serial.execute(() -> state.trailers(trailers)); + } + /** * Hands the response body to the SDK as it arrives, within the queue budget. * @@ -387,7 +414,7 @@ private boolean readAll(InputStream input) throws IOException { while (unread >= queueBytes && !done && !failed) { inbound.wait(); } - } catch (InterruptedException e) { + } catch (InterruptedException _) { Thread.currentThread().interrupt(); throw new InterruptedIOException("Interrupted reading the gRPC response"); } @@ -422,7 +449,7 @@ public void write(OutputStream output) throws IOException { while (outbound.isEmpty() && !halfClosed && !done && !failed) { outbound.wait(); } - } catch (InterruptedException e) { + } catch (InterruptedException _) { Thread.currentThread().interrupt(); throw new InterruptedIOException( "Interrupted writing the gRPC request"); diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java index 504ce74be0..da21769dd8 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientAdapterTestCase.java @@ -373,7 +373,9 @@ public void onClose(Status value, Metadata trailers) { assertEquals(Status.Code.CANCELLED, status.get().getCode()); assertEquals("recovered", call("recovered").getValue()); assertEquals(1, terminals.get()); - assertEquals(0, messages.get()); + // Responses stream as they arrive, so the one message sent before the server waited may + // be delivered before the cancellation, as with grpc-netty; never more. + assertTrue(messages.get() <= 1, "messages: " + messages.get()); } private record Pending(CountDownLatch entered, CountDownLatch cancelled) {} From 96b7b23b86063abc05ce06178909bd57f9330e15 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 5 Nov 2026 12:37:41 -0500 Subject: [PATCH 3/6] Keep early gRPC answers, and accept large messages sent on readiness A server may answer and reset the stream before the client half-closes, while the request is still written. The disconnect then failed calls whose whole response had arrived, and a writer held by flow control kept the exchange after the call released its slot. A disconnect now only wakes the reader and the writer: a complete response is still read, and a broken one fails the head or the body read. When the call ends with its request unfinished, the queued frames are dropped and the HTTP request is aborted, resetting the stream and freeing the writer. The queue rejected the next large message of a sender respecting readiness while the tail of the previous one was still queued. A message may now start while the queue is within its budget, and grow up to the message-size limit once started, so the queue stays within its budget plus one message. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../grpc/internal/RestletClientStream.java | 79 ++++++++--- .../ext/grpc/GrpcClientStreamingTestCase.java | 127 ++++++++++++++++++ 2 files changed, 189 insertions(+), 17 deletions(-) diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java index 32dd993119..ed0d34fb2a 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java @@ -85,6 +85,13 @@ final class RestletClientStream extends AbstractClientStream { private volatile Deadline deadline; private volatile boolean done; private volatile boolean failed; + + /** Whether the peer reset the stream or the connection failed. */ + private volatile boolean peerGone; + + /** Whether the request body ended, normally or not; guarded by {@link #outbound}. */ + private boolean requestEnded; + private boolean started; /** Request bytes queued; guarded by {@link #outbound}. */ @@ -93,6 +100,9 @@ final class RestletClientStream extends AbstractClientStream { /** Bytes framed for the latest request message, from the frame that started it. */ private int open; + /** Request bytes queued when the latest message started; guarded by {@link #outbound}. */ + private int messageStart; + /** Whether the SDK half-closed the request; guarded by {@link #outbound}. */ private boolean halfClosed; @@ -247,18 +257,22 @@ private void send(WritableBuffer frame, boolean end, int messages) { } /** - * Copies a frame into the queue, within its budget; must be called holding {@link #outbound}. - * One message is framed whole, possibly before the writer starts: while the queue holds only - * that message, it may grow up to one framed message. A frame counting messages starts a new - * one. + * Copies a frame into the queue, within its budget; must be called holding {@link #outbound}. A + * message may only start while the queue is within its budget, as it is for a sender that + * respects readiness; once started, it may grow up to the message-size limit, whatever the tail + * of the previous message still queued. The queue so stays within its budget plus one message. + * A frame counting messages starts a new one. * * @return The failure if the budget is exceeded, null otherwise. */ private Status enqueue(Buffer buffer, int messages) { int size = buffer.size; - int message = messages > 0 ? size : open + size; - boolean others = queued + size > message; - if (size > queueBytes - queued && (others || message > wireLimit)) { + if (messages > 0) { + messageStart = queued; + open = 0; + } + int message = open + size; + if (messageStart > queueBytes || message > wireLimit) { return Status.RESOURCE_EXHAUSTED.withDescription("Request queue byte budget exceeded"); } if (size > 0) { @@ -301,6 +315,26 @@ private void fail(Status status) { } } + /** + * Ends a request still being written when the call ended, as when a server answers before the + * half-close: its queued frames are dropped and the HTTP/2 stream is reset, which also releases + * a writer blocked by flow control. The call's own status is already reported. + */ + private void endRequest() { + boolean unfinished; + synchronized (outbound) { + unfinished = started && !requestEnded; + outbound.clear(); + queued = 0; + requestEnded = true; + outbound.notifyAll(); + } + Request active = request; + if (unfinished && active != null) { + active.abort(); + } + } + /** Wakes the request writer and the response reader, once the call failed or ended. */ private void wake() { synchronized (outbound) { @@ -324,13 +358,15 @@ private void exchange() { call.abort(); return; } + // A disconnect only wakes the reader and the writer: a peer that reset the stream + // after its whole response, as a server finishing before the half-close may, leaves + // that response to be read; one that did not fails the head or the body read. call.getDisconnectSignal() .thenAccept( - cause -> - fail( - Status.UNAVAILABLE - .withDescription("gRPC peer disconnected") - .withCause(cause))); + cause -> { + peerGone = true; + wake(); + }); // Returns at the response head; the request body is still written meanwhile. response = client.handle(call); if (!done) { @@ -411,7 +447,8 @@ private boolean readAll(InputStream input) throws IOException { int room; synchronized (inbound) { try { - while (unread >= queueBytes && !done && !failed) { + // Once the peer is gone, what the connector received is read regardless. + while (unread >= queueBytes && !done && !failed && !peerGone) { inbound.wait(); } } catch (InterruptedException _) { @@ -421,7 +458,7 @@ private boolean readAll(InputStream input) throws IOException { if (done || failed) { return false; } - room = (int) Math.min(chunk.length, queueBytes - unread); + room = (int) Math.max(1, Math.min(chunk.length, queueBytes - unread)); } int count = input.read(chunk, 0, room); if (count < 0) { @@ -446,20 +483,27 @@ public void write(OutputStream output) throws IOException { byte[] data; synchronized (outbound) { try { - while (outbound.isEmpty() && !halfClosed && !done && !failed) { + while (outbound.isEmpty() + && !halfClosed + && !done + && !failed + && !peerGone) { outbound.wait(); } } catch (InterruptedException _) { + requestEnded = true; Thread.currentThread().interrupt(); throw new InterruptedIOException( "Interrupted writing the gRPC request"); } - if (failed) { - throw new IOException("gRPC call cancelled"); + if (failed || peerGone) { + requestEnded = true; + throw new IOException("gRPC request not completed"); } data = outbound.pollFirst(); if (data == null) { // Half-closed, or the call ended first: the request body ends. + requestEnded = true; return; } queued -= data.length; @@ -572,6 +616,7 @@ public void deframerClosed(boolean partial) { if (!done) { done = true; wake(); + endRequest(); finished.accept(RestletClientStream.this); } super.deframerClosed(partial); diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java index eb4d91be66..080d865a38 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java @@ -540,6 +540,133 @@ public void onCompleted() { ((StatusRuntimeException) cause).getStatus().getCode()); } + /** + * A server may answer before the client half-closes, while the request is still written and + * held by flow control: the client gets the answer and its status, and the unfinished request + * is reset, so the only call slot is free at once. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void anEarlyAnswerEndsTheCallWhileTheRequestIsStillSent(boolean restlet) throws Exception { + adapter.setMaxConcurrentCalls(1); + start( + restlet, + ServerServiceDefinition.builder("review.Streams") + .addMethod( + UPLOAD, + ServerCalls.asyncClientStreamingCall( + out -> + new StreamObserver() { + private boolean answered; + + @Override + public void onNext(StringValue value) { + if (!answered) { + answered = true; + out.onNext(StringValue.of("enough")); + out.onCompleted(); + } + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + // Already answered. + } + })) + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, StreamObserver out) -> { + out.onNext(request); + out.onCompleted(); + })) + .build()); + var reply = new CompletableFuture(); + var closed = new CompletableFuture(); + AtomicInteger sent = new AtomicInteger(); + ClientCall call = + adapter.getChannel().newCall(UPLOAD, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onMessage(StringValue message) { + reply.complete(message.getValue()); + } + + @Override + public void onReady() { + // Never half-closes: keeps sending while ready, up to 4 MiB. + while (call.isReady() && sent.get() < 1024 && !closed.isDone()) { + call.sendMessage(StringValue.of(BLOCK)); + sent.incrementAndGet(); + } + } + + @Override + public void onClose(Status status, Metadata trailers) { + closed.complete(status); + } + }, + new Metadata()); + call.request(1); + assertEquals(Status.Code.OK, closed.get(30, TimeUnit.SECONDS).getCode()); + assertEquals("enough", reply.get(30, TimeUnit.SECONDS)); + assertEquals( + "next", + ClientCalls.blockingUnaryCall( + adapter.getChannel(), ECHO, deadline(), StringValue.of("next")) + .getValue()); + } + + /** + * With a direct callback executor, readiness may return while the tail of a large message is + * still queued: a sender respecting it may start the next large message. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void largeMessagesSentOnReadinessAreAccepted(boolean restlet) throws Exception { + adapter.setMaxQueuedBytes(16384); + adapter.setExecutor(Runnable::run); + start(restlet, streams()); + String large = "y".repeat(100 * 1024); + var result = new CompletableFuture(); + AtomicInteger sent = new AtomicInteger(); + ClientCall call = + adapter.getChannel().newCall(UPLOAD, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onMessage(StringValue message) { + result.complete(message.getValue()); + } + + @Override + public void onReady() { + while (call.isReady() && sent.get() < 2) { + call.sendMessage(StringValue.of(large)); + if (sent.incrementAndGet() == 2) { + call.halfClose(); + } + } + } + + @Override + public void onClose(Status status, Metadata trailers) { + if (!status.isOk()) { + result.completeExceptionally(status.asRuntimeException()); + } + } + }, + new Metadata()); + call.request(1); + assertEquals("2:" + (2 * large.length()), result.get(30, TimeUnit.SECONDS)); + } + private static StreamObserver single(CompletableFuture result) { return new StreamObserver<>() { @Override From c61427da0fcf565245fb7e7dcb896a808431720c Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:59:32 -0500 Subject: [PATCH 4/6] Test the gRPC request queue budget, and drain whole chunks once the peer is gone A sender ignoring readiness towards a peer that requests nothing is now tested against both peers: the call ends once with RESOURCE_EXHAUSTED and frees its only slot. Once the peer is gone, the response the connector buffered was read one byte at a time while the inbound budget was full; it is now drained in whole chunks. Co-Authored-By: Claude Opus 5.5 --- .../grpc/internal/RestletClientStream.java | 4 +- .../ext/grpc/GrpcClientStreamingTestCase.java | 71 +++++++++++++++++++ 2 files changed, 74 insertions(+), 1 deletion(-) diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java index ed0d34fb2a..91326dcb5e 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java @@ -458,7 +458,9 @@ private boolean readAll(InputStream input) throws IOException { if (done || failed) { return false; } - room = (int) Math.max(1, Math.min(chunk.length, queueBytes - unread)); + // A live peer is read within the budget; once it is gone, what the connector + // buffered is drained in whole chunks. + room = peerGone ? chunk.length : (int) Math.min(chunk.length, queueBytes - unread); } int count = input.read(chunk, 0, room); if (count < 0) { diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java index 080d865a38..8182dc89b7 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java @@ -667,6 +667,77 @@ public void onClose(Status status, Metadata trailers) { assertEquals("2:" + (2 * large.length()), result.get(30, TimeUnit.SECONDS)); } + /** + * A sender ignoring readiness towards a peer that requests nothing fills the request queue: the + * call ends once with RESOURCE_EXHAUSTED, the stream is reset and the only call slot is free. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void aSenderIgnoringReadinessFailsAtTheQueueBudget(boolean restlet) throws Exception { + adapter.setMaxConcurrentCalls(1); + adapter.setMaxQueuedBytes(16384); + start( + restlet, + ServerServiceDefinition.builder("review.Streams") + .addMethod( + CHAT, + ServerCalls.asyncBidiStreamingCall( + out -> { + // A stalled peer: it never asks for a message. + ((ServerCallStreamObserver) out) + .disableAutoRequest(); + return new StreamObserver() { + @Override + public void onNext(StringValue value) { + // Never requested. + } + + @Override + public void onError(Throwable cause) { + // The call ended; nothing to answer. + } + + @Override + public void onCompleted() { + out.onCompleted(); + } + }; + })) + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, StreamObserver out) -> { + out.onNext(request); + out.onCompleted(); + })) + .build()); + var closed = new CompletableFuture(); + AtomicInteger closes = new AtomicInteger(); + ClientCall call = adapter.getChannel().newCall(CHAT, deadline()); + call.start( + new ClientCall.Listener<>() { + @Override + public void onClose(Status status, Metadata trailers) { + closes.incrementAndGet(); + closed.complete(status); + } + }, + new Metadata()); + // Up to 4 MiB, far beyond what the peer, the connectors and the queue may hold. + for (int i = 0; i < 1024 && !closed.isDone(); i++) { + call.sendMessage(StringValue.of(BLOCK)); + } + Status status = closed.get(30, TimeUnit.SECONDS); + assertEquals(Status.Code.RESOURCE_EXHAUSTED, status.getCode()); + assertEquals("Request queue byte budget exceeded", status.getDescription()); + assertEquals( + "next", + ClientCalls.blockingUnaryCall( + adapter.getChannel(), ECHO, deadline(), StringValue.of("next")) + .getValue()); + assertEquals(1, closes.get()); + } + private static StreamObserver single(CompletableFuture result) { return new StreamObserver<>() { @Override From 494bc9f3c6fe9c12b0c289f7f3b51d1c85ff4c3c Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:38:41 -0500 Subject: [PATCH 5/6] Delimit gRPC request messages by their prefixes, and abort every ended exchange grpc-java frames a message of unknown length before counting it, so frame message counts do not mark where messages start: a single large message could be refused with RESOURCE_EXHAUSTED before the writer drained it. The queue now reads the 5-byte gRPC prefixes to find each message start and its declared length. The entity writer may return before the connector sends the last content, so the exchange is now aborted whenever the call ends; the connector ignores the abort of a completed exchange. The queue wait moves out of the entity class, and the stalled-sender tests wait for their counters to settle instead of sleeping. Co-Authored-By: Claude Opus 5.5 --- .../grpc/internal/RestletClientStream.java | 162 ++++++++++-------- .../ext/grpc/GrpcClientStreamingTestCase.java | 63 ++++++- 2 files changed, 149 insertions(+), 76 deletions(-) diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java index 91326dcb5e..ab928480eb 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java @@ -67,7 +67,7 @@ final class RestletClientStream extends AbstractClientStream { private final Client client; private final String target; private final String authority; - private final int wireLimit; + private final int limit; private final int queueBytes; private final ExecutorService io; private final Consumer finished; @@ -89,19 +89,19 @@ final class RestletClientStream extends AbstractClientStream { /** Whether the peer reset the stream or the connection failed. */ private volatile boolean peerGone; - /** Whether the request body ended, normally or not; guarded by {@link #outbound}. */ - private boolean requestEnded; - private boolean started; /** Request bytes queued; guarded by {@link #outbound}. */ private int queued; - /** Bytes framed for the latest request message, from the frame that started it. */ - private int open; + /** Bytes of the latest request message prefix seen, up to 5; guarded by {@link #outbound}. */ + private int prefix; + + /** Message length read from the prefix so far; guarded by {@link #outbound}. */ + private long declared; - /** Request bytes queued when the latest message started; guarded by {@link #outbound}. */ - private int messageStart; + /** Bytes of the latest request message still to come; guarded by {@link #outbound}. */ + private long remaining; /** Whether the SDK half-closed the request; guarded by {@link #outbound}. */ private boolean halfClosed; @@ -163,7 +163,7 @@ private RestletClientStream( ip = "[" + ip + "]"; } target = "http://" + ip + ":" + address.getPort() + "/" + method.getFullMethodName(); - wireLimit = limit + 5; + this.limit = limit; this.queueBytes = queueBytes; this.io = io; this.finished = finished; @@ -205,7 +205,7 @@ public void writeHeaders(Metadata metadata, byte[] payload) { @Override public void writeFrame(WritableBuffer frame, boolean end, boolean flush, int messages) { - send(frame, end, messages); + send(frame, end); } @Override @@ -237,11 +237,11 @@ private void begin(Metadata metadata) { } /** Queues a request frame for the writer, and records the half-close. */ - private void send(WritableBuffer frame, boolean end, int messages) { + private void send(WritableBuffer frame, boolean end) { Status failure = null; synchronized (outbound) { if (frame instanceof Buffer buffer && !done && !halfClosed) { - failure = enqueue(buffer, messages); + failure = enqueue(buffer); } if (frame != null) { frame.release(); @@ -261,22 +261,24 @@ private void send(WritableBuffer frame, boolean end, int messages) { * message may only start while the queue is within its budget, as it is for a sender that * respects readiness; once started, it may grow up to the message-size limit, whatever the tail * of the previous message still queued. The queue so stays within its budget plus one message. - * A frame counting messages starts a new one. + * Messages are delimited by their gRPC prefixes, since frames do not mark where they start: a + * message of unknown length is framed before it is counted. * * @return The failure if the budget is exceeded, null otherwise. */ - private Status enqueue(Buffer buffer, int messages) { + private Status enqueue(Buffer buffer) { int size = buffer.size; - if (messages > 0) { - messageStart = queued; - open = 0; - } - int message = open + size; - if (messageStart > queueBytes || message > wireLimit) { - return Status.RESOURCE_EXHAUSTED.withDescription("Request queue byte budget exceeded"); + for (int i = 0; i < size; ) { + if (remaining > 0) { + int skipped = (int) Math.min(remaining, size - i); + remaining -= skipped; + i += skipped; + } else if ((prefix == 0 && queued + i > queueBytes) || !prefixed(buffer.data[i++])) { + return Status.RESOURCE_EXHAUSTED.withDescription( + "Request queue byte budget exceeded"); + } } if (size > 0) { - open = message; outbound.addLast(Arrays.copyOf(buffer.data, size)); queued += size; onSendingBytes(size); @@ -284,6 +286,24 @@ private Status enqueue(Buffer buffer, int messages) { return null; } + /** + * Reads a byte of a message prefix: a compression flag, then a 4-byte big-endian length. + * + * @return False if the message exceeds the size limit. + */ + private boolean prefixed(byte value) { + if (prefix > 0) { + declared = (declared << 8) | (value & 0xFF); + } + if (++prefix < 5) { + return true; + } + remaining = declared; + prefix = 0; + declared = 0; + return remaining <= limit; + } + private void fail(Status status) { failed = true; Deadline current = deadline; @@ -315,26 +335,6 @@ private void fail(Status status) { } } - /** - * Ends a request still being written when the call ended, as when a server answers before the - * half-close: its queued frames are dropped and the HTTP/2 stream is reset, which also releases - * a writer blocked by flow control. The call's own status is already reported. - */ - private void endRequest() { - boolean unfinished; - synchronized (outbound) { - unfinished = started && !requestEnded; - outbound.clear(); - queued = 0; - requestEnded = true; - outbound.notifyAll(); - } - Request active = request; - if (unfinished && active != null) { - active.abort(); - } - } - /** Wakes the request writer and the response reader, once the call failed or ended. */ private void wake() { synchronized (outbound) { @@ -481,35 +481,7 @@ private Representation requestEntity() { return new OutputRepresentation(GRPC, Representation.UNKNOWN_SIZE) { @Override public void write(OutputStream output) throws IOException { - for (; ; ) { - byte[] data; - synchronized (outbound) { - try { - while (outbound.isEmpty() - && !halfClosed - && !done - && !failed - && !peerGone) { - outbound.wait(); - } - } catch (InterruptedException _) { - requestEnded = true; - Thread.currentThread().interrupt(); - throw new InterruptedIOException( - "Interrupted writing the gRPC request"); - } - if (failed || peerGone) { - requestEnded = true; - throw new IOException("gRPC request not completed"); - } - data = outbound.pollFirst(); - if (data == null) { - // Half-closed, or the call ended first: the request body ends. - requestEnded = true; - return; - } - queued -= data.length; - } + for (byte[] data = nextFrame(); data != null; data = nextFrame()) { output.write(data); output.flush(); int sent = data.length; @@ -519,6 +491,33 @@ public void write(OutputStream output) throws IOException { }; } + /** + * Waits for the next request frame to write. + * + * @return The frame, or null once the request ends normally: half-closed, or the call ended. + * @throws IOException If the call failed or the peer is gone, so the request is not complete. + */ + private byte[] nextFrame() throws IOException { + synchronized (outbound) { + try { + while (outbound.isEmpty() && !halfClosed && !done && !failed && !peerGone) { + outbound.wait(); + } + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); + throw new InterruptedIOException("Interrupted writing the gRPC request"); + } + if (failed || peerGone) { + throw new IOException("gRPC request not completed"); + } + byte[] data = outbound.pollFirst(); + if (data != null) { + queued -= data.length; + } + return data; + } + } + private static Metadata metadata(Series

headers) { var values = new ArrayList(); for (Header header : headers) { @@ -623,6 +622,27 @@ public void deframerClosed(boolean partial) { } super.deframerClosed(partial); } + + /** + * Ends the request once the call ended, as when a server answers before the half-close: its + * queued frames are dropped and the exchange is aborted, which resets an HTTP/2 stream + * whose request is unfinished, even once the entity writer returned, and releases a writer + * blocked by flow control. The connector ignores the abort of a completed exchange, and the + * call's own status is already reported. + */ + private void endRequest() { + boolean begun; + synchronized (outbound) { + begun = started; + outbound.clear(); + queued = 0; + outbound.notifyAll(); + } + Request active = request; + if (begun && active != null) { + active.abort(); + } + } } private static final class Buffer implements WritableBuffer { diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java index 8182dc89b7..ab1a1533f4 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcClientStreamingTestCase.java @@ -25,11 +25,14 @@ import io.grpc.stub.ServerCallStreamObserver; import io.grpc.stub.ServerCalls; import io.grpc.stub.StreamObserver; +import java.io.FilterInputStream; +import java.io.InputStream; import java.time.Duration; import java.util.ArrayList; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.AfterEach; @@ -68,6 +71,25 @@ class GrpcClientStreamingTestCase { .setType(MethodDescriptor.MethodType.BIDI_STREAMING) .setFullMethodName("review.Streams/Chat") .build(); + + /** Upload, with request messages of unknown length, which grpc-java frames before counting. */ + private static final MethodDescriptor UNKNOWN_LENGTH_UPLOAD = + UPLOAD.toBuilder( + new MethodDescriptor.Marshaller() { + @Override + public InputStream stream(StringValue value) { + return new FilterInputStream( + ECHO.getRequestMarshaller().stream(value)) {}; + } + + @Override + public StringValue parse(InputStream stream) { + return ECHO.getRequestMarshaller().parse(stream); + } + }, + ECHO.getResponseMarshaller()) + .build(); + private static final String BLOCK = "x".repeat(4096); private final GrpcClientAdapter adapter = new GrpcClientAdapter(); @@ -360,8 +382,7 @@ public void onClose(Status status, Metadata trailers) { new Metadata()); call.request(1); // Let the sender fill every buffer on the way. - Thread.sleep(1000); - int stalled = sent.get(); + int stalled = settled(sent); assertTrue(stalled < total / 2, "sent " + stalled + " of " + total + " messages"); assertEquals(0, received.get()); release.countDown(); @@ -417,8 +438,7 @@ public void onClose(Status status, Metadata trailers) { call.sendMessage(StringValue.of("go")); call.halfClose(); // Requests nothing yet: the server fills every buffer on the way, then stops. - Thread.sleep(1000); - int stalled = produced.get(); + int stalled = settled(produced); assertTrue(stalled < total / 2, "produced " + stalled + " of " + total + " messages"); assertEquals(0, received.get()); call.request(Integer.MAX_VALUE); @@ -630,6 +650,21 @@ public void onClose(Status status, Metadata trailers) { @ParameterizedTest @ValueSource(booleans = {false, true}) void largeMessagesSentOnReadinessAreAccepted(boolean restlet) throws Exception { + assertLargeMessagesSentOnReadinessAreAccepted(restlet, UPLOAD); + } + + /** + * A message of unknown length is framed before grpc-java counts it: it still starts within the + * queue budget, and grows to its size. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void largeMessagesOfUnknownLengthAreAccepted(boolean restlet) throws Exception { + assertLargeMessagesSentOnReadinessAreAccepted(restlet, UNKNOWN_LENGTH_UPLOAD); + } + + private void assertLargeMessagesSentOnReadinessAreAccepted( + boolean restlet, MethodDescriptor upload) throws Exception { adapter.setMaxQueuedBytes(16384); adapter.setExecutor(Runnable::run); start(restlet, streams()); @@ -637,7 +672,7 @@ void largeMessagesSentOnReadinessAreAccepted(boolean restlet) throws Exception { var result = new CompletableFuture(); AtomicInteger sent = new AtomicInteger(); ClientCall call = - adapter.getChannel().newCall(UPLOAD, deadline()); + adapter.getChannel().newCall(upload, deadline()); call.start( new ClientCall.Listener<>() { @Override @@ -738,6 +773,24 @@ public void onClose(Status status, Metadata trailers) { assertEquals(1, closes.get()); } + /** + * Waits until a counter stops moving for 300 ms, as the buffers on the way are full, or 10 s. + * + * @return The settled value. + */ + private static int settled(AtomicInteger counter) throws InterruptedException { + var never = new Semaphore(0); + long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + int last; + int value = counter.get(); + do { + last = value; + never.tryAcquire(300, TimeUnit.MILLISECONDS); + value = counter.get(); + } while (value != last && System.nanoTime() < end); + return value; + } + private static StreamObserver single(CompletableFuture result) { return new StreamObserver<>() { @Override From 60bcae848502e7748a2970f43c866e7ac78d3ced Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 5 Nov 2026 14:52:30 -0500 Subject: [PATCH 6/6] Walk gRPC request frames with a long-safe offset Co-Authored-By: Claude Opus 5.5 --- .../ext/grpc/internal/RestletClientStream.java | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java index ab928480eb..8f8c1fc741 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletClientStream.java @@ -268,14 +268,18 @@ private void send(WritableBuffer frame, boolean end) { */ private Status enqueue(Buffer buffer) { int size = buffer.size; - for (int i = 0; i < size; ) { + int offset = 0; + while (offset < size) { if (remaining > 0) { - int skipped = (int) Math.min(remaining, size - i); + long skipped = Math.min(remaining, (long) size - offset); remaining -= skipped; - i += skipped; - } else if ((prefix == 0 && queued + i > queueBytes) || !prefixed(buffer.data[i++])) { + offset += (int) skipped; + } else if ((prefix == 0 && queued + offset > queueBytes) + || !prefixed(buffer.data[offset])) { return Status.RESOURCE_EXHAUSTED.withDescription( "Request queue byte budget exceeded"); + } else { + offset++; } } if (size > 0) {