mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-08-04 14:57:42 +00:00
fix: Codex round 12 — arming retry safety, honest deletes, topology-free erase, bounded file holds
- 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).
This commit is contained in:
Danny Avila
parent
7186be757c
commit
acfc264bf9
8 changed files with 476 additions and 83 deletions
|
|
@ -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)) {
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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<string, string> } = {
|
||||
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<SchedulesHandlersDeps>);
|
||||
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<boolean>;
|
||||
/** Filters to file ids owned by the user. */
|
||||
filterOwnedFileIds: (fileIds: string[], userId: string) => Promise<string[]>;
|
||||
/** 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<void>;
|
||||
/** Serialized manual fire (acquires the schedule lease); null if already leased. */
|
||||
fireNow: (schedule: FireableSchedule, limits: ScheduleLimits) => Promise<FireResult | null>;
|
||||
/**
|
||||
* 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<boolean>;
|
||||
deleteSchedule: (id: string, userId: string) => Promise<ScheduleDeleteResult>;
|
||||
/** Whether this user's account deletion has begun. Fail-closed (unknown == true). */
|
||||
isUserDeleting: (userId: string) => Promise<boolean>;
|
||||
}
|
||||
|
|
@ -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<boolean> {
|
||||
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<void> {
|
||||
|
|
|
|||
|
|
@ -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<ActiveRun> & { scheduleId: string }) {
|
||||
const service = makeService(jest.fn<Promise<ActiveRun[]>, [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<Promise<ActiveRun[]>, [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<Promise<ActiveRun[]>, [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<Promise<ActiveRun[]>, [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;
|
||||
|
|
|
|||
|
|
@ -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<number>;
|
||||
};
|
||||
getAppConfig: (options?: GetAppConfigOptions) => Promise<AppConfig | undefined>;
|
||||
findUserById: (
|
||||
|
|
@ -150,7 +157,7 @@ export interface SchedulesService {
|
|||
options?: { automatic?: boolean },
|
||||
) => Promise<boolean>;
|
||||
/** Soft-deletes an owner's schedule: stop claims, abort active runs, drain, erase. */
|
||||
deleteScheduleForOwner: (scheduleId: string, userId: string) => Promise<boolean>;
|
||||
deleteScheduleForOwner: (scheduleId: string, userId: string) => Promise<ScheduleDeleteResult>;
|
||||
/**
|
||||
* 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<boolean> {
|
||||
async function deleteScheduleForOwner(
|
||||
scheduleId: string,
|
||||
userId: string,
|
||||
): Promise<ScheduleDeleteResult> {
|
||||
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';
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -60,9 +60,6 @@ export function createFileMethods(mongoose: typeof import('mongoose')): {
|
|||
data: Partial<IMongoFile> & { file_id: string },
|
||||
extraFilter?: FilterQuery<IMongoFile>,
|
||||
) => Promise<IMongoFile | null>;
|
||||
/** 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<number>;
|
||||
updateFileUsage: (data: {
|
||||
file_id: string;
|
||||
inc?: number;
|
||||
|
|
@ -411,30 +408,6 @@ export function createFileMethods(mongoose: typeof import('mongoose')): {
|
|||
}).lean<IMongoFile>();
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<number> {
|
||||
const unique = [...new Set(fileIds)];
|
||||
if (unique.length === 0) {
|
||||
return 0;
|
||||
}
|
||||
const File = mongoose.models.File as Model<IMongoFile>;
|
||||
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,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue