Roll backward downgrade compatibility from Redis 7.2 and Valkey 7.2/8.0 to Redis 7.0

redis.conf

@@ -462,6 +462,15 @@ rdbcompression yes
 # tell the loading code to skip the check.
 rdbchecksum yes
 
+# Valkey can try to load an RDB dump produced by a future version of Valkey.
+# This can only work on a best-effort basis, because future RDB versions may
+# contain information that's not known to the current version. If no new features
+# are used, it may be possible to import the data produced by a later version,
+# but loading is aborted if unknown information is encountered. Possible values
+# are 'strict' and 'relaxed'. This also applies to replication and the RESTORE
+# command.
+rdb-version-check relaxed
+
 # Enables or disables full sanitization checks for ziplist and listpack etc when
 # loading an RDB or RESTORE payload. This reduces the chances of a assertion or
 # crash later on while processing commands.
@@ -909,10 +918,10 @@ replica-priority 100
 #               commands. For instance ~* allows all the keys. The pattern
 #               is a glob-style pattern like the one of KEYS.
 #               It is possible to specify multiple patterns.
-# %R~<pattern>  Add key read pattern that specifies which keys can be read 
+# %R~<pattern>  Add key read pattern that specifies which keys can be read
 #               from.
 # %W~<pattern>  Add key write pattern that specifies which keys can be
-#               written to. 
+#               written to.
 #  allkeys      Alias for ~*
 #  resetkeys    Flush the list of allowed keys patterns.
 #  &<pattern>   Add a glob-style pattern of Pub/Sub channels that can be
@@ -939,10 +948,10 @@ replica-priority 100
 #               -@all. The user returns to the same state it has immediately
 #               after its creation.
 # (<options>)   Create a new selector with the options specified within the
-#               parentheses and attach it to the user. Each option should be 
-#               space separated. The first character must be ( and the last 
+#               parentheses and attach it to the user. Each option should be
+#               space separated. The first character must be ( and the last
 #               character must be ).
-# clearselectors            Remove all of the currently attached selectors. 
+# clearselectors            Remove all of the currently attached selectors.
 #                           Note this does not change the "root" user permissions,
 #                           which are the permissions directly applied onto the
 #                           user (outside the parentheses).
@@ -968,7 +977,7 @@ replica-priority 100
 # Basically ACL rules are processed left-to-right.
 #
 # The following is a list of command categories and their meanings:
-# * keyspace - Writing or reading from keys, databases, or their metadata 
+# * keyspace - Writing or reading from keys, databases, or their metadata
 #     in a type agnostic way. Includes DEL, RESTORE, DUMP, RENAME, EXISTS, DBSIZE,
 #     KEYS, EXPIRE, TTL, FLUSHALL, etc. Commands that may modify the keyspace,
 #     key or metadata will also have `write` category. Commands that only read
@@ -1589,8 +1598,8 @@ aof-timestamp-enabled no
 #
 # cluster-node-timeout 15000
 
-# The cluster port is the port that the cluster bus will listen for inbound connections on. When set 
-# to the default value, 0, it will be bound to the command port + 10000. Setting this value requires 
+# The cluster port is the port that the cluster bus will listen for inbound connections on. When set
+# to the default value, 0, it will be bound to the command port + 10000. Setting this value requires
 # you to specify the cluster bus port when executing cluster meet.
 # cluster-port 0
 
@@ -1725,12 +1734,12 @@ aof-timestamp-enabled no
 # PubSub message by default. (client-query-buffer-limit default value is 1gb)
 #
 # cluster-link-sendbuf-limit 0
- 
-# Clusters can configure their announced hostname using this config. This is a common use case for 
+
+# Clusters can configure their announced hostname using this config. This is a common use case for
 # applications that need to use TLS Server Name Indication (SNI) or dealing with DNS based
 # routing. By default this value is only shown as additional metadata in the CLUSTER SLOTS
-# command, but can be changed using 'cluster-preferred-endpoint-type' config. This value is 
-# communicated along the clusterbus to all nodes, setting it to an empty string will remove 
+# command, but can be changed using 'cluster-preferred-endpoint-type' config. This value is
+# communicated along the clusterbus to all nodes, setting it to an empty string will remove
 # the hostname and also propagate the removal.
 #
 # cluster-announce-hostname ""
@@ -1739,13 +1748,13 @@ aof-timestamp-enabled no
 # a user defined hostname, or by declaring they have no endpoint. Which endpoint is
 # shown as the preferred endpoint is set by using the cluster-preferred-endpoint-type
 # config with values 'ip', 'hostname', or 'unknown-endpoint'. This value controls how
-# the endpoint returned for MOVED/ASKING requests as well as the first field of CLUSTER SLOTS. 
-# If the preferred endpoint type is set to hostname, but no announced hostname is set, a '?' 
+# the endpoint returned for MOVED/ASKING requests as well as the first field of CLUSTER SLOTS.
+# If the preferred endpoint type is set to hostname, but no announced hostname is set, a '?'
 # will be returned instead.
 #
 # When a cluster advertises itself as having an unknown endpoint, it's indicating that
-# the server doesn't know how clients can reach the cluster. This can happen in certain 
-# networking situations where there are multiple possible routes to the node, and the 
+# the server doesn't know how clients can reach the cluster. This can happen in certain
+# networking situations where there are multiple possible routes to the node, and the
 # server doesn't know which one the client took. In this case, the server is expecting
 # the client to reach out on the same endpoint it used for making the last request, but use
 # the port provided in the response.
@@ -2058,7 +2067,7 @@ client-output-buffer-limit pubsub 32mb 8mb 60
 # errors or data eviction. To avoid this we can cap the accumulated memory
 # used by all client connections (all pubsub and normal clients). Once we
 # reach that limit connections will be dropped by the server freeing up
-# memory. The server will attempt to drop the connections using the most 
+# memory. The server will attempt to drop the connections using the most
 # memory first. We call this mechanism "client eviction".
 #
 # Client eviction is configured using the maxmemory-clients setting as follows:

src/cluster.c

@@ -5986,7 +5986,10 @@ int verifyDumpPayload(unsigned char *p, size_t len, uint16_t *rdbver_ptr) {
     if (rdbver_ptr) {
         *rdbver_ptr = rdbver;
     }
-    if (rdbver > RDB_VERSION) return C_ERR;
+    if ((rdbver >= RDB_FOREIGN_VERSION_MIN && rdbver <= RDB_FOREIGN_VERSION_MAX) ||
+        (rdbver > RDB_VERSION && server.rdb_version_check == RDB_VERSION_CHECK_STRICT)) {
+        return C_ERR;
+    }
 
     if (server.skip_checksum_validation)
         return C_OK;

src/config.c

@@ -159,6 +159,10 @@ configEnum propagation_error_behavior_enum[] = {
     {NULL, 0}
 };
 
+configEnum rdb_version_check_enum[] = {{"strict",  RDB_VERSION_CHECK_STRICT}, // strict: Reject future RDB versions.
+                                       {"relaxed", RDB_VERSION_CHECK_RELAXED}, // relaxed: Try parsing future RDB versions and fail only when an unknown RDB opcode or type is encountered.
+                                       {NULL,      0}};
+
 /* Output buffer limits presets. */
 clientBufferLimitsConfig clientBufferLimitsDefaults[CLIENT_TYPE_OBUF_COUNT] = {
     {0, 0, 0}, /* normal */
@@ -3055,6 +3059,7 @@ standardConfig static_configs[] = {
     createEnumConfig("propagation-error-behavior", NULL, MODIFIABLE_CONFIG, propagation_error_behavior_enum, server.propagation_error_behavior, PROPAGATION_ERR_BEHAVIOR_IGNORE, NULL, NULL),
     createEnumConfig("shutdown-on-sigint", NULL, MODIFIABLE_CONFIG | MULTI_ARG_CONFIG, shutdown_on_sig_enum, server.shutdown_on_sigint, 0, isValidShutdownOnSigFlags, NULL),
     createEnumConfig("shutdown-on-sigterm", NULL, MODIFIABLE_CONFIG | MULTI_ARG_CONFIG, shutdown_on_sig_enum, server.shutdown_on_sigterm, 0, isValidShutdownOnSigFlags, NULL),
+    createEnumConfig("rdb-version-check", NULL, MODIFIABLE_CONFIG, rdb_version_check_enum, server.rdb_version_check, RDB_VERSION_CHECK_RELAXED, NULL, NULL),
 
     /* Integer configs */
     createIntConfig("databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, server.dbnum, 16, INTEGER_CONFIG, NULL, NULL),

src/rdb.c

@@ -37,13 +37,14 @@
 #include <math.h>
 #include <fcntl.h>
 #include <sys/types.h>
+#include <stdbool.h>
 #include <sys/time.h>
 #include <sys/resource.h>
 #include <sys/wait.h>
 #include <arpa/inet.h>
 #include <sys/stat.h>
 #include <sys/param.h>
-
+#include "listpack.h"
 /* This macro is called when the internal RDB structure is corrupt */
 #define rdbReportCorruptRDB(...) rdbReportError(1, __LINE__,__VA_ARGS__)
 /* This macro is called when RDB read failed (possibly a short read) */
@@ -1720,6 +1721,90 @@ int lpPairsValidateIntegrityAndDups(unsigned char *lp, size_t size, int deep) {
     return ret;
 }
 
+/*
+ * This new function would be responsible for handling the RDB_TYPE_SET_LISTPACK
+ */
+robj *rdbLoadSetListpackObject(rio *rdb) {
+    // Step 1: Load the serialized listpack from the RDB stream.
+    robj *listpack_obj = rdbLoadStringObject(rdb);
+    if (listpack_obj == NULL) {
+        serverLog(LL_WARNING, "RDB: Failed to load string object for RDB_TYPE_SET_LISTPACK type.");
+        return NULL;
+    }
+
+    if (listpack_obj->type != OBJ_STRING || !sdsEncodedObject(listpack_obj)) {
+        serverLog(LL_WARNING, "RDB: Loaded object for RDB_TYPE_SET_LISTPACK is not a valid string type.");
+        decrRefCount(listpack_obj);
+        return NULL;
+    }
+
+    sds listpack_sds = sdsdup(listpack_obj->ptr);
+    decrRefCount(listpack_obj); // We have our own copy of the SDS now.
+
+    if (listpack_sds == NULL) {
+        serverLog(LL_WARNING, "RDB: OOM when duplicating SDS for RDB_TYPE_SET_LISTPACK.");
+        return NULL;
+    }
+
+    // Step 2: Create a new Set object for the older Redis version.
+    robj *set_obj = createSetObject();
+    if (set_obj == NULL) {
+        serverLog(LL_WARNING, "RDB: Failed to createSetObject for RDB_TYPE_SET_LISTPACK.");
+        sdsfree(listpack_sds);
+        return NULL;
+    }
+
+    // Step 3: Iterate through the listpack and add elements to the Set.
+    unsigned char *lp = (unsigned char *)listpack_sds;
+    unsigned char *fptr;
+    unsigned char *eptr;
+    unsigned char *vstr;
+    unsigned int vlen;
+    long long vlong;
+
+    fptr = lpFirst(lp);
+    eptr = fptr;
+
+    while (eptr) {
+        vstr = lpGetValue(eptr, &vlen, &vlong);
+        sds ele_sds;
+
+        if (vstr) {
+            ele_sds = sdsnewlen((const void *)vstr, vlen);
+        } else {
+            ele_sds = sdsfromlonglong(vlong);
+        }
+
+        if (ele_sds == NULL) {
+            serverLog(LL_WARNING, "RDB: OOM creating SDS for set element from RDB_TYPE_SET_LISTPACK .");
+            sdsfree(listpack_sds);
+            decrRefCount(set_obj);
+            return NULL;
+        }
+
+        // Add the element to the set.
+        // setTypeAdd in Redis 7.0.x (and similar versions) does NOT take ownership
+        // of the 'ele_sds' passed to it. It either uses the integer value directly
+        // (for intsets) or creates its own sdsdup for hashtables.
+        // Therefore, 'ele_sds' must always be freed by this function after the call.
+        setTypeAdd(set_obj, ele_sds);
+        // We don't strictly need to check the return value of setTypeAdd for freeing ele_sds,
+        // as it's never consumed by setTypeAdd. However, checking it could be useful
+        // for logging or other error handling if an element fails to be added for other reasons.
+
+        sdsfree(ele_sds); // Always free ele_sds as it's not consumed by setTypeAdd.
+
+        eptr = lpNext(lp, eptr);
+    }
+
+    // Step 4: Clean up the duplicated raw listpack SDS.
+    sdsfree(listpack_sds);
+
+    // Step 5: Return the created and populated set object.
+    serverLog(LL_VERBOSE, "RDB: Successfully loaded a RDB_TYPE_SET_LISTPACK and converted to OBJ_ENCODING_HT.");
+    return set_obj;
+}
+
 /* Load a Redis object of the specified type from the specified file.
  * On success a newly allocated object is returned, otherwise NULL.
  * When the function returns NULL and if 'error' is not NULL, the
@@ -1742,8 +1827,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
             skip = !!(server.current_client->user->flags & USER_FLAG_SANITIZE_PAYLOAD_SKIP);
         deep_integrity_validation = !skip;
     }
-
-    if (rdbtype == RDB_TYPE_STRING) {
+    if (rdbtype == RDB_TYPE_SET_LISTPACK) {
+        o = rdbLoadSetListpackObject(rdb);
+    } else if (rdbtype == RDB_TYPE_STRING) {
         /* Read string value */
         if ((o = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
         o = tryObjectEncoding(o);
@@ -2317,7 +2403,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
                 rdbReportCorruptRDB("Unknown RDB encoding type %d",rdbtype);
                 break;
         }
-    } else if (rdbtype == RDB_TYPE_STREAM_LISTPACKS || rdbtype == RDB_TYPE_STREAM_LISTPACKS_2) {
+    } else if (rdbtype == RDB_TYPE_STREAM_LISTPACKS || rdbtype == RDB_TYPE_STREAM_LISTPACKS_2 || rdbtype == RDB_TYPE_STREAM_LISTPACKS_3) {
         o = createStreamObject();
         stream *s = o->ptr;
         uint64_t listpacks = rdbLoadLen(rdb,NULL);
@@ -2394,7 +2480,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
         s->last_id.ms = rdbLoadLen(rdb,NULL);
         s->last_id.seq = rdbLoadLen(rdb,NULL);
 
-        if (rdbtype == RDB_TYPE_STREAM_LISTPACKS_2) {
+        if (rdbtype >= RDB_TYPE_STREAM_LISTPACKS_2) {
             /* Load the first entry ID. */
             s->first_id.ms = rdbLoadLen(rdb,NULL);
             s->first_id.seq = rdbLoadLen(rdb,NULL);
@@ -2461,7 +2547,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
 
             /* Load group offset. */
             uint64_t cg_offset;
-            if (rdbtype == RDB_TYPE_STREAM_LISTPACKS_2) {
+            if (rdbtype >= RDB_TYPE_STREAM_LISTPACKS_2) {
                 cg_offset = rdbLoadLen(rdb,NULL);
                 if (rioGetReadError(rdb)) {
                     rdbReportReadError("Stream cgroup offset loading failed.");
@@ -2902,18 +2988,25 @@ int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadin
     char buf[1024];
     int error;
     long long empty_keys_skipped = 0;
+    bool is_valkey_magic;
 
     rdb->update_cksum = rdbLoadProgressCallback;
     rdb->max_processing_chunk = server.loading_process_events_interval_bytes;
     if (rioRead(rdb,buf,9) == 0) goto eoferr;
     buf[9] = '\0';
-    if (memcmp(buf,"REDIS",5) != 0) {
-        serverLog(LL_WARNING,"Wrong signature trying to load DB from file");
+    if (memcmp(buf, "REDIS0", 6) == 0) {
+        is_valkey_magic = false;
+    } else if (memcmp(buf, "VALKEY", 6) == 0) {
+        is_valkey_magic = true;
+    } else {
+        serverLog(LL_WARNING, "Wrong signature trying to load DB from file");
         errno = EINVAL;
         return C_ERR;
     }
-    rdbver = atoi(buf+5);
-    if (rdbver < 1 || rdbver > RDB_VERSION) {
+    rdbver = atoi(buf + 6);
+    if (rdbver < 1 ||
+        (rdbver >= RDB_FOREIGN_VERSION_MIN && !is_valkey_magic) ||
+        (rdbver > RDB_VERSION && server.rdb_version_check == RDB_VERSION_CHECK_STRICT)) {
         serverLog(LL_WARNING,"Can't handle RDB format version %d",rdbver);
         errno = EINVAL;
         return C_ERR;
@@ -3014,9 +3107,10 @@ int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadin
                 if (rsi) rsi->repl_offset = strtoll(auxval->ptr,NULL,10);
             } else if (!strcasecmp(auxkey->ptr,"lua")) {
                 /* Won't load the script back in memory anymore. */
-            } else if (!strcasecmp(auxkey->ptr,"redis-ver")) {
-                serverLog(LL_NOTICE,"Loading RDB produced by version %s",
-                    (char*)auxval->ptr);
+            } else if (!strcasecmp(auxkey->ptr, "redis-ver")) {
+                serverLog(LL_NOTICE, "Loading RDB produced by Redis version %s", (char *)auxval->ptr);
+            } else if (!strcasecmp(auxkey->ptr, "valkey-ver")) {
+                serverLog(LL_NOTICE, "Loading RDB produced by Valkey version %s", (char *)auxval->ptr);
             } else if (!strcasecmp(auxkey->ptr,"ctime")) {
                 time_t age = time(NULL)-strtol(auxval->ptr,NULL,10);
                 if (age < 0) age = 0;

src/rdb.h

@@ -37,9 +37,27 @@
 #include "server.h"
 
 /* The current RDB version. When the format changes in a way that is no longer
- * backward compatible this number gets incremented. */
+ * backward compatible this number gets incremented.
+ *
+ * RDB 11 is the last open-source Redis RDB version, used by Valkey 7.x and 8.x.
+ *
+ * RDB 12+ are non-open-source Redis formats.
+ *
+ * Next time we bump the Valkey RDB version, use much higher version to avoid
+ * collisions with non-OSS Redis RDB versions. For example, we could use RDB
+ * version 90 for Valkey 9.0.
+ *
+ * In an RDB file/stream, we also check the magic string REDIS or VALKEY but in
+ * the DUMP/RESTORE format, there is only the RDB version number and no magic
+ * string. */
 #define RDB_VERSION 10
 
+/* Reserved range for foreign (unsupported, non-OSS) RDB format. */
+#define RDB_FOREIGN_VERSION_MIN 12
+#define RDB_FOREIGN_VERSION_MAX 79
+static_assert(RDB_VERSION < RDB_FOREIGN_VERSION_MIN || RDB_VERSION > RDB_FOREIGN_VERSION_MAX,
+              "RDB version in foreign version range");
+
 /* Defines related to the dump file format. To store 32 bits lengths for short
  * keys requires a lot of space, so we check the most significant 2 bits of
  * the first byte to interpreter the length:
@@ -95,6 +113,8 @@
 #define RDB_TYPE_ZSET_LISTPACK 17
 #define RDB_TYPE_LIST_QUICKLIST_2   18
 #define RDB_TYPE_STREAM_LISTPACKS_2 19
+#define RDB_TYPE_SET_LISTPACK 20
+#define RDB_TYPE_STREAM_LISTPACKS_3 21
 /* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */
 
 /* Test if a type is an object type. */

src/server.h

@@ -590,6 +590,11 @@ typedef enum {
     CLUSTER_ENDPOINT_TYPE_UNKNOWN_ENDPOINT /* Show NULL or empty */
 } cluster_endpoint_type;
 
+typedef enum {
+    RDB_VERSION_CHECK_STRICT = 0,
+    RDB_VERSION_CHECK_RELAXED
+} rdb_version_check_type;
+
 /* RDB active child save type. */
 #define RDB_CHILD_TYPE_NONE 0
 #define RDB_CHILD_TYPE_DISK 1     /* RDB is written to disk. */
@@ -1635,6 +1640,7 @@ struct redisServer {
     int active_defrag_enabled;
     int sanitize_dump_payload;      /* Enables deep sanitization for ziplist and listpack in RDB and RESTORE. */
     int skip_checksum_validation;   /* Disable checksum validation for RDB and RESTORE payload. */
+    int rdb_version_check;          /* Try to load RDB produced by a future version. */
     int jemalloc_bg_thread;         /* Enable jemalloc background thread */
     size_t active_defrag_ignore_bytes; /* minimum amount of fragmentation waste to start active defrag */
     int active_defrag_threshold_lower; /* minimum percentage of fragmentation to start active defrag */