improvement(cleanup): cleanup refs along in logs cleanup job

Author: icecrasher321

apps/sim/background/cleanup-logs.test.ts

@@ -0,0 +1,104 @@
+/**
+ * @vitest-environment node
+ */
+import { describe, expect, it, vi } from 'vitest'
+import { collectExecutionLargeValueKeys } from '@/background/cleanup-logs'
+
+vi.mock('@trigger.dev/sdk', () => ({
+  task: vi.fn((definition) => definition),
+}))
+
+describe('collectExecutionLargeValueKeys', () => {
+  it('collects large-value keys owned by the purged execution', () => {
+    const ownedRef = {
+      __simLargeValueRef: true,
+      version: 1,
+      id: 'lv_OWNEDREF0001',
+      kind: 'object',
+      size: 128,
+      key: 'execution/workspace-1/workflow-1/execution-1/large-value-lv_OWNEDREF0001.json',
+      executionId: 'execution-1',
+    }
+    const ownedManifest = {
+      __simLargeArrayManifest: true,
+      version: 2,
+      kind: 'array',
+      totalCount: 1,
+      chunkCount: 1,
+      byteSize: 128,
+      chunks: [
+        {
+          ref: {
+            __simLargeValueRef: true,
+            version: 1,
+            id: 'lv_CHUNKREF0001',
+            kind: 'array',
+            size: 128,
+            key: 'execution/workspace-1/workflow-1/execution-1/large-value-lv_CHUNKREF0001.json',
+            executionId: 'execution-1',
+          },
+          count: 1,
+          byteSize: 128,
+        },
+      ],
+      preview: [],
+    }
+    const inheritedRef = {
+      __simLargeValueRef: true,
+      version: 1,
+      id: 'lv_INHERITED001',
+      kind: 'object',
+      size: 128,
+      key: 'execution/workspace-1/workflow-1/source-execution/large-value-lv_INHERITED001.json',
+      executionId: 'source-execution',
+    }
+
+    const keys = collectExecutionLargeValueKeys(
+      {
+        output: {
+          ownedRef,
+          ownedManifest,
+          inheritedRef,
+        },
+      },
+      'execution-1'
+    )
+
+    expect(keys.sort()).toEqual([
+      'execution/workspace-1/workflow-1/execution-1/large-value-lv_CHUNKREF0001.json',
+      'execution/workspace-1/workflow-1/execution-1/large-value-lv_OWNEDREF0001.json',
+    ])
+  })
+
+  it('deduplicates repeated refs and ignores keyless cache-only refs', () => {
+    const ref = {
+      __simLargeValueRef: true,
+      version: 1,
+      id: 'lv_REPEATREF001',
+      kind: 'string',
+      size: 128,
+      key: 'execution/workspace-1/workflow-1/execution-1/large-value-lv_REPEATREF001.json',
+      executionId: 'execution-1',
+    }
+
+    const keys = collectExecutionLargeValueKeys(
+      {
+        first: ref,
+        second: ref,
+        keyless: {
+          __simLargeValueRef: true,
+          version: 1,
+          id: 'lv_KEYLESSREF01',
+          kind: 'string',
+          size: 128,
+          executionId: 'execution-1',
+        },
+      },
+      'execution-1'
+    )
+
+    expect(keys).toEqual([
+      'execution/workspace-1/workflow-1/execution-1/large-value-lv_REPEATREF001.json',
+    ])
+  })
+})

apps/sim/background/cleanup-logs.ts

@@ -1,14 +1,15 @@
 import { db } from '@sim/db'
-import { jobExecutionLogs, workflowExecutionLogs } from '@sim/db/schema'
+import { jobExecutionLogs, pausedExecutions, workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { task } from '@trigger.dev/sdk'
-import { and, inArray, lt } from 'drizzle-orm'
+import { and, eq, inArray, isNull, lt, notInArray, or } from 'drizzle-orm'
 import { type CleanupJobPayload, resolveCleanupScope } from '@/lib/billing/cleanup-dispatcher'
 import {
   batchDeleteByWorkspaceAndTimestamp,
   chunkedBatchDelete,
   type TableCleanupResult,
 } from '@/lib/cleanup/batch-delete'
+import { isLargeValueRef } from '@/lib/execution/payloads/large-value-ref'
 import { snapshotService } from '@/lib/logs/execution/snapshot/service'
 import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
 import { deleteFileMetadata } from '@/lib/uploads/server/metadata'
@@ -19,6 +20,60 @@ interface FileDeleteStats {
   filesTotal: number
   filesDeleted: number
   filesDeleteFailed: number
+  largeValuesTotal: number
+  largeValuesDeleted: number
+  largeValuesDeleteFailed: number
+}
+
+const RESUMABLE_PAUSED_STATUSES = ['paused', 'partially_resumed', 'cancelling']
+
+export function collectExecutionLargeValueKeys(value: unknown, executionId: string): string[] {
+  const keys = new Set<string>()
+  collectExecutionLargeValueKeysInto(value, executionId, new WeakSet<object>(), keys)
+  return Array.from(keys)
+}
+
+function getExecutionIdFromStorageKey(key: string): string | undefined {
+  const parts = key.split('/')
+  if (parts[0] !== 'execution' || parts.length < 5) {
+    return undefined
+  }
+  return parts[3]
+}
+
+function collectExecutionLargeValueKeysInto(
+  value: unknown,
+  executionId: string,
+  seen: WeakSet<object>,
+  keys: Set<string>
+): void {
+  if (!value || typeof value !== 'object') {
+    return
+  }
+
+  if (seen.has(value)) {
+    return
+  }
+
+  if (isLargeValueRef(value)) {
+    if (value.key && getExecutionIdFromStorageKey(value.key) === executionId) {
+      keys.add(value.key)
+    }
+    return
+  }
+
+  seen.add(value)
+
+  if (Array.isArray(value)) {
+    for (const item of value) {
+      collectExecutionLargeValueKeysInto(item, executionId, seen, keys)
+    }
+    return
+  }
+
+  for (const entryValue of Object.values(value)) {
+    collectExecutionLargeValueKeysInto(entryValue, executionId, seen, keys)
+  }
 }
 
 async function deleteExecutionFiles(files: unknown, stats: FileDeleteStats): Promise<void> {
@@ -41,36 +96,99 @@ async function deleteExecutionFiles(files: unknown, stats: FileDeleteStats): Pro
   )
 }
 
+async function deleteLargeValueStorageKeys(keys: string[], stats: FileDeleteStats): Promise<void> {
+  if (!isUsingCloudStorage() || keys.length === 0) return
+
+  const uniqueKeys = Array.from(new Set(keys))
+  stats.largeValuesTotal += uniqueKeys.length
+
+  await Promise.all(
+    uniqueKeys.map(async (key) => {
+      try {
+        await StorageService.deleteFile({ key, context: 'execution' })
+        await deleteFileMetadata(key)
+        stats.largeValuesDeleted++
+      } catch (error) {
+        stats.largeValuesDeleteFailed++
+        logger.error(`Failed to delete large execution value ${key}:`, { error })
+      }
+    })
+  )
+}
+
 async function cleanupWorkflowExecutionLogs(
   workspaceIds: string[],
   retentionDate: Date,
   label: string
 ): Promise<TableCleanupResult & FileDeleteStats> {
-  const fileStats: FileDeleteStats = { filesTotal: 0, filesDeleted: 0, filesDeleteFailed: 0 }
+  const fileStats: FileDeleteStats = {
+    filesTotal: 0,
+    filesDeleted: 0,
+    filesDeleteFailed: 0,
+    largeValuesTotal: 0,
+    largeValuesDeleted: 0,
+    largeValuesDeleteFailed: 0,
+  }
 
   const dbStats = await chunkedBatchDelete({
     tableDef: workflowExecutionLogs,
     workspaceIds,
     tableName: `${label}/workflow_execution_logs`,
     selectChunk: (chunkIds, limit) =>
       db
-        .select({ id: workflowExecutionLogs.id, files: workflowExecutionLogs.files })
+        .select({
+          id: workflowExecutionLogs.id,
+          executionId: workflowExecutionLogs.executionId,
+          executionData: workflowExecutionLogs.executionData,
+          files: workflowExecutionLogs.files,
+        })
         .from(workflowExecutionLogs)
+        .leftJoin(
+          pausedExecutions,
+          eq(pausedExecutions.executionId, workflowExecutionLogs.executionId)
+        )
         .where(
           and(
             inArray(workflowExecutionLogs.workspaceId, chunkIds),
-            lt(workflowExecutionLogs.startedAt, retentionDate)
+            lt(workflowExecutionLogs.startedAt, retentionDate),
+            or(
+              isNull(pausedExecutions.status),
+              notInArray(pausedExecutions.status, RESUMABLE_PAUSED_STATUSES)
+            )
           )
         )
         .limit(limit),
     onBatch: async (rows) => {
-      for (const row of rows) await deleteExecutionFiles(row.files, fileStats)
+      for (const row of rows) {
+        await deleteExecutionFiles(row.files, fileStats)
+        await deleteLargeValueStorageKeys(
+          collectExecutionLargeValueKeys(row.executionData, row.executionId),
+          fileStats
+        )
+      }
     },
   })
 
   return { ...dbStats, ...fileStats }
 }
 
+async function cleanupFreePlanOrphanedSnapshots(
+  payload: CleanupJobPayload,
+  retentionHours: number
+): Promise<void> {
+  if (payload.plan !== 'free') {
+    return
+  }
+
+  try {
+    const retentionDays = Math.floor(retentionHours / 24)
+    const snapshotsCleaned = await snapshotService.cleanupOrphanedSnapshots(retentionDays + 1)
+    logger.info(`Cleaned up ${snapshotsCleaned} orphaned snapshots`)
+  } catch (snapshotError) {
+    logger.error('Error cleaning up orphaned snapshots:', { snapshotError })
+  }
+}
+
 export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void> {
   const startTime = Date.now()
 
@@ -82,12 +200,14 @@ export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void>
 
   const { workspaceIds, retentionHours, label } = scope
 
+  const retentionDate = new Date(Date.now() - retentionHours * 60 * 60 * 1000)
+
   if (workspaceIds.length === 0) {
     logger.info(`[${label}] No workspaces to process`)
+    await cleanupFreePlanOrphanedSnapshots(payload, retentionHours)
     return
   }
 
-  const retentionDate = new Date(Date.now() - retentionHours * 60 * 60 * 1000)
   logger.info(
     `[${label}] Cleaning ${workspaceIds.length} workspaces, cutoff: ${retentionDate.toISOString()}`
   )
@@ -96,6 +216,9 @@ export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void>
   logger.info(
     `[${label}] workflow_execution_logs files: ${workflowResults.filesDeleted}/${workflowResults.filesTotal} deleted, ${workflowResults.filesDeleteFailed} failed`
   )
+  logger.info(
+    `[${label}] workflow_execution_logs large values: ${workflowResults.largeValuesDeleted}/${workflowResults.largeValuesTotal} deleted, ${workflowResults.largeValuesDeleteFailed} failed`
+  )
 
   await batchDeleteByWorkspaceAndTimestamp({
     tableDef: jobExecutionLogs,
@@ -106,16 +229,7 @@ export async function runCleanupLogs(payload: CleanupJobPayload): Promise<void>
     tableName: `${label}/job_execution_logs`,
   })
 
-  // Snapshot cleanup runs only on the free job to avoid running it N times for N enterprise workspaces.
-  if (payload.plan === 'free') {
-    try {
-      const retentionDays = Math.floor(retentionHours / 24)
-      const snapshotsCleaned = await snapshotService.cleanupOrphanedSnapshots(retentionDays + 1)
-      logger.info(`Cleaned up ${snapshotsCleaned} orphaned snapshots`)
-    } catch (snapshotError) {
-      logger.error('Error cleaning up orphaned snapshots:', { snapshotError })
-    }
-  }
+  await cleanupFreePlanOrphanedSnapshots(payload, retentionHours)
 
   const timeElapsed = (Date.now() - startTime) / 1000
   logger.info(`[${label}] Job completed in ${timeElapsed.toFixed(2)}s`)