Durability

Import Durability from durability. Use it for individually identified jobs whose results must survive retries and restarts. See recurring billing for composition with named alarms.

How it uses your Durable Object

Durability adds no storage or infrastructure of its own. It uses the Durable Object it is constructed in:

ctx.storage.sql holds its tables, so operations are as durable as the object and belong to it alone. Another object, even of the same class, has its own rows.
The object’s one alarm (ctx.storage.setAlarm) brings the object back after eviction. Every write that adds or reschedules pending work sets the alarm in the same storage transaction, and your class’s alarm() must delegate to durability.alarm().
The live instance holds everything else: concurrency permits, the running attempts, and a timer for work due before the alarm. It is rebuilt with the helper whenever the runtime creates a new instance of the object.

1/9 · The helper belongs to the object instance

Durability is constructed with this.ctx, the object’s DurableObjectState. It reads and writes only this object’s storage; another object has its own tables and its own alarm.

Open source ↗

/**
 * `Durability` example: at-least-once operations with a stable idempotency key.
 *
 * A resize request is registered durably and survives eviction, restarts, and
 * transient API failures. The handler runs at least once; passing the stable
 * call id as the resize idempotency key lets the image service collapse
 * duplicate attempts into one effect. The persisted result is read back with
 * `getResult`.
 */
import { DurableObject } from 'cloudflare:workers';
import { Durability, NonRetryableError, type DurableCall } from 'durability';

type Env = {
  IMAGE_JOBS: DurableObjectNamespace<ImageJobs>;
  /** Upstream image service; resize calls must be idempotent. */
  IMAGE_API: Fetcher;
};

type ResizeInput = { imageId: string; width: number };
type ResizeOutput = { url: string };

export class ImageJobs extends DurableObject<Env> {
  private readonly durability = new Durability({
    context: this.ctx,
    handlers: {
      resizeImage: async ({
        id,
        payload,
        signal,
      }: DurableCall<ResizeInput>): Promise<ResizeOutput> => {
        // Permanent input problems should not burn retry attempts.
        if (!Number.isInteger(payload.width) || payload.width <= 0) {
          throw new NonRetryableError(`Invalid width ${payload.width}`);
        }

        // The idempotency key lets the upstream service deduplicate a retry
        // whose previous attempt succeeded after we stopped waiting.
        const response = await this.env.IMAGE_API.fetch(
          'https://image-api/resize',
          {
            method: 'POST',
            headers: { 'Idempotency-Key': id },
            body: JSON.stringify(payload),
            signal,
          }
        );
        if (!response.ok) {
          throw new Error(`Resize failed with ${response.status}`);
        }
        return response.json();
      },
    },
    attemptTimeoutMs: 60_000,
    retries: { maxAttempts: 5 },
  });

  /** Resolves once the job is durably registered, not when it completes. */
  resize(imageId: string, width: number) {
    return this.durability.resizeImage({
      id: `resize:${imageId}:${width}`,
      payload: { imageId, width },
    });
  }

  result(imageId: string, width: number) {
    return this.durability.resizeImage.getResult(`resize:${imageId}:${width}`);
  }

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

export default {
  async fetch(request, env): Promise<Response> {
    const url = new URL(request.url);
    const [, imageId, action] = url.pathname.split('/');
    if (!imageId || !action) {
      return new Response('Expected /:imageId/resize or /:imageId/result', {
        status: 404,
      });
    }
    const width = Number(url.searchParams.get('width') ?? 512);
    const jobs = env.IMAGE_JOBS.getByName(imageId);

    if (request.method === 'POST' && action === 'resize') {
      await jobs.resize(imageId, width);
      return new Response('registered', { status: 202 });
    }
    if (action === 'result') {
      return Response.json(await jobs.result(imageId, width));
    }
    return new Response('Not found', { status: 404 });
  },
} satisfies ExportedHandler<Env>;

import { z } from 'zod';
import { traceFunction, traceMethods } from './dev.js';
import {
  DuplicateDurableCallError,
  DuplicateDurablePayloadError,
  DurableAttemptsExhaustedError,
  DurableAttemptTimeoutError,
  DurableResultSerializationError,
  isErrorInstance,
  NonRetryableError,
  reportFailure,
} from './errors.js';
import {
  createLifecycleHooks,
  type DurabilityLifecycleEvent,
  type LifecycleHook,
  type LifecycleHooks,
  type OperationLifecycleEntity,
} from './lifecycle.js';
import {
  durabilityOperationMigrations,
  durabilityMemoMigrations,
  migrateCapability,
  type DurabilityMigrationResult,
} from './migrations.js';
import {
  createPolicyResolver,
  methodOptionsSchema,
  type DurabilityMethodOptions,
  type ExecutionPolicyOptions,
  type MethodPolicy,
  type PolicyResolver,
} from './policy.js';
import {
  engineFor,
  type ActiveExecution,
  type Engine,
  type SchedulerAttachment,
} from './scheduler.js';
import { canonical, deserialize, serialize } from './serialization.js';
import {
  createMemoize,
  deleteMemosOfDroppedTable,
  memoizedKeys,
  type DurableMemoize,
} from './memoize.js';
import { listCreated, updateAttempt, type DurableTable } from './storage.js';
import {
  registerTrackedSource,
  type UntypedRecord,
  type UntypedSource,
} from './tracked-source.js';

const callRowSchema = z.object({
  id: z.string(),
  operation: z.string(),
  payload: z.string(),
  status: z.enum(['pending', 'completed', 'failed']),
  result: z.string().nullable(),
  attempt: z.number().int().nonnegative(),
  next_attempt_at: z.number().int().nonnegative(),
  last_error: z.string().nullable(),
  last_error_name: z.string().nullable(),
  completed_at: z.number().int().nonnegative().nullable(),
  created_at: z.number().int().nonnegative(),
  generation_id: z.string(),
});

type CallRow = z.infer<typeof callRowSchema>;

const callsTable: DurableTable = { name: 'durability_calls', key: 'id' };

const readCall = (sql: SqlStorage, id: string): CallRow | undefined =>
  callRowSchema
    .optional()
    .parse(
      sql.exec('SELECT * FROM durability_calls WHERE id = ?', id).toArray()[0]
    );

/**
 * The context passed to a durable operation handler for each attempt.
 *
 * @example
 * ```ts
 * const sendEmail = async ({ id, payload, signal }: DurableCall<EmailPayload>) =>
 *   fetch(payload.url, {
 *     method: 'POST',
 *     headers: { 'Idempotency-Key': id },
 *     signal,
 *   });
 * ```
 */
export type DurableCall<Payload> = {
  /**
   * The stable idempotency key supplied when the operation was registered. This
   * is the operation's logical identity: key any external deduplication or
   * once-only effect by `id`, never by `attempt`.
   */
  id: string;
  /** The handler name used to register the operation. */
  operation: string;
  /** The JSON-serializable payload supplied when the operation was registered. */
  payload: Payload;
  /**
   * The one-based attempt ordinal, durably recorded before the handler runs. It
   * is retry and logging context, not a record of what the handler did: never
   * gate once-only work on it (such as `attempt === 1`). Key such work on
   * durable state keyed by `id` instead. See the README identity glossary.
   */
  attempt: number;
  /** Aborts when the attempt exceeds its configured timeout. */
  signal: AbortSignal;
  /** Persists a callback's result by key so a retry of this operation reuses it. */
  memoize: DurableMemoize;
};

/** A function that executes one attempt of a durable operation. */
export type DurableHandler<Payload, Result> = (
  call: DurableCall<Payload>
) => Result | Promise<Result>;

export type HandlerMap = Record<string, (...args: never[]) => unknown>;

type HandlerPayload<Handler> = Handler extends (
  call: DurableCall<infer Payload>
) => unknown
  ? Payload
  : never;

type HandlerResult<Handler> = Handler extends (...args: never[]) => infer Result
  ? Awaited<Result>
  : never;

/** Result types of every operation, keyed by handler name. */
export type OperationResults<Handlers extends HandlerMap> = {
  [Operation in Extract<keyof Handlers, string>]: HandlerResult<
    Handlers[Operation]
  >;
};

type DurableOperationInput<Handler> = {
  /** Stable idempotency key; bound to the operation and its payload. */
  id: string;
  /** JSON-serializable input passed to the operation handler. */
  payload: HandlerPayload<Handler>;
};

/**
 * The persisted state and result of a durable operation.
 *
 * Operation methods only wait for durable registration, not handler completion.
 * Call `getResult` later and narrow on `status` before reading a result or error.
 */
export type DurableOperationResult<Result> =
  | {
      /** No operation exists for the supplied idempotency key. */
      status: 'not_found';
    }
  | {
      /** The operation is waiting to run or retry. */
      status: 'pending';
      /** The number of attempts already started. */
      attempt: number;
      /** Unix timestamp in milliseconds when the next attempt becomes eligible. */
      nextAttemptAt: number;
      /** The previous attempt's error message, or null before the first attempt. */
      lastError: string | null;
    }
  | {
      /** The operation exhausted its attempts or failed with a non-retryable error. */
      status: 'failed';
      /** The number of attempts that were started. */
      attempt: number;
      /** The terminal error's serialized name and message. */
      error: { name: string; message: string };
    }
  | {
      /** The operation completed successfully. */
      status: 'completed';
      /** The handler's persisted return value. */
      result: Result;
    };

type OperationEvent = Extract<
  DurabilityLifecycleEvent,
  { entityKind: 'operation' }
>;

type TypedOperationEvent<
  Operation extends string,
  Result,
> = OperationEvent extends infer Event
  ? Event extends { outcome: 'completed' }
    ? Omit<Event, 'operation' | 'result'> & {
        operation: Operation;
        /** Present only when the instance was created with `includeResults: true`. */
        result?: Result;
      }
    : Event extends object
      ? Omit<Event, 'operation'> & { operation: Operation }
      : never
  : never;

/**
 * A lifecycle event for one of this instance's operations. Completed events
 * carry the handler's typed result only when the instance is created with
 * `includeResults: true`.
 */
export type DurabilityOperationEvent<Handlers extends HandlerMap> =
  | {
      [Operation in Extract<keyof Handlers, string>]: TypedOperationEvent<
        Operation,
        HandlerResult<Handlers[Operation]>
      >;
    }[Extract<keyof Handlers, string>]
  | Extract<DurabilityLifecycleEvent, { type: 'purged' }>;

/** Registers calls for one handler and reads their persisted results. */
export type DurableOperation<Handler> = ((
  input: DurableOperationInput<Handler>
) => Promise<void>) & {
  /** Reads the persisted state for an idempotency key. */
  getResult: (
    idempotencyKey: string
  ) => Promise<DurableOperationResult<HandlerResult<Handler>>>;
};

/**
 * Configuration for durable operations.
 *
 * @example
 * ```ts
 * const options: DurabilityOptions<typeof handlers> = {
 *   attemptTimeoutMs: 30_000,
 *   retries: { maxAttempts: 5 },
 *   methods: {
 *     resizeImage: { attemptTimeoutMs: 60_000, retries: { maxAttempts: 2 } },
 *   },
 * };
 * ```
 */
export type DurabilityOptions<Handlers extends HandlerMap = HandlerMap> =
  ExecutionPolicyOptions & {
    /** Execution policy overrides keyed by handler name. */
    methods?: Partial<
      Record<Extract<keyof NoInfer<Handlers>, string>, DurabilityMethodOptions>
    >;
    /** Receives best-effort, non-durable lifecycle metrics events. */
    onLifecycleEvent?: LifecycleHook;
    /**
     * Attach the handler's return value to `completed` lifecycle events.
     * Off by default so results never reach telemetry sinks unintentionally;
     * enable only when the sink is trusted with result content.
     */
    includeResults?: boolean;
  };

/**
 * Constructor configuration: the operation handlers, their policies, and
 * either a shared `scheduler` or the Durable Object `context` (plus scheduler
 * options) to create a private one.
 */
export type DurabilityConfig<Handlers extends HandlerMap = HandlerMap> =
  DurabilityOptions<Handlers> &
    SchedulerAttachment & {
      /** Operation handlers keyed by the method names they become. */
      handlers: Handlers;
    };

const assertOperation = (
  id: string,
  existingOperation: string,
  requestedOperation: string
): void => {
  if (existingOperation !== requestedOperation) {
    throw new DuplicateDurableCallError(
      id,
      existingOperation,
      requestedOperation
    );
  }
};

const callState = (
  call: CallRow
): Exclude<DurableOperationResult<unknown>, { status: 'not_found' }> => {
  if (call.status === 'pending') {
    return {
      status: 'pending',
      attempt: call.attempt,
      nextAttemptAt: call.next_attempt_at,
      lastError: call.last_error,
    };
  }
  if (call.status === 'failed') {
    return {
      status: 'failed',
      attempt: call.attempt,
      error: {
        name: call.last_error_name ?? 'Error',
        message: call.last_error ?? 'Durable operation failed',
      },
    };
  }
  if (call.result === null) {
    throw new Error(`Completed durable call "${call.id}" has no result`);
  }
  return { status: 'completed', result: deserialize(call.result) };
};

const callRecord = (sql: SqlStorage, call: CallRow): UntypedRecord => {
  const state = callState(call);
  return {
    id: call.id,
    name: call.operation,
    generation: call.generation_id,
    createdAt: call.created_at,
    state:
      state.status === 'pending'
        ? {
            ...state,
            memoized: memoizedKeys(
              sql,
              callsTable,
              call.id,
              call.generation_id
            ),
          }
        : state,
  };
};

const operationEntity = (call: CallRow): OperationLifecycleEntity => ({
  entityKind: 'operation',
  operation: call.operation,
  id: call.id,
  generation: call.generation_id,
});

class DurabilityCore<Handlers extends HandlerMap> {
  readonly #engine: Engine;
  readonly #handlers: Handlers;
  readonly #policy: PolicyResolver<MethodPolicy>;
  readonly #lifecycle: LifecycleHooks;
  readonly #emit: LifecycleHooks['emit'];
  readonly #includeResults: boolean;
  readonly #active = new Map<string, ActiveExecution>();
  readonly #executionStarts = new Map<string, Promise<void>>();

  constructor(config: DurabilityConfig<Handlers>) {
    const { handlers } = config;
    this.#engine = engineFor(config);
    this.#handlers = handlers;
    this.#policy = createPolicyResolver(
      config,
      config.methods ?? {},
      methodOptionsSchema
    );
    this.#lifecycle = createLifecycleHooks(
      (hook) => this.#engine.emitter(hook),
      config.onLifecycleEvent
    );
    this.#emit = this.#lifecycle.emit;
    this.#includeResults = config.includeResults ?? false;

    const reserved = new Set(
      Object.getOwnPropertyNames(DurabilityCore.prototype)
    );
    for (const operation of Object.keys(handlers)) {
      if (reserved.has(operation)) {
        throw new Error(`"${operation}" is reserved by Durability`);
      }
      const method = Object.assign(
        traceFunction(
          `Durability.${operation}`,
          (input: { id: string; payload: unknown }) =>
            this.#run({ ...input, operation })
        ),
        {
          getResult: traceFunction(
            `Durability.${operation}.getResult`,
            (idempotencyKey: string) =>
              this.#getResult(operation, idempotencyKey)
          ),
        }
      );
      Object.defineProperty(this, operation, {
        value: method,
        enumerable: true,
      });
    }

    this.#engine.migrateSchema('operations', durabilityOperationMigrations);
    this.#engine.migrateSchema('memos', durabilityMemoMigrations);
    this.#engine.register({
      table: callsTable,
      runDue: (now) => this.#runDue(now),
    });
    registerTrackedSource(this, this.#trackedSource());
  }

  /** Applies or reverts the `durability_calls` schema; see {@link migrateCapability}. */
  static migrate(
    context: Pick<DurableObjectState, 'storage'>,
    target?: string | null
  ): DurabilityMigrationResult {
    const result = migrateCapability(
      context.storage,
      'operations',
      durabilityOperationMigrations,
      target
    );
    deleteMemosOfDroppedTable(context.storage.sql, callsTable);
    return result;
  }

  /** Processes due operations and re-arms the physical alarm. */
  alarm(info?: AlarmInvocationInfo): Promise<void> {
    return this.#engine.alarm(info);
  }

  /**
   * Attaches a best-effort lifecycle hook alongside `onLifecycleEvent` and
   * returns a function that detaches it.
   */
  lifecycle(
    hook: (event: DurabilityOperationEvent<Handlers>) => void | Promise<void>
  ): () => void {
    return this.#lifecycle.add(hook as LifecycleHook);
  }

  /**
   * Destructively removes operation records created strictly before the
   * timestamp, regardless of status, and returns how many were removed.
   */
  async purgeBefore(before: number): Promise<number> {
    const count = await this.#engine.purge(
      callsTable,
      before,
      this.#active,
      'Durable operation was purged'
    );
    this.#emit({
      type: 'purged',
      entityKind: 'operation',
      timestamp: Date.now(),
      before,
      count,
    });
    return count;
  }

  #trackedSource(): UntypedSource {
    const sql = this.#engine.sql;
    return {
      get: async (id) => {
        const call = readCall(sql, id);
        return call && callRecord(sql, call);
      },
      list: async ({ limit, cursor, status, name }) => {
        const page = listCreated(sql, callsTable, 'operation', {
          limit,
          after: cursor,
          status,
          name,
        });
        const records = callRowSchema
          .array()
          .parse(page.rows)
          .map((call) => callRecord(sql, call));
        return page.next === undefined
          ? { records }
          : { records, cursor: page.next };
      },
      watch: (id, listener) =>
        this.#engine.watch(callsTable, id, () => listener()),
    };
  }

  async #runDue(now: number): Promise<void> {
    const due = z
      .object({ id: z.string() })
      .array()
      .parse(
        this.#engine.sql
          .exec(
            `SELECT id FROM durability_calls
             WHERE status = 'pending' AND next_attempt_at <= ?
             ORDER BY next_attempt_at ASC
             LIMIT 100`,
            now
          )
          .toArray()
      );
    // With liveTimers an active operation reschedules itself when it settles, so
    // skipping it keeps a long handler from blocking a concurrent alarm pass.
    const pending = this.#engine.liveTimers
      ? due.filter(({ id }) => !this.#active.has(id))
      : due;
    await this.#engine.runConcurrent(pending, async ({ id }) => {
      try {
        await this.#execute(id);
      } catch {
        return;
      }
    });
  }

  async #getResult(
    operation: string,
    idempotencyKey: string
  ): Promise<DurableOperationResult<unknown>> {
    const call = readCall(this.#engine.sql, idempotencyKey);
    if (!call) {
      return { status: 'not_found' };
    }
    assertOperation(idempotencyKey, call.operation, operation);
    return callState(call);
  }

  async #run(input: {
    id: string;
    operation: string;
    payload: unknown;
  }): Promise<void> {
    const payload = serialize(input.payload);
    const now = Date.now();
    const generation = crypto.randomUUID();
    const { inserted, winner } = await this.#engine.transaction((sql) => {
      const created =
        sql
          .exec(
            `INSERT INTO durability_calls
               (id, operation, payload, status, attempt, next_attempt_at,
                created_at, generation_id)
             VALUES (?, ?, ?, 'pending', 0, ?, ?, ?)
             ON CONFLICT(id) DO NOTHING
             RETURNING id`,
            input.id,
            input.operation,
            payload,
            now,
            now,
            generation
          )
          .toArray().length > 0;
      const current = readCall(sql, input.id);
      if (!current) {
        throw new Error(`Durable call "${input.id}" was not persisted`);
      }
      assertOperation(input.id, current.operation, input.operation);
      // Reuse is idempotent only for the same data; anything else is a caller
      // bug that would otherwise be dropped silently.
      if (!created && canonical(current.payload) !== canonical(payload)) {
        throw new DuplicateDurablePayloadError(input.id, input.operation);
      }
      return { inserted: created, winner: current };
    });
    if (!inserted) {
      return;
    }

    this.#emit({
      ...operationEntity(winner),
      type: 'registered',
      timestamp: now,
      attempt: 0,
    });
    // Start eagerly only when a permit is free; otherwise the reconciled alarm picks the row up.
    const eagerPermit = this.#engine.tryAcquirePermit();
    if (!eagerPermit) {
      return;
    }
    const onError = (error: unknown) =>
      reportFailure(
        'durability.background_execution.failed',
        { operation: input.operation, id: input.id },
        error
      );
    const background = this.#execute(input.id, eagerPermit)
      .then(() => this.#engine.scheduleNextAlarm())
      .catch(onError);
    this.#engine.waitUntil(background, onError);
  }

  /**
   * Looks up a call and starts it unless the same generation is already
   * running. Lookups for one ID are serialized so two concurrent callers
   * cannot both miss the active map while awaiting storage.
   */
  async #execute(id: string, eagerPermit?: () => void): Promise<void> {
    const previousStart = this.#executionStarts.get(id) ?? Promise.resolve();
    let releaseStart!: () => void;
    const currentStart = new Promise<void>((resolve) => {
      releaseStart = resolve;
    });
    this.#executionStarts.set(id, currentStart);
    let permitTransferred = false;

    try {
      await previousStart;
      const call = readCall(this.#engine.sql, id);
      if (!call || call.status !== 'pending') {
        return;
      }
      const running = this.#active.get(id);
      if (running?.generation === call.generation_id) {
        releaseStart();
        await running.settled;
        return;
      }

      const execution = this.#start(call, eagerPermit);
      permitTransferred = true;
      releaseStart();
      return execution;
    } finally {
      releaseStart();
      if (this.#executionStarts.get(id) === currentStart) {
        this.#executionStarts.delete(id);
      }
      if (!permitTransferred) {
        eagerPermit?.();
      }
    }
  }

  #start(call: CallRow, eagerPermit: (() => void) | undefined): Promise<void> {
    const policy = this.#policy(call.operation);
    const handler = this.#handlers[call.operation] as
      | ((durableCall: DurableCall<unknown>) => unknown)
      | undefined;
    return this.#engine.startExecution(
      {
        table: callsTable,
        key: call.id,
        row: call,
        label: `operation "${call.operation}"`,
        entity: operationEntity(call),
        active: this.#active,
        policy,
        emit: this.#emit,
        includeResult: this.#includeResults,
        exhaustedError: () =>
          new DurableAttemptsExhaustedError(
            'operation',
            call.operation,
            policy.maxAttempts
          ),
        timeoutError: () =>
          new DurableAttemptTimeoutError(
            call.operation,
            policy.attemptTimeoutMs
          ),
        invoke: (attempt, signal) => {
          if (!handler) {
            throw new NonRetryableError(
              `No handler registered for operation "${call.operation}"`
            );
          }
          return handler({
            id: call.id,
            operation: call.operation,
            payload: deserialize(call.payload),
            attempt,
            signal,
            memoize: createMemoize(this.#engine.sql, callsTable, {
              key: call.id,
              generation: call.generation_id,
              attempt,
            }),
          });
        },
        complete: (result, attempt) => {
          let serialized: string;
          try {
            serialized = serialize(result);
          } catch (error) {
            throw new DurableResultSerializationError(call.operation, error);
          }
          const timestamp = Date.now();
          const updated = updateAttempt(
            this.#engine.sql,
            callsTable,
            { key: call.id, generation: call.generation_id, attempt },
            `status = 'completed', result = ?, last_error = NULL,
             last_error_name = NULL, completed_at = ?`,
            serialized,
            timestamp
          );
          return updated ? timestamp : undefined;
        },
        isTerminal: (error) =>
          isErrorInstance(error, DurableResultSerializationError),
      },
      eagerPermit
    );
  }
}

type OperationMethods<Handlers extends HandlerMap> = {
  [Operation in keyof Handlers]: DurableOperation<Handlers[Operation]>;
};

/**
 * Alarm-backed, at-least-once durable operations for one Durable Object.
 *
 * Every handler becomes a typed method on the instance. Methods resolve once
 * the call is durably registered; read outcomes through `getResult`.
 *
 * Delivery is at least once: a handler may run again after a crash, timeout, or
 * eviction. The call `id` is a stable idempotency key; pass it to external
 * systems and make effects idempotent or reconcilable to reach an effectively
 * once outcome. See the README "Delivery and safety contract".
 *
 * @example
 * ```ts
 * class ImageJobs extends DurableObject<Env> {
 *   private readonly durability = new Durability({
 *     context: this.ctx,
 *     handlers: {
 *       resizeImage: async ({ id, payload }: DurableCall<{ imageId: string }>) =>
 *         this.env.IMAGES.resize(payload.imageId, { idempotencyKey: id }),
 *     },
 *   });
 *
 *   resize(imageId: string) {
 *     return this.durability.resizeImage({ id: `resize:${imageId}`, payload: { imageId } });
 *   }
 *
 *   alarm(info?: AlarmInvocationInfo) {
 *     return this.durability.alarm(info);
 *   }
 * }
 * ```
 */
export type Durability<Handlers extends HandlerMap> = DurabilityCore<Handlers> &
  OperationMethods<Handlers>;

/** Any `Durability` instance, without the index signature of `Durability<HandlerMap>`. */
export type AnyDurability = DurabilityCore<HandlerMap>;

export interface DurabilityConstructor {
  new <Handlers extends HandlerMap>(
    config: DurabilityConfig<Handlers>
  ): Durability<Handlers>;
  migrate: typeof DurabilityCore.migrate;
}

traceMethods(DurabilityCore.prototype, 'Durability');

// TypeScript cannot declare class members derived from a generic handler map,
// so the constructor is typed separately; the constructor installs and reserves
// every operation method at runtime.
export const Durability = DurabilityCore as unknown as DurabilityConstructor;

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;
};

Details

Follow one resize from the RPC call to a completed row in the object’s storage.

Durable state: Nothing is written yet.
In-flight / in memory: The helper and its engine exist for this instance and 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/image-jobs/worker.ts

packages/durability/src/durability.ts

packages/durability/src/scheduler.ts

The helper belongs to the object instance. Durability is constructed with this.ctx, the object’s DurableObjectState. It reads and writes only this object’s storage; another object has its own tables and its own alarm. Read these source lines. Durable state: Nothing is written yet. In memory: The helper and its engine exist for this instance and are rebuilt after every eviction.
Create tables in the object’s SQLite. The constructor applies migrations synchronously through ctx.storage.sql, then registers with the engine so alarm passes include this helper’s table. Read these source lines. Durable state: durability_calls and durability_memos exist in this object’s storage. In memory: The registration with the engine is in memory and repeats on each construction.
A Worker calls the object. The fetch handler reaches the object by name and calls resize(). The object turns that into a call on the generated resizeImage method. Read these source lines. Durable state: Nothing is written until the transaction commits. In memory: The RPC call is waiting on registration.
Insert the pending row. The insert runs inside a storage transaction. A repeated ID with the same handler and payload is a no-op; a different payload throws instead of being dropped. Read these source lines. Durable state: A pending row with attempt 0 commits with the transaction. In memory: The serialized payload is held only until the write.
Set the object alarm in the same transaction. Before the transaction commits, the engine sets the object’s one alarm to the earliest pending row, but no sooner than alarmMinDelayMs (15 seconds by default). A row cannot commit without its alarm. Read these source lines. Durable state: The alarm is stored with the row it serves. In memory: Work due sooner than the alarm uses an in-memory timer that only this instance has.
Start now if a permit is free. The RPC call resolves after registration. If the engine has a free permit it starts the attempt in the background and passes the promise to ctx.waitUntil; otherwise the next timer or alarm pass picks the row up. Read these source lines. Durable state: The pending row and the alarm are already committed. In memory: The permit and the background promise exist only in this instance.
Claim the attempt before the handler runs. The engine increments the stored attempt count with a conditional UPDATE. If the attempt limit is reached, the operation fails instead of running. Read these source lines. Durable state: The attempt count is stored before any external effect can happen. In memory: The attempt is registered in the running-attempts map so the same ID cannot run twice in this instance.
The handler calls the image service. The handler runs outside the storage transaction and forwards the operation ID as the idempotency key. Read these source lines. Durable state: The row stays pending while the request is in flight. In memory: The fetch, its response, and the attempt timeout.
Store the result. Completion and memo cleanup run in one synchronous transaction. The UPDATE only matches the same generation and attempt, so a late attempt cannot overwrite a newer one. Read these source lines. Durable state: The row is completed with its JSON result. In memory: The permit is released and the running entry removed.

Evict the object mid-attempt

The object is evicted during the request. Assume the image service accepts the resize, but the runtime evicts the object before the result is stored. Everything Durability kept in memory is gone. Read these source lines. Durable state: The row is still pending with attempt 1, and the object alarm is still set. In memory: Permits, the running-attempts map, timers, and the fetch promise are lost.
The alarm starts a new instance. When the alarm fires, the runtime creates a new instance, which constructs a new Durability helper, and calls alarm(). Without this delegation, pending rows would not run again after an eviction. Read these source lines. Durable state: The alarm, the row, and its attempt count. In memory: A fresh helper with all permits free.
Read due rows from SQLite. The alarm pass selects pending rows whose retry time has passed. The interrupted operation is still pending, so it is selected. Read these source lines. Durable state: Rows are read, not changed. In memory: The due list for this pass.
Claim attempt 2 for the same ID. The interrupted attempt still counts toward maxAttempts. The new attempt receives the same operation ID and payload from the row. Read these source lines. Durable state: The stored attempt count becomes 2. In memory: A new handler call replaces the lost one.
The image service must deduplicate. The service already performed the first resize. Only its handling of the repeated Idempotency-Key prevents a second effect; nothing in the object can prove what the service committed. Read these source lines. Durable state: Exactly-once effects require the receiver’s own deduplication record. In memory: No in-memory state from the first attempt survived.

Work within alarm limits

An alarm pass throws. The runtime retries a failing alarm() only a few times and then drops it, which would strand pending rows. The engine catches the error and sets its own 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.
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. Read these source lines. Durable state: The next alarm is stored before the pass returns. In memory: Running attempts continue in this instance if it stays alive.
Pending work always has an alarm. Registration, retry scheduling, terminal failures, and purges go through this transaction, so a committed pending row always has the object alarm set for it. The in-memory timer is set only after the commit. Read these source lines. Durable state: Rows and the alarm commit together or not at all. In memory: The in-memory timer for work due before the alarm.

Behind the scenes

Durable state: durability_calls holds one row per operation ID: handler name, serialized payload, generation, status, attempt count, retry time, error, and result. Shared durability_memos holds intermediate memoized results. Pending-time and creation-time indexes support scheduling and purge.

Registration serializes the payload and inserts a pending row in a storage transaction. An existing ID must match both handler and canonical payload. The transaction also reconciles the physical alarm.
A new operation starts eagerly only if the shared scheduler has an immediately available permit. Otherwise the persisted due row waits for a timer/alarm pass.
Before calling the handler, the engine durably claims and increments its attempt. An in-memory active lock prevents overlapping same-instance execution.
Successful settlement conditionally stores the JSON result. Failure either records the next retry time or a terminal failed state. Generation/attempt checks prevent a late old execution from overwriting a purged-and-reused ID.

After eviction: the row survives, but locks and running promises do not. A pending handler can run again with the same ID. A remote effect may already have committed; receiver-side idempotency is still necessary. Abort is cooperative: a handler that ignores its signal holds the live permit until it actually settles.

Lifetime: completed and failed records stay indefinitely until destructive purgeBefore(). Purging also removes deduplication protection and memos.

Implementation: durability.ts (DurabilityCore), storage.ts (attempt claims and conditional updates).

Durable operations

import { DurableObject } from 'cloudflare:workers';
import { Durability, type DurableCall } from 'durability';

type ResizeInput = { imageId: string };

export class ImageJobs extends DurableObject<Env> {
  private readonly durability = new Durability({
    context: this.ctx,
    handlers: {
      resizeImage: async ({ id, payload }: DurableCall<ResizeInput>) =>
        this.env.IMAGES.resize(payload.imageId, { idempotencyKey: id }),
    },
    retries: { maxAttempts: 3 },
    methods: { resizeImage: { attemptTimeoutMs: 60_000 } },
  });

  resize(imageId: string) {
    return this.durability.resizeImage({
      id: `resize:${imageId}`,
      payload: { imageId },
    });
  }

  result(imageId: string) {
    return this.durability.resizeImage.getResult(`resize:${imageId}`);
  }

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

Every handler becomes a method on the instance. Handler names that collide with instance members (alarm, purgeBefore) are rejected at construction.

Reading results

const result = await this.durability.resizeImage.getResult(`resize:${imageId}`);

switch (result.status) {
  case 'not_found':
    break;
  case 'pending':
    console.log(result.attempt, result.nextAttemptAt, result.lastError);
    break;
  case 'failed':
    console.error(result.error.name, result.error.message);
    break;
  case 'completed':
    console.log(result.result);
    break;
}

The completed result type is inferred from the handler. Looking up a key that belongs to another operation throws DuplicateDurableCallError. Payloads and results must be JSON-serializable; as with JSON.stringify, undefined properties are dropped and undefined array elements become null.

An ID stays bound to its operation and payload. Registering it again with the same operation and the same data (object key order does not matter; array order and values do) is a no-op, which is what makes retries safe. Registering it with another operation throws DuplicateDurableCallError, and with a different payload DuplicateDurablePayloadError, instead of silently dropping the new call.