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

fix: honor Komodo health and prefer fresh telemetry across fleet status

Browse Source
This commit is contained in:
josh
2026-09-27 15:34:30 -04:00
parent 46a89ea4b0
commit 8ea696417f
27 changed files with 488 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/pipeline/bus.ts
+8 -2
View File
@@ -1,6 +1,7 @@
import type { StatusPlayer, GameObservation } from '../../server-observation.js';
import { EventEmitter } from 'node:events';
import type { GameTelemetry } from '../../telemetry.js';
import type { RconEndpoint, RconFailure, RconPlayer } from '../integrations/rcon.js';
import type { RconEndpoint, RconFailure } from '../integrations/rcon.js';
import type { Health } from '../fleet/health.js';
// One in-process event bus, multiplexed by the SSE endpoints. Kept separate
@@ -9,10 +10,15 @@ import type { Health } from '../fleet/health.js';
export interface ServerStatusEntry {
server: string;
containerState?: string | null;
containerCheckedAt?: string | null;
telemetryAfter?: string | null;
rconStatus?: GameObservation;
telemetry?: GameTelemetry;
players: number | null;
map: string | null;
gametype: string | null;
roster: RconPlayer[] | null;
roster: StatusPlayer[] | null;
health: Health;
lastSeenAt: string | null;
lastError: string | null;
src/lib/server/pipeline/worker.ts
+61 -142
View File
@@ -1,3 +1,4 @@
import { config } from '../config.js';
import { processProvisioning } from '../provisioning/worker.js';
import { sweep } from './sweep.js';
import { betaTick } from './beta-runner.js';
@@ -8,7 +9,8 @@ import type { RconFailure } from '../integrations/rcon.js';
import { getServer, listServers, setServerField } from '../fleet/servers.js';
import { komodoInspectHealthy } from '../integrations/komodo.js';
import { restartSettling } from '../provisioning/index.js';
import { classifyHealth } from '../fleet/health.js';
import { serverObservation } from '../fleet/status.js';
import { telemetryEvidence, latestGameSubmission } from '../fleet/status-evidence.js';
import { isProvisioning } from '../fleet/provisioning.js';
import { STATUS_REFRESH_MS } from '../../server-freshness.js';
import { recordMetrics } from '../fleet/metrics.js';
@@ -68,167 +70,84 @@ export function fanoutStatus(): Promise<void> {
async function refreshStatus(): Promise<void> {
const servers = listServers();
if (servers.length === 0) return;
if (!servers.length) return;
const entries: ServerStatusEntry[] = new Array(servers.length);
const capacities = new Map<string, number | null>();
let cursor = 0;
async function pump(): Promise<void> {
while (true) {
async function pump() {
while (cursor < servers.length) {
const i = cursor++;
if (i >= servers.length) return;
const srv = servers[i];
// Creation performs its own container, RCON, and map verification.
if (isProvisioning(srv.name)) {
if (!isProvisioning(srv.name)) {
const epoch = latestGameSubmission(srv.name);
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 }));
// A lifecycle request may be submitted while an inspection is in flight.
if (epoch === latestGameSubmission(srv.name) && !isProvisioning(srv.name)) {
setServerField(srv.name, 'container_state', container.ok ? 'running' : container.state?.toLowerCase() ?? 'unknown');
setServerField(srv.name, 'container_checked_at', new Date().toISOString());
if (!container.ok) {
setServerField(srv.name, 'last_error', null);
setServerField(srv.name, 'last_reply_snippet', null);
} else if (!telemetryEvidence(srv.name)) {
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 (epoch === latestGameSubmission(srv.name) && !isProvisioning(srv.name)) {
const now = new Date().toISOString();
setServerField(srv.name, 'last_checked_at', now);
setServerField(srv.name, 'last_reply_snippet', result.rawReply ?? null);
if (result.ok) {
capacities.set(srv.name, result.status.capacity ?? null);
setServerField(srv.name, 'last_players', result.status.players);
setServerField(srv.name, 'last_map', result.status.map);
setServerField(srv.name, 'last_gametype', result.status.gametype);
setServerField(srv.name, 'last_roster', JSON.stringify(result.status.roster));
setServerField(srv.name, 'last_seen_at', now);
setServerField(srv.name, 'last_error', null);
} else setServerField(srv.name, 'last_error', encodeError(result.reason, result.detail));
}
}
}
} else {
setServerField(srv.name, 'last_error', null);
setServerField(srv.name, 'last_reply_snippet', null);
entries[i] = {
server: srv.name, players: null, map: srv.map, gametype: null,
roster: null, health: 'provisioning', lastSeenAt: srv.last_seen_at,
lastError: null, endpoint: null, reason: null, replySnippet: null,
activeSince: null, inactiveSince: null, gameStartedAt: null, gameEndedAt: null
};
continue;
}
const container = restartSettling(srv.name)
? { ok: false as const, state: 'restarting', reason: 'Restart requested' }
: await komodoInspectHealthy(srv.name).catch((err) => ({
ok: false as const, reason: (err as Error).message
}));
const containerState = container.ok ? 'running'
: ('state' in container && typeof container.state === 'string' ? container.state.toLowerCase() : 'unknown');
setServerField(srv.name, 'container_state', containerState);
if (!container.ok) {
// Container readiness precedes RCON readiness. Old replies and
// failures describe the previous process, not this startup.
setServerField(srv.name, 'last_error', null);
setServerField(srv.name, 'last_reply_snippet', null);
entries[i] = {
server: srv.name, players: null, map: srv.map, gametype: null,
roster: null, health: classifyHealth(null, null, containerState),
lastSeenAt: srv.last_seen_at, lastError: null, endpoint: null,
reason: null, replySnippet: null, activeSince: null,
inactiveSince: null, gameStartedAt: null, gameEndedAt: null
};
continue;
}
const result = await rconStatusResult(srv.name).catch((err) => ({
ok: false as const,
endpoint: null,
reason: 'send-error' as RconFailure,
detail: (err as Error).message,
rawReply: undefined as string | undefined
}));
const now = new Date().toISOString();
if (result.ok) {
capacities.set(srv.name, result.status.capacity ?? null);
// Compare against the previously persisted count BEFORE we
// overwrite last_players — occupancy transitions are anchored
// to that boundary.
const prevPlayers = srv.last_players;
const nextPlayers = result.status.players;
let activeSince = srv.active_since;
let inactiveSince = srv.inactive_since;
// Telemetry can arrive while the RCON request is in flight.
const current = getServer(srv.name) ?? srv;
const gameStartedAt = nextPlayers === 0 ? null : current.game_started_at;
const gameEndedAt = nextPlayers === 0 ? null : current.game_ended_at;
if (nextPlayers === 0) {
const current = getServer(srv.name) ?? srv;
const status = serverObservation(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) {
setServerField(srv.name, 'active_since', current.active_since ?? status.last_seen_at);
setServerField(srv.name, 'inactive_since', null);
} else {
setServerField(srv.name, 'active_since', null);
setServerField(srv.name, 'inactive_since', current.inactive_since ?? status.last_seen_at);
setServerField(srv.name, 'game_started_at', null);
setServerField(srv.name, 'game_ended_at', null);
}
if (nextPlayers > 0 && (activeSince === null || prevPlayers === 0 || prevPlayers === null)) {
activeSince = now;
inactiveSince = null;
setServerField(srv.name, 'active_since', activeSince);
setServerField(srv.name, 'inactive_since', null);
} else if (nextPlayers === 0 && (inactiveSince === null || (prevPlayers !== null && prevPlayers > 0))) {
inactiveSince = now;
activeSince = null;
setServerField(srv.name, 'inactive_since', inactiveSince);
setServerField(srv.name, 'active_since', null);
}
// Persist last-observed state. last_seen_at is the health anchor.
setServerField(srv.name, 'last_players', nextPlayers);
setServerField(srv.name, 'last_map', result.status.map);
setServerField(srv.name, 'last_gametype', result.status.gametype);
setServerField(srv.name, 'last_roster', JSON.stringify(result.status.roster));
setServerField(srv.name, 'last_seen_at', now);
setServerField(srv.name, 'last_checked_at', now);
setServerField(srv.name, 'last_error', null);
setServerField(srv.name, 'last_reply_snippet', result.rawReply);
entries[i] = {
server: srv.name,
players: nextPlayers,
map: result.status.map,
gametype: result.status.gametype,
roster: result.status.roster,
health: classifyHealth(now, null),
lastSeenAt: now,
lastError: null,
endpoint: result.endpoint,
reason: null,
replySnippet: result.rawReply,
activeSince,
inactiveSince,
gameStartedAt,
gameEndedAt
};
} else {
// RCON unreachable — leave last_seen_at, active_since, and
// inactive_since alone (health decays to offline naturally;
// occupancy timers survive transient outages). Record the
// structured failure.
const encoded = encodeError(result.reason, result.detail);
const snippet = result.rawReply ?? null;
setServerField(srv.name, 'last_checked_at', now);
setServerField(srv.name, 'last_error', encoded);
setServerField(srv.name, 'last_reply_snippet', snippet);
entries[i] = {
server: srv.name,
players: srv.last_players,
map: srv.last_map,
gametype: srv.last_gametype,
roster: srv.last_roster ? safeJson(srv.last_roster) : null,
health: classifyHealth(srv.last_seen_at, encoded),
lastSeenAt: srv.last_seen_at,
lastError: encoded,
endpoint: result.endpoint,
reason: result.reason,
replySnippet: snippet,
activeSince: srv.active_since,
inactiveSince: srv.inactive_since,
gameStartedAt: srv.game_started_at,
gameEndedAt: srv.game_ended_at
};
}
const final = serverObservation(getServer(srv.name) ?? current);
entries[i] = { server: srv.name, players: final.last_players, map: final.last_map ?? srv.map,
gametype: final.last_gametype, roster: final.roster, health: final.health, lastSeenAt: final.last_seen_at,
lastError: final.last_error, replySnippet: final.replySnippet,
endpoint: current.game_port ? { host: current.rcon_host ?? config.rcon.host, port: current.game_port } : null, reason: final.last_error ? final.last_error.split(':')[0] as RconFailure : null,
activeSince: final.active_since, inactiveSince: final.inactive_since,
gameStartedAt: final.game_started_at, gameEndedAt: final.game_ended_at,
containerState: final.container_state, containerCheckedAt: final.container_checked_at,
telemetryAfter: final.telemetry_after, rconStatus: final.rcon_status, telemetry: final.telemetry };
}
}
await Promise.all(Array.from({ length: STATUS_CONCURRENCY }, pump));
await Promise.all(Array.from({ length: Math.min(STATUS_CONCURRENCY, servers.length) }, pump));
try {
recordMetrics(servers.map((srv, i) => ({
server: srv.name,
game: srv.game ?? 'unknown',
recordMetrics(servers.map((srv, i) => ({ server: srv.name, game: srv.game ?? 'unknown',
map: selectedZombieMap(srv.game, entries[i].map ?? srv.map, srv.map_slug)?.friendly ?? entries[i].map ?? srv.map ?? 'Unknown map',
players: capacities.has(srv.name) ? entries[i].players : null,
capacity: capacities.get(srv.name) ?? null
})));
players: entries[i].health === 'online' ? entries[i].players : null,
capacity: capacities.get(srv.name) ?? null })));
} catch (err) { console.error('[worker] metrics error:', err); }
emit({ type: 'server-status', entries });
}
function safeJson<T>(raw: string): T | null {
try {
return JSON.parse(raw) as T;
} catch {
return null;
}
}
export function startWorker(): void {
if (started) return;
started = true;