refactor(pooling): make connectFuture and reconnectTask atomic in PoolEntry

Replace the entry-monitor `synchronized` blocks that guarded `connectFuture`
and `reconnectTask` with `AtomicReference` and CAS/`getAndSet` operations.
The four `synchronized (this)` blocks in `internalConnect()`, `shutdown()`,
`handleConnectError()`, and `connectAfter()` collapse to atomic field
operations; the `synchronized` keyword on `stopReconnectTask()` is dropped.

Behaviour preserved:
* `internalConnect()` fast-path check, then `compareAndSet(null, cf)` —
  the loser's wasted build matches the previous double-checked-locking
  race; the loser's `cf.whenComplete` is not registered, same as before.
* `connectAfter()` schedules the new task outside the lock and uses
  `getAndSet` to atomically swap it in, incrementing `reconnecting` only
  on the first install and cancelling the displaced task otherwise.
* `stopReconnectTask()` and the `connectFuture.set(null)` sites become
  lock-free.

ABBA deadlock reasoning in `shutdown()` JavaDoc is reframed: with no
entry monitor to release, the field mutations are atomic/volatile and
`client.close()` is just a method call that the Netty close-callback
path is free to interleave with.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: dkasimovskiy

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

@@ -10,6 +10,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
 
@@ -144,7 +145,8 @@ final class PoolEntry {
    * <p>It will be returned to all out clients wanting to obtain this client. It is recreated only
    * if client is reconnected.
    */
-  private CompletableFuture<IProtoClient> connectFuture;
+  private final AtomicReference<CompletableFuture<IProtoClient>> connectFuture =
+      new AtomicReference<>();
 
   /** Last heartbeat state/event. */
   private volatile HeartbeatEvent lastHeartbeatEvent;
@@ -153,7 +155,7 @@ final class PoolEntry {
   private volatile Timeout heartbeatTask;
 
   /** Reconnection task. */
-  private Timeout reconnectTask;
+  private final AtomicReference<Timeout> reconnectTask = new AtomicReference<>();
 
   /** Flag signaling if heartbeat started or not. */
   private volatile boolean isHeartbeatStarted;
@@ -385,9 +387,9 @@ public void close() {
   /**
    * 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
+   * <p>Field mutations on {@link #connectFuture} and the heartbeat task are atomic/volatile, so
+   * {@code client.close()} (which acquires the {@code ConnectionImpl} monitor) and the close-event
+   * emit run without holding the entry monitor — keeping the call lock-free avoids 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.
    *
@@ -410,10 +412,8 @@ public void close() {
    * method for the same entry.
    */
   public void shutdown() {
-    synchronized (this) {
-      connectFuture = null;
-      stopHeartbeat();
-    }
+    connectFuture.set(null);
+    stopHeartbeat();
     try {
       client.close();
     } catch (Exception e) {
@@ -479,16 +479,18 @@ public void stopHeartbeat() {
   /**
    * Internal method used by reconnect task and public connect.
    *
-   * <p>See {@link #shutdown()} for the monitor-ordering reasoning; {@code client.connect()} runs
-   * outside the entry monitor for the same reason.
+   * <p>{@code client.connect()} runs outside any entry lock for the same reason as in {@link
+   * #shutdown()} — see there for the monitor-ordering reasoning. The {@link
+   * AtomicReference#compareAndSet} on {@link #connectFuture} atomically installs the freshly built
+   * future or returns the winner's future to the loser, so only one {@code
+   * whenComplete(onConnectComplete)} is registered per in-flight connect.
    *
    * @return {@link java.util.concurrent.CompletableFuture} with client
    */
   private CompletableFuture<IProtoClient> internalConnect() {
-    synchronized (this) {
-      if (connectFuture != null) {
-        return connectFuture;
-      }
+    CompletableFuture<IProtoClient> existing = connectFuture.get();
+    if (existing != null) {
+      return existing;
     }
     log.info("connect {}/{}", tag, index);
     LongTaskTimer.Sample timer = startTimer(connectTime);
@@ -506,15 +508,12 @@ private CompletableFuture<IProtoClient> internalConnect() {
                   return client.ping(firstPingOpts);
                 })
             .thenApply(r -> client);
-    synchronized (this) {
-      if (connectFuture != null) {
-        return connectFuture;
-      }
-      connectFuture = cf;
+    if (connectFuture.compareAndSet(null, cf)) {
       isShutdown.set(false);
+      cf.whenComplete(this::onConnectComplete);
+      return cf;
     }
-    cf.whenComplete(this::onConnectComplete);
-    return cf;
+    return connectFuture.get();
   }
 
   /**
@@ -554,9 +553,7 @@ private void handleConnectError(Object r, Throwable exc) {
       return;
     }
     Throwable failure = exc.getCause() != null ? exc.getCause() : exc;
-    synchronized (this) {
-      connectFuture = null;
-    }
+    connectFuture.set(null);
     log.error("connect error {}/{}: {}", tag, index, failure.toString());
     emit(listener -> listener.onConnectionFailed(tag, index, failure));
     lock();
@@ -566,18 +563,17 @@ private void handleConnectError(Object r, Throwable exc) {
 
   /** Reconnect task scheduler. */
   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);
+    log.info("reconnect {}/{} after {} ms", tag, index, reconnectAfter);
+    Timeout newTask =
+        timerService.newTimeout(
+            timeout -> internalConnect(), reconnectAfter, TimeUnit.MILLISECONDS);
+    Timeout old = reconnectTask.getAndSet(newTask);
+    if (old == null) {
+      reconnecting.incrementAndGet();
+    } else {
+      // existing task is being replaced; the existing increment in `reconnecting` carries over
+      // to the new task, so no counter change is needed here.
+      old.cancel();
     }
     emit(listener -> listener.onReconnectScheduled(tag, index, reconnectAfter));
   }
@@ -750,11 +746,11 @@ private void incHeartbeatCounters(int fail) {
   }
 
   /** Stops reconnecting task if it is active. */
-  private synchronized void stopReconnectTask() {
-    if (reconnectTask != null) {
+  private void stopReconnectTask() {
+    Timeout existing = reconnectTask.getAndSet(null);
+    if (existing != null) {
       reconnecting.decrementAndGet();
-      reconnectTask.cancel();
-      reconnectTask = null;
+      existing.cancel();
     }
   }