RedisPublisher must not drop an error when the state changes concurrently #3942 (#3943)

* RedisPublisher must not drop an error when the state changes concurrently #3942

State.onError moved the subscription to COMPLETED with a single CAS in the
loop condition. When the subscribing thread changed the state at the same
moment (DEMAND -> READING -> DEMAND while reading), the CAS failed, the loop
exited and the error was never signalled, so the subscriber waited forever.
onError now retries the CAS until the subscription is completed.

* Use explicit Mockito imports in RedisPublisherUnitTests #3942

---------

Co-authored-by: Aleksandar Todorov <a_t_todorov@yahoo.com>

Author: zenios

src/main/java/io/lettuce/core/RedisPublisher.java

@@ -713,15 +713,21 @@ void onAllDataRead(RedisSubscription<?> subscription) {
         void onError(RedisSubscription<?> subscription, Throwable t) {
 
             State state;
-            while ((state = subscription.state()) != COMPLETED && subscription.changeState(state, COMPLETED)) {
+            while ((state = subscription.state()) != COMPLETED) {
+
+                // The subscribing thread may move the state (e.g. DEMAND -> READING -> DEMAND) between the read above and
+                // the CAS. Retry instead of dropping the error, otherwise the subscriber is never terminated.
+                if (!subscription.changeState(state, COMPLETED)) {
+                    continue;
+                }
 
                 readData(subscription);
 
                 Subscriber<?> subscriber = subscription.subscriber;
                 if (subscriber != null) {
                     subscriber.onError(t);
-                    return;
                 }
+                return;
             }
         }
 

src/test/java/io/lettuce/core/RedisPublisherUnitTests.java

@@ -0,0 +1,62 @@
+package io.lettuce.core;
+
+import static io.lettuce.TestTags.UNIT_TEST;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.reactivestreams.Subscriber;
+
+import io.lettuce.core.api.StatefulConnection;
+import io.lettuce.core.codec.StringCodec;
+import io.lettuce.core.output.StatusOutput;
+import io.lettuce.core.protocol.Command;
+import io.lettuce.core.protocol.CommandType;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+
+/**
+ * Unit tests for {@link RedisPublisher}.
+ */
+@Tag(UNIT_TEST)
+class RedisPublisherUnitTests {
+
+    @Test
+    void shouldSignalErrorWhenStateChangesConcurrently() {
+
+        Command<String, String, String> command = new Command<>(CommandType.EVALSHA, new StatusOutput<>(StringCodec.UTF8),
+                null);
+
+        // Simulates the subscribing thread flipping DEMAND -> READING right before the error's CAS to COMPLETED.
+        RedisPublisher.RedisSubscription<String> subscription = new RedisPublisher.RedisSubscription<String>(
+                mock(StatefulConnection.class), command, false, ImmediateEventExecutor.INSTANCE) {
+
+            private boolean raced;
+
+            @Override
+            boolean changeState(RedisPublisher.State oldState, RedisPublisher.State newState) {
+
+                if (!raced && oldState == RedisPublisher.State.DEMAND && newState == RedisPublisher.State.COMPLETED) {
+                    raced = true;
+                    super.changeState(RedisPublisher.State.DEMAND, RedisPublisher.State.READING);
+                }
+                return super.changeState(oldState, newState);
+            }
+
+        };
+
+        Subscriber<String> subscriber = mock(Subscriber.class);
+        subscription.subscribe(subscriber);
+        subscription.changeState(RedisPublisher.State.NO_DEMAND, RedisPublisher.State.DEMAND);
+
+        RedisCommandExecutionException error = new RedisCommandExecutionException("NOSCRIPT No matching script");
+        subscription.onError(error);
+
+        verify(subscriber).onSubscribe(any());
+        verify(subscriber).onError(error);
+        verifyNoMoreInteractions(subscriber);
+    }
+
+}