From acfc264bf940c117a4f4ac41a78a3691e3de25dd Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Tue, 28 Jul 2026 13:46:43 -0400 Subject: [PATCH] =?UTF-8?q?fix:=20Codex=20round=2012=20=E2=80=94=20arming?= =?UTF-8?q?=20retry=20safety,=20honest=20deletes,=20topology-free=20erase,?= =?UTF-8?q?=20bounded=20file=20holds?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P1: a thrown (or ambiguously acknowledged) arming write now rolls the committed row back via compensateLateCreate before answering 500. Leaving it meant a client retry minted a second row and the reconciler's unarmed sweep later armed the first as well — one intended schedule became several recurring, billable ones. - deleteScheduleForOwner drains with the quiesce discipline and reports honestly (ScheduleDeleteResult): a provably job-less run is settled and erased synchronously; a live run's abort must be DELIVERED or the delete answers 503 instead of claiming a possibly still-billing generation was stopped; a delivered-but-unsettled abort answers 202 draining. - Erase-on-settle: whichever process records a run's terminal outcome also attempts the deferred erase of a deleting schedule, so draining converges in every topology — the clustered entrypoint runs no reconciler, and without this a hidden schedule's prompt survived there indefinitely. The DELETE gates stay open on this basis. - Schedule attachments now take a bounded renewable hold (extendFilesTTL, the #14470 pattern) instead of a permanent TTL unset: renewed at create/edit and each fire preflight, cleared permanently by the first real consumption, and allowed to lapse when the schedule dies first — no more uploads retained forever by a schedule deleted before its first run. The now-unused data-schemas retainFiles (added earlier in this PR) is removed. Every fix carries a test verified by neutering (fails) and restoring (passes). --- api/server/experimental.js | 9 +- api/server/routes/schedules.js | 43 +++-- packages/api/src/schedules/handlers.spec.ts | 65 ++++++- packages/api/src/schedules/handlers.ts | 76 +++++--- packages/api/src/schedules/service.spec.ts | 188 +++++++++++++++++++- packages/api/src/schedules/service.ts | 123 +++++++++++-- packages/api/src/schedules/types.ts | 27 +++ packages/data-schemas/src/methods/file.ts | 28 --- 8 files changed, 476 insertions(+), 83 deletions(-) diff --git a/api/server/experimental.js b/api/server/experimental.js index d39d55aac0..3343c81382 100644 --- a/api/server/experimental.js +++ b/api/server/experimental.js @@ -260,9 +260,12 @@ if (cluster.isMaster) { * scheduling only, and this process never arms the engine, so accepting a create or * run-now here would persist a schedule nothing will ever fire. Reads stay open so an * operator can still inspect existing schedules. DELETE stays open too: erasing a - * stored prompt needs no engine (the handler quiesces through the job store and - * refuses unsafe cases itself), and a deployment switched to clustered mode must not - * strand users with schedules they can see but never remove. + * stored prompt needs no engine and no reconciler — the delete path settles runs + * whose job is provably absent synchronously, refuses honestly (503) when a run + * cannot be confirmed stopped, and a delivered abort erases on the generation's own + * outcome write (erase-on-settle) — so a deployment switched to clustered mode is + * never stranded with schedules users can see but never remove, and never left + * holding a hidden schedule's prompt indefinitely. */ const rejectScheduleWritesUntilReady = (req, res, next) => { if (schedulesReady || SCHEDULE_ENGINE_OPTIONAL_METHODS.has(req.method)) { diff --git a/api/server/routes/schedules.js b/api/server/routes/schedules.js index 6dac17a1f1..5a9326110d 100644 --- a/api/server/routes/schedules.js +++ b/api/server/routes/schedules.js @@ -1,6 +1,11 @@ const express = require('express'); const { Permissions, PermissionTypes } = require('librechat-data-provider'); -const { isEnabled, createSchedulesHandlers, generateCheckAccess } = require('@librechat/api'); +const { + isEnabled, + SCHEDULE_FILE_HOLD, + generateCheckAccess, + createSchedulesHandlers, +} = require('@librechat/api'); const { requireJwtAuth, configMiddleware, messageIpLimiter } = require('~/server/middleware'); const { getLimits, @@ -43,22 +48,32 @@ const handlers = createSchedulesHandlers({ return (files ?? []).map((file) => file.file_id); }, markFilesUsed: async (fileIds, userId) => { - // IDEMPOTENT retention: clears the upload TTL without touching the usage counter. - // The previous implementation went through updateFilesUsage, whose `$inc: {usage}` - // means "consumed by a message" — so every retry, and every schedule PATCH that - // resent unchanged file_ids, inflated the counter with no decrement anywhere. - // Still verifies EVERY requested file was retained: a file can be deleted between - // the ownership check and here, and a silent success would persist a schedule whose - // attachments the first fire drops. - const requested = new Set(fileIds).size; - const retained = await methods.retainFiles(fileIds, { userId }); - if (retained !== requested) { - throw new Error(`attachment retention incomplete: ${retained}/${requested} files retained`); + // BOUNDED renewable hold (extendFilesTTL), not a permanent `$unset` of the upload + // TTL: permanence made a schedule deleted before its first run, an edit that + // replaced file_ids, or a failed creation leak the upload forever, since nothing + // ever restored an expiry. The first fire that actually SENDS the file clears its + // TTL through the ordinary consumption path; until then the hold is renewed at + // create/edit and each fire preflight, and lapses when the schedule stops touching + // it. Files already made permanent by a real send are skipped by construction. + // Idempotent, and never touches the usage counter (a retention is not a consumption). + // Then VERIFY every requested file still exists: one can be deleted between the + // ownership check and here, and a silent success would persist a schedule whose + // attachments the first fire drops. Existence is the check — not the hold's + // modified-count, which reads 0 for an already-permanent or already-held file. + const unique = [...new Set(fileIds)]; + await methods.extendFilesTTL(unique, SCHEDULE_FILE_HOLD, { user: userId }); + const files = await methods.getFiles({ file_id: { $in: unique }, user: userId }, null, { + file_id: 1, + }); + const present = (files ?? []).length; + if (present !== unique.length) { + throw new Error(`attachment retention incomplete: ${present}/${unique.length} files exist`); } }, fireNow: fireScheduleNow, - // Quiesce-then-erase delete: stops new claims, aborts in-flight loopback runs, - // and erases once drained (reconciler completes drain) so evidence is preserved. + // Quiesce-then-erase delete: stops new claims, settles provably job-less runs + // synchronously, aborts live ones, and reports honestly (see ScheduleDeleteResult); + // a delivered abort erases on the generation's own outcome write, in any topology. deleteSchedule: deleteScheduleForOwner, // Durable account-deletion barrier. A one-shot disable scan cannot close the // create race, so every scheduling WRITE consults the user-level flag instead. diff --git a/packages/api/src/schedules/handlers.spec.ts b/packages/api/src/schedules/handlers.spec.ts index 509f2d91d8..af2ee7c130 100644 --- a/packages/api/src/schedules/handlers.spec.ts +++ b/packages/api/src/schedules/handlers.spec.ts @@ -97,7 +97,9 @@ describe('toWireSchedule', () => { /** Minimal Express double capturing the status/body the handler settled on. */ function makeRes() { - const captured: { status?: number; body?: unknown } = {}; + const captured: { status?: number; body?: unknown; headers: Record } = { + headers: {}, + }; const res = { status(code: number) { captured.status = code; @@ -107,6 +109,10 @@ function makeRes() { captured.body = payload; return this; }, + set(name: string, value: string) { + captured.headers[name] = value; + return this; + }, }; return { res: res as unknown as Response, captured }; } @@ -235,4 +241,61 @@ describe('createSchedule late-create compensation', () => { expect(captured.status).toBe(410); expect(captured.status).not.toBe(201); }); + + /** + * A thrown (or ambiguously acknowledged) arming write must roll the committed row + * back: the client retries the failed create with a fresh UUID, the retry commits a + * second row, and the reconciler's unarmed sweep later arms the FIRST as well — one + * intended schedule becomes several recurring, billable ones. + */ + it('rolls back the committed row when the arming write throws', async () => { + const deps = makeCreateDeps({ isUserDeleting: jest.fn(async () => false) }); + (deps.methods.updateScheduleById as jest.Mock).mockRejectedValue(new Error('mongo down')); + const { res, captured } = makeRes(); + await createSchedulesHandlers(deps).createSchedule(makeCreateReq(), res); + + expect(captured.status).toBe(500); + expect(deps.methods.deleteScheduleById).toHaveBeenCalledWith(expect.any(String), 'user-1'); + }); +}); + +describe('deleteSchedule result mapping', () => { + function makeDeleteReq(): ServerRequest { + return { + params: { id: 'sched-1' }, + user: { id: 'user-1', tenantId: 't1', role: 'USER' }, + } as unknown as ServerRequest; + } + + const withResult = (result: string) => + makeCreateDeps({ + deleteSchedule: jest.fn(async () => result), + } as Partial); + + it('404s when the schedule is not found', async () => { + const { res, captured } = makeRes(); + await createSchedulesHandlers(withResult('not_found')).deleteSchedule(makeDeleteReq(), res); + expect(captured.status).toBe(404); + }); + + it('answers 200 when drained and erased', async () => { + const { res, captured } = makeRes(); + await createSchedulesHandlers(withResult('deleted')).deleteSchedule(makeDeleteReq(), res); + expect(captured.status ?? 200).toBe(200); + expect(captured.body).toEqual({ id: 'sched-1' }); + }); + + it('answers 202 while a delivered abort is still settling', async () => { + const { res, captured } = makeRes(); + await createSchedulesHandlers(withResult('draining')).deleteSchedule(makeDeleteReq(), res); + expect(captured.status).toBe(202); + expect(captured.body).toEqual({ id: 'sched-1' }); + }); + + it('refuses honestly when the active run could not be confirmed stopped', async () => { + // Reporting success would claim a possibly still-billing generation was stopped. + const { res, captured } = makeRes(); + await createSchedulesHandlers(withResult('unconfirmed')).deleteSchedule(makeDeleteReq(), res); + expect(captured.status).toBe(503); + }); }); diff --git a/packages/api/src/schedules/handlers.ts b/packages/api/src/schedules/handlers.ts index b2412137d9..5cf95b1c55 100644 --- a/packages/api/src/schedules/handlers.ts +++ b/packages/api/src/schedules/handlers.ts @@ -4,7 +4,13 @@ import { createSchedulePayloadSchema, updateSchedulePayloadSchema } from 'librec import type { TCreateSchedule, TUpdateSchedule } from 'librechat-data-provider'; import type { ScheduleMethods, ISchedule } from '@librechat/data-schemas'; import type { Response } from 'express'; -import type { ScheduleLimits, ScheduleUserContext, FireResult, FireableSchedule } from './types'; +import type { + ScheduleDeleteResult, + ScheduleUserContext, + FireableSchedule, + ScheduleLimits, + FireResult, +} from './types'; import type { ServerRequest } from '~/types'; import { isValidTimezone, cadenceIntervalMinutes, computeNextRunAt } from './cadence'; @@ -15,15 +21,17 @@ export interface SchedulesHandlersDeps { canViewAgent: (agentId: string, req: ServerRequest) => Promise; /** Filters to file ids owned by the user. */ filterOwnedFileIds: (fileIds: string[], userId: string) => Promise; - /** Clears the upload TTL on attached files so they survive until the first run. */ + /** Extends a bounded renewable upload hold on attached files so they survive to the + * first fire, which consumes them permanently; a schedule that dies first lets the + * hold lapse instead of retaining the upload forever. Throws when any file is gone. */ markFilesUsed: (fileIds: string[], userId: string) => Promise; /** Serialized manual fire (acquires the schedule lease); null if already leased. */ fireNow: (schedule: FireableSchedule, limits: ScheduleLimits) => Promise; /** * Soft-deletes a schedule with quiescing: stops new claims, aborts in-flight - * runs, and erases once drained. Returns false when not found / already deleting. + * runs, and erases once drained. See ScheduleDeleteResult for the honest states. */ - deleteSchedule: (id: string, userId: string) => Promise; + deleteSchedule: (id: string, userId: string) => Promise; /** Whether this user's account deletion has begun. Fail-closed (unknown == true). */ isUserDeleting: (userId: string) => Promise; } @@ -71,7 +79,7 @@ async function rejectIfUserDeleting( return true; } -/** Bounded attempts to clear the upload TTL on a schedule's attachments. */ +/** Bounded attempts to extend the upload hold on a schedule's attachments. */ const FILE_RETAIN_ATTEMPTS = 3; /** Public projection of a schedule — an allowlist of the `TSchedule` fields, so internal @@ -171,10 +179,11 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH } /** - * Clears the upload TTL on attached files (so they survive to the first fire), - * with bounded retry. Returns false when it exhausts retries — the caller then - * compensates (roll back the create / revert the edit) so a persisted schedule - * never references files the upload sweep is about to reap. + * Extends the bounded upload hold on attached files (so they survive to the first + * fire, which consumes them permanently), with bounded retry. Returns false when it + * exhausts retries — the caller then compensates (roll back the create / revert the + * edit) so a persisted schedule never references files the upload sweep is about to + * reap. */ async function retainFiles(fileIds: string[], userId: string): Promise { for (let attempt = 1; attempt <= FILE_RETAIN_ATTEMPTS; attempt++) { @@ -297,16 +306,29 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH res.status(410).json({ error: 'This account is being deleted' }); return; } - // ARM last. A transient write fault THROWS, leaving the schedule visible but - // unclaimed; the owner's next edit (or the reconciler's unarmed sweep) recomputes - // nextRunAt. A null result is different in kind: updateScheduleById filters out - // deleting rows, so null means the row stopped being ours between the barrier - // re-check and here — the deletion cascade marked or erased it. Falling back to the - // pre-delete snapshot would answer 201 for a schedule that is already hidden and - // pending erasure. + // ARM last. A null result means the row stopped being ours between the barrier + // re-check and here — the deletion cascade marked or erased it (updateScheduleById + // filters out deleting rows). Falling back to the pre-delete snapshot would answer + // 201 for a schedule that is already hidden and pending erasure. + // + // A THROWN (or ambiguously acknowledged) arming write must ROLL THE ROW BACK, not + // leave it: the client retries the failed create with a fresh UUID, the retry + // commits a second row, and the reconciler's unarmed sweep later arms the first as + // well — one intended schedule becomes several recurring, billable ones. With the + // row compensated away, a retry recreates exactly one. Only when BOTH compensation + // writes also fail does the unarmed row remain (it cannot fire until swept), and + // the 500 is the same either way. let armed = created; if (nextRunAt) { - const updated = await deps.methods.updateScheduleById(id, user.id, { nextRunAt }); + let updated: ISchedule | null; + try { + updated = await deps.methods.updateScheduleById(id, user.id, { nextRunAt }); + } catch (armError) { + logger.error(`[schedules] arming failed for ${id}; rolling back the create`, armError); + await compensateLateCreate(deps, id, user.id); + res.status(500).json({ error: 'Failed to create schedule. Please retry.' }); + return; + } if (updated == null) { res.status(410).json({ error: 'Schedule no longer exists' }); return; @@ -412,12 +434,26 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH // Quiesce-then-erase: disable + mark deleting (stops new claims, hides it), // abort in-flight loopback jobs, and erase once drained — so a live run's // evidence is never destroyed out from under it. - const deleted = await deps.deleteSchedule(id, requestUser(req).id); - if (!deleted) { + const result = await deps.deleteSchedule(id, requestUser(req).id); + if (result === 'not_found') { res.status(404).json({ error: 'Schedule not found' }); return; } - res.json({ id }); + // HONEST failure: at least one active run could not be confirmed stopped, so its + // generation may still be producing and billing. The schedule is already hidden + // and fenced (no new claims), and the delete is idempotent — a retry re-runs the + // drain. Reporting success here would claim the run was stopped. + if (result === 'unconfirmed') { + res.set('Retry-After', '30'); + res.status(503).json({ + error: 'Could not confirm the active run was stopped. Please retry shortly.', + }); + return; + } + // 202 for `draining`: the aborts were delivered but a generation has not yet + // recorded its terminal outcome. The schedule is hidden and erasure follows the + // settlement (erase-on-settle), in any topology. + res.status(result === 'draining' ? 202 : 200).json({ id }); } async function runScheduleNow(req: ServerRequest, res: Response): Promise { diff --git a/packages/api/src/schedules/service.spec.ts b/packages/api/src/schedules/service.spec.ts index 9079727d40..40fd2cad6f 100644 --- a/packages/api/src/schedules/service.spec.ts +++ b/packages/api/src/schedules/service.spec.ts @@ -157,16 +157,202 @@ describe('deleteScheduleForOwner', () => { const manager = jest.requireMock('../stream/GenerationJobManager').GenerationJobManager; manager.abortJob = abortJob; - await expect(service.deleteScheduleForOwner('s1', 'user-1')).resolves.toBe(true); + // A paused (not running) job is positive evidence: settled synchronously, erased. + await expect(service.deleteScheduleForOwner('s1', 'user-1')).resolves.toBe('deleted'); expect(abortJob).toHaveBeenCalled(); }); + /** Shared double set for the drain-discipline tests below. */ + function makeDeleteHarness(run: Partial & { scheduleId: string }) { + const service = makeService(jest.fn, [string]>().mockResolvedValue([])); + const methods = service.engineDeps.methods as unknown as { + markScheduleDeleting: jest.Mock; + getActiveRunsForSchedule: jest.Mock; + eraseScheduleIfDrained: jest.Mock; + recordRunOutcome: jest.Mock; + }; + methods.markScheduleDeleting = jest.fn(async () => ({ id: run.scheduleId, user: 'user-1' })); + methods.getActiveRunsForSchedule = jest.fn(async () => [run]); + methods.eraseScheduleIfDrained = jest.fn(async () => true); + methods.recordRunOutcome = jest.fn(async () => undefined); + return { service, methods }; + } + + it('does not report success when the abort of a live run is not delivered', async () => { + const { service, methods } = makeDeleteHarness({ + scheduleId: 's1', + scheduledFor: new Date('2026-01-01T00:00:00.000Z'), + conversationId: 'c1', + status: 'started', + }); + // A RUNNING identity-matched job: not settleable, must be aborted... + mockJobStore = { + getJob: jest.fn(async () => ({ + status: 'running', + createdAt: 1, + scheduleId: 's1', + scheduledFor: '2026-01-01T00:00:00.000Z', + })), + } as unknown as typeof mockJobStore; + // ...and the abort delivery FAILS (job store write rejected). + const manager = jest.requireMock('../stream/GenerationJobManager').GenerationJobManager; + manager.abortJob = jest.fn(async () => { + throw new Error('store unreachable'); + }); + + // The generation may still be producing and billing; claiming success would say + // it was stopped. The schedule stays hidden and fenced; the delete is idempotent. + await expect(service.deleteScheduleForOwner('s1', 'user-1')).resolves.toBe('unconfirmed'); + expect(methods.eraseScheduleIfDrained).not.toHaveBeenCalled(); + }); + + it('treats an unreadable job store as unknown, not absent', async () => { + const { service, methods } = makeDeleteHarness({ + scheduleId: 's1', + scheduledFor: new Date('2026-01-01T00:00:00.000Z'), + conversationId: 'c1', + status: 'started', + }); + mockJobStore = { + getJob: jest.fn(async () => { + throw new Error('redis gone'); + }), + } as unknown as typeof mockJobStore; + const manager = jest.requireMock('../stream/GenerationJobManager').GenerationJobManager; + manager.abortJob = jest.fn(async () => { + throw new Error('redis gone'); + }); + + // With the store unreadable NOTHING is proven: the row must not be settled as + // an orphan (its generation may be live) and the delete must not read as done. + await expect(service.deleteScheduleForOwner('s1', 'user-1')).resolves.toBe('unconfirmed'); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + }); + + it('settles a provably job-less run synchronously and erases without a reconciler', async () => { + const { service, methods } = makeDeleteHarness({ + scheduleId: 's1', + scheduledFor: new Date('2026-01-01T00:00:00.000Z'), + conversationId: 'c1', + status: 'started', + }); + // Confirmed absence: the lookup SUCCEEDS and returns null (a crashed fire, or a + // stale row from a previous topology). The clustered entrypoint has no reconciler, + // so deferring this row to one retained the deleted schedule indefinitely there. + mockJobStore = { getJob: jest.fn(async () => null) } as unknown as typeof mockJobStore; + + await expect(service.deleteScheduleForOwner('s1', 'user-1')).resolves.toBe('deleted'); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ scheduleId: 's1', status: 'interrupted' }), + ); + expect(methods.eraseScheduleIfDrained).toHaveBeenCalledWith('s1'); + }); + + it('reports draining when a live run was aborted but has not yet settled', async () => { + const { service, methods } = makeDeleteHarness({ + scheduleId: 's1', + scheduledFor: new Date('2026-01-01T00:00:00.000Z'), + conversationId: 'c1', + status: 'started', + }); + mockJobStore = { + getJob: jest.fn(async () => ({ + status: 'running', + createdAt: 1, + scheduleId: 's1', + scheduledFor: '2026-01-01T00:00:00.000Z', + })), + } as unknown as typeof mockJobStore; + const manager = jest.requireMock('../stream/GenerationJobManager').GenerationJobManager; + manager.abortJob = jest.fn(async () => ({ success: true })); + // The run row is still active, so the erase declines; settlement (and the + // erase-on-settle it triggers) belongs to the aborted generation's outcome write. + methods.eraseScheduleIfDrained = jest.fn(async () => false); + + await expect(service.deleteScheduleForOwner('s1', 'user-1')).resolves.toBe('draining'); + }); + afterEach(() => { mockJobStore = null; jest.restoreAllMocks(); }); }); +describe('erase-on-settle', () => { + /** + * Whichever process records a run's terminal outcome also attempts the deferred + * erase of a deleting schedule. This is what makes a delete's `draining` state + * converge in EVERY topology — the clustered entrypoint runs no reconciler, so + * without it the hidden schedule (and its prompt, which has no TTL) survived its + * last run indefinitely there. + */ + it('attempts the deferred erase after recording a terminal outcome', async () => { + const service = makeService(jest.fn, [string]>().mockResolvedValue([])); + const methods = service.engineDeps.methods as unknown as { + getScheduleById: jest.Mock; + recordRunOutcome: jest.Mock; + eraseScheduleIfDrained: jest.Mock; + }; + methods.getScheduleById = jest.fn(async () => null); + methods.recordRunOutcome = jest.fn(async () => undefined); + methods.eraseScheduleIfDrained = jest.fn(async () => true); + + await expect( + service.recordScheduleOutcome({ + scheduleId: 's1', + scheduledFor: '2026-01-01T00:00:00.000Z', + status: 'success', + }), + ).resolves.toBe(true); + expect(methods.eraseScheduleIfDrained).toHaveBeenCalledWith('s1'); + }); + + it('does not erase on a pause, which is not a settlement', async () => { + const service = makeService(jest.fn, [string]>().mockResolvedValue([])); + const methods = service.engineDeps.methods as unknown as { + getScheduleById: jest.Mock; + recordRunOutcome: jest.Mock; + eraseScheduleIfDrained: jest.Mock; + }; + methods.getScheduleById = jest.fn(async () => null); + methods.recordRunOutcome = jest.fn(async () => undefined); + methods.eraseScheduleIfDrained = jest.fn(async () => true); + + await service.recordScheduleOutcome({ + scheduleId: 's1', + scheduledFor: '2026-01-01T00:00:00.000Z', + status: 'requires_action', + }); + expect(methods.eraseScheduleIfDrained).not.toHaveBeenCalled(); + }); +}); + +describe('attachment hold renewal', () => { + it('renews the bounded upload hold at fire preflight, best-effort', async () => { + const service = makeService(jest.fn, [string]>().mockResolvedValue([])); + const methods = service.engineDeps.methods as unknown as { + extendFilesTTL: jest.Mock; + getFiles: jest.Mock; + }; + // The hold bridges upload -> first consumption; a renewal failure must not fail + // the fire (the file resolves now; at worst the hold lapses later). + methods.extendFilesTTL = jest.fn(async () => { + throw new Error('mongo down'); + }); + methods.getFiles = jest.fn(async () => [ + { file_id: 'f1', filepath: '/f1', filename: 'a.png', type: 'image/png', source: 'local' }, + ]); + + const files = await service.engineDeps.resolveFiles(['f1'], { id: 'user-1', tenantId: 't1' }); + expect(methods.extendFilesTTL).toHaveBeenCalledWith( + ['f1'], + expect.objectContaining({ renewMs: expect.any(Number), maxLifetimeMs: expect.any(Number) }), + { user: 'user-1', tenantId: 't1' }, + ); + expect(files).toHaveLength(1); + }); +}); + describe('quiesceUserSchedules drain wait', () => { afterEach(() => { mockJobStore = null; diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 8bb35844bc..48cc2f46ba 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -6,6 +6,7 @@ import type { AddressInfo } from 'net'; import type { Types } from 'mongoose'; import type { ScheduleEngineDeps, + ScheduleDeleteResult, ScheduleLimits, ScheduleUserContext, FireableSchedule, @@ -17,11 +18,11 @@ import type { BalanceUpdateFields } from '../types/balance'; import type { GetAppConfigOptions } from '../app/service'; import { generateShortLivedToken, SCHEDULE_FIRE_SCOPE, SCHEDULE_MANUAL_CLAIM } from '../crypto/jwt'; import { GenerationJobManager } from '../stream/GenerationJobManager'; +import { DEFAULT_SCHEDULE_LIMITS, SCHEDULE_FILE_HOLD } from './types'; import { buildBalanceUpdateFields } from '../middleware/balance'; import { deleteAgentCheckpoint } from '../agents/checkpointer'; import { fireSchedule, SCHEDULE_FIRE_TOKEN_TTL } from './fire'; import { getAppConfigOptionsFromUser } from '../app/service'; -import { DEFAULT_SCHEDULE_LIMITS } from './types'; import { getBalanceConfig } from '../app/config'; import { selfOriginFromAddress } from './origin'; import { startScheduleEngine } from './engine'; @@ -106,6 +107,12 @@ export interface SchedulesServiceDeps { width?: number; source: string; }> | null>; + /** Owner-scoped bounded TTL hold (`db.extendFilesTTL`-shaped). */ + extendFilesTTL: ( + fileIds: string[], + hold: { renewMs: number; maxLifetimeMs: number }, + owner: { user: string; tenantId?: string | null }, + ) => Promise; }; getAppConfig: (options?: GetAppConfigOptions) => Promise; findUserById: ( @@ -150,7 +157,7 @@ export interface SchedulesService { options?: { automatic?: boolean }, ) => Promise; /** Soft-deletes an owner's schedule: stop claims, abort active runs, drain, erase. */ - deleteScheduleForOwner: (scheduleId: string, userId: string) => Promise; + deleteScheduleForOwner: (scheduleId: string, userId: string) => Promise; /** * Quiesces all of a user's schedules ahead of account deletion (stop + abort + drain). * Returns whether the drain was CONFIRMED: false means at least one run could not be @@ -358,6 +365,13 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer // so the two never diverge. agentAccess: (agentId, user) => deps.resolveAgentFireAccess(agentId, user), resolveFiles: async (fileIds, user) => { + // Renew the bounded upload hold at every fire preflight, BEST-EFFORT: the hold + // only has to bridge upload -> first consumption (a real send clears the TTL + // permanently), and a failed renewal must not fail the fire — at worst the hold + // lapses later and resolveFiles drops the reaped file (droppedFileIds records it). + await methods + .extendFilesTTL(fileIds, SCHEDULE_FILE_HOLD, { user: user.id, tenantId: user.tenantId }) + .catch((err) => logger.warn('[schedules] attachment hold renewal failed:', err)); const files = await methods.getFiles( { file_id: { $in: fileIds }, user: user.id }, null, @@ -601,6 +615,17 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer error, autoDisableAfterFailures: limits.autoDisableAfterFailures, }); + // ERASE-ON-SETTLE: whichever process records a run's terminal outcome also + // attempts the deferred erase of a deleting schedule. This is what makes a + // delete's `draining` state converge in EVERY topology — the clustered + // entrypoint runs no reconciler, so without this the hidden schedule (and its + // prompt, which has no TTL) survived its last run indefinitely there. A cheap + // guarded no-op for live schedules (the erase filters on `deleting: true`). + if (status !== 'requires_action') { + await methods.eraseScheduleIfDrained(scheduleId).catch((err) => { + logger.warn(`[schedules] erase-on-settle failed for ${scheduleId}:`, err); + }); + } return true; } catch (err) { logger.error( @@ -696,15 +721,21 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer /** * Soft-deletes a schedule for its owner: disables + marks it `deleting` (so the * engine can no longer claim it and it disappears from the owner's list), rotates - * the claim token to fence any in-flight worker, aborts the loopback jobs of its - * active runs (evidence preserved for the reconciler), and erases immediately - * when already drained. Any still-active runs are erased by the reconciler once - * they settle. Returns false when the schedule doesn't exist / already deleting. + * the claim token to fence any in-flight worker, then DRAINS with the same evidence + * discipline as account-deletion quiesce — a run whose job is provably absent or + * settled is recorded and erased here, synchronously; a live generation is aborted + * and settles through its own outcome write (which erases on settle, so no + * reconciler is required in any topology). The result is honest: `unconfirmed` + * means at least one run could not be shown stopped, and the caller must not + * claim it was. */ - async function deleteScheduleForOwner(scheduleId: string, userId: string): Promise { + async function deleteScheduleForOwner( + scheduleId: string, + userId: string, + ): Promise { const schedule = await methods.markScheduleDeleting(scheduleId, userId); if (schedule == null) { - return false; + return 'not_found'; } const active = await methods.getActiveRunsForSchedule(scheduleId); // Resolve the checkpointer config once (only when a paused run needs pruning) in @@ -712,19 +743,73 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer const hasPausedRun = active.some( (run) => run.status === 'requires_action' && run.conversationId != null, ); - // BEST-EFFORT: markScheduleDeleting above is one-shot (a retry 404s on the already- - // deleting row), so anything that throws before the aborts strands the schedule with - // its paused job still resumable. The prune it feeds is itself best-effort. + // BEST-EFFORT: the prune this feeds is itself best-effort, and a lookup failure + // must not cost the aborts below. const checkpointer = hasPausedRun ? await resolveOwnerCheckpointer(schedule.user).catch((err) => { logger.warn(`[schedules] checkpointer lookup failed for delete ${scheduleId}:`, err); return undefined; }) : undefined; + let unconfirmed = 0; for (const run of active) { - // Preserve the aborted job for the reconciler: the run row survives (erase - // waits for it to drain), so reconcile finalizes it and clears the job. - await abortActiveRun(run, true); + // UNKNOWN is not ABSENT — the same distinction the quiesce path draws. A lookup + // that succeeded and returned null is positive evidence no generation holds this + // conversation; a lookup that THREW is evidence of nothing. + const live = run.conversationId + ? await engineDeps.getJobStatus(run.conversationId).then( + (job) => ({ known: true, job }), + () => ({ known: false, job: null }), + ) + : { known: true, job: null }; + const isThisGeneration = + live.job != null && + jobMatchesIdentity(live.job, { + scheduleId: run.scheduleId, + scheduledFor: run.scheduledFor, + }); + const settleable = live.known && !(isThisGeneration && live.job?.status === 'running'); + if (settleable) { + // Positive evidence nothing is generating: settle the row HERE so the erase + // below can proceed without any reconciler — the clustered entrypoint has + // none, and deferring these rows to it retained the deleted schedule (and its + // prompt) indefinitely in that topology. Settle BEFORE aborting: a retained + // job is the only evidence of a finished run whose outcome write failed. + const retainedOutcome = isThisGeneration + ? TERMINAL_JOB_OUTCOMES[live.job!.status] + : undefined; + const settledStatus = retainedOutcome ?? 'interrupted'; + const settled = await methods + .recordRunOutcome({ + scheduleId: run.scheduleId, + scheduledFor: run.scheduledFor, + status: settledStatus, + conversationId: run.conversationId, + ...(settledStatus === 'interrupted' ? { error: 'Schedule deleted' } : {}), + autoDisableAfterFailures: DEFAULT_SCHEDULE_LIMITS.autoDisableAfterFailures, + }) + .then( + () => true, + (err) => { + logger.warn(`[schedules] failed to settle run on delete ${scheduleId}:`, err); + return false; + }, + ); + if (settled) { + await abortActiveRun(run, false); + } else { + unconfirmed += 1; + } + } else { + // A live generation (or an unknown one): abort it, preserving the job so its + // outcome survives, and require the DELIVERY to be confirmed. An abort that + // was not delivered leaves a generation that keeps producing and billing — + // reporting this delete as a success would claim otherwise. + const aborted = await abortActiveRun(run, true); + if (!aborted) { + unconfirmed += 1; + } + } // HITL: prune the durable checkpoint of a run aborted while paused so a new turn // in this conversation can't rehydrate the stale interrupt before the Mongo TTL // reclaims it (thread_id === conversationId). Idempotent / no-op otherwise. @@ -732,8 +817,14 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer await deleteAgentCheckpoint(run.conversationId, checkpointer).catch(() => undefined); } } - await methods.eraseScheduleIfDrained(scheduleId).catch(() => undefined); - return true; + if (unconfirmed > 0) { + return 'unconfirmed'; + } + const erased = await methods.eraseScheduleIfDrained(scheduleId).catch((err) => { + logger.warn(`[schedules] erase failed for ${scheduleId}:`, err); + return false; + }); + return erased ? 'deleted' : 'draining'; } /** diff --git a/packages/api/src/schedules/types.ts b/packages/api/src/schedules/types.ts index 0708b26d60..4757dac658 100644 --- a/packages/api/src/schedules/types.ts +++ b/packages/api/src/schedules/types.ts @@ -25,6 +25,33 @@ export interface ScheduleUserContext { role?: string; } +/** + * Outcome of an owner-initiated schedule delete. + * - `deleted`: drained and erased. + * - `draining`: every active run's abort was DELIVERED; erasure follows once the + * generation records its terminal outcome (erase-on-settle), in any topology. + * - `unconfirmed`: at least one run could not be confirmed stopped (job store + * unreachable, or the abort was not delivered). The schedule stays hidden and + * fenced, but its generation may still be producing — callers must not report + * the run as stopped. + */ +export type ScheduleDeleteResult = 'not_found' | 'deleted' | 'draining' | 'unconfirmed'; + +/** + * Renewable upload hold for schedule attachments (extendFilesTTL-shaped), replacing + * the earlier permanent TTL removal, which leaked the upload forever when the schedule + * was deleted before its first run, its file_ids were replaced, or creation failed. + * Touched at create/edit and at every fire preflight; the FIRST fire that actually + * sends the file clears its TTL permanently through the ordinary consumption path, so + * the hold only has to bridge upload -> first consumption. `renewMs` covers the longest + * cadence gap (weekly) twice over; `maxLifetimeMs` bounds a schedule that never manages + * to consume (auto-disable stops its renewals long before this ceiling). + */ +export const SCHEDULE_FILE_HOLD: { renewMs: number; maxLifetimeMs: number } = { + renewMs: 14 * 24 * 60 * 60 * 1000, + maxLifetimeMs: 90 * 24 * 60 * 60 * 1000, +}; + export interface ScheduleFileRef { file_id: string; filepath?: string; diff --git a/packages/data-schemas/src/methods/file.ts b/packages/data-schemas/src/methods/file.ts index e8bc78f59c..ac428bdce0 100644 --- a/packages/data-schemas/src/methods/file.ts +++ b/packages/data-schemas/src/methods/file.ts @@ -60,9 +60,6 @@ export function createFileMethods(mongoose: typeof import('mongoose')): { data: Partial & { file_id: string }, extraFilter?: FilterQuery, ) => Promise; - /** Clears the upload TTL on files (idempotent, no usage side effect). Returns the - * number of owner-scoped files actually retained. */ - retainFiles: (fileIds: string[], ownerScope?: FileOwnerScope) => Promise; updateFileUsage: (data: { file_id: string; inc?: number; @@ -411,30 +408,6 @@ export function createFileMethods(mongoose: typeof import('mongoose')): { }).lean(); } - /** - * Clears the upload TTL on files so they survive until they are actually used, WITHOUT - * touching the usage counter. Fully idempotent: re-running it (a retry, or a schedule - * PATCH that resends unchanged file_ids) unsets fields that are already unset. - * - * `updateFileUsage` is deliberately not used for this: its `$inc: {usage}` is a - * "this file was consumed by a message" semantic, so reusing it for retention inflates - * the counter on every retry and every re-save, with no decrement anywhere. - * - * Owner-scoped fail-closed, and returns how many files were actually retained so the - * caller can detect a file deleted between the ownership check and here. - */ - async function retainFiles(fileIds: string[], ownerScope?: FileOwnerScope): Promise { - const unique = [...new Set(fileIds)]; - if (unique.length === 0) { - return 0; - } - const File = mongoose.models.File as Model; - const result = await File.updateMany(withOwnerScope({ file_id: { $in: unique } }, ownerScope), { - $unset: { expiresAt: '', temp_file_id: '' }, - }); - return result.matchedCount ?? 0; - } - /** * Increments the usage of a file identified by file_id. * @param data - The data to update, must contain file_id and the increment value for usage @@ -692,7 +665,6 @@ export function createFileMethods(mongoose: typeof import('mongoose')): { claimCodeFile, createFile, updateFile, - retainFiles, updateFileUsage, deleteFile, deleteFiles,