refactor(trigger): share the recorded-failure completion and skip the redundant error write on a timed-out recorded failure

Author: waleedlatif1

apps/sim/background/async-preprocessing-correlation.test.ts

@@ -426,6 +426,7 @@ describe('async preprocessing correlation threading', () => {
       mockWasExecutionFinalizedByCore.mockReturnValue(true)
 
       await expect(runRecordedFailure('execution-recorded-deadline')).rejects.toBe(error)
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
     })
 
     it('still fails the job when the run passed its deadline', async () => {
@@ -435,6 +436,7 @@ describe('async preprocessing correlation threading', () => {
       mockWasExecutionFinalizedByCore.mockReturnValue(true)
 
       await expect(runRecordedFailure('execution-recorded-late')).rejects.toBe(error)
+      expect(loggingSessionMockFns.mockSafeCompleteWithError).not.toHaveBeenCalled()
     })
   })
 

apps/sim/background/workflow-execution.ts

@@ -119,6 +119,13 @@ function isNonFaultPreprocessingRejection(rejection: PreprocessExecutionError):
   )
 }
 
+/** Completes the job with a failure already in the execution log, keeping it on the run's span. */
+function completeWithRecordedFailure(workflowId: string, executionId: string, error: unknown) {
+  const failure = toError(error)
+  trace.getActiveSpan()?.recordException(failure)
+  return buildRecordedJobFailureResult({ workflowId, executionId, error: failure.message })
+}
+
 function requireLegacyWorkflowJobString(value: unknown, field: string): string {
   if (typeof value !== 'string' || !value.trim()) {
     throw new Error(`Legacy workflow job ${field} must be a non-empty string`)
@@ -333,16 +340,9 @@ export async function executeWorkflowJob(
 
         // A failure core already recorded is the workflow's outcome, not a job fault; a run past
         // its deadline (checked after the await, which can outlast it) still fails the job.
-        if (
-          (await isRecordedWorkflowFailure(error, executionId, loggingSession)) &&
-          !timeoutController.isTimedOut()
-        ) {
-          trace.getActiveSpan()?.recordException(toError(error))
-          return buildRecordedJobFailureResult({
-            workflowId,
-            executionId,
-            error: toError(error).message,
-          })
+        if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) {
+          if (timeoutController.isTimedOut()) throw error
+          return completeWithRecordedFailure(workflowId, executionId, error)
         }
 
         const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
@@ -365,12 +365,7 @@ export async function executeWorkflowJob(
           isNonFaultPreprocessingRejection(preprocessingRejection) &&
           loggingSession.hasCompleted()
         ) {
-          trace.getActiveSpan()?.recordException(toError(error))
-          return buildRecordedJobFailureResult({
-            workflowId,
-            executionId,
-            error: toError(error).message,
-          })
+          return completeWithRecordedFailure(workflowId, executionId, error)
         }
 
         throw error

apps/sim/lib/core/async-jobs/recorded-failure.ts

@@ -2,12 +2,12 @@ import { isRecordLike } from '@sim/utils/object'
 import type { JobType } from '@/lib/core/async-jobs/types'
 
 /**
- * Job types whose runner reports a failure the run already recorded in its
+ * The job type whose runner reports a failure the run already recorded in its
  * execution log by completing with a {@link RecordedJobFailureResult} instead
  * of throwing, so the queue does not fault (and alert on) a user or workflow
  * error.
  */
-const RECORDED_FAILURE_JOB_TYPES: ReadonlySet<string> = new Set<JobType>(['workflow-execution'])
+const RECORDED_FAILURE_JOB_TYPE: JobType = 'workflow-execution'
 
 /**
  * Bounds the error in UTF-8 bytes so the result stays inline under Trigger.dev's
@@ -35,6 +35,7 @@ interface RecordedJobFailureResult {
   executedAt: string
 }
 
+/** The output a runner completes with when its failure is already in the execution log. */
 export function buildRecordedJobFailureResult(params: {
   workflowId: string
   executionId: string
@@ -56,6 +57,6 @@ export function buildRecordedJobFailureResult(params: {
  * runner threw.
  */
 export function getRecordedJobFailureError(type: string, output: unknown): string | undefined {
-  if (!RECORDED_FAILURE_JOB_TYPES.has(type) || !isRecordLike(output)) return undefined
+  if (type !== RECORDED_FAILURE_JOB_TYPE || !isRecordLike(output)) return undefined
   return output.success === false && typeof output.error === 'string' ? output.error : undefined
 }