Add configurable history and storage retention policies

This commit is contained in:
2026-10-05 22:48:04 -04:00
parent 7c98df4fb0
commit 3952c4c938
37 changed files with 963 additions and 82 deletions
+1
View File
@@ -29,6 +29,7 @@ export function routePermission(routeId: string | null, method: string): RoutePe
|| within(routeId, '/api/server-keys') || within(routeId, '/server-keys') || within(routeId, '/keys')) return 'admin';
if (!reading && ['/api/hosts', '/api/settings', '/api/game-files'].some(base => within(routeId, base))) return 'admin';
if (within(routeId, '/api/settings/retention')) return 'admin';
if (routeId === '/api/provisioning/[id]/override') return 'admin';
// These endpoints mix operational actions with global policy and schedule
// changes. Per-incident/job actions remain available to operators.
+10
View File
@@ -730,3 +730,13 @@ CREATE TABLE IF NOT EXISTS watchdog_actions (
cycle INTEGER NOT NULL DEFAULT 1,
UNIQUE(incident_id,cycle,action)
);
-- Explicit history holds and failure resolution never bypass dependency checks.
CREATE TABLE IF NOT EXISTS retention_controls (
category TEXT NOT NULL, record_id TEXT NOT NULL, protected INTEGER NOT NULL DEFAULT 0,
resolved_at TEXT, PRIMARY KEY(category,record_id)
);
CREATE TABLE IF NOT EXISTS retention_expirations (
category TEXT NOT NULL, record_id TEXT NOT NULL, expired_at TEXT NOT NULL,
PRIMARY KEY(category,record_id)
);
+3 -3
View File
@@ -1,7 +1,7 @@
import { db } from '../db/index.js';
import { enrichGameChatTimeline, type ChatMatch, type ChatRoundReport, type ContextualChatEntry } from '../../game-chat.js';
import { parseGameEvent } from '../../telemetry.js';
import { TELEMETRY_REQUEST_RETENTION } from '../../telemetry-diagnostics.js';
import { telemetryRetention, telemetryCutoff } from './telemetry-requests.js';
import type { ServerRow } from './servers.js';
import type { TailResult } from './logs.js';
@@ -15,7 +15,7 @@ export function serverChatTimeline(server: ServerRow, tail: TailResult): Context
// Diagnostic history survives deletion. Its receipt time must belong to the
// current registration so a reused server name cannot inherit old rounds.
const cutoff = new Date(Date.now() - TELEMETRY_REQUEST_RETENTION.maxAgeDays * 86_400_000).toISOString();
const cutoff = telemetryCutoff();
const reports = handle.prepare(`SELECT received_at AS receivedAt, event, round,
player_count AS playerCount FROM telemetry_requests
WHERE server = ? AND status = 200 AND received_at >= ?
@@ -23,7 +23,7 @@ export function serverChatTimeline(server: ServerRow, tail: TailResult): Context
AND event IN ('game_start', 'round_change', 'heartbeat', 'game_end')
AND player_count >= 0 AND (event = 'game_start' OR round >= 0)
ORDER BY id DESC LIMIT ?`).all(server.name, cutoff, server.created_at,
TELEMETRY_REQUEST_RETENTION.maxPerServer) as ChatRoundReport[];
(telemetryRetention().maxPerServer || 10000)) as ChatRoundReport[];
// Saving gameplay does not depend on diagnostic logging succeeding. Keep the
// surviving accepted snapshot useful when its request log was not recorded.
+3 -1
View File
@@ -1,3 +1,4 @@
import { fleetSettings } from '../settings.js';
import { db } from '../db/index.js';
import { ONLINE_MAX_MS } from '../../server-freshness.js';
import { parseMetricsQuery, type HeatmapCell, type MapMetrics, type MetricPoint, type MetricsData,
@@ -18,7 +19,8 @@ export function recordMetrics(samples: MetricSample[], at = Date.now()): void {
handle.transaction(() => {
const insert = handle.prepare('INSERT OR REPLACE INTO metric_samples (at, server, game, map, players, capacity) VALUES (?, ?, ?, ?, ?, ?)');
for (const s of samples) insert.run(at, s.server, s.game, s.map, s.players, s.capacity);
handle.prepare('DELETE FROM metric_samples WHERE at < ?').run(at - 31 * 86400_000);
const days = fleetSettings().metricsKeepDays;
if (days) handle.prepare('DELETE FROM metric_samples WHERE at < ?').run(at - days * 86400_000);
})();
}
+4 -1
View File
@@ -11,7 +11,7 @@ import { cancelStagedFiles, isFileApplied, queueFileChange, stagedFileChanges }
// pre-multi-script legacy rows.
export class RollbackError extends Error {
constructor(message: string, public readonly kind: 'no-such-deployment' | 'not-promote' | 'already-rolled-back') {
constructor(message: string, public readonly kind: 'history-expired' | 'no-such-deployment' | 'not-promote' | 'already-rolled-back') {
super(message);
}
}
@@ -72,6 +72,9 @@ export async function rollbackPromoteDeployment(promoteId: string): Promise<Roll
FROM deployments WHERE id = ?`)
.get(promoteId) as DeploymentRow | undefined;
if (!dep) throw new RollbackError(`no such deployment ${promoteId}`, 'no-such-deployment');
if (db().prepare("SELECT 1 FROM retention_expirations WHERE category='file-content' AND record_id=?").get(promoteId)) {
throw new RollbackError('Rollback file contents have expired under the retention policy.', 'history-expired');
}
if (dep.action !== 'promote') {
throw new RollbackError(`deployment ${promoteId} is action='${dep.action}', not 'promote'`, 'not-promote');
}
+14 -9
View File
@@ -1,10 +1,15 @@
import { fleetSettings } from '../settings.js';
import { db } from '../db/index.js';
import { TELEMETRY_REQUEST_RETENTION, type TelemetryRequestLogEntry } from '../../telemetry-diagnostics.js';
import { type TelemetryRequestLogEntry } from '../../telemetry-diagnostics.js';
const MAX_PAYLOAD_LENGTH = 16 * 1024;
// Bound storage even when requests claim many different, unregistered names.
const MAX_REQUESTS = 10_000;
const cutoff = () => new Date(Date.now() - TELEMETRY_REQUEST_RETENTION.maxAgeDays * 86_400_000).toISOString();
export function telemetryRetention() {
const settings = fleetSettings();
return { maxAgeDays: settings.telemetryKeepDays, maxPerServer: settings.telemetryMaxPerServer, maxTotal: settings.telemetryMaxTotal };
}
export const telemetryCutoff = () => telemetryRetention().maxAgeDays ? new Date(Date.now() - telemetryRetention().maxAgeDays * 86_400_000).toISOString() : '';
const cutoff = telemetryCutoff;
const textField = (value: unknown) => typeof value === 'string' ? value.slice(0, 255) : null;
export function recordTelemetryRequest(input: {
@@ -31,13 +36,13 @@ export function recordTelemetryRequest(input: {
Array.isArray(body.players) ? body.players.length : null,
textField(input.contentType), payload.slice(0, MAX_PAYLOAD_LENGTH), Number(payload.length > MAX_PAYLOAD_LENGTH)
);
handle.prepare('DELETE FROM telemetry_requests WHERE received_at < ?').run(cutoff());
handle.prepare(`DELETE FROM telemetry_requests WHERE server = ? AND id <= (
if (telemetryRetention().maxAgeDays) handle.prepare('DELETE FROM telemetry_requests WHERE received_at < ?').run(cutoff());
if (telemetryRetention().maxPerServer) handle.prepare(`DELETE FROM telemetry_requests WHERE server = ? AND id <= (
SELECT id FROM telemetry_requests WHERE server = ? ORDER BY id DESC LIMIT 1 OFFSET ?
)`).run(body.server, body.server, TELEMETRY_REQUEST_RETENTION.maxPerServer);
handle.prepare(`DELETE FROM telemetry_requests WHERE id <= (
)`).run(body.server, body.server, telemetryRetention().maxPerServer);
if (telemetryRetention().maxTotal) handle.prepare(`DELETE FROM telemetry_requests WHERE id <= (
SELECT id FROM telemetry_requests ORDER BY id DESC LIMIT 1 OFFSET ?
)`).run(MAX_REQUESTS);
)`).run(telemetryRetention().maxTotal);
})();
}
@@ -45,6 +50,6 @@ export function listTelemetryRequests(server: string): TelemetryRequestLogEntry[
const rows = db().prepare(`SELECT id, server, received_at AS receivedAt, status, error, event, map, round,
player_count AS playerCount, content_type AS contentType, payload, payload_truncated AS payloadTruncated
FROM telemetry_requests WHERE server = ? AND received_at >= ? ORDER BY id DESC LIMIT ?`)
.all(server, cutoff(), TELEMETRY_REQUEST_RETENTION.maxPerServer) as Array<Omit<TelemetryRequestLogEntry, 'payloadTruncated'> & { payloadTruncated: number }>;
.all(server, cutoff(), telemetryRetention().maxPerServer || 10000) as Array<Omit<TelemetryRequestLogEntry, 'payloadTruncated'> & { payloadTruncated: number }>;
return rows.map(row => ({ ...row, payloadTruncated: !!row.payloadTruncated }));
}
+75 -3
View File
@@ -1,3 +1,4 @@
import { fleetSettings } from '../settings.js';
import { randomUUID } from 'node:crypto';
import { mkdir, rename, stat, rm, lstat } from 'node:fs/promises';
import { dirname, join, resolve } from 'node:path';
@@ -7,7 +8,7 @@ import { getHost, getHostOperation, listHosts, queueHostOperation } from '../hos
import { ensureGameFileSchema } from './schema.js';
import { ensureHostSchema } from '../hosts/schema.js';
import { manifestChecksum, receiveGameFile, validateGamePath, validateManifest, verifyGameFile } from './storage.js';
import { MAX_UPLOAD_CHUNK_BYTES, validateUploadPlan, getUploadFileProgress, receiveUploadChunk, hashUploadedFiles, withUploadLock } from './uploads.js';
import { MAX_UPLOAD_CHUNK_BYTES, validateUploadPlan, getUploadFileProgress, receiveUploadChunk, hashUploadedFiles, withUploadLock, uploadIsActive } from './uploads.js';
import { GAME_FILE_GAMES } from '../../game-files.js';
import type { GameFileEntry, GameLibrary, GameLibraryGame, GameLibraryVersion, HostLibraryCache, LibraryCacheState, LibraryTransferProgress } from '../../game-files.js';
@@ -137,6 +138,7 @@ export function createLibraryUpload(input: Record<string, unknown>): GameLibrary
type UploadEntry = { path: string; size: number; lastModified: number };
function requireBrowserUpload(id: string): GameLibraryVersion {
const version = getLibraryVersion(id);
if (database().prepare('SELECT 1 FROM game_file_cleanup WHERE version_id=?').get(id)) throw Object.assign(new Error('Upload files expired under the retention policy.'), { status: 410 });
if (!version || version.sourceHostId !== null || !database().prepare('SELECT 1 FROM game_file_upload_entries WHERE version_id=? LIMIT 1').get(id)) {
throw Object.assign(new Error('Folder upload not found.'), { status: 404 });
}
@@ -153,6 +155,7 @@ async function uploadStorageRoot(version: GameLibraryVersion): Promise<string> {
}
export async function browserUploadStatus(id: string) {
const version = requireBrowserUpload(id);
touchUpload(id);
const status = version.state === 'ready' ? 'published' as const : publishing.has(id) ? 'verifying' as const : version.state === 'failed' ? 'failed' as const : 'pending' as const;
const root = await uploadStorageRoot(version);
const entries = uploadEntries(id);
@@ -169,15 +172,18 @@ export async function uploadBrowserFile(id: string, path: string, chunk: { offse
if (version.state === 'ready' || version.manifestSha256 || publishing.has(id)) throw Object.assign(new Error('This upload is no longer writable.'), { status: 409 });
};
checkWritable();
touchUpload(id);
const file = database().prepare('SELECT path,size,last_modified AS lastModified FROM game_file_upload_entries WHERE version_id=? AND path=?').get(id, path) as UploadEntry | undefined;
if (!file) throw new Error('File is not listed in this upload.');
const progress = await receiveUploadChunk(stagingRoot(id), file, chunk, body, checkWritable);
database().prepare("UPDATE game_file_versions SET state='importing',error=NULL WHERE id=? AND state IN ('importing','failed')").run(id);
touchUpload(id);
return progress;
}
export async function publishBrowserUpload(id: string): Promise<GameLibraryVersion> {
const version = requireBrowserUpload(id);
if (version.state === 'ready') return version;
touchUpload(id);
const existing = publishing.get(id); if (existing) return existing;
const task = withUploadLock(stagingRoot(id), async () => {
const root = await uploadStorageRoot(version);
@@ -241,6 +247,7 @@ export function createLibraryImport(input: Record<string, unknown>): GameLibrary
export function requireImportOwner(hostId: string, id: string): GameLibraryVersion {
const version = getLibraryVersion(id);
if (database().prepare('SELECT 1 FROM game_file_cleanup WHERE version_id=?').get(id)) throw Object.assign(new Error('Upload files expired under the retention policy.'), { status: 410 });
if (!version || version.sourceHostId !== hostId) throw new Error('Import not found for this host.');
return version;
}
@@ -473,7 +480,7 @@ export function reconcileHostLibraryCleanup(hostId: string): void {
const host = getHost(hostId);
if (!host?.connected || !host.credentialActive || host.mode === 'disabled' || host.decommission
|| !host.inventory.capabilities?.includes('library.remove')) return;
const latest = new Set(latestPublishedLibraries().map(version => version.id));
const latest = retainedGameFileIds();
const caches = handle.prepare('SELECT * FROM host_library_caches WHERE host_id=?').all(hostId) as CacheRow[];
for (const cache of caches) {
if (latest.has(cache.version_id) || (cache.status === 'missing' && cache.bytes_downloaded === 0) || libraryInUse(cache.version_id, hostId)) continue;
@@ -493,7 +500,8 @@ export function cleanupControllerLibraries(): Promise<void> {
const handle = database();
const current = cleaning.get(handle); if (current) return current;
const task = (async () => {
const latest = new Set(latestPublishedLibraries().map(version => version.id));
await cleanupAbandonedUploads();
const latest = retainedGameFileIds();
const versions = handle.prepare(`SELECT v.id,v.storage_location,l.game FROM game_file_versions v
JOIN game_file_libraries l ON l.id=v.library_id LEFT JOIN game_file_cleanup c ON c.version_id=v.id
WHERE v.state='ready' AND c.removed_at IS NULL`).all() as { id: string; storage_location: string; game: string }[];
@@ -521,3 +529,67 @@ export function cleanupControllerLibraries(): Promise<void> {
void task.finally(() => cleaning.delete(handle)).catch(() => {});
return task;
}
function touchUpload(id: string) {
database().prepare("UPDATE game_file_versions SET last_activity_at=? WHERE id=? AND state!='ready'").run(new Date().toISOString(),id);
}
/** Latest N published versions are retained per game; grace starts at supersession. */
export function retainedGameFileIds(settings = fleetSettings()): Set<string> {
const versions = database().prepare(`SELECT v.id,l.game,v.published_at FROM game_file_versions v
JOIN game_file_libraries l ON l.id=v.library_id WHERE v.state='ready'
ORDER BY v.published_at DESC,v.created_at DESC,v.rowid DESC`).all() as { id: string; game: string; published_at: string }[];
const kept = new Set<string>();
const seen = new Map<string,{ count: number; newerAt: string }>();
for (const v of versions) {
const prior = seen.get(v.game);
if (!prior || prior.count < settings.gameFileKeepCount
|| Date.parse(prior.newerAt) > Date.now() - settings.gameFileGraceDays * 86400_000) kept.add(v.id);
seen.set(v.game,{ count: (prior?.count ?? 0) + 1, newerAt: v.published_at });
}
for (const row of database().prepare("SELECT record_id FROM retention_controls WHERE category='game-file' AND protected=1").all() as { record_id: string }[]) kept.add(row.record_id);
return kept;
}
function abandonedUploads(settings = fleetSettings()) {
const days = settings.uploadKeepDays;
if (!days) return [];
return database().prepare(`SELECT v.id,v.storage_location,l.game FROM game_file_versions v JOIN game_file_libraries l ON l.id=v.library_id
WHERE v.state IN ('importing','failed') AND v.source_host_id='' AND COALESCE(v.last_activity_at,v.created_at)<?
AND NOT EXISTS (SELECT 1 FROM game_file_cleanup c WHERE c.version_id=v.id AND c.removed_at IS NOT NULL)
AND NOT EXISTS (SELECT 1 FROM retention_controls c WHERE c.category='upload' AND c.record_id=v.id AND c.protected=1)
ORDER BY v.created_at LIMIT 100`).all(new Date(Date.now()-days*86400_000).toISOString()) as { id: string; storage_location: string; game: string }[];
}
async function cleanupAbandonedUploads() {
for (const upload of abandonedUploads()) {
if (publishing.has(upload.id) || uploadIsActive(stagingRoot(upload.id))) continue;
await withUploadLock(stagingRoot(upload.id), async () => {
// A chunk or resume request may have refreshed the timestamp before lock admission.
if (!abandonedUploads().some(row => row.id === upload.id)) return;
if (!/^[a-zA-Z0-9_-]+$/.test(upload.id)) return;
const expected = join(gameFilesRoot(),upload.game,upload.id);
if (resolve(upload.storage_location) !== expected) return;
database().prepare('INSERT OR IGNORE INTO game_file_cleanup(version_id,started_at) VALUES(?,?)').run(upload.id,new Date().toISOString());
// Paths are controller-derived. Validate parents before recursive removal.
for (const path of [gameFilesRoot(),join(gameFilesRoot(),'.imports'),join(gameFilesRoot(),upload.game),expected,stagingRoot(upload.id),`${stagingRoot(upload.id)}.chunks`]) {
const info = await lstat(path).catch(cause => { if (cause.code === 'ENOENT') return null; throw cause; });
if (info?.isSymbolicLink()) throw new Error('Refusing upload cleanup through a symlink.');
}
for (const path of [expected,stagingRoot(upload.id),`${stagingRoot(upload.id)}.chunks`]) await rm(path,{ recursive: true,force: true });
database().prepare("UPDATE game_file_versions SET state='failed',error='Upload expired under the retention policy.' WHERE id=? AND state!='ready'").run(upload.id);
database().prepare('UPDATE game_file_cleanup SET removed_at=? WHERE version_id=?').run(new Date().toISOString(),upload.id);
});
}
}
export function gameFileRetentionPreview(settings = fleetSettings()) {
const kept = retainedGameFileIds(settings);
const versions = database().prepare(`SELECT v.id,v.size_bytes FROM game_file_versions v LEFT JOIN game_file_cleanup c ON c.version_id=v.id
WHERE v.state='ready' AND c.removed_at IS NULL`).all() as { id: string; size_bytes: number }[];
const eligible = versions.filter(v => !kept.has(v.id) && !libraryInUse(v.id) && !database().prepare("SELECT 1 FROM host_library_caches WHERE version_id=? AND (status!='missing' OR bytes_downloaded>0)").get(v.id));
const uploads = abandonedUploads(settings).filter(v => !publishing.has(v.id) && !uploadIsActive(stagingRoot(v.id)));
return [
{ category: 'game-file',label: 'Unused controller game files',eligible: eligible.length,payloadBytes: eligible.reduce((n,v)=>n+v.size_bytes,0),records: eligible.slice(0,20).map(v=>({ id: v.id })) },
{ category: 'upload',label: 'Abandoned uploads',eligible: uploads.length,payloadBytes: 0,records: uploads.slice(0,20).map(v=>({ id: v.id })) }
];
}
+2
View File
@@ -49,6 +49,8 @@ export function ensureGameFileSchema(handle: Database.Database): void {
CREATE TRIGGER IF NOT EXISTS game_file_versions_immutable BEFORE UPDATE ON game_file_versions
WHEN OLD.state = 'ready' BEGIN SELECT RAISE(ABORT, 'Published game-file versions are immutable'); END;
`);
const versionColumns = handle.prepare('PRAGMA table_info(game_file_versions)').all() as { name: string }[];
if (!versionColumns.some(column => column.name === 'last_activity_at')) handle.exec('ALTER TABLE game_file_versions ADD COLUMN last_activity_at TEXT');
const cacheColumns = handle.prepare('PRAGMA table_info(host_library_caches)').all() as { name: string }[];
if (!cacheColumns.some(column => column.name === 'bytes_verified')) handle.exec('ALTER TABLE host_library_caches ADD COLUMN bytes_verified INTEGER');
if (!cacheColumns.some(column => column.name === 'transfer_progress')) handle.exec('ALTER TABLE host_library_caches ADD COLUMN transfer_progress TEXT');
+1
View File
@@ -46,6 +46,7 @@ export function validateUploadPlan(input: unknown): UploadFilePlan[] {
interface UploadLocks { barrier: Promise<void>; writers: number; files: Map<string, Promise<void>>; active: Set<Promise<void>> }
const uploadLocks = new Map<string, UploadLocks>();
export function uploadIsActive(root: string): boolean { return uploadLocks.has(resolve(root)); }
function locksFor(key: string): UploadLocks {
let locks = uploadLocks.get(key);
if (!locks) { locks = { barrier: Promise.resolve(), writers: 0, files: new Map(), active: new Set() }; uploadLocks.set(key, locks); }
+3 -2
View File
@@ -1,3 +1,4 @@
import { fleetSettings } from '../settings.js';
import { db } from '../db/index.js';
import { ensureHostSchema } from './schema.js';
import type { CapacityCalibration, CapacitySample, CapacitySamplePage, HostInventory, ResourceAllowance } from '../../hosts.js';
@@ -89,8 +90,8 @@ export function recordInstanceResources(hostId: string, inventory: HostInventory
cpu_cores=excluded.cpu_cores,memory_bytes=excluded.memory_bytes,samples=excluded.samples,
window_start=excluded.window_start,observed_at=excluded.observed_at,computed_at=excluded.computed_at`)
.run(scope, phase, cpu, memory, rows.length, rows.at(-1)!.bucket, timestamp, timestamp);
if (scope === '*') handle.prepare('DELETE FROM ec_instance_resource_samples WHERE bucket<?')
.run(new Date(Date.parse(timestamp) - WINDOW_MS).toISOString());
if (scope === '*' && fleetSettings().instanceResourceKeepDays) handle.prepare('DELETE FROM ec_instance_resource_samples WHERE bucket<?')
.run(new Date(Date.parse(timestamp) - Math.max(WINDOW_MS, fleetSettings().instanceResourceKeepDays * 86400_000)).toISOString());
} else {
// Raise the allowance immediately on a new peak; a lower sample
// cannot reduce it until the next scheduled percentile calculation.
+10 -22
View File
@@ -1,3 +1,4 @@
import { pruneHistory } from '../retention.js';
import { recordAgentReceipt } from '../provisioning/agent-events.js';
import { EventEmitter } from 'node:events';
import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypto';
@@ -357,8 +358,8 @@ function recordResources(hostId: string, inventory: HostInventory, timestamp: st
(host_id,sampled_at,cpu_count,cpu_usage_percent,memory_total,memory_available) VALUES(?,?,?,?,?,?)`)
.run(hostId, timestamp, inventory.cpuCount ?? null, inventory.cpuUsagePercent ?? null,
inventory.memoryTotal ?? null, inventory.memoryAvailable ?? null);
database().prepare('DELETE FROM ec_host_resource_samples WHERE host_id=? AND sampled_at<?')
.run(hostId, new Date(Date.parse(timestamp) - 7 * 86400_000).toISOString());
if (fleetSettings().hostResourceKeepDays) database().prepare('DELETE FROM ec_host_resource_samples WHERE host_id=? AND sampled_at<?')
.run(hostId, new Date(Date.parse(timestamp) - fleetSettings().hostResourceKeepDays * 86400_000).toISOString());
}
export function requireReadyHost(hostId: string, options: { libraryId?: string; requiredBytes?: number; image?: string; port?: number; bindAddress?: string; protocol?: 'udp' | 'tcp' } = {}): Host {
const host = requireHost(hostId); const reasons = [...host.readinessReasons];
@@ -391,7 +392,7 @@ export function getHostOperationStatus(id: string): HostOperationStatus | null {
export function listHostOperations(hostId: string, limit = 100, summary = false): HostOperation[] {
const columns = summary ? `o.id,o.host_id,o.kind,o.instance_id,o.generation,o.state,o.error,o.created_at,o.updated_at,'{}' AS payload,NULL AS result`
: `${operationStatusColumns},${operationPayloadSql} AS payload`;
return (database().prepare(`SELECT ${columns} FROM ec_host_operations o WHERE o.host_id=? ORDER BY o.created_at DESC,o.rowid DESC LIMIT ?`)
return (database().prepare(`SELECT ${columns} FROM ec_host_operations o WHERE o.host_id=? AND o.history_expired_at IS NULL ORDER BY o.created_at DESC,o.rowid DESC LIMIT ?`)
.all(hostId, Math.min(500, Math.max(1, limit))) as OperationRow[]).map(operationFromRow);
}
/** Browse retained history without loading command payloads or credential-bearing results. */
@@ -399,6 +400,7 @@ export function listHostOperationHistory(page = 1, filter: HostOperationFilter =
const handle = database();
const pageSize = 25;
const conditions = filter === 'all' ? [] : [filter === 'routine' ? 'o.routine=1' : `(${importantOperationSql})`];
conditions.push('o.history_expired_at IS NULL');
const parameters: string[] = [];
if (scope.host) { conditions.push('o.host_id=?'); parameters.push(scope.host); }
if (scope.server) { conditions.push('o.server_name=?'); parameters.push(scope.server); }
@@ -423,22 +425,7 @@ export function listHostOperationHistory(page = 1, filter: HostOperationFilter =
}
/** Bound each cleanup pass so retained history cannot stall monitoring or receipts. */
export function pruneHostObservations(): number {
const days = fleetSettings().observationKeepDays;
if (!Number.isSafeInteger(days) || days <= 0 || days > 36500) return 0;
const handle = database();
const cutoff = new Date(Date.now() - days * 86_400_000).toISOString();
return handle.transaction(() => {
const rows = handle.prepare(`SELECT id FROM ec_host_operations WHERE observation_success=1 AND state='completed'
AND updated_at<? ORDER BY updated_at LIMIT 500`).all(cutoff) as { id: string }[];
if (!rows.length) return 0;
// Fixed SQL text prevents each batch size from adding three cached statements.
const ids = JSON.stringify(rows.map(row => row.id));
handle.prepare('DELETE FROM ec_host_operation_events WHERE operation_id IN (SELECT value FROM json_each(?))').run(ids);
handle.prepare('DELETE FROM ec_host_operation_payloads WHERE operation_id IN (SELECT value FROM json_each(?))').run(ids);
return handle.prepare('DELETE FROM ec_host_operations WHERE id IN (SELECT value FROM json_each(?))').run(ids).changes;
})();
}
export function pruneHostObservations(): number { return pruneHistory('observation'); }
/** Heartbeats can tell an executing create to stop at its next safe command boundary. */
export function pendingCreationCancellations(hostId: string): string[] {
return (database().prepare(`SELECT json_extract(payload,'$.createOperationId') AS id FROM ec_host_operations
@@ -464,11 +451,12 @@ function enqueueHostOperation(hostId: string, kind: string, payload: string | Bu
if (options.generation !== undefined && (!Number.isSafeInteger(options.generation) || options.generation < 1)) throw new Error('Instance generation must be a positive integer.');
const operation = database().transaction(() => {
if (options.idempotencyKey) {
const existing = database().prepare(`SELECT o.id,o.kind,o.instance_id,o.generation,${operationPayloadSql}=CAST(? AS TEXT) AS same_payload
const existing = database().prepare(`SELECT o.id,o.kind,o.instance_id,o.generation,o.history_expired_at,o.request_hash,${operationPayloadSql}=CAST(? AS TEXT) AS same_payload
FROM ec_host_operations o WHERE o.host_id=? AND o.idempotency_key=?`).get(payload, hostId, options.idempotencyKey) as
{ id: string; kind: string; instance_id: string | null; generation: number | null; same_payload: number } | undefined;
{ id: string; kind: string; instance_id: string | null; generation: number | null; history_expired_at: string | null; request_hash: string | null; same_payload: number } | undefined;
if (existing) {
if (existing.kind !== kind || existing.instance_id !== (options.instanceId ?? null) || existing.generation !== (options.generation ?? null) || !existing.same_payload) throw new Error('Operation idempotency key is already associated with a different request.');
const samePayload = existing.history_expired_at ? existing.request_hash === createHash('sha256').update(payload).digest('hex') : !!existing.same_payload;
if (existing.kind !== kind || existing.instance_id !== (options.instanceId ?? null) || existing.generation !== (options.generation ?? null) || !samePayload) throw new Error('Operation idempotency key is already associated with a different request.');
return getHostOperationStatus(existing.id)!;
}
}
+2
View File
@@ -88,6 +88,8 @@ export function ensureHostSchema(handle: Database.Database): void {
);
`);
const columns = handle.prepare('PRAGMA table_info(ec_host_operations)').all() as { name: string }[];
if (!columns.some(column => column.name === 'history_expired_at')) handle.exec('ALTER TABLE ec_host_operations ADD COLUMN history_expired_at TEXT');
if (!columns.some(column => column.name === 'request_hash')) handle.exec('ALTER TABLE ec_host_operations ADD COLUMN request_hash TEXT');
if (!columns.some(column => column.name === 'last_delivered_at')) handle.exec('ALTER TABLE ec_host_operations ADD COLUMN last_delivered_at TEXT');
if (!columns.some(column => column.name === 'observation_success')) {
handle.transaction(() => {
+1 -3
View File
@@ -120,9 +120,7 @@ async function runTick(): Promise<void> {
try { await processJob(job); }
catch { hold(job, 'Maintenance check failed; waiting for the next check.'); }
}
// Update attempts are also durable suppression evidence; never age them into an automatic retry.
db().prepare(`DELETE FROM maintenance_jobs WHERE kind != 'update' AND finished_at < ? AND state IN ('done','failed','cancelled')`)
.run(new Date(Date.now() - 30 * 24 * HOUR).toISOString());
}
async function processJob(job: MaintenanceJob): Promise<void> {
const server = getServer(job.server);
+4 -6
View File
@@ -1,4 +1,4 @@
import { fleetSettings } from '../settings.js';
import { pruneHistory } from '../retention.js';
import { db } from '../db/index.js';
import { emit } from './bus.js';
import { requestRestart } from '../provisioning/index.js';
@@ -22,7 +22,8 @@ export function sweep(): Promise<SweepResult> {
async function runSweep(): Promise<SweepResult> {
pruneHostObservations();
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') ORDER BY d.started_at, d.id, ds.server`).all() as PendingRow[];
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[];
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.
@@ -46,10 +47,7 @@ async function runSweep(): Promise<SweepResult> {
}
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(`DELETE FROM deployments WHERE started_at < ?
AND NOT EXISTS (SELECT 1 FROM watchdog_actions WHERE campaign_id = deployments.id) 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(new Date(Date.now() - fleetSettings().deployKeepDays * 86_400_000).toISOString());
pruneHistory('campaign');
emit({ type: 'sweep-tick', frozen: false });
return result;
}
+4 -2
View File
@@ -1,3 +1,4 @@
import { pruneHistory, pruneSampleHistory } from '../retention.js';
import { config } from '../config.js';
import { processProvisioning } from '../provisioning/worker.js';
import { sweep } from './sweep.js';
@@ -51,13 +52,14 @@ function reconcileRequesters(): void {
['campaign', 'Reconciling campaigns', sweep],
['beta', 'Reconciling beta tests', betaTick],
['maintenance', 'Checking maintenance and recovery', runMaintenance],
['game-files', 'Removing unused game files', cleanupControllerLibraries]
['game-files', 'Removing unused game files', cleanupControllerLibraries],
['retention', 'Cleaning expired history', async () => { pruneHistory(); pruneSampleHistory(); }]
];
for (const [key, label, work] of requesters) {
if (requesterFlights.has(key)) continue;
const flight = trackActivity(key, label, work)
.catch(err => { console.error(`[worker] ${key} error:`, err); })
.then(async () => { if (key !== 'game-files') await tick(true); })
.then(async () => { if (key !== 'game-files' && key !== 'retention') await tick(true); })
.finally(() => { requesterFlights.delete(key); });
requesterFlights.set(key, flight);
}
+246
View File
@@ -0,0 +1,246 @@
import { createHash } from 'node:crypto';
import { db } from './db/index.js';
import { fleetSettings } from './settings.js';
import { ensureHostSchema } from './hosts/schema.js';
import { ensureGameFileSchema } from './game-files/schema.js';
import { ensureAuthSchema } from './auth/schema.js';
import type { FleetSettings } from '../fleet-settings.js';
interface Policy {
category: string; label: string; table: string; id: string; clock: string;
days: keyof FleetSettings; eligible: string; bytes?: string; failed?: string; minimum?: boolean;
}
const terminal = "('done','warning','failed','cancelled')";
// Cancellation stops observation, not an already-delivered command. Its source
// history and before images must survive until the receipt has been reconciled.
const unsettledDeployment = `d.state NOT IN ${terminal}
OR EXISTS (SELECT 1 FROM restart_rollout r WHERE r.operation_id=d.id AND r.healthy_at IS NULL)
OR EXISTS (SELECT 1 FROM native_provisioning n JOIN ec_host_operations a ON a.id IN (n.operation_id,n.stage_operation_id)
LEFT JOIN native_instances i ON i.id=n.instance_id WHERE n.deployment_id=d.id AND
(a.state NOT IN ('completed','failed') OR (a.id=n.operation_id AND a.state='completed' AND (
(n.applied_at IS NULL AND COALESCE(json_array_length(n.files_json),0)>0)
OR (d.action='destroy' AND i.retired_at IS NULL)
OR (d.action='rebuild' AND i.retired_at IS NULL AND i.generation=n.generation
AND NOT EXISTS (SELECT 1 FROM ec_host_migrations m WHERE m.deployment_id=d.id))))))
OR EXISTS (SELECT 1 FROM ec_host_migrations m JOIN ec_host_operations a ON a.id=m.remove_operation_id
WHERE m.deployment_id=d.id AND (a.state NOT IN ('completed','failed') OR (a.state='completed' AND m.removed_at IS NULL)))`;
const unreferencedOperation = `NOT EXISTS (SELECT 1 FROM native_provisioning n WHERE o.id IN (n.operation_id,n.stage_operation_id))
AND NOT EXISTS (SELECT 1 FROM ec_host_migrations m WHERE m.remove_operation_id=o.id)
AND NOT EXISTS (SELECT 1 FROM ec_host_decommissions d WHERE d.operation_id=o.id)
AND o.kind NOT IN ('runtime.pull','library.download','library.import','library.remove','host.decommission','host.rotate-credential')
AND NOT EXISTS (SELECT 1 FROM ec_host_operations dependent WHERE dependent.state IN ('pending','accepted','running','unknown')
AND json_extract(dependent.payload,'$.createOperationId')=o.id)`;
const campaignSafe = `NOT EXISTS (SELECT 1 FROM deployment_servers s WHERE s.deployment_id=o.id AND s.server_deleted_at IS NULL
AND s.restart_state IN ('pending','held','manual'))
AND NOT EXISTS (SELECT 1 FROM watchdog_actions a WHERE a.campaign_id=o.id)
AND NOT EXISTS (SELECT 1 FROM provisioning_sources s JOIN server_deployments d ON d.id=s.operation_id
WHERE s.kind='campaign' AND s.source_id=o.id AND (${unsettledDeployment}))
AND NOT EXISTS (SELECT 1 FROM deployments d WHERE d.rolled_back_by=o.id)`;
const failedCampaign = `EXISTS (SELECT 1 FROM deployment_servers s WHERE s.deployment_id=o.id AND s.server_deleted_at IS NULL AND s.restart_state='failed')`;
function policies(): Policy[] {
return [
{ category: 'campaign', label: 'Campaigns', table: 'deployments', id: 'id', clock: 'started_at', days: 'deployKeepDays',
eligible: `${campaignSafe} AND NOT EXISTS (SELECT 1 FROM retention_controls c WHERE c.category='file-content' AND c.record_id=o.id AND c.protected=1)`, failed: failedCampaign,
bytes: `(SELECT COALESCE(SUM(COALESCE(length(content_before),0)+COALESCE(length(content_after),0)),0) FROM deployment_files WHERE deployment_id=o.id)` },
{ category: 'provisioning', label: 'Provisioning workflows', table: 'server_deployments', id: 'id', clock: 'finished_at', days: 'provisioningKeepDays', failed: "o.state='failed'", eligible: `o.state IN ${terminal} AND NOT (${unsettledDeployment.replaceAll('d.', 'o.')})
AND NOT EXISTS (SELECT 1 FROM restart_rollout r WHERE r.operation_id=o.id AND r.healthy_at IS NULL)
AND NOT EXISTS (SELECT 1 FROM native_provisioning n JOIN ec_host_operations a ON a.id IN (n.operation_id,n.stage_operation_id) WHERE n.deployment_id=o.id AND a.state NOT IN ('completed','failed'))
AND NOT EXISTS (SELECT 1 FROM ec_host_migrations m JOIN ec_host_operations a ON a.id=m.remove_operation_id WHERE m.deployment_id=o.id AND a.state NOT IN ('completed','failed'))
AND NOT EXISTS (SELECT 1 FROM deployment_servers s WHERE s.provisioning_id=o.id)
AND NOT EXISTS (SELECT 1 FROM beta_servers s WHERE s.provisioning_id=o.id)
AND NOT EXISTS (SELECT 1 FROM maintenance_jobs s WHERE s.provisioning_id=o.id)
AND NOT EXISTS (SELECT 1 FROM watchdog_actions s WHERE s.operation_id=o.id)
AND NOT EXISTS (SELECT 1 FROM native_creation_cancellations c WHERE o.id IN (c.creation_id,c.cleanup_id))
AND NOT EXISTS (SELECT 1 FROM server_creation_dependencies c WHERE c.removal_id=o.id)
AND NOT EXISTS (SELECT 1 FROM server_collection_members m JOIN server_collection_members sibling ON sibling.collection_id=m.collection_id
JOIN server_deployments d ON d.id=sibling.deployment_id WHERE m.deployment_id=o.id AND (${unsettledDeployment}))`,
bytes: "COALESCE(length(o.observation_json),0)+(SELECT COALESCE(SUM(length(recipe)),0) FROM server_launch_recipes WHERE deployment_id=o.id)" },
{ category: 'operation', label: 'Agent actions and resolved failures', table: 'ec_host_operations', id: 'id', clock: 'updated_at', days: 'actionKeepDays',
failed: "(o.state='failed' OR (o.routine=1 AND o.observation_success=0))",
eligible: `o.history_expired_at IS NULL AND o.state IN ('completed','failed') AND (o.routine=0 OR o.observation_success=0) AND ${unreferencedOperation}`,
bytes: "COALESCE(length(o.payload),0)+COALESCE(length(o.result),0)+COALESCE((SELECT length(payload) FROM ec_host_operation_payloads WHERE operation_id=o.id),0)" },
{ category: 'observation', label: 'Routine operations', table: 'ec_host_operations', id: 'id', clock: 'updated_at', days: 'observationKeepDays', eligible: "o.state='completed' AND o.observation_success=1", bytes: 'length(o.payload)+COALESCE(length(o.result),0)' },
{ category: 'maintenance', label: 'Maintenance jobs', table: 'maintenance_jobs', id: 'id', clock: 'finished_at', days: 'maintenanceKeepDays',
eligible: `o.kind!='update' AND o.state IN ('done','failed','cancelled') AND NOT EXISTS (SELECT 1 FROM server_deployments d WHERE d.id=o.provisioning_id AND (${unsettledDeployment}))` },
{ category: 'recovery', label: 'Resolved recovery incidents', table: 'watchdog_incidents', id: 'id', clock: 'finished_at', days: 'recoveryKeepDays',
eligible: `o.state='resolved' AND NOT EXISTS (SELECT 1 FROM watchdog_actions a JOIN server_deployments d ON d.id=a.operation_id WHERE a.incident_id=o.id AND (${unsettledDeployment}))
AND NOT EXISTS (SELECT 1 FROM watchdog_actions a JOIN restart_rollout r ON r.operation_id=a.operation_id WHERE a.incident_id=o.id AND r.healthy_at IS NULL)` },
{ category: 'match', label: 'Matches', table: 'leaderboard_matches', id: 'id', clock: 'last_received_at', days: 'matchKeepDays', eligible: "o.status IN ('completed','interrupted')", bytes: '0' },
{ category: 'beta', label: 'Completed betas', table: 'betas', id: 'id', clock: 'closed_at', days: 'betaKeepDays',
eligible: `o.state IN ('promoted','rolled-back') AND NOT EXISTS (SELECT 1 FROM retention_controls c WHERE c.category='beta-file-content' AND c.record_id=CAST(o.id AS TEXT) AND c.protected=1) AND NOT EXISTS (SELECT 1 FROM deployments d WHERE d.beta_id=o.id)
AND NOT EXISTS (SELECT 1 FROM provisioning_sources s JOIN server_deployments d ON d.id=s.operation_id WHERE s.kind='beta' AND s.source_id=CAST(o.id AS TEXT) AND (${unsettledDeployment}))`,
bytes: '(SELECT COALESCE(SUM(length(content)),0) FROM beta_revision_files WHERE beta_id=o.id)' },
{ category: 'file-content', label: 'Campaign file bodies', table: 'deployments', id: 'id', clock: 'finished_at', days: 'fileContentKeepDays',
eligible: `${campaignSafe} AND NOT (${failedCampaign})
AND NOT EXISTS (SELECT 1 FROM retention_controls c WHERE c.category='campaign' AND c.record_id=o.id AND c.protected=1)
AND NOT EXISTS (SELECT 1 FROM retention_expirations e WHERE e.category='file-content' AND e.record_id=o.id)`,
bytes: '(SELECT COALESCE(SUM(COALESCE(length(content_before),0)+COALESCE(length(content_after),0)),0) FROM deployment_files WHERE deployment_id=o.id)' },
{ category: 'beta-file-content', label: 'Closed beta file snapshots', table: 'betas', id: 'id', clock: 'closed_at', days: 'fileContentKeepDays',
eligible: `o.state IN ('promoted','rolled-back') AND NOT EXISTS (SELECT 1 FROM deployments d WHERE d.beta_id=o.id)
AND NOT EXISTS (SELECT 1 FROM retention_controls c WHERE c.category='beta' AND c.record_id=CAST(o.id AS TEXT) AND c.protected=1)
AND NOT EXISTS (SELECT 1 FROM retention_expirations e WHERE e.category='beta-file-content' AND e.record_id=CAST(o.id AS TEXT))
AND NOT EXISTS (SELECT 1 FROM provisioning_sources s JOIN server_deployments d ON d.id=s.operation_id WHERE s.kind='beta' AND s.source_id=CAST(o.id AS TEXT) AND (${unsettledDeployment}))`,
bytes: '(SELECT COALESCE(SUM(length(content)),0) FROM beta_revision_files WHERE beta_id=o.id)' },
{ category: 'version', label: 'Unused artifact versions', table: 'script_versions', id: 'id', clock: 'uploaded_at', days: 'versionKeepDays', minimum: false,
eligible: `NOT EXISTS (SELECT 1 FROM scripts s WHERE s.current_version_id=o.id)
AND NOT EXISTS (SELECT 1 FROM staged_file_changes s WHERE s.script_version_id=o.id)
AND ${versionReferences()}`, bytes: 'length(o.content)' },
{ category: 'audit', label: 'Authentication audit events', table: 'auth_audit', id: 'id', clock: 'created_at', days: 'auditKeepDays', eligible: '1', bytes: 'length(o.details)' }
];
}
function versionReferences(): string {
// Even cascading references pin versions: automatic cleanup must never erase
// another feature's history as a side effect of deleting an artifact.
const tables = db().prepare("SELECT name FROM sqlite_master WHERE type='table'").all() as { name: string }[];
const clauses: string[] = [];
for (const { name } of tables) {
if (name === 'plugin_imports') continue;
const quote = (s: string) => '"' + s.replaceAll('"','""') + '"';
for (const fk of db().prepare(`PRAGMA foreign_key_list(${quote(name)})`).all() as { table: string; from: string }[]) {
if (fk.table === 'script_versions') clauses.push(`NOT EXISTS (SELECT 1 FROM ${quote(name)} v WHERE v.${quote(fk.from)}=o.id)`);
}
}
return clauses.join(' AND ') || '1';
}
function initialize() { ensureHostSchema(db()); ensureAuthSchema(db()); }
function selection(policy: Policy, settings = fleetSettings()) {
const days = settings[policy.days];
const failureDays = settings.failureKeepDays;
const cut = (n: number) => new Date(Date.now() - n * 86400_000).toISOString();
const age = ['observation','operation'].includes(policy.category) ? `o.${policy.clock} < @cutoff` : policy.category === 'audit' ? "o.created_at < @numericCutoff" : `julianday(o.${policy.clock}) < julianday(@cutoff)`;
const failed = policy.failed ?? '0';
const enabled = days > 0 || policy.category === 'version' && settings.versionKeepCount > 0;
const normal = `(${enabled ? 1 : 0} AND NOT (${failed}) AND ${policy.category === 'version' && !days ? '1' : age})`;
const failure = `(${failureDays > 0 && !!policy.failed ? 1 : 0} AND (${failed}) AND EXISTS (SELECT 1 FROM retention_controls c WHERE c.category=@category AND c.record_id=CAST(o.${policy.id} AS TEXT) AND julianday(c.resolved_at)>=julianday(o.${policy.clock}) AND julianday(c.resolved_at)<julianday(@failureCutoff)))`;
const minimum = policy.minimum === false ? 0 : settings.historyKeepCount;
const minimumScope = policy.category === 'observation' ? "WHERE observation_success=1 AND state='completed'"
: policy.category === 'operation' ? "WHERE history_expired_at IS NULL AND (routine=0 OR observation_success=0)" : '';
const versionCount = policy.category === 'version' ? `AND o.id NOT IN (SELECT id FROM script_versions v WHERE v.script_id=o.script_id ORDER BY version DESC LIMIT ${settings.versionKeepCount})` : '';
return { sql: `FROM ${policy.table} o WHERE (${policy.eligible}) AND (${normal} OR ${failure})
AND NOT EXISTS (SELECT 1 FROM retention_controls c WHERE c.category=@category AND c.record_id=CAST(o.${policy.id} AS TEXT) AND c.protected=1)
AND o.rowid NOT IN (SELECT rowid FROM ${policy.table} ${minimumScope} ORDER BY ${policy.clock} DESC,rowid DESC LIMIT ${minimum}) ${versionCount}`,
params: { category: policy.category, cutoff: cut(days), numericCutoff: Date.now() - days * 86400_000, failureCutoff: cut(failureDays) } };
}
// better-sqlite3 accepts unused named bindings, keeping selection reusable.
export function retentionPreview(settings = fleetSettings()) {
initialize();
return policies().map(policy => {
const query = selection(policy, settings);
const summary = db().prepare(`SELECT COUNT(*) AS eligible,COALESCE(SUM(COALESCE(${policy.bytes ?? '0'},0)),0) AS payloadBytes ${query.sql}`).get(query.params) as { eligible: number; payloadBytes: number };
const records = db().prepare(`SELECT CAST(o.${policy.id} AS TEXT) AS id ${query.sql} ORDER BY o.${policy.clock},o.rowid LIMIT 20`).all(query.params) as { id: string }[];
return { category: policy.category, label: policy.label, ...summary, records };
});
}
export function retentionControls() {
return db().prepare('SELECT category,record_id AS id,protected,resolved_at AS resolvedAt FROM retention_controls WHERE protected=1 OR resolved_at IS NOT NULL ORDER BY category,record_id').all();
}
export function setRetentionControl(input: { category: string; id: string; protected?: boolean; resolve?: boolean }) {
initialize();
ensureGameFileSchema(db());
const policy = policies().find(policy => policy.category === input.category)
?? (['game-file','upload'].includes(input.category) ? { category: input.category, table: 'game_file_versions',id: 'id',failed: undefined } : undefined);
if (!policy || typeof input.id !== 'string' || !input.id || input.id.length > 200
|| (input.protected !== undefined && typeof input.protected !== 'boolean') || (input.resolve !== undefined && typeof input.resolve !== 'boolean')) throw new Error('Choose a valid history category and record ID.');
const row = db().prepare(`SELECT 1 FROM ${policy.table} o WHERE CAST(o.${policy.id} AS TEXT)=? ${policy.table === 'ec_host_operations' ? 'AND o.history_expired_at IS NULL' : ''} ${input.resolve ? `AND (${policy.failed ?? '0'})` : ''}`).get(input.id);
if (!row) throw new Error(input.resolve ? 'Only a failed record can be marked resolved.' : 'History record was not found.');
if (input.protected && (db().prepare('SELECT 1 FROM retention_expirations WHERE category=? AND record_id=?').get(input.category,input.id)
|| ['game-file','upload'].includes(input.category) && db().prepare('SELECT 1 FROM game_file_cleanup WHERE version_id=?').get(input.id))) {
throw new Error('These files have expired or cleanup has already started. Protection cannot restore them.');
}
db().prepare(`INSERT INTO retention_controls(category,record_id,protected,resolved_at) VALUES(?,?,?,?)
ON CONFLICT(category,record_id) DO UPDATE SET protected=COALESCE(?,protected),resolved_at=COALESCE(?,resolved_at)`)
.run(input.category,input.id,Number(input.protected ?? false),input.resolve ? new Date().toISOString() : null,
input.protected === undefined ? null : Number(input.protected),input.resolve ? new Date().toISOString() : null);
}
/** Each policy removes at most 500 eligible records per pass; dependencies are checked again under the transaction. */
export function pruneHistory(only?: string): number {
initialize();
if (only === 'observation') return pruneRoutineObservations();
let removed = only ? 0 : pruneRoutineObservations();
for (const policy of policies().filter(policy => policy.category !== 'observation' && (!only || policy.category === only))) {
removed += db().transaction(() => {
const query = selection(policy);
const rows = db().prepare(`SELECT o.rowid AS rid,CAST(o.${policy.id} AS TEXT) AS id ${query.sql} ORDER BY o.${policy.clock},o.rowid LIMIT 500`).all(query.params) as { rid: number; id: string }[];
for (const row of rows) {
if (policy.category === 'provisioning') {
db().prepare('DELETE FROM native_provisioning WHERE deployment_id=?').run(row.id);
db().prepare('DELETE FROM ec_host_migrations WHERE deployment_id=?').run(row.id);
db().prepare('DELETE FROM restart_rollout WHERE operation_id=? AND healthy_at IS NOT NULL').run(row.id);
}
if (policy.category === 'operation' || policy.category === 'observation') {
db().prepare('DELETE FROM ec_host_operation_events WHERE operation_id=?').run(row.id);
if (policy.category === 'operation') {
const payload = db().prepare(`SELECT COALESCE(p.payload,o.payload) AS payload FROM ec_host_operations o LEFT JOIN ec_host_operation_payloads p ON p.operation_id=o.id WHERE o.id=?`).get(row.id) as { payload: string };
db().prepare(`UPDATE ec_host_operations SET request_hash=?,payload='{}',result='{"historyExpired":true}',error=NULL,history_expired_at=? WHERE id=?`)
.run(createHash('sha256').update(payload.payload).digest('hex'),new Date().toISOString(),row.id);
}
db().prepare('DELETE FROM ec_host_operation_payloads WHERE operation_id=?').run(row.id);
}
if (policy.category === 'beta-file-content') {
db().prepare("UPDATE beta_revision_files SET content=X'' WHERE beta_id=?").run(row.id);
db().prepare('DELETE FROM beta_file_snapshots WHERE beta_id=?').run(row.id);
db().prepare('INSERT OR IGNORE INTO retention_expirations(category,record_id,expired_at) VALUES(?,?,?)').run(policy.category,row.id,new Date().toISOString());
} else if (policy.category === 'file-content') {
db().prepare('UPDATE deployment_files SET content_before=NULL,content_after=NULL WHERE deployment_id=?').run(row.id);
db().prepare("UPDATE staged_file_changes SET content=NULL WHERE source_kind='campaign' AND source_id=? AND state IN ('applied','cancelled')").run(row.id);
db().prepare('INSERT OR IGNORE INTO retention_expirations(category,record_id,expired_at) VALUES(?,?,?)').run(policy.category,row.id,new Date().toISOString());
} else if (policy.category !== 'operation') {
db().prepare(`DELETE FROM ${policy.table} WHERE rowid=?`).run(row.rid);
db().prepare('DELETE FROM retention_controls WHERE category=? AND record_id=?').run(policy.category,row.id);
}
}
return rows.length;
})();
}
return removed;
}
function samplePolicies(s = fleetSettings()) {
const iso = (days: number) => new Date(Date.now() - days * 86400_000).toISOString();
return [
{ category: 'host-resources', label: 'Host resource samples', table: 'ec_host_resource_samples', condition: 'sampled_at < ?', value: iso(s.hostResourceKeepDays), enabled: s.hostResourceKeepDays > 0 },
{ category: 'instance-resources', label: 'Instance resource samples', table: 'ec_instance_resource_samples', condition: 'bucket < ?', value: iso(Math.max(1,s.instanceResourceKeepDays)), enabled: s.instanceResourceKeepDays > 0 },
{ category: 'metrics', label: 'Fleet metrics samples', table: 'metric_samples', condition: 'at < ?', value: Date.now() - s.metricsKeepDays * 86400_000, enabled: s.metricsKeepDays > 0 },
{ category: 'telemetry', label: 'Telemetry diagnostics (age expiry)', table: 'telemetry_requests', condition: 'received_at < ?', value: iso(s.telemetryKeepDays), enabled: s.telemetryKeepDays > 0 }
];
}
export function sampleRetentionPreview(settings = fleetSettings()) {
initialize();
const result = samplePolicies(settings).filter(p => p.category !== 'telemetry').map(p => ({ category: p.category,label: p.label,records: [],payloadBytes: 0,
eligible: p.enabled ? (db().prepare(`SELECT COUNT(*) AS n FROM ${p.table} WHERE ${p.condition}`).get(p.value) as { n: number }).n : 0 }));
const telemetry = db().prepare(`SELECT COUNT(*) AS eligible,COALESCE(SUM(bytes),0) AS payloadBytes FROM (
SELECT received_at,length(payload) AS bytes,ROW_NUMBER() OVER (PARTITION BY server ORDER BY id DESC) AS per_server,
ROW_NUMBER() OVER (ORDER BY id DESC) AS overall FROM telemetry_requests)
WHERE (? > 0 AND received_at < ?) OR (? > 0 AND per_server > ?) OR (? > 0 AND overall > ?)`)
.get(settings.telemetryKeepDays,new Date(Date.now()-settings.telemetryKeepDays*86400_000).toISOString(),settings.telemetryMaxPerServer,
settings.telemetryMaxPerServer,settings.telemetryMaxTotal,settings.telemetryMaxTotal) as {eligible:number;payloadBytes:number};
result.push({category:'telemetry',label:'Telemetry diagnostics (age and count limits)',records:[],...telemetry});
return result;
}
export function pruneSampleHistory() {
initialize();
for (const p of samplePolicies().filter(p => p.enabled)) db().prepare(`DELETE FROM ${p.table} WHERE rowid IN (SELECT rowid FROM ${p.table} WHERE ${p.condition} LIMIT 5000)`).run(p.value);
const s = fleetSettings();
if (s.telemetryMaxPerServer) db().prepare(`DELETE FROM telemetry_requests WHERE id IN (SELECT id FROM
(SELECT id,ROW_NUMBER() OVER (PARTITION BY server ORDER BY id DESC) AS rank FROM telemetry_requests) WHERE rank>? LIMIT 5000)`).run(s.telemetryMaxPerServer);
if (s.telemetryMaxTotal) db().prepare('DELETE FROM telemetry_requests WHERE id IN (SELECT id FROM telemetry_requests ORDER BY id DESC LIMIT 5000 OFFSET ?)').run(s.telemetryMaxTotal);
}
function pruneRoutineObservations(): number {
const settings = fleetSettings();
if (!settings.observationKeepDays) return 0;
const handle = db();
return handle.transaction(() => {
const rows = handle.prepare(`SELECT id FROM ec_host_operations WHERE observation_success=1 AND state='completed'
AND updated_at<? AND NOT EXISTS (SELECT 1 FROM retention_controls c WHERE c.category='observation' AND c.record_id=ec_host_operations.id AND c.protected=1)
${settings.historyKeepCount ? `AND id NOT IN (SELECT id FROM ec_host_operations WHERE observation_success=1 AND state='completed' ORDER BY updated_at DESC LIMIT ${settings.historyKeepCount})` : ''}
ORDER BY updated_at LIMIT 500`).all(new Date(Date.now()-settings.observationKeepDays*86400_000).toISOString()) as { id: string }[];
if (!rows.length) return 0;
const ids = JSON.stringify(rows.map(row=>row.id));
handle.prepare('DELETE FROM ec_host_operation_events WHERE operation_id IN (SELECT value FROM json_each(?))').run(ids);
handle.prepare('DELETE FROM ec_host_operation_payloads WHERE operation_id IN (SELECT value FROM json_each(?))').run(ids);
return handle.prepare('DELETE FROM ec_host_operations WHERE id IN (SELECT value FROM json_each(?))').run(ids).changes;
})();
}