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 9e0f35fc172..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,