diff --git a/packages/inbox/src/delivery.ts b/packages/inbox/src/delivery.ts index a43290cff6..8141c14502 100644 --- a/packages/inbox/src/delivery.ts +++ b/packages/inbox/src/delivery.ts @@ -3,13 +3,20 @@ // group) and status so list filters and the three-column UI work without // re-deriving on every read. +import { and, eq, or } from "drizzle-orm"; import { deliverInboxItems, + mailboxKey, + principalMail, type DeliverInboxItemsOpts, type MailboxDb, type MailboxEventBus, } from "@corbits/mailbox"; -import type { MailboxDelivery, NotifyInboxItem } from "@corbits/notify/mailbox"; +import type { + MailboxDelivery, + NotifyInboxItem, + ResolveExistingMailIds, +} from "@corbits/notify/mailbox"; import { classificationFromRefs } from "./group"; @@ -74,3 +81,58 @@ export function createWorkbenchMailboxDelivery( return deliverInboxItems(db, stamped, writeOpts); }; } + +/** + * CL-7238 read-back for `@corbits/notify`'s crash window: a redelivery + * reports an already-committed row as pre-existing (`id: null`), so the + * host resolves it to its row id here and the missing dispatch rows get + * repaired instead of lost. Served from the mailbox unique mail key — + * the same `mailboxKey.inbox(source, externalId)` the write dedupes on — + * so the lookup can never disagree with the delivery about which row an + * item belongs to. Positional in, positional out, `null` where no row + * exists; the dispatch store dedupes by (mail row, sink), so a benign + * redelivery's repair never double-queues. + */ +export function createResolveExistingMailIds( + db: MailboxDb, +): ResolveExistingMailIds { + return async (items) => { + if (items.length === 0) return []; + const keyed = items.map((item) => ({ + tenantId: item.tenantId, + principalId: item.principalId, + messageKey: mailboxKey.inbox(item.source, item.externalId), + })); + const rows = await db + .select({ + id: principalMail.id, + tenantId: principalMail.tenantId, + principalId: principalMail.principalId, + messageKey: principalMail.messageKey, + }) + .from(principalMail) + .where( + or( + ...keyed.map((key) => + and( + eq(principalMail.tenantId, key.tenantId), + eq(principalMail.principalId, key.principalId), + eq(principalMail.messageKey, key.messageKey), + ), + ), + ), + ); + const idsByKey = new Map( + rows.map((row) => [ + `${row.tenantId}\n${row.principalId}\n${row.messageKey}`, + row.id, + ]), + ); + return keyed.map( + (key) => + idsByKey.get( + `${key.tenantId}\n${key.principalId}\n${key.messageKey}`, + ) ?? null, + ); + }; +} diff --git a/packages/inbox/src/index.ts b/packages/inbox/src/index.ts index 75816bc8ef..976ff600ef 100644 --- a/packages/inbox/src/index.ts +++ b/packages/inbox/src/index.ts @@ -24,6 +24,7 @@ export { type WorkbenchInboxStatus, } from "./vocabulary"; export { + createResolveExistingMailIds, createWorkbenchMailboxDelivery, type CreateWorkbenchMailboxDeliveryOpts, } from "./delivery"; diff --git a/packages/inbox/test/delivery.test.ts b/packages/inbox/test/delivery.test.ts index 54522447de..1d32ced1d7 100644 --- a/packages/inbox/test/delivery.test.ts +++ b/packages/inbox/test/delivery.test.ts @@ -20,12 +20,18 @@ import { deliverCredentialMail, deliverMentionMail, deliverRunFailureMail, + type ApprovalNotification, type NotifyDeliveryDeps, + type NotifyDispatchStore, + type SinkDeliveryResult, } from "@corbits/notify"; import { setupDatabase } from "../../../scripts/db-setup"; import { e2eDatabaseUrl } from "../../../scripts/e2e/database-url"; -import { createWorkbenchMailboxDelivery } from "../src/delivery"; +import { + createResolveExistingMailIds, + createWorkbenchMailboxDelivery, +} from "../src/delivery"; import { inboxGroupOf } from "../src/group"; import { WORKBENCH_INBOX_PRIORITIES } from "../src/vocabulary"; import { dbGate } from "../../../scripts/e2e/db-gate"; @@ -213,5 +219,91 @@ describeIfDb( await mailboxDb.close(); } }, 15000); + + test("a crash between mail write and dispatch enqueue is repaired on redelivery without doubles (CL-7238)", async () => { + const mailboxDb = createMailboxDb(scratchUrl); + try { + const innerDispatch = createInMemoryNotifyDispatchStore(); + let crashNext = true; + const dispatch: NotifyDispatchStore = { + ...innerDispatch, + enqueue: async (inputs) => { + if (crashNext) { + crashNext = false; + throw new Error( + "boom: crash between the mail write and the dispatch enqueue", + ); + } + await innerDispatch.enqueue(inputs); + }, + }; + const delivered: SinkDeliveryResult = { status: "delivered" }; + const sinks = createSinkRegistry(); + sinks.register({ + name: "always", + isEnabledFor: async () => true, + deliver: async () => delivered, + }); + const deps: NotifyDeliveryDeps = { + mail: createWorkbenchMailboxDelivery({ db: mailboxDb.db }), + addressing: { + inbox: (recipient) => `${recipient.principalId}@inbox.test`, + from: (kind) => `${kind}@notify.test`, + }, + dispatch, + resolveExistingMailIds: createResolveExistingMailIds(mailboxDb.db), + sinks, + }; + const event: ApprovalNotification = { + kind: "approval", + approvalId: generateId("approval"), + tenantId, + runId: generateId("workflowRun"), + deploymentId: generateId("workflowRun"), + toolName: "quarantine_token", + toolArguments: { repo: "corbitsdev/workbench" }, + recipients: [{ tenantId, principalId }], + createdAt: new Date().toISOString(), + }; + + // The first attempt commits its mail row through the real adapter, + // then dies before the dispatch enqueue — the crash window. + await expect(deliverApprovalMail(deps, event)).rejects.toThrow("boom"); + + // The redelivery dedupes on the mail key (no new row), resolves the + // committed row's id through the adapter read-back, and queues the + // dispatch the crash swallowed. + const redelivery = await deliverApprovalMail(deps, event); + expect(redelivery.deliveredMailboxRowIds).toEqual([]); + expect(redelivery.queuedDispatchCount).toBe(1); + + const page = await listUserMailbox(mailboxDb.db, { + tenantId, + principalId, + limit: 10, + view: "all", + priorities: WORKBENCH_INBOX_PRIORITIES, + }); + // Exactly one mail row for the redelivered event — the crash and + // both redeliveries deduped onto the first attempt's row. + const matching = page.items.filter((item) => + (item.subject ?? "").includes("quarantine_token"), + ); + expect(matching).toHaveLength(1); + const mailId = matching[0]?.id; + if (mailId === undefined) + throw new Error("expected the redelivered mail row to list"); + expect(await dispatch.listFor(mailId)).toHaveLength(1); + + // A further benign redelivery repairs the same row again, and the + // dispatch store's (mail row, sink) dedupe holds it to one row. + const third = await deliverApprovalMail(deps, event); + expect(third.deliveredMailboxRowIds).toEqual([]); + expect(third.queuedDispatchCount).toBe(1); + expect(await dispatch.listFor(mailId)).toHaveLength(1); + } finally { + await mailboxDb.close(); + } + }, 15000); }, ); 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 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); + }); +});