Working with state machines¶

A Step Functions machine that grew for years is usually one 3,000-line JSON string with the same eight states pasted sixty times. This page is a walk through porting one to Python with cfn_typed.asl: how to look at it first, how to port it without changing it, how to prove that, and what to fix afterwards. It is written from a real port of a 143-state machine, but the machine used here is made up: 23 states, three operations, with the same kinds of quirks. Everything is in examples/state_machine/ and is run by the test suite, so the code on this page works.

The machine¶

original.json is a "ledger runner": a Route choice on $.action sends the run to one of three operations.

• ImportInvoices: start a task, poll its status until $.finished, finish. Its failure handler has no retry and no catch.
• ReconcileLedger: an optional timed start, a task, a Map over the accounts, a polling loop on $.continue, finish.
• ArchiveReports: a task, a Parallel of two branches, a notification.

Each operation ends in a shared Common_Succeeded or Common_Failed, and has its own Perform_...Failed handler.

Step 0: get the definition as JSON¶

From a template, cfn-typed import --asl template.json writes the definition as a Python object (see import). From a deployed machine, aws stepfunctions describe-state-machine --state-machine-arn ARN --query definition --output text. Either way, keep the JSON file: it is the original you will compare against until the end.

Step 1: survey before you write anything¶

Count what repeats. survey.py prints the numbers that decide which helpers are worth writing:

def survey(definition: dict[str, Any]) -> list[str]:
    states: dict[str, Any] = definition["States"]
    lines = [f"{len(states)} states: {dict(Counter(s['Type'] for s in states.values()))}"]

    bodies = Counter(json.dumps({k: v for k, v in s.items() if k != "Next"}, sort_keys=True) for s in states.values() if s["Type"] == "Task")
    lines.append(f"{sum(bodies.values())} tasks, {len(bodies)} distinct bodies (ignoring Next): the fewer, the more a helper will save")

    catches = Counter(c["Next"] for s in states.values() for c in s.get("Catch", []))
    lines.append("where errors go: " + ", ".join(f"{target} x{n}" for target, n in catches.most_common()))

    loops = [name for name, s in states.items() if s["Type"] == "Wait" and states.get(s.get("Next"), {}).get("Type") == "Task"]
    lines.append(f"polling loops (a Wait that leads to a Task): {', '.join(loops) or 'none'}")

    no_handler = [n for n, s in states.items() if s["Type"] == "Task" and "Catch" not in s]
    lines.append(f"tasks with no Catch (handlers, or forgotten?): {', '.join(no_handler) or 'none'}")
    return lines

On the made-up machine:

23 states: {'Choice': 4, 'Task': 12, 'Wait': 3, 'Map': 1, 'Parallel': 1, 'Succeed': 1, 'Fail': 1}
12 tasks, 7 distinct bodies (ignoring Next): the fewer, the more a helper will save
where errors go: Perform_ImportFailed x4, Perform_ReconcileFailed x3, Common_Failed x3, Perform_ArchiveFailed x3
polling loops (a Wait that leads to a Task): Wait_ImportStatus, Wait_StartReconcile, Wait_ReconcileStaus
tasks with no Catch (handlers, or forgotten?): Perform_ImportFailed

Read it for three things. Few distinct task bodies means a task() helper will pay off; in the real machine, 71 Tasks had 17 distinct bodies. Where errors go shows the operation boundaries: each family of states that catches into the same handler is one operation, and a state that catches somewhere else is either a deliberate choice or a mistake. Polling loops are your first pattern. (Wait_StartReconcile is a timed start, not a loop; the survey is a rough tool.)

Step 2: port it without changing it¶

The rule for the first pass: keep every state name, and reproduce every quirk. The goal is a Python machine that assert_same_machine says is the original. Improving it is Step 3, and doing both at once is how a port ends up unverifiable.

Start with the things that are the same everywhere:

WORKER = "arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:acme-ledger-${Stage}-worker"
TRANSIENT = ["Lambda.ServiceException", "Lambda.AWSLambdaException", "Lambda.SdkClientException", "Lambda.TooManyRequestsException"]
RETRY_TRANSIENT, RETRY_ALL = [retry(*TRANSIENT)], [retry("States.ALL")]

Then the builders. Exit("succeeded"), Exit("error") and Exit("failed") are the three ways out of an operation: the shared success state, the shared failure state, and the operation's own failure handler. A task that should catch into its operation's handler says catch_to=FAILED; the operation wires failed to the right handler once, instead of repeating its name on every task.

SUCCEEDED, FAILED, ERROR = Exit("succeeded"), Exit("failed"), Exit("error")
# succeeded: the shared Succeed state. error: the shared Fail state. failed: this operation's own failure handler.


def task(next: str | None = None, *, retry: Any = RETRY_TRANSIENT, catch_to: str | None = FAILED, end: bool = False) -> dict[str, Any]:
    """A call to the worker Lambda. The retry set and where an error goes are the two things that differ."""
    state: dict[str, Any] = {"Type": "Task", "Resource": WORKER}
    if next:
        state["Next"] = next
    if end:
        state["End"] = True
    if retry:
        state["Retry"] = retry
    if catch_to:
        state["Catch"] = [catch(catch_to)]
    return state


def flag(variable: str, value: bool, next: str) -> ChoiceRule:
    return rule(variable, "BooleanEquals", value, next)


def status_loop(name: str, *, seconds: int | None = None, seconds_path: str | None = None, done: str, default: str = ERROR,
                catch_to: str = FAILED, wait_name: str | None = None, flag_name: str = "$.finished") -> dict[str, Any]:
    """Wait, ask the worker how it is going, and either go round again, move on to `done`, or give up at `default`."""
    wait_, check, verify = wait_name or f"Wait_{name}", f"Check_{name}", f"Verify_{name}"
    return {
        wait_: wait(check, seconds=seconds, seconds_path=seconds_path),
        check: task(verify, retry=RETRY_ALL, catch_to=catch_to),
        verify: choice(flag(flag_name, True, done), flag(flag_name, False, wait_), default=default),
    }

Each operation is a function that returns a Fragment. Its states are written with the builders, and the name of each state is the original's:

def import_invoices() -> Fragment:
    return Fragment("Start_ImportInvoices", {
        "Start_ImportInvoices": task("Wait_ImportStatus"),
        **status_loop("ImportStatus", seconds=30, done="Finalize_ImportInvoices"),
        "Finalize_ImportInvoices": task(SUCCEEDED),
        "Perform_ImportFailed": task(ERROR, retry=None, catch_to=None),  # the only handler with no retry and no catch
    }).wire("failed", "Perform_ImportFailed")


def reconcile_ledger() -> Fragment:
    return Fragment("Check_ReconcileTimings", {
        "Check_ReconcileTimings": choice(rule("$.start_time", "IsTimestamp", True, "Wait_StartReconcile"), default="Start_Reconcile"),
        "Wait_StartReconcile": wait("Start_Reconcile", timestamp_path="$.start_time"),
        "Start_Reconcile": task("Fetch_Accounts"),
        # The original catches this one into the *import* handler. Kept as it was, under its own exit name.
        "Fetch_Accounts": task("Map_Accounts", catch_to=Exit("import_failed")),
        "Map_Accounts": {
            "Type": "Map", "ItemsPath": "$.input.accounts", "MaxConcurrency": 5, "ResultPath": "$.accountResults",
            "Parameters": {"Account.$": "$$.Map.Item.Value", "Run.$": "$.run"},
            "Iterator": {"StartAt": "Reconcile_Account", "States": {"Reconcile_Account": task(end=True, retry=None, catch_to=None)}},
            "Next": "Wait_ReconcileStaus", "Retry": RETRY_TRANSIENT, "Catch": [catch(FAILED)],
        },
        # `Staus` is the original's spelling, and the status check catches into the shared failure state, not the handler.
        **status_loop("ReconcileStatus", seconds_path="$.wait_time", done="Finalize_Reconcile", default=FAILED,
                      catch_to=ERROR, wait_name="Wait_ReconcileStaus", flag_name="$.continue"),
        "Finalize_Reconcile": task(SUCCEEDED),
        "Perform_ReconcileFailed": task(ERROR, catch_to=ERROR),
    }).wire("failed", "Perform_ReconcileFailed").wire("import_failed", "Perform_ImportFailed")


def archive_reports() -> Fragment:
    return Fragment("Start_ArchiveReports", {
        "Start_ArchiveReports": task("Fan_Archive"),
        "Fan_Archive": {
            "Type": "Parallel", "Next": "Notify_Archive", "Retry": RETRY_TRANSIENT, "Catch": [catch(FAILED)],
            "Branches": [
                {"StartAt": "Render_Pdf", "States": {"Render_Pdf": task(end=True, retry=None, catch_to=None)}},
                {"StartAt": "Render_Csv", "States": {
                    "Render_Csv": task("Compress_Csv", retry=None, catch_to=None),
                    "Compress_Csv": task(end=True, retry=None, catch_to=None)}},
            ],
        },
        "Notify_Archive": task(SUCCEEDED),
        "Perform_ArchiveFailed": task(ERROR, catch_to=ERROR),
    }).wire("failed", "Perform_ArchiveFailed")

Look at what the port made you write down. Fetch_Accounts catches into the import handler, which is almost certainly a copy-paste slip, and the port gave it its own exit name (import_failed) so that the oddity is visible in the code. Wait_ReconcileStaus has a typo. Check_ReconcileStatus catches into Common_Failed where its neighbours catch into the handler. None of this was visible in the JSON.

Finally the router, and the assembly:

ACTIONS = {"ImportInvoices": import_invoices, "ReconcileLedger": reconcile_ledger, "ArchiveReports": archive_reports}


def build() -> Definition:
    machine = Fragment("Route", {"Route": choice(*[rule("$.action", "StringEquals", a, Exit(a)) for a in ACTIONS], default=ERROR)})
    for action, operation in ACTIONS.items():
        machine = machine.wire(action, operation())
    return (
        machine.include(step("Common_Succeeded", {"Type": "Succeed"})).include(step("Common_Failed", {"Type": "Fail"}))
        .wire("succeeded", "Common_Succeeded").wire("error", "Common_Failed")
        .build(Comment="Ledger runner (a made-up machine for the docs)")
    )

The test is the gate. It fails with every difference listed, which is how you find your own mistakes (the first version of this example had $.finished where the reconcile loop uses $.continue):

from cfn_typed.testing import assert_same_machine

def test_port_is_the_original() -> None:
    assert_same_machine(build(), json.loads(Path("original.json").read_text()))

Hints for porters¶

Method

• Port verbatim first, then change. Two commits, or at least two steps: one that must produce no difference, and one whose every difference you can name.
• Port one operation at a time. find_machine_differences compares whole machines, but you can compare a subset: after wiring the shared exits, fragment.states == {name: original["States"][name] for name in fragment.states}. The operation is done when that holds.
• build() is a second test. It refuses open exits (unwired exits: failed) and an unsound graph, and it names the state. Run it early and let it guide you to the exit you forgot to wire.
• Group by where errors go. The Catch target is the best clue to an operation's boundary, better than the name prefix. Names are the least reliable thing in an old machine.
• Write the survey down in a comment. "17 distinct bodies, 3 retry sets, 10 handlers" tells the next person what the helpers are for.

What to expect in an old machine

• Inconsistencies are the norm. Expect one task per operation that catches somewhere else, handlers that differ from each other, names that nearly match (Upscale_ next to UpScale_, Prestep next to PreStep), and a loop that checks $.finished where all its siblings check $.continue. Keep them in the port, name them in a comment, decide in Step 3.
• A Choice Default is not the Catch. The same state family often sends a failed check to the shared failure state and an error to the operation's handler. They are separate exits; do not merge them to tidy up.
• Order matters in lists. Choices are evaluated first-match, Retry and Catch too. assert_same_machine compares lists in order on purpose. Key order does not matter.
• Integers stay integers. BackoffRate: 2 and 2.0 compare equal in Python and render differently. retry() defaults to the int, and a diff of the rendered template stays quiet.
• A loop is not always a loop. Wait then Task is a poll only if a Choice sends control back. A timed start (TimestampPath) looks similar and is not. Check the choice before generalising.
• Some states are one-offs. Not everything is a pattern. A state that appears once is written out as a literal dict in its operation, and that is fine; resist a helper with a parameter for each exception. If a helper needs more than three or four flags to cover the cases, make two helpers or leave the odd cases literal.

Names

• State names are not free to change. They appear in execution history, CloudWatch Logs and metrics, X-Ray traces, alarms keyed on a state, and often in the worker Lambda ($$.State.Name, or branching on a name it was given). Before any rename, search every consumer you have: dashboards, alarms, the Lambda source, runbooks.
• A rename is a change, not a refactor. Prove it separately (Step 3), deploy it separately, and leave a note about executions that are in flight: a running execution keeps the definition it started with.
• Fix typos last. A typo you kept costs nothing. A fixed one has to be right everywhere at once.

Getting it into a template

• Placeholders. Strings with ${AWS::Region}, ${AWS::AccountId} or ${Stage} must become Fn::Sub. cfn_typed.asl.substituted(definition) wraps each such string. Definition is a Json-typed property, like an IAM PolicyDocument, and cfn-lint accepts Fn::Sub nested anywhere inside it. DefinitionSubstitutions is the other route, but it is for named placeholders (${HelloFunction}) whose values you supply, and cfn-lint rejects ${AWS::Region} left in a definition that is not wrapped in Fn::Sub, so use Fn::Sub for pseudo parameters. This was checked against the docs and cfn-lint, not by a deploy: deploy once to a throwaway stack and compare the deployed definition (describe-state-machine) with the original.
• Definition, not DefinitionString. CloudFormation sees a different property, so the first deploy shows a change to it even when the machine is identical. The deployed definition should be the same; compare it with the original after the first deploy.
• Size. The definition limit is 1 MB, but a machine of hundreds of states more often hits other limits first: 25,000 events of execution history, 256 KB of state payload. If an operation is already a self-contained fragment with its own exits, it can be moved into a child machine later without touching the others.

Step 3: fix what the port showed¶

Now change things, and make every change a line you can review. The first step is to prove that the tidy-up is the only change: take the original, apply the same renames to it, and compare.

RENAMES = {"Wait_ReconcileStaus": "Wait_ReconcileStatus"}


def own_handler(state: dict[str, Any]) -> dict[str, Any]:
    """Send these tasks' errors to the reconcile handler like the rest of the operation."""
    state["Catch"][0]["Next"] = "Perform_ReconcileFailed"
    return state


def build() -> Definition:
    machine = ported()
    return (
        Fragment(machine["StartAt"], machine["States"])
        .renamed(RENAMES)
        .map_states(own_handler, where=lambda s: s.get("Next") in ("Map_Accounts", "Verify_ReconcileStatus"))
        .build(Comment=machine.get("Comment", ""))
    )

renamed takes a mapping (or a function) and rewrites every state name, StartAt and the transitions that follow it, including inside Map and Parallel. map_states applies one change to the states you choose. Both return a new fragment, so the port itself stays as it was.

The test states what changes, in full:

def test_tidy_changes_what_it_says() -> None:
    original = json.loads(Path("original.json").read_text())
    expected = Fragment(original["StartAt"], original["States"]).renamed(RENAMES).build()
    assert find_machine_differences(build(), expected) == [
        'States.Check_ReconcileStatus.Catch[0].Next: "Perform_ReconcileFailed", expected "Common_Failed"',
        'States.Fetch_Accounts.Catch[0].Next: "Perform_ReconcileFailed", expected "Perform_ImportFailed"',
    ]

That list is the change description for the review: two error routes are fixed and one name is spelled correctly. Anything else in it is a bug in the tidy-up. When the original is also renamed, a rename shows up in neither list, so the list contains only behaviour changes.

Step 4: keep it honest afterwards¶

• Keep original.json and the verbatim test while the old machine is still deployed anywhere. When it is gone, the tidy version becomes the golden file: commit the rendered machine and let CI fail when it changes (assert_matches_file).
• Check the template too. assert_state_machines_valid(template) runs the graph check on what you actually deploy, and find_all_problems reports it as asl.
• Draw it for review. to_mermaid(definition) writes a diagram you can paste into a pull request. A change to the graph shows up as a change to the picture.

Where the line is¶

If you find yourself writing the same helper in two projects, that is the sign it may belong in the library. Open an issue with the helper and the shapes it covers.