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/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; } 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..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; @@ -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; 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..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 @@ -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(); 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", 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()); + } + } }