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

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, sql } from 'drizzle-orm'
 import { type CleanupJobPayload, resolveCleanupScope } from '@/lib/billing/cleanup-dispatcher'
 import {
  batchDeleteByWorkspaceAndTimestamp,
  chunkedBatchDelete,
  type TableCleanupResult,
 } from '@/lib/cleanup/batch-delete'
+import { collectLargeValueKeys } from '@/lib/execution/payloads/large-execution-value'
 import { snapshotService } from '@/lib/logs/execution/snapshot/service'
 import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
 import { deleteFileMetadata } from '@/lib/uploads/server/metadata'
@@ -19,6 +20,38 @@ interface FileDeleteStats {
  filesTotal: number
  filesDeleted: number
  filesDeleteFailed: number
+ largeValuesTotal: number
+ largeValuesDeleted: number
+ largeValuesDeleteFailed: number
+}
+
+const RESUMABLE_PAUSED_STATUSES = ['paused', 'partially_resumed', 'cancelling']
+
+async function filterLargeValueKeysWithoutRetainedReferences(
+ keys: string[],
+ deletedLogIds: string[]
+): Promise<string[]> {
+ if (keys.length === 0 || deletedLogIds.length === 0) return []
+
+ const unreferencedKeys: string[] = []
+ for (const key of Array.from(new Set(keys))) {
+ const [referencingLog] = await db
+ .select({ id: workflowExecutionLogs.id })
+ .from(workflowExecutionLogs)
+ .where(
+ and(
+ notInArray(workflowExecutionLogs.id, deletedLogIds),
+ sql`position(${key} in ${workflowExecutionLogs.executionData}::text) > 0`
+ )
+ )
+ .limit(1)
+
+ if (!referencingLog) {
+ unreferencedKeys.push(key)
+ }
+ }
+
+ return unreferencedKeys
 }
 
 async function deleteExecutionFiles(files: unknown, stats: FileDeleteStats): Promise<void> {
@@ -41,36 +74,103 @@ 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)
+ const deletedLogIds = rows.map((row) => row.id)
+ const largeValueKeys = rows.flatMap((row) => collectLargeValueKeys(row.executionData))
+ const unreferencedLargeValueKeys = await filterLargeValueKeysWithoutRetainedReferences(
+ largeValueKeys,
+ deletedLogIds
+ )
+
+ for (const row of rows) {
+ await deleteExecutionFiles(row.files, fileStats)
+ }
+ await deleteLargeValueStorageKeys(unreferencedLargeValueKeys, 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 +182,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 +198,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 +211,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`)