DurabilityScheduler

Import DurabilityScheduler from durability. Construct one per object when composing alarm-owning capabilities. A scheduler coordinates that object’s work, not work across different objects.

How it uses your Durable Object

The scheduler owns the parts of the Durable Object that its helpers would otherwise compete for:

The object’s one alarm (ctx.storage.setAlarm). Helper writes that add or reschedule pending work run in a storage transaction that also sets the alarm to the earliest pending row of every attached table. Your class’s alarm() must delegate to scheduler.alarm().
ctx.storage.sql stays with the helpers: each creates and registers its own tables, and the scheduler only reads their pending times.
The live instance holds the concurrency permits and their FIFO waiters, the early timer for work due before the alarm, and the list of attached tables. Eviction discards them; the alarm and the rows bring the work back.

1/8 · One scheduler per object instance

The class constructs one DurabilityScheduler from this.ctx. Its engine is the only code that sets the object’s alarm and its early timer, and it starts with alarmConcurrency permits, here 5, which every attached helper shares.

Open source ↗

/**
 * `DurabilityAlarms` + `DurabilityScheduler` example: recurring named alarms
 * that trigger durable operations, sharing one physical Durable Object alarm.
 *
 * Each subscription object schedules a `renew` alarm. When it fires, the
 * handler registers a durable `chargeInvoice` operation (retried on failure
 * independently of the alarm) and schedules the next occurrence. Scheduling a
 * name again replaces its pending occurrence, so `subscribe` is repeatable.
 */
import { DurableObject } from 'cloudflare:workers';
import {
  Durability,
  DurabilityAlarms,
  DurabilityScheduler,
  type DurableCall,
} from 'durability';

type Env = {
  SUBSCRIPTIONS: DurableObjectNamespace<Subscription>;
  PAYMENTS: Fetcher;
};

const renewalPeriodMs = 30 * 24 * 60 * 60 * 1000;

type ChargeInput = { customerId: string; periodStart: number };

export class Subscription extends DurableObject<Env> {
  // One object uses both helpers, so they must share a scheduler: it owns the
  // single physical alarm and one concurrency pool for all handlers.
  private readonly scheduler = new DurabilityScheduler({
    context: this.ctx,
    alarmConcurrency: 5,
  });

  private readonly durability = new Durability({
    scheduler: this.scheduler,
    handlers: {
      chargeInvoice: async ({ id, payload }: DurableCall<ChargeInput>) => {
        const response = await this.env.PAYMENTS.fetch(
          'https://payments/charge',
          {
            method: 'POST',
            headers: { 'Idempotency-Key': id },
            body: JSON.stringify(payload),
          }
        );
        if (!response.ok) {
          throw new Error(`Charge failed with ${response.status}`);
        }
      },
    },
    retries: { maxAttempts: 8 },
  });

  private readonly alarms = new DurabilityAlarms({
    scheduler: this.scheduler,
    handlers: {
      renew: async ({ idempotencyKey, scheduledTime }) => {
        const customerId = await this.ctx.storage.get<string>('customerId');
        if (customerId === undefined) {
          return; // Cancelled; let this occurrence complete without renewing.
        }
        // The alarm's idempotencyKey is stable across its retries, so a
        // retried occurrence cannot double-charge.
        await this.durability.chargeInvoice({
          id: idempotencyKey,
          payload: { customerId, periodStart: scheduledTime },
        });
        await this.alarms.renew(scheduledTime + renewalPeriodMs);
      },
    },
  });

  async subscribe(customerId: string) {
    await this.ctx.storage.put('customerId', customerId);
    await this.alarms.renew(Date.now() + renewalPeriodMs);
  }

  async cancel() {
    await this.ctx.storage.delete('customerId');
  }

  /** Destructive retention cleanup for settled records older than 90 days. */
  async purgeHistory() {
    const cutoff = Date.now() - 90 * 24 * 60 * 60 * 1000;
    return {
      operations: await this.durability.purgeBefore(cutoff),
      alarms: await this.alarms.purgeBefore(cutoff),
    };
  }

  alarm(info?: AlarmInvocationInfo) {
    return this.scheduler.alarm(info);
  }
}

export default {
  async fetch(request, env): Promise<Response> {
    const url = new URL(request.url);
    const [, customerId, action] = url.pathname.split('/');
    if (!customerId || request.method !== 'POST') {
      return new Response('Expected POST /:customerId/(subscribe|cancel)', {
        status: 404,
      });
    }
    const subscription = env.SUBSCRIPTIONS.getByName(customerId);

    if (action === 'subscribe') {
      await subscription.subscribe(customerId);
      return new Response('subscribed', { status: 201 });
    }
    if (action === 'cancel') {
      await subscription.cancel();
      return new Response('cancelled');
    }
    return new Response('Not found', { status: 404 });
  },
} satisfies ExportedHandler<Env>;

import { z } from 'zod';
import { trace, traceMethods, traceStorage } from './dev.js';
import {
  DurableRetryPolicyError,
  isErrorInstance,
  isNonRetryable,
  reportFailure,
  serializeError,
  type SerializedError,
} from './errors.js';
import {
  createEmitter,
  type DurabilityLifecycleEvent,
  type Emit,
  type LifecycleEntity,
} from './lifecycle.js';
import {
  isSchemaCurrent,
  migrateCapability,
  type MigrationCapability,
} from './migrations.js';
import { assertPositiveInteger, type AttemptPolicy } from './policy.js';
import { deleteMemos, deleteStaleMemos } from './memoize.js';
import {
  claimAttempt,
  exhaustAttempts,
  updateAttempt,
  type DurableTable,
} from './storage.js';
import type { DurableMigrations } from '@durability/storage';

/** The Durable Object state durability needs: storage plus optional `waitUntil`. */
export type DurabilityContext = Pick<DurableObjectState, 'storage'> &
  Partial<Pick<DurableObjectState, 'waitUntil'>>;

/** Options for the scheduler shared by every durability helper in one Durable Object. */
export type DurabilitySchedulerOptions = {
  /** Maximum handlers running at once across eager operations and alarm work. Defaults to 10. */
  alarmConcurrency?: number;
  /** Physical-alarm delay from 1 to 15 seconds. Defaults to 15 seconds. */
  alarmMinDelayMs?: number;
  /** Time before a running alarm hands unfinished work to a new alarm. Defaults to 14 minutes. */
  alarmHandoffMs?: number;
  /**
   * Experimental. Start due work from in-memory timers at its exact scheduled
   * time, independent of `alarmMinDelayMs`, and let newly due work run without
   * waiting behind long-running handlers. The physical alarm stays armed as the
   * eviction fallback. Off by default.
   */
  liveTimers?: boolean;
  /**
   * Called each time an alarm run fails and the scheduler re-arms its own
   * fallback alarm instead of letting the platform retry, for example to send
   * the error to an external service. Best effort: it runs after the fallback is
   * armed, a returned promise is passed to `waitUntil`, and a throw or rejection
   * is only logged, so it never changes scheduling.
   */
  onAlarmRetry?: (info: AlarmRetryInfo) => void | Promise<void>;
};

/** Details of a failed alarm run passed to `onAlarmRetry`. */
export type AlarmRetryInfo = {
  error: SerializedError;
  /** Delay until the re-armed fallback alarm fires. */
  retryInMs: number;
  /** The platform's retry count for this alarm invocation, when it provided one. */
  retryCount?: number;
  /** Whether the platform marked this invocation as a retry, when it provided one. */
  isRetry?: boolean;
  /** The owning Durable Object's ID, when the context carries one. */
  objectId?: string;
};

/** Configuration for a scheduler shared by several helpers. */
export type DurabilitySchedulerConfig = DurabilitySchedulerOptions & {
  context: DurabilityContext;
};

/**
 * How a helper attaches to a scheduler: an existing shared one, or a private
 * one created from the Durable Object context with the given scheduler options.
 */
export type SchedulerAttachment =
  | { scheduler: DurabilityScheduler; context?: never }
  | (DurabilitySchedulerConfig & { scheduler?: never });

const minAlarmDelayMs = 1_000;
const maxAlarmDelayMs = 15_000;
// setTimeout treats delays past this as zero, so anything further out can only
// be served by the physical alarm fallback.
const maxTimerMs = 2 ** 31 - 1;

const pendingHeadRowSchema = z.object({
  pk: z.string(),
  next_attempt_at: z.number().int().nonnegative(),
});
const countRowSchema = z.object({ count: z.number().int().nonnegative() });

type PhysicalAlarm = Pick<
  DurableObjectTransaction,
  'getAlarm' | 'setAlarm' | 'deleteAlarm'
>;

/** One record table attached to the scheduler: how to find and run its due work. */
export type Participant = {
  table: DurableTable;
  runDue(now: number, info: AlarmInvocationInfo | undefined): Promise<void>;
  /** Earliest due time, for tables without `status` and `next_attempt_at` columns. */
  nextPendingAt?(sql: SqlStorage): number | undefined;
  /**
   * True while the participant runs its own due work from an alarm invocation
   * it keeps open. The in-memory timer then ignores its pending time, which only
   * arms the physical alarm as the eviction fallback.
   */
  holding?(): boolean;
};

/**
 * Called after a record may have settled. Successful completions pass the
 * executed row because some tables delete it on success; other changes pass
 * nothing and listeners re-read storage.
 */
export type SettlementListener = (completed?: ExecutionSpec['row']) => void;

export type ActiveExecution = {
  generation: string;
  createdAt: number;
  settled: Promise<void>;
  controller?: AbortController;
};

/** Everything that differs between running a durable operation and a named alarm attempt. */
export type ExecutionSpec = {
  table: DurableTable;
  key: string;
  row: { generation_id: string; created_at: number };
  label: string;
  entity: LifecycleEntity;
  active: Map<string, ActiveExecution>;
  policy: AttemptPolicy;
  emit: Emit;
  /** When true, the completed event carries the handler result; omitted by default to keep results out of telemetry. */
  includeResult?: boolean;
  exhaustedError: () => Error;
  timeoutError: () => Error;
  invoke: (attempt: number, signal: AbortSignal) => unknown;
  /**
   * Persists a successful attempt and returns its settlement timestamp, or
   * undefined when a newer generation won. May throw to turn the success into
   * a failure.
   */
  complete: (result: unknown, attempt: number) => number | undefined;
  /** Kind-specific terminal rule beyond non-retryable errors and exhausted attempts. */
  isTerminal: (error: unknown) => boolean;
};

type RetryDecision = {
  error: unknown;
  terminal: boolean;
  timestamp: number;
  nextAttemptAt: number;
};

const decideRetry = (
  spec: ExecutionSpec,
  attempt: number,
  error: unknown
): RetryDecision => {
  const terminal =
    isNonRetryable(error) ||
    attempt >= spec.policy.maxAttempts ||
    spec.isTerminal(error);
  if (terminal) {
    const timestamp = Date.now();
    return { error, terminal, timestamp, nextAttemptAt: timestamp };
  }

  try {
    const rawDelay = spec.policy.delay(attempt);
    if (!Number.isFinite(rawDelay) || rawDelay < 0) {
      throw new DurableRetryPolicyError(spec.label);
    }
    const delay = Math.round(rawDelay);
    const timestamp = Date.now();
    if (
      !Number.isSafeInteger(delay) ||
      !Number.isSafeInteger(timestamp + delay)
    ) {
      throw new DurableRetryPolicyError(spec.label);
    }
    return {
      error,
      terminal: false,
      timestamp,
      nextAttemptAt: timestamp + delay,
    };
  } catch (policyError) {
    const timestamp = Date.now();
    return {
      error: isErrorInstance(policyError, DurableRetryPolicyError)
        ? policyError
        : new DurableRetryPolicyError(spec.label, policyError),
      terminal: true,
      timestamp,
      nextAttemptAt: timestamp,
    };
  }
};

/**
 * Shared machinery behind {@link DurabilityScheduler}: transactions, the physical
 * alarm, the concurrency pool, and the attempt lifecycle. Package-internal.
 */
export class Engine {
  readonly sql: SqlStorage;
  private readonly context: DurabilityContext;
  private readonly storage: DurableObjectStorage;
  private readonly alarmConcurrency: number;
  private readonly alarmMinDelayMs: number;
  private readonly alarmHandoffMs: number;
  readonly liveTimers: boolean;
  private readonly onAlarmRetry: DurabilitySchedulerOptions['onAlarmRetry'];
  private readonly participants: Participant[] = [];
  private readonly migrated = new Set<MigrationCapability>();
  private availablePermits: number;
  private readonly permitWaiters: Array<(release: () => void) => void> = [];
  private alarmRefresh: Promise<void> | undefined;
  private earlyAlarm:
    | { timestamp: number; timer: ReturnType<typeof setTimeout> }
    | undefined;
  private readonly watchers = new Map<
    string,
    Map<string, Set<SettlementListener>>
  >();
  // Keys whose attempt is currently executing, per table. The in-memory timer
  // skips these so a still-due active row cannot drive a zero-delay timer loop.
  private readonly activeByTable = new Map<string, Set<string>>();

  constructor(context: DurabilityContext, options: DurabilitySchedulerOptions) {
    this.context = context;
    this.storage = traceStorage(context.storage);
    this.sql = this.storage.sql;
    this.alarmConcurrency = options.alarmConcurrency ?? 10;
    this.alarmMinDelayMs = options.alarmMinDelayMs ?? 15_000;
    this.alarmHandoffMs = options.alarmHandoffMs ?? 14 * 60_000;
    this.liveTimers = options.liveTimers ?? false;
    this.onAlarmRetry = options.onAlarmRetry;
    if (
      this.onAlarmRetry !== undefined &&
      typeof this.onAlarmRetry !== 'function'
    ) {
      throw new TypeError('onAlarmRetry must be a function');
    }
    assertPositiveInteger('alarmConcurrency', this.alarmConcurrency);
    assertPositiveInteger('alarmMinDelayMs', this.alarmMinDelayMs);
    if (
      this.alarmMinDelayMs < minAlarmDelayMs ||
      this.alarmMinDelayMs > maxAlarmDelayMs
    ) {
      throw new RangeError(
        `alarmMinDelayMs must be between ${minAlarmDelayMs} and ${maxAlarmDelayMs}`
      );
    }
    assertPositiveInteger('alarmHandoffMs', this.alarmHandoffMs);
    this.availablePermits = this.alarmConcurrency;
  }

  register(participant: Participant): void {
    const { name } = participant.table;
    if (this.participants.some(({ table }) => table.name === name)) {
      throw new Error(`A ${name} helper is already attached to this scheduler`);
    }
    this.participants.push(participant);
  }

  waitUntil(
    promise: Promise<unknown>,
    onError: (error: unknown) => void
  ): void {
    if (!this.context.waitUntil) {
      return;
    }
    try {
      this.context.waitUntil(promise);
    } catch (error) {
      onError(error);
    }
  }

  /**
   * The owning Durable Object's ID. The public context type omits it, but the
   * `DurableObjectState` applications pass at runtime carries it.
   */
  objectId(): string | undefined {
    const id: unknown = Reflect.get(this.context, 'id');
    return id === undefined || id === null ? undefined : String(id);
  }

  emitter<
    Event extends { entityKind: string; type: string } =
      DurabilityLifecycleEvent,
  >(
    hook: ((event: Event) => void | Promise<void>) | undefined
  ): (event: Event) => void {
    return createEmitter<Event>(hook, (promise) =>
      this.context.waitUntil?.(promise)
    );
  }

  /** Applies a helper's SQLite migrations. */
  migrateSchema(
    capability: MigrationCapability,
    migrations: DurableMigrations
  ): void {
    // Helpers of one kind on one scheduler share their tables; checking the
    // schema again for each would only add statements to every construction.
    if (this.migrated.has(capability)) {
      return;
    }
    if (!isSchemaCurrent(this.sql, capability, migrations)) {
      migrateCapability(this.storage, capability, migrations, undefined);
    }
    this.migrated.add(capability);
  }

  /** Runs synchronous writes atomically without touching the physical alarm. */
  transactionSync<T>(write: (sql: SqlStorage) => T): T {
    return this.storage.transactionSync(() => write(this.sql));
  }

  /**
   * Runs the writes and reconciles the physical alarm in one storage
   * transaction, then publishes the in-memory timer only after commit.
   */
  async transaction<T>(write: (sql: SqlStorage) => T): Promise<T> {
    let nextPendingAt: number | undefined;
    const result = await this.storage.transaction(async (transaction) => {
      const value = write(this.sql);
      nextPendingAt = await this.reconcile(transaction);
      return value;
    });
    this.scheduleEarlyAlarm(nextPendingAt);
    return result;
  }

  private trackActive(table: string, key: string): void {
    let keys = this.activeByTable.get(table);
    if (!keys) {
      keys = new Set();
      this.activeByTable.set(table, keys);
    }
    keys.add(key);
  }

  private releaseActive(table: string, key: string): void {
    const keys = this.activeByTable.get(table);
    if (!keys) {
      return;
    }
    keys.delete(key);
    if (keys.size === 0) {
      this.activeByTable.delete(table);
    }
  }

  /**
   * The earliest pending time for one participant: `physical` arms the physical
   * alarm (eviction fallback) and includes active work; `timer` drives the
   * in-memory timer and, under liveTimers, skips work that is already executing
   * so a still-due active row cannot re-fire a zero-delay timer in a loop.
   */
  private pendingTimes(participant: Participant): {
    physical?: number | undefined;
    timer?: number | undefined;
  } {
    if (participant.nextPendingAt) {
      const at = participant.nextPendingAt(this.sql);
      if (at === undefined) {
        return {};
      }
      return participant.holding?.()
        ? { physical: at }
        : { physical: at, timer: at };
    }
    const active = this.activeByTable.get(participant.table.name);
    const limit = this.liveTimers ? (active?.size ?? 0) + 1 : 1;
    const rows = pendingHeadRowSchema.array().parse(
      this.sql
        .exec(
          `SELECT ${participant.table.key} AS pk, next_attempt_at
           FROM ${participant.table.name}
           WHERE status = 'pending'
           ORDER BY next_attempt_at ASC
           LIMIT ?`,
          limit
        )
        .toArray()
    );
    const physical = rows[0]?.next_attempt_at;
    if (!this.liveTimers) {
      if (physical === undefined) {
        return {};
      }
      return { physical, timer: physical };
    }
    const eligible = rows.find((row) => !active?.has(row.pk));
    return { physical, timer: eligible?.next_attempt_at };
  }

  /**
   * Arms the physical alarm from the earliest pending time (the eviction
   * fallback) and returns the earliest time the in-memory timer should fire.
   */
  private async reconcile(
    physicalAlarm: PhysicalAlarm
  ): Promise<number | undefined> {
    const physicalTimes: number[] = [];
    const timerTimes: number[] = [];
    for (const participant of this.participants) {
      const { physical, timer } = this.pendingTimes(participant);
      if (physical !== undefined) {
        physicalTimes.push(physical);
      }
      if (timer !== undefined) {
        timerTimes.push(timer);
      }
    }

    const physicalNextAt =
      physicalTimes.length === 0 ? undefined : Math.min(...physicalTimes);
    const currentAlarm = await physicalAlarm.getAlarm();
    if (physicalNextAt === undefined) {
      if (currentAlarm !== null) {
        await physicalAlarm.deleteAlarm();
      }
      return undefined;
    }

    const target = Math.max(physicalNextAt, Date.now() + this.alarmMinDelayMs);
    // Preserve a fallback inside this range so frequent reconciliation cannot push it later.
    if (
      currentAlarm === null ||
      currentAlarm < physicalNextAt ||
      currentAlarm > target
    ) {
      await physicalAlarm.setAlarm(target);
    }
    return timerTimes.length === 0 ? undefined : Math.min(...timerTimes);
  }

  private scheduleEarlyAlarm(timestamp: number | undefined): void {
    if (this.earlyAlarm?.timestamp === timestamp) {
      return;
    }
    if (this.earlyAlarm) {
      clearTimeout(this.earlyAlarm.timer);
      this.earlyAlarm = undefined;
    }
    if (timestamp === undefined) {
      return;
    }
    const delay = timestamp - Date.now();
    // Without liveTimers the in-memory timer only covers the sub-min-delay
    // window; the physical alarm handles everything beyond it. With liveTimers
    // it fires at the exact due time no matter how far out, except past the
    // timer's range where only the physical fallback can serve it.
    if (!this.liveTimers && delay >= this.alarmMinDelayMs) {
      return;
    }
    if (delay > maxTimerMs) {
      return;
    }

    const timer = setTimeout(
      () => {
        if (this.earlyAlarm?.timer !== timer) {
          return;
        }
        this.earlyAlarm = undefined;
        trace('scheduler.early_alarm', { scheduledTime: timestamp });
        const onError = (error: unknown) =>
          reportFailure(
            'durability.background_execution.failed',
            { scheduledTime: timestamp },
            error
          );
        const execution = this.alarm().catch(onError);
        this.waitUntil(execution, onError);
      },
      Math.max(delay, 0)
    );
    this.earlyAlarm = { timestamp, timer };
  }

  /** Coalesces concurrent refresh requests into one reconciliation transaction. */
  scheduleNextAlarm(): Promise<void> {
    if (this.alarmRefresh) {
      return this.alarmRefresh;
    }
    this.alarmRefresh = this.transaction(() => undefined).finally(() => {
      this.alarmRefresh = undefined;
    });
    return this.alarmRefresh;
  }

  /**
   * Aborts matching active handlers, then deletes one kind's records created
   * before the timestamp and reconciles the alarm in the same transaction.
   */
  async purge(
    table: DurableTable,
    before: number,
    active: Map<string, ActiveExecution>,
    abortReason: string
  ): Promise<number> {
    if (!Number.isSafeInteger(before) || before < 0) {
      throw new RangeError('timestamp must be a non-negative safe integer');
    }
    for (const running of active.values()) {
      if (running.createdAt < before) {
        running.controller?.abort(new Error(abortReason));
      }
    }

    const removed = await this.transaction((sql) => {
      const { count } = countRowSchema.parse(
        sql
          .exec(
            `SELECT COUNT(*) AS count FROM ${table.name} WHERE created_at < ?`,
            before
          )
          .toArray()[0]
      );
      sql.exec(`DELETE FROM ${table.name} WHERE created_at < ?`, before);
      deleteStaleMemos(sql, table);
      return count;
    });

    if (removed > 0) {
      for (const key of this.watchers.get(table.name)?.keys() ?? []) {
        this.notify(table, key);
      }
    }
    return removed;
  }

  /** In-memory only: listeners never survive eviction. */
  watch(
    table: DurableTable,
    key: string,
    listener: SettlementListener
  ): () => void {
    const byKey =
      this.watchers.get(table.name) ??
      new Map<string, Set<SettlementListener>>();
    this.watchers.set(table.name, byKey);
    const listeners = byKey.get(key) ?? new Set<SettlementListener>();
    byKey.set(key, listeners);
    listeners.add(listener);

    return () => {
      listeners.delete(listener);
      if (listeners.size === 0 && byKey.get(key) === listeners) {
        byKey.delete(key);
      }
    };
  }

  /** Call only after the change has committed. */
  notify(
    table: DurableTable,
    key: string,
    completed?: ExecutionSpec['row']
  ): void {
    for (const listener of this.watchers.get(table.name)?.get(key) ?? []) {
      listener(completed);
    }
  }

  tryAcquirePermit(): (() => void) | undefined {
    if (this.availablePermits === 0) {
      return undefined;
    }
    this.availablePermits -= 1;
    return this.createRelease();
  }

  acquirePermit(): Promise<() => void> {
    const release = this.tryAcquirePermit();
    if (release) {
      return Promise.resolve(release);
    }
    return new Promise((resolve) => {
      this.permitWaiters.push(resolve);
    });
  }

  private createRelease(): () => void {
    let released = false;
    return () => {
      if (released) {
        return;
      }
      released = true;
      const waiter = this.permitWaiters.shift();
      if (waiter) {
        waiter(this.createRelease());
      } else {
        this.availablePermits += 1;
      }
    };
  }

  async runConcurrent<Item>(
    items: Item[],
    worker: (item: Item) => Promise<void>
  ): Promise<void> {
    let nextIndex = 0;
    await Promise.all(
      Array.from(
        { length: Math.min(this.alarmConcurrency, items.length) },
        async () => {
          while (nextIndex < items.length) {
            const item = items[nextIndex++];
            if (item !== undefined) {
              // eslint-disable-next-line no-await-in-loop
              await worker(item);
            }
          }
        }
      )
    );
  }

  /**
   * Runs due work and re-arms the physical alarm. A failure that would
   * otherwise propagate to the platform is turned into a retry of our own,
   * because the platform drops an alarm after a few failed deliveries.
   */
  async alarm(info?: AlarmInvocationInfo): Promise<void> {
    try {
      await this.runAlarm(info);
    } catch (error) {
      // Throwing lets the platform retry, but it gives up after a few failures
      // and drops the alarm, stranding pending work. Keeping our own alarm
      // armed turns the failure into a retry we control instead.
      if (!(await this.armFallbackAlarm())) {
        throw error;
      }
      reportFailure(
        'durability.alarm.failed',
        { retryInMs: this.alarmMinDelayMs },
        error
      );
      this.notifyAlarmRetry(error, info);
    }
  }

  private notifyAlarmRetry(
    failure: unknown,
    invocation: AlarmInvocationInfo | undefined
  ): void {
    const hook = this.onAlarmRetry;
    if (!hook) {
      return;
    }

    const objectId = this.objectId();
    const info: AlarmRetryInfo = {
      error: serializeError(failure),
      retryInMs: this.alarmMinDelayMs,
      ...(invocation === undefined
        ? {}
        : {
            retryCount: invocation.retryCount,
            isRetry: invocation.isRetry,
          }),
      ...(objectId === undefined ? {} : { objectId }),
    };
    const report = (hookError: unknown): void =>
      reportFailure(
        'durability.alarm.retry_hook_failed',
        { retryInMs: info.retryInMs },
        hookError
      );
    try {
      const result = hook(info);
      if (result !== undefined) {
        this.waitUntil(Promise.resolve(result).catch(report), report);
      }
    } catch (hookError) {
      report(hookError);
    }
  }

  /**
   * Runs every participant's due work, hands off to a fresh alarm if the
   * invocation approaches its wall-time limit, and re-arms the physical alarm.
   */
  private async runAlarm(info?: AlarmInvocationInfo): Promise<void> {
    const startedAt = Date.now();
    const execution = Promise.all(
      this.participants.map((participant) =>
        participant.runDue(startedAt, info)
      )
    ).then(() => undefined);
    void execution.catch(() => undefined);

    const handoff = Symbol('alarm handoff');
    const remaining = this.alarmHandoffMs - (Date.now() - startedAt);
    let timeout: ReturnType<typeof setTimeout> | undefined;
    let result: void | typeof handoff;
    try {
      result = await Promise.race([
        execution,
        new Promise<typeof handoff>((resolve) => {
          timeout = setTimeout(() => resolve(handoff), Math.max(remaining, 0));
        }),
      ]);
    } finally {
      if (timeout !== undefined) {
        clearTimeout(timeout);
      }
    }

    if (result === handoff) {
      await this.armForHandoff(Date.now());
      return;
    }
    await this.scheduleNextAlarm();
  }

  /** Arms the physical alarm directly, outside any transaction; the next run reconciles properly. */
  private async armFallbackAlarm(): Promise<boolean> {
    try {
      await this.storage.setAlarm(Date.now() + this.alarmMinDelayMs);
      return true;
    } catch {
      return false;
    }
  }

  private armForHandoff(timestamp: number): Promise<void> {
    return this.storage.transaction(async (physicalAlarm) => {
      const currentAlarm = await physicalAlarm.getAlarm();
      const target = Math.max(timestamp, Date.now() + this.alarmMinDelayMs);
      if (currentAlarm === null || currentAlarm > target) {
        await physicalAlarm.setAlarm(target);
      }
    });
  }

  /**
   * Registers the execution in its active map synchronously, then claims and
   * runs one attempt in the background. The returned promise settles when the
   * attempt's outcome has been persisted, not when a timed-out handler settles.
   */
  startExecution(spec: ExecutionSpec, eagerPermit?: () => void): Promise<void> {
    const { key, row, policy } = spec;
    let settleActive!: () => void;
    const settled = new Promise<void>((resolve) => {
      settleActive = resolve;
    });
    const activeExecution: ActiveExecution = {
      generation: row.generation_id,
      createdAt: row.created_at,
      settled,
    };
    spec.active.set(key, activeExecution);
    this.trackActive(spec.table.name, key);

    let releasePermit = eagerPermit;
    let handlerStarted = false;
    let handlerSettled = false;
    let policySettled = false;
    const releaseIfSettled = (): void => {
      if (policySettled && (!handlerStarted || handlerSettled)) {
        releasePermit?.();
        settleActive();
      }
    };

    const execution = (async (): Promise<void> => {
      releasePermit ??= await this.acquirePermit();
      const attempt = claimAttempt(
        this.sql,
        spec.table,
        key,
        row.generation_id,
        policy.maxAttempts
      );
      if (attempt === undefined) {
        await this.failExhausted(spec);
        return;
      }

      const startedAt = Date.now();
      spec.emit({
        ...spec.entity,
        type: 'attempt_started',
        timestamp: startedAt,
        attempt,
      });
      const controller = new AbortController();
      activeExecution.controller = controller;
      let timeout: ReturnType<typeof setTimeout> | undefined;
      try {
        handlerStarted = true;
        const handlerResult = Promise.resolve().then(() =>
          spec.invoke(attempt, controller.signal)
        );
        void handlerResult
          .finally(() => {
            handlerSettled = true;
            releaseIfSettled();
          })
          .catch(() => undefined);
        const timeoutResult = new Promise<never>((_resolve, reject) => {
          timeout = setTimeout(() => {
            const error = spec.timeoutError();
            controller.abort(error);
            reject(error);
          }, policy.attemptTimeoutMs);
        });
        const result = await Promise.race([handlerResult, timeoutResult]);

        // One transaction: a failed memo cleanup must undo the completion, not orphan the memos.
        const timestamp = this.transactionSync((sql) => {
          const completedAt = spec.complete(result, attempt);
          if (completedAt !== undefined) {
            deleteMemos(sql, spec.table, key);
          }
          return completedAt;
        });
        if (timestamp !== undefined) {
          this.notify(spec.table, key, row);
          spec.emit({
            ...spec.entity,
            type: 'attempt_settled',
            timestamp,
            attempt,
            durationMs: Math.max(0, timestamp - startedAt),
            outcome: 'completed',
            ...(spec.includeResult ? { result } : {}),
          });
        }
      } catch (caught) {
        await this.settleFailure(spec, attempt, startedAt, caught);
      } finally {
        if (timeout !== undefined) {
          clearTimeout(timeout);
        }
      }
    })().finally(() => {
      policySettled = true;
      releaseIfSettled();
    });

    void settled.finally(() => {
      if (spec.active.get(key) === activeExecution) {
        spec.active.delete(key);
      }
      this.releaseActive(spec.table.name, key);
      // Re-arm the in-memory timer now that this key is eligible again, so work
      // the timer skipped while it executed is picked up at its exact time.
      if (this.liveTimers) {
        void this.scheduleNextAlarm().catch((error) =>
          reportFailure(
            'durability.scheduler.refresh_failed',
            { table: spec.table.name, key },
            error
          )
        );
      }
    });
    return execution;
  }

  private async failExhausted(spec: ExecutionSpec): Promise<void> {
    const { key, row, policy } = spec;
    const exhausted = serializeError(spec.exhaustedError());
    const attempt = await this.transaction((sql) => {
      const exhaustedAttempt = exhaustAttempts(
        sql,
        spec.table,
        key,
        row.generation_id,
        policy.maxAttempts,
        exhausted
      );
      if (exhaustedAttempt !== undefined) {
        deleteMemos(sql, spec.table, key);
      }
      return exhaustedAttempt;
    });
    if (attempt === undefined) {
      return;
    }

    this.notify(spec.table, key);
    spec.emit({
      ...spec.entity,
      type: 'terminal',
      timestamp: Date.now(),
      attempt,
      reason: 'attempts_exhausted',
      error: exhausted,
    });
  }

  private async settleFailure(
    spec: ExecutionSpec,
    attempt: number,
    startedAt: number,
    caught: unknown
  ): Promise<void> {
    const { key, row } = spec;
    const decision = decideRetry(spec, attempt, caught);
    const serializedError = serializeError(decision.error);
    const updated = await this.transaction((sql) => {
      const owned = updateAttempt(
        sql,
        spec.table,
        { key, generation: row.generation_id, attempt },
        'status = ?, next_attempt_at = ?, last_error = ?, last_error_name = ?',
        decision.terminal ? 'failed' : 'pending',
        decision.nextAttemptAt,
        serializedError.message,
        serializedError.name
      );
      if (owned && decision.terminal) {
        deleteMemos(sql, spec.table, key);
      }
      return owned;
    });
    if (!updated) {
      return;
    }
    if (decision.terminal) {
      this.notify(spec.table, key);
    }

    const settled = {
      ...spec.entity,
      type: 'attempt_settled' as const,
      timestamp: decision.timestamp,
      attempt,
      durationMs: Math.max(0, decision.timestamp - startedAt),
      error: serializedError,
    };
    spec.emit(
      decision.terminal
        ? { ...settled, outcome: 'failed' }
        : {
            ...settled,
            outcome: 'retry_scheduled',
            nextAttemptAt: decision.nextAttemptAt,
          }
    );
  }
}

const engines = new WeakMap<DurabilityScheduler, Engine>();

/**
 * Owns the Durable Object's physical alarm, storage, and concurrency
 * pool so {@link Durability} and {@link DurabilityAlarms} can share them.
 *
 * Construct one per Durable Object when using both helpers, pass it to each,
 * and delegate the object's `alarm` method to {@link DurabilityScheduler.alarm}.
 * A helper constructed directly from the Durable Object context creates a
 * private scheduler instead.
 *
 * @example
 * ```ts
 * class Jobs extends DurableObject<Env> {
 *   private readonly scheduler = new DurabilityScheduler({ context: this.ctx });
 *   private readonly durability = new Durability({
 *     scheduler: this.scheduler,
 *     handlers,
 *   });
 *   private readonly alarms = new DurabilityAlarms({
 *     scheduler: this.scheduler,
 *     handlers: alarmHandlers,
 *   });
 *
 *   alarm(info?: AlarmInvocationInfo) {
 *     return this.scheduler.alarm(info);
 *   }
 * }
 * ```
 */
export class DurabilityScheduler {
  constructor(config: DurabilitySchedulerConfig) {
    engines.set(this, new Engine(config.context, config));
  }

  /** Runs due work for every attached helper and re-arms the physical alarm. */
  alarm(info?: AlarmInvocationInfo): Promise<void> {
    return engineFor({ scheduler: this }).alarm(info);
  }
}

traceMethods(DurabilityScheduler.prototype, 'DurabilityScheduler');

/**
 * Resolves the engine a helper should attach to: the shared one behind a
 * scheduler, or a private one built from the Durable Object context.
 */
export const engineFor = (attachment: SchedulerAttachment): Engine => {
  if (attachment.scheduler === undefined) {
    return new Engine(attachment.context, attachment);
  }
  const engine = engines.get(attachment.scheduler);
  if (!engine) {
    throw new Error('Unknown DurabilityScheduler instance');
  }
  return engine;
};

Share one alarm between two helpers

Details

Follow the subscription example, where Durability and DurabilityAlarms share one scheduler.

Durable state: Nothing is written yet.
In-flight / in memory: The engine, its permits, and its timer are rebuilt after every eviction.

Click the diagram to open a full-screen explorer with the complete source files. Select a component to jump to its code, or select a linked code line to highlight the affected architecture. Use the ← / → keys or on-screen arrows to move through the flow; each step’s explanation sits under the diagram. Escape or the close button returns to the guide. Failure scenarios, the full step list, and durable/in-memory state notes are available under Details. These are explanatory traces of the real source, not executable simulations or guarantees about an external service.

JavaScript is disabled. Read the complete example source or follow the complete ordered walkthrough below.

examples/subscription-renewals/worker.ts

packages/durability/src/scheduler.ts

Share one alarm between two helpers

Evict the object with work pending

Eviction discards the engine. Assume the object is evicted while a renewal runs and a charge waits for a permit. Everything the engine kept in memory is gone; everything committed is still in storage. Read these source lines. Durable state: Rows in both tables, their attempt counts, and the alarm. In memory: Permits, waiters, the early timer, running keys, and handler promises are lost.
The alarm replaces the lost timer. Work that was waiting on the early timer now waits for the alarm. Every commit set the alarm no later than alarmMinDelayMs after it, so that work runs at most that much later than due. Read these source lines. Durable state: The alarm time. In memory: The timer is gone.
The alarm starts a new instance. The runtime creates a new instance, whose fields construct the scheduler and both helpers again, and calls alarm(). Without this delegation, pending rows in either table would not run again. Read these source lines. Durable state: Both tables and the alarm. In memory: A fresh engine with all permits free.
Only constructed helpers are consulted. The pass runs the callbacks of the helpers registered in this instance. Constructing every helper as a field, as the example does, ensures both tables are read. Read these source lines. Durable state: Due rows are read from both tables. In memory: The due lists for this pass.
Interrupted attempts still count. Each due row gets a new attempt through a conditional UPDATE of its stored count. The interrupted renewal used attempt 1, so this is attempt 2 with the same identity. Read these source lines. Durable state: The stored attempt count increases before the handler runs. In memory: New handler calls replace the lost ones.

Work within alarm limits

A failing pass sets its own alarm. The runtime retries a throwing alarm() only a few times and then drops it, which would strand pending rows. The engine catches the error and sets the alarm alarmMinDelayMs later instead. Read these source lines. Durable state: A new alarm time is written directly to storage. In memory: The failed pass and its error.
onAlarmRetry reports the failure. After the fallback alarm is set, the optional onAlarmRetry hook receives the error. It is best effort: a returned promise goes to waitUntil, and its own failure is only logged. Read these source lines. Durable state: Nothing beyond the fallback alarm. In memory: The hook call and its promise.
Only a failed setAlarm reaches the platform. If even the fallback setAlarm fails, the original error is thrown and the platform’s own retries apply. Once those are exhausted, the object has no alarm until another write reconciles it. Read these source lines. Durable state: Possibly no alarm at all. In memory: The error propagates to the runtime.
Hand off before the wall-time limit. If due work is still running after alarmHandoffMs (14 minutes by default), the pass sets a new alarm and returns before the runtime’s alarm wall-time limit. Running attempts continue if the instance stays alive. Read these source lines. Durable state: The next alarm is stored before the pass returns. In memory: The running attempts and their promises.

Behind the scenes

Durable state: the scheduler creates no job table of its own. Attached helpers provide their tables and due-work callbacks; the engine consults them to reconcile the object’s one physical alarm. A duplicate participant table is rejected.

Engine transactions reconcile that alarm alongside their storage changes. The next physical alarm is bounded by the earliest due work and now + alarmMinDelayMs. Near-due work can also start from an in-memory timer; experimental liveTimers extends timer scheduling further into the future.

In memory: timers, active execution locks, promises, and the FIFO concurrency pool. Eager operations and ordinary scheduled work share the default ten permits. Some capabilities, such as recurring schedule runs, explicitly maintain a separate execution limit; a shared scheduler is not a universal limit on all application work.

Recovery: a physical alarm reconstructs progress from durable participant state after eviction. Long alarm invocations hand off before the wall-time limit. If a pass itself fails, the scheduler tries to arm a fallback at alarmMinDelayMs; only failure to arm that fallback propagates to platform retries.

Never put two independent alarm-owning schedulers in one object: each can overwrite the other’s physical-alarm decision. Not every helper needs alarm work: the retained log uses the engine for storage operations but does not automatically schedule retention.

Implementation: scheduler.ts (Engine.transaction, reconcile, runAlarm, startExecution).

Using both in one Durable Object

A Durable Object has one physical alarm. When one object uses both helpers, create a DurabilityScheduler and pass it to each so they share the alarm, the storage, and the concurrency pool:

import { Durability, DurabilityAlarms, DurabilityScheduler } from 'durability';

export class ImageJobs extends DurableObject<Env> {
  private readonly scheduler = new DurabilityScheduler({
    context: this.ctx,
    alarmConcurrency: 10,
    alarmMinDelayMs: 15_000,
  });

  private readonly durability = new Durability({
    scheduler: this.scheduler,
    handlers: {
      resizeImage: async ({ payload }) =>
        this.env.IMAGES.resize(payload.imageId),
    },
  });

  private readonly alarms = new DurabilityAlarms({
    scheduler: this.scheduler,
    handlers: { cleanup: async () => this.env.IMAGES.cleanup() },
  });

  alarm(info?: AlarmInvocationInfo) {
    return this.scheduler.alarm(info);
  }
}

A helper configured with context instead of scheduler creates a private scheduler and accepts the scheduler options (alarmConcurrency, alarmMinDelayMs, alarmHandoffMs) in the same object. Two helpers on the same object must share one scheduler; otherwise each would reconcile the physical alarm against only its own records.

Delivery semantics and concurrency

Work is delivered at least once, as the delivery and safety contract describes. A process can stop after a side effect succeeds but before its completion record commits, so handlers should be idempotent and pass the call ID or alarm idempotencyKey to external services.

alarmConcurrency defaults to 10 and is one FIFO limit shared by eager operation handlers, timer-driven work, and physical-alarm work attached to the same scheduler. Newly registered operations start eagerly only when a permit is immediately available. Work due before the physical-alarm minimum enters the same queue when its in-memory timer fires.

Operation completion records are retained indefinitely so their IDs remain deduplicated. The scheduler owns the Durable Object’s alarm; compose unrelated scheduled work as named alarms instead of replacing the physical alarm.

When a run cannot finish

A Durable Object alarm whose handler throws is retried with backoff a limited number of times and then dropped. Every helper’s alarm() ends by reconciling the physical alarm, and a storage error there would otherwise propagate, so enough of them in a row would leave pending work with nothing armed. Instead, a run that fails arms the physical alarm directly at alarmMinDelayMs from now and returns, reporting durability.alarm.failed with the error; the next run reconciles properly. Only when even that setAlarm fails does the failure reach the platform and its retries.

Long-running alarm calls

Alarm invocations have a 15-minute wall-time limit. If work is still pending after alarmHandoffMs (default 14 minutes), the scheduler retains its promise in memory, arms a fallback after alarmMinDelayMs, and returns from the current invocation. The next alarm attaches to the same promise instead of starting the handler again. If the object is evicted during a handoff, the next alarm reconstructs and executes the persisted pending work.

`liveTimers` (scheduler)

liveTimers?: boolean. Due work starts from an in-memory timer at its exact scheduled time, however far beyond alarmMinDelayMs, and an alarm pass skips records that are already running instead of waiting for them, so a long handler no longer delays newly due work. The physical alarm stays armed at the usual bounded delay as the eviction fallback, and the handoff before the alarm wall-time limit is unchanged.

Gives up: every settled attempt runs one more reconciliation transaction to re-arm the timer, and timers do not survive eviction, so after one the work runs only when the physical alarm fires. Attempts interrupted by the eviction are counted as usual, so an operation with maxAttempts: 1 fails instead of running again.

`onAlarmRetry` (scheduler)

onAlarmRetry?: (info: AlarmRetryInfo) => void | Promise<void>, where info is { error: { name, message }, retryInMs, retryCount?, isRetry?, objectId? }. It is called each time an alarm run fails and the scheduler re-arms its own fallback alarm (see When a run cannot finish), so the error can go to an external service. retryInMs is alarmMinDelayMs; retryCount and isRetry come from the platform’s AlarmInvocationInfo when the invocation carried one, and objectId from the context.

The durability.alarm.failed report still happens. The hook is best effort: it runs after the fallback is armed, a returned promise is passed to waitUntil, and a throw or rejection is only reported as durability.alarm.retry_hook_failed, so it cannot change scheduling. It is not called when even the fallback cannot be armed, because the failure then reaches the platform.

const scheduler = new DurabilityScheduler({
  context: this.ctx,
  onAlarmRetry: (info) => this.env.ERRORS.send(info),
});