fix(projects): sweep only retired Project prefixes (#9021)

Retirement no longer arms the bucket-wide Project storage inventory, which
deleted every project/<id>/ prefix without a row in this database. The
database is not the authority for a bucket another deployment shares or for
a bucket that outlives a point-in-time restore, so the inventory could
permanently delete live Project files.

The reconcile event stays registered with a no-op handler so rows already
armed drain to completed, and the processor no longer revives dead-lettered
inventory rows. Per-project immediate and post-TTL sweeps are unchanged.

The Azure prefix listing no longer includes uncommitted blobs: deleteIfExists
cannot remove them, so a page of them kept the sweep continuing until Azure
garbage-collected the blocks.

Author: waleedlatif1

apps/sim/app/api/v2/uploads/[uploadId]/route.ts

@@ -85,10 +85,10 @@ export const PUT = withRouteHandler(
       try {
         await queueRetiredProjectUploadCleanup(session)
       } catch (cleanupError) {
-        logger.error(
-          'Could not schedule retired upload cleanup; retirement reconciliation will retry',
-          { uploadId: session.id, error: cleanupError }
-        )
+        logger.error('Could not schedule retired upload cleanup', {
+          uploadId: session.id,
+          error: cleanupError,
+        })
       }
       if (error instanceof LocalUploadBodyError) {
         return v2Error('BAD_REQUEST', error.message)

apps/sim/lib/core/outbox/processor.test.ts

@@ -18,9 +18,6 @@ vi.mock('@/lib/knowledge/connectors/connector-error', () => ({
   getConnectorFailureDiagnostic: () => undefined,
 }))
 vi.mock('@/lib/core/outbox/handlers', () => ({ OUTBOX_HANDLER_GROUPS: [] }))
-vi.mock('@/lib/projects/files/prefix-cleanup', () => ({
-  recoverProjectStorageReconciliation: vi.fn(),
-}))
 
 import { runOutboxProcessor } from '@/lib/core/outbox/processor'
 

apps/sim/lib/core/outbox/processor.ts

@@ -31,19 +31,6 @@ export async function runOutboxProcessor(): Promise<OutboxProcessorResult> {
     minRemainingMs: 95_000,
   })
 
-  try {
-    if (Date.now() - startedAt < OUTBOX_PROCESSOR_RECOVERY_CUTOFF_MS) {
-      const { recoverProjectStorageReconciliation } = await import(
-        '@/lib/projects/files/prefix-cleanup'
-      )
-      await recoverProjectStorageReconciliation()
-    }
-  } catch (error) {
-    logger.error('Project storage reconciliation recovery failed', {
-      error: toError(error).message,
-    })
-  }
-
   let recoveredDocuments = 0
   try {
     if (Date.now() - startedAt < OUTBOX_PROCESSOR_RECOVERY_CUTOFF_MS) {

apps/sim/lib/projects/files/__integration__/private-purge.integration.ts

@@ -36,7 +36,6 @@ import { prepareProjectStorageMutationInTx } from '@/lib/billing/storage/trackin
 import { processOutboxEventById } from '@/lib/core/outbox/service'
 import { prepareProjectsForAccountDeletion } from '@/lib/projects/account-deletion'
 import { workspaceFileStorageCleanupOutboxHandlers } from '@/lib/uploads/contexts/workspace/workspace-file-storage-cleanup-outbox'
-import * as storageService from '@/lib/uploads/core/storage-service'
 import { UPLOAD_URL_TTL_MS } from '@/lib/uploads/upload-session/provider'
 import { PROJECT_FILE_UPLOAD_BINDING_KEY } from '@/lib/uploads/upload-session/types'
 import { deleteUserAccount } from '@/lib/users/account-deletion'
@@ -456,26 +455,6 @@ describe('Private Project teardown and durable object cleanup', () => {
       await expect(readdir(join(storageRoot, `project/${f.projectId}`))).rejects.toMatchObject({
         code: 'ENOENT',
       })
-      await mkdir(dirname(join(storageRoot, upload.finalKey)), { recursive: true })
-      await writeFile(join(storageRoot, upload.finalKey), 'finished-after-empty-sweep')
-      const reconciliationId = 'project-file.storage.reconcile'
-      await db
-        .update(outboxEvent)
-        .set({ availableAt: new Date(0), attempts: 4 })
-        .where(eq(outboxEvent.id, reconciliationId))
-      expect(
-        await processOutboxEventById(reconciliationId, projectFilePrefixCleanupOutboxHandlers)
-      ).toBe('pending')
-      await expect(readFile(join(storageRoot, upload.finalKey))).rejects.toMatchObject({
-        code: 'ENOENT',
-      })
-      expect(await readFile(join(storageRoot, live.keys[0]), 'utf8')).toBe('new')
-      const [reconciled] = await db
-        .select()
-        .from(outboxEvent)
-        .where(eq(outboxEvent.id, reconciliationId))
-      expect(reconciled.attempts).toBe(0)
-      expect(reconciled.availableAt.getTime()).toBeGreaterThan(Date.now())
       const unsafe = generateId()
       await db.insert(outboxEvent).values({
         id: unsafe,
@@ -543,7 +522,7 @@ describe('Private Project teardown and durable object cleanup', () => {
     }
   )
 
-  check('retirement batches finish while sharing one inventory reconciliation event', async () => {
+  check('retirement batches finish without arming a bucket inventory', async () => {
     const f = await fixture()
     await Promise.all(
       Array.from({ length: 501 }, (_, index) => seedUpload(f, `batch-${index}.bin`))
@@ -587,175 +566,29 @@ describe('Private Project teardown and durable object cleanup', () => {
       .select()
       .from(outboxEvent)
       .where(eq(outboxEvent.eventType, 'project-file.storage.reconcile'))
-    expect(reconciliations).toHaveLength(1)
+    expect(reconciliations).toHaveLength(0)
   })
 
   check(
-    'retirement revives terminal reconciliation without replacing active progress',
+    'a previously armed storage inventory drains without deleting unowned prefixes',
     async () => {
-      const initial = await fixture()
-      await erasePrivateProject(initial)
       const reconciliationId = 'project-file.storage.reconcile'
-      for (const status of ['completed', 'dead_letter', 'pending', 'processing']) {
-        const f = await fixture()
-        const lockedAt = new Date()
-        await db
-          .update(outboxEvent)
-          .set({ status, attempts: 4, payload: { cursor: 'saved-progress' }, lockedAt })
-          .where(eq(outboxEvent.id, reconciliationId))
-        await erasePrivateProject(f)
-        const [event] = await db
-          .select()
-          .from(outboxEvent)
-          .where(eq(outboxEvent.id, reconciliationId))
-        if (status === 'completed' || status === 'dead_letter') {
-          expect(event).toMatchObject({
-            status: 'pending',
-            attempts: 0,
-            payload: {},
-            lockedAt: null,
-          })
-        } else {
-          expect(event).toMatchObject({
-            status,
-            attempts: 4,
-            payload: { cursor: 'saved-progress' },
-            lockedAt,
-          })
-        }
-      }
-      await db
-        .update(outboxEvent)
-        .set({ status: 'pending', attempts: 0, payload: {}, lockedAt: null })
-        .where(eq(outboxEvent.id, reconciliationId))
-    }
-  )
-
-  check(
-    'inventory reconciliation isolates a failed prefix and retries it next cycle while preserving live owners',
-    async () => {
-      const retired = await fixture()
-      await erasePrivateProject(retired)
-      const live = await fixture()
-      const reconciliationId = 'project-file.storage.reconcile'
-      const orphanIds = Array.from({ length: 505 }, () => generateId()).sort()
-      const obstructed = join(storageRoot, `project/${orphanIds[0]}`)
-      for (const id of orphanIds) {
-        const prefix = join(storageRoot, `project/${id}`)
-        await mkdir(prefix, { recursive: true })
-        await writeFile(join(prefix, 'late.bin'), 'late')
-      }
-      const deleteStoredFile = storageService.deleteFile
-      const deletion = vi.spyOn(storageService, 'deleteFile').mockImplementation((params) => {
-        if (params.key.startsWith(`project/${orphanIds[0]}/`)) {
-          return Promise.reject(
-            Object.assign(new Error('Fixture prefix deletion failed'), { code: 'EACCES' })
-          )
-        }
-        return deleteStoredFile(params)
+      const unowned = `project/${generateId()}/compiled/late.bin`
+      await mkdir(dirname(join(storageRoot, unowned)), { recursive: true })
+      await writeFile(join(storageRoot, unowned), 'late')
+      await db.insert(outboxEvent).values({
+        id: reconciliationId,
+        eventType: reconciliationId,
+        payload: {},
+        availableAt: new Date(0),
       })
       const { projectFilePrefixCleanupOutboxHandlers } = await import(
         '@/lib/projects/files/prefix-cleanup'
       )
-      try {
-        await db
-          .update(outboxEvent)
-          .set({ availableAt: new Date(0), payload: {}, attempts: 3 })
-          .where(eq(outboxEvent.id, reconciliationId))
-        expect(
-          await processOutboxEventById(reconciliationId, projectFilePrefixCleanupOutboxHandlers)
-        ).toBe('pending')
-        const [failed] = await db
-          .select()
-          .from(outboxEvent)
-          .where(eq(outboxEvent.id, reconciliationId))
-        expect(failed.attempts).toBe(3)
-        expect(failed.payload).toMatchObject({ failedPrefixes: 1, cursor: expect.any(String) })
-        let completedCycle = false
-        for (let attempt = 0; attempt < 10; attempt++) {
-          await db
-            .update(outboxEvent)
-            .set({ availableAt: new Date(0) })
-            .where(eq(outboxEvent.id, reconciliationId))
-          expect(
-            await processOutboxEventById(reconciliationId, projectFilePrefixCleanupOutboxHandlers)
-          ).toBe('pending')
-          const [continued] = await db
-            .select()
-            .from(outboxEvent)
-            .where(eq(outboxEvent.id, reconciliationId))
-          if (continued.availableAt.getTime() > Date.now() + 60 * 60 * 1000) {
-            expect(continued.attempts).toBe(3)
-            completedCycle = true
-            break
-          }
-          expect(continued.payload).toMatchObject({ cursor: expect.any(String) })
-        }
-        expect(completedCycle).toBe(true)
-        for (const id of orphanIds.slice(1))
-          await expect(readdir(join(storageRoot, `project/${id}`))).rejects.toMatchObject({
-            code: 'ENOENT',
-          })
-        expect(await readFile(join(storageRoot, live.keys[0]), 'utf8')).toBe('new')
-      } finally {
-        deletion.mockRestore()
-      }
-      expect(await readFile(join(obstructed, 'late.bin'), 'utf8')).toBe('late')
-      await db
-        .update(outboxEvent)
-        .set({ availableAt: new Date(0) })
-        .where(eq(outboxEvent.id, reconciliationId))
       expect(
         await processOutboxEventById(reconciliationId, projectFilePrefixCleanupOutboxHandlers)
-      ).toBe('pending')
-      await expect(readdir(obstructed)).rejects.toMatchObject({ code: 'ENOENT' })
-      const [recovered] = await db
-        .select()
-        .from(outboxEvent)
-        .where(eq(outboxEvent.id, reconciliationId))
-      expect(recovered.attempts).toBe(0)
-    }
-  )
-
-  check(
-    'periodic reconciliation recovery respects cooldown and active leases while preserving progress',
-    async () => {
-      const f = await fixture()
-      await erasePrivateProject(f)
-      const reconciliationId = 'project-file.storage.reconcile'
-      const { recoverProjectStorageReconciliation } = await import(
-        '@/lib/projects/files/prefix-cleanup'
-      )
-      const now = new Date()
-      const checkpoint = { cursor: 'provider-page', page: [f.projectId], index: 0 }
-      const old = new Date(now.getTime() - 25 * 60 * 60 * 1000)
-      for (const status of ['pending', 'processing', 'dead_letter']) {
-        await db
-          .update(outboxEvent)
-          .set({ status, attempts: 10, payload: checkpoint, processedAt: old, lockedAt: now })
-          .where(eq(outboxEvent.id, reconciliationId))
-        await recoverProjectStorageReconciliation(now)
-        const [event] = await db
-          .select()
-          .from(outboxEvent)
-          .where(eq(outboxEvent.id, reconciliationId))
-        expect(event.payload).toEqual(checkpoint)
-        expect(event).toMatchObject(
-          status === 'dead_letter'
-            ? { status: 'pending', attempts: 0, lockedAt: null, processedAt: null }
-            : { status, attempts: 10, lockedAt: now, processedAt: old }
-        )
-      }
-      await db
-        .update(outboxEvent)
-        .set({ status: 'dead_letter', attempts: 10, processedAt: now })
-        .where(eq(outboxEvent.id, reconciliationId))
-      await recoverProjectStorageReconciliation(now)
-      const [cooling] = await db
-        .select()
-        .from(outboxEvent)
-        .where(eq(outboxEvent.id, reconciliationId))
-      expect(cooling).toMatchObject({ status: 'dead_letter', attempts: 10, payload: checkpoint })
+      ).toBe('completed')
+      expect(await readFile(join(storageRoot, unowned), 'utf8')).toBe('late')
     }
   )