From 26ccccc0f55f81b7a9529fe97ab7ff0587b84a92 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 22 Oct 2026 13:17:38 -0700 Subject: [PATCH 1/2] fix(logs): cap logs-list pages that carry execution data Pages with includeTraceSpans/includeFinalOutput are capped at one materialization wave (20 rows) before the SQL read and at a 64MB stored-archive budget, first row always admitted. The cursor is minted from the last returned row, so walks never skip or repeat. Shared by v1 and v2. --- apps/docs/openapi-v2-logs.json | 4 +- apps/sim/lib/api/contracts/v2/logs.ts | 3 +- apps/sim/lib/logs/execution/trace-store.ts | 11 ++ apps/sim/lib/logs/public-queries.test.ts | 131 +++++++++++++++++++++ apps/sim/lib/logs/public-queries.ts | 50 +++++++- packages/sim-cli/src/generated/v2-api.ts | 2 +- 6 files changed, 194 insertions(+), 7 deletions(-) diff --git a/apps/docs/openapi-v2-logs.json b/apps/docs/openapi-v2-logs.json index 8348148d972..93f58aa4c67 100644 --- a/apps/docs/openapi-v2-logs.json +++ b/apps/docs/openapi-v2-logs.json @@ -213,9 +213,9 @@ "name": "limit", "in": "query", "required": false, - "description": "Maximum log entries per page. Values outside 1–1000 are truncated and clamped into that range rather than rejected. Defaults to 100.", + "description": "Maximum log entries per page. When execution data is returned (`includeTraceSpans` or `includeFinalOutput`), a page holds at most 20 entries, and fewer when their stored traces are large; follow `nextCursor` for the rest. A page whose execution data is withheld by permissions is not capped. Values outside 1–1000 are truncated and clamped into that range rather than rejected. Defaults to 100.", "schema": { - "description": "Maximum log entries per page. Values outside 1–1000 are truncated and clamped into that range rather than rejected. Defaults to 100.", + "description": "Maximum log entries per page. When execution data is returned (`includeTraceSpans` or `includeFinalOutput`), a page holds at most 20 entries, and fewer when their stored traces are large; follow `nextCursor` for the rest. A page whose execution data is withheld by permissions is not capped. Values outside 1–1000 are truncated and clamped into that range rather than rejected. Defaults to 100.", "type": "integer", "default": 100 } diff --git a/apps/sim/lib/api/contracts/v2/logs.ts b/apps/sim/lib/api/contracts/v2/logs.ts index 447ba629afb..f5847eac519 100644 --- a/apps/sim/lib/api/contracts/v2/logs.ts +++ b/apps/sim/lib/api/contracts/v2/logs.ts @@ -20,6 +20,7 @@ import { v2TimestampSchema, } from '@/lib/api/contracts/v2/shared' import { v2RunFileSchema } from '@/lib/api/contracts/v2/workflows' +import { MATERIALIZE_CONCURRENCY } from '@/lib/core/utils/concurrency' import { PERSISTED_WORKFLOW_EXECUTION_STATUSES } from '@/lib/logs/types' /** @@ -594,7 +595,7 @@ export const v2ListLogsQuerySchema = v1ListLogsQuerySchema max: 1000, fallback: 100, outOfRange: 'clamp', - description: 'Maximum log entries per page.', + description: `Maximum log entries per page. When execution data is returned (\`includeTraceSpans\` or \`includeFinalOutput\`), a page holds at most ${MATERIALIZE_CONCURRENCY} entries, and fewer when their stored traces are large; follow \`nextCursor\` for the rest. A page whose execution data is withheld by permissions is not capped.`, }), ...v2LogSortFieldSchemas, /** diff --git a/apps/sim/lib/logs/execution/trace-store.ts b/apps/sim/lib/logs/execution/trace-store.ts index 78ae6fae575..d2334e83800 100644 --- a/apps/sim/lib/logs/execution/trace-store.ts +++ b/apps/sim/lib/logs/execution/trace-store.ts @@ -28,6 +28,17 @@ export const SECRET_PROJECTION_VERSION = 1 as const */ export const TRACE_STORE_REF_KEY = 'traceStoreRef' +/** + * Bytes of the stored archive a slim `execution_data` row points at, or 0 when + * the row carries its execution data inline. Read before materializing, so a + * caller can budget a page by what it is about to load. + */ +export function executionDataArchiveBytes(executionData: unknown): number { + if (!isRecordLike(executionData)) return 0 + const ref = executionData[TRACE_STORE_REF_KEY] + return isLargeValueRef(ref) ? ref.size : 0 +} + /** * The only metadata kept inline on the slim row (everything else lives in the * externalized object). Trace presence/count survives object expiry for log diff --git a/apps/sim/lib/logs/public-queries.test.ts b/apps/sim/lib/logs/public-queries.test.ts index 6a4c415516a..d1049c9f485 100644 --- a/apps/sim/lib/logs/public-queries.test.ts +++ b/apps/sim/lib/logs/public-queries.test.ts @@ -325,3 +325,134 @@ describe('sortable public log query', () => { expect(dbChainMockFns.limit).toHaveBeenLastCalledWith(51) }) }) + +/** + * A page that carries execution data is materialized, projected, validated, and + * serialized whole, so the caller's `limit` cannot be what sizes it. The read + * narrows the page instead, and the narrowed page has to mint its cursor from + * its own last row or the walk skips every row past the cut. + */ +describe('public log page carrying execution data', () => { + const MB = 1024 * 1024 + + beforeEach(() => { + resetDbChainMock() + }) + + /** Rows in descending start order, each pointing at an archive of `archiveBytes`. */ + function archivedRows(count: number, archiveBytes: number, idPrefix = 'w', parity = 0) { + return Array.from({ length: count }, (_, index) => { + const slot = index * 2 + parity + const second = String(59 - (slot % 60)).padStart(2, '0') + const minute = String(59 - Math.floor(slot / 60)).padStart(2, '0') + return { + id: `${idPrefix}-${index}`, + startedAtCursor: `2026-08-06T00:${minute}:${second}.000000Z`, + executionData: { + traceStoreRef: { + __simLargeValueRef: true, + version: 1, + id: 'lv_bbbbbbbbbbbb', + kind: 'object', + size: archiveBytes, + key: `execution/workspace-1/workflow-1/execution-${index}/large-value-lv_bbbbbbbbbbbb.json`, + executionId: `execution-${index}`, + }, + }, + } + }) + } + + function readPage(input: { includeExecutionData: boolean; cursorKeys?: string[] }) { + return readPublicLogPage({ + filters: { workspaceId: 'workspace-1' }, + includeExecutionData: input.includeExecutionData, + sortBy: 'startedAt', + sortOrder: 'desc', + cursorKeys: input.cursorKeys, + limit: 100, + }) + } + + /** + * Follows `nextCursorKeys` to the end, serving each page the rows strictly + * after the cursor the way the keyset predicate would, and returns every + * page's ids. + */ + async function walk(rows: ReturnType) { + const pages: string[][] = [] + let cursorKeys: string[] | undefined + while (pages.length <= rows.length) { + const after = cursorKeys ? rows.findIndex((row) => row.id === cursorKeys?.[1]) + 1 : 0 + queueTableRows(schemaMock.workflowExecutionLogs, rows.slice(after)) + const page = await readPage({ includeExecutionData: true, cursorKeys }) + pages.push(page.data.map((row) => row.id)) + if (!page.nextCursorKeys) return pages + cursorKeys = page.nextCursorKeys as string[] + } + throw new Error('walk did not terminate') + } + + it('reads at most one materialization wave of rows when execution data is requested', async () => { + const rows = archivedRows(30, MB) + + const pages = await walk(rows) + + expect(dbChainMockFns.limit).toHaveBeenCalledWith(21) + expect(dbChainMockFns.limit).not.toHaveBeenCalledWith(101) + expect(pages.map((page) => page.length)).toEqual([20, 10]) + expect(pages.flat()).toEqual(rows.map((row) => row.id)) + }) + + it('cuts a page at its archive byte budget and resumes right after the cut', async () => { + const rows = archivedRows(30, 10 * MB) + + const pages = await walk(rows) + + expect(pages.map((page) => page.length)).toEqual([6, 6, 6, 6, 6]) + expect(pages.flat()).toEqual(rows.map((row) => row.id)) + }) + + it('keeps the requested page size when no execution data is read', async () => { + const rows = archivedRows(30, MB) + queueTableRows(schemaMock.workflowExecutionLogs, rows) + + const page = await readPage({ includeExecutionData: false }) + + expect(dbChainMockFns.limit).toHaveBeenLastCalledWith(101) + expect(page.data).toHaveLength(rows.length) + expect(page.nextCursorKeys).toBeNull() + }) + + it('admits one run whose archive alone exceeds the budget, so the walk still advances', async () => { + const rows = archivedRows(3, 100 * MB) + queueTableRows(schemaMock.workflowExecutionLogs, rows) + + const page = await readPage({ includeExecutionData: true }) + + expect(page.data.map((row) => row.id)).toEqual(['w-0']) + expect(page.nextCursorKeys).toEqual([rows[0].startedAtCursor, 'w-0']) + }) + + it('budgets the merged page when job runs are interleaved with materialized runs', async () => { + const workflowRows = archivedRows(10, 40 * MB, 'w', 0) + const jobRows = archivedRows(10, 0, 'j', 1).map(({ id, startedAtCursor }) => ({ + id, + startedAtCursor, + })) + queueTableRows(schemaMock.workflowExecutionLogs, workflowRows) + queueTableRows(schemaMock.jobExecutionLogs, jobRows) + + const page = await readPublicLogPage({ + filters: { workspaceId: 'workspace-1' }, + includeExecutionData: true, + includeJobRuns: true, + sortBy: 'startedAt', + sortOrder: 'desc', + limit: 100, + }) + + expect(page.data.map((row) => row.id)).toEqual(['w-0', 'j-0']) + expect(page.nextCursorKeys).toEqual([jobRows[0].startedAtCursor, 'j-0']) + }) +}) diff --git a/apps/sim/lib/logs/public-queries.ts b/apps/sim/lib/logs/public-queries.ts index 644c50b1fab..d51990ff943 100644 --- a/apps/sim/lib/logs/public-queries.ts +++ b/apps/sim/lib/logs/public-queries.ts @@ -29,6 +29,9 @@ import { textKey, timestampKey, } from '@/lib/api/list-query' +import { MATERIALIZE_CONCURRENCY } from '@/lib/core/utils/concurrency' +import { MAX_DURABLE_LARGE_VALUE_BYTES } from '@/lib/execution/payloads/limits' +import { executionDataArchiveBytes } from '@/lib/logs/execution/trace-store' import { workflowExecutionOriginSql } from '@/lib/logs/execution-origin' import { folderScopeCondition, type LogFolderScope } from '@/lib/logs/folder-scope' import { handledErrorSpanCondition } from '@/lib/logs/handled-errors' @@ -354,6 +357,44 @@ function readJobLogRows( .limit(input.limit + 1) } +/** + * Rows a page that carries execution data may hold: one + * {@link MATERIALIZE_CONCURRENCY} wave. Each row is materialized from its + * archive and secret-projected before the page is validated and serialized in + * one synchronous pass, so the caller's `limit` alone would let one request hold + * a process's heap and event loop for as many traces as it asks for. + */ +const MATERIALIZED_LOG_PAGE_MAX = MATERIALIZE_CONCURRENCY + +/** + * Stored archive bytes a page that carries execution data may load. The row + * count alone does not bound a page: archives vary by orders of magnitude, and a + * page of the largest serializes past what one string can hold. The first row is + * always admitted, so a page of one oversized run still advances the walk. + */ +const MATERIALIZED_LOG_PAGE_BYTES = MAX_DURABLE_LARGE_VALUE_BYTES + +/** + * The page size a read may return: the caller's `limit`, narrowed when the page + * carries execution data. The narrowed page still mints its cursor from its own + * last row, so a walk resumes immediately after it — the caller follows + * `nextCursor` and sees every row once. + */ +function executionDataPageLimit( + rows: readonly { executionData?: unknown }[], + input: ReadPublicLogPageInput +): number { + if (!input.includeExecutionData) return input.limit + let bytes = 0 + let admitted = 0 + for (const row of rows.slice(0, input.limit)) { + bytes += executionDataArchiveBytes(row.executionData) + if (admitted > 0 && bytes > MATERIALIZED_LOG_PAGE_BYTES) break + admitted += 1 + } + return admitted +} + /** * Reads one page of the public log sequence, ordered by any of * {@link PUBLIC_LOG_SORT_FIELDS} and resumed from a shared keyset cursor. @@ -368,8 +409,11 @@ function readJobLogRows( * another page" without a count. */ export async function readPublicLogPage( - input: ReadPublicLogPageInput + pageInput: ReadPublicLogPageInput ): Promise> { + const input = pageInput.includeExecutionData + ? { ...pageInput, limit: Math.min(pageInput.limit, MATERIALIZED_LOG_PAGE_MAX) } + : pageInput const keys = publicLogKeyset(input.sortBy, input.cursorKeys) // `folderScope` is checked separately from `jobLogsSelectable`, which reads @@ -399,7 +443,7 @@ export async function readPublicLogPage( ]) if (!includeJobRuns) { - const page = keysetPage(keys, workflowRows, input.limit) + const page = keysetPage(keys, workflowRows, executionDataPageLimit(workflowRows, input)) return { data: page.data.map((row): PublicLogListRow => ({ kind: 'workflow', ...row })), nextCursorKeys: page.nextCursorKeys, @@ -419,7 +463,7 @@ export async function readPublicLogPage( mergedKeys, input.sortOrder ) - return keysetPage(mergedKeys, merged, input.limit) + return keysetPage(mergedKeys, merged, executionDataPageLimit(merged, input)) } /** diff --git a/packages/sim-cli/src/generated/v2-api.ts b/packages/sim-cli/src/generated/v2-api.ts index c5ec0c6a84b..fa11499d45a 100644 --- a/packages/sim-cli/src/generated/v2-api.ts +++ b/packages/sim-cli/src/generated/v2-api.ts @@ -20373,7 +20373,7 @@ export const V2_OPERATIONS = { kind: 'integer', default: 100, describe: - 'Maximum log entries per page. Values outside 1–1000 are truncated and clamped into that range rather than rejected. Defaults to 100.', + 'Maximum log entries per page. When execution data is returned (`includeTraceSpans` or `includeFinalOutput`), a page holds at most 20 entries, and fewer when their stored traces are large; follow `nextCursor` for the rest. A page whose execution data is withheld by permissions is not capped. Values outside 1–1000 are truncated and clamped into that range rather than rejected. Defaults to 100.', }, cursor: { kind: 'string', From 92a7ba10c8cfc98492ccf2b2e9e34ddda831bdfd Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 22 Oct 2026 13:31:16 -0700 Subject: [PATCH 2/2] fix(logs): type the page budget over both branches' rows --- apps/sim/lib/logs/public-queries.test.ts | 1 + apps/sim/lib/logs/public-queries.ts | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/logs/public-queries.test.ts b/apps/sim/lib/logs/public-queries.test.ts index d1049c9f485..97a2586ccf0 100644 --- a/apps/sim/lib/logs/public-queries.test.ts +++ b/apps/sim/lib/logs/public-queries.test.ts @@ -449,6 +449,7 @@ describe('public log page carrying execution data', () => { includeJobRuns: true, sortBy: 'startedAt', sortOrder: 'desc', + cursorKeys: undefined, limit: 100, }) diff --git a/apps/sim/lib/logs/public-queries.ts b/apps/sim/lib/logs/public-queries.ts index d51990ff943..3649b1176be 100644 --- a/apps/sim/lib/logs/public-queries.ts +++ b/apps/sim/lib/logs/public-queries.ts @@ -381,14 +381,14 @@ const MATERIALIZED_LOG_PAGE_BYTES = MAX_DURABLE_LARGE_VALUE_BYTES * `nextCursor` and sees every row once. */ function executionDataPageLimit( - rows: readonly { executionData?: unknown }[], + rows: readonly (PublicWorkflowLogListRow | PublicLogListRow)[], input: ReadPublicLogPageInput ): number { if (!input.includeExecutionData) return input.limit let bytes = 0 let admitted = 0 for (const row of rows.slice(0, input.limit)) { - bytes += executionDataArchiveBytes(row.executionData) + if ('executionData' in row) bytes += executionDataArchiveBytes(row.executionData) if (admitted > 0 && bytes > MATERIALIZED_LOG_PAGE_BYTES) break admitted += 1 }