Add fatal step e2e test

.changeset/cyan-ducks-wonder.md

@@ -0,0 +1,5 @@
+---
+"@workflow/core": patch
+---
+
+Allow step retrying if it fails without proper cleanup

packages/core/e2e/e2e.test.ts

@@ -43,7 +43,7 @@
     const url = new URL('/api/trigger', deploymentUrl);
     url.searchParams.set('runId', runId);
 
     const res = await fetch(url);
 
     if (res.status === 202) {
       // Workflow run is still running, so we need to wait and poll again
@@ -60,7 +60,7 @@
       return res.body;
     }
 
     throw new Error(`Unexpected content type: ${contentType}`);
   }
 }
 
@@ -76,7 +76,7 @@
       workflowFile: 'workflows/98_duplicate_case.ts',
       workflowFn: 'addTenWorkflow',
     },
   ])('addTenWorkflow', { timeout: 60_000 }, async (workflow) => {
     const run = await triggerWorkflow(workflow, [123]);
     const returnValue = await getWorkflowReturnValue(run.runId);
     expect(returnValue).toBe(133);
@@ -97,25 +97,25 @@
     ]);
   });
 
   test('promiseAllWorkflow', { timeout: 60_000 }, async () => {
     const run = await triggerWorkflow('promiseAllWorkflow', []);
     const returnValue = await getWorkflowReturnValue(run.runId);
     expect(returnValue).toBe('ABC');
   });
 
   test('promiseRaceWorkflow', { timeout: 60_000 }, async () => {
     const run = await triggerWorkflow('promiseRaceWorkflow', []);
     const returnValue = await getWorkflowReturnValue(run.runId);
     expect(returnValue).toBe('B');
   });
 
   test('promiseAnyWorkflow', { timeout: 60_000 }, async () => {
     const run = await triggerWorkflow('promiseAnyWorkflow', []);
     const returnValue = await getWorkflowReturnValue(run.runId);
     expect(returnValue).toBe('B');
   });
 
   test('readableStreamWorkflow', { timeout: 60_000 }, async () => {
     const run = await triggerWorkflow('readableStreamWorkflow', []);
     const returnValue = await getWorkflowReturnValue(run.runId);
     expect(returnValue).toBeInstanceOf(ReadableStream);
@@ -145,7 +145,7 @@
       method: 'POST',
       body: JSON.stringify({ token, data: { message: 'one' } }),
     });
     expect(res.status).toBe(200);
     let body = await res.json();
     expect(body.runId).toBe(run.runId);
 
@@ -214,7 +214,7 @@
         body: JSON.stringify({ message: 'one' }),
       }
     );
     expect(res.status).toBe(202);
     const body = await res.text();
     expect(body).toBe('');
 
@@ -292,14 +292,14 @@
     expect(body).toBe('');
   });
 
   test('sleepingWorkflow', { timeout: 60_000 }, async () => {
     const run = await triggerWorkflow('sleepingWorkflow', []);
     const returnValue = await getWorkflowReturnValue(run.runId);
     expect(returnValue.startTime).toBeLessThan(returnValue.endTime);
     expect(returnValue.endTime - returnValue.startTime).toBeGreaterThan(9999);
   });
 
   test('nullByteWorkflow', { timeout: 60_000 }, async () => {
     const run = await triggerWorkflow('nullByteWorkflow', []);
     const returnValue = await getWorkflowReturnValue(run.runId);
     expect(returnValue).toBe('null byte \0');
@@ -719,4 +719,49 @@
       expect(stepCompletedEvents).toHaveLength(1);
     }
   );
+
+  test(
+    'processExitResilienceWorkflow - step self-terminates on first attempt',
+    { timeout: 60_000 },
+    async () => {
+      // This test verifies that the workflow system is resilient to fatal process crashes
+      // The step will call process.exit() on the first attempt, simulating an unhandled
+      // fatal error, but the system should recover and retry on a new process
+      const run = await triggerWorkflow('processExitResilienceWorkflow', []);
+      const returnValue = await getWorkflowReturnValue(run.runId);
+
+      // The workflow should complete successfully after the process crash
+      expect(returnValue).toMatchObject({
+        attempt: 2,
+        status: 'recovered',
+      });
+
+      // Verify the run data shows successful completion
+      const { json: runData } = await cliInspectJson(
+        `runs ${run.runId} --withData`
+      );
+      expect(runData).toMatchObject({
+        runId: run.runId,
+        status: 'completed',
+        output: { attempt: 2, status: 'recovered' },
+      });
+
+      // Query steps to verify the step was retried
+      const { json: stepsData } = await cliInspectJson(
+        `steps --runId ${run.runId} --withData`
+      );
+      expect(stepsData).toBeDefined();
+      expect(Array.isArray(stepsData)).toBe(true);
+      expect(stepsData.length).toBeGreaterThan(0);
+
+      // Find the stepThatExitsOnFirstAttempt step
+      const exitStep = stepsData.find((s: any) =>
+        s.stepName.includes('stepThatExitsOnFirstAttempt')
+      );
+      expect(exitStep).toBeDefined();
+      expect(exitStep.status).toBe('completed');
+      expect(exitStep.attempt).toBe(2);
+      expect(exitStep.output).toEqual([{ attempt: 2, status: 'recovered' }]);
+    }
+  );
 });

packages/core/src/runtime.ts

@@ -640,12 +640,14 @@ export const stepEntrypoint =
           let result: unknown;
           const attempt = step.attempt + 1;
           try {
-            if (step.status !== 'pending') {
-              // We should only be running the step if it's pending
-              // (initial state, or state set on re-try), so the step has been
+            if (!['pending', 'running'].includes(step.status)) {
+              // We should only be running the step if it's either
+              // a) pending - initial state, or state set on re-try
+              // b) running - if a step fails mid-execution, like a function timeout
+              // so the step has been
               // invoked erroneously.
               console.error(
-                `[Workflows] "${workflowRunId}" - Step invoked erroneously, expected status "pending", got "${step.status}" instead, skipping execution`
+                `[Workflows] "${workflowRunId}" - Step invoked erroneously, expected status "pending" or "running", got "${step.status}" instead, skipping execution`
               );
               span?.setAttributes({
                 ...Attribute.StepSkipped(true),
@@ -698,11 +700,24 @@ export const stepEntrypoint =
               () => stepFn(...args)
             );
 
+            // NOTE: None of the code from this point is guaranteed to run
+            // Since the step might fail or cause a function timeout and the process might be SIGKILL'd
+            // The workflow runtime must be resilient to the below code not executing on a failed step
+
             result = dehydrateStepReturnValue(result, ops);
 
             waitUntil(Promise.all(ops));
 
-            // Update the event log with the step result
+            // Mark the step as completed first. This order is important. If a concurrent
+            // execution marked the step as complete, this request should throw, and
+            // this prevent the step_completed event in the event log
+            // TODO: this should really be atomic and handled by the world
+            await world.steps.update(workflowRunId, stepId, {
+              status: 'completed',
+              output: result as Serializable,
+            });
+
+            // Then, append the event log with the step result
             await world.events.create(workflowRunId, {
               eventType: 'step_completed',
               correlationId: stepId,
@@ -711,11 +726,6 @@ export const stepEntrypoint =
               },
             });
 
-            await world.steps.update(workflowRunId, stepId, {
-              status: 'completed',
-              output: result as Serializable,
-            });
-
             span?.setAttributes({
               ...Attribute.StepStatus('completed'),
               ...Attribute.StepResultType(typeof result),

workbench/example/workflows/99_e2e.ts

@@ -512,3 +512,36 @@ async function doubleNumber(x: number) {
   'use step';
   return x * 2;
 }
+
+//////////////////////////////////////////////////////////
+
+async function stepThatExitsOnFirstAttempt() {
+  'use step';
+  const { attempt } = getStepMetadata();
+  console.log(`stepThatExitsOnFirstAttempt - attempt: ${attempt}`);
+
+  // Kill the process on the first attempt to simulate a fatal crash
+  if (attempt === 1) {
+    console.log(
+      `Attempt ${attempt} - calling process.exit() to simulate fatal crash`
+    );
+    process.exit(1);
+  }
+
+  console.log(`Attempt ${attempt} - succeeding after process recovered`);
+  return { attempt, status: 'recovered' };
+}
+
+export async function processExitResilienceWorkflow() {
+  'use workflow';
+  console.log('Starting process exit resilience workflow');
+
+  // This step should crash the process on the first attempt,
+  // but the workflow should recover and retry on a new process
+  const result = await stepThatExitsOnFirstAttempt();
+
+  console.log(
+    `Workflow completed successfully after process crash: ${JSON.stringify(result)}`
+  );
+  return result;
+}