From ac9131959f7c7026d244722d44a01dc6914c6413 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Thu, 17 Sep 2026 13:56:25 -0700 Subject: [PATCH 1/3] wip --- apps/sim/lib/execution/payloads/store.test.ts | 14 + apps/sim/lib/execution/payloads/store.ts | 4 +- .../lib/logs/execution/trace-store.test.ts | 37 ++ apps/sim/lib/logs/execution/trace-store.ts | 16 +- apps/sim/scripts/backfill-trace-spans.test.ts | 283 +++++++++++ apps/sim/scripts/backfill-trace-spans.ts | 442 +++++++++++++----- 6 files changed, 683 insertions(+), 113 deletions(-) create mode 100644 apps/sim/scripts/backfill-trace-spans.test.ts diff --git a/apps/sim/lib/execution/payloads/store.test.ts b/apps/sim/lib/execution/payloads/store.test.ts index a283778b7a2..60bcdcf05d2 100644 --- a/apps/sim/lib/execution/payloads/store.test.ts +++ b/apps/sim/lib/execution/payloads/store.test.ts @@ -248,6 +248,20 @@ describe('large execution payload store', () => { ).rejects.toThrow('Failed to persist large execution value: storage down') }) + it('preserves the database cause when metadata persistence fails after an upload', async () => { + const cause = new Error('permission denied for table workspace_files') + const error = new Error('Failed query', { cause }) + mockUploadFile.mockRejectedValueOnce(error) + await expect( + storeLargeValue({}, '{}', 2, { + workspaceId: 'workspace-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + requireDurable: true, + }) + ).rejects.toMatchObject({ cause: error }) + }) + it('materializes object-storage refs through the server helper', async () => { mockDownloadFile.mockResolvedValueOnce(Buffer.from(JSON.stringify({ ok: true }), 'utf8')) diff --git a/apps/sim/lib/execution/payloads/store.ts b/apps/sim/lib/execution/payloads/store.ts index 90762f473ba..0cb1e14385e 100644 --- a/apps/sim/lib/execution/payloads/store.ts +++ b/apps/sim/lib/execution/payloads/store.ts @@ -97,7 +97,9 @@ async function persistValue( return fileInfo.key } catch (error) { if (context.requireDurable) { - throw new Error(`Failed to persist large execution value: ${toError(error).message}`) + throw new Error(`Failed to persist large execution value: ${toError(error).message}`, { + cause: error, + }) } logger.warn('Failed to persist large execution value, keeping in memory only', { id, diff --git a/apps/sim/lib/logs/execution/trace-store.test.ts b/apps/sim/lib/logs/execution/trace-store.test.ts index 299cbfda3fa..5707e24be19 100644 --- a/apps/sim/lib/logs/execution/trace-store.test.ts +++ b/apps/sim/lib/logs/execution/trace-store.test.ts @@ -51,6 +51,43 @@ beforeEach(() => { }) 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) + + await expect( + externalizeExecutionData({ traceSpans: [] }, CONTEXT, { throwOnError: true }) + ).rejects.toBe(error) + expect(mockLogger.warn).not.toHaveBeenCalled() + }) + + it('rejects missing ownership before strict backfills write anything', async () => { + await expect( + externalizeExecutionData( + { traceSpans: [] }, + { ...CONTEXT, userId: '' }, + { throwOnError: true } + ) + ).rejects.toThrow('Trace storage requires workspaceId, workflowId, and userId') + expect(storeLargeValueMock).not.toHaveBeenCalled() + }) + + it('preserves inline completion data and logs the underlying database error', async () => { + const data = { traceSpans: [] } + storeLargeValueMock.mockRejectedValueOnce( + new Error('Failed query\nparams: private-payload', { + cause: new Error('permission denied for table workspace_files'), + }) + ) + await expect(externalizeExecutionData(data, CONTEXT)).resolves.toBe(data) + expect(mockLogger.warn).toHaveBeenCalledWith(expect.any(String), { + executionId: CONTEXT.executionId, + error: expect.objectContaining({ message: 'permission denied for table workspace_files' }), + }) + expect(JSON.stringify(mockLogger.warn.mock.calls)).not.toContain('private-payload') + }) + it('keeps the trusted Copilot binding when an externalized payload is unavailable', async () => { const correlation = { copilotToolCallId: 'tool-call-1' } const ref = { diff --git a/apps/sim/lib/logs/execution/trace-store.ts b/apps/sim/lib/logs/execution/trace-store.ts index 4851910dab4..996556ee7d7 100644 --- a/apps/sim/lib/logs/execution/trace-store.ts +++ b/apps/sim/lib/logs/execution/trace-store.ts @@ -1,5 +1,5 @@ import { createLogger } from '@sim/logger' -import { toError } from '@sim/utils/errors' +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' @@ -231,17 +231,24 @@ export function copyTraceSpansWithoutCosts(spans?: TraceSpan[]): TraceSpan[] | u * * On any failure (no scope, oversized, storage error) the original (already * cost-stripped) execution data is returned unchanged so the log is never lost. + * Backfills pass `throwOnError` to stop instead of retaining inline data. */ export async function externalizeExecutionData( executionData: Record, - context: TraceStoreWriteContext + context: TraceStoreWriteContext, + options: { throwOnError?: boolean } = {} ): Promise> { const { workspaceId, workflowId, executionId, userId } = context // workspaceId/workflowId build the storage key and can be null for // deleted-workflow rows. userId is type-guaranteed by TraceStoreWriteContext; // the falsy check is a defensive guard against an empty string. If any are // missing the durable write can't succeed, so keep the data inline. - if (!workspaceId || !workflowId || !userId) return executionData + if (!workspaceId || !workflowId || !userId) { + if (options.throwOnError) { + throw new Error('Trace storage requires workspaceId, workflowId, and userId') + } + return executionData + } try { const json = JSON.stringify(executionData) @@ -266,9 +273,10 @@ export async function externalizeExecutionData( } return slim } catch (error) { + if (options.throwOnError) throw error logger.warn('Failed to externalize execution data; keeping inline', { executionId, - error: toError(error).message, + error: describeError(error), }) return executionData } diff --git a/apps/sim/scripts/backfill-trace-spans.test.ts b/apps/sim/scripts/backfill-trace-spans.test.ts new file mode 100644 index 00000000000..7388807673d --- /dev/null +++ b/apps/sim/scripts/backfill-trace-spans.test.ts @@ -0,0 +1,283 @@ +/** + * @vitest-environment node + */ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { MAX_DURABLE_LARGE_VALUE_BYTES } from '@/lib/execution/payloads/limits' + +const { + mockRead, + mockInfo, + mockDataRead, + mockTransaction, + mockUpdate, + mockExternalize, + mockReplaceReferences, +} = vi.hoisted(() => ({ + mockRead: vi.fn(), + mockInfo: vi.fn(), + mockDataRead: vi.fn(), + mockTransaction: vi.fn(), + mockUpdate: vi.fn(), + mockExternalize: vi.fn(), + mockReplaceReferences: vi.fn(), +})) + +vi.mock('@sim/db', () => { + const db = { + select: () => { + const query = { + from: () => query, + innerJoin: () => query, + where: () => query, + orderBy: () => query, + limit: mockRead, + } + return query + }, + transaction: mockTransaction, + } + return { db, dbFor: () => db } +}) + +vi.mock('@sim/logger', () => ({ + createLogger: () => ({ info: mockInfo, error: vi.fn(), warn: vi.fn(), debug: vi.fn() }), +})) + +vi.mock('@/lib/logs/execution/trace-store', () => ({ + externalizeExecutionData: mockExternalize, + stripSpanCosts: vi.fn(), + TRACE_STORE_REF_KEY: 'traceStoreRef', +})) + +vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({ + collectLargeValueReferenceKeys: () => ['stored-key'], + replaceLargeValueReferenceKeysWithClient: mockReplaceReferences, +})) + +import { backfillTraceStorage, parseArgs, runBackfillWorkers } from '@/scripts/backfill-trace-spans' + +beforeEach(() => { + vi.resetAllMocks() + mockRead.mockImplementation((limit: number) => + limit === 0 ? Promise.resolve([]) : mockDataRead(limit) + ) + mockDataRead.mockResolvedValue([]) + mockExternalize.mockResolvedValue({ traceStoreRef: { key: 'stored-key' } }) + mockTransaction.mockImplementation(async (callback) => + callback({ + update: () => ({ set: () => ({ where: mockUpdate }) }), + }) + ) +}) + +afterEach(() => vi.useRealTimers()) + +describe('backfill options', () => { + it('defaults to four workers and supports a read-only check', () => { + expect(parseArgs([])).toEqual({ + maxBatches: Number.POSITIVE_INFINITY, + concurrency: 4, + checkOnly: false, + }) + expect(parseArgs(['--check-only', '--max-batches=2', '--concurrency=8'])).toEqual({ + maxBatches: 2, + concurrency: 8, + checkOnly: true, + }) + }) + + it.each([50, 64])('supports %i workers', (concurrency) => { + expect(parseArgs([`--concurrency=${concurrency}`]).concurrency).toBe(concurrency) + }) + + it.each([ + '--max-batches=2junk', + '--max-batches=1.5', + '--max-batches=0', + '--concurrency=65', + '--concurrency=-1', + '--concurrency=0', + '--concurrency=2=3', + '--unknown', + ])('rejects invalid input: %s', (arg) => { + expect(() => parseArgs([arg])).toThrow() + }) +}) + +describe('backfill workers', () => { + it.each([2, 50, 64])( + 'bounds active work to %i workers and visits every candidate once', + async (concurrency) => { + let active = 0 + let peak = 0 + const visited: string[] = [] + const rows = Array.from({ length: 130 }, (_, index) => ({ + id: `log-${index}`, + })) + await runBackfillWorkers(rows, concurrency, async (row) => { + active++ + peak = Math.max(peak, active) + await Promise.resolve() + visited.push(row.id) + active-- + }) + expect(peak).toBe(concurrency) + expect(visited.sort()).toEqual(rows.map((row) => row.id).sort()) + } + ) + + it('stops scheduling after a failure and drains writes already in flight', async () => { + let finishWrite = () => {} + const writing = new Promise((resolve) => { + finishWrite = resolve + }) + const error = new Error('database unavailable') + const started: string[] = [] + let completedWrite = false + const rows = [1, 2, 3, 4].map((id) => ({ id: `log-${id}` })) + const run = runBackfillWorkers(rows, 2, async ({ id }) => { + started.push(id) + if (id === 'log-2') throw error + await writing + completedWrite = true + }) + const assertion = expect(run).rejects.toBe(error) + await Promise.resolve() + expect(started).toEqual(['log-1', 'log-2']) + expect(completedWrite).toBe(false) + finishWrite() + await assertion + expect(completedWrite).toBe(true) + expect(started).toEqual(['log-1', 'log-2']) + }) +}) + +describe('trace backfill', () => { + const options = { maxBatches: 1, concurrency: 1, checkOnly: false } + const candidate = { + id: 'log-1', + workspaceId: 'workspace-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + workflowOwnerUserId: 'user-1', + payloadBytes: 128, + executionData: { traceSpans: [{ children: [{}] }] }, + } + const candidateMetadata = { id: candidate.id } + + it('fails its schema check before uploading anything and preserves the database cause', async () => { + const cause = new Error('column "size_bytes" does not exist') + const error = new Error('Failed query', { cause }) + mockRead.mockRejectedValueOnce(error) + await expect(backfillTraceStorage(options)).rejects.toBe(error) + expect(mockDataRead).not.toHaveBeenCalled() + expect(mockExternalize).not.toHaveBeenCalled() + expect(mockTransaction).not.toHaveBeenCalled() + }) + + it('check-only performs no uploads, writes, or payload reads', async () => { + await backfillTraceStorage({ ...options, checkOnly: true }) + expect(mockRead).toHaveBeenCalledTimes(5) + expect(mockRead.mock.calls.every(([limit]) => limit === 0)).toBe(true) + expect(mockExternalize).not.toHaveBeenCalled() + expect(mockTransaction).not.toHaveBeenCalled() + }) + + it('commits a durable pointer and references with the recovered owner', async () => { + mockDataRead.mockResolvedValueOnce([candidateMetadata]).mockResolvedValueOnce([candidate]) + await expect(backfillTraceStorage(options)).resolves.toEqual({ + migrated: 1, + recoveredOwners: 1, + }) + expect(mockExternalize).toHaveBeenCalledWith( + expect.objectContaining({ hasTraceSpans: true, traceSpanCount: 2 }), + { + workspaceId: 'workspace-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + userId: 'user-1', + }, + { throwOnError: true } + ) + expect(mockUpdate).toHaveBeenCalledOnce() + expect(mockReplaceReferences).toHaveBeenCalledWith( + expect.anything(), + { + workspaceId: 'workspace-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + source: 'execution_log', + }, + ['stored-key'] + ) + }) + + it('rejects an oversized payload before uploading or updating the log', async () => { + mockDataRead.mockResolvedValueOnce([candidateMetadata]).mockResolvedValueOnce([ + { + ...candidate, + executionData: null, + payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES + 1, + }, + ]) + await expect(backfillTraceStorage(options)).rejects.toMatchObject({ + cause: expect.objectContaining({ message: expect.stringContaining('backfill limit') }), + }) + expect(mockExternalize).not.toHaveBeenCalled() + expect(mockTransaction).not.toHaveBeenCalled() + }) + + it('reports progress during a slow batch and removes the timer when finished', async () => { + vi.useFakeTimers({ now: 0 }) + let markStarted = () => {} + const started = new Promise((resolve) => { + markStarted = resolve + }) + let finishUpload = () => {} + const upload = new Promise((resolve) => { + finishUpload = resolve + }) + mockDataRead.mockResolvedValueOnce([candidateMetadata]).mockResolvedValueOnce([candidate]) + mockExternalize.mockImplementation(async () => { + markStarted() + await upload + return { traceStoreRef: { key: 'stored-key' } } + }) + const run = backfillTraceStorage(options) + await started + await vi.advanceTimersByTimeAsync(5000) + expect(mockInfo).toHaveBeenCalledWith( + 'Progress: migrated 0 | skipped 0 | 0.0 rows/s | elapsed 5s' + ) + finishUpload() + await run + expect(mockInfo).toHaveBeenLastCalledWith( + 'Progress: migrated 1 | skipped 0 | 0.2 rows/s | elapsed 5s' + ) + expect(vi.getTimerCount()).toBe(0) + }) + + it('starts migrating after preflight without a full-table count or estimate', async () => { + mockDataRead.mockResolvedValueOnce([candidateMetadata]).mockResolvedValueOnce([candidate]) + await backfillTraceStorage(options) + expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 0, 100, 1]) + expect(mockExternalize).toHaveBeenCalledOnce() + }) + + it('keeps the candidate page at 100 rows with fifty workers', async () => { + await backfillTraceStorage({ ...options, concurrency: 50 }) + expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 0, 100]) + expect(mockExternalize).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]) + 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 1fb3a9fd8f3..721a39f2357 100644 --- a/apps/sim/scripts/backfill-trace-spans.ts +++ b/apps/sim/scripts/backfill-trace-spans.ts @@ -16,17 +16,41 @@ * self-hosted — and is intentionally NOT part of this script. * * Usage: - * DATABASE_URL=... bun apps/sim/scripts/backfill-trace-spans.ts [--max-batches=] + * bun apps/sim/scripts/backfill-trace-spans.ts --check-only + * bun apps/sim/scripts/backfill-trace-spans.ts --concurrency=4 [--max-batches=] + * 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= + * + * 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–64 workers; it does not set a rows-per-second target. + * Scans oldest first by started_at, with id breaking ties. --before is an + * 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. */ -import { db } from '@sim/db' -import { workflowExecutionLogs } from '@sim/db/schema' -import { toError } from '@sim/utils/errors' -import { and, asc, eq, gt, sql } from 'drizzle-orm' +import { db, dbFor } from '@sim/db' +import { + executionLargeValueDependencies, + executionLargeValueReferences, + executionLargeValues, + workflow, + workflowExecutionLogs, + workspaceFiles, +} from '@sim/db/schema' +import { createLogger } from '@sim/logger' +import { describeError, toError } from '@sim/utils/errors' +import { formatDuration } from '@sim/utils/formatting' +import { and, asc, desc, eq, sql } from 'drizzle-orm' +import { z } from 'zod' import { collectLargeValueReferenceKeys, replaceLargeValueReferenceKeysWithClient, } from '@/lib/execution/payloads/large-value-metadata' +import { MAX_DURABLE_LARGE_VALUE_BYTES } from '@/lib/execution/payloads/limits' import { externalizeExecutionData, stripSpanCosts, @@ -34,6 +58,20 @@ import { } from '@/lib/logs/execution/trace-store' const TRACE_BATCH_SIZE = 100 +const DEFAULT_CONCURRENCY = 4 +const MAX_CONCURRENCY = 64 +const PROGRESS_INTERVAL_MS = 5_000 +const logger = createLogger('BackfillTraceSpans', { logLevel: 'INFO' }) +const orderSchema = z.enum(['oldest', 'newest']) +const cutoffSchema = z.iso.datetime({ offset: true }) +const cursorSchema = z.object({ + version: z.literal(1), + order: orderSchema, + before: cutoffSchema, + startedAt: z.iso.datetime({ precision: 6 }), + id: z.string().min(1).max(512), +}) +type BackfillCursor = z.infer /** * Recursively counts trace spans (matching the completion path). Legacy rows @@ -52,137 +90,325 @@ function countTraceSpans(spans: unknown): number { interface Options { maxBatches: number + concurrency: number + checkOnly: boolean + order: z.infer + before: string + cursor?: BackfillCursor } -function parseArgs(argv: string[]): Options { - const maxBatchesArg = argv.find((a) => a.startsWith('--max-batches=')) - const maxBatches = maxBatchesArg - ? Number.parseInt(maxBatchesArg.slice('--max-batches='.length), 10) - : Number.POSITIVE_INFINITY - - if (Number.isNaN(maxBatches) || maxBatches <= 0) { - throw new Error('--max-batches must be a positive integer') +export function parseArgs(argv: string[]): Options { + const options: Options = { + maxBatches: Number.POSITIVE_INFINITY, + concurrency: DEFAULT_CONCURRENCY, + checkOnly: false, + order: 'oldest', + before: new Date().toISOString(), } + let explicitOrder = false + let explicitBefore = false + for (const arg of argv) { + if (arg === '--check-only') { + options.checkOnly = true + continue + } + const [name, value] = arg.split('=') + if (name === '--order' || name === '--before' || name === '--cursor') { + if (!value || arg.split('=').length !== 2) { + throw new Error(`${name} requires a value`) + } + if (name === '--order') { + options.order = orderSchema.parse(value) + explicitOrder = true + } else if (name === '--before') { + options.before = cutoffSchema.parse(value) + explicitBefore = true + } else { + try { + options.cursor = cursorSchema.parse( + JSON.parse(Buffer.from(value, 'base64url').toString('utf8')) + ) + } catch (error) { + throw new Error('Invalid --cursor: use a checkpoint token printed by the backfill', { + cause: error, + }) + } + } + continue + } + if (name !== '--max-batches' && name !== '--concurrency') { + throw new Error(`Unknown argument: ${arg}`) + } + const parsed = Number(value) + if ( + !/^\d+$/.test(value ?? '') || + !Number.isSafeInteger(parsed) || + parsed <= 0 || + arg.split('=').length !== 2 + ) { + throw new Error(`${name} must be a positive integer`) + } + if (name === '--max-batches') options.maxBatches = parsed + else options.concurrency = parsed + } + if (options.concurrency > MAX_CONCURRENCY) { + throw new Error(`--concurrency must be between 1 and ${MAX_CONCURRENCY}`) + } + if (options.cursor) { + if (explicitOrder && options.order !== options.cursor.order) { + throw new Error('--order must match the checkpoint cursor') + } + if (explicitBefore && options.before !== options.cursor.before) { + throw new Error('--before must match the checkpoint cursor') + } + options.order = options.cursor.order + options.before = options.cursor.before + } + return options +} + +/** Validates the metadata schema and SELECT permissions without reading payloads or writing. */ +export async function checkDatabase(): Promise { + await db.select().from(workspaceFiles).limit(0) + await db.select().from(executionLargeValueReferences).limit(0) + const execDb = dbFor('exec') + await execDb.select().from(executionLargeValues).limit(0) + await execDb.select().from(executionLargeValueDependencies).limit(0) + await db + .select({ + id: workflowExecutionLogs.id, + workspaceId: workflowExecutionLogs.workspaceId, + workflowId: workflowExecutionLogs.workflowId, + executionId: workflowExecutionLogs.executionId, + executionData: workflowExecutionLogs.executionData, + endedAt: workflowExecutionLogs.endedAt, + startedAt: workflowExecutionLogs.startedAt, + workflowOwnerUserId: workflow.userId, + }) + .from(workflowExecutionLogs) + .innerJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id)) + .limit(0) +} - return { maxBatches } +/** Stops scheduling on failure and lets active writes settle before the process exits. */ +export async function runBackfillWorkers( + rows: T[], + concurrency: number, + processRow: (row: T) => Promise +): Promise { + let cursor = 0 + let failure: Error | undefined + const worker = async () => { + while (!failure && cursor < rows.length) { + const row = rows[cursor++] + try { + await processRow(row) + } catch (error) { + failure ??= toError(error) + } + } + } + await Promise.all(Array.from({ length: Math.min(concurrency, rows.length) }, worker)) + if (failure) throw failure } /** Externalize inline heavy execution_data into the large-value store. */ -async function backfillTraceStorage( - maxBatches: number -): Promise<{ migrated: number; failed: number }> { - let migrated = 0 - let failed = 0 - // Keyset cursor by id: every row is visited at most once per run, so rows that - // can't be externalized (storage error, oversized) aren't re-selected into an - // infinite loop. A fresh re-run (cursor reset) retries any that failed. - let lastId = '' - - for (let batch = 0; batch < maxBatches; batch++) { - const rows = await db - .select({ - id: workflowExecutionLogs.id, - workspaceId: workflowExecutionLogs.workspaceId, - workflowId: workflowExecutionLogs.workflowId, - executionId: workflowExecutionLogs.executionId, - executionData: workflowExecutionLogs.executionData, - }) - .from(workflowExecutionLogs) - .where( - and( - sql`${workflowExecutionLogs.endedAt} IS NOT NULL`, - // Skip deleted-workflow rows: externalization requires a workflowId. - sql`${workflowExecutionLogs.workflowId} IS NOT NULL`, - sql`${workflowExecutionLogs.executionData} ? 'traceSpans'`, - sql`NOT (${workflowExecutionLogs.executionData} ? ${TRACE_STORE_REF_KEY})`, - lastId ? gt(workflowExecutionLogs.id, lastId) : undefined - ) - ) - .orderBy(asc(workflowExecutionLogs.id)) - .limit(TRACE_BATCH_SIZE) +export async function backfillTraceStorage( + options: Options +): Promise<{ migrated: number; recoveredOwners: number }> { + await checkDatabase() + logger.info('Database schema and read checks passed') + if (options.checkOnly) return { migrated: 0, recoveredOwners: 0 } - if (rows.length === 0) break + let migrated = 0 + let recoveredOwners = 0 + let skipped = 0 + let cursor = options.cursor + const direction = options.order === 'oldest' ? asc : desc + const pending = and( + sql`${workflowExecutionLogs.endedAt} IS NOT NULL`, + sql`${workflowExecutionLogs.workflowId} IS NOT NULL`, + sql`${workflowExecutionLogs.executionData} ? 'traceSpans'`, + sql`NOT (${workflowExecutionLogs.executionData} ? ${TRACE_STORE_REF_KEY})` + ) + const startedAt = Date.now() + logger.info('Scanning execution logs', { order: options.order, before: options.before }) + const reportProgress = () => { + 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)}` + ) + } + reportProgress() + const progressTimer = setInterval(reportProgress, PROGRESS_INTERVAL_MS) + progressTimer.unref() - for (const row of rows) { - try { - const executionData = (row.executionData ?? {}) as Record - // Derive the inline markers legacy rows lack so externalizeExecutionData - // carries them onto the slim row (they survive object expiry). - const traceSpanCount = countTraceSpans(executionData.traceSpans) - executionData.hasTraceSpans = traceSpanCount > 0 - executionData.traceSpanCount = traceSpanCount - stripSpanCosts(executionData.traceSpans) - // workspace_files.user_id (NOT NULL) needs the execution owner; legacy - // rows carry it under executionData.environment.userId. Rows without an - // owner can't be externalized — count them as failed and skip. - const environment = executionData.environment as { userId?: string } | undefined - const ownerUserId = environment?.userId - if (!ownerUserId) { - failed++ - continue - } - const slim = await externalizeExecutionData(executionData, { - workspaceId: row.workspaceId, - workflowId: row.workflowId, - executionId: row.executionId, - userId: ownerUserId, + try { + for (let batch = 0; batch < options.maxBatches; batch++) { + /** + * Seek by indexed time before checking JSON or joining workflows. Limiting + * eligible rows here can scan the entire backlog to fill one page. + * Preserve microseconds as text: a JS Date would lose cursor precision. + */ + const cursorPredicate = cursor + ? options.order === 'oldest' + ? and( + sql`${workflowExecutionLogs.startedAt} >= ${cursor.startedAt}::timestamp`, + sql`(${workflowExecutionLogs.startedAt}, ${workflowExecutionLogs.id}) > (${cursor.startedAt}::timestamp, ${cursor.id})` + ) + : and( + sql`${workflowExecutionLogs.startedAt} <= ${cursor.startedAt}::timestamp`, + sql`(${workflowExecutionLogs.startedAt}, ${workflowExecutionLogs.id}) < (${cursor.startedAt}::timestamp, ${cursor.id})` + ) + : undefined + const rows = await db + .select({ + id: workflowExecutionLogs.id, + startedAt: sql`to_char(${workflowExecutionLogs.startedAt}, 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, }) + .from(workflowExecutionLogs) + .where( + and( + sql`${workflowExecutionLogs.startedAt} < (${options.before}::timestamptz AT TIME ZONE 'UTC')`, + cursorPredicate + ) + ) + .orderBy(direction(workflowExecutionLogs.startedAt), direction(workflowExecutionLogs.id)) + .limit(TRACE_BATCH_SIZE) - if (!(TRACE_STORE_REF_KEY in slim)) { - failed++ - continue - } - - await db.transaction(async (tx) => { - await tx - .update(workflowExecutionLogs) - .set({ executionData: slim }) - .where(eq(workflowExecutionLogs.id, row.id)) + if (rows.length === 0) break - await replaceLargeValueReferenceKeysWithClient( - tx, + await runBackfillWorkers(rows, options.concurrency, async ({ id }) => { + try { + /** Reject oversized JSON in SQL before the driver materializes it. */ + const [row] = await db + .select({ + id: workflowExecutionLogs.id, + workspaceId: workflowExecutionLogs.workspaceId, + workflowId: workflowExecutionLogs.workflowId, + executionId: workflowExecutionLogs.executionId, + workflowOwnerUserId: workflow.userId, + payloadBytes: sql`octet_length(${workflowExecutionLogs.executionData}::text)`, + executionData: sql | null>`CASE + WHEN octet_length(${workflowExecutionLogs.executionData}::text) <= ${MAX_DURABLE_LARGE_VALUE_BYTES} + THEN ${workflowExecutionLogs.executionData} + ELSE NULL END`, + }) + .from(workflowExecutionLogs) + .innerJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id)) + .where(and(eq(workflowExecutionLogs.id, id), pending)) + .limit(1) + /** Skip running, deleted-workflow, already externalized, or removed rows. */ + if (!row) { + skipped++ + return + } + if (row.payloadBytes > MAX_DURABLE_LARGE_VALUE_BYTES) { + throw new Error( + `Execution payload is ${row.payloadBytes} bytes, exceeding the ${MAX_DURABLE_LARGE_VALUE_BYTES}-byte backfill limit` + ) + } + const executionData = row.executionData + if (!executionData) throw new Error('Execution data is missing') + const traceSpanCount = countTraceSpans(executionData.traceSpans) + executionData.hasTraceSpans = traceSpanCount > 0 + executionData.traceSpanCount = traceSpanCount + stripSpanCosts(executionData.traceSpans) + /** + * workspace_files.user_id (NOT NULL) needs an owner. Most rows carry + * it under executionData.environment.userId; the legacy workflow-log + * endpoint wrote an empty userId, so recover the workflow owner that + * the corrected endpoint would have persisted. + */ + const environment = executionData.environment as { userId?: string } | undefined + const storedOwnerUserId = environment?.userId + const ownerUserId = storedOwnerUserId || row.workflowOwnerUserId + const slim = await externalizeExecutionData( + executionData, { workspaceId: row.workspaceId, workflowId: row.workflowId, executionId: row.executionId, - source: 'execution_log', + userId: ownerUserId, }, - collectLargeValueReferenceKeys(slim) + { throwOnError: true } ) - }) - migrated++ - } catch (error) { - failed++ - console.error(` [trace] row ${row.id} failed: ${toError(error).message}`) - } - } + if (!(TRACE_STORE_REF_KEY in slim)) { + throw new Error('Trace storage did not return a durable reference') + } + + await db.transaction(async (tx) => { + await tx + .update(workflowExecutionLogs) + .set({ executionData: slim }) + .where(eq(workflowExecutionLogs.id, row.id)) - // Advance the cursor past this batch so failed rows aren't re-selected. - lastId = rows[rows.length - 1].id + await replaceLargeValueReferenceKeysWithClient( + tx, + { + workspaceId: row.workspaceId, + workflowId: row.workflowId, + executionId: row.executionId, + source: 'execution_log', + }, + collectLargeValueReferenceKeys(slim) + ) + }) - console.log(` [trace] batch ${batch + 1}: migrated ${migrated}, failed ${failed}`) + migrated++ + if (!storedOwnerUserId) recoveredOwners++ + } catch (error) { + throw new Error(`Backfill failed for execution log ${id}`, { cause: error }) + } + }) + + const last = rows[rows.length - 1] + cursor = { + version: 1, + order: options.order, + before: options.before, + startedAt: last.startedAt, + id: last.id, + } + logger.info('Backfill checkpoint', { + startedAt: cursor.startedAt, + id: cursor.id, + cursor: Buffer.from(JSON.stringify(cursor)).toString('base64url'), + }) + + reportProgress() + } + } finally { + clearInterval(progressTimer) + reportProgress() } - return { migrated, failed } + return { migrated, recoveredOwners } } async function main(): Promise { const options = parseArgs(process.argv.slice(2)) const startedAt = Date.now() - console.log('Backfilling trace storage (externalizing execution_data)…') - const { migrated, failed } = await backfillTraceStorage(options.maxBatches) - console.log(`Trace storage done: ${migrated} migrated, ${failed} skipped/failed.`) - - console.log(`Backfill complete in ${((Date.now() - startedAt) / 1000).toFixed(1)}s.`) -} - -main() - .catch((err) => { - console.error('Backfill failed:', err) - process.exit(1) + logger.info('Starting trace backfill', { + concurrency: options.concurrency, + checkOnly: options.checkOnly, }) - .finally(() => { - process.exit(0) + const result = await backfillTraceStorage(options) + logger.info(options.checkOnly ? 'Database check complete' : 'Backfill run finished', { + ...result, + elapsedSeconds: (Date.now() - startedAt) / 1000, }) +} + +if (import.meta.main) { + main().then( + () => process.exit(0), + (error) => { + logger.error('Backfill failed', describeError(error)) + process.exit(1) + } + ) +} From 3d97316dd82ac1e6e9ac29eb6f1c4d3e4f0d9acf Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 18 Sep 2026 11:10:36 -0700 Subject: [PATCH 2/3] fix(traces): bound and resume trace backfill workers --- apps/sim/scripts/backfill-trace-spans.test.ts | 232 ++++++++++- apps/sim/scripts/backfill-trace-spans.ts | 393 ++++++++++++------ 2 files changed, 488 insertions(+), 137 deletions(-) diff --git a/apps/sim/scripts/backfill-trace-spans.test.ts b/apps/sim/scripts/backfill-trace-spans.test.ts index 7388807673d..cae6b70d486 100644 --- a/apps/sim/scripts/backfill-trace-spans.test.ts +++ b/apps/sim/scripts/backfill-trace-spans.test.ts @@ -74,27 +74,60 @@ afterEach(() => vi.useRealTimers()) describe('backfill options', () => { it('defaults to four workers and supports a read-only check', () => { + const before = '2026-09-17T20:00:00.000Z' + vi.useFakeTimers({ now: new Date(before) }) expect(parseArgs([])).toEqual({ maxBatches: Number.POSITIVE_INFINITY, concurrency: 4, + maxInFlightMiB: 512, checkOnly: false, + order: 'oldest', + before, }) expect(parseArgs(['--check-only', '--max-batches=2', '--concurrency=8'])).toEqual({ maxBatches: 2, concurrency: 8, + maxInFlightMiB: 512, checkOnly: true, + order: 'oldest', + before, }) }) - it.each([50, 64])('supports %i workers', (concurrency) => { + it.each([50, 64, 200, 500, 512])('supports %i workers', (concurrency) => { expect(parseArgs([`--concurrency=${concurrency}`]).concurrency).toBe(concurrency) }) + it('supports a bounded payload budget independently of worker count', () => { + expect(parseArgs(['--concurrency=500', '--max-in-flight-mib=128'])).toMatchObject({ + concurrency: 500, + maxInFlightMiB: 128, + }) + }) + + it('preserves microseconds, ordering, and cutoff when resuming a checkpoint', () => { + const cursor = { + version: 1, + order: 'newest', + before: '2026-01-01T00:00:00.000Z', + startedAt: '2025-01-01T00:00:00.123456Z', + id: 'log-1', + } + const flag = `--cursor=${Buffer.from(JSON.stringify(cursor)).toString('base64url')}` + expect(parseArgs([flag])).toMatchObject({ cursor, order: cursor.order, before: cursor.before }) + expect(() => parseArgs([flag, '--order=oldest'])).toThrow('must match') + expect(() => parseArgs([flag, '--before=2026-02-01T00:00:00.000Z'])).toThrow('must match') + expect(() => parseArgs(['--cursor=invalid'])).toThrow('Invalid --cursor') + }) + it.each([ '--max-batches=2junk', '--max-batches=1.5', '--max-batches=0', - '--concurrency=65', + '--concurrency=513', + '--max-in-flight-mib=63', + '--max-in-flight-mib=4097', + '--max-in-flight-mib=1.5', '--concurrency=-1', '--concurrency=0', '--concurrency=2=3', @@ -105,13 +138,62 @@ describe('backfill options', () => { }) describe('backfill workers', () => { - it.each([2, 50, 64])( + it('reserves bytes before starting work and releases capacity for subsequent rows', async () => { + const rows = [6, 6, 4, 9, 1] + let activeBytes = 0 + let peakBytes = 0 + const visited: number[] = [] + const completed = await runBackfillWorkers( + rows, + 5, + async (bytes) => { + activeBytes += bytes + peakBytes = Math.max(peakBytes, activeBytes) + visited.push(bytes) + await Promise.resolve() + activeBytes -= bytes + }, + { byteBudget: { maxBytes: 10, sizeOf: (bytes) => bytes } } + ) + expect(completed).toBe(5) + expect(peakBytes).toBe(10) + expect(visited).toEqual(rows) + }) + + it('rejects a row larger than the byte budget without starting it', async () => { + const processRow = vi.fn() + await expect( + runBackfillWorkers([11], 5, processRow, { + byteBudget: { maxBytes: 10, sizeOf: (bytes) => bytes }, + }) + ).rejects.toThrow('byte budget') + expect(processRow).not.toHaveBeenCalled() + }) + + it('stops scheduling on shutdown while allowing started writes to settle', async () => { + const controller = new AbortController() + const finished: number[] = [] + const completed = await runBackfillWorkers( + [1, 2, 3, 4], + 2, + async (id) => { + controller.abort() + await Promise.resolve() + finished.push(id) + }, + { signal: controller.signal } + ) + expect(completed).toBe(2) + expect(finished).toEqual([1, 2]) + }) + + it.each([2, 50, 64, 500])( 'bounds active work to %i workers and visits every candidate once', async (concurrency) => { let active = 0 let peak = 0 const visited: string[] = [] - const rows = Array.from({ length: 130 }, (_, index) => ({ + const rows = Array.from({ length: 1030 }, (_, index) => ({ id: `log-${index}`, })) await runBackfillWorkers(rows, concurrency, async (row) => { @@ -153,7 +235,14 @@ describe('backfill workers', () => { }) describe('trace backfill', () => { - const options = { maxBatches: 1, concurrency: 1, checkOnly: false } + const options = { + maxBatches: 1, + concurrency: 1, + maxInFlightMiB: 512, + checkOnly: false, + order: 'oldest' as const, + before: '2026-09-17T20:00:00.000Z', + } const candidate = { id: 'log-1', workspaceId: 'workspace-1', @@ -163,7 +252,7 @@ describe('trace backfill', () => { payloadBytes: 128, executionData: { traceSpans: [{ children: [{}] }] }, } - const candidateMetadata = { id: candidate.id } + const candidateMetadata = { id: candidate.id, startedAt: '2026-09-16T20:00:00.123456Z' } it('fails its schema check before uploading anything and preserves the database cause', async () => { const cause = new Error('column "size_bytes" does not exist') @@ -183,11 +272,13 @@ describe('trace backfill', () => { expect(mockTransaction).not.toHaveBeenCalled() }) - it('commits a durable pointer and references with the recovered owner', async () => { - mockDataRead.mockResolvedValueOnce([candidateMetadata]).mockResolvedValueOnce([candidate]) + it('commits a durable pointer and references with the workflow owner', async () => { + mockDataRead + .mockResolvedValueOnce([candidateMetadata]) + .mockResolvedValueOnce([candidate]) + .mockResolvedValueOnce([candidate]) await expect(backfillTraceStorage(options)).resolves.toEqual({ migrated: 1, - recoveredOwners: 1, }) expect(mockExternalize).toHaveBeenCalledWith( expect.objectContaining({ hasTraceSpans: true, traceSpanCount: 2 }), @@ -212,6 +303,43 @@ describe('trace backfill', () => { ) }) + it.each(['execution-user', 'deleted-user', ''])( + 'uses the workflow owner while preserving execution user %j in the payload', + async (userId) => { + const executionData = { + ...structuredClone(candidate.executionData), + environment: { userId }, + } + mockDataRead + .mockResolvedValueOnce([candidateMetadata]) + .mockResolvedValueOnce([candidate]) + .mockResolvedValueOnce([{ ...candidate, executionData }]) + + await backfillTraceStorage(options) + + expect(mockExternalize).toHaveBeenCalledWith( + expect.objectContaining({ environment: { userId } }), + expect.objectContaining({ userId: candidate.workflowOwnerUserId }), + { throwOnError: true } + ) + expect(executionData.environment.userId).toBe(userId) + expect(mockUpdate).toHaveBeenCalledOnce() + } + ) + + it('fails before uploading when the workflow owner is missing', async () => { + mockDataRead + .mockResolvedValueOnce([candidateMetadata]) + .mockResolvedValueOnce([candidate]) + .mockResolvedValueOnce([{ ...candidate, workflowOwnerUserId: '' }]) + + await expect(backfillTraceStorage(options)).rejects.toMatchObject({ + cause: expect.objectContaining({ message: 'Workflow owner is missing' }), + }) + expect(mockExternalize).not.toHaveBeenCalled() + expect(mockTransaction).not.toHaveBeenCalled() + }) + it('rejects an oversized payload before uploading or updating the log', async () => { mockDataRead.mockResolvedValueOnce([candidateMetadata]).mockResolvedValueOnce([ { @@ -220,13 +348,69 @@ describe('trace backfill', () => { payloadBytes: MAX_DURABLE_LARGE_VALUE_BYTES + 1, }, ]) + await expect(backfillTraceStorage(options)).rejects.toThrow('backfill limit') + expect(mockDataRead).toHaveBeenCalledTimes(2) + 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('backfill limit') }), + cause: expect.objectContaining({ message: expect.stringContaining('grew') }), }) expect(mockExternalize).not.toHaveBeenCalled() expect(mockTransaction).not.toHaveBeenCalled() }) + it('preserves the previous checkpoint after shutdown interrupts a page', async () => { + const controller = new AbortController() + 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, { ...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 }, controller.signal)).resolves.toEqual({ + migrated: 1, + }) + expect(mockUpdate).toHaveBeenCalledOnce() + const checkpoints = mockInfo.mock.calls.filter(([message]) => message === 'Backfill checkpoint') + expect(checkpoints.length).toBeGreaterThan(0) + for (const [, checkpoint] of checkpoints) { + expect(JSON.parse(Buffer.from(checkpoint.cursor, 'base64url').toString('utf8'))).toEqual( + cursor + ) + } + }) + + it('advances the cursor over pages with no eligible payloads', async () => { + mockDataRead.mockResolvedValueOnce([candidateMetadata]).mockResolvedValueOnce([]) + await backfillTraceStorage(options) + expect(mockExternalize).not.toHaveBeenCalled() + expect(mockInfo).toHaveBeenCalledWith( + 'Backfill checkpoint', + expect.objectContaining({ + id: candidateMetadata.id, + startedAt: candidateMetadata.startedAt, + }) + ) + }) + it('reports progress during a slow batch and removes the timer when finished', async () => { vi.useFakeTimers({ now: 0 }) let markStarted = () => {} @@ -237,7 +421,10 @@ describe('trace backfill', () => { const upload = new Promise((resolve) => { finishUpload = resolve }) - mockDataRead.mockResolvedValueOnce([candidateMetadata]).mockResolvedValueOnce([candidate]) + mockDataRead + .mockResolvedValueOnce([candidateMetadata]) + .mockResolvedValueOnce([candidate]) + .mockResolvedValueOnce([candidate]) mockExternalize.mockImplementation(async () => { markStarted() await upload @@ -247,20 +434,27 @@ 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 | 0.0 rows/s | elapsed 5s', + expect.objectContaining({ rssMiB: expect.any(Number) }) ) finishUpload() await run - expect(mockInfo).toHaveBeenLastCalledWith( - 'Progress: migrated 1 | skipped 0 | 0.2 rows/s | elapsed 5s' + expect(mockInfo).toHaveBeenCalledWith( + 'Progress: migrated 1 | skipped 0 | 0.2 rows/s | elapsed 5s', + expect.objectContaining({ + stages: expect.objectContaining({ externalize: { calls: 1, averageMs: 5000 } }), + }) ) expect(vi.getTimerCount()).toBe(0) }) it('starts migrating after preflight without a full-table count or estimate', async () => { - mockDataRead.mockResolvedValueOnce([candidateMetadata]).mockResolvedValueOnce([candidate]) + mockDataRead + .mockResolvedValueOnce([candidateMetadata]) + .mockResolvedValueOnce([candidate]) + .mockResolvedValueOnce([candidate]) await backfillTraceStorage(options) - expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 0, 100, 1]) + expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 0, 100, 1, 1]) expect(mockExternalize).toHaveBeenCalledOnce() }) @@ -270,9 +464,15 @@ describe('trace backfill', () => { expect(mockExternalize).not.toHaveBeenCalled() }) + it('scales the bounded metadata page to feed higher concurrency', async () => { + await backfillTraceStorage({ ...options, concurrency: 500 }) + expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 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) diff --git a/apps/sim/scripts/backfill-trace-spans.ts b/apps/sim/scripts/backfill-trace-spans.ts index 721a39f2357..25186ab869a 100644 --- a/apps/sim/scripts/backfill-trace-spans.ts +++ b/apps/sim/scripts/backfill-trace-spans.ts @@ -21,11 +21,17 @@ * 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=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–64 workers; it does not set a rows-per-second target. + * 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 + * 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 + * advances the checkpoint; resuming rechecks its already-committed rows. * Scans oldest first by started_at, with id breaking ties. --before is an * exclusive start-time cutoff (defaults to startup). Each completed page logs * a cursor that preserves the cutoff and order for restart. --max-batches @@ -44,7 +50,7 @@ import { import { createLogger } from '@sim/logger' import { describeError, toError } from '@sim/utils/errors' import { formatDuration } from '@sim/utils/formatting' -import { and, asc, desc, eq, sql } from 'drizzle-orm' +import { and, asc, desc, eq, inArray, sql } from 'drizzle-orm' import { z } from 'zod' import { collectLargeValueReferenceKeys, @@ -59,7 +65,9 @@ import { const TRACE_BATCH_SIZE = 100 const DEFAULT_CONCURRENCY = 4 -const MAX_CONCURRENCY = 64 +const MAX_CONCURRENCY = 512 +const MIB = 1024 * 1024 +const DEFAULT_MAX_IN_FLIGHT_MIB = 512 const PROGRESS_INTERVAL_MS = 5_000 const logger = createLogger('BackfillTraceSpans', { logLevel: 'INFO' }) const orderSchema = z.enum(['oldest', 'newest']) @@ -91,6 +99,7 @@ function countTraceSpans(spans: unknown): number { interface Options { maxBatches: number concurrency: number + maxInFlightMiB: number checkOnly: boolean order: z.infer before: string @@ -101,6 +110,7 @@ export function parseArgs(argv: string[]): Options { const options: Options = { maxBatches: Number.POSITIVE_INFINITY, concurrency: DEFAULT_CONCURRENCY, + maxInFlightMiB: DEFAULT_MAX_IN_FLIGHT_MIB, checkOnly: false, order: 'oldest', before: new Date().toISOString(), @@ -136,7 +146,7 @@ export function parseArgs(argv: string[]): Options { } continue } - if (name !== '--max-batches' && name !== '--concurrency') { + if (name !== '--max-batches' && name !== '--concurrency' && name !== '--max-in-flight-mib') { throw new Error(`Unknown argument: ${arg}`) } const parsed = Number(value) @@ -149,11 +159,18 @@ export function parseArgs(argv: string[]): Options { throw new Error(`${name} must be a positive integer`) } if (name === '--max-batches') options.maxBatches = parsed + else if (name === '--max-in-flight-mib') options.maxInFlightMiB = parsed else options.concurrency = parsed } 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 + ) { + throw new Error('--max-in-flight-mib must be between 64 and 4096') + } if (options.cursor) { if (explicitOrder && options.order !== options.cursor.order) { throw new Error('--order must match the checkpoint cursor') @@ -190,38 +207,67 @@ export async function checkDatabase(): Promise { .limit(0) } -/** Stops scheduling on failure and lets active writes settle before the process exits. */ +interface WorkerOptions { + signal?: AbortSignal + byteBudget?: { maxBytes: number; sizeOf: (row: T) => number } +} + +/** Bounds active payload bytes and workers; drains writes on failure or shutdown. */ export async function runBackfillWorkers( rows: T[], concurrency: number, - processRow: (row: T) => Promise -): Promise { - let cursor = 0 + processRow: (row: T) => Promise, + { signal, byteBudget }: WorkerOptions = {} +): Promise { + let completed = 0 + let activeBytes = 0 let failure: Error | undefined - const worker = async () => { - while (!failure && cursor < rows.length) { - const row = rows[cursor++] - try { + const active = new Set>() + for (const row of rows) { + const bytes = byteBudget ? byteBudget.sizeOf(row) : 0 + if (!Number.isSafeInteger(bytes) || bytes < 0 || (byteBudget && bytes > byteBudget.maxBytes)) { + failure = new Error( + 'Candidate payload exceeds the backfill byte budget or has an invalid size' + ) + break + } + while ( + active.size >= concurrency || + (byteBudget && activeBytes + bytes > byteBudget.maxBytes) + ) { + await Promise.race(active) + } + if (failure || signal?.aborted) break + activeBytes += bytes + const task = Promise.resolve() + .then(async () => { await processRow(row) - } catch (error) { + completed++ + }) + .catch((error) => { failure ??= toError(error) - } - } + }) + .finally(() => { + activeBytes -= bytes + active.delete(task) + }) + active.add(task) } - await Promise.all(Array.from({ length: Math.min(concurrency, rows.length) }, worker)) + await Promise.all(active) if (failure) throw failure + return completed } /** Externalize inline heavy execution_data into the large-value store. */ export async function backfillTraceStorage( - options: Options -): Promise<{ migrated: number; recoveredOwners: number }> { + options: Options, + signal?: AbortSignal +): Promise<{ migrated: number }> { await checkDatabase() logger.info('Database schema and read checks passed') - if (options.checkOnly) return { migrated: 0, recoveredOwners: 0 } + if (options.checkOnly) return { migrated: 0 } let migrated = 0 - let recoveredOwners = 0 let skipped = 0 let cursor = options.cursor const direction = options.order === 'oldest' ? asc : desc @@ -232,12 +278,46 @@ export async function backfillTraceStorage( sql`NOT (${workflowExecutionLogs.executionData} ? ${TRACE_STORE_REF_KEY})` ) const startedAt = Date.now() + const timings = { + scan: { calls: 0, totalMs: 0 }, + size: { calls: 0, totalMs: 0 }, + read: { calls: 0, totalMs: 0 }, + externalize: { calls: 0, totalMs: 0 }, + commit: { calls: 0, totalMs: 0 }, + } + const measure = async (stage: keyof typeof timings, operation: () => PromiseLike) => { + const start = performance.now() + try { + return await operation() + } finally { + timings[stage].calls++ + timings[stage].totalMs += performance.now() - start + } + } + const reportCheckpoint = () => { + if (!cursor) return + logger.info('Backfill checkpoint', { + startedAt: cursor.startedAt, + id: cursor.id, + cursor: Buffer.from(JSON.stringify(cursor)).toString('base64url'), + }) + } logger.info('Scanning execution logs', { order: options.order, before: options.before }) + reportCheckpoint() const reportProgress = () => { 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} | ${rowsPerSecond.toFixed(1)} rows/s | elapsed ${formatDuration(elapsedMs)}`, + { + rssMiB: Math.round(process.memoryUsage().rss / MIB), + stages: Object.fromEntries( + Object.entries(timings).map(([stage, { calls, totalMs }]) => [ + stage, + { calls, averageMs: calls ? Math.round(totalMs / calls) : 0 }, + ]) + ), + } ) } reportProgress() @@ -246,6 +326,7 @@ export async function backfillTraceStorage( try { for (let batch = 0; batch < options.maxBatches; batch++) { + if (signal?.aborted) break /** * Seek by indexed time before checking JSON or joining workflows. Limiting * eligible rows here can scan the entire backlog to fill one page. @@ -262,107 +343,158 @@ export async function backfillTraceStorage( sql`(${workflowExecutionLogs.startedAt}, ${workflowExecutionLogs.id}) < (${cursor.startedAt}::timestamp, ${cursor.id})` ) : undefined - const rows = await db - .select({ - id: workflowExecutionLogs.id, - startedAt: sql`to_char(${workflowExecutionLogs.startedAt}, 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, - }) - .from(workflowExecutionLogs) - .where( - and( - sql`${workflowExecutionLogs.startedAt} < (${options.before}::timestamptz AT TIME ZONE 'UTC')`, - cursorPredicate + const rows = await measure('scan', () => + db + .select({ + id: workflowExecutionLogs.id, + startedAt: sql`to_char(${workflowExecutionLogs.startedAt}, 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, + }) + .from(workflowExecutionLogs) + .where( + and( + sql`${workflowExecutionLogs.startedAt} < (${options.before}::timestamptz AT TIME ZONE 'UTC')`, + cursorPredicate + ) ) - ) - .orderBy(direction(workflowExecutionLogs.startedAt), direction(workflowExecutionLogs.id)) - .limit(TRACE_BATCH_SIZE) + .orderBy(direction(workflowExecutionLogs.startedAt), direction(workflowExecutionLogs.id)) + .limit(Math.max(TRACE_BATCH_SIZE, options.concurrency * 2)) + ) if (rows.length === 0) break + if (signal?.aborted) break - await runBackfillWorkers(rows, options.concurrency, async ({ id }) => { - try { - /** Reject oversized JSON in SQL before the driver materializes it. */ - const [row] = await db - .select({ - id: workflowExecutionLogs.id, - workspaceId: workflowExecutionLogs.workspaceId, - workflowId: workflowExecutionLogs.workflowId, - executionId: workflowExecutionLogs.executionId, - workflowOwnerUserId: workflow.userId, - payloadBytes: sql`octet_length(${workflowExecutionLogs.executionData}::text)`, - executionData: sql | null>`CASE - WHEN octet_length(${workflowExecutionLogs.executionData}::text) <= ${MAX_DURABLE_LARGE_VALUE_BYTES} + /** Only sizes leave PostgreSQL until the scheduler reserves payload capacity. */ + const sizes = await measure('size', () => + db + .select({ + id: workflowExecutionLogs.id, + payloadBytes: sql`octet_length(${workflowExecutionLogs.executionData}::text)`, + }) + .from(workflowExecutionLogs) + .innerJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id)) + .where( + and( + inArray( + workflowExecutionLogs.id, + rows.map(({ id }) => id) + ), + pending + ) + ) + .limit(rows.length) + ) + 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` + ) + } + } + 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 }] + }) + skipped += rows.length - candidates.length + + const completed = await runBackfillWorkers( + candidates, + options.concurrency, + async ({ id, payloadBytes }) => { + try { + /** Reject oversized JSON in SQL before the driver materializes it. */ + const [row] = await measure('read', () => + db + .select({ + id: workflowExecutionLogs.id, + workspaceId: workflowExecutionLogs.workspaceId, + workflowId: workflowExecutionLogs.workflowId, + executionId: workflowExecutionLogs.executionId, + workflowOwnerUserId: workflow.userId, + payloadBytes: sql`octet_length(${workflowExecutionLogs.executionData}::text)`, + executionData: sql | null>`CASE + WHEN octet_length(${workflowExecutionLogs.executionData}::text) <= ${payloadBytes} THEN ${workflowExecutionLogs.executionData} ELSE NULL END`, - }) - .from(workflowExecutionLogs) - .innerJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id)) - .where(and(eq(workflowExecutionLogs.id, id), pending)) - .limit(1) - /** Skip running, deleted-workflow, already externalized, or removed rows. */ - if (!row) { - skipped++ - return - } - if (row.payloadBytes > MAX_DURABLE_LARGE_VALUE_BYTES) { - throw new Error( - `Execution payload is ${row.payloadBytes} bytes, exceeding the ${MAX_DURABLE_LARGE_VALUE_BYTES}-byte backfill limit` + }) + .from(workflowExecutionLogs) + .innerJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id)) + .where(and(eq(workflowExecutionLogs.id, id), pending)) + .limit(1) + ) + /** Skip running, deleted-workflow, already externalized, or removed rows. */ + if (!row) { + skipped++ + return + } + if (!row.workflowOwnerUserId) throw new Error('Workflow owner is missing') + if (row.payloadBytes > payloadBytes) { + throw new Error( + `Execution payload grew from ${payloadBytes} to ${row.payloadBytes} bytes after reserving memory; resume from the last checkpoint` + ) + } + const executionData = row.executionData + if (!executionData) throw new Error('Execution data is missing') + const traceSpanCount = countTraceSpans(executionData.traceSpans) + executionData.hasTraceSpans = traceSpanCount > 0 + executionData.traceSpanCount = traceSpanCount + stripSpanCosts(executionData.traceSpans) + /** + * Use the workflow's FK-backed owner for file metadata: the historical + * execution user may be deleted and stays unchanged inside the payload. + */ + const slim = await measure('externalize', () => + externalizeExecutionData( + executionData, + { + workspaceId: row.workspaceId, + workflowId: row.workflowId, + executionId: row.executionId, + userId: row.workflowOwnerUserId, + }, + { throwOnError: true } + ) ) - } - const executionData = row.executionData - if (!executionData) throw new Error('Execution data is missing') - const traceSpanCount = countTraceSpans(executionData.traceSpans) - executionData.hasTraceSpans = traceSpanCount > 0 - executionData.traceSpanCount = traceSpanCount - stripSpanCosts(executionData.traceSpans) - /** - * workspace_files.user_id (NOT NULL) needs an owner. Most rows carry - * it under executionData.environment.userId; the legacy workflow-log - * endpoint wrote an empty userId, so recover the workflow owner that - * the corrected endpoint would have persisted. - */ - const environment = executionData.environment as { userId?: string } | undefined - const storedOwnerUserId = environment?.userId - const ownerUserId = storedOwnerUserId || row.workflowOwnerUserId - const slim = await externalizeExecutionData( - executionData, - { - workspaceId: row.workspaceId, - workflowId: row.workflowId, - executionId: row.executionId, - userId: ownerUserId, - }, - { throwOnError: true } - ) - if (!(TRACE_STORE_REF_KEY in slim)) { - throw new Error('Trace storage did not return a durable reference') - } + if (!(TRACE_STORE_REF_KEY in slim)) { + throw new Error('Trace storage did not return a durable reference') + } - await db.transaction(async (tx) => { - await tx - .update(workflowExecutionLogs) - .set({ executionData: slim }) - .where(eq(workflowExecutionLogs.id, row.id)) + await measure('commit', () => + db.transaction(async (tx) => { + await tx + .update(workflowExecutionLogs) + .set({ executionData: slim }) + .where(eq(workflowExecutionLogs.id, row.id)) - await replaceLargeValueReferenceKeysWithClient( - tx, - { - workspaceId: row.workspaceId, - workflowId: row.workflowId, - executionId: row.executionId, - source: 'execution_log', - }, - collectLargeValueReferenceKeys(slim) + await replaceLargeValueReferenceKeysWithClient( + tx, + { + workspaceId: row.workspaceId, + workflowId: row.workflowId, + executionId: row.executionId, + source: 'execution_log', + }, + collectLargeValueReferenceKeys(slim) + ) + }) ) - }) - migrated++ - if (!storedOwnerUserId) recoveredOwners++ - } catch (error) { - throw new Error(`Backfill failed for execution log ${id}`, { cause: error }) + migrated++ + } catch (error) { + throw new Error(`Backfill failed for execution log ${id}`, { cause: error }) + } + }, + { + signal, + byteBudget: { + maxBytes: options.maxInFlightMiB * MIB, + sizeOf: ({ payloadBytes }) => payloadBytes, + }, } - }) + ) + + if (completed !== candidates.length) break const last = rows[rows.length - 1] cursor = { @@ -372,20 +504,17 @@ export async function backfillTraceStorage( startedAt: last.startedAt, id: last.id, } - logger.info('Backfill checkpoint', { - startedAt: cursor.startedAt, - id: cursor.id, - cursor: Buffer.from(JSON.stringify(cursor)).toString('base64url'), - }) + reportCheckpoint() reportProgress() } } finally { clearInterval(progressTimer) reportProgress() + reportCheckpoint() } - return { migrated, recoveredOwners } + return { migrated } } async function main(): Promise { @@ -394,18 +523,40 @@ async function main(): Promise { logger.info('Starting trace backfill', { concurrency: options.concurrency, + maxInFlightMiB: options.maxInFlightMiB, checkOnly: options.checkOnly, }) - const result = await backfillTraceStorage(options) - logger.info(options.checkOnly ? 'Database check complete' : 'Backfill run finished', { - ...result, - elapsedSeconds: (Date.now() - startedAt) / 1000, - }) + const controller = new AbortController() + const stop = (signal: NodeJS.Signals) => { + if (controller.signal.aborted) return + logger.warn('Stopping backfill; draining active workers before exiting', { signal }) + process.exitCode = signal === 'SIGINT' ? 130 : 143 + controller.abort() + } + process.on('SIGINT', stop) + process.on('SIGTERM', stop) + try { + const result = await backfillTraceStorage(options, controller.signal) + logger.info( + controller.signal.aborted + ? 'Backfill interrupted' + : options.checkOnly + ? 'Database check complete' + : 'Backfill run finished', + { + ...result, + elapsedSeconds: (Date.now() - startedAt) / 1000, + } + ) + } finally { + process.off('SIGINT', stop) + process.off('SIGTERM', stop) + } } if (import.meta.main) { main().then( - () => process.exit(0), + () => process.exit(process.exitCode ?? 0), (error) => { logger.error('Backfill failed', describeError(error)) process.exit(1) From e9ee2f733041842392bce2f859c87377870384c5 Mon Sep 17 00:00:00 2001 From: Theodore Li Date: Fri, 18 Sep 2026 12:19:53 -0700 Subject: [PATCH 3/3] fix(traces): use execution pool throughout backfill --- apps/sim/scripts/backfill-trace-spans.test.ts | 33 ++++++++++++++----- apps/sim/scripts/backfill-trace-spans.ts | 13 ++++---- 2 files changed, 32 insertions(+), 14 deletions(-) diff --git a/apps/sim/scripts/backfill-trace-spans.test.ts b/apps/sim/scripts/backfill-trace-spans.test.ts index cae6b70d486..57c0ae05979 100644 --- a/apps/sim/scripts/backfill-trace-spans.test.ts +++ b/apps/sim/scripts/backfill-trace-spans.test.ts @@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { MAX_DURABLE_LARGE_VALUE_BYTES } from '@/lib/execution/payloads/limits' const { + mockPrimaryRead, mockRead, mockInfo, mockDataRead, @@ -13,6 +14,7 @@ const { mockExternalize, mockReplaceReferences, } = vi.hoisted(() => ({ + mockPrimaryRead: vi.fn(), mockRead: vi.fn(), mockInfo: vi.fn(), mockDataRead: vi.fn(), @@ -23,7 +25,7 @@ const { })) vi.mock('@sim/db', () => { - const db = { + const execDb = { select: () => { const query = { from: () => query, @@ -36,7 +38,13 @@ vi.mock('@sim/db', () => { }, transaction: mockTransaction, } - return { db, dbFor: () => db } + return { + db: { select: () => ({ from: () => ({ limit: mockPrimaryRead }) }) }, + dbFor: (role: string) => { + if (role !== 'exec') throw new Error(`Unexpected database role: ${role}`) + return execDb + }, + } }) vi.mock('@sim/logger', () => ({ @@ -58,6 +66,10 @@ import { backfillTraceStorage, parseArgs, runBackfillWorkers } from '@/scripts/b beforeEach(() => { vi.resetAllMocks() + mockPrimaryRead.mockImplementation((limit: number) => { + if (limit !== 0) throw new Error('Execution payloads must use the execution pool') + return Promise.resolve([]) + }) mockRead.mockImplementation((limit: number) => limit === 0 ? Promise.resolve([]) : mockDataRead(limit) ) @@ -257,7 +269,7 @@ describe('trace backfill', () => { it('fails its schema check before uploading anything and preserves the database cause', async () => { const cause = new Error('column "size_bytes" does not exist') const error = new Error('Failed query', { cause }) - mockRead.mockRejectedValueOnce(error) + mockPrimaryRead.mockRejectedValueOnce(error) await expect(backfillTraceStorage(options)).rejects.toBe(error) expect(mockDataRead).not.toHaveBeenCalled() expect(mockExternalize).not.toHaveBeenCalled() @@ -266,13 +278,14 @@ describe('trace backfill', () => { it('check-only performs no uploads, writes, or payload reads', async () => { await backfillTraceStorage({ ...options, checkOnly: true }) - expect(mockRead).toHaveBeenCalledTimes(5) + expect(mockPrimaryRead).toHaveBeenCalledExactlyOnceWith(0) + expect(mockRead).toHaveBeenCalledTimes(4) expect(mockRead.mock.calls.every(([limit]) => limit === 0)).toBe(true) expect(mockExternalize).not.toHaveBeenCalled() expect(mockTransaction).not.toHaveBeenCalled() }) - it('commits a durable pointer and references with the workflow owner', async () => { + it('reads and commits logs and references through the execution pool with the workflow owner', async () => { mockDataRead .mockResolvedValueOnce([candidateMetadata]) .mockResolvedValueOnce([candidate]) @@ -280,6 +293,10 @@ describe('trace backfill', () => { await expect(backfillTraceStorage(options)).resolves.toEqual({ migrated: 1, }) + expect(mockPrimaryRead).toHaveBeenCalledExactlyOnceWith(0) + expect(mockRead).toHaveBeenCalledTimes(7) + expect(mockDataRead).toHaveBeenCalledTimes(3) + expect(mockTransaction).toHaveBeenCalledOnce() expect(mockExternalize).toHaveBeenCalledWith( expect.objectContaining({ hasTraceSpans: true, traceSpanCount: 2 }), { @@ -454,19 +471,19 @@ describe('trace backfill', () => { .mockResolvedValueOnce([candidate]) .mockResolvedValueOnce([candidate]) await backfillTraceStorage(options) - expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 0, 100, 1, 1]) + expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 100, 1, 1]) expect(mockExternalize).toHaveBeenCalledOnce() }) it('keeps the candidate page at 100 rows with fifty workers', async () => { await backfillTraceStorage({ ...options, concurrency: 50 }) - expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 0, 100]) + expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 100]) expect(mockExternalize).not.toHaveBeenCalled() }) it('scales the bounded metadata page to feed higher concurrency', async () => { await backfillTraceStorage({ ...options, concurrency: 500 }) - expect(mockRead.mock.calls.map(([limit]) => limit)).toEqual([0, 0, 0, 0, 0, 1000]) + 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 () => { diff --git a/apps/sim/scripts/backfill-trace-spans.ts b/apps/sim/scripts/backfill-trace-spans.ts index 25186ab869a..0862e0aee74 100644 --- a/apps/sim/scripts/backfill-trace-spans.ts +++ b/apps/sim/scripts/backfill-trace-spans.ts @@ -187,11 +187,11 @@ export function parseArgs(argv: string[]): Options { /** Validates the metadata schema and SELECT permissions without reading payloads or writing. */ export async function checkDatabase(): Promise { await db.select().from(workspaceFiles).limit(0) - await db.select().from(executionLargeValueReferences).limit(0) const execDb = dbFor('exec') + await execDb.select().from(executionLargeValueReferences).limit(0) await execDb.select().from(executionLargeValues).limit(0) await execDb.select().from(executionLargeValueDependencies).limit(0) - await db + await execDb .select({ id: workflowExecutionLogs.id, workspaceId: workflowExecutionLogs.workspaceId, @@ -267,6 +267,7 @@ export async function backfillTraceStorage( logger.info('Database schema and read checks passed') if (options.checkOnly) return { migrated: 0 } + const execDb = dbFor('exec') let migrated = 0 let skipped = 0 let cursor = options.cursor @@ -344,7 +345,7 @@ export async function backfillTraceStorage( ) : undefined const rows = await measure('scan', () => - db + execDb .select({ id: workflowExecutionLogs.id, startedAt: sql`to_char(${workflowExecutionLogs.startedAt}, 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, @@ -365,7 +366,7 @@ export async function backfillTraceStorage( /** Only sizes leave PostgreSQL until the scheduler reserves payload capacity. */ const sizes = await measure('size', () => - db + execDb .select({ id: workflowExecutionLogs.id, payloadBytes: sql`octet_length(${workflowExecutionLogs.executionData}::text)`, @@ -404,7 +405,7 @@ export async function backfillTraceStorage( try { /** Reject oversized JSON in SQL before the driver materializes it. */ const [row] = await measure('read', () => - db + execDb .select({ id: workflowExecutionLogs.id, workspaceId: workflowExecutionLogs.workspaceId, @@ -461,7 +462,7 @@ export async function backfillTraceStorage( } await measure('commit', () => - db.transaction(async (tx) => { + execDb.transaction(async (tx) => { await tx .update(workflowExecutionLogs) .set({ executionData: slim })