mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-23 18:43:21 +00:00
🪫 fix: Back Off Failed Retention Deletes (#15608)
* 🧹 fix: Stop Undeletable Files Starving the Retention Sweep `getExpiredFiles` returns the oldest `expiredAt` first, capped at `limit`, and `processDeleteRequest` leaves the record in place when storage deletion fails. Nothing records the failure, so the same files come back at the head of the next batch an hour later, forever: no backoff, no cap, and — once `limit` of them cannot be deleted — no file that expires afterwards is ever swept again. On the deployment behind #15511 that is ~29k stranded objects permanently occupying a 100-slot queue, which is why fixing the Code Interpreter side alone (LibreChat-AI/code-interpreter#85) would not have resumed deletion there. Failures are now recorded on the file. `deletionRetryAt` holds it back with a backoff doubling from one sweep interval to a day, and `deletionAttempts` retires it from the query entirely once it reaches `FILE_RETENTION_SWEEP_MAX_ATTEMPTS` (10, so roughly five days of retries). Both fields are absent on existing records and absence means "never attempted", so nothing already in the collection changes eligibility. The record itself is kept rather than deleted — the reference is what an operator needs to reconcile a bucket the sweep could not clear, and dropping it would restore the silence that made this leak invisible. Raising `FILE_RETENTION_SWEEP_MAX_ATTEMPTS` re-admits everything previously given up on, which is the supported way to resume once the storage-side failure is fixed; the give-up is logged with that instruction. The counter is incremented server-side with `$inc` so concurrent sweeps on separate nodes cannot overwrite each other's progress toward the cap, and a failure to record a failure is logged and skipped rather than aborting the rest of the batch. * 🧭 fix: Delete Code Environment Files Through the Route That Exists `deleteCodeEnvFile` tried `/sessions/:sid/objects/:fid` before falling back to `/files/:sid/:fid`. Both have been there since #13424, but only the second is mounted by any released codeapi — the first gained DELETE in LibreChat-AI/code-interpreter#85 — so every deletion paid a guaranteed 404 and a wasted round trip, and #15511 read that 404 as the whole bug. Call `/files/:sid/:fid` directly. It is the safe direction to collapse toward: codeapi has mounted it since its first release, so this works against older deployments as well as post-#85 ones, whereas keeping the other path would not. Collapsing the loop tightens two behaviours that only existed to serve it. A 405 now surfaces instead of being swallowed on the way to a second attempt; there is no second route to try, and a service that refuses the method should say so. A 404 is still treated as "already gone" — that is the only thing it can now mean — but it is logged rather than passed over in silence, because a 404 caused by a misconfigured base URL looks identical and this branch drops the file's metadata record either way. * 🔒 fix: Settle sweep retry state from the write, not the read Two findings from the review of0e52924. - `recordFailure` derived the attempt number by adding one to the count the batch had queried. Two nodes sweeping the same file read the same value, so both believed themselves to be the same attempt: each `$inc` landed, the stored count crossed `FILE_RETENTION_SWEEP_MAX_ATTEMPTS`, and neither caller ever saw the threshold. The file drops out of the query — the starvation guard still holds — but the give-up is never reported, and that log line is the operator's only notice, and carries the instruction for resuming. Return the count from the increment itself so every caller gets a distinct attempt number and exactly one observes the cap. The same staleness shortened the backoff, so `deferExpiredFile` now writes with `$max`: a deferral can only move later, and a node that computed a shorter delay cannot pull the file forward past one another node already committed. - The backoff doubled from a hard-coded hour while claiming to start from one sweep interval. At the default they coincide; away from it the schedule stops meaning anything — on a six-hour sweep the first three attempts all land on consecutive passes, and on a five-minute one the first retry skips twelve. Derive the base from `FILE_RETENTION_SWEEP_INTERVAL_MS`, floored at a minute so a pathologically short interval cannot spend the whole give-up budget on a transient outage. * 🧯 fix: Keep the give-up notice and the retry budget honest Four findings from the review ofee1bac5. - The give-up was reported after the deferral write, inside the same catch. Once the increment lands the counter is durable, so `getExpiredFiles` already excludes the file; a deferral that then failed left only the generic recording error and dropped the one line naming the file and saying how to resume. Report it as soon as the increment returns, and let the deferral fail on its own. - Retry deadlines were measured from `Date.now()` after the deletion I/O, but `startExpiredFileSweep` arms its interval before the sweep runs. A one-interval delay therefore expired just *after* the next scheduled pass, which skipped the file and pushed its first retry out by a whole extra interval — and the same drift applied to every delay that is an exact multiple. Anchor deadlines to the sweep's start instead. - The threshold used `>=`, so once two nodes pushed a file past the cap every one of them past it logged the give-up: one error per replica per exhausted file rather than one actionable notice. Only the attempt that lands exactly on the cap reports it. - Retry state outlived the content it described. `processCodeOutput` reuses a record for a repeated `(filename, conversationId)` — new bytes, new storage key, and a fresh `expiredAt` from `getRetentionExpiry` — while `createFile` and `updateFile` set only supplied fields. A record carried to the cap by its previous content stayed excluded from the sweep forever, stranding the new object exactly as this PR set out to prevent; a partially failed one started the new object's budget already spent. Both write paths now clear the fields: the budget belongs to the storage a record currently points at. The retry-state fixtures in `file.spec.ts` were seeded through `createFile`, which now clears them, so they drive the real methods the sweep uses. * 🎯 fix: Clear the retry budget only when a retention lifecycle starts Both findings from the review of8d26f9b, and both are consequences of that commit's reset rather than of the original change. - The reset was unconditional on every `createFile`/`updateFile`, on the reasoning that those paths write content. Two of them do not. `prepareImages{Local,Azure,Firebase}` call `updateFile({ file_id })` with nothing but the id — a TTL touch — every time an existing image is encoded for another chat, and the deferred preview uses the same method to transition `status`. Either handed a stranded record a fresh set of attempts and another give-up notice, on repeat, defeating the cap for the files most likely to be stranded. Gate it on the write actually setting `expiredAt`. The budget belongs to a retention lifecycle: that is the write which starts a new one, it is what `processCodeOutput` supplies when it repurposes a record, and a record with no retention deadline is never swept, so the fields are inert there anyway. - Both retry writes bumped `updatedAt`. `processCodeOutput` falls back to `updatedAt` as the writer-order stamp for records that predate `metadata.sourceDispatchedAt`, so a failed sweep landing mid-harvest read as a newer content writer and the harvest dropped its attachment. Mark them `timestamps: false`, for the reason `claimCodeFile` already does: bookkeeping is not a content write. * 🅿️ refactor: Park exhausted files instead of excluding them Three review rounds in a row found defects in how the give-up cap interacts with record reuse, each in the fix for the last. That is a design error, not a bug list: a permanent exclusion has to be bound precisely to the content lifecycle it was recorded against, and File records outlive their content. `processCodeOutput` repurposes a row for a repeated `(filename, conversationId)`, `createFile`/`updateFile` set only supplied fields, and `getRetentionExpiry` returns `{}` on a lookup failure so the row inherits its old deadline — three separate ways for bookkeeping to survive into a lifecycle it does not describe, each needing its own guard, and the two retry writes needing to be lifecycle-conditional on top. Remove the category instead. `deletionRetryAt` becomes the sweep's only hold, and reaching `FILE_RETENTION_SWEEP_MAX_ATTEMPTS` parks the file for a month rather than excluding it. The bound on the batch is the same — a stranded file costs one slot a month instead of one an hour — but a deadline that outlives its content can only delay the next object, never lose it, so nothing outside the sweep has to reason about this state at all. That deletes more than it adds: - `createFile` and `updateFile` go back to their original form. No reset, so no question of which writes install content, and `prepareImages*` and the deferred preview stop mattering here. - `getExpiredFiles` loses `maxAttempts` and its `$and`; eligibility is one `$or` on the deadline. - The interleaving race between the increment and the deferral degrades from a stranded object to a delayed one. Working through a large backlog is throughput-bound either way: every attempt costs a slot in the bounded batch, so N stranded files need N × `FILE_RETENTION_SWEEP_MAX_ATTEMPTS` passes to settle. Lower that limit when recovering a deployment that has accumulated many.
This commit is contained in:
Danny Avila
GitHub
parent
84323d8c48
commit
fe30e3db32
10 changed files with 695 additions and 86 deletions
|
|
@ -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}`,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -349,6 +349,8 @@ async function sweepExpiredFiles(options = {}) {
|
|||
return sweepExpiredFilesWithDeps(options, {
|
||||
getExpiredFiles: db.getExpiredFiles,
|
||||
processDeleteRequest,
|
||||
incrementFileDeletionAttempts: db.incrementFileDeletionAttempts,
|
||||
deferExpiredFile: db.deferExpiredFile,
|
||||
logger,
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<number>;
|
||||
deferExpiredFile: (file_id: string, deletionRetryAt: Date) => Promise<void>;
|
||||
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<ExpiredFileSweepResult> {
|
||||
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<void> => {
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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', () => {
|
||||
|
|
|
|||
|
|
@ -9,6 +9,10 @@ export type FileOwnerScope = {
|
|||
tenantId?: string | null;
|
||||
};
|
||||
|
||||
export type ExpiredFileQueryOptions = {
|
||||
now?: Date;
|
||||
};
|
||||
|
||||
function withOwnerScope<T extends FilterQuery<IMongoFile>>(
|
||||
filter: T,
|
||||
ownerScope?: FileOwnerScope,
|
||||
|
|
@ -35,7 +39,9 @@ export function createFileMethods(mongoose: typeof import('mongoose')): {
|
|||
_sortOptions?: Record<string, SortOrder> | null,
|
||||
selectFields?: Record<string, 0 | 1> | string | null,
|
||||
) => Promise<IMongoFile[] | null>;
|
||||
getExpiredFiles: (limit?: number, now?: Date) => Promise<IMongoFile[]>;
|
||||
getExpiredFiles: (limit?: number, options?: ExpiredFileQueryOptions) => Promise<IMongoFile[]>;
|
||||
incrementFileDeletionAttempts: (file_id: string) => Promise<number>;
|
||||
deferExpiredFile: (file_id: string, deletionRetryAt: Date) => Promise<void>;
|
||||
getToolFilesByIds: (
|
||||
fileIds: string[],
|
||||
toolResourceSet?: Set<EToolResources>,
|
||||
|
|
@ -130,14 +136,80 @@ export function createFileMethods(mongoose: typeof import('mongoose')): {
|
|||
return await query.sort(sortOptions).lean<IMongoFile[]>();
|
||||
}
|
||||
|
||||
async function getExpiredFiles(limit = 100, now: Date = new Date()): Promise<IMongoFile[]> {
|
||||
/**
|
||||
* 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<IMongoFile[]> {
|
||||
const File = mongoose.models.File as Model<IMongoFile>;
|
||||
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<IMongoFile[]>();
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<number> {
|
||||
const File = mongoose.models.File as Model<IMongoFile>;
|
||||
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<Pick<IMongoFile, 'deletionAttempts'> | 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<void> {
|
||||
const File = mongoose.models.File as Model<IMongoFile>;
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -152,6 +152,18 @@ const file: Schema<IMongoFile> = 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,
|
||||
|
|
|
|||
|
|
@ -76,6 +76,19 @@ export interface IMongoFile extends Omit<Document, 'model'> {
|
|||
};
|
||||
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;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue