Lambda の再帰ループ検出が EventBridge の新しい Custom Event Bus に対応したので試してみた

AWSTemplateFormatVersion: "2010-09-09"
Description: Lambda recursive loop detection test with EventBridge Custom Event Bus

Parameters:
  Boto3LayerArn:
    Type: String

Resources:
  Bus:
    Type: AWS::EventsV2::EventBus
    Properties:
      Name: recursion-test-v2

  FailureQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: recursion-test-v2-failure

  FunctionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: put-events
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action: events:PutEvents
                Resource: !Ref Bus
              - Effect: Allow
                Action: sqs:SendMessage
                Resource: !GetAtt FailureQueue.Arn

  # サブスクライバーが Lambda を呼び出すためのロール
  DeliveryRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: events.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: deliver
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action: lambda:InvokeFunction
                Resource:
                  - !GetAtt AsyncFunction.Arn
                  - !GetAtt SyncFunction.Arn
              - Effect: Allow
                Action: sqs:SendMessage
                Resource: !GetAtt FailureQueue.Arn

  AsyncFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: recursion-test-v2-async
      Runtime: python3.13
      Handler: index.handler
      Role: !GetAtt FunctionRole.Arn
      Layers:
        - !Ref Boto3LayerArn
      Environment:
        Variables:
          BUS_ARN: !Ref Bus
          DETAIL_TYPE: async
          MAX_HOP: "30"
      Code:
        ZipFile: |
          import json, os
          import boto3

          events = boto3.client("eventbridgev2")

          def handler(event, context):
              hop = event["detail"]["hop"]
              print(json.dumps({
                  "hop": hop,
                  "trace": os.environ.get("_X_AMZN_TRACE_ID"),
                  "boto3": boto3.__version__,
              }))
              # 検出されなかった場合の安全策
              if hop >= int(os.environ["MAX_HOP"]):
                  return
              events.put_events(
                  EventBusArn=os.environ["BUS_ARN"],
                  Entries=[{
                      "Source": "blog.recursion",
                      "DetailType": os.environ["DETAIL_TYPE"],
                      "Detail": json.dumps({"hop": hop + 1}),
                  }],
              )

  SyncFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: recursion-test-v2-sync
      Runtime: python3.13
      Handler: index.handler
      Role: !GetAtt FunctionRole.Arn
      Layers:
        - !Ref Boto3LayerArn
      Environment:
        Variables:
          BUS_ARN: !Ref Bus
          DETAIL_TYPE: sync
          MAX_HOP: "30"
      Code:
        ZipFile: |
          import json, os
          import boto3

          events = boto3.client("eventbridgev2")

          def handler(event, context):
              hop = event["detail"]["hop"]
              print(json.dumps({
                  "hop": hop,
                  "trace": os.environ.get("_X_AMZN_TRACE_ID"),
                  "boto3": boto3.__version__,
              }))
              # 検出されなかった場合の安全策
              if hop >= int(os.environ["MAX_HOP"]):
                  return
              events.put_events(
                  EventBusArn=os.environ["BUS_ARN"],
                  Entries=[{
                      "Source": "blog.recursion",
                      "DetailType": os.environ["DETAIL_TYPE"],
                      "Detail": json.dumps({"hop": hop + 1}),
                  }],
              )

  AsyncInvokeConfig:
    Type: AWS::Lambda::EventInvokeConfig
    Properties:
      FunctionName: !Ref AsyncFunction
      Qualifier: $LATEST
      MaximumRetryAttempts: 0
      DestinationConfig:
        OnFailure:
          Destination: !GetAtt FailureQueue.Arn

  AsyncSubscriber:
    Type: AWS::EventsV2::Subscriber
    Properties:
      Name: recursion-test-async
      EventBusArn: !Ref Bus
      FilterConfiguration:
        Filters:
          - Scope: DATA
            Pattern: '{"detail-type":["async"]}'
      BatchConfiguration:
        MaxBatchSize: 1
      InvokeConfiguration:
        TargetArn: !GetAtt AsyncFunction.Arn
        RoleArn: !GetAtt DeliveryRole.Arn
        LambdaParameters:
          InvocationType: EVENT
      OnFailureConfiguration:
        Arn: !GetAtt FailureQueue.Arn

  SyncSubscriber:
    Type: AWS::EventsV2::Subscriber
    Properties:
      Name: recursion-test-sync
      EventBusArn: !Ref Bus
      FilterConfiguration:
        Filters:
          - Scope: DATA
            Pattern: '{"detail-type":["sync"]}'
      BatchConfiguration:
        MaxBatchSize: 1
      InvokeConfiguration:
        TargetArn: !GetAtt SyncFunction.Arn
        RoleArn: !GetAtt DeliveryRole.Arn
        LambdaParameters:
          InvocationType: REQUEST_RESPONSE
      RetryPolicy:
        MaxRetryAttempts: 2
        MaxEventAgeInSeconds: 600
      OnFailureConfiguration:
        Arn: !GetAtt FailureQueue.Arn

Outputs:
  BusArn:
    Value: !Ref Bus