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

Fix transient health checks and avoid unnecessary recovery

Browse Source
This commit is contained in:
josh
2026-10-02 13:06:18 -04:00
parent 221c4c10cf
commit 65d3a99846
16 changed files with 842 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/pipeline/worker.ts
+110 -76
View File
@@ -6,7 +6,7 @@ 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, setServerField } from '../fleet/servers.js';
import { getServer, listServers, setServerField, type ServerRow } from '../fleet/servers.js';
import { komodoInspectHealthy } from '../integrations/komodo.js';
import { restartSettling } from '../provisioning/index.js';
import { serverObservation } from '../fleet/status.js';
@@ -58,92 +58,126 @@ function encodeError(reason: RconFailure, detail: string): string {
return `${reason}:${detail}`;
}
let statusInFlight: Promise<void> | null = null;
// Deduplicate queued/running checks per server. A slow server must not prevent
// completed servers from being polled again on the next 30-second tick.
const pendingStatus = new Map<string, Promise<number | null>>();
const statusQueue: Array<() => Promise<void>> = [];
let activeStatus = 0;
export function fanoutStatus(): Promise<void> {
if (statusInFlight) return statusInFlight;
statusInFlight = refreshStatus().finally(() => { statusInFlight = null; });
return statusInFlight;
function drainStatusQueue(): void {
while (activeStatus < STATUS_CONCURRENCY && statusQueue.length) {
const next = statusQueue.shift()!;
activeStatus++;
void next().finally(() => { activeStatus--; drainStatusQueue(); });
}
}
async function refreshStatus(): Promise<void> {
function queueStatus(server: string): Promise<number | null> {
const pending = pendingStatus.get(server);
if (pending) return pending;
let finish!: (capacity: number | null) => void;
const work = new Promise<number | null>(resolve => { finish = resolve; });
pendingStatus.set(server, work);
statusQueue.push(async () => {
let capacity: number | null = null;
try {
capacity = await refreshServerStatus(server);
// Consumers expect a full snapshot. Re-read it now so neither game
// data nor container evidence waits for the slowest fleet member.
emit({ type: 'server-status', entries: listServers().map(statusEntry) });
} catch (err) { console.error(`[worker] status error for ${server}:`, err); }
finally { pendingStatus.delete(server); finish(capacity); }
});
drainStatusQueue();
return work;
}
export async function fanoutStatus(): Promise<void> {
const servers = listServers();
if (!servers.length) return;
const entries: ServerStatusEntry[] = new Array(servers.length);
const capacities = new Map<string, number | null>();
let cursor = 0;
async function pump() {
while (cursor < servers.length) {
const i = cursor++;
const srv = servers[i];
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 {
const results = await Promise.all(servers.map(server => queueStatus(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.
recordMetrics(listServers().map(srv => {
const current = serverObservation(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,
capacity: capacities.get(srv.name) ?? null };
}));
} catch (err) { console.error('[worker] metrics error:', err); }
}
async function refreshServerStatus(server: string): Promise<number | null> {
const srv = getServer(server);
if (!srv) return null;
let capacity: number | null = null;
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 }));
// A lifecycle request may be submitted while an inspection is in flight.
if (unchanged()) {
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);
}
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);
} 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 (unchanged()) {
const now = new Date().toISOString();
setServerField(srv.name, 'last_checked_at', now);
setServerField(srv.name, 'last_reply_snippet', result.rawReply ?? null);
if (result.ok) {
capacity = 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));
}
}
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 };
}
} else {
setServerField(srv.name, 'last_error', null);
setServerField(srv.name, 'last_reply_snippet', null);
}
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);
}
}
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',
map: selectedZombieMap(srv.game, entries[i].map ?? srv.map, srv.map_slug)?.friendly ?? entries[i].map ?? srv.map ?? 'Unknown map',
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 });
return capacity;
}
function statusEntry(srv: ServerRow): ServerStatusEntry {
const final = serverObservation(srv);
return { 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: srv.game_port ? { host: srv.rcon_host ?? config.rcon.host, port: srv.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 };
}
export function startWorker(): void {