fix(files): release file history through the outbox after the current object is gone

Author: waleedlatif1

apps/sim/background/cleanup-file-versions.ts

@@ -8,7 +8,7 @@ import type { CleanupJobPayload } from '@/lib/billing/cleanup-dispatcher'
 import type { PlanCategory } from '@/lib/billing/plan-helpers'
 import { DEFAULT_DELETE_CHUNK_SIZE } from '@/lib/cleanup/batch-delete'
 import { retentionCleanupQueue } from '@/lib/cleanup/queue'
-import { StorageService } from '@/lib/uploads'
+import { enqueueWorkspaceFileStorageCleanups } from '@/lib/uploads/contexts/workspace/workspace-file-storage-cleanup-outbox'
 import {
   FILE_VERSION_RETENTION_KEEP_LATEST,
   MAX_SUPERSEDED_FILE_VERSIONS,
@@ -85,7 +85,6 @@ function selectExpiredVersions(fileIds: string[], cutoff: Date, maxSuperseded: n
   const ranked = cleanupDb
     .select({
       id: workspaceFileVersion.id,
-      key: workspaceFileVersion.key,
       supersededAt: workspaceFileVersion.supersededAt,
       rank: sql<number>`row_number() over (partition by ${workspaceFileVersion.fileId} order by ${workspaceFileVersion.version} desc)`.as(
         'rank'
@@ -101,7 +100,7 @@ function selectExpiredVersions(fileIds: string[], cutoff: Date, maxSuperseded: n
     .as('ranked')
 
   return cleanupDb
-    .select({ id: ranked.id, key: ranked.key })
+    .select({ id: ranked.id })
     .from(ranked)
     .where(
       and(
@@ -113,39 +112,34 @@ function selectExpiredVersions(fileIds: string[], cutoff: Date, maxSuperseded: n
 }
 
 /**
- * Deletes the stored objects first and only then the rows whose objects are gone, so a failed
- * delete leaves its row — and the next run retries it — instead of orphaning the object.
+ * Deletes the expired version rows and enqueues their stored objects on the storage-cleanup outbox
+ * in the same transaction, so a row never outlives its release and every released object is
+ * deleted durably — the outbox retries failures and treats an already-missing object as done.
  */
-async function deleteVersions(rows: Array<{ id: string; key: string }>, label: string) {
-  const failedKeys = new Set<string>()
-  for (const batch of chunkArray(rows, DEFAULT_DELETE_CHUNK_SIZE)) {
-    const deletion = await StorageService.deleteFiles(
-      batch.map((row) => row.key),
-      'workspace'
-    )
-    for (const { key, error } of deletion.failed) {
-      failedKeys.add(key)
-      logger.error(`[${label}] Failed to delete file version object ${key}`, { error })
-    }
-  }
-  const removable = rows.filter((row) => !failedKeys.has(row.key))
+async function deleteVersions(rows: Array<{ id: string }>) {
   let deleted = 0
-  for (const batch of chunkArray(removable, DEFAULT_DELETE_CHUNK_SIZE)) {
-    const removed = await cleanupDb
-      .delete(workspaceFileVersion)
-      .where(
-        and(
-          inArray(
-            workspaceFileVersion.id,
-            batch.map((row) => row.id)
-          ),
-          isNotNull(workspaceFileVersion.supersededAt)
+  for (const batch of chunkArray(rows, DEFAULT_DELETE_CHUNK_SIZE)) {
+    deleted += await cleanupDb.transaction(async (tx) => {
+      const removed = await tx
+        .delete(workspaceFileVersion)
+        .where(
+          and(
+            inArray(
+              workspaceFileVersion.id,
+              batch.map((row) => row.id)
+            ),
+            isNotNull(workspaceFileVersion.supersededAt)
+          )
         )
+        .returning({ key: workspaceFileVersion.key })
+      await enqueueWorkspaceFileStorageCleanups(
+        tx,
+        removed.map((row) => row.key)
       )
-      .returning({ id: workspaceFileVersion.id })
-    deleted += removed.length
+      return removed.length
+    })
   }
-  return { deleted, failed: rows.length - removable.length }
+  return deleted
 }
 
 export async function runCleanupFileVersions(payload: CleanupJobPayload): Promise<void> {
@@ -163,25 +157,21 @@ export async function runCleanupFileVersions(payload: CleanupJobPayload): Promis
   )
 
   let deleted = 0
-  let failed = 0
   for (const group of chunkArray(workspaceIds, WORKSPACES_PER_QUERY)) {
     const candidates = await selectCandidateFileIds(group, cutoff, maxSuperseded)
     for (const fileIds of chunkArray(candidates, FILES_PER_QUERY)) {
       for (let batch = 0; batch < MAX_BATCHES_PER_CHUNK; batch++) {
         const expired = await selectExpiredVersions(fileIds, cutoff, maxSuperseded)
         if (expired.length === 0) break
-        const result = await deleteVersions(expired, label)
-        deleted += result.deleted
-        failed += result.failed
-        if (expired.length < VERSIONS_PER_BATCH || result.deleted === 0) break
+        const removed = await deleteVersions(expired)
+        deleted += removed
+        if (expired.length < VERSIONS_PER_BATCH || removed === 0) break
       }
     }
   }
 
   const elapsed = ((Date.now() - startTime) / 1000).toFixed(2)
-  logger.info(
-    `[${label}] File version cleanup: ${deleted} deleted, ${failed} failed in ${elapsed}s`
-  )
+  logger.info(`[${label}] File version cleanup: ${deleted} released in ${elapsed}s`)
 }
 
 export const cleanupFileVersionsTask = task({

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

@@ -129,18 +129,25 @@ describe('cleanup soft deletes', () => {
     })
   })
 
-  it('releases the version history of expired workspace files before purging their objects', async () => {
+  it('releases version history only for files whose current object was deleted', async () => {
     mockSelectRowsByIdChunks
       .mockResolvedValueOnce([])
       .mockResolvedValueOnce([])
       .mockResolvedValueOnce([
         {
-          id: 'file-1',
-          key: 'workspace/ws-1/file-1',
+          id: 'file-purged',
+          key: 'workspace/ws-1/file-purged',
           workspaceId: 'ws-1',
           context: 'workspace',
           sizeBytes: 5,
         },
+        {
+          id: 'file-failed',
+          key: 'workspace/ws-1/file-failed',
+          workspaceId: 'ws-1',
+          context: 'workspace',
+          sizeBytes: 4,
+        },
         {
           id: 'chat-file',
           key: 'mothership/chat-file',
@@ -149,16 +156,20 @@ describe('cleanup soft deletes', () => {
           sizeBytes: 3,
         },
       ])
+    mockDeleteFiles.mockResolvedValueOnce({
+      deleted: 1,
+      failed: [{ key: 'workspace/ws-1/file-failed', error: 'storage unavailable' }],
+    })
 
     await runCleanupSoftDeletes(basePayload)
 
     expect(mockReleaseExpiredWorkspaceFileVersions).toHaveBeenCalledWith(
       expect.anything(),
-      ['file-1'],
+      ['file-purged'],
       expect.any(Date)
     )
-    expect(mockReleaseExpiredWorkspaceFileVersions.mock.invocationCallOrder[0]).toBeLessThan(
-      mockDeleteFiles.mock.invocationCallOrder[0]
+    expect(mockDeleteFiles.mock.invocationCallOrder[0]).toBeLessThan(
+      mockReleaseExpiredWorkspaceFileVersions.mock.invocationCallOrder[0]
     )
   })
 

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

@@ -900,12 +900,16 @@ export async function runCleanupSoftDeletes(
     chatCleanup = await prepareChatCleanup([...doomedChatIds], label)
   }
 
+  const fileCleanup = await cleanupWorkspaceFileStorage(fileScope)
+  /**
+   * Only files whose current object is gone are committed to the purge, so a file whose object
+   * deletion failed keeps its history for the retry instead of losing it ahead of its own delete.
+   */
   await releaseExpiredWorkspaceFileVersions(
     cleanupDb,
-    fileScope.multiContextRows.filter((row) => row.context === 'workspace').map((row) => row.id),
+    fileCleanup.multiContextRows.filter((row) => row.context === 'workspace').map((row) => row.id),
     retentionDate
   )
-  const fileCleanup = await cleanupWorkspaceFileStorage(fileScope)
   if (budgets && fileCleanup.filesFailed) throw new Error('File storage cleanup failed')
 
   let totalDeleted = 0

apps/sim/lib/uploads/contexts/workspace/__integration__/file-versions.integration.ts

@@ -14,7 +14,7 @@ import {
   workspaceFileVersion,
 } from '@sim/db/schema'
 import { generateId } from '@sim/utils/id'
-import { asc, eq, inArray, sql } from 'drizzle-orm'
+import { and, asc, eq, inArray, sql } from 'drizzle-orm'
 import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
 
 const fixtureStorage = vi.hoisted(() => ({ root: '' }))
@@ -434,9 +434,15 @@ describe('workspace file version history in PostgreSQL', () => {
     const events = await db
       .select({ id: outboxEvent.id, payload: outboxEvent.payload })
       .from(outboxEvent)
-      .where(eq(outboxEvent.eventType, WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT))
-    const enqueuedKeys = events.map((event) => (event.payload as { key: string }).key)
-    expect(enqueuedKeys).toEqual(expect.arrayContaining(releasedKeys))
+      .where(
+        and(
+          eq(outboxEvent.eventType, WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT),
+          inArray(sql<string>`${outboxEvent.payload}::jsonb ->> 'key'`, releasedKeys)
+        )
+      )
+    expect(events.map((event) => (event.payload as { key: string }).key).sort()).toEqual(
+      [...releasedKeys].sort()
+    )
     await db.delete(outboxEvent).where(
       inArray(
         outboxEvent.id,
@@ -464,6 +470,10 @@ describe('workspace file version history in PostgreSQL', () => {
         sql`${workspaceFileVersion.fileId} = ${fixture.fileId} AND ${workspaceFileVersion.supersededAt} IS NOT NULL`
       )
 
+    const prunedKeys = (await versionRows(fixture.fileId))
+      .filter((row) => row.version <= 3)
+      .map((row) => row.key)
+
     await runCleanupFileVersions({
       plan: 'pro',
       workspaceIds: [fixture.workspaceId],
@@ -474,5 +484,21 @@ describe('workspace file version history in PostgreSQL', () => {
     const rows = await versionRows(fixture.fileId)
     expect(rows.map((row) => row.version)).toEqual([4, 5, 6, 7, 8, 9, 10, 11, 12, 13])
     expect(rows.at(-1)?.supersededAt).toBeNull()
+    const events = await db
+      .select({ id: outboxEvent.id })
+      .from(outboxEvent)
+      .where(
+        and(
+          eq(outboxEvent.eventType, WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT),
+          inArray(sql<string>`${outboxEvent.payload}::jsonb ->> 'key'`, prunedKeys)
+        )
+      )
+    expect(events).toHaveLength(prunedKeys.length)
+    await db.delete(outboxEvent).where(
+      inArray(
+        outboxEvent.id,
+        events.map((event) => event.id)
+      )
+    )
   })
 })