Delete idle session files after the retention window (OPT-1317)

service/.env.example

@@ -9,6 +9,19 @@ REDIS_HOST=localhost
 REDIS_PORT=6379
 REDIS_PASSWORD=mysecretpassword
 
+# Session-file retention. Count-only by default; true enables deletion.
+# 48h retention + at least 24h session TTL = 72h age.
+# /metrics exposes codeapi_file_retention_{prefixes,bytes,complete}.
+# Large candidate lists spill to private temporary files under TMPDIR.
+# Each pass reclaims spill directories quiet for more than 24 hours.
+CODEAPI_FILE_RETENTION_HOURS=48
+CODEAPI_FILE_RETENTION_ENABLED=false
+CODEAPI_FILE_RETENTION_INTERVAL_MS=60000
+CODEAPI_FILE_RETENTION_WORK_PER_TICK=1000
+# Additional cross-session prefixes if build artifacts share MINIO_BUCKET.
+# runner/ and CODEAPI_CHECKPOINT_PREFIX are always excluded.
+CODEAPI_FILE_RETENTION_EXCLUDED_PREFIXES=
+
 # -----------------------------------------------------------------------------
 # Stripe: https://shipfa.st/docs/features/payments
 # -----------------------------------------------------------------------------

service/src/config.ts

@@ -311,6 +311,12 @@ export const env = {
   FETCH_MAX_REQUESTS: Number(process.env.FETCH_MAX_REQUESTS) || 120, // 120 requests per minute
   // Redis Key Cache Config
   SESSION_CACHE_TTL: Number(process.env.SESSION_CACHE_TTL) || 86400,
+  FILE_RETENTION_HOURS: configuredNumber(process.env.CODEAPI_FILE_RETENTION_HOURS, 48),
+  FILE_RETENTION_ENABLED: process.env.CODEAPI_FILE_RETENTION_ENABLED === 'true',
+  FILE_RETENTION_INTERVAL_MS: configuredNumber(process.env.CODEAPI_FILE_RETENTION_INTERVAL_MS, 60_000),
+  FILE_RETENTION_WORK_PER_TICK: configuredNumber(process.env.CODEAPI_FILE_RETENTION_WORK_PER_TICK, 1000),
+  FILE_RETENTION_EXCLUDED_PREFIXES: (process.env.CODEAPI_FILE_RETENTION_EXCLUDED_PREFIXES ?? '')
+    .split(',').map(value => value.trim()).filter(Boolean),
   /** Strict tenant isolation. When true, sessionKey resolution fails closed
    *  (500) on requests whose auth context lacks `tenantId`, instead of
    *  silently falling back to the `'legacy'` tenant prefix. Default OFF in

service/src/file-retention.test.ts

@@ -0,0 +1,280 @@
+import { afterEach, describe, expect, test } from 'bun:test';
+import { Readable } from 'node:stream';
+import { mkdtempSync, readdirSync } from 'node:fs';
+import { mkdir, rm, utimes, writeFile } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import type { Client } from 'minio';
+import { FileRetentionSweep, type RetentionSummary } from './file-retention';
+
+const now = Date.parse('2026-10-06T12:00:00Z');
+const hour = 3600_000;
+type ObjectEntry = { name: string; size: number; lastModified: Date };
+
+const fixtures: { sweep: FileRetentionSweep; tempDirectory: string }[] = [];
+afterEach(async () => {
+  for (const fixture of fixtures.splice(0)) {
+    await fixture.sweep.close();
+    await rm(fixture.tempDirectory, { recursive: true, force: true });
+  }
+});
+
+function fixture(enabled = true, workPerTick = 1000) {
+  const tempDirectory = mkdtempSync(join(tmpdir(), 'retention-test-'));
+  let onRemove: (() => void) | undefined;
+  const objects = new Map<string, ObjectEntry>();
+  const live = new Set<string>();
+  const deleted: string[][] = [];
+  let visits = 0;
+  let listing = 0;
+  let onList: ((prefix: string, count: number) => void) | undefined;
+  const put = (name: string, age = 73, size = 10): void => {
+    objects.set(name, { name, size, lastModified: new Date(now - age * hour) });
+  };
+  const store = {
+    listObjectsV2(_bucket: string, prefix: string, recursive: boolean) {
+      onList?.(prefix, ++listing);
+      const snapshot = [...objects.values()].filter(object => object.name.startsWith(prefix));
+      const entries = recursive ? snapshot : [...new Set(snapshot.map(object => object.name.split('/')[0] + '/'))]
+        .sort().map(prefix => ({ prefix }));
+      return Readable.from((async function* () {
+        for (const entry of entries) { visits++; yield entry; }
+      })());
+    },
+    async removeObjects(_bucket: string, names: string[]) {
+      onRemove?.();
+      deleted.push([...names]);
+      for (const name of names) objects.delete(name);
+      return [];
+    },
+  } as unknown as Pick<Client, 'listObjectsV2' | 'removeObjects'>;
+  const sweep = new FileRetentionSweep(store, {
+    exists: async key => Number(live.has(key.slice('session:'.length))),
+  }, 'files', {
+    enabled, retentionHours: 48, sessionTtlSeconds: 86400, workPerTick,
+    tempDirectory,
+    excludedPrefixes: ['rtsx-checkpoints/', 'runner/', 'custom-build/'],
+  }, () => now);
+  fixtures.push({ sweep, tempDirectory });
+  return { put, objects, live, deleted, sweep, tempDirectory, visits: () => visits,
+    onList: (callback: typeof onList) => { onList = callback; },
+    onRemove: (callback: typeof onRemove) => { onRemove = callback; } };
+}
+
+async function finish(sweep: FileRetentionSweep): Promise<RetentionSummary> {
+  for (let tick = 0; tick < 10000; tick++) {
+    const summary = await sweep.tick();
+    if (summary.complete) return summary;
+  }
+  throw new Error('Sweep did not finish');
+}
+
+describe('session file retention', () => {
+  test('deletes old idle files in bounded pages and is idempotent', async () => {
+    const f = fixture();
+    for (let i = 0; i < 205; i++) f.put(`old/${i}.txt`);
+    const summary = await finish(f.sweep);
+    expect(f.objects.size).toBe(0);
+    expect(f.deleted.map(page => page.length)).toEqual([100, 100, 5]);
+    expect(summary.deleted).toEqual({ prefixes: 1, bytes: 2050 });
+    expect((await finish(f.sweep)).eligible).toEqual({ prefixes: 0, bytes: 0 });
+  });
+
+  test('keeps the whole prefix for recent objects, live sessions and cross-session storage', async () => {
+    const f = fixture();
+    f.put('recent/old.txt');
+    f.put('recent/new.txt', 1);
+    f.put('boundary/exact.txt', 72);
+    f.put('live/a.txt');
+    f.live.add('live');
+    f.put('rtsx-checkpoints/rt_id/1.tar.gz', 100);
+    f.put('runner/runner-build.zip', 100);
+    f.put('custom-build/artifact.zip', 100);
+    const summary = await finish(f.sweep);
+    expect(f.deleted).toEqual([]);
+    expect(summary.keptAge).toEqual({ prefixes: 2, bytes: 30 });
+    expect(summary.keptLive).toEqual({ prefixes: 1, bytes: 10 });
+    expect(summary.keptExclusion).toEqual({ prefixes: 3, bytes: 30 });
+  });
+
+  test('dry-run reports candidate prefixes and bytes without deleting', async () => {
+    const f = fixture(false);
+    f.put('old/a.txt', 73, 20);
+    f.put('old/b.txt', 100, 30);
+    f.put('recent/a.txt', 5, 7);
+    f.put('live/a.txt', 90, 8);
+    f.live.add('live');
+    f.put('rtsx-checkpoints/rt/1.tar.gz', 100, 9);
+    const summary = await finish(f.sweep);
+    expect(summary.dryRun).toBe(true);
+    expect(summary.eligible).toEqual({ prefixes: 1, bytes: 50 });
+    expect(summary.deleted).toEqual({ prefixes: 0, bytes: 0 });
+    expect(summary.keptAge).toEqual({ prefixes: 1, bytes: 7 });
+    expect(summary.keptLive).toEqual({ prefixes: 1, bytes: 8 });
+    expect(summary.keptExclusion).toEqual({ prefixes: 1, bytes: 9 });
+    expect(f.deleted).toEqual([]);
+    expect(f.objects.size).toBe(5);
+  });
+
+  test('rechecks the whole prefix and aborts when a new write arrived', async () => {
+    const f = fixture();
+    f.put('old/a.txt');
+    f.onList((prefix, count) => {
+      if (prefix === 'old/' && count === 3) f.put('old/b.txt', 0);
+    });
+    expect((await finish(f.sweep)).keptAge).toEqual({ prefixes: 1, bytes: 20 });
+    expect(f.deleted).toEqual([]);
+  });
+
+  test('detects a same-key overwrite during recheck', async () => {
+    const f = fixture();
+    f.put('old/a.txt');
+    f.onList((prefix, count) => {
+      if (prefix === 'old/' && count === 3) f.put('old/a.txt', 0);
+    });
+    await finish(f.sweep);
+    expect(f.deleted).toEqual([]);
+  });
+
+  test('resumes a large prefix across bounded ticks without starving later prefixes', async () => {
+    const f = fixture(true, 2);
+    for (let i = 0; i < 20; i++) f.put(`a/${i}.txt`);
+    f.put('z/last.txt');
+    expect((await f.sweep.tick()).complete).toBe(false);
+    expect(f.deleted).toEqual([]);
+    await finish(f.sweep);
+    expect(f.objects.size).toBe(0);
+  });
+
+  test('a newly live mapping during recheck preserves the prefix', async () => {
+    const f = fixture();
+    f.put('old/a.txt');
+    f.onList((_prefix, count) => { if (count === 3) f.live.add('old'); });
+    expect((await finish(f.sweep)).keptLive).toEqual({ prefixes: 1, bytes: 10 });
+    expect(f.deleted).toEqual([]);
+  });
+
+  test('a fresh key after more than one delete page keeps every old key', async () => {
+    const f = fixture();
+    for (let i = 0; i < 205; i++) f.put(`old/a-${String(i).padStart(3, '0')}.txt`);
+    f.onList((prefix, count) => {
+      if (prefix === 'old/' && count === 4) f.put('old/z-new.txt', 0);
+    });
+    const summary = await finish(f.sweep);
+    expect(f.deleted).toEqual([]);
+    expect(f.objects.size).toBe(206);
+    expect(summary.keptAge).toEqual({ prefixes: 1, bytes: 2060 });
+    expect(summary.eligible).toEqual({ prefixes: 0, bytes: 0 });
+  });
+
+  test('tolerates another replica removing a prefix after it was listed', async () => {
+    const f = fixture(true, 1);
+    f.put('old/a.txt');
+    // Start one pass, then remove its object before its eventual delete call.
+    await f.sweep.tick();
+    f.objects.delete('old/a.txt');
+    await finish(f.sweep);
+    expect(f.objects.size).toBe(0);
+  });
+
+  test('dry-run counts large prefixes and cleans up spilled candidates', async () => {
+    const f = fixture(false);
+    for (let i = 0; i < 10001; i++) f.put(`large/${i}.txt`);
+    const summary = await finish(f.sweep);
+    expect(summary.eligible).toEqual({ prefixes: 1, bytes: 100010 });
+    expect(f.deleted).toEqual([]);
+    expect(f.objects.size).toBe(10001);
+    expect(readdirSync(f.tempDirectory)).toEqual([]);
+  });
+
+  test('fully deletes a prefix above the memory threshold from spill pages', async () => {
+    const f = fixture();
+    for (let i = 0; i < 10205; i++) f.put(`large/${i}.txt`);
+    let usedSpill = false;
+    f.onRemove(() => { usedSpill ||= readdirSync(f.tempDirectory).length === 1; });
+    const summary = await finish(f.sweep);
+    expect(summary.deleted).toEqual({ prefixes: 1, bytes: 102050 });
+    expect(f.objects.size).toBe(0);
+    expect(usedSpill).toBe(true);
+    expect(f.deleted.every(page => page.length <= 100)).toBe(true);
+    expect(readdirSync(f.tempDirectory)).toEqual([]);
+  });
+
+  test('a fresh trailing key in a spilled prefix preserves everything and removes the spill', async () => {
+    const f = fixture();
+    for (let i = 0; i < 10205; i++) f.put(`large/${i}.txt`);
+    f.onList((prefix, count) => {
+      if (prefix === 'large/' && count === 4) f.put('large/z-last.txt', 0);
+    });
+    const summary = await finish(f.sweep);
+    expect(summary.keptAge).toEqual({ prefixes: 1, bytes: 102060 });
+    expect(summary.eligible).toEqual({ prefixes: 0, bytes: 0 });
+    expect(f.deleted).toEqual([]);
+    expect(f.objects.size).toBe(10206);
+    expect(readdirSync(f.tempDirectory)).toEqual([]);
+  });
+
+  test('a mapping appearing during the final listing prevents every delete page', async () => {
+    const f = fixture();
+    for (let i = 0; i < 205; i++) f.put(`old/${i}.txt`);
+    f.onList((_prefix, count) => { if (count === 4) f.live.add('old'); });
+    const summary = await finish(f.sweep);
+    expect(summary.keptLive).toEqual({ prefixes: 1, bytes: 2050 });
+    expect(f.deleted).toEqual([]);
+    expect(f.objects.size).toBe(205);
+  });
+
+  test('deletion errors remove the candidate spill', async () => {
+    const f = fixture();
+    for (let i = 0; i < 10001; i++) f.put(`large/${i}.txt`);
+    let usedSpill = false;
+    f.onRemove(() => {
+      usedSpill = readdirSync(f.tempDirectory).length === 1;
+      throw new Error('delete failed');
+    });
+    await expect(finish(f.sweep)).rejects.toThrow('delete failed');
+    expect(usedSpill).toBe(true);
+    expect(readdirSync(f.tempDirectory)).toEqual([]);
+    expect(f.objects.size).toBe(10001);
+  });
+
+  test('shutdown during a resumed final listing removes its spill without deleting', async () => {
+    const f = fixture(true, 100);
+    for (let i = 0; i < 10205; i++) f.put(`large/${i}.txt`);
+    for (let tick = 0; tick < 400; tick++) {
+      await f.sweep.tick();
+      if (readdirSync(f.tempDirectory).length) break;
+    }
+    expect(readdirSync(f.tempDirectory)).toHaveLength(1);
+    expect(f.deleted).toEqual([]);
+    await f.sweep.close();
+    expect(readdirSync(f.tempDirectory)).toEqual([]);
+    expect(f.objects.size).toBe(10205);
+  });
+
+  test('startup removes only abandoned spills older than 24 hours', async () => {
+    const f = fixture();
+    const makeSpill = async (name: string, directoryAge: number, fileAge: number): Promise<void> => {
+      const directory = join(f.tempDirectory, name);
+      await mkdir(directory);
+      const file = join(directory, 'candidates.jsonl');
+      await writeFile(file, '{"name":"old/a.txt","size":10}\n');
+      const fileTime = new Date(now - fileAge * hour);
+      const directoryTime = new Date(now - directoryAge * hour);
+      await utimes(file, fileTime, fileTime);
+      await utimes(directory, directoryTime, directoryTime);
+    };
+    await makeSpill('codeapi-file-retention-abandoned', 25, 25);
+    await makeSpill('codeapi-file-retention-recent', 1, 1);
+    await makeSpill('codeapi-file-retention-active-file', 25, 1);
+    await makeSpill('codeapi-file-retention-boundary', 25, 24);
+    await makeSpill('unrelated-directory', 25, 25);
+    await finish(f.sweep);
+    expect(readdirSync(f.tempDirectory).sort()).toEqual([
+      'codeapi-file-retention-active-file',
+      'codeapi-file-retention-boundary',
+      'codeapi-file-retention-recent',
+      'unrelated-directory',
+    ]);
+  });
+});