Remove testing data

1 files changed, 0 insertions, 5387 deletions

diff --git a/examples/redis-unstable/src/replication.c b/examples/redis-unstable/src/replication.c
deleted file mode 100644
index 309d6c4..0000000
--- a/examples/redis-unstable/src/replication.c
+++ /dev/null
@@ -1,5387 +0,0 @@
-/* Asynchronous replication implementation.
- *
- * Copyright (c) 2009-Present, Redis Ltd.
- * All rights reserved.
- *
- * Copyright (c) 2024-present, Valkey contributors.
- * All rights reserved.
- *
- * Licensed under your choice of (a) the Redis Source Available License 2.0
- * (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the
- * GNU Affero General Public License v3 (AGPLv3).
- *
- * Portions of this file are available under BSD3 terms; see REDISCONTRIBUTIONS for more information.
- */
-
-/*
- * replication.c - Replication Management
- *
- * This file contains the implementation of Redis's replication logic, which
- * enables data synchronization between master and replica instances.
- * It handles:
- * - Master-to-replica synchronization
- * - Full and partial resynchronizations
- * - Replication backlog management
- * - State machines for replica operations
- * - RDB Channel for Full Sync  (lookup "rdb channel for full sync")
- */
-
-#include "server.h"
-#include "cluster.h"
-#include "cluster_slot_stats.h"
-#include "bio.h"
-#include "functions.h"
-#include "connection.h"
-#include "cluster_asm.h"
-
-#include <memory.h>
-#include <sys/time.h>
-#include <unistd.h>
-#include <fcntl.h>
-#include <sys/socket.h>
-#include <sys/stat.h>
-
-void replicationDiscardCachedMaster(void);
-void replicationResurrectCachedMaster(connection *conn);
-void replicationSendAck(void);
-int replicaPutOnline(client *slave);
-void replicaStartCommandStream(client *slave);
-int cancelReplicationHandshake(int reconnect);
-static void rdbChannelFullSyncWithMaster(connection *conn);
-static int rdbChannelAbort(void);
-static void rdbChannelBufferReplData(connection *conn);
-static void rdbChannelReplDataBufInit(void);
-static void rdbChannelStreamReplDataToDb(void);
-static void rdbChannelCleanup(void);
-
-/* We take a global flag to remember if this instance generated an RDB
- * because of replication, so that we can remove the RDB file in case
- * the instance is configured to have no persistence. */
-int RDBGeneratedByReplication = 0;
-
-
-/* A reference to diskless loading rio to abort it asynchronously. It's needed
- * for rdbchannel replication. While loading from rdbchannel connection, we may
- * yield back to eventloop. If main channel connection detects a network problem
- * we want to abort loading. It calls rioAbort() in this case, so next rioRead()
- * from rdbchannel connection will return error to cancel loading safely. */
-static rio *disklessLoadingRio = NULL;
-
-/* --------------------------- Utility functions ---------------------------- */
-
-/* Returns 1 if the replica is rdbchannel and there is an associated main
- * channel slave with that. */
-int replicationCheckHasMainChannel(client *replica) {
-    if (!(replica->flags & CLIENT_REPL_RDB_CHANNEL) ||
-        !replica->main_ch_client_id ||
-        lookupClientByID(replica->main_ch_client_id) == NULL)
-    {
-        return 0;
-    }
-    return 1;
-}
-
-/* During rdb channel replication, replica opens two connections. From master
- * POV, these connections are distinct replicas in server.slaves. This function
- * counts associated replicas as one and returns logical replica count. */
-unsigned long replicationLogicalReplicaCount(void) {
-    unsigned long count = 0;
-    listNode *ln;
-    listIter li;
-
-    listRewind(server.slaves,&li);
-    while ((ln = listNext(&li))) {
-        client *replica = listNodeValue(ln);
-        if (!replicationCheckHasMainChannel(replica))
-            count++;
-    }
-    return count;
-}
-
-int replicaFromIOThreadHasPendingRead(client *c) {
-    serverAssert(c->tid != IOTHREAD_MAIN_THREAD_ID);
-
-    int pending_read;
-    atomicGetWithSync(c->pending_read, pending_read);
-    return pending_read;
-}
-
-/* Send replicas to their respective IO threads if it has pending reads or
- * writes. Otherwise it remains in main thread so it can check for new data in
- * the replication buffer ASAP. */
-void putReplicasInPendingClientsToIOThreads(void) {
-    if (server.io_threads_num <= 1) return;
-
-    serverAssert(pthread_equal(pthread_self(), server.main_thread_id));
-
-    listIter li;
-    listNode *ln;
-    listRewind(server.slaves,&li);
-    while((ln = listNext(&li))) {
-        client *replica = listNodeValue(ln);
-
-        /* We only care about replicas that need to run on IO thread but are
-         * currently in main */
-        if (replica->tid == IOTHREAD_MAIN_THREAD_ID ||
-            replica->running_tid != IOTHREAD_MAIN_THREAD_ID)
-        {
-            continue;
-        }
-
-        /* Skip the replica if it's scheduled for close */
-        if (replica->flags & CLIENT_CLOSE_ASAP) continue;
-
-        /* The call to clientHasPendingReplies may seem redundant but in the
-         * case of replica being in IO thread we can have the following case:
-         * replica gets back to main thread after sending the repl buffer it
-         * knows about. In the mean time main thread has accumulated new repl
-         * data. In that case the replica's client wouldn't have been put in
-         * the pending write queue but will still have new repl data it needs to
-         * send, so we make sure to check for that and send it back to IO thread
-         * if so. On the other hand if replica gets back to main thread before
-         * any new repl data has accumulated then after a new cmd is propagated
-         * the replica will be put in the pending write queue as usual so we
-         * need to check for that also.
-         * In addition, if the replica client has pending read events, we should
-         * also send them to the IO thread. */
-        if (replica->flags & CLIENT_PENDING_WRITE ||
-            clientHasPendingReplies(replica) ||
-            replicaFromIOThreadHasPendingRead(replica))
-        {
-            enqueuePendingClienstToIOThreads(replica);
-        }
-    }
-}
-
-/* Run some cron tasks for a connected master client. Return 1 when the client
- * is freed, 0 otherwise. */
-int replicationCronRunMasterClient(void) {
-    if (!server.masterhost || !server.master) return 0;
-
-    if (server.master->running_tid != IOTHREAD_MAIN_THREAD_ID) return 0;
-
-    /* Timed out master when we are an already connected slave? */
-    if (server.repl_state == REPL_STATE_CONNECTED &&
-        (time(NULL)-server.master->lastinteraction) > server.repl_timeout)
-    {
-        serverLog(LL_WARNING,"MASTER timeout: no data nor PING received...");
-        freeClient(server.master);
-        return 1;
-    }
-
-    /* Send ACK to master from time to time.
-     * Note that we do not send periodic acks to masters that don't
-     * support PSYNC and replication offsets. */
-    if (!(server.master->flags & CLIENT_PRE_PSYNC))
-        replicationSendAck();
-
-    return 0;
-}
-
-ConnectionType *connTypeOfReplication(void) {
-    if (server.tls_replication) {
-        return connectionTypeTls();
-    }
-
-    return connectionTypeTcp();
-}
-
-/* Return the pointer to a string representing the slave ip:listening_port
- * pair. Mostly useful for logging, since we want to log a slave using its
- * IP address and its listening port which is more clear for the user, for
- * example: "Closing connection with replica 10.1.2.3:6380". */
-char *replicationGetSlaveName(client *c) {
-    static char buf[NET_HOST_PORT_STR_LEN];
-    char ip[NET_IP_STR_LEN];
-
-    ip[0] = '\0';
-    buf[0] = '\0';
-    if (c->slave_addr ||
-        connAddrPeerName(c->conn,ip,sizeof(ip),NULL) != -1)
-    {
-        char *addr = c->slave_addr ? c->slave_addr : ip;
-        if (c->slave_listening_port)
-            formatAddr(buf,sizeof(buf),addr,c->slave_listening_port);
-        else
-            snprintf(buf,sizeof(buf),"%s:<unknown-replica-port>",addr);
-    } else {
-        snprintf(buf,sizeof(buf),"client id #%llu",
-            (unsigned long long) c->id);
-    }
-    return buf;
-}
-
-/* Plain unlink() can block for quite some time in order to actually apply
- * the file deletion to the filesystem. This call removes the file in a
- * background thread instead. We actually just do close() in the thread,
- * by using the fact that if there is another instance of the same file open,
- * the foreground unlink() will only remove the fs name, and deleting the
- * file's storage space will only happen once the last reference is lost. */
-int bg_unlink(const char *filename) {
-    int fd = open(filename,O_RDONLY|O_NONBLOCK);
-    if (fd == -1) {
-        /* Can't open the file? Fall back to unlinking in the main thread. */
-        return unlink(filename);
-    } else {
-        /* The following unlink() removes the name but doesn't free the
-         * file contents because a process still has it open. */
-        int retval = unlink(filename);
-        if (retval == -1) {
-            /* If we got an unlink error, we just return it, closing the
-             * new reference we have to the file. */
-            int old_errno = errno;
-            close(fd);  /* This would overwrite our errno. So we saved it. */
-            errno = old_errno;
-            return -1;
-        }
-        bioCreateCloseJob(fd, 0, 0);
-        return 0; /* Success. */
-    }
-}
-
-/* ---------------------------------- MASTER -------------------------------- */
-
-void createReplicationBacklog(void) {
-    serverAssert(server.repl_backlog == NULL);
-    server.repl_backlog = zmalloc(sizeof(replBacklog));
-    server.repl_backlog->ref_repl_buf_node = NULL;
-    server.repl_backlog->unindexed_count = 0;
-    server.repl_backlog->blocks_index = raxNew();
-    server.repl_backlog->histlen = 0;
-    /* We don't have any data inside our buffer, but virtually the first
-     * byte we have is the next byte that will be generated for the
-     * replication stream. */
-    server.repl_backlog->offset = server.master_repl_offset+1;
-}
-
-/* This function is called when the user modifies the replication backlog
- * size at runtime. It is up to the function to resize the buffer and setup it
- * so that it contains the same data as the previous one (possibly less data,
- * but the most recent bytes, or the same data and more free space in case the
- * buffer is enlarged). */
-void resizeReplicationBacklog(void) {
-    if (server.repl_backlog_size < CONFIG_REPL_BACKLOG_MIN_SIZE)
-        server.repl_backlog_size = CONFIG_REPL_BACKLOG_MIN_SIZE;
-    if (server.repl_backlog)
-        incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL);
-}
-
-void freeReplicationBacklog(void) {
-    serverAssert(listLength(server.slaves) == 0);
-    if (server.repl_backlog == NULL) return;
-
-    /* Decrease the start buffer node reference count. */
-    if (server.repl_backlog->ref_repl_buf_node) {
-        replBufBlock *o = listNodeValue(
-            server.repl_backlog->ref_repl_buf_node);
-        serverAssert(o->refcount == 1); /* Last reference. */
-        o->refcount--;
-    }
-
-    /* Replication buffer blocks are completely released when we free the
-     * backlog, since the backlog is released only when there are no replicas
-     * and the backlog keeps the last reference of all blocks. */
-    freeReplicationBacklogRefMemAsync(server.repl_buffer_blocks,
-                            server.repl_backlog->blocks_index);
-    resetReplicationBuffer();
-    zfree(server.repl_backlog);
-    server.repl_backlog = NULL;
-}
-
-/* To make search offset from replication buffer blocks quickly
- * when replicas ask partial resynchronization, we create one index
- * block every REPL_BACKLOG_INDEX_PER_BLOCKS blocks. */
-void createReplicationBacklogIndex(listNode *ln) {
-    server.repl_backlog->unindexed_count++;
-    if (server.repl_backlog->unindexed_count >= REPL_BACKLOG_INDEX_PER_BLOCKS) {
-        replBufBlock *o = listNodeValue(ln);
-        uint64_t encoded_offset = htonu64(o->repl_offset);
-        raxInsert(server.repl_backlog->blocks_index,
-                  (unsigned char*)&encoded_offset, sizeof(uint64_t),
-                  ln, NULL);
-        server.repl_backlog->unindexed_count = 0;
-    }
-}
-
-/* Rebase replication buffer blocks' offset since the initial
- * setting offset starts from 0 when master restart. */
-void rebaseReplicationBuffer(long long base_repl_offset) {
-    raxFree(server.repl_backlog->blocks_index);
-    server.repl_backlog->blocks_index = raxNew();
-    server.repl_backlog->unindexed_count = 0;
-
-    listIter li;
-    listNode *ln;
-    listRewind(server.repl_buffer_blocks, &li);
-    while ((ln = listNext(&li))) {
-        replBufBlock *o = listNodeValue(ln);
-        o->repl_offset += base_repl_offset;
-        createReplicationBacklogIndex(ln);
-    }
-}
-
-void resetReplicationBuffer(void) {
-    server.repl_buffer_mem = 0;
-    server.repl_buffer_blocks = listCreate();
-    listSetFreeMethod(server.repl_buffer_blocks, zfree);
-}
-
-int canFeedReplicaReplBuffer(client *replica) {
-    /* Don't feed replicas that only want the RDB or main channels of migration
-     * destinations which need filtered stream for migrating slot ranges. */
-    if (replica->flags & CLIENT_REPL_RDBONLY ||
-        replica->flags & CLIENT_ASM_MIGRATING) return 0;
-
-    /* Don't feed replicas that are still waiting for BGSAVE to start. */
-    if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_START ||
-        replica->replstate == SLAVE_STATE_WAIT_RDB_CHANNEL) return 0;
-
-    /* Don't feed replicas that are going to be closed ASAP. */
-    if (replica->flags & CLIENT_CLOSE_ASAP) return 0;
-
-    return 1;
-}
-
-/* Create the replication backlog if needed. */
-void createReplicationBacklogIfNeeded(void) {
-    if (listLength(server.slaves) == 1 && server.repl_backlog == NULL) {
-        /* When we create the backlog from scratch, we always use a new
-         * replication ID and clear the ID2, since there is no valid
-         * past history. */
-        changeReplicationId();
-        clearReplicationId2();
-        createReplicationBacklog();
-        serverLog(LL_NOTICE,"Replication backlog created, my new "
-                            "replication IDs are '%s' and '%s'",
-                            server.replid, server.replid2);
-    }
-}
-/* Similar with 'prepareClientToWrite', note that we must call this function
- * before feeding replication stream into global replication buffer, since
- * clientHasPendingReplies in prepareClientToWrite will access the global
- * replication buffer to make judgements. */
-int prepareReplicasToWrite(void) {
-    listIter li;
-    listNode *ln;
-    int prepared = 0;
-
-    listRewind(server.slaves,&li);
-    while((ln = listNext(&li))) {
-        client *slave = ln->value;
-        if (!canFeedReplicaReplBuffer(slave)) continue;
-        if (prepareClientToWrite(slave) == C_ERR) continue;
-        prepared++;
-    }
-
-    return prepared;
-}
-
-/* Wrapper for feedReplicationBuffer() that takes Redis string objects
- * as input. */
-void feedReplicationBufferWithObject(robj *o) {
-    char llstr[LONG_STR_SIZE];
-    void *p;
-    size_t len;
-
-    if (o->encoding == OBJ_ENCODING_INT) {
-        len = ll2string(llstr,sizeof(llstr),(long)o->ptr);
-        p = llstr;
-    } else {
-        len = sdslen(o->ptr);
-        p = o->ptr;
-    }
-    feedReplicationBuffer(p,len);
-}
-
-/* Generally, we only have one replication buffer block to trim when replication
- * backlog size exceeds our setting and no replica reference it. But if replica
- * clients disconnect, we need to free many replication buffer blocks that are
- * referenced. It would cost much time if there are a lots blocks to free, that
- * will freeze server, so we trim replication backlog incrementally. */
-void incrementalTrimReplicationBacklog(size_t max_blocks) {
-    serverAssert(server.repl_backlog != NULL);
-
-    size_t trimmed_blocks = 0;
-    while (server.repl_backlog->histlen > server.repl_backlog_size &&
-           trimmed_blocks < max_blocks)
-    {
-        /* We never trim backlog to less than one block. */
-        if (listLength(server.repl_buffer_blocks) <= 1) break;
-
-        /* Replicas increment the refcount of the first replication buffer block
-         * they refer to, in that case, we don't trim the backlog even if
-         * backlog_histlen exceeds backlog_size. This implicitly makes backlog
-         * bigger than our setting, but makes the master accept partial resync as
-         * much as possible. So that backlog must be the last reference of
-         * replication buffer blocks. */
-        listNode *first = listFirst(server.repl_buffer_blocks);
-        serverAssert(first == server.repl_backlog->ref_repl_buf_node);
-        replBufBlock *fo = listNodeValue(first);
-        if (fo->refcount != 1) break;
-
-        /* We don't try trim backlog if backlog valid size will be lessen than
-         * setting backlog size once we release the first repl buffer block. */
-        if (server.repl_backlog->histlen - (long long)fo->size <=
-            server.repl_backlog_size) break;
-
-        /* Decr refcount and release the first block later. */
-        fo->refcount--;
-        trimmed_blocks++;
-        server.repl_backlog->histlen -= fo->size;
-
-        /* Go to use next replication buffer block node. */
-        listNode *next = listNextNode(first);
-        server.repl_backlog->ref_repl_buf_node = next;
-        serverAssert(server.repl_backlog->ref_repl_buf_node != NULL);
-        /* Incr reference count to keep the new head node. */
-        ((replBufBlock *)listNodeValue(next))->refcount++;
-
-        /* Remove the node in recorded blocks. */
-        uint64_t encoded_offset = htonu64(fo->repl_offset);
-        raxRemove(server.repl_backlog->blocks_index,
-            (unsigned char*)&encoded_offset, sizeof(uint64_t), NULL);
-
-        /* Delete the first node from global replication buffer. */
-        serverAssert(fo->refcount == 0 && fo->used == fo->size);
-        server.repl_buffer_mem -= (fo->size +
-            sizeof(listNode) + sizeof(replBufBlock));
-        listDelNode(server.repl_buffer_blocks, first);
-    }
-
-    /* Set the offset of the first byte we have in the backlog. */
-    server.repl_backlog->offset = server.master_repl_offset -
-                              server.repl_backlog->histlen + 1;
-}
-
-/* Free replication buffer blocks that are referenced by this client. */
-void freeReplicaReferencedReplBuffer(client *replica) {
-    serverAssert(replica->running_tid == IOTHREAD_MAIN_THREAD_ID);
-
-    if (replica->ref_repl_buf_node != NULL) {
-        /* Decrease the start buffer node reference count. */
-        replBufBlock *o = listNodeValue(replica->ref_repl_buf_node);
-        serverAssert(o->refcount > 0);
-        o->refcount--;
-        incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL);
-    }
-    replica->ref_repl_buf_node = NULL;
-    replica->ref_block_pos = 0;
-}
-
-/* Append bytes into the global replication buffer list, replication backlog and
- * all replica clients use replication buffers collectively, this function replace
- * 'addReply*', 'feedReplicationBacklog' for replicas and replication backlog,
- * First we add buffer into global replication buffer block list, and then
- * update replica / replication-backlog referenced node and block position. */
-void feedReplicationBuffer(char *s, size_t len) {
-    static long long repl_block_id = 0;
-
-    if (server.repl_backlog == NULL) return;
-
-    clusterSlotStatsIncrNetworkBytesOutForReplication(len);
-
-    /* Update the current cmd's keys with the commands replication bytes*/
-    hotkeyMetrics metrics = {0, len};
-    hotkeyStatsUpdateCurrentCmd(server.hotkeys, metrics);
-
-    while(len > 0) {
-        size_t start_pos = 0; /* The position of referenced block to start sending. */
-        listNode *start_node = NULL; /* Replica/backlog starts referenced node. */
-        int add_new_block = 0; /* Create new block if current block is total used. */
-        listNode *ln = listLast(server.repl_buffer_blocks);
-        replBufBlock *tail = ln ? listNodeValue(ln) : NULL;
-
-        /* Append to tail string when possible. */
-        if (tail && tail->size > tail->used) {
-            start_node = listLast(server.repl_buffer_blocks);
-            start_pos = tail->used;
-            /* Copy the part we can fit into the tail, and leave the rest for a
-             * new node */
-            size_t avail = tail->size - tail->used;
-            size_t copy = (avail >= len) ? len : avail;
-            memcpy(tail->buf + tail->used, s, copy);
-            tail->used += copy;
-            s += copy;
-            len -= copy;
-            server.master_repl_offset += copy;
-            server.repl_backlog->histlen += copy;
-        }
-        if (len) {
-            /* Create a new node, make sure it is allocated to at
-             * least PROTO_REPLY_CHUNK_BYTES */
-            size_t usable_size;
-            /* Avoid creating nodes smaller than PROTO_REPLY_CHUNK_BYTES, so that we can append more data into them,
-             * and also avoid creating nodes bigger than repl_backlog_size / 16, so that we won't have huge nodes that can't
-             * trim when we only still need to hold a small portion from them. */
-            size_t limit = max((size_t)server.repl_backlog_size / 16, (size_t)PROTO_REPLY_CHUNK_BYTES);
-            size_t size = min(max(len, (size_t)PROTO_REPLY_CHUNK_BYTES), limit);
-            tail = zmalloc_usable(size + sizeof(replBufBlock), &usable_size);
-            /* Take over the allocation's internal fragmentation */
-            tail->size = usable_size - sizeof(replBufBlock);
-            size_t copy = (tail->size >= len) ? len : tail->size;
-            tail->used = copy;
-            tail->refcount = 0;
-            tail->repl_offset = server.master_repl_offset + 1;
-            tail->id = repl_block_id++;
-            memcpy(tail->buf, s, copy);
-            listAddNodeTail(server.repl_buffer_blocks, tail);
-            /* We also count the list node memory into replication buffer memory. */
-            server.repl_buffer_mem += (usable_size + sizeof(listNode));
-            add_new_block = 1;
-            if (start_node == NULL) {
-                start_node = listLast(server.repl_buffer_blocks);
-                start_pos = 0;
-            }
-            s += copy;
-            len -= copy;
-            server.master_repl_offset += copy;
-            server.repl_backlog->histlen += copy;
-        }
-
-        /* For output buffer of replicas. */
-        listIter li;
-        listRewind(server.slaves,&li);
-        while((ln = listNext(&li))) {
-            client *slave = ln->value;
-            if (!canFeedReplicaReplBuffer(slave)) continue;
-
-            /* Update shared replication buffer start position. */
-            if (slave->ref_repl_buf_node == NULL) {
-                slave->ref_repl_buf_node = start_node;
-                slave->ref_block_pos = start_pos;
-                /* Only increase the start block reference count. */
-                ((replBufBlock *)listNodeValue(start_node))->refcount++;
-            }
-
-            /* Check output buffer limit only when add new block. */
-            if (add_new_block) closeClientOnOutputBufferLimitReached(slave, 1);
-        }
-
-        /* For replication backlog */
-        if (server.repl_backlog->ref_repl_buf_node == NULL) {
-            server.repl_backlog->ref_repl_buf_node = start_node;
-            /* Only increase the start block reference count. */
-            ((replBufBlock *)listNodeValue(start_node))->refcount++;
-
-            /* Replication buffer must be empty before adding replication stream
-             * into replication backlog. */
-            serverAssert(add_new_block == 1 && start_pos == 0);
-        }
-        if (add_new_block) {
-            createReplicationBacklogIndex(listLast(server.repl_buffer_blocks));
-
-            /* It is important to trim after adding replication data to keep the backlog size close to
-             * repl_backlog_size in the common case. We wait until we add a new block to avoid repeated
-             * unnecessary trimming attempts when small amounts of data are added. See comments in
-             * freeMemoryGetNotCountedMemory() for details on replication backlog memory tracking. */
-            incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL);
-        }
-    }
-}
-
-/* Propagate write commands to replication stream.
- *
- * This function is used if the instance is a master: we use the commands
- * received by our clients in order to create the replication stream.
- * Instead if the instance is a replica and has sub-replicas attached, we use
- * replicationFeedStreamFromMasterStream() */
-void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
-    int j, len;
-    char llstr[LONG_STR_SIZE];
-
-    /* In case we propagate a command that doesn't touch keys (PING, REPLCONF) we
-     * pass dbid=-1 that indicate there is no need to replicate `select` command. */
-    serverAssert(dictid == -1 || (dictid >= 0 && dictid < server.dbnum));
-
-    /* If the instance is not a top level master, return ASAP: we'll just proxy
-     * the stream of data we receive from our master instead, in order to
-     * propagate *identical* replication stream. In this way this slave can
-     * advertise the same replication ID as the master (since it shares the
-     * master replication history and has the same backlog and offsets). */
-    if (server.masterhost != NULL) return;
-
-    /* If current client is marked as master, we will proxy the command stream
-     * to our slaves instead of replicating them, that also happens when being
-     * in atomic slot migration. */
-    if (server.current_client && server.current_client->flags & CLIENT_MASTER) return;
-
-    /* If there aren't slaves, and there is no backlog buffer to populate,
-     * we can return ASAP. */
-    if (server.repl_backlog == NULL && listLength(slaves) == 0) {
-        /* We increment the repl_offset anyway, since we use that for tracking AOF fsyncs
-         * even when there's no replication active. This code will not be reached if AOF
-         * is also disabled. */
-        server.master_repl_offset += 1;
-        return;
-    }
-
-    /* We can't have slaves attached and no backlog. */
-    serverAssert(!(listLength(slaves) != 0 && server.repl_backlog == NULL));
-
-    /* Update the time of sending replication stream to replicas. */
-    server.repl_stream_lastio = server.unixtime;
-
-    /* Must install write handler for all replicas first before feeding
-     * replication stream. */
-    prepareReplicasToWrite();
-
-    /* Send SELECT command to every slave if needed. */
-    if (dictid != -1 && server.slaveseldb != dictid) {
-        robj *selectcmd;
-
-        /* For a few DBs we have pre-computed SELECT command. */
-        if (dictid >= 0 && dictid < PROTO_SHARED_SELECT_CMDS) {
-            selectcmd = shared.select[dictid];
-        } else {
-            int dictid_len;
-
-            dictid_len = ll2string(llstr,sizeof(llstr),dictid);
-            selectcmd = createObject(OBJ_STRING,
-                sdscatprintf(sdsempty(),
-                "*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n",
-                dictid_len, llstr));
-        }
-
-        feedReplicationBufferWithObject(selectcmd);
-
-        /* Although the SELECT command is not associated with any slot,
-         * its per-slot network-bytes-out accumulation is made by the above function call.
-         * To cancel-out this accumulation, below adjustment is made. */
-        clusterSlotStatsDecrNetworkBytesOutForReplication(sdslen(selectcmd->ptr));
-
-        if (dictid < 0 || dictid >= PROTO_SHARED_SELECT_CMDS)
-            decrRefCount(selectcmd);
-
-        server.slaveseldb = dictid;
-    }
-
-    /* Write the command to the replication buffer if any. */
-    char aux[LONG_STR_SIZE+3];
-
-    /* Add the multi bulk reply length. */
-    aux[0] = '*';
-    len = ll2string(aux+1,sizeof(aux)-1,argc);
-    aux[len+1] = '\r';
-    aux[len+2] = '\n';
-    feedReplicationBuffer(aux,len+3);
-
-    for (j = 0; j < argc; j++) {
-        long objlen = stringObjectLen(argv[j]);
-
-        /* We need to feed the buffer with the object as a bulk reply
-         * not just as a plain string, so create the $..CRLF payload len
-         * and add the final CRLF */
-        aux[0] = '$';
-        len = ll2string(aux+1,sizeof(aux)-1,objlen);
-        aux[len+1] = '\r';
-        aux[len+2] = '\n';
-        feedReplicationBuffer(aux,len+3);
-        feedReplicationBufferWithObject(argv[j]);
-        feedReplicationBuffer(aux+len+1,2);
-    }
-}
-
-/* This is a debugging function that gets called when we detect something
- * wrong with the replication protocol: the goal is to peek into the
- * replication backlog and show a few final bytes to make simpler to
- * guess what kind of bug it could be. */
-void showLatestBacklog(void) {
-    if (server.repl_backlog == NULL) return;
-    if (listLength(server.repl_buffer_blocks) == 0) return;
-    if (server.hide_user_data_from_log) {
-        serverLog(LL_NOTICE,"hide-user-data-from-log is on, skip logging backlog content to avoid spilling PII.");
-        return;
-    }
-
-    size_t dumplen = 256;
-    if (server.repl_backlog->histlen < (long long)dumplen)
-        dumplen = server.repl_backlog->histlen;
-
-    sds dump = sdsempty();
-    listNode *node = listLast(server.repl_buffer_blocks);
-    while(dumplen) {
-        if (node == NULL) break;
-        replBufBlock *o = listNodeValue(node);
-        size_t thislen = o->used >= dumplen ? dumplen : o->used;
-        sds head = sdscatrepr(sdsempty(), o->buf+o->used-thislen, thislen);
-        sds tmp = sdscatsds(head, dump);
-        sdsfree(dump);
-        dump = tmp;
-        dumplen -= thislen;
-        node = listPrevNode(node);
-    }
-
-    /* Finally log such bytes: this is vital debugging info to
-     * understand what happened. */
-    serverLog(LL_NOTICE,"Latest backlog is: '%s'", dump);
-    sdsfree(dump);
-}
-
-/* This function is used in order to proxy what we receive from our master
- * to our sub-slaves. Besides, we also proxy the replication stream from
- * the source node when being in atomic slot migration. */
-void replicationFeedStreamFromMasterStream(char *buf, size_t buflen) {
-    /* There must be replication backlog if having attached slaves. */
-    if (listLength(server.slaves)) serverAssert(server.repl_backlog != NULL);
-    if (server.repl_backlog) {
-        /* Must install write handler for all replicas first before feeding
-         * replication stream. */
-        prepareReplicasToWrite();
-        feedReplicationBuffer(buf,buflen);
-    } else if (server.masterhost == NULL && server.aof_enabled) {
-        /* We increment the repl_offset anyway, since we use that for tracking
-         * AOF fsyncs even when there's no replication active. This code will
-         * not be reached if AOF is also disabled.
-         *
-         * As we skip feeding the replication buffer in atomic slot migration,
-         * so here we need to update the replication offset manually. */
-        server.master_repl_offset += 1;
-    }
-}
-
-void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv, int argc) {
-    /* Fast path to return if the monitors list is empty or the server is in loading. */
-    if (monitors == NULL || listLength(monitors) == 0 || server.loading) return;
-    listNode *ln;
-    listIter li;
-    int j;
-    sds cmdrepr = sdsnew("+");
-    robj *cmdobj;
-    struct timeval tv;
-
-    gettimeofday(&tv,NULL);
-    cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec);
-    if (c->flags & CLIENT_SCRIPT) {
-        cmdrepr = sdscatprintf(cmdrepr,"[%d lua] ",dictid);
-    } else if (c->flags & CLIENT_UNIX_SOCKET) {
-        cmdrepr = sdscatprintf(cmdrepr,"[%d unix:%s] ",dictid,server.unixsocket);
-    } else {
-        cmdrepr = sdscatprintf(cmdrepr,"[%d %s] ",dictid,getClientPeerId(c));
-    }
-
-    for (j = 0; j < argc; j++) {
-        if (argv[j]->encoding == OBJ_ENCODING_INT) {
-            cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)argv[j]->ptr);
-        } else {
-            cmdrepr = sdscatrepr(cmdrepr,(char*)argv[j]->ptr,
-                        sdslen(argv[j]->ptr));
-        }
-        if (j != argc-1)
-            cmdrepr = sdscatlen(cmdrepr," ",1);
-    }
-    cmdrepr = sdscatlen(cmdrepr,"\r\n",2);
-    cmdobj = createObject(OBJ_STRING,cmdrepr);
-
-    listRewind(monitors,&li);
-    while((ln = listNext(&li))) {
-        client *monitor = ln->value;
-        /* Do not show internal commands to non-internal clients. */
-        if (c->realcmd && (c->realcmd->flags & CMD_INTERNAL) && !(monitor->flags & CLIENT_INTERNAL)) {
-            continue;
-        }
-        addReply(monitor,cmdobj);
-        updateClientMemUsageAndBucket(monitor);
-    }
-    decrRefCount(cmdobj);
-}
-
-/* Feed the slave 'c' with the replication backlog starting from the
- * specified 'offset' up to the end of the backlog. */
-long long addReplyReplicationBacklog(client *c, long long offset) {
-    serverAssert(c->running_tid == IOTHREAD_MAIN_THREAD_ID);
-
-    long long skip;
-
-    serverLog(LL_DEBUG, "[PSYNC] Replica request offset: %lld", offset);
-
-    if (server.repl_backlog->histlen == 0) {
-        serverLog(LL_DEBUG, "[PSYNC] Backlog history len is zero");
-        return 0;
-    }
-
-    serverLog(LL_DEBUG, "[PSYNC] Backlog size: %lld",
-             server.repl_backlog_size);
-    serverLog(LL_DEBUG, "[PSYNC] First byte: %lld",
-             server.repl_backlog->offset);
-    serverLog(LL_DEBUG, "[PSYNC] History len: %lld",
-             server.repl_backlog->histlen);
-
-    /* Compute the amount of bytes we need to discard. */
-    skip = offset - server.repl_backlog->offset;
-    serverLog(LL_DEBUG, "[PSYNC] Skipping: %lld", skip);
-
-    /* Iterate recorded blocks, quickly search the approximate node. */
-    listNode *node = NULL;
-    if (raxSize(server.repl_backlog->blocks_index) > 0) {
-        uint64_t encoded_offset = htonu64(offset);
-        raxIterator ri;
-        raxStart(&ri, server.repl_backlog->blocks_index);
-        raxSeek(&ri, ">", (unsigned char*)&encoded_offset, sizeof(uint64_t));
-        if (raxEOF(&ri)) {
-            /* No found, so search from the last recorded node. */
-            raxSeek(&ri, "$", NULL, 0);
-            raxPrev(&ri);
-            node = (listNode *)ri.data;
-        } else {
-            raxPrev(&ri); /* Skip the sought node. */
-            /* We should search from the prev node since the offset of current
-             * sought node exceeds searching offset. */
-            if (raxPrev(&ri))
-                node = (listNode *)ri.data;
-            else
-                node = server.repl_backlog->ref_repl_buf_node;
-        }
-        raxStop(&ri);
-    } else {
-        /* No recorded blocks, just from the start node to search. */
-        node = server.repl_backlog->ref_repl_buf_node;
-    }
-
-    /* Search the exact node. */
-    while (node != NULL) {
-        replBufBlock *o = listNodeValue(node);
-        if (o->repl_offset + (long long)o->used >= offset) break;
-        node = listNextNode(node);
-    }
-    serverAssert(node != NULL);
-
-    /* Install a writer handler first.*/
-    prepareClientToWrite(c);
-    /* Setting output buffer of the replica. */
-    replBufBlock *o = listNodeValue(node);
-    o->refcount++;
-    c->ref_repl_buf_node = node;
-    c->ref_block_pos = offset - o->repl_offset;
-
-    return server.repl_backlog->histlen - skip;
-}
-
-/* Return the offset to provide as reply to the PSYNC command received
- * from the slave. The returned value is only valid immediately after
- * the BGSAVE process started and before executing any other command
- * from clients. */
-long long getPsyncInitialOffset(void) {
-    return server.master_repl_offset;
-}
-
-/* Send a FULLRESYNC reply in the specific case of a full resynchronization,
- * as a side effect setup the slave for a full sync in different ways:
- *
- * 1) Remember, into the slave client structure, the replication offset
- *    we sent here, so that if new slaves will later attach to the same
- *    background RDB saving process (by duplicating this client output
- *    buffer), we can get the right offset from this slave.
- * 2) Set the replication state of the slave to WAIT_BGSAVE_END so that
- *    we start accumulating differences from this point.
- * 3) Force the replication stream to re-emit a SELECT statement so
- *    the new slave incremental differences will start selecting the
- *    right database number.
- *
- * Normally this function should be called immediately after a successful
- * BGSAVE for replication was started, or when there is one already in
- * progress that we attached our slave to. */
-int replicationSetupSlaveForFullResync(client *slave, long long offset) {
-    char buf[128];
-    int buflen;
-
-    slave->psync_initial_offset = offset;
-    slave->replstate = SLAVE_STATE_WAIT_BGSAVE_END;
-    /* We are going to accumulate the incremental changes for this
-     * slave as well. Set slaveseldb to -1 in order to force to re-emit
-     * a SELECT statement in the replication stream. */
-    server.slaveseldb = -1;
-
-    /* Slots snapshot. */
-    if (slave->flags & CLIENT_REPL_RDB_CHANNEL &&
-        slave->slave_req & SLAVE_REQ_SLOTS_SNAPSHOT)
-    {
-        /* Start to deliver the commands stream on migrating slots. */
-        asmSlotSnapshotAndStreamStart(slave->task);
-
-        buflen = snprintf(buf, sizeof(buf), "+SLOTSSNAPSHOT\r\n");
-        if (connWrite(slave->conn, buf, buflen) != buflen) {
-            freeClientAsync(slave);
-            return C_ERR;
-        }
-        return C_OK;
-    }
-
-    /* Don't send this reply to slaves that approached us with
-     * the old SYNC command. */
-    if (!(slave->flags & CLIENT_PRE_PSYNC)) {
-        if (slave->flags & CLIENT_REPL_RDB_CHANNEL) {
-            /* This slave is rdbchannel. Find its associated main channel and
-             * change its state so we can deliver replication stream from now
-             * on, in parallel to rdb. */
-            uint64_t id = slave->main_ch_client_id;
-            client *c = lookupClientByID(id);
-            if (c && c->replstate == SLAVE_STATE_WAIT_RDB_CHANNEL) {
-                c->replstate = SLAVE_STATE_SEND_BULK_AND_STREAM;
-                serverLog(LL_NOTICE, "Starting to deliver RDB and replication stream to replica: %s",
-                          replicationGetSlaveName(c));
-            } else {
-                serverLog(LL_WARNING, "Starting to deliver RDB to replica %s"
-                                      " but it has no associated main channel",
-                                      replicationGetSlaveName(slave));
-            }
-        }
-        buflen = snprintf(buf,sizeof(buf),"+FULLRESYNC %s %lld\r\n",
-                          server.replid,offset);
-        if (connWrite(slave->conn,buf,buflen) != buflen) {
-            freeClientAsync(slave);
-            return C_ERR;
-        }
-    }
-    return C_OK;
-}
-
-/* This function handles the PSYNC command from the point of view of a
- * master receiving a request for partial resynchronization.
- *
- * On success return C_OK, otherwise C_ERR is returned and we proceed
- * with the usual full resync. */
-int masterTryPartialResynchronization(client *c, long long psync_offset) {
-    long long psync_len;
-    char *master_replid = c->argv[1]->ptr;
-    char buf[128];
-    int buflen;
-
-    /* Is the replication ID of this master the same advertised by the wannabe
-     * slave via PSYNC? If the replication ID changed this master has a
-     * different replication history, and there is no way to continue.
-     *
-     * Note that there are two potentially valid replication IDs: the ID1
-     * and the ID2. The ID2 however is only valid up to a specific offset. */
-    if (strcasecmp(master_replid, server.replid) &&
-        (strcasecmp(master_replid, server.replid2) ||
-         psync_offset > server.second_replid_offset))
-    {
-        /* Replid "?" is used by slaves that want to force a full resync. */
-        if (master_replid[0] != '?') {
-            if (strcasecmp(master_replid, server.replid) &&
-                strcasecmp(master_replid, server.replid2))
-            {
-                serverLog(LL_NOTICE,"Partial resynchronization not accepted: "
-                    "Replication ID mismatch (Replica asked for '%s', my "
-                    "replication IDs are '%s' and '%s')",
-                    master_replid, server.replid, server.replid2);
-            } else {
-                serverLog(LL_NOTICE,"Partial resynchronization not accepted: "
-                    "Requested offset for second ID was %lld, but I can reply "
-                    "up to %lld", psync_offset, server.second_replid_offset);
-            }
-        } else {
-            serverLog(LL_NOTICE,"Full resync requested by replica %s %s",
-                replicationGetSlaveName(c),
-                c->flags & CLIENT_REPL_RDB_CHANNEL ? "(rdb-channel)" : "");
-        }
-        goto need_full_resync;
-    }
-
-    /* We still have the data our slave is asking for? */
-    if (!server.repl_backlog ||
-        psync_offset < server.repl_backlog->offset ||
-        psync_offset > (server.repl_backlog->offset + server.repl_backlog->histlen))
-    {
-        serverLog(LL_NOTICE,
-            "Unable to partial resync with replica %s for lack of backlog (Replica request was: %lld).", replicationGetSlaveName(c), psync_offset);
-        if (psync_offset > server.master_repl_offset) {
-            serverLog(LL_WARNING,
-                "Warning: replica %s tried to PSYNC with an offset that is greater than the master replication offset.", replicationGetSlaveName(c));
-        }
-        goto need_full_resync;
-    }
-
-    /* If we reached this point, we are able to perform a partial resync:
-     * 1) Set client state to make it a slave.
-     * 2) Inform the client we can continue with +CONTINUE
-     * 3) Send the backlog data (from the offset to the end) to the slave. */
-    c->flags |= CLIENT_SLAVE;
-    c->replstate = SLAVE_STATE_ONLINE;
-    c->repl_ack_time = server.unixtime;
-    c->repl_start_cmd_stream_on_ack = 0;
-    listAddNodeTail(server.slaves,c);
-    /* We can't use the connection buffers since they are used to accumulate
-     * new commands at this stage. But we are sure the socket send buffer is
-     * empty so this write will never fail actually. */
-    if (c->slave_capa & SLAVE_CAPA_PSYNC2) {
-        buflen = snprintf(buf,sizeof(buf),"+CONTINUE %s\r\n", server.replid);
-    } else {
-        buflen = snprintf(buf,sizeof(buf),"+CONTINUE\r\n");
-    }
-    if (connWrite(c->conn,buf,buflen) != buflen) {
-        freeClientAsync(c);
-        return C_OK;
-    }
-    psync_len = addReplyReplicationBacklog(c,psync_offset);
-    serverLog(LL_NOTICE,
-        "Partial resynchronization request from %s accepted. Sending %lld bytes of backlog starting from offset %lld.",
-            replicationGetSlaveName(c),
-            psync_len, psync_offset);
-    /* Note that we don't need to set the selected DB at server.slaveseldb
-     * to -1 to force the master to emit SELECT, since the slave already
-     * has this state from the previous connection with the master. */
-
-    refreshGoodSlavesCount();
-
-    /* Fire the replica change modules event. */
-    moduleFireServerEvent(REDISMODULE_EVENT_REPLICA_CHANGE,
-                          REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE,
-                          NULL);
-
-    return C_OK; /* The caller can return, no full resync needed. */
-
-need_full_resync:
-    /* We need a full resync for some reason... Note that we can't
-     * reply to PSYNC right now if a full SYNC is needed. The reply
-     * must include the master offset at the time the RDB file we transfer
-     * is generated, so we need to delay the reply to that moment. */
-    return C_ERR;
-}
-
-/* Start a BGSAVE for replication goals, which is, selecting the disk or
- * socket target depending on the configuration, and making sure that
- * the script cache is flushed before to start.
- *
- * The mincapa argument is the bitwise AND among all the slaves capabilities
- * of the slaves waiting for this BGSAVE, so represents the slave capabilities
- * all the slaves support. Can be tested via SLAVE_CAPA_* macros.
- *
- * Side effects, other than starting a BGSAVE:
- *
- * 1) Handle the slaves in WAIT_START state, by preparing them for a full
- *    sync if the BGSAVE was successfully started, or sending them an error
- *    and dropping them from the list of slaves.
- *
- * 2) Flush the Lua scripting script cache if the BGSAVE was actually
- *    started.
- *
- * Returns C_OK on success or C_ERR otherwise. */
-int startBgsaveForReplication(int mincapa, int req) {
-    int retval;
-    int socket_target = 0;
-    listIter li;
-    listNode *ln;
-
-    /* We use a socket target if slave can handle the EOF marker and we're configured to do diskless syncs.
-     * Note that in case we're creating a "filtered" RDB (functions-only, for example) we also force socket replication
-     * to avoid overwriting the snapshot RDB file with filtered data. */
-    socket_target = (server.repl_diskless_sync || req & SLAVE_REQ_RDB_MASK) && (mincapa & SLAVE_CAPA_EOF);
-    /* `SYNC` should have failed with error if we don't support socket and require a filter, assert this here */
-    serverAssert(socket_target || !(req & SLAVE_REQ_RDB_MASK));
-
-    int slots_req = req & SLAVE_REQ_SLOTS_SNAPSHOT;
-    serverLog(LL_NOTICE,"Starting BGSAVE for SYNC with target: %s%s",
-        socket_target ? (slots_req ? "slot migration destination socket" : "replicas sockets") : "disk",
-        (req & SLAVE_REQ_RDB_CHANNEL) ? " (rdb-channel)" : "");
-
-    rdbSaveInfo rsi, *rsiptr;
-    rsiptr = rdbPopulateSaveInfo(&rsi);
-    /* Only do rdbSave* when rsiptr is not NULL,
-     * otherwise slave will miss repl-stream-db. */
-    if (rsiptr) {
-        if (socket_target)
-            retval = rdbSaveToSlavesSockets(req,rsiptr);
-        else {
-            /* Keep the page cache since it'll get used soon */
-            retval = rdbSaveBackground(req, server.rdb_filename, rsiptr, RDBFLAGS_REPLICATION | RDBFLAGS_KEEP_CACHE);
-        }
-        if (server.repl_debug_pause & REPL_DEBUG_AFTER_FORK)
-            debugPauseProcess();
-    } else {
-        serverLog(LL_WARNING,"BGSAVE for replication: replication information not available, can't generate the RDB file right now. Try later.");
-        retval = C_ERR;
-    }
-
-    /* If we succeeded to start a BGSAVE with disk target, let's remember
-     * this fact, so that we can later delete the file if needed. Note
-     * that we don't set the flag to 1 if the feature is disabled, otherwise
-     * it would never be cleared: the file is not deleted. This way if
-     * the user enables it later with CONFIG SET, we are fine. */
-    if (retval == C_OK && !socket_target && server.rdb_del_sync_files)
-        RDBGeneratedByReplication = 1;
-
-    /* If we failed to BGSAVE, remove the slaves waiting for a full
-     * resynchronization from the list of slaves, inform them with
-     * an error about what happened, close the connection ASAP. */
-    if (retval == C_ERR) {
-        serverLog(LL_WARNING,"BGSAVE for replication failed");
-        listRewind(server.slaves,&li);
-        while((ln = listNext(&li))) {
-            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,
-                    "BGSAVE failed, replication can't continue");
-                slave->flags |= CLIENT_CLOSE_AFTER_REPLY;
-            }
-        }
-        return retval;
-    }
-
-    /* If the target is socket, rdbSaveToSlavesSockets() already setup
-     * the slaves for a full resync. Otherwise for disk target do it now.*/
-    if (!socket_target) {
-        listRewind(server.slaves,&li);
-        while((ln = listNext(&li))) {
-            client *slave = ln->value;
-
-            if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
-                /* Check slave has the exact requirements */
-                if (slave->slave_req != req)
-                    continue;
-                replicationSetupSlaveForFullResync(slave, getPsyncInitialOffset());
-            }
-        }
-    }
-
-    return retval;
-}
-
-/* SYNC and PSYNC command implementation. */
-void syncCommand(client *c) {
-    /* ignore SYNC if already slave or in monitor mode */
-    if (c->flags & CLIENT_SLAVE) return;
-
-    /* Check if this is a failover request to a replica with the same replid and
-     * become a master if so. */
-    if (c->argc > 3 && !strcasecmp(c->argv[0]->ptr,"psync") && 
-        !strcasecmp(c->argv[3]->ptr,"failover"))
-    {
-        serverLog(LL_NOTICE, "Failover request received for replid %s.",
-            (unsigned char *)c->argv[1]->ptr);
-        if (!server.masterhost) {
-            addReplyError(c, "PSYNC FAILOVER can't be sent to a master.");
-            return;
-        }
-
-        if (!strcasecmp(c->argv[1]->ptr,server.replid)) {
-            if (server.cluster_enabled) {
-                clusterPromoteSelfToMaster();
-            } else {
-                replicationUnsetMaster();
-            }
-            sds client = catClientInfoString(sdsempty(),c);
-            serverLog(LL_NOTICE,
-                "MASTER MODE enabled (failover request from '%s')",client);
-            sdsfree(client);
-        } else {
-            addReplyError(c, "PSYNC FAILOVER replid must match my replid.");
-            return;            
-        }
-    }
-
-    /* Don't let replicas sync with us while we're failing over */
-    if (server.failover_state != NO_FAILOVER) {
-        addReplyError(c,"-NOMASTERLINK Can't SYNC while failing over");
-        return;
-    }
-
-    /* Refuse SYNC requests if we are a slave but the link with our master
-     * is not ok... */
-    if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED) {
-        addReplyError(c,"-NOMASTERLINK Can't SYNC while not connected with my master");
-        return;
-    }
-
-    /* SYNC can't be issued when the server has pending data to send to
-     * the client about already issued commands. We need a fresh reply
-     * buffer registering the differences between the BGSAVE and the current
-     * dataset, so that we can copy to other slaves if needed. */
-    if (clientHasPendingReplies(c)) {
-        addReplyError(c,"SYNC and PSYNC are invalid with pending output");
-        return;
-    }
-
-    /* Fail sync if slave doesn't support EOF capability but wants a filtered RDB. This is because we force filtered
-     * RDB's to be generated over a socket and not through a file to avoid conflicts with the snapshot files. Forcing
-     * use of a socket is handled, if needed, in `startBgsaveForReplication`. */
-    if (c->slave_req & SLAVE_REQ_RDB_MASK && !(c->slave_capa & SLAVE_CAPA_EOF)) {
-        addReplyError(c,"Filtered replica requires EOF capability");
-        return;
-    }
-
-    serverLog(LL_NOTICE,"Replica %s asks for synchronization",
-        replicationGetSlaveName(c));
-
-    /* Try a partial resynchronization if this is a PSYNC command.
-     * If it fails, we continue with usual full resynchronization, however
-     * when this happens replicationSetupSlaveForFullResync will replied
-     * with:
-     *
-     * +FULLRESYNC <replid> <offset>
-     *
-     * So the slave knows the new replid and offset to try a PSYNC later
-     * if the connection with the master is lost. */
-    if (!strcasecmp(c->argv[0]->ptr,"psync")) {
-        long long psync_offset;
-        if (getLongLongFromObjectOrReply(c, c->argv[2], &psync_offset, NULL) != C_OK) {
-            serverLog(LL_WARNING, "Replica %s asks for synchronization but with a wrong offset",
-                      replicationGetSlaveName(c));
-            return;
-        }
-
-        if (masterTryPartialResynchronization(c, psync_offset) == C_OK) {
-            server.stat_sync_partial_ok++;
-            return; /* No full resync needed, return. */
-        } else {
-            char *master_replid = c->argv[1]->ptr;
-
-            /* Increment stats for failed PSYNCs, but only if the
-             * replid is not "?", as this is used by slaves to force a full
-             * resync on purpose when they are not able to partially
-             * resync. */
-            if (master_replid[0] != '?') server.stat_sync_partial_err++;
-            if (c->slave_capa & SLAVE_CAPA_RDB_CHANNEL_REPL) {
-                int len;
-                char buf[128];
-                /* Replica is capable of rdbchannel replication. This is
-                 * replica's main channel. Let replica know full sync is needed.
-                 * Replica will open another connection (rdbchannel). Once rdb
-                 * delivery starts, we'll stream repl data to the main channel.*/
-                c->flags |= CLIENT_SLAVE;
-                c->replstate = SLAVE_STATE_WAIT_RDB_CHANNEL;
-                c->repl_ack_time = server.unixtime;
-                listAddNodeTail(server.slaves, c);
-                createReplicationBacklogIfNeeded();
-
-                serverLog(LL_NOTICE,
-                          "Replica %s is capable of rdb channel synchronization, and partial sync isn't possible. "
-                          "Full sync will continue with dedicated rdb channel.",
-                          replicationGetSlaveName(c));
-
-                /* Send +RDBCHANNELSYNC with client id so we can associate replica connections on master.*/
-                len = snprintf(buf, sizeof(buf), "+RDBCHANNELSYNC %llu\r\n",
-                               (unsigned long long) c->id);
-                if (connWrite(c->conn, buf, strlen(buf)) != len)
-                    freeClientAsync(c);
-
-                return;
-            }
-        }
-    } else {
-        /* If a slave uses SYNC, we are dealing with an old implementation
-         * of the replication protocol (like redis-cli --slave). Flag the client
-         * so that we don't expect to receive REPLCONF ACK feedbacks. */
-        c->flags |= CLIENT_PRE_PSYNC;
-    }
-
-    /* Full resynchronization. */
-    server.stat_sync_full++;
-
-    /* Setup the slave as one waiting for BGSAVE to start. The following code
-     * paths will change the state if we handle the slave differently. */
-    c->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
-    if (server.repl_disable_tcp_nodelay)
-        connDisableTcpNoDelay(c->conn); /* Non critical if it fails. */
-    c->repldbfd = -1;
-    c->flags |= CLIENT_SLAVE;
-    listAddNodeTail(server.slaves,c);
-
-    /* Create the replication backlog if needed. */
-    createReplicationBacklogIfNeeded();
-
-    /* Keep the client in the main thread to avoid data races between the
-     * connWrite call in startBgsaveForReplication and the client's event
-     * handler in IO threads. */
-    if (c->tid != IOTHREAD_MAIN_THREAD_ID) keepClientInMainThread(c);
-
-    /* CASE 1: BGSAVE is in progress, with disk target. */
-    if (server.child_type == CHILD_TYPE_RDB &&
-        server.rdb_child_type == RDB_CHILD_TYPE_DISK)
-    {
-        /* Ok a background save is in progress. Let's check if it is a good
-         * one for replication, i.e. if there is another slave that is
-         * registering differences since the server forked to save. */
-        client *slave;
-        listNode *ln;
-        listIter li;
-
-        listRewind(server.slaves,&li);
-        while((ln = listNext(&li))) {
-            slave = ln->value;
-            /* If the client needs a buffer of commands, we can't use
-             * a replica without replication buffer. */
-            if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END &&
-                (!(slave->flags & CLIENT_REPL_RDBONLY) ||
-                 (c->flags & CLIENT_REPL_RDBONLY)))
-                break;
-        }
-        /* To attach this slave, we check that it has at least all the
-         * capabilities of the slave that triggered the current BGSAVE
-         * and its exact requirements. */
-        if (ln && ((c->slave_capa & slave->slave_capa) == slave->slave_capa) &&
-            c->slave_req == slave->slave_req) {
-            /* Perfect, the server is already registering differences for
-             * another slave. Set the right state, and copy the buffer.
-             * We don't copy buffer if clients don't want. */
-            if (!(c->flags & CLIENT_REPL_RDBONLY))
-                copyReplicaOutputBuffer(c,slave);
-            replicationSetupSlaveForFullResync(c,slave->psync_initial_offset);
-            serverLog(LL_NOTICE,"Waiting for end of BGSAVE for SYNC");
-        } else {
-            /* No way, we need to wait for the next BGSAVE in order to
-             * register differences. */
-            serverLog(LL_NOTICE,"Can't attach the replica to the current BGSAVE. Waiting for next BGSAVE for SYNC");
-        }
-
-    /* CASE 2: BGSAVE is in progress, with socket target. */
-    } else if (server.child_type == CHILD_TYPE_RDB &&
-               server.rdb_child_type == RDB_CHILD_TYPE_SOCKET)
-    {
-        /* There is an RDB child process but it is writing directly to
-         * children sockets. We need to wait for the next BGSAVE
-         * in order to synchronize. */
-        serverLog(LL_NOTICE,"Current BGSAVE has socket target. Waiting for next BGSAVE for SYNC");
-
-    /* CASE 3: There is no BGSAVE is in progress. */
-    } else {
-        if (server.repl_diskless_sync && (c->slave_capa & SLAVE_CAPA_EOF) &&
-            server.repl_diskless_sync_delay)
-        {
-            /* Diskless replication RDB child is created inside
-             * replicationCron() since we want to delay its start a
-             * few seconds to wait for more slaves to arrive. */
-            serverLog(LL_NOTICE,"Delay next BGSAVE for diskless SYNC");
-        } else {
-            /* We don't have a BGSAVE in progress, let's start one. Diskless
-             * or disk-based mode is determined by replica's capacity. */
-            if (!hasActiveChildProcess()) {
-                startBgsaveForReplication(c->slave_capa, c->slave_req);
-            } else {
-                serverLog(LL_NOTICE,
-                    "No BGSAVE in progress, but another BG operation is active. "
-                    "BGSAVE for replication delayed");
-            }
-        }
-    }
-    return;
-}
-
-/* REPLCONF <option> <value> <option> <value> ...
- * This command is used by a replica in order to configure the replication
- * process before starting it with the SYNC command.
- * This command is also used by a master in order to get the replication
- * offset from a replica.
- *
- * Currently we support these options:
- *
- * - listening-port <port>
- * - ip-address <ip>
- * What is the listening ip and port of the Replica redis instance, so that
- * the master can accurately lists replicas and their listening ports in the
- * INFO output.
- *
- * - capa <eof|psync2|rdb-channel-repl>
- * What is the capabilities of this instance.
- * eof: supports EOF-style RDB transfer for diskless replication.
- * psync2: supports PSYNC v2, so understands +CONTINUE <new repl ID>.
- *
- * - ack <offset> [fack <aofofs>]
- * Replica informs the master the amount of replication stream that it
- * processed so far, and optionally the replication offset fsynced to the AOF file.
- * This special pattern doesn't reply to the caller.
- *
- * - getack <dummy>
- * Unlike other subcommands, this is used by master to get the replication
- * offset from a replica.
- *
- * - rdb-only <0|1>
- * Only wants RDB snapshot without replication buffer.
- *
- * - rdb-filter-only <include-filters>
- * Define "include" filters for the RDB snapshot. Currently we only support
- * a single include filter: "functions". Passing an empty string "" will
- * result in an empty RDB.
- *
- * - main-ch-client-id <client-id>
- * Replica's main channel informs master that this is the main channel of the
- * rdb channel identified by the client-id. */
-void replconfCommand(client *c) {
-    int j;
-
-    if ((c->argc % 2) == 0) {
-        /* Number of arguments must be odd to make sure that every
-         * option has a corresponding value. */
-        addReplyErrorObject(c,shared.syntaxerr);
-        return;
-    }
-
-    /* Process every option-value pair. */
-    for (j = 1; j < c->argc; j+=2) {
-        if (!strcasecmp(c->argv[j]->ptr,"listening-port")) {
-            long port;
-
-            if ((getLongFromObjectOrReply(c,c->argv[j+1],
-                    &port,NULL) != C_OK))
-                return;
-            c->slave_listening_port = port;
-        } else if (!strcasecmp(c->argv[j]->ptr,"ip-address")) {
-            sds addr = c->argv[j+1]->ptr;
-            if (sdslen(addr) < NET_HOST_STR_LEN) {
-                if (c->slave_addr) sdsfree(c->slave_addr);
-                c->slave_addr = sdsdup(addr);
-            } else {
-                addReplyErrorFormat(c,"REPLCONF ip-address provided by "
-                    "replica instance is too long: %zd bytes", sdslen(addr));
-                return;
-            }
-        } else if (!strcasecmp(c->argv[j]->ptr,"capa")) {
-            /* Ignore capabilities not understood by this master. */
-            if (!strcasecmp(c->argv[j+1]->ptr,"eof"))
-                c->slave_capa |= SLAVE_CAPA_EOF;
-            else if (!strcasecmp(c->argv[j+1]->ptr,"psync2"))
-                c->slave_capa |= SLAVE_CAPA_PSYNC2;
-            else if (!strcasecmp(c->argv[j+1]->ptr,"rdb-channel-repl") && server.repl_rdb_channel &&
-                     server.repl_diskless_sync) {
-                c->slave_capa |= SLAVE_CAPA_RDB_CHANNEL_REPL;
-            }
-        } else if (!strcasecmp(c->argv[j]->ptr,"ack")) {
-            /* REPLCONF ACK is used by slave to inform the master the amount
-             * of replication stream that it processed so far. It is an
-             * internal only command that normal clients should never use. */
-            long long offset;
-
-            if (!(c->flags & CLIENT_SLAVE)) return;
-            if ((getLongLongFromObject(c->argv[j+1], &offset) != C_OK))
-                return;
-            if (offset > c->repl_ack_off)
-                c->repl_ack_off = offset;
-            if (c->argc > j+3 && !strcasecmp(c->argv[j+2]->ptr,"fack")) {
-                if ((getLongLongFromObject(c->argv[j+3], &offset) != C_OK))
-                    return;
-                if (offset > c->repl_aof_off)
-                    c->repl_aof_off = offset;
-            }
-            c->repl_ack_time = server.unixtime;
-            /* If this was a diskless replication, we need to really put
-             * the slave online when the first ACK is received (which
-             * confirms slave is online and ready to get more data). This
-             * allows for simpler and less CPU intensive EOF detection
-             * when streaming RDB files.
-             * There's a chance the ACK got to us before we detected that the
-             * bgsave is done (since that depends on cron ticks), so run a
-             * quick check first (instead of waiting for the next ACK. */
-            if (server.child_type == CHILD_TYPE_RDB && c->replstate == SLAVE_STATE_WAIT_BGSAVE_END)
-                checkChildrenDone();
-            if (c->repl_start_cmd_stream_on_ack && c->replstate == SLAVE_STATE_ONLINE)
-                replicaStartCommandStream(c);
-            /* If state is send_bulk_and_stream, it means this is the main
-             * channel of the slave in rdbchannel replication. Normally, slave
-             * will be put online after rdb fork is completed. There is chance
-             * that 'ack' might be received before we detect bgsave is done. */
-            if (c->replstate == SLAVE_STATE_SEND_BULK_AND_STREAM)
-                replicaPutOnline(c);
-            /* Note: this command does not reply anything! */
-            return;
-        } else if (!strcasecmp(c->argv[j]->ptr,"getack")) {
-            /* REPLCONF GETACK is used in order to request an ACK ASAP
-             * to the slave. */
-            if (server.masterhost && server.master) replicationSendAck();
-            return;
-        } else if (!strcasecmp(c->argv[j]->ptr,"rdb-only")) {
-           /* REPLCONF RDB-ONLY is used to identify the client only wants
-            * RDB snapshot without replication buffer. */
-            long rdb_only = 0;
-            if (getRangeLongFromObjectOrReply(c,c->argv[j+1],
-                    0,1,&rdb_only,NULL) != C_OK)
-                return;
-            if (rdb_only == 1) {
-                c->flags |= CLIENT_REPL_RDBONLY;
-                /* If replicas ask for RDB only, We can apply the background
-                 * RDB transfer optimization based on the configurations. */
-                if (server.repl_rdb_channel && server.repl_diskless_sync)
-                    c->slave_req |= SLAVE_REQ_RDB_CHANNEL;
-            } else {
-                c->flags &= ~CLIENT_REPL_RDBONLY;
-                c->slave_req &= ~SLAVE_REQ_RDB_CHANNEL;
-            }
-        } else if (!strcasecmp(c->argv[j]->ptr,"rdb-filter-only")) {
-            /* REPLCONFG RDB-FILTER-ONLY is used to define "include" filters
-             * for the RDB snapshot. Currently we only support a single
-             * include filter: "functions". In the future we may want to add
-             * other filters like key patterns, key types, non-volatile, module
-             * aux fields, ...
-             * We might want to add the complementing "RDB-FILTER-EXCLUDE" to
-             * filter out certain data. */
-            int filter_count, i;
-            sds *filters;
-            if (!(filters = sdssplitargs(c->argv[j+1]->ptr, &filter_count))) {
-                addReplyError(c, "Missing rdb-filter-only values");
-                return;
-            }
-            /* By default filter out all parts of the rdb */
-            c->slave_req |= SLAVE_REQ_RDB_EXCLUDE_DATA;
-            c->slave_req |= SLAVE_REQ_RDB_EXCLUDE_FUNCTIONS;
-            for (i = 0; i < filter_count; i++) {
-                if (!strcasecmp(filters[i], "functions"))
-                    c->slave_req &= ~SLAVE_REQ_RDB_EXCLUDE_FUNCTIONS;
-                else {
-                    addReplyErrorFormat(c, "Unsupported rdb-filter-only option: %s", (char*)filters[i]);
-                    sdsfreesplitres(filters, filter_count);
-                    return;
-                }
-            }
-            sdsfreesplitres(filters, filter_count);
-        } else if (!strcasecmp(c->argv[j]->ptr, "rdb-channel")) {
-            long rdb_channel = 0;
-            if (getRangeLongFromObjectOrReply(c, c->argv[j + 1], 0, 1, &rdb_channel, NULL) != C_OK)
-                return;
-            if (rdb_channel == 1) {
-                c->flags |= CLIENT_REPL_RDB_CHANNEL;
-            } else {
-                c->flags &= ~CLIENT_REPL_RDB_CHANNEL;
-            }
-        } else if (!strcasecmp(c->argv[j]->ptr, "main-ch-client-id")) {
-            /* REPLCONF main-ch-client-id <client-id> is used to identify
-             * the current replica rdb channel with existing main channel
-             * connection. */
-            long long client_id = 0;
-            client *main_ch;
-            if (getLongLongFromObjectOrReply(c, c->argv[j + 1], &client_id, NULL) != C_OK)
-                return;
-            main_ch = lookupClientByID(client_id);
-            if (!main_ch || main_ch->replstate != SLAVE_STATE_WAIT_RDB_CHANNEL) {
-                addReplyErrorFormat(c, "Unrecognized RDB client id: %lld", client_id);
-                return;
-            }
-            c->main_ch_client_id = (uint64_t)client_id;
-            /* Inherit the rdb-no-compress request from the main channel. */
-            if (main_ch->slave_req & SLAVE_REQ_RDB_NO_COMPRESS)
-                c->slave_req |= SLAVE_REQ_RDB_NO_COMPRESS;
-        } else if (!strcasecmp(c->argv[j]->ptr, "rdb-no-compress")) {
-            long rdb_no_compress = 0;
-            if (getRangeLongFromObjectOrReply(c, c->argv[j + 1], 0, 1, &rdb_no_compress, NULL) != C_OK)
-                return;
-            if (rdb_no_compress == 1) {
-                c->slave_req |= SLAVE_REQ_RDB_NO_COMPRESS;
-            } else {
-                c->slave_req &= ~SLAVE_REQ_RDB_NO_COMPRESS;
-            }
-        } else {
-            addReplyErrorFormat(c,"Unrecognized REPLCONF option: %s",
-                (char*)c->argv[j]->ptr);
-            return;
-        }
-    }
-    addReply(c,shared.ok);
-}
-
-/* This function puts a replica in the online state, and should be called just
- * after a replica received the RDB file for the initial synchronization.
- *
- * It does a few things:
- * 1) Put the slave in ONLINE state.
- * 2) Update the count of "good replicas".
- * 3) Trigger the module event.
- *
- * the return value indicates that the replica should be disconnected.
- * */
-int replicaPutOnline(client *slave) {
-    if (slave->flags & CLIENT_REPL_RDBONLY) {
-        slave->replstate = SLAVE_STATE_RDB_TRANSMITTED;
-        /* The client asked for RDB only so we should close it ASAP */
-        serverLog(LL_NOTICE,
-                  "RDB transfer completed, rdb only replica (%s) should be disconnected asap",
-                  replicationGetSlaveName(slave));
-        return 0;
-    }
-
-    /* Don't put migration destination client online. */
-    if (slave->flags & CLIENT_ASM_MIGRATING) return 0;
-
-    slave->replstate = SLAVE_STATE_ONLINE;
-    slave->repl_ack_time = server.unixtime; /* Prevent false timeout. */
-
-    refreshGoodSlavesCount();
-    /* Fire the replica change modules event. */
-    moduleFireServerEvent(REDISMODULE_EVENT_REPLICA_CHANGE,
-                          REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE,
-                          NULL);
-    serverLog(LL_NOTICE,"Synchronization with replica %s succeeded",
-        replicationGetSlaveName(slave));
-    return 1;
-}
-
-/* This function should be called just after a replica received the RDB file
- * for the initial synchronization, and we are finally ready to send the
- * incremental stream of commands.
- *
- * It does a few things:
- * 1) Close the replica's connection async if it doesn't need replication
- *    commands buffer stream, since it actually isn't a valid replica.
- * 2) Make sure the writable event is re-installed, since when calling the SYNC
- *    command we had no replies and it was disabled, and then we could
- *    accumulate output buffer data without sending it to the replica so it
- *    won't get mixed with the RDB stream. */
-void replicaStartCommandStream(client *slave) {
-    serverAssert(!(slave->flags & CLIENT_REPL_RDBONLY));
-    slave->repl_start_cmd_stream_on_ack = 0;
-
-    putClientInPendingWriteQueue(slave);
-}
-
-/* We call this function periodically to remove an RDB file that was
- * generated because of replication, in an instance that is otherwise
- * without any persistence. We don't want instances without persistence
- * to take RDB files around, this violates certain policies in certain
- * environments. */
-void removeRDBUsedToSyncReplicas(void) {
-    /* If the feature is disabled, return ASAP but also clear the
-     * RDBGeneratedByReplication flag in case it was set. Otherwise if the
-     * feature was enabled, but gets disabled later with CONFIG SET, the
-     * flag may remain set to one: then next time the feature is re-enabled
-     * via CONFIG SET we have it set even if no RDB was generated
-     * because of replication recently. */
-    if (!server.rdb_del_sync_files) {
-        RDBGeneratedByReplication = 0;
-        return;
-    }
-
-    if (allPersistenceDisabled() && RDBGeneratedByReplication) {
-        client *slave;
-        listNode *ln;
-        listIter li;
-
-        int delrdb = 1;
-        listRewind(server.slaves,&li);
-        while((ln = listNext(&li))) {
-            slave = ln->value;
-            if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START ||
-                slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END ||
-                slave->replstate == SLAVE_STATE_SEND_BULK)
-            {
-                delrdb = 0;
-                break; /* No need to check the other replicas. */
-            }
-        }
-        if (delrdb) {
-            struct stat sb;
-            if (lstat(server.rdb_filename,&sb) != -1) {
-                RDBGeneratedByReplication = 0;
-                serverLog(LL_NOTICE,
-                    "Removing the RDB file used to feed replicas "
-                    "in a persistence-less instance");
-                bg_unlink(server.rdb_filename);
-            }
-        }
-    }
-}
-
-/* Close the repldbfd and reclaim the page cache if the client hold
- * the last reference to replication DB */
-void closeRepldbfd(client *myself) {
-    listNode *ln;
-    listIter li;
-    int reclaim = 1;
-    listRewind(server.slaves,&li);
-    while((ln = listNext(&li))) {
-        client *slave = ln->value;
-        if (slave != myself && slave->replstate == SLAVE_STATE_SEND_BULK) {
-            reclaim = 0;
-            break;
-        }
-    }
-
-    if (reclaim) {
-        bioCreateCloseJob(myself->repldbfd, 0, 1);
-    } else {
-        close(myself->repldbfd);
-    }
-    myself->repldbfd = -1;
-}
-
-void sendBulkToSlave(connection *conn) {
-    client *slave = connGetPrivateData(conn);
-    char buf[PROTO_IOBUF_LEN];
-    ssize_t nwritten, buflen;
-
-    /* Before sending the RDB file, we send the preamble as configured by the
-     * replication process. Currently the preamble is just the bulk count of
-     * the file in the form "$<length>\r\n". */
-    if (slave->replpreamble) {
-        nwritten = connWrite(conn,slave->replpreamble,sdslen(slave->replpreamble));
-        if (nwritten == -1) {
-            serverLog(LL_WARNING,
-                "Write error sending RDB preamble to replica: %s",
-                connGetLastError(conn));
-            freeClient(slave);
-            return;
-        }
-        atomicIncr(server.stat_net_repl_output_bytes, nwritten);
-        sdsrange(slave->replpreamble,nwritten,-1);
-        if (sdslen(slave->replpreamble) == 0) {
-            sdsfree(slave->replpreamble);
-            slave->replpreamble = NULL;
-            /* fall through sending data. */
-        } else {
-            return;
-        }
-    }
-
-    /* If the preamble was already transferred, send the RDB bulk data. */
-    if (lseek(slave->repldbfd,slave->repldboff,SEEK_SET) == -1) {
-	serverLog(LL_WARNING,"Failed to lseek the RDB file to offset %lld for replica %s: %s",
-	    (long long)slave->repldboff, replicationGetSlaveName(slave), strerror(errno));
-	freeClient(slave);
-	return;
-    }
-    buflen = read(slave->repldbfd,buf,PROTO_IOBUF_LEN);
-    if (buflen <= 0) {
-        serverLog(LL_WARNING,"Read error sending DB to replica: %s",
-            (buflen == 0) ? "premature EOF" : strerror(errno));
-        freeClient(slave);
-        return;
-    }
-    if ((nwritten = connWrite(conn,buf,buflen)) == -1) {
-        if (connGetState(conn) != CONN_STATE_CONNECTED) {
-            serverLog(LL_WARNING,"Write error sending DB to replica: %s",
-                connGetLastError(conn));
-            freeClient(slave);
-        }
-        return;
-    }
-    slave->repldboff += nwritten;
-    atomicIncr(server.stat_net_repl_output_bytes, nwritten);
-    if (slave->repldboff == slave->repldbsize) {
-        closeRepldbfd(slave);
-        connSetWriteHandler(slave->conn,NULL);
-        if (!replicaPutOnline(slave)) {
-            freeClient(slave);
-            return;
-        }
-        replicaStartCommandStream(slave);
-    }
-}
-
-/* Remove one write handler from the list of connections waiting to be writable
- * during rdb pipe transfer. */
-void rdbPipeWriteHandlerConnRemoved(struct connection *conn) {
-    if (!connHasWriteHandler(conn))
-        return;
-    connSetWriteHandler(conn, NULL);
-    client *slave = connGetPrivateData(conn);
-    slave->repl_last_partial_write = 0;
-    server.rdb_pipe_numconns_writing--;
-    /* if there are no more writes for now for this conn, or write error: */
-    if (server.rdb_pipe_numconns_writing == 0) {
-        if (aeCreateFileEvent(server.el, server.rdb_pipe_read, AE_READABLE, rdbPipeReadHandler,NULL) == AE_ERR) {
-            serverPanic("Unrecoverable error creating server.rdb_pipe_read file event.");
-        }
-    }
-}
-
-/* Called in diskless master during transfer of data from the rdb pipe, when
- * the replica becomes writable again. */
-void rdbPipeWriteHandler(struct connection *conn) {
-    serverAssert(server.rdb_pipe_bufflen>0);
-    client *slave = connGetPrivateData(conn);
-    ssize_t nwritten;
-    if ((nwritten = connWrite(conn, server.rdb_pipe_buff + slave->repldboff,
-                              server.rdb_pipe_bufflen - slave->repldboff)) == -1)
-    {
-        if (connGetState(conn) == CONN_STATE_CONNECTED)
-            return; /* equivalent to EAGAIN */
-        serverLog(LL_WARNING,"Write error sending DB to replica: %s",
-            connGetLastError(conn));
-        freeClient(slave);
-        return;
-    } else {
-        slave->repldboff += nwritten;
-        atomicIncr(server.stat_net_repl_output_bytes, nwritten);
-        if (slave->repldboff < server.rdb_pipe_bufflen) {
-            slave->repl_last_partial_write = server.unixtime;
-            return; /* more data to write.. */
-        }
-    }
-    rdbPipeWriteHandlerConnRemoved(conn);
-}
-
-/* Called in diskless master, when there's data to read from the child's rdb pipe */
-void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask) {
-    UNUSED(mask);
-    UNUSED(clientData);
-    UNUSED(eventLoop);
-    int i;
-    if (!server.rdb_pipe_buff)
-        server.rdb_pipe_buff = zmalloc(PROTO_IOBUF_LEN);
-    serverAssert(server.rdb_pipe_numconns_writing==0);
-
-    while (1) {
-        server.rdb_pipe_bufflen = read(fd, server.rdb_pipe_buff, PROTO_IOBUF_LEN);
-        if (server.rdb_pipe_bufflen < 0) {
-            if (errno == EAGAIN || errno == EWOULDBLOCK)
-                return;
-            serverLog(LL_WARNING,"Diskless rdb transfer, read error sending DB to replicas: %s", strerror(errno));
-            for (i=0; i < server.rdb_pipe_numconns; i++) {
-                connection *conn = server.rdb_pipe_conns[i];
-                if (!conn)
-                    continue;
-                client *slave = connGetPrivateData(conn);
-                freeClient(slave);
-                server.rdb_pipe_conns[i] = NULL;
-            }
-            killRDBChild();
-            return;
-        }
-
-        if (server.rdb_pipe_bufflen == 0) {
-            /* EOF - write end was closed. */
-            int stillUp = 0;
-            aeDeleteFileEvent(server.el, server.rdb_pipe_read, AE_READABLE);
-            for (i=0; i < server.rdb_pipe_numconns; i++)
-            {
-                connection *conn = server.rdb_pipe_conns[i];
-                if (!conn)
-                    continue;
-                stillUp++;
-            }
-            serverLog(LL_NOTICE,"Diskless rdb transfer, done reading from pipe, %d replicas still up.", stillUp);
-            /* Now that the replicas have finished reading, notify the child that it's safe to exit. 
-             * When the server detects the child has exited, it can mark the replica as online, and
-             * start streaming the replication buffers. */
-            close(server.rdb_child_exit_pipe);
-            server.rdb_child_exit_pipe = -1;
-            return;
-        }
-
-        int stillAlive = 0;
-        for (i=0; i < server.rdb_pipe_numconns; i++)
-        {
-            ssize_t nwritten;
-            connection *conn = server.rdb_pipe_conns[i];
-            if (!conn)
-                continue;
-
-            client *slave = connGetPrivateData(conn);
-            if ((nwritten = connWrite(conn, server.rdb_pipe_buff, server.rdb_pipe_bufflen)) == -1) {
-                if (connGetState(conn) != CONN_STATE_CONNECTED) {
-                    serverLog(LL_WARNING,"Diskless rdb transfer, write error sending DB to replica: %s",
-                        connGetLastError(conn));
-                    freeClient(slave);
-                    server.rdb_pipe_conns[i] = NULL;
-                    continue;
-                }
-                /* An error and still in connected state, is equivalent to EAGAIN */
-                slave->repldboff = 0;
-            } else {
-                /* Note: when use diskless replication, 'repldboff' is the offset
-                 * of 'rdb_pipe_buff' sent rather than the offset of entire RDB. */
-                slave->repldboff = nwritten;
-                atomicIncr(server.stat_net_repl_output_bytes, nwritten);
-            }
-            /* If we were unable to write all the data to one of the replicas,
-             * setup write handler (and disable pipe read handler, below) */
-            if (nwritten != server.rdb_pipe_bufflen) {
-                slave->repl_last_partial_write = server.unixtime;
-                server.rdb_pipe_numconns_writing++;
-                connSetWriteHandler(conn, rdbPipeWriteHandler);
-            }
-            stillAlive++;
-        }
-
-        if (stillAlive == 0) {
-            serverLog(LL_WARNING,"Diskless rdb transfer, last replica dropped, killing fork child.");
-            /* Avoid deleting events after killRDBChild as it may trigger new bgsaves for other replicas. */
-            aeDeleteFileEvent(server.el, server.rdb_pipe_read, AE_READABLE); 
-            killRDBChild();
-            break;
-        }
-        /* Remove the pipe read handler if at least one write handler was set. */
-        else if (server.rdb_pipe_numconns_writing) {
-            aeDeleteFileEvent(server.el, server.rdb_pipe_read, AE_READABLE);
-            break;
-        }
-    }
-}
-
-/* This function is called at the end of every background saving.
- *
- * The argument bgsaveerr is C_OK if the background saving succeeded
- * otherwise C_ERR is passed to the function.
- * The 'type' argument is the type of the child that terminated
- * (if it had a disk or socket target). */
-void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
-    listNode *ln;
-    listIter li;
-
-    /* Note: there's a chance we got here from within the REPLCONF ACK command
-     * so we must avoid using freeClient, otherwise we'll crash on our way up. */
-
-    listRewind(server.slaves,&li);
-    while((ln = listNext(&li))) {
-        client *slave = ln->value;
-
-        /* We can get here via freeClient()->killRDBChild()->checkChildrenDone(). skip disconnected slaves. */
-        if (!slave->conn) continue;
-
-        if (slave->replstate == SLAVE_STATE_SEND_BULK_AND_STREAM) {
-            /* This is the main channel of the slave that received the RDB.
-             * Put it online if RDB delivery is successful. */
-            if (bgsaveerr == C_OK) {
-                /* Notify the task that the snapshot bulk delivery is done */
-                if (slave->flags & CLIENT_ASM_MIGRATING)
-                    asmSlotSnapshotSucceed(slave->task);
-                replicaPutOnline(slave);
-            } else {
-                freeClientAsync(slave);
-            }
-        } else if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
-            struct redis_stat buf;
-
-            if (bgsaveerr != C_OK) {
-                /* Notify the task that the snapshot bulk delivery failed */
-                if (slave->flags & CLIENT_ASM_MIGRATING)
-                    asmSlotSnapshotFailed(slave->task);
-                freeClientAsync(slave);
-                serverLog(LL_WARNING,"SYNC failed. BGSAVE child returned an error");
-                continue;
-            }
-
-            /* If this was an RDB on disk save, we have to prepare to send
-             * the RDB from disk to the slave socket. Otherwise if this was
-             * already an RDB -> Slaves socket transfer, used in the case of
-             * diskless replication, our work is trivial, we can just put
-             * the slave online. */
-            if (type == RDB_CHILD_TYPE_SOCKET) {
-                /* Slots snapshot */
-                if (slave->slave_req & SLAVE_REQ_SLOTS_SNAPSHOT) {
-                    serverLog(LL_NOTICE, "Streamed slots snapshot transfer succeeded");
-                    freeClientAsync(slave);
-                    continue;
-                }
-
-                serverLog(LL_NOTICE,
-                    "Streamed RDB transfer with replica %s succeeded (socket). Waiting for REPLCONF ACK from replica to enable streaming",
-                        replicationGetSlaveName(slave));
-                /* Note: we wait for a REPLCONF ACK message from the replica in
-                 * order to really put it online (install the write handler
-                 * so that the accumulated data can be transferred). However
-                 * we change the replication state ASAP, since our slave
-                 * is technically online now.
-                 *
-                 * So things work like that:
-                 *
-                 * 1. We end transferring the RDB file via socket.
-                 * 2. The replica is put ONLINE but the write handler
-                 *    is not installed.
-                 * 3. The replica however goes really online, and pings us
-                 *    back via REPLCONF ACK commands.
-                 * 4. Now we finally install the write handler, and send
-                 *    the buffers accumulated so far to the replica.
-                 *
-                 * But why we do that? Because the replica, when we stream
-                 * the RDB directly via the socket, must detect the RDB
-                 * EOF (end of file), that is a special random string at the
-                 * end of the RDB (for streamed RDBs we don't know the length
-                 * in advance). Detecting such final EOF string is much
-                 * simpler and less CPU intensive if no more data is sent
-                 * after such final EOF. So we don't want to glue the end of
-                 * the RDB transfer with the start of the other replication
-                 * data. */
-                if (!replicaPutOnline(slave)) {
-                    freeClientAsync(slave);
-                    continue;
-                }
-                slave->repl_start_cmd_stream_on_ack = 1;
-            } else {
-                if ((slave->repldbfd = open(server.rdb_filename,O_RDONLY)) == -1 ||
-                    redis_fstat(slave->repldbfd,&buf) == -1) {
-                    freeClientAsync(slave);
-                    serverLog(LL_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno));
-                    continue;
-                }
-                slave->repldboff = 0;
-                slave->repldbsize = buf.st_size;
-                slave->replstate = SLAVE_STATE_SEND_BULK;
-                slave->replpreamble = sdscatprintf(sdsempty(),"$%lld\r\n",
-                    (unsigned long long) slave->repldbsize);
-
-                connSetWriteHandler(slave->conn,NULL);
-                if (connSetWriteHandler(slave->conn,sendBulkToSlave) == C_ERR) {
-                    freeClientAsync(slave);
-                    continue;
-                }
-            }
-        }
-    }
-}
-
-/* Change the current instance replication ID with a new, random one.
- * This will prevent successful PSYNCs between this master and other
- * slaves, so the command should be called when something happens that
- * alters the current story of the dataset. */
-void changeReplicationId(void) {
-    getRandomHexChars(server.replid,CONFIG_RUN_ID_SIZE);
-    server.replid[CONFIG_RUN_ID_SIZE] = '\0';
-}
-
-/* Clear (invalidate) the secondary replication ID. This happens, for
- * example, after a full resynchronization, when we start a new replication
- * history. */
-void clearReplicationId2(void) {
-    memset(server.replid2,'0',sizeof(server.replid));
-    server.replid2[CONFIG_RUN_ID_SIZE] = '\0';
-    server.second_replid_offset = -1;
-}
-
-/* Use the current replication ID / offset as secondary replication
- * ID, and change the current one in order to start a new history.
- * This should be used when an instance is switched from slave to master
- * so that it can serve PSYNC requests performed using the master
- * replication ID. */
-void shiftReplicationId(void) {
-    memcpy(server.replid2,server.replid,sizeof(server.replid));
-    /* We set the second replid offset to the master offset + 1, since
-     * the slave will ask for the first byte it has not yet received, so
-     * we need to add one to the offset: for example if, as a slave, we are
-     * sure we have the same history as the master for 50 bytes, after we
-     * are turned into a master, we can accept a PSYNC request with offset
-     * 51, since the slave asking has the same history up to the 50th
-     * byte, and is asking for the new bytes starting at offset 51. */
-    server.second_replid_offset = server.master_repl_offset+1;
-    changeReplicationId();
-    serverLog(LL_NOTICE,"Setting secondary replication ID to %s, valid up to offset: %lld. New replication ID is %s", server.replid2, server.second_replid_offset, server.replid);
-}
-
-/* ----------------------------------- SLAVE -------------------------------- */
-
-/* Replication: Replica side. */
-void slaveGetPortStr(char *buf, size_t size) {
-    long long port;
-    if (server.slave_announce_port) {
-        port = server.slave_announce_port;
-    } else if (server.tls_replication && server.tls_port) {
-        port = server.tls_port;
-    } else {
-        port = server.port;
-    }
-    ll2string(buf, size, port);
-}
-
-/* Returns 1 if the given replication state is a handshake state,
- * 0 otherwise. */
-int slaveIsInHandshakeState(void) {
-    return server.repl_state >= REPL_STATE_RECEIVE_PING_REPLY &&
-           server.repl_state <= REPL_STATE_RECEIVE_PSYNC_REPLY;
-}
-
-/* Avoid the master to detect the slave is timing out while loading the
- * RDB file in initial synchronization. We send a single newline character
- * that is valid protocol but is guaranteed to either be sent entirely or
- * not, since the byte is indivisible.
- *
- * The function is called in two contexts: while we flush the current
- * data with emptyData(), and while we load the new data received as an
- * RDB file from the master. */
-void replicationSendNewlineToMaster(void) {
-    static time_t newline_sent;
-    if (time(NULL) != newline_sent) {
-        newline_sent = time(NULL);
-        /* Pinging back in this stage is best-effort. */
-        if (server.repl_transfer_s) connWrite(server.repl_transfer_s, "\n", 1);
-    }
-}
-
-/* Callback used by emptyData() while flushing away old data to load
- * the new dataset received by the master or to clear partial db if loading
- * fails. */
-void replicationEmptyDbCallback(dict *d) {
-    UNUSED(d);
-    if (server.repl_state == REPL_STATE_TRANSFER)
-        replicationSendNewlineToMaster();
-
-    processEventsWhileBlocked();
-}
-
-/* Function to flush old db or the partial db on error. */
-static void rdbLoadEmptyDbFunc(void) {
-    serverAssert(server.loading);
-
-    serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Flushing old data");
-    int empty_db_flags = server.repl_slave_lazy_flush ? EMPTYDB_ASYNC :
-                                                        EMPTYDB_NO_FLAGS;
-
-    emptyData(-1, empty_db_flags, replicationEmptyDbCallback);
-}
-
-/* Once we have a link with the master and the synchronization was
- * performed, this function materializes the master client we store
- * at server.master, starting from the specified file descriptor. */
-void replicationCreateMasterClient(connection *conn, int dbid) {
-    server.master = createClient(conn);
-    if (conn)
-        connSetReadHandler(server.master->conn, readQueryFromClient);
-
-    /**
-     * Important note:
-     * The CLIENT_DENY_BLOCKING flag is not, and should not, be set here.
-     * For commands like BLPOP, it makes no sense to block the master
-     * connection, and such blocking attempt will probably cause deadlock and
-     * break the replication. We consider such a thing as a bug because
-     * commands as BLPOP should never be sent on the replication link.
-     * A possible use-case for blocking the replication link is if a module wants
-     * to pass the execution to a background thread and unblock after the
-     * execution is done. This is the reason why we allow blocking the replication
-     * connection. */
-    server.master->flags |= CLIENT_MASTER;
-
-    /* Allocate a private query buffer for the master client instead of using the reusable query buffer.
-     * This is done because the master's query buffer data needs to be preserved for my sub-replicas to use. */
-    server.master->querybuf = sdsempty();
-    server.master->authenticated = 1;
-    server.master->reploff = server.master_initial_offset;
-    server.master->read_reploff = server.master->reploff;
-    server.master->user = NULL; /* This client can do everything. */
-    memcpy(server.master->replid, server.master_replid,
-        sizeof(server.master_replid));
-    /* If master offset is set to -1, this master is old and is not
-     * PSYNC capable, so we flag it accordingly. */
-    if (server.master->reploff == -1)
-        server.master->flags |= CLIENT_PRE_PSYNC;
-    if (dbid != -1) selectDb(server.master,dbid);
-}
-
-static int useDisklessLoad(void) {
-    /* compute boolean decision to use diskless load */
-    int enabled = server.repl_diskless_load == REPL_DISKLESS_LOAD_ALWAYS || server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB ||
-           (server.repl_diskless_load == REPL_DISKLESS_LOAD_WHEN_DB_EMPTY && dbTotalServerKeyCount()==0);
-
-    if (enabled) {
-        /* Check all modules handle read errors, otherwise it's not safe to use diskless load. */
-        if (server.repl_diskless_load != REPL_DISKLESS_LOAD_ALWAYS && !moduleAllDatatypesHandleErrors()) {
-            serverLog(LL_NOTICE,
-                "Skipping diskless-load because there are modules that don't handle read errors.");
-            enabled = 0;
-        }
-        /* Check all modules handle async replication, otherwise it's not safe to use diskless load. */
-        else if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB && !moduleAllModulesHandleReplAsyncLoad()) {
-            serverLog(LL_NOTICE,
-                "Skipping diskless-load because there are modules that are not aware of async replication.");
-            enabled = 0;
-        }
-    }
-    return enabled;
-}
-
-/* Helper function for readSyncBulkPayload() to initialize tempDb
- * before socket-loading the new db from master. The tempDb may be populated
- * by swapMainDbWithTempDb or freed by disklessLoadDiscardTempDb later. */
-redisDb *disklessLoadInitTempDb(void) {
-    return initTempDb();
-}
-
-/* Helper function for readSyncBulkPayload() to discard our tempDb
- * when the loading succeeded or failed. */
-void disklessLoadDiscardTempDb(redisDb *tempDb) {
-    discardTempDb(tempDb);
-}
-
-/* If we know we got an entirely different data set from our master
- * we have no way to incrementally feed our replicas after that.
- * We want our replicas to resync with us as well, if we have any sub-replicas.
- * This is useful on readSyncBulkPayload in places where we just finished transferring db. */
-void replicationAttachToNewMaster(void) { 
-    /* Replica starts to apply data from new master, we must discard the cached
-     * master structure. */
-    serverAssert(server.master == NULL);
-    replicationDiscardCachedMaster();
-
-    disconnectSlaves(); /* Force our replicas to resync with us as well. */
-    freeReplicationBacklog(); /* Don't allow our chained replicas to PSYNC. */
-}
-
-/* Asynchronously read the SYNC payload we receive from a master */
-#define REPL_MAX_WRITTEN_BEFORE_FSYNC (1024*1024*8) /* 8 MB */
-void readSyncBulkPayload(connection *conn) {
-    char buf[PROTO_IOBUF_LEN];
-    ssize_t nread, readlen, nwritten;
-    int use_diskless_load = useDisklessLoad();
-    int rdbchannel = (conn == server.repl_rdb_transfer_s);
-    int empty_db_flags = server.repl_slave_lazy_flush ? EMPTYDB_ASYNC :
-                                                        EMPTYDB_NO_FLAGS;
-    off_t left;
-
-    /* Static vars used to hold the EOF mark, and the last bytes received
-     * from the server: when they match, we reached the end of the transfer. */
-    static char eofmark[CONFIG_RUN_ID_SIZE];
-    static char lastbytes[CONFIG_RUN_ID_SIZE];
-    static int usemark = 0;
-
-    /* If repl_transfer_size == -1 we still have to read the bulk length
-     * from the master reply. */
-    if (server.repl_transfer_size == -1) {
-        nread = connSyncReadLine(conn,buf,1024,server.repl_syncio_timeout*1000);
-        if (nread == -1) {
-            serverLog(LL_WARNING,
-                "I/O error reading bulk count from MASTER: %s",
-                connGetLastError(conn));
-            goto error;
-        } else {
-            /* nread here is returned by connSyncReadLine(), which calls syncReadLine() and
-             * convert "\r\n" to '\0' so 1 byte is lost. */
-            atomicIncr(server.stat_net_repl_input_bytes, nread+1);
-        }
-
-        if (buf[0] == '-') {
-            serverLog(LL_WARNING,
-                "MASTER aborted replication with an error: %s",
-                buf+1);
-            goto error;
-        } else if (buf[0] == '\0') {
-            /* At this stage just a newline works as a PING in order to take
-             * the connection live. So we refresh our last interaction
-             * timestamp. */
-            server.repl_transfer_lastio = server.unixtime;
-            return;
-        } else if (buf[0] != '$') {
-            serverLog(LL_WARNING,"Bad protocol from MASTER, the first byte is not '$' (we received '%s'), are you sure the host and port are right?", buf);
-            goto error;
-        }
-
-        /* There are two possible forms for the bulk payload. One is the
-         * usual $<count> bulk format. The other is used for diskless transfers
-         * when the master does not know beforehand the size of the file to
-         * transfer. In the latter case, the following format is used:
-         *
-         * $EOF:<40 bytes delimiter>
-         *
-         * At the end of the file the announced delimiter is transmitted. The
-         * delimiter is long and random enough that the probability of a
-         * collision with the actual file content can be ignored. */
-        if (strncmp(buf+1,"EOF:",4) == 0 && strlen(buf+5) >= CONFIG_RUN_ID_SIZE) {
-            usemark = 1;
-            memcpy(eofmark,buf+5,CONFIG_RUN_ID_SIZE);
-            memset(lastbytes,0,CONFIG_RUN_ID_SIZE);
-            /* Set any repl_transfer_size to avoid entering this code path
-             * at the next call. */
-            server.repl_transfer_size = 0;
-            serverLog(LL_NOTICE,
-                "MASTER <-> REPLICA sync: receiving streamed RDB from master with EOF %s",
-                use_diskless_load? "to parser":"to disk");
-        } else {
-            usemark = 0;
-            server.repl_transfer_size = strtol(buf+1,NULL,10);
-            serverLog(LL_NOTICE,
-                "MASTER <-> REPLICA sync: receiving %lld bytes from master %s",
-                (long long) server.repl_transfer_size,
-                use_diskless_load? "to parser":"to disk");
-        }
-        return;
-    }
-
-    if (!use_diskless_load) {
-        /* Read the data from the socket, store it to a file and search
-         * for the EOF. */
-        if (usemark) {
-            readlen = sizeof(buf);
-        } else {
-            left = server.repl_transfer_size - server.repl_transfer_read;
-            readlen = (left < (signed)sizeof(buf)) ? left : (signed)sizeof(buf);
-        }
-
-        nread = connRead(conn,buf,readlen);
-        if (nread <= 0) {
-            if (connGetState(conn) == CONN_STATE_CONNECTED) {
-                /* equivalent to EAGAIN */
-                return;
-            }
-            serverLog(LL_WARNING,"I/O error trying to sync with MASTER: %s",
-                (nread == -1) ? connGetLastError(conn) : "connection lost");
-            cancelReplicationHandshake(1);
-            return;
-        }
-        atomicIncr(server.stat_net_repl_input_bytes, nread);
-
-        /* When a mark is used, we want to detect EOF asap in order to avoid
-         * writing the EOF mark into the file... */
-        int eof_reached = 0;
-
-        if (usemark) {
-            /* Update the last bytes array, and check if it matches our
-             * delimiter. */
-            if (nread >= CONFIG_RUN_ID_SIZE) {
-                memcpy(lastbytes,buf+nread-CONFIG_RUN_ID_SIZE,
-                       CONFIG_RUN_ID_SIZE);
-            } else {
-                int rem = CONFIG_RUN_ID_SIZE-nread;
-                memmove(lastbytes,lastbytes+nread,rem);
-                memcpy(lastbytes+rem,buf,nread);
-            }
-            if (memcmp(lastbytes,eofmark,CONFIG_RUN_ID_SIZE) == 0)
-                eof_reached = 1;
-        }
-
-        /* Update the last I/O time for the replication transfer (used in
-         * order to detect timeouts during replication), and write what we
-         * got from the socket to the dump file on disk. */
-        server.repl_transfer_lastio = server.unixtime;
-        if ((nwritten = write(server.repl_transfer_fd,buf,nread)) != nread) {
-            serverLog(LL_WARNING,
-                "Write error or short write writing to the DB dump file "
-                "needed for MASTER <-> REPLICA synchronization: %s",
-                (nwritten == -1) ? strerror(errno) : "short write");
-            goto error;
-        }
-        server.repl_transfer_read += nread;
-
-        /* Delete the last 40 bytes from the file if we reached EOF. */
-        if (usemark && eof_reached) {
-            if (ftruncate(server.repl_transfer_fd,
-                server.repl_transfer_read - CONFIG_RUN_ID_SIZE) == -1)
-            {
-                serverLog(LL_WARNING,
-                    "Error truncating the RDB file received from the master "
-                    "for SYNC: %s", strerror(errno));
-                goto error;
-            }
-        }
-
-        /* Sync data on disk from time to time, otherwise at the end of the
-         * transfer we may suffer a big delay as the memory buffers are copied
-         * into the actual disk. */
-        if (server.repl_transfer_read >=
-            server.repl_transfer_last_fsync_off + REPL_MAX_WRITTEN_BEFORE_FSYNC)
-        {
-            off_t sync_size = server.repl_transfer_read -
-                              server.repl_transfer_last_fsync_off;
-            rdb_fsync_range(server.repl_transfer_fd,
-                server.repl_transfer_last_fsync_off, sync_size);
-            server.repl_transfer_last_fsync_off += sync_size;
-        }
-
-        /* Check if the transfer is now complete */
-        if (!usemark) {
-            if (server.repl_transfer_read == server.repl_transfer_size)
-                eof_reached = 1;
-        }
-
-        /* If the transfer is yet not complete, we need to read more, so
-         * return ASAP and wait for the handler to be called again. */
-        if (!eof_reached) return;
-    }
-
-    /* We reach this point in one of the following cases:
-     *
-     * 1. The replica is using diskless replication, that is, it reads data
-     *    directly from the socket to the Redis memory, without using
-     *    a temporary RDB file on disk. In that case we just block and
-     *    read everything from the socket.
-     *
-     * 2. Or when we are done reading from the socket to the RDB file, in
-     *    such case we want just to read the RDB file in memory. */
-
-    /* We need to stop any AOF rewriting child before flushing and parsing
-     * the RDB, otherwise we'll create a copy-on-write disaster. */
-    if (server.aof_state != AOF_OFF) stopAppendOnly();
-    /* Also try to stop save RDB child before flushing and parsing the RDB:
-     * 1. Ensure background save doesn't overwrite synced data after being loaded.
-     * 2. Avoid copy-on-write disaster. */
-    if (server.child_type == CHILD_TYPE_RDB) {
-        if (!use_diskless_load) {
-            serverLog(LL_NOTICE,
-                "Replica is about to load the RDB file received from the "
-                "master, but there is a pending RDB child running. "
-                "Killing process %ld and removing its temp file to avoid "
-                "any race",
-                (long) server.child_pid);
-        }
-        killRDBChild();
-    }
-
-    /* Attach to the new master immediately if we are not using swapdb. */
-    if (!use_diskless_load || server.repl_diskless_load != REPL_DISKLESS_LOAD_SWAPDB)
-        replicationAttachToNewMaster();
-
-    /* Before loading the DB into memory we need to delete the readable
-     * handler, otherwise it will get called recursively since
-     * rdbLoad() will call the event loop to process events from time to
-     * time for non blocking loading. */
-    connSetReadHandler(conn, NULL);
-    
-    serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Loading DB in memory");
-    rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
-    if (use_diskless_load) {
-        rio rdb;
-        redisDb *dbarray;
-        functionsLibCtx* functions_lib_ctx;
-        int asyncLoading = 0;
-
-        if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
-            moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD,
-                                  REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_STARTED,
-                                  NULL);
-            /* Async loading means we continue serving read commands during full resync, and
-             * "swap" the new db with the old db only when loading is done.
-             * It is enabled only on SWAPDB diskless replication when master replication ID hasn't changed,
-             * because in that state the old content of the db represents a different point in time of the same
-             * data set we're currently receiving from the master. */
-            if (memcmp(server.replid, server.master_replid, CONFIG_RUN_ID_SIZE) == 0) {
-                asyncLoading = 1;
-            }
-        }
-
-        /* Set disklessLoadingRio before calling emptyData() which may yield
-         * back to networking. */
-        rioInitWithConn(&rdb,conn,server.repl_transfer_size);
-        disklessLoadingRio = &rdb;
-
-        /* Empty db */
-        loadingSetFlags(NULL, server.repl_transfer_size, asyncLoading);
-        if (server.repl_diskless_load != REPL_DISKLESS_LOAD_SWAPDB) {
-            serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Flushing old data");
-            /* Note that inside loadingSetFlags(), server.loading is set.
-             * replicationEmptyDbCallback() may yield back to event-loop to
-             * reply -LOADING. */
-            emptyData(-1, empty_db_flags, replicationEmptyDbCallback);
-        }
-        loadingFireEvent(RDBFLAGS_REPLICATION);
-
-        if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
-            dbarray = disklessLoadInitTempDb();
-            functions_lib_ctx = functionsLibCtxCreate();
-        } else {
-            dbarray = server.db;
-            functions_lib_ctx = functionsLibCtxGetCurrent();
-            functionsLibCtxClear(functions_lib_ctx);
-        }
-
-        /* Put the socket in blocking mode to simplify RDB transfer.
-         * We'll restore it when the RDB is received. */
-        connBlock(conn);
-        connRecvTimeout(conn, server.repl_timeout*1000);
-
-        int loadingFailed = 0;
-        rdbLoadingCtx loadingCtx = { .dbarray = dbarray, .functions_lib_ctx = functions_lib_ctx };
-        if (rdbLoadRioWithLoadingCtx(&rdb,RDBFLAGS_REPLICATION,&rsi,&loadingCtx) != C_OK) {
-            /* RDB loading failed. */
-            serverLog(LL_WARNING,
-                      "Failed trying to load the MASTER synchronization DB "
-                      "from socket, check server logs.");
-            loadingFailed = 1;
-        } else if (usemark) {
-            /* Verify the end mark is correct. */
-            if (!rioRead(&rdb, buf, CONFIG_RUN_ID_SIZE) ||
-                memcmp(buf, eofmark, CONFIG_RUN_ID_SIZE) != 0)
-            {
-                serverLog(LL_WARNING, "Replication stream EOF marker is broken");
-                loadingFailed = 1;
-            }
-        }
-        disklessLoadingRio = NULL;
-
-        if (loadingFailed) {
-            rioFreeConn(&rdb, NULL);
-
-            if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
-                /* Discard potentially partially loaded tempDb. */
-                moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD,
-                                      REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_ABORTED,
-                                      NULL);
-
-                disklessLoadDiscardTempDb(dbarray);
-                functionsLibCtxFree(functions_lib_ctx);
-                serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Discarding temporary DB in background");
-            } else {
-                /* Remove the half-loaded data in case we started with an empty replica. */
-                emptyData(-1,empty_db_flags,replicationEmptyDbCallback);
-            }
-
-            /* Note that replicationEmptyDbCallback() may yield back to event
-             * loop to reply -LOADING if flushing the db takes a long time. So,
-             * stopLoading() must be called after emptyData() above. */
-            stopLoading(0);
-
-            /* This must be called after stopLoading(0) as it checks loading
-             * flag in case of rdbchannel replication. */
-            cancelReplicationHandshake(1);
-
-            /* 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;
-        }
-
-        /* RDB loading succeeded if we reach this point. */
-        if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
-            /* Cancel all ASM trim jobs as we are about to swap the main db. */
-            asmCancelTrimJobs();
-            /* We will soon swap main db with tempDb and replicas will start
-             * to apply data from new master, we must discard the cached
-             * master structure and force resync of sub-replicas. */
-            replicationAttachToNewMaster();
-
-            serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Swapping active DB with loaded DB");
-            swapMainDbWithTempDb(dbarray);
-
-            /* swap existing functions ctx with the temporary one */
-            functionsLibCtxSwapWithCurrent(functions_lib_ctx);
-
-            moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD,
-                        REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_COMPLETED,
-                        NULL);
-
-            /* Delete the old db as it's useless now. */
-            disklessLoadDiscardTempDb(dbarray);
-            serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Discarding old DB in background");
-        }
-
-        /* Inform about db change, as replication was diskless and didn't cause a save. */
-        server.dirty++;
-
-        stopLoading(1);
-
-        /* Cleanup and restore the socket to the original state to continue
-         * with the normal replication. */
-        rioFreeConn(&rdb, NULL);
-        connNonBlock(conn);
-        connRecvTimeout(conn,0);
-    } else {
-
-        /* Make sure the new file (also used for persistence) is fully synced
-         * (not covered by earlier calls to rdb_fsync_range). */
-        if (fsync(server.repl_transfer_fd) == -1) {
-            serverLog(LL_WARNING,
-                "Failed trying to sync the temp DB to disk in "
-                "MASTER <-> REPLICA synchronization: %s",
-                strerror(errno));
-            cancelReplicationHandshake(1);
-            return;
-        }
-
-        /* Rename rdb like renaming rewrite aof asynchronously. */
-        int old_rdb_fd = open(server.rdb_filename,O_RDONLY|O_NONBLOCK);
-        if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
-            serverLog(LL_WARNING,
-                "Failed trying to rename the temp DB into %s in "
-                "MASTER <-> REPLICA synchronization: %s",
-                server.rdb_filename, strerror(errno));
-            cancelReplicationHandshake(1);
-            if (old_rdb_fd != -1) close(old_rdb_fd);
-            return;
-        }
-        /* Close old rdb asynchronously. */
-        if (old_rdb_fd != -1) bioCreateCloseJob(old_rdb_fd, 0, 0);
-
-        /* Sync the directory to ensure rename is persisted */
-        if (fsyncFileDir(server.rdb_filename) == -1) {
-            serverLog(LL_WARNING,
-                "Failed trying to sync DB directory %s in "
-                "MASTER <-> REPLICA synchronization: %s",
-                server.rdb_filename, strerror(errno));
-            cancelReplicationHandshake(1);
-            return;
-        }
-
-        if (rdbLoadWithEmptyFunc(server.rdb_filename,&rsi,RDBFLAGS_REPLICATION,rdbLoadEmptyDbFunc) != RDB_OK) {
-            serverLog(LL_WARNING,
-                "Failed trying to load the MASTER synchronization "
-                "DB from disk, check server logs.");
-            cancelReplicationHandshake(1);
-            if (server.rdb_del_sync_files && allPersistenceDisabled()) {
-                serverLog(LL_NOTICE,"Removing the RDB file obtained from "
-                                    "the master. This replica has persistence "
-                                    "disabled");
-                bg_unlink(server.rdb_filename);
-            }
-
-            /* Note that there's no point in restarting the AOF on sync failure,
-               it'll be restarted when sync succeeds or replica promoted. */
-            return;
-        }
-
-        /* Cleanup. */
-        if (server.rdb_del_sync_files && allPersistenceDisabled()) {
-            serverLog(LL_NOTICE,"Removing the RDB file obtained from "
-                                "the master. This replica has persistence "
-                                "disabled");
-            bg_unlink(server.rdb_filename);
-        }
-
-        zfree(server.repl_transfer_tmpfile);
-        close(server.repl_transfer_fd);
-        server.repl_transfer_fd = -1;
-        server.repl_transfer_tmpfile = NULL;
-    }
-
-    /* Final setup of the connected slave <- master link */
-    replicationCreateMasterClient(server.repl_transfer_s,rsi.repl_stream_db);
-    server.repl_state = REPL_STATE_CONNECTED;
-    server.repl_down_since = 0;
-    server.repl_up_since = server.unixtime;
-
-    if (server.repl_disconnect_start_time != 0) {
-        server.repl_total_disconnect_time += server.unixtime - server.repl_disconnect_start_time;
-        server.repl_disconnect_start_time = 0;
-    }
-    /* Fire the master link modules event. */
-    moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
-                          REDISMODULE_SUBEVENT_MASTER_LINK_UP,
-                          NULL);
-
-    /* After a full resynchronization we use the replication ID and
-     * offset of the master. The secondary ID / offset are cleared since
-     * we are starting a new history. */
-    memcpy(server.replid,server.master->replid,sizeof(server.replid));
-    server.master_repl_offset = server.master->reploff;
-    clearReplicationId2();
-
-    /* Let's create the replication backlog if needed. Slaves need to
-     * accumulate the backlog regardless of the fact they have sub-slaves
-     * or not, in order to behave correctly if they are promoted to
-     * masters after a failover. */
-    if (server.repl_backlog == NULL) createReplicationBacklog();
-    serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Finished with success");
-
-    if (server.supervised_mode == SUPERVISED_SYSTEMD) {
-        redisCommunicateSystemd("STATUS=MASTER <-> REPLICA sync: Finished with success. Ready to accept connections in read-write mode.\n");
-    }
-
-    /* Send the initial ACK immediately to put this replica in online state. */
-    if (usemark) replicationSendAck();
-
-    /* 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) {
-        serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Starting AOF after a successful sync");
-        startAppendOnlyWithRetry();
-    }
-
-    /* Stream accumulated replication buffer to the db and finalize fullsync */
-    if (rdbchannel) {
-        if (server.repl_rdb_transfer_s) {
-            connClose(server.repl_rdb_transfer_s);
-            server.repl_rdb_transfer_s = NULL;
-        }
-        rdbChannelStreamReplDataToDb();
-    }
-
-    return;
-
-error:
-    cancelReplicationHandshake(1);
-    return;
-}
-
-char *receiveSynchronousResponse(connection *conn) {
-    char buf[256];
-    /* Read the reply from the server. */
-    if (connSyncReadLine(conn,buf,sizeof(buf),server.repl_syncio_timeout*1000) == -1)
-    {
-        serverLog(LL_WARNING, "Failed to read response from the server: %s", connGetLastError(conn));
-        return NULL;
-    }
-    server.repl_transfer_lastio = server.unixtime;
-    return sdsnew(buf);
-}
-
-/* Send a pre-formatted multi-bulk command to the connection. */
-char* sendCommandRaw(connection *conn, sds cmd) {
-    if (connSyncWrite(conn,cmd,sdslen(cmd),server.repl_syncio_timeout*1000) == -1) {
-        return sdscatprintf(sdsempty(),"-Writing to master: %s",
-                connGetLastError(conn));
-    }
-    return NULL;
-}
-
-/* Compose a multi-bulk command and send it to the connection.
- * Used to send AUTH and REPLCONF commands to the master before starting the
- * replication.
- *
- * Takes a list of char* arguments, terminated by a NULL argument.
- *
- * The command returns an sds string representing the result of the
- * operation. On error the first byte is a "-".
- */
-char *sendCommand(connection *conn, ...) {
-    va_list ap;
-    sds cmd = sdsempty();
-    sds cmdargs = sdsempty();
-    size_t argslen = 0;
-    char *arg;
-
-    /* Create the command to send to the master, we use redis binary
-     * protocol to make sure correct arguments are sent. This function
-     * is not safe for all binary data. */
-    va_start(ap,conn);
-    while(1) {
-        arg = va_arg(ap, char*);
-        if (arg == NULL) break;
-        cmdargs = sdscatprintf(cmdargs,"$%zu\r\n%s\r\n",strlen(arg),arg);
-        argslen++;
-    }
-
-    cmd = sdscatprintf(cmd,"*%zu\r\n",argslen);
-    cmd = sdscatsds(cmd,cmdargs);
-    sdsfree(cmdargs);
-
-    va_end(ap);
-    char* err = sendCommandRaw(conn, cmd);
-    sdsfree(cmd);
-    if(err)
-        return err;
-    return NULL;
-}
-
-/* Compose a multi-bulk command and send it to the connection. 
- * Used to send AUTH and REPLCONF commands to the master before starting the
- * replication.
- *
- * argv_lens is optional, when NULL, strlen is used.
- *
- * The command returns an sds string representing the result of the
- * operation. On error the first byte is a "-".
- */
-char *sendCommandArgv(connection *conn, int argc, char **argv, size_t *argv_lens) {
-    sds cmd = sdsempty();
-    char *arg;
-    int i;
-
-    /* Create the command to send to the master. */
-    cmd = sdscatfmt(cmd,"*%i\r\n",argc);
-    for (i=0; i<argc; i++) {
-        int len;
-        arg = argv[i];
-        len = argv_lens ? argv_lens[i] : strlen(arg);
-        cmd = sdscatfmt(cmd,"$%i\r\n",len);
-        cmd = sdscatlen(cmd,arg,len);
-        cmd = sdscatlen(cmd,"\r\n",2);
-    }
-    char* err = sendCommandRaw(conn, cmd);
-    sdsfree(cmd);
-    if (err)
-        return err;
-    return NULL;
-}
-
-/* Try a partial resynchronization with the master if we are about to reconnect.
- * If there is no cached master structure, at least try to issue a
- * "PSYNC ? -1" command in order to trigger a full resync using the PSYNC
- * command in order to obtain the master replid and the master replication
- * global offset.
- *
- * This function is designed to be called from syncWithMaster(), so the
- * following assumptions are made:
- *
- * 1) We pass the function an already connected socket "fd".
- * 2) This function does not close the file descriptor "fd". However in case
- *    of successful partial resynchronization, the function will reuse
- *    'fd' as file descriptor of the server.master client structure.
- *
- * The function is split in two halves: if read_reply is 0, the function
- * writes the PSYNC command on the socket, and a new function call is
- * needed, with read_reply set to 1, in order to read the reply of the
- * command. This is useful in order to support non blocking operations, so
- * that we write, return into the event loop, and read when there are data.
- *
- * When read_reply is 0 the function returns PSYNC_WRITE_ERR if there
- * was a write error, or PSYNC_WAIT_REPLY to signal we need another call
- * with read_reply set to 1. However even when read_reply is set to 1
- * the function may return PSYNC_WAIT_REPLY again to signal there were
- * insufficient data to read to complete its work. We should re-enter
- * into the event loop and wait in such a case.
- *
- * The function returns:
- *
- * PSYNC_CONTINUE: If the PSYNC command succeeded and we can continue.
- * PSYNC_FULLRESYNC: If PSYNC is supported but a full resync is needed.
- *                   In this case the master replid and global replication
- *                   offset is saved.
- * PSYNC_NOT_SUPPORTED: If the server does not understand PSYNC at all and
- *                      the caller should fall back to SYNC.
- * PSYNC_WRITE_ERROR: There was an error writing the command to the socket.
- * PSYNC_WAIT_REPLY: Call again the function with read_reply set to 1.
- * PSYNC_TRY_LATER: Master is currently in a transient error condition.
- *
- * Notable side effects:
- *
- * 1) As a side effect of the function call the function removes the readable
- *    event handler from "fd", unless the return value is PSYNC_WAIT_REPLY.
- * 2) server.master_initial_offset is set to the right value according
- *    to the master reply. This will be used to populate the 'server.master'
- *    structure replication offset.
- */
-
-#define PSYNC_WRITE_ERROR 0
-#define PSYNC_WAIT_REPLY 1
-#define PSYNC_CONTINUE 2
-#define PSYNC_FULLRESYNC 3
-#define PSYNC_NOT_SUPPORTED 4
-#define PSYNC_TRY_LATER 5
-#define PSYNC_FULLRESYNC_RDBCHANNEL 6
-int slaveTryPartialResynchronization(connection *conn, int read_reply) {
-    char *psync_replid;
-    char psync_offset[32];
-    sds reply;
-
-    /* Writing half */
-    if (!read_reply) {
-        /* Initially set master_initial_offset to -1 to mark the current
-         * master replid and offset as not valid. Later if we'll be able to do
-         * a FULL resync using the PSYNC command we'll set the offset at the
-         * right value, so that this information will be propagated to the
-         * client structure representing the master into server.master. */
-        server.master_initial_offset = -1;
-
-        if (server.cached_master) {
-            psync_replid = server.cached_master->replid;
-            snprintf(psync_offset,sizeof(psync_offset),"%lld", server.cached_master->reploff+1);
-            serverLog(LL_NOTICE,"Trying a partial resynchronization (request %s:%s).", psync_replid, psync_offset);
-        } else {
-            serverLog(LL_NOTICE,"Partial resynchronization not possible (no cached master)");
-            psync_replid = "?";
-            memcpy(psync_offset,"-1",3);
-        }
-
-        /* Issue the PSYNC command, if this is a master with a failover in
-         * progress then send the failover argument to the replica to cause it
-         * to become a master */
-        if (server.failover_state == FAILOVER_IN_PROGRESS) {
-            reply = sendCommand(conn,"PSYNC",psync_replid,psync_offset,"FAILOVER",NULL);
-        } else {
-            reply = sendCommand(conn,"PSYNC",psync_replid,psync_offset,NULL);
-        }
-
-        if (reply != NULL) {
-            serverLog(LL_WARNING,"Unable to send PSYNC to master: %s",reply);
-            sdsfree(reply);
-            connSetReadHandler(conn, NULL);
-            return PSYNC_WRITE_ERROR;
-        }
-        return PSYNC_WAIT_REPLY;
-    }
-
-    /* Reading half */
-    reply = receiveSynchronousResponse(conn);
-    /* Master did not reply to PSYNC */
-    if (reply == NULL) {
-        connSetReadHandler(conn, NULL);
-        serverLog(LL_WARNING, "Master did not reply to PSYNC, will try later");
-        return PSYNC_TRY_LATER;
-    }
-
-    if (sdslen(reply) == 0) {
-        /* The master may send empty newlines after it receives PSYNC
-         * and before to reply, just to keep the connection alive. */
-        sdsfree(reply);
-        return PSYNC_WAIT_REPLY;
-    }
-
-    connSetReadHandler(conn, NULL);
-
-    if (!strncmp(reply,"+FULLRESYNC",11)) {
-        char *replid = NULL, *offset = NULL;
-
-        /* FULL RESYNC, parse the reply in order to extract the replid
-         * and the replication offset. */
-        replid = strchr(reply,' ');
-        if (replid) {
-            replid++;
-            offset = strchr(replid,' ');
-            if (offset) offset++;
-        }
-        if (!replid || !offset || (offset-replid-1) != CONFIG_RUN_ID_SIZE) {
-            serverLog(LL_WARNING,
-                "Master replied with wrong +FULLRESYNC syntax.");
-            /* This is an unexpected condition, actually the +FULLRESYNC
-             * reply means that the master supports PSYNC, but the reply
-             * format seems wrong. To stay safe we blank the master
-             * replid to make sure next PSYNCs will fail. */
-            memset(server.master_replid,0,CONFIG_RUN_ID_SIZE+1);
-        } else {
-            memcpy(server.master_replid, replid, offset-replid-1);
-            server.master_replid[CONFIG_RUN_ID_SIZE] = '\0';
-            server.master_initial_offset = strtoll(offset,NULL,10);
-            serverLog(LL_NOTICE,"Full resync from master: %s:%lld",
-                server.master_replid,
-                server.master_initial_offset);
-        }
-        sdsfree(reply);
-        return PSYNC_FULLRESYNC;
-    }
-
-    if (!strncmp(reply, "+RDBCHANNELSYNC", strlen("+RDBCHANNELSYNC"))) {
-        char *client_id = strchr(reply,' ');
-        if (client_id)
-            client_id++;
-
-        if (!client_id) {
-            serverLog(LL_WARNING,
-                      "Master replied with wrong +RDBCHANNELSYNC syntax: %s", reply);
-            sdsfree(reply);
-            return PSYNC_NOT_SUPPORTED;
-        }
-        server.repl_main_ch_client_id = strtoll(client_id, NULL, 10);;
-        /* A response of +RDBCHANNELSYNC from the master implies that partial
-         * synchronization is not possible and that the master supports full
-         * sync using dedicated RDB channel. Full sync will continue that way.*/
-        serverLog(LL_NOTICE, "PSYNC is not possible, initialize RDB channel.");
-        sdsfree(reply);
-        return PSYNC_FULLRESYNC_RDBCHANNEL;
-    }
-
-    if (!strncmp(reply,"+CONTINUE",9)) {
-        /* Partial resync was accepted. */
-        serverLog(LL_NOTICE,
-            "Successful partial resynchronization with master.");
-
-        /* Check the new replication ID advertised by the master. If it
-         * changed, we need to set the new ID as primary ID, and set
-         * secondary ID as the old master ID up to the current offset, so
-         * that our sub-slaves will be able to PSYNC with us after a
-         * disconnection. */
-        char *start = reply+10;
-        char *end = reply+9;
-        while(end[0] != '\r' && end[0] != '\n' && end[0] != '\0') end++;
-        if (end-start == CONFIG_RUN_ID_SIZE) {
-            char new[CONFIG_RUN_ID_SIZE+1];
-            memcpy(new,start,CONFIG_RUN_ID_SIZE);
-            new[CONFIG_RUN_ID_SIZE] = '\0';
-
-            if (strcmp(new,server.cached_master->replid)) {
-                /* Master ID changed. */
-                serverLog(LL_NOTICE,"Master replication ID changed to %s",new);
-
-                /* Set the old ID as our ID2, up to the current offset+1. */
-                memcpy(server.replid2,server.cached_master->replid,
-                    sizeof(server.replid2));
-                server.second_replid_offset = server.master_repl_offset+1;
-
-                /* Update the cached master ID and our own primary ID to the
-                 * new one. */
-                memcpy(server.replid,new,sizeof(server.replid));
-                memcpy(server.cached_master->replid,new,sizeof(server.replid));
-
-                /* Disconnect all the sub-slaves: they need to be notified. */
-                disconnectSlaves();
-            }
-        }
-
-        /* Setup the replication to continue. */
-        sdsfree(reply);
-        replicationResurrectCachedMaster(conn);
-
-        /* If this instance was restarted and we read the metadata to
-         * PSYNC from the persistence file, our replication backlog could
-         * be still not initialized. Create it. */
-        if (server.repl_backlog == NULL) createReplicationBacklog();
-        return PSYNC_CONTINUE;
-    }
-
-    /* If we reach this point we received either an error (since the master does
-     * not understand PSYNC or because it is in a special state and cannot
-     * serve our request), or an unexpected reply from the master.
-     *
-     * Return PSYNC_NOT_SUPPORTED on errors we don't understand, otherwise
-     * return PSYNC_TRY_LATER if we believe this is a transient error. */
-
-    if (!strncmp(reply,"-NOMASTERLINK",13) ||
-        !strncmp(reply,"-LOADING",8))
-    {
-        serverLog(LL_NOTICE,
-            "Master is currently unable to PSYNC "
-            "but should be in the future: %s", reply);
-        sdsfree(reply);
-        return PSYNC_TRY_LATER;
-    }
-
-    if (strncmp(reply,"-ERR",4)) {
-        /* If it's not an error, log the unexpected event. */
-        serverLog(LL_WARNING,
-            "Unexpected reply to PSYNC from master: %s", reply);
-    } else {
-        serverLog(LL_NOTICE,
-            "Master does not support PSYNC or is in "
-            "error state (reply: %s)", reply);
-    }
-    sdsfree(reply);
-    return PSYNC_NOT_SUPPORTED;
-}
-
-/* This handler fires when the non blocking connect was able to
- * establish a connection with the master. */
-void syncWithMaster(connection *conn) {
-    char tmpfile[256], *err = NULL;
-    int dfd = -1, maxtries = 5;
-    int psync_result;
-    static int replconf_rdb_no_compress = 0;
-
-    /* If this event fired after the user turned the instance into a master
-     * with SLAVEOF NO ONE we must just return ASAP. */
-    if (server.repl_state == REPL_STATE_NONE) {
-        connClose(conn);
-        return;
-    }
-
-    /* Check for errors in the socket: after a non blocking connect() we
-     * may find that the socket is in error state. */
-    if (connGetState(conn) != CONN_STATE_CONNECTED) {
-        serverLog(LL_WARNING,"Error condition on socket for SYNC: %s",
-                connGetLastError(conn));
-        goto error;
-    }
-
-    /* Send a PING to check the master is able to reply without errors. */
-    if (server.repl_state == REPL_STATE_CONNECTING) {
-        serverLog(LL_NOTICE,"Non blocking connect for SYNC fired the event.");
-        /* Delete the writable event so that the readable event remains
-         * registered and we can wait for the PONG reply. */
-        connSetReadHandler(conn, syncWithMaster);
-        connSetWriteHandler(conn, NULL);
-        server.repl_state = REPL_STATE_RECEIVE_PING_REPLY;
-        /* Send the PING, don't check for errors at all, we have the timeout
-         * that will take care about this. */
-        err = sendCommand(conn,"PING",NULL);
-        if (err) goto write_error;
-        return;
-    }
-
-    /* Receive the PONG command. */
-    if (server.repl_state == REPL_STATE_RECEIVE_PING_REPLY) {
-        err = receiveSynchronousResponse(conn);
-
-        /* The master did not reply */
-        if (err == NULL) goto no_response_error;
-
-        /* We accept only two replies as valid, a positive +PONG reply
-         * (we just check for "+") or an authentication error.
-         * Note that older versions of Redis replied with "operation not
-         * permitted" instead of using a proper error code, so we test
-         * both. */
-        if (err[0] != '+' &&
-            strncmp(err,"-NOAUTH",7) != 0 &&
-            strncmp(err,"-NOPERM",7) != 0 &&
-            strncmp(err,"-ERR operation not permitted",28) != 0)
-        {
-            serverLog(LL_WARNING,"Error reply to PING from master: '%s'",err);
-            sdsfree(err);
-            goto error;
-        } else {
-            serverLog(LL_NOTICE,
-                "Master replied to PING, replication can continue...");
-        }
-        sdsfree(err);
-        err = NULL;
-        server.repl_state = REPL_STATE_SEND_HANDSHAKE;
-    }
-
-    if (server.repl_state == REPL_STATE_SEND_HANDSHAKE) {
-        /* AUTH with the master if required. */
-        if (server.masterauth) {
-            char *args[3] = {"AUTH",NULL,NULL};
-            size_t lens[3] = {4,0,0};
-            int argc = 1;
-            if (server.masteruser) {
-                args[argc] = server.masteruser;
-                lens[argc] = strlen(server.masteruser);
-                argc++;
-            }
-            args[argc] = server.masterauth;
-            lens[argc] = sdslen(server.masterauth);
-            argc++;
-            err = sendCommandArgv(conn, argc, args, lens);
-            if (err) goto write_error;
-        }
-
-        /* Set the slave port, so that Master's INFO command can list the
-         * slave listening port correctly. */
-        {
-            char buf[LONG_STR_SIZE];
-
-            slaveGetPortStr(buf, sizeof(buf));
-            err = sendCommand(conn,"REPLCONF",
-                    "listening-port",buf, NULL);
-            if (err) goto write_error;
-        }
-
-        /* Set the slave ip, so that Master's INFO command can list the
-         * slave IP address port correctly in case of port forwarding or NAT.
-         * Skip REPLCONF ip-address if there is no slave-announce-ip option set. */
-        if (server.slave_announce_ip) {
-            err = sendCommand(conn,"REPLCONF",
-                    "ip-address",server.slave_announce_ip, NULL);
-            if (err) goto write_error;
-        }
-
-        /* If we are not going to save the RDB to disk, request that RDB
-         * compression be disabled, which speeds up RDB delivery. */
-        replconf_rdb_no_compress = 0;
-        if (useDisklessLoad()) {
-            replconf_rdb_no_compress = 1;
-            err = sendCommand(conn, "REPLCONF", "rdb-no-compress", "1", NULL);
-            if (err) goto write_error;
-        }
-
-        /* Inform the master of our (slave) capabilities.
-         *
-         * EOF: supports EOF-style RDB transfer for diskless replication.
-         * PSYNC2: supports PSYNC v2, so understands +CONTINUE <new repl ID>.
-         *
-         * The master will ignore capabilities it does not understand. */
-        err = sendCommand(conn,"REPLCONF",
-                          "capa","eof","capa","psync2",
-                          server.repl_rdb_channel ? "capa" : NULL, "rdb-channel-repl", NULL);
-
-        if (err) goto write_error;
-
-        server.repl_state = REPL_STATE_RECEIVE_AUTH_REPLY;
-        return;
-    }
-
-    if (server.repl_state == REPL_STATE_RECEIVE_AUTH_REPLY && !server.masterauth)
-        server.repl_state = REPL_STATE_RECEIVE_PORT_REPLY;
-
-    /* Receive AUTH reply. */
-    if (server.repl_state == REPL_STATE_RECEIVE_AUTH_REPLY) {
-        err = receiveSynchronousResponse(conn);
-        if (err == NULL) goto no_response_error;
-        if (err[0] == '-') {
-            serverLog(LL_WARNING,"Unable to AUTH to MASTER: %s",err);
-            sdsfree(err);
-            goto error;
-        }
-        sdsfree(err);
-        err = NULL;
-        server.repl_state = REPL_STATE_RECEIVE_PORT_REPLY;
-        return;
-    }
-
-    /* Receive REPLCONF listening-port reply. */
-    if (server.repl_state == REPL_STATE_RECEIVE_PORT_REPLY) {
-        err = receiveSynchronousResponse(conn);
-        if (err == NULL) goto no_response_error;
-        /* Ignore the error if any, not all the Redis versions support
-         * REPLCONF listening-port. */
-        if (err[0] == '-') {
-            serverLog(LL_NOTICE,"(Non critical) Master does not understand "
-                                "REPLCONF listening-port: %s", err);
-        }
-        sdsfree(err);
-        server.repl_state = REPL_STATE_RECEIVE_IP_REPLY;
-        return;
-    }
-
-    if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY && !server.slave_announce_ip)
-        server.repl_state = REPL_STATE_RECEIVE_COMP_REPLY;
-
-    /* Receive REPLCONF ip-address reply. */
-    if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY) {
-        err = receiveSynchronousResponse(conn);
-        if (err == NULL) goto no_response_error;
-        /* Ignore the error if any, not all the Redis versions support
-         * REPLCONF ip-address. */
-        if (err[0] == '-') {
-            serverLog(LL_NOTICE,"(Non critical) Master does not understand "
-                                "REPLCONF ip-address: %s", err);
-        }
-        sdsfree(err);
-        server.repl_state = REPL_STATE_RECEIVE_COMP_REPLY;
-        return;
-    }
-
-    if (server.repl_state == REPL_STATE_RECEIVE_COMP_REPLY && !replconf_rdb_no_compress)
-        server.repl_state = REPL_STATE_RECEIVE_CAPA_REPLY;
-
-    /* Receive REPLCONF rdb-no-compress reply. */
-    if (server.repl_state == REPL_STATE_RECEIVE_COMP_REPLY) {
-        err = receiveSynchronousResponse(conn);
-        if (err == NULL) goto no_response_error;
-        /* Ignore the error if any, not all the Redis versions support
-         * REPLCONF rdb-no-compress. */
-        if (err[0] == '-') {
-            serverLog(LL_NOTICE,"(Non critical) Master does not understand "
-                                "REPLCONF rdb-no-compress: %s", err);
-        }
-        sdsfree(err);
-        server.repl_state = REPL_STATE_RECEIVE_CAPA_REPLY;
-        return;
-    }
-
-    /* Receive CAPA reply. */
-    if (server.repl_state == REPL_STATE_RECEIVE_CAPA_REPLY) {
-        err = receiveSynchronousResponse(conn);
-        if (err == NULL) goto no_response_error;
-        /* Ignore the error if any, not all the Redis versions support
-         * REPLCONF capa. */
-        if (err[0] == '-') {
-            serverLog(LL_NOTICE,"(Non critical) Master does not understand "
-                                  "REPLCONF capa: %s", err);
-        }
-        sdsfree(err);
-        err = NULL;
-        server.repl_state = REPL_STATE_SEND_PSYNC;
-    }
-
-    /* Try a partial resynchronization. If we don't have a cached master
-     * slaveTryPartialResynchronization() will at least try to use PSYNC
-     * to start a full resynchronization so that we get the master replid
-     * and the global offset, to try a partial resync at the next
-     * reconnection attempt. */
-    if (server.repl_state == REPL_STATE_SEND_PSYNC) {
-        if (slaveTryPartialResynchronization(conn,0) == PSYNC_WRITE_ERROR) {
-            err = sdsnew("Write error sending the PSYNC command.");
-            abortFailover("Write error to failover target");
-            goto write_error;
-        }
-        server.repl_state = REPL_STATE_RECEIVE_PSYNC_REPLY;
-        return;
-    }
-
-    /* If reached this point, we should be in REPL_STATE_RECEIVE_PSYNC_REPLY. */
-    if (server.repl_state != REPL_STATE_RECEIVE_PSYNC_REPLY) {
-        serverLog(LL_WARNING,"syncWithMaster(): state machine error, "
-                             "state should be RECEIVE_PSYNC_REPLY but is %d",
-                             server.repl_state);
-        goto error;
-    }
-
-    psync_result = slaveTryPartialResynchronization(conn,1);
-    if (psync_result == PSYNC_WAIT_REPLY) return; /* Try again later... */
-
-    /* Check the status of the planned failover. We expect PSYNC_CONTINUE,
-     * but there is nothing technically wrong with a full resync which
-     * could happen in edge cases. */
-    if (server.failover_state == FAILOVER_IN_PROGRESS) {
-        if (psync_result == PSYNC_CONTINUE ||
-            psync_result == PSYNC_FULLRESYNC ||
-            psync_result == PSYNC_FULLRESYNC_RDBCHANNEL)
-        {
-            clearFailoverState();
-        } else {
-            abortFailover("Failover target rejected psync request");
-            return;
-        }
-    }
-
-    /* If the master is in an transient error, we should try to PSYNC
-     * from scratch later, so go to the error path. This happens when
-     * the server is loading the dataset or is not connected with its
-     * master and so forth. */
-    if (psync_result == PSYNC_TRY_LATER) goto error;
-
-    /* Note: if PSYNC does not return WAIT_REPLY, it will take care of
-     * uninstalling the read handler from the file descriptor. */
-
-    if (psync_result == PSYNC_CONTINUE) {
-        serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Master accepted a Partial Resynchronization.");
-        if (server.supervised_mode == SUPERVISED_SYSTEMD) {
-            redisCommunicateSystemd("STATUS=MASTER <-> REPLICA sync: Partial Resynchronization accepted. Ready to accept connections in read-write mode.\n");
-        }
-        return;
-    }
-
-    /* Fall back to SYNC if needed. Otherwise psync_result == PSYNC_FULLRESYNC
-     * and the server.master_replid and master_initial_offset are
-     * already populated. */
-    if (psync_result == PSYNC_NOT_SUPPORTED) {
-        serverLog(LL_NOTICE,"Retrying with SYNC...");
-        if (connSyncWrite(conn,"SYNC\r\n",6,server.repl_syncio_timeout*1000) == -1) {
-            serverLog(LL_WARNING,"I/O error writing to MASTER: %s",
-                connGetLastError(conn));
-            goto error;
-        }
-    }
-
-    /* Prepare a suitable temp file for bulk transfer */
-    if (!useDisklessLoad()) {
-        while(maxtries--) {
-            snprintf(tmpfile,256,
-                "temp-%d.%ld.rdb",(int)server.unixtime,(long int)getpid());
-            dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
-            if (dfd != -1) break;
-            sleep(1);
-        }
-        if (dfd == -1) {
-            serverLog(LL_WARNING,"Opening the temp file needed for MASTER <-> REPLICA synchronization: %s",strerror(errno));
-            goto error;
-        }
-        server.repl_transfer_tmpfile = zstrdup(tmpfile);
-        server.repl_transfer_fd = dfd;
-    }
-
-    server.repl_transfer_size = -1;
-    server.repl_transfer_read = 0;
-    server.repl_transfer_last_fsync_off = 0;
-    server.repl_transfer_lastio = server.unixtime;
-
-    /* Using rdb channel replication, the master responded +RDBCHANNELSYNC.
-     * We need to initialize the RDB channel. */
-    if (psync_result == PSYNC_FULLRESYNC_RDBCHANNEL) {
-        /* Create RDB connection */
-        server.repl_rdb_transfer_s = connCreate(server.el, connTypeOfReplication());
-        if (connConnect(server.repl_rdb_transfer_s, server.masterhost,
-                        server.masterport, server.bind_source_addr,
-                        rdbChannelFullSyncWithMaster) == C_ERR) {
-            serverLog(LL_WARNING, "Unable to connect to master: %s", connGetLastError(server.repl_rdb_transfer_s));
-            goto error;
-        }
-        server.repl_rdb_ch_state = REPL_RDB_CH_SEND_HANDSHAKE;
-        connSetReadHandler(server.repl_transfer_s, NULL);
-        return;
-    }
-
-    /* Setup the non blocking download of the bulk file. */
-    if (connSetReadHandler(conn, readSyncBulkPayload)
-            == C_ERR)
-    {
-        char conninfo[CONN_INFO_LEN];
-        serverLog(LL_WARNING,
-            "Can't create readable event for SYNC: %s (%s)",
-            strerror(errno), connGetInfo(conn, conninfo, sizeof(conninfo)));
-        goto error;
-    }
-
-    server.repl_state = REPL_STATE_TRANSFER;
-    return;
-
-no_response_error: /* Handle receiveSynchronousResponse() error when master has no reply */
-    serverLog(LL_WARNING, "Master did not respond to command during SYNC handshake");
-    /* Fall through to regular error handling */
-
-error:
-    if (dfd != -1) close(dfd);
-    connClose(conn);
-    if (server.repl_rdb_transfer_s)
-        connClose(server.repl_rdb_transfer_s);
-    server.repl_rdb_transfer_s = NULL;
-    server.repl_transfer_s = NULL;
-    if (server.repl_transfer_fd != -1)
-        close(server.repl_transfer_fd);
-    if (server.repl_transfer_tmpfile)
-        zfree(server.repl_transfer_tmpfile);
-    server.repl_transfer_tmpfile = NULL;
-    server.repl_transfer_fd = -1;
-    server.repl_state = REPL_STATE_CONNECT;
-    return;
-
-write_error: /* Handle sendCommand() errors. */
-    serverLog(LL_WARNING,"Sending command to master in replication handshake: %s", err);
-    sdsfree(err);
-    goto error;
-}
-
-int connectWithMaster(void) {
-    server.repl_current_sync_attempts++;
-    server.repl_total_sync_attempts++;
-    server.repl_transfer_s = connCreate(server.el, connTypeOfReplication());
-    if (connConnect(server.repl_transfer_s, server.masterhost, server.masterport,
-                server.bind_source_addr, syncWithMaster) == C_ERR) {
-        serverLog(LL_WARNING,"Unable to connect to MASTER: %s",
-                connGetLastError(server.repl_transfer_s));
-        connClose(server.repl_transfer_s);
-        server.repl_transfer_s = NULL;
-        return C_ERR;
-    }
-
-
-    server.repl_transfer_lastio = server.unixtime;
-    server.repl_state = REPL_STATE_CONNECTING;
-    serverLog(LL_NOTICE,"MASTER <-> REPLICA sync started");
-    return C_OK;
-}
-
-/* This function can be called when a non blocking connection is currently
- * in progress to undo it.
- * Never call this function directly, use cancelReplicationHandshake() instead.
- */
-void undoConnectWithMaster(void) {
-    connClose(server.repl_transfer_s);
-    server.repl_transfer_s = NULL;
-}
-
-/* Abort the async download of the bulk dataset while SYNC-ing with master.
- * Never call this function directly, use cancelReplicationHandshake() instead.
- */
-void replicationAbortSyncTransfer(void) {
-    serverAssert(server.repl_state == REPL_STATE_TRANSFER);
-    undoConnectWithMaster();
-    if (server.repl_disconnect_start_time == 0)
-        server.repl_disconnect_start_time = server.unixtime;
-    if (server.repl_transfer_fd!=-1) {
-        close(server.repl_transfer_fd);
-        bg_unlink(server.repl_transfer_tmpfile);
-        zfree(server.repl_transfer_tmpfile);
-        server.repl_transfer_tmpfile = NULL;
-        server.repl_transfer_fd = -1;
-    }
-}
-
-/* This function aborts a non blocking replication attempt if there is one
- * in progress, by canceling the non-blocking connect attempt or
- * the initial bulk transfer.
- *
- * If there was a replication handshake in progress 1 is returned and
- * the replication state (server.repl_state) set to REPL_STATE_CONNECT.
- *
- * Otherwise zero is returned and no operation is performed at all. */
-int cancelReplicationHandshake(int reconnect) {
-    if (rdbChannelAbort() != C_OK)
-        return 1;
-
-    if (server.repl_state == REPL_STATE_TRANSFER) {
-        replicationAbortSyncTransfer();
-        server.repl_state = REPL_STATE_CONNECT;
-    } else if (server.repl_state == REPL_STATE_CONNECTING ||
-               slaveIsInHandshakeState())
-    {
-        undoConnectWithMaster();
-        server.repl_state = REPL_STATE_CONNECT;
-    } else {
-        return 0;
-    }
-
-    if (!reconnect)
-        return 1;
-
-    /* try to re-connect without waiting for replicationCron, this is needed
-     * for the "diskless loading short read" test. */
-    serverLog(LL_NOTICE,"Reconnecting to MASTER %s:%d after failure",
-        server.masterhost, server.masterport);
-    connectWithMaster();
-
-    return 1;
-}
-
-/* Set replication to the specified master address and port. */
-void replicationSetMaster(char *ip, int port) {
-    int was_master = server.masterhost == NULL;
-
-    sdsfree(server.masterhost);
-    server.masterhost = NULL;
-    if (server.master) {
-        freeClient(server.master);
-    }
-    disconnectAllBlockedClients(); /* Clients blocked in master, now slave. */
-
-    /* Setting masterhost only after the call to freeClient since it calls
-     * replicationHandleMasterDisconnection which can trigger a re-connect
-     * directly from within that call. */
-    server.masterhost = sdsnew(ip);
-    server.masterport = port;
-
-    /* Update oom_score_adj */
-    setOOMScoreAdj(-1);
-
-    /* Here we don't disconnect with replicas, since they may hopefully be able
-     * to partially resync with us. We will disconnect with replicas and force
-     * them to resync with us when changing replid on partially resync with new
-     * master, or finishing transferring RDB and preparing loading DB on full
-     * sync with new master. */
-
-    cancelReplicationHandshake(0);
-    /* Before destroying our master state, create a cached master using
-     * our own parameters, to later PSYNC with the new master. */
-    if (was_master) {
-        replicationDiscardCachedMaster();
-        replicationCacheMasterUsingMyself();
-    }
-
-    /* Fire the role change modules event. */
-    moduleFireServerEvent(REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED,
-                          REDISMODULE_EVENT_REPLROLECHANGED_NOW_REPLICA,
-                          NULL);
-
-    /* Fire the master link modules event. */
-    if (server.repl_state == REPL_STATE_CONNECTED)
-        moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
-                              REDISMODULE_SUBEVENT_MASTER_LINK_DOWN,
-                              NULL);
-
-    server.repl_state = REPL_STATE_CONNECT;
-    server.repl_current_sync_attempts = 0;
-    server.repl_total_sync_attempts = 0;
-    serverLog(LL_NOTICE,"Connecting to MASTER %s:%d",
-        server.masterhost, server.masterport);
-    connectWithMaster();
-}
-
-/* Cancel replication, setting the instance as a master itself. */
-void replicationUnsetMaster(void) {
-    if (server.masterhost == NULL) return; /* Nothing to do. */
-
-    /* Fire the master link modules event. */
-    if (server.repl_state == REPL_STATE_CONNECTED)
-        moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
-                              REDISMODULE_SUBEVENT_MASTER_LINK_DOWN,
-                              NULL);
-
-    /* Clear masterhost first, since the freeClient calls
-     * replicationHandleMasterDisconnection which can attempt to re-connect. */
-    sdsfree(server.masterhost);
-    server.masterhost = NULL;
-    if (server.master) freeClient(server.master);
-    replicationDiscardCachedMaster();
-    cancelReplicationHandshake(0);
-    /* When a slave is turned into a master, the current replication ID
-     * (that was inherited from the master at synchronization time) is
-     * used as secondary ID up to the current offset, and a new replication
-     * ID is created to continue with a new replication history. */
-    shiftReplicationId();
-    /* Disconnecting all the slaves is required: we need to inform slaves
-     * of the replication ID change (see shiftReplicationId() call). However
-     * the slaves will be able to partially resync with us, so it will be
-     * a very fast reconnection. */
-    disconnectSlaves();
-    server.repl_state = REPL_STATE_NONE;
-    /* Reset the attempts number. */
-    server.repl_current_sync_attempts = 0;
-    server.repl_total_sync_attempts = 0;
-    /* We need to make sure the new master will start the replication stream
-     * with a SELECT statement. This is forced after a full resync, but
-     * with PSYNC version 2, there is no need for full resync after a
-     * master switch. */
-    server.slaveseldb = -1;
-
-    /* Update oom_score_adj */
-    setOOMScoreAdj(-1);
-
-    /* Once we turn from slave to master, we consider the starting time without
-     * slaves (that is used to count the replication backlog time to live) as
-     * starting from now. Otherwise the backlog will be freed after a
-     * failover if slaves do not connect immediately. */
-    server.repl_no_slaves_since = server.unixtime;
-    
-    /* Reset up and down time so it'll be ready for when we turn into replica again. */
-    server.repl_down_since = 0;
-    server.repl_up_since = 0;
-    /* Fire the role change modules event. */
-    moduleFireServerEvent(REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED,
-                          REDISMODULE_EVENT_REPLROLECHANGED_NOW_MASTER,
-                          NULL);
-
-    /* Restart the AOF subsystem in case we shut it down during a sync when
-     * we were still a slave. */
-    if (server.aof_enabled && server.aof_state == AOF_OFF) {
-        serverLog(LL_NOTICE, "Restarting AOF after becoming master");
-        startAppendOnlyWithRetry();
-    }
-}
-
-/* This function is called when the slave lose the connection with the
- * master into an unexpected way. */
-void replicationHandleMasterDisconnection(void) {
-    /* Fire the master link modules event. */
-    if (server.repl_state == REPL_STATE_CONNECTED)
-        moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
-                              REDISMODULE_SUBEVENT_MASTER_LINK_DOWN,
-                              NULL);
-
-    server.master = NULL;
-    if (server.repl_state == REPL_STATE_CONNECTED)
-        server.repl_current_sync_attempts = 0;
-    server.repl_state = REPL_STATE_CONNECT;
-    server.repl_down_since = server.unixtime;
-    server.repl_up_since = 0;
-    server.repl_num_master_disconnection++;
-
-    /* If we are in the loop of streaming accumulated buffers, discard the
-     * buffer and clean up the rdbchannel state. The outer loop will abort once
-     * it detects that the master client has been disconnected. For details,
-     * see rdbChannelStreamReplDataToDb() */
-    if (server.repl_main_ch_state & REPL_MAIN_CH_STREAMING_BUF)
-        rdbChannelCleanup();
-
-    if (server.repl_disconnect_start_time == 0)
-        server.repl_disconnect_start_time = server.unixtime;
-    /* We lost connection with our master, don't disconnect slaves yet,
-     * maybe we'll be able to PSYNC with our master later. We'll disconnect
-     * the slaves only if we'll have to do a full resync with our master. */
-
-    /* Try to re-connect immediately rather than wait for replicationCron
-     * waiting 1 second may risk backlog being recycled. */
-    if (server.masterhost) {
-        serverLog(LL_NOTICE,"Reconnecting to MASTER %s:%d",
-            server.masterhost, server.masterport);
-        connectWithMaster();
-    }
-}
-
-/* Rdb channel for full sync
- *
- * - During a full sync, when master is delivering RDB to the replica, incoming
- *   write commands are kept in a replication buffer in order to be sent to the
- *   replica once RDB delivery is completed. If RDB delivery takes a long time,
- *   it might create memory pressure on master. Also, once a replica connection
- *   accumulates replication data which is larger than output buffer limits,
- *   master will kill replica connection. This may cause a replication failure.
- *
- *   The main benefit of the rdb channel replication is streaming incoming
- *   commands in parallel to the RDB delivery. This approach shifts replication
- *   stream buffering to the replica and reduces load on master. We do this by
- *   opening another connection for RDB delivery. The main channel on replica
- *   will be receiving replication stream while rdb channel is receiving the RDB.
- *
- *   This feature also helps to reduce master's main process CPU load. By
- *   opening a dedicated connection for the RDB transfer, the bgsave process has
- *   direct access to the new connection and it will stream RDB directly to the
- *   replicas. Before this change, due to TLS connection restriction, the bgsave
- *   process was writing RDB bytes to a pipe and the main process was forwarding
- *   it to the replica. This is no longer necessary, the main process can avoid
- *   these expensive socket read/write syscalls.
- *
- *  Implementation
- *  - When replica connects to the master, it sends 'rdb-channel-repl' as part
- *    of capability exchange to let master to know replica supports rdb channel.
- *  - When replica lacks sufficient data for PSYNC, master sends +RDBCHANNELSYNC
- *    reply with replica's client id. As the next step, the replica opens a new
- *    connection (rdb-channel) and configures it against the master with the
- *    appropriate capabilities and requirements. It also sends given client id
- *    back to master over rdbchannel so that master can associate these
- *    channels (initial replica connection will be referred as main-channel)
- *    Then, replica requests fullsync using the RDB channel.
- *  - Prior to forking, master attaches the replica's main channel to the
- *    replication backlog to deliver replication stream starting at the snapshot
- *    end offset.
- *  - The master main process sends replication stream via the main channel,
- *    while the bgsave process sends the RDB directly to the replica via the
- *    rdb-channel. Replica accumulates replication stream in a local buffer,
- *    while the RDB is being loaded into the memory.
- *  - Once the replica completes loading the rdb, it drops the rdb channel and
- *    streams the accumulated replication stream into the db. Sync is completed.
- *
- * * Replica state machine *
- *
- * Main channel state
- * ┌───────────────────┐
- * │RECEIVE_PING_REPLY │
- * └────────┬──────────┘
- *          │ +PONG
- * ┌────────▼──────────┐
- * │SEND_HANDSHAKE     │                     RDB channel state
- * └────────┬──────────┘            ┌───────────────────────────────┐
- *          │+OK                ┌───► RDB_CH_SEND_HANDSHAKE         │
- * ┌────────▼──────────┐        │   └──────────────┬────────────────┘
- * │RECEIVE_AUTH_REPLY │        │    REPLCONF main-ch-client-id <clientid>
- * └────────┬──────────┘        │   ┌──────────────▼────────────────┐
- *          │+OK                │   │ RDB_CH_RECEIVE_AUTH_REPLY     │
- * ┌────────▼──────────┐        │   └──────────────┬────────────────┘
- * │RECEIVE_PORT_REPLY │        │                  │ +OK
- * └────────┬──────────┘        │   ┌──────────────▼────────────────┐
- *          │+OK                │   │  RDB_CH_RECEIVE_REPLCONF_REPLY│
- * ┌────────▼──────────┐        │   └──────────────┬────────────────┘
- * │RECEIVE_IP_REPLY   │        │                  │ +OK
- * └────────┬──────────┘        │   ┌──────────────▼────────────────┐
- *          │+OK                │   │ RDB_CH_RECEIVE_FULLRESYNC     │
- * ┌────────▼──────────┐        │   └──────────────┬────────────────┘
- * │RECEIVE_CAPA_REPLY │        │                  │+FULLRESYNC
- * └────────┬──────────┘        │                  │Rdb delivery
- *          │                   │   ┌──────────────▼────────────────┐
- * ┌────────▼──────────┐        │   │ RDB_CH_RDB_LOADING            │
- * │SEND_PSYNC         │        │   └──────────────┬────────────────┘
- * └─┬─────────────────┘        │                  │ Done loading
- *   │PSYNC (use cached-master) │                  │
- * ┌─▼─────────────────┐        │                  │
- * │RECEIVE_PSYNC_REPLY│        │    ┌────────────►│ Replica streams replication
- * └─┬─────────────────┘        │    │             │ buffer into memory
- *   │                          │    │             │
- *   │+RDBCHANNELSYNC client-id │    │             │
- *   ├──────┬───────────────────┘    │             │
- *   │      │ Main channel           │             │
- *   │      │ accumulates repl data  │             │
- *   │   ┌──▼────────────────┐       │     ┌───────▼───────────┐
- *   │   │ REPL_TRANSFER     ├───────┘     │    CONNECTED      │
- *   │   └───────────────────┘             └────▲───▲──────────┘
- *   │                                          │   │
- *   │                                          │   │
- *   │  +FULLRESYNC    ┌───────────────────┐    │   │
- *   ├────────────────► REPL_TRANSFER      ├────┘   │
- *   │                 └───────────────────┘        │
- *   │  +CONTINUE                                   │
- *   └──────────────────────────────────────────────┘
- */
-
-/* Replication: Replica side. */
-static int rdbChannelSendHandshake(connection *conn, sds *err) {
-    /* AUTH with the master if required. */
-    if (server.masterauth) {
-        char *args[] = {"AUTH", NULL, NULL};
-        size_t lens[] = {4, 0, 0};
-        int argc = 1;
-        if (server.masteruser) {
-            args[argc] = server.masteruser;
-            lens[argc] = strlen(server.masteruser);
-            argc++;
-        }
-        args[argc] = server.masterauth;
-        lens[argc] = sdslen(server.masterauth);
-        argc++;
-        *err = sendCommandArgv(conn, argc, args, lens);
-        if (*err) {
-            serverLog(LL_WARNING, "Error sending AUTH to master in rdb channel replication handshake: %s", *err);
-            return C_ERR;
-        }
-    }
-
-    char buf[LONG_STR_SIZE];
-    slaveGetPortStr(buf, sizeof(buf));
-
-    char cid[LONG_STR_SIZE];
-    ull2string(cid, sizeof(cid), server.repl_main_ch_client_id);
-
-    *err = sendCommand(conn, "REPLCONF", "capa", "eof", "rdb-only", "1",
-                       "rdb-channel", "1", "main-ch-client-id", cid,
-                       "listening-port", buf, NULL);
-    if (*err) {
-        serverLog(LL_WARNING, "Error sending REPLCONF command to master in rdb channel handshake: %s", *err);
-        return C_ERR;
-    }
-
-    if (connSetReadHandler(conn, rdbChannelFullSyncWithMaster) == C_ERR) {
-        char conninfo[CONN_INFO_LEN];
-        serverLog(LL_WARNING, "Can't create readable event for SYNC: %s (%s)",
-                  strerror(errno), connGetInfo(conn, conninfo, sizeof(conninfo)));
-        return C_ERR;
-    }
-    return C_OK;
-}
-
-/* Replication: Replica side. */
-static int rdbChannelHandleAuthReply(connection *conn, sds *err) {
-    *err = receiveSynchronousResponse(conn);
-    if (*err == NULL) {
-        serverLog(LL_WARNING, "Master did not respond to auth command during rdb channel handshake");
-        return C_ERR;
-    }
-    if ((*err)[0] == '-') {
-        serverLog(LL_WARNING, "Unable to AUTH to master: %s", *err);
-        return C_ERR;
-    }
-    server.repl_rdb_ch_state = REPL_RDB_CH_RECEIVE_REPLCONF_REPLY;
-    return C_OK;
-}
-
-/* Replication: Replica side. */
-static int rdbChannelHandleReplconfReply(connection *conn, sds *err) {
-    *err = receiveSynchronousResponse(conn);
-    if (*err == NULL) {
-        serverLog(LL_WARNING, "Master did not respond to replconf command during rdb channel handshake");
-        return C_ERR;
-    }
-    if (*err[0] == '-') {
-        serverLog(LL_WARNING, "Master replied error to replconf: %s", *err);
-        return C_ERR;
-    }
-    sdsfree(*err);
-
-    if (server.repl_debug_pause & REPL_DEBUG_BEFORE_RDB_CHANNEL)
-        debugPauseProcess();
-
-    /* Request rdb from master */
-    *err = sendCommand(conn, "PSYNC", "?", "-1", NULL);
-    if (*err) {
-        serverLog(LL_WARNING, "I/O error writing to Master: %s", *err);
-        return C_ERR;
-    }
-
-    return C_OK;
-}
-
-/* Replication: Replica side. */
-static int rdbChannelHandleFullresyncReply(connection *conn, sds *err) {
-    char *replid = NULL, *offset = NULL;
-
-    *err = receiveSynchronousResponse(conn);
-    if (*err == NULL)
-        return C_ERR;
-
-    if (*err[0] == '\0') {
-        /* Retry again later */
-        serverLog(LL_DEBUG, "Received empty psync reply");
-        return C_RETRY;
-    }
-
-    /* FULL RESYNC, parse the reply in order to extract the replid
-     * and the replication offset. */
-    replid = strchr(*err,' ');
-    if (replid) {
-        replid++;
-        offset = strchr(replid, ' ');
-        if (offset) offset++;
-    }
-    if (!replid || !offset || (offset-replid-1) != CONFIG_RUN_ID_SIZE) {
-        serverLog(LL_WARNING, "Received unexpected psync reply: %s", *err);
-        return C_ERR;
-    }
-    memcpy(server.master_replid, replid, offset-replid-1);
-    server.master_replid[CONFIG_RUN_ID_SIZE] = '\0';
-    server.master_initial_offset = strtoll(offset,NULL,10);
-
-    /* Prepare the main and rdb channels for rdb and repl stream delivery.*/
-    server.repl_state = REPL_STATE_TRANSFER;
-    rdbChannelReplDataBufInit();
-
-    serverLog(LL_NOTICE, "Starting to receive RDB and replication stream in parallel.");
-
-    /* Setup connection to accumulate repl data.  */
-    server.repl_main_ch_state = REPL_MAIN_CH_ACCUMULATE_BUF;
-    if (connSetReadHandler(server.repl_transfer_s,
-                           rdbChannelBufferReplData) != C_OK)
-    {
-        serverLog(LL_WARNING, "Can't set read handler for main channel: %s",
-                  strerror(errno));
-        return C_ERR;
-    }
-
-    /* Prepare RDB channel connection for RDB download. */
-    if (connSetReadHandler(server.repl_rdb_transfer_s,
-                           readSyncBulkPayload) != C_OK)
-    {
-        char inf[CONN_INFO_LEN];
-        serverLog(LL_WARNING,
-                  "Can't create readable event for rdb channel connection: %s (%s)",
-                  strerror(errno),
-                  connGetInfo(server.repl_rdb_transfer_s, inf, sizeof(inf)));
-        return C_ERR;
-    }
-
-    return C_OK;
-}
-
-/* Replication: Replica side.
- * This connection handler is used to initialize the RDB channel connection.*/
-static void rdbChannelFullSyncWithMaster(connection *conn) {
-    int ret = 0;
-    char *err = NULL;
-    serverAssert(conn == server.repl_rdb_transfer_s);
-
-    /* Check for errors in the socket: after a non blocking connect() we
-     * may find that the socket is in error state. */
-    if (connGetState(conn) != CONN_STATE_CONNECTED) {
-        serverLog(LL_WARNING, "Error condition on socket for rdb channel replication: %s",
-                  connGetLastError(conn));
-        goto error;
-    }
-    switch (server.repl_rdb_ch_state) {
-        case REPL_RDB_CH_SEND_HANDSHAKE:
-            ret = rdbChannelSendHandshake(conn, &err);
-            if (ret == C_OK)
-                server.repl_rdb_ch_state = REPL_RDB_CH_RECEIVE_AUTH_REPLY;
-            break;
-        case REPL_RDB_CH_RECEIVE_AUTH_REPLY:
-            if (server.masterauth) {
-                ret = rdbChannelHandleAuthReply(conn, &err);
-                if (ret == C_OK)
-                    server.repl_rdb_ch_state = REPL_RDB_CH_RECEIVE_REPLCONF_REPLY;
-                /* Wait for next bulk before trying to read replconf reply. */
-                break;
-            }
-            server.repl_rdb_ch_state = REPL_RDB_CH_RECEIVE_REPLCONF_REPLY;
-            /* fall through */
-        case REPL_RDB_CH_RECEIVE_REPLCONF_REPLY:
-            ret = rdbChannelHandleReplconfReply(conn, &err);
-            if (ret == C_OK)
-                server.repl_rdb_ch_state = REPL_RDB_CH_RECEIVE_FULLRESYNC;
-            break;
-        case REPL_RDB_CH_RECEIVE_FULLRESYNC:
-            ret = rdbChannelHandleFullresyncReply(conn, &err);
-            if (ret == C_OK)
-                server.repl_rdb_ch_state = REPL_RDB_CH_RDB_LOADING;
-            break;
-        default:
-            serverPanic("Unknown rdb channel state: %d", server.repl_rdb_ch_state);
-    }
-
-    if (ret == C_ERR)
-        goto error;
-
-    sdsfree(err);
-    return;
-
-error:
-    if (err) {
-        serverLog(LL_WARNING, "rdb channel sync failed with error: %s", err);
-        sdsfree(err);
-    }
-    if (server.repl_transfer_s) {
-        connClose(server.repl_transfer_s);
-        server.repl_transfer_s = NULL;
-    }
-    server.repl_state = REPL_STATE_CONNECT;
-    rdbChannelAbort();
-}
-
-void replDataBufInit(replDataBuf *buf) {
-    serverAssert(buf->blocks == NULL);
-    buf->size = 0;
-    buf->used = 0;
-    buf->last_num_blocks = 0;
-    buf->mem_used = 0;
-    buf->blocks = listCreate();
-    buf->blocks->free = zfree;
-}
-
-void replDataBufClear(replDataBuf *buf) {
-    if (buf->blocks) listRelease(buf->blocks);
-    buf->blocks = NULL;
-    buf->size = 0;
-    buf->used = 0;
-    buf->last_num_blocks = 0;
-    buf->mem_used = 0;
-}
-
-/* Replication: Replica side.
- * Initialize replica's local replication buffer to accumulate repl stream
- * during rdb channel sync. */
-static void rdbChannelReplDataBufInit(void) {
-    replDataBufInit(&server.repl_full_sync_buffer);
-}
-
-/* Replication: Replica side.
- * Clear replica's local replication buffer */
-static void rdbChannelReplDataBufClear(void) {
-    replDataBufClear(&server.repl_full_sync_buffer);
-}
-
-/* Generic function to read data from connection into the last block. */
-static int replDataBufReadIntoLastBlock(connection *conn, replDataBuf *buf,
-                                    void (*error_handler)(connection *conn))
-{
-    atomicIncr(server.stat_io_reads_processed[IOTHREAD_MAIN_THREAD_ID], 1);
-
-    replDataBufBlock *block = listNodeValue(listLast(buf->blocks));
-    serverAssert(block && block->size > block->used);
-
-    int nread = connRead(conn, block->buf + block->used, block->size - block->used);
-    if (nread <= 0) {
-        if (nread == 0 || connGetState(conn) != CONN_STATE_CONNECTED) {
-            error_handler(conn);
-        }
-        return -1;
-    }
-
-    block->used += nread;
-    if (buf) buf->used += nread;
-    atomicIncr(server.stat_net_repl_input_bytes, nread);
-
-    return nread;
-}
-
-/* Generic function to read data from connection into a buffer. */
-void replDataBufReadFromConn(connection *conn, replDataBuf *buf, void (*error_handler)(connection *conn)) {
-    const int buflen = 1024 * 1024;
-    const int minread = 16 * 1024;
-    int nread = 0;
-    int needs_read = 1;
-
-    listNode *ln = listLast(buf->blocks);
-    replDataBufBlock *tail = ln ? listNodeValue(ln) : NULL;
-
-    /* Try to append last node. */
-    if (tail && tail->size > tail->used) {
-        nread = replDataBufReadIntoLastBlock(conn, buf, error_handler);
-        if (nread <= 0)
-            return;
-
-        /* If buffer is filled fully, there might be more data in socket buffer.
-         * Only read again if we've read small amount (less than minread). */
-        needs_read = (tail->size == tail->used) && nread < minread;
-    }
-
-    if (needs_read) {
-        unsigned long long limit;
-        size_t usable_size;
-
-        /* For accumulation limit, if 'replica-full-sync-buffer-limit' is set,
-         * we'll use it. Otherwise, 'client-output-buffer-limit <replica>' is
-         * the limit.*/
-        limit = server.repl_full_sync_buffer_limit;
-        if (limit == 0)
-            limit = server.client_obuf_limits[CLIENT_TYPE_SLAVE].hard_limit_bytes;
-
-        if (limit != 0 && buf->size > limit) {
-            /* Currently this function is only used for replication and slots sync.
-             * Log accordingly, maybe should be extendable in the future. */
-            if (server.masterhost)
-                serverLog(LL_NOTICE, "Replication buffer limit has been reached (%llu bytes), "
-                    "stopped buffering replication stream. Further accumulation may occur on master side.", limit);
-            else
-                serverLog(LL_NOTICE, "Slots sync buffer limit has been reached (%llu bytes), "
-                    "stopped buffering slots sync stream. Further accumulation may occur on source side.", limit);
-
-            connSetReadHandler(conn, NULL);
-            return;
-        }
-
-        tail = zmalloc_usable(buflen, &usable_size);
-        tail->size = usable_size - sizeof(replDataBufBlock);
-        tail->used = 0;
-
-        listAddNodeTail(buf->blocks, tail);
-        buf->size += tail->size;
-        buf->mem_used += usable_size + sizeof(listNode);
-
-        /* Update buffer's peak */
-        if (buf->peak < buf->size)
-            buf->peak = buf->size;
-
-        replDataBufReadIntoLastBlock(conn, buf, error_handler);
-    }
-}
-
-/* Replication: Replica side.
- * Main channel read error handler */
-static void readReplBufferErrorHandler(connection *conn) {
-    serverLog(LL_WARNING, "Main channel error while reading from master: %s",
-              connGetLastError(conn));
-    cancelReplicationHandshake(1);
-}
-
-/* Replication: Replica side.
- * Read handler for buffering incoming repl data during RDB download/loading. */
-static void rdbChannelBufferReplData(connection *conn) {
-    replDataBuf *buf = &server.repl_full_sync_buffer;
-
-    if (server.repl_main_ch_state & REPL_MAIN_CH_STREAMING_BUF) {
-        /* While streaming accumulated buffers, we continue reading from the
-         * master to prevent accumulation on master side as much as possible.
-         * However, we aim to drain buffer eventually. To ensure we consume more
-         * than we read, we'll read at most one block after two blocks of
-         * buffers are consumed. */
-        if (listLength(buf->blocks) + 1 >= buf->last_num_blocks)
-            return;
-        buf->last_num_blocks = listLength(buf->blocks);
-    }
-
-    replDataBufReadFromConn(conn, buf, readReplBufferErrorHandler);
-}
-
-/* Generic function to stream replDataBuf data into database
- * Returns C_OK on success, C_ERR on error */
-int replDataBufStreamToDb(replDataBuf *buf, replDataBufToDbCtx *ctx) {
-    listNode *n;
-    int ret = C_OK;
-    client *c = ctx->client;
-
-    blockingOperationStarts();
-    while ((n = listFirst(buf->blocks))) {
-        replDataBufBlock *o = listNodeValue(n);
-        listUnlinkNode(buf->blocks, n);
-        zfree(n);
-
-        size_t processed = 0;
-        while (processed < o->used) {
-            size_t bytes = min(PROTO_IOBUF_LEN, o->used - processed);
-            c->querybuf = sdscatlen(c->querybuf, &o->buf[processed], bytes);
-            c->read_reploff += (long long int) bytes;
-            c->lastinteraction = server.unixtime;
-
-            /* We don't expect error return value but just in case. */
-            ret = processInputBuffer(c);
-            if (ret != C_OK) break;
-
-            processed += bytes;
-            buf->used -= bytes;
-
-            if (server.repl_debug_pause & REPL_DEBUG_ON_STREAMING_REPL_BUF)
-                debugPauseProcess();
-
-            /* Check if we should yield back to the event loop */
-            if (server.loading_process_events_interval_bytes &&
-                ((ctx->applied_offset + bytes) / server.loading_process_events_interval_bytes >
-                  ctx->applied_offset / server.loading_process_events_interval_bytes))
-            {
-                ctx->yield_callback(ctx);
-                processEventsWhileBlocked();
-            }
-            ctx->applied_offset += bytes;
-
-            /* Check if we should continue processing */
-            if (!ctx->should_continue(ctx)) {
-                ret = C_ERR;
-                break;
-            }
-
-            /* Streaming buffer into the database more slowly is useful in order
-             * to test certain edge cases. */
-            if (server.key_load_delay) debugDelay(server.key_load_delay);
-        }
-        size_t size = o->size;
-        zfree(o);
-
-        /* Break the loop if there is an error. */
-        if (ret != C_OK) break;
-
-        /* Update stats */
-        buf->size -= size;
-        buf->mem_used -= (size + sizeof(listNode) + sizeof(replDataBufBlock));
-    }
-    blockingOperationEnds();
-
-    return ret;
-}
-
-/* Replication: Replica side.
- * Yield callback for streaming replDataBuf to database */
-static void rdbChannelStreamYieldCallback(void *ctx) {
-    UNUSED(ctx);
-    replicationSendNewlineToMaster();
-}
-
-/* Replication: Replica side.
- * Global variable to track number of master disconnection.
- * Used to detect master disconnection when streaming replDataBuf to database */
-static uint64_t ReplNumMasterDisconnection = 0;
-
-/* Replication: Replica side.
- * Check if we should continue streaming replDataBuf to database */
-static int rdbChannelStreamShouldContinue(void *ctx) {
-    replDataBufToDbCtx *context = ctx;
-
-    /* Check if master client was freed in processEventsWhileBlocked().
-     * It can happen if we receive 'replicaof' command or 'client kill'
-     * command for the master. */
-    if (ReplNumMasterDisconnection != server.repl_num_master_disconnection ||
-        !server.repl_full_sync_buffer.blocks ||
-        context->client->flags & CLIENT_CLOSE_ASAP)
-    {
-        return 0;
-    }
-    return 1;
-}
-
-/* Replication: Replica side.
- * Streams accumulated replication data into the database. */
-static void rdbChannelStreamReplDataToDb(void) {
-    int ret = C_OK, close_asap = 0;
-    client *c = server.master;
-
-    /* Save repl_num_master_disconnection to figure out if master gets
-     * disconnected when we yield back to processEventsWhileBlocked() */
-    ReplNumMasterDisconnection = server.repl_num_master_disconnection;
-
-    server.repl_main_ch_state |= REPL_MAIN_CH_STREAMING_BUF;
-    serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Starting to stream replication buffer into the db"
-                         " (%zu bytes).", server.repl_full_sync_buffer.used);
-    if (!server.repl_full_sync_buffer.blocks)
-        goto out;
-
-    /* Mark the peek buffer block count. We'll use it to verify we consume
-     * faster than we read from the master. */
-    server.repl_full_sync_buffer.last_num_blocks = listLength(server.repl_full_sync_buffer.blocks);
-    /* Set read handler to continue accumulating during streaming */
-    connSetReadHandler(c->conn, rdbChannelBufferReplData);
-
-    replDataBufToDbCtx ctx = {
-        .client = c,
-        .applied_offset = 0,
-        .should_continue = rdbChannelStreamShouldContinue,
-        .yield_callback = rdbChannelStreamYieldCallback,
-    };
-
-    ret = replDataBufStreamToDb(&server.repl_full_sync_buffer, &ctx);
-
-out:
-    /* If main channel state is CLOSE_ASAP, it means main channel faced a
-     * problem while RDB is being loaded or while we are applying the
-     * accumulated buffer. It stopped replication stream buffering. It's okay
-     * though. We streamed whatever we have into the db, now we can free master
-     * client and replica can try psync. */
-    close_asap = (server.repl_main_ch_state & REPL_MAIN_CH_CLOSE_ASAP);
-
-    if (ret == C_OK) {
-        serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Successfully streamed replication buffer into the db (%zu bytes in total)",
-                             ctx.applied_offset);
-        /* Revert the read handler */
-        if (!close_asap && connSetReadHandler(c->conn, readQueryFromClient) != C_OK) {
-            serverLog(LL_WARNING,
-                      "Can't create readable event for master client: %s",
-                      strerror(errno));
-            close_asap = 1;
-        }
-    } else {
-        serverLog(LL_WARNING, "Master client was freed while streaming accumulated replication data to db.");
-        close_asap = 1;
-    }
-
-    /* If master is disconnected, state should have been cleaned up
-     * already. Otherwise, we do it here. */
-    if (ReplNumMasterDisconnection == server.repl_num_master_disconnection) {
-        rdbChannelCleanup();
-        if (server.master && close_asap)
-            freeClient(server.master);
-    }
-}
-
-static void rdbChannelCleanup(void) {
-    server.repl_rdb_ch_state = REPL_RDB_CH_STATE_NONE;
-    server.repl_main_ch_state = REPL_MAIN_CH_NONE;
-    rdbChannelReplDataBufClear();
-}
-
-/* Replication: Replica side.
- * On rdb channel failure, close rdb-connection and reset state.
- * Return C_OK if cleanup is done. Otherwise, returns C_ERR which means cleanup
- * will be done asynchronously. */
-static int rdbChannelAbort(void) {
-    if (server.repl_rdb_ch_state == REPL_RDB_CH_STATE_NONE)
-        return C_OK;
-
-    /* This function may also be called if a problem is detected on the main
-     * channel. In this case, we handle the situation differently based on
-     * the current state:
-     * - If we started loading the RDB file and the RDB is disk-based, we mark
-     *   the main channel's state as CLOSE_ASAP and defer the failure handling
-     *   until after the RDB has been loaded. This way we allow the replica to
-     *   retry psync after the RDB is loaded.
-     * - For diskless loading, we cannot safely free the rdb channel connection
-     *   object. Instead, we mark the RIO object as aborted so the next
-     *   rioRead() will fail safely.
-     * - If the RDB has already been loaded, and we are streaming the
-     *   accumulated buffer to the database, we mark the main connection
-     *   as CLOSE_ASAP and wait until the accumulated buffer is drained.
-     *   Once done, the replica can attempt psync with the offset it has. */
-    int async_cleanup = (server.repl_rdb_transfer_s && server.loading) ||
-                        (server.repl_main_ch_state & REPL_MAIN_CH_STREAMING_BUF);
-    if (async_cleanup) {
-        if (server.repl_rdb_transfer_s && server.loading) {
-            serverLog(LL_NOTICE, "Aborting rdb channel sync while loading the RDB.");
-
-            if (disklessLoadingRio)
-                /* Mark rio with abort flag, next rioRead() will return error.*/
-                rioAbort(disklessLoadingRio);
-            else {
-                /* For disk based loading, we can wait until loading is done.
-                 * This way, replica will have a chance for a successful psync
-                 * later.*/
-                serverLog(LL_NOTICE, "After loading RDB, replica will try psync with master.");
-            }
-        }
-
-        if (server.repl_transfer_s)
-            connSetReadHandler(server.repl_transfer_s, NULL);
-
-        server.repl_main_ch_state |= REPL_MAIN_CH_CLOSE_ASAP;
-        return C_ERR;
-    }
-
-    serverLog(LL_NOTICE, "Aborting rdb channel sync");
-
-    if (server.repl_rdb_transfer_s) {
-        connClose(server.repl_rdb_transfer_s);
-        server.repl_rdb_transfer_s = NULL;
-    }
-    if (server.repl_transfer_fd != -1) {
-        close(server.repl_transfer_fd);
-        server.repl_transfer_fd = -1;
-    }
-    if (server.repl_transfer_tmpfile) {
-        bg_unlink(server.repl_transfer_tmpfile);
-        zfree(server.repl_transfer_tmpfile);
-        server.repl_transfer_tmpfile = NULL;
-    }
-    rdbChannelCleanup();
-    return C_OK;
-}
-
-void replicaofCommand(client *c) {
-    /* SLAVEOF is not allowed in cluster mode as replication is automatically
-     * configured using the current address of the master node. */
-    if (server.cluster_enabled) {
-        addReplyError(c,"REPLICAOF not allowed in cluster mode.");
-        return;
-    }
-
-    if (server.failover_state != NO_FAILOVER) {
-        addReplyError(c,"REPLICAOF not allowed while failing over.");
-        return;
-    }
-
-    /* The special host/port combination "NO" "ONE" turns the instance
-     * into a master. Otherwise the new master address is set. */
-    if (!strcasecmp(c->argv[1]->ptr,"no") &&
-        !strcasecmp(c->argv[2]->ptr,"one")) {
-        if (server.masterhost) {
-            replicationUnsetMaster();
-            sds client = catClientInfoString(sdsempty(),c);
-            serverLog(LL_NOTICE,"MASTER MODE enabled (user request from '%s')",
-                client);
-            sdsfree(client);
-        }
-    } else {
-        long port;
-
-        if (c->flags & CLIENT_SLAVE)
-        {
-            /* If a client is already a replica they cannot run this command,
-             * because it involves flushing all replicas (including this
-             * client) */
-            addReplyError(c, "Command is not valid when client is a replica.");
-            return;
-        }
-
-        if (getRangeLongFromObjectOrReply(c, c->argv[2], 0, 65535, &port,
-                                          "Invalid master port") != C_OK)
-            return;
-
-        /* Check if we are already attached to the specified master */
-        if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr)
-            && server.masterport == port) {
-            serverLog(LL_NOTICE,"REPLICAOF would result into synchronization "
-                                "with the master we are already connected "
-                                "with. No operation performed.");
-            addReplySds(c,sdsnew("+OK Already connected to specified "
-                                 "master\r\n"));
-            return;
-        }
-        /* There was no previous master or the user specified a different one,
-         * we can continue. */
-        replicationSetMaster(c->argv[1]->ptr, port);
-        sds client = catClientInfoString(sdsempty(),c);
-        serverLog(LL_NOTICE,"REPLICAOF %s:%d enabled (user request from '%s')",
-            server.masterhost, server.masterport, client);
-        sdsfree(client);
-    }
-    addReply(c,shared.ok);
-}
-
-/* ROLE command: provide information about the role of the instance
- * (master or slave) and additional information related to replication
- * in an easy to process format. */
-void roleCommand(client *c) {
-    if (server.sentinel_mode) {
-        sentinelRoleCommand(c);
-        return;
-    }
-
-    if (server.masterhost == NULL) {
-        listIter li;
-        listNode *ln;
-        void *mbcount;
-        int slaves = 0;
-
-        addReplyArrayLen(c,3);
-        addReplyBulkCBuffer(c,"master",6);
-        addReplyLongLong(c,server.master_repl_offset);
-        mbcount = addReplyDeferredLen(c);
-        listRewind(server.slaves,&li);
-        while((ln = listNext(&li))) {
-            client *slave = ln->value;
-            char ip[NET_IP_STR_LEN], *slaveaddr = slave->slave_addr;
-
-            if (!slaveaddr) {
-                if (connAddrPeerName(slave->conn,ip,sizeof(ip),NULL) == -1)
-                    continue;
-                slaveaddr = ip;
-            }
-            if (slave->replstate != SLAVE_STATE_ONLINE) continue;
-            addReplyArrayLen(c,3);
-            addReplyBulkCString(c,slaveaddr);
-            addReplyBulkLongLong(c,slave->slave_listening_port);
-            addReplyBulkLongLong(c,slave->repl_ack_off);
-            slaves++;
-        }
-        setDeferredArrayLen(c,mbcount,slaves);
-    } else {
-        char *slavestate = NULL;
-
-        addReplyArrayLen(c,5);
-        addReplyBulkCBuffer(c,"slave",5);
-        addReplyBulkCString(c,server.masterhost);
-        addReplyLongLong(c,server.masterport);
-        if (slaveIsInHandshakeState()) {
-            slavestate = "handshake";
-        } else {
-            switch(server.repl_state) {
-            case REPL_STATE_NONE: slavestate = "none"; break;
-            case REPL_STATE_CONNECT: slavestate = "connect"; break;
-            case REPL_STATE_CONNECTING: slavestate = "connecting"; break;
-            case REPL_STATE_TRANSFER: slavestate = "sync"; break;
-            case REPL_STATE_CONNECTED: slavestate = "connected"; break;
-            default: slavestate = "unknown"; break;
-            }
-        }
-        addReplyBulkCString(c,slavestate);
-        addReplyLongLong(c,server.master ? server.master->reploff : -1);
-    }
-}
-
-/* Send a REPLCONF ACK command to the master to inform it about the current
- * processed offset. If we are not connected with a master, the command has
- * no effects. */
-void replicationSendAck(void) {
-    client *c = server.master;
-
-    if (c != NULL) {
-        int send_fack = server.fsynced_reploff != -1;
-        c->flags |= CLIENT_MASTER_FORCE_REPLY;
-        addReplyArrayLen(c,send_fack ? 5 : 3);
-        addReplyBulkCString(c,"REPLCONF");
-        addReplyBulkCString(c,"ACK");
-        addReplyBulkLongLong(c,c->reploff);
-        if (send_fack) {
-            addReplyBulkCString(c,"FACK");
-            addReplyBulkLongLong(c,server.fsynced_reploff);
-        }
-        c->flags &= ~CLIENT_MASTER_FORCE_REPLY;
-        /* Accumulation from above replies must be reset back to 0 manually,
-         * as this subroutine does not invoke resetClient(). */
-        c->net_output_bytes_curr_cmd = 0;
-    }
-}
-
-/* ---------------------- MASTER CACHING FOR PSYNC -------------------------- */
-
-/* In order to implement partial synchronization we need to be able to cache
- * our master's client structure after a transient disconnection.
- * It is cached into server.cached_master and flushed away using the following
- * functions. */
-
-/* This function is called by freeClient() in order to cache the master
- * client structure instead of destroying it. freeClient() will return
- * ASAP after this function returns, so every action needed to avoid problems
- * with a client that is really "suspended" has to be done by this function.
- *
- * The other functions that will deal with the cached master are:
- *
- * replicationDiscardCachedMaster() that will make sure to kill the client
- * as for some reason we don't want to use it in the future.
- *
- * replicationResurrectCachedMaster() that is used after a successful PSYNC
- * handshake in order to reactivate the cached master.
- */
-void replicationCacheMaster(client *c) {
-    serverAssert(server.master != NULL && server.cached_master == NULL);
-    serverAssert(server.master->tid == IOTHREAD_MAIN_THREAD_ID);
-    serverLog(LL_NOTICE,"Caching the disconnected master state.");
-
-    /* Unlink the client from the server structures. */
-    unlinkClient(c);
-
-    /* Reset the master client so that's ready to accept new commands:
-     * we want to discard the non processed query buffers and non processed
-     * offsets, including pending transactions, already populated arguments,
-     * pending outputs to the master. */
-    sdsclear(server.master->querybuf);
-    server.master->qb_pos = 0;
-    server.master->repl_applied = 0;
-    server.master->read_reploff = server.master->reploff;
-    server.master->reploff_next = 0;
-    if (c->flags & CLIENT_MULTI) discardTransaction(c);
-    listEmpty(c->reply);
-    c->sentlen = 0;
-    c->reply_bytes = 0;
-    c->bufpos = 0;
-    resetClient(c, -1);
-    resetClientQbufState(c);
-
-    /* Save the master. Server.master will be set to null later by
-     * replicationHandleMasterDisconnection(). */
-    server.cached_master = server.master;
-
-    /* Invalidate the Peer ID cache. */
-    if (c->peerid) {
-        sdsfree(c->peerid);
-        c->peerid = NULL;
-    }
-    /* Invalidate the Sock Name cache. */
-    if (c->sockname) {
-        sdsfree(c->sockname);
-        c->sockname = NULL;
-    }
-
-    /* Caching the master happens instead of the actual freeClient() call,
-     * so make sure to adjust the replication state. This function will
-     * also set server.master to NULL. */
-    replicationHandleMasterDisconnection();
-}
-
-/* This function is called when a master is turned into a slave, in order to
- * create from scratch a cached master for the new client, that will allow
- * to PSYNC with the slave that was promoted as the new master after a
- * failover.
- *
- * Assuming this instance was previously the master instance of the new master,
- * the new master will accept its replication ID, and potential also the
- * current offset if no data was lost during the failover. So we use our
- * current replication ID and offset in order to synthesize a cached master. */
-void replicationCacheMasterUsingMyself(void) {
-    serverLog(LL_NOTICE,
-        "Before turning into a replica, using my own master parameters "
-        "to synthesize a cached master: I may be able to synchronize with "
-        "the new master with just a partial transfer.");
-
-    /* This will be used to populate the field server.master->reploff
-     * by replicationCreateMasterClient(). We'll later set the created
-     * master as server.cached_master, so the replica will use such
-     * offset for PSYNC. */
-    server.master_initial_offset = server.master_repl_offset;
-
-    /* The master client we create can be set to any DBID, because
-     * the new master will start its replication stream with SELECT. */
-    replicationCreateMasterClient(NULL,-1);
-
-    /* Use our own ID / offset. */
-    memcpy(server.master->replid, server.replid, sizeof(server.replid));
-
-    /* Set as cached master. */
-    unlinkClient(server.master);
-    server.cached_master = server.master;
-    server.master = NULL;
-}
-
-/* Free a cached master, called when there are no longer the conditions for
- * a partial resync on reconnection. */
-void replicationDiscardCachedMaster(void) {
-    if (server.cached_master == NULL) return;
-
-    serverLog(LL_NOTICE,"Discarding previously cached master state.");
-    server.cached_master->flags &= ~CLIENT_MASTER;
-    freeClient(server.cached_master);
-    server.cached_master = NULL;
-}
-
-/* Turn the cached master into the current master, using the file descriptor
- * passed as argument as the socket for the new master.
- *
- * This function is called when successfully setup a partial resynchronization
- * so the stream of data that we'll receive will start from where this
- * master left. */
-void replicationResurrectCachedMaster(connection *conn) {
-    serverAssert(server.cached_master->tid == IOTHREAD_MAIN_THREAD_ID);
-
-    server.master = server.cached_master;
-    server.cached_master = NULL;
-    server.master->conn = conn;
-    connSetPrivateData(server.master->conn, server.master);
-    server.master->flags &= ~(CLIENT_CLOSE_AFTER_REPLY|CLIENT_CLOSE_ASAP);
-    server.master->authenticated = 1;
-    server.master->lastinteraction = server.unixtime;
-    server.repl_state = REPL_STATE_CONNECTED;
-    server.repl_down_since = 0;
-    server.repl_up_since = server.unixtime;
-    if (server.repl_disconnect_start_time != 0) {
-        server.repl_total_disconnect_time += server.unixtime - server.repl_disconnect_start_time;
-        server.repl_disconnect_start_time = 0;
-    }
-    /* Fire the master link modules event. */
-    moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
-                          REDISMODULE_SUBEVENT_MASTER_LINK_UP,
-                          NULL);
-
-    /* Re-add to the list of clients. */
-    linkClient(server.master);
-    if (connSetReadHandler(server.master->conn, readQueryFromClient)) {
-        serverLog(LL_WARNING,"Error resurrecting the cached master, impossible to add the readable handler: %s", strerror(errno));
-        freeClientAsync(server.master); /* Close ASAP. */
-    }
-
-    /* We may also need to install the write handler as well if there is
-     * pending data in the write buffers. */
-    if (clientHasPendingReplies(server.master)) {
-        if (connSetWriteHandler(server.master->conn, sendReplyToClient)) {
-            serverLog(LL_WARNING,"Error resurrecting the cached master, impossible to add the writable handler: %s", strerror(errno));
-            freeClientAsync(server.master); /* Close ASAP. */
-        }
-    }
-}
-
-/* ------------------------- MIN-SLAVES-TO-WRITE  --------------------------- */
-
-/* This function counts the number of slaves with lag <= min-slaves-max-lag.
- * If the option is active, the server will prevent writes if there are not
- * enough connected slaves with the specified lag (or less). */
-void refreshGoodSlavesCount(void) {
-    listIter li;
-    listNode *ln;
-    int good = 0;
-
-    if (!server.repl_min_slaves_to_write ||
-        !server.repl_min_slaves_max_lag) return;
-
-    listRewind(server.slaves,&li);
-    while((ln = listNext(&li))) {
-        client *slave = ln->value;
-        time_t lag = server.unixtime - slave->repl_ack_time;
-
-        if (slave->replstate == SLAVE_STATE_ONLINE &&
-            lag <= server.repl_min_slaves_max_lag) good++;
-    }
-    server.repl_good_slaves_count = good;
-}
-
-/* return true if status of good replicas is OK. otherwise false */
-int checkGoodReplicasStatus(void) {
-    return server.masterhost || /* not a primary status should be OK */
-           !server.repl_min_slaves_max_lag || /* Min slave max lag not configured */
-           !server.repl_min_slaves_to_write || /* Min slave to write not configured */
-           server.repl_good_slaves_count >= server.repl_min_slaves_to_write; /* check if we have enough slaves */
-}
-
-/* ----------------------- SYNCHRONOUS REPLICATION --------------------------
- * Redis synchronous replication design can be summarized in points:
- *
- * - Redis masters have a global replication offset, used by PSYNC.
- * - Master increment the offset every time new commands are sent to slaves.
- * - Slaves ping back masters with the offset processed so far.
- *
- * So synchronous replication adds a new WAIT command in the form:
- *
- *   WAIT <num_replicas> <milliseconds_timeout>
- *
- * That returns the number of replicas that processed the query when
- * we finally have at least num_replicas, or when the timeout was
- * reached.
- *
- * The command is implemented in this way:
- *
- * - Every time a client processes a command, we remember the replication
- *   offset after sending that command to the slaves.
- * - When WAIT is called, we ask slaves to send an acknowledgement ASAP.
- *   The client is blocked at the same time (see blocked.c).
- * - Once we receive enough ACKs for a given offset or when the timeout
- *   is reached, the WAIT command is unblocked and the reply sent to the
- *   client.
- */
-
-/* This just set a flag so that we broadcast a REPLCONF GETACK command
- * to all the slaves in the beforeSleep() function. Note that this way
- * we "group" all the clients that want to wait for synchronous replication
- * in a given event loop iteration, and send a single GETACK for them all. */
-void replicationRequestAckFromSlaves(void) {
-    server.get_ack_from_slaves = 1;
-}
-
-/* Return the number of slaves that already acknowledged the specified
- * replication offset. */
-int replicationCountAcksByOffset(long long offset) {
-    listIter li;
-    listNode *ln;
-    int count = 0;
-
-    listRewind(server.slaves,&li);
-    while((ln = listNext(&li))) {
-        client *slave = ln->value;
-
-        if (slave->replstate != SLAVE_STATE_ONLINE) continue;
-        if (slave->repl_ack_off >= offset) count++;
-    }
-    return count;
-}
-
-/* Return the number of replicas that already acknowledged the specified
- * replication offset being AOF fsynced. */
-int replicationCountAOFAcksByOffset(long long offset) {
-    listIter li;
-    listNode *ln;
-    int count = 0;
-
-    listRewind(server.slaves,&li);
-    while((ln = listNext(&li))) {
-        client *slave = ln->value;
-
-        if (slave->replstate != SLAVE_STATE_ONLINE) continue;
-        if (slave->repl_aof_off >= offset) count++;
-    }
-    return count;
-}
-
-/* WAIT for N replicas to acknowledge the processing of our latest
- * write command (and all the previous commands). */
-void waitCommand(client *c) {
-    mstime_t timeout;
-    long numreplicas, ackreplicas;
-    long long offset = c->woff;
-
-    if (server.masterhost) {
-        addReplyError(c,"WAIT cannot be used with replica instances. Please also note that since Redis 4.0 if a replica is configured to be writable (which is not the default) writes to replicas are just local and are not propagated.");
-        return;
-    }
-
-    /* Argument parsing. */
-    if (getLongFromObjectOrReply(c,c->argv[1],&numreplicas,NULL) != C_OK)
-        return;
-    if (getTimeoutFromObjectOrReply(c,c->argv[2],&timeout,UNIT_MILLISECONDS)
-        != C_OK) return;
-
-    /* First try without blocking at all. */
-    ackreplicas = replicationCountAcksByOffset(c->woff);
-    if (ackreplicas >= numreplicas || c->flags & CLIENT_DENY_BLOCKING) {
-        addReplyLongLong(c,ackreplicas);
-        return;
-    }
-
-    /* Otherwise block the client and put it into our list of clients
-     * waiting for ack from slaves. */
-    blockForReplication(c,timeout,offset,numreplicas);
-
-    /* Make sure that the server will send an ACK request to all the slaves
-     * before returning to the event loop. */
-    replicationRequestAckFromSlaves();
-}
-
-/* WAIT for N replicas and / or local master to acknowledge our latest
- * write command got synced to the disk. */
-void waitaofCommand(client *c) {
-    mstime_t timeout;
-    long numreplicas, numlocal, ackreplicas, acklocal;
-
-    /* Argument parsing. */
-    if (getRangeLongFromObjectOrReply(c,c->argv[1],0,1,&numlocal,NULL) != C_OK)
-        return;
-    if (getPositiveLongFromObjectOrReply(c,c->argv[2],&numreplicas,NULL) != C_OK)
-        return;
-    if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout,UNIT_MILLISECONDS) != C_OK)
-        return;
-
-    if (server.masterhost) {
-        addReplyError(c,"WAITAOF cannot be used with replica instances. Please also note that writes to replicas are just local and are not propagated.");
-        return;
-    }
-    if (numlocal && !server.aof_enabled) {
-        addReplyError(c, "WAITAOF cannot be used when numlocal is set but appendonly is disabled.");
-        return;
-    }
-
-    /* First try without blocking at all. */
-    ackreplicas = replicationCountAOFAcksByOffset(c->woff);
-    acklocal = server.fsynced_reploff >= c->woff;
-    if ((ackreplicas >= numreplicas && acklocal >= numlocal) || c->flags & CLIENT_DENY_BLOCKING) {
-        addReplyArrayLen(c,2);
-        addReplyLongLong(c,acklocal);
-        addReplyLongLong(c,ackreplicas);
-        return;
-    }
-
-    /* Otherwise block the client and put it into our list of clients
-     * waiting for ack from slaves. */
-    blockForAofFsync(c,timeout,c->woff,numlocal,numreplicas);
-
-    /* Make sure that the server will send an ACK request to all the slaves
-     * before returning to the event loop. */
-    replicationRequestAckFromSlaves();
-}
-
-/* This is called by unblockClient() to perform the blocking op type
- * specific cleanup. We just remove the client from the list of clients
- * waiting for replica acks. Never call it directly, call unblockClient()
- * instead. */
-void unblockClientWaitingReplicas(client *c) {
-    listNode *ln = listSearchKey(server.clients_waiting_acks,c);
-    serverAssert(ln != NULL);
-    listDelNode(server.clients_waiting_acks,ln);
-    updateStatsOnUnblock(c, 0, 0, 0);
-}
-
-/* Check if there are clients blocked in WAIT or WAITAOF that can be unblocked
- * since we received enough ACKs from slaves. */
-void processClientsWaitingReplicas(void) {
-    long long last_offset = 0;
-    long long last_aof_offset = 0;
-    int last_numreplicas = 0;
-    int last_aof_numreplicas = 0;
-
-    listIter li;
-    listNode *ln;
-
-    listRewind(server.clients_waiting_acks,&li);
-    while((ln = listNext(&li))) {
-        int numlocal = 0;
-        int numreplicas = 0;
-
-        client *c = ln->value;
-        int is_wait_aof = c->bstate.btype == BLOCKED_WAITAOF;
-
-        if (is_wait_aof && c->bstate.numlocal && !server.aof_enabled) {
-            addReplyError(c, "WAITAOF cannot be used when numlocal is set but appendonly is disabled.");
-            unblockClient(c, 1);
-            continue;
-        }
-
-        /* Every time we find a client that is satisfied for a given
-         * offset and number of replicas, we remember it so the next client
-         * may be unblocked without calling replicationCountAcksByOffset()
-         * or calling replicationCountAOFAcksByOffset()
-         * if the requested offset / replicas were equal or less. */
-        if (!is_wait_aof && last_offset && last_offset >= c->bstate.reploffset &&
-                           last_numreplicas >= c->bstate.numreplicas)
-        {
-            numreplicas = last_numreplicas;
-        } else if (is_wait_aof && last_aof_offset && last_aof_offset >= c->bstate.reploffset &&
-                    last_aof_numreplicas >= c->bstate.numreplicas)
-        {
-            numreplicas = last_aof_numreplicas;
-        } else {
-            numreplicas = is_wait_aof ?
-                replicationCountAOFAcksByOffset(c->bstate.reploffset) :
-                replicationCountAcksByOffset(c->bstate.reploffset);
-
-            /* Check if the number of replicas is satisfied. */
-            if (numreplicas < c->bstate.numreplicas) continue;
-
-            if (is_wait_aof) {
-                last_aof_offset = c->bstate.reploffset;
-                last_aof_numreplicas = numreplicas;
-            } else {
-                last_offset = c->bstate.reploffset;
-                last_numreplicas = numreplicas;
-            }
-        }
-
-        /* Check if the local constraint of WAITAOF is served */
-        if (is_wait_aof) {
-            numlocal = server.fsynced_reploff >= c->bstate.reploffset;
-            if (numlocal < c->bstate.numlocal) continue;
-        }
-
-        /* Reply before unblocking, because unblock client calls reqresAppendResponse */
-        if (is_wait_aof) {
-            /* WAITAOF has an array reply */
-            addReplyArrayLen(c, 2);
-            addReplyLongLong(c, numlocal);
-            addReplyLongLong(c, numreplicas);
-        } else {
-            addReplyLongLong(c, numreplicas);
-        }
-
-        unblockClient(c, 1);
-    }
-}
-
-/* Return the slave replication offset for this instance, that is
- * the offset for which we already processed the master replication stream. */
-long long replicationGetSlaveOffset(void) {
-    long long offset = 0;
-
-    if (server.masterhost != NULL) {
-        if (server.master) {
-            offset = server.master->reploff;
-        } else if (server.cached_master) {
-            offset = server.cached_master->reploff;
-        }
-    }
-    /* offset may be -1 when the master does not support it at all, however
-     * this function is designed to return an offset that can express the
-     * amount of data processed by the master, so we return a positive
-     * integer. */
-    if (offset < 0) offset = 0;
-    return offset;
-}
-
-/* --------------------------- REPLICATION CRON  ---------------------------- */
-
-/* Replication cron function, called 1 time per second. */
-void replicationCron(void) {
-    /* Check failover status first, to see if we need to start
-     * handling the failover. */
-    updateFailoverStatus();
-
-    /* Non blocking connection timeout? */
-    if (server.masterhost &&
-        (server.repl_state == REPL_STATE_CONNECTING ||
-         slaveIsInHandshakeState()) &&
-         (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
-    {
-        serverLog(LL_WARNING,"Timeout connecting to the MASTER...");
-        cancelReplicationHandshake(1);
-    }
-
-    /* Bulk transfer I/O timeout? */
-    if (server.masterhost && server.repl_state == REPL_STATE_TRANSFER &&
-        (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
-    {
-        serverLog(LL_WARNING,"Timeout receiving bulk data from MASTER... If the problem persists try to set the 'repl-timeout' parameter in redis.conf to a larger value.");
-        cancelReplicationHandshake(1);
-    }
-
-    /* Check if we should connect to a MASTER */
-    if (server.repl_state == REPL_STATE_CONNECT) {
-        serverLog(LL_NOTICE,"Connecting to MASTER %s:%d",
-            server.masterhost, server.masterport);
-        connectWithMaster();
-    }
-
-    replicationCronRunMasterClient();
-
-    /* If we have attached slaves, PING them from time to time.
-     * So slaves can implement an explicit timeout to masters, and will
-     * be able to detect a link disconnection even if the TCP connection
-     * will not actually go down. */
-    listIter li;
-    listNode *ln;
-    robj *ping_argv[1];
-
-    /* First, send PING according to ping_slave_period. The reason why master
-     * sends PING is to keep the connection with replica active, so master need
-     * not send PING to replicas if already sent replication stream in the past
-     * repl_ping_slave_period time. */
-    if (server.masterhost == NULL && listLength(server.slaves) &&
-        server.unixtime >= server.repl_stream_lastio + server.repl_ping_slave_period)
-    {
-        /* Note that we don't send the PING if the clients are paused during
-         * a Redis Cluster manual failover: the PING we send will otherwise
-         * alter the replication offsets of master and slave, and will no longer
-         * match the one stored into 'mf_master_offset' state. */
-        int manual_failover_in_progress =
-            ((server.cluster_enabled &&
-              clusterManualFailoverTimeLimit()) ||
-            server.failover_end_time) &&
-            isPausedActionsWithUpdate(PAUSE_ACTION_REPLICA);
-
-        if (!manual_failover_in_progress) {
-            ping_argv[0] = shared.ping;
-            replicationFeedSlaves(server.slaves, -1,
-                ping_argv, 1);
-        }
-    }
-
-    /* Second, send a newline to all the slaves in pre-synchronization
-     * stage, that is, slaves waiting for the master to create the RDB file.
-     *
-     * Also send the a newline to all the chained slaves we have, if we lost
-     * connection from our master, to keep the slaves aware that their
-     * master is online. This is needed since sub-slaves only receive proxied
-     * data from top-level masters, so there is no explicit pinging in order
-     * to avoid altering the replication offsets. This special out of band
-     * pings (newlines) can be sent, they will have no effect in the offset.
-     *
-     * The newline will be ignored by the slave but will refresh the
-     * last interaction timer preventing a timeout. In this case we ignore the
-     * ping period and refresh the connection once per second since certain
-     * timeouts are set at a few seconds (example: PSYNC response). */
-    listRewind(server.slaves,&li);
-    while((ln = listNext(&li))) {
-        client *slave = ln->value;
-
-        int is_presync =
-            (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START ||
-            (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END &&
-             server.rdb_child_type != RDB_CHILD_TYPE_SOCKET));
-
-        if (is_presync && !(slave->flags & CLIENT_CLOSE_ASAP)) {
-            connWrite(slave->conn, "\n", 1);
-        }
-    }
-
-    /* Disconnect timedout slaves. */
-    if (listLength(server.slaves)) {
-        listIter li;
-        listNode *ln;
-
-        listRewind(server.slaves,&li);
-        while((ln = listNext(&li))) {
-            client *slave = ln->value;
-
-            if (slave->replstate == SLAVE_STATE_ONLINE) {
-                if (slave->flags & CLIENT_PRE_PSYNC)
-                    continue;
-                if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout) {
-                    serverLog(LL_WARNING, "Disconnecting timedout replica (streaming sync): %s",
-                          replicationGetSlaveName(slave));
-                    freeClient(slave);
-                    continue;
-                }
-            }
-            /* We consider disconnecting only diskless replicas because disk-based replicas aren't fed
-             * by the fork child so if a disk-based replica is stuck it doesn't prevent the fork child
-             * from terminating. */
-            if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END && server.rdb_child_type == RDB_CHILD_TYPE_SOCKET) {
-                if (slave->repl_last_partial_write != 0 &&
-                    (server.unixtime - slave->repl_last_partial_write) > server.repl_timeout)
-                {
-                    serverLog(LL_WARNING, "Disconnecting timedout replica (full sync): %s",
-                          replicationGetSlaveName(slave));
-                    freeClient(slave);
-                    continue;
-                }
-            }
-        }
-    }
-
-    /* If this is a master without attached slaves and there is a replication
-     * backlog active, in order to reclaim memory we can free it after some
-     * (configured) time. Note that this cannot be done for slaves: slaves
-     * without sub-slaves attached should still accumulate data into the
-     * backlog, in order to reply to PSYNC queries if they are turned into
-     * masters after a failover. */
-    if (listLength(server.slaves) == 0 && server.repl_backlog_time_limit &&
-        server.repl_backlog && server.masterhost == NULL)
-    {
-        time_t idle = server.unixtime - server.repl_no_slaves_since;
-
-        if (idle > server.repl_backlog_time_limit) {
-            /* When we free the backlog, we always use a new
-             * replication ID and clear the ID2. This is needed
-             * because when there is no backlog, the master_repl_offset
-             * is not updated, but we would still retain our replication
-             * ID, leading to the following problem:
-             *
-             * 1. We are a master instance.
-             * 2. Our slave is promoted to master. It's repl-id-2 will
-             *    be the same as our repl-id.
-             * 3. We, yet as master, receive some updates, that will not
-             *    increment the master_repl_offset.
-             * 4. Later we are turned into a slave, connect to the new
-             *    master that will accept our PSYNC request by second
-             *    replication ID, but there will be data inconsistency
-             *    because we received writes. */
-            changeReplicationId();
-            clearReplicationId2();
-            freeReplicationBacklog();
-            serverLog(LL_NOTICE,
-                "Replication backlog freed after %d seconds "
-                "without connected replicas.",
-                (int) server.repl_backlog_time_limit);
-        }
-    }
-
-    replicationStartPendingFork();
-
-    /* Remove the RDB file used for replication if Redis is not running
-     * with any persistence. */
-    removeRDBUsedToSyncReplicas();
-
-    /* Sanity check replication buffer, the first block of replication buffer blocks
-     * must be referenced by someone, since it will be freed when not referenced,
-     * otherwise, server will OOM. also, its refcount must not be more than
-     * replicas number + 1(replication backlog). */
-    if (listLength(server.repl_buffer_blocks) > 0) {
-        replBufBlock *o = listNodeValue(listFirst(server.repl_buffer_blocks));
-        serverAssert(o->refcount > 0 &&
-            o->refcount <= (int)listLength(server.slaves)+1);
-    }
-
-    /* Refresh the number of slaves with lag <= min-slaves-max-lag. */
-    refreshGoodSlavesCount();
-}
-
-int shouldStartChildReplication(int *mincapa_out, int *req_out) {
-    /* We should start a BGSAVE good for replication if we have slaves in
-     * WAIT_BGSAVE_START state.
-     *
-     * In case of diskless replication, we make sure to wait the specified
-     * number of seconds (according to configuration) so that other slaves
-     * have the time to arrive before we start streaming. */
-    if (!hasActiveChildProcess()) {
-        time_t idle, max_idle = 0;
-        int slaves_waiting = 0;
-        int mincapa;
-        int req;
-        int first = 1;
-        listNode *ln;
-        listIter li;
-
-        listRewind(server.slaves,&li);
-        while((ln = listNext(&li))) {
-            client *slave = ln->value;
-            if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
-                if (first) {
-                    /* Get first slave's requirements */
-                    req = slave->slave_req;
-                } else if (req != slave->slave_req) {
-                    /* Skip slaves that don't match */
-                    continue;
-                }
-                idle = server.unixtime - slave->lastinteraction;
-                /* If the slave requests a slots snapshot, we should start BGSAVE
-                 * immediately since it can't share the RDB with other slaves. */
-                if (slave->slave_req & SLAVE_REQ_SLOTS_SNAPSHOT)
-                    idle = server.repl_diskless_sync_delay; /* Threshold for BGSAVE */
-                if (idle > max_idle) max_idle = idle;
-                slaves_waiting++;
-                mincapa = first ? slave->slave_capa : (mincapa & slave->slave_capa);
-                first = 0;
-            }
-        }
-
-        if (slaves_waiting &&
-            (!server.repl_diskless_sync ||
-             (server.repl_diskless_sync_max_replicas > 0 &&
-              slaves_waiting >= server.repl_diskless_sync_max_replicas) ||
-             max_idle >= server.repl_diskless_sync_delay))
-        {
-            if (mincapa_out)
-                *mincapa_out = mincapa;
-            if (req_out)
-                *req_out = req;
-            return 1;
-        }
-    }
-
-    return 0;
-}
-
-void replicationStartPendingFork(void) {
-    int mincapa = -1;
-    int req = -1;
-
-    if (shouldStartChildReplication(&mincapa, &req)) {
-        /* Start the BGSAVE. The called function may start a
-         * BGSAVE with socket target or disk target depending on the
-         * configuration and slaves capabilities and requirements. */
-        startBgsaveForReplication(mincapa, req);
-    }
-}
-
-/* Find replica at IP:PORT from replica list */
-static client *findReplica(char *host, int port) {
-    listIter li;
-    listNode *ln;
-    client *replica;
-
-    listRewind(server.slaves,&li);
-    while((ln = listNext(&li))) {
-        replica = ln->value;
-        char ip[NET_IP_STR_LEN], *replicaip = replica->slave_addr;
-
-        if (!replicaip) {
-            if (connAddrPeerName(replica->conn, ip, sizeof(ip), NULL) == -1)
-                continue;
-            replicaip = ip;
-        }
-
-        if (!strcasecmp(host, replicaip) &&
-                (port == replica->slave_listening_port))
-            return replica;
-    }
-
-    return NULL;
-}
-
-const char *getFailoverStateString(void) {
-    switch(server.failover_state) {
-        case NO_FAILOVER: return "no-failover";
-        case FAILOVER_IN_PROGRESS: return "failover-in-progress";
-        case FAILOVER_WAIT_FOR_SYNC: return "waiting-for-sync";
-        default: return "unknown";
-    }
-}
-
-/* Resets the internal failover configuration, this needs
- * to be called after a failover either succeeds or fails
- * as it includes the client unpause. */
-void clearFailoverState(void) {
-    server.failover_end_time = 0;
-    server.force_failover = 0;
-    zfree(server.target_replica_host);
-    server.target_replica_host = NULL;
-    server.target_replica_port = 0;
-    server.failover_state = NO_FAILOVER;
-    unpauseActions(PAUSE_DURING_FAILOVER);
-}
-
-/* Abort an ongoing failover if one is going on. */
-void abortFailover(const char *err) {
-    if (server.failover_state == NO_FAILOVER) return;
-
-    if (server.target_replica_host) {
-        serverLog(LL_NOTICE,"FAILOVER to %s:%d aborted: %s",
-            server.target_replica_host,server.target_replica_port,err);  
-    } else {
-        serverLog(LL_NOTICE,"FAILOVER to any replica aborted: %s",err);  
-    }
-    if (server.failover_state == FAILOVER_IN_PROGRESS) {
-        replicationUnsetMaster();
-    }
-    clearFailoverState();
-}
-
-/* 
- * FAILOVER [TO <HOST> <PORT> [FORCE]] [ABORT] [TIMEOUT <timeout>]
- * 
- * This command will coordinate a failover between the master and one
- * of its replicas. The happy path contains the following steps:
- * 1) The master will initiate a client pause write, to stop replication
- * traffic.
- * 2) The master will periodically check if any of its replicas has
- * consumed the entire replication stream through acks. 
- * 3) Once any replica has caught up, the master will itself become a replica.
- * 4) The master will send a PSYNC FAILOVER request to the target replica, which
- * if accepted will cause the replica to become the new master and start a sync.
- * 
- * FAILOVER ABORT is the only way to abort a failover command, as replicaof
- * will be disabled. This may be needed if the failover is unable to progress. 
- * 
- * The optional arguments [TO <HOST> <IP>] allows designating a specific replica
- * to be failed over to.
- * 
- * FORCE flag indicates that even if the target replica is not caught up,
- * failover to it anyway. This must be specified with a timeout and a target
- * HOST and IP.
- * 
- * TIMEOUT <timeout> indicates how long should the primary wait for 
- * a replica to sync up before aborting. If not specified, the failover
- * will attempt forever and must be manually aborted.
- */
-void failoverCommand(client *c) {
-    if (!clusterAllowFailoverCmd(c)) {
-        return;
-    }
-
-    /* Handle special case for abort */
-    if ((c->argc == 2) && !strcasecmp(c->argv[1]->ptr,"abort")) {
-        if (server.failover_state == NO_FAILOVER) {
-            addReplyError(c, "No failover in progress.");
-            return;
-        }
-
-        abortFailover("Failover manually aborted");
-        addReply(c,shared.ok);
-        return;
-    }
-
-    long timeout_in_ms = 0;
-    int force_flag = 0;
-    long port = 0;
-    char *host = NULL;
-
-    /* Parse the command for syntax and arguments. */
-    for (int j = 1; j < c->argc; j++) {
-        if (!strcasecmp(c->argv[j]->ptr,"timeout") && (j + 1 < c->argc) &&
-            timeout_in_ms == 0)
-        {
-            if (getLongFromObjectOrReply(c,c->argv[j + 1],
-                        &timeout_in_ms,NULL) != C_OK) return;
-            if (timeout_in_ms <= 0) {
-                addReplyError(c,"FAILOVER timeout must be greater than 0");
-                return;
-            }
-            j++;
-        } else if (!strcasecmp(c->argv[j]->ptr,"to") && (j + 2 < c->argc) &&
-            !host) 
-        {
-            if (getLongFromObjectOrReply(c,c->argv[j + 2],&port,NULL) != C_OK)
-                return;
-            host = c->argv[j + 1]->ptr;
-            j += 2;
-        } else if (!strcasecmp(c->argv[j]->ptr,"force") && !force_flag) {
-            force_flag = 1;
-        } else {
-            addReplyErrorObject(c,shared.syntaxerr);
-            return;
-        }
-    }
-
-    if (server.failover_state != NO_FAILOVER) {
-        addReplyError(c,"FAILOVER already in progress.");
-        return;
-    }
-
-    if (server.masterhost) {
-        addReplyError(c,"FAILOVER is not valid when server is a replica.");
-        return;
-    }
-
-    if (listLength(server.slaves) == 0) {
-        addReplyError(c,"FAILOVER requires connected replicas.");
-        return; 
-    }
-
-    if (force_flag && (!timeout_in_ms || !host)) {
-        addReplyError(c,"FAILOVER with force option requires both a timeout "
-            "and target HOST and IP.");
-        return;     
-    }
-
-    /* If a replica address was provided, validate that it is connected. */
-    if (host) {
-        client *replica = findReplica(host, port);
-
-        if (replica == NULL) {
-            addReplyError(c,"FAILOVER target HOST and PORT is not "
-                            "a replica.");
-            return;
-        }
-
-        /* Check if requested replica is online */
-        if (replica->replstate != SLAVE_STATE_ONLINE) {
-            addReplyError(c,"FAILOVER target replica is not online.");
-            return;
-        }
-
-        server.target_replica_host = zstrdup(host);
-        server.target_replica_port = port;
-        serverLog(LL_NOTICE,"FAILOVER requested to %s:%ld.",host,port);
-    } else {
-        serverLog(LL_NOTICE,"FAILOVER requested to any replica.");
-    }
-
-    mstime_t now = commandTimeSnapshot();
-    if (timeout_in_ms) {
-        server.failover_end_time = now + timeout_in_ms;
-    }
-    
-    server.force_failover = force_flag;
-    server.failover_state = FAILOVER_WAIT_FOR_SYNC;
-    /* Cancel all ASM tasks when starting failover */
-    clusterAsmCancel(NULL, "failover requested");
-    /* Cluster failover will unpause eventually */
-    pauseActions(PAUSE_DURING_FAILOVER,
-                 LLONG_MAX,
-                 PAUSE_ACTIONS_CLIENT_WRITE_SET);
-    addReply(c,shared.ok);
-}
-
-/* Failover cron function, checks coordinated failover state. 
- *
- * Implementation note: The current implementation calls replicationSetMaster()
- * to start the failover request, this has some unintended side effects if the
- * failover doesn't work like blocked clients will be unblocked and replicas will
- * be disconnected. This could be optimized further.
- */
-void updateFailoverStatus(void) {
-    if (server.failover_state != FAILOVER_WAIT_FOR_SYNC) return;
-    mstime_t now = server.mstime;
-
-    /* Check if failover operation has timed out */
-    if (server.failover_end_time && server.failover_end_time <= now) {
-        if (server.force_failover) {
-            serverLog(LL_NOTICE,
-                "FAILOVER to %s:%d time out exceeded, failing over.",
-                server.target_replica_host, server.target_replica_port);
-            server.failover_state = FAILOVER_IN_PROGRESS;
-            /* If timeout has expired force a failover if requested. */
-            replicationSetMaster(server.target_replica_host,
-                server.target_replica_port);
-            return;
-        } else {
-            /* Force was not requested, so timeout. */
-            abortFailover("Replica never caught up before timeout");
-            return;
-        }
-    }
-
-    /* Check to see if the replica has caught up so failover can start */
-    client *replica = NULL;
-    if (server.target_replica_host) {
-        replica = findReplica(server.target_replica_host, 
-            server.target_replica_port);
-    } else {
-        listIter li;
-        listNode *ln;
-
-        listRewind(server.slaves,&li);
-        /* Find any replica that has matched our repl_offset */
-        while((ln = listNext(&li))) {
-            replica = ln->value;
-            if (replica->repl_ack_off == server.master_repl_offset) {
-                char ip[NET_IP_STR_LEN], *replicaaddr = replica->slave_addr;
-
-                if (!replicaaddr) {
-                    if (connAddrPeerName(replica->conn,ip,sizeof(ip),NULL) == -1)
-                        continue;
-                    replicaaddr = ip;
-                }
-
-                /* We are now failing over to this specific node */
-                server.target_replica_host = zstrdup(replicaaddr);
-                server.target_replica_port = replica->slave_listening_port;
-                break;
-            }
-        }
-    }
-
-    /* We've found a replica that is caught up */
-    if (replica && (replica->repl_ack_off == server.master_repl_offset)) {
-        server.failover_state = FAILOVER_IN_PROGRESS;
-        serverLog(LL_NOTICE,
-                "Failover target %s:%d is synced, failing over.",
-                server.target_replica_host, server.target_replica_port);
-        /* Designated replica is caught up, failover to it. */
-        replicationSetMaster(server.target_replica_host,
-            server.target_replica_port);
-    }
-}