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

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