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

Code before:
    case REPL_DUAL_CHANNEL_RECEIVE_REPLCONF_REPLY:
        ret = dualChannelReplHandleReplconfReply(conn, &err);
        if (ret == C_OK) server.repl_rdb_channel_state = REPL_DUAL_CHANNEL_RECEIVE_ENDOFF;
        break;
    case REPL_DUAL_CHANNEL_RECEIVE_ENDOFF:
        ret = dualChannelReplHandleEndOffsetResponse(conn, &err);
        if (ret == C_OK) server.repl_rdb_channel_state = REPL_DUAL_CHANNEL_RDB_LOAD;
        break;
    default:
        serverPanic("Unexpected dual replication state: %d", server.repl_rdb_channel_state);
    }
    if (ret == C_ERR) goto error;
    sdsfree(err);
    return;

error:
    if (err) {
        serverLog(LL_WARNING, "Dual channel sync failed with error %s", err);
        sdsfree(err);
    }
    if (server.repl_transfer_s) {
        connClose(server.repl_transfer_s);
        server.repl_transfer_s = NULL;
    }
    if (server.repl_rdb_transfer_s) {
        connClose(server.repl_rdb_transfer_s);
        server.repl_rdb_transfer_s = NULL;
    }
    cleanupTransferResources();
    server.repl_state = REPL_STATE_CONNECT;
    replicationAbortDualChannelSyncTransfer();
}

/* Replication: Replica side.
 * Initialize server.pending_repl_data infrastructure, we will allocate the buffer
 * itself once we need it */
void replDataBufInit(void) {
    serverAssert(server.pending_repl_data.blocks == NULL);
    server.pending_repl_data.len = 0;
    server.pending_repl_data.peak = 0;
    server.pending_repl_data.blocks = listCreate();
    server.pending_repl_data.blocks->free = zfree;
}

/* Replication: Replica side.
 * Track the local repl-data buffer streaming progress and serve clients from time to time */
void replStreamProgressCallback(size_t offset, int readlen, time_t *last_progress_callback) {
    time_t now = mstime();
    if (server.loading_process_events_interval_bytes &&
        ((offset + readlen) / server.loading_process_events_interval_bytes >
         offset / server.loading_process_events_interval_bytes) &&

Review comment (suggestion):
replicationAbortDualChannelSyncTransfer also some cleanups, like replicationAbortDualChannelSyncTransfer will close repl_rdb_transfer_s and repl_transfer_fd. We can remove it in here. Please let me know if you want it removed.

Code after:
        /* fall through */
    case REPL_DUAL_CHANNEL_RECEIVE_REPLCONF_REPLY:
        ret = dualChannelReplHandleReplconfReply(conn, &err);
        if (ret == C_OK) server.repl_rdb_channel_state = REPL_DUAL_CHANNEL_RECEIVE_ENDOFF;
        break;
    case REPL_DUAL_CHANNEL_RECEIVE_ENDOFF:
        ret = dualChannelReplHandleEndOffsetResponse(conn, &err);
        if (ret == C_OK) server.repl_rdb_channel_state = REPL_DUAL_CHANNEL_RDB_LOAD;
        break;
    default:
        serverPanic("Unexpected dual replication state: %d", server.repl_rdb_channel_state);
    }
    if (ret == C_ERR) goto error;
    sdsfree(err);
    return;

error:
    if (err) {
        serverLog(LL_WARNING, "Dual channel sync failed with error %s", err);
        sdsfree(err);
    }
    if (server.repl_transfer_s) {
        connClose(server.repl_transfer_s);
        server.repl_transfer_s = NULL;
    }
    replicationAbortDualChannelSyncTransfer();
    server.repl_state = REPL_STATE_CONNECT;
}

/* Replication: Replica side.
 * Initialize server.pending_repl_data infrastructure, we will allocate the buffer
 * itself once we need it */
void replDataBufInit(void) {
    serverAssert(server.pending_repl_data.blocks == NULL);
    server.pending_repl_data.len = 0;
    server.pending_repl_data.peak = 0;
    server.pending_repl_data.blocks = listCreate();
    server.pending_repl_data.blocks->free = zfree;
}

/* Replication: Replica side.
 * Track the local repl-data buffer streaming progress and serve clients from time to time */
void replStreamProgressCallback(size_t offset, int readlen, time_t *last_progress_callback) {
    time_t now = mstime();
    if (server.loading_process_events_interval_bytes &&
        ((offset + readlen) / server.loading_process_events_interval_bytes >
         offset / server.loading_process_events_interval_bytes) &&
        (now - *last_progress_callback > server.loading_process_events_interval_ms)) {
        replicationSendNewlineToPrimary();
        processEventsWhileBlocked();
        *last_progress_callback = now;