add pseudo-replica mode

src/cluster_legacy.c

@@ -6648,6 +6648,11 @@ int clusterCommandSpecial(client *c) {
             return 1;
         }
 
+        if (server.pseudo_replica) {
+            addReplyError(c, "CLUSTER REPLICATE not allowed in pseudo-replica mode.");
+            return 1;
+        }
+
         /* If the instance is currently a primary, it should have no assigned
          * slots nor keys to accept to replicate some other node.
          * Replicas can switch to another primary without issues. */

src/config.c

@@ -2331,6 +2331,14 @@ static int isValidActiveDefrag(int val, const char **err) {
     return 1;
 }
 
+static int isValidPseudoReplica(int val, const char **err) {
+    if (server.primary_host && val) {
+        *err = "Server is already a replica";
+        return 0;
+    }
+    return 1;
+}
+
 static int isValidClusterConfigFile(char *val, const char **err) {
     if (!strcmp(val, "")) {
         *err = "cluster-config-file can't be empty";
@@ -2949,6 +2957,11 @@ static int setConfigReplicaOfOption(standardConfig *config, sds *argv, int argc,
         return 0;
     }
 
+    if (server.pseudo_replica) {
+        *err = "REPLICAOF not allowed in pseudo-replica mode";
+        return 0;
+    }
+
     sdsfree(server.primary_host);
     server.primary_host = NULL;
     if (!strcasecmp(argv[0], "no") && !strcasecmp(argv[1], "one")) {
@@ -3136,6 +3149,7 @@ standardConfig static_configs[] = {
     createBoolConfig("enable-debug-assert", NULL, IMMUTABLE_CONFIG | HIDDEN_CONFIG, server.enable_debug_assert, 0, NULL, NULL),
     createBoolConfig("cluster-slot-stats-enabled", NULL, MODIFIABLE_CONFIG, server.cluster_slot_stats_enabled, 0, NULL, NULL),
     createBoolConfig("hide-user-data-from-log", NULL, MODIFIABLE_CONFIG, server.hide_user_data_from_log, 1, NULL, NULL),
+    createBoolConfig("pseudo-replica", NULL, MODIFIABLE_CONFIG, server.pseudo_replica, 0, isValidPseudoReplica, NULL),
 
     /* String Configs */
     createStringConfig("aclfile", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, server.acl_filename, "", NULL, NULL),

src/db.c

@@ -385,7 +385,7 @@ robj *dbRandomKey(serverDb *db) {
         key = dictGetKey(de);
         keyobj = createStringObject(key, sdslen(key));
         if (dbFindExpiresWithDictIndex(db, key, randomDictIndex)) {
-            if (allvolatile && server.primary_host && --maxtries == 0) {
+            if (allvolatile && (server.primary_host || server.pseudo_replica) && --maxtries == 0) {
                 /* If the DB is composed only of keys with an expire set,
                  * it could happen that all the keys are already logically
                  * expired in the repilca, so the function cannot stop because
@@ -1826,6 +1826,9 @@ keyStatus expireIfNeededWithDictIndex(serverDb *db, robj *key, int flags, int di
     if (server.primary_host != NULL) {
         if (server.current_client && (server.current_client->flag.primary)) return KEY_VALID;
         if (!(flags & EXPIRE_FORCE_DELETE_EXPIRED)) return KEY_EXPIRED;
+    } else if (server.pseudo_replica) {
+        if (server.current_client && (server.current_client->flag.pseudo_master)) return KEY_VALID;
+        if (!(flags & EXPIRE_FORCE_DELETE_EXPIRED)) return KEY_EXPIRED;
     }
 
     /* In some cases we're explicitly instructed to return an indication of a

src/evict.c

@@ -546,8 +546,8 @@ int performEvictions(void) {
         goto update_metrics;
     }
 
-    if (server.maxmemory_policy == MAXMEMORY_NO_EVICTION) {
-        result = EVICT_FAIL; /* We need to free memory, but policy forbids. */
+    if (server.maxmemory_policy == MAXMEMORY_NO_EVICTION || server.pseudo_replica) {
+        result = EVICT_FAIL; /* We need to free memory, but policy forbids or we are in 'pseudo-replica' mode. */
         goto update_metrics;
     }
 

src/expire.c

@@ -521,7 +521,7 @@ int checkAlreadyExpired(long long when) {
      *
      * Instead we add the already expired key to the database with expire time
      * (possibly in the past) and wait for an explicit DEL from the primary. */
-    return (when <= commandTimeSnapshot() && !server.loading && !server.primary_host);
+    return (when <= commandTimeSnapshot() && !server.loading && !server.primary_host && !server.pseudo_replica);
 }
 
 #define EXPIRE_NX (1 << 0)

src/replication.c

@@ -1272,6 +1272,10 @@ void syncCommand(client *c) {
  * - rdb-channel <1|0>
  * Used to identify the client as a replica's rdb connection in an dual channel
  * sync session.
+ *
+ * - pseudo-master <0|1>
+ * Set this connection behaving like a master if server.pseudo_replica is true.
+ * Sync tools can set their connections into 'pseudo-master' state to visit expired keys.
  * */
 void replconfCommand(client *c) {
     int j;
@@ -1418,6 +1422,16 @@ void replconfCommand(client *c) {
                 return;
             }
             c->associated_rdb_client_id = (uint64_t)client_id;
+        } else if (!strcasecmp(c->argv[j]->ptr, "pseudo-master")) {
+            long pseudo_master = 0;
+            if (getRangeLongFromObjectOrReply(c, c->argv[j + 1],
+                                              0, 1, &pseudo_master, NULL) != C_OK)
+                return;
+            if (pseudo_master == 1) {
+                c->flag.pseudo_master = 1;
+            } else {
+                c->flag.pseudo_master = 0;
+            }
         } else {
             addReplyErrorFormat(c, "Unrecognized REPLCONF option: %s", (char *)c->argv[j]->ptr);
             return;
@@ -3961,6 +3975,11 @@ void replicaofCommand(client *c) {
         return;
     }
 
+    if (server.pseudo_replica) {
+        addReplyError(c, "REPLICAOF not allowed in pseudo-replica mode.");
+        return;
+    }
+
     if (server.failover_state != NO_FAILOVER) {
         addReplyError(c, "REPLICAOF not allowed while failing over.");
         return;

src/server.c

@@ -1054,7 +1054,7 @@ void clientsCron(void) {
 void databasesCron(void) {
     /* Expire keys by random sampling. Not required for replicas
      * as primary will synthesize DELs for us. */
-    if (server.active_expire_enabled) {
+    if (server.active_expire_enabled && !server.pseudo_replica) {
         if (iAmPrimary()) {
             activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);
         } else {
@@ -1651,7 +1651,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
 
     /* Run a fast expire cycle (the called function will return
      * ASAP if a fast cycle is not needed). */
-    if (server.active_expire_enabled && iAmPrimary()) activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST);
+    if (server.active_expire_enabled && !server.pseudo_replica && iAmPrimary()) activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST);
 
     if (moduleCount()) {
         moduleFireServerEvent(VALKEYMODULE_EVENT_EVENTLOOP, VALKEYMODULE_SUBEVENT_EVENTLOOP_BEFORE_SLEEP, NULL);
@@ -2057,6 +2057,7 @@ void initServerConfig(void) {
     server.extended_redis_compat = 0;
     server.pause_cron = 0;
     server.dict_resizing = 1;
+    server.pseudo_replica = 0;
 
     server.latency_tracking_info_percentiles_len = 3;
     server.latency_tracking_info_percentiles = zmalloc(sizeof(double) * (server.latency_tracking_info_percentiles_len));

src/server.h

@@ -1224,7 +1224,8 @@ typedef struct ClientFlags {
                                             * knows that it does not need the cache and required a full sync. With this
                                             * flag, we won't cache the primary in freeClient. */
     uint64_t fake : 1;                     /* This is a fake client without a real connection. */
-    uint64_t reserved : 5;                 /* Reserved for future use */
+    uint64_t pseudo_master : 1;            /* This client is a pseudo master */
+    uint64_t reserved : 4;                 /* Reserved for future use */
 } ClientFlags;
 
 typedef struct client {
@@ -2070,6 +2071,8 @@ struct valkeyServer {
     char primary_replid[CONFIG_RUN_ID_SIZE + 1]; /* Primary PSYNC runid. */
     long long primary_initial_offset;            /* Primary PSYNC offset. */
     int repl_replica_lazy_flush;                 /* Lazy FLUSHALL before loading DB? */
+    /* Pseudo Replica */
+    int pseudo_replica; /* If true, server is a pseudo replica. */
     /* Synchronous replication. */
     list *clients_waiting_acks; /* Clients waiting in WAIT or WAITAOF. */
     int get_ack_from_replicas;  /* If true we send REPLCONF GETACK. */

tests/unit/expire.tcl

@@ -832,6 +832,80 @@ start_server {tags {"expire"}} {
         close_replication_stream $repl
         assert_equal [r debug set-active-expire 1] {OK}
     } {} {needs:debug}
+
+    test {Pseudo-replica mode should forbid active expiration} {
+        r flushall
+
+        r config set pseudo-replica yes
+        assert_equal [r replconf pseudo-master 1] {OK}
+
+        r set foo1 bar PX 1
+        r set foo2 bar PX 1
+        after 100
+
+        assert_equal [r dbsize] {2}
+
+        assert_equal [r replconf pseudo-master 0] {OK}
+        r config set pseudo-replica no
+
+        # Verify all keys have expired
+        wait_for_condition 40 100 {
+            [r dbsize] eq 0
+        } else {
+            fail "Keys did not actively expire."
+        }
+    }
+
+    test {Pseudo-replica mode should forbid lazy expiration} {
+        r flushall
+        r debug set-active-expire 0 
+
+        r config set pseudo-replica yes
+        assert_equal [r replconf pseudo-master 1] {OK}
+
+        r set foo1 1 PX 1
+        after 10
+
+        r get foo1
+        assert_equal [r dbsize] {1}
+
+        assert_equal [r replconf pseudo-master 0] {OK}
+        r config set pseudo-replica no
+
+        r get foo1
+
+        assert_equal [r dbsize] {0}
+
+        assert_equal [r debug set-active-expire 1] {OK}
+    } {} {needs:debug}
+
+    test {RANDOMKEY can return expired key in Pseudo-replica mode} {
+        r flushall
+
+        r config set pseudo-replica yes
+        assert_equal [r replconf pseudo-master 1] {OK}
+
+        r set foo1 bar PX 1
+        after 10
+
+        set client [valkey [srv "host"] [srv "port"] 0 $::tls]
+        if {!$::singledb} {
+            $client select 9
+        }
+        assert_equal [$client ttl foo1] {-2}
+
+        assert_equal [r randomkey] {foo1}
+
+        assert_equal [r replconf pseudo-master 0] {OK}
+        r config set pseudo-replica no
+
+        # Verify all keys have expired
+        wait_for_condition 40 100 {
+            [r dbsize] eq 0
+        } else {
+            fail "Keys did not actively expire."
+        }
+    }
 }
 
 start_cluster 1 0 {tags {"expire external:skip cluster"}} {

tests/unit/maxmemory.tcl

@@ -611,3 +611,21 @@ start_server {tags {"maxmemory" "external:skip"}} {
         assert {[r object freq foo] == 5}
     }
 }
+
+start_server {tags {"maxmemory" "external:skip"}} {
+    test {Pseudo-replica mode should forbid eviction} {
+        r set key val
+        r config set pseudo-replica yes
+        assert_equal [r replconf pseudo-master 1] {OK}  
+        r config set maxmemory-policy allkeys-lru
+        r config set maxmemory 1      
+
+        assert_equal [r dbsize] {1}
+        assert_error {OOM command not allowed*} {r set key1 val1}
+
+        assert_equal [r replconf pseudo-master 0] {OK}  
+        r config set pseudo-replica no
+
+        assert_equal [r dbsize] {0}
+    }
+}

valkey.conf

@@ -801,6 +801,13 @@ replica-priority 100
 #
 # replica-ignore-disk-write-errors no
 
+# Make the master behave like a replica, which forbids expiration and evcition.
+# This is useful for sync tools, because expiration and evcition may cause the data corruption.
+# Sync tools can set their connections into 'pseudo-master' state by REPLCONF PSEUDO-MASTER to
+# behave like a master(i.e. visit expired keys).
+#
+# pseudo-replica no
+
 # -----------------------------------------------------------------------------
 # By default, Sentinel includes all replicas in its reports. A replica
 # can be excluded from Sentinel's announcements. An unannounced replica