diff --git a/api/server/controllers/UserController.js b/api/server/controllers/UserController.js index f74dadeb88..040b2ae21b 100644 --- a/api/server/controllers/UserController.js +++ b/api/server/controllers/UserController.js @@ -485,8 +485,14 @@ const deleteUserController = async (req, res) => { try { await restoreUserSchedulesFromDeletion(user.id, scheduleSuspensionToken); } catch (restoreError) { + // Every retry is exhausted at this point. The fence is still released below on + // purpose: retaining it would refuse this live account's schedule writes AND make + // `beginAgentTriggerUserDeletion` report `in_progress` forever, blocking the retry + // that is the convergence path — a later attempt re-suspends by ADOPTING this + // snapshot, so its cancel restores these exact rows. Log the token so the state is + // recoverable directly if that never happens. logger.error( - '[deleteUserController] Failed to restore suspended schedules after a cancelled deletion', + `[deleteUserController] Failed to restore suspended schedules after a cancelled deletion; they remain disabled for user ${user.id} under suspension token ${scheduleSuspensionToken}`, restoreError, ); } diff --git a/config/delete-user.js b/config/delete-user.js index c6221d722a..533a239f2f 100644 --- a/config/delete-user.js +++ b/config/delete-user.js @@ -240,9 +240,17 @@ async function gracefulExit(code = 0) { // schedule writes/claims are refused, so this restore cannot race an owner PATCH nor be // superseded by a second deletion attempt re-suspending these rows under a new token. if (scheduleSuspensionToken != null && !userDeleted) { + // Retried inside the method; the fence is still released below on purpose, since + // retaining it would block the retry that is the convergence path. Print the token + // so a restore that never converges stays recoverable by hand. await runAsSystem(() => methods.restoreUserSchedulesFromDeletion(uid, scheduleSuspensionToken), - ).catch((error) => console.error('Failed to restore suspended schedules:', error)); + ).catch((error) => + console.error( + `Failed to restore suspended schedules; they remain disabled for user ${uid} under suspension token ${scheduleSuspensionToken}:`, + error, + ), + ); } if (deletionFence != null && !userDeleted) { await runAsSystem(() => methods.cancelAgentTriggerUserPurge(uid, deletionFence)).catch( diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 976e5ee0a4..0723883ac2 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -1,9 +1,9 @@ import { logger, runAsSystem } from '@librechat/data-schemas'; import type { IScheduleRun } from '@librechat/data-schemas'; import type { ScheduleEngineDeps, JobState } from './types'; +import { hasAbortInFlight, hasResumeHandoffInFlight, retainedOutcome } from './types'; import { isShutdownInProgress, registerShutdownTask } from '~/app/shutdown'; import { fireSchedule, BALANCE_SKIP_DISABLE_THRESHOLD } from './fire'; -import { hasAbortInFlight, hasResumeHandoffInFlight } from './types'; import { computeNextRunAt } from './cadence'; const TICK_MS = 30_000; @@ -32,33 +32,6 @@ function jobIdentityMatches(jobState: JobState | null, run: IScheduleRun): boole return jobFor === run.scheduledFor.getTime(); } -/** - * The outcome the generation owner intended for a RETAINED `complete` job, or `success` - * when it left none (a fire from before the stamp, or a store that never carried it). - * An unrecognized value is treated as `success` rather than forwarded: it crossed a - * serialization boundary, and `recordRunOutcome` would reject a status outside its - * union — losing a rare refinement beats failing the recovery write outright. - */ -function retainedOutcome( - jobState: JobState | null, - fallback: 'success' | 'error', -): { - status: 'success' | 'error' | 'interrupted' | 'skipped_balance'; - error?: string; -} { - const stamped = jobState?.scheduleOutcome; - if (stamped === 'skipped_balance') { - return { status: 'skipped_balance' }; - } - if (stamped === 'interrupted') { - return { status: 'interrupted', error: jobState?.scheduleOutcomeError }; - } - if (stamped === 'error' || fallback === 'error') { - return { status: 'error', error: jobState?.scheduleOutcomeError ?? 'Run ended in error' }; - } - return { status: fallback }; -} - export type ScheduleEngine = { stop: () => void; /** Exposed for tests and the run-now handler: one full claim/fire pass. */ diff --git a/packages/api/src/schedules/erasure.spec.ts b/packages/api/src/schedules/erasure.spec.ts index cc50fce2a4..d6f0d99fb2 100644 --- a/packages/api/src/schedules/erasure.spec.ts +++ b/packages/api/src/schedules/erasure.spec.ts @@ -29,6 +29,7 @@ async function sweepOnce(options: { methods: methods as unknown as ScheduleMethods, getJobStatus: jest.fn(async () => options.job ?? null), getTriggerDelivery: jest.fn(async () => null), + clearReconciledJob: jest.fn(async () => undefined), canInferOwnerDeathFromMissingJob: options.canInferOwnerDeathFromMissingJob, }); await jest.advanceTimersByTimeAsync(5 * 60_000); @@ -119,16 +120,18 @@ describe('topology-safe dead-delivery convergence', () => { ]), }; const getTriggerDelivery = jest.fn(async () => options.delivery ?? null); + const clearReconciledJob = jest.fn(async () => undefined); const sweep = startScheduleErasureSweep({ methods: methods as unknown as ScheduleMethods, getJobStatus: jest.fn(async () => options.job ?? null), getTriggerDelivery: getTriggerDelivery as never, + clearReconciledJob, // Defaults to the UNSAFE topology to prove this path never depends on it. canInferOwnerDeathFromMissingJob: options.canInferOwnerDeathFromMissingJob ?? false, }); await jest.advanceTimersByTimeAsync(5 * 60_000); sweep.stop(); - return { methods, getTriggerDelivery }; + return { methods, getTriggerDelivery, clearReconciledJob }; } it('settles a dead delivery as error even in an unsafe topology (positive evidence)', async () => { @@ -173,6 +176,105 @@ describe('topology-safe dead-delivery convergence', () => { expect(methods.recordRunOutcome).not.toHaveBeenCalled(); }); + /** + * The gap this closes: a generation that FINISHED but whose owner exhausted every Mongo + * outcome retry leaves a preserved terminal job as the only evidence. The armed engine's + * reconciler replays that; the clustered entrypoint arms no engine, so before this the run + * kept its `started` row — and its global capacity slot — until the job store expired it. + */ + it('settles a live schedule from its retained terminal job, then releases the evidence', async () => { + const { methods, clearReconciledJob } = await convergeOnce({ + job: { + status: 'complete', + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + } as unknown as JobState, + }); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ + scheduleId: 'schedule-1', + status: 'success', + conversationId: 'conversation-1', + }), + ); + expect(clearReconciledJob).toHaveBeenCalledWith('conversation-1', { + scheduleId: 'schedule-1', + scheduledFor: new Date('2026-08-17T12:00:00.000Z'), + }); + }); + + /** Presence of an identity-matched job is positive evidence in EVERY topology, so this + * must not depend on the owner-death inference the unsafe fallback refuses. */ + it('honors the owner-stamped outcome over the generic terminal status', async () => { + const { methods } = await convergeOnce({ + job: { + status: 'complete', + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + scheduleOutcome: 'skipped_balance', + } as unknown as JobState, + }); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ status: 'skipped_balance' }), + ); + }); + + it('carries the stamped failure reason from a retained error job', async () => { + const { methods } = await convergeOnce({ + job: { + status: 'error', + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + scheduleOutcomeError: 'provider exploded', + } as unknown as JobState, + }); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ status: 'error', error: 'provider exploded' }), + ); + }); + + /** A pre-start abort reserves a conversationId but never creates the conversation, so + * projecting it would point the schedule card at a chat that does not exist. */ + it('clears the reserved conversationId when the generation never emitted its created event', async () => { + const { methods } = await convergeOnce({ + job: { + status: 'aborted', + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + createdEventEmitted: false, + } as unknown as JobState, + }); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.objectContaining({ status: 'interrupted', clearConversationId: true }), + ); + expect(methods.recordRunOutcome).toHaveBeenCalledWith( + expect.not.objectContaining({ conversationId: 'conversation-1' }), + ); + }); + + /** A terminal job belonging to a REPLACEMENT turn (same conversationId, no scheduled + * identity) must never finalize this run, nor have its evidence deleted for it. */ + it("ignores a terminal job that lost this occurrence's identity", async () => { + const { methods, clearReconciledJob } = await convergeOnce({ + job: { status: 'complete' } as unknown as JobState, + }); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + expect(clearReconciledJob).not.toHaveBeenCalled(); + }); + + it('never settles a terminal job while its abort is still in flight', async () => { + const { methods, clearReconciledJob } = await convergeOnce({ + run: { abortRequestedAt: new Date() } as Partial, + job: { + status: 'aborted', + scheduleId: 'schedule-1', + scheduledFor: '2026-08-17T12:00:00.000Z', + } as unknown as JobState, + }); + expect(methods.recordRunOutcome).not.toHaveBeenCalled(); + expect(clearReconciledJob).not.toHaveBeenCalled(); + }); + it('ignores a legacy reservation carrying no deliveryKey', async () => { const { methods, getTriggerDelivery } = await convergeOnce({ run: { deliveryKey: undefined } as Partial, @@ -213,6 +315,7 @@ describe('dead-delivery certainty fence', () => { status: 'dead', lastError: { code: 'x', message: 'timed out', certainty: options.certainty }, })) as never, + clearReconciledJob: jest.fn(async () => undefined), canInferOwnerDeathFromMissingJob: options.canInferOwnerDeathFromMissingJob, }); await jest.advanceTimersByTimeAsync(5 * 60_000); diff --git a/packages/api/src/schedules/erasure.ts b/packages/api/src/schedules/erasure.ts index 902dbfde5b..1f0c357cfe 100644 --- a/packages/api/src/schedules/erasure.ts +++ b/packages/api/src/schedules/erasure.ts @@ -1,7 +1,7 @@ import { logger, runAsSystem } from '@librechat/data-schemas'; import type { ScheduleMethods, IScheduleRun } from '@librechat/data-schemas'; import type { JobState, ScheduleEngineDeps } from './types'; -import { hasResumeHandoffInFlight, hasAbortInFlight } from './types'; +import { hasResumeHandoffInFlight, hasAbortInFlight, retainedOutcome } from './types'; import { registerShutdownTask } from '~/app/shutdown'; const SWEEP_MS = 5 * 60_000; @@ -10,9 +10,10 @@ const SWEEP_BATCH = 100; /** Runs with no readable job older than this are presumed owner-dead (matches the * engine reconciler's orphan cutoff). */ const ABANDONED_RUN_AGE_MS = 30 * 60_000; -/** Grace before a jobless reservation's delivery is consulted, so an accepted delivery - * still creating its generation is never settled mid-handoff. */ -const DEAD_DELIVERY_MIN_AGE_MS = 2 * 60_000; +/** Grace before a live schedule's reservation is converged, so an accepted delivery still + * creating its generation — or an owner still writing its terminal outcome — is never + * settled mid-handoff. */ +const STRANDED_RUN_MIN_AGE_MS = 2 * 60_000; /** Terminal job status → the run outcome it proves (mirror of the quiesce map). */ const TERMINAL_JOB_OUTCOMES: Record = { @@ -41,6 +42,9 @@ export interface ScheduleErasureDeps { * POSITIVE shared evidence that no generation owns the reservation, so settling on it * is topology-safe (unlike absence). */ getTriggerDelivery: ScheduleEngineDeps['getTriggerDelivery']; + /** Deletes a run's retained terminal job once its outcome is durable. Identity-guarded, + * so a replacement generation reusing the conversationId is never destroyed. */ + clearReconciledJob: ScheduleEngineDeps['clearReconciledJob']; /** Whether absence in this process's job store proves absence deployment-wide. * False for the process-local fallback whose scheduler refused unsafe topology. */ canInferOwnerDeathFromMissingJob: boolean; @@ -126,44 +130,104 @@ async function settleAbandonedRuns(deps: ScheduleErasureDeps, scheduleId: string } /** - * Settles reservations whose durable trigger delivery is DEAD, for live (non-deleting) - * schedules too. + * Settles a run from its OWN identity-matched terminal job, then releases the retained + * evidence — the clustered mirror of the engine reconciler's retained-job branch. + */ +async function settleFromRetainedJob( + deps: ScheduleErasureDeps, + run: IScheduleRun, + job: JobState, +): Promise { + const terminal = TERMINAL_JOB_OUTCOMES[job.status]; + if (terminal == null) { + // `running`/`requires_action`: the owner is alive and owns the settlement. + return; + } + // An `aborted` job flips the moment abortJob wins its status CAS — BEFORE its owner + // unwinds and persists — so it is the abort fence, not this status, that decides. The + // in-flight check in the caller already deferred those; reaching here means the owner + // is past its presumed-alive window and this is the backstop. + const intended = + terminal === 'interrupted' + ? { status: 'interrupted' as const, error: undefined } + : retainedOutcome(job, terminal === 'error' ? 'error' : 'success'); + await deps.methods.recordRunOutcome({ + scheduleId: run.scheduleId, + scheduledFor: run.scheduledFor, + status: intended.status, + // A pre-start abort reserved a conversationId but never created the conversation; + // projecting it would point the card at a chat that does not exist. + ...(terminal === 'interrupted' && job.createdEventEmitted !== true + ? { clearConversationId: true } + : { conversationId: run.conversationId }), + error: intended.error, + autoDisableAfterFailures: Number.MAX_SAFE_INTEGER, + }); + // AFTER the outcome write, never before: the retained job is the only surviving + // evidence if that write fails, and a preserved job carries no `completedAt`, so the + // store's finished-job sweep can never reap it on its own. + await deps.clearReconciledJob(run.conversationId as string, { + scheduleId: run.scheduleId, + scheduledFor: run.scheduledFor, + }); +} + +/** + * Converges the stranded reservations of LIVE (non-deleting) schedules, on POSITIVE + * EVIDENCE ONLY. * * `fireSchedule` reserves the run and its global capacity slot before the delivery reaches - * the chat route, so a pre-generation rejection (interactive limiter, PII, moderation) dead- - * letters the delivery while the run stays `started`. The engine's reconciler maps that to an - * `error`, but the clustered entrypoint arms no engine — a delivery that was queued before a - * restart into clustered mode would otherwise hold its capacity slot indefinitely. + * the chat route, and the clustered entrypoint arms no engine — so two states would + * otherwise hold that slot indefinitely: * - * This is POSITIVE-EVIDENCE-ONLY and therefore safe in every topology: a `dead` delivery is - * durable shared state proving no generation owns the reservation. It never infers owner - * death from a missing job, never claims, fires, or advances, and defers anything fenced by - * an in-flight abort or resume hand-off. Auto-disable policy is deliberately NOT applied here - * (the armed engine owns that); the failure is recorded and the slot released. + * - A RETAINED TERMINAL JOB. The generation finished, but its owner's Mongo outcome write + * exhausted every retry (`recordScheduleOutcome` returns false) and left the preserved + * job as the only surviving evidence. The armed engine's reconciler replays exactly + * this; with no engine nothing did, so the run stayed `started` and its retained + * generation evidence lived until store expiry. + * - A DEAD DELIVERY. A pre-generation rejection (interactive limiter, PII, moderation) + * dead-letters the delivery while the run stays `started` with no job at all. + * + * Both are safe in EVERY topology because both read positive, durable evidence instead of + * inferring owner death from absence: an identity-matched job is authoritative wherever it + * is observed (a shared store shows the real generation; a process-local store can only be + * showing this process's own), and a `dead` delivery is shared state proving no generation + * owns the reservation. This never claims, fires, or advances, and defers anything fenced + * by an in-flight abort or resume hand-off. Auto-disable policy is deliberately NOT applied + * (the armed engine owns that): the run settles and frees its slot, the streak is untouched. */ -async function settleDeadDeliveries(deps: ScheduleErasureDeps): Promise { +async function settleStrandedRuns(deps: ScheduleErasureDeps): Promise { const runs = await deps.methods.getRunsForReconciliation( - new Date(Date.now() - DEAD_DELIVERY_MIN_AGE_MS), + new Date(Date.now() - STRANDED_RUN_MIN_AGE_MS), SWEEP_BATCH, ); const now = Date.now(); for (const run of runs) { try { - if (run.status !== 'started' || !run.deliveryKey) { + // `started` is the capacity-consuming state this pass exists to release. A paused + // row holds no slot, and its approval-expiry path owns its own durable retry. + if (run.status !== 'started') { continue; } if (hasAbortInFlight(run, now) || hasResumeHandoffInFlight(run, now)) { continue; } - // A live generation owns the reservation regardless of delivery state; only an - // absent/unknown job leaves the delivery as the authority. Unknown defers. const job = run.conversationId ? await deps.getJobStatus(run.conversationId).then( (state) => ({ known: true, state }), () => ({ known: false, state: null }), ) : { known: true, state: null }; - if (!job.known || jobMatchesRun(job.state, run)) { + // Unknown is not gone: never settle on a failed job-store read. + if (!job.known) { + continue; + } + if (jobMatchesRun(job.state, run)) { + await settleFromRetainedJob(deps, run, job.state as JobState); + continue; + } + // No job of THIS occurrence's identity, so the durable delivery is the authority. + if (!run.deliveryKey) { continue; } const delivery = await deps.getTriggerDelivery(run.deliveryKey); @@ -192,7 +256,7 @@ async function settleDeadDeliveries(deps: ScheduleErasureDeps): Promise { autoDisableAfterFailures: Number.MAX_SAFE_INTEGER, }); } catch (err) { - logger.warn(`[schedules] dead-delivery settle failed for ${run.scheduleId}:`, err); + logger.warn(`[schedules] stranded-run settle failed for ${run.scheduleId}:`, err); } } } @@ -234,10 +298,11 @@ export function startScheduleErasureSweep(deps: ScheduleErasureDeps): ScheduleEr await deps.methods .markEraseAttempted(deleting.map((schedule) => schedule.id)) .catch((err) => logger.warn('[schedules] failed to stamp erase attempts:', err)); - // Live schedules too: converge reservations whose durable delivery is dead, so a - // pre-generation rejection cannot hold a global capacity slot where no engine is armed. - await settleDeadDeliveries(deps).catch((err) => - logger.warn('[schedules] dead-delivery convergence pass failed:', err), + // Live schedules too: converge reservations left stranded by a retained terminal + // job or a dead delivery, so neither can hold a global capacity slot — and no + // generation evidence can outlive its run — where no engine is armed. + await settleStrandedRuns(deps).catch((err) => + logger.warn('[schedules] stranded-run convergence pass failed:', err), ); }); } catch (err) { diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index 163562751d..393a724228 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -646,6 +646,10 @@ export function createSchedulesService( // Positive-evidence delivery convergence: a dead delivery settles its reservation // and frees the global capacity slot even where no engine is armed. getTriggerDelivery: engineDeps.getTriggerDelivery, + // Releases a run's retained terminal job once its outcome is durable, so preserved + // evidence (kept without `completedAt`, and therefore invisible to the store's + // finished-job sweep) cannot outlive the run it belonged to. + clearReconciledJob: engineDeps.clearReconciledJob, // If topology itself prevented arming, this process's missing job says // nothing about peer liveness. If only index creation failed, the topology // proof still holds and the existing owner-death backstop remains valid. diff --git a/packages/api/src/schedules/types.ts b/packages/api/src/schedules/types.ts index 4a1c250751..52c5febb42 100644 --- a/packages/api/src/schedules/types.ts +++ b/packages/api/src/schedules/types.ts @@ -88,6 +88,33 @@ export function hasResumeHandoffInFlight( ); } +/** + * The outcome the generation owner intended for a RETAINED `complete` job, or `success` + * when it left none (a fire from before the stamp, or a store that never carried it). + * An unrecognized value is treated as `success` rather than forwarded: it crossed a + * serialization boundary, and `recordRunOutcome` would reject a status outside its + * union — losing a rare refinement beats failing the recovery write outright. + */ +export function retainedOutcome( + jobState: JobState | null, + fallback: 'success' | 'error', +): { + status: 'success' | 'error' | 'interrupted' | 'skipped_balance'; + error?: string; +} { + const stamped = jobState?.scheduleOutcome; + if (stamped === 'skipped_balance') { + return { status: 'skipped_balance' }; + } + if (stamped === 'interrupted') { + return { status: 'interrupted', error: jobState?.scheduleOutcomeError }; + } + if (stamped === 'error' || fallback === 'error') { + return { status: 'error', error: jobState?.scheduleOutcomeError ?? 'Run ended in error' }; + } + return { status: fallback }; +} + export function hasAbortInFlight(run: { abortRequestedAt?: Date }, now: number): boolean { if (run.abortRequestedAt == null) { return false; diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index 4875bb2355..b50d590348 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -1170,6 +1170,63 @@ describe('reversible account-deletion suspension', () => { expect(restored.deletionSuspension).toBeUndefined(); }); + /** + * The restore's callers cannot retry it later: a cancelled deletion restores while the + * user-deletion fence is still armed and then releases that fence, after which nothing + * re-drives this. A single transient write failure would therefore leave a LIVE account + * holding silently disabled, next-run-less schedules. + */ + it('retries a transient write failure instead of stranding the rows disabled', 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'); + + const bulkWrite = jest.spyOn(Schedule, 'bulkWrite'); + bulkWrite.mockImplementationOnce(() => { + throw new Error('transient primary stepdown'); + }); + + await methods.restoreUserSchedulesFromDeletion(schedule.user, 'attempt-1'); + bulkWrite.mockRestore(); + + const restored = await getSchedule(schedule.id); + expect(restored.enabled).toBe(true); + expect(restored.nextRunAt?.getTime()).toBe(armedAt.getTime()); + expect(restored.deletionSuspension).toBeUndefined(); + }); + + /** Retrying is only safe because each attempt re-reads the rows STILL carrying the + * token: a partially-applied unordered write converges on exactly the stragglers, and + * an already-restored row is simply no longer in the set. */ + it('converges on the stragglers of a partially applied restore', async () => { + const userId = new mongoose.Types.ObjectId(); + const first = await methods.createSchedule(scheduleData({ user: userId })); + const second = await methods.createSchedule(scheduleData({ user: userId })); + await methods.suspendUserSchedulesForDeletion(userId, 'attempt-1'); + + // Exactly what a partially-applied unordered bulkWrite leaves behind: one row already + // restored and out of the token set, one still suspended. + await methods.restoreUserSchedulesFromDeletion(userId, 'attempt-1'); + await Schedule.updateOne( + { id: second.id }, + { + $set: { + enabled: false, + deletionSuspension: { token: 'attempt-1', enabled: true }, + }, + $unset: { nextRunAt: 1 }, + }, + ); + + await methods.restoreUserSchedulesFromDeletion(userId, 'attempt-1'); + + for (const id of [first.id, second.id]) { + const restored = await getSchedule(id); + expect(restored.enabled).toBe(true); + expect(restored.deletionSuspension).toBeUndefined(); + } + }); + 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 })); diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 5a21abb279..1b9e0fcc2d 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -1713,15 +1713,44 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche await tenantSafeBulkWrite(Schedule(), ops as AnyBulkWriteOperation[], { ordered: false }); } + const RESTORE_ATTEMPTS = 3; + /** * 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. + * + * RETRIED, because its callers cannot retry it later. A cancelled deletion restores while + * the user-deletion fence is still armed and then releases that fence; once released, + * nothing re-drives this, and a single transient failure would strand the owner's live + * account with silently disabled, next-run-less schedules. Retrying is safe: every attempt + * re-reads only the rows STILL carrying this token, so a partially-applied unordered write + * converges on exactly the stragglers, and a fully-applied one finds nothing and returns. + * Throws the last failure so callers still surface it. */ async function restoreUserSchedulesFromDeletion( userId: string | Types.ObjectId, token: string, + ): Promise { + let lastError: unknown; + for (let attempt = 1; attempt <= RESTORE_ATTEMPTS; attempt++) { + try { + await restoreSuspendedBatch(userId, token); + return; + } catch (error) { + lastError = error; + if (attempt < RESTORE_ATTEMPTS) { + await new Promise((resolve) => setTimeout(resolve, 100 * attempt)); + } + } + } + throw lastError; + } + + async function restoreSuspendedBatch( + userId: string | Types.ObjectId, + token: string, ): Promise { const suspended = await Schedule() .find({ user: userId, 'deletionSuspension.token': token, deleting: { $ne: true } })