From fab7b2c6607f5143f775dc7df5fce7708367fd36 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Aug 2026 21:36:56 +0000 Subject: [PATCH] fix(schedules): converge terminal runs in clustered workers, retry suspension restore Fourth Codex round on 74f7af8. Both findings are in code from this closeout. P2 - a clustered worker never settled a run whose generation finished but whose outcome write failed. `recordScheduleOutcome` retries three times and then returns false; the owner honors that by PRESERVING the terminal job as the only surviving evidence, and the armed engine's reconciler replays exactly that. The clustered entrypoint arms no engine, and its sweep only covered deleting schedules (settleAbandonedRuns) and dead deliveries - an identity-matched job was skipped outright. So an ordinary live schedule's run stayed `started`, held its GLOBAL capacity slot, and kept a preserved job that carries no `completedAt` and is therefore invisible to the store's finished-job sweep: both leaked until store expiry. The live-schedule pass now also converges from a retained terminal job, mirroring the reconciler branch it stands in for: honor the owner's stamped outcome over the generic status (so a balance refusal still walks its streak rather than resetting it), clear the reserved conversationId when the generation never emitted its created event, and delete the retained job only AFTER the outcome write is durable. This stays positive-evidence-only and safe in every topology. Presence, not absence, is the evidence: 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 - which is why it needs no canInferOwnerDeathFromMissingJob fence, unlike the absence-based paths. The in-flight abort/resume fences still defer, so an `aborted` job cannot settle a run whose owner may still be persisting. Both cases now share one pass over one window rather than two, and `retainedOutcome` moved to types.ts so the sweep reuses the engine's mapping instead of duplicating it (and stays independent of the engine). P2 - a failed restore stranded a live account's schedules. Cancelling an account deletion restores the suspended rows while the deletion fence is still armed and then releases that fence; nothing re-drives the restore afterwards, so one transient write failure left the user with silently disabled, next-run-less schedules. The restore is now retried at the single choke point both the HTTP controller and the CLI share. Retrying is safe because each attempt re-reads only the rows STILL carrying the token: a partially-applied unordered write converges on exactly the stragglers, and a fully-applied one finds nothing. The fence is still released when every attempt fails, deliberately: retaining it would refuse the live account's schedule writes AND make beginAgentTriggerUserDeletion report `in_progress` forever, blocking the retry that is the convergence path (a later attempt adopts this snapshot, so its cancel restores these exact rows). Both callers now log the user id and suspension token so the state stays recoverable by hand if that never happens. Tests: retained terminal job settled and its evidence released, stamped outcome preferred over the generic status, stamped failure reason carried, reserved conversationId cleared for a never-created conversation, identity-mismatched terminal job ignored, abort-in-flight deferred; restore retried past a transient write failure and converging on a partially applied restore. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq --- api/server/controllers/UserController.js | 8 +- config/delete-user.js | 10 +- packages/api/src/schedules/engine.ts | 29 +---- packages/api/src/schedules/erasure.spec.ts | 105 +++++++++++++++- packages/api/src/schedules/erasure.ts | 117 ++++++++++++++---- packages/api/src/schedules/service.ts | 4 + packages/api/src/schedules/types.ts | 27 ++++ .../src/methods/schedule.methods.spec.ts | 57 +++++++++ packages/data-schemas/src/methods/schedule.ts | 29 +++++ 9 files changed, 329 insertions(+), 57 deletions(-) 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 } })