fix(execution): guard ban stops against transfers and refusing runs

- A ban records stops only for workspaces the user still owns, re-read in the
  recording transaction and held FOR SHARE until it commits, so a concurrent
  ownership transfer waits rather than having its new owner's runs stopped.
  Only workspace rows are locked, so no lock order is inverted.
- Trigger.dev's workspace sweep counts only cancelled runs toward its cap, with
  a separate bound on failures, so runs that keep refusing cancellation no
  longer hold every retry at the front of the listing. cancelByExecution is
  unchanged.

Author: waleedlatif1

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

@@ -924,3 +924,27 @@ describe('TriggerDevJobQueue workspace cancellation', () => {
     )
   })
 })
+
+describe('TriggerDevJobQueue workspace cancellation past refusing runs', () => {
+  beforeEach(() => {
+    mockList.mockReset()
+    mockCancel.mockReset()
+  })
+
+  it('reaches later runs when the first ones keep refusing cancellation', async () => {
+    const listed = Array.from({ length: 15 }, (_, index) => ({
+      id: `run-${index}`,
+      tags: ['workspaceId:removed'],
+    }))
+    mockList.mockImplementation(() => createListPage(listed))
+    const refusing = new Set(['run-0', 'run-1', 'run-2'])
+    mockCancel.mockImplementation(async (runId: string) => {
+      if (refusing.has(runId)) throw new MockApiError(500, { message: 'refused' })
+    })
+    const queue = new TriggerDevJobQueue()
+
+    await expect(queue.cancelByWorkspace('removed', { maxRuns: 5 })).rejects.toThrow('refused')
+
+    expect(mockCancel).toHaveBeenCalledWith('run-9')
+  })
+})

apps/sim/lib/core/async-jobs/backends/trigger-dev.ts

@@ -675,7 +675,6 @@ export class TriggerDevJobQueue implements JobQueueBackend {
   ): Promise<WorkspaceJobCancellationResult> {
     const workspaceTag = buildWorkspaceTag(workspaceId)
     const state: CancellationScanState = { cancelledJobs: 0, failureCount: 0 }
-    let inspectedRuns = 0
     let complete = true
 
     await scanAndCancelTriggerRuns({
@@ -688,16 +687,24 @@ export class TriggerDevJobQueue implements JobQueueBackend {
           from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000),
           limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE,
         }),
+      /**
+       * Only cancelled runs count toward the cap, so runs that keep refusing cancellation cannot
+       * hold every retry at the front of the listing; a separate bound on failures keeps one call
+       * finite. Both are counted per chunk, so a call may overshoot `maxRuns` by under a chunk.
+       */
       shouldStop: () => {
-        if (inspectedRuns < options.maxRuns && !options.signal?.aborted) return false
+        if (
+          state.cancelledJobs < options.maxRuns &&
+          state.failureCount < options.maxRuns &&
+          !options.signal?.aborted
+        ) {
+          return false
+        }
         complete = false
         return true
       },
-      selectCandidate: (run) => {
-        if (!run.tags.includes(workspaceTag)) return undefined
-        inspectedRuns += 1
-        return { id: run.id, verifyPayload: false }
-      },
+      selectCandidate: (run) =>
+        run.tags.includes(workspaceTag) ? { id: run.id, verifyPayload: false } : undefined,
       state,
     })
 

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

@@ -1,13 +1,18 @@
 /**
- * An in-process run is stopped by aborting it here, which needs no Redis. Failure mode: the
- * signal deadline passes before the Redis writes start, and the local runs are left going.
+ * Failure modes:
+ * - the signal deadline passes before the Redis writes start, and the local runs are left going
+ * - a ban stops a previously archived workspace whose ownership moved on, and its new owner's
+ *   runs are stopped
  */
+import { db } from '@sim/db'
+import { workspace } from '@sim/db/schema'
+import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing/mocks/database.mock'
 import { describe, expect, it, vi } from 'vitest'
 import {
   registerManualExecutionAborter,
   unregisterManualExecutionAborter,
 } from '@/lib/execution/manual-cancellation'
-import { signalExecutionStops } from '@/lib/execution/removal-stop-records'
+import { recordExecutionStops, signalExecutionStops } from '@/lib/execution/removal-stop-records'
 
 describe('signalling removed executions', () => {
   it('aborts the runs this process holds even after the signal deadline has passed', async () => {
@@ -30,3 +35,19 @@ describe('signalling removed executions', () => {
     expect(abort).toHaveBeenCalledOnce()
   })
 })
+
+describe('recording a banned owner’s stops', () => {
+  it('records nothing for a workspace the user no longer owns', async () => {
+    resetDbChainMock()
+    queueTableRows(workspace, [])
+
+    await expect(
+      db.transaction((tx) =>
+        recordExecutionStops(tx, { workspaceIds: ['transferred'], ownerId: 'banned-user' })
+      )
+    ).resolves.toEqual([])
+
+    expect(dbChainMockFns.for).toHaveBeenCalledWith('share')
+    expect(dbChainMockFns.insert).not.toHaveBeenCalled()
+  })
+})

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

@@ -3,10 +3,10 @@ import { resumeQueue, workflow, workflowExecutionLogs, workspace } from '@sim/db
 import { createLogger } from '@sim/logger'
 import { describeError } from '@sim/utils/errors'
 import { chunkArray } from '@sim/utils/helpers'
-import { and, eq, inArray, or, type SQL } from 'drizzle-orm'
+import { and, asc, eq, inArray, or, type SQL } from 'drizzle-orm'
 import { enqueueOutboxEvents, MAX_BULK_ENQUEUE_EVENTS } from '@/lib/core/outbox/service'
 import { mapWithConcurrency } from '@/lib/core/utils/concurrency'
-import type { DbOrTx } from '@/lib/db/types'
+import type { DbOrTx, DbTransaction } from '@/lib/db/types'
 import { markExecutionCancelled } from '@/lib/execution/cancellation'
 import { abortManualExecution } from '@/lib/execution/manual-cancellation'
 import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event'
@@ -181,10 +181,28 @@ export async function enqueueExecutionStops(
  * Returns the executions, to signal at once with {@link signalExecutionStops} after commit.
  */
 export async function recordExecutionStops(
-  executor: DbOrTx,
-  scope: ExecutionStopScope
+  executor: DbTransaction,
+  scope: ExecutionStopScope & {
+    /**
+     * Stop only the workspaces this user still owns. Their rows are held `FOR SHARE` until the
+     * caller commits, so an ownership transfer racing the stop waits for it instead of having its
+     * new owner's runs stopped. Only workspace rows are locked, so no lock order is inverted.
+     */
+    ownerId?: string
+  }
 ): Promise<ExecutionStopPlan['executions']> {
-  const plan = await collectExecutionStops(executor, scope)
+  let { workspaceIds } = scope
+  if (scope.ownerId && workspaceIds.length > 0) {
+    const stillOwned = await executor
+      .select({ id: workspace.id })
+      .from(workspace)
+      .where(and(inArray(workspace.id, [...workspaceIds]), eq(workspace.ownerId, scope.ownerId)))
+      .orderBy(asc(workspace.id))
+      .for('share')
+    workspaceIds = stillOwned.map((row) => row.id)
+    if (workspaceIds.length === 0 && !scope.workflowsCreatedBy) return []
+  }
+  const plan = await collectExecutionStops(executor, { ...scope, workspaceIds })
   await enqueueExecutionStops(executor, plan)
   return plan.executions
 }

apps/sim/lib/workflows/lifecycle.ts

@@ -332,7 +332,9 @@ export async function disableUserResources(userId: string): Promise<void> {
   const stopRuns = async (workspaceIds: string[]): Promise<void> => {
     if (workspaceIds.length === 0) return
     try {
-      const executions = await db.transaction((tx) => recordExecutionStops(tx, { workspaceIds }))
+      const executions = await db.transaction((tx) =>
+        recordExecutionStops(tx, { workspaceIds, ownerId: userId })
+      )
       await signalExecutionStops(executions)
     } catch (error) {
       stopFailed = true