Fix retention query performance and cleanup scheduling

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 '[]'" }

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 (

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<SweepResult> {
     return sweepInFlight;
 }
 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')
         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<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());
-    pruneHistory('campaign');
     emit({ type: 'sweep-tick', frozen: false });
     return result;
 }

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<void> | null = null;
 let tickRequested = false;
 const requesterFlights = new Map<string, Promise<void>>();
+let nextRetentionAt = 0;
 
 function tick(requestAnother = false): Promise<void> {
     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); })

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})`;
+    // Keep age outside the correlated failure checks so SQLite can seek the
-    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)))`;
+    // clock index before examining dependencies. Resolved failures start from
+    // their controls, not another scan of the entire history table.
+    const candidates: string[] = [];
+    if (enabled) candidates.push(`SELECT o.rowid FROM ${policy.table} o
+        WHERE ${policy.category === 'version' && !days ? '1' : age} AND NOT (${failed})`);
+    if (failureDays > 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(@failureCutoff)
+        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) } };

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);

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');

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');