Merge branch 'master' into fix/handle-manual-prices

.github/workflows/budibase_ci.yml

@@ -76,6 +76,18 @@ jobs:
             yarn check:types
           fi
 
+  helm-lint:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repo
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0
+
+      - name: Use Node.js 18.x
+        uses: azure/setup-helm@v3
+      - run: cd charts/budibase && helm lint .
+
   test-libraries:
     runs-on: ubuntu-latest
     steps:

charts/budibase/templates/app-service-deployment.yaml

@@ -227,6 +227,14 @@ spec:
         resources:
         {{- toYaml . | nindent 10 }}
         {{ end }}
+        {{ if .Values.services.apps.command }}
+        command:
+        {{- toYaml .Values.services.apps.command | nindent 10 }}
+        {{ end }}
+        {{ if .Values.services.apps.args }}
+        args:
+        {{- toYaml .Values.services.apps.args | nindent 10 }}
+        {{ end }}
     {{- with .Values.affinity }}
       affinity:
         {{- toYaml . | nindent 8 }}
@@ -244,12 +252,4 @@ spec:
       {{ end }}
       restartPolicy: Always
       serviceAccountName: ""
-      {{ if .Values.services.apps.command }}
-      command:
-      {{- toYaml .Values.services.apps.command | nindent 8 }}
-      {{ end }}
-      {{ if .Values.services.apps.args }}
-      args:
-      {{- toYaml .Values.services.apps.args | nindent 8 }}
-      {{ end }}
 status: {}

charts/budibase/templates/automation-worker-service-deployment.yaml

@@ -227,6 +227,13 @@ spec:
         resources:
         {{- toYaml . | nindent 10 }}
         {{ end }}
+        command:
+        {{- toYaml .Values.services.automationWorkers.command | nindent 10 }}
+        {{ end }}
+        {{ if .Values.services.automationWorkers.args }}
+        args:
+        {{- toYaml .Values.services.automationWorkers.args | nindent 10 }}
+        {{ end }}
     {{- with .Values.affinity }}
       affinity:
         {{- toYaml . | nindent 8 }}
@@ -245,12 +252,5 @@ spec:
       restartPolicy: Always
       serviceAccountName: ""
       {{ if .Values.services.automationWorkers.command }}}
-      command:
-      {{- toYaml .Values.services.automationWorkers.command | nindent 8 }}
-      {{ end }}
-      {{ if .Values.services.automationWorkers.args }}
-      args:
-      {{- toYaml .Values.services.automationWorkers.args | nindent 8 }}
-      {{ end }}
 status: {}
 {{- end }}

charts/budibase/templates/worker-service-deployment.yaml

@@ -213,6 +213,14 @@ spec:
         resources:
         {{- toYaml . | nindent 10 }}
         {{ end }}
+        {{ if .Values.services.worker.command }}
+        command:
+        {{- toYaml .Values.services.worker.command | nindent 10 }}
+        {{ end }}
+        {{ if .Values.services.worker.args }}
+        args:
+        {{- toYaml .Values.services.worker.args | nindent 10 }}
+        {{ end }}
     {{- with .Values.affinity }}
       affinity:
         {{- toYaml . | nindent 8 }}
@@ -230,12 +238,4 @@ spec:
       {{ end }}
       restartPolicy: Always
       serviceAccountName: ""
-      {{ if .Values.services.worker.command }}
-      command:
-      {{- toYaml .Values.services.worker.command | nindent 8 }}
-      {{ end }}
-      {{ if .Values.services.worker.args }}
-      args:
-      {{- toYaml .Values.services.worker.args | nindent 8 }}
-      {{ end }}
 status: {}

hosting/single/runner.sh

@@ -79,7 +79,7 @@ redis-server --requirepass $REDIS_PASSWORD > /dev/stdout 2>&1 &
 /bbcouch-runner.sh &
 
 # only start minio if use s3 isn't passed
-if [[ -z "${USE_S3}"]]; then
+if [[ -z "${USE_S3}" ]]; then
   /minio/minio server --console-address ":9001" ${DATA_DIR}/minio > /dev/stdout 2>&1 &
 fi
 

packages/backend-core/src/queue/inMemoryQueue.ts

@@ -15,6 +15,7 @@ function newJob(queue: string, message: any) {
     timestamp: Date.now(),
     queue: queue,
     data: message,
+    opts: {},
   }
 }
 

packages/backend-core/src/timers/timers.ts

@@ -30,7 +30,7 @@ export class ExecutionTimeTracker {
     return new ExecutionTimeTracker(limitMs)
   }
 
-  constructor(private limitMs: number) {}
+  constructor(readonly limitMs: number) {}
 
   private totalTimeMs = 0
 
@@ -46,6 +46,10 @@ export class ExecutionTimeTracker {
     }
   }
 
+  get elapsedMS() {
+    return this.totalTimeMs
+  }
+
   private checkLimit() {
     if (this.totalTimeMs > this.limitMs) {
       throw new ExecutionTimeoutError(

packages/server/src/automations/utils.ts

@@ -16,6 +16,7 @@ import {
 } from "@budibase/types"
 import sdk from "../sdk"
 import { automationsEnabled } from "../features"
+import tracer from "dd-trace"
 
 const REBOOT_CRON = "@reboot"
 const WH_STEP_ID = definitions.WEBHOOK.stepId
@@ -39,26 +40,62 @@ function loggingArgs(job: AutomationJob) {
 }
 
 export async function processEvent(job: AutomationJob) {
-  const appId = job.data.event.appId!
+  return tracer.trace(
-  const automationId = job.data.automation._id!
+    "processEvent",
-  const task = async () => {
+    { resource: "automation" },
-    try {
+    async span => {
-      // need to actually await these so that an error can be captured properly
+      const appId = job.data.event.appId!
-      console.log("automation running", ...loggingArgs(job))
+      const automationId = job.data.automation._id!
 
-      const runFn = () => Runner.run(job)
+      span?.addTags({
-      const result = await quotas.addAutomation(runFn, {
+        appId,
         automationId,
+        job: {
+          id: job.id,
+          name: job.name,
+          attemptsMade: job.attemptsMade,
+          opts: {
+            attempts: job.opts.attempts,
+            priority: job.opts.priority,
+            delay: job.opts.delay,
+            repeat: job.opts.repeat,
+            backoff: job.opts.backoff,
+            lifo: job.opts.lifo,
+            timeout: job.opts.timeout,
+            jobId: job.opts.jobId,
+            removeOnComplete: job.opts.removeOnComplete,
+            removeOnFail: job.opts.removeOnFail,
+            stackTraceLimit: job.opts.stackTraceLimit,
+            preventParsingData: job.opts.preventParsingData,
+          },
+        },
       })
-      console.log("automation completed", ...loggingArgs(job))
-      return result
-    } catch (err) {
-      console.error(`automation was unable to run`, err, ...loggingArgs(job))
-      return { err }
-    }
-  }
 
-  return await context.doInAutomationContext({ appId, automationId, task })
+      const task = async () => {
+        try {
+          // need to actually await these so that an error can be captured properly
+          console.log("automation running", ...loggingArgs(job))
+
+          const runFn = () => Runner.run(job)
+          const result = await quotas.addAutomation(runFn, {
+            automationId,
+          })
+          console.log("automation completed", ...loggingArgs(job))
+          return result
+        } catch (err) {
+          span?.addTags({ error: true })
+          console.error(
+            `automation was unable to run`,
+            err,
+            ...loggingArgs(job)
+          )
+          return { err }
+        }
+      }
+
+      return await context.doInAutomationContext({ appId, automationId, task })
+    }
+  )
 }
 
 export async function updateTestHistory(

packages/server/src/integrations/rest.ts

@@ -131,7 +131,10 @@ class RestIntegration implements IntegrationBase {
     let data, raw, headers
     const contentType = response.headers.get("content-type") || ""
     try {
-      if (contentType.includes("application/json")) {
+      if (response.status === 204) {
+        data = []
+        raw = []
+      } else if (contentType.includes("application/json")) {
         data = await response.json()
         raw = JSON.stringify(data)
       } else if (

packages/server/src/integrations/tests/rest.spec.ts

@@ -186,9 +186,15 @@ describe("REST Integration", () => {
   })
 
   describe("response", () => {
-    function buildInput(json: any, text: any, header: any) {
+    const contentTypes = ["application/json", "text/plain", "application/xml"]
+    function buildInput(
+      json: any,
+      text: any,
+      header: any,
+      status: number = 200
+    ) {
       return {
-        status: 200,
+        status,
         json: json ? async () => json : undefined,
         text: text ? async () => text : undefined,
         headers: {
@@ -225,6 +231,18 @@ describe("REST Integration", () => {
       expect(output.extra.raw).toEqual(text)
       expect(output.extra.headers["content-type"]).toEqual("application/xml")
     })
+
+    test.each(contentTypes)(
+      "should not throw an error on 204 no content",
+      async contentType => {
+        const input = buildInput(undefined, null, contentType, 204)
+        const output = await config.integration.parseResponse(input)
+        expect(output.data).toEqual([])
+        expect(output.extra.raw).toEqual([])
+        expect(output.info.code).toEqual(204)
+        expect(output.extra.headers["content-type"]).toEqual(contentType)
+      }
+    )
   })
 
   describe("authentication", () => {

packages/server/src/jsRunner.ts

@@ -2,35 +2,44 @@ import vm from "vm"
 import env from "./environment"
 import { setJSRunner } from "@budibase/string-templates"
 import { context, timers } from "@budibase/backend-core"
+import tracer from "dd-trace"
 
 type TrackerFn = <T>(f: () => T) => T
 
 export function init() {
   setJSRunner((js: string, ctx: vm.Context) => {
-    const perRequestLimit = env.JS_PER_REQUEST_TIME_LIMIT_MS
+    return tracer.trace("runJS", {}, span => {
-    let track: TrackerFn = f => f()
+      const perRequestLimit = env.JS_PER_REQUEST_TIME_LIMIT_MS
-    if (perRequestLimit) {
+      let track: TrackerFn = f => f()
-      const bbCtx = context.getCurrentContext()
+      if (perRequestLimit) {
-      if (bbCtx) {
+        const bbCtx = context.getCurrentContext()
-        if (!bbCtx.jsExecutionTracker) {
+        if (bbCtx) {
-          bbCtx.jsExecutionTracker =
+          if (!bbCtx.jsExecutionTracker) {
-            timers.ExecutionTimeTracker.withLimit(perRequestLimit)
+            bbCtx.jsExecutionTracker =
+              timers.ExecutionTimeTracker.withLimit(perRequestLimit)
+          }
+          track = bbCtx.jsExecutionTracker.track.bind(bbCtx.jsExecutionTracker)
+          span?.addTags({
+            js: {
+              limitMS: bbCtx.jsExecutionTracker.limitMs,
+              elapsedMS: bbCtx.jsExecutionTracker.elapsedMS,
+            },
+          })
         }
-        track = bbCtx.jsExecutionTracker.track.bind(bbCtx.jsExecutionTracker)
       }
-    }
 
-    ctx = {
+      ctx = {
-      ...ctx,
+        ...ctx,
-      alert: undefined,
+        alert: undefined,
-      setInterval: undefined,
+        setInterval: undefined,
-      setTimeout: undefined,
+        setTimeout: undefined,
-    }
+      }
-    vm.createContext(ctx)
+      vm.createContext(ctx)
-    return track(() =>
+      return track(() =>
-      vm.runInNewContext(js, ctx, {
+        vm.runInNewContext(js, ctx, {
-        timeout: env.JS_PER_EXECUTION_TIME_LIMIT_MS,
+          timeout: env.JS_PER_EXECUTION_TIME_LIMIT_MS,
-      })
+        })
-    )
+      )
+    })
   })
 }

packages/server/src/threads/automation.ts

@@ -34,6 +34,7 @@ import { cloneDeep } from "lodash/fp"
 import { performance } from "perf_hooks"
 import * as sdkUtils from "../sdk/utils"
 import env from "../environment"
+import tracer from "dd-trace"
 
 threadUtils.threadSetup()
 const FILTER_STEP_ID = actions.BUILTIN_ACTION_DEFINITIONS.FILTER.stepId
@@ -242,278 +243,347 @@ class Orchestrator {
   }
 
   async execute(): Promise<any> {
-    // this will retrieve from context created at start of thread
+    return tracer.trace(
-    this._context.env = await sdkUtils.getEnvironmentVariables()
+      "Orchestrator.execute",
-    let automation = this._automation
+      { resource: "automation" },
-    let stopped = false
+      async span => {
-    let loopStep: AutomationStep | undefined = undefined
+        span?.addTags({
+          appId: this._appId,
+          automationId: this._automation._id,
+        })
 
-    let stepCount = 0
+        // this will retrieve from context created at start of thread
-    let loopStepNumber: any = undefined
+        this._context.env = await sdkUtils.getEnvironmentVariables()
-    let loopSteps: LoopStep[] | undefined = []
+        let automation = this._automation
-    let metadata
+        let stopped = false
-    let timeoutFlag = false
+        let loopStep: AutomationStep | undefined = undefined
-    let wasLoopStep = false
-    let timeout = this._job.data.event.timeout
-    // check if this is a recurring automation,
-    if (isProdAppID(this._appId) && isRecurring(automation)) {
-      metadata = await this.getMetadata()
-      const shouldStop = await this.checkIfShouldStop(metadata)
-      if (shouldStop) {
-        return
-      }
-    }
-    const start = performance.now()
-    for (let step of automation.definition.steps) {
-      if (timeoutFlag) {
-        break
-      }
 
-      if (timeout) {
+        let stepCount = 0
-        setTimeout(() => {
+        let loopStepNumber: any = undefined
-          timeoutFlag = true
+        let loopSteps: LoopStep[] | undefined = []
-        }, timeout || 12000)
+        let metadata
-      }
+        let timeoutFlag = false
-
+        let wasLoopStep = false
-      stepCount++
+        let timeout = this._job.data.event.timeout
-      let input: any,
+        // check if this is a recurring automation,
-        iterations = 1,
+        if (isProdAppID(this._appId) && isRecurring(automation)) {
-        iterationCount = 0
+          span?.addTags({ recurring: true })
-
+          metadata = await this.getMetadata()
-      if (step.stepId === LOOP_STEP_ID) {
+          const shouldStop = await this.checkIfShouldStop(metadata)
-        loopStep = step
+          if (shouldStop) {
-        loopStepNumber = stepCount
+            span?.addTags({ shouldStop: true })
-        continue
+            return
-      }
-
-      if (loopStep) {
-        input = await processObject(loopStep.inputs, this._context)
-        iterations = getLoopIterations(loopStep as LoopStep)
-      }
-      for (let index = 0; index < iterations; index++) {
-        let originalStepInput = cloneDeep(step.inputs)
-        // Handle if the user has set a max iteration count or if it reaches the max limit set by us
-        if (loopStep && input.binding) {
-          let tempOutput = { items: loopSteps, iterations: iterationCount }
-          try {
-            loopStep.inputs.binding = automationUtils.typecastForLooping(
-              loopStep as LoopStep,
-              loopStep.inputs as LoopInput
-            )
-          } catch (err) {
-            this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
-              status: AutomationErrors.INCORRECT_TYPE,
-              success: false,
-            })
-            loopSteps = undefined
-            loopStep = undefined
-            break
-          }
-          let item = []
-          if (
-            typeof loopStep.inputs.binding === "string" &&
-            loopStep.inputs.option === "String"
-          ) {
-            item = automationUtils.stringSplit(loopStep.inputs.binding)
-          } else if (Array.isArray(loopStep.inputs.binding)) {
-            item = loopStep.inputs.binding
-          }
-          this._context.steps[loopStepNumber] = {
-            currentItem: item[index],
-          }
-
-          // The "Loop" binding in the front end is "fake", so replace it here so the context can understand it
-          // Pretty hacky because we need to account for the row object
-          for (let [key, value] of Object.entries(originalStepInput)) {
-            if (typeof value === "object") {
-              for (let [innerKey, innerValue] of Object.entries(
-                originalStepInput[key]
-              )) {
-                if (typeof innerValue === "string") {
-                  originalStepInput[key][innerKey] =
-                    automationUtils.substituteLoopStep(
-                      innerValue,
-                      `steps.${loopStepNumber}`
-                    )
-                } else if (typeof value === "object") {
-                  for (let [innerObject, innerValue] of Object.entries(
-                    originalStepInput[key][innerKey]
-                  )) {
-                    originalStepInput[key][innerKey][innerObject] =
-                      automationUtils.substituteLoopStep(
-                        innerValue as string,
-                        `steps.${loopStepNumber}`
-                      )
-                  }
-                }
-              }
-            } else {
-              if (typeof value === "string") {
-                originalStepInput[key] = automationUtils.substituteLoopStep(
-                  value,
-                  `steps.${loopStepNumber}`
-                )
-              }
-            }
-          }
-
-          if (
-            index === env.AUTOMATION_MAX_ITERATIONS ||
-            index === parseInt(loopStep.inputs.iterations)
-          ) {
-            this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
-              status: AutomationErrors.MAX_ITERATIONS,
-              success: true,
-            })
-            loopSteps = undefined
-            loopStep = undefined
-            break
-          }
-
-          let isFailure = false
-          const currentItem = this._context.steps[loopStepNumber]?.currentItem
-          if (currentItem && typeof currentItem === "object") {
-            isFailure = Object.keys(currentItem).some(value => {
-              return currentItem[value] === loopStep?.inputs.failure
-            })
-          } else {
-            isFailure = currentItem && currentItem === loopStep.inputs.failure
-          }
-
-          if (isFailure) {
-            this.updateContextAndOutput(loopStepNumber, step, tempOutput, {
-              status: AutomationErrors.FAILURE_CONDITION,
-              success: false,
-            })
-            loopSteps = undefined
-            loopStep = undefined
-            break
           }
         }
-
+        const start = performance.now()
-        // execution stopped, record state for that
+        for (let step of automation.definition.steps) {
-        if (stopped) {
+          const stepSpan = tracer.startSpan("Orchestrator.execute.step", {
-          this.updateExecutionOutput(step.id, step.stepId, {}, STOPPED_STATUS)
+            childOf: span,
-          continue
+          })
-        }
+          stepSpan.addTags({
-
+            resource: "automation",
-        // If it's a loop step, we need to manually add the bindings to the context
+            step: {
-        let stepFn = await this.getStepFunctionality(step.stepId)
+              stepId: step.stepId,
-        let inputs = await processObject(originalStepInput, this._context)
+              id: step.id,
-        inputs = automationUtils.cleanInputValues(inputs, step.schema.inputs)
+              name: step.name,
-
+              type: step.type,
-        try {
+              title: step.stepTitle,
-          // appId is always passed
+              internal: step.internal,
-          const outputs = await stepFn({
+              deprecated: step.deprecated,
-            inputs: inputs,
+            },
-            appId: this._appId,
-            emitter: this._emitter,
-            context: this._context,
           })
 
-          this._context.steps[stepCount] = outputs
+          let input: any,
-          // if filter causes us to stop execution don't break the loop, set a var
+            iterations = 1,
-          // so that we can finish iterating through the steps and record that it stopped
+            iterationCount = 0
-          if (step.stepId === FILTER_STEP_ID && !outputs.result) {
-            stopped = true
-            this.updateExecutionOutput(step.id, step.stepId, step.inputs, {
-              ...outputs,
-              ...STOPPED_STATUS,
-            })
-            continue
-          }
-          if (loopStep && loopSteps) {
-            loopSteps.push(outputs)
-          } else {
-            this.updateExecutionOutput(
-              step.id,
-              step.stepId,
-              step.inputs,
-              outputs
-            )
-          }
-        } catch (err) {
-          console.error(`Automation error - ${step.stepId} - ${err}`)
-          return err
-        }
 
-        if (loopStep) {
+          try {
-          iterationCount++
+            if (timeoutFlag) {
-          if (index === iterations - 1) {
+              span?.addTags({ timedOut: true })
+              break
+            }
+
+            if (timeout) {
+              setTimeout(() => {
+                timeoutFlag = true
+              }, timeout || 12000)
+            }
+
+            stepCount++
+            if (step.stepId === LOOP_STEP_ID) {
+              loopStep = step
+              loopStepNumber = stepCount
+              continue
+            }
+
+            if (loopStep) {
+              input = await processObject(loopStep.inputs, this._context)
+              iterations = getLoopIterations(loopStep as LoopStep)
+              stepSpan?.addTags({ step: { iterations } })
+            }
+            for (let index = 0; index < iterations; index++) {
+              let originalStepInput = cloneDeep(step.inputs)
+              // Handle if the user has set a max iteration count or if it reaches the max limit set by us
+              if (loopStep && input.binding) {
+                let tempOutput = {
+                  items: loopSteps,
+                  iterations: iterationCount,
+                }
+                try {
+                  loopStep.inputs.binding = automationUtils.typecastForLooping(
+                    loopStep as LoopStep,
+                    loopStep.inputs as LoopInput
+                  )
+                } catch (err) {
+                  this.updateContextAndOutput(
+                    loopStepNumber,
+                    step,
+                    tempOutput,
+                    {
+                      status: AutomationErrors.INCORRECT_TYPE,
+                      success: false,
+                    }
+                  )
+                  loopSteps = undefined
+                  loopStep = undefined
+                  break
+                }
+                let item = []
+                if (
+                  typeof loopStep.inputs.binding === "string" &&
+                  loopStep.inputs.option === "String"
+                ) {
+                  item = automationUtils.stringSplit(loopStep.inputs.binding)
+                } else if (Array.isArray(loopStep.inputs.binding)) {
+                  item = loopStep.inputs.binding
+                }
+                this._context.steps[loopStepNumber] = {
+                  currentItem: item[index],
+                }
+
+                // The "Loop" binding in the front end is "fake", so replace it here so the context can understand it
+                // Pretty hacky because we need to account for the row object
+                for (let [key, value] of Object.entries(originalStepInput)) {
+                  if (typeof value === "object") {
+                    for (let [innerKey, innerValue] of Object.entries(
+                      originalStepInput[key]
+                    )) {
+                      if (typeof innerValue === "string") {
+                        originalStepInput[key][innerKey] =
+                          automationUtils.substituteLoopStep(
+                            innerValue,
+                            `steps.${loopStepNumber}`
+                          )
+                      } else if (typeof value === "object") {
+                        for (let [innerObject, innerValue] of Object.entries(
+                          originalStepInput[key][innerKey]
+                        )) {
+                          originalStepInput[key][innerKey][innerObject] =
+                            automationUtils.substituteLoopStep(
+                              innerValue as string,
+                              `steps.${loopStepNumber}`
+                            )
+                        }
+                      }
+                    }
+                  } else {
+                    if (typeof value === "string") {
+                      originalStepInput[key] =
+                        automationUtils.substituteLoopStep(
+                          value,
+                          `steps.${loopStepNumber}`
+                        )
+                    }
+                  }
+                }
+
+                if (
+                  index === env.AUTOMATION_MAX_ITERATIONS ||
+                  index === parseInt(loopStep.inputs.iterations)
+                ) {
+                  this.updateContextAndOutput(
+                    loopStepNumber,
+                    step,
+                    tempOutput,
+                    {
+                      status: AutomationErrors.MAX_ITERATIONS,
+                      success: true,
+                    }
+                  )
+                  loopSteps = undefined
+                  loopStep = undefined
+                  break
+                }
+
+                let isFailure = false
+                const currentItem =
+                  this._context.steps[loopStepNumber]?.currentItem
+                if (currentItem && typeof currentItem === "object") {
+                  isFailure = Object.keys(currentItem).some(value => {
+                    return currentItem[value] === loopStep?.inputs.failure
+                  })
+                } else {
+                  isFailure =
+                    currentItem && currentItem === loopStep.inputs.failure
+                }
+
+                if (isFailure) {
+                  this.updateContextAndOutput(
+                    loopStepNumber,
+                    step,
+                    tempOutput,
+                    {
+                      status: AutomationErrors.FAILURE_CONDITION,
+                      success: false,
+                    }
+                  )
+                  loopSteps = undefined
+                  loopStep = undefined
+                  break
+                }
+              }
+
+              // execution stopped, record state for that
+              if (stopped) {
+                this.updateExecutionOutput(
+                  step.id,
+                  step.stepId,
+                  {},
+                  STOPPED_STATUS
+                )
+                continue
+              }
+
+              // If it's a loop step, we need to manually add the bindings to the context
+              let stepFn = await this.getStepFunctionality(step.stepId)
+              let inputs = await processObject(originalStepInput, this._context)
+              inputs = automationUtils.cleanInputValues(
+                inputs,
+                step.schema.inputs
+              )
+
+              try {
+                // appId is always passed
+                const outputs = await stepFn({
+                  inputs: inputs,
+                  appId: this._appId,
+                  emitter: this._emitter,
+                  context: this._context,
+                })
+
+                this._context.steps[stepCount] = outputs
+                // if filter causes us to stop execution don't break the loop, set a var
+                // so that we can finish iterating through the steps and record that it stopped
+                if (step.stepId === FILTER_STEP_ID && !outputs.result) {
+                  stopped = true
+                  this.updateExecutionOutput(
+                    step.id,
+                    step.stepId,
+                    step.inputs,
+                    {
+                      ...outputs,
+                      ...STOPPED_STATUS,
+                    }
+                  )
+                  continue
+                }
+                if (loopStep && loopSteps) {
+                  loopSteps.push(outputs)
+                } else {
+                  this.updateExecutionOutput(
+                    step.id,
+                    step.stepId,
+                    step.inputs,
+                    outputs
+                  )
+                }
+              } catch (err) {
+                console.error(`Automation error - ${step.stepId} - ${err}`)
+                return err
+              }
+
+              if (loopStep) {
+                iterationCount++
+                if (index === iterations - 1) {
+                  loopStep = undefined
+                  this._context.steps.splice(loopStepNumber, 1)
+                  break
+                }
+              }
+            }
+          } finally {
+            stepSpan?.finish()
+          }
+
+          if (loopStep && iterations === 0) {
             loopStep = undefined
+            this.executionOutput.steps.splice(loopStepNumber + 1, 0, {
+              id: step.id,
+              stepId: step.stepId,
+              outputs: {
+                status: AutomationStepStatus.NO_ITERATIONS,
+                success: true,
+              },
+              inputs: {},
+            })
+
             this._context.steps.splice(loopStepNumber, 1)
-            break
+            iterations = 1
+          }
+
+          // Delete the step after the loop step as it's irrelevant, since information is included
+          // in the loop step
+          if (wasLoopStep && !loopStep) {
+            this._context.steps.splice(loopStepNumber + 1, 1)
+            wasLoopStep = false
+          }
+          if (loopSteps && loopSteps.length) {
+            let tempOutput = {
+              success: true,
+              items: loopSteps,
+              iterations: iterationCount,
+            }
+            this.executionOutput.steps.splice(loopStepNumber + 1, 0, {
+              id: step.id,
+              stepId: step.stepId,
+              outputs: tempOutput,
+              inputs: step.inputs,
+            })
+            this._context.steps[loopStepNumber] = tempOutput
+
+            wasLoopStep = true
+            loopSteps = []
           }
         }
-      }
 
-      if (loopStep && iterations === 0) {
+        const end = performance.now()
-        loopStep = undefined
+        const executionTime = end - start
-        this.executionOutput.steps.splice(loopStepNumber + 1, 0, {
-          id: step.id,
-          stepId: step.stepId,
-          outputs: {
-            status: AutomationStepStatus.NO_ITERATIONS,
-            success: true,
-          },
-          inputs: {},
-        })
 
-        this._context.steps.splice(loopStepNumber, 1)
+        console.info(
-        iterations = 1
+          `Automation ID: ${automation._id} Execution time: ${executionTime} milliseconds`,
-      }
+          {
+            _logKey: "automation",
+            executionTime,
+          }
+        )
 
-      // Delete the step after the loop step as it's irrelevant, since information is included
+        // store the logs for the automation run
-      // in the loop step
+        try {
-      if (wasLoopStep && !loopStep) {
+          await storeLog(this._automation, this.executionOutput)
-        this._context.steps.splice(loopStepNumber + 1, 1)
+        } catch (e: any) {
-        wasLoopStep = false
+          if (e.status === 413 && e.request?.data) {
-      }
+            // if content is too large we shouldn't log it
-      if (loopSteps && loopSteps.length) {
+            delete e.request.data
-        let tempOutput = {
+            e.request.data = { message: "removed due to large size" }
-          success: true,
+          }
-          items: loopSteps,
+          logging.logAlert("Error writing automation log", e)
-          iterations: iterationCount,
         }
-        this.executionOutput.steps.splice(loopStepNumber + 1, 0, {
+        if (isProdAppID(this._appId) && isRecurring(automation) && metadata) {
-          id: step.id,
+          await this.updateMetadata(metadata)
-          stepId: step.stepId,
+        }
-          outputs: tempOutput,
+        return this.executionOutput
-          inputs: step.inputs,
-        })
-        this._context.steps[loopStepNumber] = tempOutput
-
-        wasLoopStep = true
-        loopSteps = []
-      }
-    }
-
-    const end = performance.now()
-    const executionTime = end - start
-
-    console.info(
-      `Automation ID: ${automation._id} Execution time: ${executionTime} milliseconds`,
-      {
-        _logKey: "automation",
-        executionTime,
       }
     )
-
-    // store the logs for the automation run
-    try {
-      await storeLog(this._automation, this.executionOutput)
-    } catch (e: any) {
-      if (e.status === 413 && e.request?.data) {
-        // if content is too large we shouldn't log it
-        delete e.request.data
-        e.request.data = { message: "removed due to large size" }
-      }
-      logging.logAlert("Error writing automation log", e)
-    }
-    if (isProdAppID(this._appId) && isRecurring(automation) && metadata) {
-      await this.updateMetadata(metadata)
-    }
-    return this.executionOutput
   }
 }