From 324a40114f8f312f56e8533786a5e16467e5614b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 26 Sep 2026 17:52:00 -0700 Subject: [PATCH] fix(store): allocate uids in Postgres and enforce uid and Message-ID uniqueness appendMessage took its uid from a counter read when the store opened, and Message-ID dedupe checked the in-memory mirror before inserting, so concurrent writers produced duplicate uids and duplicate messages. It now bumps mailbox_state with UPDATE ... RETURNING inside the insert's transaction, the way moveNativeMailboxMessage does, and inserts with ON CONFLICT DO NOTHING on the Message-ID and sender, resolving to null on a duplicate. write and persist rely on that instead of checking first. appendMessage takes the host-authorized sender: write passes fromAddress, send the caller's sender address, and persist the envelope sender, so a forged From header cannot suppress another sender's mail. The synchronous append cannot allocate atomically; its JSDoc and the README point hosts at the async write paths. Migration 0008 adds sender_address (backfilled from from_address, or '' when a row has none), a unique index on (tenant_id, principal_id, folder, uid) and a partial unique index on (tenant_id, principal_id, folder, message_id, sender_address). Existing duplicates are resolved first without deleting a row: the oldest row keeps its uid and later ones move past both uid_next and the mailbox's highest uid, since a racing 0.1.0 append could leave uid_next behind; a mailbox with no mailbox_state row gets one; and a repeated Message-ID from one sender keeps its row and raw frame with only the cached message_id cleared. --- README.md | 2 + e2e/migrations.test.ts | 77 +++++++++++ e2e/native-store.test.ts | 6 +- e2e/persist.test.ts | 30 +++++ e2e/upgrade-from-0.1.0.test.ts | 76 ++++++++++- e2e/write-failure.test.ts | 2 +- e2e/write.test.ts | 33 +++++ migrations/0008_unique_uid_and_message_id.sql | 97 ++++++++++++++ src/mount.ts | 3 + src/native-store.ts | 120 +++++++++++++----- src/persist.ts | 15 +-- src/schema-check.ts | 1 + src/schema.ts | 21 +++ src/write.ts | 5 +- 14 files changed, 438 insertions(+), 50 deletions(-) create mode 100644 migrations/0008_unique_uid_and_message_id.sql diff --git a/README.md b/README.md index 1717f87..35817ed 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,8 @@ Takes the same arguments as `@intx/db`'s `runMigrations`. `schema` holds the hos | `assertMailboxScope`, `assertMailboxTenantId`, `assertMailboxFrameBytes`, `MailboxScopeIdsSchema` | Scope and frame-size checks the write paths run. | | `MAX_MAILBOX_PAGE_LIMIT`, `MAX_MAILBOX_RECIPIENTS`, `MAX_MAILBOX_FRAME_BYTES` | Page size (200), recipients per frame, and frame size (1 MiB) limits. | +The interface's synchronous `append` must return a uid before it reaches Postgres, so it cannot allocate one atomically and two stores appending at once can lose a write. Write with `appendMessage`, `writeMailboxMessage` or `createMailboxPersist` instead. + ## Using with Interchange The host's tenant middleware sets `tenant` and `principal` on the context; the routes read the caller from there, the same principal `requireGrant` authorizes. Grant each person `mailbox:*` with the `read`, `create` and `manage` actions they need. diff --git a/e2e/migrations.test.ts b/e2e/migrations.test.ts index 1c80d6c..0e16aaa 100644 --- a/e2e/migrations.test.ts +++ b/e2e/migrations.test.ts @@ -85,6 +85,7 @@ describe("runMailboxMigrations", () => { "raw", "references", "refs", + "sender_address", "subject", "tenant_id", "to_addresses", @@ -109,10 +110,12 @@ describe("runMailboxMigrations", () => { // filter, and the thread's own oldest-first keyset. // The schema.ts parity suite below holds schema.ts to this same list. expect(mailIndexes.map((i) => i.indexname)).toEqual([ + "principal_mail_folder_message_id_sender_address_idx", "principal_mail_pkey", "principal_mail_refs_idx", "principal_mail_tenant_id_principal_id_created_at_id_asc_idx", "principal_mail_tenant_id_principal_id_created_at_id_idx", + "principal_mail_tenant_id_principal_id_folder_uid_idx", "principal_mail_tenant_id_principal_id_message_id_idx", "principal_mail_tenant_id_principal_id_message_key_idx", ]); @@ -154,6 +157,80 @@ describe("runMailboxMigrations", () => { }); }); + test("0008 renumbers duplicate uids and clears repeated Message-IDs without dropping a row", async () => { + await fromEmpty(async ({ db }) => { + await applyMailboxMigrations(db, "public"); + await admin.unsafe(` + DROP INDEX "mailbox"."principal_mail_tenant_id_principal_id_folder_uid_idx"; + DROP INDEX "mailbox"."principal_mail_folder_message_id_sender_address_idx"; + INSERT INTO "tenant" VALUES ('t8') ON CONFLICT DO NOTHING; + INSERT INTO "principal" VALUES ('p8', 't8', 'p8') ON CONFLICT DO NOTHING; + INSERT INTO "mailbox"."mailbox_state" VALUES ('t8', 'p8', 'INBOX', 1, 3, 3); + INSERT INTO "mailbox"."principal_mail" + ("id", "tenant_id", "principal_id", "address", "direction", "raw", + "message_id", "sender_address", "created_at", "uid", "modseq") + VALUES + ('a', 't8', 'p8', 'x', 'inbound', '\\x00', '', 'f@x', '2026-01-01', 1, 1), + ('b', 't8', 'p8', 'x', 'inbound', '\\x00', '', 'f@x', '2026-01-02', 2, 2), + ('c', 't8', 'p8', 'x', 'inbound', '\\x00', '', 'f@x', '2026-01-03', 2, 3), + ('d', 't8', 'p8', 'x', 'inbound', '\\x00', NULL, 'f@x', '2026-01-04', 2, 3), + ('e', 't8', 'p8', 'x', 'inbound', '\\x00', '', 'g@x', '2026-01-05', 5, 4); + `); + + await applyMailboxMigrations(db, "public"); + + const rows = await admin< + { id: string; uid: string; message_id: string | null }[] + >`SELECT "id", "uid", "message_id" FROM "mailbox"."principal_mail" + WHERE "tenant_id" = 't8' ORDER BY "id"`; + expect(rows.map((r) => [r.id, Number(r.uid), r.message_id])).toEqual([ + ["a", 1, ""], + ["b", 2, null], + ["c", 6, ""], + ["d", 7, null], + ["e", 5, ""], + ]); + const [state] = await admin<{ uid_next: string }[]>` + SELECT "uid_next" FROM "mailbox"."mailbox_state" WHERE "tenant_id" = 't8'`; + expect(Number(state!.uid_next)).toBe(8); + }); + }); + + test("0008 renumbers a mailbox with no mailbox_state row and creates one past its uids", async () => { + await fromEmpty(async ({ db }) => { + await applyMailboxMigrations(db, "public"); + await admin.unsafe(` + DROP INDEX "mailbox"."principal_mail_tenant_id_principal_id_folder_uid_idx"; + INSERT INTO "tenant" VALUES ('t8') ON CONFLICT DO NOTHING; + INSERT INTO "principal" VALUES ('p8', 't8', 'p8') ON CONFLICT DO NOTHING; + INSERT INTO "mailbox"."principal_mail" + ("id", "tenant_id", "principal_id", "address", "direction", "raw", + "sender_address", "created_at", "folder", "uid", "modseq") + VALUES + ('a', 't8', 'p8', 'x', 'inbound', '\\x00', 'f@x', '2026-01-01', 'Archive', 1, 1), + ('b', 't8', 'p8', 'x', 'inbound', '\\x00', 'f@x', '2026-01-02', 'Archive', 1, 2), + ('c', 't8', 'p8', 'x', 'inbound', '\\x00', 'f@x', '2026-01-03', 'Archive', 2, 3); + `); + + await applyMailboxMigrations(db, "public"); + + const rows = await admin<{ id: string; uid: string }[]>` + SELECT "id", "uid" FROM "mailbox"."principal_mail" + WHERE "tenant_id" = 't8' ORDER BY "id"`; + expect(rows.map((r) => [r.id, Number(r.uid)])).toEqual([ + ["a", 1], + ["b", 3], + ["c", 2], + ]); + const state = await admin<{ uid_next: string; highest_modseq: string }[]>` + SELECT "uid_next", "highest_modseq" FROM "mailbox"."mailbox_state" + WHERE "tenant_id" = 't8' AND "folder" = 'Archive'`; + expect( + state.map((r) => [Number(r.uid_next), Number(r.highest_modseq)]), + ).toEqual([[4, 3]]); + }); + }); + test("0005 leaves uid and modseq NOT NULL: every write path is the native store now", async () => { await fromEmpty(async ({ db }) => { await applyMailboxMigrations(db, "public"); diff --git a/e2e/native-store.test.ts b/e2e/native-store.test.ts index 14c3265..291c734 100644 --- a/e2e/native-store.test.ts +++ b/e2e/native-store.test.ts @@ -129,7 +129,11 @@ describe("native MailboxStore over the principal mailbox tables", () => { // Moving again gets the next uid in Archive, not a collision. const inbox2 = await store.open("INBOX"); - const uid2 = inbox2.append(new Uint8Array([10]), envelope(), []); + const uid2 = inbox2.append( + new Uint8Array([10]), + envelope({ messageId: "" }), + [], + ); await inbox2.settled; const secondMove = await store.move("INBOX", uid2, "Archive"); expect(secondMove).toBe(2); diff --git a/e2e/persist.test.ts b/e2e/persist.test.ts index f5c4270..73f88f8 100644 --- a/e2e/persist.test.ts +++ b/e2e/persist.test.ts @@ -599,3 +599,33 @@ describe("frame size and recipient hard caps", () => { expect(calls).toHaveLength(2); }); }); + +describe("Message-ID dedupe", () => { + test("keys on the authorized envelope sender, not the From header", async () => { + const persist = createMailboxPersist(db, { + upstream: async () => undefined, + authorizeSender: () => ACTIVE, + }); + // A forged From reusing SENDER's Message-ID from another envelope sender + // lands first, and SENDER's own mail still delivers. + await persist(args({ senderAddress: "intruder@acme.example" })); + await persist(args()); + // A retry from SENDER with a reformatted From is still a duplicate. + await persist( + args({ + raw: buildMailFrame({ + from: `Heartbeat <${SENDER}>`, + to: "usr_user-1@acme.example", + subject: "Run finished", + body: "Body", + messageId: "", + }), + }), + ); + const rows = await rowsFor("acme", "user-1"); + expect(rows.map((r) => r.senderAddress).sort()).toEqual([ + SENDER, + "intruder@acme.example", + ]); + }); +}); diff --git a/e2e/upgrade-from-0.1.0.test.ts b/e2e/upgrade-from-0.1.0.test.ts index a262649..576a3db 100644 --- a/e2e/upgrade-from-0.1.0.test.ts +++ b/e2e/upgrade-from-0.1.0.test.ts @@ -4,6 +4,7 @@ import { buildMailFrame, createMailboxPersist, moveNativeMailboxMessage, + openNativeMailboxStore as openPublishedStore, runMailboxMigrations as runPublishedMigrations, } from "@corbits/mailbox-0.1.0"; import { runMailboxMigrations, type MailboxDb } from "../src/index.js"; @@ -23,15 +24,17 @@ afterAll(async () => { type Snapshot = { mail: { folder: string; flags: string[] }[]; state: unknown[]; + columns: string[]; + indexes: string[]; }; async function snapshot(db: MailboxDb): Promise { const [row] = await db.execute<{ snapshot: Snapshot }>(sql` SELECT jsonb_build_object( - 'mail', (SELECT jsonb_agg(to_jsonb(m) ORDER BY m."id") FROM "mailbox"."principal_mail" m), + 'mail', (SELECT jsonb_agg(to_jsonb(m) - 'sender_address' ORDER BY m."id") FROM "mailbox"."principal_mail" m), 'state', (SELECT jsonb_agg(to_jsonb(s) ORDER BY s."tenant_id", s."principal_id", s."folder") FROM "mailbox"."mailbox_state" s), - 'columns', (SELECT jsonb_agg(jsonb_build_array(table_name, column_name, data_type, is_nullable, column_default) + 'columns', (SELECT jsonb_agg(concat_ws(' ', table_name, column_name, data_type, is_nullable, column_default) ORDER BY table_name, column_name) FROM information_schema.columns WHERE table_schema = 'mailbox' AND table_name <> 'corbits_mailbox_migrations'), @@ -102,8 +105,73 @@ test("a database migrated by the published 0.1.0 runner upgrades with every row await runMailboxMigrations(config, { schema: "public" }); expect(await ledgerExists(db)).toBe(false); - expect(await snapshot(db)).toEqual(before); + const after = await snapshot(db); + // 0008 adds sender_address, backfilled from the From, and the uid and + // Message-ID unique indexes, and nothing else. + expect(after.mail).toEqual(before.mail); + expect(after.state).toEqual(before.state); + expect(after.columns.filter((c) => !before.columns.includes(c))).toEqual([ + "principal_mail sender_address text NO ''::text", + ]); + expect(before.columns.filter((c) => !after.columns.includes(c))).toEqual([]); + const senders = await db.execute<{ same: boolean }>( + sql`SELECT "sender_address" = "from_address" AS "same" FROM "mailbox"."principal_mail"`, + ); + expect(senders.map((r) => r.same)).toEqual([true, true]); + expect(after.indexes.filter((def) => !before.indexes.includes(def))).toEqual([ + "CREATE UNIQUE INDEX principal_mail_folder_message_id_sender_address_idx ON mailbox.principal_mail USING btree (tenant_id, principal_id, folder, message_id, sender_address) WHERE (message_id IS NOT NULL)", + "CREATE UNIQUE INDEX principal_mail_tenant_id_principal_id_folder_uid_idx ON mailbox.principal_mail USING btree (tenant_id, principal_id, folder, uid)", + ]); await runMailboxMigrations(config, { schema: "public" }); - expect(await snapshot(db)).toEqual(before); + expect(await snapshot(db)).toEqual(after); +}); + +test("a 0.1.0 database whose racing appends left duplicate uids and uid_next behind upgrades", async () => { + const { db, config, close } = await createEmptyTestDb(); + try { + await runPublishedMigrations(db); + await seedScope(db, "t2", "carol"); + const scope = { tenantId: "t2", principalId: "carol", folder: "INBOX" }; + const envelope = (messageId: string) => ({ + messageId, + from: "f@t2.example", + to: ["carol@t2.example"], + subject: messageId, + date: new Date(), + inReplyTo: undefined, + references: [], + interchangeType: undefined, + interchangeCorrelationId: undefined, + }); + // Two 0.1.0 stores open at uid_next 1: one appends uids 1-3, then the other + // appends uid 1 and writes uid_next back to 2. + const first = await openPublishedStore(db, scope); + const second = await openPublishedStore(db, scope); + for (const id of ["", "", ""]) { + first.append(new Uint8Array([1]), envelope(id), []); + } + await first.settled; + second.append(new Uint8Array([1]), envelope(""), []); + await second.settled; + + await runMailboxMigrations(config, { schema: "public" }); + + const rows = await db.execute<{ message_id: string; uid: string }>( + sql`SELECT "message_id", "uid" FROM "mailbox"."principal_mail" + WHERE "tenant_id" = 't2' ORDER BY "uid"`, + ); + expect(rows.map((r) => [r.message_id, Number(r.uid)])).toEqual([ + ["", 1], + ["", 2], + ["", 3], + ["", 4], + ]); + const [state] = await db.execute<{ uid_next: string }>( + sql`SELECT "uid_next" FROM "mailbox"."mailbox_state" WHERE "tenant_id" = 't2'`, + ); + expect(Number(state!.uid_next)).toBe(5); + } finally { + await close(); + } }); diff --git a/e2e/write-failure.test.ts b/e2e/write-failure.test.ts index 6cccbbe..242f6bb 100644 --- a/e2e/write-failure.test.ts +++ b/e2e/write-failure.test.ts @@ -102,7 +102,7 @@ describe("a refused append", () => { interchangeCorrelationId: undefined, }; await expect( - store.appendMessage(new Uint8Array([1]), envelope, []), + store.appendMessage(new Uint8Array([1]), envelope, [], envelope.from), ).rejects.toThrow(); expect(store.uidNext).toBe(1); expect(store.highestModSeq).toBe(0); diff --git a/e2e/write.test.ts b/e2e/write.test.ts index 1f16bbf..c0a5545 100644 --- a/e2e/write.test.ts +++ b/e2e/write.test.ts @@ -193,3 +193,36 @@ describe("deliverInboxItems", () => { expect(results[0]!.id).not.toBeNull(); }); }); + +describe("concurrent appends to one mailbox", () => { + test("five writes get five distinct uids", async () => { + const written = await Promise.all( + [0, 1, 2, 3, 4].map((i) => + writeMailboxMessage(db, args({ subject: `m${i}` })), + ), + ); + expect(written.map((w) => w?.uid).sort()).toEqual([1, 2, 3, 4, 5]); + const store = await openNativeMailboxStore(db, { + tenantId: "t1", + principalId: "p1", + folder: "INBOX", + }); + expect(store.messages.map((m) => m.uid)).toEqual([1, 2, 3, 4, 5]); + expect(store.uidNext).toBe(6); + }); + + test("three writes with one Message-ID land one row", async () => { + const written = await Promise.all( + [0, 1, 2].map(() => + writeMailboxMessage(db, args({ messageId: "" })), + ), + ); + expect(written.filter((w) => w !== null)).toHaveLength(1); + const store = await openNativeMailboxStore(db, { + tenantId: "t1", + principalId: "p1", + folder: "INBOX", + }); + expect(store.messages).toHaveLength(1); + }); +}); diff --git a/migrations/0008_unique_uid_and_message_id.sql b/migrations/0008_unique_uid_and_message_id.sql new file mode 100644 index 0000000..2a703df --- /dev/null +++ b/migrations/0008_unique_uid_and_message_id.sql @@ -0,0 +1,97 @@ +-- One row per uid, and per Message-ID from one sender, in each (tenant, +-- principal, folder) mailbox. The sender is the host-authorized envelope +-- sender, not the From header, so a forged From cannot suppress mail; rows +-- written before it was recorded fall back to their From, or to '' when they +-- have none, so the key is never NULL. Before 0.2.0 +-- appends raced on an in-memory counter, so a database may already hold +-- duplicates; they are resolved first, without deleting a row. Duplicate uids +-- keep the oldest row on its uid and move each later one past both uid_next +-- and the mailbox's highest uid, since a racing 0.1.0 writer could leave +-- uid_next behind. A later copy of a Message-ID keeps its row and raw frame, +-- and only its cached "message_id" is cleared. +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'mailbox' AND table_name = 'principal_mail' + AND column_name = 'sender_address' + ) THEN + ALTER TABLE "mailbox"."principal_mail" + ADD COLUMN "sender_address" text NOT NULL DEFAULT ''; + UPDATE "mailbox"."principal_mail" + SET "sender_address" = "from_address" + WHERE "from_address" IS NOT NULL; + END IF; +END $$; + +DO $$ +BEGIN + IF to_regclass('"mailbox"."principal_mail_tenant_id_principal_id_folder_uid_idx"') IS NULL THEN + WITH dup AS ( + SELECT "id", "tenant_id", "principal_id", "folder", + row_number() OVER ( + PARTITION BY "tenant_id", "principal_id", "folder", "uid" + ORDER BY "created_at", "id" + ) AS "n" + FROM "mailbox"."principal_mail" + ), top AS ( + SELECT "tenant_id", "principal_id", "folder", max("uid") AS "uid" + FROM "mailbox"."principal_mail" + GROUP BY "tenant_id", "principal_id", "folder" + ), moved AS ( + SELECT d."id", + GREATEST(coalesce(s."uid_next", 1) - 1, t."uid") + row_number() OVER ( + PARTITION BY d."tenant_id", d."principal_id", d."folder" + ORDER BY d."id" + ) AS "uid" + FROM dup AS d + JOIN top AS t + ON t."tenant_id" = d."tenant_id" AND t."principal_id" = d."principal_id" + AND t."folder" = d."folder" + LEFT JOIN "mailbox"."mailbox_state" AS s + ON s."tenant_id" = d."tenant_id" AND s."principal_id" = d."principal_id" + AND s."folder" = d."folder" + WHERE d."n" > 1 + ) + UPDATE "mailbox"."principal_mail" AS pm + SET "uid" = moved."uid" + FROM moved + WHERE pm."id" = moved."id"; + INSERT INTO "mailbox"."mailbox_state" + ("tenant_id", "principal_id", "folder", "uid_validity", "uid_next", "highest_modseq") + SELECT "tenant_id", "principal_id", "folder", + extract(epoch FROM min("created_at"))::bigint, + max("uid") + 1, + max("modseq") + FROM "mailbox"."principal_mail" + GROUP BY "tenant_id", "principal_id", "folder" + ON CONFLICT ("tenant_id", "principal_id", "folder") DO UPDATE + SET "uid_next" = GREATEST("mailbox_state"."uid_next", excluded."uid_next"), + "highest_modseq" = GREATEST("mailbox_state"."highest_modseq", excluded."highest_modseq"); + CREATE UNIQUE INDEX "principal_mail_tenant_id_principal_id_folder_uid_idx" + ON "mailbox"."principal_mail" ("tenant_id", "principal_id", "folder", "uid"); + END IF; +END $$; + +DO $$ +BEGIN + IF to_regclass('"mailbox"."principal_mail_folder_message_id_sender_address_idx"') IS NULL THEN + UPDATE "mailbox"."principal_mail" AS pm + SET "message_id" = NULL + FROM ( + SELECT "id", + row_number() OVER ( + PARTITION BY "tenant_id", "principal_id", "folder", "message_id", + "sender_address" + ORDER BY "created_at", "id" + ) AS "n" + FROM "mailbox"."principal_mail" + WHERE "message_id" IS NOT NULL + ) AS dup + WHERE pm."id" = dup."id" AND dup."n" > 1; + CREATE UNIQUE INDEX "principal_mail_folder_message_id_sender_address_idx" + ON "mailbox"."principal_mail" + ("tenant_id", "principal_id", "folder", "message_id", "sender_address") + WHERE "message_id" IS NOT NULL; + END IF; +END $$; diff --git a/src/mount.ts b/src/mount.ts index fbe24d2..27b4c7b 100644 --- a/src/mount.ts +++ b/src/mount.ts @@ -527,7 +527,10 @@ export function createMailboxRoutes( interchangeCorrelationId: undefined, }, [], + fromAddress, ); + // A freshly minted Message-ID cannot already be in Sent. + if (uid === null) throw new Error(`Sent already holds ${messageId}`); 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 3e82394..87e0f59 100644 --- a/src/native-store.ts +++ b/src/native-store.ts @@ -1,4 +1,4 @@ -import { sql } from "drizzle-orm"; +import { sql, TransactionRollbackError } from "drizzle-orm"; import type { MailboxStore, StoredEnvelope, @@ -41,9 +41,13 @@ function pgTextArrayLiteral(items: readonly string[]): string { * 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. + * `settled` never rejects, and `append` must return its uid synchronously, so + * it takes the uid from this instance's counter and cannot allocate it + * atomically: two instances appending at once collide on the unique uid index + * and the later write is lost. Hosts write through `appendMessage`, + * `writeMailboxMessage` or `createMailboxPersist` instead, which allocate the + * uid from `mailbox_state` inside the insert's transaction and reject when the + * insert fails. */ export type NativeMailboxStore = MailboxStore & { readonly tenantId: string; @@ -51,12 +55,18 @@ 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. */ + /** + * Appends with a uid allocated atomically in Postgres. Resolves to that uid + * once the row has landed, or to `null` when this mailbox already holds the + * envelope's Message-ID from `sender`, the host-authorized envelope sender. + * Rejects when the insert fails. + */ appendMessage( raw: Uint8Array, envelope: StoredEnvelope, flags: string[], - ): Promise; + sender: string, + ): Promise; }; // Postgres text and jsonb refuse U+0000, so a NUL in any cached header or @@ -204,9 +214,12 @@ export async function openNativeMailboxStore( // caller that needs to observe a failure awaits the promise `enqueue` // returns, not `store.settled`. let settled: Promise = Promise.resolve(); - function enqueue(work: () => Promise): Promise { - const done = settled.then(work).then(() => undefined); - settled = done.catch(() => undefined); + function enqueue(work: () => Promise): Promise { + const done = settled.then(work); + settled = done.then( + () => undefined, + () => undefined, + ); return done; } @@ -218,7 +231,7 @@ export async function openNativeMailboxStore( return msg; } - function appendLocal( + function append( raw: Uint8Array, rawEnvelope: StoredEnvelope, flags: string[], @@ -239,17 +252,17 @@ export async function openNativeMailboxStore( messages.push(message); byUid.set(uid, message); - const written = enqueue(() => + 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") + "subject", "from_address", "sender_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.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[] @@ -263,7 +276,7 @@ export async function openNativeMailboxStore( `); }), ); - return { uid, modseq, message, written }; + return uid; } const store: NativeMailboxStore = { @@ -284,25 +297,68 @@ export async function openNativeMailboxStore( return settled; }, - append(raw, envelope, flags) { - return appendLocal(raw, envelope, flags).uid; - }, + append, - 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; + async appendMessage(raw, rawEnvelope, flags, sender) { + const envelope = cleanEnvelope(rawEnvelope); + const rowId = crypto.randomUUID(); + const landed = await enqueue(() => + db.transaction(async (tx) => { + const [counters] = await tx.execute<{ + uid: string | number; + modseq: string | number; + }>(sql` + UPDATE "mailbox"."mailbox_state" + SET "uid_next" = "uid_next" + 1, "highest_modseq" = "highest_modseq" + 1 + WHERE "tenant_id" = ${tenantId} AND "principal_id" = ${principalId} AND "folder" = ${folder} + RETURNING "uid_next" - 1 AS "uid", "highest_modseq" AS "modseq" + `); + const uid = Number(counters!.uid); + const modseq = Number(counters!.modseq); + // The mailbox_state row lock taken above serializes appends to this + // mailbox, so a concurrent writer with the same Message-ID has + // either committed (and this conflicts) or waits behind this one. + const inserted = await tx.execute(sql` + INSERT INTO "mailbox"."principal_mail" + ("id", "tenant_id", "principal_id", "address", "direction", "raw", + "subject", "from_address", "sender_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}, + ${withoutNul(sender)}, ${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[] + ) + ON CONFLICT ("tenant_id", "principal_id", "folder", "message_id", "sender_address") + WHERE "message_id" IS NOT NULL + DO NOTHING + RETURNING "id" + `); + if (inserted.length === 0) { + tx.rollback(); + } + return { uid, modseq }; + }), + ).catch((err: unknown) => { + if (err instanceof TransactionRollbackError) return null; throw err; - } + }); + if (landed === null) return null; + + const { uid, modseq } = landed; + state.uidNext = Math.max(state.uidNext, uid + 1); + state.highestModSeq = Math.max(state.highestModSeq, modseq); + const message: StoredMessage & { rowId: string } = { + rowId, + uid, + modseq, + flags: new Set(flags), + envelope, + }; + messages.push(message); + byUid.set(uid, message); return uid; }, diff --git a/src/persist.ts b/src/persist.ts index bc0d52f..8ced5ea 100644 --- a/src/persist.ts +++ b/src/persist.ts @@ -207,21 +207,16 @@ export function createMailboxPersist( const references = decoded?.references ?? []; // One append per recipient, into their own INBOX — the native store has - // no multi-recipient batch. Deduped on messageId within each recipient's - // mailbox: a retried frame (same Message-ID) never delivers twice to the - // same principal. + // no multi-recipient batch. Deduped on messageId and the authorized + // envelope sender within each recipient's mailbox: a retried frame never + // delivers twice to the same principal, and a forged From cannot suppress + // another sender's mail. for (const recipient of resolved) { const store = await openNativeMailboxStore(db, { tenantId: auth.tenantId, principalId: recipient.principalId, folder: "INBOX", }); - if ( - messageId !== null && - store.messages.some((m) => m.envelope.messageId === messageId) - ) { - continue; - } const uid = await store.appendMessage( raw, { @@ -236,7 +231,9 @@ export function createMailboxPersist( interchangeCorrelationId: undefined, }, [], + senderAddress, ); + if (uid === null) continue; announce({ id: `${auth.tenantId}:${recipient.principalId}:INBOX:${uid}`, tenantId: auth.tenantId, diff --git a/src/schema-check.ts b/src/schema-check.ts index d0d4f43..5194bd5 100644 --- a/src/schema-check.ts +++ b/src/schema-check.ts @@ -35,6 +35,7 @@ import { principalMail } from "./schema.js"; */ const DATA_TYPE_BY_SQL_TYPE: Record = { text: "text", + bigint: "bigint", jsonb: "jsonb", bytea: "bytea", timestamp: "timestamp without time zone", diff --git a/src/schema.ts b/src/schema.ts index c029bfc..37ed978 100644 --- a/src/schema.ts +++ b/src/schema.ts @@ -1,4 +1,5 @@ import { + bigint, customType, index, jsonb, @@ -93,7 +94,14 @@ export const principalMail = mailboxPgSchema.table( raw: bytea("raw").notNull(), subject: text("subject"), fromAddress: text("from_address"), + // The host-authorized envelope sender, the Message-ID dedupe key's sender + // half. Added by migration `0008_unique_uid_and_message_id`. + senderAddress: text("sender_address").notNull().default(""), messageKey: text("message_key"), + // The IMAP mailbox and uid migration `0004_native_mailbox_store` added, + // named here for the unique index below. + folder: text("folder").notNull().default("INBOX"), + uid: bigint("uid", { mode: "number" }).notNull(), // The threading headers, cached off `raw` at write for the SAME reason // `subject` and `from_address` are: list never loads the MIME frame, so a // client renders a thread from the projection alone instead of fetching @@ -170,6 +178,19 @@ export const principalMail = mailboxPgSchema.table( t.principalId, t.messageId, ), + // One row per uid in each mailbox. Created by migration + // `0008_unique_uid_and_message_id`. + uniqueIndex("principal_mail_tenant_id_principal_id_folder_uid_idx").on( + t.tenantId, + t.principalId, + t.folder, + t.uid, + ), + // The append dedupe key: one row per Message-ID from one envelope sender + // in each mailbox. Created by migration `0008_unique_uid_and_message_id`. + uniqueIndex("principal_mail_folder_message_id_sender_address_idx") + .on(t.tenantId, t.principalId, t.folder, t.messageId, t.senderAddress) + .where(sql`${t.messageId} IS NOT NULL`), // `readMailboxThread`'s own keyset order: oldest first, `(created_at, id)`, // scoped the same way every other access path here is. The list-path index // above covers the same three columns in the opposite direction and a diff --git a/src/write.ts b/src/write.ts index 57906db..8157a28 100644 --- a/src/write.ts +++ b/src/write.ts @@ -222,9 +222,6 @@ export async function writeMailboxMessage( principalId: args.principalId, folder, }); - if (store.messages.some((m) => m.envelope.messageId === messageId)) { - return null; - } const uid = await store.appendMessage( raw, { @@ -239,7 +236,9 @@ export async function writeMailboxMessage( interchangeCorrelationId: undefined, }, [], + args.fromAddress, ); + if (uid === null) return null; const id = `${args.tenantId}:${args.principalId}:${folder}:${uid}`; if (bus) {