fix(traces): make trace backfills bounded and resumable (#7974)

* wip

* fix(traces): bound and resume trace backfill workers

* fix(traces): use execution pool throughout backfill

Author: TheodoreSpeaks

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'))
 

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,

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 = {

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<string, unknown>,
-  context: TraceStoreWriteContext
+  context: TraceStoreWriteContext,
+  options: { throwOnError?: boolean } = {}
 ): Promise<Record<string, unknown>> {
   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
   }