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

Pace collection startups and clarify deployment queue progress

Browse Source
This commit is contained in:
josh
2026-09-19 20:33:08 -04:00
parent 1d073779fe
commit f171889dfb
18 changed files with 693 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/fleet/collections.ts
+10 -4
View File
@@ -3,7 +3,7 @@ import type { CollectionDetail, CollectionPreview, CollectionSummary } from '$li
import { db } from '../db/index.js';
import { config, komodoConfigured, rconConfigured } from '../config.js';
import { blueprintEntries, summarizeRecipe, type BlueprintEntry } from './launch-recipes.js';
import { nextAvailablePort, nextLobbyNumber, queueCreateServer } from './create.js';
import { creationQueue, nextAvailablePort, nextLobbyNumber, queueCreateServer } from './create.js';
import { getServerKey } from './server-keys.js';
import { parseServerName, zombieMapBySlug, zombieMaps } from './map-aliases.js';
@@ -70,7 +70,7 @@ export function previewCollection(blueprintId: unknown): CollectionPreview {
db().prepare('DELETE FROM server_collection_previews WHERE expires_at < ? AND collection_id IS NULL').run(new Date().toISOString());
db().prepare('INSERT INTO server_collection_previews (id, expires_at, blueprint_name, entries, servers) VALUES (?, ?, ?, ?, ?)')
.run(id, expiresAt, blueprintName, JSON.stringify(entries), JSON.stringify(servers));
return { id, expiresAt, blueprintName, servers, blockers, ready: blockers.length === 0,
return { id, expiresAt, blueprintName, launchConcurrency: config.createConcurrency, servers, blockers, ready: blockers.length === 0,
entries: entries.map((entry, index) => ({ count: entry.count, ...counts[index], recipe: summarizeRecipe(entry.recipe) })) };
})();
}
@@ -123,7 +123,10 @@ export function queueCollection(input: { previewId?: unknown; keys?: unknown }):
const summarySql = `SELECT c.id, c.name, c.created_at, COUNT(d.id) AS total,
COALESCE(SUM(d.state IN ('done', 'warning')), 0) AS done,
COALESCE(SUM(d.state IN ('failed', 'interrupted')), 0) AS failed,
COALESCE(SUM(d.state IN ('queued', 'running', 'awaiting-health', 'waiting')), 0) AS active
COALESCE(SUM(d.state IN ('queued', 'running', 'awaiting-health', 'waiting')), 0) AS active,
COALESCE(SUM(d.state = 'queued'), 0) AS queued,
COALESCE(SUM(d.state = 'running'), 0) AS provisioning,
COALESCE(SUM(d.state = 'awaiting-health'), 0) AS verifying
FROM server_launch_collections c
LEFT JOIN server_collection_members m ON m.collection_id = c.id
LEFT JOIN server_deployments d ON d.id = m.deployment_id`;
@@ -139,5 +142,8 @@ export function getCollection(id: string): CollectionDetail | null {
EXISTS(SELECT 1 FROM servers s WHERE s.name = d.server) AS server_exists
FROM server_collection_members m JOIN server_deployments d ON d.id = m.deployment_id
JOIN server_creation_jobs j ON j.deployment_id = d.id WHERE m.collection_id = ? ORDER BY m.position`).all(id) as CollectionDetail['servers'];
return { collection, servers };
const queue = creationQueue();
const positions = new Map(queue.queued.map((job, index) => [job.id, index + 1]));
return { collection, queue: { concurrency: config.createConcurrency, active: queue.active, queued: queue.queued.length },
servers: servers.map(server => ({ ...server, queuePosition: positions.get(server.id) ?? null })) };
}
src/lib/server/fleet/create.ts
+85 -16
View File
@@ -1,5 +1,5 @@
import { rconGetPlayerCount, rconStatusResult } from '../integrations/rcon.js';
import { beginDeployment, deploymentStage, finishDeployment, REMOVING_SERVER, resumeCleanup } from './deployments.js';
import { beginDeployment, deploymentStage, finishDeployment, recordDeploymentEvent, REMOVING_SERVER, resumeCleanup, resumeVerification } from './deployments.js';
import { randomUUID } from 'node:crypto';
import type { LaunchPreview } from '$lib/deployment-plan.js';
import { blueprintRecipe, captureRecipe, recipeForDeployment, summarizeRecipe, type LaunchRecipe } from './launch-recipes.js';
@@ -234,14 +234,17 @@ function nextName(input: CreateServerInput): string | null {
return lobby === null ? null : `${base}_${lobby}`;
}
// Await provisioning for internal callers; completion still requires worker health verification.
// Internal callers use the same durable queue and capacity limit as HTTP launches.
// Await this pass's provisioning when possible; a busy host leaves the job queued.
export async function createServer(input: CreateServerInput): Promise<CreateServerResult> {
if (creating || destroying.size > 0) return { ok: false, kind: 'no-slot', message: 'Another server operation is in progress. Please retry shortly.' };
const queued = queueCreateServer(input);
if (!queued.ok) return queued;
const existing = db().prepare('SELECT state FROM server_deployments WHERE id = ?').get(queued.deploymentId) as { state: string };
if (existing.state !== 'queued') return queued;
return await provisionCreation(queued.deploymentId);
const outcomes = await creationPass();
const outcome = outcomes.get(queued.deploymentId);
if (outcome instanceof Error) throw outcome;
return outcome ?? queued;
}
function prepareCreation(input: CreateServerInput): PreparedCreation | Extract<CreateServerResult, { ok: false }> {
@@ -378,6 +381,12 @@ async function createServerUnlocked(input: CreateServerInput, prepared: Prepared
deploymentStage(operation, 'Starting container');
const deployRes = await komodoDeploy(name);
if (!deployRes.ok) {
if (deployRes.kind === 'unreachable' || (deployRes.httpStatus !== undefined && deployRes.httpStatus >= 500)) {
// A timeout or server error may arrive after the remote boot began.
// Preserve files, identity, and capacity until readiness is known.
recordDeploymentEvent(operation, 'Startup response uncertain; verifying the existing server: ' + deployRes.message);
return { ok: true, name, port, deploymentId: operation };
}
// Container never started — undo everything so retrying is clean.
await komodoDeleteDeployment(name).catch(() => {});
db().prepare('DELETE FROM servers WHERE name = ?').run(name);
@@ -596,30 +605,71 @@ async function provisionCreation(id: string): Promise<CreateServerResult> {
}
return result;
} catch (err) {
finishDeployment(id, { ok: false, message: (err as Error).message });
const server = getServer(job.server);
if (server?.game_port === job.port) {
// A lost response may have happened after Komodo started the server.
// Keep its slot and observe readiness instead of starting another boot.
db().prepare("UPDATE server_deployments SET state = 'awaiting-health' WHERE id = ?").run(id);
recordDeploymentEvent(id, 'Startup response interrupted: ' + (err as Error).message);
deploymentStage(id, 'Verifying existing server after interrupted startup');
} else finishDeployment(id, { ok: false, message: (err as Error).message });
throw err;
} finally { creating = false; }
}
let creationTick: Promise<void> | null = null;
export function processCreations(): Promise<void> {
type CreationOutcomes = Map<string, CreateServerResult | Error>;
let creationTick: Promise<CreationOutcomes> | null = null;
function creationPass(): Promise<CreationOutcomes> {
if (creationTick) return creationTick;
creationTick = runCreations().finally(() => { creationTick = null; });
return creationTick;
}
async function runCreations(): Promise<void> {
if (creating || destroying.size > 0) return;
const queued = db().prepare("SELECT id FROM server_deployments WHERE action = 'create' AND state = 'queued' ORDER BY started_at").all() as { id: string }[];
for (const { id } of queued) {
if (creating || destroying.size > 0) break;
await provisionCreation(id).catch(err => console.error('[creation] provisioning failed:', err));
export async function processCreations(): Promise<void> {
await creationPass();
}
/** One host-wide FIFO, shared by collections and individual launches. */
export function creationQueue() {
type Entry = { id: string; server: string; state: string; stage: string };
const jobs = db().prepare(`SELECT id, server, state, stage FROM server_deployments
WHERE action = 'create' AND state IN ('queued', 'running', 'awaiting-health')
ORDER BY started_at, rowid`).all() as Entry[];
return { active: jobs.filter(job => job.state !== 'queued'), queued: jobs.filter(job => job.state === 'queued') };
}
async function runCreations(): Promise<CreationOutcomes> {
const outcomes: CreationOutcomes = new Map();
if (creating || destroying.size > 0) return outcomes;
// Restart recovery is observation-only: never repeat external creation or
// deployment. A known interrupted startup must retain its readiness gate.
const interrupted = db().prepare("SELECT id FROM server_deployments WHERE action = 'create' AND state = 'interrupted' ORDER BY started_at, rowid").all() as { id: string }[];
for (const { id } of interrupted) {
if (resumeVerification(id)) continue;
// A queued removal does not make the interrupted container disappear.
// Keep its slot until removal finishes, without resuming normal work or
// confusing it with a newer creation of the same server identity.
const removing = db().prepare(`SELECT 1 FROM server_deployments d
JOIN server_creation_jobs j ON j.deployment_id = d.id
JOIN servers s ON s.name = j.server AND s.game_port = j.port
WHERE d.id = ? AND EXISTS (SELECT 1 FROM server_deployments newer WHERE newer.server = d.server
AND newer.rowid > d.rowid AND newer.action = 'destroy' AND newer.state IN ('waiting', 'running'))
AND NOT EXISTS (SELECT 1 FROM server_deployments newer WHERE newer.server = d.server
AND newer.rowid > d.rowid AND newer.action = 'create')`).get(id);
if (removing) {
db().prepare("UPDATE server_deployments SET state = 'awaiting-health', finished_at = NULL, error = NULL WHERE id = ?").run(id);
deploymentStage(id, 'Waiting for requested server removal after interrupted startup');
}
}
const checking = db().prepare("SELECT j.* FROM server_creation_jobs j JOIN server_deployments d ON d.id = j.deployment_id WHERE d.state = 'awaiting-health'").all() as CreationJob[];
// Check only previously started servers. Newly launched servers keep their
// slot until a later pass, even if a health probe would answer immediately.
const checking = db().prepare(`SELECT j.* FROM server_creation_jobs j JOIN server_deployments d ON d.id = j.deployment_id
WHERE d.state = 'awaiting-health' ORDER BY d.started_at, d.rowid`).all() as CreationJob[];
for (const job of checking) {
const id = job.deployment_id;
if (!getServer(job.server)) {
finishDeployment(id, { ok: false, message: 'Server was removed before creation health checks completed' });
if (getServer(job.server)?.game_port !== job.port) {
finishDeployment(id, { ok: false, message: 'Server was removed or replaced before creation health checks completed' });
continue;
}
if (db().prepare("SELECT id FROM server_deployments WHERE server = ? AND action = 'destroy' AND state IN ('waiting', 'running')").get(job.server)) continue;
@@ -639,4 +689,23 @@ async function runCreations(): Promise<void> {
deploymentStage(id, 'Waiting for health verification: ' + (err as Error).message);
}
}
// Snapshot the budget once. Fast success/failure must not turn a single
// worker pass into an unbounded stream of expensive container startups.
const queue = creationQueue();
const available = Math.max(0, config.createConcurrency - queue.active.length);
for (const { id } of queue.queued.slice(0, available)) {
if (creating || destroying.size > 0) break;
// API calls can cancel a selected job or resume an older startup while
// a previous provision awaits Komodo. Recheck without expanding this pass.
if (creationQueue().active.length >= config.createConcurrency) break;
if (!db().prepare("SELECT 1 FROM server_deployments WHERE id = ? AND action = 'create' AND state = 'queued'").get(id)) continue;
try { outcomes.set(id, await provisionCreation(id)); }
catch (err) {
const failure = err instanceof Error ? err : new Error(String(err));
outcomes.set(id, failure);
console.error('[creation] provisioning failed:', failure);
}
}
return outcomes;
}