From 3ee51f80cf479b7666600db05036e8880a31c51d Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 14:59:42 -0700 Subject: [PATCH 1/9] fix(execution): stop in-flight workflow runs when an account is deleted, removed or banned No removal path stopped workflow executions. Account deletion cascades the doomed workspaces' log rows away, after which no cancel path can find the run, so it kept spending hosted provider keys for an account that no longer existed until it crashed on foreign keys at finalization. A ban archived the workspaces and stopped nothing; an admin removal cascaded them the same way. - Account deletion reads the running executions (and claimed resumes, which run under their own id) of the doomed workspaces in its transaction, while the workspace rows are locked, and records one outbox stop per workspace with the erasure. After commit the stops are delivered a few at a time; a stop that cannot be delivered stays in the outbox for its worker, and the committed deletion is never reported as failed. - A stop sets the cancellation flag each engine watches and sweeps the job queue for the workspace: a new JobQueueBackend.cancelByWorkspace cancels queued, delayed and running workflow jobs by the workspaceId tag on Trigger.dev (pages of 100, ten cancels at a time, 1,000 runs per attempt, continued without spending an attempt) and by metadata.workspaceId on the database queue, so queued runs no longer fail preprocessing later. - A missing Redis completes the stop: the flag is process-local by design and no retry could reach further. A refused Redis write or a failed sweep retries. - A ban stops the runs of the workspaces it archived; the admin remove-user hooks read the runs before the bare user delete and record them after it commits. Both sit in page module graphs that cannot load the delivery, so they signal in-flight runs at once and leave the queue sweep to the outbox worker. --- apps/sim/lib/auth/auth.ts | 48 ++- .../lib/core/async-jobs/backends/database.ts | 41 +++ .../async-jobs/backends/trigger-dev.test.ts | 181 ++++++++++ .../core/async-jobs/backends/trigger-dev.ts | 64 ++++ apps/sim/lib/core/async-jobs/types.ts | 35 ++ apps/sim/lib/core/outbox/handlers.ts | 8 + apps/sim/lib/execution/removal-stop-event.ts | 2 + .../sim/lib/execution/removal-stop-records.ts | 235 ++++++++++++ .../execution/removal-stops.integration.ts | 335 ++++++++++++++++++ apps/sim/lib/execution/removal-stops.test.ts | 89 +++++ apps/sim/lib/execution/removal-stops.ts | 131 +++++++ apps/sim/lib/users/account-deletion.ts | 13 + apps/sim/lib/workflows/lifecycle.ts | 48 ++- packages/testing/src/mocks/async-jobs.mock.ts | 1 + 14 files changed, 1223 insertions(+), 8 deletions(-) create mode 100644 apps/sim/lib/execution/removal-stop-event.ts create mode 100644 apps/sim/lib/execution/removal-stop-records.ts create mode 100644 apps/sim/lib/execution/removal-stops.integration.ts create mode 100644 apps/sim/lib/execution/removal-stops.test.ts create mode 100644 apps/sim/lib/execution/removal-stops.ts diff --git a/apps/sim/lib/auth/auth.ts b/apps/sim/lib/auth/auth.ts index 4efd9cf70c5..1381502965f 100644 --- a/apps/sim/lib/auth/auth.ts +++ b/apps/sim/lib/auth/auth.ts @@ -146,6 +146,7 @@ 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' @@ -181,6 +182,14 @@ 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() + /** * 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; @@ -469,11 +478,44 @@ 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. 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. */ delete: { - after: async () => { + 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), + }) + } + }, + 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 invalidateReportingUsagePrefixes() }, }, diff --git a/apps/sim/lib/core/async-jobs/backends/database.ts b/apps/sim/lib/core/async-jobs/backends/database.ts index 05c6b9bb01f..27588386758 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.ts +++ b/apps/sim/lib/core/async-jobs/backends/database.ts @@ -18,6 +18,8 @@ import { type JobStatus, type JobType, validateMaxDurationSeconds, + WORKSPACE_EXECUTION_JOB_TYPES, + type WorkspaceJobCancellationResult, } from '@/lib/core/async-jobs/types' import { recordExecutionCancellationBackendResult } from '@/lib/core/execution-limits/metrics' @@ -517,6 +519,45 @@ 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. + */ + async cancelByWorkspace(workspaceId: string): Promise { + 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}` + ) + ) + .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) + } + + logger.info('Cancelled database jobs for workspace', { + workspaceId, + cancelledJobs: cancelled.length, + }) + return { cancelledJobs: cancelled.length, complete: true } + } + cancelByKey(cancelKey: string): boolean { const controller = inlineCancelKeyControllers.get(cancelKey) if (!controller) return false diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts index 92df6bc14f0..2de17ae39b0 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts @@ -743,3 +743,184 @@ describe('TriggerDevJobQueue cancellation', () => { expect(maxActiveCancellations).toBe(10) }) }) + +describe('TriggerDevJobQueue workspace cancellation', () => { + interface FakeRun { + id: string + taskIdentifier: string + tags: string[] + status: string + } + + /** Stands in for Trigger.dev: runs carry the tags enqueue sent, and listing honors its filters. */ + function installFakeTrigger() { + const runs: FakeRun[] = [] + mockTrigger.mockImplementation( + async (taskIdentifier: string, _payload?: unknown, options?: unknown) => { + const run = { + id: `run-${runs.length}`, + taskIdentifier, + tags: (options as { tags?: string[] } | undefined)?.tags ?? [], + status: 'QUEUED', + } + runs.push(run) + return { id: run.id } + } + ) + mockList.mockImplementation((query?: unknown) => { + const filter = query as { tag: string; taskIdentifier: string[]; status: string[] } + const matches = runs.filter( + (run) => + run.tags.includes(filter.tag) && + filter.taskIdentifier.includes(run.taskIdentifier) && + filter.status.includes(run.status) + ) + return createListPage(matches) + }) + mockCancel.mockImplementation(async (runId: string) => { + const run = runs.find((candidate) => candidate.id === runId) + if (run) run.status = 'CANCELED' + }) + return runs + } + + function workspaceRun(workspaceId: string, executionId: string) { + return { + metadata: { + workspaceId, + correlation: { + executionId, + requestId: `request-${executionId}`, + source: 'workflow' as const, + workflowId: `workflow-${executionId}`, + }, + }, + } + } + + beforeEach(() => { + mockResolveTriggerRegion.mockResolvedValue('us-east-1') + mockList.mockReset() + mockCancel.mockReset() + mockTrigger.mockReset() + }) + + it('cancels every workflow run enqueued for the workspace and nothing else', async () => { + const runs = installFakeTrigger() + const queue = new TriggerDevJobQueue() + await queue.enqueue('workflow-execution', {}, workspaceRun('removed', 'a')) + await queue.enqueue('webhook-execution', {}, workspaceRun('removed', 'b')) + await queue.enqueue('schedule-execution', {}, workspaceRun('removed', 'c')) + await queue.enqueue('resume-execution', {}, workspaceRun('removed', 'd')) + await queue.enqueue( + 'workflow-group-cell', + {}, + { + ...workspaceRun('removed', 'e'), + tags: ['tableId:t', 'rowId:r', 'group:g'], + } + ) + await queue.enqueue('workflow-execution', {}, workspaceRun('kept', 'f')) + await queue.enqueue('cleanup-logs', {}, { metadata: { workspaceId: 'removed' } }) + + await expect(queue.cancelByWorkspace('removed', { maxRuns: 100 })).resolves.toEqual({ + cancelledJobs: 5, + complete: true, + }) + + expect(runs.filter((run) => run.status === 'CANCELED').map((run) => run.id)).toEqual([ + 'run-0', + 'run-1', + 'run-2', + 'run-3', + 'run-4', + ]) + }) + + it('stops at its run cap and reports the sweep incomplete until a later call finishes it', async () => { + const runs = installFakeTrigger() + const queue = new TriggerDevJobQueue() + for (let index = 0; index < 25; index++) { + await queue.enqueue('workflow-execution', {}, workspaceRun('removed', `run-${index}`)) + } + + await expect(queue.cancelByWorkspace('removed', { maxRuns: 10 })).resolves.toEqual({ + cancelledJobs: 10, + complete: false, + }) + await expect(queue.cancelByWorkspace('removed', { maxRuns: 10 })).resolves.toEqual({ + cancelledJobs: 10, + complete: false, + }) + await expect(queue.cancelByWorkspace('removed', { maxRuns: 10 })).resolves.toEqual({ + cancelledJobs: 5, + complete: true, + }) + expect(runs.every((run) => run.status === 'CANCELED')).toBe(true) + }) + + it('reports a sweep that reached exactly its cap as complete', async () => { + installFakeTrigger() + const queue = new TriggerDevJobQueue() + for (let index = 0; index < 10; index++) { + await queue.enqueue('workflow-execution', {}, workspaceRun('removed', `run-${index}`)) + } + + await expect(queue.cancelByWorkspace('removed', { maxRuns: 10 })).resolves.toEqual({ + cancelledJobs: 10, + complete: true, + }) + }) + + it('cancels the remaining runs before reporting one it could not cancel', async () => { + const runs = installFakeTrigger() + const queue = new TriggerDevJobQueue() + for (let index = 0; index < 12; index++) { + await queue.enqueue('workflow-execution', {}, workspaceRun('removed', `run-${index}`)) + } + const cancel = mockCancel.getMockImplementation() + mockCancel.mockImplementation(async (runId: string) => { + if (runId === 'run-3') throw new MockApiError(500, { message: 'unavailable' }) + return cancel?.(runId) + }) + + await expect(queue.cancelByWorkspace('removed', { maxRuns: 100 })).rejects.toThrow( + 'unavailable' + ) + + expect(runs.filter((run) => run.status === 'CANCELED')).toHaveLength(11) + }) + + it('cancels at most ten runs at a time', async () => { + installFakeTrigger() + const queue = new TriggerDevJobQueue() + for (let index = 0; index < 35; index++) { + await queue.enqueue('workflow-execution', {}, workspaceRun('removed', `run-${index}`)) + } + const cancel = mockCancel.getMockImplementation() + let active = 0 + let maxActive = 0 + mockCancel.mockImplementation(async (runId: string) => { + active += 1 + maxActive = Math.max(maxActive, active) + await Promise.resolve() + active -= 1 + return cancel?.(runId) + }) + + await queue.cancelByWorkspace('removed', { maxRuns: 100 }) + + expect(maxActive).toBe(10) + }) + + it('reports a failed listing so the caller retries', async () => { + mockList.mockImplementation(() => { + throw new MockApiError(503, { message: 'list unavailable' }) + }) + const queue = new TriggerDevJobQueue() + + await expect(queue.cancelByWorkspace('removed', { maxRuns: 100 })).rejects.toThrow( + 'list unavailable' + ) + }) +}) diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts index 3ef4eaf5a61..befb17a990d 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts @@ -24,6 +24,9 @@ import { type JobType, TERMINAL_JOB_STATUSES, validateMaxDurationSeconds, + WORKSPACE_EXECUTION_JOB_TYPES, + type WorkspaceJobCancellationOptions, + type WorkspaceJobCancellationResult, } from '@/lib/core/async-jobs/types' import { recordExecutionCancellationBackendResult } from '@/lib/core/execution-limits/metrics' @@ -38,6 +41,7 @@ const ACTIVE_TRIGGER_RUN_STATUSES = [ ] as const const CANCELLATION_LIST_PAGE_SIZE = 25 const CANCELLATION_OPERATION_CHUNK_SIZE = 10 +const WORKSPACE_CANCELLATION_LIST_PAGE_SIZE = 100 function classifyTriggerEnqueueError(error: unknown): AsyncJobEnqueueError { if (error instanceof ApiError && error.status && error.status >= 400 && error.status < 500) { return new AsyncJobEnqueueError(toError(error).message, { @@ -657,6 +661,66 @@ export class TriggerDevJobQueue implements JobQueueBackend { } } + async cancelByWorkspace( + workspaceId: string, + options: WorkspaceJobCancellationOptions + ): Promise { + const workspaceTag = `workspaceId:${workspaceId}` + const state: CancellationScanState = { cancelledJobs: 0, failureCount: 0 } + let pending: string[] = [] + let inspectedRuns = 0 + let complete = true + + const flush = async (): Promise => { + const runIds = pending + pending = [] + const results = await Promise.allSettled(runIds.map((runId) => this.cancelJob(runId))) + for (const result of results) { + if (result.status === 'rejected') recordCancellationCandidateFailure(state, result.reason) + else state.cancelledJobs += 1 + } + } + + try { + for await (const run of runs.list({ + tag: workspaceTag, + taskIdentifier: [...WORKSPACE_EXECUTION_JOB_TYPES], + status: [...ACTIVE_TRIGGER_RUN_STATUSES], + from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000), + limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE, + })) { + if (inspectedRuns >= options.maxRuns || options.signal?.aborted) { + complete = false + break + } + if (!run.tags.includes(workspaceTag)) continue + inspectedRuns += 1 + pending.push(run.id) + if (pending.length >= CANCELLATION_OPERATION_CHUNK_SIZE) await flush() + } + } catch (error) { + recordCancellationCandidateFailure(state, error) + } finally { + await flush() + } + + if (state.firstError) { + logger.error('Failed to cancel trigger.dev runs for workspace', { + workspaceId, + cancelledJobs: state.cancelledJobs, + failureCount: state.failureCount, + error: state.firstError.message, + }) + throw state.firstError + } + logger.info('Cancelled trigger.dev runs for workspace', { + workspaceId, + cancelledJobs: state.cancelledJobs, + complete, + }) + return { cancelledJobs: state.cancelledJobs, complete } + } + cancelByKey(_cancelKey: string): boolean { // No in-process AbortControllers to abort — trigger.dev runs are cancelled // by jobId or via tag sweep (see `cancelCellRunsByTags`). Callers that diff --git a/apps/sim/lib/core/async-jobs/types.ts b/apps/sim/lib/core/async-jobs/types.ts index c65b5df56fd..8efb0e9893a 100644 --- a/apps/sim/lib/core/async-jobs/types.ts +++ b/apps/sim/lib/core/async-jobs/types.ts @@ -176,6 +176,31 @@ export const EXECUTION_JOB_TYPES_BY_CANCELLATION_SCOPE = { resume: ['resume-execution'], } as const satisfies Record +/** + * Every job type that runs a workflow on a workspace's behalf. Each one is enqueued with + * `metadata.workspaceId`, so a workspace-wide stop can find them without knowing their executions. + */ +export const WORKSPACE_EXECUTION_JOB_TYPES = [ + 'workflow-execution', + 'schedule-execution', + 'webhook-execution', + 'resume-execution', + 'workflow-group-cell', +] as const satisfies readonly JobType[] + +export interface WorkspaceJobCancellationOptions { + /** Most runs one call may inspect; the rest are left for the caller's next call. */ + maxRuns: number + /** Stops the sweep between pages once the caller's time is up. */ + signal?: AbortSignal +} + +export interface WorkspaceJobCancellationResult { + cancelledJobs: number + /** False when the sweep stopped at `maxRuns` or its signal, so active runs may remain. */ + complete: boolean +} + export type AsyncJobEnqueueAcceptance = 'rejected' | 'unknown' interface AsyncJobEnqueueErrorOptions { @@ -311,6 +336,16 @@ export interface JobQueueBackend { * Returns `true` if a matching controller was found and aborted. */ cancelByKey(cancelKey: string): boolean + + /** + * Cancel every queued, delayed or running workflow job (`WORKSPACE_EXECUTION_JOB_TYPES`) + * enqueued for a workspace, for when the workspace itself is being removed. Throws when any + * listing or cancellation failed, after attempting the rest, so the caller can retry. + */ + cancelByWorkspace( + workspaceId: string, + options: WorkspaceJobCancellationOptions + ): Promise } export type AsyncBackendType = 'trigger-dev' | 'database' diff --git a/apps/sim/lib/core/outbox/handlers.ts b/apps/sim/lib/core/outbox/handlers.ts index 77c088f6b52..240f019b531 100644 --- a/apps/sim/lib/core/outbox/handlers.ts +++ b/apps/sim/lib/core/outbox/handlers.ts @@ -14,6 +14,7 @@ import { import { MEMBER_BILLING_RECONCILIATION_EVENT_TYPE } from '@/lib/billing/organizations/membership-reconciliation-event' import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events' import type { LazyOutboxHandlerGroup } from '@/lib/core/outbox/service' +import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' import { DIRECT_GRANT_EMAIL_EVENT_TYPE } from '@/lib/invitations/direct-grant-event' import { SLACK_SEARCH_TURN_EVENT } from '@/lib/knowledge/application/slack-search/turn-event' import { @@ -186,6 +187,13 @@ export const OUTBOX_HANDLER_GROUPS: readonly LazyOutboxHandlerGroup[] = [ return organizationResourceCleanupOutboxHandlers }, }, + { + events: [EXECUTION_REMOVAL_STOP_EVENT], + load: async () => { + const { executionRemovalStopOutboxHandlers } = await import('@/lib/execution/removal-stops') + return executionRemovalStopOutboxHandlers + }, + }, { events: [INBOX_CLEANUP_EVENT], load: async () => { diff --git a/apps/sim/lib/execution/removal-stop-event.ts b/apps/sim/lib/execution/removal-stop-event.ts new file mode 100644 index 00000000000..c2b01a49ccb --- /dev/null +++ b/apps/sim/lib/execution/removal-stop-event.ts @@ -0,0 +1,2 @@ +/** Outbox event that stops the workflow runs of a removed workspace, workflow set or account. */ +export const EXECUTION_REMOVAL_STOP_EVENT = 'execution.removal-stop' diff --git a/apps/sim/lib/execution/removal-stop-records.ts b/apps/sim/lib/execution/removal-stop-records.ts new file mode 100644 index 00000000000..98273cb716e --- /dev/null +++ b/apps/sim/lib/execution/removal-stop-records.ts @@ -0,0 +1,235 @@ +import type { db } from '@sim/db' +import { resumeQueue, workflow, workflowExecutionLogs, workspace } from '@sim/db/schema' +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 { 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 { markExecutionCancelled } from '@/lib/execution/cancellation' +import { abortManualExecution } from '@/lib/execution/manual-cancellation' +import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' + +const logger = createLogger('ExecutionRemovalStopRecords') + +/** Executions one event signals; bounds the payload of a workspace with many live runs. */ +const EXECUTIONS_PER_EVENT = 500 +const SIGNAL_CONCURRENCY = 10 + +/** + * What is being removed. Runs of a removed workspace or workflow lose the rows the ordinary + * cancel path reads (or, for an archive, the access they ran under), so they are stopped here: + * every in-flight execution gets the cancellation signal its engine watches, and every queued, + * delayed or running job enqueued for a removed workspace is cancelled in the job queue. + */ +interface ExecutionStopScope { + workspaceIds: readonly string[] + /** An account whose workflows are removed wherever they live, not only in `workspaceIds`. */ + workflowsCreatedBy?: string +} + +interface PlannedExecutionStop { + executionId: string + executionDeadlineAt: Date | null + workspaceId: string | null +} + +/** Read before a removal erases the rows it names; enqueue with {@link enqueueExecutionStops}. */ +export interface ExecutionStopPlan { + workspaceIds: string[] + executions: PlannedExecutionStop[] +} + +export interface ExecutionStopPayload { + /** Workspace whose queued and running jobs this event sweeps. */ + workspaceId?: string + executions: { executionId: string; executionDeadlineAt: string | null }[] +} + +function scopeCondition(executor: DbOrTx, scope: ExecutionStopScope): SQL | undefined { + const conditions: SQL[] = [] + if (scope.workspaceIds.length > 0) { + conditions.push(inArray(workflowExecutionLogs.workspaceId, [...scope.workspaceIds])) + } + if (scope.workflowsCreatedBy) { + conditions.push( + inArray( + workflowExecutionLogs.workflowId, + executor + .select({ id: workflow.id }) + .from(workflow) + .where(eq(workflow.userId, scope.workflowsCreatedBy)) + ) + ) + } + return conditions.length > 1 ? or(...conditions) : conditions[0] +} + +/** + * Reads the executions a removal must stop. Run it in the removal's transaction (or, where the + * 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( + executor: DbOrTx, + scope: ExecutionStopScope +): Promise { + const workspaceIds = [...new Set(scope.workspaceIds)] + const inScope = scopeCondition(executor, scope) + if (!inScope) return { workspaceIds, executions: [] } + + const running = await executor + .select({ + executionId: workflowExecutionLogs.executionId, + executionDeadlineAt: workflowExecutionLogs.executionDeadlineAt, + workspaceId: workflowExecutionLogs.workspaceId, + }) + .from(workflowExecutionLogs) + .where(and(eq(workflowExecutionLogs.status, 'running'), inScope)) + const resuming = await executor + .select({ + executionId: resumeQueue.newExecutionId, + executionDeadlineAt: workflowExecutionLogs.executionDeadlineAt, + workspaceId: workflowExecutionLogs.workspaceId, + }) + .from(resumeQueue) + .innerJoin( + workflowExecutionLogs, + eq(workflowExecutionLogs.executionId, resumeQueue.parentExecutionId) + ) + .where(and(eq(resumeQueue.status, 'claimed'), inScope)) + + const executions = new Map() + for (const row of [...running, ...resuming]) { + if (!executions.has(row.executionId)) executions.set(row.executionId, row) + } + return { workspaceIds, executions: [...executions.values()] } +} + +/** + * For a bare `DELETE FROM "user"`, which cascades every workspace the account owns (archived or + * not) and every workflow it created, wherever that workflow lives. + */ +export async function collectAccountRemovalStops( + executor: DbOrTx, + userId: string +): Promise { + const owned = await executor + .select({ id: workspace.id }) + .from(workspace) + .where(eq(workspace.ownerId, userId)) + return collectExecutionStops(executor, { + workspaceIds: owned.map((row) => row.id), + workflowsCreatedBy: userId, + }) +} + +function toPayloadExecutions( + executions: readonly PlannedExecutionStop[] +): ExecutionStopPayload['executions'] { + return executions.map(({ executionId, executionDeadlineAt }) => ({ + executionId, + executionDeadlineAt: executionDeadlineAt?.toISOString() ?? null, + })) +} + +/** One event per removed workspace carries its sweep; executions beyond one event's share spill over. */ +function buildPayloads(plan: ExecutionStopPlan): ExecutionStopPayload[] { + const removedWorkspaces = new Set(plan.workspaceIds) + const byWorkspace = new Map( + plan.workspaceIds.map((workspaceId) => [workspaceId, []]) + ) + const elsewhere: PlannedExecutionStop[] = [] + for (const execution of plan.executions) { + if (execution.workspaceId && removedWorkspaces.has(execution.workspaceId)) { + byWorkspace.get(execution.workspaceId)?.push(execution) + } else { + elsewhere.push(execution) + } + } + + const payloads: ExecutionStopPayload[] = [] + for (const [workspaceId, executions] of byWorkspace) { + const [first = [], ...rest] = chunkArray(executions, EXECUTIONS_PER_EVENT) + payloads.push({ workspaceId, executions: toPayloadExecutions(first) }) + for (const chunk of rest) payloads.push({ executions: toPayloadExecutions(chunk) }) + } + for (const chunk of chunkArray(elsewhere, EXECUTIONS_PER_EVENT)) { + payloads.push({ executions: toPayloadExecutions(chunk) }) + } + return payloads +} + +/** + * Records the stops as outbox events. In the removal's transaction they commit or roll back with + * it; the outbox worker retries each one until it is delivered. + */ +export async function enqueueExecutionStops( + executor: Pick, + plan: ExecutionStopPlan +): Promise { + const eventIds: string[] = [] + for (const chunk of chunkArray(buildPayloads(plan), MAX_BULK_ENQUEUE_EVENTS)) { + eventIds.push(...(await enqueueOutboxEvents(executor, EXECUTION_REMOVAL_STOP_EVENT, chunk))) + } + return eventIds +} + +interface RecordedExecutionStops { + /** The outbox events; deliver them now with `deliverExecutionStops` where that module may load. */ + eventIds: string[] + executions: PlannedExecutionStop[] +} + +/** {@link collectExecutionStops} then {@link enqueueExecutionStops}, in the caller's transaction. */ +export async function recordExecutionStops( + executor: DbOrTx, + scope: ExecutionStopScope +): Promise { + const plan = await collectExecutionStops(executor, scope) + return { eventIds: await enqueueExecutionStops(executor, plan), executions: plan.executions } +} + +/** + * Sets the cancellation flag each execution's engine watches, and aborts the ones this process + * runs, without the queue sweep or the retry the outbox event adds. Callers whose module graph + * cannot load the delivery stop in-flight runs with it at once and leave the rest to the outbox + * worker. Returns how many flags Redis refused; never throws. + * + * Without Redis at all the flag is process-local by design (`markExecutionCancelled`) and a + * retry from any process could reach no further, so that outcome does not count as refused. + */ +export async function signalExecutionStops( + executions: readonly Pick[], + signal?: AbortSignal +): Promise { + let refused = 0 + await mapWithConcurrency(executions, SIGNAL_CONCURRENCY, async (execution) => { + if (signal?.aborted) { + refused += 1 + return + } + try { + const result = await markExecutionCancelled(execution.executionId, { + executionDeadlineAt: execution.executionDeadlineAt, + }) + if (!result.durablyRecorded && result.reason !== 'redis_unavailable') refused += 1 + } catch (error) { + refused += 1 + logger.warn('Failed to signal an execution being stopped', { + executionId: execution.executionId, + error: describeError(error), + }) + } + try { + abortManualExecution(execution.executionId) + } catch (error) { + logger.warn('Failed to abort a local execution being stopped', { + executionId: execution.executionId, + error: describeError(error), + }) + } + }) + return refused +} diff --git a/apps/sim/lib/execution/removal-stops.integration.ts b/apps/sim/lib/execution/removal-stops.integration.ts new file mode 100644 index 00000000000..f96fe5e448b --- /dev/null +++ b/apps/sim/lib/execution/removal-stops.integration.ts @@ -0,0 +1,335 @@ +/** + * Removing an account, banning it, or deleting its user row directly stops the workflow runs the + * removal orphans, against real PostgreSQL and Redis through the production removal paths. A + * stopped execution is one whose cancellation flag, the one its engine watches, is set in Redis; + * a stopped queued job is a cancelled `async_jobs` row (the database queue these suites run on). + */ +import { authMock } from '@sim/testing/mocks/auth.mock' +import { afterAll, describe, expect, it, vi } from 'vitest' + +const { redisUrl, inheritedRedisUrl } = await vi.hoisted(async () => { + const { readTestRedisUrl } = await import('@sim/db/testing/test-infrastructure') + const url = readTestRedisUrl() + const inheritedRedisUrl = process.env.REDIS_URL + if (url) process.env.REDIS_URL = url + return { redisUrl: url, inheritedRedisUrl } +}) + +vi.mock('@/lib/auth', () => authMock) + +import { db } from '@sim/db' +import { + asyncJobs, + outboxEvent, + pausedExecutions, + permissions, + resumeQueue, + user, + workflow, + workflowExecutionLogs, + workflowExecutionSnapshots, + workspace, +} from '@sim/db/schema' +import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' +import { generateId } from '@sim/utils/id' +import { and, eq, inArray, sql } from 'drizzle-orm' +import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis' +import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' +import { + collectAccountRemovalStops, + enqueueExecutionStops, + signalExecutionStops, +} from '@/lib/execution/removal-stop-records' +import { deliverExecutionStops } from '@/lib/execution/removal-stops' +import { deleteUserAccount } from '@/lib/users/account-deletion' +import { disableUserResources } from '@/lib/workflows/lifecycle' + +const userIds: string[] = [] +const workspaceIds: string[] = [] +const snapshotIds: string[] = [] +const executionIds: string[] = [] +const jobIds: string[] = [] + +async function seedUser() { + const id = generateId() + userIds.push(id) + const now = new Date() + await db.insert(user).values({ + id, + name: 'Execution removal fixture', + email: `${id}@execution-removal.test`, + emailVerified: true, + createdAt: now, + updatedAt: now, + }) + return id +} + +async function seedWorkspace(ownerId: string, writers: string[] = []) { + const id = generateId() + workspaceIds.push(id) + await insertWorkspaceFixture(db, { + id, + name: 'Execution removal fixture', + ownerId, + billedAccountUserId: ownerId, + }) + await db.insert(permissions).values( + [ownerId, ...writers].map((userId) => ({ + id: generateId(), + userId, + entityType: 'workspace', + entityId: id, + permissionType: userId === ownerId ? ('admin' as const) : ('write' as const), + })) + ) + return id +} + +async function seedWorkflow(userId: string, workspaceId: string) { + const id = generateId() + const now = new Date() + await db.insert(workflow).values({ + id, + userId, + workspaceId, + name: 'Execution removal fixture', + lastSynced: now, + createdAt: now, + updatedAt: now, + }) + return id +} + +/** A run whose engine is still executing: its log row says `running`. */ +async function seedExecution( + workspaceId: string, + workflowId: string, + status: 'running' | 'pending' = 'running' +) { + const snapshotId = generateId() + snapshotIds.push(snapshotId) + await db.insert(workflowExecutionSnapshots).values({ + id: snapshotId, + workflowId, + stateHash: snapshotId, + stateData: {}, + }) + const executionId = generateId() + executionIds.push(executionId) + await db.insert(workflowExecutionLogs).values({ + id: generateId(), + workflowId, + workspaceId, + executionId, + stateSnapshotId: snapshotId, + level: 'info', + status, + trigger: 'manual', + startedAt: new Date(), + executionData: {}, + }) + return executionId +} + +/** 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') + const pausedExecutionId = generateId() + await db.insert(pausedExecutions).values({ + id: pausedExecutionId, + workflowId, + executionId: parentExecutionId, + executionSnapshot: {}, + pausePoints: {}, + totalPauseCount: 1, + status: 'paused', + }) + const resumeExecutionId = generateId() + executionIds.push(resumeExecutionId) + await db.insert(resumeQueue).values({ + id: generateId(), + pausedExecutionId, + parentExecutionId, + newExecutionId: resumeExecutionId, + contextId: generateId(), + status: 'claimed', + }) + return resumeExecutionId +} + +/** A job the queue has not started yet. */ +async function seedQueuedJob(workspaceId: string) { + const id = `run_${generateId()}` + jobIds.push(id) + await db.insert(asyncJobs).values({ + id, + type: 'webhook-execution', + payload: {}, + status: 'pending', + metadata: { workspaceId }, + }) + return id +} + +async function isCancelled(executionId: string) { + return (await getRedisClient()?.exists(`execution:cancel:${executionId}`)) === 1 +} + +async function jobStatus(jobId: string) { + const [row] = await db + .select({ status: asyncJobs.status }) + .from(asyncJobs) + .where(eq(asyncJobs.id, jobId)) + return row?.status +} + +/** Every stop event naming one of this test's executions or workspaces. */ +async function stopEvents(ids: string[]) { + const rows = await db + .select({ id: outboxEvent.id, status: outboxEvent.status, payload: outboxEvent.payload }) + .from(outboxEvent) + .where(eq(outboxEvent.eventType, EXECUTION_REMOVAL_STOP_EVENT)) + return rows.filter((row) => ids.some((id) => JSON.stringify(row.payload).includes(id))) +} + +afterAll(async () => { + if (inheritedRedisUrl === undefined) Reflect.deleteProperty(process.env, 'REDIS_URL') + else process.env.REDIS_URL = inheritedRedisUrl +}) + +describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan', () => { + afterAll(async () => { + const redis = getRedisClient() + if (redis && executionIds.length > 0) { + await redis.del(...executionIds.map((id) => `execution:cancel:${id}`)) + } + const scoped = [...executionIds, ...workspaceIds] + for (const id of scoped) { + await db + .delete(outboxEvent) + .where( + and( + eq(outboxEvent.eventType, EXECUTION_REMOVAL_STOP_EVENT), + sql`${outboxEvent.payload}::text like ${`%${id}%`}` + ) + ) + } + if (jobIds.length) await db.delete(asyncJobs).where(inArray(asyncJobs.id, jobIds)) + if (executionIds.length) { + await db + .delete(workflowExecutionLogs) + .where(inArray(workflowExecutionLogs.executionId, executionIds)) + } + if (snapshotIds.length) { + await db + .delete(workflowExecutionSnapshots) + .where(inArray(workflowExecutionSnapshots.id, snapshotIds)) + } + if (workspaceIds.length) await deleteWorkspaceFixture(db, inArray(workspace.id, workspaceIds)) + if (userIds.length) { + await db.delete(permissions).where(inArray(permissions.userId, userIds)) + await db.delete(user).where(inArray(user.id, userIds)) + } + await closeRedisConnection() + }) + + it('stops the running, resuming and queued runs of an erased account’s workspaces, and no one else’s', async () => { + const departingId = await seedUser() + const bystanderId = await seedUser() + const doomedWorkspaceId = await seedWorkspace(departingId) + const doomedWorkflowId = await seedWorkflow(departingId, doomedWorkspaceId) + const running = await seedExecution(doomedWorkspaceId, doomedWorkflowId) + const resuming = await seedActiveResume(doomedWorkspaceId, doomedWorkflowId) + const queued = await seedQueuedJob(doomedWorkspaceId) + const keptWorkspaceId = await seedWorkspace(bystanderId) + const keptRunning = await seedExecution( + keptWorkspaceId, + await seedWorkflow(bystanderId, keptWorkspaceId) + ) + const keptQueued = await seedQueuedJob(keptWorkspaceId) + + await deleteUserAccount(departingId) + + expect(await isCancelled(running)).toBe(true) + expect(await isCancelled(resuming)).toBe(true) + expect(await jobStatus(queued)).toBe('cancelled') + expect((await stopEvents([doomedWorkspaceId])).map((event) => event.status)).toEqual([ + 'completed', + ]) + expect(await isCancelled(keptRunning)).toBe(false) + expect(await jobStatus(keptQueued)).toBe('pending') + }) + + it('stops a banned owner’s running runs at once and leaves its queued runs to the outbox worker', async () => { + const bannedId = await seedUser() + const bystanderId = await seedUser() + const ownedWorkspaceId = await seedWorkspace(bannedId) + const running = await seedExecution( + ownedWorkspaceId, + await seedWorkflow(bannedId, ownedWorkspaceId) + ) + const queued = await seedQueuedJob(ownedWorkspaceId) + const keptWorkspaceId = await seedWorkspace(bystanderId, [bannedId]) + const keptRunning = await seedExecution( + keptWorkspaceId, + await seedWorkflow(bystanderId, keptWorkspaceId) + ) + const keptQueued = await seedQueuedJob(keptWorkspaceId) + + await disableUserResources(bannedId) + + const [archived] = await db + .select({ archivedAt: workspace.archivedAt }) + .from(workspace) + .where(eq(workspace.id, ownedWorkspaceId)) + expect(archived.archivedAt).not.toBeNull() + expect(await isCancelled(running)).toBe(true) + expect(await isCancelled(keptRunning)).toBe(false) + const recorded = await stopEvents([ownedWorkspaceId]) + expect(recorded.map((event) => event.status)).toEqual(['pending']) + + await deliverExecutionStops( + recorded.map((event) => event.id), + { test: 'outbox worker' } + ) + + expect(await jobStatus(queued)).toBe('cancelled') + expect(await jobStatus(keptQueued)).toBe('pending') + }) + + /** + * Mirrors the Better Auth user delete hooks around `/admin/remove-user`: read before the bare + * delete, enqueued and signalled once it has committed. The delivery stands in for the outbox + * worker that sweeps the queue. + */ + it('stops the runs of workflows a directly deleted user created in someone else’s workspace', async () => { + const removedId = await seedUser() + const bystanderId = await seedUser() + const sharedWorkspaceId = await seedWorkspace(bystanderId, [removedId]) + const removedUsersRun = await seedExecution( + sharedWorkspaceId, + await seedWorkflow(removedId, sharedWorkspaceId) + ) + const bystandersRun = await seedExecution( + sharedWorkspaceId, + await seedWorkflow(bystanderId, sharedWorkspaceId) + ) + const bystandersQueued = await seedQueuedJob(sharedWorkspaceId) + + const plan = await collectAccountRemovalStops(db, removedId) + await db.delete(user).where(eq(user.id, removedId)) + const eventIds = await enqueueExecutionStops(db, plan) + await signalExecutionStops(plan.executions) + + expect(await isCancelled(removedUsersRun)).toBe(true) + expect(await isCancelled(bystandersRun)).toBe(false) + + await deliverExecutionStops(eventIds, { test: 'outbox worker' }) + + expect(await jobStatus(bystandersQueued)).toBe('pending') + expect((await stopEvents([removedUsersRun])).map((event) => event.status)).toEqual([ + 'completed', + ]) + }) +}) diff --git a/apps/sim/lib/execution/removal-stops.test.ts b/apps/sim/lib/execution/removal-stops.test.ts new file mode 100644 index 00000000000..0f8e34512c1 --- /dev/null +++ b/apps/sim/lib/execution/removal-stops.test.ts @@ -0,0 +1,89 @@ +/** + * The outbox outcome of a removal stop decides whether a run keeps going: an outcome that + * completes too early drops the stop, one that never completes dead-letters it. Failure modes: + * - 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 + * - 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 + */ +import { asyncJobsMock, asyncJobsMockFns } from '@sim/testing/mocks/async-jobs.mock' +import { redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +vi.mock('@/lib/core/async-jobs', () => asyncJobsMock) + +import type { OutboxEventContext } from '@/lib/core/outbox/service' +import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' +import { executionRemovalStopOutboxHandlers } from '@/lib/execution/removal-stops' + +const { mockJobQueue } = asyncJobsMockFns +const stopRemovedExecutions = executionRemovalStopOutboxHandlers[EXECUTION_REMOVAL_STOP_EVENT] + +function context(): OutboxEventContext { + return { + eventId: 'event-1', + eventType: EXECUTION_REMOVAL_STOP_EVENT, + attempts: 0, + maxAttempts: 10, + signal: new AbortController().signal, + checkpointPayload: vi.fn(), + } +} + +const payload = { + workspaceId: 'workspace-1', + executions: [{ executionId: 'execution-1', executionDeadlineAt: null }], +} + +describe('execution removal stop outcome', () => { + beforeEach(() => { + redisConfigMockFns.mockGetRedisClient.mockReturnValue(null) + mockJobQueue.cancelByWorkspace.mockReset().mockResolvedValue({ + cancelledJobs: 0, + complete: true, + }) + }) + + it('completes when there is no Redis to record the cancellation in', async () => { + await expect(stopRemovedExecutions(payload, context())).resolves.toBeUndefined() + }) + + it('retries when Redis refuses the cancellation, after still sweeping the queue', async () => { + redisConfigMockFns.mockGetRedisClient.mockReturnValue({ + eval: vi.fn().mockRejectedValue(new Error('READONLY')), + publish: vi.fn().mockResolvedValue(1), + }) + + await expect(stopRemovedExecutions(payload, context())).rejects.toThrow('not recorded') + expect(mockJobQueue.cancelByWorkspace).toHaveBeenCalledWith('workspace-1', expect.anything()) + }) + + it('retries when the queue sweep fails', async () => { + mockJobQueue.cancelByWorkspace.mockRejectedValue(new Error('provider unavailable')) + + await expect(stopRemovedExecutions(payload, context())).rejects.toThrow('provider unavailable') + }) + + it('continues without spending an attempt when the sweep stops at its cap after progress', async () => { + mockJobQueue.cancelByWorkspace.mockResolvedValue({ cancelledJobs: 1000, complete: false }) + + await expect(stopRemovedExecutions(payload, context())).resolves.toMatchObject({ + outcome: 'deferred', + consumeAttempt: false, + }) + }) + + it('retries when the sweep stops without cancelling anything', async () => { + mockJobQueue.cancelByWorkspace.mockResolvedValue({ cancelledJobs: 0, complete: false }) + + await expect(stopRemovedExecutions(payload, context())).rejects.toThrow('no progress') + }) + + it('sweeps no queue for executions outside a removed workspace', async () => { + await expect( + stopRemovedExecutions({ executions: payload.executions }, context()) + ).resolves.toBeUndefined() + expect(mockJobQueue.cancelByWorkspace).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/execution/removal-stops.ts b/apps/sim/lib/execution/removal-stops.ts new file mode 100644 index 00000000000..37bf3671160 --- /dev/null +++ b/apps/sim/lib/execution/removal-stops.ts @@ -0,0 +1,131 @@ +import { createLogger } from '@sim/logger' +import { describeError } from '@sim/utils/errors' +import { isRecordLike } from '@sim/utils/object' +import { getInlineJobQueue, getJobQueue } from '@/lib/core/async-jobs' +import { + continueOutboxHandler, + type OutboxHandler, + type OutboxHandlerRegistry, + processOutboxEventById, + withOutboxHandlerTimeout, +} from '@/lib/core/outbox/service' +import { mapWithConcurrency } from '@/lib/core/utils/concurrency' +import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' +import { + type ExecutionStopPayload, + signalExecutionStops, +} from '@/lib/execution/removal-stop-records' + +const logger = createLogger('ExecutionRemovalStops') + +/** Queue runs one handler attempt cancels before handing the rest to its next run. */ +const SWEEP_MAX_RUNS_PER_ATTEMPT = 1_000 +const HANDLER_TIMEOUT_MS = 5 * 60 * 1000 +const DELIVERY_CONCURRENCY = 4 + +function parsePayload(payload: unknown): ExecutionStopPayload { + if (!isRecordLike(payload) || !Array.isArray(payload.executions)) { + throw new Error('Execution removal stop payload is malformed') + } + let workspaceId: string | undefined + if (payload.workspaceId !== undefined) { + if (typeof payload.workspaceId !== 'string' || payload.workspaceId.length === 0) { + throw new Error('Execution removal stop payload has an invalid workspaceId') + } + workspaceId = payload.workspaceId + } + const executions = payload.executions.map((entry: unknown) => { + if ( + !isRecordLike(entry) || + typeof entry.executionId !== 'string' || + entry.executionId.length === 0 || + (entry.executionDeadlineAt !== null && typeof entry.executionDeadlineAt !== 'string') + ) { + throw new Error('Execution removal stop payload has an invalid execution') + } + return { executionId: entry.executionId, executionDeadlineAt: entry.executionDeadlineAt } + }) + return { ...(workspaceId ? { workspaceId } : {}), executions } +} + +/** + * 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. + */ +async function sweepWorkspaceJobs(workspaceId: string, signal: AbortSignal) { + const queues = new Set([await getJobQueue(), await getInlineJobQueue()]) + const results = await Promise.allSettled( + [...queues].map((queue) => + queue.cancelByWorkspace(workspaceId, { maxRuns: SWEEP_MAX_RUNS_PER_ATTEMPT, signal }) + ) + ) + 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 + } + return { cancelledJobs, complete } +} + +/** + * Idempotent, so a retry repeats the whole event: a repeated signal and a repeated cancel of a + * run that already ended are both no-ops. + */ +const stopRemovedExecutions: OutboxHandler = withOutboxHandlerTimeout( + async (rawPayload, context) => { + const payload = parsePayload(rawPayload) + const unrecorded = await signalExecutionStops( + payload.executions.map(({ executionId, executionDeadlineAt }) => ({ + executionId, + executionDeadlineAt: executionDeadlineAt ? new Date(executionDeadlineAt) : null, + })), + context.signal + ) + const sweep = payload.workspaceId + ? await sweepWorkspaceJobs(payload.workspaceId, context.signal) + : { cancelledJobs: 0, complete: true } + if (unrecorded > 0) { + throw new Error(`${unrecorded} execution cancellations were not recorded`) + } + if (!sweep.complete) { + if (sweep.cancelledJobs === 0) throw new Error('Workspace job sweep made no progress') + return continueOutboxHandler('Workspace job sweep reached its per-attempt cap') + } + }, + HANDLER_TIMEOUT_MS +) + +export const executionRemovalStopOutboxHandlers = { + [EXECUTION_REMOVAL_STOP_EVENT]: stopRemovedExecutions, +} satisfies OutboxHandlerRegistry + +/** + * Delivers committed stops now, a few at a time, and never throws: a stop that cannot be + * delivered stays in the outbox for its worker to retry, so a removal that already committed is + * never reported as failed. + */ +export async function deliverExecutionStops( + eventIds: readonly string[], + logContext: Record +): Promise { + await mapWithConcurrency(eventIds, DELIVERY_CONCURRENCY, async (eventId) => { + try { + const result = await processOutboxEventById(eventId, executionRemovalStopOutboxHandlers) + if (result !== 'completed') { + logger.warn('Execution removal stop left to the outbox worker', { + ...logContext, + eventId, + result, + }) + } + } catch (error) { + logger.warn('Execution removal stop failed inline; the outbox worker retries it', { + ...logContext, + eventId, + error: describeError(error), + }) + } + }) +} diff --git a/apps/sim/lib/users/account-deletion.ts b/apps/sim/lib/users/account-deletion.ts index ca0c883f95f..9d7c3326061 100644 --- a/apps/sim/lib/users/account-deletion.ts +++ b/apps/sim/lib/users/account-deletion.ts @@ -39,6 +39,8 @@ import { } from '@/lib/billing/organizations/resource-handoff' import { OrchestrationError } from '@/lib/core/orchestration/types' import type { DbOrTx } from '@/lib/db/types' +import { recordExecutionStops } from '@/lib/execution/removal-stop-records' +import { deliverExecutionStops } from '@/lib/execution/removal-stops' import { type StoppedRun, stopLiveRunsTx } from '@/lib/mothership/async-runs/repository' import { enqueueMemoryCleanup } from '@/lib/mothership/memories/cleanup-outbox' import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery' @@ -782,6 +784,7 @@ export async function deleteUserAccount(userId: string): Promise 0) { + /** + * Workflow runs in the doomed workspaces lose their log rows to the cascade below, after + * which no cancel path can find them, and they would keep spending hosted keys for an + * account that no longer exists. The workspace rows are locked, so no new run can log + * here before commit; the stops commit with the erasure and are delivered after it. + */ + const executionStops = await recordExecutionStops(tx, { workspaceIds: doomedWorkspaceIds }) + executionStopEventIds = executionStops.eventIds /** * Re-checked here rather than trusted from the plan: a workspace that * gained a member since the preview is no longer private, and deleting it @@ -1019,6 +1030,8 @@ export async function deleteUserAccount(userId: string): Promise { const requestId = generateRequestId() @@ -310,14 +317,41 @@ export async function disableUserResources(userId: string): Promise { .where(and(eq(workspace.ownerId, userId), isNull(workspace.archivedAt))) const failures: { workspaceId: string; error: unknown }[] = [] + const archivedWorkspaceIds: string[] = [] // Sequential: environments of one Project contend on its exclusive lock. for (const row of ownedWorkspaces) { try { - await archiveWorkspace(row.id, { requestId, expectedOwnerId: userId }) + const result = await archiveWorkspace(row.id, { requestId, expectedOwnerId: userId }) + if (result.archived) archivedWorkspaceIds.push(row.id) } catch (error) { failures.push({ workspaceId: row.id, error }) } } + + /** + * After the archives commit, which refuse new runs, so every run admitted before them is found. + * In-flight runs are signalled here; the recorded stop's queue sweep and retries run in the + * outbox worker, since this module sits in page graphs that must not load that delivery. + */ + let stopError: unknown + if (archivedWorkspaceIds.length > 0) { + try { + const { recordExecutionStops, signalExecutionStops } = await import( + '@/lib/execution/removal-stop-records' + ) + const { executions } = await db.transaction((tx) => + recordExecutionStops(tx, { workspaceIds: archivedWorkspaceIds }) + ) + await signalExecutionStops(executions) + } catch (error) { + stopError = error + logger.error(`[${requestId}] Failed to stop runs for banned user ${userId}`, { + workspaceIds: archivedWorkspaceIds, + error: getErrorMessage(error), + }) + } + } + if (failures.length > 0) { logger.error(`[${requestId}] Failed to archive workspaces for banned user ${userId}`, { failures: failures.map(({ workspaceId, error }) => ({ @@ -326,13 +360,17 @@ export async function disableUserResources(userId: string): Promise { })), }) throw new AggregateError( - failures.map((failure) => failure.error), + [ + ...failures.map((failure) => failure.error), + ...(stopError !== undefined ? [stopError] : []), + ], `Failed to archive ${failures.length} of ${ownedWorkspaces.length} workspaces for banned user ${userId}` ) } + if (stopError !== undefined) throw stopError logger.info( - `[${requestId}] Disabled resources for user ${userId}: archived ${ownedWorkspaces.length} workspaces, deleted API keys` + `[${requestId}] Disabled resources for user ${userId}: archived ${ownedWorkspaces.length} workspaces, deleted API keys, stopped their runs` ) } diff --git a/packages/testing/src/mocks/async-jobs.mock.ts b/packages/testing/src/mocks/async-jobs.mock.ts index d7ea9872872..863662b8da4 100644 --- a/packages/testing/src/mocks/async-jobs.mock.ts +++ b/packages/testing/src/mocks/async-jobs.mock.ts @@ -37,6 +37,7 @@ export const mockJobQueue = { cancelJob: vi.fn(), cancelByExecution: vi.fn(), cancelByKey: vi.fn(), + cancelByWorkspace: vi.fn(), } /** From 040ade01d8fd80cd58cc6c1467658fc691924d9d Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 15:11:28 -0700 Subject: [PATCH 2/9] fix(execution): keep removal stops off the deletion and ban critical paths - Account deletion no longer awaits the full stop delivery before purging storage. A slow Trigger.dev listing could hold the response for minutes and widen the window in which a restart orphans erased objects. It now signals in-flight runs fire-and-forget right after commit, ahead of every other post-commit step, under a 10-second deadline; the committed outbox stops repeat the signal and sweep the job queue in the outbox worker. - A failure to record a ban's stops is logged instead of thrown. The ban and archives have committed, so a 500 helped no one and a retry finds no unarchived workspace to stop. - Claimed resumes are found from the scope's running and paused log rows and joined through resume_queue's parent index, not by scanning resume_queue for claimed rows inside the deletion transaction. - Tests pin the database queue's job-type filter, so a workspace sweep never cancels a non-workflow job such as log cleanup. --- .../core/async-jobs/backends/database.test.ts | 28 ++++++++++++++- .../sim/lib/execution/removal-stop-records.ts | 36 +++++++++++-------- .../execution/removal-stops.integration.ts | 35 +++++++++++------- apps/sim/lib/execution/removal-stops.ts | 36 ------------------- apps/sim/lib/users/account-deletion.ts | 24 ++++++++----- apps/sim/lib/workflows/lifecycle.ts | 11 ++---- 6 files changed, 90 insertions(+), 80 deletions(-) diff --git a/apps/sim/lib/core/async-jobs/backends/database.test.ts b/apps/sim/lib/core/async-jobs/backends/database.test.ts index 0046926f940..c5880c464d2 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.test.ts +++ b/apps/sim/lib/core/async-jobs/backends/database.test.ts @@ -1,4 +1,9 @@ -import { dbChainMockFns, resetDbChainMock } from '@sim/testing/mocks/database.mock' +import { asyncJobs } from '@sim/db' +import { + dbChainMockFns, + flattenMockConditions, + resetDbChainMock, +} from '@sim/testing/mocks/database.mock' import { sleep } from '@sim/utils/helpers' import { beforeEach, describe, expect, it, vi } from 'vitest' @@ -503,3 +508,24 @@ 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') + + 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', + ]) + }) +}) diff --git a/apps/sim/lib/execution/removal-stop-records.ts b/apps/sim/lib/execution/removal-stop-records.ts index 98273cb716e..8be52a9bb25 100644 --- a/apps/sim/lib/execution/removal-stop-records.ts +++ b/apps/sim/lib/execution/removal-stop-records.ts @@ -16,6 +16,8 @@ const logger = createLogger('ExecutionRemovalStopRecords') /** Executions one event signals; bounds the payload of a workspace with many live runs. */ const EXECUTIONS_PER_EVENT = 500 const SIGNAL_CONCURRENCY = 10 +/** Bounds an inline signal pass; whatever it leaves, the outbox stop signals again. */ +const SIGNAL_DEADLINE_MS = 10_000 /** * What is being removed. Runs of a removed workspace or workflow lose the rows the ordinary @@ -87,18 +89,22 @@ async function collectExecutionStops( }) .from(workflowExecutionLogs) .where(and(eq(workflowExecutionLogs.status, 'running'), inScope)) + /** Driven from the scope's log rows, so `resume_queue` is read through its parent index. */ const resuming = await executor .select({ executionId: resumeQueue.newExecutionId, executionDeadlineAt: workflowExecutionLogs.executionDeadlineAt, workspaceId: workflowExecutionLogs.workspaceId, }) - .from(resumeQueue) + .from(workflowExecutionLogs) .innerJoin( - workflowExecutionLogs, - eq(workflowExecutionLogs.executionId, resumeQueue.parentExecutionId) + resumeQueue, + and( + eq(resumeQueue.parentExecutionId, workflowExecutionLogs.executionId), + eq(resumeQueue.status, 'claimed') + ) ) - .where(and(eq(resumeQueue.status, 'claimed'), inScope)) + .where(and(inArray(workflowExecutionLogs.status, ['running', 'pending']), inScope)) const executions = new Map() for (const row of [...running, ...resuming]) { @@ -176,26 +182,25 @@ export async function enqueueExecutionStops( return eventIds } -interface RecordedExecutionStops { - /** The outbox events; deliver them now with `deliverExecutionStops` where that module may load. */ - eventIds: string[] - executions: PlannedExecutionStop[] -} - -/** {@link collectExecutionStops} then {@link enqueueExecutionStops}, in the caller's transaction. */ +/** + * {@link collectExecutionStops} then {@link enqueueExecutionStops}, in the caller's transaction. + * Returns the executions, to signal at once with {@link signalExecutionStops} after commit. + */ export async function recordExecutionStops( executor: DbOrTx, scope: ExecutionStopScope -): Promise { +): Promise { const plan = await collectExecutionStops(executor, scope) - return { eventIds: await enqueueExecutionStops(executor, plan), executions: plan.executions } + await enqueueExecutionStops(executor, plan) + return plan.executions } /** * Sets the cancellation flag each execution's engine watches, and aborts the ones this process * runs, without the queue sweep or the retry the outbox event adds. Callers whose module graph * cannot load the delivery stop in-flight runs with it at once and leave the rest to the outbox - * worker. Returns how many flags Redis refused; never throws. + * worker. Returns how many flags Redis refused or the deadline (`signal`, else + * {@link SIGNAL_DEADLINE_MS}) left unsent; never throws. * * Without Redis at all the flag is process-local by design (`markExecutionCancelled`) and a * retry from any process could reach no further, so that outcome does not count as refused. @@ -204,9 +209,10 @@ export async function signalExecutionStops( executions: readonly Pick[], signal?: AbortSignal ): Promise { + const deadline = signal ?? AbortSignal.timeout(SIGNAL_DEADLINE_MS) let refused = 0 await mapWithConcurrency(executions, SIGNAL_CONCURRENCY, async (execution) => { - if (signal?.aborted) { + if (deadline.aborted) { refused += 1 return } diff --git a/apps/sim/lib/execution/removal-stops.integration.ts b/apps/sim/lib/execution/removal-stops.integration.ts index f96fe5e448b..8c819f95820 100644 --- a/apps/sim/lib/execution/removal-stops.integration.ts +++ b/apps/sim/lib/execution/removal-stops.integration.ts @@ -34,13 +34,14 @@ import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/ import { generateId } from '@sim/utils/id' import { and, eq, inArray, sql } from 'drizzle-orm' import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis' +import { processOutboxEventById } from '@/lib/core/outbox/service' import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' import { collectAccountRemovalStops, enqueueExecutionStops, signalExecutionStops, } from '@/lib/execution/removal-stop-records' -import { deliverExecutionStops } from '@/lib/execution/removal-stops' +import { executionRemovalStopOutboxHandlers } from '@/lib/execution/removal-stops' import { deleteUserAccount } from '@/lib/users/account-deletion' import { disableUserResources } from '@/lib/workflows/lifecycle' @@ -159,12 +160,12 @@ async function seedActiveResume(workspaceId: string, workflowId: string) { } /** A job the queue has not started yet. */ -async function seedQueuedJob(workspaceId: string) { +async function seedQueuedJob(workspaceId: string, type = 'webhook-execution') { const id = `run_${generateId()}` jobIds.push(id) await db.insert(asyncJobs).values({ id, - type: 'webhook-execution', + type, payload: {}, status: 'pending', metadata: { workspaceId }, @@ -193,6 +194,13 @@ async function stopEvents(ids: string[]) { return rows.filter((row) => ids.some((id) => JSON.stringify(row.payload).includes(id))) } +/** Processes the stop events the way the outbox worker does. */ +async function runOutboxWorker(eventIds: string[]) { + for (const eventId of eventIds) { + await processOutboxEventById(eventId, executionRemovalStopOutboxHandlers) + } +} + afterAll(async () => { if (inheritedRedisUrl === undefined) Reflect.deleteProperty(process.env, 'REDIS_URL') else process.env.REDIS_URL = inheritedRedisUrl @@ -242,6 +250,7 @@ describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan', const running = await seedExecution(doomedWorkspaceId, doomedWorkflowId) const resuming = await seedActiveResume(doomedWorkspaceId, doomedWorkflowId) const queued = await seedQueuedJob(doomedWorkspaceId) + const unrelatedJob = await seedQueuedJob(doomedWorkspaceId, 'cleanup-logs') const keptWorkspaceId = await seedWorkspace(bystanderId) const keptRunning = await seedExecution( keptWorkspaceId, @@ -251,9 +260,15 @@ describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan', await deleteUserAccount(departingId) - expect(await isCancelled(running)).toBe(true) - expect(await isCancelled(resuming)).toBe(true) + await expect.poll(() => isCancelled(running)).toBe(true) + await expect.poll(() => isCancelled(resuming)).toBe(true) + const recorded = await stopEvents([doomedWorkspaceId]) + expect(recorded.map((event) => event.status)).toEqual(['pending']) + + await runOutboxWorker(recorded.map((event) => event.id)) + expect(await jobStatus(queued)).toBe('cancelled') + expect(await jobStatus(unrelatedJob)).toBe('pending') expect((await stopEvents([doomedWorkspaceId])).map((event) => event.status)).toEqual([ 'completed', ]) @@ -289,10 +304,7 @@ describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan', const recorded = await stopEvents([ownedWorkspaceId]) expect(recorded.map((event) => event.status)).toEqual(['pending']) - await deliverExecutionStops( - recorded.map((event) => event.id), - { test: 'outbox worker' } - ) + await runOutboxWorker(recorded.map((event) => event.id)) expect(await jobStatus(queued)).toBe('cancelled') expect(await jobStatus(keptQueued)).toBe('pending') @@ -300,8 +312,7 @@ describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan', /** * Mirrors the Better Auth user delete hooks around `/admin/remove-user`: read before the bare - * delete, enqueued and signalled once it has committed. The delivery stands in for the outbox - * worker that sweeps the queue. + * delete, enqueued and signalled once it has committed. */ it('stops the runs of workflows a directly deleted user created in someone else’s workspace', async () => { const removedId = await seedUser() @@ -325,7 +336,7 @@ describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan', expect(await isCancelled(removedUsersRun)).toBe(true) expect(await isCancelled(bystandersRun)).toBe(false) - await deliverExecutionStops(eventIds, { test: 'outbox worker' }) + await runOutboxWorker(eventIds) expect(await jobStatus(bystandersQueued)).toBe('pending') expect((await stopEvents([removedUsersRun])).map((event) => event.status)).toEqual([ diff --git a/apps/sim/lib/execution/removal-stops.ts b/apps/sim/lib/execution/removal-stops.ts index 37bf3671160..566053ce96a 100644 --- a/apps/sim/lib/execution/removal-stops.ts +++ b/apps/sim/lib/execution/removal-stops.ts @@ -1,27 +1,20 @@ -import { createLogger } from '@sim/logger' -import { describeError } from '@sim/utils/errors' import { isRecordLike } from '@sim/utils/object' import { getInlineJobQueue, getJobQueue } from '@/lib/core/async-jobs' import { continueOutboxHandler, type OutboxHandler, type OutboxHandlerRegistry, - processOutboxEventById, withOutboxHandlerTimeout, } from '@/lib/core/outbox/service' -import { mapWithConcurrency } from '@/lib/core/utils/concurrency' import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' import { type ExecutionStopPayload, signalExecutionStops, } from '@/lib/execution/removal-stop-records' -const logger = createLogger('ExecutionRemovalStops') - /** Queue runs one handler attempt cancels before handing the rest to its next run. */ const SWEEP_MAX_RUNS_PER_ATTEMPT = 1_000 const HANDLER_TIMEOUT_MS = 5 * 60 * 1000 -const DELIVERY_CONCURRENCY = 4 function parsePayload(payload: unknown): ExecutionStopPayload { if (!isRecordLike(payload) || !Array.isArray(payload.executions)) { @@ -100,32 +93,3 @@ const stopRemovedExecutions: OutboxHandler = withOutboxHandlerTimeout( export const executionRemovalStopOutboxHandlers = { [EXECUTION_REMOVAL_STOP_EVENT]: stopRemovedExecutions, } satisfies OutboxHandlerRegistry - -/** - * Delivers committed stops now, a few at a time, and never throws: a stop that cannot be - * delivered stays in the outbox for its worker to retry, so a removal that already committed is - * never reported as failed. - */ -export async function deliverExecutionStops( - eventIds: readonly string[], - logContext: Record -): Promise { - await mapWithConcurrency(eventIds, DELIVERY_CONCURRENCY, async (eventId) => { - try { - const result = await processOutboxEventById(eventId, executionRemovalStopOutboxHandlers) - if (result !== 'completed') { - logger.warn('Execution removal stop left to the outbox worker', { - ...logContext, - eventId, - result, - }) - } - } catch (error) { - logger.warn('Execution removal stop failed inline; the outbox worker retries it', { - ...logContext, - eventId, - error: describeError(error), - }) - } - }) -} diff --git a/apps/sim/lib/users/account-deletion.ts b/apps/sim/lib/users/account-deletion.ts index 9d7c3326061..2cc251e0b1b 100644 --- a/apps/sim/lib/users/account-deletion.ts +++ b/apps/sim/lib/users/account-deletion.ts @@ -39,8 +39,11 @@ import { } from '@/lib/billing/organizations/resource-handoff' import { OrchestrationError } from '@/lib/core/orchestration/types' import type { DbOrTx } from '@/lib/db/types' -import { recordExecutionStops } from '@/lib/execution/removal-stop-records' -import { deliverExecutionStops } from '@/lib/execution/removal-stops' +import { + type ExecutionStopPlan, + recordExecutionStops, + signalExecutionStops, +} from '@/lib/execution/removal-stop-records' import { type StoppedRun, stopLiveRunsTx } from '@/lib/mothership/async-runs/repository' import { enqueueMemoryCleanup } from '@/lib/mothership/memories/cleanup-outbox' import { deliverRunStops } from '@/lib/mothership/request/application/stop-delivery' @@ -784,7 +787,7 @@ export async function deleteUserAccount(userId: string): Promise { * In-flight runs are signalled here; the recorded stop's queue sweep and retries run in the * outbox worker, since this module sits in page graphs that must not load that delivery. */ - let stopError: unknown if (archivedWorkspaceIds.length > 0) { try { const { recordExecutionStops, signalExecutionStops } = await import( '@/lib/execution/removal-stop-records' ) - const { executions } = await db.transaction((tx) => + const executions = await db.transaction((tx) => recordExecutionStops(tx, { workspaceIds: archivedWorkspaceIds }) ) await signalExecutionStops(executions) } catch (error) { - stopError = error + /** Not thrown: the ban and archives committed, and a retry finds no unarchived workspace. */ logger.error(`[${requestId}] Failed to stop runs for banned user ${userId}`, { workspaceIds: archivedWorkspaceIds, error: getErrorMessage(error), @@ -360,14 +359,10 @@ export async function disableUserResources(userId: string): Promise { })), }) throw new AggregateError( - [ - ...failures.map((failure) => failure.error), - ...(stopError !== undefined ? [stopError] : []), - ], + failures.map((failure) => failure.error), `Failed to archive ${failures.length} of ${ownedWorkspaces.length} workspaces for banned user ${userId}` ) } - if (stopError !== undefined) throw stopError logger.info( `[${requestId}] Disabled resources for user ${userId}: archived ${ownedWorkspaces.length} workspaces, deleted API keys, stopped their runs` From b65ee91aae6e6e9c67bf9daef02a522c0c15a6b5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 15:22:12 -0700 Subject: [PATCH 3/9] test(execution): give each removal-stop fixture workflow a unique name --- apps/sim/lib/execution/removal-stops.integration.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/sim/lib/execution/removal-stops.integration.ts b/apps/sim/lib/execution/removal-stops.integration.ts index 8c819f95820..9c6fe26c9a7 100644 --- a/apps/sim/lib/execution/removal-stops.integration.ts +++ b/apps/sim/lib/execution/removal-stops.integration.ts @@ -94,7 +94,7 @@ async function seedWorkflow(userId: string, workspaceId: string) { id, userId, workspaceId, - name: 'Execution removal fixture', + name: `Execution removal fixture ${id}`, lastSynced: now, createdAt: now, updatedAt: now, From 86faeabfea94dc78806f854896d676a7d0e58c74 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 15:35:14 -0700 Subject: [PATCH 4/9] 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. --- apps/sim/lib/auth/admin-removal-stops.test.ts | 59 +++++++++++++++ apps/sim/lib/auth/admin-removal-stops.ts | 55 ++++++++++++++ apps/sim/lib/auth/auth.ts | 45 ++---------- .../core/async-jobs/backends/database.test.ts | 38 +++++++++- .../lib/core/async-jobs/backends/database.ts | 71 ++++++++++-------- .../sim/lib/execution/removal-stop-records.ts | 2 +- .../execution/removal-stops.integration.ts | 37 ++++++---- apps/sim/lib/execution/removal-stops.ts | 25 +++++-- apps/sim/lib/workflows/lifecycle.ts | 72 ++++++++++++------- 9 files changed, 289 insertions(+), 115 deletions(-) create mode 100644 apps/sim/lib/auth/admin-removal-stops.test.ts create mode 100644 apps/sim/lib/auth/admin-removal-stops.ts diff --git a/apps/sim/lib/auth/admin-removal-stops.test.ts b/apps/sim/lib/auth/admin-removal-stops.test.ts new file mode 100644 index 00000000000..022d2f04f8f --- /dev/null +++ b/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() + }) +}) diff --git a/apps/sim/lib/auth/admin-removal-stops.ts b/apps/sim/lib/auth/admin-removal-stops.ts new file mode 100644 index 00000000000..3683a87a3e8 --- /dev/null +++ b/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() + +/** + * 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 { + 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 { + 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) + } +} diff --git a/apps/sim/lib/auth/auth.ts b/apps/sim/lib/auth/auth.ts index 1381502965f..110d53f4060 100644 --- a/apps/sim/lib/auth/auth.ts +++ b/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() - /** * 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() }, }, diff --git a/apps/sim/lib/core/async-jobs/backends/database.test.ts b/apps/sim/lib/core/async-jobs/backends/database.test.ts index c5880c464d2..a32f2b9b3ed 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.test.ts +++ b/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() + }) }) diff --git a/apps/sim/lib/core/async-jobs/backends/database.ts b/apps/sim/lib/core/async-jobs/backends/database.ts index 27588386758..03a6cb34887 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.ts +++ b/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 @@ -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 { - 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 { + 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 { diff --git a/apps/sim/lib/execution/removal-stop-records.ts b/apps/sim/lib/execution/removal-stop-records.ts index 8be52a9bb25..10cd6921335 100644 --- a/apps/sim/lib/execution/removal-stop-records.ts +++ b/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 { diff --git a/apps/sim/lib/execution/removal-stops.integration.ts b/apps/sim/lib/execution/removal-stops.integration.ts index 9c6fe26c9a7..88a13df314c 100644 --- a/apps/sim/lib/execution/removal-stops.integration.ts +++ b/apps/sim/lib/execution/removal-stops.integration.ts @@ -1,6 +1,7 @@ /** * Removing an account, banning it, or deleting its user row directly stops the workflow runs the - * removal orphans, against real PostgreSQL and Redis through the production removal paths. A + * removal orphans, against real PostgreSQL and Redis through the production removal paths (for + * `/admin/remove-user`, the handlers of its Better Auth delete hooks around a bare delete). A * stopped execution is one whose cancellation flag, the one its engine watches, is set in Redis; * a stopped queued job is a cancelled `async_jobs` row (the database queue these suites run on). */ @@ -33,14 +34,10 @@ import { import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' import { generateId } from '@sim/utils/id' import { and, eq, inArray, sql } from 'drizzle-orm' +import { holdAdminRemovalStops, stopAdminRemovedRuns } from '@/lib/auth/admin-removal-stops' import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis' import { processOutboxEventById } from '@/lib/core/outbox/service' import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' -import { - collectAccountRemovalStops, - enqueueExecutionStops, - signalExecutionStops, -} from '@/lib/execution/removal-stop-records' import { executionRemovalStopOutboxHandlers } from '@/lib/execution/removal-stops' import { deleteUserAccount } from '@/lib/users/account-deletion' import { disableUserResources } from '@/lib/workflows/lifecycle' @@ -285,6 +282,15 @@ describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan', await seedWorkflow(bannedId, ownedWorkspaceId) ) const queued = await seedQueuedJob(ownedWorkspaceId) + const earlierArchivedId = await seedWorkspace(bannedId) + const runningInEarlierArchived = await seedExecution( + earlierArchivedId, + await seedWorkflow(bannedId, earlierArchivedId) + ) + await db + .update(workspace) + .set({ archivedAt: new Date() }) + .where(eq(workspace.id, earlierArchivedId)) const keptWorkspaceId = await seedWorkspace(bystanderId, [bannedId]) const keptRunning = await seedExecution( keptWorkspaceId, @@ -300,19 +306,26 @@ describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan', .where(eq(workspace.id, ownedWorkspaceId)) expect(archived.archivedAt).not.toBeNull() expect(await isCancelled(running)).toBe(true) + expect(await isCancelled(runningInEarlierArchived)).toBe(true) expect(await isCancelled(keptRunning)).toBe(false) const recorded = await stopEvents([ownedWorkspaceId]) expect(recorded.map((event) => event.status)).toEqual(['pending']) + /** Admitted before the archive, logged after the stop was recorded. */ + const loggedLate = await seedExecution( + ownedWorkspaceId, + await seedWorkflow(bannedId, ownedWorkspaceId) + ) await runOutboxWorker(recorded.map((event) => event.id)) + expect(await isCancelled(loggedLate)).toBe(true) expect(await jobStatus(queued)).toBe('cancelled') expect(await jobStatus(keptQueued)).toBe('pending') }) /** - * Mirrors the Better Auth user delete hooks around `/admin/remove-user`: read before the bare - * delete, enqueued and signalled once it has committed. + * Drives the handlers Better Auth runs as the user delete's `before` and `after` hooks around + * the bare delete `/admin/remove-user` issues. */ it('stops the runs of workflows a directly deleted user created in someone else’s workspace', async () => { const removedId = await seedUser() @@ -328,15 +341,15 @@ describe.runIf(Boolean(redisUrl))('removals stop the workflow runs they orphan', ) const bystandersQueued = await seedQueuedJob(sharedWorkspaceId) - const plan = await collectAccountRemovalStops(db, removedId) + const removed = { id: removedId } + await holdAdminRemovalStops(removed) await db.delete(user).where(eq(user.id, removedId)) - const eventIds = await enqueueExecutionStops(db, plan) - await signalExecutionStops(plan.executions) + await stopAdminRemovedRuns(removed) expect(await isCancelled(removedUsersRun)).toBe(true) expect(await isCancelled(bystandersRun)).toBe(false) - await runOutboxWorker(eventIds) + await runOutboxWorker((await stopEvents([removedUsersRun])).map((event) => event.id)) expect(await jobStatus(bystandersQueued)).toBe('pending') expect((await stopEvents([removedUsersRun])).map((event) => event.status)).toEqual([ diff --git a/apps/sim/lib/execution/removal-stops.ts b/apps/sim/lib/execution/removal-stops.ts index 566053ce96a..865730906ce 100644 --- a/apps/sim/lib/execution/removal-stops.ts +++ b/apps/sim/lib/execution/removal-stops.ts @@ -1,3 +1,4 @@ +import { db } from '@sim/db' import { isRecordLike } from '@sim/utils/object' import { getInlineJobQueue, getJobQueue } from '@/lib/core/async-jobs' import { @@ -8,6 +9,7 @@ import { } from '@/lib/core/outbox/service' import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' import { + collectExecutionStops, type ExecutionStopPayload, signalExecutionStops, } from '@/lib/execution/removal-stop-records' @@ -69,13 +71,26 @@ async function sweepWorkspaceJobs(workspaceId: string, signal: AbortSignal) { const stopRemovedExecutions: OutboxHandler = withOutboxHandlerTimeout( async (rawPayload, context) => { const payload = parsePayload(rawPayload) - const unrecorded = await signalExecutionStops( - payload.executions.map(({ executionId, executionDeadlineAt }) => ({ + const executions = new Map( + payload.executions.map(({ executionId, executionDeadlineAt }) => [ executionId, - executionDeadlineAt: executionDeadlineAt ? new Date(executionDeadlineAt) : null, - })), - context.signal + { + executionId, + executionDeadlineAt: executionDeadlineAt ? new Date(executionDeadlineAt) : null, + }, + ]) ) + /** + * An archived workspace keeps its rows, so a run that passed admission before the archive + * but logged after the stop was recorded is found here. An erased workspace has none left. + */ + if (payload.workspaceId) { + const { executions: live } = await collectExecutionStops(db, { + workspaceIds: [payload.workspaceId], + }) + for (const execution of live) executions.set(execution.executionId, execution) + } + const unrecorded = await signalExecutionStops([...executions.values()], context.signal) const sweep = payload.workspaceId ? await sweepWorkspaceJobs(payload.workspaceId, context.signal) : { cancelledJobs: 0, complete: true } diff --git a/apps/sim/lib/workflows/lifecycle.ts b/apps/sim/lib/workflows/lifecycle.ts index d48adb2dbfc..beb6c74c69f 100644 --- a/apps/sim/lib/workflows/lifecycle.ts +++ b/apps/sim/lib/workflows/lifecycle.ts @@ -12,7 +12,7 @@ import { } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' -import { and, eq, inArray, isNull } from 'drizzle-orm' +import { and, eq, inArray, isNotNull, isNull } from 'drizzle-orm' import { env } from '@/lib/core/config/env' import { PlatformEvents } from '@/lib/core/telemetry' import { generateRequestId } from '@/lib/core/utils/request' @@ -294,8 +294,9 @@ export async function archiveWorkflowsByIdsInWorkspace( * Disables all resources owned by a banned user by archiving every workspace * they own (cascading to workflows, chats, KBs, tables, files, etc.), * deleting their personal API keys, and stopping the workflow runs in progress - * in the archived workspaces. Each archive records Stop for its Chat runs, which - * they read on their next run-control call. + * in every workspace they own once it is archived, including ones archived + * earlier. Each archive records Stop for its Chat runs, which they read on their + * next run-control call. * * Runs the user started in workspaces they do not own keep going. The log row * has no actor column, and the user it records is the billed account for @@ -316,41 +317,58 @@ export async function disableUserResources(userId: string): Promise { .from(workspace) .where(and(eq(workspace.ownerId, userId), isNull(workspace.archivedAt))) - const failures: { workspaceId: string; error: unknown }[] = [] - const archivedWorkspaceIds: string[] = [] - // Sequential: environments of one Project contend on its exclusive lock. - for (const row of ownedWorkspaces) { - try { - const result = await archiveWorkspace(row.id, { requestId, expectedOwnerId: userId }) - if (result.archived) archivedWorkspaceIds.push(row.id) - } catch (error) { - failures.push({ workspaceId: row.id, error }) - } - } - + const { recordExecutionStops, signalExecutionStops } = await import( + '@/lib/execution/removal-stop-records' + ) /** - * After the archives commit, which refuse new runs, so every run admitted before them is found. - * In-flight runs are signalled here; the recorded stop's queue sweep and retries run in the - * outbox worker, since this module sits in page graphs that must not load that delivery. + * Records the workspaces' stops and signals their in-flight runs; the recorded stop's queue + * sweep and retries run in the outbox worker, since this module sits in page graphs that must + * not load that delivery. Logged, not thrown: the ban and archives have committed, and a retry + * would find no unarchived workspace to stop. */ - if (archivedWorkspaceIds.length > 0) { + const stopRuns = async (workspaceIds: string[]): Promise => { + if (workspaceIds.length === 0) return try { - const { recordExecutionStops, signalExecutionStops } = await import( - '@/lib/execution/removal-stop-records' - ) - const executions = await db.transaction((tx) => - recordExecutionStops(tx, { workspaceIds: archivedWorkspaceIds }) - ) + const executions = await db.transaction((tx) => recordExecutionStops(tx, { workspaceIds })) await signalExecutionStops(executions) } catch (error) { - /** Not thrown: the ban and archives committed, and a retry finds no unarchived workspace. */ logger.error(`[${requestId}] Failed to stop runs for banned user ${userId}`, { - workspaceIds: archivedWorkspaceIds, + workspaceIds, error: getErrorMessage(error), }) } } + const failures: { workspaceId: string; error: unknown }[] = [] + const stopping: Promise[] = [] + const stoppedWorkspaceIds = new Set() + // Sequential: environments of one Project contend on its exclusive lock. + for (const row of ownedWorkspaces) { + try { + await archiveWorkspace(row.id, { + requestId, + expectedOwnerId: userId, + /** Called once the archive commits, before its slower post-commit cleanup. */ + deliverStops: () => { + stoppedWorkspaceIds.add(row.id) + stopping.push(stopRuns([row.id])) + }, + }) + } catch (error) { + failures.push({ workspaceId: row.id, error }) + } + } + + /** Workspaces archived before this ban still have the runs an archive leaves going. */ + const earlierArchived = await db + .select({ id: workspace.id }) + .from(workspace) + .where(and(eq(workspace.ownerId, userId), isNotNull(workspace.archivedAt))) + stopping.push( + stopRuns(earlierArchived.map((row) => row.id).filter((id) => !stoppedWorkspaceIds.has(id))) + ) + await Promise.all(stopping) + if (failures.length > 0) { logger.error(`[${requestId}] Failed to archive workspaces for banned user ${userId}`, { failures: failures.map(({ workspaceId, error }) => ({ From 197e901bdf85385b461c36c27d353a2273b761ff Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 15:43:49 -0700 Subject: [PATCH 5/9] refactor(execution): tighten removal stops after final audit - disableUserResources reads owned workspaces once and stops earlier-archived ones up front, dropping the post-archive read that could fail the ban - Trigger.dev cancelByWorkspace reuses scanAndCancelTriggerRuns; workspace tag shares one builder with enqueue - database queue shares abortInlineJob with cancelJob and returns completeness from control flow - removal stop handler parses its payload with zod and signals while it sweeps - drop the unused event-id return and redundant bucketing state --- .../lib/core/async-jobs/backends/database.ts | 33 ++++----- .../core/async-jobs/backends/trigger-dev.ts | 71 +++++++++---------- .../sim/lib/execution/removal-stop-records.ts | 18 ++--- apps/sim/lib/execution/removal-stops.ts | 64 ++++++----------- apps/sim/lib/workflows/lifecycle.ts | 22 ++---- 5 files changed, 84 insertions(+), 124 deletions(-) diff --git a/apps/sim/lib/core/async-jobs/backends/database.ts b/apps/sim/lib/core/async-jobs/backends/database.ts index 03a6cb34887..7790eb77a9b 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.ts +++ b/apps/sim/lib/core/async-jobs/backends/database.ts @@ -55,6 +55,15 @@ function rowToJob(row: AsyncJobRow): Job { const inlineAbortControllers = new Map() const inlineExecutionControllers = new Map>() +/** 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 @@ -424,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 @@ -538,7 +541,6 @@ export class DatabaseJobQueue implements JobQueueBackend { 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() @@ -556,21 +558,16 @@ export class DatabaseJobQueue implements JobQueueBackend { ) .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) abortInlineJob(id) cancelledJobs += cancelled.length if (cancelled.length < batchSize) { - complete = true - break + logger.info('Cancelled database jobs for workspace', { workspaceId, cancelledJobs }) + return { cancelledJobs, complete: true } } } - logger.info('Cancelled database jobs for workspace', { workspaceId, cancelledJobs, complete }) - return { cancelledJobs, complete } + logger.info('Paused database job sweep for workspace', { workspaceId, cancelledJobs }) + return { cancelledJobs, complete: false } } cancelByKey(cancelKey: string): boolean { diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts index befb17a990d..17cbf317338 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts @@ -128,6 +128,10 @@ function buildExecutionTag(executionId: string): string { return tag.length <= 128 ? tag : `executionIdHash:${sha256Hex(executionId)}` } +function buildWorkspaceTag(workspaceId: string): string { + return `workspaceId:${workspaceId}` +} + function buildWorkflowTag(workflowId: string): string { const tag = `workflowId:${workflowId}` return tag.length <= 128 ? tag : `workflowIdHash:${sha256Hex(workflowId)}` @@ -172,10 +176,13 @@ interface CancellationListRun { } interface CancellationScanOptions { - binding: ExecutionJobBinding + /** Required by any candidate that sets `verifyPayload`. */ + binding?: ExecutionJobBinding cancelJob: (jobId: string) => Promise listRuns: () => AsyncIterable selectCandidate: (run: CancellationListRun) => CancellationCandidate | undefined + /** Checked before each listed run; true ends the scan there. */ + shouldStop?: () => boolean state: CancellationScanState } @@ -186,7 +193,7 @@ function recordCancellationCandidateFailure(state: CancellationScanState, error: async function processCancellationCandidateBatch( candidates: readonly CancellationCandidate[], - binding: ExecutionJobBinding, + binding: ExecutionJobBinding | undefined, state: CancellationScanState, cancelJob: (jobId: string) => Promise ): Promise { @@ -200,7 +207,7 @@ async function processCancellationCandidateBatch( uniqueCandidates.map(async (candidate) => { if (candidate.verifyPayload) { const legacyRun = await runs.retrieve(candidate.id) - if (!payloadMatchesExecution(legacyRun.payload, binding)) return false + if (!binding || !payloadMatchesExecution(legacyRun.payload, binding)) return false } await cancelJob(candidate.id) return true @@ -232,6 +239,7 @@ async function scanAndCancelTriggerRuns(options: CancellationScanOptions): Promi try { for await (const run of options.listRuns()) { + if (options.shouldStop?.()) break const candidate = options.selectCandidate(run) if (!candidate) continue candidates.push(candidate) @@ -665,44 +673,33 @@ export class TriggerDevJobQueue implements JobQueueBackend { workspaceId: string, options: WorkspaceJobCancellationOptions ): Promise { - const workspaceTag = `workspaceId:${workspaceId}` + const workspaceTag = buildWorkspaceTag(workspaceId) const state: CancellationScanState = { cancelledJobs: 0, failureCount: 0 } - let pending: string[] = [] let inspectedRuns = 0 let complete = true - const flush = async (): Promise => { - const runIds = pending - pending = [] - const results = await Promise.allSettled(runIds.map((runId) => this.cancelJob(runId))) - for (const result of results) { - if (result.status === 'rejected') recordCancellationCandidateFailure(state, result.reason) - else state.cancelledJobs += 1 - } - } - - try { - for await (const run of runs.list({ - tag: workspaceTag, - taskIdentifier: [...WORKSPACE_EXECUTION_JOB_TYPES], - status: [...ACTIVE_TRIGGER_RUN_STATUSES], - from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000), - limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE, - })) { - if (inspectedRuns >= options.maxRuns || options.signal?.aborted) { - complete = false - break - } - if (!run.tags.includes(workspaceTag)) continue + await scanAndCancelTriggerRuns({ + cancelJob: (jobId) => this.cancelJob(jobId), + listRuns: () => + runs.list({ + tag: workspaceTag, + taskIdentifier: [...WORKSPACE_EXECUTION_JOB_TYPES], + status: [...ACTIVE_TRIGGER_RUN_STATUSES], + from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000), + limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE, + }), + shouldStop: () => { + if (inspectedRuns < options.maxRuns && !options.signal?.aborted) return false + complete = false + return true + }, + selectCandidate: (run) => { + if (!run.tags.includes(workspaceTag)) return undefined inspectedRuns += 1 - pending.push(run.id) - if (pending.length >= CANCELLATION_OPERATION_CHUNK_SIZE) await flush() - } - } catch (error) { - recordCancellationCandidateFailure(state, error) - } finally { - await flush() - } + return { id: run.id, verifyPayload: false } + }, + state, + }) if (state.firstError) { logger.error('Failed to cancel trigger.dev runs for workspace', { @@ -748,7 +745,7 @@ function buildTags(options?: EnqueueOptions): string[] { const workflowId = meta?.correlation?.workflowId ?? (typeof meta?.workflowId === 'string' ? meta.workflowId : undefined) - if (meta?.workspaceId) tags.push(`workspaceId:${meta.workspaceId}`) + if (meta?.workspaceId) tags.push(buildWorkspaceTag(meta.workspaceId)) if (workflowId) tags.push(buildWorkflowTag(workflowId)) if (meta?.userId) tags.push(`userId:${meta.userId}`) diff --git a/apps/sim/lib/execution/removal-stop-records.ts b/apps/sim/lib/execution/removal-stop-records.ts index 10cd6921335..1ffd22dc4cc 100644 --- a/apps/sim/lib/execution/removal-stop-records.ts +++ b/apps/sim/lib/execution/removal-stop-records.ts @@ -43,7 +43,7 @@ export interface ExecutionStopPlan { executions: PlannedExecutionStop[] } -export interface ExecutionStopPayload { +interface ExecutionStopPayload { /** Workspace whose queued and running jobs this event sweeps. */ workspaceId?: string executions: { executionId: string; executionDeadlineAt: string | null }[] @@ -89,7 +89,6 @@ export async function collectExecutionStops( }) .from(workflowExecutionLogs) .where(and(eq(workflowExecutionLogs.status, 'running'), inScope)) - /** Driven from the scope's log rows, so `resume_queue` is read through its parent index. */ const resuming = await executor .select({ executionId: resumeQueue.newExecutionId, @@ -142,17 +141,14 @@ function toPayloadExecutions( /** One event per removed workspace carries its sweep; executions beyond one event's share spill over. */ function buildPayloads(plan: ExecutionStopPlan): ExecutionStopPayload[] { - const removedWorkspaces = new Set(plan.workspaceIds) const byWorkspace = new Map( plan.workspaceIds.map((workspaceId) => [workspaceId, []]) ) const elsewhere: PlannedExecutionStop[] = [] for (const execution of plan.executions) { - if (execution.workspaceId && removedWorkspaces.has(execution.workspaceId)) { - byWorkspace.get(execution.workspaceId)?.push(execution) - } else { - elsewhere.push(execution) - } + const removed = execution.workspaceId ? byWorkspace.get(execution.workspaceId) : undefined + if (removed) removed.push(execution) + else elsewhere.push(execution) } const payloads: ExecutionStopPayload[] = [] @@ -174,12 +170,10 @@ function buildPayloads(plan: ExecutionStopPlan): ExecutionStopPayload[] { export async function enqueueExecutionStops( executor: Pick, plan: ExecutionStopPlan -): Promise { - const eventIds: string[] = [] +): Promise { for (const chunk of chunkArray(buildPayloads(plan), MAX_BULK_ENQUEUE_EVENTS)) { - eventIds.push(...(await enqueueOutboxEvents(executor, EXECUTION_REMOVAL_STOP_EVENT, chunk))) + await enqueueOutboxEvents(executor, EXECUTION_REMOVAL_STOP_EVENT, chunk) } - return eventIds } /** diff --git a/apps/sim/lib/execution/removal-stops.ts b/apps/sim/lib/execution/removal-stops.ts index 865730906ce..d79e5736c0d 100644 --- a/apps/sim/lib/execution/removal-stops.ts +++ b/apps/sim/lib/execution/removal-stops.ts @@ -1,5 +1,5 @@ import { db } from '@sim/db' -import { isRecordLike } from '@sim/utils/object' +import { z } from 'zod' import { getInlineJobQueue, getJobQueue } from '@/lib/core/async-jobs' import { continueOutboxHandler, @@ -8,40 +8,24 @@ import { withOutboxHandlerTimeout, } from '@/lib/core/outbox/service' import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' -import { - collectExecutionStops, - type ExecutionStopPayload, - signalExecutionStops, -} from '@/lib/execution/removal-stop-records' +import { collectExecutionStops, signalExecutionStops } from '@/lib/execution/removal-stop-records' /** Queue runs one handler attempt cancels before handing the rest to its next run. */ const SWEEP_MAX_RUNS_PER_ATTEMPT = 1_000 const HANDLER_TIMEOUT_MS = 5 * 60 * 1000 -function parsePayload(payload: unknown): ExecutionStopPayload { - if (!isRecordLike(payload) || !Array.isArray(payload.executions)) { - throw new Error('Execution removal stop payload is malformed') - } - let workspaceId: string | undefined - if (payload.workspaceId !== undefined) { - if (typeof payload.workspaceId !== 'string' || payload.workspaceId.length === 0) { - throw new Error('Execution removal stop payload has an invalid workspaceId') - } - workspaceId = payload.workspaceId - } - const executions = payload.executions.map((entry: unknown) => { - if ( - !isRecordLike(entry) || - typeof entry.executionId !== 'string' || - entry.executionId.length === 0 || - (entry.executionDeadlineAt !== null && typeof entry.executionDeadlineAt !== 'string') - ) { - throw new Error('Execution removal stop payload has an invalid execution') - } - return { executionId: entry.executionId, executionDeadlineAt: entry.executionDeadlineAt } - }) - return { ...(workspaceId ? { workspaceId } : {}), executions } -} +const payloadSchema = z.object({ + workspaceId: z.string().min(1).optional(), + executions: z.array( + z.object({ + executionId: z.string().min(1), + executionDeadlineAt: z + .string() + .nullable() + .transform((deadline) => (deadline ? new Date(deadline) : null)), + }) + ), +}) /** * Sweeps the primary queue and the database queue that inline webhook runs use even when @@ -70,15 +54,9 @@ async function sweepWorkspaceJobs(workspaceId: string, signal: AbortSignal) { */ const stopRemovedExecutions: OutboxHandler = withOutboxHandlerTimeout( async (rawPayload, context) => { - const payload = parsePayload(rawPayload) + const payload = payloadSchema.parse(rawPayload) const executions = new Map( - payload.executions.map(({ executionId, executionDeadlineAt }) => [ - executionId, - { - executionId, - executionDeadlineAt: executionDeadlineAt ? new Date(executionDeadlineAt) : null, - }, - ]) + payload.executions.map((execution) => [execution.executionId, execution]) ) /** * An archived workspace keeps its rows, so a run that passed admission before the archive @@ -90,10 +68,12 @@ const stopRemovedExecutions: OutboxHandler = withOutboxHandlerTimeout( }) for (const execution of live) executions.set(execution.executionId, execution) } - const unrecorded = await signalExecutionStops([...executions.values()], context.signal) - const sweep = payload.workspaceId - ? await sweepWorkspaceJobs(payload.workspaceId, context.signal) - : { cancelledJobs: 0, complete: true } + const [unrecorded, sweep] = await Promise.all([ + signalExecutionStops([...executions.values()], context.signal), + payload.workspaceId + ? sweepWorkspaceJobs(payload.workspaceId, context.signal) + : { cancelledJobs: 0, complete: true }, + ]) if (unrecorded > 0) { throw new Error(`${unrecorded} execution cancellations were not recorded`) } diff --git a/apps/sim/lib/workflows/lifecycle.ts b/apps/sim/lib/workflows/lifecycle.ts index beb6c74c69f..535672df5f8 100644 --- a/apps/sim/lib/workflows/lifecycle.ts +++ b/apps/sim/lib/workflows/lifecycle.ts @@ -12,7 +12,7 @@ import { } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' -import { and, eq, inArray, isNotNull, isNull } from 'drizzle-orm' +import { and, eq, inArray, isNull } from 'drizzle-orm' import { env } from '@/lib/core/config/env' import { PlatformEvents } from '@/lib/core/telemetry' import { generateRequestId } from '@/lib/core/utils/request' @@ -312,10 +312,11 @@ export async function disableUserResources(userId: string): Promise { /** Revoked first: the keys are the ban's access boundary, and an archive may fail. */ await db.delete(apiKey).where(eq(apiKey.userId, userId)) - const ownedWorkspaces = await db - .select({ id: workspace.id }) + const owned = await db + .select({ id: workspace.id, archivedAt: workspace.archivedAt }) .from(workspace) - .where(and(eq(workspace.ownerId, userId), isNull(workspace.archivedAt))) + .where(eq(workspace.ownerId, userId)) + const ownedWorkspaces = owned.filter((row) => !row.archivedAt) const { recordExecutionStops, signalExecutionStops } = await import( '@/lib/execution/removal-stop-records' @@ -340,8 +341,8 @@ export async function disableUserResources(userId: string): Promise { } const failures: { workspaceId: string; error: unknown }[] = [] - const stopping: Promise[] = [] - const stoppedWorkspaceIds = new Set() + /** Workspaces archived before this ban still have the runs an archive leaves going. */ + const stopping = [stopRuns(owned.filter((row) => row.archivedAt).map((row) => row.id))] // Sequential: environments of one Project contend on its exclusive lock. for (const row of ownedWorkspaces) { try { @@ -350,7 +351,6 @@ export async function disableUserResources(userId: string): Promise { expectedOwnerId: userId, /** Called once the archive commits, before its slower post-commit cleanup. */ deliverStops: () => { - stoppedWorkspaceIds.add(row.id) stopping.push(stopRuns([row.id])) }, }) @@ -359,14 +359,6 @@ export async function disableUserResources(userId: string): Promise { } } - /** Workspaces archived before this ban still have the runs an archive leaves going. */ - const earlierArchived = await db - .select({ id: workspace.id }) - .from(workspace) - .where(and(eq(workspace.ownerId, userId), isNotNull(workspace.archivedAt))) - stopping.push( - stopRuns(earlierArchived.map((row) => row.id).filter((id) => !stoppedWorkspaceIds.has(id))) - ) await Promise.all(stopping) if (failures.length > 0) { From cd543967844135e5620e289259f8310ea6d70aa8 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 15:59:54 -0700 Subject: [PATCH 6/9] 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. --- .../core/async-jobs/backends/database.test.ts | 10 +++++++++ .../lib/core/async-jobs/backends/database.ts | 17 +++++++++++++-- .../sim/lib/execution/removal-stop-records.ts | 2 +- .../execution/removal-stops.integration.ts | 16 +++++++++++--- apps/sim/lib/execution/removal-stops.test.ts | 8 +++++++ apps/sim/lib/execution/removal-stops.ts | 21 ++++++++++++++----- apps/sim/lib/workflows/lifecycle.ts | 4 +++- 7 files changed, 66 insertions(+), 12 deletions(-) diff --git a/apps/sim/lib/core/async-jobs/backends/database.test.ts b/apps/sim/lib/core/async-jobs/backends/database.test.ts index a32f2b9b3ed..5b4cce31ca9 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.test.ts +++ b/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)) diff --git a/apps/sim/lib/core/async-jobs/backends/database.ts b/apps/sim/lib/core/async-jobs/backends/database.ts index 7790eb77a9b..972a220d5db 100644 --- a/apps/sim/lib/core/async-jobs/backends/database.ts +++ b/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 } } } diff --git a/apps/sim/lib/execution/removal-stop-records.ts b/apps/sim/lib/execution/removal-stop-records.ts index 1ffd22dc4cc..97655c6339b 100644 --- a/apps/sim/lib/execution/removal-stop-records.ts +++ b/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() for (const row of [...running, ...resuming]) { diff --git a/apps/sim/lib/execution/removal-stops.integration.ts b/apps/sim/lib/execution/removal-stops.integration.ts index 88a13df314c..a74d38b518e 100644 --- a/apps/sim/lib/execution/removal-stops.integration.ts +++ b/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']) diff --git a/apps/sim/lib/execution/removal-stops.test.ts b/apps/sim/lib/execution/removal-stops.test.ts index 0f8e34512c1..f33e7dff222 100644 --- a/apps/sim/lib/execution/removal-stops.test.ts +++ b/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 }) diff --git a/apps/sim/lib/execution/removal-stops.ts b/apps/sim/lib/execution/removal-stops.ts index d79e5736c0d..93e6de89801 100644 --- a/apps/sim/lib/execution/removal-stops.ts +++ b/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() + 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 } } diff --git a/apps/sim/lib/workflows/lifecycle.ts b/apps/sim/lib/workflows/lifecycle.ts index 535672df5f8..b7bbde2aca1 100644 --- a/apps/sim/lib/workflows/lifecycle.ts +++ b/apps/sim/lib/workflows/lifecycle.ts @@ -327,12 +327,14 @@ export async function disableUserResources(userId: string): Promise { * 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 => { 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 { } 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'}` ) } From e6fb2802611739337e3a24c12959b1d91c3d5e98 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 16:10:12 -0700 Subject: [PATCH 7/9] fix(execution): abort local runs before the removal signal deadline can skip them The signal deadline bounds only the Redis flag writes. Aborting a run this process holds needs no Redis, so every execution is aborted first, even after the deadline has passed. Corrects the ban's note on why a failed stop record is logged: banning again records the stops again. --- .../execution/removal-stop-records.test.ts | 32 +++++++++++++++++++ .../sim/lib/execution/removal-stop-records.ts | 23 +++++++------ apps/sim/lib/workflows/lifecycle.ts | 5 +-- 3 files changed, 48 insertions(+), 12 deletions(-) create mode 100644 apps/sim/lib/execution/removal-stop-records.test.ts diff --git a/apps/sim/lib/execution/removal-stop-records.test.ts b/apps/sim/lib/execution/removal-stop-records.test.ts new file mode 100644 index 00000000000..d74194e4c8c --- /dev/null +++ b/apps/sim/lib/execution/removal-stop-records.test.ts @@ -0,0 +1,32 @@ +/** + * 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. + */ +import { describe, expect, it, vi } from 'vitest' +import { + registerManualExecutionAborter, + unregisterManualExecutionAborter, +} from '@/lib/execution/manual-cancellation' +import { 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 () => { + const abort = vi.fn() + registerManualExecutionAborter('execution-local', abort) + const deadline = new AbortController() + deadline.abort() + + try { + await expect( + signalExecutionStops( + [{ executionId: 'execution-local', executionDeadlineAt: null }], + deadline.signal + ) + ).resolves.toBe(1) + } finally { + unregisterManualExecutionAborter('execution-local', abort) + } + + expect(abort).toHaveBeenCalledOnce() + }) +}) diff --git a/apps/sim/lib/execution/removal-stop-records.ts b/apps/sim/lib/execution/removal-stop-records.ts index 97655c6339b..9e2dd0a4cb1 100644 --- a/apps/sim/lib/execution/removal-stop-records.ts +++ b/apps/sim/lib/execution/removal-stop-records.ts @@ -193,8 +193,9 @@ export async function recordExecutionStops( * Sets the cancellation flag each execution's engine watches, and aborts the ones this process * runs, without the queue sweep or the retry the outbox event adds. Callers whose module graph * cannot load the delivery stop in-flight runs with it at once and leave the rest to the outbox - * worker. Returns how many flags Redis refused or the deadline (`signal`, else - * {@link SIGNAL_DEADLINE_MS}) left unsent; never throws. + * worker. The deadline (`signal`, else {@link SIGNAL_DEADLINE_MS}) bounds only the Redis writes; + * every execution this process runs is aborted first. Returns how many flags Redis refused or + * the deadline left unsent; never throws. * * Without Redis at all the flag is process-local by design (`markExecutionCancelled`) and a * retry from any process could reach no further, so that outcome does not count as refused. @@ -203,6 +204,16 @@ export async function signalExecutionStops( executions: readonly Pick[], signal?: AbortSignal ): Promise { + for (const { executionId } of executions) { + try { + abortManualExecution(executionId) + } catch (error) { + logger.warn('Failed to abort a local execution being stopped', { + executionId, + error: describeError(error), + }) + } + } const deadline = signal ?? AbortSignal.timeout(SIGNAL_DEADLINE_MS) let refused = 0 await mapWithConcurrency(executions, SIGNAL_CONCURRENCY, async (execution) => { @@ -222,14 +233,6 @@ export async function signalExecutionStops( error: describeError(error), }) } - try { - abortManualExecution(execution.executionId) - } catch (error) { - logger.warn('Failed to abort a local execution being stopped', { - executionId: execution.executionId, - error: describeError(error), - }) - } }) return refused } diff --git a/apps/sim/lib/workflows/lifecycle.ts b/apps/sim/lib/workflows/lifecycle.ts index b7bbde2aca1..b25f2f0dd67 100644 --- a/apps/sim/lib/workflows/lifecycle.ts +++ b/apps/sim/lib/workflows/lifecycle.ts @@ -324,8 +324,9 @@ export async function disableUserResources(userId: string): Promise { /** * Records the workspaces' stops and signals their in-flight runs; the recorded stop's queue * sweep and retries run in the outbox worker, since this module sits in page graphs that must - * not load that delivery. Logged, not thrown: the ban and archives have committed, and a retry - * would find no unarchived workspace to stop. + * not load that delivery. Logged, not thrown: the ban and archives have committed, and banning + * the user again records the stops again, since every owned workspace is stopped, archived or + * not. */ let stopFailed = false const stopRuns = async (workspaceIds: string[]): Promise => { From 85d2cd25e057270ce073d21d626b25d9d69f74d5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 16:23:02 -0700 Subject: [PATCH 8/9] 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. --- .../async-jobs/backends/trigger-dev.test.ts | 24 ++++++++++++++++ .../core/async-jobs/backends/trigger-dev.ts | 21 +++++++++----- .../execution/removal-stop-records.test.ts | 27 ++++++++++++++++-- .../sim/lib/execution/removal-stop-records.ts | 28 +++++++++++++++---- apps/sim/lib/workflows/lifecycle.ts | 4 ++- 5 files changed, 88 insertions(+), 16 deletions(-) diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts index 2de17ae39b0..75da9a44683 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts +++ b/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') + }) +}) diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts index 17cbf317338..506d925d390 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts @@ -675,7 +675,6 @@ export class TriggerDevJobQueue implements JobQueueBackend { ): Promise { 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, }) diff --git a/apps/sim/lib/execution/removal-stop-records.test.ts b/apps/sim/lib/execution/removal-stop-records.test.ts index d74194e4c8c..b3f43dcd2ec 100644 --- a/apps/sim/lib/execution/removal-stop-records.test.ts +++ b/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() + }) +}) diff --git a/apps/sim/lib/execution/removal-stop-records.ts b/apps/sim/lib/execution/removal-stop-records.ts index 9e2dd0a4cb1..15452986fb6 100644 --- a/apps/sim/lib/execution/removal-stop-records.ts +++ b/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 { - 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 } diff --git a/apps/sim/lib/workflows/lifecycle.ts b/apps/sim/lib/workflows/lifecycle.ts index b25f2f0dd67..f5e2269e452 100644 --- a/apps/sim/lib/workflows/lifecycle.ts +++ b/apps/sim/lib/workflows/lifecycle.ts @@ -332,7 +332,9 @@ export async function disableUserResources(userId: string): Promise { const stopRuns = async (workspaceIds: string[]): Promise => { 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 From 7dd7ff528296c50a4a818be5e933f57575464198 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 16 Oct 2026 16:38:10 -0700 Subject: [PATCH 9/9] fix(execution): sweep a removed workspace's Trigger.dev runs of any age The workspace sweep searched only runs created in the last 14 days, so an older run still queued or waiting survived the removal. It now passes only an upper time bound, which Trigger.dev treats as no lower bound, and falls back to the retention window only if Trigger.dev refuses that search as too expensive before returning a run. cancelByExecution is unchanged. --- .../async-jobs/backends/trigger-dev.test.ts | 38 +++++++++++++++ .../core/async-jobs/backends/trigger-dev.ts | 48 +++++++++++++++---- 2 files changed, 78 insertions(+), 8 deletions(-) diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts index 75da9a44683..c11d8276713 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.test.ts @@ -948,3 +948,41 @@ describe('TriggerDevJobQueue workspace cancellation past refusing runs', () => { expect(mockCancel).toHaveBeenCalledWith('run-9') }) }) + +describe('TriggerDevJobQueue workspace cancellation of long-lived runs', () => { + beforeEach(() => { + mockList.mockReset() + mockCancel.mockReset().mockResolvedValue(undefined) + }) + + it('searches every active run of the workspace, however long ago it was created', async () => { + mockList.mockReturnValue(createListPage([{ id: 'run-old', tags: ['workspaceId:removed'] }])) + const queue = new TriggerDevJobQueue() + + await queue.cancelByWorkspace('removed', { maxRuns: 100 }) + + const [filter] = mockList.mock.calls[0] as [Record] + expect(filter.from).toBeUndefined() + expect(filter.period).toBeUndefined() + expect(filter.to).toBeInstanceOf(Date) + expect(mockCancel).toHaveBeenCalledWith('run-old') + }) + + it('falls back to the retention window when Trigger.dev refuses the unbounded search', async () => { + mockList + .mockImplementationOnce(() => { + throw new MockApiError(422, { message: 'narrow the time window' }) + }) + .mockReturnValueOnce(createListPage([{ id: 'run-recent', tags: ['workspaceId:removed'] }])) + const queue = new TriggerDevJobQueue() + + await expect(queue.cancelByWorkspace('removed', { maxRuns: 100 })).resolves.toEqual({ + cancelledJobs: 1, + complete: true, + }) + + const [fallback] = mockList.mock.calls[1] as [Record] + expect(fallback.from).toBeInstanceOf(Date) + expect(mockCancel).toHaveBeenCalledWith('run-recent') + }) +}) diff --git a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts index 506d925d390..a31ea5916b4 100644 --- a/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts +++ b/apps/sim/lib/core/async-jobs/backends/trigger-dev.ts @@ -252,6 +252,45 @@ async function scanAndCancelTriggerRuns(options: CancellationScanOptions): Promi } } +/** + * Lists a workspace's active workflow runs however old they are: a removal must reach a run that + * has been queued or waiting for longer than any recent window. Trigger.dev searches only the + * last 7 days unless a time bound is given, and passing only `to` removes the lower bound. When + * Trigger.dev refuses that search as too expensive (422) before returning a run, the listing + * falls back to the queue-wait retention window rather than sweeping nothing. + */ +async function* listActiveWorkspaceRuns( + workspaceId: string, + workspaceTag: string +): AsyncIterable { + const filter = { + tag: workspaceTag, + taskIdentifier: [...WORKSPACE_EXECUTION_JOB_TYPES], + status: [...ACTIVE_TRIGGER_RUN_STATUSES], + limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE, + } + let listed = false + try { + for await (const run of runs.list({ ...filter, to: new Date() })) { + listed = true + yield run + } + return + } catch (error) { + if (listed || !(error instanceof ApiError) || error.status !== 422) throw error + logger.warn( + 'Trigger.dev refused an unbounded workspace run search; using the retention window', + { + workspaceId, + } + ) + } + yield* runs.list({ + ...filter, + from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000), + }) +} + /** * Maps trigger.dev task IDs to our JobType */ @@ -679,14 +718,7 @@ export class TriggerDevJobQueue implements JobQueueBackend { await scanAndCancelTriggerRuns({ cancelJob: (jobId) => this.cancelJob(jobId), - listRuns: () => - runs.list({ - tag: workspaceTag, - taskIdentifier: [...WORKSPACE_EXECUTION_JOB_TYPES], - status: [...ACTIVE_TRIGGER_RUN_STATUSES], - from: new Date(Date.now() - JOB_PENDING_RETENTION_HOURS * 60 * 60 * 1000), - limit: WORKSPACE_CANCELLATION_LIST_PAGE_SIZE, - }), + listRuns: () => listActiveWorkspaceRuns(workspaceId, workspaceTag), /** * 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