fix status of execution with for-each and if-then

Author: kevinkim-ogp

packages/backend/src/helpers/compute-for-each-parameters.ts

@@ -103,8 +103,8 @@ export function computeForEachParameters({
   }
 
   if (
-    executionStep.appKey === TOOLBOX_APP_KEY &&
-    executionStep.key === TOOLBOX_ACTIONS.FOR_EACH
+    executionStep?.appKey === TOOLBOX_APP_KEY &&
+    executionStep?.key === TOOLBOX_ACTIONS.FOR_EACH
   ) {
     dataValue = get(data, forEachKeyPath as string) || ''
   } else if (currentStepIndex > forEachStepIndex) {

packages/backend/src/models/execution-step.ts

@@ -1,5 +1,7 @@
 import { IExecutionStepMetadata, IJSONObject } from '@plumber/types'
 
+import { raw } from 'objection'
+
 import appConfig from '@/config/app'
 
 import Base from './base'
@@ -81,6 +83,68 @@ class ExecutionStep extends Base {
 
     return `${appConfig.baseUrl}/apps/${this.appKey}/assets/favicon.svg`
   }
+
+  static async getForEachExecutionSteps(executionId: string) {
+    return ExecutionStep.query()
+      .with('latest_steps', (builder) => {
+        /**
+         * NOTE: there is a known issue with knex where 'groupBy' are placed at the end of the 'unionAll' query.
+         * the workaround is to unionAll both queries with 'true' to wrap the subequery.
+         */
+        builder
+          .unionAll((qb) => {
+            qb.select(
+              'step_id',
+              raw('max(created_at) as max_created_at'),
+              raw('min(created_at) as min_created_at'),
+            )
+              .from('execution_steps')
+              .groupBy('step_id')
+              .where('execution_id', '=', executionId)
+              .where(raw("metadata = '{}'::jsonb"))
+              .withSoftDeleted()
+          }, true)
+          .unionAll((qb) => {
+            qb.select(
+              'step_id',
+              raw('max(created_at) as max_created_at'),
+              raw('min(created_at) as min_created_at'),
+            )
+              .from('execution_steps')
+              .groupBy('step_id', raw("metadata->>'iteration'"))
+              .where('execution_id', '=', executionId)
+              .where(raw("metadata != '{}'::jsonb"))
+              .withSoftDeleted()
+          }, true)
+          .withSoftDeleted()
+      })
+      .join('latest_steps', (builder) => {
+        builder
+          .on('execution_steps.step_id', '=', 'latest_steps.step_id')
+          .andOn(
+            'execution_steps.created_at',
+            '=',
+            'latest_steps.max_created_at',
+          )
+      })
+      .select('execution_steps.*', 'min_created_at')
+      .withSoftDeleted()
+      .orderBy('min_created_at', 'asc')
+  }
+
+  static async getForEachExecutionState(executionId: string) {
+    const executionSteps = await ExecutionStep.getForEachExecutionSteps(
+      executionId,
+    )
+    return {
+      hasLastIterationRun: executionSteps.some(
+        (step) => step.metadata?.isLastIteration && step.metadata?.isLastStep,
+      ),
+      areAllStepsSuccessful: executionSteps.every(
+        (step) => step.status === 'success',
+      ),
+    }
+  }
 }
 
 export default ExecutionStep

packages/backend/src/services/action.ts

@@ -171,6 +171,20 @@ export const processAction = async (options: ProcessActionOptions) => {
       ? 'success'
       : 'failure'
 
+  /**
+   * FOR-EACH + IF-THEN SPECIAL CASE
+   * when there are if-then actions in the for-each, not all steps may run so the lastStep check may not work
+   * if-then uses stop-execution to terminate the flow, so we need to set the isLastStep to true
+   * so that the execution status is set to success
+   */
+  if (
+    !testRun &&
+    forEachStepIndex > -1 &&
+    runResult.nextStep?.command === 'stop-execution'
+  ) {
+    metadata.isLastStep = true
+  }
+
   const executionStep = await execution
     .$relatedQuery('executionSteps')
     .insertAndFetch({

packages/backend/src/workers/helpers/make-action-worker.ts

@@ -168,35 +168,50 @@ export function makeActionWorker(
 
         /**
          * FOR-EACH SPECIAL CASE
-         * 1. nextStep is null for the for-each execution step, return if its not the last iteration
-         *    so that we do not prematurely set the execution status to success and it can be
-         *    reflected accurately as waiting
-         * 2. if any iteration fails, the execution is set to failed
-         * 3. if all iterations are successful, the execution is set to success
+         * nextStep is null for the for-each execution step, return if its not the last iteration
+         * so that we do not prematurely set the execution status to success and it can be
+         * reflected accurately as waiting
          */
         if (!nextStep && isForEach) {
           return
         }
 
         if (!nextStep) {
+          /**
+           * FOR-EACH SPECIAL CASE
+           * default state is null (waiting for all iterations to execute)
+           * if any iteration fails, the execution is immediately set to failure
+           * if all iterations are successful, the execution is set to success
+           */
           if (nextStepMetadata?.iteration) {
+            const { hasLastIterationRun, areAllStepsSuccessful } =
+              await ExecutionStep.getForEachExecutionState(executionId)
+
+            // end of the execution: all iterations ran successfully up to the last iteration and last step
             if (
               nextStepMetadata?.isLastIteration &&
               nextStepMetadata?.isLastStep
             ) {
-              const executionSteps = await ExecutionStep.query().where({
-                execution_id: executionId,
-              })
-
-              const areAllStepsSuccessful = executionSteps.every(
-                (step) => step.status === 'success',
-              )
               if (!areAllStepsSuccessful) {
                 await Execution.setStatus(executionId, 'failure')
                 return
               }
             } else {
-              return
+              // handle failures and retries
+              if (nextStepMetadata?.isLastStep) {
+                if (!areAllStepsSuccessful) {
+                  await Execution.setStatus(executionId, 'failure')
+                  return
+                }
+
+                // if the last iteration has not run, it means this flow is still executing
+                // we return early to preserve the waiting state of the execution
+                if (!hasLastIterationRun) {
+                  return
+                }
+              } else {
+                return
+              }
             }
           }
 

packages/types/index.d.ts

@@ -127,6 +127,8 @@ export interface IExecutionStep {
 export interface IExecutionStepMetadata {
   isMock?: boolean
   iteration?: number
+  isLastIteration?: boolean
+  isLastStep?: boolean
 }
 
 export interface IExecution {