Add configurable beta test plans and staged rollout
This commit is contained in:
@@ -0,0 +1,182 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { open, stat } from 'node:fs/promises';
|
||||
import { db } from '../db/index.js';
|
||||
import { parseBetaPlan, type BetaPlan, type BetaTestView } from '../../beta-plan.js';
|
||||
import { applySample, applySignal, emptyMetrics, evaluateRequirements, type TestMetrics, type TestSignal } from './beta-evaluation.js';
|
||||
import { gameLogKey, resolveLogPath, type LogKey } from './logs.js';
|
||||
|
||||
export interface TestRun { id: string; beta_id: number; server: string; phase: number; created_at: string; observing_at: string | null; metrics: string; cursors: string }
|
||||
export interface StoredPlan { beta_id: number; plan: string; outcome: string; reason: string | null }
|
||||
interface Cursor { position: number; identity: string; partial: string; anchor?: string; generation?: string; session?: string; joined?: boolean; started?: number }
|
||||
const nowIso = () => new Date().toISOString();
|
||||
export const storedPlan = (id: number) => db().prepare('SELECT * FROM beta_test_plans WHERE beta_id = ?').get(id) as StoredPlan | undefined;
|
||||
export const currentRun = (id: number, server: string) => db().prepare('SELECT * FROM beta_test_runs WHERE beta_id = ? AND server = ? AND retired_at IS NULL').get(id, server) as TestRun | undefined;
|
||||
export function activity(id: number, message: string) { db().prepare('INSERT INTO beta_events (beta_id, message) VALUES (?, ?)').run(id, message); }
|
||||
export function setOutcome(id: number, outcome: string, reason: string | null): void {
|
||||
const previous = storedPlan(id);
|
||||
if (!previous || (previous.outcome === outcome && previous.reason === reason)) return;
|
||||
db().prepare('UPDATE beta_test_plans SET outcome = ?, reason = ? WHERE beta_id = ?').run(outcome, reason, id);
|
||||
activity(id, `Test plan ${outcome}${reason ? `: ${reason}` : ''}`);
|
||||
}
|
||||
export function readMetrics(run: TestRun): TestMetrics { return { ...emptyMetrics(), ...JSON.parse(run.metrics) }; }
|
||||
export function saveMetrics(run: TestRun, metrics: TestMetrics): void { db().prepare('UPDATE beta_test_runs SET metrics = ? WHERE id = ?').run(JSON.stringify(metrics), run.id); }
|
||||
export function evidence(run: TestRun, eventId: string, kind: string, payload: unknown): boolean {
|
||||
return db().prepare('INSERT OR IGNORE INTO beta_test_evidence (run_id, event_id, received_at, kind, payload) VALUES (?, ?, ?, ?, ?)')
|
||||
.run(run.id, eventId, nowIso(), kind, JSON.stringify(payload)).changes > 0;
|
||||
}
|
||||
|
||||
/** Snapshot log offsets BEFORE restart: pre-beta log entries can never qualify. */
|
||||
export async function prepareRun(betaId: number, server: string, phase = 0): Promise<TestRun> {
|
||||
const cursors: Record<string, Cursor> = {};
|
||||
for (const key of [gameLogKey(server), 'console'] as LogKey[]) {
|
||||
const path = resolveLogPath(server, key);
|
||||
const info = path ? await stat(path).catch(() => null) : null;
|
||||
let anchor = '';
|
||||
if (path && info?.size) {
|
||||
const file = await open(path, 'r');
|
||||
try {
|
||||
const buffer = Buffer.alloc(Math.min(info.size, 128));
|
||||
await file.read(buffer, 0, buffer.length, info.size - buffer.length);
|
||||
anchor = buffer.toString('base64');
|
||||
} finally { await file.close(); }
|
||||
}
|
||||
cursors[key] = { position: info?.size ?? 0, identity: info ? `${info.dev}:${info.ino}` : '', partial: '', anchor, generation: randomUUID() };
|
||||
}
|
||||
const id = randomUUID();
|
||||
db().transaction(() => {
|
||||
db().prepare('UPDATE beta_test_runs SET retired_at = ? WHERE beta_id = ? AND server = ? AND retired_at IS NULL').run(nowIso(), betaId, server);
|
||||
db().prepare('INSERT INTO beta_test_runs (id, beta_id, server, phase, created_at, metrics, cursors) VALUES (?, ?, ?, ?, ?, ?, ?)')
|
||||
.run(id, betaId, server, phase, nowIso(), JSON.stringify(emptyMetrics()), JSON.stringify(cursors));
|
||||
})();
|
||||
return currentRun(betaId, server)!;
|
||||
}
|
||||
|
||||
export function acceptTestSignal(server: string, runId: string, eventId: string, signal: TestSignal): { accepted: boolean; reason?: string } {
|
||||
return db().transaction(() => {
|
||||
const run = db().prepare(`SELECT r.* FROM beta_test_runs r JOIN betas b ON b.id = r.beta_id
|
||||
JOIN beta_test_plans p ON p.beta_id = b.id WHERE r.id = ? AND r.server = ? AND r.retired_at IS NULL
|
||||
AND b.state IN ('running','ready-promote') AND p.outcome IN ('active','passed')`).get(runId, server) as TestRun | undefined;
|
||||
if (!run) return { accepted: false, reason: 'Run is inactive or superseded; obtain the current run at process startup' };
|
||||
if (db().prepare('SELECT 1 FROM beta_test_evidence WHERE run_id = ? AND event_id = ?').get(runId, `telemetry:${eventId}`)) return { accepted: false, reason: 'Duplicate event' };
|
||||
const plan = parseBetaPlan(JSON.parse(storedPlan(run.beta_id)!.plan));
|
||||
const metrics = readMetrics(run);
|
||||
if (!applySignal(plan, metrics, signal, Date.now())) return { accepted: false, reason: 'Duplicate, out-of-order, closed, or unknown session/boot' };
|
||||
evidence(run, `telemetry:${eventId}`, signal.event, signal);
|
||||
saveMetrics(run, metrics);
|
||||
return { accepted: true };
|
||||
})();
|
||||
}
|
||||
|
||||
export function parseTestSignal(body: Record<string, unknown>): TestSignal {
|
||||
if (!['boot', 'session_start', 'round_change', 'match_end', 'feature'].includes(String(body.event))) throw new Error('Unknown telemetry event');
|
||||
for (const key of ['runId', 'eventId', 'bootId'] as const) {
|
||||
if (typeof body[key] !== 'string' || !/^[a-zA-Z0-9_.:-]{1,128}$/.test(body[key] as string)) throw new Error(`${key} is required (1–128 identifier characters)`);
|
||||
}
|
||||
if (body.event !== 'boot' && (typeof body.sessionId !== 'string' || !/^[a-zA-Z0-9_.:-]{1,128}$/.test(body.sessionId))) throw new Error('sessionId is required');
|
||||
for (const key of ['sequence', 'round', 'players']) if (!Number.isSafeInteger(body[key]) || (body[key] as number) < 0) throw new Error(`${key} must be a nonnegative integer`);
|
||||
if (body.event === 'feature' && (typeof body.name !== 'string' || !/^[a-zA-Z0-9_.:-]{1,120}$/.test(body.name))) throw new Error('Feature name is required');
|
||||
return { event: body.event as TestSignal['event'], bootId: body.bootId as string, sessionId: body.sessionId as string ?? '',
|
||||
sequence: body.sequence as number, round: body.round as number, players: body.players as number, name: body.name as string ?? '' };
|
||||
}
|
||||
|
||||
/** Bounded incremental reads, with persistent byte cursors and partial lines. */
|
||||
export async function observeRun(run: TestRun, plan: BetaPlan): Promise<void> {
|
||||
const cursors = JSON.parse(run.cursors) as Record<string, Cursor>;
|
||||
const observed: Array<{ key: LogKey; line: string; id: string }> = [];
|
||||
let logAvailable = false;
|
||||
for (const key of [gameLogKey(run.server), 'console'] as LogKey[]) {
|
||||
const path = resolveLogPath(run.server, key);
|
||||
const file = path ? await open(path, 'r').catch(() => null) : null;
|
||||
if (!file) continue;
|
||||
try {
|
||||
const info = await file.stat();
|
||||
if (key === 'console') logAvailable = true;
|
||||
const identity = `${info.dev}:${info.ino}`;
|
||||
let cursor = cursors[key] ?? { position: 0, identity, partial: '' };
|
||||
let anchorChanged = false;
|
||||
if (cursor.anchor && info.size >= cursor.position) {
|
||||
const expected = Buffer.from(cursor.anchor, 'base64');
|
||||
const actual = Buffer.alloc(expected.length);
|
||||
await file.read(actual, 0, actual.length, cursor.position - actual.length);
|
||||
anchorChanged = !actual.equals(expected);
|
||||
}
|
||||
if (cursor.identity !== identity || info.size < cursor.position || anchorChanged) {
|
||||
cursor = { position: 0, identity, partial: '', generation: randomUUID() };
|
||||
observed.push({ key, line: '', id: `rotation:${key}:${randomUUID()}` });
|
||||
}
|
||||
if (info.size - cursor.position > 2_000_000) {
|
||||
setOutcome(run.beta_id, 'inconclusive', `${run.server}: log reader fell behind; evidence is incomplete`);
|
||||
return;
|
||||
}
|
||||
const buffer = Buffer.alloc(info.size - cursor.position);
|
||||
const { bytesRead } = await file.read(buffer, 0, buffer.length, cursor.position);
|
||||
const combined = Buffer.concat([Buffer.from(cursor.partial, 'base64'), buffer.subarray(0, bytesRead)]);
|
||||
let start = 0;
|
||||
for (let i = 0; i < combined.length; i++) if (combined[i] === 10) {
|
||||
observed.push({ key, line: combined.subarray(start, i).toString('utf8'), id: `${key}:${identity}:${cursor.generation ?? ''}:${cursor.position + i - Buffer.from(cursor.partial, 'base64').length}` });
|
||||
start = i + 1;
|
||||
}
|
||||
cursor.position += bytesRead;
|
||||
const anchor = Buffer.alloc(Math.min(cursor.position, 128));
|
||||
await file.read(anchor, 0, anchor.length, cursor.position - anchor.length);
|
||||
cursor.anchor = anchor.toString('base64');
|
||||
cursor.partial = combined.subarray(start).toString('base64');
|
||||
if (cursor.partial.length > 100_000) cursor.partial = ''; // malformed giant line cannot grow forever
|
||||
cursor.identity = identity;
|
||||
cursors[key] = cursor;
|
||||
} finally { await file.close(); }
|
||||
}
|
||||
db().transaction(() => {
|
||||
const fresh = currentRun(run.beta_id, run.server);
|
||||
if (!fresh || fresh.id !== run.id) return;
|
||||
const m = readMetrics(fresh);
|
||||
m.logAvailable = logAvailable;
|
||||
for (const entry of observed) {
|
||||
const cursor = cursors[entry.key];
|
||||
if (!entry.line) { evidence(run, entry.id, 'log-rotation', { log: entry.key }); continue; }
|
||||
if (entry.key === 'console') {
|
||||
if (/script (?:runtime )?error|unhandled exception|script exception|fatal error/i.test(entry.line) && evidence(run, entry.id, 'script-error', { line: entry.line.slice(0, 1000) })) m.logErrors++;
|
||||
continue;
|
||||
}
|
||||
if (/InitGame:/.test(entry.line)) { cursor.session = entry.id; cursor.joined = false; cursor.started = undefined; }
|
||||
if (cursor.session && /\sJ;[^;]*;\d+;/.test(entry.line) && !cursor.joined) { cursor.joined = true; cursor.started = Date.now(); }
|
||||
if (/ExitLevel:\s*executed/.test(entry.line) && cursor.session && cursor.joined) {
|
||||
// Unqualified match counts use logs. Round/population-qualified counts use session telemetry exclusively.
|
||||
if (evidence(run, `match:${cursor.session}`, 'completed-match', { session: cursor.session, line: entry.line, minutes: (Date.now() - (cursor.started ?? Date.now())) / 60_000 })) {
|
||||
for (const r of plan.requirements) if (r.kind === 'matches' && r.minRound === 0 && r.minPlayers <= 1
|
||||
&& Date.now() - (cursor.started ?? Date.now()) >= r.minMinutes * 60_000) m.values[r.id] = (m.values[r.id] ?? 0) + 1;
|
||||
}
|
||||
cursor.joined = false;
|
||||
}
|
||||
if (/ShutdownGame:/.test(entry.line)) cursor.joined = false;
|
||||
}
|
||||
const server = db().prepare('SELECT last_checked_at, last_seen_at, last_players, last_error, container_state FROM servers WHERE name = ?').get(run.server) as {
|
||||
last_checked_at: string | null; last_seen_at: string | null; last_players: number | null; last_error: string | null; container_state: string | null };
|
||||
const sample = server.last_checked_at ? Date.parse(server.last_checked_at) : 0;
|
||||
const freshSample = sample >= Date.parse(run.observing_at ?? run.created_at) && Date.now() - sample <= 60_000;
|
||||
const healthy = freshSample && !server.last_error && server.container_state === 'running' && server.last_seen_at === server.last_checked_at;
|
||||
const at = freshSample && server.container_state === 'running' ? sample : Date.now();
|
||||
// Startup requires fresh status; don't count an absent first observation as an outage.
|
||||
const before = m.wasHealthy;
|
||||
const lastSample = m.lastSample;
|
||||
applySample(plan, m, at, healthy, healthy ? server.last_players ?? 0 : 0);
|
||||
if (freshSample && sample !== lastSample) evidence(run, `sample:${sample}`, 'status-sample', { healthy, players: server.last_players, checkedAt: server.last_checked_at });
|
||||
if (before !== m.wasHealthy) evidence(run, `health:${at}`, 'health', { healthy, players: server.last_players, reason: server.last_error });
|
||||
saveMetrics(run, m);
|
||||
db().prepare('UPDATE beta_test_runs SET cursors = ? WHERE id = ?').run(JSON.stringify(cursors), run.id);
|
||||
})();
|
||||
}
|
||||
|
||||
export function testView(betaId: number): BetaTestView | null {
|
||||
const stored = storedPlan(betaId);
|
||||
if (!stored) return null;
|
||||
const plan = parseBetaPlan(JSON.parse(stored.plan));
|
||||
const runs = db().prepare('SELECT * FROM beta_test_runs WHERE beta_id = ? AND retired_at IS NULL ORDER BY server').all(betaId) as TestRun[];
|
||||
return { plan, outcome: stored.outcome, reason: stored.reason,
|
||||
servers: runs.map(run => {
|
||||
const progress = evaluateRequirements(plan, readMetrics(run), run.phase);
|
||||
return { server: run.server, runId: run.id, phase: run.phase, passed: progress.every(r => r.passed), progress };
|
||||
}),
|
||||
evidence: db().prepare(`SELECT e.id, r.server, e.received_at, e.kind, e.payload FROM beta_test_evidence e
|
||||
JOIN beta_test_runs r ON r.id = e.run_id WHERE r.beta_id = ? ORDER BY e.id DESC LIMIT 200`).all(betaId) as BetaTestView['evidence'] };
|
||||
}
|
||||
Reference in New Issue
Block a user