fix(account-deletion): retry a saved sweep position once before restarting its family

A single failed listing from a saved page token restarted the family from
the prefix start, so a transient provider error at a deep position threw
the position away. The first failure now records tokenFailures: 1 and
rethrows, so the outbox retries from the same position under its normal
attempt budget; only a second failure from that token restarts the family.
Any successful listing clears the count.

Author: waleedlatif1

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

@@ -331,28 +331,50 @@ describe('deleted workspace storage sweep', () => {
     expect(checkpointPayload).toHaveBeenCalledWith({ family: 1, pageToken: null })
   })
 
-  it('walks the family from the start when its saved position no longer lists', async () => {
-    vi.mocked(listStorageKeyPage).mockRejectedValueOnce(
-      new Error('The continuation token is invalid')
-    )
+  it('retries a saved position that failed to list once instead of walking from the start', async () => {
+    let payload: Record<string, unknown> = {
+      version: 1,
+      workspaceId: 'wsdoomed1',
+      pageToken: 'workspace/wsdoomed1/a.txt',
+    }
+    const checkpointPayload = vi.fn(async (patch: Record<string, unknown>) => {
+      payload = { ...payload, ...patch }
+    })
+    vi.mocked(listStorageKeyPage).mockRejectedValueOnce(new Error('Service unavailable'))
 
-    await expect(
-      sweep({ version: 1, workspaceId: 'wsdoomed1', pageToken: 'expired-token' }, outboxContext())
-    ).resolves.toBe(undefined)
+    await expect(sweep(payload, outboxContext({ checkpointPayload }))).rejects.toThrow(
+      'Service unavailable'
+    )
+    for (const key of DOOMED_KEYS) expect(await exists(key)).toBe(true)
+    expect(payload).toMatchObject({ pageToken: 'workspace/wsdoomed1/a.txt', tokenFailures: 1 })
 
-    for (const key of DOOMED_KEYS) expect(await exists(key)).toBe(false)
+    await expect(sweep(payload, outboxContext({ checkpointPayload }))).resolves.toBe(undefined)
+    expect(await exists('workspace/wsdoomed1/a.txt')).toBe(true)
+    expect(await exists('workspace/wsdoomed1/nested/b.txt')).toBe(false)
+    expect(payload).toMatchObject({ tokenFailures: 0 })
   })
 
-  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)
+  it.each([
+    { name: 'no longer lists', pageToken: 'expired-token', rejects: true },
+    { name: 'is not a key under the prefix', pageToken: 'zz-other-provider', rejects: false },
+  ])(
+    'walks the family from the start once its saved position $name on a second attempt',
+    async ({ pageToken, rejects }) => {
+      let payload: Record<string, unknown> = { version: 1, workspaceId: 'wsdoomed1', pageToken }
+      const checkpointPayload = vi.fn(async (patch: Record<string, unknown>) => {
+        payload = { ...payload, ...patch }
+      })
+      const invalid = new Error('The continuation token is invalid')
+      if (rejects)
+        vi.mocked(listStorageKeyPage).mockRejectedValueOnce(invalid).mockRejectedValueOnce(invalid)
 
-    for (const key of DOOMED_KEYS) expect(await exists(key)).toBe(false)
-  })
+      await expect(sweep(payload, outboxContext({ checkpointPayload }))).rejects.toThrow()
+      for (const key of DOOMED_KEYS) expect(await exists(key)).toBe(true)
+
+      await expect(sweep(payload, outboxContext({ checkpointPayload }))).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 = await putKept(500)
@@ -362,7 +384,7 @@ describe('deleted workspace storage sweep', () => {
     const checkpointPayload = vi.fn(async () => {})
 
     const outcome = await sweep(
-      { version: 1, workspaceId: 'wsdoomed1', pageToken: 'expired-token' },
+      { version: 1, workspaceId: 'wsdoomed1', pageToken: 'expired-token', tokenFailures: 1 },
       deadlineAfter(2, { checkpointPayload })
     )
 
@@ -404,6 +426,8 @@ describe('deleted workspace storage sweep', () => {
     { version: 1, workspaceId: 'wsdoomed1', pageToken: 'x'.repeat(4097) },
     { version: 1, workspaceId: 'wsdoomed1', resumes: -1 },
     { version: 1, workspaceId: 'wsdoomed1', resumes: '1' },
+    { version: 1, workspaceId: 'wsdoomed1', tokenFailures: 2 },
+    { version: 1, workspaceId: 'wsdoomed1', tokenFailures: '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

@@ -110,6 +110,11 @@ interface WorkspaceSweepPayload {
   pageToken?: string
   /** Runs continued on a moved listing position alone; see {@link MAX_POSITION_RESUMES}. */
   resumes: number
+  /**
+   * Set once listing from `pageToken` failed. A second failure from the same token restarts the
+   * family; a single one is retried in place, since providers do not reliably mark a dead cursor.
+   */
+  tokenFailed: boolean
 }
 
 interface ObjectCleanupPayload {
@@ -147,7 +152,16 @@ function parseWorkspaceSweep(value: unknown): WorkspaceSweepPayload {
   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 } : {}) }
+  const tokenFailures = value.tokenFailures ?? 0
+  if (tokenFailures !== 0 && tokenFailures !== 1)
+    throw new Error('Invalid workspace storage sweep page token failure count')
+  return {
+    workspaceId: value.workspaceId,
+    family,
+    resumes,
+    tokenFailed: tokenFailures === 1,
+    ...(pageToken ? { pageToken } : {}),
+  }
 }
 
 function isSafeKey(key: string): boolean {
@@ -243,19 +257,22 @@ interface PrefixSweepResult {
  * does not keep, following the listing's page token so kept objects never hold back later ones. A
  * page that neither advances the token nor removes anything is `stalled` rather than re-listed
  * forever. A saved token is provider-specific and may no longer list (an expired or rejected
- * cursor), so a failed first listing from one retries once from the start of the prefix.
+ * cursor). The first failed listing from one is recorded and rethrown, so the outbox retries from
+ * the same position; when that token has failed before, the walk restarts from the prefix start.
  */
 async function sweepWorkspacePrefix(
   prefix: string,
   storageContext: StorageContext,
   context: OutboxEventContext,
   isReferenced: (keys: readonly string[]) => Promise<Set<string>>,
   workspaceExists: () => Promise<boolean>,
-  startToken: string | undefined
+  startToken: string | undefined,
+  startTokenFailed: boolean
 ): Promise<PrefixSweepResult> {
   let deleted = 0
   let pageToken = startToken
   let restarted = false
+  let failureRecorded = startTokenFailed
   for (;;) {
     if (outboxDeadlineNear(context)) return { deleted, status: 'partial', pageToken, restarted }
     let page: StorageKeyPage
@@ -264,6 +281,10 @@ async function sweepWorkspacePrefix(
     } catch (error) {
       if (pageToken === undefined || pageToken !== startToken || restarted) throw error
       context.signal.throwIfAborted()
+      if (!failureRecorded) {
+        await context.checkpointPayload({ tokenFailures: 1 })
+        throw error
+      }
       logger.warn('Workspace storage sweep could not resume its saved listing position', {
         prefix,
         error: describeError(error),
@@ -272,6 +293,10 @@ async function sweepWorkspacePrefix(
       pageToken = undefined
       continue
     }
+    if (failureRecorded) {
+      await context.checkpointPayload({ tokenFailures: 0 })
+      failureRecorded = false
+    }
     if (page.keys.some((key) => !key.startsWith(prefix)))
       throw new Error('Storage provider returned a key outside the workspace prefix')
     const owners = page.keys.flatMap((key) => sidecarOwner(key) ?? [])
@@ -336,7 +361,8 @@ const sweepWorkspaceStorage: OutboxHandler<unknown> = async (raw, context) => {
       context,
       isReferenced,
       workspaceExists,
-      startToken
+      startToken,
+      family === payload.family && payload.tokenFailed
     )
     if (result.status === 'refused') return
     if (result.deleted > 0) progressed = true