yc-software · time-attack · Sep 11, 2026 · Sep 11, 2026 · Sep 11, 2026 · Sep 11, 2026
diff --git 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,21 @@ 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 type DueCron = Cron & { scheduledAt: number };
+
+interface CronSlotClaim {
+  id: string;
+  cron: Cron;
+  scheduledAt: number;
+  claimedAt: number;
+  executionRevision: number;
+  priorLastFiredAt?: number;
+}
+
 export interface CronStore {
   create(input: CreateCronInput): Promise<Cron>;
   get(id: string): Promise<Cron | null>;
@@ -78,9 +93,15 @@ export interface CronStore {
   markFired(id: string, at: number, scheduledAt?: number): Promise<void>;
   markAttempted(id: string, at: number): Promise<void>;
   defer(id: string, until: number): Promise<void>;
-  claimSlot(id: string, scheduledAt: number, at: number): Promise<boolean>;
-  unclaimSlot(id: string, scheduledAt: number, at: number, priorLastFiredAt: number | undefined): Promise<void>;
-  due(now: number): Promise<Array<Cron & { scheduledAt: number }>>;
+  claimSlot(id: string, scheduledAt: number, at: number): Promise<CronSlotClaim | null>;
+  completeSlot(id: string, claim: CronSlotClaim): Promise<void>;
+  releaseSlot(id: string, claim: CronSlotClaim, deferUntil?: number): Promise<void>;
+  failSlot(id: string, claim: CronSlotClaim, failedAt: number): Promise<number | undefined>;
+  completeDueSlot(id: string, cron: DueCron, at: number): Promise<void>;
+  deferDueSlot(id: string, cron: DueCron, until: number): Promise<void>;
+  failDueSlot(id: string, cron: DueCron, failedAt: number): Promise<number | undefined>;
+  disableDueSlot(id: string, cron: DueCron): Promise<void>;
+  due(now: number): Promise<DueCron[]>;
 }
 
 export function isDeferred(cron: Pick<Cron, "deferUntil">, now: number): boolean {
@@ -93,12 +114,100 @@ function normalizeTitle(title: string | undefined): string | undefined {
   return trimmed.length > 80 ? `${trimmed.slice(0, 79)}...` : trimmed;
 }
 
+function mergeFields(cron: Cron, fields: Partial<Cron>): Cron {
+  const next = { ...cron };
+  for (const [key, value] of Object.entries(fields)) {
+    if (value === undefined) delete (next as Record<string, unknown>)[key];
+    else (next as Record<string, unknown>)[key] = value;
+  }
+  return next;
+}
+
+function clearFailureState(cron: Cron): Cron {
+  const { failureBackoff: _failureBackoff, ...rest } = cron;
+  return { ...rest, failureGeneration: 0 };
+}
+
+function clearAttemptState(cron: Cron): Cron {
+  const { activeClaimId: _activeClaimId, ...rest } = clearFailureState(cron);
+  return rest;
+}
+
+function requireAtomicUpdate(backing: DurableMap<Cron>): NonNullable<DurableMap<Cron>["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));
+}
+
+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 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): boolean {
+  return (
+    cron.enabled &&
+    !cron.archived &&
+    generation(cron.executionRevision) === generation(expected.executionRevision) &&
+    sameFailureState(cron, expected) &&
+    recoverNextFireAt(cron.schedule, cron.createdAt, cron.lastFiredAt, cron.nextFireAt) === expected.scheduledAt
+  );
+}
+
 export function createCronStore(
   backing: DurableMap<Cron> = createMemoryMap<Cron>(),
   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<Cron>,
+    kind: "execution" | "schedule" = "execution",
+  ): Promise<Cron | null> =>
+    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);
@@ -121,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 } : {}),
@@ -150,17 +261,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, patch.schedule !== undefined ? "schedule" : "execution");
     },
     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 +335,143 @@ export function createCronStore(
         lastFiredAt: at,
         nextFireAt: advanceNextFireAt(cron.schedule, advanceFrom),
         deferUntil: undefined,
+        activeClaimId: undefined,
+        failureBackoff: undefined,
+        failureGeneration: 0,
       });
     },
     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,
+          executionRevision: generation(cron.executionRevision),
+          ...(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 sameFailureSlot = cron.failureBackoff?.scheduledAt === scheduledAt;
+        return mergeFields(cron, {
+          lastFiredAt: at,
+          nextFireAt,
+          deferUntil: undefined,
+          activeClaimId: claimId,
+          failureBackoff: sameFailureSlot ? cron.failureBackoff : undefined,
+          failureGeneration: sameFailureSlot ? generation(cron.failureGeneration) : 0,
+        });
+      });
+      return claim;
+    },
+    async completeSlot(id, claim) {
+      await requireAtomicUpdate(backing)(id, (cron) => {
+        if (cron.activeClaimId !== claim.id) return cron;
+        return mergeFields(cron, {
+          activeClaimId: undefined,
+          failureBackoff: undefined,
+          failureGeneration: 0,
+          ...(claim.cron.schedule.everyMs === undefined && claim.cron.schedule.cron === undefined
+            ? { enabled: false }
+            : {}),
+        });
+      });
     },
-    async unclaimSlot(id, scheduledAt, at, priorLastFiredAt) {
-      const restore = (cron: Cron): Cron => {
-        if (cron.lastFiredAt !== at) return cron;
-        const { lastFiredAt: _dropped, ...rest } = cron;
+    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,
+          ...(!configurationChanged && 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;
+        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, {
+          lastFiredAt: claim.priorLastFiredAt,
+          nextFireAt: claim.scheduledAt,
+          activeClaimId: undefined,
+          failureBackoff: { scheduledAt: claim.scheduledAt, failures },
+          failureGeneration: nextGeneration(cron.failureGeneration),
+          deferUntil,
+        });
+      });
+      return deferUntil;
+    },
+    async completeDueSlot(id, expected, at) {
+      await requireAtomicUpdate(backing)(id, (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: cron.deferUntil === expected.deferUntil ? undefined : cron.deferUntil,
+          failureBackoff: undefined,
+          failureGeneration: 0,
+          ...(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 {
-          ...rest,
-          ...(priorLastFiredAt !== undefined ? { lastFiredAt: priorLastFiredAt } : {}),
-          nextFireAt: scheduledAt,
+          ...cron,
+          failureBackoff: { scheduledAt: expected.scheduledAt, failures },
+          failureGeneration: nextGeneration(cron.failureGeneration),
+          deferUntil,
         };
-      };
-      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 });
+      });
+      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 });
     },
     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<Cron & { scheduledAt: number }> = [];