fix(tables): re-check update batches under a moved schema, own-key reads

- The background update runner re-reads a batch's rows inside the batch
  transaction and re-validates them merged with the re-derived patch when
  the schema moved since the page was checked, as the inline bulk update
  does; an unchanged schema adds no query. updatePageByIds takes an async
  per-batch hook with the transaction for this.
- batchInsertRowsWithTx returns the definition it validated against, and
  batchInsertRows dispatches its insert triggers with it.
- Row cells and patch keys are read as own properties (Object.hasOwn), so a
  legacy column keyed by a prototype name such as `constructor` is not
  seen in rows or patches that do not hold it.
- The long schema-wait test asserts the write is waiting on the schema lock
  before the holder commits.

Author: waleedlatif1

apps/sim/lib/table/import-data.ts

@@ -326,7 +326,7 @@ export async function importAppendRows(
         generateId().slice(0, 8),
         { uniqueColumnsLocked: true }
       )
-      inserted.push(...batchInserted)
+      inserted.push(...batchInserted.rows)
     }
     return { inserted, table: working }
   })

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

@@ -681,15 +681,20 @@ export interface PagePatch {
 /**
  * Applies a JSONB-merge patch (`data || patchJson`) to a page of row ids, committed in
  * UPDATE_BATCH_SIZE chunks (each its own transaction, 60s timeout) so a large background update
- * makes incremental, resumable progress. Each batch takes its patch from `patchFor`, called inside
- * the batch's transaction with the definition `revalidate` read there, so a caller can derive it
- * from the live schema. Returns the number of rows updated.
+ * makes incremental, resumable progress. Each batch takes its patch from `prepare`, called inside
+ * the batch's transaction with the definition `revalidate` read there and the batch's row ids, so a
+ * caller can derive it, and check the rows it merges into, against the live schema. Returns the
+ * number of rows updated.
  */
 export async function updatePageByIds(
   tableId: string,
   workspaceId: string,
   rowIds: string[],
-  patchFor: (table: TableDefinition | undefined) => PagePatch | null,
+  prepare: (
+    trx: DbTransaction,
+    table: TableDefinition | undefined,
+    batch: string[]
+  ) => Promise<PagePatch | null>,
   /** Proof the caller asserted the update lock (see `mutation-locks.ts`). */
   _proof: MutationProof<'update'>,
   /** Re-asserts the lock inside each batch transaction. See {@link guardBatch}. */
@@ -701,7 +706,7 @@ export async function updatePageByIds(
     const batch = rowIds.slice(i, i + TABLE_LIMITS.UPDATE_BATCH_SIZE)
     const rows = await db.transaction(async (trx) => {
       await setTableTxTimeouts(trx, { statementMs: 60_000 })
-      const patch = patchFor(await guardBatch(trx, tableId, revalidate))
+      const patch = await prepare(trx, await guardBatch(trx, tableId, revalidate), batch)
       if (!patch) return []
       return mutateTableRowsWithSecretProvenance(trx, {
         rows: batch.map((rowId) => ({ rowId, provenance: patch.secretProvenance })),

apps/sim/lib/table/rows/row-writes.integration.ts

@@ -9,7 +9,7 @@ import { userTableDefinitions, userTableRows } from '@sim/db/schema'
 import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
 import { storageServiceMock, storageServiceMockFns } from '@sim/testing/mocks/storage-service.mock'
 import { tableBillingMock, tableBillingMockFns } from '@sim/testing/mocks/table-billing.mock'
-import { tableTriggerMock } from '@sim/testing/mocks/table-trigger.mock'
+import { tableTriggerMock, tableTriggerMockFns } from '@sim/testing/mocks/table-trigger.mock'
 import { tableWorkflowColumnsMock } from '@sim/testing/mocks/table-workflow-columns.mock'
 import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
@@ -97,6 +97,27 @@ async function rowsVersion(tableId: string): Promise<number> {
 const textColumns = (...ids: string[]): ColumnDefinition[] =>
   ids.map((id) => ({ id, name: id, type: 'string' }))
 
+/** Sessions waiting on the table's schema lock, matched by the key's hash as `pg_locks` shows it. */
+async function schemaLockWaiters(tableId: string): Promise<number> {
+  const [{ waiting }] = await control<{ waiting: number }[]>`
+    WITH lock AS (SELECT hashtextextended(${`user_table_schema:${tableId}`}, 0) AS key)
+    SELECT count(*)::int AS waiting FROM pg_locks l CROSS JOIN lock
+    WHERE l.locktype = 'advisory' AND NOT l.granted
+      AND l.classid = ((lock.key >> 32) & 4294967295)::oid
+      AND l.objid = (lock.key & 4294967295)::oid`
+  return waiting
+}
+
+/** Polls until exactly `expected` sessions wait on the table's schema lock, and asserts it. */
+async function untilSchemaLockWaiters(tableId: string, expected: number) {
+  let waiting = 0
+  for (let attempt = 0; attempt < 400 && waiting !== expected; attempt++) {
+    await sleep(5)
+    waiting = await schemaLockWaiters(tableId)
+  }
+  expect(waiting).toBe(expected)
+}
+
 describe('table row writes against real PostgreSQL', () => {
   beforeAll(async () => {
     await control`INSERT INTO "user" (id, name, email, email_verified, created_at, updated_at)
@@ -1050,6 +1071,61 @@ describe('table row writes against real PostgreSQL', () => {
       }
     )
 
+    it('fires the insert trigger of a batch insert with the column names of the live schema', async () => {
+      const table = await seededTable()
+      const stale: TableDefinition = {
+        ...table,
+        schema: {
+          columns: table.schema.columns.map((column) =>
+            column.id === 'note' ? { ...column, name: 'old_note' } : column
+          ),
+        },
+      }
+      tableTriggerMockFns.mockFireTableTrigger.mockClear()
+
+      await batchInsertRows(
+        {
+          tableId: table.id,
+          workspaceId,
+          rows: [{ key: 'k3', note: 'n' }],
+          secretProvenance: undefined,
+          capabilityGovernedUserId: null,
+        },
+        stale,
+        'stale-schema'
+      )
+
+      const [trigger] = tableTriggerMockFns.mockFireTableTrigger.mock.calls
+      const schema = trigger?.[6] as { columns: ColumnDefinition[] }
+      expect(schema.columns.map((column) => column.name)).toContain('note')
+    })
+
+    it('does not count a legacy unique column named after an object prototype key as patched', async () => {
+      const table = await createTable([
+        { name: 'constructor', type: 'string', unique: true },
+        { id: 'note', name: 'note', type: 'string' },
+      ])
+      await seedRows(table.id, [
+        { id: `${table.id}-a`, data: { constructor: 'a', note: 'n' }, orderKey: 'a0' },
+        { id: `${table.id}-b`, data: { constructor: 'b', note: 'n' }, orderKey: 'a1' },
+      ])
+
+      const result = await updateRowsByFilter(
+        table,
+        {
+          filter: { note: 'n' },
+          data: { note: 'patched' },
+          // The limited path: the paged one skips rows created after its JS-clock cutoff.
+          limit: 10,
+          secretProvenance: undefined,
+          capabilityGovernedUserId: null,
+        },
+        'prototype-key'
+      )
+
+      expect(result.affectedCount).toBe(2)
+    })
+
     it('refuses a bulk update writing one value to rows of a column made unique since its snapshot', async () => {
       const table = await seededTable()
       await updateColumnConstraints(
@@ -1120,17 +1196,7 @@ describe('table row writes against real PostgreSQL', () => {
           () => 'rejected'
         )
 
-        let waiting = 0
-        for (let attempt = 0; attempt < 400 && waiting === 0; attempt++) {
-          await sleep(5)
-          ;[{ waiting }] = await control<{ waiting: number }[]>`
-            WITH lock AS (SELECT hashtextextended(${schemaLockKey}, 0) AS key)
-            SELECT count(*)::int AS waiting FROM pg_locks l CROSS JOIN lock
-            WHERE l.locktype = 'advisory' AND NOT l.granted
-              AND l.classid = ((lock.key >> 32) & 4294967295)::oid
-              AND l.objid = (lock.key & 4294967295)::oid`
-        }
-        expect(waiting).toBe(1)
+        await untilSchemaLockWaiters(table.id, 1)
         expect(await Promise.race([settled, sleep(50).then(() => 'pending')])).toBe('pending')
 
         await holder`COMMIT`
@@ -1143,7 +1209,7 @@ describe('table row writes against real PostgreSQL', () => {
 
     /**
      * Holds the table's schema lock exclusively, starts `write`, and commits once `write` is seen
-     * waiting and `holdMs` has passed. Settles as `write` does.
+     * waiting on it and `holdMs` more has passed. Settles as `write` does.
      */
     async function writeBehindSchemaLock(
       table: TableDefinition,
@@ -1156,6 +1222,7 @@ describe('table row writes against real PostgreSQL', () => {
         await holder`SELECT pg_advisory_xact_lock(hashtextextended(${`user_table_schema:${table.id}`}, 0))`
         const pending = write()
         pending.catch(() => {})
+        await untilSchemaLockWaiters(table.id, 1)
         await sleep(holdMs)
         await holder`COMMIT`
         return await pending
@@ -1301,25 +1368,6 @@ describe('table row writes against real PostgreSQL', () => {
       return { table, run, retry }
     }
 
-    async function schemaLockWaiters(tableId: string): Promise<number> {
-      const [{ waiting }] = await control<{ waiting: number }[]>`
-        WITH lock AS (SELECT hashtextextended(${`user_table_schema:${tableId}`}, 0) AS key)
-        SELECT count(*)::int AS waiting FROM pg_locks l CROSS JOIN lock
-        WHERE l.locktype = 'advisory' AND NOT l.granted
-          AND l.classid = ((lock.key >> 32) & 4294967295)::oid
-          AND l.objid = (lock.key & 4294967295)::oid`
-      return waiting
-    }
-
-    async function untilSchemaLockWaiters(tableId: string, expected: number) {
-      let waiting = 0
-      for (let attempt = 0; attempt < 400 && waiting !== expected; attempt++) {
-        await sleep(5)
-        waiting = await schemaLockWaiters(tableId)
-      }
-      expect(waiting).toBe(expected)
-    }
-
     /**
      * Commits `change` between the job's first and second batch. A held schema lock stops the
      * first batch; `change` then queues behind it, and a waiting lock is granted in queue order, so
@@ -1478,6 +1526,28 @@ describe('table row writes against real PostgreSQL', () => {
       expect(count).toBe(TABLE_LIMITS.UPDATE_BATCH_SIZE)
     })
 
+    it('refuses a batch whose re-derived patch grows a row past the size limit', async () => {
+      const { table, run } = await seededJob({ note: 7 })
+      await changeSchemaUnderLock(table.id, `jsonb_set(schema, '{columns,3,type}', '"number"')`)
+      const bigRowId = `${table.id}-${String(TABLE_LIMITS.UPDATE_BATCH_SIZE + 1).padStart(4, '0')}`
+      // Stored jsonb orders keys by length, so a merged row reads kind, email, filler, then note.
+      const email = `e${TABLE_LIMITS.UPDATE_BATCH_SIZE + 1}@example.test`
+      const base = Buffer.byteLength(JSON.stringify({ kind: 'seed', email, filler: '', note: 7 }))
+      await control`UPDATE user_table_rows
+        SET data = data || jsonb_build_object('filler', repeat('x', ${getMaxRowSizeBytes() - base}))
+        WHERE id = ${bigRowId}`
+
+      const [job, change] = await changeBetweenBatches(table.id, run, () =>
+        changeSchemaUnderLock(table.id, `jsonb_set(schema, '{columns,3,type}', '"string"')`)
+      )
+
+      expect(change.status).toBe('fulfilled')
+      expect(job.status === 'rejected' && job.reason).toBeInstanceOf(UpdatePatchRejectedError)
+      const [{ count }] = await control<{ count: number }[]>`SELECT count(*)::int AS count
+        FROM user_table_rows WHERE table_id = ${table.id} AND data ? 'note'`
+      expect(count).toBe(TABLE_LIMITS.UPDATE_BATCH_SIZE)
+    })
+
     it('writes a date patch given as an epoch number', async () => {
       const { table, run } = await seededJob({ due: 1704067200000 })
 

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

@@ -310,17 +310,17 @@ export async function batchInsertRows(
     addedRows: data.rows.length,
   })
 
-  const result = await db.transaction((trx) =>
+  const { rows, table: written } = await db.transaction((trx) =>
     batchInsertRowsWithTx(trx, data, table, requestId, { ...options, lockSchema: true })
   )
   notifyTableRowUsage({
     workspaceId: table.workspaceId,
     currentRowCount: table.rowCount,
-    addedRows: result.length,
+    addedRows: rows.length,
     limit: rowLimit,
   })
-  dispatchAfterBatchInsert(table, result, requestId, data.userId, data.capabilityGovernedUserId)
-  return result
+  dispatchAfterBatchInsert(written, rows, requestId, data.userId, data.capabilityGovernedUserId)
+  return rows
 }
 
 /** Options for the transaction-bound whole-row writers. */
@@ -345,7 +345,8 @@ export interface BatchInsertOptions extends TxRowWriteOptions {
  * Transaction-bound variant of `batchInsertRows`. Validates rows and unique
  * constraints, then performs INSERTs inside the provided transaction. Caller
  * is responsible for opening the transaction. Use when row inserts must be
- * atomic with other writes (e.g., schema mutations) on the same tx.
+ * atomic with other writes (e.g., schema mutations) on the same tx. Returns the inserted rows and
+ * the definition they were validated against, for the caller's post-commit dispatch.
  *
  * Capacity is NOT checked here (it would mean a billing-pool read inside the tx).
  * Callers gate it before opening the tx — see `batchInsertRows` and the import paths.
@@ -361,7 +362,7 @@ export async function batchInsertRowsWithTx(
   snapshot: TableDefinition,
   requestId: string,
   options: BatchInsertOptions = {}
-): Promise<TableRow[]> {
+): Promise<{ rows: TableRow[]; table: TableDefinition }> {
   assertRowInsert(snapshot)
   const timeouts = { statementMs: 60_000 }
   const table = options.lockSchema ? await lockLiveTableSchema(trx, snapshot, timeouts) : snapshot
@@ -456,7 +457,7 @@ export async function batchInsertRowsWithTx(
   }))
 
   await options.readProvenance?.capture(trx, result)
-  return result
+  return { rows: result, table }
 }
 
 /**
@@ -1913,7 +1914,7 @@ export async function updateRow(
         mergedExecutions = applyExecutionsPatch(existingRow.executions, effectiveExecutionsPatch)
       }
       const persistedData = jsonbMergePatch(
-        Object.keys(data.data).filter((columnId) => columnId in mergedData),
+        Object.keys(data.data).filter((columnId) => Object.hasOwn(mergedData, columnId)),
         mergedData
       )
       const patchedUniqueColumns = getUniqueColumns(live.schema).filter((column) =>
@@ -2188,8 +2189,8 @@ async function assertBulkUpdateUnique(
   rows: BulkUpdateMatch[],
   patch: RowData
 ): Promise<void> {
-  const uniqueColumnsInUpdate = getUniqueColumns(table.schema).filter(
-    (column) => getColumnId(column) in patch
+  const uniqueColumnsInUpdate = getUniqueColumns(table.schema).filter((column) =>
+    Object.hasOwn(patch, getColumnId(column))
   )
   if (uniqueColumnsInUpdate.length === 0 || rows.length === 0) return
   if (rows.length > 1) {
@@ -2723,8 +2724,8 @@ export async function batchUpdateRows(
             `Row ${update.rowId}: ${refit.errors.join(', ')}`
           )
         }
-        update.changedColumnIds = update.changedColumnIds.filter(
-          (columnId) => columnId in update.mergedData
+        update.changedColumnIds = update.changedColumnIds.filter((columnId) =>
+          Object.hasOwn(update.mergedData, columnId)
         )
         const cleared = deriveExecClearsForDataPatch(
           request.data,

apps/sim/lib/table/rows/unique-locks.ts

@@ -17,7 +17,7 @@ import { type AdvisoryXactLockRequest, acquireAdvisoryXactLocks } from '@/lib/db
 import { getColumnId } from '@/lib/table/column-keys'
 import type { DbTransaction } from '@/lib/table/planner'
 import type { RowData, TableDefinition } from '@/lib/table/types'
-import { getUniqueColumns, uniqueValueKey } from '@/lib/table/validation'
+import { cellOf, getUniqueColumns, uniqueValueKey } from '@/lib/table/validation'
 
 const UNIQUE_LOCK_TAG = 'user_table_unique_value'
 
@@ -63,7 +63,7 @@ export async function lockUniqueValues(
     const columnId = getColumnId(column)
     if (columnIds && !columnIds.has(columnId)) continue
     for (const row of rows) {
-      const value = row[columnId]
+      const value = cellOf(row, columnId)
       if (value === null || value === undefined) continue
       const key = `${tableKey}:${columnId}:${uniqueValueKey(value, column)}`
       if (!valueKeys.has(key) && valueKeys.size === MAX_VALUE_LOCKS) {

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

@@ -23,18 +23,22 @@ vi.mock('@/lib/table/billing', () => tableBillingMock)
 
 vi.mock('@/lib/table/rows/live-schema', () => tableRowsLiveSchemaMock)
 
-vi.mock('@/lib/table/validation', async (importOriginal) => ({
-  uniqueValueKey: (await importOriginal<typeof import('@/lib/table/validation')>()).uniqueValueKey,
-  validateRowSize: vi.fn(() => ({ valid: true, errors: [] })),
-  validateRowAgainstSchema: vi.fn(() => ({ valid: true, errors: [] })),
-  coerceRowToSchema: vi.fn(() => ({ valid: true, errors: [] })),
-  coerceRowValues: vi.fn(),
-  validateTableName: vi.fn(() => ({ valid: true, errors: [] })),
-  validateTableSchema: vi.fn(() => ({ valid: true, errors: [] })),
-  getUniqueColumns: vi.fn(() => []),
-  checkUniqueConstraintsDb: vi.fn(async () => ({ valid: true, errors: [] })),
-  checkBatchUniqueConstraintsDb: vi.fn(async () => ({ valid: true, errors: [] })),
-}))
+vi.mock('@/lib/table/validation', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('@/lib/table/validation')>()
+  return {
+    cellOf: actual.cellOf,
+    uniqueValueKey: actual.uniqueValueKey,
+    validateRowSize: vi.fn(() => ({ valid: true, errors: [] })),
+    validateRowAgainstSchema: vi.fn(() => ({ valid: true, errors: [] })),
+    coerceRowToSchema: vi.fn(() => ({ valid: true, errors: [] })),
+    coerceRowValues: vi.fn(),
+    validateTableName: vi.fn(() => ({ valid: true, errors: [] })),
+    validateTableSchema: vi.fn(() => ({ valid: true, errors: [] })),
+    getUniqueColumns: vi.fn(() => []),
+    checkUniqueConstraintsDb: vi.fn(async () => ({ valid: true, errors: [] })),
+    checkBatchUniqueConstraintsDb: vi.fn(async () => ({ valid: true, errors: [] })),
+  }
+})
 
 /**
  * Inspects the queued `trx.execute(...)` calls for SQL containing `substring`.