diff --git a/src/lib/server/fleet/health-observations.ts b/src/lib/server/fleet/health-observations.ts new file mode 100644 index 0000000..7577e63 --- /dev/null +++ b/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 { result: T; observedAt: string } +interface Identity { key: string; port: number | null; checkedAt: string | null } +interface Entry { + identity: Identity; + startedAt: number; + pending?: Promise>; + observation?: Observation; +} + +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(server: string, fresh: boolean, probe: Probe, + entries: Map>, request: (server: string) => Promise): Promise> { + 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 = { 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>>>(); +const rcon = new Map>>>(); + +/** 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); +} diff --git a/src/lib/server/maintenance/recovery.ts b/src/lib/server/maintenance/recovery.ts index 629eb51..4c0abfb 100644 --- a/src/lib/server/maintenance/recovery.ts +++ b/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 { } interface Observation { healthy: boolean; actionable: boolean; reason: string; containerState: string | null; players?: number; source?: 'rcon' | 'telemetry' } -async function observe(server: string, operationId?: string): Promise { +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); - 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 { 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) { diff --git a/src/lib/server/pipeline/worker.ts b/src/lib/server/pipeline/worker.ts index 1c8398c..dd5862e 100644 --- a/src/lib/server/pipeline/worker.ts +++ b/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 { 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 { 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) { diff --git a/tests/unit/health-observations.test.ts b/tests/unit/health-observations.test.ts new file mode 100644 index 0000000..5435216 --- /dev/null +++ b/tests/unit/health-observations.test.ts @@ -0,0 +1,259 @@ +import { randomUUID } from 'node:crypto'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { newFixture, type Fixture } from '../helpers/fixture.js'; + +vi.mock('../../src/lib/server/integrations/komodo.js', () => ({ komodoInspectHealthy: vi.fn() })); +vi.mock('../../src/lib/server/integrations/rcon.js', () => ({ rconStatusResult: vi.fn() })); + +const server = 'managed'; +const now = Date.parse('2026-10-05T16:00:00Z'); +const healthy = { ok: true as const, state: 'running' }; +const playing = { ok: true as const, endpoint: { host: 'localhost', port: 28960 }, rawReply: 'status', + status: { players: 2, map: 'zombie_theater', gametype: 'zom', roster: [] } }; +let fixture: Fixture; + +beforeEach(async () => { + vi.resetModules(); vi.clearAllMocks(); + fixture = await newFixture([]); + vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(now); +}); +afterEach(async () => { vi.useRealTimers(); await fixture.cleanup(); vi.restoreAllMocks(); }); + +async function setup() { + const { db } = await import('../../src/lib/server/db/index.js'); + const database = db(); + database.prepare('INSERT INTO servers(name,komodo_name,created_at,game_port) VALUES(?,?,?,28960)') + .run(server, server, new Date(now).toISOString()); + const { komodoInspectHealthy } = await import('../../src/lib/server/integrations/komodo.js'); + const { rconStatusResult } = await import('../../src/lib/server/integrations/rcon.js'); + const inspect = vi.mocked(komodoInspectHealthy).mockResolvedValue(healthy); + const rcon = vi.mocked(rconStatusResult).mockResolvedValue(playing); + return { database, inspect, rcon, ...await import('../../src/lib/server/fleet/health-observations.js') }; +} + +async function native(api: Awaited>) { + const hosts = await import('../../src/lib/server/hosts/index.js'); + const host = hosts.createHost({ name: 'Monitoring host' }); + hosts.enrollHost(hosts.createEnrollment(host.id).token, { protocolVersion: 1 }); + const instanceId = randomUUID(); + api.database.prepare(`INSERT INTO native_instances(id,server,host_id,library_id,image,port,created_at) + VALUES(?,?,?,'library','image',28960,?)`).run(instanceId, server, host.id, new Date(now).toISOString()); + return { host, instanceId, hosts }; +} + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise(done => { resolve = done; }); + return { promise, resolve }; +} + +describe('shared monitoring observations', () => { + 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; + const request = kind === 'inspect' ? api.inspect : api.rcon; + const first = await call(server); + expect(first.observedAt).toBe(new Date(now).toISOString()); + vi.setSystemTime(now + 29_999); + expect(await call(server)).toEqual(first); + expect(request).toHaveBeenCalledTimes(1); + vi.setSystemTime(now + 30_000); + expect((await call(server)).observedAt).toBe(new Date(now + 30_000).toISOString()); + expect(request).toHaveBeenCalledTimes(2); + }); + + it('shares overlapping monitoring requests, but keeps inspection and RCON separate', async () => { + const api = await setup(); + const pending = deferred(); + api.inspect.mockReturnValueOnce(pending.promise); + const observations = [api.inspectForMonitoring(server), api.inspectForMonitoring(server)]; + await api.rconForMonitoring(server); + expect(api.inspect).toHaveBeenCalledTimes(1); + expect(api.rcon).toHaveBeenCalledTimes(1); + vi.setSystemTime(now + 5000); + pending.resolve(healthy); + expect(await Promise.all(observations)).toEqual(Array(2).fill({ result: healthy, observedAt: new Date(now + 5000).toISOString() })); + }); + + it('timestamps a slow result on completion without making it reusable after its request expires', async () => { + const api = await setup(); + const pending = deferred(); + api.inspect.mockReturnValueOnce(pending.promise); + const observation = api.inspectForMonitoring(server); + await Promise.resolve(); + vi.setSystemTime(now + 31_000); + pending.resolve(healthy); + expect(await observation).toEqual({ result: healthy, observedAt: new Date(now + 31_000).toISOString() }); + await api.inspectForMonitoring(server); + expect(api.inspect).toHaveBeenCalledTimes(2); + }); + + it('reuses the saved completion timestamp while expiring from the earlier request start', async () => { + const api = await setup(); + const pending = deferred(); + api.inspect.mockReturnValueOnce(pending.promise); + const observation = api.inspectForMonitoring(server); + await Promise.resolve(); + vi.setSystemTime(now + 20_000); + pending.resolve(healthy); + const completed = await observation; + api.database.prepare('UPDATE servers SET container_checked_at=? WHERE name=?').run(completed.observedAt, server); + vi.setSystemTime(now + 25_000); + expect(await api.inspectForMonitoring(server)).toEqual(completed); + expect(api.inspect).toHaveBeenCalledTimes(1); + vi.setSystemTime(now + 30_000); + await api.inspectForMonitoring(server); + expect(api.inspect).toHaveBeenCalledTimes(2); + }); + + it('does not reuse observations from a future clock', async () => { + const api = await setup(); + await api.inspectForMonitoring(server); + vi.setSystemTime(now - 1); + await api.inspectForMonitoring(server); + expect(api.inspect).toHaveBeenCalledTimes(2); + }); + + it('does not cache failed results or rejected requests', async () => { + const api = await setup(); + api.inspect.mockResolvedValueOnce({ ok: false, state: 'stopped', reason: 'Stopped' }); + expect((await api.inspectForMonitoring(server)).result.ok).toBe(false); + await api.inspectForMonitoring(server); + expect(api.inspect).toHaveBeenCalledTimes(2); + api.rcon.mockRejectedValueOnce(new Error('Unavailable')); + await expect(api.rconForMonitoring(server)).rejects.toThrow('Unavailable'); + await api.rconForMonitoring(server); + expect(api.rcon).toHaveBeenCalledTimes(2); + }); + + it('forced checks bypass completed results and evict old success when the forced result fails', async () => { + const api = await setup(); + await api.inspectForMonitoring(server); + api.inspect.mockResolvedValueOnce({ ok: false, state: 'stopped', reason: 'Stopped' }); + expect((await api.inspectForMonitoring(server, true)).result.ok).toBe(false); + await api.inspectForMonitoring(server); + expect(api.inspect).toHaveBeenCalledTimes(3); + }); + + it('forced checks bypass wrapper overlap and an older reply cannot replace the newer cache entry', async () => { + const api = await setup(); + const old = deferred(); + api.inspect.mockReturnValueOnce(old.promise); + const first = api.inspectForMonitoring(server); + await Promise.resolve(); + vi.setSystemTime(now + 1000); + const latest = await api.inspectForMonitoring(server, true); + old.resolve(healthy); await first; + expect(await api.inspectForMonitoring(server)).toEqual(latest); + expect(api.inspect).toHaveBeenCalledTimes(2); + }); + + it.each(['registration', 'lifecycle', 'desired state', 'stop'] as const)('invalidates cached observations after a %s change', async change => { + const api = await setup(); + await api.inspectForMonitoring(server); + if (change === 'registration') api.database.prepare('UPDATE servers SET created_at=? WHERE name=?').run(new Date(now + 1).toISOString(), server); + if (change === 'lifecycle') api.database.prepare('INSERT INTO restart_rollout(server,requested_at) VALUES(?,?)').run(server, new Date(now + 1).toISOString()); + if (change === 'desired state') api.database.prepare("UPDATE servers SET desired_state='stopped' WHERE name=?").run(server); + if (change === 'stop') api.database.prepare(`INSERT INTO server_deployments(id,server,action,state,stage,started_at,submitted_at,triggered_by) + VALUES('stop',?,'stop','awaiting-health','Stopping',?,?,'ui')`).run(server, new Date(now).toISOString(), new Date(now + 1).toISOString()); + await api.inspectForMonitoring(server); + expect(api.inspect).toHaveBeenCalledTimes(2); + }); + + it('invalidates integration changes and accepts legacy port discovery by the successful RCON probe', async () => { + const api = await setup(); + await api.inspectForMonitoring(server); + api.database.prepare("UPDATE servers SET komodo_name='replacement' WHERE name=?").run(server); + await api.inspectForMonitoring(server); + expect(api.inspect).toHaveBeenCalledTimes(2); + api.database.prepare('UPDATE servers SET game_port=NULL WHERE name=?').run(server); + api.rcon.mockImplementationOnce(async () => { + api.database.prepare('UPDATE servers SET game_port=28960 WHERE name=?').run(server); + return playing; + }); + expect((await api.rconForMonitoring(server)).result).toEqual(playing); + await api.rconForMonitoring(server); + expect(api.rcon).toHaveBeenCalledTimes(1); + api.database.prepare('UPDATE servers SET game_port=28961 WHERE name=?').run(server); + await api.rconForMonitoring(server); + expect(api.rcon).toHaveBeenCalledTimes(2); + }); + + it.each(['inspect', 'rcon'] as const)('does not reuse %s after a newer manual observation', async kind => { + const api = await setup(); + const call = kind === 'inspect' ? api.inspectForMonitoring : api.rconForMonitoring; + const request = kind === 'inspect' ? api.inspect : api.rcon; + const field = kind === 'inspect' ? 'container_checked_at' : 'last_checked_at'; + await call(server); + vi.setSystemTime(now + 1000); + api.database.prepare(`UPDATE servers SET ${field}=? WHERE name=?`).run(new Date(now + 1000).toISOString(), server); + await call(server); + expect(request).toHaveBeenCalledTimes(2); + }); + + it('rejects a previous generation completing after replacement without discarding the new generation cache', async () => { + const api = await setup(); + const { instanceId } = await native(api); + const pending = deferred(); + api.inspect.mockReturnValueOnce(pending.promise); + const old = api.inspectForMonitoring(server); + const rejected = expect(old).rejects.toThrow('Server changed'); + await Promise.resolve(); + api.database.prepare('UPDATE native_instances SET generation=2 WHERE id=?').run(instanceId); + const current = await api.inspectForMonitoring(server); + pending.resolve(healthy); await rejected; + expect(await api.inspectForMonitoring(server)).toEqual(current); + expect(api.inspect).toHaveBeenCalledTimes(2); + }); + + it('rejects a result when a lifecycle request arrives while its probe is in flight', async () => { + const api = await setup(); + const pending = deferred(); + api.rcon.mockReturnValueOnce(pending.promise); + const old = api.rconForMonitoring(server); + const rejected = expect(old).rejects.toThrow('Server changed'); + await Promise.resolve(); + api.database.prepare('INSERT INTO restart_rollout(server,requested_at) VALUES(?,?)').run(server, new Date(now + 1).toISOString()); + pending.resolve(playing); await rejected; + await api.rconForMonitoring(server); + expect(api.rcon).toHaveBeenCalledTimes(2); + }); + + it('discards a slow result when a newer manual observation was saved during the request', async () => { + const api = await setup(); + const pending = deferred(); + api.rcon.mockReturnValueOnce(pending.promise); + const old = api.rconForMonitoring(server); + const rejected = expect(old).rejects.toBeInstanceOf(api.MonitoringChangedError); + await Promise.resolve(); + vi.setSystemTime(now + 1000); + api.database.prepare('UPDATE servers SET last_checked_at=? WHERE name=?').run(new Date().toISOString(), server); + pending.resolve(playing); await rejected; + await api.rconForMonitoring(server); + expect(api.rcon).toHaveBeenCalledTimes(2); + }); + + it('discards cached native success while its host is offline and probes again after reconnecting', async () => { + const api = await setup(); + const { host, hosts } = await native(api); + await api.inspectForMonitoring(server); + api.database.prepare('UPDATE ec_hosts SET last_heartbeat=NULL WHERE id=?').run(host.id); + await expect(api.inspectForMonitoring(server)).rejects.toThrow('Host agent is offline'); + expect(api.inspect).toHaveBeenCalledTimes(1); + hosts.heartbeat(host.id, { protocolVersion: 1 }); + await api.inspectForMonitoring(server); + expect(api.inspect).toHaveBeenCalledTimes(2); + }); + + it('rejects an otherwise successful result if the native host went offline during the probe', async () => { + const api = await setup(); + const { host } = await native(api); + const pending = deferred(); + api.inspect.mockReturnValueOnce(pending.promise); + const observation = api.inspectForMonitoring(server); + const rejected = expect(observation).rejects.toThrow('Host agent is offline'); + await Promise.resolve(); + api.database.prepare('UPDATE ec_hosts SET last_heartbeat=NULL WHERE id=?').run(host.id); + pending.resolve(healthy); await rejected; + }); +}); diff --git a/tests/unit/server-status.test.ts b/tests/unit/server-status.test.ts index a10fa58..832022e 100644 --- a/tests/unit/server-status.test.ts +++ b/tests/unit/server-status.test.ts @@ -30,6 +30,51 @@ async function setup() { } describe('database fleet and container readiness', () => { + it('shares routine observations without renewing their timestamps and keeps manual probes fresh', async () => { + const { inspect, rcon, fanoutStatus, getServer } = await setup(); + const { inspectForMonitoring, rconForMonitoring } = await import('../../src/lib/server/fleet/health-observations.js'); + vi.useFakeTimers({ toFake: ['Date'] }); + const at = Date.parse('2026-10-05T12:00:00Z'); + vi.setSystemTime(at); + inspect.mockResolvedValue({ ok: true, state: 'running' }); + rcon.mockResolvedValue({ ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: 'status', + status: { players: 2, map: 'zombie_theater', gametype: 'zom', roster: [] } }); + await inspectForMonitoring('managed'); + await rconForMonitoring('managed'); + vi.setSystemTime(at + 10_000); + await fanoutStatus(); + expect(inspect).toHaveBeenCalledTimes(1); + expect(rcon).toHaveBeenCalledTimes(1); + expect(getServer('managed')).toMatchObject({ container_checked_at: new Date(at).toISOString(), last_seen_at: new Date(at).toISOString() }); + + const { POST } = await import('../../src/routes/api/servers/[name]/probe-rcon/+server.js'); + rcon.mockResolvedValue({ ok: false, endpoint: null, reason: 'timeout', detail: 'manual failure' }); + await POST({ params: { name: 'managed' } } as never); + expect(rcon).toHaveBeenCalledTimes(2); + vi.setSystemTime(at + 20_000); + await fanoutStatus(); + expect(rcon).toHaveBeenCalledTimes(3); + expect(getServer('managed')).toMatchObject({ last_error: 'timeout:manual failure', last_seen_at: new Date(at).toISOString() }); + }); + + it('does not overwrite a newer manual observation with an earlier background reply', async () => { + const { db, inspect, rcon, fanoutStatus, getServer } = await setup(); + vi.useFakeTimers({ toFake: ['Date'] }); + const at = Date.parse('2026-10-05T12:00:00Z'); + vi.setSystemTime(at); + inspect.mockResolvedValue({ ok: true, state: 'running' }); + let release!: (result: Awaited>) => void; + rcon.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + const pending = fanoutStatus(); + await vi.waitFor(() => expect(rcon).toHaveBeenCalledTimes(1)); + vi.setSystemTime(at + 10_000); + db.prepare("UPDATE servers SET last_checked_at=?,last_error='timeout:manual failure' WHERE name='managed'").run(new Date().toISOString()); + release({ ok: true, endpoint: { host: 'localhost', port: 28960 }, rawReply: 'old reply', + status: { players: 0, map: 'zombie_theater', gametype: 'zom', roster: [] } }); + await pending; + expect(getServer('managed')).toMatchObject({ last_error: 'timeout:manual failure', last_seen_at: null, last_checked_at: new Date(at + 10_000).toISOString() }); + }); + it('refreshes a ten-server fleet while one check is slow, without duplicating or exceeding four checks', async () => { const { db, inspect, rcon, fanoutStatus, getServer } = await setup(); const { subscribe } = await import('../../src/lib/server/pipeline/bus.js'); @@ -391,6 +436,7 @@ describe('database fleet and container readiness', () => { it.each(['unknown', undefined, 'throw'])('automatically recovers through RCON when inspection returns %s', async state => { const { inspect, rcon, fanoutStatus, getServer } = await setup(); + vi.useFakeTimers({ toFake: ['Date'] }); const { GET } = await import('../../src/routes/api/servers/+server.js'); const { subscribe } = await import('../../src/lib/server/pipeline/bus.js'); if (state === 'throw') inspect.mockRejectedValue(new Error('Inspection timed out')); @@ -409,6 +455,7 @@ describe('database fleet and container readiness', () => { })] })); rcon.mockResolvedValue({ ok: false, endpoint: null, reason: 'timeout', detail: 'no reply' }); + vi.setSystemTime(Date.now() + 30_000); await fanoutStatus(); expect((await (await GET()).json()).servers[0]).toMatchObject({ health: 'stale', last_error: 'timeout:no reply' }); });