244 lines
13 KiB
TypeScript
244 lines
13 KiB
TypeScript
import { createHash } from 'node:crypto';
|
|
import { config } from '../config.js';
|
|
import { db } from '../db/index.js';
|
|
import { getHost, queueHostOperation, waitHostOperationStatus } from '../hosts/index.js';
|
|
import { nativeInstance, type NativeInstance } from '../backends/native.js';
|
|
import { nativeHealth } from '../backends/native.js';
|
|
import { rconStatusResult, agentRconStatusResult, type AgentRconReply } from '../integrations/rcon.js';
|
|
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;
|
|
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 {}
|
|
export class MonitoringSkippedError 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,
|
|
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);
|
|
if (!sameIdentity(current, latest) || 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 nativeHealth>>>>();
|
|
const rcon = new Map<string, Entry<Awaited<ReturnType<typeof rconStatusResult>>>>();
|
|
|
|
interface MonitoringStatus {
|
|
identity: string;
|
|
container: Observation<Awaited<ReturnType<typeof nativeHealth>>>;
|
|
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 nativeHealth>>> | null {
|
|
const row = getServer(server);
|
|
if (!row?.native_instance_id || !row.host_id) return null;
|
|
const 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
|
|
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;
|
|
return { observedAt: new Date(at).toISOString(), result: state.healthy
|
|
? { ok: true, state: state.state } : { ok: false, state: state.state, reason: `Container ${state.state}` } };
|
|
}
|
|
|
|
type RconObservation = Observation<Awaited<ReturnType<typeof rconStatusResult>>>;
|
|
|
|
/** Prime a whole host before the per-host queue admits just its first two servers. */
|
|
export function prepareMonitoringRcon(servers: string[]): Map<string, () => Promise<RconObservation>> {
|
|
const probes = new Map<string, () => Promise<RconObservation>>();
|
|
const batches = new Map<string, { instances: NativeInstance[]; result: Promise<{ results: Array<{ instanceId: string; generation: number; reply: AgentRconReply }> }> }>();
|
|
for (const server of servers) {
|
|
const instance = nativeInstance(server);
|
|
if (!instance) continue;
|
|
const host = getHost(instance.host_id);
|
|
if (!host?.connected || !host.inventory.capabilities?.includes('rcon.status-batch')) continue;
|
|
const container = heartbeatInspection(server);
|
|
if (container && !container.result.ok && container.result.state !== 'unknown') continue;
|
|
const before = identity(server, 'rcon');
|
|
const request = async () => {
|
|
let batch = batches.get(instance.host_id);
|
|
if (!batch) {
|
|
const instances: NativeInstance[] = [];
|
|
// observe() invokes each request in the same microtask turn;
|
|
// enqueue afterward so cached or shared probes need no batch slot.
|
|
const result = Promise.resolve().then(async () => {
|
|
const operation = withOperationSource('Status monitor', () => queueHostOperation(instance.host_id, 'rcon.status-batch', {
|
|
instances: instances.map(value => ({ instanceId: value.id, generation: value.generation }))
|
|
}));
|
|
// Allow delivery and receipt overhead beyond the agent's 20-second budget.
|
|
const receipt = await waitHostOperationStatus(operation.id, 25_000);
|
|
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 }> };
|
|
});
|
|
batch = { instances, result }; batches.set(instance.host_id, batch);
|
|
}
|
|
batch.instances.push(instance);
|
|
const endpoint = { host: `agent:${instance.host_id}`, port: instance.port };
|
|
try {
|
|
const receipt = await batch.result;
|
|
const replies = receipt.results.filter(value => value.instanceId === instance.id && value.generation === instance.generation);
|
|
if (replies.length !== 1 || typeof replies[0].reply?.text !== 'string') throw new Error('Host status batch omitted a valid instance result.');
|
|
if (replies[0].reply.reason === 'deadline') throw new MonitoringSkippedError('Host status batch budget expired before this server was checked.');
|
|
return agentRconStatusResult(replies[0].reply, endpoint);
|
|
} catch (error) {
|
|
if (error instanceof MonitoringSkippedError) throw error;
|
|
return { ok: false as const, endpoint, reason: 'timeout' as const, detail: (error as Error).message };
|
|
}
|
|
};
|
|
// 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 }));
|
|
probes.set(server, async () => {
|
|
const result = await pending;
|
|
if ('error' in result) throw result.error;
|
|
const current = identity(server, 'rcon');
|
|
if (!sameIdentity(before, current) || newerEvidence(current, Date.parse(result.value.observedAt))) {
|
|
throw new MonitoringChangedError('Server changed before consuming the host status batch.');
|
|
}
|
|
return result.value;
|
|
});
|
|
}
|
|
return probes;
|
|
}
|
|
|
|
/** 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, nativeHealth);
|
|
}
|
|
export function rconForMonitoring(server: string, fresh = false) {
|
|
if (fresh) statuses.delete(server);
|
|
return observe(server, fresh, 'rcon', rcon, rconStatusResult);
|
|
}
|