import { randomUUID } from 'node:crypto'; import { db } from '../db/index.js'; import { nativeInstance } from '../backends/native.js'; import { getServer, listServers } from '../fleet/servers.js'; import { hasActiveProvisioning } from '../fleet/provisioning.js'; import { inspectForMonitoring, rconForMonitoring, monitoringStatus, monitoringIdentity, MonitoringChangedError } from '../fleet/health-observations.js'; import { ONLINE_MAX_MS } from '../../server-freshness.js'; import { latestGameSubmission, telemetryEvidence, telemetryReport } from '../fleet/status-evidence.js'; import { prepareDriftCampaign, createDriftRebuildCampaign, previewDrift } from '../fleet/drift.js'; import { cancelProvisioning, requestWatchdogLifecycle, escalateWatchdogRestart } from '../provisioning/requests.js'; import { deploymentStage, finishDeployment } from '../provisioning/history.js'; import { recordProgress } from '../provisioning/progress.js'; import { withOperationSource } from '../hosts/operation-context.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_DRIFT_SCAN_MS = 5 * 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 server_deleted_at IS NULL 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 sampledAt = kind === 'health' ? details.observedAt ?? iso() : iso(); const sampledTime = Date.parse(sampledAt); const previous = active(server, kind); if (previous) { // Repeated UI scans must not manufacture a sustained failure. const sampleGap = sampledTime - Date.parse(previous.last_seen_at); const actionable = (details as WatchdogEvidence & { actionable?: boolean }).actionable; const prior = JSON.parse(previous.details) as WatchdogEvidence & { actionable?: boolean }; const newSample = sampleGap > 0 && Date.now() - Date.parse(prior.countedAt ?? previous.last_seen_at) >= WATCHDOG_SCAN_MS; const priorActionable = prior.actionable; // A successful routine/manual probe between watchdog samples breaks // the failure streak, even if our next probe happens to fail again. const after = latestGameSubmission(server); const successful = [getServer(server)?.last_seen_at, telemetryReport(server)?.receivedAt] .some(at => at && Date.parse(at) > Date.parse(previous.last_seen_at) && Date.parse(at) <= sampledTime && (!after || Date.parse(at) > Date.parse(after))); const reset = kind === 'health' && (successful || (actionable === true && priorActionable !== true) || prior.identity !== details.identity || sampleGap > WATCHDOG_SCAN_MS + ONLINE_MAX_MS); db().prepare('UPDATE watchdog_incidents SET details=?, last_seen_at=?, failures=?, first_seen_at=? WHERE id=?') .run(JSON.stringify({ ...details, countedAt: newSample || reset ? iso() : prior.countedAt ?? previous.last_seen_at }), newSample || reset ? sampledAt : previous.last_seen_at, reset ? 1 : previous.failures + (newSample ? 1 : 0), reset ? sampledAt : 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, countedAt: iso() }), sampledAt, sampledAt); event(id, summary); return incident(id)!; } /** Call only after verifying container readiness and current game evidence. */ export function resolveRecoveredHealthOperation(operationId: string, players: number, source: 'rcon' | 'telemetry' | null): boolean { return db().transaction(() => { const operation = db().prepare(`SELECT d.server, d.action, d.state, d.submitted_at, d.triggered_by, i.id AS incident_id FROM server_deployments d JOIN provisioning_sources s ON s.operation_id = d.id AND s.kind = 'watchdog' JOIN watchdog_incidents i ON i.id = s.source_id AND i.server = d.server WHERE d.id = ? AND d.server_deleted_at IS NULL AND i.server_deleted_at IS NULL AND i.kind = 'health' AND i.state IN ('open','recovering') AND NOT EXISTS (SELECT 1 FROM provisioning_sources other WHERE other.operation_id = d.id AND other.kind != 'watchdog')`) .get(operationId) as { server: string; action: string; state: string; submitted_at: string | null; triggered_by: string; incident_id: string } | undefined; if (!operation || getServer(operation.server)?.desired_state !== 'running') return false; const waiting = operation.state === 'waiting' && operation.submitted_at === null; if (waiting && !operation.triggered_by.startsWith('watchdog:')) return false; // Cancelling our unsubmitted request is safe when later queued work // exists, and lets that work proceed. Submitted recovery must retain // exclusive ownership before accepting evidence or releasing its slot. if (!waiting && hasActiveProvisioning(operation.server, operationId)) return false; if (!waiting && (operation.action !== 'restart' || !['awaiting-health','failed'].includes(operation.state))) return false; const message = waiting ? 'Server recovered before automatic recovery was submitted' : 'Fresh container and game health verified after restart'; if (waiting && !cancelProvisioning(operationId, message)) return false; const now = iso(); db().prepare(`UPDATE restart_rollout SET checked_at = ?, healthy_at = ?, players = ?, error = NULL WHERE server = ? AND operation_id = ? AND healthy_at IS NULL`).run(now, now, players, operation.server, operationId); db().prepare('UPDATE server_deployments SET checked_at = ?, players = ? WHERE id = ?').run(now, players, operationId); recordProgress(operationId, { blocker: null, observedAt: now, recovery: { state: 'released', at: now, operationId, message }, checks: [{ key: 'container', label: 'Container readiness', source: 'agent', result: 'pass', checkedAt: now }, { key: 'game', label: 'Game status', source, result: 'pass', actual: players, checkedAt: now }] }); if (operation.state === 'awaiting-health') { deploymentStage(operationId, message); finishDeployment(operationId, { ok: true }); } transition(operation.incident_id, 'resolved', message); return true; })(); } 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 scanSuppressed(server: string, kind?: 'health' | 'drift'): boolean { // An explicit retry can reclaim its own earlier attempt's recovery slot. // Keep other findings and every active lifecycle operation suppressed. const retained = db().prepare(`SELECT r.operation_id FROM restart_rollout r JOIN server_deployments d ON d.id = r.operation_id AND d.server_deleted_at IS NULL JOIN watchdog_actions a ON a.operation_id = d.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 i.server = r.server AND i.server_deleted_at IS NULL AND i.state = 'open' AND a.cycle < i.cycle AND (? IS NULL OR i.kind = ?) LIMIT 1`).get(server, kind ?? null, kind ?? null) as { operation_id: string } | undefined; return hasActiveProvisioning(server, retained?.operation_id); } 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 bs.server_deleted_at IS NULL AND b.state IN ('running','ready-promote') LIMIT 1`).get(server) || !!db().prepare(`SELECT 1 FROM deployment_servers ds WHERE ds.server=? AND ds.server_deleted_at IS NULL 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 (row.server_deleted_at) return { ok: false as const, status: 409, error: 'This server was deleted; its recovery history cannot be changed.' }; 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 { const rows = db().prepare("SELECT * FROM watchdog_incidents WHERE server_deleted_at IS NULL AND 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; // Continue observing our own recovery, but do not verify or escalate // against files and health owned by another lifecycle operation. if (hasActiveProvisioning(row.server, action.operation_id)) 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 || hasActiveProvisioning(row.server, action.operation_id)) continue; if (plan.files.length) { transition(row.id, 'attention', 'Rebuild completed but drift remains; inspect the campaign and server files'); continue; } } catch (err) { if (!hasActiveProvisioning(row.server, action.operation_id)) 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; // A recovery deadline is a reason to check again, not evidence that // rebuilding is necessary. The worker may not have polled this tick yet. const observation = await observe(row.server, action.operation_id); if (!observation || incident(row.id)?.cycle !== row.cycle || hasActiveProvisioning(row.server, action.operation_id)) continue; if (observation.healthy) { resolveRecoveredHealthOperation(action.operation_id, observation.players!, observation.source ?? null); continue; } if (!observation.actionable) 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; observedAt: string; identity: string; players?: number; source?: 'rcon' | 'telemetry' } async function observe(server: string, operationId?: string, fresh = false): Promise { const epoch = latestGameSubmission(server); const suppressed = () => (operationId ? hasActiveProvisioning(server, operationId) : scanSuppressed(server)) || protectedWork(server) || getServer(server)?.desired_state !== 'running' || epoch !== latestGameSubmission(server); // Routine watchdog passes only consume the status monitor. Manual checks and // recovery verification still obtain fresh, independently verified evidence. const snapshot = !fresh && !operationId ? monitoringStatus(server) : null; if (!fresh && !operationId && !snapshot) return null; const inspected = snapshot?.container ?? await inspectForMonitoring(server, true).catch(error => { if (error instanceof MonitoringChangedError) return null; throw error; }); if (!inspected) return null; const health = inspected.result; if (suppressed()) return null; const identity = snapshot?.identity ?? monitoringIdentity(server); const containerEvidence = { identity, observedAt: inspected.observedAt }; const state = health.state?.toLowerCase() ?? null; if (!health.ok) return { ...containerEvidence, healthy: false, actionable: !!state && ['stopped','exited','created','dead','unhealthy','restarting','starting'].includes(state), reason: health.reason, containerState: state }; const telemetry = telemetryEvidence(server); if (telemetry) return { identity, observedAt: telemetry.receivedAt, healthy: true, actionable: false, reason: 'Container healthy with fresh telemetry', containerState: state, players: telemetry.players, source: 'telemetry' }; const probed = snapshot ? snapshot.game : await rconForMonitoring(server, true).catch(error => { if (error instanceof MonitoringChangedError) return null; throw error; }); if (!probed) return null; const rcon = probed.result; if (suppressed()) return null; const latest = telemetryEvidence(server); const gameEvidence = { identity: snapshot?.identity ?? monitoringIdentity(server), observedAt: latest?.receivedAt ?? probed.observedAt }; if (rcon.ok || latest) return { ...gameEvidence, healthy: true, actionable: false, reason: 'Container healthy with fresh game status', containerState: state, players: latest?.players ?? (rcon.ok ? rcon.status.players : 0), source: latest ? 'telemetry' : 'rcon' }; return { ...gameEvidence, healthy: false, actionable: rcon.reason === 'timeout' || rcon.reason === 'send-error', reason: `Game status unavailable: ${rcon.reason}; waiting for player-count verification before recovery`, containerState: state }; } async function scanServer(server: string, forceScan: boolean): Promise { const row = getServer(server); if (!row || row.desired_state === 'stopped') return; if (scanSuppressed(server) || protectedWork(server)) return; const observation = await observe(server, undefined, forceScan); // Probes can overlap an API request; recheck ownership before queuing anything. if (!observation || scanSuppressed(server) || protectedWork(server) || getServer(server)?.desired_state === 'stopped') return; if (observation.identity !== monitoringIdentity(server)) return; if (!scanSuppressed(server, 'health') && observation.healthy) { const current = active(server, 'health'); if (current) transition(current.id, 'resolved', 'Fresh game status and host agent health verified'); } else if (!scanSuppressed(server, 'health')) { const problem = detect(server, 'health', observation.reason, observation); const startup = ['starting','restarting'].includes(observation.containerState ?? ''); const oldEnough = Date.parse(observation.observedAt) - 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.healthy && !observation.actionable) || scanSuppressed(server, 'drift')) return; // File trees change far less often than liveness. Persist the last successful // scan per registration/process so health checks do not rescan every minute. // Manual checks and new lifecycle submissions always get fresh file evidence. const driftKey = `watchdog_drift_scan:${server}`; const instance = nativeInstance(server); const identity = JSON.stringify([row.created_at, instance?.id, instance?.generation, latestGameSubmission(server)]); let previous: { identity?: string; at?: string } = {}; try { previous = JSON.parse(setting(driftKey) ?? '{}'); } catch { /* rescan legacy/invalid state */ } const age = Date.now() - Date.parse(previous?.at ?? ''); if (!forceScan && previous?.identity === identity && age >= 0 && age < WATCHDOG_DRIFT_SCAN_MS) return; const prepared = await prepareDriftCampaign(server); if (scanSuppressed(server, 'drift') || protectedWork(server) || getServer(server)?.desired_state === 'stopped') return; setSetting(driftKey, JSON.stringify({ identity, at: iso() })); 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 | null = null; export function watchdogTick(forceScan = false): Promise { if (inFlight) return inFlight; inFlight = withOperationSource(forceScan ? 'Manual maintenance check' : 'Maintenance watchdog', () => run(forceScan)) .finally(() => { inFlight = null; }); return inFlight; } async function run(forceScan: boolean): Promise { 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 (!listServers().some(server => nativeInstance(server.name))) { setSetting('watchdog_scan_error', 'Register a native host instance 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, forceScan); } catch (err) { if (!hasActiveProvisioning(server.name)) 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 server_deleted_at IS NULL AND state != 'resolved'").all() as WatchdogIncident[]; for (const row of abandoned) { if (getServer(row.server) || hasActiveProvisioning(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')); }