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

Add Prettier and ESLint tooling (Refs #77)

Browse Source
This commit is contained in:
josh
2026-10-09 15:56:13 -04:00
parent 8ed059b8d3
commit 16469e2d43
711 changed files with 99737 additions and 28286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/pipeline/activity.ts
+11 -4
View File
@@ -6,12 +6,19 @@ let sequence = 0;
/** Live diagnostics only. Durable operation milestones belong in provisioning history. */
export function workerActivity(): WorkerActivity[] {
return [...active.values()].map(value => ({ ...value }));
return [...active.values()].map((value) => ({ ...value }));
}
export async function trackActivity<T>(key: string, label: string, work: () => Promise<T>): Promise<T> {
export async function trackActivity<T>(
key: string,
label: string,
work: () => Promise<T>
): Promise<T> {
if (active.has(key)) key = `${key}:${++sequence}`;
active.set(key, { key, label, startedAt: new Date().toISOString() });
try { return await withOperationSource(currentOperationSource() ?? label, work); }
finally { active.delete(key); }
try {
return await withOperationSource(currentOperationSource() ?? label, work);
} finally {
active.delete(key);
}
}
src/lib/server/pipeline/beta-plan-runner.ts
+177 -50
View File
@@ -1,35 +1,67 @@
import { db } from '../db/index.js';
import { currentBetaRevision } from '../fleet/beta-revisions.js';
import { betaOperationInFlight, deployBetaServers, isActiveBetaServer, listBetaServers, rollbackBeta } from '../fleet/beta.js';
import { activity, currentRun, observeRun, prepareRun, readMetrics, saveMetrics, setOutcome, storedPlan, type StoredPlan } from '../fleet/beta-tests.js';
import {
betaOperationInFlight,
deployBetaServers,
isActiveBetaServer,
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 { requestRestart, releaseRestartRequest } from '../provisioning/index.js';
function stage(id: number, server: string, next: BetaStage, error: string | null = null): void {
const result = db().prepare('UPDATE beta_servers SET stage = ?, error = ?, updated_at = ? WHERE beta_id = ? AND server = ? AND server_deleted_at IS NULL')
const result = db()
.prepare(
'UPDATE beta_servers SET stage = ?, error = ?, updated_at = ? WHERE beta_id = ? AND server = ? AND server_deleted_at IS NULL'
)
.run(next, error, new Date().toISOString(), id, server);
if (result.changes) emit({ type: 'beta-server-stage', betaId: id, server, stage: next, error });
}
export function betaState(id: number, ready: boolean): void {
const state = ready ? 'ready-promote' : 'running';
const result = db().prepare("UPDATE betas SET state = ? WHERE id = ? AND state IN ('running','ready-promote') AND state <> ?").run(state, id, state);
const result = db()
.prepare(
"UPDATE betas SET state = ? WHERE id = ? AND state IN ('running','ready-promote') AND state <> ?"
)
.run(state, id, state);
if (result.changes) emit({ type: 'beta-state', betaId: id, state });
// A revoked pass must not leave this version labelled tested unless another beta still proves it.
db().prepare(`UPDATE script_versions SET state = ? WHERE id IN (SELECT script_version_id FROM beta_scripts WHERE beta_id = ?)
db()
.prepare(
`UPDATE script_versions SET state = ? WHERE id IN (SELECT script_version_id FROM beta_scripts WHERE beta_id = ?)
AND id NOT IN (SELECT current_version_id FROM scripts WHERE current_version_id IS NOT NULL)
AND NOT EXISTS (SELECT 1 FROM beta_scripts bs JOIN betas b ON b.id = bs.beta_id
WHERE bs.script_version_id = script_versions.id AND b.id <> ? AND b.state IN ('ready-promote','promoted'))`)
WHERE bs.script_version_id = script_versions.id AND b.id <> ? AND b.state IN ('ready-promote','promoted'))`
)
.run(ready ? 'tested' : 'untested', id, id);
}
export async function rollbackTestPlan(id: number, plan: BetaPlan): Promise<void> {
const result = await rollbackBeta(id, { restart: plan.rollbackRestart });
if (!result.deploymentId) activity(id, 'Cancelled queued changes; no deployed files need restoration');
else if (plan.rollbackRestart === 'when-empty') activity(id, 'Queued file restoration and rolling restarts; waiting for empty servers');
else activity(id, 'File restoration is held; resume the rollback campaign to apply it with a safe restart');
if (!result.deploymentId)
activity(id, 'Cancelled queued changes; no deployed files need restoration');
else if (plan.rollbackRestart === 'when-empty')
activity(id, 'Queued file restoration and rolling restarts; waiting for empty servers');
else
activity(
id,
'File restoration is held; resume the rollback campaign to apply it with a safe restart'
);
if (result.warnings.length) activity(id, result.warnings.join('; '));
}
@@ -37,27 +69,41 @@ async function fail(id: number, plan: BetaPlan, reason: string): Promise<void> {
betaState(id, false);
setOutcome(id, plan.failureAction === 'pause' ? 'paused' : 'failed', reason);
if (plan.failureAction === 'rollback') {
try { await rollbackTestPlan(id, plan); }
catch (error) { setOutcome(id, 'failed', `Automatic rollback failed: ${(error as Error).message}`); }
try {
await rollbackTestPlan(id, plan);
} catch (error) {
setOutcome(id, 'failed', `Automatic rollback failed: ${(error as Error).message}`);
}
}
}
let inFlight: Promise<void> | null = null;
export function plannedBetaTick(): Promise<void> {
if (inFlight) return inFlight;
inFlight = tick().finally(() => { inFlight = null; });
inFlight = tick().finally(() => {
inFlight = null;
});
return inFlight;
}
async function tick(): Promise<void> {
const plans = db().prepare(`SELECT p.* FROM beta_test_plans p JOIN betas b ON b.id = p.beta_id
WHERE b.state IN ('running','ready-promote') AND p.outcome IN ('active','passed') ORDER BY p.beta_id`).all() as StoredPlan[];
const plans = db()
.prepare(
`SELECT p.* FROM beta_test_plans p JOIN betas b ON b.id = p.beta_id
WHERE b.state IN ('running','ready-promote') AND p.outcome IN ('active','passed') ORDER BY p.beta_id`
)
.all() as StoredPlan[];
for (const stored of plans) {
if (betaOperationInFlight(stored.beta_id)) continue;
try { await advancePlan(stored); }
catch (error) {
try {
await advancePlan(stored);
} catch (error) {
const reason = `Test runner: ${(error as Error).message}`;
try { await fail(stored.beta_id, parseBetaPlan(JSON.parse(stored.plan)), reason); }
catch { betaState(stored.beta_id, false); setOutcome(stored.beta_id, 'failed', reason); }
try {
await fail(stored.beta_id, parseBetaPlan(JSON.parse(stored.plan)), reason);
} catch {
betaState(stored.beta_id, false);
setOutcome(stored.beta_id, 'failed', reason);
}
}
}
}
@@ -68,13 +114,20 @@ async function advancePlan(stored: StoredPlan): Promise<void> {
const servers = listBetaServers(id);
// Preserve the original coverage and evidence. A replacement with the same
// name cannot complete a test that began on a different server.
const missing = servers.filter(server => !isActiveBetaServer(id, server.server));
const missing = servers.filter((server) => !isActiveBetaServer(id, server.server));
if (missing.length) {
betaState(id, false);
setOutcome(id, 'paused', `Test server deleted: ${missing.map(server => server.server).join(', ')}. Roll back this beta and start a new test.`);
setOutcome(
id,
'paused',
`Test server deleted: ${missing.map((server) => server.server).join(', ')}. Roll back this beta and start a new test.`
);
return;
}
if (stored.outcome !== 'passed' && Date.now() - Date.parse(currentBetaRevision(id).started_at) >= plan.timeoutHours * 3_600_000) {
if (
stored.outcome !== 'passed' &&
Date.now() - Date.parse(currentBetaRevision(id).started_at) >= plan.timeoutHours * 3_600_000
) {
betaState(id, false);
setOutcome(id, 'inconclusive', 'Deadline reached without enough qualifying evidence');
return;
@@ -83,19 +136,38 @@ async function advancePlan(stored: StoredPlan): Promise<void> {
if (!isActiveBetaServer(id, server.server)) return;
const freshPlan = storedPlan(id)!;
if (!['active', 'passed'].includes(freshPlan.outcome)) return;
if (server.stage === 'failed') { await fail(id, plan, `${server.server}: ${server.error ?? 'server failed'}`); return; }
if (server.stage === 'failed') {
await fail(id, plan, `${server.server}: ${server.error ?? 'server failed'}`);
return;
}
if (server.stage === 'deploying' || server.stage === 'restarting') {
stage(id, server.server, 'failed', `${server.stage === 'deploying' ? 'deploy: ' : ''}Operation interrupted; review the server before starting a fresh attempt`);
await fail(id, plan, `${server.server}: interrupted ${server.stage}`); return;
stage(
id,
server.server,
'failed',
`${server.stage === 'deploying' ? 'deploy: ' : ''}Operation interrupted; review the server before starting a fresh attempt`
);
await fail(id, plan, `${server.server}: interrupted ${server.stage}`);
return;
}
if (server.stage === 'queued') continue;
if (server.stage === 'awaiting-empty') {
const waitingRun = currentRun(id, server.server);
if (waitingRun?.observing_at) {
await observeRun(waitingRun, plan);
if (!['active', 'passed'].includes(storedPlan(id)!.outcome)) { betaState(id, false); return; }
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 (!['active', 'passed'].includes(storedPlan(id)!.outcome)) {
betaState(id, false);
return;
}
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;
}
}
// Prepare evidence once before handing off the restart. Empty-server
// checks and remote execution belong entirely to Provisioning.
@@ -113,61 +185,116 @@ async function advancePlan(stored: StoredPlan): Promise<void> {
if (!isActiveBetaServer(id, server.server)) return;
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)) {
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;
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 = ?')
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; }
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);
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');
}
const run = currentRun(id, server.server);
if (!run?.observing_at) continue;
await observeRun(run, plan);
if (!isActiveBetaServer(id, server.server)) return;
if (!['active', 'passed'].includes(storedPlan(id)!.outcome)) { betaState(id, false); return; }
if (!['active', 'passed'].includes(storedPlan(id)!.outcome)) {
betaState(id, false);
return;
}
const metrics = readMetrics(currentRun(id, server.server)!);
const progress = evaluateRequirements(plan, metrics, run.phase);
const broken = progress.find(p => p.failed);
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)) {
const broken = progress.find((p) => p.failed);
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);
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');
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');
}
const updated = listBetaServers(id);
if (updated.some(server => !isActiveBetaServer(id, server.server))) return;
const queued = updated.filter(s => s.stage === 'queued');
if (queued.length && updated.filter(s => s.stage !== 'queued').every(s => s.stage === 'ready')) {
const wave = queued.slice(0, plan.waveSize || queued.length).map(s => s.server);
if (updated.some((server) => !isActiveBetaServer(id, server.server))) return;
const queued = updated.filter((s) => s.stage === 'queued');
if (
queued.length &&
updated.filter((s) => s.stage !== 'queued').every((s) => s.stage === 'ready')
) {
const wave = queued.slice(0, plan.waveSize || queued.length).map((s) => s.server);
activity(id, `Previous wave passed; deploying ${wave.join(', ')}`);
await deployBetaServers(id, wave);
betaState(id, false);
return;
}
const registry = db().prepare('SELECT server AS name, game, map FROM beta_test_coverage WHERE beta_id = ?').all(id) as Array<{ name: string; game: string | null; map: string | null }>;
const registry = db()
.prepare('SELECT server AS name, game, map FROM beta_test_coverage WHERE beta_id = ?')
.all(id) as Array<{ name: string; game: string | null; map: string | null }>;
// All waves must be activated, even when the final coverage rule is a quorum.
// A passing server proves only its assigned files. A quorum or coverage
// group must not pass an unrelated game/map's untested bundle member.
const ready = !queued.length && updated.every(s => !!currentRun(id, s.server)?.observing_at)
&& coveragePassed(plan, updated.map(s => ({ ...registry.find(r => r.name === s.server)!, passed: s.stage === 'ready' })))
&& !db().prepare(`SELECT 1 FROM beta_scripts files WHERE files.beta_id = ? AND NOT EXISTS (
const ready =
!queued.length &&
updated.every((s) => !!currentRun(id, s.server)?.observing_at) &&
coveragePassed(
plan,
updated.map((s) => ({
...registry.find((r) => r.name === s.server)!,
passed: s.stage === 'ready'
}))
) &&
!db()
.prepare(
`SELECT 1 FROM beta_scripts files WHERE files.beta_id = ? AND NOT EXISTS (
SELECT 1 FROM beta_server_scripts members JOIN beta_servers servers
ON servers.beta_id = members.beta_id AND servers.server = members.server
WHERE members.beta_id = files.beta_id AND members.script_id = files.script_id
AND servers.server_deleted_at IS NULL AND servers.stage = 'ready'
) LIMIT 1`).get(id);
) LIMIT 1`
)
.get(id);
betaState(id, ready);
setOutcome(id, ready ? 'passed' : 'active', null);
}
src/lib/server/pipeline/beta-runner.ts
+11 -4
View File
@@ -72,7 +72,9 @@ function reconcileBetaState(betaId: number): void {
const versionRows = db()
.prepare(`SELECT script_version_id FROM beta_scripts WHERE beta_id = ?`)
.all(betaId) as Array<{ script_version_id: number }>;
const markTested = db().prepare(`UPDATE script_versions SET state = 'tested' WHERE id = ?`);
const markTested = db().prepare(
`UPDATE script_versions SET state = 'tested' WHERE id = ?`
);
for (const v of versionRows) markTested.run(v.script_version_id);
}
emit({ type: 'beta-state', betaId, state: 'ready-promote' });
@@ -83,7 +85,9 @@ let betaInFlight: Promise<{ processed: number }> | null = null;
export function betaTick(): Promise<{ processed: number }> {
if (betaInFlight) return betaInFlight;
betaInFlight = runBetaTick().finally(() => { betaInFlight = null; });
betaInFlight = runBetaTick().finally(() => {
betaInFlight = null;
});
return betaInFlight;
}
@@ -127,7 +131,7 @@ async function advance(row: BetaServerActiveRow): Promise<void> {
case 'restarting':
case 'awaiting-health': {
const operation = requestRestart(row.server, { kind: 'beta', id: String(row.beta_id) });
if (['failed','interrupted','cancelled'].includes(operation.state)) {
if (['failed', 'interrupted', 'cancelled'].includes(operation.state)) {
transition(row.beta_id, row.server, 'failed', operation.error ?? operation.stage);
return;
}
@@ -135,7 +139,10 @@ async function advance(row: BetaServerActiveRow): Promise<void> {
transition(row.beta_id, row.server, 'awaiting-empty', operation.stage);
return;
}
db().prepare('UPDATE beta_servers SET restarted_at = ? WHERE beta_id = ? AND server = ?')
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);
src/lib/server/pipeline/bus.ts
+16 -6
View File
@@ -85,22 +85,32 @@ const changedBetas = new Set<number>();
export function invalidateBeta(betaId: number): void {
const scheduled = changedBetas.size > 0;
changedBetas.add(betaId);
if (!scheduled) queueMicrotask(() => {
const ids = [...changedBetas]; changedBetas.clear();
for (const betaId of ids) bus.emit('event', { type: 'beta-changed', betaId });
});
if (!scheduled)
queueMicrotask(() => {
const ids = [...changedBetas];
changedBetas.clear();
for (const betaId of ids) bus.emit('event', { type: 'beta-changed', betaId });
});
}
let invalidationPending = false;
export function invalidateFleet(): void {
if (invalidationPending) return;
invalidationPending = true;
// Flush after nested database transactions have committed.
queueMicrotask(() => { invalidationPending = false; bus.emit('event', { type: 'fleet-invalidated' }); });
queueMicrotask(() => {
invalidationPending = false;
bus.emit('event', { type: 'fleet-invalidated' });
});
}
export function emit(event: PipelineEvent): void {
bus.emit('event', event);
if (event.type === 'beta-opened' || event.type === 'beta-state' || event.type === 'beta-server-stage') invalidateFleet();
if (
event.type === 'beta-opened' ||
event.type === 'beta-state' ||
event.type === 'beta-server-stage'
)
invalidateFleet();
if ('betaId' in event && event.type !== 'beta-changed') invalidateBeta(event.betaId);
}
src/lib/server/pipeline/event-stream.ts
+34 -23
View File
@@ -13,7 +13,11 @@ interface EventStreamOptions {
initialFrames?: Uint8Array[];
}
export function eventStreamResponse({ accepts, encode, initialFrames = [] }: EventStreamOptions): Response {
export function eventStreamResponse({
accepts,
encode,
initialFrames = []
}: EventStreamOptions): Response {
let heartbeat: ReturnType<typeof setInterval> | undefined;
let unsubscribe: (() => void) | undefined;
let closed = false;
@@ -25,30 +29,37 @@ export function eventStreamResponse({ accepts, encode, initialFrames = [] }: Eve
unsubscribe = undefined;
};
const stream = new ReadableStream<Uint8Array>({
start(controller) {
const send = (frame: () => Uint8Array) => {
const stream = new ReadableStream<Uint8Array>(
{
start(controller) {
const send = (frame: () => Uint8Array) => {
if (closed) return;
if ((controller.desiredSize ?? 0) <= 0) {
// Reconnect for fresh state instead of silently losing deltas.
// close() does not invoke cancel(), so release resources here.
cleanup();
controller.close();
return;
}
try {
controller.enqueue(frame());
} catch (cause) {
cleanup();
controller.error(cause);
}
};
send(() => hello);
for (const frame of initialFrames) send(() => frame);
if (closed) return;
if ((controller.desiredSize ?? 0) <= 0) {
// Reconnect for fresh state instead of silently losing deltas.
// close() does not invoke cancel(), so release resources here.
cleanup();
controller.close();
return;
}
try { controller.enqueue(frame()); }
catch (cause) { cleanup(); controller.error(cause); }
};
send(() => hello);
for (const frame of initialFrames) send(() => frame);
if (closed) return;
unsubscribe = subscribe(event => {
if (accepts(event)) send(() => encode(event));
});
heartbeat = setInterval(() => send(() => ping), HEARTBEAT_MS);
unsubscribe = subscribe((event) => {
if (accepts(event)) send(() => encode(event));
});
heartbeat = setInterval(() => send(() => ping), HEARTBEAT_MS);
},
cancel: cleanup
},
cancel: cleanup
}, { highWaterMark: MAX_QUEUED_FRAMES });
{ highWaterMark: MAX_QUEUED_FRAMES }
);
return new Response(stream, {
headers: {
src/lib/server/pipeline/sweep.ts
+68 -18
View File
@@ -3,47 +3,97 @@ import { emit } from './bus.js';
import { requestRestart } from '../provisioning/index.js';
interface PendingRow {
deployment_id: string; server: string; restart_state: string;
restarted_at: string | null; files_changed: number;
players: number | null; error: string | null;
deployment_id: string;
server: string;
restart_state: string;
restarted_at: string | null;
files_changed: number;
players: number | null;
error: string | null;
}
type SweepResult = { checked: number; restarted: number; healthy: number; held: number; failed: number };
type SweepResult = {
checked: number;
restarted: number;
healthy: number;
held: number;
failed: number;
};
let sweepInFlight: Promise<SweepResult> | null = null;
const nowIso = () => new Date().toISOString();
/** 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; });
sweepInFlight = runSweep().finally(() => {
sweepInFlight = null;
});
return sweepInFlight;
}
async function runSweep(): Promise<SweepResult> {
const pending = db().prepare(`SELECT ds.* FROM deployment_servers ds JOIN deployments d ON d.id = ds.deployment_id
const pending = db()
.prepare(
`SELECT ds.* FROM deployment_servers ds JOIN deployments d ON d.id = ds.deployment_id
WHERE ds.server_deleted_at IS NULL AND ds.restart_state IN ('pending','failed')
AND NOT EXISTS (SELECT 1 FROM retention_controls c WHERE c.category='campaign' AND c.record_id=ds.deployment_id AND c.resolved_at IS NOT NULL) ORDER BY d.started_at, d.id, ds.server`).all() as PendingRow[];
AND NOT EXISTS (SELECT 1 FROM retention_controls c WHERE c.category='campaign' AND c.record_id=ds.deployment_id AND c.resolved_at IS NOT NULL) 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) {
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;
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 = ?,
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);
if (state !== row.restart_state || operation.players !== row.players || error !== row.error) {
emit({ type: 'server-state', deploymentId: row.deployment_id, server: row.server, state, players: operation.players, error });
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
);
if (
state !== row.restart_state ||
operation.players !== row.players ||
error !== row.error
) {
emit({
type: 'server-state',
deploymentId: row.deployment_id,
server: row.server,
state,
players: operation.players,
error
});
}
}
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 server_deleted_at IS NULL AND restart_state IN ('pending','failed','held'))`).run(nowIso());
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 server_deleted_at IS NULL AND restart_state IN ('pending','failed','held'))`
)
.run(nowIso());
emit({ type: 'sweep-tick', frozen: false });
return result;
}
src/lib/server/pipeline/worker.ts
+244 -64
View File
@@ -10,7 +10,16 @@ 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 {
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';
@@ -46,10 +55,17 @@ function tick(requestAnother = false): Promise<void> {
tickInFlight = (async () => {
do {
tickRequested = false;
try { await trackActivity('provisioning', 'Processing provisioning requests', () => processProvisioning({ wait: false })); }
catch (err) { console.error('[worker] provisioning error:', err); }
try {
await trackActivity('provisioning', 'Processing provisioning requests', () =>
processProvisioning({ wait: false })
);
} catch (err) {
console.error('[worker] provisioning error:', err);
}
} while (tickRequested);
})().finally(() => { tickInFlight = null; });
})().finally(() => {
tickInFlight = null;
});
return tickInFlight;
}
@@ -64,7 +80,15 @@ function reconcileRequesters(): void {
['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(); }]
[
'retention',
'Cleaning expired history',
async () => {
await drainRetention();
pruneAutoscaling();
pruneRebalancing();
}
]
];
for (const [key, label, work] of requesters) {
if (requesterFlights.has(key)) continue;
@@ -73,9 +97,21 @@ function reconcileRequesters(): void {
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); });
.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);
}
}
@@ -92,47 +128,73 @@ function encodeError(reason: RconFailure, detail: string): string {
// 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>> }
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 });
});
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); });
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> {
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); }
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; });
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));
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); }
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;
@@ -146,31 +208,63 @@ 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.telemetryEvidence(server.name)?.players.length !== 0).map(server => server.name), snapshot);
const results = await Promise.all(servers.map(server => queueStatus(server, snapshot, probes.get(server.name))));
const probes = prepareMonitoringRcon(
servers
.filter(
(server) =>
!pendingStatus.has(server.name) &&
!snapshot.provisioning(server.name) &&
!snapshot.settling(server.name) &&
snapshot.telemetryEvidence(server.name)?.players.length !== 0
)
.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); }
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> {
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 = {}; };
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);
@@ -178,25 +272,57 @@ async function refreshServerStatus(server: string, snapshot: MonitoringSnapshot,
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. */ }
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;
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 });
? {
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';
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
@@ -210,7 +336,9 @@ async function refreshServerStatus(server: string, snapshot: MonitoringSnapshot,
// 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 => {
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.
@@ -218,13 +346,26 @@ async function refreshServerStatus(server: string, snapshot: MonitoringSnapshot,
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 };
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) {
if (
unchanged() &&
previous &&
(previous.last_checked_at ?? '') <= probe.observedAt
) {
gameObservation = probe;
const now = probe.observedAt;
updates.last_checked_at = now;
@@ -235,10 +376,14 @@ async function refreshServerStatus(server: string, snapshot: MonitoringSnapshot,
} 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 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.rcon_failures =
consecutive && previous.last_checked_at === now
? failures
: Math.min(2, failures + 1);
updates.last_error = encodeError(result.reason, result.detail);
}
}
@@ -257,17 +402,37 @@ async function refreshServerStatus(server: string, snapshot: MonitoringSnapshot,
return capacity;
}
export function statusEntry(srv: ServerRow, observe = serverObservation, capacity?: PlayerCapacityObservation): ServerStatusEntry {
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,
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 };
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 {
@@ -276,13 +441,26 @@ export function startWorker(): void {
void tick();
reconcileRequesters();
void fanoutStatus().catch(err => { console.error('[worker] status fanout error:', err); });
tickTimer = setInterval(() => { void tick(); reconcileRequesters(); }, TICK_MS);
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); });
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');
console.log(
'[worker] started — sweep every',
TICK_MS,
'ms, status every',
STATUS_REFRESH_MS,
'ms'
);
}
export function stopWorker(): void {
@@ -303,4 +481,6 @@ export async function kick(): Promise<void> {
}
/** Agent receipts need observation promptly, without launching another fleet scan. */
export async function kickProvisioning(): Promise<void> { await tick(true); }
export async function kickProvisioning(): Promise<void> {
await tick(true);
}