From e271f313253aae73094f2d97bafa49a336f83def Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Tue, 6 Oct 2026 23:06:04 -0400 Subject: [PATCH] Require revocable authorization for automatic lobby removal --- agent/ec_agent.py | 36 +++++- agent/tests/test_agent.py | 55 ++++++++++ src/lib/autoscaling.ts | 1 + src/lib/server/autoscaling/admission.ts | 1 + src/lib/server/autoscaling/authorization.ts | 52 +++++++++ src/lib/server/autoscaling/index.ts | 46 ++++++-- src/lib/server/autoscaling/planner.ts | 1 + src/lib/server/autoscaling/snapshot.ts | 4 +- src/lib/server/provisioning/history.ts | 4 +- src/lib/server/provisioning/instances.ts | 11 +- src/lib/server/provisioning/native.ts | 17 ++- .../server/provisioning/safety-override.ts | 1 + .../[id]/authorize-removal/+server.ts | 9 ++ static/agent/ec-agent.py | 36 +++++- static/agent/manifest.json | 4 +- tests/unit/autoscaling-dry-run.test.ts | 2 +- tests/unit/autoscaling-planner.test.ts | 1 + tests/unit/autoscaling-removal.test.ts | 103 ++++++++++++++++++ 18 files changed, 357 insertions(+), 27 deletions(-) create mode 100644 src/lib/server/autoscaling/authorization.ts create mode 100644 src/routes/api/agent/operations/[id]/authorize-removal/+server.ts create mode 100644 tests/unit/autoscaling-removal.test.ts diff --git a/agent/ec_agent.py b/agent/ec_agent.py index 9c98576..502d503 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.6.0" +VERSION = "1.7.0" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -71,6 +71,10 @@ class Rejected(AgentError): """A definitive precondition failure, before any destructive action.""" +class AutoscalingInvalidated(Rejected): + """This removal needs a new policy decision, never a later local retry.""" + + def systemctl(*args): try: result = subprocess.run(["systemctl", *args], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=60, check=False) @@ -1356,7 +1360,7 @@ class Agent: if line.startswith("PRETTY_NAME="): os_name = line.split("=", 1)[1].strip('"') update = self.update_inventory() - capabilities = ["library.remove", "instance.cancel-create", "instance.safety-override", "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", "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 []) 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)), "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): @@ -1581,6 +1585,24 @@ class Agent: raise AgentError("Staged file checksum mismatch") return content + def authorize_autoscaling_removal(self, instance_id, generation, payload): + authorization = payload.get("autoscaling") + if not isinstance(authorization, dict) or not self.executing_id or payload.get("safetyOverride") or payload.get("stageId"): + raise AutoscalingInvalidated("Invalid autoscaling removal authorization") + started = time.monotonic() + try: + response = self.request("POST", "/api/agent/operations/" + urllib.parse.quote(self.executing_id, safe="") + "/authorize-removal", authorization) + except AgentError: + raise AutoscalingInvalidated("Autoscaling removal authorization unavailable; request a new decision") from None + ttl = response.get("validForMs") if isinstance(response, dict) else None + if not isinstance(response, dict) or response.get("authorized") is not True or isinstance(ttl, bool) or not isinstance(ttl, (int, float)) or not 0 < ttl <= 5000: + raise AutoscalingInvalidated("Autoscaling removal authorization was withdrawn or expired") + # Occupancy can change during the controller round trip. Check again + # within the short claim window, immediately before the durable checkpoint. + self.safe_to_disrupt(instance_id, generation) + if (time.monotonic() - started) * 1000 >= ttl: + raise AutoscalingInvalidated("Autoscaling removal authorization expired before mutation") + def lifecycle(self, kind, instance_id, generation, payload): if kind == "instance.cancel-create": return self.cancel_creation(instance_id, generation, payload) @@ -1610,6 +1632,8 @@ class Agent: # The authenticated controller records explicit administrator approval # for this immutable operation. Ownership and generation checks still apply. override_safety = payload.get("safetyOverride") is True + if "autoscaling" in payload and (kind != "instance.remove" or override_safety): + raise AutoscalingInvalidated("Autoscaling only permits removal with player safety checks") if kind == "instance.rebuild": next_generation = payload.get("nextGeneration", payload.get("newGeneration")) if isinstance(next_generation, bool) or not isinstance(next_generation, int) or next_generation <= generation: @@ -1618,6 +1642,8 @@ class Agent: self.validate_spec(spec) if not override_safety: self.safe_to_disrupt(instance_id, generation) + if "autoscaling" in payload: + self.authorize_autoscaling_removal(instance_id, generation, payload) self.checkpoint({"phase": "disrupting", "containerId": health.get("containerId"), "startedAt": health.get("startedAt")}) if payload.get("stageId") and kind == "instance.restart": before = self.activate(instance_id, generation, payload["stageId"], override_safety=override_safety) @@ -2203,7 +2229,11 @@ class Agent: try: result = self.dispatch(operation) except Held as err: + if operation.get("kind") == "instance.remove" and "autoscaling" in operation.get("payload", {}): + return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error=str(err)) return self.journal.update(operation["id"], "accepted", {"blocked": "occupied" if "Players" in str(err) else "unknown", "message": str(err)}) + except AutoscalingInvalidated as err: + return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error=str(err)) except Exception as err: # The command may already have changed Docker or files. Never infer # failure means nothing happened, and never retry without inspection. @@ -2239,6 +2269,8 @@ class Agent: instance_id, generation = operation.get("instanceId"), operation.get("generation") checkpoint = json.loads(row["checkpoint"]) if row.get("checkpoint") else {} try: + if kind == "instance.remove" and "autoscaling" in payload and not checkpoint: + return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error="Interrupted autoscaling removal requires a new decision") if kind == "instance.cancel-create": return self.journal.update(operation["id"], "completed", self.cancel_creation(instance_id, generation, payload)) if kind == "instance.create": diff --git a/agent/tests/test_agent.py b/agent/tests/test_agent.py index 4f4124b..8e93378 100644 --- a/agent/tests/test_agent.py +++ b/agent/tests/test_agent.py @@ -959,6 +959,61 @@ class AgentTests(unittest.TestCase): self.agent.rcon = lambda *_: {"ok": True, "players": 0} self.assertEqual(self.agent.execute(operation)["state"], "completed") + def autoscaling_operation(self, op_id="autoscaling-remove"): + return self.operation("instance.remove", {"autoscaling": {"policyId": "policy", "revision": 1, "expiresAt": 2000000}}, op_id) + + def test_autoscaling_occupied_or_unknown_removal_never_retries_later(self): + self.create() + for index, players in enumerate((2, None)): + operation = self.autoscaling_operation(f"autoscaling-{index}") + self.agent.rcon = lambda *_, count=players: {"ok": True, "players": count} + with patch.object(self.agent, "request", side_effect=AssertionError("Occupied removal requested authorization")): + result = self.agent.execute(operation) + self.assertEqual(result["state"], "failed") + self.assertTrue(json.loads(result["result"])["autoscalingInvalidated"]) + self.agent.rcon = lambda *_: {"ok": True, "players": 0} + self.assertEqual(self.agent.execute(operation)["state"], "failed") + self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls)) + + def test_autoscaling_removal_requires_fresh_one_shot_controller_authorization(self): + self.create() + self.agent.rcon = lambda *_: {"ok": True, "players": 0} + with patch.object(self.agent, "request", return_value={"authorized": False}) as request: + self.assertEqual(self.agent.execute(self.autoscaling_operation("revoked"))["state"], "failed") + self.assertTrue(request.call_args.args[1].endswith("/revoked/authorize-removal")) + with patch.object(self.agent, "request", return_value={"authorized": True, "validForMs": 5000}): + self.assertEqual(self.agent.execute(self.autoscaling_operation("valid"))["state"], "completed") + self.assertEqual(sum(args[:2] == ("container", "rm") for args in self.docker.calls), 1) + + def test_autoscaling_rechecks_players_after_authorization_and_rejects_expired_claim(self): + self.create() + with patch.object(self.agent, "rcon", side_effect=[{"ok": True, "players": 0}, {"ok": True, "players": 1}]), patch.object(self.agent, "request", return_value={"authorized": True, "validForMs": 5000}): + self.assertEqual(self.agent.execute(self.autoscaling_operation("joined"))["state"], "failed") + self.agent.rcon = lambda *_: {"ok": True, "players": 0} + with patch.object(self.agent, "request", return_value={"authorized": True, "validForMs": 5000}), patch.object(module.time, "monotonic", side_effect=[0, 6]): + self.assertEqual(self.agent.execute(self.autoscaling_operation("expired"))["state"], "failed") + self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls)) + + def test_autoscaling_restart_before_mutation_invalidates_instead_of_replaying(self): + self.create() + operation = self.autoscaling_operation() + self.agent.journal.accept(operation) + self.agent.journal.update(operation["id"], "running") + self.agent.journal.close() + self.agent = Agent(self.config, self.docker) + with patch.object(self.agent, "request", side_effect=AssertionError("Recovered command requested execution")): + result = self.agent.execute(operation) + self.assertEqual(result["state"], "failed") + self.assertTrue(json.loads(result["result"])["autoscalingInvalidated"]) + self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls)) + + def test_autoscaling_cannot_use_admin_override(self): + self.create() + operation = self.autoscaling_operation() + operation["payload"]["safetyOverride"] = True + self.assertEqual(self.agent.execute(operation)["state"], "failed") + self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls)) + def test_occupancy_hold_explains_local_rcon_failure_without_exposing_reply(self): self.create() operation = self.operation("instance.restart") diff --git a/src/lib/autoscaling.ts b/src/lib/autoscaling.ts index 507f690..8041937 100644 --- a/src/lib/autoscaling.ts +++ b/src/lib/autoscaling.ts @@ -13,6 +13,7 @@ export interface ScalingMember { name: string; identity: string; state: 'occupied' | 'empty' | 'creating' | 'removing' | 'stopped' | 'unknown' | 'protected'; players: number | null; observedAt: number | null; source: string | null; emptySince: number | null; blocker: string | null; operationId: string | null; + emptyStartedAt?: number | null; } export interface ScalingSnapshot { members: ScalingMember[]; reservedNames: string[]; keyedNames: string[]; blockers: string[]; diff --git a/src/lib/server/autoscaling/admission.ts b/src/lib/server/autoscaling/admission.ts index 5c089eb..b81621b 100644 --- a/src/lib/server/autoscaling/admission.ts +++ b/src/lib/server/autoscaling/admission.ts @@ -19,6 +19,7 @@ export function scalingAdmission(operationId: string): string | null { else { const policy: ScalingPolicy = { ...JSON.parse(row.settings), id: row.id, mode: row.mode, revision: row.revision, nextActionAt: 0, pausedReason: null, updatedAt: '' }; const requested = JSON.parse(intent.decision) as ScalingDecision; + if (requested.action === 'destroy' && row.mode !== 'create-destroy') return 'Automatic removal is disabled.'; const snapshot = scalingSnapshot(policy, Date.now(), operationId); if (requested.action === 'create') { snapshot.reservedNames = snapshot.reservedNames.filter(n => n !== requested.server); diff --git a/src/lib/server/autoscaling/authorization.ts b/src/lib/server/autoscaling/authorization.ts new file mode 100644 index 0000000..ff952da --- /dev/null +++ b/src/lib/server/autoscaling/authorization.ts @@ -0,0 +1,52 @@ +import { db } from '../db/index.js'; +import { scalingAdmission, scalingIntent } from './admission.js'; +import { getHost } from '../hosts/index.js'; + +export const autoscalingRemovalCapability = 'instance.autoscaling-remove'; +/** Authorization is attached to the immutable host command inside its submission transaction. */ +export function removalAuthorization(operationId: string, hostId: string): object | null { + const intent = scalingIntent(operationId); + if (!intent) return null; + if (!getHost(hostId)?.inventory.capabilities?.includes(autoscalingRemovalCapability)) throw new Error('Update the host agent before automatic removal.'); + const reason = scalingAdmission(operationId); + if (reason) throw new Error(reason); + const expiresAt = Date.now() + 60_000; + db().prepare('INSERT INTO autoscaling_authorizations(operation_id,policy_id,revision,expires_at) VALUES(?,?,?,?)') + .run(operationId, intent.policy_id, intent.revision, expiresAt); + return { policyId: intent.policy_id, revision: intent.revision, expiresAt }; +} + +/** Claim is the cancellation boundary. Lost replies never authorize a second claim. */ +export function claimRemovalAuthorization(hostId: string, commandId: string, input: Record): { authorized: boolean; validForMs?: number; reason?: string } { + return db().transaction(() => { + const now = Date.now(); + const row = db().prepare(`SELECT a.*,d.state AS deployment_state,o.state AS command_state FROM autoscaling_authorizations a + JOIN native_provisioning n ON n.deployment_id=a.operation_id JOIN server_deployments d ON d.id=n.deployment_id + JOIN native_instances i ON i.id=n.instance_id AND i.generation=n.generation AND i.retired_at IS NULL + JOIN ec_host_operations o ON o.id=n.operation_id AND o.host_id=i.host_id AND o.instance_id=i.id AND o.generation=i.generation + WHERE o.id=? AND o.host_id=? AND d.action='destroy' AND d.server_deleted_at IS NULL`).get(commandId, hostId) as + { operation_id: string; policy_id: string; revision: number; expires_at: number; claimed_at: number | null; revoked_at: number | null; deployment_state: string; command_state: string } | undefined; + if (!row || input.policyId !== row.policy_id || input.revision !== row.revision || input.expiresAt !== row.expires_at + || row.claimed_at !== null || row.revoked_at !== null || row.expires_at <= now + || row.deployment_state !== 'running' || !['pending','accepted','running'].includes(row.command_state)) return { authorized: false, reason: 'Removal authorization expired, changed, or was already claimed.' }; + const reason = scalingAdmission(row.operation_id); + if (reason) { db().prepare('UPDATE autoscaling_authorizations SET revoked_at=? WHERE operation_id=?').run(now, row.operation_id); return { authorized: false, reason }; } + db().prepare('UPDATE autoscaling_authorizations SET claimed_at=? WHERE operation_id=? AND claimed_at IS NULL').run(now, row.operation_id); + // Only a short final occupancy check may occur after this boundary. + return { authorized: true, validForMs: Math.min(5000, row.expires_at - now) }; + })(); +} + +export function revokePolicyRemovals(policyId: string): void { + if (!db().prepare('SELECT 1 FROM autoscaling_authorizations WHERE policy_id=?').get(policyId)) return; + db().prepare('UPDATE autoscaling_authorizations SET revoked_at=? WHERE policy_id=? AND claimed_at IS NULL AND revoked_at IS NULL').run(Date.now(), policyId); + const at = new Date().toISOString(); + const commands = db().prepare(`SELECT o.id FROM ec_host_operations o JOIN native_provisioning n ON n.operation_id=o.id + JOIN autoscaling_authorizations a ON a.operation_id=n.deployment_id + WHERE a.policy_id=? AND a.claimed_at IS NULL AND a.revoked_at IS NOT NULL AND o.state='pending' AND o.last_delivered_at IS NULL`).all(policyId) as { id: string }[]; + for (const command of commands) { + const result = JSON.stringify({ autoscalingInvalidated: true }); + db().prepare("UPDATE ec_host_operations SET state='failed',result=?,error='Autoscaling removal withdrawn before delivery.',updated_at=? WHERE id=?").run(result, at, command.id); + db().prepare("INSERT INTO ec_host_operation_events(operation_id,state,created_at,error) VALUES(?,'failed',?,'Autoscaling removal withdrawn before delivery.')").run(command.id, at); + } +} diff --git a/src/lib/server/autoscaling/index.ts b/src/lib/server/autoscaling/index.ts index 27ff008..a6ccb16 100644 --- a/src/lib/server/autoscaling/index.ts +++ b/src/lib/server/autoscaling/index.ts @@ -6,6 +6,11 @@ import { scalingSnapshot } from './snapshot.js'; import { nativeLaunchSelection, queueNativeCreationIntent } from '../provisioning/native.js'; import { cancelCreation } from '../provisioning/cancel-creation.js'; import { scalingAdmission } from './admission.js'; +import { destroyServer } from '../provisioning/instances.js'; +import { cancelRemoval } from '../provisioning/history.js'; +import { autoscalingRemovalCapability, revokePolicyRemovals } from './authorization.js'; +import { getHost } from '../hosts/index.js'; +import { getServer } from '../fleet/servers.js'; export class ScalingError extends Error { constructor(message: string, public status = 400) { super(message); } } interface PolicyRow { id: string; revision: number; settings: string; mode: ScalingPolicy['mode']; next_action_at: number; paused_reason: string | null; updated_at: string } @@ -23,11 +28,14 @@ export function previewScaling(input: unknown, id?: string, revision?: number): nextActionAt: existing?.nextActionAt ?? 0, pausedReason: null, updatedAt: existing?.updatedAt ?? '' }; const snapshot = scalingSnapshot(policy); const blockers: string[] = []; - if (policy.mode === 'create-destroy') blockers.push('Automatic removal is not available yet.'); if (liveScaling(policy.mode)) { if (snapshot.keyedNames.length < policy.maximum) blockers.push(`Save keys for at least ${policy.maximum} lobby identities before activating this policy.`); blockers.push(...snapshot.blockers); for (const member of snapshot.members) if (!['occupied', 'empty'].includes(member.state)) blockers.push(`${member.name}: ${member.blocker ?? member.state}. Settle this work before activation.`); + if (policy.mode === 'create-destroy') for (const member of snapshot.members) { + const hostId = getServer(member.name)?.host_id; + if (!hostId || !getHost(hostId)?.inventory.capabilities?.includes(autoscalingRemovalCapability)) blockers.push(`Update the host agent for ${member.name} before enabling automatic removal.`); + } if (db().prepare('SELECT 1 FROM autoscaling_decisions WHERE policy_id=? AND resolved=0').get(policy.id)) blockers.push('Settle the current autoscaling operation before editing a live policy.'); try { nativeLaunchSelection({ automatic: true }, policy.game); } catch (error) { blockers.push((error as Error).message); } } @@ -43,6 +51,7 @@ export function saveScaling(input: unknown, id?: string, revision?: number, toke if (!token || preview.token !== token) throw new ScalingError('The policy preview changed. Review it again before saving.', 409); if (preview.blockers.length) throw new ScalingError(preview.blockers[0]); const policy = preview.policy, settings = scalingSettings(policy), nextId = id ?? randomUUID(), at = new Date().toISOString(); + if (id) revokePolicyRemovals(id); if (!id && db().prepare('SELECT 1 FROM autoscaling_policies WHERE game=? AND map_slug=?').get(policy.game, policy.mapSlug)) throw new ScalingError('A policy already exists for this map.', 409); db().prepare(`INSERT INTO autoscaling_policies(id,game,map_slug,settings,mode,updated_at) VALUES(?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET settings=excluded.settings,mode=excluded.mode,revision=autoscaling_policies.revision+1,paused_reason=NULL,updated_at=excluded.updated_at`).run(nextId, policy.game, policy.mapSlug, JSON.stringify(settings), policy.mode, at); @@ -59,7 +68,7 @@ export function saveScaling(input: unknown, id?: string, revision?: number, toke export function setScalingGlobalPause(paused: boolean): void { db().transaction(() => { db().prepare("INSERT INTO settings(key,value) VALUES('autoscaling_paused',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value").run(paused ? '1' : '0'); - if (paused) for (const policy of listScalingPolicies()) reconcileScaling(policy); + if (paused) for (const policy of listScalingPolicies()) { revokePolicyRemovals(policy.id); reconcileScaling(policy); } })(); } @@ -69,18 +78,26 @@ function reconcileScaling(policy: ScalingPolicy): void { LEFT JOIN native_provisioning n ON n.deployment_id=d.id WHERE a.policy_id=? AND a.resolved=0`).all(policy.id) as { id: string; revision: number; operation_id: string; action: string; state: string; submitted_at: string | null; command_id: string | null }[]; for (const intent of intents) { - if (intent.action === 'create' && !intent.command_id && !intent.submitted_at + if (intent.action === 'create' && ['queued','waiting','cancelled'].includes(intent.state) && !intent.command_id && !intent.submitted_at && (scalingAdmission(intent.operation_id) || intent.state === 'cancelled')) cancelCreation(intent.operation_id, 'autoscaling-withdrawal', false); + if (intent.action === 'destroy' && scalingAdmission(intent.operation_id)) { + revokePolicyRemovals(policy.id); + if (!intent.command_id) cancelRemoval(intent.operation_id, false); + } const current = db().prepare('SELECT state,server_deleted_at FROM server_deployments WHERE id=?').get(intent.operation_id) as { state: string; server_deleted_at: string | null }; const terminal = ['done','warning','failed','cancelled'].includes(current.state); - const command = intent.command_id ? db().prepare('SELECT state FROM ec_host_operations WHERE id=?').get(intent.command_id) as { state: string } | undefined : undefined; + const command = intent.command_id ? db().prepare('SELECT state,result FROM ec_host_operations WHERE id=?').get(intent.command_id) as { state: string; result: string | null } | undefined : undefined; + const invalidated = command?.state === 'failed' && command.result && JSON.parse(command.result).autoscalingInvalidated === true; const receiptPending = current.state === 'cancelled' && !current.server_deleted_at && command?.state === 'completed' && !!db().prepare('SELECT 1 FROM native_provisioning WHERE deployment_id=? AND applied_at IS NULL').get(intent.operation_id); const settled = (!intent.command_id || !!command && ['completed','failed'].includes(command.state)) && !receiptPending; - if ((current.state === 'failed' || current.state === 'interrupted' || command?.state === 'unknown') && liveScaling(policy.mode)) { + if (!invalidated && (current.state === 'failed' || current.state === 'interrupted' || command?.state === 'unknown') && liveScaling(policy.mode)) { db().prepare('UPDATE autoscaling_policies SET paused_reason=?,updated_at=? WHERE id=?').run(`Inspect failed or uncertain Provisioning operation ${intent.operation_id} before resuming.`, new Date().toISOString(), policy.id); } - if (terminal && settled) db().prepare('UPDATE autoscaling_decisions SET resolved=1 WHERE id=?').run(intent.id); + if (terminal && settled) { + db().prepare('UPDATE autoscaling_decisions SET resolved=1 WHERE id=?').run(intent.id); + if (intent.action === 'destroy' && (invalidated || current.state === 'cancelled')) db().prepare("UPDATE autoscaling_members SET observation=json_set(observation,'$.emptySince',NULL) WHERE policy_id=?").run(policy.id); + } } if (!liveScaling(policy.mode) && !db().prepare('SELECT 1 FROM autoscaling_decisions WHERE policy_id=? AND resolved=0').get(policy.id)) { db().prepare('UPDATE autoscaling_policies SET owns_map=0 WHERE id=?').run(policy.id); @@ -100,6 +117,9 @@ export async function runAutoscaling(): Promise { if (db().prepare('SELECT 1 FROM autoscaling_decisions WHERE policy_id=? AND resolved=0').get(policy.id)) snapshot.blockers.unshift('Waiting for the current autoscaling operation to settle.'); const decision = planScaling(policy, snapshot, now); for (const member of snapshot.members) { + // Keep the observation interval that justified a pending removal; + // admission reads fresh evidence and revokes it on any new blocker. + if (member.state === 'removing' && db().prepare('SELECT 1 FROM autoscaling_decisions WHERE policy_id=? AND operation_id=? AND resolved=0').get(policy.id, member.operationId)) continue; const observation = emptyObservation(member, { identity: member.identity, observedAt: member.observedAt, emptySince: member.emptySince, evaluatedAt: now }, now); db().prepare(`INSERT INTO autoscaling_members(policy_id,server,identity,observation) VALUES(?,?,?,?) ON CONFLICT(policy_id,server) DO UPDATE SET identity=excluded.identity,observation=excluded.observation`).run(policy.id, member.name, member.identity, JSON.stringify(observation)); @@ -108,7 +128,7 @@ export async function runAutoscaling(): Promise { const evidence = JSON.stringify({ members: snapshot.members, reservations: snapshot.reservedNames, keyCount: snapshot.keyedNames.length, blockers: snapshot.blockers }); const fingerprint = createHash('sha256').update(JSON.stringify([policy.revision, decision, snapshot.members.map(m => [m.name, m.identity, m.state])])).digest('hex'); const last = db().prepare('SELECT id,fingerprint FROM autoscaling_decisions WHERE policy_id=? ORDER BY evaluated_at DESC,rowid DESC LIMIT 1').get(policy.id) as { id: string; fingerprint: string } | undefined; - const execute = liveScaling(policy.mode) && decision.action === 'create'; + const execute = liveScaling(policy.mode) && (decision.action === 'create' || policy.mode === 'create-destroy' && decision.action === 'destroy'); const decisionId = randomUUID(); if (!execute && last?.fingerprint === fingerprint) db().prepare('UPDATE autoscaling_decisions SET last_seen_at=?,evidence=? WHERE id=?').run(now, evidence, last.id); else db().prepare('INSERT INTO autoscaling_decisions(id,policy_id,revision,evaluated_at,last_seen_at,fingerprint,decision,evidence) VALUES(?,?,?,?,?,?,?,?)') @@ -117,10 +137,12 @@ export async function runAutoscaling(): Promise { db().prepare('UPDATE autoscaling_decisions SET resolved=0 WHERE id=?').run(decisionId); try { db().transaction(() => { - const selection = nativeLaunchSelection({ automatic: true }, policy.game); - const result = queueNativeCreationIntent({ server: decision.server!, game: policy.game, mapSlug: policy.mapSlug, selection, - createdBy: `autoscaling:${policy.id}`, autoscalingDecisionId: decisionId }); + const result = decision.action === 'create' + ? queueNativeCreationIntent({ server: decision.server!, game: policy.game, mapSlug: policy.mapSlug, + selection: nativeLaunchSelection({ automatic: true }, policy.game), createdBy: `autoscaling:${policy.id}`, autoscalingDecisionId: decisionId }) + : destroyServer(decision.server!, `autoscaling:${policy.id}`, decisionId); if (!result.ok) throw new Error(result.message); + if (!result.deploymentId) throw new Error('Provisioning did not return an operation identity.'); db().prepare('UPDATE autoscaling_decisions SET operation_id=? WHERE id=?').run(result.deploymentId, decisionId); db().prepare("INSERT INTO provisioning_sources(operation_id,kind,source_id,server) VALUES(?,'autoscaling',?,?)").run(result.deploymentId, policy.id, decision.server); db().prepare('UPDATE autoscaling_policies SET next_action_at=? WHERE id=?').run(now + policy.cooldownMinutes * 60_000, policy.id); @@ -128,7 +150,7 @@ export async function runAutoscaling(): Promise { })(); } catch (error) { db().prepare('UPDATE autoscaling_decisions SET resolved=1 WHERE id=?').run(decisionId); - db().prepare('UPDATE autoscaling_policies SET paused_reason=? WHERE id=?').run(`Creation could not be queued: ${(error as Error).message}`, policy.id); + db().prepare('UPDATE autoscaling_policies SET paused_reason=? WHERE id=?').run(`Scaling action could not be queued: ${(error as Error).message}`, policy.id); } } db().prepare(`INSERT OR IGNORE INTO autoscaling_samples(policy_id,sampled_at,mode,occupied,empty,unknown,deficit) VALUES(?,?,?,?,?,?,?)`) @@ -160,5 +182,5 @@ export function scalingReport(): ScalingReport { COALESCE(SUM(CASE WHEN unknown>0 THEN MIN(duration,15000) ELSE 0 END)/60000.0,0) AS unknownMinutes FROM intervals WHERE duration<=30000`).get(policy.id) as ScalingView['totals']; return { policy, snapshot, decision, token: '', blockers: [], history, lastEvaluatedAt: history[0]?.lastSeenAt ?? null, totals }; }); - return { policies, globalPaused: scalingGloballyPaused(), liveAvailable: true, removalAvailable: false }; + return { policies, globalPaused: scalingGloballyPaused(), liveAvailable: true, removalAvailable: true }; } diff --git a/src/lib/server/autoscaling/planner.ts b/src/lib/server/autoscaling/planner.ts index 17a12c7..0476192 100644 --- a/src/lib/server/autoscaling/planner.ts +++ b/src/lib/server/autoscaling/planner.ts @@ -7,6 +7,7 @@ export function emptyObservation(member: ScalingMember, previous: EmptyObservati const stamp = member.observedAt; const empty = member.state === 'empty' && stamp !== null && stamp <= now && now - stamp <= ONLINE_MAX_MS; const continuous = empty && previous?.identity === member.identity && previous.emptySince !== null + && (!member.emptyStartedAt || previous.emptySince >= member.emptyStartedAt) && previous.observedAt !== null && stamp >= previous.observedAt && stamp - previous.observedAt <= ONLINE_MAX_MS && now >= previous.evaluatedAt && now - previous.evaluatedAt <= ONLINE_MAX_MS; return { identity: member.identity, observedAt: stamp, emptySince: empty ? continuous ? previous!.emptySince : stamp : null, evaluatedAt: now }; diff --git a/src/lib/server/autoscaling/snapshot.ts b/src/lib/server/autoscaling/snapshot.ts index 25af71a..6faa54e 100644 --- a/src/lib/server/autoscaling/snapshot.ts +++ b/src/lib/server/autoscaling/snapshot.ts @@ -53,7 +53,9 @@ export function scalingSnapshot(policy: ScalingPolicy, now = Date.now(), exceptO else if (operation) { state = operation.action === 'create' ? 'creating' : operation.action === 'destroy' ? 'removing' : 'protected'; blocker = `Provisioning ${operation.id} is ${operation.state}`; } else if (row.desired_state === 'stopped' || ['stopped','exited','created','dead'].includes(row.container_state ?? '')) state = 'stopped'; else if (observation.health === 'online' && players !== null && observedAt !== null && observedAt <= now && freshStatusTimestamp(new Date(observedAt).toISOString(), now)) state = players > 0 ? 'occupied' : 'empty'; - return { name: row.name, identity, state, players, observedAt, source: source ?? null, emptySince: null, blocker, operationId: operation?.id ?? null }; + const transitions = [row.inactive_since, observation.inactive_since].map(at => Date.parse(at ?? '')).filter(at => Number.isFinite(at) && at <= now); + return { name: row.name, identity, state, players, observedAt, source: source ?? null, emptySince: null, + emptyStartedAt: transitions.length ? Math.max(...transitions) : null, blocker, operationId: operation?.id ?? null }; }); for (const row of reservations.filter(matches)) if (!members.some(m => m.name === row.name)) members.push({ name: row.name, identity: 'reserved', state: 'creating', players: null, observedAt: null, source: null, emptySince: null, blocker: 'Creation is pending.', operationId: operations.find(o => o.server === row.name)?.id ?? null }); const retained = handle.prepare('SELECT server,observation FROM autoscaling_members WHERE policy_id=?').all(policy.id) as { server: string; observation: string }[]; diff --git a/src/lib/server/provisioning/history.ts b/src/lib/server/provisioning/history.ts index 8c6f510..7a6cda3 100644 --- a/src/lib/server/provisioning/history.ts +++ b/src/lib/server/provisioning/history.ts @@ -148,13 +148,13 @@ export function canCancelRemoval(id: string): boolean { } /** Competes atomically with the worker's claim, before any external removal. */ -export function cancelRemoval(id: string): boolean { +export function cancelRemoval(id: string, pausePolicy = true): boolean { return db().transaction(() => { const finishedAt = new Date().toISOString(); const result = db().prepare(`UPDATE server_deployments SET state = 'cancelled', stage = 'Removal cancelled', finished_at = ?, error = NULL WHERE id = ? AND server_deleted_at IS NULL AND action = 'destroy' AND state = 'waiting' AND stage != ?`).run(finishedAt, id, REMOVING_SERVER); if (!result.changes) return false; - pauseScalingForOperation(id); + if (pausePolicy) pauseScalingForOperation(id); recordProgress(id, { blocker: null, checks: [], event: { kind: 'outcome', message: 'Queued removal cancelled. The server and its files were kept.', details: { state: 'cancelled', finishedAt } } }); diff --git a/src/lib/server/provisioning/instances.ts b/src/lib/server/provisioning/instances.ts index 3db0429..e03c197 100644 --- a/src/lib/server/provisioning/instances.ts +++ b/src/lib/server/provisioning/instances.ts @@ -1,4 +1,5 @@ -import { scalingOwnershipBlocker } from '../autoscaling/ownership.js'; +import { scalingOwnershipBlocker, assertScalingOwnership } from '../autoscaling/ownership.js'; +import { mapIdentity } from '../autoscaling/snapshot.js'; import { db } from '../db/index.js'; import { activeBetaRemovalBlocker, getServer } from '../fleet/servers.js'; import type { GameMode, ZombieGame } from '../fleet/map-aliases.js'; @@ -68,9 +69,13 @@ function hasActiveCreation(name: string): boolean { return !!db().prepare("SELECT 1 FROM server_deployments WHERE server = ? AND server_deleted_at IS NULL AND action = 'create' AND state IN ('queued', 'running', 'awaiting-health')").get(name); } -export function destroyServer(name: string, triggeredBy = 'ui'): DestroyServerResult & { deploymentId?: string } { +export function destroyServer(name: string, triggeredBy = 'ui', autoscalingDecisionId?: string): DestroyServerResult & { deploymentId?: string } { const owned = scalingOwnershipBlocker(name); - if (owned) return { ok: false, kind: 'conflict', message: owned, warnings: [] }; + if (autoscalingDecisionId) { + const identity = mapIdentity(getServer(name) ?? { name }); + if (!identity) return { ok: false, kind: 'conflict', message: 'Map identity unavailable.', warnings: [] }; + assertScalingOwnership(identity.game, identity.mapSlug, autoscalingDecisionId); + } else if (owned) return { ok: false, kind: 'conflict', message: owned, warnings: [] }; const cancelled = cancelQueuedNativeCreation(name, triggeredBy); if (cancelled) return { ok: true, warnings: [], deploymentId: cancelled }; const blocker = activeBetaRemovalBlocker(name); diff --git a/src/lib/server/provisioning/native.ts b/src/lib/server/provisioning/native.ts index 36bbc02..1643e75 100644 --- a/src/lib/server/provisioning/native.ts +++ b/src/lib/server/provisioning/native.ts @@ -1,4 +1,6 @@ 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 } from './concurrency.js'; import { isIP } from 'node:net'; @@ -18,7 +20,7 @@ import { prepareNativeLaunch, evaluateNativePlacement, type PreparedNativeLaunch import { getServerKey, saveServerKey } from '../fleet/server-keys.js'; import { activeBetaRemovalBlocker, getServer } from '../fleet/servers.js'; import { nextLobbyNumber, type CreateServerInput, type CreateServerResult } from './instances.js'; -import { beginDeployment, deploymentStage, finishDeployment, REMOVING_SERVER } from './history.js'; +import { beginDeployment, deploymentStage, finishDeployment, REMOVING_SERVER, cancelRemoval } from './history.js'; import { recordProgress, type CheckInput } from './progress.js'; import { trackActivity } from '../pipeline/activity.js'; import { fleetSettings } from '../settings.js'; @@ -659,7 +661,11 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { JOIN watchdog_incidents i ON i.id=s.source_id JOIN server_deployments d ON d.id=s.operation_id WHERE s.operation_id=? AND s.kind='watchdog' AND i.kind='health' AND d.triggered_by LIKE 'watchdog:%'`).get(job.id); if (recoveredWatchdog && occupancy.players !== null) { resolveRecoveredHealthOperation(job.id, occupancy.players, occupancy.source); return false; } - if (occupancy.players !== 0) { hold(job, occupancy.players !== null ? `Waiting for ${occupancy.players} player(s) to leave.` : 'Waiting for a fresh local player count.'); return false; } + if (occupancy.players !== 0) { + if (job.action === 'destroy' && scalingIntent(job.id)) cancelRemoval(job.id, false); + else hold(job, occupancy.players !== null ? `Waiting for ${occupancy.players} player(s) to leave.` : 'Waiting for a fresh local player count.'); + return false; + } } return true; }; @@ -729,6 +735,13 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { ? 'Host eligibility checked; player safety checks bypassed by admin override' : 'Host eligibility and safety checks passed' } }); const payload: Record = files.length ? { stageId: job.id } : {}; if (overrideSafety) payload.safetyOverride = true; + if (job.action === 'destroy') { + const authorization = removalAuthorization(job.id, instance!.host_id); + if (authorization) { + if (overrideSafety) throw new Error('Autoscaling never bypasses player safety.'); + payload.autoscaling = authorization; + } + } if (prepared) { deploymentStage(job.id, 'Preparing launch files for the host agent', { phase: 'prepare', recipeStatus: 'attempted', checkpoint: { key: 'startup-queue', label: 'Startup capacity available' } }); diff --git a/src/lib/server/provisioning/safety-override.ts b/src/lib/server/provisioning/safety-override.ts index cdd96f8..63e8db6 100644 --- a/src/lib/server/provisioning/safety-override.ts +++ b/src/lib/server/provisioning/safety-override.ts @@ -19,6 +19,7 @@ export function safetyOverrideAvailability(id: string): { applicable: boolean; u AND d.state IN ('queued','waiting') AND d.submitted_at IS NULL AND n.operation_id IS NULL AND NOT EXISTS (SELECT 1 FROM ec_host_migrations m WHERE m.deployment_id=d.id) AND NOT EXISTS (SELECT 1 FROM native_creation_cancellations c WHERE c.cleanup_id=d.id) + AND NOT EXISTS (SELECT 1 FROM autoscaling_decisions a WHERE a.operation_id=d.id) AND NOT EXISTS (SELECT 1 FROM provisioning_safety_overrides o WHERE o.deployment_id=d.id)`) .get(id) as { host_id: string } | undefined; if (!target) return { applicable: false, unavailableReason: null }; diff --git a/src/routes/api/agent/operations/[id]/authorize-removal/+server.ts b/src/routes/api/agent/operations/[id]/authorize-removal/+server.ts new file mode 100644 index 0000000..54eea2e --- /dev/null +++ b/src/routes/api/agent/operations/[id]/authorize-removal/+server.ts @@ -0,0 +1,9 @@ +import { json } from '@sveltejs/kit'; +import { authenticateAgent, hostApiError, objectBody } from '$lib/server/hosts/http.js'; +import { claimRemovalAuthorization } from '$lib/server/autoscaling/authorization.js'; + +export async function POST({ request, params }: { request: Request; params: { id: string } }) { + const host = authenticateAgent(request); + try { return json(claimRemovalAuthorization(host.id, params.id, await objectBody(request, 4096))); } + catch (cause) { return hostApiError(cause); } +} diff --git a/static/agent/ec-agent.py b/static/agent/ec-agent.py index 9c98576..502d503 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.6.0" +VERSION = "1.7.0" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -71,6 +71,10 @@ class Rejected(AgentError): """A definitive precondition failure, before any destructive action.""" +class AutoscalingInvalidated(Rejected): + """This removal needs a new policy decision, never a later local retry.""" + + def systemctl(*args): try: result = subprocess.run(["systemctl", *args], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=60, check=False) @@ -1356,7 +1360,7 @@ class Agent: if line.startswith("PRETTY_NAME="): os_name = line.split("=", 1)[1].strip('"') update = self.update_inventory() - capabilities = ["library.remove", "instance.cancel-create", "instance.safety-override", "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", "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 []) 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)), "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): @@ -1581,6 +1585,24 @@ class Agent: raise AgentError("Staged file checksum mismatch") return content + def authorize_autoscaling_removal(self, instance_id, generation, payload): + authorization = payload.get("autoscaling") + if not isinstance(authorization, dict) or not self.executing_id or payload.get("safetyOverride") or payload.get("stageId"): + raise AutoscalingInvalidated("Invalid autoscaling removal authorization") + started = time.monotonic() + try: + response = self.request("POST", "/api/agent/operations/" + urllib.parse.quote(self.executing_id, safe="") + "/authorize-removal", authorization) + except AgentError: + raise AutoscalingInvalidated("Autoscaling removal authorization unavailable; request a new decision") from None + ttl = response.get("validForMs") if isinstance(response, dict) else None + if not isinstance(response, dict) or response.get("authorized") is not True or isinstance(ttl, bool) or not isinstance(ttl, (int, float)) or not 0 < ttl <= 5000: + raise AutoscalingInvalidated("Autoscaling removal authorization was withdrawn or expired") + # Occupancy can change during the controller round trip. Check again + # within the short claim window, immediately before the durable checkpoint. + self.safe_to_disrupt(instance_id, generation) + if (time.monotonic() - started) * 1000 >= ttl: + raise AutoscalingInvalidated("Autoscaling removal authorization expired before mutation") + def lifecycle(self, kind, instance_id, generation, payload): if kind == "instance.cancel-create": return self.cancel_creation(instance_id, generation, payload) @@ -1610,6 +1632,8 @@ class Agent: # The authenticated controller records explicit administrator approval # for this immutable operation. Ownership and generation checks still apply. override_safety = payload.get("safetyOverride") is True + if "autoscaling" in payload and (kind != "instance.remove" or override_safety): + raise AutoscalingInvalidated("Autoscaling only permits removal with player safety checks") if kind == "instance.rebuild": next_generation = payload.get("nextGeneration", payload.get("newGeneration")) if isinstance(next_generation, bool) or not isinstance(next_generation, int) or next_generation <= generation: @@ -1618,6 +1642,8 @@ class Agent: self.validate_spec(spec) if not override_safety: self.safe_to_disrupt(instance_id, generation) + if "autoscaling" in payload: + self.authorize_autoscaling_removal(instance_id, generation, payload) self.checkpoint({"phase": "disrupting", "containerId": health.get("containerId"), "startedAt": health.get("startedAt")}) if payload.get("stageId") and kind == "instance.restart": before = self.activate(instance_id, generation, payload["stageId"], override_safety=override_safety) @@ -2203,7 +2229,11 @@ class Agent: try: result = self.dispatch(operation) except Held as err: + if operation.get("kind") == "instance.remove" and "autoscaling" in operation.get("payload", {}): + return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error=str(err)) return self.journal.update(operation["id"], "accepted", {"blocked": "occupied" if "Players" in str(err) else "unknown", "message": str(err)}) + except AutoscalingInvalidated as err: + return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error=str(err)) except Exception as err: # The command may already have changed Docker or files. Never infer # failure means nothing happened, and never retry without inspection. @@ -2239,6 +2269,8 @@ class Agent: instance_id, generation = operation.get("instanceId"), operation.get("generation") checkpoint = json.loads(row["checkpoint"]) if row.get("checkpoint") else {} try: + if kind == "instance.remove" and "autoscaling" in payload and not checkpoint: + return self.journal.update(operation["id"], "failed", {"autoscalingInvalidated": True}, error="Interrupted autoscaling removal requires a new decision") if kind == "instance.cancel-create": return self.journal.update(operation["id"], "completed", self.cancel_creation(instance_id, generation, payload)) if kind == "instance.create": diff --git a/static/agent/manifest.json b/static/agent/manifest.json index 6889268..2d6d71a 100644 --- a/static/agent/manifest.json +++ b/static/agent/manifest.json @@ -1,8 +1,8 @@ { - "version": "1.6.0", + "version": "1.7.0", "protocolVersion": 1, "platform": "debian-amd64", - "sha256": "bcc0a207125f2a1ad754c294decf2238064995ba93183b845a589a41a41925f6", + "sha256": "4b40206f477dcf35f0506281b3dddca32b57214a964bcf5169c8611dc705132d", "updaterProtocolVersion": 1, "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" } diff --git a/tests/unit/autoscaling-dry-run.test.ts b/tests/unit/autoscaling-dry-run.test.ts index b36c16c..167ad18 100644 --- a/tests/unit/autoscaling-dry-run.test.ts +++ b/tests/unit/autoscaling-dry-run.test.ts @@ -31,7 +31,7 @@ it('rejects duplicate policies, stale edits and changed previews', async () => { expect(() => scaling.saveScaling(settings, undefined, undefined, preview.token)).toThrow('already exists'); expect(() => scaling.previewScaling(settings, saved.id, 0)).toThrow('changed'); expect(() => scaling.saveScaling({ ...settings, maximum: 5 }, saved.id, saved.revision, preview.token)).toThrow('preview changed'); - expect(() => scaling.saveScaling({ ...settings, mode: 'create-destroy' }, saved.id, saved.revision, scaling.previewScaling({ ...settings, mode: 'create-destroy' }, saved.id, saved.revision).token)).toThrow('not available'); + expect(() => scaling.saveScaling({ ...settings, mode: 'create-destroy' }, saved.id, saved.revision, scaling.previewScaling({ ...settings, mode: 'create-destroy' }, saved.id, saved.revision).token)).toThrow('Save keys'); }); it('pauses evaluations and prunes expired observations without deleting current history', async () => { diff --git a/tests/unit/autoscaling-planner.test.ts b/tests/unit/autoscaling-planner.test.ts index 0517eef..356ded4 100644 --- a/tests/unit/autoscaling-planner.test.ts +++ b/tests/unit/autoscaling-planner.test.ts @@ -61,5 +61,6 @@ describe('continuous empty evidence', () => { expect(emptyObservation(member(1, 'unknown'), previous, now).emptySince).toBeNull(); expect(emptyObservation({ ...member(1), identity: 'replacement', observedAt: now + 1 }, previous, now + 1).emptySince).toBe(now + 1); expect(emptyObservation({ ...member(1), observedAt: now + 120_000 }, previous, now + 120_000).emptySince).toBe(now + 120_000); + expect(emptyObservation({ ...member(1), emptyStartedAt: now + 10_000, observedAt: now + 15_000 }, previous, now + 15_000).emptySince).toBe(now + 15_000); }); }); diff --git a/tests/unit/autoscaling-removal.test.ts b/tests/unit/autoscaling-removal.test.ts new file mode 100644 index 0000000..0abb9d7 --- /dev/null +++ b/tests/unit/autoscaling-removal.test.ts @@ -0,0 +1,103 @@ +import { afterEach, beforeEach, 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/integrations/rcon.js', () => ({ rconGetPlayerCount: vi.fn(async () => 0) })); +let fixture: Fixture; +beforeEach(async () => { vi.resetModules(); fixture = await newFixture([]); }); +afterEach(async () => { vi.useRealTimers(); await fixture.cleanup(); vi.restoreAllMocks(); }); +const names = ['t6_zombie_town_1', 't6_zombie_town_2']; +const settings = { game: 't6', mapSlug: 'town', maximum: 2, mode: 'create-destroy', emptyMinutes: 1, cooldownMinutes: 1 }; + +async function setup() { + const host = await prepareNativeHost(names); + const { db } = await import('../../src/lib/server/db/index.js'); + const scaling = await import('../../src/lib/server/autoscaling/index.js'); + const authorization = await import('../../src/lib/server/autoscaling/authorization.js'); + db().prepare("UPDATE ec_hosts SET inventory=json_set(inventory,'$.capabilities',json(?)) WHERE id=?").run(JSON.stringify([authorization.autoscalingRemovalCapability]), host.host.id); + const base = Date.now(); vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(base); + for (const name of names) { + db().prepare("INSERT INTO servers(name,game,map,map_slug,container_state,last_players) VALUES(?,'t6','zm_transit','town','running',0)").run(name); + db().prepare('INSERT INTO server_keys(server_id,pluto_key) VALUES(?,?)').run(name, `key-${name}`); + } + const fresh = (offset: number) => { + vi.setSystemTime(base + offset); const at = new Date().toISOString(); + db().prepare('UPDATE servers SET container_checked_at=?,last_seen_at=?,last_checked_at=?').run(at, at, at); + db().prepare('UPDATE ec_hosts SET last_heartbeat=?').run(at); + }; + fresh(0); + const saved = scaling.saveScaling(settings, undefined, undefined, scaling.previewScaling(settings).token); + for (const offset of [0, 30_000, 60_000]) { fresh(offset); await scaling.runAutoscaling(); } + const intent = db().prepare('SELECT operation_id FROM autoscaling_decisions WHERE resolved=0').get() as { operation_id: string }; + expect(intent?.operation_id).toBeTruthy(); + const backend = await import('../../src/lib/server/backends/native.js'); + vi.spyOn(backend, 'nativeRequest').mockResolvedValue({ healthy: true, state: 'running', exists: true } as never); + const { processProvisioning } = await import('../../src/lib/server/provisioning/worker.js'); + const dispatch = async () => { + await processProvisioning(); + const row = db().prepare('SELECT operation_id FROM native_provisioning WHERE deployment_id=?').get(intent.operation_id) as { operation_id: string }; + expect(row.operation_id).toBeTruthy(); + const { getHostOperation } = await import('../../src/lib/server/hosts/index.js'); + const command = getHostOperation(row.operation_id)!; + return { id: row.operation_id, payload: command.payload.autoscaling as Record }; + }; + return { db, scaling, authorization, policy: saved, operationId: intent.operation_id, hostId: host.host.id, fresh, dispatch, processProvisioning }; +} + +it('removes only after a full observed empty interval and claims authorization once', async () => { + const s = await setup(); + const { enableSafetyOverride } = await import('../../src/lib/server/provisioning/safety-override.js'); + expect(enableSafetyOverride(s.operationId, { id: 'admin', username: 'admin' }).ok).toBe(false); + const command = await s.dispatch(); + expect(s.authorization.claimRemovalAuthorization('wrong-host', command.id, command.payload).authorized).toBe(false); + expect(s.authorization.claimRemovalAuthorization(s.hostId, command.id, command.payload)).toMatchObject({ authorized: true, validForMs: 5000 }); + expect(s.authorization.claimRemovalAuthorization(s.hostId, command.id, command.payload).authorized).toBe(false); +}); + +it('withdraws a queued removal when a player arrives and restarts the empty interval', async () => { + const s = await setup(); + s.db().prepare('UPDATE servers SET last_players=1 WHERE name=?').run(names[0]); + await s.scaling.runAutoscaling(); + expect(s.db().prepare('SELECT state FROM server_deployments WHERE id=?').get(s.operationId)).toEqual({ state: 'cancelled' }); + s.db().prepare('UPDATE servers SET last_players=0').run(); + s.fresh(90_000); await s.scaling.runAutoscaling(); + expect(s.db().prepare('SELECT COUNT(*) AS n FROM autoscaling_decisions WHERE resolved=0').get()).toEqual({ n: 0 }); +}); + +it('rechecks controller occupancy after the async probe even before monitoring refreshes', async () => { + const s = await setup(); + const rcon = await import('../../src/lib/server/integrations/rcon.js'); + vi.mocked(rcon.rconGetPlayerCount).mockResolvedValueOnce(1); + await s.processProvisioning(); + expect(s.db().prepare('SELECT state FROM server_deployments WHERE id=?').get(s.operationId)).toEqual({ state: 'cancelled' }); + expect(s.db().prepare('SELECT COUNT(*) AS n FROM autoscaling_authorizations').get()).toEqual({ n: 0 }); +}); + +it('revokes a delivered removal on policy pause without claiming it was undone', async () => { + const s = await setup(); const command = await s.dispatch(); + s.db().prepare('UPDATE ec_host_operations SET last_delivered_at=? WHERE id=?').run(new Date().toISOString(), command.id); + const off = { ...settings, mode: 'off' }; + s.scaling.saveScaling(off, s.policy.id, s.policy.revision, s.scaling.previewScaling(off, s.policy.id, s.policy.revision).token); + expect(s.authorization.claimRemovalAuthorization(s.hostId, command.id, command.payload).authorized).toBe(false); + expect(s.db().prepare('SELECT owns_map FROM autoscaling_policies').get()).toEqual({ owns_map: 1 }); + expect(s.db().prepare('SELECT state FROM server_deployments WHERE id=?').get(s.operationId)).toEqual({ state: 'running' }); +}); + +it('withdraws undelivered commands and reconciles invalidation without a retry or failure pause', async () => { + const s = await setup(); const command = await s.dispatch(); + s.scaling.setScalingGlobalPause(true); + expect(s.db().prepare('SELECT state FROM ec_host_operations WHERE id=?').get(command.id)).toEqual({ state: 'failed' }); + await s.processProvisioning(); await s.scaling.runAutoscaling(); + expect(s.scaling.getScalingPolicy(s.policy.id)?.pausedReason).toBeNull(); + expect(s.db().prepare('SELECT COUNT(*) AS n FROM autoscaling_decisions WHERE resolved=0').get()).toEqual({ n: 0 }); +}); + +it('rejects expired, changed revision, and newly protected removal authorizations', async () => { + const s = await setup(); const command = await s.dispatch(); + expect(s.authorization.claimRemovalAuthorization(s.hostId, command.id, { ...command.payload, revision: 999 }).authorized).toBe(false); + s.db().prepare("INSERT INTO maintenance_jobs(server,requested_at) VALUES(?,?)").run(names[1], new Date().toISOString()); + expect(s.authorization.claimRemovalAuthorization(s.hostId, command.id, command.payload).authorized).toBe(false); + s.db().prepare('DELETE FROM maintenance_jobs').run(); + s.fresh(130_000); + expect(s.authorization.claimRemovalAuthorization(s.hostId, command.id, command.payload).authorized).toBe(false); +});