improvement(redis-cleanup): schedule, async workflow, hitl bae64 cache cleanup

Author: icecrasher321

apps/sim/background/schedule-execution.ts

@@ -21,6 +21,7 @@ import {
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
+import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server'
 import {
   executeWorkflowCore,
   wasExecutionFinalizedByCore,
@@ -348,6 +349,13 @@ async function runWorkflowExecution({
     })
 
     throw error
+  } finally {
+    void cleanupExecutionBase64Cache(executionId).catch((cleanupError) => {
+      logger.warn(`[${requestId}] Failed to cleanup base64 cache`, {
+        executionId,
+        error: toError(cleanupError).message,
+      })
+    })
   }
 }
 

apps/sim/background/workflow-execution.ts

@@ -7,6 +7,7 @@ import { createTimeoutAbortController, getTimeoutErrorMessage } from '@/lib/core
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
+import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server'
 import {
   executeWorkflowCore,
   wasExecutionFinalizedByCore,
@@ -196,6 +197,13 @@ export async function executeWorkflowJob(payload: WorkflowExecutionPayload) {
       })
 
       throw error
+    } finally {
+      void cleanupExecutionBase64Cache(executionId).catch((cleanupError) => {
+        logger.warn(`[${requestId}] Failed to cleanup base64 cache`, {
+          executionId,
+          error: toError(cleanupError).message,
+        })
+      })
     }
   })
 }

apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts

@@ -16,6 +16,7 @@ import {
 import { compactBlockLogs, compactExecutionPayload } from '@/lib/execution/payloads/serializer'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
+import { cleanupExecutionBase64Cache } from '@/lib/uploads/utils/user-file-base64.server'
 import { executeWorkflowCore } from '@/lib/workflows/executor/execution-core'
 import type { ExecutionEvent } from '@/lib/workflows/executor/execution-events'
 import { ExecutionSnapshot } from '@/executor/execution/snapshot'
@@ -1363,6 +1364,12 @@ export class PauseResumeManager {
           })
         })
       }
+      void cleanupExecutionBase64Cache(resumeExecutionId).catch((error) => {
+        logger.warn('Failed to cleanup base64 cache for resume execution', {
+          resumeExecutionId,
+          error: toError(error).message,
+        })
+      })
     }
 
     if (executionError || !result) {