Cancel active provisioning jobs when rolling back betas
This commit is contained in:
@@ -4,6 +4,8 @@ import { choosePlacement, availableHostPort, type PlacementOptions } from '../ho
|
||||
import { randomBytes, randomUUID } from 'node:crypto';
|
||||
import { posix } from 'node:path';
|
||||
import { db } from '../db/index.js';
|
||||
import { ensureHostSchema } from '../hosts/schema.js';
|
||||
import { unsettledCancelledCommand } from './cancellation.js';
|
||||
import { config, relDestForKind, serverDir } from '../config.js';
|
||||
import { nativeInstance, nativeRequest, unplacedNativeCreation, type NativeInstance } from '../backends/native.js';
|
||||
import { getHost, getHostOperationStatus, queueHostOperation, queueHostOperationSerialized, releaseHostPorts, reserveHostPort } from '../hosts/index.js';
|
||||
@@ -244,7 +246,64 @@ function applyReceipt(job: Job, result: Record<string, unknown>) {
|
||||
})();
|
||||
}
|
||||
|
||||
function hold(job: Job, message: string) { deploymentStage(job.id, message, { blocker: { code: 'native-host', message, automatic: true } }); }
|
||||
/** Retain confirmed file snapshots and instance identity even after observation is cancelled. */
|
||||
export function reconcileNativeProvisioningReceipt(id: string): void {
|
||||
const job = db().prepare(`SELECT d.*,n.instance_id,n.generation,n.operation_id,n.files_json,n.applied_at
|
||||
FROM server_deployments d JOIN native_provisioning n ON n.deployment_id=d.id
|
||||
WHERE d.id=? AND d.server_deleted_at IS NULL`).get(id) as Job | undefined;
|
||||
if (!job) return;
|
||||
const migration = hostMigration(id);
|
||||
if (job.state === 'cancelled' && migration?.remove_operation_id && !migration.removed_at) {
|
||||
const removal = getHostOperationStatus(migration.remove_operation_id);
|
||||
if (removal?.state === 'failed') {
|
||||
db().prepare('DELETE FROM ec_host_ports WHERE host_id=? AND instance_id=?').run(migration.target_host_id, migration.instance_id);
|
||||
} else if (removal?.state === 'completed') {
|
||||
const original = nativeInstance(job.server);
|
||||
if (original?.id === migration.instance_id && original.host_id === migration.source_host_id && original.generation === job.generation) {
|
||||
db().prepare('UPDATE native_instances SET retired_at=? WHERE id=?').run(removal.updatedAt, original.id);
|
||||
releaseHostPorts(original.id);
|
||||
db().prepare('DELETE FROM servers WHERE name=?').run(job.server);
|
||||
db().prepare('UPDATE ec_host_migrations SET removed_at=? WHERE deployment_id=?').run(removal.updatedAt, id);
|
||||
recordProgress(id, { event: { kind: 'cleanup', message: 'Cancelled migration: source removal completed; replacement creation was not started.' } });
|
||||
}
|
||||
}
|
||||
}
|
||||
if (!job.operation_id) return;
|
||||
const operation = getHostOperationStatus(job.operation_id);
|
||||
if (operation?.state === 'failed') {
|
||||
for (const file of JSON.parse(job.files_json ?? '[]') as FilePlan[]) {
|
||||
db().prepare("UPDATE staged_file_changes SET state=? WHERE id=? AND state='applying'")
|
||||
.run(job.state === 'cancelled' ? 'cancelled' : 'pending', file.id);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (operation?.state !== 'completed') return;
|
||||
const instance = nativeInstance(job.server);
|
||||
if (!instance || instance.id !== job.instance_id) return;
|
||||
if (job.action === 'destroy') {
|
||||
if (instance.generation !== job.generation) return;
|
||||
db().prepare('UPDATE native_instances SET retired_at=? WHERE id=?').run(operation.updatedAt, instance.id);
|
||||
releaseHostPorts(instance.id);
|
||||
db().prepare('DELETE FROM servers WHERE name=?').run(job.server);
|
||||
finishCreationCancellation(job.id);
|
||||
return;
|
||||
}
|
||||
if (job.action === 'rebuild' && !hostMigration(id) && instance.generation === job.generation) {
|
||||
db().prepare('UPDATE native_instances SET generation=generation+1 WHERE id=?').run(instance.id);
|
||||
// Replacements queued while the delivered rebuild was settling bound the
|
||||
// previous generation. Only untouched, newer jobs may follow this receipt.
|
||||
if (job.state === 'cancelled') db().prepare(`UPDATE native_provisioning SET generation=?
|
||||
WHERE instance_id=? AND generation=? AND operation_id IS NULL AND stage_operation_id IS NULL
|
||||
AND deployment_id IN (SELECT newer.id FROM server_deployments newer
|
||||
WHERE newer.rowid>(SELECT rowid FROM server_deployments WHERE id=?)
|
||||
AND newer.server_deleted_at IS NULL AND newer.state IN ('queued','waiting'))`)
|
||||
.run(job.generation + 1, instance.id, job.generation, id);
|
||||
} else if (instance.generation !== job.generation && !(job.action === 'rebuild' && instance.generation === job.generation + 1)) return;
|
||||
applyReceipt(job, (operation.result ?? {}) as Record<string, unknown>);
|
||||
db().prepare('UPDATE native_provisioning SET applied_at=COALESCE(applied_at,?) WHERE deployment_id=?').run(operation.updatedAt, id);
|
||||
}
|
||||
|
||||
function hold(job: Job, message: string) { if (!stillActive(job)) return; deploymentStage(job.id, message, { blocker: { code: 'native-host', message, automatic: true } }); }
|
||||
function stillActive(job: Job): boolean {
|
||||
return !!db().prepare(`SELECT 1 FROM server_deployments WHERE id=? AND server_deleted_at IS NULL
|
||||
AND state IN ('queued','waiting','running','awaiting-health','interrupted')`).get(job.id);
|
||||
@@ -259,6 +318,26 @@ function complete(job: Job) {
|
||||
|
||||
export async function processNativeProvisioning(eligibility: (operation: { id: string; server: string; action: string; state: string; stage: string }) => string | null,
|
||||
options: { wait?: boolean } = {}): Promise<void> {
|
||||
ensureHostSchema(db());
|
||||
const cancelled = db().prepare(`SELECT d.id FROM server_deployments d
|
||||
JOIN native_provisioning n ON n.deployment_id=d.id
|
||||
LEFT JOIN ec_host_operations o ON o.id=n.operation_id
|
||||
LEFT JOIN ec_host_migrations m ON m.deployment_id=d.id
|
||||
LEFT JOIN ec_host_operations removal ON removal.id=m.remove_operation_id
|
||||
WHERE d.state='cancelled' AND d.server_deleted_at IS NULL AND (
|
||||
(o.state='completed' AND (n.applied_at IS NULL OR d.action='destroy'))
|
||||
OR (o.state='failed' AND EXISTS (SELECT 1 FROM staged_file_changes f JOIN provisioning_sources s
|
||||
ON s.kind=f.source_kind AND s.source_id=f.source_id AND s.server=f.server
|
||||
WHERE s.operation_id=d.id AND f.state='applying'))
|
||||
OR (m.removed_at IS NULL AND (removal.state='completed' OR (removal.state='failed'
|
||||
AND EXISTS (SELECT 1 FROM ec_host_ports p WHERE p.host_id=m.target_host_id AND p.instance_id=m.instance_id))))
|
||||
)`).all() as { id: string }[];
|
||||
for (const row of cancelled) {
|
||||
try { reconcileNativeProvisioningReceipt(row.id); }
|
||||
catch (error) {
|
||||
deploymentStage(row.id, 'Cancelled; waiting to reconcile the delivered command: ' + (error as Error).message);
|
||||
}
|
||||
}
|
||||
reconcileHostDrains();
|
||||
const jobs = db().prepare(`SELECT d.*,n.instance_id,n.generation,n.stage_operation_id,n.operation_id,n.files_json,n.applied_at,q.selection
|
||||
FROM server_deployments d LEFT JOIN native_provisioning n ON n.deployment_id=d.id
|
||||
@@ -346,6 +425,8 @@ async function placeQueuedCreation(job: Job, instance: NativeInstance | null): P
|
||||
|
||||
async function processJob(job: Job, eligibility: (job: Job) => string | null) {
|
||||
if (!stillActive(job)) return;
|
||||
const unsettled = unsettledCancelledCommand(job.server, job.id);
|
||||
if (unsettled) { hold(job, 'Waiting for the already-delivered command from cancelled operation ' + unsettled + ' to settle.'); return; }
|
||||
let instance = nativeInstance(job.server);
|
||||
if (job.action === 'create' && !job.operation_id && !job.stage_operation_id && !job.submitted_at) {
|
||||
// A resumed bound job owns its original identity. An unplaced intent
|
||||
@@ -514,6 +595,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
|
||||
bindAddress: instance.bind_address, excludeInstanceId: instance.id, thorough: false, dispatch: true });
|
||||
if (job.action === 'create' || job.action === 'rebuild') assertReadyLibrary(instance.host_id, instance.library_id);
|
||||
const checkOccupancy = async (): Promise<boolean> => {
|
||||
if (!stillActive(job)) return false;
|
||||
// Fast controller check keeps occupied lobbies out of the operation slots.
|
||||
// The agent repeats this immediately before its local mutation.
|
||||
deploymentStage(job.id, 'Checking container and player occupancy before submission', { phase: 'eligibility', blocker: null });
|
||||
|
||||
Reference in New Issue
Block a user