Repository: redis/lettuce
Pull request #2597 (2024-01-17 11:35:58)

Fix: make sure FIFO order between write and notify channel active
Fix https://github.com/lettuce-io/lettuce-core/issues/2598 

If `write()` comes in between the line `this.channel = channel` and `sharedLock.doExclusive(() -> {}` in `notifyChannelActive()`, the FIFO order is violated since the previous entered commands have not been flushed yet.
See the test case `writeShouldGuaranteeFIFOOrder()`
Make sure that:

- [Y] You have read the [contribution guidelines](https://github.com/lettuce-io/lettuce-core/blob/main/.github/CONTRIBUTING.md).
- [N] You have created a feature request first to discuss your contribution intent. Please reference the feature request ticket number in the pull request. (No since it is a bug fix instead of a feature).
- [Y] You use the code formatters provided [here](https://github.com/lettuce-io/lettuce-core/blob/main/formatting.xml) and have them applied to your changes. Don’t submit any formatting related changes.
- [Y] You submit test cases (unit or integration tests) that back your changes.

 <!--
Great! Live long and prosper.
-->



Patch:
diff --git a/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java b/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java
index 07fbacf77..502ce4a76 100644
--- a/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java
+++ b/src/main/java/io/lettuce/core/protocol/DefaultEndpoint.java
@@ -195,9 +195,9 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
             }
 
             if (autoFlushCommands) {
-
-                if (isConnected()) {
-                    writeToChannelAndFlush(command);
+                Channel channel = this.channel;
+                if (isConnected(channel)) {
+                    writeToChannelAndFlush(channel, command);
                 } else {
                     writeToDisconnectedBuffer(command);
                 }
@@ -236,9 +236,9 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
             }
 
             if (autoFlushCommands) {
-
-                if (isConnected()) {
-                    writeToChannelAndFlush(commands);
+                Channel channel = this.channel;
+                if (isConnected(channel)) {
+                    writeToChannelAndFlush(channel, commands);
                 } else {
                     writeToDisconnectedBuffer(commands);
                 }
@@ -288,10 +288,9 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
             return new RedisException("Connection is closed");
         }
 
+        final boolean connected = isConnected(this.channel);
         if (usesBoundedQueues()) {
 
-            boolean connected = isConnected();
-
             if (QUEUE_SIZE.get(this) + commands > clientOptions.getRequestQueueSize()) {
                 return new RedisException("Request queue size exceeded: " + clientOptions.getRequestQueueSize()
                         + ". Commands are not accepted until the queue size drops.");
@@ -308,7 +307,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
             }
         }
 
-        if (!isConnected() && rejectCommandsWhileDisconnected) {
+        if (!connected && rejectCommandsWhileDisconnected) {
             return new RedisException("Currently not connected. Commands are rejected.");
         }
 
@@ -370,11 +369,11 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
         commandBuffer.add(command);
     }
 
-    private void writeToChannelAndFlush(RedisCommand<?, ?, ?> command) {
+    private void writeToChannelAndFlush(Channel channel, RedisCommand<?, ?, ?> command) {
 
         QUEUE_SIZE.incrementAndGet(this);
 
-        ChannelFuture channelFuture = channelWriteAndFlush(command);
+        ChannelFuture channelFuture = channelWriteAndFlush(channel, command);
 
         if (reliability == Reliability.AT_MOST_ONCE) {
             // cancel on exceptions and remove from queue, because there is no housekeeping
@@ -387,7 +386,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
         }
     }
 
-    private void writeToChannelAndFlush(Collection<? extends RedisCommand<?, ?, ?>> commands) {
+    private void writeToChannelAndFlush(Channel channel, Collection<? extends RedisCommand<?, ?, ?>> commands) {
 
         QUEUE_SIZE.addAndGet(this, commands.size());
 
@@ -395,7 +394,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
 
             // cancel on exceptions and remove from queue, because there is no housekeeping
             for (RedisCommand<?, ?, ?> command : commands) {
-                channelWrite(command).addListener(AtMostOnceWriteListener.newInstance(this, command));
+                channelWrite(channel, command).addListener(AtMostOnceWriteListener.newInstance(this, command));
             }
         }
 
@@ -403,14 +402,14 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
 
             // commands are ok to stay within the queue, reconnect will retrigger them
             for (RedisCommand<?, ?, ?> command : commands) {
-                channelWrite(command).addListener(RetryListener.newInstance(this, command));
+                channelWrite(channel, command).addListener(RetryListener.newInstance(this, command));
             }
         }
 
-        channelFlush();
+        channelFlush(channel);
     }
 
-    private void channelFlush() {
+    private void channelFlush(Channel channel) {
 
         if (debugEnabled) {
             logger.debug("{} write() channelFlush", logPrefix());
@@ -419,7 +418,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
         channel.flush();
     }
 
-    private ChannelFuture channelWrite(RedisCommand<?, ?, ?> command) {
+    private ChannelFuture channelWrite(Channel channel, RedisCommand<?, ?, ?> command) {
 
         if (debugEnabled) {
             logger.debug("{} write() channelWrite command {}", logPrefix(), command);
@@ -428,7 +427,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
         return channel.write(command);
     }
 
-    private ChannelFuture channelWriteAndFlush(RedisCommand<?, ?, ?> command) {
+    private ChannelFuture channelWriteAndFlush(Channel channel, RedisCommand<?, ?, ?> command) {
 
         if (debugEnabled) {
             logger.debug("{} write() writeAndFlush command {}", logPrefix(), command);
@@ -441,7 +440,6 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
     public void notifyChannelActive(Channel channel) {
 
         this.logPrefix = null;
-        this.channel = channel;
         this.connectionError = null;
 
         if (isClosed()) {
@@ -456,6 +454,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
         }
 
         sharedLock.doExclusive(() -> {
+            this.channel = channel;
 
             try {
                 // Move queued commands to buffer before issuing any commands because of connection activation.
@@ -478,7 +477,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
                     inActivation = false;
                 }
 
-                flushCommands(disconnectedBuffer);
+                flushCommands(channel, disconnectedBuffer);
             } catch (Exception e) {
 
                 if (debugEnabled) {
@@ -525,7 +524,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
             doExclusive(this::drainCommands).forEach(cmd -> cmd.completeExceptionally(t));
         }
 
-        if (!isConnected()) {
+        if (!isConnected(this.channel)) {
             connectionError = t;
         }
     }
@@ -538,16 +537,16 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
     @Override
     @SuppressWarnings({ "rawtypes", "unchecked" })
     public void flushCommands() {
-        flushCommands(commandBuffer);
+        flushCommands(this.channel, commandBuffer);
     }
 
-    private void flushCommands(Queue<RedisCommand<?, ?, ?>> queue) {
+    private void flushCommands(Channel channel, Queue<RedisCommand<?, ?, ?>> queue) {
 
         if (debugEnabled) {
             logger.debug("{} flushCommands()", logPrefix());
         }
 
-        if (isConnected()) {
+        if (isConnected(channel)) {
 
             List<RedisCommand<?, ?, ?>> commands = sharedLock.doExclusive(() -> {
 
@@ -563,7 +562,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
             }
 
             if (!commands.isEmpty()) {
-                writeToChannelAndFlush(commands);
+                writeToChannelAndFlush(channel, commands);
             }
         }
     }
@@ -626,10 +625,10 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
 
     private Channel getOpenChannel() {
 
-        Channel currentChannel = this.channel;
+        Channel channel = this.channel;
 
-        if (currentChannel != null) {
-            return currentChannel;
+        if (channel != null /* && channel.isOpen() is this deliberately omitted? */) {
+            return channel;
         }
 
         return null;
@@ -646,6 +645,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
             logger.debug("{} reset()", logPrefix());
         }
 
+        Channel channel = this.channel;
         if (channel != null) {
             channel.pipeline().fireUserEventTriggered(new ConnectionEvents.Reset());
         }
@@ -718,9 +718,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
                 }
             }
 
-            if (isConnected()) {
-                flushCommands(disconnectedBuffer);
-            }
+            flushCommands(this.channel, disconnectedBuffer);
         });
     }
 
@@ -802,9 +800,7 @@ public class DefaultEndpoint implements RedisChannelWriter, Endpoint, PushHandle
         }
     }
 
-    private boolean isConnected() {
-
-        Channel channel = this.channel;
+    private boolean isConnected(Channel channel) {
         return channel != null && channel.isActive();
     }
 


Test patch:
diff --git a/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java b/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java
index 9b0a8cd7b..06354e5d6 100644
--- a/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java
+++ b/src/test/java/io/lettuce/core/protocol/DefaultEndpointUnitTests.java
@@ -124,6 +124,34 @@ class DefaultEndpointUnitTests {
         sut.setConnectionFacade(connectionFacade);
     }
 
+    @Test
+    void writeShouldGuaranteeFIFOOrder() {
+        sut.write(Collections.singletonList(new Command<>(CommandType.SELECT, new StatusOutput<>(StringCodec.UTF8))));
+
+        sut.registerConnectionWatchdog(connectionWatchdog);
+        doAnswer(i -> sut.write(new Command<>(CommandType.AUTH, new StatusOutput<>(StringCodec.UTF8)))).when(connectionWatchdog)
+                .arm();
+        when(channel.isActive()).thenReturn(true);
+
+        sut.notifyChannelActive(channel);
+
+        DefaultChannelPromise promise = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE);
+
+        when(channel.writeAndFlush(any())).thenAnswer(invocation -> {
+            if (invocation.getArguments()[0] instanceof RedisCommand) {
+                queue.add((RedisCommand) invocation.getArguments()[0]);
+            }
+
+            if (invocation.getArguments()[0] instanceof Collection) {
+                queue.addAll((Collection) invocation.getArguments()[0]);
+            }
+            return promise;
+        });
+
+        assertThat(queue).hasSize(2).first().hasFieldOrPropertyWithValue("type", CommandType.SELECT);
+        assertThat(queue).hasSize(2).last().hasFieldOrPropertyWithValue("type", CommandType.AUTH);
+    }
+
     @Test
     void writeConnectedShouldWriteCommandToChannel() {