From 1405114a1435e693e780eef54d0d84023b1b9c17 Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Mon, 5 Oct 2026 10:48:38 -0400 Subject: [PATCH] Keep native collection launches responsive --- README.md | 7 +- docs/deployment-blueprints.md | 4 + package.json | 9 +- src/lib/server/backends/native.ts | 4 +- src/lib/server/db/index.ts | 3 + src/lib/server/db/schema.sql | 1 + src/lib/server/fleet/launch-recipes.ts | 23 +- src/lib/server/hosts/delivery.ts | 62 +++++ src/lib/server/hosts/index.ts | 103 ++++++-- src/lib/server/hosts/schema.ts | 4 + src/lib/server/provisioning/history.ts | 6 +- src/lib/server/provisioning/instances.ts | 18 +- .../server/provisioning/launch-preparation.ts | 71 ++++++ src/lib/server/provisioning/launch-worker.ts | 38 +++ src/lib/server/provisioning/native.ts | 185 +++++++++----- src/routes/api/agent/operations/+server.ts | 10 +- .../api/agent/operations/[id]/+server.ts | 4 +- tests/helpers/fixture.ts | 2 + tests/unit/agent-operation-delivery.test.ts | 142 +++++++++++ .../unit/collection-match-performance.test.ts | 7 +- .../launch-recipe-summary-migration.test.ts | 46 ++++ tests/unit/launch-recipes.test.ts | 31 +++ tests/unit/native-hosts.test.ts | 64 ++++- tests/unit/native-launch-preparation.test.ts | 227 ++++++++++++++++++ tests/unit/native-provisioning.test.ts | 34 ++- vite.launch.config.ts | 16 ++ 26 files changed, 998 insertions(+), 123 deletions(-) create mode 100644 src/lib/server/hosts/delivery.ts create mode 100644 src/lib/server/provisioning/launch-preparation.ts create mode 100644 src/lib/server/provisioning/launch-worker.ts create mode 100644 tests/unit/agent-operation-delivery.test.ts create mode 100644 tests/unit/launch-recipe-summary-migration.test.ts create mode 100644 tests/unit/native-launch-preparation.test.ts create mode 100644 vite.launch.config.ts diff --git a/README.md b/README.md index 06bec90..ae13318 100644 --- a/README.md +++ b/README.md @@ -151,9 +151,10 @@ npm run build ``` Builds regenerate the downloadable agent, standalone updater and checksum manifest, -and bundle the metrics/leaderboard workers under `build/reports`. `npm run dev` -and `npm test` compile those workers automatically. After editing a report -implementation during development, restart `npm run dev` to rebuild its worker. +and bundle the metrics/leaderboard workers under `build/reports` and native launch +preparation under `build/launch`. `npm run dev` and `npm test` compile those workers +automatically. After editing a report or launch-preparation implementation during +development, restart `npm run dev` to rebuild its worker. Reports use separate read-only SQLite connections and bounded revision-aware caches; metrics snapshots can be reused for up to 30 seconds when their inputs are unchanged. diff --git a/docs/deployment-blueprints.md b/docs/deployment-blueprints.md index 48af6af..5845237 100644 --- a/docs/deployment-blueprints.md +++ b/docs/deployment-blueprints.md @@ -167,6 +167,10 @@ This applies to collection collections, single-server launches, and **Launch aga Previews and queued jobs store file metadata without copying script or plugin contents for every lobby. Provisioning captures the bytes when each job starts. +Native launch preparation reads files and encodes command payloads in a worker +thread. The controller rechecks capacity, promotions, credentials, and safety +before submitting those commands. Agent delivery streams the stored payloads in +small chunks, while progress polling reads metadata without loading file bytes. The files selected at provisioning, their hashes and version metadata, composed configuration, selected map, and image reference are recorded in SQLite for diff --git a/package.json b/package.json index 863578d..e60ccd2 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "type": "module", "scripts": { "dev": "vite dev", - "build": "vite build && npm run build:reports -- --outDir build/reports", + "build": "vite build && npm run build:reports -- --outDir build/reports && npm run build:launch -- --outDir build/launch", "preview": "vite preview", "check": "svelte-kit sync && svelte-check --tsconfig ./tsconfig.json", "test": "vitest run", @@ -17,9 +17,10 @@ "prebuild": "npm run agent:package", "test:agent": "python3 -m unittest discover -s agent/tests -v", "build:reports": "vite build --config vite.reports.config.ts", - "predev": "npm run build:reports", - "pretest": "npm run build:reports", - "pretest:watch": "npm run build:reports" + "build:launch": "vite build --config vite.launch.config.ts", + "predev": "npm run build:reports && npm run build:launch", + "pretest": "npm run build:reports && npm run build:launch", + "pretest:watch": "npm run build:reports && npm run build:launch" }, "dependencies": { "@fontsource/alfa-slab-one": "^5.3.0", diff --git a/src/lib/server/backends/native.ts b/src/lib/server/backends/native.ts index a6816d2..399f0e0 100644 --- a/src/lib/server/backends/native.ts +++ b/src/lib/server/backends/native.ts @@ -1,6 +1,6 @@ import { randomUUID } from 'node:crypto'; import { db } from '../db/index.js'; -import { getHost, queueHostOperation, waitHostOperation } from '../hosts/index.js'; +import { getHost, queueHostOperation, waitHostOperationStatus } from '../hosts/index.js'; import { trackActivity } from '../pipeline/activity.js'; import { latestGameSubmission } from '../fleet/lifecycle-evidence.js'; @@ -43,7 +43,7 @@ async function executeRequest(instance: NativeInstance, kind: string, payload instanceId: instance.id, generation: instance.generation, idempotencyKey: key ?? randomUUID() }); const result = await trackActivity(`agent:${operation.id}`, `Waiting for ${kind} on ${instance.server}`, - () => waitHostOperation(operation.id, 25_000)); + () => waitHostOperationStatus(operation.id, 25_000)); if (result.state !== 'completed') throw new Error(result.error || `Agent operation ${result.id} is ${result.state}; its original ID is retained.`); // Do not apply replies from a retired generation to a replacement server. const current = nativeInstance(instance.server); diff --git a/src/lib/server/db/index.ts b/src/lib/server/db/index.ts index e4adb39..663e468 100644 --- a/src/lib/server/db/index.ts +++ b/src/lib/server/db/index.ts @@ -345,6 +345,9 @@ export function db(): Database.Database { { name: 'mode', ddl: "ALTER TABLE server_launch_collections ADD COLUMN mode TEXT NOT NULL DEFAULT 'fill'" }, { name: 'retained_servers', ddl: "ALTER TABLE server_launch_collections ADD COLUMN retained_servers TEXT NOT NULL DEFAULT '[]'" } ]); + migrateTable(handle, 'server_launch_recipes', [ + { name: 'recipe_summary', ddl: 'ALTER TABLE server_launch_recipes ADD COLUMN recipe_summary TEXT' } + ]); migrateServerHistory(handle); migrateTable(handle, 'server_collection_previews', [ { name: 'mode', ddl: "ALTER TABLE server_collection_previews ADD COLUMN mode TEXT NOT NULL DEFAULT 'fill'" }, diff --git a/src/lib/server/db/schema.sql b/src/lib/server/db/schema.sql index cc8c118..27546f3 100644 --- a/src/lib/server/db/schema.sql +++ b/src/lib/server/db/schema.sql @@ -522,6 +522,7 @@ CREATE TABLE IF NOT EXISTS server_launch_previews ( CREATE TABLE IF NOT EXISTS server_launch_recipes ( deployment_id TEXT PRIMARY KEY REFERENCES server_deployments(id) ON DELETE CASCADE, recipe TEXT NOT NULL, + recipe_summary TEXT, blueprint_name TEXT ); diff --git a/src/lib/server/fleet/launch-recipes.ts b/src/lib/server/fleet/launch-recipes.ts index db3f979..32eb5e2 100644 --- a/src/lib/server/fleet/launch-recipes.ts +++ b/src/lib/server/fleet/launch-recipes.ts @@ -109,10 +109,29 @@ function buildRecipe(game: ZombieGame, mapSlug: string, includeContent: boolean) configFile: configFileForGame(game), configSource, files }; } -export function summarizeRecipe(recipe: LaunchRecipePlan): LaunchRecipeSummary { +export function summarizeRecipe(recipe: LaunchRecipePlan, fingerprint = sha256Buffer(Buffer.from(JSON.stringify(recipe)))): LaunchRecipeSummary { const { format: _format, files, ...summary } = recipe; return { ...summary, files: files.map(({ content: _content, ...file }) => file), - fingerprint: sha256Buffer(Buffer.from(JSON.stringify(recipe))) }; + fingerprint }; +} + +/** Keep the small detail response beside the immutable attempted file bytes. */ +export function recipeStorage(recipe: LaunchRecipePlan): { recipe: string; recipeSummary: string } { + const encoded = JSON.stringify(recipe); + return { recipe: encoded, recipeSummary: JSON.stringify(summarizeRecipe(recipe, sha256Buffer(Buffer.from(encoded)))) }; +} + +export function recipeSummaryForDeployment(id: string): { recipe: LaunchRecipeSummary; blueprintName: string | null } | null { + const row = db().prepare('SELECT recipe_summary, blueprint_name FROM server_launch_recipes WHERE deployment_id = ?') + .get(id) as { recipe_summary: string | null; blueprint_name: string | null } | undefined; + if (!row) return null; + if (row.recipe_summary !== null) return { recipe: JSON.parse(row.recipe_summary), blueprintName: row.blueprint_name }; + // Old snapshots remain readable. Populate on first access, without making + // controller startup scan and decode every retained plugin in its history. + const launch = recipeForDeployment(id)!; + const summary = summarizeRecipe(launch.recipe); + db().prepare('UPDATE server_launch_recipes SET recipe_summary = ? WHERE deployment_id = ?').run(JSON.stringify(summary), id); + return { recipe: summary, blueprintName: row.blueprint_name }; } export function recipeForDeployment(id: string): { recipe: LaunchRecipePlan; blueprintName: string | null } | null { diff --git a/src/lib/server/hosts/delivery.ts b/src/lib/server/hosts/delivery.ts new file mode 100644 index 0000000..ed37c73 --- /dev/null +++ b/src/lib/server/hosts/delivery.ts @@ -0,0 +1,62 @@ +import { setImmediate } from 'node:timers/promises'; +import { getHostOperationPayload, type OperationDeliveryRow } from './index.js'; + +// Even non-ASCII content stays below 64 KiB per encoded chunk. +const CHUNK_CHARACTERS = 16 * 1024; +const CHUNK_BYTES = 64 * 1024; +const encoder = new TextEncoder(); + +function* jsonParts(operations: OperationDeliveryRow[], serverTime: string): Generator { + yield '{"operations":['; + for (const [index, operation] of operations.entries()) { + if (index) yield ','; + const metadata = JSON.stringify({ + id: operation.id, hostId: operation.host_id, kind: operation.kind, + instanceId: operation.instance_id, generation: operation.generation, state: operation.state, + error: operation.error, createdAt: operation.created_at, updatedAt: operation.updated_at + }); + yield metadata.slice(0, -1) + ',"payload":'; + // Load only the immutable payload currently being sent, after the + // stream has yielded its metadata and had a chance to serve health work. + yield getHostOperationPayload(operation.host_id, operation.id); + yield ',"result":'; + yield operation.result ?? 'null'; + yield '}'; + } + yield '],"serverTime":' + JSON.stringify(serverTime) + '}'; +} + +function* encodedChunks(operations: OperationDeliveryRow[], serverTime: string): Generator { + for (const part of jsonParts(operations, serverTime)) { + if (Buffer.isBuffer(part)) { + for (let start = 0; start < part.length; start += CHUNK_BYTES) yield part.subarray(start, start + CHUNK_BYTES); + continue; + } + for (let start = 0; start < part.length;) { + let end = Math.min(start + CHUNK_CHARACTERS, part.length); + // A UTF-16 surrogate pair must be encoded together, including when + // user-authored configuration text falls on a chunk boundary. + const last = part.charCodeAt(end - 1); + if (end < part.length && last >= 0xd800 && last <= 0xdbff) end--; + yield encoder.encode(part.slice(start, end)); + start = end; + } + } +} + +/** Bound encoding work and yield between chunks so health requests keep flowing. */ +export function hostOperationsResponse(operations: OperationDeliveryRow[], serverTime = new Date().toISOString()): Response { + const chunks = encodedChunks(operations, serverTime); + let cancelled = false; + const stream = new ReadableStream({ + async pull(controller) { + await setImmediate(); + if (cancelled) return; + const next = chunks.next(); + if (next.done) controller.close(); + else controller.enqueue(next.value); + }, + cancel() { cancelled = true; chunks.return(undefined); } + }); + return new Response(stream, { headers: { 'Content-Type': 'application/json; charset=utf-8', 'Cache-Control': 'no-store' } }); +} diff --git a/src/lib/server/hosts/index.ts b/src/lib/server/hosts/index.ts index 2bc4e69..48a4658 100644 --- a/src/lib/server/hosts/index.ts +++ b/src/lib/server/hosts/index.ts @@ -30,7 +30,7 @@ function decommissionRow(hostId: string): DecommissionRow | undefined { } function presentDecommission(row: DecommissionRow | undefined): HostDecommission | null { return row ? { operationId: row.operation_id, stage: row.stage, requestedAt: row.requested_at, updatedAt: row.updated_at, - error: row.error, retryable: row.stage === 'failed' && getHostOperation(row.operation_id)?.state === 'failed' } : null; + error: row.error, retryable: row.stage === 'failed' && getHostOperationStatus(row.operation_id)?.state === 'failed' } : null; } function hostFromRow(row: HostRow): Host { const inventory: HostInventory = JSON.parse(row.inventory); @@ -137,7 +137,7 @@ export function receiveDecommissionReceipt(operationId: string, token: string, i if (!row || !timingSafeEqual(Buffer.from(row.token_hash, 'hex'), Buffer.from(digest(token), 'hex'))) throw new HostDeletionError('A valid cleanup receipt credential is required.', 401); if (!['cleaned', 'completed', 'failed'].includes(input.stage as string)) throw new HostDeletionError('Invalid cleanup receipt stage.', 400); if (row.stage === 'completed') return { acknowledged: true as const }; - const operation = getHostOperation(operationId)!; + const operation = getHostOperationStatus(operationId)!; if (!['accepted','running','unknown'].includes(operation.state)) throw new HostDeletionError('Cleanup has not been accepted by the host.', 409); const timestamp = now(); if (input.stage === 'failed') { @@ -314,41 +314,82 @@ export function requireReadyHost(hostId: string, options: { libraryId?: string; return host; } -type OperationRow = { id: string; host_id: string; kind: string; instance_id: string | null; generation: number | null; payload: string; state: HostOperationState; result: string | null; error: string | null; created_at: string; updated_at: string }; -function operationFromRow(row: OperationRow): HostOperation { return { id: row.id, hostId: row.host_id, kind: row.kind, instanceId: row.instance_id, generation: row.generation, payload: JSON.parse(row.payload), state: row.state, result: row.result === null ? null : JSON.parse(row.result), error: row.error, createdAt: row.created_at, updatedAt: row.updated_at }; } -export function getHostOperation(id: string): HostOperation | null { const row = database().prepare('SELECT * FROM ec_host_operations WHERE id=?').get(id) as OperationRow | undefined; return row ? operationFromRow(row) : null; } -export function listHostOperations(hostId: string, limit = 100, summary = false): HostOperation[] { return (database().prepare(`SELECT ${summary ? `id,host_id,kind,instance_id,generation,state,error,created_at,updated_at,'{}' AS payload,NULL AS result` : '*'} FROM ec_host_operations WHERE host_id=? ORDER BY created_at DESC,rowid DESC LIMIT ?`).all(hostId, Math.min(500, Math.max(1, limit))) as OperationRow[]).map(operationFromRow); } +export type OperationRow = { id: string; host_id: string; kind: string; instance_id: string | null; generation: number | null; payload: string; state: HostOperationState; result: string | null; error: string | null; created_at: string; updated_at: string }; +export type OperationDeliveryRow = Omit; +export type HostOperationStatus = Omit; +const operationStatusColumns = 'o.id,o.host_id,o.kind,o.instance_id,o.generation,o.state,o.result,o.error,o.created_at,o.updated_at'; +const operationPayloadSql = 'COALESCE((SELECT p.payload FROM ec_host_operation_payloads p WHERE p.operation_id=o.id),o.payload)'; +function operationStatusFromRow(row: Omit): HostOperationStatus { return { id: row.id, hostId: row.host_id, kind: row.kind, instanceId: row.instance_id, generation: row.generation, state: row.state, result: row.result === null ? null : JSON.parse(row.result), error: row.error, createdAt: row.created_at, updatedAt: row.updated_at }; } +function operationFromRow(row: OperationRow): HostOperation { return { ...operationStatusFromRow(row), payload: JSON.parse(row.payload) }; } +export function getHostOperation(id: string): HostOperation | null { + const row = database().prepare(`SELECT ${operationStatusColumns},${operationPayloadSql} AS payload FROM ec_host_operations o WHERE o.id=?`) + .get(id) as OperationRow | undefined; + return row ? operationFromRow(row) : null; +} +/** Polling and receipts need the outcome, never another copy of launch files. */ +export function getHostOperationStatus(id: string): HostOperationStatus | null { + const row = database().prepare('SELECT id,host_id,kind,instance_id,generation,state,result,error,created_at,updated_at FROM ec_host_operations WHERE id=?') + .get(id) as Omit | undefined; + return row ? operationStatusFromRow(row) : null; +} +export function listHostOperations(hostId: string, limit = 100, summary = false): HostOperation[] { + const columns = summary ? `o.id,o.host_id,o.kind,o.instance_id,o.generation,o.state,o.error,o.created_at,o.updated_at,'{}' AS payload,NULL AS result` + : `${operationStatusColumns},${operationPayloadSql} AS payload`; + return (database().prepare(`SELECT ${columns} FROM ec_host_operations o WHERE o.host_id=? ORDER BY o.created_at DESC,o.rowid DESC LIMIT ?`) + .all(hostId, Math.min(500, Math.max(1, limit))) as OperationRow[]).map(operationFromRow); +} /** Heartbeats can tell an executing create to stop 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); } -export function queueHostOperation(hostId: string, kind: string, payload: Record, options: { instanceId?: string; generation?: number; idempotencyKey?: string; id?: string } = {}): HostOperation { +type QueueHostOperationOptions = { instanceId?: string; generation?: number; idempotencyKey?: string; id?: string }; +export function queueHostOperation(hostId: string, kind: string, payload: Record, options: QueueHostOperationOptions = {}): HostOperation { + if (!payload || typeof payload !== 'object' || Array.isArray(payload)) throw new Error('Operation payload must be an object.'); + const operation = enqueueHostOperation(hostId, kind, JSON.stringify(payload), options); + return getHostOperation(operation.id)!; +} +/** Keep immutable launch bytes separate so receipts and delivery do not rewrite large payloads. */ +export function queueHostOperationSerialized(hostId: string, kind: string, payload: Buffer, options: QueueHostOperationOptions = {}, routingPayload: Record = {}): HostOperationStatus { + if (!Buffer.isBuffer(payload) || payload[0] !== 123 || payload[payload.length - 1] !== 125) throw new Error('Serialized operation payload must be a JSON object.'); + if (!routingPayload || typeof routingPayload !== 'object' || Array.isArray(routingPayload)) throw new Error('Operation routing payload must be an object.'); + return enqueueHostOperation(hostId, kind, payload, options, JSON.stringify(routingPayload)); +} +function enqueueHostOperation(hostId: string, kind: string, payload: string | Buffer, options: QueueHostOperationOptions, routingPayload?: string): HostOperationStatus { const host = requireHost(hostId); if (host.decommission) throw new Error('Host cleanup is in progress; new operations are blocked.'); if (kind === 'host.decommission') throw new Error('Use the host deletion workflow to request cleanup.'); if (!/^[a-z][a-z0-9_.-]{0,79}$/.test(kind)) throw new Error('Invalid operation kind.'); - if (!payload || typeof payload !== 'object' || Array.isArray(payload)) throw new Error('Operation payload must be an object.'); if (options.generation !== undefined && (!Number.isSafeInteger(options.generation) || options.generation < 1)) throw new Error('Instance generation must be a positive integer.'); const operation = database().transaction(() => { if (options.idempotencyKey) { - const existing = database().prepare('SELECT * FROM ec_host_operations WHERE host_id=? AND idempotency_key=?').get(hostId, options.idempotencyKey) as OperationRow | undefined; + const existing = database().prepare(`SELECT o.id,o.kind,o.instance_id,o.generation,${operationPayloadSql}=CAST(? AS TEXT) AS same_payload + FROM ec_host_operations o WHERE o.host_id=? AND o.idempotency_key=?`).get(payload, hostId, options.idempotencyKey) as + { id: string; kind: string; instance_id: string | null; generation: number | null; same_payload: number } | undefined; if (existing) { - if (existing.kind !== kind || existing.instance_id !== (options.instanceId ?? null) || existing.generation !== (options.generation ?? null) || existing.payload !== JSON.stringify(payload)) throw new Error('Operation idempotency key is already associated with a different request.'); - return operationFromRow(existing); + if (existing.kind !== kind || existing.instance_id !== (options.instanceId ?? null) || existing.generation !== (options.generation ?? null) || !existing.same_payload) throw new Error('Operation idempotency key is already associated with a different request.'); + return getHostOperationStatus(existing.id)!; } } if (['instance.create','create'].includes(kind)) requireReadyHost(hostId); if (host.mode === 'disabled' && kind !== 'instance.cancel-create') throw new Error('Host is disabled. Enable it before requesting operations.'); const id = options.id ?? randomUUID(); const timestamp = now(); - database().prepare('INSERT INTO ec_host_operations(id,host_id,kind,instance_id,generation,idempotency_key,payload,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?)').run(id, hostId, kind, options.instanceId ?? null, options.generation ?? null, options.idempotencyKey ?? null, JSON.stringify(payload), timestamp, timestamp); + database().prepare('INSERT INTO ec_host_operations(id,host_id,kind,instance_id,generation,idempotency_key,payload,created_at,updated_at) VALUES(?,?,?,?,?,?,CAST(? AS TEXT),?,?)').run(id, hostId, kind, options.instanceId ?? null, options.generation ?? null, options.idempotencyKey ?? null, routingPayload ?? payload, timestamp, timestamp); + if (routingPayload !== undefined) database().prepare('INSERT INTO ec_host_operation_payloads(operation_id,payload) VALUES(?,CAST(? AS TEXT))').run(id, payload); database().prepare('INSERT INTO ec_host_operation_events(operation_id,state,created_at) VALUES(?,?,?)').run(id, 'pending', timestamp); - return getHostOperation(id)!; + return getHostOperationStatus(id)!; })(); queueMicrotask(() => workEvents.emit(hostId)); return operation; } export function pendingHostOperations(hostId: string, lane: 'all' | 'read' | 'health' | 'write' = 'all'): HostOperation[] { + return pendingHostOperationRows(hostId, lane).map(operationFromRow); +} +type OperationLane = 'all' | 'read' | 'health' | 'write'; +export function pendingHostOperationRows(hostId: string, lane?: OperationLane, includePayload?: true): OperationRow[]; +export function pendingHostOperationRows(hostId: string, lane: OperationLane, includePayload: false): OperationDeliveryRow[]; +/** Delivery snapshots retain status/results while large immutable payloads stay in SQLite. */ +export function pendingHostOperationRows(hostId: string, lane: OperationLane = 'all', includePayload = true): OperationDeliveryRow[] { const host = requireHost(hostId); if (host.mode === 'disabled' || !host.connected) return []; // Older agents hold activity.lock across the entire delivered batch, including @@ -361,8 +402,8 @@ export function pendingHostOperations(hostId: string, lane: 'all' | 'read' | 'he // Accepted/running/unknown commands keep the same immutable ID. The agent // consults its durable journal before acknowledging or executing a replay. const readOnly = `(kind IN ('host.inspect','instance.inspect','files.read','files.list','files.stat','logs.read') - OR (kind='rcon.execute' AND json_extract(payload,'$.command') IN ('status','getstatus','version')))`; - const healthOnly = `(kind='instance.inspect' OR (kind='rcon.execute' AND json_extract(payload,'$.command') IN ('status','getstatus','version')))`; + OR (kind='rcon.execute' AND json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version')))`; + const healthOnly = `(kind='instance.inspect' OR (kind='rcon.execute' AND json_extract(${operationPayloadSql},'$.command') IN ('status','getstatus','version')))`; const healthLane = host.inventory.capabilities?.includes('operations.health-lane'); const laneFilter = lane === 'all' ? '' : lane === 'health' ? ` AND ${healthOnly}` : ` AND ${lane === 'write' ? 'NOT ' : ''}${readOnly}${lane === 'read' && healthLane ? ` AND NOT ${healthOnly}` : ''}`; @@ -370,23 +411,31 @@ export function pendingHostOperations(hostId: string, lane: 'all' | 'read' | 'he // in small batches to amortize delivery without handing out a long backlog. const limit = healthLane ? lane === 'health' ? 1 : lane === 'read' ? 4 : 25 : 25; return database().transaction(() => { - const rows = database().prepare(`SELECT * FROM ec_host_operations o WHERE host_id=? AND state IN ('pending','accepted','running','unknown') + const columns = includePayload ? `${operationStatusColumns},${operationPayloadSql} AS payload` : operationStatusColumns; + const rows = database().prepare(`SELECT ${columns} FROM ec_host_operations o WHERE host_id=? AND state IN ('pending','accepted','running','unknown') ${laneFilter} AND (? OR state != 'pending' OR kind NOT IN ('instance.create','create')) 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) ORDER BY CASE WHEN kind='instance.cancel-create' THEN 0 ELSE 1 END,COALESCE(last_delivered_at,''),created_at,rowid LIMIT ${limit}`) - .all(hostId, Number(host.readiness === 'ready'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false)) as OperationRow[]; + .all(hostId, Number(host.readiness === 'ready'), Number(host.inventory.capabilities?.includes('instance.cancel-create') ?? false)) as OperationDeliveryRow[]; const deliveredAt = now(); for (const operation of rows) database().prepare('UPDATE ec_host_operations SET last_delivered_at=? WHERE id=?').run(deliveredAt, operation.id); - return rows.map(operationFromRow); + return rows; })(); } -export function reportHostOperation(hostId: string, id: string, input: { state: unknown; result?: unknown; error?: unknown }): HostOperation { +/** Payloads never change after enqueueing; fetch one delivery's UTF-8 bytes without decoding. */ +export function getHostOperationPayload(hostId: string, id: string): Buffer { + const row = database().prepare(`SELECT CAST(${operationPayloadSql} AS BLOB) AS payload FROM ec_host_operations o WHERE o.id=? AND o.host_id=?`) + .get(id, hostId) as { payload: Buffer } | undefined; + if (!row) throw new Error('Host operation was not found for this host.'); + return row.payload; +} +export function reportHostOperationStatus(hostId: string, id: string, input: { state: unknown; result?: unknown; error?: unknown }): HostOperationStatus { const operation = database().transaction(() => { requireHost(hostId); - const operation = getHostOperation(id); + const operation = getHostOperationStatus(id); if (!operation || operation.hostId !== hostId) throw new Error('Operation was not found for this host.'); if (operation.kind === 'host.decommission' && input.state === 'completed') throw new Error('Host removal requires the cleanup worker completion receipt.'); if (!['accepted','running','completed','failed','unknown'].includes(input.state as string)) throw new Error('Invalid operation state.'); @@ -407,21 +456,29 @@ export function reportHostOperation(hostId: string, id: string, input: { state: recordAgentReceipt(id, state, now()); } } - return getHostOperation(id)!; + return getHostOperationStatus(id)!; })(); queueMicrotask(() => workEvents.emit(`operation:${id}`)); return operation; } -export async function waitHostOperation(id: string, timeoutMs = 30_000): Promise { +export function reportHostOperation(hostId: string, id: string, input: { state: unknown; result?: unknown; error?: unknown }): HostOperation { + reportHostOperationStatus(hostId, id, input); + return getHostOperation(id)!; +} +export async function waitHostOperationStatus(id: string, timeoutMs = 30_000): Promise { const deadline = Date.now() + timeoutMs; while (true) { - const operation = getHostOperation(id); + const operation = getHostOperationStatus(id); if (!operation) throw new Error('Host operation was not found.'); if (['completed','failed','unknown'].includes(operation.state)) return operation; if (Date.now() >= deadline) throw new Error(`Host operation ${id} is awaiting acknowledgement; reconcile this operation ID before retrying.`); await waitForHostWork(`operation:${id}`, new AbortController().signal, Math.min(1000, Math.max(1, deadline - Date.now()))); } } +export async function waitHostOperation(id: string, timeoutMs = 30_000): Promise { + await waitHostOperationStatus(id, timeoutMs); + return getHostOperation(id)!; +} export function hostPortAvailable(hostId: string, candidate: number, bindAddress?: string, protocol: 'udp' | 'tcp' = 'udp', instanceId?: string): boolean { const host = requireHost(hostId); port(candidate); const bind = bindAddress ?? host.bindAddress; diff --git a/src/lib/server/hosts/schema.ts b/src/lib/server/hosts/schema.ts index cb12041..e6d311f 100644 --- a/src/lib/server/hosts/schema.ts +++ b/src/lib/server/hosts/schema.ts @@ -33,6 +33,10 @@ export function ensureHostSchema(handle: Database.Database): void { -- history. The queue index cannot narrow those lookups beyond host_id, -- causing a history scan for every queued command on Node's main thread. CREATE INDEX IF NOT EXISTS ec_host_operations_kind ON ec_host_operations(host_id, kind, created_at); + CREATE TABLE IF NOT EXISTS ec_host_operation_payloads ( + operation_id TEXT PRIMARY KEY REFERENCES ec_host_operations(id) ON DELETE CASCADE, + payload TEXT NOT NULL + ); CREATE TABLE IF NOT EXISTS ec_host_operation_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, operation_id TEXT NOT NULL REFERENCES ec_host_operations(id), state TEXT NOT NULL, created_at TEXT NOT NULL, error TEXT diff --git a/src/lib/server/provisioning/history.ts b/src/lib/server/provisioning/history.ts index 65a4f80..d3ae310 100644 --- a/src/lib/server/provisioning/history.ts +++ b/src/lib/server/provisioning/history.ts @@ -1,6 +1,6 @@ import { randomUUID } from 'node:crypto'; import { db } from '../db/index.js'; -import { recipeForDeployment, summarizeRecipe } from '../fleet/launch-recipes.js'; +import { recipeSummaryForDeployment } from '../fleet/launch-recipes.js'; import { actionLabels, type ProvisioningAction } from '../../provisioning.js'; import type { ProvisioningCapacity, ProvisioningDeployment, ProvisioningDetail, ProvisioningSource } from '../../provisioning-detail.js'; import { fleetSettings } from '../settings.js'; @@ -111,7 +111,7 @@ export function getDeployment(id: string, options: { events?: 'all' | 'recent' } const { observation_json: _snapshot, phase_key: _phase, phase_started_at: _phaseStarted, observed_at: _observed, ...deployment } = row; const sources = (db().prepare('SELECT kind, source_id, server FROM provisioning_sources WHERE operation_id = ? ORDER BY kind, source_id') .all(id) as Omit[]).map(sourceDetails); - const launch = recipeForDeployment(id); + const launch = recipeSummaryForDeployment(id); const collection = db().prepare(`SELECT c.id, c.name FROM ( SELECT collection_id, deployment_id FROM server_collection_members UNION ALL SELECT collection_id, deployment_id FROM server_collection_removals) m @@ -120,7 +120,7 @@ export function getDeployment(id: string, options: { events?: 'all' | 'recent' } canCancelCreation: canCancelCreation(id), cancellation: cancellation ?? null, workerActivity: workerActivity(), events: deploymentEvents(id, options.events === 'recent'), activity: { ...activitySummary(id), failureExcerpt: latestFailureEvent(id) }, sources, - recipe: launch ? summarizeRecipe(launch.recipe) : null, blueprintName: launch?.blueprintName ?? null, + recipe: launch?.recipe ?? null, blueprintName: launch?.blueprintName ?? null, canResumeVerification: canResumeVerification(id), canResumeCleanup: canResumeCleanup(id), canResumeRebuild: canResumeRebuild(id), canCancelRemoval: canCancelRemoval(id), collection: collection ?? null, progress: readProgress(row), capacity: capacityFor(deployment, generatedAt), diff --git a/src/lib/server/provisioning/instances.ts b/src/lib/server/provisioning/instances.ts index fdc0574..168c5e3 100644 --- a/src/lib/server/provisioning/instances.ts +++ b/src/lib/server/provisioning/instances.ts @@ -4,7 +4,7 @@ import { probePlayerEvidence, probeServerStatus } from '../fleet/status-evidence import { beginDeployment, deploymentStage, finishDeployment, recordDeploymentEvent, REMOVING_SERVER, resumeCleanup, resumeVerification, claimRemoval } from './history.js'; import { randomUUID } from 'node:crypto'; import type { LaunchPreview } from '$lib/deployment-plan.js'; -import { blueprintRecipe, captureRecipe, captureRecipePlan, recipeForDeployment, summarizeRecipe, type LaunchRecipe, type LaunchRecipePlan } from '../fleet/launch-recipes.js'; +import { blueprintRecipe, captureRecipe, captureRecipePlan, recipeForDeployment, recipeStorage, summarizeRecipe, type LaunchRecipe, type LaunchRecipePlan } from '../fleet/launch-recipes.js'; import { mkdir, chmod, rm, writeFile } from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { config, destOfKind, serverDir, gameOfServer, configFileForGame } from '../config.js'; @@ -240,9 +240,10 @@ export function queueCreateServer(input: CreateServerInput, releasingServers: st if (!prepared.ok) return prepared; // A preview records choices, not version pins. Promotions remain authoritative. const currentRecipe = captureRecipePlan(input.game ?? 't5', prepared.mapEntry.slug); + const storedRecipe = recipeStorage(currentRecipe); const id = beginDeployment(prepared.name, 'create', input.createdBy); - db().prepare('INSERT INTO server_launch_recipes (deployment_id, recipe, blueprint_name) VALUES (?, ?, ?)') - .run(id, JSON.stringify(currentRecipe), preview?.blueprint_name ?? null); + db().prepare('INSERT INTO server_launch_recipes (deployment_id, recipe, recipe_summary, blueprint_name) VALUES (?, ?, ?, ?)') + .run(id, storedRecipe.recipe, storedRecipe.recipeSummary, preview?.blueprint_name ?? null); if (preview) db().prepare('UPDATE server_launch_previews SET deployment_id = ? WHERE id = ?').run(id, preview.id); db().prepare("UPDATE server_deployments SET state = 'queued' WHERE id = ?").run(id); db().prepare('INSERT INTO server_creation_jobs (deployment_id, server, port, game_mode, map_slug, expected_map) VALUES (?, ?, ?, ?, ?, ?)') @@ -275,7 +276,9 @@ export function queueRebuildServer(name: string, triggeredBy = 'ui'): CreateServ db().prepare('INSERT INTO server_creation_jobs (deployment_id, server, port, game_mode, map_slug, expected_map) VALUES (?, ?, ?, ?, ?, ?)') .run(id, name, server.game_port, 'zombie', map.slug, map.canonical); db().prepare('INSERT INTO server_rebuild_jobs (deployment_id) VALUES (?)').run(id); - db().prepare('INSERT INTO server_launch_recipes (deployment_id, recipe) VALUES (?, ?)').run(id, JSON.stringify(recipe)); + const storedRecipe = recipeStorage(recipe); + db().prepare('INSERT INTO server_launch_recipes (deployment_id, recipe, recipe_summary) VALUES (?, ?, ?)') + .run(id, storedRecipe.recipe, storedRecipe.recipeSummary); deploymentStage(id, 'Waiting to rebuild the empty server', { phase: 'eligibility', recipeStatus: 'queued', observedAt: null, blocker: { code: 'eligibility', message: 'Waiting for safe rebuild eligibility checks.', automatic: true } }); return { ok: true as const, name, port: server.game_port!, deploymentId: id }; @@ -399,10 +402,11 @@ async function createServerUnlocked(input: CreateServerInput, prepared: Prepared // Resolve once more at provisioning so promotions made while queued also apply. // Persist the exact attempted files for history, never as input to future launches. const recipe = captureRecipe(game, mapEntry.slug); + const storedRecipe = recipeStorage(recipe); db().transaction(() => { - db().prepare(`INSERT INTO server_launch_recipes (deployment_id, recipe) VALUES (?, ?) - ON CONFLICT(deployment_id) DO UPDATE SET recipe = excluded.recipe`) - .run(operation, JSON.stringify(recipe)); + db().prepare(`INSERT INTO server_launch_recipes (deployment_id, recipe, recipe_summary) VALUES (?, ?, ?) + ON CONFLICT(deployment_id) DO UPDATE SET recipe = excluded.recipe, recipe_summary = excluded.recipe_summary`) + .run(operation, storedRecipe.recipe, storedRecipe.recipeSummary); deploymentStage(operation, 'Preparing server files', { phase: 'prepare', recipeStatus: 'attempted', blocker: null, checks: [], checkpoint: { key: 'startup-queue', label: 'Admitted to startup' }, event: { kind: 'recipe', message: 'Promoted files selected for this creation attempt.', details: { capturedAt: recipe.capturedAt } } }); diff --git a/src/lib/server/provisioning/launch-preparation.ts b/src/lib/server/provisioning/launch-preparation.ts new file mode 100644 index 0000000..e18f140 --- /dev/null +++ b/src/lib/server/provisioning/launch-preparation.ts @@ -0,0 +1,71 @@ +import { Worker } from 'node:worker_threads'; +import { resolve } from 'node:path'; +import { config } from '../config.js'; +import { db } from '../db/index.js'; +import type { NativeInstance } from '../backends/native.js'; +import type { LaunchRecipePlan } from '../fleet/launch-recipes.js'; + +export interface NativeLaunchInput { + game: 't5' | 't6'; + mapSlug: string; + native?: LaunchRecipePlan['native']; + instance: NativeInstance; + serverKey: string; + rconPassword: string; + rebuild: boolean; +} + +export interface PreparedNativeLaunch { + recipe: LaunchRecipePlan; + recipeSummary: string; + recipeJson: Uint8Array; + payloadJson: Uint8Array; +} + +let worker: Worker | null = null; +let sequence = 0; +const pending = new Map void; + reject: (error: Error) => void; + timer: NodeJS.Timeout; +}>(); + +function stop(error: Error, current = worker): Promise | undefined { + if (current !== worker) return; + worker = null; + for (const request of pending.values()) { clearTimeout(request.timer); request.reject(error); } + pending.clear(); + return current?.terminate(); +} + +/** The reader and JSON encoder run off the controller's request thread. */ +export function prepareNativeLaunch(input: NativeLaunchInput): Promise { + db(); // Migrations must finish before the independent read-only connection opens. + if (!worker) { + const current = worker = new Worker(resolve(process.env.NODE_ENV === 'production' + ? 'build/launch/worker.mjs' : '.svelte-kit/launch/worker.mjs'), { + workerData: { dbPath: resolve(config.dbPath) } + }); + current.on('message', ({ id, value, error }: { id: number; value: PreparedNativeLaunch; error?: string }) => { + if (current !== worker) return; + const request = pending.get(id); + if (!request) return; + pending.delete(id); clearTimeout(request.timer); + if (error) request.reject(new Error(error)); else request.resolve(value); + if (!pending.size) current.unref(); + }); + current.on('error', error => { void stop(error, current); }); + current.on('exit', () => { void stop(new Error('Launch preparation worker stopped; preparation will retry.'), current); }); + } + worker.ref(); + const id = ++sequence; + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { void stop(new Error('Launch preparation timed out; preparation will retry.')); }, 60_000); + pending.set(id, { resolve, reject, timer }); + worker!.postMessage({ id, input }); + }); +} + +export async function closeLaunchPreparation(): Promise { + await stop(new Error('Launch preparation stopped.')); +} diff --git a/src/lib/server/provisioning/launch-worker.ts b/src/lib/server/provisioning/launch-worker.ts new file mode 100644 index 0000000..083bf7e --- /dev/null +++ b/src/lib/server/provisioning/launch-worker.ts @@ -0,0 +1,38 @@ +import { parentPort } from 'node:worker_threads'; +import { posix } from 'node:path'; +import { db } from '../db/index.js'; +import { relDestForKind } from '../config.js'; +import { captureRecipe, summarizeRecipe } from '../fleet/launch-recipes.js'; +import { sha256Buffer } from '../fleet/files.js'; +import type { NativeLaunchInput } from './launch-preparation.js'; + +parentPort!.on('message', ({ id, input }: { id: number; input: NativeLaunchInput }) => { + try { + // All file versions and composed configuration belong to one snapshot. + const recipe = db().transaction(() => ({ ...captureRecipe(input.game, input.mapSlug), + native: input.native, image: input.instance.image }))(); + const instance = input.instance; + const payload = { + name: instance.server, game: recipe.game, libraryId: instance.library_id, image: instance.image, + port: instance.port, bindAddress: instance.bind_address, rconPassword: input.rconPassword, + env: { PLUTAINER_SERVER_NAME: instance.server, PLUTAINER_PORT: String(instance.port), + PLUTAINER_GAME: recipe.game === 't6' ? 't6zm' : 't5sp', PLUTAINER_CONFIG_FILE: recipe.configFile, + PLUTAINER_MAP_ROTATE: 'false', + PLUTAINER_EXTRA_ARGS: `+exec electric_cherry_map.cfg +set tws_server_id ${instance.server} +map_rotate`, + PLUTO_SERVER_KEY: input.serverKey }, + files: recipe.files.map(file => ({ path: posix.join(relDestForKind(file.kind, recipe.game), file.path), + contentBase64: file.content, sha256: file.sha256 })), + ...(input.rebuild ? { nextGeneration: instance.generation + 1 } : {}) + }; + const encode = (value: unknown) => new TextEncoder().encode(JSON.stringify(value)); + const recipeJson = encode(recipe); + const payloadJson = encode(payload); + const plan = { ...recipe, files: recipe.files.map(({ content: _content, ...file }) => file) }; + const recipeSummary = JSON.stringify(summarizeRecipe(plan, sha256Buffer(Buffer.from(recipeJson.buffer)))); + // Transfer ownership of the encoded bytes; do not clone giant strings or objects. + parentPort!.postMessage({ id, value: { recipe: plan, recipeSummary, recipeJson, payloadJson } }, + [recipeJson.buffer, payloadJson.buffer]); + } catch (error) { + parentPort!.postMessage({ id, error: error instanceof Error ? error.message : 'Launch preparation failed.' }); + } +}); diff --git a/src/lib/server/provisioning/native.ts b/src/lib/server/provisioning/native.ts index d9b85b1..05e3b83 100644 --- a/src/lib/server/provisioning/native.ts +++ b/src/lib/server/provisioning/native.ts @@ -1,13 +1,15 @@ import { isIP } from 'node:net'; +import { setImmediate as yieldToController } from 'node:timers/promises'; import { CapacityError, choosePlacement, availableHostPort, type PlacementOptions } from '../hosts/capacity.js'; import { randomBytes, randomUUID } from 'node:crypto'; import { posix } from 'node:path'; import { db } from '../db/index.js'; import { config, relDestForKind, serverDir } from '../config.js'; import { nativeInstance, nativeRequest, type NativeInstance } from '../backends/native.js'; -import { getHost, getHostOperation, listHosts, queueHostOperation, releaseHostPorts, reserveHostPort } from '../hosts/index.js'; +import { getHost, getHostOperationStatus, listHosts, queueHostOperation, queueHostOperationSerialized, releaseHostPorts, reserveHostPort } from '../hosts/index.js'; import { newestReadyLibraryId, assertReadyLibrary } from '../game-files/index.js'; -import { blueprintRecipe, captureRecipe, captureRecipePlan, recipeForDeployment, summarizeRecipe, type LaunchRecipe, type LaunchRecipePlan } from '../fleet/launch-recipes.js'; +import { blueprintRecipe, captureRecipePlan, recipeForDeployment, recipeStorage, summarizeRecipe, type LaunchRecipePlan } from '../fleet/launch-recipes.js'; +import { prepareNativeLaunch, type PreparedNativeLaunch } from './launch-preparation.js'; import { getServerKey, saveServerKey } from '../fleet/server-keys.js'; import { activeBetaRemovalBlocker, getServer } from '../fleet/servers.js'; import { nextLobbyNumber, type CreateServerInput, type CreateServerResult } from './instances.js'; @@ -125,7 +127,9 @@ export function queueNativeCreation(input: CreateServerInput): CreateServerResul db().prepare("UPDATE server_deployments SET state='queued' WHERE id=?").run(id); db().prepare('INSERT INTO server_creation_jobs(deployment_id,server,port,game_mode,map_slug,expected_map) VALUES(?,?,?,?,?,?)') .run(id, name, port, 'zombie', recipe.mapSlug, recipe.expectedMap); - db().prepare('INSERT INTO server_launch_recipes(deployment_id,recipe,blueprint_name) VALUES(?,?,?)').run(id, JSON.stringify(recipe), preview?.blueprint_name ?? null); + const stored = recipeStorage(recipe); + db().prepare('INSERT INTO server_launch_recipes(deployment_id,recipe,recipe_summary,blueprint_name) VALUES(?,?,?,?)') + .run(id, stored.recipe, stored.recipeSummary, preview?.blueprint_name ?? null); if (preview) db().prepare('UPDATE server_launch_previews SET deployment_id=?,port=? WHERE id=?').run(id, port, preview.id); deploymentStage(id, 'Waiting for provisioning evaluation', { phase: 'eligibility', recipeStatus: 'queued' }); return { ok: true as const, name, port, deploymentId: id }; @@ -164,7 +168,8 @@ export function queueNativeRebuild(name: string, triggeredBy: string): CreateSer db().prepare("UPDATE server_deployments SET state='waiting' WHERE id=?").run(id); db().prepare('INSERT INTO server_creation_jobs(deployment_id,server,port,game_mode,map_slug,expected_map) VALUES(?,?,?,?,?,?)') .run(id, name, instance.port, 'zombie', recipe.mapSlug, recipe.expectedMap); - db().prepare('INSERT INTO server_launch_recipes(deployment_id,recipe) VALUES(?,?)').run(id, JSON.stringify(recipe)); + const stored = recipeStorage(recipe); + db().prepare('INSERT INTO server_launch_recipes(deployment_id,recipe,recipe_summary) VALUES(?,?,?)').run(id, stored.recipe, stored.recipeSummary); deploymentStage(id, 'Waiting to rebuild when empty', { phase: 'eligibility', recipeStatus: 'queued' }); return { ok: true as const, name, port: instance.port, deploymentId: id }; })(); @@ -212,16 +217,6 @@ function applyReceipt(job: Job, result: Record) { })(); } -function createPayload(instance: NativeInstance, recipe: LaunchRecipe) { - const key = getServerKey(instance.server); - if (!key) throw new Error('The saved server key was removed.'); - return { name: instance.server, game: recipe.game, libraryId: instance.library_id, image: instance.image, port: instance.port, bindAddress: instance.bind_address, - rconPassword: config.rcon.password || randomBytes(32).toString('hex'), - env: { PLUTAINER_SERVER_NAME: instance.server, PLUTAINER_PORT: String(instance.port), PLUTAINER_GAME: recipe.game === 't6' ? 't6zm' : 't5sp', - PLUTAINER_CONFIG_FILE: recipe.configFile, PLUTAINER_MAP_ROTATE: 'false', PLUTAINER_EXTRA_ARGS: `+exec electric_cherry_map.cfg +set tws_server_id ${instance.server} +map_rotate`, PLUTO_SERVER_KEY: key }, - files: recipe.files.map(file => ({ path: posix.join(relDestForKind(file.kind, recipe.game), file.path), contentBase64: file.content, sha256: file.sha256 })) }; -} - function hold(job: Job, message: string) { deploymentStage(job.id, message, { blocker: { code: 'native-host', message, automatic: true } }); } function stillActive(job: Job): boolean { return !!db().prepare(`SELECT 1 FROM server_deployments WHERE id=? AND server_deleted_at IS NULL @@ -267,6 +262,24 @@ export async function processNativeProvisioning(eligibility: (operation: { id: s } const activeJobs = new Map>(); +// Preparation owns startup capacity until its command is committed or discarded. +const preparingJobs = new Set(); +function otherPreparations(id: string): number { return preparingJobs.size - Number(preparingJobs.has(id)); } + +let launchCommit: Promise = Promise.resolve(); +function commitPreparedLaunch(commit: () => void): Promise { + // Worker replies can arrive together. Persist one launch per event-loop turn + // so a batch of large SQLite writes cannot hold off requests and health timers. + const next = launchCommit.then(async () => { await yieldToController(); commit(); }); + launchCommit = next.catch(() => {}); + return next; +} + +function sameLaunchFiles(first: LaunchRecipePlan, second: LaunchRecipePlan): boolean { + const selected = (recipe: LaunchRecipePlan) => ({ game: recipe.game, mapSlug: recipe.mapSlug, + expectedMap: recipe.expectedMap, configFile: recipe.configFile, configSource: recipe.configSource, files: recipe.files }); + return JSON.stringify(selected(first)) === JSON.stringify(selected(second)); +} /** Reservations can move only while no remote work has ever been submitted. */ function placeQueuedCreation(job: Job, instance: NativeInstance): NativeInstance { @@ -285,7 +298,8 @@ function placeQueuedCreation(job: Job, instance: NativeInstance): NativeInstance .run(host.id, bind, port, network.publicAddress ?? host.publicAddress, network.publicPort ?? null, instance.id); db().prepare('UPDATE server_creation_jobs SET port=? WHERE deployment_id=?').run(port, job.id); recipe.native = { ...prior, hostId: host.id, bindAddress: bind, publicAddress: network.publicAddress ?? host.publicAddress, publicPort: network.publicPort }; - db().prepare('UPDATE server_launch_recipes SET recipe=? WHERE deployment_id=?').run(JSON.stringify(recipe), job.id); + const stored = recipeStorage(recipe); + db().prepare('UPDATE server_launch_recipes SET recipe=?,recipe_summary=? WHERE deployment_id=?').run(stored.recipe, stored.recipeSummary, job.id); db().prepare('UPDATE server_launch_previews SET port=? WHERE deployment_id=?').run(port, job.id); deploymentStage(job.id, `Placed on ${host.name} after CPU, memory, history and port checks`, { phase: 'eligibility' }); return nativeInstance(job.server)!; @@ -302,7 +316,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { const settings = fleetSettings(); const active = db().prepare(`SELECT COUNT(*) AS n FROM server_deployments WHERE id!=? AND server_deleted_at IS NULL AND action IN ('create','rebuild') AND state IN ('running','awaiting-health')`).get(job.id) as { n: number }; - if (active.n >= settings.createConcurrency) { hold(job, 'Waiting for startup capacity.'); return; } + if (active.n + otherPreparations(job.id) >= settings.createConcurrency) { hold(job, 'Waiting for startup capacity.'); return; } instance = placeQueuedCreation(job, instance); } const host = getHost(instance.host_id); @@ -311,7 +325,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { hold(job, 'Host is unavailable; existing server state is unknown. Waiting for reconnection.'); return; } if (job.operation_id) { - const operation = getHostOperation(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')) { 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.' @@ -381,7 +395,8 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { if (!status.ok) { hold(job, `Waiting for game status: ${status.reason}`); return; } if (job.action === 'create' || job.action === 'rebuild') { deploymentStage(job.id, 'Checking selected engine map', { phase: 'verify-map', blocker: null }); - const recipe = recipeForDeployment(job.id)!.recipe; + const recipe = db().prepare('SELECT expected_map AS expectedMap FROM server_creation_jobs WHERE deployment_id=?') + .get(job.id) as { expectedMap: string }; check({ key: 'map', label: 'Selected engine map verified', source: status.source, result: status.map === recipe.expectedMap ? 'pass' : 'waiting', expected: recipe.expectedMap, actual: status.map }, 'verify-map'); if (status.map !== recipe.expectedMap) { hold(job, `Waiting for selected map ${recipe.expectedMap}; observed ${status.map ?? 'unknown'}.`); return; } @@ -416,7 +431,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { hold(job, 'Staging and verifying files on the host.'); return; } if (job.stage_operation_id) { - const stage = getHostOperation(job.stage_operation_id); + const stage = getHostOperationStatus(job.stage_operation_id); if (stage?.state === 'failed') { finishDeployment(job.id, { ok: false, message: stage.error ?? 'File staging failed.' }); return; } if (stage?.state !== 'completed') { hold(job, 'Waiting for verified remote file staging.'); return; } if (files.some(file => (db().prepare('SELECT state FROM staged_file_changes WHERE id=?').get(file.id) as { state: string } | undefined)?.state !== 'pending')) { @@ -426,7 +441,7 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { const settings = fleetSettings(); const count = (actions: string[]) => (db().prepare(`SELECT COUNT(*) AS n FROM server_deployments WHERE id != ? AND server_deleted_at IS NULL AND action IN (${actions.map(() => '?').join(',')}) AND state IN ('running','awaiting-health')`).get(job.id, ...actions) as { n: number }).n; - if ((job.action === 'create' || job.action === 'rebuild') && count(['create','rebuild']) >= settings.createConcurrency) { + if ((job.action === 'create' || job.action === 'rebuild') && count(['create','rebuild']) + otherPreparations(job.id) >= settings.createConcurrency) { deploymentStage(job.id, 'Waiting for startup capacity.', { phase: 'startup-queue', blocker: { code: 'startup-capacity', message: 'Waiting for startup capacity.', automatic: true } }); return; } @@ -435,65 +450,103 @@ async function processJob(job: Job, eligibility: (job: Job) => string | null) { if (job.action === 'create') choosePlacement({ hostId: instance.host_id, libraryId: instance.library_id, image: instance.image, bindAddress: instance.bind_address, excludeInstanceId: instance.id, thorough: true, dispatch: true }); if (job.action === 'create' || job.action === 'rebuild') assertReadyLibrary(instance.host_id, instance.library_id); - if (job.action !== 'create' && job.action !== 'start') { + const checkOccupancy = async (): Promise => { // Fast controller check keeps occupied lobbies out of the operation slots. // The agent repeats this immediately before its local mutation. 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; + if (!stillActive(job)) return false; if (!['stopped','exited','created','dead','missing'].includes(health.state)) { const occupancy = await probePlayerEvidence(job.server); - if (!stillActive(job)) return; + if (!stillActive(job)) return false; const recoveredWatchdog = health.healthy && occupancy.players !== null && db().prepare(`SELECT 1 FROM provisioning_sources s 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; } - 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; } + 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; } } - } + return true; + }; + if (job.action !== 'create' && job.action !== 'start' && !await checkOccupancy()) return; const current = db().prepare('SELECT state FROM server_deployments WHERE id=? AND server_deleted_at IS NULL').get(job.id) as { state: string } | undefined; if (!current || !['queued','waiting','interrupted'].includes(current.state) || eligibility(job)) return; - db().transaction(() => { - const live = nativeInstance(job.server); - if (live?.id !== instance!.id || live.generation !== instance!.generation) throw new Error('Instance changed during preflight.'); - if (job.action === 'destroy' || job.action === 'rebuild') { - const beta = activeBetaRemovalBlocker(job.server); - if (beta) { hold(job, beta); return; } - } - if (JSON.stringify(filesFor(job)) !== JSON.stringify(files)) { - finishDeployment(job.id, { ok: false, message: 'File requests changed during preflight. Review and submit a fresh activation.' }); return; - } - // Async occupancy checks can overlap another job claiming the final slot. - if ((job.action === 'create' || job.action === 'rebuild') && count(['create','rebuild']) >= settings.createConcurrency) { - deploymentStage(job.id, 'Waiting for startup capacity.', { phase: 'startup-queue', blocker: { code: 'startup-capacity', message: 'Waiting for startup capacity.', automatic: true } }); return; - } - if (job.action === 'destroy' && count(['destroy']) >= settings.destroyConcurrency) { hold(job, 'Waiting for destruction capacity.'); return; } - if (job.action === 'start' || job.action === 'restart' || job.action === 'rebuild') { - 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, new Date().toISOString(), job.id); - } - recordProgress(job.id, { blocker: null, checkpoint: { key: 'eligibility', label: 'Host eligibility and safety checks passed' } }); - let payload: Record = files.length ? { stageId: job.id } : {}; + let prepared: PreparedNativeLaunch | undefined; + let preparedKey: string | null = null; + try { if (job.action === 'create' || job.action === 'rebuild') { - deploymentStage(job.id, 'Preparing launch files for the host agent', { phase: 'prepare', recipeStatus: 'attempted', - checkpoint: { key: 'startup-queue', label: 'Startup capacity available' } }); + if (count(['create','rebuild']) + otherPreparations(job.id) >= fleetSettings().createConcurrency) { + hold(job, 'Waiting for startup capacity.'); return; + } + preparingJobs.add(job.id); + preparedKey = getServerKey(job.server) ?? null; + if (!preparedKey) throw new Error('The saved server key was removed.'); const prior = recipeForDeployment(job.id)!.recipe as NativeRecipe; - const recipe = { ...captureRecipe(prior.game, prior.mapSlug), native: prior.native, image: instance!.image }; - payload = createPayload(instance!, recipe); - if (job.action === 'rebuild') payload.nextGeneration = instance!.generation + 1; - db().prepare('UPDATE server_launch_recipes SET recipe=? WHERE deployment_id=?').run(JSON.stringify(recipe), job.id); - if (job.action === 'create') db().prepare(`INSERT INTO servers(name,komodo_type,komodo_name,game,map,game_mode,map_slug,host_dir,config_file,game_id,image,game_port,created_by,container_state) - VALUES(?,'deployment','',?,?,'zombie',?,?,?,?,?,?,'native','starting')`).run(job.server, recipe.game, recipe.expectedMap, recipe.mapSlug, serverDir(job.server), - recipe.configFile, recipe.game === 't6' ? 't6zm' : 't5sp', recipe.image, instance!.port); - for (const file of recipe.files) recordScript(job.server, file.path, file.versionId, file.sha256, file.size, 'native-create'); - recordProgress(job.id, { checkpoint: { key: 'prepare', label: 'Launch files selected and command payload prepared' } }); + deploymentStage(job.id, 'Preparing launch files for the host agent', { phase: 'prepare', + checkpoint: { key: 'startup-queue', label: 'Startup capacity available' } }); + prepared = await prepareNativeLaunch({ game: prior.game, mapSlug: prior.mapSlug, native: prior.native, + instance, serverKey: preparedKey, rconPassword: config.rcon.password || randomBytes(32).toString('hex'), + rebuild: job.action === 'rebuild' }); + if (job.action === 'rebuild' && !await checkOccupancy()) return; } - const operation = queueHostOperation(instance!.host_id, `instance.${job.action === 'destroy' ? 'remove' : job.action}`, payload, - { instanceId: instance!.id, generation: instance!.generation, idempotencyKey: job.id }); - db().prepare('UPDATE native_provisioning SET operation_id=? WHERE deployment_id=?').run(operation.id, job.id); - for (const file of files) db().prepare("UPDATE staged_file_changes SET state='applying' WHERE id=? AND state='pending'").run(file.id); - db().prepare("UPDATE server_deployments SET state='running',submitted_at=? WHERE id=?").run(new Date().toISOString(), job.id); - deploymentStage(job.id, `Waiting for host agent to acknowledge ${job.action}`, { phase: 'container-action', blocker: null, - command: { state: 'attempted', message: `Durable agent operation ${operation.id}; reconciliation always uses this ID.` } }); - })(); + const commit = db().transaction(() => { + const state = db().prepare('SELECT state FROM server_deployments WHERE id=? AND server_deleted_at IS NULL').get(job.id) as { state: string } | undefined; + if (!state || !['queued','waiting','interrupted'].includes(state.state) || eligibility(job)) return; + const live = nativeInstance(job.server); + if (live?.id !== instance!.id || live.generation !== instance!.generation) throw new Error('Instance changed during preflight.'); + const liveHost = getHost(instance!.host_id); + if (!liveHost?.connected || liveHost.mode === 'disabled') { hold(job, 'Host is unavailable; waiting for reconnection.'); return; } + if (job.action === 'destroy' || job.action === 'rebuild') { + const beta = activeBetaRemovalBlocker(job.server); + if (beta) { hold(job, beta); return; } + } + if (JSON.stringify(filesFor(job)) !== JSON.stringify(files)) { + finishDeployment(job.id, { ok: false, message: 'File requests changed during preflight. Review and submit a fresh activation.' }); return; + } + // Async occupancy checks can overlap another job claiming the final slot. + if ((job.action === 'create' || job.action === 'rebuild') && count(['create','rebuild']) + otherPreparations(job.id) >= fleetSettings().createConcurrency) { + deploymentStage(job.id, 'Waiting for startup capacity.', { phase: 'startup-queue', blocker: { code: 'startup-capacity', message: 'Waiting for startup capacity.', automatic: true } }); return; + } + if (job.action === 'destroy' && count(['destroy']) >= fleetSettings().destroyConcurrency) { hold(job, 'Waiting for destruction capacity.'); return; } + if (prepared) { + if (getServerKey(job.server) !== preparedKey || !sameLaunchFiles(prepared.recipe, + captureRecipePlan(prepared.recipe.game, prepared.recipe.mapSlug))) { + hold(job, 'Launch files or credentials changed during preparation; preparing the current selection next pass.'); return; + } + assertReadyLibrary(instance!.host_id, instance!.library_id); + if (job.action === 'create') choosePlacement({ hostId: instance!.host_id, libraryId: instance!.library_id, image: instance!.image, + bindAddress: instance!.bind_address, excludeInstanceId: instance!.id, thorough: true, dispatch: true }); + } + if (job.action === 'start' || job.action === 'restart' || job.action === 'rebuild') { + 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, new Date().toISOString(), job.id); + } + recordProgress(job.id, { blocker: null, checkpoint: { key: 'eligibility', label: 'Host eligibility and safety checks passed' } }); + const payload: Record = files.length ? { stageId: job.id } : {}; + if (prepared) { + deploymentStage(job.id, 'Preparing launch files for the host agent', { phase: 'prepare', recipeStatus: 'attempted', + checkpoint: { key: 'startup-queue', label: 'Startup capacity available' } }); + const recipe = prepared.recipe; + db().prepare('UPDATE server_launch_recipes SET recipe=CAST(? AS TEXT),recipe_summary=? WHERE deployment_id=?') + .run(Buffer.from(prepared.recipeJson.buffer, prepared.recipeJson.byteOffset, prepared.recipeJson.byteLength), prepared.recipeSummary, job.id); + if (job.action === 'create') db().prepare(`INSERT INTO servers(name,komodo_type,komodo_name,game,map,game_mode,map_slug,host_dir,config_file,game_id,image,game_port,created_by,container_state) + VALUES(?,'deployment','',?,?,'zombie',?,?,?,?,?,?,'native','starting')`).run(job.server, recipe.game, recipe.expectedMap, recipe.mapSlug, serverDir(job.server), + recipe.configFile, recipe.game === 't6' ? 't6zm' : 't5sp', recipe.image, instance!.port); + for (const file of recipe.files) recordScript(job.server, file.path, file.versionId, file.sha256, file.size, 'native-create'); + recordProgress(job.id, { checkpoint: { key: 'prepare', label: 'Launch files selected and command payload prepared' } }); + } + const operationOptions = { instanceId: instance!.id, generation: instance!.generation, idempotencyKey: job.id }; + const kind = `instance.${job.action === 'destroy' ? 'remove' : job.action}`; + const operation = prepared ? queueHostOperationSerialized(instance!.host_id, kind, + Buffer.from(prepared.payloadJson.buffer, prepared.payloadJson.byteOffset, prepared.payloadJson.byteLength), operationOptions, + { name: instance!.server, game: prepared.recipe.game, libraryId: instance!.library_id }) + : queueHostOperation(instance!.host_id, kind, payload, operationOptions); + db().prepare('UPDATE native_provisioning SET operation_id=? WHERE deployment_id=?').run(operation.id, job.id); + for (const file of files) db().prepare("UPDATE staged_file_changes SET state='applying' WHERE id=? AND state='pending'").run(file.id); + db().prepare("UPDATE server_deployments SET state='running',submitted_at=? WHERE id=?").run(new Date().toISOString(), job.id); + deploymentStage(job.id, `Waiting for host agent to acknowledge ${job.action}`, { phase: 'container-action', blocker: null, + command: { state: 'attempted', message: `Durable agent operation ${operation.id}; reconciliation always uses this ID.` } }); + }); + if (prepared) await commitPreparedLaunch(commit); + else commit(); + } finally { preparingJobs.delete(job.id); } } diff --git a/src/routes/api/agent/operations/+server.ts b/src/routes/api/agent/operations/+server.ts index 735442d..bf7b135 100644 --- a/src/routes/api/agent/operations/+server.ts +++ b/src/routes/api/agent/operations/+server.ts @@ -1,17 +1,17 @@ -import { json } from '@sveltejs/kit'; -import { pendingHostOperations, waitForHostWork } from '$lib/server/hosts/index.js'; +import { pendingHostOperationRows, waitForHostWork } from '$lib/server/hosts/index.js'; import { authenticateAgent } from '$lib/server/hosts/http.js'; import { reconcileHostPreparation } from '$lib/server/hosts/preparation.js'; +import { hostOperationsResponse } from '$lib/server/hosts/delivery.js'; export async function GET({ request }: { request: Request }) { const host = authenticateAgent(request); reconcileHostPreparation(host.id); const params = new URL(request.url).searchParams; const lane = params.get('lane') === 'health' ? 'health' : params.get('lane') === 'read' ? 'read' : params.get('lane') === 'write' ? 'write' : 'all'; - let operations = pendingHostOperations(host.id, lane); + let operations = pendingHostOperationRows(host.id, lane, false); if (!operations.length && params.get('wait') === '20') { await waitForHostWork(host.id, request.signal); authenticateAgent(request); // Revocation still applies after waiting. - operations = pendingHostOperations(host.id, lane); + operations = pendingHostOperationRows(host.id, lane, false); } - return json({ operations, serverTime: new Date().toISOString() }, { headers: { 'Cache-Control': 'no-store' } }); + return hostOperationsResponse(operations); } diff --git a/src/routes/api/agent/operations/[id]/+server.ts b/src/routes/api/agent/operations/[id]/+server.ts index 1a71333..371497c 100644 --- a/src/routes/api/agent/operations/[id]/+server.ts +++ b/src/routes/api/agent/operations/[id]/+server.ts @@ -1,12 +1,12 @@ import { json } from '@sveltejs/kit'; -import { reportHostOperation } from '$lib/server/hosts/index.js'; +import { reportHostOperationStatus } from '$lib/server/hosts/index.js'; import { authenticateAgent, hostApiError, objectBody } from '$lib/server/hosts/http.js'; import { kickProvisioning } from '$lib/server/pipeline/worker.js'; export async function POST({ request, params }: { request: Request; params: { id: string } }) { const host = authenticateAgent(request); try { const body = await objectBody(request, 128 * 1024 * 1024); - const operation = reportHostOperation(host.id, params.id, { state: body.state, result: body.result, error: body.error }); + const operation = reportHostOperationStatus(host.id, params.id, { state: body.state, result: body.result, error: body.error }); if (['completed', 'failed', 'unknown'].includes(operation.state) && (operation.kind.startsWith('instance.') && operation.kind !== 'instance.inspect' || operation.kind === 'files.stage')) { void kickProvisioning().catch(error => console.error('[agent] provisioning wake failed:', error)); diff --git a/tests/helpers/fixture.ts b/tests/helpers/fixture.ts index 1a02c4b..e0b2e1a 100644 --- a/tests/helpers/fixture.ts +++ b/tests/helpers/fixture.ts @@ -49,6 +49,8 @@ export async function newFixture(servers: string[] = ['t5_zombie_theater_1']): P try { const { closeReports } = await import('../../src/lib/server/reports/index.js'); await closeReports(); + const { closeLaunchPreparation } = await import('../../src/lib/server/provisioning/launch-preparation.js'); + await closeLaunchPreparation(); const mod = (await import('../../src/lib/server/db/index.js')) as { closeDb: () => void; }; diff --git a/tests/unit/agent-operation-delivery.test.ts b/tests/unit/agent-operation-delivery.test.ts new file mode 100644 index 0000000..82d7816 --- /dev/null +++ b/tests/unit/agent-operation-delivery.test.ts @@ -0,0 +1,142 @@ +import { setImmediate } from 'node:timers/promises'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { newFixture, type Fixture } from '../helpers/fixture.js'; + +vi.mock('../../src/lib/server/hosts/preparation.js', () => ({ reconcileHostPreparation: vi.fn() })); + +let fixture: Fixture; +const inventory = { + protocolVersion: 1, dockerRunning: true, directoriesWritable: true, storageAvailable: 100 * 1024 ** 3, + runtimeImages: ['plutainer:test'], libraries: [{ id: 'test-library', state: 'ready' }], + capabilities: ['instance.cancel-create', 'operations.health-lane'] +}; + +beforeEach(async () => { vi.resetModules(); fixture = await newFixture([]); }); +afterEach(async () => { vi.restoreAllMocks(); vi.useRealTimers(); await fixture.cleanup(); }); + +async function enrolled() { + const hosts = await import('../../src/lib/server/hosts/index.js'); + const host = hosts.createHost({ name: 'Delivery host' }); + const credentials = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory); + const { GET } = await import('../../src/routes/api/agent/operations/+server.js'); + const request = (query = '') => new Request(`https://ec.test/api/agent/operations${query}`, { + headers: { authorization: `Bearer ${credentials.credential}` } + }); + return { ...hosts, host, GET, request }; +} + +describe('agent operation delivery', () => { + it('streams large persisted launch files without parsing payloads and yields to controller work', async () => { + const api = await enrolled(); + const payload = { + // Place a Unicode character across the first payload byte chunk. + value: 'x'.repeat(64 * 1024 - '{"value":"'.length - 1) + '🧟 café "quoted"\n', + files: [{ path: 'plugins/runtime.dll', contentBase64: 'A'.repeat(8 * 1024 ** 2) }] + }; + const operation = api.queueHostOperation(api.host.id, 'instance.create', payload, + { instanceId: 'instance-one', generation: 2 }); + const result = { phase: 'prepared', description: '世界', details: [null, true, 42] }; + api.reportHostOperation(api.host.id, operation.id, { state: 'running', result }); + const expected = api.getHostOperation(operation.id)!; + const serialized = JSON.stringify(payload); + const originalParse = JSON.parse; + const parse = vi.spyOn(JSON, 'parse').mockImplementation((value, reviver) => { + if (value === serialized) throw new Error('Delivery must not deserialize launch files.'); + return originalParse(value, reviver); + }); + + const response = await api.GET({ request: api.request('?lane=write') }); + expect(response.headers.get('cache-control')).toBe('no-store'); + expect(response.headers.get('content-type')).toContain('application/json'); + let complete = false; + let interleaved = false; + const controllerWork = setImmediate().then(() => { interleaved = !complete; }); + const chunks: Uint8Array[] = []; + const reader = response.body!.getReader(); + while (true) { + const next = await reader.read(); + if (next.done) break; + expect(next.value.byteLength).toBeLessThanOrEqual(64 * 1024); + chunks.push(next.value); + } + complete = true; + await controllerWork; + expect(interleaved).toBe(true); + expect(chunks.length).toBeGreaterThan(100); + parse.mockRestore(); + const body = JSON.parse(Buffer.concat(chunks).toString('utf8')); + expect(body).toEqual({ operations: [expected], serverTime: expect.any(String) }); + expect(Number.isFinite(Date.parse(body.serverTime))).toBe(true); + }); + + it('selects without payloads, loads one payload at a time, and retains the selected receipt snapshot', async () => { + const api = await enrolled(); + const payload = Buffer.from(JSON.stringify({ file: 'A'.repeat(512 * 1024) })); + api.queueHostOperationSerialized(api.host.id, 'instance.create', payload); + const second = api.queueHostOperationSerialized(api.host.id, 'instance.create', payload); + api.reportHostOperationStatus(api.host.id, second.id, { state: 'accepted', result: { phase: 'selected' } }); + const { db } = await import('../../src/lib/server/db/index.js'); + const prepare = vi.spyOn(db(), 'prepare'); + const payloadReads = () => prepare.mock.calls.filter(([sql]) => sql.startsWith('SELECT CAST(') && sql.includes(' AS BLOB) AS payload')).length; + + const response = await api.GET({ request: api.request('?lane=write') }); + const selection = prepare.mock.calls.find(([sql]) => sql.includes('FROM ec_host_operations o'))![0]; + const projection = selection.split(' FROM ')[0]; + expect(projection).not.toContain('payload'); + expect(projection).not.toContain('*'); + expect(payloadReads()).toBe(0); + // A receipt arriving during a streamed response must not mix a newer + // result with the state and timestamps chosen for this delivery. + api.reportHostOperationStatus(api.host.id, second.id, { state: 'running', result: { phase: 'newer' } }); + const reader = response.body!.getReader(); + const first = await reader.read(); + expect(first.done).toBe(false); + expect(payloadReads()).toBe(0); + const chunks: Uint8Array[] = [first.value!]; + let checkedNextPayload = false; + while (true) { + const next = await reader.read(); + if (next.done) break; + if (payloadReads() === 2 && !checkedNextPayload) { + expect(Buffer.concat(chunks).includes(payload)).toBe(true); + checkedNextPayload = true; + } + chunks.push(next.value); + } + expect(payloadReads()).toBe(2); + expect(checkedNextPayload).toBe(true); + const body = JSON.parse(Buffer.concat(chunks).toString('utf8')); + expect(body.operations).toHaveLength(2); + expect(body.operations[1]).toMatchObject({ id: second.id, state: 'accepted', result: { phase: 'selected' } }); + expect(body.operations[0].payload).toEqual(JSON.parse(payload.toString('utf8'))); + expect(body.operations[1].payload).toEqual(body.operations[0].payload); + const other = api.createHost({ name: 'Other host' }); + expect(() => api.getHostOperationPayload(other.id, second.id)).toThrow('not found for this host'); + }); + + it('retains health lanes, delivery rotation, and cancellation suppression through the streamed route', async () => { + const api = await enrolled(); + const operations = Array.from({ length: 27 }, (_, index) => api.queueHostOperation(api.host.id, 'instance.restart', { index })); + const inspect = api.queueHostOperation(api.host.id, 'instance.inspect', {}); + const first = await (await api.GET({ request: api.request('?lane=write') })).json(); + expect(first.operations.map((operation: { id: string }) => operation.id)).toEqual(operations.slice(0, 25).map(operation => operation.id)); + const second = await (await api.GET({ request: api.request('?lane=write') })).json(); + expect(second.operations.slice(0, 2).map((operation: { id: string }) => operation.id)).toEqual(operations.slice(25).map(operation => operation.id)); + const health = await (await api.GET({ request: api.request('?lane=health') })).json(); + expect(health.operations).toEqual([inspect]); + + const create = api.queueHostOperation(api.host.id, 'instance.create', {}); + const cancellation = api.queueHostOperation(api.host.id, 'instance.cancel-create', { createOperationId: create.id }); + const cancelled = await (await api.GET({ request: api.request('?lane=write') })).json(); + expect(cancelled.operations[0].id).toBe(cancellation.id); + expect(cancelled.operations.some((operation: { id: string }) => operation.id === create.id)).toBe(false); + }); + + it('rechecks credentials after waking a long poll before sending work', async () => { + const api = await enrolled(); + const waiting = api.GET({ request: api.request('?lane=health&wait=20') }); + api.revokeCredentials(api.host.id); + api.queueHostOperation(api.host.id, 'instance.inspect', {}); + await expect(waiting).rejects.toMatchObject({ status: 401 }); + }); +}); diff --git a/tests/unit/collection-match-performance.test.ts b/tests/unit/collection-match-performance.test.ts index 5470252..b5c4656 100644 --- a/tests/unit/collection-match-performance.test.ts +++ b/tests/unit/collection-match-performance.test.ts @@ -125,10 +125,11 @@ describe('matching a native collection', () => { await new Promise(resolve => setImmediate(resolve)); await processProvisioning(); - const dispatched = db().prepare(`SELECT n.deployment_id,o.payload FROM native_provisioning n - JOIN ec_host_operations o ON o.id=n.operation_id WHERE o.kind='instance.create'`).all() as { deployment_id: string; payload: string }[]; + const dispatched = db().prepare(`SELECT n.deployment_id,o.id FROM native_provisioning n + JOIN ec_host_operations o ON o.id=n.operation_id WHERE o.kind='instance.create'`).all() as { deployment_id: string; id: string }[]; expect(dispatched).toHaveLength(1); - const payload = JSON.parse(dispatched[0].payload) as { files: { path: string; contentBase64: string; sha256: string }[] }; + const { getHostOperation } = await import('../../src/lib/server/hosts/index.js'); + const payload = getHostOperation(dispatched[0].id)!.payload as { files: { path: string; contentBase64: string; sha256: string }[] }; expect(payload.files.find(file => file.path.endsWith('/fixture.dll'))).toMatchObject({ sha256: version.sha256, contentBase64: content.toString('base64') }); diff --git a/tests/unit/launch-recipe-summary-migration.test.ts b/tests/unit/launch-recipe-summary-migration.test.ts new file mode 100644 index 0000000..b91974e --- /dev/null +++ b/tests/unit/launch-recipe-summary-migration.test.ts @@ -0,0 +1,46 @@ +import Database from 'better-sqlite3'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import schema from '../../src/lib/server/db/schema.sql?raw'; +import { newFixture, type Fixture } from '../helpers/fixture.js'; +import type { LaunchRecipe } from '../../src/lib/server/fleet/launch-recipes.js'; + +let fixture: Fixture; +beforeEach(async () => { vi.resetModules(); fixture = await newFixture([]); }); +afterEach(async () => { vi.restoreAllMocks(); await fixture.cleanup(); }); + +describe('persisted launch recipe summaries', () => { + it('lazily upgrades legacy snapshots once without changing fingerprints or retained bytes', async () => { + const recipe: LaunchRecipe = { + format: 1, game: 't5', mapSlug: 'theater', mapName: 'Kino der Toten', expectedMap: 'zombie_theater', + capturedAt: '2025-01-01T00:00:00.000Z', image: 'plutainer:test', configFile: 'dedicated_sp.cfg', configSource: 'Defaults', + files: [{ name: 'Old plugin', kind: 'plugin', path: 'legacy.dll', versionId: 1, version: 1, + sha256: 'a'.repeat(64), size: 2 * 1024 ** 2, content: Buffer.alloc(2 * 1024 ** 2, 42).toString('base64') }] + }; + const legacy = new Database(fixture.dbPath); + const encoded = JSON.stringify(recipe, null, 2); + try { + legacy.exec(schema.replace(' recipe_summary TEXT,\n', '')); + legacy.exec(`INSERT INTO server_deployments(id,server,action,state,stage,started_at,triggered_by) + VALUES('legacy','retired-server','create','done','Complete','2025-01-01','fixture');`); + legacy.prepare('INSERT INTO server_launch_recipes(deployment_id,recipe,blueprint_name) VALUES(?,?,?)') + .run('legacy', encoded, 'Old collection'); + } finally { legacy.close(); } + const { db, closeDb } = await import('../../src/lib/server/db/index.js'); + const { recipeSummaryForDeployment, recipeForDeployment, summarizeRecipe } = await import('../../src/lib/server/fleet/launch-recipes.js'); + const expected = { recipe: summarizeRecipe(recipe), blueprintName: 'Old collection' }; + expect(db().prepare('SELECT recipe_summary FROM server_launch_recipes').get()).toEqual({ recipe_summary: null }); + expect(recipeSummaryForDeployment('legacy')).toEqual(expected); + closeDb(); + const parse = JSON.parse; + const parsing = vi.spyOn(JSON, 'parse').mockImplementation((value, reviver) => { + if (value.length > 1024 * 1024) throw new Error('Legacy recipe was decoded again.'); + return parse(value, reviver); + }); + try { + expect(recipeSummaryForDeployment('legacy')).toEqual(expected); + expect(recipeSummaryForDeployment('missing')).toBeNull(); + } finally { parsing.mockRestore(); } + expect(recipeForDeployment('legacy')).toEqual({ recipe, blueprintName: 'Old collection' }); + expect(db().prepare('SELECT recipe FROM server_launch_recipes').get()).toEqual({ recipe: encoded }); + }); +}); diff --git a/tests/unit/launch-recipes.test.ts b/tests/unit/launch-recipes.test.ts index 772ef42..30ce810 100644 --- a/tests/unit/launch-recipes.test.ts +++ b/tests/unit/launch-recipes.test.ts @@ -46,6 +46,37 @@ async function seedArtifacts() { } describe('launch recipes and previews', () => { + it('keeps detail polling independent of retained plugin bytes after launching', async () => { + const { createScript } = await import('../../src/lib/server/fleet/scripts.js'); + const { db, closeDb } = await import('../../src/lib/server/db/index.js'); + const plugin = createScript({ name: 'Large utility plugin', kind: 'plugin', scopeType: 'game', scopeValue: 't5', + targetRelPath: 'large.dll', content: Buffer.alloc(8 * 1024 ** 2, 42) }); + db().prepare('UPDATE scripts SET current_version_id = ? WHERE id = ?').run(plugin.version.id, plugin.script.id); + const { queueCreateServer, processCreations } = await import('../../src/lib/server/fleet/create.js'); + const { getDeployment } = await import('../../src/lib/server/fleet/deployments.js'); + const { recipeForDeployment, summarizeRecipe } = await import('../../src/lib/server/fleet/launch-recipes.js'); + const queued = queueCreateServer(input); + if (!queued.ok) throw new Error(queued.message); + const queuedSummary = getDeployment(queued.deploymentId)!.recipe!; + await processCreations(); + const full = recipeForDeployment(queued.deploymentId)!.recipe; + const expected = summarizeRecipe(full); + expect(expected.fingerprint).not.toBe(queuedSummary.fingerprint); + expect(full.files.find(file => file.path === 'large.dll')?.content).toBe(plugin.version.content.toString('base64')); + closeDb(); + const parse = JSON.parse; + const parsing = vi.spyOn(JSON, 'parse').mockImplementation((value, reviver) => { + if (value.length > 1024 * 1024) throw new Error('Detail polling parsed plugin bytes.'); + return parse(value, reviver); + }); + try { + for (let read = 0; read < 3; read++) expect(getDeployment(queued.deploymentId)?.recipe).toEqual(expected); + const row = db().prepare('SELECT length(recipe_summary) AS bytes FROM server_launch_recipes WHERE deployment_id = ?') + .get(queued.deploymentId) as { bytes: number }; + expect(row.bytes).toBeLessThan(10_000); + } finally { parsing.mockRestore(); } + }); + it('previews without reserving capacity, saving keys, exposing bytes, or calling Komodo', async () => { await seedArtifacts(); const { previewCreation } = await import('../../src/lib/server/fleet/create.js'); diff --git a/tests/unit/native-hosts.test.ts b/tests/unit/native-hosts.test.ts index e164b98..a273e9c 100644 --- a/tests/unit/native-hosts.test.ts +++ b/tests/unit/native-hosts.test.ts @@ -15,6 +15,68 @@ async function enrolled() { } describe('native host enrollment and scoped authentication', () => { + it('queues serialized launch files and processes receipts without hydrating plugin payloads', async () => { + const api = await enrolled(); + const { db } = await import('../../src/lib/server/db/index.js'); + const payload = { files: [{ path: 'plugins/fixture.dll', contentBase64: Buffer.alloc(8 * 1024 ** 2, 42).toString('base64') }] }; + const bytes = Buffer.from(JSON.stringify(payload)); + const options = { instanceId: 'immutable-instance', generation: 1, idempotencyKey: 'large-launch' }; + const parse = JSON.parse; + const parsing = vi.spyOn(JSON, 'parse').mockImplementation((value, reviver) => { + if (value.length > 1024 * 1024) throw new Error('Launch payload was unnecessarily parsed.'); + return parse(value, reviver); + }); + let id: string; + try { + const operation = api.queueHostOperationSerialized(api.host.id, 'instance.create', bytes, options); + id = operation.id; + expect(operation).not.toHaveProperty('payload'); + expect(api.queueHostOperationSerialized(api.host.id, 'instance.create', bytes, options).id).toBe(id); + expect(() => api.queueHostOperationSerialized(api.host.id, 'instance.create', Buffer.from('{}'), options)).toThrow('different request'); + const stored = db().prepare('SELECT typeof(payload) AS kind,count(*) AS count FROM ec_host_operations WHERE idempotency_key=?').get(options.idempotencyKey); + expect(stored).toEqual({ kind: 'text', count: 1 }); + const waiting = api.waitHostOperationStatus(id); + expect(api.reportHostOperationStatus(api.host.id, id, { state: 'accepted' }).state).toBe('accepted'); + expect(api.reportHostOperationStatus(api.host.id, id, { state: 'running' }).state).toBe('running'); + api.reportHostOperationStatus(api.host.id, id, { state: 'completed', result: { healthy: true } }); + expect(await waiting).toMatchObject({ id, state: 'completed', result: { healthy: true } }); + expect(api.reportHostOperationStatus(api.host.id, id, { state: 'running' }).state).toBe('completed'); + expect(api.getHostOperationStatus(id)).not.toHaveProperty('payload'); + const other = api.createHost({ name: 'Other host' }); + expect(() => api.reportHostOperationStatus(other.id, id, { state: 'completed' })).toThrow('this host'); + } finally { parsing.mockRestore(); } + // Full-data APIs retain their original contract for explicit readers. + expect(api.getHostOperation(id!)?.payload).toEqual(payload); + expect(api.reportHostOperation(api.host.id, id!, { state: 'running' }).payload).toEqual(payload); + expect((await api.waitHostOperation(id!)).payload).toEqual(payload); + }); + + it('keeps serialized launch routing rows small while full-data APIs hydrate separate immutable bytes', async () => { + const api = await enrolled(); + const { db } = await import('../../src/lib/server/db/index.js'); + const routing = { libraryId: 'test-library', name: 't5_zombie_theater_1', game: 't5' }; + const payload = { ...routing, files: [{ contentBase64: 'A'.repeat(512 * 1024) }] }; + const bytes = Buffer.from(JSON.stringify(payload)); + const launch = api.queueHostOperationSerialized(api.host.id, 'instance.create', bytes, {}, routing); + const inline = api.queueHostOperation(api.host.id, 'instance.inspect', { name: 'inline-legacy' }); + expect(api.getHostOperation(launch.id)?.payload).toEqual(payload); + expect(api.listHostOperations(api.host.id).find(operation => operation.id === launch.id)?.payload).toEqual(payload); + expect(api.pendingHostOperations(api.host.id, 'write').map(operation => operation.payload)).toEqual([payload]); + expect(api.pendingHostOperations(api.host.id, 'read')).toEqual([inline]); + expect(api.pendingHostOperationRows(api.host.id, 'write', false)[0]).not.toHaveProperty('payload'); + api.reportHostOperationStatus(api.host.id, launch.id, { state: 'accepted' }); + api.reportHostOperationStatus(api.host.id, launch.id, { state: 'running', result: { phase: 'starting' } }); + const stored = db().prepare(`SELECT payload,last_delivered_at FROM ec_host_operations WHERE id=?`).get(launch.id) as + { payload: string; last_delivered_at: string | null }; + expect(stored.payload).toBe(JSON.stringify(routing)); + expect(stored.last_delivered_at).not.toBeNull(); + expect(db().prepare('SELECT length(CAST(payload AS BLOB)) AS bytes FROM ec_host_operation_payloads WHERE operation_id=?').get(launch.id)) + .toEqual({ bytes: bytes.length }); + expect(api.getHostOperationPayload(api.host.id, launch.id)).toEqual(bytes); + expect(api.getHostOperationPayload(api.host.id, inline.id).toString('utf8')).toBe(JSON.stringify(inline.payload)); + expect(db().prepare('SELECT 1 FROM ec_host_operation_payloads WHERE operation_id=?').get(inline.id)).toBeUndefined(); + }); + it('narrows cancellation lookups by kind instead of scanning completed health history', async () => { const api = await enrolled(); api.heartbeat(api.host.id, { ...inventory, capabilities: ['instance.cancel-create', 'operations.health-lane'] }); @@ -29,7 +91,7 @@ describe('native host enrollment and scoped authentication', () => { const inspect = api.queueHostOperation(api.host.id, 'instance.inspect', {}); const prepare = vi.spyOn(handle, 'prepare'); expect(api.pendingHostOperations(api.host.id, 'health').map(op => op.id)).toEqual([inspect.id]); - const query = prepare.mock.calls.map(([sql]) => sql).find(sql => sql.includes('SELECT * FROM ec_host_operations o'))!; + const query = prepare.mock.calls.map(([sql]) => sql).find(sql => sql.includes('FROM ec_host_operations o WHERE host_id=?'))!; prepare.mockRestore(); // Assert the actual poll's access path, avoiding hardware-dependent timing thresholds. const plan = handle.prepare(`EXPLAIN QUERY PLAN ${query}`).all(api.host.id, 1, 1) as { detail: string }[]; diff --git a/tests/unit/native-launch-preparation.test.ts b/tests/unit/native-launch-preparation.test.ts new file mode 100644 index 0000000..eeb44f6 --- /dev/null +++ b/tests/unit/native-launch-preparation.test.ts @@ -0,0 +1,227 @@ +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { newFixture, type Fixture } from '../helpers/fixture.js'; + +vi.mock('../../src/lib/server/integrations/plutonium.js', () => ({ + checkRelease: vi.fn(async () => {}), freshRelease: () => 5354, probeVersion: vi.fn(async () => 5354) +})); +vi.mock('../../src/lib/server/pipeline/worker.js', () => ({ kick: vi.fn(async () => {}), kickProvisioning: vi.fn(async () => {}) })); + +let fixture: Fixture; +let hostId: string; +let libraryId: string; +const image = 'example.test/plutainer:test'; +const releases: (() => void)[] = []; +const passes: Promise[] = []; + +beforeEach(async () => { + vi.resetModules(); vi.clearAllMocks(); + fixture = await newFixture([]); + vi.stubEnv('GAME_FILES_ROOT', join(fixture.root, 'libraries')); + vi.stubEnv('PLUTAINER_IMAGE', image); vi.stubEnv('RCON_PASSWORD', 'test-password'); + vi.stubEnv('KOMODO_ADDRESS', ''); vi.stubEnv('KOMODO_API_KEY', ''); vi.stubEnv('KOMODO_API_SECRET', ''); + const { createHost, createEnrollment, enrollHost, heartbeat } = await import('../../src/lib/server/hosts/index.js'); + hostId = createHost({ name: 'Preparation test host', portMin: 28960, portMax: 28980 }).id; + const inventory = { cpuCount: 32, cpuUsagePercent: 0, memoryTotal: 64 * 1024 ** 3, + memoryAvailable: 60 * 1024 ** 3, protocolVersion: 1, dockerRunning: true, + directoriesWritable: true, storageAvailable: 1_000_000_000_000, runtimeImages: [image] }; + enrollHost(createEnrollment(hostId).token, inventory); + const { createLibraryImport, updateHostLibraryCaches } = await import('../../src/lib/server/game-files/index.js'); + const { db } = await import('../../src/lib/server/db/index.js'); + libraryId = createLibraryImport({ hostId, game: 't5' }).id; + db().prepare("UPDATE game_file_versions SET state='ready',manifest='[]',manifest_sha256=?,size_bytes=1,published_at=? WHERE id=?") + .run('a'.repeat(64), new Date().toISOString(), libraryId); + updateHostLibraryCaches(hostId, [{ libraryVersionId: libraryId, status: 'ready', manifestSha256: 'a'.repeat(64) }]); + heartbeat(hostId, { ...inventory, libraries: [{ id: libraryId, state: 'ready' }] }); +}); + +afterEach(async () => { + for (const release of releases.splice(0)) release(); + await Promise.allSettled(passes.splice(0)); + vi.restoreAllMocks(); + await fixture.cleanup(); + vi.unstubAllEnvs(); +}); + +async function queue() { + const { queueNativeCreation } = await import('../../src/lib/server/provisioning/native.js'); + const result = queueNativeCreation({ game: 't5', gameMode: 'zombie', mapSlug: 'theater', + hostId, libraryId, image, plutoKey: 'saved-key' }); + if (!result.ok) throw new Error(result.message); + return result; +} + +async function runPass() { + const { processNativeProvisioning } = await import('../../src/lib/server/provisioning/native.js'); + const pass = processNativeProvisioning(() => null); + passes.push(pass); + await pass; +} + +async function operationFor(deploymentId: string) { + const { db } = await import('../../src/lib/server/db/index.js'); + return db().prepare(`SELECT o.id FROM native_provisioning n JOIN ec_host_operations o ON o.id=n.operation_id + WHERE n.deployment_id=?`).get(deploymentId) as { id: string } | undefined; +} + +/** Pause after real bytes are captured so mutations exercise the final admission checks. */ +async function pausePreparation() { + const preparation = await import('../../src/lib/server/provisioning/launch-preparation.js'); + const prepare = preparation.prepareNativeLaunch; + let started!: () => void; + let release!: () => void; + const ready = new Promise(resolve => { started = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + releases.push(release); + const spy = vi.spyOn(preparation, 'prepareNativeLaunch').mockImplementationOnce(async input => { + const result = await prepare(input); + started(); + await gate; + return result; + }); + return { ready, release, spy }; +} + +describe('asynchronous native launch preparation', () => { + it('keeps controller timers and launch details responsive while preparing multiple large plugins', async () => { + const { createScript } = await import('../../src/lib/server/fleet/scripts.js'); + const { db } = await import('../../src/lib/server/db/index.js'); + const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js'); + const { getDeployment } = await import('../../src/lib/server/provisioning/history.js'); + saveFleetSettings({ ...fleetSettings(), createConcurrency: 4 }); + const { script, version } = createScript({ name: 'Large launch plugin', kind: 'plugin', targetRelPath: 'fixture.dll', + scopeType: 'game', scopeValue: 't5', content: Buffer.alloc(8 * 1024 ** 2, 42) }); + db().prepare('UPDATE scripts SET current_version_id=? WHERE id=?').run(version.id, script.id); + const jobs = await Promise.all(Array.from({ length: 4 }, queue)); + const hosts = await import('../../src/lib/server/hosts/index.js'); + const queueCommand = hosts.queueHostOperationSerialized; + let yielded = true; + let uninterruptedCommits = 0; + const committing = vi.spyOn(hosts, 'queueHostOperationSerialized').mockImplementation((...args) => { + if (!yielded) uninterruptedCommits++; + yielded = false; + setImmediate(() => { yielded = true; }); + return queueCommand(...args); + }); + const parse = JSON.parse; + const stringify = JSON.stringify; + const parsing = vi.spyOn(JSON, 'parse').mockImplementation((raw, reviver) => { + if (typeof raw === 'string' && raw.length > 1_000_000) throw new Error('Large file payload parsed on the controller thread'); + return parse(raw, reviver); + }); + const encoding = vi.spyOn(JSON, 'stringify').mockImplementation((...args: Parameters) => { + const encoded = stringify(...args); + if (encoded && encoded.length > 1_000_000) throw new Error('Large file payload serialized on the controller thread'); + return encoded; + }); + let ticks = 0; + const timer = setInterval(() => { ticks++; }, 1); + try { + const pass = runPass(); + await new Promise(resolve => setTimeout(resolve, 0)); + for (const job of jobs) expect(getDeployment(job.deploymentId)?.recipe?.files) + .toContainEqual(expect.objectContaining({ path: 'fixture.dll', size: 8 * 1024 ** 2 })); + await pass; + expect(ticks).toBeGreaterThan(1); + expect(committing).toHaveBeenCalledTimes(4); + expect(uninterruptedCommits).toBe(0); + for (const job of jobs) { + expect(await operationFor(job.deploymentId)).toBeDefined(); + expect(getDeployment(job.deploymentId)?.recipe?.files) + .toContainEqual(expect.objectContaining({ path: 'fixture.dll', sha256: version.sha256 })); + } + } finally { + clearInterval(timer); parsing.mockRestore(); encoding.mockRestore(); committing.mockRestore(); + } + }, 30_000); + + it('does not resurrect an unsubmitted creation cancelled while its files are prepared', async () => { + const job = await queue(); + const paused = await pausePreparation(); + const pass = runPass(); + await paused.ready; + const { cancelCreation } = await import('../../src/lib/server/provisioning/cancel-creation.js'); + expect(cancelCreation(job.deploymentId)).toEqual(expect.any(String)); + paused.release(); + await pass; + const { db } = await import('../../src/lib/server/db/index.js'); + expect(await operationFor(job.deploymentId)).toBeUndefined(); + expect(db().prepare('SELECT name FROM servers WHERE name=?').get(job.name)).toBeUndefined(); + expect(db().prepare('SELECT instance_id FROM ec_host_ports').all()).toEqual([]); + expect(db().prepare('SELECT state FROM server_deployments WHERE id=?').get(job.deploymentId)).toEqual({ state: 'cancelled' }); + }); + + it('discards prepared bytes when the promoted version changes before dispatch', async () => { + const { createScript, uploadNewVersion } = await import('../../src/lib/server/fleet/scripts.js'); + const { db } = await import('../../src/lib/server/db/index.js'); + const { script, version } = createScript({ name: 'Launch plugin', kind: 'plugin', targetRelPath: 'fixture.dll', + scopeType: 'game', scopeValue: 't5', content: Buffer.from('before promotion') }); + db().prepare('UPDATE scripts SET current_version_id=? WHERE id=?').run(version.id, script.id); + const job = await queue(); + const paused = await pausePreparation(); + const pass = runPass(); + await paused.ready; + const content = Buffer.from('newly promoted plugin'); + const latest = uploadNewVersion(script.id, content); + db().prepare('UPDATE scripts SET current_version_id=? WHERE id=?').run(latest.id, script.id); + paused.release(); + await pass; + expect(await operationFor(job.deploymentId)).toBeUndefined(); + await runPass(); + const operation = await operationFor(job.deploymentId); + expect(operation).toBeDefined(); + const { getHostOperation } = await import('../../src/lib/server/hosts/index.js'); + const payload = getHostOperation(operation!.id)!.payload as { files: { contentBase64: string; sha256: string }[] }; + expect(payload.files).toContainEqual(expect.objectContaining({ contentBase64: content.toString('base64'), sha256: latest.sha256 })); + }); + + it('prepares only one launch while the single startup slot is reserved', async () => { + const first = await queue(); + const second = await queue(); + const paused = await pausePreparation(); + const pass = runPass(); + await paused.ready; + expect(paused.spy).toHaveBeenCalledTimes(1); + const { processNativeProvisioning } = await import('../../src/lib/server/provisioning/native.js'); + await processNativeProvisioning(() => null, { wait: false }); + expect(paused.spy).toHaveBeenCalledTimes(1); + paused.release(); + await pass; + expect(await operationFor(first.deploymentId)).toBeDefined(); + expect(await operationFor(second.deploymentId)).toBeUndefined(); + }); + + it('keeps a prepared launch queued if its host is disabled before dispatch', async () => { + const job = await queue(); + const paused = await pausePreparation(); + const pass = runPass(); + await paused.ready; + const { updateHost } = await import('../../src/lib/server/hosts/index.js'); + updateHost(hostId, { mode: 'disabled' }); + paused.release(); + await pass; + expect(await operationFor(job.deploymentId)).toBeUndefined(); + updateHost(hostId, { mode: 'active' }); + await runPass(); + expect(await operationFor(job.deploymentId)).toBeDefined(); + }); + + it('rechecks a reduced startup limit before submitting prepared files', async () => { + const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js'); + saveFleetSettings({ ...fleetSettings(), createConcurrency: 2 }); + const first = await queue(); + await runPass(); + expect(await operationFor(first.deploymentId)).toBeDefined(); + const second = await queue(); + const paused = await pausePreparation(); + const pass = runPass(); + await paused.ready; + saveFleetSettings({ ...fleetSettings(), createConcurrency: 1 }); + paused.release(); + await pass; + expect(await operationFor(second.deploymentId)).toBeUndefined(); + saveFleetSettings({ ...fleetSettings(), createConcurrency: 2 }); + await runPass(); + expect(await operationFor(second.deploymentId)).toBeDefined(); + }); +}); diff --git a/tests/unit/native-provisioning.test.ts b/tests/unit/native-provisioning.test.ts index 991ecad..0d27b57 100644 --- a/tests/unit/native-provisioning.test.ts +++ b/tests/unit/native-provisioning.test.ts @@ -246,7 +246,9 @@ describe('native instance provisioning', () => { try { const { getDeployment } = await import('../../src/lib/server/provisioning/history.js'); expect(getDeployment(first.deploymentId)!.progress?.phase).toBe('verify-container'); - expect((await operationFor(second.deploymentId))?.kind).toBe('instance.create'); + await vi.waitFor(async () => { + expect((await operationFor(second.deploymentId))?.kind).toBe('instance.create'); + }, { timeout: 5_000 }); } finally { release({ healthy: true, state: 'running' } as never); await pending; @@ -268,7 +270,9 @@ describe('native instance provisioning', () => { try { const second = await queue(); await processProvisioning({ wait: false }); - expect((await operationFor(second.deploymentId))?.kind).toBe('instance.create'); + await vi.waitFor(async () => { + expect((await operationFor(second.deploymentId))?.kind).toBe('instance.create'); + }, { timeout: 5_000 }); expect(vi.mocked(nativeRequest)).toHaveBeenCalledTimes(1); } finally { release({ healthy: true, state: 'running' } as never); @@ -401,6 +405,32 @@ describe('native instance provisioning', () => { await advance(); expect(nativeInstance(job.name)?.generation).toBe(2); }); + it('holds a rebuild when players join while its launch files are being prepared', async () => { + const job = await queue(); await advance(); 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); + const preparation = await import('../../src/lib/server/provisioning/launch-preparation.js'); + const prepare = preparation.prepareNativeLaunch; + let started!: () => void; + let release!: () => void; + const ready = new Promise(resolve => { started = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const spy = vi.spyOn(preparation, 'prepareNativeLaunch').mockImplementationOnce(async input => { + const result = await prepare(input); + started(); await gate; + return result; + }); + const pending = advance(); + try { + await ready; + remote.players = 2; + release(); await pending; + expect(await operationFor(rebuild.deploymentId)).toBeNull(); + remote.players = 0; + await advance(); + expect((await operationFor(rebuild.deploymentId))?.kind).toBe('instance.rebuild'); + } finally { release(); await pending; spy.mockRestore(); } + }); it('finishes native rebuild verification without waiting for its own recovery slot to be released', async () => { const job = await queue(); await advance(); await acknowledge(job.deploymentId); const { queueNativeRebuild } = await import('../../src/lib/server/provisioning/native.js'); diff --git a/vite.launch.config.ts b/vite.launch.config.ts new file mode 100644 index 0000000..79e9314 --- /dev/null +++ b/vite.launch.config.ts @@ -0,0 +1,16 @@ +import { defineConfig } from 'vite'; +import { resolve } from 'node:path'; + +// Reuse launch composition with an independent read-only database connection. +export default defineConfig({ + plugins: [{ name: 'launch-database', enforce: 'pre', resolveId(source) { + if (source.endsWith('/db/index.js')) return resolve('src/lib/server/reports/database.ts'); + } }], + build: { + ssr: true, outDir: '.svelte-kit/launch', emptyOutDir: true, + rollupOptions: { + input: 'src/lib/server/provisioning/launch-worker.ts', external: ['better-sqlite3'], + output: { entryFileNames: 'worker.mjs' } + } + } +});