fix: revert protectMode upon channelActive event, complete non-retrya…

…ble activation command in drainStackUponChannelInactive()
redis · Aug 17, 2024 · 6df7b23 · 6df7b23
1 parent 7a54c8b
commit 6df7b23
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 12 deletions.
diff --git a/src/main/java/io/lettuce/core/protocol/CommandHandler.java b/src/main/java/io/lettuce/core/protocol/CommandHandler.java
@@ -190,13 +190,18 @@ void setBuffer(ByteBuf buffer) {
         return drainCommands(stack);
     }
 
-    private Deque<RedisCommand<?, ?, ?>> drainStack() {
+    private Deque<RedisCommand<?, ?, ?>> drainStackUponChannelInactive() {
         final Deque<RedisCommand<?, ?, ?>> target = new ArrayDeque<>(stack.size());
 
         RedisCommand<?, ?, ?> cmd;
         while ((cmd = stack.poll()) != null) {
-            if (!cmd.isDone() && !ActivationCommand.isActivationCommand(cmd)) {
-                target.add(cmd);
+            if (!cmd.isDone()) {
+                if (!ActivationCommand.isActivationCommand(cmd)) {
+                    target.add(cmd);
+                } else {
+                    cmd.completeExceptionally(
+                            new RedisConnectionException("activation command won't be retried upon channel inactive"));
+                }
             }
         }
 
@@ -379,7 +384,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         endpoint.notifyChannelInactive(ctx.channel());
         Deque<RedisCommand<?, ?, ?>> autoBatchFlushRetryableDrainQueuedCommands = UnmodifiableDeque.emptyDeque();
         if (supportsAutoBatchFlush) {
-            autoBatchFlushRetryableDrainQueuedCommands = drainStack();
+            autoBatchFlushRetryableDrainQueuedCommands = drainStackUponChannelInactive();
         } else {
             endpoint.notifyDrainQueuedCommands(this);
         }

diff --git a/src/main/java/io/lettuce/core/protocol/DefaultAutoBatchFlushEndpoint.java b/src/main/java/io/lettuce/core/protocol/DefaultAutoBatchFlushEndpoint.java
@@ -319,6 +319,7 @@ public void notifyChannelActive(Channel channel) {
 
         this.logPrefix = null;
         this.connectionError = null;
+        this.inProtectMode = false;
 
         if (!CHANNEL.compareAndSet(this, DummyContextualChannelInstances.CHANNEL_CONNECTING, contextualChannel)) {
             channel.close();
@@ -398,7 +399,7 @@ public void notifyChannelInactive(Channel channel) {
 
     @Override
     public void notifyChannelInactiveAfterWatchdogDecision(Channel channel,
-            Deque<RedisCommand<?, ?, ?>> retryableQueuedCommands) {
+            Deque<RedisCommand<?, ?, ?>> retryablePendingCommands) {
         final ContextualChannel inactiveChan = this.channel;
         if (!inactiveChan.context.initialState.isConnected()) {
             logger.error("[unexpected][{}] notifyChannelInactive: channel initial state not connected", logPrefix());
@@ -446,7 +447,7 @@ public void notifyChannelInactiveAfterWatchdogDecision(Channel channel,
             CHANNEL.set(this, DummyContextualChannelInstances.CHANNEL_ENDPOINT_CLOSED);
         }
         inactiveChan.context
-                .setCloseStatus(new ConnectionContext.CloseStatus(willReconnect, retryableQueuedCommands, exception));
+                .setCloseStatus(new ConnectionContext.CloseStatus(willReconnect, retryablePendingCommands, exception));
         trySetEndpointQuiescence(inactiveChan);
     }
 
@@ -945,11 +946,11 @@ private <K, V, T> RedisCommand<K, V, T> processActivationCommand(RedisCommand<K,
 
     private Throwable validateWrite(ContextualChannel chan, int commands, boolean isActivationCommand) {
         if (isClosed()) {
-            return new RedisException("Connection is closed");
+            return new RedisException("Endpoint is closed");
         }
 
         final Throwable localConnectionErr = connectionError;
-        if (localConnectionErr != null /* different logic of DefaultEndpoint */) {
+        if (localConnectionErr != null /* attention: different logic of DefaultEndpoint */) {
             return localConnectionErr;
         }
 
@@ -961,18 +962,19 @@ private Throwable validateWrite(ContextualChannel chan, int commands, boolean is
 
         final ConnectionContext.State initialState = chan.context.initialState;
         final boolean rejectCommandsWhileDisconnectedLocal = this.rejectCommandsWhileDisconnected || isActivationCommand;
+        final String rejectDesc = isActivationCommand ? "isActivationCommand" : "rejectCommandsWhileDisconnected";
         switch (initialState) {
             case ENDPOINT_CLOSED:
                 return new RedisException("Connection is closed");
             case RECONNECT_FAILED:
-                return failedToReconnectReason;
+                return getFailedToReconnectReason();
             case WILL_RECONNECT:
             case CONNECTING:
-                return rejectCommandsWhileDisconnectedLocal
-                        ? new RedisException("Currently not connected. Commands are rejected.")
+                return rejectCommandsWhileDisconnectedLocal ? new RedisException("Currently not connected and " + rejectDesc)
                         : null;
             case CONNECTED:
-                return !chan.isActive() && rejectCommandsWhileDisconnectedLocal ? new RedisException("Channel is closed")
+                return !chan.isActive() && rejectCommandsWhileDisconnectedLocal
+                        ? new RedisException("Channel is inactive and " + rejectDesc)
                         : null;
             default:
                 throw new IllegalStateException("unexpected state: " + initialState);