diff --git a/packages/api/src/schedules/fire.spec.ts b/packages/api/src/schedules/fire.spec.ts index 44675bbb3f..f5400a124c 100644 --- a/packages/api/src/schedules/fire.spec.ts +++ b/packages/api/src/schedules/fire.spec.ts @@ -413,11 +413,32 @@ describe('fireSchedule', () => { expect(global.fetch).toHaveBeenCalledTimes(1); }); + it('does not reserve a run when the claim already lapsed during preflight', async () => { + const { methods, runs } = makeMethods(); + // The preflight (user/config/permission/balance/attachment queries) outlasted the + // 5-minute lease and another worker re-claimed the occurrence. + (methods.revalidateClaim as jest.Mock).mockResolvedValue(false); + mockFetch(async () => okResponse()); + const result = await fireSchedule(makeDeps(methods), makeSchedule(), LIMITS, dueAt()); + expect(result.skipped).toBe('superseded'); + // Reserving here would win the occurrence's unique row: the FRESH claimer would then + // see `duplicate` and advance without firing, while this worker's own revalidation + // fails and rollbackReservation deliberately retains the row (leaseBy changed). The + // occurrence would be lost with its capacity slot held until the orphan sweep. + expect(methods.reserveStartedRun).not.toHaveBeenCalled(); + expect([...runs.entries()].some(([k]) => k.startsWith('sched-1:'))).toBe(false); + expect(global.fetch).not.toHaveBeenCalled(); + // The lease is handed back by holder so the fresh claimer is not left waiting. + expect(methods.releaseLeaseByHolder).toHaveBeenCalledWith('sched-1', 'inst-1'); + }); + it('preserves the reserved run for reconcile when the lease was taken over', async () => { const { methods, runs } = makeMethods(); // An owner edit/takeover superseded this fire after it reserved its run, which is // the path that now drives rollbackReservation (capacity no longer inserts at all). - (methods.revalidateClaim as jest.Mock).mockResolvedValue(false); + // Valid at the pre-reserve check, superseded by the pre-POST one: that ordering IS + // the scenario, since a claim already dead before reserving never reserves at all. + (methods.revalidateClaim as jest.Mock).mockResolvedValueOnce(true).mockResolvedValue(false); // Simulate a lease takeover: this worker no longer holds the claim. (methods.holdsLease as jest.Mock).mockResolvedValue(false); mockFetch(async () => okResponse()); @@ -433,7 +454,7 @@ describe('fireSchedule', () => { const { methods, runs } = makeMethods(); // Account deletion hard-deleted the schedule after this fire reserved its run: // revalidation fails, the lease is not held AND the schedule no longer exists. - (methods.revalidateClaim as jest.Mock).mockResolvedValue(false); + (methods.revalidateClaim as jest.Mock).mockResolvedValueOnce(true).mockResolvedValue(false); (methods.holdsLease as jest.Mock).mockResolvedValue(false); (methods.scheduleExists as jest.Mock).mockResolvedValue(false); mockFetch(async () => okResponse()); diff --git a/packages/api/src/schedules/fire.ts b/packages/api/src/schedules/fire.ts index 79d06ded96..dcd7a6a217 100644 --- a/packages/api/src/schedules/fire.ts +++ b/packages/api/src/schedules/fire.ts @@ -252,6 +252,20 @@ export async function fireSchedule( } }; + /** + * Steps aside from a superseded fire (owner edit/delete, or a lease-expiry re-claim). + * `advance()` is fenced on the OLD claim token, which the edit rotated, so it no-ops + * and would leave this worker's lease held until its TTL — reporting the edited + * schedule / Run now as "already in progress" though no run was dispatched. Releasing + * by HOLDER makes it immediately re-claimable, and correctly no-ops after a takeover + * (leaseBy changed) so the new holder's lease is never stripped. + */ + const releaseSupersededLease = async () => { + if (schedule.leaseBy != null) { + await methods.releaseLeaseByHolder(schedule.id, schedule.leaseBy); + } + }; + if (nextRunAt == null) { await methods.disableSchedule(schedule.id, 'invalid_schedule', claimToken); await advance(); @@ -368,6 +382,23 @@ export async function fireSchedule( (id) => !files.some((file) => file.file_id === id), ); + // Revalidate BEFORE reserving, not only before the POST below. The preflight above + // (user, config, permission, balance, attachment queries) can outlast the 5-minute + // lease, and a stale worker that reserves anyway wins the occurrence's unique row: + // the fresh claimer then sees `duplicate` and advances without firing, while this + // worker's own revalidation fails and rollbackReservation deliberately RETAINS the + // row (the lease takeover changed leaseBy). The occurrence is lost and its global + // capacity slot stays held until the 30-minute orphan sweep. Nothing is reserved yet + // here, so a superseded fire simply steps aside. + if ( + claimToken != null && + !(await methods.revalidateClaim(schedule.id, claimToken, !options?.manual)) + ) { + await releaseSupersededLease(); + await advance(); + return { fired: false, skipped: 'superseded' as const }; + } + // Pre-generate the conversation id and reserve the run row up front. The // loopback POST reuses it (streamId === conversationId), so reconciliation can // ALWAYS locate this occurrence's job — even if the post-accept detail write @@ -457,15 +488,7 @@ export async function fireSchedule( !(await methods.revalidateClaim(schedule.id, claimToken, !options?.manual)) ) { await rollbackReservation(); - // This fire is superseded (owner edit/delete). advance() is fenced on the OLD - // claim token, which the edit rotated, so it no-ops and would leave this - // worker's lease held until its TTL — reporting the edited schedule / Run now - // as "already in progress" though no run was dispatched. Release our own lease - // by holder (leaseBy) so it's immediately re-claimable; a takeover changed - // leaseBy, so this correctly no-ops there and never strips the new holder's lease. - if (schedule.leaseBy != null) { - await methods.releaseLeaseByHolder(schedule.id, schedule.leaseBy); - } + await releaseSupersededLease(); await advance(); return { fired: false, skipped: 'superseded' as const }; }