fix(cleanup): guard storage retries by file generation

Author: TheodoreSpeaks

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

@@ -21,13 +21,11 @@ const { storage, prepareChat, executeChat, hardDelete, billing, decrement, reRoo
   })
 )
 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')>()
+vi.mock('@/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] })
+    enqueueOutboxEvent: vi.fn(async (_tx: unknown, eventType: string, payload: unknown) => {
+      const id = `test-outbox-${outbox.size}`
+      outbox.set(id, { eventType, payload })
       return id
     }),
     processOutboxEventById: vi.fn(async (id: string, handlers: OutboxHandlerRegistry) => {
@@ -55,7 +53,7 @@ vi.mock('@/background/cleanup-soft-deletes', () => ({
 }))
 vi.mock('@/lib/uploads', () => ({
   isUsingCloudStorage: () => true,
-  StorageService: { deleteFiles: storage },
+  StorageService: { deleteFile: storage },
 }))
 vi.mock('@/lib/cleanup/chat-cleanup', () => ({ prepareChatCleanup: prepareChat }))
 vi.mock('@/lib/knowledge/documents/service', () => ({ hardDeleteDocuments: hardDelete }))
@@ -88,7 +86,7 @@ beforeEach(() => {
   vi.clearAllMocks()
   resetDbChainMock()
   outbox.clear()
-  storage.mockResolvedValue({ deleted: 1, failed: [] })
+  storage.mockResolvedValue(undefined)
   prepareChat.mockResolvedValue({ execute: executeChat })
 })
 
@@ -154,7 +152,7 @@ describe('requested cleanup stages', () => {
   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' }] })
+    storage.mockRejectedValue(new Error('unavailable'))
     const run = control('workflowLogs', false)
     await expect(runBoundedLogScope(scope, run)).rejects.toThrow('storage cleanup is incomplete')
     expect(dbChainMockFns.delete).toHaveBeenCalledOnce()

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

@@ -13,8 +13,7 @@ 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')>()),
+vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({
   registerLargeValueOwner: vi.fn().mockResolvedValue(true),
   addLargeValueReference: vi.fn().mockResolvedValue(undefined),
 }))

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

@@ -22,8 +22,7 @@ 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')>()),
+vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({
   registerLargeValueOwner: vi.fn().mockResolvedValue(true),
   addLargeValueReference: vi.fn().mockResolvedValue(undefined),
 }))

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

@@ -7,8 +7,7 @@ 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')>()),
+vi.mock('@/lib/execution/payloads/large-value-metadata', () => ({
   registerLargeValueOwner: vi.fn().mockResolvedValue(true),
   addLargeValueReference: vi.fn().mockResolvedValue(undefined),
 }))

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

@@ -53,7 +53,7 @@ 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. 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.
+- 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. Each event captures metadata IDs, content versions, and deletion state; retries skip changed or newly registered bindings and lock matching rows through a cancellable 15-second storage deletion. 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
 

apps/sim/lib/cleanup/bounded.postgres.integration.ts

@@ -13,8 +13,8 @@ import {
 import { processOutboxEventById } from '@/lib/core/outbox/service'
 import { lockLargeValueKeysForReference } from '@/lib/execution/payloads/large-value-lock'
 
-const { deleteStorageFiles } = vi.hoisted(() => ({ deleteStorageFiles: vi.fn() }))
-vi.mock('@/lib/uploads', () => ({ StorageService: { deleteFiles: deleteStorageFiles } }))
+const { deleteStorageFile } = vi.hoisted(() => ({ deleteStorageFile: vi.fn() }))
+vi.mock('@/lib/uploads', () => ({ StorageService: { deleteFile: deleteStorageFile } }))
 
 const url = new URL(process.env.DATABASE_URL ?? '')
 if (url.hostname !== '127.0.0.1' || url.pathname !== '/bounded_cleanup_test') {
@@ -43,7 +43,7 @@ beforeAll(async () => {
     sql`CREATE TABLE execution_large_values (key text PRIMARY KEY, workspace_id text, deleted_at timestamp)`
   )
   await db.execute(
-    sql`CREATE TABLE workspace_files (key text PRIMARY KEY, context text, deleted_at timestamp)`
+    sql`CREATE TABLE workspace_files (id text PRIMARY KEY DEFAULT gen_random_uuid()::text, key text NOT NULL, context text, deleted_at timestamp, content_updated_at timestamp NOT NULL DEFAULT now())`
   )
   await db.execute(sql`CREATE TABLE workflow_execution_logs (execution_id text)`)
   await db.execute(sql`CREATE TABLE paused_executions (execution_id text, status text)`)
@@ -56,7 +56,7 @@ beforeAll(async () => {
   )
 })
 beforeEach(async () => {
-  deleteStorageFiles.mockReset()
+  deleteStorageFile.mockReset()
   await db.execute(sql`TRUNCATE outbox_event`)
   await db.execute(
     sql`TRUNCATE execution_large_value_references, execution_large_value_dependencies, execution_large_values, workflow_execution_logs, paused_executions, workspace_files`
@@ -269,7 +269,10 @@ describe('large-value reference lifecycle locks', () => {
       const table = kind === 'metadata' ? 'execution_large_values' : 'workspace_files'
       if (kind === 'metadata')
         await db.execute(sql`INSERT INTO execution_large_values VALUES ('live-key','one',NULL)`)
-      else await db.execute(sql`INSERT INTO workspace_files VALUES ('live-key','execution',NULL)`)
+      else
+        await db.execute(
+          sql`INSERT INTO workspace_files (key,context,deleted_at) VALUES ('live-key','execution',NULL)`
+        )
       await db.transaction(async (tx) => {
         await lockLargeValueKeysForReference(tx, ['live-key'])
         await expect(
@@ -310,43 +313,104 @@ describe('durable retention storage cleanup', () => {
     ).rejects.toThrow('abort transaction')
     expect(await count('roots')).toBe(6)
     expect((await db.execute(sql`SELECT id FROM outbox_event`)).length).toBe(0)
-    expect(deleteStorageFiles).not.toHaveBeenCalled()
+    expect(deleteStorageFile).not.toHaveBeenCalled()
   })
   it('retries storage from the outbox after the root is gone', async () => {
     const [eventId] = await cleanupQuery(async (tx) => {
       await tx.delete(roots).where(eq(roots.id, 'a'))
       return enqueueRetentionStorageCleanup(tx, ['blob'], 'execution', 1)
     })
-    deleteStorageFiles.mockResolvedValueOnce({
-      deleted: 0,
-      failed: [{ key: 'blob', error: 'offline' }],
-    })
+    deleteStorageFile.mockRejectedValueOnce(new Error('offline'))
     await expect(
       processRetentionStorageCleanup(control(1), 'workflows', [eventId])
     ).rejects.toThrow('incomplete: pending')
     expect(await count('roots')).toBe(5)
-    const [pending] = await db.execute<{ status: string; payload: { keys: string[] } }>(
-      sql`SELECT status, payload FROM outbox_event WHERE id = ${eventId}`
-    )
+    const [pending] = await db.execute<{
+      status: string
+      payload: { files: Array<{ key: string }> }
+    }>(sql`SELECT status, payload FROM outbox_event WHERE id = ${eventId}`)
     expect(pending.status).toBe('pending')
-    expect(pending.payload.keys).toEqual(['blob'])
+    expect(pending.payload.files.map((file) => file.key)).toEqual(['blob'])
     await db.execute(
       sql`UPDATE outbox_event SET available_at = now() - interval '1 second' WHERE id = ${eventId}`
     )
-    deleteStorageFiles.mockResolvedValueOnce({ deleted: 1, failed: [] })
+    deleteStorageFile.mockResolvedValueOnce(undefined)
     await expect(processOutboxEventById(eventId, retentionStorageOutboxHandlers)).resolves.toBe(
       'completed'
     )
-    expect(deleteStorageFiles).toHaveBeenCalledTimes(2)
+    expect(deleteStorageFile).toHaveBeenCalledTimes(2)
     expect(await count('roots')).toBe(5)
   })
+  it.each(['restore', 'replace', 'new-binding'] as const)(
+    'does not delete a newer live binding on retry: %s',
+    async (change) => {
+      if (change !== 'new-binding') {
+        await db.execute(sql`INSERT INTO workspace_files(id,key,context,deleted_at)
+          VALUES ('original','blob','execution',now() - interval '1 day')`)
+      }
+      const [eventId] = await cleanupQuery((tx) =>
+        enqueueRetentionStorageCleanup(tx, ['blob'], 'execution', 1, true)
+      )
+      deleteStorageFile.mockRejectedValueOnce(new Error('offline'))
+      await expect(processOutboxEventById(eventId, retentionStorageOutboxHandlers)).resolves.toBe(
+        'pending'
+      )
+      if (change === 'restore') {
+        await db.execute(sql`UPDATE workspace_files SET deleted_at = NULL,
+          content_updated_at = content_updated_at + interval '1 second' WHERE id = 'original'`)
+      } else {
+        // A replacement may coexist with an older tombstone for the same key.
+        await db.execute(
+          sql`INSERT INTO workspace_files(id,key,context) VALUES ('replacement','blob','execution')`
+        )
+      }
+      await db.execute(
+        sql`UPDATE outbox_event SET available_at = now() - interval '1 second' WHERE id = ${eventId}`
+      )
+      await expect(processOutboxEventById(eventId, retentionStorageOutboxHandlers)).resolves.toBe(
+        'completed'
+      )
+      expect(deleteStorageFile).toHaveBeenCalledTimes(1)
+      const active = await db.execute(
+        sql`SELECT id FROM workspace_files WHERE key = 'blob' AND deleted_at IS NULL`
+      )
+      expect(active).toHaveLength(1)
+    }
+  )
+  it('holds the captured binding lock through storage deletion and tombstones only that identity', async () => {
+    await db.execute(
+      sql`INSERT INTO workspace_files(id,key,context) VALUES ('original','blob','execution')`
+    )
+    const [eventId] = await cleanupQuery((tx) =>
+      enqueueRetentionStorageCleanup(tx, ['blob'], 'execution', 1, true)
+    )
+    deleteStorageFile.mockImplementationOnce(async () => {
+      await expect(
+        runOutsideTransactionContext(() =>
+          db.transaction(async (tx) => {
+            await tx.execute(sql`SET LOCAL lock_timeout = '100ms'`)
+            await tx.execute(
+              sql`UPDATE workspace_files SET content_updated_at = now() WHERE id = 'original'`
+            )
+          })
+        )
+      ).rejects.toThrow()
+    })
+    await expect(processOutboxEventById(eventId, retentionStorageOutboxHandlers)).resolves.toBe(
+      'completed'
+    )
+    expect(deleteStorageFile).toHaveBeenCalledOnce()
+    expect(
+      await db.execute(sql`SELECT id FROM workspace_files WHERE deleted_at IS NULL`)
+    ).toHaveLength(0)
+  })
   it('bounds and deduplicates the persisted key batches', async () => {
     await cleanupQuery((tx) =>
       enqueueRetentionStorageCleanup(tx, ['a', 'b', 'a', 'c'], 'workspace', 2)
     )
-    const events = await db.execute<{ payload: { keys: string[] } }>(
+    const events = await db.execute<{ payload: { files: Array<{ key: string }> } }>(
       sql`SELECT payload FROM outbox_event ORDER BY created_at, id`
     )
-    expect(events.map((event) => event.payload.keys.length).sort()).toEqual([1, 2])
+    expect(events.map((event) => event.payload.files.length).sort()).toEqual([1, 2])
   })
 })

apps/sim/lib/cleanup/bounded.ts

@@ -18,7 +18,7 @@ export async function setCleanupTimeouts(tx: Pick<CleanupTransaction, 'execute'>
   await tx.execute(sql`SET LOCAL statement_timeout = '5s'`)
 }
 
-/** One short DB transaction; callers do storage/backend work outside this callback. */
+/** One bounded DB transaction; storage under a binding lock must use a cancellable deadline. */
 export const cleanupQuery: CleanupQuery = (query) =>
   dbFor('cleanup').transaction(async (tx) => {
     await setCleanupTimeouts(tx)