improvement(tables): cut the DB round trips a table read and write spend on protocol (#8104)

* improvement(tables): cut the DB round trips a table read and write spend on protocol

The grid's first page spent more time on round trips than on work. Four of them
were avoidable:

- `pendingDeleteMask` probed `table_jobs` on every read, though the table a
  request just loaded already carries its latest non-export job, and the
  `one_active_per_table` unique index makes that row the running delete when one
  exists. Callers that hold a table across a long walk (the export stream, the
  snapshot builder) keep probing per page, so a delete starting mid-walk still
  begins masking.
- The run-state sidecar was read for every table, including the ones that
  declare no workflow group and therefore cannot have a row — four chunked
  queries on a 1000-row page, all returning nothing.
- The drain opened a transaction per batch. The guards are fixed for the call,
  so each extra batch paid `BEGIN` + `set_config` + `COMMIT` for nothing.
- `setTableTxTimeouts` issued three `SET LOCAL` statements; `set_config(…, true)`
  is the same thing and fits in one round trip, as the read guards already do.

A 1000-row page goes from 22 statements to 14, a 50-row page from 15 to 13, and
every write transaction drops two.

* review(tables): drop the delete-mask elision and correct the provenance snapshot doc

Reading the delete job from the table a request already loaded widened a race
the mask probe has always had — a job committing between the check and the row
read is missed either way, but trusting the loaded fields moves the check two
queries earlier. Closing it properly means evaluating the job inside the row
read's own snapshot, which is a larger change than this one, so the elision is
removed and `pending-delete-mask.ts` is back to what it was.

The three remaining reductions are untouched: they were the bulk of the win,
and each is a read this code cannot need rather than a read it takes on faith.

Also updates `TableRowProvenanceReader`'s doc, which still described one
repeatable-read transaction per batch.

Author: waleedlatif1

apps/sim/lib/table/__tests__/service-filter-threading.test.ts

@@ -50,6 +50,7 @@ vi.mock('@/lib/table/rows/executions', () => ({
   })),
   loadExecutionsByRow: mockLoadExecutionsByRow,
   loadExecutionsForRow: vi.fn(async () => ({})),
+  tableMayHaveRunState: vi.fn(() => true),
   writeExecutionsPatch: vi.fn(async () => 'wrote'),
 }))
 
@@ -65,6 +66,7 @@ vi.mock('@/lib/table/validation', () => ({
   checkBatchUniqueConstraintsDb: vi.fn(async () => ({ valid: true, errors: [] })),
 }))
 
+import { tableMayHaveRunState } from '@/lib/table/rows/executions'
 import {
   deleteRow,
   deleteRowsByFilter,
@@ -666,3 +668,40 @@ describe('queryRows byte budget', () => {
     })
   })
 })
+
+/**
+ * The run-state sidecar read the row path used to make unconditionally, one round trip (four on a
+ * full page) for tables that cannot hold a single sidecar row. This pins that the query is
+ * actually skipped rather than merely ignored, and that the fallback still runs.
+ */
+describe('queryRows run-state elision', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+    vi.mocked(tableMayHaveRunState).mockReturnValue(true)
+  })
+
+  it('skips the run-state read for a table that can hold none, still reporting empty executions', async () => {
+    vi.mocked(tableMayHaveRunState).mockReturnValue(false)
+    dbChainMockFns.limit.mockResolvedValueOnce([])
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      { id: 'row-1', data: {}, position: 0, orderKey: 'a0', createdAt: null, updatedAt: null },
+    ])
+
+    const result = await queryRows(TABLE, { limit: 5, includeTotal: false }, 'req-1')
+
+    expect(mockLoadExecutionsByRow).not.toHaveBeenCalled()
+    expect(result.rows[0].executions).toEqual({})
+  })
+
+  it('reads run state for a table that can hold it', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([])
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      { id: 'row-1', data: {}, position: 0, orderKey: 'a0', createdAt: null, updatedAt: null },
+    ])
+
+    await queryRows(TABLE, { limit: 5, includeTotal: false }, 'req-1')
+
+    expect(mockLoadExecutionsByRow).toHaveBeenCalledTimes(1)
+  })
+})

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

@@ -68,6 +68,31 @@ function findExecutedRawSql(substring: string): string | undefined {
   return undefined
 }
 
+/** Every string reachable from a drizzle `sql` fragment — its literal chunks AND its bound values. */
+function collectStrings(node: unknown, out: string[] = []): string[] {
+  if (typeof node === 'string') out.push(node)
+  else if (Array.isArray(node)) for (const entry of node) collectStrings(entry, out)
+  else if (node && typeof node === 'object')
+    for (const entry of Object.values(node as Record<string, unknown>)) collectStrings(entry, out)
+  return out
+}
+
+/**
+ * Whether one `set_config(<setting>, '<value>', true)` guard was executed.
+ *
+ * The guards bind their values as parameters, so the setting name lives in the statement's
+ * literal chunks while the duration lives in its bound values — asserting on a rendered string
+ * would only re-check the placeholder.
+ */
+function executedTxTimeout(setting: string, value: string): boolean {
+  return dbChainMockFns.execute.mock.calls.some(([arg]) => {
+    const strings = collectStrings(arg)
+    return (
+      strings.some((entry) => entry.includes(`set_config('${setting}'`)) && strings.includes(value)
+    )
+  })
+}
+
 /**
  * The `data` payload of the last `.set(...)` row write. `updateRow` always writes a JSONB merge
  * (`data = data || {changed}::jsonb`), so this is a `sql` fragment exposing `{ strings, values }`.
@@ -420,11 +445,9 @@ describe('mutation paths — SET LOCAL timeouts', () => {
       insertRow({ tableId: 'tbl-1', data: { name: 'a' }, workspaceId: 'ws-1' }, TABLE, 'req-1')
     ).rejects.toBeDefined()
 
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '10000ms'")).toBeDefined()
-    expect(findExecutedRawSql("SET LOCAL lock_timeout = '3000ms'")).toBeDefined()
-    expect(
-      findExecutedRawSql("SET LOCAL idle_in_transaction_session_timeout = '5000ms'")
-    ).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '10000ms')).toBe(true)
+    expect(executedTxTimeout('lock_timeout', '3000ms')).toBe(true)
+    expect(executedTxTimeout('idle_in_transaction_session_timeout', '5000ms')).toBe(true)
   })
 
   it('batchInsertRows raises statement_timeout to 60s', async () => {
@@ -436,7 +459,7 @@ describe('mutation paths — SET LOCAL timeouts', () => {
       )
     ).rejects.toBeDefined()
 
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '60000ms'")).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '60000ms')).toBe(true)
   })
 
   it('replaceTableRows scales statement_timeout with (existing + new) row count', async () => {
@@ -450,7 +473,7 @@ describe('mutation paths — SET LOCAL timeouts', () => {
     )
 
     // (100_000 + 50_000) × 3ms/row = 450_000ms; above 120_000 floor, below 600_000 cap
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '450000ms'")).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '450000ms')).toBe(true)
   })
 
   it('replaceTableRows caps scaled timeout at 10 minutes for very large tables', async () => {
@@ -459,7 +482,7 @@ describe('mutation paths — SET LOCAL timeouts', () => {
     await replaceTableRows({ tableId: 'tbl-1', workspaceId: 'ws-1', rows: [] }, hugeTable, 'req-1')
 
     // 10M × 3ms = 30M ms, capped at 600_000ms (10 min)
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '600000ms'")).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '600000ms')).toBe(true)
   })
 
   it('replaceTableRows uses the 120s floor on small tables', async () => {
@@ -472,7 +495,7 @@ describe('mutation paths — SET LOCAL timeouts', () => {
     )
 
     // 12 × 3ms = 36ms → floored at 120_000ms
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '120000ms'")).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '120000ms')).toBe(true)
   })
 
   it('renameColumn is metadata-only — no per-row JSONB rewrite regardless of row count', async () => {
@@ -491,7 +514,7 @@ describe('mutation paths — SET LOCAL timeouts', () => {
     await deleteColumn({ tableId: 'tbl-1', columnName: 'age' }, 'req-1')
 
     // 100 × 2ms = 200ms → floored at 60_000ms
-    expect(findExecutedRawSql("SET LOCAL statement_timeout = '60000ms'")).toBeDefined()
+    expect(executedTxTimeout('statement_timeout', '60000ms')).toBe(true)
   })
 
   it('replaceTableRows acquires the per-table advisory lock to serialize concurrent replaces', async () => {

apps/sim/lib/table/rows/executions.test.ts

@@ -4,8 +4,12 @@
 import { dbChainMock, dbChainMockFns, resetDbChainMock } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import type { DbOrTx } from '@/lib/db/types'
-import { loadExecutionsByRow, writeExecutionsPatch } from '@/lib/table/rows/executions'
-import type { RowExecutionMetadata } from '@/lib/table/types'
+import {
+  loadExecutionsByRow,
+  tableMayHaveRunState,
+  writeExecutionsPatch,
+} from '@/lib/table/rows/executions'
+import type { RowExecutionMetadata, TableSchema } from '@/lib/table/types'
 
 const EXECUTION_STATE: RowExecutionMetadata = {
   status: 'running',
@@ -258,3 +262,39 @@ describe('loadExecutionsByRow', () => {
     expect(byRow.size).toBe(750)
   })
 })
+
+describe('tableMayHaveRunState', () => {
+  const column = (overrides: Partial<TableSchema['columns'][number]> = {}) => ({
+    id: 'col_1',
+    name: 'title',
+    type: 'string' as const,
+    ...overrides,
+  })
+
+  it('is false for a schema that declares no group', () => {
+    expect(tableMayHaveRunState({ columns: [column()] })).toBe(false)
+    expect(tableMayHaveRunState({ columns: [column()], workflowGroups: [] })).toBe(false)
+  })
+
+  it('is true once the schema declares a group', () => {
+    expect(
+      tableMayHaveRunState({
+        columns: [column()],
+        workflowGroups: [{ id: 'group-1' }] as TableSchema['workflowGroups'],
+      })
+    ).toBe(true)
+  })
+
+  /**
+   * A column still pointing at a group is group state whatever the group list says, so an
+   * unexpected schema shape must keep the sidecar read rather than silently drop run state.
+   */
+  it('is true for a column that still names a group the list has lost', () => {
+    expect(
+      tableMayHaveRunState({
+        columns: [column({ workflowGroupId: 'group-1' })],
+        workflowGroups: [],
+      })
+    ).toBe(true)
+  })
+})

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

@@ -40,6 +40,24 @@ interface LoadExecutionsOptions {
   budgetBytes?: number
 }
 
+/**
+ * Whether a table can have any run-state sidecar at all.
+ *
+ * `tableRowExecutions` is keyed by `(rowId, groupId)`, and every writer takes its `groupId` from
+ * a group on the table's own schema. Group and column deletes strip the matching sidecar rows in
+ * the same transaction that removes the group ({@link stripGroupExecutions}), so a schema that
+ * declares no group cannot have a surviving row — the sidecar read would return nothing, and the
+ * caller would fill in the same empty map it gets by skipping.
+ *
+ * Both signals are checked rather than just `workflowGroups`: a column still carrying a
+ * `workflowGroupId` means the table has group state whatever the group list looks like, so an
+ * unexpected schema shape keeps the query instead of silently dropping run state.
+ */
+export function tableMayHaveRunState(schema: TableSchema): boolean {
+  if (schema.workflowGroups && schema.workflowGroups.length > 0) return true
+  return schema.columns.some((column) => column.workflowGroupId !== undefined)
+}
+
 /**
  * Loads `tableRowExecutions` rows for the given row ids and groups them into a
  * `Map<rowId, RowExecutions>` suitable for plugging into `TableRow.executions`.

apps/sim/lib/table/rows/secret-provenance.ts

@@ -959,8 +959,10 @@ export async function loadTableRowSecretProvenance(
 
 /**
  * Collects only returned row values while their database snapshot is still valid.
- * Readers use one repeatable-read transaction per bounded batch; writers capture
- * after stamping and before releasing row locks. Nothing is reloaded after commit.
+ * A read captures inside one repeatable-read transaction spanning every batch of
+ * its page, so a row and the sidecar captured for it always come from the same
+ * snapshot; writers capture after stamping and before releasing row locks.
+ * Nothing is reloaded after commit.
  */
 export class TableRowProvenanceReader {
   private readonly accumulator: ResolvedSecretTraceProvenanceAccumulator