Skip to content

Commit 8e2c0b9

Browse files
TheodoreSpeaksTheodoreSpeaks
committed
fix(api): make execution polling resume-aware
1 parent 6236bb2 commit 8e2c0b9

6 files changed

Lines changed: 190 additions & 60 deletions

File tree

‎apps/sim/app/api/resume/[workflowId]/[executionId]/[contextId]/route.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -269,7 +269,7 @@ describe('POST /api/resume/[workflowId]/[executionId]/[contextId]', () => {
269269
'resume-execution',
270270
expect.objectContaining({ resumeExecutionId: 'resume-execution-1' }),
271271
expect.objectContaining({
272-
jobId: 'resume-execution:resume-execution-1',
272+
jobId: 'resume-execution:resume-entry-1',
273273
metadata: expect.objectContaining({ workflowId: WORKFLOW_ID }),
274274
})
275275
)

‎apps/sim/app/api/resume/[workflowId]/[executionId]/[contextId]/route.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -327,7 +327,7 @@ export const POST = withRouteHandler(
327327
try {
328328
const jobQueue = await getJobQueue()
329329
queueJobId = await jobQueue.enqueue('resume-execution', resumePayload, {
330-
jobId: `${RESUME_EXECUTION_JOB_ID_PREFIX}${enqueueResult.resumeExecutionId}`,
330+
jobId: `${RESUME_EXECUTION_JOB_ID_PREFIX}${enqueueResult.resumeEntryId}`,
331331
metadata: { workflowId, workspaceId: workflow.workspaceId, userId },
332332
})
333333
logger.info('Enqueued async resume execution', {

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

Lines changed: 61 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -3,22 +3,25 @@
33
*/
44
import { beforeEach, describe, expect, it, vi } from 'vitest'
55

6-
const { MockApiError, mockResolveTriggerRegion, mockTrigger } = vi.hoisted(() => {
7-
class MockApiError extends Error {
8-
constructor(
9-
readonly status: number | undefined,
10-
message: string
11-
) {
12-
super(message)
6+
const { MockApiError, mockListRuns, mockResolveTriggerRegion, mockRetrieveRun, mockTrigger } =
7+
vi.hoisted(() => {
8+
class MockApiError extends Error {
9+
constructor(
10+
readonly status: number | undefined,
11+
message: string
12+
) {
13+
super(message)
14+
}
1315
}
14-
}
1516

16-
return {
17-
MockApiError,
18-
mockResolveTriggerRegion: vi.fn(),
19-
mockTrigger: vi.fn(),
20-
}
21-
})
17+
return {
18+
MockApiError,
19+
mockListRuns: vi.fn(),
20+
mockResolveTriggerRegion: vi.fn(),
21+
mockRetrieveRun: vi.fn(),
22+
mockTrigger: vi.fn(),
23+
}
24+
})
2225

2326
vi.mock('@trigger.dev/core/v3', () => ({
2427
taskContext: { isInsideTask: false },
@@ -28,7 +31,8 @@ vi.mock('@trigger.dev/sdk', () => ({
2831
ApiError: MockApiError,
2932
runs: {
3033
cancel: vi.fn(),
31-
retrieve: vi.fn(),
34+
list: mockListRuns,
35+
retrieve: mockRetrieveRun,
3236
},
3337
tasks: {
3438
batchTriggerAndWait: vi.fn(),
@@ -63,6 +67,7 @@ describe('TriggerDevJobQueue enqueue', () => {
6367
expect.objectContaining({
6468
idempotencyKey: 'workflow:1',
6569
idempotencyKeyTTL: '14d',
70+
tags: ['jobId:workflow:1'],
6671
})
6772
)
6873
})
@@ -113,3 +118,44 @@ describe('TriggerDevJobQueue enqueue', () => {
113118
expect(mockTrigger).not.toHaveBeenCalled()
114119
})
115120
})
121+
122+
describe('TriggerDevJobQueue getJob', () => {
123+
beforeEach(() => {
124+
vi.clearAllMocks()
125+
})
126+
127+
it('resolves a deterministic job ID through its Trigger.dev tag', async () => {
128+
mockRetrieveRun
129+
.mockRejectedValueOnce(new MockApiError(404, 'run not found'))
130+
.mockResolvedValueOnce({
131+
id: 'run-1',
132+
taskIdentifier: 'workflow-execution',
133+
payload: { workflowId: 'workflow-1' },
134+
status: 'COMPLETED',
135+
createdAt: '2026-08-05T12:00:00.000Z',
136+
finishedAt: '2026-08-05T12:00:05.000Z',
137+
attemptCount: 1,
138+
output: { output: { answer: 42 } },
139+
})
140+
mockListRuns.mockReturnValueOnce(
141+
(async function* () {
142+
yield { id: 'run-1' }
143+
})()
144+
)
145+
const queue = new TriggerDevJobQueue()
146+
147+
const job = await queue.getJob('workflow-execution:execution-1')
148+
149+
expect(mockListRuns).toHaveBeenCalledWith({
150+
tag: 'jobId:workflow-execution:execution-1',
151+
limit: 1,
152+
})
153+
expect(mockRetrieveRun).toHaveBeenNthCalledWith(2, 'run-1')
154+
expect(job).toMatchObject({
155+
id: 'workflow-execution:execution-1',
156+
status: 'completed',
157+
output: { output: { answer: 42 } },
158+
metadata: { workflowId: 'workflow-1' },
159+
})
160+
})
161+
})

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

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -189,7 +189,26 @@ export class TriggerDevJobQueue implements JobQueueBackend {
189189

190190
async getJob(jobId: string): Promise<Job | null> {
191191
try {
192-
const run = await runs.retrieve(jobId)
192+
let run: Awaited<ReturnType<typeof runs.retrieve>>
193+
try {
194+
run = await runs.retrieve(jobId)
195+
} catch (error) {
196+
const isNotFound =
197+
(error instanceof Error && error.message.toLowerCase().includes('not found')) ||
198+
(error && typeof error === 'object' && 'status' in error && error.status === 404)
199+
if (!isNotFound) throw error
200+
201+
let runId: string | undefined
202+
for await (const candidate of runs.list({ tag: `jobId:${jobId}`, limit: 1 })) {
203+
runId = candidate.id
204+
break
205+
}
206+
if (!runId) {
207+
logger.debug('Job not found in trigger.dev', { jobId })
208+
return null
209+
}
210+
run = await runs.retrieve(runId)
211+
}
193212

194213
const payload = run.payload as Record<string, unknown>
195214
const metadata: JobMetadata = {
@@ -270,6 +289,7 @@ function buildTags(options?: EnqueueOptions): string[] {
270289
const tags: string[] = []
271290
const meta = options?.metadata
272291

292+
if (options?.jobId) tags.push(`jobId:${options.jobId}`)
273293
if (meta?.workspaceId) tags.push(`workspaceId:${meta.workspaceId}`)
274294
if (meta?.workflowId) tags.push(`workflowId:${meta.workflowId}`)
275295
if (meta?.userId) tags.push(`userId:${meta.userId}`)

‎apps/sim/lib/workflows/executor/execution-status.test.ts‎

Lines changed: 50 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@ describe('getWorkflowExecutionStatus queue projection', () => {
3030
beforeEach(() => {
3131
vi.clearAllMocks()
3232
resetDbChainMock()
33-
queueTableRows(schemaMock.workflowExecutionLogs, [])
3433
})
3534

3635
it('projects a queued workflow job as an execution resource', async () => {
@@ -57,8 +56,9 @@ describe('getWorkflowExecutionStatus queue projection', () => {
5756
expect(mockGetJob).toHaveBeenCalledWith('workflow-execution:execution-1')
5857
})
5958

60-
it('uses the resume execution ID when the queued work is a resume attempt', async () => {
61-
mockGetJob.mockResolvedValueOnce(null).mockResolvedValueOnce({
59+
it('uses the resume entry ID when the queued work is a resume attempt', async () => {
60+
queueTableRows(schemaMock.resumeQueue, [{ id: 'resume-entry-1' }])
61+
mockGetJob.mockResolvedValueOnce({
6262
status: 'processing',
6363
createdAt: new Date('2026-08-05T12:00:00.000Z'),
6464
startedAt: new Date('2026-08-05T12:00:01.000Z'),
@@ -73,17 +73,56 @@ describe('getWorkflowExecutionStatus queue projection', () => {
7373
status: 'running',
7474
startedAt: '2026-08-05T12:00:01.000Z',
7575
})
76-
expect(mockGetJob).toHaveBeenNthCalledWith(2, 'resume-execution:execution-1')
76+
expect(mockGetJob).toHaveBeenCalledWith('resume-execution:resume-entry-1')
77+
})
78+
79+
it('projects an active resume ahead of the existing paused log', async () => {
80+
queueTableRows(schemaMock.workflowExecutionLogs, [
81+
{
82+
executionId: 'execution-1',
83+
workflowId: 'workflow-1',
84+
status: 'paused',
85+
},
86+
])
87+
queueTableRows(schemaMock.resumeQueue, [{ id: 'resume-entry-1' }])
88+
mockGetJob.mockResolvedValueOnce({
89+
status: 'pending',
90+
createdAt: new Date('2026-08-05T12:00:00.000Z'),
91+
metadata: { workflowId: 'workflow-1' },
92+
})
93+
94+
const status = await getWorkflowExecutionStatus(input)
95+
96+
expect(status).toMatchObject({
97+
executionId: 'execution-1',
98+
status: 'queued',
99+
paused: null,
100+
})
101+
})
102+
103+
it('returns completed queue output when requested', async () => {
104+
mockGetJob.mockResolvedValueOnce({
105+
status: 'completed',
106+
createdAt: new Date('2026-08-05T12:00:00.000Z'),
107+
completedAt: new Date('2026-08-05T12:00:05.000Z'),
108+
output: { output: { answer: 42 } },
109+
metadata: { workflowId: 'workflow-1' },
110+
})
111+
112+
const status = await getWorkflowExecutionStatus({ ...input, includeOutput: true })
113+
114+
expect(status).toMatchObject({
115+
status: 'completed',
116+
finalOutput: { answer: 42 },
117+
})
77118
})
78119

79120
it('does not expose a queue record belonging to another workflow', async () => {
80-
mockGetJob
81-
.mockResolvedValueOnce({
82-
status: 'pending',
83-
createdAt: new Date('2026-08-05T12:00:00.000Z'),
84-
metadata: { workflowId: 'workflow-2' },
85-
})
86-
.mockResolvedValueOnce(null)
121+
mockGetJob.mockResolvedValueOnce({
122+
status: 'pending',
123+
createdAt: new Date('2026-08-05T12:00:00.000Z'),
124+
metadata: { workflowId: 'workflow-2' },
125+
})
87126

88127
await expect(getWorkflowExecutionStatus(input)).resolves.toBeNull()
89128
})

‎apps/sim/lib/workflows/executor/execution-status.ts‎

Lines changed: 56 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,9 @@
11
import { db } from '@sim/db'
2-
import { pausedExecutions, workflowExecutionLogs } from '@sim/db/schema'
2+
import { pausedExecutions, resumeQueue, workflowExecutionLogs } from '@sim/db/schema'
33
import { and, eq } from 'drizzle-orm'
44
import type { WorkflowExecutionStatusResponse } from '@/lib/api/contracts/workflows'
55
import { getJobQueue } from '@/lib/core/async-jobs'
6+
import type { Job } from '@/lib/core/async-jobs/types'
67
import {
78
collectFunctionalBlockOutputs,
89
type FunctionalExecutionDataSource,
@@ -86,6 +87,38 @@ function extractError(executionData: unknown): string | null {
8687
return null
8788
}
8889

90+
function extractJobFinalOutput(output: unknown): unknown | null {
91+
if (!output || typeof output !== 'object' || !('output' in output)) return null
92+
return (output as Record<string, unknown>).output ?? null
93+
}
94+
95+
function projectQueueJob(
96+
job: Job,
97+
input: Pick<GetWorkflowExecutionStatusInput, 'executionId' | 'includeOutput' | 'workflowId'>
98+
): WorkflowExecutionStatusResponse {
99+
const status: WorkflowExecutionStatusResponse['status'] =
100+
job.status === 'pending' ? 'queued' : job.status === 'processing' ? 'running' : job.status
101+
const startedAt = job.startedAt ?? job.createdAt
102+
const endedAt = job.completedAt ?? null
103+
104+
return {
105+
executionId: input.executionId,
106+
workflowId: input.workflowId,
107+
status,
108+
trigger: job.metadata.correlation?.triggerType ?? 'api',
109+
level: status === 'failed' ? 'error' : 'info',
110+
startedAt: startedAt.toISOString(),
111+
endedAt: endedAt?.toISOString() ?? null,
112+
totalDurationMs: endedAt ? endedAt.getTime() - startedAt.getTime() : null,
113+
paused: null,
114+
cost: null,
115+
error: status === 'failed' ? (job.error ?? 'Execution failed') : null,
116+
finalOutput:
117+
input.includeOutput && status === 'completed' ? extractJobFinalOutput(job.output) : null,
118+
blockOutputs: null,
119+
}
120+
}
121+
89122
export interface GetWorkflowExecutionStatusInput {
90123
workflowId: string
91124
executionId: string
@@ -121,42 +154,34 @@ export async function getWorkflowExecutionStatus(
121154
)
122155
.limit(1)
123156

124-
if (!logRow) {
125-
const jobQueue = await getJobQueue()
126-
const jobIds = [
127-
`${WORKFLOW_EXECUTION_JOB_ID_PREFIX}${executionId}`,
128-
`${RESUME_EXECUTION_JOB_ID_PREFIX}${executionId}`,
129-
]
157+
const [activeResume] = await db
158+
.select({ id: resumeQueue.id })
159+
.from(resumeQueue)
160+
.where(
161+
and(
162+
eq(resumeQueue.parentExecutionId, executionId),
163+
eq(resumeQueue.newExecutionId, executionId),
164+
eq(resumeQueue.status, 'claimed')
165+
)
166+
)
167+
.limit(1)
168+
169+
const queueJobIds = [
170+
...(activeResume ? [`${RESUME_EXECUTION_JOB_ID_PREFIX}${activeResume.id}`] : []),
171+
...(!logRow ? [`${WORKFLOW_EXECUTION_JOB_ID_PREFIX}${executionId}`] : []),
172+
]
130173

131-
for (const jobId of jobIds) {
174+
if (queueJobIds.length > 0) {
175+
const jobQueue = await getJobQueue()
176+
for (const jobId of queueJobIds) {
132177
const job = await jobQueue.getJob(jobId)
133178
if (!job || job.metadata.workflowId !== workflowId) continue
134-
135-
const status: WorkflowExecutionStatusResponse['status'] =
136-
job.status === 'pending' ? 'queued' : job.status === 'processing' ? 'running' : job.status
137-
const startedAt = job.startedAt ?? job.createdAt
138-
const endedAt = job.completedAt ?? null
139-
140-
return {
141-
executionId,
142-
workflowId,
143-
status,
144-
trigger: job.metadata.correlation?.triggerType ?? 'api',
145-
level: status === 'failed' ? 'error' : 'info',
146-
startedAt: startedAt.toISOString(),
147-
endedAt: endedAt?.toISOString() ?? null,
148-
totalDurationMs: endedAt ? endedAt.getTime() - startedAt.getTime() : null,
149-
paused: null,
150-
cost: null,
151-
error: status === 'failed' ? (job.error ?? 'Execution failed') : null,
152-
finalOutput: null,
153-
blockOutputs: null,
154-
}
179+
return projectQueueJob(job, { executionId, includeOutput, workflowId })
155180
}
156-
157-
return null
158181
}
159182

183+
if (!logRow) return null
184+
160185
const [pausedRow] = await db
161186
.select({
162187
id: pausedExecutions.id,

0 commit comments

Comments
 (0)