From 45b7339f385e47e01f5343ace521410f0cb87281 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 14 Oct 2026 08:17:19 -0700 Subject: [PATCH] fix(projects): sweep only retired Project prefixes Retirement no longer arms the bucket-wide Project storage inventory, which deleted every project// 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. --- .../app/api/v2/uploads/[uploadId]/route.ts | 8 +- apps/sim/lib/core/outbox/processor.test.ts | 3 - apps/sim/lib/core/outbox/processor.ts | 13 - .../private-purge.integration.ts | 193 +------------ apps/sim/lib/projects/files/prefix-cleanup.ts | 265 +----------------- 5 files changed, 28 insertions(+), 454 deletions(-) diff --git a/apps/sim/app/api/v2/uploads/[uploadId]/route.ts b/apps/sim/app/api/v2/uploads/[uploadId]/route.ts index 42ddd56e355..3dceddbdfd2 100644 --- a/apps/sim/app/api/v2/uploads/[uploadId]/route.ts +++ b/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) diff --git a/apps/sim/lib/core/outbox/processor.test.ts b/apps/sim/lib/core/outbox/processor.test.ts index 5c41d849cae..4676a76503e 100644 --- a/apps/sim/lib/core/outbox/processor.test.ts +++ b/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' diff --git a/apps/sim/lib/core/outbox/processor.ts b/apps/sim/lib/core/outbox/processor.ts index 6bfab7c43bb..995e70b57d2 100644 --- a/apps/sim/lib/core/outbox/processor.ts +++ b/apps/sim/lib/core/outbox/processor.ts @@ -31,19 +31,6 @@ export async function runOutboxProcessor(): Promise { 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) { diff --git a/apps/sim/lib/projects/files/__integration__/private-purge.integration.ts b/apps/sim/lib/projects/files/__integration__/private-purge.integration.ts index 99e7d4c3c09..3993fce5a51 100644 --- a/apps/sim/lib/projects/files/__integration__/private-purge.integration.ts +++ b/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') } ) diff --git a/apps/sim/lib/projects/files/prefix-cleanup.ts b/apps/sim/lib/projects/files/prefix-cleanup.ts index a001f752b9d..13786cb2eae 100644 --- a/apps/sim/lib/projects/files/prefix-cleanup.ts +++ b/apps/sim/lib/projects/files/prefix-cleanup.ts @@ -1,15 +1,12 @@ import { opendir, rmdir } from 'node:fs/promises' import { dirname, join } from 'node:path' import { db } from '@sim/db' -import { outboxEvent, project, uploadSession } from '@sim/db/schema' -import { createLogger } from '@sim/logger' +import { project, uploadSession } from '@sim/db/schema' import { describeError } from '@sim/utils/errors' import { isRecordLike } from '@sim/utils/object' -import { compareStrings } from '@sim/utils/string' -import { and, asc, eq, gt, inArray, isNull, lte, sql } from 'drizzle-orm' +import { and, asc, eq, gt, inArray, isNull, sql } from 'drizzle-orm' import { continueOutboxHandler, - deferOutboxHandler, enqueueOutboxEvent, type OutboxEventContext, type OutboxHandler, @@ -35,8 +32,6 @@ import { PROJECT_FILE_UPLOAD_BINDING_KEY } from '@/lib/uploads/upload-session/ty const PAGE_SIZE = 500 const CLOCK_SKEW_MS = 60_000 -const RECONCILE_INTERVAL_MS = 24 * 60 * 60 * 1000 -const logger = createLogger('ProjectStorageCleanup') type UploadCleanup = Pick< UploadSessionRecord, @@ -48,37 +43,6 @@ interface CleanupPayload { uploadCursor?: number } -interface ReconcilePayload { - cursor: string | null - page: string[] - index: number - unfinished: string[] - nextCursor: string | null - failedPrefixes: number -} - -/** Resume an existing inventory after its provider-outage retry budget and cooldown expire. */ -export async function recoverProjectStorageReconciliation(now = new Date()): Promise { - await db - .update(outboxEvent) - .set({ - status: 'pending', - attempts: 0, - lockedAt: null, - lastError: null, - processedAt: null, - availableAt: now, - }) - .where( - and( - eq(outboxEvent.id, PROJECT_STORAGE_RECONCILE_EVENT), - eq(outboxEvent.eventType, PROJECT_STORAGE_RECONCILE_EVENT), - eq(outboxEvent.status, 'dead_letter'), - lte(outboxEvent.processedAt, new Date(now.getTime() - RECONCILE_INTERVAL_MS)) - ) - ) -} - /** Expired signatures stop new transfers, but a transfer already in progress may finish later. */ function projectUploadCleanupAvailableAt(now = new Date()): Date { return new Date(now.getTime() + UPLOAD_URL_TTL_MS + CLOCK_SKEW_MS) @@ -139,27 +103,6 @@ async function enqueueSweeps( await enqueueOutboxEvent(executor, PROJECT_FILE_PREFIX_CLEANUP_EVENT, payload, { availableAt: projectUploadCleanupAvailableAt(retiredAt), }) - await executor - .insert(outboxEvent) - .values({ - id: PROJECT_STORAGE_RECONCILE_EVENT, - eventType: PROJECT_STORAGE_RECONCILE_EVENT, - payload: {}, - availableAt: projectUploadCleanupAvailableAt(retiredAt), - }) - .onConflictDoUpdate({ - target: outboxEvent.id, - set: { - status: 'pending', - payload: {}, - attempts: 0, - lockedAt: null, - lastError: null, - processedAt: null, - availableAt: projectUploadCleanupAvailableAt(retiredAt), - }, - setWhere: inArray(outboxEvent.status, ['completed', 'dead_letter']), - }) } /** Must follow Project and payer locks and precede directory/file locks in the retirement transaction. */ @@ -317,7 +260,7 @@ async function listKeys(prefix: string, signal: AbortSignal): Promise if (!config.containerName) throw new Error('Project blob container missing') const pages = (await getBlobServiceClient()) .getContainerClient(config.containerName) - .listBlobsFlat({ prefix, includeUncommitedBlobs: true, abortSignal: signal }) + .listBlobsFlat({ prefix, abortSignal: signal }) .byPage({ maxPageSize: PAGE_SIZE }) const page = await pages.next() return page.done ? [] : page.value.segment.blobItems.map((object) => object.name) @@ -361,199 +304,6 @@ async function sweepPrefix(projectId: string, context: OutboxEventContext): Prom return keys.length < PAGE_SIZE } -function readProjectId(prefix: string): string { - const match = /^project\/([a-zA-Z0-9_-]+)\/$/.exec(prefix) - if (!match) throw new Error('Storage provider returned an invalid Project prefix') - return match[1] -} - -async function listProjectPrefixes(cursor: string | null, signal: AbortSignal) { - signal.throwIfAborted() - if (USE_S3_STORAGE) { - const { ListObjectsV2Command } = await import('@aws-sdk/client-s3') - const { getS3Client } = await import('@/lib/uploads/providers/s3/client') - const page = await getS3Client().send( - new ListObjectsV2Command({ - Bucket: getStorageConfig('project').bucket, - Prefix: 'project/', - Delimiter: '/', - MaxKeys: PAGE_SIZE, - ...(cursor ? { ContinuationToken: cursor } : {}), - }), - { abortSignal: signal } - ) - return { - page: (page.CommonPrefixes ?? []).map((entry) => readProjectId(entry.Prefix ?? '')), - nextCursor: page.NextContinuationToken ?? null, - } - } - if (USE_BLOB_STORAGE) { - const { getBlobServiceClient } = await import('@/lib/uploads/providers/blob/client') - const config = getStorageConfig('project') - if (!config.containerName) throw new Error('Project blob container missing') - const pages = (await getBlobServiceClient()) - .getContainerClient(config.containerName) - .listBlobsByHierarchy('/', { - prefix: 'project/', - includeUncommitedBlobs: true, - abortSignal: signal, - }) - .byPage({ maxPageSize: PAGE_SIZE, ...(cursor ? { continuationToken: cursor } : {}) }) - const page = await pages.next() - return page.done - ? { page: [], nextCursor: null } - : { - page: (page.value.segment.blobPrefixes ?? []).map((entry) => readProjectId(entry.name)), - nextCursor: page.value.continuationToken || null, - } - } - if (USE_GCS_STORAGE) { - const { getGcsClient } = await import('@/lib/uploads/providers/google-cloud-storage/client') - const config = getStorageConfig('project') - if (!config.bucket) throw new Error('Project bucket missing') - const [, nextQuery, response] = await (await getGcsClient()).bucket(config.bucket).getFiles({ - prefix: 'project/', - delimiter: '/', - maxResults: PAGE_SIZE, - autoPaginate: false, - ...(cursor ? { pageToken: cursor } : {}), - }) - signal.throwIfAborted() - if ( - !isRecordLike(response) || - (response.prefixes !== undefined && !Array.isArray(response.prefixes)) - ) - throw new Error('Invalid Project storage listing') - const prefixes: unknown[] = response.prefixes ?? [] - return { - page: prefixes.map((prefix) => { - if (typeof prefix !== 'string') throw new Error('Invalid Project storage prefix') - return readProjectId(prefix) - }), - nextCursor: nextQuery?.pageToken ?? null, - } - } - let directory - try { - directory = await opendir(join(UPLOAD_DIR_SERVER, 'project')) - } catch (error) { - if (describeError(error).code === 'ENOENT') return { page: [], nextCursor: null } - throw error - } - const ids: string[] = [] - for await (const entry of directory) { - signal.throwIfAborted() - if (!entry.isDirectory() || (cursor && compareStrings(entry.name, cursor) <= 0)) continue - if (ids.length > PAGE_SIZE && compareStrings(entry.name, ids[ids.length - 1]) >= 0) continue - ids.push(readProjectId(`project/${entry.name}/`)) - ids.sort(compareStrings) - if (ids.length > PAGE_SIZE + 1) ids.pop() - } - const page = ids.slice(0, PAGE_SIZE) - return { page, nextCursor: ids.length > PAGE_SIZE ? page[page.length - 1] : null } -} - -function parseReconcilePayload(raw: unknown): ReconcilePayload { - if (!isRecordLike(raw)) throw new Error('Invalid Project reconciliation payload') - const cursor = raw.cursor ?? null - const nextCursor = raw.nextCursor ?? null - const page = raw.page ?? [] - const unfinished = raw.unfinished ?? [] - const index = raw.index ?? 0 - const failedPrefixes = raw.failedPrefixes ?? 0 - if ( - (cursor !== null && typeof cursor !== 'string') || - (nextCursor !== null && typeof nextCursor !== 'string') || - !Array.isArray(page) || - page.length > PAGE_SIZE || - !Array.isArray(unfinished) || - unfinished.length > PAGE_SIZE || - typeof index !== 'number' || - !Number.isSafeInteger(index) || - index < 0 || - index > page.length || - typeof failedPrefixes !== 'number' || - !Number.isSafeInteger(failedPrefixes) || - failedPrefixes < 0 - ) - throw new Error('Invalid Project reconciliation checkpoint') - function ownerIds(values: unknown[]): string[] { - return values.map((id) => { - if (typeof id !== 'string') throw new Error('Invalid Project reconciliation identity') - prefixFor(id) - return id - }) - } - return { - cursor, - nextCursor, - page: ownerIds(page), - unfinished: ownerIds(unfinished), - index, - failedPrefixes, - } -} - -const reconcilePrefixes: OutboxHandler = async (raw, context) => { - const payload = parseReconcilePayload(raw) - if (!payload.page.length) { - Object.assign(payload, await listProjectPrefixes(payload.cursor, context.signal), { - index: 0, - unfinished: [], - }) - await context.checkpointPayload({ ...payload }) - } - const owners = payload.page.length - ? await db.select({ id: project.id }).from(project).where(inArray(project.id, payload.page)) - : [] - const existing = new Set(owners.map((owner) => owner.id)) - for (let index = payload.index; index < payload.page.length; index++) { - context.signal.throwIfAborted() - const id = payload.page[index] - if (!existing.has(id)) { - try { - if (!(await sweepPrefix(id, context))) payload.unfinished.push(id) - } catch (error) { - context.signal.throwIfAborted() - payload.failedPrefixes++ - logger.warn('Project storage prefix cleanup failed; retrying on the next inventory', { - projectId: id, - error, - }) - } - } - payload.index = index + 1 - await context.checkpointPayload({ - index: payload.index, - unfinished: payload.unfinished, - failedPrefixes: payload.failedPrefixes, - }) - if (context.deadlineAt && Date.now() + 5000 >= context.deadlineAt) - return continueOutboxHandler('Continuing Project storage reconciliation') - } - if (payload.unfinished.length) { - await context.checkpointPayload({ page: payload.unfinished, index: 0, unfinished: [] }) - return continueOutboxHandler('Continuing partially cleaned Project prefixes') - } - await context.checkpointPayload({ - cursor: payload.nextCursor, - nextCursor: null, - page: [], - index: 0, - unfinished: [], - failedPrefixes: payload.nextCursor ? payload.failedPrefixes : 0, - }) - if (payload.nextCursor) return continueOutboxHandler('Continuing Project storage inventory') - return { - ...deferOutboxHandler( - `Project storage inventory reconciled with ${payload.failedPrefixes} failed prefixes`, - RECONCILE_INTERVAL_MS, - false - ), - resetAttempts: payload.failedPrefixes === 0, - } -} - const cleanupPrefix: OutboxHandler = async (raw, context) => { const payload = parsePayload(raw) const [owner] = await db @@ -595,7 +345,14 @@ const cleanupPrefix: OutboxHandler = async (raw, context) => { return continueOutboxHandler('Continuing Project storage cleanup') } +/** + * Retained only to drain rows armed by the former bucket inventory. The inventory deleted every + * `project//` prefix without a matching row, which is unsafe because this database is not the + * authority for a bucket shared with another deployment or restored to an earlier point in time. + */ +const drainRetiredStorageInventory: OutboxHandler = async () => {} + export const projectFilePrefixCleanupOutboxHandlers = { [PROJECT_FILE_PREFIX_CLEANUP_EVENT]: cleanupPrefix, - [PROJECT_STORAGE_RECONCILE_EVENT]: reconcilePrefixes, + [PROJECT_STORAGE_RECONCILE_EVENT]: drainRetiredStorageInventory, } satisfies OutboxHandlerRegistry