From fbf06294867d4ea7a8a8d9b86a254ef28bea8e6b Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Tue, 6 Oct 2026 14:03:28 -0400 Subject: [PATCH] Separate staged command payloads from provisioning state --- src/lib/server/backends/schema.ts | 8 ++++++ src/lib/server/hosts/schema.ts | 9 ++++++ src/lib/server/provisioning/native.ts | 38 +++++++++++++++++--------- tests/unit/native-provisioning.test.ts | 12 ++++++++ 4 files changed, 54 insertions(+), 13 deletions(-) diff --git a/src/lib/server/backends/schema.ts b/src/lib/server/backends/schema.ts index 0736c54..7f3c696 100644 --- a/src/lib/server/backends/schema.ts +++ b/src/lib/server/backends/schema.ts @@ -43,6 +43,14 @@ export function migrateNativeInstances(handle: Database.Database): void { ); CREATE INDEX IF NOT EXISTS idx_ec_host_migrations_instance ON ec_host_migrations(instance_id); `); + if (!handle.prepare("SELECT 1 FROM settings WHERE key='native_file_plans_v1'").get()) { + handle.transaction(() => { + handle.exec(`UPDATE native_provisioning SET files_json=(SELECT json_group_array(json_remove(value,'$.contentBase64')) + FROM json_each(native_provisioning.files_json)) WHERE files_json IS NOT NULL; + INSERT INTO settings(key,value) VALUES('native_file_plans_v1','1');`); + })(); + } + } /** Accepted native creations receive their host port only when provisioning places them. */ diff --git a/src/lib/server/hosts/schema.ts b/src/lib/server/hosts/schema.ts index d46f89e..6bfcca3 100644 --- a/src/lib/server/hosts/schema.ts +++ b/src/lib/server/hosts/schema.ts @@ -122,6 +122,15 @@ export function ensureHostSchema(handle: Database.Database): void { CREATE INDEX IF NOT EXISTS ec_host_operations_important_history ON ec_host_operations(created_at) WHERE ${importantOperationSql.replaceAll('o.', '')}`); handle.exec('CREATE TABLE IF NOT EXISTS ec_host_schema_migrations (name TEXT PRIMARY KEY)'); + if (!handle.prepare('SELECT 1 FROM ec_host_schema_migrations WHERE name=?').get('staged-operation-payloads')) { + handle.transaction(() => { + handle.exec(`INSERT OR IGNORE INTO ec_host_operation_payloads(operation_id,payload) + SELECT id,payload FROM ec_host_operations WHERE kind='files.stage' AND json_type(payload,'$.files')='array'; + UPDATE ec_host_operations SET payload=json_object('stageId',json_extract(payload,'$.stageId')) + WHERE kind='files.stage' AND json_type(payload,'$.files')='array';`); + handle.prepare('INSERT INTO ec_host_schema_migrations(name) VALUES(?)').run('staged-operation-payloads'); + })(); + } if (!handle.prepare('SELECT 1 FROM ec_host_schema_migrations WHERE name=?').get('completed-status-batches')) { handle.transaction(() => { // Reclassify existing completed batches without discarding their per-server replies. diff --git a/src/lib/server/provisioning/native.ts b/src/lib/server/provisioning/native.ts index 1573593..c854030 100644 --- a/src/lib/server/provisioning/native.ts +++ b/src/lib/server/provisioning/native.ts @@ -36,8 +36,8 @@ import { safetyOverride, safetyOverrideCapability } from './safety-override.js'; export type NativeSelection = import('../../hosts.js').NativePlacement; type NativeRecipe = LaunchRecipePlan & { native?: NativeSelection }; interface Job { selection?: string | null; id: string; server: string; action: ProvisioningAction; state: string; submitted_at: string | null; stage: string; - instance_id: string; generation: number; stage_operation_id: string | null; operation_id: string | null; files_json: string | null; applied_at: string | null } -interface FilePlan { id: number; path: string; contentBase64: string | null; sha256: string | null; checkBefore: boolean; expectedBeforeSha: string | null } + instance_id: string; generation: number; stage_operation_id: string | null; operation_id: string | null; applied_at: string | null } +interface FilePlan { id: number; path: string; sha256: string | null; checkBefore: boolean; expectedBeforeSha: string | null } /** Validate a launch request without selecting a host or consuming capacity. */ export function nativeLaunchSelection(input: Record, game: string): NativeSelection { @@ -215,7 +215,7 @@ export function queueNativeRebuild(name: string, triggeredBy: string): CreateSer } catch (error) { return { ok: false, kind: 'bad-request', message: (error as Error).message }; } } -function filesFor(job: Job): FilePlan[] { +function filesFor(job: Job): Array { const rows = db().prepare(`SELECT f.* 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=? AND f.server=? AND f.state='pending' ORDER BY f.id`).all(job.id, job.server) as StagedFileChange[]; @@ -224,10 +224,18 @@ function filesFor(job: Job): FilePlan[] { checkBefore: !!row.check_before, expectedBeforeSha: row.expected_before_sha })); } +function fileMetadata({ contentBase64, ...file }: FilePlan & { contentBase64?: string | null }): FilePlan { return file; } + +function storedFiles(job: Job): FilePlan[] { + const row = db().prepare('SELECT files_json FROM native_provisioning WHERE deployment_id=?').get(job.id) as { files_json: string | null } | undefined; + return (JSON.parse(row?.files_json ?? '[]') as FilePlan[]).map(fileMetadata); +} + /** Agent receipts are the authoritative before images; do not read a controller mount. */ function applyReceipt(job: Job, result: Record) { - if (job.applied_at || !job.files_json) return; - const plans = JSON.parse(job.files_json) as FilePlan[]; + if (job.applied_at) return; + const plans = storedFiles(job); + if (!plans.length) return; const before = result.before as { id?: number; path: string; contentBase64: string | null }[] | undefined; db().transaction(() => { for (const plan of plans) { @@ -257,7 +265,7 @@ function applyReceipt(job: Job, result: Record) { /** 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 + const job = db().prepare(`SELECT d.*,n.instance_id,n.generation,n.operation_id,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; @@ -280,7 +288,7 @@ export function reconcileNativeProvisioningReceipt(id: string): void { 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[]) { + for (const file of storedFiles(job)) { db().prepare("UPDATE staged_file_changes SET state=? WHERE id=? AND state='applying'") .run(job.state === 'cancelled' ? 'cancelled' : 'pending', file.id); } @@ -348,7 +356,7 @@ export async function processNativeProvisioning(eligibility: (operation: { id: s } } 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 + const jobs = db().prepare(`SELECT d.*,n.instance_id,n.generation,n.stage_operation_id,n.operation_id,n.applied_at,q.selection FROM server_deployments d LEFT JOIN native_provisioning n ON n.deployment_id=d.id LEFT JOIN native_creation_intents q ON q.deployment_id=d.id WHERE (n.deployment_id IS NOT NULL OR q.deployment_id IS NOT NULL) AND 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[]; @@ -494,7 +502,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { // post-mutation failures remain unknown and retain their capacity. db().transaction(() => { db().prepare('DELETE FROM restart_rollout WHERE operation_id=?').run(job.id); - const files = JSON.parse(job.files_json ?? '[]') as FilePlan[]; + const files = storedFiles(job); for (const file of files) db().prepare("UPDATE staged_file_changes SET state='pending' WHERE id=? AND state='applying'").run(file.id); finishDeployment(job.id, { ok: false, message: operation.error ?? 'Agent operation failed before mutation.' }); })(); @@ -585,10 +593,14 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { if (job.action === 'destroy' || job.action === 'rebuild') { const beta = activeBetaRemovalBlocker(job.server); if (beta) { hold(job, beta); return; } } - const files = job.files_json ? JSON.parse(job.files_json) as FilePlan[] : filesFor(job); + const files = job.stage_operation_id ? storedFiles(job) : filesFor(job); if (files.length && !job.stage_operation_id) { - const stage = queueHostOperation(instance.host_id, 'files.stage', { stageId: job.id, files }, { instanceId: instance.id, generation: instance.generation, idempotencyKey: `${job.id}:stage` }); - db().prepare('UPDATE native_provisioning SET stage_operation_id=?,files_json=? WHERE deployment_id=?').run(stage.id, JSON.stringify(files), job.id); + db().transaction(() => { + const stage = queueHostOperationSerialized(instance.host_id, 'files.stage', Buffer.from(JSON.stringify({ stageId: job.id, files })), + { instanceId: instance.id, generation: instance.generation, idempotencyKey: `${job.id}:stage` }, { stageId: job.id }); + db().prepare('UPDATE native_provisioning SET stage_operation_id=?,files_json=? WHERE deployment_id=?') + .run(stage.id, JSON.stringify(files.map(fileMetadata)), job.id); + })(); hold(job, 'Staging and verifying files on the host.'); return; } if (job.stage_operation_id) { @@ -671,7 +683,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { const beta = activeBetaRemovalBlocker(job.server); if (beta) { hold(job, beta); return; } } - if (JSON.stringify(filesFor(job)) !== JSON.stringify(files)) { + if (JSON.stringify(filesFor(job).map(fileMetadata)) !== JSON.stringify(files.map(fileMetadata))) { finishDeployment(job.id, { ok: false, message: 'File requests changed during preflight. Review and submit a fresh activation.' }); return; } const capacityBlocker = concurrencyBlocker(job.id, job.action, instance!.host_id); diff --git a/tests/unit/native-provisioning.test.ts b/tests/unit/native-provisioning.test.ts index 5130088..9110afc 100644 --- a/tests/unit/native-provisioning.test.ts +++ b/tests/unit/native-provisioning.test.ts @@ -708,6 +708,18 @@ describe('native instance provisioning', () => { const { getHostOperation, reportHostOperation } = await import('../../src/lib/server/hosts/index.js'); const stage = getHostOperation(staged.stage_operation_id)!; expect(stage.kind).toBe('files.stage'); expect(await operationFor(restart.id)).toBeNull(); + const routing = db().prepare('SELECT payload FROM ec_host_operations WHERE id=?').get(stage.id) as { payload: string }; + expect(JSON.parse(routing.payload)).toEqual({ stageId: restart.id }); + const plan = db().prepare('SELECT files_json FROM native_provisioning WHERE deployment_id=?').get(restart.id) as { files_json: string }; + expect(JSON.parse(plan.files_json)[0]).not.toHaveProperty('contentBase64'); + expect(db().prepare('SELECT payload FROM ec_host_operation_payloads WHERE operation_id=?').get(stage.id)) + .toEqual({ payload: JSON.stringify(stage.payload) }); + const { closeDb } = await import('../../src/lib/server/db/index.js'); + closeDb(); + const { pendingHostOperations } = await import('../../src/lib/server/hosts/index.js'); + expect(pendingHostOperations(hostId).find(operation => operation.id === stage.id)?.payload).toEqual(stage.payload); + await advance(); + expect((db().prepare('SELECT stage_operation_id FROM native_provisioning WHERE deployment_id=?').get(restart.id) as { stage_operation_id: string }).stage_operation_id).toBe(stage.id); expect(stagedFileChanges('campaign', 'campaign')[0].state).toBe('pending'); reportHostOperation(hostId, stage.id, { state: 'completed', result: { state: 'staged' } }); await advance(); expect(await operationFor(restart.id)).toBeNull();