From 2fc25432611c88141d0ddb3127bb29e3ad31412b Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 22 Sep 2026 17:45:15 -0700 Subject: [PATCH 1/2] fix(traces): allow explicitly skipping oversized backfill records --- apps/sim/scripts/backfill-trace-spans.test.ts | 174 +++++++++++++++--- apps/sim/scripts/backfill-trace-spans.ts | 39 +++- 2 files changed, 178 insertions(+), 35 deletions(-) diff --git a/apps/sim/scripts/backfill-trace-spans.test.ts b/apps/sim/scripts/backfill-trace-spans.test.ts index 57c0ae05979..37fc9b8b3f0 100644 --- a/apps/sim/scripts/backfill-trace-spans.test.ts +++ b/apps/sim/scripts/backfill-trace-spans.test.ts @@ -8,6 +8,7 @@ const { mockPrimaryRead, mockRead, mockInfo, + mockWarn, mockDataRead, mockTransaction, mockUpdate, @@ -17,6 +18,7 @@ const { mockPrimaryRead: vi.fn(), mockRead: vi.fn(), mockInfo: vi.fn(), + mockWarn: vi.fn(), mockDataRead: vi.fn(), mockTransaction: vi.fn(), mockUpdate: vi.fn(), @@ -48,7 +50,7 @@ vi.mock('@sim/db', () => { }) vi.mock('@sim/logger', () => ({ - createLogger: () => ({ info: mockInfo, error: vi.fn(), warn: vi.fn(), debug: vi.fn() }), + createLogger: () => ({ info: mockInfo, error: vi.fn(), warn: mockWarn, debug: vi.fn() }), })) vi.mock('@/lib/logs/execution/trace-store', () => ({ @@ -93,6 +95,7 @@ describe('backfill options', () => { concurrency: 4, maxInFlightMiB: 512, checkOnly: false, + skipOversized: false, order: 'oldest', before, }) @@ -101,6 +104,7 @@ describe('backfill options', () => { concurrency: 8, maxInFlightMiB: 512, checkOnly: true, + skipOversized: false, order: 'oldest', before, }) @@ -117,6 +121,11 @@ describe('backfill options', () => { }) }) + it('requires explicit opt-in to skip oversized records', () => { + expect(parseArgs([]).skipOversized).toBe(false) + expect(parseArgs(['--skip-oversized']).skipOversized).toBe(true) + }) + it('preserves microseconds, ordering, and cutoff when resuming a checkpoint', () => { const cursor = { version: 1, @@ -143,6 +152,8 @@ describe('backfill options', () => { '--concurrency=-1', '--concurrency=0', '--concurrency=2=3', + '--skip-oversized=false', + '--skip-oversized=true', '--unknown', ])('rejects invalid input: %s', (arg) => { expect(() => parseArgs([arg])).toThrow() @@ -252,6 +263,7 @@ describe('trace backfill', () => { concurrency: 1, maxInFlightMiB: 512, checkOnly: false, + skipOversized: false, order: 'oldest' as const, before: '2026-09-17T20:00:00.000Z', } @@ -292,6 +304,7 @@ describe('trace backfill', () => { .mockResolvedValueOnce([candidate]) await expect(backfillTraceStorage(options)).resolves.toEqual({ migrated: 1, + skippedOversized: 0, }) expect(mockPrimaryRead).toHaveBeenCalledExactlyOnceWith(0) expect(mockRead).toHaveBeenCalledTimes(7) @@ -369,23 +382,114 @@ describe('trace backfill', () => { expect(mockDataRead).toHaveBeenCalledTimes(2) expect(mockExternalize).not.toHaveBeenCalled() expect(mockTransaction).not.toHaveBeenCalled() + expect(mockWarn).not.toHaveBeenCalled() + expect(mockInfo.mock.calls.some(([message]) => message === 'Backfill checkpoint')).toBe(false) }) - it('fails before uploading when the payload grows past its reserved capacity', async () => { + it('reports oversized rows without reading them and migrates later pages when opted in', async () => { + const oversizedMetadata = { ...candidateMetadata, id: 'log-oversized' } + const oversizedSize = { + id: oversizedMetadata.id, + payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES + 1, + } + const nextMetadata = { + id: 'log-next', + startedAt: '2026-09-16T20:00:01.123456Z', + } + const nextCandidate = { ...candidate, id: nextMetadata.id, executionId: 'execution-next' } mockDataRead - .mockResolvedValueOnce([candidateMetadata]) + .mockResolvedValueOnce([candidateMetadata, oversizedMetadata]) + .mockResolvedValueOnce([candidate, oversizedSize]) .mockResolvedValueOnce([candidate]) - .mockResolvedValueOnce([ - { ...candidate, payloadBytes: candidate.payloadBytes + 1, executionData: null }, - ]) - await expect(backfillTraceStorage(options)).rejects.toMatchObject({ - cause: expect.objectContaining({ message: expect.stringContaining('grew') }), + .mockResolvedValueOnce([nextMetadata]) + .mockResolvedValueOnce([nextCandidate]) + .mockResolvedValueOnce([nextCandidate]) + + await expect( + backfillTraceStorage({ ...options, maxBatches: 2, skipOversized: true }) + ).resolves.toEqual({ migrated: 2, skippedOversized: 1 }) + + expect(mockDataRead).toHaveBeenCalledTimes(6) + expect(mockExternalize).toHaveBeenCalledTimes(2) + expect(mockUpdate).toHaveBeenCalledTimes(2) + expect(mockWarn).toHaveBeenCalledExactlyOnceWith( + 'Skipping oversized execution log; leaving data inline', + { + executionLogId: oversizedMetadata.id, + payloadBytes: oversizedSize.payloadBytes, + limitBytes: MAX_DURABLE_LARGE_VALUE_BYTES, + } + ) + expect(mockInfo).toHaveBeenCalledWith( + 'Backfill checkpoint', + expect.objectContaining(oversizedMetadata) + ) + expect(mockInfo).toHaveBeenLastCalledWith( + 'Backfill checkpoint', + expect.objectContaining(nextMetadata) + ) + expect(mockInfo).toHaveBeenCalledWith( + expect.stringContaining('migrated 2 | skipped 1 (oversized 1)'), + expect.anything() + ) + }) + + it.each(['oldest', 'newest'] as const)( + 'advances a fully oversized page in %s order without fetching or writing payloads', + async (order) => { + mockDataRead + .mockResolvedValueOnce([candidateMetadata]) + .mockResolvedValueOnce([ + { id: candidate.id, payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES + 1 }, + ]) + + await expect( + backfillTraceStorage({ ...options, order, skipOversized: true }) + ).resolves.toEqual({ migrated: 0, skippedOversized: 1 }) + + expect(mockDataRead).toHaveBeenCalledTimes(2) + expect(mockExternalize).not.toHaveBeenCalled() + expect(mockTransaction).not.toHaveBeenCalled() + expect(mockInfo).toHaveBeenCalledWith( + 'Backfill checkpoint', + expect.objectContaining(candidateMetadata) + ) + } + ) + + it('allows a payload exactly at the durable size limit with skipping enabled', async () => { + const atLimit = { ...candidate, payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES } + mockDataRead + .mockResolvedValueOnce([candidateMetadata]) + .mockResolvedValueOnce([atLimit]) + .mockResolvedValueOnce([atLimit]) + + await expect(backfillTraceStorage({ ...options, skipOversized: true })).resolves.toEqual({ + migrated: 1, + skippedOversized: 0, }) - expect(mockExternalize).not.toHaveBeenCalled() - expect(mockTransaction).not.toHaveBeenCalled() + expect(mockWarn).not.toHaveBeenCalled() + expect(mockUpdate).toHaveBeenCalledOnce() }) - it('preserves the previous checkpoint after shutdown interrupts a page', async () => { + it.each([false, true])( + 'still rejects payload growth with skipOversized=%s', + async (skipOversized) => { + mockDataRead + .mockResolvedValueOnce([candidateMetadata]) + .mockResolvedValueOnce([candidate]) + .mockResolvedValueOnce([ + { ...candidate, payloadBytes: candidate.payloadBytes + 1, executionData: null }, + ]) + await expect(backfillTraceStorage({ ...options, skipOversized })).rejects.toMatchObject({ + cause: expect.objectContaining({ message: expect.stringContaining('grew') }), + }) + expect(mockExternalize).not.toHaveBeenCalled() + expect(mockTransaction).not.toHaveBeenCalled() + } + ) + + it('preserves the previous checkpoint when shutdown interrupts a page with an oversized row', async () => { const controller = new AbortController() const cursor = { version: 1 as const, @@ -395,15 +499,26 @@ describe('trace backfill', () => { id: 'previous-log', } mockDataRead - .mockResolvedValueOnce([candidateMetadata, { ...candidateMetadata, id: 'log-2' }]) - .mockResolvedValueOnce([candidate, { ...candidate, id: 'log-2' }]) + .mockResolvedValueOnce([ + candidateMetadata, + { ...candidateMetadata, id: 'log-2' }, + { ...candidateMetadata, id: 'log-oversized' }, + ]) + .mockResolvedValueOnce([ + candidate, + { ...candidate, id: 'log-2' }, + { id: 'log-oversized', payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES + 1 }, + ]) .mockResolvedValueOnce([candidate]) mockExternalize.mockImplementationOnce(async () => { controller.abort() return { traceStoreRef: { key: 'stored-key' } } }) - await expect(backfillTraceStorage({ ...options, cursor }, controller.signal)).resolves.toEqual({ + await expect( + backfillTraceStorage({ ...options, cursor, skipOversized: true }, controller.signal) + ).resolves.toEqual({ migrated: 1, + skippedOversized: 1, }) expect(mockUpdate).toHaveBeenCalledOnce() const checkpoints = mockInfo.mock.calls.filter(([message]) => message === 'Backfill checkpoint') @@ -451,13 +566,13 @@ describe('trace backfill', () => { await started await vi.advanceTimersByTimeAsync(5000) expect(mockInfo).toHaveBeenCalledWith( - 'Progress: migrated 0 | skipped 0 | 0.0 rows/s | elapsed 5s', + 'Progress: migrated 0 | skipped 0 (oversized 0) | 0.0 rows/s | elapsed 5s', expect.objectContaining({ rssMiB: expect.any(Number) }) ) finishUpload() await run expect(mockInfo).toHaveBeenCalledWith( - 'Progress: migrated 1 | skipped 0 | 0.2 rows/s | elapsed 5s', + 'Progress: migrated 1 | skipped 0 (oversized 0) | 0.2 rows/s | elapsed 5s', expect.objectContaining({ stages: expect.objectContaining({ externalize: { calls: 1, averageMs: 5000 } }), }) @@ -486,15 +601,20 @@ describe('trace backfill', () => { expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 1000]) }) - it('does not update the log or start the next candidate after storage fails', async () => { - mockDataRead - .mockResolvedValueOnce([candidateMetadata, { ...candidateMetadata, id: 'log-2' }]) - .mockResolvedValueOnce([candidate, { ...candidate, id: 'log-2' }]) - .mockResolvedValueOnce([candidate]) - const error = new Error('storage denied') - mockExternalize.mockRejectedValueOnce(error) - await expect(backfillTraceStorage(options)).rejects.toMatchObject({ cause: error }) - expect(mockExternalize).toHaveBeenCalledOnce() - expect(mockTransaction).not.toHaveBeenCalled() - }) + it.each([false, true])( + 'still stops on storage failures with skipOversized=%s', + async (skipOversized) => { + mockDataRead + .mockResolvedValueOnce([candidateMetadata, { ...candidateMetadata, id: 'log-2' }]) + .mockResolvedValueOnce([candidate, { ...candidate, id: 'log-2' }]) + .mockResolvedValueOnce([candidate]) + const error = new Error('storage denied') + mockExternalize.mockRejectedValueOnce(error) + await expect(backfillTraceStorage({ ...options, skipOversized })).rejects.toMatchObject({ + cause: error, + }) + expect(mockExternalize).toHaveBeenCalledOnce() + expect(mockTransaction).not.toHaveBeenCalled() + } + ) }) diff --git a/apps/sim/scripts/backfill-trace-spans.ts b/apps/sim/scripts/backfill-trace-spans.ts index 0862e0aee74..b1213a11091 100644 --- a/apps/sim/scripts/backfill-trace-spans.ts +++ b/apps/sim/scripts/backfill-trace-spans.ts @@ -21,6 +21,7 @@ * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=50 * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=50 --order=newest --before= * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=50 --cursor= + * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=10 --skip-oversized --cursor= * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=200 --max-in-flight-mib=512 --cursor= * * Uses the configured database URLs and object storage. Stops on the first @@ -36,6 +37,10 @@ * exclusive start-time cutoff (defaults to startup). Each completed page logs * a cursor that preserves the cutoff and order for restart. --max-batches * limits pages examined, including pages with no eligible payloads. + * --skip-oversized explicitly leaves payloads above the durable storage limit + * inline, logs each skipped ID and size, and reports their count. Completed + * pages advance past these rows; retry them from an earlier checkpoint after + * addressing their size. All other failures still stop the run. */ import { db, dbFor } from '@sim/db' @@ -101,6 +106,7 @@ interface Options { concurrency: number maxInFlightMiB: number checkOnly: boolean + skipOversized: boolean order: z.infer before: string cursor?: BackfillCursor @@ -112,6 +118,7 @@ export function parseArgs(argv: string[]): Options { concurrency: DEFAULT_CONCURRENCY, maxInFlightMiB: DEFAULT_MAX_IN_FLIGHT_MIB, checkOnly: false, + skipOversized: false, order: 'oldest', before: new Date().toISOString(), } @@ -122,6 +129,10 @@ export function parseArgs(argv: string[]): Options { options.checkOnly = true continue } + if (arg === '--skip-oversized') { + options.skipOversized = true + continue + } const [name, value] = arg.split('=') if (name === '--order' || name === '--before' || name === '--cursor') { if (!value || arg.split('=').length !== 2) { @@ -262,14 +273,15 @@ export async function runBackfillWorkers( export async function backfillTraceStorage( options: Options, signal?: AbortSignal -): Promise<{ migrated: number }> { +): Promise<{ migrated: number; skippedOversized: number }> { await checkDatabase() logger.info('Database schema and read checks passed') - if (options.checkOnly) return { migrated: 0 } + if (options.checkOnly) return { migrated: 0, skippedOversized: 0 } const execDb = dbFor('exec') let migrated = 0 let skipped = 0 + let skippedOversized = 0 let cursor = options.cursor const direction = options.order === 'oldest' ? asc : desc const pending = and( @@ -309,7 +321,7 @@ export async function backfillTraceStorage( const elapsedMs = Date.now() - startedAt const rowsPerSecond = elapsedMs > 0 ? migrated / (elapsedMs / 1000) : 0 logger.info( - `Progress: migrated ${migrated} | skipped ${skipped} | ${rowsPerSecond.toFixed(1)} rows/s | elapsed ${formatDuration(elapsedMs)}`, + `Progress: migrated ${migrated} | skipped ${skipped} (oversized ${skippedOversized}) | ${rowsPerSecond.toFixed(1)} rows/s | elapsed ${formatDuration(elapsedMs)}`, { rssMiB: Math.round(process.memoryUsage().rss / MIB), stages: Object.fromEntries( @@ -384,14 +396,24 @@ export async function backfillTraceStorage( ) .limit(rows.length) ) + const bytesById = new Map() for (const { id, payloadBytes } of sizes) { if (payloadBytes > MAX_DURABLE_LARGE_VALUE_BYTES) { - throw new Error( - `Execution log ${id} exceeds the ${MAX_DURABLE_LARGE_VALUE_BYTES}-byte backfill limit` - ) + if (!options.skipOversized) { + throw new Error( + `Execution log ${id} is ${payloadBytes} bytes, exceeding the ${MAX_DURABLE_LARGE_VALUE_BYTES}-byte backfill limit; use --skip-oversized to leave oversized logs inline and continue` + ) + } + skippedOversized++ + logger.warn('Skipping oversized execution log; leaving data inline', { + executionLogId: id, + payloadBytes, + limitBytes: MAX_DURABLE_LARGE_VALUE_BYTES, + }) + continue } + bytesById.set(id, payloadBytes) } - const bytesById = new Map(sizes.map(({ id, payloadBytes }) => [id, payloadBytes])) const candidates = rows.flatMap(({ id }) => { const payloadBytes = bytesById.get(id) return payloadBytes === undefined ? [] : [{ id, payloadBytes }] @@ -515,7 +537,7 @@ export async function backfillTraceStorage( reportCheckpoint() } - return { migrated } + return { migrated, skippedOversized } } async function main(): Promise { @@ -526,6 +548,7 @@ async function main(): Promise { concurrency: options.concurrency, maxInFlightMiB: options.maxInFlightMiB, checkOnly: options.checkOnly, + skipOversized: options.skipOversized, }) const controller = new AbortController() const stop = (signal: NodeJS.Signals) => { From aeee328299b505efede2057ab3830deffb188cdd Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Tue, 22 Sep 2026 18:28:35 -0700 Subject: [PATCH 2/2] fix(traces): support larger execution trace archives --- apps/sim/lib/execution/payloads/limits.ts | 1 + .../payloads/materialization.server.ts | 9 +- apps/sim/lib/execution/payloads/store.test.ts | 56 ++++- apps/sim/lib/execution/payloads/store.ts | 32 ++- .../execution/trace-store-storage.test.ts | 134 ++++++++++++ .../lib/logs/execution/trace-store.test.ts | 28 ++- apps/sim/lib/logs/execution/trace-store.ts | 15 +- apps/sim/scripts/backfill-trace-spans.test.ts | 206 ++++++------------ apps/sim/scripts/backfill-trace-spans.ts | 57 ++--- 9 files changed, 336 insertions(+), 202 deletions(-) create mode 100644 apps/sim/lib/logs/execution/trace-store-storage.test.ts diff --git a/apps/sim/lib/execution/payloads/limits.ts b/apps/sim/lib/execution/payloads/limits.ts index 4166e14d66c..e06cff1d4ea 100644 --- a/apps/sim/lib/execution/payloads/limits.ts +++ b/apps/sim/lib/execution/payloads/limits.ts @@ -1,4 +1,5 @@ export const MAX_DURABLE_LARGE_VALUE_BYTES = 64 * 1024 * 1024 +export const MAX_TRACE_ARCHIVE_BYTES = 512 * 1024 * 1024 export const MAX_INLINE_MATERIALIZATION_BYTES = 16 * 1024 * 1024 export const MAX_FUNCTION_FILE_BYTES = 64 * 1024 * 1024 export const MAX_FUNCTION_INLINE_BYTES = 10 * 1024 * 1024 diff --git a/apps/sim/lib/execution/payloads/materialization.server.ts b/apps/sim/lib/execution/payloads/materialization.server.ts index 8e8035a90f7..46ce92ecdfc 100644 --- a/apps/sim/lib/execution/payloads/materialization.server.ts +++ b/apps/sim/lib/execution/payloads/materialization.server.ts @@ -75,12 +75,15 @@ function getLogger(options: ExecutionMaterializationContext): Logger { return options.logger ?? logger } -export function assertDurableLargeValueSize(size: number): void { - if (size > MAX_DURABLE_LARGE_VALUE_BYTES) { +export function assertDurableLargeValueSize( + size: number, + limitBytes = MAX_DURABLE_LARGE_VALUE_BYTES +): void { + if (size > limitBytes) { throw new ExecutionResourceLimitError({ resource: 'execution_payload_bytes', attemptedBytes: size, - limitBytes: MAX_DURABLE_LARGE_VALUE_BYTES, + limitBytes, }) } } diff --git a/apps/sim/lib/execution/payloads/store.test.ts b/apps/sim/lib/execution/payloads/store.test.ts index 60bcdcf05d2..fa830cc438c 100644 --- a/apps/sim/lib/execution/payloads/store.test.ts +++ b/apps/sim/lib/execution/payloads/store.test.ts @@ -8,12 +8,19 @@ import { clearLargeValueCacheForTests, materializeLargeValueRefSync, } from '@/lib/execution/payloads/cache' -import { MAX_DURABLE_LARGE_VALUE_BYTES } from '@/lib/execution/payloads/limits' +import { + MAX_DURABLE_LARGE_VALUE_BYTES, + MAX_TRACE_ARCHIVE_BYTES, +} from '@/lib/execution/payloads/limits' import { readLargeValueRefFromStorage, readUserFileContent, } from '@/lib/execution/payloads/materialization.server' -import { materializeLargeValueRef, storeLargeValue } from '@/lib/execution/payloads/store' +import { + materializeLargeValueRef, + storeExecutionTraceArchive, + storeLargeValue, +} from '@/lib/execution/payloads/store' import { EXECUTION_RESOURCE_LIMIT_CODE } from '@/lib/execution/resource-errors' const { @@ -350,6 +357,51 @@ describe('large execution payload store', () => { requireDurable: true, }) ).rejects.toMatchObject({ code: EXECUTION_RESOURCE_LIMIT_CODE }) + expect(mockUploadFile).not.toHaveBeenCalled() + }) + + it('admits a trace archive at its separate size cap with durable ownership', async () => { + const ref = await storeExecutionTraceArchive({}, '{}', MAX_TRACE_ARCHIVE_BYTES, { + workspaceId: 'workspace-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + userId: 'user-1', + }) + + expect(mockUploadFile).toHaveBeenCalledOnce() + expect(mockRegisterLargeValueOwner).toHaveBeenCalledWith( + expect.objectContaining({ key: ref.key, size: MAX_TRACE_ARCHIVE_BYTES }), + [] + ) + expect(materializeLargeValueRefSync(ref, { executionId: 'execution-1' })).toBeUndefined() + }) + + it('rejects archives above the trace cap before upload or metadata writes', async () => { + await expect( + storeExecutionTraceArchive({}, '{}', MAX_TRACE_ARCHIVE_BYTES + 1, { + workspaceId: 'workspace-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + userId: 'user-1', + }) + ).rejects.toMatchObject({ code: EXECUTION_RESOURCE_LIMIT_CODE }) + expect(mockUploadFile).not.toHaveBeenCalled() + expect(mockRegisterLargeValueOwner).not.toHaveBeenCalled() + }) + + it('requires durable storage for trace archives even if the caller disables it', async () => { + mockUploadFile.mockRejectedValueOnce(new Error('storage unavailable')) + + await expect( + storeExecutionTraceArchive({}, '{}', 2, { + workspaceId: 'workspace-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + userId: 'user-1', + requireDurable: false, + }) + ).rejects.toThrow('storage unavailable') + expect(mockRegisterLargeValueOwner).not.toHaveBeenCalled() }) it('bounds explicit server-side materialization', async () => { diff --git a/apps/sim/lib/execution/payloads/store.ts b/apps/sim/lib/execution/payloads/store.ts index 0cb1e14385e..7b67fa8094d 100644 --- a/apps/sim/lib/execution/payloads/store.ts +++ b/apps/sim/lib/execution/payloads/store.ts @@ -9,6 +9,10 @@ import { type LargeValueKind, type LargeValueRef, } from '@/lib/execution/payloads/large-value-ref' +import { + MAX_DURABLE_LARGE_VALUE_BYTES, + MAX_TRACE_ARCHIVE_BYTES, +} from '@/lib/execution/payloads/limits' import { assertDurableLargeValueSize, assertInlineMaterializationSize, @@ -164,7 +168,33 @@ export async function storeLargeValue( size: number, context: LargeValueStoreContext ): Promise { - assertDurableLargeValueSize(size) + return persistLargeValue(value, json, size, context, MAX_DURABLE_LARGE_VALUE_BYTES) +} + +/** Stores a completed execution archive with a larger cap than individual workflow values. */ +export async function storeExecutionTraceArchive( + value: Record, + json: string, + size: number, + context: LargeValueStoreContext +): Promise { + return persistLargeValue( + value, + json, + size, + { ...context, requireDurable: true }, + MAX_TRACE_ARCHIVE_BYTES + ) +} + +async function persistLargeValue( + value: unknown, + json: string, + size: number, + context: LargeValueStoreContext, + limitBytes: number +): Promise { + assertDurableLargeValueSize(size, limitBytes) const referencedKeys = collectLargeValueKeys(value) const id = `lv_${generateShortId(12)}` let key = await persistValue(id, json, context) diff --git a/apps/sim/lib/logs/execution/trace-store-storage.test.ts b/apps/sim/lib/logs/execution/trace-store-storage.test.ts new file mode 100644 index 00000000000..00ebdc1cef7 --- /dev/null +++ b/apps/sim/lib/logs/execution/trace-store-storage.test.ts @@ -0,0 +1,134 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { clearLargeValueCacheForTests } from '@/lib/execution/payloads/cache' +import { + MAX_DURABLE_LARGE_VALUE_BYTES, + MAX_TRACE_ARCHIVE_BYTES, +} from '@/lib/execution/payloads/limits' +import { storeLargeValue } from '@/lib/execution/payloads/store' +import { EXECUTION_RESOURCE_LIMIT_CODE } from '@/lib/execution/resource-errors' +import { + externalizeExecutionData, + materializeExecutionData, + TRACE_STORE_REF_KEY, +} from '@/lib/logs/execution/trace-store' + +const { mockUploadFile, mockDownloadFile, mockRegisterOwner, mockAddReference } = vi.hoisted( + () => ({ + mockUploadFile: vi.fn(), + mockDownloadFile: vi.fn(), + mockRegisterOwner: vi.fn(), + mockAddReference: vi.fn(), + }) +) + +/** Scale the two caps down to exercise real serialization and storage reads with small fixtures. */ +vi.mock('@/lib/execution/payloads/limits', async (importOriginal) => ({ + ...(await importOriginal()), + MAX_DURABLE_LARGE_VALUE_BYTES: 1024, + MAX_TRACE_ARCHIVE_BYTES: 4096, +})) + +vi.mock('@/lib/uploads', () => ({ + StorageService: { uploadFile: mockUploadFile, downloadFile: mockDownloadFile }, +})) + +vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({ + registerLargeValueOwner: mockRegisterOwner, + addLargeValueReference: mockAddReference, +})) + +const CONTEXT = { + workspaceId: 'workspace-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + userId: 'user-1', +} + +beforeEach(() => { + vi.clearAllMocks() + clearLargeValueCacheForTests() + mockRegisterOwner.mockResolvedValue(true) + mockUploadFile.mockImplementation(async ({ customKey, file }) => { + mockDownloadFile.mockResolvedValue(file) + return { key: customKey } + }) +}) + +describe('trace archive storage round trip', () => { + it('uploads an archive above the ordinary value cap and reads it back after a cache miss', async () => { + const data = { + traceSpans: [{ id: 'span-1', output: 'é'.repeat(MAX_DURABLE_LARGE_VALUE_BYTES) }], + traceSpanCount: 1, + hasTraceSpans: true, + } + const json = JSON.stringify(data) + const size = Buffer.byteLength(json, 'utf8') + expect(size).toBeGreaterThan(MAX_DURABLE_LARGE_VALUE_BYTES) + expect(size).toBeLessThan(MAX_TRACE_ARCHIVE_BYTES) + + await expect(storeLargeValue(data, json, size, CONTEXT)).rejects.toMatchObject({ + code: EXECUTION_RESOURCE_LIMIT_CODE, + }) + expect(mockUploadFile).not.toHaveBeenCalled() + + const slim = await externalizeExecutionData(data, CONTEXT, { throwOnError: true }) + expect(slim).toEqual({ + [TRACE_STORE_REF_KEY]: expect.objectContaining({ size, key: expect.any(String) }), + traceSpanCount: 1, + hasTraceSpans: true, + }) + expect(mockRegisterOwner).toHaveBeenCalledWith( + expect.objectContaining({ + workspaceId: CONTEXT.workspaceId, + workflowId: CONTEXT.workflowId, + executionId: CONTEXT.executionId, + size, + }), + [] + ) + clearLargeValueCacheForTests() + + await expect(materializeExecutionData(slim, CONTEXT)).resolves.toEqual(data) + expect(mockDownloadFile).toHaveBeenCalledExactlyOnceWith({ + key: expect.any(String), + context: 'execution', + maxBytes: size, + }) + expect(mockAddReference).not.toHaveBeenCalled() + }) + + it('rejects an over-limit archive before uploading it', async () => { + await expect( + externalizeExecutionData( + { traceSpans: [{ output: 'x'.repeat(MAX_TRACE_ARCHIVE_BYTES) }] }, + CONTEXT, + { throwOnError: true } + ) + ).rejects.toMatchObject({ code: EXECUTION_RESOURCE_LIMIT_CODE }) + expect(mockUploadFile).not.toHaveBeenCalled() + expect(mockRegisterOwner).not.toHaveBeenCalled() + }) + + it('bounds stored archive reads even if a reference declares a larger size', async () => { + const data = { traceSpans: [], hasTraceSpans: false } + const slim = await externalizeExecutionData(data, CONTEXT, { throwOnError: true }) + clearLargeValueCacheForTests() + + await expect( + materializeExecutionData( + { + ...slim, + [TRACE_STORE_REF_KEY]: { + ...(slim[TRACE_STORE_REF_KEY] as Record), + size: MAX_TRACE_ARCHIVE_BYTES + 1, + }, + }, + CONTEXT + ) + ).resolves.toEqual({ hasTraceSpans: false }) + expect(mockDownloadFile).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/logs/execution/trace-store.test.ts b/apps/sim/lib/logs/execution/trace-store.test.ts index 5707e24be19..614147802ab 100644 --- a/apps/sim/lib/logs/execution/trace-store.test.ts +++ b/apps/sim/lib/logs/execution/trace-store.test.ts @@ -3,13 +3,17 @@ */ import { beforeEach, describe, expect, it, vi } from 'vitest' -const { decryptSecretMock, materializeLargeValueRefMock, storeLargeValueMock, mockLogger } = - vi.hoisted(() => ({ - decryptSecretMock: vi.fn(), - materializeLargeValueRefMock: vi.fn(), - storeLargeValueMock: vi.fn(), - mockLogger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, - })) +const { + decryptSecretMock, + materializeLargeValueRefMock, + storeExecutionTraceArchiveMock, + mockLogger, +} = vi.hoisted(() => ({ + decryptSecretMock: vi.fn(), + materializeLargeValueRefMock: vi.fn(), + storeExecutionTraceArchiveMock: vi.fn(), + mockLogger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, +})) vi.mock('@sim/logger', () => ({ createLogger: () => mockLogger, @@ -21,7 +25,7 @@ vi.mock('@/lib/core/security/encryption', () => ({ vi.mock('@/lib/execution/payloads/store', () => ({ materializeLargeValueRef: materializeLargeValueRefMock, - storeLargeValue: storeLargeValueMock, + storeExecutionTraceArchive: storeExecutionTraceArchiveMock, })) import { @@ -54,7 +58,7 @@ describe('execution data storage', () => { it('propagates the original storage failure for strict backfills', async () => { const cause = new Error('column "size_bytes" does not exist') const error = new Error('Failed query', { cause }) - storeLargeValueMock.mockRejectedValueOnce(error) + storeExecutionTraceArchiveMock.mockRejectedValueOnce(error) await expect( externalizeExecutionData({ traceSpans: [] }, CONTEXT, { throwOnError: true }) @@ -70,12 +74,12 @@ describe('execution data storage', () => { { throwOnError: true } ) ).rejects.toThrow('Trace storage requires workspaceId, workflowId, and userId') - expect(storeLargeValueMock).not.toHaveBeenCalled() + expect(storeExecutionTraceArchiveMock).not.toHaveBeenCalled() }) it('preserves inline completion data and logs the underlying database error', async () => { const data = { traceSpans: [] } - storeLargeValueMock.mockRejectedValueOnce( + storeExecutionTraceArchiveMock.mockRejectedValueOnce( new Error('Failed query\nparams: private-payload', { cause: new Error('permission denied for table workspace_files'), }) @@ -100,7 +104,7 @@ describe('execution data storage', () => { executionId: 'execution-1', preview: { unsafe: 'must-not-remain-inline' }, } as const - storeLargeValueMock.mockResolvedValue(ref) + storeExecutionTraceArchiveMock.mockResolvedValue(ref) materializeLargeValueRefMock.mockRejectedValue(new Error('object unavailable')) const slim = await externalizeExecutionData( diff --git a/apps/sim/lib/logs/execution/trace-store.ts b/apps/sim/lib/logs/execution/trace-store.ts index 996556ee7d7..7162ecc5456 100644 --- a/apps/sim/lib/logs/execution/trace-store.ts +++ b/apps/sim/lib/logs/execution/trace-store.ts @@ -2,7 +2,11 @@ import { createLogger } from '@sim/logger' import { describeError, toError } from '@sim/utils/errors' import { isRecordLike, omit } from '@sim/utils/object' import { isLargeValueRef } from '@/lib/execution/payloads/large-value-ref' -import { materializeLargeValueRef, storeLargeValue } from '@/lib/execution/payloads/store' +import { MAX_TRACE_ARCHIVE_BYTES } from '@/lib/execution/payloads/limits' +import { + materializeLargeValueRef, + storeExecutionTraceArchive, +} from '@/lib/execution/payloads/store' import { FunctionalOutputsUnavailableError } from '@/lib/logs/execution/functional-outputs' import { projectTraceSpansForSecrets } from '@/lib/logs/execution/trace-secret-projection' import type { TraceSpan } from '@/lib/logs/types' @@ -254,15 +258,12 @@ export async function externalizeExecutionData( const json = JSON.stringify(executionData) const size = Buffer.byteLength(json, 'utf8') - // storeLargeValue persists to the execution bucket with a conforming key and - // registers owner + dependency closure (trace -> nested span large values), - // so GC keeps nested children alive while this run's log row exists. - const ref = await storeLargeValue(executionData, json, size, { + /** Register the archive owner and dependencies so nested span values survive with the log. */ + const ref = await storeExecutionTraceArchive(executionData, json, size, { workspaceId, workflowId, executionId, userId, - requireDurable: true, }) const { preview: _preview, ...slimRef } = ref @@ -316,7 +317,7 @@ export async function materializeExecutionData( workspaceId: context.workspaceId, workflowId, executionId: context.executionId, - maxBytes: ref.size, + maxBytes: Math.min(ref.size, MAX_TRACE_ARCHIVE_BYTES), // Read-only: the value is already referenced by its own execution; don't // re-register (or fail) on every view/export. trackReference: false, diff --git a/apps/sim/scripts/backfill-trace-spans.test.ts b/apps/sim/scripts/backfill-trace-spans.test.ts index 37fc9b8b3f0..c8fe20caaa4 100644 --- a/apps/sim/scripts/backfill-trace-spans.test.ts +++ b/apps/sim/scripts/backfill-trace-spans.test.ts @@ -2,13 +2,15 @@ * @vitest-environment node */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -import { MAX_DURABLE_LARGE_VALUE_BYTES } from '@/lib/execution/payloads/limits' +import { + MAX_DURABLE_LARGE_VALUE_BYTES, + MAX_TRACE_ARCHIVE_BYTES, +} from '@/lib/execution/payloads/limits' const { mockPrimaryRead, mockRead, mockInfo, - mockWarn, mockDataRead, mockTransaction, mockUpdate, @@ -18,7 +20,6 @@ const { mockPrimaryRead: vi.fn(), mockRead: vi.fn(), mockInfo: vi.fn(), - mockWarn: vi.fn(), mockDataRead: vi.fn(), mockTransaction: vi.fn(), mockUpdate: vi.fn(), @@ -50,7 +51,7 @@ vi.mock('@sim/db', () => { }) vi.mock('@sim/logger', () => ({ - createLogger: () => ({ info: mockInfo, error: vi.fn(), warn: mockWarn, debug: vi.fn() }), + createLogger: () => ({ info: mockInfo, error: vi.fn(), warn: vi.fn(), debug: vi.fn() }), })) vi.mock('@/lib/logs/execution/trace-store', () => ({ @@ -95,7 +96,6 @@ describe('backfill options', () => { concurrency: 4, maxInFlightMiB: 512, checkOnly: false, - skipOversized: false, order: 'oldest', before, }) @@ -104,7 +104,6 @@ describe('backfill options', () => { concurrency: 8, maxInFlightMiB: 512, checkOnly: true, - skipOversized: false, order: 'oldest', before, }) @@ -121,11 +120,6 @@ describe('backfill options', () => { }) }) - it('requires explicit opt-in to skip oversized records', () => { - expect(parseArgs([]).skipOversized).toBe(false) - expect(parseArgs(['--skip-oversized']).skipOversized).toBe(true) - }) - it('preserves microseconds, ordering, and cutoff when resuming a checkpoint', () => { const cursor = { version: 1, @@ -152,8 +146,6 @@ describe('backfill options', () => { '--concurrency=-1', '--concurrency=0', '--concurrency=2=3', - '--skip-oversized=false', - '--skip-oversized=true', '--unknown', ])('rejects invalid input: %s', (arg) => { expect(() => parseArgs([arg])).toThrow() @@ -263,7 +255,6 @@ describe('trace backfill', () => { concurrency: 1, maxInFlightMiB: 512, checkOnly: false, - skipOversized: false, order: 'oldest' as const, before: '2026-09-17T20:00:00.000Z', } @@ -304,7 +295,6 @@ describe('trace backfill', () => { .mockResolvedValueOnce([candidate]) await expect(backfillTraceStorage(options)).resolves.toEqual({ migrated: 1, - skippedOversized: 0, }) expect(mockPrimaryRead).toHaveBeenCalledExactlyOnceWith(0) expect(mockRead).toHaveBeenCalledTimes(7) @@ -375,81 +365,28 @@ describe('trace backfill', () => { { ...candidate, executionData: null, - payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES + 1, + payloadBytes: MAX_TRACE_ARCHIVE_BYTES + 1, }, ]) - await expect(backfillTraceStorage(options)).rejects.toThrow('backfill limit') + await expect(backfillTraceStorage(options)).rejects.toThrow('trace archive limit') expect(mockDataRead).toHaveBeenCalledTimes(2) expect(mockExternalize).not.toHaveBeenCalled() expect(mockTransaction).not.toHaveBeenCalled() - expect(mockWarn).not.toHaveBeenCalled() - expect(mockInfo.mock.calls.some(([message]) => message === 'Backfill checkpoint')).toBe(false) }) - it('reports oversized rows without reading them and migrates later pages when opted in', async () => { - const oversizedMetadata = { ...candidateMetadata, id: 'log-oversized' } - const oversizedSize = { - id: oversizedMetadata.id, - payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES + 1, - } - const nextMetadata = { - id: 'log-next', - startedAt: '2026-09-16T20:00:01.123456Z', - } - const nextCandidate = { ...candidate, id: nextMetadata.id, executionId: 'execution-next' } - mockDataRead - .mockResolvedValueOnce([candidateMetadata, oversizedMetadata]) - .mockResolvedValueOnce([candidate, oversizedSize]) - .mockResolvedValueOnce([candidate]) - .mockResolvedValueOnce([nextMetadata]) - .mockResolvedValueOnce([nextCandidate]) - .mockResolvedValueOnce([nextCandidate]) - - await expect( - backfillTraceStorage({ ...options, maxBatches: 2, skipOversized: true }) - ).resolves.toEqual({ migrated: 2, skippedOversized: 1 }) - - expect(mockDataRead).toHaveBeenCalledTimes(6) - expect(mockExternalize).toHaveBeenCalledTimes(2) - expect(mockUpdate).toHaveBeenCalledTimes(2) - expect(mockWarn).toHaveBeenCalledExactlyOnceWith( - 'Skipping oversized execution log; leaving data inline', - { - executionLogId: oversizedMetadata.id, - payloadBytes: oversizedSize.payloadBytes, - limitBytes: MAX_DURABLE_LARGE_VALUE_BYTES, - } - ) - expect(mockInfo).toHaveBeenCalledWith( - 'Backfill checkpoint', - expect.objectContaining(oversizedMetadata) - ) - expect(mockInfo).toHaveBeenLastCalledWith( - 'Backfill checkpoint', - expect.objectContaining(nextMetadata) - ) - expect(mockInfo).toHaveBeenCalledWith( - expect.stringContaining('migrated 2 | skipped 1 (oversized 1)'), - expect.anything() - ) - }) - - it.each(['oldest', 'newest'] as const)( - 'advances a fully oversized page in %s order without fetching or writing payloads', - async (order) => { + it.each([MAX_DURABLE_LARGE_VALUE_BYTES + 1, MAX_TRACE_ARCHIVE_BYTES])( + 'uploads and commits a %i-byte archive without skipping it', + async (payloadBytes) => { + const large = { ...candidate, payloadBytes } mockDataRead .mockResolvedValueOnce([candidateMetadata]) - .mockResolvedValueOnce([ - { id: candidate.id, payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES + 1 }, - ]) - - await expect( - backfillTraceStorage({ ...options, order, skipOversized: true }) - ).resolves.toEqual({ migrated: 0, skippedOversized: 1 }) + .mockResolvedValueOnce([large]) + .mockResolvedValueOnce([large]) - expect(mockDataRead).toHaveBeenCalledTimes(2) - expect(mockExternalize).not.toHaveBeenCalled() - expect(mockTransaction).not.toHaveBeenCalled() + await expect(backfillTraceStorage(options)).resolves.toEqual({ migrated: 1 }) + expect(mockExternalize).toHaveBeenCalledOnce() + expect(mockUpdate).toHaveBeenCalledOnce() + expect(mockReplaceReferences).toHaveBeenCalledOnce() expect(mockInfo).toHaveBeenCalledWith( 'Backfill checkpoint', expect.objectContaining(candidateMetadata) @@ -457,39 +394,46 @@ describe('trace backfill', () => { } ) - it('allows a payload exactly at the durable size limit with skipping enabled', async () => { - const atLimit = { ...candidate, payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES } + it('requires enough byte budget before fetching a larger archive and preserves the checkpoint', async () => { + const cursor = { + version: 1 as const, + order: options.order, + before: options.before, + startedAt: '2025-01-01T00:00:00.123456Z', + id: 'previous-log', + } mockDataRead .mockResolvedValueOnce([candidateMetadata]) - .mockResolvedValueOnce([atLimit]) - .mockResolvedValueOnce([atLimit]) + .mockResolvedValueOnce([ + { id: candidate.id, payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES + 1 }, + ]) - await expect(backfillTraceStorage({ ...options, skipOversized: true })).resolves.toEqual({ - migrated: 1, - skippedOversized: 0, - }) - expect(mockWarn).not.toHaveBeenCalled() - expect(mockUpdate).toHaveBeenCalledOnce() + await expect(backfillTraceStorage({ ...options, maxInFlightMiB: 64, cursor })).rejects.toThrow( + 'increase the byte budget' + ) + expect(mockDataRead).toHaveBeenCalledTimes(2) + expect(mockExternalize).not.toHaveBeenCalled() + expect(mockTransaction).not.toHaveBeenCalled() + const checkpoints = mockInfo.mock.calls.filter(([message]) => message === 'Backfill checkpoint') + expect(checkpoints.length).toBeGreaterThan(0) + expect(checkpoints.every(([, value]) => value.id === cursor.id)).toBe(true) }) - it.each([false, true])( - 'still rejects payload growth with skipOversized=%s', - async (skipOversized) => { - mockDataRead - .mockResolvedValueOnce([candidateMetadata]) - .mockResolvedValueOnce([candidate]) - .mockResolvedValueOnce([ - { ...candidate, payloadBytes: candidate.payloadBytes + 1, executionData: null }, - ]) - await expect(backfillTraceStorage({ ...options, skipOversized })).rejects.toMatchObject({ - cause: expect.objectContaining({ message: expect.stringContaining('grew') }), - }) - expect(mockExternalize).not.toHaveBeenCalled() - expect(mockTransaction).not.toHaveBeenCalled() - } - ) + it('fails before uploading when the payload grows past its reserved capacity', async () => { + mockDataRead + .mockResolvedValueOnce([candidateMetadata]) + .mockResolvedValueOnce([candidate]) + .mockResolvedValueOnce([ + { ...candidate, payloadBytes: candidate.payloadBytes + 1, executionData: null }, + ]) + await expect(backfillTraceStorage(options)).rejects.toMatchObject({ + cause: expect.objectContaining({ message: expect.stringContaining('grew') }), + }) + expect(mockExternalize).not.toHaveBeenCalled() + expect(mockTransaction).not.toHaveBeenCalled() + }) - it('preserves the previous checkpoint when shutdown interrupts a page with an oversized row', async () => { + it('preserves the previous checkpoint after shutdown interrupts a page', async () => { const controller = new AbortController() const cursor = { version: 1 as const, @@ -499,26 +443,15 @@ describe('trace backfill', () => { id: 'previous-log', } mockDataRead - .mockResolvedValueOnce([ - candidateMetadata, - { ...candidateMetadata, id: 'log-2' }, - { ...candidateMetadata, id: 'log-oversized' }, - ]) - .mockResolvedValueOnce([ - candidate, - { ...candidate, id: 'log-2' }, - { id: 'log-oversized', payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES + 1 }, - ]) + .mockResolvedValueOnce([candidateMetadata, { ...candidateMetadata, id: 'log-2' }]) + .mockResolvedValueOnce([candidate, { ...candidate, id: 'log-2' }]) .mockResolvedValueOnce([candidate]) mockExternalize.mockImplementationOnce(async () => { controller.abort() return { traceStoreRef: { key: 'stored-key' } } }) - await expect( - backfillTraceStorage({ ...options, cursor, skipOversized: true }, controller.signal) - ).resolves.toEqual({ + await expect(backfillTraceStorage({ ...options, cursor }, controller.signal)).resolves.toEqual({ migrated: 1, - skippedOversized: 1, }) expect(mockUpdate).toHaveBeenCalledOnce() const checkpoints = mockInfo.mock.calls.filter(([message]) => message === 'Backfill checkpoint') @@ -566,13 +499,13 @@ describe('trace backfill', () => { await started await vi.advanceTimersByTimeAsync(5000) expect(mockInfo).toHaveBeenCalledWith( - 'Progress: migrated 0 | skipped 0 (oversized 0) | 0.0 rows/s | elapsed 5s', + 'Progress: migrated 0 | skipped 0 | 0.0 rows/s | elapsed 5s', expect.objectContaining({ rssMiB: expect.any(Number) }) ) finishUpload() await run expect(mockInfo).toHaveBeenCalledWith( - 'Progress: migrated 1 | skipped 0 (oversized 0) | 0.2 rows/s | elapsed 5s', + 'Progress: migrated 1 | skipped 0 | 0.2 rows/s | elapsed 5s', expect.objectContaining({ stages: expect.objectContaining({ externalize: { calls: 1, averageMs: 5000 } }), }) @@ -601,20 +534,15 @@ describe('trace backfill', () => { expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 1000]) }) - it.each([false, true])( - 'still stops on storage failures with skipOversized=%s', - async (skipOversized) => { - mockDataRead - .mockResolvedValueOnce([candidateMetadata, { ...candidateMetadata, id: 'log-2' }]) - .mockResolvedValueOnce([candidate, { ...candidate, id: 'log-2' }]) - .mockResolvedValueOnce([candidate]) - const error = new Error('storage denied') - mockExternalize.mockRejectedValueOnce(error) - await expect(backfillTraceStorage({ ...options, skipOversized })).rejects.toMatchObject({ - cause: error, - }) - expect(mockExternalize).toHaveBeenCalledOnce() - expect(mockTransaction).not.toHaveBeenCalled() - } - ) + it('does not update the log or start the next candidate after storage fails', async () => { + mockDataRead + .mockResolvedValueOnce([candidateMetadata, { ...candidateMetadata, id: 'log-2' }]) + .mockResolvedValueOnce([candidate, { ...candidate, id: 'log-2' }]) + .mockResolvedValueOnce([candidate]) + const error = new Error('storage denied') + mockExternalize.mockRejectedValueOnce(error) + await expect(backfillTraceStorage(options)).rejects.toMatchObject({ cause: error }) + expect(mockExternalize).toHaveBeenCalledOnce() + expect(mockTransaction).not.toHaveBeenCalled() + }) }) diff --git a/apps/sim/scripts/backfill-trace-spans.ts b/apps/sim/scripts/backfill-trace-spans.ts index b1213a11091..22e9b89b60e 100644 --- a/apps/sim/scripts/backfill-trace-spans.ts +++ b/apps/sim/scripts/backfill-trace-spans.ts @@ -21,14 +21,15 @@ * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=50 * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=50 --order=newest --before= * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=50 --cursor= - * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=10 --skip-oversized --cursor= * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=200 --max-in-flight-mib=512 --cursor= * * Uses the configured database URLs and object storage. Stops on the first * failure after draining active workers; reruns skip committed rows. * Reports migrated rows, throughput, and elapsed time every five seconds. * Concurrency accepts 1–512 workers; it does not set a rows-per-second target. - * Payload reads share a serialized-byte budget (512 MiB by default). Parsed + * Trace archives are capped at 512 MiB; individual workflow values keep their + * separate 64 MiB cap. Payload reads share a serialized-byte budget (512 MiB + * by default), so larger archives reduce effective concurrency. Parsed * objects, serialization copies, and the shared cache use additional memory. * Reports RSS and cumulative average timings per stage without counting rows. * SIGINT/SIGTERM stop scheduling and drain active writes. A partial page never @@ -37,10 +38,6 @@ * exclusive start-time cutoff (defaults to startup). Each completed page logs * a cursor that preserves the cutoff and order for restart. --max-batches * limits pages examined, including pages with no eligible payloads. - * --skip-oversized explicitly leaves payloads above the durable storage limit - * inline, logs each skipped ID and size, and reports their count. Completed - * pages advance past these rows; retry them from an earlier checkpoint after - * addressing their size. All other failures still stop the run. */ import { db, dbFor } from '@sim/db' @@ -61,7 +58,7 @@ import { collectLargeValueReferenceKeys, replaceLargeValueReferenceKeysWithClient, } from '@/lib/execution/payloads/large-value-metadata' -import { MAX_DURABLE_LARGE_VALUE_BYTES } from '@/lib/execution/payloads/limits' +import { MAX_TRACE_ARCHIVE_BYTES } from '@/lib/execution/payloads/limits' import { externalizeExecutionData, stripSpanCosts, @@ -106,7 +103,6 @@ interface Options { concurrency: number maxInFlightMiB: number checkOnly: boolean - skipOversized: boolean order: z.infer before: string cursor?: BackfillCursor @@ -118,7 +114,6 @@ export function parseArgs(argv: string[]): Options { concurrency: DEFAULT_CONCURRENCY, maxInFlightMiB: DEFAULT_MAX_IN_FLIGHT_MIB, checkOnly: false, - skipOversized: false, order: 'oldest', before: new Date().toISOString(), } @@ -129,10 +124,6 @@ export function parseArgs(argv: string[]): Options { options.checkOnly = true continue } - if (arg === '--skip-oversized') { - options.skipOversized = true - continue - } const [name, value] = arg.split('=') if (name === '--order' || name === '--before' || name === '--cursor') { if (!value || arg.split('=').length !== 2) { @@ -176,10 +167,7 @@ export function parseArgs(argv: string[]): Options { if (options.concurrency > MAX_CONCURRENCY) { throw new Error(`--concurrency must be between 1 and ${MAX_CONCURRENCY}`) } - if ( - options.maxInFlightMiB < MAX_DURABLE_LARGE_VALUE_BYTES / MIB || - options.maxInFlightMiB > 4096 - ) { + if (options.maxInFlightMiB < 64 || options.maxInFlightMiB > 4096) { throw new Error('--max-in-flight-mib must be between 64 and 4096') } if (options.cursor) { @@ -273,15 +261,14 @@ export async function runBackfillWorkers( export async function backfillTraceStorage( options: Options, signal?: AbortSignal -): Promise<{ migrated: number; skippedOversized: number }> { +): Promise<{ migrated: number }> { await checkDatabase() logger.info('Database schema and read checks passed') - if (options.checkOnly) return { migrated: 0, skippedOversized: 0 } + if (options.checkOnly) return { migrated: 0 } const execDb = dbFor('exec') let migrated = 0 let skipped = 0 - let skippedOversized = 0 let cursor = options.cursor const direction = options.order === 'oldest' ? asc : desc const pending = and( @@ -321,7 +308,7 @@ export async function backfillTraceStorage( const elapsedMs = Date.now() - startedAt const rowsPerSecond = elapsedMs > 0 ? migrated / (elapsedMs / 1000) : 0 logger.info( - `Progress: migrated ${migrated} | skipped ${skipped} (oversized ${skippedOversized}) | ${rowsPerSecond.toFixed(1)} rows/s | elapsed ${formatDuration(elapsedMs)}`, + `Progress: migrated ${migrated} | skipped ${skipped} | ${rowsPerSecond.toFixed(1)} rows/s | elapsed ${formatDuration(elapsedMs)}`, { rssMiB: Math.round(process.memoryUsage().rss / MIB), stages: Object.fromEntries( @@ -396,24 +383,19 @@ export async function backfillTraceStorage( ) .limit(rows.length) ) - const bytesById = new Map() for (const { id, payloadBytes } of sizes) { - if (payloadBytes > MAX_DURABLE_LARGE_VALUE_BYTES) { - if (!options.skipOversized) { - throw new Error( - `Execution log ${id} is ${payloadBytes} bytes, exceeding the ${MAX_DURABLE_LARGE_VALUE_BYTES}-byte backfill limit; use --skip-oversized to leave oversized logs inline and continue` - ) - } - skippedOversized++ - logger.warn('Skipping oversized execution log; leaving data inline', { - executionLogId: id, - payloadBytes, - limitBytes: MAX_DURABLE_LARGE_VALUE_BYTES, - }) - continue + if (payloadBytes > MAX_TRACE_ARCHIVE_BYTES) { + throw new Error( + `Execution log ${id} is ${payloadBytes} bytes, exceeding the ${MAX_TRACE_ARCHIVE_BYTES}-byte trace archive limit` + ) + } + if (payloadBytes > options.maxInFlightMiB * MIB) { + throw new Error( + `Execution log ${id} is ${payloadBytes} bytes, exceeding --max-in-flight-mib=${options.maxInFlightMiB}; increase the byte budget to migrate it` + ) } - bytesById.set(id, payloadBytes) } + const bytesById = new Map(sizes.map(({ id, payloadBytes }) => [id, payloadBytes])) const candidates = rows.flatMap(({ id }) => { const payloadBytes = bytesById.get(id) return payloadBytes === undefined ? [] : [{ id, payloadBytes }] @@ -537,7 +519,7 @@ export async function backfillTraceStorage( reportCheckpoint() } - return { migrated, skippedOversized } + return { migrated } } async function main(): Promise { @@ -548,7 +530,6 @@ async function main(): Promise { concurrency: options.concurrency, maxInFlightMiB: options.maxInFlightMiB, checkOnly: options.checkOnly, - skipOversized: options.skipOversized, }) const controller = new AbortController() const stop = (signal: NodeJS.Signals) => {