From e1fb588fb2e26e2936ecfcfa77477f0e586201d5 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 10:32:39 -0700 Subject: [PATCH 1/2] Add tests for delivery thread race createDeliveryThread is select-then-insert on (kind='delivery', runRef) with no unique index backing the read, so a redelivered routine event can create two delivery threads for one run. Assert the fix CL-7130 already established for root/reply threads: a concurrency test proving two racing createDeliveryThread calls converge on one row, and a migration test proving existing duplicate delivery threads get deduped (with references repointed) before the unique index is added. These currently fail: no unique index or insert-first path exists yet for delivery threads. --- packages/chat/test/migrations.test.ts | 186 +++++++++++++++++++++ packages/chat/test/threads.drizzle.test.ts | 47 +++++- 2 files changed, 224 insertions(+), 9 deletions(-) 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(); + } + }); }); From 3d09599312e04d621674fc530a5829d4db676976 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 10:33:13 -0700 Subject: [PATCH 2/2] Delivery threads: dedupe and enforce a unique key createDeliveryThread was select-then-insert on (kind='delivery', runRef) with no unique constraint backing the read, unlike its ensureRootThread/anchoredReplyThread siblings fixed for the same class of bug in CL-7130. A redelivered routine event could insert a second delivery thread for the same run, splitting replies across both. Migration 0026 dedupes any existing duplicate delivery threads (keeping the oldest row, repointing thread membership and message thread pointers off the dropped rows) and adds a partial unique index on workbench_threads scoped to kind='delivery' with a non-null run_ref. createDeliveryThread now inserts with onConflictDoNothing and re-selects on conflict, matching its siblings' pattern exactly. Fixes CL-7199. --- packages/chat/src/migrations.ts | 51 ++++++++++++++++++++++ packages/chat/src/schema.ts | 7 +++ packages/chat/src/threads.ts | 75 +++++++++++++++++++++++---------- 3 files changed, 110 insertions(+), 23 deletions(-) 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"),