From 4c35a7b097e8f27f931d4594dd27c4339bc55b1f Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Mon, 24 Aug 2026 11:10:55 -0700 Subject: [PATCH 1/2] fix(webhooks): requeue deliveries dropped by retryable setup infrastructure failures --- apps/sim/background/webhook-execution.test.ts | 171 +++++++++++++-- apps/sim/background/webhook-execution.ts | 196 +++++++++++++++++- apps/sim/lib/core/config/env.ts | 3 + .../errors/retryable-infrastructure.test.ts | 65 ++++++ .../core/errors/retryable-infrastructure.ts | 44 ++++ apps/sim/lib/core/idempotency/service.test.ts | 41 ++++ apps/sim/lib/core/idempotency/service.ts | 9 +- apps/sim/lib/execution/preprocessing.test.ts | 80 +++++++ apps/sim/lib/execution/preprocessing.ts | 97 +++++---- 9 files changed, 642 insertions(+), 64 deletions(-) create mode 100644 apps/sim/lib/core/errors/retryable-infrastructure.test.ts diff --git a/apps/sim/background/webhook-execution.test.ts b/apps/sim/background/webhook-execution.test.ts index 29baecab743..1cf18b1af9b 100644 --- a/apps/sim/background/webhook-execution.test.ts +++ b/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) => 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' + ) + }) }) diff --git a/apps/sim/background/webhook-execution.ts b/apps/sim/background/webhook-execution.ts index 94d23ceef50..132dbd7651f 100644 --- a/apps/sim/background/webhook-execution.ts +++ b/apps/sim/background/webhook-execution.ts @@ -2,8 +2,10 @@ 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 { generateId } from '@sim/utils/id' import { isRecordLike } from '@sim/utils/object' +import { backoffWithJitter } from '@sim/utils/retry' import { task, timeout } from '@trigger.dev/sdk' import { eq } from 'drizzle-orm' import { @@ -14,7 +16,15 @@ import { assertBillingAttributionSnapshot, type BillingAttributionSnapshot, } from '@/lib/billing/core/billing-attribution' +import { getJobQueue } from '@/lib/core/async-jobs' import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types' +import { env, envNumber } from '@/lib/core/config/env' +import { + describeRetryableInfrastructureError, + isRetryableInfrastructureError, + isRetryableSetupError, + RetryableSetupError, +} from '@/lib/core/errors/retryable-infrastructure' import { capExecutionTimeoutMs, createTimeoutAbortController, @@ -22,6 +32,7 @@ import { getExecutionDeadlineAt, getTimeoutErrorMessage, RESERVATION_TTL_BUFFER_MS, + toTriggerMaxDurationSeconds, } from '@/lib/core/execution-limits' import { IdempotencyService, @@ -289,6 +300,122 @@ export type WebhookExecutionPayload = { triggerTimestampMs?: number /** Trusted attempt budget resolved before the webhook enters the queue. */ executionTimeoutMs?: number + /** + * How many times this delivery was already requeued after a retryable + * infrastructure failure during setup (before any block ran). Absent on + * first delivery and on legacy queued jobs. + */ + infraRetryCount?: number +} + +const WEBHOOK_INFRA_RETRY_BASE_MS = envNumber(env.WEBHOOK_INFRA_RETRY_BASE_MS, 30_000, { + min: 1, + integer: true, +}) + +const WEBHOOK_INFRA_RETRY_MAX_MS = envNumber(env.WEBHOOK_INFRA_RETRY_MAX_MS, 5 * 60_000, { + min: 1, + integer: true, +}) + +/** Set to 0 to disable setup-failure requeues and restore fail-on-first-error behavior. */ +export const WEBHOOK_INFRA_RETRY_MAX_ATTEMPTS = envNumber(env.WEBHOOK_INFRA_RETRY_MAX_ATTEMPTS, 5, { + min: 0, + integer: true, +}) + +function hasRemainingWebhookInfraRetry(payload: WebhookExecutionPayload): boolean { + return (payload.infraRetryCount ?? 0) < WEBHOOK_INFRA_RETRY_MAX_ATTEMPTS +} + +/** Bounded, jittered delay for webhook setup-failure requeues. Attempt is 1-indexed. */ +function calculateWebhookInfraRetryDelayMs(retryAttempt: number): number { + return Math.min( + WEBHOOK_INFRA_RETRY_MAX_MS, + Math.round( + backoffWithJitter(retryAttempt, null, { + baseMs: WEBHOOK_INFRA_RETRY_BASE_MS, + maxMs: WEBHOOK_INFRA_RETRY_MAX_MS, + }) + ) + ) +} + +/** + * Re-enqueues a delivery whose setup failed on retryable infrastructure, + * preserving the execution identity (execution id, request id, idempotency + * inputs) so the retry is the same delivery, not a duplicate. Returns false + * when the enqueue itself fails, in which case the caller must surface the + * original error so the run fails loudly instead of losing the delivery + * silently. + */ +async function requeueWebhookExecutionAfterSetupFailure( + payload: WebhookExecutionPayload, + correlation: AsyncExecutionCorrelation, + error: RetryableSetupError +): Promise { + const retryAttempt = (payload.infraRetryCount ?? 0) + 1 + const delayMs = calculateWebhookInfraRetryDelayMs(retryAttempt) + + try { + const retryPayload: WebhookExecutionPayload = { + ...payload, + executionId: correlation.executionId, + requestId: correlation.requestId, + correlation, + infraRetryCount: retryAttempt, + } + const jobId = await (await getJobQueue()).enqueue('webhook-execution', retryPayload, { + delayMs, + metadata: { + workflowId: payload.workflowId, + workspaceId: payload.workspaceId, + userId: payload.userId, + correlation, + }, + maxDurationSeconds: toTriggerMaxDurationSeconds(payload.executionTimeoutMs), + /** + * 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 + * server-side. + */ + runner: async (_queuedPayload: unknown, signal: AbortSignal) => { + await sleep(delayMs) + if (signal.aborted) return undefined + return executeWebhookJob(retryPayload, signal) + }, + }) + + logger.warn( + `[${correlation.requestId}] Requeued webhook execution after retryable setup failure`, + { + workflowId: payload.workflowId, + webhookId: payload.webhookId, + executionId: correlation.executionId, + provider: payload.provider, + retryAttempt, + maxAttempts: WEBHOOK_INFRA_RETRY_MAX_ATTEMPTS, + delayMs, + jobId, + error: error.message, + cause: error.cause, + } + ) + return true + } catch (enqueueError) { + logger.error( + `[${correlation.requestId}] Failed to requeue webhook execution after setup failure`, + { + workflowId: payload.workflowId, + webhookId: payload.webhookId, + executionId: correlation.executionId, + retryAttempt, + error: enqueueError, + } + ) + return false + } } export async function executeWebhookJob( @@ -381,6 +508,29 @@ export async function executeWebhookJob( return result } catch (error) { await releaseExecutionSlot(executionId) + + /** + * 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. + */ + 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, + } + } throw error } }) @@ -424,7 +574,8 @@ export async function resolveWebhookExecutionProviderConfig< } catch (error) { const errorMessage = toError(error).message throw new Error( - `Failed to resolve webhook provider config for ${provider} webhook ${webhookRecord.id}: ${errorMessage}` + `Failed to resolve webhook provider config for ${provider} webhook ${webhookRecord.id}: ${errorMessage}`, + { cause: toError(error) } ) } } @@ -503,6 +654,7 @@ async function executeWebhookJobInternal( checkRateLimit: false, checkDeployment: false, skipUsageLimits: admissionCompleted, + suppressRetryableFailureLogs: hasRemainingWebhookInfraRetry(payload), workspaceId: payload.workspaceId, loggingSession, billingAttribution: payload.billingAttribution, @@ -511,7 +663,12 @@ async function executeWebhookJobInternal( }) if (!preprocessResult.success) { - throw new Error(preprocessResult.error?.message || 'Preprocessing failed in background job') + const failure = preprocessResult.error + const failureMessage = failure?.message || 'Preprocessing failed in background job' + if (failure && failure.statusCode >= 500 && failure.retryable === true) { + throw new RetryableSetupError(failureMessage, { cause: failure.cause }) + } + throw new Error(failureMessage) } const { actorUserId, billingAttribution, workflowRecord } = preprocessResult @@ -548,6 +705,15 @@ async function executeWebhookJobInternal( const workflowVariables = (workflowRecord.variables as Record) || {} let deploymentVersionId: string | undefined + /** + * Flipped immediately before `executeWorkflowCore` is invoked. While false, + * no block has run and no execution effect exists, so a retryable + * infrastructure error may be surfaced as a `RetryableSetupError` and the + * whole delivery safely re-attempted. Once true, errors are never + * reclassified as retryable — retrying after the executor started could + * double-run the workflow. + */ + let workflowCoreStarted = false try { const workflowStatePromise = payload.deploymentVersionId @@ -802,6 +968,7 @@ async function executeWebhookJobInternal( [] ) + workflowCoreStarted = true const executionResult = await executeWorkflowCore({ snapshot, callbacks: {}, @@ -839,6 +1006,28 @@ async function executeWebhookJobInternal( const errorMessage = toError(error).message const errorStack = error instanceof Error ? error.stack : undefined + /** + * Mirrors the schedule executor's setup boundary: an infrastructure error + * raised before the workflow core started left no execution effect, so it + * is surfaced as a `RetryableSetupError` — releasing the idempotency claim + * and, while attempts remain, requeueing without recording a terminal + * failed row for an attempt that will be retried. Exhausted retries fall + * through to normal failure handling but still throw typed so a provider + * redelivery is not rejected for a run that never happened. + */ + const retryableSetupCause = + !workflowCoreStarted && isRetryableInfrastructureError(error) + ? describeRetryableInfrastructureError(error) + : undefined + if (retryableSetupCause && hasRemainingWebhookInfraRetry(payload)) { + logger.warn(`[${requestId}] Retryable setup failure before webhook workflow started`, { + workflowId: payload.workflowId, + provider: payload.provider, + cause: retryableSetupCause, + }) + throw new RetryableSetupError(errorMessage, { cause: retryableSetupCause }) + } + logger.error( `[${requestId}] Webhook execution failed`, loggingSession.projectDiagnosticError(error, { @@ -906,6 +1095,9 @@ async function executeWebhookJobInternal( ) } + if (retryableSetupCause) { + throw new RetryableSetupError(errorMessage, { cause: retryableSetupCause }) + } throw error } } diff --git a/apps/sim/lib/core/config/env.ts b/apps/sim/lib/core/config/env.ts index cf5fa33c642..c39c3310ec5 100644 --- a/apps/sim/lib/core/config/env.ts +++ b/apps/sim/lib/core/config/env.ts @@ -345,6 +345,9 @@ export const env = createEnv({ SCHEDULE_INFRA_RETRY_BASE_MS: z.string().optional().default('60000'), SCHEDULE_INFRA_RETRY_MAX_MS: z.string().optional().default('300000'), SCHEDULE_INFRA_RETRY_MAX_ATTEMPTS: z.string().optional().default('10'), + WEBHOOK_INFRA_RETRY_BASE_MS: z.string().optional().default('30000'), + WEBHOOK_INFRA_RETRY_MAX_MS: z.string().optional().default('300000'), + WEBHOOK_INFRA_RETRY_MAX_ATTEMPTS: z.string().optional().default('5'), // Cloud Storage - AWS S3 AWS_REGION: z.string().optional(), // AWS region for S3 buckets diff --git a/apps/sim/lib/core/errors/retryable-infrastructure.test.ts b/apps/sim/lib/core/errors/retryable-infrastructure.test.ts new file mode 100644 index 00000000000..f2424908058 --- /dev/null +++ b/apps/sim/lib/core/errors/retryable-infrastructure.test.ts @@ -0,0 +1,65 @@ +/** + * @vitest-environment node + */ + +import { describe, expect, it } from 'vitest' +import { + describeRetryableInfrastructureError, + isRetryableInfrastructureError, + isRetryableSetupError, + RetryableSetupError, +} from '@/lib/core/errors/retryable-infrastructure' + +function errorWithCode(code: string): Error { + return Object.assign(new Error(`error ${code}`), { code }) +} + +describe('isRetryableInfrastructureError', () => { + it('recognizes postgres.js client connection failure codes', () => { + for (const code of [ + 'CONNECT_TIMEOUT', + 'CONNECTION_CLOSED', + 'CONNECTION_ENDED', + 'CONNECTION_DESTROYED', + ]) { + expect(isRetryableInfrastructureError(errorWithCode(code))).toBe(true) + } + }) + + it('recognizes node syscall and postgres server codes', () => { + expect(isRetryableInfrastructureError(errorWithCode('ECONNRESET'))).toBe(true) + expect(isRetryableInfrastructureError(errorWithCode('57P01'))).toBe(true) + }) + + it('walks the cause chain to find a retryable code', () => { + const wrapped = new Error('Failed to resolve webhook provider config', { + cause: errorWithCode('CONNECT_TIMEOUT'), + }) + expect(isRetryableInfrastructureError(wrapped)).toBe(true) + expect(describeRetryableInfrastructureError(wrapped)).toMatchObject({ + code: 'CONNECT_TIMEOUT', + }) + }) + + it('does not classify semantic SQL errors as retryable', () => { + expect(isRetryableInfrastructureError(errorWithCode('42703'))).toBe(false) + expect(isRetryableInfrastructureError(new Error('workflow not found'))).toBe(false) + expect(isRetryableInfrastructureError(undefined)).toBe(false) + }) +}) + +describe('RetryableSetupError', () => { + it('is recognized by its guard and preserves the cause', () => { + const cause = { code: 'CONNECT_TIMEOUT' } + const error = new RetryableSetupError('Internal error while fetching workflow', { cause }) + + expect(isRetryableSetupError(error)).toBe(true) + expect(error.name).toBe('RetryableSetupError') + expect(error.cause).toBe(cause) + }) + + it('does not match plain errors', () => { + expect(isRetryableSetupError(new Error('boom'))).toBe(false) + expect(isRetryableSetupError(undefined)).toBe(false) + }) +}) diff --git a/apps/sim/lib/core/errors/retryable-infrastructure.ts b/apps/sim/lib/core/errors/retryable-infrastructure.ts index 4eed69c6425..30ef8103986 100644 --- a/apps/sim/lib/core/errors/retryable-infrastructure.ts +++ b/apps/sim/lib/core/errors/retryable-infrastructure.ts @@ -35,6 +35,22 @@ const RETRYABLE_APP_ERROR_CODES = new Set([ 'CONNECTION_POOL_EXHAUSTED', ]) +/** + * postgres.js raises connection-level failures with its own `code` values + * rather than Node syscall codes: `CONNECT_TIMEOUT` when a connection cannot + * be established within `connect_timeout`, `CONNECTION_CLOSED` when the socket + * drops with queries pending, `CONNECTION_ENDED`/`CONNECTION_DESTROYED` when + * the pool is shut down or torn down underneath a query. All mean the query + * did not complete against a reachable server — the same class as + * `ECONNRESET`/`ETIMEDOUT` above. + */ +const RETRYABLE_PG_CLIENT_ERROR_CODES = new Set([ + 'CONNECT_TIMEOUT', + 'CONNECTION_CLOSED', + 'CONNECTION_ENDED', + 'CONNECTION_DESTROYED', +]) + function getErrorChain(error: unknown): Array> { const chain: Array> = [] let current: unknown = error @@ -58,6 +74,7 @@ export function describeRetryableInfrastructureError( (code && RETRYABLE_DB_ERROR_CODES.has(code)) || (code && RETRYABLE_NETWORK_ERROR_CODES.has(code)) || (code && RETRYABLE_APP_ERROR_CODES.has(code)) || + (code && RETRYABLE_PG_CLIENT_ERROR_CODES.has(code)) || (errno && RETRYABLE_NETWORK_ERROR_CODES.has(errno)) ) { return { @@ -76,3 +93,30 @@ export function describeRetryableInfrastructureError( export function isRetryableInfrastructureError(error: unknown): boolean { return Boolean(describeRetryableInfrastructureError(error)) } + +/** + * A retryable infrastructure failure raised strictly BEFORE the guarded + * operation performed any effect (no workflow block ran, no mutation + * committed). Throwing it is a contract: + * + * - the whole operation may be safely re-attempted from scratch, and + * - {@link IdempotencyService.executeWithIdempotency} releases the claim + * instead of memoizing the failure, so a re-attempt (or a provider + * redelivery) can claim and run rather than being rejected for the + * result TTL. + * + * Only construct one at a point where the "no effect yet" invariant is + * provable — e.g. behind a `workflowCoreStarted` guard. + */ +export class RetryableSetupError extends Error { + readonly isRetryableSetup = true as const + + constructor(message: string, options?: { cause?: unknown }) { + super(message, options) + this.name = 'RetryableSetupError' + } +} + +export function isRetryableSetupError(error: unknown): error is RetryableSetupError { + return error instanceof Error && (error as Partial).isRetryableSetup === true +} diff --git a/apps/sim/lib/core/idempotency/service.test.ts b/apps/sim/lib/core/idempotency/service.test.ts index de4aeea6eac..09efc7306b7 100644 --- a/apps/sim/lib/core/idempotency/service.test.ts +++ b/apps/sim/lib/core/idempotency/service.test.ts @@ -22,6 +22,7 @@ vi.mock('@/lib/core/storage', () => ({ resetStorageMethod: () => {}, })) +import { RetryableSetupError } from '@/lib/core/errors/retryable-infrastructure' import { IdempotencyService, WEBHOOK_IN_PROGRESS_LEASE_SECONDS, @@ -310,3 +311,43 @@ describe('IdempotencyService in-progress deadlines', () => { expect(condition.match(/::jsonb/g)).toHaveLength(2) }) }) + +describe('IdempotencyService retryable setup failures', () => { + it('releases the claim instead of memoizing when the operation throws a RetryableSetupError', async () => { + redisEvalMock.mockResolvedValueOnce([1, '']).mockResolvedValueOnce(1) + + const setupError = new RetryableSetupError('Internal error while fetching workflow') + await expect( + webhookIdempotency.executeWithIdempotency( + 'sim', + 'wh_1:setup-failure', + vi.fn().mockRejectedValue(setupError) + ) + ).rejects.toBe(setupError) + + const claimSerialized = JSON.parse((redisEvalMock.mock.calls[0] as unknown[])[3] as string) + const followUpCall = redisEvalMock.mock.calls[1] as unknown[] + // The follow-up must be the owner-fenced DELETE (release), not the failed-result store. + expect(followUpCall[0]).toContain("return redis.call('DEL', KEYS[1])") + expect(followUpCall[3]).toBe(claimSerialized.claimToken) + }) + + it('memoizes plain operation failures so duplicate deliveries stay rejected', async () => { + redisEvalMock.mockResolvedValueOnce([1, '']).mockResolvedValueOnce(1) + + await expect( + webhookIdempotency.executeWithIdempotency( + 'sim', + 'wh_1:plain-failure', + vi.fn().mockRejectedValue(new Error('handler blew up')) + ) + ).rejects.toThrow('handler blew up') + + const followUpCall = redisEvalMock.mock.calls[1] as unknown[] + expect(followUpCall[0]).toContain('SETEX') + expect(JSON.parse(followUpCall[5] as string)).toMatchObject({ + success: false, + status: 'failed', + }) + }) +}) diff --git a/apps/sim/lib/core/idempotency/service.ts b/apps/sim/lib/core/idempotency/service.ts index b3cbac52b25..aa885aa8e55 100644 --- a/apps/sim/lib/core/idempotency/service.ts +++ b/apps/sim/lib/core/idempotency/service.ts @@ -6,6 +6,7 @@ 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 { getMaxExecutionTimeout } from '@/lib/core/execution-limits' import { getStorageMethod, type StorageMethod } from '@/lib/core/storage' import { extractProviderIdentifierFromBody } from '@/lib/webhooks/providers' @@ -633,7 +634,13 @@ export class IdempotencyService { } catch (error) { const errorMessage = getErrorMessage(error, 'Unknown error') - if (this.config.retryFailures) { + /** + * A `RetryableSetupError` certifies the operation failed before any + * effect happened, so there is no outcome to memoize: release the claim + * so a re-attempt or provider redelivery can run instead of being + * rejected with the cached failure for the whole result TTL. + */ + if (this.config.retryFailures || isRetryableSetupError(error)) { await this.deleteKey( claimResult.normalizedKey, claimResult.storageMethod, diff --git a/apps/sim/lib/execution/preprocessing.test.ts b/apps/sim/lib/execution/preprocessing.test.ts index a39e4b500af..03f8a83c1f8 100644 --- a/apps/sim/lib/execution/preprocessing.test.ts +++ b/apps/sim/lib/execution/preprocessing.test.ts @@ -247,6 +247,86 @@ describe('preprocessExecution logPreprocessingErrors option', () => { }) }) +describe('preprocessExecution suppressRetryableFailureLogs option', () => { + const baseOptions = { + workflowId: 'workflow-1', + userId: 'owner-1', + triggerType: 'webhook' as const, + executionId: 'execution-1', + requestId: 'request-1', + checkDeployment: false, + checkRateLimit: false, + workspaceId: 'workspace-1', + } + + function makeLoggingSession() { + return { + safeStart: vi.fn().mockResolvedValue(true), + safeCompleteWithError: vi.fn().mockResolvedValue(undefined), + } + } + + it('skips the failure row for a retryable infrastructure failure', async () => { + workflowAuthzMockFns.mockGetActiveWorkflowRecord.mockRejectedValueOnce( + Object.assign(new Error('write CONNECT_TIMEOUT'), { code: 'CONNECT_TIMEOUT' }) + ) + const loggingSession = makeLoggingSession() + + const result = await preprocessExecution({ + ...baseOptions, + suppressRetryableFailureLogs: true, + loggingSession: loggingSession as any, + }) + + expect(result).toMatchObject({ + success: false, + error: { + message: 'Internal error while fetching workflow', + statusCode: 500, + retryable: true, + }, + }) + expect(loggingSession.safeStart).not.toHaveBeenCalled() + }) + + it('still records non-retryable failures while suppression is on', async () => { + workflowAuthzMockFns.mockGetActiveWorkflowRecord.mockRejectedValueOnce( + new Error('column "unknown" does not exist') + ) + const loggingSession = makeLoggingSession() + + const result = await preprocessExecution({ + ...baseOptions, + suppressRetryableFailureLogs: true, + loggingSession: loggingSession as any, + }) + + expect(result).toMatchObject({ + success: false, + error: { statusCode: 500, retryable: false }, + }) + expect(loggingSession.safeStart).toHaveBeenCalled() + }) + + it('records retryable failures when the option is absent', async () => { + workflowAuthzMockFns.mockGetActiveWorkflowRecord.mockRejectedValueOnce( + Object.assign(new Error('write CONNECT_TIMEOUT'), { code: 'CONNECT_TIMEOUT' }) + ) + const loggingSession = makeLoggingSession() + + const result = await preprocessExecution({ + ...baseOptions, + loggingSession: loggingSession as any, + }) + + expect(result).toMatchObject({ + success: false, + error: { statusCode: 500, retryable: true }, + }) + expect(loggingSession.safeStart).toHaveBeenCalled() + }) +}) + describe('preprocessExecution ban gate', () => { const baseOptions = { workflowId: 'workflow-1', diff --git a/apps/sim/lib/execution/preprocessing.ts b/apps/sim/lib/execution/preprocessing.ts index 84f41b41c41..c68dda80679 100644 --- a/apps/sim/lib/execution/preprocessing.ts +++ b/apps/sim/lib/execution/preprocessing.ts @@ -83,6 +83,14 @@ export interface PreprocessExecutionOptions { skipConcurrencyReservation?: boolean /** Skip execution-log error rows when the caller presents the failure itself. */ logPreprocessingErrors?: boolean + /** + * Skip execution-log error rows ONLY for retryable infrastructure failures + * (`statusCode >= 500 && retryable`). Set by callers that requeue such + * failures under the same execution id, so an attempt that will be retried + * does not leave a terminal failed row behind; the caller passes `false` + * again on its final attempt so an exhausted retry still records the row. + */ + suppressRetryableFailureLogs?: boolean workspaceId?: string loggingSession?: LoggingSession @@ -176,6 +184,7 @@ export async function preprocessExecution( skipUsageLimits = false, skipConcurrencyReservation = false, logPreprocessingErrors = true, + suppressRetryableFailureLogs = false, workspaceId: providedWorkspaceId, loggingSession: providedLoggingSession, triggerData, @@ -191,6 +200,10 @@ export async function preprocessExecution( const recordPreprocessingError: typeof logPreprocessingError = (args) => logPreprocessingErrors ? logPreprocessingError(args) : Promise.resolve() + /** True when this failure's log row is deferred to a caller that will requeue it. */ + const isFailureLogSuppressed = (failure: PreprocessExecutionError): boolean => + suppressRetryableFailureLogs && failure.statusCode >= 500 && failure.retryable === true + logger.info(`[${requestId}] Starting execution preprocessing`, { workflowId, userId, @@ -239,27 +252,27 @@ export async function preprocessExecution( } catch (error) { logger.error(`[${requestId}] Error fetching workflow`, { error, workflowId }) - await recordPreprocessingError({ - workflowId, - executionId, - triggerType, - requestId, - userId: userId || 'unknown', - workspaceId: providedWorkspaceId || '', - errorMessage: 'Internal error while fetching workflow', - loggingSession: providedLoggingSession, - triggerData, - }) - - return { - success: false, - error: { - message: 'Internal error while fetching workflow', - statusCode: 500, - retryable: isRetryableInfrastructureError(error), - cause: describeRetryableInfrastructureError(error), - }, + const failure: PreprocessExecutionError = { + message: 'Internal error while fetching workflow', + statusCode: 500, + retryable: isRetryableInfrastructureError(error), + cause: describeRetryableInfrastructureError(error), } + if (!isFailureLogSuppressed(failure)) { + await recordPreprocessingError({ + workflowId, + executionId, + triggerType, + requestId, + userId: userId || 'unknown', + workspaceId: providedWorkspaceId || '', + errorMessage: 'Internal error while fetching workflow', + loggingSession: providedLoggingSession, + triggerData, + }) + } + + return { success: false, error: failure } } } else if (workflowRecord.archivedAt) { logger.warn(`[${requestId}] Prefetched workflow is archived: ${workflowId}`) @@ -381,27 +394,27 @@ export async function preprocessExecution( } catch (error) { logger.error(`[${requestId}] Error resolving billing attribution`, { error, workflowId }) const errorLogUserId = userId || 'unknown' - await recordPreprocessingError({ - workflowId, - executionId, - triggerType, - requestId, - userId: errorLogUserId, - workspaceId, - errorMessage: BILLING_ERROR_MESSAGES.BILLING_ERROR_GENERIC, - loggingSession: providedLoggingSession, - triggerData, - }) - - return { - success: false, - error: { - message: 'Error resolving billing account', - statusCode: 500, - retryable: isRetryableInfrastructureError(error), - cause: describeRetryableInfrastructureError(error), - }, + const failure: PreprocessExecutionError = { + message: 'Error resolving billing account', + statusCode: 500, + retryable: isRetryableInfrastructureError(error), + cause: describeRetryableInfrastructureError(error), } + if (!isFailureLogSuppressed(failure)) { + await recordPreprocessingError({ + workflowId, + executionId, + triggerType, + requestId, + userId: errorLogUserId, + workspaceId, + errorMessage: BILLING_ERROR_MESSAGES.BILLING_ERROR_GENERIC, + loggingSession: providedLoggingSession, + triggerData, + }) + } + + return { success: false, error: failure } } const plan = billingAttribution.payerSubscription?.plan as SubscriptionPlan | undefined @@ -700,7 +713,7 @@ export async function preprocessExecution( const readGateFailure = banFailure ?? usageResult.failure if (readGateFailure) { - if (readGateFailure.recordError) { + if (readGateFailure.recordError && !isFailureLogSuppressed(readGateFailure.response.error)) { await recordPreprocessingError(readGateFailure.recordError) } return readGateFailure.response @@ -708,7 +721,7 @@ export async function preprocessExecution( const rateLimitFailure = await runRateLimitGate() if (rateLimitFailure) { - if (rateLimitFailure.recordError) { + if (rateLimitFailure.recordError && !isFailureLogSuppressed(rateLimitFailure.response.error)) { await recordPreprocessingError(rateLimitFailure.recordError) } return rateLimitFailure.response From 95e763ecb6ccac7c52db11ca93674629a062569d Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Mon, 24 Aug 2026 11:21:22 -0700 Subject: [PATCH 2/2] fix(webhooks): restore terminal log on failed requeue and make retry backoff abort-aware --- apps/sim/background/webhook-execution.test.ts | 13 ++- apps/sim/background/webhook-execution.ts | 85 +++++++++++++++---- .../hosted-key/hosted-key-rate-limiter.ts | 27 +----- packages/utils/src/helpers.test.ts | 43 +++++++++- packages/utils/src/helpers.ts | 25 ++++++ 5 files changed, 147 insertions(+), 46 deletions(-) diff --git a/apps/sim/background/webhook-execution.test.ts b/apps/sim/background/webhook-execution.test.ts index 1cf18b1af9b..822e908413a 100644 --- a/apps/sim/background/webhook-execution.test.ts +++ b/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' }), + }) + ) }) }) diff --git a/apps/sim/background/webhook-execution.ts b/apps/sim/background/webhook-execution.ts index 132dbd7651f..b758c54932f 100644 --- a/apps/sim/background/webhook-execution.ts +++ b/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 { + 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 } diff --git a/apps/sim/lib/core/rate-limiter/hosted-key/hosted-key-rate-limiter.ts b/apps/sim/lib/core/rate-limiter/hosted-key/hosted-key-rate-limiter.ts index 2638a385b21..9289e2bd155 100644 --- a/apps/sim/lib/core/rate-limiter/hosted-key/hosted-key-rate-limiter.ts +++ b/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 { - if (!signal) return sleep(ms) - if (signal.aborted) return Promise.resolve() - return new Promise((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 diff --git a/packages/utils/src/helpers.test.ts b/packages/utils/src/helpers.test.ts index b967117bacf..cf7f83ee5fa 100644 --- a/packages/utils/src/helpers.test.ts +++ b/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') diff --git a/packages/utils/src/helpers.ts b/packages/utils/src/helpers.ts index cbc8337b26f..cd1fb98d64e 100644 --- a/packages/utils/src/helpers.ts +++ b/packages/utils/src/helpers.ts @@ -6,6 +6,31 @@ export function sleep(ms: number): Promise { 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 { + if (!signal) return sleep(ms) + if (signal.aborted) return Promise.resolve() + return new Promise((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 = () => {}