From 447f52547fcfae58cad1db9896501f09c4471cec Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 21 Sep 2026 01:52:22 -0700 Subject: [PATCH 1/2] improvement(tables): cut the DB round trips a table read and write spend on protocol MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../app/api/v1/tables/[tableId]/rows/route.ts | 2 + .../service-filter-threading.test.ts | 88 +++++++++ .../lib/table/__tests__/update-row.test.ts | 43 ++++- apps/sim/lib/table/application/rows.test.ts | 1 + apps/sim/lib/table/application/rows.ts | 4 + apps/sim/lib/table/rows/executions.test.ts | 44 ++++- apps/sim/lib/table/rows/executions.ts | 18 ++ .../sim/lib/table/rows/pending-delete-mask.ts | 43 ++++- apps/sim/lib/table/rows/service.ts | 175 +++++++++++------- apps/sim/lib/table/tx.ts | 24 ++- apps/sim/lib/table/types.ts | 10 + 11 files changed, 361 insertions(+), 91 deletions(-) diff --git a/apps/sim/app/api/v1/tables/[tableId]/rows/route.ts b/apps/sim/app/api/v1/tables/[tableId]/rows/route.ts index 3a497fdb39c..74c28ee0a80 100644 --- a/apps/sim/app/api/v1/tables/[tableId]/rows/route.ts +++ b/apps/sim/app/api/v1/tables/[tableId]/rows/route.ts @@ -185,6 +185,8 @@ export const GET = withRouteHandler(async (request: NextRequest, context: TableR offset: validated.offset, includeTotal: validated.includeTotal, withExecutions: false, + // `table` was loaded a few lines above, for this read. + trustLoadedJob: true, }, requestId ) 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..aabbbd312a8 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,89 @@ describe('queryRows byte budget', () => { }) }) }) + +/** + * Two reads the row path used to make unconditionally, each one round trip on the grid's hot + * page. Both are now answered from state the caller already holds; these pin that the query is + * actually skipped rather than merely ignored, and that the fallbacks still run. + */ +describe('queryRows round-trip elision', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + vi.mocked(tableMayHaveRunState).mockReturnValue(true) + }) + + const hydrated = ( + fields: Partial> + ): TableDefinition => ({ ...TABLE, jobStatus: null, jobType: null, ...fields }) + + it('skips the delete-job probe when the hydrated table shows no running delete', async () => { + await queryRows( + hydrated({}), + { limit: 5, includeTotal: false, withExecutions: false, trustLoadedJob: true }, + 'req-1' + ) + + // With no probe, the drain batch is the FIRST bounded query rather than the second. + expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 6) + }) + + it('still probes when the hydrated table shows a running delete job', async () => { + await queryRows( + hydrated({ jobStatus: 'running', jobType: 'delete' }), + { limit: 5, includeTotal: false, withExecutions: false, trustLoadedJob: true }, + 'req-1' + ) + + expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 1) + expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(2, 6) + }) + + /** An unhydrated definition cannot rule the job out, so it keeps the lookup it always had. */ + it('still probes when the table carries no job fields at all', async () => { + await queryRows( + TABLE, + { limit: 5, includeTotal: false, withExecutions: false, trustLoadedJob: true }, + 'req-1' + ) + + expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 1) + expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(2, 6) + }) + + /** + * A caller that holds its table across a long walk (the export stream) must keep re-asking, so + * a delete job starting mid-walk still begins masking its doomed rows. + */ + it('still probes for a caller that did not vouch for its table', async () => { + await queryRows(hydrated({}), { limit: 5, includeTotal: false, withExecutions: false }, 'req-1') + + expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 1) + expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(2, 6) + }) + + 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/application/rows.test.ts b/apps/sim/lib/table/application/rows.test.ts index e8b7974d555..e46be5f23c0 100644 --- a/apps/sim/lib/table/application/rows.test.ts +++ b/apps/sim/lib/table/application/rows.test.ts @@ -769,6 +769,7 @@ describe('row query and upsert application semantics', () => { includeTotal: false, withExecutions: false, runStateBudgetBytes: TABLE_LIMITS.MAX_ROW_RUN_STATE_BYTES, + trustLoadedJob: true, }, expect.any(String) ) diff --git a/apps/sim/lib/table/application/rows.ts b/apps/sim/lib/table/application/rows.ts index 284e8a794ca..9438813686c 100644 --- a/apps/sim/lib/table/application/rows.ts +++ b/apps/sim/lib/table/application/rows.ts @@ -436,6 +436,8 @@ export const listTableRows = defineAuthorizedTableUseCase({ includeTotal: false, withExecutions: input.includeRunState ?? false, runStateBudgetBytes: TABLE_LIMITS.MAX_ROW_RUN_STATE_BYTES, + // `context.table` was loaded by this use case's own resolver, for this read. + trustLoadedJob: true, }, requestId(input) ) @@ -573,6 +575,8 @@ export const queryTableRows = defineAuthorizedTableUseCase({ withExecutions: input.includeRunState ?? false, runStateBudgetBytes: TABLE_LIMITS.MAX_ROW_RUN_STATE_BYTES, columnIds, + // `context.table` was loaded by this use case's own resolver, for this read. + trustLoadedJob: true, }, requestId(input), readProvenance 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/pending-delete-mask.ts b/apps/sim/lib/table/rows/pending-delete-mask.ts index 5036657e307..7d3095008f0 100644 --- a/apps/sim/lib/table/rows/pending-delete-mask.ts +++ b/apps/sim/lib/table/rows/pending-delete-mask.ts @@ -9,6 +9,43 @@ import type { TableDefinition, TableDeleteJobPayload } from '@/lib/table/types' const logger = createLogger('TablePendingDeleteMask') +/** + * Whether {@link PendingDeleteMaskOptions.trustLoadedJob} can rule a running delete job out. + * + * Every table loaded through `getTableById` / `listTables` already carries its latest non-export + * job, folded into that same SELECT as a lateral, so a caller that loaded its table for this very + * read already holds the answer and the lookup is pure overhead on the hot path. + * + * The derivation is exact, not a heuristic. `table_jobs_one_active_per_table` is unique on + * `table_id WHERE status = 'running' AND type <> 'export'` — the same predicate the lateral + * filters on — so a running delete job is the ONLY running non-export job on its table, and no + * further non-export job can be inserted while it holds that slot. It is therefore the newest + * non-export job by `started_at`, which is exactly the row the lateral returns. + * + * `jobStatus === undefined` means the fields were never hydrated (a `TableDefinition` assembled + * by some other path), which is indistinguishable from "no job" in the shape alone — so that case + * falls back to the query rather than assuming. A hydrated table with no job has + * `jobStatus: null`. + */ +function hydratedJobRulesOutDelete(table: TableDefinition): boolean { + if (table.jobStatus === undefined) return false + return !(table.jobStatus === 'running' && table.jobType === 'delete') +} + +export interface PendingDeleteMaskOptions { + /** + * Answer from `table`'s own latest-job fields when they rule a running delete out, instead of + * querying for one. + * + * Only for a caller whose `table` was loaded for this read: the fields are then as fresh as the + * query would have been. A caller that loads a table once and then pages for a while — the + * export runner, the snapshot builder — must NOT set this, because a delete job starting + * mid-walk would never appear in its snapshot and its later pages would stop masking doomed + * rows. Those callers keep re-asking per page, which is what makes the mask appear mid-walk. + */ + trustLoadedJob?: boolean +} + /** * Visibility mask for a running delete job: returns a clause keeping only rows the job will NOT * delete, or `undefined` when no delete job is running. The job's persisted scope @@ -20,7 +57,11 @@ const logger = createLogger('TablePendingDeleteMask') * `(doomed) IS NOT TRUE` rather than `NOT (doomed)`: JSONB predicates evaluate to NULL on missing * cells, and those rows are NOT selected for deletion (NULL ≠ TRUE) — they must stay visible. */ -export async function pendingDeleteMask(table: TableDefinition): Promise { +export async function pendingDeleteMask( + table: TableDefinition, + options?: PendingDeleteMaskOptions +): Promise { + if (options?.trustLoadedJob && hydratedJobRulesOutDelete(table)) return undefined const [job] = await db .select({ payload: tableJobs.payload }) .from(tableJobs) diff --git a/apps/sim/lib/table/rows/service.ts b/apps/sim/lib/table/rows/service.ts index 1bbd2e96418..4df092d8c19 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 { @@ -1041,7 +1042,9 @@ export async function findRowMatches( if (columnIds.length === 0) return { matches: [], truncated: false } // Same visibility rule as queryRows: don't surface rows a running delete job will remove. - const deleteMask = await pendingDeleteMask(table) + // Search is always one shot against a table its caller just loaded, so the table's own + // latest-job fields are as fresh as a lookup would be — see `PendingDeleteMaskOptions`. + const deleteMask = await pendingDeleteMask(table, { trustLoadedJob: true }) const baseConditions = and( eq(userTableRows.tableId, table.id), @@ -1161,6 +1164,7 @@ export async function queryRows( withExecutions = true, runStateBudgetBytes, columnIds, + trustLoadedJob = false, } = options const tableName = USER_TABLE_ROWS_SQL_NAME @@ -1168,7 +1172,7 @@ export async function queryRows( // Hide rows a running delete job is about to remove — both the page and the count below share // this clause, so totals stay consistent with the visible rows. - const deleteMask = await pendingDeleteMask(table) + const deleteMask = await pendingDeleteMask(table, { trustLoadedJob }) const baseConditions = and( eq(userTableRows.tableId, table.id), @@ -1250,13 +1254,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 +1424,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 +1452,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) + `) } /** diff --git a/apps/sim/lib/table/types.ts b/apps/sim/lib/table/types.ts index 98747c75ed6..9d77c3b17f6 100644 --- a/apps/sim/lib/table/types.ts +++ b/apps/sim/lib/table/types.ts @@ -626,6 +626,16 @@ export interface QueryOptions { * Mutually exclusive with `sort`. May be combined with `offset` (a compound * cursor seeks the anchor, then offsets past unkeyed rows consumed after it). */ after?: TableRowsCursor + /** + * Answer "is a delete job running?" from `table`'s own latest-job fields rather than querying + * for one, saving a round trip on every read. + * + * Set it only when `table` was loaded for this very read — a request handler that resolved its + * table and is about to return. A caller that loads a table once and then pages for a while + * (the export stream) must leave it off, so a delete job starting mid-walk still begins masking + * its doomed rows partway through. + */ + trustLoadedJob?: boolean /** * When true (default), runs a `COUNT(*)` and returns `totalCount` as a number. * Pass `false` to skip the count query (grid UI doesn't need it); `totalCount` From bd69492e9c6cd6f3ae0eee5072a1e6ba5ce18c6e Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Mon, 21 Sep 2026 01:59:16 -0700 Subject: [PATCH 2/2] review(tables): drop the delete-mask elision and correct the provenance snapshot doc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../app/api/v1/tables/[tableId]/rows/route.ts | 2 - .../service-filter-threading.test.ts | 57 ++----------------- apps/sim/lib/table/application/rows.test.ts | 1 - apps/sim/lib/table/application/rows.ts | 4 -- .../sim/lib/table/rows/pending-delete-mask.ts | 43 +------------- apps/sim/lib/table/rows/secret-provenance.ts | 6 +- apps/sim/lib/table/rows/service.ts | 7 +-- apps/sim/lib/table/types.ts | 10 ---- 8 files changed, 11 insertions(+), 119 deletions(-) diff --git a/apps/sim/app/api/v1/tables/[tableId]/rows/route.ts b/apps/sim/app/api/v1/tables/[tableId]/rows/route.ts index 74c28ee0a80..3a497fdb39c 100644 --- a/apps/sim/app/api/v1/tables/[tableId]/rows/route.ts +++ b/apps/sim/app/api/v1/tables/[tableId]/rows/route.ts @@ -185,8 +185,6 @@ export const GET = withRouteHandler(async (request: NextRequest, context: TableR offset: validated.offset, includeTotal: validated.includeTotal, withExecutions: false, - // `table` was loaded a few lines above, for this read. - trustLoadedJob: true, }, requestId ) 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 aabbbd312a8..d17890020e0 100644 --- a/apps/sim/lib/table/__tests__/service-filter-threading.test.ts +++ b/apps/sim/lib/table/__tests__/service-filter-threading.test.ts @@ -670,66 +670,17 @@ describe('queryRows byte budget', () => { }) /** - * Two reads the row path used to make unconditionally, each one round trip on the grid's hot - * page. Both are now answered from state the caller already holds; these pin that the query is - * actually skipped rather than merely ignored, and that the fallbacks still run. + * 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 round-trip elision', () => { +describe('queryRows run-state elision', () => { beforeEach(() => { vi.clearAllMocks() resetDbChainMock() vi.mocked(tableMayHaveRunState).mockReturnValue(true) }) - const hydrated = ( - fields: Partial> - ): TableDefinition => ({ ...TABLE, jobStatus: null, jobType: null, ...fields }) - - it('skips the delete-job probe when the hydrated table shows no running delete', async () => { - await queryRows( - hydrated({}), - { limit: 5, includeTotal: false, withExecutions: false, trustLoadedJob: true }, - 'req-1' - ) - - // With no probe, the drain batch is the FIRST bounded query rather than the second. - expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 6) - }) - - it('still probes when the hydrated table shows a running delete job', async () => { - await queryRows( - hydrated({ jobStatus: 'running', jobType: 'delete' }), - { limit: 5, includeTotal: false, withExecutions: false, trustLoadedJob: true }, - 'req-1' - ) - - expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 1) - expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(2, 6) - }) - - /** An unhydrated definition cannot rule the job out, so it keeps the lookup it always had. */ - it('still probes when the table carries no job fields at all', async () => { - await queryRows( - TABLE, - { limit: 5, includeTotal: false, withExecutions: false, trustLoadedJob: true }, - 'req-1' - ) - - expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 1) - expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(2, 6) - }) - - /** - * A caller that holds its table across a long walk (the export stream) must keep re-asking, so - * a delete job starting mid-walk still begins masking its doomed rows. - */ - it('still probes for a caller that did not vouch for its table', async () => { - await queryRows(hydrated({}), { limit: 5, includeTotal: false, withExecutions: false }, 'req-1') - - expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(1, 1) - expect(dbChainMockFns.limit).toHaveBeenNthCalledWith(2, 6) - }) - 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([]) diff --git a/apps/sim/lib/table/application/rows.test.ts b/apps/sim/lib/table/application/rows.test.ts index e46be5f23c0..e8b7974d555 100644 --- a/apps/sim/lib/table/application/rows.test.ts +++ b/apps/sim/lib/table/application/rows.test.ts @@ -769,7 +769,6 @@ describe('row query and upsert application semantics', () => { includeTotal: false, withExecutions: false, runStateBudgetBytes: TABLE_LIMITS.MAX_ROW_RUN_STATE_BYTES, - trustLoadedJob: true, }, expect.any(String) ) diff --git a/apps/sim/lib/table/application/rows.ts b/apps/sim/lib/table/application/rows.ts index 9438813686c..284e8a794ca 100644 --- a/apps/sim/lib/table/application/rows.ts +++ b/apps/sim/lib/table/application/rows.ts @@ -436,8 +436,6 @@ export const listTableRows = defineAuthorizedTableUseCase({ includeTotal: false, withExecutions: input.includeRunState ?? false, runStateBudgetBytes: TABLE_LIMITS.MAX_ROW_RUN_STATE_BYTES, - // `context.table` was loaded by this use case's own resolver, for this read. - trustLoadedJob: true, }, requestId(input) ) @@ -575,8 +573,6 @@ export const queryTableRows = defineAuthorizedTableUseCase({ withExecutions: input.includeRunState ?? false, runStateBudgetBytes: TABLE_LIMITS.MAX_ROW_RUN_STATE_BYTES, columnIds, - // `context.table` was loaded by this use case's own resolver, for this read. - trustLoadedJob: true, }, requestId(input), readProvenance diff --git a/apps/sim/lib/table/rows/pending-delete-mask.ts b/apps/sim/lib/table/rows/pending-delete-mask.ts index 7d3095008f0..5036657e307 100644 --- a/apps/sim/lib/table/rows/pending-delete-mask.ts +++ b/apps/sim/lib/table/rows/pending-delete-mask.ts @@ -9,43 +9,6 @@ import type { TableDefinition, TableDeleteJobPayload } from '@/lib/table/types' const logger = createLogger('TablePendingDeleteMask') -/** - * Whether {@link PendingDeleteMaskOptions.trustLoadedJob} can rule a running delete job out. - * - * Every table loaded through `getTableById` / `listTables` already carries its latest non-export - * job, folded into that same SELECT as a lateral, so a caller that loaded its table for this very - * read already holds the answer and the lookup is pure overhead on the hot path. - * - * The derivation is exact, not a heuristic. `table_jobs_one_active_per_table` is unique on - * `table_id WHERE status = 'running' AND type <> 'export'` — the same predicate the lateral - * filters on — so a running delete job is the ONLY running non-export job on its table, and no - * further non-export job can be inserted while it holds that slot. It is therefore the newest - * non-export job by `started_at`, which is exactly the row the lateral returns. - * - * `jobStatus === undefined` means the fields were never hydrated (a `TableDefinition` assembled - * by some other path), which is indistinguishable from "no job" in the shape alone — so that case - * falls back to the query rather than assuming. A hydrated table with no job has - * `jobStatus: null`. - */ -function hydratedJobRulesOutDelete(table: TableDefinition): boolean { - if (table.jobStatus === undefined) return false - return !(table.jobStatus === 'running' && table.jobType === 'delete') -} - -export interface PendingDeleteMaskOptions { - /** - * Answer from `table`'s own latest-job fields when they rule a running delete out, instead of - * querying for one. - * - * Only for a caller whose `table` was loaded for this read: the fields are then as fresh as the - * query would have been. A caller that loads a table once and then pages for a while — the - * export runner, the snapshot builder — must NOT set this, because a delete job starting - * mid-walk would never appear in its snapshot and its later pages would stop masking doomed - * rows. Those callers keep re-asking per page, which is what makes the mask appear mid-walk. - */ - trustLoadedJob?: boolean -} - /** * Visibility mask for a running delete job: returns a clause keeping only rows the job will NOT * delete, or `undefined` when no delete job is running. The job's persisted scope @@ -57,11 +20,7 @@ export interface PendingDeleteMaskOptions { * `(doomed) IS NOT TRUE` rather than `NOT (doomed)`: JSONB predicates evaluate to NULL on missing * cells, and those rows are NOT selected for deletion (NULL ≠ TRUE) — they must stay visible. */ -export async function pendingDeleteMask( - table: TableDefinition, - options?: PendingDeleteMaskOptions -): Promise { - if (options?.trustLoadedJob && hydratedJobRulesOutDelete(table)) return undefined +export async function pendingDeleteMask(table: TableDefinition): Promise { const [job] = await db .select({ payload: tableJobs.payload }) .from(tableJobs) 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 4df092d8c19..b6cfb9b7ccb 100644 --- a/apps/sim/lib/table/rows/service.ts +++ b/apps/sim/lib/table/rows/service.ts @@ -1042,9 +1042,7 @@ export async function findRowMatches( if (columnIds.length === 0) return { matches: [], truncated: false } // Same visibility rule as queryRows: don't surface rows a running delete job will remove. - // Search is always one shot against a table its caller just loaded, so the table's own - // latest-job fields are as fresh as a lookup would be — see `PendingDeleteMaskOptions`. - const deleteMask = await pendingDeleteMask(table, { trustLoadedJob: true }) + const deleteMask = await pendingDeleteMask(table) const baseConditions = and( eq(userTableRows.tableId, table.id), @@ -1164,7 +1162,6 @@ export async function queryRows( withExecutions = true, runStateBudgetBytes, columnIds, - trustLoadedJob = false, } = options const tableName = USER_TABLE_ROWS_SQL_NAME @@ -1172,7 +1169,7 @@ export async function queryRows( // Hide rows a running delete job is about to remove — both the page and the count below share // this clause, so totals stay consistent with the visible rows. - const deleteMask = await pendingDeleteMask(table, { trustLoadedJob }) + const deleteMask = await pendingDeleteMask(table) const baseConditions = and( eq(userTableRows.tableId, table.id), diff --git a/apps/sim/lib/table/types.ts b/apps/sim/lib/table/types.ts index 9d77c3b17f6..98747c75ed6 100644 --- a/apps/sim/lib/table/types.ts +++ b/apps/sim/lib/table/types.ts @@ -626,16 +626,6 @@ export interface QueryOptions { * Mutually exclusive with `sort`. May be combined with `offset` (a compound * cursor seeks the anchor, then offsets past unkeyed rows consumed after it). */ after?: TableRowsCursor - /** - * Answer "is a delete job running?" from `table`'s own latest-job fields rather than querying - * for one, saving a round trip on every read. - * - * Set it only when `table` was loaded for this very read — a request handler that resolved its - * table and is about to return. A caller that loads a table once and then pages for a while - * (the export stream) must leave it off, so a delete job starting mid-walk still begins masking - * its doomed rows partway through. - */ - trustLoadedJob?: boolean /** * When true (default), runs a `COUNT(*)` and returns `totalCount` as a number. * Pass `false` to skip the count query (grid UI doesn't need it); `totalCount`