diff --git a/apps/sim/app/api/logs/export/route.test.ts b/apps/sim/app/api/logs/export/route.test.ts index 52cfac5171d..a1a035d429a 100644 --- a/apps/sim/app/api/logs/export/route.test.ts +++ b/apps/sim/app/api/logs/export/route.test.ts @@ -8,7 +8,7 @@ import { import { permissionsMock, permissionsMockFns } from '@sim/testing/mocks/permissions.mock' import { createMockRequest } from '@sim/testing/mocks/request.mock' import { traceStoreMock, traceStoreMockFns } from '@sim/testing/mocks/trace-store.mock' -import { beforeEach, describe, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, type MockInstance, vi } from 'vitest' const { mockExpandFolderIdsWithDescendants, mockMapWithConcurrency } = vi.hoisted(() => ({ mockExpandFolderIdsWithDescendants: vi.fn(), @@ -30,6 +30,7 @@ vi.mock('@/lib/core/utils/concurrency', () => ({ mapWithConcurrency: mockMapWithConcurrency, })) +import { toCsvRow } from '@/lib/core/utils/csv' import { capabilityRefusal } from '@/lib/permission-groups/capabilities' import { GET } from '@/app/api/logs/export/route' @@ -73,7 +74,20 @@ function flattenConditions(condition: unknown): Array> { } describe('GET /api/logs/export', () => { + /** + * Every read advances past the export's CPU slice, so each piece of a row ends its slice. One + * clock for the whole file, never moving backwards, keeps the process-wide slice deadline it + * sets meaningful for every later test. + */ + let fakeClockMs = 0 + let clock: MockInstance<() => number> | undefined + beforeEach(() => { + fakeClockMs = Math.max(fakeClockMs, performance.now()) + clock = vi.spyOn(performance, 'now').mockImplementation(() => { + fakeClockMs += 20 + return fakeClockMs + }) resetDbChainMock() mockGetSession.mockResolvedValue({ user: { id: 'user-1' } }) mockCheckWorkspaceAccess.mockResolvedValue({ hasAccess: true }) @@ -93,6 +107,10 @@ describe('GET /api/logs/export', () => { ) }) + afterEach(() => { + clock?.mockRestore() + }) + it('returns only the CSV header when workspace access is denied', async () => { mockCheckWorkspaceAccess.mockResolvedValueOnce({ hasAccess: false }) @@ -216,6 +234,64 @@ describe('GET /api/logs/export', () => { await expect(Promise.all([pendingRead, cancellation])).resolves.toBeDefined() }) + it('streams one large trace row in pieces with event-loop turns between them', async () => { + const content = 'say "hi", then wait\n'.repeat(100_000) + const traceSpans = [ + { id: 'root', name: 'Workflow Execution', children: [{ output: { content } }] }, + ] + queueTableRows(workflowExecutionLogs, [ + logRow(0, { executionData: { message: 'm', traceSpans } }), + ]) + + const response = await GET(makeRequest(), undefined) + if (!response.body) throw new Error('expected a streamed export body') + const reader = response.body.getReader() + const decoder = new TextDecoder() + const header = decoder.decode((await reader.read()).value) + + let turns = 0 + let streaming = true + const tick = () => { + if (!streaming) return + turns += 1 + setImmediate(tick) + } + setImmediate(tick) + const turnsAtEachRowChunk: number[] = [] + let row = '' + try { + while (true) { + const { done, value } = await reader.read() + if (done) break + row += decoder.decode(value, { stream: true }) + turnsAtEachRowChunk.push(turns) + } + } finally { + streaming = false + } + const turnsBetweenRowChunks = turnsAtEachRowChunk + .slice(1) + .map((turnsAtChunk, index) => turnsAtChunk - turnsAtEachRowChunk[index]) + + expect(header).toMatch(/^startedAt,.*\n$/) + expect(turnsAtEachRowChunk.length).toBeGreaterThan(1) + expect(turnsBetweenRowChunks.some((delta) => delta > 0)).toBe(true) + expect(row).toBe( + `${toCsvRow([ + logRow(0).startedAt.toISOString(), + 'info', + 'Workflow', + 'manual', + '500', + '0.01', + 'workflow-1', + 'execution-0', + 'm', + JSON.stringify(traceSpans), + ])}\n` + ) + }) + it('blanks the cost column and span spend when the group withholds cost', async () => { mockGetUserPermissionConfig.mockResolvedValue({ hideCostInfo: true }) queueTableRows(workflowExecutionLogs, [ diff --git a/apps/sim/app/api/logs/export/route.ts b/apps/sim/app/api/logs/export/route.ts index ad4cca9381c..73468428ae4 100644 --- a/apps/sim/app/api/logs/export/route.ts +++ b/apps/sim/app/api/logs/export/route.ts @@ -6,7 +6,8 @@ import { and, desc, eq, lt, or, type SQL, sql } from 'drizzle-orm' import { type NextRequest, NextResponse } from 'next/server' import { getSession } from '@/lib/auth' import { MATERIALIZE_CONCURRENCY, mapWithConcurrency } from '@/lib/core/utils/concurrency' -import { formatCsvValue, toCsvRow } from '@/lib/core/utils/csv' +import { formatCsvValue, toCsvLinePieces, toCsvRow } from '@/lib/core/utils/csv' +import { yieldWhenSliceExpires } from '@/lib/core/utils/event-loop' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { materializeExecutionDataForDisplay } from '@/lib/logs/execution/trace-store' import { withheldSpendData } from '@/lib/logs/fetch-log-detail' @@ -223,7 +224,7 @@ export const GET = withRouteHandler(async (request: NextRequest) => { error: getErrorMessage(rowError), }) } - const line = toCsvRow([ + const pieces = toCsvLinePieces([ formatCsvValue(row.startedAt), formatCsvValue(row.level), formatCsvValue(row.workflowName), @@ -235,7 +236,14 @@ export const GET = withRouteHandler(async (request: NextRequest) => { formatCsvValue(message), formatCsvValue(tracesJson), ]) - yield encoder.encode(`${line}\n`) + /** + * One trace can serialize to tens of megabytes, and escaping it in one run would hold + * the event loop for every other request on this server; small rows share a slice. + */ + for (const piece of pieces) { + yield encoder.encode(piece) + await yieldWhenSliceExpires() + } } } diff --git a/apps/sim/lib/core/utils/csv.test.ts b/apps/sim/lib/core/utils/csv.test.ts index c2c938f1337..2f207cecae2 100644 --- a/apps/sim/lib/core/utils/csv.test.ts +++ b/apps/sim/lib/core/utils/csv.test.ts @@ -1,5 +1,10 @@ import { describe, expect, it } from 'vitest' -import { formatCsvValue, neutralizeCsvFormula, toCsvRow } from '@/lib/core/utils/csv' +import { + formatCsvValue, + neutralizeCsvFormula, + toCsvLinePieces, + toCsvRow, +} from '@/lib/core/utils/csv' describe('CSV formatting', () => { it.each(['=1+1', '+1+1', '-1+1', '@SUM(A1)', '\t=1+1', '\r=1+1'])( @@ -19,4 +24,26 @@ describe('CSV formatting', () => { 'plain,"with,comma","with""quote","with\nnewline","with\rreturn"' ) }) + + it.each([1, 2, 3, 7, 0, -5, 1.5, Number.NaN])( + 'splits a line into pieces that encode to the same bytes as the whole line (piece size %s)', + (maxPieceChars) => { + const values = [ + 'plain', + 'a"b,c', + '\u{1F600}"x\u{1F600}\n\u{1F600}', + '', + '\uD83D\u{1F600}', + '\uDE00\u{1F600}\uDE00', + 'lone\uD83D', + 'tail\u{1F600}', + ] + const pieces = [...toCsvLinePieces(values, maxPieceChars)] + const encoder = new TextEncoder() + const pieceBytes = Buffer.concat(pieces.map((piece) => encoder.encode(piece))) + + expect(pieces.join('')).toBe(`${toCsvRow(values)}\n`) + expect(pieceBytes.equals(encoder.encode(`${toCsvRow(values)}\n`))).toBe(true) + } + ) }) diff --git a/apps/sim/lib/core/utils/csv.ts b/apps/sim/lib/core/utils/csv.ts index 3b59d3c947e..c743094f313 100644 --- a/apps/sim/lib/core/utils/csv.ts +++ b/apps/sim/lib/core/utils/csv.ts @@ -17,10 +17,65 @@ export function formatCsvValue( return String(value) } +const CSV_QUOTE_TRIGGER = /[",\n\r]/ +const CSV_LINE_PIECE_CHARS = 256 * 1024 + export function toCsvRow(values: string[]): string { return values.map(escapeCsvField).join(',') } function escapeCsvField(field: string): string { - return /[",\n\r]/.test(field) ? `"${field.replace(/"/g, '""')}"` : field + return CSV_QUOTE_TRIGGER.test(field) ? `"${field.replace(/"/g, '""')}"` : field +} + +/** + * Yields `toCsvRow(values) + '\n'` as consecutive pieces of about `maxPieceChars` source + * characters, so a caller can yield to the event loop between pieces instead of escaping a + * multi-megabyte field in one synchronous run. A piece never ends between the two halves of a + * surrogate pair, so encoding each piece separately produces the same UTF-8 bytes as encoding the + * whole line; a lone surrogate stays lone either way. A non-finite size falls back to the default + * and anything below one becomes one. + */ +export function* toCsvLinePieces( + values: readonly string[], + maxPieceChars = CSV_LINE_PIECE_CHARS +): Generator { + const pieceChars = Number.isFinite(maxPieceChars) + ? Math.max(1, Math.floor(maxPieceChars)) + : CSV_LINE_PIECE_CHARS + let pending = '' + for (let index = 0; index < values.length; index++) { + if (index > 0) pending += ',' + const field = values[index] + const quoted = CSV_QUOTE_TRIGGER.test(field) + if (quoted) pending += '"' + let start = 0 + while (start < field.length) { + let end = Math.min(start + pieceChars, field.length) + if ( + end < field.length && + isHighSurrogate(field.charCodeAt(end - 1)) && + isLowSurrogate(field.charCodeAt(end)) + ) { + end += 1 + } + const slice = field.slice(start, end) + pending += quoted ? slice.replace(/"/g, '""') : slice + start = end + if (pending.length >= pieceChars) { + yield pending + pending = '' + } + } + if (quoted) pending += '"' + } + yield `${pending}\n` +} + +function isHighSurrogate(code: number): boolean { + return code >= 0xd800 && code <= 0xdbff +} + +function isLowSurrogate(code: number): boolean { + return code >= 0xdc00 && code <= 0xdfff } diff --git a/apps/sim/lib/core/utils/event-loop.ts b/apps/sim/lib/core/utils/event-loop.ts new file mode 100644 index 00000000000..2200c539dc4 --- /dev/null +++ b/apps/sim/lib/core/utils/event-loop.ts @@ -0,0 +1,35 @@ +const CPU_SLICE_MS = 10 + +let sliceEndsAt = 0 +let turnScheduled = false +const turnQueue: Array<() => void> = [] + +/** + * Cooperative yield for CPU-bound request work. Awaits settle as microtasks, which never let the + * event loop service I/O, so a loop that only awaits holds the whole process until it finishes. + * Call this between units of work: it returns immediately while the current slice lasts and + * otherwise waits for a macrotask turn. + * + * The slice is process-wide and shared by every caller in the process, not per caller or per + * request: each turn hands one fresh slice to one waiter. Releasing every waiter together would + * let each finish its in-flight unit before I/O runs again, so a turn would grow with concurrency + * instead of staying at one slice plus one unit. + */ +export async function yieldWhenSliceExpires(): Promise { + if (performance.now() < sliceEndsAt) return + await new Promise((resolve) => { + turnQueue.push(resolve) + scheduleTurn() + }) +} + +function scheduleTurn(): void { + if (turnScheduled) return + turnScheduled = true + setImmediate(() => { + turnScheduled = false + sliceEndsAt = performance.now() + CPU_SLICE_MS + turnQueue.shift()?.() + if (turnQueue.length > 0) scheduleTurn() + }) +} diff --git a/apps/sim/lib/logs/execution/trace-secret-projection.ts b/apps/sim/lib/logs/execution/trace-secret-projection.ts index c22f28f8fe0..57c733b4231 100644 --- a/apps/sim/lib/logs/execution/trace-secret-projection.ts +++ b/apps/sim/lib/logs/execution/trace-secret-projection.ts @@ -1,6 +1,7 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' import { isPlainRecord, omit } from '@sim/utils/object' +import { yieldWhenSliceExpires } from '@/lib/core/utils/event-loop' import { isLargeArrayManifest, LARGE_ARRAY_MANIFEST_MARKER, @@ -36,7 +37,6 @@ const MAX_LARGE_VALUES = 1_024 const MAX_LARGE_VALUE_CHAIN_DEPTH = 32 const MAX_LARGE_MANIFEST_CHUNKS = MAX_LARGE_VALUES const MAX_LARGE_MANIFEST_ITEMS = MAX_CONTENT_NODES -const PROJECTION_SLICE_MS = 10 const OMIT = Symbol('omit-trace-content') const LARGE_VALUE_REF_KEYS = new Set([ LARGE_VALUE_REF_MARKER, @@ -168,38 +168,6 @@ function createProjectionContext( } } -let projectionSliceEndsAt = 0 -let projectionTurnScheduled = false -const projectionTurnQueue: Array<() => void> = [] - -/** - * Projection is CPU-bound and its awaits settle as microtasks, which never let the event loop - * service I/O, so without a macrotask yield one large trace — or a page of them projected - * concurrently — holds the whole process until it finishes. - * - * The slice is process-wide and handed to one waiting projection per event-loop turn: releasing - * them together would let each finish its in-flight field before I/O runs again, so a turn would - * grow with concurrency instead of staying at one slice plus one field. - */ -async function yieldWhenSliceExpires(): Promise { - if (performance.now() < projectionSliceEndsAt) return - await new Promise((resolve) => { - projectionTurnQueue.push(resolve) - scheduleProjectionTurn() - }) -} - -function scheduleProjectionTurn(): void { - if (projectionTurnScheduled) return - projectionTurnScheduled = true - setImmediate(() => { - projectionTurnScheduled = false - projectionSliceEndsAt = performance.now() + PROJECTION_SLICE_MS - projectionTurnQueue.shift()?.() - if (projectionTurnQueue.length > 0) scheduleProjectionTurn() - }) -} - function visitNode(state: TraversalState, depth: number): void { state.nodes += 1 if (state.nodes > MAX_CONTENT_NODES) {