use rdb to start appendonly after primary-replica full sync

Signed-off-by: Ray Cao <RayCao125@gmail.com>

Author: RayaCoo

src/aof.c

@@ -1005,6 +1005,104 @@ int startAppendOnly(void) {
     return C_OK;
 }
 
+int tryRestartAppendOnlyWithRdb(void) {
+    serverAssert(server.aof_state == AOF_OFF);
+
+    int newfd = -1;
+    sds new_base_filename = NULL;
+    sds new_base_filepath = NULL;
+    sds new_incr_filename = NULL;
+    sds new_incr_filepath = NULL;
+    aofManifest *temp_am = NULL;
+
+    if (dirCreateIfMissing(server.aof_dirname) == -1) {
+        serverLog(LL_WARNING, "Can't open or create append-only dir %s: %s", server.aof_dirname, strerror(errno));
+        goto cleanup;
+    }
+
+    serverAssert(server.aof_manifest != NULL);
+
+    /* Dup a temporary aof_manifest for subsequent modifications. */
+    temp_am = aofManifestDup(server.aof_manifest);
+
+    /* Get a new BASE file name and mark the previous (if we have)
+     * as the HISTORY type. */
+    new_base_filename = getNewBaseFileNameAndMarkPreAsHistory(temp_am, server.aof_use_rdb_preamble);
+    serverAssert(new_base_filename != NULL);
+    new_base_filepath = makePath(server.aof_dirname, new_base_filename);
+    if (rename(server.rdb_filename, new_base_filepath) == -1) {
+        serverLog(LL_WARNING, "Error trying to rename the RDB file %s into %s: %s", server.rdb_filename,
+                  new_base_filepath, strerror(errno));
+        goto cleanup;
+    }
+    sdsfree(new_base_filepath);
+
+    /* Get next new incr aof name. */
+    new_incr_filename = getNewIncrAofName(temp_am);
+    new_incr_filepath = makePath(server.aof_dirname, new_incr_filename);
+    newfd = open(new_incr_filepath, O_WRONLY | O_TRUNC | O_CREAT, 0644);
+    sdsfree(new_incr_filepath);
+    if (newfd == -1) {
+        serverLog(LL_WARNING, "Can't open the append-only file %s: %s", new_incr_filename, strerror(errno));
+        goto cleanup;
+    }
+
+    /* Change the AOF file type in 'incr_aof_list' from AOF_FILE_TYPE_INCR
+     * to AOF_FILE_TYPE_HIST, and move them to the 'history_aof_list'. */
+    markRewrittenIncrAofAsHistory(temp_am);
+
+    /* Persist AOF Manifest. */
+    if (persistAofManifest(temp_am) == C_ERR) {
+        serverLog(LL_WARNING, "Can't open the append-only file %s: %s", new_incr_filename, strerror(errno));
+        goto cleanup;
+    }
+    /* We can safely let `server.aof_manifest` point to 'temp_am' and free the previous one. */
+    aofManifestFreeAndUpdate(temp_am);
+
+    aofDelHistoryFiles();
+
+    atomic_store_explicit(&server.fsynced_reploff_pending, server.primary_repl_offset, memory_order_relaxed);
+    /* Update the fsynced replication offset that just now become valid. */
+    long long fsynced_reploff_pending =
+        atomic_load_explicit(&server.fsynced_reploff_pending, memory_order_relaxed);
+    server.fsynced_reploff = fsynced_reploff_pending;
+
+    /* If AOF fsync error in bio job, we just ignore it and log the event. */
+    int aof_bio_fsync_status = atomic_load_explicit(&server.aof_bio_fsync_status, memory_order_relaxed);
+    if (aof_bio_fsync_status == C_ERR) {
+        atomic_store_explicit(&server.aof_bio_fsync_status, C_OK, memory_order_relaxed);
+    }
+
+    /* If AOF was in error state, we just ignore it and log the event. */
+    if (server.aof_last_write_status == C_ERR) {
+        server.aof_last_write_status = C_OK;
+    }
+
+    server.aof_last_fsync = server.mstime;
+    /* Reset the aof_last_incr_size. */
+    server.aof_last_incr_size = 0;
+    /* Reset the aof_last_incr_fsync_offset. */
+    server.aof_last_incr_fsync_offset = 0;
+
+    server.aof_fd = newfd;
+    server.aof_rewrite_base_size = getAppendOnlyFileSize(new_base_filename, NULL);
+    server.aof_current_size = server.aof_rewrite_base_size;
+    server.aof_lastbgrewrite_status = C_OK;
+    server.stat_aofrw_consecutive_failures = 0;
+    server.aof_state = AOF_ON;
+
+    return C_OK;
+
+cleanup:
+    if (temp_am) aofManifestFree(temp_am);
+    if (new_base_filename) sdsfree(new_base_filename);
+    if (new_base_filepath) sdsfree(new_base_filepath);
+    if (new_incr_filename) sdsfree(new_incr_filename);
+    if (new_incr_filepath) sdsfree(new_incr_filepath);
+    if (newfd != -1) close(newfd);
+    return C_ERR;
+}
+
 /* This is a wrapper to the write syscall in order to retry on short writes
  * or if the syscall gets interrupted. It could look strange that we retry
  * on short writes given that we are writing to a block device: normally if

src/replication.c

@@ -2474,7 +2474,7 @@ void readSyncBulkPayload(connection *conn) {
         }
 
         /* Cleanup. */
-        if (server.rdb_del_sync_files && allPersistenceDisabled()) {
+        if (!(server.aof_enabled && server.aof_use_rdb_preamble) && server.rdb_del_sync_files && allPersistenceDisabled()) {
             serverLog(LL_NOTICE, "Removing the RDB file obtained from "
                                  "the primary. This replica has persistence "
                                  "disabled");
@@ -2524,7 +2524,9 @@ void readSyncBulkPayload(connection *conn) {
     /* Restart the AOF subsystem now that we finished the sync. This
      * will trigger an AOF rewrite, and when done will start appending
      * to the new file. */
-    if (server.aof_enabled) restartAOFAfterSYNC();
+    if (server.aof_enabled && server.aof_use_rdb_preamble) tryRetartAppendOnlyWithRdb();
+
+    if (server.aof_enabled && server.aof_state == AOF_OFF) restartAOFAfterSYNC();
 
     /* In case of dual channel replication sync we want to close the RDB connection
      * once the connection is established */
@@ -3872,7 +3874,6 @@ int connectWithPrimary(void) {
         return C_ERR;
     }
 
-
     server.repl_transfer_lastio = server.unixtime;
     server.repl_state = REPL_STATE_CONNECTING;
     serverLog(LL_NOTICE, "PRIMARY <-> REPLICA sync started");

src/server.h

@@ -2984,6 +2984,7 @@ int rewriteAppendOnlyFileBackground(void);
 int loadAppendOnlyFiles(aofManifest *am);
 void stopAppendOnly(void);
 int startAppendOnly(void);
+int tryRestartAppendOnlyWithRdb(void);
 void backgroundRewriteDoneHandler(int exitcode, int bysignal);
 void killAppendOnlyChild(void);
 void restartAOFAfterSYNC(void);