fix: handle failed connection attempts once

CHANGELOG.md

@@ -30,6 +30,26 @@
   `VHANDSHAKE_NOT_COMPLETE`. The check now makes a synchronous `callrw` to every replicaset master,
   which reopens a router connection that vshard closed after a failed handshake and waits for the
   new one, requires every master to be `available`, and logs the replica and bucket states.
+- Fixed a bug when a server that accepts the TCP connection but does not send the greeting was
+  reported as "Connection closed by server". The connect future is now failed only with
+  `TimeoutException`, whose message names the address and the timeout, and no
+  `CLOSE_BY_REMOTE` event is fired. Previously the connection pool handled every such attempt
+  twice, so the reconnect chains multiplied while the server stayed silent. The greeting timer of
+  an attempt that the server closed no longer closes the channel of the next attempt.
+- Fixed reconnect chains multiplying when a server (or a proxy in front of it) accepts the TCP
+  connection and closes it before the greeting: the close was reported both by the failed connect
+  future and by a `CLOSE_BY_REMOTE` event, and the pool handled each report. Close events are now
+  fired only for a connection that has received the greeting.
+- Fixed a leak when graceful shutdown is enabled: every connect attempt added one more
+  `box.shutdown` callback to the same watcher, so a client that kept reconnecting accumulated
+  callbacks, and a `box.shutdown` event after a successful connect ran the callback once per
+  previous attempt. The callback is now added once per client.
+- Fixed the `pool.connect.time` timer of the connection pool: it was stopped only when an
+  attempt succeeded, so every failed attempt left an active sample behind.
+- Fixed `reconnectAfter` being ignored for the connections created with the client: the client
+  builder set it on the pool after the connections had been created with the default of 1000 ms,
+  and the pool did not pass the new value to existing connections. `withReconnectAfter` and the
+  `reconnect-after` property of Spring Data now take effect.
 
 ### Documentation
 

tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java

@@ -16,6 +16,7 @@
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -97,6 +98,10 @@ public class IProtoClientImpl implements IProtoClient {
   private final List<String> toUnwatch;
   protected final Map<Long, IProtoStateMachine> fsmRegistry;
   private final Map<String, Watcher> watchers;
+
+  /** Set once the {@code box.shutdown} callback is added; it survives reconnects. */
+  private final AtomicBoolean shutdownWatcherRegistered = new AtomicBoolean();
+
   protected final Timer timerService;
   private final WatcherOptions watcherOpts;
   private CompletableFuture<Integer> serverProtocolVersion;
@@ -185,7 +190,7 @@ public CompletableFuture<Void> connect(InetSocketAddress address, long timeoutMs
   @Override
   public CompletableFuture<Void> connect(
       InetSocketAddress address, long timeoutMs, boolean gracefulShutdown) {
-    if (gracefulShutdown) {
+    if (gracefulShutdown && shutdownWatcherRegistered.compareAndSet(false, true)) {
       watch(SHUTDOWN_EVENT_KEY, this::shutdownEventCallback);
     }
 

tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java

@@ -44,7 +44,9 @@ private enum CloseReason {
     /** Closed by client close reason. */
     CLOSED_BY_CLIENT(ERR_CLOSED_BY_CLIENT),
     /** Closed by shutdown close reason. */
-    CLOSED_BY_SHUTDOWN(ERR_CLOSED_BY_SHUTDOWN);
+    CLOSED_BY_SHUTDOWN(ERR_CLOSED_BY_SHUTDOWN),
+    /** Closed by the client because the greeting did not arrive in time. */
+    CLOSED_BY_TIMEOUT(ERR_CLOSED_BY_TIMEOUT);
 
     private final String message;
 
@@ -98,6 +100,9 @@ public String getMessage() {
   private static final String ERR_CLOSED_BY_CLIENT = "Connection closed by client";
   private static final String ERR_CLOSED_BY_SERVER = "Connection closed by server";
   private static final String ERR_CLOSED_BY_SHUTDOWN = "Connection closed by shutdown";
+  private static final String ERR_CLOSED_BY_TIMEOUT = "Connection closed by connect timeout";
+  private static final String ERR_GREETING_TIMEOUT =
+      "Connection timeout: no greeting from %s within %d ms";
   private static final String ERR_NOT_CONNECTED = "Connection is not established";
   private static final String ERR_SEND_FAILURE = "Failed to send IProto message: %s";
 
@@ -118,6 +123,16 @@ public String getMessage() {
   private Consumer<IProtoResponse> consumer;
   private CompletableFuture<Greeting> connectPromise;
   private Timeout timeoutHandler;
+
+  /** Greeting promise of the current attempt; a timer of an earlier attempt must not touch it. */
+  private CompletableFuture<Greeting> greetingPromise;
+
+  /**
+   * Whether the current attempt has received the greeting. A close of a connection that has never
+   * been established is reported only through the failed connect future, not by a close event.
+   */
+  private volatile boolean established;
+
   private int idleTimeout;
 
   /**
@@ -190,13 +205,16 @@ public synchronized CompletableFuture<Greeting> connect(InetSocketAddress addres
       log.debug("{} is connecting to {}", this, address);
 
       CompletableFuture<Greeting> promise = new CompletableFuture<>();
-      scheduleGreetingTimeout(promise, timeoutMs);
+      greetingPromise = promise;
+      established = false;
+      scheduleGreetingTimeout(promise, address, timeoutMs);
       connectPromise =
           promise.thenApply(
               greeting -> {
                 cancelGreetingTimeout();
                 log.info("{} connected to {}", this, address);
                 this.greeting = greeting;
+                established = true;
                 state.set(State.READY);
                 return greeting;
               });
@@ -358,6 +376,15 @@ private synchronized void onChannelClose(ChannelFuture f) {
     String message;
     ConnectionCloseEvent event;
     CloseReason lastCloseReason = closedBy.get();
+    if (lastCloseReason == CloseReason.CLOSED_BY_TIMEOUT) {
+      // The greeting timer has already failed the connect promise with TimeoutException, so the
+      // greeting can no longer complete it: the connection has never been established and there
+      // is no close event to fire.
+      state.set(State.CLOSED);
+      channel = null;
+      connectPromise = null;
+      return;
+    }
     switch (lastCloseReason) {
       case CLOSED_BY_SHUTDOWN:
         message = lastCloseReason.getMessage();
@@ -377,13 +404,14 @@ private synchronized void onChannelClose(ChannelFuture f) {
     if (notify) {
       failConnectPromise(ex);
     }
+    final boolean fireEvent = notify && established;
 
     channel = null;
     connectPromise = null;
     greeting = null;
     state.set(State.CLOSED);
 
-    if (notify) {
+    if (fireEvent) {
       fire(event, ex);
     }
   }
@@ -416,23 +444,42 @@ private void fire(ConnectionCloseEvent event, Throwable ex) {
     }
   }
 
-  private void scheduleGreetingTimeout(CompletableFuture<Greeting> promise, long timeoutMs) {
+  private void scheduleGreetingTimeout(
+      CompletableFuture<Greeting> promise, InetSocketAddress address, long timeoutMs) {
     log.debug("Schedule timeout for connect in milliseconds: {}", timeoutMs);
     timeoutHandler =
         this.timerService.newTimeout(
             timeoutHandler -> {
-              if (promise.isDone()) {
-                return;
-              }
-              if (state.get() == State.CONNECTING || state.get() == State.WAITING) {
-                channel.pipeline().close();
-                promise.completeExceptionally(new TimeoutException("Connection timeout"));
+              // Fail the promise before closing the channel: if the greeting has already
+              // completed it, the connection is established and must not be closed here.
+              if (promise.completeExceptionally(
+                  new TimeoutException(String.format(ERR_GREETING_TIMEOUT, address, timeoutMs)))) {
+                closeOnGreetingTimeout(promise);
               }
             },
             timeoutMs,
             TimeUnit.MILLISECONDS);
   }
 
+  /**
+   * Closes the channel of a connection whose greeting promise has been failed by the timeout.
+   *
+   * <p>The close reason is set before the channel is closed, so the close is not reported as a
+   * close by server: the connection has never been established. A timer of an earlier attempt does
+   * nothing: its channel is already closed, and the current one belongs to another attempt.
+   *
+   * @param promise greeting promise of the attempt the timer was scheduled for
+   */
+  private synchronized void closeOnGreetingTimeout(CompletableFuture<Greeting> promise) {
+    if (promise != greetingPromise) {
+      return;
+    }
+    if (state.get() != State.CONNECTING && state.get() != State.WAITING) {
+      return;
+    }
+    closeChannel(CloseReason.CLOSED_BY_TIMEOUT);
+  }
+
   private void cancelGreetingTimeout() {
     if (timeoutHandler == null) {
       return;

tarantool-core/src/test/java/io/tarantool/core/ShutdownWatcherRegistrationTest.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core;
+
+import java.lang.reflect.Field;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.tarantool.core.connection.ConnectionFactory;
+
+/** Reconnect attempts must not add a new {@code box.shutdown} callback each time. */
+public class ShutdownWatcherRegistrationTest {
+
+  private static final int ATTEMPTS = 5;
+
+  private EventLoopGroup group;
+  private Timer timer;
+  private IProtoClientImpl client;
+
+  @BeforeEach
+  public void setUp() {
+    group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory());
+    timer = new HashedWheelTimer();
+    Bootstrap bootstrap =
+        new Bootstrap()
+            .group(group)
+            .channel(NioSocketChannel.class)
+            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
+    client = new IProtoClientImpl(new ConnectionFactory(bootstrap, timer), timer);
+  }
+
+  @AfterEach
+  public void tearDown() throws Exception {
+    client.close();
+    timer.stop();
+    group.shutdownGracefully().get();
+  }
+
+  @Test
+  public void test_failedConnects_shouldRegisterShutdownCallbackOnce() throws Exception {
+    InetSocketAddress refused = closedPortAddress();
+
+    for (int i = 0; i < ATTEMPTS; i++) {
+      CompletableFuture<Void> attempt = client.connect(refused, 1000, true);
+      assertThrows(ExecutionException.class, () -> attempt.get(5, TimeUnit.SECONDS));
+    }
+
+    assertEquals(1, shutdownCallbacksCount());
+  }
+
+  private static InetSocketAddress closedPortAddress() throws Exception {
+    try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) {
+      return new InetSocketAddress(socket.getInetAddress(), socket.getLocalPort());
+    }
+  }
+
+  // The callbacks are not exposed by the API, so the test reads them directly.
+  @SuppressWarnings("unchecked")
+  private int shutdownCallbacksCount() throws Exception {
+    Field watchersField = IProtoClientImpl.class.getDeclaredField("watchers");
+    watchersField.setAccessible(true);
+    Map<String, Watcher> watchers = (Map<String, Watcher>) watchersField.get(client);
+    Field callbacksField = Watcher.class.getDeclaredField("callbacks");
+    callbacksField.setAccessible(true);
+    return ((List<?>) callbacksField.get(watchers.get("box.shutdown"))).size();
+  }
+}