mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-22 18:13:27 +00:00
fix(schedules): Wave B.1 — reversible account-deletion schedule suspension
Thread r3804518383. Account-deletion quiesce marked every schedule `deleting`, disabled it, and cleared nextRunAt destructively. When a later cascade step (or the drain itself) failed, the controller cancelled the user-deletion fence — restoring the user — but the schedules stayed `deleting` and were erased by the sweep, silently losing all of a live user's scheduled prompts. Replace the destructive marking with a REVERSIBLE, token-fenced suspension: - suspendUserSchedulesForDeletion(userId, token) snapshots each schedule's prior enabled/nextRunAt under a per-attempt token, then fences firing (disable, clear nextRunAt, rotate claimToken). It never sets `deleting`, so a suspended row is not erasure-eligible. Snapshotting reads then bulkWrites (a classic update cannot copy field values under DocumentDB), fenced per row so an already- suspended/soft-deleted/edited row is left alone; idempotent per token. - restoreUserSchedulesFromDeletion(userId, token) reverses it, re-enabling and re-arming only rows still carrying the exact attempt token and not independently deleted — so an owner-deleted or newer-attempt-suspended schedule is never resurrected. - deleteUserController generates the attempt token, passes it to quiesce, and on any failure that cancels the user-deletion fence restores the suspended rows. A successful deletion hard-deletes them (and their snapshots) via the existing cascade and never restores. Adds a `deletionSuspension` embedded field (excluded from the wire schedule), data-method regression tests (suspend/restore/fence/idempotency/no-resurrect), and controller tests (restore on drain-false and post-quiesce cascade failure, no restore on success). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq
This commit is contained in:
Claude
parent
69d863fbee
commit
005305ac83
9 changed files with 314 additions and 37 deletions
|
|
@ -33,7 +33,11 @@ const {
|
|||
purgeAgentTriggerDeliveriesForUser,
|
||||
} = require('~/server/services/Agents/triggers');
|
||||
const { getAppConfig } = require('~/server/services/Config');
|
||||
const { quiesceUserSchedules } = require('~/server/services/Schedules');
|
||||
const { randomUUID } = require('node:crypto');
|
||||
const {
|
||||
quiesceUserSchedules,
|
||||
restoreUserSchedulesFromDeletion,
|
||||
} = require('~/server/services/Schedules');
|
||||
const { getLogStores } = require('~/cache');
|
||||
const db = require('~/models');
|
||||
|
||||
|
|
@ -361,6 +365,7 @@ const updateUserPluginsController = async (req, res) => {
|
|||
const deleteUserController = async (req, res) => {
|
||||
const { user } = req;
|
||||
let triggerDeletionFence;
|
||||
let scheduleSuspensionToken;
|
||||
let userDeleted = false;
|
||||
|
||||
try {
|
||||
|
|
@ -396,7 +401,12 @@ const deleteUserController = async (req, res) => {
|
|||
}
|
||||
await drainAgentTriggerDeliveriesForUser(user.id);
|
||||
await subagentThreadTaskStore.cancelAndDrainForOwner(user.id, user.tenantId);
|
||||
if (!(await quiesceUserSchedules(user.id))) {
|
||||
// Reversibly suspend the user's schedules under a per-attempt token BEFORE draining.
|
||||
// A later cascade step (or this drain) can still fail and cancel the deletion, and the
|
||||
// catch below restores exactly this attempt's rows — so a failed deletion never leaves
|
||||
// a live user with silently disabled, erasure-eligible schedules.
|
||||
scheduleSuspensionToken = randomUUID();
|
||||
if (!(await quiesceUserSchedules(user.id, scheduleSuspensionToken))) {
|
||||
throw new Error('Scheduled executions could not be confirmed stopped');
|
||||
}
|
||||
const activeAgentRuns = await GenerationJobManager.getCleanupBlockingJobIdsForUser(
|
||||
|
|
@ -475,6 +485,20 @@ const deleteUserController = async (req, res) => {
|
|||
logger.error('[deleteUserController] Failed to release trigger deletion fence', fenceError);
|
||||
}
|
||||
}
|
||||
// The account survives this failed attempt, so its schedules must too: restore the
|
||||
// exact rows this attempt suspended (re-enabled/re-armed from their snapshot). Fenced
|
||||
// to the token, so a schedule the owner deleted meanwhile is not resurrected. A
|
||||
// successful deletion never reaches here (userDeleted short-circuits it).
|
||||
if (scheduleSuspensionToken != null && !userDeleted) {
|
||||
try {
|
||||
await restoreUserSchedulesFromDeletion(user.id, scheduleSuspensionToken);
|
||||
} catch (restoreError) {
|
||||
logger.error(
|
||||
'[deleteUserController] Failed to restore suspended schedules after a cancelled deletion',
|
||||
restoreError,
|
||||
);
|
||||
}
|
||||
}
|
||||
logger.error('[deleteUserController]', err);
|
||||
return res.status(500).json({ message: 'Something went wrong.' });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ const mockCancelAgentTriggerUserPurge = jest.fn().mockResolvedValue(true);
|
|||
const mockPurgeAgentTriggerDeliveriesForUser = jest.fn().mockResolvedValue(undefined);
|
||||
const mockCancelAndDrainSubagentThreads = jest.fn().mockResolvedValue(undefined);
|
||||
const mockQuiesceUserSchedules = jest.fn().mockResolvedValue(true);
|
||||
const mockRestoreUserSchedules = jest.fn().mockResolvedValue(undefined);
|
||||
|
||||
jest.mock('@librechat/data-schemas', () => {
|
||||
const actual = jest.requireActual('@librechat/data-schemas');
|
||||
|
|
@ -104,6 +105,7 @@ jest.mock('~/server/services/Endpoints/agents/subagentThreadStore', () => ({
|
|||
|
||||
jest.mock('~/server/services/Schedules', () => ({
|
||||
quiesceUserSchedules: (...args) => mockQuiesceUserSchedules(...args),
|
||||
restoreUserSchedulesFromDeletion: (...args) => mockRestoreUserSchedules(...args),
|
||||
}));
|
||||
|
||||
jest.mock('~/server/services/Files/process', () => ({
|
||||
|
|
@ -369,7 +371,7 @@ describe('deleteUserController', () => {
|
|||
);
|
||||
expect(mockDrainAgentTriggerDeliveriesForUser).toHaveBeenCalledWith(userId.toString());
|
||||
expect(mockCancelAndDrainSubagentThreads).toHaveBeenCalledWith(userId.toString(), undefined);
|
||||
expect(mockQuiesceUserSchedules).toHaveBeenCalledWith(userId.toString());
|
||||
expect(mockQuiesceUserSchedules).toHaveBeenCalledWith(userId.toString(), expect.any(String));
|
||||
expect(beginAgentTriggerUserDeletion.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
mockPrepareAgentTriggerUserPurge.mock.invocationCallOrder[0],
|
||||
);
|
||||
|
|
@ -394,6 +396,8 @@ describe('deleteUserController', () => {
|
|||
expect(mockPurgeAgentTriggerDeliveriesForUser).toHaveBeenCalledWith(userId.toString());
|
||||
expect(cancelAgentTriggerUserDeletion).not.toHaveBeenCalled();
|
||||
expect(mockCancelAgentTriggerUserPurge).not.toHaveBeenCalled();
|
||||
// A successful deletion hard-deletes the schedules; it must never restore them.
|
||||
expect(mockRestoreUserSchedules).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('aborts generations admitted before the deletion fence before erasing messages', async () => {
|
||||
|
|
@ -440,6 +444,11 @@ describe('deleteUserController', () => {
|
|||
expect(mockCancelAgentTriggerUserPurge).toHaveBeenCalledWith(userIdString, deletionFence);
|
||||
expect(cancelAgentTriggerUserDeletion).toHaveBeenCalledWith(userIdString, deletionFence);
|
||||
expect(deleteUserById).not.toHaveBeenCalled();
|
||||
// Account survives -> its suspended schedules are restored under the quiesce token.
|
||||
expect(mockRestoreUserSchedules).toHaveBeenCalledWith(
|
||||
userIdString,
|
||||
mockQuiesceUserSchedules.mock.calls[0][1],
|
||||
);
|
||||
});
|
||||
|
||||
it('fails closed before data cleanup when detached subagents do not drain', async () => {
|
||||
|
|
@ -487,6 +496,11 @@ describe('deleteUserController', () => {
|
|||
expect(mockCancelAgentTriggerUserPurge).toHaveBeenCalledWith(userIdString, deletionFence);
|
||||
expect(cancelAgentTriggerUserDeletion).toHaveBeenCalledWith(userIdString, deletionFence);
|
||||
expect(deleteUserById).not.toHaveBeenCalled();
|
||||
// Account survives -> its suspended schedules are restored under the quiesce token.
|
||||
expect(mockRestoreUserSchedules).toHaveBeenCalledWith(
|
||||
userIdString,
|
||||
mockQuiesceUserSchedules.mock.calls[0][1],
|
||||
);
|
||||
});
|
||||
|
||||
it('should remove the user from all groups via $pullAll', async () => {
|
||||
|
|
|
|||
|
|
@ -90,6 +90,7 @@ module.exports = {
|
|||
isScheduleLive: invoke('isScheduleLive'),
|
||||
deleteScheduleForOwner: invoke('deleteScheduleForOwner'),
|
||||
quiesceUserSchedules: invoke('quiesceUserSchedules'),
|
||||
restoreUserSchedulesFromDeletion: invoke('restoreUserSchedulesFromDeletion'),
|
||||
initializeScheduleEngine: invoke('initializeScheduleEngine'),
|
||||
initializeScheduleErasureSweep: invoke('initializeScheduleErasureSweep'),
|
||||
recordExpiredScheduleApproval,
|
||||
|
|
|
|||
|
|
@ -46,7 +46,8 @@ function makeService(
|
|||
): ReturnType<typeof createSchedulesService> {
|
||||
recordRunOutcome = jest.fn(async () => undefined);
|
||||
const methods = {
|
||||
disableUserSchedulesForDeletion: jest.fn(async () => undefined),
|
||||
suspendUserSchedulesForDeletion: jest.fn(async () => undefined),
|
||||
restoreUserSchedulesFromDeletion: jest.fn(async () => undefined),
|
||||
getActiveRunsForUser,
|
||||
countActiveRuns: jest.fn(async () => 0),
|
||||
requestRunAbort: jest.fn(async () => true),
|
||||
|
|
@ -685,7 +686,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
.mockResolvedValue([]); // subsequent polls: drained
|
||||
const service = makeService(getActive);
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
// Each poll waits one interval; advance twice so the loop observes the drain.
|
||||
await jest.advanceTimersByTimeAsync(250);
|
||||
await jest.advanceTimersByTimeAsync(250);
|
||||
|
|
@ -705,7 +706,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
const getActive = jest.fn<Promise<ActiveRun[]>, [string]>().mockResolvedValue([run()]); // never drains
|
||||
const service = makeService(getActive);
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
// Advance past the full bounded timeout; the loop must give up, not hang, and must
|
||||
// report the drain as UNCONFIRMED so deletion defers rather than destroying.
|
||||
await jest.advanceTimersByTimeAsync(10_000);
|
||||
|
|
@ -738,7 +739,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
})),
|
||||
} as unknown as typeof mockJobStore;
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
await jest.advanceTimersByTimeAsync(10_000);
|
||||
// The pause hand-off's writes are still in flight; confirming the drain here
|
||||
// let the destructive cascade run before they landed. Deferral is bounded: the
|
||||
|
|
@ -770,7 +771,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
deleteJob,
|
||||
} as unknown as typeof mockJobStore;
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
await jest.advanceTimersByTimeAsync(500);
|
||||
await expect(pending).resolves.toBe(true);
|
||||
|
||||
|
|
@ -812,7 +813,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
deleteJob: jest.fn(async () => undefined),
|
||||
} as unknown as typeof mockJobStore;
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
await jest.advanceTimersByTimeAsync(10_000);
|
||||
// Unconfirmed, so the caller defers deletion (503 + Retry-After) instead of
|
||||
// destroying data the owner is still writing.
|
||||
|
|
@ -841,7 +842,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
deleteJob: jest.fn(async () => undefined),
|
||||
} as unknown as typeof mockJobStore;
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
await jest.advanceTimersByTimeAsync(500);
|
||||
await expect(pending).resolves.toBe(true);
|
||||
expect(recordRunOutcome).toHaveBeenCalledWith(
|
||||
|
|
@ -867,7 +868,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
} as unknown as typeof mockJobStore;
|
||||
recordRunOutcome.mockRejectedValue(new Error('mongo down'));
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
await jest.advanceTimersByTimeAsync(11_000);
|
||||
// The row never settles, so the drain cannot confirm and deletion defers.
|
||||
await expect(pending).resolves.toBe(false);
|
||||
|
|
@ -883,7 +884,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
|
||||
// Nothing to abort and nothing to drain, so the quiesce is trivially CONFIRMED and
|
||||
// the deletion cascade may proceed to its destructive steps.
|
||||
await expect(service.quiesceUserSchedules('user-1')).resolves.toBe(true);
|
||||
await expect(service.quiesceUserSchedules('user-1', 'attempt-1')).resolves.toBe(true);
|
||||
// Only the initial collection read; the drain loop is skipped for an empty set.
|
||||
expect(getActive).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
|
@ -906,7 +907,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
const getActive = jest.fn<Promise<ActiveRun[]>, [string]>().mockResolvedValue([pausedRun()]);
|
||||
const service = makeService(getActive);
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
await jest.advanceTimersByTimeAsync(10_000);
|
||||
// Terminalizing here would drop the row out of the active set, so the drain would
|
||||
// report nothing to wait for and the destructive cascade could delete messages the
|
||||
|
|
@ -929,7 +930,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
const getActive = jest.fn<Promise<ActiveRun[]>, [string]>().mockResolvedValue([pausedRun()]);
|
||||
const service = makeService(getActive);
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
await jest.advanceTimersByTimeAsync(10_000);
|
||||
await pending;
|
||||
// An approval that will never be consumed must not keep the account undeletable.
|
||||
|
|
@ -948,7 +949,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
const getActive = jest.fn<Promise<ActiveRun[]>, [string]>().mockResolvedValue([pausedRun()]);
|
||||
const service = makeService(getActive);
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
await jest.advanceTimersByTimeAsync(10_000);
|
||||
// A thrown lookup is evidence of NOTHING. Reading it as "genuinely paused" would
|
||||
// terminalize a row whose resumed generation may still be running, and the drain
|
||||
|
|
@ -967,7 +968,7 @@ describe('quiesceUserSchedules drain wait', () => {
|
|||
const getActive = jest.fn<Promise<ActiveRun[]>, [string]>().mockResolvedValue([pausedRun()]);
|
||||
const service = makeService(getActive);
|
||||
|
||||
const pending = service.quiesceUserSchedules('user-1');
|
||||
const pending = service.quiesceUserSchedules('user-1', 'attempt-1');
|
||||
await jest.advanceTimersByTimeAsync(10_000);
|
||||
await pending;
|
||||
expect(recordRunOutcome).toHaveBeenCalled();
|
||||
|
|
|
|||
|
|
@ -212,12 +212,20 @@ export interface SchedulesService {
|
|||
/** Soft-deletes an owner's schedule: stop claims, abort active runs, drain, erase. */
|
||||
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
|
||||
* confirmed settled, and the caller must NOT proceed to destructive deletion — the
|
||||
* durable barrier keeps refusing new work while a later pass finishes the cascade.
|
||||
* Quiesces all of a user's schedules ahead of account deletion (reversible suspension +
|
||||
* abort + drain). `token` identifies this deletion attempt and is what a later
|
||||
* {@link restoreUserSchedulesFromDeletion} restores against. Returns whether the drain was
|
||||
* CONFIRMED: false means at least one run could not be confirmed settled, and the caller
|
||||
* must NOT proceed to destructive deletion — the durable barrier keeps refusing new work
|
||||
* while a later pass finishes the cascade.
|
||||
*/
|
||||
quiesceUserSchedules: (userId: string) => Promise<boolean>;
|
||||
quiesceUserSchedules: (userId: string, token: string) => Promise<boolean>;
|
||||
/**
|
||||
* Reverses a quiesce whose account deletion was cancelled, re-enabling and re-arming only
|
||||
* the rows this exact attempt suspended. Safe to call even if quiesce never suspended a
|
||||
* row (no matching token → no-op).
|
||||
*/
|
||||
restoreUserSchedulesFromDeletion: (userId: string, token: string) => Promise<void>;
|
||||
/** Arms the scheduler for THIS process. v1 is single-process only; the clustered
|
||||
* entrypoint does not start it. Returns undefined when index creation failed, or when
|
||||
* the topology cannot be shown safe (process-local job store with no single-process
|
||||
|
|
@ -1277,13 +1285,15 @@ export function createSchedulesService(
|
|||
}
|
||||
|
||||
/**
|
||||
* Quiesces every schedule of a user ahead of account deletion: marks them all
|
||||
* non-claimable (so no new occurrence fires while the cascade runs) and aborts
|
||||
* the loopback jobs of any in-flight runs, so a scheduled generation cannot keep
|
||||
* persisting messages after the account's messages/conversations are deleted.
|
||||
* Quiesces every schedule of a user ahead of account deletion: REVERSIBLY suspends them
|
||||
* under the attempt `token` (non-claimable, so no new occurrence fires while the cascade
|
||||
* runs) and aborts the loopback jobs of any in-flight runs, so a scheduled generation
|
||||
* cannot keep persisting messages after the account's messages/conversations are deleted.
|
||||
* The suspension is reversible so a cancelled deletion can restore it (see
|
||||
* restoreUserSchedulesFromDeletion) rather than stranding a live user with erased rows.
|
||||
*/
|
||||
async function quiesceUserSchedules(userId: string): Promise<boolean> {
|
||||
await methods.disableUserSchedulesForDeletion(userId);
|
||||
async function quiesceUserSchedules(userId: string, token: string): Promise<boolean> {
|
||||
await methods.suspendUserSchedulesForDeletion(userId, token);
|
||||
const active = await methods.getActiveRunsForUser(userId);
|
||||
const unconfirmed: string[] = [];
|
||||
for (const run of active) {
|
||||
|
|
@ -1426,6 +1436,16 @@ export function createSchedulesService(
|
|||
return confirmed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reverses a quiesce whose account deletion was cancelled (a controller failure that
|
||||
* released the user-deletion fence). Re-enables and re-arms only the rows this exact
|
||||
* attempt suspended; a schedule the owner independently deleted, or one a newer attempt
|
||||
* re-suspended, is left as-is.
|
||||
*/
|
||||
async function restoreUserSchedulesFromDeletion(userId: string, token: string): Promise<void> {
|
||||
await methods.restoreUserSchedulesFromDeletion(userId, token);
|
||||
}
|
||||
|
||||
return {
|
||||
getLimits,
|
||||
engineDeps,
|
||||
|
|
@ -1438,6 +1458,7 @@ export function createSchedulesService(
|
|||
isScheduleLive,
|
||||
deleteScheduleForOwner,
|
||||
quiesceUserSchedules,
|
||||
restoreUserSchedulesFromDeletion,
|
||||
initializeScheduleEngine,
|
||||
initializeScheduleErasureSweep,
|
||||
};
|
||||
|
|
|
|||
|
|
@ -1058,6 +1058,102 @@ describe('deleteSchedulesByUser', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('reversible account-deletion suspension', () => {
|
||||
async function rawSchedule(id: string): Promise<ISchedule | null> {
|
||||
return Schedule.findOne({ id }).lean<ISchedule>();
|
||||
}
|
||||
|
||||
it('suspends under a token, fencing firing while snapshotting the prior state', async () => {
|
||||
const armedAt = new Date('2030-01-01T00:00:00Z');
|
||||
const schedule = await methods.createSchedule(scheduleData({ nextRunAt: armedAt }));
|
||||
const before = await getSchedule(schedule.id);
|
||||
|
||||
await methods.suspendUserSchedulesForDeletion(schedule.user, 'attempt-1');
|
||||
|
||||
const after = await getSchedule(schedule.id);
|
||||
expect(after.enabled).toBe(false);
|
||||
expect(after.nextRunAt).toBeUndefined();
|
||||
expect(after.claimToken).not.toBe(before.claimToken);
|
||||
// Reversible, NOT destructive: never marked for erasure.
|
||||
expect(after.deleting).not.toBe(true);
|
||||
expect(after.deletionSuspension?.token).toBe('attempt-1');
|
||||
expect(after.deletionSuspension?.enabled).toBe(true);
|
||||
expect(after.deletionSuspension?.nextRunAt?.getTime()).toBe(armedAt.getTime());
|
||||
});
|
||||
|
||||
it('restores enabled + next-run from the snapshot and clears the suspension', async () => {
|
||||
const armedAt = new Date('2030-01-01T00:00:00Z');
|
||||
const schedule = await methods.createSchedule(scheduleData({ nextRunAt: armedAt }));
|
||||
await methods.suspendUserSchedulesForDeletion(schedule.user, 'attempt-1');
|
||||
|
||||
await methods.restoreUserSchedulesFromDeletion(schedule.user, 'attempt-1');
|
||||
|
||||
const restored = await getSchedule(schedule.id);
|
||||
expect(restored.enabled).toBe(true);
|
||||
expect(restored.nextRunAt?.getTime()).toBe(armedAt.getTime());
|
||||
expect(restored.deletionSuspension).toBeUndefined();
|
||||
});
|
||||
|
||||
it('restores a previously DISABLED schedule as disabled with no next run', async () => {
|
||||
const schedule = await methods.createSchedule(scheduleData({ enabled: false }));
|
||||
await Schedule.updateOne({ id: schedule.id }, { $unset: { nextRunAt: 1 } });
|
||||
await methods.suspendUserSchedulesForDeletion(schedule.user, 'attempt-1');
|
||||
await methods.restoreUserSchedulesFromDeletion(schedule.user, 'attempt-1');
|
||||
|
||||
const restored = await getSchedule(schedule.id);
|
||||
expect(restored.enabled).toBe(false);
|
||||
expect(restored.nextRunAt).toBeUndefined();
|
||||
expect(restored.deletionSuspension).toBeUndefined();
|
||||
});
|
||||
|
||||
it('restore is fenced to the exact attempt token', async () => {
|
||||
const schedule = await methods.createSchedule(scheduleData());
|
||||
await methods.suspendUserSchedulesForDeletion(schedule.user, 'attempt-1');
|
||||
|
||||
// A different attempt's restore must not touch this suspension.
|
||||
await methods.restoreUserSchedulesFromDeletion(schedule.user, 'attempt-2');
|
||||
|
||||
const after = await getSchedule(schedule.id);
|
||||
expect(after.enabled).toBe(false);
|
||||
expect(after.deletionSuspension?.token).toBe('attempt-1');
|
||||
});
|
||||
|
||||
it('cannot resurrect a schedule the owner independently deleted mid-suspension', async () => {
|
||||
const schedule = await methods.createSchedule(scheduleData());
|
||||
await methods.suspendUserSchedulesForDeletion(schedule.user, 'attempt-1');
|
||||
// The owner deletes this one schedule while the account deletion is in flight.
|
||||
await methods.markScheduleDeleting(schedule.id, schedule.user);
|
||||
|
||||
await methods.restoreUserSchedulesFromDeletion(schedule.user, 'attempt-1');
|
||||
|
||||
const after = await rawSchedule(schedule.id);
|
||||
expect(after?.deleting).toBe(true);
|
||||
expect(after?.enabled).toBe(false); // NOT re-enabled by restore
|
||||
});
|
||||
|
||||
it('is idempotent per token — a retry does not overwrite the snapshot', async () => {
|
||||
const armedAt = new Date('2030-01-01T00:00:00Z');
|
||||
const schedule = await methods.createSchedule(scheduleData({ nextRunAt: armedAt }));
|
||||
await methods.suspendUserSchedulesForDeletion(schedule.user, 'attempt-1');
|
||||
// A retry of the SAME attempt must not re-snapshot the now-disabled state.
|
||||
await methods.suspendUserSchedulesForDeletion(schedule.user, 'attempt-1');
|
||||
|
||||
const after = await getSchedule(schedule.id);
|
||||
expect(after.deletionSuspension?.enabled).toBe(true);
|
||||
expect(after.deletionSuspension?.nextRunAt?.getTime()).toBe(armedAt.getTime());
|
||||
});
|
||||
|
||||
it('a successful deletion removes suspended rows and their snapshots', async () => {
|
||||
const userId = new mongoose.Types.ObjectId();
|
||||
const schedule = await methods.createSchedule(scheduleData({ user: userId }));
|
||||
await methods.suspendUserSchedulesForDeletion(userId, 'attempt-1');
|
||||
|
||||
await methods.deleteSchedulesByUser(userId);
|
||||
|
||||
expect(await rawSchedule(schedule.id)).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe('acquireManualRunLease / releaseLease', () => {
|
||||
it('serializes concurrent run-now attempts and can be released without advancing', async () => {
|
||||
const schedule = await methods.createSchedule(
|
||||
|
|
|
|||
|
|
@ -1,12 +1,13 @@
|
|||
import { randomUUID } from 'node:crypto';
|
||||
import type { ScheduleRunStatus, ScheduleDisabledReason } from 'librechat-data-provider';
|
||||
import type { Model, Types } from 'mongoose';
|
||||
import type { Model, Types, AnyBulkWriteOperation } from 'mongoose';
|
||||
import type {
|
||||
ISchedule,
|
||||
IScheduleDocument,
|
||||
IScheduleRun,
|
||||
IScheduleRunDocument,
|
||||
} from '~/types/schedule';
|
||||
import { tenantSafeBulkWrite } from '~/utils/tenantBulkWrite';
|
||||
import { createIndexesWithRetry } from '~/utils/retry';
|
||||
|
||||
const DUPLICATE_KEY = 11000;
|
||||
|
|
@ -240,7 +241,14 @@ export type ScheduleMethods = {
|
|||
markScheduleDeleting: (id: string, userId: string | Types.ObjectId) => Promise<ISchedule | null>;
|
||||
getActiveRunsForSchedule: (scheduleId: string) => Promise<IScheduleRun[]>;
|
||||
getActiveRunsForUser: (userId: string | Types.ObjectId) => Promise<IScheduleRun[]>;
|
||||
disableUserSchedulesForDeletion: (userId: string | Types.ObjectId) => Promise<void>;
|
||||
suspendUserSchedulesForDeletion: (
|
||||
userId: string | Types.ObjectId,
|
||||
token: string,
|
||||
) => Promise<void>;
|
||||
restoreUserSchedulesFromDeletion: (
|
||||
userId: string | Types.ObjectId,
|
||||
token: string,
|
||||
) => Promise<void>;
|
||||
getDeletingSchedules: (limit: number) => Promise<ISchedule[]>;
|
||||
markEraseAttempted: (ids: string[]) => Promise<void>;
|
||||
getScheduleByClientRequestId: (
|
||||
|
|
@ -1644,14 +1652,99 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
* the engine cannot fire a new occurrence while the cascade runs. Rotates each
|
||||
* claim token to fence any in-flight worker.
|
||||
*/
|
||||
async function disableUserSchedulesForDeletion(userId: string | Types.ObjectId): Promise<void> {
|
||||
await Schedule().updateMany(
|
||||
{ user: userId, deleting: { $ne: true } },
|
||||
{
|
||||
$set: { enabled: false, deleting: true, claimToken: randomUUID() },
|
||||
$unset: { nextRunAt: 1 },
|
||||
/**
|
||||
* REVERSIBLE quiesce ahead of account deletion. Snapshots each schedule's pre-suspension
|
||||
* enabled/next-run state under this attempt's `token`, then fences firing (disable, clear
|
||||
* `nextRunAt`, rotate `claimToken`). It deliberately does NOT set `deleting`: a suspended
|
||||
* row must not be erased, because a deletion the controller later cancels has to restore
|
||||
* it. On a successful deletion the hard-delete cascade removes the row (and this snapshot)
|
||||
* anyway. Snapshotting needs each row's current values, which a classic (DocumentDB-safe)
|
||||
* update cannot copy, so it reads then `bulkWrite`s — fenced per row so a row already
|
||||
* suspended by this token, soft-deleted, or edited out from under the read is left alone.
|
||||
*/
|
||||
async function suspendUserSchedulesForDeletion(
|
||||
userId: string | Types.ObjectId,
|
||||
token: string,
|
||||
): Promise<void> {
|
||||
const schedules = await Schedule()
|
||||
.find({
|
||||
user: userId,
|
||||
deleting: { $ne: true },
|
||||
'deletionSuspension.token': { $ne: token },
|
||||
})
|
||||
.select('id enabled nextRunAt')
|
||||
.lean<Array<Pick<ISchedule, 'id' | 'enabled' | 'nextRunAt'>>>();
|
||||
if (schedules.length === 0) {
|
||||
return;
|
||||
}
|
||||
const ops = schedules.map((schedule) => ({
|
||||
updateOne: {
|
||||
filter: {
|
||||
id: schedule.id,
|
||||
user: userId,
|
||||
deleting: { $ne: true },
|
||||
'deletionSuspension.token': { $ne: token },
|
||||
},
|
||||
update: {
|
||||
$set: {
|
||||
enabled: false,
|
||||
claimToken: randomUUID(),
|
||||
deletionSuspension: {
|
||||
token,
|
||||
enabled: schedule.enabled === true,
|
||||
...(schedule.nextRunAt != null ? { nextRunAt: schedule.nextRunAt } : {}),
|
||||
},
|
||||
},
|
||||
$unset: { nextRunAt: 1 as const },
|
||||
},
|
||||
},
|
||||
);
|
||||
}));
|
||||
await tenantSafeBulkWrite(Schedule(), ops as AnyBulkWriteOperation[], { ordered: false });
|
||||
}
|
||||
|
||||
/**
|
||||
* Reverses {@link suspendUserSchedulesForDeletion} when a deletion attempt is cancelled.
|
||||
* Restores enabled/next-run from the snapshot and clears the suspension — but ONLY for
|
||||
* rows still carrying this exact attempt's token and not independently soft-deleted, so a
|
||||
* schedule the owner deleted (or a newer attempt re-suspended) is never resurrected.
|
||||
*/
|
||||
async function restoreUserSchedulesFromDeletion(
|
||||
userId: string | Types.ObjectId,
|
||||
token: string,
|
||||
): Promise<void> {
|
||||
const suspended = await Schedule()
|
||||
.find({ user: userId, 'deletionSuspension.token': token, deleting: { $ne: true } })
|
||||
.select('id deletionSuspension')
|
||||
.lean<Array<Pick<ISchedule, 'id' | 'deletionSuspension'>>>();
|
||||
if (suspended.length === 0) {
|
||||
return;
|
||||
}
|
||||
const ops = suspended.map((schedule) => {
|
||||
const snapshot = schedule.deletionSuspension;
|
||||
const restoreNextRunAt = snapshot?.nextRunAt;
|
||||
return {
|
||||
updateOne: {
|
||||
filter: {
|
||||
id: schedule.id,
|
||||
user: userId,
|
||||
'deletionSuspension.token': token,
|
||||
deleting: { $ne: true },
|
||||
},
|
||||
update: {
|
||||
$set: {
|
||||
enabled: snapshot?.enabled === true,
|
||||
claimToken: randomUUID(),
|
||||
...(restoreNextRunAt != null ? { nextRunAt: restoreNextRunAt } : {}),
|
||||
},
|
||||
$unset: {
|
||||
deletionSuspension: 1 as const,
|
||||
...(restoreNextRunAt == null ? { nextRunAt: 1 as const } : {}),
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
});
|
||||
await tenantSafeBulkWrite(Schedule(), ops as AnyBulkWriteOperation[], { ordered: false });
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -1873,7 +1966,8 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
|
|||
markScheduleDeleting,
|
||||
getActiveRunsForSchedule,
|
||||
getActiveRunsForUser,
|
||||
disableUserSchedulesForDeletion,
|
||||
suspendUserSchedulesForDeletion,
|
||||
restoreUserSchedulesFromDeletion,
|
||||
getDeletingSchedules,
|
||||
markEraseAttempted,
|
||||
getScheduleByClientRequestId,
|
||||
|
|
|
|||
|
|
@ -118,6 +118,22 @@ const scheduleSchema: Schema<IScheduleDocument> = new Schema(
|
|||
type: Boolean,
|
||||
default: false,
|
||||
},
|
||||
/**
|
||||
* Reversible account-deletion suspension. Set at quiesce under a per-attempt token
|
||||
* that snapshots the pre-suspension enabled/next-run state. A deletion cancelled by a
|
||||
* controller failure restores exactly the rows carrying its token; a successful
|
||||
* deletion hard-deletes the row (and this snapshot with it). Distinct from `deleting`:
|
||||
* a suspended row is NOT eligible for erasure, so a failed attempt cannot lose it.
|
||||
*/
|
||||
deletionSuspension: {
|
||||
type: {
|
||||
token: { type: String, required: true },
|
||||
enabled: { type: Boolean, required: true },
|
||||
nextRunAt: { type: Date },
|
||||
},
|
||||
default: undefined,
|
||||
_id: false,
|
||||
},
|
||||
/** Erased tombstone: content is gone, only the create-idempotency identity
|
||||
* remains for a bounded retry window (TTL below). A delayed create retry that
|
||||
* matches this key must answer "deleted", not resurrect the recurring work. */
|
||||
|
|
|
|||
|
|
@ -28,6 +28,16 @@ export interface ISchedule {
|
|||
/** Owner-config generation; bumped only by an owner edit. */
|
||||
configRevision?: number;
|
||||
deleting?: boolean;
|
||||
/** Reversible account-deletion suspension. Set at quiesce under a per-attempt token
|
||||
* that snapshots the pre-suspension enabled/next-run state, so a deletion that is
|
||||
* later cancelled restores exactly this row (fenced to its token) instead of leaving
|
||||
* a live user with silently disabled schedules. Distinct from `deleting`, which marks
|
||||
* a row for erasure. */
|
||||
deletionSuspension?: {
|
||||
token: string;
|
||||
enabled: boolean;
|
||||
nextRunAt?: Date;
|
||||
};
|
||||
erased?: boolean;
|
||||
erasedAt?: Date;
|
||||
slot?: number;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue