Align READONLY and READWRITE with Redis cluster semantics

libs/cluster/Server/ClusterConfig.cs

@@ -168,22 +168,22 @@ public bool HasAssignedSlots(ushort workerId)
         /// 3. Local slots for a replica are those slots served by its primary only for read operations
         /// </summary>
         /// <param name="slot">Slot to check</param>
-        /// <param name="readWriteSession">Used to override write restrictions for non-local slots that are replicas of the slot owner</param>
+        /// <param name="enableReplicaReads">Whether a replica can serve reads for slots owned by its primary</param>
         /// <returns>True if slot is owned by this node, false otherwise</returns>
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public bool IsLocal(ushort slot, bool readWriteSession = true)
-            => slotMap[slot].workerId == LOCAL_WORKER_ID || IsLocalExpensive(slot, readWriteSession);
+        public bool IsLocal(ushort slot, bool enableReplicaReads = true)
+            => slotMap[slot].workerId == LOCAL_WORKER_ID || IsLocalExpensive(slot, enableReplicaReads);
 
         /// <summary>
         /// If slot in MIGRATE state then it must have been set by original owner, so we keep treating it like a local slot and serve requests if the key has not yet migrated.
         /// If it is a read command and this is a replica the associated slot should be assigned to this node's primary in order for the read request to be served.
         /// </summary>
         /// <param name="slot"></param>
-        /// <param name="readWriteSession"></param>
+        /// <param name="enableReplicaReads"></param>
         /// <returns></returns>
-        private bool IsLocalExpensive(ushort slot, bool readWriteSession)
+        private bool IsLocalExpensive(ushort slot, bool enableReplicaReads)
             => slotMap[slot]._state == SlotState.MIGRATING ||
-            (readWriteSession &&
+            (enableReplicaReads &&
             workers[1].Role == NodeRole.REPLICA &&
             slotMap[slot]._workerId > 1 &&
             LocalNodePrimaryId != null &&

libs/cluster/Server/ClusterManagerSlotState.cs

@@ -229,7 +229,7 @@ public bool TryPrepareSlotForImport(int slot, string nodeid, out ReadOnlySpan<by
                     return false;
                 }
 
-                if (current.IsLocal((ushort)slot, readWriteSession: false))
+                if (current.IsLocal((ushort)slot, enableReplicaReads: false))
                 {
                     errorMessage = Encoding.ASCII.GetBytes($"ERR This is a local hash slot {slot} and is already imported");
                     return false;
@@ -289,7 +289,7 @@ public bool TryPrepareSlotsForImport(HashSet<int> slots, string nodeid, out Read
                 foreach (var slot in slots)
                 {
                     // Can only import remote slots
-                    if (current.IsLocal((ushort)slot, readWriteSession: false))
+                    if (current.IsLocal((ushort)slot, enableReplicaReads: false))
                     {
                         errorMessage = Encoding.ASCII.GetBytes($"ERR This is a local hash slot {slot} and is already imported");
                         return false;

libs/cluster/Server/Replication/ReplicaOps/ReplicaDiskbasedSync.cs

@@ -346,7 +346,7 @@ public AofAddress TryReplicaDiskbasedRecovery(
 
                 // Mark this txn run as a read-write session if we are replaying as a replica
                 // This is necessary to ensure that the stored procedure can perform write operations if needed
-                clusterProvider.replicationManager.aofProcessor.SetReadWriteSession();
+                clusterProvider.replicationManager.aofProcessor.SetInternalWriteSession();
 
                 return this.replicationOffset;
             }

libs/cluster/Server/Replication/ReplicaOps/ReplicaDisklessSync.cs

@@ -245,7 +245,7 @@ public AofAddress TryReplicaDisklessRecovery(SyncMetadata primarySyncMetadata, o
 
                 // Mark this txn run as a read-write session if we are replaying as a replica
                 // This is necessary to ensure that the stored procedure can perform write operations if needed
-                clusterProvider.replicationManager.aofProcessor.SetReadWriteSession();
+                clusterProvider.replicationManager.aofProcessor.SetInternalWriteSession();
 
                 return this.replicationOffset;
             }

libs/cluster/Session/ClusterSession.cs

@@ -34,15 +34,15 @@ internal sealed partial class ClusterSession : IClusterSession
 
         public long LocalCurrentEpoch => _localCurrentEpoch;
 
-        /// <summary>
-        /// Indicates if this is a session that allows for reads and writes
-        /// </summary>
-        bool readWriteSession = false;
+        bool readOnlySession;
+        bool internalWriteSession;
 
-        public bool ReadWriteSession => clusterProvider.clusterManager.CurrentConfig.IsPrimary || readWriteSession;
+        public bool ReadOnlySession => readOnlySession;
+        public bool IsInternalWriteSession => internalWriteSession;
 
-        public void SetReadOnlySession() => readWriteSession = false;
-        public void SetReadWriteSession() => readWriteSession = true;
+        public void SetReadOnlySession() => readOnlySession = true;
+        public void SetReadWriteSession() => readOnlySession = false;
+        public void SetInternalWriteSession() => internalWriteSession = true;
 
         /// <inheritdoc/>
         public bool IsReplicating { get; private set; }

libs/cluster/Session/MigrateCommand.cs

@@ -195,7 +195,7 @@ private bool NetworkTryMIGRATE(out bool invalidParameters)
 
                         // Check if all keys are local R/W because we migrate keys and need to be able to delete them
                         var slot = HashSlotUtils.HashSlot(currKeySlice);
-                        if (!current.IsLocal(slot, readWriteSession: false))
+                        if (!current.IsLocal(slot, enableReplicaReads: false))
                         {
                             pstate = MigrateCmdParseState.SLOTNOTLOCAL;
                             continue;
@@ -247,7 +247,7 @@ private bool NetworkTryMIGRATE(out bool invalidParameters)
                         }
 
                         // Check if slot is local and can be migrated
-                        if (!current.IsLocal((ushort)slot, readWriteSession: false))
+                        if (!current.IsLocal((ushort)slot, enableReplicaReads: false))
                         {
                             pstate = MigrateCmdParseState.SLOTNOTLOCAL;
                             slotParseError = slot;
@@ -299,7 +299,7 @@ private bool NetworkTryMIGRATE(out bool invalidParameters)
                             }
 
                             // Check if slot is not owned by current node or cluster mode is not enabled
-                            if (!current.IsLocal((ushort)slot, readWriteSession: false))
+                            if (!current.IsLocal((ushort)slot, enableReplicaReads: false))
                             {
                                 pstate = MigrateCmdParseState.SLOTNOTLOCAL;
                                 slotParseError = slot;

libs/cluster/Session/SlotVerification/ClusterSlotVerify.cs

@@ -26,7 +26,7 @@ private ClusterSlotVerificationResult SingleKeySlotVerify(ref ClusterConfig conf
             ClusterSlotVerificationResult SingleKeyReadSlotVerify(ref ClusterConfig config, ref PinnedSpanByte keySlice)
             {
                 var _slot = slot == -1 ? HashSlotUtils.HashSlot(keySlice) : (ushort)slot;
-                var IsLocal = config.IsLocal(_slot);
+                var IsLocal = config.IsLocal(_slot, enableReplicaReads: readOnlySession);
                 var state = config.GetState(_slot);
 
                 // If local check we can serve request or redirect with ask
@@ -69,7 +69,7 @@ ClusterSlotVerificationResult SingleKeyReadWriteSlotVerify(bool waitForStableSlo
                 var _slot = slot == -1 ? HashSlotUtils.HashSlot(keySlice) : (ushort)slot;
 
             tryAgain:
-                var IsLocal = config.IsLocal(_slot, readWriteSession: readWriteSession);
+                var IsLocal = config.IsLocal(_slot, enableReplicaReads: internalWriteSession);
                 var state = config.GetState(_slot);
 
                 if (waitForStableSlot && state is SlotState.IMPORTING or SlotState.MIGRATING)
@@ -79,7 +79,7 @@ ClusterSlotVerificationResult SingleKeyReadWriteSlotVerify(bool waitForStableSlo
                 }
 
                 // Redirect r/w requests towards primary
-                if (config.LocalNodeRole == NodeRole.REPLICA && !readWriteSession)
+                if (config.LocalNodeRole == NodeRole.REPLICA && !internalWriteSession)
                     return new(SlotVerifiedState.MOVED, _slot);
 
                 if (IsLocal)

libs/resources/RespCommandsDocs.json

@@ -6130,7 +6130,7 @@
   {
     "Command": "READWRITE",
     "Name": "READWRITE",
-    "Summary": "Enables read-write queries for a connection to a Reids Cluster replica node.",
+    "Summary": "Disables read queries for a connection to a Redis Cluster replica node.",
     "Group": "Cluster",
     "Complexity": "O(1)"
   },

libs/server/AOF/AofProcessor.cs

@@ -93,14 +93,14 @@ public sealed unsafe partial class AofProcessor
         RangeIndexManager activeRangeIndexManager;
 
         /// <summary>
-        /// Set ReadWriteSession on the cluster session (NOTE: used for replaying stored procedures only)
+        /// Allow the cluster session to apply writes while replaying stored procedures
         /// </summary>
-        public void SetReadWriteSession()
+        public void SetInternalWriteSession()
         {
             for (var i = 0; i < storeWrapper.serverOptions.AofVirtualSublogCount; i++)
             {
                 var respServerSession = aofReplayCoordinator.GetReplayContext(i).respServerSession;
-                respServerSession.clusterSession.SetReadWriteSession();
+                respServerSession.clusterSession.SetInternalWriteSession();
             }
         }
 

libs/server/Cluster/IClusterSession.cs

@@ -18,9 +18,14 @@ public interface IClusterSession
         string RemoteNodeId { get; }
 
         /// <summary>
-        /// Type of session
+        /// Whether this connection may serve read-only commands from a replica
         /// </summary>
-        bool ReadWriteSession { get; }
+        bool ReadOnlySession { get; }
+
+        /// <summary>
+        /// Whether this internal session may apply writes while replaying the AOF
+        /// </summary>
+        bool IsInternalWriteSession { get; }
 
         /// <summary>
         /// If the current session is part of an active replication stream (set on first APPENDLOG, including the init handshake).
@@ -35,15 +40,20 @@ public interface IClusterSession
         IGarnetServer Server { get; set; }
 
         /// <summary>
-        /// Make this cluster session a read-only session
+        /// Allow this connection to serve read-only commands from a replica
         /// </summary>
         void SetReadOnlySession();
 
         /// <summary>
-        /// Make this cluster session a read-write session
+        /// Restore the default behavior of redirecting commands from a replica
         /// </summary>
         void SetReadWriteSession();
 
+        /// <summary>
+        /// Allow the internal AOF replay session to apply writes on a replica
+        /// </summary>
+        void SetInternalWriteSession();
+
         /// <summary>
         /// Local current epoch
         /// </summary>

libs/server/Config/RuntimeServerConfig.cs

@@ -114,9 +114,8 @@ void SetReadOnly(ServerConfigType t, string name, ConfigKind kind,
             // rejects them because they are constants or physical (require restart). Their CONFIG GET value
             // is computed by the per-option formatter, which reads directly from the startup
             // GarnetServerOptions (the read-only fall-through) — no runtime slot is used.
-            // NOTE: slave-read-only is intentionally NOT here: it is a per-session value (READWRITE/READONLY,
-            // https://redis.io/docs/latest/commands/readwrite/) and is handled directly by the CONFIG GET
-            // handler, which has the calling session in scope.
+            // NOTE: slave-read-only is intentionally not here because it is a fixed compatibility setting
+            // handled directly by CONFIG GET.
             SetReadOnly(ServerConfigType.TIMEOUT, "timeout",
                 ConfigKind.Int32 | ConfigKind.Seconds | ConfigKind.TimeSpan, static _ => "0", ConfigTimeUnit.Seconds);
             SetReadOnly(ServerConfigType.SAVE, "save", ConfigKind.String, static _ => "");

libs/server/Resp/BasicCommands.cs

@@ -999,7 +999,7 @@ private bool NetworkFLUSHDB()
                 return AbortWithWrongNumberOfArguments(nameof(RespCommand.FLUSHDB));
             }
 
-            if (storeWrapper.serverOptions.EnableCluster && storeWrapper.clusterProvider.IsReplica() && !clusterSession.ReadWriteSession)
+            if (storeWrapper.serverOptions.EnableCluster && storeWrapper.clusterProvider.IsReplica() && !clusterSession.IsInternalWriteSession)
             {
                 while (!RespWriteUtils.TryWriteError(CmdStrings.RESP_ERR_FLUSHALL_READONLY_REPLICA, ref dcurr, dend))
                     SendAndReset();
@@ -1021,7 +1021,7 @@ private bool NetworkFLUSHALL()
                 return AbortWithWrongNumberOfArguments(nameof(RespCommand.FLUSHALL));
             }
 
-            if (storeWrapper.serverOptions.EnableCluster && storeWrapper.clusterProvider.IsReplica() && !clusterSession.ReadWriteSession)
+            if (storeWrapper.serverOptions.EnableCluster && storeWrapper.clusterProvider.IsReplica() && !clusterSession.IsInternalWriteSession)
             {
                 return AbortWithErrorMessage(CmdStrings.RESP_ERR_FLUSHALL_READONLY_REPLICA);
             }
@@ -1034,7 +1034,7 @@ private bool NetworkFLUSHALL()
         }
 
         /// <summary>
-        /// Mark this session as readonly session
+        /// Allow this connection to serve read-only commands from a replica
         /// </summary>
         /// <returns></returns>
         private bool NetworkREADONLY()
@@ -1047,7 +1047,7 @@ private bool NetworkREADONLY()
         }
 
         /// <summary>
-        /// Mark this session as readwrite
+        /// Restore the default behavior of redirecting commands from a replica
         /// </summary>
         /// <returns></returns>
         private bool NetworkREADWRITE()

libs/server/ServerConfig.cs

@@ -18,8 +18,7 @@ public static ServerConfigType GetConfig(Span<byte> parameter)
             if (parameter.SequenceEqual("*"u8))
                 return ServerConfigType.ALL;
 
-            // slave-read-only is a per-session value (READWRITE/READONLY) and is resolved by the CONFIG GET
-            // handler which has the session in scope; it is not part of the runtime config table.
+            // slave-read-only is a fixed compatibility setting handled directly by CONFIG GET.
             if (parameter.SequenceEqual("SLAVE-READ-ONLY"u8))
                 return ServerConfigType.SLAVE_READ_ONLY;
 
@@ -51,7 +50,7 @@ private unsafe bool NetworkCONFIG_GET()
                 if (serverConfigType == ServerConfigType.ALL)
                 {
                     parameters = [.. RuntimeServerConfig.RuntimeTypes];
-                    // slave-read-only is session-scoped and not part of the table, so include it explicitly.
+                    // slave-read-only is not part of the runtime table, so include it explicitly.
                     parameters.Add(ServerConfigType.SLAVE_READ_ONLY);
                     returnAll = true;
                     continue;
@@ -74,10 +73,8 @@ private unsafe bool NetworkCONFIG_GET()
                     string name, value;
                     if (configType == ServerConfigType.SLAVE_READ_ONLY)
                     {
-                        // Per-session value: a session is read-only only when it is on a replica and has not
-                        // opted into writes via READWRITE (see https://redis.io/docs/latest/commands/readwrite/).
                         name = "slave-read-only";
-                        value = clusterSession == null || clusterSession.ReadWriteSession ? "no" : "yes";
+                        value = "yes";
                     }
                     else
                     {

test/cluster/Garnet.test.cluster.replication/ReplicationTests/ClusterReplicationBaseTests.cs

@@ -291,6 +291,45 @@ public void ClusterCheckpointRetrieveDisableStorageTier([Values] bool performRMW
             ClusterSRPrimaryCheckpointRetrieve(performRMW, disableObjects, false, true);
         }
 
+        [Test, Order(5)]
+        [Category("REPLICATION")]
+        public void ReadWriteSessionDoesNotEnableReplicaWritesAfterFailover()
+        {
+            const int primaryIndex = 0;
+            const int replicaIndex = 1;
+            const string key = "readwrite-failover-key";
+            const string value = "value";
+
+            context.CreateInstances(2, enableAOF: true, useTLS: useTLS, asyncReplay: asyncReplay, sublogCount: sublogCount);
+            context.CreateConnection(useTLS: useTLS);
+            _ = context.clusterTestUtils.SimpleSetupCluster(primary_count: 1, replica_count: 1, logger: context.logger);
+
+            using var session = context.clusterTestUtils.CreateGarnetClientSession(primaryIndex, useTLS: useTLS);
+            session.Connect();
+            ClassicAssert.AreEqual("OK", session.ExecuteAsync("READWRITE").GetAwaiter().GetResult());
+            ClassicAssert.AreEqual("OK", session.ExecuteAsync("SET", key, value).GetAwaiter().GetResult());
+            context.clusterTestUtils.WaitForReplicaAofSync(primaryIndex, replicaIndex, context.logger);
+
+            _ = context.clusterTestUtils.ClusterFailover(replicaIndex, logger: context.logger);
+            context.clusterTestUtils.WaitForNoFailover(replicaIndex, context.logger);
+            context.clusterTestUtils.WaitForFailoverCompleted(replicaIndex, context.logger);
+            context.clusterTestUtils.WaitForReplicaRecovery(primaryIndex, context.logger);
+            context.clusterTestUtils.WaitForReplicaAofSync(replicaIndex, primaryIndex, context.logger);
+
+            var slot = ClusterTestUtils.HashSlot(Encoding.ASCII.GetBytes(key));
+            var exception = Assert.Throws<Exception>(() => session.ExecuteAsync("GET", key).GetAwaiter().GetResult());
+            StringAssert.StartsWith($"MOVED {slot} ", exception.Message);
+
+            exception = Assert.Throws<Exception>(() => session.ExecuteAsync("SET", key, "local-value").GetAwaiter().GetResult());
+            StringAssert.StartsWith($"MOVED {slot} ", exception.Message);
+
+            exception = Assert.Throws<Exception>(() => session.ExecuteAsync("FLUSHALL").GetAwaiter().GetResult());
+            ClassicAssert.AreEqual("ERR You can't write against a read only replica.", exception.Message);
+
+            ClassicAssert.AreEqual("OK", session.ExecuteAsync("READONLY").GetAwaiter().GetResult());
+            ClassicAssert.AreEqual(value, session.ExecuteAsync("GET", key).GetAwaiter().GetResult());
+        }
+
         [Test, Order(6)]
         [Category("REPLICATION")]
         public void ClusterSRPrimaryCheckpointRetrieve([Values] bool performRMW, [Values] bool disableObjects, [Values] bool manySegments)