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

Fix a relocated ABBA deadlock: `PoolEntry.connect()`/`internalConnect()` no longer hold the entry monitor across `client.connect()`, which would deadlock with the Netty close-callback path on the shared `ConnectionImpl` monitor during a close/reconnect overlap.
Fix `onConnectionClosed` accounting: the `PoolEntry` shutdown idempotency flag is now reset per connection generation, so a KILL/reconnect cycle emits one close event per generation instead of suppressing all closes after the first.
`IProtoClientPoolImpl.forEach()` snapshots clients under the pool lock and invokes the action outside it, avoiding `ConcurrentModificationException` under concurrent `setGroups()` without holding the lock across a user callback.

Author: dkasimovskiy

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

@@ -167,13 +167,8 @@ final class PoolEntry {
   private volatile boolean isLocked;
 
   /**
-   * Idempotency flag for {@link #shutdown()}, scoped to a single connection generation.
-   *
-   * <p>Guarantees that the close listener is invoked only once per generation even if both {@link
-   * #close()} and {@link #shutdown()} are called, or shutdown is invoked multiple times due to
-   * overlapping connection error events. It is reset in {@link #internalConnect()} when a new
-   * connection generation begins, so a KILL/reconnect cycle still emits one {@code
-   * onConnectionClosed} per generation.
+   * Per-generation idempotency flag for {@link #shutdown()} close-event emit; reset in {@link
+   * #internalConnect()} when a new connection generation begins.
    */
   private final AtomicBoolean isShutdown = new AtomicBoolean(false);
 
@@ -431,13 +426,8 @@ public void stopHeartbeat() {
   /**
    * Internal method used by reconnect task and public connect.
    *
-   * <p>The {@code client.connect()} chain is built without holding the entry monitor: holding it
-   * across {@code client.connect()} (which takes the {@code ConnectionImpl} monitor) 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.
-   * Concurrent callers are de-duplicated via the {@link #connectFuture} field; if two threads race
-   * past the first check, the second {@code client.connect()} is a no-op because {@code
-   * ConnectionImpl.connect()} only opens a channel on a {@code CLOSED -> CONNECTING} transition.
+   * <p>See {@link #shutdown()} for the monitor-ordering reasoning; {@code client.connect()} runs
+   * outside the entry monitor for the same reason.
    *
    * @return {@link java.util.concurrent.CompletableFuture} with client
    */
@@ -465,8 +455,6 @@ private CompletableFuture<IProtoClient> internalConnect() {
             .thenApply(r -> client);
     synchronized (this) {
       if (connectFuture != null) {
-        // ponytail: lost the race; ConnectionImpl's CLOSED->CONNECTING CAS already de-duped the
-        // channel, so our `cf` wraps the same promise and is safely discarded.
         return connectFuture;
       }
       connectFuture = cf;
@@ -505,10 +493,6 @@ 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
    */

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

@@ -94,10 +94,8 @@ protected void execLua(TarantoolContainer<?> container, String command) {
 
   protected int getActiveConnectionsCount(TarantoolContainer<?> tt) {
     try {
-      // 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.
+      // box.stat.net().CONNECTIONS.current is updated asynchronously by the IProto worker;
+      // the loop's fiber.sleep lets it drain pending connections before we read.
       String lua =
           "local last = box.stat.net().CONNECTIONS.current;"
               + " for i = 1, 50 do"
@@ -119,12 +117,8 @@ protected int getActiveConnectionsCountDelta(TarantoolContainer<?> tt, int basel
   }
 
   /**
-   * 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.
+   * Retries {@link #getActiveConnectionsCount} until it equals {@code expected} — see there for why
+   * a single read is unreliable.
    *
    * @param tt the Tarantool container under test
    * @param expected the expected number of active connections