cleanup code

Author: icecrasher321

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

@@ -164,6 +164,7 @@ vi.mock('@/lib/cleanup/batch-delete', () => ({
 }))
 
 vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({
+  LIVE_PAUSED_REFERENCE_STATUSES: ['paused', 'partially_resumed', 'cancelling'],
   markLargeValuesDeleted: mockMarkLargeValuesDeleted,
   pruneLargeValueMetadata: mockPruneLargeValueMetadata,
   unreferencedLargeValuePredicate: vi.fn(() => ({ op: 'unreferencedLargeValuePredicate' })),

apps/sim/background/cleanup-logs.ts

@@ -19,6 +19,7 @@ import {
   type TableCleanupResult,
 } from '@/lib/cleanup/batch-delete'
 import {
+  LIVE_PAUSED_REFERENCE_STATUSES,
   markLargeValuesDeleted,
   pruneLargeValueMetadata,
   unreferencedLargeValuePredicate,
@@ -35,8 +36,6 @@ interface FileDeleteStats {
   filesDeleteFailed: number
 }
 
-const RESUMABLE_PAUSED_STATUSES = ['paused', 'partially_resumed', 'cancelling']
-
 const WORKFLOW_LOG_CLEANUP_BATCH_SIZE = 500
 const WORKFLOW_LOG_CLEANUP_MAX_BATCHES = 50
 const WORKFLOW_LOG_CLEANUP_ROW_LIMIT =
@@ -254,7 +253,7 @@ async function cleanupLegacyLargeExecutionValues(
                       SELECT 1
                       FROM ${pausedExecutions} AS ref_pe
                       WHERE ref_pe.execution_id = ref.execution_id
-                        AND ref_pe.status = ANY(${RESUMABLE_PAUSED_STATUSES}::text[])
+                        AND ref_pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
                     )
                   )
                 )
@@ -277,7 +276,7 @@ async function cleanupLegacyLargeExecutionValues(
                     SELECT 1
                     FROM ${pausedExecutions} AS parent_owner_pe
                     WHERE parent_owner_pe.execution_id = parent_value.owner_execution_id
-                      AND parent_owner_pe.status = ANY(${RESUMABLE_PAUSED_STATUSES}::text[])
+                      AND parent_owner_pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
                   )
                   OR EXISTS (
                     SELECT 1
@@ -298,7 +297,7 @@ async function cleanupLegacyLargeExecutionValues(
                             SELECT 1
                             FROM ${pausedExecutions} AS parent_ref_pe
                             WHERE parent_ref_pe.execution_id = parent_ref.execution_id
-                              AND parent_ref_pe.status = ANY(${RESUMABLE_PAUSED_STATUSES}::text[])
+                              AND parent_ref_pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
                           )
                         )
                       )
@@ -314,7 +313,7 @@ async function cleanupLegacyLargeExecutionValues(
               SELECT 1
               FROM ${pausedExecutions} AS pe
               WHERE pe.execution_id = split_part(${workspaceFiles.key}, '/', 4)
-                AND pe.status = ANY(${RESUMABLE_PAUSED_STATUSES}::text[])
+                AND pe.status = ANY(${LIVE_PAUSED_REFERENCE_STATUSES}::text[])
             )`
           )
         )
@@ -395,7 +394,7 @@ async function cleanupWorkflowExecutionLogs(
             lt(workflowExecutionLogs.startedAt, retentionDate),
             or(
               isNull(pausedExecutions.status),
-              notInArray(pausedExecutions.status, RESUMABLE_PAUSED_STATUSES)
+              notInArray(pausedExecutions.status, [...LIVE_PAUSED_REFERENCE_STATUSES])
             )
           )
         )

apps/sim/lib/execution/payloads/large-value-metadata.ts

@@ -19,7 +19,7 @@ const LARGE_VALUE_METADATA_WRITE_CHUNK_SIZE = 500
 const LARGE_VALUE_METADATA_WORKSPACE_CHUNK_SIZE = 50
 const LARGE_VALUE_METADATA_PRUNE_BATCH_SIZE = 1_000
 const LARGE_VALUE_METADATA_PRUNE_MAX_ROWS_PER_TABLE = 5_000
-const LIVE_PAUSED_REFERENCE_STATUSES = ['paused', 'partially_resumed', 'cancelling'] as const
+export const LIVE_PAUSED_REFERENCE_STATUSES = ['paused', 'partially_resumed', 'cancelling'] as const
 
 export interface LargeValueOwner {
   key: string

apps/sim/lib/execution/payloads/store.test.ts

@@ -155,6 +155,27 @@ describe('large execution payload store', () => {
     expect(mockDeleteFileMetadata).not.toHaveBeenCalled()
   })
 
+  it('falls back to memory-only refs for non-durable writes when orphan cleanup fails', async () => {
+    const value = { payload: 'x'.repeat(2048) }
+    const json = JSON.stringify(value)
+    mockRegisterLargeValueOwner.mockResolvedValueOnce(false)
+    mockDeleteFiles.mockImplementationOnce(async ([key]) => ({
+      deleted: 0,
+      failed: [{ key }],
+    }))
+
+    const ref = await storeLargeValue(value, json, Buffer.byteLength(json, 'utf8'), {
+      workspaceId: 'workspace-1',
+      workflowId: 'workflow-1',
+      executionId: 'execution-1',
+      userId: 'user-1',
+    })
+
+    expect(ref.key).toBeUndefined()
+    expect(materializeLargeValueRefSync(ref, { executionId: 'execution-1' })).toEqual(value)
+    expect(mockDeleteFileMetadata).not.toHaveBeenCalled()
+  })
+
   it('passes nested large value refs to owner metadata registration', async () => {
     const nestedKey =
       'execution/workspace-1/workflow-1/source-execution/large-value-lv_abcdefghijkl.json'

apps/sim/lib/execution/payloads/store.ts

@@ -175,8 +175,8 @@ export async function storeLargeValue(
   if (key) {
     const registered = await registerPersistedValueOwner(key, size, referencedKeys, context)
     if (!registered) {
-      const cleaned = await deleteUntrackedPersistedValue(key)
-      if (!cleaned || context.requireDurable) {
+      await deleteUntrackedPersistedValue(key)
+      if (context.requireDurable) {
         throw new Error('Failed to persist large execution value metadata')
       }
       key = undefined