fix(execution): fail a paused run whose log was never finalized instead of publishing it

Author: waleedlatif1

apps/sim/lib/workflows/executor/pause-persistence.integration.ts

@@ -1,22 +1,20 @@
 /**
- * Pause publication against real PostgreSQL: a paused run becomes resumable only after
- * its log has been finalized out of `running`, so an immediate resume finds a claimable log.
+ * Pause publication against real PostgreSQL: a paused run becomes resumable only once its
+ * log has been finalized out of `running`, so an immediate resume finds a claimable log.
  */
 import { db } from '@sim/db'
 import {
   pausedExecutions,
-  resumeQueue,
   user,
   workflow,
   workflowExecutionLogs,
   workflowExecutionSnapshots,
   workspace,
 } from '@sim/db/schema'
 import { createDeferred } from '@sim/testing'
-import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { eq } from 'drizzle-orm'
-import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest'
 import {
   type BillingAttributionSnapshot,
   resolveBillingAttribution,
@@ -31,14 +29,10 @@ const ids = {
   owner: `pause-publish-owner-${generateId()}`,
   workspace: generateId(),
   workflow: generateId(),
-  execution: generateId(),
 }
 
 const CONTEXT_ID = 'approval'
 
-/** Long enough for an unguarded publish to commit; a guarded one never resolves while held. */
-const UNGUARDED_PUBLISH_WINDOW_MS = 250
-
 const workflowState: WorkflowState = {
   blocks: {
     start: {
@@ -56,7 +50,10 @@ const workflowState: WorkflowState = {
   parallels: {},
 }
 
-function pausedResult(billingAttribution: BillingAttributionSnapshot): ExecutionResult {
+function pausedResult(
+  executionId: string,
+  billingAttribution: BillingAttributionSnapshot
+): ExecutionResult {
   return {
     success: true,
     output: {},
@@ -77,7 +74,7 @@ function pausedResult(billingAttribution: BillingAttributionSnapshot): Execution
         metadata: {
           workflowId: ids.workflow,
           workspaceId: ids.workspace,
-          executionId: ids.execution,
+          executionId,
           userId: ids.owner,
           billingAttribution,
         },
@@ -87,17 +84,34 @@ function pausedResult(billingAttribution: BillingAttributionSnapshot): Execution
   }
 }
 
-async function logStatus() {
+/** Starts a run whose log is `running`, as the core leaves it when execution returns. */
+async function startRun() {
+  const executionId = generateId()
+  const billingAttribution = await resolveBillingAttribution({
+    actorUserId: ids.owner,
+    workspaceId: ids.workspace,
+  })
+  const loggingSession = new LoggingSession(ids.workflow, executionId, 'api', 'pause-publish')
+  await loggingSession.safeStart({
+    userId: ids.owner,
+    workspaceId: ids.workspace,
+    billingAttribution,
+    workflowState,
+  })
+  return { executionId, loggingSession, result: pausedResult(executionId, billingAttribution) }
+}
+
+async function logStatus(executionId: string) {
   const [row] = await db
     .select({ status: workflowExecutionLogs.status })
     .from(workflowExecutionLogs)
-    .where(eq(workflowExecutionLogs.executionId, ids.execution))
+    .where(eq(workflowExecutionLogs.executionId, executionId))
   return row?.status
 }
 
-function resume() {
+function resume(executionId: string) {
   return PauseResumeManager.enqueueOrStartResume({
-    executionId: ids.execution,
+    executionId,
     workflowId: ids.workflow,
     contextId: CONTEXT_ID,
     resumeInput: {},
@@ -132,8 +146,12 @@ beforeAll(async () => {
   })
 })
 
+afterEach(() => {
+  vi.restoreAllMocks()
+})
+
 afterAll(async () => {
-  await db.delete(resumeQueue).where(eq(resumeQueue.parentExecutionId, ids.execution))
+  // Deleting a paused execution cascades to its resume queue entries.
   await db.delete(pausedExecutions).where(eq(pausedExecutions.workflowId, ids.workflow))
   await db.delete(workflowExecutionLogs).where(eq(workflowExecutionLogs.workflowId, ids.workflow))
   await db
@@ -144,44 +162,53 @@ afterAll(async () => {
 })
 
 describe('handlePostExecutionPauseState', () => {
-  it('publishes a pause only after the run log is finalized, so an immediate resume finds a claimable log', async () => {
-    const billingAttribution = await resolveBillingAttribution({
-      actorUserId: ids.owner,
-      workspaceId: ids.workspace,
-    })
-    const loggingSession = new LoggingSession(ids.workflow, ids.execution, 'api', 'pause-publish')
-    await loggingSession.safeStart({
-      userId: ids.owner,
-      workspaceId: ids.workspace,
-      billingAttribution,
-      workflowState,
-    })
-    expect(await logStatus()).toBe('running')
+  it('publishes a pause only after its log is finalized, so an immediate resume finds a claimable log', async () => {
+    const { executionId, loggingSession, result } = await startRun()
 
     /** Holds the core's background log finalizer open, as a slow trace projection would. */
     const finalizer = createDeferred<void>()
     loggingSession.setPostExecutionPromise(
       finalizer.promise.then(() => loggingSession.safeCompleteWithPause({ traceSpans: [] }))
     )
 
+    const persistPauseResult = PauseResumeManager.persistPauseResult
+    const logFinalizedAtPublish: boolean[] = []
+    vi.spyOn(PauseResumeManager, 'persistPauseResult').mockImplementation((args) => {
+      logFinalizedAtPublish.push(loggingSession.hasCompleted())
+      return persistPauseResult.call(PauseResumeManager, args)
+    })
+
     const publish = handlePostExecutionPauseState({
-      result: pausedResult(billingAttribution),
+      result,
       workflowId: ids.workflow,
-      executionId: ids.execution,
+      executionId,
+      loggingSession,
+    })
+    finalizer.resolve()
+    await publish
+
+    expect(logFinalizedAtPublish).toEqual([true])
+    expect(await logStatus(executionId)).toBe('pending')
+    await expect(resume(executionId)).resolves.toMatchObject({ status: 'starting' })
+  })
+
+  it('fails the run instead of publishing a pause whose log was never finalized', async () => {
+    const { executionId, loggingSession, result } = await startRun()
+
+    /** The core's finalizer swallows its own failures, so a lost pause write still settles. */
+    loggingSession.setPostExecutionPromise(Promise.resolve())
+
+    await handlePostExecutionPauseState({
+      result,
+      workflowId: ids.workflow,
+      executionId,
       loggingSession,
     })
-    await Promise.race([publish, sleep(UNGUARDED_PUBLISH_WINDOW_MS)])
 
-    expect(await logStatus()).toBe('running')
-    await expect(resume()).rejects.toMatchObject({
+    expect(await logStatus(executionId)).toBe('failed')
+    await expect(resume(executionId)).rejects.toMatchObject({
       name: 'ResumeAdmissionError',
       statusCode: 404,
     })
-
-    finalizer.resolve()
-    await publish
-
-    expect(await logStatus()).toBe('pending')
-    await expect(resume()).resolves.toMatchObject({ status: 'starting' })
   })
 })

apps/sim/lib/workflows/executor/pause-persistence.ts

@@ -21,14 +21,15 @@ interface HandlePostExecutionPauseStateArgs {
  * Every caller of `executeWorkflowCore` must call this after execution completes
  * to ensure HITL pause state is persisted to the database and queued resumes are drained.
  *
- * - If execution is paused with a valid snapshot: persists to `paused_executions` table
+ * - If execution is paused but its log was never finalized: marks execution as failed
  * - If execution is paused without a snapshot: marks execution as failed
+ * - If execution is paused with a valid snapshot: persists to `paused_executions` table
  * - If execution is not paused: processes any queued resume entries
  *
- * A pause is published only after the core's post-execution logging settles. The
- * core finalizes the run log in the background, and a resume claims that log only
- * once it has left `running`, so publishing first lets an immediate resume be
- * rejected as no longer resumable.
+ * A pause is published only after the core's post-execution logging has persisted
+ * the paused log. A resume claims that log only once it has left `running`, so a
+ * pause published before then (or without it ever happening) is rejected as no
+ * longer resumable.
  */
 export async function handlePostExecutionPauseState({
   result,
@@ -39,7 +40,10 @@ export async function handlePostExecutionPauseState({
 }: HandlePostExecutionPauseStateArgs): Promise<void> {
   if (result.status === 'paused') {
     await loggingSession.waitForPostExecution()
-    if (!result.snapshotSeed) {
+    if (!loggingSession.hasCompleted()) {
+      logger.error('Paused execution log was not finalized', { executionId })
+      await loggingSession.markAsFailed('Failed to record paused execution')
+    } else if (!result.snapshotSeed) {
       logger.error('Missing snapshot seed for paused execution', { executionId })
       await loggingSession.markAsFailed('Missing snapshot seed for paused execution')
     } else {