use endian.h for expires too

build.yaml

@@ -21,13 +21,13 @@ machamp:
         parent: make-build
         # https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
         builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
-        command: ./runtest --clients 4 --verbose --tls
+        command: ./runtest --clients 4 --verbose 
       cluster-test:
         type: cmd
         parent: make-build
         # https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
         builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
-        command: ./runtest-cluster --tls
+        command: ./runtest-cluster 
       sentinel-test:
         type: cmd
         parent: make-build

pkg/deb/master_changelog

@@ -1,86 +1,3 @@
-keydb (6:6.3.4-1distribution_placeholder) codename_placeholder; urgency=medium
-
-  * Add keydb_modstatsd, a module for providing keydb stats to a local statsd server. Can be found under src/modules/keydb_modstatsd
-  * Fixed FreeBSD compile(Thanks to @bra-fsn)
-  * Fixed a bug in s3 bucket config that blocked startup(Thanks to Alexandre Chichmanian)
-  * Fixed a bug causing crash if keys command is called after a blocking command (Fixes issues #659, #605, #532)
-  * Added proper error checking of replica configs, now KeyDB will throw an error if replica-of config is passed before active-replica or multimaster configs
-  * Fixed double free bug in lazy free
-  * Added "overload-protect-percent" config, when enabled this will load shed clients whenever CPU usage exceeds configured value
-  * Added "availability-zone" config, this can be passed any string which will then be reported by info command
-  * Updated eviction logic to account for total system memory availability using sysinfo
-  * Fixed a bug where repl-backlog-size config was modified in keydb.conf with the runtime value during config rewrite
-  * Fixed a possible deadlock when running CLIENT KILL with large number of clients
-  * Fixed a bug where KeyDB would overcommit memory during fork BG save
-  * Fixed a bug with disk repl_backlog causing double free
-  * Fixed integer overflow issue in the temp rdb file naming (Thanks to @karthyuom)
-  * Fixed compile issue with GCC 13.1.1 (Thanks to @michieldwitte)
-  * Removed expireset and restored redis expire behaviour
-  * Fixed a bug causing forked processes to hang, specifically affecting RDB and AOF(Fixes issues #675, #619)
-  * Added "CLUSTER REPLICATE NO ONE" to turn a replica into an empty primary
-  * Added RDB-less full sync, can be enabled with config 'enable-keydb-fastsync'
-  * Added "flash-disable-key-cache" config to disable key cache which stores every key(but not value) in memory
-  * Moved cluster slot to key map to being stored by storage provider, rather than in memory
-  * Moved expires to be stored by storage provider, rather than in memory
-  * Enabled expiry/eviction from storage provider, previously only data stored in memory could be expired/evicted(Fixes #645 along with the 2 previous updates)
-  * Fixed a bug where swapdb result was not recovered after keydb restarts in FLASH mode (Thanks to @karthyuom)
-  * Fixed double free bug when fast sync was canceled early
-  * Fixed a bug where a temp rdb file with zero bytes is generated in flash mode (Thanks to @karthyuom)
-  * Fixed a bug where flash CF options are being reset to default after flushall (Thanks to @karthyuom)(Fixes issue #717)
-  * Updated debug reload command to work with flash
-  * Added support for KEYDB_PASSWORD env variable(Thanks to @einar-pexip)
-  * Compile with flash in Dockerfile(Thanks to @der-eismann)
-  * Fix broken redis-cli symlink (Thanks to @pimvandenbroek)
-  * Use tini for alpine docker(Thanks to @rofafor)
-
--- Malavan Sotheeswaran <msotheeswaran@snapchat.com>  Sun, 29 Oct 2023 8:39:21 +0000
-
-keydb (6:6.3.3-1distribution_placeholder) codename_placeholder; urgency=medium
-
-  * To help accelerate our development efforts for KeyDB, this will be the last release containing support for Centos 7, Ubuntu 16.04, Ubuntu 18.04, Debian 9 and 32-bit builds
-  * This release contains fixes for 11 issues along with improvements to the KeyDB FLASH feature:
-  * Fixed race condition with expireset access (Issue #597)
-  * Fixed keys command with lua, and added keydb as an alternative to redis as lua variable to access db (Issue #562 and #594)
-  * Fixed hang in aof child (Issue #554)
-  * Fixed leaking fds from RDB save (Issue #453 and #584)
-  * Added config for S3 RDB load/save (Issue #584)
-  * Enabled active defrag during forkless background save to improve average memory efficiency (Issue #460)
-  * Fixed crash with fork background save during replication (Issue #567)
-  * RocksDB has been updated to v7.9.2
-  * Fixed missing slot_to_key map in FLASH cluster mode (Issue #574)
-  * Added keyspace notifications for keys loaded from FLASH at startup
-  * Fixed race condition in prefetchKeysAsync with FLASH enabled (Issue #571)
-  * Fixed integer overflow in rand family of commands(Issue #631, #632, #633)
-  * Fixed bad value in hincrbyfloat(Issue #634)
-  * Fixed OOM hang in rand family of commands(Issue #635)
-  * Added config to limit count of return values in rand family of commands(Issue #636)
-
--- Ben Schermel <ben@eqalpha.com>  Tue, 18 Apr 2023 10:00:37 +0000
-
-keydb (6:6.3.2-1distribution_placeholder) codename_placeholder; urgency=medium
-
-  * This release contains Beta level support for KeyDB FLASH, new ASYNC commands, latency improvements and a number of bug fixes. 
-  * KeyDB FLASH is included as a Beta feature. FLASH allows KeyDB to persist data to the storage medium it is written to, avoiding the need for AOF/RDB files. KeyDB uses rocksdb as the persistent storage provider and can be enabled with config "storage-provider flash /path/to/rocksdb/output". Read more at https://docs.keydb.dev/docs/flash/
-  * In addition to GET/MGET, ASYNC support has been added for the following commands: HGET, HMGET, HKEYS, HVALS, HGETALL, HSCAN and can be enable with config "enable-async-commands yes”
-  * Packaging support for Ubuntu 22.04 (Jammy) and Debian 12 (Bookworm) has been included with this release
-  * Added new soft shutdown feature, can be enabled with config "soft-shutdown yes".
-  * If soft shutdown is enabled, instead of shutting down right away, the server will wait until all clients have disconnected, and will reject all new connection attempts.
-  * Fixed memory leak with tls certificates when tls allowlist is enabled
-  * Fixed bug in rdb load with flash enabled to ensure all dbs are safe to load (previously only checked db[0])
-  * Fixed race conditions in rdb load and replication
-  * Fixed memory access of rdb file after it should have been deleted
-  * Fixed integer overflow bug in flash(Issue #486)
-  * Improve TLS latency by queueing new commands before executing instead of after
-  * Removed O(n) count of memory usage from info command(replaced by O(1) estimate)
-  * Improved latency of clearing large number of flash DBs(Thanks to Paul Chen for this fix)(Issue #516)
-  * replaced sprintf with snprintf to avoid potential security bugs
-  * Fixed bug where a failed move due to key already existing in move target would result in the key being removed from move source(Thanks to Paul Chen for this fix)(Issue #497)
-  * Fixed usage of deprecated OpenSSL api in OpenSSL v>3.0.1(Issue #392)
-  * Imported security fixes from Redis (CVE-2023-22458 CVE-2022-35977)
-  * Other fixed issues: #480 #477 #454 #452 #303 #425 #492 #541
-
--- Ben Schermel <ben@eqalpha.com>  Fri, 20 Jan 2023 20:00:37 +0000
-
 keydb (6:6.3.1-1distribution_placeholder) codename_placeholder; urgency=medium
 
   * This point release contains fixes to bugs related to expires, active-rep, and rdb saving

src/SnapshotPayloadParseState.cpp

@@ -136,33 +136,14 @@ void SnapshotPayloadParseState::flushQueuedKeys() {
     int idb = current_database;
     serverAssert(vecqueuedKeys.size() == vecqueuedVals.size());
     auto sizePrev = vecqueuedKeys.size();
+    (*insertsInFlight)++;
+    std::weak_ptr<std::atomic<int>> insertsInFlightTmp = insertsInFlight; // C++ GRRRRRRRRRRRRRRRR, we don't want to capute "this" because that's dangerous
     if (current_database < cserver.dbnum) {
-        if (g_pserver->m_pstorageFactory) {
-            (*insertsInFlight)++;
-            std::weak_ptr<std::atomic<int>> insertsInFlightTmp = insertsInFlight; // C++ GRRRRRRRRRRRRRRRR, we don't want to capute "this" because that's dangerous
-            g_pserver->asyncworkqueue->AddWorkFunction([idb, vecqueuedKeys = std::move(this->vecqueuedKeys), vecqueuedKeysCb = std::move(this->vecqueuedKeysCb), vecqueuedVals = std::move(this->vecqueuedVals), vecqueuedValsCb = std::move(this->vecqueuedValsCb), insertsInFlightTmp, pallocator = m_spallocator.release()]() mutable {
-                g_pserver->db[idb]->bulkDirectStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
-                (*(insertsInFlightTmp.lock()))--;
-                delete pallocator;
-            });
-        } else {
-            for (size_t ival = 0; ival < vecqueuedKeys.size(); ++ival) {
-                size_t offset = 0;
-                auto spexpire = deserializeExpire(vecqueuedVals[ival], vecqueuedValsCb[ival], &offset);    
-                auto o = deserializeStoredObject(vecqueuedVals[ival] + offset, vecqueuedValsCb[ival] - offset);
-                sds sdsKey = sdsnewlen(vecqueuedKeys[ival], -static_cast<ssize_t>(vecqueuedKeysCb[ival]));
-                if (dbMerge(g_pserver->db[idb], sdsKey, o, false)) {
-                    if (spexpire != nullptr)
-                        g_pserver->db[idb]->setExpire(sdsKey, std::move(*spexpire));
-                } else {
-                    sdsfree(sdsKey);
-                }
-            }
-            vecqueuedKeys.clear();
-            vecqueuedKeysCb.clear();
-            vecqueuedVals.clear();
-            vecqueuedValsCb.clear();
-        }
+        g_pserver->asyncworkqueue->AddWorkFunction([idb, vecqueuedKeys = std::move(this->vecqueuedKeys), vecqueuedKeysCb = std::move(this->vecqueuedKeysCb), vecqueuedVals = std::move(this->vecqueuedVals), vecqueuedValsCb = std::move(this->vecqueuedValsCb), insertsInFlightTmp, pallocator = m_spallocator.release()]() mutable {
+            g_pserver->db[idb]->bulkDirectStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
+            (*(insertsInFlightTmp.lock()))--;
+            delete pallocator;
+        });
     } else {
         // else drop the data
         vecqueuedKeys.clear();

src/config.cpp

@@ -2983,7 +2983,6 @@ standardConfig configs[] = {
     createIntConfig("overload-protect-percent", NULL, MODIFIABLE_CONFIG, 0, 200, g_pserver->overload_protect_threshold, 0, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("force-eviction-percent", NULL, MODIFIABLE_CONFIG, 0, 100, g_pserver->force_eviction_percent, 0, INTEGER_CONFIG, NULL, NULL),
     createBoolConfig("enable-async-rehash", NULL, MODIFIABLE_CONFIG, g_pserver->enable_async_rehash, 1, NULL, NULL),
-    createBoolConfig("enable-keydb-fastsync", NULL, MODIFIABLE_CONFIG, g_pserver->fEnableFastSync, 0, NULL, NULL),
 
 #ifdef USE_OPENSSL
     createIntConfig("tls-port", NULL, MODIFIABLE_CONFIG, 0, 65535, g_pserver->tls_port, 0, INTEGER_CONFIG, NULL, updateTLSPort), /* TCP port. */

src/db.cpp

@@ -56,6 +56,7 @@ int expireIfNeeded(redisDb *db, robj *key, robj *o);
 void slotToKeyUpdateKeyCore(const char *key, size_t keylen, int add);
 
 std::unique_ptr<expireEntry> deserializeExpire(const char *str, size_t cch, size_t *poffset);
+sds serializeStoredObjectAndExpire(robj_roptr o);
 
 dictType dictChangeDescType {
     dictSdsHash,                /* hash function */

src/networking.cpp

@@ -163,6 +163,7 @@ client *createClient(connection *conn, int iel) {
     c->multibulklen = 0;
     c->bulklen = -1;
     c->sentlen = 0;
+    c->sentlenAsync = 0;
     c->flags = 0;
     c->fPendingAsyncWrite = FALSE;
     c->fPendingAsyncWriteHandler = FALSE;

src/redis-cli.c

@@ -2524,10 +2524,7 @@ cleanup:
 
 static int clusterManagerNodeConnect(clusterManagerNode *node) {
     if (node->context) redisFree(node->context);
-    struct timeval tv;
-    tv.tv_sec = config.cluster_manager_command.timeout / 1000;
-    tv.tv_usec = (config.cluster_manager_command.timeout % 1000) * 1000;
-    node->context = redisConnectWithTimeout(node->ip, node->port, tv);
+    node->context = redisConnect(node->ip, node->port);
     if (!node->context->err && config.tls) {
         const char *err = NULL;
         if (cliSecureConnection(node->context, config.sslconfig, &err) == REDIS_ERR && err) {
@@ -5629,10 +5626,7 @@ static int clusterManagerCommandImport(int argc, char **argv) {
     char *reply_err = NULL;
     redisReply *src_reply = NULL;
     // Connect to the source node.
-    struct timeval tv;
-    tv.tv_sec = config.cluster_manager_command.timeout / 1000;
-    tv.tv_usec = (config.cluster_manager_command.timeout % 1000) * 1000;
-    redisContext *src_ctx = redisConnectWithTimeout(src_ip, src_port, tv);
+    redisContext *src_ctx = redisConnect(src_ip, src_port);
     if (src_ctx->err) {
         success = 0;
         fprintf(stderr,"Could not connect to KeyDB at %s:%d: %s.\n", src_ip,

src/replication.cpp

@@ -1019,6 +1019,8 @@ public:
     void addReplica(client *replica) {
         replicas.push_back(replica);
         replicationSetupSlaveForFullResync(replica,getPsyncInitialOffset());
+        // Optimize the socket for bulk transfer
+        //connDisableTcpNoDelay(replica->conn);
     }
 
     bool isActive() const { return !replicas.empty(); }
@@ -1091,7 +1093,7 @@ public:
         reply->used += size;
     }
 
-    void addLongLongWithPrefix(long long val, char prefix) {
+    void addLongWithPrefix(long val, char prefix) {
         char buf[128];
         int len;
 
@@ -1103,19 +1105,15 @@ public:
     }
 
     void addArrayLen(int len) {
-        addLongLongWithPrefix(len, '*');
+        addLongWithPrefix(len, '*');
     }
 
     void addLong(long val) {
-        addLongLongWithPrefix(val, ':');
-    }
-
-    void addLongLong(long long val) {
-        addLongLongWithPrefix(val, ':');
+        addLongWithPrefix(val, ':');
     }
 
     void addString(const char *s, unsigned long len) {
-        addLongLongWithPrefix(len, '$');
+        addLongWithPrefix(len, '$');
         addData(s, len);
         addData("\r\n", 2);
     }
@@ -1151,19 +1149,22 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
     // TODO: This needs to be on a background thread
     int retval = C_OK;
     serverAssert(GlobalLocksAcquired());
-    serverLog(LL_NOTICE, "Starting fast full sync with target: %s", "disk");
+    serverLog(LL_NOTICE, "Starting storage provider fast full sync with target: %s", "disk");
 
     std::shared_ptr<replicationBuffer> spreplBuf = std::make_shared<replicationBuffer>();
     listNode *ln;
     listIter li;
+    client *replica = nullptr;
     listRewind(g_pserver->slaves, &li);
-    while ((ln = listNext(&li))) {
+    while (replica == nullptr && (ln = listNext(&li))) {
         client *replicaCur = (client*)listNodeValue(ln);
-        if ((replicaCur->slave_capa & SLAVE_CAPA_KEYDB_FASTSYNC) && (replicaCur->replstate == SLAVE_STATE_WAIT_BGSAVE_START)) {
-            spreplBuf->addReplica(replicaCur);
+        if ((replicaCur->slave_capa & SLAVE_CAPA_ROCKSDB_SNAPSHOT) && (replicaCur->replstate == SLAVE_STATE_WAIT_BGSAVE_START)) {
+            replica = replicaCur;
+            spreplBuf->addReplica(replica);
             replicaCur->replstate = SLAVE_STATE_FASTSYNC_TX;
         }
     }
+    serverAssert(replica != nullptr);
 
     spreplBuf->addArrayLen(2);   // Two sections: Metadata and databases
 
@@ -1178,7 +1179,7 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
             spreplBuf->addString(rsi->repl_id, CONFIG_RUN_ID_SIZE);
         spreplBuf->addArrayLen(2);
             spreplBuf->addString("repl-offset", 11);
-            spreplBuf->addLongLong(rsi->master_repl_offset);
+            spreplBuf->addLong(rsi->master_repl_offset);
 
     if (dictSize(g_pserver->lua_scripts)) {
         dictEntry *de;
@@ -1194,13 +1195,13 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
         di = NULL; /* So that we don't release it again on error. */
     }
 
-    std::shared_ptr<std::vector<const redisDbPersistentDataSnapshot*>> spvecsnapshot = std::make_shared<std::vector<const redisDbPersistentDataSnapshot*>>();
+    std::shared_ptr<std::vector<std::unique_ptr<const StorageCache>>> spvecspsnapshot = std::make_shared<std::vector<std::unique_ptr<const StorageCache>>>();
     for (int idb = 0; idb < cserver.dbnum; ++idb) {
-        spvecsnapshot->emplace_back(g_pserver->db[idb]->createSnapshot(getMvccTstamp(), false));
+        spvecspsnapshot->emplace_back(g_pserver->db[idb]->CloneStorageCache());
     }
     aeReleaseLock();
 
-    g_pserver->asyncworkqueue->AddWorkFunction([spreplBuf = std::move(spreplBuf), spvecsnapshot = std::move(spvecsnapshot)]{
+    g_pserver->asyncworkqueue->AddWorkFunction([spreplBuf = std::move(spreplBuf), spvecspsnapshot = std::move(spvecspsnapshot)]{
         int retval = C_OK;
         auto timeStart = ustime();
         auto lastLogTime = timeStart;
@@ -1210,16 +1211,15 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
         // Databases
         replBuf.addArrayLen(cserver.dbnum);
         for (int idb = 0; idb < cserver.dbnum; ++idb) {
-            auto &spsnapshot = (*spvecsnapshot)[idb];
-            size_t snapshotDeclaredCount = spsnapshot->size();
+            auto &spsnapshot = (*spvecspsnapshot)[idb];
+            size_t snapshotDeclaredCount = spsnapshot->count();
             replBuf.addArrayLen(snapshotDeclaredCount);
             size_t count = 0;
-            bool result = spsnapshot->iterate_threadsafe([&replBuf, &count, &cbData, &lastLogTime, &cbLastUpdate](const char *strKey, robj_roptr o) -> bool{
+            bool result = spsnapshot->enumerate([&replBuf, &count, &cbData, &lastLogTime, &cbLastUpdate](const char *rgchKey, size_t cchKey, const void *rgchVal, size_t cchVal) -> bool{
                 replBuf.addArrayLen(2);
 
-                replBuf.addString(strKey, sdslen(strKey));
-                sds strT = serializeStoredObjectAndExpire(o);
-                replBuf.addString(strT, sdslen(strT));
+                replBuf.addString(rgchKey, cchKey);
+                replBuf.addString((const char *)rgchVal, cchVal);
                 ++count;
                 if ((count % 8092) == 0) {
                     auto curTime = ustime();
@@ -1230,8 +1230,7 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
                         lastLogTime = ustime();
                     }
                 }
-                cbData += sdslen(strKey) + sdslen(strT);
-                sdsfree(strT);
+                cbData += cchKey + cchVal;
                 return replBuf.isActive();
             });
 
@@ -1303,7 +1302,7 @@ int startBgsaveForReplication(int mincapa) {
     /* Only do rdbSave* when rsiptr is not NULL,
      * otherwise replica will miss repl-stream-db. */
     if (rsiptr) {
-        if (mincapa & SLAVE_CAPA_KEYDB_FASTSYNC && FFastSyncEnabled())
+        if (mincapa & SLAVE_CAPA_ROCKSDB_SNAPSHOT && g_pserver->m_pstorageFactory)
             retval = rdbSaveSnapshotForReplication(rsiptr);
         else if (socket_target)
             retval = rdbSaveToSlavesSockets(rsiptr);
@@ -1482,7 +1481,7 @@ void syncCommand(client *c) {
     }
 
     /* CASE 0: Fast Sync */
-    if (c->slave_capa & SLAVE_CAPA_KEYDB_FASTSYNC && FFastSyncEnabled()) {
+    if ((c->slave_capa & SLAVE_CAPA_ROCKSDB_SNAPSHOT) && g_pserver->m_pstorageFactory) {
         serverLog(LL_NOTICE,"Fast SYNC on next replication cycle");
     /* CASE 1: BGSAVE is in progress, with disk target. */
     } else if (g_pserver->FRdbSaveInProgress() &&
@@ -1673,8 +1672,8 @@ void replconfCommand(client *c) {
                 c->slave_capa |= SLAVE_CAPA_PSYNC2;
             else if (!strcasecmp((const char*)ptrFromObj(c->argv[j+1]), "activeExpire"))
                 c->slave_capa |= SLAVE_CAPA_ACTIVE_EXPIRE;
-            else if (!strcasecmp((const char*)ptrFromObj(c->argv[j+1]), "keydb-fastsync"))
-                c->slave_capa |= SLAVE_CAPA_KEYDB_FASTSYNC;
+            else if (!strcasecmp((const char*)ptrFromObj(c->argv[j+1]), "rocksdb-snapshot-load"))
+                c->slave_capa |= SLAVE_CAPA_ROCKSDB_SNAPSHOT;
 
             fCapaCommand = true;
         } else if (!strcasecmp((const char*)ptrFromObj(c->argv[j]),"ack")) {
@@ -1739,14 +1738,10 @@ void replconfCommand(client *c) {
 
     if (fCapaCommand) {
         sds reply = sdsnew("+OK");
-        if (g_pserver->fActiveReplica) {
+        if (g_pserver->fActiveReplica)
             reply = sdscat(reply, " active-replica");
-        }
-        if ((c->slave_capa & SLAVE_CAPA_KEYDB_FASTSYNC) && FFastSyncEnabled()) {
-            reply = sdscat(reply, " keydb-fastsync-save");
-        } else {
-            c->slave_capa = (c->slave_capa & (~SLAVE_CAPA_KEYDB_FASTSYNC)); // never try to fast sync for this as they won't expect it
-        }
+        if (g_pserver->m_pstorageFactory && (c->slave_capa & SLAVE_CAPA_ROCKSDB_SNAPSHOT) && !g_pserver->fActiveReplica)
+            reply = sdscat(reply, " rocksdb-snapshot-save");
         reply = sdscat(reply, "\r\n");
         addReplySds(c, reply);
     } else {
@@ -2520,7 +2515,6 @@ bool readSnapshotBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi
         }
     }
 
-    serverAssert(mi->parseState != nullptr);
     for (int iter = 0; iter < 10; ++iter) {
         if (mi->parseState->shouldThrottle())
             return false;
@@ -2531,16 +2525,11 @@ bool readSnapshotBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi
 
         auto nread = connRead(conn, mi->bulkreadBuffer+qblen, readlen);
         if (nread <= 0) {
-            if (connGetState(conn) != CONN_STATE_CONNECTED) {
-                serverLog(LL_WARNING,"I/O error trying to sync with MASTER: %s",
-                    (nread == -1) ? strerror(errno) : "connection lost");
+            if (connGetState(conn) != CONN_STATE_CONNECTED)
                 cancelReplicationHandshake(mi, true);
-            }
             return false;
         }
-        g_pserver->stat_net_input_bytes += nread;
         mi->repl_transfer_lastio = g_pserver->unixtime;
-        mi->repl_transfer_read += nread;
         sdsIncrLen(mi->bulkreadBuffer,nread);
 
         size_t offset = 0;
@@ -2612,9 +2601,8 @@ bool readSnapshotBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi
             rsi.repl_stream_db = mi->parseState->getMetaDataLongLong("repl-stream-db");
             rsi.repl_offset = mi->parseState->getMetaDataLongLong("repl-offset");
             sds str = mi->parseState->getMetaDataStr("repl-id");
-            if (sdslen(str) == CONFIG_RUN_ID_SIZE) {
+            if (sdslen(str) == CONFIG_RUN_ID_SIZE+1) {
                 memcpy(rsi.repl_id, str, CONFIG_RUN_ID_SIZE+1);
-                rsi.repl_id_is_set = 1;
             }
 
             fFinished = true;
@@ -3013,7 +3001,7 @@ void readSyncBulkPayload(connection *conn) {
         return;
     }
 
-    if (mi->isKeydbFastsync) {
+    if (mi->isRocksdbSnapshotRepl) {
         if (!readSnapshotBulkPayload(conn, mi, rsi))
             return;
     } else {
@@ -3026,15 +3014,17 @@ void readSyncBulkPayload(connection *conn) {
 
     /* Final setup of the connected slave <- master link */
     replicationCreateMasterClient(mi,mi->repl_transfer_s,rsi.repl_stream_db);
-    if (mi->isKeydbFastsync) {
+    if (mi->isRocksdbSnapshotRepl) {
         /* We need to handle the case where the initial querybuf data was read by fast sync */
         /* This should match the work readQueryFromClient would do for a master client */
         mi->master->querybuf = sdscatsds(mi->master->querybuf, mi->bulkreadBuffer);
-        mi->master->pending_querybuf = sdscatsds(mi->master->pending_querybuf, mi->bulkreadBuffer);
-        mi->master->read_reploff += sdslen(mi->bulkreadBuffer);
-
         sdsfree(mi->bulkreadBuffer);
         mi->bulkreadBuffer = nullptr;
+
+        mi->master->pending_querybuf = sdscatlen(mi->master->pending_querybuf,
+            mi->master->querybuf,sdslen(mi->master->querybuf));
+
+        mi->master->read_reploff += sdslen(mi->master->querybuf);
     }
     mi->repl_transfer_s = nullptr;
     mi->repl_state = REPL_STATE_CONNECTED;
@@ -3073,13 +3063,13 @@ void readSyncBulkPayload(connection *conn) {
     }
 
     /* Send the initial ACK immediately to put this replica in online state. */
-    if (usemark || mi->isKeydbFastsync) replicationSendAck(mi);
+    if (usemark || mi->isRocksdbSnapshotRepl) replicationSendAck(mi);
 
     /* Restart the AOF subsystem now that we finished the sync. This
      * will trigger an AOF rewrite, and when done will start appending
      * to the new file. */
     if (g_pserver->aof_enabled) restartAOFAfterSYNC();
-    if (mi->isKeydbFastsync)
+    if (mi->isRocksdbSnapshotRepl)
         readQueryFromClient(conn); // There may be querybuf data we just appeneded
     return;
 }
@@ -3405,15 +3395,15 @@ void parseMasterCapa(redisMaster *mi, sds strcapa)
     char *pchEnd = szStart;
 
     mi->isActive = false;
-    mi->isKeydbFastsync = false;
+    mi->isRocksdbSnapshotRepl = false;
     for (;;)
     {
         if (*pchEnd == ' ' || *pchEnd == '\0') {
             // Parse the word
             if (strncmp(szStart, "active-replica", pchEnd - szStart) == 0) {
                 mi->isActive = true;
-            } else if (strncmp(szStart, "keydb-fastsync-save", pchEnd - szStart) == 0) {
-                mi->isKeydbFastsync = true;
+            } else if (strncmp(szStart, "rocksdb-snapshot-save", pchEnd - szStart) == 0) {
+                mi->isRocksdbSnapshotRepl = true;
             }
             szStart = pchEnd + 1;
         }
@@ -3560,9 +3550,9 @@ retry_connect:
             "capa","psync2",
             "capa","activeExpire",
         };
-        if (FFastSyncEnabled() && g_pserver->repl_diskless_load != REPL_DISKLESS_LOAD_SWAPDB) {
+        if (g_pserver->m_pstorageFactory && !g_pserver->fActiveReplica && g_pserver->repl_diskless_load != REPL_DISKLESS_LOAD_SWAPDB) {
             veccapabilities.push_back("capa");
-            veccapabilities.push_back("keydb-fastsync");
+            veccapabilities.push_back("rocksdb-snapshot-load");
         }
 
         err = sendCommandArgv(conn, veccapabilities.size(), veccapabilities.data(), nullptr);
@@ -3745,7 +3735,7 @@ retry_connect:
     }
 
     /* Prepare a suitable temp file for bulk transfer */
-    if (!useDisklessLoad() && !mi->isKeydbFastsync) {
+    if (!useDisklessLoad() && !mi->isRocksdbSnapshotRepl) {
         while(maxtries--) {
             auto dt = std::chrono::system_clock::now().time_since_epoch();
             auto dtMillisecond = std::chrono::duration_cast<std::chrono::milliseconds>(dt);
@@ -3874,10 +3864,6 @@ int cancelReplicationHandshake(redisMaster *mi, int reconnect) {
         delete mi->parseState;
         mi->parseState = nullptr;
     }
-    if (mi->bulkreadBuffer) {
-        sdsfree(mi->bulkreadBuffer);
-        mi->bulkreadBuffer = nullptr;
-    }
 
     if (mi->repl_state == REPL_STATE_TRANSFER) {
         replicationAbortSyncTransfer(mi);
@@ -4343,6 +4329,7 @@ void replicationCacheMaster(redisMaster *mi, client *c) {
     if (c->flags & CLIENT_MULTI) discardTransaction(c);
     listEmpty(c->reply);
     c->sentlen = 0;
+    c->sentlenAsync = 0;
     c->reply_bytes = 0;
     c->bufpos = 0;
     resetClient(c);

src/sds.h

@@ -418,10 +418,7 @@ public:
     sdsstring &operator=(const sdsstring &other)
     {
         sdsfree(m_str);
-        if (other.m_str != nullptr)
-            m_str = sdsdup(other.m_str);
-        else
-            m_str = nullptr;
+        m_str = sdsdup(other.m_str);
         return *this;
     }
 

src/server.cpp

@@ -5102,8 +5102,8 @@ int processCommand(client *c, int callFlags) {
 
             sds str = sdsempty();
             for (int j = 0; j < numkeys; j++) {
-                str = sdscatsds(str, (sds)ptrFromObj(c->argv[keyindex[j]]));
-                str = sdscat(str, " ");
+                sdscatsds(str, (sds)ptrFromObj(c->argv[keyindex[j]]));
+                sdscat(str, " ");
             }
         
             if (numkeys > 0)

src/server.h

@@ -620,7 +620,7 @@ typedef enum {
 #define SLAVE_CAPA_EOF (1<<0)    /* Can parse the RDB EOF streaming format. */
 #define SLAVE_CAPA_PSYNC2 (1<<1) /* Supports PSYNC2 protocol. */
 #define SLAVE_CAPA_ACTIVE_EXPIRE (1<<2) /* Will the slave perform its own expirations? (Don't send delete) */
-#define SLAVE_CAPA_KEYDB_FASTSYNC (1<<3)
+#define SLAVE_CAPA_ROCKSDB_SNAPSHOT (1<<3)
 
 /* Synchronous read timeout - replica side */
 #define CONFIG_REPL_SYNCIO_TIMEOUT 5
@@ -1610,6 +1610,7 @@ struct client {
     unsigned long long reply_bytes; /* Tot bytes of objects in reply list. */
     size_t sentlen;         /* Amount of bytes already sent in the current
                                buffer or object being sent. */
+    size_t sentlenAsync;    /* same as sentlen buf for async buffers (which are a different stream) */
     time_t ctime;           /* Client creation time. */
     long duration;          /* Current command duration. Used for measuring latency of blocking/non-blocking cmds */
     time_t lastinteraction; /* Time of the last interaction, used for timeout */
@@ -1870,7 +1871,7 @@ struct redisMaster {
     long long master_initial_offset;           /* Master PSYNC offset. */
 
     bool isActive = false;
-    bool isKeydbFastsync = false;
+    bool isRocksdbSnapshotRepl = false;
     int repl_state;          /* Replication status if the instance is a replica */
     off_t repl_transfer_size; /* Size of RDB to read from master during sync. */
     off_t repl_transfer_read; /* Amount of RDB read from master during sync. */
@@ -2687,7 +2688,6 @@ struct redisServer {
 
     int fActiveReplica;                          /* Can this replica also be a master? */
     int fWriteDuringActiveLoad;                  /* Can this active-replica write during an RDB load? */
-    int fEnableFastSync = false;
 
     // Format:
     //  Lower 20 bits: a counter incrementing for each command executed in the same millisecond
@@ -3182,7 +3182,6 @@ void trimStringObjectIfNeeded(robj *o);
 robj *deserializeStoredObject(const void *data, size_t cb);
 std::unique_ptr<expireEntry> deserializeExpire(const char *str, size_t cch, size_t *poffset);
 sds serializeStoredObject(robj_roptr o, sds sdsPrefix = nullptr);
-sds serializeStoredObjectAndExpire(robj_roptr o);
 
 #define sdsEncodedObject(objptr) (objptr->encoding == OBJ_ENCODING_RAW || objptr->encoding == OBJ_ENCODING_EMBSTR)
 
@@ -3972,10 +3971,6 @@ inline int ielFromEventLoop(const aeEventLoop *eventLoop)
     return iel;
 }
 
-inline bool FFastSyncEnabled() {
-    return g_pserver->fEnableFastSync && !g_pserver->fActiveReplica;
-}
-
 inline int FCorrectThread(client *c)
 {
     return (c->conn == nullptr)

src/version.h

@@ -1,5 +1,5 @@
-#define KEYDB_REAL_VERSION "6.3.4"
-#define KEYDB_VERSION_NUM 0x00060304
+#define KEYDB_REAL_VERSION "255.255.255"
+#define KEYDB_VERSION_NUM 0x00ffffff
 extern const char *KEYDB_SET_VERSION;   // Unlike real version, this can be overriden by the config
 
 enum VersionCompareResult

tests/integration/replication.tcl

@@ -584,7 +584,6 @@ start_server {tags {"repl"}} {
     set master [srv 0 client]
     $master config set repl-diskless-sync yes
     $master config set repl-diskless-sync-delay 1
-    $master config set enable-keydb-fastsync no
     set master_host [srv 0 host]
     set master_port [srv 0 port]
     set master_pid [srv 0 pid]
@@ -728,8 +727,6 @@ start_server {tags {"repl"}} {
             }
         }
     }
-
-    $master config set enable-keydb-fastsync yes
 }
 
 if 0 {
@@ -870,10 +867,8 @@ test {replicaof right after disconnection} {
 test {Kill rdb child process if its dumping RDB is not useful} {
     start_server {tags {"repl"}} {
         set slave1 [srv 0 client]
-        $slave1 config set enable-keydb-fastsync no
         start_server {} {
             set slave2 [srv 0 client]
-            $slave2 config set enable-keydb-fastsync no
             start_server {} {
                 set master [srv 0 client]
                 set master_host [srv 0 host]

tests/unit/maxmemory.tcl

@@ -182,8 +182,6 @@ proc test_slave_buffers {test_name cmd_count payload_len limit_memory pipeline}
                 fail "Replication not started."
             }
 
-            after 100
-
             # measure used memory after the slave connected and set maxmemory
             set orig_used [s -1 used_memory]
             set orig_client_buf [s -1 mem_clients_normal]

tests/unit/tls-auditlog.tcl

@@ -51,7 +51,7 @@ if {$::tls} {
         }
     }
 
-    test {TLS Audit Log: Able to connect while with SAN having a comprehensive list} {
+    test {TLS: Able to connect while with SAN having a comprehensive list} {
         start_server {tags {"tls"} overrides {tls-auditlog-blocklist {dummy.keydb.dev san2.keydb.dev other.keydb.dev}}} {
             catch {r PING} e
             assert_match {PONG} $e
@@ -65,7 +65,7 @@ if {$::tls} {
         }
     }
 
-    test {TLS Audit Log: Able to connect while with SAN having a comprehensive list with wildcards} {
+    test {TLS Audit LogTLS: Able to connect while with SAN having a comprehensive list with wildcards} {
         start_server {tags {"tls"} overrides {tls-auditlog-blocklist {dummy.* san*.dev other.*}}} {
             catch {r PING} e
             assert_match {PONG} $e