Add an optional durable outbound delivery backlog #1376

Motivation

A delayed delivery or large fanout needs an authoritative owner for both prepared payload and remaining recipients. A wake token in a queue is insufficient if a worker loses the token or stops after a backend removes it but before processing completes. Suspension can make these gaps last much longer, but restart-safe ownership is useful without FEP-f799 as well.

Provide an optional KV-backed backlog that owns work before the first enqueue/wake and can recover through sweeping without new traffic. This is distinct from #1333's durable cancellation of a send invocation and #1311's interruption of a local attempt. Do not introduce either cancellation API as an implicit prerequisite here.

Proposed API and use

Add a registration to Federatable<TData>:

setDeliveryBacklog(options: {
  maxRetention: Temporal.DurationLike;
  store?: AtomicKvStore; keyPrefix?: KvKey;
  intake?: "open" | "closed";
  legacyQueue?: "drain" | "adopt-on-hold";
  sweepInterval?: Temporal.DurationLike;
  heldRecheckInterval?: Temporal.DurationLike;
  claimBatchSize?: number;
  leaseDuration?: Temporal.DurationLike;
}): DeliveryBacklog<TData>;

interface DeliveryBacklog<TData> {
  getDepth(ctx: Context<TData>): Promise<BacklogDepth>;
  inspect(ctx: Context<TData>, options?: PageOptions): Promise<Page<BacklogView>>;
  runOnce(ctx: Context<TData>, options?: WorkOptions):
    Promise<{ processed: number; more: boolean }>;
}

Depth distinguishes entries, intents, held, leased, quarantined and pending recipients. Inspection returns IDs, inbox/activity identifiers, due/retention time, lease and wait reason, never the secret prepared signing payload.

const backlog = federation.setDeliveryBacklog({
  maxRetention: { days: 7 },
});
await ctx.sendActivity({ identifier: "alice" }, recipients, activity);
const depth = await backlog.getDepth(ctx);
const page = await backlog.inspect(ctx, { limit: 20 });

// In a cron-owned deployment:
const batch = await backlog.runOnce(ctx, { limit: 100 });
// Schedule another invocation when batch.more is true.

This is a proposed opt-in. The default path accepts backlog ownership before returning; explicit immediate delivery remains direct. The existing queue/startQueue() worker owner can drive the backlog, including a deliberately supported queue-less cron path. Registration/import starts no I/O and needs no initialize() call. With legacyQueue: "drain", pre-existing queue messages finish through the legacy path while new default sends use backlog ownership. With legacyQueue: "adopt-on-hold", a legacy message encountering a hold is durably adopted before the old queue attempt is completed. The intake-closed fallback uses a configured legacy queue, independently of which migration mode is selected.

Ownership and atomic transitions

Persist the prepared canonical payload, resolved recipient snapshot and immutable retention origin before the first signal. Fanout owns a parent intent; expansion atomically inserts a chunk of children and advances its cursor. A conflicting direct batch rolls back the entire batch, and a conflicting expansion chunk is parked with a typed reason rather than silently skipping recipients. Deterministic child IDs alone do not stop re-insertion after completed children were deleted.

Use getMany() snapshots, versioned membership/index roots and casMany() checks for ownership, current cursor, lease and each affected record. Bound chunks to backend limits without splitting one atomic commit. Due-work discovery must recover through persistent index/sweep state rather than treating a queue hint or partial list scan as complete authority. Measure index size/contention and validate the supported atomic key scope on actual adapters.

Claiming, renewing and settling work require the current fencing token. A lease only protects storage writes; an expired worker may already have sent a POST, so unknown acknowledgment can duplicate it. Quarantine unsupported payload versions and expansion conflicts with finite retention and an explicit drop observation, not silent successful acknowledgment. Existing queue signing secrets remain private internal data; this feature does not require redesigning every existing queue credential format.

Preserve sequential accepted order for the same key/destination, including an earlier unexpanded intent. Do not promise global wall-clock order across concurrent producers or old and new queues. A stronger backend order can be an implementation improvement rather than the public minimum.

Suspension, breaker and rollout

Standalone maxRetention runs from immutable ownership time. If receiving is present, retain its independent deferred bound, actual circuit-held time to live (TTL) and existing retry-stop decisions. Perform retention/lease checks and the notice gate before a side-effecting half-open probe at each real POST. Explicitly connect queued breaker behavior on the cron-only backlog path; do not broaden legacy immediate breaker coverage.

Exclude only known FEP-f799 hold intervals, counting overlaps once, from ordinary retry-age accounting. This never extends the backlog maxRetention, the receiver's hard deferred bound or circuit-held TTL. Persist an accounting-slice cap or retain release facts so worker downtime is not credited without a bound. Circuit/quarantine/storage-unavailable time is not FEP-f799 hold credit. A wake only accelerates reevaluation; it is not an authority decision. Handle wake-versus-leased-settle races and preserve sweep fallback. An empty or short claim batch, for example a PostgreSQL SKIP LOCKED scan, does not prove no work remains.

Use intake: "closed" to drain existing ownership before removing the feature. New default sends use a configured legacy queue; without one, return a typed DeliveryBacklogIntakeClosedError rather than silently sending synchronously. Verify both intents and entries are drained. Preserve existing no-backlog behavior.

A generic atomic KV implementation was chosen over a mandatory backend-specific backlog store. A queue sidecar without source ownership, in-process-only locks, hidden direct fallback and exactly-once claims were rejected. A future storage optimization SPI can follow measured need; it is not required in the first public API. Whichever of this feature and suspension receiving ships second owns the integration tests between them.

Shipping criteria and docs

Test crashes before/after wake, direct batch conflict, expansion/cursor replay, deleted-child reinsertion, lease expiry during POST, wake while leased, absent index rows, restart without traffic, ordering, overlapping holds, downtime credit and intake close with and without a configured legacy queue. Run against a persistent atomic adapter that passes the shared conformance suite. General backlog operation must work with suspension disabled.

Add docs/manual/delivery-backlog.md or equivalent complete guidance, and update mq.md, tasks.md, send.md, deploy.md, circuit-breaker.md, implemented monitoring/logging APIs and the suspension guide's integration section. Revise framework skill assumptions about queues, JSDoc, a runnable recovery/drain example and the package fragment. Keep permanent send cancellation in #1333 out of this issue's completion claim.