fix(execution): stop in-flight workflow runs when an account is deleted, removed or banned

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 {
@@ -469,11 +470,15 @@ export const auth = betterAuth({
       },
       /**
        * The admin plugin's `/admin/remove-user` deletes the user row through Better Auth, so its
-       * usage rows cascade away without `deleteUserAccount`. Better Auth runs this once the delete
-       * commits, which is when the stored reporting prefixes that counted them must be discarded.
+       * 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.
        */
       delete: {
-        after: async () => {
+        before: holdAdminRemovalStops,
+        after: async (user) => {
+          await stopAdminRemovedRuns(user)
           await invalidateReportingUsagePrefixes()
         },
       },

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

@@ -1,4 +1,10 @@
-import { dbChainMockFns, resetDbChainMock } from '@sim/testing/mocks/database.mock'
+import { asyncJobs } from '@sim/db'
+import {
+  dbChainMockFns,
+  flattenMockConditions,
+  queueTableRows,
+  resetDbChainMock,
+} from '@sim/testing/mocks/database.mock'
 import { sleep } from '@sim/utils/helpers'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
@@ -503,3 +509,69 @@ describe('DatabaseJobQueue a workflow job that completed with a recorded failure
     })
   })
 })
+
+describe('DatabaseJobQueue workspace cancellation', () => {
+  beforeEach(() => {
+    resetDbChainMock()
+  })
+
+  it('cancels only workflow job types, so the workspace’s other jobs keep running', async () => {
+    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
+    )
+    expect(typeFilter?.values).toEqual([
+      'workflow-execution',
+      'schedule-execution',
+      'webhook-execution',
+      'resume-execution',
+      '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('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))
+
+    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

@@ -18,10 +18,14 @@ import {
   type JobStatus,
   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']>
@@ -51,6 +55,15 @@ function rowToJob(row: AsyncJobRow): Job {
 const inlineAbortControllers = new Map<string, AbortController>()
 const inlineExecutionControllers = new Map<string, Map<AbortController, JobType>>()
 
+/** Aborts the runner this process holds for the job, if any. */
+function abortInlineJob(jobId: string): boolean {
+  const controller = inlineAbortControllers.get(jobId)
+  if (!controller) return false
+  controller.abort(new DOMException('user', 'AbortError'))
+  inlineAbortControllers.delete(jobId)
+  return true
+}
+
 /**
  * Per-cancel-key abort controllers for the `batchEnqueueAndWait` direct-call
  * path. Distinct from `inlineAbortControllers` (which keys by jobId) — this
@@ -420,13 +433,7 @@ export class DatabaseJobQueue implements JobQueueBackend {
     // Abort any in-process inline execution first so the running workflow
     // observes the signal and stops mid-flight. Then mark the row cancelled so
     // any future poller skips it without treating a user action as a failure.
-    const controller = inlineAbortControllers.get(jobId)
-    let aborted = false
-    if (controller) {
-      controller.abort(new DOMException('user', 'AbortError'))
-      inlineAbortControllers.delete(jobId)
-      aborted = true
-    }
+    const aborted = abortInlineJob(jobId)
 
     const now = new Date()
     await db
@@ -517,6 +524,65 @@ 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. Works in batches
+   * up to `maxRuns`, checking `signal` between them, like the Trigger.dev sweep.
+   */
+  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
+    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 })
+
+      for (const { id } of cancelled) abortInlineJob(id)
+      cancelledJobs += cancelled.length
+      if (cancelled.length < batchSize) {
+        /**
+         * 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 }
+      }
+    }
+
+    logger.info('Paused database job sweep for workspace', { workspaceId, cancelledJobs })
+    return { cancelledJobs, complete: false }
+  }
+
   cancelByKey(cancelKey: string): boolean {
     const controller = inlineCancelKeyControllers.get(cancelKey)
     if (!controller) return false