diff --git a/agent/ec_agent.py b/agent/ec_agent.py index 01cb0df..e50a31d 100644 --- a/agent/ec_agent.py +++ b/agent/ec_agent.py @@ -33,7 +33,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.8.0" +VERSION = "1.8.1" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -1394,14 +1394,18 @@ class Agent: if line.startswith("PRETTY_NAME="): os_name = line.split("=", 1)[1].strip('"') update = self.update_inventory() - capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else []) + capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else []) plutonium_cache = self.plutonium_cache_inventory() return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), **({"plutoniumCache": plutonium_cache} if plutonium_cache else {}), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})} - def safe_to_disrupt(self, instance_id, generation): + def safe_to_disrupt(self, instance_id, generation, *, allow_unhealthy=False): health = self.inspect(instance_id, generation) if health["state"] in ("missing", "stopped", "exited", "created", "dead"): return + # Provisioning recovery may treat explicitly unhealthy containers as + # empty. Reinspect here so recovered containers still require occupancy. + if allow_unhealthy and health["state"] == "unhealthy": + return if not health["running"]: raise Held("Container state is unknown; disruption held") result = self.rcon(instance_id, generation, "status") @@ -1857,6 +1861,9 @@ class Agent: raise AutoscalingInvalidated("Autoscaling removal authorization expired before mutation") def lifecycle(self, kind, instance_id, generation, payload): + unhealthy_recovery = payload.get("unhealthyRecovery") is True + if unhealthy_recovery and (kind not in ("instance.restart", "instance.rebuild") or payload.get("stageId") or "autoscaling" in payload): + raise AgentError("Unhealthy recovery only permits restart or rebuild without file activation or autoscaling") if kind == "instance.cancel-create": return self.cancel_creation(instance_id, generation, payload) if kind == "instance.create": @@ -1891,10 +1898,10 @@ class Agent: next_generation = payload.get("nextGeneration", payload.get("newGeneration")) if isinstance(next_generation, bool) or not isinstance(next_generation, int) or next_generation <= generation: raise AgentError("Rebuild must use a newer immutable generation") - spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration")} + spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration", "unhealthyRecovery")} self.validate_spec(spec) if not override_safety: - self.safe_to_disrupt(instance_id, generation) + self.safe_to_disrupt(instance_id, generation, allow_unhealthy=unhealthy_recovery) if "autoscaling" in payload: self.authorize_autoscaling_removal(instance_id, generation, payload) self.checkpoint({"phase": "disrupting", "containerId": health.get("containerId"), "startedAt": health.get("startedAt")}) @@ -1928,29 +1935,46 @@ class Agent: Repeating this cleanup after an ambiguous Docker result is safe. """ create_id = identifier(payload.get("createOperationId")) + targets = [(generation, create_id)] + if "recoveryRebuildOperationId" in payload or "recoveryGeneration" in payload: + rebuild_id = identifier(payload.get("recoveryRebuildOperationId")) + recovery_generation = payload.get("recoveryGeneration") + if isinstance(recovery_generation, bool) or not isinstance(recovery_generation, int) or recovery_generation != generation + 1: + raise Rejected("Cancellation recovery must target the next generation") + saved = self.journal.db.execute("SELECT request FROM operations WHERE id=?", (rebuild_id,)).fetchone() + if saved: + rebuild = json.loads(saved["request"]) + rebuild_payload = rebuild.get("payload", {}) + if rebuild["kind"] != "instance.rebuild" or rebuild["instanceId"] != instance_id or rebuild["generation"] != generation or rebuild_payload.get("nextGeneration", rebuild_payload.get("newGeneration")) != recovery_generation or rebuild_payload.get("unhealthyRecovery") is not True: + raise Rejected("Cancellation does not match the creation recovery rebuild") + targets.append((recovery_generation, rebuild_id)) marker = self.instance_dir(instance_id, generation).parent / "current.json" - if marker.exists() and json.loads(marker.read_text())["generation"] > generation: + if marker.exists() and json.loads(marker.read_text())["generation"] > targets[-1][0]: raise Rejected("Cancelled creation was superseded by a newer generation") saved = self.journal.db.execute("SELECT request FROM operations WHERE id=?", (create_id,)).fetchone() if saved: original = json.loads(saved["request"]) if original["kind"] != "instance.create" or original["instanceId"] != instance_id or original["generation"] != generation: raise Rejected("Cancellation does not match the original creation") - health = self.inspect(instance_id, generation) - if health["exists"] and health.get("creationOperation") != create_id: - raise Rejected("Container belongs to a different creation operation") + health_by_generation = {} + for target_generation, operation_id in targets: + health = self.inspect(instance_id, target_generation) + if health["exists"] and health.get("creationOperation") != operation_id: + raise Rejected("Container belongs to a different creation operation") + health_by_generation[target_generation] = health self.checkpoint({"phase": "cancelling-creation", "createOperationId": create_id}) - # Fence before deleting: a crash must never permit the old creation to run again. - self.mark_removed(instance_id, generation) - if health["exists"]: - self.docker.call("container", "rm", "--force", "--volumes", self.container_name(instance_id, generation)) - after = self.inspect(instance_id, generation) - if after["exists"]: - raise AgentError("Cancelled creation container still exists") - directory = self.instance_dir(instance_id, generation) - if directory.exists(): - shutil.rmtree(directory) - return {**after, "cancelledCreation": create_id} + # Fence both generations before deleting: delayed create/rebuild work + # must never recreate either target after cancellation or a crash. + self.mark_removed(instance_id, targets[-1][0]) + for target_generation, _ in targets: + if health_by_generation[target_generation]["exists"]: + self.docker.call("container", "rm", "--force", "--volumes", self.container_name(instance_id, target_generation)) + if self.inspect(instance_id, target_generation)["exists"]: + raise AgentError("Cancelled creation container still exists") + directory = self.instance_dir(instance_id, target_generation) + if directory.exists(): + shutil.rmtree(directory) + return {**self.inspect(instance_id, generation), "cancelledCreation": create_id} def stage(self, instance_id, generation, payload, operation_id): directory = self.instance_dir(instance_id, generation) @@ -2453,7 +2477,7 @@ class Agent: current_marker = self.instance_dir(instance_id, generation).parent / "current.json" if current_marker.exists(): current = json.loads(current_marker.read_text()) - if current["generation"] > generation or current["generation"] == generation and current.get("removed") and kind not in ("instance.inspect", "instance.remove", "instance.cancel-create"): + if kind != "instance.cancel-create" and (current["generation"] > generation or current["generation"] == generation and current.get("removed") and kind not in ("instance.inspect", "instance.remove")): raise AgentError("Instance generation has been superseded or removed") if kind.startswith("instance."): return self.lifecycle(kind, instance_id, generation, payload) diff --git a/agent/tests/test_agent.py b/agent/tests/test_agent.py index 8e93378..7316b21 100644 --- a/agent/tests/test_agent.py +++ b/agent/tests/test_agent.py @@ -1043,6 +1043,112 @@ class AgentTests(unittest.TestCase): self.assertEqual(self.agent.execute(self.operation("instance.stop"))["state"], "accepted") self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls)) + def assert_unhealthy_recovery(self, kind): + spec = self.create() + self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} + payload = {"unhealthyRecovery": True} + if kind == "instance.rebuild": + payload.update({**spec, "nextGeneration": 2}) + operation = self.operation(kind, payload) + with patch.object(self.agent, "rcon", side_effect=AssertionError("Unhealthy recovery queried player status")): + result = self.agent.execute(operation) + self.assertEqual(result["state"], "completed", result) + calls = list(self.docker.calls) + self.assertEqual(self.agent.execute(operation)["state"], "completed") + self.assertEqual(self.docker.calls, calls) + target_generation = 2 if kind == "instance.rebuild" else 1 + self.assertTrue(self.agent.inspect(INSTANCE, target_generation)["healthy"]) + self.assertNotIn("unhealthyRecovery", self.agent.spec(INSTANCE, target_generation)) + + def test_unhealthy_recovery_restarts_without_player_status(self): + self.assert_unhealthy_recovery("instance.restart") + self.assertIn("instance.unhealthy-recovery", self.agent.inventory()["capabilities"]) + + def test_unhealthy_recovery_rebuilds_without_player_status(self): + self.assert_unhealthy_recovery("instance.rebuild") + + def test_unhealthy_recovery_keeps_safety_for_other_health_states(self): + spec = self.create() + container = self.docker.containers[self.agent.container_name(INSTANCE, 1)] + for kind in ("instance.restart", "instance.rebuild"): + payload = {"unhealthyRecovery": True} + if kind == "instance.rebuild": + payload.update({"spec": spec, "nextGeneration": 2}) + for state in ("healthy", "starting", "restarting", "unknown"): + with self.subTest(kind=kind, state=state): + container["State"] = {"Status": "unknown" if state == "unknown" else "running", + "Running": state != "unknown", "Health": {"Status": state}} + with patch.object(self.agent, "rcon", return_value={"ok": True, "players": 2}) as rcon: + result = self.agent.execute(self.operation(kind, payload, f"{kind.split('.')[-1]}-{state}")) + self.assertEqual(result["state"], "accepted", result) + self.assertEqual(rcon.call_count, 0 if state == "unknown" else 1) + self.assertFalse(any(call[:2] in (("container", "restart"), ("container", "stop"), ("container", "rm")) for call in self.docker.calls)) + + def test_unhealthy_recovery_rechecks_health_before_disruption(self): + spec = self.create() + container = self.docker.containers[self.agent.container_name(INSTANCE, 1)] + inspect = self.docker.inspect + for kind in ("instance.restart", "instance.rebuild"): + container["State"]["Health"] = {"Status": "unhealthy"} + observations = [] + + def recovered(name): + observations.append(name) + if len(observations) == 2: + container["State"]["Health"] = {"Status": "healthy"} + return inspect(name) + + payload = {"unhealthyRecovery": True} + if kind == "instance.rebuild": + payload.update({"spec": spec, "nextGeneration": 2}) + with patch.object(self.docker, "inspect", side_effect=recovered), patch.object(self.agent, "rcon", return_value={"ok": True, "players": 2}) as rcon: + result = self.agent.execute(self.operation(kind, payload, kind.split(".")[-1])) + self.assertEqual(result["state"], "accepted", result) + rcon.assert_called_once() + self.assertFalse(any(call[:2] in (("container", "restart"), ("container", "stop"), ("container", "rm")) for call in self.docker.calls)) + + def test_unhealthy_recovery_requires_boolean_true(self): + self.create() + self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} + for index, value in enumerate((False, None, "true", 1, {})): + with self.subTest(value=value), patch.object(self.agent, "rcon", return_value={"ok": False, "players": None, "reason": "unreachable"}) as rcon: + result = self.agent.execute(self.operation("instance.restart", {"unhealthyRecovery": value}, f"attempt-{index}")) + self.assertEqual(result["state"], "accepted", result) + rcon.assert_called_once() + self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls)) + + def test_unhealthy_recovery_rejects_other_lifecycle_actions_and_activation(self): + self.create() + self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} + attempts = [(kind, {}) for kind in ("instance.stop", "instance.remove", "instance.start", "instance.create")] + attempts.extend([("instance.restart", {"stageId": "campaign"}), ("instance.restart", {"autoscaling": {}})]) + calls = list(self.docker.calls) + for index, (kind, payload) in enumerate(attempts): + with self.subTest(kind=kind, payload=payload): + result = self.agent.execute(self.operation(kind, {**payload, "unhealthyRecovery": True}, f"attempt-{index}")) + self.assertEqual(result["state"], "failed", result) + self.assertIn("Unhealthy recovery only permits", result["error"]) + self.assertEqual(self.docker.calls, calls) + with patch.object(self.agent, "rcon", return_value={"ok": True, "players": 2}) as rcon: + result = self.agent.execute(self.operation("files.activate", {"stageId": "campaign", "unhealthyRecovery": True})) + self.assertEqual(result["state"], "accepted", result) + rcon.assert_called_once() + + def test_unhealthy_recovery_preserves_ownership_and_generation_checks(self): + spec = self.create() + container = self.docker.containers[self.agent.container_name(INSTANCE, 1)] + container["State"]["Health"] = {"Status": "unhealthy"} + labels = container["Config"]["Labels"] + container["Config"]["Labels"] = {} + result = self.agent.execute(self.operation("instance.restart", {"unhealthyRecovery": True}, "unowned")) + self.assertEqual(result["state"], "failed", result) + container["Config"]["Labels"] = labels + result = self.agent.execute(self.operation("instance.rebuild", {"unhealthyRecovery": True, "spec": spec, "nextGeneration": 1}, "same-generation")) + self.assertEqual(result["state"], "failed", result) + result = self.agent.execute(self.operation("instance.restart", {"unhealthyRecovery": True}, "stale", generation=0)) + self.assertEqual(result["state"], "failed", result) + self.assertFalse(any(call[:2] in (("container", "restart"), ("container", "stop"), ("container", "rm")) for call in self.docker.calls)) + def assert_override_disruption(self, kind): spec = self.create() payload = {"safetyOverride": True} @@ -1241,6 +1347,88 @@ class AgentTests(unittest.TestCase): self.assertEqual(result["state"], "failed") self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls)) + def test_cancel_creation_removes_completed_recovery_rebuild(self): + spec = self.create() + self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} + rebuild = self.operation("instance.rebuild", {"spec": spec, "nextGeneration": 2, "unhealthyRecovery": True}, "rebuild") + self.assertEqual(self.agent.execute(rebuild)["state"], "completed") + cancel = self.operation("instance.cancel-create", {"createOperationId": "create", "recoveryRebuildOperationId": "rebuild", "recoveryGeneration": 2}, "cancel") + with patch.object(self.agent, "rcon", side_effect=AssertionError("Cancellation queried player status")): + result = self.agent.execute(cancel) + self.assertEqual(result["state"], "completed", result) + self.assertFalse(json.loads(result["result"])["exists"]) + for generation in (1, 2): + self.assertFalse(self.agent.inspect(INSTANCE, generation)["exists"]) + self.assertFalse(self.agent.instance_dir(INSTANCE, generation).exists()) + self.assertEqual(json.loads((self.agent.instance_dir(INSTANCE, 1).parent / "current.json").read_text()), {"generation": 2, "removed": True}) + calls = list(self.docker.calls) + self.assertEqual(self.agent.execute(cancel)["state"], "completed") + self.assertEqual(self.docker.calls, calls) + + def test_cancel_creation_fences_undelivered_recovery_rebuild(self): + spec = self.create() + cancel = self.operation("instance.cancel-create", {"createOperationId": "create", "recoveryRebuildOperationId": "late-rebuild", "recoveryGeneration": 2}, "cancel") + self.assertEqual(self.agent.execute(cancel)["state"], "completed") + rebuild = self.operation("instance.rebuild", {"spec": spec, "nextGeneration": 2, "unhealthyRecovery": True}, "late-rebuild") + result = self.agent.execute(rebuild) + self.assertEqual(result["state"], "failed", result) + self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"]) + self.assertFalse(self.agent.inspect(INSTANCE, 2)["exists"]) + self.assertEqual(sum(call[:2] == ("container", "create") for call in self.docker.calls), 1) + + def test_cancel_creation_reconciles_recovery_rebuild_removal(self): + spec = self.create() + self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"} + self.docker.lose_reply_after = "start" + rebuild = self.operation("instance.rebuild", {"spec": spec, "nextGeneration": 2, "unhealthyRecovery": True}, "rebuild") + self.assertEqual(self.agent.execute(rebuild)["state"], "unknown") + self.docker.lose_reply_after = "rm" + cancel = self.operation("instance.cancel-create", {"createOperationId": "create", "recoveryRebuildOperationId": "rebuild", "recoveryGeneration": 2}, "cancel") + self.assertEqual(self.agent.execute(cancel)["state"], "unknown") + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + self.assertEqual(self.agent.execute(cancel)["state"], "completed") + for generation in (1, 2): + self.assertFalse(self.agent.inspect(INSTANCE, generation)["exists"]) + self.assertFalse(self.agent.instance_dir(INSTANCE, generation).exists()) + + def test_cancel_creation_rejects_mismatched_recovery_rebuild(self): + spec = self.create() + for index, (kind, instance_id, generation, changes) in enumerate(( + ("instance.restart", INSTANCE, 1, {}), + ("instance.rebuild", "another-instance", 1, {}), + ("instance.rebuild", INSTANCE, 2, {}), + ("instance.rebuild", INSTANCE, 1, {"nextGeneration": 3}), + ("instance.rebuild", INSTANCE, 1, {"unhealthyRecovery": False}), + )): + rebuild = self.operation(kind, {"spec": spec, "nextGeneration": 2, "unhealthyRecovery": True, **changes}, f"rebuild-{index}", generation=generation) + rebuild["instanceId"] = instance_id + self.agent.journal.accept(rebuild) + cancel = self.operation("instance.cancel-create", {"createOperationId": "create", "recoveryRebuildOperationId": rebuild["id"], "recoveryGeneration": 2}, f"cancel-{index}") + result = self.agent.execute(cancel) + self.assertEqual(result["state"], "failed", result) + self.assertIn("does not match", result["error"]) + self.assertTrue(self.agent.inspect(INSTANCE, 1)["exists"]) + self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls)) + + def test_cancel_creation_rejects_unrelated_recovery_container_or_newer_generation(self): + self.create() + original = self.docker.containers[self.agent.container_name(INSTANCE, 1)] + unrelated = json.loads(json.dumps(original)) + unrelated["Config"]["Labels"].update({module.LABEL + ".generation": "2", module.LABEL + ".operation": "unrelated"}) + self.docker.containers[self.agent.container_name(INSTANCE, 2)] = unrelated + payload = {"createOperationId": "create", "recoveryRebuildOperationId": "rebuild", "recoveryGeneration": 2} + result = self.agent.execute(self.operation("instance.cancel-create", payload, "unrelated-container")) + self.assertEqual(result["state"], "failed", result) + self.assertIn("different creation", result["error"]) + marker = self.agent.instance_dir(INSTANCE, 1).parent / "current.json" + module.atomic_json(marker, {"generation": 3}) + result = self.agent.execute(self.operation("instance.cancel-create", payload, "newer-generation")) + self.assertEqual(result["state"], "failed", result) + self.assertIn("superseded", result["error"]) + self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls)) + self.assertEqual(json.loads(marker.read_text()), {"generation": 3}) + def test_unowned_container_never_mutated(self): self.create() self.docker.containers[self.agent.container_name(INSTANCE, 1)]["Config"]["Labels"] = {} diff --git a/docs/guides/provisioning.md b/docs/guides/provisioning.md index bc4baf2..d97f018 100644 --- a/docs/guides/provisioning.md +++ b/docs/guides/provisioning.md @@ -71,6 +71,14 @@ making another request. ### Health and runtime checks +If the host agent explicitly reports an unhealthy container during Create or +Rebuild verification, that operation forces one restart, then forces one rebuild +if the container remains unhealthy. Recovery preserves the selected launch files, +credentials, server identity, and port. If both attempts leave it unhealthy, the +operation fails for inspection instead of repeating recovery. Starting, restarting, +and unknown container states continue waiting for readiness. This requires host +agent 1.8.1 or newer. + Create and Rebuild finish only after RCON reports a runtime revision at least as new as the current Plutonium production release. The worker checks the production manifest and records the runtime revision for fleet and server version status as @@ -114,6 +122,11 @@ campaign, beta, maintenance, or provisioning work. Duplicate pending requests fo the same action return the existing operation; a different action returns a conflict. Requests and events survive application restarts. +Rebuild and recovery within Create or Rebuild treat an explicitly unhealthy +container as empty without requiring a player count. The agent checks health +again before disruption; if the container has recovered, normal player checks +apply. This exception does not apply to unknown health or other lifecycle actions. + For a failed container stuck waiting for player status, an administrator can open its queued or waiting **Restart**, **Stop**, **Rebuild**, or **Destroy** operation and choose **Admin override…**. The confirmation modal explains the interruption diff --git a/src/lib/server/backends/schema.ts b/src/lib/server/backends/schema.ts index 5463496..3366f1c 100644 --- a/src/lib/server/backends/schema.ts +++ b/src/lib/server/backends/schema.ts @@ -46,6 +46,9 @@ export function migrateNativeInstances(handle: Database.Database): void { if (!(handle.pragma('table_info(ec_host_migrations)') as { name: string }[]).some(column => column.name === 'runtime_password')) { handle.exec('ALTER TABLE ec_host_migrations ADD COLUMN runtime_password TEXT'); } + if (!(handle.pragma('table_info(native_provisioning)') as { name: string }[]).some(column => column.name === 'recovery_step')) { + handle.exec("ALTER TABLE native_provisioning ADD COLUMN recovery_step TEXT CHECK (recovery_step IN ('restart','rebuild'))"); + } 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')) diff --git a/src/lib/server/hosts/index.ts b/src/lib/server/hosts/index.ts index 610c0c2..18a0924 100644 --- a/src/lib/server/hosts/index.ts +++ b/src/lib/server/hosts/index.ts @@ -466,10 +466,13 @@ export function listHostOperationHistory(page = 1, filter: HostOperationFilter = /** Bound each cleanup pass so retained history cannot stall monitoring or receipts. */ export function pruneHostObservations(): number { return pruneHistory('observation'); } -/** Heartbeats can tell an executing create to stop at its next safe command boundary. */ +/** Heartbeats stop either launch generation at its next safe command boundary. */ export function pendingCreationCancellations(hostId: string): string[] { - return (database().prepare(`SELECT json_extract(payload,'$.createOperationId') AS id FROM ec_host_operations - WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown')`).all(hostId) as { id: string }[]).map(row => row.id); + const rows = database().prepare(`SELECT json_extract(payload,'$.createOperationId') AS creation_id, + json_extract(payload,'$.recoveryRebuildOperationId') AS recovery_id FROM ec_host_operations + WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown')`) + .all(hostId) as { creation_id: string; recovery_id: string | null }[]; + return rows.flatMap(row => row.recovery_id ? [row.creation_id, row.recovery_id] : [row.creation_id]); } type QueueHostOperationOptions = { instanceId?: string; generation?: number; idempotencyKey?: string; id?: string }; export function queueHostOperation(hostId: string, kind: string, payload: Record, options: QueueHostOperationOptions = {}): HostOperation { @@ -537,8 +540,11 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = ' // busy indefinitely. Drain the old batch by withholding ALL further work; // heartbeats and receipts still flow, and the new capability releases this hold. // This must live in the controller so the fix also works before an agent update. - if (!host.inventory.capabilities?.includes('instance.cancel-create') && database().prepare(`SELECT 1 FROM ec_host_operations - WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown') LIMIT 1`).get(hostId)) return []; + if (database().prepare(`SELECT 1 FROM ec_host_operations + WHERE host_id=? AND kind='instance.cancel-create' AND state IN ('pending','accepted','running','unknown') + AND (NOT ? OR (json_extract(payload,'$.recoveryRebuildOperationId') IS NOT NULL AND NOT ?)) LIMIT 1`) + .get(hostId, Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false), + Number(host.inventory.capabilities?.includes('instance.unhealthy-recovery') ?? false))) return []; // Accepted/running/unknown commands keep the same immutable ID. The agent // consults its durable journal before acknowledging or executing a replay. const healthOnly = `(kind IN ('instance.inspect','rcon.status-batch') OR (kind='rcon.execute' AND json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version')))`; @@ -557,6 +563,9 @@ export function pendingHostOperationRows(hostId: string, lane: OperationLane = ' AND (kind!='instance.cancel-create' OR ?) AND NOT EXISTS (SELECT 1 FROM ec_host_operations cancellation WHERE cancellation.host_id=o.host_id AND cancellation.kind='instance.cancel-create' AND json_extract(cancellation.payload,'$.createOperationId')=o.id) + AND NOT EXISTS (SELECT 1 FROM native_creation_cancellations cancellation + JOIN native_provisioning creation ON creation.deployment_id=cancellation.creation_id + WHERE creation.operation_id=o.id) ${lane === 'health' ? "AND (state NOT IN ('accepted','running') OR MAX(updated_at,COALESCE(last_delivered_at,'')) <= ?)" : ''} ORDER BY CASE WHEN kind='instance.cancel-create' THEN 0 WHEN kind='plutonium.prepare' THEN 2 ELSE 1 END, COALESCE(last_delivered_at,''),created_at,rowid LIMIT ${limit}`) diff --git a/src/lib/server/provisioning/cancel-creation.ts b/src/lib/server/provisioning/cancel-creation.ts index 2fb1ccb..fa9ee86 100644 --- a/src/lib/server/provisioning/cancel-creation.ts +++ b/src/lib/server/provisioning/cancel-creation.ts @@ -2,14 +2,16 @@ import { pauseScalingForOperation } from '../autoscaling/ownership.js'; import { db } from '../db/index.js'; import { releaseServerKey } from '../fleet/server-keys.js'; import { nativeInstance } from '../backends/native.js'; -import { queueHostOperation, releaseHostPorts } from '../hosts/index.js'; +import { getHostOperationStatus, queueHostOperation, releaseHostPorts } from '../hosts/index.js'; import { beginDeployment, cancelRemoval, deploymentStage, finishDeployment } from './history.js'; import { recordProgress } from './progress.js'; export function canCancelCreation(id: string): boolean { return !!db().prepare(`SELECT 1 FROM server_deployments d LEFT JOIN native_provisioning n ON n.deployment_id=d.id - LEFT JOIN native_instances i ON i.id=n.instance_id AND i.generation=n.generation AND i.retired_at IS NULL + LEFT JOIN native_instances i ON i.id=n.instance_id AND i.retired_at IS NULL + AND (i.generation=n.generation OR (n.recovery_step='rebuild' AND i.generation=n.generation+1 + AND n.applied_at IS NOT NULL)) WHERE d.id=? AND d.action='create' AND d.server_deleted_at IS NULL AND (i.id IS NOT NULL OR (n.deployment_id IS NULL AND EXISTS (SELECT 1 FROM native_creation_intents intent WHERE intent.deployment_id=d.id) @@ -29,9 +31,9 @@ export function cancelCreation(id: string, triggeredBy = 'ui', pausePolicy = tru if (prior) return prior.cleanup_id; if (!canCancelCreation(id)) return null; if (pausePolicy) pauseScalingForOperation(id); - const job = db().prepare(`SELECT d.server,n.operation_id,n.stage_operation_id FROM server_deployments d + const job = db().prepare(`SELECT d.server,n.operation_id,n.stage_operation_id,n.generation,n.recovery_step FROM server_deployments d LEFT JOIN native_provisioning n ON n.deployment_id=d.id WHERE d.id=?`).get(id) as - { server: string; operation_id: string | null; stage_operation_id: string | null }; + { server: string; operation_id: string | null; stage_operation_id: string | null; generation: number | null; recovery_step: 'restart' | 'rebuild' | null }; const instance = nativeInstance(job.server); // Replace a removal still waiting on occupancy, including one whose asynchronous probe is outstanding. for (const removal of db().prepare("SELECT id FROM server_deployments WHERE server=? AND action='destroy' AND state='waiting' AND server_deleted_at IS NULL").all(job.server) as { id: string }[]) { @@ -54,8 +56,30 @@ export function cancelCreation(id: string, triggeredBy = 'ui', pausePolicy = tru finishCreationCancellation(cleanupId); } else { if (!instance) throw new Error('The original native instance is unavailable.'); - const operation = queueHostOperation(instance.host_id, 'instance.cancel-create', { createOperationId: job.operation_id ?? id }, - { instanceId: instance.id, generation: instance.generation, idempotencyKey: cleanupId }); + let generation = instance.generation; + const payload: { createOperationId: string; recoveryRebuildOperationId?: string; recoveryGeneration?: number } = { createOperationId: job.operation_id ?? id }; + if (job.recovery_step) { + // Recovery replaces operation_id, but cancellation must retain the + // immutable ownership label from the original launch. + const original = db().prepare(`SELECT id FROM ec_host_operations WHERE host_id=? AND instance_id=? + AND generation=? AND kind='instance.create' ORDER BY rowid DESC LIMIT 1`) + .get(instance.host_id, instance.id, job.generation) as { id: string } | undefined; + if (!original || !job.generation) throw new Error('The original creation command is unavailable.'); + payload.createOperationId = original.id; + generation = job.generation; + if (job.recovery_step === 'rebuild') { + const recovery = job.operation_id ? getHostOperationStatus(job.operation_id) : null; + if (!recovery || recovery.kind !== 'instance.rebuild' || recovery.instanceId !== instance.id || recovery.generation !== generation) { + throw new Error('The creation recovery command is unavailable.'); + } + // A delivered rebuild may have created the next generation + // before its receipt arrives. Fence and remove both identities. + payload.recoveryRebuildOperationId = recovery.id; + payload.recoveryGeneration = generation + 1; + } + } + const operation = queueHostOperation(instance.host_id, 'instance.cancel-create', payload, + { instanceId: instance.id, generation, idempotencyKey: cleanupId }); db().prepare('UPDATE native_provisioning SET operation_id=? WHERE deployment_id=?').run(operation.id, cleanupId); db().prepare('UPDATE server_deployments SET submitted_at=? WHERE id=?').run(new Date().toISOString(), cleanupId); deploymentStage(cleanupId, 'Waiting for host agent to cancel creation and remove its partial instance', { phase: 'container-action', blocker: null, diff --git a/src/lib/server/provisioning/concurrency.ts b/src/lib/server/provisioning/concurrency.ts index 6a004e4..4e13242 100644 --- a/src/lib/server/provisioning/concurrency.ts +++ b/src/lib/server/provisioning/concurrency.ts @@ -14,7 +14,7 @@ export function lifecycleHolders(): Holder[] { // Find capacity owners through their active state or outstanding command. // A correlated OR over d used to walk every retained job and fetch its agent // receipt on every admission, even when all that history was completed. - return db().prepare(`SELECT d.id,d.action,i.host_id,m.target_host_id, + return db().prepare(`SELECT d.id,CASE WHEN n.recovery_step='rebuild' THEN 'rebuild' ELSE d.action END AS action,i.host_id,m.target_host_id, CASE WHEN m.remove_operation_id IS NOT NULL AND m.removed_at IS NULL THEN 1 ELSE 0 END AS removing FROM server_deployments d LEFT JOIN native_provisioning n ON n.deployment_id=d.id diff --git a/src/lib/server/provisioning/history.ts b/src/lib/server/provisioning/history.ts index 898760f..9d713a7 100644 --- a/src/lib/server/provisioning/history.ts +++ b/src/lib/server/provisioning/history.ts @@ -107,7 +107,9 @@ function capacityFor(deployment: ProvisioningDeployment, evaluatedAt: string): P pools.push({ kind: 'startup', limit: settings.createConcurrency, used: holders.length, holders, requestOrder: order < 0 ? null : order + 1, ownsSlot: holders.some(holder => holder.id === deployment.id) }); } - if (['start', 'restart', 'rebuild'].includes(deployment.action)) { + const creationRecovery = deployment.action === 'create' && db().prepare(`SELECT 1 FROM native_provisioning + WHERE deployment_id=? AND recovery_step IS NOT NULL`).get(deployment.id); + if (['start', 'restart', 'rebuild'].includes(deployment.action) || creationRecovery) { const holders = db().prepare(`SELECT operation_id AS id, server FROM restart_rollout WHERE healthy_at IS NULL ORDER BY requested_at, server`).all() as { id: string | null; server: string }[]; pools.push({ kind: 'restart', limit: settings.restartConcurrency, used: holders.length, holders, diff --git a/src/lib/server/provisioning/launch-preparation.ts b/src/lib/server/provisioning/launch-preparation.ts index a019c95..2a2ea6c 100644 --- a/src/lib/server/provisioning/launch-preparation.ts +++ b/src/lib/server/provisioning/launch-preparation.ts @@ -15,6 +15,7 @@ export interface NativeLaunchInput { rconPassword: string; rebuild: boolean; safetyOverride?: boolean; + unhealthyRecovery?: boolean; /** Rebalance replays recorded deployed bytes instead of current promoted files. */ frozenRecipeId?: string; } diff --git a/src/lib/server/provisioning/launch-worker.ts b/src/lib/server/provisioning/launch-worker.ts index 3e0a11c..b2e4b58 100644 --- a/src/lib/server/provisioning/launch-worker.ts +++ b/src/lib/server/provisioning/launch-worker.ts @@ -44,7 +44,9 @@ parentPort!.on('message', (message: { id: number; kind: 'launch'; input: NativeL PLUTO_SERVER_KEY: input.serverKey }, files: recipe.files.map(file => ({ path: posix.join(relDestForKind(file.kind, recipe.game), file.path), contentSha256: file.sha256, sha256: file.sha256 })), - ...(input.rebuild ? { nextGeneration: instance.generation + 1, ...(input.safetyOverride ? { safetyOverride: true } : {}) } : {}) + ...(input.rebuild ? { nextGeneration: instance.generation + 1, + ...(input.safetyOverride ? { safetyOverride: true } : {}), + ...(input.unhealthyRecovery ? { unhealthyRecovery: true } : {}) } : {}) }; const encode = (value: unknown) => new TextEncoder().encode(JSON.stringify(value)); const blobs = [...new Map(recipe.files.map(file => [file.sha256, diff --git a/src/lib/server/provisioning/native.ts b/src/lib/server/provisioning/native.ts index 1d99212..4bfeb3d 100644 --- a/src/lib/server/provisioning/native.ts +++ b/src/lib/server/provisioning/native.ts @@ -2,7 +2,7 @@ import { assertScalingOwnership } from '../autoscaling/ownership.js'; import { removalAuthorization } from '../autoscaling/authorization.js'; import { scalingIntent } from '../autoscaling/admission.js'; import { putFileBlob, getFileBlob, blobSize } from '../db/file-blobs.js'; -import { concurrencyBlocker, preparingJobs, committedRebalanceStartup } from './concurrency.js'; +import { concurrencyBlocker, preparingJobs, committedRebalanceStartup, lifecycleHolders } from './concurrency.js'; import { isIP } from 'node:net'; import { setImmediate as yieldToController } from 'node:timers/promises'; import { choosePlacement, availableHostPort, type PlacementOptions } from '../hosts/capacity.js'; @@ -39,12 +39,15 @@ import { cancelCreation, finishCreationCancellation } from './cancel-creation.js 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'; 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; applied_at: string | null } + instance_id: string; generation: number; stage_operation_id: string | null; operation_id: string | null; applied_at: string | null; + recovery_step: 'restart' | 'rebuild' | null } interface FilePlan { id: number; path: string; sha256: string | null; checkBefore: boolean; expectedBeforeSha: string | null } +const unhealthyRecoveryCapability = 'instance.unhealthy-recovery'; /** Validate a launch request without selecting a host or consuming capacity. */ export function nativeLaunchSelection(input: Record, game: string): NativeSelection { @@ -242,7 +245,7 @@ function storedFiles(job: Job): FilePlan[] { /** Agent receipts are the authoritative before images; do not read a controller mount. */ function applyReceipt(job: Job, result: Record) { - if (job.applied_at) return; + if (job.applied_at || job.recovery_step) return; const plans = storedFiles(job); if (!plans.length) return; const before = result.before as { id?: number; path: string; contentBase64: string | null; contentSha256?: string }[] | undefined; @@ -276,7 +279,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.applied_at + const job = db().prepare(`SELECT d.*,n.instance_id,n.generation,n.operation_id,n.applied_at,n.recovery_step 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; @@ -309,6 +312,13 @@ export function reconcileNativeProvisioningReceipt(id: string): void { if (operation?.state !== 'completed') return; const instance = nativeInstance(job.server); if (!instance || instance.id !== job.instance_id) return; + if (job.state === 'cancelled' && job.action === 'create' + && db().prepare('SELECT 1 FROM native_creation_cancellations WHERE creation_id=?').get(id)) { + // Cleanup owns both generations of a cancelled recovery rebuild. Do not + // advance the registry underneath its already-bound removal command. + db().prepare('UPDATE native_provisioning SET applied_at=COALESCE(applied_at,?) WHERE deployment_id=?').run(operation.updatedAt, id); + return; + } if (job.action === 'destroy') { if (instance.generation !== job.generation) return; db().prepare('UPDATE native_instances SET retired_at=? WHERE id=?').run(operation.updatedAt, instance.id); @@ -318,7 +328,7 @@ export function reconcileNativeProvisioningReceipt(id: string): void { finishCreationCancellation(job.id); return; } - if (job.action === 'rebuild' && !hostMigration(id) && instance.generation === job.generation) { + 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. @@ -328,7 +338,7 @@ export function reconcileNativeProvisioningReceipt(id: string): void { 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 && !(job.action === 'rebuild' && instance.generation === job.generation + 1)) return; + } else if (instance.generation !== job.generation && !(operation.kind === 'instance.rebuild' && instance.generation === job.generation + 1)) return; applyReceipt(job, (operation.result ?? {}) as Record); db().prepare('UPDATE native_provisioning SET applied_at=COALESCE(applied_at,?) WHERE deployment_id=?').run(operation.updatedAt, id); } @@ -342,10 +352,78 @@ function complete(job: Job) { db().transaction(() => { db().prepare('UPDATE native_provisioning SET verified_at=? WHERE deployment_id=?').run(new Date().toISOString(), job.id); db().prepare('UPDATE restart_rollout SET healthy_at=?,error=NULL WHERE operation_id=?').run(new Date().toISOString(), job.id); + if (job.recovery_step) recordProgress(job.id, { recovery: { state: 'released', operationId: job.id, + message: 'Container and game health verified after provisioning recovery.' } }); finishDeployment(job.id, { ok: true }); })(); } +/** Continue the same launch with at most one restart and one replacement. */ +function queueUnhealthyRecovery(job: Job, instance: NativeInstance): void { + db().transaction(() => { + if (!stillActive(job)) return; + const live = nativeInstance(job.server); + if (live?.id !== instance.id || live.generation !== instance.generation) return; + if (job.recovery_step === 'rebuild') { + const message = 'Container is still unhealthy after recovery restart and rebuild. Inspect the server before retrying.'; + db().prepare('UPDATE restart_rollout SET error=? WHERE operation_id=?').run(message, job.id); + finishDeployment(job.id, { ok: false, message }); + return; + } + const host = getHost(instance.host_id); + if (!host?.connected || host.mode === 'disabled') { hold(job, 'Host is unavailable; waiting for reconnection.'); return; } + if (!host.inventory.capabilities?.includes(unhealthyRecoveryCapability)) { + hold(job, 'Update this host’s agent to recover unhealthy containers during provisioning.'); return; + } + const step = job.recovery_step === 'restart' ? 'rebuild' : 'restart'; + const settings = fleetSettings(); + const slots = db().prepare(`SELECT i.host_id FROM restart_rollout r LEFT JOIN native_instances i + ON i.server=r.server AND i.retired_at IS NULL WHERE r.healthy_at IS NULL AND (r.operation_id IS NULL OR r.operation_id!=?)`) + .all(job.id) as { host_id: string | null }[]; + if (settings.restartConcurrency > 0 && slots.length >= settings.restartConcurrency) { + hold(job, 'Waiting for restart recovery capacity.'); return; + } + if (settings.restartConcurrencyPerHost > 0 && slots.filter(slot => slot.host_id === instance.host_id).length >= settings.restartConcurrencyPerHost) { + hold(job, 'Waiting for restart recovery capacity on this host.'); return; + } + let payload: Record = { unhealthyRecovery: true }; + if (step === 'rebuild') { + const beta = activeBetaRemovalBlocker(job.server); + if (beta) { hold(job, beta); return; } + if (job.action !== 'rebuild' && settings.rebuildConcurrency > 0) { + if (lifecycleHolders().filter(holder => holder.id !== job.id && holder.action === 'rebuild').length >= settings.rebuildConcurrency) { + hold(job, 'Waiting for rebuild capacity.'); return; + } + } + assertReadyLibrary(instance.host_id, instance.library_id); + // Recover the files and credentials selected for this launch, even + // if new versions have been promoted while readiness was pending. + const launch = db().prepare(`SELECT COALESCE(p.payload,o.payload) AS payload FROM ec_host_operations o + JOIN native_instances i ON ${runtimeLaunchSql} + LEFT JOIN ec_host_operation_payloads p ON p.operation_id=o.id + WHERE i.id=? AND o.state='completed' AND o.history_expired_at IS NULL ORDER BY o.rowid DESC LIMIT 1`) + .get(instance.id) as { payload: string } | undefined; + if (!launch) { hold(job, 'Waiting for the original launch payload before recovery rebuild.'); return; } + payload = { ...JSON.parse(launch.payload), nextGeneration: instance.generation + 1, unhealthyRecovery: true }; + delete payload.safetyOverride; + } + const operation = queueHostOperationSerialized(instance.host_id, `instance.${step}`, Buffer.from(JSON.stringify(payload)), + { instanceId: instance.id, generation: instance.generation, idempotencyKey: `${job.id}:unhealthy:${step}` }, + { name: instance.server, unhealthyRecovery: true }); + const at = new Date().toISOString(); + db().prepare('UPDATE native_provisioning SET operation_id=?,generation=?,recovery_step=?,applied_at=NULL WHERE deployment_id=?') + .run(operation.id, instance.generation, step, job.id); + db().prepare("UPDATE server_deployments SET state='running',submitted_at=?,error=NULL,finished_at=NULL WHERE id=?").run(at, job.id); + 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, at, job.id); + const message = `Container unhealthy; ${step === 'restart' ? 'forcing a restart' : 'forcing a rebuild after restart recovery failed'}.`; + deploymentStage(job.id, message, { phase: 'container-action', blocker: null, checks: [], + command: { state: 'attempted', message: `Recovery ${step} queued as agent operation ${operation.id}.` }, + recovery: { state: 'held', operationId: job.id, message: 'Recovery runs within this provisioning operation.' }, + event: { kind: 'unhealthy-recovery', message, details: { step, operationId: operation.id, generation: instance.generation } } }); + })(); +} + export async function processNativeProvisioning(eligibility: (operation: { id: string; server: string; action: string; state: string; stage: string }) => string | null, options: { wait?: boolean } = {}): Promise { ensureHostSchema(db()); @@ -369,7 +447,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.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,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`).all() as Job[]; @@ -485,7 +563,8 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { job.instance_id = instance.id; job.generation = instance.generation; } const migration = hostMigration(job.id); - if (!instance || instance.id !== job.instance_id || (instance.generation !== job.generation && !(job.action === 'rebuild' && !migration && instance.generation === job.generation + 1))) { + const rebuilding = job.recovery_step === 'rebuild' || (!job.recovery_step && job.action === 'rebuild' && !migration); + if (!instance || instance.id !== job.instance_id || (instance.generation !== job.generation && !(rebuilding && instance.generation === job.generation + 1))) { finishDeployment(job.id, { ok: false, message: 'Original instance was retired or replaced; no command was sent.' }); return; } if (migration && !migration.removed_at) { @@ -499,7 +578,10 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { } if (job.operation_id) { const operation = getHostOperationStatus(job.operation_id); - if (operation?.kind === 'instance.cancel-create' && !['completed', 'failed'].includes(operation.state) && !host.inventory.capabilities?.includes('instance.cancel-create')) { + const recoveryCleanup = operation?.kind === 'instance.cancel-create' && !!db().prepare(`SELECT 1 FROM ec_host_operations + WHERE id=? AND json_extract(payload,'$.recoveryRebuildOperationId') IS NOT NULL`).get(operation.id); + if (operation?.kind === 'instance.cancel-create' && !['completed', 'failed'].includes(operation.state) + && (!host.inventory.capabilities?.includes('instance.cancel-create') || (recoveryCleanup && !host.inventory.capabilities?.includes(unhealthyRecoveryCapability)))) { hold(job, hostAgentUpdatesEnabled(host.inventory) ? 'Waiting for the host agent update. New command delivery is paused so its current batch can finish and the automatic updater can run. Cleanup resumes after the updated agent connects.' : 'Update the host agent to support creation cancellation: rerun the current installer on this host. Automatic updates are not enabled. New command delivery is paused; cleanup and port release await the updated agent.'); return; @@ -514,7 +596,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { })(); return; } - if (migration) { + if (migration && !job.recovery_step) { if (Date.now() - Date.parse(operation.updatedAt) >= 60_000) { db().transaction(() => { db().prepare('DELETE FROM restart_rollout WHERE operation_id=?').run(job.id); @@ -552,15 +634,17 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { })(); return; } - if (job.action === 'rebuild' && !migration && instance.generation === job.generation) { + 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); + db().prepare('UPDATE native_provisioning SET applied_at=COALESCE(applied_at,?) WHERE deployment_id=?').run(operation.updatedAt, job.id); 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.` }, - checkpoint: { key: 'container-action', label: `Agent confirmed ${job.action}`, at: operation.updatedAt } }); + const action = job.recovery_step ? `recovery ${job.recovery_step}` : job.action; + recordProgress(job.id, { command: { state: 'accepted', message: `Agent confirmed ${action}; verifying the resulting instance.` }, + checkpoint: { key: job.recovery_step ? `recovery-${job.recovery_step}` : 'container-action', label: `Agent confirmed ${action}`, at: operation.updatedAt } }); } const checks: CheckInput[] = []; const check = (value: CheckInput, phase: import('../../provisioning-detail.js').ProvisioningPhase) => { @@ -580,7 +664,11 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { 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; } + if (!health.healthy) { + if (health.state === 'unhealthy' && (job.action === 'create' || job.action === 'rebuild')) queueUnhealthyRecovery(job, instance); + else 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; @@ -669,6 +757,12 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { 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 false; + if (job.action === 'rebuild' && health.state === 'unhealthy') { + if (!getHost(instance.host_id)?.inventory.capabilities?.includes(unhealthyRecoveryCapability)) { + hold(job, 'Update this host’s agent to recover unhealthy containers during provisioning.'); return false; + } + return true; + } if (!['stopped','exited','created','dead','missing'].includes(health.state)) { const occupancy = await probePlayerEvidence(job.server); if (!stillActive(job)) return false; @@ -708,7 +802,8 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { preparedPassword = config.rcon.password || randomBytes(32).toString('hex'); prepared = await prepareNativeLaunch({ game: prior.game, mapSlug: prior.mapSlug, native: prior.native, instance, serverKey: preparedKey, rconPassword: preparedPassword, - rebuild: job.action === 'rebuild' && !migration, safetyOverride: overrideSafety, frozenRecipeId }); + rebuild: job.action === 'rebuild' && !migration, safetyOverride: overrideSafety, + unhealthyRecovery: host.inventory.capabilities?.includes(unhealthyRecoveryCapability), frozenRecipeId }); if (job.action === 'rebuild' && !migration && !await checkOccupancy()) return; } const commit = db().transaction(() => { diff --git a/src/lib/server/retention.ts b/src/lib/server/retention.ts index 6f71898..12c6773 100644 --- a/src/lib/server/retention.ts +++ b/src/lib/server/retention.ts @@ -22,12 +22,26 @@ const unsettledDeployment = `d.state NOT IN ${terminal} (a.state NOT IN ('completed','failed') 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 (d.action='rebuild' AND i.retired_at IS NULL AND i.generation=n.generation - AND NOT EXISTS (SELECT 1 FROM ec_host_migrations m WHERE m.deployment_id=d.id)))))) + OR (a.kind='instance.rebuild' AND i.retired_at IS NULL AND i.generation=n.generation))))) OR EXISTS (SELECT 1 FROM ec_host_migrations m JOIN ec_host_operations a ON a.id=m.remove_operation_id WHERE m.deployment_id=d.id AND (a.state NOT IN ('completed','failed') OR (a.state='completed' AND m.removed_at IS NULL)))`; +// Recovery replaces operation_id with its latest command. The original launch +// still supplies rebuild files and the ownership label used by creation cleanup. +const recoveryLaunch = `EXISTS (SELECT 1 FROM native_provisioning n + JOIN server_deployments d ON d.id=n.deployment_id + JOIN native_instances i ON i.id=n.instance_id AND i.retired_at IS NULL + WHERE n.recovery_step IS NOT NULL AND d.server_deleted_at IS NULL AND o.instance_id=i.id AND o.host_id=i.host_id + AND ((o.kind='instance.create' AND o.generation=n.generation) + OR (o.kind='instance.rebuild' AND o.generation+1=n.generation)) + AND ((${unsettledDeployment}) OR (d.action='create' AND d.state IN ('failed','cancelled') + AND i.generation BETWEEN n.generation AND n.generation+1 + AND NOT EXISTS (SELECT 1 FROM server_deployments newer WHERE newer.server=d.server + AND newer.server_deleted_at IS NULL AND newer.rowid>d.rowid AND newer.state!='cancelled' + AND NOT (newer.action='destroy' AND newer.state='waiting' + AND NOT EXISTS (SELECT 1 FROM native_provisioning p WHERE p.deployment_id=newer.id AND p.operation_id IS NOT NULL))))))`; const unreferencedOperation = `NOT EXISTS (SELECT 1 FROM native_provisioning n WHERE o.id IN (n.operation_id,n.stage_operation_id)) AND NOT EXISTS (SELECT 1 FROM native_instances i WHERE i.retired_at IS NULL AND ${runtimeLaunchSql}) + AND NOT (${recoveryLaunch}) AND NOT EXISTS (SELECT 1 FROM ec_host_migrations m WHERE m.remove_operation_id=o.id) AND NOT EXISTS (SELECT 1 FROM ec_host_decommissions d WHERE d.operation_id=o.id) AND o.kind NOT IN ('runtime.pull','library.download','library.import','library.remove','host.decommission','host.rotate-credential') @@ -97,6 +111,7 @@ function policies(): Policy[] { failed: "o.state='failed'", eligible: `o.state IN ('completed','failed') AND o.kind IN ('instance.create','instance.rebuild','instance.restart','instance.start','files.stage','files.read') AND NOT EXISTS (SELECT 1 FROM retention_controls c WHERE c.category='operation' AND c.record_id=o.id AND c.protected=1) + AND NOT (${recoveryLaunch}) AND NOT EXISTS (SELECT 1 FROM native_provisioning n JOIN server_deployments d ON d.id=n.deployment_id WHERE o.id IN (n.operation_id,n.stage_operation_id) AND ((${unsettledDeployment}) OR EXISTS ( SELECT 1 FROM retention_controls c WHERE c.category IN ('provisioning','workflow-payload') AND c.record_id=d.id AND c.protected=1))) diff --git a/static/agent/ec-agent.py b/static/agent/ec-agent.py index 01cb0df..e50a31d 100644 --- a/static/agent/ec-agent.py +++ b/static/agent/ec-agent.py @@ -33,7 +33,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.8.0" +VERSION = "1.8.1" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -1394,14 +1394,18 @@ class Agent: if line.startswith("PRETTY_NAME="): os_name = line.split("=", 1)[1].strip('"') update = self.update_inventory() - capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else []) + capabilities = ["library.remove", "plutonium.cache", "instance.cancel-create", "instance.safety-override", "instance.unhealthy-recovery", "instance.autoscaling-remove", "operations.read-lane", "operations.health-lane", "rcon.status-batch", "instance.heartbeat-state", "files.manifest", "files.blobs"] + (["host.decommission"] if self.can_uninstall() else []) + (["agent.update"] if update["enabled"] else []) plutonium_cache = self.plutonium_cache_inventory() return {"hostname": socket.gethostname(), "agentVersion": VERSION, "protocolVersion": 1, "capabilities": capabilities, "agentUpdate": update, "agentLanes": self.lane_inventory(), "os": os_name, "osVersion": platform.release(), "architecture": platform.machine(), "cpu": platform.processor() or platform.machine(), "cpuCount": os.cpu_count() or 0, **resources, **({"cpuUsagePercent": cpu_usage} if cpu_usage is not None else {}), "memoryTotal": memory.get("MemTotal", 0), "memoryAvailable": memory.get("MemAvailable", 0), "storageTotal": usage.total, "storageAvailable": usage.free, "dockerVersion": docker_version, "dockerRunning": docker_running, "directoriesWritable": os.access(self.root, os.W_OK), "runtimeImages": sorted(set(images)), **({"plutoniumCache": plutonium_cache} if plutonium_cache else {}), "libraries": libraries, "libraryCaches": caches, "currentOperation": self.current_operation, "portRange": self.config.get("portRange", {"min": 28960, "max": 29060})} - def safe_to_disrupt(self, instance_id, generation): + def safe_to_disrupt(self, instance_id, generation, *, allow_unhealthy=False): health = self.inspect(instance_id, generation) if health["state"] in ("missing", "stopped", "exited", "created", "dead"): return + # Provisioning recovery may treat explicitly unhealthy containers as + # empty. Reinspect here so recovered containers still require occupancy. + if allow_unhealthy and health["state"] == "unhealthy": + return if not health["running"]: raise Held("Container state is unknown; disruption held") result = self.rcon(instance_id, generation, "status") @@ -1857,6 +1861,9 @@ class Agent: raise AutoscalingInvalidated("Autoscaling removal authorization expired before mutation") def lifecycle(self, kind, instance_id, generation, payload): + unhealthy_recovery = payload.get("unhealthyRecovery") is True + if unhealthy_recovery and (kind not in ("instance.restart", "instance.rebuild") or payload.get("stageId") or "autoscaling" in payload): + raise AgentError("Unhealthy recovery only permits restart or rebuild without file activation or autoscaling") if kind == "instance.cancel-create": return self.cancel_creation(instance_id, generation, payload) if kind == "instance.create": @@ -1891,10 +1898,10 @@ class Agent: next_generation = payload.get("nextGeneration", payload.get("newGeneration")) if isinstance(next_generation, bool) or not isinstance(next_generation, int) or next_generation <= generation: raise AgentError("Rebuild must use a newer immutable generation") - spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration")} + spec = payload.get("spec") or {key: value for key, value in payload.items() if key not in ("nextGeneration", "newGeneration", "unhealthyRecovery")} self.validate_spec(spec) if not override_safety: - self.safe_to_disrupt(instance_id, generation) + self.safe_to_disrupt(instance_id, generation, allow_unhealthy=unhealthy_recovery) if "autoscaling" in payload: self.authorize_autoscaling_removal(instance_id, generation, payload) self.checkpoint({"phase": "disrupting", "containerId": health.get("containerId"), "startedAt": health.get("startedAt")}) @@ -1928,29 +1935,46 @@ class Agent: Repeating this cleanup after an ambiguous Docker result is safe. """ create_id = identifier(payload.get("createOperationId")) + targets = [(generation, create_id)] + if "recoveryRebuildOperationId" in payload or "recoveryGeneration" in payload: + rebuild_id = identifier(payload.get("recoveryRebuildOperationId")) + recovery_generation = payload.get("recoveryGeneration") + if isinstance(recovery_generation, bool) or not isinstance(recovery_generation, int) or recovery_generation != generation + 1: + raise Rejected("Cancellation recovery must target the next generation") + saved = self.journal.db.execute("SELECT request FROM operations WHERE id=?", (rebuild_id,)).fetchone() + if saved: + rebuild = json.loads(saved["request"]) + rebuild_payload = rebuild.get("payload", {}) + if rebuild["kind"] != "instance.rebuild" or rebuild["instanceId"] != instance_id or rebuild["generation"] != generation or rebuild_payload.get("nextGeneration", rebuild_payload.get("newGeneration")) != recovery_generation or rebuild_payload.get("unhealthyRecovery") is not True: + raise Rejected("Cancellation does not match the creation recovery rebuild") + targets.append((recovery_generation, rebuild_id)) marker = self.instance_dir(instance_id, generation).parent / "current.json" - if marker.exists() and json.loads(marker.read_text())["generation"] > generation: + if marker.exists() and json.loads(marker.read_text())["generation"] > targets[-1][0]: raise Rejected("Cancelled creation was superseded by a newer generation") saved = self.journal.db.execute("SELECT request FROM operations WHERE id=?", (create_id,)).fetchone() if saved: original = json.loads(saved["request"]) if original["kind"] != "instance.create" or original["instanceId"] != instance_id or original["generation"] != generation: raise Rejected("Cancellation does not match the original creation") - health = self.inspect(instance_id, generation) - if health["exists"] and health.get("creationOperation") != create_id: - raise Rejected("Container belongs to a different creation operation") + health_by_generation = {} + for target_generation, operation_id in targets: + health = self.inspect(instance_id, target_generation) + if health["exists"] and health.get("creationOperation") != operation_id: + raise Rejected("Container belongs to a different creation operation") + health_by_generation[target_generation] = health self.checkpoint({"phase": "cancelling-creation", "createOperationId": create_id}) - # Fence before deleting: a crash must never permit the old creation to run again. - self.mark_removed(instance_id, generation) - if health["exists"]: - self.docker.call("container", "rm", "--force", "--volumes", self.container_name(instance_id, generation)) - after = self.inspect(instance_id, generation) - if after["exists"]: - raise AgentError("Cancelled creation container still exists") - directory = self.instance_dir(instance_id, generation) - if directory.exists(): - shutil.rmtree(directory) - return {**after, "cancelledCreation": create_id} + # Fence both generations before deleting: delayed create/rebuild work + # must never recreate either target after cancellation or a crash. + self.mark_removed(instance_id, targets[-1][0]) + for target_generation, _ in targets: + if health_by_generation[target_generation]["exists"]: + self.docker.call("container", "rm", "--force", "--volumes", self.container_name(instance_id, target_generation)) + if self.inspect(instance_id, target_generation)["exists"]: + raise AgentError("Cancelled creation container still exists") + directory = self.instance_dir(instance_id, target_generation) + if directory.exists(): + shutil.rmtree(directory) + return {**self.inspect(instance_id, generation), "cancelledCreation": create_id} def stage(self, instance_id, generation, payload, operation_id): directory = self.instance_dir(instance_id, generation) @@ -2453,7 +2477,7 @@ class Agent: current_marker = self.instance_dir(instance_id, generation).parent / "current.json" if current_marker.exists(): current = json.loads(current_marker.read_text()) - if current["generation"] > generation or current["generation"] == generation and current.get("removed") and kind not in ("instance.inspect", "instance.remove", "instance.cancel-create"): + if kind != "instance.cancel-create" and (current["generation"] > generation or current["generation"] == generation and current.get("removed") and kind not in ("instance.inspect", "instance.remove")): raise AgentError("Instance generation has been superseded or removed") if kind.startswith("instance."): return self.lifecycle(kind, instance_id, generation, payload) diff --git a/static/agent/manifest.json b/static/agent/manifest.json index 176a290..7d384d8 100644 --- a/static/agent/manifest.json +++ b/static/agent/manifest.json @@ -1,8 +1,8 @@ { - "version": "1.8.0", + "version": "1.8.1", "protocolVersion": 1, "platform": "debian-amd64", - "sha256": "eb36afb3113b92304da42fc78ac7f3b8218456f33c5bc1c0f57b018bbc2ad65d", + "sha256": "09c722864c73ddb76404ea994d9a194892a2226d8d9006de8dcac9a6e2591760", "updaterProtocolVersion": 1, "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" } diff --git a/tests/unit/concurrency.test.ts b/tests/unit/concurrency.test.ts index 1596a88..7c910e0 100644 --- a/tests/unit/concurrency.test.ts +++ b/tests/unit/concurrency.test.ts @@ -44,6 +44,20 @@ it('enforces per-host removal and recovery limits independently of other hosts', expect(concurrencyBlocker('b','restart','other')).toBeNull(); }); +it('counts a creation recovery rebuild against rebuild capacity without losing its startup reservation', async () => { + const { db } = await import('../../src/lib/server/db/index.js'); + const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js'); + const { concurrencyBlocker, lifecycleHolders } = await import('../../src/lib/server/provisioning/concurrency.js'); + db().prepare("INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at) VALUES('i','t5_test','host','lib','img',1,'2026-01-01')").run(); + db().prepare("INSERT INTO server_deployments(id,server,action,state,stage,started_at,triggered_by) VALUES('a','t5_test','create','running','recovery','2026-01-01','test')").run(); + db().prepare("INSERT INTO native_provisioning(deployment_id,instance_id,generation,recovery_step) VALUES('a','i',1,'rebuild')").run(); + saveFleetSettings({ ...fleetSettings(), rebuildConcurrency: 1, createConcurrencyPerHost: 1 }); + expect(lifecycleHolders()).toEqual([expect.objectContaining({ id: 'a', action: 'rebuild' })]); + expect(concurrencyBlocker('b', 'rebuild', 'other')).toContain('rebuild'); + expect(concurrencyBlocker('b', 'create', 'host')).toContain('startup'); + expect(concurrencyBlocker('a', 'rebuild', 'host')).toBeNull(); +}); + it('retains every capacity owner across failure, cancellation and migration without counting completed history', async () => { const { db } = await import('../../src/lib/server/db/index.js'); const { createHost } = await import('../../src/lib/server/hosts/index.js'); @@ -99,7 +113,7 @@ it('seeks capacity owners through active indexes instead of scanning retained jo let sql: string; try { expect(lifecycleHolders()).toEqual([]); - sql = prepare.mock.calls.find(([sql])=>sql.startsWith('SELECT d.id,d.action'))![0]; + sql = prepare.mock.calls.find(([sql])=>sql.startsWith('SELECT d.id,'))![0]; } finally { prepare.mockRestore(); } const plan = h.prepare(`EXPLAIN QUERY PLAN ${sql}`).all('[]') as {detail:string}[]; const details = plan.map(row=>row.detail).join('\n'); diff --git a/tests/unit/native-provisioning.test.ts b/tests/unit/native-provisioning.test.ts index d7ea220..b47e0b1 100644 --- a/tests/unit/native-provisioning.test.ts +++ b/tests/unit/native-provisioning.test.ts @@ -18,7 +18,7 @@ vi.mock('../../src/lib/server/rebalancing/admission.js', () => ({ let queuedKeys = 0; let fixture: Fixture; let hostId: string; let libraryId: string; const image = 'example.test/plutainer:test'; -const inventory = () => ({ cpuCount: 16, cpuUsagePercent: 10, memoryTotal: 64 * 1024 ** 3, memoryAvailable: 60 * 1024 ** 3, protocolVersion: 1, dockerRunning: true, directoriesWritable: true, storageAvailable: 100_000_000_000, runtimeImages: [image], libraries: libraryId ? [{ id: libraryId, state: 'ready' as const }] : [] }); +const inventory = () => ({ cpuCount: 16, cpuUsagePercent: 10, memoryTotal: 64 * 1024 ** 3, memoryAvailable: 60 * 1024 ** 3, protocolVersion: 1, capabilities: ['instance.unhealthy-recovery'], dockerRunning: true, directoriesWritable: true, storageAvailable: 100_000_000_000, runtimeImages: [image], libraries: libraryId ? [{ id: libraryId, state: 'ready' as const }] : [] }); beforeEach(async () => { vi.resetModules(); vi.clearAllMocks(); remote.state = 'running'; remote.players = 0; remote.map = 'zombie_theater'; libraryId = ''; queuedKeys = 0; fixture = await newFixture([]); @@ -1503,6 +1503,201 @@ describe('automatic capacity admission and placement', () => { }); }); +describe('unhealthy recovery within native create and rebuild operations', () => { + async function launch(action: 'create' | 'rebuild') { + const job = await queue(); await advance(); + if (action === 'create') return job; + await acknowledge(job.deploymentId); + const { queueNativeRebuild } = await import('../../src/lib/server/provisioning/native.js'); + const rebuild = queueNativeRebuild(job.name, 'test'); + if (!rebuild.ok) throw new Error(rebuild.message); + await advance(); + return rebuild; + } + + it.each(['create', 'rebuild'] as const)('recovers an unhealthy %s with one restart then a rebuild using the original launch payload', async action => { + const { config } = await import('../../src/lib/server/config.js'); + Object.assign(config.rcon, { password: '' }); + const job = await launch(action); + const original = (await operationFor(job.deploymentId))!; + const currentGeneration = action === 'rebuild' ? 2 : 1; + const { nativeInstance } = await import('../../src/lib/server/backends/native.js'); + const { db, closeDb } = await import('../../src/lib/server/db/index.js'); + const { rconStatusResult } = await import('../../src/lib/server/integrations/rcon.js'); + vi.mocked(rconStatusResult).mockClear(); + remote.state = 'unhealthy'; remote.players = 4; + Object.assign(config.rcon, { password: 'changed-after-original-launch' }); + await acknowledge(job.deploymentId); + const restart = (await operationFor(job.deploymentId))!; + const { getDeployment } = await import('../../src/lib/server/provisioning/history.js'); + expect(restart).toMatchObject({ kind: 'instance.restart', state: 'pending', generation: currentGeneration, + instanceId: original.instanceId, payload: { unhealthyRecovery: true } }); + expect(getDeployment(job.deploymentId)!.capacity.pools).toEqual(expect.arrayContaining([ + expect.objectContaining({ kind: 'startup', ownsSlot: true }), + expect.objectContaining({ kind: 'restart', ownsSlot: true }) + ])); + expect(restart.id).not.toBe(original.id); + expect(nativeInstance(job.name)?.generation).toBe(currentGeneration); + expect(db().prepare('SELECT recovery_step,generation FROM native_provisioning WHERE deployment_id=?').get(job.deploymentId)) + .toEqual({ recovery_step: 'restart', generation: currentGeneration }); + expect(rconStatusResult).not.toHaveBeenCalled(); + + // Restarting the controller must preserve the current recovery command. + closeDb(); await advance(); + expect((await operationFor(job.deploymentId))!.id).toBe(restart.id); + await acknowledge(job.deploymentId); + const rebuild = (await operationFor(job.deploymentId))!; + expect(rebuild).toMatchObject({ kind: 'instance.rebuild', state: 'pending', generation: currentGeneration, + instanceId: original.instanceId, payload: { unhealthyRecovery: true, nextGeneration: currentGeneration + 1, + files: original.payload.files, env: original.payload.env, rconPassword: original.payload.rconPassword, + libraryId: original.payload.libraryId, image: original.payload.image } }); + expect(rebuild.id).not.toBe(restart.id); + expect(nativeInstance(job.name)?.generation).toBe(currentGeneration); + expect(rconStatusResult).not.toHaveBeenCalled(); + expect(db().prepare('SELECT recovery_step,generation FROM native_provisioning WHERE deployment_id=?').get(job.deploymentId)) + .toEqual({ recovery_step: 'rebuild', generation: currentGeneration }); + remote.state = 'running'; remote.players = 0; + await acknowledge(job.deploymentId); + expect(nativeInstance(job.name)?.generation).toBe(currentGeneration + 1); + expect(await detail(job.deploymentId)).toMatchObject({ action, state: 'done' }); + expect(getDeployment(job.deploymentId)!.progress?.recovery?.state).toBe('released'); + expect(getDeployment(job.deploymentId)!.capacity.pools.every(pool => !pool.ownsSlot)).toBe(true); + await advance(); + expect((await operationFor(job.deploymentId))!.id).toBe(rebuild.id); + expect(nativeInstance(job.name)?.generation).toBe(currentGeneration + 1); + }); + + it.each(['create', 'rebuild'] as const)('finishes %s after a successful recovery restart without another rebuild', async action => { + const job = await launch(action); + remote.state = 'unhealthy'; await acknowledge(job.deploymentId); + const restart = (await operationFor(job.deploymentId))!; + expect(restart.kind).toBe('instance.restart'); + remote.state = 'running'; await acknowledge(job.deploymentId); await advance(); + expect(await detail(job.deploymentId)).toMatchObject({ action, state: 'done' }); + expect((await operationFor(job.deploymentId))!.id).toBe(restart.id); + const { nativeInstance } = await import('../../src/lib/server/backends/native.js'); + expect(nativeInstance(job.name)?.generation).toBe(action === 'rebuild' ? 2 : 1); + const { db } = await import('../../src/lib/server/db/index.js'); + expect(db().prepare("SELECT COUNT(*) AS count FROM ec_host_operations WHERE kind='instance.rebuild'").get()) + .toEqual({ count: action === 'rebuild' ? 1 : 0 }); + }); + + it.each(['restart', 'rebuild'] as const)('retains an unknown recovery %s receipt without replaying or advancing recovery', async step => { + const job = await launch('create'); + remote.state = 'unhealthy'; await acknowledge(job.deploymentId); + if (step === 'rebuild') await acknowledge(job.deploymentId); + const operation = (await operationFor(job.deploymentId))!; + expect(operation.kind).toBe(`instance.${step}`); + const { reportHostOperation } = await import('../../src/lib/server/hosts/index.js'); + reportHostOperation(hostId, operation.id, { state: 'unknown', error: 'Reply lost during recovery' }); + const { closeDb, db } = await import('../../src/lib/server/db/index.js'); closeDb(); + const { interruptDeployments } = await import('../../src/lib/server/provisioning/history.js'); interruptDeployments(); + await advance(); await advance(); + expect((await operationFor(job.deploymentId))!.id).toBe(operation.id); + expect(db().prepare("SELECT COUNT(*) AS count FROM ec_host_operations WHERE kind IN ('instance.create','instance.restart','instance.rebuild')").get()) + .toEqual({ count: step === 'restart' ? 2 : 3 }); + expect((await detail(job.deploymentId)).state).not.toBe('failed'); + remote.state = 'running'; await acknowledge(job.deploymentId); + expect((await detail(job.deploymentId)).state).toBe('done'); + }); + + it.each(['starting', 'restarting', 'unknown'])('waits for %s container readiness without forcing recovery', async state => { + const job = await launch('create'); + const original = (await operationFor(job.deploymentId))!; + remote.state = state; await acknowledge(job.deploymentId); await advance(); + expect((await operationFor(job.deploymentId))!.id).toBe(original.id); + expect(await detail(job.deploymentId)).toMatchObject({ state: 'awaiting-health', stage: `Waiting for container readiness: ${state}` }); + const { db } = await import('../../src/lib/server/db/index.js'); + expect(db().prepare('SELECT recovery_step FROM native_provisioning WHERE deployment_id=?').get(job.deploymentId)) + .toEqual({ recovery_step: null }); + }); + + it.each(['create', 'rebuild'] as const)('does not submit recovery if %s is cancelled during its health probe', async action => { + const job = await launch(action); + const original = (await operationFor(job.deploymentId))!; + const { reportHostOperation } = await import('../../src/lib/server/hosts/index.js'); + reportHostOperation(hostId, original.id, { state: 'completed' }); + const { nativeRequest } = await import('../../src/lib/server/backends/native.js'); + const { cancelProvisioningJob } = await import('../../src/lib/server/provisioning/cancellation.js'); + let release!: (value: never) => void; + vi.mocked(nativeRequest).mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + const pending = advance(); + await vi.waitFor(() => expect(release).toBeTypeOf('function')); + expect(cancelProvisioningJob(job.deploymentId)).toBe(true); + release({ healthy: false, state: 'unhealthy' } as never); + await pending; await advance(); + expect((await detail(job.deploymentId)).state).toBe('cancelled'); + expect((await operationFor(job.deploymentId))!.id).toBe(original.id); + const { db } = await import('../../src/lib/server/db/index.js'); + expect(db().prepare("SELECT id FROM ec_host_operations WHERE kind='instance.restart'").all()).toEqual([]); + }); + + it.each(['create', 'rebuild'] as const)('reconciles a delivered recovery rebuild after its %s flow is cancelled', async action => { + const job = await launch(action); + remote.state = 'unhealthy'; await acknowledge(job.deploymentId); await acknowledge(job.deploymentId); + const rebuild = (await operationFor(job.deploymentId))!; + expect(rebuild.kind).toBe('instance.rebuild'); + const { pendingHostOperations, reportHostOperation } = await import('../../src/lib/server/hosts/index.js'); + const { cancelProvisioningJob } = await import('../../src/lib/server/provisioning/cancellation.js'); + pendingHostOperations(hostId); + expect(cancelProvisioningJob(job.deploymentId)).toBe(true); + reportHostOperation(hostId, rebuild.id, { state: 'completed' }); + await advance(); await advance(); + const { nativeInstance } = await import('../../src/lib/server/backends/native.js'); + expect(nativeInstance(job.name)?.generation).toBe(action === 'rebuild' ? 3 : 2); + expect((await detail(job.deploymentId)).state).toBe('cancelled'); + expect((await operationFor(job.deploymentId))!.id).toBe(rebuild.id); + }); + + it('fails clearly after an unhealthy recovery rebuild instead of looping', async () => { + const job = await launch('create'); + remote.state = 'unhealthy'; + await acknowledge(job.deploymentId); + await acknowledge(job.deploymentId); + const rebuild = (await operationFor(job.deploymentId))!; + expect(rebuild.kind).toBe('instance.rebuild'); + await acknowledge(job.deploymentId); await advance(); + expect(await detail(job.deploymentId)).toMatchObject({ state: 'failed', error: expect.stringMatching(/unhealthy.*restart.*rebuild/i) }); + expect((await operationFor(job.deploymentId))!.id).toBe(rebuild.id); + const { db } = await import('../../src/lib/server/db/index.js'); + expect(db().prepare("SELECT COUNT(*) AS count FROM ec_host_operations WHERE kind IN ('instance.create','instance.restart','instance.rebuild')").get()) + .toEqual({ count: 3 }); + }); + + it.each(['create', 'rebuild'] as const)('holds unhealthy %s verification until the agent supports recovery', async action => { + const job = await launch(action); + const original = (await operationFor(job.deploymentId))!; + const { heartbeat } = await import('../../src/lib/server/hosts/index.js'); + heartbeat(hostId, { ...inventory(), capabilities: [] }); + remote.state = 'unhealthy'; await acknowledge(job.deploymentId); await advance(); + expect((await operationFor(job.deploymentId))!.id).toBe(original.id); + expect(await detail(job.deploymentId)).toMatchObject({ state: 'awaiting-health', stage: expect.stringMatching(/agent/i) }); + heartbeat(hostId, inventory()); await advance(); + expect(await operationFor(job.deploymentId)).toMatchObject({ kind: 'instance.restart', payload: { unhealthyRecovery: true } }); + }); + + it.each([false, true])('rebuilds an already unhealthy container without requiring RCON emptiness once supported (supported=%s)', async supported => { + const created = await launch('create'); await acknowledge(created.deploymentId); + const { heartbeat } = await import('../../src/lib/server/hosts/index.js'); + if (!supported) heartbeat(hostId, { ...inventory(), capabilities: [] }); + const { queueNativeRebuild } = await import('../../src/lib/server/provisioning/native.js'); + const rebuild = queueNativeRebuild(created.name, 'test'); + if (!rebuild.ok) throw new Error(rebuild.message); + const { rconStatusResult } = await import('../../src/lib/server/integrations/rcon.js'); + vi.mocked(rconStatusResult).mockClear(); + remote.state = 'unhealthy'; remote.players = 3; + await advance(); + if (!supported) { + expect(await operationFor(rebuild.deploymentId)).toBeNull(); + expect((await detail(rebuild.deploymentId)).stage).toMatch(/agent/i); + heartbeat(hostId, inventory()); await advance(); + } + expect(await operationFor(rebuild.deploymentId)).toMatchObject({ kind: 'instance.rebuild', generation: 1, + payload: { unhealthyRecovery: true, nextGeneration: 2 } }); + expect(rconStatusResult).not.toHaveBeenCalled(); + }); +}); + describe('native provisioning cancellation and rollback', () => { it('cancels an unhealthy restart and prevents a late health response from reviving it', async () => { const job = await queue(); await advance(); await acknowledge(job.deploymentId); diff --git a/tests/unit/native-recovery-cancellation.test.ts b/tests/unit/native-recovery-cancellation.test.ts new file mode 100644 index 0000000..a2f6d2f --- /dev/null +++ b/tests/unit/native-recovery-cancellation.test.ts @@ -0,0 +1,115 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { newFixture, type Fixture } from '../helpers/fixture.js'; +import { prepareNativeHost } from '../helpers/native-host.js'; + +vi.mock('../../src/lib/server/pipeline/worker.js', () => ({ kick: vi.fn(async () => {}), kickProvisioning: vi.fn(async () => {}) })); + +let fixture: Fixture; +const server = 't5_zombie_theater_1'; +beforeEach(async () => { + vi.resetModules(); + fixture = await newFixture([]); +}); +afterEach(async () => { await fixture.cleanup(); }); + +async function recovering(step: 'restart' | 'rebuild') { + const { host } = await prepareNativeHost([server]); + const hosts = await import('../../src/lib/server/hosts/index.js'); + const { db } = await import('../../src/lib/server/db/index.js'); + const { nativeInstance } = await import('../../src/lib/server/backends/native.js'); + const { beginDeployment, getDeployment } = await import('../../src/lib/server/provisioning/history.js'); + const { cancelCreation, canCancelCreation } = await import('../../src/lib/server/provisioning/cancel-creation.js'); + const { processNativeProvisioning, reconcileNativeProvisioningReceipt } = await import('../../src/lib/server/provisioning/native.js'); + hosts.heartbeat(host.id, { ...hosts.getHost(host.id)!.inventory, capabilities: ['instance.cancel-create', 'instance.unhealthy-recovery'] }); + db().prepare('INSERT INTO servers(name) VALUES(?)').run(server); + const instance = nativeInstance(server)!; + hosts.reserveHostPort(host.id, instance.id, { port: instance.port, protocol: 'udp' }); + hosts.reserveHostPort(host.id, instance.id, { port: instance.port, protocol: 'tcp' }); + const deploymentId = beginDeployment(server, 'create'); + const original = hosts.queueHostOperation(host.id, 'instance.create', {}, { instanceId: instance.id, generation: 1 }); + hosts.reportHostOperation(host.id, original.id, { state: 'completed', result: { state: 'unhealthy', healthy: false } }); + const recovery = hosts.queueHostOperation(host.id, `instance.${step}`, { unhealthyRecovery: true, ...(step === 'rebuild' ? { nextGeneration: 2 } : {}) }, + { instanceId: instance.id, generation: 1 }); + db().prepare('UPDATE native_provisioning SET operation_id=?,recovery_step=? WHERE deployment_id=?').run(recovery.id, step, deploymentId); + db().prepare("UPDATE server_deployments SET state='awaiting-health' WHERE id=?").run(deploymentId); + const commandFor = (id: string) => { + const row = db().prepare('SELECT operation_id FROM native_provisioning WHERE deployment_id=?').get(id) as { operation_id: string }; + return hosts.getHostOperation(row.operation_id)!; + }; + return { host, instance, deploymentId, original, recovery, ...hosts, db, nativeInstance, cancelCreation, canCancelCreation, + getDeployment, reconcileNativeProvisioningReceipt, commandFor, advance: () => processNativeProvisioning(() => null) }; +} + +describe('cancelling creation during unhealthy recovery', () => { + it('keeps the original creation identity and suppresses recovery restart replay', async () => { + const api = await recovering('restart'); + expect(api.pendingHostOperations(api.host.id).map(operation => operation.id)).toContain(api.recovery.id); + const cleanup = api.cancelCreation(api.deploymentId)!; + const removal = api.commandFor(cleanup); + expect(removal).toMatchObject({ kind: 'instance.cancel-create', generation: 1, payload: { createOperationId: api.original.id } }); + expect(removal.payload).toEqual({ createOperationId: api.original.id }); + expect(api.pendingCreationCancellations(api.host.id)).toEqual([api.original.id]); + expect(api.pendingHostOperations(api.host.id).filter(operation => operation.kind.startsWith('instance.')).map(operation => operation.id)).toEqual([removal.id]); + api.reportHostOperation(api.host.id, removal.id, { state: 'completed', result: { exists: false } }); + await api.advance(); + expect(api.getDeployment(cleanup)?.deployment.state).toBe('done'); + expect(api.getHostOperation(api.recovery.id)?.state).toBe('failed'); + expect(api.db().prepare('SELECT * FROM ec_host_ports').all()).toEqual([]); + }); + + it.each(['pending', 'running', 'unknown', 'completed'] as const)('cleans up both rebuild generations when recovery is %s', async state => { + const api = await recovering('rebuild'); + if (state !== 'pending') { + api.pendingHostOperations(api.host.id); + api.reportHostOperation(api.host.id, api.recovery.id, { state, result: { state: 'unhealthy', healthy: false } }); + } + if (state === 'completed') api.reconcileNativeProvisioningReceipt(api.deploymentId); + const registeredGeneration = state === 'completed' ? 2 : 1; + expect(api.nativeInstance(server)?.generation).toBe(registeredGeneration); + expect(api.canCancelCreation(api.deploymentId)).toBe(true); + const cleanup = api.cancelCreation(api.deploymentId)!; + const removal = api.commandFor(cleanup); + expect(removal).toMatchObject({ kind: 'instance.cancel-create', generation: 1, + payload: { createOperationId: api.original.id, recoveryRebuildOperationId: api.recovery.id, recoveryGeneration: 2 } }); + expect(api.pendingCreationCancellations(api.host.id)).toEqual([api.original.id, api.recovery.id]); + expect(api.db().prepare('SELECT generation FROM native_provisioning WHERE deployment_id=?').get(cleanup)).toEqual({ generation: registeredGeneration }); + expect(api.pendingHostOperations(api.host.id).filter(operation => operation.kind.startsWith('instance.')).map(operation => operation.id)).toEqual([removal.id]); + // A delivered rebuild may report success after cancellation, while the + // cleanup command already owns deletion of both possible containers. + api.reportHostOperation(api.host.id, api.recovery.id, { state: 'completed', result: { state: 'unhealthy', healthy: false } }); + api.reconcileNativeProvisioningReceipt(api.deploymentId); + expect(api.nativeInstance(server)?.generation).toBe(registeredGeneration); + expect(api.db().prepare('SELECT * FROM ec_host_ports').all()).toHaveLength(2); + api.reportHostOperation(api.host.id, removal.id, { state: 'completed', result: { exists: false } }); + await api.advance(); + expect(api.getDeployment(cleanup)?.deployment.state).toBe('done'); + expect(api.getDeployment(api.deploymentId)?.deployment).toMatchObject({ state: 'cancelled', server_deleted_at: expect.any(String) }); + expect(api.nativeInstance(server)).toBeNull(); + expect(api.db().prepare('SELECT * FROM ec_host_ports').all()).toEqual([]); + expect(api.db().prepare('SELECT name FROM servers WHERE name=?').get(server)).toBeUndefined(); + expect(api.pendingCreationCancellations(api.host.id)).toEqual([]); + expect(api.cancelCreation(api.deploymentId)).toBe(cleanup); + }); + + it('does not adopt a changed generation without a completed recovery receipt', async () => { + const api = await recovering('rebuild'); + api.db().prepare('UPDATE native_instances SET generation=2 WHERE id=?').run(api.instance.id); + expect(api.canCancelCreation(api.deploymentId)).toBe(false); + expect(api.cancelCreation(api.deploymentId)).toBeNull(); + api.reportHostOperation(api.host.id, api.recovery.id, { state: 'completed' }); + api.db().prepare('UPDATE native_instances SET generation=3 WHERE id=?').run(api.instance.id); + expect(api.cancelCreation(api.deploymentId)).toBeNull(); + }); + + it('holds extended cleanup until the host supports recovery cancellation', async () => { + const api = await recovering('rebuild'); + const cleanup = api.cancelCreation(api.deploymentId)!; + api.heartbeat(api.host.id, { ...api.getHost(api.host.id)!.inventory, capabilities: ['instance.cancel-create'] }); + expect(api.pendingHostOperations(api.host.id)).toEqual([]); + await api.advance(); + expect(api.getDeployment(cleanup)?.deployment.stage).toContain('Update the host agent'); + expect(api.db().prepare('SELECT * FROM ec_host_ports').all()).toHaveLength(2); + api.heartbeat(api.host.id, { ...api.getHost(api.host.id)!.inventory, capabilities: ['instance.cancel-create', 'instance.unhealthy-recovery'] }); + expect(api.pendingHostOperations(api.host.id).filter(operation => operation.kind.startsWith('instance.')).map(operation => operation.id)).toEqual([api.commandFor(cleanup).id]); + }); +}); diff --git a/tests/unit/retention.test.ts b/tests/unit/retention.test.ts index 8a280c2..2aacf5b 100644 --- a/tests/unit/retention.test.ts +++ b/tests/unit/retention.test.ts @@ -230,3 +230,83 @@ it('protects cancelled workflows and linked file history through ambiguous and u 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'); + const host = hosts.createHost({ name: 'Recovery host' }); + const { config } = await import('../../src/lib/server/config.js'); + hosts.enrollHost(hosts.createEnrollment(host.id).token, { protocolVersion: 1, dockerRunning: true, + directoriesWritable: true, storageAvailable: 100_000_000_000, runtimeImages: [config.defaultRuntimeImage], + libraries: [{ id: 'lib', state: 'ready' }] }); + const generation = action === 'create' ? 1 : 2; + a.db.prepare("INSERT INTO native_instances(id,server,host_id,generation,library_id,image,port,created_at) VALUES('instance','s',?,?,'lib','image',1,?)") + .run(host.id, generation, old); + a.db.prepare("INSERT INTO server_deployments(id,server,action,state,stage,started_at,triggered_by) VALUES('recovering','s',?,'awaiting-health','Recovery',?,'test')") + .run(action === 'create' ? 'create' : 'rebuild', old); + const sha256 = putFileBlob(Buffer.from('original launch files'), a.db); + const launch = hosts.queueHostOperation(host.id, action === 'rebuild' ? 'instance.rebuild' : 'instance.create', { + name: 's', files: [{ path: 'configs/server.cfg', sha256, contentSha256: sha256 }] + }, { instanceId: 'instance', generation: action === 'rebuild' ? 1 : generation, + idempotencyKey: action === 'migration' ? 'recovering:migration-create:0' : 'recovering' }); + hosts.reportHostOperation(host.id, launch.id, { state: 'completed' }); + const restart = hosts.queueHostOperation(host.id, 'instance.restart', { unhealthyRecovery: true }, { instanceId: 'instance', generation }); + a.db.prepare("INSERT INTO native_provisioning(deployment_id,instance_id,generation,operation_id,recovery_step) VALUES('recovering','instance',?,?,'restart')") + .run(generation, restart.id); + a.db.prepare('UPDATE ec_host_operations SET updated_at=?').run(old); + a.saveFleetSettings({ ...a.fleetSettings(), actionKeepDays: 1, payloadKeepDays: 1, historyKeepCount: 0 }); + return { ...a, hosts, host, generation, launch, restart, sha256 }; +} + +it.each(['create', 'rebuild', 'migration'] as const)('keeps original %s launch files and ownership through recovery and releases them after success', async action => { + const a = await recoveryLaunch(action); + expect(a.pruneHistory('operation-payload')).toBe(0); + expect(a.hosts.getHostOperation(a.launch.id)?.payload.files).toEqual([ + { path: 'configs/server.cfg', sha256: a.sha256, contentBase64: Buffer.from('original launch files').toString('base64') } + ]); + a.hosts.reportHostOperation(a.host.id, a.restart.id, { state: 'completed' }); + const rebuild = a.hosts.queueHostOperation(a.host.id, 'instance.rebuild', { unhealthyRecovery: true, nextGeneration: a.generation + 1 }, + { instanceId: 'instance', generation: a.generation }); + a.hosts.reportHostOperation(a.host.id, rebuild.id, { state: 'completed' }); + a.db.prepare("UPDATE native_provisioning SET operation_id=?,recovery_step='rebuild',applied_at=? WHERE deployment_id='recovering'") + .run(rebuild.id, old); + a.db.prepare("UPDATE native_instances SET generation=generation+1 WHERE id='instance'").run(); + a.db.prepare('UPDATE ec_host_operations SET updated_at=?').run(old); + a.pruneHistory('operation-payload'); + a.pruneHistory('operation'); + expect(a.db.prepare('SELECT history_expired_at FROM ec_host_operations WHERE id=?').get(a.launch.id)) + .toEqual({ history_expired_at: null }); + expect(a.hosts.getHostOperation(a.launch.id)?.payload.files).toEqual([ + { path: 'configs/server.cfg', sha256: a.sha256, contentBase64: Buffer.from('original launch files').toString('base64') } + ]); + a.db.prepare("UPDATE server_deployments SET state='done',finished_at=? WHERE id='recovering'").run(old); + a.pruneHistory('operation-payload'); + expect(a.hosts.getHostOperation(a.launch.id)?.payload.files).toEqual([{ path: 'configs/server.cfg', sha256: a.sha256 }]); + a.pruneHistory('operation'); + expect(a.db.prepare('SELECT history_expired_at FROM ec_host_operations WHERE id=?').get(a.launch.id)) + .toEqual({ history_expired_at: expect.any(String) }); +}); + +it.each(['cancelled', 'failed'])('keeps original ownership for a %s recovered creation until a later lifecycle operation supersedes cleanup', async state => { + const a = await recoveryLaunch('create'); + a.hosts.reportHostOperation(a.host.id, a.restart.id, { state: 'completed' }); + const rebuild = a.hosts.queueHostOperation(a.host.id, 'instance.rebuild', { unhealthyRecovery: true, nextGeneration: 2 }, + { instanceId: 'instance', generation: 1 }); + a.hosts.reportHostOperation(a.host.id, rebuild.id, { state: 'completed' }); + a.db.prepare("UPDATE native_provisioning SET operation_id=?,recovery_step='rebuild',applied_at=? WHERE deployment_id='recovering'") + .run(rebuild.id, old); + a.db.prepare("UPDATE native_instances SET generation=2 WHERE id='instance'").run(); + a.db.prepare('UPDATE ec_host_operations SET updated_at=?').run(old); + a.db.prepare("UPDATE server_deployments SET state=?,finished_at=? WHERE id='recovering'").run(state, old); + const { canCancelCreation } = await import('../../src/lib/server/provisioning/cancel-creation.js'); + expect(canCancelCreation('recovering')).toBe(true); + a.pruneHistory('operation'); + expect(a.db.prepare('SELECT history_expired_at FROM ec_host_operations WHERE id=?').get(a.launch.id)) + .toEqual({ history_expired_at: null }); + a.db.prepare("INSERT INTO server_deployments(id,server,action,state,stage,started_at,finished_at,triggered_by) VALUES('replacement','s','restart','done','Done',?,?,'test')") + .run(old, old); + expect(canCancelCreation('recovering')).toBe(false); + a.pruneHistory('operation'); + expect(a.db.prepare('SELECT history_expired_at FROM ec_host_operations WHERE id=?').get(a.launch.id)) + .toEqual({ history_expired_at: expect.any(String) }); +});