From e7d4c4019174ccda903f69d4f772e1aee964567c Mon Sep 17 00:00:00 2001 From: Thor Haaland Date: Tue, 6 Oct 2026 22:52:06 +0000 Subject: [PATCH 1/4] feat: sweep expired session files with dry-run retention counts --- service/.env.example | 10 ++ service/src/config.ts | 6 ++ service/src/file-retention.test.ts | 160 ++++++++++++++++++++++++++++ service/src/file-retention.ts | 165 +++++++++++++++++++++++++++++ service/src/file-server.ts | 44 +++++++- service/src/metrics.ts | 15 +++ 6 files changed, 399 insertions(+), 1 deletion(-) create mode 100644 service/src/file-retention.test.ts create mode 100644 service/src/file-retention.ts diff --git a/service/.env.example b/service/.env.example index e06ec4c0..768b2d61 100644 --- a/service/.env.example +++ b/service/.env.example @@ -9,6 +9,16 @@ REDIS_HOST=localhost REDIS_PORT=6379 REDIS_PASSWORD=mysecretpassword +# Session-file retention. Default is count-only; enable only after Thor approves +# a complete dry-run pass. 48h retention + at least 24h session TTL = 72h age. +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..2064fd6c --- /dev/null +++ b/service/src/file-retention.test.ts @@ -0,0 +1,160 @@ +import { describe, expect, test } from 'bun:test'; +import { Readable } from 'node:stream'; +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 }; + +function fixture(enabled = true, workPerTick = 1000) { + 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[]) { + 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, + excludedPrefixes: ['rtsx-checkpoints/', 'runner/', 'custom-build/'], + }, () => now); + return { put, objects, live, deleted, sweep, visits: () => visits, + onList: (callback: typeof onList) => { onList = 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 object in the deletion listing is never sent for deletion', async () => { + const f = fixture(); + f.put('old/a.txt'); + f.onList((prefix, count) => { + if (prefix === 'old/' && count === 4) f.put('old/new.txt', 0); + }); + await finish(f.sweep); + expect(f.deleted).toEqual([]); + expect(f.objects.has('old/new.txt')).toBe(true); + expect(f.objects.has('old/a.txt')).toBe(true); + }); + + 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); + }); +}); diff --git a/service/src/file-retention.ts b/service/src/file-retention.ts new file mode 100644 index 00000000..1f192e26 --- /dev/null +++ b/service/src/file-retention.ts @@ -0,0 +1,165 @@ +import type { BucketItem, Client } from 'minio'; + +export interface RetentionOptions { + retentionHours: number; + sessionTtlSeconds: number; + enabled: boolean; + workPerTick: number; + excludedPrefixes: 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; + +/** 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 { + // 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; + } + add(this.summary.eligible, checkedBytes); + if (!this.options.enabled) continue; + + let page: BucketItem[] = []; + let deletedBytes = 0; + let deletedObjects = 0; + const remove = async (): Promise => { + // Never issue a prefix deletion. Only exact, old keys from this page. + // This client has no conditional delete. An overwrite after this + // listing can still race the delete; fixing that requires versioned + // deletes or coordination with every writer, not another stat call. + 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; + page = []; + }; + for await (const object of this.store.listObjectsV2(this.bucket, prefix, true)) { + if ((object.lastModified?.getTime() ?? Infinity) >= cutoff || + await this.sessions.exists(`session:${sessionId}`)) { + page = []; + break; + } + page.push(object); + if (page.length === DELETE_PAGE_SIZE) { + await remove(); + } + yield; + } + if (page.length && !await this.sessions.exists(`session:${sessionId}`)) await remove(); + if (deletedObjects) add(this.summary.deleted, deletedBytes); + } + } +} 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', From 27af96c9eed53946646729f7c01cae5c84578cb0 Mon Sep 17 00:00:00 2001 From: Thor Haaland Date: Tue, 6 Oct 2026 23:01:08 +0000 Subject: [PATCH 2/4] fix: finish retention listing before deleting bounded candidates --- service/.env.example | 6 ++-- service/src/file-retention.test.ts | 54 ++++++++++++++++++++++++++---- service/src/file-retention.ts | 50 +++++++++++++++++---------- service/src/file-server.ts | 2 +- 4 files changed, 86 insertions(+), 26 deletions(-) diff --git a/service/.env.example b/service/.env.example index 768b2d61..1680ac6b 100644 --- a/service/.env.example +++ b/service/.env.example @@ -9,8 +9,10 @@ REDIS_HOST=localhost REDIS_PORT=6379 REDIS_PASSWORD=mysecretpassword -# Session-file retention. Default is count-only; enable only after Thor approves -# a complete dry-run pass. 48h retention + at least 24h session TTL = 72h age. +# 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}. +# Prefixes over 10000 objects are kept and reported as keptTooLarge. CODEAPI_FILE_RETENTION_HOURS=48 CODEAPI_FILE_RETENTION_ENABLED=false CODEAPI_FILE_RETENTION_INTERVAL_MS=60000 diff --git a/service/src/file-retention.test.ts b/service/src/file-retention.test.ts index 2064fd6c..34bb349c 100644 --- a/service/src/file-retention.test.ts +++ b/service/src/file-retention.test.ts @@ -136,16 +136,17 @@ describe('session file retention', () => { expect(f.deleted).toEqual([]); }); - test('a fresh object in the deletion listing is never sent for deletion', async () => { + test('a fresh key after more than one delete page keeps every old key', async () => { const f = fixture(); - f.put('old/a.txt'); + 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/new.txt', 0); + if (prefix === 'old/' && count === 4) f.put('old/z-new.txt', 0); }); - await finish(f.sweep); + const summary = await finish(f.sweep); expect(f.deleted).toEqual([]); - expect(f.objects.has('old/new.txt')).toBe(true); - expect(f.objects.has('old/a.txt')).toBe(true); + 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 () => { @@ -157,4 +158,45 @@ describe('session file retention', () => { await finish(f.sweep); expect(f.objects.size).toBe(0); }); + + test('dry-run reports an oversized prefix as kept rather than eligible', 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.keptTooLarge).toEqual({ prefixes: 1, bytes: 100010 }); + expect(summary.eligible).toEqual({ prefixes: 0, bytes: 0 }); + expect(f.deleted).toEqual([]); + }); + + test('a prefix exactly at the collection cap can be deleted', async () => { + const f = fixture(); + for (let i = 0; i < 10000; i++) f.put(`large/${i}.txt`); + const summary = await finish(f.sweep); + expect(summary.deleted).toEqual({ prefixes: 1, bytes: 100000 }); + expect(summary.keptTooLarge).toEqual({ prefixes: 0, bytes: 0 }); + expect(f.objects.size).toBe(0); + }); + + test('exceeding the cap during the final listing keeps the entire prefix', async () => { + const f = fixture(); + for (let i = 0; i < 10000; i++) f.put(`large/${i}.txt`); + f.onList((prefix, count) => { + if (prefix === 'large/' && count === 4) f.put('large/z-last.txt'); + }); + const summary = await finish(f.sweep); + expect(summary.keptTooLarge).toEqual({ prefixes: 1, bytes: 100010 }); + expect(summary.eligible).toEqual({ prefixes: 0, bytes: 0 }); + expect(f.deleted).toEqual([]); + expect(f.objects.size).toBe(10001); + }); + + 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); + }); }); diff --git a/service/src/file-retention.ts b/service/src/file-retention.ts index 1f192e26..d82dd994 100644 --- a/service/src/file-retention.ts +++ b/service/src/file-retention.ts @@ -17,11 +17,13 @@ export type RetentionSummary = { keptAge: Count; keptLive: Count; keptExclusion: Count; + keptTooLarge: Count; }; type Store = Pick; type Sessions = { exists(key: string): Promise }; const DELETE_PAGE_SIZE = 100; +const MAX_PREFIX_OBJECTS = 10_000; /** One bounded pass resumes across ticks. Listings and deletes use the existing * MinIO client, whose listing stream fetches pages lazily. No bucket-wide key @@ -50,7 +52,7 @@ export class FileRetentionSweep { this.summary = { dryRun: !this.options.enabled, complete: false, eligible: count(), deleted: count(), keptAge: count(), - keptLive: count(), keptExclusion: count(), + keptLive: count(), keptExclusion: count(), keptTooLarge: count(), }; this.pass = this.scan(); } @@ -110,6 +112,7 @@ export class FileRetentionSweep { add(this.summary.keptLive, bytes); continue; } if (newest >= cutoff) { add(this.summary.keptAge, bytes); continue; } + if (objects > MAX_PREFIX_OBJECTS) { add(this.summary.keptTooLarge, bytes); continue; } // Re-read the entire prefix before deleting any key. A newly written // object keeps the whole prefix, not just that object. @@ -129,13 +132,39 @@ export class FileRetentionSweep { checkedBytes !== bytes || checkedObjects !== objects) { add(this.summary.keptAge, checkedBytes); continue; } - add(this.summary.eligible, checkedBytes); + // Finish the final listing before removing even the first page. A fresh + // key may sort after hundreds of old ones. Retain at most the cap while + // still counting the full prefix for the summary. + const candidates: BucketItem[] = []; + 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); + if (finalObjects <= MAX_PREFIX_OBJECTS) candidates.push(object); + yield; + } + if (await this.sessions.exists(`session:${sessionId}`)) { + add(this.summary.keptLive, finalBytes); continue; + } + if (finalObjects > MAX_PREFIX_OBJECTS) { + add(this.summary.keptTooLarge, 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 page: BucketItem[] = []; let deletedBytes = 0; let deletedObjects = 0; - const remove = async (): Promise => { + for (let offset = 0; offset < candidates.length; offset += DELETE_PAGE_SIZE) { + if (await this.sessions.exists(`session:${sessionId}`)) break; + const page = candidates.slice(offset, offset + DELETE_PAGE_SIZE); // Never issue a prefix deletion. Only exact, old keys from this page. // This client has no conditional delete. An overwrite after this // listing can still race the delete; fixing that requires versioned @@ -144,21 +173,8 @@ export class FileRetentionSweep { 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; - page = []; - }; - for await (const object of this.store.listObjectsV2(this.bucket, prefix, true)) { - if ((object.lastModified?.getTime() ?? Infinity) >= cutoff || - await this.sessions.exists(`session:${sessionId}`)) { - page = []; - break; - } - page.push(object); - if (page.length === DELETE_PAGE_SIZE) { - await remove(); - } yield; } - if (page.length && !await this.sessions.exists(`session:${sessionId}`)) await remove(); if (deletedObjects) add(this.summary.deleted, deletedBytes); } } diff --git a/service/src/file-server.ts b/service/src/file-server.ts index b12f4d8c..af5f92d0 100644 --- a/service/src/file-server.ts +++ b/service/src/file-server.ts @@ -821,7 +821,7 @@ function startRetention(): void { try { const summary = await retentionSweep!.tick(); logger.info('File retention sweep', summary); - for (const outcome of ['eligible', 'deleted', 'keptAge', 'keptLive', 'keptExclusion'] as const) { + for (const outcome of ['eligible', 'deleted', 'keptAge', 'keptLive', 'keptExclusion', 'keptTooLarge'] as const) { const labels = { outcome, dry_run: String(summary.dryRun) }; fileRetentionPrefixes.set(labels, summary[outcome].prefixes); fileRetentionBytes.set(labels, summary[outcome].bytes); From 116f366608a20584ac69cb80365671c58c8b5068 Mon Sep 17 00:00:00 2001 From: Thor Haaland Date: Tue, 6 Oct 2026 23:04:56 +0000 Subject: [PATCH 3/4] fix: spill large retention candidates instead of exempting prefixes --- service/.env.example | 2 +- service/src/file-retention.test.ts | 82 ++++++++++++--- service/src/file-retention.ts | 160 ++++++++++++++++++++--------- service/src/file-server.ts | 2 +- 4 files changed, 181 insertions(+), 65 deletions(-) diff --git a/service/.env.example b/service/.env.example index 1680ac6b..6e4f053b 100644 --- a/service/.env.example +++ b/service/.env.example @@ -12,7 +12,7 @@ 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}. -# Prefixes over 10000 objects are kept and reported as keptTooLarge. +# Large candidate lists spill to private temporary files under TMPDIR. CODEAPI_FILE_RETENTION_HOURS=48 CODEAPI_FILE_RETENTION_ENABLED=false CODEAPI_FILE_RETENTION_INTERVAL_MS=60000 diff --git a/service/src/file-retention.test.ts b/service/src/file-retention.test.ts index 34bb349c..b0448726 100644 --- a/service/src/file-retention.test.ts +++ b/service/src/file-retention.test.ts @@ -1,5 +1,9 @@ -import { describe, expect, test } from 'bun:test'; +import { afterEach, describe, expect, test } from 'bun:test'; import { Readable } from 'node:stream'; +import { mkdtempSync, readdirSync } from 'node:fs'; +import { rm } 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'; @@ -7,7 +11,17 @@ 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[][] = []; @@ -28,6 +42,7 @@ function fixture(enabled = true, workPerTick = 1000) { })()); }, async removeObjects(_bucket: string, names: string[]) { + onRemove?.(); deleted.push([...names]); for (const name of names) objects.delete(name); return []; @@ -37,10 +52,13 @@ function fixture(enabled = true, workPerTick = 1000) { 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); - return { put, objects, live, deleted, sweep, visits: () => visits, - onList: (callback: typeof onList) => { onList = callback; } }; + 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 { @@ -159,35 +177,41 @@ describe('session file retention', () => { expect(f.objects.size).toBe(0); }); - test('dry-run reports an oversized prefix as kept rather than eligible', async () => { + 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.keptTooLarge).toEqual({ prefixes: 1, bytes: 100010 }); - expect(summary.eligible).toEqual({ prefixes: 0, bytes: 0 }); + expect(summary.eligible).toEqual({ prefixes: 1, bytes: 100010 }); expect(f.deleted).toEqual([]); + expect(f.objects.size).toBe(10001); + expect(readdirSync(f.tempDirectory)).toEqual([]); }); - test('a prefix exactly at the collection cap can be deleted', async () => { + test('fully deletes a prefix above the memory threshold from spill pages', async () => { const f = fixture(); - for (let i = 0; i < 10000; i++) f.put(`large/${i}.txt`); + 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: 100000 }); - expect(summary.keptTooLarge).toEqual({ prefixes: 0, bytes: 0 }); + 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('exceeding the cap during the final listing keeps the entire prefix', async () => { + test('a fresh trailing key in a spilled prefix preserves everything and removes the spill', async () => { const f = fixture(); - for (let i = 0; i < 10000; i++) f.put(`large/${i}.txt`); + 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'); + if (prefix === 'large/' && count === 4) f.put('large/z-last.txt', 0); }); const summary = await finish(f.sweep); - expect(summary.keptTooLarge).toEqual({ prefixes: 1, bytes: 100010 }); + 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(10001); + expect(f.objects.size).toBe(10206); + expect(readdirSync(f.tempDirectory)).toEqual([]); }); test('a mapping appearing during the final listing prevents every delete page', async () => { @@ -199,4 +223,32 @@ describe('session file retention', () => { 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); + }); }); diff --git a/service/src/file-retention.ts b/service/src/file-retention.ts index d82dd994..eda423ae 100644 --- a/service/src/file-retention.ts +++ b/service/src/file-retention.ts @@ -1,4 +1,8 @@ -import type { BucketItem, Client } from 'minio'; +import type { Client } from 'minio'; +import { createReadStream } from 'node:fs'; +import { mkdtemp, open, rm, type FileHandle } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; export interface RetentionOptions { retentionHours: number; @@ -6,6 +10,7 @@ export interface RetentionOptions { enabled: boolean; workPerTick: number; excludedPrefixes: string[]; + tempDirectory?: string; } type Count = { prefixes: number; bytes: number }; @@ -17,13 +22,75 @@ export type RetentionSummary = { keptAge: Count; keptLive: Count; keptExclusion: Count; - keptTooLarge: Count; }; type Store = Pick; type Sessions = { exists(key: string): Promise }; const DELETE_PAGE_SIZE = 100; -const MAX_PREFIX_OBJECTS = 10_000; +const IN_MEMORY_OBJECTS = 10_000; + +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, 'codeapi-file-retention-')); + 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 @@ -52,7 +119,7 @@ export class FileRetentionSweep { this.summary = { dryRun: !this.options.enabled, complete: false, eligible: count(), deleted: count(), keptAge: count(), - keptLive: count(), keptExclusion: count(), keptTooLarge: count(), + keptLive: count(), keptExclusion: count(), }; this.pass = this.scan(); } @@ -112,7 +179,6 @@ export class FileRetentionSweep { add(this.summary.keptLive, bytes); continue; } if (newest >= cutoff) { add(this.summary.keptAge, bytes); continue; } - if (objects > MAX_PREFIX_OBJECTS) { add(this.summary.keptTooLarge, bytes); continue; } // Re-read the entire prefix before deleting any key. A newly written // object keeps the whole prefix, not just that object. @@ -132,50 +198,48 @@ export class FileRetentionSweep { checkedBytes !== bytes || checkedObjects !== objects) { add(this.summary.keptAge, checkedBytes); continue; } - // Finish the final listing before removing even the first page. A fresh - // key may sort after hundreds of old ones. Retain at most the cap while - // still counting the full prefix for the summary. - const candidates: BucketItem[] = []; - 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); - if (finalObjects <= MAX_PREFIX_OBJECTS) candidates.push(object); - yield; - } - if (await this.sessions.exists(`session:${sessionId}`)) { - add(this.summary.keptLive, finalBytes); continue; - } - if (finalObjects > MAX_PREFIX_OBJECTS) { - add(this.summary.keptTooLarge, 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 (let offset = 0; offset < candidates.length; offset += DELETE_PAGE_SIZE) { - if (await this.sessions.exists(`session:${sessionId}`)) break; - const page = candidates.slice(offset, offset + DELETE_PAGE_SIZE); - // Never issue a prefix deletion. Only exact, old keys from this page. - // This client has no conditional delete. An overwrite after this - // listing can still race the delete; fixing that requires versioned - // deletes or coordination with every writer, not another stat call. - 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; + // 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(); } - if (deletedObjects) add(this.summary.deleted, deletedBytes); } } } diff --git a/service/src/file-server.ts b/service/src/file-server.ts index af5f92d0..b12f4d8c 100644 --- a/service/src/file-server.ts +++ b/service/src/file-server.ts @@ -821,7 +821,7 @@ function startRetention(): void { try { const summary = await retentionSweep!.tick(); logger.info('File retention sweep', summary); - for (const outcome of ['eligible', 'deleted', 'keptAge', 'keptLive', 'keptExclusion', 'keptTooLarge'] as const) { + 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); From 028b08d9b3497b89cac42f3a66b975aed04f7e58 Mon Sep 17 00:00:00 2001 From: Thor Haaland Date: Tue, 6 Oct 2026 23:08:50 +0000 Subject: [PATCH 4/4] fix: reclaim abandoned retention spills after 24 hours --- service/.env.example | 1 + service/src/file-retention.test.ts | 28 ++++++++++++++++++++++++++- service/src/file-retention.ts | 31 ++++++++++++++++++++++++++++-- 3 files changed, 57 insertions(+), 3 deletions(-) diff --git a/service/.env.example b/service/.env.example index 6e4f053b..6de33405 100644 --- a/service/.env.example +++ b/service/.env.example @@ -13,6 +13,7 @@ REDIS_PASSWORD=mysecretpassword # 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 diff --git a/service/src/file-retention.test.ts b/service/src/file-retention.test.ts index b0448726..da1fc36a 100644 --- a/service/src/file-retention.test.ts +++ b/service/src/file-retention.test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect, test } from 'bun:test'; import { Readable } from 'node:stream'; import { mkdtempSync, readdirSync } from 'node:fs'; -import { rm } from 'node:fs/promises'; +import { mkdir, rm, utimes, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import type { Client } from 'minio'; @@ -251,4 +251,30 @@ describe('session file retention', () => { 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 index eda423ae..5f8e378e 100644 --- a/service/src/file-retention.ts +++ b/service/src/file-retention.ts @@ -1,6 +1,6 @@ import type { Client } from 'minio'; import { createReadStream } from 'node:fs'; -import { mkdtemp, open, rm, type FileHandle } from 'node:fs/promises'; +import { lstat, mkdtemp, open, opendir, rm, type FileHandle } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -28,6 +28,32 @@ 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 }; @@ -42,7 +68,7 @@ class CandidateSpool { async append(candidate: Candidate): Promise { if (!this.directory && this.buffered.length === IN_MEMORY_OBJECTS) { - this.directory = await mkdtemp(join(this.root, 'codeapi-file-retention-')); + this.directory = await mkdtemp(join(this.root, SPILL_PREFIX)); this.file = await open(join(this.directory, 'candidates.jsonl'), 'wx', 0o600); await this.flush(); } @@ -145,6 +171,7 @@ export class FileRetentionSweep { } 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 +