sc-site-sync/lib/siteLock.js

// The site-wide apply lock: at most ONE apply or rollback runs on a site (tenant)
// at a time, across every cluster worker and every server sharing the database.
//
// The lock is one row of the raw _ss_apply_lock table (per tenant schema), held
// under a LEASE that the holder renews RENEWALS_PER_LEASE times per LEASE_MS while
// it works, so the lock of a holder that stops (crash, kill, restart) lapses within
// LEASE_MS. Acquiring is one atomic statement each way -- INSERT ... ON CONFLICT DO
// NOTHING while the lock is free, else an UPDATE that only matches a lapsed lease --
// so of any number of racing callers exactly one wins.
//
// Every apply and rollback runs under the lock (apply.applyPlan,
// rollback.rollbackApply), so its holder knows no other apply or rollback is
// running: a plan still marked applying / rolling_back was abandoned by a holder
// that stopped. acquire() therefore returns each such plan to the status it was
// claimed from (_ss_plans.claimed_from), exactly as a failed apply or rollback
// restores it: an interrupted apply is pending again (its restore point rolls back
// what it changed), an interrupted rollback can be retried.
//
// A holder that stalls past its lease (event loop blocked, database unreachable)
// can lose the lock to another caller. It notices at its next checkpoint
// (assertHeld) and stops without writing plan statuses again: the caller that took
// over has already recovered them.
//
// Lease times are read from the DATABASE's clock (dbNowMs), never a server's own,
// so servers sharing the database agree on when a lease lapses even when their
// clocks disagree.

const db = require("@saltcorn/data/db");

const planLib         = require("./plan");
const { randomUuid }  = require("./ids");
const { PLAN_STATUS, APPLY_LOCK_TABLE } = require("./constants");

// The one row per tenant schema.
const LOCK_NAME = "apply";
// How long a holder's lock outlives its last renewal: the wait before a stopped
// holder's plans can be recovered.
const LEASE_MS = 60 * 1000;
// Renewals per lease: a holder only loses the lock after missing all but the last.
const RENEWALS_PER_LEASE = 4;
// SQLite reports a write it could not start within its busy timeout (another
// process is writing -- usually the running apply) as SQLITE_BUSY: retry the
// claim this many times, this far apart, before refusing.
const SQLITE_BUSY    = "SQLITE_BUSY";
const BUSY_RETRIES   = 3;
const BUSY_RETRY_MS  = 250;

// The plan statuses only a running apply / rollback holds, and where an abandoned
// one returns when it has no claimed_from (claimed before 0.0.4): an apply is
// pending again, a rollback applied again (so it can be retried).
const RECOVERY = [
    { inFlight: PLAN_STATUS.APPLYING,     fallback: PLAN_STATUS.PENDING },
    { inFlight: PLAN_STATUS.ROLLING_BACK, fallback: PLAN_STATUS.APPLIED }
];
const IN_FLIGHT = RECOVERY.map((r) => r.inFlight);

// How the lock stands for an admin (operations.applyLockView): nothing running, an
// apply or rollback running, or plans left in flight by a process that stopped.
const LOCK_STATES = {
    IDLE:        "idle",
    RUNNING:     "running",
    INTERRUPTED: "interrupted"
};


class LockBusyError extends Error {
    constructor(message, holder) {
        super(message);
        this.name   = "LockBusyError";
        this.holder = holder;
    }
}


class LockLostError extends Error {
    constructor(message) {
        super(message);
        this.name = "LockLostError";
    }
}


const lockRef = () => `${db.getTenantSchemaPrefix()}${APPLY_LOCK_TABLE}`;


// SQL for the database clock's current time in epoch milliseconds.
const dbNowMs = () => {
    return db.isSQLite ? "CAST((julianday('now') - 2440587.5) * 86400000 AS INTEGER)" : "CAST(EXTRACT(EPOCH FROM clock_timestamp()) * 1000 AS BIGINT)";
};


// Take the lock row for `holder`: insert it while the lock is free, else take over
// a lapsed lease. True only for the one caller that won. (Placeholders stay in
// order of appearance: SQLite numbers `$n` parameters as it meets them.)
const claimRow = async (ref, holder, operation, planId, leaseMs) => {
    const values = [holder, operation, planId, new Date().toISOString(), leaseMs];
    const inserted = await db.query(
        `INSERT INTO ${ref} (holder, operation, plan_id, acquired_at, expires_ms, lock_name) VALUES ($1,$2,$3,$4,${dbNowMs()} + $5,$6) ` +
        `ON CONFLICT DO NOTHING RETURNING holder`,
        values.concat([LOCK_NAME])
    );
    if (inserted.rows.length === 1) {
        return true;
    }
    const taken = await db.query(
        `UPDATE ${ref} SET holder = $1, operation = $2, plan_id = $3, acquired_at = $4, expires_ms = ${dbNowMs()} + $5 ` +
        `WHERE lock_name = $6 AND expires_ms < ${dbNowMs()} RETURNING holder`,
        values.concat([LOCK_NAME])
    );
    return taken.rows.length === 1;
};


// claimRow, riding out SQLITE_BUSY: false as soon as a live holder can be read
// (reads still work while another process writes), a retry otherwise, and
// LockBusyError when the database stays busy.
const claimRowPatiently = async (ref, holder, operation, planId, leaseMs) => {
    for (let attempt = 0; ; attempt++) {
        try {
            return await claimRow(ref, holder, operation, planId, leaseMs);
        } catch (e) {
            if (e.code !== SQLITE_BUSY) {
                throw e;
            }
            const state = await lockState().catch(() => null);
            if (state && state.live) {
                return false;
            }
            if (attempt >= BUSY_RETRIES) {
                throw new LockBusyError("the site database is busy; try again in a moment", null);
            }
            await new Promise((resolve) => setTimeout(resolve, BUSY_RETRY_MS));
        }
    }
};


// Return every plan left in flight to the status it was claimed from. Only safe
// while holding the lock (no apply or rollback can be running then).
const recoverAbandoned = async () => {
    const recovered = [];
    for (const r of RECOVERY) {
        const rs = await db.query(
            `UPDATE ${db.getTenantSchemaPrefix()}_ss_plans SET status = COALESCE(claimed_from, $1) WHERE status = $2 RETURNING plan_id, status`,
            [r.fallback, r.inFlight]
        );
        for (const row of rs.rows) {
            recovered.push({ plan_id: row.plan_id, was: r.inFlight, status: row.status });
        }
    }
    return recovered;
};


// The held lock: renewed in the background until released.
const makeLease = (ref, holder, leaseMs) => {
    let lost = false;
    const renew = async () => {
        if (lost) {
            return false;
        }
        const rs = await db.query(
            `UPDATE ${ref} SET expires_ms = ${dbNowMs()} + $1 WHERE lock_name = $2 AND holder = $3 RETURNING holder`,
            [leaseMs, LOCK_NAME, holder]
        );
        lost = rs.rows.length === 0;
        return !lost;
    };
    // A failed renewal (database hiccup) is retried on the next tick; only finding
    // the row gone or taken over loses the lock.
    const timer = setInterval(() => {
        renew().catch(() => {});
    }, Math.floor(leaseMs / RENEWALS_PER_LEASE));
    timer.unref();
    return {
        recovered: [],
        renew,
        assertHeld: async () => {
            if (!(await renew())) {
                throw new LockLostError(`lost the site lock after stalling longer than its ${leaseMs / 1000}s lease, so stopped here; another apply or rollback may have taken over and recovered this plan`);
            }
        },
        release: async () => {
            clearInterval(timer);
            // A release that fails lapses with the lease instead.
            await db.query(`DELETE FROM ${ref} WHERE lock_name = $1 AND holder = $2`, [LOCK_NAME, holder]).catch(() => {});
        }
    };
};


// Who holds the lock, read-only: null when free; else the operation, its plan,
// when it started, when its lease runs out, and live (false: its holder stopped
// renewing, so the next acquire takes over).
const lockState = async () => {
    const rs = await db.query(`SELECT *, ${dbNowMs()} AS db_now_ms FROM ${lockRef()} WHERE lock_name = $1`, [LOCK_NAME]);
    if (rs.rows.length === 0) {
        return null;
    }
    const row     = rs.rows[0];
    const expires = Number(row.expires_ms);
    return {
        operation:   row.operation,
        plan_id:     row.plan_id,
        acquired_at: row.acquired_at,
        expires_at:  new Date(expires).toISOString(),
        live:        expires >= Number(row.db_now_ms)
    };
};


const busyMessage = (holder) => {
    if (!holder) {
        return "another apply or rollback was just finishing on this site; try again";
    }
    return `another ${holder.operation} is running on this site (plan ${holder.plan_id}, since ${holder.acquired_at}); ` +
        `try again once it has finished. If the process running it stopped, its lock lapses by ${holder.expires_at} ` +
        `and the next apply or rollback recovers that plan.`;
};


// Acquire the lock for `operation` ("apply" / "rollback") of planId, or throw
// LockBusyError. The returned lease lists the abandoned plans it recovered.
const acquire = async ({ operation, planId, leaseMs = LEASE_MS }) => {
    const ref    = lockRef();
    const holder = randomUuid();
    const won    = await claimRowPatiently(ref, holder, operation, planId || null, leaseMs);
    if (!won) {
        const state = await lockState();
        throw new LockBusyError(busyMessage(state), state);
    }
    const lease = makeLease(ref, holder, leaseMs);
    try {
        lease.recovered = await recoverAbandoned();
    } catch (e) {
        await lease.release();
        throw e;
    }
    return lease;
};


// Run fn(lease) holding the lock. With `lease` (a step of an operation that holds
// it already, such as the apply inside a rollback) fn runs under that lease, with
// nothing recovered of its own.
const withSiteLock = async ({ operation, planId, lease }, fn) => {
    if (lease) {
        return await fn(Object.assign({}, lease, { recovered: [] }));
    }
    const held = await acquire({ operation, planId });
    try {
        return await fn(held);
    } finally {
        await held.release();
    }
};


// Run fn() for a plan this caller claimed (moved to an in-flight status). On
// success the plan moves to `done`, on failure back to `prior` -- unless the lock
// was lost meanwhile: the caller that took it over has recovered the plan already
// and may be working on it. A failure carries `planId`.
const finishClaimed = async (lease, planId, { prior, done }, fn) => {
    try {
        const result = await fn();
        await lease.assertHeld();
        await planLib.setPlanStatus(planId, done);
        return result;
    } catch (e) {
        if (!(e instanceof LockLostError) && (await lease.renew().catch(() => false))) {
            await planLib.setPlanStatus(planId, prior);
        }
        e.planId = planId;
        throw e;
    }
};


// Where an in-flight plan goes when recovered (row: status + claimed_from).
const recoveredStatus = (row) => {
    const r = RECOVERY.find((x) => x.inFlight === row.status);
    return r ? row.claimed_from || r.fallback : row.status;
};


// Warnings telling the caller which abandoned plans its lease recovered.
const recoveryWarnings = (lease) => {
    return lease.recovered.map((r) => {
        const next = r.was === PLAN_STATUS.APPLYING
            ? "this site may be partly changed by it: roll it back from its restore point, or apply it again"
            : "its rollback can be retried";
        return `plan ${r.plan_id} was left ${r.was} by a process that stopped part-way; it is ${r.status} again (${next})`;
    });
};


module.exports = {
    IN_FLIGHT,
    LOCK_STATES,
    LockBusyError,
    LockLostError,
    acquire,
    withSiteLock,
    finishClaimed,
    lockState,
    recoveredStatus,
    recoveryWarnings
};