Bound exclusive lock acquisition in SharedLock to fail fast on a dead…

src/main/java/io/lettuce/core/cluster/ClusterNodeEndpoint.java

@@ -51,11 +51,28 @@ public CompletableFuture<Void> closeAsync() {
 
         logger.debug("{} closeAsync()", logPrefix());
 
+        Throwable drainError = null;
         if (clusterChannelWriter != null) {
-            retriggerCommands(doExclusive(this::drainCommands));
+            try {
+                retriggerCommands(doExclusive(this::drainCommands));
+            } catch (Throwable t) {
+                drainError = t;
+            }
+        }
+
+        CompletableFuture<Void> superFuture = super.closeAsync();
+
+        if (drainError != null) {
+            final Throwable ex = drainError;
+            return superFuture.handle((v, t) -> {
+                if (t != null && t != ex) {
+                    ex.addSuppressed(t);
+                }
+                throw (ex instanceof RuntimeException) ? (RuntimeException) ex : new RedisException(ex);
+            });
         }
 
-        return super.closeAsync();
+        return superFuture;
     }
 
     protected void retriggerCommands(Collection<RedisCommand<?, ?, ?>> commands) {

src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java

@@ -605,14 +605,38 @@ public CompletableFuture<Void> closeAsync() {
                 connectionWatchdog.prepareClose();
             }
 
-            cancelBufferedCommands("Close");
+            Throwable cancelError = null;
+            try {
+                cancelBufferedCommands("Close");
+            } catch (Throwable t) {
+                cancelError = t;
+                try {
+                    cancelCommands("Close", drainCommands(), RedisCommand::cancel);
+                } catch (Throwable drainEx) {
+                    cancelError.addSuppressed(drainEx);
+                }
+            }
 
             Channel channel = getOpenChannel();
 
             if (channel != null) {
-                Futures.adapt(channel.close(), closeFuture);
+                if (cancelError != null) {
+                    final Throwable ex = cancelError;
+                    channel.close().addListener(future -> {
+                        if (!future.isSuccess() && future.cause() != null) {
+                            ex.addSuppressed(future.cause());
+                        }
+                        closeFuture.completeExceptionally(ex);
+                    });
+                } else {
+                    Futures.adapt(channel.close(), closeFuture);
+                }
             } else {
-                closeFuture.complete(null);
+                if (cancelError != null) {
+                    closeFuture.completeExceptionally(cancelError);
+                } else {
+                    closeFuture.complete(null);
+                }
             }
         }
 

src/main/java/io/lettuce/core/protocol/SharedLock.java

@@ -1,11 +1,14 @@
 package io.lettuce.core.protocol;
 
+import java.time.Duration;
 import java.util.WeakHashMap;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Supplier;
 
+import io.lettuce.core.RedisException;
 import io.lettuce.core.internal.LettuceAssert;
 
 /**
@@ -43,6 +46,28 @@ class SharedLock {
 
     private volatile Thread exclusiveLockOwner;
 
+    private static final Duration DEFAULT_EXCLUSIVE_LOCK_TIMEOUT = Duration.ofSeconds(30);
+
+    private final long exclusiveLockTimeoutNanos;
+
+    /**
+     * Create a {@link SharedLock} with the {@link #DEFAULT_EXCLUSIVE_LOCK_TIMEOUT default} exclusive-lock timeout.
+     */
+    SharedLock() {
+        this(DEFAULT_EXCLUSIVE_LOCK_TIMEOUT);
+    }
+
+    /**
+     * Create a {@link SharedLock} with a configurable exclusive-lock acquisition timeout.
+     *
+     * @param exclusiveLockTimeout the maximum time an exclusive acquisition waits before failing fast with a
+     *        {@link RedisException}; must not be {@code null}.
+     */
+    SharedLock(Duration exclusiveLockTimeout) {
+        LettuceAssert.notNull(exclusiveLockTimeout, "Exclusive lock timeout must not be null");
+        this.exclusiveLockTimeoutNanos = exclusiveLockTimeout.toNanos();
+    }
+
     /**
      * Wait for stateLock and increment writers. Will wait if stateLock is locked and if writer counter is negative.
      */
@@ -108,12 +133,13 @@ <T> T doExclusive(Supplier<T> supplier) {
 
         LettuceAssert.notNull(supplier, "Supplier must not be null");
 
-        lock.lock();
+        long deadline = System.nanoTime() + exclusiveLockTimeoutNanos;
+        acquireExclusiveGuard(deadline);
         try {
 
             try {
 
-                lockWritersExclusive();
+                lockWritersExclusive(deadline);
                 return supplier.get();
             } finally {
                 unlockWritersExclusive();
@@ -124,10 +150,55 @@ <T> T doExclusive(Supplier<T> supplier) {
     }
 
     /**
-     * Wait for stateLock and no writers. Must be used in an outer {@code synchronized} block to prevent interleaving with other
-     * methods using writers. Sets writers to a negative value to create a lock for {@link #incrementWriters()}.
+     * Acquire the guarding {@link #lock} for an exclusive operation, bounded by {@code deadline}. If the lock was leaked by a
+     * thread that died inside the guarded region (issue #3804) a plain {@code lock.lock()} would park the caller - frequently a
+     * Netty event-loop thread - forever. Fail fast with a {@link RedisException} instead so the endpoint can be rebuilt. The
+     * acquisition remains reentrant, so a thread that already holds the lock re-acquires immediately.
+     * <p>
+     * Acquisition continues in a loop bounded by {@code deadline} while catching {@link InterruptedException} and restoring the
+     * interrupt flag afterwards, so callers with their interrupt flag already set (such as during
+     * {@code close()}/{@code closeAsync()}) do not fail an otherwise successful exclusive operation solely due to the interrupt
+     * under brief contention.
+     *
+     * @param deadline the {@link System#nanoTime()} timestamp by which the acquisition must complete.
      */
-    private void lockWritersExclusive() {
+    private void acquireExclusiveGuard(long deadline) {
+
+        boolean interrupted = false;
+        boolean acquired = lock.tryLock();
+        try {
+            while (!acquired) {
+                long timeoutNanos = deadline - System.nanoTime();
+                if (timeoutNanos <= 0) {
+                    break;
+                }
+                try {
+                    acquired = lock.tryLock(timeoutNanos, TimeUnit.NANOSECONDS);
+                } catch (InterruptedException e) {
+                    interrupted = true;
+                }
+            }
+        } finally {
+            if (interrupted) {
+                Thread.currentThread().interrupt();
+            }
+        }
+
+        if (!acquired) {
+            throw new RedisException("Timed out after " + TimeUnit.NANOSECONDS.toMillis(exclusiveLockTimeoutNanos)
+                    + "ms acquiring the exclusive SharedLock; the lock holder likely died inside the guarded region. "
+                    + "The endpoint must be rebuilt to recover.");
+        }
+    }
+
+    /**
+     * Wait for stateLock and no writers, bounded by {@code deadline}. Must be used in an outer {@code synchronized} block to
+     * prevent interleaving with other methods using writers. Sets writers to a negative value to create a lock for
+     * {@link #incrementWriters()}.
+     *
+     * @param deadline the {@link System#nanoTime()} timestamp by which writer draining must complete.
+     */
+    private void lockWritersExclusive(long deadline) {
 
         if (exclusiveLockOwner == Thread.currentThread()) {
             WRITERS.decrementAndGet(this);
@@ -138,12 +209,36 @@ private void lockWritersExclusive() {
         try {
             for (;;) {
 
-                // allow reentrant exclusive lock by comparing writers count and threadWriters count
+                // allow reentrant exclusive lock by comparing writers count and threadWriters
+                // count
                 int threadWriterCount = getThreadWriterCount();
                 if (WRITERS.compareAndSet(this, threadWriterCount, -1)) {
                     exclusiveLockOwner = Thread.currentThread();
                     return;
                 }
+
+                // A leaked shared writer (a thread that died before decrementWriters()) keeps
+                // the writer count above this thread's own count, so the CAS above can never
+                // succeed.
+                // Or if exclusive mode was abandoned (writers is negative, e.g. a previous
+                // exclusive holder died),
+                // the CAS also cannot succeed. Bound the spin and fail fast instead of burning
+                // a CPU core forever
+                // (issues #3804, #3880).
+                if (System.nanoTime() - deadline >= 0) {
+                    long currentWriters = WRITERS.get(this);
+                    if (currentWriters < 0) {
+                        throw new RedisException("Timed out after " + TimeUnit.NANOSECONDS.toMillis(exclusiveLockTimeoutNanos)
+                                + "ms acquiring the exclusive SharedLock; the exclusive lock was likely abandoned by a previous holder. "
+                                + "The endpoint must be rebuilt to recover.");
+                    }
+
+                    throw new RedisException("Timed out after " + TimeUnit.NANOSECONDS.toMillis(exclusiveLockTimeoutNanos)
+                            + "ms waiting for " + currentWriters + " shared writer(s) to drain while acquiring the "
+                            + "exclusive SharedLock; a shared writer was likely leaked. The endpoint must be rebuilt to recover.");
+                }
+
+                Thread.yield();
             }
         } finally {
             lock.unlock();

src/test/java/io/lettuce/core/cluster/ClusterNodeEndpointUnitTests.java

@@ -6,7 +6,9 @@
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
+import java.time.Duration;
 import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
@@ -145,6 +147,34 @@ void closeWithBufferedCommandsFails() {
         assertThatThrownBy(() -> TestFutures.awaitOrTimeout(command)).isInstanceOf(RedisException.class);
     }
 
+    @Test
+    void closeAsyncClosesSuperWhenSharedLockTimesOut() throws Exception {
+
+        Class<?> sharedLockClass = Class.forName("io.lettuce.core.protocol.SharedLock");
+        java.lang.reflect.Constructor<?> ctor = sharedLockClass.getDeclaredConstructor(Duration.class);
+        ctor.setAccessible(true);
+        Object sharedLock = ctor.newInstance(Duration.ofMillis(50));
+        ReflectionTestUtils.setField(sut, "sharedLock", sharedLock);
+
+        java.lang.reflect.Method incrementWriters = sharedLockClass.getDeclaredMethod("incrementWriters");
+        incrementWriters.setAccessible(true);
+
+        Thread leaker = new Thread(() -> {
+            try {
+                incrementWriters.invoke(sharedLock);
+            } catch (Exception ignored) {
+            }
+        });
+        leaker.start();
+        leaker.join(1000);
+
+        CompletableFuture<Void> closeFuture = sut.closeAsync();
+        assertThat(closeFuture).isNotNull();
+        assertThat(sut.isClosed()).isTrue();
+        assertThat(closeFuture.isCompletedExceptionally()).isTrue();
+        assertThatThrownBy(closeFuture::join).hasCauseInstanceOf(RedisException.class);
+    }
+
     private void prepareNewEndpoint() {
         sut = new ClusterNodeEndpoint(clientOptions, clientResources, clusterChannelWriter);
         disconnectedBuffer = (Queue) ReflectionTestUtils.getField(sut, "disconnectedBuffer");

src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java

@@ -5,7 +5,9 @@
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
+import java.lang.reflect.Field;
 import java.nio.channels.ClosedChannelException;
+import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Queue;
@@ -47,6 +49,7 @@
 import io.netty.channel.DefaultChannelPromise;
 import io.netty.channel.EventLoop;
 import io.netty.handler.codec.EncoderException;
+import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.concurrent.ImmediateEventExecutor;
 
 /**
@@ -378,6 +381,99 @@ void retryListenerDoesNotRetryCompletedCommands() {
         verify(channel, never()).writeAndFlush(command);
     }
 
+    @Test
+    void closeAsyncSucceedsWhenThreadIsInterrupted() {
+
+        when(channel.close()).thenReturn(mock(ChannelFuture.class));
+        sut.notifyChannelActive(channel);
+
+        Thread.currentThread().interrupt();
+        try {
+            CompletableFuture<Void> closeFuture = sut.closeAsync();
+            assertThat(closeFuture).isNotNull();
+            assertThat(sut.isClosed()).isTrue();
+            assertThat(Thread.currentThread().isInterrupted()).isTrue();
+        } finally {
+            Thread.interrupted();
+        }
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    void closeAsyncCompletesExceptionallyWhenSharedLockTimesOut() throws Exception {
+
+        SharedLock sharedLock = new SharedLock(Duration.ofMillis(50));
+        Field sharedLockField = DefaultEndpoint.class.getDeclaredField("sharedLock");
+        sharedLockField.setAccessible(true);
+        sharedLockField.set(sut, sharedLock);
+
+        ChannelFuture channelCloseFuture = mock(ChannelFuture.class);
+        when(channel.close()).thenReturn(channelCloseFuture);
+        sut.notifyChannelActive(channel);
+
+        Thread leaker = new Thread(sharedLock::incrementWriters);
+        leaker.start();
+        leaker.join(1000);
+
+        CompletableFuture<Void> closeFuture = sut.closeAsync();
+        assertThat(closeFuture).isNotNull();
+        assertThat(sut.isClosed()).isTrue();
+
+        verify(channel).close();
+
+        ArgumentCaptor<GenericFutureListener> captor = ArgumentCaptor.forClass(GenericFutureListener.class);
+        verify(channelCloseFuture).addListener(captor.capture());
+        when(channelCloseFuture.isSuccess()).thenReturn(true);
+        captor.getValue().operationComplete(channelCloseFuture);
+
+        assertThat(closeFuture).isCompletedExceptionally();
+        assertThatThrownBy(closeFuture::join).hasCauseInstanceOf(RedisException.class).hasMessageContaining("shared writer");
+
+        assertThat(sut.closeAsync()).isSameAs(closeFuture);
+    }
+
+    @Test
+    void closeAsyncCompletesExceptionallyWhenSharedLockTimesOutAndChannelIsNull() throws Exception {
+
+        SharedLock sharedLock = new SharedLock(Duration.ofMillis(50));
+        Field sharedLockField = DefaultEndpoint.class.getDeclaredField("sharedLock");
+        sharedLockField.setAccessible(true);
+        sharedLockField.set(sut, sharedLock);
+
+        Thread leaker = new Thread(sharedLock::incrementWriters);
+        leaker.start();
+        leaker.join(1000);
+
+        CompletableFuture<Void> closeFuture = sut.closeAsync();
+        assertThat(closeFuture).isNotNull();
+        assertThat(sut.isClosed()).isTrue();
+        assertThat(closeFuture).isCompletedExceptionally();
+        assertThatThrownBy(closeFuture::join).hasCauseInstanceOf(RedisException.class).hasMessageContaining("shared writer");
+    }
+
+    @Test
+    void closeAsyncCancelsBufferedCommandsWhenSharedLockTimesOutAndChannelIsNull() throws Exception {
+
+        SharedLock sharedLock = new SharedLock(Duration.ofMillis(50));
+        Field sharedLockField = DefaultEndpoint.class.getDeclaredField("sharedLock");
+        sharedLockField.setAccessible(true);
+        sharedLockField.set(sut, sharedLock);
+
+        RedisCommand<String, String, String> command = new AsyncCommand<>(
+                new Command<>(CommandType.PING, new StatusOutput<>(StringCodec.UTF8)));
+        sut.write(command);
+
+        Thread leaker = new Thread(sharedLock::incrementWriters);
+        leaker.start();
+        leaker.join(1000);
+
+        CompletableFuture<Void> closeFuture = sut.closeAsync();
+        assertThat(closeFuture).isNotNull();
+        assertThat(sut.isClosed()).isTrue();
+        assertThat(closeFuture).isCompletedExceptionally();
+        assertThat(command.isCancelled()).isTrue();
+    }
+
     @Test
     void shouldWrapActivationCommands() {