From 2adadb0d0549feca189bfde88eda9f5128f92b39 Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Sun, 4 Oct 2026 14:50:59 -0400 Subject: [PATCH] Fix native provisioning scheduling and progress reporting --- docs/native-hosts.md | 26 ++++ src/lib/provisioning-detail.ts | 4 + src/lib/server/backends/native.ts | 4 +- src/lib/server/fleet/drift.ts | 96 +++++++++----- src/lib/server/hosts/index.ts | 8 +- src/lib/server/pipeline/activity.ts | 16 +++ src/lib/server/pipeline/worker.ts | 52 +++++--- src/lib/server/provisioning/agent-events.ts | 27 ++++ src/lib/server/provisioning/history.ts | 7 +- src/lib/server/provisioning/native.ts | 94 ++++++++++--- src/lib/server/provisioning/progress.ts | 2 +- src/lib/server/provisioning/worker.ts | 28 +++- src/lib/ui/provisioning-presentation.ts | 25 +++- .../api/agent/operations/[id]/+server.ts | 11 +- src/routes/provisioning/[id]/+page.svelte | 23 ++++ tests/unit/drift-kinds.test.ts | 31 +++++ tests/unit/native-provisioning.test.ts | 123 ++++++++++++++++++ tests/unit/worker-scheduling.test.ts | 76 +++++++++++ 18 files changed, 569 insertions(+), 84 deletions(-) create mode 100644 src/lib/server/pipeline/activity.ts create mode 100644 src/lib/server/provisioning/agent-events.ts create mode 100644 tests/unit/worker-scheduling.test.ts diff --git a/docs/native-hosts.md b/docs/native-hosts.md index f5d06a0..a09b196 100644 --- a/docs/native-hosts.md +++ b/docs/native-hosts.md @@ -374,3 +374,29 @@ coverage is not evidence that a real T5/T6 server has been deployed here. Run: Legacy backend code remains for existing database registrations. New native launches do not require its configuration, API, network or mounted directories. + +### Provisioning progress and diagnostics + +Native launches distinguish initial eligibility evaluation from an actual wait +for startup capacity. The operation page records payload preparation, command +submission, agent acknowledgement and execution, and separate container, game, +map and Plutonium revision checks. Native actions use host-agent steps rather +than legacy Komodo deployment steps. Evidence remains available after completion. + +The **Host agent execution** panel reads the durable command state directly, +including its queue time and latest report time. Agent receipts are recorded in +the provisioning history when received; they do not wait for a fleet scan. +An agent command marked completed still requires EC readiness verification. +The **Controller activity** panel shows current observations and their elapsed +time while the operation is active. It is a live snapshot, not historical proof +of which task delayed a previous deployment. + +Campaign, beta and maintenance observers run independently of provisioning. +New requests and completed agent commands wake provisioning; an arrival during a +pass schedules another pass. File drift scans use bounded batches of four reads +per scan to reduce agent polling round trips. Startup and disruption safety gates +remain enforced at command admission. + +Older completed operations whose progress snapshots were already discarded +cannot recover missing check results from this change. Their surviving events +and host command records remain available without inventing verification times. diff --git a/src/lib/provisioning-detail.ts b/src/lib/provisioning-detail.ts index 1897bdb..4d392fd 100644 --- a/src/lib/provisioning-detail.ts +++ b/src/lib/provisioning-detail.ts @@ -95,6 +95,9 @@ export interface ProvisioningCapacity { }[]; } export interface ProvisioningDetail { + backend?: 'native' | 'komodo'; + workerActivity?: WorkerActivity[]; + agentOperation?: { id: string; hostId: string; state: string; updatedAt: string; createdAt: string; error: string | null } | null; deployment: ProvisioningDeployment; events: ProvisioningEvent[]; activity: { total: number; latestId: number | null; failureExcerpt: ProvisioningEvent | null }; @@ -111,6 +114,7 @@ export interface ProvisioningDetail { warningMessages: string[]; generatedAt: string; } +export interface WorkerActivity { key: string; label: string; startedAt: string } export interface ProvisioningEventPage { events: ProvisioningEvent[]; total: number; diff --git a/src/lib/server/backends/native.ts b/src/lib/server/backends/native.ts index d455a6e..4e4e6f4 100644 --- a/src/lib/server/backends/native.ts +++ b/src/lib/server/backends/native.ts @@ -1,6 +1,7 @@ import { randomUUID } from 'node:crypto'; import { db } from '../db/index.js'; import { getHost, queueHostOperation, waitHostOperation } from '../hosts/index.js'; +import { trackActivity } from '../pipeline/activity.js'; export interface NativeInstance { id: string; server: string; host_id: string; generation: number; library_id: string; @@ -23,7 +24,8 @@ export async function nativeRequest>(instance: Nativ const operation = queueHostOperation(instance.host_id, kind, payload, { instanceId: instance.id, generation: instance.generation, idempotencyKey: key ?? randomUUID() }); - const result = await waitHostOperation(operation.id, 25_000); + const result = await trackActivity(`agent:${operation.id}`, `Waiting for ${kind} on ${instance.server}`, + () => waitHostOperation(operation.id, 25_000)); if (result.state !== 'completed') throw new Error(result.error || `Agent operation ${result.id} is ${result.state}; its original ID is retained.`); // Do not apply replies from a retired generation to a replacement server. const current = nativeInstance(instance.server); diff --git a/src/lib/server/fleet/drift.ts b/src/lib/server/fleet/drift.ts index da2611f..409fb19 100644 --- a/src/lib/server/fleet/drift.ts +++ b/src/lib/server/fleet/drift.ts @@ -13,6 +13,25 @@ import { selectedZombieMap } from './map-aliases.js'; import { getServer, listServers } from './servers.js'; import { promotedArtifacts } from './promoted-artifacts.js'; import { queueFileChange } from './staged-files.js'; +import { trackActivity } from '../pipeline/activity.js'; + +/** Queue a small batch for the agent's next poll, preserving input order. */ +async function scanBatch(items: readonly T[], work: (item: T) => Promise): Promise { + const results: R[] = new Array(items.length); + let cursor = 0; + let failed = false; + async function pump() { + while (!failed && cursor < items.length) { + const index = cursor++; + try { results[index] = await work(items[index]); } + catch (error) { failed = true; throw error; } + } + } + const workers = await Promise.allSettled(Array.from({ length: Math.min(4, items.length) }, pump)); + const failure = workers.find(result => result.status === 'rejected'); + if (failure?.status === 'rejected') throw failure.reason; + return results; +} // Destinations drift scans, in a stable order so the UI shows scripts first. // Baseline shares the config destination — one physical dir, two logical kinds. @@ -71,22 +90,26 @@ async function readExistingFile(path: string): Promise { async function walkFiles(root: string): Promise { const out: string[] = []; - async function recur(dir: string): Promise { - let entries; - try { - entries = await readdir(dir, { withFileTypes: true }); - } catch (error) { - if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; - throw error; - } - for (const e of entries) { - if (DRIFT_IGNORE_FILENAMES.has(e.name)) continue; - const abs = join(dir, e.name); - if (e.isDirectory()) await recur(abs); - else if (e.isFile()) out.push(toPosix(relative(root, abs))); - } + let directories = [root]; + while (directories.length) { + const children = await scanBatch(directories, async dir => { + let entries; + try { entries = await readdir(dir, { withFileTypes: true }); } + catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return []; + throw error; + } + const next: string[] = []; + for (const entry of entries) { + if (DRIFT_IGNORE_FILENAMES.has(entry.name)) continue; + const abs = join(dir, entry.name); + if (entry.isDirectory()) next.push(abs); + else if (entry.isFile()) out.push(toPosix(relative(root, abs))); + } + return next; + }); + directories = children.flat(); } - await recur(root); return out.sort(); } @@ -98,6 +121,10 @@ async function walkFiles(root: string): Promise { * artifacts with the promoted versions. */ export async function scanServerDrift(server: string): Promise { + return trackActivity(`drift:${server}`, `Checking files on ${server}`, () => scanServerFiles(server)); +} + +async function scanServerFiles(server: string): Promise { const srv = getServer(server); if (!srv || srv.destroyed_at || hasActiveBeta(server)) return []; // A missing mount is unavailable evidence, not an empty managed tree. @@ -121,27 +148,28 @@ export async function scanServerDrift(server: string): Promise { const onDisk = await walkFiles(dest); if (onDisk.length === 0) continue; - for (const relPath of onDisk) { - if (managed.has(`${kind}:${relPath}`)) continue; - if (keepDefaultConfig && kind === 'config' && relPath === configFileOf(server)) continue; + const untracked = await scanBatch(onDisk, async (relPath): Promise => { + if (managed.has(`${kind}:${relPath}`)) return null; + if (keepDefaultConfig && kind === 'config' && relPath === configFileOf(server)) return null; if (kind === 'config' && relPath === MAP_CONFIG_FILE && srv?.created_by - && srv.game_mode === 'zombie' && getServerMapConfig(srv.game, srv.map, srv.map_slug)) continue; + && srv.game_mode === 'zombie' && getServerMapConfig(srv.game, srv.map, srv.map_slug)) return null; const abs = join(dest, relPath); try { - const st = await stat(abs); const buf = await readFile(abs); - drift.push({ + return { server, relPath, - size: st.size, + size: buf.byteLength, sha256: sha256Buffer(buf), kind, driftKind: 'untracked' - }); + }; } catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; // A file disappeared between walk and stat; the next scan sees it. } - } + return null; + }); + drift.push(...untracked.filter((entry): entry is DriftEntry => entry !== null)); } await requireServerDirectory(server); @@ -171,18 +199,19 @@ async function scanManagedForServer(server: string): Promise { // Inventory records the last managed write, not the current disk bytes. // Check both so edits and outdated versions are detected on tracked files. - for (const { kind, script, version } of promotedArtifacts(srv.game, srv.map)) { + const managed = await scanBatch(promotedArtifacts(srv.game, srv.map), async ({ kind, script, version }): Promise => { const content = await readExistingFile(join(destOfKind(server, kind), script.target_rel_path)); const inventory = getServerScript(server, script.target_rel_path); if (content?.equals(version.content) && inventory?.sha256 === version.sha256 - && inventory.script_version_id === version.id) continue; - out.push({ + && inventory.script_version_id === version.id) return null; + return { server, relPath: script.target_rel_path, kind, driftKind: content ? 'replace' : 'missing', size: version.size, sha256: version.sha256, scriptId: script.id, scriptVersionId: version.id - }); - } + }; + }); + out.push(...managed.filter((entry): entry is DriftEntry => entry !== null)); // Composed cfg: the effective dedicated_sp.cfg is baseline + all promoted // overlays for this server's scope. It's "missing" when either the file @@ -329,15 +358,14 @@ export interface PreparedDriftCampaign { export async function prepareDriftCampaign(server: string): Promise { const servers = driftTargets([server]); const entries = await scanServerDrift(server); - const files: PreparedDriftCampaign['files'] = []; - for (const entry of entries) { + const files = await scanBatch(entries, async entry => { const before = await readExistingFile(join(destOfKind(server, entry.kind), entry.relPath)); const after = entry.driftKind === 'untracked' ? null : await resolveManagedEntry(server, entry); if (entry.driftKind !== 'untracked' && !after) throw new Error(`Cannot resolve promoted content for ${entry.relPath}`); - files.push({ before, after, file: { server, relPath: entry.relPath, kind: entry.kind, driftKind: entry.driftKind, + return { before, after, file: { server, relPath: entry.relPath, kind: entry.kind, driftKind: entry.driftKind, beforeSha256: before ? sha256Buffer(before) : null, afterSha256: after ? sha256Buffer(after) : null, - beforeSize: before?.byteLength ?? null, afterSize: after?.byteLength ?? null } }); - } + beforeSize: before?.byteLength ?? null, afterSize: after?.byteLength ?? null } }; + }); await requireServerDirectory(server); if (hasActiveBeta(server)) throw new Error('Active beta owns this server; reconciliation deferred'); return { plan: buildPlan(servers, files.map(({ file }) => file)), files }; diff --git a/src/lib/server/hosts/index.ts b/src/lib/server/hosts/index.ts index dabf90a..9a5ab33 100644 --- a/src/lib/server/hosts/index.ts +++ b/src/lib/server/hosts/index.ts @@ -1,3 +1,4 @@ +import { recordAgentReceipt } from '../provisioning/agent-events.js'; import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypto'; import { isIP } from 'node:net'; import { db } from '../db/index.js'; @@ -342,7 +343,12 @@ export function reportHostOperation(hostId: string, id: string, input: { state: if (state === 'failed') database().prepare("UPDATE ec_host_decommissions SET stage='failed',error=?,updated_at=? WHERE operation_id=?").run(error ?? 'The agent could not start cleanup. Repair the installation, then retry deletion.', now(), id); else database().prepare("UPDATE ec_host_decommissions SET stage='running',updated_at=? WHERE operation_id=? AND stage IN ('queued','running')").run(now(), id); } - if (operation.state !== state) database().prepare('INSERT INTO ec_host_operation_events(operation_id,state,created_at,error) VALUES(?,?,?,?)').run(id, state, now(), error); + if (operation.state !== state) { + database().prepare('INSERT INTO ec_host_operation_events(operation_id,state,created_at,error) VALUES(?,?,?,?)').run(id, state, now(), error); + if (operation.kind.startsWith('instance.') && operation.kind !== 'instance.inspect' || operation.kind === 'files.stage') { + recordAgentReceipt(id, state, now()); + } + } return getHostOperation(id)!; })(); } diff --git a/src/lib/server/pipeline/activity.ts b/src/lib/server/pipeline/activity.ts new file mode 100644 index 0000000..f569906 --- /dev/null +++ b/src/lib/server/pipeline/activity.ts @@ -0,0 +1,16 @@ +import type { WorkerActivity } from '../../provisioning-detail.js'; + +const active = new Map(); +let sequence = 0; + +/** Live diagnostics only. Durable operation milestones belong in provisioning history. */ +export function workerActivity(): WorkerActivity[] { + return [...active.values()].map(value => ({ ...value })); +} + +export async function trackActivity(key: string, label: string, work: () => Promise): Promise { + if (active.has(key)) key = `${key}:${++sequence}`; + active.set(key, { key, label, startedAt: new Date().toISOString() }); + try { return await work(); } + finally { active.delete(key); } +} diff --git a/src/lib/server/pipeline/worker.ts b/src/lib/server/pipeline/worker.ts index 79d8927..f93797e 100644 --- a/src/lib/server/pipeline/worker.ts +++ b/src/lib/server/pipeline/worker.ts @@ -16,6 +16,7 @@ import { STATUS_REFRESH_MS } from '../../server-freshness.js'; import { recordMetrics } from '../fleet/metrics.js'; import { selectedZombieMap } from '../../map-aliases.js'; import { runMaintenance } from '../maintenance/index.js'; +import { trackActivity } from './activity.js'; const TICK_MS = 15_000; // Cap concurrent UDP RCON calls so a 15-server fleet doesn't blast the @@ -27,26 +28,38 @@ let tickTimer: NodeJS.Timeout | null = null; let statusTimer: NodeJS.Timeout | null = null; let tickInFlight: Promise | null = null; +let tickRequested = false; +const requesterFlights = new Map>(); -function tick(): Promise { +function tick(requestAnother = false): Promise { + if (requestAnother) tickRequested = true; if (tickInFlight) return tickInFlight; - tickInFlight = runTick().finally(() => { tickInFlight = null; }); + tickInFlight = (async () => { + do { + tickRequested = false; + try { await trackActivity('provisioning', 'Processing provisioning requests', () => processProvisioning({ wait: false })); } + catch (err) { console.error('[worker] provisioning error:', err); } + } while (tickRequested); + })().finally(() => { tickInFlight = null; }); return tickInFlight; } -async function reconcileRequesters(): Promise { - try { await sweep(); } catch (err) { console.error('[worker] campaign error:', err); } - try { await betaTick(); } catch (err) { console.error('[worker] beta error:', err); } - try { await runMaintenance(); } catch (err) { console.error('[worker] maintenance error:', err); } -} - -async function runTick(): Promise { - // Callers submit durable requests first. Exactly one owner admits and runs - // lifecycle work, then callers observe the resulting jobs by ID. - await reconcileRequesters(); - try { await processProvisioning(); } - catch (err) { console.error('[worker] provisioning error:', err); } - await reconcileRequesters(); +function reconcileRequesters(): void { + // Observers submit durable intent independently. Provisioning still owns + // admission and rechecks source eligibility immediately before mutation. + const requesters: [string, string, () => Promise][] = [ + ['campaign', 'Reconciling campaigns', sweep], + ['beta', 'Reconciling beta tests', betaTick], + ['maintenance', 'Checking maintenance and recovery', runMaintenance] + ]; + for (const [key, label, work] of requesters) { + if (requesterFlights.has(key)) continue; + const flight = trackActivity(key, label, work) + .catch(err => { console.error(`[worker] ${key} error:`, err); }) + .then(async () => { await tick(true); }) + .finally(() => { requesterFlights.delete(key); }); + requesterFlights.set(key, flight); + } } /** @@ -185,8 +198,9 @@ export function startWorker(): void { started = true; void tick(); + reconcileRequesters(); void fanoutStatus(); - tickTimer = setInterval(tick, TICK_MS); + tickTimer = setInterval(() => { void tick(); reconcileRequesters(); }, TICK_MS); statusTimer = setInterval(fanoutStatus, STATUS_REFRESH_MS); console.log('[worker] started — sweep every', TICK_MS, 'ms, status every', STATUS_REFRESH_MS, 'ms'); @@ -205,5 +219,9 @@ export function stopWorker(): void { * fires immediately after a state change instead of waiting for the next tick. */ export async function kick(): Promise { - await tick(); + reconcileRequesters(); + await tick(true); } + +/** Agent receipts need observation promptly, without launching another fleet scan. */ +export async function kickProvisioning(): Promise { await tick(true); } diff --git a/src/lib/server/provisioning/agent-events.ts b/src/lib/server/provisioning/agent-events.ts new file mode 100644 index 0000000..8ed4fde --- /dev/null +++ b/src/lib/server/provisioning/agent-events.ts @@ -0,0 +1,27 @@ +import { db } from '../db/index.js'; +import { recordProgress } from './progress.js'; + +/** Record receipts in the report transaction, independently of the execution worker. */ +export function recordAgentReceipt(operationId: string, state: string, at: string): void { + // Host operations also serve installations that have no provisioning schema yet. + if (!db().prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name='native_provisioning'").get()) return; + const job = db().prepare(`SELECT d.id, d.phase_key, n.operation_id FROM native_provisioning n + JOIN server_deployments d ON d.id=n.deployment_id + WHERE (n.operation_id=? OR n.stage_operation_id=?) AND d.server_deleted_at IS NULL + AND d.state IN ('queued','waiting','running','awaiting-health','interrupted')`).get(operationId, operationId) as + { id: string; phase_key: string | null; operation_id: string | null } | undefined; + if (!job) return; + const staging = job.operation_id !== operationId; + const message = `${staging ? 'File staging' : 'Host agent command'} ${state} (${operationId})`; + if (!staging && job.phase_key === 'container-action') { + db().prepare('UPDATE server_deployments SET stage=? WHERE id=?').run(`Host agent command ${state}`, job.id); + } + recordProgress(job.id, { + blocker: null, + event: { kind: 'agent', message, details: { operationId, state, receivedAt: at } }, + ...(['accepted', 'running', 'completed'].includes(state) ? { + checkpoint: { key: `${staging ? 'staging' : 'agent'}-${state}`, label: message, at }, + ...(!staging ? { command: { state: 'accepted' as const, message: 'Host agent acknowledged the command.', at } } : {}) + } : {}) + }); +} diff --git a/src/lib/server/provisioning/history.ts b/src/lib/server/provisioning/history.ts index 25dfb79..045ea94 100644 --- a/src/lib/server/provisioning/history.ts +++ b/src/lib/server/provisioning/history.ts @@ -7,6 +7,7 @@ import { fleetSettings } from '../settings.js'; import { activitySummary, deploymentEvents, latestFailureEvent } from './activity.js'; import { readProgress, recordProgress, type ProgressUpdate } from './progress.js'; import { bindNativeOperation } from './native.js'; +import { workerActivity } from '../pipeline/activity.js'; export function beginDeployment(server: string, action: ProvisioningAction, triggeredBy = 'ui'): string { const id = randomUUID(); @@ -101,6 +102,9 @@ export function getDeployment(id: string, options: { events?: 'all' | 'recent' } (ProvisioningDeployment & { observation_json: string | null; phase_key: string | null; phase_started_at: string | null; observed_at: string | null }) | undefined; if (!row) return null; const generatedAt = new Date().toISOString(); + const native = db().prepare('SELECT instance_id FROM native_provisioning WHERE deployment_id=?').get(id); + const agentOperation = native ? db().prepare(`SELECT o.id, o.host_id AS hostId, o.state, o.updated_at AS updatedAt, o.created_at AS createdAt, o.error + FROM native_provisioning n JOIN ec_host_operations o ON o.id=n.operation_id WHERE n.deployment_id=?`).get(id) as ProvisioningDetail['agentOperation'] : null; const { observation_json: _snapshot, phase_key: _phase, phase_started_at: _phaseStarted, observed_at: _observed, ...deployment } = row; const sources = (db().prepare('SELECT kind, source_id, server FROM provisioning_sources WHERE operation_id = ? ORDER BY kind, source_id') .all(id) as Omit[]).map(sourceDetails); @@ -109,7 +113,8 @@ export function getDeployment(id: string, options: { events?: 'all' | 'recent' } SELECT collection_id, deployment_id FROM server_collection_members UNION ALL SELECT collection_id, deployment_id FROM server_collection_removals) m JOIN server_launch_collections c ON c.id = m.collection_id WHERE m.deployment_id = ? ORDER BY c.created_at, c.id LIMIT 1`).get(id) as { id: string; name: string } | undefined; - return { deployment, events: deploymentEvents(id, options.events === 'recent'), + return { deployment, backend: native ? 'native' as const : 'komodo' as const, agentOperation: agentOperation ?? null, + workerActivity: workerActivity(), events: deploymentEvents(id, options.events === 'recent'), activity: { ...activitySummary(id), failureExcerpt: latestFailureEvent(id) }, sources, recipe: launch ? summarizeRecipe(launch.recipe) : null, blueprintName: launch?.blueprintName ?? null, canResumeVerification: canResumeVerification(id), canResumeCleanup: canResumeCleanup(id), diff --git a/src/lib/server/provisioning/native.ts b/src/lib/server/provisioning/native.ts index ef25d9e..793dde2 100644 --- a/src/lib/server/provisioning/native.ts +++ b/src/lib/server/provisioning/native.ts @@ -10,7 +10,8 @@ import { getServerKey, saveServerKey } from '../fleet/server-keys.js'; import { activeBetaRemovalBlocker, getServer } from '../fleet/servers.js'; import { nextLobbyNumber, type CreateServerInput, type CreateServerResult } from './instances.js'; import { beginDeployment, deploymentStage, finishDeployment, REMOVING_SERVER } from './history.js'; -import { recordProgress } from './progress.js'; +import { recordProgress, type CheckInput } from './progress.js'; +import { trackActivity } from '../pipeline/activity.js'; import { fleetSettings } from '../settings.js'; import { resolveRecoveredHealthOperation } from '../maintenance/recovery.js'; import { probeServerStatus, probePlayerEvidence } from '../fleet/status-evidence.js'; @@ -126,7 +127,7 @@ export function queueNativeCreation(input: CreateServerInput): CreateServerResul .run(id, name, port, 'zombie', recipe.mapSlug, recipe.expectedMap); db().prepare('INSERT INTO server_launch_recipes(deployment_id,recipe,blueprint_name) VALUES(?,?,?)').run(id, JSON.stringify(recipe), preview?.blueprint_name ?? null); if (preview) db().prepare('UPDATE server_launch_previews SET deployment_id=? WHERE id=?').run(id, preview.id); - deploymentStage(id, 'Queued for native host deployment', { phase: 'startup-queue', recipeStatus: 'queued' }); + deploymentStage(id, 'Waiting for provisioning evaluation', { phase: 'eligibility', recipeStatus: 'queued' }); return { ok: true as const, name, port, deploymentId: id }; })(); } catch (error) { return { ok: false, kind: 'bad-request', message: (error as Error).message }; } @@ -230,6 +231,10 @@ function createPayload(instance: NativeInstance, recipe: LaunchRecipe) { } function hold(job: Job, message: string) { 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); +} function complete(job: Job) { db().transaction(() => { db().prepare('UPDATE native_provisioning SET verified_at=? WHERE deployment_id=?').run(new Date().toISOString(), job.id); @@ -238,7 +243,8 @@ function complete(job: Job) { })(); } -export async function processNativeProvisioning(eligibility: (operation: { id: string; server: string; action: string; state: string; stage: string }) => string | null): Promise { +export async function processNativeProvisioning(eligibility: (operation: { id: string; server: string; action: string; state: string; stage: string }) => string | null, + options: { wait?: boolean } = {}): Promise { // Draining prevents admission immediately and retires existing lobbies only through the normal safety queue. for (const host of listHosts().filter(host => host.mode === 'draining' && host.connected)) { const instances = db().prepare('SELECT * FROM native_instances WHERE host_id=? AND retired_at IS NULL').all(host.id) as NativeInstance[]; @@ -253,12 +259,23 @@ export async function processNativeProvisioning(eligibility: (operation: { id: s const jobs = db().prepare(`SELECT d.*,n.instance_id,n.generation,n.stage_operation_id,n.operation_id,n.files_json,n.applied_at FROM native_provisioning n JOIN server_deployments d ON d.id=n.deployment_id WHERE d.server_deleted_at IS NULL AND d.state IN ('queued','waiting','running','awaiting-health','interrupted') ORDER BY d.started_at,d.rowid`).all() as Job[]; - for (const job of jobs) { - try { await processJob(job, eligibility); } - catch (error) { hold(job, (error as Error).message); } - } + // One host's observation must not prevent another job being admitted. + // Per-job ownership also covers overlapping direct calls and worker passes. + const work = jobs.map(job => { + const pending = activeJobs.get(job.id); + if (pending) return pending; + const work = trackActivity(`provisioning:${job.id}`, `Evaluating ${job.action} for ${job.server}`, + () => processJob(job, eligibility)) + .catch(error => { if (stillActive(job)) hold(job, (error as Error).message); }) + .finally(() => { activeJobs.delete(job.id); }); + activeJobs.set(job.id, work); + return work; + }); + if (options.wait !== false) await Promise.all(work); } +const activeJobs = new Map>(); + async function processJob(job: Job, eligibility: (job: Job) => string | null) { let instance = nativeInstance(job.server); if (!instance || instance.id !== job.instance_id || (instance.generation !== job.generation && !(job.action === 'rebuild' && instance.generation === job.generation + 1))) { @@ -289,10 +306,11 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { } if (job.action === 'destroy') { db().transaction(() => { - deploymentStage(job.id, REMOVING_SERVER, { checkpoint: { key: 'delete-deployment', label: 'Agent confirmed instance and container removal' } }); + deploymentStage(job.id, REMOVING_SERVER, { phase: 'cleanup', checkpoint: { key: 'container-action', label: 'Agent confirmed instance and container removal', at: operation.updatedAt } }); db().prepare('UPDATE native_instances SET retired_at=? WHERE id=?').run(new Date().toISOString(), instance!.id); releaseHostPorts(instance!.id); db().prepare('DELETE FROM servers WHERE name=?').run(job.server); + recordProgress(job.id, { checkpoint: { key: 'cleanup', label: 'Instance retired and port reservations released' } }); complete(job); })(); return; @@ -304,28 +322,56 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { applyReceipt(job, (operation.result ?? {}) as Record); if (job.state !== 'awaiting-health') { db().prepare("UPDATE server_deployments SET state='awaiting-health',error=NULL,finished_at=NULL WHERE id=?").run(job.id); - recordProgress(job.id, { command: { state: 'accepted', message: `Agent confirmed ${job.action}; verifying the resulting instance.` }, phase: 'verify-container' }); + recordProgress(job.id, { command: { state: 'accepted', message: `Agent confirmed ${job.action}; verifying the resulting instance.` }, + checkpoint: { key: 'container-action', label: `Agent confirmed ${job.action}`, at: operation.updatedAt } }); } + const checks: CheckInput[] = []; + const check = (value: CheckInput, phase: import('../../provisioning-detail.js').ProvisioningPhase) => { + checks.push({ ...value, checkedAt: new Date().toISOString() }); + recordProgress(job.id, { checks, ...(value.result === 'pass' ? { checkpoint: { key: phase, label: value.label } } : {}) }); + }; + deploymentStage(job.id, 'Checking container state with host agent', { phase: job.action === 'stop' ? 'verify-stopped' : 'verify-container', blocker: null, checks }); const health = await nativeRequest<{ healthy: boolean; state: string }>(instance, 'instance.inspect'); + if (!stillActive(job)) return; if (job.action === 'stop') { - if (!['stopped','exited','created','dead'].includes(health.state)) { hold(job, 'Waiting for the agent to verify a stopped container.'); return; } + const stopped = ['stopped','exited','created','dead'].includes(health.state); + check({ key: 'stopped', label: 'Stopped container verified', source: 'agent', result: stopped ? 'pass' : 'waiting', actual: health.state }, 'verify-stopped'); + if (!stopped) { hold(job, 'Waiting for the agent to verify a stopped container.'); return; } db().prepare("UPDATE servers SET container_state=?,last_players=NULL,last_roster=NULL WHERE name=?").run(health.state, job.server); complete(job); return; } + check({ key: 'container', label: 'Container readiness verified', source: 'agent', result: health.healthy ? 'pass' : 'waiting', actual: health.state }, 'verify-container'); if (!health.healthy) { hold(job, `Waiting for container readiness: ${health.state}`); return; } + deploymentStage(job.id, 'Checking game status', { phase: 'verify-game', blocker: null }); const status = await probeServerStatus(job.server, operation.updatedAt); + if (!stillActive(job)) return; + check({ key: 'game', label: 'Fresh game status verified', source: status.ok ? status.source : null, + result: status.ok ? 'pass' : 'unknown', actual: status.ok ? status.players : null, + message: status.ok ? null : status.reason }, 'verify-game'); if (!status.ok) { hold(job, `Waiting for game status: ${status.reason}`); return; } if (job.action === 'create' || job.action === 'rebuild') { + deploymentStage(job.id, 'Checking selected engine map', { phase: 'verify-map', blocker: null }); const recipe = recipeForDeployment(job.id)!.recipe; + check({ key: 'map', label: 'Selected engine map verified', source: status.source, + result: status.map === recipe.expectedMap ? 'pass' : 'waiting', expected: recipe.expectedMap, actual: status.map }, 'verify-map'); if (status.map !== recipe.expectedMap) { hold(job, `Waiting for selected map ${recipe.expectedMap}; observed ${status.map ?? 'unknown'}.`); return; } + deploymentStage(job.id, 'Checking production Plutonium release', { phase: 'verify-plutonium', blocker: null }); await checkRelease(); - const revision = await probeVersion(job.server, { operationId: job.id }); const target = freshRelease(); + if (!stillActive(job)) return; + const target = freshRelease(); + checks.push({ key: 'plutonium-release', label: 'Production Plutonium release', source: 'manifest', + result: target === null ? 'unknown' : 'pass', actual: target }); + recordProgress(job.id, { checks }); + deploymentStage(job.id, 'Checking running Plutonium revision', { phase: 'verify-plutonium', blocker: null }); + const revision = await probeVersion(job.server, { operationId: job.id }); + if (!stillActive(job)) return; + check({ key: 'plutonium-version', label: 'Plutonium revision verified', source: 'rcon', expected: target, actual: revision, + result: revision === null || target === null ? 'unknown' : revision < target ? 'fail' : 'pass' }, 'verify-plutonium'); if (revision === null || target === null) { hold(job, 'Waiting to verify the Plutonium runtime revision.'); return; } if (revision < target) { finishDeployment(job.id, { ok: false, message: `Running Plutonium r${revision} is older than production r${target}.` }); return; } } - db().prepare("UPDATE servers SET container_state='running',container_checked_at=? WHERE name=?").run(new Date().toISOString(), job.server); - recordProgress(job.id, { blocker: null, checks: [{ key: 'container', label: 'Container readiness', source: 'agent', result: 'pass', actual: health.state }, - { key: 'game', label: 'Game status', source: status.source, result: 'pass', actual: status.players }] }); + db().prepare('UPDATE servers SET container_state=?,container_checked_at=? WHERE name=?').run(health.state, new Date().toISOString(), job.server); + recordProgress(job.id, { blocker: null, checks }); complete(job); return; } const reason = eligibility(job); @@ -350,16 +396,23 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { const settings = fleetSettings(); const count = (actions: string[]) => (db().prepare(`SELECT COUNT(*) AS n FROM server_deployments WHERE id != ? AND server_deleted_at IS NULL AND action IN (${actions.map(() => '?').join(',')}) AND state IN ('running','awaiting-health')`).get(job.id, ...actions) as { n: number }).n; - if ((job.action === 'create' || job.action === 'rebuild') && count(['create','rebuild']) >= settings.createConcurrency) { hold(job, 'Waiting for startup capacity.'); return; } + if ((job.action === 'create' || job.action === 'rebuild') && count(['create','rebuild']) >= settings.createConcurrency) { + deploymentStage(job.id, 'Waiting for startup capacity.', { phase: 'startup-queue', + blocker: { code: 'startup-capacity', message: 'Waiting for startup capacity.', automatic: true } }); return; + } if (job.action === 'destroy' && count(['destroy']) >= settings.destroyConcurrency) { hold(job, 'Waiting for destruction capacity.'); return; } + deploymentStage(job.id, 'Checking host eligibility', { phase: 'eligibility', blocker: null }); if (job.action === 'create') requireReadyHost(instance.host_id, { image: instance.image, requiredBytes: 1024 ** 3 }); if (job.action === 'create' || job.action === 'rebuild') requireReadyLibrary(instance.host_id, instance.library_id); if (job.action !== 'create' && job.action !== 'start') { // 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 }); const health = await nativeRequest<{ state: string; healthy?: boolean }>(instance, 'instance.inspect'); + if (!stillActive(job)) return; if (!['stopped','exited','created','dead','missing'].includes(health.state)) { const occupancy = await probePlayerEvidence(job.server); + if (!stillActive(job)) return; const recoveredWatchdog = health.healthy && occupancy.players !== null && db().prepare(`SELECT 1 FROM provisioning_sources s JOIN watchdog_incidents i ON i.id=s.source_id JOIN server_deployments d ON d.id=s.operation_id WHERE s.operation_id=? AND s.kind='watchdog' AND i.kind='health' AND d.triggered_by LIKE 'watchdog:%'`).get(job.id); @@ -379,12 +432,20 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { if (JSON.stringify(filesFor(job)) !== JSON.stringify(files)) { finishDeployment(job.id, { ok: false, message: 'File requests changed during preflight. Review and submit a fresh activation.' }); return; } + // Async occupancy checks can overlap another job claiming the final slot. + if ((job.action === 'create' || job.action === 'rebuild') && count(['create','rebuild']) >= settings.createConcurrency) { + deploymentStage(job.id, 'Waiting for startup capacity.', { phase: 'startup-queue', blocker: { code: 'startup-capacity', message: 'Waiting for startup capacity.', automatic: true } }); return; + } + if (job.action === 'destroy' && count(['destroy']) >= settings.destroyConcurrency) { hold(job, 'Waiting for destruction capacity.'); return; } if (job.action === 'start' || job.action === 'restart' || job.action === 'rebuild') { db().prepare(`INSERT INTO restart_rollout(server,requested_at,operation_id) VALUES(?,?,?) ON CONFLICT(server) DO UPDATE SET requested_at=excluded.requested_at,operation_id=excluded.operation_id,healthy_at=NULL,error=NULL`).run(job.server, new Date().toISOString(), job.id); } + recordProgress(job.id, { blocker: null, checkpoint: { key: 'eligibility', label: 'Host eligibility and safety checks passed' } }); let payload: Record = files.length ? { stageId: job.id } : {}; if (job.action === 'create' || job.action === 'rebuild') { + deploymentStage(job.id, 'Preparing launch files for the host agent', { phase: 'prepare', recipeStatus: 'attempted', + checkpoint: { key: 'startup-queue', label: 'Startup capacity available' } }); const prior = recipeForDeployment(job.id)!.recipe as NativeRecipe; const recipe: NativeRecipe = { ...captureRecipe(prior.game, prior.mapSlug), native: prior.native, image: instance!.image }; payload = createPayload(instance!, recipe); @@ -394,13 +455,14 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { VALUES(?,'deployment','',?,?,'zombie',?,?,?,?,?,?,'native','starting')`).run(job.server, recipe.game, recipe.expectedMap, recipe.mapSlug, serverDir(job.server), recipe.configFile, recipe.game === 't6' ? 't6zm' : 't5sp', recipe.image, instance!.port); for (const file of recipe.files) recordScript(job.server, file.path, file.versionId, file.sha256, file.size, 'native-create'); + recordProgress(job.id, { checkpoint: { key: 'prepare', label: 'Launch files selected and command payload prepared' } }); } const operation = queueHostOperation(instance!.host_id, `instance.${job.action === 'destroy' ? 'remove' : job.action}`, payload, { instanceId: instance!.id, generation: instance!.generation, idempotencyKey: job.id }); db().prepare('UPDATE native_provisioning SET operation_id=? WHERE deployment_id=?').run(operation.id, job.id); for (const file of files) db().prepare("UPDATE staged_file_changes SET state='applying' WHERE id=? AND state='pending'").run(file.id); db().prepare("UPDATE server_deployments SET state='running',submitted_at=? WHERE id=?").run(new Date().toISOString(), job.id); - deploymentStage(job.id, `Host agent ${job.action} queued`, { phase: 'container-action', blocker: null, + deploymentStage(job.id, `Waiting for host agent to acknowledge ${job.action}`, { phase: 'container-action', blocker: null, command: { state: 'attempted', message: `Durable agent operation ${operation.id}; reconciliation always uses this ID.` } }); })(); } diff --git a/src/lib/server/provisioning/progress.ts b/src/lib/server/provisioning/progress.ts index dd4aa81..c02d375 100644 --- a/src/lib/server/provisioning/progress.ts +++ b/src/lib/server/provisioning/progress.ts @@ -36,7 +36,7 @@ export function readProgress(row: { observation_json?: string | null }): Provisi || (v.blocker.label !== undefined && !nullableString(v.blocker.label)))) return null; if (v.checks.length > 10 || !v.checks.every(c => object(c) && ['container', 'game', 'map', 'plutonium-release', 'plutonium-version', 'occupancy', 'stopped'].includes(String(c.key)) && typeof c.label === 'string' && ['pass', 'waiting', 'fail', 'unknown'].includes(String(c.result)) - && (c.source === null || ['komodo', 'rcon', 'telemetry', 'combined', 'manifest'].includes(String(c.source))) + && (c.source === null || ['komodo', 'agent', 'rcon', 'telemetry', 'combined', 'manifest'].includes(String(c.source))) && typeof c.checkedAt === 'string' && nullableString(c.receivedAt) && scalar(c.expected) && scalar(c.actual) && nullableString(c.message))) return null; if (v.checkpoints.length > 100 || !v.checkpoints.every(c => object(c) && typeof c.key === 'string' && typeof c.label === 'string' && typeof c.at === 'string' && (c.phase === null || phase(c.phase)))) return null; diff --git a/src/lib/server/provisioning/worker.ts b/src/lib/server/provisioning/worker.ts index ecd2283..028f8d2 100644 --- a/src/lib/server/provisioning/worker.ts +++ b/src/lib/server/provisioning/worker.ts @@ -14,6 +14,7 @@ import type { ProvisioningBlocker } from '../../provisioning-detail.js'; import { recordProgress } from './progress.js'; import { nativeInstance, nativeDeployment } from '../backends/native.js'; import { processNativeProvisioning } from './native.js'; +import { trackActivity } from '../pipeline/activity.js'; const stoppedStates = new Set(['stopped', 'exited', 'created', 'dead']); export const isStopped = (state?: string) => stoppedStates.has(state?.toLowerCase() ?? ''); @@ -91,16 +92,26 @@ function blocker(server: string, id: string, action: RuntimeAction): Provisionin interface Operation { id: string; server: string; action: RuntimeAction; state: string; stage: string } let ticking: Promise | null = null; -export function processProvisioning(): Promise { - if (ticking) return ticking; - ticking = run().finally(() => { ticking = null; }); - return ticking; +let awaitedPass: Promise | null = null; +export function processProvisioning(options: { wait?: boolean } = {}): Promise { + if (options.wait !== false && awaitedPass) return awaitedPass; + // Native jobs own their individual flights. A slow observation must not + // prevent the next scheduler pass admitting newly arrived native work. + const native = nativePass(options); + if (!ticking) ticking = trackActivity('legacy-provisioning', 'Processing legacy provisioning', run) + .finally(() => { ticking = null; }); + if (options.wait === false) { + void ticking.catch(error => console.error('[worker] legacy provisioning error:', error)); + return native; + } + awaitedPass = Promise.all([native, ticking]).then(() => {}).finally(() => { awaitedPass = null; }); + return awaitedPass; } export const processLifecycle = processProvisioning; -async function run(): Promise { - await processNativeProvisioning(operation => { +function nativePass(options: { wait?: boolean }): Promise { + return processNativeProvisioning(operation => { if (operation.action === 'create') return null; if (operation.action === 'destroy') { const other = db().prepare(`SELECT id FROM server_deployments WHERE server=? AND id!=? AND server_deleted_at IS NULL @@ -112,7 +123,10 @@ async function run(): Promise { if (eligibility === 'cancelled') { cancelProvisioning(operation.id, 'Requester cancelled or replaced this operation'); return 'Requester cancelled this operation.'; } if (eligibility === 'held') return 'Waiting for the requester to resume.'; return blocker(operation.server, operation.id, typed.action)?.message ?? null; - }); + }, options); +} + +async function run(): Promise { await processCreations(); await processDestructions(); await refreshRestartRollout(); diff --git a/src/lib/ui/provisioning-presentation.ts b/src/lib/ui/provisioning-presentation.ts index d38a74f..838a677 100644 --- a/src/lib/ui/provisioning-presentation.ts +++ b/src/lib/ui/provisioning-presentation.ts @@ -19,7 +19,13 @@ const request: GroupDefinition = { key: 'request', label: 'Request and eligibili const creation: GroupDefinition = { key: 'create', label: 'Create and start', phases: ['startup-queue', 'prepare', 'create-deployment', 'container-action'] }; const removal: GroupDefinition = { key: 'remove', label: 'Remove the instance', phases: ['remove-container', 'delete-deployment', 'cleanup'] }; const readiness: GroupDefinition = { key: 'verify', label: 'Verify readiness', phases: ['verify-container', 'verify-game', 'verify-map', 'verify-plutonium'] }; -function definitions(action: ProvisioningAction): GroupDefinition[] { +function definitions(action: ProvisioningAction, native = false): GroupDefinition[] { + if (native) { + const command: GroupDefinition = { key: 'command', label: `${actionLabels[action]} on host agent`, phases: ['container-action'] }; + if (action === 'create' || action === 'rebuild') return [request, + { key: 'create', label: action === 'rebuild' ? 'Rebuild on host agent' : 'Create and start on host agent', phases: ['startup-queue', 'prepare', 'container-action'] }, readiness]; + if (action === 'destroy') return [request, command, { key: 'cleanup', label: 'Retire instance', phases: ['cleanup'] }]; + } if (action === 'create') return [{ ...request, phases: ['requested'] }, creation, readiness]; if (action === 'rebuild') return [request, removal, { ...creation, label: 'Create a fresh instance' }, readiness]; if (action === 'destroy') return [request, removal]; @@ -30,7 +36,7 @@ function definitions(action: ProvisioningAction): GroupDefinition[] { export function provisioningGroups(detail: ProvisioningDetail): ProgressGroup[] { const progress = detail.progress; if (!progress) return []; - const groups = definitions(detail.deployment.action); + const groups = definitions(detail.deployment.action, detail.backend === 'native'); const phases = groups.flatMap(group => group.phases); const current = progress.phase ? phases.indexOf(progress.phase) : -1; const active = activeProvisioningStates.includes(detail.deployment.state); @@ -48,7 +54,10 @@ export function provisioningGroups(detail: ProvisioningDetail): ProgressGroup[] : isCurrent && active ? progress.blocker || ['waiting', 'queued'].includes(detail.deployment.state) ? 'waiting' : 'current' : current >= 0 && index >= current && detail.deployment.state === 'cancelled' ? 'cancelled' : active && current >= 0 && index > current ? 'pending' : 'unknown'; - const label = key === 'container-action' && ['start', 'restart', 'stop'].includes(detail.deployment.action) + const label = detail.backend === 'native' && key === 'startup-queue' ? 'Admit to startup capacity' + : detail.backend === 'native' && key === 'prepare' ? 'Prepare launch payload' + : detail.backend === 'native' && key === 'container-action' ? `${actionLabels[detail.deployment.action]} instance through host agent` + : key === 'container-action' && ['start', 'restart', 'stop'].includes(detail.deployment.action) ? `${actionLabels[detail.deployment.action]} existing container` : phaseLabels[key]; return { key, label, state, at: check?.checkedAt ?? checkpoint?.at ?? (isCurrent ? progress.phaseStartedAt : null) }; }); @@ -93,7 +102,7 @@ export function provisioningOutcome(detail: ProvisioningDetail) { else if (progress?.blocker) description = progress.blocker.automatic ? 'The worker will recheck this condition and continue automatically when eligible.' : 'Review the blocking condition below before this operation can continue.'; - else if (operation.state === 'queued') description = 'The request is saved. It will start when its dependencies and startup capacity allow.'; + else if (operation.state === 'queued') description = 'The request is saved and awaiting provisioning evaluation. No capacity blocker has been recorded.'; else if (operation.state === 'waiting') description = 'Waiting for the worker’s eligibility checks. The recorded stage below describes the last known condition.'; else description = operation.action === 'destroy' ? 'Removal is in progress. Confirmed checkpoints and remaining cleanup are recorded below.' : operation.action === 'rebuild' ? 'The old instance is removed before a fresh instance is created in its reserved slot. Follow each checkpoint below.' @@ -151,6 +160,12 @@ export function isProvisioningEventPage(value: unknown): value is ProvisioningEv /** Validate every nested shape consumed by the page before replacing its last good snapshot. */ export function isProvisioningDetail(value: unknown, id: string): value is ProvisioningDetail { if (!record(value) || !record(value.deployment)) return false; + if (value.backend !== undefined && !['native', 'komodo'].includes(value.backend as string)) return false; + if (value.workerActivity !== undefined && (!Array.isArray(value.workerActivity) || !value.workerActivity.every(item => + record(item) && text(item.key) && text(item.label) && timestamp(item.startedAt)))) return false; + if (value.agentOperation !== undefined && value.agentOperation !== null && !(record(value.agentOperation) + && text(value.agentOperation.id) && text(value.agentOperation.hostId) && text(value.agentOperation.state) + && timestamp(value.agentOperation.updatedAt) && timestamp(value.agentOperation.createdAt) && nullableText(value.agentOperation.error))) return false; const operation = value.deployment; if (operation.id !== id || !provisioningActions.includes(operation.action as ProvisioningAction) || !['server', 'state', 'stage', 'triggered_by', 'warnings'].every(key => text(operation[key])) @@ -189,7 +204,7 @@ export function isProvisioningDetail(value: unknown, id: string): value is Provi if (!Array.isArray(progress.checks) || !progress.checks.every(check => record(check) && ['container', 'game', 'map', 'plutonium-release', 'plutonium-version', 'occupancy', 'stopped'].includes(check.key as string) && ['pass', 'waiting', 'fail', 'unknown'].includes(check.result as string) - && [null, 'komodo', 'rcon', 'telemetry', 'combined', 'manifest'].includes(check.source as null | string) + && [null, 'komodo', 'agent', 'rcon', 'telemetry', 'combined', 'manifest'].includes(check.source as null | string) && text(check.label) && timestamp(check.checkedAt) && nullableTime(check.receivedAt) && scalar(check.expected) && scalar(check.actual) && nullableText(check.message))) return false; if (!Array.isArray(progress.checkpoints) || !progress.checkpoints.every(checkpoint => record(checkpoint) diff --git a/src/routes/api/agent/operations/[id]/+server.ts b/src/routes/api/agent/operations/[id]/+server.ts index 10fb3e6..1a71333 100644 --- a/src/routes/api/agent/operations/[id]/+server.ts +++ b/src/routes/api/agent/operations/[id]/+server.ts @@ -1,8 +1,17 @@ import { json } from '@sveltejs/kit'; import { reportHostOperation } from '$lib/server/hosts/index.js'; import { authenticateAgent, hostApiError, objectBody } from '$lib/server/hosts/http.js'; +import { kickProvisioning } from '$lib/server/pipeline/worker.js'; export async function POST({ request, params }: { request: Request; params: { id: string } }) { const host = authenticateAgent(request); - try { const body = await objectBody(request, 128 * 1024 * 1024); const operation = reportHostOperation(host.id, params.id, { state: body.state, result: body.result, error: body.error }); return json({ operation: { id: operation.id, state: operation.state } }); } + try { + const body = await objectBody(request, 128 * 1024 * 1024); + const operation = reportHostOperation(host.id, params.id, { state: body.state, result: body.result, error: body.error }); + if (['completed', 'failed', 'unknown'].includes(operation.state) + && (operation.kind.startsWith('instance.') && operation.kind !== 'instance.inspect' || operation.kind === 'files.stage')) { + void kickProvisioning().catch(error => console.error('[agent] provisioning wake failed:', error)); + } + return json({ operation: { id: operation.id, state: operation.state } }); + } catch (cause) { return hostApiError(cause); } } diff --git a/src/routes/provisioning/[id]/+page.svelte b/src/routes/provisioning/[id]/+page.svelte index 46adb81..dcc43b9 100644 --- a/src/routes/provisioning/[id]/+page.svelte +++ b/src/routes/provisioning/[id]/+page.svelte @@ -112,6 +112,29 @@