Redis 5.0 RC2.

This way we let big endian systems to still load old RDB versions.
However newver versions will be saved and loaded in a way that make RDB
expires cross-endian again. Thanks to @oranagra for the reporting and
the discussion about this problem, leading to this fix.

00-RELEASENOTES

@@ -1,16 +1,234 @@
-Hello! This file is just a placeholder, since this is the "unstable" branch
-of Redis, the place where all the development happens.
+Redis 5.0 release notes
+=======================
 
-There is no release notes for this branch, it gets forked into another branch
-every time there is a partial feature freeze in order to eventually create
-a new stable release.
+--------------------------------------------------------------------------------
+Upgrade urgency levels:
 
-Usually "unstable" is stable enough for you to use it in development environments
-however you should never use it in production environments. It is possible
-to download the latest stable release here:
+LOW:      No need to upgrade unless there are new features you want to use.
+MODERATE: Program an upgrade of the server, but it's not urgent.
+HIGH:     There is a critical bug that may affect a subset of users. Upgrade!
+CRITICAL: There is a critical bug affecting MOST USERS. Upgrade ASAP.
+SECURITY: There are security fixes in the release.
+--------------------------------------------------------------------------------
 
-    http://download.redis.io/releases/redis-stable.tar.gz
+================================================================================
+Redis 5.0 RC2     Released Wed Jun 13 12:49:13 CEST 2018
+================================================================================
 
-More information is available at http://redis.io
+Upgrade urgency CRITICAL: This release fixes important security issues.
+                    HIGH: This release fixes a SCAN commands family bug.
+                MODERATE: This release fixes a PSYNC2 edge case with expires.
+                MODERATE: Sentinel related fixes.
+                     LOW: All the other issues
 
-Happy hacking!
+NOTE: This release breaks API compatibility with certain APIs that were
+introduced in Redis 5.0 RC1. Notably ZPOP* now returns score/element in reverse
+order. XINFO <key> special form was removed, now XINFO STREAM <key> must be
+used to obtain general information about the stream.
+
+Redis 5.0 RC2 ixes a number of important issues:
+
+* Important security issues related to the Lua scripting engine.
+  Please check https://github.com/antirez/redis/issues/5017
+  for more information.
+
+* A bug with SCAN, SSCAN, HSCAN and ZSCAN, that may not return all the elements.
+  We also add a regression test that can trigger the issue often when present, and
+  may in theory be able to find unrelated regressions.
+
+* A PSYNC2 bug is fixed: Redis should not expire keys when saving RDB files
+  because otherwise it is no longer possible to use such RDB file as a base
+  for partial resynchronization. It no longer represents the right state.
+
+* Compatibility of AOF with RDB preamble when the RDB checksum is disabled.
+
+* Sentinel bug that in some cases prevented Sentinel to detect that the master
+  was down immediately. A delay was added to the detection.
+
+* XREADGROUP would crash when the master had attacked slaves.
+
+* Replication and events generation of several streams command were fixed.
+
+* XREVRANGE fixed, in some cases it could not return elements, or crash the
+  server, or in general not behave correctly.
+
+* ZPOP can now unblock multiple clients in a sane way.
+
+* Other minor issues.
+
+Moreover this release adds new features:
+
+* XGROUP DESTROY and XGROUP SETID.
+
+* RDB loading speedup.
+
+* Configurable stream macro node limits (number of elements / bytes).
+
+* More smaller improvements.
+
+The following is the list of commits composing the release, please check
+the commit messages and authors for credits.
+
+antirez in commit 9fdcc159:
+ Security: fix redis-cli buffer overflow.
+ 1 file changed, 16 insertions(+), 11 deletions(-)
+
+antirez in commit cf760071:
+ Security: fix Lua struct package offset handling.
+ 1 file changed, 6 insertions(+), 2 deletions(-)
+
+antirez in commit a57595ca:
+ Security: more cmsgpack fixes by @soloestoy.
+ 1 file changed, 7 insertions(+)
+
+antirez in commit 8783fb94:
+ Security: update Lua struct package for security.
+ 1 file changed, 23 insertions(+), 23 deletions(-)
+
+antirez in commit 8cb9344b:
+ Security: fix Lua cmsgpack library stack overflow.
+ 1 file changed, 3 insertions(+)
+
+赵磊 in commit 59080f60:
+ Fix dictScan(): It can't scan all buckets when dict is shrinking.
+ 1 file changed, 14 insertions(+), 11 deletions(-)
+
+dejun.xdj in commit ac2a824a:
+ Fix redis-cli memory leak when sending set preference command.
+ 1 file changed, 2 insertions(+)
+
+dejun.xdj in commit c7197ff5:
+ Check if the repeat value is positive in while loop of cliSendCommand().
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+dejun.xdj in commit 3f77777f:
+ Change the type of repeat argument to long for function cliSendCommand.
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+dejun.xdj in commit 7a565d72:
+ Fix negtive repeat command value issue.
+ 1 file changed, 11 insertions(+), 3 deletions(-)
+
+dejun.xdj in commit 64bf60fb:
+ Detect and stop saving history for auth command with repeat option.
+ 1 file changed, 17 insertions(+), 10 deletions(-)
+
+dejun.xdj in commit 5bed12aa:
+ Change the warning message a little bit to avoid trademark issuses.
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+dejun.xdj in commit d71c4961:
+ Stop saving auth command in redis-cli history.
+ 1 file changed, 4 insertions(+), 2 deletions(-)
+
+dejun.xdj in commit fca99e41:
+ Add warning message when using password on command line
+ 1 file changed, 1 insertion(+)
+
+antirez in commit 01407a3a:
+ Don't expire keys while loading RDB from AOF preamble.
+ 3 files changed, 5 insertions(+), 5 deletions(-)
+
+WuYunlong in commit fb5408cf:
+ Fix rdb save by allowing dumping of expire keys, so that when we add a new slave, and do a failover, eighter by manual or not, other local slaves will delete the expired keys properly.
+ 2 files changed, 3 insertions(+), 7 deletions(-)
+
+antirez in commit 0b8b6df4:
+ Backport hiredis issue 525 fix to compile on FreeBSD.
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+antirez in commit e98627c5:
+ Add INIT INFO to the provided init script.
+ 1 file changed, 8 insertions(+)
+
+antirez in commit 17f5de89:
+ Fix ae.c when a timer finalizerProc adds an event.
+ 2 files changed, 10 insertions(+), 6 deletions(-)
+
+antirez in commit 266e6423:
+ Sentinel: fix delay in detecting ODOWN.
+ 1 file changed, 9 insertions(+), 5 deletions(-)
+
+zhaozhao.zz in commit eafaf172:
+ AOF & RDB: be compatible with rdbchecksum no
+ 1 file changed, 9 insertions(+), 7 deletions(-)
+
+huijing.whj in commit 4630da37:
+ fix int overflow problem in freeMemoryIfNeeded
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+================================================================================
+Redis 5.0 RC1   Released Tue May 29 14:14:11 CEST 2018
+================================================================================
+
+Upgrade urgency LOW: This is the first RC of Redis 5.
+
+Introduction to the Redis 5 release
+===================================
+
+Redis 5 is a release focused on a few important features. While Redis 4
+was very very focused on operations, Redis 5 changes are mostly user-facing,
+with the implementation of new data types and operations on top of existing
+types. The following are the major features of this release:
+
+1. The new Stream data type. https://redis.io/topics/streams-intro
+2. New Redis modules APIs: Timers and Cluster API.
+3. RDB now store LFU and LRU information.
+4. The cluster manager was ported from Ruby (redis-trib.rb) to C code
+   inside redis-cli. Check `redis-cli --cluster help` for more info.
+5. New sorted set commands: ZPOPMIN/MAX and blocking variants.
+6. Active defragmentation version 2.
+7. Improvemenets in HyperLogLog implementations.
+8. Better memory reporting capabilities.
+9. Many commands with sub-commands now have an HELP subcommand.
+10. Better performances when clients connect and disconnect often.
+11. Many bug fixes and other random improvements.
+12. Jemalloc was upgraded to version 5.1
+
+Thanks to all the users and developers who made this release possible.
+We'll follow up with more RC releases, until the code looks production ready
+and we don't get reports of serious issues for a while.
+
+A special thank you for the amount of work put into this release
+(in decreasing number of commits) by:
+
+Fabio Nicotra, <artix2@gmail.com>
+Soloestoy <zhaozhao.zz@alibaba-inc.com>
+Itamar Haber <itamar@redislabs.com>
+Oran Agra <oran@redislabs.com>
+Dvir Volk <dvirsky@gmail.com>
+dejun.xdj <dejun.xdj@alibaba-inc.com>
+Guy Benoish <guy.benoish@redislabs.com>
+Charsyam <charsyam@gmail.com>
+Otmar Ertl <otmar.ertl@gmail.com>
+Jan-Erik Rediger <janerik@fnordig.de>
+Spinlock <wnzheng@gmail.com>
+
+Migrating from 4.0 to 5.0
+=========================
+
+Redis 4.0 is mostly a strict subset of 5.0, you should not have any problem
+upgrading your application from 4.0 to 5.0. However this is a list of small
+non-backward compatible changes introduced in the 5.0 release:
+
+* redis-cli now implements the cluster management tool. We still ship the
+  old redis-trib, but new fixes will be implemented only in redis-cli.
+  See `redis-cli --cluster help` for more info.
+
+* The RDB format changed. Redis 5.0 is still able to read 4.0 (and all the
+  past versions) files, but not the other way around.
+
+* Certain log formats and sentences are different in Redis 5.0.
+
+--------------------------------------------------------------------------------
+
+Credits: For each release, a list of changes with the relative author is
+provided. Where not specified the implementation and design is done by
+Salvatore Sanfilippo. Thanks to Redis Labs for making all this possible.
+Also many thanks to all the other contributors and the amazing community
+we have.
+
+Commit messages may contain additional credits.
+
+Enjoy,
+Salvatore

deps/lua/src/lua_cmsgpack.c

@@ -385,6 +385,7 @@ void mp_encode_lua_table_as_array(lua_State *L, mp_buf *buf, int level) {
 #endif
 
     mp_encode_array(L,buf,len);
+    luaL_checkstack(L, 1, "in function mp_encode_lua_table_as_array");
     for (j = 1; j <= len; j++) {
         lua_pushnumber(L,j);
         lua_gettable(L,-2);
@@ -400,6 +401,7 @@ void mp_encode_lua_table_as_map(lua_State *L, mp_buf *buf, int level) {
      * Lua API, we need to iterate a first time. Note that an alternative
      * would be to do a single run, and then hack the buffer to insert the
      * map opcodes for message pack. Too hackish for this lib. */
+    luaL_checkstack(L, 3, "in function mp_encode_lua_table_as_map");
     lua_pushnil(L);
     while(lua_next(L,-2)) {
         lua_pop(L,1); /* remove value, keep key for next iteration. */
@@ -515,10 +517,14 @@ int mp_pack(lua_State *L) {
     if (nargs == 0)
         return luaL_argerror(L, 0, "MessagePack pack needs input.");
 
+    if (!lua_checkstack(L, nargs))
+        return luaL_argerror(L, 0, "Too many arguments for MessagePack pack.");
+
     buf = mp_buf_new(L);
     for(i = 1; i <= nargs; i++) {
         /* Copy argument i to top of stack for _encode processing;
          * the encode function pops it from the stack when complete. */
+        luaL_checkstack(L, 1, "in function mp_check");
         lua_pushvalue(L, i);
 
         mp_encode_lua_type(L,buf,0);
@@ -547,6 +553,7 @@ void mp_decode_to_lua_array(lua_State *L, mp_cur *c, size_t len) {
     int index = 1;
 
     lua_newtable(L);
+    luaL_checkstack(L, 1, "in function mp_decode_to_lua_array");
     while(len--) {
         lua_pushnumber(L,index++);
         mp_decode_to_lua_type(L,c);
@@ -821,6 +828,9 @@ int mp_unpack_full(lua_State *L, int limit, int offset) {
          * subtract the entire buffer size from the unprocessed size
          * to get our next start offset */
         int offset = len - c.left;
+
+        luaL_checkstack(L, 1, "in function mp_unpack_full");
+
         /* Return offset -1 when we have have processed the entire buffer. */
         lua_pushinteger(L, c.left == 0 ? -1 : offset);
         /* Results are returned with the arg elements still

deps/lua/src/lua_struct.c

@@ -1,7 +1,7 @@
 /*
 ** {======================================================
 ** Library for packing/unpacking structures.
-** $Id: struct.c,v 1.4 2012/07/04 18:54:29 roberto Exp $
+** $Id: struct.c,v 1.7 2018/05/11 22:04:31 roberto Exp $
 ** See Copyright Notice at the end of this file
 ** =======================================================
 */
@@ -15,8 +15,8 @@
 ** h/H - signed/unsigned short
 ** l/L - signed/unsigned long
 ** T   - size_t
-** i/In - signed/unsigned integer with size `n' (default is size of int)
-** cn - sequence of `n' chars (from/to a string); when packing, n==0 means
+** i/In - signed/unsigned integer with size 'n' (default is size of int)
+** cn - sequence of 'n' chars (from/to a string); when packing, n==0 means
         the whole string; when unpacking, n==0 means use the previous
         read number as the string length
 ** s - zero-terminated string
@@ -89,14 +89,12 @@ typedef struct Header {
 } Header;
 
 
-static int getnum (lua_State *L, const char **fmt, int df) {
+static int getnum (const char **fmt, int df) {
   if (!isdigit(**fmt))  /* no number? */
     return df;  /* return default value */
   else {
     int a = 0;
     do {
-      if (a > (INT_MAX / 10) || a * 10 > (INT_MAX - (**fmt - '0')))
-        luaL_error(L, "integral size overflow");
       a = a*10 + *((*fmt)++) - '0';
     } while (isdigit(**fmt));
     return a;
@@ -117,9 +115,9 @@ static size_t optsize (lua_State *L, char opt, const char **fmt) {
     case 'f':  return sizeof(float);
     case 'd':  return sizeof(double);
     case 'x': return 1;
-    case 'c': return getnum(L, fmt, 1);
+    case 'c': return getnum(fmt, 1);
     case 'i': case 'I': {
-      int sz = getnum(L, fmt, sizeof(int));
+      int sz = getnum(fmt, sizeof(int));
       if (sz > MAXINTSIZE)
         luaL_error(L, "integral size %d is larger than limit of %d",
                        sz, MAXINTSIZE);
@@ -152,7 +150,7 @@ static void controloptions (lua_State *L, int opt, const char **fmt,
     case '>': h->endian = BIG; return;
     case '<': h->endian = LITTLE; return;
     case '!': {
-      int a = getnum(L, fmt, MAXALIGN);
+      int a = getnum(fmt, MAXALIGN);
       if (!isp2(a))
         luaL_error(L, "alignment %d is not a power of 2", a);
       h->align = a;
@@ -295,21 +293,26 @@ static int b_unpack (lua_State *L) {
   const char *fmt = luaL_checkstring(L, 1);
   size_t ld;
   const char *data = luaL_checklstring(L, 2, &ld);
-  size_t pos = luaL_optinteger(L, 3, 1) - 1;
+  size_t pos = luaL_optinteger(L, 3, 1);
+  luaL_argcheck(L, pos > 0, 3, "offset must be 1 or greater");
+  pos--; /* Lua indexes are 1-based, but here we want 0-based for C
+          * pointer math. */
+  int n = 0;  /* number of results */
   defaultoptions(&h);
-  lua_settop(L, 2);
   while (*fmt) {
     int opt = *fmt++;
     size_t size = optsize(L, opt, &fmt);
     pos += gettoalign(pos, &h, opt, size);
-    luaL_argcheck(L, pos+size <= ld, 2, "data string too short");
-    luaL_checkstack(L, 1, "too many results");
+    luaL_argcheck(L, size <= ld && pos <= ld - size,
+                   2, "data string too short");
+    /* stack space for item + next position */
+    luaL_checkstack(L, 2, "too many results");
     switch (opt) {
       case 'b': case 'B': case 'h': case 'H':
       case 'l': case 'L': case 'T': case 'i':  case 'I': {  /* integer types */
         int issigned = islower(opt);
         lua_Number res = getinteger(data+pos, h.endian, issigned, size);
-        lua_pushnumber(L, res);
+        lua_pushnumber(L, res); n++;
         break;
       }
       case 'x': {
@@ -319,25 +322,26 @@ static int b_unpack (lua_State *L) {
         float f;
         memcpy(&f, data+pos, size);
         correctbytes((char *)&f, sizeof(f), h.endian);
-        lua_pushnumber(L, f);
+        lua_pushnumber(L, f); n++;
         break;
       }
       case 'd': {
         double d;
         memcpy(&d, data+pos, size);
         correctbytes((char *)&d, sizeof(d), h.endian);
-        lua_pushnumber(L, d);
+        lua_pushnumber(L, d); n++;
         break;
       }
       case 'c': {
         if (size == 0) {
-          if (!lua_isnumber(L, -1))
-            luaL_error(L, "format `c0' needs a previous size");
+          if (n == 0 || !lua_isnumber(L, -1))
+            luaL_error(L, "format 'c0' needs a previous size");
           size = lua_tonumber(L, -1);
-          lua_pop(L, 1);
-          luaL_argcheck(L, pos+size <= ld, 2, "data string too short");
+          lua_pop(L, 1); n--;
+          luaL_argcheck(L, size <= ld && pos <= ld - size,
+                           2, "data string too short");
         }
-        lua_pushlstring(L, data+pos, size);
+        lua_pushlstring(L, data+pos, size); n++;
         break;
       }
       case 's': {
@@ -345,15 +349,15 @@ static int b_unpack (lua_State *L) {
         if (e == NULL)
           luaL_error(L, "unfinished string in data");
         size = (e - (data+pos)) + 1;
-        lua_pushlstring(L, data+pos, size - 1);
+        lua_pushlstring(L, data+pos, size - 1); n++;
         break;
       }
       default: controloptions(L, opt, &fmt, &h);
     }
     pos += size;
   }
-  lua_pushinteger(L, pos + 1);
-  return lua_gettop(L) - 2;
+  lua_pushinteger(L, pos + 1);  /* next position */
+  return n + 1;
 }
 
 
@@ -399,7 +403,7 @@ LUALIB_API int luaopen_struct (lua_State *L) {
 
 
 /******************************************************************************
-* Copyright (C) 2010-2012 Lua.org, PUC-Rio.  All rights reserved.
+* Copyright (C) 2010-2018 Lua.org, PUC-Rio.  All rights reserved.
 *
 * Permission is hereby granted, free of charge, to any person obtaining
 * a copy of this software and associated documentation files (the

redis.conf

@@ -1106,6 +1106,17 @@ zset-max-ziplist-value 64
 # composed of many HyperLogLogs with cardinality in the 0 - 15000 range.
 hll-sparse-max-bytes 3000
 
+# Streams macro node max size / items. The stream data structure is a radix
+# tree of big nodes that encode multiple items inside. Using this configuration
+# it is possible to configure how big a single node can be in bytes, and the
+# maximum number of items it may contain before switching to a new node when
+# appending new stream entries. If any of the following settings are set to
+# zero, the limit is ignored, so for instance it is possible to set just a
+# max entires limit by setting max-bytes to 0 and max-entries to the desired
+# value.
+stream-node-max-bytes 4096
+stream-node-max-entries 100
+
 # Active rehashing uses 1 millisecond every 100 milliseconds of CPU time in
 # order to help rehashing the main Redis hash table (the one mapping top-level
 # keys to values). The hash table implementation Redis uses (see dict.c)

src/blocked.c

@@ -314,8 +314,9 @@ void handleClientsBlockedOnKeys(void) {
                 if (de) {
                     list *clients = dictGetVal(de);
                     int numclients = listLength(clients);
+                    unsigned long zcard = zsetLength(o);
 
-                    while(numclients--) {
+                    while(numclients-- && zcard) {
                         listNode *clientnode = listFirst(clients);
                         client *receiver = clientnode->value;
 
@@ -332,6 +333,7 @@ void handleClientsBlockedOnKeys(void) {
                                      ? ZSET_MIN : ZSET_MAX;
                         unblockClient(receiver);
                         genericZpopCommand(receiver,&rl->key,1,where,1,NULL);
+                        zcard--;
 
                         /* Replicate the command. */
                         robj *argv[2];
@@ -396,12 +398,6 @@ void handleClientsBlockedOnKeys(void) {
                                            1);
                             }
 
-                            /* Note that after we unblock the client, 'gt'
-                             * and other receiver->bpop stuff are no longer
-                             * valid, so we must do the setup above before
-                             * this call. */
-                            unblockClient(receiver);
-
                             /* Emit the two elements sub-array consisting of
                              * the name of the stream and the data we
                              * extracted from it. Wrapped in a single-item
@@ -417,6 +413,12 @@ void handleClientsBlockedOnKeys(void) {
                             streamReplyWithRange(receiver,s,&start,NULL,
                                                  receiver->bpop.xread_count,
                                                  0, group, consumer, 0, &pi);
+
+                            /* Note that after we unblock the client, 'gt'
+                             * and other receiver->bpop stuff are no longer
+                             * valid, so we must do the setup above before
+                             * this call. */
+                            unblockClient(receiver);
                         }
                     }
                 }

src/cluster.c

@@ -2120,7 +2120,7 @@ void clusterWriteHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
     nwritten = write(fd, link->sndbuf, sdslen(link->sndbuf));
     if (nwritten <= 0) {
         serverLog(LL_DEBUG,"I/O error writing to node link: %s",
-            strerror(errno));
+            (nwritten == -1) ? strerror(errno) : "short write");
         handleLinkIOError(link);
         return;
     }
@@ -2377,7 +2377,7 @@ void clusterSendPing(clusterLink *link, int type) {
      * same time.
      *
      * Since we have non-voting slaves that lower the probability of an entry
-     * to feature our node, we set the number of entires per packet as
+     * to feature our node, we set the number of entries per packet as
      * 10% of the total nodes we have. */
     wanted = floor(dictSize(server.cluster->nodes)/10);
     if (wanted < 3) wanted = 3;

src/config.c

@@ -509,14 +509,16 @@ void loadServerConfigFromString(char *config) {
             server.rdb_filename = zstrdup(argv[1]);
         } else if (!strcasecmp(argv[0],"active-defrag-threshold-lower") && argc == 2) {
             server.active_defrag_threshold_lower = atoi(argv[1]);
-            if (server.active_defrag_threshold_lower < 0) {
-                err = "active-defrag-threshold-lower must be 0 or greater";
+            if (server.active_defrag_threshold_lower < 0 ||
+                server.active_defrag_threshold_lower > 1000) {
+                err = "active-defrag-threshold-lower must be between 0 and 1000";
                 goto loaderr;
             }
         } else if (!strcasecmp(argv[0],"active-defrag-threshold-upper") && argc == 2) {
             server.active_defrag_threshold_upper = atoi(argv[1]);
-            if (server.active_defrag_threshold_upper < 0) {
-                err = "active-defrag-threshold-upper must be 0 or greater";
+            if (server.active_defrag_threshold_upper < 0 ||
+                server.active_defrag_threshold_upper > 1000) {
+                err = "active-defrag-threshold-upper must be between 0 and 1000";
                 goto loaderr;
             }
         } else if (!strcasecmp(argv[0],"active-defrag-ignore-bytes") && argc == 2) {
@@ -547,6 +549,10 @@ void loadServerConfigFromString(char *config) {
             server.hash_max_ziplist_entries = memtoll(argv[1], NULL);
         } else if (!strcasecmp(argv[0],"hash-max-ziplist-value") && argc == 2) {
             server.hash_max_ziplist_value = memtoll(argv[1], NULL);
+        } else if (!strcasecmp(argv[0],"stream-node-max-bytes") && argc == 2) {
+            server.stream_node_max_bytes = memtoll(argv[1], NULL);
+        } else if (!strcasecmp(argv[0],"stream-node-max-entries") && argc == 2) {
+            server.stream_node_max_entries = atoi(argv[1]);
         } else if (!strcasecmp(argv[0],"list-max-ziplist-entries") && argc == 2){
             /* DEAD OPTION */
         } else if (!strcasecmp(argv[0],"list-max-ziplist-value") && argc == 2) {
@@ -1083,6 +1089,10 @@ void configSetCommand(client *c) {
       "hash-max-ziplist-entries",server.hash_max_ziplist_entries,0,LLONG_MAX) {
     } config_set_numerical_field(
       "hash-max-ziplist-value",server.hash_max_ziplist_value,0,LLONG_MAX) {
+    } config_set_numerical_field(
+      "stream-node-max-bytes",server.stream_node_max_bytes,0,LLONG_MAX) {
+    } config_set_numerical_field(
+      "stream-node-max-entries",server.stream_node_max_entries,0,LLONG_MAX) {
     } config_set_numerical_field(
       "list-max-ziplist-size",server.list_max_ziplist_size,INT_MIN,INT_MAX) {
     } config_set_numerical_field(
@@ -1102,7 +1112,7 @@ void configSetCommand(client *c) {
     } config_set_numerical_field(
       "slowlog-max-len",ll,0,LLONG_MAX) {
       /* Cast to unsigned. */
-        server.slowlog_max_len = (unsigned)ll;
+        server.slowlog_max_len = (unsigned long)ll;
     } config_set_numerical_field(
       "latency-monitor-threshold",server.latency_monitor_threshold,0,LLONG_MAX){
     } config_set_numerical_field(
@@ -1267,6 +1277,10 @@ void configGetCommand(client *c) {
             server.hash_max_ziplist_entries);
     config_get_numerical_field("hash-max-ziplist-value",
             server.hash_max_ziplist_value);
+    config_get_numerical_field("stream-node-max-bytes",
+            server.stream_node_max_bytes);
+    config_get_numerical_field("stream-node-max-entries",
+            server.stream_node_max_entries);
     config_get_numerical_field("list-max-ziplist-size",
             server.list_max_ziplist_size);
     config_get_numerical_field("list-compress-depth",
@@ -2056,6 +2070,8 @@ int rewriteConfig(char *path) {
     rewriteConfigNotifykeyspaceeventsOption(state);
     rewriteConfigNumericalOption(state,"hash-max-ziplist-entries",server.hash_max_ziplist_entries,OBJ_HASH_MAX_ZIPLIST_ENTRIES);
     rewriteConfigNumericalOption(state,"hash-max-ziplist-value",server.hash_max_ziplist_value,OBJ_HASH_MAX_ZIPLIST_VALUE);
+    rewriteConfigNumericalOption(state,"stream-node-max-bytes",server.stream_node_max_bytes,OBJ_STREAM_NODE_MAX_BYTES);
+    rewriteConfigNumericalOption(state,"stream-node-max-entries",server.stream_node_max_entries,OBJ_STREAM_NODE_MAX_ENTRIES);
     rewriteConfigNumericalOption(state,"list-max-ziplist-size",server.list_max_ziplist_size,OBJ_LIST_MAX_ZIPLIST_SIZE);
     rewriteConfigNumericalOption(state,"list-compress-depth",server.list_compress_depth,OBJ_LIST_COMPRESS_DEPTH);
     rewriteConfigNumericalOption(state,"set-max-intset-entries",server.set_max_intset_entries,OBJ_SET_MAX_INTSET_ENTRIES);

src/db.c

@@ -942,16 +942,18 @@ void moveCommand(client *c) {
 }
 
 /* Helper function for dbSwapDatabases(): scans the list of keys that have
- * one or more blocked clients for B[LR]POP or other list blocking commands
- * and signal the keys are ready if they are lists. See the comment where
- * the function is used for more info. */
+ * one or more blocked clients for B[LR]POP or other blocking commands
+ * and signal the keys as ready if they are of the right type. See the comment
+ * where the function is used for more info. */
 void scanDatabaseForReadyLists(redisDb *db) {
     dictEntry *de;
     dictIterator *di = dictGetSafeIterator(db->blocking_keys);
     while((de = dictNext(di)) != NULL) {
         robj *key = dictGetKey(de);
         robj *value = lookupKey(db,key,LOOKUP_NOTOUCH);
-        if (value && (value->type == OBJ_LIST || value->type == OBJ_STREAM))
+        if (value && (value->type == OBJ_LIST ||
+                      value->type == OBJ_STREAM ||
+                      value->type == OBJ_ZSET))
             signalKeyAsReady(db, key);
     }
     dictReleaseIterator(di);

src/debug.c

@@ -287,9 +287,10 @@ void debugCommand(client *c) {
         const char *help[] = {
 "assert -- Crash by assertion failed.",
 "change-repl-id -- Change the replication IDs of the instance. Dangerous, should be used only for testing the replication subsystem.",
-"crash-and-recovery <milliseconds> -- Hard crash and restart after <milliseconds> delay.",
+"crash-and-recover <milliseconds> -- Hard crash and restart after <milliseconds> delay.",
 "digest -- Outputs an hex signature representing the current DB content.",
 "htstats <dbid> -- Return hash table statistics of the specified Redis database.",
+"htstats-key <key> -- Like htstats but for the hash table stored as key's value.",
 "loadaof -- Flush the AOF buffers on disk and reload the AOF in memory.",
 "lua-always-replicate-commands (0|1) -- Setting it to 1 makes Lua replication defaulting to replicating single commands, without the script having to enable effects replication.",
 "object <key> -- Show low level info about key and associated value.",
@@ -547,6 +548,34 @@ NULL
         stats = sdscat(stats,buf);
 
         addReplyBulkSds(c,stats);
+    } else if (!strcasecmp(c->argv[1]->ptr,"htstats-key") && c->argc == 3) {
+        robj *o;
+        dict *ht = NULL;
+
+        if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nokeyerr))
+                == NULL) return;
+
+        /* Get the hash table reference from the object, if possible. */
+        switch (o->encoding) {
+        case OBJ_ENCODING_SKIPLIST:
+            {
+                zset *zs = o->ptr;
+                ht = zs->dict;
+            }
+            break;
+        case OBJ_ENCODING_HT:
+            ht = o->ptr;
+            break;
+        }
+
+        if (ht == NULL) {
+            addReplyError(c,"The value stored at the specified key is not "
+                            "represented using an hash table");
+        } else {
+            char buf[4096];
+            dictGetStats(buf,sizeof(buf),ht);
+            addReplyBulkCString(c,buf);
+        }
     } else if (!strcasecmp(c->argv[1]->ptr,"change-repl-id") && c->argc == 2) {
         serverLog(LL_WARNING,"Changing replication IDs after receiving DEBUG change-repl-id");
         changeReplicationId();

src/dict.c

@@ -146,14 +146,14 @@ int dictResize(dict *d)
 /* Expand or create the hash table */
 int dictExpand(dict *d, unsigned long size)
 {
-    dictht n; /* the new hash table */
-    unsigned long realsize = _dictNextPower(size);
-
     /* the size is invalid if it is smaller than the number of
      * elements already inside the hash table */
     if (dictIsRehashing(d) || d->ht[0].used > size)
         return DICT_ERR;
 
+    dictht n; /* the new hash table */
+    unsigned long realsize = _dictNextPower(size);
+
     /* Rehashing to the same table size is not useful. */
     if (realsize == d->ht[0].size) return DICT_ERR;
 
@@ -858,6 +858,15 @@ unsigned long dictScan(dict *d,
             de = next;
         }
 
+        /* Set unmasked bits so incrementing the reversed cursor
+         * operates on the masked bits */
+        v |= ~m0;
+
+        /* Increment the reverse cursor */
+        v = rev(v);
+        v++;
+        v = rev(v);
+
     } else {
         t0 = &d->ht[0];
         t1 = &d->ht[1];
@@ -892,22 +901,16 @@ unsigned long dictScan(dict *d,
                 de = next;
             }
 
-            /* Increment bits not covered by the smaller mask */
-            v = (((v | m0) + 1) & ~m0) | (v & m0);
+            /* Increment the reverse cursor not covered by the smaller mask.*/
+            v |= ~m1;
+            v = rev(v);
+            v++;
+            v = rev(v);
 
             /* Continue while bits covered by mask difference is non-zero */
         } while (v & (m0 ^ m1));
     }
 
-    /* Set unmasked bits so incrementing the reversed cursor
-     * operates on the masked bits of the smaller table */
-    v |= ~m0;
-
-    /* Increment the reverse cursor */
-    v = rev(v);
-    v++;
-    v = rev(v);
-
     return v;
 }
 

src/help.h

@@ -1,4 +1,4 @@
-/* Automatically generated by utils/generate-command-help.rb, do not edit. */
+/* Automatically generated by generate-command-help.rb, do not edit. */
 
 #ifndef __REDIS_HELP_H
 #define __REDIS_HELP_H
@@ -17,7 +17,8 @@ static char *commandGroups[] = {
     "scripting",
     "hyperloglog",
     "cluster",
-    "geo"
+    "geo",
+    "stream"
 };
 
 struct commandHelp {
@@ -82,6 +83,16 @@ struct commandHelp {
     "Pop a value from a list, push it to another list and return it; or block until one is available",
     2,
     "2.2.0" },
+    { "BZPOPMAX",
+    "key [key ...] timeout",
+    "Remove and return the member with the highest score from one or more sorted sets, or block until one is available",
+    4,
+    "5.0.0" },
+    { "BZPOPMIN",
+    "key [key ...] timeout",
+    "Remove and return the member with the lowest score from one or more sorted sets, or block until one is available",
+    4,
+    "5.0.0" },
     { "CLIENT GETNAME",
     "-",
     "Get the current connection name",
@@ -318,12 +329,12 @@ struct commandHelp {
     0,
     "1.2.0" },
     { "FLUSHALL",
-    "-",
+    "[ASYNC]",
     "Remove all keys from all databases",
     9,
     "1.0.0" },
     { "FLUSHDB",
-    "-",
+    "[ASYNC]",
     "Remove all keys from the current database",
     9,
     "1.0.0" },
@@ -532,6 +543,36 @@ struct commandHelp {
     "Trim a list to the specified range",
     2,
     "1.0.0" },
+    { "MEMORY DOCTOR",
+    "-",
+    "Outputs memory problems report",
+    9,
+    "4.0.0" },
+    { "MEMORY HELP",
+    "-",
+    "Show helpful text about the different subcommands",
+    9,
+    "4.0.0" },
+    { "MEMORY MALLOC-STATS",
+    "-",
+    "Show allocator internal stats",
+    9,
+    "4.0.0" },
+    { "MEMORY PURGE",
+    "-",
+    "Ask the allocator to release memory",
+    9,
+    "4.0.0" },
+    { "MEMORY STATS",
+    "-",
+    "Show memory usage details",
+    9,
+    "4.0.0" },
+    { "MEMORY USAGE",
+    "key [SAMPLES count]",
+    "Estimate the memory usage of a key",
+    9,
+    "4.0.0" },
     { "MGET",
     "key [key ...]",
     "Get the values of all the given keys",
@@ -723,7 +764,7 @@ struct commandHelp {
     10,
     "3.2.0" },
     { "SCRIPT EXISTS",
-    "script [script ...]",
+    "sha1 [sha1 ...]",
     "Check existence of scripts in the script cache.",
     10,
     "2.6.0" },
@@ -758,7 +799,7 @@ struct commandHelp {
     8,
     "1.0.0" },
     { "SET",
-    "key value [EX seconds] [PX milliseconds] [NX|XX]",
+    "key value [expiration EX seconds|PX milliseconds] [NX|XX]",
     "Set the string value of a key",
     1,
     "1.0.0" },
@@ -867,6 +908,11 @@ struct commandHelp {
     "Add multiple sets and store the resulting set in a key",
     3,
     "1.0.0" },
+    { "SWAPDB",
+    "index index",
+    "Swaps two Redis databases",
+    8,
+    "4.0.0" },
     { "SYNC",
     "-",
     "Internal command used for replication",
@@ -877,6 +923,11 @@ struct commandHelp {
     "Return the current server time",
     9,
     "2.6.0" },
+    { "TOUCH",
+    "key [key ...]",
+    "Alters the last access time of a key(s). Returns the number of existing keys specified.",
+    0,
+    "3.2.1" },
     { "TTL",
     "key",
     "Get the time to live for a key",
@@ -887,6 +938,11 @@ struct commandHelp {
     "Determine the type stored at key",
     0,
     "1.0.0" },
+    { "UNLINK",
+    "key [key ...]",
+    "Delete a key asynchronously in another thread. Otherwise it is just as DEL, but non blocking.",
+    0,
+    "4.0.0" },
     { "UNSUBSCRIBE",
     "[channel [channel ...]]",
     "Stop listening for messages posted to the given channels",
@@ -907,6 +963,41 @@ struct commandHelp {
     "Watch the given keys to determine execution of the MULTI/EXEC block",
     7,
     "2.2.0" },
+    { "XADD",
+    "key ID field string [field string ...]",
+    "Appends a new entry to a stream",
+    14,
+    "5.0.0" },
+    { "XLEN",
+    "key",
+    "Return the number of entires in a stream",
+    14,
+    "5.0.0" },
+    { "XPENDING",
+    "key group [start end count] [consumer]",
+    "Return information and entries from a stream conusmer group pending entries list, that are messages fetched but never acknowledged.",
+    14,
+    "5.0.0" },
+    { "XRANGE",
+    "key start end [COUNT count]",
+    "Return a range of elements in a stream, with IDs matching the specified IDs interval",
+    14,
+    "5.0.0" },
+    { "XREAD",
+    "[COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]",
+    "Return never seen elements in multiple streams, with IDs greater than the ones reported by the caller for each stream. Can block.",
+    14,
+    "5.0.0" },
+    { "XREADGROUP",
+    "GROUP group consumer [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]",
+    "Return new entries from a stream using a consumer group, or access the history of the pending entries for a given consumer. Can block.",
+    14,
+    "5.0.0" },
+    { "XREVRANGE",
+    "key end start [COUNT count]",
+    "Return a range of elements in a stream, with IDs matching the specified IDs interval, in reverse order (from greater to smaller IDs) compared to XRANGE",
+    14,
+    "5.0.0" },
     { "ZADD",
     "key [NX|XX] [CH] [INCR] score member [score member ...]",
     "Add one or more members to a sorted set, or update its score if it already exists",
@@ -937,6 +1028,16 @@ struct commandHelp {
     "Count the number of members in a sorted set between a given lexicographical range",
     4,
     "2.8.9" },
+    { "ZPOPMAX",
+    "key [count]",
+    "Remove and return members with the highest scores in a sorted set",
+    4,
+    "5.0.0" },
+    { "ZPOPMIN",
+    "key [count]",
+    "Remove and return members with the lowest scores in a sorted set",
+    4,
+    "5.0.0" },
     { "ZRANGE",
     "key start stop [WITHSCORES]",
     "Return a range of members in a sorted set, by index",

src/hyperloglog.c

@@ -429,14 +429,14 @@ uint64_t MurmurHash64A (const void * key, int len, unsigned int seed) {
     }
 
     switch(len & 7) {
-    case 7: h ^= (uint64_t)data[6] << 48;
-    case 6: h ^= (uint64_t)data[5] << 40;
-    case 5: h ^= (uint64_t)data[4] << 32;
-    case 4: h ^= (uint64_t)data[3] << 24;
-    case 3: h ^= (uint64_t)data[2] << 16;
-    case 2: h ^= (uint64_t)data[1] << 8;
+    case 7: h ^= (uint64_t)data[6] << 48; /* fall-thru */
+    case 6: h ^= (uint64_t)data[5] << 40; /* fall-thru */
+    case 5: h ^= (uint64_t)data[4] << 32; /* fall-thru */
+    case 4: h ^= (uint64_t)data[3] << 24; /* fall-thru */
+    case 3: h ^= (uint64_t)data[2] << 16; /* fall-thru */
+    case 2: h ^= (uint64_t)data[1] << 8; /* fall-thru */
     case 1: h ^= (uint64_t)data[0];
-            h *= m;
+            h *= m; /* fall-thru */
     };
 
     h ^= h >> r;

src/lzf_d.c

@@ -86,6 +86,8 @@ lzf_decompress (const void *const in_data,  unsigned int in_len,
 #ifdef lzf_movsb
           lzf_movsb (op, ip, ctrl);
 #else
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wimplicit-fallthrough"
           switch (ctrl)
             {
               case 32: *op++ = *ip++; case 31: *op++ = *ip++; case 30: *op++ = *ip++; case 29: *op++ = *ip++;
@@ -97,6 +99,7 @@ lzf_decompress (const void *const in_data,  unsigned int in_len,
               case  8: *op++ = *ip++; case  7: *op++ = *ip++; case  6: *op++ = *ip++; case  5: *op++ = *ip++;
               case  4: *op++ = *ip++; case  3: *op++ = *ip++; case  2: *op++ = *ip++; case  1: *op++ = *ip++;
             }
+#pragma GCC diagnostic pop
 #endif
         }
       else /* back reference */
@@ -163,17 +166,17 @@ lzf_decompress (const void *const in_data,  unsigned int in_len,
 
                 break;
 
-              case 9: *op++ = *ref++;
-              case 8: *op++ = *ref++;
-              case 7: *op++ = *ref++;
-              case 6: *op++ = *ref++;
-              case 5: *op++ = *ref++;
-              case 4: *op++ = *ref++;
-              case 3: *op++ = *ref++;
-              case 2: *op++ = *ref++;
-              case 1: *op++ = *ref++;
+              case 9: *op++ = *ref++; /* fall-thru */
+              case 8: *op++ = *ref++; /* fall-thru */
+              case 7: *op++ = *ref++; /* fall-thru */
+              case 6: *op++ = *ref++; /* fall-thru */
+              case 5: *op++ = *ref++; /* fall-thru */
+              case 4: *op++ = *ref++; /* fall-thru */
+              case 3: *op++ = *ref++; /* fall-thru */
+              case 2: *op++ = *ref++; /* fall-thru */
+              case 1: *op++ = *ref++; /* fall-thru */
               case 0: *op++ = *ref++; /* two octets more */
-                      *op++ = *ref++;
+                      *op++ = *ref++; /* fall-thru */
             }
 #endif
         }

src/object.c

@@ -968,7 +968,7 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
         listRewind(server.clients,&li);
         while((ln = listNext(&li))) {
             client *c = listNodeValue(ln);
-            if (c->flags & CLIENT_SLAVE)
+            if (c->flags & CLIENT_SLAVE && !(c->flags & CLIENT_MONITOR))
                 continue;
             mem += getClientOutputBufferMemoryUsage(c);
             mem += sdsAllocSize(c->querybuf);
@@ -1161,10 +1161,10 @@ void objectCommand(client *c) {
 
     if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
         const char *help[] = {
-"encoding <key> -- Return the kind of internal representation used in order to store the value associated with a key.",
-"freq <key> -- Return the access frequency index of the key. The returned integer is proportional to the logarithm of the recent access frequency of the key.",
-"idletime <key> -- Return the idle time of the key, that is the approximated number of seconds elapsed since the last access to the key.",
-"refcount <key> -- Return the number of references of the value associated with the specified key.",
+"ENCODING <key> -- Return the kind of internal representation used in order to store the value associated with a key.",
+"FREQ <key> -- Return the access frequency index of the key. The returned integer is proportional to the logarithm of the recent access frequency of the key.",
+"IDLETIME <key> -- Return the idle time of the key, that is the approximated number of seconds elapsed since the last access to the key.",
+"REFCOUNT <key> -- Return the number of references of the value associated with the specified key.",
 NULL
         };
         addReplyHelp(c, help);

src/rdb.c

@@ -100,6 +100,9 @@ int rdbLoadType(rio *rdb) {
     return type;
 }
 
+/* This is only used to load old databases stored with the RDB_OPCODE_EXPIRETIME
+ * opcode. New versions of Redis store using the RDB_OPCODE_EXPIRETIME_MS
+ * opcode. */
 time_t rdbLoadTime(rio *rdb) {
     int32_t t32;
     rdbLoadRaw(rdb,&t32,4);
@@ -108,12 +111,26 @@ time_t rdbLoadTime(rio *rdb) {
 
 int rdbSaveMillisecondTime(rio *rdb, long long t) {
     int64_t t64 = (int64_t) t;
+    memrev64ifbe(&t64); /* Store in little endian. */
     return rdbWriteRaw(rdb,&t64,8);
 }
 
-long long rdbLoadMillisecondTime(rio *rdb) {
+/* This function loads a time from the RDB file. It gets the version of the
+ * RDB because, unfortunately, before Redis 5 (RDB version 9), the function
+ * failed to convert data to/from little endian, so RDB files with keys having
+ * expires could not be shared between big endian and little endian systems
+ * (because the expire time will be totally wrong). The fix for this is just
+ * to call memrev64ifbe(), however if we fix this for all the RDB versions,
+ * this call will introduce an incompatibility for big endian systems:
+ * after upgrading to Redis version 5 they will no longer be able to load their
+ * own old RDB files. Because of that, we instead fix the function only for new
+ * RDB versions, and load older RDB versions as we used to do in the past,
+ * allowing big endian systems to load their own old RDB files. */
+long long rdbLoadMillisecondTime(rio *rdb, int rdbver) {
     int64_t t64;
     rdbLoadRaw(rdb,&t64,8);
+    if (rdbver >= 9) /* Check the top comment of this function. */
+        memrev64ifbe(&t64); /* Convert in big endian if the system is BE. */
     return (long long)t64;
 }
 
@@ -988,8 +1005,7 @@ size_t rdbSavedObjectLen(robj *o) {
  * On error -1 is returned.
  * On success if the key was actually saved 1 is returned, otherwise 0
  * is returned (the key was already expired). */
-int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime)
-{
+int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime) {
     int savelru = server.maxmemory_policy & MAXMEMORY_FLAG_LRU;
     int savelfu = server.maxmemory_policy & MAXMEMORY_FLAG_LFU;
 
@@ -1001,7 +1017,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime)
 
     /* Save the LRU info. */
     if (savelru) {
-        int idletime = estimateObjectIdleTime(val);
+        uint64_t idletime = estimateObjectIdleTime(val);
         idletime /= 1000; /* Using seconds is enough and requires less space.*/
         if (rdbSaveType(rdb,RDB_OPCODE_IDLE) == -1) return -1;
         if (rdbSaveLen(rdb,idletime) == -1) return -1;
@@ -1441,6 +1457,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
         o = createZsetObject();
         zs = o->ptr;
 
+        if (zsetlen > DICT_HT_INITIAL_SIZE)
+            dictExpand(zs->dict,zsetlen);
+
         /* Load every single element of the sorted set. */
         while(zsetlen--) {
             sds sdsele;
@@ -1509,6 +1528,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
             sdsfree(value);
         }
 
+        if (o->encoding == OBJ_ENCODING_HT && len > DICT_HT_INITIAL_SIZE)
+            dictExpand(o->ptr,len);
+
         /* Load remaining fields and values into the hash table */
         while (o->encoding == OBJ_ENCODING_HT && len > 0) {
             len--;
@@ -1683,7 +1705,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
                 unsigned char rawid[sizeof(streamID)];
                 rdbLoadRaw(rdb,rawid,sizeof(rawid));
                 streamNACK *nack = streamCreateNACK(NULL);
-                nack->delivery_time = rdbLoadMillisecondTime(rdb);
+                nack->delivery_time = rdbLoadMillisecondTime(rdb,RDB_VERSION);
                 nack->delivery_count = rdbLoadLen(rdb,NULL);
                 if (!raxInsert(cgroup->pel,rawid,sizeof(rawid),nack,NULL))
                     rdbExitReportCorruptRDB("Duplicated gobal PEL entry "
@@ -1702,7 +1724,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
                 streamConsumer *consumer = streamLookupConsumer(cgroup,cname,
                                            1);
                 sdsfree(cname);
-                consumer->seen_time = rdbLoadMillisecondTime(rdb);
+                consumer->seen_time = rdbLoadMillisecondTime(rdb,RDB_VERSION);
 
                 /* Load the PEL about entries owned by this specific
                  * consumer. */
@@ -1867,7 +1889,7 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
         } else if (type == RDB_OPCODE_EXPIRETIME_MS) {
             /* EXPIRETIME_MS: milliseconds precision expire times introduced
              * with RDB v3. Like EXPIRETIME but no with more precision. */
-            expiretime = rdbLoadMillisecondTime(rdb);
+            expiretime = rdbLoadMillisecondTime(rdb,rdbver);
             continue; /* Read next opcode. */
         } else if (type == RDB_OPCODE_FREQ) {
             /* FREQ: LFU frequency. */

src/redis-cli.c

@@ -258,20 +258,25 @@ static long long mstime(void) {
 }
 
 static void cliRefreshPrompt(void) {
-    int len;
-
     if (config.eval_ldb) return;
-    if (config.hostsocket != NULL)
-        len = snprintf(config.prompt,sizeof(config.prompt),"redis %s",
-                       config.hostsocket);
-    else
-        len = anetFormatAddr(config.prompt, sizeof(config.prompt),
-                           config.hostip, config.hostport);
+
+    sds prompt = sdsempty();
+    if (config.hostsocket != NULL) {
+        prompt = sdscatfmt(prompt,"redis %s",config.hostsocket);
+    } else {
+        char addr[256];
+        anetFormatAddr(addr, sizeof(addr), config.hostip, config.hostport);
+        prompt = sdscatlen(prompt,addr,strlen(addr));
+    }
+
     /* Add [dbnum] if needed */
     if (config.dbnum != 0)
-        len += snprintf(config.prompt+len,sizeof(config.prompt)-len,"[%d]",
-            config.dbnum);
-    snprintf(config.prompt+len,sizeof(config.prompt)-len,"> ");
+        prompt = sdscatfmt(prompt,"[%i]",config.dbnum);
+
+    /* Copy the prompt in the static buffer. */
+    prompt = sdscatlen(prompt,"> ",2);
+    snprintf(config.prompt,sizeof(config.prompt),"%s",prompt);
+    sdsfree(prompt);
 }
 
 /* Return the name of the dotfile for the specified 'dotfilename'.
@@ -1075,6 +1080,8 @@ static int cliSendCommand(int argc, char **argv, long repeat) {
     if (!strcasecmp(command,"info") ||
         (argc >= 2 && !strcasecmp(command,"debug") &&
                        !strcasecmp(argv[1],"htstats")) ||
+        (argc >= 2 && !strcasecmp(command,"debug") &&
+                       !strcasecmp(argv[1],"htstats-key")) ||
         (argc >= 2 && !strcasecmp(command,"memory") &&
                       (!strcasecmp(argv[1],"malloc-stats") ||
                        !strcasecmp(argv[1],"doctor"))) ||
@@ -3088,7 +3095,7 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
                 currentNode->flags |= CLUSTER_MANAGER_FLAG_FAIL;
             else if (strcmp(flag, "slave") == 0) {
                 currentNode->flags |= CLUSTER_MANAGER_FLAG_SLAVE;
-                if (master_id == 0) {
+                if (master_id != NULL) {
                     if (currentNode->replicate) sdsfree(currentNode->replicate);
                     currentNode->replicate = sdsnew(master_id);
                 }
@@ -4887,7 +4894,7 @@ static int clusterManagerCommandRebalance(int argc, char **argv) {
         listRewind(involved, &li);
         while ((ln = listNext(&li)) != NULL) {
             clusterManagerNode *n = ln->value;
-            if (n->balance < 0 && total_balance > 0) {
+            if (n->balance <= 0 && total_balance > 0) {
                 n->balance--;
                 total_balance--;
             }
@@ -5091,7 +5098,7 @@ static int clusterManagerCommandImport(int argc, char **argv) {
 
     // Build a slot -> node map
     clusterManagerNode  *slots_map[CLUSTER_MANAGER_SLOTS];
-    memset(slots_map, 0, sizeof(slots_map) / sizeof(clusterManagerNode *));
+    memset(slots_map, 0, sizeof(slots_map));
     listIter li;
     listNode *ln;
     for (i = 0; i < CLUSTER_MANAGER_SLOTS; i++) {
@@ -5575,7 +5582,7 @@ static void getRDB(void) {
         nwritten = write(fd, buf, nread);
         if (nwritten != nread) {
             fprintf(stderr,"Error writing data to file: %s\n",
-                strerror(errno));
+                (nwritten == -1) ? strerror(errno) : "short write");
             exit(1);
         }
         payload -= nread;

src/replication.c

@@ -1105,7 +1105,7 @@ void restartAOF() {
 #define REPL_MAX_WRITTEN_BEFORE_FSYNC (1024*1024*8) /* 8 MB */
 void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
     char buf[4096];
-    ssize_t nread, readlen;
+    ssize_t nread, readlen, nwritten;
     off_t left;
     UNUSED(el);
     UNUSED(privdata);
@@ -1206,8 +1206,9 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
     }
 
     server.repl_transfer_lastio = server.unixtime;
-    if (write(server.repl_transfer_fd,buf,nread) != nread) {
-        serverLog(LL_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchronization: %s", strerror(errno));
+    if ((nwritten = write(server.repl_transfer_fd,buf,nread)) != nread) {
+        serverLog(LL_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchronization: %s", 
+            (nwritten == -1) ? strerror(errno) : "short write");
         goto error;
     }
     server.repl_transfer_read += nread;

src/server.c

@@ -845,13 +845,14 @@ int clientsCronResizeQueryBuffer(client *c) {
 
     /* There are two conditions to resize the query buffer:
      * 1) Query buffer is > BIG_ARG and too big for latest peak.
-     * 2) Client is inactive and the buffer is bigger than 1k. */
-    if (((querybuf_size > PROTO_MBULK_BIG_ARG) &&
-         (querybuf_size/(c->querybuf_peak+1)) > 2) ||
-         (querybuf_size > 1024 && idletime > 2))
+     * 2) Query buffer is > BIG_ARG and client is idle. */
+    if (querybuf_size > PROTO_MBULK_BIG_ARG &&
+         ((querybuf_size/(c->querybuf_peak+1)) > 2 ||
+          idletime > 2))
     {
-        /* Only resize the query buffer if it is actually wasting space. */
-        if (sdsavail(c->querybuf) > 1024) {
+        /* Only resize the query buffer if it is actually wasting
+         * at least a few kbytes. */
+        if (sdsavail(c->querybuf) > 1024*4) {
             c->querybuf = sdsRemoveFreeSpace(c->querybuf);
         }
     }
@@ -1485,6 +1486,8 @@ void initServerConfig(void) {
     server.zset_max_ziplist_entries = OBJ_ZSET_MAX_ZIPLIST_ENTRIES;
     server.zset_max_ziplist_value = OBJ_ZSET_MAX_ZIPLIST_VALUE;
     server.hll_sparse_max_bytes = CONFIG_DEFAULT_HLL_SPARSE_MAX_BYTES;
+    server.stream_node_max_bytes = OBJ_STREAM_NODE_MAX_BYTES;
+    server.stream_node_max_entries = OBJ_STREAM_NODE_MAX_ENTRIES;
     server.shutdown_asap = 0;
     server.cluster_enabled = 0;
     server.cluster_node_timeout = CLUSTER_DEFAULT_NODE_TIMEOUT;

src/server.h

@@ -339,7 +339,7 @@ typedef long long mstime_t; /* millisecond time type. */
 /* Anti-warning macro... */
 #define UNUSED(V) ((void) V)
 
-#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^32 elements */
+#define ZSKIPLIST_MAXLEVEL 64 /* Should be enough for 2^64 elements */
 #define ZSKIPLIST_P 0.25      /* Skiplist P = 1/4 */
 
 /* Append only defines */
@@ -348,12 +348,14 @@ typedef long long mstime_t; /* millisecond time type. */
 #define AOF_FSYNC_EVERYSEC 2
 #define CONFIG_DEFAULT_AOF_FSYNC AOF_FSYNC_EVERYSEC
 
-/* Zip structure related defaults */
+/* Zipped structures related defaults */
 #define OBJ_HASH_MAX_ZIPLIST_ENTRIES 512
 #define OBJ_HASH_MAX_ZIPLIST_VALUE 64
 #define OBJ_SET_MAX_INTSET_ENTRIES 512
 #define OBJ_ZSET_MAX_ZIPLIST_ENTRIES 128
 #define OBJ_ZSET_MAX_ZIPLIST_VALUE 64
+#define OBJ_STREAM_NODE_MAX_BYTES 4096
+#define OBJ_STREAM_NODE_MAX_ENTRIES 100
 
 /* List defaults */
 #define OBJ_LIST_MAX_ZIPLIST_SIZE -2
@@ -780,7 +782,7 @@ typedef struct zskiplistNode {
     struct zskiplistNode *backward;
     struct zskiplistLevel {
         struct zskiplistNode *forward;
-        unsigned int span;
+        unsigned long span;
     } level[];
 } zskiplistNode;
 
@@ -1177,6 +1179,8 @@ struct redisServer {
     size_t zset_max_ziplist_entries;
     size_t zset_max_ziplist_value;
     size_t hll_sparse_max_bytes;
+    size_t stream_node_max_bytes;
+    int64_t stream_node_max_entries;
     /* List parameters */
     int list_max_ziplist_size;
     int list_compress_depth;
@@ -1624,7 +1628,7 @@ void zzlNext(unsigned char *zl, unsigned char **eptr, unsigned char **sptr);
 void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr);
 unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec *range);
 unsigned char *zzlLastInRange(unsigned char *zl, zrangespec *range);
-unsigned int zsetLength(const robj *zobj);
+unsigned long zsetLength(const robj *zobj);
 void zsetConvert(robj *zobj, int encoding);
 void zsetConvertToZiplistIfNeeded(robj *zobj, size_t maxelelen);
 int zsetScore(robj *zobj, sds member, double *score);

src/siphash.c

@@ -142,12 +142,12 @@ uint64_t siphash(const uint8_t *in, const size_t inlen, const uint8_t *k) {
     }
 
     switch (left) {
-    case 7: b |= ((uint64_t)in[6]) << 48;
-    case 6: b |= ((uint64_t)in[5]) << 40;
-    case 5: b |= ((uint64_t)in[4]) << 32;
-    case 4: b |= ((uint64_t)in[3]) << 24;
-    case 3: b |= ((uint64_t)in[2]) << 16;
-    case 2: b |= ((uint64_t)in[1]) << 8;
+    case 7: b |= ((uint64_t)in[6]) << 48; /* fall-thru */
+    case 6: b |= ((uint64_t)in[5]) << 40; /* fall-thru */
+    case 5: b |= ((uint64_t)in[4]) << 32; /* fall-thru */
+    case 4: b |= ((uint64_t)in[3]) << 24; /* fall-thru */
+    case 3: b |= ((uint64_t)in[2]) << 16; /* fall-thru */
+    case 2: b |= ((uint64_t)in[1]) << 8; /* fall-thru */
     case 1: b |= ((uint64_t)in[0]); break;
     case 0: break;
     }
@@ -202,12 +202,12 @@ uint64_t siphash_nocase(const uint8_t *in, const size_t inlen, const uint8_t *k)
     }
 
     switch (left) {
-    case 7: b |= ((uint64_t)siptlw(in[6])) << 48;
-    case 6: b |= ((uint64_t)siptlw(in[5])) << 40;
-    case 5: b |= ((uint64_t)siptlw(in[4])) << 32;
-    case 4: b |= ((uint64_t)siptlw(in[3])) << 24;
-    case 3: b |= ((uint64_t)siptlw(in[2])) << 16;
-    case 2: b |= ((uint64_t)siptlw(in[1])) << 8;
+    case 7: b |= ((uint64_t)siptlw(in[6])) << 48; /* fall-thru */
+    case 6: b |= ((uint64_t)siptlw(in[5])) << 40; /* fall-thru */
+    case 5: b |= ((uint64_t)siptlw(in[4])) << 32; /* fall-thru */
+    case 4: b |= ((uint64_t)siptlw(in[3])) << 24; /* fall-thru */
+    case 3: b |= ((uint64_t)siptlw(in[2])) << 16; /* fall-thru */
+    case 2: b |= ((uint64_t)siptlw(in[1])) << 8; /* fall-thru */
     case 1: b |= ((uint64_t)siptlw(in[0])); break;
     case 0: break;
     }

src/t_stream.c

@@ -41,6 +41,7 @@
 #define STREAM_ITEM_FLAG_SAMEFIELDS (1<<1)  /* Same fields as master entry. */
 
 void streamFreeCG(streamCG *cg);
+void streamFreeNACK(streamNACK *na);
 size_t streamReplyWithRangeFromConsumerPEL(client *c, stream *s, streamID *start, streamID *end, size_t count, streamConsumer *consumer);
 
 /* -----------------------------------------------------------------------
@@ -171,7 +172,7 @@ int streamCompareID(streamID *a, streamID *b) {
  * if the ID was generated by the function. However the function may return
  * C_ERR if an ID was given via 'use_id', but adding it failed since the
  * current top ID is greater or equal. */
-int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id, streamID *use_id) {
+int streamAppendItem(stream *s, robj **argv, int64_t numfields, streamID *added_id, streamID *use_id) {
     /* If an ID was given, check that it's greater than the last entry ID
      * or return an error. */
     if (use_id && streamCompareID(use_id,&s->last_id) <= 0) return C_ERR;
@@ -221,7 +222,7 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
      * +-------+---------+------------+---------+--/--+---------+---------+-+
      *
      * count and deleted just represent respectively the total number of
-     * entires inside the listpack that are valid, and marked as deleted
+     * entries inside the listpack that are valid, and marked as deleted
      * (delted flag in the entry flags set). So the total number of items
      * actually inside the listpack (both deleted and not) is count+deleted.
      *
@@ -234,10 +235,24 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
      *
      * The "0" entry at the end is the same as the 'lp-count' entry in the
      * regular stream entries (see below), and marks the fact that there are
-     * no more entires, when we scan the stream from right to left. */
+     * no more entries, when we scan the stream from right to left. */
+
+    /* First of all, check if we can append to the current macro node or
+     * if we need to switch to the next one. 'lp' will be set to NULL if
+     * the current node is full. */
+    if (lp != NULL) {
+        if (server.stream_node_max_bytes &&
+            lp_bytes > server.stream_node_max_bytes)
+        {
+            lp = NULL;
+        } else if (server.stream_node_max_entries) {
+            int64_t count = lpGetInteger(lpFirst(lp));
+            if (count > server.stream_node_max_entries) lp = NULL;
+        }
+    }
 
     int flags = STREAM_ITEM_FLAG_NONE;
-    if (lp == NULL || lp_bytes > STREAM_BYTES_PER_LISTPACK) {
+    if (lp == NULL || lp_bytes > server.stream_node_max_bytes) {
         master_id = id;
         streamEncodeID(rax_key,&id);
         /* Create the listpack having the master entry ID and fields. */
@@ -245,7 +260,7 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
         lp = lpAppendInteger(lp,1); /* One item, the one we are adding. */
         lp = lpAppendInteger(lp,0); /* Zero deleted so far. */
         lp = lpAppendInteger(lp,numfields);
-        for (int i = 0; i < numfields; i++) {
+        for (int64_t i = 0; i < numfields; i++) {
             sds field = argv[i*2]->ptr;
             lp = lpAppend(lp,(unsigned char*)field,sdslen(field));
         }
@@ -270,10 +285,10 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
 
         /* Check if the entry we are adding, have the same fields
          * as the master entry. */
-        int master_fields_count = lpGetInteger(lp_ele);
+        int64_t master_fields_count = lpGetInteger(lp_ele);
         lp_ele = lpNext(lp,lp_ele);
         if (numfields == master_fields_count) {
-            int i;
+            int64_t i;
             for (i = 0; i < master_fields_count; i++) {
                 sds field = argv[i*2]->ptr;
                 int64_t e_len;
@@ -317,14 +332,14 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
     lp = lpAppendInteger(lp,id.seq - master_id.seq);
     if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS))
         lp = lpAppendInteger(lp,numfields);
-    for (int i = 0; i < numfields; i++) {
+    for (int64_t i = 0; i < numfields; i++) {
         sds field = argv[i*2]->ptr, value = argv[i*2+1]->ptr;
         if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS))
             lp = lpAppend(lp,(unsigned char*)field,sdslen(field));
         lp = lpAppend(lp,(unsigned char*)value,sdslen(value));
     }
     /* Compute and store the lp-count field. */
-    int lp_count = numfields;
+    int64_t lp_count = numfields;
     lp_count += 3; /* Add the 3 fixed fields flags + ms-diff + seq-diff. */
     if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS)) {
         /* If the item is not compressed, it also has the fields other than
@@ -564,7 +579,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
                 /* If we are going backward, read the number of elements this
                  * entry is composed of, and jump backward N times to seek
                  * its start. */
-                int lp_count = lpGetInteger(si->lp_ele);
+                int64_t lp_count = lpGetInteger(si->lp_ele);
                 if (lp_count == 0) { /* We reached the master entry. */
                     si->lp = NULL;
                     si->lp_ele = NULL;
@@ -627,12 +642,17 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
              * forward, or seek the previous entry if we are going
              * backward. */
             if (!si->rev) {
-                int to_discard = (flags & STREAM_ITEM_FLAG_SAMEFIELDS) ?
-                                    *numfields : *numfields*2;
+                int64_t to_discard = (flags & STREAM_ITEM_FLAG_SAMEFIELDS) ?
+                                      *numfields : *numfields*2;
                 for (int64_t i = 0; i < to_discard; i++)
                     si->lp_ele = lpNext(si->lp,si->lp_ele);
             } else {
-                int prev_times = 4; /* flag + id ms/seq diff + numfields. */
+                int64_t prev_times = 4; /* flag + id ms + id seq + one more to
+                                           go back to the previous entry "count"
+                                           field. */
+                /* If the entry was not flagged SAMEFIELD we also read the
+                 * number of fields, so go back one more. */
+                if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS)) prev_times++;
                 while(prev_times--) si->lp_ele = lpPrev(si->lp,si->lp_ele);
             }
         }
@@ -690,6 +710,9 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
     aux = lpGetInteger(p);
     lp = lpReplaceInteger(lp,&p,aux+1);
 
+    /* Update the number of entries counter. */
+    si->stream->length--;
+
     /* Re-seek the iterator to fix the now messed up state. */
     streamID start, end;
     if (si->rev) {
@@ -814,7 +837,7 @@ void streamPropagateXCLAIM(client *c, robj *key, robj *group, robj *id, streamNA
  * Note that this function is recursive in certian cases. When it's called
  * with a non NULL group and consumer argument, it may call
  * streamReplyWithRangeFromConsumerPEL() in order to get entries from the
- * consumer pending entires list. However such a function will then call
+ * consumer pending entries list. However such a function will then call
  * streamReplyWithRange() in order to emit single entries (found in the
  * PEL by ID) to the client. This is the use case for the STREAM_RWR_RAWENTRIES
  * flag.
@@ -867,18 +890,41 @@ size_t streamReplyWithRange(client *c, stream *s, streamID *start, streamID *end
 
         /* If a group is passed, we need to create an entry in the
          * PEL (pending entries list) of this group *and* this consumer.
-         * Note that we are sure about the fact the message is not already
-         * associated with some other consumer, because if we reached this
-         * loop the IDs the user is requesting are greater than any message
-         * delivered for this group. */
+         *
+         * Note that we cannot be sure about the fact the message is not
+         * already owned by another consumer, because the admin is able
+         * to change the consumer group last delivered ID using the
+         * XGROUP SETID command. So if we find that there is already
+         * a NACK for the entry, we need to associate it to the new
+         * consumer. */
         if (group && !(flags & STREAM_RWR_NOACK)) {
             unsigned char buf[sizeof(streamID)];
             streamEncodeID(buf,&id);
+
+            /* Try to add a new NACK. Most of the time this will work and
+             * will not require extra lookups. We'll fix the problem later
+             * if we find that there is already a entry for this ID. */
             streamNACK *nack = streamCreateNACK(consumer);
             int retval = 0;
-            retval += raxInsert(group->pel,buf,sizeof(buf),nack,NULL);
-            retval += raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
-            serverAssert(retval == 2); /* Make sure entry was inserted. */
+            retval += raxTryInsert(group->pel,buf,sizeof(buf),nack,NULL);
+            retval += raxTryInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
+
+            /* Now we can check if the entry was already busy, and
+             * in that case reassign the entry to the new consumer. */
+            if (retval == 0) {
+                streamFreeNACK(nack);
+                nack = raxFind(group->pel,buf,sizeof(buf));
+                serverAssert(nack != raxNotFound);
+                raxRemove(nack->consumer->pel,buf,sizeof(buf),NULL);
+                /* Update the consumer and idle time. */
+                nack->consumer = consumer;
+                nack->delivery_time = mstime();
+                nack->delivery_count++;
+                /* Add the entry in the new consumer local PEL. */
+                raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
+            } else if (retval == 1) {
+                serverPanic("NACK half-created. Should not be possible.");
+            }
 
             /* Propagate as XCLAIM. */
             if (spi) {
@@ -1269,14 +1315,13 @@ void xreadCommand(client *c) {
          * starting from now. */
         int id_idx = i - streams_arg - streams_count;
         robj *key = c->argv[i-streams_count];
-        robj *o;
+        robj *o = lookupKeyRead(c->db,key);
+        if (o && checkType(c,o,OBJ_STREAM)) goto cleanup;
         streamCG *group = NULL;
 
         /* If a group was specified, than we need to be sure that the
          * key and group actually exist. */
         if (groupname) {
-            o = lookupKeyRead(c->db,key);
-            if (o && checkType(c,o,OBJ_STREAM)) goto cleanup;
             if (o == NULL ||
                 (group = streamLookupCG(o->ptr,groupname->ptr)) == NULL)
             {
@@ -1290,8 +1335,6 @@ void xreadCommand(client *c) {
         }
 
         if (strcmp(c->argv[i]->ptr,"$") == 0) {
-            o = lookupKeyRead(c->db,key);
-            if (o && checkType(c,o,OBJ_STREAM)) goto cleanup;
             if (o) {
                 stream *s = o->ptr;
                 ids[id_idx] = s->last_id;
@@ -1336,7 +1379,7 @@ void xreadCommand(client *c) {
             /* Emit the two elements sub-array consisting of the name
              * of the stream and the data we extracted from it. */
             addReplyMultiBulkLen(c,2);
-            addReplyBulk(c,c->argv[i+streams_arg]);
+            addReplyBulk(c,c->argv[streams_arg+i]);
             streamConsumer *consumer = NULL;
             if (groups) consumer = streamLookupConsumer(groups[i],
                                                         consumername->ptr,1);
@@ -1516,13 +1559,13 @@ uint64_t streamDelConsumer(streamCG *cg, sds name) {
 
 /* XGROUP CREATE <key> <groupname> <id or $>
  * XGROUP SETID <key> <id or $>
- * XGROUP DELGROUP <key> <groupname>
+ * XGROUP DESTROY <key> <groupname>
  * XGROUP DELCONSUMER <key> <groupname> <consumername> */
 void xgroupCommand(client *c) {
     const char *help[] = {
 "CREATE      <key> <groupname> <id or $>  -- Create a new consumer group.",
 "SETID       <key> <groupname> <id or $>  -- Set the current group ID.",
-"DELGROUP    <key> <groupname>            -- Remove the specified group.",
+"DESTROY     <key> <groupname>            -- Remove the specified group.",
 "DELCONSUMER <key> <groupname> <consumer> -- Remove the specified conusmer.",
 "HELP                                     -- Prints this help.",
 NULL
@@ -1535,14 +1578,13 @@ NULL
     /* Lookup the key now, this is common for all the subcommands but HELP. */
     if (c->argc >= 4) {
         robj *o = lookupKeyWriteOrReply(c,c->argv[2],shared.nokeyerr);
-        if (o == NULL) return;
+        if (o == NULL || checkType(c,o,OBJ_STREAM)) return;
         s = o->ptr;
         grpname = c->argv[3]->ptr;
 
         /* Certain subcommands require the group to exist. */
         if ((cg = streamLookupCG(s,grpname)) == NULL &&
             (!strcasecmp(opt,"SETID") ||
-             !strcasecmp(opt,"DELGROUP") ||
              !strcasecmp(opt,"DELCONSUMER")))
         {
             addReplyErrorFormat(c, "-NOGROUP No such consumer group '%s' "
@@ -1564,18 +1606,42 @@ NULL
         if (cg) {
             addReply(c,shared.ok);
             server.dirty++;
+            notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-create",
+                                c->argv[2],c->db->id);
         } else {
             addReplySds(c,
                 sdsnew("-BUSYGROUP Consumer Group name already exists\r\n"));
         }
     } else if (!strcasecmp(opt,"SETID") && c->argc == 5) {
-    } else if (!strcasecmp(opt,"DELGROUP") && c->argc == 4) {
+        streamID id;
+        if (!strcmp(c->argv[4]->ptr,"$")) {
+            id = s->last_id;
+        } else if (streamParseIDOrReply(c,c->argv[4],&id,0) != C_OK) {
+            return;
+        }
+        cg->last_id = id;
+        addReply(c,shared.ok);
+        server.dirty++;
+        notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-setid",c->argv[2],c->db->id);
+    } else if (!strcasecmp(opt,"DESTROY") && c->argc == 4) {
+        if (cg) {
+            raxRemove(s->cgroups,(unsigned char*)grpname,sdslen(grpname),NULL);
+            streamFreeCG(cg);
+            addReply(c,shared.cone);
+            server.dirty++;
+            notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-destroy",
+                                c->argv[2],c->db->id);
+        } else {
+            addReply(c,shared.czero);
+        }
     } else if (!strcasecmp(opt,"DELCONSUMER") && c->argc == 5) {
         /* Delete the consumer and returns the number of pending messages
          * that were yet associated with such a consumer. */
         long long pending = streamDelConsumer(cg,c->argv[4]->ptr);
         addReplyLongLong(c,pending);
         server.dirty++;
+        notifyKeyspaceEvent(NOTIFY_STREAM,"xgroup-delconsumer",
+                            c->argv[2],c->db->id);
     } else if (!strcasecmp(opt,"HELP")) {
         addReplyHelp(c, help);
     } else {
@@ -1728,8 +1794,10 @@ void xpendingCommand(client *c) {
 
         /* If a consumer name was mentioned but it does not exist, we can
          * just return an empty array. */
-        if (consumername && consumer == NULL)
+        if (consumername && consumer == NULL) {
             addReplyMultiBulkLen(c,0);
+            return;
+        }
 
         rax *pel = consumer ? consumer->pel : group->pel;
         unsigned char startkey[sizeof(streamID)];
@@ -1785,7 +1853,7 @@ void xpendingCommand(client *c) {
  * becomes the specified <consumer>. If the minimum idle time specified
  * is zero, messages are claimed regardless of their idle time.
  *
- * All the messages that cannot be found inside the pending entires list
+ * All the messages that cannot be found inside the pending entries list
  * are ignored, but in case the FORCE option is used. In that case we
  * create the NACK (representing a not yet acknowledged message) entry in
  * the consumer group PEL.
@@ -1970,7 +2038,7 @@ void xclaimCommand(client *c) {
             nack->delivery_time = deliverytime;
             /* Set the delivery attempts counter if given. */
             if (retrycount >= 0) nack->delivery_count = retrycount;
-            /* Add the entry in the new cosnumer local PEL. */
+            /* Add the entry in the new consumer local PEL. */
             raxInsert(consumer->pel,buf,sizeof(buf),nack,NULL);
             /* Send the reply for this entry. */
             if (justid) {
@@ -1999,7 +2067,7 @@ void xclaimCommand(client *c) {
 void xdelCommand(client *c) {
     robj *o;
 
-    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL
+    if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL
         || checkType(c,o,OBJ_STREAM)) return;
     stream *s = o->ptr;
 
@@ -2040,7 +2108,7 @@ void xtrimCommand(client *c) {
 
     /* If the key does not exist, we are ok returning zero, that is, the
      * number of elements removed from the stream. */
-    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL
+    if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL
         || checkType(c,o,OBJ_STREAM)) return;
     stream *s = o->ptr;
 
@@ -2093,14 +2161,12 @@ void xtrimCommand(client *c) {
 /* XINFO CONSUMERS key group
  * XINFO GROUPS <key>
  * XINFO STREAM <key>
- * XINFO <key> (alias of XINFO STREAM key)
  * XINFO HELP. */
 void xinfoCommand(client *c) {
     const char *help[] = {
 "CONSUMERS <key> <groupname>  -- Show consumer groups of group <groupname>.",
 "GROUPS <key>                 -- Show the stream consumer groups.",
 "STREAM <key>                 -- Show information about the stream.",
-"<key>                        -- Alias for STREAM <key>.",
 "HELP                         -- Print this help.",
 NULL
     };
@@ -2112,20 +2178,19 @@ NULL
     if (!strcasecmp(c->argv[1]->ptr,"HELP")) {
         addReplyHelp(c, help);
         return;
+    } else if (c->argc < 3) {
+        addReplyError(c,"syntax error, try 'XINFO HELP'");
+        return;
     }
 
-    /* Handle the fact that no subcommand means "STREAM". */
-    if (c->argc == 2) {
-        opt = "STREAM";
-        key = c->argv[1];
-    } else {
-        opt = c->argv[1]->ptr;
-        key = c->argv[2];
-    }
+    /* With the exception of HELP handled before any other sub commands, all
+     * the ones are in the form of "<subcommand> <key>". */
+    opt = c->argv[1]->ptr;
+    key = c->argv[2];
 
     /* Lookup the key now, this is common for all the subcommands but HELP. */
     robj *o = lookupKeyWriteOrReply(c,key,shared.nokeyerr);
-    if (o == NULL) return;
+    if (o == NULL || checkType(c,o,OBJ_STREAM)) return;
     s = o->ptr;
 
     /* Dispatch the different subcommands. */
@@ -2180,9 +2245,7 @@ NULL
             addReplyLongLong(c,raxSize(cg->pel));
         }
         raxStop(&ri);
-    } else if (c->argc == 2 ||
-               (!strcasecmp(opt,"STREAM") && c->argc == 3))
-    {
+    } else if (!strcasecmp(opt,"STREAM") && c->argc == 3) {
         /* XINFO STREAM <key> (or the alias XINFO <key>). */
         addReplyMultiBulkLen(c,12);
         addReplyStatus(c,"length");

src/t_zset.c

@@ -1100,8 +1100,8 @@ unsigned char *zzlDeleteRangeByRank(unsigned char *zl, unsigned int start, unsig
  * Common sorted set API
  *----------------------------------------------------------------------------*/
 
-unsigned int zsetLength(const robj *zobj) {
-    int length = -1;
+unsigned long zsetLength(const robj *zobj) {
+    unsigned long length = 0;
     if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
         length = zzlLength(zobj->ptr);
     } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
@@ -1878,7 +1878,7 @@ void zuiClearIterator(zsetopsrc *op) {
     }
 }
 
-int zuiLength(zsetopsrc *op) {
+unsigned long zuiLength(zsetopsrc *op) {
     if (op->subject == NULL)
         return 0;
 
@@ -2085,7 +2085,11 @@ int zuiFind(zsetopsrc *op, zsetopval *val, double *score) {
 }
 
 int zuiCompareByCardinality(const void *s1, const void *s2) {
-    return zuiLength((zsetopsrc*)s1) - zuiLength((zsetopsrc*)s2);
+    unsigned long first = zuiLength((zsetopsrc*)s1);
+    unsigned long second = zuiLength((zsetopsrc*)s2);
+    if (first > second) return 1;
+    if (first < second) return -1;
+    return 0;
 }
 
 #define REDIS_AGGR_SUM 1
@@ -2129,7 +2133,7 @@ void zunionInterGenericCommand(client *c, robj *dstkey, int op) {
     zsetopsrc *src;
     zsetopval zval;
     sds tmp;
-    unsigned int maxelelen = 0;
+    size_t maxelelen = 0;
     robj *dstobj;
     zset *dstzset;
     zskiplistNode *znode;
@@ -2363,8 +2367,8 @@ void zrangeGenericCommand(client *c, int reverse) {
     int withscores = 0;
     long start;
     long end;
-    int llen;
-    int rangelen;
+    long llen;
+    long rangelen;
 
     if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != C_OK) ||
         (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != C_OK)) return;
@@ -2671,7 +2675,7 @@ void zcountCommand(client *c) {
     robj *key = c->argv[1];
     robj *zobj;
     zrangespec range;
-    int count = 0;
+    unsigned long count = 0;
 
     /* Parse the range arguments */
     if (zslParseRange(c->argv[2],c->argv[3],&range) != C_OK) {
@@ -2748,7 +2752,7 @@ void zlexcountCommand(client *c) {
     robj *key = c->argv[1];
     robj *zobj;
     zlexrangespec range;
-    int count = 0;
+    unsigned long count = 0;
 
     /* Parse the range arguments */
     if (zslParseLexRange(c->argv[2],c->argv[3],&range) != C_OK) {
@@ -3163,8 +3167,8 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
             signalModifiedKey(c->db,key);
         }
 
-        addReplyDouble(c,score);
         addReplyBulkCBuffer(c,ele,sdslen(ele));
+        addReplyDouble(c,score);
         sdsfree(ele);
         arraylen += 2;
 
@@ -3216,9 +3220,9 @@ void blockingGenericZpopCommand(client *c, int where) {
                 return;
             } else {
                 if (zsetLength(o) != 0) {
-                    /* Non empty zset, this is like a normal Z[REV]POP. */
+                    /* Non empty zset, this is like a normal ZPOP[MIN|MAX]. */
                     genericZpopCommand(c,&c->argv[j],1,where,1,NULL);
-                    /* Replicate it as an Z[REV]POP instead of BZ[REV]POP. */
+                    /* Replicate it as an ZPOP[MIN|MAX] instead of BZPOP[MIN|MAX]. */
                     rewriteClientCommandVector(c,2,
                         where == ZSET_MAX ? shared.zpopmax : shared.zpopmin,
                         c->argv[j]);

src/version.h

@@ -1 +1 @@
-#define REDIS_VERSION "999.999.999"
+#define REDIS_VERSION "4.9.102"

src/ziplist.c

@@ -27,7 +27,7 @@
  * traversal.
  *
  * <uint16_t zllen> is the number of entries. When there are more than
- * 2^16-2 entires, this value is set to 2^16-1 and we need to traverse the
+ * 2^16-2 entries, this value is set to 2^16-1 and we need to traverse the
  * entire list to know how many items it holds.
  *
  * <uint8_t zlend> is a special entry representing the end of the ziplist.
@@ -256,7 +256,7 @@
 #define ZIPLIST_ENTRY_END(zl)   ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-1)
 
 /* Increment the number of items field in the ziplist header. Note that this
- * macro should never overflow the unsigned 16 bit integer, since entires are
+ * macro should never overflow the unsigned 16 bit integer, since entries are
  * always pushed one at a time. When UINT16_MAX is reached we want the count
  * to stay there to signal that a full scan is needed to get the number of
  * items inside the ziplist. */

src/zmalloc.c

@@ -30,6 +30,7 @@
 
 #include <stdio.h>
 #include <stdlib.h>
+#include <stdint.h>
 
 /* This function provide us access to the original libc free(). This is useful
  * for instance to free results obtained by backtrace_symbols(). We need
@@ -418,7 +419,7 @@ size_t zmalloc_get_memory_size(void) {
     mib[0] = CTL_HW;
 #if defined(HW_REALMEM)
     mib[1] = HW_REALMEM;        /* FreeBSD. ----------------- */
-#elif defined(HW_PYSMEM)
+#elif defined(HW_PHYSMEM)
     mib[1] = HW_PHYSMEM;        /* Others. ------------------ */
 #endif
     unsigned int size = 0;      /* 32-bit */

tests/unit/scan.tcl

@@ -236,4 +236,50 @@ start_server {tags {"scan"}} {
         set first_score [lindex $res 1]
         assert {$first_score != 0}
     }
+
+    test "SCAN regression test for issue #4906" {
+        for {set k 0} {$k < 100} {incr k} {
+            r del set
+            r sadd set x; # Make sure it's not intset encoded
+            set toremove {}
+            unset -nocomplain found
+            array set found {}
+
+            # Populate the set
+            set numele [expr {100+[randomInt 1000]}]
+            for {set j 0} {$j < $numele} {incr j} {
+                r sadd set $j
+                if {$j >= 100} {
+                    lappend toremove $j
+                }
+            }
+
+            # Start scanning
+            set cursor 0
+            set iteration 0
+            set del_iteration [randomInt 10]
+            while {!($cursor == 0 && $iteration != 0)} {
+                lassign [r sscan set $cursor] cursor items
+
+                # Mark found items. We expect to find from 0 to 99 at the end
+                # since those elements will never be removed during the scanning.
+                foreach i $items {
+                    set found($i) 1
+                }
+                incr iteration
+                # At some point remove most of the items to trigger the
+                # rehashing to a smaller hash table.
+                if {$iteration == $del_iteration} {
+                    r srem set {*}$toremove
+                }
+            }
+
+            # Verify that SSCAN reported everything from 0 to 99
+            for {set j 0} {$j < 100} {incr j} {
+                if {![info exists found($j)]} {
+                    fail "SSCAN element missing $j"
+                }
+            }
+        }
+    }
 }

tests/unit/type/stream.tcl

@@ -253,4 +253,20 @@ start_server {
             }
         }
     }
+
+    test {XREVRANGE regression test for issue #5006} {
+        # Add non compressed entries
+        r xadd teststream 1234567891230 key1 value1
+        r xadd teststream 1234567891240 key2 value2
+        r xadd teststream 1234567891250 key3 value3
+
+        # Add SAMEFIELD compressed entries
+        r xadd teststream2 1234567891230 key1 value1
+        r xadd teststream2 1234567891240 key1 value2
+        r xadd teststream2 1234567891250 key1 value3
+
+        assert_equal [r xrevrange teststream 1234567891245 -] {{1234567891240-0 {key2 value2}} {1234567891230-0 {key1 value1}}}
+
+        assert_equal [r xrevrange teststream2 1234567891245 -] {{1234567891240-0 {key1 value2}} {1234567891230-0 {key1 value1}}}
+    }
 }

tests/unit/type/zset.tcl

@@ -653,11 +653,11 @@ start_server {tags {"zset"}} {
             r del zset
             assert_equal {} [r zpopmin zset]
             create_zset zset {-1 a 1 b 2 c 3 d 4 e}
-            assert_equal {-1 a} [r zpopmin zset]
-            assert_equal {1 b} [r zpopmin zset]
-            assert_equal {4 e} [r zpopmax zset]
-            assert_equal {3 d} [r zpopmax zset]
-            assert_equal {2 c} [r zpopmin zset]
+            assert_equal {a -1} [r zpopmin zset]
+            assert_equal {b 1} [r zpopmin zset]
+            assert_equal {e 4} [r zpopmax zset]
+            assert_equal {d 3} [r zpopmax zset]
+            assert_equal {c 2} [r zpopmin zset]
             assert_equal 0 [r exists zset]
             r set foo bar
             assert_error "*WRONGTYPE*" {r zpopmin foo}
@@ -669,8 +669,8 @@ start_server {tags {"zset"}} {
             assert_equal {} [r zpopmin z1 2]
             assert_error "*WRONGTYPE*" {r zpopmin foo 2}
             create_zset z1 {0 a 1 b 2 c 3 d}
-            assert_equal {0 a 1 b} [r zpopmin z1 2]
-            assert_equal {3 d 2 c} [r zpopmax z1 2]
+            assert_equal {a 0 b 1} [r zpopmin z1 2]
+            assert_equal {d 3 c 2} [r zpopmax z1 2]
         }
 
         test "BZPOP with a single existing sorted set - $encoding" {
@@ -678,11 +678,11 @@ start_server {tags {"zset"}} {
             create_zset zset {0 a 1 b 2 c}
 
             $rd bzpopmin zset 5
-            assert_equal {zset 0 a} [$rd read]
+            assert_equal {zset a 0} [$rd read]
             $rd bzpopmin zset 5
-            assert_equal {zset 1 b} [$rd read]
+            assert_equal {zset b 1} [$rd read]
             $rd bzpopmax zset 5
-            assert_equal {zset 2 c} [$rd read]
+            assert_equal {zset c 2} [$rd read]
             assert_equal 0 [r exists zset]
         }
 
@@ -692,16 +692,16 @@ start_server {tags {"zset"}} {
             create_zset z2 {3 d 4 e 5 f}
 
             $rd bzpopmin z1 z2 5
-            assert_equal {z1 0 a} [$rd read]
+            assert_equal {z1 a 0} [$rd read]
             $rd bzpopmax z1 z2 5
-            assert_equal {z1 2 c} [$rd read]
+            assert_equal {z1 c 2} [$rd read]
             assert_equal 1 [r zcard z1]
             assert_equal 3 [r zcard z2]
 
             $rd bzpopmax z2 z1 5
-            assert_equal {z2 5 f} [$rd read]
+            assert_equal {z2 f 5} [$rd read]
             $rd bzpopmin z2 z1 5
-            assert_equal {z2 3 d} [$rd read]
+            assert_equal {z2 d 3} [$rd read]
             assert_equal 1 [r zcard z1]
             assert_equal 1 [r zcard z2]
         }
@@ -711,9 +711,9 @@ start_server {tags {"zset"}} {
             r del z1
             create_zset z2 {3 d 4 e 5 f}
             $rd bzpopmax z1 z2 5
-            assert_equal {z2 5 f} [$rd read]
+            assert_equal {z2 f 5} [$rd read]
             $rd bzpopmin z2 z1 5
-            assert_equal {z2 3 d} [$rd read]
+            assert_equal {z2 d 3} [$rd read]
             assert_equal 0 [r zcard z1]
             assert_equal 1 [r zcard z2]
         }
@@ -1107,7 +1107,7 @@ start_server {tags {"zset"}} {
             r del zset
             r zadd zset 1 bar
             $rd read
-        } {zset 1 bar}
+        } {zset bar 1}
 
         test "BZPOPMIN, ZADD + DEL + SET should not awake blocked client" {
             set rd [redis_deferring_client]
@@ -1124,7 +1124,7 @@ start_server {tags {"zset"}} {
             r del zset
             r zadd zset 1 bar
             $rd read
-        } {zset 1 bar}
+        } {zset bar 1}
 
         test "BZPOPMIN with same key multiple times should work" {
             set rd [redis_deferring_client]
@@ -1133,18 +1133,18 @@ start_server {tags {"zset"}} {
             # Data arriving after the BZPOPMIN.
             $rd bzpopmin z1 z2 z2 z1 0
             r zadd z1 0 a
-            assert_equal [$rd read] {z1 0 a}
+            assert_equal [$rd read] {z1 a 0}
             $rd bzpopmin z1 z2 z2 z1 0
             r zadd z2 1 b
-            assert_equal [$rd read] {z2 1 b}
+            assert_equal [$rd read] {z2 b 1}
 
             # Data already there.
             r zadd z1 0 a
             r zadd z2 1 b
             $rd bzpopmin z1 z2 z2 z1 0
-            assert_equal [$rd read] {z1 0 a}
+            assert_equal [$rd read] {z1 a 0}
             $rd bzpopmin z1 z2 z2 z1 0
-            assert_equal [$rd read] {z2 1 b}
+            assert_equal [$rd read] {z2 b 1}
         }
 
         test "MULTI/EXEC is isolated from the point of view of BZPOPMIN" {
@@ -1157,7 +1157,7 @@ start_server {tags {"zset"}} {
             r zadd zset 2 c
             r exec
             $rd read
-        } {zset 0 a}
+        } {zset a 0}
 
         test "BZPOPMIN with variadic ZADD" {
             set rd [redis_deferring_client]
@@ -1167,7 +1167,7 @@ start_server {tags {"zset"}} {
             if {$::valgrind} {after 100}
             assert_equal 2 [r zadd zset -1 foo 1 bar]
             if {$::valgrind} {after 100}
-            assert_equal {zset -1 foo} [$rd read]
+            assert_equal {zset foo -1} [$rd read]
             assert_equal {bar} [r zrange zset 0 -1]
         }
 
@@ -1177,7 +1177,7 @@ start_server {tags {"zset"}} {
             $rd bzpopmin zset 0
             after 1000
             r zadd zset 0 foo
-            assert_equal {zset 0 foo} [$rd read]
+            assert_equal {zset foo 0} [$rd read]
         }
     }
 

utils/generate-command-help.rb

@@ -14,7 +14,8 @@ GROUPS = [
   "scripting",
   "hyperloglog",
   "cluster",
-  "geo"
+  "geo",
+  "stream"
 ].freeze
 
 GROUPS_BY_NAME = Hash[*