state-machine.asl.yml

StartAt: Normalize Input
States:
  # Start of pre-processing
  Normalize Input:
    Comment: Normalizes input data
    Type: Task
    InputPath: "$"
    Resource: "${NormalizeInputLambdaFunctionArn}"
    Parameters:
      Input.$: "$"
      StateMachine:
        Id.$: "$$.StateMachine.Id"
    ResultPath: "$"
    OutputPath: "$"
    Next: JobReceived Callbacks Map
    Retry:
      - ErrorEquals:
          - MissingTaskTypeError
        MaxAttempts: 0
      - ErrorEquals:
          - States.ALL
        IntervalSeconds: 1
        MaxAttempts: 2
        BackoffRate: 1
    Catch:
      - ErrorEquals:
          - States.ALL
        ResultPath: "$.Error"
        Next: Normalize Input Catcher
  Normalize Input Catcher:
    Comment: Add a parameter so normalize errors can be identified
    Type: Pass
    InputPath: "$"
    Result: NORMALIZE_INPUT_ERROR
    ResultPath: "$.State"
    OutputPath: "$"
    Next: Add Empty TaskResults

  JobReceived Callbacks Map:
    Type: Map
    Comment: Iterates over all callback endpoints to indicate the job was received
    InputPath: "$"
    ItemsPath: "$.Job.Callbacks"
    Parameters:
      Callback.$: "$$.Map.Item.Value"
      StateMachine:
        Id.$: "$$.StateMachine.Id"
      Execution:
        Id.$: "$$.Execution.Id"
      Message: # The JSON of this value will be sent to endpoints
        JobReceived:
          Job:
            Id.$: "$.Job.Id"
          Execution:
            Id.$: "$$.Execution.Id"
          State: RECEIVED
    ResultPath: "$.Void" # The output of the iterator states is discarded
    OutputPath: "$"
    Next: Ingest Source File
    MaxConcurrency: 0
    Iterator:
      StartAt: Send JobReceived Callback
      States:
        Send JobReceived Callback:
          Type: Task
          Comment: >-
            Sends a job received message for a single callback endpoint in
            the iterator
          InputPath: "$"
          Resource: "${CallbackLambdaFunctionArn}"
          ResultPath: "$"
          OutputPath: "$"
          End: true
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2

  Ingest Source File:
    Comment: Creates an artifact of the source file in S3
    Type: Task
    InputPath: "$"
    Parameters:
      Job:
        Source.$: "$.Job.Source"
      Execution:
        Id.$: "$$.Execution.Id"
    Resource: "${IngestLambdaFunctionArn}"
    ResultPath: "$.Artifact"
    OutputPath: "$"
    Next: Detect Source File Type
    Retry:
      - ErrorEquals:
          - UnknownSourceModeError
        MaxAttempts: 0
      - ErrorEquals:
          - States.ALL
        IntervalSeconds: 5
        MaxAttempts: 3
        BackoffRate: 2
    Catch:
      - ErrorEquals:
          - States.ALL
        ResultPath: "$.Error"
        Next: Ingest Source File Catcher
  Ingest Source File Catcher:
    Comment: Add a parameter so ingest errors can be identified
    Type: Pass
    InputPath: "$"
    Result: SOURCE_FILE_INGEST_ERROR
    ResultPath: "$.State"
    OutputPath: "$"
    Next: Add Empty TaskResults

  Detect Source File Type:
    Comment: Detects the file type of the source file artifact
    Type: Task
    InputPath: "$"
    Parameters:
      Artifact.$: "$.Artifact"
    Resource: "${SourceTypeLambdaFunctionArn}"
    ResultPath: "$.Artifact.Descriptor"
    OutputPath: "$"
    Next: Tasks Map
    Retry:
      - ErrorEquals:
          - States.ALL
        IntervalSeconds: 5
        MaxAttempts: 3
        BackoffRate: 2
    Catch:
      - ErrorEquals:
          - States.ALL
        ResultPath: "$.Error"
        Next: Detect Source File Type Catcher
  Detect Source File Type Catcher:
    Comment: Add a parameter so type detection errors can be identified
    Type: Pass
    InputPath: "$"
    Result: SOURCE_FILE_TYPE_DETECTION_ERROR
    ResultPath: "$.State"
    OutputPath: "$"
    Next: Add Empty TaskResults
  # End of pre-processing

  # Start of task execution
  Tasks Map:
    # The key output of this map state is the TaskResults array. There will
    # be one element for each task, whether it succeeded or failed. Successful
    # task elements will have a `Task` property, and failed tasks will have a
    # `Type` property.
    Type: Map
    Comment: Iterates over all tasks included in the job
    InputPath: "$"
    ItemsPath: "$.Job.Tasks"
    Parameters:
      # Parameters' value is what is passed to each state within the iterator,
      # but each state decides independetly what is passed to its backing
      # resource (Lambda, etc). For example, Callbacks is available to each
      # state, but usually not passed into Lambdas, since most tasks don't
      # utilize that data.
      Job:
        Id.$: "$.Job.Id"
      Execution:
        Id.$: "$$.Execution.Id"
      Artifact.$: "$.Artifact"
      TaskIndex.$: "$$.Map.Item.Index"
      Task.$: "$$.Map.Item.Value" # This value is a task defined in the job input, e.g., { "Type": "Copy" }
      Callbacks.$: "$.Job.Callbacks"
    ResultPath: "$.TaskResults"
    OutputPath: "$"
    Next: JobResult Callback Map
    MaxConcurrency: 0
    Iterator:
      StartAt: Route Task By Type
      States:
        Route Task By Type:
          Type: Choice
          Comment: >-
            For the current task being iterated, routes to the appropriate
            state
          Default: Unknown Task Type
          Choices:
            - Variable: "$.Task.Type"
              StringEquals: Inspect
              Next: Inspect Task Handler
            - And:
                - Variable: "$.Task.Type"
                  StringEquals: Copy
                - Variable: "$.Task.Mode"
                  StringEquals: AWS/S3
              Next: Copy Task Handler
            - And:
                - Variable: "$.Task.Type"
                  StringEquals: Copy
                - Variable: "$.Task.Mode"
                  StringEquals: FTP/Passive
              Next: FTP Copy Task Fargate Execution
            - And:
                - Variable: "$.Task.Type"
                  StringEquals: Copy
                - Variable: "$.Task.Mode"
                  StringEquals: FTP/Active
              Next: FTP Copy Task Fargate Execution
            - And:
                - Variable: "$.Task.Type"
                  StringEquals: Copy
                - Variable: "$.Task.Mode"
                  StringEquals: FTP/Auto
              Next: FTP Copy Task Fargate Execution
            - Variable: "$.Task.Type"
              StringEquals: Transcode
              Next: Transcode Task Fargate Execution
            - Variable: "$.Task.Type"
              StringEquals: Image
              Next: Image Task Handler
            - Variable: "$.Task.Type"
              StringEquals: Transcribe
              Next: Transcribe Task Job Start
            - Variable: "$.Task.Type"
              StringEquals: WavWrap
              Next: WAV Wrap Task Handler
            - Variable: "$.Task.Type"
              StringEquals: DetectSilence
              Next: Detect Silence Task Handler
            - Variable: "$.Task.Type"
              StringEquals: DetectTone
              Next: Detect Tone Task Handler
            - Variable: "$.Task.Type"
              StringEquals: Waveform
              Next: Waveform Task Handler

        # This is a no-op. No callbacks are sent, and any record of this task
        # will be filtered out of the job result callbacks
        Unknown Task Type:
          Type: Pass
          Comment: Handles and blackholes tasks with an unknown type
          End: true
          ResultPath: "$"
          Result:
            Task: "Null"

        # Task operations
        # - The InputPath and OutputPath of all of these must be $, so that
        #   all data is available to states down the line.
        # - The ResultPath for states that are returning the final output of a
        #   task's operation (i.e., states immediately prior to the callbacks
        #   map) must be $.TaskResult, and their return value must be the
        #   expected standard task output (e.g., { Task: 'Copy' })
        # - All task operation states, not just the final state, should catch
        #   to the TaskResult Error Callback Map.
        Inspect Task Handler:
          Type: Task
          Comment: Inspects the artifact file
          Resource: "${InspectMediaLambdaFunctionArn}"
          InputPath: "$"
          Parameters:
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"
          ResultPath: "$.TaskResult"
          OutputPath: "$"
          Next: TaskResult Callbacks Map
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        Copy Task Handler:
          Type: Task
          Comment: Copies the artifact
          Resource: "${CopyLambdaFunctionArn}"
          InputPath: "$"
          Parameters:
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"
          ResultPath: "$.TaskResult"
          OutputPath: "$"
          Next: TaskResult Callbacks Map
          Retry:
            - ErrorEquals:
                - UnknownCopyTaskModeError
              MaxAttempts: 0
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        FTP Copy Task Fargate Execution:
          Type: Task
          Comment: Sends artifact to an FTP destination
          Resource: arn:aws:states:::ecs:runTask.sync
          InputPath: "$"
          Parameters:
            Cluster: "${EcsClusterArn}"
            LaunchType: FARGATE
            Overrides:
              ContainerOverrides:
                - Environment:
                    - Name: STATE_MACHINE_ARN
                      Value.$: "$$.StateMachine.Id"
                    - Name: STATE_MACHINE_NAME
                      Value.$: "$$.StateMachine.Name"
                    - Name: STATE_MACHINE_EXECUTION_ID
                      Value.$: "$$.Execution.Id"
                    - Name: STATE_MACHINE_JOB_ID
                      Value.$: "$.Job.Id"
                    - Name: STATE_MACHINE_TASK_INDEX
                      Value.$: States.Format('{}', $.TaskIndex)
                    - Name: STATE_MACHINE_AWS_REGION
                      Value: "${AwsRegion}"
                    - Name: STATE_MACHINE_ARTIFACT_BUCKET_NAME
                      Value.$: "$.Artifact.BucketName"
                    - Name: STATE_MACHINE_ARTIFACT_OBJECT_KEY
                      Value.$: "$.Artifact.ObjectKey"
                    - Name: STATE_MACHINE_TASK_JSON
                      Value.$: States.JsonToString($.Task)
                  Name: "${FtpCopyContainerName}"
            NetworkConfiguration:
              AwsvpcConfiguration:
                AssignPublicIp: ENABLED
                SecurityGroups:
                  - "${FtpCopyEcsTaskSecurityGroupId}"
                Subnets:
                  - "${VpcPublicSubnet1}"
                  - "${VpcPublicSubnet2}"
            PropagateTags: TASK_DEFINITION
            TaskDefinition: "${FtpCopyEcsTaskDefinitionArn}"
          ResultPath: "$.Void"
          OutputPath: "$"
          Next: FTP Copy Task Results Formatter
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 10
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        FTP Copy Task Results Formatter:
          Type: Task
          Comment: Formats the output of an FTP copy task
          Resource: "${FtpCopyTaskOutputLambdaFunctionArn}"
          InputPath: "$"
          Parameters:
            Job:
              Id.$: "$.Job.Id"
            Execution:
              Id.$: "$$.Execution.Id"
            Task.$: "$.Task"
            TaskIteratorIndex.$: "$.TaskIndex"
          ResultPath: "$.TaskResult"
          OutputPath: "$"
          Next: TaskResult Callbacks Map
          Retry:
            - ErrorEquals:
                - FtpOperationError
              MaxAttempts: 0
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        Transcode Task Fargate Execution:
          Type: Task
          Comment: Transcodes multimedia artifact
          Resource: arn:aws:states:::ecs:runTask.sync
          InputPath: "$"
          Parameters:
            Cluster: "${EcsClusterArn}"
            LaunchType: FARGATE
            Overrides:
              ContainerOverrides:
                - Environment:
                    - Name: STATE_MACHINE_ARN
                      Value.$: "$$.StateMachine.Id"
                    - Name: STATE_MACHINE_NAME
                      Value.$: "$$.StateMachine.Name"
                    - Name: STATE_MACHINE_EXECUTION_ID
                      Value.$: "$$.Execution.Id"
                    - Name: STATE_MACHINE_JOB_ID
                      Value.$: "$.Job.Id"
                    - Name: STATE_MACHINE_TASK_INDEX
                      Value.$: States.Format('{}', $.TaskIndex)
                    - Name: STATE_MACHINE_S3_DESTINATION_WRITER_ROLE
                      Value: "${S3DestinationWriterRoleArn}"
                    - Name: STATE_MACHINE_AWS_REGION
                      Value: "${AwsRegion}"
                    - Name: STATE_MACHINE_ARTIFACT_BUCKET_NAME
                      Value.$: "$.Artifact.BucketName"
                    - Name: STATE_MACHINE_ARTIFACT_OBJECT_KEY
                      Value.$: "$.Artifact.ObjectKey"
                    - Name: STATE_MACHINE_DESTINATION_JSON
                      Value.$: States.JsonToString($.Task.Destination)
                    - Name: STATE_MACHINE_DESTINATION_MODE
                      Value.$: "$.Task.Destination.Mode"
                    - Name: STATE_MACHINE_DESTINATION_BUCKET_NAME
                      Value.$: "$.Task.Destination.BucketName"
                    - Name: STATE_MACHINE_DESTINATION_OBJECT_KEY
                      Value.$: "$.Task.Destination.ObjectKey"
                    - Name: STATE_MACHINE_DESTINATION_FORMAT
                      Value.$: "$.Task.Format"
                    - Name: STATE_MACHINE_FFMPEG_GLOBAL_OPTIONS
                      Value.$: "$.Task.FFmpeg.GlobalOptions"
                    - Name: STATE_MACHINE_FFMPEG_INPUT_FILE_OPTIONS
                      Value.$: "$.Task.FFmpeg.InputFileOptions"
                    - Name: STATE_MACHINE_FFMPEG_OUTPUT_FILE_OPTIONS
                      Value.$: "$.Task.FFmpeg.OutputFileOptions"
                    # Guaranteed and required values should be listed and
                    # accessed directly. Optional or conditional values can be
                    # pulled from the JSON, to avoid state machine runtime
                    # errors
                    - Name: STATE_MACHINE_TASK_JSON
                      Value.$: States.JsonToString($.Task)
                    - Name: STATE_MACHINE_ARTIFACT_JSON
                      Value.$: States.JsonToString($.Artifact)
                  Name: "${TranscodeContainerName}"
            NetworkConfiguration:
              AwsvpcConfiguration:
                AssignPublicIp: ENABLED
                Subnets:
                  - "${VpcPublicSubnet1}"
                  - "${VpcPublicSubnet2}"
            PropagateTags: TASK_DEFINITION
            TaskDefinition: "${TranscodeEcsTaskDefinitionArn}"
          ResultPath: "$.Void"
          OutputPath: "$"
          Next: Transcode Task Results Formatter
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 15
              MaxAttempts: 5
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        Transcode Task Results Formatter:
          Type: Task
          Comment: Formats the output of a transcode task
          Resource: "${TranscodeTaskOutputLambdaFunctionArn}"
          InputPath: "$"
          Parameters:
            Job:
              Id.$: "$.Job.Id"
            Execution:
              Id.$: "$$.Execution.Id"
            Task.$: "$.Task"
            TaskIteratorIndex.$: "$.TaskIndex"
          ResultPath: "$.TaskResult"
          OutputPath: "$"
          Next: TaskResult Callbacks Map
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        Image Task Handler:
          Type: Task
          Comment: Manipulates and converts image artifacts
          Resource: "${ImageTransformLambdaFunctionArn}"
          InputPath: "$"
          Parameters:
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"
          ResultPath: "$.TaskResult"
          OutputPath: "$"
          Next: TaskResult Callbacks Map
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        WAV Wrap Task Handler:
          Type: Task
          Comment: WAV wraps audio artifacts
          Resource: "${WavWrapLambdaFunctionArn}"
          InputPath: "$"
          Parameters:
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"
          ResultPath: "$.TaskResult"
          OutputPath: "$"
          Next: TaskResult Callbacks Map
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        Detect Silence Task Handler:
          Type: Task
          Comment: Detects silence in audio files
          Resource: "${SilenceDetectionLambdaFunctionArn}"
          InputPath: "$"
          Parameters:
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"
          ResultPath: "$.TaskResult"
          OutputPath: "$"
          Next: TaskResult Callbacks Map
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        Detect Tone Task Handler:
          Type: Task
          Comment: Detects tones in audio files
          Resource: "${ToneDetectionLambdaFunctionArn}"
          InputPath: "$"
          Parameters:
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"
          ResultPath: "$.TaskResult"
          OutputPath: "$"
          Next: TaskResult Callbacks Map
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        Waveform Task Handler:
          Type: Task
          Comment: Generates waveform data from audio files
          Resource: "${WaveformLambdaFunctionArn}"
          InputPath: "$"
          Parameters:
            Job:
              Id.$: "$.Job.Id"
            Artifact.$: "$.Artifact"
            Task.$: "$.Task"
          ResultPath: "$.TaskResult"
          OutputPath: "$"
          Next: TaskResult Callbacks Map
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        Transcribe Task Job Start:
          Type: Task
          Comment: Transcribes audio from artifacts
          Resource: arn:${AwsPartition}:states:::lambda:invoke.waitForTaskToken
          InputPath: "$"
          Parameters:
            FunctionName: "${TranscriptionJobStartLambdaFunctionArn}"
            Payload:
              Job:
                Id.$: "$.Job.Id"
              Execution:
                Id.$: "$$.Execution.Id"
              Artifact.$: "$.Artifact"
              Task.$: "$.Task"
              TaskIteratorIndex.$: "$.TaskIndex"
              TaskToken.$: "$$.Task.Token"
          ResultPath: "$.TranscriptionJob"
          OutputPath: "$"
          Next: Transcribe Task Job Success Handler
          Retry:
            - ErrorEquals:
                - InvalidTranscribeTaskInputError
                - UnknownDestinationModeError
              MaxAttempts: 0
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map
        Transcribe Task Job Success Handler:
          Type: Task
          Comment: Formats the output of a transcribe task
          Resource: "${TranscriptionJobResultsLambdaFunctionArn}"
          InputPath: "$"
          Parameters:
            Task.$: "$.Task"
            TranscriptionJob.$: "$.TranscriptionJob"
          ResultPath: "$.TaskResult"
          OutputPath: "$"
          Next: TaskResult Callbacks Map
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2
          Catch:
            - ErrorEquals:
                - States.ALL
              ResultPath: "$.Error"
              Next: TaskResult Error Callback Map

        # Task callbacks
        # All tasks operations end up at one of these callback states,
        # depending on whether the task was successful or not. The value for
        # a given task in the TaskResults array coming out of the Tasks Map
        # will be the output whichever of these callback states is used.
        #
        # For failed tasks, the task callback will include information about
        # the error. This information is not included in the job result
        # callback.
        #
        # The output for a successful task is the result of the task.
        # The output for a failed task is the original input task definition.
        TaskResult Callbacks Map:
          Type: Map
          Comment: >-
            Iterates over all callback endpoints to send messages when tasks
            are successful
          InputPath: "$"
          ItemsPath: "$.Callbacks"
          Parameters:
            Callback.$: "$$.Map.Item.Value"
            StateMachine:
              Id.$: "$$.StateMachine.Id"
            Execution:
              Id.$: "$$.Execution.Id"
            TaskIteratorIndex.$: "$.TaskIndex"
            Message: # The JSON of this value will be sent to endpoints
              Task.$: "$.Task"
              TaskResult:
                Job:
                  Id.$: "$.Job.Id"
                Execution:
                  Id.$: "$$.Execution.Id"
                Result.$: "$.TaskResult"
          ResultPath: "$.Void" # The output of the iterator states is discarded
          OutputPath: "$.TaskResult"
          End: true
          MaxConcurrency: 0
          Iterator:
            StartAt: Send TaskResult Callback
            States:
              Send TaskResult Callback:
                Type: Task
                Comment: >-
                  Sends a callback message to a single endpoint when tasks
                  are successful
                InputPath: "$"
                Resource: "${CallbackLambdaFunctionArn}"
                ResultPath: "$"
                OutputPath: "$"
                End: true
                Retry:
                  - ErrorEquals:
                      - States.ALL
                    IntervalSeconds: 5
                    MaxAttempts: 3
                    BackoffRate: 2
        TaskResult Error Callback Map:
          Type: Map
          Comment: >-
            Iterates over all callback endpoints to send messages when tasks
            fail
          InputPath: "$"
          ItemsPath: "$.Callbacks"
          Parameters:
            Callback.$: "$$.Map.Item.Value"
            StateMachine:
              Id.$: "$$.StateMachine.Id"
            Execution:
              Id.$: "$$.Execution.Id"
            TaskIteratorIndex.$: "$.TaskIndex"
            Message: # The JSON of this value will be sent to endpoints
              Task.$: "$.Task"
              TaskResult:
                Job:
                  Id.$: "$.Job.Id"
                Execution:
                  Id.$: "$$.Execution.Id"
                Error.$: "$.Error"
          ResultPath: "$.Void" # The output of the iterator states is discarded
          OutputPath: "$.Task"
          End: true
          MaxConcurrency: 0
          Iterator:
            StartAt: Send TaskResult Error Callback
            States:
              Send TaskResult Error Callback:
                Type: Task
                Comment: >-
                  Sends a callback message to a single endpoint when tasks
                  fail
                InputPath: "$"
                Resource: "${CallbackLambdaFunctionArn}"
                ResultPath: "$"
                OutputPath: "$"
                End: true
                Retry:
                  - ErrorEquals:
                      - States.ALL
                    IntervalSeconds: 5
                    MaxAttempts: 3
                    BackoffRate: 2
    Catch:
      - ErrorEquals:
          - States.ALL
        ResultPath: "$.Error"
        Next: Tasks Map Catcher
  Tasks Map Catcher:
    Comment: >-
      Add a parameter so task map errors can be identified. Is NOT responsible
      for catching individual task execution errors (like if a Copy task fails)
    Type: Pass
    InputPath: "$"
    Result: ITERATOR_ERROR
    ResultPath: "$.State"
    OutputPath: "$"
    Next: Add Empty TaskResults
  # End of task execution

  # States below this will expect $.TaskResults to exist
  Add Empty TaskResults:
    Comment: >-
      Add a TaskResults key with an empty array value, for cases where the task
      iterator does not succeed and it's not otherwise added
    Type: Pass
    InputPath: "$"
    Result: []
    ResultPath: "$.TaskResults"
    OutputPath: "$"
    Next: JobResult Callback Map

  # Start of post-processing
  JobResult Callback Map:
    Type: Map
    Comment: Iterates over all callback endpoints to send job results
    InputPath: "$"
    ItemsPath: "$.Job.Callbacks"
    Parameters:
      Callback.$: "$$.Map.Item.Value"
      StateMachine:
        Id.$: "$$.StateMachine.Id"
      Execution:
        Id.$: "$$.Execution.Id"
      Message: # The JSON of this value will be sent to endpoints
        JobResult:
          Job:
            Id.$: "$.Job.Id"
          Execution:
            Id.$: "$$.Execution.Id"
          State.$: "$.State"
          # Task/Type = Null elements are filtered out.
          # Successful tasks will have a Task property
          TaskResults.$: "$.TaskResults.[?(@.Task && @.Task != 'Null')]"
          # Failed tasks will have a Type property
          FailedTasks.$: "$.TaskResults.[?(@.Type && @.Type != 'Null')]"
    ResultPath: "$.Void" # The output of the iterator states is discarded
    OutputPath: "$"
    Next: Serialized Jobs Map
    MaxConcurrency: 0
    Iterator:
      StartAt: Send JobResult Callback
      States:
        Send JobResult Callback:
          Type: Task
          Comment: >-
            Sends a callback message to a single endpoint in the iterator with
            a job result
          InputPath: "$"
          Resource: "${CallbackLambdaFunctionArn}"
          ResultPath: "$"
          OutputPath: "$"
          End: true
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2

  Serialized Jobs Map:
    Type: Map
    Comment: Iterates over all serialized jobs and sends them to SNS
    InputPath: "$"
    ItemsPath: "$.Job.SerializedJobs"
    Parameters:
      Execution:
        Id.$: "$$.Execution.Id"
      ExecutionTrace.$: "$.Job.ExecutionTrace"
      SerializedJob.$: "$$.Map.Item.Value"
    ResultPath: "$.Void" # The output of the iterator states is discarded
    OutputPath: "$"
    Next: Normalize Output
    MaxConcurrency: 0
    Iterator:
      StartAt: Start Serialized Job Execution
      States:
        Start Serialized Job Execution:
          Type: Task
          Comment: Sends a serialized job to the job execution SNS topic
          InputPath: "$"
          Resource: "${JobSerializerLambdaFunctionArn}"
          ResultPath: "$"
          OutputPath: "$"
          End: true
          Retry:
            - ErrorEquals:
                - States.ALL
              IntervalSeconds: 5
              MaxAttempts: 3
              BackoffRate: 2

  Normalize Output:
    Comment: Normalizes output data
    Type: Task
    InputPath: "$"
    Retry:
      - ErrorEquals:
          - States.ALL
        IntervalSeconds: 1
        MaxAttempts: 2
        BackoffRate: 1
    Parameters:
      StateMachine:
        Id.$: "$$.StateMachine.Id"
      Message:
        JobResult:
          Job:
            Id.$: "$.Job.Id"
          Execution:
            Id.$: "$$.Execution.Id"
          State.$: "$.State"
          TaskResults.$: "$.TaskResults.[?(@.Task && @.Task != 'Null')]"
          FailedTasks.$: "$.TaskResults.[?(@.Type && @.Type != 'Null')]"
    Resource: "${NormalizeOutputLambdaFunctionArn}"
    ResultPath: "$"
    OutputPath: "$"
    End: true
  # End of post-processing