fix(execution): queued execution finalization and async correlation (#…

…3535) * fix(execution): finalize runs before wrapper recovery * fix(async): preserve execution correlation across queued runs * fix(webhooks): pass correlation into preprocessing * style(webhooks): normalize webhook executor formatting * fix(async): avoid pre-starting queued execution logs Let executeWorkflowCore own normal-path logging start so queued workflow and schedule executions persist the richer deployment and environment metadata instead of an earlier placeholder start record. * fix(async): harden execution finalization guards Prevent leaked core finalization markers from accumulating while keeping outer recovery paths idempotent. Preserve best-effort logging completion by reusing settled completion promises instead of reopening duplicate terminal writes. * fix(async): preserve outcomes during cleanup Keep execution finalization cleanup best-effort so cancellation cleanup failures do not overwrite successful or failed outcomes. Restore webhook processor formatting to the repository Biome style to avoid noisy formatter churn. * fix(async): keep execution finalization state consistent Retry minimal logging for early failures, only mark core finalization after a log row actually completes, and let paused completions fall back cleanly. * fix(async): clean stale finalization guards Scan all finalized execution ids during TTL cleanup so refreshed keys cannot keep expired guards alive, and cover the reused-id ordering regression. * fix(async): retry failed error finalization Allow error finalization to retry after a non-error completion and fallback both fail, and always persist failed/error semantics for completeWithError. * fix(webhooks): reuse preprocessing execution ids Thread preprocessing execution identity into queued webhook execution so both phases share the same correlation and logs. --------- Co-authored-by: test <test@example.com>
simstudioai · waleedlatif1 · Mar 17, 2026 · Mar 13, 2026 · Mar 13, 2026 · Mar 14, 2026
commit 92290029f09b05638850dfaccfec14c825b43376
diff --git a/apps/sim/app/api/schedules/execute/route.test.ts b/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',
+          },
+        },
+      }
+    )
+  })
 })
diff --git a/apps/sim/app/api/schedules/execute/route.ts b/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}`

diff --git a/apps/sim/app/api/webhooks/trigger/[path]/route.test.ts b/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', () => ({
@@ -350,21 +355,28 @@ vi.mock('@/lib/webhooks/processor', () => ({
         return null
       }
     ),
-  checkWebhookPreprocessing: vi
-    .fn()
-    .mockResolvedValue({ error: null, actorUserId: 'test-user-id' }),
+  checkWebhookPreprocessing: vi.fn().mockResolvedValue({
+    error: null,
+    actorUserId: 'test-user-id',
+    executionId: 'preprocess-execution-id',
+    correlation: {
+      executionId: 'preprocess-execution-id',
+      requestId: 'mock-request-id',
+      source: 'webhook',
+      workflowId: 'test-workflow-id',
+      webhookId: 'generic-webhook-id',
+      path: 'test-path',
+      provider: 'generic',
+      triggerType: 'webhook',
+    },
+  }),
   formatProviderErrorResponse: vi.fn().mockImplementation((_webhook, error, status) => {
     const { NextResponse } = require('next/server')
     return NextResponse.json({ error }, { status })
   }),
   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', () => ({
@@ -419,7 +431,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)
 
@@ -494,6 +506,47 @@ 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',
+          actorUserId: 'test-user-id',
+          executionId: 'preprocess-execution-id',
+          correlation: {
+            executionId: 'preprocess-execution-id',
+            requestId: 'mock-request-id',
+            source: 'webhook',
+            workflowId: 'test-workflow-id',
+            webhookId: 'generic-webhook-id',
+            path: 'test-path',
+            provider: 'generic',
+            triggerType: 'webhook',
+          },
+        })
+      )
+    })
+
     it('should process generic webhook without authentication', async () => {
       testData.webhooks.push({
         id: 'generic-webhook-id',
@@ -514,7 +567,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)
 
@@ -544,7 +597,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)
     })
@@ -575,7 +628,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)
     })
@@ -610,7 +663,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)
       }
@@ -645,7 +698,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)
       }
@@ -668,7 +721,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')
@@ -696,7 +749,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')
@@ -716,7 +769,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')
@@ -744,7 +797,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')
@@ -772,7 +825,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')
@@ -797,7 +850,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(

diff --git a/apps/sim/app/api/webhooks/trigger/[path]/route.ts b/apps/sim/app/api/webhooks/trigger/[path]/route.ts
@@ -144,6 +144,8 @@ export async function POST(
       requestId,
       path,
       actorUserId: preprocessResult.actorUserId,
+      executionId: preprocessResult.executionId,
+      correlation: preprocessResult.correlation,
     })
     responses.push(response)
   }