fix(async-jobs): finish cancellation scans after root errors

Author: icecrasher321

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

@@ -17,6 +17,7 @@ const runs = new Map<
 >()
 
 beforeEach(() => {
+  triggerSdkMockFns.mockRunsList.mockReset()
   runs.clear()
   triggerSdkMockFns.mockTasksTrigger.mockImplementation(async (type, payload) => {
     const id = `run_${generateId()}`

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

@@ -1,8 +1,9 @@
+import { idempotencyKey } from '@sim/db/schema'
 import {
   asyncJobsRegionMock,
   asyncJobsRegionMockFns,
 } from '@sim/testing/mocks/async-jobs-region.mock'
-import { dbChainMockFns } from '@sim/testing/mocks/database.mock'
+import { dbChainMockFns, queueTableRows } from '@sim/testing/mocks/database.mock'
 import { getMockLogger } from '@sim/testing/mocks/logger.mock'
 import {
   MockTriggerApiError as MockApiError,
@@ -329,7 +330,7 @@ describe('TriggerDevJobQueue status mapping', () => {
 
 describe('TriggerDevJobQueue cancellation', () => {
   beforeEach(() => {
-    mockList.mockReturnValue(
+    mockList.mockReset().mockReturnValue(
       createListPage([
         {
           id: 'run-1',
@@ -476,6 +477,49 @@ describe('TriggerDevJobQueue cancellation', () => {
     })
   })
 
+  it.each(['receipt', 'retrieve', 'cancel'] as const)(
+    'continues every discovery phase after a root %s failure',
+    async (failurePhase) => {
+      const failure = new Error(`root ${failurePhase} unavailable`)
+      const payload = { workflowId: 'workflow-1', executionId: 'execution-1' }
+      const cancelled = new Set<string>()
+      if (failurePhase === 'receipt') {
+        dbChainMockFns.limit.mockRejectedValueOnce(failure)
+      } else {
+        queueTableRows(idempotencyKey, [{ result: { runId: 'run_root' } }])
+      }
+      mockRetrieve.mockImplementation(async (id: string) => {
+        if (id === 'run_root' && failurePhase === 'retrieve') throw failure
+        return { id, taskIdentifier: 'workflow-execution', status: 'QUEUED', payload }
+      })
+      mockCancel.mockImplementation(async (id: string) => {
+        if (id === 'run_root') throw failure
+        cancelled.add(id)
+      })
+      mockList
+        .mockReturnValueOnce(
+          createListPage([
+            { id: 'tagged', tags: ['workflowId:workflow-1', 'executionId:execution-1'] },
+          ])
+        )
+        .mockReturnValueOnce(
+          createListPage([{ id: 'legacy-tagged', tags: ['workflowId:workflow-1'] }])
+        )
+        .mockReturnValueOnce(createListPage([{ id: 'legacy-untagged', tags: [] }]))
+
+      await expect(
+        new TriggerDevJobQueue().cancelByExecution(
+          {
+            ...payload,
+            rootJobId: 'workflow-execution:execution-1',
+          },
+          'standalone'
+        )
+      ).rejects.toBe(failure)
+      expect(cancelled).toEqual(new Set(['tagged', 'legacy-tagged', 'legacy-untagged']))
+    }
+  )
+
   it('cancels legacy workflow-tagged runs only after payload verification', async () => {
     mockList.mockReturnValueOnce(createListPage([])).mockReturnValueOnce(
       createListPage([

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

@@ -551,16 +551,20 @@ export class TriggerDevJobQueue implements JobQueueBackend {
 
     try {
       if (binding.rootJobId) {
-        const root = await this.getJob(binding.rootJobId)
-        if (
-          root &&
-          (allowedTaskIdentifiers as readonly string[]).includes(root.type) &&
-          (root.status === JOB_STATUS.PENDING || root.status === JOB_STATUS.PROCESSING) &&
-          payloadMatchesExecution(root.payload, binding)
-        ) {
-          await this.cancelJob(root.id)
-          cancelledRootRunId = root.id
-          state.cancelledJobs += 1
+        try {
+          const root = await this.getJob(binding.rootJobId)
+          if (
+            root &&
+            (allowedTaskIdentifiers as readonly string[]).includes(root.type) &&
+            (root.status === JOB_STATUS.PENDING || root.status === JOB_STATUS.PROCESSING) &&
+            payloadMatchesExecution(root.payload, binding)
+          ) {
+            await this.cancelJob(root.id)
+            cancelledRootRunId = root.id
+            state.cancelledJobs += 1
+          }
+        } catch (error) {
+          recordCancellationCandidateFailure(state, error)
         }
       }