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

Code before:
}

void receiveRDBinBioThreadSingleChannel(connection *conn) {
    UNUSED(conn);
    receiveRDBinBioThread(0);
}

void receiveRDBinBioThreadDualChannel(connection *conn) {
    UNUSED(conn);
    receiveRDBinBioThread(1);
}

/* Replication: Replica side.
 * Upon dual-channel sync failure, close rdb-connection, reset repl-state, reset
 * provisional primary struct, and free local replication buffer. */
void replicationAbortDualChannelSyncTransfer(void) {
    serverAssert(server.repl_rdb_channel_state != REPL_DUAL_CHANNEL_STATE_NONE);
    dualChannelServerLog(LL_NOTICE, "Aborting dual channel sync");
    if (server.repl_rdb_transfer_s) {
        connClose(server.repl_rdb_transfer_s);
        server.repl_rdb_transfer_s = NULL;
    }
    cleanupTransferResources();
    server.repl_rdb_channel_state = REPL_DUAL_CHANNEL_STATE_NONE;
    server.repl_provisional_primary.read_reploff = 0;
    server.repl_provisional_primary.reploff = 0;
    server.repl_provisional_primary.conn = NULL;
    server.repl_provisional_primary.dbid = -1;
    server.rdb_client_id = -1;
    freePendingReplDataBuf();
    return;
}

/* Replication: Primary side.
 * Send current replication offset to replica. Use the following structure:
 * $ENDOFF:<repl-offset> <primary-repl-id> <current-db-id> <client-id> */
int sendCurrentOffsetToReplica(client *replica) {
    char buf[128];
    int buflen;
    buflen = snprintf(buf, sizeof(buf), "$ENDOFF:%lld %s %d %llu\r\n", server.primary_repl_offset, server.replid,
                      server.replicas_eldb, (long long unsigned int)replica->id);
    dualChannelServerLog(LL_NOTICE, "Sending to replica %s RDB end offset %lld and client-id %llu",
                         replicationGetReplicaName(replica), server.primary_repl_offset,
                         (long long unsigned int)replica->id);
    if (connSyncWrite(replica->conn, buf, buflen, server.repl_syncio_timeout * 1000) != buflen) {
        freeClientAsync(replica);
        return C_ERR;
    }
    return C_OK;
}


Review comment (suggestion):
This could lead to a use-after-free scenario or the background thread accessing freed memory.
The async free function (`freePendingReplDataBufAsync`) will try to access the list later in a background thread, but the pointer has already been nullified. 

Code after:
    sds err = sendCommandRaw(conn, cmd);
    sdsfree(cmd);
    if (err) return err;
    return NULL;
}

/* Replication: Replica side.
 * Returns an sds represent this replica port to be used by the primary (mostly
 * for logs) */
sds getReplicaPortString(void) {
    long long replica_port;
    if (server.replica_announce_port) {
        replica_port = server.replica_announce_port;
    } else if (server.tls_replication && server.tls_port) {
        replica_port = server.tls_port;
    } else {
        replica_port = server.port;
    }
    return sdsfromlonglong(replica_port);
}

/* Replication: Replica side.
 * Free replica's local replication buffer */
void freePendingReplDataBuf(void) {
    if (server.pending_repl_data.blocks) freePendingReplDataBufAsync(server.pending_repl_data.blocks);
    server.pending_repl_data.blocks = NULL;
    server.pending_repl_data.len = 0;
}

void receiveRDBinBioThread(bool is_dual_channel) {
    serverLog(LL_NOTICE, "Replica main thread creating Bio thread to save RDB to disk");
    connection *conn;
    /* Thread safety: revoke the main thread's access to the connection while the Bio thread uses it.
     * We'll restore it once the Bio thread terminates */
    if (is_dual_channel) {
        conn = server.repl_rdb_transfer_s;
        server.repl_rdb_transfer_s = NULL;
    } else {
        conn = server.repl_transfer_s;
        server.repl_transfer_s = NULL;
    }
    connSetReadHandler(conn, NULL);
    bioCreateSaveRDBToDiskJob(conn, is_dual_channel);
}

void receiveRDBinBioThreadSingleChannel(connection *conn) {
    UNUSED(conn);
    receiveRDBinBioThread(0);
}

void receiveRDBinBioThreadDualChannel(connection *conn) {