fix test

* ship packages with flash option

* deb12 fix

Co-authored-by: benschermel <bschermel@snapchat.com>

.github/workflows/ci.yml

@@ -21,10 +21,10 @@ jobs:
     - name: test-tls
       run: |
         sudo apt-get -y install tcl tcl-tls
-        ./runtest --clients 2 --verbose --tls
+        ./runtest --clients 1 --verbose --tls --config server-threads 3
     - name: cluster-test
       run: |
-        ./runtest-cluster --tls
+        ./runtest-cluster --tls --config server-threads 3
     - name: sentinel test
       run: |
           ./runtest-sentinel
@@ -69,15 +69,3 @@ jobs:
           sudo apt-get -y remove libzstd || true
           sudo apt-get -y install uuid-dev libcurl4-openssl-dev libbz2-dev zlib1g-dev libsnappy-dev liblz4-dev libzstd-dev libgflags-dev
           make KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror' MALLOC=libc -j2
-  build-ubuntu-32bit:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: actions/checkout@v2
-      with:
-        submodules: recursive
-    - name: make
-      run: |
-          sudo dpkg --add-architecture i386
-          sudo apt-get update
-          sudo apt-get -y install gcc-multilib g++-multilib libc6-dev-i386 lib32z1 uuid-dev:i386 libcurl4-openssl-dev:i386
-          make KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror' 32bit -j2

README.md

@@ -104,6 +104,12 @@ Avoid forwarding RREPLAY messages to other masters? WARNING: This setting is dan
 If you would like KeyDB to dump and load directly to AWS S3 this option specifies the bucket.  Using this option with the traditional RDB options will result in KeyDB backing up twice to both locations.  If both are specified KeyDB will first attempt to load from the local dump file and if that fails load from S3.  This requires the AWS CLI tools to be installed and configured which are used under the hood to transfer the data.
 
 
+```
+storage-provider flash /path/to/flash
+```
+If you would like to use KeyDB FLASH storage, specify the storage medium followed by the directory path on your local SSD volume. Note that this feature is still considered experimental and should be used with discretion. See [FLASH Documentation](https://docs.keydb.dev/docs/flash) for more details on configuration and setting up your FLASH volume. 
+
+
 Building KeyDB
 --------------
 
@@ -145,6 +151,12 @@ installed):
     % ./utils/gen-test-certs.sh
     % ./runtest --tls
 
+To build with KeyDB FLASH support, use:
+
+    % make ENABLE_FLASH=yes
+
+***Note that the KeyDB FLASH feature is considered experimental (beta) and should used with discretion
+
 
 Fixing build problems with dependencies or cached build options
 ---------

keydb.conf

@@ -2073,4 +2073,18 @@ replica-weighting-factor 2
 #
 # By default this is enabled
 #
-active-client-balancing yes
+active-client-balancing yes
+
+# Enable FLASH support (Experimental Feature)
+# storage-provider flash /path/to/flash/db
+
+# Blob support is a way to store very large objects (>200MB) on disk
+# The files are automatically cleaned up when KeyDB exits and are only
+# for temporary use.  This helps reduce memory pressure for very large
+# data items at the cost of some performance.
+#
+# By default this config is disable.  When enabled the disk associated
+# with KeyDB's working directory will be used.  If there is insufficient
+# disk space or any other I/O error KeyDB will instead use memory.
+#
+# blob-support false

pkg/deb/conf/keydb.conf

@@ -2063,3 +2063,17 @@ server-threads 2
 #
 # By default KeyDB sets this to 2.
 replica-weighting-factor 2
+
+# Enable FLASH support (Experimental Feature)
+# storage-provider flash /path/to/flash/db
+
+# Blob support is a way to store very large objects (>200MB) on disk
+# The files are automatically cleaned up when KeyDB exits and are only
+# for temporary use.  This helps reduce memory pressure for very large
+# data items at the cost of some performance.
+# 
+# By default this config is disable.  When enabled the disk associated
+# with KeyDB's working directory will be used.  If there is insufficient
+# disk space or any other I/O error KeyDB will instead use memory.
+# 
+# blob-support false

pkg/deb/deb-buildsource.sh

@@ -19,6 +19,11 @@ elif [ "$distributor" == "Ubuntu" ]; then
 fi
 codename=$(lsb_release --codename --short)
 date=$(date +%a," "%d" "%b" "%Y" "%T)
+
+# overwrite debian bookworm version until updated
+if [ $codename == "bookworm" ]; then
+    distname=+deb12u1
+fi
 pkg_name=keydb-$majorv:$version$distname
 
 # create build tree

pkg/deb/debian/rules

@@ -4,6 +4,7 @@ include /usr/share/dpkg/buildflags.mk
 
 export BUILD_TLS=yes
 export USE_SYSTEMD=yes
+export ENABLE_FLASH=yes
 export CFLAGS CPPFLAGS LDFLAGS
 export DEB_BUILD_MAINT_OPTIONS = hardening=+all
 export DEB_LDFLAGS_MAINT_APPEND = -ldl -latomic $(LUA_LDFLAGS)

pkg/deb/debian_dh9/rules

@@ -10,6 +10,7 @@ include /usr/share/dpkg/buildflags.mk
 
 export BUILD_TLS=yes
 export USE_SYSTEMD=yes
+export ENABLE_FLASH=yes
 export CFLAGS CPPFLAGS LDFLAGS
 export DEB_BUILD_MAINT_OPTIONS = hardening=+all
 export DEB_LDFLAGS_MAINT_APPEND = -ldl -latomic $(LUA_LDFLAGS)

pkg/docker/Dockerfile_Alpine

@@ -21,6 +21,10 @@ RUN set -eux; \
                 bash \
                 perl \
                 git \
+                bzip2-dev \
+                zstd-dev \
+                lz4-dev \
+                snappy-dev \
         ; \
         cd /tmp && git clone --branch $BRANCH https://github.com/Snapchat/KeyDB.git --recursive; \
         cd /tmp/KeyDB; \

pkg/rpm/keydb_build/keydb_rpm/etc/keydb/keydb.conf

@@ -2063,3 +2063,17 @@ server-threads 2
 #
 # By default KeyDB sets this to 2.
 replica-weighting-factor 2
+
+# Enable FLASH support (Experimental Feature)
+# storage-provider flash /path/to/flash/db
+
+# Blob support is a way to store very large objects (>200MB) on disk
+# The files are automatically cleaned up when KeyDB exits and are only
+# for temporary use.  This helps reduce memory pressure for very large
+# data items at the cost of some performance.
+# 
+# By default this config is disable.  When enabled the disk associated
+# with KeyDB's working directory will be used.  If there is insufficient
+# disk space or any other I/O error KeyDB will instead use memory.
+# 
+# blob-support false

runtest-moduleapi

@@ -37,5 +37,6 @@ $TCLSH tests/test_helper.tcl \
 --single unit/moduleapi/hash \
 --single unit/moduleapi/zset \
 --single unit/moduleapi/stream \
+--single unit/moduleapi/load \
 --config server-threads 3 \
 "${@}"

src/config.cpp

@@ -2587,7 +2587,7 @@ static int updateMaxclients(long long val, long long prev, const char **err) {
         adjustOpenFilesLimit();
         if (g_pserver->maxclients != val) {
             static char msg[128];
-            sprintf(msg, "The operating system is not able to handle the specified number of clients, try with %d", g_pserver->maxclients);
+            snprintf(msg, 128, "The operating system is not able to handle the specified number of clients, try with %d", g_pserver->maxclients);
             *err = msg;
             if (g_pserver->maxclients > prev) {
                 g_pserver->maxclients = prev;
@@ -2626,7 +2626,7 @@ static int updateMaxclients(long long val, long long prev, const char **err) {
 
                 if (res != AE_OK){
                     static char msg[128];
-                    sprintf(msg, "Failed to post the request to change setsize for Thread %d", iel);
+                    snprintf(msg, 128,"Failed to post the request to change setsize for Thread %d", iel);
                     *err = msg;
                     return 0;
                 }

src/db.cpp

@@ -2607,6 +2607,14 @@ void clusterStorageLoadCallback(const char *rgchkey, size_t cch, void *)
     slotToKeyUpdateKeyCore(rgchkey, cch, true /*add*/);
 }
 
+void moduleLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
+    if (g_pserver->cluster_enabled) {
+        clusterStorageLoadCallback(rgchKey, cchKey, data);
+    }
+    robj *keyobj = createEmbeddedStringObject(rgchKey, cchKey);
+    moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", keyobj, *(int *)data);
+}
+
 void redisDb::initialize(int id)
 {
     redisDbPersistentData::initialize();
@@ -2625,8 +2633,8 @@ void redisDb::storageProviderInitialize()
 {
     if (g_pserver->m_pstorageFactory != nullptr)
     {
-        IStorageFactory::key_load_iterator itr = (g_pserver->cluster_enabled) ? clusterStorageLoadCallback : nullptr;
-        this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, nullptr));
+        IStorageFactory::key_load_iterator itr = moduleLoadCallback;
+        this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, &id));
     }
 }
 

src/dict.cpp

@@ -1577,7 +1577,7 @@ char *stringFromLongLong(long long value) {
     int len;
     char *s;
 
-    len = sprintf(buf,"%lld",value);
+    len = snprintf(buf,32,"%lld",value);
     s = zmalloc(len+1);
     memcpy(s, buf, len);
     s[len] = '\0';

src/endianconv.c

@@ -112,15 +112,15 @@ int endianconvTest(int argc, char *argv[], int accurate) {
     UNUSED(argv);
     UNUSED(accurate);
 
-    sprintf(buf,"ciaoroma");
+    snprintf(buf,32,"ciaoroma");
     memrev16(buf);
     printf("%s\n", buf);
 
-    sprintf(buf,"ciaoroma");
+    snprintf(buf,32,"ciaoroma");
     memrev32(buf);
     printf("%s\n", buf);
 
-    sprintf(buf,"ciaoroma");
+    snprintf(buf,32,"ciaoroma");
     memrev64(buf);
     printf("%s\n", buf);
 

src/keydb-diagnostic-tool.cpp

@@ -904,7 +904,7 @@ int main(int argc, const char **argv) {
 
     while (self_threads < config.max_threads) {
         for (int i = 0; i < config.numclients; i++) {
-            sprintf(command, "SET %d %s\r\n", self_threads * config.numclients + i, set_value);
+            snprintf(command, 63, "SET %d %s\r\n", self_threads * config.numclients + i, set_value);
             createClient(command, strlen(command), NULL,self_threads);
         }
 

src/networking.cpp

@@ -145,7 +145,7 @@ client *createClient(connection *conn, int iel) {
     client_id = g_pserver->next_client_id.fetch_add(1);
     c->iel = iel;
     c->id = client_id;
-    sprintf(c->lock.szName, "client %" PRIu64, client_id);
+    snprintf(c->lock.szName, 56, "client %" PRIu64, client_id);
     c->resp = 2;
     c->conn = conn;
     c->name = NULL;
@@ -777,11 +777,11 @@ void setDeferredAggregateLen(client *c, void *node, long length, char prefix) {
          * we return NULL in addReplyDeferredLen() */
         if (node == NULL) return;
         char lenstr[128];
-        size_t lenstr_len = sprintf(lenstr, "%c%ld\r\n", prefix, length);
+        size_t lenstr_len = snprintf(lenstr, 128, "%c%ld\r\n", prefix, length);
         setDeferredReply(c, node, lenstr, lenstr_len);
     } else {
         char lenstr[128];
-        int lenstr_len = sprintf(lenstr, "%c%ld\r\n", prefix, length);
+        int lenstr_len = snprintf(lenstr, 128, "%c%ld\r\n", prefix, length);
 
         size_t idxSplice = (size_t)node;
         serverAssert(idxSplice <= c->replyAsync->used);

src/redis-cli.c

@@ -6835,17 +6835,17 @@ void bytesToHuman(char *s, long long n) {
     }
     if (n < 1024) {
         /* Bytes */
-        sprintf(s,"%lldB",n);
+        snprintf(s,256,"%lldB",n);
         return;
     } else if (n < (1024*1024)) {
         d = (double)n/(1024);
-        sprintf(s,"%.2fK",d);
+        snprintf(s,256,"%.2fK",d);
     } else if (n < (1024LL*1024*1024)) {
         d = (double)n/(1024*1024);
-        sprintf(s,"%.2fM",d);
+        snprintf(s,256,"%.2fM",d);
     } else if (n < (1024LL*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024);
-        sprintf(s,"%.2fG",d);
+        snprintf(s,256,"%.2fG",d);
     }
 }
 
@@ -6875,12 +6875,12 @@ static void statMode(void) {
         for (j = 0; j < 20; j++) {
             long k;
 
-            sprintf(buf,"db%d:keys",j);
+            snprintf(buf,64,"db%d:keys",j);
             k = getLongInfoField(reply->str,buf);
             if (k == LONG_MIN) continue;
             aux += k;
         }
-        sprintf(buf,"%ld",aux);
+        snprintf(buf,64,"%ld",aux);
         printf("%-11s",buf);
 
         /* Used memory */
@@ -6890,23 +6890,23 @@ static void statMode(void) {
 
         /* Clients */
         aux = getLongInfoField(reply->str,"connected_clients");
-        sprintf(buf,"%ld",aux);
+        snprintf(buf,64,"%ld",aux);
         printf(" %-8s",buf);
 
         /* Blocked (BLPOPPING) Clients */
         aux = getLongInfoField(reply->str,"blocked_clients");
-        sprintf(buf,"%ld",aux);
+        snprintf(buf,64,"%ld",aux);
         printf("%-8s",buf);
 
         /* Requests */
         aux = getLongInfoField(reply->str,"total_commands_processed");
-        sprintf(buf,"%ld (+%ld)",aux,requests == 0 ? 0 : aux-requests);
+        snprintf(buf,64,"%ld (+%ld)",aux,requests == 0 ? 0 : aux-requests);
         printf("%-19s",buf);
         requests = aux;
 
         /* Connections */
         aux = getLongInfoField(reply->str,"total_connections_received");
-        sprintf(buf,"%ld",aux);
+        snprintf(buf,64,"%ld",aux);
         printf(" %-12s",buf);
 
         /* Children */

src/redismodule.h

@@ -326,6 +326,7 @@ static const RedisModuleEvent
 #define REDISMODULE_SUBEVENT_LOADING_ENDED 3
 #define REDISMODULE_SUBEVENT_LOADING_FAILED 4
 #define _REDISMODULE_SUBEVENT_LOADING_NEXT 5
+#define REDISMODULE_SUBEVENT_LOADING_FLASH_START 6
 
 #define REDISMODULE_SUBEVENT_CLIENT_CHANGE_CONNECTED 0
 #define REDISMODULE_SUBEVENT_CLIENT_CHANGE_DISCONNECTED 1

src/replication.cpp

@@ -1142,7 +1142,7 @@ public:
     }
 };
 
-int rdbSaveSnapshotForReplication(struct rdbSaveInfo *rsi) {
+int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
     // TODO: This needs to be on a background thread
     int retval = C_OK;
     serverAssert(GlobalLocksAcquired());

src/server.cpp

@@ -4067,12 +4067,6 @@ void initServer(void) {
         }
     }
 
-    /* We have to initialize storage providers after the cluster has been initialized */
-    for (int idb = 0; idb < cserver.dbnum; ++idb)
-    {
-        g_pserver->db[idb]->storageProviderInitialize();
-    }
-
     saveMasterStatusToStorage(false); // eliminate the repl-offset field
     
     /* Initialize ACL default password if it exists */
@@ -4085,6 +4079,15 @@ void initServer(void) {
  * Thread Local Storage initialization collides with dlopen call.
  * see: https://sourceware.org/bugzilla/show_bug.cgi?id=19329 */
 void InitServerLast() {
+
+    /* We have to initialize storage providers after the cluster has been initialized */
+    moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_FLASH_START, NULL);
+    for (int idb = 0; idb < cserver.dbnum; ++idb)
+    {
+        g_pserver->db[idb]->storageProviderInitialize();
+    }
+    moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_ENDED, NULL);
+
     bioInit();
     set_jemalloc_bg_thread(cserver.jemalloc_bg_thread);
     g_pserver->initial_memory_usage = zmalloc_used_memory();
@@ -5472,25 +5475,25 @@ void bytesToHuman(char *s, unsigned long long n) {
 
     if (n < 1024) {
         /* Bytes */
-        sprintf(s,"%lluB",n);
+        snprintf(s,256,"%lluB",n);
     } else if (n < (1024*1024)) {
         d = (double)n/(1024);
-        sprintf(s,"%.2fK",d);
+        snprintf(s,256,"%.2fK",d);
     } else if (n < (1024LL*1024*1024)) {
         d = (double)n/(1024*1024);
-        sprintf(s,"%.2fM",d);
+        snprintf(s,256,"%.2fM",d);
     } else if (n < (1024LL*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024);
-        sprintf(s,"%.2fG",d);
+        snprintf(s,256,"%.2fG",d);
     } else if (n < (1024LL*1024*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024*1024);
-        sprintf(s,"%.2fT",d);
+        snprintf(s,256,"%.2fT",d);
     } else if (n < (1024LL*1024*1024*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024*1024*1024);
-        sprintf(s,"%.2fP",d);
+        snprintf(s,256,"%.2fP",d);
     } else {
         /* Let's hope we never need this */
-        sprintf(s,"%lluB",n);
+        snprintf(s,256,"%lluB",n);
     }
 }
 

src/server.h

@@ -1909,6 +1909,12 @@ struct MasterSaveInfo {
             masterhost = sdsstring(sdsdup(mi.masterhost));
         masterport = mi.masterport;
     }
+    MasterSaveInfo(const MasterSaveInfo &other) {
+        masterhost = other.masterhost;
+        masterport = other.masterport;
+        memcpy(master_replid, other.master_replid, sizeof(master_replid));
+        master_initial_offset = other.master_initial_offset;
+    }
 
     MasterSaveInfo &operator=(const MasterSaveInfo &other) {
         masterhost = other.masterhost;

src/t_nhash.cpp

@@ -263,7 +263,7 @@ sds writeJsonValue(sds output, const char *valIn, size_t cchIn) {
                     serverAssert(!FSimpleJsonEscapeCh(valIn[ich]));
                     if (FExtendedJsonEscapeCh(valIn[ich])) {
                         dst[ichDst++] = '\\'; dst[ichDst++] = 'u';
-                        sprintf(dst + ichDst, "%4x", valIn[ich]);
+                        snprintf(dst + ichDst, cchIn+cchEscapeExtra-ichDst, "%4x", valIn[ich]);
                         ichDst += 4;
                     } else {
                         dst[ichDst++] = valIn[ich];

src/ziplist.c

@@ -1704,17 +1704,17 @@ static unsigned char *createIntList() {
     unsigned char *zl = ziplistNew();
     char buf[32];
 
-    sprintf(buf, "100");
+    snprintf(buf, 32, "100");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
-    sprintf(buf, "128000");
+    snprintf(buf, 32, "128000");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
-    sprintf(buf, "-100");
+    snprintf(buf, 32, "-100");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_HEAD);
-    sprintf(buf, "4294967296");
+    snprintf(buf, 32, "4294967296");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_HEAD);
-    sprintf(buf, "non integer");
+    snprintf(buf, 32, "non integer");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
-    sprintf(buf, "much much longer non integer");
+    snprintf(buf, 32, "much much longer non integer");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
     return zl;
 }
@@ -2228,7 +2228,7 @@ int ziplistTest(int argc, char **argv, int accurate) {
         char buf[32];
         int i,len;
         for (i = 0; i < 1000; i++) {
-            len = sprintf(buf,"%d",i);
+            len = snprintf(buf,32,"%d",i);
             zl = ziplistPush(zl,(unsigned char*)buf,len,ZIPLIST_TAIL);
         }
         for (i = 0; i < 1000; i++) {
@@ -2375,13 +2375,13 @@ int ziplistTest(int argc, char **argv, int accurate) {
                 } else {
                     switch(rand() % 3) {
                     case 0:
-                        buflen = sprintf(buf,"%lld",(0LL + rand()) >> 20);
+                        buflen = snprintf(buf,1024,"%lld",(0LL + rand()) >> 20);
                         break;
                     case 1:
-                        buflen = sprintf(buf,"%lld",(0LL + rand()));
+                        buflen = snprintf(buf,1024,"%lld",(0LL + rand()));
                         break;
                     case 2:
-                        buflen = sprintf(buf,"%lld",(0LL + rand()) << 20);
+                        buflen = snprintf(buf,1024,"%lld",(0LL + rand()) << 20);
                         break;
                     default:
                         assert(NULL);
@@ -2410,7 +2410,7 @@ int ziplistTest(int argc, char **argv, int accurate) {
 
                 assert(ziplistGet(p,&sstr,&slen,&sval));
                 if (sstr == NULL) {
-                    buflen = sprintf(buf,"%lld",sval);
+                    buflen = snprintf(buf,1024,"%lld",sval);
                 } else {
                     buflen = slen;
                     memcpy(buf,sstr,buflen);

src/zmalloc.cpp

@@ -521,7 +521,7 @@ int jemalloc_purge() {
     unsigned narenas = 0;
     size_t sz = sizeof(unsigned);
     if (!je_mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
-        sprintf(tmp, "arena.%d.purge", narenas);
+        snprintf(tmp, 32, "arena.%d.purge", narenas);
         if (!je_mallctl(tmp, NULL, 0, NULL, 0))
             return 0;
     }

tests/modules/Makefile

@@ -39,7 +39,8 @@ TEST_MODULES = \
     defragtest.so \
     hash.so \
     zset.so \
-    stream.so
+    stream.so \
+    load.so
 
 
 .PHONY: all

tests/modules/hooks.c

@@ -212,6 +212,7 @@ void loadingCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void
         case REDISMODULE_SUBEVENT_LOADING_RDB_START: keyname = "loading-rdb-start"; break;
         case REDISMODULE_SUBEVENT_LOADING_AOF_START: keyname = "loading-aof-start"; break;
         case REDISMODULE_SUBEVENT_LOADING_REPL_START: keyname = "loading-repl-start"; break;
+        case REDISMODULE_SUBEVENT_LOADING_FLASH_START: keyname = "loading-flash-start"; break;
         case REDISMODULE_SUBEVENT_LOADING_ENDED: keyname = "loading-end"; break;
         case REDISMODULE_SUBEVENT_LOADING_FAILED: keyname = "loading-failed"; break;
     }

tests/modules/load.c

@@ -0,0 +1,94 @@
+/* Server hooks API example
+ *
+ * -----------------------------------------------------------------------------
+ *
+ * Copyright (c) 2019, Salvatore Sanfilippo <antirez at gmail dot com>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#define REDISMODULE_EXPERIMENTAL_API
+#include "redismodule.h"
+
+size_t count, finalCount;
+
+/* Client state change callback. */
+void loadCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data) {
+    REDISMODULE_NOT_USED(ctx);
+    REDISMODULE_NOT_USED(e);
+    REDISMODULE_NOT_USED(data);
+
+    if (sub == REDISMODULE_SUBEVENT_LOADING_FLASH_START || sub == REDISMODULE_SUBEVENT_LOADING_RDB_START || sub == REDISMODULE_SUBEVENT_LOADING_AOF_START || sub == REDISMODULE_SUBEVENT_LOADING_REPL_START) {
+        count = 0;
+        finalCount = 0;
+    } else if (sub == REDISMODULE_SUBEVENT_LOADING_ENDED) {
+        finalCount = count;
+    }
+}
+
+int loadKeyCallback(RedisModuleCtx *ctx, int type,  const char *event, RedisModuleString *key) {
+    REDISMODULE_NOT_USED(ctx);
+    REDISMODULE_NOT_USED(type);
+    REDISMODULE_NOT_USED(event);
+
+    const char *keyname = RedisModule_StringPtrLen(key, NULL);
+
+    RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_NOTICE, "Loaded key: %s", keyname);
+
+    count++;
+    return 0;
+}
+
+int LoadCount_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    REDISMODULE_NOT_USED(argv);
+    RedisModule_AutoMemory(ctx); /* Use automatic memory management. */
+
+    if (argc != 1) return RedisModule_WrongArity(ctx);
+
+    RedisModule_ReplyWithLongLong(ctx, finalCount);
+
+    return REDISMODULE_OK;
+}
+
+/* This function must be present on each Redis module. It is used in order to
+ * register the commands into the Redis server. */
+int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    REDISMODULE_NOT_USED(argv);
+    REDISMODULE_NOT_USED(argc);
+
+    if (RedisModule_Init(ctx,"load",1,REDISMODULE_APIVER_1)
+        == REDISMODULE_ERR) return REDISMODULE_ERR;
+
+    RedisModule_SubscribeToServerEvent(ctx,
+        RedisModuleEvent_Loading, loadCallback);
+    RedisModule_SubscribeToKeyspaceEvents(ctx,
+        REDISMODULE_NOTIFY_LOADED, loadKeyCallback);
+
+    if (RedisModule_CreateCommand(ctx, "load.count",
+        LoadCount_RedisCommand,"readonly",1,1,1) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+    return REDISMODULE_OK;
+}

tests/support/keydb.tcl

@@ -66,6 +66,33 @@ proc redis {{server 127.0.0.1} {port 6379} {defer 0} {tls 0} {tlsoptions {}} {re
     interp alias {} ::redis::redisHandle$id {} ::redis::__dispatch__ $id
 }
 
+# On recent versions of tcl-tls/OpenSSL, reading from a dropped connection
+# results with an error we need to catch and mimic the old behavior.
+proc ::redis::redis_safe_read {fd len} {
+    if {$len == -1} {
+        set err [catch {set val [read $fd]} msg]
+    } else {
+        set err [catch {set val [read $fd $len]} msg]
+    }
+    if {!$err} {
+        return $val
+    }
+    if {[string match "*connection abort*" $msg]} {
+        return {}
+    }
+    error $msg
+}
+
+proc ::redis::redis_safe_gets {fd} {
+    if {[catch {set val [gets $fd]} msg]} {
+        if {[string match "*connection abort*" $msg]} {
+            return {}
+        }
+        error $msg
+    }
+    return $val
+}
+
 # This is a wrapper to the actual dispatching procedure that handles
 # reconnection if needed.
 proc ::redis::__dispatch__ {id method args} {
@@ -146,6 +173,10 @@ proc ::redis::__method__read {id fd} {
     ::redis::redis_read_reply $id $fd
 }
 
+proc ::redis::__method__rawread {id fd {len -1}} {
+    return [redis_safe_read $fd $len]
+}
+
 proc ::redis::__method__write {id fd buf} {
     ::redis::redis_write $fd $buf
 }
@@ -192,8 +223,8 @@ proc ::redis::redis_writenl {fd buf} {
 }
 
 proc ::redis::redis_readnl {fd len} {
-    set buf [read $fd $len]
-    read $fd 2 ; # discard CR LF
+    set buf [redis_safe_read $fd $len]
+    redis_safe_read $fd 2 ; # discard CR LF
     return $buf
 }
 
@@ -239,11 +270,11 @@ proc ::redis::redis_read_map {id fd} {
 }
 
 proc ::redis::redis_read_line fd {
-    string trim [gets $fd]
+    string trim [redis_safe_gets $fd]
 }
 
 proc ::redis::redis_read_null fd {
-    gets $fd
+    redis_safe_gets $fd
     return {}
 }
 
@@ -260,7 +291,7 @@ proc ::redis::redis_read_reply {id fd} {
     }
 
     while {1} {
-        set type [read $fd 1]
+        set type [redis_safe_read $fd 1]
         switch -exact -- $type {
             _ {return [redis_read_null $fd]}
             : -

tests/unit/moduleapi/load.tcl

@@ -0,0 +1,15 @@
+set testmodule [file normalize tests/modules/load.so]
+
+if {$::flash_enabled} {
+    start_server [list tags [list "modules"] overrides [list storage-provider {flash ./rocks.db.master} databases 256 loadmodule $testmodule]] {
+        test "Module is notified of keys loaded from flash" {
+            r flushall
+            r set foo bar
+            r set bar foo
+            r set foobar barfoo
+            assert_equal [r load.count] 0
+            r debug reload
+            assert_equal [r load.count] 3
+        }
+    }
+}

tests/unit/obuf-limits.tcl

@@ -111,7 +111,7 @@ start_server {tags {"obuf-limits"} overrides { server-threads 1 }} {
 
         # Read nothing
         set fd [$rd channel]
-        assert_equal {} [read $fd]
+        assert_equal {} [$rd rawread]
     }
 
     # Note: This test assumes that what's written with one write, will be read by redis in one read.
@@ -151,8 +151,7 @@ start_server {tags {"obuf-limits"} overrides { server-threads 1 }} {
         assert_equal "PONG" [r ping]
         set clients [r client list]
         assert_no_match "*name=multicommands*" $clients
-        set fd [$rd2 channel]
-        assert_equal {} [read $fd]
+        assert_equal {} [$rd2 rawread]
     }
 
     test {Execute transactions completely even if client output buffer limit is enforced} {