mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
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.
This commit is contained in:
@@ -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 }))
|
||||
})
|
||||
})
|
||||
|
||||
@@ -141,7 +141,7 @@ describe('writeWorkflowGroupState', () => {
|
||||
},
|
||||
TABLE,
|
||||
CONTEXT.requestId,
|
||||
{ dataWriteMode: 'patch', computedWrite: true }
|
||||
{ computedWrite: true }
|
||||
)
|
||||
expect(mockAppendTableEvent).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
|
||||
@@ -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) =>
|
||||
|
||||
@@ -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<string, RowExecutionMetadata | null>
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user