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

Allow revising active betas with fresh tests and history

Browse Source
This commit is contained in:
josh
2026-10-05 21:02:58 -04:00
parent 9d4ff992a4
commit f8d0ba3936
21 changed files with 713 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/fleet/beta-revisions.ts
+29
View File
@@ -0,0 +1,29 @@
import { db } from '../db/index.js';
import type { BetaRevision } from '../../beta-revision.js';
export function currentBetaRevision(id: number): { revision: number; started_at: string } {
return db().prepare(`SELECT revision, started_at FROM beta_revisions WHERE beta_id = ? ORDER BY revision DESC LIMIT 1`).get(id) as
{ revision: number; started_at: string } | undefined
?? { revision: 1, started_at: (db().prepare('SELECT opened_at FROM betas WHERE id = ?').get(id) as { opened_at: string }).opened_at };
}
/** Called in the revision transaction, before replacing the pinned bundle. */
export function archiveBetaRevision(id: number, revision: number, startedAt: string, notes: string | null): void {
db().prepare('INSERT OR IGNORE INTO beta_revisions (beta_id, revision, started_at, notes) VALUES (?, ?, ?, ?)')
.run(id, revision, startedAt, notes);
db().prepare(`INSERT OR IGNORE INTO beta_revision_files
(beta_id, revision, script_id, name, kind, path, version_id, version, release_version, sha256, content)
SELECT bs.beta_id, ?, s.id, s.name, s.kind, s.target_rel_path, v.id, v.version, v.release_version, v.sha256, v.content
FROM beta_scripts bs JOIN scripts s ON s.id = bs.script_id JOIN script_versions v ON v.id = bs.script_version_id
WHERE bs.beta_id = ?`).run(revision, id);
}
export function betaRevisionHistory(id: number): BetaRevision[] {
const rows = db().prepare('SELECT revision, started_at, notes, results_json FROM beta_revisions WHERE beta_id = ? ORDER BY revision DESC')
.all(id) as Array<{ revision: number; started_at: string; notes: string | null; results_json: string | null }>;
return rows.map(({ results_json, ...row }) => ({ ...row,
results: results_json ? JSON.parse(results_json) : null,
files: db().prepare(`SELECT script_id, name, kind, path, version_id, version, release_version, sha256
FROM beta_revision_files WHERE beta_id = ? AND revision = ? ORDER BY script_id`).all(id, row.revision) as BetaRevision['files']
}));
}
src/lib/server/fleet/beta-tests.ts
+4 -3
View File
@@ -6,6 +6,7 @@ import { parseBetaPlan, type BetaPlan, type BetaTestView } from '../../beta-plan
import type { GameTelemetry } from '../../telemetry.js';
import { applySample, applyGameEvent, emptyMetrics, evaluateRequirements, type TestMetrics } from './beta-evaluation.js';
import { ONLINE_MAX_MS } from '../../server-freshness.js';
import { currentBetaRevision } from './beta-revisions.js';
export interface TestRun { id: string; beta_id: number; server: string; phase: number; created_at: string; observing_at: string | null; metrics: string }
export interface StoredPlan { beta_id: number; plan: string; outcome: string; reason: string | null }
@@ -33,8 +34,8 @@ export async function prepareRun(betaId: number, server: string, phase = 0): Pro
if (!activeParticipant(betaId, server)) throw new Error('Original test server was deleted; start a new beta');
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) VALUES (?, ?, ?, ?, ?, ?)')
.run(randomUUID(), betaId, server, phase, nowIso(), JSON.stringify(emptyMetrics()));
db().prepare('INSERT INTO beta_test_runs (id, beta_id, server, phase, created_at, metrics, revision) VALUES (?, ?, ?, ?, ?, ?, ?)')
.run(randomUUID(), betaId, server, phase, nowIso(), JSON.stringify(emptyMetrics()), currentBetaRevision(betaId).revision);
})();
return currentRun(betaId, server)!;
}
@@ -97,5 +98,5 @@ export function testView(betaId: number): BetaTestView | null {
}),
evidence: db().prepare(`SELECT e.id, r.server, s.server_deleted_at, e.received_at, e.kind, e.payload FROM beta_test_evidence e
JOIN beta_test_runs r ON r.id = e.run_id JOIN beta_servers s ON s.beta_id = r.beta_id AND s.server = r.server
WHERE r.beta_id = ? ORDER BY e.id DESC LIMIT 200`).all(betaId) as BetaTestView['evidence'] };
WHERE r.beta_id = ? AND r.revision = ? ORDER BY e.id DESC LIMIT 200`).all(betaId, currentBetaRevision(betaId).revision) as BetaTestView['evidence'] };
}
src/lib/server/fleet/beta.ts
+125 -3
View File
@@ -1,5 +1,6 @@
import { parseBetaPlan, type BetaPlan } from '../../beta-plan.js';
import { storedPlan } from './beta-tests.js';
import { storedPlan, testView, activity, setOutcome } from './beta-tests.js';
import { archiveBetaRevision, currentBetaRevision } from './beta-revisions.js';
import type { BetaPromotionPreview } from '../../beta-promotion.js';
const operations = new Set<number>();
@@ -28,17 +29,18 @@ import {
listVersions,
serversMatchingScope,
setScope,
uploadNewVersion,
type ScopeType,
type ScriptRow,
type ScriptVersionRow
} from './scripts.js';
import { composeServerCfg } from './config-compose.js';
import { getServerScript, removeScript } from './inventory.js';
import { cancelStagedFiles, isFileApplied, queueFileChange } from './staged-files.js';
import { cancelStagedFiles, isFileApplied, queueFileChange, stagedFileChanges } from './staged-files.js';
import { emit } from '../pipeline/bus.js';
import { cancelProvisioningJob } from '../provisioning/cancellation.js';
import { reconcileNativeProvisioningReceipt } from '../provisioning/native.js';
import { requestRestart } from '../provisioning/requests.js';
import { requestRestart, releaseRestartRequest } from '../provisioning/requests.js';
import type { BetaStage } from '../pipeline/bus.js';
// Whether an artifact writes a dedicated on-disk file (script/plugin) or
@@ -497,6 +499,124 @@ export async function openBeta(input: OpenBetaInput): Promise<OpenBetaResult> {
};
}
export interface BetaRevisionUpload { scriptId: number; content: Buffer; releaseVersion?: string }
/** Replace existing bundle members; commit files, deadline and evidence reset together. */
export async function reviseBeta(betaId: number, expectedRevision: number, uploads: BetaRevisionUpload[], notes = ''): Promise<{ revision: number }> {
// Drain a runner already observing the previous revision before resetting it.
const { betaTick } = await import('../pipeline/beta-runner.js');
await betaTick();
return exclusive(betaId, async () => {
const beta = getBeta(betaId);
if (!beta) throw new BetaError('Beta not found', 'no-such-beta');
if (!['running', 'ready-promote'].includes(beta.state)) throw new BetaError('Beta is closed', 'invalid-state');
const current = currentBetaRevision(betaId);
if (current.revision !== expectedRevision) throw new BetaError('The beta changed. Reload and review the current revision.', 'preview-changed');
if (!uploads.length || new Set(uploads.map(file => file.scriptId)).size !== uploads.length) {
throw new BetaError('Choose at least one replacement, once per file.', 'invalid-state');
}
if (notes.length > 1000) throw new BetaError('Change notes must be at most 1000 characters.', 'invalid-state');
const bundle = listBetaScripts(betaId);
for (const file of uploads) {
const member = bundle.find(row => row.script_id === file.scriptId);
if (!member) throw new BetaError('Updates can only replace files already in this beta.', 'invalid-state');
if (!file.content.length) throw new BetaError('Choose non-empty replacement files.', 'invalid-state');
if (getVersion(member.script_version_id)!.sha256 === sha256Buffer(file.content)) {
throw new BetaError(`${getScript(file.scriptId)!.name}: the replacement is identical to the current version.`, 'invalid-state');
}
}
const servers = listBetaServers(betaId);
if (!servers.length || servers.some(server => !isActiveBetaServer(betaId, server.server))) {
throw new BetaError('An original beta server is missing. Roll back and open a new beta.', 'invalid-state');
}
ensureHostSchema(db());
assertNativeActivationSettled(betaId);
if (stagedFileChanges('beta', String(betaId)).some(row => row.state === 'applying')) {
throw new BetaError('A file write was interrupted. Resolve it before updating the beta.', 'conflict');
}
const jobs = db().prepare(`SELECT DISTINCT d.id, d.state, d.submitted_at, n.operation_id FROM server_deployments d
JOIN provisioning_sources s ON s.operation_id = d.id LEFT JOIN native_provisioning n ON n.deployment_id = d.id
WHERE s.kind = 'beta' AND s.source_id = ?`).all(String(betaId)) as
Array<{ id: string; state: string; submitted_at: string | null; operation_id: string | null }>;
if (jobs.some(job => ['running', 'awaiting-health'].includes(job.state)
|| (job.state === 'interrupted' && servers.some(server => server.provisioning_id === job.id))
|| (['waiting', 'queued'].includes(job.state) && (job.submitted_at || job.operation_id)))) {
throw new BetaError('Wait for the current activation and recovery to finish before updating this beta.', 'conflict');
}
const stored = storedPlan(betaId);
const plan = stored ? parseBetaPlan(JSON.parse(stored.plan)) : null;
const revision = current.revision + 1;
const at = nowIso();
db().transaction(() => {
archiveBetaRevision(betaId, current.revision, current.started_at, null);
db().prepare('UPDATE beta_revisions SET results_json = ? WHERE beta_id = ? AND revision = ?')
.run(JSON.stringify({ state: beta.state, test: testView(betaId), servers }), betaId, current.revision);
for (const upload of uploads) {
const version = uploadNewVersion(upload.scriptId, upload.content, notes.trim() || undefined, upload.releaseVersion);
db().prepare('UPDATE beta_scripts SET script_version_id = ? WHERE beta_id = ? AND script_id = ?')
.run(version.id, betaId, upload.scriptId);
db().prepare('UPDATE betas SET script_version_id = ? WHERE id = ? AND script_id = ?')
.run(version.id, betaId, upload.scriptId);
}
// Preflight all targets before cancelling work. Errors roll back the
// uploads and history as well as the beta changes.
const changes = new Map(servers.map(server => [server.server, revisionFileChanges(betaId, server.server)]));
// Betas created before staged activation may only have deployed_at
// and snapshots. Record their first writes before replacing staging.
for (const server of servers.filter(server => server.deployed_at)) {
for (const change of changes.get(server.server)!) {
if (isFileApplied('beta', String(betaId), server.server, change.kind, change.relPath) === null) {
db().prepare('INSERT OR IGNORE INTO beta_file_changes (beta_id, server, rel_path) VALUES (?, ?, ?)')
.run(betaId, server.server, change.relPath);
}
}
}
for (const job of jobs.filter(job => ['waiting', 'queued'].includes(job.state))) {
cancelProvisioningJob(job.id, `Superseded by beta #${betaId} revision ${revision}`);
}
for (const server of servers) releaseRestartRequest(server.server, { kind: 'beta', id: String(betaId) });
// New IDs prevent historical host receipts from resolving to a newer
// revision's desired bytes. Original before images stay intact.
db().prepare("DELETE FROM staged_file_changes WHERE source_kind = 'beta' AND source_id = ?").run(String(betaId));
db().prepare('UPDATE beta_test_runs SET retired_at = ? WHERE beta_id = ? AND retired_at IS NULL').run(at, betaId);
db().prepare(`UPDATE beta_servers SET stage = 'queued', error = NULL, deployed_at = NULL,
restarted_at = NULL, healthy_at = NULL, match_seen_at = NULL, updated_at = ? WHERE beta_id = ?`).run(at, betaId);
db().prepare("UPDATE betas SET state = 'running' WHERE id = ?").run(betaId);
setOutcome(betaId, 'active', null);
archiveBetaRevision(betaId, revision, at, notes.trim() || null);
// The original first wave follows selection order; later waves are
// ordered by the runner. Preserve that canary selection on revision.
const firstWave = db().prepare('SELECT server FROM beta_servers WHERE beta_id = ? ORDER BY rowid LIMIT ?')
.all(betaId, plan?.waveSize || servers.length) as { server: string }[];
for (const server of firstWave) {
for (const change of changes.get(server.server)!) queueFileChange(change);
db().prepare("UPDATE beta_servers SET stage = 'awaiting-empty' WHERE beta_id = ? AND server = ?").run(betaId, server.server);
}
activity(betaId, `Revision ${revision} submitted; fresh tests and testing window${notes.trim() ? `: ${notes.trim()}` : ''}`);
})();
emit({ type: 'beta-state', betaId, state: 'running' });
return { revision };
});
}
function revisionFileChanges(betaId: number, name: string): Parameters<typeof queueFileChange>[0][] {
const resolved = listBetaScripts(betaId).map(row => ({ script: getScript(row.script_id)!, version: getVersion(row.script_version_id)! }));
const server = getServer(name)!;
const changes: Parameters<typeof queueFileChange>[0][] = resolved.filter(row => !isCollapsingKind(row.script.kind))
.map(({ script, version }) => ({ sourceKind: 'beta', sourceId: String(betaId), server: name, kind: script.kind,
relPath: script.target_rel_path, content: version.content, scriptVersionId: version.id, inventorySource: `beta:${betaId}` }));
if (resolved.some(row => isCollapsingKind(row.script.kind))) {
const composed = composeCfgForServer(name, server.game, server.map, {
inBetaBaseline: resolved.find(row => row.script.kind === 'baseline')?.version,
inBetaOverlays: resolved.filter(row => row.script.kind === 'config')
});
if (!composed) throw new BetaError(`No baseline available for ${name}.`, 'invalid-state');
changes.push({ sourceKind: 'beta', sourceId: String(betaId), server: name, kind: 'config', relPath: configFileOf(name),
content: Buffer.from(composed.text), inventorySource: `beta:${betaId}` });
}
return changes;
}
/** Apply a reserved wave only when its predecessors have passed. */
export async function deployBetaServers(betaId: number, names: string[]): Promise<void> {
return exclusive(betaId, () => deployBetaServersInternal(betaId, names));
@@ -620,6 +740,8 @@ async function rollbackBetaInternal(betaId: number, options: { restart?: 'manual
const restorations: Array<{ server: string; kind: ArtifactKind; relPath: string; content: Buffer | null; versionId: number | null }> = [];
const changed = db().prepare('SELECT 1 FROM beta_file_changes WHERE beta_id = ? AND server = ? AND rel_path = ?');
const applied = (bs: BetaServerRow, kind: ArtifactKind, path: string) => {
// A queued revision may supersede an applied one before its own activation.
if (changed.get(betaId, bs.server, path)) return true;
const staged = isFileApplied('beta', String(betaId), bs.server, kind, path);
if (staged !== null) return staged;
// Existing betas predating staging wrote their files before deployed_at.
src/lib/server/fleet/staged-files.ts
+16 -9
View File
@@ -80,6 +80,21 @@ function readExisting(path: string): Buffer | null {
catch (err) { if ((err as NodeJS.ErrnoException).code === 'ENOENT') return null; throw err; }
}
/** The first activation owns the rollback image, including an absent file. */
export function captureBetaBeforeImage(row: StagedFileChange, before: Buffer | null): void {
if (db().prepare('SELECT 1 FROM beta_file_changes WHERE beta_id = ? AND server = ? AND rel_path = ?')
.get(row.source_id, row.server, row.rel_path)) return;
const prior = getServerScript(row.server, row.rel_path);
if (before !== null) db().prepare(`INSERT OR IGNORE INTO beta_file_snapshots
(beta_id, server, rel_path, prior_content, prior_sha256, prior_size) VALUES (?, ?, ?, ?, ?, ?)`)
.run(row.source_id, row.server, row.rel_path, before, sha256Buffer(before), before.length);
db().prepare(`UPDATE beta_server_scripts SET had_original = ?, prior_version_id = ? WHERE beta_id = ? AND server = ?
AND script_id IN (SELECT id FROM scripts WHERE target_rel_path = ? AND kind = ?)`)
.run(before === null ? 0 : 1, prior?.script_version_id ?? null, row.source_id, row.server, row.rel_path, row.kind);
db().prepare('INSERT INTO beta_file_changes (beta_id, server, rel_path) VALUES (?, ?, ?)')
.run(row.source_id, row.server, row.rel_path);
}
/** Only called after occupancy checks and reservation of the restart slot.
* Synchronous file replacement keeps cancellation, new requests, and deletion
* from interleaving between the final gates and dispatching the restart.
@@ -111,17 +126,9 @@ export function applyStagedFilesForOperation(operationId: string, server: string
for (const { row, path } of plans) {
// Earlier coalesced campaigns may have just updated the same path.
const before = readExisting(path);
const prior = getServerScript(server, row.rel_path);
db().transaction(() => {
if (row.source_kind === 'beta') {
if (before !== null) db().prepare(`INSERT OR IGNORE INTO beta_file_snapshots
(beta_id, server, rel_path, prior_content, prior_sha256, prior_size) VALUES (?, ?, ?, ?, ?, ?)`)
.run(row.source_id, server, row.rel_path, before, sha256Buffer(before), before.length);
db().prepare(`UPDATE beta_server_scripts SET had_original = ?, prior_version_id = ? WHERE beta_id = ? AND server = ?
AND script_id IN (SELECT id FROM scripts WHERE target_rel_path = ? AND kind = ?)`)
.run(before === null ? 0 : 1, prior?.script_version_id ?? null, row.source_id, server, row.rel_path, row.kind);
db().prepare('INSERT OR IGNORE INTO beta_file_changes (beta_id, server, rel_path) VALUES (?, ?, ?)')
.run(row.source_id, server, row.rel_path);
captureBetaBeforeImage(row, before);
}
if (row.deployment_file_id && !row.preserve_before_image) db().prepare(`UPDATE deployment_files SET
verb = ?, content_before = ?, sha256_before = ? WHERE id = ?`)