From 1b6fd611b0ffef4408bd5ad6c7b377f762033fff Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 15 Sep 2026 09:38:32 -0700 Subject: [PATCH 1/2] test(notify): prove CL-7238 crash-window dispatch loss (red) Fault-injection tests: a throw between the mail write and the dispatch enqueue leaves a committed mail row whose redelivery dedupes, so the dispatch row is never queued. No src change; the two repair tests fail. --- packages/notify/test/deliver.test.ts | 126 +++++++++++++++++++++++++++ 1 file changed, 126 insertions(+) diff --git a/packages/notify/test/deliver.test.ts b/packages/notify/test/deliver.test.ts index 84f8808735..ffa8e5f307 100644 --- a/packages/notify/test/deliver.test.ts +++ b/packages/notify/test/deliver.test.ts @@ -13,7 +13,9 @@ import { type MailboxDelivery, type NotifyAddressing, type NotifyDeliveryDeps, + type NotifyDispatchStore, type NotifyInboxItem, + type ResolveExistingMailIds, type SinkDeliveryResult, } from "../src/index"; @@ -251,3 +253,127 @@ describe("deliverNotification", () => { expect(second.deliveredMailboxRowIds).toEqual([]); }); }); + +describe("deliverNotification crash window (CL-7238)", () => { + function durableMailbox(): { + mail: MailboxDelivery; + resolveExistingMailIds: ResolveExistingMailIds; + } { + const rows = new Map(); + let next = 0; + const keyOf = (item: NotifyInboxItem) => + `${item.tenantId}\n${item.principalId}\n${item.externalId}`; + const mail: MailboxDelivery = async (items, opts) => + items.map((item) => { + const existing = rows.get(keyOf(item)); + if (existing !== undefined) + return { messageKey: item.externalId, id: null }; + next += 1; + const id = `mail-${next}`; + rows.set(keyOf(item), id); + opts?.enqueue?.({ id, item }); + return { messageKey: item.externalId, id }; + }); + const resolveExistingMailIds: ResolveExistingMailIds = async (items) => + items.map((item) => rows.get(keyOf(item)) ?? null); + return { mail, resolveExistingMailIds }; + } + + function crashableDispatch(): { + dispatch: NotifyDispatchStore; + armCrash: () => void; + } { + const inner = createInMemoryNotifyDispatchStore(); + let crashNext = false; + const dispatch: NotifyDispatchStore = { + ...inner, + enqueue: async (inputs) => { + if (crashNext) { + crashNext = false; + throw new Error( + "boom: crash between the mail write and the dispatch enqueue", + ); + } + await inner.enqueue(inputs); + }, + }; + return { + dispatch, + armCrash: () => { + crashNext = true; + }, + }; + } + + function crashDeps( + mail: MailboxDelivery, + dispatch: NotifyDispatchStore, + resolveExistingMailIds?: ResolveExistingMailIds, + ): NotifyDeliveryDeps { + const sinks = createSinkRegistry(); + const delivered: SinkDeliveryResult = { status: "delivered" }; + sinks.register({ + name: "always", + isEnabledFor: async () => true, + deliver: async () => delivered, + }); + const deps: NotifyDeliveryDeps = { mail, addressing, dispatch, sinks }; + return resolveExistingMailIds === undefined + ? deps + : { ...deps, resolveExistingMailIds }; + } + + test("a redelivery repairs the crash window without double-queueing", async () => { + const { mail, resolveExistingMailIds } = durableMailbox(); + const { dispatch, armCrash } = crashableDispatch(); + const deps = crashDeps(mail, dispatch, resolveExistingMailIds); + + armCrash(); + await expect(deliverNotification(deps, approval)).rejects.toThrow("boom"); + + const redelivery = await deliverNotification(deps, approval); + expect(redelivery.deliveredMailboxRowIds).toEqual([]); + expect(redelivery.queuedDispatchCount).toBe(1); + expect(await dispatch.listFor("mail-1")).toHaveLength(1); + + const third = await deliverNotification(deps, approval); + expect(third.queuedDispatchCount).toBe(1); + expect(await dispatch.listFor("mail-1")).toHaveLength(1); + }); + + test("a redelivery without the read-back still loses the dispatch", async () => { + const { mail } = durableMailbox(); + const { dispatch, armCrash } = crashableDispatch(); + const deps = crashDeps(mail, dispatch); + + armCrash(); + await expect(deliverNotification(deps, approval)).rejects.toThrow("boom"); + + const redelivery = await deliverNotification(deps, approval); + expect(redelivery).toEqual({ + deliveredMailboxRowIds: [], + queuedDispatchCount: 0, + }); + expect(await dispatch.listFor("mail-1")).toHaveLength(0); + }); + + test("a mixed batch dispatches fresh and repaired rows together", async () => { + const { mail, resolveExistingMailIds } = durableMailbox(); + const { dispatch } = crashableDispatch(); + const deps = crashDeps(mail, dispatch, resolveExistingMailIds); + + await deliverNotification(deps, approval); + const mixed = await deliverNotification(deps, { + ...approval, + recipients: [ + { tenantId: "tnt_1", principalId: "prn_1" }, + { tenantId: "tnt_1", principalId: "prn_2" }, + ], + }); + + expect(mixed.deliveredMailboxRowIds).toEqual(["mail-2"]); + expect(mixed.queuedDispatchCount).toBe(2); + expect(await dispatch.listFor("mail-1")).toHaveLength(1); + expect(await dispatch.listFor("mail-2")).toHaveLength(1); + }); +}); From bd4ed604181d0e3fa4ea721ae497aa605c8cf0a7 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 15 Sep 2026 09:38:37 -0700 Subject: [PATCH 2/2] fix(notify): reconcile mail write with dispatch enqueue (CL-7238) deliverNotification now resolves pre-existing mail ids through an optional host read-back and enqueues dispatch for them; the dispatch store dedupes by (mail row, sink), so the repair never double-queues. Without the read-back a crash-window redelivery still loses the dispatch (pinned by test); the inbox adapter owns that wiring. --- packages/notify/src/deliver.ts | 50 ++++++++++++++++++++++++++++------ packages/notify/src/index.ts | 1 + packages/notify/src/mailbox.ts | 13 +++++++++ 3 files changed, 56 insertions(+), 8 deletions(-) diff --git a/packages/notify/src/deliver.ts b/packages/notify/src/deliver.ts index 505137d323..3a1069b2bf 100644 --- a/packages/notify/src/deliver.ts +++ b/packages/notify/src/deliver.ts @@ -7,12 +7,13 @@ // dispatch row per (mail row, enabled sink). A sink is never called from this // path: the mail is the durable record, and a copy of it is the worker's job. // -// That queuing step is NOT atomic with the mail write, and nothing -// reconciles the two if a crash or a throw lands between them (CL-7238): -// a mail row can end up committed with no dispatch row ever queued for it. -// Closing that gap needs a seam `@corbits/mailbox` doesn't expose today -// (an in-transaction hook, or a way to read back an existing row's id by -// key) — see CL-7238 for why this can't be fixed from this package alone. +// That queuing step is NOT atomic with the mail write: a crash or a +// throw between them leaves a mail row committed with no dispatch row +// queued (CL-7238). The mail delivery reports such rows with a `null` +// id, so a redelivery resolves their ids through the host's +// `resolveExistingMailIds` read-back and repairs the missing dispatch +// rows; the dispatch store dedupes by (mail row, sink), so the repair +// never double-queues. import { parseNotificationEvent, type ApprovalNotification, @@ -25,6 +26,7 @@ import type { MailboxDelivery, NotifyAddressing, NotifyInboxItem, + ResolveExistingMailIds, } from "./mailbox"; import { notificationExternalId, renderNotification } from "./render"; import type { SinkRegistry } from "./sinks"; @@ -38,6 +40,15 @@ export interface NotifyDeliveryDeps { readonly addressing: NotifyAddressing; readonly dispatch: NotifyDispatchStore; readonly sinks: SinkRegistry; + /** + * Read-back for the CL-7238 crash window: when the mail delivery + * reports an item as pre-existing (`id: null`), the row was committed + * by an earlier attempt that died before its dispatch enqueue. + * Resolving those items to their row ids lets this call repair the + * missing dispatch rows; without it a redelivery after such a crash + * still loses the dispatch. The host owns this wiring. + */ + readonly resolveExistingMailIds?: ResolveExistingMailIds; } export interface NotifyDeliveryReport { @@ -75,8 +86,9 @@ export async function deliverNotification( input: unknown, ): Promise { const event = parseNotificationEvent(input); + const items = toInboxItems(event, deps.addressing); const written: { id: string; tenantId: string; principalId: string }[] = []; - await deps.mail(toInboxItems(event, deps.addressing), { + const results = await deps.mail(items, { enqueue: ({ id, item }) => { written.push({ id, @@ -86,8 +98,30 @@ export async function deliverNotification( }, }); + // CL-7238: a `null` id marks a row committed by an earlier attempt + // that died before its dispatch enqueue (or a benign redelivery). + // Resolve those ids so the enqueue below repairs the missing rows; + // without the read-back there is nothing to enqueue for them. + const deduped = items.filter( + (_, index) => (results[index]?.id ?? null) === null, + ); + const repaired: { id: string; tenantId: string; principalId: string }[] = []; + if (deduped.length > 0 && deps.resolveExistingMailIds !== undefined) { + const resolved = await deps.resolveExistingMailIds(deduped); + deduped.forEach((item, index) => { + const id = resolved[index]; + if (id !== undefined && id !== null) { + repaired.push({ + id, + tenantId: item.tenantId, + principalId: item.principalId, + }); + } + }); + } + const queued: EnqueueDispatchInput[] = []; - for (const row of written) { + for (const row of [...written, ...repaired]) { const enabled = await deps.sinks.listEnabledFor({ tenantId: row.tenantId, principalId: row.principalId, diff --git a/packages/notify/src/index.ts b/packages/notify/src/index.ts index b89fac083d..076e60dbfe 100644 --- a/packages/notify/src/index.ts +++ b/packages/notify/src/index.ts @@ -84,4 +84,5 @@ export type { NotifyDeliverOpts, NotifyInboxItem, NotifyMailRef, + ResolveExistingMailIds, } from "./mailbox"; diff --git a/packages/notify/src/mailbox.ts b/packages/notify/src/mailbox.ts index 4cbd768570..556225e9b4 100644 --- a/packages/notify/src/mailbox.ts +++ b/packages/notify/src/mailbox.ts @@ -37,11 +37,24 @@ export type NotifyDeliverOpts = { readonly enqueue?: (delivered: { id: string; item: NotifyInboxItem }) => void; }; +/** Results align positionally with `items`, mirroring the vendor delivery. */ export type MailboxDelivery = ( items: NotifyInboxItem[], opts?: NotifyDeliverOpts, ) => Promise; +/** + * Read-back for mail rows an earlier delivery committed before dying + * ahead of its dispatch enqueue (CL-7238). The input lists the items + * the mail delivery just reported as pre-existing (`id: null`); the + * result aligns positionally with the input, with `null` where no row + * exists. The host serves this from the mailbox's unique mail key — a + * lookup the seam doesn't expose today, so the host owns the wiring. + */ +export type ResolveExistingMailIds = ( + items: readonly NotifyInboxItem[], +) => Promise; + /** * How a notification addresses mail. The host owns its own address space — * a mailbox address is a deployment fact, not a product rule — so both halves