From c6b101a40066fb8cb37342382679b4d56f39ddca Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 26 Jul 2026 11:47:05 -0700 Subject: [PATCH 1/3] fix(tables): write single-cell edits as an atomic JSONB patch (fix row-level LWW) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A table row stores every cell in one jsonb `data` column, and the row-edit paths called updateRow in the default replace mode — a full-object read-modify-write. Two users (or a user and an agent/copilot) editing DIFFERENT cells of the same row concurrently would clobber each other: whichever write committed last overwrote the whole row from its stale snapshot. Pass dataWriteMode: 'patch' (Postgres `data = data || patch::jsonb`) from the four partial-update callers — the UI row route, the v1 API row route, the copilot table tool, and the workflow-group cancel-write — so each cell edit merges atomically under the row lock. upsertRow (whole-row by design) and updateRowsByFilter (already uses ||) are unchanged. computedWrite stays unset on the user paths, preserving the update lock. Adds a route test asserting the UI route opts into patch mode; the updateRow patch mechanism itself is already covered. --- .../[tableId]/rows/[rowId]/route.test.ts | 82 +++++++++++++++++++ .../api/table/[tableId]/rows/[rowId]/route.ts | 9 +- .../v1/tables/[tableId]/rows/[rowId]/route.ts | 7 +- .../copilot/tools/server/table/user-table.ts | 6 +- apps/sim/lib/table/workflow-columns.ts | 6 +- 5 files changed, 106 insertions(+), 4 deletions(-) create mode 100644 apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts diff --git a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts new file mode 100644 index 00000000000..3fdf297e0c2 --- /dev/null +++ b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts @@ -0,0 +1,82 @@ +/** + * @vitest-environment node + */ +import { NextRequest } from 'next/server' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockAuth, mockCheckAccess, mockUpdateRow } = vi.hoisted(() => ({ + mockAuth: vi.fn(), + mockCheckAccess: vi.fn(), + mockUpdateRow: vi.fn(), +})) + +vi.mock('@/lib/auth/hybrid', () => ({ + checkSessionOrInternalAuth: mockAuth, +})) + +vi.mock('@/lib/table', () => ({ + updateRow: mockUpdateRow, + deleteRow: vi.fn(), +})) + +vi.mock('@/app/api/table/row-wire', () => ({ + rowWireTranslators: () => ({ + dataIn: (data: unknown) => data, + dataOut: (data: unknown) => data, + }), +})) + +vi.mock('@/app/api/table/utils', async () => { + const { NextResponse } = await import('next/server') + return { + checkAccess: mockCheckAccess, + accessError: (result: { status: number }) => + NextResponse.json({ error: 'denied' }, { status: result.status }), + rowWriteErrorResponse: () => undefined, + tableLockErrorResponse: () => undefined, + rootErrorMessage: (error: unknown) => (error instanceof Error ? error.message : ''), + } +}) + +import { PATCH } from '@/app/api/table/[tableId]/rows/[rowId]/route' + +const TABLE = { + id: 'tbl_1', + name: 'People', + workspaceId: 'ws-1', + schema: { columns: [{ id: 'col_1', name: 'name', type: 'text' }] }, +} + +function patchRequest(body: unknown) { + return new NextRequest('http://localhost:3000/api/table/tbl_1/rows/row_1', { + method: 'PATCH', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body), + }) +} + +describe('PATCH /api/table/[tableId]/rows/[rowId]', () => { + beforeEach(() => { + vi.clearAllMocks() + mockAuth.mockResolvedValue({ success: true, userId: 'user-1', authType: 'session' }) + mockCheckAccess.mockResolvedValue({ ok: true, table: TABLE }) + mockUpdateRow.mockResolvedValue({ + id: 'row_1', + data: { col_1: 'Bob' }, + position: 0, + createdAt: new Date('2026-01-01T00:00:00.000Z'), + updatedAt: new Date('2026-01-01T00:00:01.000Z'), + }) + }) + + it("writes only the edited cells (dataWriteMode: 'patch') so a concurrent edit to another cell of the same row is not clobbered", async () => { + const res = await PATCH(patchRequest({ workspaceId: 'ws-1', data: { col_1: 'Bob' } }), { + params: Promise.resolve({ tableId: 'tbl_1', rowId: 'row_1' }), + }) + + expect(res.status).toBe(200) + expect(mockUpdateRow).toHaveBeenCalledTimes(1) + // The 4th argument is the options object; it must opt into cell-atomic JSONB patch mode. + expect(mockUpdateRow.mock.calls[0][3]).toEqual({ dataWriteMode: 'patch' }) + }) +}) diff --git a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts index 82ec048ca84..a6aa02ddc43 100644 --- a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts +++ b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts @@ -133,6 +133,12 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR } const wire = rowWireTranslators(authResult.authType, table.schema as TableSchema) + // Write only the edited cells via a Postgres JSONB concat (`data = data || patch`) instead of + // replacing the whole `data` object. A table row stores every cell in one jsonb column, so a + // full-object replace is last-write-wins across the ROW: two users editing different cells of + // the same row concurrently would clobber each other. `patch` merges under the row lock, so + // each cell edit is atomic. `computedWrite` stays unset — this is a user edit, still subject to + // the normal update lock (unlike the workflow engine writing its own output cells). const updatedRow = await updateRow( { tableId, @@ -142,7 +148,8 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR actorUserId: authResult.userId, }, table, - requestId + requestId, + { dataWriteMode: 'patch' } ) // Only `null` when a `cancellationGuard` is supplied and the SQL guard // rejects the write — this route doesn't pass one, so reaching null is a bug. diff --git a/apps/sim/app/api/v1/tables/[tableId]/rows/[rowId]/route.ts b/apps/sim/app/api/v1/tables/[tableId]/rows/[rowId]/route.ts index 2a7ea2fe7a5..1f94852b107 100644 --- a/apps/sim/app/api/v1/tables/[tableId]/rows/[rowId]/route.ts +++ b/apps/sim/app/api/v1/tables/[tableId]/rows/[rowId]/route.ts @@ -139,6 +139,10 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR const idByName = buildIdByName(table.schema as TableSchema) const toNamedRow = namedRowMapper((table.schema as TableSchema).columns) + // Patch only the edited cells (`data = data || patch`) so concurrent edits to different cells + // of the same row don't clobber each other — a row stores all its cells in one jsonb column, + // so a full-object replace is last-write-wins across the whole row. See the app route for the + // full rationale. const updatedRow = await updateRow( { tableId, @@ -148,7 +152,8 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR actorUserId, }, table, - requestId + requestId, + { dataWriteMode: 'patch' } ) // No `cancellationGuard` is passed here, so `updateRow` can't return null // from this caller. Defensive narrowing for TypeScript. diff --git a/apps/sim/lib/copilot/tools/server/table/user-table.ts b/apps/sim/lib/copilot/tools/server/table/user-table.ts index a88e1dee903..2706d1e0572 100644 --- a/apps/sim/lib/copilot/tools/server/table/user-table.ts +++ b/apps/sim/lib/copilot/tools/server/table/user-table.ts @@ -727,6 +727,9 @@ export const userTableServerTool: BaseServerTool assertNotAborted() const idByName = buildIdByName(table.schema) const toNamedRow = namedRowMapper(table.schema.columns) + // Patch only the edited cells so a copilot edit doesn't clobber a concurrent user edit + // to a different cell of the same row (all cells share one jsonb column — a full-object + // replace is last-write-wins across the row). See the table row route for the rationale. const updatedRow = await updateRow( { tableId: args.tableId, @@ -736,7 +739,8 @@ export const userTableServerTool: BaseServerTool actorUserId: context.userId, }, table, - requestId + requestId, + { dataWriteMode: 'patch' } ) if (!updatedRow) { // Only the cell-task path passes a `cancellationGuard`; this caller diff --git a/apps/sim/lib/table/workflow-columns.ts b/apps/sim/lib/table/workflow-columns.ts index 9e0f35fc172..c0533306a7f 100644 --- a/apps/sim/lib/table/workflow-columns.ts +++ b/apps/sim/lib/table/workflow-columns.ts @@ -627,7 +627,11 @@ export async function cancelWorkflowGroupRuns( executionsPatch: m.executionsPatch, }, table, - `wfgrp-cancel-${m.rowId}` + `wfgrp-cancel-${m.rowId}`, + // This write only touches execution state (data is `{}`). `patch` makes the data write a + // true no-op (`data || '{}'`), whereas the default replace would rewrite the whole jsonb + // column and could revert a user's concurrent edit to a cell of the same row. + { dataWriteMode: 'patch' } ).catch((err) => { logger.error(`Failed to write cancelled state for row ${m.rowId}:`, err) }) From 7aff1ee1d1e05b779aa35ffafffa727ff89e78f9 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sun, 26 Jul 2026 11:52:35 -0700 Subject: [PATCH 2/3] test(tables): drop banned inline error-message pattern in route test mock --- apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts index 3fdf297e0c2..35d16175971 100644 --- a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts +++ b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts @@ -34,7 +34,7 @@ vi.mock('@/app/api/table/utils', async () => { NextResponse.json({ error: 'denied' }, { status: result.status }), rowWriteErrorResponse: () => undefined, tableLockErrorResponse: () => undefined, - rootErrorMessage: (error: unknown) => (error instanceof Error ? error.message : ''), + rootErrorMessage: () => '', } }) From e7c903bb277f92fcf60c6582966ac5b00d0aa435 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 27 Jul 2026 09:01:14 -0700 Subject: [PATCH 3/3] refactor(tables): make the atomic cell-merge the only row-write path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit of the opt-in-per-caller approach found it incomplete and the wrong shape: - `replace` mode never actually replaces — updateRow computes mergedData = {...existing, ...patch}, which only ever adds/overwrites keys, never deletes. So replace is content-identical to patch and differs ONLY by being racy (full-object read-modify-write = row-level last-write-wins). No caller benefits from it. - Two structurally identical `data:{}` exec-clears in background/workflow-column- execution.ts were left in replace mode, exposed to the same clobber the PR fixes. - batchUpdateRows (the multi-cell grid/copilot path) still full-replaced with no patch option at all, so the PR's own goal was only half-met. Remove the dataWriteMode option entirely and make updateRow ALWAYS write the changed cells via a shared jsonbMergePatch helper (data = data || {changed}::jsonb). Apply the same helper per-row in batchUpdateRows. This auto-covers every updateRow caller — including the two exec-clears — and closes the grid path, with the user-facing route and copilot callers now unchanged vs staging (the fix lives entirely in the service layer). computedWrite (the workflow-engine lock exemption) is untouched. Tests updated to assert the always-on JSONB merge for both updateRow and batchUpdateRows. --- .../[tableId]/rows/[rowId]/route.test.ts | 82 ---------------- .../api/table/[tableId]/rows/[rowId]/route.ts | 9 +- .../v1/tables/[tableId]/rows/[rowId]/route.ts | 7 +- .../copilot/tools/server/table/user-table.ts | 6 +- .../lib/table/__tests__/update-row.test.ts | 95 +++++++++++++------ apps/sim/lib/table/cell-write.test.ts | 2 +- apps/sim/lib/table/cell-write.ts | 2 +- apps/sim/lib/table/rows/service.ts | 40 ++++---- apps/sim/lib/table/workflow-columns.ts | 8 +- 9 files changed, 100 insertions(+), 151 deletions(-) delete mode 100644 apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts diff --git a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts deleted file mode 100644 index 35d16175971..00000000000 --- a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.test.ts +++ /dev/null @@ -1,82 +0,0 @@ -/** - * @vitest-environment node - */ -import { NextRequest } from 'next/server' -import { beforeEach, describe, expect, it, vi } from 'vitest' - -const { mockAuth, mockCheckAccess, mockUpdateRow } = vi.hoisted(() => ({ - mockAuth: vi.fn(), - mockCheckAccess: vi.fn(), - mockUpdateRow: vi.fn(), -})) - -vi.mock('@/lib/auth/hybrid', () => ({ - checkSessionOrInternalAuth: mockAuth, -})) - -vi.mock('@/lib/table', () => ({ - updateRow: mockUpdateRow, - deleteRow: vi.fn(), -})) - -vi.mock('@/app/api/table/row-wire', () => ({ - rowWireTranslators: () => ({ - dataIn: (data: unknown) => data, - dataOut: (data: unknown) => data, - }), -})) - -vi.mock('@/app/api/table/utils', async () => { - const { NextResponse } = await import('next/server') - return { - checkAccess: mockCheckAccess, - accessError: (result: { status: number }) => - NextResponse.json({ error: 'denied' }, { status: result.status }), - rowWriteErrorResponse: () => undefined, - tableLockErrorResponse: () => undefined, - rootErrorMessage: () => '', - } -}) - -import { PATCH } from '@/app/api/table/[tableId]/rows/[rowId]/route' - -const TABLE = { - id: 'tbl_1', - name: 'People', - workspaceId: 'ws-1', - schema: { columns: [{ id: 'col_1', name: 'name', type: 'text' }] }, -} - -function patchRequest(body: unknown) { - return new NextRequest('http://localhost:3000/api/table/tbl_1/rows/row_1', { - method: 'PATCH', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify(body), - }) -} - -describe('PATCH /api/table/[tableId]/rows/[rowId]', () => { - beforeEach(() => { - vi.clearAllMocks() - mockAuth.mockResolvedValue({ success: true, userId: 'user-1', authType: 'session' }) - mockCheckAccess.mockResolvedValue({ ok: true, table: TABLE }) - mockUpdateRow.mockResolvedValue({ - id: 'row_1', - data: { col_1: 'Bob' }, - position: 0, - createdAt: new Date('2026-01-01T00:00:00.000Z'), - updatedAt: new Date('2026-01-01T00:00:01.000Z'), - }) - }) - - it("writes only the edited cells (dataWriteMode: 'patch') so a concurrent edit to another cell of the same row is not clobbered", async () => { - const res = await PATCH(patchRequest({ workspaceId: 'ws-1', data: { col_1: 'Bob' } }), { - params: Promise.resolve({ tableId: 'tbl_1', rowId: 'row_1' }), - }) - - expect(res.status).toBe(200) - expect(mockUpdateRow).toHaveBeenCalledTimes(1) - // The 4th argument is the options object; it must opt into cell-atomic JSONB patch mode. - expect(mockUpdateRow.mock.calls[0][3]).toEqual({ dataWriteMode: 'patch' }) - }) -}) diff --git a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts index a6aa02ddc43..82ec048ca84 100644 --- a/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts +++ b/apps/sim/app/api/table/[tableId]/rows/[rowId]/route.ts @@ -133,12 +133,6 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR } const wire = rowWireTranslators(authResult.authType, table.schema as TableSchema) - // Write only the edited cells via a Postgres JSONB concat (`data = data || patch`) instead of - // replacing the whole `data` object. A table row stores every cell in one jsonb column, so a - // full-object replace is last-write-wins across the ROW: two users editing different cells of - // the same row concurrently would clobber each other. `patch` merges under the row lock, so - // each cell edit is atomic. `computedWrite` stays unset — this is a user edit, still subject to - // the normal update lock (unlike the workflow engine writing its own output cells). const updatedRow = await updateRow( { tableId, @@ -148,8 +142,7 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR actorUserId: authResult.userId, }, table, - requestId, - { dataWriteMode: 'patch' } + requestId ) // Only `null` when a `cancellationGuard` is supplied and the SQL guard // rejects the write — this route doesn't pass one, so reaching null is a bug. diff --git a/apps/sim/app/api/v1/tables/[tableId]/rows/[rowId]/route.ts b/apps/sim/app/api/v1/tables/[tableId]/rows/[rowId]/route.ts index 1f94852b107..2a7ea2fe7a5 100644 --- a/apps/sim/app/api/v1/tables/[tableId]/rows/[rowId]/route.ts +++ b/apps/sim/app/api/v1/tables/[tableId]/rows/[rowId]/route.ts @@ -139,10 +139,6 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR const idByName = buildIdByName(table.schema as TableSchema) const toNamedRow = namedRowMapper((table.schema as TableSchema).columns) - // Patch only the edited cells (`data = data || patch`) so concurrent edits to different cells - // of the same row don't clobber each other — a row stores all its cells in one jsonb column, - // so a full-object replace is last-write-wins across the whole row. See the app route for the - // full rationale. const updatedRow = await updateRow( { tableId, @@ -152,8 +148,7 @@ export const PATCH = withRouteHandler(async (request: NextRequest, context: RowR actorUserId, }, table, - requestId, - { dataWriteMode: 'patch' } + requestId ) // No `cancellationGuard` is passed here, so `updateRow` can't return null // from this caller. Defensive narrowing for TypeScript. diff --git a/apps/sim/lib/copilot/tools/server/table/user-table.ts b/apps/sim/lib/copilot/tools/server/table/user-table.ts index 2706d1e0572..a88e1dee903 100644 --- a/apps/sim/lib/copilot/tools/server/table/user-table.ts +++ b/apps/sim/lib/copilot/tools/server/table/user-table.ts @@ -727,9 +727,6 @@ export const userTableServerTool: BaseServerTool assertNotAborted() const idByName = buildIdByName(table.schema) const toNamedRow = namedRowMapper(table.schema.columns) - // Patch only the edited cells so a copilot edit doesn't clobber a concurrent user edit - // to a different cell of the same row (all cells share one jsonb column — a full-object - // replace is last-write-wins across the row). See the table row route for the rationale. const updatedRow = await updateRow( { tableId: args.tableId, @@ -739,8 +736,7 @@ export const userTableServerTool: BaseServerTool actorUserId: context.userId, }, table, - requestId, - { dataWriteMode: 'patch' } + requestId ) if (!updatedRow) { // Only the cell-task path passes a `cancellationGuard`; this caller diff --git a/apps/sim/lib/table/__tests__/update-row.test.ts b/apps/sim/lib/table/__tests__/update-row.test.ts index a97d5dd526a..467edef2944 100644 --- a/apps/sim/lib/table/__tests__/update-row.test.ts +++ b/apps/sim/lib/table/__tests__/update-row.test.ts @@ -1,11 +1,13 @@ /** * @vitest-environment node */ -import { dbChainMockFns, resetDbChainMock } from '@sim/testing' +import { tableRowExecutions, userTableRows } from '@sim/db/schema' +import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing' import { beforeEach, describe, expect, it, vi } from 'vitest' import { deleteColumn, renameColumn } from '@/lib/table/columns/service' import { batchInsertRows, + batchUpdateRows, insertRow, replaceTableRows, updateRow, @@ -64,6 +66,17 @@ function findExecutedRawSql(substring: string): string | undefined { return undefined } +/** + * The `data` payload of the last `.set(...)` row write. `updateRow` always writes a JSONB merge + * (`data = data || {changed}::jsonb`), so this is a `sql` fragment exposing `{ strings, values }`. + */ +function lastSetDataSql(): { strings?: string[]; values?: unknown[] } | undefined { + const payload = dbChainMockFns.set.mock.calls.at(-1)?.[0] as + | { data?: { strings?: string[]; values?: unknown[] } } + | undefined + return payload?.data +} + const EXISTING_ROW = { id: 'row-1', tableId: 'tbl-1', @@ -109,10 +122,15 @@ describe('updateRow — partial merge', () => { 'req-1' ) + // The returned row is the full merge… expect(result.data).toEqual({ name: 'Alice', age: 31 }) - expect(dbChainMockFns.set).toHaveBeenCalledWith( - expect.objectContaining({ data: { name: 'Alice', age: 31 } }) - ) + // …but the WRITE is a JSONB merge of only the changed cell (`data = data || {age:31}`), so the + // unchanged `name` column is never re-written — that's what keeps concurrent edits to different + // cells of the same row from clobbering each other. + const data = lastSetDataSql() + expect(data?.strings?.join('')).toContain(' || ') + expect(data?.values).toContain(JSON.stringify({ age: 31 })) + expect(data?.values).not.toContain(JSON.stringify({ name: 'Alice', age: 31 })) }) it('allows updating a single column without affecting others', async () => { @@ -123,9 +141,7 @@ describe('updateRow — partial merge', () => { ) expect(result.data).toEqual({ name: 'Bob', age: 30 }) - expect(dbChainMockFns.set).toHaveBeenCalledWith( - expect.objectContaining({ data: { name: 'Bob', age: 30 } }) - ) + expect(lastSetDataSql()?.values).toContain(JSON.stringify({ name: 'Bob' })) }) it('allows explicitly nulling a field while preserving others', async () => { @@ -136,9 +152,8 @@ describe('updateRow — partial merge', () => { ) expect(result.data).toEqual({ name: 'Alice', age: null }) - expect(dbChainMockFns.set).toHaveBeenCalledWith( - expect.objectContaining({ data: { name: 'Alice', age: null } }) - ) + // A cleared cell is written as present-with-null, not dropped. + expect(lastSetDataSql()?.values).toContain(JSON.stringify({ age: null })) }) it('handles a full-row update correctly (idempotent merge)', async () => { @@ -151,23 +166,6 @@ describe('updateRow — partial merge', () => { expect(result.data).toEqual({ name: 'Bob', age: 25 }) }) - it('sends only changed keys in JSONB patch mode while returning merged data', async () => { - const result = await updateRow( - { tableId: 'tbl-1', rowId: 'row-1', data: { age: 31 }, workspaceId: 'ws-1' }, - TABLE, - 'req-1', - { dataWriteMode: 'patch' } - ) - - expect(result?.data).toEqual({ name: 'Alice', age: 31 }) - const setPayload = dbChainMockFns.set.mock.calls.at(-1)?.[0] as - | { data?: { strings?: string[]; values?: unknown[] } } - | undefined - expect(setPayload?.data?.strings?.join('')).toContain(' || ') - expect(setPayload?.data?.values).toContain(JSON.stringify({ age: 31 })) - expect(setPayload?.data?.values).not.toContain(JSON.stringify({ name: 'Alice', age: 31 })) - }) - it('throws when the row does not exist', async () => { dbChainMockFns.limit.mockResolvedValueOnce([]) @@ -419,3 +417,46 @@ describe('mutation paths — SET LOCAL timeouts', () => { expect(findExecutedSqlContaining('hashtextextended')).toBe(true) }) }) + +describe('batchUpdateRows — per-row partial merge', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + }) + + it('writes each row as a JSONB merge of only its changed cells', async () => { + queueTableRows(userTableRows, [ + { id: 'row-1', data: { name: 'Alice', age: 30 } }, + { id: 'row-2', data: { name: 'Carol', age: 40 } }, + ]) + queueTableRows(tableRowExecutions, []) // loadExecutionsByRow → no sidecar rows + + await batchUpdateRows( + { + tableId: 'tbl-1', + workspaceId: 'ws-1', + updates: [ + { rowId: 'row-1', data: { age: 31 } }, + { rowId: 'row-2', data: { name: 'Dave' } }, + ], + }, + TABLE, + 'req-1' + ) + + // Each row write is `data || {changed}::jsonb`, carrying ONLY that row's changed cell — so a + // batch edit can't clobber a concurrent edit to another cell of the same row. + const writes = dbChainMockFns.set.mock.calls + .map((c) => c[0] as { data?: { strings?: string[]; values?: unknown[] } }) + .filter((p) => Array.isArray(p?.data?.strings)) + expect(writes.length).toBe(2) + for (const w of writes) { + expect(w.data?.strings?.join('')).toContain(' || ') + } + const values = writes.flatMap((w) => w.data?.values ?? []) + expect(values).toContain(JSON.stringify({ age: 31 })) + expect(values).toContain(JSON.stringify({ name: 'Dave' })) + // Never the whole row. + expect(values).not.toContain(JSON.stringify({ name: 'Alice', age: 31 })) + }) +}) diff --git a/apps/sim/lib/table/cell-write.test.ts b/apps/sim/lib/table/cell-write.test.ts index cd6eb08dbe7..bc0a65bea0a 100644 --- a/apps/sim/lib/table/cell-write.test.ts +++ b/apps/sim/lib/table/cell-write.test.ts @@ -141,7 +141,7 @@ describe('writeWorkflowGroupState', () => { }, TABLE, CONTEXT.requestId, - { dataWriteMode: 'patch', computedWrite: true } + { computedWrite: true } ) expect(mockAppendTableEvent).toHaveBeenCalledWith( expect.objectContaining({ diff --git a/apps/sim/lib/table/cell-write.ts b/apps/sim/lib/table/cell-write.ts index d32b0c988f2..4b1257d2021 100644 --- a/apps/sim/lib/table/cell-write.ts +++ b/apps/sim/lib/table/cell-write.ts @@ -86,7 +86,7 @@ export async function writeWorkflowGroupState( requestId, // `computedWrite` is what lets a workflow column keep populating on an // update-locked table; the lock still covers user-authored columns. - { dataWriteMode: 'patch', computedWrite: true } + { computedWrite: true } ) } else { result = await db.transaction((trx) => diff --git a/apps/sim/lib/table/rows/service.ts b/apps/sim/lib/table/rows/service.ts index 2f68b404918..41eedd066f8 100644 --- a/apps/sim/lib/table/rows/service.ts +++ b/apps/sim/lib/table/rows/service.ts @@ -1169,11 +1169,6 @@ class GuardRejected extends Error { * @throws Error if row not found or validation fails */ export interface UpdateRowOptions { - /** - * `patch` sends only changed keys to Postgres via JSONB concatenation while - * retaining the same merged-row validation and returned shape. - */ - dataWriteMode?: 'replace' | 'patch' /** * Marks the write as the workflow/enrichment engine filling its own output * cells, which exempts it from the update lock. Set by `cell-write.ts` only — @@ -1182,6 +1177,22 @@ export interface UpdateRowOptions { computedWrite?: boolean } +/** + * A row stores every cell in one jsonb `data` column, so a row update writes the changed cells as + * an in-DB JSONB merge (`data = data || {changed}::jsonb`) rather than replacing the whole object. + * Postgres evaluates the concat against the current committed row under its write lock, so + * concurrent edits to DIFFERENT cells of the same row both survive instead of the last writer + * clobbering the row from a stale read. Values come from the caller's already-coerced merged row; + * only the changed keys are sent, so an empty change set is a no-op on `data`. Whole-row + * replacement is `upsertRow`'s job, not this path (an update only ever adds/overwrites cells — + * it never deletes a key — so a full-object write would only differ by being racy). + */ +function jsonbMergePatch(changedColumnIds: string[], coercedRow: RowData): SQL { + const patch: RowData = {} + for (const columnId of changedColumnIds) patch[columnId] = coercedRow[columnId] + return sql`${userTableRows.data} || ${JSON.stringify(patch)}::jsonb` +} + export async function updateRow( data: UpdateRowData, table: TableDefinition, @@ -1241,14 +1252,7 @@ export async function updateRow( } const now = new Date() - const persistedDataPatch: RowData = {} - for (const columnId of Object.keys(data.data)) { - persistedDataPatch[columnId] = mergedData[columnId] - } - const persistedData = - options.dataWriteMode === 'patch' - ? sql`${userTableRows.data} || ${JSON.stringify(persistedDataPatch)}::jsonb` - : mergedData + const persistedData = jsonbMergePatch(Object.keys(data.data), mergedData) // Cell-task partial writes pass `cancellationGuard` so the upsert into // `tableRowExecutions` is a no-op when (a) a stop click already wrote @@ -1597,6 +1601,7 @@ export async function batchUpdateRows( const mergedUpdates: Array<{ rowId: string + changedColumnIds: string[] mergedData: RowData mergedExecutions: RowExecutions executionsPatch?: Record @@ -1631,6 +1636,7 @@ export async function batchUpdateRows( mergedUpdates.push({ rowId: update.rowId, + changedColumnIds: Object.keys(update.data), mergedData: merged, mergedExecutions, executionsPatch: effectiveExecutionsPatch, @@ -1660,11 +1666,13 @@ export async function batchUpdateRows( for (let i = 0; i < mergedUpdates.length; i += TABLE_LIMITS.UPDATE_BATCH_SIZE) { const batch = mergedUpdates.slice(i, i + TABLE_LIMITS.UPDATE_BATCH_SIZE) // Update row data in parallel; sidecar exec writes are sequential per - // row (each goes through writeExecutionsPatch's per-key upsert). - const dataPromises = batch.map(({ rowId, mergedData }) => + // row (each goes through writeExecutionsPatch's per-key upsert). Each row + // merges its changed cells via JSONB concat (see jsonbMergePatch) so a + // batch edit can't clobber a concurrent edit to another cell of the row. + const dataPromises = batch.map(({ rowId, changedColumnIds, mergedData }) => trx .update(userTableRows) - .set({ data: mergedData, updatedAt: now }) + .set({ data: jsonbMergePatch(changedColumnIds, mergedData), updatedAt: now }) .where(eq(userTableRows.id, rowId)) ) await Promise.all(dataPromises) diff --git a/apps/sim/lib/table/workflow-columns.ts b/apps/sim/lib/table/workflow-columns.ts index c0533306a7f..4054113d4e2 100644 --- a/apps/sim/lib/table/workflow-columns.ts +++ b/apps/sim/lib/table/workflow-columns.ts @@ -618,6 +618,8 @@ export async function cancelWorkflowGroupRuns( // re-enqueueing what we just cancelled. await Promise.allSettled( mutations.map((m) => + // Only touches execution state — `data: {}` is a no-op on the row's cells (updateRow merges + // the empty patch), so this cancel can't revert a user's concurrent edit to the same row. updateRow( { tableId, @@ -627,11 +629,7 @@ export async function cancelWorkflowGroupRuns( executionsPatch: m.executionsPatch, }, table, - `wfgrp-cancel-${m.rowId}`, - // This write only touches execution state (data is `{}`). `patch` makes the data write a - // true no-op (`data || '{}'`), whereas the default replace would rewrite the whole jsonb - // column and could revert a user's concurrent edit to a cell of the same row. - { dataWriteMode: 'patch' } + `wfgrp-cancel-${m.rowId}` ).catch((err) => { logger.error(`Failed to write cancelled state for row ${m.rowId}:`, err) })