创建顺序执行的Step Function时遇RuntimeError问题求助

问题原因 #

你遇到的RuntimeError: Error: State 'Pass' already has a next state错误，本质是代码重复给初始的Pass状态设置next节点。每次循环调用definition.next()时，definition始终指向最初的Pass状态——第一次循环给Pass加了第一个Lambda任务，第二次循环又尝试给同一个Pass加第二个任务，而Step Function规则要求一个状态只能有一个next指向，因此触发报错。

解决方案 #

需要维护一个变量跟踪当前链的末尾节点，每次将新任务追加到链的最后，而不是重复修改初始状态。修正后的代码如下：

def build_state_machine_definition(validators: list) -> dict:
    """Builds a state machine definition for a given list of tests"""

    # 初始化初始状态和链跟踪变量
    initial_pass = sfn.Pass(self, "Pass")
    current_chain = sfn.Chain.start(initial_pass)

    # 遍历验证器列表，逐个添加Lambda任务
    for i, validator in enumerate(validators):
        lambda_name = validator.split("/")[-2]
        print("lambda_name: " + str(lambda_name))
        lambda_function = _lambda.Function(
            self,
            f"{lambda_name}-lambda",
            runtime=_lambda.Runtime.PYTHON_3_8,
            code=_lambda.Code.from_asset(validator),
            handler="lambda_function.lambda_handler",
            timeout=Duration.seconds(30),
        )

        # 创建Lambda调用任务（区分最后一个任务的命名）
        task_suffix = "Validation" if i == len(validators) - 1 else "Validate"
        lambda_task = tasks.LambdaInvoke(
            self, f"{lambda_name}-{task_suffix}", lambda_function=lambda_function
        )
        
        # 将任务追加到当前链末尾，并更新跟踪变量
        current_chain = current_chain.next(lambda_task)

    return current_chain


state_machine_definition = build_state_machine_definition(lambda_handler_paths)

# 创建Step Function状态机
sfn.StateMachine(
    self,
    "StateMachine",
    definition=state_machine_definition,
    timeout=Duration.minutes(10),
)

关键修改点 #

• 新增current_chain变量，用于跟踪链的最新末尾节点，避免重复修改初始的Pass状态
• 每次追加任务后，将current_chain更新为current_chain.next(lambda_task)的返回值，确保后续任务始终接在之前任务的后面
• 优化了任务命名逻辑，用统一的方式生成任务ID，代码更简洁

内容的提问来源于stack exchange，提问作者aphexlog