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

refactor: make provisioning own instance lifecycle operations

Browse Source
This commit is contained in:
josh
2026-09-27 14:45:00 -04:00
parent cfed584614
commit df265d1900
44 changed files with 1791 additions and 1609 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/pipeline/beta-plan-runner.ts
+28 -35
View File
@@ -1,12 +1,10 @@
import { db } from '../db/index.js';
import { komodoConfigured, rconConfigured } from '../config.js';
import { rconGetPlayerCount } from '../integrations/rcon.js';
import { betaOperationInFlight, deployBetaServers, getBeta, listBetaServers, rollbackBeta } from '../fleet/beta.js';
import { activity, currentRun, observeRun, prepareRun, readMetrics, saveMetrics, setOutcome, storedPlan, type StoredPlan } from '../fleet/beta-tests.js';
import { coveragePassed, evaluateRequirements } from '../fleet/beta-evaluation.js';
import { parseBetaPlan, type BetaPlan } from '../../beta-plan.js';
import { emit, type BetaStage } from './bus.js';
import { restartInRollout, rolloutHealth, rolloutWaitReason } from './restart-rollout.js';
import { requestRestart, releaseRestartRequest } from '../provisioning/index.js';
function stage(id: number, server: string, next: BetaStage, error: string | null = null): void {
db().prepare('UPDATE beta_servers SET stage = ?, error = ?, updated_at = ? WHERE beta_id = ? AND server = ?')
@@ -80,11 +78,6 @@ async function advancePlan(stored: StoredPlan): Promise<void> {
setOutcome(id, 'inconclusive', 'Deadline reached without enough qualifying evidence');
return;
}
if (!komodoConfigured() || !rconConfigured()) {
betaState(id, false);
setOutcome(id, 'active', 'Waiting for Komodo and RCON configuration');
return;
}
for (const server of servers) {
const freshPlan = storedPlan(id)!;
if (!['active', 'passed'].includes(freshPlan.outcome)) return;
@@ -102,35 +95,33 @@ async function advancePlan(stored: StoredPlan): Promise<void> {
const broken = evaluateRequirements(plan, readMetrics(currentRun(id, server.server)!), waitingRun.phase).find(p => p.failed);
if (broken) { await fail(id, plan, `${server.server}: ${broken.label}: ${broken.detail}`); return; }
}
if (rolloutWaitReason(server.server)) continue;
if (await rconGetPlayerCount(server.server) !== 0) continue;
const oldRun = currentRun(id, server.server);
const phase = oldRun ? oldRun.phase + 1 : 0;
// Fail closed after interrupted restart requests; never silently reuse earlier evidence.
stage(id, server.server, 'restarting');
const run = await prepareRun(id, server.server, phase);
if (oldRun) {
const previous = readMetrics(oldRun);
const metrics = readMetrics(run);
metrics.outages = previous.outages;
saveMetrics(run, metrics);
// Prepare evidence once before handing off the restart. Empty-server
// checks and remote execution belong entirely to Provisioning.
let run = currentRun(id, server.server);
if (!run || run.observing_at) {
const oldRun = run;
run = await prepareRun(id, server.server, oldRun ? oldRun.phase + 1 : 0);
if (oldRun) {
const metrics = readMetrics(run);
metrics.outages = readMetrics(oldRun).outages;
saveMetrics(run, metrics);
}
}
const restart = await restartInRollout(server.server);
if (restart.kind === 'queued') {
db().prepare('DELETE FROM beta_test_runs WHERE id = ?').run(run.id);
if (oldRun) db().prepare('UPDATE beta_test_runs SET retired_at = NULL WHERE id = ?').run(oldRun.id);
stage(id, server.server, 'awaiting-empty', restart.message); continue;
}
if (restart.kind === 'failed') { stage(id, server.server, 'failed', restart.message); await fail(id, plan, restart.message); return; }
db().prepare('UPDATE beta_servers SET restarted_at = ?, healthy_at = NULL WHERE beta_id = ? AND server = ?')
.run(new Date().toISOString(), id, server.server);
stage(id, server.server, 'awaiting-health');
continue;
}
if (server.stage === 'awaiting-health') {
const health = rolloutHealth(server.server);
if (!health.ok) continue;
const now = new Date().toISOString();
if (server.stage === 'awaiting-empty' || server.stage === 'awaiting-health') {
const operation = requestRestart(server.server, { kind: 'beta', id: String(id) });
if (['failed','interrupted','cancelled'].includes(operation.state)) {
const reason = operation.error ?? operation.stage;
stage(id, server.server, 'failed', reason); await fail(id, plan, reason); return;
}
if (!operation.submitted_at || operation.state === 'running') {
stage(id, server.server, 'awaiting-empty', operation.stage);
continue;
}
db().prepare('UPDATE beta_servers SET restarted_at = ? WHERE beta_id = ? AND server = ?')
.run(operation.submitted_at, id, server.server);
if (operation.state !== 'done') { stage(id, server.server, 'awaiting-health', operation.stage); continue; }
const now = operation.finished_at ?? new Date().toISOString();
db().prepare('UPDATE beta_servers SET healthy_at = ? WHERE beta_id = ? AND server = ?').run(now, id, server.server);
db().prepare('UPDATE beta_test_runs SET observing_at = ? WHERE beta_id = ? AND server = ? AND retired_at IS NULL').run(now, id, server.server);
stage(id, server.server, 'observing');
@@ -145,6 +136,8 @@ async function advancePlan(stored: StoredPlan): Promise<void> {
if (broken) { stage(id, server.server, 'failed', broken.detail); await fail(id, plan, `${server.server}: ${broken.label}: ${broken.detail}`); return; }
const recoveryTarget = Math.max(0, ...plan.requirements.filter(r => r.kind === 'recovery').map(r => r.target));
if (run.phase < recoveryTarget && progress.filter(p => !plan.requirements.some(r => r.id === p.id && r.kind === 'recovery')).every(p => p.passed)) {
releaseRestartRequest(server.server, { kind: 'beta', id: String(id) });
db().prepare('UPDATE beta_servers SET restarted_at = NULL, healthy_at = NULL WHERE beta_id = ? AND server = ?').run(id, server.server);
stage(id, server.server, 'awaiting-empty');
activity(id, `${server.server}: recovery cycle ${run.phase + 1}/${recoveryTarget} queued; gameplay requirements will be repeated`);
} else stage(id, server.server, progress.every(p => p.passed) ? 'ready' : 'observing');
src/lib/server/pipeline/beta-runner.ts
+15 -51
View File
@@ -1,9 +1,7 @@
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 } from '../integrations/rcon.js';
import { refreshRestartRollout, restartInRollout, rolloutHealth, rolloutWaitReason } from './restart-rollout.js';
import { requestRestart } from '../provisioning/index.js';
import { plannedBetaTick } from './beta-plan-runner.js';
interface BetaServerActiveRow {
@@ -20,10 +18,6 @@ function nowIso(): string {
return new Date().toISOString();
}
function integrationsOn(): boolean {
return komodoConfigured() && rconConfigured();
}
const setStage = () =>
db().prepare(
`UPDATE beta_servers
@@ -94,9 +88,7 @@ export function betaTick(): Promise<{ processed: number }> {
}
async function runBetaTick(): Promise<{ processed: number }> {
if (integrationsOn()) await refreshRestartRollout();
await plannedBetaTick();
if (!integrationsOn()) return { processed: 0 };
const rows = db()
.prepare(
@@ -131,49 +123,21 @@ async function runBetaTick(): Promise<{ processed: number }> {
async function advance(row: BetaServerActiveRow): Promise<void> {
switch (row.stage) {
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');
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';
transition(row.beta_id, row.server, next);
return;
}
case 'restarting':
case 'awaiting-health': {
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.
setStage().run(
'awaiting-health',
h.reason,
nowIso(),
'awaiting-health',
'awaiting-health',
nowIso(),
'awaiting-health',
'awaiting-health',
nowIso(),
'awaiting-health',
nowIso(),
row.beta_id,
row.server
);
const operation = requestRestart(row.server, { kind: 'beta', id: String(row.beta_id) });
if (['failed','interrupted','cancelled'].includes(operation.state)) {
transition(row.beta_id, row.server, 'failed', operation.error ?? operation.stage);
return;
}
if (!operation.submitted_at || operation.state === 'running') {
transition(row.beta_id, row.server, 'awaiting-empty', operation.stage);
return;
}
db().prepare('UPDATE beta_servers SET restarted_at = ? WHERE beta_id = ? AND server = ?')
.run(operation.submitted_at, row.beta_id, row.server);
if (row.wait_for_health && operation.state !== 'done') {
transition(row.beta_id, row.server, 'awaiting-health', operation.stage);
return;
}
const next = row.wait_for_match ? 'awaiting-match' : 'ready';
src/lib/server/pipeline/restart-rollout.ts
+1 -157
View File
@@ -1,157 +1 @@
import { config } from '../config.js';
import { db } from '../db/index.js';
import { komodoInspectHealthy, komodoRestart, komodoRuntimeAction } from '../integrations/komodo.js';
import { rconStatusResult } from '../integrations/rcon.js';
import { beginDeployment, deploymentStage, finishDeployment } from '../fleet/deployments.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;
operation_id: string | null;
}
function slotFor(server: string): RestartSlot | undefined {
return db().prepare('SELECT * FROM restart_rollout WHERE server = ?').get(server) as RestartSlot | undefined;
}
/** The queued restart may still expose the previous healthy container. */
export function restartSettling(server: string): boolean {
const slot = slotFor(server);
return requestsInFlight.has(server) || !!(slot && !slot.healthy_at
&& Date.now() - Date.parse(slot.requested_at) < HEALTH_POLL_MS);
}
export function rolloutWaitReason(server: string, operationId = ''): string | null {
if (db().prepare(`SELECT 1 FROM server_deployments WHERE server = ? AND id != ?
AND ((action IN ('start','stop','rebuild') AND state IN ('running','awaiting-health'))
OR (action = 'destroy' AND state IN ('waiting','running')))`).get(server, operationId)) {
return 'Waiting for the current provisioning operation';
}
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, options: { operationId?: string; action?: 'start' | 'restart'; triggeredBy?: string } = {}): Promise<RolloutRestartResult> {
let operationId = options.operationId;
const waitReason = db().transaction(() => {
const reason = rolloutWaitReason(server, operationId);
if (reason) return reason;
operationId ??= beginDeployment(server, options.action ?? 'restart', options.triggeredBy ?? 'automation');
db().prepare(`INSERT INTO restart_rollout (server, requested_at, operation_id) VALUES (?, ?, ?)
ON CONFLICT(server) DO UPDATE SET requested_at = excluded.requested_at,
checked_at = NULL, healthy_at = NULL, players = NULL, error = NULL, operation_id = excluded.operation_id`)
.run(server, new Date().toISOString(), operationId);
deploymentStage(operationId, options.action === 'start' ? 'Starting container' : 'Restarting container');
return null;
}).immediate();
if (waitReason) return { kind: 'queued', message: waitReason };
requestsInFlight.add(server);
db().prepare(`UPDATE servers SET container_state = 'restarting', last_error = NULL, last_reply_snippet = NULL,
game_started_at = NULL, game_ended_at = NULL WHERE name = ?`).run(server);
try {
const result = options.action === 'start' ? await komodoRuntimeAction(server, 'start') : await komodoRestart(server);
if (result.ok) {
db().prepare("UPDATE server_deployments SET state = 'awaiting-health' WHERE id = ?").run(operationId);
deploymentStage(operationId!, 'Waiting for container and RCON health');
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);
}
finishDeployment(operationId!, { ok: false, message: `komodo: ${result.message}` });
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);
finishDeployment(operationId!, { ok: false, message });
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);
if (slot.operation_id && db().prepare("SELECT 1 FROM server_deployments WHERE id = ? AND state = 'awaiting-health'").get(slot.operation_id)) {
if (error) deploymentStage(slot.operation_id, 'Waiting for health: ' + error);
else {
deploymentStage(slot.operation_id, 'Verified container and RCON health');
finishDeployment(slot.operation_id, { ok: true });
}
}
}
}
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' };
}
export { restartSettling, rolloutWaitReason, refreshRestartRollout, rolloutHealth } from '../provisioning/restarts.js';
src/lib/server/pipeline/sweep.ts
+31 -184
View File
@@ -1,202 +1,49 @@
import { config, komodoConfigured, rconConfigured } from '../config.js';
import { config } from '../config.js';
import { db } from '../db/index.js';
import { emit } from './bus.js';
import { rconGetPlayerCount } from '../integrations/rcon.js';
import { setServerField } from '../fleet/servers.js';
import { refreshRestartRollout, restartInRollout, rolloutHealth, rolloutWaitReason } from './restart-rollout.js';
import { requestRestart } from '../provisioning/index.js';
interface PendingRow {
deployment_id: string;
server: string;
attempts: number;
restarted_at: string | null;
files_changed: number;
deployment_id: string; server: string; restart_state: string;
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;
const nowIso = () => new Date().toISOString();
function nowIso(): string {
return new Date().toISOString();
}
function integrationsOn(): boolean {
return komodoConfigured() && rconConfigured();
}
/**
* Two-phase sweep of every pending/failed deployment_server row.
*
* Phase 1 (restarted_at IS NULL): the "wait for empty then restart" phase.
* - Poll RCON for player count. Busy or unreachable → stay pending.
* - Empty → call Komodo restart, set restarted_at, stay pending so the
* row picks up phase 2 on the next tick.
*
* Phase 2 (restarted_at IS NOT NULL): the "wait for healthy" phase.
* - Ask Komodo whether the container is running AND probe RCON.
* - Both green → done. Either red → stay pending; keep polling.
*
* 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.
*/
/** Submit activation requests and mirror their results. Provisioning owns all execution. */
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()
.prepare(
`UPDATE deployment_servers
SET restart_state = 'manual', checked_at = ?, error = 'integrations not configured'
WHERE restart_state IN ('pending','failed')`
)
.run(nowIso());
emit({ type: 'sweep-tick', frozen: false, reason: 'integrations not configured' });
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 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) {
emit({ type: 'sweep-tick', frozen: false });
return { checked: 0, restarted: 0, healthy: 0, held: 0, failed: 0 };
}
type Outcome = { state: 'done' | 'failed' | 'pending'; players: number | null; error: string | null; restartedNow: boolean };
// 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;
let healthy = 0;
let held = 0;
let failed = 0;
const applyOutcome = db().prepare(
`UPDATE deployment_servers
SET restart_state = ?, checked_at = ?, players = ?, error = ?, attempts = attempts + 1,
restarted_at = CASE WHEN ? = 1 THEN ? ELSE restarted_at END
WHERE deployment_id = ? AND server = ?`
);
const pending = db().prepare(`SELECT ds.* 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[];
const result: SweepResult = { checked: 0, restarted: 0, healthy: 0, held: 0, failed: 0 };
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
? 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++;
if (outcome.state === 'pending') held++;
if (outcome.state === 'failed') failed++;
// Cache last known player count for the dashboard.
if (outcome.players !== null) {
setServerField(row.server, 'last_players', outcome.players);
setServerField(row.server, 'last_checked_at', nowIso());
}
}
const now = nowIso();
applyOutcome.run(
outcome.state,
now,
outcome.players,
outcome.error,
outcome.restartedNow ? 1 : 0,
now,
row.deployment_id,
row.server
);
emit({
type: 'server-state',
deploymentId: row.deployment_id,
server: row.server,
state: outcome.state,
players: outcome.players,
error: outcome.error
});
if (!row.files_changed) continue; // A failed file write is not a restart request.
const operation = requestRestart(row.server, { kind: 'campaign', id: row.deployment_id });
result.checked++;
const state = operation.state === 'done' ? 'done'
: ['failed','interrupted','cancelled'].includes(operation.state) ? 'failed' : 'pending';
const restartedAt = ['awaiting-health','done'].includes(operation.state) ? operation.submitted_at : row.restarted_at;
if (restartedAt && !row.restarted_at) result.restarted++;
if (state === 'done') result.healthy++;
if (state === 'pending') result.held++;
if (state === 'failed') result.failed++;
const error = state === 'done' ? null : operation.error ?? operation.stage;
db().prepare(`UPDATE deployment_servers SET restart_state = ?, checked_at = ?, players = ?, error = ?,
attempts = attempts + CASE WHEN ? IS NOT NULL AND checked_at IS NOT ? THEN 1 ELSE 0 END,
restarted_at = COALESCE(restarted_at, ?) WHERE deployment_id = ? AND server = ?`)
.run(state, operation.checked_at, operation.players, error, operation.checked_at, operation.checked_at, restartedAt, row.deployment_id, row.server);
emit({ type: 'server-state', deploymentId: row.deployment_id, server: row.server, state, players: operation.players, error });
}
// Finish deployments where every server row is non-pending/non-failed.
db()
.prepare(
`UPDATE deployments
SET finished_at = ?
WHERE finished_at IS NULL
AND NOT EXISTS (
SELECT 1 FROM deployment_servers
WHERE deployment_id = deployments.id
AND restart_state IN ('pending','failed')
)`
)
.run(nowIso());
const cutoff = new Date(Date.now() - config.deployKeepDays * 24 * 60 * 60 * 1000).toISOString();
db()
.prepare(
`DELETE FROM deployments
WHERE started_at < ?
AND NOT EXISTS (
SELECT 1 FROM deployment_servers
WHERE deployment_id = deployments.id
AND restart_state IN ('pending','failed')
)`
)
.run(cutoff);
db().prepare(`UPDATE deployments SET finished_at = ? WHERE finished_at IS NULL AND NOT EXISTS (
SELECT 1 FROM deployment_servers WHERE deployment_id = deployments.id AND restart_state IN ('pending','failed','held'))`).run(nowIso());
db().prepare(`DELETE FROM deployments WHERE started_at < ? AND NOT EXISTS (
SELECT 1 FROM deployment_servers WHERE deployment_id = deployments.id AND restart_state IN ('pending','failed','held'))`)
.run(new Date(Date.now() - config.deployKeepDays * 86_400_000).toISOString());
emit({ type: 'sweep-tick', frozen: false });
return { checked, restarted, healthy, held, failed };
}
// 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 };
}
if (players > 0) {
return { state: 'pending', players, error: null, restartedNow: false };
}
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 };
}
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.
function runHealthPhase(server: string): { state: 'done' | 'pending'; players: number | null; error: string | null; restartedNow: boolean } {
const h = rolloutHealth(server);
if (!h.ok) {
return { state: 'pending', players: null, error: h.reason, restartedNow: false };
}
return { state: 'done', players: h.players, error: null, restartedNow: false };
return result;
}
src/lib/server/pipeline/worker.ts
+13 -20
View File
@@ -1,5 +1,4 @@
import { processDestructions, processCreations } from '../fleet/create.js';
import { processLifecycle } from '../fleet/lifecycle.js';
import { processProvisioning } from '../provisioning/worker.js';
import { sweep } from './sweep.js';
import { betaTick } from './beta-runner.js';
import { emit } from './bus.js';
@@ -8,7 +7,7 @@ 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 './restart-rollout.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';
@@ -33,25 +32,19 @@ function tick(): Promise<void> {
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); }
}
async function runTick(): Promise<void> {
try { await processCreations(); }
catch (err) { console.error('[worker] creation error:', err); }
try { await processDestructions(); }
catch (err) { console.error('[worker] destruction error:', err); }
try { await processLifecycle(); }
// 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); }
try {
await sweep();
} catch (err) {
console.error('[worker] sweep error:', err);
}
try {
await betaTick();
} catch (err) {
console.error('[worker] beta-runner error:', err);
}
try { await maintenanceTick(); }
catch (err) { console.error('[worker] maintenance error:', err); }
await reconcileRequesters();
}
/**