fix(notify): reconcile mail write with dispatch enqueue (CL-7238)

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<NotifyDeliveryReport> {
   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,

packages/notify/src/index.ts

@@ -84,4 +84,5 @@ export type {
   NotifyDeliverOpts,
   NotifyInboxItem,
   NotifyMailRef,
+  ResolveExistingMailIds,
 } from "./mailbox";

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<NotifyDeliveredItem[]>;
 
+/**
+ * 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<readonly (string | null)[]>;
+
 /**
  * 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

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<string, string>();
+    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);
+  });
+});