refactor(tables): wrapper owns row-write transactions; trx no longer crosses the boundary

Author: TheodoreSpeaks

apps/sim/lib/table/__tests__/table-wrapper.test.ts

@@ -39,8 +39,9 @@ vi.mock('@/lib/table/sql', () => ({
 
 import {
   buildOrderedRowValues,
-  closeGapAfterDelete,
-  compactAfterDelete,
+  deleteOrderedRow,
+  deleteOrderedRowsByIds,
+  insertOrderedRow,
   queryRows,
   reserveBatchPositions,
   reserveInsertPosition,
@@ -123,23 +124,66 @@ describe('table-wrapper', () => {
     })
   })
 
-  describe('closeGapAfterDelete', () => {
-    it('shifts trailing rows down with a single UPDATE', async () => {
-      await closeGapAfterDelete(trx, 'tbl-1', 3)
+  describe('insertOrderedRow', () => {
+    it('reserves an append position and inserts in one transaction', async () => {
+      dbChainMockFns.where.mockResolvedValueOnce([{ maxPos: 2 }]) // nextRowPosition
+      dbChainMockFns.returning.mockResolvedValueOnce([
+        { id: 'row-1', data: { a: 1 }, position: 3, createdAt: new Date(), updatedAt: new Date() },
+      ])
+      const row = await insertOrderedRow({
+        tableId: 'tbl-1',
+        workspaceId: 'ws-1',
+        data: { a: 1 },
+        rowId: 'row-1',
+        now: new Date('2024-01-01'),
+      })
+      expect(row.id).toBe('row-1')
+      expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(1)
+      expect(dbChainMockFns.insert).toHaveBeenCalledTimes(1)
+    })
+  })
+
+  describe('deleteOrderedRow', () => {
+    it('returns false when no row matches', async () => {
+      dbChainMockFns.returning.mockResolvedValueOnce([])
+      expect(await deleteOrderedRow({ tableId: 'tbl-1', rowId: 'x', workspaceId: 'ws-1' })).toBe(
+        false
+      )
+      expect(dbChainMockFns.update).not.toHaveBeenCalled()
+    })
+
+    it('deletes and closes the gap when a row matches', async () => {
+      dbChainMockFns.returning.mockResolvedValueOnce([{ position: 4 }])
+      expect(await deleteOrderedRow({ tableId: 'tbl-1', rowId: 'r', workspaceId: 'ws-1' })).toBe(
+        true
+      )
+      // closeGapAfterDelete → one shift-down UPDATE.
       expect(dbChainMockFns.update).toHaveBeenCalledTimes(1)
-      expect(dbChainMockFns.set).toHaveBeenCalledTimes(1)
     })
   })
 
-  describe('compactAfterDelete', () => {
-    it('runs a recompaction UPDATE (scoped form)', async () => {
-      await compactAfterDelete(trx, 'tbl-1', 4)
-      expect(trx.execute).toHaveBeenCalledTimes(1)
+  describe('deleteOrderedRowsByIds', () => {
+    it('no-ops on an empty id list', async () => {
+      expect(
+        await deleteOrderedRowsByIds({ tableId: 'tbl-1', workspaceId: 'ws-1', rowIds: [] })
+      ).toEqual([])
+      expect(dbChainMockFns.transaction).not.toHaveBeenCalled()
     })
 
-    it('runs a recompaction UPDATE (full form)', async () => {
-      await compactAfterDelete(trx, 'tbl-1')
-      expect(trx.execute).toHaveBeenCalledTimes(1)
+    it('deletes in a transaction and recompacts when rows were removed', async () => {
+      dbChainMockFns.returning.mockResolvedValueOnce([
+        { id: 'a', position: 1 },
+        { id: 'b', position: 3 },
+      ])
+      const deleted = await deleteOrderedRowsByIds({
+        tableId: 'tbl-1',
+        workspaceId: 'ws-1',
+        rowIds: ['a', 'b'],
+      })
+      expect(deleted).toHaveLength(2)
+      expect(dbChainMockFns.transaction).toHaveBeenCalledTimes(1)
+      // Recompaction runs as a raw UPDATE via trx.execute.
+      expect(trx.execute).toHaveBeenCalled()
     })
   })
 

apps/sim/lib/table/service.ts

@@ -28,15 +28,17 @@ import { CSV_MAX_BATCH_SIZE } from './import'
 import { buildFilterClause } from './sql'
 import {
   acquireRowOrderLock,
-  closeGapAfterDelete,
-  compactAfterDelete,
+  deleteOrderedRow,
+  deleteOrderedRowsByIds,
+  insertOrderedRow,
   loadExecutionsByRow,
   loadExecutionsForRow,
   queryRows as queryRowsWrapper,
   reserveBatchPositions,
   reserveInsertPosition,
 } from './table-wrapper'
 import { fireTableTrigger } from './trigger'
+import { type DbTransaction, setTableTxTimeouts } from './tx'
 import type {
   AddWorkflowGroupData,
   BatchInsertData,
@@ -100,29 +102,6 @@ export class TableConflictError extends Error {
 
 export type TableScope = 'active' | 'archived' | 'all'
 
-type DbTransaction = Parameters<Parameters<typeof db.transaction>[0]>[0]
-
-/**
- * Sets per-transaction Postgres timeouts via `SET LOCAL`.
- *
- * `lock_timeout` is the critical one: without it, a waiter inherits the full
- * `statement_timeout` clock, so one stuck writer can drain the pool.
- *
- * Safe under pgBouncer transaction pooling — `SET LOCAL` is transaction-scoped
- * and cleared at COMMIT/ROLLBACK before the session returns to the pool.
- */
-async function setTableTxTimeouts(
-  trx: DbTransaction,
-  opts?: { statementMs?: number; lockMs?: number; idleMs?: number }
-) {
-  const s = opts?.statementMs ?? 10_000
-  const l = opts?.lockMs ?? 3_000
-  const i = opts?.idleMs ?? 5_000
-  await trx.execute(sql.raw(`SET LOCAL statement_timeout = '${s}ms'`))
-  await trx.execute(sql.raw(`SET LOCAL lock_timeout = '${l}ms'`))
-  await trx.execute(sql.raw(`SET LOCAL idle_in_transaction_session_timeout = '${i}ms'`))
-}
-
 /**
  * Serializes schema/metadata read-modify-writes for a single table so
  * concurrent mutators can't clobber each other's `schema` JSONB
@@ -996,24 +975,14 @@ export async function insertRow(
   // (migration 0198): a single conditional UPDATE on user_table_definitions
   // increments row_count iff row_count < max_rows, taking the row lock
   // atomically. No app-level FOR UPDATE / COUNT needed.
-  const [row] = await db.transaction(async (trx) => {
-    await setTableTxTimeouts(trx)
-
-    const targetPosition = await reserveInsertPosition(trx, data.tableId, data.position)
-
-    return trx
-      .insert(userTableRows)
-      .values({
-        id: rowId,
-        tableId: data.tableId,
-        workspaceId: data.workspaceId,
-        data: data.data,
-        position: targetPosition,
-        createdAt: now,
-        updatedAt: now,
-        ...(data.userId ? { createdBy: data.userId } : {}),
-      })
-      .returning()
+  const row = await insertOrderedRow({
+    tableId: data.tableId,
+    workspaceId: data.workspaceId,
+    data: data.data,
+    rowId,
+    position: data.position,
+    createdBy: data.userId,
+    now,
   })
 
   logger.info(`[${requestId}] Inserted row ${rowId} into table ${data.tableId}`)
@@ -2307,23 +2276,8 @@ export async function deleteRow(
   workspaceId: string,
   requestId: string
 ): Promise<void> {
-  await db.transaction(async (trx) => {
-    await setTableTxTimeouts(trx)
-    const [deleted] = await trx
-      .delete(userTableRows)
-      .where(
-        and(
-          eq(userTableRows.id, rowId),
-          eq(userTableRows.tableId, tableId),
-          eq(userTableRows.workspaceId, workspaceId)
-        )
-      )
-      .returning({ position: userTableRows.position })
-
-    if (!deleted) throw new Error('Row not found')
-
-    await closeGapAfterDelete(trx, tableId, deleted.position)
-  })
+  const deleted = await deleteOrderedRow({ tableId, rowId, workspaceId })
+  if (!deleted) throw new Error('Row not found')
 
   logger.info(`[${requestId}] Deleted row ${rowId} from table ${tableId}`)
 }
@@ -2707,28 +2661,11 @@ export async function deleteRowsByFilter(
   }
 
   const rowIds = matchingRows.map((r) => r.id)
-  const minDeletedPos = matchingRows.reduce(
-    (min, r) => (r.position < min ? r.position : min),
-    matchingRows[0].position
-  )
-
-  await db.transaction(async (trx) => {
-    await setTableTxTimeouts(trx, { statementMs: 60_000 })
-    for (let i = 0; i < rowIds.length; i += TABLE_LIMITS.DELETE_BATCH_SIZE) {
-      const batch = rowIds.slice(i, i + TABLE_LIMITS.DELETE_BATCH_SIZE)
-      await trx.delete(userTableRows).where(
-        and(
-          eq(userTableRows.tableId, table.id),
-          eq(userTableRows.workspaceId, table.workspaceId),
-          sql`${userTableRows.id} = ANY(ARRAY[${sql.join(
-            batch.map((id) => sql`${id}`),
-            sql`, `
-          )}])`
-        )
-      )
-    }
 
-    await compactAfterDelete(trx, table.id, minDeletedPos)
+  await deleteOrderedRowsByIds({
+    tableId: table.id,
+    workspaceId: table.workspaceId,
+    rowIds,
   })
 
   logger.info(`[${requestId}] Deleted ${matchingRows.length} rows from table ${table.id}`)
@@ -2752,36 +2689,10 @@ export async function deleteRowsByIds(
 ): Promise<BulkDeleteByIdsResult> {
   const uniqueRequestedRowIds = Array.from(new Set(data.rowIds))
 
-  const deletedRows = await db.transaction(async (trx) => {
-    await setTableTxTimeouts(trx, { statementMs: 60_000 })
-    const deleted: { id: string; position: number }[] = []
-    for (let i = 0; i < uniqueRequestedRowIds.length; i += TABLE_LIMITS.DELETE_BATCH_SIZE) {
-      const batch = uniqueRequestedRowIds.slice(i, i + TABLE_LIMITS.DELETE_BATCH_SIZE)
-      const rows = await trx
-        .delete(userTableRows)
-        .where(
-          and(
-            eq(userTableRows.tableId, data.tableId),
-            eq(userTableRows.workspaceId, data.workspaceId),
-            sql`${userTableRows.id} = ANY(ARRAY[${sql.join(
-              batch.map((id) => sql`${id}`),
-              sql`, `
-            )}])`
-          )
-        )
-        .returning({ id: userTableRows.id, position: userTableRows.position })
-      deleted.push(...rows)
-    }
-
-    if (deleted.length > 0) {
-      const minDeletedPos = deleted.reduce(
-        (min, r) => (r.position < min ? r.position : min),
-        deleted[0].position
-      )
-      await compactAfterDelete(trx, data.tableId, minDeletedPos)
-    }
-
-    return deleted
+  const deletedRows = await deleteOrderedRowsByIds({
+    tableId: data.tableId,
+    workspaceId: data.workspaceId,
+    rowIds: uniqueRequestedRowIds,
   })
 
   const deletedIds = deletedRows.map((r) => r.id)