fix(async): preserve execution correlation across queued runs

Author: PlaneInABottle

apps/sim/app/api/schedules/execute/route.test.ts

@@ -107,7 +107,11 @@ vi.mock('@sim/db', () => ({
   },
 }))
 
-import { GET } from '@/app/api/schedules/execute/route'
+vi.mock('uuid', () => ({
+  v4: vi.fn().mockReturnValue('schedule-execution-1'),
+}))
+
+import { GET } from './route'
 
 const SINGLE_SCHEDULE = [
   {
@@ -204,4 +208,44 @@ describe('Scheduled Workflow Execution API Route', () => {
     const data = await response.json()
     expect(data).toHaveProperty('executedCount', 2)
   })
+
+  it('should enqueue preassigned correlation metadata for schedules', async () => {
+    mockDbReturning.mockReturnValue(SINGLE_SCHEDULE)
+
+    const response = await GET(createMockRequest())
+
+    expect(response.status).toBe(200)
+    expect(mockEnqueue).toHaveBeenCalledWith(
+      'schedule-execution',
+      expect.objectContaining({
+        scheduleId: 'schedule-1',
+        workflowId: 'workflow-1',
+        executionId: 'schedule-execution-1',
+        requestId: 'test-request-id',
+        correlation: {
+          executionId: 'schedule-execution-1',
+          requestId: 'test-request-id',
+          source: 'schedule',
+          workflowId: 'workflow-1',
+          scheduleId: 'schedule-1',
+          triggerType: 'schedule',
+          scheduledFor: '2025-01-01T00:00:00.000Z',
+        },
+      }),
+      {
+        metadata: {
+          workflowId: 'workflow-1',
+          correlation: {
+            executionId: 'schedule-execution-1',
+            requestId: 'test-request-id',
+            source: 'schedule',
+            workflowId: 'workflow-1',
+            scheduleId: 'schedule-1',
+            triggerType: 'schedule',
+            scheduledFor: '2025-01-01T00:00:00.000Z',
+          },
+        },
+      }
+    )
+  })
 })

apps/sim/app/api/schedules/execute/route.ts

@@ -2,6 +2,7 @@ import { db, workflowDeploymentVersion, workflowSchedule } from '@sim/db'
 import { createLogger } from '@sim/logger'
 import { and, eq, isNull, lt, lte, not, or, sql } from 'drizzle-orm'
 import { type NextRequest, NextResponse } from 'next/server'
+import { v4 as uuidv4 } from 'uuid'
 import { verifyCronAuth } from '@/lib/auth/internal'
 import { getJobQueue, shouldExecuteInline } from '@/lib/core/async-jobs'
 import { generateRequestId } from '@/lib/core/utils/request'
@@ -57,10 +58,23 @@ export async function GET(request: NextRequest) {
 
     const queuePromises = dueSchedules.map(async (schedule) => {
       const queueTime = schedule.lastQueuedAt ?? queuedAt
+      const executionId = uuidv4()
+      const correlation = {
+        executionId,
+        requestId,
+        source: 'schedule' as const,
+        workflowId: schedule.workflowId,
+        scheduleId: schedule.id,
+        triggerType: 'schedule',
+        scheduledFor: schedule.nextRunAt?.toISOString(),
+      }
 
       const payload = {
         scheduleId: schedule.id,
         workflowId: schedule.workflowId,
+        executionId,
+        requestId,
+        correlation,
         blockId: schedule.blockId || undefined,
         cronExpression: schedule.cronExpression || undefined,
         lastRanAt: schedule.lastRanAt?.toISOString(),
@@ -71,7 +85,7 @@ export async function GET(request: NextRequest) {
 
       try {
         const jobId = await jobQueue.enqueue('schedule-execution', payload, {
-          metadata: { workflowId: schedule.workflowId },
+          metadata: { workflowId: schedule.workflowId, correlation },
         })
         logger.info(
           `[${requestId}] Queued schedule execution task ${jobId} for workflow ${schedule.workflowId}`

apps/sim/app/api/webhooks/trigger/[path]/route.test.ts

@@ -101,6 +101,7 @@ const {
   processWebhookMock,
   executeMock,
   getWorkspaceBilledAccountUserIdMock,
+  queueWebhookExecutionMock,
 } = vi.hoisted(() => ({
   generateRequestHashMock: vi.fn().mockResolvedValue('test-hash-123'),
   validateSlackSignatureMock: vi.fn().mockResolvedValue(true),
@@ -125,6 +126,10 @@ const {
     .mockImplementation(async (workspaceId: string | null | undefined) =>
       workspaceId ? 'test-user-id' : null
     ),
+  queueWebhookExecutionMock: vi.fn().mockImplementation(async () => {
+    const { NextResponse } = await import('next/server')
+    return NextResponse.json({ message: 'Webhook processed' })
+  }),
 }))
 
 vi.mock('@trigger.dev/sdk', () => ({
@@ -333,12 +338,7 @@ vi.mock('@/lib/webhooks/processor', () => ({
   }),
   shouldSkipWebhookEvent: vi.fn().mockReturnValue(false),
   handlePreDeploymentVerification: vi.fn().mockReturnValue(null),
-  queueWebhookExecution: vi.fn().mockImplementation(async () => {
-    // Call processWebhookMock so tests can verify it was called
-    processWebhookMock()
-    const { NextResponse } = await import('next/server')
-    return NextResponse.json({ message: 'Webhook processed' })
-  }),
+  queueWebhookExecution: queueWebhookExecutionMock,
 }))
 
 vi.mock('drizzle-orm/postgres-js', () => ({
@@ -353,7 +353,7 @@ vi.mock('@/lib/core/utils/request', () => requestUtilsMock)
 
 process.env.DATABASE_URL = 'postgresql://test:test@localhost:5432/test'
 
-import { POST } from '@/app/api/webhooks/trigger/[path]/route'
+import { POST } from './route'
 
 describe('Webhook Trigger API Route', () => {
   beforeEach(() => {
@@ -393,7 +393,7 @@ describe('Webhook Trigger API Route', () => {
 
     const params = Promise.resolve({ path: 'non-existent-path' })
 
-    const response = await POST(req, { params })
+    const response = await POST(req as any, { params })
 
     expect(response.status).toBe(404)
 
@@ -402,6 +402,32 @@ describe('Webhook Trigger API Route', () => {
   })
 
   describe('Generic Webhook Authentication', () => {
+    it('passes correlation-bearing request context into webhook queueing', async () => {
+      testData.webhooks.push({
+        id: 'generic-webhook-id',
+        provider: 'generic',
+        path: 'test-path',
+        isActive: true,
+        providerConfig: { requireAuth: false },
+        workflowId: 'test-workflow-id',
+      })
+
+      const req = createMockRequest('POST', { event: 'test', id: 'test-123' })
+      const params = Promise.resolve({ path: 'test-path' })
+
+      const response = await POST(req as any, { params })
+
+      expect(response.status).toBe(200)
+      expect(queueWebhookExecutionMock).toHaveBeenCalledOnce()
+      const call = queueWebhookExecutionMock.mock.calls[0]
+      expect(call[0]).toEqual(expect.objectContaining({ id: 'generic-webhook-id' }))
+      expect(call[1]).toEqual(expect.objectContaining({ id: 'test-workflow-id' }))
+      expect(call[2]).toEqual(expect.objectContaining({ event: 'test', id: 'test-123' }))
+      expect(call[4]).toEqual(
+        expect.objectContaining({ requestId: 'mock-request-id', path: 'test-path' })
+      )
+    })
+
     it('should process generic webhook without authentication', async () => {
       testData.webhooks.push({
         id: 'generic-webhook-id',
@@ -422,7 +448,7 @@ describe('Webhook Trigger API Route', () => {
       const req = createMockRequest('POST', { event: 'test', id: 'test-123' })
       const params = Promise.resolve({ path: 'test-path' })
 
-      const response = await POST(req, { params })
+      const response = await POST(req as any, { params })
 
       expect(response.status).toBe(200)
 
@@ -452,7 +478,7 @@ describe('Webhook Trigger API Route', () => {
       const req = createMockRequest('POST', { event: 'bearer.test' }, headers)
       const params = Promise.resolve({ path: 'test-path' })
 
-      const response = await POST(req, { params })
+      const response = await POST(req as any, { params })
 
       expect(response.status).toBe(200)
     })
@@ -483,7 +509,7 @@ describe('Webhook Trigger API Route', () => {
       const req = createMockRequest('POST', { event: 'custom.header.test' }, headers)
       const params = Promise.resolve({ path: 'test-path' })
 
-      const response = await POST(req, { params })
+      const response = await POST(req as any, { params })
 
       expect(response.status).toBe(200)
     })
@@ -518,7 +544,7 @@ describe('Webhook Trigger API Route', () => {
         const req = createMockRequest('POST', { event: 'case.test' }, headers)
         const params = Promise.resolve({ path: 'test-path' })
 
-        const response = await POST(req, { params })
+        const response = await POST(req as any, { params })
 
         expect(response.status).toBe(200)
       }
@@ -553,7 +579,7 @@ describe('Webhook Trigger API Route', () => {
         const req = createMockRequest('POST', { event: 'custom.case.test' }, headers)
         const params = Promise.resolve({ path: 'test-path' })
 
-        const response = await POST(req, { params })
+        const response = await POST(req as any, { params })
 
         expect(response.status).toBe(200)
       }
@@ -576,7 +602,7 @@ describe('Webhook Trigger API Route', () => {
       const req = createMockRequest('POST', { event: 'wrong.token.test' }, headers)
       const params = Promise.resolve({ path: 'test-path' })
 
-      const response = await POST(req, { params })
+      const response = await POST(req as any, { params })
 
       expect(response.status).toBe(401)
       expect(await response.text()).toContain('Unauthorized - Invalid authentication token')
@@ -604,7 +630,7 @@ describe('Webhook Trigger API Route', () => {
       const req = createMockRequest('POST', { event: 'wrong.custom.test' }, headers)
       const params = Promise.resolve({ path: 'test-path' })
 
-      const response = await POST(req, { params })
+      const response = await POST(req as any, { params })
 
       expect(response.status).toBe(401)
       expect(await response.text()).toContain('Unauthorized - Invalid authentication token')
@@ -624,7 +650,7 @@ describe('Webhook Trigger API Route', () => {
       const req = createMockRequest('POST', { event: 'no.auth.test' })
       const params = Promise.resolve({ path: 'test-path' })
 
-      const response = await POST(req, { params })
+      const response = await POST(req as any, { params })
 
       expect(response.status).toBe(401)
       expect(await response.text()).toContain('Unauthorized - Invalid authentication token')
@@ -652,7 +678,7 @@ describe('Webhook Trigger API Route', () => {
       const req = createMockRequest('POST', { event: 'exclusivity.test' }, headers)
       const params = Promise.resolve({ path: 'test-path' })
 
-      const response = await POST(req, { params })
+      const response = await POST(req as any, { params })
 
       expect(response.status).toBe(401)
       expect(await response.text()).toContain('Unauthorized - Invalid authentication token')
@@ -680,7 +706,7 @@ describe('Webhook Trigger API Route', () => {
       const req = createMockRequest('POST', { event: 'wrong.header.name.test' }, headers)
       const params = Promise.resolve({ path: 'test-path' })
 
-      const response = await POST(req, { params })
+      const response = await POST(req as any, { params })
 
       expect(response.status).toBe(401)
       expect(await response.text()).toContain('Unauthorized - Invalid authentication token')
@@ -705,7 +731,7 @@ describe('Webhook Trigger API Route', () => {
       const req = createMockRequest('POST', { event: 'no.token.config.test' }, headers)
       const params = Promise.resolve({ path: 'test-path' })
 
-      const response = await POST(req, { params })
+      const response = await POST(req as any, { params })
 
       expect(response.status).toBe(401)
       expect(await response.text()).toContain(