misc: de-duplicate the code by adding replicationRestoreOffsetFromSaveInfo

Signed-off-by: arthur.lee <[email protected]>

Author: arthurkiller

src/aof.c

@@ -1470,32 +1470,7 @@ int loadSingleAppendOnlyFile(char *filename) {
             goto cleanup;
         } else {
             /* Restore the replication ID / offset from the RDB file. */
-            if (rsi.repl_id_is_set && rsi.repl_offset != -1 && rsi.repl_stream_db != -1) {
-                rsi_is_valid = 1;
-                if (!iAmPrimary()) {
-                    memcpy(server.replid, rsi.repl_id, sizeof(server.replid));
-                    server.primary_repl_offset = rsi.repl_offset;
-                    if (!server.primary && !server.cached_primary) {
-                        /* only cache myself primary if replica did not synced to its primary node yet */
-                        replicationCachePrimaryUsingMyself();
-                        selectDb(server.cached_primary, rsi.repl_stream_db);
-                    }
-                    serverLog(LL_NOTICE, "Loading preamble rdb changed replication info, replid: %s, primary_repl_offset: %lld",
-                              server.replid, server.primary_repl_offset);
-                } else {
-                    memcpy(server.replid2, rsi.repl_id, sizeof(server.replid));
-                    server.second_replid_offset = rsi.repl_offset + 1;
-                    server.primary_repl_offset += rsi.repl_offset;
-                    serverAssert(server.repl_backlog);
-                    server.repl_backlog->offset = server.primary_repl_offset - server.repl_backlog->histlen + 1;
-                    rebaseReplicationBuffer(rsi.repl_offset);
-                    server.repl_no_replicas_since = time(NULL);
-                    serverLog(LL_NOTICE, "Loading preamble rdb changed replication info, replid2: %s, secondary_repl_offset: "
-                                         "%lld, primary_repl_offset: %lld",
-                              server.replid2, server.second_replid_offset, server.primary_repl_offset);
-                }
-            }
-
+            rsi_is_valid = replicationRestoreOffsetFromSaveInfo(&rsi, true);
             loadingAbsProgress(ftello(fp));
             last_progress_report_size = ftello(fp);
             if (old_style) serverLog(LL_NOTICE, "Reading the remaining AOF tail...");

src/rdb.c

@@ -3952,3 +3952,41 @@ rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi) {
     }
     return NULL;
 }
+
+/* Restore the replication ID / offset from the RDB file
+ * return 1 if rdbSaveInfo is valid */
+int replicationRestoreOffsetFromSaveInfo(rdbSaveInfo *rsi, bool is_aof_preamble) {
+    int rsi_is_valid = 0;
+    if (rsi == NULL) return rsi_is_valid;
+    if (rsi->repl_id_is_set && rsi->repl_offset != -1 && rsi->repl_stream_db != -1) {
+        /* Note that older implementations may save a repl_stream_db
+         * of -1 inside the RDB file in a wrong way, see more
+         * information in function rdbPopulateSaveInfo. */
+        rsi_is_valid = 1;
+        if (!iAmPrimary()) {
+            memcpy(server.replid, rsi->repl_id, sizeof(server.replid));
+            server.primary_repl_offset = rsi->repl_offset;
+            if (!is_aof_preamble || (!server.primary && !server.cached_primary)) {
+                /* If this is a replica, create a cached primary from this
+                 * information, in order to allow partial resynchronizations
+                 * with primaries. For AOF, only cache the primary if replica
+                 * has not synced to its primary node yet. */
+                replicationCachePrimaryUsingMyself();
+                selectDb(server.cached_primary, rsi->repl_stream_db);
+            }
+        } else {
+            /* If this is a primary, we can save the replication info
+             * as secondary ID and offset, in order to allow replicas
+             * to partial resynchronizations with primaries. */
+            memcpy(server.replid2, rsi->repl_id, sizeof(server.replid));
+            server.second_replid_offset = rsi->repl_offset + 1;
+            /* Rebase primary_repl_offset from rsi.repl_offset. */
+            server.primary_repl_offset += rsi->repl_offset;
+            serverAssert(server.repl_backlog);
+            server.repl_backlog->offset = server.primary_repl_offset - server.repl_backlog->histlen + 1;
+            rebaseReplicationBuffer(rsi->repl_offset);
+            server.repl_no_replicas_since = time(NULL);
+        }
+    }
+    return rsi_is_valid;
+}

src/rdb.h

@@ -212,5 +212,6 @@ int rdbFunctionLoad(rio *rdb, int ver, functionsLibCtx *lib_ctx, int rdbflags, s
 int rdbSaveRio(int req, rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi);
 ssize_t rdbSaveFunctions(rio *rdb);
 rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi);
+int replicationRestoreOffsetFromSaveInfo(rdbSaveInfo *rsi, bool is_aof_preamble);
 
 #endif

src/server.c

@@ -6926,36 +6926,7 @@ void loadDataFromDisk(void) {
         int rdb_load_ret = rdbLoad(server.rdb_filename, &rsi, rdb_flags);
         if (rdb_load_ret == RDB_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 &&
-                /* Note that older implementations may save a repl_stream_db
-                 * of -1 inside the RDB file in a wrong way, see more
-                 * information in function rdbPopulateSaveInfo. */
-                rsi.repl_stream_db != -1) {
-                rsi_is_valid = 1;
-                if (!iAmPrimary()) {
-                    memcpy(server.replid, rsi.repl_id, sizeof(server.replid));
-                    server.primary_repl_offset = rsi.repl_offset;
-                    /* If this is a replica, create a cached primary from this
-                     * information, in order to allow partial resynchronizations
-                     * with primaries. */
-                    replicationCachePrimaryUsingMyself();
-                    selectDb(server.cached_primary, rsi.repl_stream_db);
-                } else {
-                    /* If this is a primary, we can save the replication info
-                     * as secondary ID and offset, in order to allow replicas
-                     * to partial resynchronizations with primaries. */
-                    memcpy(server.replid2, rsi.repl_id, sizeof(server.replid));
-                    server.second_replid_offset = rsi.repl_offset + 1;
-                    /* Rebase primary_repl_offset from rsi.repl_offset. */
-                    server.primary_repl_offset += rsi.repl_offset;
-                    serverAssert(server.repl_backlog);
-                    server.repl_backlog->offset = server.primary_repl_offset - server.repl_backlog->histlen + 1;
-                    rebaseReplicationBuffer(rsi.repl_offset);
-                    server.repl_no_replicas_since = time(NULL);
-                }
-            }
+            rsi_is_valid = replicationRestoreOffsetFromSaveInfo(&rsi, false);
         } else if (rdb_load_ret != RDB_NOT_EXIST) {
             serverLog(LL_WARNING, "Fatal error loading the DB, check server logs. Exiting.");
             exit(1);