diff --git a/packages/chat/src/migrations.ts b/packages/chat/src/migrations.ts index d9bae76be..16b59d9d1 100644 --- a/packages/chat/src/migrations.ts +++ b/packages/chat/src/migrations.ts @@ -468,6 +468,57 @@ export const chatMigrations: readonly ChatMigration[] = [ WHERE "kind" = 'reply'; `, }, + { + // CL-7199: `createDeliveryThread` select-then-inserted with no + // unique constraint backing the read, same class of bug as 0025's + // root/reply case — a redelivered routine event could insert a + // second delivery thread for the same run. Dedupe existing + // duplicates — keeping the oldest row per (tenant, workbench, + // run_ref) — before the partial unique index below makes a repeat + // impossible, repointing every reference to a dropped duplicate's + // id first. Rows with a null `run_ref` are excluded from the + // dedupe: the window partition below groups nulls together, but + // the partial index treats them as distinct, so collapsing them + // here would delete rows the index was never going to reject. + name: "0026_workbench_threads_delivery_key", + sql: ` + CREATE TEMP TABLE "delivery_thread_dedupe_map" ON COMMIT DROP AS + SELECT "id" AS "drop_id", "keep_id" FROM ( + SELECT + "id", + first_value("id") OVER ( + PARTITION BY "tenant_id", "workbench_id", "run_ref" + ORDER BY "created_at", "id" + ) AS "keep_id" + FROM "chat"."workbench_threads" + WHERE "kind" = 'delivery' AND "run_ref" IS NOT NULL + ) "ranked" + WHERE "id" <> "keep_id"; + + UPDATE "chat"."workbench_thread_messages" "wtm" + SET "thread_id" = "m"."keep_id" + FROM "delivery_thread_dedupe_map" "m" + WHERE "wtm"."thread_id" = "m"."drop_id"; + + UPDATE "chat"."workbench_messages" "wm" + SET "thread_id" = "m"."keep_id" + FROM "delivery_thread_dedupe_map" "m" + WHERE "wm"."thread_id" = "m"."drop_id"; + + UPDATE "chat"."workbench_threads" "wt" + SET "parent_thread_id" = "m"."keep_id" + FROM "delivery_thread_dedupe_map" "m" + WHERE "wt"."parent_thread_id" = "m"."drop_id"; + + DELETE FROM "chat"."workbench_threads" "wt" + USING "delivery_thread_dedupe_map" "m" + WHERE "wt"."id" = "m"."drop_id"; + + CREATE UNIQUE INDEX IF NOT EXISTS "workbench_threads_delivery_key" + ON "chat"."workbench_threads" ("tenant_id", "workbench_id", "run_ref") + WHERE "kind" = 'delivery' AND "run_ref" IS NOT NULL; + `, + }, ]; /** diff --git a/packages/chat/src/schema.ts b/packages/chat/src/schema.ts index 372cb12e1..1db70257f 100644 --- a/packages/chat/src/schema.ts +++ b/packages/chat/src/schema.ts @@ -237,6 +237,13 @@ export const workbenchThreads = chatSchema.table( uniqueIndex("workbench_threads_reply_key") .on(table.tenantId, table.workbenchId, table.parentMessageId) .where(sql`${table.kind} = 'reply'`), + // `run_ref` is nullable (root/reply rows never set it), and a + // partial unique index treats NULLs as distinct rather than equal + // — so the predicate excludes them explicitly rather than relying + // on every delivery-thread caller to always supply one. + uniqueIndex("workbench_threads_delivery_key") + .on(table.tenantId, table.workbenchId, table.runRef) + .where(sql`${table.kind} = 'delivery' AND ${table.runRef} IS NOT NULL`), ], ); diff --git a/packages/chat/src/threads.ts b/packages/chat/src/threads.ts index f865f9c1e..ea63e0286 100644 --- a/packages/chat/src/threads.ts +++ b/packages/chat/src/threads.ts @@ -7,7 +7,7 @@ // mention fan-out (see codec.ts). Message-id reply correlation is a // workbench concern here — do not fork Interchange mail to change that. -import { and, eq, asc } from "drizzle-orm"; +import { and, eq, asc, sql } from "drizzle-orm"; import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; import { workbenchThreadMessages, workbenchThreads } from "./schema"; @@ -358,14 +358,6 @@ export type ThreadDb< TSchema extends Record = Record, > = PostgresJsDatabase; -function requireReturningRow(rows: readonly T[], what: string): T { - const row = rows[0]; - if (row === undefined) { - throw new Error(`expected ${what} row from returning()`); - } - return row; -} - function mapThreadRow( row: typeof workbenchThreads.$inferSelect, ): WorkbenchThread { @@ -556,23 +548,36 @@ export function createDrizzleThreadStore< return reselected; } + async function selectDeliveryThread( + tenantId: string, + workbenchId: string, + runRef: string, + ): Promise { + const rows = await db + .select() + .from(workbenchThreads) + .where( + and( + eq(workbenchThreads.tenantId, tenantId), + eq(workbenchThreads.workbenchId, workbenchId), + eq(workbenchThreads.kind, "delivery"), + eq(workbenchThreads.runRef, runRef), + ), + ) + .limit(1); + return rows[0] ? mapThreadRow(rows[0]) : undefined; + } + return { ensureRootThread, async createDeliveryThread(input) { - const existing = await db - .select() - .from(workbenchThreads) - .where( - and( - eq(workbenchThreads.tenantId, input.tenantId), - eq(workbenchThreads.workbenchId, input.workbenchId), - eq(workbenchThreads.kind, "delivery"), - eq(workbenchThreads.runRef, input.runRef), - ), - ) - .limit(1); - if (existing[0]) return mapThreadRow(existing[0]); + // Insert-first, not select-then-insert: two concurrent first + // deliveries of the same run both attempt the insert, the + // partial unique index (tenant_id, workbench_id, run_ref) WHERE + // kind = 'delivery' AND run_ref IS NOT NULL serializes them, and + // the loser's empty `returning()` re-selects the winner's row + // rather than creating a duplicate delivery thread. const id = newThreadId(); const inserted = await db .insert(workbenchThreads) @@ -586,8 +591,32 @@ export function createDrizzleThreadStore< runRef: input.runRef, title: input.title ?? null, }) + .onConflictDoNothing({ + target: [ + workbenchThreads.tenantId, + workbenchThreads.workbenchId, + workbenchThreads.runRef, + ], + // Matches `workbench_threads_delivery_key`'s predicate in + // ./schema.ts exactly — the ON CONFLICT arbiter predicate + // must match the partial index's own predicate, not just be + // implied by it. + where: sql`${workbenchThreads.kind} = 'delivery' AND ${workbenchThreads.runRef} IS NOT NULL`, + }) .returning(); - return mapThreadRow(requireReturningRow(inserted, "delivery thread")); + const row = inserted[0]; + if (row) return mapThreadRow(row); + const reselected = await selectDeliveryThread( + input.tenantId, + input.workbenchId, + input.runRef, + ); + if (!reselected) { + throw new Error( + "expected delivery thread row after conflicting insert", + ); + } + return reselected; }, openReplyThread: (input) => anchoredReplyThread(input, "reply"), diff --git a/packages/chat/test/migrations.test.ts b/packages/chat/test/migrations.test.ts index 1df9b6426..872c9f77b 100644 --- a/packages/chat/test/migrations.test.ts +++ b/packages/chat/test/migrations.test.ts @@ -46,6 +46,7 @@ const migrationNames = [ "0023_drop_workbench_host_arm", "0024_workbench_launch_sources_digest", "0025_workbench_threads_unique_key", + "0026_workbench_threads_delivery_key", ]; describeIfDb("applyChatMigrations", () => { @@ -173,6 +174,12 @@ describeIfDb("applyChatMigrations", () => { ); expect(threadIndexNames).toContain("workbench_threads_root_key"); expect(threadIndexNames).toContain("workbench_threads_reply_key"); + + // CL-7199: `createDeliveryThread` insert-then-reselects on + // conflict the same way; this partial unique index is what makes + // the conflict possible instead of a silent duplicate delivery + // thread per run. + expect(threadIndexNames).toContain("workbench_threads_delivery_key"); } finally { await sql.end(); } @@ -385,3 +392,182 @@ describeIfDb("0025_workbench_threads_unique_key dedupe", () => { } }, 120000); }); + +describeIfDb("0026_workbench_threads_delivery_key dedupe", () => { + const scratchUrl = scratchUrlFor( + databaseUrl ?? "postgres://localhost:5432/unused", + ).replace("_chat_migrations_test", "_chat_migrations_delivery_dedupe_test"); + const scratchTarget = new URL(scratchUrl); + const scratchDatabase = scratchTarget.pathname.replace(/^\//, ""); + + beforeAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); + await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`); + } finally { + await maintenance.end(); + } + }, 20000); + + afterAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); + } finally { + await maintenance.end(); + } + }, 20000); + + test("collapses duplicate delivery threads per run ref, repoints thread_id and parent_thread_id references, leaves null-run-ref rows untouched, and rejects a repeat insert", async () => { + // Replay every migration through 0025 by hand, seeding the + // duplicates 0026 must clean up before it exists to forbid them — + // then let `applyChatMigrations` run 0026 for real. + const preDeliveryKeyMigrations = chatMigrations.filter( + (migration) => migration.name !== "0026_workbench_threads_delivery_key", + ); + const seed = postgres(scratchUrl, { max: 1, onnotice: () => undefined }); + try { + await seed.unsafe(`CREATE SCHEMA IF NOT EXISTS "chat"`); + await seed.unsafe( + `CREATE TABLE IF NOT EXISTS "chat"."chat_migrations" ` + + `(name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`, + ); + for (const migration of preDeliveryKeyMigrations) { + await seed.begin(async (tx) => { + await tx.unsafe(migration.sql); + await tx.unsafe( + `INSERT INTO "chat"."chat_migrations" (name) VALUES ($1)`, + [migration.name], + ); + }); + } + + await seed.unsafe(` + INSERT INTO "chat"."workbench_threads" + (id, tenant_id, workbench_id, kind, run_ref, created_at) + VALUES + ('thr_delivery_old', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', 'run_1', now() - interval '1 hour'), + ('thr_delivery_new', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', 'run_1', now()) + `); + // Two delivery rows with a null run_ref: the window partition in + // the dedupe groups nulls together (unlike the partial unique + // index it backstops, which treats nulls as distinct), so these + // must survive untouched rather than being collapsed to one row. + await seed.unsafe(` + INSERT INTO "chat"."workbench_threads" + (id, tenant_id, workbench_id, kind, run_ref, created_at) + VALUES + ('thr_delivery_null_a', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', NULL, now() - interval '1 hour'), + ('thr_delivery_null_b', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', NULL, now()) + `); + await seed.unsafe(` + INSERT INTO "chat"."workbench_thread_messages" + (tenant_id, workbench_id, thread_id, message_id) + VALUES ('tnt_ddedupe', 'wb_ddedupe', 'thr_delivery_new', 'msg_ddedupe') + `); + // A message parked in the dropped delivery duplicate — proves + // `workbench_messages.thread_id` is repointed too. + await seed.unsafe(` + INSERT INTO "chat"."workbench_messages" + (id, tenant_id, workbench_id, sender_address, thread_id, parts) + VALUES ( + 'msg_in_dropped_delivery', 'tnt_ddedupe', 'wb_ddedupe', 'addr_1', + 'thr_delivery_new', '[]'::jsonb + ) + `); + // A reply thread anchored directly on the dropped delivery + // duplicate (a reply to a message that lived in it) — proves + // `workbench_threads.parent_thread_id` is repointed, matching + // `containerThreadFor`/`resolveThreadAnchor` letting a delivery + // thread act as a reply's anchor. + await seed.unsafe(` + INSERT INTO "chat"."workbench_threads" + (id, tenant_id, workbench_id, kind, parent_message_id, parent_thread_id, created_at) + VALUES ( + 'thr_child_of_dropped_delivery', 'tnt_ddedupe', 'wb_ddedupe', 'reply', + 'msg_child', 'thr_delivery_new', now() + ) + `); + } finally { + await seed.end(); + } + + const report = await applyChatMigrations(scratchUrl); + expect(report.applied).toEqual(["0026_workbench_threads_delivery_key"]); + + const verify = postgres(scratchUrl, { max: 1, onnotice: () => undefined }); + try { + const deliveries = await verify.unsafe( + `SELECT id FROM "chat"."workbench_threads" ` + + `WHERE tenant_id = 'tnt_ddedupe' AND workbench_id = 'wb_ddedupe' ` + + `AND kind = 'delivery' AND run_ref = 'run_1'`, + ); + expect(deliveries.map((row) => String(row["id"]))).toEqual([ + "thr_delivery_old", + ]); + + const nullRunRefRows = await verify.unsafe( + `SELECT id FROM "chat"."workbench_threads" ` + + `WHERE tenant_id = 'tnt_ddedupe' AND workbench_id = 'wb_ddedupe' ` + + `AND kind = 'delivery' AND run_ref IS NULL`, + ); + expect(nullRunRefRows.map((row) => String(row["id"])).sort()).toEqual([ + "thr_delivery_null_a", + "thr_delivery_null_b", + ]); + + const membership = await verify.unsafe( + `SELECT thread_id FROM "chat"."workbench_thread_messages" WHERE message_id = 'msg_ddedupe'`, + ); + expect(String(membership[0]?.["thread_id"])).toBe("thr_delivery_old"); + + const messageThreadId = await verify.unsafe( + `SELECT thread_id FROM "chat"."workbench_messages" WHERE id = 'msg_in_dropped_delivery'`, + ); + expect(String(messageThreadId[0]?.["thread_id"])).toBe( + "thr_delivery_old", + ); + + const childParentThreadId = await verify.unsafe( + `SELECT parent_thread_id FROM "chat"."workbench_threads" WHERE id = 'thr_child_of_dropped_delivery'`, + ); + expect(String(childParentThreadId[0]?.["parent_thread_id"])).toBe( + "thr_delivery_old", + ); + + // The delivery unique index holds: a second delivery row for the + // same (tenant, workbench, run_ref) key is now rejected rather + // than silently accepted as a second duplicate. + await expect( + (async () => { + await verify.unsafe( + `INSERT INTO "chat"."workbench_threads" ` + + `(id, tenant_id, workbench_id, kind, run_ref) ` + + `VALUES ('thr_delivery_conflict', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', 'run_1')`, + ); + })(), + ).rejects.toThrow(); + + // A null run_ref never conflicts, matching the partial index's + // predicate. + await verify.unsafe( + `INSERT INTO "chat"."workbench_threads" ` + + `(id, tenant_id, workbench_id, kind, run_ref) ` + + `VALUES ('thr_delivery_null_c', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', NULL)`, + ); + } finally { + await verify.end(); + } + }, 120000); +}); diff --git a/packages/chat/test/threads.drizzle.test.ts b/packages/chat/test/threads.drizzle.test.ts index 2a3ed34cb..03852fcf9 100644 --- a/packages/chat/test/threads.drizzle.test.ts +++ b/packages/chat/test/threads.drizzle.test.ts @@ -2,15 +2,16 @@ // checkout still runs the unit gates), mirroring // `reactions.drizzle.test.ts`. Runs against its own scratch database. // -// `threads.test.ts` proves `ensureRootThread`/`openReplyThread`'s -// idempotency against the in-memory store, which can never actually -// race (no `await` between its read and write). This exercises the -// real `createDrizzleThreadStore` path, where two concurrent first -// writers for the same root or reply key really do race at the -// database: proves the fix (insert with `onConflictDoNothing` backed -// by the partial unique index, then re-select on conflict — never -// select-then-insert) never throws a raw unique-violation and both -// callers converge on the same thread row (CL-7130). +// `threads.test.ts` proves `ensureRootThread`/`openReplyThread`/ +// `createDeliveryThread`'s idempotency against the in-memory store, +// which can never actually race (no `await` between its read and +// write). This exercises the real `createDrizzleThreadStore` path, +// where two concurrent first writers for the same root, reply, or +// delivery key really do race at the database: proves the fix (insert +// with `onConflictDoNothing` backed by the partial unique index, then +// re-select on conflict — never select-then-insert) never throws a raw +// unique-violation and both callers converge on the same thread row +// (CL-7130, CL-7199). import { afterAll, beforeAll, describe, expect, test } from "bun:test"; import { drizzle } from "drizzle-orm/postgres-js"; import postgres from "postgres"; @@ -122,4 +123,32 @@ describeIfDb("createDrizzleThreadStore: concurrent first writers", () => { await sql.end(); } }); + + test("two concurrent createDeliveryThread calls for the same run ref never throw a unique-violation and agree on one row", async () => { + const sql = postgres(scratchUrl, { max: 5, onnotice: () => undefined }); + try { + const store = createDrizzleThreadStore(drizzle(sql)); + const input = { + tenantId: TENANT, + workbenchId: WORKBENCH, + runRef: "run_race", + }; + + const [first, second] = await Promise.all([ + store.createDeliveryThread(input), + store.createDeliveryThread(input), + ]); + + expect(first.id).toBe(second.id); + + const rows = await sql.unsafe( + `SELECT id FROM "chat"."workbench_threads" ` + + `WHERE tenant_id = $1 AND workbench_id = $2 AND kind = 'delivery' AND run_ref = $3`, + [TENANT, WORKBENCH, "run_race"], + ); + expect(rows).toHaveLength(1); + } finally { + await sql.end(); + } + }); });