electric-cherry/src/lib/server/provisioning/cancellation.ts

import { pauseScalingForOperation } from '../autoscaling/ownership.js';
import { db } from '../db/index.js';
import { recordProgress } from './progress.js';
import { requestRebalanceStopForOperation } from '../rebalancing/admission.js';

export const rebalanceCancellationGuard = `NOT (d.triggered_by LIKE 'rebalance:%' AND EXISTS (
    SELECT 1 FROM ec_host_migrations m WHERE m.deployment_id=d.id AND m.remove_operation_id IS NOT NULL))`;

export function rebalanceCancellationBlocker(id: string): string | null {
    return db()
        .prepare(
            `SELECT 1 FROM server_deployments d WHERE d.id=? AND NOT (${rebalanceCancellationGuard})`
        )
        .get(id)
        ? 'This rebalance move has committed source removal and must finish recreating the lobby. Stop the rebalance run to withdraw its remaining moves.'
        : null;
}

export function provisioningCancelled(id: string): boolean {
    return !!db()
        .prepare("SELECT 1 FROM server_deployments WHERE id=? AND state='cancelled'")
        .get(id);
}

export function assertProvisioningNotCancelled(id: string): void {
    if (provisioningCancelled(id)) throw new Error('Provisioning operation cancelled');
}

export function canCancelProvisioningJob(id: string): boolean {
    return !!db()
        .prepare(
            `SELECT 1 FROM server_deployments d WHERE d.id=? AND d.server_deleted_at IS NULL
        AND d.state IN ('queued','waiting','running','awaiting-health','interrupted') AND ${rebalanceCancellationGuard}`
        )
        .get(id);
}

/** Cancel remaining work, not effects already performed by the host. */
export function cancelProvisioningJob(
    id: string,
    reason = 'Cancelled by operator',
    options: { preserveRebalanceRun?: boolean } = {}
): boolean {
    return db().transaction(() => {
        if (provisioningCancelled(id)) return true;
        if (!canCancelProvisioningJob(id)) return false;
        const rebalance = !!db()
            .prepare(
                "SELECT 1 FROM server_deployments WHERE id=? AND triggered_by LIKE 'rebalance:%'"
            )
            .get(id);
        if (rebalance) {
            if (!options.preserveRebalanceRun) requestRebalanceStopForOperation(id);
        } else pauseScalingForOperation(id);
        const at = new Date().toISOString();
        // Only commands never delivered can be withdrawn. Delivered commands retain
        // their immutable IDs and receipts; replacements wait for their resolution.
        const commands = db()
            .prepare(
                `SELECT o.id FROM ec_host_operations o
            WHERE o.state='pending' AND o.last_delivered_at IS NULL AND o.id IN (
                SELECT operation_id FROM native_provisioning WHERE deployment_id=?
                UNION SELECT stage_operation_id FROM native_provisioning WHERE deployment_id=?
                UNION SELECT remove_operation_id FROM ec_host_migrations WHERE deployment_id=?)`
            )
            .all(id, id, id) as { id: string }[];
        for (const command of commands) {
            db()
                .prepare(
                    "UPDATE ec_host_operations SET state='failed',error=?,updated_at=? WHERE id=?"
                )
                .run(reason, at, command.id);
            db()
                .prepare(
                    "INSERT INTO ec_host_operation_events(operation_id,state,created_at,error) VALUES(?,'failed',?,?)"
                )
                .run(command.id, at, reason);
        }
        db()
            .prepare(
                `UPDATE staged_file_changes SET state='cancelled' WHERE state='pending' AND EXISTS (
            SELECT 1 FROM provisioning_sources s WHERE s.operation_id=? AND s.kind=staged_file_changes.source_kind
                AND s.source_id=staged_file_changes.source_id AND s.server=staged_file_changes.server)`
            )
            .run(id);
        db().prepare('DELETE FROM restart_rollout WHERE operation_id=?').run(id);
        db()
            .prepare(
                "UPDATE server_deployments SET state='cancelled',finished_at=?,stage=?,error=NULL WHERE id=?"
            )
            .run(at, reason, id);
        recordProgress(id, {
            blocker: null,
            checks: [],
            recovery: {
                state: 'released',
                at,
                operationId: id,
                message:
                    'Recovery capacity released by cancellation; this does not certify server health.'
            },
            event: {
                kind: 'outcome',
                message: reason,
                details: { state: 'cancelled', finishedAt: at, error: null, warnings: [] }
            }
        });
        return true;
    })();
}

/** A cancelled controller job must not let a replacement overtake a delivered mutation. */
export function unsettledCancelledCommand(
    server: string,
    replacementId: string
): {
    deploymentId: string;
    operationId: string;
    state: string;
    error: string | null;
} | null {
    const row = db()
        .prepare(
            `SELECT d.id AS deploymentId,o.id AS operationId,o.state,o.error FROM server_deployments d
        JOIN native_provisioning n ON n.deployment_id=d.id
        LEFT JOIN ec_host_migrations m ON m.deployment_id=d.id
        JOIN ec_host_operations o ON o.id=n.operation_id OR o.id=m.remove_operation_id
        WHERE d.server=? AND d.server_deleted_at IS NULL AND d.state='cancelled'
        AND (o.state NOT IN ('completed','failed')
            OR (o.state='completed' AND o.id=n.operation_id AND n.files_json IS NOT NULL AND n.applied_at IS NULL)
            OR (o.state='failed' AND o.id=n.operation_id AND o.kind='instance.rebuild'
                AND json_type(o.result,'$.rebuildFailure') IS NOT NULL AND n.applied_at IS NULL))
        AND NOT EXISTS (SELECT 1 FROM native_creation_cancellations c WHERE c.creation_id=d.id AND c.cleanup_id=?)
        ORDER BY d.rowid,o.rowid LIMIT 1`
        )
        .get(server, replacementId) as
        | { deploymentId: string; operationId: string; state: string; error: string | null }
        | undefined;
    return row ?? null;
}