Obuf limit, exit during loop in *RAND* commands and KEYS (#11676)

Related to the hang reported in #11671
Currently, redis can disconnect a client due to reaching output buffer limit,
it'll also avoid feeding that output buffer with more data, but it will keep
running the loop in the command (despite the client already being marked for
disconnection)

This PR is an attempt to mitigate the problem, specifically for commands that
are easy to abuse, specifically: KEYS, HRANDFIELD, SRANDMEMBER, ZRANDMEMBER.
The RAND family of commands can take a negative COUNT argument (which is not
bound to the number of elements in the key), so it's enough to create a key
with one field, and then these commands can be used to hang redis.
For KEYS the caller can use the existing keyspace in redis (if big enough).

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/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

src/sort.cpp

@@ -321,8 +321,10 @@ void sortCommand(client *c) {
     default: vectorlen = 0; serverPanic("Bad SORT type"); /* Avoid GCC warning */
     }
 
-    /* Perform LIMIT start,count sanity checking. */
-    start = (limit_start < 0) ? 0 : limit_start;
+    /* Perform LIMIT start,count sanity checking.
+     * And avoid integer overflow by limiting inputs to object sizes. */
+    start = min(max(limit_start, 0), vectorlen);
+    limit_count = min(max(limit_count, -1), vectorlen);
     end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
     if (start >= vectorlen) {
         start = vectorlen-1;

src/t_hash.cpp

@@ -1062,6 +1062,8 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
                 addReplyBulkCBuffer(c, key, sdslen(key));
                 if (withvalues)
                     addReplyBulkCBuffer(c, value, sdslen(value));
+                if (c->flags & CLIENT_CLOSE_ASAP)
+                    break;
             }
         } else if (hash->encoding == OBJ_ENCODING_ZIPLIST) {
             ziplistEntry *keys, *vals = NULL;
@@ -1075,6 +1077,8 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
                 count -= sample_count;
                 ziplistRandomPairs((unsigned char*)ptrFromObj(hash), sample_count, keys, vals);
                 harndfieldReplyWithZiplist(c, sample_count, keys, vals);
+                if (c->flags & CLIENT_CLOSE_ASAP)
+                    break;
             }
             zfree(keys);
             zfree(vals);

src/t_set.cpp

@@ -706,6 +706,8 @@ void srandmemberWithCountCommand(client *c) {
             } else {
                 addReplyBulkCBuffer(c,ele,sdslen(ele));
             }
+            if (c->flags & CLIENT_CLOSE_ASAP)
+                break;
         }
         return;
     }

src/t_string.cpp

@@ -38,8 +38,14 @@ int getGenericCommand(client *c);
  * String Commands
  *----------------------------------------------------------------------------*/
 
-static int checkStringLength(client *c, long long size) {
-    if (!(c->flags & CLIENT_MASTER) && size > g_pserver->proto_max_bulk_len) {
+static int checkStringLength(client *c, long long size, long long append) {
+    if (c->flags & CLIENT_MASTER)
+        return C_OK;
+    /* 'uint64_t' cast is there just to prevent undefined behavior on overflow */
+    long long total = (uint64_t)size + append;
+    /* Test configured max-bulk-len represending a limit of the biggest string object,
+     * and also test for overflow. */
+    if (g_pserver->proto_max_bulk_len || total < size || total < append) {
         addReplyError(c,"string exceeds maximum allowed size (proto-max-bulk-len)");
         return C_ERR;
     }
@@ -445,7 +451,7 @@ void setrangeCommand(client *c) {
         }
 
         /* Return when the resulting string exceeds allowed size */
-        if (checkStringLength(c,offset+sdslen(value)) != C_OK)
+        if (checkStringLength(c,offset,sdslen(value)) != C_OK)
             return;
 
         o = createObject(OBJ_STRING,sdsnewlen(NULL, offset+sdslen(value)));
@@ -465,7 +471,7 @@ void setrangeCommand(client *c) {
         }
 
         /* Return when the resulting string exceeds allowed size */
-        if (checkStringLength(c,offset+sdslen(value)) != C_OK)
+        if (checkStringLength(c,offset,sdslen(value)) != C_OK)
             return;
 
         /* Create a copy when the object is shared or encoded. */
@@ -685,8 +691,7 @@ void appendCommand(client *c) {
 
         /* "append" is an argument, so always an sds */
         append = c->argv[2];
-        totlen = stringObjectLen(o)+sdslen((sds)ptrFromObj(append));
-        if (checkStringLength(c,totlen) != C_OK)
+        if (checkStringLength(c,stringObjectLen(o),sdslen((sds)ptrFromObj(append))) != C_OK)
             return;
 
         /* Append the value */

src/t_zset.cpp

@@ -4054,6 +4054,8 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
                 addReplyBulkCBuffer(c, key, sdslen(key));
                 if (withscores)
                     addReplyDouble(c, *(double*)dictGetVal(de));
+                if (c->flags & CLIENT_CLOSE_ASAP)
+                    break;
             }
         } else if (zsetobj->encoding == OBJ_ENCODING_ZIPLIST) {
             ziplistEntry *keys, *vals = NULL;
@@ -4067,6 +4069,8 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
                 count -= sample_count;
                 ziplistRandomPairs((unsigned char*)ptrFromObj(zsetobj), sample_count, keys, vals);
                 zarndmemberReplyWithZiplist(c, sample_count, keys, vals);
+                if (c->flags & CLIENT_CLOSE_ASAP)
+                    break;
             }
             zfree(keys);
             zfree(vals);

tests/unit/obuf-limits.tcl

@@ -182,4 +182,20 @@ start_server {tags {"obuf-limits"} overrides { server-threads 1 }} {
         assert_equal "v2" [r get k2]
         assert_equal "v3" [r get k3]
     }
+
+    test "Obuf limit, HRANDFIELD with huge count stopped mid-run" {
+        r config set client-output-buffer-limit {normal 1000000 0 0}
+        r hset myhash a b
+        catch {r hrandfield myhash -999999999} e
+        assert_match "*I/O error*" $e
+        reconnect
+    }
+
+    test "Obuf limit, KEYS stopped mid-run" {
+        r config set client-output-buffer-limit {normal 100000 0 0}
+        populate 1000 "long-key-name-prefix-of-100-chars-------------------------------------------------------------------"
+        catch {r keys *} e
+        assert_match "*I/O error*" $e
+        reconnect
+    }
 }

tests/unit/sort.tcl

@@ -315,4 +315,15 @@ start_server {
             }
         }
     }
+
+    test {SETRANGE with huge offset} {
+        r lpush L 2 1 0
+        # expecting a different outcome on 32 and 64 bit systems
+        foreach value {9223372036854775807 2147483647} {
+            catch {[r sort_ro L by a limit 2 $value]} res
+            if {![string match "2" $res] && ![string match "*out of range*" $res]} {
+                assert_not_equal $res "expecting an error or 2"
+            }
+        }
+    }
 }

tests/unit/type/string.tcl

@@ -574,4 +574,14 @@ start_server {tags {"string"}} {
     test {LCS indexes with match len and minimum match len} {
         dict get [r STRALGO LCS IDX KEYS virus1 virus2 WITHMATCHLEN MINMATCHLEN 5] matches
     } {{{1 222} {13 234} 222}}
+
+    test {SETRANGE with huge offset} {
+        foreach value {9223372036854775807 2147483647} {
+            catch {[r setrange K $value A]} res
+            # expecting a different error on 32 and 64 bit systems
+            if {![string match "*string exceeds maximum allowed size*" $res] && ![string match "*out of range*" $res]} {
+                assert_equal $res "expecting an error"
+           }
+        }
+    }
 }