init iteration

compile fix
somehow missed reference to m_setexpire
2023-08-22 22:48:26 -07:00 · 2023-08-22 13:18:14 -07:00 · 2023-08-22 13:03:03 -07:00 · 2023-08-22 12:48:28 -07:00 · 2023-06-29 21:50:05 -07:00 · 2023-06-28 17:44:36 -07:00
40 changed files with 213 additions and 487 deletions
@@ -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
@@ -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
@@ -53,7 +53,7 @@ RUN set -eux; \
        grep -E '^ *createBoolConfig[(]"protected-mode",.*, *1 *,.*[)],$' ./src/config.cpp; \
        sed -ri 's!^( *createBoolConfig[(]"protected-mode",.*, *)1( *,.*[)],)$!\10\2!' ./src/config.cpp; \
        grep -E '^ *createBoolConfig[(]"protected-mode",.*, *0 *,.*[)],$' ./src/config.cpp; \
-        make -j$(nproc) BUILD_TLS=yes ENABLE_FLASH=yes; \
+        make -j$(nproc) BUILD_TLS=yes; \
        cd src; \
        strip keydb-cli keydb-benchmark keydb-check-rdb keydb-check-aof keydb-diagnostic-tool keydb-sentinel keydb-server; \
        mv keydb-server keydb-cli keydb-benchmark keydb-check-rdb keydb-check-aof keydb-diagnostic-tool keydb-sentinel /usr/local/bin/; \
@@ -83,7 +83,6 @@ RUN \
        sed -i 's/^\(logfile .*\)$/# \1/' /etc/keydb/keydb.conf; \
        sed -i 's/protected-mode yes/protected-mode no/g' /etc/keydb/keydb.conf; \
        sed -i 's/^\(bind .*\)$/# \1/' /etc/keydb/keydb.conf; \
-        cd /usr/local/bin; \
        ln -s keydb-cli redis-cli; \
        cd /etc/keydb; \
        ln -s keydb.conf redis.conf; \
@@ -97,10 +96,6 @@ RUN set -eux; \
        echo 'if [ "${1#-}" != "$1" ] || [ "${1%.conf}" != "$1" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
        echo '        set -- keydb-server "$@"' >> /usr/local/bin/docker-entrypoint.sh; \
        echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
-        echo '# if KEYDB_PASSWORD is set, add it to the arguments' >> /usr/local/bin/docker-entrypoint.sh; \
-        echo 'if [ -n "$KEYDB_PASSWORD" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
-        echo '        set -- "$@" --requirepass "${KEYDB_PASSWORD}"' >> /usr/local/bin/docker-entrypoint.sh; \
-        echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
        echo "# allow the container to be started with `--user`" >> /usr/local/bin/docker-entrypoint.sh; \
        echo 'if [ "$1" = "keydb-server" -a "$(id -u)" = "0" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
        echo "        find . \! -user keydb -exec chown keydb '{}' +" >> /usr/local/bin/docker-entrypoint.sh; \
@@ -1,11 +1,11 @@
-FROM alpine:3.18
+FROM alpine:3.12
 # add our user and group first to make sure their IDs get assigned consistently, regardless of whatever dependencies get added
 RUN addgroup -S -g 1000 keydb && adduser -S -G keydb -u 999 keydb
 RUN mkdir -p /etc/keydb
 ARG BRANCH
 RUN set -eux; \
        \
-        apk add --no-cache su-exec tini; \
+        apk add --no-cache su-exec; \
        apk add --no-cache --virtual .build-deps \
                coreutils \
                gcc \
@@ -32,7 +32,7 @@ RUN set -eux; \
        grep -E '^ *createBoolConfig[(]"protected-mode",.*, *1 *,.*[)],$' ./src/config.cpp; \
        sed -ri 's!^( *createBoolConfig[(]"protected-mode",.*, *)1( *,.*[)],)$!\10\2!' ./src/config.cpp; \
        grep -E '^ *createBoolConfig[(]"protected-mode",.*, *0 *,.*[)],$' ./src/config.cpp; \
-        make -j$(nproc) BUILD_TLS=yes ENABLE_FLASH=yes; \
+        make -j$(nproc) BUILD_TLS=yes; \
        cd src; \
        strip keydb-cli keydb-benchmark keydb-check-rdb keydb-check-aof keydb-diagnostic-tool keydb-sentinel keydb-server; \
        mv keydb-server keydb-cli keydb-benchmark keydb-check-rdb keydb-check-aof keydb-diagnostic-tool keydb-sentinel /usr/local/bin/; \
@@ -54,7 +54,6 @@ RUN set -eux; \
        sed -i 's/^\(logfile .*\)$/# \1/' /etc/keydb/keydb.conf; \
        sed -i 's/protected-mode yes/protected-mode no/g' /etc/keydb/keydb.conf; \
        sed -i 's/^\(bind .*\)$/# \1/' /etc/keydb/keydb.conf; \
-        cd /usr/local/bin; \
        ln -s keydb-cli redis-cli; \
        cd /etc/keydb; \
        ln -s keydb.conf redis.conf; \
@@ -68,10 +67,6 @@ RUN set -eux; \
        echo 'if [ "${1#-}" != "$1" ] || [ "${1%.conf}" != "$1" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
        echo '        set -- keydb-server "$@"' >> /usr/local/bin/docker-entrypoint.sh; \
        echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
-        echo '# if KEYDB_PASSWORD is set, add it to the arguments' >> /usr/local/bin/docker-entrypoint.sh; \
-        echo 'if [ -n "$KEYDB_PASSWORD" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
-        echo '        set -- "$@" --requirepass "${KEYDB_PASSWORD}"' >> /usr/local/bin/docker-entrypoint.sh; \
-        echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
        echo "# allow the container to be started with `--user`" >> /usr/local/bin/docker-entrypoint.sh; \
        echo 'if [ "$1" = "keydb-server" -a "$(id -u)" = "0" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
        echo "        find . \! -user keydb -exec chown keydb '{}' +" >> /usr/local/bin/docker-entrypoint.sh; \
@@ -81,6 +76,6 @@ RUN set -eux; \
        chmod +x /usr/local/bin/docker-entrypoint.sh
 VOLUME /data
 WORKDIR /data
-ENTRYPOINT ["tini", "--", "docker-entrypoint.sh"]
+ENTRYPOINT ["docker-entrypoint.sh"]
 EXPOSE 6379
 CMD ["keydb-server", "/etc/keydb/keydb.conf"]
@@ -44,7 +44,7 @@ public:
        endWriteBatch();
    }

-    virtual std::vector<std::string> getExpirationCandidates(unsigned int count) = 0;
+    virtual std::vector<std::string> getExpirationCandidates(unsigned int count, long long time) = 0;
    virtual std::vector<std::string> getEvictionCandidates(unsigned int count) = 0;
    virtual void setExpire(const char *key, size_t cchKey, long long expire) = 0;
    virtual void removeExpire(const char *key, size_t cchKey, long long expire) = 0;
@@ -358,7 +358,6 @@ endif
 OS := $(shell cat /etc/os-release | grep ID= | head -n 1 | cut -d'=' -f2)
 ifeq ($(OS),alpine)
    FINAL_CXXFLAGS+=-DUNW_LOCAL_ONLY
-	FINAL_CXXFLAGS+=-DALPINE
    FINAL_LIBS += -lunwind
 endif

@@ -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();
@@ -51,7 +51,7 @@ public:
    bool enumerate(IStorage::callback fn) const { return m_spstorage->enumerate(fn); }
    bool enumerate_hashslot(IStorage::callback fn, unsigned int hashslot) const { return m_spstorage->enumerate_hashslot(fn, hashslot); }

-    std::vector<std::string> getExpirationCandidates(unsigned int count) { return m_spstorage->getExpirationCandidates(count); }
+    std::vector<std::string> getExpirationCandidates(unsigned int count, long long time) { return m_spstorage->getExpirationCandidates(count, time); }
    std::vector<std::string> getEvictionCandidates(unsigned int count) { return m_spstorage->getEvictionCandidates(count); }
    void setExpire(const char *key, size_t cchKey, long long expire) { m_spstorage->setExpire(key, cchKey, expire); }
    void removeExpire(const char *key, size_t cchKey, long long expire) { m_spstorage->removeExpire(key, cchKey, expire); }
@@ -113,8 +113,6 @@ enum class AE_ASYNC_OP
    PostCppFunction,
    DeleteFileEvent,
    CreateFileEvent,
-    PostAsynDBFunction,
-
 };

 struct aeCommand
@@ -127,8 +125,6 @@ struct aeCommand
        aePostFunctionProc *proc;
        aeFileProc *fproc;
        std::function<void()> *pfn;
-        aePostFunctionTokenProc* tproc;
-
    };
    void *clientData;
 };
@@ -179,17 +175,6 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
            (*cmd.pfn)();

            delete cmd.pfn;
-            break;
-        }
-        case AE_ASYNC_OP::PostAsynDBFunction:
-        {   //added to support async api IStorage
-           if (cmd.fLock && !ulock.owns_lock()) {
-                g_forkLock.releaseRead();
-                ulock.lock();
-                g_forkLock.acquireRead();
-            }  
-            ((aePostFunctionTokenProc*)cmd.tproc)(eventLoop,(StorageToken*)cmd.clientData);
-            break; 
        }
            break;
        }
@@ -271,20 +256,6 @@ int aePostFunction(aeEventLoop *eventLoop, aePostFunctionProc *proc, void *arg)
    return AE_OK;
 }

-int aePostFunction(aeEventLoop *eventLoop, aePostFunctionTokenProc *proc, StorageToken *token)
-{
-  //added to support async api IStorage
-    aeCommand cmd = {};
-    cmd.op = AE_ASYNC_OP::PostAsynDBFunction; 
-    cmd.tproc = proc;
-    cmd.clientData = (void*)token; 
-    cmd.fLock = true;
-    auto size = write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
-    if (size != sizeof(cmd))
-        return AE_ERR;
-    return AE_OK;
-}
-
 int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fLock, bool fForceQueue)
 {
    if (eventLoop == g_eventLoopThisThread && !fForceQueue)
@@ -890,7 +861,7 @@ void aeReleaseForkLock()

 void aeForkLockInChild()
 {
-    g_forkLock.setMulti(false);
+    g_forkLock.setNotify(false);
 }

 int aeThreadOwnsLock()
@@ -72,15 +72,13 @@ extern "C" {
 #define AE_NOTUSED(V) ((void) V)

 struct aeEventLoop;
+
 /* Types and data structures */
 typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
 typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
 typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientData);
 typedef void aeBeforeSleepProc(struct aeEventLoop *eventLoop);
 typedef void aePostFunctionProc(void *pvArgs);
-//added to support async api IStorage
-struct StorageToken;
-typedef void aePostFunctionTokenProc(struct aeEventLoop *el, struct StorageToken *token);

 /* File event structure */
 typedef struct aeFileEvent {
@@ -136,8 +134,6 @@ int aePostFunction(aeEventLoop *eventLoop, aePostFunctionProc *proc, void *arg);
 #ifdef __cplusplus
 }   // EXTERN C
 int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fLock = true, bool fForceQueue = false);
-//added to support async api IStorage
-int aePostFunction(aeEventLoop *eventLoop, aePostFunctionTokenProc *proc, StorageToken *token);
 extern "C" {
 #endif
 void aeDeleteEventLoop(aeEventLoop *eventLoop);
@@ -5991,9 +5991,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
                /* If it is not the first key, make sure it is exactly
                 * the same key as the first we saw. */
                if (!equalStringObjects(firstkey,thiskey)) {
-                    clusterNode* nThisKey = g_pserver->cluster->slots[thisslot];
-
-                    if ((slot != thisslot) && (nThisKey != n || migrating_slot || importing_slot || g_pserver->cluster->migrating_slots_to[slot] != nullptr || g_pserver->cluster->importing_slots_from[slot] != nullptr)) {
+                    if (slot != thisslot) {
                        /* Error: multiple keys from different slots. */
                        getKeysFreeResult(&result);
                        if (error_code)
@@ -2946,7 +2946,7 @@ standardConfig configs[] = {
    createLongLongConfig("latency-monitor-threshold", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, g_pserver->latency_monitor_threshold, 0, INTEGER_CONFIG, NULL, NULL),
    createLongLongConfig("proto-max-bulk-len", NULL, MODIFIABLE_CONFIG, 1024*1024, LLONG_MAX, g_pserver->proto_max_bulk_len, 512ll*1024*1024, MEMORY_CONFIG, NULL, NULL), /* Bulk request max size */
    createLongLongConfig("stream-node-max-entries", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, g_pserver->stream_node_max_entries, 100, INTEGER_CONFIG, NULL, NULL),
-    createLongLongConfig("repl-backlog-size", NULL, MODIFIABLE_CONFIG, 1, LLONG_MAX, g_pserver->repl_backlog_config_size, 1024*1024, MEMORY_CONFIG, NULL, updateReplBacklogSize), /* Default: 1mb */
+    createLongLongConfig("repl-backlog-size", NULL, MODIFIABLE_CONFIG, 1, LLONG_MAX, g_pserver->repl_backlog_size, 1024*1024, MEMORY_CONFIG, NULL, updateReplBacklogSize), /* Default: 1mb */
    createLongLongConfig("repl-backlog-disk-reserve", NULL, IMMUTABLE_CONFIG, 0, LLONG_MAX, cserver.repl_backlog_disk_size, 0, MEMORY_CONFIG, NULL, NULL),
    createLongLongConfig("max-snapshot-slip", NULL, MODIFIABLE_CONFIG, 0, 5000, g_pserver->snapshot_slip, 400, 0, NULL, NULL),
    createLongLongConfig("max-rand-count", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX/2, g_pserver->rand_total_threshold, LONG_MAX/2, 0, NULL, NULL),
@@ -2982,8 +2982,6 @@ standardConfig configs[] = {
    createSDSConfig("availability-zone", NULL, MODIFIABLE_CONFIG, 0, g_pserver->sdsAvailabilityZone, "", NULL, NULL),
    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. */
@@ -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 */
@@ -2702,6 +2703,11 @@ void redisDbPersistentData::prepOverwriteForSnapshot(char *key)
        auto itr = m_pdbSnapshot->find_cached_threadsafe(key);
        if (itr.key() != nullptr)
        {
+            if (itr.val()->FExpires()) {
+                // Note: I'm sure we could handle this, but its too risky at the moment.
+                //  There are known bugs doing this with expires
+                return;
+            }
            sds keyNew = sdsdupshared(itr.key());
            if (dictAdd(m_pdictTombstone, keyNew, (void*)dictHashKey(m_pdict, key)) != DICT_OK)
                sdsfree(keyNew);
@@ -2739,13 +2745,6 @@ void redisDbPersistentData::clear(void(callback)(void*))
    if (m_spstorage != nullptr)
        m_spstorage->clear(callback);
    dictEmpty(m_pdictTombstone,callback);
-
-    // To avoid issues with async rehash we completly free the old dict and create a fresh one
-    dictRelease(m_pdict);
-    dictRelease(m_pdictTombstone);
-    m_pdict = dictCreate(&dbDictType, this);
-    m_pdictTombstone = dictCreate(&dbTombstoneDictType, this);
-
    m_pdbSnapshot = nullptr;
    m_numexpires = 0;
 }
@@ -2881,10 +2880,10 @@ LNotFound:
            {
                dictAdd(m_pdict, sdsNewKey, o);
                
-                o->SetFExpires(spexpire != nullptr);
                if (spexpire != nullptr) {
                    o->expire = std::move(*spexpire);
                }
+                o->SetFExpires(spexpire != nullptr);
                g_pserver->stat_storage_provider_read_hits++;
            } else {
                sdsfree(sdsNewKey);
@@ -79,7 +79,6 @@ void printCrashReport(void);
 void bugReportEnd(int killViaSignal, int sig);
 void logStackTrace(void *eip, int uplevel);
 void getTempFileName(char tmpfile[], int tmpfileNum);
-bool initializeStorageProvider(const char **err);

 /* ================================= Debugging ============================== */

@@ -309,7 +308,7 @@ void computeDatasetDigest(unsigned char *final) {
        mixDigest(final,&aux,sizeof(aux));

        /* Iterate this DB writing every entry */
-        db->iterate_threadsafe([final](const char *key, robj_roptr o)->bool {
+        db->iterate_threadsafe([final, db](const char *key, robj_roptr o)->bool {
            unsigned char digest[20];
            robj *keyobj;

@@ -536,75 +535,30 @@ NULL
                return;
            }
        }
-        
-        if (g_pserver->m_pstorageFactory) {
-            protectClient(c);

-            for (int idb = 0; idb < cserver.dbnum; ++idb) {
-                if (g_pserver->db[idb]->processChanges(false))
-                    g_pserver->db[idb]->commitChanges();
-                g_pserver->db[idb]->storageProviderDelete();
-            }
-            delete g_pserver->metadataDb;
-
-            if (flush) emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
-
-            delete g_pserver->m_pstorageFactory;
-
-            const char *err;
-            if (!initializeStorageProvider(&err))
-            {
-                serverLog(LL_WARNING, "Failed to initialize storage provider: %s",err);
-                exit(EXIT_FAILURE);
-            }
-
-            g_pserver->metadataDb = g_pserver->m_pstorageFactory->createMetadataDb();
-            for (int idb = 0; idb < cserver.dbnum; ++idb)
-            {
-                int dbid = idb;
-                std::string dbid_key = "db-" + std::to_string(idb);
-                g_pserver->metadataDb->retrieve(dbid_key.c_str(), dbid_key.length(), [&](const char *, size_t, const void *data, size_t){
-                    dbid = *(int*)data;
-                });
-                delete g_pserver->db[idb];
-                g_pserver->db[idb] = new (MALLOC_LOCAL) redisDb();
-                g_pserver->db[idb]->initialize(dbid);
-            }
-
-            moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_FLASH_START, NULL);
-            for (int idb = 0; idb < cserver.dbnum; ++idb)
-            {
-                g_pserver->db[idb]->storageProviderInitialize();
-                g_pserver->db[idb]->trackChanges(false);
-            }
-            moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_ENDED, NULL);
-            
-            unprotectClient(c);
-        } else {
-            /* The default behavior is to save the RDB file before loading
-            * it back. */
-            if (save) {
-                rdbSaveInfo rsi, *rsiptr;
-                rsiptr = rdbPopulateSaveInfo(&rsi);
-                if (rdbSave(nullptr, rsiptr) != C_OK) {
-                    addReply(c,shared.err);
-                    return;
-                }
-            }
-
-            /* The default behavior is to remove the current dataset from
-            * memory before loading the RDB file, however when MERGE is
-            * used together with NOFLUSH, we are able to merge two datasets. */
-            if (flush) emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
-
-            protectClient(c);
-            int ret = rdbLoadFile(g_pserver->rdb_filename,NULL,flags);
-            unprotectClient(c);
-            if (ret != C_OK) {
-                addReplyError(c,"Error trying to load the RDB dump");
+        /* The default behavior is to save the RDB file before loading
+         * it back. */
+        if (save) {
+            rdbSaveInfo rsi, *rsiptr;
+            rsiptr = rdbPopulateSaveInfo(&rsi);
+            if (rdbSave(nullptr, rsiptr) != C_OK) {
+                addReply(c,shared.err);
                return;
            }
        }
+
+        /* The default behavior is to remove the current dataset from
+         * memory before loading the RDB file, however when MERGE is
+         * used together with NOFLUSH, we are able to merge two datasets. */
+        if (flush) emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
+
+        protectClient(c);
+        int ret = rdbLoadFile(g_pserver->rdb_filename,NULL,flags);
+        unprotectClient(c);
+        if (ret != C_OK) {
+            addReplyError(c,"Error trying to load the RDB dump");
+            return;
+        }
        serverLog(LL_WARNING,"DB reloaded by DEBUG RELOAD");
        addReply(c,shared.ok);
    } else if (!strcasecmp(szFromObj(c->argv[1]),"loadaof")) {
@@ -518,13 +518,11 @@ void pexpireMemberAtCommand(client *c)
            } while (sampled == 0 ||
                    (expired*100/sampled) > config_cycle_acceptable_stale);
        } else {
-            long prev_expired;
-            long long now = mstime();
+            long long now;
            size_t tried = 0;
            std::vector<std::string> keys;
            do {
-                prev_expired = total_expired;
-                keys = db->getStorageCache()->getExpirationCandidates(ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP);
+                keys = db->getStorageCache()->getExpirationCandidates(ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP, now);
                for (std::string key : keys) {
                    robj* keyobj = createStringObject(key.c_str(), key.size());
                    db->find(szFromObj(keyobj));
@@ -535,7 +533,7 @@ void pexpireMemberAtCommand(client *c)
                }
                total_sampled += keys.size();
                elapsed = ustime()-start;
-            } while (keys.size() > 0 && (elapsed < timelimit) && (total_expired - prev_expired) > 0);
+            } while (keys.size() > 0 && (elapsed < timelimit));

            if (ustime()-start > timelimit) {
                timelimit_exit = 1;
@@ -160,8 +160,6 @@ public:

    expireEntry &operator=(expireEntry &&e)
    {
-        if (FFat())
-            delete pfatentry();
        s = e.s;
        e.s.m_whenAndPtrUnion = 0;
        e.s.fFat = false;
@@ -169,8 +167,6 @@ public:
    }

    expireEntry &operator=(expireEntry &e) {
-        if (FFat())
-            delete pfatentry();
        if (e.FFat()) {
            s.m_whenAndPtrUnion = reinterpret_cast<long long>(new expireEntryFat(*e.pfatentry()));
            s.fFat = true;
@@ -19,7 +19,6 @@ static size_t getMemKey(std::string key) {
    }
    return 0;
 # else
-    (void)key;
    return 0;
 # endif
 }
@@ -690,9 +690,8 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) {
 }

 /* Free the context after the user function was called. */
-void moduleFreeContext(RedisModuleCtx *ctx, bool propogate) {
-    if (propogate)
-        moduleHandlePropagationAfterCommandCallback(ctx);
+void moduleFreeContext(RedisModuleCtx *ctx) {
+    moduleHandlePropagationAfterCommandCallback(ctx);
    autoMemoryCollect(ctx);
    poolAllocRelease(ctx);
    if (ctx->postponed_arrays) {
@@ -7058,7 +7057,7 @@ RedisModuleString *RM_DictPrev(RedisModuleCtx *ctx, RedisModuleDictIter *di, voi
 /* Compare the element currently pointed by the iterator to the specified
 * element given by key/keylen, according to the operator 'op' (the set of
 * valid operators are the same valid for RedisModule_DictIteratorStart).
- * If the comparison is successful the command returns REDISMODULE_OK
+ * If the comparision is successful the command returns REDISMODULE_OK
 * otherwise REDISMODULE_ERR is returned.
 *
 * This is useful when we want to just emit a lexicographical range, so
@@ -4,7 +4,7 @@ OBJECT_FILES := modmain.o
 MODSNAP_CXX_FLAGS := -std=gnu++14

 %.o: %.cpp
-	$(CXX) -o $@ -c $< $(MODULE_FLAGS) -I../../../deps/cpp-statsd-client/include $(MODSNAP_CXX_FLAGS) -g
+	$(CXX) -o $@ -c $< $(MODULE_FLAGS) -I../../../deps/cpp-statsd-client/include $(MODSNAP_CXX_FLAGS)

 modstatsd.so: $(OBJECT_FILES)
 	$(CXX) -shared $(OBJECT_FILES) -o modstatsd.so
@@ -76,7 +76,7 @@ std::string m_strPrefix { "keydb" };
 const std::regex g_replica_or_db_info_regex { "^(slave|db)(\\d+)" };
 const char *g_string_counter_separator = "__";
 const uint64_t g_stats_buffer_size_bytes = 1600;
-std::string nodeName;
+utsname sysName;
 int unameResult;

 enum class StatsD_Type {
@@ -582,7 +582,7 @@ void event_cron_handler(struct RedisModuleCtx *ctx, RedisModuleEvent eid, uint64
        size_t startTime = ustime();

 #ifdef __linux__
-	/* Log CPU Usage */
+        /* Log CPU Usage */
 	static long long s_mscpuLast = 0;
 	struct rusage self_ru;
 	getrusage(RUSAGE_SELF, &self_ru);
@@ -591,7 +591,7 @@ void event_cron_handler(struct RedisModuleCtx *ctx, RedisModuleEvent eid, uint64
 			+ (self_ru.ru_stime.tv_sec * 1000) + (self_ru.ru_stime.tv_usec / 1000);


-	g_stats->gauge("cpu_load_perc", ((double)(mscpuCur - s_mscpuLast) / ((curTime - lastTime)*1000))*100, false /* prefixOnly */);   
+        g_stats->gauge("cpu_load_perc", (double)(mscpuCur - s_mscpuLast) / ((curTime - lastTime)*1000), true /* prefixOnly */);   
 	s_mscpuLast = mscpuCur;
 #endif

@@ -600,7 +600,7 @@ void event_cron_handler(struct RedisModuleCtx *ctx, RedisModuleEvent eid, uint64

        /* node name */
        if (unameResult == 0) {
-            g_stats->increment("node_name" + std::string(g_string_counter_separator) + nodeName);
+            g_stats->increment("node_name" + std::string(g_string_counter_separator) + sysName.nodename);
        }

        /* Log INFO Fields */
@@ -653,7 +653,7 @@ void event_cron_handler(struct RedisModuleCtx *ctx, RedisModuleEvent eid, uint64
        reply = RedisModule_Call(ctx, "dbsize", "");
        long long keys = RedisModule_CallReplyInteger(reply);
        RedisModule_FreeCallReply(reply);
-        g_stats->gauge("keys", keys, false /* prefixOnly */);
+        g_stats->gauge("keys", keys);
        RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_DEBUG, "Emitting metric \"keys\": %llu", keys);
        g_stats->timing("emit_keys_metric_time_taken_us", ustime() - commandStartTime);

@@ -672,20 +672,17 @@ extern "C" int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv,
    RedisModule_AutoMemory(ctx);
    /* Use pod name if available*/
    const char *podName = getenv("POD_NAME");
-    utsname sysName;
    unameResult = uname(&sysName);
-    if (unameResult == 0) {
-        nodeName = std::string(sysName.nodename);
-        std::replace(nodeName.begin(), nodeName.end(), '.', '-');
-    }
    if (podName != nullptr) {
        m_strPrefix = podName;
-        std::replace(m_strPrefix.begin(), m_strPrefix.end(), '.', '-');
    }
-    else if (unameResult == 0) {
-        m_strPrefix = nodeName;
-        unameResult = 1;
+    else {
+        if (unameResult == 0) {
+            m_strPrefix = std::string(sysName.nodename);
+            unameResult = 1;
+        }
    }
+    std::replace(m_strPrefix.begin(), m_strPrefix.end(), '.', '-');

    for (int iarg = 0; iarg < argc; ++iarg) {
        size_t len = 0;
@@ -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;
@@ -46,7 +46,6 @@ robj *createObject(int type, void *ptr) {
    char *oB = (char*)zcalloc(sizeof(robj)+mvccExtraBytes, MALLOC_SHARED);
    robj *o = reinterpret_cast<robj*>(oB + mvccExtraBytes);
    
-    new (o) redisObject;
    o->type = type;
    o->encoding = OBJ_ENCODING_RAW;
    o->m_ptr = ptr;
@@ -419,7 +418,6 @@ void decrRefCount(robj_roptr o) {
        case OBJ_NESTEDHASH: freeNestedHashObject(o); break;
        default: serverPanic("Unknown object type"); break;
        }
-        o->~redisObject();
        if (g_pserver->fActiveReplica) {
            zfree(reinterpret_cast<redisObjectExtended*>(o.unsafe_robjcast())-1);
        } else {
@@ -2542,7 +2542,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int *error, uint64_t mvcc_ts
         * encoding version in the lower 10 bits of the module ID. */
        void *ptr = mt->rdb_load(&io,moduleid&1023);
        if (io.ctx) {
-            moduleFreeContext(io.ctx, false /* propogate */);
+            moduleFreeContext(io.ctx);
            zfree(io.ctx);
        }

@@ -8,46 +8,38 @@ class readWriteLock {
    int m_readCount = 0;
    int m_writeCount = 0;
    bool m_writeWaiting = false;
-    bool m_multi = true;
+    bool m_notify = true;
 public:
    readWriteLock(const char *name) : m_readLock(name), m_writeLock(name) {}

    void acquireRead() {
-        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (m_multi) {
-            rm.lock();
-            while (m_writeCount > 0 || m_writeWaiting)
-                m_cv.wait(rm);
-        }
+        std::unique_lock<fastlock> rm(m_readLock);
+        while (m_writeCount > 0 || m_writeWaiting)
+            m_cv.wait(rm);
        m_readCount++;
    }
    
    bool tryAcquireRead() {
        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (m_multi) {
-            if (!rm.try_lock())
-                return false;
-            if (m_writeCount > 0 || m_writeWaiting)
-                return false;
-        }
+        if (!rm.try_lock())
+            return false;
+        if (m_writeCount > 0 || m_writeWaiting)
+            return false;
        m_readCount++;
        return true;
    }

    void acquireWrite(bool exclusive = true) {
-        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (m_multi) {
-            rm.lock();
-            m_writeWaiting = true;
-            while (m_readCount > 0)
+        std::unique_lock<fastlock> rm(m_readLock);
+        m_writeWaiting = true;
+        while (m_readCount > 0)
+            m_cv.wait(rm);
+        if (exclusive) {
+            /* Another thread might have the write lock while we have the read lock
+               but won't be able to release it until they can acquire the read lock
+               so release the read lock and try again instead of waiting to avoid deadlock */
+            while(!m_writeLock.try_lock())
                m_cv.wait(rm);
-            if (exclusive) {
-                /* Another thread might have the write lock while we have the internal lock
-                but won't be able to release it until they can acquire the internal lock
-                so release the internal lock and try again instead of waiting to avoid deadlock */
-                while(!m_writeLock.try_lock())
-                    m_cv.wait(rm);
-            }
        }
        m_writeCount++;
        m_writeWaiting = false;
@@ -60,38 +52,32 @@ public:

    bool tryAcquireWrite(bool exclusive = true) {
        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (m_multi) {
-            if (!rm.try_lock())
+        if (!rm.try_lock())
+            return false;
+        if (m_readCount > 0)
+            return false;
+        if (exclusive)
+            if (!m_writeLock.try_lock())
                return false;
-            if (m_readCount > 0)
-                return false;
-            if (exclusive)
-                if (!m_writeLock.try_lock())
-                    return false;
-        }
        m_writeCount++;
        return true;
    }

    void releaseRead() {
-        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (m_multi) {
-            rm.lock();
-            m_cv.notify_all();
-        }
+        std::unique_lock<fastlock> rm(m_readLock);
        m_readCount--;
+        if (m_notify)
+            m_cv.notify_all();
    }

    void releaseWrite(bool exclusive = true) {
-        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
+        std::unique_lock<fastlock> rm(m_readLock);
        serverAssert(m_writeCount > 0);
-        if (m_multi) {
-            rm.lock();
-            if (exclusive)
-                m_writeLock.unlock();
-            m_cv.notify_all();
-        }
+        if (exclusive)
+            m_writeLock.unlock();
        m_writeCount--;
+        if (m_notify)
+            m_cv.notify_all();
    }

    void downgradeWrite(bool exclusive = true) {
@@ -99,8 +85,8 @@ public:
        acquireRead();
    }

-    void setMulti(bool multi) {
-        m_multi = multi;
+    void setNotify(bool notify) {
+        m_notify = notify;
    }

    bool hasReader() {
@@ -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,
@@ -188,8 +188,6 @@ int bg_unlink(const char *filename) {
 /* ---------------------------------- MASTER -------------------------------- */

 bool createDiskBacklog() {
-    if (g_pserver->repl_backlog_disk != nullptr)
-        return true;    // already exists
    // Lets create some disk backed pages and add them here
    std::string path = "./repl-backlog-temp" + std::to_string(gettid());
 #if (defined __APPLE__ || defined __FreeBSD__)
@@ -1019,6 +1017,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 +1091,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 +1103,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 +1147,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 +1177,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 +1193,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 +1209,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 +1228,7 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
                        lastLogTime = ustime();
                    }
                }
-                cbData += sdslen(strKey) + sdslen(strT);
-                sdsfree(strT);
+                cbData += cchKey + cchVal;
                return replBuf.isActive();
            });

@@ -1303,7 +1300,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 +1479,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 +1670,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 +1736,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 +2513,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 +2523,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 +2599,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 +2999,7 @@ void readSyncBulkPayload(connection *conn) {
        return;
    }

-    if (mi->isKeydbFastsync) {
+    if (mi->isRocksdbSnapshotRepl) {
        if (!readSnapshotBulkPayload(conn, mi, rsi))
            return;
    } else {
@@ -3026,15 +3012,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 +3061,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 +3393,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 +3548,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,12 +3733,12 @@ retry_connect:
    }

    /* Prepare a suitable temp file for bulk transfer */
-    if (!useDisklessLoad() && !mi->isKeydbFastsync) {
+    if (!useDisklessLoad()) {
        while(maxtries--) {
            auto dt = std::chrono::system_clock::now().time_since_epoch();
            auto dtMillisecond = std::chrono::duration_cast<std::chrono::milliseconds>(dt);
            snprintf(tmpfile,sizeof(tmpfile),
-                "temp-%lld.%ld.rdb",(long long)dtMillisecond.count(),(long int)getpid());
+                "temp-%d.%ld.rdb",(int)dtMillisecond.count(),(long int)getpid());
            dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
            if (dfd != -1) break;
            sleep(1);
@@ -3874,10 +3862,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 +4327,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);
@@ -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;
    }

@@ -2137,7 +2137,7 @@ void databasesCron(bool fMainThread) {
                ::dict *dict = g_pserver->db[rehash_db]->dictUnsafeKeyOnly();
                /* Are we async rehashing? And if so is it time to re-calibrate? */
                /* The recalibration limit is a prime number to ensure balancing across threads */
-                if (g_pserver->enable_async_rehash && rehashes_per_ms > 0 && async_rehashes < 131 && !cserver.active_defrag_enabled && cserver.cthreads > 1 && dictSize(dict) > 2048 && dictIsRehashing(dict) && !g_pserver->loading && aeLockContention() > 1) {
+                if (rehashes_per_ms > 0 && async_rehashes < 131 && !cserver.active_defrag_enabled && cserver.cthreads > 1 && dictSize(dict) > 2048 && dictIsRehashing(dict) && !g_pserver->loading) {
                    serverTL->rehashCtl = dictRehashAsyncStart(dict, rehashes_per_ms * ((1000 / g_pserver->hz) / 10));  // Estimate 10% CPU time spent in lock contention
                    if (serverTL->rehashCtl)
                        ++async_rehashes;
@@ -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)
@@ -5663,8 +5663,7 @@ sds genRedisInfoString(const char *section) {
            "lru_clock:%u\r\n"
            "executable:%s\r\n"
            "config_file:%s\r\n"
-            "availability_zone:%s\r\n"
-            "features:%s\r\n",
+            "availability_zone:%s\r\n",
            KEYDB_SET_VERSION,
            redisGitSHA1(),
            strtol(redisGitDirty(),NULL,10) > 0,
@@ -5691,8 +5690,7 @@ sds genRedisInfoString(const char *section) {
            lruclock,
            cserver.executable ? cserver.executable : "",
            cserver.configfile ? cserver.configfile : "",
-            g_pserver->sdsAvailabilityZone,
-            "cluster_mget");
+            g_pserver->sdsAvailabilityZone);
    }

    /* Clients */
@@ -7413,7 +7411,7 @@ static void validateConfiguration()
        exit(EXIT_FAILURE);
    }

-    g_pserver->repl_backlog_size = g_pserver->repl_backlog_config_size; // this is normally set in the update logic, but not on initial config
+    g_pserver->repl_backlog_config_size = g_pserver->repl_backlog_size; // this is normally set in the update logic, but not on initial config
 }

 int iAmMaster(void) {
@@ -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
@@ -966,7 +966,6 @@ struct redisObjectExtended {

 typedef struct redisObject {
    friend redisObject *createEmbeddedStringObject(const char *ptr, size_t len);
-    friend redisObject *createObject(int type, void *ptr);
 protected:
    redisObject() {}

@@ -1610,6 +1609,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 +1870,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. */
@@ -2305,7 +2305,6 @@ struct redisServer {
    rax *errors;                /* Errors table */
    int activerehashing;        /* Incremental rehash in serverCron() */
    int active_defrag_running;  /* Active defragmentation running (holds current scan aggressiveness) */
-    int enable_async_rehash = 1;    /* Should we use the async rehash feature? */
    int cronloops;              /* Number of times the cron function run */
    char runid[CONFIG_RUN_ID_SIZE+1];  /* ID always different at every exec. */
    int sentinel_mode;          /* True if this instance is a Sentinel. */
@@ -2687,7 +2686,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
@@ -2919,7 +2917,7 @@ int moduleGetCommandKeysViaAPI(struct redisCommand *cmd, robj **argv, int argc,
 moduleType *moduleTypeLookupModuleByID(uint64_t id);
 void moduleTypeNameByID(char *name, uint64_t moduleid);
 const char *moduleTypeModuleName(moduleType *mt);
-void moduleFreeContext(struct RedisModuleCtx *ctx, bool propogate = true);
+void moduleFreeContext(struct RedisModuleCtx *ctx);
 void unblockClientFromModule(client *c);
 void moduleHandleBlockedClients(int iel);
 void moduleBlockedClientTimedOut(client *c);
@@ -3182,7 +3180,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)

@@ -3924,7 +3921,7 @@ void updateActiveReplicaMastersFromRsi(rdbSaveInfo *rsi);
 uint64_t getMvccTstamp();
 void incrementMvccTstamp();

-#if __GNUC__ >= 7 && !defined(NO_DEPRECATE_FREE) && !defined(ALPINE)
+#if __GNUC__ >= 7 && !defined(NO_DEPRECATE_FREE)
 [[deprecated]]
 void *calloc(size_t count, size_t size) noexcept;
 [[deprecated]]
@@ -3972,10 +3969,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)
@@ -587,12 +587,8 @@ bool redisDbPersistentDataSnapshot::iterate_threadsafe_core(std::function<bool(c
                if (!fKeyOnly)
                {
                    size_t offset = 0;
-                    std::unique_ptr<expireEntry> spexpire = deserializeExpire((const char*)data, cbData, &offset);
+                    deserializeExpire((const char*)data, cbData, &offset);
                    o = deserializeStoredObject(reinterpret_cast<const char*>(data)+offset, cbData-offset);
-                    o->SetFExpires(spexpire != nullptr);
-                    if (spexpire != nullptr) {
-                        o->expire = std::move(*spexpire);
-                    }
                }
                fContinue = fn(sdsKey, o);
                if (o != nullptr)
@@ -7,8 +7,6 @@
 #include "../cluster.h"
 #include "rocksdbfactor_internal.h"

-template class std::basic_string<char>;
-
 static const char keyprefix[] = INTERNAL_KEY_PREFIX;

 rocksdb::Options DefaultRocksDBOptions();
@@ -26,9 +24,8 @@ bool FInternalKey(const char *key, size_t cch)

 std::string getPrefix(unsigned int hashslot)
 {
-    HASHSLOT_PREFIX_TYPE slot = HASHSLOT_PREFIX_ENDIAN((HASHSLOT_PREFIX_TYPE)hashslot);
-    char *hash_char = (char *)&slot;
-    return std::string(hash_char, HASHSLOT_PREFIX_BYTES);
+    char *hash_char = (char *)&hashslot;
+    return std::string(hash_char + (sizeof(unsigned int) - 2), 2);
 }

 std::string prefixKey(const char *key, size_t cchKey)
@@ -147,28 +144,16 @@ void RocksDBStorageProvider::retrieve(const char *key, size_t cchKey, callbackSi
 size_t RocksDBStorageProvider::clear()
 {
    size_t celem = count();
-    auto options = m_spdb->GetOptions(m_spcolfamily.get());
    auto status = m_spdb->DropColumnFamily(m_spcolfamily.get());
    auto strName = m_spcolfamily->GetName();

    rocksdb::ColumnFamilyHandle *handle = nullptr;
-    rocksdb::ColumnFamilyOptions cf_options(options);
-
+    rocksdb::ColumnFamilyOptions cf_options(m_pfactory->RocksDbOptions());
    m_spdb->CreateColumnFamily(cf_options, strName, &handle);
    m_spcolfamily = std::shared_ptr<rocksdb::ColumnFamilyHandle>(handle);

    if (!status.ok())
        throw status.ToString();
-   
-    status = m_spdb->DropColumnFamily(m_spexpirecolfamily.get());
-    strName = m_spexpirecolfamily->GetName();
-
-    m_spdb->CreateColumnFamily(cf_options, strName, &handle);
-    m_spexpirecolfamily = std::shared_ptr<rocksdb::ColumnFamilyHandle>(handle);
-
-    if (!status.ok())
-        throw status.ToString();
-
    m_count = 0;
    return celem;
 }
@@ -187,7 +172,7 @@ bool RocksDBStorageProvider::enumerate(callback fn) const
        if (FInternalKey(it->key().data(), it->key().size()))
            continue;
        ++count;
-        bool fContinue = fn(it->key().data()+HASHSLOT_PREFIX_BYTES, it->key().size()-HASHSLOT_PREFIX_BYTES, it->value().data(), it->value().size());
+        bool fContinue = fn(it->key().data()+2, it->key().size()-2, it->value().data(), it->value().size());
        if (!fContinue)
            break;
    }
@@ -206,17 +191,17 @@ bool RocksDBStorageProvider::enumerate_hashslot(callback fn, unsigned int hashsl
    std::string prefix = getPrefix(hashslot);
    std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spcolfamily.get()));
    size_t count = 0;
-    for (it->Seek(prefix); it->Valid(); it->Next()) {
+    for (it->Seek(prefix.c_str()); it->Valid(); it->Next()) {
        if (FInternalKey(it->key().data(), it->key().size()))
            continue;
-        if (HASHSLOT_PREFIX_RECOVER(*(HASHSLOT_PREFIX_TYPE *)it->key().data()) != hashslot)
+        if (strncmp(it->key().data(),prefix.c_str(),2) != 0)
            break;
        ++count;
-        bool fContinue = fn(it->key().data()+HASHSLOT_PREFIX_BYTES, it->key().size()-HASHSLOT_PREFIX_BYTES, it->value().data(), it->value().size());
+        bool fContinue = fn(it->key().data()+2, it->key().size()-2, it->value().data(), it->value().size());
        if (!fContinue)
            break;
    }
-    bool full_iter = !it->Valid() || (HASHSLOT_PREFIX_RECOVER(*(HASHSLOT_PREFIX_TYPE *)it->key().data()) != hashslot);
+    bool full_iter = !it->Valid() || (strncmp(it->key().data(),prefix.c_str(),2) != 0);
    if (full_iter && count != g_pserver->cluster->slots_keys_count[hashslot])
    {
        printf("WARNING: rocksdb hashslot count mismatch");
@@ -230,8 +215,7 @@ void RocksDBStorageProvider::setExpire(const char *key, size_t cchKey, long long
 {
    rocksdb::Status status;
    std::unique_lock<fastlock> l(m_lock);
-    long long beExpire = htobe64(expire);
-    std::string prefix((const char *)&beExpire,sizeof(long long));
+    std::string prefix((const char *)&expire,sizeof(long long));
    std::string strKey(key, cchKey);
    if (m_spbatch != nullptr)
        status = m_spbatch->Put(m_spexpirecolfamily.get(), rocksdb::Slice(prefix + strKey), rocksdb::Slice(strKey));
@@ -245,8 +229,7 @@ void RocksDBStorageProvider::removeExpire(const char *key, size_t cchKey, long l
 {
    rocksdb::Status status;
    std::unique_lock<fastlock> l(m_lock);
-    long long beExpire = htobe64(expire);
-    std::string prefix((const char *)&beExpire,sizeof(long long));
+    std::string prefix((const char *)&expire,sizeof(long long));
    std::string strKey(key, cchKey);
    std::string fullKey = prefix + strKey;
    if (!FExpireExists(fullKey))
@@ -259,14 +242,21 @@ void RocksDBStorageProvider::removeExpire(const char *key, size_t cchKey, long l
        throw status.ToString();
 }

-std::vector<std::string> RocksDBStorageProvider::getExpirationCandidates(unsigned int count)
+std::vector<std::string> RocksDBStorageProvider::getExpirationCandidates(unsigned int count, long long time)
 {
    std::vector<std::string> result;
    std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spexpirecolfamily.get()));
-    for (it->SeekToFirst(); it->Valid() && result.size() < count; it->Next()) {
+    for (it->SeekToFirst(); it->Valid() && (*((long long *)it->key().data()) < time) && (result.size() < count); it->Next()) {
        if (FInternalKey(it->key().data(), it->key().size()))
            continue;
        result.emplace_back(it->value().data(), it->value().size());
+        rocksdb::Status status;
+        if (m_spbatch)
+            status = m_spbatch->Delete(m_spexpirecolfamily.get(), it->key());
+        else
+            status = m_spdb->Delete(WriteOptions(), m_spexpirecolfamily.get(), it->key());
+        if (!status.ok())
+            throw status.ToString();
    }
    return result;
 }
@@ -283,11 +273,11 @@ std::vector<std::string> RocksDBStorageProvider::getEvictionCandidates(unsigned
        for (it->Seek(randomHashSlot()); it->Valid() && result.size() < count; it->Next()) {
            if (FInternalKey(it->key().data(), it->key().size()))
                continue;
-            result.emplace_back(it->key().data() + HASHSLOT_PREFIX_BYTES, it->key().size() - HASHSLOT_PREFIX_BYTES);
+            result.emplace_back(it->key().data() + 2, it->key().size() - 2);
        }
    } else {
        std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spexpirecolfamily.get()));
-        for (it->SeekToFirst(); it->Valid() && result.size() < count; it->Next()) {
+        for (it->SeekToFirst(); it->Valid() && result.size() < 16; it->Next()) {
            if (FInternalKey(it->key().data(), it->key().size()))
                continue;
            result.emplace_back(it->value().data(), it->value().size());
@@ -4,21 +4,9 @@
 #include "../IStorage.h"
 #include <rocksdb/db.h>
 #include <rocksdb/utilities/write_batch_with_index.h>
-#ifdef __APPLE__
-#include <libkern/OSByteOrder.h>
-#define htole16 OSSwapHostToLittleInt16
-#define le16toh OSSwapLittleToHostInt16
-#define htobe64 OSSwapHostToBigInt64
-#else
-#include <endian.h>
-#endif
 #include "../fastlock.h"

 #define INTERNAL_KEY_PREFIX "\x00\x04\x03\x00\x05\x02\x04"
-#define HASHSLOT_PREFIX_TYPE uint16_t
-#define HASHSLOT_PREFIX_BYTES sizeof(HASHSLOT_PREFIX_TYPE)
-#define HASHSLOT_PREFIX_ENDIAN htole16
-#define HASHSLOT_PREFIX_RECOVER le16toh
 static const char count_key[] = INTERNAL_KEY_PREFIX "__keydb__count\1";
 static const char version_key[] = INTERNAL_KEY_PREFIX "__keydb__version\1";
 static const char meta_key[] = INTERNAL_KEY_PREFIX "__keydb__metadata\1";
@@ -48,7 +36,7 @@ public:
    virtual bool enumerate(callback fn) const override;
    virtual bool enumerate_hashslot(callback fn, unsigned int hashslot) const override;

-    virtual std::vector<std::string> getExpirationCandidates(unsigned int count) override;
+    virtual std::vector<std::string> getExpirationCandidates(unsigned int count, long long time) override;
    virtual std::vector<std::string> getEvictionCandidates(unsigned int count) override;
    virtual void setExpire(const char *key, size_t cchKey, long long expire) override;
    virtual void removeExpire(const char *key, size_t cchKey, long long expire) override;
@@ -72,7 +72,7 @@ RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum, cons
    rocksdb::DB *db = nullptr;

    auto options = RocksDbOptions();
-    options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(HASHSLOT_PREFIX_BYTES));
+    options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(2));

    for (int idb = 0; idb < dbnum; ++idb)
    {
@@ -196,7 +196,7 @@ IStorage *RocksDBStorageFactory::create(int db, key_load_iterator iter, void *pr
                printf("\tDatabase %d was not shutdown cleanly, recomputing metrics\n", db);
            fFirstRealKey = false;
            if (iter != nullptr)
-                iter(it->key().data()+HASHSLOT_PREFIX_BYTES, it->key().size()-HASHSLOT_PREFIX_BYTES, privdata);
+                iter(it->key().data()+2, it->key().size()-2, privdata);
            ++count;
        }
    }
@@ -29,7 +29,7 @@ public:
    virtual bool enumerate_hashslot(callback fn, unsigned int hashslot) const override;
    virtual size_t count() const override;

-    virtual std::vector<std::string> getExpirationCandidates(unsigned int) override { return std::vector<std::string>(); }
+    virtual std::vector<std::string> getExpirationCandidates(unsigned int, long long) override { return std::vector<std::string>(); }
    virtual std::vector<std::string> getEvictionCandidates(unsigned int) override { return std::vector<std::string>(); }
    virtual void setExpire(const char *, size_t, long long) override {}
    virtual void removeExpire(const char *, size_t, long long) override {}
@@ -2036,7 +2036,7 @@ struct zsetopsrc {
                zset *zs;
                zskiplistNode *node;
            } sl;
-        } iterzset;
+        } zset;
    } iter;
 };

@@ -2085,7 +2085,7 @@ void zuiInitIterator(zsetopsrc *op) {
        /* Sorted sets are traversed in reverse order to optimize for
         * the insertion of the elements in a new list as in
         * ZDIFF/ZINTER/ZUNION */
-        iterzset *it = (iterzset*)&op->iter.iterzset;
+        iterzset *it = (iterzset*)&op->iter.zset;
        if (op->encoding == OBJ_ENCODING_ZIPLIST) {
            it->zl.zl = (unsigned char*)op->subject->m_ptr;
            it->zl.eptr = ziplistIndex(it->zl.zl,-2);
@@ -2118,7 +2118,7 @@ void zuiClearIterator(zsetopsrc *op) {
            serverPanic("Unknown set encoding");
        }
    } else if (op->type == OBJ_ZSET) {
-        iterzset *it = &op->iter.iterzset;
+        iterzset *it = &op->iter.zset;
        if (op->encoding == OBJ_ENCODING_ZIPLIST) {
            UNUSED(it); /* skip */
        } else if (op->encoding == OBJ_ENCODING_SKIPLIST) {
@@ -2194,7 +2194,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) {
            serverPanic("Unknown set encoding");
        }
    } else if (op->type == OBJ_ZSET) {
-        iterzset *it = &op->iter.iterzset;
+        iterzset *it = &op->iter.zset;
        if (op->encoding == OBJ_ENCODING_ZIPLIST) {
            /* No need to check both, but better be explicit. */
            if (it->zl.eptr == NULL || it->zl.sptr == NULL)
@@ -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
@@ -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]
@@ -81,6 +81,21 @@ if {$::flash_enabled} {
            }
        }

+        test { EXPIRE reduces used disk space } {
+            r flushall
+            r set testkey foo ex 10
+            r debug flush-storage
+            set used [r debug get-storage-usage]
+            assert_equal {1} [r dbsize]
+            wait_for_condition 50 1000 {
+                [r dbsize] == 0
+            } else {
+                fail "key is not expired"
+            }
+            set after [r debug get-storage-usage]
+            assert {$after < $used}
+        }
+
        test { SUBKEY EXPIRE persists after cache flush } {
            r flushall
            r sadd testkey foo bar baz
@@ -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]
@@ -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
Author	SHA1	Message	Date
Malavan Sotheeswaran	717805e191	init iteration	2023-08-22 22:48:26 -07:00
Malavan Sotheeswaran	8f8b61f211	compile fix	2023-08-22 13:18:14 -07:00
Malavan Sotheeswaran	fca3c77823	somehow missed reference to m_setexpire	2023-08-22 13:03:03 -07:00
Malavan Sotheeswaran	22a0ad9a23	Merge remote-tracking branch 'internal/main' into flash_expiration	2023-08-22 12:48:28 -07:00
Malavan Sotheeswaran	747eb8067e	disk space test	2023-06-29 21:50:05 -07:00
Malavan Sotheeswaran	1d547f89e1	1 second between each check	2023-06-28 17:44:36 -07:00
Malavan Sotheeswaran	9281b728d5	use dedicated db for load test to avoid loading other keys	2023-06-28 15:50:24 -07:00
Malavan Sotheeswaran	04cc2c01c6	change expire test to wait_for_condition	2023-06-28 00:27:51 -07:00
Malavan Sotheeswaran	f681ca1a9f	Merge remote-tracking branch 'internal/main' into flash_expiration	2023-06-27 15:25:47 -07:00
Malavan Sotheeswaran	336202eba5	strict <	2023-06-27 11:43:13 -07:00
Malavan Sotheeswaran	597dd12498	flash eviction test working	2023-06-27 00:06:32 -07:00
Malavan Sotheeswaran	7f86f4f876	WIP flash eviction tests	2023-06-26 22:06:38 -07:00
Malavan Sotheeswaran	ea7838eb46	fix test	2023-06-26 21:35:09 -07:00
Malavan Sotheeswaran	eb2018a423	reject versions before 6.3.4	2023-06-23 20:33:45 -07:00
Malavan Sotheeswaran	0eb118731e	fix subkey exprires	2023-06-23 18:31:19 -07:00
Malavan Sotheeswaran	a0a264a11c	figured out the bug!	2023-06-23 18:20:36 -07:00
Malavan Sotheeswaran	79b3490f49	fix warnings	2023-06-23 16:52:46 -07:00
Malavan Sotheeswaran	178df4ddc6	respect sample count	2023-06-23 16:32:50 -07:00
Malavan Sotheeswaran	6504e7965a	bake expire modifications into storage cache insert and erase	2023-06-23 12:21:47 -07:00
Malavan Sotheeswaran	7a9d4bbc31	some refactoring to avoid duplicate code	2023-06-23 12:03:08 -07:00
Malavan Sotheeswaran	0c0f8581f5	brackets to avoid ambiguos else	2023-06-21 19:15:11 -07:00
Malavan Sotheeswaran	967bd78658	only delete if exists	2023-06-21 19:12:38 -07:00
Malavan Sotheeswaran	605f369d65	remove expires from storage provider when removed from memory	2023-06-21 18:34:49 -07:00
Malavan Sotheeswaran	c6871a3179	run tests with flash	2023-06-21 16:36:47 -07:00
Malavan Sotheeswaran	f0d9466561	missing nullptr check	2023-06-14 15:11:27 -07:00
Malavan Sotheeswaran	ec3b1eb5c4	handle all keys case	2023-06-14 15:07:50 -07:00
Malavan Sotheeswaran	e248fec27f	for all keys we need to remove the prefix	2023-06-14 14:20:17 -07:00
Malavan Sotheeswaran	eea8932aa7	unit test doesn't make sense	2023-06-13 23:34:59 -07:00
Malavan Sotheeswaran	98eaebd4c8	accurate config update message	2023-06-13 22:52:26 -07:00
Malavan Sotheeswaran	8e41dc3ed2	fix spacing	2023-06-13 19:19:48 -07:00
Malavan Sotheeswaran	e1afce8a4a	remove extra blank line	2023-06-13 19:19:04 -07:00
Malavan Sotheeswaran	63dedf054f	add unit test for flash eviction	2023-06-13 19:12:26 -07:00
Malavan Sotheeswaran	5f1884fc36	include vector	2023-06-12 22:55:36 -07:00
Malavan Sotheeswaran	e54fd922d9	update expire column during fast sync	2023-06-12 21:51:40 -07:00
Malavan Sotheeswaran	b81ba8bfc9	Merge branch 'main' into flash_expiration	2023-06-12 15:09:37 -07:00
Malavan Sotheeswaran	f737084170	mstime not ustime	2023-05-26 03:05:29 -07:00
Malavan Sotheeswaran	49f48d808d	double check expire time	2023-05-26 03:00:15 -07:00
Malavan Sotheeswaran	71045949c2	fix include and crash	2023-05-26 02:42:19 -07:00
Malavan Sotheeswaran	ef1483e1d4	fix crash	2023-05-25 14:01:19 -07:00
Malavan Sotheeswaran	708b8d941b	actually add expires to rocksdb	2023-05-25 00:17:12 -07:00
Malavan Sotheeswaran	12fd46da0b	add ability to set expires in storge provider	2023-05-24 21:57:12 -07:00
Malavan Sotheeswaran	5aff96c577	fixes	2023-05-24 15:07:15 -07:00
Malavan Sotheeswaran	5cb2ead650	refactor for time limited expire cycle	2023-05-24 14:55:41 -07:00
Malavan Sotheeswaran	6853b15d50	add random hashslot impl	2023-05-23 21:26:01 -07:00
Malavan Sotheeswaran	f701161d0a	more progress	2023-05-23 21:10:40 -07:00
Malavan Sotheeswaran	6a32023cfc	use correct key variable	2023-05-23 21:05:23 -07:00
Malavan Sotheeswaran	6eb595d6b4	remove expire from memory when evicting from cache	2023-05-23 20:50:58 -07:00
Malavan Sotheeswaran	edae100767	refactor towards new design	2023-05-23 20:47:15 -07:00
Malavan Sotheeswaran	90fc43b849	add get expire and get eviction candidate impls for rocksdb	2023-05-18 10:19:49 -07:00
Malavan Sotheeswaran	d4857720f2	adding KeyDB side of flash expiration/eviction	2023-05-17 13:46:43 -07:00
Malavan Sotheeswaran	cdfca6a42b	add machamp script	2023-05-11 21:20:26 -07:00
Malavan Sotheeswaran	bae5c0dad1	switch machamp to main branch	2023-05-11 21:06:31 -07:00
Malavan Sotheeswaran	8b5fced206	add machamp	2023-05-11 21:05:59 -07:00
Malavan Sotheeswaran	e6c0e14cdd	add fixed length prefix extractor	2023-05-11 12:46:47 -07:00
Malavan Sotheeswaran	e47624dc8e	add option to show logs of 1 server at a time	2023-05-10 22:21:27 -07:00
Malavan Sotheeswaran	76e103b9bf	fix warnings	2023-05-10 15:25:39 -07:00
Malavan Sotheeswaran	ac1bd8eec4	add assert to ensure correctness	2023-05-10 14:19:17 -07:00
Malavan Sotheeswaran	9bfb89bfd8	forgot ;	2023-05-10 14:16:18 -07:00
Malavan Sotheeswaran	eb565ae8f7	check for count mismatch	2023-05-10 14:12:19 -07:00
Malavan Sotheeswaran	bc37934f6f	forgot return statement in getKeysInSlot	2023-05-09 21:10:35 -07:00
Malavan Sotheeswaran	e0007d9687	add --flash option for tests	2023-05-09 21:07:54 -07:00
Malavan Sotheeswaran	e52792fe2a	add hashslot enum to test storage provider	2023-05-09 13:52:37 -07:00
Malavan Sotheeswaran	880ca46788	fix compile	2023-05-09 13:49:19 -07:00
Malavan Sotheeswaran	6e1d6146e0	implement slot to keys with rocksdb	2023-05-09 12:36:48 -07:00
Malavan Sotheeswaran	ce248d7b66	don't prefix internal keys	2023-05-09 11:30:08 -07:00
Malavan Sotheeswaran	4bdd2ed5f2	remove hashslot prefix when loading from rocksdb	2023-05-08 21:53:56 -07:00
Malavan Sotheeswaran	e7ba8452bd	Merge remote-tracking branch 'origin/main' into flash_slot_to_key	2023-05-08 13:03:48 -07:00
Malavan Sotheeswaran	b9f049a25d	don't need to offset internal keys	2023-05-05 15:45:09 -07:00
Malavan Sotheeswaran	e69bfd5da6	add hash prefixing to rocksdb	2023-05-05 12:22:29 -07:00