fix(cleanup): coordinate deletion with resource eligibility

Author: TheodoreSpeaks

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

@@ -120,18 +120,56 @@ describe('requested cleanup stages', () => {
     expect(run.progress.stages.jobLogs).toMatchObject({ selected: 2, deleted: 2 })
     expect(dbChainMockFns.delete).toHaveBeenCalledTimes(2)
   })
-  it('stops before root deletion if attached log storage fails', async () => {
+  it('records committed log deletion if attached storage cleanup fails', async () => {
     queueTableRows(schemaMock.workflowExecutionLogs, [{ id: 'one', files: [{ key: 'blob' }] }])
+    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')
-    expect(dbChainMockFns.delete).not.toHaveBeenCalled()
+    expect(dbChainMockFns.delete).toHaveBeenCalledOnce()
     expect(run.progress.stages.workflowLogs).toMatchObject({
       selected: 1,
-      deleted: 0,
+      deleted: 1,
       filesFailed: 1,
     })
   })
+  it('does not remove files of a log protected after selection', async () => {
+    queueTableRows(schemaMock.workflowExecutionLogs, [{ id: 'one', files: [{ key: 'blob' }] }])
+    dbChainMockFns.returning.mockResolvedValueOnce([])
+    const run = control('workflowLogs', false)
+    await runBoundedLogScope(scope, run)
+    expect(storage).not.toHaveBeenCalled()
+    expect(dbChainMockFns.update).not.toHaveBeenCalled()
+    expect(run.progress.stages.workflowLogs).toMatchObject({ selected: 1, deleted: 0, skipped: 1 })
+  })
+  it.each(['largeValues', 'legacyLargeValues'] as const)(
+    'does not remove a %s key that fails the final liveness claim',
+    async (type) => {
+      queueTableRows(
+        type === 'largeValues' ? schemaMock.executionLargeValues : schemaMock.workspaceFiles,
+        [{ key: 'referenced-key' }]
+      )
+      dbChainMockFns.returning.mockResolvedValueOnce([])
+      const run = control(type, false)
+      await runBoundedLogScope(scope, run)
+      expect(storage).not.toHaveBeenCalled()
+      expect(run.progress.stages[type]).toMatchObject({ selected: 1, deleted: 0, skipped: 1 })
+    }
+  )
+  it.each(['files', 'legacyFiles'] as const)(
+    'does not remove a restored %s object',
+    async (type) => {
+      queueTableRows(type === 'files' ? schemaMock.workspaceFiles : schemaMock.workspaceFile, [
+        { id: 'one', key: 'blob', context: 'workspace', workspaceId: 'ws-one', sizeBytes: 100 },
+      ])
+      billing.mockResolvedValue({ workspaceId: 'ws-one' })
+      dbChainMockFns.returning.mockResolvedValueOnce([])
+      const run = control(type, false)
+      await runBoundedSoftDeleteScope(scope, run)
+      expect(storage).not.toHaveBeenCalled()
+      expect(run.progress.stages[type]).toMatchObject({ selected: 1, deleted: 0, skipped: 1 })
+    }
+  )
   it('keeps workflow chat side effects even with a zero chat budget', async () => {
     queueTableRows(schemaMock.workflow, [{ id: 'workflow-one' }])
     queueTableRows(schemaMock.copilotChats, [{ id: 'child-chat' }])

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

@@ -51,17 +51,6 @@ export async function runBoundedLogScope(payload: CleanupJobPayload, control: Bo
         ),
       (row) => row.id,
       async (rows) => {
-        for (const row of rows) {
-          const keys = Array.isArray(row.files)
-            ? row.files.flatMap((file) =>
-                file && typeof file === 'object' && 'key' in file && typeof file.key === 'string'
-                  ? [file.key]
-                  : []
-              )
-            : []
-          await deleteBoundedStorage(control, 'workflowLogs', keys, 'execution')
-          if (isUsingCloudStorage()) await tombstoneBoundedFiles(control, keys)
-        }
         const deleted = await control.query(async (tx) =>
           tx
             .delete(workflowExecutionLogs)
@@ -74,9 +63,20 @@ export async function runBoundedLogScope(payload: CleanupJobPayload, control: Bo
                 )
               )
             )
-            .returning({ id: workflowExecutionLogs.id })
+            .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]
+                  : []
+              )
+            : []
+          await deleteBoundedStorage(control, 'workflowLogs', keys, 'execution')
+          if (isUsingCloudStorage()) await tombstoneBoundedFiles(control, keys)
+        }
       }
     )
     await boundedDelete(
@@ -122,15 +122,22 @@ export async function runBoundedLogScope(payload: CleanupJobPayload, control: Bo
         (row) => row.key,
         async (rows) => {
           if (!isUsingCloudStorage()) return
-          const keys = rows.map((row) => row.key)
+          const selectedKeys = rows.map((row) => row.key)
+          const keys = await control.query(async (tx) => {
+            await tx
+              .select({ key: table.key })
+              .from(table)
+              .where(inArray(table.key, selectedKeys))
+              .orderBy(asc(table.key))
+              .for('update')
+            const claimed = await tx
+              .update(table)
+              .set({ deletedAt: new Date() })
+              .where(and(eligible, inArray(table.key, selectedKeys)))
+              .returning({ key: table.key })
+            return claimed.map((row) => row.key)
+          })
           await deleteBoundedStorage(control, type, keys, 'execution')
-          if (!legacy)
-            await control.query(async (tx) => {
-              await tx
-                .update(executionLargeValues)
-                .set({ deletedAt: new Date() })
-                .where(inArray(executionLargeValues.key, keys))
-            })
           await control.deleted(type, keys.length)
           await tombstoneBoundedFiles(control, keys)
         }

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

@@ -30,6 +30,7 @@ import {
   resolveCleanupOwnerScope,
 } from '@/lib/cleanup/resource-scope'
 import { hardDeleteDocuments } from '@/lib/knowledge/documents/service'
+import { cleanupKnowledgeStorageBinding } from '@/lib/knowledge/documents/storage-cleanup'
 import type { StorageContext } from '@/lib/uploads'
 import { getWorkspaceFileSize } from '@/lib/uploads/shared/types'
 import { reRootActiveFolderChildrenUnguarded } from '@/background/cleanup-soft-deletes'
@@ -121,18 +122,16 @@ export async function runBoundedSoftDeleteScope(
         lt(knowledgeBase.deletedAt, cutoff)
       ),
       {
-        before: async (kbIds) => {
+        beforeDelete: async (kbIds, tx) => {
           // Existing ledger/outbox implementation owns document and embedding deletion.
           while (true) {
             control.assertTimeRemaining()
-            const rows = await control.query(async (tx) =>
-              tx
-                .select({ id: document.id })
-                .from(document)
-                .where(inArray(document.knowledgeBaseId, kbIds))
-                .orderBy(asc(document.id))
-                .limit(control.options.batchSize)
-            )
+            const rows = await tx
+              .select({ id: document.id })
+              .from(document)
+              .where(inArray(document.knowledgeBaseId, kbIds))
+              .orderBy(asc(document.id))
+              .limit(control.options.batchSize)
             if (rows.length === 0) break
             const deleted = await hardDeleteDocuments(
               rows.map((row) => row.id),
@@ -141,7 +140,7 @@ export async function runBoundedSoftDeleteScope(
               undefined,
               undefined,
               undefined,
-              control.query
+              async (query) => query(tx)
             )
             if (deleted !== rows.length)
               throw new Error('Knowledge-base document cleanup did not delete its selected batch')
@@ -176,10 +175,8 @@ export async function runBoundedSoftDeleteScope(
           ),
           target.type === 'folders'
             ? {
-                before: (folderIds) =>
-                  control.query((tx) =>
-                    reRootActiveFolderChildrenUnguarded(folderIds, cutoff, payload.label, tx, true)
-                  ),
+                beforeDelete: (folderIds, tx) =>
+                  reRootActiveFolderChildrenUnguarded(folderIds, cutoff, payload.label, tx, true),
               }
             : {}
         )
@@ -210,12 +207,6 @@ async function cleanupFiles(control: BoundedCleanup, scope: CleanupOwnerScope, c
         ),
       (row) => row.id,
       async (rows) => {
-        await deleteBoundedStorage(
-          control,
-          'legacyFiles',
-          rows.map((row) => row.key),
-          'workspace'
-        )
         const deleted = await control.query(async (tx) =>
           tx
             .delete(workspaceFile)
@@ -228,9 +219,15 @@ async function cleanupFiles(control: BoundedCleanup, scope: CleanupOwnerScope, c
                 )
               )
             )
-            .returning({ id: workspaceFile.id })
+            .returning({ id: workspaceFile.id, key: workspaceFile.key })
         )
         await control.deleted('legacyFiles', deleted.length)
+        await deleteBoundedStorage(
+          control,
+          'legacyFiles',
+          deleted.map((row) => row.key),
+          'workspace'
+        )
       }
     )
   }
@@ -270,7 +267,6 @@ async function cleanupFiles(control: BoundedCleanup, scope: CleanupOwnerScope, c
                 return resolveStorageBillingContext(row.workspaceId, { executor })
               })
             : undefined
-        await deleteBoundedStorage(control, 'files', [row.key], row.context as StorageContext)
         const remove = async (tx: Parameters<Parameters<typeof db.transaction>[0]>[0]) => {
           const deleted = await tx
             .delete(workspaceFiles)
@@ -282,22 +278,32 @@ async function cleanupFiles(control: BoundedCleanup, scope: CleanupOwnerScope, c
                 billing ? eq(workspaceFiles.workspaceId, billing.workspaceId) : undefined
               )
             )
-            .returning({ id: workspaceFiles.id, sizeBytes: workspaceFiles.sizeBytes })
+            .returning({
+              id: workspaceFiles.id,
+              key: workspaceFiles.key,
+              sizeBytes: workspaceFiles.sizeBytes,
+            })
           if (billing)
             await decrementStorageUsageForBillingContextInTx(
               tx,
               billing,
               deleted.reduce((sum, file) => sum + getWorkspaceFileSize(file), 0)
             )
-          return deleted.length
+          return deleted
         }
         const deleted = billing
           ? await db.transaction(async (tx) => {
               await setCleanupTimeouts(tx)
               return remove(tx)
             })
           : await control.query(remove)
-        await control.deleted('files', deleted)
+        await control.deleted('files', deleted.length)
+        await deleteBoundedStorage(
+          control,
+          'files',
+          deleted.map((file) => file.key),
+          row.context as StorageContext
+        )
       }
     }
   )
@@ -317,36 +323,42 @@ async function cleanupOrphanBindings(control: BoundedCleanup, scope: CleanupOwne
     (limit, seen) =>
       control.query(async (tx) =>
         tx
-          .select({ id: workspaceFiles.id, key: workspaceFiles.key })
+          .select({
+            id: workspaceFiles.id,
+            key: workspaceFiles.key,
+            contentUpdatedAt: workspaceFiles.contentUpdatedAt,
+            workspaceId: workspaceFiles.workspaceId,
+            organizationId: workspaceFiles.organizationId,
+            userId: workspaceFiles.userId,
+          })
           .from(workspaceFiles)
           .where(and(eligible, seen.length ? notInArray(workspaceFiles.id, seen) : undefined))
           .orderBy(asc(workspaceFiles.id))
           .limit(limit)
       ),
     (row) => row.id,
     async (rows) => {
-      await deleteBoundedStorage(
-        control,
-        type,
-        rows.map((row) => row.key),
-        'knowledge-base'
-      )
-      const deleted = await control.query(async (tx) =>
-        tx
-          .update(workspaceFiles)
-          .set({ deletedAt: new Date() })
-          .where(
-            and(
-              eligible,
-              inArray(
-                workspaceFiles.id,
-                rows.map((row) => row.id)
-              )
-            )
-          )
-          .returning({ id: workspaceFiles.id })
-      )
-      await control.deleted(type, deleted.length)
+      for (const row of rows) {
+        control.assertTimeRemaining()
+        const deleted = await cleanupKnowledgeStorageBinding(
+          {
+            version: 1,
+            documentId: `orphan:${row.id}`,
+            fileId: row.id,
+            key: row.key,
+            contentUpdatedAt: row.contentUpdatedAt.toISOString(),
+            workspaceId: row.workspaceId,
+            organizationId: row.organizationId,
+            userId: row.userId,
+          },
+          AbortSignal.timeout(15_000),
+          control.query
+        )
+        if (deleted) {
+          await control.deleted(type, 1)
+          await control.files(type, 1, 0)
+        }
+      }
     }
   )
 }

apps/sim/executor/execution/block-executor.test.ts

@@ -31,6 +31,11 @@ const { mockUploadFile, mockDownloadFile, mockMaskBatch } = vi.hoisted(() => ({
   mockMaskBatch: vi.fn(),
 }))
 
+vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({
+  registerLargeValueOwner: vi.fn().mockResolvedValue(true),
+  addLargeValueReference: vi.fn().mockResolvedValue(undefined),
+}))
+
 vi.mock('@/ee/access-control/utils/permission-check', () => ({
   validateBlockType: vi.fn(),
 }))

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

@@ -9,6 +9,11 @@ import { navigatePathAsync } from '@/executor/variables/resolvers/reference-asyn
 import type { ResolutionContext } from './reference'
 import { WorkflowResolver } from './workflow'
 
+vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({
+  registerLargeValueOwner: vi.fn().mockResolvedValue(true),
+  addLargeValueReference: vi.fn().mockResolvedValue(undefined),
+}))
+
 vi.mock('@/lib/workflows/variables/variable-manager', () => ({
   VariableManager: {
     resolveForExecution: vi.fn((value) => value),

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

@@ -42,7 +42,7 @@ A call with **no query parameters retains the existing scheduled cleanup behavio
 
 ## What the budget means
 
-Limits count **selected roots**, including roots restored before deletion. They do not count all physical rows affected by foreign keys. Two workflows can cascade into thousands of blocks, edges, chats, and messages. Mandatory attached-file and backend cleanup follows selected parents even when its standalone type budget is zero. Knowledge-base deletion retains the existing document accounting and storage-cleanup outbox behavior.
+Limits count **selected roots**, including roots restored before deletion. They do not count all physical rows affected by foreign keys. Two workflows can cascade into thousands of blocks, edges, chats, and messages. Mandatory attached-file and backend cleanup follows selected parents even when its standalone type budget is zero. Knowledge-base deletion retains the existing document accounting and storage-cleanup outbox behavior. Its child changes and parent deletion share one locked transaction, as do folder re-rooting and deletion; a failure rolls all of them back.
 
 `largeValues`/`legacyLargeValues` count object keys. `orphanKnowledgeBaseBindings` counts bindings soft-deleted after object cleanup. Metadata pruning counts its selected metadata records. A dry run previews the current state; it does not reserve rows for a later deletion run.
 
@@ -51,9 +51,9 @@ Limits count **selected roots**, including roots restored before deletion. They
 - One coordinator walks owner scopes sequentially with shared budgets. It creates no child cleanup jobs.
 - Logs and soft deletes share the named Trigger queue `retention-cleanup`, concurrency 1, including newly dispatched scheduled jobs. No per-type concurrency keys are used.
 - 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.
+- 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. This change does not add recovery for the existing gap when chat backend/storage cleanup fails after parent rows have committed.
+- 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.
 
 ## Rollout