fix(webhooks): run a delivery whose dedup claim reply was lost or late

apps/sim/lib/core/idempotency/claim-faults.integration.ts

@@ -0,0 +1,213 @@
+/**
+ * The Redis claim under a lost or late reply, against real Redis. A TCP proxy between ioredis and
+ * Redis lets the claim script run and then drops the connection before its reply (ioredis replays
+ * the command on reconnect) or holds the reply past the command timeout. Skipped without
+ * `TEST_REDIS_URL`; the Redis client accessor is the only fixture.
+ */
+import { createServer, type Server, type Socket, connect as tcpConnect } from 'node:net'
+import { readTestRedisUrl } from '@sim/db/testing/test-infrastructure'
+import { redisConfigMock, redisConfigMockFns } from '@sim/testing/mocks/redis-config.mock'
+import { sleep } from '@sim/utils/helpers'
+import { generateId } from '@sim/utils/id'
+import Redis from 'ioredis'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+import { isRetryableSetupError } from '@/lib/core/errors/retryable-infrastructure'
+import { IdempotencyService } from '@/lib/core/idempotency/service'
+
+vi.mock('@/lib/core/config/redis', () => redisConfigMock)
+
+const redisUrl = readTestRedisUrl()
+const COMMAND_TIMEOUT_MS = 500
+const CLAIM_SCRIPT_MARKER = 'local claimed'
+
+type ClaimFault = { kind: 'drop-reply' } | { kind: 'delay-reply'; ms: number }
+
+/** Forwards bytes both ways; applies one armed fault to the next reply after a claim script. */
+class ClaimFaultProxy {
+  private server: Server | undefined
+  private armed: ClaimFault | undefined
+  claimsForwarded = 0
+
+  async start(upstream: URL): Promise<number> {
+    this.server = createServer((client) => this.pipe(client, upstream))
+    await new Promise<void>((resolve) => this.server?.listen(0, '127.0.0.1', resolve))
+    const address = this.server.address()
+    if (!address || typeof address === 'string') throw new Error('proxy has no port')
+    return address.port
+  }
+
+  arm(fault: ClaimFault): void {
+    this.armed = fault
+  }
+
+  async stop(): Promise<void> {
+    await new Promise((resolve) => this.server?.close(resolve))
+  }
+
+  private pipe(client: Socket, upstream: URL): void {
+    const server = tcpConnect(Number(upstream.port || 6379), upstream.hostname)
+    let pendingFault: ClaimFault | undefined
+    // Keeps the stream's tail so a marker that TCP split across chunks still matches.
+    let tail = ''
+    client.on('data', (chunk) => {
+      const seen = tail + chunk.toString('latin1')
+      tail = seen.slice(-(CLAIM_SCRIPT_MARKER.length - 1))
+      if (seen.includes(CLAIM_SCRIPT_MARKER)) {
+        this.claimsForwarded++
+        if (this.armed) {
+          pendingFault = this.armed
+          this.armed = undefined
+        }
+      }
+      server.write(chunk)
+    })
+    server.on('data', (chunk) => {
+      const fault = pendingFault
+      pendingFault = undefined
+      if (fault?.kind === 'drop-reply') {
+        client.destroy()
+        server.destroy()
+        return
+      }
+      if (fault?.kind === 'delay-reply') {
+        server.pause()
+        setTimeout(() => {
+          if (!client.destroyed) client.write(chunk)
+          server.resume()
+        }, fault.ms)
+        return
+      }
+      if (!client.destroyed) client.write(chunk)
+    })
+    client.on('close', () => server.destroy())
+    server.on('close', () => client.destroy())
+    client.on('error', () => server.destroy())
+    server.on('error', () => client.destroy())
+  }
+}
+
+describe.runIf(Boolean(redisUrl))('Redis idempotency claim with a lost or late reply', () => {
+  const proxy = new ClaimFaultProxy()
+  let redis: Redis
+  let direct: Redis
+  const namespace = `claim-faults-${generateId()}`
+  const service = new IdempotencyService({
+    namespace,
+    inProgressTtlSeconds: 60 * 60 * 2,
+    storeResultBody: false,
+    forceStorage: 'redis',
+  })
+
+  beforeAll(async () => {
+    const upstream = new URL(redisUrl ?? '')
+    const port = await proxy.start(upstream)
+    redis = new Redis({
+      host: '127.0.0.1',
+      port,
+      db: Number(upstream.pathname.slice(1) || 0),
+      lazyConnect: true,
+      commandTimeout: COMMAND_TIMEOUT_MS,
+      maxRetriesPerRequest: 5,
+      retryStrategy: () => 50,
+    })
+    await redis.connect()
+    direct = new Redis(redisUrl ?? '', { lazyConnect: true, maxRetriesPerRequest: 0 })
+    await direct.connect()
+  })
+
+  beforeEach(() => {
+    redisConfigMockFns.mockGetRedisClient.mockReturnValue(redis)
+  })
+
+  afterAll(async () => {
+    redis?.disconnect()
+    const keys = (await direct?.keys(`idempotency:${namespace}:*`)) ?? []
+    if (keys.length > 0) await direct.del(...keys)
+    await direct?.quit()
+    await proxy.stop()
+  })
+
+  async function storedStatus(identifier: string): Promise<string | undefined> {
+    const keys = await direct.keys(`idempotency:*:${identifier}`)
+    if (keys.length === 0) return undefined
+    const value = await direct.get(keys[0])
+    return value ? (JSON.parse(value) as { status?: string }).status : undefined
+  }
+
+  it('runs the operation once when the claim applied but its reply was lost', async () => {
+    const identifier = generateId()
+    const operation = vi.fn(async () => 'ran')
+    const claimsBefore = proxy.claimsForwarded
+
+    proxy.arm({ kind: 'drop-reply' })
+    const execution = await service.executeOrSkipInProgress('generic', identifier, operation)
+
+    expect(proxy.claimsForwarded - claimsBefore).toBe(2)
+    expect(execution).toEqual({ outcome: 'resolved', result: 'ran' })
+    expect(operation).toHaveBeenCalledTimes(1)
+    expect(await storedStatus(identifier)).toBe('completed')
+  })
+
+  it('releases its claim and asks for a retry when the claim reply is late, so a retry runs', async () => {
+    const identifier = generateId()
+    const operation = vi.fn(async () => 'ran')
+
+    proxy.arm({ kind: 'delay-reply', ms: COMMAND_TIMEOUT_MS * 1.5 })
+    const failure = await service.executeOrSkipInProgress('generic', identifier, operation).then(
+      () => undefined,
+      (error: unknown) => error
+    )
+    await sleep(COMMAND_TIMEOUT_MS * 2)
+
+    expect(isRetryableSetupError(failure)).toBe(true)
+    expect(operation).not.toHaveBeenCalled()
+    expect(await storedStatus(identifier)).toBeUndefined()
+
+    const retry = await service.executeOrSkipInProgress('generic', identifier, operation)
+    expect(retry).toEqual({ outcome: 'resolved', result: 'ran' })
+    expect(operation).toHaveBeenCalledTimes(1)
+  })
+
+  it('keeps the original error when the stall also outlasts the release', async () => {
+    const identifier = generateId()
+    const operation = vi.fn(async () => 'ran')
+
+    proxy.arm({ kind: 'delay-reply', ms: COMMAND_TIMEOUT_MS * 3 })
+    const failure = await service.executeOrSkipInProgress('generic', identifier, operation).then(
+      () => undefined,
+      (error: unknown) => error
+    )
+    await sleep(COMMAND_TIMEOUT_MS * 4)
+
+    expect(failure).toBeInstanceOf(Error)
+    expect(isRetryableSetupError(failure)).toBe(false)
+    expect(operation).not.toHaveBeenCalled()
+  })
+
+  it("still skips a delivery whose key is another holder's live claim", async () => {
+    const identifier = generateId()
+    let holderStarted!: () => void
+    const started = new Promise<void>((resolve) => {
+      holderStarted = resolve
+    })
+    let releaseHolder!: () => void
+    const released = new Promise<void>((resolve) => {
+      releaseHolder = resolve
+    })
+    const holder = vi.fn(async () => {
+      holderStarted()
+      await released
+      return 'holder'
+    })
+    const duplicate = vi.fn(async () => 'duplicate')
+
+    const running = service.executeOrSkipInProgress('generic', identifier, holder)
+    await started
+    const second = await service.executeOrSkipInProgress('generic', identifier, duplicate)
+    releaseHolder()
+
+    expect(second).toEqual({ outcome: 'in-progress' })
+    expect(duplicate).not.toHaveBeenCalled()
+    expect(await running).toEqual({ outcome: 'resolved', result: 'holder' })
+  })
+})

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

@@ -6,7 +6,10 @@ import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { and, eq, lt, or, sql } from 'drizzle-orm'
 import { getRedisClient } from '@/lib/core/config/redis'
-import { isRetryableSetupError } from '@/lib/core/errors/retryable-infrastructure'
+import {
+  isRetryableSetupError,
+  RetryableSetupError,
+} from '@/lib/core/errors/retryable-infrastructure'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import { getStorageMethod, type StorageMethod } from '@/lib/core/storage'
 import { extractProviderIdentifierFromBody, isDeliveryIdOptional } from '@/lib/webhooks/providers'
@@ -101,10 +104,16 @@ const REDIS_KEY_PREFIX = 'idempotency:'
 const MAX_WAIT_TIME_MS = getMaxExecutionTimeout()
 const POLL_INTERVAL_MS = 1000
 
+/**
+ * The claim value carries a fresh `claimToken`, so finding exactly `ARGV[1]` means this same
+ * call already applied: ioredis replays a command whose reply a dropped connection lost, and
+ * that replay must claim rather than report its own lease as another holder's.
+ */
 const CLAIM_REDIS_LEASE_SCRIPT = `
 local claimed = redis.call('SET', KEYS[1], ARGV[1], 'EX', tonumber(ARGV[2]), 'NX')
 if claimed then return {1, ''} end
 local current = redis.call('GET', KEYS[1])
+if current == ARGV[1] then return {1, ''} end
 if current then return {0, current} end
 claimed = redis.call('SET', KEYS[1], ARGV[1], 'EX', tonumber(ARGV[2]), 'NX')
 if claimed then return {1, ''} end
@@ -295,13 +304,19 @@ export class IdempotencyService {
       1,
       Math.ceil(((inProgressResult.inProgressExpiresAt ?? Date.now()) - Date.now()) / 1000)
     )
-    const claimResponse = await redis.eval(
-      CLAIM_REDIS_LEASE_SCRIPT,
-      1,
-      redisKey,
-      JSON.stringify(inProgressResult),
-      inProgressTtlSeconds
-    )
+    let claimResponse: unknown
+    try {
+      claimResponse = await redis.eval(
+        CLAIM_REDIS_LEASE_SCRIPT,
+        1,
+        redisKey,
+        JSON.stringify(inProgressResult),
+        inProgressTtlSeconds
+      )
+    } catch (error) {
+      if (!inProgressResult.claimToken) throw error
+      return this.releaseAmbiguousRedisClaim(normalizedKey, inProgressResult.claimToken, error)
+    }
     if (!Array.isArray(claimResponse) || claimResponse.length < 2) {
       throw new Error(`Redis idempotency claim returned an invalid result: ${normalizedKey}`)
     }
@@ -332,6 +347,37 @@ export class IdempotencyService {
     }
   }
 
+  /**
+   * A claim that errors (a timed-out reply, a connection that dropped after the script ran) may
+   * still hold the key, and nothing else would release it before the lease ends: every
+   * redelivery in that window would be skipped as a duplicate. Release this call's own claim by
+   * its token, then throw a `RetryableSetupError` so the caller re-attempts. If the release fails
+   * too, the claim may still be held, so the original error is rethrown unchanged.
+   */
+  private async releaseAmbiguousRedisClaim(
+    normalizedKey: string,
+    claimToken: string,
+    error: unknown
+  ): Promise<never> {
+    try {
+      await this.deleteKey(normalizedKey, 'redis', { claimToken })
+    } catch (releaseError) {
+      logger.error(`Failed to release an idempotency claim with an unknown outcome`, {
+        normalizedKey,
+        error: getErrorMessage(error),
+        releaseError: getErrorMessage(releaseError),
+      })
+      throw error
+    }
+    logger.warn(`Released an idempotency claim with an unknown outcome`, {
+      normalizedKey,
+      error: getErrorMessage(error),
+    })
+    throw new RetryableSetupError(`Idempotency claim failed: ${getErrorMessage(error)}`, {
+      cause: error,
+    })
+  }
+
   private async atomicallyClaimDb(
     normalizedKey: string,
     inProgressResult: ProcessingResult