fix(tables): drop the per-table row-order lock from inserts

Appends mint keys in a random slot after the last key, so concurrent appends
never share a key and a batch stays contiguous. Positions are left best-effort
and may repeat; the run dispatcher finishes a tied position before advancing
its cursor. Positional inserts skip tied keys. Concurrent replaces stay
serialized by the table's unique lock, which every replace already takes.

Author: TheodoreSpeaks

apps/sim/lib/table/dispatcher.integration.ts

@@ -0,0 +1,94 @@
+/**
+ * The run dispatcher's row windows against real PostgreSQL: rows that share a `position` (two
+ * inserts that assigned it concurrently) are each dispatched exactly once, wherever the window
+ * boundary falls.
+ */
+import { db } from '@sim/db'
+import { userTableDefinitions, userTableRows } from '@sim/db/schema'
+import { readTestDatabaseUrl } from '@sim/db/testing/test-infrastructure'
+import { asyncJobsMock, asyncJobsMockFns } from '@sim/testing/mocks/async-jobs.mock'
+import { tableEventsMock } from '@sim/testing/mocks/table-events.mock'
+import {
+  tableWorkflowColumnsMock,
+  tableWorkflowColumnsMockFns,
+} from '@sim/testing/mocks/table-workflow-columns.mock'
+import { generateId } from '@sim/utils/id'
+import postgres from 'postgres'
+import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
+
+vi.mock('@/lib/core/async-jobs/config', () => asyncJobsMock)
+vi.mock('@/lib/table/events', () => tableEventsMock)
+vi.mock('@/lib/table/workflow-columns', () => tableWorkflowColumnsMock)
+
+import { insertDispatch, runDispatcherToCompletion } from '@/lib/table/dispatcher'
+
+const url = readTestDatabaseUrl()
+if (process.env.DATABASE_URL !== url) {
+  throw new Error('This suite requires only the disposable local test database')
+}
+const control = postgres(url, { max: 2, onnotice: () => {} })
+const workspaceId = generateId()
+const userId = generateId()
+
+describe('table run dispatcher against real PostgreSQL', () => {
+  beforeAll(async () => {
+    await control`INSERT INTO "user" (id, name, email, email_verified, created_at, updated_at)
+      VALUES (${userId}, 'Dispatcher fixture', ${`${userId}@example.test`}, true, now(), now())`
+    await control`INSERT INTO workspace (id, name, owner_id, billed_account_user_id)
+      VALUES (${workspaceId}, 'Dispatcher fixtures', ${userId}, ${userId})`
+  })
+
+  afterAll(async () => {
+    await control`DELETE FROM workspace WHERE id = ${workspaceId}`
+    await control`DELETE FROM "user" WHERE id = ${userId}`
+    await control.end()
+  })
+
+  it('dispatches every row once when a window boundary splits rows that share a position', async () => {
+    const tableId = generateId()
+    await db.insert(userTableDefinitions).values({
+      id: tableId,
+      workspaceId,
+      name: tableId,
+      schema: { columns: [], workflowGroups: [{ id: 'group-1' }] },
+      createdBy: userId,
+    })
+    const positions = [0, 1, 1, 1, 2, 3, 3, 4]
+    await db.insert(userTableRows).values(
+      positions.map((position, i) => ({
+        id: `${tableId}-${i}`,
+        tableId,
+        workspaceId,
+        data: {},
+        position,
+        orderKey: `a${i}`,
+      }))
+    )
+    tableWorkflowColumnsMockFns.mockBuildPendingRuns.mockImplementation(
+      (_table: unknown, rows: Array<{ id: string }>) =>
+        rows.map((row) => ({ rowId: row.id, groupId: 'group-1', tableId, workspaceId }))
+    )
+    tableWorkflowColumnsMockFns.mockBuildEnqueueItems.mockImplementation(async (runs: unknown[]) =>
+      runs.map((payload) => ({ payload }))
+    )
+    const dispatched: string[] = []
+    asyncJobsMockFns.mockJobQueue.batchEnqueueAndWait.mockImplementation(
+      async (_kind: string, items: Array<{ payload: { rowId: string } }>) => {
+        for (const item of items) dispatched.push(item.payload.rowId)
+      }
+    )
+
+    const dispatchId = await insertDispatch({
+      tableId,
+      workspaceId,
+      requestId: 'dispatcher-ties',
+      mode: 'all',
+      scope: { groupIds: ['group-1'] },
+      isManualRun: true,
+      capabilityGovernedUserId: null,
+    })
+    await runDispatcherToCompletion(dispatchId, 2)
+
+    expect([...dispatched].sort()).toEqual(positions.map((_, i) => `${tableId}-${i}`).sort())
+  })
+})

apps/sim/lib/table/dispatcher.ts

@@ -556,13 +556,34 @@ export async function dispatcherStep(
       .select()
       .from(userTableRows)
       .where(and(...filters))
-      .orderBy(asc(userTableRows.position))
+      .orderBy(asc(userTableRows.position), asc(userTableRows.id))
       .limit(windowSize)
   // Filtered scopes carry a jsonb predicate the planner can't estimate — left alone it
   // seq-scans the whole shared relation per window; keep it on the tenant's position index.
-  const chunk = hasJsonbFilter
-    ? await withSeqscanOff(async (trx) => windowQuery(trx))
-    : await windowQuery(db)
+  const runQuery = async <T>(query: (executor: DbExecutor) => PromiseLike<T>): Promise<T> =>
+    hasJsonbFilter ? withSeqscanOff(async (trx) => query(trx)) : query(db)
+  const chunk = await runQuery(windowQuery)
+
+  // Inserts assign `position` without a lock, so concurrent ones can share it. The next window
+  // starts strictly after this one's last position, so a full window takes the rest of that tie.
+  if (chunk.length === windowSize) {
+    const lastPosition = chunk[chunk.length - 1].position
+    const seen = chunk.filter((r) => r.position === lastPosition).map((r) => r.id)
+    const ties = await runQuery((executor) =>
+      executor
+        .select()
+        .from(userTableRows)
+        .where(
+          and(
+            ...filters,
+            eq(userTableRows.position, lastPosition),
+            notInArray(userTableRows.id, seen)
+          )
+        )
+        .orderBy(asc(userTableRows.id))
+    )
+    chunk.push(...ties)
+  }
 
   if (chunk.length === 0) {
     // Through the shared, guarded completion like the other two exits: this

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

@@ -16,11 +16,7 @@ import { assertRowDelete, assertRowInsert, assertSchemaMutable } from '@/lib/tab
 import { nKeysBetween } from '@/lib/table/order-key'
 import type { DbTransaction } from '@/lib/table/planner'
 import { lockLiveTableSchema, refitRowToSchema, withLiveSchema } from '@/lib/table/rows/live-schema'
-import {
-  acquireRowOrderLock,
-  guardBatch,
-  type MutationRevalidator,
-} from '@/lib/table/rows/ordering'
+import { guardBatch, type MutationRevalidator } from '@/lib/table/rows/ordering'
 import {
   createExactEmptyTableRowSecretProvenance,
   mutateTableRowsWithSecretProvenance,
@@ -60,7 +56,7 @@ 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; the caller threads each batch's
+ * contiguous order keys (no max-key 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
@@ -251,9 +247,6 @@ export async function setTableSchemaForImport(
  * parsed must be visible to the asserts in `addTableColumnsWithTx` /
  * `batchInsertRowsWithTx` / `replaceTableRowsWithTx`, which all read the
  * definition they are handed.
- *
- * Taken before `acquireRowOrderLock` so the order stays advisory → rows_pos →
- * definitions, matching every other advisory-lock holder.
  */
 async function refreshUnderLock(
   trx: DbTransaction,
@@ -297,16 +290,10 @@ 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.
+    // Lock the unique columns whole: 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
-      // columns first would lock the definition row before rows_pos, inverting
-      // the order and deadlocking concurrent inserts on this table. The lock is
-      // re-entrant, so the per-batch acquire below is a no-op.
-      await acquireRowOrderLock(trx, table.id)
       working = await addTableColumnsWithTx(trx, working, additions, ctx.requestId)
     }
     const inserted: TableRow[] = []
@@ -368,7 +355,6 @@ export async function importReplaceRows(
     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)
     }
     return replaceTableRowsWithTx(

apps/sim/lib/table/lock-order.test.ts

@@ -12,6 +12,7 @@
  */
 
 import { generateKeyBetween, generateNKeysBetween } from '@sim/utils/fractional-indexing'
+import { generateRandomBytes } from '@sim/utils/random'
 
 /**
  * Returns a key that sorts strictly between `a` and `b`. Pass `null` for an open
@@ -32,3 +33,27 @@ export function keyBetween(a: string | null, b: string | null): string {
 export function nKeysBetween(a: string | null, b: string | null, n: number): string[] {
   return generateNKeysBetween(a, b, n)
 }
+
+/** Random halvings {@link appendKeys} takes: two concurrent appends collide with odds 2^-32. */
+const APPEND_SLOT_BITS = 32
+
+/**
+ * Returns `n` ordered keys after `last`, in a slot no concurrent append to the same table picks.
+ *
+ * Appends take no lock, so two writers can read the same `last`. Each mints inside the integer
+ * range `keyBetween(last, null)` opens, narrowed by {@link APPEND_SLOT_BITS} random halvings to a
+ * private sub-range: keys never collide, and a batch stays contiguous instead of interleaving with
+ * another writer's. The next append reads the new max and moves to the next integer, so keys do
+ * not grow across appends — each carries a fixed few extra characters.
+ */
+export function appendKeys(last: string | null, n: number): string[] {
+  let lo = keyBetween(last, null)
+  let hi = keyBetween(lo, null)
+  const bits = generateRandomBytes(APPEND_SLOT_BITS / 8)
+  for (let i = 0; i < APPEND_SLOT_BITS; i++) {
+    const mid = keyBetween(lo, hi)
+    if ((bits[i >> 3] >> (i & 7)) & 1) lo = mid
+    else hi = mid
+  }
+  return nKeysBetween(lo, hi, n)
+}