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

feat: add Watchdog recovery and drift rebuild workflows

Browse Source
This commit is contained in:
josh
2026-09-27 15:13:38 -04:00
parent a7044f21a1
commit b9f5902945
36 changed files with 2131 additions and 424 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/db/index.ts
+41
View File
@@ -16,6 +16,7 @@ let _db: Database.Database | null = null;
// backfill columns; this can.
const SERVERS_COLUMNS: Array<{ name: string; ddl: string }> = [
{ name: 'container_state', ddl: 'ALTER TABLE servers ADD COLUMN container_state TEXT' },
{ name: 'desired_state', ddl: "ALTER TABLE servers ADD COLUMN desired_state TEXT NOT NULL DEFAULT 'running' CHECK (desired_state IN ('running','stopped'))" },
// Base columns that shipped in the very first schema stay here for parity
// with fresh installs — no ALTER is issued if they're already present.
{ name: 'komodo_server', ddl: 'ALTER TABLE servers ADD COLUMN komodo_server TEXT' },
@@ -246,6 +247,44 @@ function migrateProvisioningActions(handle: Database.Database): void {
} finally { handle.pragma('foreign_keys = ON'); }
}
/** Rebuild constrained tables while preserving links, indexes, triggers, and row order. */
function migrateWatchdogConstraints(handle: Database.Database): void {
for (const name of ['provisioning_sources', 'deployment_servers']) {
const { sql } = handle.prepare("SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?").get(name) as { sql: string };
const replacement = name === 'provisioning_sources'
? (sql.includes("'watchdog'") ? sql : sql.replace("'campaign','beta','maintenance'", "'campaign','beta','maintenance','watchdog'"))
: sql.replace(/server\s+TEXT NOT NULL REFERENCES servers\(name\)/, 'server TEXT NOT NULL');
if (sql === replacement) continue;
const dependent = handle.prepare("SELECT sql FROM sqlite_master WHERE tbl_name = ? AND type IN ('index','trigger') AND sql IS NOT NULL")
.all(name) as { sql: string }[];
handle.pragma('foreign_keys = OFF');
try {
handle.transaction(() => {
handle.exec(replacement.replace(name, name + '_watchdog_new'));
handle.exec(`INSERT INTO ${name}_watchdog_new (rowid, ${
(handle.prepare(`PRAGMA table_info(${name})`).all() as { name: string }[]).map(col => col.name).join(', ')
}) SELECT rowid, ${
(handle.prepare(`PRAGMA table_info(${name})`).all() as { name: string }[]).map(col => col.name).join(', ')
} FROM ${name}; DROP TABLE ${name}; ALTER TABLE ${name}_watchdog_new RENAME TO ${name};`);
for (const object of dependent) handle.exec(object.sql);
if ((handle.pragma('foreign_key_check') as unknown[]).length) throw new Error('Watchdog migration found broken references');
})();
} finally { handle.pragma('foreign_keys = ON'); }
}
}
function backfillDesiredState(handle: Database.Database): void {
if (handle.prepare("SELECT 1 FROM settings WHERE key = 'watchdog_desired_state_v1'").get()) return;
handle.transaction(() => {
// A failed stop may have reached Komodo. Preserve that explicit intent.
handle.exec(`UPDATE servers SET desired_state = 'stopped' WHERE 'stop' = (
SELECT action FROM server_deployments d WHERE d.server = servers.name AND d.state != 'cancelled'
AND d.action IN ('create','start','restart','rebuild','stop')
ORDER BY d.rowid DESC LIMIT 1);
INSERT INTO settings (key, value) VALUES ('watchdog_desired_state_v1', '1');`);
})();
}
export function db(): Database.Database {
if (_db) return _db;
@@ -260,6 +299,7 @@ export function db(): Database.Database {
dropLegacyBetaTables(handle);
handle.exec(schemaSql);
migrateProvisioningActions(handle);
migrateWatchdogConstraints(handle);
migrateTable(handle, 'restart_rollout', [{ name: 'operation_id', ddl: 'ALTER TABLE restart_rollout ADD COLUMN operation_id TEXT' }]);
migrateTable(handle, 'server_deployments', [
{ name: 'submitted_at', ddl: 'ALTER TABLE server_deployments ADD COLUMN submitted_at TEXT' },
@@ -270,6 +310,7 @@ export function db(): Database.Database {
migrateTable(handle, table, [{ name: 'provisioning_id', ddl: `ALTER TABLE ${table} ADD COLUMN provisioning_id TEXT` }]);
}
migrateServers(handle);
backfillDesiredState(handle);
// Column additions must run BEFORE any statement that references the
// new columns. schema.sql cannot include those references — see the
// comment there.
src/lib/server/db/schema.sql
+36 -2
View File
@@ -64,6 +64,7 @@ CREATE TABLE IF NOT EXISTS servers (
komodo_name TEXT NOT NULL,
komodo_server TEXT,
container_state TEXT, -- last observed Komodo runtime state
desired_state TEXT NOT NULL DEFAULT 'running' CHECK (desired_state IN ('running','stopped')),
game_port INTEGER, -- UDP RCON port; lazy-filled from Komodo
rcon_host TEXT, -- optional per-server override of RCON_HOST
game TEXT, -- 't5' etc.; static per-server (no map rotation)
@@ -299,7 +300,7 @@ CREATE INDEX IF NOT EXISTS idx_deployments_started ON deployments(started_at DES
CREATE TABLE IF NOT EXISTS deployment_servers (
deployment_id TEXT NOT NULL REFERENCES deployments(id) ON DELETE CASCADE,
server TEXT NOT NULL REFERENCES servers(name),
server TEXT NOT NULL, -- Campaign observation survives a rebuild's registry gap.
files_changed INTEGER NOT NULL,
restart_state TEXT NOT NULL, -- pending | done | failed | manual | held
players INTEGER,
@@ -445,7 +446,7 @@ CREATE TABLE IF NOT EXISTS server_rebuild_jobs (
-- history retention, while current pointers live on the caller's server rows.
CREATE TABLE IF NOT EXISTS provisioning_sources (
operation_id TEXT NOT NULL REFERENCES server_deployments(id) ON DELETE CASCADE,
kind TEXT NOT NULL CHECK (kind IN ('campaign','beta','maintenance')),
kind TEXT NOT NULL CHECK (kind IN ('campaign','beta','maintenance','watchdog')),
source_id TEXT NOT NULL,
server TEXT NOT NULL,
PRIMARY KEY (operation_id, kind, source_id, server)
@@ -622,3 +623,36 @@ CREATE TABLE IF NOT EXISTS metric_samples (
capacity INTEGER,
PRIMARY KEY (at, server)
);
-- Watchdog history deliberately survives server destruction/rebuild.
CREATE TABLE IF NOT EXISTS watchdog_incidents (
id TEXT PRIMARY KEY,
server TEXT NOT NULL,
kind TEXT NOT NULL CHECK (kind IN ('health','drift')),
state TEXT NOT NULL CHECK (state IN ('open','recovering','attention','resolved','dismissed')),
summary TEXT NOT NULL,
details TEXT NOT NULL DEFAULT '{}',
first_seen_at TEXT NOT NULL,
last_seen_at TEXT NOT NULL,
finished_at TEXT,
failures INTEGER NOT NULL DEFAULT 1,
cycle INTEGER NOT NULL DEFAULT 1
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_watchdog_active ON watchdog_incidents(server,kind) WHERE state != 'resolved';
CREATE TABLE IF NOT EXISTS watchdog_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
incident_id TEXT NOT NULL REFERENCES watchdog_incidents(id) ON DELETE CASCADE,
created_at TEXT NOT NULL,
message TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_watchdog_events ON watchdog_events(incident_id,id);
CREATE TABLE IF NOT EXISTS watchdog_actions (
id TEXT PRIMARY KEY,
incident_id TEXT NOT NULL REFERENCES watchdog_incidents(id) ON DELETE CASCADE,
action TEXT NOT NULL CHECK (action IN ('restart','rebuild')),
operation_id TEXT NOT NULL UNIQUE,
campaign_id TEXT,
created_at TEXT NOT NULL,
cycle INTEGER NOT NULL DEFAULT 1,
UNIQUE(incident_id,cycle,action)
);
src/lib/server/fleet/drift.ts
+129 -37
View File
@@ -1,22 +1,20 @@
import { mkdir, readdir, readFile, rm, stat, writeFile } from 'node:fs/promises';
import { randomUUID } from 'node:crypto';
import type { DriftPlan, DriftPlanFile } from '$lib/drift-plan.js';
import { existsSync } from 'node:fs';
import { dirname, join, posix, relative, sep } from 'node:path';
import { configFileOf, destOfKind, type ArtifactKind } from '../config.js';
import { configFileOf, destOfKind, serverDir, type ArtifactKind } from '../config.js';
import { db } from '../db/index.js';
import { emit } from '../pipeline/bus.js';
import { sha256Buffer } from './files.js';
import {
getServerScript,
listServerScripts,
recordScript,
removeScript
} from './inventory.js';
import { getBaselineForScope, getVersion } from './scripts.js';
import { getPromotedServerConfig, getServerMapConfig, MAP_CONFIG_FILE } from './server-config.js';
import { selectedZombieMap } from './map-aliases.js';
import { serverDirectories, getServer, listServers } from './servers.js';
import { getServer, listServers } from './servers.js';
import { promotedArtifacts } from './promoted-artifacts.js';
// Destinations drift scans, in a stable order so the UI shows scripts first.
@@ -54,14 +52,35 @@ function toPosix(p: string): string {
return p.split(sep).join(posix.sep);
}
function hasActiveBeta(server: string): boolean {
return !!db().prepare(`SELECT 1 FROM beta_servers bs JOIN betas b ON b.id = bs.beta_id
WHERE bs.server = ? AND b.state IN ('running', 'ready-promote') LIMIT 1`).get(server);
}
async function requireServerDirectory(server: string): Promise<void> {
try {
if (!(await stat(serverDir(server))).isDirectory()) throw new Error('path is not a directory');
} catch (error) {
throw new Error(`Server files unavailable for ${server}: ${(error as Error).message}`);
}
}
async function readExistingFile(path: string): Promise<Buffer | null> {
return readFile(path).catch((error: NodeJS.ErrnoException) => {
if (error.code === 'ENOENT') return null;
throw error;
});
}
async function walkFiles(root: string): Promise<string[]> {
const out: string[] = [];
async function recur(dir: string): Promise<void> {
let entries;
try {
entries = await readdir(dir, { withFileTypes: true });
} catch {
return;
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') return;
throw error;
}
for (const e of entries) {
if (DRIFT_IGNORE_FILENAMES.has(e.name)) continue;
@@ -76,27 +95,28 @@ async function walkFiles(root: string): Promise<string[]> {
/**
* Scan every managed destination tree on a server (scripts, configs, plugins)
* and return every file that isn't tracked in server_scripts. During a beta,
* the beta bytes live at the actual target filename and the inventory row is
* updated to reflect them, so beta files register as tracked with no
* special-casing (pre-beta bytes are snapshotted to beta_file_snapshots for
* rollback rather than left on disk as .OLD).
*
* The inventory table (server_scripts) keys on (server, rel_path) with no
* destination discriminator. Rel_paths across destinations don't collide in
* practice (`.gsc` for scripts, `.cfg` for configs, `.dll` for plugins), so
* a shared tracked-set is safe.
* and compare disk contents with promoted artifacts and inventory. An active
* beta owns its server's files until promotion or rollback; defer the whole
* server so reconciliation cannot replace experimental config or missing beta
* artifacts with the promoted versions.
*/
export async function scanServerDrift(server: string): Promise<DriftEntry[]> {
const tracked = new Set(listServerScripts(server).map((r) => r.rel_path));
const drift = await scanManagedForServer(server);
const managed = new Set(drift.map((entry) => `${entry.kind}:${entry.relPath}`));
const srv = getServer(server);
if (!srv || srv.destroyed_at || hasActiveBeta(server)) return [];
// A missing mount is unavailable evidence, not an empty managed tree.
await requireServerDirectory(server);
const drift = await scanManagedForServer(server);
// Previous inventory is an audit of installed bytes, not permission to
// retain artifacts whose scope was moved or removed from the library.
const managed = new Set(promotedArtifacts(srv.game, srv.map)
.map(({ kind, script }) => `${kind}:${script.target_rel_path}`));
const baseline = srv.game ? getBaselineForScope('game', srv.game) : undefined;
if (baseline?.current_version_id) managed.add(`config:${configFileOf(server)}`);
// Without a promoted baseline, Plutainer owns the seeded config. Keep it
// until a baseline is promoted; startup arguments handle map selection.
const keepDefaultConfig = srv?.game
&& selectedZombieMap(srv.game, srv.map, srv.map_slug)
&& !getBaselineForScope('game', srv.game)?.current_version_id;
&& !baseline?.current_version_id;
// Scan for untracked files only after resolving managed targets.
for (const kind of DRIFT_DESTS) {
@@ -105,7 +125,7 @@ export async function scanServerDrift(server: string): Promise<DriftEntry[]> {
if (onDisk.length === 0) continue;
for (const relPath of onDisk) {
if (tracked.has(relPath) || managed.has(`${kind}:${relPath}`)) continue;
if (managed.has(`${kind}:${relPath}`)) continue;
if (keepDefaultConfig && kind === 'config' && relPath === configFileOf(server)) continue;
if (kind === 'config' && relPath === MAP_CONFIG_FILE && srv?.created_by
&& srv.game_mode === 'zombie' && getServerMapConfig(srv.game, srv.map, srv.map_slug)) continue;
@@ -120,13 +140,15 @@ export async function scanServerDrift(server: string): Promise<DriftEntry[]> {
kind,
driftKind: 'untracked'
});
} catch {
/* file disappeared between walk and stat — skip */
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
// A file disappeared between walk and stat; the next scan sees it.
}
}
}
return drift;
await requireServerDirectory(server);
return hasActiveBeta(server) ? [] : drift;
}
// Every promoted artifact whose scope matches this server's game/map should
@@ -143,21 +165,23 @@ async function scanManagedForServer(server: string): Promise<DriftEntry[]> {
? getServerMapConfig(srv.game, srv.map, srv.map_slug) : null;
if (mapConfig) {
const abs = join(destOfKind(server, 'config'), MAP_CONFIG_FILE);
const content = await readFile(abs).catch(() => null);
const content = await readExistingFile(abs);
if (!content?.equals(mapConfig)) {
out.push({ server, relPath: MAP_CONFIG_FILE, kind: 'config', driftKind: content ? 'replace' : 'missing',
size: mapConfig.byteLength, sha256: sha256Buffer(mapConfig) });
}
}
// Existing files without inventory must be installed/tracked in one pass,
// never deleted first. Preserve tracked files (including beta versions).
// Inventory records the last managed write, not the current disk bytes.
// Check both so edits and outdated versions are detected on tracked files.
for (const { kind, script, version } of promotedArtifacts(srv.game, srv.map)) {
const present = existsSync(join(destOfKind(server, kind), script.target_rel_path));
if (present && getServerScript(server, script.target_rel_path)) continue;
const content = await readExistingFile(join(destOfKind(server, kind), script.target_rel_path));
const inventory = getServerScript(server, script.target_rel_path);
if (content?.equals(version.content) && inventory?.sha256 === version.sha256
&& inventory.script_version_id === version.id) continue;
out.push({
server, relPath: script.target_rel_path, kind,
driftKind: present ? 'replace' : 'missing',
driftKind: content ? 'replace' : 'missing',
size: version.size, sha256: version.sha256,
scriptId: script.id, scriptVersionId: version.id
});
@@ -176,13 +200,14 @@ async function scanManagedForServer(server: string): Promise<DriftEntry[]> {
const targetRel = configFileOf(server);
const abs = join(dest, targetRel);
const inv = getServerScript(server, targetRel);
const onDiskMatches = existsSync(abs) && inv?.sha256 === composed.sha256;
const content = await readExistingFile(abs);
const onDiskMatches = content?.equals(composed.content) && inv?.sha256 === composed.sha256;
if (!onDiskMatches) {
out.push({
server,
relPath: targetRel,
kind: 'config',
driftKind: existsSync(abs) ? 'replace' : 'missing',
driftKind: content ? 'replace' : 'missing',
size: composed.content.byteLength,
sha256: composed.sha256,
scriptId: baseline.id,
@@ -230,12 +255,8 @@ function composeExpectedCfg(
*/
export async function scanFleetDrift(): Promise<DriftEntry[]> {
const servers = listServers().map((s) => s.name);
const registered = new Set(servers);
const allDirs = new Set(await serverDirectories());
// Only touch server dirs that are both registered AND present on disk.
const eligible = servers.filter((n) => registered.has(n) && allDirs.has(n));
const out: DriftEntry[] = [];
for (const name of eligible) {
for (const name of servers) {
const rows = await scanServerDrift(name);
out.push(...rows);
}
@@ -271,6 +292,10 @@ async function describePlan(servers: string[], entries: DriftEntry[]): Promise<D
beforeSha256: before ? sha256Buffer(before) : null, afterSha256: after ? sha256Buffer(after) : null,
beforeSize: before?.byteLength ?? null, afterSize: after?.byteLength ?? null });
}
return buildPlan(servers, files);
}
function buildPlan(servers: string[], files: DriftPlanFile[]): DriftPlan {
files.sort((a, b) => `${a.server}:${a.kind}:${a.relPath}`.localeCompare(`${b.server}:${b.kind}:${b.relPath}`));
return { servers, files, scannedAt: new Date().toISOString(), fingerprint: sha256Buffer(Buffer.from(JSON.stringify({ servers, files }))) };
}
@@ -298,6 +323,62 @@ function deploymentIdNow(): string {
return `dep-${stamp}-drift-${randomUUID().slice(0, 8)}`;
}
export interface PreparedDriftCampaign {
plan: DriftPlan;
files: Array<{ file: DriftPlanFile; before: Buffer | null; after: Buffer | null }>;
}
/** Capture a rebuild's file evidence without changing the running instance. */
export async function prepareDriftCampaign(server: string): Promise<PreparedDriftCampaign> {
const servers = driftTargets([server]);
const entries = await scanServerDrift(server);
const files: PreparedDriftCampaign['files'] = [];
for (const entry of entries) {
const before = await readExistingFile(join(destOfKind(server, entry.kind), entry.relPath));
const after = entry.driftKind === 'untracked' ? null : await resolveManagedEntry(server, entry);
if (entry.driftKind !== 'untracked' && !after) throw new Error(`Cannot resolve promoted content for ${entry.relPath}`);
files.push({ before, after, file: { server, relPath: entry.relPath, kind: entry.kind, driftKind: entry.driftKind,
beforeSha256: before ? sha256Buffer(before) : null, afterSha256: after ? sha256Buffer(after) : null,
beforeSize: before?.byteLength ?? null, afterSize: after?.byteLength ?? null } });
}
await requireServerDirectory(server);
if (hasActiveBeta(server)) throw new Error('Active beta owns this server; reconciliation deferred');
return { plan: buildPlan(servers, files.map(({ file }) => file)), files };
}
/** Record the audited plan atomically with a queued rebuild, never a second restart. */
export function createDriftRebuildCampaign(
server: string, operationId: string, incidentId: string | number, prepared: PreparedDriftCampaign
): string | null {
if (!prepared.files.length) return null;
if (prepared.plan.servers.length !== 1 || prepared.plan.servers[0] !== server
|| prepared.files.some(({ file }) => file.server !== server)) throw new Error('Drift evidence belongs to another server');
if (hasActiveBeta(server)) throw new Error('Active beta owns this server; reconciliation deferred');
return db().transaction(() => {
if (!db().prepare("SELECT 1 FROM server_deployments WHERE id = ? AND server = ? AND action = 'rebuild'").get(operationId, server)) {
throw new Error('Drift campaign requires the matching rebuild operation');
}
const id = deploymentIdNow();
db().prepare("INSERT INTO deployments (id, action, started_at, triggered_by) VALUES (?, 'drift', ?, 'watchdog')")
.run(id, new Date().toISOString());
const insertFile = db().prepare(`INSERT INTO deployment_files
(deployment_id, server, verb, rel_path, sha256_before, sha256_after, content_before, content_after)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`);
for (const { file, before, after } of prepared.files) {
insertFile.run(id, server, after === null ? 'removed' : before === null ? 'added' : 'updated',
file.relPath, file.beforeSha256, file.afterSha256, before, after);
}
db().prepare(`INSERT INTO deployment_servers (deployment_id, server, files_changed, restart_state, provisioning_id)
VALUES (?, ?, ?, 'pending', ?)`).run(id, server, prepared.files.length, operationId);
db().prepare("INSERT INTO provisioning_sources (operation_id, kind, source_id, server) VALUES (?, 'campaign', ?, ?)")
.run(operationId, id, server);
db().prepare('INSERT INTO campaign_events (deployment_id, message) VALUES (?, ?)')
.run(id, `Watchdog ${incidentId}: captured drift evidence; rebuild ${operationId} will install the latest promoted setup. Files remain unchanged until provisioning rebuilds the instance.`);
emit({ type: 'deployment-opened', deploymentId: id, action: 'drift', servers: [server] });
return id;
})();
}
/**
* Detect drift on the requested servers (or the whole fleet) and either
* report the plan (dryRun) or remove untracked files and install managed
@@ -368,10 +449,19 @@ export async function runDriftDeployment(input: RunDriftInput = {}): Promise<Run
const affected: string[] = [];
for (const s of perServer) {
if (s.entries.length === 0) continue;
if (hasActiveBeta(s.server)) {
s.skipped.push('Active beta owns this server; reconciliation deferred');
continue;
}
for (const entry of s.entries) {
const dest = destOfKind(s.server, entry.kind);
const abs = join(dest, entry.relPath);
try {
await requireServerDirectory(s.server);
if (hasActiveBeta(s.server)) {
s.skipped.push(`${entry.relPath}: active beta owns this server; reconciliation deferred`);
continue;
}
const reviewed = plan?.files.find(file => file.server === s.server && file.kind === entry.kind && file.relPath === entry.relPath);
if (reviewed) {
const current = await readFile(abs).catch((error: NodeJS.ErrnoException) => { if (error.code === 'ENOENT') return null; throw error; });
@@ -383,6 +473,7 @@ export async function runDriftDeployment(input: RunDriftInput = {}): Promise<Run
if (entry.driftKind === 'untracked') {
// Delete a file inventory doesn't know about.
const content = await readFile(abs);
if (hasActiveBeta(s.server)) throw new Error('active beta owns this server; reconciliation deferred');
await rm(abs, { force: true });
removeScript(s.server, entry.relPath);
s.removed.push(entry.relPath);
@@ -403,6 +494,7 @@ export async function runDriftDeployment(input: RunDriftInput = {}): Promise<Run
throw err;
});
await mkdir(dirname(abs), { recursive: true });
if (hasActiveBeta(s.server)) throw new Error('active beta owns this server; reconciliation deferred');
await writeFile(abs, written);
recordScript(
s.server,
src/lib/server/fleet/servers.ts
+1
View File
@@ -68,6 +68,7 @@ export interface ServerRow {
komodo_name: string;
komodo_server: string | null;
container_state: string | null;
desired_state: 'running' | 'stopped';
game_port: number | null;
rcon_host: string | null;
game: string | null;
src/lib/server/fleet/status-evidence.ts
+46
View File
@@ -0,0 +1,46 @@
import { db } from '../db/index.js';
import { parseGameEvent } from '../../telemetry.js';
import { ONLINE_MAX_MS } from '../../server-freshness.js';
import { rconGetPlayerCount, rconStatusResult } from '../integrations/rcon.js';
export interface TelemetryEvidence { players: number; map: string; receivedAt: string }
/** Only validated, recently received reports can establish current game state. */
export function telemetryEvidence(server: string, after?: string | null): TelemetryEvidence | null {
const row = db().prepare('SELECT received_at, payload FROM server_telemetry WHERE server = ?').get(server) as
{ received_at: string; payload: string | null } | undefined;
if (!row?.payload) return null;
const received = Date.parse(row.received_at);
const age = Date.now() - received;
if (!Number.isFinite(age) || age < 0 || age > ONLINE_MAX_MS) return null;
if (after && (!Number.isFinite(Date.parse(after)) || received <= Date.parse(after))) return null;
try {
const event = parseGameEvent(JSON.parse(row.payload));
if (event.server !== server) return null;
return { players: event.players.length, map: event.map, receivedAt: row.received_at };
} catch { return null; }
}
/** Positive occupancy from either source always vetoes an empty-server claim. */
export async function probePlayerCount(server: string): Promise<number | null> {
const players = await rconGetPlayerCount(server).catch(() => null);
// Read after the network request: a player may join while RCON is pending.
const telemetry = telemetryEvidence(server);
if (players !== null && players > 0) return players;
if (telemetry && telemetry.players > 0) return telemetry.players;
return players === 0 || telemetry?.players === 0 ? 0 : null;
}
/** Komodo readiness remains a caller prerequisite. Old-process telemetry is excluded. */
export async function probeServerStatus(server: string, after?: string | null): Promise<
{ ok: true; players: number; map: string | null; source: 'rcon' | 'telemetry' } | { ok: false; reason: string }
> {
let reason = 'RCON unavailable';
try {
const rcon = await rconStatusResult(server);
if (rcon.ok) return { ok: true, players: Math.max(rcon.status.players, telemetryEvidence(server, after)?.players ?? 0), map: rcon.status.map, source: 'rcon' };
reason = rcon.reason;
} catch (err) { reason = (err as Error).message; }
const telemetry = telemetryEvidence(server, after);
return telemetry ? { ok: true, ...telemetry, source: 'telemetry' } : { ok: false, reason };
}
src/lib/server/pipeline/sweep.ts
+2 -1
View File
@@ -41,7 +41,8 @@ async function runSweep(): Promise<SweepResult> {
}
db().prepare(`UPDATE deployments SET finished_at = ? WHERE finished_at IS NULL AND NOT EXISTS (
SELECT 1 FROM deployment_servers WHERE deployment_id = deployments.id AND restart_state IN ('pending','failed','held'))`).run(nowIso());
db().prepare(`DELETE FROM deployments WHERE started_at < ? AND NOT EXISTS (
db().prepare(`DELETE FROM deployments WHERE started_at < ?
AND NOT EXISTS (SELECT 1 FROM watchdog_actions WHERE campaign_id = deployments.id) AND NOT EXISTS (
SELECT 1 FROM deployment_servers WHERE deployment_id = deployments.id AND restart_state IN ('pending','failed','held'))`)
.run(new Date(Date.now() - config.deployKeepDays * 86_400_000).toISOString());
emit({ type: 'sweep-tick', frozen: false });
src/lib/server/pipeline/worker.ts
+2
View File
@@ -14,6 +14,7 @@ import { STATUS_REFRESH_MS } from '../../server-freshness.js';
import { recordMetrics } from '../fleet/metrics.js';
import { selectedZombieMap } from '../../map-aliases.js';
import { maintenanceTick } from '../fleet/plutonium.js';
import { watchdogTick } from '../watchdog/index.js';
const TICK_MS = 15_000;
// Cap concurrent UDP RCON calls so a 15-server fleet doesn't blast the
@@ -36,6 +37,7 @@ async function reconcileRequesters(): Promise<void> {
try { await sweep(); } catch (err) { console.error('[worker] campaign error:', err); }
try { await betaTick(); } catch (err) { console.error('[worker] beta error:', err); }
try { await maintenanceTick(); } catch (err) { console.error('[worker] maintenance error:', err); }
try { await watchdogTick(); } catch (err) { console.error('[worker] watchdog error:', err); }
}
async function runTick(): Promise<void> {
src/lib/server/provisioning/index.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// Request/observation boundary used by APIs, campaigns, beta tests, and schedules.
// Execution is driven exclusively by the provisioning worker.
export { requestLifecycle, requestRestart, provisioningOperation, cancelProvisioning, releaseRestartRequest } from './requests.js';
export { requestLifecycle, requestRestart, requestWatchdogLifecycle, escalateWatchdogRestart, provisioningOperation, cancelProvisioning, releaseRestartRequest } from './requests.js';
export type { LifecycleOperation, RestartSource } from './requests.js';
export { queueCreateServer, destroyServer, previewCreation, creationQueue, nextAvailablePort, nextLobbyNumber } from './instances.js';
export { getDeployment, listDeployments, resumeVerification, resumeCleanup, resumeRebuild, cancelRemoval } from './history.js';
src/lib/server/provisioning/instances.ts
+20 -13
View File
@@ -1,4 +1,4 @@
import { rconGetPlayerCount, rconStatusResult } from '../integrations/rcon.js';
import { probePlayerCount, probeServerStatus } from '../fleet/status-evidence.js';
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';
@@ -229,7 +229,7 @@ export function queueCreateServer(input: CreateServerInput, releasingServers: st
}
/** Reserve the same identity and port before removing any part of the instance. */
export function queueRebuildServer(name: string): CreateServerResult {
export function queueRebuildServer(name: string, triggeredBy = 'ui'): CreateServerResult {
const server = getServer(name);
const game = gameOfServer(name);
const map = server?.map_slug && zombieMapBySlug(game, server.map_slug);
@@ -237,12 +237,12 @@ export function queueRebuildServer(name: string): CreateServerResult {
|| (server.host_dir && server.host_dir !== serverDir(name))) {
return { ok: false, kind: 'bad-request', message: 'Rebuild requires a managed server with its original map, directory, and port.' };
}
if (!config.komodo.targetServer || !config.rcon.password) return { ok: false, kind: 'unconfigured', message: 'Configure the Komodo target host and RCON before rebuilding.' };
if (!config.komodo.targetServer || (!config.rcon.password && !triggeredBy.startsWith('watchdog:'))) return { ok: false, kind: 'unconfigured', message: 'Configure the Komodo target host and RCON before rebuilding.' };
if (!getServerKey(name)) return { ok: false, kind: 'bad-request', message: `Save a server key for ${name} in Keys before rebuilding.` };
// Validate that the replacement setup can be composed before destruction.
const recipe = captureRecipe(game, map.slug);
return db().transaction(() => {
const id = beginDeployment(name, 'rebuild');
const id = beginDeployment(name, 'rebuild', triggeredBy);
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, server.game_port, 'zombie', map.slug, map.canonical);
@@ -436,6 +436,7 @@ async function createServerUnlocked(input: CreateServerInput, prepared: Prepared
// Step 5: execute/Deploy — actually start the container.
deploymentStage(operation, 'Starting container');
db().prepare('UPDATE server_deployments SET submitted_at = ? WHERE id = ?').run(new Date().toISOString(), operation);
const deployRes = await komodoDeploy(name);
if (!deployRes.ok) {
if (deployRes.kind === 'unreachable' || (deployRes.httpStatus !== undefined && deployRes.httpStatus >= 500)) {
@@ -541,7 +542,7 @@ async function runDestructions(): Promise<void> {
continue;
}
// A fresh successful zero is required; cached occupancy and timeouts are not evidence of emptiness.
let players = await rconGetPlayerCount(operation.server);
let players = await probePlayerCount(operation.server);
if (players === null) {
const health = await komodoInspectHealthy(operation.server);
if (['stopped','exited','created','dead'].includes(health.state?.toLowerCase() ?? '')) players = 0;
@@ -621,8 +622,13 @@ async function cleanupDestroyedServer(name: string, operation: string): Promise<
// Test evidence cascades from runs; inventory and snapshots cascade from servers.
try {
const tx = db().transaction((n: string) => {
db().prepare('DELETE FROM deployment_servers WHERE server = ?').run(n);
db().prepare('DELETE FROM deployment_files WHERE server = ?').run(n);
// The Watchdog's rebuild campaign observes this exact operation
// through the registry gap and retains its before/after file audit.
for (const table of ['deployment_servers', 'deployment_files']) {
db().prepare(`DELETE FROM ${table} WHERE server = ? AND deployment_id NOT IN (
SELECT campaign_id FROM watchdog_actions WHERE campaign_id IS NOT NULL)`)
.run(n);
}
db().prepare('DELETE FROM beta_servers WHERE server = ?').run(n);
db().prepare('DELETE FROM beta_server_scripts WHERE server = ?').run(n);
db().prepare('DELETE FROM beta_test_runs WHERE server = ?').run(n);
@@ -647,6 +653,7 @@ async function cleanupDestroyedServer(name: string, operation: string): Promise<
}
interface CreationJob {
deployment_id: string; server: string; port: number; game_mode: GameMode; map_slug: string; expected_map: string; triggered_by?: string;
submitted_at?: string | null;
}
async function provisionCreation(id: string): Promise<CreateServerResult> {
@@ -739,7 +746,7 @@ async function runCreations(): Promise<CreationOutcomes> {
// 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
const checking = db().prepare(`SELECT j.*, COALESCE(d.submitted_at, d.started_at) AS submitted_at 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;
@@ -751,14 +758,14 @@ async function runCreations(): Promise<CreationOutcomes> {
try {
const komodo = await komodoInspectHealthy(job.server);
if (!komodo.ok) { deploymentStage(id, 'Waiting for Komodo health: ' + komodo.reason); continue; }
const rcon = await rconStatusResult(job.server);
if (!rcon.ok) { deploymentStage(id, 'Waiting for RCON health: ' + rcon.reason); continue; }
const actualMap = rcon.status.map;
const status = await probeServerStatus(job.server, job.submitted_at);
if (!status.ok) { deploymentStage(id, 'Waiting for RCON or fresh telemetry health: ' + status.reason); continue; }
const actualMap = status.map;
if (actualMap !== job.expected_map) {
deploymentStage(id, 'Waiting for selected map: expected ' + job.expected_map + ', RCON reports ' + (actualMap ?? 'no map'));
deploymentStage(id, 'Waiting for selected map: expected ' + job.expected_map + ', ' + (status.source === 'rcon' ? 'RCON' : 'telemetry') + ' reports ' + (actualMap ?? 'no map'));
continue;
}
deploymentStage(id, 'Verified Komodo healthy, RCON healthy, and map ' + actualMap);
deploymentStage(id, 'Verified Komodo healthy, ' + (status.source === 'rcon' ? 'RCON healthy' : 'fresh telemetry') + ', and map ' + actualMap);
finishDeployment(id, { ok: true });
} catch (err) {
deploymentStage(id, 'Waiting for health verification: ' + (err as Error).message);
src/lib/server/provisioning/requests.ts
+118 -4
View File
@@ -1,14 +1,16 @@
import { db } from '../db/index.js';
import { komodoConfigured, rconConfigured } from '../config.js';
import { getServer } from '../fleet/servers.js';
import { beginDeployment, deploymentStage, recordDeploymentEvent } from './history.js';
import { beginDeployment, deploymentStage, finishDeployment, recordDeploymentEvent } from './history.js';
import { queueRebuildServer } from './instances.js';
import { restartRequestInFlight } from './restarts.js';
import { runtimeActions, type RuntimeAction } from '../../provisioning.js';
export interface LifecycleOperation {
id: string; server: string; action: string; state: string; stage: string;
submitted_at: string | null; finished_at: string | null; checked_at: string | null;
players: number | null; error: string | null;
started_at: string;
}
export interface RestartSource { kind: 'campaign' | 'beta' | 'maintenance'; id: string }
export function provisioningOperation(id: string): LifecycleOperation | undefined {
@@ -115,19 +117,131 @@ export function requestLifecycle(server: string, action: RuntimeAction) {
return db().transaction(() => {
const existing = db().prepare(`SELECT id, action FROM server_deployments WHERE server = ?
AND state IN ('queued','waiting','running','awaiting-health') ORDER BY rowid DESC LIMIT 1`).get(server) as { id: string; action: string } | undefined;
if (existing) return existing.action === action
? { ok: true as const, deploymentId: existing.id }
: { ok: false as const, status: 409, error: 'Another provisioning operation is already pending for this server.' };
if (existing) {
if (existing.action !== action) return { ok: false as const, status: 409, error: 'Another provisioning operation is already pending for this server.' };
db().prepare('UPDATE servers SET desired_state = ? WHERE name = ?').run(action === 'stop' ? 'stopped' : 'running', server);
return { ok: true as const, deploymentId: existing.id };
}
const row = getServer(server);
if (!row || row.destroyed_at) return { ok: false as const, status: 404, error: 'Server not found.' };
if (action === 'rebuild') {
const result = queueRebuildServer(server);
if (result.ok) db().prepare("UPDATE servers SET desired_state = 'running' WHERE name = ?").run(server);
return result.ok ? { ok: true as const, deploymentId: result.deploymentId }
: { ok: false as const, status: 400, error: result.message };
}
const id = beginDeployment(server, action);
db().prepare('UPDATE servers SET desired_state = ? WHERE name = ?').run(action === 'stop' ? 'stopped' : 'running', server);
db().prepare("UPDATE server_deployments SET state = 'waiting' WHERE id = ?").run(id);
deploymentStage(id, action === 'start' ? 'Waiting to start the container' : 'Waiting to verify the server is empty');
return { ok: true as const, deploymentId: id };
})();
}
type LifecycleRequestResult = { ok: true; deploymentId: string } | { ok: false; status: number; error: string };
function watchdogRequestGuard(server: string, incidentId: string): Extract<LifecycleRequestResult, { ok: false }> | null {
const incident = db().prepare("SELECT 1 FROM watchdog_incidents WHERE id = ? AND server = ? AND state IN ('open','recovering')").get(incidentId, server);
if (!incident) return { ok: false, status: 409, error: 'The Watchdog incident is no longer active.' };
if (db().prepare("SELECT 1 FROM settings WHERE key = 'watchdog_enabled' AND value = '0'").get()) {
return { ok: false, status: 409, error: 'Watchdog is paused.' };
}
const row = getServer(server);
if (!row || row.destroyed_at) return { ok: false, status: 404, error: 'Server not found.' };
if (row.desired_state === 'stopped') return { ok: false, status: 409, error: 'This server is intentionally stopped.' };
if (!komodoConfigured()) return { ok: false, status: 503, error: 'Configure Komodo before provisioning.' };
return null;
}
function existingWatchdogAction(incidentId: string, action: 'restart' | 'rebuild'): string | undefined {
return (db().prepare(`SELECT a.operation_id FROM watchdog_actions a JOIN watchdog_incidents i ON i.id = a.incident_id
WHERE a.incident_id = ? AND a.action = ? AND a.cycle = i.cycle`).get(incidentId, action) as { operation_id: string } | undefined)?.operation_id;
}
function queueWatchdogAction(server: string, action: 'restart' | 'rebuild', incidentId: string): LifecycleRequestResult {
let id: string;
if (action === 'rebuild') {
const result = queueRebuildServer(server, `watchdog:${incidentId}`);
if (!result.ok) return { ok: false, status: 400, error: result.message };
id = result.deploymentId;
} else {
id = beginDeployment(server, action, `watchdog:${incidentId}`);
db().prepare("UPDATE server_deployments SET state = 'waiting' WHERE id = ?").run(id);
deploymentStage(id, 'Waiting to verify the server is empty');
}
db().prepare("INSERT INTO provisioning_sources (operation_id, kind, source_id, server) VALUES (?, 'watchdog', ?, ?)").run(id, incidentId, server);
return { ok: true, deploymentId: id };
}
/** Watchdog submits durable work; Provisioning remains the only execution owner. */
export function requestWatchdogLifecycle(server: string, action: 'restart' | 'rebuild', incidentId: string): LifecycleRequestResult {
return db().transaction(() => {
const previous = existingWatchdogAction(incidentId, action);
if (previous) return { ok: true as const, deploymentId: previous };
const error = watchdogRequestGuard(server, incidentId);
if (error) return error;
const pending = db().prepare(`SELECT d.id, d.action, s.source_id FROM server_deployments d
LEFT JOIN provisioning_sources s ON s.operation_id = d.id AND s.kind = 'watchdog'
WHERE d.server = ? AND d.state IN ('queued','waiting','running','awaiting-health') ORDER BY d.rowid DESC LIMIT 1`)
.get(server) as { id: string; action: string; source_id: string | null } | undefined;
if (pending) return pending.source_id === incidentId && pending.action === action
? { ok: true as const, deploymentId: pending.id }
: { ok: false as const, status: 409, error: 'Another provisioning operation is already pending for this server.' };
const next = queueWatchdogAction(server, action, incidentId);
if (!next.ok) return next;
// An operator retry starts a new incident cycle. Keep the previous
// failed attempt's capacity, but let that new attempt use it instead
// of waiting forever behind its own incident's terminal operation.
const retained = db().prepare(`SELECT r.operation_id FROM restart_rollout r
JOIN server_deployments d ON d.id = r.operation_id
JOIN watchdog_actions a ON a.operation_id = d.id AND a.incident_id = ?
JOIN watchdog_incidents i ON i.id = a.incident_id
WHERE r.server = ? AND r.healthy_at IS NULL AND d.state IN ('failed','interrupted','cancelled')
AND a.cycle < i.cycle`).get(incidentId, server) as { operation_id: string } | undefined;
if (retained && !restartRequestInFlight(server)) {
db().prepare('UPDATE restart_rollout SET operation_id = ? WHERE server = ? AND operation_id = ? AND healthy_at IS NULL')
.run(next.deploymentId, server, retained.operation_id);
recordDeploymentEvent(retained.operation_id, 'Recovery capacity handed to explicit Watchdog retry ' + next.deploymentId);
recordDeploymentEvent(next.deploymentId, 'Retained recovery capacity from prior incident attempt ' + retained.operation_id);
}
return next;
})();
}
/** Atomically reserve a replacement before handing it the failed restart's capacity. */
export function escalateWatchdogRestart(incidentId: string, operationId: string, recoveryTimeoutMs = 300_000): LifecycleRequestResult {
return db().transaction(() => {
const previous = existingWatchdogAction(incidentId, 'rebuild');
if (previous) return { ok: true as const, deploymentId: previous };
const operation = provisioningOperation(operationId);
if (!operation || operation.action !== 'restart' || !['awaiting-health','failed'].includes(operation.state)
|| !db().prepare("SELECT 1 FROM provisioning_sources WHERE operation_id = ? AND kind = 'watchdog' AND source_id = ?").get(operationId, incidentId)) {
return { ok: false as const, status: 409, error: 'The restart is not eligible for Watchdog escalation.' };
}
const error = watchdogRequestGuard(operation.server, incidentId);
if (error) return error;
const pending = db().prepare(`SELECT d.id, d.action, s.source_id FROM server_deployments d
LEFT JOIN provisioning_sources s ON s.operation_id = d.id AND s.kind = 'watchdog'
WHERE d.server = ? AND d.id != ? AND d.state IN ('queued','waiting','running','awaiting-health') LIMIT 1`)
.get(operation.server, operationId) as { id: string; action: string; source_id: string | null } | undefined;
if (pending) return pending.action === 'rebuild' && pending.source_id === incidentId
? { ok: true as const, deploymentId: pending.id }
: { ok: false as const, status: 409, error: 'Another provisioning operation is already pending for this server.' };
const slot = db().prepare('SELECT operation_id, requested_at, healthy_at FROM restart_rollout WHERE server = ?').get(operation.server) as
{ operation_id: string | null; requested_at: string; healthy_at: string | null } | undefined;
const anchor = Date.parse(slot?.requested_at ?? operation.submitted_at ?? operation.started_at);
if (restartRequestInFlight(operation.server) || (slot && (slot.operation_id !== operationId || slot.healthy_at))
|| !Number.isFinite(anchor) || Date.now() - anchor < Math.max(300_000, recoveryTimeoutMs)) {
return { ok: false as const, status: 409, error: 'Waiting for the submitted restart to finish its recovery window.' };
}
const replacement = queueWatchdogAction(operation.server, 'rebuild', incidentId);
if (!replacement.ok) return replacement;
const reason = 'Watchdog restart recovery timed out; handed recovery to rebuild ' + replacement.deploymentId;
if (operation.state === 'awaiting-health') finishDeployment(operationId, { ok: false, message: reason });
else recordDeploymentEvent(operationId, reason);
if (slot) db().prepare('UPDATE restart_rollout SET operation_id = ?, error = ? WHERE server = ? AND operation_id = ? AND healthy_at IS NULL')
.run(replacement.deploymentId, reason, operation.server, operationId);
recordDeploymentEvent(replacement.deploymentId, 'Recovery handed off from restart ' + operationId + '; waiting for a fresh empty or stopped server');
return replacement;
})();
}
src/lib/server/provisioning/restarts.ts
+23 -11
View File
@@ -1,13 +1,14 @@
import { config } from '../config.js';
import { db } from '../db/index.js';
import { komodoInspectHealthy, komodoRestart, komodoRuntimeAction } from '../integrations/komodo.js';
import { rconStatusResult } from '../integrations/rcon.js';
import { probeServerStatus } from '../fleet/status-evidence.js';
import { deploymentStage, finishDeployment } from './history.js';
// Komodo queues restart work. Don't mistake the still-running old container
// for a completed restart when an API kick immediately follows the request.
const HEALTH_POLL_MS = 15_000;
const requestsInFlight = new Set<string>();
export const restartRequestInFlight = (server: string) => requestsInFlight.has(server);
interface RestartSlot {
server: string;
@@ -41,11 +42,14 @@ export function rolloutWaitReason(server: string, operationId = ''): string | nu
return 'Waiting for the current provisioning operation';
}
const slot = slotFor(server);
if (slot?.healthy_at === null) {
const ownHandoff = slot?.operation_id === operationId && !!db().prepare(`SELECT 1 FROM server_deployments d
WHERE d.id = ? AND d.action IN ('restart','rebuild') AND EXISTS (
SELECT 1 FROM provisioning_sources s WHERE s.operation_id = d.id AND s.kind = 'watchdog')`).get(operationId);
if (slot?.healthy_at === null && !ownHandoff) {
return `Waiting for the previous restart of this server to recover${slot.error ? `: ${slot.error}` : ''}`;
}
const { n } = db().prepare('SELECT COUNT(*) AS n FROM restart_rollout WHERE healthy_at IS NULL').get() as { n: number };
return n >= config.restartConcurrency ? 'Queued for a rolling restart; waiting for another server to recover' : null;
return n - (ownHandoff && slot?.healthy_at === null ? 1 : 0) >= config.restartConcurrency ? 'Queued for a rolling restart; waiting for another server to recover' : null;
}
export type RolloutRestartResult =
@@ -76,7 +80,7 @@ export async function restartInRollout(server: string, options: { operationId: s
const result = options.action === 'start' ? await komodoRuntimeAction(server, 'start') : await komodoRestart(server);
if (result.ok) {
db().prepare("UPDATE server_deployments SET state = 'awaiting-health' WHERE id = ?").run(operationId);
deploymentStage(operationId!, 'Waiting for container and RCON health');
deploymentStage(operationId!, 'Waiting for container and game health (RCON or fresh telemetry)');
return { kind: 'started' };
}
// These failures cannot have queued a remote restart. Other failures
@@ -115,6 +119,13 @@ async function refresh(): Promise<void> {
.all() as RestartSlot[];
for (const slot of slots) {
if (requestsInFlight.has(slot.server)) continue;
// Recovery capacity transferred to a rebuild remains reserved until
// teardown. Its health is verified by the creation worker, never here.
if (slot.operation_id && db().prepare(`SELECT 1 FROM server_deployments d WHERE d.id = ? AND (
(d.action = 'rebuild' AND d.state IN ('waiting','queued','running','awaiting-health'))
OR (d.action = 'restart' AND d.state = 'waiting' AND d.submitted_at IS NULL AND EXISTS (
SELECT 1 FROM provisioning_sources s WHERE s.operation_id = d.id AND s.kind = 'watchdog')))`)
.get(slot.operation_id)) continue;
if (Date.now() - Date.parse(slot.checked_at ?? slot.requested_at) < HEALTH_POLL_MS) continue;
let error: string | null = null;
let players: number | null = null;
@@ -123,22 +134,23 @@ async function refresh(): Promise<void> {
if (!health.ok) {
error = `unhealthy: ${health.reason}${health.snippet ? ` snippet=${health.snippet.slice(0, 400)}` : ''}`;
} else {
const rcon = await rconStatusResult(slot.server);
if (!rcon.ok) error = `rcon not answering: ${rcon.reason}`;
else players = rcon.status.players;
const status = await probeServerStatus(slot.server, slot.requested_at);
if (!status.ok) error = `rcon not answering and no fresh telemetry: ${status.reason}`;
else players = status.players;
}
} catch (err) {
error = `health check: ${(err as Error).message}`;
}
const now = new Date().toISOString();
db().prepare(`UPDATE restart_rollout SET checked_at = ?, healthy_at = ?, players = ?, error = ?
WHERE server = ? AND requested_at = ? AND healthy_at IS NULL`)
.run(now, error ? null : now, players, error, slot.server, slot.requested_at);
const checked = db().prepare(`UPDATE restart_rollout SET checked_at = ?, healthy_at = ?, players = ?, error = ?
WHERE server = ? AND requested_at = ? AND operation_id IS ? AND healthy_at IS NULL`)
.run(now, error ? null : now, players, error, slot.server, slot.requested_at, slot.operation_id);
if (!checked.changes) continue;
if (slot.operation_id && db().prepare("SELECT 1 FROM server_deployments WHERE id = ? AND state = 'awaiting-health'").get(slot.operation_id)) {
db().prepare('UPDATE server_deployments SET checked_at = ?, players = ? WHERE id = ?').run(now, players, slot.operation_id);
if (error) deploymentStage(slot.operation_id, 'Waiting for health: ' + error);
else {
deploymentStage(slot.operation_id, 'Verified container and RCON health');
deploymentStage(slot.operation_id, 'Verified container and game health (RCON or fresh telemetry)');
finishDeployment(slot.operation_id, { ok: true });
}
}
src/lib/server/provisioning/worker.ts
+20 -9
View File
@@ -4,7 +4,7 @@ import { getServer } from '../fleet/servers.js';
import { deploymentStage, finishDeployment, REMOVING_SERVER } from './history.js';
import { processCreations, processDestructions, rebuildServer } from './instances.js';
import { komodoInspectHealthy, komodoRuntimeAction } from '../integrations/komodo.js';
import { rconGetPlayerCount } from '../integrations/rcon.js';
import { probePlayerCount } from '../fleet/status-evidence.js';
import { refreshRestartRollout, restartInRollout, rolloutWaitReason } from './restarts.js';
import { cancelProvisioning } from './requests.js';
import type { RuntimeAction } from '../../provisioning.js';
@@ -12,14 +12,20 @@ import type { RuntimeAction } from '../../provisioning.js';
const stoppedStates = new Set(['stopped', 'exited', 'created', 'dead']);
export const isStopped = (state?: string) => stoppedStates.has(state?.toLowerCase() ?? '');
interface Source { kind: 'campaign' | 'beta' | 'maintenance'; source_id: string }
interface Source { kind: 'campaign' | 'beta' | 'maintenance' | 'watchdog'; source_id: string }
const sourcesFor = (id: string) => db().prepare('SELECT kind, source_id FROM provisioning_sources WHERE operation_id = ?').all(id) as Source[];
const priority = (sources: Source[]) => sources.some(s => s.kind === 'beta') ? 0 : sources.some(s => s.kind === 'campaign') ? 1 : sources.some(s => s.kind === 'maintenance') ? 3 : 2;
const priority = (sources: Source[]) => sources.some(s => s.kind === 'watchdog') ? 2 : sources.some(s => s.kind === 'beta') ? 0 : sources.some(s => s.kind === 'campaign') ? 1 : sources.some(s => s.kind === 'maintenance') ? 3 : 2;
/** Source work may hold or withdraw a request, but never executes it. */
function sourceEligibility(operation: Operation): 'ready' | 'held' | 'cancelled' {
const sources = sourcesFor(operation.id);
if (!sources.length) return 'ready';
// Watchdog intent gates every linked source, including its drift campaign.
for (const source of sources.filter(source => source.kind === 'watchdog')) {
if (!db().prepare("SELECT 1 FROM watchdog_incidents WHERE id = ? AND server = ? AND state IN ('open','recovering')").get(source.source_id, operation.server)
|| getServer(operation.server)?.desired_state === 'stopped') return 'cancelled';
if (db().prepare("SELECT 1 FROM settings WHERE key = 'watchdog_enabled' AND value = '0'").get()) return 'held';
}
let held = false;
for (const source of sources) {
if (source.kind === 'campaign') {
@@ -36,10 +42,10 @@ function sourceEligibility(operation: Operation): 'ready' | 'held' | 'cancelled'
if (!['awaiting-empty','restarting','awaiting-health'].includes(row.stage)) continue;
if (!row.outcome || ['active','passed'].includes(row.outcome)) return 'ready';
if (row.outcome === 'paused') held = true;
} else {
} else if (source.kind === 'maintenance') {
const row = db().prepare('SELECT state, provisioning_id FROM maintenance_jobs WHERE id = ?').get(source.source_id) as { state: string; provisioning_id: string } | undefined;
if (row?.provisioning_id === operation.id && ['queued','restarting'].includes(row.state)) return 'ready';
}
} else return 'ready';
}
return held ? 'held' : 'cancelled';
}
@@ -61,7 +67,8 @@ function blocker(server: string, id: string): string | null {
if (rank > 1 && db().prepare(`SELECT 1 FROM deployment_servers ds WHERE ds.server = ?
AND ds.restart_state IN ('pending','failed','held') AND (ds.provisioning_id IS NULL OR EXISTS (
SELECT 1 FROM server_deployments d WHERE d.id = ds.provisioning_id AND d.state IN ('queued','waiting','running','awaiting-health')))
LIMIT 1`).get(server)) return 'Waiting for campaign work on this server';
AND NOT EXISTS (SELECT 1 FROM watchdog_actions a WHERE a.operation_id = ? AND a.campaign_id = ds.deployment_id)
LIMIT 1`).get(server, id)) return 'Waiting for campaign work on this server';
return rolloutWaitReason(server, id);
}
@@ -93,14 +100,18 @@ async function run(): Promise<void> {
async function processOperation(operation: Operation): Promise<void> {
const { id, server, action } = operation;
// An explicit cleanup resume follows confirmed remote deletion. Complete
// that durable checkpoint even if the incident now needs attention or the
// Watchdog is paused; never strand the reserved replacement identity.
if (action === 'rebuild' && operation.stage === REMOVING_SERVER) { await rebuildServer(id); return; }
if (operation.state === 'waiting') {
const eligibility = sourceEligibility(operation);
if (eligibility === 'cancelled') { cancelProvisioning(id, 'Requester cancelled or replaced this operation'); return; }
if (eligibility === 'held') { deploymentStage(id, 'Waiting for the requester to resume'); return; }
}
if (action === 'rebuild' && operation.stage === REMOVING_SERVER) { await rebuildServer(id); return; }
if (!getServer(server)) { finishDeployment(id, { ok: false, message: 'Server is no longer registered.' }); return; }
if (!komodoConfigured() || !rconConfigured()) { deploymentStage(id, 'Waiting for Komodo and RCON configuration'); return; }
const watchdog = sourcesFor(id).some(source => source.kind === 'watchdog');
if (!komodoConfigured() || (!rconConfigured() && !watchdog)) { deploymentStage(id, 'Waiting for Komodo and RCON configuration'); return; }
if (operation.state === 'awaiting-health') {
const health = await komodoInspectHealthy(server);
if (isStopped(health.state)) {
@@ -114,7 +125,7 @@ async function processOperation(operation: Operation): Promise<void> {
if (reason) { deploymentStage(id, reason); return; }
const health = await komodoInspectHealthy(server);
if (action !== 'start' && !isStopped(health.state)) {
const players = await rconGetPlayerCount(server);
const players = await probePlayerCount(server);
db().prepare('UPDATE server_deployments SET players = ?, checked_at = ? WHERE id = ?').run(players, new Date().toISOString(), id);
if (players !== 0) {
deploymentStage(id, players === null ? 'Waiting for a fresh player count' : `Waiting for ${players} player(s) to leave`);
src/lib/server/watchdog/index.ts
+233
View File
@@ -0,0 +1,233 @@
import { randomUUID } from 'node:crypto';
import { db } from '../db/index.js';
import { komodoConfigured } from '../config.js';
import { getServer, listServers } from '../fleet/servers.js';
import { komodoInspectHealthy } from '../integrations/komodo.js';
import { rconStatusResult } from '../integrations/rcon.js';
import { telemetryEvidence } from '../fleet/status-evidence.js';
import { prepareDriftCampaign, createDriftRebuildCampaign, previewDrift } from '../fleet/drift.js';
import { requestWatchdogLifecycle, escalateWatchdogRestart } from '../provisioning/requests.js';
import type { WatchdogAction, WatchdogDetail, WatchdogEvidence, WatchdogIncident, WatchdogOverview, WatchdogState } from '../../watchdog.js';
export const WATCHDOG_SCAN_MS = 60_000;
export const WATCHDOG_FAILURE_MS = 120_000;
export const WATCHDOG_RECOVERY_MS = 300_000;
const iso = () => new Date().toISOString();
const pendingStates = ['queued', 'waiting', 'running', 'awaiting-health'];
const setting = (key: string) => (db().prepare('SELECT value FROM settings WHERE key = ?').get(key) as { value: string } | undefined)?.value ?? null;
function setSetting(key: string, value: string) { db().prepare('INSERT INTO settings(key,value) VALUES (?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value').run(key, value); }
export const watchdogEnabled = () => setting('watchdog_enabled') !== '0';
export function setWatchdogEnabled(enabled: boolean) { setSetting('watchdog_enabled', enabled ? '1' : '0'); }
function incident(id: string) { return db().prepare('SELECT * FROM watchdog_incidents WHERE id = ?').get(id) as WatchdogIncident | undefined; }
function active(server: string, kind: 'health' | 'drift') { return db().prepare("SELECT * FROM watchdog_incidents WHERE server = ? AND kind = ? AND state != 'resolved'").get(server, kind) as WatchdogIncident | undefined; }
function event(id: string, message: string) { db().prepare('INSERT INTO watchdog_events(incident_id,created_at,message) VALUES (?,?,?)').run(id, iso(), message); }
function transition(id: string, state: WatchdogState, summary: string) {
const previous = incident(id);
if (!previous || (previous.state === state && previous.summary === summary)) return;
db().prepare('UPDATE watchdog_incidents SET state=?, summary=?, finished_at=? WHERE id=?')
.run(state, summary, state === 'resolved' || state === 'dismissed' ? iso() : null, id);
event(id, summary);
}
function detect(server: string, kind: 'health' | 'drift', summary: string, details: WatchdogEvidence): WatchdogIncident {
const previous = active(server, kind);
if (previous) {
// Repeated UI scans must not manufacture a sustained failure.
const newSample = Date.now() - Date.parse(previous.last_seen_at) >= WATCHDOG_SCAN_MS;
const actionable = (details as WatchdogEvidence & { actionable?: boolean }).actionable;
const priorActionable = (JSON.parse(previous.details) as { actionable?: boolean }).actionable;
const reset = kind === 'health' && actionable === true && priorActionable !== true;
db().prepare('UPDATE watchdog_incidents SET details=?, last_seen_at=?, failures=?, first_seen_at=? WHERE id=?')
.run(JSON.stringify(details), newSample || reset ? iso() : previous.last_seen_at,
reset ? 1 : previous.failures + (newSample ? 1 : 0), reset ? iso() : previous.first_seen_at, previous.id);
if (previous.state === 'open') transition(previous.id, 'open', summary);
return incident(previous.id)!;
}
const id = randomUUID();
db().prepare(`INSERT INTO watchdog_incidents(id,server,kind,state,summary,details,first_seen_at,last_seen_at)
VALUES (?,?,?,'open',?,?,?,?)`).run(id, server, kind, summary, JSON.stringify(details), iso(), iso());
event(id, summary);
return incident(id)!;
}
function actions(id: string): WatchdogAction[] {
return db().prepare(`SELECT a.*, d.state AS operation_state, d.stage AS operation_stage, d.error AS operation_error
FROM watchdog_actions a LEFT JOIN server_deployments d ON d.id=a.operation_id
WHERE a.incident_id=? ORDER BY a.rowid`).all(id) as WatchdogAction[];
}
export function watchdogOverview(): WatchdogOverview {
return { enabled: watchdogEnabled(), lastScanAt: setting('watchdog_last_scan'), scanError: setting('watchdog_scan_error') || null,
incidents: db().prepare(`SELECT * FROM watchdog_incidents WHERE state != 'resolved' OR id IN (
SELECT id FROM watchdog_incidents WHERE state = 'resolved' ORDER BY first_seen_at DESC, rowid DESC LIMIT 100)
ORDER BY CASE WHEN state='resolved' THEN 1 ELSE 0 END, first_seen_at DESC, rowid DESC`).all() as WatchdogIncident[] };
}
export function watchdogDetail(id: string): WatchdogDetail | null {
const row = incident(id);
return row ? { incident: row, actions: actions(id), events: db().prepare('SELECT id,created_at,message FROM watchdog_events WHERE incident_id=? ORDER BY id').all(id) as WatchdogDetail['events'] } : null;
}
function hasPending(server: string) {
return !!db().prepare("SELECT 1 FROM server_deployments WHERE server=? AND state IN ('queued','waiting','running','awaiting-health') LIMIT 1").get(server);
}
function protectedWork(server: string): boolean {
return !!db().prepare(`SELECT 1 FROM beta_servers bs JOIN betas b ON b.id=bs.beta_id WHERE bs.server=? AND b.state IN ('running','ready-promote') LIMIT 1`).get(server)
|| !!db().prepare(`SELECT 1 FROM deployment_servers ds WHERE ds.server=? AND ds.restart_state IN ('pending','held','failed','manual')
AND NOT EXISTS (SELECT 1 FROM watchdog_actions a WHERE a.campaign_id=ds.deployment_id) LIMIT 1`).get(server);
}
function addAction(row: WatchdogIncident, operationId: string, action: 'restart' | 'rebuild', campaignId: string | null = null) {
db().prepare('INSERT INTO watchdog_actions(id,incident_id,action,operation_id,campaign_id,created_at,cycle) VALUES (?,?,?,?,?,?,?)')
.run(randomUUID(), row.id, action, operationId, campaignId, iso(), row.cycle);
transition(row.id, 'recovering', action === 'restart' ? 'Restart queued; waiting for safe recovery' : 'Rebuild queued with the latest promoted setup');
event(row.id, `${action === 'restart' ? 'Restart' : 'Rebuild'} requested: ${operationId}${campaignId ? `; campaign ${campaignId}` : ''}`);
}
export function updateWatchdogIncident(id: string, action: 'retry' | 'dismiss') {
return db().transaction(() => {
const row = incident(id);
if (!row) return { ok: false as const, status: 404, error: 'Watchdog incident not found.' };
if (actions(id).some(a => pendingStates.includes(a.operation_state ?? ''))) return { ok: false as const, status: 409, error: 'Provisioning is still active. Complete or cancel that operation first.' };
if (action === 'dismiss') {
if (!['open','attention'].includes(row.state)) return { ok: false as const, status: 409, error: 'Only detected incidents or incidents needing attention can be dismissed.' };
transition(id, 'dismissed', 'Dismissed by operator; automatic attempts remain suppressed until the condition clears');
} else {
if (!['attention','dismissed'].includes(row.state)) return { ok: false as const, status: 409, error: 'Only incidents needing attention or dismissed incidents can be retried.' };
if (!getServer(row.server)) return { ok: false as const, status: 409, error: 'The server is no longer registered; inspect its provisioning history.' };
db().prepare('UPDATE watchdog_incidents SET cycle=cycle+1, failures=1, first_seen_at=?, last_seen_at=? WHERE id=?').run(iso(), iso(), id);
transition(id, 'open', 'Operator requested a fresh recovery attempt; checking current conditions');
}
return { ok: true as const };
})();
}
async function reconcileActions(): Promise<void> {
const rows = db().prepare("SELECT * FROM watchdog_incidents WHERE state IN ('recovering','attention')").all() as WatchdogIncident[];
for (const row of rows) {
const action = actions(row.id).filter(a => a.cycle === row.cycle).at(-1);
if (!action) continue;
const operation = db().prepare('SELECT state,submitted_at,started_at,stage,error FROM server_deployments WHERE id=?').get(action.operation_id) as
{ state: string; submitted_at: string | null; started_at: string; stage: string; error: string | null } | undefined;
if (!operation) { transition(row.id, 'attention', 'Provisioning history is missing; inspect the server before retrying'); continue; }
if (['done','warning'].includes(operation.state)) {
if (row.kind === 'drift') {
try {
const plan = await previewDrift([row.server]);
if (incident(row.id)?.cycle !== row.cycle) continue;
if (plan.files.length) { transition(row.id, 'attention', 'Rebuild completed but drift remains; inspect the campaign and server files'); continue; }
} catch (err) { transition(row.id, 'attention', `Cannot verify rebuilt files: ${(err as Error).message}`); continue; }
}
transition(row.id, 'resolved', action.action === 'restart' ? 'Server recovered after restart' : 'Server recovered after rebuild');
continue;
}
if (row.state === 'attention') continue; // Exhausted attempts stay latched, including across app restarts.
if (['cancelled','interrupted'].includes(operation.state)) {
transition(row.id, 'attention', `Provisioning ${operation.state}; inspect the recorded operation before retrying`); continue;
}
if (action.action === 'rebuild') {
if (operation.state === 'failed') transition(row.id, 'attention', `Rebuild failed: ${operation.error ?? operation.stage}`);
else if (operation.state === 'awaiting-health' && Date.now() - Date.parse(operation.submitted_at ?? operation.started_at) >= WATCHDOG_RECOVERY_MS)
transition(row.id, 'attention', 'Rebuild has not recovered within five minutes; provisioning continues verification');
continue;
}
if (!['awaiting-health','failed'].includes(operation.state)) continue;
if (Date.now() - Date.parse(operation.submitted_at ?? operation.started_at) < WATCHDOG_RECOVERY_MS) continue;
if (!watchdogEnabled()) continue;
const current = getServer(row.server);
if (!current || current.desired_state === 'stopped' || protectedWork(row.server)) continue;
const result = db().transaction(() => {
const next = escalateWatchdogRestart(row.id, action.operation_id, WATCHDOG_RECOVERY_MS);
if (next.ok) addAction(row, next.deploymentId, 'rebuild');
return next;
})();
if (!result.ok && result.status !== 409 && result.status !== 503) transition(row.id, 'attention', `Could not queue rebuild: ${result.error}`);
}
}
interface Observation { healthy: boolean; actionable: boolean; reason: string; containerState: string | null }
async function observe(server: string): Promise<Observation> {
const health = await komodoInspectHealthy(server);
const state = health.state?.toLowerCase() ?? null;
if (!health.ok) return { healthy: false, actionable: !!state && ['stopped','exited','created','dead','unhealthy','restarting','starting'].includes(state), reason: health.reason, containerState: state };
if (telemetryEvidence(server)) return { healthy: true, actionable: false, reason: 'Komodo healthy with fresh telemetry', containerState: state };
const rcon = await rconStatusResult(server);
if (rcon.ok || telemetryEvidence(server)) return { healthy: true, actionable: false, reason: 'Komodo healthy with fresh game status', containerState: state };
return { healthy: false, actionable: true, reason: `Game status unavailable: ${rcon.reason}; waiting for player-count verification before recovery`, containerState: state };
}
async function scanServer(server: string): Promise<void> {
const row = getServer(server);
if (!row || row.desired_state === 'stopped') return;
if (hasPending(server) || protectedWork(server)) return;
const observation = await observe(server);
// Probes can overlap an API request; recheck ownership before queuing anything.
if (hasPending(server) || protectedWork(server) || getServer(server)?.desired_state === 'stopped') return;
if (observation.healthy) {
const current = active(server, 'health');
if (current) transition(current.id, 'resolved', 'Fresh game status and Komodo health verified');
} else {
const problem = detect(server, 'health', observation.reason, observation);
const startup = ['starting','restarting'].includes(observation.containerState ?? '');
const oldEnough = Date.now() - Date.parse(problem.first_seen_at) >= (startup ? WATCHDOG_RECOVERY_MS : WATCHDOG_FAILURE_MS);
if (problem.state === 'open' && observation.actionable && oldEnough && problem.failures >= 3 && watchdogEnabled()) {
const result = db().transaction(() => {
const next = requestWatchdogLifecycle(server, 'restart', problem.id);
if (next.ok) addAction(problem, next.deploymentId, 'restart');
return next;
})();
if (!result.ok && result.status !== 409 && result.status !== 503) transition(problem.id, 'attention', result.error);
}
// Unknown container health is an observation failure, not proof that a rebuild is safe.
if (!observation.actionable || hasPending(server)) return;
}
const prepared = await prepareDriftCampaign(server);
if (hasPending(server) || protectedWork(server) || getServer(server)?.desired_state === 'stopped') return;
if (!prepared.plan.files.length) {
const current = active(server, 'drift');
if (current) transition(current.id, 'resolved', 'Server files match the promoted setup');
return;
}
const problem = detect(server, 'drift', `${prepared.plan.files.length} file difference(s) from the promoted setup`, { files: prepared.plan.files, fingerprint: prepared.plan.fingerprint });
if (problem.state !== 'open' || !watchdogEnabled()) return;
const result = db().transaction(() => {
const next = requestWatchdogLifecycle(server, 'rebuild', problem.id);
if (next.ok) {
const campaignId = createDriftRebuildCampaign(server, next.deploymentId, problem.id, prepared);
addAction(problem, next.deploymentId, 'rebuild', campaignId);
}
return next;
})();
if (!result.ok && result.status !== 409 && result.status !== 503) transition(problem.id, 'attention', `Could not queue drift rebuild: ${result.error}`);
}
let inFlight: Promise<void> | null = null;
export function watchdogTick(forceScan = false): Promise<void> {
if (inFlight) return inFlight;
inFlight = run(forceScan).finally(() => { inFlight = null; });
return inFlight;
}
async function run(forceScan: boolean): Promise<void> {
await reconcileActions();
const lastScan = setting('watchdog_last_scan');
if (!forceScan && lastScan && Date.now() - Date.parse(lastScan) < WATCHDOG_SCAN_MS) return;
setSetting('watchdog_last_scan', iso());
if (!komodoConfigured()) {
setSetting('watchdog_scan_error', 'Configure Komodo before automatic recovery. RCON or fresh telemetry verifies game status.'); return;
}
const errors: string[] = [];
const servers = listServers();
let cursor = 0;
async function pump() {
while (cursor < servers.length) {
const server = servers[cursor++];
try { await scanServer(server.name); }
catch (err) { errors.push(`${server.name}: ${(err as Error).message}`); }
}
}
await Promise.all(Array.from({ length: Math.min(4, servers.length) }, pump));
// Preserve incidents for temporarily absent rebuild instances, but close retired registrations.
const abandoned = db().prepare("SELECT * FROM watchdog_incidents WHERE state != 'resolved'").all() as WatchdogIncident[];
for (const row of abandoned) {
if (getServer(row.server) || hasPending(row.server)) continue;
const last = db().prepare('SELECT action,state FROM server_deployments WHERE server=? ORDER BY rowid DESC LIMIT 1').get(row.server) as { action: string; state: string } | undefined;
// An absent registry can also mean a failed rebuild. Only a completed
// destruction retires that failure without requiring inspection.
if (['open','dismissed'].includes(row.state) || (last?.action === 'destroy' && ['done','warning'].includes(last.state)))
transition(row.id, 'resolved', 'Server is no longer registered');
}
setSetting('watchdog_scan_error', errors.join('\n'));
}