Repository: valkey-io/valkey
File: src/replication.c (C)

Code before:
                              "from socket, check server logs.");
        loadingFailed = 1;
    } else if (*usemark) {
        /* Verify the end mark is correct. */
        if (!rioRead(&rdb, buf, RDB_EOF_MARK_SIZE) || memcmp(buf, eofmark, RDB_EOF_MARK_SIZE) != 0) {
            serverLog(LL_WARNING, "Replication stream EOF marker is broken");
            loadingFailed = 1;
        }
    }

    if (loadingFailed) {
        stopLoading(0);
        rioFreeConn(&rdb, NULL);

        if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
            /* Discard potentially partially loaded tempDb. */
            moduleFireServerEvent(VALKEYMODULE_EVENT_REPL_ASYNC_LOAD, VALKEYMODULE_SUBEVENT_REPL_ASYNC_LOAD_ABORTED,
                                  NULL);

            disklessLoadDiscardTempDb(diskless_load_tempDb);
            disklessLoadDiscardFunctionsLibCtx(temp_functions_lib_ctx);
            serverLog(LL_NOTICE, "PRIMARY <-> REPLICA sync: Discarding temporary DB in background");
        } else {
            /* If we received RDB_INCOMPATIBLE, the old data was preserved */
            if (retval == RDB_INCOMPATIBLE) { 
                serverLog(LL_NOTICE, "PRIMARY <-> REPLICA sync: No data was loaded, skipping discard step");
            } else {
                /* Remove the half-loaded data in case the load failed for other reasons. */
                serverLog(LL_NOTICE, "PRIMARY <-> REPLICA sync: Discarding the half-loaded data");
                emptyData(-1, empty_db_flags, replicationEmptyDbCallback);
                }
        }

        /* Note that there's no point in restarting the AOF on SYNC
         * failure, it'll be restarted when sync succeeds or the replica
         * gets promoted. */
        return C_ERR;
    }

    /* RDB loading succeeded if we reach this point. */
    if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
        /* We will soon swap main db with tempDb and replicas will start
         * to apply data from new primary, we must discard the cached
         * primary structure and force resync of sub-replicas. */
        replicationAttachToNewPrimary();

        serverLog(LL_NOTICE, "PRIMARY <-> REPLICA sync: Swapping active DB with loaded DB");
        swapMainDbWithTempDb(diskless_load_tempDb);

        /* swap existing functions ctx with the temporary one */
        functionsLibCtxSwapWithCurrent(temp_functions_lib_ctx, 1);

Review comment (suggestion):
```suggestion
                serverLog(LL_NOTICE, "PRIMARY <-> REPLICA sync: RDB version incompatible, preserving old data");
```

Code after:
    moduleFireServerEvent(VALKEYMODULE_EVENT_PRIMARY_LINK_CHANGE, VALKEYMODULE_SUBEVENT_PRIMARY_LINK_UP, NULL);
    if (server.repl_state == REPL_STATE_CONNECTED) {
        /* After a full resynchronization we use the replication ID and
         * offset of the primary. The secondary ID / offset are cleared since
         * we are starting a new history. */
        memcpy(server.replid, server.primary->repl_data->replid, sizeof(server.replid));
        server.primary_repl_offset = server.primary->repl_data->reploff;
    }
    clearReplicationId2();

    /* Let's create the replication backlog if needed. Replicas need to
     * accumulate the backlog regardless of the fact they have sub-replicas
     * or not, in order to behave correctly if they are promoted to
     * primaries after a failover. */
    if (server.repl_backlog == NULL) createReplicationBacklog();
    serverLog(LL_NOTICE, "PRIMARY <-> REPLICA sync: Finished with success");

    if (server.supervised_mode == SUPERVISED_SYSTEMD) {
        serverCommunicateSystemd("STATUS=PRIMARY <-> REPLICA sync: Finished with success. Ready to accept connections "
                                 "in read-write mode.\n");
    }

    /* 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();

    /* In case of dual channel replication sync we want to close the RDB connection
     * once the connection is established */
    if (conn == server.repl_rdb_transfer_s) {
        connClose(conn);
        server.repl_rdb_transfer_s = NULL;
    }
}

int replicaLoadPrimaryRDBFromSocket(connection *conn, char *buf, char *eofmark, int *usemark, rdbSaveInfo *rsi) {
    rio rdb;
    serverDb **dbarray;
    functionsLibCtx *functions_lib_ctx;
    serverDb **diskless_load_tempDb = NULL;
    functionsLibCtx *temp_functions_lib_ctx = NULL;
    int empty_db_flags = server.repl_replica_lazy_flush ? EMPTYDB_ASYNC : EMPTYDB_NO_FLAGS;
    int asyncLoading = 0;

    if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
        /* Initialize empty tempDb dictionaries. */
        diskless_load_tempDb = disklessLoadInitTempDb();
        temp_functions_lib_ctx = disklessLoadFunctionsLibCtxCreate();

        moduleFireServerEvent(VALKEYMODULE_EVENT_REPL_ASYNC_LOAD, VALKEYMODULE_SUBEVENT_REPL_ASYNC_LOAD_STARTED, NULL);
        /* Async loading means we continue serving read commands during full resync, and