fix(tables): make the atomic per-cell merge the only row-write path (fix row-level LWW)

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 }))
+  })
+})

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({

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) =>

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<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)

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,