fix(execution): finish the removal sweep's edge cases

- The removal stop handler loads both job queues independently and sweeps
  whichever loaded, so a primary queue that fails to load no longer skips the
  database queue's sweep. The failure is still rethrown for the outbox to retry.
- A claimed resume is found under a parent log row in the legacy `paused`
  status too, one of the statuses a resume claims from.
- The database queue re-checks for a matching active job after a short batch,
  since a selected job that finished before the update reached it shrinks the
  batch while later jobs still match.
- The ban's completion log no longer reports its runs stopped when recording
  their stops failed.

Author: waleedlatif1

apps/sim/lib/core/async-jobs/backends/database.test.ts

@@ -2,6 +2,7 @@ import { asyncJobs } from '@sim/db'
 import {
   dbChainMockFns,
   flattenMockConditions,
+  queueTableRows,
   resetDbChainMock,
 } from '@sim/testing/mocks/database.mock'
 import { sleep } from '@sim/utils/helpers'
@@ -543,6 +544,15 @@ describe('DatabaseJobQueue workspace cancellation', () => {
     ).resolves.toEqual({ cancelledJobs: 230, complete: true })
   })
 
+  it('reports a short batch incomplete while matching jobs remain', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce(ids(99))
+    queueTableRows(asyncJobs, [{ id: 'job-100' }])
+
+    await expect(
+      new DatabaseJobQueue().cancelByWorkspace('workspace-1', { maxRuns: 1000 })
+    ).resolves.toEqual({ cancelledJobs: 99, complete: false })
+  })
+
   it('stops at its run cap and reports the sweep incomplete', async () => {
     dbChainMockFns.returning.mockResolvedValueOnce(ids(100)).mockResolvedValueOnce(ids(50, 100))
 

apps/sim/lib/core/async-jobs/backends/database.ts

@@ -561,8 +561,21 @@ export class DatabaseJobQueue implements JobQueueBackend {
       for (const { id } of cancelled) abortInlineJob(id)
       cancelledJobs += cancelled.length
       if (cancelled.length < batchSize) {
-        logger.info('Cancelled database jobs for workspace', { workspaceId, cancelledJobs })
-        return { cancelledJobs, complete: true }
+        /**
+         * A short batch is not proof the sweep is done: a selected job that finished before the
+         * update reached it shrinks the batch while later jobs may still match.
+         */
+        const [remaining] = await db
+          .select({ id: asyncJobs.id })
+          .from(asyncJobs)
+          .where(active)
+          .limit(1)
+        logger.info('Cancelled database jobs for workspace', {
+          workspaceId,
+          cancelledJobs,
+          complete: !remaining,
+        })
+        return { cancelledJobs, complete: !remaining }
       }
     }
 

apps/sim/lib/execution/removal-stop-records.ts

@@ -103,7 +103,7 @@ export async function collectExecutionStops(
         eq(resumeQueue.status, 'claimed')
       )
     )
-    .where(and(inArray(workflowExecutionLogs.status, ['running', 'pending']), inScope))
+    .where(and(inArray(workflowExecutionLogs.status, ['running', 'pending', 'paused']), inScope))
 
   const executions = new Map<string, PlannedExecutionStop>()
   for (const row of [...running, ...resuming]) {

apps/sim/lib/execution/removal-stops.integration.ts

@@ -103,7 +103,7 @@ async function seedWorkflow(userId: string, workspaceId: string) {
 async function seedExecution(
   workspaceId: string,
   workflowId: string,
-  status: 'running' | 'pending' = 'running'
+  status: 'running' | 'pending' | 'paused' = 'running'
 ) {
   const snapshotId = generateId()
   snapshotIds.push(snapshotId)
@@ -131,8 +131,12 @@ async function seedExecution(
 }
 
 /** A paused run whose resume a worker has claimed and is executing under its own id. */
-async function seedActiveResume(workspaceId: string, workflowId: string) {
-  const parentExecutionId = await seedExecution(workspaceId, workflowId, 'pending')
+async function seedActiveResume(
+  workspaceId: string,
+  workflowId: string,
+  parentStatus: 'pending' | 'paused' = 'pending'
+) {
+  const parentExecutionId = await seedExecution(workspaceId, workflowId, parentStatus)
   const pausedExecutionId = generateId()
   await db.insert(pausedExecutions).values({
     id: pausedExecutionId,
@@ -246,6 +250,11 @@ describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan',
     const doomedWorkflowId = await seedWorkflow(departingId, doomedWorkspaceId)
     const running = await seedExecution(doomedWorkspaceId, doomedWorkflowId)
     const resuming = await seedActiveResume(doomedWorkspaceId, doomedWorkflowId)
+    const resumingLegacyPaused = await seedActiveResume(
+      doomedWorkspaceId,
+      await seedWorkflow(departingId, doomedWorkspaceId),
+      'paused'
+    )
     const queued = await seedQueuedJob(doomedWorkspaceId)
     const unrelatedJob = await seedQueuedJob(doomedWorkspaceId, 'cleanup-logs')
     const keptWorkspaceId = await seedWorkspace(bystanderId)
@@ -259,6 +268,7 @@ describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan',
 
     await expect.poll(() => isCancelled(running)).toBe(true)
     await expect.poll(() => isCancelled(resuming)).toBe(true)
+    await expect.poll(() => isCancelled(resumingLegacyPaused)).toBe(true)
     const recorded = await stopEvents([doomedWorkspaceId])
     expect(recorded.map((event) => event.status)).toEqual(['pending'])
 

apps/sim/lib/execution/removal-stops.test.ts

@@ -4,6 +4,7 @@
  * - Redis refuses the cancellation write → retry, and the queue sweep still runs
  * - no Redis at all → the process-local flag is all there is, so the stop completes
  * - a queue sweep fails → retry
+ * - one queue fails to load → the other is still swept, then retry
  * - a sweep stops at its cap having cancelled runs → continue without spending an attempt
  * - a sweep stops at its cap having cancelled nothing → retry, so it cannot loop forever
  */
@@ -65,6 +66,13 @@ describe('execution removal stop outcome', () => {
     await expect(stopRemovedExecutions(payload, context())).rejects.toThrow('provider unavailable')
   })
 
+  it('still sweeps the database queue when the primary queue fails to load, then retries', async () => {
+    asyncJobsMockFns.mockGetJobQueue.mockRejectedValueOnce(new Error('queue unavailable'))
+
+    await expect(stopRemovedExecutions(payload, context())).rejects.toThrow('queue unavailable')
+    expect(mockJobQueue.cancelByWorkspace).toHaveBeenCalledWith('workspace-1', expect.anything())
+  })
+
   it('continues without spending an attempt when the sweep stops at its cap after progress', async () => {
     mockJobQueue.cancelByWorkspace.mockResolvedValue({ cancelledJobs: 1000, complete: false })
 

apps/sim/lib/execution/removal-stops.ts

@@ -1,6 +1,7 @@
 import { db } from '@sim/db'
 import { z } from 'zod'
 import { getInlineJobQueue, getJobQueue } from '@/lib/core/async-jobs'
+import type { JobQueueBackend } from '@/lib/core/async-jobs/types'
 import {
   continueOutboxHandler,
   type OutboxHandler,
@@ -29,10 +30,17 @@ const payloadSchema = z.object({
 
 /**
  * Sweeps the primary queue and the database queue that inline webhook runs use even when
- * Trigger.dev is primary. Both are attempted before the first failure is rethrown.
+ * Trigger.dev is primary. Each queue that loads is swept, even when the other fails to load or
+ * to sweep, before the first failure is rethrown for the outbox to retry.
  */
 async function sweepWorkspaceJobs(workspaceId: string, signal: AbortSignal) {
-  const queues = new Set([await getJobQueue(), await getInlineJobQueue()])
+  const resolved = await Promise.allSettled([getJobQueue(), getInlineJobQueue()])
+  const queues = new Set<JobQueueBackend>()
+  const failures: unknown[] = []
+  for (const result of resolved) {
+    if (result.status === 'fulfilled') queues.add(result.value)
+    else failures.push(result.reason)
+  }
   const results = await Promise.allSettled(
     [...queues].map((queue) =>
       queue.cancelByWorkspace(workspaceId, { maxRuns: SWEEP_MAX_RUNS_PER_ATTEMPT, signal })
@@ -41,10 +49,13 @@ async function sweepWorkspaceJobs(workspaceId: string, signal: AbortSignal) {
   let cancelledJobs = 0
   let complete = true
   for (const result of results) {
-    if (result.status === 'rejected') throw result.reason
-    cancelledJobs += result.value.cancelledJobs
-    complete &&= result.value.complete
+    if (result.status === 'rejected') failures.push(result.reason)
+    else {
+      cancelledJobs += result.value.cancelledJobs
+      complete &&= result.value.complete
+    }
   }
+  if (failures.length > 0) throw failures[0]
   return { cancelledJobs, complete }
 }
 

apps/sim/lib/workflows/lifecycle.ts

@@ -327,12 +327,14 @@ export async function disableUserResources(userId: string): Promise<void> {
    * not load that delivery. Logged, not thrown: the ban and archives have committed, and a retry
    * would find no unarchived workspace to stop.
    */
+  let stopFailed = false
   const stopRuns = async (workspaceIds: string[]): Promise<void> => {
     if (workspaceIds.length === 0) return
     try {
       const executions = await db.transaction((tx) => recordExecutionStops(tx, { workspaceIds }))
       await signalExecutionStops(executions)
     } catch (error) {
+      stopFailed = true
       logger.error(`[${requestId}] Failed to stop runs for banned user ${userId}`, {
         workspaceIds,
         error: getErrorMessage(error),
@@ -375,7 +377,7 @@ export async function disableUserResources(userId: string): Promise<void> {
   }
 
   logger.info(
-    `[${requestId}] Disabled resources for user ${userId}: archived ${ownedWorkspaces.length} workspaces, deleted API keys, stopped their runs`
+    `[${requestId}] Disabled resources for user ${userId}: archived ${ownedWorkspaces.length} workspaces, deleted API keys, ${stopFailed ? 'failed to stop some of their runs' : 'stopped their runs'}`
   )
 }