fix(webhooks): restore terminal log on failed requeue and make retry backoff abort-aware

Author: icecrasher321

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

@@ -661,7 +661,7 @@ describe('executeWebhookJob fault vs error handling', () => {
     expect(loggingSessionMockFns.mockSafeCompleteWithError).toHaveBeenCalled()
   })
 
-  it('faults the run when the requeue enqueue itself fails', async () => {
+  it('faults the run and restores the terminal log row when the requeue enqueue itself fails', async () => {
     executionPreprocessingMockFns.mockPreprocessExecution.mockResolvedValueOnce({
       success: false,
       error: {
@@ -675,5 +675,16 @@ describe('executeWebhookJob fault vs error handling', () => {
     await expect(executeWebhookJob(payload)).rejects.toThrow(
       'Internal error while fetching workflow'
     )
+
+    // The retry-bound attempt suppressed its failure row; a failed requeue means
+    // no retry will run, so the terminal row must be written before faulting.
+    expect(loggingSessionMockFns.mockSafeStart).toHaveBeenCalledWith(
+      expect.objectContaining({ userId: 'user-1', workspaceId: 'workspace-1' })
+    )
+    expect(loggingSessionMockFns.mockSafeCompleteWithError).toHaveBeenCalledWith(
+      expect.objectContaining({
+        error: expect.objectContaining({ message: 'Internal error while fetching workflow' }),
+      })
+    )
   })
 })

apps/sim/background/webhook-execution.ts

@@ -2,7 +2,7 @@ import { db } from '@sim/db'
 import { account, webhook } from '@sim/db/schema'
 import { createLogger, runWithRequestContext } from '@sim/logger'
 import { toError } from '@sim/utils/errors'
-import { sleep } from '@sim/utils/helpers'
+import { interruptibleSleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { isRecordLike } from '@sim/utils/object'
 import { backoffWithJitter } from '@sim/utils/retry'
@@ -377,11 +377,12 @@ async function requeueWebhookExecutionAfterSetupFailure(
       /**
        * The database backend executes jobs only through an in-process runner
        * and does not apply `delayMs` to it, so the runner sleeps out the
-       * backoff itself; the trigger.dev backend ignores this field and delays
+       * backoff itself (abort-aware, so cancellation and shutdown don't wait
+       * out the timer); the trigger.dev backend ignores this field and delays
        * server-side.
        */
       runner: async (_queuedPayload: unknown, signal: AbortSignal) => {
-        await sleep(delayMs)
+        await interruptibleSleep(delayMs, signal)
         if (signal.aborted) return undefined
         return executeWebhookJob(retryPayload, signal)
       },
@@ -418,6 +419,54 @@ async function requeueWebhookExecutionAfterSetupFailure(
   }
 }
 
+/**
+ * Restores the terminal failed execution-log row for a setup failure whose
+ * replacement enqueue failed. Attempts headed for a requeue suppress their
+ * failure row so the retry can reuse the execution id; once the requeue is
+ * known to have failed, no retry will run, so the row must be written here or
+ * the delivery faults without any execution record. Best-effort by design:
+ * the same infrastructure outage that broke setup may also break this write,
+ * in which case the faulted run remains the only signal — matching how
+ * preprocessing's own error logging degrades.
+ */
+async function recordSetupFailureWithoutRequeue(
+  payload: WebhookExecutionPayload,
+  correlation: AsyncExecutionCorrelation,
+  error: RetryableSetupError
+): Promise<void> {
+  try {
+    const loggingSession = new LoggingSession(
+      payload.workflowId,
+      correlation.executionId,
+      payload.provider,
+      correlation.requestId
+    )
+    await loggingSession.safeStart({
+      userId: payload.userId,
+      workspaceId: payload.workspaceId,
+      variables: {},
+      triggerData: { correlation },
+    })
+    await loggingSession.safeCompleteWithError({
+      error: {
+        message: error.message,
+        stackTrace: undefined,
+      },
+      traceSpans: [],
+      skipCost: true,
+    })
+  } catch (loggingError) {
+    logger.error(
+      `[${correlation.requestId}] Failed to record webhook setup failure after requeue failure`,
+      {
+        workflowId: payload.workflowId,
+        executionId: correlation.executionId,
+        error: loggingError,
+      }
+    )
+  }
+}
+
 export async function executeWebhookJob(
   payload: WebhookExecutionPayload,
   externalAbortSignal?: AbortSignal
@@ -513,23 +562,23 @@ export async function executeWebhookJob(
          * A typed setup failure certifies no block ran and the idempotency
          * claim was released, so requeueing the same delivery cannot double
          * run it; the retry re-admits usage and re-claims from scratch. When
-         * the requeue enqueue itself fails, fall through to the throw so the
-         * run fails loudly rather than dropping the delivery silently.
+         * the requeue enqueue itself fails, restore the terminal failure row
+         * the retry-bound attempt suppressed, then fall through to the throw
+         * so the run fails loudly rather than dropping the delivery silently.
          */
-        if (
-          isRetryableSetupError(error) &&
-          hasRemainingWebhookInfraRetry(payload) &&
-          (await requeueWebhookExecutionAfterSetupFailure(payload, correlation, error))
-        ) {
-          return {
-            success: false,
-            requeued: true,
-            workflowId: payload.workflowId,
-            executionId,
-            output: {},
-            executedAt: new Date().toISOString(),
-            provider: payload.provider,
+        if (isRetryableSetupError(error) && hasRemainingWebhookInfraRetry(payload)) {
+          if (await requeueWebhookExecutionAfterSetupFailure(payload, correlation, error)) {
+            return {
+              success: false,
+              requeued: true,
+              workflowId: payload.workflowId,
+              executionId,
+              output: {},
+              executedAt: new Date().toISOString(),
+              provider: payload.provider,
+            }
           }
+          await recordSetupFailureWithoutRequeue(payload, correlation, error)
         }
         throw error
       }

apps/sim/lib/core/rate-limiter/hosted-key/hosted-key-rate-limiter.ts

@@ -1,5 +1,5 @@
 import { createLogger } from '@sim/logger'
-import { sleep } from '@sim/utils/helpers'
+import { interruptibleSleep } from '@sim/utils/helpers'
 import { generateShortId } from '@sim/utils/id'
 import { getMaxExecutionTimeout } from '@/lib/core/execution-limits'
 import {
@@ -53,31 +53,6 @@ const MIN_QUEUE_RETRY_DELAY_MS = 50
  */
 const QUEUE_HEAD_POLL_MS = 200
 
-/**
- * Sleep for `ms`, resolving early if `signal` aborts. Cleans up its own timer and listener
- * so neither leaks. Callers don't need to distinguish an early (aborted) return from a normal
- * one — the surrounding wait loop re-checks its budget immediately after and bails when the
- * signal has fired. Falls back to a plain sleep when no signal is provided.
- */
-function interruptibleSleep(ms: number, signal?: AbortSignal): Promise<void> {
-  if (!signal) return sleep(ms)
-  if (signal.aborted) return Promise.resolve()
-  return new Promise<void>((resolve) => {
-    const onAbort = () => {
-      clearTimeout(timer)
-      signal.removeEventListener('abort', onAbort)
-      resolve()
-    }
-    const timer = setTimeout(() => {
-      signal.removeEventListener('abort', onAbort)
-      resolve()
-    }, ms)
-    signal.addEventListener('abort', onAbort, { once: true })
-    // Catch an abort that fired between the guard above and addEventListener.
-    if (signal.aborted) onAbort()
-  })
-}
-
 /**
  * Resolves env var names for a hosted-key prefix. Numbered pools use a
  * `{PREFIX}_COUNT` env var. Deployments that still provide one legacy singular

packages/utils/src/helpers.test.ts

@@ -2,7 +2,7 @@
  * @vitest-environment node
  */
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
-import { chunkArray, noop, sleep } from './helpers.js'
+import { chunkArray, interruptibleSleep, noop, sleep } from './helpers.js'
 
 describe('sleep', () => {
   beforeEach(() => {
@@ -30,6 +30,47 @@ describe('sleep', () => {
   })
 })
 
+describe('interruptibleSleep', () => {
+  beforeEach(() => {
+    vi.useFakeTimers()
+  })
+
+  afterEach(() => {
+    vi.useRealTimers()
+  })
+
+  it('resolves after the delay when no signal is provided', async () => {
+    const promise = interruptibleSleep(1000)
+    vi.advanceTimersByTime(1000)
+    await expect(promise).resolves.toBeUndefined()
+  })
+
+  it('resolves after the delay when the signal never aborts', async () => {
+    const controller = new AbortController()
+    const promise = interruptibleSleep(1000, controller.signal)
+    vi.advanceTimersByTime(1000)
+    await expect(promise).resolves.toBeUndefined()
+  })
+
+  it('resolves early when the signal aborts mid-sleep', async () => {
+    const controller = new AbortController()
+    let resolved = false
+    interruptibleSleep(60_000, controller.signal).then(() => {
+      resolved = true
+    })
+    vi.advanceTimersByTime(1)
+    controller.abort()
+    await Promise.resolve()
+    expect(resolved).toBe(true)
+  })
+
+  it('resolves immediately for an already-aborted signal', async () => {
+    const controller = new AbortController()
+    controller.abort()
+    await expect(interruptibleSleep(60_000, controller.signal)).resolves.toBeUndefined()
+  })
+})
+
 describe('noop', () => {
   it('is a function', () => {
     expect(typeof noop).toBe('function')

packages/utils/src/helpers.ts

@@ -6,6 +6,31 @@ export function sleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms))
 }
 
+/**
+ * Sleep for `ms`, resolving early if `signal` aborts. Cleans up its own timer and listener
+ * so neither leaks. Callers don't need to distinguish an early (aborted) return from a normal
+ * one — the surrounding wait loop re-checks its budget or the signal immediately after and
+ * bails when it has fired. Falls back to a plain sleep when no signal is provided.
+ */
+export function interruptibleSleep(ms: number, signal?: AbortSignal): Promise<void> {
+  if (!signal) return sleep(ms)
+  if (signal.aborted) return Promise.resolve()
+  return new Promise<void>((resolve) => {
+    const onAbort = () => {
+      clearTimeout(timer)
+      signal.removeEventListener('abort', onAbort)
+      resolve()
+    }
+    const timer = setTimeout(() => {
+      signal.removeEventListener('abort', onAbort)
+      resolve()
+    }, ms)
+    signal.addEventListener('abort', onAbort, { once: true })
+    // Catch an abort that fired between the guard above and addEventListener.
+    if (signal.aborted) onAbort()
+  })
+}
+
 /** No-operation function for use as default callback. */
 export const noop = () => {}