import { z } from 'zod';
import { traceFunction, traceMethods } from './dev.js';
import {
DurableRetryPolicyError,
FanoutEnqueueError,
FanoutSettlementError,
FanoutTimeoutError,
reportFailure,
serializeError,
type SerializedError,
} from './errors.js';
import {
createLifecycleHooks,
type DurabilityLifecycleEvent,
type FanoutLifecycleEntity,
type LifecycleHook,
type LifecycleHooks,
} from './lifecycle.js';
import { RoutingLoad, type EnqueueResult, type LoadSnapshot } from './load.js';
import type { MessageWrite, StoredMessage } from './stored-message.js';
import {
createMemoize,
deleteMemos,
deleteMemosOfDroppedTable,
type DurableMemoize,
} from './memoize.js';
import {
durabilityFanoutMigrations,
durabilityMemoMigrations,
migrateCapability,
} from './migrations.js';
import {
assertPositiveInteger,
createPolicyResolver,
methodOptionsSchema,
type AttemptPolicy,
type ExecutionPolicyOptions,
} from './policy.js';
import {
engineFor,
type Engine,
type SchedulerAttachment,
} from './scheduler.js';
import { deserialize, serialize } from './serialization.js';
import {
claimAttempt,
deleteAttempt,
updateAttempt,
type AttemptRef,
type DurableTable,
} from './storage.js';
const fanoutMessageRowSchema = z.object({
key: z.string(),
id: z.string(),
payload: z.string(),
targets: z.string(),
remaining: z.number().int().nonnegative(),
seq: z.number().int().nonnegative(),
created_at: z.number().int().nonnegative(),
generation_id: z.string(),
});
type FanoutMessageRow = z.infer<typeof fanoutMessageRowSchema>;
const deliveryRowSchema = z.object({
id: z.string(),
message_key: z.string(),
target_id: z.string(),
seq: z.number().int().nonnegative(),
status: z.literal('pending'),
phase: z.enum(['delivery', 'dead_letter']),
attempt: z.number().int().nonnegative(),
next_attempt_at: z.number().int().nonnegative(),
last_error: z.string().nullable(),
last_error_name: z.string().nullable(),
created_at: z.number().int().nonnegative(),
generation_id: z.string(),
dead_lettered_at: z.number().int().nonnegative().nullable(),
dead_letter_reason: z.enum(['explicit', 'exhausted']).nullable(),
});
type DeliveryRow = z.infer<typeof deliveryRowSchema>;
const deliveriesTable: DurableTable = {
name: 'durability_fanout_deliveries',
key: 'id',
};
const countRowSchema = z.object({ count: z.number().int().nonnegative() });
const seqRowSchema = z.object({ seq: z.number().int().nonnegative() });
const readManifest = (
sql: SqlStorage,
key: string
): FanoutMessageRow | undefined =>
fanoutMessageRowSchema
.optional()
.parse(
sql
.exec('SELECT * FROM durability_fanout_messages WHERE key = ?', key)
.toArray()[0]
);
const attemptOf = (row: DeliveryRow): AttemptRef => ({
key: row.id,
generation: row.generation_id,
attempt: row.attempt,
});
export type FanoutInput<Body = unknown> = {
id?: string;
body: Body;
/** Defaults to id. Redrive uses the DLQ entry ID to register fresh attempts. */
deduplicationKey?: string;
};
/** Each target receives its own settlement capability for each message. */
export type DurabilityFanoutMessage<Body = unknown> = {
id: string;
deliveryId: string;
target: string;
body: Body;
attempt: number;
enqueuedAt: number;
/**
* Persists a callback's result by key for this message and target, so a
* retried delivery reuses it. Deleted when the delivery is acked or dead-lettered.
*/
memoize: DurableMemoize;
ack(): Promise<void>;
retry(delayMs?: number): Promise<void>;
deadLetter(): Promise<void>;
};
export type FanoutTarget<Body> = ExecutionPolicyOptions &
(
| {
deliver(
messages: DurabilityFanoutMessage<Body>[]
): unknown | Promise<unknown>;
/**
* Acks every message the consumer leaves unsettled when `deliver`
* returns without throwing, in one transaction instead of one call
* per message. Explicit `retry()` and `deadLetter()` still apply.
*/
ackOnReturn?: boolean;
storage?: never;
}
| { storage: MessageWrite<Body>; deliver?: never; ackOnReturn?: never }
);
export type FanoutEnqueueOptions = {
/** Omit for all static targets. Redrive specifies only the failed target. */
targets?: readonly string[];
};
export type DurabilityFanoutConfig<Body> = SchedulerAttachment &
ExecutionPolicyOptions & {
targets: Record<string, FanoutTarget<Body>>;
routing?: RoutingLoad;
/** Receives terminal failures. Without it, terminal deliveries are dropped. */
dlq?: MessageWrite<Body>;
maxBatchSize?: number;
/**
* Holds first deliveries for up to this many milliseconds so messages
* arriving together share a batch. Defaults to 0 (deliver immediately).
*/
batchDelayMs?: number;
onLifecycleEvent?: LifecycleHook;
};
type DeadLetterReason = NonNullable<StoredMessage['failure']>['reason'];
type Claim = { row: DeliveryRow; message: FanoutMessageRow };
type Settlement = {
claim: Claim;
closed: boolean;
finished?: boolean;
completed?: boolean;
outcome?: 'ack' | 'retry' | 'deadLetter';
pending?: Promise<void>;
};
const infrastructureRetryMs = 60_000;
const entity = (claim: Claim): FanoutLifecycleEntity => ({
entityKind: 'fanout_delivery',
id: claim.message.id,
target: claim.row.target_id,
generation: claim.row.generation_id,
});
const deliveryIdentity = (row: DeliveryRow): string =>
JSON.stringify([row.generation_id, row.target_id]);
/** A lifecycle event for one of this instance's deliveries. */
export type DurabilityFanoutEvent =
| Extract<DurabilityLifecycleEvent, { entityKind: 'fanout_delivery' }>
| Extract<DurabilityLifecycleEvent, { type: 'purged' }>;
/**
* A composable DO capability: durable, independent delivery to static named
* targets. Share a scheduler with operations/alarms on the same object.
* Register targets under stable IDs; removing a target pauses its pending work.
*/
export class DurabilityFanout<Body = unknown> {
readonly #engine: Engine;
readonly #routing: RoutingLoad;
readonly #targets: Map<string, FanoutTarget<Body>>;
readonly #policies: Map<string, AttemptPolicy>;
readonly #dlq: MessageWrite<Body> | undefined;
readonly #maxBatchSize: number;
readonly #batchDelayMs: number;
readonly #lifecycle: LifecycleHooks;
readonly #emit: LifecycleHooks['emit'];
readonly #running = new Map<string, Promise<void>>();
/**
* Cached pending-delivery count. Counting the table on every enqueue costs
* O(backlog), which made a shard that fell behind fall further behind. It is
* hydrated once and kept exact by inserts and removals; a failed transaction
* may leave it wrong, so it is discarded and recounted on the next read.
*/
#pending: number | undefined;
constructor(config: DurabilityFanoutConfig<Body>) {
this.#maxBatchSize = config.maxBatchSize ?? 10;
assertPositiveInteger('maxBatchSize', this.#maxBatchSize);
this.#batchDelayMs = config.batchDelayMs ?? 0;
if (!Number.isSafeInteger(this.#batchDelayMs) || this.#batchDelayMs < 0) {
throw new RangeError('batchDelayMs must be a non-negative integer');
}
this.#targets = new Map(
Object.entries(config.targets).map(([id, target]) => [id, { ...target }])
);
if (this.#targets.size === 0) {
throw new TypeError('Fanout requires at least one static target');
}
this.#policies = new Map();
for (const [id, target] of this.#targets) {
if (
!id ||
(typeof target.deliver !== 'function' &&
typeof target.storage !== 'function')
) {
throw new TypeError(`Invalid fanout target "${id}"`);
}
this.#policies.set(
id,
createPolicyResolver(config, { [id]: target }, methodOptionsSchema)(id)
);
}
this.#engine = engineFor(config);
this.#routing = config.routing ?? new RoutingLoad();
this.#dlq = config.dlq;
this.#lifecycle = createLifecycleHooks(
(hook) => this.#engine.emitter(hook),
config.onLifecycleEvent
);
this.#emit = this.#lifecycle.emit;
this.#engine.migrateSchema('fanout', durabilityFanoutMigrations);
this.#engine.migrateSchema('memos', durabilityMemoMigrations);
this.#engine.register({
table: deliveriesTable,
runDue: () => this.#runDue(),
});
}
static migrate(
context: Pick<DurableObjectState, 'storage'>,
target?: string | null
) {
const result = migrateCapability(
context.storage,
'fanout',
durabilityFanoutMigrations,
target
);
deleteMemosOfDroppedTable(context.storage.sql, deliveriesTable);
return result;
}
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: DurabilityFanoutEvent) => void | Promise<void>
): () => void {
return this.#lifecycle.add(hook as LifecycleHook);
}
async load(): Promise<LoadSnapshot> {
return this.#routing.snapshot(this.#pendingCount());
}
/**
* Atomically accepts the whole batch, including its frozen target set.
* Definite input conflicts return success:false. Storage/commit uncertainty
* throws: a transport failure is never represented as definite rejection.
* IDs deduplicate while their registration is retained, not forever.
*/
async enqueue(
messages: FanoutInput<Body> | FanoutInput<Body>[],
options: FanoutEnqueueOptions = {}
): Promise<EnqueueResult<string[]>> {
const finish = this.#routing.begin('inbound');
let entries: Array<{ key: string; id: string; payload: string }>;
let targets: string[];
try {
targets = [...(options.targets ?? this.#targets.keys())].sort();
if (
targets.length === 0 ||
new Set(targets).size !== targets.length ||
targets.some((id) => !this.#targets.has(id))
) {
throw new FanoutEnqueueError(
'Enqueue targets must be distinct configured target IDs'
);
}
entries = (Array.isArray(messages) ? messages : [messages]).map(
(message) => {
const id = message.id ?? crypto.randomUUID();
const key = message.deduplicationKey ?? id;
if (
typeof id !== 'string' ||
!id ||
typeof key !== 'string' ||
!key
) {
throw new FanoutEnqueueError(
'Message IDs and deduplication keys must be non-empty strings'
);
}
return { key, id, payload: serialize(message.body) };
}
);
} catch (error) {
finish();
return {
success: false,
error: serializeError(error),
load: await this.load(),
};
}
let registered: Claim[];
try {
registered = await this.#transaction((sql) => {
const claims: Claim[] = [];
for (const entry of entries) {
const existing = readManifest(sql, entry.key);
if (existing) {
if (
existing.id !== entry.id ||
existing.payload !== entry.payload ||
existing.targets !== JSON.stringify(targets)
) {
throw new FanoutEnqueueError(
`Deduplication key "${entry.key}" already identifies different work`
);
}
continue;
}
const { seq } = seqRowSchema.parse(
sql
.exec(
`UPDATE durability_fanout_seq
SET next_seq = next_seq + 1
WHERE id = 1
RETURNING next_seq - 1 AS seq`
)
.toArray()[0]
);
const message: FanoutMessageRow = {
...entry,
targets: JSON.stringify(targets),
remaining: targets.length,
seq,
created_at: Date.now(),
generation_id: crypto.randomUUID(),
};
sql.exec(
`INSERT INTO durability_fanout_messages
(key, id, payload, targets, remaining, seq, created_at,
generation_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
message.key,
message.id,
message.payload,
message.targets,
message.remaining,
message.seq,
message.created_at,
message.generation_id
);
for (const target of targets) {
const row: DeliveryRow = {
id: JSON.stringify([entry.key, target]),
message_key: entry.key,
target_id: target,
seq: message.seq,
generation_id: message.generation_id,
created_at: message.created_at,
status: 'pending',
phase: 'delivery',
attempt: 0,
next_attempt_at: this.#firstAttemptAt(message.created_at),
last_error: null,
last_error_name: null,
dead_lettered_at: null,
dead_letter_reason: null,
};
const inserted =
sql
.exec(
`INSERT INTO durability_fanout_deliveries
(id, message_key, target_id, seq, status, phase, attempt,
next_attempt_at, created_at, generation_id)
VALUES (?, ?, ?, ?, 'pending', 'delivery', 0, ?, ?, ?)
ON CONFLICT(id) DO NOTHING
RETURNING id`,
row.id,
row.message_key,
row.target_id,
row.seq,
row.next_attempt_at,
row.created_at,
row.generation_id
)
.toArray().length > 0;
if (!inserted) {
throw new Error('Fanout manifest and delivery state disagree');
}
this.#adjustPending(1);
claims.push({ row, message });
}
}
return claims;
});
} catch (error) {
if (!(error instanceof FanoutEnqueueError)) {
throw error;
}
finish();
return {
success: false,
error: serializeError(error),
load: await this.load(),
};
} finally {
finish();
}
for (const claim of registered) {
this.#emit({
...entity(claim),
type: 'registered',
timestamp: claim.message.created_at,
attempt: 0,
});
}
const load = await this.load();
// Delayed first attempts are not due yet; the scheduler wakes for them.
if (registered.length > 0 && this.#batchDelayMs === 0) {
const onError = (error: unknown) =>
reportFailure('durability.fanout.failed', {}, error);
const work = this.#runDue()
.then(() => this.#engine.scheduleNextAlarm())
.catch(onError);
this.#engine.waitUntil(work, onError);
}
return { success: true, value: entries.map(({ id }) => id), load };
}
// Rounding up to a shared boundary makes every message in the window due together.
#firstAttemptAt(createdAt: number): number {
if (this.#batchDelayMs === 0) {
return createdAt;
}
return Math.ceil(createdAt / this.#batchDelayMs) * this.#batchDelayMs;
}
async #runDue(): Promise<void> {
const targets = z
.object({ target_id: z.string() })
.array()
.parse(
this.#engine.sql
.exec(
`SELECT DISTINCT target_id FROM durability_fanout_deliveries
ORDER BY target_id`
)
.toArray()
)
.map((row) => row.target_id);
// Each target advances independently; a missing or slow target cannot hide
// another target behind a global LIMIT. Each pass processes one bounded batch.
await Promise.all(
targets.map((id) => {
const running = this.#running.get(id);
if (running) {
return running;
}
const work = this.#runTarget(id).finally(() =>
this.#running.delete(id)
);
this.#running.set(id, work);
return work;
})
);
}
async #runTarget(id: string): Promise<void> {
const release = await this.#engine.acquirePermit();
try {
const due = this.#listDue(id, Date.now());
if (due.length === 0) {
return;
}
const target = this.#targets.get(id);
const policy = this.#policies.get(id);
if (!target || !policy) {
reportFailure(
'durability.fanout.target_missing',
{ target: id },
new Error(`Target "${id}" is no longer configured; retaining work`)
);
await this.#postpone(due);
return;
}
// One transaction claims the whole batch: per-message transactions made the
// pass cost O(batch) commits, which capped a shard's throughput.
const { claims, terminal } = await this.#transaction((sql) => {
const claimed: Claim[] = [];
const exhausted: Claim[] = [];
const deadline = Date.now() + policy.attemptTimeoutMs;
for (const row of due) {
const message = readManifest(sql, row.message_key);
if (!message || message.generation_id !== row.generation_id) {
throw new Error('Fanout delivery is missing its message manifest');
}
if (
row.phase === 'dead_letter' ||
row.attempt >= policy.maxAttempts
) {
exhausted.push({ row, message });
continue;
}
const attempt = claimAttempt(
sql,
deliveriesTable,
row.id,
row.generation_id,
policy.maxAttempts,
deadline
);
if (attempt === undefined) {
continue;
}
claimed.push({ row: { ...row, attempt }, message });
}
return { claims: claimed, terminal: exhausted };
});
for (const claim of terminal) {
// eslint-disable-next-line no-await-in-loop
await this.#deadLetter(
claim,
claim.row.dead_letter_reason ?? 'exhausted'
);
}
if (claims.length > 0) {
await this.#deliver(target, claims, policy);
}
} finally {
release();
}
}
async #postpone(rows: DeliveryRow[]): Promise<void> {
await this.#transaction((sql) => {
for (const row of rows) {
updateAttempt(
sql,
deliveriesTable,
attemptOf(row),
'next_attempt_at = ?',
Date.now() + infrastructureRetryMs
);
}
});
}
async #deliver(
target: FanoutTarget<Body>,
claims: Claim[],
policy: AttemptPolicy
): Promise<void> {
const startedAt = Date.now();
const finish = this.#routing.begin('outbound');
const states: Settlement[] = claims.map((claim) => ({
claim,
closed: false,
}));
const messages = states.map((state) =>
this.#message(state, policy, startedAt)
);
for (const claim of claims) {
this.#emit({
...entity(claim),
type: 'attempt_started',
timestamp: startedAt,
attempt: claim.row.attempt,
});
}
let failure: unknown = new Error(
'Consumer returned without settling the message'
);
try {
await this.#deadline(async () => {
let invocationFailed = false;
let invocationError: unknown;
try {
if (target.deliver) {
await target.deliver(messages);
} else {
const writes = await Promise.allSettled(
messages.map(async (message) => {
await target.storage({
id: message.deliveryId,
messageId: message.id,
target: message.target,
body: message.body,
enqueuedAt: message.enqueuedAt,
storedAt: Date.now(),
attempts: message.attempt,
});
await message.ack();
})
);
const rejected = writes.find(
(write) => write.status === 'rejected'
);
if (rejected?.status === 'rejected') {
throw rejected.reason;
}
}
} catch (error) {
invocationFailed = true;
invocationError = error;
}
for (const state of states) {
state.closed = true;
}
if (!invocationFailed && target.ackOnReturn) {
this.#ackUnsettled(states, startedAt);
}
// A callback failure must not bypass already-started settlements.
// The outer deadline still bounds a hung storage or RPC operation.
const settlements = await Promise.allSettled(
states.map((state) => state.pending)
);
if (invocationFailed) {
throw invocationError;
}
const rejected = settlements.find(
(settlement) => settlement.status === 'rejected'
);
if (rejected?.status === 'rejected') {
throw rejected.reason;
}
}, policy.attemptTimeoutMs);
} catch (error) {
failure = error;
} finally {
for (const state of states) {
state.closed = true;
}
finish();
}
for (const state of states) {
if (state.completed) {
continue;
}
if (state.outcome !== undefined) {
// Pending actions retain their persisted claim deadline. Failed actions
// may have committed despite throwing; consult state before retrying.
if (!state.finished) {
continue;
}
const current = deliveryRowSchema
.optional()
.parse(
this.#engine.sql
.exec(
'SELECT * FROM durability_fanout_deliveries WHERE id = ?',
state.claim.row.id
)
.toArray()[0]
);
if (
!current ||
current.generation_id !== state.claim.row.generation_id ||
current.attempt !== state.claim.row.attempt ||
current.phase === 'dead_letter'
) {
continue;
}
}
state.outcome = 'retry';
// eslint-disable-next-line no-await-in-loop
await this.#retry(
state.claim,
policy,
undefined,
serializeError(failure),
startedAt
);
}
}
/** Acks every unsettled message of a batch in one transaction. */
#ackUnsettled(states: Settlement[], startedAt: number): void {
const unsettled = states.filter((state) => state.outcome === undefined);
if (unsettled.length === 0) {
return;
}
const removed = this.#transaction((sql) => {
const ids = new Set<string>();
for (const { claim } of unsettled) {
if (this.#removeDelivery(sql, claim.row)) {
ids.add(claim.row.id);
}
}
return ids;
});
for (const state of unsettled) {
state.outcome = 'ack';
state.pending = removed
.then((ids) => {
// A row a later attempt already reclaimed stays with that attempt.
if (!ids.has(state.claim.row.id)) {
return undefined;
}
state.completed = true;
this.#emit({
...entity(state.claim),
type: 'attempt_settled',
timestamp: Date.now(),
attempt: state.claim.row.attempt,
durationMs: Math.max(0, Date.now() - startedAt),
outcome: 'completed',
});
return undefined;
})
.finally(() => {
state.finished = true;
});
void state.pending.catch(() => undefined);
}
}
#message(
state: Settlement,
policy: AttemptPolicy,
startedAt: number
): DurabilityFanoutMessage<Body> {
const { row, message } = state.claim;
const identity = {
deliveryId: deliveryIdentity(row),
target: row.target_id,
};
const settle = (
outcome: NonNullable<Settlement['outcome']>,
action: () => Promise<void>
): Promise<void> => {
if (state.closed || state.outcome !== undefined) {
return Promise.reject(
new FanoutSettlementError(message.id, state.outcome)
);
}
state.outcome = outcome;
state.pending = action()
.then(() => {
state.completed = true;
return undefined;
})
.finally(() => {
state.finished = true;
});
void state.pending.catch(() => undefined);
return state.pending;
};
return {
id: message.id,
...identity,
body: deserialize(message.payload) as Body,
enqueuedAt: message.created_at,
attempt: row.attempt,
memoize: createMemoize(this.#engine.sql, deliveriesTable, attemptOf(row)),
ack: traceFunction(
'DurabilityFanoutMessage.ack',
() =>
settle('ack', async () => {
await this.#remove(row);
this.#emit({
...entity(state.claim),
type: 'attempt_settled',
timestamp: Date.now(),
attempt: row.attempt,
durationMs: Math.max(0, Date.now() - startedAt),
outcome: 'completed',
});
}),
identity
),
retry: traceFunction(
'DurabilityFanoutMessage.retry',
(delayMs?: number) => {
if (
delayMs !== undefined &&
(!Number.isSafeInteger(delayMs) ||
delayMs < 0 ||
!Number.isSafeInteger(Date.now() + delayMs))
) {
return Promise.reject(
new RangeError(
'delayMs must produce a non-negative safe-integer timestamp'
)
);
}
return settle('retry', () =>
this.#retry(
state.claim,
policy,
delayMs,
{ name: 'Retry', message: 'Retried by consumer' },
startedAt
)
);
},
identity
),
deadLetter: traceFunction(
'DurabilityFanoutMessage.deadLetter',
() =>
settle('deadLetter', () => this.#deadLetter(state.claim, 'explicit')),
identity
),
};
}
async #remove(row: DeliveryRow): Promise<void> {
await this.#transaction((sql) => {
if (!this.#removeDelivery(sql, row)) {
throw new FanoutSettlementError(row.id);
}
});
}
/** Deletes a delivery the attempt still owns and releases its manifest share. */
#removeDelivery(sql: SqlStorage, row: DeliveryRow): boolean {
if (!deleteAttempt(sql, deliveriesTable, attemptOf(row))) {
return false;
}
deleteMemos(sql, deliveriesTable, row.id);
this.#adjustPending(-1);
sql.exec(
`UPDATE durability_fanout_messages
SET remaining = remaining - 1
WHERE key = ?`,
row.message_key
);
sql.exec(
`DELETE FROM durability_fanout_messages
WHERE key = ? AND remaining = 0`,
row.message_key
);
return true;
}
async #retry(
claim: Claim,
policy: AttemptPolicy,
override: number | undefined,
error: SerializedError,
startedAt: number
): Promise<void> {
const { row } = claim;
if (row.attempt >= policy.maxAttempts) {
await this.#deadLetter(claim, 'exhausted', error);
return;
}
let nextAttemptAt: number;
try {
const rawDelay = override ?? policy.delay(row.attempt);
const delay = Math.round(rawDelay);
nextAttemptAt = Date.now() + delay;
if (
!Number.isFinite(rawDelay) ||
rawDelay < 0 ||
!Number.isSafeInteger(nextAttemptAt)
) {
throw new DurableRetryPolicyError('fanout');
}
} catch (caught) {
await this.#deadLetter(claim, 'exhausted', serializeError(caught));
return;
}
await this.#transaction((sql) => {
const updated = updateAttempt(
sql,
deliveriesTable,
attemptOf(row),
'next_attempt_at = ?, last_error = ?, last_error_name = ?',
nextAttemptAt,
error.message,
error.name
);
if (!updated) {
throw new FanoutSettlementError(row.id);
}
});
this.#emit({
...entity(claim),
type: 'attempt_settled',
timestamp: Date.now(),
attempt: row.attempt,
durationMs: Math.max(0, Date.now() - startedAt),
outcome: 'retry_scheduled',
nextAttemptAt,
error,
});
}
async #deadLetter(
claim: Claim,
reason: DeadLetterReason,
failure?: SerializedError
): Promise<void> {
const { row, message } = claim;
const timestamp = row.dead_lettered_at ?? Date.now();
const error =
failure ??
(row.last_error === null
? null
: { name: row.last_error_name ?? 'Error', message: row.last_error });
// Without a DLQ nothing could ever accept this delivery, and retrying the
// dead-letter step would keep the object awake forever, so it is dropped.
if (!this.#dlq) {
await this.#remove(row);
this.#emitTerminal(claim, reason, error);
return;
}
// Persist intent BEFORE external I/O: eviction or timeout must never turn
// a terminal decision into another consumer delivery.
await this.#transaction((sql) => {
const updated = updateAttempt(
sql,
deliveriesTable,
attemptOf(row),
`phase = 'dead_letter', dead_lettered_at = ?, dead_letter_reason = ?,
next_attempt_at = ?, last_error = ?, last_error_name = ?`,
timestamp,
reason,
Date.now() + infrastructureRetryMs,
error?.message ?? null,
error?.name ?? null
);
if (!updated) {
throw new FanoutSettlementError(row.id);
}
});
try {
const stored: StoredMessage<Body> = {
id: deliveryIdentity(row),
messageId: message.id,
target: row.target_id,
body: deserialize(message.payload) as Body,
enqueuedAt: message.created_at,
storedAt: timestamp,
attempts: row.attempt,
failure: { reason, error },
};
await this.#deadline(() => this.#dlq!(stored), infrastructureRetryMs);
await this.#remove(row);
this.#emitTerminal(claim, reason, error);
} catch (caught) {
reportFailure(
'durability.fanout.dead_letter_failed',
{ id: message.id, target: row.target_id },
caught
);
}
}
#emitTerminal(
claim: Claim,
reason: DeadLetterReason,
error: SerializedError | null
): void {
this.#emit({
...entity(claim),
type: 'terminal',
timestamp: Date.now(),
attempt: claim.row.attempt,
reason: reason === 'explicit' ? 'dead_lettered' : 'attempts_exhausted',
error: error ?? {
name: 'DeadLetter',
message: 'Dead-lettered by consumer',
},
});
}
async #transaction<T>(write: (sql: SqlStorage) => T): Promise<T> {
try {
return await this.#engine.transaction(write);
} catch (error) {
this.#pending = undefined;
throw error;
}
}
#pendingCount(): number {
this.#pending ??= countRowSchema.parse(
this.#engine.sql
.exec(
`SELECT COUNT(*) AS count FROM durability_fanout_deliveries
WHERE status = 'pending'`
)
.toArray()[0]
).count;
return this.#pending;
}
#adjustPending(delta: number): void {
if (this.#pending !== undefined) {
this.#pending += delta;
}
}
/**
* First deliveries (attempt = 0) in sequence, then retries: two indexed
* queries instead of one sort over every pending row.
*/
#listDue(target: string, now: number): DeliveryRow[] {
const first = deliveryRowSchema.array().parse(
this.#engine.sql
.exec(
`SELECT * FROM durability_fanout_deliveries
WHERE target_id = ? AND phase = 'delivery' AND status = 'pending'
AND attempt = 0 AND next_attempt_at <= ?
ORDER BY seq ASC
LIMIT ?`,
target,
now,
this.#maxBatchSize
)
.toArray()
);
if (first.length >= this.#maxBatchSize) {
return first;
}
const retries = deliveryRowSchema.array().parse(
this.#engine.sql
.exec(
`SELECT * FROM durability_fanout_deliveries
WHERE target_id = ? AND status = 'pending'
AND attempt > 0 AND next_attempt_at <= ?
ORDER BY seq ASC
LIMIT ?`,
target,
now,
this.#maxBatchSize - first.length
)
.toArray()
);
return [...first, ...retries];
}
async #deadline(
run: () => Promise<unknown>,
timeoutMs: number
): Promise<void> {
let timer: ReturnType<typeof setTimeout> | undefined;
try {
await Promise.race([
Promise.resolve().then(run),
new Promise<never>((_resolve, reject) => {
timer = setTimeout(
() => reject(new FanoutTimeoutError(timeoutMs)),
timeoutMs
);
}),
]);
} finally {
if (timer !== undefined) {
clearTimeout(timer);
}
}
}
}
traceMethods(DurabilityFanout.prototype, 'DurabilityFanout');
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;
};