diff --git a/service/.env.example b/service/.env.example index e06ec4c0..6de33405 100644 --- a/service/.env.example +++ b/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 # ----------------------------------------------------------------------------- diff --git a/service/src/config.ts b/service/src/config.ts index 94be4e6a..7aa2e844 100644 --- a/service/src/config.ts +++ b/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 diff --git a/service/src/file-retention.test.ts b/service/src/file-retention.test.ts new file mode 100644 index 00000000..da1fc36a --- /dev/null +++ b/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(); + const live = new Set(); + 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; + 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 { + 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 => { + 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', + ]); + }); +}); diff --git a/service/src/file-retention.ts b/service/src/file-retention.ts new file mode 100644 index 00000000..5f8e378e --- /dev/null +++ b/service/src/file-retention.ts @@ -0,0 +1,272 @@ +import type { Client } from 'minio'; +import { createReadStream } from 'node:fs'; +import { lstat, mkdtemp, open, opendir, rm, type FileHandle } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +export interface RetentionOptions { + retentionHours: number; + sessionTtlSeconds: number; + enabled: boolean; + workPerTick: number; + excludedPrefixes: string[]; + tempDirectory?: string; +} + +type Count = { prefixes: number; bytes: number }; +export type RetentionSummary = { + dryRun: boolean; + complete: boolean; + eligible: Count; + deleted: Count; + keptAge: Count; + keptLive: Count; + keptExclusion: Count; +}; + +type Store = Pick; +type Sessions = { exists(key: string): Promise }; +const DELETE_PAGE_SIZE = 100; +const IN_MEMORY_OBJECTS = 10_000; +const SPILL_PREFIX = 'codeapi-file-retention-'; +const ABANDONED_SPILL_AGE_MS = 24 * 3600_000; + +/** Startup housekeeping shares the tick budget with object-store listings. + * Include directory mtime so a new, still-empty spill cannot look abandoned. + * A concurrently removed spill is already clean. Never follow symlinks. */ +async function* removeAbandonedSpills(root: string, now: number): AsyncGenerator { + const cutoff = now - ABANDONED_SPILL_AGE_MS; + for await (const entry of await opendir(root)) { + yield; + if (!entry.isDirectory() || !entry.name.startsWith(SPILL_PREFIX)) continue; + const directory = join(root, entry.name); + try { + let newest = (await lstat(directory)).mtimeMs; + for await (const file of await opendir(directory)) { + yield; + const stat = await lstat(join(directory, file.name)); + newest = Math.max(newest, stat.isFile() ? stat.mtimeMs : Infinity); + } + newest = Math.max(newest, (await lstat(directory)).mtimeMs); + if (newest < cutoff) await rm(directory, { recursive: true, force: true }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + } + } +} + +type Candidate = { name: string; size: number }; + +/** Only exact keys and sizes are spooled, never object bytes. Each sweep owns + * a private temporary directory; generator finalization removes it. */ +class CandidateSpool { + private buffered: Candidate[] = []; + private directory?: string; + private file?: FileHandle; + + constructor(private root: string) {} + + async append(candidate: Candidate): Promise { + if (!this.directory && this.buffered.length === IN_MEMORY_OBJECTS) { + this.directory = await mkdtemp(join(this.root, SPILL_PREFIX)); + this.file = await open(join(this.directory, 'candidates.jsonl'), 'wx', 0o600); + await this.flush(); + } + this.buffered.push(candidate); + if (this.file && this.buffered.length === DELETE_PAGE_SIZE) await this.flush(); + } + + private async flush(): Promise { + if (!this.buffered.length) return; + await this.file!.appendFile(this.buffered.map(item => JSON.stringify(item) + '\n').join('')); + this.buffered = []; + } + + async *pages(): AsyncGenerator { + if (!this.directory) { + for (let offset = 0; offset < this.buffered.length; offset += DELETE_PAGE_SIZE) { + yield this.buffered.slice(offset, offset + DELETE_PAGE_SIZE); + } + return; + } + await this.flush(); + await this.file!.close(); + this.file = undefined; + let page: Candidate[] = []; + let remainder = ''; + for await (const chunk of createReadStream(join(this.directory, 'candidates.jsonl'), { encoding: 'utf8' })) { + const lines = (remainder + chunk).split('\n'); + remainder = lines.pop()!; + for (const line of lines) { + page.push(JSON.parse(line) as Candidate); + if (page.length === DELETE_PAGE_SIZE) { + yield page; + page = []; + } + } + } + if (remainder) throw new Error('Incomplete file retention candidate spool'); + if (page.length) yield page; + } + + async close(): Promise { + try { + await this.file?.close(); + } finally { + if (this.directory) await rm(this.directory, { recursive: true, force: true }); + } + } +} + +/** One bounded pass resumes across ticks. Listings and deletes use the existing + * MinIO client, whose listing stream fetches pages lazily. No bucket-wide key + * array is retained. A process restart simply starts a new, idempotent pass. */ +export class FileRetentionSweep { + private pass?: AsyncGenerator; + private summary!: RetentionSummary; + + constructor( + private store: Store, + private sessions: Sessions, + private bucket: string, + private options: RetentionOptions, + private now: () => number = Date.now, + ) { + if (!Number.isFinite(options.retentionHours) || options.retentionHours < 0 || + !Number.isFinite(options.sessionTtlSeconds) || options.sessionTtlSeconds <= 0 || + !Number.isInteger(options.workPerTick) || options.workPerTick < 1) { + throw new Error('Invalid file retention configuration'); + } + } + + async tick(): Promise { + if (!this.pass) { + const count = (): Count => ({ prefixes: 0, bytes: 0 }); + this.summary = { + dryRun: !this.options.enabled, complete: false, + eligible: count(), deleted: count(), keptAge: count(), + keptLive: count(), keptExclusion: count(), + }; + this.pass = this.scan(); + } + try { + for (let work = 0; work < this.options.workPerTick; work++) { + if ((await this.pass.next()).done) { + this.summary.complete = true; + this.pass = undefined; + break; + } + } + return structuredClone(this.summary); + } catch (error) { + await this.pass?.return(); + this.pass = undefined; + throw error; + } + } + + async close(): Promise { + await this.pass?.return(); + this.pass = undefined; + } + + private async *scan(): AsyncGenerator { + yield* removeAbandonedSpills(this.options.tempDirectory ?? tmpdir(), this.now()); + // The fixed session authorization window is not an idle timer. Add at + // least 24h, or the configured longer TTL, to the requested idle retention. + const cutoff = this.now() - (this.options.retentionHours * 3600 + + Math.max(86400, this.options.sessionTtlSeconds)) * 1000; + const add = (count: Count, bytes: number): void => { + count.prefixes++; + count.bytes += bytes; + }; + for await (const entry of this.store.listObjectsV2(this.bucket, '', false)) { + yield; + const prefix = entry.prefix as string | undefined; + if (!prefix) { + add(this.summary.keptExclusion, entry.size ?? 0); + continue; + } + let bytes = 0; + let newest = 0; + let objects = 0; + let excluded = this.options.excludedPrefixes.some(value => + prefix.startsWith(value) || value.startsWith(prefix)); + for await (const object of this.store.listObjectsV2(this.bucket, prefix, true)) { + bytes += object.size; + objects++; + newest = Math.max(newest, object.lastModified?.getTime() ?? Infinity); + // Nested objects are not session files (/). + excluded ||= object.name.slice(prefix.length).includes('/'); + yield; + } + if (excluded) { add(this.summary.keptExclusion, bytes); continue; } + const sessionId = prefix.slice(0, -1); + if (await this.sessions.exists(`session:${sessionId}`)) { + add(this.summary.keptLive, bytes); continue; + } + if (newest >= cutoff) { add(this.summary.keptAge, bytes); continue; } + + // Re-read the entire prefix before deleting any key. A newly written + // object keeps the whole prefix, not just that object. + let checkedBytes = 0; + let checkedNewest = 0; + let checkedObjects = 0; + for await (const object of this.store.listObjectsV2(this.bucket, prefix, true)) { + checkedBytes += object.size; + checkedObjects++; + checkedNewest = Math.max(checkedNewest, object.lastModified?.getTime() ?? Infinity); + yield; + } + if (await this.sessions.exists(`session:${sessionId}`)) { + add(this.summary.keptLive, checkedBytes); continue; + } + if (checkedNewest >= cutoff || checkedNewest !== newest || + checkedBytes !== bytes || checkedObjects !== objects) { + add(this.summary.keptAge, checkedBytes); continue; + } + // Complete the final listing before deleting any page. Spill large + // prefixes to this process's temp directory instead of exempting them. + const candidates = new CandidateSpool(this.options.tempDirectory ?? tmpdir()); + try { + let finalBytes = 0; + let finalObjects = 0; + let finalNewest = 0; + for await (const object of this.store.listObjectsV2(this.bucket, prefix, true)) { + finalBytes += object.size; + finalObjects++; + finalNewest = Math.max(finalNewest, object.lastModified?.getTime() ?? Infinity); + await candidates.append({ name: object.name, size: object.size }); + yield; + } + if (await this.sessions.exists(`session:${sessionId}`)) { + add(this.summary.keptLive, finalBytes); continue; + } + if (finalNewest >= cutoff || finalNewest !== checkedNewest || + finalBytes !== checkedBytes || finalObjects !== checkedObjects) { + add(this.summary.keptAge, finalBytes); continue; + } + if (!finalObjects) continue; + add(this.summary.eligible, finalBytes); + if (!this.options.enabled) continue; + + let deletedBytes = 0; + let deletedObjects = 0; + for await (const page of candidates.pages()) { + if (await this.sessions.exists(`session:${sessionId}`)) break; + // This client has no conditional delete. An overwrite after listing + // can still race deletion; closing that race requires versioned + // deletes or coordination with every writer. + const errors = await this.store.removeObjects(this.bucket, page.map(object => object.name)); + if (errors.length) throw new Error(`File retention delete failed: ${JSON.stringify(errors)}`); + deletedBytes += page.reduce((sum, object) => sum + object.size, 0); + deletedObjects += page.length; + yield; + } + if (deletedObjects) add(this.summary.deleted, deletedBytes); + } finally { + await candidates.close(); + } + } + } +} diff --git a/service/src/file-server.ts b/service/src/file-server.ts index 46f404be..b12f4d8c 100644 --- a/service/src/file-server.ts +++ b/service/src/file-server.ts @@ -10,7 +10,7 @@ import type { BucketItem, BucketItemStat, ClientOptions } from 'minio'; import type { Readable } from 'stream'; import type * as tls from 'tls'; import type * as t from './types'; -import { metricsHandler, fileUploads, fileDownloads } from './metrics'; +import { metricsHandler, fileUploads, fileDownloads, fileRetentionPrefixes, fileRetentionBytes, fileRetentionComplete } from './metrics'; import { httpMetricsMiddleware } from './middleware/httpMetrics'; import { internalServiceAuthEnabled, requireInternalServiceAuth } from './internal-service-auth'; import { shutdownTelemetry, traceHttpRequest } from './telemetry'; @@ -18,6 +18,7 @@ import logger from './fileServerLogger'; import { env } from './config'; import { streamObjectToResponse } from './file-server-download'; import { redisKeepAliveOptions } from './redis-options'; +import { FileRetentionSweep } from './file-retention'; import { OWNER_BINDING_PATTERN, OWNER_DELETE_OPERATION, @@ -801,10 +802,48 @@ const port = Number(process.env.FILE_SERVER_PORT ?? 3000); const host = process.env.FILE_SERVER_HOST; let server: ReturnType | undefined; let shuttingDown = false; +let retentionSweep: FileRetentionSweep | undefined; +let retentionTimer: NodeJS.Timeout | undefined; +let retentionTick: Promise | undefined; + +function startRetention(): void { + if (!Number.isFinite(env.FILE_RETENTION_INTERVAL_MS) || env.FILE_RETENTION_INTERVAL_MS < 1) { + throw new Error('Invalid CODEAPI_FILE_RETENTION_INTERVAL_MS'); + } + retentionSweep = new FileRetentionSweep(minioClient, redisClient, bucketName, { + retentionHours: env.FILE_RETENTION_HOURS, + sessionTtlSeconds: env.SESSION_CACHE_TTL, + enabled: env.FILE_RETENTION_ENABLED, + workPerTick: env.FILE_RETENTION_WORK_PER_TICK, + excludedPrefixes: [env.CHECKPOINT_PREFIX, 'runner/', ...env.FILE_RETENTION_EXCLUDED_PREFIXES], + }); + const tick = async (): Promise => { + try { + const summary = await retentionSweep!.tick(); + logger.info('File retention sweep', summary); + for (const outcome of ['eligible', 'deleted', 'keptAge', 'keptLive', 'keptExclusion'] as const) { + const labels = { outcome, dry_run: String(summary.dryRun) }; + fileRetentionPrefixes.set(labels, summary[outcome].prefixes); + fileRetentionBytes.set(labels, summary[outcome].bytes); + } + fileRetentionComplete.set(Number(summary.complete)); + } catch (error) { + fileRetentionComplete.set(0); + logger.error('File retention sweep failed', { error }); + } finally { + if (!shuttingDown) { + retentionTimer = setTimeout(() => { retentionTick = tick(); }, env.FILE_RETENTION_INTERVAL_MS); + retentionTimer.unref(); + } + } + }; + retentionTick = tick(); +} async function startServer(): Promise { try { await initializeStorage(); + startRetention(); const onListen = () => { logger.info(`[${INSTANCE_ID}] Server running on ${host ?? '*'}:${port}`); }; @@ -829,8 +868,11 @@ async function shutdown(): Promise { if (shuttingDown) return; shuttingDown = true; logger.info(`[${INSTANCE_ID}] Shutting down file server...`); + clearTimeout(retentionTimer); try { await closeHttpServer(); + await retentionTick; + await retentionSweep?.close(); await redisClient.quit(); try { await shutdownTelemetry(); diff --git a/service/src/metrics.ts b/service/src/metrics.ts index adfd9872..1aed9725 100644 --- a/service/src/metrics.ts +++ b/service/src/metrics.ts @@ -155,6 +155,21 @@ export const fileDownloads = new Counter({ help: 'Total number of files downloaded', }); +export const fileRetentionPrefixes = new Gauge({ + name: 'codeapi_file_retention_prefixes', + help: 'Prefix counts in the current or last retention pass; do not sum replicas', + labelNames: ['outcome', 'dry_run'], +}); +export const fileRetentionBytes = new Gauge({ + name: 'codeapi_file_retention_bytes', + help: 'Byte counts in the current or last retention pass; do not sum replicas', + labelNames: ['outcome', 'dry_run'], +}); +export const fileRetentionComplete = new Gauge({ + name: 'codeapi_file_retention_complete', + help: 'Whether the most recent retention tick completed a whole bucket pass', +}); + // -- Tool call server metrics -- export const toolCalls = new Counter({ name: 'codeapi_tool_calls_total',