From 780b43acb828d597c4fcf73d2a1c1abb6035080b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 26 Sep 2026 17:49:25 -0700 Subject: [PATCH] fix(store): fail the caller when a native-store append fails The store's write queue swallowed every failure, so send, writeMailboxMessage and persist reported a uid (and announced it) for a row that never landed. appendMessage resolves only once its own insert and counter update commit, in one transaction, and on failure rolls the in-memory uidNext, highestModSeq and message back and rejects. Every append in the package goes through it. NUL is dropped from header values and cached envelope fields, since Postgres text and jsonb refuse it; raw keeps every byte. --- e2e/write-failure.test.ts | 215 ++++++++++++++++++++++++++++++++++++++ src/frame.ts | 8 +- src/mount.ts | 3 +- src/native-store.ts | 145 +++++++++++++++++-------- src/persist.ts | 3 +- src/write.ts | 3 +- 6 files changed, 326 insertions(+), 51 deletions(-) create mode 100644 e2e/write-failure.test.ts diff --git a/e2e/write-failure.test.ts b/e2e/write-failure.test.ts new file mode 100644 index 0000000..6cccbbe --- /dev/null +++ b/e2e/write-failure.test.ts @@ -0,0 +1,215 @@ +// A write that Postgres refuses must fail the caller, not report a uid for a +// row that never landed. The refusal here is a real trigger on the mail plane. +import { + afterAll, + beforeAll, + beforeEach, + describe, + expect, + test, +} from "bun:test"; +import { sql } from "drizzle-orm"; +import { + createMailboxRoutes, + type OutgoingMailboxMessage, +} from "../src/mount.js"; +import { + createInMemoryMailboxEventBus, + type MailboxEvent, +} from "../src/bus.js"; +import { openNativeMailboxStore } from "../src/native-store.js"; +import { createMailboxPersist } from "../src/persist.js"; +import { buildMailFrame } from "../src/frame.js"; +import { writeMailboxMessage } from "../src/write.js"; +import { allowAllGrants, mountAs, seedScope, withTestDb } from "./helpers.js"; +import type { MailboxDb } from "../src/db.js"; + +const SCOPE = { tenantId: "t1", principalId: "p1" }; +let db: MailboxDb; + +beforeAll(async () => { + db = await withTestDb(); + await db.execute(sql` + CREATE OR REPLACE FUNCTION "mailbox"."test_refuse_boom"() RETURNS trigger AS $$ + BEGIN + IF NEW."subject" = 'boom' THEN RAISE EXCEPTION 'refused'; END IF; + RETURN NEW; + END $$ LANGUAGE plpgsql + `); + await db.execute(sql` + CREATE TRIGGER "test_refuse_boom" BEFORE INSERT ON "mailbox"."principal_mail" + FOR EACH ROW EXECUTE FUNCTION "mailbox"."test_refuse_boom"() + `); +}); + +afterAll(async () => { + await db.execute( + sql`DROP TRIGGER IF EXISTS "test_refuse_boom" ON "mailbox"."principal_mail"`, + ); + await db.execute(sql`DROP FUNCTION IF EXISTS "mailbox"."test_refuse_boom"()`); +}); + +beforeEach(async () => { + db = await withTestDb(); + await seedScope(db, SCOPE.tenantId, SCOPE.principalId); +}); + +async function mailRows(folder: string) { + return db.execute<{ uid: string; subject: string | null }>( + sql`SELECT "uid", "subject" FROM "mailbox"."principal_mail" + WHERE "folder" = ${folder} ORDER BY "uid"`, + ); +} + +function sendApp(deliveries: OutgoingMailboxMessage[]) { + return mountAs( + SCOPE, + createMailboxRoutes({ + db, + requireGrant: allowAllGrants, + bus: createInMemoryMailboxEventBus(), + senderAddressFor: () => "p1@t1.example", + deliver: (message) => { + deliveries.push(message); + }, + }), + ); +} + +function send(app: ReturnType, subject: string) { + return app.request("/me/inbox/send", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ to: ["bob@example.com"], subject, body: "b" }), + }); +} + +describe("a refused append", () => { + test("appendMessage rejects and rolls the store's counters back", async () => { + const store = await openNativeMailboxStore(db, { + ...SCOPE, + folder: "INBOX", + }); + const envelope = { + messageId: "", + from: "a@t1.example", + to: ["p1@t1.example"], + subject: "boom", + date: new Date(), + inReplyTo: undefined, + references: [], + interchangeType: undefined, + interchangeCorrelationId: undefined, + }; + await expect( + store.appendMessage(new Uint8Array([1]), envelope, []), + ).rejects.toThrow(); + expect(store.uidNext).toBe(1); + expect(store.highestModSeq).toBe(0); + expect(store.messages).toHaveLength(0); + }); + + test("send fails and never reaches deliver", async () => { + const deliveries: OutgoingMailboxMessage[] = []; + const app = sendApp(deliveries); + expect((await send(app, "boom")).status).toBe(500); + expect(deliveries).toHaveLength(0); + expect(await mailRows("Sent")).toHaveLength(0); + + const ok = await send(app, "fine"); + expect(ok.status).toBe(200); + expect(((await ok.json()) as { uid: number }).uid).toBe(1); + }); + + test("writeMailboxMessage rejects and publishes nothing", async () => { + const bus = createInMemoryMailboxEventBus(); + const events: MailboxEvent[] = []; + bus.subscribe(SCOPE, (event) => events.push(event)); + await expect( + writeMailboxMessage( + db, + { + ...SCOPE, + address: "p1@t1.example", + fromAddress: "a@t1.example", + subject: "boom", + body: "b", + }, + bus, + ), + ).rejects.toThrow(); + expect(events).toHaveLength(0); + expect(await mailRows("INBOX")).toHaveLength(0); + }); + + test("persist announces no row it did not write", async () => { + const announced: string[] = []; + const persist = createMailboxPersist(db, { + upstream: async () => undefined, + authorizeSender: () => ({ tenantId: "t1", domain: "t1.example" }), + onRow: (row) => announced.push(row.id), + }); + await persist({ + senderAddress: "a@t1.example", + recipients: ["p1@t1.example"], + raw: buildMailFrame({ + from: "a@t1.example", + to: "p1@t1.example", + subject: "boom", + body: "b", + messageId: "", + }), + }); + expect(announced).toHaveLength(0); + expect(await mailRows("INBOX")).toHaveLength(0); + }); +}); + +describe("NUL in header and envelope fields", () => { + test("send stores the message with NUL dropped from subject and recipient", async () => { + const app = sendApp([]); + const res = await app.request("/me/inbox/send", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + to: ["bob\u0000@example.com"], + subject: "hi\u0000", + body: "b", + }), + }); + expect(res.status).toBe(200); + const store = await openNativeMailboxStore(db, { + ...SCOPE, + folder: "Sent", + }); + expect(store.messages[0]?.envelope.subject).toBe("hi"); + expect(store.messages[0]?.envelope.to).toEqual(["bob@example.com"]); + }); + + test("writeMailboxMessage stores a NUL subject without it", async () => { + const written = await writeMailboxMessage(db, { + ...SCOPE, + address: "p1@t1.example", + fromAddress: "a@t1.example", + subject: "x\u0000y", + body: "b", + }); + expect(written?.uid).toBe(1); + expect((await mailRows("INBOX"))[0]?.subject).toBe("xy"); + }); + + test("persist stores a frame whose Subject carries NUL", async () => { + const persist = createMailboxPersist(db, { + upstream: async () => undefined, + authorizeSender: () => ({ tenantId: "t1", domain: "t1.example" }), + }); + await persist({ + senderAddress: "a@t1.example", + recipients: ["p1@t1.example"], + raw: new TextEncoder().encode( + "From: a@t1.example\r\nSubject: bad\u0000\r\nMessage-ID: \r\n\r\nhi\r\n", + ), + }); + expect((await mailRows("INBOX"))[0]?.subject).toBe("bad"); + }); +}); diff --git a/src/frame.ts b/src/frame.ts index 2742e24..b57d15d 100644 --- a/src/frame.ts +++ b/src/frame.ts @@ -7,9 +7,13 @@ import { // Header values are single-line by contract; anything reaching a header // (an externally-supplied subject in particular) is flattened so it can -// never smuggle in extra headers via embedded newlines. +// never smuggle in extra headers via embedded newlines. NUL is dropped: no +// header may carry it, and Postgres text cannot store it. export function headerValue(value: string): string { - return value.replace(/[\r\n]+/g, " ").trim(); + return value + .replaceAll("\0", "") + .replace(/[\r\n]+/g, " ") + .trim(); } /** diff --git a/src/mount.ts b/src/mount.ts index 747a120..fbe24d2 100644 --- a/src/mount.ts +++ b/src/mount.ts @@ -513,7 +513,7 @@ export function createMailboxRoutes( db, inFolder(resolved, "Sent"), ); - const uid = sentStore.append( + const uid = await sentStore.appendMessage( raw, { messageId, @@ -528,7 +528,6 @@ export function createMailboxRoutes( }, [], ); - await sentStore.settled; publish(resolved, `Sent:${uid}`, "create"); await deliver({ raw, from: fromAddress, to: parsed.to, messageId }); diff --git a/src/native-store.ts b/src/native-store.ts index d650291..3e82394 100644 --- a/src/native-store.ts +++ b/src/native-store.ts @@ -40,6 +40,10 @@ function pgTextArrayLiteral(items: readonly string[]): string { * chains onto, in order, so two writes for the same mailbox never race each * other at the database. Call `await store.settled` before trusting a write * has actually landed (every test in `native-store.test.ts` does). + * + * `settled` never rejects, so a caller that must fail when its write fails + * uses `appendMessage`, which resolves only once that append has landed and + * rolls the in-memory mirror back when it has not. */ export type NativeMailboxStore = MailboxStore & { readonly tenantId: string; @@ -47,8 +51,37 @@ export type NativeMailboxStore = MailboxStore & { readonly folder: string; /** Resolves once every write queued so far has been applied to Postgres. */ readonly settled: Promise; + /** `append`, resolving once the row has landed; rejects, rolled back, when it does not. */ + appendMessage( + raw: Uint8Array, + envelope: StoredEnvelope, + flags: string[], + ): Promise; }; +// Postgres text and jsonb refuse U+0000, so a NUL in any cached header or +// envelope field would fail the whole insert. `raw` keeps every byte. +function withoutNul(value: string): string { + return value.replaceAll("\0", ""); +} + +function cleanEnvelope(envelope: StoredEnvelope): StoredEnvelope { + return { + messageId: withoutNul(envelope.messageId), + from: withoutNul(envelope.from), + to: envelope.to.map(withoutNul), + subject: withoutNul(envelope.subject), + date: envelope.date, + inReplyTo: + envelope.inReplyTo === undefined + ? undefined + : withoutNul(envelope.inReplyTo), + references: envelope.references.map(withoutNul), + interchangeType: envelope.interchangeType, + interchangeCorrelationId: envelope.interchangeCorrelationId, + }; +} + type Row = { id: string; uid: string | number; @@ -169,13 +202,12 @@ export async function openNativeMailboxStore( // never racing each other. `settled` always resolves (never rejects) so one // failed write does not wedge every later one from being attempted; a // caller that needs to observe a failure awaits the promise `enqueue` - // itself was given, not `store.settled`. + // returns, not `store.settled`. let settled: Promise = Promise.resolve(); - function enqueue(work: () => Promise): void { - settled = settled.then(work).then( - () => undefined, - () => undefined, - ); + function enqueue(work: () => Promise): Promise { + const done = settled.then(work).then(() => undefined); + settled = done.catch(() => undefined); + return done; } function requireMessage(uid: number): StoredMessage & { rowId: string } { @@ -186,6 +218,54 @@ export async function openNativeMailboxStore( return msg; } + function appendLocal( + raw: Uint8Array, + rawEnvelope: StoredEnvelope, + flags: string[], + ) { + const envelope = cleanEnvelope(rawEnvelope); + const uid = state.uidNext; + const modseq = state.highestModSeq + 1; + state.uidNext += 1; + state.highestModSeq = modseq; + const rowId = crypto.randomUUID(); + const message: StoredMessage & { rowId: string } = { + rowId, + uid, + modseq, + flags: new Set(flags), + envelope, + }; + messages.push(message); + byUid.set(uid, message); + + const written = enqueue(() => + db.transaction(async (tx) => { + await tx.execute(sql` + INSERT INTO "mailbox"."principal_mail" + ("id", "tenant_id", "principal_id", "address", "direction", "raw", + "subject", "from_address", "message_id", "in_reply_to", "references", + "to_addresses", "created_at", "folder", "uid", "modseq", "flags") + VALUES ( + ${rowId}, ${tenantId}, ${principalId}, ${envelope.from || envelope.to[0] || ""}, + 'inbound', ${Buffer.from(raw)}, ${envelope.subject}, ${envelope.from}, + ${envelope.messageId || null}, ${envelope.inReplyTo ?? null}, + ${envelope.references.length > 0 ? JSON.stringify(envelope.references) : null}, + ${envelope.to.length > 0 ? JSON.stringify(envelope.to) : null}, + ${envelope.date.toISOString()}, ${folder}, ${uid}, ${modseq}, ${pgTextArrayLiteral(flags)}::text[] + ) + `); + await tx.execute(sql` + UPDATE "mailbox"."mailbox_state" + SET "uid_next" = GREATEST("uid_next", ${uid + 1}), + "highest_modseq" = GREATEST("highest_modseq", ${modseq}) + WHERE "tenant_id" = ${tenantId} AND "principal_id" = ${principalId} AND "folder" = ${folder} + `); + }), + ); + return { uid, modseq, message, written }; + } + const store: NativeMailboxStore = { tenantId, principalId, @@ -205,45 +285,24 @@ export async function openNativeMailboxStore( }, append(raw, envelope, flags) { - const uid = state.uidNext; - const modseq = state.highestModSeq + 1; - state.uidNext += 1; - state.highestModSeq = modseq; - const rowId = crypto.randomUUID(); - const message: StoredMessage & { rowId: string } = { - rowId, - uid, - modseq, - flags: new Set(flags), - envelope, - }; - messages.push(message); - byUid.set(uid, message); + return appendLocal(raw, envelope, flags).uid; + }, - enqueue(() => - db - .execute(sql` - INSERT INTO "mailbox"."principal_mail" - ("id", "tenant_id", "principal_id", "address", "direction", "raw", - "subject", "from_address", "message_id", "in_reply_to", "references", - "to_addresses", "created_at", "folder", "uid", "modseq", "flags") - VALUES ( - ${rowId}, ${tenantId}, ${principalId}, ${envelope.from || envelope.to[0] || ""}, - 'inbound', ${Buffer.from(raw)}, ${envelope.subject}, ${envelope.from}, - ${envelope.messageId || null}, ${envelope.inReplyTo ?? null}, - ${envelope.references.length > 0 ? JSON.stringify(envelope.references) : null}, - ${envelope.to.length > 0 ? JSON.stringify(envelope.to) : null}, - ${envelope.date.toISOString()}, ${folder}, ${uid}, ${modseq}, ${pgTextArrayLiteral(flags)}::text[] - ) - `) - .then(() => - db.execute(sql` - UPDATE "mailbox"."mailbox_state" - SET "uid_next" = ${state.uidNext}, "highest_modseq" = ${state.highestModSeq} - WHERE "tenant_id" = ${tenantId} AND "principal_id" = ${principalId} AND "folder" = ${folder} - `), - ), + async appendMessage(raw, envelope, flags) { + const { uid, modseq, message, written } = appendLocal( + raw, + envelope, + flags, ); + try { + await written; + } catch (err) { + messages.splice(messages.indexOf(message), 1); + byUid.delete(uid); + if (state.uidNext === uid + 1) state.uidNext = uid; + if (state.highestModSeq === modseq) state.highestModSeq = modseq - 1; + throw err; + } return uid; }, diff --git a/src/persist.ts b/src/persist.ts index c61bcc5..bc0d52f 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -222,7 +222,7 @@ export function createMailboxPersist( ) { continue; } - const uid = store.append( + const uid = await store.appendMessage( raw, { messageId: messageId ?? "", @@ -237,7 +237,6 @@ export function createMailboxPersist( }, [], ); - await store.settled; announce({ id: `${auth.tenantId}:${recipient.principalId}:INBOX:${uid}`, tenantId: auth.tenantId, diff --git a/src/write.ts b/src/write.ts index ace81f4..57906db 100644 --- a/src/write.ts +++ b/src/write.ts @@ -225,7 +225,7 @@ export async function writeMailboxMessage( if (store.messages.some((m) => m.envelope.messageId === messageId)) { return null; } - const uid = store.append( + const uid = await store.appendMessage( raw, { messageId, @@ -240,7 +240,6 @@ export async function writeMailboxMessage( }, [], ); - await store.settled; const id = `${args.tenantId}:${args.principalId}:${folder}:${uid}`; if (bus) {