fix(account-deletion): durably purge a deleted workspace's storage by prefix

Account deletion found the objects to erase by reading their rows before the
teardown transaction and deleting them in memory after commit. Anything that
read could not see stayed in storage forever: rows written between the read
and the workspace lock, rowless objects (table snapshots, compiled documents,
table-import leftovers, execution objects whose metadata insert failed),
everything past the 100k in-memory cap, everything after a crash following
commit, and the account's private files in workspaces that survive it, whose
rows cascade away with the user.

Inside the deletion transaction, after the workspace rows are locked, account
deletion now queues delayed outbox work that commits or rolls back with it:

- one prefix sweep per deleted workspace across workspace/, table-import/,
  execution/, table-snapshots/, copilot-doc-compiled/ and workspace-logos/
  in the bucket serving each, with a trailing-slash prefix built from a
  validated id;
- exact-key cleanup for what cascades outside those prefixes: knowledge-base
  documents, Copilot uploads and other non-prefixed files of deleted
  workspaces, and the account's private files elsewhere (attachments of its
  own chats, private contexts, files with no workspace). Handed-off, Project,
  organization files and attachments of someone else's chat are not queued.

Both wait out every upload URL signed before the deletion, refuse as a no-op
when the workspace or account row exists again (restore, PITR), and keep any
key a surviving workspace_files, workspace_file, workspace_file_version,
document or execution_large_values row still references. The sweep reuses the
Project prefix cleanup's provider listing and deletion, generalized by storage
context and given an in-run provider cursor; Project cleanup still lists one
page per run and deletes exactly as before. The existing in-memory purge is
unchanged and still runs.

Large execution values now pass cleanupOnMetadataFailure, so a failed
metadata insert deletes the freshly keyed object instead of orphaning it, as
execution files already do.

Author: waleedlatif1

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

@@ -44,6 +44,10 @@ import {
   WORKSPACE_FILE_LIVE_DOC_OUTBOX_EVENT,
   WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT,
 } from '@/lib/uploads/contexts/workspace/file-outbox-events'
+import {
+  ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT,
+  ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT,
+} from '@/lib/users/storage-sweep-events'
 import { WORKFLOW_DEPLOYMENT_OUTBOX_EVENTS } from '@/lib/workflows/deployment-outbox-events'
 import { MIGRATED_INVITATION_EMAIL_EVENT_TYPE } from '@/lib/workspaces/admin-move-event'
 import {
@@ -231,6 +235,13 @@ export const OUTBOX_HANDLER_GROUPS: readonly LazyOutboxHandlerGroup[] = [
       return projectFilePrefixCleanupOutboxHandlers
     },
   },
+  {
+    events: [ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT, ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT],
+    load: async () => {
+      const { accountStorageSweepOutboxHandlers } = await import('@/lib/users/storage-sweep')
+      return accountStorageSweepOutboxHandlers
+    },
+  },
   {
     events: [PROJECT_FILE_DOCUMENT_RETIRE_EVENT],
     load: async () => {

apps/sim/lib/execution/payloads/store.local.test.ts

@@ -0,0 +1,80 @@
+import { mkdtemp, readdir, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import {
+  largeValueMetadataMock,
+  largeValueMetadataMockFns,
+} from '@sim/testing/mocks/large-value-metadata.mock'
+import { uploadsConfigMock } from '@sim/testing/mocks/uploads-config.mock'
+import {
+  uploadsMetadataMock,
+  uploadsMetadataMockFns,
+} from '@sim/testing/mocks/uploads-metadata.mock'
+import { setUploadDirServer, uploadsSetupMock } from '@sim/testing/mocks/uploads-setup.mock'
+import { describeError } from '@sim/utils/errors'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+vi.mock('@/lib/uploads/core/setup.server', () => uploadsSetupMock)
+vi.mock('@/lib/uploads/config', () => uploadsConfigMock)
+vi.mock('@/lib/uploads/server/metadata', () => uploadsMetadataMock)
+vi.mock('@/lib/execution/payloads/large-value-metadata', () => largeValueMetadataMock)
+
+import { storeLargeValue } from '@/lib/execution/payloads/store'
+
+const { mockInsertFileMetadata, mockDeleteFileMetadata } = uploadsMetadataMockFns
+const scope = { workspaceId: 'workspace-1', workflowId: 'workflow-1', executionId: 'execution-1' }
+
+let directory: string
+
+async function storedObjects(): Promise<string[]> {
+  try {
+    return await readdir(join(directory, 'execution', ...Object.values(scope)))
+  } catch (error) {
+    if (describeError(error).code === 'ENOENT') return []
+    throw error
+  }
+}
+
+describe('large value persistence against real local storage', () => {
+  beforeEach(async () => {
+    directory = await mkdtemp(join(tmpdir(), 'sim-large-value-'))
+    setUploadDirServer(directory)
+    mockInsertFileMetadata.mockReset().mockResolvedValue({ id: 'file-1' })
+    mockDeleteFileMetadata.mockReset().mockResolvedValue(true)
+    largeValueMetadataMockFns.mockRegisterLargeValueOwner.mockResolvedValue(true)
+  })
+
+  afterEach(async () => {
+    await rm(directory, { recursive: true, force: true })
+  })
+
+  it('removes the uploaded object when its metadata row cannot be written', async () => {
+    mockInsertFileMetadata.mockRejectedValueOnce(new Error('workspace no longer exists'))
+    const json = JSON.stringify({ payload: 'x'.repeat(2048) })
+
+    await expect(
+      storeLargeValue(JSON.parse(json), json, Buffer.byteLength(json), {
+        ...scope,
+        userId: 'user-1',
+        requireDurable: true,
+      })
+    ).rejects.toThrow('Failed to persist large execution value')
+
+    expect(await storedObjects()).toEqual([])
+    expect(largeValueMetadataMockFns.mockRegisterLargeValueOwner).not.toHaveBeenCalled()
+  })
+
+  it('keeps the object once its metadata row is written', async () => {
+    const json = JSON.stringify({ payload: 'y'.repeat(2048) })
+
+    const ref = await storeLargeValue(JSON.parse(json), json, Buffer.byteLength(json), {
+      ...scope,
+      userId: 'user-1',
+      requireDurable: true,
+    })
+
+    expect((await storedObjects()).filter((name) => name.endsWith('.json'))).toEqual([
+      ref.key?.split('/').pop(),
+    ])
+  })
+})

apps/sim/lib/execution/payloads/store.ts

@@ -90,6 +90,7 @@ async function persistValue(
       context: 'execution',
       preserveKey: true,
       customKey: key,
+      cleanupOnMetadataFailure: true,
       metadata: {
         originalName: `large-value-${id}.json`,
         uploadedAt: new Date().toISOString(),

apps/sim/lib/projects/files/prefix-cleanup.ts

@@ -19,6 +19,7 @@ import {
 } from '@/lib/projects/files/outbox-events'
 import {
   getStorageConfig,
+  type StorageContext,
   USE_BLOB_STORAGE,
   USE_GCS_STORAGE,
   USE_S3_STORAGE,
@@ -45,6 +46,11 @@ interface CleanupPayload {
 
 /** Expired signatures stop new transfers, but a transfer already in progress may finish later. */
 export function projectUploadCleanupAvailableAt(now = new Date()): Date {
+  return storageCleanupAvailableAt(now)
+}
+
+/** Earliest moment no upload URL signed before `now` can still write an object. */
+export function storageCleanupAvailableAt(now = new Date()): Date {
   return new Date(now.getTime() + UPLOAD_URL_TTL_MS + CLOCK_SKEW_MS)
 }
 
@@ -242,57 +248,113 @@ async function localKeys(prefix: string, signal: AbortSignal): Promise<string[]>
   return keys
 }
 
-async function listKeys(prefix: string, signal: AbortSignal): Promise<string[]> {
+/** One provider page of object keys; `done` once the listing has nothing after this page. */
+interface StorageKeyPage {
+  keys: string[]
+  done: boolean
+  /** Provider cursor for the following page, valid only within the current run. */
+  nextPageToken?: string
+}
+
+/**
+ * Lists one page of object keys under `prefix` in the bucket serving `context`. A local listing
+ * has no cursor: it returns the first keys still on disk, so callers advance by deleting them.
+ */
+export async function listStorageKeyPage(
+  prefix: string,
+  context: StorageContext,
+  signal: AbortSignal,
+  pageToken?: string
+): Promise<StorageKeyPage> {
   signal.throwIfAborted()
   if (USE_S3_STORAGE) {
     const { ListObjectsV2Command } = await import('@aws-sdk/client-s3')
     const { getS3Client } = await import('@/lib/uploads/providers/s3/client')
-    const config = getStorageConfig('project')
+    const config = getStorageConfig(context)
     const page = await getS3Client().send(
-      new ListObjectsV2Command({ Bucket: config.bucket, Prefix: prefix, MaxKeys: PAGE_SIZE }),
+      new ListObjectsV2Command({
+        Bucket: config.bucket,
+        Prefix: prefix,
+        MaxKeys: PAGE_SIZE,
+        ...(pageToken ? { ContinuationToken: pageToken } : {}),
+      }),
       { abortSignal: signal }
     )
-    return (page.Contents ?? []).flatMap((object) => (object.Key ? [object.Key] : []))
+    const nextPageToken = page.IsTruncated ? page.NextContinuationToken : undefined
+    return {
+      keys: (page.Contents ?? []).flatMap((object) => (object.Key ? [object.Key] : [])),
+      done: !nextPageToken,
+      ...(nextPageToken ? { nextPageToken } : {}),
+    }
   }
   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 config = getStorageConfig(context)
+    if (!config.containerName) throw new Error(`Blob container missing for ${context} storage`)
     const pages = (await getBlobServiceClient())
       .getContainerClient(config.containerName)
       .listBlobsFlat({ prefix, abortSignal: signal })
-      .byPage({ maxPageSize: PAGE_SIZE })
+      .byPage({ maxPageSize: PAGE_SIZE, ...(pageToken ? { continuationToken: pageToken } : {}) })
     const page = await pages.next()
-    return page.done ? [] : page.value.segment.blobItems.map((object) => object.name)
+    if (page.done) return { keys: [], done: true }
+    const nextPageToken = page.value.continuationToken || undefined
+    return {
+      keys: page.value.segment.blobItems.map((object) => object.name),
+      done: !nextPageToken,
+      ...(nextPageToken ? { nextPageToken } : {}),
+    }
   }
   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 [files] = await (await getGcsClient())
-      .bucket(config.bucket)
-      .getFiles({ prefix, maxResults: PAGE_SIZE, autoPaginate: false })
+    const config = getStorageConfig(context)
+    if (!config.bucket) throw new Error(`Bucket missing for ${context} storage`)
+    const [files, nextQuery] = await (await getGcsClient()).bucket(config.bucket).getFiles({
+      prefix,
+      maxResults: PAGE_SIZE,
+      autoPaginate: false,
+      ...(pageToken ? { pageToken } : {}),
+    })
     signal.throwIfAborted()
-    return files.map((file) => file.name)
+    const nextPageToken =
+      isRecordLike(nextQuery) && typeof nextQuery.pageToken === 'string' && nextQuery.pageToken
+        ? nextQuery.pageToken
+        : undefined
+    return {
+      keys: files.map((file) => file.name),
+      done: !nextPageToken,
+      ...(nextPageToken ? { nextPageToken } : {}),
+    }
   }
-  return localKeys(prefix, signal)
+  const keys = await localKeys(prefix, signal)
+  return { keys, done: keys.length < PAGE_SIZE }
 }
 
-async function sweepPrefix(projectId: string, context: OutboxEventContext): Promise<boolean> {
-  const prefix = prefixFor(projectId)
-  const keys = await listKeys(prefix, context.signal)
+/**
+ * Deletes listed keys, each of which must sit under `prefix`, and prunes local directories the
+ * deletions empty, never above the prefix itself. Stops before another deletion once the handler
+ * deadline is near; `complete` is false when it did.
+ */
+export async function deleteStorageKeysUnderPrefix(
+  prefix: string,
+  context: StorageContext,
+  keys: readonly string[],
+  outbox: Pick<OutboxEventContext, 'signal' | 'deadlineAt'>
+): Promise<{ deleted: number; complete: boolean }> {
+  if (prefix.length < 2 || !prefix.endsWith('/')) throw new Error('Unbounded storage prefix')
   let deleted = 0
   for (const key of keys) {
-    if (!key.startsWith(prefix)) throw new Error('Storage provider returned a foreign Project key')
-    context.signal.throwIfAborted()
-    if (deleted > 0 && context.deadlineAt && Date.now() + 5000 >= context.deadlineAt) return false
+    if (!key.startsWith(prefix))
+      throw new Error('Storage provider returned a key outside the prefix')
+    outbox.signal.throwIfAborted()
+    if (deleted > 0 && outbox.deadlineAt && Date.now() + 5000 >= outbox.deadlineAt)
+      return { deleted, complete: false }
     try {
-      await deleteFile({ key, context: 'project', signal: context.signal })
+      await deleteFile({ key, context, signal: outbox.signal })
     } catch (error) {
       if (describeError(error).code !== 'ENOENT') throw error
     }
     if (!USE_S3_STORAGE && !USE_BLOB_STORAGE && !USE_GCS_STORAGE) {
-      const root = join(UPLOAD_DIR_SERVER, 'project', projectId)
+      const root = join(UPLOAD_DIR_SERVER, prefix.slice(0, -1))
       let directory = dirname(join(UPLOAD_DIR_SERVER, key))
       while (directory === root || directory.startsWith(`${root}/`)) {
         await removeEmptyDirectory(directory)
@@ -301,7 +363,14 @@ async function sweepPrefix(projectId: string, context: OutboxEventContext): Prom
     }
     deleted++
   }
-  return keys.length < PAGE_SIZE
+  return { deleted, complete: true }
+}
+
+async function sweepPrefix(projectId: string, context: OutboxEventContext): Promise<boolean> {
+  const prefix = prefixFor(projectId)
+  const { keys } = await listStorageKeyPage(prefix, 'project', context.signal)
+  const { complete } = await deleteStorageKeysUnderPrefix(prefix, 'project', keys, context)
+  return complete && keys.length < PAGE_SIZE
 }
 
 const cleanupPrefix: OutboxHandler<unknown> = async (raw, context) => {

apps/sim/lib/users/account-deletion.ts

@@ -54,6 +54,7 @@ import {
 } from '@/lib/table/rows/executions'
 import type { StorageContext } from '@/lib/uploads'
 import { isUsingCloudStorage, StorageService } from '@/lib/uploads'
+import { enqueueAccountStorageCleanupInTx } from '@/lib/users/storage-sweep'
 import {
   listSharedResourceWorkspaceIdsForUser,
   listUnresolvedSharedResourceWorkspaces,
@@ -861,6 +862,11 @@ export async function deleteUserAccount(userId: string): Promise<AccountDeletion
      * workspace and user deletes cascade their rows away; delivered after commit.
      */
     stoppedRuns = await stopLiveRunsTx(tx, { userId })
+    /**
+     * Queued while the doomed workspaces are locked and their rows still exist: every object
+     * that cascades away below gets a durable, delayed cleanup that commits with the deletion.
+     */
+    await enqueueAccountStorageCleanupInTx(tx, { userId, workspaceIds: doomedWorkspaceIds })
     if (doomedWorkspaceIds.length > 0) {
       /**
        * Re-checked here rather than trusted from the plan: a workspace that

apps/sim/lib/users/storage-sweep-events.ts

@@ -0,0 +1,2 @@
+export const ACCOUNT_WORKSPACE_STORAGE_SWEEP_EVENT = 'account-deletion.storage.workspace-sweep'
+export const ACCOUNT_STORAGE_OBJECT_CLEANUP_EVENT = 'account-deletion.storage.object-cleanup'