fix(account-deletion): bound position-only sweep resumes and reject foreign local page tokens

Provider cursors are opaque, so a moved listing position alone no longer
counts as progress past 1000 resumes per event; past that a run that only
moved its position spends an attempt, keeping the sweep bounded even by a
cursor that revisits positions. A local page token that is not a key under
the prefix (one saved under another storage provider) now fails to list, so
the sweep restarts the family instead of skipping every key before it.

Author: waleedlatif1

apps/sim/lib/uploads/core/prefix-sweep.ts

@@ -173,6 +173,8 @@ export async function listStorageKeyPage(
         : undefined
     )
   }
+  if (pageToken && !pageToken.startsWith(prefix))
+    throw new Error('Local storage page token is not a key under the prefix')
   const keys = await localKeys(prefix, signal, pageToken)
   return keyPage(keys, keys.length < STORAGE_KEY_PAGE_SIZE ? undefined : keys[keys.length - 1])
 }

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

@@ -89,6 +89,22 @@ async function put(key: string, body = 'x'): Promise<void> {
   await writeFile(join(root, key), body)
 }
 
+/** Writes `count` objects a surviving document names, answering `lookups` reference checks. */
+async function putKept(count: number, lookups = 1): Promise<string[]> {
+  const width = String(count - 1).length
+  const keys = Array.from(
+    { length: count },
+    (_, index) => `workspace/wsdoomed1/a-kept/${String(index).padStart(width, '0')}.bin`
+  )
+  for (const key of keys) await put(key)
+  for (let lookup = 0; lookup < lookups; lookup++)
+    queueTableRows(
+      schemaMock.document,
+      keys.map((key) => ({ key }))
+    )
+  return keys
+}
+
 async function exists(key: string): Promise<boolean> {
   try {
     await stat(join(root, key))
@@ -287,16 +303,8 @@ describe('deleted workspace storage sweep', () => {
   })
 
   it('reaches objects behind more kept pages than one run can walk, spending no attempt', async () => {
-    const keptKeys = Array.from(
-      { length: 1000 },
-      (_, index) => `workspace/wsdoomed1/a-kept/${String(index).padStart(4, '0')}.bin`
-    )
-    for (const key of [...keptKeys, 'workspace/wsdoomed1/z-unkept/0.bin']) await put(key)
-    for (let call = 0; call < 30; call++)
-      queueTableRows(
-        schemaMock.document,
-        keptKeys.map((key) => ({ key }))
-      )
+    const keptKeys = await putKept(1000, 30)
+    await put('workspace/wsdoomed1/z-unkept/0.bin')
 
     const outcomes = await sweepUntilSettled({ version: 1, workspaceId: 'wsdoomed1' }, 2)
 
@@ -335,16 +343,19 @@ describe('deleted workspace storage sweep', () => {
     for (const key of DOOMED_KEYS) expect(await exists(key)).toBe(false)
   })
 
+  it('walks the family from the start when its saved position is not a key under the prefix', async () => {
+    await expect(
+      sweep(
+        { version: 1, workspaceId: 'wsdoomed1', pageToken: 'zz-other-provider' },
+        outboxContext()
+      )
+    ).resolves.toBe(undefined)
+
+    for (const key of DOOMED_KEYS) expect(await exists(key)).toBe(false)
+  })
+
   it('spends an attempt when a restart from a rejected position only walks kept objects', async () => {
-    const keptKeys = Array.from(
-      { length: 500 },
-      (_, index) => `workspace/wsdoomed1/a-kept/${String(index).padStart(3, '0')}.bin`
-    )
-    for (const key of keptKeys) await put(key)
-    queueTableRows(
-      schemaMock.document,
-      keptKeys.map((key) => ({ key }))
-    )
+    const keptKeys = await putKept(500)
     vi.mocked(listStorageKeyPage).mockRejectedValueOnce(
       new Error('The continuation token is invalid')
     )
@@ -357,7 +368,29 @@ describe('deleted workspace storage sweep', () => {
 
     expect(outcome).toMatchObject({ outcome: 'deferred' })
     expect(outcome).not.toHaveProperty('consumeAttempt')
-    expect(checkpointPayload).toHaveBeenCalledWith({ family: 0, pageToken: keptKeys.at(-1) })
+    expect(checkpointPayload).toHaveBeenCalledWith({
+      family: 0,
+      pageToken: keptKeys.at(-1),
+      resumes: 0,
+    })
+  })
+
+  it('spends an attempt on a moved position once its resume budget is gone', async () => {
+    const keptKeys = await putKept(1000)
+    const checkpointPayload = vi.fn(async () => {})
+
+    const outcome = await sweep(
+      { version: 1, workspaceId: 'wsdoomed1', resumes: 1000 },
+      deadlineAfter(1, { checkpointPayload })
+    )
+
+    expect(outcome).toMatchObject({ outcome: 'deferred' })
+    expect(outcome).not.toHaveProperty('consumeAttempt')
+    expect(checkpointPayload).toHaveBeenCalledWith({
+      family: 0,
+      pageToken: keptKeys[499],
+      resumes: 1001,
+    })
   })
 
   it.each([
@@ -369,6 +402,8 @@ describe('deleted workspace storage sweep', () => {
     { version: 1, workspaceId: 'wsdoomed1', pageToken: 42 },
     { version: 1, workspaceId: 'wsdoomed1', pageToken: '' },
     { version: 1, workspaceId: 'wsdoomed1', pageToken: 'x'.repeat(4097) },
+    { version: 1, workspaceId: 'wsdoomed1', resumes: -1 },
+    { version: 1, workspaceId: 'wsdoomed1', resumes: '1' },
   ])('rejects an unbounded or malformed payload %o', async (payload) => {
     await expect(sweep(payload, outboxContext())).rejects.toThrow()
     for (const key of DOOMED_KEYS) expect(await exists(key)).toBe(true)

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

@@ -61,6 +61,13 @@ const ROW_PAGE_SIZE = 1000
 const OBJECTS_PER_EVENT = 100
 const MAX_PAGE_TOKEN_LENGTH = 4096
 
+/**
+ * Runs one workspace sweep may continue on a moved listing position alone. Provider cursors are
+ * opaque, so nothing here proves one never revisits a position; past this, a run that only moved
+ * its position spends an attempt, keeping the sweep bounded even by a cursor that cycles.
+ */
+const MAX_POSITION_RESUMES = 1000
+
 /**
  * Ceiling on exact keys one deletion records for durable cleanup. Past it the rest are logged
  * and left; the workspace prefix sweeps are unaffected, since they list storage instead.
@@ -101,6 +108,8 @@ interface WorkspaceSweepPayload {
   family: number
   /** Listing position inside `family` a stopped run saved; absent from events queued before it. */
   pageToken?: string
+  /** Runs continued on a moved listing position alone; see {@link MAX_POSITION_RESUMES}. */
+  resumes: number
 }
 
 interface ObjectCleanupPayload {
@@ -135,7 +144,10 @@ function parseWorkspaceSweep(value: unknown): WorkspaceSweepPayload {
       pageToken.length > MAX_PAGE_TOKEN_LENGTH)
   )
     throw new Error('Invalid workspace storage sweep page token')
-  return { workspaceId: value.workspaceId, family, ...(pageToken ? { pageToken } : {}) }
+  const resumes = value.resumes ?? 0
+  if (typeof resumes !== 'number' || !Number.isSafeInteger(resumes) || resumes < 0)
+    throw new Error('Invalid workspace storage sweep resume count')
+  return { workspaceId: value.workspaceId, family, resumes, ...(pageToken ? { pageToken } : {}) }
 }
 
 function isSafeKey(key: string): boolean {
@@ -294,7 +306,8 @@ async function sweepWorkspacePrefix(
  * recovery), so the sweep stops without touching storage; it looks again before every page it
  * deletes from. A run stopped by its deadline saves its listing position, so a family whose early
  * pages are all kept still reaches the objects behind them; moving that position counts as
- * progress, except after a restart from the start, so a cursor that never lists stays bounded.
+ * progress, except after a restart from the start or past {@link MAX_POSITION_RESUMES}, so a
+ * cursor that never lists or never ends stays bounded by the attempt budget.
  */
 const sweepWorkspaceStorage: OutboxHandler<unknown> = async (raw, context) => {
   const payload = parseWorkspaceSweep(raw)
@@ -331,8 +344,10 @@ const sweepWorkspaceStorage: OutboxHandler<unknown> = async (raw, context) => {
       return deferOutboxHandler('Workspace storage sweep made no progress on a full listing page')
     if (result.status === 'partial') {
       if (result.pageToken !== startToken) {
-        await context.checkpointPayload({ family, pageToken: result.pageToken ?? null })
-        if (!result.restarted) progressed = true
+        const movedOnly = !progressed && !result.restarted
+        const resumes = payload.resumes + (movedOnly ? 1 : 0)
+        await context.checkpointPayload({ family, pageToken: result.pageToken ?? null, resumes })
+        if (movedOnly && resumes <= MAX_POSITION_RESUMES) progressed = true
       }
       return progressed
         ? continueOutboxHandler('Continuing workspace storage sweep')