From 0d147803a96888943cc557a6e2a042a43f6d3dd9 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 14:07:58 -0700 Subject: [PATCH 1/2] Add tests for concurrent workbench settings patches A PATCH that only touches chat/pinned must keep a concurrent chat/participants write. Cover the in-memory store, the HTTP route, and the drizzle row lock. --- .../chat/test/settings-patch.drizzle.test.ts | 129 ++++++++++++++++++ packages/chat/test/store.test.ts | 67 +++++++++ packages/chat/test/workbench-settings.test.ts | 50 +++++++ 3 files changed, 246 insertions(+) create mode 100644 packages/chat/test/settings-patch.drizzle.test.ts diff --git a/packages/chat/test/settings-patch.drizzle.test.ts b/packages/chat/test/settings-patch.drizzle.test.ts new file mode 100644 index 000000000..13dcac915 --- /dev/null +++ b/packages/chat/test/settings-patch.drizzle.test.ts @@ -0,0 +1,129 @@ +// DB-gated: skipped when no DATABASE_URL is reachable (a fresh +// checkout still runs the unit gates), mirroring `read-state.drizzle.test.ts`. +// Two concurrent `patchWorkbenchSettings` calls on one row must both land: +// a participants write and a `chat/pinned` write cannot clobber each other. +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import { drizzle } from "drizzle-orm/postgres-js"; +import postgres from "postgres"; + +import { e2eDatabaseUrl } from "../../../scripts/e2e/harness"; +import { applyChatMigrations } from "../src/migrations"; +import { createDrizzleChatStore } from "../src/store"; + +function scratchUrlFor(e2eUrl: string): string { + const url = new URL(e2eUrl); + const database = url.pathname.replace(/^\//, ""); + url.pathname = `/${database}_chat_settings_patch_drizzle_test`; + return url.toString(); +} + +const databaseUrl = e2eDatabaseUrl(); +const describeIfDb = databaseUrl === undefined ? describe.skip : describe; + +const TENANT = "tnt_1"; +const WORKBENCH = "run_workbench1"; +const ALICE = { address: "prn_alice@acme.example", handle: "alice" }; +const BOB = { address: "prn_bob@acme.example", handle: "bob" }; + +describeIfDb("createDrizzleChatStore: patchWorkbenchSettings", () => { + const scratchUrl = scratchUrlFor( + databaseUrl ?? "postgres://localhost:5432/unused", + ); + 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(); + } + await applyChatMigrations(scratchUrl); + }); + + 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(); + } + }); + + test("concurrent patches of participants and another key both land", async () => { + const sql = postgres(scratchUrl, { max: 5, onnotice: () => undefined }); + try { + const store = createDrizzleChatStore(drizzle(sql)); + await store.createWorkbenchSettings({ + tenantId: TENANT, + workbenchId: WORKBENCH, + settings: { + "chat/pinned": false, + "chat/participants": [ALICE], + }, + updatedBy: "prn_1", + }); + + await Promise.all([ + store.patchWorkbenchSettings({ + tenantId: TENANT, + workbenchId: WORKBENCH, + patch: { "chat/pinned": true }, + updatedBy: "prn_2", + }), + store.patchWorkbenchSettings({ + tenantId: TENANT, + workbenchId: WORKBENCH, + patch: { "chat/participants": [ALICE, BOB] }, + updatedBy: "prn_3", + }), + ]); + + const row = await store.getWorkbenchSettings(TENANT, WORKBENCH); + expect(row?.settings["chat/pinned"]).toBe(true); + expect(row?.settings["chat/participants"]).toEqual([ALICE, BOB]); + } finally { + await sql.end(); + } + }); + + test("a patch that omits chat/participants keeps the locked row's list", async () => { + const sql = postgres(scratchUrl, { max: 5, onnotice: () => undefined }); + try { + const store = createDrizzleChatStore(drizzle(sql)); + const workbenchId = "run_workbench2"; + await store.createWorkbenchSettings({ + tenantId: TENANT, + workbenchId, + settings: { + "chat/pinned": false, + "chat/participants": [ALICE, BOB], + }, + updatedBy: "prn_1", + }); + + const updated = await store.patchWorkbenchSettings({ + tenantId: TENANT, + workbenchId, + patch: { "chat/pinned": true }, + updatedBy: "prn_2", + }); + expect(updated.settings["chat/pinned"]).toBe(true); + expect(updated.settings["chat/participants"]).toEqual([ALICE, BOB]); + } finally { + await sql.end(); + } + }); +}); diff --git a/packages/chat/test/store.test.ts b/packages/chat/test/store.test.ts index fd69a33a3..48516760b 100644 --- a/packages/chat/test/store.test.ts +++ b/packages/chat/test/store.test.ts @@ -140,6 +140,73 @@ test("mutateWorkbenchParticipants rejects a missing workbench", async () => { ).rejects.toThrow(); }); +test("patchWorkbenchSettings merges only provided keys and keeps omitted ones", async () => { + const store = createInMemoryChatStore(); + const alice = { address: "prn_alice@acme.example", handle: "alice" }; + await store.createWorkbenchSettings({ + tenantId: "tnt_1", + workbenchId: "chn_1", + settings: { + "chat/pinned": false, + "chat/participants": [alice], + }, + updatedBy: "prn_1", + }); + + const updated = await store.patchWorkbenchSettings({ + tenantId: "tnt_1", + workbenchId: "chn_1", + patch: { "chat/pinned": true }, + updatedBy: "prn_2", + }); + expect(updated.settings["chat/pinned"]).toBe(true); + expect(updated.settings["chat/participants"]).toEqual([alice]); + expect(updated.updatedBy).toBe("prn_2"); + + await expect( + store.patchWorkbenchSettings({ + tenantId: "tnt_1", + workbenchId: "chn_missing", + patch: { "chat/pinned": true }, + updatedBy: "prn_1", + }), + ).rejects.toThrow(); +}); + +test("concurrent patchWorkbenchSettings of participants and another key both land", async () => { + const store = createInMemoryChatStore(); + const alice = { address: "prn_alice@acme.example", handle: "alice" }; + const bob = { address: "prn_bob@acme.example", handle: "bob" }; + await store.createWorkbenchSettings({ + tenantId: "tnt_1", + workbenchId: "chn_1", + settings: { + "chat/pinned": false, + "chat/participants": [alice], + }, + updatedBy: "prn_1", + }); + + await Promise.all([ + store.patchWorkbenchSettings({ + tenantId: "tnt_1", + workbenchId: "chn_1", + patch: { "chat/pinned": true }, + updatedBy: "prn_2", + }), + store.patchWorkbenchSettings({ + tenantId: "tnt_1", + workbenchId: "chn_1", + patch: { "chat/participants": [alice, bob] }, + updatedBy: "prn_3", + }), + ]); + + const row = await store.getWorkbenchSettings("tnt_1", "chn_1"); + expect(row?.settings["chat/pinned"]).toBe(true); + expect(row?.settings["chat/participants"]).toEqual([alice, bob]); +}); + test("getBenchSettings is undefined until a bench sets defaults, then upsertBenchSettings replaces them", async () => { const store = createInMemoryChatStore(); expect(await store.getBenchSettings("tnt_1")).toBeUndefined(); diff --git a/packages/chat/test/workbench-settings.test.ts b/packages/chat/test/workbench-settings.test.ts index 2f64a29a7..0d065f900 100644 --- a/packages/chat/test/workbench-settings.test.ts +++ b/packages/chat/test/workbench-settings.test.ts @@ -487,6 +487,56 @@ describe("PATCH /workbenches/:id/settings", () => { expect(clearedBody.contextWindow).toEqual({ value: 20, source: "inherit" }); expect(clearedBody.settings["chat/contextWindow"]).toBeNull(); }); + + test("concurrent PATCH of chat/participants and chat/pinned both land", async () => { + const deps = buildDeps(); + const app = mountAs(createChatRoutes(deps), "prn_alice"); + const { body: workbench } = await createWorkbench(app, { + kind: "workbench", + }); + + const before = await deps.store.getWorkbenchSettings( + TENANT.id, + workbench.id, + ); + const existingParticipants = before?.settings["chat/participants"]; + const extra = { + address: "ins_extra@acme.example", + handle: "extra", + }; + + const [pinnedResponse, participantsResponse] = await Promise.all([ + app.request(`/workbenches/${workbench.id}/settings`, { + method: "PATCH", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ "chat/pinned": true }), + }), + app.request(`/workbenches/${workbench.id}/settings`, { + method: "PATCH", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + "chat/participants": [ + ...(Array.isArray(existingParticipants) + ? existingParticipants + : []), + extra, + ], + }), + }), + ]); + + expect(pinnedResponse.status).toBe(200); + expect(participantsResponse.status).toBe(200); + + const stored = await deps.store.getWorkbenchSettings( + TENANT.id, + workbench.id, + ); + expect(stored?.settings["chat/pinned"]).toBe(true); + const participants = stored?.settings["chat/participants"]; + expect(Array.isArray(participants)).toBe(true); + expect(participants).toContainEqual(extra); + }); }); describe("GET/PATCH /bench/settings", () => { From 7a61fe19ea95efd6241ec2ebae95668edc20d721 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 14:08:05 -0700 Subject: [PATCH 2/2] Merge workbench settings patches under a row lock The PATCH handler no longer whole-blob-writes a pre-request snapshot. patchWorkbenchSettings SELECT ... FOR UPDATE the workbench_settings row and merges only the keys the caller sent, so a pinned-flag write cannot revert a concurrent invite. --- packages/chat/src/routes.ts | 18 ++---- packages/chat/src/store.ts | 124 ++++++++++++++++++++++++++++++++++-- 2 files changed, 127 insertions(+), 15 deletions(-) diff --git a/packages/chat/src/routes.ts b/packages/chat/src/routes.ts index 8639d02a5..4d3f348d3 100644 --- a/packages/chat/src/routes.ts +++ b/packages/chat/src/routes.ts @@ -3407,23 +3407,19 @@ export function createChatRoutes(deps: CreateChatRoutesDeps): Hono { // `chat/participants` is normalized to records on write even when // a caller PATCHes it with bare addresses (as the settings-control // wire path does) — settings always hold records, never strings. - const merged: Record = { - ...existing.settings, - ...patch, - }; + // Merge happens under a row lock against the live snapshot so a + // PATCH that omits `chat/participants` cannot revert a concurrent + // invite. + const writePatch: Record = { ...patch }; if (patch["chat/participants"] !== undefined) { - merged["chat/participants"] = parseParticipants( + writePatch["chat/participants"] = parseParticipants( patch["chat/participants"], ); } - // The settings record itself is the durable source of truth; it - // is updated before anything else here fires, so a failure - // below never leaves the record unwritten and the audit trail - // silently ahead of it. - const row = await deps.store.updateWorkbenchSettings({ + const row = await deps.store.patchWorkbenchSettings({ tenantId: tenant.id, workbenchId, - settings: merged, + patch: writePatch, updatedBy: principal.id, }); diff --git a/packages/chat/src/store.ts b/packages/chat/src/store.ts index 7dbe4ebb4..1d8a6322d 100644 --- a/packages/chat/src/store.ts +++ b/packages/chat/src/store.ts @@ -1,9 +1,12 @@ // Persistence for the two chat product tables, kept apart from route // wiring so the HTTP layer never touches drizzle directly. `settings` -// is record-as-truth: callers read and write the whole namespaced -// jsonb blob, and this module never interprets any `chat/*` key — -// that parsing lives in `routes.ts`, next to the request boundary it -// guards. +// is record-as-truth: callers read and write namespaced jsonb keys, and +// this module never interprets any `chat/*` key — that parsing lives in +// `routes.ts`, next to the request boundary it guards. Whole-blob replace +// (`updateWorkbenchSettings`) remains for callers that already hold the +// full record; `patchWorkbenchSettings` locks the row and merges only the +// keys the caller sent so concurrent PATCHes of different keys cannot +// clobber each other. // // `ChatStore` is the seam `routes.ts` actually depends on; `createDrizzleChatStore` // is its one production implementation, over the two tables in `./schema.ts`. @@ -73,6 +76,13 @@ export interface MutateWorkbenchParticipantsInput { ) => ParticipantRecord[]; } +export interface PatchWorkbenchSettingsInput { + readonly tenantId: string; + readonly workbenchId: string; + readonly patch: Record; + readonly updatedBy: string; +} + export interface ChatBenchSettingsRow { readonly tenantId: string; readonly settings: Record; @@ -146,6 +156,15 @@ export interface ChatStore { mutateWorkbenchParticipants( input: MutateWorkbenchParticipantsInput, ): Promise; + /** + * Merges `patch` onto the workbench's settings under a row lock + * (`SELECT ... FOR UPDATE` in Postgres). Only keys present in `patch` + * are written; omitted keys keep the locked snapshot's values, so a + * concurrent PATCH of a different key cannot be reverted. + */ + patchWorkbenchSettings( + input: PatchWorkbenchSettingsInput, + ): Promise; getBenchSettings(tenantId: string): Promise; upsertBenchSettings( input: UpsertBenchSettingsInput, @@ -200,6 +219,18 @@ export interface ChatStore { ): Promise; } +/** Top-level JSONB merge: only keys present in `patch` overwrite. */ +function mergeSettingsPatch( + existing: Record, + patch: Record, +): Record { + const merged: Record = { ...existing }; + for (const key of Object.keys(patch)) { + merged[key] = patch[key]; + } + return merged; +} + /** * The production `ChatStore`, backed by the `workbench_settings` and * `workbench_read_state` tables declared in `./schema.ts`. @@ -338,6 +369,49 @@ export function createDrizzleChatStore>( }); }, + async patchWorkbenchSettings(input) { + return db.transaction(async (tx) => { + const [selected] = await tx + .select() + .from(workbenchSettings) + .where( + and( + eq(workbenchSettings.tenantId, input.tenantId), + eq(workbenchSettings.workbenchId, input.workbenchId), + ), + ) + .for("update") + .limit(1); + if (selected === undefined) { + throw new Error( + `patchWorkbenchSettings: no workbench_settings row for workbench ${input.workbenchId}`, + ); + } + const existing = selected as WorkbenchSettingsRow; + const merged = mergeSettingsPatch(existing.settings, input.patch); + const [row] = await tx + .update(workbenchSettings) + .set({ + settings: merged, + updatedBy: input.updatedBy, + updatedAt: new Date(), + }) + .where( + and( + eq(workbenchSettings.tenantId, input.tenantId), + eq(workbenchSettings.workbenchId, input.workbenchId), + ), + ) + .returning(); + if (row === undefined) { + throw new Error( + `patchWorkbenchSettings: no workbench_settings row for workbench ${input.workbenchId}`, + ); + } + return row as WorkbenchSettingsRow; + }); + }, + async getBenchSettings(tenantId) { const [selected] = await db .select() @@ -467,6 +541,7 @@ export function createInMemoryChatStore(): ChatStore { const readStateByKey = new Map(); const benchSettingsByTenant = new Map(); const launchedByKey = new Set(); + const rowLocks = new Map>(); const settingsKey = (tenantId: string, workbenchId: string) => `${tenantId}:${workbenchId}`; @@ -476,6 +551,27 @@ export function createInMemoryChatStore(): ChatStore { principalId: string, ) => `${tenantId}:${workbenchId}:${principalId}`; + const withRowLock = async ( + key: string, + fn: () => Promise, + ): Promise => { + const previous = rowLocks.get(key) ?? Promise.resolve(); + let release: () => void = () => undefined; + const held = new Promise((resolve) => { + release = resolve; + }); + rowLocks.set( + key, + previous.then(() => held), + ); + await previous; + try { + return await fn(); + } finally { + release(); + } + }; + return { async createWorkbenchSettings(input) { const row: WorkbenchSettingsRow = { @@ -548,6 +644,26 @@ export function createInMemoryChatStore(): ChatStore { return row; }, + async patchWorkbenchSettings(input) { + const key = settingsKey(input.tenantId, input.workbenchId); + return withRowLock(key, async () => { + const existing = settingsByKey.get(key); + if (existing === undefined) { + throw new Error( + `patchWorkbenchSettings: no workbench_settings row for workbench ${input.workbenchId}`, + ); + } + const row: WorkbenchSettingsRow = { + ...existing, + settings: mergeSettingsPatch(existing.settings, input.patch), + updatedBy: input.updatedBy, + updatedAt: new Date(), + }; + settingsByKey.set(key, row); + return row; + }); + }, + async getBenchSettings(tenantId) { return benchSettingsByTenant.get(tenantId); },