josh/electric-cherry
1
0
Fork 0
Code Issues 18 Pull Requests Actions Packages Projects Releases Wiki Activity

Separate staged command payloads from provisioning state

Browse Source
This commit is contained in:
josh
2026-10-06 14:03:28 -04:00
parent ddb3db9537
commit fbf0629486
4 changed files with 54 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/backends/schema.ts
+8
View File
@@ -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. */
src/lib/server/hosts/schema.ts
+9
View File
@@ -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.
src/lib/server/provisioning/native.ts
+25 -13
View File
@@ -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<string, unknown>, 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<FilePlan & { contentBase64: string | null }> {
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<string, unknown>) {
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<string, unknown>) {
/** 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);