fix(tables): surface real error causes on cell-execution failures (diagnostics) (#4868)

* fix(tables): retry transient DB/Redis failures in cell execution and surface error causes Workflow-group-cell runs intermittently failed on trivial DB reads/writes under heavy fan-out, stranding cells in `running`. Investigation showed the PlanetScale and ElastiCache backends were healthy at the time — the failures are transient connection-level faults that the cell (maxAttempts: 1) had no tolerance for, and the real cause was never logged (Drizzle wraps it as "Failed query: ..." and the driver cause lives in error.cause). Resilience: - Add retryTransient (lib/table/retry-transient.ts): retries only transient infra errors (reuses isRetryableInfrastructureError; adds an ioredis command-timeout match) with jittered backoff, then rethrows. Fail-fast for everything else. - Wrap the cell's getTableById/getRowById reads, the terminal write (cell-write updateRow — idempotent via the executionId guard), and the Redis cascade-lock acquire. Diagnostics: - Add describeError (lib/core/errors/retryable-infrastructure.ts): walks the .cause chain and always returns the underlying driver cause (code/errno/ syscall + causeChain), including for unclassified errors like AbortError. - Log `cause` + a `retryable` flag (and aborted/timedOut in the cell's main catch) across the cell + finalization error paths, mirroring the existing schedule-execution pattern. Logging-only; no behavior change. This lets the next recurrence reveal the real cause and whether the retry applies. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(tables): address review feedback on cell retry resilience - retryTransient: re-check the abort signal after the backoff sleep so a cancellation during sleep stops the next attempt (don't run/return work for an already-cancelled task). - isRetryableRedisError: walk the .cause chain (mirroring the infra classifier) so wrapped Redis timeouts are recognized; drop "Connection is in subscriber mode" — that's a connection-state programming error, not a transient drop, and would just fail identically every retry. - cascade-lock: stop wrapping acquireLock in retryTransient. acquireLock is a non-idempotent SET NX, so retrying after a timed-out-but-applied first SET returns false (key already ours) and yields a false `contended` that skips the cascade. A transient Redis blip here just fails the run before pickup (no stranded cell); the dispatcher re-drives it. - Tests: cause-chain Redis match, subscriber-mode exclusion, abort-during-sleep. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(tables): drop out-of-scope abort/timeout fields from cell catch The main catch logged `aborted`/`timedOut` from `abortSignal`/`timeoutController`, but those are declared inside the outer try block (the inner try around executeWorkflow is try/finally, so this catch belongs to the outer try) and are not in scope in the catch — `next build`'s type-check failed with "Cannot find name 'abortSignal'". Local incremental `tsc --noEmit` had skipped the file and falsely passed; the Cursor/Greptile reviewers flagged this correctly. Removed the two fields. Abort/timeout is still surfaced via `cause: describeError(err)` (an aborted run shows `name: 'AbortError'` / the timeout message), so no diagnostic signal is lost. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * refactor(tables): drop in-process retry, keep cause diagnostics only In-process retry is the wrong layer for this path: the cell task is maxAttempts:1 by design, retrying on a possibly-degraded worker may not help, and it masks the very transient-failure signal we're trying to capture before we understand the root cause. Removed retryTransient entirely (file + all wrapping in cell-write, the cascade reads, and the lock acquire) and kept only the diagnostic logging. - Deleted lib/table/retry-transient.ts (+ test); cell-write and the cascade reads call getTableById/getRowById/updateRow directly again, fail-fast. - Kept describeError + `cause`/`retryable` fields across the cell + finalization catch blocks; the cell-path `retryable` flag now sources from isRetryableInfrastructureError (the canonical classifier) for consistency. Diagnostics-first: surface the real driver cause on the next recurrence, then decide the actual fix (e.g. task-level maxAttempts, or addressing the worker- side cause) from evidence rather than a speculative in-process retry. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(schedules): log error cause on scheduled-execution failure paths The scheduled-job failure paths logged the raw error (.message/stack only) — its `.cause` (the real driver error behind a Drizzle "Failed query: ..." wrapper) was never recorded, and the classified-only `describeRetryableInfrastructureError` returns undefined for unrecognized errors. A real failed run (same incident window as the cell failures) failed in `applyScheduleUpdate` with exactly this unrecorded cause. Added `cause: describeError(error)` (always-on, walks the cause chain) to the applyScheduleUpdate catch, the early-failure catch, and the unhandled-error catch — passed as a second arg so the existing message+stack still emit. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * refactor(errors): move describeError to @sim/utils/errors `describeError` is a general-purpose error/cause-chain helper — it didn't belong in `lib/core/errors/retryable-infrastructure.ts` (that module is specifically about classifying retryable infra errors, and the name read wrong for a generic diagnostic). Moved it to `@sim/utils/errors` alongside `toError`/ `getErrorMessage`/`getPostgresErrorCode`, with its own cycle-safe cause walk. - Added describeError + DescribedError + tests to packages/utils/src/errors.ts. - Reverted the describeError addition from retryable-infrastructure.ts (it keeps only isRetryableInfrastructureError / describeRetryableInfrastructureError, which are accurately named and still used by the schedule retry path). - Re-pointed all consumers (cell, logging-session, pause-persistence, schedule) to import describeError from @sim/utils/errors. The `retryable` classification flag still sources from isRetryableInfrastructureError where used. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-09-24 15:45:35 +08:00 · 2026-06-03 17:57:06 -04:00
parent 5efb47e8c5
commit aed44024d4
6 changed files with 152 additions and 13 deletions
@@ -7,7 +7,7 @@ import {
  workflowSchedule,
 } from '@sim/db'
 import { createLogger, runWithRequestContext } from '@sim/logger'
-import { toError } from '@sim/utils/errors'
+import { describeError, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { backoffWithJitter } from '@sim/utils/retry'
 import { task } from '@trigger.dev/sdk'
@@ -156,7 +156,7 @@ async function applyScheduleUpdate(

    return updatedRows.length > 0
  } catch (error) {
-    logger.error(`[${requestId}] ${context}`, error)
+    logger.error(`[${requestId}] ${context}`, error, { cause: describeError(error) })
    throw error
  }
 }
@@ -530,7 +530,13 @@ async function runWorkflowExecution({
      }
    }

-    logger.error(`[${requestId}] Early failure in scheduled workflow ${payload.workflowId}`, error)
+    logger.error(
+      `[${requestId}] Early failure in scheduled workflow ${payload.workflowId}`,
+      error,
+      {
+        cause: describeError(error),
+      }
+    )

    if (wasExecutionFinalizedByCore(error, executionId)) {
      throw error
@@ -950,7 +956,9 @@ export async function executeScheduleJob(payload: ScheduleExecutionPayload) {
          return
        }

-        logger.error(`[${requestId}] Error processing schedule ${payload.scheduleId}`, error)
+        logger.error(`[${requestId}] Error processing schedule ${payload.scheduleId}`, error, {
+          cause: describeError(error),
+        })
        await releaseClaim(
          now,
          `Failed to release schedule ${payload.scheduleId} after unhandled error`
@@ -1,12 +1,13 @@
 import { db } from '@sim/db'
 import { workflow as workflowTable } from '@sim/db/schema'
 import { createLogger, runWithRequestContext } from '@sim/logger'
-import { toError } from '@sim/utils/errors'
+import { describeError, toError } from '@sim/utils/errors'
 import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { backoffWithJitter } from '@sim/utils/retry'
 import { task } from '@trigger.dev/sdk'
 import { eq } from 'drizzle-orm'
+import { isRetryableInfrastructureError } from '@/lib/core/errors/retryable-infrastructure'
 import { createTimeoutAbortController } from '@/lib/core/execution-limits'
 import { RateLimiter } from '@/lib/core/rate-limiter/rate-limiter'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
@@ -597,8 +598,8 @@ async function runWorkflowAndWriteTerminal(
          })
          .catch((err) => {
            logger.warn(
-              `Per-block partial write failed (table=${tableId} row=${rowId} group=${groupId}):`,
-              err
+              `Per-block partial write failed (table=${tableId} row=${rowId} group=${groupId})`,
+              { cause: describeError(err), retryable: isRetryableInfrastructureError(err) }
            )
          })
      }
@@ -720,7 +721,12 @@ async function runWorkflowAndWriteTerminal(
      const message = toError(err).message
      logger.error(
        `Workflow group cell execution failed (table=${tableId} row=${rowId} group=${groupId})`,
-        { error: message, executionId }
+        {
+          error: message,
+          executionId,
+          cause: describeError(err),
+          retryable: isRetryableInfrastructureError(err),
+        }
      )
      terminalWritten = true
      await writeChain.catch(() => {})
@@ -735,7 +741,11 @@ async function runWorkflowAndWriteTerminal(
          blockErrors,
        })
      } catch (writeErr) {
-        logger.error('Also failed to write error state', { error: toError(writeErr).message })
+        logger.error('Also failed to write error state', {
+          error: toError(writeErr).message,
+          cause: describeError(writeErr),
+          retryable: isRetryableInfrastructureError(writeErr),
+        })
      }
      return 'error'
    }
@@ -1,8 +1,9 @@
 import { db } from '@sim/db'
 import { workflowExecutionLogs } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
-import { toError } from '@sim/utils/errors'
+import { describeError, toError } from '@sim/utils/errors'
 import { and, eq, sql } from 'drizzle-orm'
+import { isRetryableInfrastructureError } from '@/lib/core/errors/retryable-infrastructure'
 import { executionLogger } from '@/lib/logs/execution/logger'
 import {
  calculateCostSummary,
@@ -177,6 +178,8 @@ export class LoggingSession {
    } catch (error) {
      logger.error(`Failed to persist last started block for execution ${this.executionId}:`, {
        error: toError(error).message,
+        cause: describeError(error),
+        retryable: isRetryableInfrastructureError(error),
      })
    }
  }
@@ -193,6 +196,8 @@ export class LoggingSession {
    } catch (error) {
      logger.error(`Failed to persist last completed block for execution ${this.executionId}:`, {
        error: toError(error).message,
+        cause: describeError(error),
+        retryable: isRetryableInfrastructureError(error),
      })
    }
  }
@@ -411,6 +416,8 @@ export class LoggingSession {
        executionId: this.executionId,
        error: toError(error).message,
        stack: error instanceof Error ? error.stack : undefined,
+        cause: describeError(error),
+        retryable: isRetryableInfrastructureError(error),
      })
      throw error
    }
@@ -1057,7 +1064,11 @@ export class LoggingSession {
      this.completionAttemptFailed = true
      logger.error(
        `[${this.requestId || 'unknown'}] Cost-only fallback also failed for execution ${this.executionId}:`,
-        { error: toError(fallbackError).message }
+        {
+          error: toError(fallbackError).message,
+          cause: describeError(fallbackError),
+          retryable: isRetryableInfrastructureError(fallbackError),
+        }
      )
    }
  }
@@ -1,5 +1,6 @@
 import { createLogger } from '@sim/logger'
-import { toError } from '@sim/utils/errors'
+import { describeError, toError } from '@sim/utils/errors'
+import { isRetryableInfrastructureError } from '@/lib/core/errors/retryable-infrastructure'
 import type { LoggingSession } from '@/lib/logs/execution/logging-session'
 import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
 import type { ExecutionResult } from '@/executor/types'
@@ -46,6 +47,8 @@ export async function handlePostExecutionPauseState({
        logger.error('Failed to persist pause result', {
          executionId,
          error: toError(pauseError).message,
+          cause: describeError(pauseError),
+          retryable: isRetryableInfrastructureError(pauseError),
        })
        await loggingSession.markAsFailed(
          `Failed to persist pause state: ${toError(pauseError).message}`
@@ -59,6 +62,8 @@ export async function handlePostExecutionPauseState({
      logger.error('Failed to process queued resumes', {
        executionId,
        error: toError(resumeError).message,
+        cause: describeError(resumeError),
+        retryable: isRetryableInfrastructureError(resumeError),
      })
    }
  }
@@ -2,7 +2,7 @@
 * @vitest-environment node
 */
 import { describe, expect, it } from 'vitest'
-import { getPostgresErrorCode, toError } from './errors.js'
+import { describeError, getPostgresErrorCode, toError } from './errors.js'

 describe('toError', () => {
  it('returns the same Error when given an Error', () => {
@@ -76,3 +76,54 @@ describe('getPostgresErrorCode', () => {
    expect(getPostgresErrorCode(err1)).toBeUndefined()
  })
 })
+
+describe('describeError', () => {
+  it('reports name and message for a plain error, omitting causeChain', () => {
+    const described = describeError(new Error('boom'))
+    expect(described).toEqual({ name: 'Error', message: 'boom' })
+    expect(described.causeChain).toBeUndefined()
+  })
+
+  it('surfaces the deepest cause for a wrapped driver error', () => {
+    const driver = Object.assign(new Error('read ECONNRESET'), {
+      code: 'ECONNRESET',
+      errno: 'ECONNRESET',
+      syscall: 'read',
+    })
+    const wrapped = new Error('Failed query: select ...', { cause: driver })
+    const described = describeError(wrapped)
+    expect(described.message).toBe('read ECONNRESET')
+    expect(described.code).toBe('ECONNRESET')
+    expect(described.errno).toBe('ECONNRESET')
+    expect(described.syscall).toBe('read')
+    expect(described.causeChain).toEqual([
+      'Error: Failed query: select ...',
+      'Error: read ECONNRESET',
+    ])
+  })
+
+  it('always returns the cause for unclassified errors (AbortError)', () => {
+    const aborted = Object.assign(new Error('The operation was aborted'), { name: 'AbortError' })
+    expect(describeError(aborted)).toEqual({
+      name: 'AbortError',
+      message: 'The operation was aborted',
+    })
+  })
+
+  it('falls back to a populated description for non-Error input without throwing', () => {
+    expect(describeError('just a string')).toEqual({ name: 'Error', message: 'just a string' })
+    expect(() => describeError({ weird: true })).not.toThrow()
+  })
+
+  it('stops at depth 10 and does not loop on a cyclic cause', () => {
+    const a = new Error('a')
+    const b = new Error('b')
+    ;(a as { cause?: unknown }).cause = b
+    ;(b as { cause?: unknown }).cause = a
+    let described: ReturnType<typeof describeError> | undefined
+    expect(() => {
+      described = describeError(a)
+    }).not.toThrow()
+    expect(described?.causeChain?.length).toBeLessThanOrEqual(10)
+  })
+})
@@ -39,6 +39,60 @@ export function getPostgresConstraintName(error: unknown): string | undefined {
  return readPgErrorField(error, 'constraint_name') ?? readPgErrorField(error, 'constraint')
 }

+export interface DescribedError {
+  name: string
+  message: string
+  code?: string
+  errno?: string
+  syscall?: string
+  /** `"Name: message"` per link in the `.cause` chain, outermost first. Present only when the chain has more than one link. */
+  causeChain?: string[]
+}
+
+/**
+ * Always-on diagnostic view of an error and its `.cause` chain.
+ *
+ * Reports the fields of the DEEPEST `.cause` link, because a wrapped driver
+ * error (e.g. Drizzle's `"Failed query: ..."` wrapping an `ECONNRESET`) carries
+ * the real reason there, not on the outer wrapper. Always returns a populated
+ * object — including for non-`Error` throws and unclassified errors like
+ * `AbortError`. Cycle-safe and depth-bounded.
+ *
+ * Loggers do not serialize the non-enumerable `Error.prototype.cause`, so pass
+ * the result as an explicit structured field rather than the raw error.
+ */
+export function describeError(error: unknown): DescribedError {
+  const chain: Error[] = []
+  const seen = new Set<unknown>()
+  let current: unknown = error
+  while (current instanceof Error && !seen.has(current) && chain.length < 10) {
+    seen.add(current)
+    chain.push(current)
+    current = current.cause
+  }
+
+  if (chain.length === 0) {
+    const normalized = toError(error)
+    return { name: normalized.name, message: normalized.message }
+  }
+
+  const deepest = chain[chain.length - 1] as Error & Record<string, unknown>
+  const asString = (value: unknown): string | undefined =>
+    typeof value === 'string' ? value : undefined
+  const code = asString(deepest.code)
+  const errno = asString(deepest.errno)
+  const syscall = asString(deepest.syscall)
+
+  return {
+    name: deepest.name,
+    message: deepest.message,
+    ...(code ? { code } : {}),
+    ...(errno ? { errno } : {}),
+    ...(syscall ? { syscall } : {}),
+    ...(chain.length > 1 ? { causeChain: chain.map((e) => `${e.name}: ${e.message}`) } : {}),
+  }
+}
+
 function readPgErrorField(error: unknown, field: string): string | undefined {
  const seen = new Set<unknown>()
  let current: unknown = error