improvement(execution): memory usage for aggregated results

Author: icecrasher321

apps/sim/app/api/function/execute/route.test.ts

@@ -52,6 +52,7 @@ import { POST } from '@/app/api/function/execute/route'
 describe('Function Execute API Route', () => {
   beforeEach(() => {
     vi.clearAllMocks()
+    featureFlagsMock.isE2bEnabled = false
 
     hybridAuthMockFns.mockCheckInternalAuth.mockResolvedValue({
       success: true,
@@ -240,6 +241,33 @@ describe('Function Execute API Route', () => {
       expect(response.status).toBe(200)
       expect(data.success).toBe(true)
     })
+
+    it('rejects large refs in runtimes without ref-native helpers', async () => {
+      featureFlagsMock.isE2bEnabled = true
+      const req = createMockRequest('POST', {
+        code: 'echo "${__blockRef_0}"',
+        language: 'shell',
+        contextVariables: {
+          __blockRef_0: {
+            __simLargeValueRef: true,
+            version: 1,
+            id: 'lv_ABCDEFGHIJKL',
+            kind: 'array',
+            size: 12 * 1024 * 1024,
+            executionId: 'execution-1',
+          },
+        },
+      })
+
+      const response = await POST(req)
+      const data = await response.json()
+
+      expect(response.status).toBe(500)
+      expect(data.success).toBe(false)
+      expect(data.error).toContain(
+        'Large execution values require the JavaScript isolated-vm runtime'
+      )
+    })
   })
 
   describe('Template Variable Resolution', () => {

apps/sim/app/api/function/execute/route.ts

@@ -14,7 +14,7 @@ import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
 import { executeInE2B, executeShellInE2B } from '@/lib/execution/e2b'
 import { executeInIsolatedVM, type IsolatedVMBrokerHandler } from '@/lib/execution/isolated-vm'
 import { CodeLanguage, DEFAULT_CODE_LANGUAGE, isValidCodeLanguage } from '@/lib/execution/languages'
-import { isLargeValueRef } from '@/lib/execution/payloads/large-value-ref'
+import { containsLargeValueRef, isLargeValueRef } from '@/lib/execution/payloads/large-value-ref'
 import {
   MAX_FUNCTION_INLINE_BYTES,
   MAX_INLINE_MATERIALIZATION_BYTES,
@@ -1013,6 +1013,12 @@ export const POST = withRouteHandler(async (req: NextRequest) => {
       contextVariables = { ...codeResolution.contextVariables, ...preResolvedContextVariables }
     }
 
+    if (lang === CodeLanguage.Shell && containsLargeValueRef(contextVariables)) {
+      throw new Error(
+        'Large execution values require the JavaScript isolated-vm runtime. Select a nested field or read the value in a JavaScript function.'
+      )
+    }
+
     let jsImports = ''
     let jsRemainingCode = resolvedCode
     let hasImports = false
@@ -1124,6 +1130,12 @@ export const POST = withRouteHandler(async (req: NextRequest) => {
       !isCustomTool &&
       (lang === CodeLanguage.Python || (lang === CodeLanguage.JavaScript && hasImports))
 
+    if (useE2B && containsLargeValueRef(contextVariables)) {
+      throw new Error(
+        'Large execution values require the JavaScript isolated-vm runtime. Remove imports, select a nested field, or read the value in a JavaScript function without E2B.'
+      )
+    }
+
     if (useE2B) {
       logger.info(`[${requestId}] E2B status`, {
         enabled: isE2bEnabled,

apps/sim/executor/execution/snapshot-serializer.ts

@@ -1,8 +1,16 @@
+import { LARGE_VALUE_THRESHOLD_BYTES } from '@/lib/execution/payloads/large-value-ref'
 import type { DAG } from '@/executor/dag/builder'
 import { ExecutionSnapshot } from '@/executor/execution/snapshot'
 import type { ExecutionMetadata, SerializableExecutionState } from '@/executor/execution/types'
 import type { ExecutionContext, SerializedSnapshot } from '@/executor/types'
 
+function assertSnapshotValueIsCompact(value: unknown, label: string): void {
+  const json = JSON.stringify(value)
+  if (json && Buffer.byteLength(json, 'utf8') > LARGE_VALUE_THRESHOLD_BYTES) {
+    throw new Error(`Cannot serialize pause snapshot with oversized ${label}; compact it first.`)
+  }
+}
+
 function mapFromEntries<T>(map?: Map<string, T>): Record<string, T> | undefined {
   if (!map) return undefined
   return Object.fromEntries(map)
@@ -94,6 +102,9 @@ export function serializePauseSnapshot(
     dagIncomingEdges,
   }
 
+  assertSnapshotValueIsCompact(context.workflowVariables, 'workflow variables')
+  assertSnapshotValueIsCompact(state.loopExecutions, 'loop execution state')
+
   const workspaceId = metadataFromContext?.workspaceId ?? context.workspaceId
   if (!workspaceId) {
     throw new Error(

apps/sim/executor/handlers/variables/variables-handler.test.ts

@@ -0,0 +1,180 @@
+/**
+ * @vitest-environment node
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { clearLargeValueCacheForTests } from '@/lib/execution/payloads/cache'
+import { isLargeValueRef } from '@/lib/execution/payloads/large-value-ref'
+import { BlockType } from '@/executor/constants'
+import { VariablesBlockHandler } from '@/executor/handlers/variables/variables-handler'
+import type { ExecutionContext } from '@/executor/types'
+import type { SerializedBlock } from '@/serializer/types'
+
+const { mockUploadFile } = vi.hoisted(() => ({
+  mockUploadFile: vi.fn(),
+}))
+
+vi.mock('@/lib/uploads', () => ({
+  StorageService: {
+    uploadFile: mockUploadFile,
+  },
+}))
+
+function createContext(overrides: Partial<ExecutionContext> = {}): ExecutionContext {
+  return {
+    workflowId: 'workflow-1',
+    workspaceId: 'workspace-1',
+    executionId: 'execution-1',
+    userId: 'user-1',
+    blockStates: new Map(),
+    blockLogs: [],
+    metadata: { duration: 0 },
+    environmentVariables: {},
+    workflowVariables: {
+      'var-1': { id: 'var-1', name: 'issues', type: 'array', value: [] },
+    },
+    decisions: { router: new Map(), condition: new Map() },
+    loopExecutions: new Map(),
+    executedBlocks: new Set(),
+    activeExecutionPath: new Set(),
+    completedLoops: new Set(),
+    ...overrides,
+  }
+}
+
+function createBlock(): SerializedBlock {
+  return {
+    id: 'variables-block-1',
+    metadata: { id: BlockType.VARIABLES, name: 'Variables' },
+    position: { x: 0, y: 0 },
+    config: { tool: BlockType.VARIABLES, params: {} },
+    inputs: {},
+    outputs: {},
+    enabled: true,
+  }
+}
+
+describe('VariablesBlockHandler', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    clearLargeValueCacheForTests()
+    mockUploadFile.mockImplementation(async ({ customKey }) => ({ key: customKey }))
+  })
+
+  it('preserves small assignments inline', async () => {
+    const handler = new VariablesBlockHandler()
+    const ctx = createContext()
+    const value = [{ key: 'SIM-1', summary: 'Small issue' }]
+
+    const output = await handler.execute(ctx, createBlock(), {
+      variables: [
+        {
+          variableId: 'var-1',
+          variableName: 'issues',
+          type: 'array',
+          value,
+        },
+      ],
+    })
+
+    expect(ctx.workflowVariables?.['var-1'].value).toEqual(value)
+    expect(output).toEqual({ issues: value })
+    expect(mockUploadFile).not.toHaveBeenCalled()
+  })
+
+  it('stores oversized assignments as durable refs in variables and block output', async () => {
+    const handler = new VariablesBlockHandler()
+    const ctx = createContext()
+    const value = Array.from({ length: 120_000 }, (_, index) => ({
+      key: `SIM-${index}`,
+      summary: 'Issue summary that keeps each item small',
+    }))
+
+    const output = await handler.execute(ctx, createBlock(), {
+      variables: [
+        {
+          variableId: 'var-1',
+          variableName: 'issues',
+          type: 'array',
+          value,
+        },
+      ],
+    })
+
+    const storedValue = ctx.workflowVariables?.['var-1'].value
+    expect(isLargeValueRef(storedValue)).toBe(true)
+    expect(output.issues).toBe(storedValue)
+    expect(storedValue).toMatchObject({
+      __simLargeValueRef: true,
+      kind: 'array',
+      executionId: 'execution-1',
+    })
+    expect(mockUploadFile).toHaveBeenCalledWith(
+      expect.objectContaining({
+        context: 'execution',
+        preserveKey: true,
+        customKey: expect.stringContaining('/execution-1/large-value-'),
+      })
+    )
+  })
+
+  it('fails clearly when durable context is missing for oversized assignments', async () => {
+    const handler = new VariablesBlockHandler()
+    const ctx = createContext({ workspaceId: undefined, executionId: undefined })
+    const value = Array.from({ length: 120_000 }, (_, index) => ({
+      key: `SIM-${index}`,
+      summary: 'Issue summary that keeps each item small',
+    }))
+
+    await expect(
+      handler.execute(ctx, createBlock(), {
+        variables: [
+          {
+            variableId: 'var-1',
+            variableName: 'issues',
+            type: 'array',
+            value,
+          },
+        ],
+      })
+    ).rejects.toThrow(
+      'Cannot persist large execution value without workspace, workflow, and execution IDs'
+    )
+
+    expect(mockUploadFile).not.toHaveBeenCalled()
+  })
+
+  it('preserves existing variable metadata when compacting reassignment', async () => {
+    const handler = new VariablesBlockHandler()
+    const ctx = createContext({
+      workflowVariables: {
+        'var-1': {
+          id: 'var-1',
+          name: 'issues',
+          type: 'array',
+          value: [],
+          isExisting: true,
+        },
+      },
+    })
+    const value = [{ key: 'SIM-1', summary: 'Updated' }]
+
+    await handler.execute(ctx, createBlock(), {
+      variables: [
+        {
+          variableId: 'var-1',
+          variableName: 'issues',
+          type: 'array',
+          value,
+        },
+      ],
+    })
+
+    expect(ctx.workflowVariables?.['var-1']).toEqual({
+      id: 'var-1',
+      name: 'issues',
+      type: 'array',
+      value,
+      isExisting: true,
+    })
+  })
+})

apps/sim/executor/handlers/variables/variables-handler.ts

@@ -1,4 +1,6 @@
 import { createLogger } from '@sim/logger'
+import { toError } from '@sim/utils/errors'
+import { compactWorkflowVariableValue } from '@/lib/execution/payloads/serializer'
 import type { BlockOutput } from '@/blocks/types'
 import { BlockType } from '@/executor/constants'
 import type { BlockHandler, ExecutionContext } from '@/executor/types'
@@ -24,36 +26,47 @@ export class VariablesBlockHandler implements BlockHandler {
 
       const assignments = this.parseAssignments(inputs.variables)
 
+      const output: Record<string, any> = {}
+
       for (const assignment of assignments) {
         const existingEntry = assignment.variableId
           ? [assignment.variableId, ctx.workflowVariables[assignment.variableId]]
           : Object.entries(ctx.workflowVariables).find(
               ([_, v]) => v.name === assignment.variableName
             )
+        const value = await this.compactAssignmentValue(ctx, assignment.value)
 
         if (existingEntry?.[1]) {
           const [id, variable] = existingEntry
           ctx.workflowVariables[id] = {
             ...variable,
-            value: assignment.value,
+            value,
           }
+          output[assignment.variableName] = value
         } else {
           logger.warn(`Variable "${assignment.variableName}" not found in workflow variables`)
         }
       }
 
-      const output: Record<string, any> = {}
-      for (const assignment of assignments) {
-        output[assignment.variableName] = assignment.value
-      }
-
       return output
-    } catch (error: any) {
-      logger.error('Variables block execution failed:', error)
-      throw new Error(`Variables block execution failed: ${error.message}`)
+    } catch (error) {
+      const normalizedError = toError(error)
+      logger.error('Variables block execution failed:', normalizedError)
+      throw new Error(`Variables block execution failed: ${normalizedError.message}`)
     }
   }
 
+  private async compactAssignmentValue(ctx: ExecutionContext, value: any): Promise<any> {
+    return compactWorkflowVariableValue(value, {
+      workspaceId: ctx.workspaceId,
+      workflowId: ctx.workflowId,
+      executionId: ctx.executionId,
+      userId: ctx.userId,
+      largeValueExecutionIds: ctx.largeValueExecutionIds,
+      allowLargeValueWorkflowScope: ctx.allowLargeValueWorkflowScope,
+    })
+  }
+
   private parseAssignments(
     assignmentsInput: any
   ): Array<{ variableId?: string; variableName: string; type: string; value: any }> {