From a1a384ee0e073df1f9d784234d3c64b4a13632fc Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Thu, 8 Oct 2026 13:34:48 -0700 Subject: [PATCH 1/2] 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. --- apps/sim/lib/table/dispatcher.integration.ts | 94 ++++ apps/sim/lib/table/dispatcher.ts | 29 +- apps/sim/lib/table/import-data.ts | 22 +- apps/sim/lib/table/lock-order.test.ts | 100 ----- apps/sim/lib/table/order-key.ts | 25 ++ apps/sim/lib/table/rows/ordering.ts | 78 ++-- .../lib/table/rows/row-writes.integration.ts | 412 +++++++++++++----- apps/sim/lib/table/rows/service.ts | 61 +-- apps/sim/lib/table/rows/unique-locks.ts | 5 +- apps/sim/lib/table/service.ts | 3 +- apps/sim/lib/table/update-row.test.ts | 123 +----- apps/sim/lib/table/validation.ts | 5 +- 12 files changed, 515 insertions(+), 442 deletions(-) create mode 100644 apps/sim/lib/table/dispatcher.integration.ts delete mode 100644 apps/sim/lib/table/lock-order.test.ts diff --git a/apps/sim/lib/table/dispatcher.integration.ts b/apps/sim/lib/table/dispatcher.integration.ts new file mode 100644 index 00000000000..68616d39978 --- /dev/null +++ b/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()) + }) +}) diff --git a/apps/sim/lib/table/dispatcher.ts b/apps/sim/lib/table/dispatcher.ts index 2d812ff8b94..804725d82e9 100644 --- a/apps/sim/lib/table/dispatcher.ts +++ b/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 (query: (executor: DbExecutor) => PromiseLike): Promise => + 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 diff --git a/apps/sim/lib/table/import-data.ts b/apps/sim/lib/table/import-data.ts index ee70fb34465..e8b52936d39 100644 --- a/apps/sim/lib/table/import-data.ts +++ b/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( diff --git a/apps/sim/lib/table/lock-order.test.ts b/apps/sim/lib/table/lock-order.test.ts deleted file mode 100644 index 5191779da93..00000000000 --- a/apps/sim/lib/table/lock-order.test.ts +++ /dev/null @@ -1,100 +0,0 @@ -/** - * Lock-order regression guard: a column-creating CSV import must acquire the - * per-table row-order advisory lock (`user_table_rows_pos`) BEFORE writing - * `user_table_definitions`, matching the rows_pos → definitions order that plain - * inserts take via the row-count trigger. The opposite order deadlocks - * concurrent inserts on the same table. - */ -import { userTableDefinitions } from '@sim/db/schema' -import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing' -import { beforeEach, describe, expect, it, vi } from 'vitest' -import { importAppendRows } from '@/lib/table/import-data' -import type { TableDefinition } from '@/lib/table/types' - -vi.mock('@/lib/table/validation', () => ({ - validateRowSize: vi.fn(() => ({ valid: true, errors: [] })), - validateRowAgainstSchema: vi.fn(() => ({ valid: true, errors: [] })), - coerceRowToSchema: vi.fn(() => ({ valid: true, errors: [] })), - coerceRowValues: vi.fn((row) => row), - 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: [] })), -})) - -const TABLE: TableDefinition = { - id: 'tbl-1', - name: 'People', - description: null, - schema: { columns: [{ name: 'name', type: 'string' }] }, - metadata: null, - rowCount: 0, - maxRows: 1000, - workspaceId: 'ws-1', - createdBy: 'user-1', - locks: { schemaLocked: false, insertLocked: false, updateLocked: false, deleteLocked: false }, - archivedAt: null, - createdAt: new Date('2024-01-01'), - updatedAt: new Date('2024-01-01'), -} - -/** - * invocationCallOrder of the first `tx.execute(...)` whose SQL contains - * `substring` — checks the template strings, the interpolated values (the - * advisory key is passed as a value), and `sql.raw` output. - */ -function executeOrderContaining(substring: string): number { - const { calls, invocationCallOrder } = dbChainMockFns.execute.mock - for (let i = 0; i < calls.length; i++) { - const arg = calls[i][0] as { strings?: unknown; values?: unknown; rawSql?: unknown } | undefined - const haystacks: string[] = [] - if (Array.isArray(arg?.strings)) { - haystacks.push( - ...(arg.strings as unknown[]).filter((s): s is string => typeof s === 'string') - ) - } - if (Array.isArray(arg?.values)) { - haystacks.push(...(arg.values as unknown[]).filter((v): v is string => typeof v === 'string')) - } - if (typeof arg?.rawSql === 'string') haystacks.push(arg.rawSql) - if (haystacks.some((s) => s.includes(substring))) return invocationCallOrder[i] - } - return -1 -} - -/** invocationCallOrder of the first `tx.update(table)` call. */ -function updateOrderForTable(table: unknown): number { - const { calls, invocationCallOrder } = dbChainMockFns.update.mock - for (let i = 0; i < calls.length; i++) { - if (calls[i][0] === table) return invocationCallOrder[i] - } - return -1 -} - -describe('table import lock ordering', () => { - beforeEach(() => { - resetDbChainMock() - queueTableRows(userTableDefinitions, [TABLE]) - }) - - it('acquires the rows_pos advisory before the user_table_definitions write when adding columns', async () => { - // The lock pre-acquire and the column-creating write both run at the top of - // the import, before the batch-insert loop. The loop's order-key aggregates - // aren't fully wired in this unit mock, so tolerate a downstream error after - // the locks under test have been recorded. - await importAppendRows( - TABLE, - [{ name: 'new_col', type: 'string' }], - [{ name: 'Alice', new_col: 'x' }], - { workspaceId: 'ws-1', userId: 'user-1', requestId: 'req-1' } - ).catch(() => {}) - - const rowsPosLockOrder = executeOrderContaining('user_table_rows_pos') - const definitionsWriteOrder = updateOrderForTable(userTableDefinitions) - - expect(rowsPosLockOrder).toBeGreaterThan(0) - expect(definitionsWriteOrder).toBeGreaterThan(0) - expect(rowsPosLockOrder).toBeLessThan(definitionsWriteOrder) - }) -}) diff --git a/apps/sim/lib/table/order-key.ts b/apps/sim/lib/table/order-key.ts index b6f8304bde2..aa29119996a 100644 --- a/apps/sim/lib/table/order-key.ts +++ b/apps/sim/lib/table/order-key.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) +} diff --git a/apps/sim/lib/table/rows/ordering.ts b/apps/sim/lib/table/rows/ordering.ts index 06ea5bc198e..267343d5eb8 100644 --- a/apps/sim/lib/table/rows/ordering.ts +++ b/apps/sim/lib/table/rows/ordering.ts @@ -14,7 +14,7 @@ import { acquireAdvisoryXactLock } from '@/lib/db/advisory-locks' import type { DbOrTx } from '@/lib/db/types' import { getDeleteSnapshotBatchSize, TABLE_LIMITS } from '@/lib/table/constants' import type { MutationProof } from '@/lib/table/mutation-locks' -import { keyBetween, nKeysBetween } from '@/lib/table/order-key' +import { appendKeys, keyBetween } from '@/lib/table/order-key' import { type DbExecutor, type DbTransaction, withSeqscanOff } from '@/lib/table/planner' import { TableRowNotFoundError } from '@/lib/table/rows/errors' import { @@ -149,18 +149,10 @@ export async function nextImportStartOrderKey(tableId: string): Promise { const [{ maxPos }] = await trx .select({ @@ -174,8 +166,7 @@ export async function nextRowPosition(trx: DbTransaction, tableId: string): Prom /** * The append anchors — `max(order_key)` and the next free `position` — in ONE round trip. * - * An append needs both, and asking separately is two serial round trips inside the row-order - * advisory lock, which every other inserting request is waiting on. Postgres plans each `max()` + * An append needs both, and asking separately is two serial round trips. Postgres plans each `max()` * as its own InitPlan, so the combined statement still serves each aggregate from its own index * (`(table_id, order_key, id)` and `(table_id, position)`) with an index-only backward scan — * exactly the two plans the separate queries produced, in one statement rather than two. @@ -208,15 +199,16 @@ export async function maxOrderKey(executor: DbOrTx, tableId: string): Promise { + const [next] = await trx + .select({ orderKey: userTableRows.orderKey }) + .from(userTableRows) + .where(and(eq(userTableRows.tableId, tableId), gt(userTableRows.orderKey, key))) + .orderBy(asc(userTableRows.orderKey)) + .limit(1) + return next?.orderKey ?? null } /** * Resolves the `order_key` for an insert expressed by an anchor row id — * `afterRowId` (place directly after) or `beforeRowId` (directly before). Finds * the anchor and its adjacent key via the `(table_id, order_key, id)` index - * (O(1)) and mints a key between them. Caller holds the row-order lock. + * (O(1)) and mints a key between them, or an append key when the anchor is the last row. */ export async function resolveInsertByNeighbor( trx: DbTransaction, @@ -274,17 +282,8 @@ export async function resolveInsertByNeighbor( } if (afterRowId) { - // hi = the smallest key strictly GREATER than the anchor key. Comparing keys - // (not the `(order_key, id)` row tuple) skips past any sibling that shares the - // anchor's key, so `keyBetween` always gets strictly-ordered bounds and can't - // throw on a stray duplicate. Identical to the row tuple when keys are distinct. - const [next] = await trx - .select({ orderKey: userTableRows.orderKey }) - .from(userTableRows) - .where(and(eq(userTableRows.tableId, tableId), gt(userTableRows.orderKey, anchorKey))) - .orderBy(asc(userTableRows.orderKey)) - .limit(1) - return keyBetween(anchorKey, next?.orderKey ?? null) + const next = await keyAbove(trx, tableId, anchorKey) + return next === null ? appendKeys(anchorKey, 1)[0] : keyBetween(anchorKey, next) } // beforeRowId: lo = the largest key strictly LESS than the anchor key (distinct, @@ -302,14 +301,14 @@ export async function resolveInsertByNeighbor( * Computes fractional `order_key`s for a batch insert by appending a contiguous * run after the current max key. `order_key` is authoritative, so callers needing * exact placement pass explicit `orderKeys` (handled before this function); here - * we just append a run. Caller holds the lock. + * we just append a run (see {@link appendKeys}). */ export async function resolveBatchInsertOrderKeys( trx: DbTransaction, tableId: string, count: number ): Promise { - return nKeysBetween(await maxOrderKey(trx, tableId), null, count) + return appendKeys(await maxOrderKey(trx, tableId), count) } /** @@ -333,8 +332,7 @@ export async function insertOrderedRow(params: { /** Proof the caller asserted the insert lock (see `mutation-locks.ts`). */ proof: MutationProof<'insert'> /** - * Opens the transaction in place of the default timeouts, before the row-order lock: the - * caller's schema guard (see `live-schema.ts`), which applies the timeouts, then its unique-value + * Opens the transaction in place of the default timeouts: the caller's schema guard (see `live-schema.ts`), which applies the timeouts, then its unique-value * locks and unique check (see `unique-locks.ts`), so the check sees any concurrent insert of the * same value. */ @@ -362,21 +360,19 @@ export async function insertOrderedRow(params: { const [row] = await db.transaction(async (trx) => { if (params.validate) await params.validate(trx) else await setTableTxTimeouts(trx) - await acquireRowOrderLock(trx, tableId) // Resolve the authoritative order key from neighbor ids when given, else from the requested // position. `order_key` is authoritative — `position` is a best-effort, no-shift companion. // // A plain append needs only the two table maxima, so it reads them together - // ({@link appendAnchors}) rather than paying a second round trip under the order lock. The - // anchored and positional forms resolve their key by walking to a slot, so they still ask - // for the next position separately. + // ({@link appendAnchors}). The anchored and positional forms resolve their key by walking to a + // slot, so they still ask for the next position separately. const appending = !afterRowId && !beforeRowId && position === undefined let orderKey: string let targetPosition: number if (appending) { const anchors = await appendAnchors(trx, tableId) - orderKey = keyBetween(anchors.maxOrderKey, null) + orderKey = appendKeys(anchors.maxOrderKey, 1)[0] targetPosition = anchors.nextPosition } else { orderKey = diff --git a/apps/sim/lib/table/rows/row-writes.integration.ts b/apps/sim/lib/table/rows/row-writes.integration.ts index fe72a597c96..975a8a08fb6 100644 --- a/apps/sim/lib/table/rows/row-writes.integration.ts +++ b/apps/sim/lib/table/rows/row-writes.integration.ts @@ -1,8 +1,7 @@ /** * Row-write integration tests against the provisioned, disposable TEST_DATABASE_URL database (the - * integration setup points DATABASE_URL at it too). The unique-value locks and the row-order lock - * run for real everywhere; the `rows_version` cases also need the migrated deferred trigger and - * skip without it. + * integration setup points DATABASE_URL at it too). The unique-value locks run for real + * everywhere; the `rows_version` cases also need the migrated deferred trigger and skip without it. */ import { db } from '@sim/db' import { userTableDefinitions, userTableRows } from '@sim/db/schema' @@ -24,11 +23,10 @@ vi.mock('@/lib/uploads/core/storage-service', () => storageServiceMock) import { deleteColumn, updateColumnConstraints } from '@/lib/table/columns/service' import { getMaxRowSizeBytes } from '@/lib/table/constants' -import { bulkInsertImportBatch, importReplaceRows } from '@/lib/table/import-data' +import { bulkInsertImportBatch, importAppendRows, importReplaceRows } from '@/lib/table/import-data' import type { DbTransaction } from '@/lib/table/planner' import { readCurrentRowsVersion } from '@/lib/table/row-changes' import { lockLiveTableSchema } from '@/lib/table/rows/live-schema' -import { acquireRowOrderLock } from '@/lib/table/rows/ordering' import { batchInsertRows, batchUpdateRows, @@ -121,6 +119,88 @@ async function untilSchemaLockWaiters(tableId: string, expected: number) { expect(waiting).toBe(expected) } +type LockWaiters = { onDefinitionRow: number; onValueLock: number } + +/** The session {@link raceUnderHeldDefinitionRow} holds the definition row in, while it does. */ +let definitionRowHolder: number | null = null + +/** + * Polls until exactly `expected.onDefinitionRow` sessions wait on the held definition row (directly, + * or queued behind an earlier waiter for it) and `expected.onValueLock` wait on one of the table's + * unique locks, and returns the last count seen. Both counts are scoped to the table, so concurrent + * suites cannot skew them: a unique-lock waiter waits on the table's unique lock itself, or on a + * value lock while holding the table's unique lock shared. + */ +async function waitForLockWaiters(tableId: string, expected: LockWaiters) { + const uniqueLockKey = `user_table_unique:${tableId}` + let waiting: LockWaiters = { onDefinitionRow: 0, onValueLock: 0 } + for (let attempt = 0; attempt < 400; attempt++) { + await sleep(5) + ;[waiting] = await control` + WITH RECURSIVE behind_holder(pid) AS ( + SELECT pid FROM pg_stat_activity + WHERE ${definitionRowHolder ?? 0}::int = ANY (pg_blocking_pids(pid)) + UNION + SELECT a.pid FROM pg_stat_activity a + JOIN behind_holder b ON b.pid = ANY (pg_blocking_pids(a.pid)) + ), lock AS ( + SELECT hashtextextended(${uniqueLockKey}, 0) AS unique_key + ), advisory AS ( + SELECT l.pid, l.granted, + l.classid = ((lock.unique_key >> 32) & 4294967295)::oid + AND l.objid = (lock.unique_key & 4294967295)::oid AS is_unique + FROM pg_locks l CROSS JOIN lock + WHERE l.locktype = 'advisory' + AND l.database = (SELECT oid FROM pg_database WHERE datname = current_database()) + ) + SELECT + (SELECT count(*) FROM pg_stat_activity a JOIN behind_holder USING (pid) + WHERE a.wait_event IN ('tuple', 'transactionid'))::int AS "onDefinitionRow", + count(*) FILTER ( + WHERE NOT granted AND ( + is_unique OR pid IN (SELECT pid FROM advisory WHERE granted AND is_unique) + ) + )::int AS "onValueLock" + FROM advisory` + if ( + waiting.onDefinitionRow === expected.onDefinitionRow && + waiting.onValueLock === expected.onValueLock + ) { + break + } + } + return waiting +} + +/** + * Holds the table's definition row `FOR UPDATE` while `writes` start, and releases it only once + * exactly `onDefinitionRow` of them wait on it and `onValueLock` wait on a unique-value lock. An + * insert's row write checks the row's foreign key to the definition, after its value locks and + * unique check, so this pins every write mid-flight; releasing any earlier could let them run + * one after another and prove nothing about the race. + */ +async function raceUnderHeldDefinitionRow( + tableId: string, + writes: Array<() => Promise>, + expected: LockWaiters +): Promise[]> { + const holder = await control.reserve() + try { + await holder`BEGIN` + const [{ pid }] = await holder<{ pid: number }[]>`SELECT pg_backend_pid() AS pid` + definitionRowHolder = pid + await holder`SELECT 1 FROM user_table_definitions WHERE id = ${tableId} FOR UPDATE` + const racers = Promise.allSettled(writes.map((write) => write())) + expect(await waitForLockWaiters(tableId, expected)).toEqual(expected) + await holder`COMMIT` + return await racers + } finally { + definitionRowHolder = null + await holder`ROLLBACK`.catch(() => {}) + holder.release() + } +} + describe('table row writes against real PostgreSQL', () => { beforeAll(async () => { await control`INSERT INTO "user" (id, name, email, email_verified, created_at, updated_at) @@ -158,79 +238,6 @@ describe('table row writes against real PostgreSQL', () => { 'unique-race' ) - type LockWaiters = { onOrderLock: number; onValueLock: number } - - /** - * Polls until exactly `expected.onOrderLock` sessions wait on the table's row-order lock and - * `expected.onValueLock` wait on one of its unique locks, and returns the last count seen. Both - * counts are scoped to the table, so concurrent suites cannot skew them: a unique-lock waiter - * waits on the table's unique lock itself, or on a value lock while holding the table's unique - * lock shared. - */ - async function waitForLockWaiters(tableId: string, expected: LockWaiters) { - const orderLockKey = `user_table_rows_pos:${tableId}` - const uniqueLockKey = `user_table_unique:${tableId}` - let waiting: LockWaiters = { onOrderLock: 0, onValueLock: 0 } - for (let attempt = 0; attempt < 400; attempt++) { - await sleep(5) - ;[waiting] = await control` - WITH lock AS ( - SELECT hashtextextended(${orderLockKey}, 0) AS order_key, - hashtextextended(${uniqueLockKey}, 0) AS unique_key - ), advisory AS ( - SELECT l.pid, l.granted, - l.classid = ((lock.order_key >> 32) & 4294967295)::oid - AND l.objid = (lock.order_key & 4294967295)::oid AS is_order, - l.classid = ((lock.unique_key >> 32) & 4294967295)::oid - AND l.objid = (lock.unique_key & 4294967295)::oid AS is_unique - FROM pg_locks l CROSS JOIN lock - WHERE l.locktype = 'advisory' - AND l.database = (SELECT oid FROM pg_database WHERE datname = current_database()) - ) - SELECT - count(*) FILTER (WHERE NOT granted AND is_order)::int AS "onOrderLock", - count(*) FILTER ( - WHERE NOT granted AND NOT is_order AND ( - is_unique OR pid IN (SELECT pid FROM advisory WHERE granted AND is_unique) - ) - )::int AS "onValueLock" - FROM advisory` - if ( - waiting.onOrderLock === expected.onOrderLock && - waiting.onValueLock === expected.onValueLock - ) { - break - } - } - return waiting - } - - /** - * Holds the table's row-order lock while `writes` start, and releases it only once exactly - * `onOrderLock` of them wait on it and `onValueLock` wait on a unique-value lock. Inserts take - * their value locks and run their unique check before the row-order lock, so this pins every - * write mid-flight; releasing any earlier could let them run one after another and prove - * nothing about the race. - */ - async function raceUnderHeldOrderLock( - tableId: string, - writes: Array<() => Promise>, - expected: LockWaiters - ): Promise[]> { - const holder = await control.reserve() - try { - await holder`BEGIN` - await holder`SELECT pg_advisory_xact_lock(hashtextextended(${`user_table_rows_pos:${tableId}`}, 0))` - const racers = Promise.allSettled(writes.map((write) => write())) - expect(await waitForLockWaiters(tableId, expected)).toEqual(expected) - await holder`COMMIT` - return await racers - } finally { - await holder`ROLLBACK`.catch(() => {}) - holder.release() - } - } - async function storedCount(tableId: string, match: Record) { const [{ count }] = await control<{ count: number }[]>`SELECT count(*)::int AS count FROM user_table_rows WHERE table_id = ${tableId} AND data @> ${control.json(match)}` @@ -243,13 +250,13 @@ describe('table row writes against real PostgreSQL', () => { it('rejects the second of two concurrent single-row inserts of the same value', async () => { const table = await createTable(uniqueColumns) - const results = await raceUnderHeldOrderLock( + const results = await raceUnderHeldDefinitionRow( table.id, [ () => insertEmail(table, 'dup@example.test'), () => insertEmail(table, 'dup@example.test'), ], - { onOrderLock: 1, onValueLock: 1 } + { onDefinitionRow: 1, onValueLock: 1 } ) expect(fulfilled(results)).toBe(1) @@ -271,10 +278,14 @@ describe('table row writes against real PostgreSQL', () => { 'unique-race' ) - const results = await raceUnderHeldOrderLock(table.id, [insertScore('8'), insertScore(8)], { - onOrderLock: 1, - onValueLock: 1, - }) + const results = await raceUnderHeldDefinitionRow( + table.id, + [insertScore('8'), insertScore(8)], + { + onDefinitionRow: 1, + onValueLock: 1, + } + ) expect(fulfilled(results)).toBe(1) expect(await storedCount(table.id, { score: 8 })).toBe(1) @@ -283,10 +294,10 @@ describe('table row writes against real PostgreSQL', () => { it('lets concurrent inserts of different values proceed together', async () => { const table = await createTable(uniqueColumns) - const results = await raceUnderHeldOrderLock( + const results = await raceUnderHeldDefinitionRow( table.id, [() => insertEmail(table, 'a@example.test'), () => insertEmail(table, 'b@example.test')], - { onOrderLock: 2, onValueLock: 0 } + { onDefinitionRow: 2, onValueLock: 0 } ) expect(fulfilled(results)).toBe(2) @@ -295,7 +306,7 @@ describe('table row writes against real PostgreSQL', () => { it('rejects a single insert racing a batch insert that holds the same value', async () => { const table = await createTable(uniqueColumns) - const results = await raceUnderHeldOrderLock( + const results = await raceUnderHeldDefinitionRow( table.id, [ () => @@ -312,7 +323,7 @@ describe('table row writes against real PostgreSQL', () => { ), () => insertEmail(table, 'dup@example.test'), ], - { onOrderLock: 1, onValueLock: 1 } + { onDefinitionRow: 1, onValueLock: 1 } ) expect(fulfilled(results)).toBe(1) @@ -325,14 +336,16 @@ describe('table row writes against real PostgreSQL', () => { { id: `${table.id}-a`, data: { email: 'a@example.test' }, orderKey: 'a0' }, ]) - const results = await raceUnderHeldOrderLock( + const results = await raceUnderHeldDefinitionRow( table.id, [ () => insertEmail(table, 'dup@example.test'), async () => { // Start the edit only once the insert has passed its check and holds its locks. - expect(await waitForLockWaiters(table.id, { onOrderLock: 1, onValueLock: 0 })).toEqual({ - onOrderLock: 1, + expect( + await waitForLockWaiters(table.id, { onDefinitionRow: 1, onValueLock: 0 }) + ).toEqual({ + onDefinitionRow: 1, onValueLock: 0, }) return updateRow( @@ -349,7 +362,7 @@ describe('table row writes against real PostgreSQL', () => { ) }, ], - { onOrderLock: 1, onValueLock: 1 } + { onDefinitionRow: 1, onValueLock: 1 } ) expect(fulfilled(results)).toBe(1) @@ -359,7 +372,7 @@ describe('table row writes against real PostgreSQL', () => { it('lets a replace that adds the first unique column wait out a writer on an older schema', async () => { const table = await createTable([{ id: 'name', name: 'name', type: 'string' }]) // The writer resolved the table while it still had a unique column, so it holds the unique - // lock shared and will want the row-order lock next. + // lock shared and will write its row next. const stale: TableDefinition = { ...table, schema: { columns: [...table.schema.columns, ...uniqueColumns] }, @@ -376,7 +389,13 @@ describe('table row writes against real PostgreSQL', () => { await lockUniqueValues(trx, stale, [{ email: 'stale@example.test' }]) holdsUniqueLock() await released - await acquireRowOrderLock(trx, table.id) + await trx.insert(userTableRows).values({ + id: `${table.id}-stale`, + tableId: table.id, + workspaceId, + data: { email: 'stale@example.test' }, + orderKey: 'a0', + }) }) await holding @@ -388,8 +407,8 @@ describe('table row writes against real PostgreSQL', () => { { rows: [{ name: 'fresh', code: 'c-1' }], workspaceId }, 'unique-race' ) - expect(await waitForLockWaiters(table.id, { onOrderLock: 0, onValueLock: 1 })).toEqual({ - onOrderLock: 0, + expect(await waitForLockWaiters(table.id, { onDefinitionRow: 0, onValueLock: 1 })).toEqual({ + onDefinitionRow: 0, onValueLock: 1, }) } finally { @@ -429,7 +448,7 @@ describe('table row writes against real PostgreSQL', () => { const rows = Array.from({ length: 80 }, (_, i) => ({ email: `bulk-${i}@example.test` })) rows.push({ email: 'dup@example.test' }) - const results = await raceUnderHeldOrderLock( + const results = await raceUnderHeldDefinitionRow( table.id, [ () => @@ -446,7 +465,7 @@ describe('table row writes against real PostgreSQL', () => { ), () => insertEmail(table, 'dup@example.test'), ], - { onOrderLock: 1, onValueLock: 1 } + { onDefinitionRow: 1, onValueLock: 1 } ) expect(fulfilled(results)).toBe(1) @@ -456,14 +475,16 @@ describe('table row writes against real PostgreSQL', () => { it('rejects an import batch racing a single insert of the same value', async () => { const table = await createTable(uniqueColumns) - const results = await raceUnderHeldOrderLock( + const results = await raceUnderHeldDefinitionRow( table.id, [ () => insertEmail(table, 'dup@example.test'), async () => { // Start the import only once the insert has passed its check and holds its locks. - expect(await waitForLockWaiters(table.id, { onOrderLock: 1, onValueLock: 0 })).toEqual({ - onOrderLock: 1, + expect( + await waitForLockWaiters(table.id, { onDefinitionRow: 1, onValueLock: 0 }) + ).toEqual({ + onDefinitionRow: 1, onValueLock: 0, }) return bulkInsertImportBatch( @@ -478,7 +499,7 @@ describe('table row writes against real PostgreSQL', () => { ) }, ], - { onOrderLock: 1, onValueLock: 1 } + { onDefinitionRow: 1, onValueLock: 1 } ) expect(fulfilled(results)).toBe(1) @@ -649,20 +670,20 @@ describe('table row writes against real PostgreSQL', () => { it('locks JSON values one by one, equating objects whose keys differ only in order', async () => { const table = await createTable(jsonColumns) - const different = await raceUnderHeldOrderLock( + const different = await raceUnderHeldDefinitionRow( table.id, [() => insertMeta(table, { a: 1 }), () => insertMeta(table, { a: 1, b: 2 })], - { onOrderLock: 2, onValueLock: 0 } + { onDefinitionRow: 2, onValueLock: 0 } ) expect(fulfilled(different)).toBe(2) - const reordered = await raceUnderHeldOrderLock( + const reordered = await raceUnderHeldDefinitionRow( table.id, [ () => insertMeta(table, { x: 1, y: [1, 2] }), () => insertMeta(table, { y: [1, 2], x: 1 }), ], - { onOrderLock: 1, onValueLock: 1 } + { onDefinitionRow: 1, onValueLock: 1 } ) expect(fulfilled(reordered)).toBe(1) expect(await storedCount(table.id, { meta: { x: 1 } })).toBe(1) @@ -1545,6 +1566,185 @@ describe('table row writes against real PostgreSQL', () => { }) }) + describe('inserts without a table-wide lock', () => { + const insertName = (table: TableDefinition, name: string, placement = {}) => + insertRow( + { + tableId: table.id, + workspaceId, + data: { name }, + secretProvenance: undefined, + capabilityGovernedUserId: null, + ...placement, + }, + table, + 'lockless-insert' + ) + const batchInsertNames = (table: TableDefinition, names: string[]) => + batchInsertRows( + { + tableId: table.id, + workspaceId, + rows: names.map((name) => ({ name })), + secretProvenance: undefined, + capabilityGovernedUserId: null, + }, + table, + 'lockless-batch' + ) + const orderedNames = async (tableId: string) => + ( + await control<{ name: string; order_key: string }[]>` + SELECT data->>'name' AS name, order_key FROM user_table_rows + WHERE table_id = ${tableId} ORDER BY order_key, id` + ).map((row) => row.name) + + it('lets every insert path commit while another transaction holds the retired row-order lock', async () => { + const table = await createTable([ + { id: 'name', name: 'name', type: 'string' }, + { id: 'key', name: 'key', type: 'string', unique: true }, + ]) + await seedRows(table.id, [{ id: `${table.id}-a`, data: { name: 'a' }, orderKey: 'a0' }]) + const writes: Array<[string, () => Promise]> = [ + ['append', () => insertName(table, 'b')], + ['at a position', () => insertName(table, 'c', { position: 0 })], + ['after a row', () => insertName(table, 'd', { afterRowId: `${table.id}-a` })], + ['batch', () => batchInsertNames(table, ['e', 'f'])], + [ + 'upsert', + () => + upsertRow( + { + tableId: table.id, + workspaceId, + data: { name: 'g', key: 'g' }, + conflictTarget: 'key', + secretProvenance: undefined, + capabilityGovernedUserId: null, + }, + table, + 'lockless-upsert' + ), + ], + [ + 'import with a new column', + () => + importAppendRows( + table, + [{ name: 'extra', type: 'string' }], + [{ name: 'h', extra: 'x' }], + { workspaceId, requestId: 'lockless-import', capabilityGovernedUserId: null } + ), + ], + ] + + const holder = await control.reserve() + const elapsedMs: Record = {} + try { + await holder`BEGIN` + await holder`SELECT pg_advisory_xact_lock(hashtextextended(${`user_table_rows_pos:${table.id}`}, 0))` + for (const [name, write] of writes) { + const started = Date.now() + await write() + elapsedMs[name] = Date.now() - started + } + } finally { + await holder`ROLLBACK`.catch(() => {}) + holder.release() + } + + for (const [name] of writes) expect(elapsedMs[name], name).toBeLessThan(2_000) + expect((await orderedNames(table.id)).length).toBe(8) + }) + + it('gives appends that read the same last key distinct keys, keeping each batch together', async () => { + const table = await createTable(textColumns('name')) + await seedRows(table.id, [{ id: `${table.id}-a`, data: { name: 'seed' }, orderKey: 'a0' }]) + const batches = Array.from({ length: 4 }, (_, b) => + Array.from({ length: 5 }, (_, i) => `batch${b}-${i}`) + ) + const singles = Array.from({ length: 4 }, (_, i) => `single${i}`) + + // Every writer reads the table's last key before its row write waits on the held definition + // row, so all of them mint from the same `a0`. + const results = await raceUnderHeldDefinitionRow( + table.id, + [ + ...batches.map((names) => () => batchInsertNames(table, names)), + ...singles.map((name) => () => insertName(table, name)), + ], + { onDefinitionRow: batches.length + singles.length, onValueLock: 0 } + ) + + expect(results.filter((result) => result.status === 'rejected')).toEqual([]) + const [{ keys, distinct }] = await control<{ keys: number; distinct: number }[]>` + SELECT count(*)::int AS keys, count(DISTINCT order_key)::int AS distinct + FROM user_table_rows WHERE table_id = ${table.id}` + expect(distinct).toBe(keys) + const names = await orderedNames(table.id) + expect(names[0]).toBe('seed') + for (const batch of batches) { + const at = names.indexOf(batch[0]) + expect(names.slice(at, at + batch.length)).toEqual(batch) + } + }) + + it('inserts at a position between two rows that share a key', async () => { + const table = await createTable(textColumns('name')) + await seedRows(table.id, [ + { id: `${table.id}-0`, data: { name: 'a' }, orderKey: 'a0' }, + { id: `${table.id}-1`, data: { name: 'b' }, orderKey: 'a1' }, + { id: `${table.id}-2`, data: { name: 'c' }, orderKey: 'a1' }, + { id: `${table.id}-3`, data: { name: 'd' }, orderKey: 'a2' }, + ]) + + await insertName(table, 'new', { position: 2 }) + + const names = await orderedNames(table.id) + expect(names.indexOf('new')).toBe(3) + expect(names.at(-1)).toBe('d') + }) + + it('appends a row whose requested position is past the last row', async () => { + const table = await createTable(textColumns('name')) + await seedRows(table.id, [ + { id: `${table.id}-0`, data: { name: 'a' }, orderKey: 'a0' }, + { id: `${table.id}-1`, data: { name: 'b' }, orderKey: 'a1' }, + ]) + + await insertName(table, 'new', { position: 10 }) + + expect(await orderedNames(table.id)).toEqual(['a', 'b', 'new']) + }) + + it('leaves only the later row set when two replaces run concurrently', async () => { + const table = await createTable(textColumns('name')) + await seedRows(table.id, [{ id: `${table.id}-old`, data: { name: 'old' }, orderKey: 'a0' }]) + const replaceWith = (names: string[]) => + replaceTableRows( + { + tableId: table.id, + workspaceId, + rows: names.map((name) => ({ name })), + secretProvenance: undefined, + }, + table, + 'concurrent-replace' + ) + + const results = await Promise.allSettled([ + replaceWith(['x1', 'x2']), + replaceWith(['y1', 'y2']), + ]) + + expect(results.map((result) => result.status)).toEqual(['fulfilled', 'fulfilled']) + expect([ + ['x1', 'x2'], + ['y1', 'y2'], + ]).toContainEqual(await orderedNames(table.id)) + }) + }) + describe.skipIf(!migrated)('rows_version', () => { it('advances once for a transaction that edits cells across several statements', async () => { const table = await createTable(textColumns('name')) diff --git a/apps/sim/lib/table/rows/service.ts b/apps/sim/lib/table/rows/service.ts index 51bb937de90..c17ce40edd2 100644 --- a/apps/sim/lib/table/rows/service.ts +++ b/apps/sim/lib/table/rows/service.ts @@ -57,7 +57,6 @@ import { refitRowToSchema, } from '@/lib/table/rows/live-schema' import { - acquireRowOrderLock, type DeletedTableRow, deleteOrderedRow, deleteOrderedRowsByIds, @@ -336,8 +335,8 @@ export interface TxRowWriteOptions extends RowWriteOptions { export interface BatchInsertOptions extends TxRowWriteOptions { /** - * The caller already holds `lockUniqueColumns` for the table in this transaction (bulk imports, - * which take it before the row-order lock), so the batch takes no per-value locks of its own. + * The caller already holds `lockUniqueColumns` for the table in this transaction (bulk imports), + * so the batch takes no per-value locks of its own. */ uniqueColumnsLocked?: boolean } @@ -352,9 +351,8 @@ export interface BatchInsertOptions extends TxRowWriteOptions { * 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. * - * Takes the rows' unique-value locks (see `unique-locks.ts`) before the unique check and the - * row-order lock, so a caller must not already hold the row-order lock unless it passes - * `uniqueColumnsLocked`. Validates against `table` unless it passes `lockSchema`, so a caller that + * Takes the rows' unique-value locks (see `unique-locks.ts`) before the unique check unless the + * caller passes `uniqueColumnsLocked`. Validates against `table` unless it passes `lockSchema`, so a caller that * does not reads `table` under the table's schema lock in this transaction. */ export async function batchInsertRowsWithTx( @@ -422,7 +420,6 @@ export async function batchInsertRowsWithTx( ...(data.userId ? { createdBy: data.userId } : {}), }) - await acquireRowOrderLock(trx, data.tableId) // Undo restore passes exact saved keys; otherwise append after the current max. const orderKeys = data.orderKeys && data.orderKeys.length > 0 @@ -548,10 +545,9 @@ export async function replaceTableRows( * Capacity is NOT checked here (it would mean a billing-pool read inside the tx). * Callers gate it before opening the tx — see `replaceTableRows` and `importReplaceRows`. * - * Takes the table's unique lock before its row-order lock, so a caller already holding the - * row-order lock must take `lockUniqueColumns` first. Validates against `table` unless it passes - * `lockSchema`, so a caller that does not reads `table` under the table's schema lock in this - * transaction. + * Takes the table's unique lock exclusively, which also serializes concurrent replaces. Validates + * against `table` unless it passes `lockSchema`, so a caller that does not reads `table` under the + * table's schema lock in this transaction. */ export async function replaceTableRowsWithTx( trx: DbTransaction, @@ -632,16 +628,12 @@ export async function replaceTableRowsWithTx( if (!options.lockSchema) await setTableTxTimeouts(trx, { statementMs }) // Every current row is about to go, so any concurrent write of a unique value conflicts with the - // replacement set: hold the unique columns exclusively, ahead of the row-order lock. + // replacement set: hold the unique columns exclusively. The lock is taken even without unique + // columns, which also serializes concurrent replaces: otherwise the second's DELETE would not see + // the first's rows, and the table would end up with both row sets. A plain insert that commits + // during a replace survives it, the outcome of running right after it. await lockUniqueColumns(trx, table) - // Serialize concurrent replaces (and concurrent auto-position inserts) on the - // same table. Without this, two concurrent replaces each see their own MVCC - // snapshot for the DELETE; the second's DELETE would not observe rows the - // first inserted, so both transactions commit and the table ends up with - // the union of both row sets instead of only the last caller's rows. - await acquireRowOrderLock(trx, data.tableId) - const deleteCountRows = await trx.execute<{ count: number | string }>(sql` WITH deleted AS ( DELETE FROM ${userTableRows} @@ -810,7 +802,7 @@ export async function upsertRow( const snapshotMatchFilter = upsertConflictProbe(target, data.data) // Resolve the plan limit BEFORE the tx (the lookup is a separate pool read; doing - // it inside the tx would hold a connection + the row-order lock during it). The + // it inside the tx would hold a connection + the value lock during it). The // insert branch enforces it; the update path doesn't add a row, so it's exempt. const rowLimit = await getMaxRowsPerTable(table.workspaceId) @@ -843,7 +835,8 @@ export async function upsertRow( ) } // Holds every unique value this row writes, the conflict target included, before the lookup - // and the unique check below: concurrent upserts and inserts of the same value serialize here. + // and the unique check below: concurrent upserts and inserts of the same value serialize here, + // so the lookup already sees a racing upsert's committed row. await lockUniqueValues(trx, live, [data.data]) // Find existing row by single conflict target column @@ -876,30 +869,8 @@ export async function upsertRow( const now = new Date() - // Resolve which row (if any) we should update. If the initial SELECT missed, - // acquire the lock and re-check — a concurrent upsert may have inserted the - // matching row between our SELECT and the INSERT path; without the re-check - // both transactions would insert and bypass the app-level unique check. - let matchedRowId = existingRow?.id - let previousData = existingRow?.data as RowData | undefined - if (!matchedRowId) { - await acquireRowOrderLock(trx, data.tableId) - const [racedRow] = await trx - .select({ id: userTableRows.id, data: userTableRows.data }) - .from(userTableRows) - .where( - and( - eq(userTableRows.tableId, data.tableId), - eq(userTableRows.workspaceId, data.workspaceId), - matchFilter - ) - ) - .limit(1) - if (racedRow) { - matchedRowId = racedRow.id - previousData = racedRow.data as RowData - } - } + const matchedRowId = existingRow?.id + const previousData = existingRow?.data as RowData | undefined if (matchedRowId) { assertRowUpdate(table, patchColumnIds(data.data)) diff --git a/apps/sim/lib/table/rows/unique-locks.ts b/apps/sim/lib/table/rows/unique-locks.ts index e5b2ee8b40e..4f8e2e93d70 100644 --- a/apps/sim/lib/table/rows/unique-locks.ts +++ b/apps/sim/lib/table/rows/unique-locks.ts @@ -9,7 +9,7 @@ * * Lock order, everywhere: the table's schema lock (shared by row writes, exclusive by schema * changes; see `live-schema.ts`), then the table's unique lock and its value locks in sorted order, - * in one statement, then the table's row-order lock, then the definition row. + * in one statement, then the definition row. */ import { compareStrings } from '@sim/utils/string' @@ -83,7 +83,8 @@ export async function lockUniqueValues( * Locks every unique value of `table` exclusively, for writers that replace the table's rows * wholesale and so conflict with any concurrent write of a unique value. It locks even a table with * no unique columns yet: a writer holding an older schema may still hold the lock shared, and a - * whole-table writer that adds a unique column must already hold it before the row-order lock. + * whole-table writer that adds a unique column must already hold it before it writes rows. Taking + * it exclusively also serializes two replaces of the same table. */ export async function lockUniqueColumns(trx: DbTransaction, table: TableDefinition): Promise { await acquireAdvisoryXactLocks(trx, UNIQUE_LOCK_TAG, [ diff --git a/apps/sim/lib/table/service.ts b/apps/sim/lib/table/service.ts index 811b7816d8a..7f216babd45 100644 --- a/apps/sim/lib/table/service.ts +++ b/apps/sim/lib/table/service.ts @@ -124,8 +124,7 @@ function readLocks(row: { * * Uses an advisory lock (not `SELECT ... FOR UPDATE` on the definition row) so * it adds no edges to the row-lock graph — a FOR UPDATE here would also block the - * foreign-key check (KEY SHARE) of every row write to the table. Mirrors - * `acquireRowOrderLock`. The lock and + * foreign-key check (KEY SHARE) of every row write to the table. The lock and * the read both release at COMMIT/ROLLBACK; the wait is bounded by the * `statement_timeout` set in `setTableTxTimeouts`. */ diff --git a/apps/sim/lib/table/update-row.test.ts b/apps/sim/lib/table/update-row.test.ts index 27ca42f3bbc..62fb6e51ff3 100644 --- a/apps/sim/lib/table/update-row.test.ts +++ b/apps/sim/lib/table/update-row.test.ts @@ -3,13 +3,7 @@ import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing' import { tableBillingMock, tableBillingMockFns } from '@sim/testing/mocks/table-billing.mock' import { tableRowsLiveSchemaMock } from '@sim/testing/mocks/table-rows-live-schema.mock' import { beforeEach, describe, expect, it, vi } from 'vitest' -import { - batchUpdateRows, - insertRow, - replaceTableRows, - updateRow, - upsertRow, -} from '@/lib/table/rows/service' +import { batchUpdateRows, replaceTableRows, updateRow, upsertRow } from '@/lib/table/rows/service' import type { TableDefinition } from '@/lib/table/types' import { checkUniqueConstraintsDb, getUniqueColumns } from '@/lib/table/validation' @@ -169,66 +163,13 @@ describe('updateRow — partial merge', () => { }) }) -describe('insertRow — position race safety (migration 0198 + advisory lock)', () => { +describe('upsertRow — conflict target', () => { beforeEach(() => { vi.resetAllMocks() resetDbChainMock() vi.mocked(getUniqueColumns).mockReturnValue([]) }) - it('auto-position inserts acquire the per-table advisory lock before reading max(position)', async () => { - await expect( - insertRow({ tableId: 'tbl-1', data: { name: 'a' }, workspaceId: 'ws-1' }, TABLE, 'req-1') - ).rejects.toBeDefined() - - expect(findExecutedSqlContaining('pg_advisory_xact_lock')).toBe(true) - expect(findExecutedSqlContaining('hashtextextended')).toBe(true) - }) - - it('upsertRow skips the row-order lock on the update path (match found)', async () => { - vi.mocked(getUniqueColumns).mockReturnValue([{ name: 'name', type: 'string', unique: true }]) - dbChainMockFns.limit.mockResolvedValueOnce([ - { - id: 'row-1', - tableId: 'tbl-1', - workspaceId: 'ws-1', - data: { name: 'Alice', age: 30 }, - position: 0, - createdAt: new Date(), - updatedAt: new Date(), - }, - ]) - dbChainMockFns.returning.mockResolvedValueOnce([ - { - id: 'row-1', - tableId: 'tbl-1', - workspaceId: 'ws-1', - data: { name: 'Alice', age: 31 }, - position: 0, - createdAt: new Date(), - updatedAt: new Date(), - }, - ]) - - await upsertRow( - { - tableId: 'tbl-1', - workspaceId: 'ws-1', - data: { name: 'Alice', age: 31 }, - conflictTarget: 'name', - }, - TABLE, - 'req-1' - ) - - expect(findExecutedSqlContaining('SELECT pg_advisory_xact_lock(')).toBe(false) - }) - - /** - * The v2 surface is column-NAME-keyed and resolves `conflictTarget` to its - * storage id before this call, so the rejection has to translate back — a - * caller that sent `email` cannot act on a `col_…` id it has never seen. - */ it('upsertRow names the conflict column the caller does, not its storage id', async () => { const table: TableDefinition = { ...TABLE, @@ -256,66 +197,6 @@ describe('insertRow — position race safety (migration 0198 + advisory lock)', ) ).rejects.toThrow('Column "email" is not a unique column. Available unique columns: slug') }) - - it('upsertRow acquires the advisory lock on the insert path (no match)', async () => { - vi.mocked(getUniqueColumns).mockReturnValue([{ name: 'name', type: 'string', unique: true }]) - // Initial existing-row check + post-lock re-check both find no match. - dbChainMockFns.limit.mockResolvedValueOnce([]) - dbChainMockFns.limit.mockResolvedValueOnce([]) - - await expect( - upsertRow( - { - tableId: 'tbl-1', - workspaceId: 'ws-1', - data: { name: 'Bob', age: 25 }, - conflictTarget: 'name', - }, - TABLE, - 'req-1' - ) - ).rejects.toBeDefined() - - expect(findExecutedSqlContaining('pg_advisory_xact_lock')).toBe(true) - }) - - it('upsertRow re-checks after acquiring the lock and switches to UPDATE when a racing tx inserted the row', async () => { - vi.mocked(getUniqueColumns).mockReturnValue([{ name: 'name', type: 'string', unique: true }]) - // Initial existing-row check: no match (another tx has not committed yet). - dbChainMockFns.limit.mockResolvedValueOnce([]) - // Post-lock re-check: a racing tx just inserted the row. - const racedRow = { - id: 'row-raced', - tableId: 'tbl-1', - workspaceId: 'ws-1', - data: { name: 'Bob', age: 25 }, - position: 0, - createdAt: new Date(), - updatedAt: new Date(), - } - dbChainMockFns.limit.mockResolvedValueOnce([racedRow]) - // UPDATE returning the patched row. - dbChainMockFns.returning.mockResolvedValueOnce([ - { ...racedRow, data: { name: 'Bob', age: 26 } }, - ]) - - const result = await upsertRow( - { - tableId: 'tbl-1', - workspaceId: 'ws-1', - data: { name: 'Bob', age: 26 }, - conflictTarget: 'name', - }, - TABLE, - 'req-1' - ) - - expect(findExecutedSqlContaining('pg_advisory_xact_lock')).toBe(true) - expect(result.operation).toBe('update') - expect(result.row.id).toBe('row-raced') - expect(dbChainMockFns.update).toHaveBeenCalled() - expect(dbChainMockFns.insert).not.toHaveBeenCalled() - }) }) describe('mutation paths — SET LOCAL timeouts', () => { diff --git a/apps/sim/lib/table/validation.ts b/apps/sim/lib/table/validation.ts index ac8b5a3bd5d..42a8e084ba0 100644 --- a/apps/sim/lib/table/validation.ts +++ b/apps/sim/lib/table/validation.ts @@ -515,9 +515,8 @@ export async function checkUniqueConstraintsDb( /** * Runs `check` on the caller's open transaction with seq scans penalized, then puts back the - * transaction's previous setting. Unique checks now run inside write transactions ahead of the - * row-order lock, and leaving the flag off for the rest of the transaction slowed the locked - * position and INSERT statements that follow. Reading the previous value and setting the new one + * transaction's previous setting. Unique checks run inside write transactions, and leaving the + * flag off for the rest of the transaction slowed the position and INSERT statements that follow. Reading the previous value and setting the new one * in one statement keeps it one round trip each way; a caller that set the flag itself keeps it. */ async function withSeqscanOffOn( From f334574c943ff754f71a24d61bb2615f45819efa Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Thu, 8 Oct 2026 13:45:31 -0700 Subject: [PATCH 2/2] test(tables): order rows bytewise in the lockless-insert tests and stub the job queue locally --- apps/sim/lib/table/dispatcher.integration.ts | 9 ++++++--- apps/sim/lib/table/rows/row-writes.integration.ts | 7 ++++--- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/apps/sim/lib/table/dispatcher.integration.ts b/apps/sim/lib/table/dispatcher.integration.ts index 68616d39978..04f17305ebe 100644 --- a/apps/sim/lib/table/dispatcher.integration.ts +++ b/apps/sim/lib/table/dispatcher.integration.ts @@ -6,7 +6,6 @@ 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, @@ -16,7 +15,11 @@ 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) +const { mockBatchEnqueueAndWait } = vi.hoisted(() => ({ mockBatchEnqueueAndWait: vi.fn() })) + +vi.mock('@/lib/core/async-jobs/config', () => ({ + getJobQueue: async () => ({ batchEnqueueAndWait: mockBatchEnqueueAndWait }), +})) vi.mock('@/lib/table/events', () => tableEventsMock) vi.mock('@/lib/table/workflow-columns', () => tableWorkflowColumnsMock) @@ -72,7 +75,7 @@ describe('table run dispatcher against real PostgreSQL', () => { runs.map((payload) => ({ payload })) ) const dispatched: string[] = [] - asyncJobsMockFns.mockJobQueue.batchEnqueueAndWait.mockImplementation( + mockBatchEnqueueAndWait.mockImplementation( async (_kind: string, items: Array<{ payload: { rowId: string } }>) => { for (const item of items) dispatched.push(item.payload.rowId) } diff --git a/apps/sim/lib/table/rows/row-writes.integration.ts b/apps/sim/lib/table/rows/row-writes.integration.ts index 975a8a08fb6..4ebc908c01a 100644 --- a/apps/sim/lib/table/rows/row-writes.integration.ts +++ b/apps/sim/lib/table/rows/row-writes.integration.ts @@ -1592,11 +1592,12 @@ describe('table row writes against real PostgreSQL', () => { table, 'lockless-batch' ) + // Byte order, as migration 0228 sets on the column: `db:push` leaves the default collation. const orderedNames = async (tableId: string) => ( - await control<{ name: string; order_key: string }[]>` - SELECT data->>'name' AS name, order_key FROM user_table_rows - WHERE table_id = ${tableId} ORDER BY order_key, id` + await control<{ name: string }[]>` + SELECT data->>'name' AS name FROM user_table_rows + WHERE table_id = ${tableId} ORDER BY order_key COLLATE "C", id` ).map((row) => row.name) it('lets every insert path commit while another transaction holds the retired row-order lock', async () => {