diff --git a/api/server/services/Files/Code/crud.js b/api/server/services/Files/Code/crud.js index c7e63540d2..aaa53a82d8 100644 --- a/api/server/services/Files/Code/crud.js +++ b/api/server/services/Files/Code/crud.js @@ -84,7 +84,6 @@ async function deleteCodeEnvFile(req, file) { return; } - const missingOrUnsupportedStatuses = new Set([404, 405]); for (const [executionRouteKey, ref] of refs) { const executionProfile = ref.executionProfile ?? 'default'; const environments = @@ -127,48 +126,37 @@ async function deleteCodeEnvFile(req, file) { const bridgeWorkerId = configuredEnvironment?.workerId ?? configuredEnvironment?.pairing?.workerId; const authHeaders = await getCodeApiAuthHeaders(req, bridgeWorkerId); - const baseRequest = { - method: 'delete', - headers: { - 'User-Agent': 'LibreChat/1.0', - ...authHeaders, - ...codeExecutionHeaders({ - executionProfile, - bridgeWorkerId, - }), - }, - httpAgent: codeServerHttpAgent, - httpsAgent: codeServerHttpsAgent, - timeout: 15000, - }; - const urls = [ - `${baseURL}/sessions/${ref.storage_session_id}/objects/${ref.file_id}${query}`, - `${baseURL}/files/${ref.storage_session_id}/${ref.file_id}${query}`, - ]; - - let lastError; - let deleted = false; - for (const url of urls) { - try { - await axios({ ...baseRequest, url }); - deleted = true; - break; - } catch (error) { - lastError = error; - if (!missingOrUnsupportedStatuses.has(error.response?.status)) { - throw error; - } - } - } - if (!deleted && lastError) { - logAxiosError({ - error: lastError, - message: `Error deleting code environment file: ${lastError.message}`, + /* codeapi has mounted DELETE at `/files/:session_id/:fileId` since its + * first release. The file-server's own `/sessions/:id/objects/:fileId` + * only gained DELETE in LibreChat-AI/code-interpreter#85, so trying it + * first cost a guaranteed 404 against every older deployment. */ + try { + await axios({ + method: 'delete', + url: `${baseURL}/files/${ref.storage_session_id}/${ref.file_id}${query}`, + headers: { + 'User-Agent': 'LibreChat/1.0', + ...authHeaders, + ...codeExecutionHeaders({ + executionProfile, + bridgeWorkerId, + }), + }, + httpAgent: codeServerHttpAgent, + httpsAgent: codeServerHttpsAgent, + timeout: 15000, }); - if (lastError.response?.status === 404) { - continue; + } catch (error) { + if (error.response?.status !== 404) { + throw error; } - throw new Error(lastError.message || 'An error occurred during file deletion.'); + /* Already gone. Logged rather than swallowed: a 404 from a + * misconfigured base URL is indistinguishable from one for an absent + * object, and this branch drops the file's record either way. */ + logAxiosError({ + error, + message: `Code environment object already absent: ${error.message}`, + }); } } } diff --git a/api/server/services/Files/Code/crud.spec.js b/api/server/services/Files/Code/crud.spec.js index 4e5c1242db..e35758ee35 100644 --- a/api/server/services/Files/Code/crud.spec.js +++ b/api/server/services/Files/Code/crud.spec.js @@ -212,7 +212,7 @@ describe('Code CRUD', () => { expect(mockAxios).toHaveBeenCalledWith( expect.objectContaining({ method: 'delete', - url: 'https://code-api.example.com/sessions/session-1/objects/file-1?kind=agent&id=agent-abc', + url: 'https://code-api.example.com/files/session-1/file-1?kind=agent&id=agent-abc', headers: expect.objectContaining({ Authorization: 'Bearer codeapi-token', 'User-Agent': 'LibreChat/1.0', @@ -239,7 +239,7 @@ describe('Code CRUD', () => { expect(mockAxios).toHaveBeenCalledWith( expect.objectContaining({ - url: 'https://code-stateful.example.com/sessions/session-1/objects/file-1?kind=agent&id=agent-abc', + url: 'https://code-stateful.example.com/files/session-1/file-1?kind=agent&id=agent-abc', headers: expect.objectContaining({ 'X-CodeAPI-Expected-Profile': 'stateful', }), @@ -270,14 +270,14 @@ describe('Code CRUD', () => { expect(mockAxios).toHaveBeenNthCalledWith( 1, expect.objectContaining({ - url: expect.stringContaining('/sessions/session-1/objects/file-1'), + url: expect.stringContaining('/files/session-1/file-1'), headers: expect.objectContaining({ 'X-CodeAPI-Expected-Profile': 'default' }), }), ); expect(mockAxios).toHaveBeenNthCalledWith( 2, expect.objectContaining({ - url: expect.stringContaining('/sessions/stateful-session/objects/stateful-file'), + url: expect.stringContaining('/files/stateful-session/stateful-file'), headers: expect.objectContaining({ 'X-CodeAPI-Expected-Profile': 'stateful' }), }), ); @@ -320,33 +320,27 @@ describe('Code CRUD', () => { expect(mockAxios).not.toHaveBeenCalled(); }); - it.each([404, 405])( - 'falls back to the legacy code environment delete route after a %s', - async (status) => { - mockAxios - .mockRejectedValueOnce( - Object.assign(new Error('missing route'), { response: { status } }), - ) - .mockResolvedValueOnce({ status: 204 }); + it('never calls the file-server path that only newer codeapi mounts', async () => { + mockAxios.mockResolvedValue({ status: 204 }); - await deleteCodeEnvFile(req, file); + await deleteCodeEnvFile(req, file); - expect(mockAxios).toHaveBeenNthCalledWith( - 1, - expect.objectContaining({ - method: 'delete', - url: 'https://code-api.example.com/sessions/session-1/objects/file-1?kind=agent&id=agent-abc', - }), - ); - expect(mockAxios).toHaveBeenNthCalledWith( - 2, - expect.objectContaining({ - method: 'delete', - url: 'https://code-api.example.com/files/session-1/file-1?kind=agent&id=agent-abc', - }), - ); - }, - ); + expect(mockAxios).toHaveBeenCalledTimes(1); + expect(mockAxios).not.toHaveBeenCalledWith( + expect.objectContaining({ + url: expect.stringContaining('/sessions/session-1/objects/file-1'), + }), + ); + }); + + it('surfaces an unsupported delete method instead of retrying elsewhere', async () => { + mockAxios.mockRejectedValue( + Object.assign(new Error('method not allowed'), { response: { status: 405 } }), + ); + + await expect(deleteCodeEnvFile(req, file)).rejects.toThrow('method not allowed'); + expect(mockAxios).toHaveBeenCalledTimes(1); + }); it('skips files without a code environment ref', async () => { await deleteCodeEnvFile(req, {}); @@ -356,12 +350,12 @@ describe('Code CRUD', () => { }); it('treats missing code environment objects as already deleted', async () => { - mockAxios - .mockRejectedValueOnce(Object.assign(new Error('missing'), { response: { status: 404 } })) - .mockRejectedValueOnce(Object.assign(new Error('missing'), { response: { status: 404 } })); + mockAxios.mockRejectedValue( + Object.assign(new Error('missing'), { response: { status: 404 } }), + ); await expect(deleteCodeEnvFile(req, file)).resolves.toBeUndefined(); - expect(mockAxios).toHaveBeenCalledTimes(2); + expect(mockAxios).toHaveBeenCalledTimes(1); }); it('throws when code environment deletion fails', async () => { diff --git a/api/server/services/Files/process.js b/api/server/services/Files/process.js index c4e7d307b3..09c0b0dc57 100644 --- a/api/server/services/Files/process.js +++ b/api/server/services/Files/process.js @@ -349,6 +349,8 @@ async function sweepExpiredFiles(options = {}) { return sweepExpiredFilesWithDeps(options, { getExpiredFiles: db.getExpiredFiles, processDeleteRequest, + incrementFileDeletionAttempts: db.incrementFileDeletionAttempts, + deferExpiredFile: db.deferExpiredFile, logger, }); } diff --git a/api/server/services/Files/process.spec.js b/api/server/services/Files/process.spec.js index 14eaae5941..20562fa627 100644 --- a/api/server/services/Files/process.spec.js +++ b/api/server/services/Files/process.spec.js @@ -172,6 +172,8 @@ jest.mock('~/models', () => ({ findFileById: jest.fn(), getConvo: jest.fn(), getExpiredFiles: jest.fn(), + incrementFileDeletionAttempts: jest.fn(), + deferExpiredFile: jest.fn(), addAgentResourceFile: jest.fn().mockResolvedValue({}), removeAgentResourceFiles: jest.fn(), removeAgentResourceFilesFromAllAgents: jest.fn(), @@ -1840,6 +1842,8 @@ describe('sweepExpiredFiles', () => { expect.objectContaining({ getExpiredFiles: db.getExpiredFiles, processDeleteRequest: expect.any(Function), + incrementFileDeletionAttempts: db.incrementFileDeletionAttempts, + deferExpiredFile: db.deferExpiredFile, logger: expect.objectContaining({ error: expect.any(Function), info: expect.any(Function), diff --git a/packages/api/src/files/sweep.spec.ts b/packages/api/src/files/sweep.spec.ts index 4c5465f97e..49609bd2cc 100644 --- a/packages/api/src/files/sweep.spec.ts +++ b/packages/api/src/files/sweep.spec.ts @@ -1,6 +1,12 @@ import { EModelEndpoint, FileSources } from 'librechat-data-provider'; import type { AppConfig } from '@librechat/data-schemas'; -import { getFileRetentionSweepInterval, startExpiredFileSweep, sweepExpiredFiles } from './sweep'; +import { + sweepExpiredFiles, + startExpiredFileSweep, + getExpiredFileRetryDelay, + getFileRetentionMaxAttempts, + getFileRetentionSweepInterval, +} from './sweep'; describe('expired file sweep helpers', () => { const logger = { @@ -9,14 +15,23 @@ describe('expired file sweep helpers', () => { error: jest.fn(), }; + let deferExpiredFile: jest.Mock; + let incrementFileDeletionAttempts: jest.Mock; + let attemptCount: number; + beforeEach(() => { jest.clearAllMocks(); + attemptCount = 0; + incrementFileDeletionAttempts = jest.fn().mockImplementation(() => ++attemptCount); + deferExpiredFile = jest.fn().mockResolvedValue(undefined); delete process.env.FILE_RETENTION_SWEEP_INTERVAL_MS; + delete process.env.FILE_RETENTION_SWEEP_MAX_ATTEMPTS; }); afterEach(() => { jest.useRealTimers(); delete process.env.FILE_RETENTION_SWEEP_INTERVAL_MS; + delete process.env.FILE_RETENTION_SWEEP_MAX_ATTEMPTS; }); it('loads endpoint config and deletes expired OpenAI storage files', async () => { @@ -40,9 +55,17 @@ describe('expired file sweep helpers', () => { const result = await sweepExpiredFiles( { appConfig: {} as AppConfig, loadAppConfig, limit: 1 }, - { getExpiredFiles, processDeleteRequest, logger }, + { + getExpiredFiles, + processDeleteRequest, + incrementFileDeletionAttempts, + deferExpiredFile, + logger, + }, ); + expect(getExpiredFiles).toHaveBeenCalledWith(1); + expect(deferExpiredFile).not.toHaveBeenCalled(); expect(loadAppConfig).toHaveBeenCalledTimes(1); expect(processDeleteRequest).toHaveBeenCalledWith({ req: expect.objectContaining({ @@ -62,16 +85,217 @@ describe('expired file sweep helpers', () => { const result = await sweepExpiredFiles( { appConfig: {} as AppConfig, limit: 1 }, - { getExpiredFiles, processDeleteRequest, logger }, + { + getExpiredFiles, + processDeleteRequest, + incrementFileDeletionAttempts, + deferExpiredFile, + logger, + }, ); expect(processDeleteRequest).not.toHaveBeenCalled(); + expect(deferExpiredFile).toHaveBeenCalledWith('orphaned-file', expect.any(Date)); expect(logger.warn).toHaveBeenCalledWith( '[sweepExpiredFiles] Skipping expired file without user: orphaned-file', ); expect(result).toEqual({ scanned: 1, deleted: 0, failed: 1 }); }); + it.each([ + ['a rejected delete', { failedFileIds: ['stuck-file'], deletedFileIds: [] }], + ['a delete that resolves nothing', { failedFileIds: [], deletedFileIds: [] }], + ])('defers the file after %s', async (_case, outcome) => { + const getExpiredFiles = jest + .fn() + .mockResolvedValue([{ file_id: 'stuck-file', user: 'user-1' }]); + const processDeleteRequest = jest.fn().mockResolvedValue(outcome); + + const result = await sweepExpiredFiles( + { appConfig: {} as AppConfig, limit: 1 }, + { + getExpiredFiles, + processDeleteRequest, + incrementFileDeletionAttempts, + deferExpiredFile, + logger, + }, + ); + + const [fileId, retryAt] = deferExpiredFile.mock.calls[0]; + expect(fileId).toBe('stuck-file'); + /* Backoff comes off the attempt the increment returned, not the count + * the queried snapshot carried. */ + expect(incrementFileDeletionAttempts).toHaveBeenCalledWith('stuck-file'); + expect(retryAt.getTime() - Date.now()).toBeCloseTo(60 * 60 * 1000, -3); + expect(result).toEqual({ scanned: 1, deleted: 0, failed: 1 }); + }); + + it('defers the file when the delete throws', async () => { + const getExpiredFiles = jest + .fn() + .mockResolvedValue([{ file_id: 'stuck-file', user: 'user-1' }]); + const processDeleteRequest = jest.fn().mockRejectedValue(new Error('storage unreachable')); + + const result = await sweepExpiredFiles( + { appConfig: {} as AppConfig, limit: 1 }, + { + getExpiredFiles, + processDeleteRequest, + incrementFileDeletionAttempts, + deferExpiredFile, + logger, + }, + ); + + expect(deferExpiredFile).toHaveBeenCalledWith('stuck-file', expect.any(Date)); + expect(result).toEqual({ scanned: 1, deleted: 0, failed: 1 }); + }); + + it('reports parking once the file reaches the attempt cap', async () => { + process.env.FILE_RETENTION_SWEEP_MAX_ATTEMPTS = '3'; + attemptCount = 2; + deferExpiredFile.mockRejectedValue(new Error('mongo unreachable')); + const getExpiredFiles = jest + .fn() + .mockResolvedValue([{ file_id: 'stranded-file', user: 'user-1' }]); + const processDeleteRequest = jest + .fn() + .mockResolvedValue({ deletedFileIds: [], failedFileIds: ['stranded-file'] }); + + await sweepExpiredFiles( + { appConfig: {} as AppConfig, limit: 1 }, + { + getExpiredFiles, + processDeleteRequest, + incrementFileDeletionAttempts, + deferExpiredFile, + logger, + }, + ); + + /* The counter is durable once the increment lands, so a deferral that + * fails afterwards must not swallow the notice. */ + expect(logger.error).toHaveBeenCalledWith( + expect.stringContaining('Parking expired file stranded-file after 3 failed deletions'), + ); + }); + + it('reports the cap once, on the attempt that reaches it', async () => { + process.env.FILE_RETENTION_SWEEP_MAX_ATTEMPTS = '3'; + attemptCount = 2; + const getExpiredFiles = jest.fn().mockResolvedValue([ + { file_id: 'stranded-file', user: 'user-1' }, + { file_id: 'stranded-file', user: 'user-1' }, + ]); + const processDeleteRequest = jest + .fn() + .mockResolvedValue({ deletedFileIds: [], failedFileIds: ['stranded-file'] }); + + await sweepExpiredFiles( + { appConfig: {} as AppConfig, limit: 2 }, + { + getExpiredFiles, + processDeleteRequest, + incrementFileDeletionAttempts, + deferExpiredFile, + logger, + }, + ); + + /* Increments returned 3 and 4. Only the one that landed on the cap + * reports it — another node past the threshold would otherwise repeat + * the notice once per replica, per file. */ + const parked = logger.error.mock.calls.filter(([message]) => + String(message).includes('Parking expired file'), + ); + expect(parked).toHaveLength(1); + }); + + it('anchors retry deadlines to the start of the sweep', async () => { + const getExpiredFiles = jest.fn().mockImplementation(async () => { + /* Deletion I/O pushes wall-clock past the sweep's start. Measuring + * the deadline from here would land it just after the next scheduled + * pass, costing the file a whole extra interval. */ + jest.setSystemTime(Date.now() + 30_000); + return [{ file_id: 'stuck-file', user: 'user-1' }]; + }); + const processDeleteRequest = jest + .fn() + .mockResolvedValue({ deletedFileIds: [], failedFileIds: ['stuck-file'] }); + + jest.useFakeTimers(); + const sweepStartedAt = Date.now(); + await sweepExpiredFiles( + { appConfig: {} as AppConfig, limit: 1 }, + { + getExpiredFiles, + processDeleteRequest, + incrementFileDeletionAttempts, + deferExpiredFile, + logger, + }, + ); + + const [, retryAt] = deferExpiredFile.mock.calls[0]; + expect(retryAt.getTime()).toBe(sweepStartedAt + 60 * 60 * 1000); + }); + + it('keeps sweeping the batch when recording a failure fails', async () => { + const getExpiredFiles = jest.fn().mockResolvedValue([ + { file_id: 'stuck-file', user: 'user-1' }, + { file_id: 'deletable-file', user: 'user-1' }, + ]); + const processDeleteRequest = jest + .fn() + .mockResolvedValueOnce({ deletedFileIds: [], failedFileIds: ['stuck-file'] }) + .mockResolvedValueOnce({ deletedFileIds: ['deletable-file'], failedFileIds: [] }); + incrementFileDeletionAttempts.mockRejectedValue(new Error('mongo unreachable')); + + const result = await sweepExpiredFiles( + { appConfig: {} as AppConfig, limit: 2 }, + { + getExpiredFiles, + processDeleteRequest, + incrementFileDeletionAttempts, + deferExpiredFile, + logger, + }, + ); + + expect(logger.error).toHaveBeenCalledWith( + '[sweepExpiredFiles] Error recording failed deletion of expired file stuck-file:', + expect.any(Error), + ); + expect(result).toEqual({ scanned: 2, deleted: 1, failed: 1 }); + }); + + it('climbs to a daily ladder, then parks, and ignores unusable attempt limits', () => { + expect(getExpiredFileRetryDelay(1, 10)).toBe(60 * 60 * 1000); + expect(getExpiredFileRetryDelay(4, 10)).toBe(8 * 60 * 60 * 1000); + expect(getExpiredFileRetryDelay(9, 10)).toBe(24 * 60 * 60 * 1000); + + /* At the cap the file is parked rather than excluded, so a record + * reused for different content is delayed at worst, never stranded. */ + expect(getExpiredFileRetryDelay(10, 10)).toBe(30 * 24 * 60 * 60 * 1000); + expect(getExpiredFileRetryDelay(64, 10)).toBe(30 * 24 * 60 * 60 * 1000); + + /* The base tracks the configured cadence rather than a fixed hour. */ + process.env.FILE_RETENTION_SWEEP_INTERVAL_MS = String(5 * 60 * 1000); + expect(getExpiredFileRetryDelay(1, 10)).toBe(5 * 60 * 1000); + expect(getExpiredFileRetryDelay(3, 10)).toBe(20 * 60 * 1000); + + /* ...but never drops below the floor that protects the budget. */ + process.env.FILE_RETENTION_SWEEP_INTERVAL_MS = '5'; + expect(getExpiredFileRetryDelay(1, 10)).toBe(60 * 1000); + delete process.env.FILE_RETENTION_SWEEP_INTERVAL_MS; + + expect(getFileRetentionMaxAttempts('0')).toBe(10); + expect(getFileRetentionMaxAttempts('2.5')).toBe(10); + expect(getFileRetentionMaxAttempts(' ')).toBe(10); + expect(getFileRetentionMaxAttempts('25')).toBe(25); + }); + it('falls back to the default interval for sub-millisecond values', () => { expect(getFileRetentionSweepInterval('0.5')).toBe(60 * 60 * 1000); }); diff --git a/packages/api/src/files/sweep.ts b/packages/api/src/files/sweep.ts index 4128fe3aa4..e20e5afdd2 100644 --- a/packages/api/src/files/sweep.ts +++ b/packages/api/src/files/sweep.ts @@ -7,6 +7,10 @@ import { import type { AppConfig } from '@librechat/data-schemas'; const DEFAULT_FILE_RETENTION_SWEEP_INTERVAL_MS = 60 * 60 * 1000; +const DEFAULT_FILE_RETENTION_MAX_ATTEMPTS = 10; +const MIN_FILE_RETENTION_RETRY_BASE_MS = 60 * 1000; +const FILE_RETENTION_RETRY_MAX_MS = 24 * 60 * 60 * 1000; +const FILE_RETENTION_PARK_MS = 30 * 24 * 60 * 60 * 1000; type ExpiredFile = { file_id: string; @@ -51,6 +55,8 @@ type SweepDependencies = { req: SweepRequest; files: ExpiredFile[]; }) => Promise<{ deletedFileIds: string[]; failedFileIds: string[] }>; + incrementFileDeletionAttempts: (file_id: string) => Promise; + deferExpiredFile: (file_id: string, deletionRetryAt: Date) => Promise; logger: SweepLogger; }; @@ -86,6 +92,61 @@ export function getFileRetentionSweepInterval( return value; } +/** + * Consecutive failures after which a file is parked rather than retried on + * the ordinary ladder. + * + * Lower it when working through a large backlog: every attempt costs a slot + * in the bounded batch, so N stranded files take N × this many passes to + * settle before the queue frees up for files expiring behind them. + */ +export function getFileRetentionMaxAttempts( + maxAttempts: string | undefined = process.env.FILE_RETENTION_SWEEP_MAX_ATTEMPTS, +): number { + if (maxAttempts == null || maxAttempts.trim() === '') { + return DEFAULT_FILE_RETENTION_MAX_ATTEMPTS; + } + + const value = Number(maxAttempts); + if (!Number.isInteger(value) || value < 1) { + return DEFAULT_FILE_RETENTION_MAX_ATTEMPTS; + } + return value; +} + +/** + * Backoff before the next attempt, doubling from one sweep interval up to a + * day. `attempts` is the file's failure count including the one just + * recorded, so the first retry lands on the following sweep. + * + * A file that exhausts `maxAttempts` is parked for a month instead. Parking + * rather than excluding is what keeps this safe against record reuse: a + * code-output row is repurposed for a repeated `(filename, conversationId)` + * and keeps fields it was not asked to change, so state that survived into + * a new lifecycle would strand a different object than the one it was + * recorded against. A deadline can only delay that object; a flag would + * lose it. + * + * The base tracks the configured interval so the schedule stays meaningful + * at any cadence — a fixed hour would be inert for the first three attempts + * of a six-hour sweep, and would skip twelve passes of a five-minute one. + * The floor keeps a pathologically short interval from spending the whole + * budget on a transient outage. + */ +export function getExpiredFileRetryDelay(attempts: number, maxAttempts: number): number { + if (attempts >= maxAttempts) { + return FILE_RETENTION_PARK_MS; + } + + const interval = getFileRetentionSweepInterval(); + const base = Math.max( + interval > 0 ? interval : DEFAULT_FILE_RETENTION_SWEEP_INTERVAL_MS, + MIN_FILE_RETENTION_RETRY_BASE_MS, + ); + const exponent = Math.max(0, attempts - 1); + return Math.min(base * 2 ** exponent, FILE_RETENTION_RETRY_MAX_MS); +} + export function getExpiredFileEndpoint(source?: string): string { return source === FileSources.azure ? EModelEndpoint.azureAssistants : EModelEndpoint.assistants; } @@ -205,18 +266,69 @@ export async function resolveExpiredFileSweepConfig({ export async function sweepExpiredFiles( { appConfig, limit = 100, loadAppConfig }: ExpiredFileSweepOptions | undefined = {}, - { getExpiredFiles, processDeleteRequest, logger }: SweepDependencies, + { + getExpiredFiles, + processDeleteRequest, + incrementFileDeletionAttempts, + deferExpiredFile, + logger, + }: SweepDependencies, ): Promise { + const maxAttempts = getFileRetentionMaxAttempts(); + /* Deadlines are anchored here rather than at the moment of failure. The + * interval timer is armed before the sweep runs, so a deadline measured + * after the deletion I/O lands just past the next scheduled pass and the + * file waits a whole extra interval. */ + const sweepStartedAt = Date.now(); const files = (await getExpiredFiles(limit)) ?? []; let resolvedAppConfig = appConfig; let deleted = 0; let failed = 0; + /* Every failure has to be recorded, not just counted: an unrecorded one + * leaves the file at the head of the next `expiredAt`-ordered batch, and + * a file that can never be deleted then crowds out every file that + * expired after it for the life of the deployment. */ + const recordFailure = async (file: ExpiredFile): Promise => { + failed++; + let attempts: number; + try { + /* The attempt number comes from the increment rather than from the + * snapshot this batch queried, so two nodes sweeping the same file + * each get a distinct one. */ + attempts = await incrementFileDeletionAttempts(file.file_id); + } catch (error) { + logger.error( + `[sweepExpiredFiles] Error recording failed deletion of expired file ${file.file_id}:`, + error, + ); + return; + } + + /* Exactly the attempt that lands on the cap reports it — `>=` would + * have every later node past the threshold repeat the notice once per + * replica, per file. */ + if (attempts === maxAttempts) { + logger.error( + `[sweepExpiredFiles] Parking expired file ${file.file_id} after ${attempts} failed deletions. Its backing storage was not removed and the record is kept, so the reference survives for reconciliation; the sweep will try again in about a month rather than on every pass.`, + ); + } + + try { + await deferExpiredFile( + file.file_id, + new Date(sweepStartedAt + getExpiredFileRetryDelay(attempts, maxAttempts)), + ); + } catch (error) { + logger.error(`[sweepExpiredFiles] Error deferring expired file ${file.file_id}:`, error); + } + }; + for (const file of files) { const userId = typeof file.user === 'string' ? file.user : file.user?.toString?.(); if (!userId) { logger.warn(`[sweepExpiredFiles] Skipping expired file without user: ${file.file_id}`); - failed++; + await recordFailure(file); continue; } @@ -229,21 +341,21 @@ export async function sweepExpiredFiles( const req = createExpiredFileSweepRequest({ appConfig: resolvedAppConfig, file, userId }); const { deletedFileIds, failedFileIds } = await processDeleteRequest({ req, files: [file] }); if (failedFileIds.includes(file.file_id)) { - failed++; + await recordFailure(file); continue; } if (deletedFileIds.includes(file.file_id)) { deleted++; } else { - failed++; logger.error( `[sweepExpiredFiles] Delete request finished without resolving expired file ${file.file_id}`, ); + await recordFailure(file); } } catch (error) { - failed++; logger.error(`[sweepExpiredFiles] Error deleting expired file ${file.file_id}:`, error); + await recordFailure(file); } } diff --git a/packages/data-schemas/src/methods/file.spec.ts b/packages/data-schemas/src/methods/file.spec.ts index d4235b03fd..e1ce4f49a9 100644 --- a/packages/data-schemas/src/methods/file.spec.ts +++ b/packages/data-schemas/src/methods/file.spec.ts @@ -380,10 +380,196 @@ describe('File Methods', () => { true, ); - const files = await fileMethods.getExpiredFiles(100, now); + const files = await fileMethods.getExpiredFiles(100, { now }); expect(files.map((file) => file.file_id)).toEqual([expiredFileId]); }); + + /** Seeds a sweep-eligible file, then drives its retry state through the + * same methods the sweep itself uses. */ + const seedExpiredFile = async ({ + file_id, + expiredAt, + attempts = 0, + retryAt, + }: { + file_id: string; + expiredAt: Date; + attempts?: number; + retryAt?: Date; + }) => { + await fileMethods.createFile( + { + file_id, + user: new mongoose.Types.ObjectId(), + filename: `${file_id}.txt`, + filepath: `/uploads/${file_id}.txt`, + type: 'text/plain', + bytes: 100, + expiredAt, + }, + true, + ); + for (let i = 0; i < attempts; i++) { + await fileMethods.incrementFileDeletionAttempts(file_id); + } + if (retryAt) { + await fileMethods.deferExpiredFile(file_id, retryAt); + } + }; + + it('holds back files whose deletion backoff has not elapsed', async () => { + const now = new Date('2030-01-01T00:00:00.000Z'); + const readyFileId = uuidv4(); + const backedOffFileId = uuidv4(); + + await seedExpiredFile({ + file_id: backedOffFileId, + expiredAt: new Date('2029-12-01T00:00:00.000Z'), + attempts: 2, + retryAt: new Date('2030-01-01T00:00:01.000Z'), + }); + await seedExpiredFile({ + file_id: readyFileId, + expiredAt: new Date('2029-12-31T00:00:00.000Z'), + attempts: 2, + retryAt: new Date('2029-12-31T23:00:00.000Z'), + }); + + const files = await fileMethods.getExpiredFiles(100, { now }); + + expect(files.map((file) => file.file_id)).toEqual([readyFileId]); + }); + + it('holds a parked file back without excluding it for good', async () => { + const parkedFileId = uuidv4(); + const retriableFileId = uuidv4(); + + /* The parked file expired first, so with nothing holding it back it + * sorts to the front of every batch and a limit-sized run of them + * starves everything that expired afterwards. */ + await seedExpiredFile({ + file_id: parkedFileId, + expiredAt: new Date('2029-01-01T00:00:00.000Z'), + attempts: 10, + retryAt: new Date('2030-02-01T00:00:00.000Z'), + }); + await seedExpiredFile({ + file_id: retriableFileId, + expiredAt: new Date('2029-12-31T00:00:00.000Z'), + attempts: 9, + }); + + const parked = await fileMethods.getExpiredFiles(100, { + now: new Date('2030-01-01T00:00:00.000Z'), + }); + expect(parked.map((file) => file.file_id)).toEqual([retriableFileId]); + + /* Nothing is excluded permanently — a record reused for different + * content in the meantime gets its object swept rather than stranded. */ + const afterPark = await fileMethods.getExpiredFiles(100, { + now: new Date('2030-02-01T00:00:01.000Z'), + }); + expect(afterPark.map((file) => file.file_id)).toEqual([parkedFileId, retriableFileId]); + }); + + it('leaves retry state alone on writes that touch the record', async () => { + const fileId = uuidv4(); + await seedExpiredFile({ + file_id: fileId, + expiredAt: new Date('2029-01-01T00:00:00.000Z'), + attempts: 4, + retryAt: new Date('2030-02-01T00:00:00.000Z'), + }); + + /* `prepareImagesLocal` clears the upload TTL with nothing but the id + * on every reuse of an image, the deferred preview only transitions + * `status`, and `processCodeOutput` repurposes a row for new content. + * None of them need to reason about sweep bookkeeping: the deferral + * is a deadline, so the worst a survivor can do is delay. */ + await fileMethods.updateFile({ file_id: fileId }); + await fileMethods.updateFile({ file_id: fileId, status: 'ready' }); + + const [file] = (await fileMethods.getFiles({ file_id: fileId }))!; + expect(file.deletionAttempts).toBe(4); + expect(file.deletionRetryAt).toEqual(new Date('2030-02-01T00:00:00.000Z')); + }); + + it('records retry bookkeeping without posing as a content write', async () => { + const fileId = uuidv4(); + await seedExpiredFile({ file_id: fileId, expiredAt: new Date('2029-01-01T00:00:00.000Z') }); + const [before] = (await fileMethods.getFiles({ file_id: fileId }))!; + + /* `processCodeOutput` falls back to `updatedAt` as the writer-order + * stamp for records predating `metadata.sourceDispatchedAt`. A sweep + * that bumped it would look like a newer content writer and a + * background harvest would drop its attachment. */ + await fileMethods.incrementFileDeletionAttempts(fileId); + await fileMethods.deferExpiredFile(fileId, new Date('2030-06-01T00:00:00.000Z')); + + const [after] = (await fileMethods.getFiles({ file_id: fileId }))!; + expect(after.deletionAttempts).toBe(1); + expect(after.updatedAt).toEqual(before.updatedAt); + }); + }); + + describe('deferExpiredFile', () => { + const createDeferrableFile = async (file_id: string) => { + await fileMethods.createFile( + { + file_id, + user: new mongoose.Types.ObjectId(), + filename: 'deferred.txt', + filepath: '/uploads/deferred.txt', + type: 'text/plain', + bytes: 100, + expiredAt: new Date('2029-12-31T00:00:00.000Z'), + }, + true, + ); + }; + + it('stamps the next retry and counts the attempt', async () => { + const fileId = uuidv4(); + const retryAt = new Date('2030-02-01T00:00:00.000Z'); + await createDeferrableFile(fileId); + + expect(await fileMethods.incrementFileDeletionAttempts(fileId)).toBe(1); + await fileMethods.deferExpiredFile(fileId, retryAt); + + const [file] = (await fileMethods.getFiles({ file_id: fileId }))!; + expect(file.deletionAttempts).toBe(1); + expect(file.deletionRetryAt).toEqual(retryAt); + }); + + it('hands concurrent sweeps distinct attempt numbers', async () => { + const fileId = uuidv4(); + await createDeferrableFile(fileId); + + /* Two nodes recording a failure for the same file in the same pass. + * Deriving the attempt from the queried snapshot would give both the + * same number and neither would see the cap being reached. */ + const attempts = await Promise.all([ + fileMethods.incrementFileDeletionAttempts(fileId), + fileMethods.incrementFileDeletionAttempts(fileId), + fileMethods.incrementFileDeletionAttempts(fileId), + ]); + + expect([...attempts].sort()).toEqual([1, 2, 3]); + }); + + it('never pulls a deferral earlier than one already committed', async () => { + const fileId = uuidv4(); + const later = new Date('2030-02-01T00:00:00.000Z'); + const earlier = new Date('2030-01-01T00:00:00.000Z'); + await createDeferrableFile(fileId); + + await fileMethods.deferExpiredFile(fileId, later); + await fileMethods.deferExpiredFile(fileId, earlier); + + const [file] = (await fileMethods.getFiles({ file_id: fileId }))!; + expect(file.deletionRetryAt).toEqual(later); + }); }); describe('getToolFilesByIds', () => { diff --git a/packages/data-schemas/src/methods/file.ts b/packages/data-schemas/src/methods/file.ts index b145ea65f9..97c704f694 100644 --- a/packages/data-schemas/src/methods/file.ts +++ b/packages/data-schemas/src/methods/file.ts @@ -9,6 +9,10 @@ export type FileOwnerScope = { tenantId?: string | null; }; +export type ExpiredFileQueryOptions = { + now?: Date; +}; + function withOwnerScope>( filter: T, ownerScope?: FileOwnerScope, @@ -35,7 +39,9 @@ export function createFileMethods(mongoose: typeof import('mongoose')): { _sortOptions?: Record | null, selectFields?: Record | string | null, ) => Promise; - getExpiredFiles: (limit?: number, now?: Date) => Promise; + getExpiredFiles: (limit?: number, options?: ExpiredFileQueryOptions) => Promise; + incrementFileDeletionAttempts: (file_id: string) => Promise; + deferExpiredFile: (file_id: string, deletionRetryAt: Date) => Promise; getToolFilesByIds: ( fileIds: string[], toolResourceSet?: Set, @@ -130,14 +136,80 @@ export function createFileMethods(mongoose: typeof import('mongoose')): { return await query.sort(sortOptions).lean(); } - async function getExpiredFiles(limit = 100, now: Date = new Date()): Promise { + /** + * Expired files the sweep may attempt right now, oldest deadline first. + * + * `deletionRetryAt` is the only thing holding a file back: one that keeps + * failing is deferred on a growing backoff, and one that exhausts its + * attempts is parked far enough out to stop crowding the batch. Without + * it a permanently undeletable file sorts to the front of this bounded + * batch on every pass and starves every file that expired after it. + * + * The deferral is deliberately a deadline rather than a flag, so nothing + * here is ever excluded for good. File records are reused across content + * lifecycles — a code-output row is repurposed for a repeated + * `(filename, conversationId)`, keeping fields it was not asked to change + * — so bookkeeping that permanently excluded a row would eventually + * strand a *different* object than the one it was recorded against. A + * deadline can only ever delay that; it cannot lose it. + * + * An absent field means "never attempted", so records written before it + * existed remain eligible. + */ + async function getExpiredFiles( + limit = 100, + { now = new Date() }: ExpiredFileQueryOptions = {}, + ): Promise { const File = mongoose.models.File as Model; - return await File.find({ expiredAt: { $ne: null, $lte: now } }) + return await File.find({ + expiredAt: { $ne: null, $lte: now }, + $or: [{ deletionRetryAt: null }, { deletionRetryAt: { $lte: now } }], + }) .sort({ expiredAt: 1 }) .limit(limit) .lean(); } + /** + * Records one failed sweep deletion and returns the file's resulting + * consecutive-failure count. + * + * The count comes back from the increment itself rather than being + * re-derived from the caller's snapshot. Two nodes sweeping the same file + * would otherwise each read the same value and each believe itself to be + * the same attempt, pushing the stored count past the give-up cap while + * both still think it is below — so the give-up would never be reported. + * Returning it here gives every caller a distinct attempt number. + */ + async function incrementFileDeletionAttempts(file_id: string): Promise { + const File = mongoose.models.File as Model; + const file = await File.findOneAndUpdate( + { file_id }, + { $inc: { deletionAttempts: 1 } }, + /** `timestamps: false`: sweep bookkeeping is not a content write. + * `processCodeOutput` falls back to `updatedAt` as the writer-order + * stamp for records predating `metadata.sourceDispatchedAt`, so + * bumping it here would make a failed sweep look like a newer writer + * and a background harvest would drop its attachment. Same reasoning + * as `claimCodeFile`. */ + { new: true, projection: { deletionAttempts: 1 }, timestamps: false }, + ).lean | null>(); + + return file?.deletionAttempts ?? 0; + } + + /** + * Holds a file back from the sweep until `deletionRetryAt`. + * + * Written with `$max` so a deferral can only ever move later. A node that + * computed a shorter backoff from a lower attempt count cannot pull the + * file forward past a longer one another node already committed. + */ + async function deferExpiredFile(file_id: string, deletionRetryAt: Date): Promise { + const File = mongoose.models.File as Model; + await File.updateOne({ file_id }, { $max: { deletionRetryAt } }, { timestamps: false }).exec(); + } + /** * Retrieves tool files (files that are embedded or have a fileIdentifier) from an array of file IDs. * Note: execute_code files are handled separately by getCodeGeneratedFiles. @@ -679,6 +751,8 @@ export function createFileMethods(mongoose: typeof import('mongoose')): { findFileById, getFiles, getExpiredFiles, + incrementFileDeletionAttempts, + deferExpiredFile, getToolFilesByIds, getCodeGeneratedFiles, getUserCodeFiles, diff --git a/packages/data-schemas/src/schema/file.ts b/packages/data-schemas/src/schema/file.ts index 208ad67bdf..0b075eef68 100644 --- a/packages/data-schemas/src/schema/file.ts +++ b/packages/data-schemas/src/schema/file.ts @@ -152,6 +152,18 @@ const file: Schema = new Schema( * backing storage first, then removes this metadata record. */ type: Date, }, + deletionAttempts: { + /* Consecutive failed sweep deletions, driving the backoff below. A + * file at or past `FILE_RETENTION_SWEEP_MAX_ATTEMPTS` is parked + * rather than retried on the ordinary ladder. */ + type: Number, + }, + deletionRetryAt: { + /* Earliest next sweep attempt, backed off from `deletionAttempts`. + * The sweep's only hold: a deadline, never a flag, so a record reused + * for different content can be delayed but not stranded. */ + type: Date, + }, }, { timestamps: true, diff --git a/packages/data-schemas/src/types/file.ts b/packages/data-schemas/src/types/file.ts index 8038512144..0e6b950aa0 100644 --- a/packages/data-schemas/src/types/file.ts +++ b/packages/data-schemas/src/types/file.ts @@ -76,6 +76,19 @@ export interface IMongoFile extends Omit { }; expiresAt?: Date; expiredAt?: Date | null; + /** + * Consecutive failed retention-sweep deletions. The sweep backs off + * between attempts and parks the file once this reaches the configured + * cap, so a file whose backing storage refuses deletion cannot occupy + * the sweep's bounded queue. Absent until the first failure. + */ + deletionAttempts?: number; + /** + * Earliest time the retention sweep may retry this file, and the only + * thing holding it back. Set alongside `deletionAttempts` on every + * failure; absent until the first one. + */ + deletionRetryAt?: Date; createdAt?: Date; updatedAt?: Date; tenantId?: string;