Skip to content

Commit 104a6fc

Browse files
authored
fix(tables): make the atomic per-cell merge the only row-write path (#5970)
Table rows store all cells in one jsonb column; row updates did a full-object read-modify-write (row-level last-write-wins). updateRow now always merges changed cells via a shared jsonbMergePatch helper (data = data || {changed}::jsonb) under the row lock, and batchUpdateRows does the same per-row, so concurrent edits to different cells of the same row no longer clobber each other. Removed the never-beneficial replace mode; fix lives entirely in the service layer.
1 parent 1a4bfe4 commit 104a6fc

5 files changed

Lines changed: 96 additions & 45 deletions

File tree

apps/sim/lib/table/__tests__/update-row.test.ts

Lines changed: 68 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,13 @@
11
/**
22
* @vitest-environment node
33
*/
4-
import { dbChainMockFns, resetDbChainMock } from '@sim/testing'
4+
import { tableRowExecutions, userTableRows } from '@sim/db/schema'
5+
import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing'
56
import { beforeEach, describe, expect, it, vi } from 'vitest'
67
import { deleteColumn, renameColumn } from '@/lib/table/columns/service'
78
import {
89
batchInsertRows,
10+
batchUpdateRows,
911
insertRow,
1012
replaceTableRows,
1113
updateRow,
@@ -64,6 +66,17 @@ function findExecutedRawSql(substring: string): string | undefined {
6466
return undefined
6567
}
6668

69+
/**
70+
* The `data` payload of the last `.set(...)` row write. `updateRow` always writes a JSONB merge
71+
* (`data = data || {changed}::jsonb`), so this is a `sql` fragment exposing `{ strings, values }`.
72+
*/
73+
function lastSetDataSql(): { strings?: string[]; values?: unknown[] } | undefined {
74+
const payload = dbChainMockFns.set.mock.calls.at(-1)?.[0] as
75+
| { data?: { strings?: string[]; values?: unknown[] } }
76+
| undefined
77+
return payload?.data
78+
}
79+
6780
const EXISTING_ROW = {
6881
id: 'row-1',
6982
tableId: 'tbl-1',
@@ -109,10 +122,15 @@ describe('updateRow — partial merge', () => {
109122
'req-1'
110123
)
111124

125+
// The returned row is the full merge…
112126
expect(result.data).toEqual({ name: 'Alice', age: 31 })
113-
expect(dbChainMockFns.set).toHaveBeenCalledWith(
114-
expect.objectContaining({ data: { name: 'Alice', age: 31 } })
115-
)
127+
// …but the WRITE is a JSONB merge of only the changed cell (`data = data || {age:31}`), so the
128+
// unchanged `name` column is never re-written — that's what keeps concurrent edits to different
129+
// cells of the same row from clobbering each other.
130+
const data = lastSetDataSql()
131+
expect(data?.strings?.join('')).toContain(' || ')
132+
expect(data?.values).toContain(JSON.stringify({ age: 31 }))
133+
expect(data?.values).not.toContain(JSON.stringify({ name: 'Alice', age: 31 }))
116134
})
117135

118136
it('allows updating a single column without affecting others', async () => {
@@ -123,9 +141,7 @@ describe('updateRow — partial merge', () => {
123141
)
124142

125143
expect(result.data).toEqual({ name: 'Bob', age: 30 })
126-
expect(dbChainMockFns.set).toHaveBeenCalledWith(
127-
expect.objectContaining({ data: { name: 'Bob', age: 30 } })
128-
)
144+
expect(lastSetDataSql()?.values).toContain(JSON.stringify({ name: 'Bob' }))
129145
})
130146

131147
it('allows explicitly nulling a field while preserving others', async () => {
@@ -136,9 +152,8 @@ describe('updateRow — partial merge', () => {
136152
)
137153

138154
expect(result.data).toEqual({ name: 'Alice', age: null })
139-
expect(dbChainMockFns.set).toHaveBeenCalledWith(
140-
expect.objectContaining({ data: { name: 'Alice', age: null } })
141-
)
155+
// A cleared cell is written as present-with-null, not dropped.
156+
expect(lastSetDataSql()?.values).toContain(JSON.stringify({ age: null }))
142157
})
143158

144159
it('handles a full-row update correctly (idempotent merge)', async () => {
@@ -151,23 +166,6 @@ describe('updateRow — partial merge', () => {
151166
expect(result.data).toEqual({ name: 'Bob', age: 25 })
152167
})
153168

154-
it('sends only changed keys in JSONB patch mode while returning merged data', async () => {
155-
const result = await updateRow(
156-
{ tableId: 'tbl-1', rowId: 'row-1', data: { age: 31 }, workspaceId: 'ws-1' },
157-
TABLE,
158-
'req-1',
159-
{ dataWriteMode: 'patch' }
160-
)
161-
162-
expect(result?.data).toEqual({ name: 'Alice', age: 31 })
163-
const setPayload = dbChainMockFns.set.mock.calls.at(-1)?.[0] as
164-
| { data?: { strings?: string[]; values?: unknown[] } }
165-
| undefined
166-
expect(setPayload?.data?.strings?.join('')).toContain(' || ')
167-
expect(setPayload?.data?.values).toContain(JSON.stringify({ age: 31 }))
168-
expect(setPayload?.data?.values).not.toContain(JSON.stringify({ name: 'Alice', age: 31 }))
169-
})
170-
171169
it('throws when the row does not exist', async () => {
172170
dbChainMockFns.limit.mockResolvedValueOnce([])
173171

@@ -419,3 +417,46 @@ describe('mutation paths — SET LOCAL timeouts', () => {
419417
expect(findExecutedSqlContaining('hashtextextended')).toBe(true)
420418
})
421419
})
420+
421+
describe('batchUpdateRows — per-row partial merge', () => {
422+
beforeEach(() => {
423+
vi.clearAllMocks()
424+
resetDbChainMock()
425+
})
426+
427+
it('writes each row as a JSONB merge of only its changed cells', async () => {
428+
queueTableRows(userTableRows, [
429+
{ id: 'row-1', data: { name: 'Alice', age: 30 } },
430+
{ id: 'row-2', data: { name: 'Carol', age: 40 } },
431+
])
432+
queueTableRows(tableRowExecutions, []) // loadExecutionsByRow → no sidecar rows
433+
434+
await batchUpdateRows(
435+
{
436+
tableId: 'tbl-1',
437+
workspaceId: 'ws-1',
438+
updates: [
439+
{ rowId: 'row-1', data: { age: 31 } },
440+
{ rowId: 'row-2', data: { name: 'Dave' } },
441+
],
442+
},
443+
TABLE,
444+
'req-1'
445+
)
446+
447+
// Each row write is `data || {changed}::jsonb`, carrying ONLY that row's changed cell — so a
448+
// batch edit can't clobber a concurrent edit to another cell of the same row.
449+
const writes = dbChainMockFns.set.mock.calls
450+
.map((c) => c[0] as { data?: { strings?: string[]; values?: unknown[] } })
451+
.filter((p) => Array.isArray(p?.data?.strings))
452+
expect(writes.length).toBe(2)
453+
for (const w of writes) {
454+
expect(w.data?.strings?.join('')).toContain(' || ')
455+
}
456+
const values = writes.flatMap((w) => w.data?.values ?? [])
457+
expect(values).toContain(JSON.stringify({ age: 31 }))
458+
expect(values).toContain(JSON.stringify({ name: 'Dave' }))
459+
// Never the whole row.
460+
expect(values).not.toContain(JSON.stringify({ name: 'Alice', age: 31 }))
461+
})
462+
})

apps/sim/lib/table/cell-write.test.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -141,7 +141,7 @@ describe('writeWorkflowGroupState', () => {
141141
},
142142
TABLE,
143143
CONTEXT.requestId,
144-
{ dataWriteMode: 'patch', computedWrite: true }
144+
{ computedWrite: true }
145145
)
146146
expect(mockAppendTableEvent).toHaveBeenCalledWith(
147147
expect.objectContaining({

apps/sim/lib/table/cell-write.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@ export async function writeWorkflowGroupState(
8686
requestId,
8787
// `computedWrite` is what lets a workflow column keep populating on an
8888
// update-locked table; the lock still covers user-authored columns.
89-
{ dataWriteMode: 'patch', computedWrite: true }
89+
{ computedWrite: true }
9090
)
9191
} else {
9292
result = await db.transaction((trx) =>

apps/sim/lib/table/rows/service.ts

Lines changed: 24 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -1169,11 +1169,6 @@ class GuardRejected extends Error {
11691169
* @throws Error if row not found or validation fails
11701170
*/
11711171
export interface UpdateRowOptions {
1172-
/**
1173-
* `patch` sends only changed keys to Postgres via JSONB concatenation while
1174-
* retaining the same merged-row validation and returned shape.
1175-
*/
1176-
dataWriteMode?: 'replace' | 'patch'
11771172
/**
11781173
* Marks the write as the workflow/enrichment engine filling its own output
11791174
* cells, which exempts it from the update lock. Set by `cell-write.ts` only —
@@ -1182,6 +1177,22 @@ export interface UpdateRowOptions {
11821177
computedWrite?: boolean
11831178
}
11841179

1180+
/**
1181+
* A row stores every cell in one jsonb `data` column, so a row update writes the changed cells as
1182+
* an in-DB JSONB merge (`data = data || {changed}::jsonb`) rather than replacing the whole object.
1183+
* Postgres evaluates the concat against the current committed row under its write lock, so
1184+
* concurrent edits to DIFFERENT cells of the same row both survive instead of the last writer
1185+
* clobbering the row from a stale read. Values come from the caller's already-coerced merged row;
1186+
* only the changed keys are sent, so an empty change set is a no-op on `data`. Whole-row
1187+
* replacement is `upsertRow`'s job, not this path (an update only ever adds/overwrites cells —
1188+
* it never deletes a key — so a full-object write would only differ by being racy).
1189+
*/
1190+
function jsonbMergePatch(changedColumnIds: string[], coercedRow: RowData): SQL {
1191+
const patch: RowData = {}
1192+
for (const columnId of changedColumnIds) patch[columnId] = coercedRow[columnId]
1193+
return sql`${userTableRows.data} || ${JSON.stringify(patch)}::jsonb`
1194+
}
1195+
11851196
export async function updateRow(
11861197
data: UpdateRowData,
11871198
table: TableDefinition,
@@ -1241,14 +1252,7 @@ export async function updateRow(
12411252
}
12421253

12431254
const now = new Date()
1244-
const persistedDataPatch: RowData = {}
1245-
for (const columnId of Object.keys(data.data)) {
1246-
persistedDataPatch[columnId] = mergedData[columnId]
1247-
}
1248-
const persistedData =
1249-
options.dataWriteMode === 'patch'
1250-
? sql`${userTableRows.data} || ${JSON.stringify(persistedDataPatch)}::jsonb`
1251-
: mergedData
1255+
const persistedData = jsonbMergePatch(Object.keys(data.data), mergedData)
12521256

12531257
// Cell-task partial writes pass `cancellationGuard` so the upsert into
12541258
// `tableRowExecutions` is a no-op when (a) a stop click already wrote
@@ -1597,6 +1601,7 @@ export async function batchUpdateRows(
15971601

15981602
const mergedUpdates: Array<{
15991603
rowId: string
1604+
changedColumnIds: string[]
16001605
mergedData: RowData
16011606
mergedExecutions: RowExecutions
16021607
executionsPatch?: Record<string, RowExecutionMetadata | null>
@@ -1631,6 +1636,7 @@ export async function batchUpdateRows(
16311636

16321637
mergedUpdates.push({
16331638
rowId: update.rowId,
1639+
changedColumnIds: Object.keys(update.data),
16341640
mergedData: merged,
16351641
mergedExecutions,
16361642
executionsPatch: effectiveExecutionsPatch,
@@ -1660,11 +1666,13 @@ export async function batchUpdateRows(
16601666
for (let i = 0; i < mergedUpdates.length; i += TABLE_LIMITS.UPDATE_BATCH_SIZE) {
16611667
const batch = mergedUpdates.slice(i, i + TABLE_LIMITS.UPDATE_BATCH_SIZE)
16621668
// Update row data in parallel; sidecar exec writes are sequential per
1663-
// row (each goes through writeExecutionsPatch's per-key upsert).
1664-
const dataPromises = batch.map(({ rowId, mergedData }) =>
1669+
// row (each goes through writeExecutionsPatch's per-key upsert). Each row
1670+
// merges its changed cells via JSONB concat (see jsonbMergePatch) so a
1671+
// batch edit can't clobber a concurrent edit to another cell of the row.
1672+
const dataPromises = batch.map(({ rowId, changedColumnIds, mergedData }) =>
16651673
trx
16661674
.update(userTableRows)
1667-
.set({ data: mergedData, updatedAt: now })
1675+
.set({ data: jsonbMergePatch(changedColumnIds, mergedData), updatedAt: now })
16681676
.where(eq(userTableRows.id, rowId))
16691677
)
16701678
await Promise.all(dataPromises)

apps/sim/lib/table/workflow-columns.ts

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -618,6 +618,8 @@ export async function cancelWorkflowGroupRuns(
618618
// re-enqueueing what we just cancelled.
619619
await Promise.allSettled(
620620
mutations.map((m) =>
621+
// Only touches execution state — `data: {}` is a no-op on the row's cells (updateRow merges
622+
// the empty patch), so this cancel can't revert a user's concurrent edit to the same row.
621623
updateRow(
622624
{
623625
tableId,

0 commit comments

Comments
 (0)