fix(pooling): avoid PoolEntry/ConnectionImpl deadlock by calling client.close()/emit outside monitor

The previous attempt to fix race conditions in PoolEntry made shutdown(),
close() and handleConnectError() synchronized on the PoolEntry monitor while
they invoked client.close() (which acquires the ConnectionImpl monitor) and
emitted pool events. The Netty close-event path runs the reverse order:
ConnectionImpl.onChannelClose() (synchronized on ConnectionImpl) calls back
into PoolEntry.handleConnectError(), which then blocks on the PoolEntry
monitor. This produced an ABBA deadlock between the HashedWheelTimer worker
and the Netty event loop, which manifested as a permanent hang in
DistributingRoundRobinBalancerTest (the @timeout(25) SameThread variant
cannot interrupt CompletableFuture.join()).

The thread dump of the hung JVM confirmed:
  - 'pool-N-thread-1' BLOCKED in ConnectionImpl.closeChannel, holding PoolEntry
  - 'multiThreadIoEventLoopGroup-2-1' BLOCKED in PoolEntry.handleConnectError, holding ConnectionImpl

Fix: narrow the entry-monitor critical sections to field mutations only.
client.close() and emit(...) are now called outside the monitor, breaking
the lock-ordering cycle.

Also fix a race in internalConnect(): assign connectFuture to the local chain
BEFORE attaching whenComplete, so an inline failure completing before the
method returns cannot have its nulling of connectFuture in
handleConnectError() overwritten by a stale failed future.

Verified locally:
  - DistributingRoundRobinBalancerTest: 5/5 on 2.11.8, 5/5 on 3.5.0, no hangs
  - testDistributingRoundRobinStartWithStuckNodeA (reproducer): 5/5 stable
  - ConnectionPoolReconnectsTest + ConnectionPoolHeartbeatTest: 5/5 on 2.11.8, 3/3 on 3.5.0
  - IProtoClientPoolTest + ConnectionPoolTest (unit): 32/32

Author: opencode

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

@@ -345,15 +345,25 @@ public boolean isLocked() {
   }
 
   /** Closes client and stops heartbeat and reconnect tasks if started. */
-  public synchronized void close() {
+  public void close() {
     stopReconnectTask();
     shutdown();
   }
 
-  /** Closes client and stops heartbeat task is started. */
-  public synchronized void shutdown() {
-    connectFuture = null;
-    stopHeartbeat();
+  /**
+   * 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() {
+    synchronized (this) {
+      connectFuture = null;
+      stopHeartbeat();
+    }
     try {
       client.close();
     } catch (Exception e) {
@@ -422,6 +432,11 @@ public void stopHeartbeat() {
   /**
    * Internal method used by reconnect task and public connect.
    *
+   * <p>{@code connectFuture} is assigned to the local chain <em>before</em> {@code whenComplete} is
+   * attached, so that an inline failure completing before this method returns cannot have its
+   * nulling of {@code connectFuture} in {@link #handleConnectError(Object, Throwable)} overwritten
+   * by a stale failed future.
+   *
    * @return {@link java.util.concurrent.CompletableFuture} with client
    */
   private synchronized CompletableFuture<IProtoClient> internalConnect() {
@@ -433,7 +448,7 @@ private synchronized CompletableFuture<IProtoClient> internalConnect() {
     CompletableFuture<?> future =
         client.connect(group.getAddress(), connectTimeout, gracefulShutdown);
     String user = group.getUser();
-    connectFuture =
+    CompletableFuture<IProtoClient> cf =
         future
             .thenCompose(
                 greeting -> {
@@ -443,8 +458,9 @@ private synchronized CompletableFuture<IProtoClient> internalConnect() {
                   }
                   return client.ping(firstPingOpts);
                 })
-            .thenApply(r -> client)
-            .whenComplete(this::onConnectComplete);
+            .thenApply(r -> client);
+    connectFuture = cf;
+    cf.whenComplete(this::onConnectComplete);
     return connectFuture;
   }
 
@@ -477,19 +493,25 @@ private void onConnectComplete(Object r, Throwable exc) {
   /**
    * Handler for connection close.
    *
+   * <p>Only the {@code connectFuture} reset is performed under the entry monitor. The listener
+   * emit, {@link #lock()}, {@link #shutdown()} and {@link #connectAfter()} calls are made outside
+   * the monitor; {@code shutdown()} will itself acquire {@code ConnectionImpl} via {@code
+   * client.close()}, and the close-event path takes the same monitor — holding the entry monitor
+   * across them would deadlock.
+   *
    * @param r connection instance
    * @param exc exception which led to connection close
    */
   private void handleConnectError(Object r, Throwable exc) {
     if (exc == null) {
       return;
     }
+    Throwable failure = exc.getCause() != null ? exc.getCause() : exc;
     synchronized (this) {
-      Throwable failure = exc.getCause() != null ? exc.getCause() : exc;
       connectFuture = null;
-      log.error("connect error {}/{}: {}", tag, index, failure.toString());
-      emit(listener -> listener.onConnectionFailed(tag, index, failure));
     }
+    log.error("connect error {}/{}: {}", tag, index, failure.toString());
+    emit(listener -> listener.onConnectionFailed(tag, index, failure));
     lock();
     shutdown();
     connectAfter();
@@ -500,15 +522,17 @@ private void connectAfter() {
     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);
-      emit(listener -> listener.onReconnectScheduled(tag, index, reconnectAfter));
     }
+    emit(listener -> listener.onReconnectScheduled(tag, index, reconnectAfter));
   }
 
   /**