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

Use status monitor evidence for routine watchdog health checks

Browse Source
This commit is contained in:
josh
2026-10-05 16:13:47 -04:00
parent 91f4490d34
commit 87c5cf4b4b
8 changed files with 321 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/maintenance.md
+12 -3
View File
@@ -102,9 +102,18 @@ reachable. Finished scheduled and legacy maintenance activity is retained for
## Health recovery and configuration drift
The health observer (the watchdog) scans once per minute. Three observations over
at least two minutes are required before an automatic health restart; starting or
restarting containers get five minutes. Unknown Komodo responses are observations,
The status monitor collects container and game health every 30 seconds. Once per
minute, the watchdog evaluates that completed evidence without issuing its own
routine health probes. Samples expire after 45 seconds; missing or stale evidence
leaves health unknown and cannot justify recovery. Samples retain their original
timestamps and instance identity, and each failure sample counts at most once.
A monitoring gap or changed instance resets the sustained failure window. After
an application restart, routine watchdog checks wait for a new status-monitor pass.
**Check now** and recovery verification still request fresh evidence, and
Provisioning retains its checks before executing recovery actions.
Three observations over at least two minutes are required before an automatic
health restart; starting or restarting containers get five minutes. Unknown Komodo responses are observations,
not proof of a stopped container. Repeated manual checks do not accelerate this
window. A successful routine or manual game-status probe between scans resets
the failure streak. Missing RCON configuration, unresolved ports, and invalid
src/lib/server/fleet/health-observations.ts
+60
View File
@@ -1,3 +1,4 @@
import { createHash } from 'node:crypto';
import { config } from '../config.js';
import { db } from '../db/index.js';
import { getHost, queueHostOperation, waitHostOperationStatus } from '../hosts/index.js';
@@ -7,6 +8,7 @@ import { rconStatusResult, agentRconStatusResult, type AgentRconReply } from '..
import { withOperationSource } from '../hosts/operation-context.js';
import { getServer } from './servers.js';
import { latestGameSubmission } from './lifecycle-evidence.js';
import { ONLINE_MAX_MS } from '../../server-freshness.js';
const MAX_AGE_MS = 30_000;
const MAX_SERVERS = 1024;
@@ -100,6 +102,62 @@ async function observe<T extends { ok: boolean }>(server: string, fresh: boolean
const inspections = new Map<string, Entry<Awaited<ReturnType<typeof komodoInspectHealthy>>>>();
const rcon = new Map<string, Entry<Awaited<ReturnType<typeof rconStatusResult>>>>();
interface MonitoringStatus {
identity: string;
container: Observation<Awaited<ReturnType<typeof komodoInspectHealthy>>>;
game?: Observation<Awaited<ReturnType<typeof rconStatusResult>>>;
}
// Keep failed samples too, independently of the successful-probe cache. Reading
// this snapshot must never dispatch a request or renew an observation's age.
const statuses = new Map<string, { snapshot?: MonitoringStatus }>();
export function monitoringIdentity(server: string): string {
const container = identity(server, 'inspect');
const game = identity(server, 'rcon');
// Identity includes connection configuration; persist only its fingerprint.
return createHash('sha256').update(JSON.stringify([container.key, game.key, game.port])).digest('hex');
}
/** Capture ownership before a status pass; publish only evidence for that instance. */
export function monitoringStatusWriter(server: string) {
let before: string;
try { before = identity(server, 'inspect').key; }
catch { return () => {}; }
const slot = { snapshot: statuses.get(server)?.snapshot };
statuses.delete(server);
statuses.set(server, slot);
while (statuses.size > MAX_SERVERS) statuses.delete(statuses.keys().next().value!);
return (container: MonitoringStatus['container'], game?: MonitoringStatus['game']) => {
// A forced check or newer pass supersedes this writer, even while its
// network requests are still in flight.
if (statuses.get(server) !== slot) return;
try {
if (identity(server, 'inspect').key !== before) return;
const current = slot.snapshot;
if (current && current.container.observedAt > container.observedAt) return;
slot.snapshot = { identity: monitoringIdentity(server), container, game };
} catch { statuses.delete(server); }
};
}
/** Latest completed status-monitor pass, with stale or superseded evidence excluded. */
export function monitoringStatus(server: string): MonitoringStatus | null {
const saved = statuses.get(server)?.snapshot;
if (!saved) return null;
const fresh = (at: string) => {
const age = Date.now() - Date.parse(at);
return age >= 0 && age <= ONLINE_MAX_MS;
};
try {
if (saved.identity !== monitoringIdentity(server)
|| !fresh(saved.container.observedAt)
|| newerEvidence(identity(server, 'inspect'), Date.parse(saved.container.observedAt))) return null;
const game = saved.game && fresh(saved.game.observedAt)
&& !newerEvidence(identity(server, 'rcon'), Date.parse(saved.game.observedAt)) ? saved.game : undefined;
return { ...saved, game };
} catch { return null; }
}
/** Keep the actual sample time: a later heartbeat must not renew old evidence. */
export function heartbeatInspection(server: string): Observation<Awaited<ReturnType<typeof komodoInspectHealthy>>> | null {
const row = getServer(server);
@@ -175,10 +233,12 @@ export function prepareMonitoringRcon(servers: string[]): Map<string, () => Prom
/** Routine monitoring only. Manual diagnostics and disruption gates use fresh probes directly. */
export function inspectForMonitoring(server: string, fresh = false) {
if (fresh) statuses.delete(server);
const heartbeat = !fresh && heartbeatInspection(server);
if (heartbeat) return Promise.resolve(heartbeat);
return observe(server, fresh, 'inspect', inspections, komodoInspectHealthy);
}
export function rconForMonitoring(server: string, fresh = false) {
if (fresh) statuses.delete(server);
return observe(server, fresh, 'rcon', rcon, rconStatusResult);
}
src/lib/server/maintenance/recovery.ts
+31 -16
View File
@@ -4,7 +4,8 @@ import { komodoConfigured } from '../config.js';
import { nativeInstance } from '../backends/native.js';
import { getServer, listServers } from '../fleet/servers.js';
import { hasActiveProvisioning } from '../fleet/provisioning.js';
import { inspectForMonitoring, rconForMonitoring, MonitoringChangedError } from '../fleet/health-observations.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';
@@ -41,28 +42,34 @@ function transition(id: string, state: WatchdogState, summary: string) {
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 newSample = Date.now() - Date.parse(previous.last_seen_at) >= WATCHDOG_SCAN_MS;
const sampleGap = sampledTime - Date.parse(previous.last_seen_at);
const actionable = (details as WatchdogEvidence & { actionable?: boolean }).actionable;
const priorActionable = (JSON.parse(previous.details) as { 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) <= Date.now() && (!after || Date.parse(at) > Date.parse(after)));
const reset = kind === 'health' && (successful || (actionable === true && priorActionable !== true));
&& 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), newSample || reset ? iso() : previous.last_seen_at,
reset ? 1 : previous.failures + (newSample ? 1 : 0), reset ? iso() : previous.first_seen_at, previous.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), iso(), iso());
VALUES (?,?,?,'open',?,?,?,?)`).run(id, server, kind, summary, JSON.stringify({ ...details, countedAt: iso() }), sampledAt, sampledAt);
event(id, summary);
return incident(id)!;
}
@@ -222,23 +229,29 @@ async function reconcileActions(): Promise<void> {
}
}
interface Observation { healthy: boolean; actionable: boolean; reason: string; containerState: string | null; players?: number; source?: 'rcon' | 'telemetry' }
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<Observation | null> {
const epoch = latestGameSubmission(server);
const suppressed = () => (operationId ? hasActiveProvisioning(server, operationId) : scanSuppressed(server))
|| protectedWork(server) || getServer(server)?.desired_state !== 'running' || epoch !== latestGameSubmission(server);
const inspected = await inspectForMonitoring(server, fresh || !!operationId).catch(error => {
// 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 { healthy: false, actionable: !!state && ['stopped','exited','created','dead','unhealthy','restarting','starting'].includes(state), reason: health.reason, containerState: state };
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 { healthy: true, actionable: false, reason: 'Komodo healthy with fresh telemetry', containerState: state, players: telemetry.players, source: 'telemetry' };
const probed = await rconForMonitoring(server, fresh || !!operationId).catch(error => {
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;
});
@@ -246,9 +259,10 @@ async function observe(server: string, operationId?: string, fresh = false): Pro
const rcon = probed.result;
if (suppressed()) return null;
const latest = telemetryEvidence(server);
if (rcon.ok || latest) return { healthy: true, actionable: false, reason: 'Komodo healthy with fresh game status', containerState: state,
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 { healthy: false, actionable: rcon.reason === 'timeout' || rcon.reason === 'send-error',
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<void> {
@@ -258,13 +272,14 @@ async function scanServer(server: string, forceScan: boolean): Promise<void> {
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 Komodo 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.now() - Date.parse(problem.first_seen_at) >= (startup ? WATCHDOG_RECOVERY_MS : WATCHDOG_FAILURE_MS);
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);
src/lib/server/pipeline/worker.ts
+8 -3
View File
@@ -6,7 +6,7 @@ import { emit } from './bus.js';
import type { ServerStatusEntry } from './bus.js';
import type { RconFailure } from '../integrations/rcon.js';
import { getServer, listServers, setServerFields, type ServerRow } from '../fleet/servers.js';
import { inspectForMonitoring, rconForMonitoring, prepareMonitoringRcon, MonitoringChangedError } from '../fleet/health-observations.js';
import { inspectForMonitoring, rconForMonitoring, prepareMonitoringRcon, monitoringStatusWriter, MonitoringChangedError } from '../fleet/health-observations.js';
import { restartSettling } from '../provisioning/index.js';
import { serverObservation, observationReader } from '../fleet/status.js';
import { telemetryEvidence, latestGameSubmission } from '../fleet/status-evidence.js';
@@ -149,6 +149,9 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp
let updates: Partial<Record<keyof ServerRow, string | number | null>> = {};
const flush = () => { setServerFields(srv.name, updates); updates = {}; };
if (!isProvisioning(srv.name)) {
const publishObservation = monitoringStatusWriter(srv.name);
let gameObservation: Awaited<ReturnType<typeof rconForMonitoring>> | undefined;
const settling = restartSettling(srv.name);
const epoch = latestGameSubmission(srv.name);
const unchanged = () => {
const current = getServer(srv.name);
@@ -157,7 +160,7 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp
&& current.native_instance_id === srv.native_instance_id && current.instance_generation === srv.instance_generation;
};
const inspectionStartedAt = new Date().toISOString();
const inspected = restartSettling(srv.name)
const inspected = settling
? { result: { ok: false as const, state: 'restarting', reason: 'Restart requested' }, observedAt: inspectionStartedAt }
: await inspectForMonitoring(srv.name).catch(err => err instanceof MonitoringChangedError ? null
: { result: { ok: false as const, state: 'unknown', reason: (err as Error).message }, observedAt: inspectionStartedAt });
@@ -178,10 +181,11 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp
flush();
const probeStartedAt = new Date().toISOString();
const probe = await (probeStatus ? probeStatus() : rconForMonitoring(srv.name)).catch(err => err instanceof MonitoringChangedError ? null
: { result: { ok: false as const, reason: 'send-error' as RconFailure, detail: (err as Error).message, rawReply: undefined }, observedAt: probeStartedAt });
: { result: { ok: false as const, endpoint: null, reason: 'send-error' as RconFailure, detail: (err as Error).message, rawReply: undefined }, observedAt: probeStartedAt });
if (!probe) return null;
const result = probe.result;
if (unchanged() && (getServer(srv.name)?.last_checked_at ?? '') <= probe.observedAt) {
gameObservation = probe;
const now = probe.observedAt;
updates.last_checked_at = now;
updates.last_reply_snippet = result.rawReply ?? null;
@@ -196,6 +200,7 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp
} else updates.last_error = encodeError(result.reason, result.detail);
}
}
if (!settling && unchanged()) publishObservation(inspected, gameObservation);
}
} else {
updates.last_error = null;
src/lib/watchdog.ts
+3
View File
@@ -17,6 +17,9 @@ export interface WatchdogIncident {
operation_stage?: string | null;
}
export interface WatchdogEvidence {
observedAt?: string;
countedAt?: string;
identity?: string;
reason?: string;
containerState?: string | null;
files?: DriftPlanFile[];
tests/unit/health-observations.test.ts
+68
View File
@@ -48,6 +48,74 @@ function deferred<T>() {
}
describe('shared monitoring observations', () => {
it('retains failures for watchdog consumption without caching them as reusable probes', async () => {
const api = await setup();
api.rcon.mockResolvedValue({ ok: false, endpoint: null, reason: 'timeout', detail: 'No reply' });
const publish = api.monitoringStatusWriter(server);
publish(await api.inspectForMonitoring(server), await api.rconForMonitoring(server));
expect(api.monitoringStatus(server)).toMatchObject({
container: { observedAt: new Date(now).toISOString(), result: healthy },
game: { observedAt: new Date(now).toISOString(), result: { ok: false, reason: 'timeout' } }
});
vi.setSystemTime(now + 40_000);
expect(api.monitoringStatus(server)?.game?.observedAt).toBe(new Date(now).toISOString());
expect(api.inspect).toHaveBeenCalledTimes(1);
expect(api.rcon).toHaveBeenCalledTimes(1);
vi.setSystemTime(now + 45_001);
expect(api.monitoringStatus(server)).toBeNull();
vi.setSystemTime(now - 1);
expect(api.monitoringStatus(server)).toBeNull();
});
it.each(['registration', 'generation', 'lifecycle', 'endpoint', 'manual', 'offline'] as const)('invalidates a completed status snapshot after %s changes', async change => {
const api = await setup();
const host = await native(api);
const publish = api.monitoringStatusWriter(server);
publish(await api.inspectForMonitoring(server), await api.rconForMonitoring(server));
expect(api.monitoringStatus(server)).not.toBeNull();
if (change === 'registration') api.database.prepare('UPDATE servers SET created_at=? WHERE name=?').run(new Date(now + 1).toISOString(), server);
if (change === 'generation') api.database.prepare('UPDATE native_instances SET generation=2').run();
if (change === 'lifecycle') api.database.prepare('INSERT INTO restart_rollout(server,requested_at) VALUES(?,?)').run(server, new Date(now + 1).toISOString());
if (change === 'endpoint') api.database.prepare('UPDATE native_instances SET port=28961').run();
if (change === 'manual') api.database.prepare('UPDATE servers SET container_checked_at=? WHERE name=?').run(new Date(now + 1).toISOString(), server);
if (change === 'offline') api.database.prepare('UPDATE ec_hosts SET last_heartbeat=? WHERE id=?').run(new Date(now - 300_000).toISOString(), host.host.id);
expect(api.monitoringStatus(server)).toBeNull();
});
it('does not publish a status pass across an instance replacement', async () => {
const api = await setup();
await native(api);
const publish = api.monitoringStatusWriter(server);
const container = await api.inspectForMonitoring(server);
const game = await api.rconForMonitoring(server);
api.database.prepare('UPDATE native_instances SET generation=2').run();
publish(container, game);
expect(api.monitoringStatus(server)).toBeNull();
});
it.each(['inspect', 'rcon'] as const)('invalidates the routine snapshot when a forced %s check starts', async kind => {
const api = await setup();
const publish = api.monitoringStatusWriter(server);
const container = await api.inspectForMonitoring(server);
const game = await api.rconForMonitoring(server);
publish(container, game);
expect(api.monitoringStatus(server)).not.toBeNull();
await (kind === 'inspect' ? api.inspectForMonitoring : api.rconForMonitoring)(server, true);
publish(container, game);
expect(api.monitoringStatus(server)).toBeNull();
});
it('excludes old game evidence even when the container sample is fresh', async () => {
const api = await setup();
const game = await api.rconForMonitoring(server);
vi.setSystemTime(now + 46_000);
const publish = api.monitoringStatusWriter(server);
publish(await api.inspectForMonitoring(server), game);
expect(api.monitoringStatus(server)?.container.result).toEqual(healthy);
expect(api.monitoringStatus(server)?.game).toBeUndefined();
expect(api.rcon).toHaveBeenCalledTimes(1);
});
it.each(['inspect', 'rcon'] as const)('reuses %s results until 30 seconds after request start without refreshing their timestamp', async kind => {
const api = await setup();
const call = kind === 'inspect' ? api.inspectForMonitoring : api.rconForMonitoring;
tests/unit/host-status-batch.test.ts
+8
View File
@@ -59,6 +59,10 @@ describe('host status observations', () => {
} finally { api.complete(); await Promise.all(passes); }
expect(api.inspect).not.toHaveBeenCalled();
expect(api.getServer('server-2-4')).toMatchObject({ last_players: 0, last_map: 'mp_kino', container_checked_at: new Date(at).toISOString() });
expect(api.monitoringStatus('server-2-4')).toMatchObject({
container: { observedAt: new Date(at).toISOString(), result: { ok: true } },
game: { observedAt: api.getServer('server-2-4')?.last_seen_at, result: { ok: true } }
});
expect(api.db.prepare('SELECT COUNT(*) AS n FROM metric_samples').get()).toEqual({ n: 15 });
await api.fanoutStatus();
expect(api.operations()).toHaveLength(3);
@@ -90,6 +94,9 @@ describe('host status observations', () => {
api.heartbeat(host.id, { ...host.inventory, instanceStates: [{ ...host.inventory.instanceStates![0], healthy: false, state: 'exited' }] });
await api.fanoutStatus();
expect(api.getServer('server-0-0')?.container_state).toBe('exited');
expect(api.monitoringStatus('server-0-0')).toMatchObject({ container: {
observedAt: new Date(at).toISOString(), result: { ok: false, state: 'exited' }
} });
expect(api.operations()).toEqual([]);
});
@@ -120,6 +127,7 @@ describe('host status observations', () => {
} finally { api.complete(); await pass; }
expect(api.getServer('server-0-0')?.last_error).toBe('bad-reply:wrong RCON password');
expect(api.getServer('server-0-4')?.last_error).toContain('timeout:');
expect(api.monitoringStatus('server-0-4')?.game?.result).toMatchObject({ ok: false, reason: 'timeout' });
expect(api.operations()).toHaveLength(1);
});
tests/unit/watchdog.test.ts
+131
View File
@@ -58,6 +58,10 @@ const onlyIncident = () => { expect(incidents()).toHaveLength(1); return inciden
const operations = () => database.db().prepare('SELECT id, action, state FROM server_deployments ORDER BY rowid').all() as { id: string; action: string; state: string }[];
function advance(ms: number) { vi.setSystemTime(Date.now() + ms); }
function down() { vi.mocked(komodo.komodoInspectHealthy).mockResolvedValue({ ok: false, state: 'exited', reason: 'Container exited' }); }
async function pollStatus() {
const worker = await vi.importActual<typeof import('../../src/lib/server/pipeline/worker.js')>('../../src/lib/server/pipeline/worker.js');
await worker.fanoutStatus();
}
async function sustainedFailure() {
down();
await watchdog.watchdogTick(true);
@@ -68,24 +72,148 @@ async function sustainedFailure() {
}
describe('Watchdog detection and ownership', () => {
it('waits for status-monitor evidence without issuing its own routine probes', async () => {
down();
for (let minute = 0; minute < 4; minute++) {
await watchdog.watchdogTick();
advance(watchdog.WATCHDOG_SCAN_MS);
}
expect(komodo.komodoInspectHealthy).not.toHaveBeenCalled();
expect(rcon.rconStatusResult).not.toHaveBeenCalled();
expect(incidents()).toEqual([]);
expect(operations()).toEqual([]);
});
it.each(['container', 'game'])('requests recovery from sustained %s failures collected by status polling', async kind => {
if (kind === 'container') down();
else vi.mocked(rcon.rconStatusResult).mockResolvedValue(missingRcon);
for (let minute = 0; minute < 3; minute++) {
await pollStatus();
await watchdog.watchdogTick();
if (minute < 2) {
expect(onlyIncident().failures).toBe(minute + 1);
expect(operations()).toEqual([]);
advance(watchdog.WATCHDOG_SCAN_MS);
}
}
expect(onlyIncident()).toMatchObject({ failures: 3, state: 'recovering' });
expect(operations()).toEqual([expect.objectContaining({ action: 'restart', state: 'waiting' })]);
expect(komodo.komodoInspectHealthy).toHaveBeenCalledTimes(3);
expect(rcon.rconStatusResult).toHaveBeenCalledTimes(kind === 'game' ? 3 : 0);
});
it('does not count the same sample again or use stale data, and resets after a monitoring gap', async () => {
down();
await pollStatus();
await watchdog.watchdogTick();
const first = onlyIncident();
advance(30_000);
database.db().prepare("DELETE FROM settings WHERE key='watchdog_last_scan'").run();
await watchdog.watchdogTick();
advance(60_000);
await watchdog.watchdogTick();
expect(onlyIncident()).toMatchObject({ failures: 1, last_seen_at: first.last_seen_at });
expect(komodo.komodoInspectHealthy).toHaveBeenCalledTimes(1);
advance(60_000);
await pollStatus();
await watchdog.watchdogTick();
expect(onlyIncident()).toMatchObject({ failures: 1, first_seen_at: new Date().toISOString() });
expect(operations()).toEqual([]);
});
it('uses sample times across polling offsets without counting early or losing sustained failures', async () => {
down();
await pollStatus();
advance(20_000);
await watchdog.watchdogTick();
advance(10_000);
await pollStatus();
advance(50_000);
// The first failure is stale until the monitor supplies another sample.
await pollStatus();
await watchdog.watchdogTick();
expect(onlyIncident().failures).toBe(2);
advance(40_000);
await pollStatus();
advance(20_000);
await watchdog.watchdogTick();
expect(onlyIncident()).toMatchObject({ failures: 3, state: 'recovering' });
expect(operations()).toHaveLength(1);
});
it('breaks a routine failure streak on a successful status poll between watchdog passes', async () => {
vi.mocked(rcon.rconStatusResult).mockResolvedValue(missingRcon);
await pollStatus();
await watchdog.watchdogTick();
advance(30_000);
vi.mocked(rcon.rconStatusResult).mockResolvedValueOnce(healthyRcon);
await pollStatus();
advance(30_000);
await pollStatus();
await watchdog.watchdogTick();
expect(onlyIncident()).toMatchObject({ failures: 1, first_seen_at: new Date().toISOString() });
expect(operations()).toEqual([]);
expect(rcon.rconStatusResult).toHaveBeenCalledTimes(3);
});
it('uses fresh telemetry alongside the polled container and waits for polled RCON when telemetry expires', async () => {
const { saveTelemetry } = await import('../../src/lib/server/fleet/telemetry.js');
saveTelemetry({ ...gameEvent('heartbeat', 17, 4, server), receivedAt: new Date().toISOString() });
await pollStatus();
await watchdog.watchdogTick();
expect(incidents()).toEqual([]);
expect(rcon.rconStatusResult).not.toHaveBeenCalled();
advance(watchdog.WATCHDOG_SCAN_MS);
await watchdog.watchdogTick();
expect(incidents()).toEqual([]);
expect(rcon.rconStatusResult).not.toHaveBeenCalled();
vi.mocked(rcon.rconStatusResult).mockResolvedValue(missingRcon);
await pollStatus();
advance(30_000);
database.db().prepare("DELETE FROM settings WHERE key='watchdog_last_scan'").run();
await watchdog.watchdogTick();
expect(onlyIncident()).toMatchObject({ failures: 1, state: 'open' });
expect(rcon.rconStatusResult).toHaveBeenCalledTimes(1);
});
it('starts a new failure window for a replacement registration', async () => {
down();
await pollStatus();
await watchdog.watchdogTick();
advance(watchdog.WATCHDOG_SCAN_MS);
await pollStatus();
await watchdog.watchdogTick();
expect(onlyIncident().failures).toBe(2);
advance(watchdog.WATCHDOG_SCAN_MS);
database.db().prepare('UPDATE servers SET created_at=? WHERE name=?').run(new Date().toISOString(), server);
await pollStatus();
await watchdog.watchdogTick();
expect(onlyIncident()).toMatchObject({ failures: 1, first_seen_at: new Date().toISOString() });
expect(operations()).toEqual([]);
});
it('scans files every five minutes while retaining minute health checks and immediate manual scans', async () => {
const drift = await import('../../src/lib/server/fleet/drift.js');
const scan = vi.spyOn(drift, 'prepareDriftCampaign');
await pollStatus();
await watchdog.watchdogTick();
expect(scan).toHaveBeenCalledTimes(1);
for (let minute = 1; minute < 5; minute++) {
advance(watchdog.WATCHDOG_SCAN_MS);
await pollStatus();
await watchdog.watchdogTick();
}
expect(scan).toHaveBeenCalledTimes(1);
expect(komodo.komodoInspectHealthy).toHaveBeenCalledTimes(5);
advance(watchdog.WATCHDOG_SCAN_MS);
await pollStatus();
await watchdog.watchdogTick();
expect(scan).toHaveBeenCalledTimes(2);
await watchdog.watchdogTick(true);
expect(scan).toHaveBeenCalledTimes(3);
advance(watchdog.WATCHDOG_SCAN_MS);
database.db().prepare('UPDATE servers SET created_at=? WHERE name=?').run(new Date().toISOString(), server);
await pollStatus();
await watchdog.watchdogTick();
expect(scan).toHaveBeenCalledTimes(4);
scan.mockRestore();
@@ -94,9 +222,11 @@ describe('Watchdog detection and ownership', () => {
it('retries failed file scans on the next health scan instead of retaining successful freshness', async () => {
const drift = await import('../../src/lib/server/fleet/drift.js');
const scan = vi.spyOn(drift, 'prepareDriftCampaign').mockRejectedValueOnce(new Error('Agent unavailable'));
await pollStatus();
await watchdog.watchdogTick();
expect(watchdog.watchdogOverview().scanError).toContain('Agent unavailable');
advance(watchdog.WATCHDOG_SCAN_MS);
await pollStatus();
await watchdog.watchdogTick();
expect(scan).toHaveBeenCalledTimes(2);
expect(watchdog.watchdogOverview().scanError).toBeNull();
@@ -112,6 +242,7 @@ describe('Watchdog detection and ownership', () => {
expect(onlyIncident().failures).toBe(1);
expect(operations()).toEqual([]);
advance(watchdog.WATCHDOG_SCAN_MS);
await pollStatus();
await watchdog.watchdogTick();
expect(onlyIncident().failures).toBe(2);
advance(watchdog.WATCHDOG_SCAN_MS - 1);