Step Functions — ワークフローを設計する

注文処理の移行が始まった頃、ダイチは新しい問題に直面していた。

「注文確定処理、Lambda1個では無理だ」

注文が確定するまでのフローは複雑だった。

在庫確認（DynamoDB）
決済処理（外部決済API）
在庫引き当て（DynamoDB）
注文レコード作成（Aurora）
確認メール送信（SES）
配送システム連携（外部API）

これを1つのLambdaに詰め込むと、タイムアウトリスク、エラー時のロールバック処理の複雑化、テストの困難さという問題が生じる。

AWS Step Functions がこの問題を解決する。複数のLambda関数を「ステートマシン」として定義し、フロー制御・エラーハンドリング・リトライを宣言的に記述できる。

Step Functions の基本概念

Loading diagram...

Amazon States Language でワークフローを定義

Step Functionsのワークフローは Amazon States Language (ASL) というJSONで定義する。

# template.yaml
OrderProcessingStateMachine:
  Type: AWS::Serverless::StateMachine
  Properties:
    DefinitionUri: statemachines/order_processing.asl.json
    Policies:
      - LambdaInvokePolicy:
          FunctionName: !Ref CheckStockFunction
      - LambdaInvokePolicy:
          FunctionName: !Ref ProcessPaymentFunction
      - LambdaInvokePolicy:
          FunctionName: !Ref ReserveStockFunction
      - LambdaInvokePolicy:
          FunctionName: !Ref CreateOrderFunction
      - LambdaInvokePolicy:
          FunctionName: !Ref SendConfirmationEmailFunction
      - LambdaInvokePolicy:
          FunctionName: !Ref RefundPaymentFunction
      - CloudWatchLogsFullAccess
    Logging:
      Level: ALL
      IncludeExecutionData: true
      Destinations:
        - CloudWatchLogsLogGroup:
            LogGroupArn: !GetAtt StateMachineLogGroup.Arn

// statemachines/order_processing.asl.json
{
  "Comment": "注文処理ワークフロー",
  "StartAt": "CheckStock",
  "States": {
    "CheckStock": {
      "Type": "Task",
      "Resource": "${CheckStockFunctionArn}",
      "ResultPath": "$.stock_check",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.ServiceException", "Lambda.AWSLambdaException"],
          "IntervalSeconds": 2,
          "MaxAttempts": 3,
          "BackoffRate": 2
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "OrderFailed",
          "ResultPath": "$.error"
        }
      ],
      "Next": "IsStockAvailable"
    },
 
    "IsStockAvailable": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.stock_check.available",
          "BooleanEquals": true,
          "Next": "ProcessPayment"
        }
      ],
      "Default": "StockUnavailable"
    },
 
    "StockUnavailable": {
      "Type": "Fail",
      "Error": "StockUnavailableError",
      "Cause": "Requested item is out of stock"
    },
 
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "${ProcessPaymentFunctionArn}",
      "ResultPath": "$.payment",
      "Retry": [
        {
          "ErrorEquals": ["PaymentTemporaryError"],
          "IntervalSeconds": 5,
          "MaxAttempts": 2,
          "BackoffRate": 1.5
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["PaymentDeclinedError"],
          "Next": "PaymentDeclined",
          "ResultPath": "$.error"
        },
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "OrderFailed",
          "ResultPath": "$.error"
        }
      ],
      "Next": "ReserveStock"
    },
 
    "PaymentDeclined": {
      "Type": "Fail",
      "Error": "PaymentDeclinedError",
      "Cause": "Payment was declined by the payment provider"
    },
 
    "ReserveStock": {
      "Type": "Task",
      "Resource": "${ReserveStockFunctionArn}",
      "ResultPath": "$.reservation",
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "RefundPayment",
          "ResultPath": "$.error"
        }
      ],
      "Next": "CreateOrder"
    },
 
    "RefundPayment": {
      "Type": "Task",
      "Resource": "${RefundPaymentFunctionArn}",
      "Comment": "在庫引き当て失敗時の補償トランザクション",
      "Next": "OrderFailed"
    },
 
    "CreateOrder": {
      "Type": "Task",
      "Resource": "${CreateOrderFunctionArn}",
      "ResultPath": "$.order",
      "Next": "SendConfirmationEmail"
    },
 
    "SendConfirmationEmail": {
      "Type": "Task",
      "Resource": "${SendConfirmationEmailFunctionArn}",
      "ResultPath": "$.email",
      "Next": "OrderSucceeded"
    },
 
    "OrderSucceeded": {
      "Type": "Succeed"
    },
 
    "OrderFailed": {
      "Type": "Fail",
      "Error": "OrderProcessingError",
      "Cause": "Order processing failed"
    }
  }
}

各 Lambda 関数の実装

在庫確認 Lambda

# src/check_stock/app.rb
# frozen_string_literal: true
 
require 'aws-sdk-dynamodb'
require 'json'
 
$dynamodb = Aws::DynamoDB::Client.new
 
def lambda_handler(event:, context:)
  product_id = event['product_id']
  quantity = event['quantity']
 
  result = $dynamodb.get_item(
    table_name: ENV['PRODUCTS_TABLE_NAME'],
    key: {
      'PK' => "PRODUCT##{product_id}",
      'SK' => 'PRODUCT'
    },
    projection_expression: 'stock'
  )
 
  stock = result.item&.dig('stock')&.to_i || 0
 
  {
    product_id:,
    requested_quantity: quantity,
    available_stock: stock,
    available: stock >= quantity
  }
end

決済処理 Lambda

# src/process_payment/app.rb
# frozen_string_literal: true
 
require 'json'
require 'net/http'
require 'uri'
 
class PaymentDeclinedError < StandardError; end
class PaymentTemporaryError < StandardError; end
 
def lambda_handler(event:, context:)
  payment_method_id = event['payment_method_id']
  amount = event['amount']
  currency = event.fetch('currency', 'JPY')
  order_ref = event['order_id']
 
  # 外部決済APIの呼び出し（Stripeなど）
  response = call_payment_api(
    payment_method_id:,
    amount:,
    currency:,
    metadata: { order_ref: }
  )
 
  case response[:status]
  when 'succeeded'
    {
      payment_id: response[:id],
      status: 'succeeded',
      amount:,
      charged_at: Time.now.iso8601
    }
  when 'requires_action'
    raise PaymentDeclinedError, "Payment requires additional action"
  when 'failed'
    raise PaymentDeclinedError, "Payment declined: #{response[:failure_message]}"
  else
    raise PaymentTemporaryError, "Unexpected payment status: #{response[:status]}"
  end
end
 
private
 
def call_payment_api(payment_method_id:, amount:, currency:, metadata:)
  # 実際はStripe等のSDKを使う
  # ここではモック実装
  {
    id: "pay_#{SecureRandom.hex(12)}",
    status: 'succeeded',
    amount:,
    currency:
  }
end

補償トランザクション — 払い戻し処理

# src/refund_payment/app.rb
# frozen_string_literal: true
 
require 'json'
require 'logger'
 
$logger = Logger.new($stdout)
 
def lambda_handler(event:, context:)
  payment_id = event.dig('payment', 'payment_id')
  error_info = event['error']
 
  $logger.warn("Refunding payment #{payment_id} due to: #{error_info&.dig('Cause')}")
 
  # 決済のキャンセル/払い戻し
  refund_result = process_refund(payment_id)
 
  $logger.info("Refund completed: #{refund_result[:refund_id]}")
 
  {
    refunded: true,
    payment_id:,
    refund_id: refund_result[:refund_id],
    refunded_at: Time.now.iso8601
  }
end
 
private
 
def process_refund(payment_id)
  # Stripe refund API 等の呼び出し
  { refund_id: "re_#{SecureRandom.hex(12)}" }
end

並列実行 (Parallel State)

確認メールと配送システム連携は同時に行えるので、Parallel Stateを使う。

"NotifyAllSystems": {
  "Type": "Parallel",
  "Branches": [
    {
      "StartAt": "SendEmail",
      "States": {
        "SendEmail": {
          "Type": "Task",
          "Resource": "${SendEmailFunctionArn}",
          "End": true
        }
      }
    },
    {
      "StartAt": "NotifyShipping",
      "States": {
        "NotifyShipping": {
          "Type": "Task",
          "Resource": "${NotifyShippingFunctionArn}",
          "End": true
        }
      }
    }
  ],
  "Next": "OrderSucceeded"
}

INFO

Parallel Stateのすべてのブランチが完了して初めて次のステートに進む。一つのブランチが失敗するとParallelステート全体が失敗となり、Catchで補捉できる。

ワークフローの実行とモニタリング

Step Functions の実行を開始する

# 注文APIのLambda（ワークフロー開始側）
require 'aws-sdk-sfn'
 
$sfn = Aws::SFN::Client.new
 
def lambda_handler(event:, context:)
  body = JSON.parse(event['body'])
 
  input = {
    order_id: SecureRandom.uuid,
    user_id: body['user_id'],
    product_id: body['product_id'],
    quantity: body['quantity'],
    amount: body['amount'],
    payment_method_id: body['payment_method_id']
  }
 
  # ステートマシンの実行を開始
  execution = $sfn.start_execution(
    state_machine_arn: ENV['ORDER_STATE_MACHINE_ARN'],
    name: "order-#{input[:order_id]}",  # 実行の一意な名前
    input: JSON.generate(input)
  )
 
  {
    statusCode: 202,  # Accepted
    body: JSON.generate({
      order_id: input[:order_id],
      execution_arn: execution.execution_arn,
      message: 'Order is being processed'
    })
  }
end

# CLIで実行状態を確認
aws stepfunctions describe-execution \
  --execution-arn "arn:aws:states:ap-northeast-1:123456789:execution:OrderProcessing:order-xxx"
 
# 実行履歴（どのステートで何が起きたか）
aws stepfunctions get-execution-history \
  --execution-arn "arn:aws:states:ap-northeast-1:123456789:execution:OrderProcessing:order-xxx"

Express vs Standard ワークフロー

Step Functionsには2種類のワークフローがある。

	Standard	Express
最大実行時間	1年	5分
実行保証	Exactly-once	At-least-once
料金	ステート遷移数	実行数 + 期間
ユースケース	長期プロセス、注文管理	高頻度・短期間処理
監査ログ	完全な履歴保存	CloudWatch Logsのみ

ECサイトの注文処理には Standard Workflow が適している。「ちょうど1回」の実行保証が必要だからだ。

WARNING

Express Workflowでは同じ注文が複数回処理される可能性がある。決済処理を含むフローでは必ずStandard Workflowを使うこと。ただしExpressは秒間数千件の高スループットに対応しており、ログ処理などに向いている。

ダイチはStep Functionsの実装を終えてコードを見渡した。各Lambda関数は100行以下のシンプルなコードになった。エラーハンドリングはASLで宣言的に記述されており、コード中に複雑なtry/catchのネストがない。

「これ、テストしやすい」と同僚のサクラが言った。「各Lambda関数を独立してテストできて、フロー全体はStep Functionsが管理してくれるから」

まさにその通りだった。関心の分離が自然に実現されていた。