fix: release superseded engine claims

This commit is contained in:
Danny Avila 2026-08-17 13:15:02 -04:00
parent d107ded8cd
commit ecddfa0376
2 changed files with 36 additions and 6 deletions

View file

@ -49,6 +49,7 @@ function makeMethods(schedule: FireableSchedule) {
advanceSchedule: jest.fn(
async (_id: string, _next: Date | null, _from?: Date, _token?: string) => true,
),
releaseLeaseByHolder: jest.fn(async () => undefined),
disableSchedule: jest.fn(async (_id: string, _reason: string, _token?: string) => undefined),
releaseLease: jest.fn(async () => true),
holdsLease: jest.fn(async () => true),
@ -135,6 +136,18 @@ describe('runTick misfire skip-forward', () => {
expect(fetchSpy).not.toHaveBeenCalled();
});
it('releases its old holder when an owner edit fences the misfire advance', async () => {
const schedule = makeClaimedSchedule({ nextRunAt: staleAt() });
const methods = makeMethods(schedule);
// The edit rotated claimToken/nextRunAt after this worker claimed but preserved
// its unique lease holder, so the token-fenced advance no longer matches.
methods.advanceSchedule.mockResolvedValueOnce(false);
await tickOnce(makeDeps(methods));
expect(methods.releaseLeaseByHolder).toHaveBeenCalledWith('sched-1', 'inst-1');
});
it('disables and clears a stale occurrence whose cadence is uncomputable', async () => {
const schedule = makeClaimedSchedule({ nextRunAt: staleAt(), timezone: 'Not/AZone' });
const methods = makeMethods(schedule);

View file

@ -402,6 +402,27 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
return true;
}
const scheduledFor = schedule.nextRunAt ?? new Date();
/**
* Advances this claimed occurrence without preserving a superseded holder.
* Owner edits rotate the token/next occurrence but intentionally leave the old
* lease in place; a fenced miss therefore releases only this worker's unique
* holder. Infrastructure errors keep the lease as retry backoff, as before.
*/
const advanceClaim = async (nextRunAt: Date | null): Promise<void> => {
try {
const advanced = await deps.methods.advanceSchedule(
schedule.id,
nextRunAt,
scheduledFor,
schedule.claimToken,
);
if (!advanced && schedule.leaseBy != null) {
await deps.methods.releaseLeaseByHolder(schedule.id, schedule.leaseBy);
}
} catch {
// Leave the claim lease as bounded backoff for transient storage failures.
}
};
// Use the CLAIM's clock for the misfire cutoff: the claim wrote
// leaseUntil = now + LEASE_MS from the claiming worker's clock, so
// leaseUntil - LEASE_MS is that worker's "now" at claim — usually this very
@ -440,9 +461,7 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
// next FUTURE occurrence. Without it nextRunAt keeps pointing at the stale
// one, so every later tick reclaims and skips the same occurrence forever and
// a schedule overdue past an outage never fires again.
await deps.methods
.advanceSchedule(schedule.id, next, scheduledFor, schedule.claimToken)
.catch(() => undefined);
await advanceClaim(next);
logger.info(`[schedules] skipped stale occurrence for ${schedule.id} (misfire grace)`);
return false;
}
@ -492,9 +511,7 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
.then(() => true)
.catch(() => false);
if (disabled) {
await deps.methods
.advanceSchedule(schedule.id, null, scheduledFor, schedule.claimToken)
.catch(() => undefined);
await advanceClaim(null);
}
}
}