cloudron-box/src/locks.js

'use strict';

exports = module.exports = {
    setTaskId,

    acquire,
    wait,

    release,
    releaseAll,
    releaseByTaskId,

    TYPE_APP_TASK_PREFIX: 'app_task_',
    TYPE_APP_BACKUP_PREFIX: 'app_backup_',
    TYPE_BOX_UPDATE: 'box_update', // for the actual update and after the backup. this allows the backup before update do not block
    TYPE_BOX_UPDATE_TASK: 'box_update_task', // for scheduling the update task
    TYPE_FULL_BACKUP_TASK: 'full_backup_task', // for scheduling the backup task

    TYPE_MAIL_SERVER_RESTART: 'mail_restart',
};

const assert = require('assert'),
    BoxError = require('./boxerror.js'),
    database = require('./database.js'),
    debug = require('debug')('box:locks'),
    promiseRetry = require('./promise-retry.js');

let gTaskId = null;

function setTaskId(taskId) {
    assert.strictEqual(typeof taskId, 'string');
    gTaskId = taskId;
}

async function read() {
    const result = await database.query('SELECT version, dataJson FROM locks');
    return { version: result[0].version, data: JSON.parse(result[0].dataJson) };
}

async function write(value) {
    assert.strictEqual(typeof value.version, 'number');
    assert.strictEqual(typeof value.data, 'object');

    const result = await database.query('UPDATE locks SET dataJson=?, version=version+1 WHERE id=? AND version=?', [ JSON.stringify(value.data), 'platform', value.version ]);
    if (result.affectedRows !== 1) throw new BoxError(BoxError.CONFLICT, 'Someone updated before we did');
    debug(`write: current locks: ${JSON.stringify(value.data)}`);
}

function canAcquire(data, type) {
    assert.strictEqual(typeof data, 'object');
    assert.strictEqual(typeof type, 'string');

    if (type === exports.TYPE_BOX_UPDATE) {
        if (Object.keys(data).some(k => k.startsWith(exports.TYPE_APP_TASK_PREFIX))) return new BoxError(BoxError.BAD_STATE, 'One or more app tasks are active');
        if (Object.keys(data).some(k => k.startsWith(exports.TYPE_APP_BACKUP_PREFIX))) return new BoxError(BoxError.BAD_STATE, 'One or more app backups are active');
    } else if (type.startsWith(exports.TYPE_APP_TASK_PREFIX)) {
        if (exports.TYPE_BOX_UPDATE in data) return new BoxError(BoxError.BAD_STATE, 'Update is active');
    } else if (type === exports.TYPE_FULL_BACKUP_TASK) {
        if (exports.TYPE_BOX_UPDATE_TASK in data) return new BoxError(BoxError.BAD_STATE, 'Update task is active');
    } else if (type === exports.TYPE_BOX_UPDATE_TASK) {
        if (exports.TYPE_FULL_BACKUP_TASK in data) return new BoxError(BoxError.BAD_STATE, 'Backup task is active');
    }

    // TYPE_MAIL_SERVER_RESTART can co-run with everything

    return null;
}

async function acquire(type) {
    assert.strictEqual(typeof type, 'string');

    await promiseRetry({ times: Number.MAX_SAFE_INTEGER, interval: 100, debug, retry: (error) => error.reason === BoxError.CONFLICT }, async () => {
        const { version, data } = await read();
        if (type in data) throw new BoxError(BoxError.BAD_STATE, `Locked by ${data[type]}`);
        const error = canAcquire(data, type);
        if (error) throw error;
        data[type] = gTaskId;
        await write({ version, data });
        debug(`acquire: ${type}`);
    });
}

async function wait(type) {
    assert.strictEqual(typeof type, 'string');

    await promiseRetry({ times: Number.MAX_SAFE_INTEGER, interval: 10000, debug }, async () => await acquire(type));
}

async function release(type) {
    assert.strictEqual(typeof type, 'string');

    await promiseRetry({ times: Number.MAX_SAFE_INTEGER, interval: 100, debug, retry: (error) => error.reason === BoxError.CONFLICT }, async () => {
        const { version, data } = await read();
        if (!(type in data)) throw new BoxError(BoxError.BAD_STATE, `Lock ${type} was never acquired`);
        if (data[type] !== gTaskId) throw new BoxError(BoxError.BAD_STATE, `Task ${gTaskId} attempted to release lock ${type} acquired by ${data[type]}`);
        delete data[type];
        await write({ version, data });
        debug(`release: ${type}`);
    });
}

async function releaseAll() {
    await database.query('DELETE FROM locks');
    await database.query('INSERT INTO locks (id, dataJson) VALUES (?, ?)', [ 'platform', JSON.stringify({}) ]);
    debug('releaseAll: all locks released');
}

async function releaseByTaskId(taskId) {
    assert.strictEqual(typeof taskId, 'string');

    await promiseRetry({ times: Number.MAX_SAFE_INTEGER, interval: 100, debug, retry: (error) => error.reason === BoxError.CONFLICT }, async () => {
        const { version, data } = await read();

        for (const type of Object.keys(data)) {
            if (data[type] === taskId) {
                debug(`releaseByTaskId: task ${taskId} forgot to unlock ${type}`);
                delete data[type];
            }
        }

        await write({ version, data });

        debug(`releaseByTaskId: ${taskId}`);
    });
}