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

Fix rebuild generation reconciliation (Refs #155)

Browse Source
This commit is contained in:
josh
2026-10-14 12:36:32 -04:00
parent 61dbfb2c5f
commit dedf15d9af
17 changed files with 672 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/provisioning/native.ts
+62 -35
View File
@@ -70,7 +70,11 @@ 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';
import {
probeServerStatus,
probePlayerEvidence,
latestGameSubmission
} from '../fleet/status-evidence.js';
import { monitoringIdentity } from '../fleet/health-observations.js';
import { handoffVerifiedHealth } from './health-handoff.js';
import { checkRelease, freshRelease, probeVersion } from '../integrations/plutonium.js';
@@ -85,6 +89,12 @@ import { reconcileHostDrains } from './host-drain.js';
import { hostMigration, isRebalanceMigration, processMigrationRemoval } from './migrations.js';
import { safetyOverride, safetyOverrideCapability } from './safety-override.js';
import { runtimeLaunchSql } from '../fleet/runtime-config.js';
import {
reconcileCompletedRebuild,
assertVerificationGeneration,
GenerationRecoveryRequired,
observedGeneration
} from './rebuild-receipt.js';
export type NativeSelection = import('../../hosts.js').NativePlacement;
type NativeRecipe = LaunchRecipePlan & { native?: NativeSelection };
@@ -813,27 +823,11 @@ export function reconcileNativeProvisioningReceipt(id: string): void {
finishCreationCancellation(job.id);
return;
}
if (operation.kind === 'instance.rebuild' && 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 &&
!(operation.kind === 'instance.rebuild' && instance.generation === job.generation + 1)
)
return;
if (operation.kind === 'instance.rebuild') {
// An older applied receipt must never undo a later successful replacement.
if (job.applied_at && instance.generation > operation.generation! + 1) return;
reconcileCompletedRebuild(id);
} else if (instance.generation !== job.generation) return;
applyReceipt(job, (operation.result ?? {}) as Record<string, unknown>);
db()
.prepare(
@@ -1032,10 +1026,12 @@ export async function processNativeProvisioning(
`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 native_instances i ON i.id=n.instance_id AND i.retired_at IS NULL
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'))
(o.state='completed' AND (n.applied_at IS NULL OR d.action='destroy'
OR (o.kind='instance.rebuild' AND i.generation=o.generation)))
OR (o.state='failed' AND o.kind='instance.rebuild' AND json_type(o.result,'$.rebuildFailure') IS NOT NULL AND n.applied_at IS NULL)
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
@@ -1061,7 +1057,8 @@ export async function processNativeProvisioning(
`SELECT d.*,n.instance_id,n.generation,n.stage_operation_id,n.operation_id,n.applied_at,n.recovery_step,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`
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')
AND COALESCE(CASE WHEN json_valid(d.observation_json) THEN json_extract(d.observation_json,'$.blocker.code') END,'')!='generation-recovery' ORDER BY d.started_at,d.rowid`
)
.all() as Job[];
// One host's observation must not prevent another job being admitted.
@@ -1075,7 +1072,27 @@ export async function processNativeProvisioning(
() => processJob(job, eligibility)
)
.catch((error) => {
if (stillActive(job)) hold(job, (error as Error).message);
if (!stillActive(job)) return;
if (
error instanceof GenerationRecoveryRequired ||
/generation has been superseded or removed/i.test((error as Error).message)
) {
const instance = nativeInstance(job.server);
const message =
error instanceof GenerationRecoveryRequired
? error.message
: `Generation recovery required: controller expects ${instance?.generation ?? job.generation}, host reports ${instance ? (observedGeneration(instance) ?? 'unknown') : 'unknown'}. Reconcile the completed rebuild before resuming verification.`;
db()
.prepare(
"UPDATE server_deployments SET state='interrupted',error=? WHERE id=?"
)
.run(message, job.id);
deploymentStage(job.id, message, {
blocker: { code: 'generation-recovery', message, automatic: false },
checks: [],
recovery: { state: 'held', operationId: job.id, message }
});
} else hold(job, (error as Error).message);
})
.finally(() => {
activeJobs.delete(job.id);
@@ -1328,6 +1345,15 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
job.instance_id = instance.id;
job.generation = instance.generation;
}
// Reconcile before comparing the job snapshot or issuing any observations.
// Historical jobs can have a stale binding even though their command is correct.
if (job.operation_id) {
const receipt = getHostOperationStatus(job.operation_id);
if (receipt?.kind === 'instance.rebuild' && receipt.state === 'completed') {
instance = reconcileCompletedRebuild(job.id);
job.generation = receipt.generation!;
}
}
const migration = hostMigration(job.id);
const rebuilding =
job.recovery_step === 'rebuild' ||
@@ -1336,7 +1362,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
!instance ||
instance.id !== job.instance_id ||
(instance.generation !== job.generation &&
!(rebuilding && instance.generation === job.generation + 1))
!(rebuilding && job.operation_id && instance.generation === job.generation + 1))
) {
finishDeployment(job.id, {
ok: false,
@@ -1493,12 +1519,6 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
})();
return;
}
if (operation.kind === 'instance.rebuild' && instance.generation === job.generation) {
db()
.prepare('UPDATE native_instances SET generation=? WHERE id=? AND generation=?')
.run(job.generation + 1, instance.id, job.generation);
instance = nativeInstance(job.server)!;
}
applyReceipt(job, (operation.result ?? {}) as Record<string, unknown>);
db()
.prepare(
@@ -1542,6 +1562,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
blocker: null,
checks
});
assertVerificationGeneration(instance, operation.updatedAt);
const verificationIdentity = monitoringIdentity(job.server);
const health = await nativeRequest<{ healthy: boolean; state: string }>(
instance,
@@ -1601,7 +1622,11 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
return;
}
deploymentStage(job.id, 'Checking game status', { phase: 'verify-game', blocker: null });
const status = await probeServerStatus(job.server, operation.updatedAt);
const submission = latestGameSubmission(job.server);
const status = await probeServerStatus(
job.server,
submission && submission > operation.updatedAt ? submission : operation.updatedAt
);
if (!stillActive(job)) return;
check(
{
@@ -2255,8 +2280,10 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) {
)
: queueHostOperation(instance!.host_id, kind, payload, operationOptions);
db()
.prepare('UPDATE native_provisioning SET operation_id=? WHERE deployment_id=?')
.run(operation.id, job.id);
.prepare(
'UPDATE native_provisioning SET operation_id=?,generation=? WHERE deployment_id=?'
)
.run(operation.id, instance!.generation, job.id);
for (const file of files)
db()
.prepare(