fix(webhooks): requeue deliveries dropped by retryable setup infrastructure failures

Author: icecrasher321

apps/sim/background/webhook-execution.test.ts

@@ -25,25 +25,32 @@ const {
   mockLoadDeploymentVersionState,
   mockGetProviderHandler,
   mockSetResolvedSecretTraceRegistry,
-} = vi.hoisted(() => ({
-  mockResolveWebhookRecordProviderConfig: vi.fn(),
-  mockExecuteWorkflowCore: vi.fn(),
-  mockWasExecutionFinalizedByCore: vi.fn(),
-  mockExecuteWithIdempotency: vi.fn(),
-  mockRefreshExecutionSlotExpiry: vi.fn().mockResolvedValue(true),
-  mockReleaseExecutionSlot: vi.fn(),
-  mockGetProviderHandler: vi.fn(() => ({})),
-  mockSetResolvedSecretTraceRegistry: vi.fn(),
-  mockLoadDeploymentVersionState: vi.fn(
-    async (_workflowId: string, deploymentVersionId: string) => ({
-      blocks: {},
-      edges: [],
-      loops: {},
-      parallels: {},
-      deploymentVersionId,
-    })
-  ),
-}))
+  mockEnqueue,
+  mockGetJobQueue,
+} = vi.hoisted(() => {
+  const mockEnqueue = vi.fn()
+  return {
+    mockResolveWebhookRecordProviderConfig: vi.fn(),
+    mockExecuteWorkflowCore: vi.fn(),
+    mockWasExecutionFinalizedByCore: vi.fn(),
+    mockExecuteWithIdempotency: vi.fn(),
+    mockRefreshExecutionSlotExpiry: vi.fn().mockResolvedValue(true),
+    mockReleaseExecutionSlot: vi.fn(),
+    mockGetProviderHandler: vi.fn(() => ({})),
+    mockSetResolvedSecretTraceRegistry: vi.fn(),
+    mockLoadDeploymentVersionState: vi.fn(
+      async (_workflowId: string, deploymentVersionId: string) => ({
+        blocks: {},
+        edges: [],
+        loops: {},
+        parallels: {},
+        deploymentVersionId,
+      })
+    ),
+    mockEnqueue,
+    mockGetJobQueue: vi.fn(async () => ({ enqueue: mockEnqueue })),
+  }
+})
 
 const mockGetEffectiveEnvironmentSnapshot =
   environmentUtilsMockFns.mockGetEffectiveEnvironmentSnapshot
@@ -105,6 +112,11 @@ vi.mock('@/lib/core/execution-limits', () => ({
   getExecutionDeadlineAt: vi.fn(() => new Date(Date.now() + 120_000)),
   getTimeoutErrorMessage: vi.fn(() => 'timed out'),
   RESERVATION_TTL_BUFFER_MS: 300_000,
+  toTriggerMaxDurationSeconds: vi.fn(() => undefined),
+}))
+
+vi.mock('@/lib/core/async-jobs', () => ({
+  getJobQueue: mockGetJobQueue,
 }))
 
 vi.mock('@/lib/workflows/executor/pause-persistence', () => ({
@@ -132,6 +144,7 @@ vi.mock('@/triggers', () => ({
   isTriggerValid: vi.fn(() => false),
 }))
 
+import { isRetryableSetupError } from '@/lib/core/errors/retryable-infrastructure'
 import {
   executeWebhookJob,
   resolveWebhookExecutionProviderConfig,
@@ -242,6 +255,7 @@ describe('executeWebhookJob fault vs error handling', () => {
       }
     })
     mockGetProviderHandler.mockReturnValue({})
+    mockEnqueue.mockResolvedValue('run_retry')
     mockExecuteWithIdempotency.mockImplementation(
       (_provider: string, _key: string, operation: () => Promise<unknown>) => operation()
     )
@@ -543,4 +557,123 @@ describe('executeWebhookJob fault vs error handling', () => {
 
     expect(executionPreprocessingMockFns.mockPreprocessExecution).not.toHaveBeenCalled()
   })
+
+  it('requeues the delivery when preprocessing fails on retryable infrastructure', async () => {
+    executionPreprocessingMockFns.mockPreprocessExecution.mockResolvedValueOnce({
+      success: false,
+      error: {
+        message: 'Internal error while fetching workflow',
+        statusCode: 500,
+        retryable: true,
+        cause: { code: 'CONNECT_TIMEOUT' },
+      },
+    })
+
+    const result = await executeWebhookJob(payload)
+
+    expect(result).toMatchObject({
+      success: false,
+      requeued: true,
+      workflowId: 'workflow-1',
+      executionId: 'execution-1',
+    })
+    expect(executionPreprocessingMockFns.mockPreprocessExecution).toHaveBeenCalledWith(
+      expect.objectContaining({ suppressRetryableFailureLogs: true })
+    )
+    expect(mockEnqueue).toHaveBeenCalledTimes(1)
+    const [jobType, retryPayload, options] = mockEnqueue.mock.calls[0]
+    expect(jobType).toBe('webhook-execution')
+    expect(retryPayload).toMatchObject({
+      webhookId: 'webhook-1',
+      workflowId: 'workflow-1',
+      executionId: 'execution-1',
+      requestId: 'request-1',
+      infraRetryCount: 1,
+    })
+    expect(options.delayMs).toBeGreaterThan(0)
+    // Database backend executes only through an in-process runner; trigger.dev ignores it.
+    expect(options.runner).toBeTypeOf('function')
+    expect(mockReleaseExecutionSlot).toHaveBeenCalledWith('execution-1')
+    expect(mockExecuteWorkflowCore).not.toHaveBeenCalled()
+    // No terminal failure row for an attempt that will be retried.
+    expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+  })
+
+  it('requeues on retryable infrastructure errors thrown by setup reads', async () => {
+    dbChainMockFns.limit.mockRejectedValueOnce(
+      Object.assign(new Error('write CONNECT_TIMEOUT'), { code: 'CONNECT_TIMEOUT' })
+    )
+
+    const result = await executeWebhookJob(payload)
+
+    expect(result).toMatchObject({ success: false, requeued: true })
+    expect(mockEnqueue).toHaveBeenCalledTimes(1)
+    expect(mockExecuteWorkflowCore).not.toHaveBeenCalled()
+    expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
+  })
+
+  it('faults the run without requeueing once the retry budget is exhausted', async () => {
+    executionPreprocessingMockFns.mockPreprocessExecution.mockResolvedValueOnce({
+      success: false,
+      error: {
+        message: 'Internal error while fetching workflow',
+        statusCode: 500,
+        retryable: true,
+      },
+    })
+
+    await expect(executeWebhookJob({ ...payload, infraRetryCount: 5 })).rejects.toSatisfy(
+      (error: unknown) => isRetryableSetupError(error)
+    )
+
+    expect(executionPreprocessingMockFns.mockPreprocessExecution).toHaveBeenCalledWith(
+      expect.objectContaining({ suppressRetryableFailureLogs: false })
+    )
+    expect(mockEnqueue).not.toHaveBeenCalled()
+    expect(mockReleaseExecutionSlot).toHaveBeenCalledWith('execution-1')
+  })
+
+  it('does not requeue non-retryable preprocessing failures', async () => {
+    executionPreprocessingMockFns.mockPreprocessExecution.mockResolvedValueOnce({
+      success: false,
+      error: { message: 'Usage limit exceeded', statusCode: 402 },
+    })
+
+    await expect(executeWebhookJob(payload)).rejects.toSatisfy(
+      (error: unknown) =>
+        !isRetryableSetupError(error) && (error as Error).message === 'Usage limit exceeded'
+    )
+
+    expect(mockEnqueue).not.toHaveBeenCalled()
+  })
+
+  it('never reclassifies infrastructure errors after the workflow core started', async () => {
+    const infraError = Object.assign(new Error('Connection terminated unexpectedly'), {
+      code: 'CONNECTION_CLOSED',
+    })
+    mockExecuteWorkflowCore.mockRejectedValue(infraError)
+    mockWasExecutionFinalizedByCore.mockReturnValue(false)
+
+    await expect(executeWebhookJob(payload)).rejects.toBe(infraError)
+
+    expect(mockEnqueue).not.toHaveBeenCalled()
+    // Post-core failures keep recording the terminal row.
+    expect(loggingSessionMockFns.mockSafeCompleteWithError).toHaveBeenCalled()
+  })
+
+  it('faults the run when the requeue enqueue itself fails', async () => {
+    executionPreprocessingMockFns.mockPreprocessExecution.mockResolvedValueOnce({
+      success: false,
+      error: {
+        message: 'Internal error while fetching workflow',
+        statusCode: 500,
+        retryable: true,
+      },
+    })
+    mockEnqueue.mockRejectedValueOnce(new Error('trigger api unavailable'))
+
+    await expect(executeWebhookJob(payload)).rejects.toThrow(
+      'Internal error while fetching workflow'
+    )
+  })
 })