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

feat: unify recovery, updates, and schedules under Maintenance

Browse Source
This commit is contained in:
josh
2026-09-27 16:00:18 -04:00
parent 8ea696417f
commit fc46cb9da4
42 changed files with 1256 additions and 833 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/db/index.ts
+4
View File
@@ -310,6 +310,10 @@ export function db(): Database.Database {
for (const table of ['deployment_servers', 'beta_servers', 'maintenance_jobs']) {
migrateTable(handle, table, [{ name: 'provisioning_id', ddl: `ALTER TABLE ${table} ADD COLUMN provisioning_id TEXT` }]);
}
migrateTable(handle, 'maintenance_jobs', [
{ name: 'kind', ddl: "ALTER TABLE maintenance_jobs ADD COLUMN kind TEXT NOT NULL DEFAULT 'legacy'" },
{ name: 'origin', ddl: "ALTER TABLE maintenance_jobs ADD COLUMN origin TEXT NOT NULL DEFAULT 'manual'" }
]);
migrateServers(handle);
backfillDesiredState(handle);
// Column additions must run BEFORE any statement that references the
src/lib/server/fleet/overview.ts
+9
View File
@@ -64,6 +64,15 @@ export function fleetOperations(now = Date.now()): FleetOperation[] {
state: row.stage === 'failed' ? 'attention' : 'active', label: stages[row.stage] ?? row.stage,
detail: row.error ?? recovering.get(row.server)?.error ?? `Beta ${row.id}`, href: `/betas/${row.id}` });
}
const maintenance = handle.prepare(`SELECT m.* FROM maintenance_jobs m WHERE m.state = 'failed'
AND NOT EXISTS (SELECT 1 FROM maintenance_jobs n WHERE n.server = m.server AND n.kind = m.kind AND n.id > m.id)`)
.all() as { id: number; server: string; kind: string; detail: string | null; requested_at: string }[];
for (const job of maintenance) {
operations.push({ key: `maintenance:${job.id}`, group: `maintenance:${job.id}`, kind: 'maintenance',
server: job.server, state: 'attention', startedAt: job.requested_at,
label: job.kind === 'update' ? 'Update verification needs attention' : 'Maintenance needs attention',
detail: job.detail ?? 'Inspect the maintenance result', href: `/maintenance?view=activity&server=${encodeURIComponent(job.server)}#job-${job.id}` });
}
for (const slot of slots) {
if (representedRecovery.has(slot.server)) continue;
operations.push({ key: `recovery:${slot.server}:${slot.requested_at}`, group: `recovery:${slot.server}`,
src/lib/server/fleet/plutonium.ts
+2 -156
View File
@@ -1,156 +1,2 @@
import { db } from '../db/index.js';
import { config } from '../config.js';
import { getServer, listServers } from './servers.js';
import { rconPlutoniumVersion } from '../integrations/rcon.js';
import { requestRestart, cancelProvisioning, restartRecoveryPending } from '../provisioning/index.js';
import { nextMaintenanceAt, parsePlutoniumManifest, PLUTONIUM_MANIFEST_URL } from '../../plutonium.js';
import type { MaintenanceJob, MaintenanceSchedule, PlutoniumRelease, PlutoniumStatus, PlutoniumVersion } from '../../plutonium.js';
const HOUR = 60 * 60 * 1000;
const nowIso = () => new Date().toISOString();
function readSetting<T>(key: string, fallback: T): T {
const row = db().prepare('SELECT value FROM settings WHERE key = ?').get(key) as { value: string } | undefined;
return row ? JSON.parse(row.value) : fallback;
}
function writeSetting(key: string, value: unknown): void {
db().prepare('INSERT INTO settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value')
.run(key, JSON.stringify(value));
}
function release(): PlutoniumRelease {
return readSetting('plutonium_release', { revision: null, checkedAt: null, error: null });
}
function schedule(): MaintenanceSchedule {
// Legacy schedules were explicitly UTC. Preserve their timing until saved.
return { timeZone: 'UTC', ...readSetting('plutonium_schedule', { enabled: false, time: '06:00', nextAt: null }) };
}
export function saveMaintenanceSchedule(enabled: boolean, time: string, timeZone: string): void {
const nextAt = nextMaintenanceAt(time, new Date(), timeZone);
writeSetting('plutonium_schedule', { enabled, time, timeZone, nextAt: enabled ? nextAt : null });
}
export function plutoniumStatus(): PlutoniumStatus {
const versions = db().prepare('SELECT * FROM plutonium_versions').all() as PlutoniumVersion[];
return {
release: release(), schedule: schedule(),
servers: listServers().map(s => versions.find(v => v.server === s.name)
?? { server: s.name, revision: null, checked_at: null, error: null }),
jobs: db().prepare(`SELECT * FROM maintenance_jobs WHERE state IN ('queued','restarting')
OR id IN (SELECT id FROM maintenance_jobs ORDER BY id DESC LIMIT 100) ORDER BY id DESC`).all() as MaintenanceJob[]
};
}
async function probeVersion(server: string): Promise<number | null> {
const recovering = restartRecoveryPending(server);
const result = recovering ? { revision: null, error: 'Waiting for restart recovery.' }
: await rconPlutoniumVersion(server).catch(() => ({ revision: null, error: 'Version check failed.' }));
// Clear the old revision on failure: it is no longer evidence of what is running.
db().prepare(`INSERT INTO plutonium_versions (server, revision, checked_at, error) VALUES (?, ?, ?, ?)
ON CONFLICT(server) DO UPDATE SET revision = excluded.revision, checked_at = excluded.checked_at, error = excluded.error`)
.run(server, result.revision, nowIso(), result.error);
return result.revision;
}
let checking: Promise<void> | null = null;
export function checkPlutonium(): Promise<void> {
if (checking) return checking;
checking = check().finally(() => { checking = null; });
return checking;
}
async function check(): Promise<void> {
const previous = release();
try {
const response = await fetch(PLUTONIUM_MANIFEST_URL, { signal: AbortSignal.timeout(config.httpTimeoutMs), redirect: 'error' });
if (!response.ok) throw new Error(`Updater manifest request failed (HTTP ${response.status}).`);
const body = await response.text();
if (body.length > 2_000_000) throw new Error('Updater manifest response was too large.');
writeSetting('plutonium_release', { revision: parsePlutoniumManifest(JSON.parse(body)), checkedAt: nowIso(), error: null });
} catch (err) {
writeSetting('plutonium_release', { ...previous, checkedAt: nowIso(), error: (err as Error).message });
}
const servers = listServers();
for (let i = 0; i < servers.length; i += 4) {
await Promise.all(servers.slice(i, i + 4).map(s => probeVersion(s.name)));
}
}
export function queueMaintenance(servers: string[]): number {
const names = new Set(listServers().map(s => s.name));
if (!servers.length || servers.some(s => !names.has(s))) throw new Error('Choose existing fleet servers.');
return db().transaction(() => {
const latest = release();
// An old cached release is still displayed, but never used as a fresh target.
const target = !latest.error && latest.checkedAt && Date.now() - Date.parse(latest.checkedAt) < HOUR
? latest.revision : null;
let added = 0;
for (const server of new Set(servers)) {
added += db().prepare(`INSERT OR IGNORE INTO maintenance_jobs (server, target_revision, requested_at)
VALUES (?, ?, ?)`).run(server, target, nowIso()).changes;
const job = db().prepare("SELECT id FROM maintenance_jobs WHERE server = ? AND state IN ('queued','restarting')").get(server) as { id: number };
requestRestart(server, { kind: 'maintenance', id: String(job.id) });
}
return added;
})();
}
export function cancelMaintenance(id: number): boolean {
return db().transaction(() => {
const job = db().prepare("SELECT provisioning_id FROM maintenance_jobs WHERE id = ? AND state = 'queued'").get(id) as { provisioning_id: string | null } | undefined;
if (!job) return false;
if (job.provisioning_id && !cancelProvisioning(job.provisioning_id)) return false;
return db().prepare(`UPDATE maintenance_jobs SET state = 'cancelled', finished_at = ?, detail = 'Cancelled by operator'
WHERE id = ? AND state = 'queued'`).run(nowIso(), id).changes > 0;
})();
}
function finish(job: MaintenanceJob, state: 'done' | 'failed' | 'cancelled', detail: string) {
db().prepare('UPDATE maintenance_jobs SET state = ?, detail = ?, finished_at = ? WHERE id = ?')
.run(state, detail, nowIso(), job.id);
}
function hold(job: MaintenanceJob, detail: string) {
db().prepare('UPDATE maintenance_jobs SET detail = ? WHERE id = ?').run(detail, job.id);
}
let ticking: Promise<void> | null = null;
export function maintenanceTick(): Promise<void> {
if (ticking) return ticking;
ticking = runTick().finally(() => { ticking = null; });
return ticking;
}
async function runTick(): Promise<void> {
const latest = release();
if (!latest.checkedAt || Date.now() - Date.parse(latest.checkedAt) >= HOUR) await checkPlutonium();
db().transaction(() => {
const plan = schedule();
if (plan.enabled && plan.nextAt && plan.nextAt <= nowIso()) {
const names = listServers().map(s => s.name);
if (names.length) queueMaintenance(names);
// Coalesce missed days after downtime into one run; no restart backlog.
writeSetting('plutonium_schedule', { ...plan, nextAt: nextMaintenanceAt(plan.time, new Date(), plan.timeZone) });
}
})();
const jobs = db().prepare(`SELECT * FROM maintenance_jobs WHERE state IN ('queued','restarting') ORDER BY id`).all() as MaintenanceJob[];
for (const job of jobs) {
try { await processJob(job); }
catch { hold(job, 'Maintenance check failed; waiting for the next check.'); }
}
db().prepare(`DELETE FROM maintenance_jobs WHERE finished_at < ? AND state IN ('done','failed','cancelled')`)
.run(new Date(Date.now() - 30 * 24 * HOUR).toISOString());
}
async function processJob(job: MaintenanceJob): Promise<void> {
const server = getServer(job.server);
if (!server || server.destroyed_at) return finish(job, 'cancelled', 'Server removed.');
const operation = requestRestart(job.server, { kind: 'maintenance', id: String(job.id) });
if (['failed','interrupted','cancelled'].includes(operation.state)) {
return finish(job, operation.state === 'cancelled' ? 'cancelled' : 'failed', operation.error ?? operation.stage);
}
if (operation.state !== 'done') {
if (operation.submitted_at) {
db().prepare("UPDATE maintenance_jobs SET state = 'restarting', restarted_at = ? WHERE id = ?").run(operation.submitted_at, job.id);
db().prepare('UPDATE plutonium_versions SET revision = NULL, error = ? WHERE server = ?').run('Waiting for restart recovery.', job.server);
}
return hold(job, operation.stage);
}
const revision = await probeVersion(job.server);
if (job.target_revision && (revision === null || revision < job.target_revision)) {
return finish(job, 'failed', `Server recovered, but r${job.target_revision} was not verified (${revision === null ? 'version unknown' : `running r${revision}`}). Check Plutainer auto-update settings.`);
}
return finish(job, 'done', `Server recovered; ${revision === null ? 'running revision unknown' : `running r${revision}`}.`);
}
// Compatibility for existing jobs and API clients.
export * from "../maintenance/runtime.js";
src/lib/server/integrations/plutonium.ts
+76
View File
@@ -0,0 +1,76 @@
import { db } from '../db/index.js';
import { config } from '../config.js';
import { getServer, listServers } from '../fleet/servers.js';
import { latestGameSubmission } from '../fleet/status-evidence.js';
import { rconPlutoniumVersion } from './rcon.js';
import { komodoInspectHealthy } from './komodo.js';
import { restartRecoveryPending } from '../provisioning/index.js';
import { parsePlutoniumManifest, PLUTONIUM_MANIFEST_URL } from '../../plutonium.js';
import type { PlutoniumRelease, PlutoniumVersion } from '../../plutonium.js';
import { readSetting, writeSetting } from '../maintenance/settings.js';
export const HOUR = 60 * 60 * 1000;
const nowIso = () => new Date().toISOString();
export function release(): PlutoniumRelease {
return readSetting('plutonium_release', { revision: null, checkedAt: null, error: null });
}
export function freshRelease(): number | null {
const latest = release();
const age = latest.checkedAt ? Date.now() - Date.parse(latest.checkedAt) : NaN;
return !latest.error && age >= 0 && age < HOUR ? latest.revision : null;
}
export function runningVersions(): PlutoniumVersion[] {
const versions = db().prepare('SELECT * FROM plutonium_versions').all() as PlutoniumVersion[];
const target = freshRelease();
return listServers().map(server => {
const value = versions.find(v => v.server === server.name)
?? { server: server.name, revision: null, checked_at: null, error: null };
const age = value.checked_at ? Date.now() - Date.parse(value.checked_at) : NaN;
const epoch = latestGameSubmission(server.name);
const stopped = server.desired_state === 'stopped';
const valid = !value.error && age >= 0 && age < HOUR && (!epoch || value.checked_at! > epoch) && !restartRecoveryPending(server.name);
const revision = valid && !stopped ? value.revision : null;
return { ...value, revision, stopped,
update_status: stopped ? 'stopped' : revision === null || target === null ? 'unknown'
: revision < target ? 'outdated' : revision > target ? 'ahead' : 'current' };
});
}
export async function probeVersion(server: string): Promise<number | null> {
const epoch = latestGameSubmission(server);
const recovering = restartRecoveryPending(server);
const health = !recovering && getServer(server)?.desired_state !== 'stopped'
? await komodoInspectHealthy(server).catch(() => ({ ok: false })) : { ok: false };
let result = !health.ok ? { revision: null, error: 'Waiting for a healthy running server.' }
: await rconPlutoniumVersion(server).catch(() => ({ revision: null, error: 'Version check failed.' }));
if (!getServer(server)) return null;
if (epoch !== latestGameSubmission(server) || restartRecoveryPending(server)) {
result = { revision: null, error: 'Waiting for restart recovery.' };
}
// Clear the old revision on failure: it is no longer evidence of what is running.
db().prepare(`INSERT INTO plutonium_versions (server, revision, checked_at, error) VALUES (?, ?, ?, ?)
ON CONFLICT(server) DO UPDATE SET revision = excluded.revision, checked_at = excluded.checked_at, error = excluded.error`)
.run(server, result.revision, nowIso(), result.error);
return result.revision;
}
let checking: Promise<void> | null = null;
export function checkPlutonium(): Promise<void> {
if (checking) return checking;
checking = check().finally(() => { checking = null; });
return checking;
}
async function check(): Promise<void> {
const previous = release();
try {
const response = await fetch(PLUTONIUM_MANIFEST_URL, { signal: AbortSignal.timeout(config.httpTimeoutMs), redirect: 'error' });
if (!response.ok) throw new Error(`Updater manifest request failed (HTTP ${response.status}).`);
const body = await response.text();
if (body.length > 2_000_000) throw new Error('Updater manifest response was too large.');
writeSetting('plutonium_release', { revision: parsePlutoniumManifest(JSON.parse(body)), checkedAt: nowIso(), error: null });
} catch (err) {
writeSetting('plutonium_release', { ...previous, checkedAt: nowIso(), error: (err as Error).message });
}
const servers = listServers();
for (let i = 0; i < servers.length; i += 4) {
await Promise.all(servers.slice(i, i + 4).map(s => probeVersion(s.name)));
}
}
src/lib/server/maintenance/index.ts
+13
View File
@@ -0,0 +1,13 @@
import { maintenancePolicies } from './policies.js';
import { watchdogOverview, watchdogTick } from './recovery.js';
import { maintenanceTick, plutoniumStatus } from './runtime.js';
import type { MaintenanceOverview } from '../../maintenance.js';
export function maintenanceOverview(): MaintenanceOverview {
return { policies: maintenancePolicies(), recovery: watchdogOverview(), runtime: plutoniumStatus() };
}
export async function runMaintenance(): Promise<void> {
// One failing observer must not prevent the other capabilities from reconciling.
const results = await Promise.allSettled([maintenanceTick(), watchdogTick()]);
for (const result of results) if (result.status === 'rejected') console.error('[maintenance]', result.reason);
}
src/lib/server/maintenance/policies.ts
+25
View File
@@ -0,0 +1,25 @@
import { db } from '../db/index.js';
import type { MaintenancePolicy, MaintenancePolicies } from '../../maintenance.js';
function setting(key: string): string | undefined {
return (db().prepare('SELECT value FROM settings WHERE key = ?').get(key) as { value: string } | undefined)?.value;
}
function save(key: string, value: boolean) {
db().prepare('INSERT INTO settings(key,value) VALUES (?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value').run(key, value ? '1' : '0');
}
export function maintenancePolicies(): MaintenancePolicies {
// Preserve the old recovery switch without opting anyone into runtime upgrades.
const legacy = setting('watchdog_enabled') !== '0';
return {
paused: setting('maintenance_paused') === '1',
health: setting('maintenance_health') === undefined ? legacy : setting('maintenance_health') === '1',
drift: setting('maintenance_drift') === undefined ? legacy : setting('maintenance_drift') === '1',
updates: setting('maintenance_updates') === '1'
};
}
export function maintenanceAllowed(policy?: MaintenancePolicy): boolean {
const policies = maintenancePolicies();
return !policies.paused && (!policy || policies[policy]);
}
export function setMaintenancePolicy(policy: MaintenancePolicy, enabled: boolean) { save(`maintenance_${policy}`, enabled); }
export function pauseMaintenance(paused: boolean) { save('maintenance_paused', paused); }
src/lib/server/maintenance/recovery.ts
+242
View File
@@ -0,0 +1,242 @@
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';
import { maintenanceAllowed, setMaintenancePolicy } from './policies.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 = () => maintenanceAllowed('health') || maintenanceAllowed('drift');
export function setWatchdogEnabled(enabled: boolean) {
setSetting('watchdog_enabled', enabled ? '1' : '0');
setMaintenancePolicy('health', enabled);
setMaintenancePolicy('drift', enabled);
}
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 i.*, (SELECT d.stage FROM watchdog_actions a
JOIN server_deployments d ON d.id = a.operation_id WHERE a.incident_id = i.id AND a.cycle = i.cycle
ORDER BY a.rowid DESC LIMIT 1) AS operation_stage
FROM watchdog_incidents i 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 (!maintenanceAllowed(row.kind)) 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 && maintenanceAllowed('health')) {
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' || !maintenanceAllowed('drift')) 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'));
}
src/lib/server/maintenance/runtime.ts
+139
View File
@@ -0,0 +1,139 @@
import { db } from '../db/index.js';
import { getServer, listServers } from '../fleet/servers.js';
import { requestRestart, cancelProvisioning } from '../provisioning/index.js';
import { nextMaintenanceAt } from '../../plutonium.js';
import type { MaintenanceJob, MaintenanceSchedule, PlutoniumStatus } from '../../plutonium.js';
import { readSetting, writeSetting } from './settings.js';
import { checkPlutonium, probeVersion, release, freshRelease, runningVersions, HOUR } from '../integrations/plutonium.js';
import { maintenanceAllowed } from './policies.js';
export { checkPlutonium } from '../integrations/plutonium.js';
const nowIso = () => new Date().toISOString();
function schedule(): MaintenanceSchedule {
// Legacy schedules were explicitly UTC. Preserve their timing until saved.
return { timeZone: 'UTC', ...readSetting('plutonium_schedule', { enabled: false, time: '06:00', nextAt: null }) };
}
export function saveMaintenanceSchedule(enabled: boolean, time: string, timeZone: string): void {
const nextAt = nextMaintenanceAt(time, new Date(), timeZone);
writeSetting('plutonium_schedule', { enabled, time, timeZone, nextAt: enabled ? nextAt : null });
}
export function plutoniumStatus(): PlutoniumStatus {
return {
release: release(), schedule: schedule(), servers: runningVersions(),
jobs: db().prepare(`SELECT m.*, d.stage AS operation_stage, d.state AS operation_state
FROM maintenance_jobs m LEFT JOIN server_deployments d ON d.id = m.provisioning_id
WHERE m.state IN ('queued','restarting')
OR (m.state = 'failed' AND NOT EXISTS (SELECT 1 FROM maintenance_jobs n WHERE n.server = m.server AND n.kind = m.kind AND n.id > m.id))
OR m.id IN (SELECT id FROM maintenance_jobs ORDER BY id DESC LIMIT 100) ORDER BY m.id DESC`).all() as MaintenanceJob[]
};
}
export function queueMaintenance(servers: string[], kind: MaintenanceJob['kind'] = 'legacy', origin: MaintenanceJob['origin'] = 'manual'): number {
const names = new Set(listServers().map(s => s.name));
if (!servers.length || servers.some(s => !names.has(s))) throw new Error('Choose existing fleet servers.');
return db().transaction(() => {
const target = kind === 'schedule' ? null : freshRelease();
if (kind === 'update' && target === null) throw new Error('Check the published revision before requesting an update.');
let added = 0;
for (const server of new Set(servers)) {
if (getServer(server)?.desired_state === 'stopped') continue;
added += db().prepare(`INSERT OR IGNORE INTO maintenance_jobs (server, target_revision, requested_at, kind, origin)
VALUES (?, ?, ?, ?, ?)`).run(server, target, nowIso(), kind, origin).changes;
const job = db().prepare("SELECT id FROM maintenance_jobs WHERE server = ? AND state IN ('queued','restarting')").get(server) as { id: number };
requestRestart(server, { kind: 'maintenance', id: String(job.id) });
}
return added;
})();
}
/** One automatic attempt per server/revision. Explicit retries preserve the old outcome. */
export function queueUpdates(servers?: string[], automatic = false): number {
if (!maintenanceAllowed()) throw new Error('Maintenance is paused. Resume it before requesting updates.');
if (automatic && !maintenanceAllowed('updates')) return 0;
if (freshRelease() === null) throw new Error('Check the published revision before requesting an update.');
const versions = runningVersions();
if (servers && (!servers.length || servers.some(server => !versions.some(v => v.server === server)))) throw new Error('Choose existing fleet servers.');
const candidates = versions.filter(v => v.update_status === 'outdated' && (!servers || servers.includes(v.server)));
const names = candidates.filter(v => !automatic || !db().prepare(`SELECT 1 FROM maintenance_jobs
WHERE server = ? AND target_revision = ? AND kind = 'update'`).get(v.server, freshRelease())).map(v => v.server);
return names.length ? queueMaintenance(names, 'update', automatic ? 'automatic' : 'manual') : 0;
}
export function retryMaintenance(id: number): number {
const job = db().prepare("SELECT * FROM maintenance_jobs WHERE id = ? AND state = 'failed'").get(id) as MaintenanceJob | undefined;
if (!job) throw new Error('Only failed maintenance can be retried.');
if (!maintenanceAllowed()) throw new Error('Maintenance is paused.');
if (getServer(job.server)?.desired_state === 'stopped') throw new Error('This server is intentionally stopped.');
// Ambiguous lifecycle failures retain their slot and require diagnostics, not another request.
if (job.provisioning_id) {
const operation = db().prepare('SELECT state FROM server_deployments WHERE id = ?').get(job.provisioning_id) as { state: string } | undefined;
if (operation?.state !== 'done') throw new Error('Inspect and recover the existing Provisioning operation before retrying.');
}
return queueMaintenance([job.server], job.kind, 'manual');
}
export function cancelMaintenance(id: number): boolean {
return db().transaction(() => {
const job = db().prepare("SELECT provisioning_id FROM maintenance_jobs WHERE id = ? AND state = 'queued'").get(id) as { provisioning_id: string | null } | undefined;
if (!job) return false;
if (job.provisioning_id && !cancelProvisioning(job.provisioning_id)) return false;
return db().prepare(`UPDATE maintenance_jobs SET state = 'cancelled', finished_at = ?, detail = 'Cancelled by operator'
WHERE id = ? AND state = 'queued'`).run(nowIso(), id).changes > 0;
})();
}
function finish(job: MaintenanceJob, state: 'done' | 'failed' | 'cancelled', detail: string) {
db().prepare('UPDATE maintenance_jobs SET state = ?, detail = ?, finished_at = ? WHERE id = ?')
.run(state, detail, nowIso(), job.id);
}
function hold(job: MaintenanceJob, detail: string) {
db().prepare('UPDATE maintenance_jobs SET detail = ? WHERE id = ?').run(detail, job.id);
}
let ticking: Promise<void> | null = null;
export function maintenanceTick(): Promise<void> {
if (ticking) return ticking;
ticking = runTick().finally(() => { ticking = null; });
return ticking;
}
async function runTick(): Promise<void> {
const latest = release();
if (!latest.checkedAt || Date.now() - Date.parse(latest.checkedAt) >= HOUR) await checkPlutonium();
if (maintenanceAllowed('updates') && freshRelease() !== null) queueUpdates(undefined, true);
db().transaction(() => {
const plan = schedule();
if (maintenanceAllowed() && plan.enabled && plan.nextAt && plan.nextAt <= nowIso()) {
const names = listServers().filter(s => s.desired_state !== 'stopped').map(s => s.name);
if (names.length) queueMaintenance(names, 'schedule', 'schedule');
// Coalesce missed days after downtime into one run; no restart backlog.
writeSetting('plutonium_schedule', { ...plan, nextAt: nextMaintenanceAt(plan.time, new Date(), plan.timeZone) });
}
})();
const jobs = db().prepare(`SELECT * FROM maintenance_jobs WHERE state IN ('queued','restarting') ORDER BY id`).all() as MaintenanceJob[];
for (const job of jobs) {
try { await processJob(job); }
catch { hold(job, 'Maintenance check failed; waiting for the next check.'); }
}
// Update attempts are also durable suppression evidence; never age them into an automatic retry.
db().prepare(`DELETE FROM maintenance_jobs WHERE kind != 'update' AND finished_at < ? AND state IN ('done','failed','cancelled')`)
.run(new Date(Date.now() - 30 * 24 * HOUR).toISOString());
}
async function processJob(job: MaintenanceJob): Promise<void> {
const server = getServer(job.server);
if (!server || server.destroyed_at) return finish(job, 'cancelled', 'Server removed.');
const operation = requestRestart(job.server, { kind: 'maintenance', id: String(job.id) });
if (['failed','interrupted','cancelled'].includes(operation.state)) {
return finish(job, operation.state === 'cancelled' ? 'cancelled' : 'failed', operation.error ?? operation.stage);
}
if (operation.state !== 'done') {
if (operation.submitted_at) {
db().prepare("UPDATE maintenance_jobs SET state = 'restarting', restarted_at = ? WHERE id = ?").run(operation.submitted_at, job.id);
db().prepare('UPDATE plutonium_versions SET revision = NULL, error = ? WHERE server = ?').run('Waiting for restart recovery.', job.server);
}
return hold(job, operation.stage);
}
const revision = await probeVersion(job.server);
if (job.target_revision && (revision === null || revision < job.target_revision)) {
return finish(job, 'failed', `Server recovered, but r${job.target_revision} was not verified (${revision === null ? 'version unknown' : `running r${revision}`}). Check Plutainer auto-update settings.`);
}
return finish(job, 'done', `Server recovered; ${revision === null ? 'running revision unknown' : `running r${revision}`}.`);
}
src/lib/server/maintenance/settings.ts
+9
View File
@@ -0,0 +1,9 @@
import { db } from '../db/index.js';
export function readSetting<T>(key: string, fallback: T): T {
const row = db().prepare('SELECT value FROM settings WHERE key = ?').get(key) as { value: string } | undefined;
return row ? JSON.parse(row.value) : fallback;
}
export function writeSetting(key: string, value: unknown): void {
db().prepare('INSERT INTO settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value')
.run(key, JSON.stringify(value));
}
src/lib/server/pipeline/worker.ts
+2 -4
View File
@@ -15,8 +15,7 @@ import { isProvisioning } from '../fleet/provisioning.js';
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';
import { runMaintenance } from '../maintenance/index.js';
const TICK_MS = 15_000;
// Cap concurrent UDP RCON calls so a 15-server fleet doesn't blast the
@@ -38,8 +37,7 @@ function tick(): Promise<void> {
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); }
try { await runMaintenance(); } catch (err) { console.error('[worker] maintenance error:', err); }
}
async function runTick(): Promise<void> {
src/lib/server/provisioning/requests.ts
+4 -3
View File
@@ -1,3 +1,4 @@
import { maintenanceAllowed } from '../maintenance/policies.js';
import { db } from '../db/index.js';
import { komodoConfigured, rconConfigured } from '../config.js';
import { getServer } from '../fleet/servers.js';
@@ -141,10 +142,10 @@ export function requestLifecycle(server: string, action: RuntimeAction) {
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);
const incident = db().prepare("SELECT kind FROM watchdog_incidents WHERE id = ? AND server = ? AND state IN ('open','recovering')").get(incidentId, server) as { kind: 'health' | 'drift' } | undefined;
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.' };
if (!maintenanceAllowed(incident.kind)) {
return { ok: false, status: 409, error: 'This maintenance policy is paused.' };
}
const row = getServer(server);
if (!row || row.destroyed_at) return { ok: false, status: 404, error: 'Server not found.' };
src/lib/server/provisioning/worker.ts
+9 -3
View File
@@ -1,3 +1,4 @@
import { maintenanceAllowed } from '../maintenance/policies.js';
import { db } from '../db/index.js';
import { komodoConfigured, rconConfigured } from '../config.js';
import { getServer } from '../fleet/servers.js';
@@ -24,7 +25,8 @@ function sourceEligibility(operation: Operation): 'ready' | 'held' | 'cancelled'
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';
const incident = db().prepare('SELECT kind FROM watchdog_incidents WHERE id = ?').get(source.source_id) as { kind: 'health' | 'drift' };
if (!maintenanceAllowed(incident.kind)) return 'held';
}
let held = false;
for (const source of sources) {
@@ -43,8 +45,12 @@ function sourceEligibility(operation: Operation): 'ready' | 'held' | 'cancelled'
if (!row.outcome || ['active','passed'].includes(row.outcome)) return 'ready';
if (row.outcome === 'paused') held = true;
} 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';
const row = db().prepare('SELECT state, provisioning_id, kind, origin FROM maintenance_jobs WHERE id = ?').get(source.source_id) as { state: string; provisioning_id: string; kind: string; origin: string } | undefined;
if (row?.provisioning_id === operation.id && ['queued','restarting'].includes(row.state)) {
if (getServer(operation.server)?.desired_state === 'stopped') return 'cancelled';
if (!maintenanceAllowed() || (row.kind === 'update' && row.origin === 'automatic' && !maintenanceAllowed('updates'))) return 'held';
return 'ready';
}
} else return 'ready';
}
return held ? 'held' : 'cancelled';
src/lib/server/watchdog/index.ts
+2 -233
View File
@@ -1,233 +1,2 @@
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'));
}
// Compatibility for existing callers; Maintenance owns recovery.
export * from "../maintenance/recovery.js";