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

Defer server file changes until guarded restart activation

Browse Source 
Persist beta, promotion, drift, and rollback payloads until occupancy and restart-capacity checks pass. Preserve rollback snapshots and campaign ordering, and allow held rollback campaigns to resume safely.

Fixes #28

Validation: 805 tests passed, Svelte checks clean, production build passed.
This commit is contained in:
josh
2026-09-27 23:06:16 -04:00
parent eb2a852fdc
commit fb48f64de1
36 changed files with 1634 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/server/fleet/beta.ts
+150 -249
View File
@@ -12,11 +12,11 @@ async function exclusive<T>(id: number, action: () => Promise<T>): Promise<T> {
import { mkdir, readFile, rm, writeFile } from 'node:fs/promises';
import { existsSync } from 'node:fs';
import { dirname, join, posix } from 'node:path';
import { join, posix } from 'node:path';
import { betaArchiveOf, configFileOf, destOfKind, type ArtifactKind } from '../config.js';
import { db } from '../db/index.js';
import { sha256Buffer } from './files.js';
import { ensureDest, getServer, heldByBeta } from './servers.js';
import { getServer, heldByBeta } from './servers.js';
import {
getBaselineForScope,
getScript,
@@ -30,7 +30,8 @@ import {
type ScriptVersionRow
} from './scripts.js';
import { composeServerCfg } from './config-compose.js';
import { recordScript, removeScript, getServerScript } from './inventory.js';
import { getServerScript, removeScript } from './inventory.js';
import { cancelStagedFiles, isFileApplied, queueFileChange } from './staged-files.js';
import { emit } from '../pipeline/bus.js';
import type { BetaStage } from '../pipeline/bus.js';
@@ -207,6 +208,19 @@ function nowIso(): string {
return new Date().toISOString();
}
function unresolvedCampaignWrites(server: string): Array<{ server: string; deploymentId: string; state: string; kind: ArtifactKind; rel_path: string }> {
return db().prepare(`SELECT DISTINCT f.server, f.source_id AS deploymentId, f.state, f.kind, f.rel_path
FROM staged_file_changes f
JOIN deployment_servers ds ON ds.deployment_id = f.source_id AND ds.server = f.server
LEFT JOIN server_deployments operation ON operation.id = ds.provisioning_id
WHERE f.source_kind = 'campaign' AND f.server = ? AND ds.server_deleted_at IS NULL
AND ds.restart_state IN ('pending','failed','held','manual')
AND (f.state = 'applying' OR (f.state = 'pending'
AND (ds.restart_state IN ('held','manual') OR operation.id IS NULL
OR operation.state IN ('queued','waiting','running','awaiting-health'))))`)
.all(server) as Array<{ server: string; deploymentId: string; state: string; kind: ArtifactKind; rel_path: string }>;
}
// ---------- Reads ----------
export function getBeta(id: number): BetaRow | undefined {
@@ -370,6 +384,17 @@ export async function openBeta(input: OpenBetaInput): Promise<OpenBetaResult> {
throw new BetaError('one or more servers are out of scope', 'out-of-scope', outOfScope);
}
// A beta outranks campaigns during activation. Do not let it overtake older
// staged writes which would otherwise overwrite the tested version on closure.
// Legacy campaigns have no staged payload and retain their restart-only priority.
const queuedCampaigns = input.servers.flatMap(unresolvedCampaignWrites);
if (queuedCampaigns.length) {
const interrupted = queuedCampaigns.some(row => row.state === 'applying');
throw new BetaError(interrupted
? 'A campaign file write needs inspection. Resolve or roll back that campaign before opening a beta on this server.'
: 'Finish or resume queued file campaigns before opening a beta on these servers.', 'conflict', queuedCampaigns);
}
// Per-(server, script) conflict check. heldByBeta covers "another running
// beta already claimed this target"; the previous stale-.OLD check is no
// longer relevant now that snapshots live in the DB.
@@ -383,14 +408,6 @@ export async function openBeta(input: OpenBetaInput): Promise<OpenBetaResult> {
}
if (conflicts.length > 0) throw new BetaError('target already held by another beta', 'conflict', conflicts);
// Ensure destinations exist. The `ensureDest` helper covers only the
// script destination today; config/plugin destinations are created
// on-demand via mkdir during the write loop.
for (const name of input.servers) {
const ok = await ensureDest(name, true);
if (!ok.ok) throw new BetaError(`dest not usable for ${name}: ${ok.reason}`, 'dest-fail');
}
// DB writes in one transaction.
const primary = resolved[0]; // populates deprecated betas.* NOT NULL cols
const insertBeta = db().prepare(
@@ -430,13 +447,12 @@ export async function openBeta(input: OpenBetaInput): Promise<OpenBetaResult> {
db().prepare('INSERT INTO beta_test_coverage (beta_id, server, game, map) VALUES (?, ?, ?, ?)').run(betaId, name, selected.game, selected.map);
}
for (const r of resolved) {
const prior = getServerScript(name, r.script.target_rel_path);
insertBetaServerScript.run(
betaId,
name,
r.script.id,
prior ? 1 : 0,
prior?.script_version_id ?? null
0,
null
);
}
}
@@ -484,133 +500,61 @@ async function deployBetaServersInternal(betaId: number, names: string[]): Promi
await writeFile(join(archive, base), r.version.content);
}
const insertSnapshot = db().prepare(
`INSERT INTO beta_file_snapshots (beta_id, server, rel_path, prior_content, prior_sha256, prior_size)
VALUES (?, ?, ?, ?, ?, ?)`
);
// On-disk deploy per server. If any write fails on a server, mark the
// whole server 'failed' — we can't restart a partially deployed server
// safely. Remaining artifacts on that server are skipped.
const updateStage = db().prepare(
`UPDATE beta_servers
SET stage = ?, error = ?, deployed_at = ?, updated_at = ?
WHERE beta_id = ? AND server = ? AND server_deleted_at IS NULL`
);
// Persist the requested bytes without touching any live destination. Provisioning
// captures the original file and inventory immediately before guarded activation.
for (const name of names) {
if (!isActiveBetaServer(betaId, name)) continue;
db().prepare("UPDATE beta_servers SET stage = 'deploying', deployed_at = ? WHERE beta_id = ? AND server = ?")
.run(nowIso(), betaId, name);
const serverRow = serverRows.find((r) => r.name === name)!;
let firstErr: string | null = null;
// 1. Non-collapsing artifacts: script/plugin — write directly to the
// target filename. Pre-beta bytes go into beta_file_snapshots so
// rollback can restore them.
for (const { script, version } of nonCollapsing) {
if (firstErr) break;
const dest = destOfKind(name, script.kind);
const targetPath = join(dest, script.target_rel_path);
try {
await mkdir(join(dest, dirname(script.target_rel_path)), { recursive: true });
if (existsSync(targetPath)) {
const priorBytes = await readFile(targetPath);
if (!isActiveBetaServer(betaId, name)) break;
insertSnapshot.run(
betaId,
name,
script.target_rel_path,
priorBytes,
sha256Buffer(priorBytes),
priorBytes.byteLength
);
db().prepare('UPDATE beta_server_scripts SET had_original = 1 WHERE beta_id = ? AND server = ? AND script_id = ?').run(betaId, name, script.id);
}
db().prepare('INSERT OR IGNORE INTO beta_file_changes (beta_id, server, rel_path) VALUES (?, ?, ?)').run(betaId, name, script.target_rel_path);
if (!isActiveBetaServer(betaId, name)) break;
await writeFile(targetPath, version.content);
if (!isActiveBetaServer(betaId, name)) break;
recordScript(name, script.target_rel_path, version.id, version.sha256, version.size, `beta:${betaId}`);
} catch (err) {
firstErr = `${script.name}: ${(err as Error).message}`;
}
}
// 2. Collapsing layer: compose the effective dedicated_sp.cfg from
// baseline + all applicable overlays, write in place. Pre-beta cfg
// bytes go into beta_file_snapshots keyed on the target rel_path.
const hasCollapsing = collapsingBaseline || collapsingOverlays.length > 0;
if (!firstErr && hasCollapsing) {
try {
try {
const changes: Parameters<typeof queueFileChange>[0][] = nonCollapsing.map(({ script, version }) => ({
sourceKind: 'beta' as const, sourceId: String(betaId), server: name,
kind: script.kind, relPath: script.target_rel_path, content: version.content,
scriptVersionId: version.id, inventorySource: `beta:${betaId}`
}));
if (collapsingBaseline || collapsingOverlays.length > 0) {
const composedResult = composeCfgForServer(name, serverRow.game, serverRow.map, {
inBetaBaseline: collapsingBaseline?.version,
inBetaOverlays: collapsingOverlays
});
if (!composedResult) {
throw new Error(
`no baseline available for ${name} (game=${serverRow.game ?? '?'}) — upload one before running config overlays`
);
}
const dest = destOfKind(name, 'config');
await mkdir(dest, { recursive: true });
const targetRel = configFileOf(name);
const targetPath = join(dest, targetRel);
if (existsSync(targetPath)) {
const priorBytes = await readFile(targetPath);
if (!isActiveBetaServer(betaId, name)) continue;
insertSnapshot.run(
betaId,
name,
targetRel,
priorBytes,
sha256Buffer(priorBytes),
priorBytes.byteLength
);
throw new Error(`no baseline available for ${name} (game=${serverRow.game ?? '?'}) — upload one before running config overlays`);
}
const composedBuf = Buffer.from(composedResult.text, 'utf8');
db().prepare('INSERT OR IGNORE INTO beta_file_changes (beta_id, server, rel_path) VALUES (?, ?, ?)').run(betaId, name, targetRel);
if (!isActiveBetaServer(betaId, name)) continue;
await writeFile(targetPath, composedBuf);
if (!isActiveBetaServer(betaId, name)) continue;
recordScript(
name,
targetRel,
null,
sha256Buffer(composedBuf),
composedBuf.byteLength,
`beta:${betaId}`
);
// Archive the composed preview for operator visibility.
changes.push({ sourceKind: 'beta', sourceId: String(betaId), server: name,
kind: 'config', relPath: configFileOf(name), content: composedBuf,
scriptVersionId: null, inventorySource: `beta:${betaId}` });
await writeFile(join(archive, `${name}.${configFileOf(name)}`), composedBuf);
} catch (err) {
firstErr = `configs: ${(err as Error).message}`;
}
}
const t = nowIso();
if (!isActiveBetaServer(betaId, name)) continue;
if (firstErr) {
updateStage.run('failed', `deploy: ${firstErr}`, t, t, betaId, name);
emit({ type: 'beta-server-stage', betaId, server: name, stage: 'failed', error: `deploy: ${firstErr}` });
} else {
updateStage.run('awaiting-empty', null, t, t, betaId, name);
if (!isActiveBetaServer(betaId, name)) continue;
db().transaction(() => {
for (const change of changes) queueFileChange(change);
db().prepare("UPDATE beta_servers SET stage = 'awaiting-empty', error = NULL, deployed_at = NULL, updated_at = ? WHERE beta_id = ? AND server = ? AND server_deleted_at IS NULL")
.run(nowIso(), betaId, name);
})();
emit({ type: 'beta-server-stage', betaId, server: name, stage: 'awaiting-empty', error: null });
} catch (error) {
const message = `deploy: ${(error as Error).message}`;
db().prepare("UPDATE beta_servers SET stage = 'failed', error = ?, updated_at = ? WHERE beta_id = ? AND server = ? AND server_deleted_at IS NULL")
.run(message, nowIso(), betaId, name);
emit({ type: 'beta-server-stage', betaId, server: name, stage: 'failed', error: message });
}
}
}
// ---------- rollback ----------
export interface RollbackResult {
betaId: number;
deploymentId: string | null;
restored: string[];
dropped: string[];
warnings: string[];
}
export async function rollbackBeta(betaId: number): Promise<RollbackResult> {
return exclusive(betaId, () => rollbackBetaInternal(betaId));
export async function rollbackBeta(betaId: number, options: { restart?: 'manual' | 'when-empty' } = {}): Promise<RollbackResult> {
return exclusive(betaId, () => rollbackBetaInternal(betaId, options));
}
async function rollbackBetaInternal(betaId: number): Promise<RollbackResult> {
async function rollbackBetaInternal(betaId: number, options: { restart?: 'manual' | 'when-empty' }): Promise<RollbackResult> {
const beta = getBeta(betaId);
if (!beta) throw new BetaError(`no such beta (id=${betaId})`, 'no-such-beta');
if (beta.state !== 'running' && beta.state !== 'ready-promote') {
@@ -637,86 +581,77 @@ async function rollbackBetaInternal(betaId: number): Promise<RollbackResult> {
WHERE beta_id = ? AND server = ? AND rel_path = ?`
);
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) => {
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.
return !!changed.get(betaId, bs.server, path) || !!bs.deployed_at;
};
for (const bs of servers) {
if (!isActiveBetaServer(betaId, bs.server)) {
warnings.push(`${bs.server}: original server was deleted; retained history was skipped`);
continue;
}
if (!bs.deployed_at && bs.stage === 'queued') continue;
for (const { script } of bundle) {
if (!isActiveBetaServer(betaId, bs.server)) break;
if (isCollapsingKind(script.kind)) continue; // handled once per server below
if (storedPlan(betaId) && !db().prepare('SELECT 1 FROM beta_file_changes WHERE beta_id = ? AND server = ? AND rel_path = ?').get(betaId, bs.server, script.target_rel_path)) continue;
const dest = destOfKind(bs.server, script.kind);
const targetPath = join(dest, script.target_rel_path);
if (isCollapsingKind(script.kind) || !applied(bs, script.kind, script.target_rel_path)) continue;
const bss = getBetaServerScript(betaId, bs.server, script.id);
if (bss?.had_original) {
const snap = getSnapshot.get(betaId, bs.server, script.target_rel_path) as
| { prior_content: Buffer; prior_sha256: string; prior_size: number }
| undefined;
if (snap) {
await mkdir(join(dest, dirname(script.target_rel_path)), { recursive: true });
if (!isActiveBetaServer(betaId, bs.server)) break;
await writeFile(targetPath, snap.prior_content);
if (!isActiveBetaServer(betaId, bs.server)) break;
if (bss.prior_version_id) {
recordScript(bs.server, script.target_rel_path, bss.prior_version_id, snap.prior_sha256, snap.prior_size, `rollback:${betaId}`);
} else {
recordScript(bs.server, script.target_rel_path, null, snap.prior_sha256, snap.prior_size, `rollback:${betaId}`);
}
restored.add(bs.server);
} else {
warnings.push(`${bs.server}/${script.name}: snapshot missing — cannot restore`);
}
} else {
// No prior file — the beta wrote a fresh file. Drop it.
if (existsSync(targetPath)) await rm(targetPath, { force: true });
if (!isActiveBetaServer(betaId, bs.server)) break;
removeScript(bs.server, script.target_rel_path);
dropped.add(bs.server);
const snap = getSnapshot.get(betaId, bs.server, script.target_rel_path) as
| { prior_content: Buffer; prior_sha256: string; prior_size: number }
| undefined;
if (bss?.had_original && !snap) {
warnings.push(`${bs.server}/${script.name}: snapshot missing — cannot restore`);
continue;
}
restorations.push({ server: bs.server, kind: script.kind, relPath: script.target_rel_path,
content: snap?.prior_content ?? null, versionId: bss?.prior_version_id ?? null });
(snap ? restored : dropped).add(bs.server);
}
// Collapsing config layer: single restore for the shared cfg file.
if (!isActiveBetaServer(betaId, bs.server)) continue;
if (hasCollapsing && (!storedPlan(betaId) || db().prepare('SELECT 1 FROM beta_file_changes WHERE beta_id = ? AND server = ? AND rel_path = ?').get(betaId, bs.server, configFileOf(bs.server)))) {
const dest = destOfKind(bs.server, 'config');
const targetRel = configFileOf(bs.server);
const targetPath = join(dest, targetRel);
const targetRel = configFileOf(bs.server);
if (hasCollapsing && applied(bs, 'config', targetRel)) {
const snap = getSnapshot.get(betaId, bs.server, targetRel) as
| { prior_content: Buffer; prior_sha256: string; prior_size: number }
| undefined;
if (snap) {
await mkdir(dest, { recursive: true });
if (!isActiveBetaServer(betaId, bs.server)) continue;
await writeFile(targetPath, snap.prior_content);
if (!isActiveBetaServer(betaId, bs.server)) continue;
recordScript(bs.server, targetRel, null, snap.prior_sha256, snap.prior_size, `rollback:${betaId}`);
restored.add(bs.server);
} else if (existsSync(targetPath)) {
// No snapshot ⇒ the beta wrote the cfg into a server that had
// no pre-existing file. Remove the beta-written cfg to leave
// the server clean.
await rm(targetPath, { force: true });
if (!isActiveBetaServer(betaId, bs.server)) continue;
removeScript(bs.server, targetRel);
dropped.add(bs.server);
}
restorations.push({ server: bs.server, kind: 'config', relPath: targetRel,
content: snap?.prior_content ?? null, versionId: null });
(snap ? restored : dropped).add(bs.server);
}
}
// Snapshots are single-use — safe to drop now that the disk is restored.
db().prepare('DELETE FROM beta_file_snapshots WHERE beta_id = ?').run(betaId);
const deploymentId = restorations.length ? `dep-beta-rollback-${betaId}-${Date.now()}` : null;
db().transaction(() => {
cancelStagedFiles('beta', String(betaId));
if (deploymentId) {
db().prepare("INSERT INTO deployments (id, action, started_at, triggered_by, beta_id) VALUES (?, 'rollback', ?, ?, ?)")
.run(deploymentId, nowIso(), storedPlan(betaId) ? 'beta-test-plan' : 'ui', betaId);
for (const server of new Set(restorations.map(change => change.server))) {
db().prepare('INSERT INTO deployment_servers (deployment_id, server, files_changed, restart_state) VALUES (?, ?, ?, ?)')
.run(deploymentId, server, restorations.filter(change => change.server === server).length,
options.restart === 'manual' ? 'held' : 'pending');
}
for (const change of restorations) {
const file = db().prepare(`INSERT INTO deployment_files
(deployment_id, server, verb, rel_path, sha256_before, sha256_after, content_before, content_after)
VALUES (?, ?, ?, ?, NULL, ?, NULL, ?)`).run(deploymentId, change.server,
change.content === null ? 'removed' : 'updated', change.relPath,
change.content === null ? null : sha256Buffer(change.content), change.content);
queueFileChange({ sourceKind: 'campaign', sourceId: deploymentId, server: change.server,
kind: change.kind, relPath: change.relPath, content: change.content,
scriptVersionId: change.versionId, inventorySource: `rollback:${betaId}`,
deploymentFileId: Number(file.lastInsertRowid) });
}
}
// The queued restoration owns a durable copy before snapshots are released.
db().prepare('DELETE FROM beta_file_snapshots WHERE beta_id = ?').run(betaId);
db().prepare("UPDATE betas SET state = 'rolled-back', closed_at = ? WHERE id = ?").run(nowIso(), betaId);
})();
if (deploymentId) emit({ type: 'deployment-opened', deploymentId, action: 'rollback', servers: [...new Set(restorations.map(change => change.server))] });
// Wipe archive + close out beta.
const archive = betaArchiveOf(betaId);
if (existsSync(archive)) await rm(archive, { recursive: true, force: true });
db()
.prepare(`UPDATE betas SET state = 'rolled-back', closed_at = ? WHERE id = ?`)
.run(nowIso(), betaId);
// Per-version test-failed transition. Skip a version if (a) another beta
// still gates it at ready-promote/promoted, or (b) it's the currently-
// promoted version of its script (already-live code isn't "failed" by
@@ -742,6 +677,7 @@ async function rollbackBetaInternal(betaId: number): Promise<RollbackResult> {
return {
betaId,
deploymentId,
restored: [...restored].sort(),
dropped: [...dropped].sort(),
warnings
@@ -915,8 +851,7 @@ async function promoteBetaInternal(betaId: number, reviewedToken?: string): Prom
}
}
// Write files. Track per-plan success and per-server first-error.
const planWritten = new Map<Plan, boolean>();
// Build durable activation work; live inventory changes only at activation.
const serverErr = new Map<string, string>();
// Collapsing config layer: one composed cfg per server. Union of beta
@@ -928,6 +863,7 @@ async function promoteBetaInternal(betaId: number, reviewedToken?: string): Prom
composedContent: Buffer;
composedSha: string;
hadFile: boolean;
requiresActivation: boolean;
}
const cfgPlans: CfgPlan[] = [];
if (collapsing.length > 0) {
@@ -951,7 +887,7 @@ async function promoteBetaInternal(betaId: number, reviewedToken?: string): Prom
let priorContent: Buffer | null = null;
let priorSha: string | null = null;
let hadFile = false;
// Beta server: prior cfg bytes were snapshotted at openBeta.
// Beta server: preserve its original bytes captured at activation.
if (betaServerSet.has(name)) {
const snap = getSnapshotForPromote.get(betaId, name, targetRel) as
| { prior_content: Buffer; prior_sha256: string }
@@ -962,59 +898,32 @@ async function promoteBetaInternal(betaId: number, reviewedToken?: string): Prom
hadFile = true;
}
}
if (!hadFile && existsSync(targetPath)) {
// Scope-expansion server (or beta server with no snapshot):
// current on-disk cfg is the "before".
if (!betaServerSet.has(name) && existsSync(targetPath)) {
// Capture an initial preview; activation records the actual before image.
priorContent = await readFile(targetPath);
priorSha = sha256Buffer(priorContent);
hadFile = true;
}
cfgPlans.push({ server: name, priorContent, priorSha, composedContent: composedBuf, composedSha, hadFile });
const requiresActivation = !betaServerSet.has(name) || !existsSync(targetPath)
|| sha256Buffer(await readFile(targetPath)) !== composedSha;
cfgPlans.push({ server: name, priorContent, priorSha, composedContent: composedBuf, composedSha, hadFile, requiresActivation });
}
}
// Recheck after asynchronous snapshot reads, before any mutation. Execution uses
// the captured target lists/content so later membership changes never expand it.
checkReview();
for (const p of plans) {
checkReview();
const dest = destOfKind(p.server, p.script.kind);
try {
await mkdir(join(dest, dirname(p.script.target_rel_path)), { recursive: true });
checkReview();
await writeFile(join(dest, p.script.target_rel_path), p.version.content);
checkReview();
planWritten.set(p, true);
} catch (err) {
if (err instanceof BetaError) throw err;
planWritten.set(p, false);
if (!serverErr.has(p.server)) serverErr.set(p.server, `${p.script.name}: ${(err as Error).message}`);
}
}
// Write composed cfg to each target.
const cfgWritten = new Map<CfgPlan, boolean>();
// A scope expansion from another promotion may have been queued while this
// beta was active. Reassert the tested bytes after that older campaign, even
// when they already match the live file, so it cannot overwrite this release.
const queuedByServer = new Map(reviewed.preview.servers.map(server => [server.name, unresolvedCampaignWrites(server.name)]));
const plansToActivate = new Set(plans.filter(p => !betaServerSet.has(p.server)
|| queuedByServer.get(p.server)?.some(row => row.kind === p.script.kind && row.rel_path === p.script.target_rel_path)));
for (const cp of cfgPlans) {
checkReview();
try {
const dest = destOfKind(cp.server, 'config');
await mkdir(dest, { recursive: true });
checkReview();
const targetRel = configFileOf(cp.server);
const targetPath = join(dest, targetRel);
// Beta server already has composed content at targetPath; overwrite
// idempotently. Scope-expansion server: write in place.
await writeFile(targetPath, cp.composedContent);
checkReview();
cfgWritten.set(cp, true);
} catch (err) {
if (err instanceof BetaError) throw err;
cfgWritten.set(cp, false);
const msg = `configs: ${(err as Error).message}`;
if (!serverErr.has(cp.server)) serverErr.set(cp.server, msg);
console.error(`[promote] ${cp.server}/configs failed:`, err);
}
if (queuedByServer.get(cp.server)?.some(row => row.kind === 'config' && row.rel_path === configFileOf(cp.server))) cp.requiresActivation = true;
}
const activationServers = new Set([...Array.from(plansToActivate, plan => plan.server),
...cfgPlans.filter(plan => plan.requiresActivation).map(plan => plan.server)]);
// Build the aggregate deployment.
const depId = promoteDeploymentId(betaId);
const startedAt = nowIso();
@@ -1049,13 +958,11 @@ async function promoteBetaInternal(betaId: number, reviewedToken?: string): Prom
for (const { script, version } of bundle) {
insertDepScript.run(depId, script.id, version.id, script.current_version_id, script.kind);
}
// deployment_files — one row per successful (script, server) write for
// non-collapsing artifacts, plus one row per (server, dedicated_sp.cfg)
// for the composed config layer.
// Record the reviewed change. Scope expansion gets its actual before
// image when activated; tested servers keep their pre-beta history.
for (const p of plans) {
if (!planWritten.get(p)) continue;
const verb = p.hadFile ? 'updated' : 'added';
insertDepFile.run(
const file = insertDepFile.run(
depId,
p.server,
verb,
@@ -1065,19 +972,16 @@ async function promoteBetaInternal(betaId: number, reviewedToken?: string): Prom
p.priorContent,
p.version.content
);
recordScript(
p.server,
p.script.target_rel_path,
p.version.id,
p.version.sha256,
p.version.size,
`promote:${depId}`
);
if (plansToActivate.has(p)) {
queueFileChange({ sourceKind: 'campaign', sourceId: depId, server: p.server,
kind: p.script.kind, relPath: p.script.target_rel_path, content: p.version.content,
scriptVersionId: p.version.id, inventorySource: `promote:${depId}`,
deploymentFileId: Number(file.lastInsertRowid), preserveBeforeImage: betaServerSet.has(p.server) });
}
}
for (const cp of cfgPlans) {
if (!cfgWritten.get(cp)) continue;
const verb = cp.hadFile ? 'updated' : 'added';
insertDepFile.run(
const file = insertDepFile.run(
depId,
cp.server,
verb,
@@ -1087,14 +991,12 @@ async function promoteBetaInternal(betaId: number, reviewedToken?: string): Prom
cp.priorContent,
cp.composedContent
);
recordScript(
cp.server,
configFileOf(cp.server),
null,
cp.composedSha,
cp.composedContent.byteLength,
`promote:${depId}`
);
if (cp.requiresActivation) {
queueFileChange({ sourceKind: 'campaign', sourceId: depId, server: cp.server,
kind: 'config', relPath: configFileOf(cp.server), content: cp.composedContent,
scriptVersionId: null, inventorySource: `promote:${depId}`,
deploymentFileId: Number(file.lastInsertRowid), preserveBeforeImage: betaServerSet.has(cp.server) });
}
}
// deployment_servers — one row per unique server. Beta servers are
// already restarted; scope-only servers still need the sweep to drain
@@ -1106,9 +1008,9 @@ async function promoteBetaInternal(betaId: number, reviewedToken?: string): Prom
continue;
}
const filesChanged =
plans.filter((p) => p.server === name && planWritten.get(p)).length +
cfgPlans.filter((c) => c.server === name && cfgWritten.get(c)).length;
if (betaServerSet.has(name)) {
plans.filter((p) => p.server === name).length +
cfgPlans.filter((c) => c.server === name).length;
if (!activationServers.has(name)) {
const bsRestartedAt = db()
.prepare(`SELECT restarted_at FROM beta_servers WHERE beta_id = ? AND server = ?`)
.get(betaId, name) as { restarted_at: string | null } | undefined;
@@ -1124,16 +1026,15 @@ async function promoteBetaInternal(betaId: number, reviewedToken?: string): Prom
})();
db().prepare(`UPDATE betas SET state = 'promoted', closed_at = ? WHERE id = ?`).run(nowIso(), betaId);
// Snapshots are single-use — the promoted content is live on disk, so the
// pre-beta bytes are no longer needed.
// The deployment history owns the pre-beta bytes for future rollback.
db().prepare('DELETE FROM beta_file_snapshots WHERE beta_id = ?').run(betaId);
const archive = betaArchiveOf(betaId);
if (existsSync(archive)) await rm(archive, { recursive: true, force: true });
const landed = Array.from(new Set([
...plans.filter((p) => planWritten.get(p)).map((p) => p.server),
...cfgPlans.filter((c) => cfgWritten.get(c)).map((c) => c.server)
...plans.map((p) => p.server),
...cfgPlans.map((c) => c.server)
])).sort();
emit({ type: 'beta-state', betaId, state: 'promoted' });
@@ -1144,7 +1045,7 @@ async function promoteBetaInternal(betaId: number, reviewedToken?: string): Prom
deploymentId: depId,
landed,
scriptIds: bundle.map((b) => b.script.id),
awaitingActivation: reviewed.preview.servers.filter(server => !server.tested && !serverErr.has(server.name)).length,
awaitingActivation: [...activationServers].filter(name => !serverErr.has(name)).length,
failedServers: [...serverErr.keys()]
};
}
src/lib/server/fleet/drift.ts
+26 -29
View File
@@ -1,21 +1,18 @@
import { mkdir, readdir, readFile, rm, stat, writeFile } from 'node:fs/promises';
import { readdir, readFile, stat } from 'node:fs/promises';
import { randomUUID } from 'node:crypto';
import type { DriftPlan, DriftPlanFile } from '$lib/drift-plan.js';
import { dirname, join, posix, relative, sep } from 'node:path';
import { join, posix, relative, sep } from 'node:path';
import { configFileOf, destOfKind, serverDir, type ArtifactKind } from '../config.js';
import { db } from '../db/index.js';
import { emit } from '../pipeline/bus.js';
import { sha256Buffer } from './files.js';
import {
getServerScript,
recordScript,
removeScript
} from './inventory.js';
import { getServerScript } from './inventory.js';
import { getBaselineForScope, getVersion } from './scripts.js';
import { getPromotedServerConfig, getServerMapConfig, MAP_CONFIG_FILE } from './server-config.js';
import { selectedZombieMap } from './map-aliases.js';
import { getServer, listServers } from './servers.js';
import { promotedArtifacts } from './promoted-artifacts.js';
import { queueFileChange } from './staged-files.js';
// Destinations drift scans, in a stable order so the UI shows scripts first.
// Baseline shares the config destination — one physical dir, two logical kinds.
@@ -309,8 +306,8 @@ export async function previewDrift(servers?: string[]): Promise<DriftPlan> {
export interface RunDriftResult {
deploymentId: string | null;
// `removed` = untracked files deleted from disk (drift kind='untracked')
// `restored` = missing or existing artifacts installed and tracked
// `removed` and `restored` describe queued changes. Provisioning applies
// them only after the server passes its activation gates.
perServer: Array<{ server: string; removed: string[]; restored: string[]; skipped: string[] }>;
totalRemoved: number;
totalRestored: number;
@@ -381,9 +378,8 @@ export function createDriftRebuildCampaign(
/**
* Detect drift on the requested servers (or the whole fleet) and either
* report the plan (dryRun) or remove untracked files and install managed
* artifacts as a single deployment. The sweep restarts each affected server once
* it's empty and waits for it to come back healthy.
* report the plan (dryRun) or stage the changes as a single deployment.
* Provisioning applies them and restarts each affected server once it's empty.
*/
export async function runDriftDeployment(input: RunDriftInput = {}): Promise<RunDriftResult> {
const { dryRun = false, triggeredBy = 'ui' } = input;
@@ -471,15 +467,19 @@ export async function runDriftDeployment(input: RunDriftInput = {}): Promise<Run
}
}
if (entry.driftKind === 'untracked') {
// Delete a file inventory doesn't know about.
// Capture the reviewed deletion without touching the live tree.
const content = await readFile(abs);
if (hasActiveBeta(s.server)) throw new Error('active beta owns this server; reconciliation deferred');
await rm(abs, { force: true });
removeScript(s.server, entry.relPath);
db().transaction(() => {
const file = insertFile.run(depId, s.server, 'removed', entry.relPath, sha256Buffer(content), null, content, null);
queueFileChange({ sourceKind: 'campaign', sourceId: depId, server: s.server,
kind: entry.kind, relPath: entry.relPath, content: null,
inventorySource: `drift:${depId}`, deploymentFileId: Number(file.lastInsertRowid),
expectedBeforeSha: reviewed?.beforeSha256 ?? sha256Buffer(content) });
})();
s.removed.push(entry.relPath);
insertFile.run(depId, s.server, 'removed', entry.relPath, entry.sha256, null, content, null);
} else {
// Install the managed artifact without deleting the target first.
// Persist exact replacement bytes so waiting cannot change the plan.
const written = await resolveManagedEntry(s.server, entry);
if (!written) {
s.skipped.push(`${entry.relPath}: missing artifact could not be resolved from the DB`);
@@ -493,20 +493,17 @@ export async function runDriftDeployment(input: RunDriftInput = {}): Promise<Run
if (err.code === 'ENOENT') return null;
throw err;
});
await mkdir(dirname(abs), { recursive: true });
if (hasActiveBeta(s.server)) throw new Error('active beta owns this server; reconciliation deferred');
await writeFile(abs, written);
recordScript(
s.server,
entry.relPath,
entry.kind === 'config' ? null : entry.scriptVersionId ?? null,
sha256Buffer(written),
written.byteLength,
`drift:${depId}`
);
db().transaction(() => {
const file = insertFile.run(depId, s.server, before ? 'updated' : 'added', entry.relPath,
before ? sha256Buffer(before) : null, sha256Buffer(written), before, written);
queueFileChange({ sourceKind: 'campaign', sourceId: depId, server: s.server,
kind: entry.kind, relPath: entry.relPath, content: written,
scriptVersionId: entry.kind === 'config' ? null : entry.scriptVersionId ?? null,
inventorySource: `drift:${depId}`, deploymentFileId: Number(file.lastInsertRowid),
expectedBeforeSha: reviewed ? reviewed.beforeSha256 : before ? sha256Buffer(before) : null });
})();
s.restored.push(entry.relPath);
insertFile.run(depId, s.server, before ? 'updated' : 'added', entry.relPath,
before ? sha256Buffer(before) : null, sha256Buffer(written), before, written);
}
} catch (err) {
s.skipped.push(`${entry.relPath}: ${(err as Error).message}`);
src/lib/server/fleet/rollback.ts
+29 -50
View File
@@ -1,11 +1,8 @@
import { mkdir, rm, writeFile } from 'node:fs/promises';
import { existsSync } from 'node:fs';
import { dirname, join } from 'node:path';
import { destOfKind, type ArtifactKind } from '../config.js';
import type { ArtifactKind } from '../config.js';
import { db } from '../db/index.js';
import { emit } from '../pipeline/bus.js';
import { recordScript, removeScript } from './inventory.js';
import { getVersion } from './scripts.js';
import { cancelStagedFiles, isFileApplied, queueFileChange, stagedFileChanges } from './staged-files.js';
// Roll a promoted deployment back to whatever versions were current before
// the promote landed. Reads deployment_scripts as the source of truth for
@@ -142,58 +139,27 @@ export async function rollbackPromoteDeployment(promoteId: string): Promise<Roll
// Group files by server so each server gets exactly one deployment_servers row.
const perServer = new Map<string, FileRow[]>();
const staged = stagedFileChanges('campaign', promoteId);
// Composed baseline rows are physically staged as config files. Prefer the
// explicit link instead of inferring their destination from the library.
const kindForFile = (file: FileRow) => staged.find(row => row.deployment_file_id === file.id)?.kind ?? kindForRelPath(file.rel_path);
for (const f of fileRows) {
// A promotion can still be waiting for empty servers. Withdrawing it
// must not restore or remove files that it never installed. Rows from
// older releases have no staging record and retain their audit meaning.
const kind = kindForFile(f);
const preservesBetaSnapshot = staged.some(row => row.server === f.server && row.kind === kind
&& row.rel_path === f.rel_path && row.preserve_before_image);
if (!preservesBetaSnapshot && isFileApplied('campaign', promoteId, f.server, kind, f.rel_path) === false) continue;
const arr = perServer.get(f.server);
if (arr) arr.push(f);
else perServer.set(f.server, [f]);
}
const affectedServers: string[] = [];
const perServerFailures = new Map<string, string>();
const active = (server: string) => !!db().prepare(`SELECT 1 FROM deployment_servers ds JOIN servers s ON s.name = ds.server
WHERE ds.deployment_id = ? AND ds.server = ? AND ds.server_deleted_at IS NULL AND s.destroyed_at IS NULL`).get(promoteId, server);
for (const [server, files] of perServer) {
if (!active(server)) continue;
try {
for (const f of files) {
if (!active(server)) break;
const kind = kindForRelPath(f.rel_path);
const dest = destOfKind(server, kind);
const abs = join(dest, f.rel_path);
await mkdir(join(dest, dirname(f.rel_path)), { recursive: true });
if (!active(server)) break;
const owningScript = scriptByTarget.get(f.rel_path);
if (f.verb === 'added') {
if (existsSync(abs)) await rm(abs, { force: true });
if (!active(server)) break;
removeScript(server, f.rel_path);
} else if (f.verb === 'updated' && f.content_before) {
await writeFile(abs, f.content_before);
if (!active(server)) break;
const priorVer = owningScript?.prior_version_id
? getVersion(owningScript.prior_version_id)
: null;
recordScript(
server,
f.rel_path,
priorVer?.id ?? null,
f.sha256_before!,
f.content_before.byteLength,
`rollback:${newDepId}`
);
} else if (f.verb === 'removed' && f.content_before) {
await writeFile(abs, f.content_before);
if (!active(server)) break;
recordScript(server, f.rel_path, null, f.sha256_before!, f.content_before.byteLength, `rollback:${newDepId}`);
}
}
if (active(server)) affectedServers.push(server);
} catch (err) {
perServerFailures.set(server, (err as Error).message);
}
}
// Write the new rollback deployment + mark the original.
const insertDep = db().prepare(
`INSERT INTO deployments
@@ -216,6 +182,11 @@ export async function rollbackPromoteDeployment(promoteId: string): Promise<Roll
);
db().transaction(() => {
cancelStagedFiles('campaign', promoteId);
// Withdrawing this source leaves a shared restart available to other
// campaigns. Provisioning cancels an operation with no eligible sources.
db().prepare(`UPDATE deployment_servers SET restart_state = 'skipped', error = NULL
WHERE deployment_id = ? AND server_deleted_at IS NULL AND restart_state IN ('pending','failed','held','manual')`).run(promoteId);
insertDep.run(newDepId, startedAt);
// Mirror the promote's script list — swapped: the "current" version
// after rollback is the promote's prior; the "prior" of the rollback
@@ -231,8 +202,8 @@ export async function rollbackPromoteDeployment(promoteId: string): Promise<Roll
}
for (const [server, files] of perServer) {
if (!active(server)) continue;
if (perServerFailures.has(server)) {
insertDepServer.run(newDepId, server, 0, 'failed', perServerFailures.get(server)!);
if (files.some(file => file.verb !== 'added' && file.content_before === null)) {
insertDepServer.run(newDepId, server, 0, 'failed', 'Original file snapshot is missing; cannot restore safely');
continue;
}
insertDepServer.run(newDepId, server, files.length, 'pending', null);
@@ -241,7 +212,7 @@ export async function rollbackPromoteDeployment(promoteId: string): Promise<Roll
f.verb === 'added' ? 'removed'
: f.verb === 'removed' ? 'added'
: 'updated';
insertDepFile.run(
const inserted = insertDepFile.run(
newDepId,
server,
invertedVerb,
@@ -251,7 +222,15 @@ export async function rollbackPromoteDeployment(promoteId: string): Promise<Roll
f.content_after,
f.content_before
);
const owningScript = scriptByTarget.get(f.rel_path);
const priorVersion = owningScript?.prior_version_id ? getVersion(owningScript.prior_version_id) : null;
queueFileChange({ sourceKind: 'campaign', sourceId: newDepId, server,
kind: kindForFile(f), relPath: f.rel_path,
content: f.verb === 'added' ? null : f.content_before,
scriptVersionId: f.verb === 'updated' ? priorVersion?.id ?? null : null,
inventorySource: `rollback:${newDepId}`, deploymentFileId: Number(inserted.lastInsertRowid) });
}
affectedServers.push(server);
}
db().prepare(`UPDATE deployments SET rolled_back_by = ? WHERE id = ?`).run(newDepId, promoteId);
// Restore each affected script's current_version_id.
src/lib/server/fleet/staged-files.ts
+155
View File
@@ -0,0 +1,155 @@
import { mkdirSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs';
import { dirname, relative, resolve, isAbsolute } from 'node:path';
import { destOfKind, serverDir, type ArtifactKind } from '../config.js';
import { db } from '../db/index.js';
import { sha256Buffer } from './files.js';
import { getServerScript, recordScript, removeScript } from './inventory.js';
type SourceKind = 'campaign' | 'beta';
interface FileChange {
sourceKind: SourceKind;
sourceId: string;
server: string;
kind: ArtifactKind;
relPath: string;
content: Buffer | null;
scriptVersionId?: number | null;
inventorySource: string;
deploymentFileId?: number;
preserveBeforeImage?: boolean;
expectedBeforeSha?: string | null;
}
export interface StagedFileChange {
id: number; source_kind: SourceKind; source_id: string; server: string;
kind: ArtifactKind; rel_path: string; content: Buffer | null;
script_version_id: number | null; inventory_source: string;
deployment_file_id: number | null; preserve_before_image: number;
check_before: number; expected_before_sha: string | null;
state: 'pending' | 'applying' | 'applied' | 'cancelled'; applied_at: string | null;
}
function targetPath(server: string, kind: ArtifactKind, relPath: string): string {
const root = resolve(destOfKind(server, kind));
const target = resolve(root, relPath);
const rel = relative(root, target);
if (!rel || rel === '..' || rel.startsWith('../') || isAbsolute(rel)) throw new Error('File target must be inside its artifact directory');
return target;
}
/** Capture immutable desired bytes without changing the server or its inventory. */
export function queueFileChange(change: FileChange): void {
targetPath(change.server, change.kind, change.relPath);
db().prepare(`INSERT INTO staged_file_changes (source_kind, source_id, server, kind, rel_path, content,
script_version_id, inventory_source, deployment_file_id, preserve_before_image, check_before, expected_before_sha)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(source_kind, source_id, server, kind, rel_path) DO UPDATE SET
content = excluded.content, script_version_id = excluded.script_version_id,
inventory_source = excluded.inventory_source, deployment_file_id = excluded.deployment_file_id,
preserve_before_image = excluded.preserve_before_image, check_before = excluded.check_before,
expected_before_sha = excluded.expected_before_sha
WHERE staged_file_changes.state = 'pending'`)
.run(change.sourceKind, change.sourceId, change.server, change.kind, change.relPath, change.content,
change.scriptVersionId ?? null, change.inventorySource, change.deploymentFileId ?? null,
change.preserveBeforeImage ? 1 : 0, change.expectedBeforeSha !== undefined ? 1 : 0, change.expectedBeforeSha ?? null);
}
export function stagedFileChanges(sourceKind: SourceKind, sourceId: string, server?: string): StagedFileChange[] {
return db().prepare(`SELECT * FROM staged_file_changes WHERE source_kind = ? AND source_id = ?
${server === undefined ? '' : 'AND server = ?'} ORDER BY id`)
.all(sourceKind, sourceId, ...(server === undefined ? [] : [server])) as StagedFileChange[];
}
export function cancelStagedFiles(sourceKind: SourceKind, sourceId: string, server?: string): void {
db().prepare(`UPDATE staged_file_changes SET state = 'cancelled' WHERE source_kind = ? AND source_id = ?
AND state = 'pending' ${server === undefined ? '' : 'AND server = ?'}`)
.run(sourceKind, sourceId, ...(server === undefined ? [] : [server]));
}
export function hasAppliedFiles(sourceKind: SourceKind, sourceId: string, server?: string): boolean {
return stagedFileChanges(sourceKind, sourceId, server).some(row => row.state === 'applied' || row.state === 'applying');
}
/** null identifies legacy work that wrote files before staging was introduced. */
export function isFileApplied(sourceKind: SourceKind, sourceId: string, server: string, kind: ArtifactKind, relPath: string): boolean | null {
const row = stagedFileChanges(sourceKind, sourceId, server).find(row => row.kind === kind && row.rel_path === relPath);
return row ? row.state === 'applied' || row.state === 'applying' : null;
}
function readExisting(path: string): Buffer | null {
try { return readFileSync(path); }
catch (err) { if ((err as NodeJS.ErrnoException).code === 'ENOENT') return null; throw err; }
}
/** 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.
*/
export function applyStagedFilesForOperation(operationId: string, server: string): void {
const rows = db().prepare(`SELECT f.* FROM staged_file_changes f JOIN provisioning_sources s
ON s.kind = f.source_kind AND s.source_id = f.source_id AND s.server = f.server
WHERE s.operation_id = ? AND f.server = ? AND f.state IN ('pending','applying') ORDER BY f.id`)
.all(operationId, server) as StagedFileChange[];
if (!rows.length) return;
if (!statSync(serverDir(server)).isDirectory()) throw new Error('Server directory is unavailable');
// Preflight every target before changing any files. An interrupted write
// needs inspection/rollback, never an automatic replay.
const plans = rows.map(row => {
if (row.state === 'applying') throw new Error('A previous file write was interrupted; inspect or roll back the campaign');
const source = row.source_kind === 'campaign'
? db().prepare(`SELECT 1 FROM deployment_servers WHERE deployment_id = ? AND server = ?
AND server_deleted_at IS NULL AND provisioning_id = ? AND restart_state IN ('pending','failed')`).get(row.source_id, server, operationId)
: db().prepare(`SELECT 1 FROM beta_servers bs JOIN betas b ON b.id = bs.beta_id WHERE bs.beta_id = ? AND bs.server = ?
AND bs.server_deleted_at IS NULL AND bs.provisioning_id = ? AND b.state IN ('running','ready-promote')`).get(row.source_id, server, operationId);
if (!source) throw new Error('File change requester is no longer eligible');
const path = targetPath(server, row.kind, row.rel_path);
const before = readExisting(path);
if (row.check_before && (before === null ? null : sha256Buffer(before)) !== row.expected_before_sha) {
throw new Error(`${row.rel_path}: changed while queued; scan and review again`);
}
return { row, path, before };
});
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);
}
if (row.deployment_file_id && !row.preserve_before_image) db().prepare(`UPDATE deployment_files SET
verb = ?, content_before = ?, sha256_before = ? WHERE id = ?`)
.run(row.content === null ? 'removed' : before === null ? 'added' : 'updated', before,
before === null ? null : sha256Buffer(before), row.deployment_file_id);
db().prepare("UPDATE staged_file_changes SET state = 'applying' WHERE id = ?").run(row.id);
})();
if (row.content === null) {
rmSync(path, { force: true });
removeScript(server, row.rel_path);
} else {
mkdirSync(dirname(path), { recursive: true });
// Preserve existing ownership, permissions, and symlink behavior.
// The durable applying checkpoint prevents replay after a partial write.
writeFileSync(path, row.content);
// A queued version can be deleted independently; the staged bytes
// still apply, but its optional inventory link must remain valid.
const version = row.script_version_id !== null && db().prepare('SELECT 1 FROM script_versions WHERE id = ?').get(row.script_version_id)
? row.script_version_id : null;
recordScript(server, row.rel_path, version, sha256Buffer(row.content), row.content.length, row.inventory_source);
}
db().prepare("UPDATE staged_file_changes SET state = 'applied', applied_at = ? WHERE id = ?")
.run(new Date().toISOString(), row.id);
if (row.source_kind === 'campaign') db().prepare('INSERT INTO campaign_events (deployment_id, message) VALUES (?, ?)')
.run(row.source_id, `${server}: applied ${row.rel_path}`);
}
for (const betaId of new Set(rows.filter(row => row.source_kind === 'beta').map(row => row.source_id))) {
db().prepare('UPDATE beta_servers SET deployed_at = ?, updated_at = ? WHERE beta_id = ? AND server = ? AND server_deleted_at IS NULL')
.run(new Date().toISOString(), new Date().toISOString(), betaId, server);
}
}