smartbotic/nodes/sdcpp/sdcpp-job-wait.js

/**
 * @node sdcpp-job-wait
 * @name SD.cpp Wait For Job
 * @category sdcpp
 * @version 1.0.0
 * @description Poll an sdcpp-restapi job until it finishes, fails or the wait runs out
 * @icon hourglass
 */

const configSchema = {
    type: 'object',
    properties: {
        serverUrl: {
            type: 'string',
            title: 'Server URL',
            description: 'Base address of the sdcpp-restapi server',
            default: 'http://localhost:8077'
        },
        credentialId: {
            type: 'string',
            title: 'Credential',
            description: 'A basic credential holding the sdcpp-restapi username and password',
            dynamicOptions: {
                source: 'credentials',
                filter: { type: ['basic'] }
            }
        },
        jobId: {
            type: 'string',
            title: 'Job ID',
            description: 'Id of the job to wait for, normally {{$json.jobId}} from an SD.cpp Generate node'
        },
        pollInterval: {
            type: 'number',
            title: 'Poll Interval (ms)',
            description: 'How long to pause between status checks. The server renders serially, so polling faster than a second gains nothing',
            default: 2000
        },
        maxWait: {
            type: 'number',
            title: 'Maximum Wait (ms)',
            description: 'Give up after this long and report the job as still running. This holds one execution slot for the whole wait, so keep it close to how long a render really takes on this server',
            default: 600000
        },
        failOnTimeout: {
            type: 'boolean',
            title: 'Fail If The Wait Runs Out',
            description: 'Throw when the maximum wait passes and the job is still going. Leave off to return the unfinished job and decide in the workflow. The job itself keeps running either way',
            default: true
        },
        failOnJobFailure: {
            type: 'boolean',
            title: 'Fail If The Job Failed',
            description: 'Throw when the job ends in the failed or cancelled state, instead of returning it as data',
            default: true
        },
        timeout: {
            type: 'number',
            title: 'Request Timeout (ms)',
            description: 'Timeout for each individual status request, not for the wait as a whole',
            default: 30000
        }
    },
    required: ['credentialId']
};

const inputSchema = {
    type: 'object',
    properties: {
        data: { type: 'any' }
    }
};

const outputSchema = {
    type: 'object',
    properties: {
        jobId: { type: 'string' },
        status: { type: 'string' },
        type: { type: 'string' },
        done: { type: 'boolean', description: 'False only when the wait ran out while the job was still going' },
        succeeded: { type: 'boolean' },
        timedOut: { type: 'boolean', description: 'True when the maximum wait passed before the job finished' },
        step: { type: 'number' },
        totalSteps: { type: 'number' },
        percent: { type: 'number' },
        outputs: { type: 'array', description: 'Relative output paths, ready for an SD.cpp Fetch Output node' },
        urls: { type: 'array' },
        error: { type: 'string' },
        waitedMs: { type: 'number', description: 'How long this node actually waited' },
        polls: { type: 'number', description: 'How many status requests it made' },
        params: { type: 'object' },
        modelSettings: { type: 'object' }
    }
};

function normalizeServer(url) {
    const value = String(url || '').trim();
    if (!value) {
        throw new Error('SD.cpp: a server URL is required, such as http://localhost:8077');
    }
    return value.replace(/\/+$/, '');
}

function readCredential(credentialId) {
    const auth = smartbotic.credentials.get(credentialId);
    if (!auth || auth.success !== true) {
        throw new Error('SD.cpp: could not read the credential: ' +
            ((auth && auth.error) || 'unknown error'));
    }

    const value = auth.headerValue || '';
    if (value.indexOf('Basic ') !== 0) {
        throw new Error('SD.cpp: the credential must be a basic one, holding the sdcpp-restapi ' +
            'username and password');
    }

    const decoded = smartbotic.utils.base64Decode(value.substring(6));
    const separator = decoded.indexOf(':');
    if (separator < 1) {
        throw new Error('SD.cpp: the credential is malformed, expected a username and a password');
    }

    return {
        username: decoded.substring(0, separator),
        password: decoded.substring(separator + 1)
    };
}

function call(options) {
    const response = smartbotic.http.request(options);

    let body = response.data;
    if (typeof body === 'string' && body.length > 0) {
        try {
            body = JSON.parse(body);
        } catch (e) {
            const snippet = body.substring(0, 200).replace(/\s+/g, ' ');
            throw new Error('SD.cpp: ' + options.what + ' returned HTTP ' + response.status +
                ' with a body that is not JSON: ' + snippet);
        }
    }

    if (response.status < 200 || response.status >= 300) {
        const detail = (body && (body.message || body.error)) || ('HTTP ' + response.status);
        throw new Error('SD.cpp: ' + options.what + ' failed: ' + detail);
    }

    return body || {};
}

function login(server, credential, timeout) {
    const session = call({
        method: 'POST',
        url: server + '/auth/login',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
            username: credential.username,
            password: credential.password
        }),
        timeout: timeout,
        what: 'signing in'
    });

    if (!session.token) {
        throw new Error('SD.cpp: the server accepted the login but returned no token');
    }
    return session.token;
}

const TERMINAL = ['completed', 'failed', 'cancelled', 'deleted'];

function shapeJob(job, server) {
    const status = job.status || 'unknown';
    const progress = job.progress || {};
    const step = progress.step !== undefined ? progress.step : 0;
    const totalSteps = progress.total_steps !== undefined ? progress.total_steps : 0;
    const outputs = job.outputs || [];

    const urls = [];
    for (let i = 0; i < outputs.length; i++) {
        urls.push(server + '/output/' + outputs[i]);
    }

    return {
        jobId: job.job_id || '',
        status: status,
        type: job.type || '',
        done: TERMINAL.indexOf(status) !== -1,
        succeeded: status === 'completed',
        step: step,
        totalSteps: totalSteps,
        percent: totalSteps > 0 ? Math.round((step / totalSteps) * 100) : -1,
        outputs: outputs,
        urls: urls,
        error: job.error || '',
        createdAt: job.created_at || '',
        startedAt: job.started_at || '',
        completedAt: job.completed_at || '',
        params: job.params || {},
        modelSettings: job.model_settings || {}
    };
}

async function execute(config, input, context) {
    const server = normalizeServer(config.serverUrl);
    const timeout = config.timeout || 30000;
    const pollInterval = config.pollInterval > 0 ? config.pollInterval : 2000;
    const maxWait = config.maxWait > 0 ? config.maxWait : 600000;

    const jobId = String(config.jobId || '').trim();
    if (!jobId) {
        throw new Error('SD.cpp: a job id is required');
    }

    const credential = readCredential(config.credentialId);

    // Signing in once and reusing the token for every poll. A long render can
    // mean hundreds of checks, and logging in for each one would triple the
    // request count for no reason.
    const token = login(server, credential, timeout);

    const startedAt = Date.now();
    let polls = 0;
    let result = null;

    while (true) {
        const job = call({
            method: 'GET',
            url: server + '/queue/' + encodeURIComponent(jobId),
            headers: { 'Authorization': 'Bearer ' + token },
            timeout: timeout,
            what: 'reading job ' + jobId
        });

        polls++;
        result = shapeJob(job, server);

        if (result.done) {
            break;
        }

        // Check the clock before sleeping rather than after, so the node cannot
        // overshoot the maximum wait by a whole poll interval.
        const elapsed = Date.now() - startedAt;
        if (elapsed + pollInterval >= maxWait) {
            result.timedOut = true;
            break;
        }

        smartbotic.utils.sleep(pollInterval);
    }

    result.waitedMs = Date.now() - startedAt;
    result.polls = polls;
    result.timedOut = result.timedOut === true;

    if (result.timedOut) {
        // The job is untouched and still queued on the server. Reporting the id
        // back means a later run can pick it up rather than re-rendering it.
        smartbotic.log.warn('SD.cpp: stopped waiting for job ' + jobId + ' after ' +
            result.waitedMs + 'ms, it is still ' + result.status);
        if (config.failOnTimeout !== false) {
            throw new Error('SD.cpp: job ' + jobId + ' was still ' + result.status +
                ' after ' + result.waitedMs + 'ms. It is still running on the server; ' +
                'raise the maximum wait or check it again later with SD.cpp Job Status');
        }
        return result;
    }

    smartbotic.log.info('SD.cpp: job ' + jobId + ' finished as ' + result.status +
        ' after ' + result.waitedMs + 'ms and ' + polls + ' polls');

    if (config.failOnJobFailure !== false &&
        (result.status === 'failed' || result.status === 'cancelled')) {
        throw new Error('SD.cpp: job ' + jobId + ' ended as ' + result.status + ': ' +
            (result.error || 'no reason given'));
    }

    return result;
}

module.exports = { configSchema, inputSchema, outputSchema, execute };