improvement(tables): bump rows version once at commit and serialize same-value unique writes (#8335)

* improvement(tables): bump a table's rows version once at commit

* fix(tables): serialize concurrent writes of the same unique value with per-value advisory locks

* fix(tables): take the batched advisory locks on a transaction

* fix(tables): cap value locks at the per-connection lock budget and lock unique columns on every import batch

Lower the per-transaction value-lock cap to 64 so concurrent writers cannot
exhaust the shared lock table; larger writes keep the exclusive column-lock
fallback. Import batches now take the unique-column locks and run their unique
check inside their transaction, batchUpdateRows rejects two updates writing the
same unique value and locks and checks only the unique columns each update
changes, and the 0390 comment describes the mechanism only.

* chore(tables): share one unique-value key between the check and the locks

* fix(tables): lock a table's unique values under one table key so wide schemas stay within the lock budget

* fix(tables): key unique JSON values by canonical form so key order never hides a batch duplicate

Author: waleedlatif1

apps/sim/lib/db/advisory-locks.ts

@@ -44,3 +44,30 @@ export async function tryAcquireAdvisoryXactLock(
   )
   return Boolean(lock?.acquired)
 }
+
+/** One lock of an {@link acquireAdvisoryXactLocks} set. */
+export interface AdvisoryXactLockRequest {
+  key: string
+  /** Takes the lock in shared mode, which conflicts only with exclusive holders of `key`. */
+  shared: boolean
+}
+
+/**
+ * Blocks until every transaction-scoped advisory lock in `locks` is held, in one round trip.
+ * `jsonb_array_elements` emits elements in array order and each row's lock call runs before the
+ * next row is read, so the locks are taken exactly in the order given: callers that must not
+ * deadlock pass their locks in one global order. The locks release on commit or rollback.
+ */
+export async function acquireAdvisoryXactLocks(
+  tx: DbTransaction,
+  tag: string,
+  locks: readonly AdvisoryXactLockRequest[]
+): Promise<void> {
+  if (locks.length === 0) return
+  await tx.execute(sql`
+    SELECT CASE WHEN (lock ->> 'shared')::boolean
+      THEN pg_advisory_xact_lock_shared(hashtextextended(lock ->> 'key', 0))
+      ELSE pg_advisory_xact_lock(hashtextextended(lock ->> 'key', 0))
+    END
+    FROM jsonb_array_elements(${JSON.stringify(locks)}::jsonb) AS lock ${lockTag(tag)}`)
+}

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

@@ -25,6 +25,7 @@ import {
   mutateTableRowsWithSecretProvenance,
 } from '@/lib/table/rows/secret-provenance'
 import { batchInsertRowsWithTx, replaceTableRowsWithTx } from '@/lib/table/rows/service'
+import { lockUniqueColumns } from '@/lib/table/rows/unique-locks'
 import { addTableColumnsWithTx, auditTableColumnsAdded, getTableById } from '@/lib/table/service'
 import type {
   ReplaceRowsResult,
@@ -58,10 +59,12 @@ export interface BulkImportBatch {
  * Inserts one batch of rows for an async import in a single committed statement.
  *
  * Differs from {@link batchInsertRowsWithTx} for the bulk-load case: caller-supplied
- * contiguous order keys (no `acquireRowOrderLock` scan — an
- * import owns its hidden table as the sole writer), no `RETURNING`, and **no
- * `fireTableTrigger` / `runWorkflowColumn`** (a 1M-row import must not dispatch a
- * workflow run per row). `row_count` is maintained set-based by the statement-level
+ * contiguous order keys (no `acquireRowOrderLock` scan; the caller threads each batch's
+ * anchor from the previous one), no `RETURNING`, and **no `fireTableTrigger` /
+ * `runWorkflowColumn`** (a 1M-row import must not dispatch a workflow run per row).
+ * Append and replace imports run this against the live table, so other writers can
+ * race it: the batch holds the table's unique columns exclusively while it checks and
+ * inserts. `row_count` is maintained set-based by the statement-level
  * trigger. There is no surrounding transaction and no rollback: each batch commits on
  * its own, so committed batches persist even if a later batch fails.
  *
@@ -99,25 +102,9 @@ export async function bulkInsertImportBatch(
     }
   }
 
-  const uniqueColumns = getUniqueColumns(table.schema)
-  if (uniqueColumns.length > 0) {
-    const uniqueResult = await checkBatchUniqueConstraintsDb(
-      data.tableId,
-      data.rows,
-      table.schema,
-      db
-    )
-    if (!uniqueResult.valid) {
-      throw new OrchestrationError(
-        'validation',
-        uniqueResult.errors.map((e) => `Row ${e.row + 1}: ${e.errors.join(', ')}`).join('; ')
-      )
-    }
-  }
-
   const now = new Date()
-  // Import worker is the table's sole writer; append keys after the anchor the caller threads
-  // from the previous batch's last key — no per-batch max(order_key) scan over a growing table.
+  // Append keys after the anchor the caller threads from the previous batch's last key — no
+  // per-batch max(order_key) scan over a growing table.
   const orderKeys = nKeysBetween(data.afterOrderKey ?? null, null, data.rows.length)
   const rowsToInsert = data.rows.map((rowData, i) => ({
     id: `row_${generateId().replace(/-/g, '')}`,
@@ -133,6 +120,22 @@ export async function bulkInsertImportBatch(
 
   const inserted = await db.transaction(async (trx) => {
     await guardBatch(trx, data.tableId, revalidate)
+    if (getUniqueColumns(table.schema).length > 0) {
+      // The whole-table unique lock, not per-value: a batch is far more values than the value-lock cap.
+      await lockUniqueColumns(trx, table)
+      const uniqueResult = await checkBatchUniqueConstraintsDb(
+        data.tableId,
+        data.rows,
+        table.schema,
+        trx
+      )
+      if (!uniqueResult.valid) {
+        throw new OrchestrationError(
+          'validation',
+          uniqueResult.errors.map((e) => `Row ${e.row + 1}: ${e.errors.join(', ')}`).join('; ')
+        )
+      }
+    }
     return mutateTableRowsWithSecretProvenance(trx, {
       rows: rowsToInsert.map((row) => ({
         rowId: row.id,
@@ -282,6 +285,9 @@ export async function importAppendRows(
   })
   const result = await db.transaction(async (trx) => {
     let working = await refreshUnderLock(trx, table)
+    // Lock the unique columns whole, ahead of the row-order lock: per-value locks for every row of
+    // an import would flood the server's lock table.
+    await lockUniqueColumns(trx, working)
     if (additions.length > 0) {
       // Take the row-order lock before creating columns so this path uses the
       // same rows_pos → user_table_definitions order as plain inserts. Creating
@@ -305,7 +311,8 @@ export async function importAppendRows(
           secretProvenance: batch.map(createExactEmptyTableRowSecretProvenance),
         },
         working,
-        generateId().slice(0, 8)
+        generateId().slice(0, 8),
+        { uniqueColumnsLocked: true }
       )
       inserted.push(...batchInserted)
     }
@@ -347,6 +354,7 @@ export async function importReplaceRows(
   })
   const result = await db.transaction(async (trx) => {
     let working = await refreshUnderLock(trx, table)
+    await lockUniqueColumns(trx, working)
     if (additions.length > 0) {
       await acquireRowOrderLock(trx, table.id)
       working = await addTableColumnsWithTx(trx, working, additions, requestId)

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

@@ -332,6 +332,11 @@ export async function insertOrderedRow(params: {
   secretProvenance?: TableRowSecretProvenanceWrite
   /** Proof the caller asserted the insert lock (see `mutation-locks.ts`). */
   proof: MutationProof<'insert'>
+  /**
+   * Runs first in the transaction, before the row-order lock: the caller's unique-value locks and
+   * unique check (see `unique-locks.ts`), so the check sees any concurrent insert of the same value.
+   */
+  assertUnique?: (trx: DbTransaction) => Promise<void>
 }): Promise<{
   id: string
   data: RowData
@@ -354,6 +359,7 @@ export async function insertOrderedRow(params: {
   } = params
   const [row] = await db.transaction(async (trx) => {
     await setTableTxTimeouts(trx)
+    await params.assertUnique?.(trx)
     await acquireRowOrderLock(trx, tableId)
 
     // Resolve the authoritative order key from neighbor ids when given, else from the requested