From e6912ceb57d25aa9d54d719dfb5a6a6d4942d102 Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Mon, 5 Oct 2026 13:09:44 -0400 Subject: [PATCH] Add configurable retention for successful routine operations --- agent/ec_agent.py | 78 +++++++- agent/tests/test_agent.py | 188 ++++++++++++++++++ src/lib/fleet-settings.ts | 14 +- src/lib/server/db/fleet-settings.ts | 5 +- src/lib/server/hosts/index.ts | 25 ++- src/lib/server/hosts/observations.ts | 18 ++ src/lib/server/hosts/schema.ts | 11 + src/lib/server/pipeline/sweep.ts | 2 + src/routes/api/agent/heartbeat/+server.ts | 4 +- src/routes/settings/+page.svelte | 3 + static/agent/ec-agent.py | 78 +++++++- static/agent/manifest.json | 4 +- tests/unit/fleet-settings.test.ts | 37 +++- tests/unit/host-observation-retention.test.ts | 155 +++++++++++++++ 14 files changed, 590 insertions(+), 32 deletions(-) create mode 100644 src/lib/server/hosts/observations.ts create mode 100644 tests/unit/host-observation-retention.test.ts diff --git a/agent/ec_agent.py b/agent/ec_agent.py index 8b0373c..187ad6b 100644 --- a/agent/ec_agent.py +++ b/agent/ec_agent.py @@ -32,7 +32,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.4.8" +VERSION = "1.4.9" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -516,6 +516,19 @@ class ReadWriteGate: self.condition.notify_all() +def read_operation(operation): + if not isinstance(operation, dict): + return False + if operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"): + return True + payload = operation.get("payload") + return operation.get("kind") == "rcon.execute" and isinstance(payload, dict) and payload.get("command") in ("status", "getstatus", "version") + + +def valid_observation_keep_days(value): + return type(value) is int and 0 <= value <= 36500 + + class Journal: def __init__(self, path, recover=True): self.db = sqlite3.connect(path) @@ -525,8 +538,14 @@ class Journal: self.db.execute("PRAGMA journal_mode=WAL") self.db.execute("PRAGMA synchronous=FULL") self.db.execute("CREATE TABLE IF NOT EXISTS operations (id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)") - if "checkpoint" not in {row[1] for row in self.db.execute("PRAGMA table_info(operations)")}: + columns = {row[1] for row in self.db.execute("PRAGMA table_info(operations)")} + if "checkpoint" not in columns: self.db.execute("ALTER TABLE operations ADD COLUMN checkpoint TEXT") + if "reported_at" not in columns: + self.db.execute("ALTER TABLE operations ADD COLUMN reported_at REAL") + if "retention_eligible" not in columns: + self.db.execute("ALTER TABLE operations ADD COLUMN retention_eligible INTEGER NOT NULL DEFAULT 0") + self.db.execute("CREATE INDEX IF NOT EXISTS operations_observation_retention ON operations(updated,id) WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1") self.db.execute("UPDATE operations SET state='unknown', error='Agent restarted during execution; inspect before reconciliation', updated=? WHERE state='running'", (time.time(),)) self.db.commit() @@ -551,10 +570,37 @@ class Journal: return dict(row) def update(self, op_id, state, result=None, error=None): - self.db.execute("UPDATE operations SET state=?, result=?, error=?, updated=? WHERE id=?", (state, json.dumps(result) if result is not None else None, error, time.time(), op_id)) + self.db.execute("UPDATE operations SET state=?, result=?, error=?, updated=?, reported_at=NULL, retention_eligible=0 WHERE id=?", (state, json.dumps(result) if result is not None else None, error, time.time(), op_id)) self.db.commit() return self.get(op_id) + def confirm_completed(self, row): + """Only an acknowledged, unchanged completion can become disposable.""" + if row["state"] != "completed": + return + eligible = False + try: + result = json.loads(row["result"]) if row.get("result") is not None else None + eligible = read_operation(json.loads(row["request"])) and not row.get("error") and not ( + isinstance(result, dict) and (result.get("ok") is False or result.get("healthy") is False or result.get("error") not in (None, "")) + ) + except (ValueError, TypeError, KeyError): + pass # Malformed historical records are retained for inspection. + self.db.execute("""UPDATE operations SET reported_at=COALESCE(reported_at,?), retention_eligible=? + WHERE id=? AND state='completed' AND updated=? AND result IS ? AND error IS ?""", + (time.time(), int(bool(eligible)), row["id"], row["updated"], row.get("result"), row.get("error"))) + self.db.commit() + + def prune_observations(self, keep_days): + if not valid_observation_keep_days(keep_days) or keep_days == 0: + return 0 + cutoff = time.time() - keep_days * 86400 + removed = self.db.execute("""DELETE FROM operations WHERE id IN ( + SELECT id FROM operations WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1 + AND updated= next_prune: + next_prune = time.monotonic() + 60 + with self.activity(read_only=True) as acquired: + if acquired and not self.update_pending(): + if journal.prune_observations(self.observation_keep_days) == 500: + next_prune = time.monotonic() + 15 response = self.request("GET", f"/api/agent/operations?lane={lane}&wait=20") handled = False for operation in response.get("operations", []): @@ -2242,7 +2302,7 @@ class Agent: continue row, _ = journal.accept(operation) if row["state"] in ("accepted", "unknown", "running"): - self.report(journal.update(operation["id"], "running")) + self.report(journal.update(operation["id"], "running"), journal=journal) with self.read_gate.read(), self.instance_lock(operation): try: result = self.dispatch(operation) @@ -2251,7 +2311,7 @@ class Agent: row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable") # Receipt delivery can take seconds; it must not # hold the instance lock needed by health checks. - self.report(row) + self.report(row, journal=journal) handled = True finally: with self.read_claim_lock: diff --git a/agent/tests/test_agent.py b/agent/tests/test_agent.py index 0271dd6..b4c0477 100644 --- a/agent/tests/test_agent.py +++ b/agent/tests/test_agent.py @@ -57,6 +57,115 @@ class Response(io.BytesIO): self.headers = headers or {} +class JournalRetentionTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.path = Path(self.temp.name) / "operations.sqlite3" + self.journal = Journal(self.path) + self.now = 2000000 + self.sequence = 0 + + def tearDown(self): + self.journal.close() + self.temp.cleanup() + + def record(self, kind="files.stat", payload=None, result=None, state="completed", error=None, at=None, confirmed=True): + self.sequence += 1 + operation = {"id": f"operation-{self.sequence}", "kind": kind, "payload": payload or {}} + self.journal.accept(operation) + with patch.object(module.time, "time", return_value=self.now - 172800 if at is None else at): + row = self.journal.update(operation["id"], state, result, error) + if confirmed: + self.journal.confirm_completed(row) + return row + + def prune(self, days=1): + with patch.object(module.time, "time", return_value=self.now): + return self.journal.prune_observations(days) + + def test_cutoff_requires_an_older_confirmed_success(self): + old = self.record(at=self.now - 86401) + boundary = self.record(at=self.now - 86400) + recent = self.record(at=self.now - 10) + unconfirmed = self.record(confirmed=False) + self.assertEqual(self.prune(), 1) + with self.assertRaisesRegex(AgentError, "Unknown operation"): + self.journal.get(old["id"]) + for row in (boundary, recent, unconfirmed): + self.assertEqual(self.journal.get(row["id"])["state"], "completed") + + def test_zero_and_invalid_retention_preserve_every_record(self): + row = self.record() + for days in (0, -1, 36501, True, False, None, "1", 1.0, 1.5): + with self.subTest(days=days): + self.assertEqual(self.prune(days), 0) + self.assertEqual(self.journal.get(row["id"])["state"], "completed") + + def test_all_read_kinds_are_eligible_but_mutations_keep_their_replay_journal(self): + for kind in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"): + self.record(kind, result={"healthy": True}) + for command in ("status", "getstatus", "version"): + self.record("rcon.execute", {"command": command}, {"ok": True}) + mutations = [self.record(kind) for kind in ("instance.restart", "instance.stop", "instance.create", "files.stage", "files.activate", "credentials.rotate")] + mutations.append(self.record("rcon.execute", {"command": "map_restart"}, {"ok": True})) + self.assertEqual(self.prune(), 10) + for row in mutations: + saved, accepted = self.journal.accept(json.loads(row["request"])) + self.assertFalse(accepted) + self.assertEqual(saved["id"], row["id"]) + + def test_errors_failures_and_unfinished_operations_are_retained(self): + retained = [self.record(result=result) for result in ({"ok": False}, {"healthy": False}, {"error": "partial result"}, {"error": False}, {"error": []})] + retained.append(self.record(error="read did not finish")) + retained.extend(self.record(state=state) for state in ("accepted", "running", "unknown", "failed")) + self.assertEqual(self.prune(), 0) + for row in retained: + self.assertEqual(self.journal.get(row["id"])["state"], row["state"]) + + def test_old_receipt_cannot_confirm_a_changed_or_reexecuted_result(self): + old = self.record(confirmed=False) + with patch.object(module.time, "time", return_value=self.now - 90000): + current = self.journal.update(old["id"], "completed", {"new": "result"}) + self.journal.confirm_completed(old) + self.assertIsNone(self.journal.get(old["id"])["reported_at"]) + self.assertEqual(self.prune(), 0) + self.journal.confirm_completed(current) + self.assertEqual(self.prune(), 1) + + def test_pruning_uses_bounded_indexed_batches(self): + template = self.record() + self.journal.db.executemany("""INSERT INTO operations(id,fingerprint,request,state,result,updated,reported_at,retention_eligible) + VALUES(?,?,?,'completed',NULL,?,?,1)""", [ + (f"batch-{index}", template["fingerprint"], template["request"], self.now - 172800, self.now - 172800) + for index in range(504) + ]) + self.journal.db.commit() + self.assertEqual(self.prune(), 500) + self.assertEqual(self.journal.db.execute("SELECT COUNT(*) FROM operations").fetchone()[0], 5) + self.assertEqual(self.prune(), 5) + plan = self.journal.db.execute("""EXPLAIN QUERY PLAN SELECT id FROM operations + WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1 AND updated = { deployKeepDays: 'Campaign history retention', + observationKeepDays: 'Routine operation history retention', restartConcurrency: 'Concurrent restarts', createConcurrency: 'Concurrent server startups', destroyConcurrency: 'Concurrent server destructions' @@ -27,11 +30,14 @@ export function validateFleetSettings(input: unknown): FleetSettings { } for (const key of Object.keys(defaultFleetSettings) as (keyof FleetSettings)[]) { const value = values[key]; - if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 1) { - throw new Error(`${fleetSettingLabels[key]} must be a positive whole number.`); + const minimum = key === 'observationKeepDays' ? 0 : 1; + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < minimum) { + throw new Error(`${fleetSettingLabels[key]} must be ${minimum === 0 ? 'zero or a positive whole number' : 'a positive whole number'}.`); + } + if ((key === 'deployKeepDays' || key === 'observationKeepDays') && value > 36500) { + throw new Error(`${fleetSettingLabels[key]} must be at most 36500 days.`); } - if (key === 'deployKeepDays' && value > 36500) throw new Error('Campaign history retention must be at most 36500 days.'); } - return { deployKeepDays: values.deployKeepDays as number, restartConcurrency: values.restartConcurrency as number, + return { deployKeepDays: values.deployKeepDays as number, observationKeepDays: values.observationKeepDays as number, restartConcurrency: values.restartConcurrency as number, createConcurrency: values.createConcurrency as number, destroyConcurrency: values.destroyConcurrency as number }; } diff --git a/src/lib/server/db/fleet-settings.ts b/src/lib/server/db/fleet-settings.ts index e27f807..b9c71f6 100644 --- a/src/lib/server/db/fleet-settings.ts +++ b/src/lib/server/db/fleet-settings.ts @@ -8,8 +8,9 @@ export function initializeFleetSettings(handle: Database.Database): void { const saved = handle.prepare('SELECT value FROM settings WHERE key = ?').get(FLEET_SETTINGS_KEY) as { value: string } | undefined; if (saved) { const settings = JSON.parse(saved.value); - if (!Object.hasOwn(settings, 'destroyConcurrency')) { - const upgraded = validateFleetSettings({ ...settings, destroyConcurrency: defaultFleetSettings.destroyConcurrency }); + if (!Object.hasOwn(settings, 'destroyConcurrency') || !Object.hasOwn(settings, 'observationKeepDays')) { + const upgraded = validateFleetSettings({ destroyConcurrency: defaultFleetSettings.destroyConcurrency, + observationKeepDays: defaultFleetSettings.observationKeepDays, ...settings }); handle.prepare('UPDATE settings SET value = ? WHERE key = ?').run(JSON.stringify(upgraded), FLEET_SETTINGS_KEY); } return; diff --git a/src/lib/server/hosts/index.ts b/src/lib/server/hosts/index.ts index 6cee07e..d1b32be 100644 --- a/src/lib/server/hosts/index.ts +++ b/src/lib/server/hosts/index.ts @@ -4,6 +4,8 @@ import { createHash, randomBytes, randomUUID, timingSafeEqual } from 'node:crypt import { isIP } from 'node:net'; import { db } from '../db/index.js'; import { ensureHostSchema } from './schema.js'; +import { successfulObservation } from './observations.js'; +import { fleetSettings } from '../settings.js'; import { recordInstanceResources, validateInstanceResources } from './calibration.js'; import { refreshHostAvailability, scheduleAvailabilityRefresh } from './availability.js'; import type { Host, HostDecommission, HostDeletion, HostInventory, HostMode, HostOperation, HostOperationHistoryPage, HostOperationState } from '../../hosts.js'; @@ -372,6 +374,24 @@ export function listHostOperationHistory(page = 1): HostOperationHistoryPage { return { operations, page: currentPage, pageSize, total, totalPages }; })(); } + +/** 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 row.id); + const placeholders = ids.map(() => '?').join(','); + handle.prepare(`DELETE FROM ec_host_operation_events WHERE operation_id IN (${placeholders})`).run(...ids); + handle.prepare(`DELETE FROM ec_host_operation_payloads WHERE operation_id IN (${placeholders})`).run(...ids); + return handle.prepare(`DELETE FROM ec_host_operations WHERE id IN (${placeholders})`).run(...ids).changes; + })(); +} /** 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 @@ -479,7 +499,10 @@ export function reportHostOperationStatus(hostId: string, id: string, input: { s const error = input.error === undefined || input.error === null ? null : String(input.error).slice(0, 4000); const result = input.result === undefined ? null : JSON.stringify(input.result); if (result !== null && Buffer.byteLength(result) > 128 * 1024 * 1024) throw new Error('Operation result exceeds the limit.'); - database().prepare('UPDATE ec_host_operations SET state=?,result=?,error=?,updated_at=? WHERE id=?').run(state, result, error, now(), id); + const observation = state === 'completed' && !error && successfulObservation(operation.kind, + operation.kind === 'rcon.execute' ? JSON.parse(getHostOperationPayload(hostId, id).toString('utf8')) : {}, input.result); + database().prepare('UPDATE ec_host_operations SET state=?,result=?,error=?,updated_at=?,observation_success=? WHERE id=?') + .run(state, result, error, now(), Number(observation), id); if (operation.kind === 'host.decommission') { if (state === 'failed') database().prepare("UPDATE ec_host_decommissions SET stage='failed',error=?,updated_at=? WHERE operation_id=?").run(error ?? 'The agent could not start cleanup. Repair the installation, then retry deletion.', now(), id); else database().prepare("UPDATE ec_host_decommissions SET stage='running',updated_at=? WHERE operation_id=? AND stage IN ('queued','running')").run(now(), id); diff --git a/src/lib/server/hosts/observations.ts b/src/lib/server/hosts/observations.ts new file mode 100644 index 0000000..6466a95 --- /dev/null +++ b/src/lib/server/hosts/observations.ts @@ -0,0 +1,18 @@ +export const observationKinds = ['host.inspect', 'instance.inspect', 'files.read', 'files.list', 'files.stat', 'files.manifest', 'logs.read']; +const commands = ['status', 'getstatus', 'version']; + +export function successfulObservation(kind: string, payload: Record, result: unknown): boolean { + const routine = observationKinds.includes(kind) || kind === 'rcon.execute' && commands.includes(String(payload.command)); + const value = result && typeof result === 'object' ? result as Record : {}; + return routine && value.ok !== false && value.healthy !== false && (value.error == null || value.error === ''); +} + +/** Uses the operation alias o; serialized payloads may live outside the routing row. */ +export const routineObservationSql = `(o.kind IN (${observationKinds.map(kind => `'${kind}'`).join(',')}) + OR (o.kind='rcon.execute' AND json_extract(COALESCE( + (SELECT p.payload FROM ec_host_operation_payloads p WHERE p.operation_id=o.id),o.payload),'$.command') IN ('status','getstatus','version')))`; + +export const successfulObservationSql = `(o.state='completed' AND COALESCE(o.error,'')='' AND ${routineObservationSql} + AND CASE WHEN o.result IS NULL THEN 1 WHEN json_valid(o.result) + THEN json_extract(o.result,'$.ok') IS NOT 0 AND json_extract(o.result,'$.healthy') IS NOT 0 + AND COALESCE(json_extract(o.result,'$.error'),'')='' ELSE 0 END)`; diff --git a/src/lib/server/hosts/schema.ts b/src/lib/server/hosts/schema.ts index b3d15a0..6c2ae47 100644 --- a/src/lib/server/hosts/schema.ts +++ b/src/lib/server/hosts/schema.ts @@ -1,4 +1,5 @@ import type Database from 'better-sqlite3'; +import { successfulObservationSql } from './observations.js'; const initialized = new WeakSet(); export function ensureHostSchema(handle: Database.Database): void { @@ -28,6 +29,7 @@ export function ensureHostSchema(handle: Database.Database): void { payload TEXT NOT NULL, state TEXT NOT NULL DEFAULT 'pending' CHECK (state IN ('pending','accepted','running','completed','failed','unknown')), result TEXT, error TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, last_delivered_at TEXT, + observation_success INTEGER NOT NULL DEFAULT 0, UNIQUE(host_id, idempotency_key) ); CREATE INDEX IF NOT EXISTS ec_host_operations_queue ON ec_host_operations(host_id, state, created_at); @@ -45,6 +47,7 @@ export function ensureHostSchema(handle: Database.Database): void { id INTEGER PRIMARY KEY AUTOINCREMENT, operation_id TEXT NOT NULL REFERENCES ec_host_operations(id), state TEXT NOT NULL, created_at TEXT NOT NULL, error TEXT ); + CREATE INDEX IF NOT EXISTS ec_host_operation_events_operation ON ec_host_operation_events(operation_id); CREATE TABLE IF NOT EXISTS ec_host_ports ( host_id TEXT NOT NULL REFERENCES ec_hosts(id), instance_id TEXT NOT NULL, bind_address TEXT NOT NULL, protocol TEXT NOT NULL CHECK (protocol IN ('udp','tcp')), @@ -82,6 +85,14 @@ 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 === '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(() => { + handle.exec('ALTER TABLE ec_host_operations ADD COLUMN observation_success INTEGER NOT NULL DEFAULT 0'); + handle.exec(`UPDATE ec_host_operations AS o SET observation_success=1 WHERE ${successfulObservationSql}`); + })(); + } + handle.exec(`CREATE INDEX IF NOT EXISTS ec_host_operations_observation_retention ON ec_host_operations(updated_at) + WHERE observation_success=1 AND state='completed'`); const hostColumns = handle.prepare('PRAGMA table_info(ec_hosts)').all() as { name: string }[]; if (!hostColumns.some(column => column.name === 'deleted_at')) handle.exec('ALTER TABLE ec_hosts ADD COLUMN deleted_at TEXT'); const allowanceColumns = handle.prepare('PRAGMA table_info(ec_resource_allowances)').all() as { name: string }[]; diff --git a/src/lib/server/pipeline/sweep.ts b/src/lib/server/pipeline/sweep.ts index 5c7a330..525f7b6 100644 --- a/src/lib/server/pipeline/sweep.ts +++ b/src/lib/server/pipeline/sweep.ts @@ -2,6 +2,7 @@ import { fleetSettings } from '../settings.js'; import { db } from '../db/index.js'; import { emit } from './bus.js'; import { requestRestart } from '../provisioning/index.js'; +import { pruneHostObservations } from '../hosts/index.js'; interface PendingRow { deployment_id: string; server: string; restart_state: string; @@ -18,6 +19,7 @@ export function sweep(): Promise { return sweepInFlight; } async function runSweep(): Promise { + 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[]; const result: SweepResult = { checked: 0, restarted: 0, healthy: 0, held: 0, failed: 0 }; diff --git a/src/routes/api/agent/heartbeat/+server.ts b/src/routes/api/agent/heartbeat/+server.ts index aa219e8..40e56b4 100644 --- a/src/routes/api/agent/heartbeat/+server.ts +++ b/src/routes/api/agent/heartbeat/+server.ts @@ -3,6 +3,7 @@ import { heartbeat, pendingCreationCancellations } from '$lib/server/hosts/index import { authenticateAgent, hostApiError, objectBody } from '$lib/server/hosts/http.js'; import { reconcileHostPreparation } from '$lib/server/hosts/preparation.js'; import { reconcileHostLibraryCleanup, cleanupControllerLibraries } from '$lib/server/game-files/index.js'; +import { fleetSettings } from '$lib/server/settings.js'; export async function POST({ request }: { request: Request }) { const authenticated = authenticateAgent(request); try { @@ -16,6 +17,7 @@ export async function POST({ request }: { request: Request }) { reconcileHostPreparation(host.id); reconcileHostLibraryCleanup(host.id); await cleanupControllerLibraries(); - return json({ host, cancelledCreations: pendingCreationCancellations(host.id), heartbeatSeconds: 15, serverTime: new Date().toISOString() }); + return json({ host, cancelledCreations: pendingCreationCancellations(host.id), heartbeatSeconds: 15, + observationKeepDays: fleetSettings().observationKeepDays, serverTime: new Date().toISOString() }); } catch (cause) { return hostApiError(cause); } } diff --git a/src/routes/settings/+page.svelte b/src/routes/settings/+page.svelte index 649203f..decf18c 100644 --- a/src/routes/settings/+page.svelte +++ b/src/routes/settings/+page.svelte @@ -54,6 +54,9 @@

Default: 7 days. Removes eligible old campaign records and file diffs on the next cleanup check. Pending, failed, held, and health-recovery-linked campaigns are kept. Provisioning history is retained.

+ + +

Default: 1 day. Removes successful monitoring and file-read operation records on the next cleanup check. Failures and actions are retained. Set to 0 to keep these records indefinitely.

Default: 1 across the fleet. Campaigns, betas, maintenance, and lifecycle actions share recovery slots until container and game health are verified.

diff --git a/static/agent/ec-agent.py b/static/agent/ec-agent.py index 8b0373c..187ad6b 100644 --- a/static/agent/ec-agent.py +++ b/static/agent/ec-agent.py @@ -32,7 +32,7 @@ import urllib.parse import urllib.request import uuid -VERSION = "1.4.8" +VERSION = "1.4.9" LABEL = "io.electric-cherry" MAX_FILE = 32 * 1024 * 1024 MANIFEST_MAX_ENTRIES = 10000 @@ -516,6 +516,19 @@ class ReadWriteGate: self.condition.notify_all() +def read_operation(operation): + if not isinstance(operation, dict): + return False + if operation.get("kind") in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"): + return True + payload = operation.get("payload") + return operation.get("kind") == "rcon.execute" and isinstance(payload, dict) and payload.get("command") in ("status", "getstatus", "version") + + +def valid_observation_keep_days(value): + return type(value) is int and 0 <= value <= 36500 + + class Journal: def __init__(self, path, recover=True): self.db = sqlite3.connect(path) @@ -525,8 +538,14 @@ class Journal: self.db.execute("PRAGMA journal_mode=WAL") self.db.execute("PRAGMA synchronous=FULL") self.db.execute("CREATE TABLE IF NOT EXISTS operations (id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)") - if "checkpoint" not in {row[1] for row in self.db.execute("PRAGMA table_info(operations)")}: + columns = {row[1] for row in self.db.execute("PRAGMA table_info(operations)")} + if "checkpoint" not in columns: self.db.execute("ALTER TABLE operations ADD COLUMN checkpoint TEXT") + if "reported_at" not in columns: + self.db.execute("ALTER TABLE operations ADD COLUMN reported_at REAL") + if "retention_eligible" not in columns: + self.db.execute("ALTER TABLE operations ADD COLUMN retention_eligible INTEGER NOT NULL DEFAULT 0") + self.db.execute("CREATE INDEX IF NOT EXISTS operations_observation_retention ON operations(updated,id) WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1") self.db.execute("UPDATE operations SET state='unknown', error='Agent restarted during execution; inspect before reconciliation', updated=? WHERE state='running'", (time.time(),)) self.db.commit() @@ -551,10 +570,37 @@ class Journal: return dict(row) def update(self, op_id, state, result=None, error=None): - self.db.execute("UPDATE operations SET state=?, result=?, error=?, updated=? WHERE id=?", (state, json.dumps(result) if result is not None else None, error, time.time(), op_id)) + self.db.execute("UPDATE operations SET state=?, result=?, error=?, updated=?, reported_at=NULL, retention_eligible=0 WHERE id=?", (state, json.dumps(result) if result is not None else None, error, time.time(), op_id)) self.db.commit() return self.get(op_id) + def confirm_completed(self, row): + """Only an acknowledged, unchanged completion can become disposable.""" + if row["state"] != "completed": + return + eligible = False + try: + result = json.loads(row["result"]) if row.get("result") is not None else None + eligible = read_operation(json.loads(row["request"])) and not row.get("error") and not ( + isinstance(result, dict) and (result.get("ok") is False or result.get("healthy") is False or result.get("error") not in (None, "")) + ) + except (ValueError, TypeError, KeyError): + pass # Malformed historical records are retained for inspection. + self.db.execute("""UPDATE operations SET reported_at=COALESCE(reported_at,?), retention_eligible=? + WHERE id=? AND state='completed' AND updated=? AND result IS ? AND error IS ?""", + (time.time(), int(bool(eligible)), row["id"], row["updated"], row.get("result"), row.get("error"))) + self.db.commit() + + def prune_observations(self, keep_days): + if not valid_observation_keep_days(keep_days) or keep_days == 0: + return 0 + cutoff = time.time() - keep_days * 86400 + removed = self.db.execute("""DELETE FROM operations WHERE id IN ( + SELECT id FROM operations WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1 + AND updated= next_prune: + next_prune = time.monotonic() + 60 + with self.activity(read_only=True) as acquired: + if acquired and not self.update_pending(): + if journal.prune_observations(self.observation_keep_days) == 500: + next_prune = time.monotonic() + 15 response = self.request("GET", f"/api/agent/operations?lane={lane}&wait=20") handled = False for operation in response.get("operations", []): @@ -2242,7 +2302,7 @@ class Agent: continue row, _ = journal.accept(operation) if row["state"] in ("accepted", "unknown", "running"): - self.report(journal.update(operation["id"], "running")) + self.report(journal.update(operation["id"], "running"), journal=journal) with self.read_gate.read(), self.instance_lock(operation): try: result = self.dispatch(operation) @@ -2251,7 +2311,7 @@ class Agent: row = journal.update(operation["id"], "failed", error=str(err) if isinstance(err, AgentError) else "Observation unavailable") # Receipt delivery can take seconds; it must not # hold the instance lock needed by health checks. - self.report(row) + self.report(row, journal=journal) handled = True finally: with self.read_claim_lock: diff --git a/static/agent/manifest.json b/static/agent/manifest.json index a2031f1..e910cce 100644 --- a/static/agent/manifest.json +++ b/static/agent/manifest.json @@ -1,8 +1,8 @@ { - "version": "1.4.8", + "version": "1.4.9", "protocolVersion": 1, "platform": "debian-amd64", - "sha256": "45eb0ef619fb0ceef2ac843657fd7fdaf8cfa610951f0ddfabdbf5942abc6e8c", + "sha256": "c0a02509a8307b2abb75e5450bf3f39596252f26e041ed938d662bc7127bec62", "updaterProtocolVersion": 1, "updaterSha256": "066bb896726600ece9ce72697ce523c9a707cc9a621505931ba0892dec59d3c8" } diff --git a/tests/unit/fleet-settings.test.ts b/tests/unit/fleet-settings.test.ts index 5c23a30..923423f 100644 --- a/tests/unit/fleet-settings.test.ts +++ b/tests/unit/fleet-settings.test.ts @@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { newFixture, type Fixture } from '../helpers/fixture.js'; let fixture: Fixture; -const defaults = { deployKeepDays: 7, restartConcurrency: 1, createConcurrency: 1, destroyConcurrency: 1 }; +const defaults = { deployKeepDays: 7, observationKeepDays: 1, restartConcurrency: 1, createConcurrency: 1, destroyConcurrency: 1 }; beforeEach(async () => { vi.resetModules(); fixture = await newFixture([]); @@ -14,7 +14,7 @@ describe('fleet settings API and persistence', () => { it('returns defaults and saves all settings across a module reload and database reopen', async () => { const { GET, PUT } = await import('../../src/routes/api/settings/+server.js'); expect(await GET().json()).toEqual(defaults); - const saved = { deployKeepDays: 30, restartConcurrency: 3, createConcurrency: 2, destroyConcurrency: 4 }; + const saved = { deployKeepDays: 30, observationKeepDays: 5, restartConcurrency: 3, createConcurrency: 2, destroyConcurrency: 4 }; const response = await PUT({ request: new Request('http://localhost/api/settings', { method: 'PUT', body: JSON.stringify(saved) }) }); expect(response.status).toBe(200); expect(await response.json()).toEqual(saved); @@ -37,7 +37,7 @@ describe('fleet settings API and persistence', () => { vi.stubEnv('DEPLOYMENT_RESTART_CONCURRENCY', '4'); vi.stubEnv('DEPLOYMENT_CREATE_CONCURRENCY', '2'); const { fleetSettings } = await import('../../src/lib/server/settings.js'); - expect(fleetSettings()).toEqual({ deployKeepDays: 14, restartConcurrency: 4, createConcurrency: 2, destroyConcurrency: 1 }); + expect(fleetSettings()).toEqual({ deployKeepDays: 14, observationKeepDays: 1, restartConcurrency: 4, createConcurrency: 2, destroyConcurrency: 1 }); closeDb(); vi.stubEnv('DEPLOY_KEEP_DAYS', '7'); expect(fleetSettings().deployKeepDays).toBe(14); @@ -53,18 +53,47 @@ describe('fleet settings API and persistence', () => { vi.stubEnv('DEPLOYMENT_RESTART_CONCURRENCY', '9'); vi.stubEnv('DEPLOYMENT_CREATE_CONCURRENCY', '8'); const { GET } = await import('../../src/routes/api/settings/+server.js'); - const upgraded = { ...previous, destroyConcurrency: 1 }; + const upgraded = { ...previous, destroyConcurrency: 1, observationKeepDays: 1 }; expect(await GET().json()).toEqual(upgraded); expect(JSON.parse((db().prepare("SELECT value FROM settings WHERE key = 'fleet_settings'").get() as { value: string }).value)).toEqual(upgraded); closeDb(); expect(await GET().json()).toEqual(upgraded); }); + it('adds routine observation retention to modern saved settings without changing any existing values', async () => { + const { db, closeDb } = await import('../../src/lib/server/db/index.js'); + const previous = { deployKeepDays: 45, restartConcurrency: 3, createConcurrency: 2, destroyConcurrency: 4 }; + db().prepare("UPDATE settings SET value = ? WHERE key = 'fleet_settings'").run(JSON.stringify(previous)); + db().prepare("INSERT INTO settings(key,value) VALUES('unrelated','keep')").run(); + closeDb(); + vi.stubEnv('DEPLOY_KEEP_DAYS', 'bad'); + const { GET } = await import('../../src/routes/api/settings/+server.js'); + const upgraded = { ...previous, observationKeepDays: 1 }; + expect(await GET().json()).toEqual(upgraded); + expect(JSON.parse((db().prepare("SELECT value FROM settings WHERE key = 'fleet_settings'").get() as { value: string }).value)).toEqual(upgraded); + expect(db().prepare("SELECT value FROM settings WHERE key='unrelated'").get()).toEqual({ value: 'keep' }); + closeDb(); + expect(await GET().json()).toEqual(upgraded); + }); + + it.each([0, 36500])('persists routine observation retention of %s days across database reopen', async observationKeepDays => { + const { GET, PUT } = await import('../../src/routes/api/settings/+server.js'); + const saved = { ...defaults, observationKeepDays }; + const response = await PUT({ request: new Request('http://localhost/api/settings', { method: 'PUT', body: JSON.stringify(saved) }) }); + expect(response.status).toBe(200); + expect(await response.json()).toEqual(saved); + const { closeDb } = await import('../../src/lib/server/db/index.js'); + closeDb(); + expect(await GET().json()).toEqual(saved); + }); + it.each([ 'null', '[]', '{}', '{', ...['deployKeepDays', 'restartConcurrency', 'createConcurrency', 'destroyConcurrency'].flatMap(key => [0, -1, 1.5, '2', true, null, 9007199254740992].map(value => JSON.stringify({ ...defaults, [key]: value }))), + ...[-1, 1.5, '2', true, null, 36501, 9007199254740992].map(value => JSON.stringify({ ...defaults, observationKeepDays: value })), JSON.stringify({ deployKeepDays: 7, restartConcurrency: 1, createConcurrency: 1 }), + JSON.stringify({ deployKeepDays: 7, restartConcurrency: 1, createConcurrency: 1, destroyConcurrency: 1 }), JSON.stringify({ ...defaults, deployKeepDays: 36501 }), JSON.stringify({ ...defaults, unknown: 1 }) ])('rejects invalid input without partially saving: %s', async body => { diff --git a/tests/unit/host-observation-retention.test.ts b/tests/unit/host-observation-retention.test.ts new file mode 100644 index 0000000..42e4ae0 --- /dev/null +++ b/tests/unit/host-observation-retention.test.ts @@ -0,0 +1,155 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { newFixture, type Fixture } from '../helpers/fixture.js'; + +const now = Date.parse('2026-10-05T16:00:00Z'); +const day = 86_400_000; +const at = (age: number) => new Date(now - age).toISOString(); +const inventory = { protocolVersion: 1, dockerRunning: true, directoriesWritable: true, + storageAvailable: 100 * 1024 ** 3, runtimeImages: ['plutainer:test'], libraries: [{ id: 'library', state: 'ready' }] }; +let fixture: Fixture; + +beforeEach(async () => { + vi.resetModules(); fixture = await newFixture([]); + vi.useFakeTimers({ toFake: ['Date'] }); vi.setSystemTime(now); +}); +afterEach(async () => { vi.useRealTimers(); vi.restoreAllMocks(); await fixture.cleanup(); }); + +async function setup() { + const hosts = await import('../../src/lib/server/hosts/index.js'); + const { db, closeDb } = await import('../../src/lib/server/db/index.js'); + const { fleetSettings, saveFleetSettings } = await import('../../src/lib/server/settings.js'); + const host = hosts.createHost({ name: 'Retention host' }); + const credentials = hosts.enrollHost(hosts.createEnrollment(host.id).token, inventory); + const handle = db(); + function add(kind: string, options: { state?: string; payload?: Record; result?: unknown; + error?: string; age?: number; serialized?: boolean; routing?: Record } = {}) { + const payload = options.payload ?? {}; + const operation = options.serialized + ? hosts.queueHostOperationSerialized(host.id, kind, Buffer.from(JSON.stringify(payload)), {}, options.routing ?? {}) + : hosts.queueHostOperation(host.id, kind, payload); + const state = options.state ?? 'completed'; + if (state !== 'pending') hosts.reportHostOperationStatus(host.id, operation.id, { + state, result: options.result === undefined ? { ok: true } : options.result, error: options.error + }); + handle.prepare('UPDATE ec_host_operations SET created_at=?,updated_at=? WHERE id=?') + .run(at(10 * day), at(options.age ?? 2 * day), operation.id); + return operation.id; + } + return { ...hosts, handle, db, closeDb, host, credentials, fleetSettings, saveFleetSettings, add }; +} + +describe('routine host observation retention', () => { + it('removes aged successful reads, manifests, and read-only RCON while retaining failures, actions, and recent evidence', async () => { + const api = await setup(); + const removed = ['host.inspect', 'instance.inspect', 'files.read', 'files.stat', 'files.list', 'files.manifest', 'logs.read'] + .map(kind => api.add(kind)); + removed.push(...['status', 'getstatus', 'version'].map(command => api.add('rcon.execute', { payload: { command } }))); + const retained = [ + ...['pending', 'accepted', 'running', 'failed', 'unknown'].map(state => api.add('files.read', { state })), + ...['instance.restart', 'instance.stop', 'instance.remove', 'files.stage', 'files.activate', 'files.rollback', 'runtime.pull', 'unknown.read'] + .map(kind => api.add(kind)), + api.add('rcon.execute', { payload: { command: 'map_restart' } }), + api.add('rcon.execute', { payload: { command: 'status; map_restart' } }), + api.add('rcon.execute', { payload: { command: 'status' }, result: { ok: false, reason: 'timeout' } }), + api.add('instance.inspect', { result: { healthy: false, state: 'stopped' } }), + api.add('files.read', { error: 'Read failed' }), + api.add('files.read', { result: { error: 'Read failed' } }), + api.add('files.stat', { age: day }), + api.add('files.read', { age: day - 1 }) + ]; + expect(api.pruneHostObservations()).toBe(removed.length); + const ids = (api.handle.prepare('SELECT id FROM ec_host_operations').all() as { id: string }[]).map(row => row.id); + expect(ids.sort()).toEqual(retained.sort()); + expect(api.pruneHostObservations()).toBe(0); + }); + + it('classifies full serialized RCON payloads and deletes their events and immutable bytes together', async () => { + const api = await setup(); + const read = api.add('rcon.execute', { serialized: true, payload: { command: 'status' }, routing: { command: 'map_restart' } }); + const action = api.add('rcon.execute', { serialized: true, payload: { command: 'map_restart' }, routing: { command: 'status' } }); + expect(api.handle.prepare('SELECT COUNT(*) AS count FROM ec_host_operation_payloads').get()).toEqual({ count: 2 }); + expect(api.pruneHostObservations()).toBe(1); + expect(api.getHostOperationStatus(read)).toBeNull(); + expect(api.getHostOperation(action)?.payload).toEqual({ command: 'map_restart' }); + expect(api.handle.prepare('SELECT DISTINCT operation_id FROM ec_host_operation_events').all()).toEqual([{ operation_id: action }]); + expect(api.handle.prepare('SELECT operation_id FROM ec_host_operation_payloads').all()).toEqual([{ operation_id: action }]); + expect(api.handle.pragma('foreign_key_check')).toEqual([]); + }); + + it('honors disabled retention and changes to the saved period using the last receipt time', async () => { + const api = await setup(); + const old = api.add('files.read', { age: 4 * day }); + const recentReceipt = api.add('files.read', { age: 2 * day }); + api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays: 0 }); + expect(api.pruneHostObservations()).toBe(0); + api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays: 3 }); + expect(api.pruneHostObservations()).toBe(1); + expect(api.getHostOperationStatus(old)).toBeNull(); + expect(api.getHostOperationStatus(recentReceipt)).not.toBeNull(); + api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays: 1 }); + expect(api.pruneHostObservations()).toBe(1); + }); + + it('bounds each cleanup batch and seeks through the retention index despite extensive retained action history', async () => { + const api = await setup(); + api.handle.transaction(() => { + const insert = api.handle.prepare(`INSERT INTO ec_host_operations + (id,host_id,kind,payload,state,created_at,updated_at,observation_success) + VALUES(?,?,?,'{}','completed',?,?,?)`); + for (let index = 0; index < 2000; index++) insert.run(`action-${index}`, api.host.id, 'instance.restart', at(10 * day), at(10 * day), 0); + for (let index = 0; index < 503; index++) insert.run(`read-${index}`, api.host.id, 'files.read', at(2 * day), at(2 * day), 1); + })(); + const prepare = vi.spyOn(api.handle, 'prepare'); + expect(api.pruneHostObservations()).toBe(500); + const query = prepare.mock.calls.map(([sql]) => sql).find(sql => sql.includes('SELECT id FROM ec_host_operations WHERE observation_success'))!; + prepare.mockRestore(); + const plan = api.handle.prepare(`EXPLAIN QUERY PLAN ${query}`).all(at(day)) as { detail: string }[]; + expect(plan.some(row => row.detail.includes('USING INDEX ec_host_operations_observation_retention'))).toBe(true); + expect(plan.some(row => row.detail.includes('TEMP B-TREE'))).toBe(false); + expect(api.pruneHostObservations()).toBe(3); + expect(api.pruneHostObservations()).toBe(0); + expect(api.handle.prepare("SELECT COUNT(*) AS count FROM ec_host_operations WHERE kind='instance.restart'").get()).toEqual({ count: 2000 }); + }); + + it('backfills legacy observation classification once while preserving actions and negative or invalid results', async () => { + const api = await setup(); + const read = api.add('files.manifest'); + const serializedRead = api.add('rcon.execute', { serialized: true, payload: { command: 'status' } }); + const retained = [api.add('instance.restart'), api.add('rcon.execute', { payload: { command: 'map_restart' } }), + api.add('instance.inspect', { result: { healthy: false } }), api.add('files.read', { result: { ok: false } }), + api.add('files.read', { error: 'Unavailable' }), api.add('files.read', { result: { error: 'Unavailable' } }), + api.add('files.stat', { state: 'unknown' })]; + const malformed = api.add('files.read'); + api.handle.prepare('UPDATE ec_host_operations SET result=? WHERE id=?').run('{not-json', malformed); + retained.push(malformed); + api.handle.exec('DROP INDEX ec_host_operations_observation_retention'); + api.handle.exec('ALTER TABLE ec_host_operations DROP COLUMN observation_success'); + api.closeDb(); + + expect(api.pruneHostObservations()).toBe(2); + expect(api.getHostOperationStatus(read)).toBeNull(); + expect(api.getHostOperationStatus(serializedRead)).toBeNull(); + const handle = api.db(); + expect((handle.prepare('SELECT id FROM ec_host_operations').all() as { id: string }[]).map(row => row.id).sort()).toEqual(retained.sort()); + // Subsequent boots honor the recorded classification instead of repeatedly + // deriving it from historical rows that were deliberately preserved. + handle.prepare("UPDATE ec_host_operations SET result='{}' WHERE id=?").run(malformed); + api.closeDb(); + expect(api.pruneHostObservations()).toBe(0); + expect(api.getHostOperationStatus(malformed)).not.toBeNull(); + }); + + it('returns the current shared retention setting to an agent on every heartbeat', async () => { + const api = await setup(); + const { POST } = await import('../../src/routes/api/agent/heartbeat/+server.js'); + for (const observationKeepDays of [7, 0]) { + api.saveFleetSettings({ ...api.fleetSettings(), observationKeepDays }); + const response = await POST({ request: new Request('https://ec.test/api/agent/heartbeat', { + method: 'POST', headers: { authorization: `Bearer ${api.credentials.credential}`, 'Content-Type': 'application/json' }, + body: JSON.stringify({ inventory }) + }) }); + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ observationKeepDays, heartbeatSeconds: 15 }); + } + }); +});