fix(cleanup): persist storage retries and complete test fixtures

Author: TheodoreSpeaks

apps/sim/app/api/webhooks/outbox/process/route.ts

@@ -9,6 +9,7 @@ import { enterpriseOwnerClaimOutboxHandlers } from '@/lib/billing/enterprise-own
 import { enterpriseIssuanceOutboxHandlers } from '@/lib/billing/enterprise-provisioning'
 import { membershipBillingOutboxHandlers } from '@/lib/billing/organizations/membership-reconciliation'
 import { billingOutboxHandlers } from '@/lib/billing/webhooks/outbox-handlers'
+import { retentionStorageOutboxHandlers } from '@/lib/cleanup/storage-outbox'
 import { processOutboxEvents } from '@/lib/core/outbox/service'
 import { DeadlineExceededError } from '@/lib/core/utils/deadline'
 import { generateRequestId } from '@/lib/core/utils/request'
@@ -33,6 +34,7 @@ export const dynamic = 'force-dynamic'
 export const maxDuration = 800
 
 const handlers = {
+  ...retentionStorageOutboxHandlers,
   ...slackSearchOutboxHandlers,
   ...adminInvitationOperationOutboxHandlers,
   ...adminMemberOperationOutboxHandlers,

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

@@ -7,6 +7,7 @@ import {
   schemaMock,
 } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
+import type { OutboxHandlerRegistry } from '@/lib/core/outbox/service'
 
 const { storage, prepareChat, executeChat, hardDelete, billing, decrement, reRoot } = vi.hoisted(
   () => ({
@@ -19,6 +20,35 @@ const { storage, prepareChat, executeChat, hardDelete, billing, decrement, reRoo
     reRoot: vi.fn(),
   })
 )
+const outbox = vi.hoisted(() => new Map<string, { eventType: string; payload: unknown }>())
+vi.mock('@/lib/core/outbox/service', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('@/lib/core/outbox/service')>()
+  return {
+    ...actual,
+    enqueueOutboxEvent: vi.fn(async (...args: Parameters<typeof actual.enqueueOutboxEvent>) => {
+      const id = await actual.enqueueOutboxEvent(...args)
+      outbox.set(id, { eventType: args[1], payload: args[2] })
+      return id
+    }),
+    processOutboxEventById: vi.fn(async (id: string, handlers: OutboxHandlerRegistry) => {
+      const event = outbox.get(id)
+      if (!event) throw new Error('Missing test outbox event')
+      try {
+        await handlers[event.eventType](event.payload, {
+          eventId: id,
+          eventType: event.eventType,
+          attempts: 0,
+          maxAttempts: 48,
+          signal: new AbortController().signal,
+          checkpointPayload: async () => {},
+        })
+        return 'completed'
+      } catch {
+        return 'pending'
+      }
+    }),
+  }
+})
 vi.mock('@/background/cleanup-logs', () => ({ legacyLargeValuePredicate: vi.fn() }))
 vi.mock('@/background/cleanup-soft-deletes', () => ({
   reRootActiveFolderChildrenUnguarded: reRoot,
@@ -57,6 +87,7 @@ function control(type: CleanupType, dryRun = true, limit = 1) {
 beforeEach(() => {
   vi.clearAllMocks()
   resetDbChainMock()
+  outbox.clear()
   storage.mockResolvedValue({ deleted: 1, failed: [] })
   prepareChat.mockResolvedValue({ execute: executeChat })
 })
@@ -125,7 +156,7 @@ describe('requested cleanup stages', () => {
     dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'one', files: [{ key: 'blob' }] }])
     storage.mockResolvedValue({ deleted: 0, failed: [{ key: 'blob', error: 'unavailable' }] })
     const run = control('workflowLogs', false)
-    await expect(runBoundedLogScope(scope, run)).rejects.toThrow('storage deletions failed')
+    await expect(runBoundedLogScope(scope, run)).rejects.toThrow('storage cleanup is incomplete')
     expect(dbChainMockFns.delete).toHaveBeenCalledOnce()
     expect(run.progress.stages.workflowLogs).toMatchObject({
       selected: 1,
@@ -215,7 +246,7 @@ describe('bounded file billing', () => {
       { id: 'file-one', key: 'blob', context: 'workspace', workspaceId: 'ws-one', sizeBytes: 100 },
     ])
     billing.mockResolvedValue({ workspaceId: 'ws-one' })
-    dbChainMockFns.returning.mockResolvedValue([{ id: 'file-one', sizeBytes: 40 }])
+    dbChainMockFns.returning.mockResolvedValue([{ id: 'file-one', key: 'blob', sizeBytes: 40 }])
     const run = control('files', false)
     await runBoundedSoftDeleteScope({ ...scope, workspaceIds: ['ws-one', 'ws-two'] }, run)
     expect(decrement).toHaveBeenCalledWith(

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

@@ -12,7 +12,10 @@ import type { CleanupJobPayload } from '@/lib/billing/cleanup-dispatcher'
 import type { BoundedCleanup } from '@/lib/cleanup/bounded'
 import { boundedDelete } from '@/lib/cleanup/bounded-delete'
 import { pruneBoundedLargeValueMetadata } from '@/lib/cleanup/bounded-large-value-metadata'
-import { deleteBoundedStorage, tombstoneBoundedFiles } from '@/lib/cleanup/bounded-storage'
+import {
+  enqueueRetentionStorageCleanup,
+  processRetentionStorageCleanup,
+} from '@/lib/cleanup/storage-outbox'
 import {
   LIVE_PAUSED_REFERENCE_STATUSES,
   unreferencedLargeValuePredicate,
@@ -51,8 +54,8 @@ export async function runBoundedLogScope(payload: CleanupJobPayload, control: Bo
         ),
       (row) => row.id,
       async (rows) => {
-        const deleted = await control.query(async (tx) =>
-          tx
+        const { deleted, events } = await control.query(async (tx) => {
+          const deleted = await tx
             .delete(workflowExecutionLogs)
             .where(
               and(
@@ -64,19 +67,28 @@ export async function runBoundedLogScope(payload: CleanupJobPayload, control: Bo
               )
             )
             .returning({ id: workflowExecutionLogs.id, files: workflowExecutionLogs.files })
-        )
-        await control.deleted('workflowLogs', deleted.length)
-        for (const row of deleted) {
-          const keys = Array.isArray(row.files)
-            ? row.files.flatMap((file) =>
-                file && typeof file === 'object' && 'key' in file && typeof file.key === 'string'
-                  ? [file.key]
-                  : []
+          const keys = deleted.flatMap((row) =>
+            Array.isArray(row.files)
+              ? row.files.flatMap((file) =>
+                  file && typeof file === 'object' && 'key' in file && typeof file.key === 'string'
+                    ? [file.key]
+                    : []
+                )
+              : []
+          )
+          const events = isUsingCloudStorage()
+            ? await enqueueRetentionStorageCleanup(
+                tx,
+                keys,
+                'execution',
+                control.options.batchSize,
+                true
               )
             : []
-          await deleteBoundedStorage(control, 'workflowLogs', keys, 'execution')
-          if (isUsingCloudStorage()) await tombstoneBoundedFiles(control, keys)
-        }
+          return { deleted, events }
+        })
+        await control.deleted('workflowLogs', deleted.length)
+        await processRetentionStorageCleanup(control, 'workflowLogs', events)
       }
     )
     await boundedDelete(
@@ -123,7 +135,7 @@ export async function runBoundedLogScope(payload: CleanupJobPayload, control: Bo
         async (rows) => {
           if (!isUsingCloudStorage()) return
           const selectedKeys = rows.map((row) => row.key)
-          const keys = await control.query(async (tx) => {
+          const { keys, events } = await control.query(async (tx) => {
             await tx
               .select({ key: table.key })
               .from(table)
@@ -135,11 +147,18 @@ export async function runBoundedLogScope(payload: CleanupJobPayload, control: Bo
               .set({ deletedAt: new Date() })
               .where(and(eligible, inArray(table.key, selectedKeys)))
               .returning({ key: table.key })
-            return claimed.map((row) => row.key)
+            const keys = claimed.map((row) => row.key)
+            const events = await enqueueRetentionStorageCleanup(
+              tx,
+              keys,
+              'execution',
+              control.options.batchSize,
+              true
+            )
+            return { keys, events }
           })
-          await deleteBoundedStorage(control, type, keys, 'execution')
+          await processRetentionStorageCleanup(control, type, events)
           await control.deleted(type, keys.length)
-          await tombstoneBoundedFiles(control, keys)
         }
       )
     }

apps/sim/background/cleanup-soft-deletes-bounded.ts

@@ -21,17 +21,20 @@ import {
 } from '@/lib/billing/storage'
 import { type BoundedCleanup, setCleanupTimeouts } from '@/lib/cleanup/bounded'
 import { boundedDelete } from '@/lib/cleanup/bounded-delete'
-import { deleteBoundedStorage } from '@/lib/cleanup/bounded-storage'
 import type { CleanupType } from '@/lib/cleanup/bounded-types'
 import { prepareChatCleanup } from '@/lib/cleanup/chat-cleanup'
 import {
   type CleanupOwnerScope,
   cleanupOwnerCondition,
   resolveCleanupOwnerScope,
 } from '@/lib/cleanup/resource-scope'
+import {
+  enqueueRetentionStorageCleanup,
+  processRetentionStorageCleanup,
+} from '@/lib/cleanup/storage-outbox'
 import { hardDeleteDocuments } from '@/lib/knowledge/documents/service'
 import { cleanupKnowledgeStorageBinding } from '@/lib/knowledge/documents/storage-cleanup'
-import type { StorageContext } from '@/lib/uploads'
+import { isUsingCloudStorage, type StorageContext } from '@/lib/uploads'
 import { getWorkspaceFileSize } from '@/lib/uploads/shared/types'
 import { reRootActiveFolderChildrenUnguarded } from '@/background/cleanup-soft-deletes'
 
@@ -207,8 +210,8 @@ async function cleanupFiles(control: BoundedCleanup, scope: CleanupOwnerScope, c
         ),
       (row) => row.id,
       async (rows) => {
-        const deleted = await control.query(async (tx) =>
-          tx
+        const { deleted, events } = await control.query(async (tx) => {
+          const deleted = await tx
             .delete(workspaceFile)
             .where(
               and(
@@ -220,14 +223,18 @@ async function cleanupFiles(control: BoundedCleanup, scope: CleanupOwnerScope, c
               )
             )
             .returning({ id: workspaceFile.id, key: workspaceFile.key })
-        )
+          const events = isUsingCloudStorage()
+            ? await enqueueRetentionStorageCleanup(
+                tx,
+                deleted.map((row) => row.key),
+                'workspace',
+                control.options.batchSize
+              )
+            : []
+          return { deleted, events }
+        })
         await control.deleted('legacyFiles', deleted.length)
-        await deleteBoundedStorage(
-          control,
-          'legacyFiles',
-          deleted.map((row) => row.key),
-          'workspace'
-        )
+        await processRetentionStorageCleanup(control, 'legacyFiles', events)
       }
     )
   }
@@ -289,21 +296,24 @@ async function cleanupFiles(control: BoundedCleanup, scope: CleanupOwnerScope, c
               billing,
               deleted.reduce((sum, file) => sum + getWorkspaceFileSize(file), 0)
             )
-          return deleted
+          const events = isUsingCloudStorage()
+            ? await enqueueRetentionStorageCleanup(
+                tx,
+                deleted.map((file) => file.key),
+                row.context as StorageContext,
+                control.options.batchSize
+              )
+            : []
+          return { deleted, events }
         }
-        const deleted = billing
+        const { deleted, events } = billing
           ? await db.transaction(async (tx) => {
               await setCleanupTimeouts(tx)
               return remove(tx)
             })
           : await control.query(remove)
         await control.deleted('files', deleted.length)
-        await deleteBoundedStorage(
-          control,
-          'files',
-          deleted.map((file) => file.key),
-          row.context as StorageContext
-        )
+        await processRetentionStorageCleanup(control, 'files', events)
       }
     }
   )

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

@@ -13,6 +13,12 @@ const { mockUploadFile } = vi.hoisted(() => ({
   mockUploadFile: vi.fn(),
 }))
 
+vi.mock('@/lib/execution/payloads/large-value-metadata', async (importOriginal) => ({
+  ...(await importOriginal<typeof import('@/lib/execution/payloads/large-value-metadata')>()),
+  registerLargeValueOwner: vi.fn().mockResolvedValue(true),
+  addLargeValueReference: vi.fn().mockResolvedValue(undefined),
+}))
+
 vi.mock('@/lib/uploads', () => ({
   StorageService: {
     uploadFile: mockUploadFile,

apps/sim/executor/orchestrators/loop.test.ts

@@ -22,6 +22,12 @@ const mockLogger =
     vi.mocked(createLogger).mock.calls.findIndex(([name]) => name === 'LoopOrchestrator')
   ].value
 
+vi.mock('@/lib/execution/payloads/large-value-metadata', async (importOriginal) => ({
+  ...(await importOriginal<typeof import('@/lib/execution/payloads/large-value-metadata')>()),
+  registerLargeValueOwner: vi.fn().mockResolvedValue(true),
+  addLargeValueReference: vi.fn().mockResolvedValue(undefined),
+}))
+
 vi.mock('@/lib/execution/isolated-vm', () => ({
   executeInIsolatedVM: mockExecuteInIsolatedVM,
 }))

apps/sim/executor/variables/resolvers/block.test.ts

@@ -7,6 +7,12 @@ import { navigatePathAsync } from '@/executor/variables/resolvers/reference-asyn
 import { BlockResolver } from './block'
 import { RESOLVED_EMPTY, type ResolutionContext } from './reference'
 
+vi.mock('@/lib/execution/payloads/large-value-metadata', async (importOriginal) => ({
+  ...(await importOriginal<typeof import('@/lib/execution/payloads/large-value-metadata')>()),
+  registerLargeValueOwner: vi.fn().mockResolvedValue(true),
+  addLargeValueReference: vi.fn().mockResolvedValue(undefined),
+}))
+
 vi.mock('@/lib/uploads/server/metadata', () => ({
   insertImmutableFileMetadata: vi.fn().mockResolvedValue({ id: 'execution-payload-file' }),
   insertFileMetadata: vi.fn().mockResolvedValue({ id: 'execution-payload-file' }),

apps/sim/lib/cleanup/bounded-cleanup.md

@@ -53,11 +53,11 @@ Limits count **selected roots**, including roots restored before deletion. They
 - Bounded dispatch sets one attempt and a 180-second hard maximum. The worker stops starting new root batches after 120 seconds; cancellable child preparation also observes that work deadline.
 - Cleanup SQL uses transaction-local **500ms lock_timeout** and **5s statement_timeout**. The billable-file delete and storage decrement remain atomic. Orphan knowledge-base storage cleanup reuses the binding lock held by document creation, with a 15-second storage deadline. Large-value reference writers lock the value before registering references; cleanup locks and rechecks it before claiming a tombstone.
 - Trigger `cleanup` metadata reports each requested type's `selected`, `deleted`, `skipped`, `filesDeleted`, and `filesFailed`, plus stage, duration, and stop reason: `budgets_exhausted`, `scopes_exhausted`, `time_budget`, or `failed`.
-- A failure stops subsequent stages, preserves completed progress, and fails the run. External deletion is not transactional with Postgres. A hard process termination may leave the last metadata checkpoint behind actual effects. Do not blindly replay failed runs: inspect the failed stage and any external work already completed. Log and file storage is removed only for rows returned by the guarded delete. A storage failure after that commit can leave orphaned objects; a failed large-value storage deletion can leave a tombstone with remaining bytes. This change does not add recovery for those gaps or for chat backend/storage cleanup failures after parent deletion. Inspect failed runs before resubmitting.
+- A failure stops subsequent stages, preserves completed progress, and fails the run. External deletion is not transactional with Postgres. A hard process termination may leave the last metadata checkpoint behind actual effects. Do not blindly replay failed runs: inspect the failed stage and any external work already completed. Log and file storage is removed only for rows returned by the guarded delete. Their keys, and claimed large-value keys, enter `retention.storage.cleanup` outbox events in the same database transaction. The run attempts those events immediately; the existing outbox worker retries failures without selecting more roots. Inspect pending/dead-letter events when a run fails. Chat backend/storage cleanup still has its existing post-parent-delete recovery gap.
 
 ## Rollout
 
-1. Deploy Trigger workers first, then the API. Keep the existing production cleanup schedules disabled.
+1. Deploy Trigger workers first, then the API. Keep the existing production cleanup schedules disabled. Ensure the existing outbox processor is running so persisted storage failures can retry.
 2. Let old cleanup runs finish or cancel them before manual draining. Old queued jobs may retain their previous queue/version settings; the new queue cannot serialize against those runs.
 3. Run a preview, then one small delete invocation. Wait for the Trigger run to finish. Check database CPU, query latency, lock waits, replication lag, and application errors against their normal baseline.
 4. Repeat with a fresh request ID. Increase either the per-call budget or call frequency gradually, holding the other steady. Pause submissions on timeouts, failures, or database/application degradation. Avoid overlapping caller loops even though the Trigger queue serializes these jobs.