PSYNC2: Save replication ID/offset on RDB file.psync2

This means that stopping a slave and restarting it will still make it
able to PSYNC with the master. Moreover the master itself will retain
its ID/offset, in case it gets turned into a slave, or if a slave will
try to PSYNC with it with an exactly updated offset (otherwise there is
no backlog).

This change was possible thanks to PSYNC v2 that makes saving the current
replication state much simpler.

4 files changed, 29 insertions, 3 deletions

diff --git a/src/rdb.c b/src/rdb.c
index aa9c631de..765e13374 100644
--- a/src/rdb.c
+++ b/src/rdb.c
@@ -855,6 +855,8 @@ int rdbSaveInfoAuxFields(rio *rdb, int flags, rdbSaveInfo *rsi) {
         }
     }
     if (rdbSaveAuxFieldStrInt(rdb,"aof-preamble",aof_preamble) == -1) return -1;
+    if (rdbSaveAuxFieldStrStr(rdb,"repl-id",server.replid) == -1) return -1;
+    if (rdbSaveAuxFieldStrInt(rdb,"repl-offset",server.master_repl_offset) == -1) return -1;
     return 1;
 }
 
@@ -1513,6 +1515,13 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi) {
                     (char*)auxval->ptr);
             } else if (!strcasecmp(auxkey->ptr,"repl-stream-db")) {
                 if (rsi) rsi->repl_stream_db = atoi(auxval->ptr);
+            } else if (!strcasecmp(auxkey->ptr,"repl-id")) {
+                if (rsi && sdslen(auxval->ptr) == CONFIG_RUN_ID_SIZE) {
+                    memcpy(rsi->repl_id,auxval->ptr,CONFIG_RUN_ID_SIZE+1);
+                    rsi->repl_id_is_set = 1;
+                }
+            } else if (!strcasecmp(auxkey->ptr,"repl-offset")) {
+                if (rsi) rsi->repl_offset = strtoll(auxval->ptr,NULL,10);
             } else {
                 /* We ignore fields we don't understand, as by AUX field
                  * contract. */
diff --git a/src/replication.c b/src/replication.c
index 1962b3a42..1a9b2e574 100644
--- a/src/replication.c
+++ b/src/replication.c
@@ -39,7 +39,6 @@
 
 void replicationDiscardCachedMaster(void);
 void replicationResurrectCachedMaster(int newfd);
-void replicationCacheMasterUsingMyself(void);
 void replicationSendAck(void);
 void putSlaveOnline(client *slave);
 int cancelReplicationHandshake(void);
diff --git a/src/server.c b/src/server.c
index b94490a33..d17ded9b0 100644
--- a/src/server.c
+++ b/src/server.c
@@ -3423,9 +3423,20 @@ void loadDataFromDisk(void) {
         if (loadAppendOnlyFile(server.aof_filename) == C_OK)
             serverLog(LL_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000);
     } else {
-        if (rdbLoad(server.rdb_filename,NULL) == C_OK) {
+        rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
+        if (rdbLoad(server.rdb_filename,&rsi) == C_OK) {
             serverLog(LL_NOTICE,"DB loaded from disk: %.3f seconds",
                 (float)(ustime()-start)/1000000);
+
+            /* Restore the replication ID / offset from the RDB file. */
+            if (rsi.repl_id_is_set && rsi.repl_offset != -1) {
+                memcpy(server.replid,rsi.repl_id,sizeof(server.replid));
+                server.master_repl_offset = rsi.repl_offset;
+                /* If we are a slave, create a cached master from this
+                 * information, in order to allow partial resynchronizations
+                 * with masters. */
+                if (server.masterhost) replicationCacheMasterUsingMyself();
+            }
         } else if (errno != ENOENT) {
             serverLog(LL_WARNING,"Fatal error loading the DB: %s. Exiting.",strerror(errno));
             exit(1);
diff --git a/src/server.h b/src/server.h
index 8aa1d6fcb..b7f909933 100644
--- a/src/server.h
+++ b/src/server.h
@@ -813,10 +813,16 @@ struct redisMemOverhead {
  * select the correct DB and are able to accept the stream coming from the
  * top-level master. */
 typedef struct rdbSaveInfo {
+    /* Used saving and loading. */
     int repl_stream_db;  /* DB to select in server.master client. */
+
+    /* Used only loading. */
+    int repl_id_is_set;  /* True if repl_id field is set. */
+    char repl_id[CONFIG_RUN_ID_SIZE+1];     /* Replication ID. */
+    long long repl_offset;                  /* Replication offset. */
 } rdbSaveInfo;
 
-#define RDB_SAVE_INFO_INIT {-1}
+#define RDB_SAVE_INFO_INIT {-1,0,"000000000000000000000000000000",-1}
 
 /*-----------------------------------------------------------------------------
  * Global server state
@@ -1441,6 +1447,7 @@ int replicationSetupSlaveForFullResync(client *slave, long long offset);
 void changeReplicationId(void);
 void clearReplicationId2(void);
 void chopReplicationBacklog(void);
+void replicationCacheMasterUsingMyself(void);
 
 /* Generic persistence functions */
 void startLoading(FILE *fp);