Redis 3.0.5

Kinda related to #2770.

00-RELEASENOTES

@@ -10,6 +10,30 @@ 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.
 --------------------------------------------------------------------------------
 
+--[ Redis 3.0.5 ] Release date: 15 Oct 2015
+
+Upgrade urgency: MODERATE, the most important thing is a fix in the replication
+                 code that may make the slave hanging forever if the master
+                 remains with an open socket even if it is no longer able to
+                 reply.
+
+* [FIX] MOVE now moves the TTL as well. A bug lasting forever... finally
+        fixed thanks to Andy Grunwald that reported it.
+        (reported by Andy Grunwald, fixed by Salvatore Sanfilippo)
+* [FIX] Fix a false positive in HSTRLEN test.
+* [FIX] Fix a bug in redis-cli --pipe mode that was not able to read back
+        replies from the server incrementally. Now a mass import will use
+        a lot less memory, and you can use --pipe to do incremental streaming.
+        (reported by Twitter user @fsaintjacques, fixed by Salvatore
+        Sanfilippo)
+* [FIX] Slave detection of master timeout. (fixed by Kevin McGehee, refactoring
+        and regression test by Salvatore Sanfilippo)
+
+* [NEW] Cluster: redis-trib fix can fix an additional case for opens lots.
+        (Salvatore Sanfilippo)
+* [NEW] Cluster: redis-trib import support for --copy and --replace options
+        (David Thomson)
+
 --[ Redis 3.0.4 ] Release date: 8 Sep 2015
 
 Upgrade urgency: HIGH for Redis and Sentinel. However note that in order to
@@ -34,7 +58,7 @@ Upgrade urgency: HIGH for Redis and Sentinel. However note that in order to
         an issue reported by "GeorgeBJ" user at Github)
 * [FIX] redis-cli --scan iteration fixed when returned cursor overflows
         32 bit signed integer. (Ofir Luzon, Yuval Inbar)
-* [FIX] Senitnel: fixed a bug during the master switch process, where for a
+* [FIX] Sentinel: fixed a bug during the master switch process, where for a
         failed conditional check, the new configuration is rewritten, during
         a small window of time, in a corrupted way where the master is
         also reported to be one of the slaves. This bug is rare to trigger

redis.conf

@@ -1,4 +1,9 @@
-# Redis configuration file example
+# Redis configuration file example.
+#
+# Note that in order to read the configuration file, Redis must be
+# started with the file path as first argument:
+#
+# ./redis-server /path/to/redis.conf
 
 # Note on units: when memory size is needed, it is possible to specify
 # it in the usual form of 1k 5GB 4M and so forth:

src/db.c

@@ -714,7 +714,7 @@ void moveCommand(redisClient *c) {
     robj *o;
     redisDb *src, *dst;
     int srcid;
-    long long dbid;
+    long long dbid, expire;
 
     if (server.cluster_enabled) {
         addReplyError(c,"MOVE is not allowed in cluster mode");
@@ -748,6 +748,7 @@ void moveCommand(redisClient *c) {
         addReply(c,shared.czero);
         return;
     }
+    expire = getExpire(c->db,c->argv[1]);
 
     /* Return zero if the key already exists in the target DB */
     if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
@@ -755,6 +756,7 @@ void moveCommand(redisClient *c) {
         return;
     }
     dbAdd(dst,c->argv[1],o);
+    if (expire != -1) setExpire(dst,c->argv[1],expire);
     incrRefCount(o);
 
     /* OK! key moved, free the entry in the source DB */

src/redis-cli.c

@@ -1397,6 +1397,7 @@ static void getRDB(void) {
  * Bulk import (pipe) mode
  *--------------------------------------------------------------------------- */
 
+#define PIPEMODE_WRITE_LOOP_MAX_BYTES (128*1024)
 static void pipeMode(void) {
     int fd = context->fd;
     long long errors = 0, replies = 0, obuf_len = 0, obuf_pos = 0;
@@ -1473,6 +1474,8 @@ static void pipeMode(void) {
 
         /* Handle the writable state: we can send protocol to the server. */
         if (mask & AE_WRITABLE) {
+            ssize_t loop_nwritten = 0;
+
             while(1) {
                 /* Transfer current buffer to server. */
                 if (obuf_len != 0) {
@@ -1489,6 +1492,7 @@ static void pipeMode(void) {
                     }
                     obuf_len -= nwritten;
                     obuf_pos += nwritten;
+                    loop_nwritten += nwritten;
                     if (obuf_len != 0) break; /* Can't accept more data. */
                 }
                 /* If buffer is empty, load from stdin. */
@@ -1524,7 +1528,8 @@ static void pipeMode(void) {
                         obuf_pos = 0;
                     }
                 }
-                if (obuf_len == 0 && eof) break;
+                if ((obuf_len == 0 && eof) ||
+                    loop_nwritten > PIPEMODE_WRITE_LOOP_MAX_BYTES) break;
             }
         }
 

src/redis-trib.rb

@@ -496,6 +496,10 @@ class RedisTrib
         #         importing state in 1 slot. That's trivial to address.
         if migrating.length == 1 && importing.length == 1
             move_slot(migrating[0],importing[0],slot,:verbose=>true,:fix=>true)
+        # Case 2: There are multiple nodes that claim the slot as importing,
+        # they probably got keys about the slot after a restart so opened
+        # the slot. In this case we just move all the keys to the owner
+        # according to the configuration.
         elsif migrating.length == 0 && importing.length > 0
             xputs ">>> Moving all the #{slot} slot keys to its owner #{owner}"
             importing.each {|node|
@@ -504,8 +508,14 @@ class RedisTrib
                 xputs ">>> Setting #{slot} as STABLE in #{node}"
                 node.r.cluster("setslot",slot,"stable")
             }
+        # Case 3: There are no slots claiming to be in importing state, but
+        # there is a migrating node that actually don't have any key. We
+        # can just close the slot, probably a reshard interrupted in the middle.
+        elsif importing.length == 0 && migrating.length == 1 &&
+              migrating[0].r.cluster("getkeysinslot",slot,10).length == 0
+            migrating[0].r.cluster("setslot",slot,"stable")
         else
-            xputs "[ERR] Sorry, Redis-trib can't fix this slot yet (work in progress)"
+            xputs "[ERR] Sorry, Redis-trib can't fix this slot yet (work in progress). Slot is set as migrating in #{migrating.join(",")}, as importing in #{importing.join(",")}, owner is #{owner}"
         end
     end
 
@@ -812,7 +822,7 @@ class RedisTrib
                     source.r.client.call(["migrate",target.info[:host],target.info[:port],key,0,15000])
                 rescue => e
                     if o[:fix] && e.to_s =~ /BUSYKEY/
-                        xputs "*** Target key #{key} exists. Replace it for FIX."
+                        xputs "*** Target key #{key} exists. Replacing it for FIX."
                         source.r.client.call(["migrate",target.info[:host],target.info[:port],key,0,15000,:replace])
                     else
                         puts ""
@@ -1139,7 +1149,9 @@ class RedisTrib
     def import_cluster_cmd(argv,opt)
         source_addr = opt['from']
         xputs ">>> Importing data from #{source_addr} to cluster #{argv[1]}"
-
+        use_copy = opt['copy']
+        use_replace = opt['replace']
+        
         # Check the existing cluster.
         load_cluster_info_from_node(argv[0])
         check_cluster
@@ -1174,7 +1186,10 @@ class RedisTrib
                 print "Migrating #{k} to #{target}: "
                 STDOUT.flush
                 begin
-                    source.client.call(["migrate",target.info[:host],target.info[:port],k,0,15000])
+                    cmd = ["migrate",target.info[:host],target.info[:port],k,0,15000]
+                    cmd << :copy if use_copy
+                    cmd << :replace if use_replace
+                    source.client.call(cmd)
                 rescue => e
                     puts e
                 else
@@ -1334,7 +1349,7 @@ COMMANDS={
 ALLOWED_OPTIONS={
     "create" => {"replicas" => true},
     "add-node" => {"slave" => false, "master-id" => true},
-    "import" => {"from" => :required},
+    "import" => {"from" => :required, "copy" => false, "replace" => false},
     "reshard" => {"from" => true, "to" => true, "slots" => true, "yes" => false}
 }
 

src/redis.h

@@ -272,6 +272,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #define REDIS_REPL_NONE 0 /* No active replication */
 #define REDIS_REPL_CONNECT 1 /* Must connect to master */
 #define REDIS_REPL_CONNECTING 2 /* Connecting to master */
+/* --- Handshake states, must be ordered --- */
 #define REDIS_REPL_RECEIVE_PONG 3 /* Wait for PING reply */
 #define REDIS_REPL_SEND_AUTH 4 /* Send AUTH to master */
 #define REDIS_REPL_RECEIVE_AUTH 5 /* Wait for AUTH reply */
@@ -281,6 +282,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #define REDIS_REPL_RECEIVE_CAPA 9 /* Wait for REPLCONF reply */
 #define REDIS_REPL_SEND_PSYNC 10 /* Send PSYNC */
 #define REDIS_REPL_RECEIVE_PSYNC 11 /* Wait for PSYNC reply */
+/* --- End of handshake states --- */
 #define REDIS_REPL_TRANSFER 12 /* Receiving .rdb from master */
 #define REDIS_REPL_CONNECTED 13 /* Connected to master */
 

src/replication.c

@@ -909,6 +909,13 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
 
 /* ----------------------------------- SLAVE -------------------------------- */
 
+/* Returns 1 if the given replication state is a handshake state,
+ * 0 otherwise. */
+int slaveIsInHandshakeState(void) {
+    return server.repl_state >= REDIS_REPL_RECEIVE_PONG &&
+           server.repl_state <= REDIS_REPL_RECEIVE_PSYNC;
+}
+
 /* Abort the async download of the bulk dataset while SYNC-ing with master */
 void replicationAbortSyncTransfer(void) {
     redisAssert(server.repl_state == REDIS_REPL_TRANSFER);
@@ -1196,6 +1203,7 @@ char *sendSynchronousCommand(int flags, int fd, ...) {
             return sdscatprintf(sdsempty(),"-Reading from master: %s",
                     strerror(errno));
         }
+        server.repl_transfer_lastio = server.unixtime;
         return sdsnew(buf);
     }
     return NULL;
@@ -1626,7 +1634,7 @@ void undoConnectWithMaster(void) {
     int fd = server.repl_transfer_s;
 
     redisAssert(server.repl_state == REDIS_REPL_CONNECTING ||
-                server.repl_state == REDIS_REPL_RECEIVE_PONG);
+                slaveIsInHandshakeState());
     aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);
     close(fd);
     server.repl_transfer_s = -1;
@@ -1645,7 +1653,7 @@ int cancelReplicationHandshake(void) {
     if (server.repl_state == REDIS_REPL_TRANSFER) {
         replicationAbortSyncTransfer();
     } else if (server.repl_state == REDIS_REPL_CONNECTING ||
-             server.repl_state == REDIS_REPL_RECEIVE_PONG)
+               slaveIsInHandshakeState())
     {
         undoConnectWithMaster();
     } else {
@@ -1780,22 +1788,17 @@ void roleCommand(redisClient *c) {
         addReplyBulkCBuffer(c,"slave",5);
         addReplyBulkCString(c,server.masterhost);
         addReplyLongLong(c,server.masterport);
-        switch(server.repl_state) {
-        case REDIS_REPL_NONE: slavestate = "none"; break;
-        case REDIS_REPL_CONNECT: slavestate = "connect"; break;
-        case REDIS_REPL_CONNECTING: slavestate = "connecting"; break;
-        case REDIS_REPL_RECEIVE_PONG:
-        case REDIS_REPL_SEND_AUTH:
-        case REDIS_REPL_RECEIVE_AUTH:
-        case REDIS_REPL_SEND_PORT:
-        case REDIS_REPL_RECEIVE_PORT:
-        case REDIS_REPL_SEND_CAPA:
-        case REDIS_REPL_RECEIVE_CAPA:
-        case REDIS_REPL_SEND_PSYNC:
-        case REDIS_REPL_RECEIVE_PSYNC: slavestate = "handshake"; break;
-        case REDIS_REPL_TRANSFER: slavestate = "sync"; break;
-        case REDIS_REPL_CONNECTED: slavestate = "connected"; break;
-        default: slavestate = "unknown"; break;
+        if (slaveIsInHandshakeState()) {
+            slavestate = "handshake";
+        } else {
+            switch(server.repl_state) {
+            case REDIS_REPL_NONE: slavestate = "none"; break;
+            case REDIS_REPL_CONNECT: slavestate = "connect"; break;
+            case REDIS_REPL_CONNECTING: slavestate = "connecting"; break;
+            case REDIS_REPL_TRANSFER: slavestate = "sync"; break;
+            case REDIS_REPL_CONNECTED: slavestate = "connected"; break;
+            default: slavestate = "unknown"; break;
+            }
         }
         addReplyBulkCString(c,slavestate);
         addReplyLongLong(c,server.master ? server.master->reploff : -1);
@@ -2186,8 +2189,8 @@ void replicationCron(void) {
     /* Non blocking connection timeout? */
     if (server.masterhost &&
         (server.repl_state == REDIS_REPL_CONNECTING ||
-         server.repl_state == REDIS_REPL_RECEIVE_PONG) &&
-        (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
+         slaveIsInHandshakeState()) &&
+         (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
     {
         redisLog(REDIS_WARNING,"Timeout connecting to the MASTER...");
         undoConnectWithMaster();

src/version.h

@@ -1 +1 @@
-#define REDIS_VERSION "3.0.4"
+#define REDIS_VERSION "3.0.5"

tests/integration/replication.tcl

@@ -1,3 +1,56 @@
+proc log_file_matches {log pattern} {
+    set fp [open $log r]
+    set content [read $fp]
+    close $fp
+    string match $pattern $content
+}
+
+start_server {tags {"repl"}} {
+    set slave [srv 0 client]
+    set slave_host [srv 0 host]
+    set slave_port [srv 0 port]
+    set slave_log [srv 0 stdout]
+    start_server {} {
+        set master [srv 0 client]
+        set master_host [srv 0 host]
+        set master_port [srv 0 port]
+
+        # Configure the master in order to hang waiting for the BGSAVE
+        # operation, so that the slave remains in the handshake state.
+        $master config set repl-diskless-sync yes
+        $master config set repl-diskless-sync-delay 1000
+
+        # Use a short replication timeout on the slave, so that if there
+        # are no bugs the timeout is triggered in a reasonable amount
+        # of time.
+        $slave config set repl-timeout 5
+
+        # Start the replication process...
+        $slave slaveof $master_host $master_port
+
+        test {Slave enters handshake} {
+            wait_for_condition 50 1000 {
+                [string match *handshake* [$slave role]]
+            } else {
+                fail "Slave does not enter handshake state"
+            }
+        }
+
+        # But make the master unable to send
+        # the periodic newlines to refresh the connection. The slave
+        # should detect the timeout.
+        $master debug sleep 10
+
+        test {Slave is able to detect timeout during handshake} {
+            wait_for_condition 50 1000 {
+                [log_file_matches $slave_log "*Timeout connecting to the MASTER*"]
+            } else {
+                fail "Slave is not able to detect timeout"
+            }
+        }
+    }
+}
+
 start_server {tags {"repl"}} {
     set A [srv 0 client]
     set A_host [srv 0 host]

tests/unit/basic.tcl

@@ -433,6 +433,32 @@ start_server {tags {"basic"}} {
         set e
     } {*ERR*index out of range}
 
+    test {MOVE can move key expire metadata as well} {
+        r select 10
+        r flushdb
+        r select 9
+        r set mykey foo ex 100
+        r move mykey 10
+        assert {[r ttl mykey] == -2}
+        r select 10
+        assert {[r ttl mykey] > 0 && [r ttl mykey] <= 100}
+        assert {[r get mykey] eq "foo"}
+        r select 9
+    }
+
+    test {MOVE does not create an expire if it does not exist} {
+        r select 10
+        r flushdb
+        r select 9
+        r set mykey foo
+        r move mykey 10
+        assert {[r ttl mykey] == -2}
+        r select 10
+        assert {[r ttl mykey] == -1}
+        assert {[r get mykey] eq "foo"}
+        r select 9
+    }
+
     test {SET/GET keys in different DBs} {
         r set a hello
         r set b world

tests/unit/type/hash.tcl

@@ -2,8 +2,8 @@ start_server {tags {"hash"}} {
     test {HSET/HLEN - Small hash creation} {
         array set smallhash {}
         for {set i 0} {$i < 8} {incr i} {
-            set key [randstring 0 8 alpha]
-            set val [randstring 0 8 alpha]
+            set key __avoid_collisions__[randstring 0 8 alpha]
+            set val __avoid_collisions__[randstring 0 8 alpha]
             if {[info exists smallhash($key)]} {
                 incr i -1
                 continue
@@ -21,8 +21,8 @@ start_server {tags {"hash"}} {
     test {HSET/HLEN - Big hash creation} {
         array set bighash {}
         for {set i 0} {$i < 1024} {incr i} {
-            set key [randstring 0 8 alpha]
-            set val [randstring 0 8 alpha]
+            set key __avoid_collisions__[randstring 0 8 alpha]
+            set val __avoid_collisions__[randstring 0 8 alpha]
             if {[info exists bighash($key)]} {
                 incr i -1
                 continue
@@ -33,7 +33,7 @@ start_server {tags {"hash"}} {
         list [r hlen bighash]
     } {1024}
 
-    test {Is the big hash encoded with a ziplist?} {
+    test {Is the big hash encoded with an hash table?} {
         assert_encoding hashtable bighash
     }