fix(webhooks): skip duplicate in-progress deliveries and bound the polling claim lease

Author: waleedlatif1

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

@@ -32,7 +32,7 @@ const {
   mockResolveWebhookRecordProviderConfig,
   mockExecuteWorkflowCore,
   mockWasExecutionFinalizedByCore,
-  mockExecuteWithIdempotency,
+  mockExecuteOrSkipInProgress,
   mockGetProviderHandler,
   mockSetResolvedSecretTraceRegistry,
   mockExecutionSnapshot,
@@ -42,7 +42,7 @@ const {
     mockResolveWebhookRecordProviderConfig: vi.fn(),
     mockExecuteWorkflowCore: vi.fn(),
     mockWasExecutionFinalizedByCore: vi.fn(),
-    mockExecuteWithIdempotency: vi.fn(),
+    mockExecuteOrSkipInProgress: vi.fn(),
     mockGetProviderHandler: vi.fn(() => ({})),
     mockSetResolvedSecretTraceRegistry: vi.fn(),
     mockExecutionSnapshot: vi.fn(),
@@ -84,7 +84,7 @@ vi.mock('@/lib/workflows/executor/execution-core', () => ({
 vi.mock('@/lib/core/idempotency', () => ({
   IdempotencyService: { createWebhookIdempotencyKey: vi.fn(() => 'idempotency-key') },
   webhookIdempotency: {
-    executeWithIdempotency: mockExecuteWithIdempotency,
+    executeOrSkipInProgress: mockExecuteOrSkipInProgress,
   },
 }))
 
@@ -254,8 +254,11 @@ describe('executeWebhookJob fault vs error handling', () => {
       .mockResolvedValue(undefined)
     mockGetProviderHandler.mockReturnValue({})
     mockEnqueue.mockReset().mockResolvedValue('run_retry')
-    mockExecuteWithIdempotency.mockImplementation(
-      (_provider: string, _key: string, operation: () => Promise<unknown>) => operation()
+    mockExecuteOrSkipInProgress.mockImplementation(
+      async (_provider: string, _key: string, operation: () => Promise<unknown>) => ({
+        outcome: 'resolved',
+        result: await operation(),
+      })
     )
     executionPreprocessingMockFns.mockPreprocessExecution.mockResolvedValue({
       success: true,
@@ -634,14 +637,30 @@ describe('executeWebhookJob fault vs error handling', () => {
       workflowId: 'workflow-1',
       executionId: 'original-execution',
     }
-    mockExecuteWithIdempotency.mockResolvedValueOnce(cachedResult)
+    mockExecuteOrSkipInProgress.mockResolvedValueOnce({ outcome: 'resolved', result: cachedResult })
 
     await expect(executeWebhookJob(payload)).resolves.toBe(cachedResult)
 
     expect(executionPreprocessingMockFns.mockPreprocessExecution).not.toHaveBeenCalled()
     expect(mockReleaseExecutionSlot).toHaveBeenCalledWith('execution-1')
   })
 
+  it('acknowledges a duplicate of an in-progress delivery without running it and frees its reservation', async () => {
+    mockExecuteOrSkipInProgress.mockResolvedValueOnce({ outcome: 'in-progress' })
+
+    await expect(executeWebhookJob(payload)).resolves.toMatchObject({
+      success: true,
+      duplicate: true,
+      workflowId: 'workflow-1',
+      executionId: 'execution-1',
+    })
+
+    expect(executionPreprocessingMockFns.mockPreprocessExecution).not.toHaveBeenCalled()
+    expect(mockExecuteWorkflowCore).not.toHaveBeenCalled()
+    expect(mockEnqueue).not.toHaveBeenCalled()
+    expect(mockReleaseExecutionSlot).toHaveBeenCalledExactlyOnceWith('execution-1')
+  })
+
   it('rejects queued webhook work without an immutable attribution snapshot', async () => {
     await expect(
       executeWebhookJob({
@@ -705,7 +724,7 @@ describe('executeWebhookJob fault vs error handling', () => {
 
     expect(redisGet).toHaveBeenCalledWith('usage:reservation:execution-1')
     expect(result).toMatchObject({ success: false, requeued: true })
-    expect(mockExecuteWithIdempotency).not.toHaveBeenCalled()
+    expect(mockExecuteOrSkipInProgress).not.toHaveBeenCalled()
     expect(mockExecuteWorkflowCore).not.toHaveBeenCalled()
     expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
     expect(mockReleaseExecutionSlot).toHaveBeenCalledExactlyOnceWith('execution-1')
@@ -750,7 +769,7 @@ describe('executeWebhookJob fault vs error handling', () => {
     })
 
     expect(mockEnqueue).not.toHaveBeenCalled()
-    expect(mockExecuteWithIdempotency).not.toHaveBeenCalled()
+    expect(mockExecuteOrSkipInProgress).not.toHaveBeenCalled()
     expect(mockReleaseExecutionSlot).toHaveBeenCalledExactlyOnceWith('execution-1')
     expect(loggingSessionMockFns.mockSafeStart).toHaveBeenCalledTimes(1)
     expect(loggingSessionMockFns.mockSafeCompleteWithError).toHaveBeenCalledExactlyOnceWith(
@@ -784,12 +803,12 @@ describe('executeWebhookJob fault vs error handling', () => {
 
     expect(mockReleaseExecutionSlot).toHaveBeenCalledExactlyOnceWith('execution-1')
     expect(mockEnqueue).not.toHaveBeenCalled()
-    expect(mockExecuteWithIdempotency).not.toHaveBeenCalled()
+    expect(mockExecuteOrSkipInProgress).not.toHaveBeenCalled()
   })
 
   it('does not treat an ambiguous idempotency claim timeout as a safe setup retry', async () => {
     const error = new Error('Command timed out')
-    mockExecuteWithIdempotency.mockRejectedValueOnce(error)
+    mockExecuteOrSkipInProgress.mockRejectedValueOnce(error)
 
     await expect(executeWebhookJob(payload)).rejects.toBe(error)
 

apps/sim/background/webhook-execution.ts

@@ -589,7 +589,7 @@ export async function executeWebhookJob(
         )
       }
 
-      const result = await webhookIdempotency.executeWithIdempotency(
+      const execution = await webhookIdempotency.executeOrSkipInProgress(
         authenticatedPayload.provider,
         idempotencyKey,
         runOperation,
@@ -604,7 +604,26 @@ export async function executeWebhookJob(
       if (!operationStarted) {
         await releaseExecutionSlot(executionId)
       }
-      return result
+      if (execution.outcome === 'in-progress') {
+        // Ingress already acknowledged this delivery and nothing reads this job's result,
+        // so waiting on the live holder would only pin the machine and queue slot.
+        logger.info(`[${requestId}] Skipping duplicate webhook delivery already in progress`, {
+          webhookId: authenticatedPayload.webhookId,
+          workflowId: authenticatedPayload.workflowId,
+          provider: authenticatedPayload.provider,
+          executionId,
+        })
+        return {
+          success: true,
+          duplicate: true,
+          workflowId: authenticatedPayload.workflowId,
+          executionId,
+          output: {},
+          executedAt: new Date().toISOString(),
+          provider: authenticatedPayload.provider,
+        }
+      }
+      return execution.result
     })
   } catch (error) {
     await releaseExecutionSlot(executionId)

apps/sim/lib/core/idempotency/service.test.ts

@@ -1,10 +1,12 @@
+import { flushMicrotasks } from '@sim/testing/helpers/async'
 import { dbChainMockFns, resetDbChainMock } from '@sim/testing/mocks/database.mock'
 import { redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 
-const { redisDelMock, redisEvalMock } = vi.hoisted(() => ({
+const { redisDelMock, redisEvalMock, redisGetMock } = vi.hoisted(() => ({
   redisDelMock: vi.fn(),
   redisEvalMock: vi.fn(),
+  redisGetMock: vi.fn(),
 }))
 
 vi.mock('@/lib/core/storage', () => ({
@@ -18,11 +20,16 @@ vi.mock('@/lib/core/storage', () => ({
 import { RetryableSetupError } from '@/lib/core/errors/retryable-infrastructure'
 import {
   IdempotencyService,
+  pollingIdempotency,
   WEBHOOK_IN_PROGRESS_LEASE_SECONDS,
   webhookIdempotency,
 } from '@/lib/core/idempotency/service'
 
-redisConfigMockFns.mockGetRedisClient.mockReturnValue({ del: redisDelMock, eval: redisEvalMock })
+redisConfigMockFns.mockGetRedisClient.mockReturnValue({
+  del: redisDelMock,
+  eval: redisEvalMock,
+  get: redisGetMock,
+})
 
 const SEVEN_DAYS_SECONDS = 60 * 60 * 24 * 7
 
@@ -32,6 +39,8 @@ afterEach(() => {
 
 beforeEach(() => {
   resetDbChainMock()
+  redisEvalMock.mockReset()
+  redisGetMock.mockReset()
 })
 
 describe('IdempotencyService.createWebhookIdempotencyKey', () => {
@@ -223,26 +232,41 @@ describe('IdempotencyService in-progress deadlines', () => {
     expect(WEBHOOK_IN_PROGRESS_LEASE_SECONDS).toBeLessThan(SEVEN_DAYS_SECONDS)
   })
 
-  it('leases an untimed webhook claim for the bounded lease, not the seven-day dedupe window', async () => {
-    vi.useFakeTimers()
-    vi.setSystemTime(new Date('2026-08-03T12:00:00.000Z'))
-    redisEvalMock.mockResolvedValue([1, ''])
-
-    await webhookIdempotency.atomicallyClaim('gmail', 'wh_1:untimed-delivery')
-
-    const [, , redisKey, serialized, ttlSeconds] = redisEvalMock.mock.calls[0] as [
-      string,
-      number,
-      string,
-      string,
-      number,
-    ]
-    expect(redisKey).toBe('idempotency:webhook:gmail:wh_1:untimed-delivery')
-    expect(ttlSeconds).toBe(WEBHOOK_IN_PROGRESS_LEASE_SECONDS)
-    expect(JSON.parse(serialized).inProgressExpiresAt).toBe(
-      Date.now() + WEBHOOK_IN_PROGRESS_LEASE_SECONDS * 1000
-    )
-  })
+  it.each([
+    {
+      name: 'webhook',
+      service: webhookIdempotency,
+      leaseSeconds: WEBHOOK_IN_PROGRESS_LEASE_SECONDS,
+      dedupeSeconds: SEVEN_DAYS_SECONDS,
+    },
+    {
+      name: 'polling',
+      service: pollingIdempotency,
+      leaseSeconds: 5 * 60,
+      dedupeSeconds: 60 * 60 * 24 * 3,
+    },
+  ])(
+    'leases an untimed $name claim for its bounded lease so a crashed holder frees the key, not the dedupe window',
+    async ({ name, service, leaseSeconds, dedupeSeconds }) => {
+      vi.useFakeTimers()
+      vi.setSystemTime(new Date('2026-08-03T12:00:00.000Z'))
+      redisEvalMock.mockResolvedValue([1, ''])
+
+      await service.atomicallyClaim('gmail', 'wh_1:untimed-delivery')
+
+      const [, , redisKey, serialized, ttlSeconds] = redisEvalMock.mock.calls[0] as [
+        string,
+        number,
+        string,
+        string,
+        number,
+      ]
+      expect(redisKey).toBe(`idempotency:${name}:gmail:wh_1:untimed-delivery`)
+      expect(ttlSeconds).toBe(leaseSeconds)
+      expect(ttlSeconds).toBeLessThan(dedupeSeconds)
+      expect(JSON.parse(serialized).inProgressExpiresAt).toBe(Date.now() + leaseSeconds * 1000)
+    }
+  )
 
   it('keeps the seven-day dedupe window on a completed webhook result', async () => {
     redisEvalMock.mockResolvedValueOnce([1, '']).mockResolvedValueOnce(1)
@@ -305,6 +329,78 @@ describe('IdempotencyService in-progress deadlines', () => {
   })
 })
 
+describe('IdempotencyService duplicate of an in-progress operation', () => {
+  const liveClaim = () =>
+    JSON.stringify({
+      success: false,
+      status: 'in-progress',
+      startedAt: Date.now(),
+      inProgressExpiresAt: Date.now() + WEBHOOK_IN_PROGRESS_LEASE_SECONDS * 1000,
+      claimToken: 'other-holder',
+    })
+
+  it('returns in-progress at once instead of polling the live holder when skipping', async () => {
+    vi.useFakeTimers()
+    vi.setSystemTime(new Date('2026-08-03T12:00:00.000Z'))
+    const holder = liveClaim()
+    redisEvalMock.mockResolvedValueOnce([0, holder])
+    redisGetMock.mockResolvedValue(holder)
+    const operation = vi.fn()
+    const settled = vi.fn()
+
+    void webhookIdempotency
+      .executeOrSkipInProgress('gmail', 'wh_1:running-delivery', operation)
+      .then(settled, settled)
+    await flushMicrotasks(10)
+
+    expect(settled).toHaveBeenCalledExactlyOnceWith({ outcome: 'in-progress' })
+    expect(operation).not.toHaveBeenCalled()
+    expect(redisGetMock).not.toHaveBeenCalled()
+  })
+
+  it('still replays a completed result and rethrows a failed one when skipping', async () => {
+    const service = new IdempotencyService({ forceStorage: 'redis' })
+    redisEvalMock
+      .mockResolvedValueOnce([
+        0,
+        JSON.stringify({ success: true, status: 'completed', result: 'first-run' }),
+      ])
+      .mockResolvedValueOnce([
+        0,
+        JSON.stringify({ success: false, status: 'failed', error: 'first run failed' }),
+      ])
+
+    await expect(service.executeOrSkipInProgress('provider', 'done', vi.fn())).resolves.toEqual({
+      outcome: 'resolved',
+      result: 'first-run',
+    })
+    await expect(service.executeOrSkipInProgress('provider', 'failed', vi.fn())).rejects.toThrow(
+      'first run failed'
+    )
+  })
+
+  it('keeps waiting for the live holder by default, so a Stripe-style caller never acknowledges early', async () => {
+    vi.useFakeTimers()
+    vi.setSystemTime(new Date('2026-08-03T12:00:00.000Z'))
+    const service = new IdempotencyService({ forceStorage: 'redis' })
+    const holder = liveClaim()
+    redisEvalMock.mockResolvedValueOnce([0, holder])
+    redisGetMock
+      .mockResolvedValueOnce(holder)
+      .mockResolvedValueOnce(
+        JSON.stringify({ success: true, status: 'completed', result: 'holder-result' })
+      )
+    const settled = vi.fn()
+
+    void service.executeWithIdempotency('stripe', 'evt_1', vi.fn()).then(settled, settled)
+    await flushMicrotasks(10)
+    expect(settled).not.toHaveBeenCalled()
+
+    await vi.advanceTimersByTimeAsync(1_000)
+    expect(settled).toHaveBeenCalledExactlyOnceWith('holder-result')
+  })
+})
+
 describe('IdempotencyService retryable setup failures', () => {
   it('releases the claim instead of memoizing when the operation throws a RetryableSetupError', async () => {
     redisEvalMock.mockResolvedValueOnce([1, '']).mockResolvedValueOnce(1)