import type { Clock } from "./clock"; export type PendingReceiptSettlement = | { readonly status: "fulfilled"; readonly value: Value } | { readonly status: "rejected"; readonly error: unknown }; export type PendingReceiptDisposalPolicy = | { readonly kind: "resolve"; readonly value: (identity: { scope: string; messageId: string }) => Value; } | { readonly kind: "reject"; readonly error: (identity: { scope: string; messageId: string; }) => unknown; }; export interface PendingReceiptLedgerOptions { readonly clock: Pick; readonly maxPendingEntries: number; readonly maxSettledEntries: number; readonly settledRetentionMs: number; readonly disposal: PendingReceiptDisposalPolicy; } export type PendingReceiptClaim = | { readonly kind: "fresh"; readonly receipt: Promise; } | { readonly kind: "duplicate"; readonly receipt: Promise; readonly state: "pending" | "settled"; } | { readonly kind: "conflict" } | { readonly kind: "capacity-rejected" } | { readonly kind: "disposed" }; interface Entry { readonly scope: string; readonly messageId: string; readonly fingerprint: string; readonly receipt: Promise; readonly resolve: (value: Value) => void; readonly reject: (error: unknown) => void; state: "pending" | "settled"; settledAt?: number; settlementSequence?: number; } /** * Process-local retry receipt ownership. Pending entries are reserved before * the admission factory runs and can never be evicted by capacity pressure. */ export class PendingReceiptLedger { private readonly scopes = new Map>>(); private pendingCount = 0; private settledCount = 0; private settlementSequence = 0; private disposed = false; constructor(private readonly options: PendingReceiptLedgerOptions) { if ( !Number.isInteger(options.maxPendingEntries) || options.maxPendingEntries < 0 || !Number.isInteger(options.maxSettledEntries) || options.maxSettledEntries < 0 || !Number.isFinite(options.settledRetentionMs) || options.settledRetentionMs < 0 ) { throw new Error("Pending receipt ledger limits must be non-negative"); } } claim(input: { readonly scope: string; readonly messageId: string; readonly fingerprint: string; readonly start: () => Value | PromiseLike; readonly retention?: ( settlement: PendingReceiptSettlement, ) => "retain" | "remove"; }): PendingReceiptClaim { if (this.disposed) return { kind: "disposed" }; this.pruneExpired(); if (this.disposed) return { kind: "disposed" }; const entries = this.scopes.get(input.scope); const existing = entries?.get(input.messageId); if (existing) { if (existing.fingerprint !== input.fingerprint) { return { kind: "conflict" }; } return { kind: "duplicate", receipt: existing.receipt, state: existing.state, }; } if (this.pendingCount >= this.options.maxPendingEntries) { return { kind: "capacity-rejected" }; } let resolve!: (value: Value) => void; let reject!: (error: unknown) => void; const receipt = new Promise((resolveReceipt, rejectReceipt) => { resolve = resolveReceipt; reject = rejectReceipt; }); const entry: Entry = { scope: input.scope, messageId: input.messageId, fingerprint: input.fingerprint, receipt, resolve, reject, state: "pending", }; const scopeEntries = entries ?? new Map>(); if (!entries) this.scopes.set(input.scope, scopeEntries); scopeEntries.set(input.messageId, entry); this.pendingCount += 1; let started: Value | PromiseLike; try { started = input.start(); } catch (error) { this.settle(input.scope, input.messageId, entry, input.retention, { status: "rejected", error, }); return { kind: "fresh", receipt }; } void Promise.resolve(started).then( (value) => this.settle(input.scope, input.messageId, entry, input.retention, { status: "fulfilled", value, }), (error) => this.settle(input.scope, input.messageId, entry, input.retention, { status: "rejected", error, }), ); return { kind: "fresh", receipt }; } hasReceipt(scope: string, messageId: string): boolean { return this.scopes.get(scope)?.has(messageId) === true; } dispose(): void { if (this.disposed) return; this.disposed = true; const entries = [...this.scopes.values()].flatMap((scope) => [ ...scope.values(), ]); this.scopes.clear(); this.pendingCount = 0; this.settledCount = 0; for (const entry of entries) { if (entry.state !== "pending") continue; entry.state = "settled"; try { if (this.options.disposal.kind === "resolve") { entry.resolve(this.options.disposal.value(entry)); } else { entry.reject(this.options.disposal.error(entry)); } } catch (error) { entry.reject(error); } } } private settle( scope: string, messageId: string, entry: Entry, retention: | ((settlement: PendingReceiptSettlement) => "retain" | "remove") | undefined, settlement: PendingReceiptSettlement, ): void { if (entry.state !== "pending") return; entry.state = "settled"; this.pendingCount -= 1; if (settlement.status === "fulfilled") { entry.resolve(settlement.value); } else { entry.reject(settlement.error); } let remove = false; try { remove = retention?.(settlement) === "remove"; } catch { remove = true; } if ( this.disposed || remove || this.scopes.get(scope)?.get(messageId) !== entry ) { this.deleteIfCurrent(scope, messageId, entry); return; } let settledAt: number; try { settledAt = this.options.clock.now(); } catch { this.deleteIfCurrent(scope, messageId, entry); return; } if (this.disposed || this.scopes.get(scope)?.get(messageId) !== entry) { this.deleteIfCurrent(scope, messageId, entry); return; } entry.settledAt = settledAt; entry.settlementSequence = this.settlementSequence++; this.settledCount += 1; this.compactSettled(); } private pruneExpired(): void { let now: number; try { now = this.options.clock.now(); } catch { return; } for (const [scope, entries] of this.scopes) { for (const [messageId, entry] of entries) { if ( entry.state === "settled" && entry.settledAt !== undefined && now - entry.settledAt >= this.options.settledRetentionMs ) { this.deleteIfCurrent(scope, messageId, entry); } } } } private compactSettled(): void { while (this.settledCount > this.options.maxSettledEntries) { let oldest: | { scope: string; messageId: string; entry: Entry } | undefined; for (const [scope, entries] of this.scopes) { for (const [messageId, entry] of entries) { if ( entry.state !== "settled" || entry.settlementSequence === undefined ) { continue; } if ( !oldest || entry.settlementSequence < (oldest.entry.settlementSequence ?? Number.MAX_SAFE_INTEGER) ) { oldest = { scope, messageId, entry }; } } } if (!oldest) return; this.deleteIfCurrent(oldest.scope, oldest.messageId, oldest.entry); } } private deleteIfCurrent( scope: string, messageId: string, entry: Entry, ): void { const entries = this.scopes.get(scope); if (entries?.get(messageId) !== entry) return; entries.delete(messageId); if (entries.size === 0) this.scopes.delete(scope); if (entry.state === "settled" && entry.settledAt !== undefined) { this.settledCount -= 1; } } }