fix(execution): retry transient database failures during execution setup (#7681)

* fix(execution): retry transient database failures during execution setup

A dropped Postgres connection during workflow execution setup killed the
run permanently. The first read in preprocessing is the workflow fetch; an
ECONNRESET there surfaced as "Internal error while fetching workflow", and
because background executions run with maxAttempts 1 there was no retry.

Route the read-only setup operations through the existing
withDatabaseReadRetry helper so a dropped connection is retried in place,
before any effect exists. The retried operations are all reads, so the
rate-limit token debit and the concurrency reservation are never re-entered.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BYi7yz8qo98ziQWZmRpqb8

* test(execution): type the logging-session factory instead of casting

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BYi7yz8qo98ziQWZmRpqb8

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: waleedlatif1

apps/sim/lib/db/read-retry.ts

@@ -38,14 +38,20 @@ export function isTransientDatabaseReadError(error: unknown): boolean {
  * rebuild the query outside any transaction and must have no side effects or locks.
  * A failed connection cannot establish whether a write committed, so writes and
  * transactions must never use this helper.
+ *
+ * `label` names the read in the retry log line so callers that wrap several can tell which flapped.
  */
-export async function withDatabaseReadRetry<T>(read: () => Promise<T>): Promise<T> {
+export async function withDatabaseReadRetry<T>(
+  read: () => Promise<T>,
+  options: { label?: string } = {}
+): Promise<T> {
   for (let attempt = 1; ; attempt++) {
     try {
       return await read()
     } catch (error) {
       if (attempt >= 3 || !isTransientDatabaseReadError(error)) throw error
       logger.warn('Retrying transient database read', {
+        label: options.label,
         attempt,
         code: getPostgresErrorCode(error),
       })

apps/sim/lib/execution/preprocessing.test.ts

@@ -5,15 +5,18 @@
 import { loggingSessionMock, workflowAuthzMockFns } from '@sim/testing'
 import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
 import { ADMISSION_ERROR_CODE } from '@/lib/core/admission/transient-failure'
+import type { LoggingSession } from '@/lib/logs/execution/logging-session'
 
 const {
+  mockSleep,
   mockCheckAttributedUsageLimits,
   mockCheckRateLimit,
   mockGetActivelyBannedUserIds,
   mockReserveExecutionSlot,
   mockResolveBillingAttribution,
   mockResolveSystemBillingAttribution,
 } = vi.hoisted(() => ({
+  mockSleep: vi.fn().mockResolvedValue(undefined),
   mockCheckAttributedUsageLimits: vi.fn(),
   mockCheckRateLimit: vi.fn(),
   mockGetActivelyBannedUserIds: vi.fn().mockResolvedValue([]),
@@ -22,6 +25,9 @@ const {
   mockResolveSystemBillingAttribution: vi.fn(),
 }))
 
+vi.mock('@sim/utils/helpers', () => ({
+  sleep: mockSleep,
+}))
 vi.mock('@/lib/auth/ban', () => ({
   getActivelyBannedUserIds: mockGetActivelyBannedUserIds,
 }))
@@ -248,6 +254,10 @@ describe('preprocessExecution logPreprocessingErrors option', () => {
 })
 
 describe('preprocessExecution suppressRetryableFailureLogs option', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
   const baseOptions = {
     workflowId: 'workflow-1',
     userId: 'owner-1',
@@ -266,16 +276,21 @@ describe('preprocessExecution suppressRetryableFailureLogs option', () => {
     }
   }
 
+  /** Preprocessing only reaches `safeStart`/`safeCompleteWithError`, so the mock stands in for the full session. */
+  function asLoggingSession(session: ReturnType<typeof makeLoggingSession>): LoggingSession {
+    return session as unknown as LoggingSession
+  }
+
   it('skips the failure row for a retryable infrastructure failure', async () => {
-    workflowAuthzMockFns.mockGetActiveWorkflowRecord.mockRejectedValueOnce(
+    workflowAuthzMockFns.mockGetActiveWorkflowRecord.mockRejectedValue(
       Object.assign(new Error('write CONNECT_TIMEOUT'), { code: 'CONNECT_TIMEOUT' })
     )
     const loggingSession = makeLoggingSession()
 
     const result = await preprocessExecution({
       ...baseOptions,
       suppressRetryableFailureLogs: true,
-      loggingSession: loggingSession as any,
+      loggingSession: asLoggingSession(loggingSession),
     })
 
     expect(result).toMatchObject({
@@ -298,7 +313,7 @@ describe('preprocessExecution suppressRetryableFailureLogs option', () => {
     const result = await preprocessExecution({
       ...baseOptions,
       suppressRetryableFailureLogs: true,
-      loggingSession: loggingSession as any,
+      loggingSession: asLoggingSession(loggingSession),
     })
 
     expect(result).toMatchObject({
@@ -308,15 +323,32 @@ describe('preprocessExecution suppressRetryableFailureLogs option', () => {
     expect(loggingSession.safeStart).toHaveBeenCalled()
   })
 
+  it('retries the workflow fetch before surfacing a transient failure', async () => {
+    workflowAuthzMockFns.mockGetActiveWorkflowRecord.mockRejectedValue(
+      Object.assign(new Error('read ECONNRESET'), { code: 'ECONNRESET' })
+    )
+
+    const result = await preprocessExecution({
+      ...baseOptions,
+      loggingSession: asLoggingSession(makeLoggingSession()),
+    })
+
+    expect(workflowAuthzMockFns.mockGetActiveWorkflowRecord).toHaveBeenCalledTimes(3)
+    expect(result).toMatchObject({
+      success: false,
+      error: { message: 'Internal error while fetching workflow', retryable: true },
+    })
+  })
+
   it('records retryable failures when the option is absent', async () => {
-    workflowAuthzMockFns.mockGetActiveWorkflowRecord.mockRejectedValueOnce(
+    workflowAuthzMockFns.mockGetActiveWorkflowRecord.mockRejectedValue(
       Object.assign(new Error('write CONNECT_TIMEOUT'), { code: 'CONNECT_TIMEOUT' })
     )
     const loggingSession = makeLoggingSession()
 
     const result = await preprocessExecution({
       ...baseOptions,
-      loggingSession: loggingSession as any,
+      loggingSession: asLoggingSession(loggingSession),
     })
 
     expect(result).toMatchObject({

apps/sim/lib/execution/preprocessing.ts

@@ -35,6 +35,7 @@ import {
 } from '@/lib/core/execution-limits/metrics'
 import { RateLimiter } from '@/lib/core/rate-limiter/rate-limiter'
 import type { SubscriptionPlan } from '@/lib/core/rate-limiter/types'
+import { withDatabaseReadRetry } from '@/lib/db/read-retry'
 import { LoggingSession, type SessionStartParams } from '@/lib/logs/execution/logging-session'
 import type { CoreTriggerType } from '@/stores/logs/filters/types'
 
@@ -236,7 +237,9 @@ export async function preprocessExecution(
   let workflowRecord: WorkflowRecord | null = prefetchedWorkflowRecord ?? null
   if (!workflowRecord) {
     try {
-      workflowRecord = await getActiveWorkflowRecord(workflowId)
+      workflowRecord = await withDatabaseReadRetry(() => getActiveWorkflowRecord(workflowId), {
+        label: 'getActiveWorkflowRecord',
+      })
 
       if (!workflowRecord) {
         logger.warn(`[${requestId}] Workflow not found: ${workflowId}`)
@@ -297,7 +300,9 @@ export async function preprocessExecution(
       },
     }
   } else {
-    const activeWorkflow = await getActiveWorkflowRecord(workflowId)
+    const activeWorkflow = await withDatabaseReadRetry(() => getActiveWorkflowRecord(workflowId), {
+      label: 'getActiveWorkflowRecord',
+    })
     if (!activeWorkflow) {
       logger.warn(`[${requestId}] Workflow archived before execution started: ${workflowId}`)
       return {
@@ -365,7 +370,10 @@ export async function preprocessExecution(
     }
 
     if (!actorUserId) {
-      billingAttribution = await resolveSystemBillingAttribution(workspaceId)
+      billingAttribution = await withDatabaseReadRetry(
+        () => resolveSystemBillingAttribution(workspaceId),
+        { label: 'resolveSystemBillingAttribution' }
+      )
       actorUserId = billingAttribution.actorUserId
       logger.info(`[${requestId}] Using atomically resolved system actor and payer`, {
         actorUserId,
@@ -402,7 +410,11 @@ export async function preprocessExecution(
     }
 
     if (!billingAttribution) {
-      billingAttribution = await resolveBillingAttribution({ actorUserId, workspaceId })
+      const attributionInput = { actorUserId, workspaceId }
+      billingAttribution = await withDatabaseReadRetry(
+        () => resolveBillingAttribution(attributionInput),
+        { label: 'resolveBillingAttribution' }
+      )
     }
   } catch (error) {
     logger.error(`[${requestId}] Error resolving billing attribution`, { error, workflowId })
@@ -487,7 +499,10 @@ export async function preprocessExecution(
       banCandidateIds.push(userId)
     }
     try {
-      const bannedUserIds = await getActivelyBannedUserIds(banCandidateIds)
+      const bannedUserIds = await withDatabaseReadRetry(
+        () => getActivelyBannedUserIds(banCandidateIds),
+        { label: 'getActivelyBannedUserIds' }
+      )
       if (bannedUserIds.length > 0) {
         logger.warn(`[${requestId}] Execution blocked: banned account`, {
           workflowId,
@@ -558,7 +573,10 @@ export async function preprocessExecution(
     if (skipUsageLimits) return { failure: null, snapshot: null }
     let snapshot: UsageSnapshot | null = null
     try {
-      const usageCheck = await checkAttributedUsageLimits(billingAttribution)
+      const usageCheck = await withDatabaseReadRetry(
+        () => checkAttributedUsageLimits(billingAttribution),
+        { label: 'checkAttributedUsageLimits' }
+      )
       snapshot = usageCheck.payerUsage
         ? {
             ...usageCheck.payerUsage,

apps/sim/lib/workflows/executor/execution-core.test.ts

@@ -318,7 +318,8 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
       loggingSession: loggingSession as any,
     })
 
-    await Promise.resolve()
+    // setImmediate, not a fixed hop count: the assertion is about ordering, not how many microtasks precede the loads
+    await new Promise((resolve) => setImmediate(resolve))
 
     expect(callOrder).toContain('load-workflow:start')
     expect(callOrder).toContain('load-env:start')

apps/sim/lib/workflows/executor/execution-core.ts

@@ -19,6 +19,7 @@ import {
   getTimeoutErrorMessage,
   isTimeoutAbortReason,
 } from '@/lib/core/execution-limits'
+import { withDatabaseReadRetry } from '@/lib/db/read-retry'
 import { getExecutionEnvironment } from '@/lib/environment/utils'
 import { clearExecutionCancellation } from '@/lib/execution/cancellation'
 import { warmLargeValueRefs } from '@/lib/execution/payloads/hydration'
@@ -381,7 +382,11 @@ export async function executeWorkflowCore(
   options: ExecuteWorkflowCoreOptions
 ): Promise<ExecutionResult> {
   const workspaceId = options.snapshot.metadata.workspaceId
-  const rows = workspaceId ? await getCustomBlockRowsForWorkspace(workspaceId) : []
+  const rows = workspaceId
+    ? await withDatabaseReadRetry(() => getCustomBlockRowsForWorkspace(workspaceId), {
+        label: 'getCustomBlockRowsForWorkspace',
+      })
+    : []
   return withCustomBlockOverlay(rows, () => executeWorkflowCoreImpl(options))
 }
 
@@ -560,8 +565,11 @@ async function executeWorkflowCoreImpl(
     }
 
     const [workflowState, env] = await Promise.all([
-      loadWorkflowState(),
-      getExecutionEnvironment(personalEnvUserId, workspaceEnvUserId, providedWorkspaceId),
+      withDatabaseReadRetry(loadWorkflowState, { label: 'loadWorkflowState' }),
+      withDatabaseReadRetry(
+        () => getExecutionEnvironment(personalEnvUserId, workspaceEnvUserId, providedWorkspaceId),
+        { label: 'getExecutionEnvironment' }
+      ),
     ])
 
     const { blocks, loops, parallels } = workflowState
@@ -847,12 +855,16 @@ async function executeWorkflowCoreImpl(
     // stage (below) and the block-outputs stage (threaded into the executor).
     // Stored rules are the source of truth; absence yields the disabled default
     // with one indexed lookup and no masking cost for non-PII organizations.
-    const [row] = await db
-      .select({ orgSettings: organization.dataRetentionSettings })
-      .from(workspace)
-      .leftJoin(organization, eq(organization.id, workspace.organizationId))
-      .where(eq(workspace.id, providedWorkspaceId))
-      .limit(1)
+    const [row] = await withDatabaseReadRetry(
+      () =>
+        db
+          .select({ orgSettings: organization.dataRetentionSettings })
+          .from(workspace)
+          .leftJoin(organization, eq(organization.id, workspace.organizationId))
+          .where(eq(workspace.id, providedWorkspaceId))
+          .limit(1),
+      { label: 'resolvePiiRedactionPolicy' }
+    )
     const piiRedaction: EffectivePiiRedaction = resolveEffectivePiiRedaction({
       orgSettings: row?.orgSettings,
       workspaceId: providedWorkspaceId,
@@ -933,7 +945,10 @@ async function executeWorkflowCoreImpl(
         (block) => block.id === resolvedTriggerBlockId
       )
       if (entryBlock && isRunMetadataEnabled(entryBlock)) {
-        const runIdentity = await resolveStartBlockRunIdentity(metadata.principal)
+        const runIdentity = await withDatabaseReadRetry(
+          () => resolveStartBlockRunIdentity(metadata.principal),
+          { label: 'resolveStartBlockRunIdentity' }
+        )
         startRunMetadata = {
           ...runIdentity,
           workspaceId: providedWorkspaceId,