Support Valkey 9.1 rollback compatibility with Redis 6.0

redis.conf

@@ -364,6 +364,15 @@ rdb-del-sync-files no
 # Note that you must specify a directory here, not a file name.
 dir ./
 
+# Redis can try to load an RDB dump produced by a future version of Redis/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
+
 ################################# REPLICATION #################################
 
 # Master-Replica replication. Use replicaof to make a Redis instance a copy of

src/Makefile

@@ -245,7 +245,7 @@ endif
 
 REDIS_SERVER_NAME=redis-server$(PROG_SUFFIX)
 REDIS_SENTINEL_NAME=redis-sentinel$(PROG_SUFFIX)
-REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o gopher.o tracking.o connection.o tls.o sha256.o timeout.o setcpuaffinity.o mt19937-64.o
+REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o gopher.o tracking.o connection.o tls.o sha256.o timeout.o setcpuaffinity.o mt19937-64.o rdb_downgrade_compat.o
 REDIS_CLI_NAME=redis-cli$(PROG_SUFFIX)
 REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o ae.o crcspeed.o crc64.o siphash.o crc16.o mt19937-64.o
 REDIS_BENCHMARK_NAME=redis-benchmark$(PROG_SUFFIX)

src/cluster.c

@@ -31,6 +31,7 @@
 #include "server.h"
 #include "cluster.h"
 #include "endianconv.h"
+#include "rdb_downgrade_compat.h"
 
 #include <sys/types.h>
 #include <sys/socket.h>
@@ -171,13 +172,30 @@ int clusterLoadConfig(char *filename) {
             n = createClusterNode(argv[0],0);
             clusterAddNode(n);
         }
-        /* Address and port */
-        if ((p = strrchr(argv[1],':')) == NULL) {
+        /* Address and port. The format is ip:port[,hostname] */
+        int aux_argc;
+        sds *aux_argv = sdssplitlen(argv[1],sdslen(argv[1]),",",1,&aux_argc);
+        if (aux_argv == NULL) {
+            sdsfreesplitres(argv,argc);
+            goto fmterr;
+        }
+
+        if (aux_argc > 1) {
+            if (sdslen(aux_argv[1]) > 0) {
+                sdsfree(n->hostname);
+                n->hostname = sdsnew(aux_argv[1]);
+            } else {
+                if (n->hostname) sdsclear(n->hostname);
+            }
+        }
+
+        if ((p = strrchr(aux_argv[0],':')) == NULL) {
+            sdsfreesplitres(aux_argv,aux_argc);
             sdsfreesplitres(argv,argc);
             goto fmterr;
         }
         *p = '\0';
-        memcpy(n->ip,argv[1],strlen(argv[1])+1);
+        memcpy(n->ip,aux_argv[0],strlen(aux_argv[0])+1);
         char *port = p+1;
         char *busp = strchr(port,'@');
         if (busp) {
@@ -189,6 +207,10 @@ int clusterLoadConfig(char *filename) {
          * In this case we set it to the default offset of 10000 from the
          * base port. */
         n->cport = busp ? atoi(busp) : n->port + CLUSTER_PORT_INCR;
+        sdsfreesplitres(aux_argv,aux_argc);
+
+        /* The plaintext port for client in a TLS cluster (n->pport) is not
+         * stored in nodes.conf. It is received later over the bus protocol. */
 
         /* Parse flags */
         p = s = argv[2];
@@ -786,6 +808,7 @@ clusterNode *createClusterNode(char *nodename, int flags) {
     node->data_received = 0;
     node->fail_time = 0;
     node->link = NULL;
+    node->hostname = NULL;
     memset(node->ip,0,sizeof(node->ip));
     node->port = 0;
     node->cport = 0;
@@ -956,6 +979,7 @@ void freeClusterNode(clusterNode *n) {
     if (n->link) freeClusterLink(n->link);
     listRelease(n->fail_reports);
     zfree(n->slaves);
+    if (n->hostname) sdsfree(n->hostname);
     zfree(n);
 }
 
@@ -1752,12 +1776,12 @@ int clusterProcessPacket(clusterLink *link) {
 
         explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
         explen += (sizeof(clusterMsgDataGossip)*count);
-        if (totlen != explen) return 1;
+        if (totlen < explen) return 1;
     } else if (type == CLUSTERMSG_TYPE_FAIL) {
         uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 
         explen += sizeof(clusterMsgDataFail);
-        if (totlen != explen) return 1;
+        if (totlen < explen) return 1;
     } else if (type == CLUSTERMSG_TYPE_PUBLISH) {
         uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 
@@ -1777,7 +1801,7 @@ int clusterProcessPacket(clusterLink *link) {
         uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 
         explen += sizeof(clusterMsgDataUpdate);
-        if (totlen != explen) return 1;
+        if (totlen < explen) return 1;
     } else if (type == CLUSTERMSG_TYPE_MODULE) {
         uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 
@@ -4123,11 +4147,15 @@ sds clusterGenNodeDescription(clusterNode *node) {
     sds ci;
 
     /* Node coordinates */
-    ci = sdscatprintf(sdsempty(),"%.40s %s:%d@%d ",
+    ci = sdscatprintf(sdsempty(),"%.40s %s:%d@%d",
         node->name,
         node->ip,
         node->port,
         node->cport);
+    if (node->hostname && sdslen(node->hostname) > 0) {
+        ci = sdscatfmt(ci,",%s",node->hostname);
+    }
+    ci = sdscat(ci," ");
 
     /* Flags */
     ci = representClusterNodeFlags(ci, node->flags);
@@ -4944,7 +4972,7 @@ void createDumpPayload(rio *payload, robj *o, robj *key) {
  * instance and that the checksum is ok.
  * If the DUMP payload looks valid C_OK is returned, otherwise C_ERR
  * is returned. */
-int verifyDumpPayload(unsigned char *p, size_t len) {
+int verifyDumpPayload(unsigned char *p, size_t len, int *rdbverptr) {
     unsigned char *footer;
     uint16_t rdbver;
     uint64_t crc;
@@ -4955,12 +4983,18 @@ int verifyDumpPayload(unsigned char *p, size_t len) {
 
     /* Verify RDB version */
     rdbver = (footer[1] << 8) | footer[0];
-    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;
+    }
 
     /* Verify CRC64 */
     crc = crc64(0,p,len-8);
     memrev64ifbe(&crc);
-    return (memcmp(&crc,footer+2,8) == 0) ? C_OK : C_ERR;
+    if (memcmp(&crc,footer+2,8) != 0) return C_ERR;
+
+    if (rdbverptr) *rdbverptr = rdbver;
+    return C_OK;
 }
 
 /* DUMP keyname
@@ -4988,7 +5022,7 @@ void dumpCommand(client *c) {
 void restoreCommand(client *c) {
     long long ttl, lfu_freq = -1, lru_idle = -1, lru_clock = -1;
     rio payload;
-    int j, type, replace = 0, absttl = 0;
+    int j, type, rdbver, replace = 0, absttl = 0;
     robj *obj;
 
     /* Parse additional options */
@@ -5041,15 +5075,15 @@ void restoreCommand(client *c) {
     }
 
     /* Verify RDB version and data checksum. */
-    if (verifyDumpPayload(c->argv[3]->ptr,sdslen(c->argv[3]->ptr)) == C_ERR)
+    if (verifyDumpPayload(c->argv[3]->ptr,sdslen(c->argv[3]->ptr),&rdbver) == C_ERR)
     {
         addReplyError(c,"DUMP payload version or checksum are wrong");
         return;
     }
 
     rioInitWithBuffer(&payload,c->argv[3]->ptr);
     if (((type = rdbLoadObjectType(&payload)) == -1) ||
-        ((obj = rdbLoadObject(type,&payload,key->ptr)) == NULL))
+        ((obj = rdbLoadObjectCompat(type,&payload,key->ptr,NULL,rdbver)) == NULL))
     {
         addReplyError(c,"Bad data format");
         return;

src/cluster.h

@@ -137,6 +137,7 @@ typedef struct clusterNode {
     int cport;                  /* Latest known cluster port of this node. */
     clusterLink *link;          /* TCP/IP link with this node */
     list *fail_reports;         /* List of nodes signaling this as failing */
+    sds hostname;
 } clusterNode;
 
 typedef struct clusterState {

src/config.c

@@ -113,6 +113,12 @@ configEnum oom_score_adj_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 */
@@ -2327,6 +2333,7 @@ standardConfig configs[] = {
     createEnumConfig("maxmemory-policy", NULL, MODIFIABLE_CONFIG, maxmemory_policy_enum, server.maxmemory_policy, MAXMEMORY_NO_EVICTION, NULL, NULL),
     createEnumConfig("appendfsync", NULL, MODIFIABLE_CONFIG, aof_fsync_enum, server.aof_fsync, AOF_FSYNC_EVERYSEC, NULL, NULL),
     createEnumConfig("oom-score-adj", NULL, MODIFIABLE_CONFIG, oom_score_adj_enum, server.oom_score_adj, OOM_SCORE_ADJ_NO, NULL, updateOOMScoreAdj),
+    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/help.h

@@ -974,7 +974,7 @@ struct commandHelp {
     8,
     "1.0.0" },
     { "SET",
-    "key value [EX seconds|PX milliseconds|KEEPTTL] [NX|XX]",
+    "key value [EX seconds|PX milliseconds|EXAT unix-time-seconds|PXAT unix-time-milliseconds|KEEPTTL] [NX|XX]",
     "Set the string value of a key",
     1,
     "1.0.0" },

src/listpack.c

@@ -383,6 +383,22 @@ uint32_t lpCurrentEncodedSize(unsigned char *p) {
     return 0;
 }
 
+/* Return the number of bytes needed to encode the type and optional string
+ * length of the listpack element pointed to by 'p'. */
+static uint32_t lpCurrentEncodedSizeBytes(unsigned char *p) {
+    if (LP_ENCODING_IS_7BIT_UINT(p[0])) return 1;
+    if (LP_ENCODING_IS_6BIT_STR(p[0])) return 1;
+    if (LP_ENCODING_IS_13BIT_INT(p[0])) return 1;
+    if (LP_ENCODING_IS_16BIT_INT(p[0])) return 1;
+    if (LP_ENCODING_IS_24BIT_INT(p[0])) return 1;
+    if (LP_ENCODING_IS_32BIT_INT(p[0])) return 1;
+    if (LP_ENCODING_IS_64BIT_INT(p[0])) return 1;
+    if (LP_ENCODING_IS_12BIT_STR(p[0])) return 2;
+    if (LP_ENCODING_IS_32BIT_STR(p[0])) return 5;
+    if (p[0] == LP_EOF) return 1;
+    return 0;
+}
+
 /* Skip the current entry returning the next. It is invalid to call this
  * function if the current element is the EOF element at the end of the
  * listpack, however, while this function is used to implement lpNext(),
@@ -752,6 +768,59 @@ uint32_t lpBytes(unsigned char *lp) {
     return lpGetTotalBytes(lp);
 }
 
+/* Validate a listpack entry and advance 'pp' to the next entry. */
+static int lpValidateNext(unsigned char *lp, unsigned char **pp, size_t lpbytes) {
+    unsigned char *p = *pp;
+    if (!p || p < lp + LP_HDR_SIZE || p > lp + lpbytes - 1) return 0;
+
+    if (*p == LP_EOF) {
+        if (p != lp + lpbytes - 1) return 0;
+        *pp = NULL;
+        return 1;
+    }
+
+    size_t offset = p - lp;
+    uint32_t lenbytes = lpCurrentEncodedSizeBytes(p);
+    if (!lenbytes || lenbytes > lpbytes - offset - 1) return 0;
+
+    unsigned long entrylen = lpCurrentEncodedSize(p);
+    unsigned long encoded_backlen = lpEncodeBacklen(NULL, entrylen);
+    if (entrylen > lpbytes - offset - 1 ||
+        encoded_backlen > lpbytes - offset - entrylen - 1) {
+        return 0;
+    }
+
+    p += entrylen + encoded_backlen;
+    if (lpDecodeBacklen(p - 1) != entrylen) return 0;
+
+    *pp = p;
+    return 1;
+}
+
+/* Validate the listpack header and, when 'deep' is non-zero, every entry.
+ * The callback is invoked only after an entry has passed bounds validation. */
+int lpValidateIntegrity(unsigned char *lp, size_t size, int deep,
+                        listpackValidateEntryCB entry_cb, void *cb_userdata) {
+    if (size < LP_HDR_SIZE + 1) return 0;
+    if (lpGetTotalBytes(lp) != size) return 0;
+    if (lp[size - 1] != LP_EOF) return 0;
+    if (!deep) return 1;
+
+    uint32_t count = 0;
+    uint32_t numele = lpGetNumElements(lp);
+    unsigned char *p = lp + LP_HDR_SIZE;
+    while (p && p[0] != LP_EOF) {
+        unsigned char *entry = p;
+        if (!lpValidateNext(lp, &p, size)) return 0;
+        if (entry_cb && !entry_cb(entry, numele, cb_userdata)) return 0;
+        count++;
+    }
+
+    if (p != lp + size - 1) return 0;
+    if (numele != LP_HDR_NUMELE_UNKNOWN && numele != count) return 0;
+    return 1;
+}
+
 /* Seek the specified element and returns the pointer to the seeked element.
  * Positive indexes specify the zero-based element to seek from the head to
  * the tail, negative indexes specify elements starting from the tail, where
@@ -800,4 +869,3 @@ unsigned char *lpSeek(unsigned char *lp, long index) {
         return ele;
     }
 }
-

src/listpack.h

@@ -35,6 +35,7 @@
 #ifndef __LISTPACK_H
 #define __LISTPACK_H
 
+#include <stddef.h>
 #include <stdint.h>
 
 #define LP_INTBUF_SIZE 21 /* 20 digits of -2^63 + 1 null term = 21. */
@@ -57,5 +58,9 @@ unsigned char *lpNext(unsigned char *lp, unsigned char *p);
 unsigned char *lpPrev(unsigned char *lp, unsigned char *p);
 uint32_t lpBytes(unsigned char *lp);
 unsigned char *lpSeek(unsigned char *lp, long index);
+typedef int (*listpackValidateEntryCB)(unsigned char *p, unsigned int head_count,
+                                      void *userdata);
+int lpValidateIntegrity(unsigned char *lp, size_t size, int deep,
+                        listpackValidateEntryCB entry_cb, void *cb_userdata);
 
 #endif