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

Code before:
int proceedWithSlotExportJobConnecting(slotMigrationJob *job, bool *completed) {
    serverAssert(job->type == SLOT_MIGRATION_EXPORT && job->conn);
    *completed = false;

    switch (connGetState(job->conn)) {
    case CONN_STATE_CONNECTED:
        *completed = true;
        return C_OK;
    case CONN_STATE_CONNECTING: return C_OK;
    default:
        return C_ERR;
    }
}

/* Perform the authentication steps needed to authenticate a slot migration
 * job's connection. Return C_ERR if an error is encountered, or C_OK if the
 * authentication is done. */
int performSlotExportJobAuthentication(slotMigrationJob *job) {
    serverAssert(job->type == SLOT_MIGRATION_EXPORT);
    if (!server.primary_auth) {
        return C_OK;
    }
    sds err = replicationSendAuth(job->conn);
    if (err) {
        serverLog(LL_WARNING,
                  "Failed to send AUTH command for slot migration %s: %s",
                  job->description, err);
        sdsfree(err);
        return C_ERR;
    }
    err = receiveSynchronousResponse(job->conn);
    if (err == NULL) {
        serverLog(LL_WARNING,
                  "Received no response to AUTH command for slot migration %s",
                  job->description);
        return C_ERR;
    }
    if (err[0] == '-') {
        serverLog(LL_WARNING,
                  "Failed to AUTH for slot migration %s: %s",
                  job->description, err);
        sdsfree(err);
        return C_ERR;
    }
    serverLog(LL_NOTICE,
              "Successfully authenticated slot migration %s",
              job->description);
    sdsfree(err);
    return C_OK;
}


Review comment (suggestion):
Ideally we wouldn't use `receiveSynchronousResponse` (e.g. in case the response isn't fully received) but in practice I think this should work good enough.

We can iterate on it if needed

Code after:
}

/* Proceed with a previously started connection. If no problems have occurred,
 * C_OK is returned and completed is set. C_ERR will be returned if an issue was
 * encountered. */
int proceedWithSlotExportJobConnecting(slotMigrationJob *job, bool *completed) {
    serverAssert(job->type == SLOT_MIGRATION_EXPORT && job->conn);
    *completed = false;

    switch (connGetState(job->conn)) {
    case CONN_STATE_CONNECTED:
        *completed = true;
        return C_OK;
    case CONN_STATE_CONNECTING: return C_OK;
    default:
        return C_ERR;
    }
}

/* Read a response to the AUTH command, moving to the next stage of the migration
 * if the response is a success. If there is an error, fail the migration with the
 * error message. */
void slotMigrationJobReadAuthResponse(connection *conn) {
    slotMigrationJob *job = (slotMigrationJob *)connGetPrivateData(conn);

    sds err = receiveSynchronousResponse(job->conn);
    if (err == NULL) {
        finishSlotMigrationJob(job, SLOT_MIGRATION_JOB_FAILED, "Target node did not respond to AUTH command");
        return;
    }
    if (err[0] == '-') {
        sds status_msg = sdscatfmt(sdsempty(), "Failed to AUTH to target node: %s", err);
        finishSlotMigrationJob(job, SLOT_MIGRATION_JOB_FAILED, status_msg);
        sdsfree(err);
        sdsfree(status_msg);
        return;
    }

    sdsfree(err);
    serverLog(LL_NOTICE, "Successfully authenticated slot migration %s", job->description);
    updateSlotMigrationJobState(job, SLOT_EXPORT_SEND_ESTABLISH);
    proceedWithSlotMigration(job);
}

/* Perform the authentication steps needed to authenticate a slot migration
 * job's connection. */
void slotMigrationJobSendAuth(slotMigrationJob *job) {
    serverAssert(job->type == SLOT_MIGRATION_EXPORT);
    serverAssert(server.primary_auth);

    sds err = replicationSendAuth(job->conn);