fix: handle failed connection attempts once

CHANGELOG.md

@@ -2,6 +2,29 @@
 
 ## [Unreleased]
 
+### Bug fixes
+
+- 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`, 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.
+
 ## [1.7.3] - 2026-10-30
 
 ### Pool

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,8 @@ 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_CONNECTION_TIMEOUT = "Connection timeout";
   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 +122,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 +204,16 @@ public synchronized CompletableFuture<Greeting> connect(InetSocketAddress addres
       log.debug("{} is connecting to {}", this, address);
 
       CompletableFuture<Greeting> promise = new CompletableFuture<>();
+      greetingPromise = promise;
+      established = false;
       scheduleGreetingTimeout(promise, 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 +375,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 +403,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);
     }
   }
@@ -421,18 +448,35 @@ private void scheduleGreetingTimeout(CompletableFuture<Greeting> promise, long t
     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(ERR_CONNECTION_TIMEOUT))) {
+                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();
+  }
+}

tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java

@@ -8,7 +8,10 @@
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.net.ConnectException;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
 import java.net.SocketException;
 import java.nio.channels.ClosedChannelException;
 import java.time.Duration;
@@ -397,18 +400,69 @@ public void testConnectByMultipleThreads() throws InterruptedException {
     assertEquals(CONCURRENT_THREADS_COUNT, success);
   }
 
+  /*
+   * Server accepts the connection but does not send the greeting. The connect future fails only
+   * with TimeoutException, and no CLOSE_BY_REMOTE event is fired: the client closes the channel
+   * itself.
+   */
   @Test
   @DisabledIfEnvironmentVariable(named = "TARANTOOL_VERSION", matches = "3.*")
   public void testConnectWithGreetingTimeout() throws InterruptedException {
-    Connection client = factory.create();
+    AtomicReference<Throwable> closedByRemote = new AtomicReference<>();
+    Connection client =
+        factory
+            .create()
+            .onClose(ConnectionCloseEvent.CLOSE_BY_REMOTE, (c, ex) -> closedByRemote.set(ex));
     InetSocketAddress address = new InetSocketAddress(tt.getHost(), tt.getMappedPort(3306));
     CompletableFuture<Greeting> future = client.connect(address, 1_000);
     Exception ex = assertThrows(CompletionException.class, future::join);
     Throwable cause = ex.getCause();
-    assertTrue(
-        cause instanceof TimeoutException || cause instanceof ConnectionClosedException,
-        "Exception should be either TimeoutException or ConnectionClosedException");
+    assertEquals(TimeoutException.class, cause.getClass());
     assertEquals("Connection timeout", cause.getMessage());
+    // Without the fix the close event follows the timeout within milliseconds.
+    Thread.sleep(500);
+    assertNull(closedByRemote.get(), "greeting timeout is reported as closed by server");
+  }
+
+  /*
+   * The server closes the first attempt before the greeting; the second attempt waits for the
+   * greeting of a silent server. The greeting timer of the first attempt must not close the second
+   * one: it fails with TimeoutException after its own timeout.
+   */
+  @Test
+  @DisabledIfEnvironmentVariable(named = "TARANTOOL_VERSION", matches = "3.*")
+  public void testGreetingTimerOfClosedAttemptDoesNotCloseNextAttempt() throws Exception {
+    long timeoutMs = 1_000;
+    long pauseMs = 500;
+    Connection client = factory.create();
+    try (ServerSocket closingServer = new ServerSocket(0, 50, InetAddress.getLoopbackAddress())) {
+      Thread closer =
+          new Thread(
+              () -> {
+                try (Socket socket = closingServer.accept()) {
+                  // closed by try-with-resources before the greeting
+                } catch (IOException ignored) {
+                  // the test fails on its own assertions
+                }
+              });
+      closer.setDaemon(true);
+      closer.start();
+      InetSocketAddress closing =
+          new InetSocketAddress(closingServer.getInetAddress(), closingServer.getLocalPort());
+      assertThrows(CompletionException.class, client.connect(closing, timeoutMs)::join);
+    }
+    Thread.sleep(pauseMs);
+
+    InetSocketAddress silent = new InetSocketAddress(tt.getHost(), tt.getMappedPort(3306));
+    long start = System.nanoTime();
+    CompletableFuture<Greeting> second = client.connect(silent, timeoutMs);
+    Throwable cause = assertThrows(CompletionException.class, second::join).getCause();
+    long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+
+    // Without the fix the timer of the first attempt closes the second one about
+    // timeoutMs - pauseMs after it starts, and it fails as closed by server.
+    assertEquals(TimeoutException.class, cause.getClass(), cause.toString());
+    assertTrue(elapsedMs >= timeoutMs - 100, "second attempt failed after " + elapsedMs + " ms");
   }
 
   @Test

tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java

@@ -501,6 +501,8 @@ public void setReconnectAfter(long reconnectAfter) throws IllegalArgumentExcepti
     if (reconnectAfter <= 0) {
       throw new IllegalArgumentException("reconnect period should be positive number");
     }
+    entries.forEach(
+        (tag, entryGroup) -> entryGroup.forEach(pe -> pe.setReconnectAfter(reconnectAfter)));
     this.reconnectAfter = reconnectAfter;
   }