Fix/discovery wait ready time units

core/src/main/java/tech/ydb/core/impl/YdbTransportImpl.java

@@ -6,6 +6,7 @@
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 import com.google.common.base.Strings;
 import org.slf4j.Logger;
@@ -135,11 +136,14 @@ public AuthCallOptions getAuthCallOptions() {
     protected GrpcChannel getChannel(GrpcRequestSettings settings) {
         EndpointRecord endpoint = endpointPool.getEndpoint(channelPool.getReadyEndpoints(), settings);
         if (endpoint == null) {
-            long timeout = -1;
+            // negative value tells waitReady to use the default discovery timeout
+            long timeoutMs = -1;
             if (settings.getDeadlineAfter() != 0) {
-                timeout = settings.getDeadlineAfter() - System.nanoTime();
+                long leftNanos = settings.getDeadlineAfter() - System.nanoTime();
+                // an already expired deadline must not fall back to the default timeout
+                timeoutMs = Math.max(TimeUnit.NANOSECONDS.toMillis(leftNanos), 1);
             }
-            discovery.waitReady(timeout);
+            discovery.waitReady(timeoutMs);
             endpoint = endpointPool.getEndpoint(Collections.emptySet(), settings);
         }
         return channelPool.getChannel(endpoint);

core/src/test/java/tech/ydb/core/impl/YdbTransportImplTest.java

@@ -8,9 +8,11 @@
 import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 import javax.annotation.Nonnull;
 
@@ -27,6 +29,7 @@
 import tech.ydb.core.UnexpectedResultException;
 import tech.ydb.core.grpc.GrpcRequestSettings;
 import tech.ydb.core.grpc.GrpcTransport;
+import tech.ydb.core.grpc.GrpcTransportBuilder;
 import tech.ydb.core.grpc.YdbHeaders;
 import tech.ydb.core.impl.pool.EndpointRecord;
 import tech.ydb.core.impl.pool.ManagedChannelFactory;
@@ -183,6 +186,69 @@ public void asyncBuildGoodTest() {
         Assert.assertTrue(isReady.isDone());
     }
 
+    @Test
+    public void asyncWaitingForReadyTest() throws Exception {
+        CountDownLatch discoveryLatch = new CountDownLatch(1);
+        Queue<Runnable> lazyTasks = new ConcurrentLinkedQueue<>();
+        Executor lazyExecutor = (Runnable r) -> {
+            lazyTasks.add(r);
+            discoveryLatch.countDown();
+        };
+
+        Mockito.when(discoveryChannel.newCall(Mockito.eq(DiscoveryServiceGrpc.getListEndpointsMethod()), Mockito.any()))
+                .thenReturn(MockedCall.discovery(lazyExecutor, "self", new EndpointRecord("node", 2136)));
+        Mockito.when(transportChannel.newCall(Mockito.eq(DiscoveryServiceGrpc.getWhoAmIMethod()), Mockito.any()))
+                .thenReturn(MockedCall.whoAmICall("i am node"));
+
+        Duration discoveryTimeout = Duration.ofSeconds(5);
+        Duration callTimeout = Duration.ofMillis(50);
+
+        Assert.assertEquals(0, lazyTasks.size());
+        try (GrpcTransport transport = GrpcTransport.forConnectionString("grpc://mocked:2136/local")
+                .withInitMode(GrpcTransportBuilder.InitMode.ASYNC)
+                .withDiscoveryTimeout(discoveryTimeout)
+                .withChannelFactoryBuilder(builder -> channelFactory)
+                .build()
+        ) {
+            Assert.assertTrue(discoveryLatch.await(1, TimeUnit.SECONDS));
+            Assert.assertEquals(1, lazyTasks.size()); // a discovery call
+
+            long startedAt = System.nanoTime();
+            CompletableFuture<Result<DiscoveryProtos.WhoAmIResponse>> call1 = CompletableFuture.supplyAsync(
+                    () ->transport.unaryCall(
+                            DiscoveryServiceGrpc.getWhoAmIMethod(),
+                            GrpcRequestSettings.newBuilder().withDeadline(callTimeout).build(),
+                            DiscoveryProtos.WhoAmIRequest.newBuilder().build()
+                    ).join(),
+                    testScheduler
+            );
+
+            Assert.assertFalse(call1.isDone());
+            Result<DiscoveryProtos.WhoAmIResponse> res1 = call1.get(5, TimeUnit.SECONDS);
+            Assert.assertEquals(StatusCode.CLIENT_INTERNAL_ERROR, res1.getStatus().getCode());
+            Assert.assertEquals(1, lazyTasks.size()); // a new call wasn't executed
+
+            long nanos = System.nanoTime() - startedAt;
+            Assert.assertTrue(nanos >=  callTimeout.toNanos());
+            Assert.assertTrue(nanos < discoveryTimeout.toNanos());
+
+            CompletableFuture<Result<DiscoveryProtos.WhoAmIResponse>> call2 = CompletableFuture.supplyAsync(
+                    () ->transport.unaryCall(
+                            DiscoveryServiceGrpc.getWhoAmIMethod(),
+                            GrpcRequestSettings.newBuilder().build(),
+                            DiscoveryProtos.WhoAmIRequest.newBuilder().build()
+                    ).join(),
+                    testScheduler
+            );
+            Assert.assertFalse(call2.isDone());
+            Assert.assertEquals(1, lazyTasks.size()); // a new call wasn't executed
+
+            lazyTasks.poll().run(); // complete discovery
+            Result<DiscoveryProtos.WhoAmIResponse> res2 = call2.join();
+            Assert.assertTrue(res2.isSuccess());
+        }
+    }
+
     @Test
     public void failFastOnMissingPort() {
         String endpoint = "127.1.2.3";