several bugfixes to diskless replication and the CAPA mechanism

in diskless replication - master not notifing the slave that rdb transfer
terminated on error, and lets slave wait for replication timeout

when starting a replication for a certain mincapa, we must take care to
exclude slaves that didn't declare that capa, otherwise they may get something
that they can't handle

3 files changed, 12 insertions, 5 deletions

diff --git a/src/rdb.c b/src/rdb.c
index 28985b2a6..4f46d3580 100644
--- a/src/rdb.c
+++ b/src/rdb.c
@@ -1907,7 +1907,7 @@ void backgroundSaveDoneHandler(int exitcode, int bysignal) {
 
 /* Spawn an RDB child that writes the RDB to the sockets of the slaves
  * that are currently in SLAVE_STATE_WAIT_BGSAVE_START state. */
-int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
+int rdbSaveToSlavesSockets(int mincapa, rdbSaveInfo *rsi) {
     int *fds;
     uint64_t *clientids;
     int numfds;
@@ -1940,6 +1940,9 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
         client *slave = ln->value;
 
         if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
+            /* Check slave has at least the minimum capabilities */
+            if ((mincapa & slave->slave_capa) != mincapa)
+                continue;
             clientids[numfds] = slave->id;
             fds[numfds++] = slave->fd;
             replicationSetupSlaveForFullResync(slave,getPsyncInitialOffset());
diff --git a/src/rdb.h b/src/rdb.h
index ecb066fb0..fd6cbaf33 100644
--- a/src/rdb.h
+++ b/src/rdb.h
@@ -132,7 +132,7 @@ int rdbSaveObjectType(rio *rdb, robj *o);
 int rdbLoadObjectType(rio *rdb);
 int rdbLoad(char *filename, rdbSaveInfo *rsi);
 int rdbSaveBackground(char *filename, rdbSaveInfo *rsi);
-int rdbSaveToSlavesSockets(rdbSaveInfo *rsi);
+int rdbSaveToSlavesSockets(int mincapa, rdbSaveInfo *rsi);
 void rdbRemoveTempFile(pid_t childpid);
 int rdbSave(char *filename, rdbSaveInfo *rsi);
 ssize_t rdbSaveObject(rio *rdb, robj *o);
diff --git a/src/replication.c b/src/replication.c
index 064d2bece..328382570 100644
--- a/src/replication.c
+++ b/src/replication.c
@@ -575,7 +575,7 @@ int startBgsaveForReplication(int mincapa) {
      * otherwise slave will miss repl-stream-db. */
     if (rsiptr) {
         if (socket_target)
-            retval = rdbSaveToSlavesSockets(rsiptr);
+            retval = rdbSaveToSlavesSockets(mincapa, rsiptr);
         else
             retval = rdbSaveBackground(server.rdb_filename,rsiptr);
     } else {
@@ -593,6 +593,7 @@ int startBgsaveForReplication(int mincapa) {
             client *slave = ln->value;
 
             if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
+                slave->replstate = REPL_STATE_NONE;
                 slave->flags &= ~CLIENT_SLAVE;
                 listDelNode(server.slaves,ln);
                 addReplyError(slave,
@@ -611,8 +612,11 @@ int startBgsaveForReplication(int mincapa) {
             client *slave = ln->value;
 
             if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
-                    replicationSetupSlaveForFullResync(slave,
-                            getPsyncInitialOffset());
+                /* Check slave has at least the minimum capabilities */
+                if ((mincapa & slave->slave_capa) != mincapa)
+                    continue;
+                replicationSetupSlaveForFullResync(slave,
+                        getPsyncInitialOffset());
             }
         }
     }