fix(processing): unblock error responses and isolate run-count failures

Remove unnecessary `await waitForCompletion()` from non-SSE and SSE error
paths where no `markAsFailed()` follows — these were blocking error responses
on log persistence for no reason. Wrap `updateWorkflowRunCounts` in its own
try/catch so a run-count DB failure cannot prevent session completion, billing,
and trace span persistence.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: waleedlatif1

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -22,7 +22,6 @@ import { createExecutionEventWriter, setExecutionMeta } from '@/lib/execution/ev
 import { processInputFileFields } from '@/lib/execution/files'
 import { preprocessExecution } from '@/lib/execution/preprocessing'
 import { LoggingSession } from '@/lib/logs/execution/logging-session'
-import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
 import {
   cleanupExecutionBase64Cache,
   hydrateUserFilesWithBase64,
@@ -642,6 +641,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
           logger.info(`[${requestId}] Non-SSE execution timed out`, {
             timeoutMs: timeoutController.timeoutMs,
           })
+          await loggingSession.waitForCompletion()
           await loggingSession.markAsFailed(timeoutErrorMessage)
 
           return NextResponse.json(
@@ -698,20 +698,6 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
 
         const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
 
-        // Fire-and-forget: execution-core.ts already handles logging via its own
-        // fire-and-forget call. The `completed` guard in LoggingSession prevents
-        // double-writes, so this is a no-op — but we avoid awaiting it to reduce
-        // error-response latency.
-        const { traceSpans, totalDuration } = executionResult
-          ? buildTraceSpans(executionResult)
-          : { traceSpans: [], totalDuration: 0 }
-
-        void loggingSession.safeCompleteWithError({
-          totalDurationMs: totalDuration || executionResult?.metadata?.duration,
-          error: { message: errorMessage },
-          traceSpans,
-        })
-
         return NextResponse.json(
           {
             success: false,
@@ -1041,6 +1027,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
               logger.error(`[${requestId}] Missing snapshot seed for paused execution`, {
                 executionId,
               })
+              await loggingSession.waitForCompletion()
               await loggingSession.markAsFailed('Missing snapshot seed for paused execution')
             } else {
               try {
@@ -1056,6 +1043,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
                   executionId,
                   error: pauseError instanceof Error ? pauseError.message : String(pauseError),
                 })
+                await loggingSession.waitForCompletion()
                 await loggingSession.markAsFailed(
                   `Failed to persist pause state: ${pauseError instanceof Error ? pauseError.message : String(pauseError)}`
                 )
@@ -1072,6 +1060,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
                 timeoutMs: timeoutController.timeoutMs,
               })
 
+              await loggingSession.waitForCompletion()
               await loggingSession.markAsFailed(timeoutErrorMessage)
 
               sendEvent({
@@ -1133,15 +1122,6 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id:
           logger.error(`[${requestId}] SSE execution failed: ${errorMessage}`, { isTimeout })
 
           const executionResult = hasExecutionResult(error) ? error.executionResult : undefined
-          const { traceSpans, totalDuration } = executionResult
-            ? buildTraceSpans(executionResult)
-            : { traceSpans: [], totalDuration: 0 }
-
-          await loggingSession.safeCompleteWithError({
-            totalDurationMs: totalDuration || executionResult?.metadata?.duration,
-            error: { message: errorMessage },
-            traceSpans,
-          })
 
           sendEvent({
             type: 'execution:error',

apps/sim/lib/logs/execution/logging-session.ts

@@ -91,6 +91,8 @@ export class LoggingSession {
   private completed = false
   /** Synchronous flag to prevent concurrent completion attempts (race condition guard) */
   private completing = false
+  /** Tracks the in-flight completion promise so callers can await it */
+  private completionPromise: Promise<void> | null = null
   private accumulatedCost: AccumulatedCost = {
     total: BASE_EXECUTION_CHARGE,
     input: 0,
@@ -694,7 +696,27 @@ export class LoggingSession {
     }
   }
 
+  /**
+   * Wait for any in-flight fire-and-forget completion to finish.
+   * Used by callers (e.g. markAsFailed) that need to ensure completion
+   * has settled before overwriting execution status.
+   */
+  async waitForCompletion(): Promise<void> {
+    if (this.completionPromise) {
+      try {
+        await this.completionPromise
+      } catch {
+        /* already handled by safe* wrapper */
+      }
+    }
+  }
+
   async safeComplete(params: SessionCompleteParams = {}): Promise<void> {
+    this.completionPromise = this._safeCompleteImpl(params)
+    return this.completionPromise
+  }
+
+  private async _safeCompleteImpl(params: SessionCompleteParams = {}): Promise<void> {
     try {
       await this.complete(params)
     } catch (error) {
@@ -714,6 +736,11 @@ export class LoggingSession {
   }
 
   async safeCompleteWithError(params?: SessionErrorCompleteParams): Promise<void> {
+    this.completionPromise = this._safeCompleteWithErrorImpl(params)
+    return this.completionPromise
+  }
+
+  private async _safeCompleteWithErrorImpl(params?: SessionErrorCompleteParams): Promise<void> {
     try {
       await this.completeWithError(params)
     } catch (error) {
@@ -735,6 +762,11 @@ export class LoggingSession {
   }
 
   async safeCompleteWithCancellation(params?: SessionCancelledParams): Promise<void> {
+    this.completionPromise = this._safeCompleteWithCancellationImpl(params)
+    return this.completionPromise
+  }
+
+  private async _safeCompleteWithCancellationImpl(params?: SessionCancelledParams): Promise<void> {
     try {
       await this.completeWithCancellation(params)
     } catch (error) {
@@ -755,6 +787,11 @@ export class LoggingSession {
   }
 
   async safeCompleteWithPause(params?: SessionPausedParams): Promise<void> {
+    this.completionPromise = this._safeCompleteWithPauseImpl(params)
+    return this.completionPromise
+  }
+
+  private async _safeCompleteWithPauseImpl(params?: SessionPausedParams): Promise<void> {
     try {
       await this.completeWithPause(params)
     } catch (error) {

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

@@ -368,7 +368,11 @@ export async function executeWorkflowCore(
         const { traceSpans, totalDuration } = buildTraceSpans(result)
 
         if (result.success && result.status !== 'paused') {
-          await updateWorkflowRunCounts(workflowId)
+          try {
+            await updateWorkflowRunCounts(workflowId)
+          } catch (runCountError) {
+            logger.error(`[${requestId}] Failed to update run counts`, { error: runCountError })
+          }
         }
 
         if (result.status === 'cancelled') {