From 05666702b3de81d62585c87aac3e918ecd9f2d56 Mon Sep 17 00:00:00 2001 From: Sina Matian Date: Fri, 11 Sep 2026 13:55:19 -0700 Subject: [PATCH 1/5] Back off failed cron fire retries --- src/cron/cron-store.ts | 165 ++++++++++++++++++++++++++---------- src/cron/scheduler.ts | 56 ++++++++---- src/types.ts | 7 ++ test/cron-queue.test.ts | 73 ++++++++++++++-- test/cron-scheduler.test.ts | 131 +++++++++++++++++++++++++++- test/cron-store.test.ts | 128 ++++++++++++++++++++++++---- 6 files changed, 475 insertions(+), 85 deletions(-) diff --git a/src/cron/cron-store.ts b/src/cron/cron-store.ts index df0bdc4e1..cd0a2f67a 100644 --- a/src/cron/cron-store.ts +++ b/src/cron/cron-store.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import type { Cron, CronFireLogEntry, @@ -19,7 +20,6 @@ import { buildTriggerBase, contentPart, createDeduped, - setTriggerRecipientConsent, type CreateTriggerInput, } from "../triggers/trigger-store.ts"; import { hashId } from "../util/crypto.ts"; @@ -57,6 +57,18 @@ export const FIRE_RETENTION_MS = 90 * 24 * 60 * 60 * 1000; export const FIRE_RETENTION_KEEP_PER_CRON = 100; +const FAILURE_BACKOFF_BASE_MS = 5_000; +const FAILURE_BACKOFF_MAX_MS = 5 * 60_000; +const FAILURE_BACKOFF_MAX_FAILURES = 7; + +export interface CronSlotClaim { + id: string; + cron: Cron; + scheduledAt: number; + claimedAt: number; + priorLastFiredAt?: number; +} + export interface CronStore { create(input: CreateCronInput): Promise; get(id: string): Promise; @@ -78,8 +90,10 @@ export interface CronStore { markFired(id: string, at: number, scheduledAt?: number): Promise; markAttempted(id: string, at: number): Promise; defer(id: string, until: number): Promise; - claimSlot(id: string, scheduledAt: number, at: number): Promise; - unclaimSlot(id: string, scheduledAt: number, at: number, priorLastFiredAt: number | undefined): Promise; + claimSlot(id: string, scheduledAt: number, at: number): Promise; + completeSlot(id: string, claim: CronSlotClaim): Promise; + releaseSlot(id: string, claim: CronSlotClaim, deferUntil?: number): Promise; + failSlot(id: string, claim: CronSlotClaim, failedAt: number): Promise; due(now: number): Promise>; } @@ -93,12 +107,43 @@ function normalizeTitle(title: string | undefined): string | undefined { return trimmed.length > 80 ? `${trimmed.slice(0, 79)}...` : trimmed; } +function mergeFields(cron: Cron, fields: Partial): Cron { + const next = { ...cron }; + for (const [key, value] of Object.entries(fields)) { + if (value === undefined) delete (next as Record)[key]; + else (next as Record)[key] = value; + } + return next; +} + +function clearAttemptState(cron: Cron): Cron { + const { activeClaimId: _activeClaimId, failureBackoff: _failureBackoff, ...rest } = cron; + return rest; +} + +function requireAtomicUpdate(backing: DurableMap): NonNullable["update"]> { + if (!backing.update) throw new Error("cron store requires atomic durable-map updates"); + return backing.update; +} + +function failureBackoffMs(failures: number): number { + return Math.min(FAILURE_BACKOFF_MAX_MS, FAILURE_BACKOFF_BASE_MS * 2 ** (failures - 1)); +} + export function createCronStore( backing: DurableMap = createMemoryMap(), opts?: { staleRunningMs?: number; fires?: CronFireStore }, ): CronStore { const staleRunningMs = opts?.staleRunningMs ?? DEFAULT_FIRE_RUNNING_STALE_MS; const fires = opts?.fires ?? createMemoryCronFireStore(); + const updateCron = (id: string, fields: Partial, resetAttempt = true): Promise => { + if (backing.update) + return backing.update(id, (cron) => mergeFields(resetAttempt ? clearAttemptState(cron) : cron, fields)); + return backing.merge( + id, + resetAttempt ? { ...fields, activeClaimId: undefined, failureBackoff: undefined } : fields, + ); + }; return { async create(input) { assertNoEscalation(input); @@ -150,17 +195,17 @@ export function createCronStore( if (patch.members !== undefined) fields.members = patch.members; if (patch.runAs !== undefined) fields.runAs = patch.runAs; if (patch.unattendedGrants !== undefined) fields.unattendedGrants = patch.unattendedGrants; - return backing.merge(id, fields); + return updateCron(id, fields, Object.keys(fields).length > 0); }, delete: (id) => backing.delete(id), async setEnabled(id, enabled) { - await backing.merge(id, { enabled, ...(enabled ? { archived: false } : {}) }); + await updateCron(id, { enabled, ...(enabled ? { archived: false } : {}) }); }, async setDestination(id, destination) { - await backing.merge(id, { destination }); + await updateCron(id, { destination }); }, - setRecipientConsent(id, recipientConsent) { - return setTriggerRecipientConsent(backing, id, recipientConsent); + async setRecipientConsent(id, recipientConsent) { + await updateCron(id, { recipientConsent }); }, async beginFire(id, entry, opts) { if ((await backing.get(id)) === null) return { begun: false }; @@ -224,55 +269,85 @@ export function createCronStore( lastFiredAt: at, nextFireAt: advanceNextFireAt(cron.schedule, advanceFrom), deferUntil: undefined, + activeClaimId: undefined, + failureBackoff: undefined, }); }, async claimSlot(id, scheduledAt, at) { - let claimed = false; - const transform = (cron: Cron): Cron => { - claimed = false; - if (cron.archived || !cron.enabled) return cron; + const claimId = randomUUID(); + let claim: CronSlotClaim | null = null; + await requireAtomicUpdate(backing)(id, (cron) => { + claim = null; + if (cron.archived || !cron.enabled || isDeferred(cron, at)) return cron; if (recoverNextFireAt(cron.schedule, cron.createdAt, cron.lastFiredAt, cron.nextFireAt) !== scheduledAt) return cron; - claimed = true; + claim = { + id: claimId, + cron, + scheduledAt, + claimedAt: at, + ...(cron.lastFiredAt !== undefined ? { priorLastFiredAt: cron.lastFiredAt } : {}), + }; const advanceFrom = isCalendarSchedule(cron.schedule) ? scheduledAt : at; - const next = advanceNextFireAt(cron.schedule, advanceFrom); - const { nextFireAt: _dropped, deferUntil: _cleared, ...rest } = cron; - return { ...rest, lastFiredAt: at, ...(next !== undefined ? { nextFireAt: next } : {}) }; - }; - if (backing.update) { - await backing.update(id, transform); - return claimed; - } - const cron = await backing.get(id); - if (!cron) return false; - const next = transform(cron); - if (!claimed) return false; - await backing.merge(id, { lastFiredAt: next.lastFiredAt, nextFireAt: next.nextFireAt, deferUntil: undefined }); - return true; + const nextFireAt = advanceNextFireAt(cron.schedule, advanceFrom); + const failureBackoff = cron.failureBackoff?.scheduledAt === scheduledAt ? cron.failureBackoff : undefined; + return mergeFields(cron, { + lastFiredAt: at, + nextFireAt, + deferUntil: undefined, + activeClaimId: claimId, + failureBackoff, + }); + }); + return claim; }, - async unclaimSlot(id, scheduledAt, at, priorLastFiredAt) { - const restore = (cron: Cron): Cron => { - if (cron.lastFiredAt !== at) return cron; - const { lastFiredAt: _dropped, ...rest } = cron; - return { - ...rest, - ...(priorLastFiredAt !== undefined ? { lastFiredAt: priorLastFiredAt } : {}), - nextFireAt: scheduledAt, - }; - }; - if (backing.update) { - await backing.update(id, restore); - return; - } - const cron = await backing.get(id); - if (!cron || cron.lastFiredAt !== at) return; - await backing.merge(id, { lastFiredAt: priorLastFiredAt, nextFireAt: scheduledAt }); + async completeSlot(id, claim) { + await requireAtomicUpdate(backing)(id, (cron) => { + if (cron.activeClaimId !== claim.id) return cron; + return mergeFields(cron, { activeClaimId: undefined, failureBackoff: undefined }); + }); + }, + async releaseSlot(id, claim, deferUntil) { + await requireAtomicUpdate(backing)(id, (cron) => { + if (cron.activeClaimId !== claim.id || cron.lastFiredAt !== claim.claimedAt) return cron; + return mergeFields(cron, { + lastFiredAt: claim.priorLastFiredAt, + nextFireAt: claim.scheduledAt, + activeClaimId: undefined, + ...(deferUntil !== undefined ? { deferUntil: Math.max(cron.deferUntil ?? 0, deferUntil) } : {}), + }); + }); + }, + async failSlot(id, claim, failedAt) { + let deferUntil: number | undefined; + await requireAtomicUpdate(backing)(id, (cron) => { + deferUntil = undefined; + if (cron.activeClaimId !== claim.id || cron.lastFiredAt !== claim.claimedAt) return cron; + const failures = + cron.failureBackoff?.scheduledAt === claim.scheduledAt + ? Math.min(FAILURE_BACKOFF_MAX_FAILURES, cron.failureBackoff.failures + 1) + : 1; + deferUntil = Math.max(cron.deferUntil ?? 0, failedAt + failureBackoffMs(failures)); + return mergeFields(cron, { + lastFiredAt: claim.priorLastFiredAt, + nextFireAt: claim.scheduledAt, + activeClaimId: undefined, + failureBackoff: { scheduledAt: claim.scheduledAt, failures }, + deferUntil, + }); + }); + return deferUntil; }, async markAttempted(id, at) { await backing.merge(id, { lastAttemptAt: at }); }, async defer(id, until) { - await backing.merge(id, { deferUntil: until }); + if (backing.update) { + await backing.update(id, (cron) => ({ ...cron, deferUntil: Math.max(cron.deferUntil ?? 0, until) })); + return; + } + const cron = await backing.get(id); + if (cron) await backing.merge(id, { deferUntil: Math.max(cron.deferUntil ?? 0, until) }); }, async due(now) { const due: Array = []; diff --git a/src/cron/scheduler.ts b/src/cron/scheduler.ts index b65f018e0..35102067e 100644 --- a/src/cron/scheduler.ts +++ b/src/cron/scheduler.ts @@ -8,7 +8,7 @@ import { type TurnResult, } from "../types.ts"; import type { IdentityService } from "../identity/identity-service.ts"; -import { isDeferred, type CronStore } from "./cron-store.ts"; +import { isDeferred, type CronSlotClaim, type CronStore } from "./cron-store.ts"; import type { DeliveryStore } from "../delivery/delivery-store.ts"; import type { IdempotencyStore } from "../idempotency/idempotency-store.ts"; import { runTrigger, type TriggerDeps } from "../triggers/run-trigger.ts"; @@ -31,7 +31,7 @@ const FIRE_GC_INTERVAL_MS = 6 * 60 * 60_000; const BUSY_DEFER_MS = 30_000; const BUSY_DEFER_MAX_LATE_MS = 10 * 60_000; -type FireResult = { authzFailed: boolean; deferred?: boolean }; +type FireResult = { authzFailed: boolean; deferUntil?: number }; type RunNowResult = | { started: true; fireKey: string; settled: Promise } @@ -276,8 +276,7 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { status: "deferred", note: `session busy — retrying at ${utcMinute(deferUntil)}`, }); - await deps.crons.defer(cron.id, deferUntil); - return { authzFailed: false, deferred: true }; + return { authzFailed: false, deferUntil }; } if (outcome.ran || outcome.authzFailed) { await deps.crons.recordFire(cron.id, { @@ -292,10 +291,7 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { ...(outcome.sessionId ? { sessionId: outcome.sessionId } : {}), }); } - if (outcome.authzFailed) { - await deps.crons.setEnabled(cron.id, false); - return { authzFailed: true }; - } + if (outcome.authzFailed) return { authzFailed: true }; if (isOneShotSchedule(cron.schedule)) await deps.crons.setEnabled(cron.id, false); return { authzFailed: false }; } @@ -342,11 +338,31 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { console.warn(`[scheduler] fan-out capped: firing ${batch.length}/${due.length} due crons this tick`); } for (const cron of batch) { + let claim: CronSlotClaim | null = null; try { - const { authzFailed, deferred } = await fire(cron, t, `cron:${cron.id}:${cron.scheduledAt}`, cron.scheduledAt); - if (!authzFailed && !deferred) await deps.crons.markFired(cron.id, t, cron.scheduledAt); + claim = await deps.crons.claimSlot(cron.id, cron.scheduledAt, t); + if (!claim) continue; + const { authzFailed, deferUntil } = await fire( + claim.cron, + t, + `cron:${cron.id}:${cron.scheduledAt}`, + cron.scheduledAt, + ); + if (authzFailed || deferUntil !== undefined) { + await deps.crons.releaseSlot(cron.id, claim, deferUntil); + if (authzFailed) await deps.crons.setEnabled(cron.id, false); + } else { + await deps.crons.completeSlot(cron.id, claim); + } } catch (e) { console.error("[scheduler] fire failed:", errMessage(e)); + if (claim) { + try { + await deps.crons.failSlot(cron.id, claim, now()); + } catch (transitionError) { + console.error("[scheduler] failure backoff failed:", errMessage(transitionError)); + } + } } } }; @@ -402,17 +418,21 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { await deps.jobQueue!.enqueueFire({ ...job, notBefore: cron.deferUntil! }); return; } - if (!(await deps.crons.claimSlot(job.cronId, slot, t))) return; + const claim = await deps.crons.claimSlot(job.cronId, slot, t); + if (!claim) return; try { - const { authzFailed, deferred } = await fire(cron, t, `cron:${cron.id}:${slot}`, slot); - if (authzFailed || deferred) { - await deps.crons.unclaimSlot(job.cronId, slot, t, cron.lastFiredAt); - if (deferred) await enqueueNext(job.cronId); + const { authzFailed, deferUntil } = await fire(claim.cron, t, `cron:${cron.id}:${slot}`, slot); + if (authzFailed || deferUntil !== undefined) { + await deps.crons.releaseSlot(job.cronId, claim, deferUntil); + if (authzFailed) await deps.crons.setEnabled(job.cronId, false); + if (deferUntil !== undefined) await enqueueNext(job.cronId); return; } + await deps.crons.completeSlot(job.cronId, claim); } catch (e) { console.error("[scheduler] fire failed:", errMessage(e)); - await deps.crons.unclaimSlot(job.cronId, slot, t, cron.lastFiredAt); + const deferUntil = await deps.crons.failSlot(job.cronId, claim, now()); + if (deferUntil !== undefined) await deps.jobQueue!.enqueueFire({ ...job, notBefore: deferUntil }); return; } await enqueueNext(job.cronId); @@ -466,7 +486,9 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { : { started: false, reason: "unavailable" }; } const settled = fire(cron, t, fireKey).then( - () => undefined, + async ({ authzFailed }) => { + if (authzFailed) await deps.crons.setEnabled(cronId, false); + }, (e: unknown) => console.error("%s", `[scheduler] manual fire of cron ${cronId} failed:`, errMessage(e)), ); return { started: true, fireKey, settled }; diff --git a/src/types.ts b/src/types.ts index 8eef23602..b4e479837 100644 --- a/src/types.ts +++ b/src/types.ts @@ -235,11 +235,18 @@ export interface CronFireNote { by?: string; } +export interface CronFailureBackoff { + scheduledAt: number; + failures: number; +} + export interface Cron extends TriggerBase { schedule: CronSchedule; nextFireAt?: number; lastAttemptAt?: number; deferUntil?: number; + failureBackoff?: CronFailureBackoff; + activeClaimId?: string; title?: string; archived?: boolean; action?: string; diff --git a/test/cron-queue.test.ts b/test/cron-queue.test.ts index 95a78c8f6..0f3334100 100644 --- a/test/cron-queue.test.ts +++ b/test/cron-queue.test.ts @@ -32,14 +32,26 @@ async function until(cond: () => boolean, ms: number): Promise { while (!cond() && Date.now() < deadline) await new Promise((r) => setTimeout(r, 100)); } -function instance(calls: TurnRequest[], turnMs = 0, fires?: CronFireStore): { scheduler: Scheduler; crons: CronStore } { +async function untilAsync(cond: () => Promise, ms: number): Promise { + const deadline = Date.now() + ms; + while (!(await cond()) && Date.now() < deadline) await new Promise((r) => setTimeout(r, 100)); +} + +function instance( + calls: TurnRequest[], + turnMs = 0, + fires?: CronFireStore, + runOverride?: (req: TurnRequest) => Promise, +): { scheduler: Scheduler; crons: CronStore } { const maps = createPostgresMapFactory(URL!); const crons = createCronStore(maps.map(CRONS_TABLE), fires ? { fires } : undefined); - const run = async (req: TurnRequest): Promise => { - calls.push(req); - if (turnMs) await new Promise((r) => setTimeout(r, turnMs)); - return { status: "ok", reply: "QUEUE-OUTPUT" }; - }; + const run = + runOverride ?? + (async (req: TurnRequest): Promise => { + calls.push(req); + if (turnMs) await new Promise((r) => setTimeout(r, turnMs)); + return { status: "ok", reply: "QUEUE-OUTPUT" }; + }); const scheduler = createScheduler({ crons, deliveries: createDeliveryStore(), @@ -83,6 +95,55 @@ test( }, ); +test( + "pg-boss queue: thrown failures persist a cooldown across instances without changing the fire key", + { skip, timeout: 120_000 }, + async () => { + const calls: Array<{ req: TurnRequest; at: number }> = []; + const a = instance([], 0, undefined, async (req) => { + calls.push({ req, at: Date.now() }); + throw new Error("provider down"); + }); + const b = instance([], 0, undefined, async (req) => { + calls.push({ req, at: Date.now() }); + throw new Error("provider down"); + }); + a.scheduler.start(1_000); + b.scheduler.start(1_000); + try { + const cron = await a.crons.create({ + schedule: { firstFireAt: Date.now() + 500 }, + action: "durable failing queue fire", + owner: "U3", + createdBy: "U3", + ownerScopeId: scopeId("personal", "U3"), + }); + await until(() => calls.length >= 1, 30_000); + assert.equal(calls.length, 1); + await untilAsync(async () => (await b.crons.get(cron.id))?.failureBackoff?.failures === 1, 5_000); + const persisted = await b.crons.get(cron.id); + assert.deepEqual(persisted?.failureBackoff, { scheduledAt: cron.nextFireAt, failures: 1 }); + assert.ok((persisted?.deferUntil ?? 0) >= calls[0]!.at + 5_000); + await new Promise((resolve) => setTimeout(resolve, 2_000)); + assert.equal(calls.length, 1, "a sibling reconcile cannot redeliver before the durable hold"); + + await until(() => calls.length >= 2, 20_000); + assert.equal(calls.length, 2); + assert.ok(calls[1]!.at - calls[0]!.at >= 4_500); + assert.equal(calls[1]!.req.idempotencyKey, calls[0]!.req.idempotencyKey); + await untilAsync(async () => (await a.crons.get(cron.id))?.failureBackoff?.failures === 2, 5_000); + assert.deepEqual((await a.crons.get(cron.id))?.failureBackoff, { + scheduledAt: cron.nextFireAt, + failures: 2, + }); + } finally { + a.scheduler.stop(); + b.scheduler.stop(); + await new Promise((resolve) => setTimeout(resolve, 500)); + } + }, +); + test( "pg-boss queue: a recurring cron chains fires with unique slots, and a schedule edit invalidates queued slots", { skip, timeout: 120_000 }, diff --git a/test/cron-scheduler.test.ts b/test/cron-scheduler.test.ts index 2b7937bc2..0a2ef1fbc 100644 --- a/test/cron-scheduler.test.ts +++ b/test/cron-scheduler.test.ts @@ -533,7 +533,7 @@ test("a failing interval cron does not starve later due crons across ticks", asy assert.deepEqual( calls.map((call) => call.idempotencyKey), - [`cron:${failing.id}:1`, `cron:${succeeding.id}:1`, `cron:${failing.id}:1`, `cron:${succeeding.id}:3000`], + [`cron:${failing.id}:1`, `cron:${succeeding.id}:1`, `cron:${succeeding.id}:3000`], ); }); @@ -1282,6 +1282,66 @@ test("a note written by someone other than the fire renders attributed, and mult assert.doesNotMatch(input, /Notes from last fire agent/); }); +test("interval mode backs off thrown failures without delaying healthy crons", async () => { + const crons = createCronStore(); + const calls: TurnRequest[] = []; + let clock = 1_000; + const scheduler = createScheduler({ + crons, + deliveries: createDeliveryStore(), + idempotency: createIdempotencyStore(), + identity: createIdentityService(), + run: async (req) => { + calls.push(req); + if (req.text.includes("broken task")) throw new Error("provider down"); + return { status: "ok", reply: "done" }; + }, + now: () => clock, + }); + const broken = await crons.create({ + schedule: { firstFireAt: 1_000 }, + action: "broken task", + owner: "U1", + createdBy: "U1", + ownerScopeId: scopeId("personal", "U1"), + }); + const healthy = await crons.create({ + schedule: { firstFireAt: 1_000 }, + action: "healthy task", + owner: "U2", + createdBy: "U2", + ownerScopeId: scopeId("personal", "U2"), + }); + + await scheduler.tick(clock); + assert.equal(calls.length, 2, "the failing fire does not stop its healthy sibling"); + assert.equal((await crons.get(healthy.id))!.enabled, false); + let failed = (await crons.get(broken.id))!; + assert.equal(failed.deferUntil, 6_000); + assert.deepEqual(failed.failureBackoff, { scheduledAt: 1_000, failures: 1 }); + + clock = 5_999; + await scheduler.tick(clock); + assert.equal(calls.length, 2); + clock = 6_000; + await scheduler.tick(clock); + assert.equal(calls.length, 3); + failed = (await crons.get(broken.id))!; + assert.equal(failed.deferUntil, 16_000); + assert.deepEqual(failed.failureBackoff, { scheduledAt: 1_000, failures: 2 }); + + for (const expected of [36_000, 76_000, 156_000, 316_000, 616_000, 916_000]) { + clock = failed.deferUntil!; + await scheduler.tick(clock); + failed = (await crons.get(broken.id))!; + assert.equal(failed.deferUntil, expected); + } + assert.deepEqual( + calls.filter((call) => call.text.includes("broken task")).map((call) => call.idempotencyKey), + Array(8).fill(`cron:${broken.id}:1000`), + ); +}); + function busyOnce(calls: TurnRequest[]) { return async (req: TurnRequest): Promise => { calls.push(req); @@ -1430,3 +1490,72 @@ test("queue mode: a busy fire releases its slot and is re-queued to run after th assert.deepEqual(enqueued.pop(), { cronId: cron.id, scheduledAt: 36_000 }, "the next slot is chained without a hold"); scheduler.stop(); }); + +test("queue mode keeps early jobs and reconciliation behind durable failure backoff", async () => { + const crons = createCronStore(); + const calls: TurnRequest[] = []; + let clock = 1_000; + let onFire: ((job: { cronId: string; scheduledAt: number; notBefore?: number }) => Promise) | undefined; + let onTick: (() => Promise) | undefined; + const enqueued: Array<{ cronId: string; scheduledAt: number; notBefore?: number }> = []; + const scheduler = createScheduler({ + crons, + deliveries: createDeliveryStore(), + idempotency: createIdempotencyStore(), + identity: createIdentityService(), + run: async (req) => { + calls.push(req); + throw new Error("provider down"); + }, + now: () => clock, + jobQueue: { + async start(handlers) { + onFire = handlers.onFire; + onTick = handlers.onTick; + }, + async enqueueFire(job) { + enqueued.push(job); + }, + healthy: () => true, + async stop() {}, + }, + }); + scheduler.start(1_000); + const cron = await crons.create({ + schedule: { firstFireAt: 1 }, + action: "broken queue task", + owner: "U1", + createdBy: "U1", + ownerScopeId: scopeId("personal", "U1"), + }); + for (let i = 0; i < 20 && (!onFire || !onTick); i++) await new Promise((resolve) => setImmediate(resolve)); + + await onTick!(); + assert.deepEqual(enqueued.pop(), { cronId: cron.id, scheduledAt: 1 }); + await onFire!({ cronId: cron.id, scheduledAt: 1 }); + assert.equal(calls.length, 1); + assert.deepEqual(enqueued.pop(), { cronId: cron.id, scheduledAt: 1, notBefore: 6_000 }); + + clock = 1_001; + await onTick!(); + assert.deepEqual(enqueued.pop(), { cronId: cron.id, scheduledAt: 1, notBefore: 6_000 }); + await onFire!({ cronId: cron.id, scheduledAt: 1 }); + assert.equal(calls.length, 1, "an already-queued early job rechecks the fresh hold"); + assert.deepEqual(enqueued.pop(), { cronId: cron.id, scheduledAt: 1, notBefore: 6_000 }); + + clock = 6_000; + await onFire!({ cronId: cron.id, scheduledAt: 1 }); + assert.equal(calls.length, 2); + assert.deepEqual(enqueued.pop(), { cronId: cron.id, scheduledAt: 1, notBefore: 16_000 }); + + for (const notBefore of [36_000, 76_000, 156_000, 316_000, 616_000, 916_000]) { + clock = (await crons.get(cron.id))!.deferUntil!; + await onFire!({ cronId: cron.id, scheduledAt: 1 }); + assert.deepEqual(enqueued.pop(), { cronId: cron.id, scheduledAt: 1, notBefore }); + } + assert.deepEqual( + calls.map((call) => call.idempotencyKey), + Array(8).fill(`cron:${cron.id}:1`), + ); + scheduler.stop(); +}); diff --git a/test/cron-store.test.ts b/test/cron-store.test.ts index cccc1ba19..12a57b34b 100644 --- a/test/cron-store.test.ts +++ b/test/cron-store.test.ts @@ -387,8 +387,11 @@ test("create dedup survives a 'restart': a fresh store over the same backing sti test("claimSlot: exactly one claimant wins a slot; the claim advances the schedule", async () => { const store = createCronStore(); const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000_000 } }); - assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_500), true, "the first claim wins"); - assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_600), false, "a second claim on the same slot loses"); + await store.update(cron.id, { action: "fresh action" }); + const claim = await store.claimSlot(cron.id, 1_000_000, 1_000_500); + assert.ok(claim, "the first claim wins"); + assert.equal(claim.cron.action, "fresh action", "the claim carries the configuration read under its store lock"); + assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_600), null, "a second claim on the same slot loses"); const after = (await store.get(cron.id))!; assert.equal(after.lastFiredAt, 1_000_500); assert.equal(after.nextFireAt, 1_060_500, "the claim advances like markFired"); @@ -410,37 +413,130 @@ test("claimSlot: concurrent claimants on one slot still yield exactly one winner test("claimSlot: refuses a stale slot, a disabled cron, and an archived cron", async () => { const store = createCronStore(); const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000_000 } }); - assert.equal(await store.claimSlot(cron.id, 999, 1_000_500), false, "a slot that isn't the next fire is stale"); + assert.equal(await store.claimSlot(cron.id, 999, 1_000_500), null, "a slot that isn't the next fire is stale"); await store.setEnabled(cron.id, false); - assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_500), false, "a disabled cron cannot be claimed"); + assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_500), null, "a disabled cron cannot be claimed"); await store.setEnabled(cron.id, true); await store.update(cron.id, { archived: true }); - assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_500), false, "an archived cron cannot be claimed"); - assert.equal(await store.claimSlot("missing", 1_000_000, 1_000_500), false); + assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_500), null, "an archived cron cannot be claimed"); + assert.equal(await store.claimSlot("missing", 1_000_000, 1_000_500), null); }); test("claimSlot: a one-shot claim consumes the slot for good", async () => { const store = createCronStore(); const cron = await store.create({ ...base, schedule: { firstFireAt: 1_000_000 } }); - assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_500), true); + assert.ok(await store.claimSlot(cron.id, 1_000_000, 1_000_500)); const after = (await store.get(cron.id))!; assert.equal(after.nextFireAt, undefined, "no next fire remains"); - assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_600), false); + assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_600), null); }); -test("unclaimSlot: restores a failed claim so the slot retries, and only that exact claim", async () => { +test("releaseSlot restores a claimed slot, and only its exact claim", async () => { const store = createCronStore(); const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000_000 } }); await store.markFired(cron.id, 940_000); - const prior = (await store.get(cron.id))!.lastFiredAt; - assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_000_500), true); - await store.unclaimSlot(cron.id, 1_000_000, 1_000_500, prior); + const first = await store.claimSlot(cron.id, 1_000_000, 1_000_500); + assert.ok(first); + await store.releaseSlot(cron.id, first); const restored = (await store.get(cron.id))!; assert.equal(restored.lastFiredAt, 940_000); assert.equal(restored.nextFireAt, 1_000_000, "the slot is claimable again"); - assert.equal(await store.claimSlot(cron.id, 1_000_000, 1_001_000), true); - await store.unclaimSlot(cron.id, 1_000_000, 999, prior); - assert.equal((await store.get(cron.id))!.lastFiredAt, 1_001_000, "an unclaim for a different claim is a no-op"); + const second = await store.claimSlot(cron.id, 1_000_000, 1_001_000); + assert.ok(second); + await store.releaseSlot(cron.id, first); + assert.equal((await store.get(cron.id))!.lastFiredAt, 1_001_000, "a release for an older claim is a no-op"); +}); + +test("failed slot backoff grows to five minutes and survives store reloads", async () => { + const backing = createMemoryMap(); + let store = createCronStore(backing); + const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000 } }); + let at = 1_000; + const delays = [5_000, 10_000, 20_000, 40_000, 80_000, 160_000, 300_000, 300_000]; + for (let i = 0; i < delays.length; i++) { + const claim = await store.claimSlot(cron.id, 1_000, at); + assert.ok(claim); + assert.equal(await store.failSlot(cron.id, claim, at), at + delays[i]!); + const failed = (await store.get(cron.id))!; + assert.deepEqual(failed.failureBackoff, { scheduledAt: 1_000, failures: Math.min(i + 1, 7) }); + assert.equal(failed.nextFireAt, 1_000); + assert.equal(failed.deferUntil, at + delays[i]!); + assert.deepEqual(await store.due(failed.deferUntil - 1), []); + at = failed.deferUntil; + store = createCronStore(backing); + } +}); + +test("slot success, a new slot, and an action edit reset failure history", async () => { + const store = createCronStore(); + const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000 } }); + const failedClaim = await store.claimSlot(cron.id, 1_000, 1_000); + assert.ok(failedClaim); + await store.failSlot(cron.id, failedClaim, 1_000); + + const retryClaim = await store.claimSlot(cron.id, 1_000, 6_000); + assert.ok(retryClaim); + assert.deepEqual((await store.get(cron.id))!.failureBackoff, { scheduledAt: 1_000, failures: 1 }); + await store.completeSlot(cron.id, retryClaim); + assert.equal((await store.get(cron.id))!.failureBackoff, undefined); + + const nextSlot = (await store.get(cron.id))!.nextFireAt!; + const nextClaim = await store.claimSlot(cron.id, nextSlot, nextSlot); + assert.ok(nextClaim); + await store.failSlot(cron.id, nextClaim, nextSlot); + assert.deepEqual((await store.get(cron.id))!.failureBackoff, { scheduledAt: nextSlot, failures: 1 }); + + const editedClaim = await store.claimSlot(cron.id, nextSlot, nextSlot + 5_000); + assert.ok(editedClaim); + await store.update(cron.id, { action: "new action" }); + await store.failSlot(cron.id, editedClaim, nextSlot + 5_000); + const edited = (await store.get(cron.id))!; + assert.equal(edited.action, "new action"); + assert.equal(edited.failureBackoff, undefined); + assert.equal(edited.activeClaimId, undefined); + + const rescheduled = await store.create({ + ...base, + action: "reschedule me", + schedule: { everyMs: 60_000, firstFireAt: 2_000 }, + }); + const staleScheduleClaim = await store.claimSlot(rescheduled.id, 2_000, 2_000); + assert.ok(staleScheduleClaim); + await store.update(rescheduled.id, { schedule: { everyMs: 120_000, firstFireAt: 500_000 } }); + await store.failSlot(rescheduled.id, staleScheduleClaim, 2_000); + const afterScheduleEdit = (await store.get(rescheduled.id))!; + assert.deepEqual(afterScheduleEdit.schedule, { everyMs: 120_000, firstFireAt: 500_000 }); + assert.equal(afterScheduleEdit.nextFireAt, 500_000); + assert.equal(afterScheduleEdit.failureBackoff, undefined); +}); + +test("stale failures cannot replace a newer claim, and busy deferrals preserve error history and later holds", async () => { + const store = createCronStore(); + const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000 } }); + const first = await store.claimSlot(cron.id, 1_000, 1_000); + assert.ok(first); + await store.releaseSlot(cron.id, first); + const second = await store.claimSlot(cron.id, 1_000, 1_001); + assert.ok(second); + await store.failSlot(cron.id, first, 1_001); + assert.equal((await store.get(cron.id))!.activeClaimId, second.id); + await store.failSlot(cron.id, second, 1_001); + + const retry = await store.claimSlot(cron.id, 1_000, 6_001); + assert.ok(retry); + await store.defer(cron.id, 40_000); + await store.releaseSlot(cron.id, retry, 30_000); + let stored = (await store.get(cron.id))!; + assert.equal(stored.deferUntil, 40_000); + assert.deepEqual(stored.failureBackoff, { scheduledAt: 1_000, failures: 1 }); + assert.equal(await store.claimSlot(cron.id, 1_000, 39_999), null, "claim rechecks the fresh durable hold"); + + const afterBusy = await store.claimSlot(cron.id, 1_000, 40_000); + assert.ok(afterBusy); + await store.failSlot(cron.id, afterBusy, 40_000); + stored = (await store.get(cron.id))!; + assert.deepEqual(stored.failureBackoff, { scheduledAt: 1_000, failures: 2 }); + assert.equal(stored.deferUntil, 50_000); }); test("beginFire journals a running entry; recordFire closes the same row with the outcome", async () => { @@ -824,7 +920,7 @@ test("a deferred cron is not due until its deferral passes, and firing clears th await store.defer(cron.id, slot + 90_000); const next = await store.get(cron.id); - assert.equal(await store.claimSlot(cron.id, next!.nextFireAt!, slot + 90_000), true); + assert.ok(await store.claimSlot(cron.id, next!.nextFireAt!, slot + 90_000)); assert.equal((await store.get(cron.id))?.deferUntil, undefined, "claimSlot clears the deferral"); }); From 4c3314ff06c468bda83f297b74c423d02d2de006 Mon Sep 17 00:00:00 2001 From: Sina Matian Date: Fri, 11 Sep 2026 14:03:32 -0700 Subject: [PATCH 2/5] Test cron backoff across queue restart --- test/cron-queue.test.ts | 30 ++++++++++++++++++++---------- 1 file changed, 20 insertions(+), 10 deletions(-) diff --git a/test/cron-queue.test.ts b/test/cron-queue.test.ts index 0f3334100..2ee0bd0f7 100644 --- a/test/cron-queue.test.ts +++ b/test/cron-queue.test.ts @@ -100,14 +100,14 @@ test( { skip, timeout: 120_000 }, async () => { const calls: Array<{ req: TurnRequest; at: number }> = []; - const a = instance([], 0, undefined, async (req) => { + const failingRun = async (req: TurnRequest): Promise => { calls.push({ req, at: Date.now() }); throw new Error("provider down"); - }); - const b = instance([], 0, undefined, async (req) => { - calls.push({ req, at: Date.now() }); - throw new Error("provider down"); - }); + }; + const a = instance([], 0, undefined, failingRun); + const b = instance([], 0, undefined, failingRun); + let originalsStopped = false; + let restarted: { scheduler: Scheduler; crons: CronStore } | undefined; a.scheduler.start(1_000); b.scheduler.start(1_000); try { @@ -127,18 +127,28 @@ test( await new Promise((resolve) => setTimeout(resolve, 2_000)); assert.equal(calls.length, 1, "a sibling reconcile cannot redeliver before the durable hold"); + a.scheduler.stop(); + b.scheduler.stop(); + originalsStopped = true; + await new Promise((resolve) => setTimeout(resolve, 500)); + restarted = instance([], 0, undefined, failingRun); + restarted.scheduler.start(1_000); + await until(() => calls.length >= 2, 20_000); assert.equal(calls.length, 2); assert.ok(calls[1]!.at - calls[0]!.at >= 4_500); assert.equal(calls[1]!.req.idempotencyKey, calls[0]!.req.idempotencyKey); - await untilAsync(async () => (await a.crons.get(cron.id))?.failureBackoff?.failures === 2, 5_000); - assert.deepEqual((await a.crons.get(cron.id))?.failureBackoff, { + await untilAsync(async () => (await restarted!.crons.get(cron.id))?.failureBackoff?.failures === 2, 5_000); + assert.deepEqual((await restarted.crons.get(cron.id))?.failureBackoff, { scheduledAt: cron.nextFireAt, failures: 2, }); } finally { - a.scheduler.stop(); - b.scheduler.stop(); + if (!originalsStopped) { + a.scheduler.stop(); + b.scheduler.stop(); + } + restarted?.scheduler.stop(); await new Promise((resolve) => setTimeout(resolve, 500)); } }, From 8912c2f96377e023e1856e5410bfe3c0ee558b41 Mon Sep 17 00:00:00 2001 From: Sina Matian Date: Fri, 11 Sep 2026 14:29:17 -0700 Subject: [PATCH 3/5] Keep interval cron slots recoverable --- src/cron/cron-store.ts | 85 +++++++++++++++++++++++++++++++-- src/cron/scheduler.ts | 42 ++++++++++------- test/cron-queue.test.ts | 94 +++++++++++++++++++++++++++++++------ test/cron-scheduler.test.ts | 90 ++++++++++++++++++++++++++++++++++- test/cron-store.test.ts | 30 ++++++++++++ 5 files changed, 302 insertions(+), 39 deletions(-) diff --git a/src/cron/cron-store.ts b/src/cron/cron-store.ts index cd0a2f67a..8798c811e 100644 --- a/src/cron/cron-store.ts +++ b/src/cron/cron-store.ts @@ -61,6 +61,8 @@ const FAILURE_BACKOFF_BASE_MS = 5_000; const FAILURE_BACKOFF_MAX_MS = 5 * 60_000; const FAILURE_BACKOFF_MAX_FAILURES = 7; +export type DueCron = Cron & { scheduledAt: number }; + export interface CronSlotClaim { id: string; cron: Cron; @@ -94,7 +96,11 @@ export interface CronStore { completeSlot(id: string, claim: CronSlotClaim): Promise; releaseSlot(id: string, claim: CronSlotClaim, deferUntil?: number): Promise; failSlot(id: string, claim: CronSlotClaim, failedAt: number): Promise; - due(now: number): Promise>; + completeDueSlot(id: string, cron: DueCron, at: number): Promise; + deferDueSlot(id: string, cron: DueCron, until: number): Promise; + failDueSlot(id: string, cron: DueCron, failedAt: number): Promise; + disableDueSlot(id: string, cron: DueCron): Promise; + due(now: number): Promise; } export function isDeferred(cron: Pick, now: number): boolean { @@ -130,6 +136,40 @@ function failureBackoffMs(failures: number): number { return Math.min(FAILURE_BACKOFF_MAX_MS, FAILURE_BACKOFF_BASE_MS * 2 ** (failures - 1)); } +function nextFailureCount(cron: Cron, scheduledAt: number): number { + if (cron.failureBackoff?.scheduledAt !== scheduledAt) return 1; + const failures = cron.failureBackoff.failures; + if (!Number.isFinite(failures) || failures < 1) return 1; + return Math.min(FAILURE_BACKOFF_MAX_FAILURES, Math.floor(failures) + 1); +} + +function attemptIdentity(cron: Cron): string { + return contentPart([ + cron.schedule, + cron.title, + cron.action, + cron.message, + cron.loopId, + cron.destination, + cron.runAs, + cron.members, + cron.unattendedGrants, + cron.recipientConsent, + ]); +} + +function matchesDueSlot(cron: Cron, expected: DueCron, allowDisabledOneShot = false): boolean { + const enabled = + cron.enabled || + (allowDisabledOneShot && expected.schedule.everyMs === undefined && expected.schedule.cron === undefined); + return ( + enabled && + !cron.archived && + attemptIdentity(cron) === attemptIdentity(expected) && + recoverNextFireAt(cron.schedule, cron.createdAt, cron.lastFiredAt, cron.nextFireAt) === expected.scheduledAt + ); +} + export function createCronStore( backing: DurableMap = createMemoryMap(), opts?: { staleRunningMs?: number; fires?: CronFireStore }, @@ -323,10 +363,7 @@ export function createCronStore( await requireAtomicUpdate(backing)(id, (cron) => { deferUntil = undefined; if (cron.activeClaimId !== claim.id || cron.lastFiredAt !== claim.claimedAt) return cron; - const failures = - cron.failureBackoff?.scheduledAt === claim.scheduledAt - ? Math.min(FAILURE_BACKOFF_MAX_FAILURES, cron.failureBackoff.failures + 1) - : 1; + const failures = nextFailureCount(cron, claim.scheduledAt); deferUntil = Math.max(cron.deferUntil ?? 0, failedAt + failureBackoffMs(failures)); return mergeFields(cron, { lastFiredAt: claim.priorLastFiredAt, @@ -338,6 +375,44 @@ export function createCronStore( }); return deferUntil; }, + async completeDueSlot(id, expected, at) { + await requireAtomicUpdate(backing)(id, (cron) => { + if (!matchesDueSlot(cron, expected, true)) return cron; + const advanceFrom = isCalendarSchedule(cron.schedule) ? expected.scheduledAt : at; + return mergeFields(cron, { + lastFiredAt: at, + nextFireAt: advanceNextFireAt(cron.schedule, advanceFrom), + deferUntil: undefined, + failureBackoff: undefined, + ...(cron.schedule.everyMs === undefined && cron.schedule.cron === undefined ? { enabled: false } : {}), + }); + }); + }, + async deferDueSlot(id, expected, until) { + await requireAtomicUpdate(backing)(id, (cron) => + matchesDueSlot(cron, expected) ? { ...cron, deferUntil: Math.max(cron.deferUntil ?? 0, until) } : cron, + ); + }, + async failDueSlot(id, expected, failedAt) { + let deferUntil: number | undefined; + await requireAtomicUpdate(backing)(id, (cron) => { + deferUntil = undefined; + if (!matchesDueSlot(cron, expected)) return cron; + const failures = nextFailureCount(cron, expected.scheduledAt); + deferUntil = Math.max(cron.deferUntil ?? 0, failedAt + failureBackoffMs(failures)); + return { + ...cron, + failureBackoff: { scheduledAt: expected.scheduledAt, failures }, + deferUntil, + }; + }); + return deferUntil; + }, + async disableDueSlot(id, expected) { + await requireAtomicUpdate(backing)(id, (cron) => + matchesDueSlot(cron, expected) ? mergeFields(clearAttemptState(cron), { enabled: false }) : cron, + ); + }, async markAttempted(id, at) { await backing.merge(id, { lastAttemptAt: at }); }, diff --git a/src/cron/scheduler.ts b/src/cron/scheduler.ts index 35102067e..536e5b3fc 100644 --- a/src/cron/scheduler.ts +++ b/src/cron/scheduler.ts @@ -8,7 +8,7 @@ import { type TurnResult, } from "../types.ts"; import type { IdentityService } from "../identity/identity-service.ts"; -import { isDeferred, type CronSlotClaim, type CronStore } from "./cron-store.ts"; +import { isDeferred, type CronStore, type DueCron } from "./cron-store.ts"; import type { DeliveryStore } from "../delivery/delivery-store.ts"; import type { IdempotencyStore } from "../idempotency/idempotency-store.ts"; import { runTrigger, type TriggerDeps } from "../triggers/run-trigger.ts"; @@ -320,6 +320,7 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { } }; + const intervalInFlight = new Set(); const fireDue = async (t: number): Promise => { const due = await deps.crons.due(t); let batch = due; @@ -338,31 +339,36 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { console.warn(`[scheduler] fan-out capped: firing ${batch.length}/${due.length} due crons this tick`); } for (const cron of batch) { - let claim: CronSlotClaim | null = null; + const fireKey = `cron:${cron.id}:${cron.scheduledAt}`; + if (intervalInFlight.has(fireKey)) continue; + intervalInFlight.add(fireKey); + let admitted: DueCron | null = null; try { - claim = await deps.crons.claimSlot(cron.id, cron.scheduledAt, t); - if (!claim) continue; - const { authzFailed, deferUntil } = await fire( - claim.cron, - t, - `cron:${cron.id}:${cron.scheduledAt}`, - cron.scheduledAt, - ); - if (authzFailed || deferUntil !== undefined) { - await deps.crons.releaseSlot(cron.id, claim, deferUntil); - if (authzFailed) await deps.crons.setEnabled(cron.id, false); - } else { - await deps.crons.completeSlot(cron.id, claim); - } + const current = await deps.crons.get(cron.id); + if ( + !current || + current.archived || + !current.enabled || + isDeferred(current, t) || + nextSlot(current) !== cron.scheduledAt + ) + continue; + admitted = { ...current, scheduledAt: cron.scheduledAt }; + const { authzFailed, deferUntil } = await fire(admitted, t, fireKey, cron.scheduledAt); + if (authzFailed) await deps.crons.disableDueSlot(cron.id, admitted); + else if (deferUntil !== undefined) await deps.crons.deferDueSlot(cron.id, admitted, deferUntil); + else await deps.crons.completeDueSlot(cron.id, admitted, t); } catch (e) { console.error("[scheduler] fire failed:", errMessage(e)); - if (claim) { + if (admitted) { try { - await deps.crons.failSlot(cron.id, claim, now()); + await deps.crons.failDueSlot(cron.id, admitted, now()); } catch (transitionError) { console.error("[scheduler] failure backoff failed:", errMessage(transitionError)); } } + } finally { + intervalInFlight.delete(fireKey); } } }; diff --git a/test/cron-queue.test.ts b/test/cron-queue.test.ts index 2ee0bd0f7..cae4f9757 100644 --- a/test/cron-queue.test.ts +++ b/test/cron-queue.test.ts @@ -15,6 +15,7 @@ const skip = URL ? false : "set DATABASE_URL (a Postgres) to run the cron queue const SCHEMA = "pgboss_cron_queue_test"; const CRONS_TABLE = "cron_queue_test_crons"; +const INTERVAL_CRONS_TABLE = "cron_interval_test_crons"; const IDEM_TABLE = "cron_queue_test_idempotency"; before(async () => { @@ -23,7 +24,7 @@ before(async () => { const p = new pg.Pool({ connectionString: URL }); await p.query("DROP TABLE IF EXISTS qm_schema_migrations CASCADE"); await p.query(`DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE`); - await p.query(`DROP TABLE IF EXISTS ${CRONS_TABLE}, ${IDEM_TABLE}`); + await p.query(`DROP TABLE IF EXISTS ${CRONS_TABLE}, ${INTERVAL_CRONS_TABLE}, ${IDEM_TABLE}`); await p.end(); }); @@ -95,13 +96,70 @@ test( }, ); +test( + "Postgres interval scheduler: a thrown fire persists backoff without pre-consuming its slot", + { skip }, + async () => { + const maps = createPostgresMapFactory(URL!); + const crons = createCronStore(maps.map(INTERVAL_CRONS_TABLE)); + const calls: TurnRequest[] = []; + let clock = 1_000; + const scheduler = createScheduler({ + crons, + deliveries: createDeliveryStore(), + idempotency: createIdempotencyStore(maps.map(IDEM_TABLE)), + identity: createIdentityService(), + run: async (req) => { + calls.push(req); + throw new Error("provider down"); + }, + now: () => clock, + }); + const cron = await crons.create({ + schedule: { firstFireAt: 1 }, + action: "durable failing interval fire", + owner: "U4", + createdBy: "U4", + ownerScopeId: scopeId("personal", "U4"), + }); + + await scheduler.tick(clock); + const reloaded = createCronStore(createPostgresMapFactory(URL!).map(INTERVAL_CRONS_TABLE)); + let stored = (await reloaded.get(cron.id))!; + assert.equal(stored.lastFiredAt, undefined); + assert.equal(stored.nextFireAt, 1); + assert.equal(stored.deferUntil, 6_000); + assert.deepEqual(stored.failureBackoff, { scheduledAt: 1, failures: 1 }); + + clock = 5_999; + await scheduler.tick(clock); + assert.equal(calls.length, 1); + clock = 6_000; + await scheduler.tick(clock); + stored = (await reloaded.get(cron.id))!; + assert.equal(calls.length, 2); + assert.equal(stored.nextFireAt, 1); + assert.equal(stored.deferUntil, 16_000); + assert.deepEqual(stored.failureBackoff, { scheduledAt: 1, failures: 2 }); + assert.equal(calls[1]!.idempotencyKey, calls[0]!.idempotencyKey); + }, +); + test( "pg-boss queue: thrown failures persist a cooldown across instances without changing the fire key", { skip, timeout: 120_000 }, async () => { const calls: Array<{ req: TurnRequest; at: number }> = []; + let signalFirstStarted!: () => void; + let releaseFirstFailure!: () => void; + const firstStarted = new Promise((resolve) => (signalFirstStarted = resolve)); + const firstFailureReleased = new Promise((resolve) => (releaseFirstFailure = resolve)); const failingRun = async (req: TurnRequest): Promise => { calls.push({ req, at: Date.now() }); + if (calls.length === 1) { + signalFirstStarted(); + await firstFailureReleased; + } throw new Error("provider down"); }; const a = instance([], 0, undefined, failingRun); @@ -118,30 +176,36 @@ test( createdBy: "U3", ownerScopeId: scopeId("personal", "U3"), }); - await until(() => calls.length >= 1, 30_000); + await firstStarted; assert.equal(calls.length, 1); - await untilAsync(async () => (await b.crons.get(cron.id))?.failureBackoff?.failures === 1, 5_000); - const persisted = await b.crons.get(cron.id); - assert.deepEqual(persisted?.failureBackoff, { scheduledAt: cron.nextFireAt, failures: 1 }); - assert.ok((persisted?.deferUntil ?? 0) >= calls[0]!.at + 5_000); - await new Promise((resolve) => setTimeout(resolve, 2_000)); - assert.equal(calls.length, 1, "a sibling reconcile cannot redeliver before the durable hold"); - + releaseFirstFailure(); + await untilAsync(async () => (await b.crons.get(cron.id))?.failureBackoff !== undefined, 5_000); a.scheduler.stop(); b.scheduler.stop(); originalsStopped = true; await new Promise((resolve) => setTimeout(resolve, 500)); + + const persisted = (await b.crons.get(cron.id))!; + assert.equal(persisted.failureBackoff?.scheduledAt, cron.nextFireAt); + assert.ok((persisted.failureBackoff?.failures ?? 0) >= 1); + assert.ok((persisted.deferUntil ?? 0) >= calls[0]!.at + 5_000); + const callsBeforeRestart = calls.length; + const failuresBeforeRestart = persisted.failureBackoff!.failures; restarted = instance([], 0, undefined, failingRun); restarted.scheduler.start(1_000); - await until(() => calls.length >= 2, 20_000); - assert.equal(calls.length, 2); - assert.ok(calls[1]!.at - calls[0]!.at >= 4_500); - assert.equal(calls[1]!.req.idempotencyKey, calls[0]!.req.idempotencyKey); - await untilAsync(async () => (await restarted!.crons.get(cron.id))?.failureBackoff?.failures === 2, 5_000); + await until(() => calls.length > callsBeforeRestart, 20_000); + assert.equal(calls.length, callsBeforeRestart + 1); + const restartedCall = calls.at(-1)!; + assert.ok(restartedCall.at >= persisted.deferUntil! - 100); + assert.equal(restartedCall.req.idempotencyKey, calls[0]!.req.idempotencyKey); + await untilAsync( + async () => (await restarted!.crons.get(cron.id))?.failureBackoff?.failures === failuresBeforeRestart + 1, + 5_000, + ); assert.deepEqual((await restarted.crons.get(cron.id))?.failureBackoff, { scheduledAt: cron.nextFireAt, - failures: 2, + failures: failuresBeforeRestart + 1, }); } finally { if (!originalsStopped) { diff --git a/test/cron-scheduler.test.ts b/test/cron-scheduler.test.ts index 0a2ef1fbc..7e47d43da 100644 --- a/test/cron-scheduler.test.ts +++ b/test/cron-scheduler.test.ts @@ -2,7 +2,7 @@ import { test } from "node:test"; import assert from "node:assert/strict"; import { createScheduler } from "../src/cron/scheduler.ts"; import { runNowSettled } from "./support/settle.ts"; -import { createCronStore } from "../src/cron/cron-store.ts"; +import { createCronStore, type CronStore } from "../src/cron/cron-store.ts"; import { createDeliveryStore } from "../src/delivery/delivery-store.ts"; import { createIdempotencyStore } from "../src/idempotency/idempotency-store.ts"; import { createIdentityService } from "../src/identity/identity-service.ts"; @@ -1342,6 +1342,94 @@ test("interval mode backs off thrown failures without delaying healthy crons", a ); }); +test("an in-flight interval fire leaves its durable slot recoverable and overlapping ticks do not duplicate it", async () => { + const backing = createMemoryMap(); + const crons = createCronStore(backing); + let started!: () => void; + let finish!: () => void; + const runStarted = new Promise((resolve) => (started = resolve)); + const runFinished = new Promise((resolve) => (finish = resolve)); + let calls = 0; + const scheduler = createScheduler({ + crons, + deliveries: createDeliveryStore(), + idempotency: createIdempotencyStore(), + identity: createIdentityService(), + run: async () => { + calls++; + started(); + await runFinished; + return { status: "ok", reply: "done" }; + }, + now: () => 1_000, + }); + const cron = await crons.create({ + schedule: { firstFireAt: 1_000 }, + action: "slow task", + owner: "U1", + createdBy: "U1", + ownerScopeId: scopeId("personal", "U1"), + }); + + const firstTick = scheduler.tick(1_000); + await runStarted; + await scheduler.tick(1_001); + assert.equal(calls, 1); + const reloaded = createCronStore(backing); + const stored = (await reloaded.get(cron.id))!; + assert.equal(stored.lastFiredAt, undefined); + assert.equal(stored.nextFireAt, 1_000); + assert.deepEqual( + (await reloaded.due(1_001)).map((due) => due.id), + [cron.id], + ); + + finish(); + await firstTick; + const completed = (await reloaded.get(cron.id))!; + assert.equal(completed.lastFiredAt, 1_000); + assert.equal(completed.enabled, false); +}); + +test("a failed interval backoff write leaves the original slot recoverable after reload", async () => { + const backing = createMemoryMap(); + const baseCrons = createCronStore(backing); + const crons: CronStore = { + ...baseCrons, + async failDueSlot() { + throw new Error("backoff store unavailable"); + }, + }; + const scheduler = createScheduler({ + crons, + deliveries: createDeliveryStore(), + idempotency: createIdempotencyStore(), + identity: createIdentityService(), + run: async () => { + throw new Error("provider down"); + }, + now: () => 1_000, + }); + const cron = await crons.create({ + schedule: { firstFireAt: 1_000 }, + action: "failing task", + owner: "U1", + createdBy: "U1", + ownerScopeId: scopeId("personal", "U1"), + }); + + await scheduler.tick(1_000); + const reloaded = createCronStore(backing); + const stored = (await reloaded.get(cron.id))!; + assert.equal(stored.lastFiredAt, undefined); + assert.equal(stored.nextFireAt, 1_000); + assert.equal(stored.failureBackoff, undefined); + assert.deepEqual( + (await reloaded.due(1_001)).map((due) => due.id), + [cron.id], + ); +}); + function busyOnce(calls: TurnRequest[]) { return async (req: TurnRequest): Promise => { calls.push(req); diff --git a/test/cron-store.test.ts b/test/cron-store.test.ts index 12a57b34b..ec927a814 100644 --- a/test/cron-store.test.ts +++ b/test/cron-store.test.ts @@ -467,6 +467,17 @@ test("failed slot backoff grows to five minutes and survives store reloads", asy } }); +test("malformed persisted failure counts restart at the first bounded delay", async () => { + const backing = createMemoryMap(); + const store = createCronStore(backing); + const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000 } }); + await backing.merge(cron.id, { failureBackoff: { scheduledAt: 1_000, failures: Number.NaN } }); + const claim = await store.claimSlot(cron.id, 1_000, 1_000); + assert.ok(claim); + assert.equal(await store.failSlot(cron.id, claim, 1_000), 6_000); + assert.deepEqual((await store.get(cron.id))!.failureBackoff, { scheduledAt: 1_000, failures: 1 }); +}); + test("slot success, a new slot, and an action edit reset failure history", async () => { const store = createCronStore(); const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000 } }); @@ -510,6 +521,25 @@ test("slot success, a new slot, and an action edit reset failure history", async assert.equal(afterScheduleEdit.failureBackoff, undefined); }); +test("interval slot transitions ignore stale action and schedule snapshots", async () => { + const store = createCronStore(); + const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000 } }); + const [stale] = await store.due(1_000); + assert.ok(stale); + await store.update(cron.id, { action: "edited action" }); + await store.failDueSlot(cron.id, stale, 1_000); + assert.equal((await store.get(cron.id))!.failureBackoff, undefined); + + const [beforeScheduleEdit] = await store.due(1_000); + assert.ok(beforeScheduleEdit); + await store.update(cron.id, { schedule: { everyMs: 120_000, firstFireAt: 500_000 } }); + await store.completeDueSlot(cron.id, beforeScheduleEdit, 1_000); + const edited = (await store.get(cron.id))!; + assert.deepEqual(edited.schedule, { everyMs: 120_000, firstFireAt: 500_000 }); + assert.equal(edited.nextFireAt, 500_000); + assert.equal(edited.lastFiredAt, undefined); +}); + test("stale failures cannot replace a newer claim, and busy deferrals preserve error history and later holds", async () => { const store = createCronStore(); const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000 } }); From 24fa3e0efa999d0db41d8a4ba3d2cdb2404d336f Mon Sep 17 00:00:00 2001 From: Sina Matian Date: Fri, 11 Sep 2026 14:58:36 -0700 Subject: [PATCH 4/5] Fence cron retries across edits --- src/cron/cron-store.ts | 119 +++++++++++++++++++++--------- src/cron/scheduler.ts | 15 ++-- src/types.ts | 2 + test/cron-scheduler.test.ts | 74 +++++++++++++++++++ test/cron-store.test.ts | 140 ++++++++++++++++++++++++++++++++++++ test/postgres-map.test.ts | 49 ++++++++++++- 6 files changed, 355 insertions(+), 44 deletions(-) diff --git a/src/cron/cron-store.ts b/src/cron/cron-store.ts index 8798c811e..9261f6197 100644 --- a/src/cron/cron-store.ts +++ b/src/cron/cron-store.ts @@ -68,6 +68,7 @@ export interface CronSlotClaim { cron: Cron; scheduledAt: number; claimedAt: number; + executionRevision: number; priorLastFiredAt?: number; } @@ -122,8 +123,13 @@ function mergeFields(cron: Cron, fields: Partial): Cron { return next; } +function clearFailureState(cron: Cron): Cron { + const { failureBackoff: _failureBackoff, ...rest } = cron; + return { ...rest, failureGeneration: 0 }; +} + function clearAttemptState(cron: Cron): Cron { - const { activeClaimId: _activeClaimId, failureBackoff: _failureBackoff, ...rest } = cron; + const { activeClaimId: _activeClaimId, ...rest } = clearFailureState(cron); return rest; } @@ -143,29 +149,29 @@ function nextFailureCount(cron: Cron, scheduledAt: number): number { return Math.min(FAILURE_BACKOFF_MAX_FAILURES, Math.floor(failures) + 1); } -function attemptIdentity(cron: Cron): string { - return contentPart([ - cron.schedule, - cron.title, - cron.action, - cron.message, - cron.loopId, - cron.destination, - cron.runAs, - cron.members, - cron.unattendedGrants, - cron.recipientConsent, - ]); +function generation(value: number | undefined): number { + if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 0 || value === Number.MAX_SAFE_INTEGER) + return 0; + return value; +} + +function nextGeneration(value: number | undefined): number { + return generation(value) + 1; +} + +function sameFailureState(cron: Cron, expected: Cron): boolean { + return ( + generation(cron.failureGeneration) === generation(expected.failureGeneration) && + contentPart(cron.failureBackoff) === contentPart(expected.failureBackoff) + ); } -function matchesDueSlot(cron: Cron, expected: DueCron, allowDisabledOneShot = false): boolean { - const enabled = - cron.enabled || - (allowDisabledOneShot && expected.schedule.everyMs === undefined && expected.schedule.cron === undefined); +function matchesDueSlot(cron: Cron, expected: DueCron): boolean { return ( - enabled && + cron.enabled && !cron.archived && - attemptIdentity(cron) === attemptIdentity(expected) && + generation(cron.executionRevision) === generation(expected.executionRevision) && + sameFailureState(cron, expected) && recoverNextFireAt(cron.schedule, cron.createdAt, cron.lastFiredAt, cron.nextFireAt) === expected.scheduledAt ); } @@ -176,14 +182,32 @@ export function createCronStore( ): CronStore { const staleRunningMs = opts?.staleRunningMs ?? DEFAULT_FIRE_RUNNING_STALE_MS; const fires = opts?.fires ?? createMemoryCronFireStore(); - const updateCron = (id: string, fields: Partial, resetAttempt = true): Promise => { - if (backing.update) - return backing.update(id, (cron) => mergeFields(resetAttempt ? clearAttemptState(cron) : cron, fields)); - return backing.merge( - id, - resetAttempt ? { ...fields, activeClaimId: undefined, failureBackoff: undefined } : fields, - ); - }; + const updateCron = ( + id: string, + fields: Partial, + kind: "execution" | "schedule" = "execution", + ): Promise => + requireAtomicUpdate(backing)(id, (cron) => { + const effectiveFields = { ...fields }; + let effectiveKind = kind; + if ( + kind === "schedule" && + fields.schedule !== undefined && + contentPart(fields.schedule) === contentPart(cron.schedule) + ) { + delete effectiveFields.schedule; + delete effectiveFields.nextFireAt; + effectiveKind = "execution"; + } + const updated = mergeFields(cron, effectiveFields); + const unchanged = Object.keys(effectiveFields).every((key) => { + if (key === "archived" && !cron.archived && updated.archived === false) return true; + return contentPart(cron[key as keyof Cron]) === contentPart(updated[key as keyof Cron]); + }); + if (unchanged) return cron; + const revised = { ...updated, executionRevision: nextGeneration(cron.executionRevision) }; + return effectiveKind === "schedule" ? clearAttemptState(revised) : clearFailureState(revised); + }); return { async create(input) { assertNoEscalation(input); @@ -206,6 +230,8 @@ export function createCronStore( return createDeduped(backing, contentId, (id) => ({ ...buildTriggerBase(input, id, now), schedule, + executionRevision: 0, + failureGeneration: 0, ...(nextFireAt !== undefined ? { nextFireAt } : {}), ...(title ? { title } : {}), ...(input.action !== undefined ? { action: input.action } : {}), @@ -235,7 +261,7 @@ export function createCronStore( if (patch.members !== undefined) fields.members = patch.members; if (patch.runAs !== undefined) fields.runAs = patch.runAs; if (patch.unattendedGrants !== undefined) fields.unattendedGrants = patch.unattendedGrants; - return updateCron(id, fields, Object.keys(fields).length > 0); + return updateCron(id, fields, patch.schedule !== undefined ? "schedule" : "execution"); }, delete: (id) => backing.delete(id), async setEnabled(id, enabled) { @@ -311,6 +337,7 @@ export function createCronStore( deferUntil: undefined, activeClaimId: undefined, failureBackoff: undefined, + failureGeneration: 0, }); }, async claimSlot(id, scheduledAt, at) { @@ -326,17 +353,19 @@ export function createCronStore( cron, scheduledAt, claimedAt: at, + executionRevision: generation(cron.executionRevision), ...(cron.lastFiredAt !== undefined ? { priorLastFiredAt: cron.lastFiredAt } : {}), }; const advanceFrom = isCalendarSchedule(cron.schedule) ? scheduledAt : at; const nextFireAt = advanceNextFireAt(cron.schedule, advanceFrom); - const failureBackoff = cron.failureBackoff?.scheduledAt === scheduledAt ? cron.failureBackoff : undefined; + const sameFailureSlot = cron.failureBackoff?.scheduledAt === scheduledAt; return mergeFields(cron, { lastFiredAt: at, nextFireAt, deferUntil: undefined, activeClaimId: claimId, - failureBackoff, + failureBackoff: sameFailureSlot ? cron.failureBackoff : undefined, + failureGeneration: sameFailureSlot ? generation(cron.failureGeneration) : 0, }); }); return claim; @@ -344,17 +373,27 @@ export function createCronStore( async completeSlot(id, claim) { await requireAtomicUpdate(backing)(id, (cron) => { if (cron.activeClaimId !== claim.id) return cron; - return mergeFields(cron, { activeClaimId: undefined, failureBackoff: undefined }); + return mergeFields(cron, { + activeClaimId: undefined, + failureBackoff: undefined, + failureGeneration: 0, + ...(claim.cron.schedule.everyMs === undefined && claim.cron.schedule.cron === undefined + ? { enabled: false } + : {}), + }); }); }, async releaseSlot(id, claim, deferUntil) { await requireAtomicUpdate(backing)(id, (cron) => { if (cron.activeClaimId !== claim.id || cron.lastFiredAt !== claim.claimedAt) return cron; + const configurationChanged = generation(cron.executionRevision) !== claim.executionRevision; return mergeFields(cron, { lastFiredAt: claim.priorLastFiredAt, nextFireAt: claim.scheduledAt, activeClaimId: undefined, - ...(deferUntil !== undefined ? { deferUntil: Math.max(cron.deferUntil ?? 0, deferUntil) } : {}), + ...(!configurationChanged && deferUntil !== undefined + ? { deferUntil: Math.max(cron.deferUntil ?? 0, deferUntil) } + : {}), }); }); }, @@ -363,6 +402,13 @@ export function createCronStore( await requireAtomicUpdate(backing)(id, (cron) => { deferUntil = undefined; if (cron.activeClaimId !== claim.id || cron.lastFiredAt !== claim.claimedAt) return cron; + if (generation(cron.executionRevision) !== claim.executionRevision) { + return mergeFields(cron, { + lastFiredAt: claim.priorLastFiredAt, + nextFireAt: claim.scheduledAt, + activeClaimId: undefined, + }); + } const failures = nextFailureCount(cron, claim.scheduledAt); deferUntil = Math.max(cron.deferUntil ?? 0, failedAt + failureBackoffMs(failures)); return mergeFields(cron, { @@ -370,6 +416,7 @@ export function createCronStore( nextFireAt: claim.scheduledAt, activeClaimId: undefined, failureBackoff: { scheduledAt: claim.scheduledAt, failures }, + failureGeneration: nextGeneration(cron.failureGeneration), deferUntil, }); }); @@ -377,13 +424,14 @@ export function createCronStore( }, async completeDueSlot(id, expected, at) { await requireAtomicUpdate(backing)(id, (cron) => { - if (!matchesDueSlot(cron, expected, true)) return cron; + if (!matchesDueSlot(cron, expected)) return cron; const advanceFrom = isCalendarSchedule(cron.schedule) ? expected.scheduledAt : at; return mergeFields(cron, { lastFiredAt: at, nextFireAt: advanceNextFireAt(cron.schedule, advanceFrom), - deferUntil: undefined, + deferUntil: cron.deferUntil === expected.deferUntil ? undefined : cron.deferUntil, failureBackoff: undefined, + failureGeneration: 0, ...(cron.schedule.everyMs === undefined && cron.schedule.cron === undefined ? { enabled: false } : {}), }); }); @@ -403,6 +451,7 @@ export function createCronStore( return { ...cron, failureBackoff: { scheduledAt: expected.scheduledAt, failures }, + failureGeneration: nextGeneration(cron.failureGeneration), deferUntil, }; }); diff --git a/src/cron/scheduler.ts b/src/cron/scheduler.ts index 536e5b3fc..a1786a1c5 100644 --- a/src/cron/scheduler.ts +++ b/src/cron/scheduler.ts @@ -31,7 +31,7 @@ const FIRE_GC_INTERVAL_MS = 6 * 60 * 60_000; const BUSY_DEFER_MS = 30_000; const BUSY_DEFER_MAX_LATE_MS = 10 * 60_000; -type FireResult = { authzFailed: boolean; deferUntil?: number }; +type FireResult = { authzFailed: boolean; deferUntil?: number; disable?: boolean }; type RunNowResult = | { started: true; fireKey: string; settled: Promise } @@ -216,8 +216,7 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { status: result.status ?? "ok", ...(result.note ? { note: truncate(result.note, CRON_FIRE_REPLY_MAX_CHARS) } : {}), }); - if (isOneShotSchedule(cron.schedule)) await deps.crons.setEnabled(cron.id, false); - return { authzFailed: false }; + return { authzFailed: false, ...(isOneShotSchedule(cron.schedule) ? { disable: true } : {}) }; } const mentionRoster = await cronMentionRoster(deps, cron).catch(() => undefined); let outcome: Awaited>; @@ -291,9 +290,8 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { ...(outcome.sessionId ? { sessionId: outcome.sessionId } : {}), }); } - if (outcome.authzFailed) return { authzFailed: true }; - if (isOneShotSchedule(cron.schedule)) await deps.crons.setEnabled(cron.id, false); - return { authzFailed: false }; + if (outcome.authzFailed) return { authzFailed: true, disable: true }; + return { authzFailed: false, ...(isOneShotSchedule(cron.schedule) ? { disable: true } : {}) }; } let lastStrandedSweep = 0; @@ -439,6 +437,7 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { console.error("[scheduler] fire failed:", errMessage(e)); const deferUntil = await deps.crons.failSlot(job.cronId, claim, now()); if (deferUntil !== undefined) await deps.jobQueue!.enqueueFire({ ...job, notBefore: deferUntil }); + else await enqueueNext(job.cronId); return; } await enqueueNext(job.cronId); @@ -492,8 +491,8 @@ export function createScheduler(deps: SchedulerDeps): Scheduler { : { started: false, reason: "unavailable" }; } const settled = fire(cron, t, fireKey).then( - async ({ authzFailed }) => { - if (authzFailed) await deps.crons.setEnabled(cronId, false); + async ({ disable }) => { + if (disable) await deps.crons.setEnabled(cronId, false); }, (e: unknown) => console.error("%s", `[scheduler] manual fire of cron ${cronId} failed:`, errMessage(e)), ); diff --git a/src/types.ts b/src/types.ts index b4e479837..5a0ca7b82 100644 --- a/src/types.ts +++ b/src/types.ts @@ -246,6 +246,8 @@ export interface Cron extends TriggerBase { lastAttemptAt?: number; deferUntil?: number; failureBackoff?: CronFailureBackoff; + failureGeneration?: number; + executionRevision?: number; activeClaimId?: string; title?: string; archived?: boolean; diff --git a/test/cron-scheduler.test.ts b/test/cron-scheduler.test.ts index 7e47d43da..92516e893 100644 --- a/test/cron-scheduler.test.ts +++ b/test/cron-scheduler.test.ts @@ -1024,6 +1024,80 @@ test("queue mode: fires claim the slot before running, and stale or lost claims scheduler.stop(); }); +test("queue mode: one-shot failures and busy releases survive in-flight edits", async () => { + for (const transition of ["failure", "busy"] as const) { + for (const edit of ["title", "action", "idempotent-enable"] as const) { + const crons = createCronStore(); + const calls: TurnRequest[] = []; + const enqueued: Array<{ cronId: string; scheduledAt: number; notBefore?: number }> = []; + let onFire: ((job: { cronId: string; scheduledAt: number; notBefore?: number }) => Promise) | undefined; + let signalStarted!: () => void; + let release!: () => void; + const started = new Promise((resolve) => (signalStarted = resolve)); + const released = new Promise((resolve) => (release = resolve)); + const scheduler = createScheduler({ + crons, + deliveries: createDeliveryStore(), + idempotency: createIdempotencyStore(), + identity: createIdentityService(), + now: () => 1_000, + run: async (req) => { + calls.push(req); + signalStarted(); + await released; + if (transition === "failure") throw new Error("provider down"); + return { status: "refused", refusalKind: "session_busy", reason: "busy" }; + }, + jobQueue: { + async start(handlers) { + onFire = handlers.onFire; + }, + async enqueueFire(job) { + enqueued.push(job); + }, + healthy: () => true, + async stop() {}, + }, + }); + scheduler.start(1_000); + for (let i = 0; i < 20 && !onFire; i++) await new Promise((resolve) => setImmediate(resolve)); + const cron = await crons.create({ + schedule: { firstFireAt: 1 }, + action: "original action", + owner: "U1", + createdBy: "U1", + ownerScopeId: scopeId("personal", "U1"), + }); + enqueued.length = 0; + const firing = onFire!({ cronId: cron.id, scheduledAt: 1 }); + await started; + if (edit === "title") await crons.update(cron.id, { title: "edited title" }); + else if (edit === "action") await crons.update(cron.id, { action: "edited action" }); + else await crons.setEnabled(cron.id, true); + release(); + await firing; + + const stored = (await crons.get(cron.id))!; + const changed = edit !== "idempotent-enable"; + assert.equal(stored.enabled, true, `${transition}/${edit}: one-shot remains enabled`); + assert.equal(stored.lastFiredAt, undefined, `${transition}/${edit}: claim cursor is restored`); + assert.equal(stored.nextFireAt, 1, `${transition}/${edit}: original slot is restored`); + assert.equal(stored.activeClaimId, undefined); + if (changed) { + assert.equal(stored.deferUntil, undefined, `${transition}/${edit}: stale cooldown is discarded`); + assert.equal(stored.failureBackoff, undefined); + assert.deepEqual(enqueued.at(-1), { cronId: cron.id, scheduledAt: 1 }); + } else { + const notBefore = transition === "failure" ? 6_000 : 31_000; + assert.equal(stored.deferUntil, notBefore); + assert.deepEqual(enqueued.at(-1), { cronId: cron.id, scheduledAt: 1, notBefore }); + } + assert.equal(calls.length, 1); + scheduler.stop(); + } + } +}); + test("queue mode: while the queue runs, the interval scheduler's leader lease is held as a guard", async (t) => { t.mock.timers.enable({ apis: ["setInterval"] }); const heldKeys: string[] = []; diff --git a/test/cron-store.test.ts b/test/cron-store.test.ts index ec927a814..d36f8b45f 100644 --- a/test/cron-store.test.ts +++ b/test/cron-store.test.ts @@ -500,11 +500,14 @@ test("slot success, a new slot, and an action edit reset failure history", async const editedClaim = await store.claimSlot(cron.id, nextSlot, nextSlot + 5_000); assert.ok(editedClaim); await store.update(cron.id, { action: "new action" }); + assert.equal((await store.get(cron.id))!.activeClaimId, editedClaim.id); await store.failSlot(cron.id, editedClaim, nextSlot + 5_000); const edited = (await store.get(cron.id))!; assert.equal(edited.action, "new action"); assert.equal(edited.failureBackoff, undefined); + assert.equal(edited.deferUntil, undefined); assert.equal(edited.activeClaimId, undefined); + assert.equal(edited.nextFireAt, nextSlot); const rescheduled = await store.create({ ...base, @@ -521,6 +524,84 @@ test("slot success, a new slot, and an action edit reset failure history", async assert.equal(afterScheduleEdit.failureBackoff, undefined); }); +test("one-shot queue claims survive failure and busy release after non-schedule edits", async () => { + for (const transition of ["failure", "busy"] as const) { + for (const edit of ["title", "action", "idempotent-enable"] as const) { + const backing = createMemoryMap(); + const store = createCronStore(backing); + const cron = await store.create({ ...base, schedule: { firstFireAt: 1_000 } }); + await backing.merge(cron.id, { + failureBackoff: { scheduledAt: 1_000, failures: 2 }, + failureGeneration: 2, + }); + const claim = await store.claimSlot(cron.id, 1_000, 1_000); + assert.ok(claim); + if (edit === "title") await store.update(cron.id, { title: "edited title" }); + else if (edit === "action") await store.update(cron.id, { action: "edited action" }); + else await store.setEnabled(cron.id, true); + assert.equal((await store.get(cron.id))!.activeClaimId, claim.id); + + const until = + transition === "failure" + ? await store.failSlot(cron.id, claim, 1_000) + : await store.releaseSlot(cron.id, claim, 31_000); + const stored = (await store.get(cron.id))!; + const changed = edit !== "idempotent-enable"; + assert.equal(stored.enabled, true, `${transition}/${edit}: the one-shot remains enabled`); + assert.equal(stored.lastFiredAt, undefined, `${transition}/${edit}: the prior cursor is restored`); + assert.equal(stored.nextFireAt, 1_000, `${transition}/${edit}: the original slot is restored`); + assert.equal(stored.activeClaimId, undefined); + if (changed) { + assert.equal(until, undefined, `${transition}/${edit}: an obsolete outcome adds no cooldown`); + assert.equal(stored.deferUntil, undefined); + assert.equal(stored.failureBackoff, undefined); + } else if (transition === "failure") { + assert.equal(until, 21_000); + assert.equal(stored.deferUntil, 21_000); + assert.deepEqual(stored.failureBackoff, { scheduledAt: 1_000, failures: 3 }); + } else { + assert.equal(until, undefined); + assert.equal(stored.deferUntil, 31_000); + assert.deepEqual(stored.failureBackoff, { scheduledAt: 1_000, failures: 2 }); + } + } + } +}); + +test("no-op cron setters preserve claim ownership, failure state, and execution revision", async () => { + const backing = createMemoryMap(); + const store = createCronStore(backing); + const consent = { recipientId: "U2", status: "accepted" as const, decidedAt: 50 }; + const destination = { type: "slack" as const, target: "C1" }; + const cron = await store.create({ + ...base, + title: "same title", + destination, + recipientConsent: consent, + schedule: { firstFireAt: 1_000 }, + }); + await backing.merge(cron.id, { + failureBackoff: { scheduledAt: 1_000, failures: 2 }, + failureGeneration: 2, + }); + const claim = await store.claimSlot(cron.id, 1_000, 1_000); + assert.ok(claim); + const before = (await store.get(cron.id))!; + + await store.setEnabled(cron.id, true); + await store.setDestination(cron.id, destination); + await store.setRecipientConsent(cron.id, consent); + await store.update(cron.id, { title: "same title", action: "x", schedule: { firstFireAt: 1_000 } }); + await store.markAttempted(cron.id, 2_000); + await store.setFireNote(cron.id, { text: "bookkeeping", at: 2_000 }); + + const after = (await store.get(cron.id))!; + assert.equal(after.activeClaimId, before.activeClaimId); + assert.equal(after.executionRevision, before.executionRevision); + assert.equal(after.failureGeneration, before.failureGeneration); + assert.deepEqual(after.failureBackoff, before.failureBackoff); +}); + test("interval slot transitions ignore stale action and schedule snapshots", async () => { const store = createCronStore(); const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000 } }); @@ -540,6 +621,65 @@ test("interval slot transitions ignore stale action and schedule snapshots", asy assert.equal(edited.lastFiredAt, undefined); }); +test("interval outcome CAS rejects duplicate failures and ABA configuration edits", async () => { + const store = createCronStore(); + const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000 } }); + const [admitted] = await store.due(1_000); + assert.ok(admitted); + assert.equal(await store.failDueSlot(cron.id, admitted, 1_000), 6_000); + assert.equal(await store.failDueSlot(cron.id, admitted, 1_000), undefined); + let stored = (await store.get(cron.id))!; + assert.deepEqual(stored.failureBackoff, { scheduledAt: 1_000, failures: 1 }); + assert.equal(stored.failureGeneration, 1); + assert.equal(stored.deferUntil, 6_000); + + const [retry] = await store.due(6_000); + assert.ok(retry); + await store.update(cron.id, { action: "B" }); + await store.update(cron.id, { action: "x" }); + assert.equal((await store.get(cron.id))!.executionRevision, 2); + await store.failDueSlot(cron.id, retry, 6_000); + await store.completeDueSlot(cron.id, retry, 6_000); + stored = (await store.get(cron.id))!; + assert.equal(stored.action, "x"); + assert.equal(stored.lastFiredAt, undefined); + assert.equal(stored.nextFireAt, 1_000); + assert.equal(stored.failureBackoff, undefined); +}); + +test("interval success preserves a newer busy hold and cannot clear a newer failure generation", async () => { + const store = createCronStore(); + const held = await store.create({ + ...base, + action: "held success", + schedule: { everyMs: 60_000, firstFireAt: 1_000 }, + }); + const [heldAdmission] = await store.due(1_000); + assert.ok(heldAdmission); + await store.deferDueSlot(held.id, heldAdmission, 40_000); + await store.completeDueSlot(held.id, heldAdmission, 1_000); + const completed = (await store.get(held.id))!; + assert.equal(completed.lastFiredAt, 1_000); + assert.equal(completed.nextFireAt, 61_000); + assert.equal(completed.deferUntil, 40_000); + + const failed = await store.create({ + ...base, + action: "failed sibling", + schedule: { everyMs: 60_000, firstFireAt: 1_000 }, + }); + const failureAdmission = (await store.due(1_000)).find((due) => due.id === failed.id); + assert.ok(failureAdmission); + await store.failDueSlot(failed.id, failureAdmission, 1_000); + await store.completeDueSlot(failed.id, failureAdmission, 1_000); + const preserved = (await store.get(failed.id))!; + assert.equal(preserved.lastFiredAt, undefined); + assert.equal(preserved.nextFireAt, 1_000); + assert.deepEqual(preserved.failureBackoff, { scheduledAt: 1_000, failures: 1 }); + assert.equal(preserved.failureGeneration, 1); + assert.equal(preserved.deferUntil, 6_000); +}); + test("stale failures cannot replace a newer claim, and busy deferrals preserve error history and later holds", async () => { const store = createCronStore(); const cron = await store.create({ ...base, schedule: { everyMs: 60_000, firstFireAt: 1_000 } }); diff --git a/test/postgres-map.test.ts b/test/postgres-map.test.ts index ea8bc169e..140f319cb 100644 --- a/test/postgres-map.test.ts +++ b/test/postgres-map.test.ts @@ -21,7 +21,7 @@ before(async () => { const p = new pg.Pool({ connectionString: URL }); await p.query("DROP TABLE IF EXISTS qm_schema_migrations CASCADE"); await p.query( - "DROP TABLE IF EXISTS map_widgets, map_crons, map_keychain_creds, map_keychain_grants, map_keychain_asks, process_sessions, durable_map_versions CASCADE", + "DROP TABLE IF EXISTS map_widgets, map_crons, map_cron_races, map_keychain_creds, map_keychain_grants, map_keychain_asks, process_sessions, durable_map_versions CASCADE", ); await p.end(); }); @@ -84,6 +84,53 @@ test("pg map: an artifact store rides the map (a cron round-trips through Postgr assert.equal(got?.lastFiredAt, 123); }); +test("pg map: cron claim recovery and interval failure CAS survive independent store instances", { skip }, async () => { + const map = () => createPostgresMapFactory(URL!).map("map_cron_races"); + const first = createCronStore(map()); + const second = createCronStore(map()); + const cron = await first.create({ + schedule: { firstFireAt: 1_000 }, + action: "A", + ownerScopeId: scopeId("personal", "U1"), + owner: "U1", + createdBy: "U1", + }); + const claim = await first.claimSlot(cron.id, 1_000, 1_000); + assert.ok(claim); + await second.update(cron.id, { action: "B" }); + assert.equal(await first.failSlot(cron.id, claim, 1_000), undefined); + + const reloaded = createCronStore(map()); + let stored = (await reloaded.get(cron.id))!; + assert.equal(stored.action, "B"); + assert.equal(stored.lastFiredAt, undefined); + assert.equal(stored.nextFireAt, 1_000); + assert.equal(stored.deferUntil, undefined); + const [admitted] = await reloaded.due(1_000); + assert.ok(admitted); + + const results = await Promise.all([ + first.failDueSlot(cron.id, admitted, 1_000), + second.failDueSlot(cron.id, admitted, 1_000), + ]); + assert.equal(results.filter((result) => result === 6_000).length, 1); + assert.equal(results.filter((result) => result === undefined).length, 1); + stored = (await createCronStore(map()).get(cron.id))!; + assert.deepEqual(stored.failureBackoff, { scheduledAt: 1_000, failures: 1 }); + assert.equal(stored.failureGeneration, 1); + + const [retry] = await reloaded.due(6_000); + assert.ok(retry); + await first.update(cron.id, { action: "C" }); + await second.update(cron.id, { action: "B" }); + await reloaded.failDueSlot(cron.id, retry, 6_000); + stored = (await createCronStore(map()).get(cron.id))!; + assert.equal(stored.action, "B"); + assert.equal(stored.executionRevision, 3); + assert.equal(stored.failureBackoff, undefined); + assert.equal(stored.deferUntil, 6_000); +}); + test( "pg map: putIfAbsent inserts once and returns the existing row on a conflict (atomic claim)", { skip }, From 46cf70cf8b98d0006f33e23d892fd1f15181097e Mon Sep 17 00:00:00 2001 From: Sina Matian Date: Fri, 11 Sep 2026 15:36:28 -0700 Subject: [PATCH 5/5] Stop exporting internal cron retry types --- src/cron/cron-store.ts | 2 +- src/types.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/cron/cron-store.ts b/src/cron/cron-store.ts index 9261f6197..b74acc61a 100644 --- a/src/cron/cron-store.ts +++ b/src/cron/cron-store.ts @@ -63,7 +63,7 @@ const FAILURE_BACKOFF_MAX_FAILURES = 7; export type DueCron = Cron & { scheduledAt: number }; -export interface CronSlotClaim { +interface CronSlotClaim { id: string; cron: Cron; scheduledAt: number; diff --git a/src/types.ts b/src/types.ts index 5a0ca7b82..37419619b 100644 --- a/src/types.ts +++ b/src/types.ts @@ -235,7 +235,7 @@ export interface CronFireNote { by?: string; } -export interface CronFailureBackoff { +interface CronFailureBackoff { scheduledAt: number; failures: number; }