From c4e2ff4384a83c9f8adac0c56d130614603b86a1 Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Thu, 8 Oct 2026 14:24:52 -0400 Subject: [PATCH] Reconcile rebuilds interrupted before replacement creation --- agent/ec_agent.py | 74 ++++++++- agent/tests/test_agent.py | 168 ++++++++++++++++++++ docs/guides/native-hosts.md | 10 ++ docs/guides/provisioning.md | 10 ++ src/lib/server/provisioning/cancellation.ts | 16 +- src/lib/server/provisioning/native.ts | 60 ++++++- src/lib/server/retention.ts | 5 +- static/agent/ec-agent.py | 74 ++++++++- static/agent/manifest.json | 4 +- tests/unit/native-provisioning.test.ts | 135 ++++++++++++++++ tests/unit/retention.test.ts | 29 ++++ 11 files changed, 571 insertions(+), 14 deletions(-) diff --git a/agent/ec_agent.py b/agent/ec_agent.py index 6c7583c..c883576 100644 --- a/agent/ec_agent.py +++ b/agent/ec_agent.py @@ -35,7 +35,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.8.3" +VERSION = "1.8.4" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -2665,12 +2665,84 @@ class Agent: atomic_json(self.config_path, self.config) return True + def reconcile_uncreated_rebuild(self, operation, row): + """Settle removal only when the replacement create was never submitted. + + The write lane and instance lock exclude the original Python execution. + A terminal journal receipt prevents any later replay of that command. + """ + if row["state"] != "unknown": + return row + instance_id, generation = instance_identity(operation.get("instanceId"), operation.get("generation")) + payload = operation.get("payload", {}) + next_generation = payload.get("nextGeneration", payload.get("newGeneration")) + if type(next_generation) is not int or next_generation <= generation: + return row + previous = json.loads(row["result"]) if row.get("result") else {} + if not isinstance(previous, dict): + return row + directory = scoped(self.root, f"instances/{instance_id}/{next_generation}") + + def unchanged_generation(): + try: + marker = scoped(self.root, f"instances/{instance_id}/current.json") + current = json.loads(marker.read_text()) + except FileNotFoundError: + return True + return isinstance(current, dict) and type(current.get("generation")) is int and current["generation"] == generation and current.get("removed", False) is False + + def both_missing(): + for target in (generation, next_generation): + health = self.inspect(instance_id, target) + if health.get("exists") is not False or health.get("running") is not False or health.get("state") != "missing": + return False + return True + + if not unchanged_generation() or not both_missing(): + return row + # Preparation can leave runtime/configuration files before create's + # checkpoint. A fresh operation must not inherit those partial files. + expected_spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration", "unhealthyRecovery")} + try: + saved_spec = json.loads(scoped(directory, "spec.json").read_text()) + except FileNotFoundError: + pass + else: + if saved_spec != expected_spec: + return row + try: + metadata = directory.lstat() + except FileNotFoundError: + pass + else: + if not stat_module.S_ISDIR(metadata.st_mode): + return row + shutil.rmtree(directory) + # Also sync when a previous reconciliation removed the directory but + # crashed before recording the terminal receipt. + try: + sync_parent(directory) + except FileNotFoundError: + pass + if not unchanged_generation() or not both_missing(): + return row + original_error = row.get("error") + result = {**previous, "reconciled": True, "originalError": original_error, + "rebuildFailure": {"phase": "disrupting", "generation": generation, "nextGeneration": next_generation, + "oldContainerMissing": True, "replacementContainerMissing": True, "replacementFilesRemoved": True}} + message = "Old container removed; replacement creation failed." + if original_error: + message += " " + original_error + return self.journal.update(operation["id"], "failed", result, error=message) + def reconcile(self, operation, row): """Prove desired state with immutable ownership; never resubmit a command.""" kind, payload = operation["kind"], operation.get("payload", {}) instance_id, generation = operation.get("instanceId"), operation.get("generation") checkpoint = json.loads(row["checkpoint"]) if row.get("checkpoint") else {} try: + if kind == "instance.rebuild" and checkpoint.get("phase") == "disrupting": + return self.reconcile_uncreated_rebuild(operation, row) if kind == "instance.remove" and "autoscaling" in payload and not checkpoint: return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error="Interrupted autoscaling removal requires a new decision") if kind == "instance.cancel-create": diff --git a/agent/tests/test_agent.py b/agent/tests/test_agent.py index 137f224..7be3e77 100644 --- a/agent/tests/test_agent.py +++ b/agent/tests/test_agent.py @@ -1604,6 +1604,174 @@ class AgentTests(unittest.TestCase): self.assertEqual(self.agent.execute(operation)["state"], "completed") self.assertFalse(old.exists()) + def interrupted_rebuild_before_create(self): + payload = self.create() + self.agent.rcon = lambda *_: {"ok": True, "players": 0} + operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2}, "interrupted-rebuild") + + def interrupted_seed(app, image): + (app / "partial-runtime").write_text("incomplete") + raise OSError(module.errno.ENOSPC, "disk full") + + with patch.object(self.agent, "seed_plutonium", side_effect=interrupted_seed): + row = self.agent.execute(operation) + self.assertEqual(row["state"], "unknown", row) + self.assertEqual(json.loads(row["checkpoint"])["phase"], "disrupting") + self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"]) + self.assertFalse(self.agent.inspect(INSTANCE, 2)["exists"]) + return operation, row + + def test_rebuild_reconciles_removal_before_create_without_replaying(self): + operation, interrupted = self.interrupted_rebuild_before_create() + original = self.agent.instance_dir(INSTANCE, 1) + partial = self.agent.instance_dir(INSTANCE, 2) + module.atomic_json(partial / "spec.json", operation["payload"]["spec"]) + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + calls = list(self.docker.calls) + failed = self.agent.execute(operation) + self.assertEqual(failed["state"], "failed", failed) + result = json.loads(failed["result"]) + self.assertEqual(result["rebuildFailure"], {"phase": "disrupting", "generation": 1, "nextGeneration": 2, + "oldContainerMissing": True, "replacementContainerMissing": True, + "replacementFilesRemoved": True}) + self.assertTrue(result["reconciled"]) + self.assertEqual(result["unexpectedError"], json.loads(interrupted["result"])["unexpectedError"]) + self.assertEqual(result["originalError"], interrupted["error"]) + self.assertTrue(failed["error"].startswith("Old container removed; replacement creation failed.")) + self.assertIn(interrupted["error"], failed["error"]) + self.assertEqual((original / "app/configs/server.cfg").read_bytes(), b"original") + self.assertFalse(partial.exists()) + self.assertEqual(json.loads((original.parent / "current.json").read_text()), {"generation": 1}) + self.assertEqual(self.agent.execute(operation), failed) + self.assertEqual(self.docker.calls, calls) + replacement = self.operation("instance.rebuild", operation["payload"], "replacement-rebuild") + self.assertEqual(self.agent.execute(replacement)["state"], "completed") + self.assertFalse((partial / "app/partial-runtime").exists()) + self.assertEqual(self.agent.inspect(INSTANCE, 2)["creationOperation"], replacement["id"]) + calls = list(self.docker.calls) + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + self.assertEqual(self.agent.execute(operation), failed) + self.assertEqual(self.docker.calls, calls) + + def test_rebuild_removal_reconciles_when_marker_is_absent(self): + operation, _ = self.interrupted_rebuild_before_create() + marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json" + marker.unlink() + row = self.agent.execute(operation) + self.assertEqual(row["state"], "failed", row) + self.assertIn("rebuildFailure", json.loads(row["result"])) + self.assertFalse(marker.exists()) + self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists()) + + def test_rebuild_lost_removal_receipt_settles_without_recreating_or_deleting_twice(self): + payload = self.create() + self.agent.rcon = lambda *_: {"ok": True, "players": 0} + self.docker.lose_reply_after = "rm" + operation = self.operation("instance.rebuild", {"spec": payload, "nextGeneration": 2}, "rebuild") + self.assertEqual(self.agent.execute(operation)["state"], "unknown") + self.assertFalse(self.agent.instance_dir(INSTANCE, 2).exists()) + calls = list(self.docker.calls) + result = self.agent.execute(operation) + self.assertEqual(result["state"], "failed", result) + self.assertTrue(json.loads(result["result"])["rebuildFailure"]["replacementFilesRemoved"]) + self.assertEqual(self.docker.calls, calls) + self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists()) + + def test_rebuild_removal_requires_both_containers_positively_missing(self): + operation, interrupted = self.interrupted_rebuild_before_create() + partial = self.agent.instance_dir(INSTANCE, 2) + for generation in (1, 2): + for status, foreign in (("running", False), ("created", False), ("exited", False), ("unknown", False), ("running", True)): + with self.subTest(generation=generation, status=status, foreign=foreign): + name = self.agent.container_name(INSTANCE, generation) + labels = {} if foreign else {module.LABEL + ".managed": "true", module.LABEL + ".instance": INSTANCE, + module.LABEL + ".generation": str(generation)} + self.docker.containers[name] = {"Config": {"Labels": labels}, "State": {"Status": status, "Running": status == "running"}} + self.assertEqual(self.agent.execute(operation), interrupted) + self.assertTrue(partial.exists()) + self.docker.containers.pop(name) + with patch.object(self.agent, "inspect", return_value={"exists": False, "running": False, "state": "unknown"}): + self.assertEqual(self.agent.execute(operation), interrupted) + with patch.object(self.agent, "inspect", side_effect=AgentError("Docker unavailable")): + self.assertEqual(self.agent.execute(operation), interrupted) + self.assertTrue(partial.exists()) + + def test_rebuild_removal_does_not_settle_after_a_create_checkpoint(self): + operation, interrupted = self.interrupted_rebuild_before_create() + for phase in ("creating", "starting", "unknown"): + with self.subTest(phase=phase): + self.agent.journal.checkpoint(operation["id"], {"phase": phase, "generation": 2}) + row = self.agent.execute(operation) + self.assertEqual(row["state"], "unknown", row) + self.assertEqual(row["result"], interrupted["result"]) + self.assertTrue(self.agent.instance_dir(INSTANCE, 2).exists()) + self.agent.journal.db.execute("UPDATE operations SET checkpoint=NULL WHERE id=?", (operation["id"],)) + self.agent.journal.db.commit() + self.assertEqual(self.agent.execute(operation)["state"], "unknown") + self.assertTrue(self.agent.instance_dir(INSTANCE, 2).exists()) + + def test_rebuild_removal_rejects_ambiguous_generation_markers(self): + operation, interrupted = self.interrupted_rebuild_before_create() + marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json" + for current in ({"generation": 0}, {"generation": 2}, {"generation": 3}, {"generation": True}, {"generation": "1"}, + {"generation": 1, "removed": True}, {"generation": 1, "removed": None}, {}, []): + with self.subTest(current=current): + module.atomic_json(marker, current) + self.assertEqual(self.agent.execute(operation), interrupted) + self.assertTrue(self.agent.instance_dir(INSTANCE, 2).exists()) + self.assertEqual(json.loads(marker.read_text()), current) + marker.write_text("invalid JSON") + self.assertEqual(self.agent.execute(operation), interrupted) + + def test_rebuild_removal_keeps_unknown_for_unrelated_partial_files_and_symlinks(self): + operation, interrupted = self.interrupted_rebuild_before_create() + partial = self.agent.instance_dir(INSTANCE, 2) + module.atomic_json(partial / "spec.json", {**operation["payload"]["spec"], "name": "unrelated"}) + self.assertEqual(self.agent.execute(operation), interrupted) + self.assertTrue(partial.exists()) + (partial / "spec.json").unlink() + outside = self.agent.root / "retained-files" + partial.rename(outside) + partial.symlink_to(outside, target_is_directory=True) + self.assertEqual(self.agent.execute(operation), interrupted) + self.assertEqual((outside / "app/partial-runtime").read_text(), "incomplete") + partial.unlink() + outside.rename(partial) + marker = partial.parent / "current.json" + retained_marker = self.agent.root / "retained-current.json" + marker.rename(retained_marker) + marker.symlink_to(retained_marker) + self.assertEqual(self.agent.execute(operation), interrupted) + self.assertTrue(partial.exists()) + + def test_rebuild_removal_keeps_cleanup_errors_and_changed_generation_unresolved(self): + operation, interrupted = self.interrupted_rebuild_before_create() + partial = self.agent.instance_dir(INSTANCE, 2) + with patch.object(module.shutil, "rmtree", side_effect=PermissionError("cleanup denied")): + self.assertEqual(self.agent.execute(operation), interrupted) + self.assertTrue(partial.exists()) + cleanup = module.shutil.rmtree + + def superseded(path): + cleanup(path) + module.atomic_json(partial.parent / "current.json", {"generation": 3}) + + with patch.object(module.shutil, "rmtree", side_effect=superseded): + self.assertEqual(self.agent.execute(operation), interrupted) + self.assertEqual(json.loads((partial.parent / "current.json").read_text()), {"generation": 3}) + self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists()) + + def test_rebuild_removal_waits_for_durable_partial_file_cleanup(self): + operation, interrupted = self.interrupted_rebuild_before_create() + with patch.object(module, "sync_parent", side_effect=OSError("sync unavailable")): + self.assertEqual(self.agent.execute(operation), interrupted) + self.assertFalse(self.agent.instance_dir(INSTANCE, 2).exists()) + self.assertEqual(self.agent.execute(operation), interrupted) + self.assertEqual(self.agent.execute(operation)["state"], "failed") + self.assertTrue(self.agent.instance_dir(INSTANCE, 1).exists()) + def test_traversal_and_symlinks_are_rejected(self): self.create() app = self.agent.instance_dir(INSTANCE, 1) / "app" diff --git a/docs/guides/native-hosts.md b/docs/guides/native-hosts.md index 83bc22b..b589a43 100644 --- a/docs/guides/native-hosts.md +++ b/docs/guides/native-hosts.md @@ -415,6 +415,16 @@ Otherwise the result remains **Unknown** and holds dependent work. Running alone is insufficient evidence that a particular restart happened. EC neither assumes an unreachable server stopped nor launches a duplicate replacement elsewhere. +An interrupted rebuild can settle as **Failed** when its last checkpoint is +`disrupting`, Docker confirms both the old and replacement containers are absent, +and any existing generation marker still matches the original instance. The agent removes +only the partial replacement files, preserves the old generation's files and +marker, and retains the original diagnostic in its failure report. It does not +replay the interrupted rebuild. A separately queued rebuild can then proceed +through normal eligibility checks. A later creation checkpoint, a changed +generation, an inspection error, or failed partial-file cleanup leaves the result +**Unknown** and keeps dependent work waiting. + ### Player checks and activation Campaign bytes are staged and verified while players remain connected. Only diff --git a/docs/guides/provisioning.md b/docs/guides/provisioning.md index d97f018..9cdf931 100644 --- a/docs/guides/provisioning.md +++ b/docs/guides/provisioning.md @@ -200,6 +200,16 @@ guards still apply. **Resume rebuild cleanup** completes cleanup before creating the replacement in its reserved slot. Resuming the same operation preserves its earlier failure and resume events. Uncertain commands have no generic Retry action. +Cancelling an operation does not stop a command already delivered to the host. +A later rebuild waits for that command to settle. Its waiting message includes +the host command's state and last reported error; **View cancelled operation** +opens the original command details. If an interrupted rebuild removed its old +container before creating the replacement, the host can confirm failure after +verifying that both containers are absent and the instance identity is unchanged. +The original diagnostic remains available, and the queued rebuild can continue. +Uncertain inspections keep it waiting; the old command is never replayed merely +because its receipt was lost. + Create and rebuild capture a setup when queued and select the latest promoted files again when creation begins. The page distinguishes **Setup captured when queued** from **Files selected for this attempt**. A queued capture does not pin diff --git a/src/lib/server/provisioning/cancellation.ts b/src/lib/server/provisioning/cancellation.ts index 78efdd7..ed81020 100644 --- a/src/lib/server/provisioning/cancellation.ts +++ b/src/lib/server/provisioning/cancellation.ts @@ -60,14 +60,20 @@ export function cancelProvisioningJob(id: string, reason = 'Cancelled by operato } /** A cancelled controller job must not let a replacement overtake a delivered mutation. */ -export function unsettledCancelledCommand(server: string, replacementId: string): string | null { - const row = db().prepare(`SELECT d.id FROM server_deployments d +export function unsettledCancelledCommand(server: string, replacementId: string): { + deploymentId: string; operationId: string; state: string; error: string | null; +} | null { + const row = db().prepare(`SELECT d.id AS deploymentId,o.id AS operationId,o.state,o.error FROM server_deployments d JOIN native_provisioning n ON n.deployment_id=d.id LEFT JOIN ec_host_migrations m ON m.deployment_id=d.id JOIN ec_host_operations o ON o.id=n.operation_id OR o.id=m.remove_operation_id WHERE d.server=? AND d.server_deleted_at IS NULL AND d.state='cancelled' AND (o.state NOT IN ('completed','failed') - OR (o.state='completed' AND o.id=n.operation_id AND n.files_json IS NOT NULL AND n.applied_at IS NULL)) - AND NOT EXISTS (SELECT 1 FROM native_creation_cancellations c WHERE c.creation_id=d.id AND c.cleanup_id=?) LIMIT 1`).get(server, replacementId) as { id: string } | undefined; - return row?.id ?? null; + OR (o.state='completed' AND o.id=n.operation_id AND n.files_json IS NOT NULL AND n.applied_at IS NULL) + OR (o.state='failed' AND o.id=n.operation_id AND o.kind='instance.rebuild' + AND json_type(o.result,'$.rebuildFailure') IS NOT NULL AND n.applied_at IS NULL)) + AND NOT EXISTS (SELECT 1 FROM native_creation_cancellations c WHERE c.creation_id=d.id AND c.cleanup_id=?) + ORDER BY d.rowid,o.rowid LIMIT 1`).get(server, replacementId) as + { deploymentId: string; operationId: string; state: string; error: string | null } | undefined; + return row ?? null; } diff --git a/src/lib/server/provisioning/native.ts b/src/lib/server/provisioning/native.ts index 4bfeb3d..f309f83 100644 --- a/src/lib/server/provisioning/native.ts +++ b/src/lib/server/provisioning/native.ts @@ -13,7 +13,7 @@ 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, listHosts, getHostOperationStatus, queueHostOperation, queueHostOperationSerialized, releaseHostPorts, reserveHostPort } from '../hosts/index.js'; +import { getHost, listHosts, getHostOperationStatus, queueHostOperation, queueHostOperationSerialized, releaseHostPorts, reserveHostPort, type HostOperationStatus } from '../hosts/index.js'; import { newestReadyLibraryId, assertReadyLibrary } from '../game-files/index.js'; import { blueprintRecipe, captureRecipePlan, recipeForDeployment, recipeSummaryForDeployment, recipeStorage, summarizeRecipe, type LaunchRecipePlan } from '../fleet/launch-recipes.js'; import { prepareNativeLaunch, evaluateNativePlacement, type PreparedNativeLaunch } from './launch-preparation.js'; @@ -277,6 +277,48 @@ function applyReceipt(job: Job, result: Record) { })(); } +/** Consume a conclusive partial failure without treating removal as a healthy rebuild. */ +function applyIncompleteRebuildReceipt(job: Job, operation: HostOperationStatus): void { + const result = operation.result as { reconciled?: unknown; rebuildFailure?: { + phase?: unknown; generation?: unknown; nextGeneration?: unknown; + oldContainerMissing?: unknown; replacementContainerMissing?: unknown; replacementFilesRemoved?: unknown; + } | null } | null; + if (!result || !Object.prototype.hasOwnProperty.call(result, 'rebuildFailure')) return; + const failure = result.rebuildFailure; + if (operation.state !== 'failed' || operation.kind !== 'instance.rebuild' || result.reconciled !== true + || !failure || failure.phase !== 'disrupting' || failure.generation !== job.generation + || failure.nextGeneration !== job.generation + 1 || failure.oldContainerMissing !== true + || failure.replacementContainerMissing !== true || failure.replacementFilesRemoved !== true) { + throw new Error('The interrupted rebuild receipt does not confirm safe replacement cleanup. Inspect the original host command.'); + } + db().transaction(() => { + const binding = db().prepare('SELECT operation_id,applied_at FROM native_provisioning WHERE deployment_id=?') + .get(job.id) as { operation_id: string | null; applied_at: string | null } | undefined; + if (binding?.operation_id !== operation.id) throw new Error('The interrupted rebuild command no longer belongs to this operation.'); + if (binding.applied_at) return; + const requested = db().prepare(`SELECT COALESCE(json_extract(COALESCE(p.payload,o.payload),'$.nextGeneration'), + json_extract(COALESCE(p.payload,o.payload),'$.newGeneration')) AS nextGeneration FROM ec_host_operations o + LEFT JOIN ec_host_operation_payloads p ON p.operation_id=o.id WHERE o.id=?`).get(operation.id) as + { nextGeneration: number | null } | undefined; + if (requested?.nextGeneration !== failure.nextGeneration) { + throw new Error('The interrupted rebuild receipt does not match its requested replacement generation.'); + } + const instance = nativeInstance(job.server); + if (!instance || instance.id !== job.instance_id || instance.generation !== job.generation + || operation.instanceId !== instance.id || operation.hostId !== instance.host_id || operation.generation !== instance.generation) { + throw new Error('The interrupted rebuild belongs to a different instance or generation. Inspect its host command.'); + } + db().prepare("UPDATE servers SET container_state='missing',container_checked_at=?,last_players=NULL,last_roster=NULL WHERE name=?") + .run(operation.updatedAt, job.server); + db().prepare('UPDATE native_provisioning SET applied_at=? WHERE deployment_id=?').run(operation.updatedAt, job.id); + recordProgress(job.id, { checks: [{ key: 'container', label: 'Container state after interrupted rebuild', source: 'agent', + result: 'fail', actual: 'missing', checkedAt: operation.updatedAt, + message: 'The old container is absent and replacement creation never started. Partial replacement files were removed.' }], + event: { kind: 'reconciled', message: 'Interrupted rebuild settled: old container removed; replacement creation failed.', + details: { operationId: operation.id, generation: job.generation, nextGeneration: failure.nextGeneration } } }); + })(); +} + /** 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.applied_at,n.recovery_step @@ -303,6 +345,7 @@ export function reconcileNativeProvisioningReceipt(id: string): void { if (!job.operation_id) return; const operation = getHostOperationStatus(job.operation_id); if (operation?.state === 'failed') { + applyIncompleteRebuildReceipt(job, operation); 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); @@ -434,6 +477,7 @@ export async function processNativeProvisioning(eligibility: (operation: { id: s 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 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 WHERE s.operation_id=d.id AND f.state='applying')) @@ -542,7 +586,14 @@ async function placeQueuedCreation(job: Job, instance: NativeInstance | null, el 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; } + if (unsettled) { + const pendingReceipt = unsettled.state === 'completed' || unsettled.state === 'failed'; + const message = `Waiting for the host command from cancelled operation ${unsettled.deploymentId} ${pendingReceipt ? 'to be reconciled' : 'to settle'}. Agent state: ${unsettled.state}.` + + (unsettled.error ? ` Last error: ${unsettled.error.slice(0, 500)}` : ''); + deploymentStage(job.id, message, { blocker: { code: 'cancelled-command', message, automatic: true, + href: `/provisioning/${encodeURIComponent(unsettled.deploymentId)}`, label: 'View cancelled operation' } }); + 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 @@ -588,6 +639,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { } if (!operation || operation.state === 'unknown') { hold(job, 'Agent result is unknown. Reconciling the original operation; no replacement or repeated command will be sent.'); return; } if (operation.state === 'failed') { + applyIncompleteRebuildReceipt(job, operation); if (job.action === 'destroy' && scalingIntent(job.id) && (operation.result as { autoscalingInvalidated?: boolean } | null)?.autoscalingInvalidated === true) { db().transaction(() => { db().prepare("UPDATE server_deployments SET state='cancelled',stage='Automatic removal withdrawn; new empty interval required',finished_at=?,error=NULL WHERE id=?") @@ -607,8 +659,8 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { hold(job, `${operation.error ?? 'Destination creation failed before mutation.'} Retrying after fresh capacity checks.`); return; } - // The agent reports failed only when no mutation began. Ambiguous - // post-mutation failures remain unknown and retain their capacity. + // Failed commands either never mutated or have a conclusively + // reconciled partial rebuild above. Ambiguous outcomes retain capacity. db().transaction(() => { db().prepare('DELETE FROM restart_rollout WHERE operation_id=?').run(job.id); const files = storedFiles(job); diff --git a/src/lib/server/retention.ts b/src/lib/server/retention.ts index 12c6773..d1f160f 100644 --- a/src/lib/server/retention.ts +++ b/src/lib/server/retention.ts @@ -19,7 +19,10 @@ const unsettledDeployment = `d.state NOT IN ${terminal} OR EXISTS (SELECT 1 FROM restart_rollout r WHERE r.operation_id=d.id AND r.healthy_at IS NULL) OR EXISTS (SELECT 1 FROM native_provisioning n JOIN ec_host_operations a ON a.id IN (n.operation_id,n.stage_operation_id) LEFT JOIN native_instances i ON i.id=n.instance_id WHERE n.deployment_id=d.id AND - (a.state NOT IN ('completed','failed') OR (a.id=n.operation_id AND a.state='completed' AND ( + (a.state NOT IN ('completed','failed') + OR (a.state='failed' AND a.kind='instance.rebuild' AND a.id=n.operation_id + AND json_type(a.result,'$.rebuildFailure') IS NOT NULL AND n.applied_at IS NULL) + OR (a.id=n.operation_id AND a.state='completed' AND ( (n.applied_at IS NULL AND COALESCE(json_array_length(n.files_json),0)>0) OR (d.action='destroy' AND i.retired_at IS NULL) OR (a.kind='instance.rebuild' AND i.retired_at IS NULL AND i.generation=n.generation))))) diff --git a/static/agent/ec-agent.py b/static/agent/ec-agent.py index 6c7583c..c883576 100644 --- a/static/agent/ec-agent.py +++ b/static/agent/ec-agent.py @@ -35,7 +35,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.8.3" +VERSION = "1.8.4" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -2665,12 +2665,84 @@ class Agent: atomic_json(self.config_path, self.config) return True + def reconcile_uncreated_rebuild(self, operation, row): + """Settle removal only when the replacement create was never submitted. + + The write lane and instance lock exclude the original Python execution. + A terminal journal receipt prevents any later replay of that command. + """ + if row["state"] != "unknown": + return row + instance_id, generation = instance_identity(operation.get("instanceId"), operation.get("generation")) + payload = operation.get("payload", {}) + next_generation = payload.get("nextGeneration", payload.get("newGeneration")) + if type(next_generation) is not int or next_generation <= generation: + return row + previous = json.loads(row["result"]) if row.get("result") else {} + if not isinstance(previous, dict): + return row + directory = scoped(self.root, f"instances/{instance_id}/{next_generation}") + + def unchanged_generation(): + try: + marker = scoped(self.root, f"instances/{instance_id}/current.json") + current = json.loads(marker.read_text()) + except FileNotFoundError: + return True + return isinstance(current, dict) and type(current.get("generation")) is int and current["generation"] == generation and current.get("removed", False) is False + + def both_missing(): + for target in (generation, next_generation): + health = self.inspect(instance_id, target) + if health.get("exists") is not False or health.get("running") is not False or health.get("state") != "missing": + return False + return True + + if not unchanged_generation() or not both_missing(): + return row + # Preparation can leave runtime/configuration files before create's + # checkpoint. A fresh operation must not inherit those partial files. + expected_spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration", "unhealthyRecovery")} + try: + saved_spec = json.loads(scoped(directory, "spec.json").read_text()) + except FileNotFoundError: + pass + else: + if saved_spec != expected_spec: + return row + try: + metadata = directory.lstat() + except FileNotFoundError: + pass + else: + if not stat_module.S_ISDIR(metadata.st_mode): + return row + shutil.rmtree(directory) + # Also sync when a previous reconciliation removed the directory but + # crashed before recording the terminal receipt. + try: + sync_parent(directory) + except FileNotFoundError: + pass + if not unchanged_generation() or not both_missing(): + return row + original_error = row.get("error") + result = {**previous, "reconciled": True, "originalError": original_error, + "rebuildFailure": {"phase": "disrupting", "generation": generation, "nextGeneration": next_generation, + "oldContainerMissing": True, "replacementContainerMissing": True, "replacementFilesRemoved": True}} + message = "Old container removed; replacement creation failed." + if original_error: + message += " " + original_error + return self.journal.update(operation["id"], "failed", result, error=message) + def reconcile(self, operation, row): """Prove desired state with immutable ownership; never resubmit a command.""" kind, payload = operation["kind"], operation.get("payload", {}) instance_id, generation = operation.get("instanceId"), operation.get("generation") checkpoint = json.loads(row["checkpoint"]) if row.get("checkpoint") else {} try: + if kind == "instance.rebuild" and checkpoint.get("phase") == "disrupting": + return self.reconcile_uncreated_rebuild(operation, row) if kind == "instance.remove" and "autoscaling" in payload and not checkpoint: return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error="Interrupted autoscaling removal requires a new decision") if kind == "instance.cancel-create": diff --git a/static/agent/manifest.json b/static/agent/manifest.json index acab6d2..2be4dff 100644 --- a/static/agent/manifest.json +++ b/static/agent/manifest.json @@ -1,8 +1,8 @@ { - "version": "1.8.3", + "version": "1.8.4", "protocolVersion": 1, "platform": "debian-amd64", - "sha256": "388569ffd4566dee63584970838d4d3ae7d2b8ef92675ae7627c31389dcbfec0", + "sha256": "7e7c0049161a6f64743210b787a8ec30bd32d726c30587962bc5c705d94dfbac", "updaterProtocolVersion": 1, "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" } diff --git a/tests/unit/native-provisioning.test.ts b/tests/unit/native-provisioning.test.ts index 478f5d8..0034e0d 100644 --- a/tests/unit/native-provisioning.test.ts +++ b/tests/unit/native-provisioning.test.ts @@ -1946,6 +1946,141 @@ describe('native provisioning cancellation and rollback', () => { }); }); +describe('partial rebuild failure reconciliation', () => { + const originalError = 'Replacement game library could not be prepared'; + const failure = (generation = 1) => ({ reconciled: true, originalError, rebuildFailure: { + phase: 'disrupting', generation, nextGeneration: generation + 1, + oldContainerMissing: true, replacementContainerMissing: true, replacementFilesRemoved: true + } }); + + async function cancelledRebuild() { + const job = await queue(); await advance(); await acknowledge(job.deploymentId); + const { requestLifecycle } = await import('../../src/lib/server/provisioning/requests.js'); + const first = requestLifecycle(job.name, 'rebuild'); + if (!first.ok) throw new Error(first.error); + await advance(); + const command = (await operationFor(first.deploymentId))!; + const { pendingHostOperations, reportHostOperation } = await import('../../src/lib/server/hosts/index.js'); + pendingHostOperations(hostId); + reportHostOperation(hostId, command.id, { state: 'unknown', error: originalError }); + const { cancelProvisioningJob } = await import('../../src/lib/server/provisioning/cancellation.js'); + expect(cancelProvisioningJob(first.deploymentId)).toBe(true); + const replacement = requestLifecycle(job.name, 'rebuild'); + if (!replacement.ok) throw new Error(replacement.error); + return { job, first, command, replacement }; + } + + it('records missing containers before releasing a queued rebuild without replaying the cancelled command', async () => { + const { job, first, command, replacement } = await cancelledRebuild(); + const { db, closeDb } = await import('../../src/lib/server/db/index.js'); + const { getDeployment } = await import('../../src/lib/server/provisioning/history.js'); + const { getHostOperation, reportHostOperation, pendingHostOperations } = await import('../../src/lib/server/hosts/index.js'); + const { nativeInstance, nativeRequest } = await import('../../src/lib/server/backends/native.js'); + const { getServerKey } = await import('../../src/lib/server/fleet/server-keys.js'); + const { unsettledCancelledCommand } = await import('../../src/lib/server/provisioning/cancellation.js'); + const ports = db().prepare('SELECT host_id,instance_id,port,protocol FROM ec_host_ports ORDER BY protocol').all(); + await advance(); + expect(await operationFor(replacement.deploymentId)).toBeNull(); + expect(getDeployment(replacement.deploymentId)?.progress?.blocker).toMatchObject({ code: 'cancelled-command', + href: `/provisioning/${first.deploymentId}`, label: 'View cancelled operation', + message: expect.stringContaining(originalError) }); + expect(getDeployment(replacement.deploymentId)?.progress?.blocker?.message).toContain('unknown'); + reportHostOperation(hostId, command.id, { state: 'failed', + error: `Old container removed; replacement creation failed. ${originalError}`, result: failure() }); + expect(unsettledCancelledCommand(job.name, replacement.deploymentId)).not.toBeNull(); + expect(db().prepare('SELECT applied_at FROM native_provisioning WHERE deployment_id=?').get(first.deploymentId)) + .toEqual({ applied_at: null }); + remote.state = 'missing'; + vi.mocked(nativeRequest).mockImplementationOnce(async () => { + expect(db().prepare('SELECT container_state,last_players FROM servers WHERE name=?').get(job.name)) + .toEqual({ container_state: 'missing', last_players: null }); + expect(db().prepare('SELECT applied_at FROM native_provisioning WHERE deployment_id=?').get(first.deploymentId)) + .toEqual({ applied_at: expect.any(String) }); + expect(unsettledCancelledCommand(job.name, replacement.deploymentId)).toBeNull(); + return { exists: false, healthy: false, state: 'missing' } as never; + }); + closeDb(); await advance(); + const next = (await operationFor(replacement.deploymentId))!; + expect(next).toMatchObject({ kind: 'instance.rebuild', generation: 1, instanceId: command.instanceId, + payload: { nextGeneration: 2, env: { PLUTO_SERVER_KEY: 'saved-key' } } }); + expect(next.id).not.toBe(command.id); + expect(nativeInstance(job.name)).toMatchObject({ id: command.instanceId, generation: 1, retired_at: null }); + expect(getServerKey(job.name)).toBe('saved-key'); + expect(db().prepare('SELECT host_id,instance_id,port,protocol FROM ec_host_ports ORDER BY protocol').all()).toEqual(ports); + expect((await detail(first.deploymentId)).state).toBe('cancelled'); + expect(getHostOperation(command.id)).toMatchObject({ state: 'failed', result: failure() }); + expect(pendingHostOperations(hostId).some(operation => operation.id === command.id)).toBe(false); + await advance(); + expect((await operationFor(replacement.deploymentId))!.id).toBe(next.id); + expect(db().prepare("SELECT COUNT(*) AS count FROM ec_host_operations WHERE kind='instance.rebuild'").get()).toEqual({ count: 2 }); + remote.state = 'running'; await acknowledge(replacement.deploymentId); + expect((await detail(replacement.deploymentId)).state).toBe('done'); + expect(nativeInstance(job.name)?.generation).toBe(2); + expect((await detail(first.deploymentId)).state).toBe('cancelled'); + }); + + it.each(['disrupting', 'creating'])('keeps an ambiguous %s rebuild blocked even when inspections currently see no containers', async phase => { + const { job, first, command, replacement } = await cancelledRebuild(); + const { reportHostOperation } = await import('../../src/lib/server/hosts/index.js'); + reportHostOperation(hostId, command.id, { state: 'unknown', error: originalError, + result: { checkpoint: { phase }, state: 'missing', oldContainerMissing: true, replacementContainerMissing: true } }); + remote.state = 'missing'; + const { nativeRequest, nativeInstance } = await import('../../src/lib/server/backends/native.js'); + vi.mocked(nativeRequest).mockClear(); + const { closeDb, db } = await import('../../src/lib/server/db/index.js'); closeDb(); + await advance(); await advance(); + expect(await operationFor(replacement.deploymentId)).toBeNull(); + expect(vi.mocked(nativeRequest)).not.toHaveBeenCalled(); + expect((await operationFor(first.deploymentId))!.id).toBe(command.id); + expect(nativeInstance(job.name)?.generation).toBe(1); + expect(db().prepare('SELECT applied_at FROM native_provisioning WHERE deployment_id=?').get(first.deploymentId)) + .toEqual({ applied_at: null }); + expect(db().prepare("SELECT COUNT(*) AS count FROM ec_host_operations WHERE kind='instance.rebuild'").get()).toEqual({ count: 1 }); + }); + + it.each(['creation started', 'replacement files remain', 'wrong target generation', 'command target mismatch'])('does not release a replacement from invalid partial evidence: %s', async evidence => { + const { first, command, replacement } = await cancelledRebuild(); + const result = failure(); + if (evidence === 'creation started') result.rebuildFailure.phase = 'creating'; + if (evidence === 'replacement files remain') result.rebuildFailure.replacementFilesRemoved = false; + if (evidence === 'wrong target generation') result.rebuildFailure.nextGeneration = 3; + const { db } = await import('../../src/lib/server/db/index.js'); + if (evidence === 'command target mismatch') db().prepare("UPDATE ec_host_operation_payloads SET payload=json_set(payload,'$.nextGeneration',3) WHERE operation_id=?") + .run(command.id); + const { reportHostOperation } = await import('../../src/lib/server/hosts/index.js'); + reportHostOperation(hostId, command.id, { state: 'failed', error: originalError, result }); + remote.state = 'missing'; await advance(); + expect(await operationFor(replacement.deploymentId)).toBeNull(); + expect(db().prepare('SELECT applied_at FROM native_provisioning WHERE deployment_id=?').get(first.deploymentId)) + .toEqual({ applied_at: null }); + expect(db().prepare("SELECT COUNT(*) AS count FROM ec_host_operations WHERE kind='instance.rebuild'").get()).toEqual({ count: 1 }); + }); + + it('records an uncancelled partial rebuild failure without claiming replacement generation or successful health', async () => { + const job = await queue(); await advance(); await acknowledge(job.deploymentId); + const { requestLifecycle } = await import('../../src/lib/server/provisioning/requests.js'); + const first = requestLifecycle(job.name, 'rebuild'); + if (!first.ok) throw new Error(first.error); + await advance(); + const command = (await operationFor(first.deploymentId))!; + const { reportHostOperation } = await import('../../src/lib/server/hosts/index.js'); + reportHostOperation(hostId, command.id, { state: 'failed', result: failure(), + error: `Old container removed; replacement creation failed. ${originalError}` }); + await advance(); + const { nativeInstance } = await import('../../src/lib/server/backends/native.js'); + const { db } = await import('../../src/lib/server/db/index.js'); + expect((await detail(first.deploymentId)).state).toBe('failed'); + expect(nativeInstance(job.name)).toMatchObject({ generation: 1, retired_at: null }); + expect(db().prepare('SELECT container_state,last_players FROM servers WHERE name=?').get(job.name)) + .toEqual({ container_state: 'missing', last_players: null }); + expect(db().prepare('SELECT applied_at,verified_at FROM native_provisioning WHERE deployment_id=?').get(first.deploymentId)) + .toEqual({ applied_at: expect.any(String), verified_at: null }); + expect(db().prepare('SELECT COUNT(*) AS count FROM ec_host_ports WHERE instance_id=?').get(command.instanceId)).toEqual({ count: 2 }); + const { getServerKey } = await import('../../src/lib/server/fleet/server-keys.js'); + expect(getServerKey(job.name)).toBe('saved-key'); + }); +}); + describe('cancelled native mutation bookkeeping', () => { it.each([null, 'before beta'])('keeps the original host receipt across beta revisions (original=%s)', async original => { const job = await queue(); await advance(); await acknowledge(job.deploymentId); diff --git a/tests/unit/retention.test.ts b/tests/unit/retention.test.ts index 2aacf5b..f8d97d2 100644 --- a/tests/unit/retention.test.ts +++ b/tests/unit/retention.test.ts @@ -231,6 +231,35 @@ it('protects cancelled workflows and linked file history through ambiguous and u expect(a.db.pragma('foreign_key_check')).toEqual([]); }); +it('retains a cancelled rebuild and its source until the partial-failure receipt is applied', async () => { + const a = await api(); + const hosts = await import('../../src/lib/server/hosts/index.js'); + const host = hosts.createHost({ name: 'Host' }); + a.db.prepare("INSERT INTO servers(name,container_state) VALUES('s','running')").run(); + a.db.prepare("INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at) VALUES('instance','s',?,'lib','image',1,?)").run(host.id, old); + a.db.prepare("INSERT INTO server_deployments(id,server,action,state,stage,started_at,finished_at,triggered_by) VALUES('cancelled','s','rebuild','cancelled','cancelled',?,?,'test')").run(old, old); + const command = hosts.queueHostOperation(host.id, 'instance.rebuild', { nextGeneration: 2 }, { instanceId: 'instance', generation: 1 }); + a.db.prepare("INSERT INTO native_provisioning(deployment_id,instance_id,generation,operation_id) VALUES('cancelled','instance',1,?)").run(command.id); + a.db.prepare("INSERT INTO deployments(id,action,started_at,finished_at) VALUES('campaign','promote',?,?)").run(old, old); + a.db.prepare("INSERT INTO provisioning_sources(operation_id,kind,source_id,server) VALUES('cancelled','campaign','campaign','s')").run(); + hosts.reportHostOperation(host.id, command.id, { state: 'failed', error: 'Replacement creation failed', result: { + reconciled: true, rebuildFailure: { phase: 'disrupting', generation: 1, nextGeneration: 2, + oldContainerMissing: true, replacementContainerMissing: true, replacementFilesRemoved: true } + } }); + a.saveFleetSettings({ ...a.fleetSettings(), deployKeepDays: 1, provisioningKeepDays: 1, historyKeepCount: 0 }); + expect(a.pruneHistory('campaign')).toBe(0); + expect(a.pruneHistory('provisioning')).toBe(0); + expect(a.db.prepare("SELECT applied_at FROM native_provisioning WHERE deployment_id='cancelled'").get()).toEqual({ applied_at: null }); + + const { reconcileNativeProvisioningReceipt } = await import('../../src/lib/server/provisioning/native.js'); + reconcileNativeProvisioningReceipt('cancelled'); + expect(a.db.prepare("SELECT applied_at FROM native_provisioning WHERE deployment_id='cancelled'").get()).toEqual({ applied_at: expect.any(String) }); + expect(a.db.prepare("SELECT container_state FROM servers WHERE name='s'").get()).toEqual({ container_state: 'missing' }); + expect(a.pruneHistory('campaign')).toBe(1); + expect(a.pruneHistory('provisioning')).toBe(1); + expect(a.db.pragma('foreign_key_check')).toEqual([]); +}); + async function recoveryLaunch(action: 'create' | 'rebuild' | 'migration') { const a = await api(); const hosts = await import('../../src/lib/server/hosts/index.js');