[BUG] Moves key to another DB, the source key was removed if the move failed due to the key exists in the destination db #497 (#498)

Co-authored-by: Paul Chen <mingchen@Mings-MacBook-Pro.local>

pkg/deb/deb-buildsource.sh

@@ -23,8 +23,8 @@ pkg_name=keydb-$majorv:$version$distname
 
 # create build tree
 cd ../../../
-tar -czvf keydb_$version.orig.tar.gz --force-local keydb-internal
-cd keydb-internal/pkg/deb/
+tar -czvf keydb_$version.orig.tar.gz --force-local KeyDB
+cd KeyDB/pkg/deb/
 mkdir -p $pkg_name/tmp
 if [[ "$codename" == "xenial" ]] || [[ "$codename" == "stretch" ]]; then
 	cp -r debian_dh9 $pkg_name/tmp/debian

pkg/deb/debian/control

@@ -26,8 +26,8 @@ Build-Depends:
  libzstd-dev <!nocheck>
 Standards-Version: 4.2.1
 Homepage: https://docs.keydb.dev/
-Vcs-Git: https://github.com/EQ-Alpha/KeyDB.git
-Vcs-Browser: https://github.com/EQ-Alpha/KeyDB
+Vcs-Git: https://github.com/Snapchat/KeyDB.git
+Vcs-Browser: https://github.com/Snapchat/KeyDB
 
 Package: keydb
 Architecture: all

pkg/deb/debian_dh9/control

@@ -25,8 +25,8 @@ Build-Depends:
  libzstd-dev <!nocheck>
 Standards-Version: 4.2.1
 Homepage: https://docs.keydb.dev/
-Vcs-Git: https://github.com/EQ-Alpha/KeyDB.git
-Vcs-Browser: https://github.com/EQ-Alpha/KeyDB
+Vcs-Git: https://github.com/Snapchat/KeyDB.git
+Vcs-Browser: https://github.com/Snapchat/KeyDB
 
 Package: keydb
 Architecture: all

pkg/deb/master_changelog

@@ -1,3 +1,11 @@
+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
+  * Issues fixed: #419, #422, #428
+  * PRs: #426, #429, #431, #433
+
+-- Ben Schermel <ben@eqalpha.com>  Mon, 23 May 2022 20:00:37 +0000
+
 keydb (6:6.3.0-1distribution_placeholder) codename_placeholder; urgency=medium
 
   * This release open sources KeyDB Enterprise features into the open source project along with PSYNC for active replication

pkg/docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM ubuntu:20.04
+FROM ubuntu:18.04
 SHELL ["/bin/bash","-c"]
 RUN groupadd -r keydb && useradd -r -g keydb keydb
 # use gosu for easy step-down from root: https://github.com/tianon/gosu/releases
@@ -77,7 +77,7 @@ RUN \
         mkdir /data && chown keydb:keydb /data; \
         mkdir /flash && chown keydb:keydb /flash; \
         mkdir -p /etc/keydb; \
-        cp /tmp/keydb-internal/keydb.conf /etc/keydb/; \
+        cp /tmp/KeyDB/keydb.conf /etc/keydb/; \
         sed -i 's/^\(daemonize .*\)$/# \1/' /etc/keydb/keydb.conf; \
         sed -i 's/^\(dir .*\)$/# \1\ndir \/data/' /etc/keydb/keydb.conf; \
         sed -i 's/^\(logfile .*\)$/# \1/' /etc/keydb/keydb.conf; \
@@ -110,8 +110,3 @@ ENV KEYDB_PRO_DIRECTORY=/usr/local/bin/
 ENTRYPOINT ["docker-entrypoint.sh"]
 EXPOSE 6379
 CMD ["keydb-server","/etc/keydb/keydb.conf"]
-
-
-
-
-

pkg/docker/Dockerfile_Alpine

@@ -0,0 +1,77 @@
+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; \
+        apk add --no-cache --virtual .build-deps \
+                coreutils \
+                gcc \
+                linux-headers \
+                make \
+                musl-dev \
+                openssl-dev \
+                git \
+                util-linux-dev \
+                curl-dev \
+                g++ \
+                libunwind-dev \
+                bash \
+                perl \
+                git \
+        ; \
+        cd /tmp && git clone --branch $BRANCH https://github.com/Snapchat/KeyDB.git --recursive; \
+        cd /tmp/KeyDB; \
+        # disable protected mode as it relates to docker
+        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; \
+        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/; \
+        runDeps="$( \
+                scanelf --needed --nobanner --format '%n#p' --recursive /usr/local \
+                        | tr ',' '\n' \
+                        | sort -u \
+                        | awk 'system("[ -e /usr/local/lib/" $1 " ]") == 0 { next } { print "so:" $1 }' \
+        )"; \
+        apk add --no-network --virtual .keydb-rundeps $runDeps; \
+        apk del --no-network .build-deps; \
+        # create working directories and organize files
+        mkdir /data && chown keydb:keydb /data; \
+        mkdir /flash && chown keydb:keydb /flash; \
+        mkdir -p /etc/keydb; \
+        cp /tmp/KeyDB/keydb.conf /etc/keydb/; \
+        sed -i 's/^\(daemonize .*\)$/# \1/' /etc/keydb/keydb.conf; \
+        sed -i 's/^\(dir .*\)$/# \1\ndir \/data/' /etc/keydb/keydb.conf; \
+        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; \
+        ln -s keydb-cli redis-cli; \
+        cd /etc/keydb; \
+        ln -s keydb.conf redis.conf; \
+        rm -rf /tmp/*
+# generate entrypoint script
+RUN set -eux; \
+        echo '#!/bin/sh' > /usr/local/bin/docker-entrypoint.sh; \
+        echo 'set -e' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo "# first arg is '-f' or '--some-option'" >> /usr/local/bin/docker-entrypoint.sh; \
+        echo "# or first arg is `something.conf`" >> /usr/local/bin/docker-entrypoint.sh; \
+        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 "# 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; \
+        echo '        exec su-exec keydb "$0" "$@"' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo 'exec "$@"' >> /usr/local/bin/docker-entrypoint.sh; \
+        chmod +x /usr/local/bin/docker-entrypoint.sh
+VOLUME /data
+WORKDIR /data
+ENTRYPOINT ["docker-entrypoint.sh"]
+EXPOSE 6379
+CMD ["keydb-server", "/etc/keydb/keydb.conf"]

src/StorageCache.cpp

@@ -203,6 +203,7 @@ void StorageCache::beginWriteBatch() {
 }
 
 void StorageCache::emergencyFreeCache() {
+    std::unique_lock<fastlock> ul(m_lock);
     dict *d = m_pdict;
     m_pdict = nullptr;
     if (d != nullptr) {

src/db.cpp

@@ -1568,6 +1568,12 @@ void moveCommand(client *c) {
         return;
     }
 
+    /* Return zero if the key already exists in the target DB */
+    if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
+        addReply(c,shared.czero);
+        return;
+    }
+
     std::unique_ptr<expireEntry> spexpire;
     {   // scope pexpireOld
     std::unique_lock<fastlock> ul(g_expireLock);
@@ -1582,12 +1588,6 @@ void moveCommand(client *c) {
     dbDelete(src,c->argv[1]);
     g_pserver->dirty++;
 
-    /* Return zero if the key already exists in the target DB */
-    if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
-        addReply(c,shared.czero);
-        decrRefCount(o);
-        return;
-    }
     dbAdd(dst,c->argv[1],o);
     if (spexpire != nullptr) setExpire(c,dst,c->argv[1],std::move(*spexpire));
 
@@ -2639,7 +2639,10 @@ bool redisDbPersistentData::insert(char *key, robj *o, bool fAssumeNew, dict_ite
         ensure(key);
     dictEntry *de;
     int res = dictAdd(m_pdict, key, o, &de);
-    serverAssert(FImplies(fAssumeNew, res == DICT_OK));
+    if (!FImplies(fAssumeNew, res == DICT_OK)) {
+        serverLog(LL_WARNING,
+            "Assumed new key %s existed in DB.", key);
+    }
     if (res == DICT_OK)
     {
 #ifdef CHECKED_BUILD
@@ -3292,6 +3295,10 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
     auto cmd = lookupCommand(szFromObj(command.argv[0]));
     if (cmd == nullptr)
         return false; // Bad command? It's not for us to judge, just bail
+    
+    if (command.argc < std::abs(cmd->arity))
+        return false; // Invalid number of args
+    
     int numkeys = getKeysFromCommand(cmd, command.argv, command.argc, &result);
     for (int ikey = 0; ikey < numkeys; ++ikey)
     {

src/gc.h

@@ -16,7 +16,17 @@ class GarbageCollector
     struct EpochHolder
     {
         uint64_t tstamp;
-        std::vector<std::unique_ptr<T>> m_vecObjs;
+        std::unique_ptr<std::vector<std::unique_ptr<T>>> m_spvecObjs;
+
+        EpochHolder() {
+            m_spvecObjs = std::make_unique<std::vector<std::unique_ptr<T>>>();
+        }
+
+        // Support move operators
+        EpochHolder(EpochHolder &&other) = default;
+        EpochHolder &operator=(EpochHolder &&) = default;
+
+
 
         bool operator<(uint64_t tstamp) const
         {
@@ -108,12 +118,12 @@ public:
         {
             EpochHolder e;
             e.tstamp = m_epochNext+1;
-            e.m_vecObjs.push_back(std::move(sp));
+            e.m_spvecObjs->push_back(std::move(sp));
             m_listepochs.emplace_back(std::move(e));
         }
         else
         {
-            itr->m_vecObjs.push_back(std::move(sp));
+            itr->m_spvecObjs->push_back(std::move(sp));
         }
     }
 
@@ -123,4 +133,4 @@ private:
     std::list<EpochHolder> m_listepochs;
     std::unordered_set<uint64_t> m_setepochOutstanding;
     uint64_t m_epochNext = 0;
-};
+};

src/networking.cpp

@@ -1985,9 +1985,9 @@ void ProcessPendingAsyncWrites()
          * writes may have been signalled without having been copied to the replyAsync buffer,
          * thus causing the buffer to be NULL */ 
         if (c->replyAsync != nullptr){
-            int size = c->replyAsync->used;
+            size_t size = c->replyAsync->used;
 
-            if (listLength(c->reply) == 0 && size <= (PROTO_REPLY_CHUNK_BYTES - c->bufpos)) {
+            if (listLength(c->reply) == 0 && size <= static_cast<size_t>(PROTO_REPLY_CHUNK_BYTES - c->bufpos)) {
                 memcpy(c->buf + c->bufpos, c->replyAsync->buf(), size);
                 c->bufpos += size;
             } else {

src/object.cpp

@@ -1091,7 +1091,7 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
     mem_total += g_pserver->initial_memory_usage;
 
     mem = 0;
-    if (g_pserver->repl_backlog)
+    if (g_pserver->repl_backlog && g_pserver->repl_backlog != g_pserver->repl_backlog_disk)
         mem += zmalloc_size(g_pserver->repl_backlog);
     mh->repl_backlog = mem;
     mem_total += mem;

src/redis-cli.c

@@ -5485,8 +5485,8 @@ static int clusterManagerCommandRebalance(int argc, char **argv) {
             listAddNodeTail(lsrc, src);
             table = clusterManagerComputeReshardTable(lsrc, numslots);
             listRelease(lsrc);
-            int table_len = (int) listLength(table);
-            if (!table || table_len != numslots) {
+            int table_len = 0;
+            if (!table || (table_len = (int) listLength(table)) != numslots) {
                 clusterManagerLogErr("*** Assertion failed: Reshard table "
                                      "!= number of slots");
                 result = 0;

src/replication.cpp

@@ -190,7 +190,7 @@ int bg_unlink(const char *filename) {
 bool createDiskBacklog() {
     // Lets create some disk backed pages and add them here
     std::string path = "./repl-backlog-temp" + std::to_string(gettid());
-#ifdef __APPLE__
+#if (defined __APPLE__ || defined __FreeBSD__)
     int fd = open(path.c_str(), O_CREAT | O_RDWR, S_IRUSR | S_IWUSR);
 #else
     int fd = open(path.c_str(), O_CREAT | O_RDWR | O_LARGEFILE, S_IRUSR | S_IWUSR);
@@ -1132,6 +1132,14 @@ public:
             replica->repl_put_online_on_ack = 1;
         }
     }
+
+    void abort() {
+        for (auto replica : replicas) {
+            // Close the connection to force a resync
+            freeClientAsync(replica);
+        }
+        replicas.clear();
+    }
 };
 
 int rdbSaveSnapshotForReplication(struct rdbSaveInfo *rsi) {
@@ -1227,7 +1235,11 @@ int rdbSaveSnapshotForReplication(struct rdbSaveInfo *rsi) {
                 retval = C_ERR;
                 break;
             }
-            serverAssert(count == snapshotDeclaredCount);
+            if (count != snapshotDeclaredCount) {
+                serverLog(LL_WARNING, "Replication BUG: Count of keys sent does not match actual count.  Aborting full sync.");
+                replBuf.abort();
+                break;
+            }
         }
         
         replBuf.end();

src/server.cpp

@@ -2888,7 +2888,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
     static thread_local bool fFirstRun = true;
     // note: we also copy the DB pointer in case a DB swap is done while the lock is released
     std::vector<redisDb*> vecdb;    // note we cache the database pointer in case a dbswap is done while the lock is released
-    if (cserver.storage_memory_model == STORAGE_WRITETHROUGH && g_pserver->m_pstorageFactory != nullptr && !g_pserver->loading)
+    if (cserver.storage_memory_model == STORAGE_WRITETHROUGH && !g_pserver->loading)
     {
         if (!fFirstRun) {
             mstime_t storage_process_latency;
@@ -5053,6 +5053,8 @@ bool client::asyncCommand(std::function<void(const redisDbPersistentDataSnapshot
                             std::function<void(const redisDbPersistentDataSnapshot *)> &&postFn) 
 {
     serverAssert(FCorrectThread(this));
+    if (serverTL->in_eval)
+        return false;   // we cannot block clients in EVAL
     const redisDbPersistentDataSnapshot *snapshot = nullptr;
     if (!(this->flags & (CLIENT_MULTI | CLIENT_BLOCKED)))
         snapshot = this->db->createSnapshot(this->mvccCheckpoint, false /* fOptional */);

src/storage/teststorageprovider.cpp

@@ -26,12 +26,10 @@ TestStorageProvider::~TestStorageProvider()
 {
 }
 
-void TestStorageProvider::insert(const char *key, size_t cchKey, void *data, size_t cb, bool fOverwrite)
+void TestStorageProvider::insert(const char *key, size_t cchKey, void *data, size_t cb, bool /*fOverwrite*/)
 {
     auto strkey = std::string(key, cchKey);
-    bool fActuallyExists = m_map.find(strkey) != m_map.end();
-    serverAssert(fActuallyExists == fOverwrite);
-    m_map.insert(std::make_pair(strkey, std::string((char*)data, cb)));
+    m_map[strkey] = std::string((char*)data, cb);
 }
 
 

src/t_stream.cpp

@@ -789,13 +789,13 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
              * update it after (and if) we actually remove the entry */
             unsigned char *pcopy = p;
 
-            int flags = lpGetInteger(p);
+            int64_t flags = lpGetInteger(p);
             p = lpNext(lp, p); /* Skip flags. */
-            int to_skip;
+            int64_t to_skip;
 
-            int ms_delta = lpGetInteger(p);
+            int64_t ms_delta = lpGetInteger(p);
             p = lpNext(lp, p); /* Skip ID ms delta */
-            int seq_delta = lpGetInteger(p);
+            int64_t seq_delta = lpGetInteger(p);
             p = lpNext(lp, p); /* Skip ID seq delta */
 
             streamID currid = {0}; /* For MINID */
@@ -1125,7 +1125,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
              * the first entry emitted for this listpack, then we already
              * emitted the current entry, and have to go back to the previous
              * one. */
-            int lp_count = lpGetInteger(si->lp_ele);
+            int64_t lp_count = lpGetInteger(si->lp_ele);
             while(lp_count--) si->lp_ele = lpPrev(si->lp,si->lp_ele);
             /* Seek lp-count of prev entry. */
             si->lp_ele = lpPrev(si->lp,si->lp_ele);
@@ -1155,7 +1155,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
 
             /* Get the flags entry. */
             si->lp_flags = si->lp_ele;
-            int flags = lpGetInteger(si->lp_ele);
+            int64_t flags = lpGetInteger(si->lp_ele);
             si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek ID. */
 
             /* Get the ID: it is encoded as difference between the master
@@ -1264,7 +1264,7 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
      * deleted entries in the listpack header.
      *
      * We start flagging: */
-    int flags = lpGetInteger(si->lp_flags);
+    int64_t flags = lpGetInteger(si->lp_flags);
     flags |= STREAM_ITEM_FLAG_DELETED;
     lp = lpReplaceInteger(lp,&si->lp_flags,flags);
 

tests/unit/scripting.tcl

@@ -126,6 +126,10 @@ start_server {tags {"scripting"}} {
         r select 9
         set res
     } {original value}
+    
+    test {EVAL background command} {
+	r eval {redis.call("SCAN", "0", "MATCH", "key*", "COUNT", 1000)} 0
+    } {}
 
     if 0 {
         test {EVAL - Script can't run more than configured time limit} {

tests/unit/type/stream.tcl

@@ -172,6 +172,15 @@ start_server {
         assert_equal [r XRANGE mystream - +] {{3-0 {f v}} {4-0 {f v}} {5-0 {f v}}}
     }
 
+    test {XTRIM with MINID option, big delta from master record} {
+        r DEL mystream
+        r XADD mystream 1-0 f v
+        r XADD mystream 1641544570597-0 f v
+        r XADD mystream 1641544570597-1 f v
+        r XTRIM mystream MINID 1641544570597-0
+        assert_equal [r XRANGE mystream - +] {{1641544570597-0 {f v}} {1641544570597-1 {f v}}}
+    }
+
     test {XADD mass insertion and XLEN} {
         r DEL mystream
         r multi