fix(execution): close the review's gaps in removal stops

- A ban records and signals each workspace's stops as soon as its archive
  commits, through the archive's post-commit stop hook, instead of after every
  archive's socket and webhook cleanup. It also stops runs in the owner's
  workspaces archived before the ban, which an archive leaves running.
- The outbox handler re-reads a workspace's running executions when it runs.
  An archived workspace keeps its rows, so a run admitted before the ban that
  logged after its stop was recorded is still signalled.
- The admin remove-user hook handlers move to lib/auth/admin-removal-stops.ts,
  and signal the removed runs even when recording their stops fails.
- The database queue's workspace sweep works in batches of 100 up to the
  caller's cap and checks its abort signal between them, like the Trigger.dev
  sweep, rather than one unbounded update reported as complete.

Author: waleedlatif1

apps/sim/lib/auth/admin-removal-stops.test.ts

@@ -0,0 +1,59 @@
+/**
+ * The `/admin/remove-user` hooks hand a plan from `before` to `after` across a non-transactional
+ * delete. Failure modes:
+ * - the outbox write fails after the delete committed → the in-flight runs must still be signalled
+ * - `after` runs for a different delete than the one `before` read → nothing is stopped
+ */
+import { workflowExecutionLogs, workspace } from '@sim/db/schema'
+import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing/mocks/database.mock'
+import { redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { holdAdminRemovalStops, stopAdminRemovedRuns } from '@/lib/auth/admin-removal-stops'
+
+const mockEval = vi.fn()
+
+function queueRunningExecution(executionId: string) {
+  queueTableRows(workspace, [])
+  queueTableRows(workflowExecutionLogs, [
+    { executionId, executionDeadlineAt: null, workspaceId: 'workspace-elsewhere' },
+  ])
+  queueTableRows(workflowExecutionLogs, [])
+}
+
+describe('admin user removal stops', () => {
+  beforeEach(() => {
+    resetDbChainMock()
+    mockEval.mockReset().mockResolvedValue(1)
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue({ eval: mockEval })
+  })
+
+  it('signals the removed user’s runs even when their stops cannot be recorded', async () => {
+    const user = { id: 'user-1' }
+    queueRunningExecution('execution-1')
+    await holdAdminRemovalStops(user)
+    dbChainMockFns.insert.mockImplementationOnce(() => {
+      throw new Error('outbox unavailable')
+    })
+
+    await expect(stopAdminRemovedRuns(user)).resolves.toBeUndefined()
+
+    expect(mockEval).toHaveBeenCalledWith(
+      expect.any(String),
+      1,
+      'execution:cancel:execution-1',
+      expect.anything(),
+      expect.anything(),
+      expect.anything(),
+      expect.anything()
+    )
+  })
+
+  it('stops nothing for a delete whose runs it did not read', async () => {
+    queueRunningExecution('execution-2')
+    await holdAdminRemovalStops({ id: 'user-2' })
+
+    await stopAdminRemovedRuns({ id: 'user-2' })
+
+    expect(mockEval).not.toHaveBeenCalled()
+  })
+})

apps/sim/lib/auth/admin-removal-stops.ts

@@ -0,0 +1,55 @@
+import { db } from '@sim/db'
+import { createLogger } from '@sim/logger'
+import { getErrorMessage } from '@sim/utils/errors'
+import {
+  collectAccountRemovalStops,
+  type ExecutionStopPlan,
+  enqueueExecutionStops,
+  signalExecutionStops,
+} from '@/lib/execution/removal-stop-records'
+
+const logger = createLogger('AdminRemovalStops')
+
+/**
+ * Plans read by {@link holdAdminRemovalStops} and acted on by {@link stopAdminRemovedRuns},
+ * keyed by the user entity Better Auth passes to both hooks of one delete. A delete that fails
+ * never reaches `after`, so its plan is dropped with the entity.
+ */
+const pendingStops = new WeakMap<object, ExecutionStopPlan>()
+
+/**
+ * The user delete's `before` hook: `/admin/remove-user` deletes the user row outside any
+ * transaction, and the cascade erases the rows naming its runs, so they are read now. A failed
+ * read lets the removal proceed as before rather than blocking an admin.
+ */
+export async function holdAdminRemovalStops(user: { id: string }): Promise<void> {
+  try {
+    pendingStops.set(user, await collectAccountRemovalStops(db, user.id))
+  } catch (error) {
+    logger.error('Failed to read the runs a user removal must stop', {
+      userId: user.id,
+      error: getErrorMessage(error),
+    })
+  }
+}
+
+/**
+ * The user delete's `after` hook, once the delete has committed: signals the in-flight runs at
+ * once, whether or not their stops could be recorded, and records the stops for the outbox
+ * worker's queue sweep and retries. Never throws.
+ */
+export async function stopAdminRemovedRuns(user: { id: string }): Promise<void> {
+  const plan = pendingStops.get(user)
+  pendingStops.delete(user)
+  if (!plan) return
+  try {
+    await enqueueExecutionStops(db, plan)
+  } catch (error) {
+    logger.error('Failed to record the stops of a removed user’s runs', {
+      userId: user.id,
+      error: getErrorMessage(error),
+    })
+  } finally {
+    await signalExecutionStops(plan.executions)
+  }
+}

apps/sim/lib/auth/auth.ts

@@ -38,6 +38,7 @@ import { readAttributionProperties } from '@/lib/analytics/attribution'
 import { FREEBUFF_CLICK_ID_COOKIE } from '@/lib/analytics/freebuff'
 import { reportFreebuffConversion } from '@/lib/analytics/freebuff.server'
 import { getAccessControlConfig, isEmailBlockedByAccessControl } from '@/lib/auth/access-control'
+import { holdAdminRemovalStops, stopAdminRemovedRuns } from '@/lib/auth/admin-removal-stops'
 import { createAnonymousSession, ensureAnonymousUserExists } from '@/lib/auth/anonymous'
 import { buildConnectorProviders } from '@/lib/auth/connectors/providers'
 import {
@@ -146,7 +147,6 @@ import {
   consumeOAuthCredentialDraftBinding,
   processCredentialDraft,
 } from '@/lib/credentials/draft-processor'
-import type { ExecutionStopPlan } from '@/lib/execution/removal-stop-records'
 import { sendEmail } from '@/lib/messaging/email/mailer'
 import { getFromEmailAddress, getPersonalEmailFrom } from '@/lib/messaging/email/utils'
 import { quickValidateEmail } from '@/lib/messaging/email/validation'
@@ -182,14 +182,6 @@ import { SSO_TRUSTED_PROVIDERS } from '@/ee/sso/constants'
 
 const logger = createLogger('Auth')
 
-/**
- * Workflow runs `/admin/remove-user` must stop, read by the user delete's `before` hook while the
- * rows naming them exist and enqueued by its `after` hook once the delete has committed. Keyed by
- * the user entity Better Auth passes to both hooks of the same delete; a delete that fails never
- * reaches `after`, so its entry is dropped with the entity.
- */
-const adminRemovalExecutionStops = new WeakMap<object, ExecutionStopPlan>()
-
 /**
  * A Stripe price that maps to no catalog plan leaves the stored plan in place. That is expected
  * for enterprise (priced per customer) and for a legacy plan still on the price it was sold at;
@@ -481,41 +473,12 @@ export const auth = betterAuth({
        * usage rows cascade away without `deleteUserAccount`. Better Auth runs `after` once the
        * delete commits, which is when the stored reporting prefixes that counted them must be
        * discarded, and when the runs the cascade orphaned are stopped: in-flight ones at once,
-       * queued ones by the outbox worker, since this module cannot load that delivery. The
-       * delete is not transactional, so those runs are read just before it; a failed read lets
-       * the removal proceed as before rather than blocking an admin.
+       * queued ones by the outbox worker, since this module cannot load that delivery.
        */
       delete: {
-        before: async (user) => {
-          try {
-            const { collectAccountRemovalStops } = await import(
-              '@/lib/execution/removal-stop-records'
-            )
-            adminRemovalExecutionStops.set(user, await collectAccountRemovalStops(db, user.id))
-          } catch (error) {
-            logger.error('[databaseHooks.user.delete.before] Failed to read runs to stop', {
-              userId: user.id,
-              error: getErrorMessage(error),
-            })
-          }
-        },
+        before: holdAdminRemovalStops,
         after: async (user) => {
-          const executionStops = adminRemovalExecutionStops.get(user)
-          adminRemovalExecutionStops.delete(user)
-          if (executionStops) {
-            try {
-              const { enqueueExecutionStops, signalExecutionStops } = await import(
-                '@/lib/execution/removal-stop-records'
-              )
-              await enqueueExecutionStops(db, executionStops)
-              await signalExecutionStops(executionStops.executions)
-            } catch (error) {
-              logger.error('[databaseHooks.user.delete.after] Failed to stop removed runs', {
-                userId: user.id,
-                error: getErrorMessage(error),
-              })
-            }
-          }
+          await stopAdminRemovedRuns(user)
           await invalidateReportingUsagePrefixes()
         },
       },

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

@@ -515,7 +515,7 @@ describe('DatabaseJobQueue workspace cancellation', () => {
   })
 
   it('cancels only workflow job types, so the workspace’s other jobs keep running', async () => {
-    await new DatabaseJobQueue().cancelByWorkspace('workspace-1')
+    await new DatabaseJobQueue().cancelByWorkspace('workspace-1', { maxRuns: 1000 })
 
     const typeFilter = flattenMockConditions(dbChainMockFns.where.mock.calls[0][0]).find(
       (node) => node.type === 'inArray' && node.column === asyncJobs.type
@@ -528,4 +528,40 @@ describe('DatabaseJobQueue workspace cancellation', () => {
       'workflow-group-cell',
     ])
   })
+
+  const ids = (count: number, offset = 0) =>
+    Array.from({ length: count }, (_, index) => ({ id: `job-${offset + index}` }))
+
+  it('keeps cancelling in batches until a short batch shows none are left', async () => {
+    dbChainMockFns.returning
+      .mockResolvedValueOnce(ids(100))
+      .mockResolvedValueOnce(ids(100, 100))
+      .mockResolvedValueOnce(ids(30, 200))
+
+    await expect(
+      new DatabaseJobQueue().cancelByWorkspace('workspace-1', { maxRuns: 1000 })
+    ).resolves.toEqual({ cancelledJobs: 230, complete: true })
+  })
+
+  it('stops at its run cap and reports the sweep incomplete', async () => {
+    dbChainMockFns.returning.mockResolvedValueOnce(ids(100)).mockResolvedValueOnce(ids(50, 100))
+
+    await expect(
+      new DatabaseJobQueue().cancelByWorkspace('workspace-1', { maxRuns: 150 })
+    ).resolves.toEqual({ cancelledJobs: 150, complete: false })
+    expect(dbChainMockFns.returning).toHaveBeenCalledTimes(2)
+  })
+
+  it('cancels nothing once its caller’s time is up', async () => {
+    const controller = new AbortController()
+    controller.abort()
+
+    await expect(
+      new DatabaseJobQueue().cancelByWorkspace('workspace-1', {
+        maxRuns: 1000,
+        signal: controller.signal,
+      })
+    ).resolves.toEqual({ cancelledJobs: 0, complete: false })
+    expect(dbChainMockFns.update).not.toHaveBeenCalled()
+  })
 })

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

@@ -19,11 +19,13 @@ import {
   type JobType,
   validateMaxDurationSeconds,
   WORKSPACE_EXECUTION_JOB_TYPES,
+  type WorkspaceJobCancellationOptions,
   type WorkspaceJobCancellationResult,
 } from '@/lib/core/async-jobs/types'
 import { recordExecutionCancellationBackendResult } from '@/lib/core/execution-limits/metrics'
 
 const logger = createLogger('DatabaseJobQueue')
+const WORKSPACE_CANCELLATION_BATCH_SIZE = 100
 
 type AsyncJobRow = typeof asyncJobs.$inferSelect
 type Runner = NonNullable<EnqueueOptions['runner']>
@@ -523,39 +525,52 @@ export class DatabaseJobQueue implements JobQueueBackend {
    * Cancels the workspace's pending and processing rows, so no process can claim them, and
    * aborts the runners this process holds for them. A runner another process holds stops through
    * its execution's cancellation signal instead. Batches run through `batchEnqueueAndWait` keep
-   * no row and are left to that signal and their dispatch's own cancellation.
+   * no row and are left to that signal and their dispatch's own cancellation. Works in batches
+   * up to `maxRuns`, checking `signal` between them, like the Trigger.dev sweep.
    */
-  async cancelByWorkspace(workspaceId: string): Promise<WorkspaceJobCancellationResult> {
-    const now = new Date()
-    const cancelled = await db
-      .update(asyncJobs)
-      .set({
-        status: JOB_STATUS.CANCELLED,
-        completedAt: now,
-        error: 'Cancelled',
-        updatedAt: now,
-      })
-      .where(
-        and(
-          inArray(asyncJobs.status, [JOB_STATUS.PENDING, JOB_STATUS.PROCESSING]),
-          inArray(asyncJobs.type, [...WORKSPACE_EXECUTION_JOB_TYPES]),
-          sql`${asyncJobs.metadata}->>'workspaceId' = ${workspaceId}`
+  async cancelByWorkspace(
+    workspaceId: string,
+    options: WorkspaceJobCancellationOptions
+  ): Promise<WorkspaceJobCancellationResult> {
+    const active = and(
+      inArray(asyncJobs.status, [JOB_STATUS.PENDING, JOB_STATUS.PROCESSING]),
+      inArray(asyncJobs.type, [...WORKSPACE_EXECUTION_JOB_TYPES]),
+      sql`${asyncJobs.metadata}->>'workspaceId' = ${workspaceId}`
+    )
+    let cancelledJobs = 0
+    let complete = false
+    while (cancelledJobs < options.maxRuns && !options.signal?.aborted) {
+      const batchSize = Math.min(WORKSPACE_CANCELLATION_BATCH_SIZE, options.maxRuns - cancelledJobs)
+      const now = new Date()
+      const cancelled = await db
+        .update(asyncJobs)
+        .set({ status: JOB_STATUS.CANCELLED, completedAt: now, error: 'Cancelled', updatedAt: now })
+        .where(
+          and(
+            active,
+            inArray(
+              asyncJobs.id,
+              db.select({ id: asyncJobs.id }).from(asyncJobs).where(active).limit(batchSize)
+            )
+          )
         )
-      )
-      .returning({ id: asyncJobs.id })
+        .returning({ id: asyncJobs.id })
 
-    for (const { id } of cancelled) {
-      const controller = inlineAbortControllers.get(id)
-      if (!controller) continue
-      controller.abort(new DOMException('user', 'AbortError'))
-      inlineAbortControllers.delete(id)
+      for (const { id } of cancelled) {
+        const controller = inlineAbortControllers.get(id)
+        if (!controller) continue
+        controller.abort(new DOMException('user', 'AbortError'))
+        inlineAbortControllers.delete(id)
+      }
+      cancelledJobs += cancelled.length
+      if (cancelled.length < batchSize) {
+        complete = true
+        break
+      }
     }
 
-    logger.info('Cancelled database jobs for workspace', {
-      workspaceId,
-      cancelledJobs: cancelled.length,
-    })
-    return { cancelledJobs: cancelled.length, complete: true }
+    logger.info('Cancelled database jobs for workspace', { workspaceId, cancelledJobs, complete })
+    return { cancelledJobs, complete }
   }
 
   cancelByKey(cancelKey: string): boolean {

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

@@ -73,7 +73,7 @@ function scopeCondition(executor: DbOrTx, scope: ExecutionStopScope): SQL | unde
  * removal is not transactional, immediately before it): a cascade erases these rows. An active
  * resume runs under its own execution id, so the claimed resume of a paused run is included.
  */
-async function collectExecutionStops(
+export async function collectExecutionStops(
   executor: DbOrTx,
   scope: ExecutionStopScope
 ): Promise<ExecutionStopPlan> {