diff --git a/apps/sim/lib/table/__tests__/service-filter-threading.test.ts b/apps/sim/lib/table/__tests__/service-filter-threading.test.ts index a7089a000f4..d17890020e0 100644 --- a/apps/sim/lib/table/__tests__/service-filter-threading.test.ts +++ b/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) + }) +}) diff --git a/apps/sim/lib/table/__tests__/update-row.test.ts b/apps/sim/lib/table/__tests__/update-row.test.ts index 8370f7b192d..8a43ec0e093 100644 --- a/apps/sim/lib/table/__tests__/update-row.test.ts +++ b/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)) collectStrings(entry, out) + return out +} + +/** + * Whether one `set_config(, '', 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 () => { diff --git a/apps/sim/lib/table/rows/executions.test.ts b/apps/sim/lib/table/rows/executions.test.ts index 01ad6b1c56d..e0c38cbf429 100644 --- a/apps/sim/lib/table/rows/executions.test.ts +++ b/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 = {}) => ({ + 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) + }) +}) diff --git a/apps/sim/lib/table/rows/executions.ts b/apps/sim/lib/table/rows/executions.ts index 3ed355573c6..935a39605aa 100644 --- a/apps/sim/lib/table/rows/executions.ts +++ b/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` suitable for plugging into `TableRow.executions`. diff --git a/apps/sim/lib/table/rows/secret-provenance.ts b/apps/sim/lib/table/rows/secret-provenance.ts index 2dc90cf0bdf..b36583d0546 100644 --- a/apps/sim/lib/table/rows/secret-provenance.ts +++ b/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 diff --git a/apps/sim/lib/table/rows/service.ts b/apps/sim/lib/table/rows/service.ts index 1bbd2e96418..b6cfb9b7ccb 100644 --- a/apps/sim/lib/table/rows/service.ts +++ b/apps/sim/lib/table/rows/service.ts @@ -47,6 +47,7 @@ import { deriveExecClearsForDataPatch, loadExecutionsByRow, loadExecutionsForRow, + tableMayHaveRunState, writeExecutionsPatch, } from '@/lib/table/rows/executions' import { @@ -1250,13 +1251,14 @@ export async function queryRows( * route does not expose — where it previously rendered. Callers that publish * the ceiling pass it; callers that do not keep the unbounded read they had. */ - const executionsByRow = withExecutions - ? await loadExecutionsByRow( - db, - rows.map((r) => r.id), - runStateBudgetBytes === undefined ? undefined : { budgetBytes: runStateBudgetBytes } - ) - : null + const executionsByRow = + withExecutions && tableMayHaveRunState(table.schema) + ? await loadExecutionsByRow( + db, + rows.map((r) => r.id), + runStateBudgetBytes === undefined ? undefined : { budgetBytes: runStateBudgetBytes } + ) + : null logger.info( `[${requestId}] Queried ${rows.length} rows from table ${table.id} (total: ${totalCount}, bytes: ${fetched.bytes}, more: ${fetched.hasMore})` @@ -1419,7 +1421,12 @@ export async function fetchRowsBounded(params: BoundedFetchParams): Promise { + const runBatch = ( + trx: DbTransaction, + batchSeek: TableRowsCursor | undefined, + batchOffset: number, + ask: number + ) => { const buildQuery = (executor: DbExecutor) => { // `order_key` is nullable (rows predating the backfill, and forked rows that // inherit a NULL key). A bare row-constructor comparison evaluates to NULL for @@ -1442,72 +1449,93 @@ export async function fetchRowsBounded(params: BoundedFetchParams): Promise 0 ? query.offset(batchOffset) : query } - return withReadGuards( - async (trx) => { - const batch = await buildQuery(trx) - let cut = false - const returnedRows: Array = [] - for (const fetchedRow of batch) { - // Project before measuring: the budget is a promise about the response, - // so columns the caller will never receive must not count against it. - const row = columnIds - ? { ...fetchedRow, data: projectRowData(fetchedRow.data as RowData, columnIds) } - : fetchedRow - const rowBytes = Buffer.byteLength(JSON.stringify(row.data)) - const rowStoredBytes = columnIds - ? Buffer.byteLength(JSON.stringify(fetchedRow.data)) - : rowBytes - if (cutBytes !== undefined && rows.length > 0 && bytes + rowBytes > cutBytes) { - // Unbounded queries promise the ENTIRE result — a partial page would be - // silent truncation, so fail fast instead (the drain has only fetched - // ~budget bytes at this point, never the whole table). - if (limit === undefined) { - throw new TableQueryValidationError( - `Query result exceeds the ${Math.floor(cutBytes / (1024 * 1024))}MB limit. Add a filter or a limit to narrow the result.`, - 'TABLE_QUERY_RESULT_TOO_LARGE' - ) - } - // Bounded page, byte cut opted in: `row` is the witness. Requires a - // non-empty page so a single over-budget row is still returned alone. - hasMore = true - cut = true - break - } - // Limit cut: `row` is the +1 peek witness. - if (rows.length === limit) { - hasMore = true - cut = true - break - } - rows.push(row) - returnedRows.push(row) - bytes += rowBytes - storedBytes += rowStoredBytes - consumedSinceAnchor++ - if (rowBytes > maxRowBytes) maxRowBytes = rowBytes - if (rowStoredBytes > maxStoredRowBytes) maxStoredRowBytes = rowStoredBytes - if (keysetValid && row.orderKey) { - anchor = { orderKey: row.orderKey, id: row.id } - anchorOffset = 0 - consumedSinceAnchor = 0 + return (async () => { + const batch = await buildQuery(trx) + let cut = false + const returnedRows: Array = [] + for (const fetchedRow of batch) { + // Project before measuring: the budget is a promise about the response, + // so columns the caller will never receive must not count against it. + const row = columnIds + ? { ...fetchedRow, data: projectRowData(fetchedRow.data as RowData, columnIds) } + : fetchedRow + const rowBytes = Buffer.byteLength(JSON.stringify(row.data)) + const rowStoredBytes = columnIds + ? Buffer.byteLength(JSON.stringify(fetchedRow.data)) + : rowBytes + if (cutBytes !== undefined && rows.length > 0 && bytes + rowBytes > cutBytes) { + // Unbounded queries promise the ENTIRE result — a partial page would be + // silent truncation, so fail fast instead (the drain has only fetched + // ~budget bytes at this point, never the whole table). + if (limit === undefined) { + throw new TableQueryValidationError( + `Query result exceeds the ${Math.floor(cutBytes / (1024 * 1024))}MB limit. Add a filter or a limit to narrow the result.`, + 'TABLE_QUERY_RESULT_TOO_LARGE' + ) } + // Bounded page, byte cut opted in: `row` is the witness. Requires a + // non-empty page so a single over-budget row is still returned alone. + hasMore = true + cut = true + break } - await params.readProvenance?.capture(trx, returnedRows) - return { cut, batchLength: batch.length } - }, - { seqscanOff: sorted, repeatableRead: Boolean(params.readProvenance) } - ) + // Limit cut: `row` is the +1 peek witness. + if (rows.length === limit) { + hasMore = true + cut = true + break + } + rows.push(row) + returnedRows.push(row) + bytes += rowBytes + storedBytes += rowStoredBytes + consumedSinceAnchor++ + if (rowBytes > maxRowBytes) maxRowBytes = rowBytes + if (rowStoredBytes > maxStoredRowBytes) maxStoredRowBytes = rowStoredBytes + if (keysetValid && row.orderKey) { + anchor = { orderKey: row.orderKey, id: row.id } + anchorOffset = 0 + consumedSinceAnchor = 0 + } + } + await params.readProvenance?.capture(trx, returnedRows) + return { cut, batchLength: batch.length } + })() } - while (true) { - const limitRemaining = limit === undefined ? Number.POSITIVE_INFINITY : limit - rows.length - const target = Math.min(nextBatchRows(), limitRemaining) - const ask = target + 1 - const { cut, batchLength } = await runBatch(anchor, anchorOffset + consumedSinceAnchor, ask) - if (cut) break - // Short batch = the source is exhausted; hasMore stays false. - if (batchLength < ask) break - } + /** + * One transaction for the whole drain, not one per batch. + * + * The guards are identical on every batch — `seqscanOff` and `repeatableRead` are fixed for the + * call — so opening a transaction per batch bought nothing and cost `BEGIN`, the `set_config` + * statement and `COMMIT` on each one. A 1000-row page drains in two batches, so that was six + * round trips of pure protocol on the critical path of the grid's first read. + * + * Row visibility is unchanged. Under READ COMMITTED (the unprovenance path) every statement + * still takes its own snapshot, so a batch sees exactly what a separate transaction would have. + * Under REPEATABLE READ (the provenance path) the batches now share one snapshot instead of + * taking one each, which is strictly more consistent: a row and the sidecar captured for it can + * no longer come from different points in time across a batch boundary. + */ + await withReadGuards( + async (trx) => { + while (true) { + const limitRemaining = limit === undefined ? Number.POSITIVE_INFINITY : limit - rows.length + const target = Math.min(nextBatchRows(), limitRemaining) + const ask = target + 1 + const { cut, batchLength } = await runBatch( + trx, + anchor, + anchorOffset + consumedSinceAnchor, + ask + ) + if (cut) break + // Short batch = the source is exhausted; hasMore stays false. + if (batchLength < ask) break + } + }, + { seqscanOff: sorted, repeatableRead: Boolean(params.readProvenance) } + ) return { rows, diff --git a/apps/sim/lib/table/tx.ts b/apps/sim/lib/table/tx.ts index 4e0b8b2dcbe..43e5f57bf2d 100644 --- a/apps/sim/lib/table/tx.ts +++ b/apps/sim/lib/table/tx.ts @@ -11,13 +11,22 @@ import type { DbTransaction } from '@/lib/table/planner' const TIMEOUT_CAP_MS = 10 * 60_000 /** - * Sets per-transaction Postgres timeouts via `SET LOCAL`. + * Sets per-transaction Postgres timeouts, in ONE round trip. * * `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. + * `set_config(name, value, is_local => true)` is exactly `SET LOCAL` — transaction-scoped, dying + * with the commit, and reverting the same way on a savepoint rollback — but it is a function + * call, so all three fit in one `SELECT`. Three separate `SET LOCAL` statements each cost their + * own round trip, and every table write opens a transaction that begins with this call, so that + * was two wasted round trips on every insert, update, delete, import and column change. It also + * takes the values as bound parameters, which `SET LOCAL` cannot — the reason this is one + * statement rather than three semicolon-joined ones, since a bound parameter forces the extended + * protocol and that rejects multiple commands per message. + * + * Safe under pgBouncer transaction pooling — the settings are transaction-scoped and cleared at + * COMMIT/ROLLBACK before the session returns to the pool. */ export async function setTableTxTimeouts( trx: DbTransaction, @@ -26,9 +35,12 @@ export async function setTableTxTimeouts( 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'`)) + await trx.execute(sql` + select + set_config('statement_timeout', ${`${s}ms`}, true), + set_config('lock_timeout', ${`${l}ms`}, true), + set_config('idle_in_transaction_session_timeout', ${`${i}ms`}, true) + `) } /**