From 41fcce5e3eacdf13747211acee057101df8cae21 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Wed, 4 Nov 2026 21:58:31 -0500 Subject: [PATCH 1/3] Host client-streaming and bidirectional gRPC methods The server host read the whole request before handing it to the SDK, so GrpcServerAdapter rejected client-streaming and bidirectional methods. Read the request on its own virtual thread instead, handing it to the SDK as it arrives: the handler returns once the response head is ready, so responses stream out while the client still sends, and a bidirectional call exchanges messages both ways over one HTTP/2 stream. Request 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, so an unread request pushes back on the client. Message size stays bounded by the SDK's deframer. A request stream that breaks while read now ends the call as CANCELLED, like a peer reset in grpc-netty, and a read stuck after the call ended is still reset. Part of #1561 (R6a). Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 11 +- .../restlet/ext/grpc/GrpcServerAdapter.java | 22 +- .../ext/grpc/internal/RestletServerHost.java | 96 ++++-- .../ext/grpc/GrpcServerAdapterTestCase.java | 308 ++++++++++++++++-- readme.md | 4 +- 5 files changed, 388 insertions(+), 53 deletions(-) diff --git a/changes.md b/changes.md index 3674696dc2..28fa95112c 100644 --- a/changes.md +++ b/changes.md @@ -95,15 +95,18 @@ Changes log bounded messages/calls, one-shot SSE responses, and request cancellation. Sessions, listening GET streams, replay and Ikanos revision compatibility are not supported. Borrowed HTTP connectors are not stopped by the MCP client. - - Added experimental SDK-native gRPC unary and bounded server-streaming hosting on a - borrowed Restlet/Netty HTTP/2 listener. Generated and dynamic services can share the + - Added experimental SDK-native gRPC hosting of unary, server-streaming, + client-streaming and bidirectional methods on a borrowed Restlet/Netty HTTP/2 listener. + Request messages are read as the application requests them, within the queue budget, + so an unread request pushes back on the client, while responses stream out; a + bidirectional call exchanges messages both ways over one HTTP/2 stream. Generated and dynamic services can share the listener with REST routes. grpc-java owns service dispatch, framing and status semantics; Restlet owns HTTP I/O and finite message, queue, concurrency and call-duration budgets. Cancellation and response completion terminate each logical transport exactly once; admission remains held until response transmission or cancellation. Adapter shutdown leaves the borrowed listener and supplied callback executor running and reports timeout - explicitly. Client-streaming, bidirectional streaming, TLS configuration and physical - gRPC connection controls are not supported by the server adapter. + 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. Regression tests cover authority mapping, cancellation before and after headers, diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java index 8f698ed6dd..81f2acb29c 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/GrpcServerAdapter.java @@ -30,9 +30,11 @@ /** * Socketless grpc-java server on a borrowed Restlet HTTP/2 listener. * - *

Supports unary requests and server-streamed responses, with finite message, queue, concurrency - * and duration budgets. Does not implement TLS or physical connection controls. The host - * application owns the HTTP listener and any supplied callback executor. + *

Supports unary, server-streaming, client-streaming and bidirectional methods, with finite + * message, queue, concurrency and duration budgets. Request messages are read as the application + * requests them, and responses are sent as they are produced, so a bidirectional call can exchange + * messages both ways over one HTTP/2 stream. Does not implement TLS or physical connection + * controls. The host application owns the HTTP listener and any supplied callback executor. * *

Register services before {@link #start()}, attach them to a router with {@link #attachTo}, and * configure the host listener for plaintext HTTP/2. This adapter is single-use. Stopping it cancels @@ -93,6 +95,12 @@ public int getMaxMessageBytes() { } } + /** + * Sets the per-call queue budget, for each direction: response bytes waiting for the network, + * and request bytes read but not yet consumed by the messages the application requested. + * + * @param value At least 16384 bytes. + */ public void setMaxQueuedBytes(int value) { synchronized (lock) { configurable(); @@ -201,11 +209,11 @@ public synchronized void start() throws Exception { configurable(); for (var service : List.copyOf(services)) { for (var method : service.getMethods()) { - var type = method.getMethodDescriptor().getType(); - if (type != io.grpc.MethodDescriptor.MethodType.UNARY - && type != io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING) { + if (method.getMethodDescriptor().getType() + == io.grpc.MethodDescriptor.MethodType.UNKNOWN) { throw new UnsupportedOperationException( - "Only unary and server-streaming methods are supported"); + "Methods of unknown type are not supported: " + + method.getMethodDescriptor().getFullMethodName()); } } } diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java index 21a5af88ca..342f607643 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java @@ -323,6 +323,10 @@ private final class Call implements ServerTransport { private final InternalLogId id = InternalLogId.allocate(Call.class, "Restlet logical call"); private final ArrayDeque queue = new ArrayDeque<>(); + + /** Guards {@link #unread}, and wakes the request reader. */ + private final Object inbound = new Object(); + private final CountDownLatch head = new CountDownLatch(1); private final AtomicBoolean finished = new AtomicBoolean(); private final AtomicLong operations = new AtomicLong(); @@ -339,6 +343,9 @@ private final class Call implements ServerTransport { private Metadata trailers; private int queued; + /** Request bytes handed to the SDK but not consumed yet; guarded by {@link #inbound}. */ + private long unread; + /** Bytes framed for the latest message, from the frame that started it. */ private int open; @@ -410,19 +417,12 @@ private void begin() throws IOException, InterruptedException { transportListener.streamCreated(stream, method, headers); stream.state.onStreamAllocated(); }); - byte[] bytes; - try (var input = request.getEntity().getStream()) { - bytes = input == null ? new byte[0] : read(input); - } - if (bytes.length > maxBytes + 5) { - cancel(Status.RESOURCE_EXHAUSTED.withDescription("Request byte budget exceeded")); + InputStream input = request.getEntity().getStream(); + if (input == null) { + deliver(new byte[0], true); } else { - serial.execute( - () -> { - if (!reported) { - stream.state.inboundDataReceived(ReadableBuffers.wrap(bytes), true); - } - }); + // Read apart, so that the response can start while the client still sends. + Thread.ofVirtual().name("restlet-grpc-server-read").start(() -> readAll(input)); } if (!head.await(duration.toNanos(), TimeUnit.NANOSECONDS)) { cancel(Status.DEADLINE_EXCEEDED.withDescription("Response headers timed out")); @@ -449,12 +449,65 @@ private Attributes transportAttributes() { return result.build(); } - private byte[] read(InputStream input) throws IOException { - enterIo(); - try { - return input.readNBytes(maxBytes + 6); - } finally { - blocked.set(0); + /** + * Hands the request body to the SDK as it arrives. Bytes handed over but not yet consumed + * by the SDK's deframer, which only reads the messages the application requested, stay + * within the queue budget: a slow consumer stops the reads, and so the client. + */ + private void readAll(InputStream input) { + try (input) { + byte[] chunk = new byte[16384]; + for (; ; ) { + int room; + synchronized (inbound) { + while (unread >= queueBytes && !finished.get()) { + inbound.wait(); + } + if (finished.get()) { + return; + } + room = (int) Math.min(chunk.length, queueBytes - unread); + } + int count; + enterIo(); + try { + count = input.read(chunk, 0, room); + } finally { + blocked.set(0); + } + if (count < 0) { + deliver(new byte[0], true); + return; + } + if (count > 0) { + synchronized (inbound) { + unread += count; + } + deliver(java.util.Arrays.copyOf(chunk, count), false); + } + } + } catch (InterruptedException e) { + cancel(Status.CANCELLED.withDescription("gRPC request read interrupted")); + } catch (IOException | RuntimeException e) { + cancel(Status.CANCELLED.withDescription("gRPC request read failed").withCause(e)); + } + } + + /** Hands request bytes to the SDK, in {@link #serial} after the stream's creation. */ + private void deliver(byte[] bytes, boolean end) { + serial.execute( + () -> { + if (!reported) { + stream.state.inboundDataReceived(ReadableBuffers.wrap(bytes), end); + } + }); + } + + /** Frees inbound budget once the SDK's deframer consumed request bytes. */ + private void consumed(int count) { + synchronized (inbound) { + unread -= count; + inbound.notifyAll(); } } @@ -584,6 +637,9 @@ private boolean claim() { if (timeout != null) { timeout.cancel(false); } + synchronized (inbound) { + inbound.notifyAll(); + } return true; } @@ -750,7 +806,9 @@ public void runOnTransportThread(Runnable task) { } @Override - public void bytesRead(int count) {} + public void bytesRead(int count) { + call.consumed(count); + } @Override public void deframeFailed(Throwable cause) { 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 434bfa6dd6..2cc83b262e 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 @@ -78,6 +78,16 @@ class GrpcServerAdapterTestCase { .setType(MethodDescriptor.MethodType.SERVER_STREAMING) .setFullMethodName("review.Server/Watch") .build(); + private static final MethodDescriptor UPLOAD = + ECHO.toBuilder() + .setType(MethodDescriptor.MethodType.CLIENT_STREAMING) + .setFullMethodName("review.Server/Upload") + .build(); + private static final MethodDescriptor CHAT = + ECHO.toBuilder() + .setType(MethodDescriptor.MethodType.BIDI_STREAMING) + .setFullMethodName("review.Server/Chat") + .build(); private final GrpcServerAdapter adapter = new GrpcServerAdapter(); private Server http; private ManagedChannel channel; @@ -369,33 +379,288 @@ void requestAndResponseMessageBudgetsAreEnforced() throws Exception { } @Test - void unsupportedStreamingRegistrationFailsExplicitly() { + void unknownMethodTypesFailExplicitly() { adapter.getServices() .add( ServerServiceDefinition.builder("review.Server") .addMethod( ECHO.toBuilder() - .setType( - MethodDescriptor.MethodType - .CLIENT_STREAMING) + .setType(MethodDescriptor.MethodType.UNKNOWN) .build(), - ServerCalls.asyncClientStreamingCall( - out -> - new io.grpc.stub.StreamObserver< - StringValue>() { - @Override - public void onNext(StringValue value) {} - - @Override - public void onError(Throwable cause) {} - - @Override - public void onCompleted() {} - })) + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver + out) -> out.onCompleted())) .build()); assertThrows(UnsupportedOperationException.class, adapter::start); } + /** Counts the messages and characters of an uploaded stream. */ + private static ServerServiceDefinition upload() { + return ServerServiceDefinition.builder("review.Server") + .addMethod( + UPLOAD, + ServerCalls.asyncClientStreamingCall( + out -> + new io.grpc.stub.StreamObserver() { + private int messages; + private long characters; + + @Override + public void onNext(StringValue value) { + messages++; + characters += value.getValue().length(); + } + + @Override + public void onError(Throwable cause) {} + + @Override + public void onCompleted() { + out.onNext( + StringValue.of( + messages + ":" + characters)); + out.onCompleted(); + } + })) + .build(); + } + + @Test + void clientStreamingReceivesEveryMessage() throws Exception { + start(upload()); + var result = new java.util.concurrent.CompletableFuture(); + var requests = + ClientCalls.asyncClientStreamingCall( + channel.newCall( + UPLOAD, + CallOptions.DEFAULT.withDeadlineAfter(15, TimeUnit.SECONDS)), + observer(result)); + for (int i = 0; i < 200; i++) { + requests.onNext(StringValue.of("x".repeat(4096))); + } + requests.onCompleted(); + assertEquals("200:" + (200 * 4096), result.get(15, TimeUnit.SECONDS)); + } + + /** Each reply is received before the next request is sent, over one HTTP/2 stream. */ + @Test + void bidirectionalCallsExchangeMessagesBothWays() throws Exception { + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + CHAT, + ServerCalls.asyncBidiStreamingCall( + out -> + new io.grpc.stub.StreamObserver() { + @Override + public void onNext(StringValue value) { + out.onNext( + StringValue.of( + "re:" + value.getValue())); + } + + @Override + public void onError(Throwable cause) {} + + @Override + public void onCompleted() { + out.onCompleted(); + } + })) + .build()); + var replies = new java.util.concurrent.LinkedBlockingQueue(); + var done = new java.util.concurrent.CompletableFuture(); + var requests = + ClientCalls.asyncBidiStreamingCall( + channel.newCall( + CHAT, CallOptions.DEFAULT.withDeadlineAfter(15, TimeUnit.SECONDS)), + new io.grpc.stub.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(10, TimeUnit.SECONDS)); + } + requests.onCompleted(); + done.get(10, TimeUnit.SECONDS); + assertTrue(replies.isEmpty()); + } + + /** + * Request bytes the application didn't ask for stay within the queue budget plus HTTP/2 flow + * control windows, so a client that respects readiness stops sending. + */ + @Test + void unrequestedMessagesPushBackOnTheClient() throws Exception { + adapter.setMaxQueuedBytes(32768); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger received = new AtomicInteger(); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + UPLOAD, + ServerCalls.asyncClientStreamingCall( + out -> { + var observer = + (ServerCallStreamObserver) out; + observer.disableAutoRequest(); + Thread.startVirtualThread( + () -> { + try { + release.await(); + } catch (InterruptedException e) { + return; + } + observer.request(Integer.MAX_VALUE); + }); + return new io.grpc.stub.StreamObserver() { + @Override + public void onNext(StringValue value) { + received.incrementAndGet(); + } + + @Override + public void onError(Throwable cause) {} + + @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 java.util.concurrent.CompletableFuture(); + ClientCall call = + channel.newCall( + UPLOAD, CallOptions.DEFAULT.withDeadlineAfter(30, TimeUnit.SECONDS)); + 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("x".repeat(4096))); + 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); + 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 requested message larger than the queue budget is consumed as it arrives. */ + @Test + void messagesLargerThanTheQueueBudgetStreamThrough() throws Exception { + adapter.setMaxQueuedBytes(16384); + start(echo()); + String large = "x".repeat(512 * 1024); + assertEquals(large, call(large)); + } + + @Test + void cancellingAStreamingCallReleasesItsSlot() throws Exception { + adapter.setMaxConcurrentCalls(1); + CountDownLatch cancelled = new CountDownLatch(1); + CountDownLatch receiving = new CountDownLatch(1); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + CHAT, + ServerCalls.asyncBidiStreamingCall( + out -> { + ((ServerCallStreamObserver) out) + .setOnCancelHandler(cancelled::countDown); + return new io.grpc.stub.StreamObserver() { + @Override + public void onNext(StringValue value) { + receiving.countDown(); + } + + @Override + public void onError(Throwable cause) {} + + @Override + public void onCompleted() { + out.onCompleted(); + } + }; + })) + .addMethod( + ECHO, + ServerCalls.asyncUnaryCall( + (StringValue request, + io.grpc.stub.StreamObserver out) -> { + out.onNext(request); + out.onCompleted(); + })) + .build()); + ClientCall call = + channel.newCall(CHAT, CallOptions.DEFAULT.withDeadlineAfter(15, TimeUnit.SECONDS)); + call.start(new ClientCall.Listener<>() {}, new Metadata()); + call.sendMessage(StringValue.of("open")); + assertTrue(receiving.await(10, TimeUnit.SECONDS)); + call.cancel("client gave up", null); + assertTrue(cancelled.await(10, TimeUnit.SECONDS)); + assertEquals("next", call("next")); + } + + private static io.grpc.stub.StreamObserver observer( + java.util.concurrent.CompletableFuture result) { + return new io.grpc.stub.StreamObserver<>() { + @Override + public void onNext(StringValue value) { + result.complete(value.getValue()); + } + + @Override + public void onError(Throwable cause) { + result.completeExceptionally(cause); + } + + @Override + public void onCompleted() {} + }; + } + private static byte[] frame(String value) throws IOException { var bytes = new ByteArrayOutputStream(); var data = new DataOutputStream(bytes); @@ -834,9 +1099,8 @@ public int read() throws IOException { () -> {}); Response response = new Response(request); adapter.handle(request, response); - assertEquals( - String.valueOf(Status.Code.INVALID_ARGUMENT.value()), - trailersAfterWriting(response)); + // A request stream that breaks while read is a peer reset, as grpc-netty reports it. + assertEquals(String.valueOf(Status.Code.CANCELLED.value()), trailersAfterWriting(response)); } @Test @@ -862,7 +1126,9 @@ public int read() throws IOException { try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { executor.submit(() -> adapter.handle(request, response)).get(10, TimeUnit.SECONDS); } - assertEquals(0, reset.getCount()); + // The request is read apart from the handler, which returns once the call ends; the read + // still stuck after that is reset shortly after. + assertTrue(reset.await(10, TimeUnit.SECONDS)); assertEquals( String.valueOf(Status.Code.DEADLINE_EXCEEDED.value()), trailersAfterWriting(response)); diff --git a/readme.md b/readme.md index 436ae556d6..6e0c91d525 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: unary and server-streaming services, and unary client, over HTTP/2 | +| [gRPC](https://grpc.io/) | `org.restlet.ext.grpc` | Experimental: services of every streaming type, and unary client, 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 client and bidirectional streaming for gRPC, and MCP sessions and resumable streams. Feedback is welcome in [Discussions](https://github.com/restlet/restlet-framework-java/discussions). +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). ## Learn more From 0fe93726da3ce7eb4c6f90ed793bcf4f647e745c Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 5 Nov 2026 06:07:28 -0500 Subject: [PATCH 2/3] Track gRPC reads and writes apart, and ignore the server's own body close The request reader and the response writer now run concurrently but shared one in-progress marker, so a read ending while a response write was stuck cleared the writer's marker, and a later deadline or shutdown no longer reset that write. Track reads and writes separately, and reset the stream if either stays stuck after the call ended. When a bidirectional service finishes before the client half-closes, the Netty server closes the unread request body after the terminal write, and the reader took that for a peer failure, cancelling a completed call before its delivery. Ignore read failures once the response ended; the delivery and disconnect signals settle the call. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../ext/grpc/internal/RestletServerHost.java | 37 ++-- .../ext/grpc/GrpcServerAdapterTestCase.java | 164 ++++++++++++++++++ 2 files changed, 190 insertions(+), 11 deletions(-) diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java index 342f607643..49fc73f343 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java @@ -334,8 +334,14 @@ private final class Call implements ServerTransport { /** 0 before the application callbacks, 1 while they may run, 2 once released. */ private final AtomicInteger callbackState = new AtomicInteger(); - /** The network read or write in progress, 0 if none. */ - private final AtomicLong blocked = new AtomicLong(); + /** + * The network read in progress, 0 if none. Reads and writes run concurrently, so each has + * its own marker: the end of one must not hide the other from {@link #abortIfStuck}. + */ + private final AtomicLong reading = new AtomicLong(); + + /** The network write in progress, 0 if none. */ + private final AtomicLong writing = new AtomicLong(); private ServerTransportListener transportListener; private SdkStream stream; @@ -469,11 +475,11 @@ private void readAll(InputStream input) { room = (int) Math.min(chunk.length, queueBytes - unread); } int count; - enterIo(); + enterIo(reading); try { count = input.read(chunk, 0, room); } finally { - blocked.set(0); + reading.set(0); } if (count < 0) { deliver(new byte[0], true); @@ -489,6 +495,13 @@ private void readAll(InputStream input) { } catch (InterruptedException e) { cancel(Status.CANCELLED.withDescription("gRPC request read interrupted")); } catch (IOException | RuntimeException e) { + synchronized (queue) { + if (terminal) { + // The response ended, so the server closed the unread request body; the + // delivery or disconnect signal settles the call, not this read. + return; + } + } cancel(Status.CANCELLED.withDescription("gRPC request read failed").withCause(e)); } } @@ -529,12 +542,12 @@ public void write(OutputStream output) throws IOException { } queued -= data.length; } - enterIo(); + enterIo(writing); try { output.write(data); output.flush(); } finally { - blocked.set(0); + writing.set(0); } int sent = data.length; serial.execute(() -> stream.state.onSentBytes(sent)); @@ -552,8 +565,8 @@ public void write(OutputStream output) throws IOException { } /** Marks the start of a network read or write, which only the peer may otherwise end. */ - private void enterIo() { - blocked.set(operations.incrementAndGet()); + private void enterIo(AtomicLong marker) { + marker.set(operations.incrementAndGet()); if (finished.get()) { abortIfStuck(); } @@ -565,8 +578,9 @@ private void enterIo() { * elsewhere was already woken by the end of the call. */ private void abortIfStuck() { - long operation = blocked.get(); - if (operation == 0) { + long read = reading.get(); + long write = writing.get(); + if (read == 0 && write == 0) { return; } Thread.startVirtualThread( @@ -577,7 +591,8 @@ private void abortIfStuck() { Thread.currentThread().interrupt(); return; } - if (blocked.get() == operation) { + if ((read != 0 && reading.get() == read) + || (write != 0 && writing.get() == write)) { request.abort(); } }); 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 2cc83b262e..fbe5e050ca 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 @@ -1184,6 +1184,170 @@ public void write(int value) } } + /** The end of the request, read while a response write is blocked, doesn't hide that write. */ + @Test + void requestEndDoesNotHideABlockedResponseWrite() throws Exception { + adapter.getServices() + .add( + ServerServiceDefinition.builder("review.Server") + .addMethod( + CHAT, + ServerCalls.asyncBidiStreamingCall( + out -> + new io.grpc.stub.StreamObserver< + StringValue>() { + @Override + public void onNext(StringValue value) { + out.onNext(StringValue.of("first")); + } + + @Override + public void onError(Throwable cause) {} + + @Override + public void onCompleted() {} + })) + .build()); + adapter.start(); + CountDownLatch reset = new CountDownLatch(1); + CountDownLatch writing = new CountDownLatch(1); + byte[] frame = frame("watch"); + InputStream body = + new InputStream() { + private boolean sent; + + @Override + public int read() throws IOException { + throw new UnsupportedOperationException(); + } + + @Override + public int read(byte[] buffer, int offset, int length) throws IOException { + if (!sent) { + sent = true; + System.arraycopy(frame, 0, buffer, offset, frame.length); + return frame.length; + } + // The request ends only once the response write is blocked. + try { + writing.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return -1; + } + }; + Request request = grpcRequest(CHAT, body, reset::countDown); + Response response = new Response(request); + adapter.handle(request, response); + try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { + var writer = + executor.submit( + () -> { + response.getEntity() + .write( + new OutputStream() { + @Override + public void write(int value) + throws IOException { + writing.countDown(); + try { + // Bounded, so that a missing reset + // fails + // the test instead of hanging it. + reset.await(30, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + throw new IOException("stream reset"); + } + }); + return null; + }); + assertTrue(writing.await(10, TimeUnit.SECONDS)); + // Let the reader see the end of the request, then end the call. + Thread.sleep(100); + adapter.stop(); + assertTrue(reset.await(10, TimeUnit.SECONDS)); + assertThrows( + java.util.concurrent.ExecutionException.class, + () -> writer.get(10, TimeUnit.SECONDS)); + } + } + + /** + * A bidirectional service may finish before the client half-closes; the server then closes the + * request body, which must not turn the completed call into a cancelled one. + */ + @Test + void earlyServerCompletionIsNotACancellation() throws Exception { + CountDownLatch completed = new CountDownLatch(1); + CountDownLatch cancelled = new CountDownLatch(1); + adapter.getInterceptors() + .add( + new io.grpc.ServerInterceptor() { + @Override + public io.grpc.ServerCall.Listener interceptCall( + io.grpc.ServerCall call, + Metadata headers, + io.grpc.ServerCallHandler next) { + return new io.grpc.ForwardingServerCallListener + .SimpleForwardingServerCallListener<>( + next.startCall(call, headers)) { + @Override + public void onComplete() { + completed.countDown(); + super.onComplete(); + } + + @Override + public void onCancel() { + cancelled.countDown(); + super.onCancel(); + } + }; + } + }); + start( + ServerServiceDefinition.builder("review.Server") + .addMethod( + CHAT, + ServerCalls.asyncBidiStreamingCall( + out -> + new io.grpc.stub.StreamObserver() { + @Override + public void onNext(StringValue value) { + out.onNext(value); + out.onCompleted(); + } + + @Override + public void onError(Throwable cause) {} + + @Override + public void onCompleted() {} + })) + .build()); + var status = new java.util.concurrent.CompletableFuture(); + // The first call on a fresh channel also waits for the HTTP/2 connection setup. + ClientCall call = + channel.newCall(CHAT, CallOptions.DEFAULT.withDeadlineAfter(30, TimeUnit.SECONDS)); + call.start( + new ClientCall.Listener<>() { + @Override + public void onClose(Status value, Metadata trailers) { + status.complete(value); + } + }, + new Metadata()); + call.request(1); + // The client never half-closes. + call.sendMessage(StringValue.of("only")); + assertEquals(Status.Code.OK, status.get(30, TimeUnit.SECONDS).getCode()); + assertTrue(completed.await(10, TimeUnit.SECONDS)); + assertFalse(cancelled.await(200, TimeUnit.MILLISECONDS)); + } + @Test void callbacksMayUseTheAdapterWhileItStops() throws Exception { CountDownLatch entered = new CountDownLatch(1); From fb73debab39b999172dd728ad8da8005cee7643d Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 5 Nov 2026 06:58:07 -0500 Subject: [PATCH 3/3] Restore the interrupt in the gRPC request reader, and tidy its tests The request reader caught InterruptedException without restoring the thread's interrupt status (SonarCloud S2142). Restore it before cancelling the call, and do the same in a test helper. Comment the listener methods the tests don't use, and use unnamed catch variables. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../ext/grpc/internal/RestletServerHost.java | 3 +- .../ext/grpc/GrpcServerAdapterTestCase.java | 39 +++++++++++++------ 2 files changed, 30 insertions(+), 12 deletions(-) diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java index 49fc73f343..42fcde9a1a 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/RestletServerHost.java @@ -492,7 +492,8 @@ private void readAll(InputStream input) { deliver(java.util.Arrays.copyOf(chunk, count), false); } } - } catch (InterruptedException e) { + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); cancel(Status.CANCELLED.withDescription("gRPC request read interrupted")); } catch (IOException | RuntimeException e) { synchronized (queue) { 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 fbe5e050ca..da4d3b61b4 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 @@ -461,7 +461,9 @@ public void onNext(StringValue value) { } @Override - public void onError(Throwable cause) {} + public void onError(Throwable cause) { + /* Not used by this test. */ + } @Override public void onCompleted() { @@ -522,7 +524,8 @@ void unrequestedMessagesPushBackOnTheClient() throws Exception { () -> { try { release.await(); - } catch (InterruptedException e) { + } catch (InterruptedException _) { + Thread.currentThread().interrupt(); return; } observer.request(Integer.MAX_VALUE); @@ -534,7 +537,9 @@ public void onNext(StringValue value) { } @Override - public void onError(Throwable cause) {} + public void onError(Throwable cause) { + /* Not used by this test. */ + } @Override public void onCompleted() { @@ -616,7 +621,9 @@ public void onNext(StringValue value) { } @Override - public void onError(Throwable cause) {} + public void onError(Throwable cause) { + /* Not used by this test. */ + } @Override public void onCompleted() { @@ -657,7 +664,9 @@ public void onError(Throwable cause) { } @Override - public void onCompleted() {} + public void onCompleted() { + /* Not used by this test. */ + } }; } @@ -1202,10 +1211,14 @@ public void onNext(StringValue value) { } @Override - public void onError(Throwable cause) {} + public void onError(Throwable cause) { + /* Not used by this test. */ + } @Override - public void onCompleted() {} + public void onCompleted() { + /* Not used by this test. */ + } })) .build()); adapter.start(); @@ -1231,7 +1244,7 @@ public int read(byte[] buffer, int offset, int length) throws IOException { // The request ends only once the response write is blocked. try { writing.await(); - } catch (InterruptedException e) { + } catch (InterruptedException _) { Thread.currentThread().interrupt(); } return -1; @@ -1256,7 +1269,7 @@ public void write(int value) // fails // the test instead of hanging it. reset.await(30, TimeUnit.SECONDS); - } catch (InterruptedException e) { + } catch (InterruptedException _) { Thread.currentThread().interrupt(); } throw new IOException("stream reset"); @@ -1322,10 +1335,14 @@ public void onNext(StringValue value) { } @Override - public void onError(Throwable cause) {} + public void onError(Throwable cause) { + /* Not used by this test. */ + } @Override - public void onCompleted() {} + public void onCompleted() { + /* Not used by this test. */ + } })) .build()); var status = new java.util.concurrent.CompletableFuture();