487 lines
19 KiB
TypeScript
487 lines
19 KiB
TypeScript
import { drainRetention } from '../retention.js';
|
|
import { collectFleetChat } from '../fleet/chat.js';
|
|
import { runAutoscaling, pruneAutoscaling } from '../autoscaling/index.js';
|
|
import { runRebalancing, pruneRebalancing } from '../rebalancing/index.js';
|
|
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 type { RconFailure } from '../integrations/rcon.js';
|
|
import { setServerFields, type ServerRow } from '../fleet/servers.js';
|
|
import {
|
|
inspectForMonitoring,
|
|
rconForMonitoring,
|
|
prepareMonitoringRcon,
|
|
monitoringStatus,
|
|
monitoringStatusWriter,
|
|
monitoringIdentity,
|
|
MonitoringChangedError,
|
|
MonitoringSkippedError
|
|
} from '../fleet/health-observations.js';
|
|
import { serverObservation } from '../fleet/status.js';
|
|
import { successfulStatusFields, activityFields } from '../fleet/status-updates.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';
|
|
import { cleanupControllerLibraries } from '../game-files/index.js';
|
|
import { trackActivity } from './activity.js';
|
|
import { MonitoringSnapshot } from '../fleet/monitoring-snapshot.js';
|
|
import { observedPlayerCapacity } from '../fleet/player-capacity.js';
|
|
import type { PlayerCapacityObservation } from '../../server-capacity.js';
|
|
import { withHostRequestContext } from '../hosts/request-context.js';
|
|
import { withOperationSource } from '../hosts/operation-context.js';
|
|
import { failureReplySnippet } from '../integrations/rcon-diagnostics.js';
|
|
|
|
const TICK_MS = 15_000;
|
|
const RETENTION_INTERVAL_MS = 5 * 60_000;
|
|
// Match each agent's two health lanes without making unrelated hosts wait.
|
|
const STATUS_CONCURRENCY_PER_HOST = 2;
|
|
|
|
let started = false;
|
|
let tickTimer: NodeJS.Timeout | null = null;
|
|
let statusTimer: NodeJS.Timeout | null = null;
|
|
|
|
let tickInFlight: Promise<void> | null = null;
|
|
let tickRequested = false;
|
|
const requesterFlights = new Map<string, Promise<void>>();
|
|
let nextRetentionAt = 0;
|
|
|
|
function tick(requestAnother = false): Promise<void> {
|
|
if (requestAnother) tickRequested = true;
|
|
if (tickInFlight) return tickInFlight;
|
|
tickInFlight = (async () => {
|
|
do {
|
|
tickRequested = false;
|
|
try {
|
|
await trackActivity('provisioning', 'Processing provisioning requests', () =>
|
|
processProvisioning({ wait: false })
|
|
);
|
|
} catch (err) {
|
|
console.error('[worker] provisioning error:', err);
|
|
}
|
|
} while (tickRequested);
|
|
})().finally(() => {
|
|
tickInFlight = null;
|
|
});
|
|
return tickInFlight;
|
|
}
|
|
|
|
function reconcileRequesters(): void {
|
|
// Observers submit durable intent independently. Provisioning still owns
|
|
// admission and rechecks source eligibility immediately before mutation.
|
|
const requesters: [string, string, () => Promise<unknown>][] = [
|
|
['campaign', 'Reconciling campaigns', sweep],
|
|
['beta', 'Reconciling beta tests', betaTick],
|
|
['maintenance', 'Checking maintenance and recovery', runMaintenance],
|
|
['autoscaling', 'Evaluating lobby autoscaling', runAutoscaling],
|
|
['rebalancing', 'Reconciling rebalance moves', runRebalancing],
|
|
['game-files', 'Removing unused game files', cleanupControllerLibraries],
|
|
['chat', 'Saving match chat', collectFleetChat],
|
|
[
|
|
'retention',
|
|
'Cleaning expired history',
|
|
async () => {
|
|
await drainRetention();
|
|
pruneAutoscaling();
|
|
pruneRebalancing();
|
|
}
|
|
]
|
|
];
|
|
for (const [key, label, work] of requesters) {
|
|
if (requesterFlights.has(key)) continue;
|
|
if (key === 'retention') {
|
|
if (Date.now() < nextRetentionAt) continue;
|
|
nextRetentionAt = Date.now() + RETENTION_INTERVAL_MS;
|
|
}
|
|
const flight = trackActivity(key, label, work)
|
|
.catch((err) => {
|
|
console.error(`[worker] ${key} error:`, err);
|
|
})
|
|
.then(async (result) => {
|
|
if (
|
|
key !== 'game-files' &&
|
|
key !== 'retention' &&
|
|
key !== 'chat' &&
|
|
(!['autoscaling', 'rebalancing'].includes(key) || result === true)
|
|
)
|
|
await tick(true);
|
|
})
|
|
.finally(() => {
|
|
requesterFlights.delete(key);
|
|
});
|
|
requesterFlights.set(key, flight);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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}`;
|
|
}
|
|
|
|
// 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>>();
|
|
interface StatusQueue {
|
|
active: number;
|
|
queued: Array<() => Promise<void>>;
|
|
}
|
|
const statusQueues = new Map<string, StatusQueue>();
|
|
const changedStatus = new Map<string, ServerStatusEntry>();
|
|
function publishStatus(entry: ServerStatusEntry) {
|
|
const scheduled = changedStatus.size > 0;
|
|
changedStatus.set(entry.server, entry);
|
|
if (!scheduled)
|
|
queueMicrotask(() => {
|
|
const entries = [...changedStatus.values()];
|
|
changedStatus.clear();
|
|
emit({ type: 'server-status', delta: true, entries });
|
|
});
|
|
}
|
|
|
|
function drainStatusQueue(host: string, queue: StatusQueue): void {
|
|
while (queue.active < STATUS_CONCURRENCY_PER_HOST && queue.queued.length) {
|
|
const next = queue.queued.shift()!;
|
|
queue.active++;
|
|
void next().finally(() => {
|
|
queue.active--;
|
|
drainStatusQueue(host, queue);
|
|
});
|
|
}
|
|
if (!queue.active && !queue.queued.length) statusQueues.delete(host);
|
|
}
|
|
|
|
function queueStatus(
|
|
srv: ServerRow,
|
|
snapshot: MonitoringSnapshot,
|
|
probe?: () => ReturnType<typeof rconForMonitoring>
|
|
): Promise<number | null> {
|
|
const server = srv.name;
|
|
const pending = pendingStatus.get(server);
|
|
if (pending) return pending;
|
|
// Host queues keep slow instances from consuming other hosts' slots.
|
|
const host = srv.host_id ? `native:${srv.host_id}` : `unregistered:${srv.name}`;
|
|
let queue = statusQueues.get(host);
|
|
if (!queue) {
|
|
queue = { active: 0, queued: [] };
|
|
statusQueues.set(host, queue);
|
|
}
|
|
let finish!: (capacity: number | null) => void;
|
|
const work = new Promise<number | null>((resolve) => {
|
|
finish = resolve;
|
|
});
|
|
pendingStatus.set(server, work);
|
|
queue.queued.push(async () => {
|
|
let capacity: number | null = null;
|
|
try {
|
|
capacity = await withOperationSource('Status monitor', () =>
|
|
refreshServerStatus(server, snapshot, probe)
|
|
);
|
|
// Publish this server's latest evidence without waiting for the fleet.
|
|
const current = snapshot.get(server)?.server;
|
|
if (current)
|
|
publishStatus(
|
|
statusEntry(current, snapshot.observe, observedPlayerCapacity(server, snapshot))
|
|
);
|
|
} catch (err) {
|
|
console.error(`[worker] status error for ${server}:`, err);
|
|
} finally {
|
|
pendingStatus.delete(server);
|
|
finish(capacity);
|
|
}
|
|
});
|
|
drainStatusQueue(host, queue);
|
|
return work;
|
|
}
|
|
|
|
export function fanoutStatus(): Promise<void> {
|
|
return withHostRequestContext(statusPass);
|
|
}
|
|
|
|
async function statusPass(): Promise<void> {
|
|
const snapshot = new MonitoringSnapshot();
|
|
const servers = snapshot.servers();
|
|
if (!servers.length) return;
|
|
const probes = prepareMonitoringRcon(
|
|
servers
|
|
.filter(
|
|
(server) =>
|
|
!pendingStatus.has(server.name) &&
|
|
!snapshot.provisioning(server.name) &&
|
|
!snapshot.settling(server.name) &&
|
|
!snapshot.canSkipRcon(server.name)
|
|
)
|
|
.map((server) => server.name),
|
|
snapshot
|
|
);
|
|
const results = await Promise.all(
|
|
servers.map((server) => queueStatus(server, snapshot, probes.get(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.
|
|
snapshot.refresh();
|
|
recordMetrics(
|
|
snapshot.servers().map((srv) => {
|
|
const current = snapshot.observe(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,
|
|
snapshot: MonitoringSnapshot,
|
|
probeStatus?: () => ReturnType<typeof rconForMonitoring>
|
|
): Promise<number | null> {
|
|
snapshot.refresh(server);
|
|
const srv = snapshot.get(server)?.server;
|
|
if (!srv) return null;
|
|
let capacity: number | null = null;
|
|
let updates: Partial<Record<keyof ServerRow, string | number | null>> = {};
|
|
const flush = () => {
|
|
setServerFields(srv.name, updates);
|
|
snapshot.patch(srv.name, updates as Partial<ServerRow>);
|
|
updates = {};
|
|
};
|
|
if (!snapshot.provisioning(srv.name)) {
|
|
const previousGame = monitoringStatus(srv.name, snapshot)?.game;
|
|
const publishObservation = monitoringStatusWriter(srv.name, snapshot);
|
|
let gameObservation: Awaited<ReturnType<typeof rconForMonitoring>> | undefined;
|
|
const settling = snapshot.settling(srv.name);
|
|
const epoch = snapshot.get(srv.name)?.submission;
|
|
let identity: string | undefined;
|
|
try {
|
|
identity = monitoringIdentity(srv.name, snapshot);
|
|
} catch {
|
|
/* Offline hosts still need an unavailable sample. */
|
|
}
|
|
const unchanged = () => {
|
|
const current = snapshot.get(srv.name)?.server;
|
|
if (identity)
|
|
try {
|
|
if (identity !== monitoringIdentity(srv.name, snapshot)) return false;
|
|
} catch {
|
|
return false;
|
|
}
|
|
return (
|
|
epoch === snapshot.get(srv.name)?.submission &&
|
|
!snapshot.provisioning(srv.name) &&
|
|
current?.created_at === srv.created_at &&
|
|
current.desired_state === srv.desired_state &&
|
|
current.native_instance_id === srv.native_instance_id &&
|
|
current.instance_generation === srv.instance_generation
|
|
);
|
|
};
|
|
const inspectionStartedAt = new Date().toISOString();
|
|
const inspected = settling
|
|
? {
|
|
result: { ok: false as const, state: 'restarting', reason: 'Restart requested' },
|
|
observedAt: inspectionStartedAt
|
|
}
|
|
: await inspectForMonitoring(srv.name, false, snapshot).catch((err) =>
|
|
err instanceof MonitoringChangedError
|
|
? null
|
|
: {
|
|
result: {
|
|
ok: false as const,
|
|
state: 'unknown',
|
|
reason: (err as Error).message
|
|
},
|
|
observedAt: inspectionStartedAt
|
|
}
|
|
);
|
|
snapshot.refresh(srv.name);
|
|
if (!inspected) return null;
|
|
const container = inspected.result;
|
|
// A lifecycle request may be submitted while an inspection is in flight.
|
|
if (
|
|
unchanged() &&
|
|
(snapshot.get(srv.name)?.server?.container_checked_at ?? '') <= inspected.observedAt
|
|
) {
|
|
updates.container_state = container.ok
|
|
? 'running'
|
|
: (container.state?.toLowerCase() ?? 'unknown');
|
|
updates.container_checked_at = inspected.observedAt;
|
|
// An unavailable inspection is not evidence that the game stopped.
|
|
// Probe it just as a manual RCON check would; explicit lifecycle or
|
|
// unhealthy states still take precedence over game replies.
|
|
const unavailable = updates.container_state === 'unknown';
|
|
if (!container.ok && !unavailable) {
|
|
updates.last_error = null;
|
|
updates.rcon_failures = 0;
|
|
updates.last_reply_snippet = null;
|
|
} else if (unavailable || !snapshot.canSkipRcon(srv.name)) {
|
|
// Occupied telemetry rosters still need RCON for player pings.
|
|
flush();
|
|
const probeStartedAt = new Date().toISOString();
|
|
const probe = await (
|
|
probeStatus ? probeStatus() : rconForMonitoring(srv.name, false, snapshot)
|
|
).catch((err) => {
|
|
snapshot.refresh(srv.name);
|
|
if (err instanceof MonitoringSkippedError) {
|
|
// The container was checked, but game evidence retains its original age.
|
|
if (unchanged()) publishObservation(inspected, previousGame);
|
|
return null;
|
|
}
|
|
if (err instanceof MonitoringChangedError) return null;
|
|
return {
|
|
result: {
|
|
ok: false as const,
|
|
endpoint: null,
|
|
reason: 'send-error' as RconFailure,
|
|
detail: (err as Error).message,
|
|
rawReply: undefined
|
|
},
|
|
observedAt: probeStartedAt
|
|
};
|
|
});
|
|
snapshot.refresh(srv.name);
|
|
if (!probe) return null;
|
|
const result = probe.result;
|
|
const previous = snapshot.get(srv.name)?.server;
|
|
if (
|
|
unchanged() &&
|
|
previous &&
|
|
(previous.last_checked_at ?? '') <= probe.observedAt
|
|
) {
|
|
gameObservation = probe;
|
|
const now = probe.observedAt;
|
|
updates.last_checked_at = now;
|
|
updates.last_reply_snippet = failureReplySnippet(result);
|
|
if (result.ok) {
|
|
capacity = result.status.capacity ?? null;
|
|
Object.assign(updates, successfulStatusFields(result, now));
|
|
} else {
|
|
// Only distinct accepted checks in the current lifecycle count.
|
|
// Skipped/superseded checks return above without changing the streak.
|
|
const consecutive =
|
|
previous.last_error &&
|
|
(!epoch || (previous.last_checked_at ?? '') > epoch);
|
|
const failures = consecutive ? previous.rcon_failures || 1 : 0;
|
|
updates.rcon_failures =
|
|
consecutive && previous.last_checked_at === now
|
|
? failures
|
|
: Math.min(2, failures + 1);
|
|
updates.last_error = encodeError(result.reason, result.detail);
|
|
}
|
|
}
|
|
}
|
|
if (!settling && unchanged()) publishObservation(inspected, gameObservation);
|
|
}
|
|
} else {
|
|
updates.last_error = null;
|
|
updates.rcon_failures = 0;
|
|
updates.last_reply_snippet = null;
|
|
}
|
|
const current = { ...(snapshot.get(srv.name)?.server ?? srv), ...updates } as ServerRow;
|
|
const status = snapshot.observe(current);
|
|
Object.assign(updates, activityFields(current, status));
|
|
flush();
|
|
return capacity;
|
|
}
|
|
|
|
export function statusEntry(
|
|
srv: ServerRow,
|
|
observe = serverObservation,
|
|
capacity?: PlayerCapacityObservation
|
|
): ServerStatusEntry {
|
|
const final = observe(srv);
|
|
return {
|
|
server: srv.name,
|
|
players: final.last_players,
|
|
capacity: capacity?.capacity ?? null,
|
|
capacityObservedAt: capacity?.capacity_observed_at ?? null,
|
|
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,
|
|
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 {
|
|
if (started) return;
|
|
started = true;
|
|
|
|
void tick();
|
|
reconcileRequesters();
|
|
void fanoutStatus().catch((err) => {
|
|
console.error('[worker] status fanout error:', err);
|
|
});
|
|
tickTimer = setInterval(() => {
|
|
void tick();
|
|
reconcileRequesters();
|
|
}, TICK_MS);
|
|
statusTimer = setInterval(() => {
|
|
void fanoutStatus().catch((err) => {
|
|
console.error('[worker] status fanout error:', err);
|
|
});
|
|
}, 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> {
|
|
reconcileRequesters();
|
|
await tick(true);
|
|
}
|
|
|
|
/** Agent receipts need observation promptly, without launching another fleet scan. */
|
|
export async function kickProvisioning(): Promise<void> {
|
|
await tick(true);
|
|
}
|