fix: report gRPC deadline races as cancellations

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

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<ClientInterceptor> getInterceptors() {
         return interceptors;
     }

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;

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;
@@ -43,6 +44,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;
@@ -57,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;
 
@@ -128,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)) {
@@ -183,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;

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());
     }
 
@@ -390,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();

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();
     }

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",

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<byte[], byte[]> method() {
         var marshaller =
                 new MethodDescriptor.Marshaller<byte[]>() {
                     @Override
@@ -60,57 +61,95 @@ public byte[] parse(InputStream input) {
                         throw new UnsupportedOperationException("No response is expected");
                     }
                 };
-        var method =
-                MethodDescriptor.<byte[], byte[]>newBuilder()
-                        .setType(MethodDescriptor.MethodType.UNARY)
-                        .setFullMethodName("review.Unary/Echo")
-                        .setRequestMarshaller(marshaller)
-                        .setResponseMarshaller(marshaller)
-                        .build();
+        return MethodDescriptor.<byte[], byte[]>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> 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> 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> 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> 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());
+        }
+    }
 }