diff --git a/src/lib/server/fleet/health-observations.ts b/src/lib/server/fleet/health-observations.ts index de1d5b6..4509006 100644 --- a/src/lib/server/fleet/health-observations.ts +++ b/src/lib/server/fleet/health-observations.ts @@ -8,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 type { MonitoringSnapshot } from './monitoring-snapshot.js'; import { ONLINE_MAX_MS } from '../../server-freshness.js'; const MAX_AGE_MS = 30_000; @@ -25,19 +26,20 @@ interface Entry { export class MonitoringChangedError extends Error {} export class MonitoringSkippedError extends Error {} -function identity(server: string, probe: Probe): Identity { - const row = getServer(server); +function identity(server: string, probe: Probe, snapshot?: MonitoringSnapshot): Identity { + const row = snapshot ? snapshot.get(server)?.server : 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')) { + const native = snapshot ? snapshot.get(server)?.native : row.native_instance_id ? nativeInstance(server) : null; + if (row.native_instance_id && (!native || (snapshot ? snapshot.get(server)?.host : 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 + const savedLifecycle = snapshot?.get(server)?.lifecycle; + const lifecycle = snapshot ? savedLifecycle ? { id: savedLifecycle.id, action: savedLifecycle.action, submitted_at: savedLifecycle.submitted_at } : null : 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, + key: JSON.stringify([row.created_at, row.desired_state, snapshot ? snapshot.get(server)?.submission : latestGameSubmission(server), lifecycle, native ? [native.host_id, native.id, native.generation] : null, probe === 'rcon' ? [row.rcon_host, config.rcon] : null]), port: probe === 'rcon' ? native?.port ?? row.game_port : null, @@ -57,9 +59,9 @@ function reusable(startedAt: number): boolean { } async function observe(server: string, fresh: boolean, probe: Probe, - entries: Map>, request: (server: string) => Promise): Promise> { + entries: Map>, request: (server: string) => Promise, snapshot?: MonitoringSnapshot, batch = false): Promise> { let current: Identity; - try { current = identity(server, probe); } + try { current = identity(server, probe, snapshot); } catch (error) { entries.delete(server); throw error; } const saved = entries.get(server); const savedAt = saved?.observation ? Date.parse(saved.observation.observedAt) : saved?.startedAt ?? 0; @@ -75,7 +77,8 @@ async function observe(server: string, fresh: boolean 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); + if (!batch) snapshot?.refresh(server); + const latest = identity(server, probe, snapshot); if (!sameIdentity(current, latest) || newerEvidence(latest, entry.startedAt)) { throw new MonitoringChangedError('Server changed during monitoring; the previous observation was discarded.'); } @@ -105,17 +108,17 @@ interface MonitoringStatus { // this snapshot must never dispatch a request or renew an observation's age. const statuses = new Map(); -export function monitoringIdentity(server: string): string { - const container = identity(server, 'inspect'); - const game = identity(server, 'rcon'); +export function monitoringIdentity(server: string, snapshot?: MonitoringSnapshot): string { + const container = identity(server, 'inspect', snapshot); + const game = identity(server, 'rcon', snapshot); // 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) { +export function monitoringStatusWriter(server: string, snapshot?: MonitoringSnapshot) { let before: string; - try { before = identity(server, 'inspect').key; } + try { before = identity(server, 'inspect', snapshot).key; } catch { return () => {}; } const slot = { snapshot: statuses.get(server)?.snapshot }; statuses.delete(server); @@ -126,16 +129,16 @@ export function monitoringStatusWriter(server: string) { // network requests are still in flight. if (statuses.get(server) !== slot) return; try { - if (identity(server, 'inspect').key !== before) return; + if (identity(server, 'inspect', snapshot).key !== before) return; const current = slot.snapshot; if (current && current.container.observedAt > container.observedAt) return; - slot.snapshot = { identity: monitoringIdentity(server), container, game }; + slot.snapshot = { identity: monitoringIdentity(server, snapshot), container, game }; } catch { statuses.delete(server); } }; } /** Latest completed status-monitor pass, with stale or superseded evidence excluded. */ -export function monitoringStatus(server: string): MonitoringStatus | null { +export function monitoringStatus(server: string, snapshot?: MonitoringSnapshot): MonitoringStatus | null { const saved = statuses.get(server)?.snapshot; if (!saved) return null; const fresh = (at: string) => { @@ -143,28 +146,28 @@ export function monitoringStatus(server: string): MonitoringStatus | null { return age >= 0 && age <= ONLINE_MAX_MS; }; try { - if (saved.identity !== monitoringIdentity(server) + if (saved.identity !== monitoringIdentity(server, snapshot) || !fresh(saved.container.observedAt) - || newerEvidence(identity(server, 'inspect'), Date.parse(saved.container.observedAt))) return null; + || newerEvidence(identity(server, 'inspect', snapshot), 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; + && !newerEvidence(identity(server, 'rcon', snapshot), 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>> | null { - const row = getServer(server); +export function heartbeatInspection(server: string, snapshot?: MonitoringSnapshot): Observation>> | null { + const row = snapshot ? snapshot.get(server)?.server : getServer(server); if (!row?.native_instance_id || !row.host_id) return null; - const host = getHost(row.host_id); + const host = snapshot ? snapshot.get(server)?.host : getHost(row.host_id); if (!host?.connected || !host.inventory.capabilities?.includes('instance.heartbeat-state')) return null; const state = host.inventory.instanceStates?.find(state => state.instanceId === row.native_instance_id && state.generation === row.instance_generation); if (!state) return null; const at = Date.parse(state.observedAt); - const lifecycle = db().prepare(`SELECT MAX(MAX(submitted_at,COALESCE(finished_at,submitted_at))) AS at + const lifecycle = snapshot ? { at: snapshot.get(server)?.lifecycleAt } : db().prepare(`SELECT MAX(MAX(submitted_at,COALESCE(finished_at,submitted_at))) AS at FROM server_deployments WHERE server=? AND server_deleted_at IS NULL`).get(server) as { at: string | null }; if (!reusable(at) || at < Date.parse(row.created_at) || at < Date.parse(lifecycle.at ?? '') - || at < Date.parse(latestGameSubmission(server) ?? '') || at < Date.parse(row.container_checked_at ?? '')) return null; + || at < Date.parse((snapshot ? snapshot.get(server)?.submission : latestGameSubmission(server)) ?? '') || at < Date.parse(row.container_checked_at ?? '')) return null; return { observedAt: new Date(at).toISOString(), result: state.healthy ? { ok: true, state: state.state } : { ok: false, state: state.state, reason: `Container ${state.state}` } }; } @@ -172,17 +175,17 @@ export function heartbeatInspection(server: string): Observation>>; /** Prime a whole host before the per-host queue admits just its first two servers. */ -export function prepareMonitoringRcon(servers: string[]): Map Promise> { +export function prepareMonitoringRcon(servers: string[], snapshot?: MonitoringSnapshot): Map Promise> { const probes = new Map Promise>(); const batches = new Map }> }>(); for (const server of servers) { - const instance = nativeInstance(server); + const instance = snapshot ? snapshot.get(server)?.native : nativeInstance(server); if (!instance) continue; - const host = getHost(instance.host_id); + const host = snapshot ? snapshot.get(server)?.host : getHost(instance.host_id); if (!host?.connected || !host.inventory.capabilities?.includes('rcon.status-batch')) continue; - const container = heartbeatInspection(server); + const container = heartbeatInspection(server, snapshot); if (container && !container.result.ok && container.result.state !== 'unknown') continue; - const before = identity(server, 'rcon'); + const before = identity(server, 'rcon', snapshot); const request = async () => { let batch = batches.get(instance.host_id); if (!batch) { @@ -195,6 +198,7 @@ export function prepareMonitoringRcon(servers: string[]): Map Prom })); // Allow delivery and receipt overhead beyond the agent's 20-second budget. const receipt = await waitHostOperationStatus(operation.id, 25_000); + snapshot?.refresh(); if (receipt.state !== 'completed') throw new Error(receipt.error ?? 'Host status batch did not complete.'); return receipt.result as { results: Array<{ instanceId: string; generation: number; reply: AgentRconReply }> }; }); @@ -216,11 +220,11 @@ export function prepareMonitoringRcon(servers: string[]): Map Prom // Observe before dispatch, including servers still waiting for their // queue slot. Consume failures too, so a failed batch is never retried // once per server or left as an unhandled rejected promise. - const pending = observe(server, false, 'rcon', rcon, request).then(value => ({ value }), error => ({ error })); + const pending = observe(server, false, 'rcon', rcon, request, snapshot, true).then(value => ({ value }), error => ({ error })); probes.set(server, async () => { const result = await pending; if ('error' in result) throw result.error; - const current = identity(server, 'rcon'); + const current = identity(server, 'rcon', snapshot); if (!sameIdentity(before, current) || newerEvidence(current, Date.parse(result.value.observedAt))) { throw new MonitoringChangedError('Server changed before consuming the host status batch.'); } @@ -231,13 +235,13 @@ export function prepareMonitoringRcon(servers: string[]): Map Prom } /** Routine monitoring only. Manual diagnostics and disruption gates use fresh probes directly. */ -export function inspectForMonitoring(server: string, fresh = false) { +export function inspectForMonitoring(server: string, fresh = false, snapshot?: MonitoringSnapshot) { if (fresh) statuses.delete(server); - const heartbeat = !fresh && heartbeatInspection(server); + const heartbeat = !fresh && heartbeatInspection(server, snapshot); if (heartbeat) return Promise.resolve(heartbeat); - return observe(server, fresh, 'inspect', inspections, nativeHealth); + return observe(server, fresh, 'inspect', inspections, nativeHealth, snapshot); } -export function rconForMonitoring(server: string, fresh = false) { +export function rconForMonitoring(server: string, fresh = false, snapshot?: MonitoringSnapshot) { if (fresh) statuses.delete(server); - return observe(server, fresh, 'rcon', rcon, rconStatusResult); + return observe(server, fresh, 'rcon', rcon, rconStatusResult, snapshot); } diff --git a/src/lib/server/fleet/monitoring-snapshot.ts b/src/lib/server/fleet/monitoring-snapshot.ts new file mode 100644 index 0000000..371dcb3 --- /dev/null +++ b/src/lib/server/fleet/monitoring-snapshot.ts @@ -0,0 +1,91 @@ +import { db } from '../db/index.js'; +import { ensureHostSchema } from '../hosts/schema.js'; +import { HEARTBEAT_MAX_AGE_MS } from '../hosts/index.js'; +import { parseHostInventory } from '../hosts/request-context.js'; +import type { Host } from '../../hosts.js'; +import type { NativeInstance } from '../backends/native.js'; +import type { ServerRow } from './servers.js'; +import { parseGameEvent, type GameTelemetry } from '../../telemetry.js'; +import { ONLINE_MAX_MS } from '../../server-freshness.js'; +import { serverObservation } from './status.js'; + +const nativeFields = ['id','server','host_id','generation','library_id','image','bind_address','port','public_address','public_port','created_at','retired_at']; +const sql = `SELECT s.*, n.id AS native_instance_id,n.host_id,n.generation AS instance_generation,n.library_id AS game_library_id, + CASE WHEN n.id IS NOT NULL THEN json_object(${nativeFields.map(key => `'${key}',n.${key}`).join(',')}) END AS instance_json, + h.inventory AS host_inventory,h.last_heartbeat AS host_heartbeat, + EXISTS(SELECT 1 FROM ec_host_credentials c WHERE c.host_id=h.id AND c.revoked_at IS NULL) AS credential_active, + (SELECT json_object('id',d.id,'action',d.action,'state',d.state,'submitted_at',d.submitted_at) + FROM server_deployments d WHERE d.server=s.name AND d.server_deleted_at IS NULL ORDER BY d.rowid DESC LIMIT 1) AS lifecycle_json, + (SELECT MAX(MAX(d.submitted_at,COALESCE(d.finished_at,d.submitted_at))) FROM server_deployments d + WHERE d.server=s.name AND d.server_deleted_at IS NULL) AS lifecycle_at, + (SELECT MAX(at) FROM (SELECT MAX(d.submitted_at) AS at FROM server_deployments d + WHERE d.server=s.name AND d.server_deleted_at IS NULL AND d.action IN ('create','start','restart','rebuild') + UNION ALL SELECT r.requested_at)) AS submission, + r.requested_at AS restart_at,r.healthy_at AS restart_healthy,t.received_at AS telemetry_at,t.payload AS telemetry_json + FROM servers s LEFT JOIN native_instances n ON n.server=s.name AND n.retired_at IS NULL + LEFT JOIN ec_hosts h ON h.id=n.host_id AND h.deleted_at IS NULL + LEFT JOIN restart_rollout r ON r.server=s.name LEFT JOIN server_telemetry t ON t.server=s.name`; +interface Row extends ServerRow { + instance_json: string | null; host_inventory: string | null; host_heartbeat: string | null; credential_active: number; + lifecycle_json: string | null; lifecycle_at: string | null; submission: string | null; + restart_at: string | null; restart_healthy: string | null; telemetry_at: string | null; telemetry_json: string | null; +} +interface Record { + server: ServerRow; native: NativeInstance | null; + host: Pick | null; + lifecycle: { id: string; action: string; state: string; submitted_at: string | null } | null; + lifecycleAt: string | null; submission: string | null; telemetry: GameTelemetry | null; + restartAt: string | null; restartHealthy: string | null; +} + +/** A pass shares its reads; each asynchronous evidence boundary refreshes in one joined query. */ +export class MonitoringSnapshot { + private records = new Map(); + constructor() { ensureHostSchema(db()); this.refresh(); } + refresh(server?: string): void { + const rows = server === undefined ? db().prepare(sql + ' WHERE s.destroyed_at IS NULL ORDER BY s.name').all() as Row[] + : db().prepare(sql + ' WHERE s.name=? AND s.destroyed_at IS NULL').all(server) as Row[]; + if (server === undefined) this.records.clear(); else this.records.delete(server); + for (const row of rows) { + const { instance_json,host_inventory,host_heartbeat,credential_active,lifecycle_json,lifecycle_at,submission, + restart_at,restart_healthy,telemetry_at,telemetry_json,...fields } = row; + const connected = !!credential_active && !!host_heartbeat && Date.now() - Date.parse(host_heartbeat) <= HEARTBEAT_MAX_AGE_MS; + let telemetry: GameTelemetry | null = null; + if (telemetry_json) try { + const saved = JSON.parse(telemetry_json) as GameTelemetry; + const event = parseGameEvent(saved); + if (event.server === row.name) telemetry = { ...event, receivedAt: telemetry_at!, timing: saved.timing }; + } catch { /* Invalid telemetry cannot establish health. */ } + this.records.set(row.name, { server: fields, native: instance_json ? JSON.parse(instance_json) : null, + host: host_inventory ? { connected, connectionState: connected ? 'online' : 'offline', + inventory: parseHostInventory(row.host_id!, host_inventory) } : null, + lifecycle: lifecycle_json ? JSON.parse(lifecycle_json) : null, lifecycleAt: lifecycle_at, submission, telemetry, + restartAt: restart_at, restartHealthy: restart_healthy }); + } + } + get(server: string) { return this.records.get(server); } + servers() { return [...this.records.values()].map(record => record.server); } + provisioning(server: string) { + const operation = this.get(server)?.lifecycle; + return !!operation && ['create','rebuild','start'].includes(operation.action) && ['queued','running','awaiting-health'].includes(operation.state); + } + settling(server: string) { + const record = this.get(server); + return !!record?.restartAt && !record.restartHealthy && Date.now() - Date.parse(record.restartAt) < 15_000; + } + telemetryEvidence(server: string) { + const record = this.get(server), report = record?.telemetry; + if (!report) return null; + const received = Date.parse(report.receivedAt), age = Date.now() - received; + return Number.isFinite(age) && age >= 0 && age <= ONLINE_MAX_MS + && (!record?.submission || received > Date.parse(record.submission)) ? report : null; + } + observe = (server: ServerRow) => serverObservation(server, { + provisioning: this.provisioning(server.name), settling: this.settling(server.name), + submission: this.get(server.name)?.submission ?? null, telemetry: this.get(server.name)?.telemetry ?? null + }); + patch(server: string, fields: Partial) { + const record = this.get(server); + if (record) record.server = { ...record.server, ...fields }; + } +} diff --git a/src/lib/server/pipeline/worker.ts b/src/lib/server/pipeline/worker.ts index 4d8b870..3bf30c7 100644 --- a/src/lib/server/pipeline/worker.ts +++ b/src/lib/server/pipeline/worker.ts @@ -6,18 +6,17 @@ import { betaTick } from './beta-runner.js'; 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, monitoringStatus, monitoringStatusWriter, MonitoringChangedError, MonitoringSkippedError } 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'; -import { isProvisioning } from '../fleet/provisioning.js'; +import { setServerFields, type ServerRow } from '../fleet/servers.js'; +import { inspectForMonitoring, rconForMonitoring, prepareMonitoringRcon, monitoringStatus, monitoringStatusWriter, monitoringIdentity, MonitoringChangedError, MonitoringSkippedError } from '../fleet/health-observations.js'; +import { serverObservation } from '../fleet/status.js'; import { STATUS_REFRESH_MS } from '../../server-freshness.js'; import { recordMetrics } from '../fleet/metrics.js'; import { selectedZombieMap } from '../../map-aliases.js'; import { runMaintenance } from '../maintenance/index.js'; import { cleanupControllerLibraries } from '../game-files/index.js'; import { trackActivity } from './activity.js'; +import { MonitoringSnapshot } from '../fleet/monitoring-snapshot.js'; +import { withHostRequestContext } from '../hosts/request-context.js'; import { withOperationSource } from '../hosts/operation-context.js'; const TICK_MS = 15_000; @@ -104,7 +103,7 @@ function drainStatusQueue(host: string, queue: StatusQueue): void { if (!queue.active && !queue.queued.length) statusQueues.delete(host); } -function queueStatus(srv: ServerRow, probe?: () => ReturnType): Promise { +function queueStatus(srv: ServerRow, snapshot: MonitoringSnapshot, probe?: () => ReturnType): Promise { const server = srv.name; const pending = pendingStatus.get(server); if (pending) return pending; @@ -118,10 +117,10 @@ function queueStatus(srv: ServerRow, probe?: () => ReturnType { let capacity: number | null = null; try { - capacity = await withOperationSource('Status monitor', () => refreshServerStatus(server, probe)); + capacity = await withOperationSource('Status monitor', () => refreshServerStatus(server, snapshot, probe)); // Publish this server's latest evidence without waiting for the fleet. - const current = getServer(server); - if (current) publishStatus(statusEntry(current)); + const current = snapshot.get(server)?.server; + if (current) publishStatus(statusEntry(current, snapshot.observe)); } catch (err) { console.error(`[worker] status error for ${server}:`, err); } finally { pendingStatus.delete(server); finish(capacity); } }); @@ -129,19 +128,24 @@ function queueStatus(srv: ServerRow, probe?: () => ReturnType { - const servers = listServers(); +export function fanoutStatus(): Promise { + return withHostRequestContext(statusPass); +} + +async function statusPass(): Promise { + const snapshot = new MonitoringSnapshot(); + const servers = snapshot.servers(); if (!servers.length) return; const probes = prepareMonitoringRcon(servers.filter(server => !pendingStatus.has(server.name) - && !isProvisioning(server.name) && !restartSettling(server.name) && !telemetryEvidence(server.name)).map(server => server.name)); - const results = await Promise.all(servers.map(server => queueStatus(server, probes.get(server.name)))); + && !snapshot.provisioning(server.name) && !snapshot.settling(server.name) && !snapshot.telemetryEvidence(server.name)).map(server => server.name), snapshot); + const results = await Promise.all(servers.map(server => queueStatus(server, snapshot, probes.get(server.name)))); const capacities = new Map(servers.map((server, i) => [server.name, results[i]])); try { // Metrics remain complete fleet collections, using freshness at capture // time rather than the status saved at the beginning of a long scan. - const observe = observationReader(); - recordMetrics(listServers().map(srv => { - const current = observe(srv); + snapshot.refresh(); + recordMetrics(snapshot.servers().map(srv => { + const current = snapshot.observe(srv); return { server: srv.name, game: srv.game ?? 'unknown', map: selectedZombieMap(srv.game, current.last_map ?? srv.map, srv.map_slug)?.friendly ?? current.last_map ?? srv.map ?? 'Unknown map', players: current.health === 'online' ? current.last_players : null, @@ -150,33 +154,38 @@ export async function fanoutStatus(): Promise { } catch (err) { console.error('[worker] metrics error:', err); } } -async function refreshServerStatus(server: string, probeStatus?: () => ReturnType): Promise { - const srv = getServer(server); +async function refreshServerStatus(server: string, snapshot: MonitoringSnapshot, probeStatus?: () => ReturnType): Promise { + snapshot.refresh(server); + const srv = snapshot.get(server)?.server; if (!srv) return null; let capacity: number | null = null; let updates: Partial> = {}; - const flush = () => { setServerFields(srv.name, updates); updates = {}; }; - if (!isProvisioning(srv.name)) { - const previousGame = monitoringStatus(srv.name)?.game; - const publishObservation = monitoringStatusWriter(srv.name); + const flush = () => { setServerFields(srv.name, updates); snapshot.patch(srv.name, updates as Partial); updates = {}; }; + if (!snapshot.provisioning(srv.name)) { + const previousGame = monitoringStatus(srv.name, snapshot)?.game; + const publishObservation = monitoringStatusWriter(srv.name, snapshot); let gameObservation: Awaited> | undefined; - const settling = restartSettling(srv.name); - const epoch = latestGameSubmission(srv.name); + const settling = snapshot.settling(srv.name); + const epoch = snapshot.get(srv.name)?.submission; + let identity: string | undefined; + try { identity = monitoringIdentity(srv.name, snapshot); } catch { /* Offline hosts still need an unavailable sample. */ } const unchanged = () => { - const current = getServer(srv.name); - return epoch === latestGameSubmission(srv.name) && !isProvisioning(srv.name) + const current = snapshot.get(srv.name)?.server; + if (identity) try { if (identity !== monitoringIdentity(srv.name, snapshot)) return false; } catch { return false; } + return epoch === snapshot.get(srv.name)?.submission && !snapshot.provisioning(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 = settling ? { result: { ok: false as const, state: 'restarting', reason: 'Restart requested' }, observedAt: inspectionStartedAt } - : await inspectForMonitoring(srv.name).catch(err => err instanceof MonitoringChangedError ? null + : await inspectForMonitoring(srv.name, false, snapshot).catch(err => err instanceof MonitoringChangedError ? null : { result: { ok: false as const, state: 'unknown', reason: (err as Error).message }, observedAt: inspectionStartedAt }); + snapshot.refresh(srv.name); if (!inspected) return null; const container = inspected.result; // A lifecycle request may be submitted while an inspection is in flight. - if (unchanged() && (getServer(srv.name)?.container_checked_at ?? '') <= inspected.observedAt) { + if (unchanged() && (snapshot.get(srv.name)?.server?.container_checked_at ?? '') <= inspected.observedAt) { updates.container_state = container.ok ? 'running' : container.state?.toLowerCase() ?? 'unknown'; updates.container_checked_at = inspected.observedAt; // An unavailable inspection is not evidence that the game stopped. @@ -187,10 +196,11 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp updates.last_error = null; updates.rcon_failures = 0; updates.last_reply_snippet = null; - } else if (unavailable || !telemetryEvidence(srv.name)) { + } else if (unavailable || !snapshot.telemetryEvidence(srv.name)) { flush(); const probeStartedAt = new Date().toISOString(); - const probe = await (probeStatus ? probeStatus() : rconForMonitoring(srv.name)).catch(err => { + const probe = await (probeStatus ? probeStatus() : rconForMonitoring(srv.name, false, snapshot)).catch(err => { + snapshot.refresh(srv.name); if (err instanceof MonitoringSkippedError) { // The container was checked, but game evidence retains its original age. if (unchanged()) publishObservation(inspected, previousGame); @@ -199,9 +209,10 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp if (err instanceof MonitoringChangedError) return null; return { result: { ok: false as const, endpoint: null, reason: 'send-error' as RconFailure, detail: (err as Error).message, rawReply: undefined }, observedAt: probeStartedAt }; }); + snapshot.refresh(srv.name); if (!probe) return null; const result = probe.result; - const previous = getServer(srv.name); + const previous = snapshot.get(srv.name)?.server; if (unchanged() && previous && (previous.last_checked_at ?? '') <= probe.observedAt) { gameObservation = probe; const now = probe.observedAt; @@ -234,9 +245,8 @@ async function refreshServerStatus(server: string, probeStatus?: () => ReturnTyp updates.rcon_failures = 0; updates.last_reply_snippet = null; } - flush(); - const current = getServer(srv.name) ?? srv; - const status = serverObservation(current); + const current = { ...(snapshot.get(srv.name)?.server ?? srv), ...updates } as ServerRow; + const status = snapshot.observe(current); // Occupancy timers and match clearing follow the same source as the UI. if (status.health === 'online' && status.last_players !== null) { if (status.last_players > 0) { diff --git a/tests/unit/host-status-batch.test.ts b/tests/unit/host-status-batch.test.ts index b96480f..0093d65 100644 --- a/tests/unit/host-status-batch.test.ts +++ b/tests/unit/host-status-batch.test.ts @@ -45,6 +45,24 @@ async function setup(count = 5, hostCount = 1) { } describe('host status observations', () => { + it('bounds monitoring reads below five per server across thirty instances', async () => { + const api = await setup(10, 3); + const hosts = await import('../../src/lib/server/hosts/index.js'); + const wait = vi.spyOn(hosts, 'waitHostOperationStatus').mockImplementation(async id => { + const operation = hosts.getHostOperation(id)!; + return { ...operation, state: 'completed', result: { results: (operation.payload.instances as { instanceId: string; generation: number }[]) + .map(target => ({ ...target, reply: { ok: true, text: 'map: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n' } })) } }; + }); + const prepare = vi.spyOn(api.db, 'prepare'); + try { + await api.fanoutStatus(); + const reads = prepare.mock.calls.filter(([sql]) => /^\s*(SELECT|WITH)\b/i.test(sql)); + expect(reads.length).toBeLessThan(30 * 5); + expect(reads.filter(([sql]) => sql.includes('instance_json')).length).toBeLessThanOrEqual(30 * 3 + 5); + expect(api.getServer('server-2-9')?.last_players).toBe(0); + } finally { prepare.mockRestore(); wait.mockRestore(); } + }); + it('uses one batch per host for a fifteen-server pass and shares overlapping scans', async () => { const api = await setup(5, 3); const hosts = await import('../../src/lib/server/hosts/index.js');