From 8afe970dac244295b88bb8405426145b30013b6a Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:07:14 -0700 Subject: [PATCH 1/6] Add tests for createMailboxPersist's resolveRefs option Covers refs landing inside the same transaction (visible to a bus subscriber on the create event), a resolver called once per frame regardless of recipient count, a throwing resolver leaving zero rows while upstream still completes, and over-cap refs truncated to MAX_MAILBOX_REFS. --- src/persist.test.ts | 108 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 108 insertions(+) diff --git a/src/persist.test.ts b/src/persist.test.ts index fdf5455..0b350b1 100644 --- a/src/persist.test.ts +++ b/src/persist.test.ts @@ -9,6 +9,7 @@ import { type MailboxPersistArgs, type SenderAuthorization, type PersistedMailboxRow, + type ResolveMailboxRefs, } from "./persist.js"; import { createInMemoryMailboxEventBus } from "./bus.js"; import { buildMailFrame } from "./frame.js"; @@ -17,8 +18,10 @@ import { withTestDb, seedScope } from "./test-helpers.js"; import type { MailboxDb } from "./db.js"; import { MAX_MAILBOX_FRAME_BYTES, + MAX_MAILBOX_REFS, assertMailboxFrameBytes, } from "./write.js"; +import { getMailboxMessage, type MailboxRef } from "./read.js"; let db: MailboxDb; @@ -520,6 +523,111 @@ describe("transport insert idempotency", () => { }); }); +describe("resolveRefs", () => { + test("refs are visible to a bus subscriber on the create event", async () => { + const bus = createInMemoryMailboxEventBus(); + const seenIds: string[] = []; + bus.subscribe({ tenantId: "acme", principalId: "user-1" }, (e) => + seenIds.push(e.id), + ); + const refs: MailboxRef[] = [{ kind: "workbench", id: "thread-1" }]; + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + bus, + resolveRefs: () => refs, + }); + + await persist(args()); + + // By the time the bus fires, the insert has already committed — a + // subscriber reading the row by the announced id sees refs already + // stamped, not a later-arriving update. + expect(seenIds).toHaveLength(1); + const message = await getMailboxMessage(db, { + tenantId: "acme", + principalId: "user-1", + id: seenIds[0]!, + }); + expect(message?.refs).toEqual(refs); + }); + + test("resolveRefs is called once for three recipients, and every row gets its refs", async () => { + await seedScope(db, "acme", "user-3"); + const refs: MailboxRef[] = [{ kind: "workbench", id: "thread-1" }]; + const calls: unknown[] = []; + const resolveRefs: ResolveMailboxRefs = (a) => { + calls.push(a); + return refs; + }; + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs, + }); + + await persist( + args({ + recipients: [ + "usr_user-1@acme.example", + "usr_user-2@acme.example", + "usr_user-3@acme.example", + ], + }), + ); + + expect(calls).toHaveLength(1); + for (const principalId of ["user-1", "user-2", "user-3"]) { + const [row] = await rowsFor("acme", principalId); + const message = await getMailboxMessage(db, { + tenantId: "acme", + principalId, + id: row!.id, + }); + expect(message?.refs).toEqual(refs); + } + }); + + test("a throwing resolveRefs writes zero rows; upstream still completes", async () => { + const { upstream, result } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => { + throw new Error("resolver exploded"); + }, + }); + + expect(await persist(args())).toBe(result); + expect(await rowsFor("acme", "user-1")).toHaveLength(0); + }); + + test("over-cap refs from resolveRefs are truncated to MAX_MAILBOX_REFS", async () => { + const refs: MailboxRef[] = Array.from({ length: MAX_MAILBOX_REFS + 5 }, (_, i) => ({ + kind: "workbench", + id: `thread-${i}`, + })); + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => refs, + }); + + await persist(args()); + + const [row] = await rowsFor("acme", "user-1"); + const message = await getMailboxMessage(db, { + tenantId: "acme", + principalId: "user-1", + id: row!.id, + }); + expect(message?.refs?.length).toBe(MAX_MAILBOX_REFS); + }); +}); + describe("frame size and recipient hard caps", () => { test("assertMailboxFrameBytes accepts at-cap and throws RangeError one byte over", () => { // Dual-write swallows the RangeError inside attemptMailboxWrite; the pure From c367aa8ad0354583dd7e3de1fbcdcca417e35cd6 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:07:20 -0700 Subject: [PATCH 2/6] createMailboxPersist: accept refs at insert time resolveRefs(args) is called once per frame, before the transaction opens, receiving the persist args plus the resolved senderAuthorization and the decoded frame (or null). Its result is validated with MailboxRefArraySchema and capped at MAX_MAILBOX_REFS (boundRefs, now exported from write.ts), then stored on every recipient row inside the existing single transaction, so the post-commit bus event already carries refs. A throwing resolver follows the wrapper's existing dual-write contract: logged, upstream unaffected, no mailbox row for that frame. --- src/persist.ts | 58 +++++++++++++++++++++++++++++++++++++++++++++++++- src/write.ts | 2 +- 2 files changed, 58 insertions(+), 2 deletions(-) diff --git a/src/persist.ts b/src/persist.ts index 7a85395..44c7047 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -17,15 +17,18 @@ import { createHash } from "node:crypto"; import { and, eq, inArray, sql } from "drizzle-orm"; +import { type } from "arktype"; import { getLogger } from "@intx/log"; import { hostPrincipal, mailbox, principalMail } from "./schema.js"; import type { MailboxDb } from "./db.js"; import { publishMailboxEvent, type MailboxEventBus } from "./bus.js"; -import { decodeMailFrame, parseMsgIdList } from "./frame.js"; +import { decodeMailFrame, parseMsgIdList, type DecodedFrame } from "./frame.js"; import { resolveMailboxRecipients } from "./recipients.js"; +import { MailboxRefArraySchema, type MailboxRef } from "./read.js"; import { assertMailboxScope, assertMailboxFrameBytes, + boundRefs, } from "./write.js"; const logger = getLogger(["corbits-mailbox", "persist"]); @@ -91,6 +94,28 @@ export type PersistedMailboxRow = { senderAddress: string; }; +/** + * Host seam for stamping every recipient row of one frame with the same + * `refs`. Called once per frame, before the transaction opens — NOT once per + * recipient — so a host pointing every row at the same upstream entity + * (`{ kind: "workbench", id }`) does one lookup, not N. + * + * The result is validated with `MailboxRefArraySchema` and capped at + * `MAX_MAILBOX_REFS` the same way `writeMailboxMessage`'s `refs` argument is; + * see `boundRefs`. Returning `undefined` (or an empty array) stores no refs. + * + * A throwing `resolveRefs` is handled exactly like a mailbox-write failure + * under the dual-write contract: logged, upstream still runs (it already + * ran, or still will, independently of this), and no mailbox row is written + * for that frame. See ARCHITECTURE.md's persist section. + */ +export type ResolveMailboxRefs = ( + args: MailboxPersistArgs & { + senderAuthorization: SenderAuthorization; + decoded: DecodedFrame | null; + }, +) => Promise | MailboxRef[] | undefined; + export type CreateMailboxPersistOpts = { /** The host's own persist path. Always called, for every frame. */ upstream: (args: MailboxPersistArgs) => Promise; @@ -99,6 +124,12 @@ export type CreateMailboxPersistOpts = { bus?: MailboxEventBus; /** Best-effort hook per inserted row; a throw is logged, never propagated. */ onRow?: (row: PersistedMailboxRow) => void; + /** + * Resolve the `refs` every recipient row of one frame gets, INSIDE the + * existing single transaction — so the post-commit bus event and any SSE + * subscriber already see them. See `ResolveMailboxRefs`. + */ + resolveRefs?: ResolveMailboxRefs; }; /** @@ -232,6 +263,30 @@ export function createMailboxPersist( // an upgrade's backfill would have produced for the same frame. const inReplyTo = parseMsgIdList(decoded?.headers.get("in-reply-to"))[0] ?? null; + // Resolved ONCE per frame, before the transaction — every recipient row + // gets the same refs, and a resolver that hits an upstream entity does one + // lookup regardless of recipient count. A throw here propagates out of + // `writeMailboxRows` exactly like any other pre-transaction failure: + // `attemptMailboxWrite` catches and logs it, upstream still stands, and no + // mailbox row is written for this frame. + let refs: MailboxRef[] | undefined; + if (opts.resolveRefs) { + const resolved = await opts.resolveRefs({ + senderAddress, + recipients, + raw, + senderAuthorization: auth, + decoded, + }); + if (resolved !== undefined && resolved.length > 0) { + const validated = MailboxRefArraySchema(resolved); + if (validated instanceof type.errors) { + throw new RangeError(`invalid mailbox refs: ${validated.summary}`); + } + refs = boundRefs(validated, null); + } + } + // Mail rows and their management rows commit together: the management row // is created eagerly with the message (see `writeMailboxMessage`), and a // message without one is unreachable by every mutation. messageKey makes @@ -252,6 +307,7 @@ export function createMailboxPersist( fromAddress, messageId, inReplyTo, + refs: refs ?? null, messageKey: transportMessageKey( messageId, raw, diff --git a/src/write.ts b/src/write.ts index f6c2a18..c65b978 100644 --- a/src/write.ts +++ b/src/write.ts @@ -92,7 +92,7 @@ export function assertMailboxFrameBytes(raw: Uint8Array): void { // `messageKey` is the caller's own identifier and is absent for externally // delivered mail, which is never deduped — the warning below carries whatever // the caller actually supplied rather than minting an id nobody can correlate. -function boundRefs( +export function boundRefs( refs: MailboxRef[] | undefined, messageKey: string | null, ): MailboxRef[] | undefined { From 7f9a52ce1ce4be535f30a6feab5eda46c7e5410a Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:07:24 -0700 Subject: [PATCH 3/6] Update docs: createMailboxPersist's resolveRefs option Document the seam in ARCHITECTURE.md's persist section, add a README snippet, and a CHANGELOG entry. --- ARCHITECTURE.md | 33 ++++++++++++++++++++++++--------- CHANGELOG.md | 9 +++++++++ README.md | 18 ++++++++++++++++++ 3 files changed, 51 insertions(+), 9 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 2b41d04..f64c295 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -58,15 +58,16 @@ tables foreign-key to both. Nothing changed in Interchange to make that work — the coupling lives entirely on this side. One further seam lives outside `mountMailbox`, on the write side: -`createMailboxPersist(db, { upstream, authorizeSender, bus?, onRow? })` wraps a -host's own mail-persist function so every addressed principal also gets a -durable row. `authorizeSender(address) => { tenantId, domain } | null` is the -host's decision — whether a sender address belongs to a *live* agent instance -is not a schema fact. Returning `null` skips the mailbox write entirely while -the frame still goes upstream. On the recipient side the package does consult -the control plane: an address whose local part matches no known principal in -the authorized tenant is skipped with a warning rather than minting a phantom -mailbox row, and never costs the frame's real recipients their durable copy. +`createMailboxPersist(db, { upstream, authorizeSender, bus?, onRow?, resolveRefs? })` +wraps a host's own mail-persist function so every addressed principal also +gets a durable row. `authorizeSender(address) => { tenantId, domain } | null` +is the host's decision — whether a sender address belongs to a *live* agent +instance is not a schema fact. Returning `null` skips the mailbox write +entirely while the frame still goes upstream. On the recipient side the +package does consult the control plane: an address whose local part matches +no known principal in the authorized tenant is skipped with a warning rather +than minting a phantom mailbox row, and never costs the frame's real +recipients their durable copy. The wrapper's contract is **dual-write independence in both directions**: an `upstream` throw still attempts the mailbox write and then re-throws the @@ -75,6 +76,20 @@ persist that upstream already completed. Under retry, the mailbox side is idempotent: package-owned transport `messageKey`s plus `onConflictDoNothing` collapse duplicate frames without failing the call or re-announcing. +`resolveRefs(args)` — `args` is `MailboxPersistArgs` plus the resolved +`senderAuthorization` and the `decoded` frame (or `null` if the parser +rejected it) — is called ONCE per frame, before the transaction opens, not +once per recipient: a host pointing every row at the same upstream entity +does one lookup, not N. Its result is validated with `MailboxRefArraySchema` +and capped at `MAX_MAILBOX_REFS` the same way `writeMailboxMessage`'s `refs` +argument is (excess entries truncated, logged, never a throw), then stored on +every recipient row of that frame INSIDE the same transaction — so the +post-commit bus `create` event and any SSE subscriber already see refs on the +row once it's readable. A throwing `resolveRefs` is handled exactly like any +other pre-transaction failure: it falls under the same dual-write contract as +the rest of the wrapper — logged, upstream unaffected (already ran, or still +will, independently of this), and no mailbox row for that frame. + `resolvePrincipal`'s signature is identical across the Corbits cores, so a host mounting more than one passes the same function to each. diff --git a/CHANGELOG.md b/CHANGELOG.md index 24dd77c..77a70fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,15 @@ always called out under their own heading. ### Added +- **`createMailboxPersist` accepts `resolveRefs` at insert time.** The new + `resolveRefs(args)` option — `args` is the persist args plus the resolved + `senderAuthorization` and the `decoded` frame (or `null`) — is called once + per frame, before the transaction opens, and its result (validated with + `MailboxRefArraySchema`, capped at `MAX_MAILBOX_REFS`) is stored on every + recipient row inside that same transaction, so the post-commit bus event + already carries refs. A throwing `resolveRefs` follows the existing + dual-write contract: logged, upstream unaffected, no mailbox row for that + frame. - **Threading headers on the frame and in the list projection.** `buildMailFrame` accepts `references` — the thread's ancestry, oldest first — and emits it as a folded `References:` header; `In-Reply-To` diff --git a/README.md b/README.md index 357ddc4..dd03fd3 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,24 @@ Requires `@intx` 0.2.2 or newer. See [ARCHITECTURE.md](./ARCHITECTURE.md) for the data model. +## Dual-write persist + +```ts +import { createMailboxPersist } from "@corbits/mailbox"; + +const persist = createMailboxPersist(db, { + upstream: hostMailTransport.persist, + authorizeSender: (address) => resolveActiveInstance(address), + // Called once per frame, before the transaction — every recipient row + // gets the same refs, so a bus subscriber sees them on the `create` event. + resolveRefs: ({ decoded }) => + decoded ? [{ kind: "workbench", id: decoded.messageId ?? "" }] : undefined, +}); +``` + +See ARCHITECTURE.md's persist section for the full contract, including +`resolveRefs`'s dual-write-failure semantics. + ## Install ```sh From 46b0779c1a2503320fde8cbf19f90a857104725b Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:17:15 -0700 Subject: [PATCH 4/6] Add tests for resolveRefs freeze, NULL refs, invalid-output, and truncation semantics --- src/persist.test.ts | 112 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 112 insertions(+) diff --git a/src/persist.test.ts b/src/persist.test.ts index 0b350b1..a5ef006 100644 --- a/src/persist.test.ts +++ b/src/persist.test.ts @@ -626,6 +626,118 @@ describe("resolveRefs", () => { }); expect(message?.refs?.length).toBe(MAX_MAILBOX_REFS); }); + + test("retried frame with a different resolver result keeps the FIRST refs", async () => { + let call = 0; + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => [{ kind: "workbench", id: `attempt-${++call}` }], + }); + + await persist(args()); + await persist(args()); + + const all = await rowsFor("acme", "user-1"); + expect(all).toHaveLength(1); + expect(call).toBe(2); + const message = await getMailboxMessage(db, { + tenantId: "acme", + principalId: "user-1", + id: all[0]!.id, + }); + expect(message?.refs).toEqual([{ kind: "workbench", id: "attempt-1" }]); + }); + + test("resolver returning undefined stores SQL NULL, not []", async () => { + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => undefined, + }); + + await persist(args()); + + const [row] = await rowsFor("acme", "user-1"); + expect(row!.refs).toBeNull(); + }); + + test("resolver returning [] stores SQL NULL, not []", async () => { + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => [], + }); + + await persist(args()); + + const [row] = await rowsFor("acme", "user-1"); + expect(row!.refs).toBeNull(); + }); + + test("schema-invalid resolver output writes zero rows; upstream result still returned", async () => { + const { upstream, result } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => [{ kind: 42, id: "x" }] as unknown as MailboxRef[], + }); + + expect(await persist(args())).toBe(result); + expect(await rowsFor("acme", "user-1")).toHaveLength(0); + }); + + test("resolver runs after upstream resolves (serial), adding its latency to the call", async () => { + const order: string[] = []; + const persist = createMailboxPersist(db, { + upstream: async () => { + await Bun.sleep(150); + order.push("upstream"); + return [{ delivered: true }]; + }, + authorizeSender: () => ACTIVE, + resolveRefs: async () => { + order.push("resolver-start"); + await Bun.sleep(150); + order.push("resolver-end"); + return undefined; + }, + }); + + const t0 = performance.now(); + await persist(args()); + const elapsed = performance.now() - t0; + + expect(order).toEqual(["upstream", "resolver-start", "resolver-end"]); + expect(elapsed).toBeGreaterThanOrEqual(290); + }); + + test("the 21st ref (a workbench ref appended last) is silently dropped", async () => { + const refs: MailboxRef[] = Array.from({ length: MAX_MAILBOX_REFS }, (_, i) => ({ + kind: "thread", + id: `t-${i}`, + })); + refs.push({ kind: "workbench", id: "must-be-present" }); + const { upstream } = recordingUpstream(); + const persist = createMailboxPersist(db, { + upstream, + authorizeSender: () => ACTIVE, + resolveRefs: () => refs, + }); + + await persist(args()); + + const [row] = await rowsFor("acme", "user-1"); + const message = await getMailboxMessage(db, { + tenantId: "acme", + principalId: "user-1", + id: row!.id, + }); + expect(message?.refs?.some((r) => r.kind === "workbench")).toBe(false); + }); }); describe("frame size and recipient hard caps", () => { From cd7c052d83c2d76c6603dbd1c0974505e9eceb3d Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:17:34 -0700 Subject: [PATCH 5/6] persist: tag resolveRefs failures with their stage, log messageId/senderAddress --- src/persist.ts | 46 ++++++++++++++++++++++++++++++++++++---------- src/write.ts | 3 +++ 2 files changed, 39 insertions(+), 10 deletions(-) diff --git a/src/persist.ts b/src/persist.ts index 44c7047..1c55434 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -33,6 +33,21 @@ import { const logger = getLogger(["corbits-mailbox", "persist"]); +/** + * Tags a thrown error with the persist stage that produced it, so the + * dual-write failure log can name `resolveRefs` specifically instead of a + * generic "mailbox write failed". The wrapped error is what's logged and + * (never here) rethrown — see `attemptMailboxWrite`. + */ +class MailboxPersistStageError extends Error { + readonly stage: string; + constructor(stage: string, cause: unknown) { + super(cause instanceof Error ? cause.message : String(cause), { cause }); + this.name = "MailboxPersistStageError"; + this.stage = stage; + } +} + // Cap the sender-controlled recipient list before resolve / inArray / multi-row // insert. Matches MAX_BULK_MAILBOX_IDS posture: hard refuse, never clamp. export const MAX_MAILBOX_RECIPIENTS = 50; @@ -271,19 +286,24 @@ export function createMailboxPersist( // mailbox row is written for this frame. let refs: MailboxRef[] | undefined; if (opts.resolveRefs) { - const resolved = await opts.resolveRefs({ - senderAddress, - recipients, - raw, - senderAuthorization: auth, - decoded, - }); - if (resolved !== undefined && resolved.length > 0) { - const validated = MailboxRefArraySchema(resolved); + let resolvedRefs: MailboxRef[] | undefined; + try { + resolvedRefs = await opts.resolveRefs({ + senderAddress, + recipients, + raw, + senderAuthorization: auth, + decoded, + }); + } catch (err) { + throw new MailboxPersistStageError("resolveRefs", err); + } + if (resolvedRefs !== undefined && resolvedRefs.length > 0) { + const validated = MailboxRefArraySchema(resolvedRefs); if (validated instanceof type.errors) { throw new RangeError(`invalid mailbox refs: ${validated.summary}`); } - refs = boundRefs(validated, null); + refs = boundRefs(validated, messageId, { senderAddress }); } } @@ -361,8 +381,14 @@ export function createMailboxPersist( try { await writeMailboxRows(args); } catch (err) { + // Decoded independently of `writeMailboxRows`'s own decode: the throw + // may have happened before that decode ran (e.g. authorizeSender), and + // this log line must still correlate to a messageId when one exists. + const messageId = decodeMailFrame(args.raw)?.messageId ?? null; logger.error("mailbox write failed for mail from {senderAddress}", { senderAddress: args.senderAddress, + messageId, + ...(err instanceof MailboxPersistStageError ? { stage: err.stage } : {}), error: err instanceof Error ? err : new Error(String(err)), }); } diff --git a/src/write.ts b/src/write.ts index c65b978..3e092fd 100644 --- a/src/write.ts +++ b/src/write.ts @@ -95,6 +95,8 @@ export function assertMailboxFrameBytes(raw: Uint8Array): void { export function boundRefs( refs: MailboxRef[] | undefined, messageKey: string | null, + /** Extra correlation fields merged into the truncation log line, e.g. `senderAddress` on the persist path. */ + extra?: Record, ): MailboxRef[] | undefined { if (refs === undefined || refs.length === 0) return undefined; if (refs.length <= MAX_MAILBOX_REFS) return refs; @@ -102,6 +104,7 @@ export function boundRefs( messageKey, received: refs.length, kept: MAX_MAILBOX_REFS, + ...extra, }); return refs.slice(0, MAX_MAILBOX_REFS); } From e44808b2ff2a4567cd6a0859b710fa7e5c18bbf9 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Fri, 4 Sep 2026 18:17:42 -0700 Subject: [PATCH 6/6] Update docs: resolveRefs freeze-on-first-insert, serial-after-upstream, truncation ordering --- ARCHITECTURE.md | 29 ++++++++++++++++++++--------- README.md | 6 ++++++ src/persist.ts | 21 ++++++++++++++++++--- 3 files changed, 44 insertions(+), 12 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f64c295..265f584 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -80,15 +80,26 @@ collapse duplicate frames without failing the call or re-announcing. `senderAuthorization` and the `decoded` frame (or `null` if the parser rejected it) — is called ONCE per frame, before the transaction opens, not once per recipient: a host pointing every row at the same upstream entity -does one lookup, not N. Its result is validated with `MailboxRefArraySchema` -and capped at `MAX_MAILBOX_REFS` the same way `writeMailboxMessage`'s `refs` -argument is (excess entries truncated, logged, never a throw), then stored on -every recipient row of that frame INSIDE the same transaction — so the -post-commit bus `create` event and any SSE subscriber already see refs on the -row once it's readable. A throwing `resolveRefs` is handled exactly like any -other pre-transaction failure: it falls under the same dual-write contract as -the rest of the wrapper — logged, upstream unaffected (already ran, or still -will, independently of this), and no mailbox row for that frame. +does one lookup, not N. It runs AFTER `upstream` resolves and serially with +it, so its latency adds to the call rather than overlapping. Its result is +validated with `MailboxRefArraySchema` and capped at `MAX_MAILBOX_REFS` the +same way `writeMailboxMessage`'s `refs` argument is (excess entries +truncated from the end of the list, logged with `messageId` and +`senderAddress`, never a throw) — so a resolver must return a small set with +the load-bearing ref FIRST, since anything past the cap is silently dropped. +Refs are then stored on every recipient row of that frame INSIDE the same +transaction — so the post-commit bus `create` event and any SSE subscriber +already see refs on the row once it's readable. + +Refs are frozen at the FIRST successful insert for a given frame: a retry +(same idempotency key) still calls `resolveRefs` — it is not skipped — but +because `onConflictDoNothing` writes no row on a retry, a different result +from that second call is simply discarded; only the first call's refs ever +land. A throwing `resolveRefs` is handled exactly like any other +pre-transaction failure: it falls under the same dual-write contract as the +rest of the wrapper — logged (naming `resolveRefs` as the failing stage), +upstream unaffected (already ran, or still will, independently of this), and +no mailbox row for that frame. `resolvePrincipal`'s signature is identical across the Corbits cores, so a host mounting more than one passes the same function to each. diff --git a/README.md b/README.md index dd03fd3..392222d 100644 --- a/README.md +++ b/README.md @@ -25,6 +25,12 @@ const persist = createMailboxPersist(db, { }); ``` +`resolveRefs` runs after `upstream` resolves and serially with it, and its +refs are frozen at the frame's first successful insert — a retry still runs +the resolver but a different result on that later call is discarded. Return +a small set with the load-bearing ref first: the list is capped at +`MAX_MAILBOX_REFS` by truncating from the end. + See ARCHITECTURE.md's persist section for the full contract, including `resolveRefs`'s dual-write-failure semantics. diff --git a/src/persist.ts b/src/persist.ts index 1c55434..12f172c 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -115,14 +115,29 @@ export type PersistedMailboxRow = { * recipient — so a host pointing every row at the same upstream entity * (`{ kind: "workbench", id }`) does one lookup, not N. * + * Runs AFTER `upstream` resolves, and serially with it — not concurrently — + * so its latency adds to the call. This keeps refs available before the + * mailbox transaction opens without racing `upstream`'s own effects. + * + * Refs are frozen at the FIRST successful insert for a frame: a retried + * frame (same idempotency key) that reaches `resolveRefs` again still runs + * the resolver — it is not skipped — but a different result is discarded, + * since `onConflictDoNothing` means no row is written for the retry. Do not + * rely on a resolver's return value being applied on any call after the + * first that actually inserts. + * * The result is validated with `MailboxRefArraySchema` and capped at * `MAX_MAILBOX_REFS` the same way `writeMailboxMessage`'s `refs` argument is; * see `boundRefs`. Returning `undefined` (or an empty array) stores no refs. + * Because excess entries are truncated rather than rejected, a resolver + * MUST return a small set with the load-bearing ref FIRST — anything past + * `MAX_MAILBOX_REFS` is silently dropped from the end of the list. * * A throwing `resolveRefs` is handled exactly like a mailbox-write failure - * under the dual-write contract: logged, upstream still runs (it already - * ran, or still will, independently of this), and no mailbox row is written - * for that frame. See ARCHITECTURE.md's persist section. + * under the dual-write contract: logged (naming `resolveRefs` as the failing + * stage), upstream still runs (it already ran, or still will, independently + * of this), and no mailbox row is written for that frame. See + * ARCHITECTURE.md's persist section. */ export type ResolveMailboxRefs = ( args: MailboxPersistArgs & {