fix(async): harden execution finalization guards

Prevent leaked core finalization markers from accumulating while keeping outer recovery paths idempotent. Preserve best-effort logging completion by reusing settled completion promises instead of reopening duplicate terminal writes.

Author: PlaneInABottle

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

@@ -1,4 +1,8 @@
-import { describe, expect, it, vi } from 'vitest'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { completeWorkflowExecutionMock } = vi.hoisted(() => ({
+  completeWorkflowExecutionMock: vi.fn(),
+}))
 
 vi.mock('@sim/db', () => ({
   db: {},
@@ -25,7 +29,7 @@ vi.mock('drizzle-orm', () => ({
 vi.mock('@/lib/logs/execution/logger', () => ({
   executionLogger: {
     startWorkflowExecution: vi.fn(),
-    completeWorkflowExecution: vi.fn(),
+    completeWorkflowExecution: completeWorkflowExecutionMock,
   },
 }))
 
@@ -50,26 +54,39 @@ vi.mock('@/lib/logs/execution/logging-factory', () => ({
 import { LoggingSession } from './logging-session'
 
 describe('LoggingSession completion retries', () => {
-  it('clears failed completion promise so error finalization can retry', async () => {
-    const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1') as any
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
 
-    const successFinalizeError = new Error('success finalize failed')
-    session.complete = vi.fn().mockRejectedValue(successFinalizeError)
-    session.completeWithCostOnlyLog = vi.fn().mockRejectedValue(successFinalizeError)
-    session.completeWithError = vi.fn().mockResolvedValue(undefined)
+  it('keeps completion best-effort when full completion and fallback both fail', async () => {
+    const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1')
 
-    await expect(session.safeComplete({ finalOutput: { ok: true } })).rejects.toThrow(
-      'success finalize failed'
-    )
+    completeWorkflowExecutionMock
+      .mockRejectedValueOnce(new Error('success finalize failed'))
+      .mockRejectedValueOnce(new Error('cost only failed'))
+
+    await expect(session.safeComplete({ finalOutput: { ok: true } })).resolves.toBeUndefined()
 
     await expect(
       session.safeCompleteWithError({
         error: { message: 'fallback error finalize' },
       })
     ).resolves.toBeUndefined()
 
-    expect(session.complete).toHaveBeenCalledTimes(1)
-    expect(session.completeWithCostOnlyLog).toHaveBeenCalledTimes(1)
-    expect(session.completeWithError).toHaveBeenCalledTimes(1)
+    expect(completeWorkflowExecutionMock).toHaveBeenCalledTimes(2)
+  })
+
+  it('reuses the settled completion promise for repeated completion attempts', async () => {
+    const session = new LoggingSession('workflow-1', 'execution-1', 'api', 'req-1')
+
+    completeWorkflowExecutionMock
+      .mockRejectedValueOnce(new Error('success finalize failed'))
+      .mockRejectedValueOnce(new Error('cost only failed'))
+
+    await expect(session.safeComplete({ finalOutput: { ok: true } })).resolves.toBeUndefined()
+
+    await expect(session.safeComplete({ finalOutput: { ok: true } })).resolves.toBeUndefined()
+
+    expect(completeWorkflowExecutionMock).toHaveBeenCalledTimes(2)
   })
 })

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

@@ -25,6 +25,8 @@ type TriggerData = Record<string, unknown> & {
 
 const logger = createLogger('LoggingSession')
 
+type CompletionAttempt = 'complete' | 'error' | 'cancelled' | 'paused'
+
 export interface SessionStartParams {
   userId?: string
   workspaceId: string
@@ -98,6 +100,8 @@ export class LoggingSession {
   private completing = false
   /** Tracks the in-flight completion promise so callers can await it */
   private completionPromise: Promise<void> | null = null
+  private completionAttempt: CompletionAttempt | null = null
+  private completionAttemptFailed = false
   private accumulatedCost: AccumulatedCost = {
     total: BASE_EXECUTION_CHARGE,
     input: 0,
@@ -695,15 +699,32 @@ export class LoggingSession {
     }
   }
 
-  async safeComplete(params: SessionCompleteParams = {}): Promise<void> {
-    if (this.completionPromise) return this.completionPromise
-    this.completionPromise = this._safeCompleteImpl(params).catch((error) => {
-      this.completionPromise = null
+  private shouldStartNewCompletionAttempt(attempt: CompletionAttempt): boolean {
+    return this.completionAttemptFailed && this.completionAttempt !== 'error' && attempt === 'error'
+  }
+
+  private runCompletionAttempt(
+    attempt: CompletionAttempt,
+    run: () => Promise<void>
+  ): Promise<void> {
+    if (this.completionPromise && !this.shouldStartNewCompletionAttempt(attempt)) {
+      return this.completionPromise
+    }
+
+    this.completionAttempt = attempt
+    this.completionAttemptFailed = false
+    this.completionPromise = run().catch((error) => {
+      this.completionAttemptFailed = true
       throw error
     })
+
     return this.completionPromise
   }
 
+  async safeComplete(params: SessionCompleteParams = {}): Promise<void> {
+    return this.runCompletionAttempt('complete', () => this._safeCompleteImpl(params))
+  }
+
   private async _safeCompleteImpl(params: SessionCompleteParams = {}): Promise<void> {
     try {
       await this.complete(params)
@@ -724,12 +745,7 @@ export class LoggingSession {
   }
 
   async safeCompleteWithError(params?: SessionErrorCompleteParams): Promise<void> {
-    if (this.completionPromise) return this.completionPromise
-    this.completionPromise = this._safeCompleteWithErrorImpl(params).catch((error) => {
-      this.completionPromise = null
-      throw error
-    })
-    return this.completionPromise
+    return this.runCompletionAttempt('error', () => this._safeCompleteWithErrorImpl(params))
   }
 
   private async _safeCompleteWithErrorImpl(params?: SessionErrorCompleteParams): Promise<void> {
@@ -754,12 +770,9 @@ export class LoggingSession {
   }
 
   async safeCompleteWithCancellation(params?: SessionCancelledParams): Promise<void> {
-    if (this.completionPromise) return this.completionPromise
-    this.completionPromise = this._safeCompleteWithCancellationImpl(params).catch((error) => {
-      this.completionPromise = null
-      throw error
-    })
-    return this.completionPromise
+    return this.runCompletionAttempt('cancelled', () =>
+      this._safeCompleteWithCancellationImpl(params)
+    )
   }
 
   private async _safeCompleteWithCancellationImpl(params?: SessionCancelledParams): Promise<void> {
@@ -783,12 +796,7 @@ export class LoggingSession {
   }
 
   async safeCompleteWithPause(params?: SessionPausedParams): Promise<void> {
-    if (this.completionPromise) return this.completionPromise
-    this.completionPromise = this._safeCompleteWithPauseImpl(params).catch((error) => {
-      this.completionPromise = null
-      throw error
-    })
-    return this.completionPromise
+    return this.runCompletionAttempt('paused', () => this._safeCompleteWithPauseImpl(params))
   }
 
   private async _safeCompleteWithPauseImpl(params?: SessionPausedParams): Promise<void> {

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

@@ -89,7 +89,11 @@ vi.mock('@/serializer', () => ({
   })),
 }))
 
-import { executeWorkflowCore, wasExecutionFinalizedByCore } from './execution-core'
+import {
+  executeWorkflowCore,
+  FINALIZED_EXECUTION_ID_TTL_MS,
+  wasExecutionFinalizedByCore,
+} from './execution-core'
 
 describe('executeWorkflowCore terminal finalization sequencing', () => {
   const loggingSession = {
@@ -127,6 +131,7 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
 
   beforeEach(() => {
     vi.clearAllMocks()
+    vi.useRealTimers()
 
     loadWorkflowFromNormalizedTablesMock.mockResolvedValue({
       blocks: {
@@ -330,7 +335,47 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
 
     expect(safeCompleteWithErrorMock).toHaveBeenCalledTimes(1)
     expect(wasExecutionFinalizedByCore('engine failed', 'execution-1')).toBe(true)
-    expect(wasExecutionFinalizedByCore('engine failed', 'execution-1')).toBe(false)
+    expect(wasExecutionFinalizedByCore('engine failed', 'execution-1')).toBe(true)
+  })
+
+  it('expires stale finalized execution ids for callers that never consume the guard', async () => {
+    vi.useFakeTimers()
+    vi.setSystemTime(new Date('2026-03-13T00:00:00.000Z'))
+
+    executorExecuteMock.mockRejectedValue('engine failed')
+
+    await expect(
+      executeWorkflowCore({
+        snapshot: {
+          ...createSnapshot(),
+          metadata: {
+            ...createSnapshot().metadata,
+            executionId: 'execution-stale',
+          },
+        } as any,
+        callbacks: {},
+        loggingSession: loggingSession as any,
+      })
+    ).rejects.toBe('engine failed')
+
+    vi.setSystemTime(new Date(Date.now() + FINALIZED_EXECUTION_ID_TTL_MS + 1))
+
+    await expect(
+      executeWorkflowCore({
+        snapshot: {
+          ...createSnapshot(),
+          metadata: {
+            ...createSnapshot().metadata,
+            executionId: 'execution-fresh',
+          },
+        } as any,
+        callbacks: {},
+        loggingSession: loggingSession as any,
+      })
+    ).rejects.toBe('engine failed')
+
+    expect(wasExecutionFinalizedByCore('engine failed', 'execution-stale')).toBe(false)
+    expect(wasExecutionFinalizedByCore('engine failed', 'execution-fresh')).toBe(true)
   })
 
   it('falls back to error finalization when success finalization rejects', async () => {

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

@@ -116,18 +116,39 @@ type ExecutionErrorWithFinalizationFlag = Error & {
   executionFinalizedByCore?: boolean
 }
 
-const finalizedExecutionIds = new Set<string>()
+export const FINALIZED_EXECUTION_ID_TTL_MS = 5 * 60 * 1000
+
+const finalizedExecutionIds = new Map<string, number>()
+
+function cleanupExpiredFinalizedExecutionIds(now = Date.now()): void {
+  for (const [executionId, expiresAt] of finalizedExecutionIds.entries()) {
+    if (expiresAt > now) {
+      break
+    }
+
+    finalizedExecutionIds.delete(executionId)
+  }
+}
+
+function rememberFinalizedExecutionId(executionId: string): void {
+  const now = Date.now()
+
+  cleanupExpiredFinalizedExecutionIds(now)
+  finalizedExecutionIds.set(executionId, now + FINALIZED_EXECUTION_ID_TTL_MS)
+}
 
 function markExecutionFinalizedByCore(error: unknown, executionId: string): void {
-  finalizedExecutionIds.add(executionId)
+  rememberFinalizedExecutionId(executionId)
 
   if (error instanceof Error) {
     ;(error as ExecutionErrorWithFinalizationFlag).executionFinalizedByCore = true
   }
 }
 
 export function wasExecutionFinalizedByCore(error: unknown, executionId?: string): boolean {
-  if (executionId && finalizedExecutionIds.delete(executionId)) {
+  cleanupExpiredFinalizedExecutionIds()
+
+  if (executionId && finalizedExecutionIds.has(executionId)) {
     return true
   }