259 lines
12 KiB
TypeScript
259 lines
12 KiB
TypeScript
import { processProvisioning } from '../provisioning/worker.js';
|
|
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, setServerField } from '../fleet/servers.js';
|
|
import { komodoInspectHealthy } from '../integrations/komodo.js';
|
|
import { restartSettling } from '../provisioning/index.js';
|
|
import { classifyHealth } from '../fleet/health.js';
|
|
import { isProvisioning } from '../fleet/provisioning.js';
|
|
import { STATUS_REFRESH_MS } from '../../server-freshness.js';
|
|
import { recordMetrics } from '../fleet/metrics.js';
|
|
import { selectedZombieMap } from '../../map-aliases.js';
|
|
import { maintenanceTick } from '../fleet/plutonium.js';
|
|
import { watchdogTick } from '../watchdog/index.js';
|
|
|
|
const TICK_MS = 15_000;
|
|
// Cap concurrent UDP RCON calls so a 15-server fleet doesn't blast the
|
|
// host all at once.
|
|
const STATUS_CONCURRENCY = 4;
|
|
|
|
let started = false;
|
|
let tickTimer: NodeJS.Timeout | null = null;
|
|
let statusTimer: NodeJS.Timeout | null = null;
|
|
|
|
let tickInFlight: Promise<void> | null = null;
|
|
|
|
function tick(): Promise<void> {
|
|
if (tickInFlight) return tickInFlight;
|
|
tickInFlight = runTick().finally(() => { tickInFlight = null; });
|
|
return tickInFlight;
|
|
}
|
|
|
|
async function reconcileRequesters(): Promise<void> {
|
|
try { await sweep(); } catch (err) { console.error('[worker] campaign error:', err); }
|
|
try { await betaTick(); } catch (err) { console.error('[worker] beta error:', err); }
|
|
try { await maintenanceTick(); } catch (err) { console.error('[worker] maintenance error:', err); }
|
|
try { await watchdogTick(); } catch (err) { console.error('[worker] watchdog error:', err); }
|
|
}
|
|
|
|
async function runTick(): Promise<void> {
|
|
// Callers submit durable requests first. Exactly one owner admits and runs
|
|
// lifecycle work, then callers observe the resulting jobs by ID.
|
|
await reconcileRequesters();
|
|
try { await processProvisioning(); }
|
|
catch (err) { console.error('[worker] provisioning error:', err); }
|
|
await reconcileRequesters();
|
|
}
|
|
|
|
/**
|
|
* Encode a structured RCON failure as `"reason:detail"` so `last_error`
|
|
* stays a single column but the UI can split it apart when rendering the
|
|
* diagnostic panel. Reasons are stable; details are human-facing.
|
|
*/
|
|
function encodeError(reason: RconFailure, detail: string): string {
|
|
return `${reason}:${detail}`;
|
|
}
|
|
|
|
let statusInFlight: Promise<void> | null = null;
|
|
|
|
export function fanoutStatus(): Promise<void> {
|
|
if (statusInFlight) return statusInFlight;
|
|
statusInFlight = refreshStatus().finally(() => { statusInFlight = null; });
|
|
return statusInFlight;
|
|
}
|
|
|
|
async function refreshStatus(): Promise<void> {
|
|
const servers = listServers();
|
|
if (servers.length === 0) 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) {
|
|
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)) {
|
|
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) {
|
|
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
|
|
};
|
|
}
|
|
}
|
|
}
|
|
await Promise.all(Array.from({ length: STATUS_CONCURRENCY }, 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: capacities.has(srv.name) ? 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;
|
|
|
|
void tick();
|
|
void fanoutStatus();
|
|
tickTimer = setInterval(tick, TICK_MS);
|
|
statusTimer = setInterval(fanoutStatus, STATUS_REFRESH_MS);
|
|
|
|
console.log('[worker] started — sweep every', TICK_MS, 'ms, status every', STATUS_REFRESH_MS, 'ms');
|
|
}
|
|
|
|
export function stopWorker(): void {
|
|
if (tickTimer) clearInterval(tickTimer);
|
|
if (statusTimer) clearInterval(statusTimer);
|
|
tickTimer = null;
|
|
statusTimer = null;
|
|
started = false;
|
|
}
|
|
|
|
/**
|
|
* External trigger — called from `/api/push` and beta endpoints so a sweep
|
|
* fires immediately after a state change instead of waiting for the next tick.
|
|
*/
|
|
export async function kick(): Promise<void> {
|
|
await tick();
|
|
}
|