From ee00c608593bb056095f36da1b18a6927ae70864 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 02:13:07 +0000 Subject: [PATCH] fix(schedules): fence the clustered pause replay against a concurrent resume claim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex round 5, on code I added in f735341. The finding is correct and the exposure is one I introduced. The pause replay lives in a sweep that runs in EVERY clustered replica, so several sweepers can observe the same unprojected pause. Its guards were all derived from an in-memory row SNAPSHOT: hasResumeHandoffInFlight reads the snapshot's resumeClaimedAt, and recordRunOutcome's pause branch matched any row currently in `started`/`requires_action` with no fence of its own. The race: sweeper A projects the pause and frees the slot; the owner's approval then claims a fresh one (markRunResumeClaimed takes the row to `started` WITH resumeClaimedAt in one write); sweeper B, still holding the pre-projection snapshot, passes its hand-off check and replays — `$unset: { capacitySlot, resumeClaimedAt }` under a continuation that is already running. The run reverts to `requires_action` while its generation proceeds outside global capacity. The engine's reconciler makes the same call and has the same snapshot-derived guard, but v1 arms exactly one engine, so it has no concurrent racer; the sweep is the first thing to run this transition in parallel. Left the engine alone rather than widening the change: its `requires_action` re-affirmation is deliberate and single-writer. Fixed where the race is, in the write itself. `recordRunOutcome` takes an optional `requireNoResumeClaim`, which adds `resumeClaimedAt: { $exists: false }` to the pause filter, and the sweep sets it. Because the stamp is written in the SAME update that moves the row to `started`, its absence is atomic proof no resume owns the row. The flag is deliberately NOT set by the generation owner: its own re-pause legitimately clears the stamp as the hand-off's completion signal. The fence cannot block the recovery it exists to enable: markRunResumeClaimed only matches `requires_action`, so a genuinely stuck `started` row can carry no resume claim — it is in fact blocking its own approval until this replay frees it. Tests: a replay from a stale snapshot leaves a resume-claimed row's status, slot, and claim stamp intact, while a stuck row with no claim still recovers; and the sweep is asserted to send the fence. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq --- packages/api/src/schedules/erasure.spec.ts | 3 + packages/api/src/schedules/erasure.ts | 6 ++ .../src/methods/schedule.methods.spec.ts | 69 +++++++++++++++++++ packages/data-schemas/src/methods/schedule.ts | 17 +++++ 4 files changed, 95 insertions(+) diff --git a/packages/api/src/schedules/erasure.spec.ts b/packages/api/src/schedules/erasure.spec.ts index 064c45df1a..3bb6a815af 100644 --- a/packages/api/src/schedules/erasure.spec.ts +++ b/packages/api/src/schedules/erasure.spec.ts @@ -281,6 +281,9 @@ describe('topology-safe dead-delivery convergence', () => { scheduleId: 'schedule-1', status: 'requires_action', conversationId: 'conversation-1', + // Every replica runs this sweep, so the write itself must be fenced against a + // resume that claimed the row after this snapshot was taken. + requireNoResumeClaim: true, }), ); // The job is still LIVE awaiting approval — its evidence must not be released. diff --git a/packages/api/src/schedules/erasure.ts b/packages/api/src/schedules/erasure.ts index 82ac826db2..3109f29b4e 100644 --- a/packages/api/src/schedules/erasure.ts +++ b/packages/api/src/schedules/erasure.ts @@ -153,6 +153,12 @@ async function settleFromObservedJob( status: 'requires_action', conversationId: run.conversationId, autoDisableAfterFailures: Number.MAX_SAFE_INTEGER, + // Every clustered replica runs this sweep, so N sweepers can observe the same + // unprojected pause. The `run` above is a SNAPSHOT: once one sweeper projects the + // pause, the owner's approval can claim a fresh slot, and a peer still holding the + // pre-projection snapshot would pass hasResumeHandoffInFlight and unset that slot + // and claim stamp under the running continuation. Fence it in the write itself. + requireNoResumeClaim: true, }); return; } diff --git a/packages/data-schemas/src/methods/schedule.methods.spec.ts b/packages/data-schemas/src/methods/schedule.methods.spec.ts index b50d590348..67b81f559f 100644 --- a/packages/data-schemas/src/methods/schedule.methods.spec.ts +++ b/packages/data-schemas/src/methods/schedule.methods.spec.ts @@ -2266,6 +2266,75 @@ describe('updateScheduleById revision fence', () => { }); }); +describe('pause replay is fenced against a concurrent resume claim', () => { + /** + * Every clustered replica runs the erasure sweep, so several can observe the same + * unprojected pause. The first projects it; the owner's approval then claims a fresh + * slot; a peer still holding the PRE-projection snapshot passes its in-memory + * hand-off check and replays the pause — unsetting `capacitySlot` and + * `resumeClaimedAt` under a continuation that is already running. The fence has to + * live in the write, not in the caller's snapshot. + */ + it('refuses to replay a pause onto a row a resume has claimed', async () => { + const schedule = await methods.createSchedule(scheduleData()); + const scheduledFor = new Date('2030-03-01T00:00:00Z'); + await methods.reserveStartedRun( + runData(schedule, { scheduledFor, conversationId: 'convo-race', capacitySlot: 0 }), + ); + // Sweeper A wins: the row leaves `started` and frees its slot. + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'requires_action', + conversationId: 'convo-race', + autoDisableAfterFailures: Number.MAX_SAFE_INTEGER, + requireNoResumeClaim: true, + }); + // The approval now claims a FRESH slot; this is the state a stale peer must not touch. + const claimed = await methods.markRunResumeClaimed(schedule.id, scheduledFor, 3); + expect(claimed).toEqual({ capacitySlot: 3 }); + + // Sweeper B replays from its pre-projection snapshot. + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'requires_action', + conversationId: 'convo-race', + autoDisableAfterFailures: Number.MAX_SAFE_INTEGER, + requireNoResumeClaim: true, + }); + + const run = await ScheduleRun.findOne({ scheduleId: schedule.id, scheduledFor }).lean(); + expect(run?.status).toBe('started'); + expect(run?.capacitySlot).toBe(3); + expect(run?.resumeClaimedAt).toBeDefined(); + }); + + /** The fence must not block the recovery it exists to enable: a genuinely stuck + * `started` row carries no resume claim, because markRunResumeClaimed only matches + * `requires_action` — the stuck row blocks its own approval until this replay runs. */ + it('still recovers a stuck row that carries no resume claim', async () => { + const schedule = await methods.createSchedule(scheduleData()); + const scheduledFor = new Date('2030-03-02T00:00:00Z'); + await methods.reserveStartedRun( + runData(schedule, { scheduledFor, conversationId: 'convo-stuck', capacitySlot: 1 }), + ); + + await methods.recordRunOutcome({ + scheduleId: schedule.id, + scheduledFor, + status: 'requires_action', + conversationId: 'convo-stuck', + autoDisableAfterFailures: Number.MAX_SAFE_INTEGER, + requireNoResumeClaim: true, + }); + + const run = await ScheduleRun.findOne({ scheduleId: schedule.id, scheduledFor }).lean(); + expect(run?.status).toBe('requires_action'); + expect(run?.capacitySlot).toBeUndefined(); + }); +}); + describe('reconciliation rotates the paused window', () => { /** * Paused runs accumulate without bound (no capacity slot, no occurrence block) and diff --git a/packages/data-schemas/src/methods/schedule.ts b/packages/data-schemas/src/methods/schedule.ts index 1b9e0fcc2d..0fc64f5cf6 100644 --- a/packages/data-schemas/src/methods/schedule.ts +++ b/packages/data-schemas/src/methods/schedule.ts @@ -114,6 +114,20 @@ export interface RecordRunOutcomeParams { /** Consecutive-balance-skip auto-disable threshold; required to settle a run as * `skipped_balance` (a mid-generation balance refusal), ignored otherwise. */ balanceSkipDisableThreshold?: number; + /** + * RECOVERY REPLAY of a pause, from an actor holding a possibly-stale row snapshot: + * apply the transition ONLY while no resume claim exists. Without it, two clustered + * sweepers observing the same unprojected pause race — the first projects it, the + * owner's approval then claims a fresh slot (`started` + `resumeClaimedAt`), and the + * second still passes its snapshot-based hand-off check and unsets the capacity slot + * and claim stamp out from under the running continuation. `markRunResumeClaimed` + * sets the stamp in the SAME write that takes the row to `started`, so its absence is + * an atomic proof that no resume owns this row. + * + * Never set by the generation owner: its own re-pause legitimately clears the stamp as + * the hand-off's completion signal. + */ + requireNoResumeClaim?: boolean; } /** Result of claiming/leasing a schedule: the snapshot plus the fencing token to carry. */ @@ -1291,6 +1305,9 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche scheduleId: params.scheduleId, scheduledFor: params.scheduledFor, status: { $in: ['started', 'requires_action'] }, + // See requireNoResumeClaim: fences a stale-snapshot recovery replay against a + // resume that claimed the row after the snapshot was taken. + ...(params.requireNoResumeClaim ? { resumeClaimedAt: { $exists: false } } : {}), }, { $set: {