diff --git a/src/lib/server/db/index.ts b/src/lib/server/db/index.ts index 90e66d7..812fac3 100644 --- a/src/lib/server/db/index.ts +++ b/src/lib/server/db/index.ts @@ -353,6 +353,8 @@ export function db(): Database.Database { migrateTable(handle, 'beta_test_runs', [{ name: 'revision', ddl: 'ALTER TABLE beta_test_runs ADD COLUMN revision INTEGER NOT NULL DEFAULT 1' }]); migrateTable(handle, 'deployment_scripts', DEPLOYMENT_SCRIPTS_COLUMNS); migrateTable(handle, 'deployments', DEPLOYMENTS_COLUMNS); + // Older databases gain rolled_back_by above, after schema.sql runs. + handle.exec('CREATE INDEX IF NOT EXISTS idx_deployments_rolled_back_by ON deployments(rolled_back_by)'); migrateTable(handle, 'server_launch_collections', [ { name: 'mode', ddl: "ALTER TABLE server_launch_collections ADD COLUMN mode TEXT NOT NULL DEFAULT 'fill'" }, { name: 'retained_servers', ddl: "ALTER TABLE server_launch_collections ADD COLUMN retained_servers TEXT NOT NULL DEFAULT '[]'" } diff --git a/src/lib/server/db/schema.sql b/src/lib/server/db/schema.sql index 76de4c5..2abdea6 100644 --- a/src/lib/server/db/schema.sql +++ b/src/lib/server/db/schema.sql @@ -522,6 +522,7 @@ CREATE TABLE IF NOT EXISTS provisioning_sources ( server TEXT NOT NULL, PRIMARY KEY (operation_id, kind, source_id, server) ); +CREATE INDEX IF NOT EXISTS idx_provisioning_sources_source ON provisioning_sources(kind, source_id); -- Blueprints retain map/count selections only. The legacy column name remains -- compatible with older snapshot-based definitions, read as selections at runtime. @@ -732,6 +733,7 @@ CREATE TABLE IF NOT EXISTS watchdog_actions ( cycle INTEGER NOT NULL DEFAULT 1, UNIQUE(incident_id,cycle,action) ); +CREATE INDEX IF NOT EXISTS idx_watchdog_actions_campaign ON watchdog_actions(campaign_id); -- Explicit history holds and failure resolution never bypass dependency checks. CREATE TABLE IF NOT EXISTS retention_controls ( diff --git a/src/lib/server/pipeline/sweep.ts b/src/lib/server/pipeline/sweep.ts index e308de0..645145a 100644 --- a/src/lib/server/pipeline/sweep.ts +++ b/src/lib/server/pipeline/sweep.ts @@ -1,8 +1,6 @@ -import { pruneHistory } from '../retention.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; @@ -20,7 +18,6 @@ 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') 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[]; @@ -47,7 +44,6 @@ async function runSweep(): Promise { } 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()); - pruneHistory('campaign'); emit({ type: 'sweep-tick', frozen: false }); return result; } diff --git a/src/lib/server/pipeline/worker.ts b/src/lib/server/pipeline/worker.ts index 6c86f01..4d8b870 100644 --- a/src/lib/server/pipeline/worker.ts +++ b/src/lib/server/pipeline/worker.ts @@ -21,6 +21,7 @@ import { trackActivity } from './activity.js'; import { withOperationSource } from '../hosts/operation-context.js'; const TICK_MS = 15_000; +const RETENTION_INTERVAL_MS = 5 * 60_000; // Match each agent's two health lanes without making unrelated hosts wait. const STATUS_CONCURRENCY_PER_HOST = 2; @@ -31,6 +32,7 @@ let statusTimer: NodeJS.Timeout | null = null; let tickInFlight: Promise | null = null; let tickRequested = false; const requesterFlights = new Map>(); +let nextRetentionAt = 0; function tick(requestAnother = false): Promise { if (requestAnother) tickRequested = true; @@ -57,6 +59,10 @@ function reconcileRequesters(): void { ]; for (const [key, label, work] of requesters) { if (requesterFlights.has(key)) continue; + if (key === 'retention') { + if (Date.now() < nextRetentionAt) continue; + nextRetentionAt = Date.now() + RETENTION_INTERVAL_MS; + } const flight = trackActivity(key, label, work) .catch(err => { console.error(`[worker] ${key} error:`, err); }) .then(async () => { if (key !== 'game-files' && key !== 'retention') await tick(true); }) diff --git a/src/lib/server/retention.ts b/src/lib/server/retention.ts index d0569d1..1809c24 100644 --- a/src/lib/server/retention.ts +++ b/src/lib/server/retention.ts @@ -108,16 +108,24 @@ 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 age = ['campaign','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) 0 && policy.failed) candidates.push(`SELECT o.rowid FROM retention_controls c + JOIN ${policy.table} o ON o.${policy.id}=c.record_id + WHERE c.category=@category AND julianday(c.resolved_at)=julianday(o.${policy.clock}) AND (${failed})`); 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}) + return { sql: `FROM ${policy.table} o WHERE ${candidates.length ? `o.rowid IN (${candidates.join(' UNION ALL ')})` : '0'} AND (${policy.eligible}) 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) } }; diff --git a/tests/unit/retention.test.ts b/tests/unit/retention.test.ts index 51b7588..4d4bd28 100644 --- a/tests/unit/retention.test.ts +++ b/tests/unit/retention.test.ts @@ -22,6 +22,53 @@ it('previews eligible campaigns, respects latest-N and explicit holds, and treat a.saveFleetSettings({ ...a.fleetSettings(),deployKeepDays: 0,historyKeepCount: 0 }); expect(a.pruneHistory('campaign')).toBe(0); }); +it('seeks expired campaigns and uses indexed dependency checks even with failure retention enabled', async () => { + const a = await api(); + const now = new Date().toISOString(); + const insert = a.db.prepare("INSERT INTO deployments(id,action,started_at,finished_at) VALUES(?,'drift',?,?)"); + a.db.transaction(() => { + for (let n=0;n<8000;n++) insert.run(`recent-${n}`,now,now); + })(); + a.saveFleetSettings({...a.fleetSettings(),deployKeepDays:7,failureKeepDays:7}); + const prepare = vi.spyOn(a.db,'prepare'); + let sql: string; + try { + expect(a.pruneHistory('campaign')).toBe(0); + sql = prepare.mock.calls.find(([sql]) => sql.startsWith('SELECT o.rowid AS rid'))![0]; + } finally { prepare.mockRestore(); } + const plan = a.db.prepare(`EXPLAIN QUERY PLAN ${sql}`).all({category:'campaign',cutoff:old,failureCutoff:old}) as {detail:string}[]; + const details = plan.map(row=>row.detail).join('\n'); + expect(details).toMatch(/SEARCH o USING (?:COVERING )?INDEX idx_deployments_started \(started_at<\?\)/); + for (const index of ['idx_deployments_rolled_back_by','idx_watchdog_actions_campaign','idx_provisioning_sources_source']) { + expect(details).toContain(index); + } + expect(details).not.toMatch(/SCAN [odas]\b/); +}); +it('selects resolved failures separately while preserving age, protection and batch limits', async () => { + const a = await api(); + const now = new Date().toISOString(); + for (const id of ['normal','unresolved','resolved','held','new-resolution','changed']) { + a.db.prepare("INSERT INTO deployments(id,action,started_at,finished_at) VALUES(?,'drift',?,?)").run(id,old,old); + if (id !== 'normal') a.db.prepare("INSERT INTO deployment_servers(deployment_id,server,files_changed,restart_state) VALUES(?,'s',1,'failed')").run(id); + } + for (const id of ['resolved','held','new-resolution','changed']) { + a.setRetentionControl({category:'campaign',id,resolve:true,protected:id==='held'}); + if (id !== 'new-resolution') a.db.prepare("UPDATE retention_controls SET resolved_at=? WHERE record_id=?").run(old,id); + } + a.db.prepare("UPDATE deployments SET started_at=? WHERE id='changed'").run(now); + // Failure retention works independently of normal campaign retention. + a.saveFleetSettings({...a.fleetSettings(),deployKeepDays:0,failureKeepDays:7}); + expect(a.retentionPreview().find(p=>p.category==='campaign')?.records).toEqual([{id:'resolved'}]); + expect(a.pruneHistory('campaign')).toBe(1); + a.saveFleetSettings({...a.fleetSettings(),deployKeepDays:7}); + const insert = a.db.prepare("INSERT INTO deployments(id,action,started_at) VALUES(?,'drift',?)"); + a.db.transaction(() => { for (let n=0;n<600;n++) insert.run(`old-${n}`,old); })(); + expect(a.pruneHistory('campaign')).toBe(500); + expect(a.pruneHistory('campaign')).toBe(101); + expect(a.db.prepare('SELECT id FROM deployments ORDER BY id').all()).toEqual([ + {id:'changed'},{id:'held'},{id:'new-resolution'},{id:'unresolved'} + ]); +}); it('expires file bodies without silently turning rollback into file deletion', async () => { const a = await api(); a.db.prepare("INSERT INTO deployments(id,action,started_at,finished_at) VALUES('p','promote',?,?)").run(old,old); diff --git a/tests/unit/sweep.test.ts b/tests/unit/sweep.test.ts index ab0ffc8..989c1c7 100644 --- a/tests/unit/sweep.test.ts +++ b/tests/unit/sweep.test.ts @@ -16,6 +16,16 @@ afterEach(async () => { }); describe('sweep', () => { + it('leaves history cleanup to the retention requester', async () => { + const retention = await import('../../src/lib/server/retention.js'); + const prune = vi.spyOn(retention,'pruneHistory'); + try { + const { sweep } = await import('../../src/lib/server/pipeline/sweep.js'); + await sweep(); + expect(prune).not.toHaveBeenCalled(); + } finally { prune.mockRestore(); } + }); + it('emits server state only when state, players, or error changes', async () => { const { sweep } = await import('../../src/lib/server/pipeline/sweep.js'); const { seedServerRegistry } = await import('../../src/lib/server/fleet/servers.js'); diff --git a/tests/unit/worker-scheduling.test.ts b/tests/unit/worker-scheduling.test.ts index 6a392ce..af12164 100644 --- a/tests/unit/worker-scheduling.test.ts +++ b/tests/unit/worker-scheduling.test.ts @@ -6,6 +6,7 @@ vi.mock('../../src/lib/server/pipeline/beta-runner.js', () => ({ betaTick: vi.fn vi.mock('../../src/lib/server/maintenance/index.js', () => ({ runMaintenance: vi.fn() })); vi.mock('../../src/lib/server/game-files/index.js', () => ({ cleanupControllerLibraries: vi.fn(async () => {}) })); vi.mock('../../src/lib/server/provisioning/worker.js', () => ({ processProvisioning: vi.fn(async () => {}) })); +vi.mock('../../src/lib/server/retention.js', () => ({ pruneHistory: vi.fn(), pruneSampleHistory: vi.fn() })); function deferred() { let resolve!: () => void; @@ -31,6 +32,30 @@ async function setup() { } describe('independent provisioning scheduling', () => { + it('runs retention at most every five minutes across ticks and explicit kicks', async () => { + const s = await setup(); + const { pruneHistory, pruneSampleHistory } = await import('../../src/lib/server/retention.js'); + vi.useFakeTimers(); + try { + s.startWorker(); + expect(pruneHistory).toHaveBeenCalledTimes(1); + expect(pruneSampleHistory).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(285_000); + await s.kick(); + expect(pruneHistory).toHaveBeenCalledTimes(1); + expect(pruneSampleHistory).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(15_000); + expect(pruneHistory).toHaveBeenCalledTimes(2); + expect(pruneSampleHistory).toHaveBeenCalledTimes(2); + expect(s.process.mock.calls.length).toBeGreaterThan(2); + } finally { + s.stopWorker(); s.background.resolve(); + await vi.advanceTimersByTimeAsync(0); + vi.useRealTimers(); + await vi.waitFor(() => expect(s.workerActivity()).toEqual([])); + } + }); + it('runs controller library cleanup on worker ticks without overlapping a slow cleanup', async () => { const s = await setup(); const { cleanupControllerLibraries } = await import('../../src/lib/server/game-files/index.js');