Disable request timeout if grpc-timeout is omitted while `useClient…

…TimeoutHeader` is enabled. (line#5144)

Motivation:

gRPC specification says:
> If Timeout is omitted a server should assume an infinite timeout.
> Client implementations are free to send a default minimum timeout
> based on their deployment requirements.

Armeria gRPC implementation does not follow the specification and uses the server's timeout as is.

Modifications:

- Clear a request timeout if no `grpc-timeout` header is set when `useClientTimeoutHeader` enabled

Result:

Armeria gRPC service no longer schedules a timeout if `grpc-timeout` header is omitted when `GrpcServiceBuilder.userClientTimeoutHeader()` is enabled.

grpc/src/main/java/com/linecorp/armeria/server/grpc/AbstractUnframedGrpcService.java

@@ -63,17 +63,20 @@
 import io.grpc.ServerServiceDefinition;
 import io.grpc.Status;
 import io.grpc.Status.Code;
+import io.netty.util.AttributeKey;
 
 /**
  * Common part of the {@link UnframedGrpcService} and {@link HttpJsonTranscodingService}.
  */
 abstract class AbstractUnframedGrpcService extends SimpleDecoratingHttpService implements GrpcService {
 
+    private static final Logger logger = LoggerFactory.getLogger(AbstractUnframedGrpcService.class);
+    static final AttributeKey<Boolean> IS_UNFRAMED_GRPC =
+            AttributeKey.valueOf(AbstractUnframedGrpcService.class, "IS_UNFRAMED_GRPC");
+
     private final GrpcService delegate;
     private final UnframedGrpcErrorHandler unframedGrpcErrorHandler;
 
-    private static final Logger logger = LoggerFactory.getLogger(AbstractUnframedGrpcService.class);
-
     /**
      * Creates a new instance that decorates the specified {@link HttpService}.
      */
@@ -142,6 +145,7 @@ protected void frameAndServe(
             @Nullable Function<HttpData, HttpData> responseBodyConverter,
             MediaType responseContentType) {
         final HttpRequest grpcRequest;
+        ctx.setAttr(IS_UNFRAMED_GRPC, true);
         try (ArmeriaMessageFramer framer = new ArmeriaMessageFramer(
                 ctx.alloc(), ArmeriaMessageFramer.NO_MAX_OUTBOUND_MESSAGE_SIZE, false)) {
             final HttpData frame;
@@ -194,7 +198,7 @@ static void deframeAndRespond(ServiceRequestContext ctx,
             PooledObjects.close(grpcResponse.content());
             res.completeExceptionally(new NullPointerException("grpcStatusCode must not be null"));
             logger.warn("{} A gRPC response must have the {} header. response: {}",
-                    ctx, GrpcHeaderNames.GRPC_STATUS, grpcResponse);
+                        ctx, GrpcHeaderNames.GRPC_STATUS, grpcResponse);
             return;
         }
         Status grpcStatus = Status.fromCodeValue(Integer.parseInt(grpcStatusCode));

grpc/src/main/java/com/linecorp/armeria/server/grpc/FramedGrpcService.java

@@ -243,6 +243,15 @@ protected HttpResponse doPost(ServiceRequestContext ctx, HttpRequest req) throws
                                     ctx, defaultHeaders.get(serializationFormat).toBuilder(),
                                     GrpcStatus.fromThrowable(statusFunction, ctx, e, metadata), metadata));
                 }
+            } else {
+                if (Boolean.TRUE.equals(ctx.attr(AbstractUnframedGrpcService.IS_UNFRAMED_GRPC))) {
+                    // For unframed gRPC, we use the default timeout.
+                } else {
+                    // For framed gRPC, as per gRPC specification, if timeout is omitted a server should assume
+                    // an infinite timeout.
+                    // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#protocol
+                    ctx.clearRequestTimeout();
+                }
             }
         }
 

grpc/src/test/java/com/linecorp/armeria/client/grpc/GrpcClientTimeoutTest.java

@@ -36,7 +36,7 @@
 import com.google.common.util.concurrent.Uninterruptibles;
 
 import com.linecorp.armeria.server.ServerBuilder;
-import com.linecorp.armeria.server.ServiceRequestContext;
+import com.linecorp.armeria.server.annotation.Blocking;
 import com.linecorp.armeria.server.grpc.GrpcService;
 import com.linecorp.armeria.testing.junit5.server.ServerExtension;
 
@@ -85,6 +85,18 @@ protected void configure(ServerBuilder sb) throws Exception {
         }
     };
 
+    @RegisterExtension
+    static ServerExtension serverWithNoClientTimeout = new ServerExtension() {
+        @Override
+        protected void configure(ServerBuilder sb) throws Exception {
+            sb.requestTimeoutMillis(2000);
+            sb.service(GrpcService.builder()
+                                  .useClientTimeoutHeader(false)
+                                  .addService(new SlowService())
+                                  .build());
+        }
+    };
+
     @Test
     void clientTimeout() throws InterruptedException {
         final TestServiceBlockingStub client =
@@ -105,9 +117,21 @@ void clientTimeout() throws InterruptedException {
         });
     }
 
+    @Test
+    void clientNoDeadline() {
+        final TestServiceBlockingStub client =
+                GrpcClients.builder(server.httpUri())
+                           // Server disables the timeout if grpc-timeout is not specified and
+                           // useClientTimeoutHeaders is set to true.
+                           .responseTimeoutMillis(0)
+                           .build(TestServiceBlockingStub.class);
+        final SimpleResponse simpleResponse = client.unaryCall(SimpleRequest.getDefaultInstance());
+        assertThat(simpleResponse.getUsername()).isEqualTo("Armeria");
+    }
+
     @Test
     void serverTimeout() throws InterruptedException {
-        final TestServiceBlockingStub client = GrpcClients.builder(server.httpUri())
+        final TestServiceBlockingStub client = GrpcClients.builder(serverWithNoClientTimeout.httpUri())
                                                           .responseTimeoutMillis(0)
                                                           .build(TestServiceBlockingStub.class);
 
@@ -127,19 +151,16 @@ void serverTimeout() throws InterruptedException {
     }
 
     private static class SlowService extends TestServiceImplBase {
+        @Blocking
         @Override
         public void unaryCall(SimpleRequest request, StreamObserver<SimpleResponse> responseObserver) {
-            ServiceRequestContext.current()
-                                 .blockingTaskExecutor()
-                                 .submit(() -> {
-                                     // Defer response
-                                     logger.debug("Perform a long running task.");
-                                     Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
-                                     responseObserver.onNext(SimpleResponse.newBuilder()
-                                                                           .setUsername("Armeria")
-                                                                           .build());
-                                     responseObserver.onCompleted();
-                                 });
+            // Defer response
+            logger.debug("Perform a long running task.");
+            Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
+            responseObserver.onNext(SimpleResponse.newBuilder()
+                                                  .setUsername("Armeria")
+                                                  .build());
+            responseObserver.onCompleted();
         }
     }
 }