diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index ee201f8d02..95c09736a3 100644 --- a/api/server/controllers/agents/request.js +++ b/api/server/controllers/agents/request.js @@ -866,11 +866,6 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit scheduledFor, status: 'requires_action', conversationId: streamId, - // Segment 0: this generation came from the fire, which creates the run - // with no resumeSeq. The approval card is emitted BEFORE this write, so a - // fast resume can promote the run (bumping the epoch) first — without this - // token the stale pause would demote the run that resume just started. - expectResumeSeq: 0, }); } logger.debug( diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index afcb48e05d..8d3a3c1d57 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -23,13 +23,7 @@ const { getMCPRequestContext, cleanupMCPRequestContextForReq, } = require('~/server/services/MCPRequestContext'); -const { - recordScheduleOutcome, - reserveScheduledResume, - markScheduledResumeClaimed, - commitScheduledResume, - releaseScheduledResume, -} = require('~/server/services/Schedules'); +const { recordScheduleOutcome } = require('~/server/services/Schedules'); const { saveMessage, getConvo, getMessages } = require('~/models'); /** @@ -542,60 +536,12 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) return res.status(429).json({ error: 'Too many concurrent requests' }); } - // Atomically reserve the schedule's active slot BEFORE claiming the approval, so a - // deferral (overlap/capacity/gone) leaves the approval claimable. This promotes - // the paused run requires_action -> started (single-active partial index enforces - // per-schedule overlap) and reserve-then-verifies global fireConcurrency. Crucially - // the caller does NOT release on a lost claim below: whichever request wins the - // approval drives whatever is `started`, and an unclaimed promotion self-heals when - // the reconciler surfaces the still-paused job (so a losing racer never flips the - // winner's active row). - let resumeLease = null; - if (job.metadata?.scheduleId) { - let reservation; - try { - reservation = await reserveScheduledResume( - job.metadata.scheduleId, - job.metadata.scheduledFor, - ); - } catch (err) { - // A store/config error here throws BEFORE the run's own try/finally, so - // release the pending-request slot taken above — otherwise it leaks until its - // TTL and spuriously 429s the user's later resume/chat attempts. - await decrementPendingRequest(userId); - logger.error('[ResumeAgentController] Scheduled resume reservation failed', err); - return res.status(500).json({ error: 'Failed to resume' }); - } - if (reservation.outcome !== 'ok') { - await decrementPendingRequest(userId); - logger.debug( - `[ResumeAgentController] Deferring scheduled resume (${reservation.outcome}): ${streamId}`, - ); - const deferrals = { - gone: { status: 410, error: 'This schedule no longer exists' }, - capacity: { status: 409, error: 'Scheduled run capacity reached; try again shortly' }, - overlap: { - status: 409, - error: 'Another run for this schedule is in progress; try again shortly', - }, - held: { - status: 409, - error: 'This scheduled run is already resuming; try again shortly', - }, - }; - const { status, error } = deferrals[reservation.outcome] ?? deferrals.overlap; - return res.status(status).json({ error }); - } - // Own the durable RESUMING lease: every later transition is fenced on this holder, - // and resumeSeq is the CAS token any pause written by THIS segment must carry. - resumeLease = { - scheduleId: job.metadata.scheduleId, - scheduledFor: job.metadata.scheduledFor, - holder: reservation.holder, - resumeSeq: reservation.resumeSeq, - }; - } - + // v1 scope: the scheduler does NOT orchestrate resumes. A scheduled run that pauses + // for approval released its capacity slot on pause and is handed off — from here it is + // an ordinary paused conversation the user resumes through the chat UI. The scheduler + // only OBSERVES the eventual outcome below (so the schedule card reflects it); it does + // not reserve, lease, or fence this resume. Scheduled HITL orchestration is a + // fast-follow behind the experimental flag. // Atomically claim the resume. The single winner drives the run; a racing second // submit (double-click, two tabs) gets false and must not re-drive — that would // re-execute tools and double-bill. Do NOT release the reservation on a lost claim: @@ -606,42 +552,13 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) claimed = await GenerationJobManager.approvals.resolve(streamId, pendingAction.actionId); } catch (err) { await decrementPendingRequest(userId); - // Pre-claim failure: roll the reservation back so the approval stays actionable - // and the capacity slot isn't held by a resume that never started. - if (resumeLease) { - await releaseScheduledResume( - resumeLease.scheduleId, - resumeLease.scheduledFor, - resumeLease.holder, - ).catch(() => undefined); - } logger.error('[ResumeAgentController] Failed to claim resume', err); return res.status(500).json({ error: 'Failed to resume' }); } if (!claimed) { await decrementPendingRequest(userId); - // Still pre-claim: roll the reservation back so the approval stays actionable and - // the capacity slot is freed rather than held by a resume that never ran. - if (resumeLease) { - await releaseScheduledResume( - resumeLease.scheduleId, - resumeLease.scheduledFor, - resumeLease.holder, - ).catch(() => undefined); - } return res.status(409).json({ error: 'This action was already resolved or has expired' }); } - // The approval is consumed: from here a crash must roll FORWARD (the reconciler must - // finish or terminalize this run), so the lease is no longer adoptable by a retry. - if (resumeLease) { - await markScheduledResumeClaimed( - resumeLease.scheduleId, - resumeLease.scheduledFor, - resumeLease.holder, - ).catch((err) => - logger.error('[ResumeAgentController] Failed to mark resume lease claimed', err), - ); - } // Seed the run-scoped MCP request-context store BEFORE the ACK: once `res.json` // finishes the response, a later `getMCPRequestContext(req, res)` (from tool loading) @@ -739,17 +656,6 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) // the paused job's createdAt — used by the re-pause CAS pre-check + checkpoint prune to // avoid acting on a job a newer request has since replaced. client.jobCreatedAt = job.createdAt; - // Reconstruction succeeded, so the RESUMING phase is over: drop the lease and let - // the row be an ordinary `started` run again. A crash before this point leaves the - // lease for the reconciler (post-claim => roll forward), never a silently stranded - // `running` job the reconciler would skip. - if (resumeLease) { - await commitScheduledResume( - resumeLease.scheduleId, - resumeLease.scheduledFor, - resumeLease.holder, - ).catch((err) => logger.error('[ResumeAgentController] Failed to commit resume lease', err)); - } client.responseMessageId = job.metadata.responseMessageId; client.parentMessageId = job.metadata.userMessage?.messageId ?? Constants.NO_PARENT; // Read the pre-pause content BEFORE swapping the store's content reference: the @@ -801,9 +707,6 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) scheduledFor: job.metadata.scheduledFor, status: 'requires_action', conversationId, - // Epoch CAS: if a newer resume already bumped resumeSeq, this pause belongs to - // a superseded segment and must not demote the run that resume promoted. - expectResumeSeq: resumeLease?.resumeSeq, }); } return; diff --git a/api/server/services/Schedules/index.js b/api/server/services/Schedules/index.js index 69f92a3444..d6d6700443 100644 --- a/api/server/services/Schedules/index.js +++ b/api/server/services/Schedules/index.js @@ -28,10 +28,6 @@ module.exports = { fireScheduleNow: service.fireScheduleNow, recordScheduleOutcome: service.recordScheduleOutcome, isScheduleLive: service.isScheduleLive, - reserveScheduledResume: service.reserveScheduledResume, - markScheduledResumeClaimed: service.markScheduledResumeClaimed, - commitScheduledResume: service.commitScheduledResume, - releaseScheduledResume: service.releaseScheduledResume, deleteScheduleForOwner: service.deleteScheduleForOwner, quiesceUserSchedules: service.quiesceUserSchedules, initializeScheduleEngine: service.initializeScheduleEngine, diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 0f6f92b8c1..c44c441c69 100644 --- a/packages/api/src/schedules/engine.ts +++ b/packages/api/src/schedules/engine.ts @@ -96,38 +96,10 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine { conversationId: run.conversationId, error, autoDisableAfterFailures: runLimits.autoDisableAfterFailures, - // Fence the reconciler's own pause on the epoch it OBSERVED: several awaits - // separate the read from this write, and a resume landing in between must - // not be demoted by a sweep that is now looking at a stale segment. - ...(status === 'requires_action' ? { expectResumeSeq: run.resumeSeq ?? 0 } : {}), // Terminal bookkeeping is fenced on the config the run started under, so a // reconciled outcome cannot auto-disable a schedule the owner has edited. ...(run.configRevision != null ? { expectConfigRevision: run.configRevision } : {}), }); - // RESUME-LEASE RECOVERY. A resume that died between consuming the approval and - // reconstructing its generation leaves the row `started` with a live-looking - // lease, and the `running` skip below would step over it forever. Once the - // lease deadline passes, decide by phase: - // post-claim (resumeClaimedAt set) -> the approval is spent, so it can never - // be re-offered; roll FORWARD by terminalizing as interrupted. - // pre-claim -> the approval was never consumed, so roll BACK to - // requires_action (releaseResumeLease frees the slot) and let the user retry. - if ( - run.resumeHolder != null && - run.resumeExpiresAt != null && - run.resumeExpiresAt.getTime() < Date.now() - ) { - if (run.resumeClaimedAt != null) { - await finalize('interrupted', 'Resume did not complete'); - } else { - await deps.methods.releaseResumeLease( - run.scheduleId, - run.scheduledFor, - run.resumeHolder, - ); - } - continue; - } if (jobStatus === 'running') { continue; } diff --git a/packages/api/src/schedules/service.ts b/packages/api/src/schedules/service.ts index e5e285e924..321aef1e9f 100644 --- a/packages/api/src/schedules/service.ts +++ b/packages/api/src/schedules/service.ts @@ -1,4 +1,3 @@ -import { randomUUID } from 'node:crypto'; import { logger, runAsSystem, tenantStorage } from '@librechat/data-schemas'; import { getRefillEligibilityDate, Permissions, PermissionTypes } from 'librechat-data-provider'; import type { ScheduleMethods, AppConfig, IBalance } from '@librechat/data-schemas'; @@ -30,25 +29,6 @@ import { isEnabled } from '../utils/common'; /** Recordable terminal/paused run outcome, as accepted by `recordRunOutcome`. */ type ScheduleRunOutcomeStatus = Parameters[0]['status']; -/** - * Outcome of the pre-claim resume reservation for a HITL resume. `gone` = the - * schedule was deleted/soft-deleted (non-resumable); `overlap` = another occurrence - * is active; `capacity` = the global cap is saturated; each defers with the approval - * left unconsumed. `ok` = reserved (or already active), let the claim proceed. - */ -export type ResumeCheck = 'ok' | 'gone' | 'overlap' | 'capacity' | 'held'; - -/** - * Outcome of the durable resume reservation. `ok` carries the lease identity the - * caller must fence every later transition on: `holder` for commit/release and - * `resumeSeq` as the CAS token for any pause written by this segment. - */ -export type ResumeReservation = - | { outcome: 'ok'; holder: string; resumeSeq: number } - /** Not a scheduled resume (ordinary interactive HITL): there is no lease to drive. */ - | { outcome: 'not-scheduled' } - | { outcome: Exclude }; - /** Whether a persisted job still carries a given scheduled occurrence's identity. */ function jobMatchesIdentity(job: SerializableJobData, identity: JobIdentity): boolean { if (job.scheduleId !== identity.scheduleId || job.scheduledFor == null) { @@ -63,8 +43,6 @@ export interface RecordScheduleOutcomeInput { status: ScheduleRunOutcomeStatus; conversationId?: string; error?: string; - /** Pause CAS token: the resumeSeq the writing segment observed. */ - expectResumeSeq?: number; } /** @@ -122,38 +100,6 @@ export interface SchedulesService { * missed it) — aborting before any messages are persisted. */ isScheduleLive: (scheduleId: string, expectedConfigRevision?: number) => Promise; - /** - * Acquires the durable RESUMING lease for a HITL resume, BEFORE the approval claim - * (so a deferral leaves the approval claimable). Per-schedule overlap is enforced by - * the single-active partial index and the GLOBAL fireConcurrency cap by the unique - * capacity-slot index, so two resumes of different schedules can never both admit. - * 'gone' = deleted/terminal; 'overlap' = another occurrence is active; 'capacity' = - * cap saturated; 'held' = another live resume attempt owns this occurrence. On 'ok' - * the caller MUST drive the lease: markResumeClaimed after consuming the approval, - * then commitResumeLease on success or releaseResumeLease while still pre-claim. - */ - reserveScheduledResume: ( - scheduleId: string, - scheduledFor: string | Date, - ) => Promise; - /** Marks the resume lease as having consumed the approval (crash must roll forward). */ - markScheduledResumeClaimed: ( - scheduleId: string, - scheduledFor: string | Date, - holder: string, - ) => Promise; - /** Commits a resume once its generation is reconstructed and running. */ - commitScheduledResume: ( - scheduleId: string, - scheduledFor: string | Date, - holder: string, - ) => Promise; - /** Rolls a still-unclaimed resume back to `requires_action`, freeing its slot. */ - releaseScheduledResume: ( - scheduleId: string, - scheduledFor: string | Date, - holder: string, - ) => Promise; /** Soft-deletes an owner's schedule: stop claims, abort active runs, drain, erase. */ deleteScheduleForOwner: (scheduleId: string, userId: string) => Promise; /** @@ -225,9 +171,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer } const MANUAL_RUN_LEASE_MS = 5 * 60 * 1000; - // Lease held by an in-flight HITL resume. Long enough to cover reconstruction, short - // enough that a crashed PRE-CLAIM resume is reclaimable without a human. - const RESUME_LEASE_MS = 2 * 60 * 1000; // Bounded wait for aborted scheduled runs to settle during account-deletion quiesce, // before the message/conversation cascade runs. Long enough to cover a generation that // already returned from the model finishing its persistence; capped so account deletion @@ -535,7 +478,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer status, conversationId, error, - expectResumeSeq, }: RecordScheduleOutcomeInput): Promise { if (!scheduleId || !scheduledFor) { return true; @@ -555,7 +497,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer conversationId, error, autoDisableAfterFailures: limits.autoDisableAfterFailures, - expectResumeSeq, // Fence terminal bookkeeping/auto-disable to the config this run started // under, so an owner edit or re-enable since then is never acted on. expectConfigRevision: run?.configRevision, @@ -598,107 +539,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer return true; } - /** - * Reservation for a HITL resume, run BEFORE the approval claim so a deferral - * leaves the approval claimable. Checks existence/overlap/capacity READ-ONLY - * first (a null schedule -> 'gone'; another `started` occurrence -> 'overlap'; the - * owner's fireConcurrency saturated -> 'capacity'), then promotes the run into the - * single active slot WITHOUT any rollback. No rollback is the key correctness - * property: whichever request wins the approval drives whatever is `started`, so a - * losing racer can never flip the winner's active row back to requires_action. - * Per-schedule overlap is hard-enforced by the partial unique index; the global - * cap is a best-effort soft cap here (concurrent resumes of DIFFERENT schedules - * can transiently overshoot by the number racing, self-healing when they settle) — - * the fire path remains the hard-enforced cap for new load, and enforcing it - * atomically here would require either a rollback that races the claim takeover or - * a drift-prone global counter. - */ - async function reserveScheduledResume( - scheduleId: string, - scheduledFor: string | Date, - ): Promise { - if (!scheduleId || !scheduledFor) { - return { outcome: 'not-scheduled' }; - } - // A resume restarts a billed generation, so the global stop applies here too. - // Reported as 'gone' so the approval is left unconsumed and the caller defers. - if (await engineDeps.isGloballyDisabled()) { - return { outcome: 'gone' }; - } - // getScheduleById hides deleted/soft-deleted schedules, so a null here means the - // owner already deleted the schedule — its paused run must not be resumable even - // if the delete's best-effort abort raced. Reject before touching the run. - const schedule = await methods.getScheduleById(scheduleId); - if (schedule == null) { - return { outcome: 'gone' }; - } - const when = new Date(scheduledFor); - // Overlap = a DIFFERENT occurrence is currently `started`. Exclude this run's own - // row: if its pause bookkeeping failed transiently the row can still be `started`, - // and treating that as an overlap would wrongly reject resuming THIS same occurrence. - if (await methods.hasOtherActiveRun(scheduleId, when)) { - return { outcome: 'overlap' }; - } - const owner = await engineDeps.getUserContext(schedule.user); - const limits = await getLimits(owner ?? undefined); - const holder = randomUUID(); - // ATOMIC global capacity: the slot (not a count) is the contended resource, so two - // resumes of DIFFERENT schedules can no longer both pass a cap-1 check — the unique - // partial index rejects the loser and the allocator advances it to the next free - // slot, or reports 'capacity' when genuinely saturated. - const reserved = await runAsSystem(() => - withCapacitySlot( - limits.fireConcurrency, - () => methods.getCapacityOccupancy(), - async (slot) => { - const lease = await methods.acquireResumeLease({ - scheduleId, - scheduledFor: when, - holder, - ttlMs: RESUME_LEASE_MS, - capacitySlot: slot, - }); - return lease.outcome === 'slot-taken' ? 'slot-taken' : { claimed: lease }; - }, - ), - ); - if (reserved === 'capacity') { - return { outcome: 'capacity' }; - } - const lease = reserved.claimed; - if (lease.outcome === 'acquired') { - return { outcome: 'ok', holder: lease.holder, resumeSeq: lease.resumeSeq }; - } - // 'held' = another live resume attempt owns this occurrence (previously collapsed - // into a lossy 'missing' -> 'ok', which admitted both racers). 'overlap'/'gone' - // defer with the approval left unconsumed. - return { outcome: lease.outcome === 'held' ? 'held' : lease.outcome }; - } - - async function markScheduledResumeClaimed( - scheduleId: string, - scheduledFor: string | Date, - holder: string, - ): Promise { - return methods.markResumeClaimed(scheduleId, new Date(scheduledFor), holder, RESUME_LEASE_MS); - } - - async function commitScheduledResume( - scheduleId: string, - scheduledFor: string | Date, - holder: string, - ): Promise { - return methods.commitResumeLease(scheduleId, new Date(scheduledFor), holder); - } - - async function releaseScheduledResume( - scheduleId: string, - scheduledFor: string | Date, - holder: string, - ): Promise { - return methods.releaseResumeLease(scheduleId, new Date(scheduledFor), holder); - } - /** Aborts an active run's loopback job (identity-guarded). Returns whether the * abort was delivered (false when the job wasn't reachable — e.g. a peer worker's * private store, or a transient error). */ @@ -831,10 +671,6 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer fireScheduleNow, recordScheduleOutcome, isScheduleLive, - reserveScheduledResume, - markScheduledResumeClaimed, - commitScheduledResume, - releaseScheduledResume, deleteScheduleForOwner, quiesceUserSchedules, initializeScheduleEngine, diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index 1ef36a6c3e..d94aff1aaf 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -1087,238 +1087,3 @@ describe('account-deletion barrier (delete vs create)', () => { expect(pending.some((u) => u._id.toString() === pendingUser._id.toString())).toBe(true); }); }); - -describe('lifecycle barriers: stale pause vs resume, and concurrent resumes at cap-1', () => { - const scheduledFor = new Date('2026-07-20T12:00:00Z'); - - it('a stale requires_action write cannot demote an already-resumed run', async () => { - const schedule = await methods.createSchedule(scheduleData()); - await methods.insertScheduleRun(runData(schedule, { scheduledFor })); - - // Segment 0 pauses. This mirrors the INITIAL pause exactly as request.js writes it: - // expectResumeSeq 0, against a row that has no resumeSeq field at all. - await methods.recordRunOutcome({ - scheduleId: schedule.id, - scheduledFor, - status: 'requires_action', - conversationId: 'convo-1', - autoDisableAfterFailures: 3, - expectResumeSeq: 0, - }); - expect((await getRun(schedule.id, scheduledFor)).status).toBe('requires_action'); - - // A resume promotes the run and BUMPS the epoch, opening segment 2. - const lease = await methods.acquireResumeLease({ - scheduleId: schedule.id, - scheduledFor, - holder: 'resume-a', - ttlMs: 60_000, - capacitySlot: 0, - }); - expect(lease.outcome).toBe('acquired'); - const resumedEpoch = lease.outcome === 'acquired' ? lease.resumeSeq : -1; - expect(resumedEpoch).toBeGreaterThan(0); - - // The now-STALE pause callback from segment 1 fires late, still carrying epoch 0. - // It must be dropped: demoting here would make a live, running generation look - // paused and free its slot for a concurrent occurrence. - await methods.recordRunOutcome({ - scheduleId: schedule.id, - scheduledFor, - status: 'requires_action', - conversationId: 'convo-1', - autoDisableAfterFailures: 3, - expectResumeSeq: 0, - }); - expect((await getRun(schedule.id, scheduledFor)).status).toBe('started'); - - // The CURRENT segment's pause still lands. - await methods.recordRunOutcome({ - scheduleId: schedule.id, - scheduledFor, - status: 'requires_action', - conversationId: 'convo-1', - autoDisableAfterFailures: 3, - expectResumeSeq: resumedEpoch, - }); - expect((await getRun(schedule.id, scheduledFor)).status).toBe('requires_action'); - }); - - it('only one of two resumes of DIFFERENT schedules admits at cap-1', async () => { - // Two distinct schedules, each with its own paused occurrence. Per-schedule overlap - // cannot arbitrate this: the contended resource is the GLOBAL capacity slot. - const a = await methods.createSchedule(scheduleData()); - const b = await methods.createSchedule(scheduleData()); - await methods.insertScheduleRun(runData(a, { scheduledFor, status: 'requires_action' })); - await methods.insertScheduleRun(runData(b, { scheduledFor, status: 'requires_action' })); - - // Both race for the SAME slot 0 (a cap of 1 leaves exactly one). - const [first, second] = await Promise.all([ - methods.acquireResumeLease({ - scheduleId: a.id, - scheduledFor, - holder: 'resume-a', - ttlMs: 60_000, - capacitySlot: 0, - }), - methods.acquireResumeLease({ - scheduleId: b.id, - scheduledFor, - holder: 'resume-b', - ttlMs: 60_000, - capacitySlot: 0, - }), - ]); - - const outcomes = [first.outcome, second.outcome].sort(); - expect(outcomes).toEqual(['acquired', 'slot-taken']); - // Exactly one run holds the slot, so the cap was never exceeded. - expect(await ScheduleRun.countDocuments({ status: 'started', capacitySlot: 0 })).toBe(1); - }); - - it('a released pre-claim resume frees its capacity slot for the loser', async () => { - const a = await methods.createSchedule(scheduleData()); - await methods.insertScheduleRun(runData(a, { scheduledFor, status: 'requires_action' })); - - const lease = await methods.acquireResumeLease({ - scheduleId: a.id, - scheduledFor, - holder: 'resume-a', - ttlMs: 60_000, - capacitySlot: 0, - }); - expect(lease.outcome).toBe('acquired'); - - // Rolling back a still-unclaimed resume must return the row to requires_action AND - // release the slot, so the approval stays actionable and capacity is not leaked. - expect(await methods.releaseResumeLease(a.id, scheduledFor, 'resume-a')).toBe(true); - const row = await getRun(a.id, scheduledFor); - expect(row.status).toBe('requires_action'); - expect(row.capacitySlot).toBeUndefined(); - expect(await ScheduleRun.countDocuments({ status: 'started', capacitySlot: 0 })).toBe(0); - }); -}); - -describe('resume lease: duplicate attempts and crash recovery', () => { - const scheduledFor = new Date('2026-07-20T12:00:00Z'); - - async function pausedRun() { - const schedule = await methods.createSchedule(scheduleData()); - await methods.insertScheduleRun(runData(schedule, { scheduledFor, status: 'requires_action' })); - return schedule; - } - - it('a delayed duplicate resume cannot demote the winner after it committed', async () => { - const schedule = await pausedRun(); - - // Winner acquires, consumes the approval, reconstructs, commits. Commit clears the - // holder, so the row is an ordinary `started` run again. - const winner = await methods.acquireResumeLease({ - scheduleId: schedule.id, - scheduledFor, - holder: 'winner', - ttlMs: 60_000, - capacitySlot: 0, - }); - expect(winner.outcome).toBe('acquired'); - await methods.markResumeClaimed(schedule.id, scheduledFor, 'winner', 60_000); - expect(await methods.commitResumeLease(schedule.id, scheduledFor, 'winner')).toBe(true); - - // A delayed duplicate submit for the SAME action now arrives. It can still adopt the - // holderless row, but it will LOSE approvals.resolve and then release. - const loser = await methods.acquireResumeLease({ - scheduleId: schedule.id, - scheduledFor, - holder: 'loser', - ttlMs: 60_000, - capacitySlot: 0, - }); - expect(loser.outcome).toBe('acquired'); - await methods.releaseResumeLease(schedule.id, scheduledFor, 'loser'); - - // The winner's run must still be live. Demoting here would knock a running - // generation back to requires_action and free its capacity slot underneath it. - const row = await getRun(schedule.id, scheduledFor); - expect(row.status).toBe('started'); - expect(row.capacitySlot).toBe(0); - expect(row.resumeHolder).toBeUndefined(); - }); - - it('still rolls back a resume that genuinely promoted a paused run', async () => { - const schedule = await pausedRun(); - const lease = await methods.acquireResumeLease({ - scheduleId: schedule.id, - scheduledFor, - holder: 'solo', - ttlMs: 60_000, - capacitySlot: 0, - }); - expect(lease.outcome).toBe('acquired'); - - // Pre-claim failure: this attempt promoted the row, so release MUST demote it and - // free the slot, leaving the approval actionable. - expect(await methods.releaseResumeLease(schedule.id, scheduledFor, 'solo')).toBe(true); - const row = await getRun(schedule.id, scheduledFor); - expect(row.status).toBe('requires_action'); - expect(row.capacitySlot).toBeUndefined(); - }); - - it('a second live attempt cannot steal an unexpired lease', async () => { - const schedule = await pausedRun(); - await methods.acquireResumeLease({ - scheduleId: schedule.id, - scheduledFor, - holder: 'first', - ttlMs: 60_000, - capacitySlot: 0, - }); - const second = await methods.acquireResumeLease({ - scheduleId: schedule.id, - scheduledFor, - holder: 'second', - ttlMs: 60_000, - capacitySlot: 1, - }); - expect(second.outcome).toBe('held'); - }); - - it('an EXPIRED pre-claim lease is adoptable; a post-claim one is not', async () => { - const preClaim = await pausedRun(); - await methods.acquireResumeLease({ - scheduleId: preClaim.id, - scheduledFor, - holder: 'crashed', - ttlMs: -1_000, // already expired - capacitySlot: 0, - }); - // Never consumed the approval, so a retry may take it over. - const adopted = await methods.acquireResumeLease({ - scheduleId: preClaim.id, - scheduledFor, - holder: 'retry', - ttlMs: 60_000, - capacitySlot: 0, - }); - expect(adopted.outcome).toBe('acquired'); - - const postClaim = await pausedRun(); - await methods.acquireResumeLease({ - scheduleId: postClaim.id, - scheduledFor, - holder: 'crashed-2', - ttlMs: -1_000, - capacitySlot: 1, - }); - // The approval WAS consumed, so this must roll forward (reconciler terminalizes it) - // rather than be re-run by another attempt. - await methods.markResumeClaimed(postClaim.id, scheduledFor, 'crashed-2', -1_000); - const stolen = await methods.acquireResumeLease({ - scheduleId: postClaim.id, - scheduledFor, - holder: 'retry-2', - ttlMs: 60_000, - capacitySlot: 2, - }); - expect(stolen.outcome).toBe('held'); - }); -}); diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 63a56b0778..d9c26a0b56 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -68,9 +68,6 @@ export interface RecordRunOutcomeParams { error?: string; durationMs?: number; autoDisableAfterFailures: number; - /** Pause CAS token: the resumeSeq the writing segment observed. A pause whose epoch - * no longer matches is dropped, so it cannot demote an already-resumed run. */ - expectResumeSeq?: number; /** The configRevision this run started under. Terminal bookkeeping / auto-disable is * skipped when the owner has since edited the schedule (revision moved on). */ expectConfigRevision?: number; @@ -87,29 +84,6 @@ export type StartedRunReservation = | { run: IScheduleRun } | { conflict: 'duplicate' | 'overlap' | 'slot-taken' }; -export interface AcquireResumeLeaseParams { - scheduleId: string; - scheduledFor: Date; - /** Unique identity of this resume attempt; every later transition is fenced on it. */ - holder: string; - ttlMs: number; - /** Global capacity slot to claim if the row does not already hold one. */ - capacitySlot: number; -} - -/** - * `acquired` = this attempt owns the occurrence; `held` = another live attempt owns it; - * `overlap` = a DIFFERENT occurrence holds the schedule's active slot; `slot-taken` = - * the global capacity slot collided (allocator should try the next one); `gone` = the - * occurrence no longer exists or is already terminal. - */ -export type ResumeLeaseResult = - | { outcome: 'acquired'; holder: string; resumeSeq: number; capacitySlot?: number } - | { outcome: 'held' } - | { outcome: 'overlap' } - | { outcome: 'slot-taken' } - | { outcome: 'gone' }; - /** Outcome of promoting a paused occurrence back into the single active slot. */ export type PromoteRunResult = 'promoted' | 'overlap' | 'missing'; @@ -157,15 +131,6 @@ export type ScheduleMethods = { insertScheduleRun: (data: Partial) => Promise; reserveStartedRun: (data: Partial) => Promise; getCapacityOccupancy: () => Promise<{ takenSlots: number[]; unslotted: number }>; - acquireResumeLease: (params: AcquireResumeLeaseParams) => Promise; - markResumeClaimed: ( - scheduleId: string, - scheduledFor: Date, - holder: string, - ttlMs: number, - ) => Promise; - commitResumeLease: (scheduleId: string, scheduledFor: Date, holder: string) => Promise; - releaseResumeLease: (scheduleId: string, scheduledFor: Date, holder: string) => Promise; requestRunAbort: (scheduleId: string, scheduledFor: Date) => Promise; getRun: (scheduleId: string, scheduledFor: Date) => Promise; setRunFireDetails: ( @@ -636,160 +601,6 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche return { takenSlots, unslotted }; } - /** - * Acquires the durable RESUMING lease for a HITL resume, atomically: promotes the - * occurrence into the single active slot, claims a global capacity slot, mints the - * next `resumeSeq` (the CAS token that fences stale pause writes), and stamps the - * holder so only this attempt may drive the row. - * - * Adoptable states (the $or): a paused row; a row already `started` with NO holder - * (its pause bookkeeping never landed); or a row whose PRE-CLAIM lease expired (a - * crashed resume that never consumed the approval). A post-claim lease is never - * stolen — that attempt must roll forward, not be re-run. - */ - async function acquireResumeLease(params: AcquireResumeLeaseParams): Promise { - const { scheduleId, scheduledFor, holder, ttlMs, capacitySlot } = params; - try { - const row = await ScheduleRun() - .findOneAndUpdate( - { - scheduleId, - scheduledFor, - $or: [ - { status: 'requires_action' }, - { status: 'started', resumeHolder: { $exists: false } }, - { - status: 'started', - resumeClaimedAt: { $exists: false }, - $expr: { $lt: [{ $ifNull: ['$resumeExpiresAt', new Date(0)] }, '$$NOW'] }, - }, - ], - }, - [ - { - $set: { - status: 'started', - resumeSeq: { $add: [{ $ifNull: ['$resumeSeq', 0] }, 1] }, - resumeHolder: holder, - resumeExpiresAt: { $add: ['$$NOW', ttlMs] }, - capacitySlot, - // Whether this lease ADOPTED a row that was already `started` rather than - // promoting a paused one. Expressions in a pipeline $set see the PRE-stage - // document, so this observes the status before the line above changes it. - // Release must never demote an adopted row: a committed resume clears its - // holder, so a late duplicate attempt can otherwise adopt a LIVE running - // run and roll it back to requires_action. - resumeAdopted: { $eq: ['$status', 'started'] }, - }, - }, - { $unset: ['resumeClaimedAt'] }, - ], - { new: true }, - ) - .lean(); - if (row == null) { - // Distinguish "another attempt holds it" from "the occurrence is gone/terminal" - // instead of collapsing both to a lossy 'missing'. - const current = await ScheduleRun() - .findOne({ scheduleId, scheduledFor }) - .select('status resumeHolder') - .lean<{ status?: ScheduleRunStatus; resumeHolder?: string }>(); - if (current == null) { - return { outcome: 'gone' }; - } - if (current.resumeHolder != null && current.status === 'started') { - return { outcome: 'held' }; - } - return { outcome: 'gone' }; - } - return { - outcome: 'acquired', - holder, - resumeSeq: row.resumeSeq ?? 0, - capacitySlot: row.capacitySlot, - }; - } catch (error) { - if (isCapacitySlotConflict(error)) { - return { outcome: 'slot-taken' }; - } - if (isActiveRunConflict(error)) { - return { outcome: 'overlap' }; - } - throw error; - } - } - - /** Marks the resume lease as having consumed the approval. After this point a crash - * must roll FORWARD (the approval is spent) and the lease is no longer adoptable. */ - async function markResumeClaimed( - scheduleId: string, - scheduledFor: Date, - holder: string, - ttlMs: number, - ): Promise { - const result = await ScheduleRun().updateOne( - { scheduleId, scheduledFor, resumeHolder: holder }, - [{ $set: { resumeClaimedAt: '$$NOW', resumeExpiresAt: { $add: ['$$NOW', ttlMs] } } }], - ); - return (result.matchedCount ?? 0) > 0; - } - - /** Commits the resume: the generation is reconstructed and running, so the row is a - * normal `started` run again and the lease fields are dropped. Holder-fenced. */ - async function commitResumeLease( - scheduleId: string, - scheduledFor: Date, - holder: string, - ): Promise { - const result = await ScheduleRun().updateOne( - { scheduleId, scheduledFor, resumeHolder: holder }, - { $unset: { resumeHolder: 1, resumeExpiresAt: 1, resumeClaimedAt: 1, resumeAdopted: 1 } }, - ); - return (result.matchedCount ?? 0) > 0; - } - - /** Rolls a PRE-CLAIM resume back to `requires_action`, freeing the capacity slot so - * the approval stays actionable. Refuses once the approval was claimed (post-claim - * must terminalize instead) and is holder-fenced so a stale attempt can't demote. */ - async function releaseResumeLease( - scheduleId: string, - scheduledFor: Date, - holder: string, - ): Promise { - // Demote ONLY a row this holder actually promoted out of `requires_action`. An - // ADOPTED row was already `started` (a committed resume, or one whose pause - // bookkeeping never landed), and demoting it would knock a live generation back to - // requires_action and free its capacity slot underneath it. - const promoted = await ScheduleRun().updateOne( - { - scheduleId, - scheduledFor, - resumeHolder: holder, - resumeClaimedAt: { $exists: false }, - resumeAdopted: { $ne: true }, - }, - { - $set: { status: 'requires_action' }, - $unset: { resumeHolder: 1, resumeExpiresAt: 1, capacitySlot: 1, resumeAdopted: 1 }, - }, - ); - if ((promoted.matchedCount ?? 0) > 0) { - return true; - } - // Adopted row: drop only OUR lease so another attempt can take it, leaving the run - // (and its slot) exactly as we found it. - const released = await ScheduleRun().updateOne( - { - scheduleId, - scheduledFor, - resumeHolder: holder, - resumeClaimedAt: { $exists: false }, - }, - { $unset: { resumeHolder: 1, resumeExpiresAt: 1, resumeAdopted: 1 } }, - ); - return (released.matchedCount ?? 0) > 0; - } - /** Records that an abort was requested WITHOUT freeing the capacity slot: the run * keeps counting against fireConcurrency until its generation owner confirms * settlement by writing a terminal outcome. */ @@ -900,24 +711,11 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche // instead hide the pause from the card until some later terminal outcome. Guard on // a matching active run first so a spoofed scheduleId can't write a card. Keyed on // existence, not modification, so a retried pause still re-affirms the card. - // EPOCH CAS: a pause may only land on the segment that produced it. A resume - // bumps resumeSeq, so a stale requires_action callback from the PREVIOUS segment - // no longer matches and cannot demote the run the resume just promoted. - // `{field: null}` matches null OR missing in Mongo, so epoch 0 must be expressed - // as $in [0, null] — a never-resumed row has no resumeSeq at all, and a plain - // equality on 0 would never match it (silently disabling the fence). - let epochFilter: Record = {}; - if (params.expectResumeSeq === 0) { - epochFilter = { resumeSeq: { $in: [0, null] } }; - } else if (params.expectResumeSeq != null) { - epochFilter = { resumeSeq: params.expectResumeSeq }; - } const activeRun = await ScheduleRun() .findOne({ scheduleId: params.scheduleId, scheduledFor: params.scheduledFor, status: { $in: ['started', 'requires_action'] }, - ...epochFilter, }) .select('_id') .lean(); @@ -937,7 +735,6 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche scheduleId: params.scheduleId, scheduledFor: params.scheduledFor, status: { $in: ['started', 'requires_action'] }, - ...epochFilter, }, { $set: { @@ -972,13 +769,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche // actually stopped, so this is the ONLY place the global capacity slot is // released. An abort request alone does not free it (see requestRunAbort). // Any in-flight resume lease ends with the run. - $unset: { - capacitySlot: 1, - resumeHolder: 1, - resumeExpiresAt: 1, - resumeClaimedAt: 1, - resumeAdopted: 1, - }, + $unset: { capacitySlot: 1 }, }, ); // No-match guard: never touch schedule bookkeeping without a matching run @@ -1277,10 +1068,6 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche insertScheduleRun, reserveStartedRun, getCapacityOccupancy, - acquireResumeLease, - markResumeClaimed, - commitResumeLease, - releaseResumeLease, requestRunAbort, getRun, setRunFireDetails, diff --git a/packages/data-schemas/src/schema/scheduleRun.ts b/packages/data-schemas/src/schema/scheduleRun.ts index 68f4da4146..3aa941bfb3 100644 --- a/packages/data-schemas/src/schema/scheduleRun.ts +++ b/packages/data-schemas/src/schema/scheduleRun.ts @@ -59,32 +59,6 @@ const scheduleRunSchema: Schema = new Schema( bookkept: { type: Boolean, }, - /** Monotonic per-occurrence segment counter. Incremented by every resume - * reservation; a pause write must CAS on the epoch its segment observed, so a - * stale `requires_action` callback can never demote an already-resumed run. */ - resumeSeq: { - type: Number, - min: 0, - }, - /** Identity of the in-flight resume attempt. Presence means RESUMING: the run is - * `started` but its generation has not been reconstructed yet. */ - resumeHolder: { - type: String, - }, - /** Deadline for the current resume phase, so a crashed resume is reclaimable. */ - resumeExpiresAt: { - type: Date, - }, - /** Set once the approval claim succeeded; discriminates pre-claim (rollback-safe) - * from post-claim (must roll forward) recovery. */ - resumeClaimedAt: { - type: Date, - }, - /** True when the lease ADOPTED an already-`started` row instead of promoting a - * paused one. Release must not demote an adopted row (it may be a live run). */ - resumeAdopted: { - type: Boolean, - }, /** Global concurrency slot held while `started`. The unique partial index below * turns fireConcurrency into a DB-enforced bound instead of a racy count. */ capacitySlot: { @@ -109,18 +83,17 @@ const scheduleRunSchema: Schema = new Schema( ); scheduleRunSchema.index({ scheduleId: 1, scheduledFor: 1 }, { unique: true }); -// At most ONE active (`started`) run per schedule, enforced by the DB rather than -// a read-then-write check. Makes both the fire-path overlap skip and the HITL -// resume promotion atomic: a second occurrence inserting/promoting to `started` -// while one is already active fails with a duplicate-key error instead of racing. +// At most ONE active (`started`) run per schedule, enforced by the DB rather than a +// read-then-write check: a second occurrence inserting while one is already active +// fails with a duplicate-key error instead of racing. scheduleRunSchema.index( { scheduleId: 1 }, { unique: true, partialFilterExpression: { status: 'started' } }, ); // GLOBAL fireConcurrency, enforced by the DB rather than a read-then-compare count. -// Every transition into `started` (fire insert, resume promotion) claims a slot in -// [0, fireConcurrency) in the SAME write; a duplicate slot is rejected atomically, so -// two concurrent admissions of DIFFERENT schedules can never both pass a cap-1 check. +// A fire claims a slot in [0, fireConcurrency) in the SAME write that inserts the run; +// a duplicate slot is rejected atomically, so two concurrent admissions of DIFFERENT +// schedules can never both pass a cap-1 check. // Partial + $exists so legacy slotless rows (written before this field) never collide. scheduleRunSchema.index( { capacitySlot: 1 }, diff --git a/packages/data-schemas/src/types/schedule.ts b/packages/data-schemas/src/types/schedule.ts index 26c998c0b3..3ba0ea81bd 100644 --- a/packages/data-schemas/src/types/schedule.ts +++ b/packages/data-schemas/src/types/schedule.ts @@ -59,15 +59,6 @@ export interface IScheduleRun { droppedFileIds?: string[]; durationMs?: number; bookkept?: boolean; - /** Monotonic segment counter; the CAS token a pause write must match. */ - resumeSeq?: number; - /** Identity of the in-flight resume attempt (presence == RESUMING). */ - resumeHolder?: string; - resumeExpiresAt?: Date; - /** Set once the approval claim succeeded (pre-claim vs post-claim recovery). */ - resumeClaimedAt?: Date; - /** True when the lease adopted an already-`started` row rather than promoting a paused one. */ - resumeAdopted?: boolean; /** Global concurrency slot held while `started`. */ capacitySlot?: number; /** When an abort was requested; capacity is held until settlement is confirmed. */