From 58436416be659a7841e4f36adcc9ff8f9d960371 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Thu, 29 Oct 2026 21:57:11 -0400 Subject: [PATCH 1/3] Add a getter for the gRPC client callback executor Pair setExecutor with getExecutor, so that every GrpcClientAdapter setting has a getter, following the Restlet API conventions used across the RPC extensions. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../main/java/org/restlet/ext/grpc/GrpcClientAdapter.java | 5 +++++ 1 file changed, 5 insertions(+) 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 3847e932e0..48950242a0 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 @@ -136,6 +136,11 @@ public synchronized void setExecutor(Executor value) { executor = Objects.requireNonNull(value, "executor"); } + /** Returns the borrowed SDK callback executor, or null when the SDK default is used. */ + public synchronized Executor getExecutor() { + return executor; + } + public List getInterceptors() { return interceptors; } From bdc41d8ab03a0ab9bb468c58a87233d4a99cd248 Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Fri, 30 Oct 2026 09:26:06 -0400 Subject: [PATCH 2/3] Tighten null-safety in the gRPC client streams and tests Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- .../ext/grpc/internal/GuardedClientStream.java | 1 + .../ext/grpc/internal/RestletClientStream.java | 1 + .../restlet/ext/grpc/GrpcClientAdapterTestCase.java | 5 +++-- .../restlet/ext/grpc/GrpcHttpValidationTestCase.java | 8 +++++--- .../java/org/restlet/ext/grpc/GrpcModuleTestCase.java | 11 ++++++----- 5 files changed, 16 insertions(+), 10 deletions(-) diff --git a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/GuardedClientStream.java b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/GuardedClientStream.java index e640b9e924..656c7a5eee 100644 --- a/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/GuardedClientStream.java +++ b/org.restlet.ext.grpc/src/main/java/org/restlet/ext/grpc/internal/GuardedClientStream.java @@ -19,6 +19,7 @@ import java.io.InputStream; /** Delayed SDK streams must turn synchronous framing failures into terminal call status. */ +@SuppressWarnings("null") final class GuardedClientStream implements ClientStream { private final ClientStream delegate; private final int maxBytes; 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 d52ca2e315..2890b89baa 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 @@ -43,6 +43,7 @@ import org.restlet.util.Series; /** SDK-framed, bounded unary exchange. No gRPC frame/status codec is implemented here. */ +@SuppressWarnings("null") final class RestletClientStream extends AbstractClientStream { private final SerializingExecutor serial = new SerializingExecutor(Runnable::run); private final State state; 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 eb15d91ec9..8cad53ac38 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 @@ -188,10 +188,11 @@ void preservesTrailersOnlyErrorAndRecovers() throws Exception { assertThrows(StatusRuntimeException.class, () -> call("bad")); assertEquals(Status.Code.INVALID_ARGUMENT, error.getStatus().getCode()); assertEquals("native % error: caf\u00e9", error.getStatus().getDescription()); + Metadata trailers = java.util.Objects.requireNonNull(error.getTrailers()); assertEquals( List.of("one", "two"), - com.google.common.collect.ImmutableList.copyOf(error.getTrailers().getAll(TEXT))); - assertArrayEquals(new byte[] {0, (byte) 255, 7}, error.getTrailers().get(BINARY)); + com.google.common.collect.ImmutableList.copyOf(trailers.getAll(TEXT))); + assertArrayEquals(new byte[] {0, (byte) 255, 7}, trailers.get(BINARY)); assertEquals("recovered", call("recovered").getValue()); } diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java index 3321577c08..5739bc72c1 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcHttpValidationTestCase.java @@ -160,7 +160,9 @@ void missingGrpcStatusIsNotSuccess() throws Exception { response -> response.setEntity("", MediaType.valueOf("application/grpc"))); Status status = failure(); assertEquals(Status.Code.UNKNOWN, status.getCode(), status::toString); - assertTrue(status.getDescription().contains("missing GRPC status")); + assertTrue( + java.util.Objects.requireNonNull(status.getDescription()) + .contains("missing GRPC status")); assertRecovered(); } @@ -265,8 +267,8 @@ void multipleUnaryResponseMessagesFailAndRecover() throws Exception { assertEquals(Status.Code.CANCELLED, status.getCode(), status::toString); assertEquals(Status.Code.INTERNAL, Status.fromThrowable(status.getCause()).getCode()); assertTrue( - Status.fromThrowable(status.getCause()) - .getDescription() + java.util.Objects.requireNonNull( + Status.fromThrowable(status.getCause()).getDescription()) .contains("More than one value")); assertRecovered(); } diff --git a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcModuleTestCase.java b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcModuleTestCase.java index 1133978ae6..81ff8796d1 100644 --- a/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcModuleTestCase.java +++ b/org.restlet.ext.grpc/src/test/java/org/restlet/ext/grpc/GrpcModuleTestCase.java @@ -66,7 +66,7 @@ void nativeChannelResolvesWithoutSdkNetworkTransportOrInternalExports(@TempDir P String modulePath = paths.stream() .distinct() - .map(Path::toString) + .map(path -> path.toString()) .collect(Collectors.joining(File.pathSeparator)); Path descriptor = temp.resolve("module-info.java"); Path example = temp.resolve("Main.java"); @@ -104,13 +104,14 @@ public static void main(String[] args) throws Exception { """); Path classes = temp.resolve("classes"); var diagnostics = new java.io.ByteArrayOutputStream(); + var diagnosticStream = new java.io.PrintStream(diagnostics, true); assertEquals( 0, - javax.tools.ToolProvider.getSystemJavaCompiler() + java.util.spi.ToolProvider.findFirst("javac") + .orElseThrow() .run( - null, - diagnostics, - diagnostics, + diagnosticStream, + diagnosticStream, "--module-path", modulePath, "-d", From 2b9209e21e1abe976fd7291ba693e054f31fd2bc Mon Sep 17 00:00:00 2001 From: Jerome Louvel <374450+jlouvel@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:16:17 -0500 Subject: [PATCH 3/3] Report gRPC deadline races as cancellations When the client deadline expires, the grpc-java peer resets the stream with CANCEL, and the Restlet HTTP/2 client reports it as a closed connection. Report such failures after the deadline as CANCELLED, like grpc-netty, so that the SDK call turns them into DEADLINE_EXCEEDED instead of UNAVAILABLE. Make a first call in the deadline tests before the timed call, so that class loading and the HTTP/2 connection no longer exhaust their 3-second deadlines in a cold JVM. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Jerome Louvel <374450+jlouvel@users.noreply.github.com> --- changes.md | 4 + .../grpc/internal/RestletClientStream.java | 22 +++- .../ext/grpc/GrpcClientAdapterTestCase.java | 19 +++ .../internal/RestletClientStreamTestCase.java | 119 ++++++++++++------ 4 files changed, 123 insertions(+), 41 deletions(-) diff --git a/changes.md b/changes.md index b80f352af9..d625a433ac 100644 --- a/changes.md +++ b/changes.md @@ -242,6 +242,10 @@ Changes log - The gRPC client could reject a call with `RESOURCE_EXHAUSTED` ("gRPC call limit exceeded") right after the end of the previous one: the call slot was released after the caller was notified. Issue #1622. + - A gRPC client call whose deadline expired could end with `UNAVAILABLE` instead of + `DEADLINE_EXCEEDED`: the peer resets the stream when its deadline expires, which the + Restlet HTTP/2 client reported as a closed connection. Such failures after the deadline + now end the call with `DEADLINE_EXCEEDED`, as with grpc-netty. - The HTTP/1.1 server connector no longer closes a kept-alive connection without notice when the application answers without reading the request body (for example an early 401, 403 or 406). The unread body is discarded when it is already received, or when at most 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 2890b89baa..5b6e6c37bd 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 @@ -11,6 +11,7 @@ import io.grpc.Attributes; import io.grpc.CallOptions; import io.grpc.ClientStreamTracer; +import io.grpc.Deadline; import io.grpc.InternalMetadata; import io.grpc.Metadata; import io.grpc.MethodDescriptor; @@ -58,6 +59,7 @@ final class RestletClientStream extends AbstractClientStream { private Metadata headers; private volatile Request request; private volatile Thread worker; + private volatile Deadline deadline; private volatile boolean done; private boolean sent; @@ -129,6 +131,12 @@ public Attributes getAttributes() { return Attributes.EMPTY; } + @Override + public void setDeadline(Deadline value) { + deadline = value; + super.setDeadline(value); + } + @Override public void setAuthority(String value) { if (!authority.equals(value)) { @@ -184,10 +192,22 @@ public void cancel(Status status) { } private void fail(Status status) { + Deadline current = deadline; + Status reported = + (status.getCode() == Status.Code.UNAVAILABLE + && current != null + && current.isExpired()) + // A peer whose deadline expired resets the stream with CANCEL, which + // Restlet reports as a closed connection. Report it like grpc-netty, so + // that the SDK call turns it into DEADLINE_EXCEEDED. + ? Status.CANCELLED + .withDescription("gRPC stream ended after the deadline") + .withCause(status.getCause()) + : status; serial.execute( () -> { if (!done) { - state.fail(status); + state.fail(reported); } }); Request active = 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 8cad53ac38..bf06dc2919 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 @@ -391,9 +391,28 @@ private Pending startBlocked() throws Exception { })) .build(); start(service); + warmUp(); return new Pending(entered, cancelled); } + /** + * Makes a first call so that later deadlines measure the call, not class loading and the HTTP/2 + * connection: in a cold JVM they can exhaust a 3-second deadline before the request reaches the + * peer. The warm-up is retried when it hits such a deadline itself. + */ + private void warmUp() { + for (int attempt = 1; ; attempt++) { + try { + assertEquals("warm", call("warm").getValue()); + return; + } catch (StatusRuntimeException e) { + if (e.getStatus().getCode() != Status.Code.DEADLINE_EXCEEDED || attempt == 5) { + throw e; + } + } + } + } + @Test void cancellationBeforeHeadersHasOneTerminalCallbackAndRecovery() throws Exception { Pending pending = startBlocked(); 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 41bfd05f82..e6b3ba14e4 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 @@ -13,6 +13,7 @@ import io.grpc.CallOptions; import io.grpc.ClientStreamTracer; +import io.grpc.Deadline; import io.grpc.Metadata; import io.grpc.MethodDescriptor; import io.grpc.Status; @@ -22,6 +23,7 @@ import java.io.InputStream; import java.net.InetAddress; import java.net.InetSocketAddress; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; @@ -46,8 +48,7 @@ void cancelledTimersDoNotStayQueued() { } } - @Test - void shutdownBeforeListenerInstallationIsReplayedOnceAndReleasesAdmission() { + private static MethodDescriptor method() { var marshaller = new MethodDescriptor.Marshaller() { @Override @@ -60,57 +61,95 @@ public byte[] parse(InputStream input) { throw new UnsupportedOperationException("No response is expected"); } }; - var method = - MethodDescriptor.newBuilder() - .setType(MethodDescriptor.MethodType.UNARY) - .setFullMethodName("review.Unary/Echo") - .setRequestMarshaller(marshaller) - .setResponseMarshaller(marshaller) - .build(); + return MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNARY) + .setFullMethodName("review.Unary/Echo") + .setRequestMarshaller(marshaller) + .setResponseMarshaller(marshaller) + .build(); + } + + private static RestletClientStream stream(ExecutorService io, AtomicInteger released) { + return new RestletClientStream( + new Client(Protocol.HTTP), + new InetSocketAddress(InetAddress.getLoopbackAddress(), 8080), + "localhost:8080", + method(), + new Metadata(), + CallOptions.DEFAULT, + new ClientStreamTracer[0], + 128, + io, + ignored -> released.incrementAndGet()); + } + + /** Starts the stream with a listener recording its terminal status. */ + private static void start( + RestletClientStream stream, AtomicReference status, Runnable onClose) { + new GuardedClientStream(stream, 128) + .start( + new ClientStreamListener() { + @Override + public void headersRead(Metadata headers) {} + + @Override + public void closed( + Status value, RpcProgress progress, Metadata trailers) { + status.set(value); + onClose.run(); + } + + @Override + public void messagesAvailable(MessageProducer producer) {} + + @Override + public void onReady() {} + }); + } + + @Test + void shutdownBeforeListenerInstallationIsReplayedOnceAndReleasesAdmission() { AtomicInteger released = new AtomicInteger(); AtomicInteger terminals = new AtomicInteger(); AtomicReference status = new AtomicReference<>(); try (var io = Executors.newVirtualThreadPerTaskExecutor()) { - var stream = - new RestletClientStream( - new Client(Protocol.HTTP), - new InetSocketAddress(InetAddress.getLoopbackAddress(), 8080), - "localhost:8080", - method, - new Metadata(), - CallOptions.DEFAULT, - new ClientStreamTracer[0], - 128, - io, - ignored -> released.incrementAndGet()); + var stream = stream(io, released); Status shutdown = Status.UNAVAILABLE.withDescription("shutdown before start"); stream.cancel(shutdown); stream.cancel(Status.CANCELLED); assertEquals(0, terminals.get()); assertEquals(0, released.get()); - new GuardedClientStream(stream, 128) - .start( - new ClientStreamListener() { - @Override - public void headersRead(Metadata headers) {} - - @Override - public void closed( - Status value, RpcProgress progress, Metadata trailers) { - status.set(value); - terminals.incrementAndGet(); - } - - @Override - public void messagesAvailable(MessageProducer producer) {} - - @Override - public void onReady() {} - }); + start(stream, status, terminals::incrementAndGet); stream.cancel(Status.CANCELLED); assertSame(shutdown, status.get()); assertEquals(1, terminals.get()); assertEquals(1, released.get()); } } + + @Test + void transportFailureAfterTheDeadlineIsReportedAsCancelled() { + AtomicReference status = new AtomicReference<>(); + try (var io = Executors.newVirtualThreadPerTaskExecutor()) { + var stream = stream(io, new AtomicInteger()); + stream.setDeadline(Deadline.after(-1, TimeUnit.SECONDS)); + start(stream, status, () -> {}); + var cause = new java.io.IOException("Connection closed by peer"); + stream.cancel(Status.UNAVAILABLE.withCause(cause)); + assertEquals(Status.Code.CANCELLED, status.get().getCode()); + assertSame(cause, status.get().getCause()); + } + } + + @Test + void transportFailureBeforeTheDeadlineStaysUnavailable() { + AtomicReference status = new AtomicReference<>(); + try (var io = Executors.newVirtualThreadPerTaskExecutor()) { + var stream = stream(io, new AtomicInteger()); + stream.setDeadline(Deadline.after(1, TimeUnit.MINUTES)); + start(stream, status, () -> {}); + stream.cancel(Status.UNAVAILABLE); + assertEquals(Status.Code.UNAVAILABLE, status.get().getCode()); + } + } }