127 lines
4.9 KiB
JavaScript
127 lines
4.9 KiB
JavaScript
'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');
|
|
}
|
|
|
|
// identify programming errors in tasks that forgot to clean up locks
|
|
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}`);
|
|
});
|
|
}
|