import { config } from '../config.js'; 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 { 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'; import { selectedZombieMap } from '../../map-aliases.js'; import { runMaintenance } from '../maintenance/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 | null = null; function tick(): Promise { if (tickInFlight) return tickInFlight; tickInFlight = runTick().finally(() => { tickInFlight = null; }); return tickInFlight; } async function reconcileRequesters(): Promise { try { await sweep(); } catch (err) { console.error('[worker] campaign error:', err); } try { await betaTick(); } catch (err) { console.error('[worker] beta error:', err); } try { await runMaintenance(); } catch (err) { console.error('[worker] maintenance error:', err); } } async function runTick(): Promise { // 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 | null = null; export function fanoutStatus(): Promise { if (statusInFlight) return statusInFlight; statusInFlight = refreshStatus().finally(() => { statusInFlight = null; }); return statusInFlight; } async function refreshStatus(): Promise { const servers = listServers(); if (!servers.length) return; const entries: ServerStatusEntry[] = new Array(servers.length); const capacities = new Map(); 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 { 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); } } 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: 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 }); } 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 { await tick(); }