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

Share status monitoring snapshots across each fleet pass

Browse Source
This commit is contained in:
josh
2026-10-06 14:47:14 -04:00
parent 562b770725
commit daefb22be6
4 changed files with 195 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/fleet/health-observations.ts
+41 -37
View File
@@ -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<T> {
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<T extends { ok: boolean }>(server: string, fresh: boolean, probe: Probe,
entries: Map<string, Entry<T>>, request: (server: string) => Promise<T>): Promise<Observation<T>> {
entries: Map<string, Entry<T>>, request: (server: string) => Promise<T>, snapshot?: MonitoringSnapshot, batch = false): Promise<Observation<T>> {
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<T extends { ok: boolean }>(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<string, { snapshot?: MonitoringStatus }>();
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<Awaited<ReturnType<typeof nativeHealth>>> | null {
const row = getServer(server);
export function heartbeatInspection(server: string, snapshot?: MonitoringSnapshot): Observation<Awaited<ReturnType<typeof nativeHealth>>> | 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<Awaited<ReturnT
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>> {
export function prepareMonitoringRcon(servers: string[], snapshot?: MonitoringSnapshot): 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);
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<string, () => 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<string, () => 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<string, () => 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);
}