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

Roll deployment restarts through a shared fleet-wide recovery limit

Browse Source
This commit is contained in:
josh
2026-09-17 11:25:06 -04:00
parent 3f0d6dd87d
commit 8e04a190b3
12 changed files with 579 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/pipeline/beta-runner.ts
+30 -18
View File
@@ -2,9 +2,9 @@ import { db } from '../db/index.js';
import { emit } from './bus.js';
import type { BetaStage } from './bus.js';
import { komodoConfigured, rconConfigured } from '../config.js';
import { rconGetPlayerCount, rconStatusResult } from '../integrations/rcon.js';
import { komodoRestart, komodoInspectHealthy } from '../integrations/komodo.js';
import { rconGetPlayerCount } from '../integrations/rcon.js';
import { scanForMatchEnd } from '../fleet/match-log.js';
import { refreshRestartRollout, restartInRollout, rolloutHealth, rolloutWaitReason } from './restart-rollout.js';
interface BetaServerActiveRow {
beta_id: number;
@@ -84,8 +84,17 @@ function reconcileBetaState(betaId: number): void {
}
}
export async function betaTick(): Promise<{ processed: number }> {
let betaInFlight: Promise<{ processed: number }> | null = null;
export function betaTick(): Promise<{ processed: number }> {
if (betaInFlight) return betaInFlight;
betaInFlight = runBetaTick().finally(() => { betaInFlight = null; });
return betaInFlight;
}
async function runBetaTick(): Promise<{ processed: number }> {
if (!integrationsOn()) return { processed: 0 };
await refreshRestartRollout();
const rows = db()
.prepare(
@@ -95,7 +104,8 @@ export async function betaTick(): Promise<{ processed: number }> {
FROM beta_servers bs
JOIN betas b ON b.id = bs.beta_id
WHERE b.state = 'running'
AND bs.stage IN ('awaiting-empty','restarting','awaiting-health','awaiting-match')`
AND bs.stage IN ('awaiting-empty','restarting','awaiting-health','awaiting-match')
ORDER BY b.id, bs.server`
)
.all() as BetaServerActiveRow[];
@@ -117,19 +127,24 @@ export async function betaTick(): Promise<{ processed: number }> {
async function advance(row: BetaServerActiveRow): Promise<void> {
switch (row.stage) {
case 'awaiting-empty': {
case 'awaiting-empty':
case 'restarting': {
const waitReason = rolloutWaitReason(row.server);
if (waitReason) {
transition(row.beta_id, row.server, 'awaiting-empty', waitReason);
return;
}
const players = await rconGetPlayerCount(row.server);
if (players === null) return; // hold — RCON unreachable, try next tick
if (players > 0) return; // hold — still populated
transition(row.beta_id, row.server, 'restarting');
// Fall through to restart on the same tick for lower latency.
await advance({ ...row, stage: 'restarting' });
return;
}
case 'restarting': {
const r = await komodoRestart(row.server);
if (!r.ok) {
transition(row.beta_id, row.server, 'failed', `komodo: ${r.message}`);
const r = await restartInRollout(row.server);
if (r.kind === 'queued') {
transition(row.beta_id, row.server, 'awaiting-empty', r.message);
return;
}
if (r.kind === 'failed') {
transition(row.beta_id, row.server, 'failed', r.message);
return;
}
const next = row.wait_for_health ? 'awaiting-health' : row.wait_for_match ? 'awaiting-match' : 'ready';
@@ -137,14 +152,13 @@ async function advance(row: BetaServerActiveRow): Promise<void> {
return;
}
case 'awaiting-health': {
const h = await komodoInspectHealthy(row.server);
const h = rolloutHealth(row.server);
if (!h.ok) {
// Record the failure reason (including any snippet) so the
// beta detail UI can surface why we're stuck.
const snippet = h.snippet ? ` snippet=${h.snippet.slice(0, 400)}` : '';
setStage().run(
'awaiting-health',
`unhealthy: ${h.reason}${snippet}`,
h.reason,
nowIso(),
'awaiting-health',
'awaiting-health',
@@ -159,8 +173,6 @@ async function advance(row: BetaServerActiveRow): Promise<void> {
);
return;
}
const rcon = await rconStatusResult(row.server);
if (!rcon.ok) return; // container reports up but game not answering yet
const next = row.wait_for_match ? 'awaiting-match' : 'ready';
transition(row.beta_id, row.server, next);
return;
src/lib/server/pipeline/restart-rollout.ts
+125
View File
@@ -0,0 +1,125 @@
import { config } from '../config.js';
import { db } from '../db/index.js';
import { komodoInspectHealthy, komodoRestart } from '../integrations/komodo.js';
import { rconStatusResult } from '../integrations/rcon.js';
// Komodo queues restart work. Don't mistake the still-running old container
// for a completed restart when an API kick immediately follows the request.
const HEALTH_POLL_MS = 15_000;
const requestsInFlight = new Set<string>();
interface RestartSlot {
server: string;
requested_at: string;
checked_at: string | null;
healthy_at: string | null;
players: number | null;
error: string | null;
}
function slotFor(server: string): RestartSlot | undefined {
return db().prepare('SELECT * FROM restart_rollout WHERE server = ?').get(server) as RestartSlot | undefined;
}
export function rolloutWaitReason(server: string): string | null {
const slot = slotFor(server);
if (slot?.healthy_at === null) {
return `Waiting for the previous restart of this server to recover${slot.error ? `: ${slot.error}` : ''}`;
}
const { n } = db().prepare('SELECT COUNT(*) AS n FROM restart_rollout WHERE healthy_at IS NULL').get() as { n: number };
return n >= config.restartConcurrency ? 'Queued for a rolling restart; waiting for another server to recover' : null;
}
export type RolloutRestartResult =
| { kind: 'started' }
| { kind: 'queued'; message: string }
| { kind: 'failed'; message: string };
/** Reserve capacity before any async I/O; callers still check for players first. */
export async function restartInRollout(server: string): Promise<RolloutRestartResult> {
const waitReason = db().transaction(() => {
const reason = rolloutWaitReason(server);
if (reason) return reason;
db().prepare(`INSERT INTO restart_rollout (server, requested_at) VALUES (?, ?)
ON CONFLICT(server) DO UPDATE SET requested_at = excluded.requested_at,
checked_at = NULL, healthy_at = NULL, players = NULL, error = NULL`)
.run(server, new Date().toISOString());
return null;
}).immediate();
if (waitReason) return { kind: 'queued', message: waitReason };
requestsInFlight.add(server);
try {
const result = await komodoRestart(server);
if (result.ok) return { kind: 'started' };
// These failures cannot have queued a remote restart. Other failures
// (especially a timeout) may have: retain capacity until recovery.
if (result.kind === 'no-name' || result.kind === 'unconfigured') {
db().prepare('DELETE FROM restart_rollout WHERE server = ?').run(server);
} else {
db().prepare('UPDATE restart_rollout SET error = ? WHERE server = ?').run(`komodo: ${result.message}`, server);
}
return { kind: 'failed', message: `komodo: ${result.message}` };
} catch (err) {
const message = `komodo: ${(err as Error).message}`;
db().prepare('UPDATE restart_rollout SET error = ? WHERE server = ?').run(message, server);
return { kind: 'failed', message };
} finally {
// Start the grace period at completion, including slow/failed requests.
db().prepare('UPDATE restart_rollout SET requested_at = ? WHERE server = ?')
.run(new Date().toISOString(), server);
requestsInFlight.delete(server);
}
}
let refreshInFlight: Promise<void> | null = null;
/** Poll all occupied slots, even if their originating beta was already closed. */
export function refreshRestartRollout(): Promise<void> {
if (refreshInFlight) return refreshInFlight;
refreshInFlight = refresh().finally(() => { refreshInFlight = null; });
return refreshInFlight;
}
async function refresh(): Promise<void> {
// Adopt work started before this feature was installed. INSERT OR IGNORE
// preserves completed slots and never resets a newer restart's progress.
db().prepare(`INSERT OR IGNORE INTO restart_rollout (server, requested_at)
SELECT server, MAX(restarted_at) FROM deployment_servers
WHERE restart_state IN ('pending','failed') AND restarted_at IS NOT NULL GROUP BY server`).run();
db().prepare(`INSERT OR IGNORE INTO restart_rollout (server, requested_at)
SELECT bs.server, MAX(COALESCE(bs.restarted_at, bs.updated_at))
FROM beta_servers bs JOIN betas b ON b.id = bs.beta_id
WHERE b.state = 'running' AND bs.stage IN ('restarting','awaiting-health') GROUP BY bs.server`).run();
const slots = db().prepare('SELECT * FROM restart_rollout WHERE healthy_at IS NULL ORDER BY requested_at, server')
.all() as RestartSlot[];
for (const slot of slots) {
if (requestsInFlight.has(slot.server)) continue;
if (Date.now() - Date.parse(slot.checked_at ?? slot.requested_at) < HEALTH_POLL_MS) continue;
let error: string | null = null;
let players: number | null = null;
try {
const health = await komodoInspectHealthy(slot.server);
if (!health.ok) {
error = `unhealthy: ${health.reason}${health.snippet ? ` snippet=${health.snippet.slice(0, 400)}` : ''}`;
} else {
const rcon = await rconStatusResult(slot.server);
if (!rcon.ok) error = `rcon not answering: ${rcon.reason}`;
else players = rcon.status.players;
}
} catch (err) {
error = `health check: ${(err as Error).message}`;
}
const now = new Date().toISOString();
db().prepare(`UPDATE restart_rollout SET checked_at = ?, healthy_at = ?, players = ?, error = ?
WHERE server = ? AND requested_at = ? AND healthy_at IS NULL`)
.run(now, error ? null : now, players, error, slot.server, slot.requested_at);
}
}
export function rolloutHealth(server: string): { ok: true; players: number } | { ok: false; reason: string } {
const slot = slotFor(server);
if (slot?.healthy_at) return { ok: true, players: slot.players ?? 0 };
return { ok: false, reason: slot?.error ?? 'Waiting for restart recovery and health checks' };
}
src/lib/server/pipeline/sweep.ts
+37 -25
View File
@@ -1,17 +1,21 @@
import { config, komodoConfigured, rconConfigured } from '../config.js';
import { db } from '../db/index.js';
import { emit } from './bus.js';
import { rconGetPlayerCount, rconStatusResult } from '../integrations/rcon.js';
import { komodoRestart, komodoInspectHealthy } from '../integrations/komodo.js';
import { rconGetPlayerCount } from '../integrations/rcon.js';
import { setServerField } from '../fleet/servers.js';
import { refreshRestartRollout, restartInRollout, rolloutHealth, rolloutWaitReason } from './restart-rollout.js';
interface PendingRow {
deployment_id: string;
server: string;
attempts: number;
restarted_at: string | null;
files_changed: number;
}
type SweepResult = { checked: number; restarted: number; healthy: number; held: number; failed: number };
let sweepInFlight: Promise<SweepResult> | null = null;
function nowIso(): string {
return new Date().toISOString();
}
@@ -35,7 +39,13 @@ function integrationsOn(): boolean {
* Same-server dedup is per-tick per-phase so multiple concurrent deployments
* targeting the same server only cost one RCON+Komodo round trip per tick.
*/
export async function sweep(): Promise<{ checked: number; restarted: number; healthy: number; held: number; failed: number }> {
export function sweep(): Promise<SweepResult> {
if (sweepInFlight) return sweepInFlight;
sweepInFlight = runSweep().finally(() => { sweepInFlight = null; });
return sweepInFlight;
}
async function runSweep(): Promise<SweepResult> {
if (!integrationsOn()) {
// Freeze pending rows as manual — they won't drain without integrations.
db()
@@ -49,11 +59,15 @@ export async function sweep(): Promise<{ checked: number; restarted: number; hea
return { checked: 0, restarted: 0, healthy: 0, held: 0, failed: 0 };
}
// Recovery frees capacity before admitting any more servers. This runs
// even without pending deployment rows (betas share the same slots).
await refreshRestartRollout();
const pending = db()
.prepare(
`SELECT deployment_id, server, attempts, restarted_at
FROM deployment_servers
WHERE restart_state IN ('pending','failed')`
`SELECT ds.deployment_id, ds.server, ds.attempts, ds.restarted_at, ds.files_changed
FROM deployment_servers ds JOIN deployments d ON d.id = ds.deployment_id
WHERE ds.restart_state IN ('pending','failed')
ORDER BY d.started_at, d.id, ds.server`
)
.all() as PendingRow[];
if (pending.length === 0) {
@@ -65,6 +79,7 @@ export async function sweep(): Promise<{ checked: number; restarted: number; hea
// Split dedup caches by phase — the two phases talk to different endpoints.
const preRestart = new Map<string, Outcome>();
const postRestart = new Map<string, Outcome>();
const recoveringServers = new Set(pending.filter((row) => row.restarted_at).map((row) => row.server));
let checked = 0;
let restarted = 0;
@@ -80,13 +95,18 @@ export async function sweep(): Promise<{ checked: number; restarted: number; hea
);
for (const row of pending) {
// A failed file write isn't a restart request. Preserve the original
// failure until the operator repairs/retries that deployment.
if (row.files_changed === 0) continue;
const cache = row.restarted_at ? postRestart : preRestart;
let outcome = cache.get(row.server);
if (!outcome) {
checked++;
outcome = row.restarted_at
? await runHealthPhase(row.server)
: await runRestartPhase(row.server);
? runHealthPhase(row.server)
: recoveringServers.has(row.server)
? { state: 'pending', players: null, error: 'Queued behind an earlier restart of this server', restartedNow: false }
: await runRestartPhase(row.server);
cache.set(row.server, outcome);
if (outcome.state === 'done' && row.restarted_at) healthy++;
if (outcome.restartedNow) restarted++;
@@ -152,6 +172,8 @@ export async function sweep(): Promise<{ checked: number; restarted: number; hea
// Phase 1: wait for the server to be empty, then Komodo-restart it.
async function runRestartPhase(server: string): Promise<{ state: 'done' | 'failed' | 'pending'; players: number | null; error: string | null; restartedNow: boolean }> {
const waitReason = rolloutWaitReason(server);
if (waitReason) return { state: 'pending', players: null, error: waitReason, restartedNow: false };
const players = await rconGetPlayerCount(server);
if (players === null) {
return { state: 'pending', players: null, error: 'rcon unreachable', restartedNow: false };
@@ -159,32 +181,22 @@ async function runRestartPhase(server: string): Promise<{ state: 'done' | 'faile
if (players > 0) {
return { state: 'pending', players, error: null, restartedNow: false };
}
const r = await komodoRestart(server);
if (r.ok) {
const r = await restartInRollout(server);
if (r.kind === 'started') {
// Restart triggered — flip restarted_at so the next tick runs the
// health phase. Still 'pending' overall until health confirms.
return { state: 'pending', players: 0, error: null, restartedNow: true };
}
if (r.kind === 'no-name') {
return { state: 'failed', players: 0, error: r.message, restartedNow: false };
}
return { state: 'failed', players: 0, error: `komodo: ${r.message}`, restartedNow: false };
return { state: r.kind === 'queued' ? 'pending' : 'failed', players: 0, error: r.message, restartedNow: false };
}
// Phase 2: after Komodo-restart, wait until the container reports running AND
// RCON is answering. Only then does the deployment_server row transition to
// 'done'. If either check fails we stay pending and try again next tick.
async function runHealthPhase(server: string): Promise<{ state: 'done' | 'failed' | 'pending'; players: number | null; error: string | null; restartedNow: boolean }> {
const h = await komodoInspectHealthy(server);
function runHealthPhase(server: string): { state: 'done' | 'pending'; players: number | null; error: string | null; restartedNow: boolean } {
const h = rolloutHealth(server);
if (!h.ok) {
// Include a truncated response snippet (when the extractor is the
// failure) so the operator can spot which Komodo field to teach.
const snippet = h.snippet ? ` snippet=${h.snippet.slice(0, 400)}` : '';
return { state: 'pending', players: null, error: `unhealthy: ${h.reason}${snippet}`, restartedNow: false };
return { state: 'pending', players: null, error: h.reason, restartedNow: false };
}
const rcon = await rconStatusResult(server);
if (!rcon.ok) {
return { state: 'pending', players: null, error: `rcon not answering: ${rcon.reason}`, restartedNow: false };
}
return { state: 'done', players: rcon.status.players, error: null, restartedNow: false };
return { state: 'done', players: h.players, error: null, restartedNow: false };
}
src/lib/server/pipeline/worker.ts
+9 -1
View File
@@ -18,7 +18,15 @@ let started = false;
let tickTimer: NodeJS.Timeout | null = null;
let statusTimer: NodeJS.Timeout | null = null;
async function tick(): Promise<void> {
let tickInFlight: Promise<void> | null = null;
function tick(): Promise<void> {
if (tickInFlight) return tickInFlight;
tickInFlight = runTick().finally(() => { tickInFlight = null; });
return tickInFlight;
}
async function runTick(): Promise<void> {
try {
await sweep();
} catch (err) {