Share recent health observations between background monitors

src/lib/server/fleet/health-observations.ts

@@ -0,0 +1,108 @@
+import { config } from '../config.js';
+import { db } from '../db/index.js';
+import { getHost } from '../hosts/index.js';
+import { nativeInstance } from '../backends/native.js';
+import { komodoInspectHealthy } from '../integrations/komodo.js';
+import { rconStatusResult } from '../integrations/rcon.js';
+import { getServer } from './servers.js';
+import { latestGameSubmission } from './lifecycle-evidence.js';
+
+const MAX_AGE_MS = 30_000;
+const MAX_SERVERS = 1024;
+type Probe = 'inspect' | 'rcon';
+interface Observation<T> { result: T; observedAt: string }
+interface Identity { key: string; port: number | null; checkedAt: string | null }
+interface Entry<T> {
+    identity: Identity;
+    startedAt: number;
+    pending?: Promise<Observation<T>>;
+    observation?: Observation<T>;
+}
+
+export class MonitoringChangedError extends Error {}
+
+function identity(server: string, probe: Probe): Identity {
+    const row = getServer(server);
+    if (!row || row.destroyed_at) throw new MonitoringChangedError('Server is no longer registered; monitoring evidence is unavailable.');
+    const native = row.native_instance_id ? nativeInstance(server) : null;
+    if (row.native_instance_id && (!native || getHost(native.host_id)?.connectionState !== 'online')) {
+        throw new Error('Host agent is offline; monitoring evidence is unavailable.');
+    }
+    // Stop requests also invalidate observations, even though they do not
+    // establish a new game-telemetry epoch in latestGameSubmission().
+    const lifecycle = db().prepare(`SELECT id, action, submitted_at FROM server_deployments
+        WHERE server=? AND server_deleted_at IS NULL ORDER BY rowid DESC LIMIT 1`).get(server) ?? null;
+    return {
+        key: JSON.stringify([row.created_at, row.desired_state, latestGameSubmission(server), lifecycle,
+            native ? [native.host_id, native.id, native.generation] : null,
+            native ? null : [row.komodo_type, row.komodo_name, row.komodo_server, config.komodo, config.httpTimeoutMs],
+            probe === 'rcon' ? [row.rcon_host, config.rcon] : null]),
+        port: probe === 'rcon' ? native?.port ?? row.game_port : null,
+        checkedAt: probe === 'rcon' ? row.last_checked_at : row.container_checked_at
+    };
+}
+
+function sameIdentity(left: Identity, right: Identity): boolean {
+    return left.key === right.key && left.port === right.port;
+}
+function newerEvidence(current: Identity, startedAt: number): boolean {
+    return Date.parse(current.checkedAt ?? '') > startedAt;
+}
+function reusable(startedAt: number): boolean {
+    const age = Date.now() - startedAt;
+    return age >= 0 && age < MAX_AGE_MS;
+}
+
+async function observe<T extends { ok: boolean }>(server: string, fresh: boolean, probe: Probe,
+    entries: Map<string, Entry<T>>, request: (server: string) => Promise<T>): Promise<Observation<T>> {
+    let current: Identity;
+    try { current = identity(server, probe); }
+    catch (error) { entries.delete(server); throw error; }
+    const saved = entries.get(server);
+    const savedAt = saved?.observation ? Date.parse(saved.observation.observedAt) : saved?.startedAt ?? 0;
+    if (!fresh && saved && sameIdentity(saved.identity, current) && !newerEvidence(current, savedAt)) {
+        if (saved.pending) return saved.pending;
+        if (saved.observation && reusable(saved.startedAt)) return saved.observation;
+    }
+
+    // Replace the previous slot before dispatch, including for forced probes.
+    // A forced failure must not leave an older successful result reusable.
+    const entry: Entry<T> = { identity: current, startedAt: Date.now() };
+    entries.delete(server);
+    entries.set(server, entry);
+    while (entries.size > MAX_SERVERS) entries.delete(entries.keys().next().value!);
+    const pending = Promise.resolve().then(() => request(server)).then(result => {
+        const latest = identity(server, probe);
+        // Legacy RCON may discover and save its first port during this request.
+        // Accept that enrichment only when all other identity fields agree.
+        const endpoint = 'endpoint' in result ? result.endpoint : null;
+        const learnedPort = probe === 'rcon' && !current.port && result.ok
+            && endpoint !== null && typeof endpoint === 'object' && 'port' in endpoint && endpoint.port === latest.port;
+        if (current.key !== latest.key || (current.port !== latest.port && !learnedPort)
+            || newerEvidence(latest, entry.startedAt)) {
+            throw new MonitoringChangedError('Server changed during monitoring; the previous observation was discarded.');
+        }
+        const observation = { result, observedAt: new Date().toISOString() };
+        if (entries.get(server) === entry && result.ok && reusable(entry.startedAt)) {
+            entry.identity = latest;
+            entry.observation = observation;
+        }
+        return observation;
+    }).finally(() => {
+        entry.pending = undefined;
+        if (entries.get(server) === entry && !entry.observation) entries.delete(server);
+    });
+    entry.pending = pending;
+    return pending;
+}
+
+const inspections = new Map<string, Entry<Awaited<ReturnType<typeof komodoInspectHealthy>>>>();
+const rcon = new Map<string, Entry<Awaited<ReturnType<typeof rconStatusResult>>>>();
+
+/** Routine monitoring only. Manual diagnostics and disruption gates use fresh probes directly. */
+export function inspectForMonitoring(server: string, fresh = false) {
+    return observe(server, fresh, 'inspect', inspections, komodoInspectHealthy);
+}
+export function rconForMonitoring(server: string, fresh = false) {
+    return observe(server, fresh, 'rcon', rcon, rconStatusResult);
+}

src/lib/server/maintenance/recovery.ts

@@ -4,8 +4,7 @@ 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 { komodoInspectHealthy } from '../integrations/komodo.js';
-import { rconStatusResult } from '../integrations/rcon.js';
+import { inspectForMonitoring, rconForMonitoring, MonitoringChangedError } from '../fleet/health-observations.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';
@@ -223,17 +222,27 @@ async function reconcileActions(): Promise<void> {
 }
 
 interface Observation { healthy: boolean; actionable: boolean; reason: string; containerState: string | null; players?: number; source?: 'rcon' | 'telemetry' }
-async function observe(server: string, operationId?: string): Promise<Observation | null> {
+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 health = await komodoInspectHealthy(server);
+    const inspected = await inspectForMonitoring(server, fresh || !!operationId).catch(error => {
+        if (error instanceof MonitoringChangedError) return null;
+        throw error;
+    });
+    if (!inspected) return null;
+    const health = inspected.result;
     if (suppressed()) return null;
     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 };
     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 rcon = await rconStatusResult(server);
+    const probed = await rconForMonitoring(server, fresh || !!operationId).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);
     if (rcon.ok || latest) return { healthy: true, actionable: false, reason: 'Komodo healthy with fresh game status', containerState: state,
@@ -245,7 +254,7 @@ async function scanServer(server: string, forceScan: boolean): Promise<void> {
     const row = getServer(server);
     if (!row || row.desired_state === 'stopped') return;
     if (scanSuppressed(server) || protectedWork(server)) return;
-    const observation = await observe(server);
+    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 (!scanSuppressed(server, 'health') && observation.healthy) {

src/lib/server/pipeline/worker.ts

@@ -4,10 +4,9 @@ import { sweep } from './sweep.js';
 import { betaTick } from './beta-runner.js';
 import { emit } from './bus.js';
 import type { ServerStatusEntry } from './bus.js';
-import { rconStatusResult } from '../integrations/rcon.js';
 import type { RconFailure } from '../integrations/rcon.js';
 import { getServer, listServers, setServerFields, type ServerRow } from '../fleet/servers.js';
-import { komodoInspectHealthy } from '../integrations/komodo.js';
+import { inspectForMonitoring, rconForMonitoring, 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';
@@ -143,15 +142,23 @@ async function refreshServerStatus(server: string): Promise<number | null> {
     const flush = () => { setServerFields(srv.name, updates); updates = {}; };
     if (!isProvisioning(srv.name)) {
         const epoch = latestGameSubmission(srv.name);
-        const unchanged = () => epoch === latestGameSubmission(srv.name) && !isProvisioning(srv.name)
-            && getServer(srv.name)?.created_at === srv.created_at;
-        const container = restartSettling(srv.name)
-            ? { ok: false as const, state: 'restarting', reason: 'Restart requested' }
-            : await komodoInspectHealthy(srv.name).catch(err => ({ ok: false as const, state: 'unknown', reason: (err as Error).message }));
+        const unchanged = () => {
+            const current = getServer(srv.name);
+            return epoch === latestGameSubmission(srv.name) && !isProvisioning(srv.name)
+                && current?.created_at === srv.created_at && current.desired_state === srv.desired_state
+                && current.native_instance_id === srv.native_instance_id && current.instance_generation === srv.instance_generation;
+        };
+        const inspectionStartedAt = new Date().toISOString();
+        const inspected = restartSettling(srv.name)
+            ? { 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 });
+        if (!inspected) return null;
+        const container = inspected.result;
         // A lifecycle request may be submitted while an inspection is in flight.
-        if (unchanged()) {
+        if (unchanged() && (getServer(srv.name)?.container_checked_at ?? '') <= inspected.observedAt) {
             updates.container_state = container.ok ? 'running' : container.state?.toLowerCase() ?? 'unknown';
-            updates.container_checked_at = new Date().toISOString();
+            updates.container_checked_at = inspected.observedAt;
             // An unavailable inspection is not evidence that the game stopped.
             // Probe it just as a manual RCON check would; explicit lifecycle or
             // unhealthy states still take precedence over game replies.
@@ -161,10 +168,13 @@ async function refreshServerStatus(server: string): Promise<number | null> {
                 updates.last_reply_snippet = null;
             } else if (unavailable || !telemetryEvidence(srv.name)) {
                 flush();
-                const result = await rconStatusResult(srv.name).catch(err => ({ ok: false as const,
-                    reason: 'send-error' as RconFailure, detail: (err as Error).message, rawReply: undefined }));
-                if (unchanged()) {
-                    const now = new Date().toISOString();
+                const probeStartedAt = new Date().toISOString();
+                const probe = await 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 });
+                if (!probe) return null;
+                const result = probe.result;
+                if (unchanged() && (getServer(srv.name)?.last_checked_at ?? '') <= probe.observedAt) {
+                    const now = probe.observedAt;
                     updates.last_checked_at = now;
                     updates.last_reply_snippet = result.rawReply ?? null;
                     if (result.ok) {