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) {