Sample client enable/disable module API

Redis 6.0.5

00-RELEASENOTES

@@ -11,6 +11,76 @@ CRITICAL: There is a critical bug affecting MOST USERS. Upgrade ASAP.
 SECURITY: There are security fixes in the release.
 --------------------------------------------------------------------------------
 
+================================================================================
+Redis 6.0.5     Released Tue Jun 09 11:56:08 CEST 2020
+================================================================================
+
+Upgrade urgency MODERATE: several bugs with moderate impact are fixed here.
+
+The most important issues are listed here:
+
+* Fix handling of speical chars in ACL LOAD.
+* Make Redis Cluster more robust about operation errors that may lead
+  to two clusters to mix together.
+* Revert the sendfile() implementation of RDB transfer. It causes some delay.
+* Fix TLS certificate loading for chained certificates.
+* Fix AOF rewirting of KEEPTTL SET option.
+* Fix MULTI/EXEC behavior during -BUSY script errors.
+
+And this is the full list of commits:
+
+antirez in commit ee8dd01bb:
+ Temporary fix for #7353 issue about EVAL during -BUSY.
+ 1 file changed, 9 insertions(+)
+
+xhe in commit a4a856d53:
+ return the correct proto version HELLO should return the current proto version, while the code hardcoded 3
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+Oran Agra in commit e2046b300:
+ Don't queue commands in an already aborted MULTI state
+ 1 file changed, 7 insertions(+)
+
+Oran Agra in commit b35fdf1de:
+ Avoid rejecting WATCH / UNWATCH, like MULTI/EXEC/DISCARD
+ 1 file changed, 4 insertions(+), 2 deletions(-)
+
+zhaozhao.zz in commit 1d7bf208c:
+ AOF: append origin SET if no expire option
+ 2 files changed, 23 insertions(+), 8 deletions(-)
+
+Oran Agra in commit 676445ad9:
+ fix disconnectSlaves, to try to free each slave.
+ 1 file changed, 1 deletion(-)
+
+zhaozhao.zz in commit 4846c0c8a:
+ donot free protected client in freeClientsInAsyncFreeQueue
+ 1 file changed, 9 insertions(+), 3 deletions(-)
+
+Oran Agra in commit f33de403e:
+ fix pingoff  test race
+ 1 file changed, 1 insertion(+)
+
+Kevin Fwu in commit 49af4d07e:
+ Fix TLS certificate loading for chained certificates.
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+antirez in commit 329fddbda:
+ Revert "Implements sendfile for redis."
+ 2 files changed, 2 insertions(+), 55 deletions(-)
+
+antirez in commit 925a2cd5a:
+ Revert "avoid using sendfile if tls-replication is enabled"
+ 1 file changed, 27 insertions(+), 34 deletions(-)
+
+Liu Zhen in commit 84a7a9058:
+ fix clusters mixing accidentally by gossip
+ 1 file changed, 10 insertions(+), 2 deletions(-)
+
+antirez in commit cd63359a1:
+ Fix handling of special chars in ACL LOAD.
+ 1 file changed, 8 insertions(+), 4 deletions(-)
+
 ================================================================================
 Redis 6.0.4     Released Thu May 28 11:36:45 CEST 2020
 ================================================================================

README.md

@@ -1,6 +1,5 @@
 ![Current Release](https://img.shields.io/github/release/JohnSully/KeyDB.svg)
 ![CI](https://github.com/JohnSully/KeyDB/workflows/CI/badge.svg?branch=unstable)
-[![Join the chat at https://gitter.im/KeyDB/community](https://badges.gitter.im/KeyDB/community.svg)](https://gitter.im/KeyDB/community?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 [![StackShare](http://img.shields.io/badge/tech-stack-0690fa.svg?style=flat)](https://stackshare.io/eq-alpha-technology-inc/eq-alpha-technology-inc)
 
 ##### New! Want to extend KeyDB with Javascript?  Try [ModJS](https://github.com/JohnSully/ModJS)

keydb.conf

@@ -1811,6 +1811,15 @@ jemalloc-bg-thread yes
 # Set bgsave child process to cpu affinity 1,10,11
 # bgsave_cpulist 1,10-11
 
+# The minimum number of clients on a thread before KeyDB assigns new connections to a different thread
+#  Tuning this parameter is a tradeoff between locking overhead and distributing the workload over multiple cores
+# min-clients-per-thread 50
+
+# Avoid forwarding RREPLAY messages to other masters?
+#   WARNING: This setting is dangerous! You must be certain all masters are connected to each
+#   other in a true mesh topology or data loss will occur!
+#   This command can be used to reduce multimaster bus traffic
+# multi-master-no-forward no
 
 # Path to directory for file backed scratchpad.  The file backed scratchpad
 # reduces memory requirements by storing rarely accessed data on disk 

src/Makefile

@@ -47,6 +47,11 @@ endif
 
 USEASM?=true
 
+ifeq ($(NOMVCC),)
+	CFLAGS+= -DENABLE_MVCC
+	CXXFLAGS+= -DENABLE_MVCC
+endif
+
 ifneq ($(SANITIZE),)
 	CFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE
 	CXXFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE
@@ -402,7 +407,7 @@ bench: $(REDIS_BENCHMARK_NAME)
 	$(MAKE) CXXFLAGS="-m32" CFLAGS="-m32" LDFLAGS="-m32"
 
 gcov:
-	$(MAKE) REDIS_CFLAGS="-fprofile-arcs -ftest-coverage -DCOVERAGE_TEST" REDIS_LDFLAGS="-fprofile-arcs -ftest-coverage"
+	$(MAKE) REDIS_CXXFLAGS="-fprofile-arcs -ftest-coverage -DCOVERAGE_TEST" REDIS_CFLAGS="-fprofile-arcs -ftest-coverage -DCOVERAGE_TEST" REDIS_LDFLAGS="-fprofile-arcs -ftest-coverage"
 
 noopt:
 	$(MAKE) OPTIMIZATION="-O0"

src/acl.cpp

@@ -735,10 +735,11 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
  * EEXIST: You are adding a key pattern after "*" was already added. This is
  *         almost surely an error on the user side.
  * ENODEV: The password you are trying to remove from the user does not exist.
- * EBADMSG: The hash you are trying to add is not a valid hash. 
+ * EBADMSG: The hash you are trying to add is not a valid hash.
  */
 int ACLSetUser(user *u, const char *op, ssize_t oplen) {
     if (oplen == -1) oplen = strlen(op);
+    if (oplen == 0) return C_OK; /* Empty string is a no-operation. */
     if (!strcasecmp(op,"on")) {
         u->flags |= USER_FLAG_ENABLED;
         u->flags &= ~USER_FLAG_DISABLED;
@@ -1300,7 +1301,7 @@ sds ACLLoadFromFile(const char *filename) {
         if (lines[i][0] == '\0') continue;
 
         /* Split into arguments */
-        argv = sdssplitargs(lines[i],&argc);
+        argv = sdssplitlen(lines[i],sdslen(lines[i])," ",1,&argc);
         if (argv == NULL) {
             errors = sdscatprintf(errors,
                      "%s:%d: unbalanced quotes in acl line. ",
@@ -1332,11 +1333,14 @@ sds ACLLoadFromFile(const char *filename) {
             continue;
         }
 
-        /* Try to process the line using the fake user to validate iif
-         * the rules are able to apply cleanly. */
+        /* Try to process the line using the fake user to validate if
+         * the rules are able to apply cleanly. At this stage we also
+         * trim trailing spaces, so that we don't have to handle that
+         * in ACLSetUser(). */
         ACLSetUser(fakeuser,"reset",-1);
         int j;
         for (j = 2; j < argc; j++) {
+            argv[j] = sdstrim(argv[j],"\t\r\n");
             if (ACLSetUser(fakeuser,argv[j],sdslen(argv[j])) != C_OK) {
                 const char *errmsg = ACLSetUserStringError();
                 errors = sdscatprintf(errors,

src/ae.cpp

@@ -234,7 +234,7 @@ int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
 
     int ret = AE_OK;
     
-    aeCommand cmd = {};
+    aeCommand cmd;
     cmd.op = AE_ASYNC_OP::CreateFileEvent;
     cmd.fd = fd;
     cmd.mask = mask;
@@ -258,9 +258,11 @@ int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
     
     if (fSynchronous)
     {
-        std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::defer_lock);
+        {
+        std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::adopt_lock);
         cmd.pctl->cv.wait(ulock);
         ret = cmd.pctl->rval;
+        }
         delete cmd.pctl;
     }
 
@@ -294,14 +296,13 @@ int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynch
     }
 
     aeCommand cmd = {};
-    memset(&cmd, 0, sizeof(aeCommand));
     cmd.op = AE_ASYNC_OP::PostCppFunction;
     cmd.pfn = new (MALLOC_LOCAL) std::function<void()>(fn);
     cmd.pctl = nullptr;
     cmd.fLock = fLock;
     if (fSynchronous)
     {
-        cmd.pctl = new (MALLOC_LOCAL) aeCommandControl();
+        cmd.pctl = new (MALLOC_LOCAL) aeCommandControl;
         cmd.pctl->mutexcv.lock();
     }
 
@@ -312,9 +313,11 @@ int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynch
     int ret = AE_OK;
     if (fSynchronous)
     {
-        std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::defer_lock);
+        {
+        std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::adopt_lock);
         cmd.pctl->cv.wait(ulock);
         ret = cmd.pctl->rval;
+        }
         delete cmd.pctl;
     }
     return ret;
@@ -454,7 +457,7 @@ void aeDeleteFileEventAsync(aeEventLoop *eventLoop, int fd, int mask)
 {
     if (eventLoop == g_eventLoopThisThread)
         return aeDeleteFileEvent(eventLoop, fd, mask);
-    aeCommand cmd;
+    aeCommand cmd = {};
     cmd.op = AE_ASYNC_OP::DeleteFileEvent;
     cmd.fd = fd;
     cmd.mask = mask;

src/aof.cpp

@@ -671,19 +671,23 @@ sds catCommandForAofAndActiveReplication(sds buf, struct redisCommand *cmd, robj
     } else if (cmd->proc == setCommand && argc > 3) {
         int i;
         robj *exarg = NULL, *pxarg = NULL;
-        /* Translate SET [EX seconds][PX milliseconds] to SET and PEXPIREAT */
-        buf = catAppendOnlyGenericCommand(buf,3,argv);
         for (i = 3; i < argc; i ++) {
             if (!strcasecmp(szFromObj(argv[i]), "ex")) exarg = argv[i+1];
             if (!strcasecmp(szFromObj(argv[i]), "px")) pxarg = argv[i+1];
         }
         serverAssert(!(exarg && pxarg));
-        if (exarg)
-            buf = catAppendOnlyExpireAtCommand(buf,cserver.expireCommand,argv[1],
-                                               exarg);
-        if (pxarg)
-            buf = catAppendOnlyExpireAtCommand(buf,cserver.pexpireCommand,argv[1],
-                                               pxarg);
+        if (exarg || pxarg) {
+            /* Translate SET [EX seconds][PX milliseconds] to SET and PEXPIREAT */
+            buf = catAppendOnlyGenericCommand(buf,3,argv);
+            if (exarg)
+                buf = catAppendOnlyExpireAtCommand(buf,cserver.expireCommand,argv[1],
+                                                   exarg);
+            if (pxarg)
+                buf = catAppendOnlyExpireAtCommand(buf,cserver.pexpireCommand,argv[1],
+                                                   pxarg);
+        } else {
+            buf = catAppendOnlyGenericCommand(buf,argc,argv);
+        }
     } else if (cmd->proc == expireMemberCommand || cmd->proc == expireMemberAtCommand ||
         cmd->proc == pexpireMemberAtCommand) {
         /* Translate subkey expire commands to PEXPIREMEMBERAT */

src/cluster.cpp

@@ -1501,7 +1501,10 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
             }
         } else {
             /* If it's not in NOADDR state and we don't have it, we
-             * start a handshake process against this IP/PORT pairs.
+             * add it to our trusted dict with exact nodeid and flag.
+             * Note that we cannot simply start a handshake against
+             * this IP/PORT pairs, since IP/PORT can be reused already,
+             * otherwise we risk joining another cluster.
              *
              * Note that we require that the sender of this gossip message
              * is a well known node in our cluster, otherwise we risk
@@ -1510,7 +1513,12 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
                 !(flags & CLUSTER_NODE_NOADDR) &&
                 !clusterBlacklistExists(g->nodename))
             {
-                clusterStartHandshake(g->ip,ntohs(g->port),ntohs(g->cport));
+                clusterNode *node;
+                node = createClusterNode(g->nodename, flags);
+                memcpy(node->ip,g->ip,NET_IP_STR_LEN);
+                node->port = ntohs(g->port);
+                node->cport = ntohs(g->cport);
+                clusterAddNode(node);
             }
         }
 

src/config.cpp

@@ -518,7 +518,7 @@ void loadServerConfigFromString(char *config) {
             err = "KeyDB not compliled with scratch-file support.";
             goto loaderr;
 #endif
-        } else if (!strcasecmp(argv[0],"server-threads") && argc == 2) {
+        } else if ((!strcasecmp(argv[0],"server-threads") || !strcasecmp(argv[0],"io-threads")) && argc == 2) {
             cserver.cthreads = atoi(argv[1]);
             if (cserver.cthreads <= 0 || cserver.cthreads > MAX_EVENT_LOOPS) {
                 err = "Invalid number of threads specified";
@@ -2167,6 +2167,13 @@ static int updateMaxclients(long long val, long long prev, const char **err) {
     return 1;
 }
 
+static int validateMultiMasterNoForward(int val, const char **) {
+    if (val) {
+        serverLog(LL_WARNING, "WARNING: multi-master-no-forward is set, you *must* use a mesh topology or dataloss will occur");
+    }
+    return 1;
+}
+
 #ifdef USE_OPENSSL
 static int updateTlsCfg(char *val, char *prev, const char **err) {
     UNUSED(val);
@@ -2185,6 +2192,7 @@ static int updateTlsCfgBool(int val, int prev, const char **err) {
 }
 #endif  /* USE_OPENSSL */
 
+int fDummy = false;
 standardConfig configs[] = {
     /* Bool configs */
     createBoolConfig("rdbchecksum", NULL, IMMUTABLE_CONFIG, g_pserver->rdb_checksum, 1, NULL, NULL),
@@ -2200,6 +2208,7 @@ standardConfig configs[] = {
     createBoolConfig("lazyfree-lazy-eviction", NULL, MODIFIABLE_CONFIG, g_pserver->lazyfree_lazy_eviction, 0, NULL, NULL),
     createBoolConfig("lazyfree-lazy-expire", NULL, MODIFIABLE_CONFIG, g_pserver->lazyfree_lazy_expire, 0, NULL, NULL),
     createBoolConfig("lazyfree-lazy-server-del", NULL, MODIFIABLE_CONFIG, g_pserver->lazyfree_lazy_server_del, 0, NULL, NULL),
+    createBoolConfig("lazyfree-lazy-user-del", NULL, MODIFIABLE_CONFIG, g_pserver->lazyfree_lazy_user_del , 0, NULL, NULL),
     createBoolConfig("repl-disable-tcp-nodelay", NULL, MODIFIABLE_CONFIG, g_pserver->repl_disable_tcp_nodelay, 0, NULL, NULL),
     createBoolConfig("repl-diskless-sync", NULL, MODIFIABLE_CONFIG, g_pserver->repl_diskless_sync, 0, NULL, NULL),
     createBoolConfig("aof-rewrite-incremental-fsync", NULL, MODIFIABLE_CONFIG, g_pserver->aof_rewrite_incremental_fsync, 1, NULL, NULL),
@@ -2220,6 +2229,7 @@ standardConfig configs[] = {
     createBoolConfig("cluster-enabled", NULL, IMMUTABLE_CONFIG, g_pserver->cluster_enabled, 0, NULL, NULL),
     createBoolConfig("appendonly", NULL, MODIFIABLE_CONFIG, g_pserver->aof_enabled, 0, NULL, updateAppendonly),
     createBoolConfig("cluster-allow-reads-when-down", NULL, MODIFIABLE_CONFIG, g_pserver->cluster_allow_reads_when_down, 0, NULL, NULL),
+    createBoolConfig("multi-master-no-forward", NULL, MODIFIABLE_CONFIG, cserver.multimaster_no_forward, 0, validateMultiMasterNoForward, NULL),
 
     /* String Configs */
     createStringConfig("aclfile", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->acl_filename, "", NULL, NULL),

src/connection.cpp

@@ -164,7 +164,7 @@ static int connSocketWrite(connection *conn, const void *data, size_t data_len)
     int ret = write(conn->fd, data, data_len);
     if (ret < 0 && errno != EAGAIN) {
         conn->last_errno = errno;
-        conn->state = CONN_STATE_ERROR;
+        conn->state.store(CONN_STATE_ERROR, std::memory_order_relaxed);
     }
 
     return ret;
@@ -173,10 +173,10 @@ static int connSocketWrite(connection *conn, const void *data, size_t data_len)
 static int connSocketRead(connection *conn, void *buf, size_t buf_len) {
     int ret = read(conn->fd, buf, buf_len);
     if (!ret) {
-        conn->state = CONN_STATE_CLOSED;
+        conn->state.store(CONN_STATE_CLOSED, std::memory_order_release);
     } else if (ret < 0 && errno != EAGAIN) {
         conn->last_errno = errno;
-        conn->state = CONN_STATE_ERROR;
+        conn->state.store(CONN_STATE_ERROR, std::memory_order_release);
     }
 
     return ret;
@@ -253,14 +253,14 @@ static void connSocketEventHandler(struct aeEventLoop *el, int fd, void *clientD
     UNUSED(fd);
     connection *conn = (connection*)clientData;
 
-    if (conn->state == CONN_STATE_CONNECTING &&
+    if (conn->state.load(std::memory_order_relaxed) == CONN_STATE_CONNECTING &&
             (mask & AE_WRITABLE) && conn->conn_handler) {
 
         if (connGetSocketError(conn)) {
             conn->last_errno = errno;
-            conn->state = CONN_STATE_ERROR;
+            conn->state.store(CONN_STATE_ERROR, std::memory_order_release);
         } else {
-            conn->state = CONN_STATE_CONNECTED;
+            conn->state.store(CONN_STATE_CONNECTED, std::memory_order_release);
         }
 
         if (!conn->write_handler) aeDeleteFileEvent(serverTL->el,conn->fd,AE_WRITABLE);
@@ -407,7 +407,7 @@ int connRecvTimeout(connection *conn, long long ms) {
 }
 
 int connGetState(connection *conn) {
-    return conn->state;
+    return conn->state.load(std::memory_order_relaxed);
 }
 
 void connSetThreadAffinity(connection *conn, int cpu) {

src/db.cpp

@@ -92,9 +92,11 @@ static robj *lookupKey(redisDb *db, robj *key, int flags) {
 
         updateDbValAccess(de, flags);
 
+#ifdef ENABLE_MVCC
         if (flags & LOOKUP_UPDATEMVCC) {
             val->mvcc_tstamp = getMvccTstamp();
         }
+#endif
         return val;
     } else {
         return NULL;
@@ -206,7 +208,9 @@ int dbAddCore(redisDb *db, robj *key, robj *val) {
     serverAssert(!val->FExpires());
     sds copy = sdsdup(szFromObj(key));
     int retval = dictAdd(db->pdict, copy, val);
+#ifdef ENABLE_MVCC
     val->mvcc_tstamp = key->mvcc_tstamp = getMvccTstamp();
+#endif
 
     if (retval == DICT_OK)
     {
@@ -256,7 +260,9 @@ void dbOverwriteCore(redisDb *db, dictEntry *de, robj *key, robj *val, bool fUpd
     if (fUpdateMvcc) {
         if (val->getrefcount(std::memory_order_relaxed) == OBJ_SHARED_REFCOUNT)
             val = dupStringObject(val);
+#ifdef ENABLE_MVCC
         val->mvcc_tstamp = getMvccTstamp();
+#endif
     }
 
     dictSetVal(db->pdict, de, val);
@@ -290,13 +296,15 @@ int dbMerge(redisDb *db, robj *key, robj *val, int fReplace)
         if (de == nullptr)
             return (dbAddCore(db, key, val) == DICT_OK);
 
+#ifdef ENABLE_MVCC
         robj *old = (robj*)dictGetVal(de);
         if (old->mvcc_tstamp <= val->mvcc_tstamp)
         {
             dbOverwriteCore(db, de, key, val, false, true);
             return true;
         }
-        
+#endif
+
         return false;
     }
     else
@@ -683,6 +691,13 @@ void existsCommand(client *c) {
     addReplyLongLong(c,count);
 }
 
+void mexistsCommand(client *c) {
+    addReplyArrayLen(c, c->argc - 1);
+    for (int j = 1; j < c->argc; ++j) {
+        addReplyBool(c, lookupKeyRead(c->db, c->argv[j]));
+    }
+}
+
 void selectCommand(client *c) {
     long id;
 
@@ -1219,8 +1234,7 @@ int dbSwapDatabases(long id1, long id2) {
     if (id1 < 0 || id1 >= cserver.dbnum ||
         id2 < 0 || id2 >= cserver.dbnum) return C_ERR;
     if (id1 == id2) return C_OK;
-    redisDb aux; 
-    memcpy(&aux, &g_pserver->db[id1], sizeof(redisDb));
+    redisDb aux(g_pserver->db[id1]);
     redisDb *db1 = &g_pserver->db[id1], *db2 = &g_pserver->db[id2];
 
     /* Swap hash tables. Note that we don't swap blocking_keys,

src/help.h

@@ -1325,7 +1325,12 @@ struct commandHelp {
     "Rename a hash key, copying the value.",
     4,
     "6.5.3"
-    }
+    },
+    { "KEYDB.MEXISTS",
+    "key [key ...]",
+    "Determine if a key exists",
+    0,
+    "6.5.12" },
 };
 
 #endif

src/module.cpp

@@ -4604,6 +4604,18 @@ int moduleUnblockClientByHandle(RedisModuleBlockedClient *bc, void *privdata) {
     return REDISMODULE_OK;
 }
 
+void RM_DisableClientAsync(RedisModuleCtx *ctx) {
+    ctx->client->postFunction([](client *c) {
+        connSetReadHandler(c->conn, nullptr, true);
+    });
+}
+
+void RM_EnableClientAsync(RedisModuleCtx *ctx) {
+    ctx->client->postFunction([](client *c) {
+        connSetReadHandler(c->conn, readQueryFromClient, true /* fThreadSafe */);
+    });
+}
+
 /* This API is used by the Redis core to unblock a client that was blocked
  * by a module. */
 void moduleUnblockClient(client *c) {
@@ -4895,14 +4907,17 @@ void RM_FreeThreadSafeContext(RedisModuleCtx *ctx) {
     zfree(ctx);
 }
 
+static bool g_fModuleThread = false;
 /* Acquire the server lock before executing a thread safe API call.
  * This is not needed for `RedisModule_Reply*` calls when there is
  * a blocked client connected to the thread safe context. */
 void RM_ThreadSafeContextLock(RedisModuleCtx *ctx) {
     UNUSED(ctx);
-    moduleAcquireGIL(FALSE /*fServerThread*/);
-    if (serverTL == nullptr)
+    if (serverTL == nullptr) {
         serverTL = &g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN];    // arbitrary module threads get the main thread context
+        g_fModuleThread = true;
+    }
+    moduleAcquireGIL(FALSE /*fServerThread*/);
 }
 
 /* Release the server lock after a thread safe API call was executed. */
@@ -4911,10 +4926,20 @@ void RM_ThreadSafeContextUnlock(RedisModuleCtx *ctx) {
     moduleReleaseGIL(FALSE /*fServerThread*/);
 }
 
+// A module may be triggered synchronously in a non-module context.  In this scenario we don't lock again
+//  as the server thread acquisition is sufficient.  If we did try to lock we would deadlock
+static bool FModuleCallBackLock(bool fServerThread)
+{
+    return !fServerThread && aeThreadOwnsLock() && !g_fModuleThread && s_cAcquisitionsServer > 0;
+}
 void moduleAcquireGIL(int fServerThread) {
     std::unique_lock<std::mutex> lock(s_mutex);
     int *pcheck = fServerThread ? &s_cAcquisitionsModule : &s_cAcquisitionsServer;
 
+    if (FModuleCallBackLock(fServerThread)) {
+        return;
+    }
+
     while (*pcheck > 0)
         s_cv.wait(lock);
 
@@ -4932,6 +4957,10 @@ void moduleAcquireGIL(int fServerThread) {
 void moduleReleaseGIL(int fServerThread) {
     std::unique_lock<std::mutex> lock(s_mutex);
 
+    if (FModuleCallBackLock(fServerThread)) {
+        return;
+    }
+
     if (fServerThread)
     {
         --s_cAcquisitionsServer;

src/multi.cpp

@@ -59,6 +59,13 @@ void queueMultiCommand(client *c) {
     multiCmd *mc;
     int j;
 
+    /* No sense to waste memory if the transaction is already aborted.
+     * this is useful in case client sends these in a pipeline, or doesn't
+     * bother to read previous responses and didn't notice the multi was already
+     * aborted. */
+    if (c->flags & CLIENT_DIRTY_EXEC)
+        return;
+
     c->mstate.commands = (multiCmd*)zrealloc(c->mstate.commands,
             sizeof(multiCmd)*(c->mstate.count+1), MALLOC_LOCAL);
     mc = c->mstate.commands+c->mstate.count;
@@ -131,6 +138,15 @@ void execCommand(client *c) {
         return;
     }
 
+    /* If we are in -BUSY state, flag the transaction and return the
+     * -BUSY error, like Redis <= 5. This is a temporary fix, may be changed
+     *  ASAP, see issue #7353 on Github. */
+    if (g_pserver->lua_timedout) {
+        flagTransaction(c);
+        addReply(c, shared.slowscripterr);
+        return;
+    }
+
     /* Check if we need to abort the EXEC because:
      * 1) Some WATCHed key was touched.
      * 2) There was a previous error while queueing commands.

src/networking.cpp

@@ -1645,7 +1645,7 @@ int freeClientsInAsyncFreeQueue(int iel) {
     while((ln = listNext(&li))) 
     {
         client *c = (client*)listNodeValue(ln);
-        if (c->iel == iel)
+        if (c->iel == iel && !(c->flags & CLIENT_PROTECTED))
         {
             vecclientsFree.push_back(c);
             listDelNode(g_pserver->clients_to_close, ln);
@@ -1928,9 +1928,12 @@ int handleClientsWithPendingWrites(int iel, int aof_state) {
         }
     }
 
-    AeLocker locker;
-    locker.arm(nullptr);
-    ProcessPendingAsyncWrites();
+    if (listLength(serverTL->clients_pending_asyncwrite))
+    {
+        AeLocker locker;
+        locker.arm(nullptr);
+        ProcessPendingAsyncWrites();
+    }
 
     return processed;
 }
@@ -3057,7 +3060,7 @@ void helloCommand(client *c) {
     addReplyBulkCString(c,KEYDB_SET_VERSION);
 
     addReplyBulkCString(c,"proto");
-    addReplyLongLong(c,3);
+    addReplyLongLong(c,ver);
 
     addReplyBulkCString(c,"id");
     addReplyLongLong(c,c->id);

src/object.cpp

@@ -46,7 +46,9 @@ robj *createObject(int type, void *ptr) {
     o->encoding = OBJ_ENCODING_RAW;
     o->m_ptr = ptr;
     o->setrefcount(1);
+#ifdef ENABLE_MVCC
     o->mvcc_tstamp = OBJ_MVCC_INVALID;
+#endif
 
     /* Set the LRU to the current lruclock (minutes resolution), or
      * alternatively the LFU counter. */
@@ -101,7 +103,9 @@ robj *createEmbeddedStringObject(const char *ptr, size_t len) {
     o->type = OBJ_STRING;
     o->encoding = OBJ_ENCODING_EMBSTR;
     o->setrefcount(1);
+#ifdef ENABLE_MVCC
     o->mvcc_tstamp = OBJ_MVCC_INVALID;
+#endif
 
     if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU) {
         o->lru = (LFUGetTimeInMinutes()<<8) | LFU_INIT_VAL;
@@ -129,8 +133,13 @@ robj *createEmbeddedStringObject(const char *ptr, size_t len) {
  *
  * The current limit of 52 is chosen so that the biggest string object
  * we allocate as EMBSTR will still fit into the 64 byte arena of jemalloc. */
+#ifdef ENABLE_MVCC
 #define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 48
-static_assert((sizeof(redisObject)+OBJ_ENCODING_EMBSTR_SIZE_LIMIT-8) == 64, "Max EMBSTR obj should be 64 bytes total");
+#else
+#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 256
+#endif
+
+//static_assert((sizeof(redisObject)+OBJ_ENCODING_EMBSTR_SIZE_LIMIT-8) == 64, "Max EMBSTR obj should be 64 bytes total");
 robj *createStringObject(const char *ptr, size_t len) {
     if (len <= OBJ_ENCODING_EMBSTR_SIZE_LIMIT)
         return createEmbeddedStringObject(ptr,len);
@@ -1317,10 +1326,12 @@ NULL
          * because we update the access time only
          * when the key is read or overwritten. */
         addReplyLongLong(c,LFUDecrAndReturn(o));
+#ifdef ENABLE_MVCC
     } else if (!strcasecmp(szFromObj(c->argv[1]), "lastmodified") && c->argc == 3) {
         if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
                 == NULL) return;
         addReplyLongLong(c, (g_pserver->mstime - (o->mvcc_tstamp >> MVCC_MS_SHIFT)) / 1000);
+#endif
     } else {
         addReplySubcommandSyntaxError(c);
     }

src/rdb.cpp

@@ -1089,8 +1089,10 @@ int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, expireEntry *pexpire) {
     }
 
     char szT[32];
+#ifdef ENABLE_MVCC
     snprintf(szT, 32, "%" PRIu64, val->mvcc_tstamp);
     if (rdbSaveAuxFieldStrStr(rdb,"mvcc-tstamp", szT) == -1) return -1;
+#endif
 
     /* Save type, key, value */
     if (rdbSaveObjectType(rdb,val) == -1) return -1;
@@ -2046,7 +2048,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, uint64_t mvcc_tstamp) {
         return NULL;
     }
 
+#ifdef ENABLE_MVCC
     o->mvcc_tstamp = mvcc_tstamp;
+#endif
     serverAssert(!o->FExpires());
     return o;
 }
@@ -2398,7 +2402,11 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
             key = nullptr;
             goto eoferr;
         }
+#ifdef ENABLE_MVCC
         bool fStaleMvccKey = (rsi) ? val->mvcc_tstamp < rsi->mvccMinThreshold : false;
+#else
+        bool fStaleMvccKey = false;
+#endif
 
         /* Check if the key already expired. This function is used when loading
          * an RDB file from disk, either at startup, or when an RDB was
@@ -2412,9 +2420,8 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
             if (fStaleMvccKey && !fExpiredKey && rsi != nullptr && rsi->mi != nullptr && rsi->mi->staleKeyMap != nullptr && lookupKeyRead(db, &keyobj) == nullptr) {
                 // We have a key that we've already deleted and is not back in our database.
                 //  We'll need to inform the sending master of the delete if it is also a replica of us
-                robj *objKeyDup = createStringObject(key, sdslen(key));
-                rsi->mi->staleKeyMap->operator[](db - g_pserver->db).push_back(objKeyDup);
-                decrRefCount(objKeyDup);
+                robj_sharedptr objKeyDup(createStringObject(key, sdslen(key)));
+                rsi->mi->staleKeyMap->operator[](db - g_pserver->db).push_back(objKeyDup);                
             }
             fLastKeyExpired = true;
             sdsfree(key);

src/replication.cpp

@@ -1370,30 +1370,28 @@ void sendBulkToSlave(connection *conn) {
      * try to use sendfile system call if supported, unless tls is enabled.
      * fallback to normal read+write otherwise. */
     nwritten = 0;
-    if (!nwritten) {
-        ssize_t buflen;
-        char buf[PROTO_IOBUF_LEN];
+    ssize_t buflen;
+    char buf[PROTO_IOBUF_LEN];
 
-        lseek(replica->repldbfd,replica->repldboff,SEEK_SET);
-        buflen = read(replica->repldbfd,buf,PROTO_IOBUF_LEN);
-        if (buflen <= 0) {
-            serverLog(LL_WARNING,"Read error sending DB to replica: %s",
-                (buflen == 0) ? "premature EOF" : strerror(errno));
+    lseek(replica->repldbfd,replica->repldboff,SEEK_SET);
+    buflen = read(replica->repldbfd,buf,PROTO_IOBUF_LEN);
+    if (buflen <= 0) {
+        serverLog(LL_WARNING,"Read error sending DB to replica: %s",
+            (buflen == 0) ? "premature EOF" : strerror(errno));
+        ul.unlock();
+        aeLock.arm(nullptr);
+        freeClient(replica);
+        return;
+    }
+    if ((nwritten = connWrite(conn,buf,buflen)) == -1) {
+        if (connGetState(conn) != CONN_STATE_CONNECTED) {
+            serverLog(LL_WARNING,"Write error sending DB to replica: %s",
+                connGetLastError(conn));
             ul.unlock();
             aeLock.arm(nullptr);
             freeClient(replica);
-            return;
-        }
-        if ((nwritten = connWrite(conn,buf,buflen)) == -1) {
-            if (connGetState(conn) != CONN_STATE_CONNECTED) {
-                serverLog(LL_WARNING,"Write error sending DB to replica: %s",
-                    connGetLastError(conn));
-                ul.unlock();
-                aeLock.arm(nullptr);
-                freeClient(replica);
-            }
-            return;
         }
+        return;
     }
 
     replica->repldboff += nwritten;
@@ -2069,7 +2067,6 @@ void readSyncBulkPayload(connection *conn) {
      *
      * 2. Or when we are done reading from the socket to the RDB file, in
      *    such case we want just to read the RDB file in memory. */
-    serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Flushing old data");
 
     /* We need to stop any AOF rewriting child before flusing and parsing
      * the RDB, otherwise we'll create a copy-on-write disaster. */
@@ -2089,7 +2086,10 @@ void readSyncBulkPayload(connection *conn) {
      * (Where disklessLoadMakeBackups left server.db empty) because we
      * want to execute all the auxiliary logic of emptyDb (Namely,
      * fire module events) */
-    emptyDb(-1,empty_db_flags,replicationEmptyDbCallback);
+    if (!fUpdate) {
+        serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Flushing old data");
+        emptyDb(-1,empty_db_flags,replicationEmptyDbCallback);
+    }
 
     /* Before loading the DB into memory we need to delete the readable
      * handler, otherwise it will get called recursively since
@@ -3151,15 +3151,46 @@ void replicaofCommand(client *c) {
         return;
     }
 
-    /* The special host/port combination "NO" "ONE" turns the instance
-     * into a master. Otherwise the new master address is set. */
-    if (!strcasecmp((const char*)ptrFromObj(c->argv[1]),"no") &&
+    if (c->argc > 3) {
+        if (c->argc != 4) {
+            addReplyError(c, "Invalid arguments");
+            return;
+        }
+        if (!strcasecmp((const char*)ptrFromObj(c->argv[1]),"remove")) {
+            listIter li;
+            listNode *ln;
+            bool fRemoved = false;
+            long port;
+            string2l(szFromObj(c->argv[3]), sdslen(szFromObj(c->argv[3])), &port);
+        LRestart:
+            listRewind(g_pserver->masters, &li);
+            while ((ln = listNext(&li))) {
+                redisMaster *mi = (redisMaster*)listNodeValue(ln);
+                if (mi->masterport != port)
+                    continue;
+                if (sdscmp(szFromObj(c->argv[2]), mi->masterhost) == 0) {
+                    replicationUnsetMaster(mi);
+                    fRemoved = true;
+                    goto LRestart;
+                }
+            }
+            if (!fRemoved) {
+                addReplyError(c, "Master not found");
+                return;
+            } else if (listLength(g_pserver->masters) == 0) {
+                goto LLogNoMaster;
+            }
+        }
+    } else if (!strcasecmp((const char*)ptrFromObj(c->argv[1]),"no") &&
         !strcasecmp((const char*)ptrFromObj(c->argv[2]),"one")) {
+        /* The special host/port combination "NO" "ONE" turns the instance
+         * into a master. Otherwise the new master address is set. */
         if (listLength(g_pserver->masters)) {
             while (listLength(g_pserver->masters))
             {
                 replicationUnsetMaster((redisMaster*)listNodeValue(listFirst(g_pserver->masters)));
             }
+        LLogNoMaster:
             sds client = catClientInfoString(sdsempty(),c);
             serverLog(LL_NOTICE,"MASTER MODE enabled (user request from '%s')",
                 client);
@@ -4218,7 +4249,7 @@ void replicaReplayCommand(client *c)
     serverTL->current_client = current_clientSave;
 
     // call() will not propogate this for us, so we do so here
-    if (!s_pstate->FCancelled() && s_pstate->FFirst())
+    if (!s_pstate->FCancelled() && s_pstate->FFirst() && !cserver.multimaster_no_forward)
         alsoPropagate(cserver.rreplayCommand,c->db->id,c->argv,c->argc,PROPAGATE_AOF|PROPAGATE_REPL);
     
     s_pstate->Pop();

src/server.cpp

@@ -247,6 +247,10 @@ struct redisCommand redisCommandTable[] = {
      "read-only fast @keyspace",
      0,NULL,1,-1,1,0,0,0},
 
+    {"keydb.mexists",mexistsCommand,-2,
+     "read-only fast @keyspace",
+     0,NULL,1,-1,1,0,0,0},
+
     {"setbit",setbitCommand,4,
      "write use-memory @bitmap",
      0,NULL,1,1,1,0,0,0},
@@ -768,7 +772,7 @@ struct redisCommand redisCommandTable[] = {
      "admin no-script ok-stale",
      0,NULL,0,0,0,0,0,0},
 
-    {"replicaof",replicaofCommand,3,
+    {"replicaof",replicaofCommand,-3,
      "admin no-script ok-stale",
      0,NULL,0,0,0,0,0,0},
 
@@ -809,11 +813,11 @@ struct redisCommand redisCommandTable[] = {
      0,NULL,0,0,0,0,0,0},
 
     {"watch",watchCommand,-2,
-     "no-script fast @transaction",
+     "no-script fast ok-loading ok-stale @transaction",
      0,NULL,1,-1,1,0,0,0},
 
     {"unwatch",unwatchCommand,1,
-     "no-script fast @transaction",
+     "no-script fast ok-loading ok-stale @transaction",
      0,NULL,0,0,0,0,0,0},
 
     {"cluster",clusterCommand,-2,
@@ -3866,6 +3870,8 @@ int processCommand(client *c, int callFlags, AeLocker &locker) {
           c->cmd->proc != multiCommand &&
           c->cmd->proc != execCommand &&
           c->cmd->proc != discardCommand &&
+          c->cmd->proc != watchCommand &&
+          c->cmd->proc != unwatchCommand &&
         !(c->cmd->proc == shutdownCommand &&
           c->argc == 2 &&
           tolower(((char*)ptrFromObj(c->argv[1]))[0]) == 'n') &&

src/server.h

@@ -166,22 +166,24 @@ class robj_sharedptr
 
 public:
     robj_sharedptr()
-        : m_ptr(nullptr)
-        {}
-    robj_sharedptr(redisObject *ptr)
-        : m_ptr(ptr)
-        {
+    : m_ptr(nullptr)
+    {}
+    explicit robj_sharedptr(redisObject *ptr)
+    : m_ptr(ptr)
+    {
+        if(m_ptr)
             incrRefCount(ptr);
-        }
+    }
     ~robj_sharedptr()
     {
         if (m_ptr)
             decrRefCount(m_ptr);
     }
     robj_sharedptr(const robj_sharedptr& other)
-    {
-        m_ptr = other.m_ptr;
-        incrRefCount(m_ptr);
+    : m_ptr(other.m_ptr)
+    {        
+        if(m_ptr)
+            incrRefCount(m_ptr);
     }
 
     robj_sharedptr(robj_sharedptr&& other)
@@ -192,41 +194,19 @@ public:
 
     robj_sharedptr &operator=(const robj_sharedptr& other)
     {
-        if (m_ptr)
-            decrRefCount(m_ptr);
-        m_ptr = other.m_ptr;
-        incrRefCount(m_ptr);
+        robj_sharedptr tmp(other);
+        using std::swap;
+        swap(m_ptr, tmp.m_ptr);
         return *this;
     }
     robj_sharedptr &operator=(redisObject *ptr)
     {
-        if (m_ptr)
-            decrRefCount(m_ptr);
-        m_ptr = ptr;
-        incrRefCount(m_ptr);
+        robj_sharedptr tmp(ptr);
+        using std::swap;
+        swap(m_ptr, tmp.m_ptr);
         return *this;
     }
-
-    bool operator==(const robj_sharedptr &other) const
-    {
-        return m_ptr == other.m_ptr;
-    }
-
-    bool operator==(const void *p) const
-    {
-        return m_ptr == p;
-    }
-
-    bool operator!=(const robj_sharedptr &other) const
-    {
-        return m_ptr != other.m_ptr;
-    }
-
-    bool operator!=(const void *p) const
-    {
-        return m_ptr != p;
-    }
-
+    
     redisObject* operator->() const
     {
         return m_ptr;
@@ -237,7 +217,7 @@ public:
         return !m_ptr;
     }
 
-    operator bool() const{
+    explicit operator bool() const{
         return !!m_ptr;
     }
 
@@ -247,8 +227,39 @@ public:
     }
 
     redisObject *get() { return m_ptr; }
+    const redisObject *get() const { return m_ptr; }
 };
 
+inline bool operator==(const robj_sharedptr &lhs, const robj_sharedptr &rhs)
+{
+    return lhs.get() == rhs.get();
+}
+
+inline bool operator!=(const robj_sharedptr &lhs, const robj_sharedptr &rhs)
+{
+    return !(lhs == rhs);
+}
+
+inline bool operator==(const robj_sharedptr &lhs, const void *p)
+{
+    return lhs.get() == p;
+}
+
+inline bool operator==(const void *p, const robj_sharedptr &rhs)
+{
+    return rhs == p;
+}
+
+inline bool operator!=(const robj_sharedptr &lhs, const void *p)
+{
+    return !(lhs == p);
+}
+
+inline bool operator!=(const void *p, const robj_sharedptr &rhs)
+{
+    return !(rhs == p);
+}
+
 /* Error codes */
 #define C_OK                    0
 #define C_ERR                   -1
@@ -794,7 +805,9 @@ typedef struct redisObject {
 private:
     mutable std::atomic<unsigned> refcount {0};
 public:
+#ifdef ENABLE_MVCC
     uint64_t mvcc_tstamp;
+#endif
     void *m_ptr;
 
     inline bool FExpires() const { return refcount.load(std::memory_order_relaxed) >> 31; }
@@ -805,7 +818,11 @@ public:
     void addref() const { refcount.fetch_add(1, std::memory_order_relaxed); }
     unsigned release() const { return refcount.fetch_sub(1, std::memory_order_seq_cst) & ~(1U << 31); }
 } robj;
+#ifdef ENABLE_MVCC
 static_assert(sizeof(redisObject) == 24, "object size is critical, don't increase");
+#else
+static_assert(sizeof(redisObject) == 16, "object size is critical, don't increase");
+#endif
 
 __attribute__((always_inline)) inline const void *ptrFromObj(robj_roptr &o)
 {
@@ -1637,6 +1654,7 @@ struct redisServerConst {
     unsigned char uuid[UUID_BINARY_LEN];         /* This server's UUID - populated on boot */
     bool fUsePro = false;
     int thread_min_client_threshold = 50;
+    int multimaster_no_forward;
 };
 
 struct redisServer {
@@ -2810,6 +2828,7 @@ void getCommand(client *c);
 void delCommand(client *c);
 void unlinkCommand(client *c);
 void existsCommand(client *c);
+void mexistsCommand(client *c);
 void setbitCommand(client *c);
 void getbitCommand(client *c);
 void bitfieldCommand(client *c);

src/storage.h

@@ -1,8 +1,6 @@
 #ifndef __STORAGE_H__
 #define __STORAGE_H__
 
-#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 48   // Note: also defined in object.c - should always match
-
 #ifdef __cplusplus
 extern "C" {
 #endif

src/tls.cpp

@@ -226,7 +226,7 @@ int tlsConfigure(redisTLSContextConfig *ctx_config) {
     SSL_CTX_set_ecdh_auto(ctx, 1);
 #endif
 
-    if (SSL_CTX_use_certificate_file(ctx, ctx_config->cert_file, SSL_FILETYPE_PEM) <= 0) {
+    if (SSL_CTX_use_certificate_chain_file(ctx, ctx_config->cert_file) <= 0) {
         ERR_error_string_n(ERR_get_error(), errbuf, sizeof(errbuf));
         serverLog(LL_WARNING, "Failed to load certificate: %s: %s", ctx_config->cert_file, errbuf);
         goto error;

src/tracking.cpp

@@ -101,8 +101,8 @@ void disableTracking(client *c) {
 
 /* Set the client 'c' to track the prefix 'prefix'. If the client 'c' is
  * already registered for the specified prefix, no operation is performed. */
-void enableBcastTrackingForPrefix(client *c, const char *prefix, size_t plen) {
-    bcastState *bs = (bcastState*)raxFind(PrefixTable,(unsigned char*)prefix,sdslen(prefix));
+static void enableBcastTrackingForPrefix(client *c, const char *prefix, size_t plen) {
+    bcastState *bs = (bcastState*)raxFind(PrefixTable,(unsigned char*)prefix,plen);
     /* If this is the first client subscribing to such prefix, create
      * the prefix in the table. */
     if (bs == raxNotFound) {

tests/integration/psync2-pingoff.tcl

@@ -64,6 +64,7 @@ start_server {} {
         # make sure replication is still alive and kicking
         $R(1) incr x
         wait_for_condition 50 1000 {
+            [status $R(0) loading] == 0 &&
             [$R(0) get x] == 1
         } else {
             fail "replica didn't get incr"

tests/integration/replication-active.tcl

@@ -237,3 +237,18 @@ start_server {tags {"active-repl"} overrides {active-replica yes}} {
     }
 }
 }
+
+start_server {tags {"active-repl"} overrides {active-replica yes}} {
+    set slave [srv 0 client]
+    set slave_host [srv 0 host]
+    set slave_port [srv 0 port]
+    start_server {tags {"active-repl"} overrides { active-replica yes}} {
+        r set testkeyB bar
+        test {Active Replica Merges Database On Sync} {
+            $slave set testkeyA foo
+            r replicaof $slave_host $slave_port
+	    after 1000
+	    assert_equal 2 [r dbsize]
+	}
+    }
+}

tests/integration/replication-multimaster.tcl

@@ -1,4 +1,6 @@
 foreach topology {mesh ring} {
+
+foreach noforward [expr {[string equal $topology "mesh"] ? {no yes} : {no}}] {
 start_server {tags {"multi-master"} overrides {hz 500 active-replica yes multi-master yes}} {
 start_server {overrides {hz 500 active-replica yes multi-master yes}} {
 start_server {overrides {hz 500 active-replica yes multi-master yes}} {
@@ -8,8 +10,12 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
         set R($j) [srv [expr 0-$j] client]
         set R_host($j) [srv [expr 0-$j] host]
         set R_port($j) [srv [expr 0-$j] port]
+
+	$R($j) config set multi-master-no-forward $noforward
     }
 
+    set topology_name "$topology[expr {[string equal $noforward "yes"] ? " no-forward" : ""}]"
+
     # Initialize as mesh
     if [string equal $topology "mesh"] {
     for {set j 0} {$j < 4} {incr j} {
@@ -31,7 +37,7 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
         $R(3) replicaof $R_host(2) $R_port(2)
     }
 
-    test "$topology all nodes up" {
+    test "$topology_name all nodes up" {
         for {set j 0} {$j < 4} {incr j} {
             wait_for_condition 50 100 {
                 [string match {*master_global_link_status:up*} [$R($j) info replication]]
@@ -41,7 +47,7 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
         }
     }
 
-    test "$topology replicates to all nodes" {
+    test "$topology_name replicates to all nodes" {
         $R(0) set testkey foo
         after 500
         for {set n 0} {$n < 4} {incr n} {
@@ -53,7 +59,7 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
         }
     }
 
-    test "$topology replicates only once" {
+    test "$topology_name replicates only once" {
         $R(0) set testkey 1
         after 500
         #wait_for_condition 50 100 {
@@ -74,7 +80,7 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
         }
     }
 
-    test "$topology transaction replicates only once" {
+    test "$topology_name transaction replicates only once" {
         for {set j 0} {$j < 1000} {incr j} {
             $R(0) set testkey 1
             $R(0) multi
@@ -95,3 +101,4 @@ start_server {overrides {hz 500 active-replica yes multi-master yes}} {
 }
 }
 }
+}

tests/modules/hooks.c

@@ -30,6 +30,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
+#define REDISMODULE_EXPERIMENTAL_API 1
 #include "redismodule.h"
 #include <stdio.h>
 #include <string.h>
@@ -143,10 +144,12 @@ void flushdbCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void
 {
     REDISMODULE_NOT_USED(e);
 
+    RedisModule_ThreadSafeContextLock(ctx);
     RedisModuleFlushInfo *fi = data;
     char *keyname = (sub == REDISMODULE_SUBEVENT_FLUSHDB_START) ?
         "flush-start" : "flush-end";
     LogNumericEvent(ctx, keyname, fi->dbnum);
+    RedisModule_ThreadSafeContextUnlock(ctx);
 }
 
 void roleChangeCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data)

tests/unit/expire.tcl

@@ -313,4 +313,14 @@ start_server {tags {"expire"}} {
         after 3000
         assert_equal [r dbsize] 0
     }
+
+    test {SET - use KEEPTTL option, TTL should not be removed after loadaof} {
+        r config set appendonly yes
+        r set foo bar EX 100
+        r set foo bar2 KEEPTTL
+        after 2000
+        r debug loadaof
+        set ttl [r ttl foo]
+        assert {$ttl <= 98 && $ttl > 90}
+    }
 }

tests/unit/multi.tcl

@@ -363,6 +363,9 @@ start_server {tags {"multi"}} {
         set xx [r get xx]
         # make sure that either the whole transcation passed or none of it (we actually expect none)
         assert { $xx == 1 || $xx == 3}
+        # Discard the transaction since EXEC likely got -BUSY error
+        # so the client is still in MULTI state.
+        catch { $rd2 discard ;$rd2 read } e
         # check that the connection is no longer in multi state
         $rd2 ping asdf
         set pong [$rd2 read]