aws2-9 Exercise 3: Retry and Catch Measured, Definition and Template Mutations, and a Cross-Resource Checker ======================================================================== Run with Windows 11, Python 3.14 (py launcher) and a venv with moto 5.2.4 (server mode on port 5000), boto3 1.43.112, cfn-lint 1.57.2, PyYAML; Step Functions Local from the Docker image amazon/aws-stepfunctions-local on port 8083 with a mock config (Docker Desktop 29.8.2); the SAM CLI 1.168.0 from the chapter 8 environment for the Lambda run. Scratch folder C:/lwaws/c9, nothing installed globally. Dummy credentials (test/test) are used with the emulators. VERIFICATION LEVEL: 1 (no AWS account, nothing deployed); statements about the real services come from the AWS documentation read on 2026-10-13 and are marked as such in the chapter. Outputs below are saved from the real runs of 2026-10-13. --- MockConfigFile.json --- { "StateMachines": { "RetryDemo": { "TestCases": { "SucceedAfterTwo": {"Charge": "ThrowTwiceThenOk"}, "AlwaysFails": {"Charge": "AlwaysTransient"}, "OtherError": {"Charge": "Declined"} } }, "CappedRetry": {"TestCases": {"AlwaysFails": {"Charge": "AlwaysTransient"}}}, "UncappedRetry": {"TestCases": {"AlwaysFails": {"Charge": "AlwaysTransient"}}}, "JitterRetry": {"TestCases": {"AlwaysFails": {"Charge": "AlwaysTransient"}}}, "TimeoutNoRetry": { "TestCases": { "TimeoutThenOk": {"Charge": "TimeoutThenOk"}, "OtherThenOk": {"Charge": "OtherThenOk"} } } }, "MockedResponses": { "ThrowTwiceThenOk": { "0-1": {"Throw": {"Error": "Payment.Transient", "Cause": "temporary failure"}}, "2": {"Return": {"StatusCode": 200, "Payload": {"charged": true}}} }, "AlwaysTransient": { "0-3": {"Throw": {"Error": "Payment.Transient", "Cause": "temporary failure"}} }, "Declined": { "0": {"Throw": {"Error": "Payment.Declined", "Cause": "card declined"}} }, "TimeoutThenOk": { "0": {"Throw": {"Error": "States.Timeout", "Cause": "timed out"}}, "1": {"Return": {"StatusCode": 200, "Payload": {"charged": true}}} }, "OtherThenOk": { "0": {"Throw": {"Error": "Payment.Other", "Cause": "something else"}}, "1": {"Return": {"StatusCode": 200, "Payload": {"charged": true}}} } } } --- how Step Functions Local was started (Docker Desktop on Windows; MSYS_NO_PATHCONV=1 in Git Bash) --- docker run -d --name sfnlocal -p 8083:8083 --mount type=bind,readonly,source=C:/lwaws/c9/MockConfigFile.json,destination=/home/StepFunctionsLocal/MockConfigFile.json -e SFN_MOCK_CONFIG=/home/StepFunctionsLocal/MockConfigFile.json amazon/aws-stepfunctions-local (afterwards: docker rm -f sfnlocal) --- Lambda partial-batch run (from c9/lambda_batch with the SAM CLI of chapter 8) --- sam local generate-event sqs receive-message > ev_one.json # one record, body 'Hello from SQS!' (five records made from it: ids id1..id5, bodies with order A-1..A-5, poison true for A-2 and A-4) sam build ; sam local invoke BatchFn -e ev_five.json --- sfn_retry.py --- """Retry/Catch semantics from the Step Functions error-handling page (read 2026-10-13), run on Step Functions Local (Docker image amazon/aws-stepfunctions-local with a mock config; AWS marks it 'unsupported' and without feature parity). The Task result or error is scripted by the mock config; the retry delays are measured from the execution history. State machine names must match the names in MockConfigFile.json.""" import json, sys, time from botocore.exceptions import ClientError from common import client sfn = client("stepfunctions", "http://127.0.0.1:8083") ROLE = "arn:aws:iam::123456789012:role/DummyRole" FN = "arn:aws:lambda:us-east-1:123456789012:function:charge" CREATED = {} def machine(retry, catch=True): task = {"Type": "Task", "Resource": "arn:aws:states:::lambda:invoke", "Parameters": {"FunctionName": FN, "Payload.$": "$"}, "Retry": retry, "Next": "Done"} if catch: task["Catch"] = [{"ErrorEquals": ["States.ALL"], "ResultPath": "$.error", "Next": "Refund"}] return {"StartAt": "Charge", "States": {"Charge": task, "Done": {"Type": "Succeed"}, "Refund": {"Type": "Pass", "Result": "refunded", "End": True}}} def run(name, definition, case, wait=90): if name not in CREATED: try: CREATED[name] = sfn.create_state_machine(name=name, definition=json.dumps(definition), roleArn=ROLE)["stateMachineArn"] except ClientError as e: raise SystemExit(f"{name}: {e.response['Error']['Code']}: {e.response['Error']['Message'][:200]}") ex = sfn.start_execution(stateMachineArn=f"{CREATED[name]}#{case}", name=f"{case}-{int(time.time() * 1000)}")["executionArn"] t0 = time.time() while time.time() - t0 < wait: d = sfn.describe_execution(executionArn=ex) if d["status"] != "RUNNING": break time.sleep(0.4) ev = sfn.get_execution_history(executionArn=ex, maxResults=500)["events"] return d, ev def gaps(events): """seconds from each failed attempt to the next scheduled attempt""" last_fail, out = None, [] for e in sorted(events, key=lambda e: e["id"]): if e["type"] == "TaskFailed": last_fail = e["timestamp"] elif e["type"] == "TaskScheduled" and last_fail is not None: out.append(round((e["timestamp"] - last_fail).total_seconds(), 1)); last_fail = None return out def attempts(events): return sum(1 for e in events if e["type"] == "TaskScheduled") def path(events): return [e["stateEnteredEventDetails"]["name"] for e in sorted(events, key=lambda e: e["id"]) if e["type"].endswith("StateEntered") and e["type"] != "ExecutionStarted"] ROWS = [] def show(label, d, ev, docs): g = gaps(ev) out = (d.get("output") or d.get("error") or "")[:40] ROWS.append((label, attempts(ev), g, d["status"], path(ev))) print(f"{label:60s} {attempts(ev):<4d} {str(g):18s} {d['status']:10s} {'>'.join(path(ev)):22s} [docs: {docs}]") return g R1 = [{"ErrorEquals": ["Payment.Transient"], "IntervalSeconds": 1, "MaxAttempts": 3, "BackoffRate": 2}] print(f"{'scenario':60s} {'tries':4s} {'waits (s)':18s} {'status':10s} {'states entered':22s}") d, ev = run("RetryDemo", machine(R1), "SucceedAfterTwo"); show("2 transient failures, then success (1 s, x2, max 3)", d, ev, "waits 1 then 2; success") d, ev = run("RetryDemo", machine(R1), "AlwaysFails"); show("always fails: retries used up, then Catch", d, ev, "3 retries (waits 1,2,4) then Catch") d, ev = run("RetryDemo", machine(R1), "OtherError"); show("error name not in ErrorEquals: no retry, Catch", d, ev, "no retry; Catch") R2 = [{"ErrorEquals": ["Payment.Transient"], "IntervalSeconds": 3, "MaxAttempts": 3, "BackoffRate": 2, "MaxDelaySeconds": 5}] d, ev = run("CappedRetry", machine(R2), "AlwaysFails"); show("interval 3, x2, max 3, MaxDelaySeconds 5", d, ev, "waits 3, 5, 5") R3 = [{"ErrorEquals": ["Payment.Transient"], "IntervalSeconds": 3, "MaxAttempts": 3, "BackoffRate": 2}] d, ev = run("UncappedRetry", machine(R3), "AlwaysFails"); show("interval 3, x2, max 3, no cap", d, ev, "waits 3, 6, 12") R4 = [{"ErrorEquals": ["Payment.Transient"], "IntervalSeconds": 2, "MaxAttempts": 3, "BackoffRate": 2, "JitterStrategy": "FULL"}] for i in (1, 2): d, ev = run("JitterRetry", machine(R4), "AlwaysFails"); show(f"JitterStrategy FULL, interval 2, x2, max 3 (run {i})", d, ev, "waits random in [0,2], [0,4], [0,8]") R5 = [{"ErrorEquals": ["States.Timeout"], "MaxAttempts": 0}, {"ErrorEquals": ["States.ALL"]}] d, ev = run("TimeoutNoRetry", machine(R5), "TimeoutThenOk"); show("docs' 'retry all except States.Timeout': Timeout thrown", d, ev, "States.Timeout never retried") d, ev = run("TimeoutNoRetry", machine(R5), "OtherThenOk"); show("same policy: another error thrown once", d, ev, "retried once, success") json.dump([(a, b, c, d_, e) for a, b, c, d_, e in ROWS], open("out/sfn_retry_rows.json", "w")) --- py sfn_retry.py (saved output of the real run) --- scenario tries waits (s) status states entered 2 transient failures, then success (1 s, x2, max 3) 3 [1.0, 2.0] SUCCEEDED Charge>Done [docs: waits 1 then 2; success] always fails: retries used up, then Catch 4 [1.0, 2.0, 4.0] SUCCEEDED Charge>Refund [docs: 3 retries (waits 1,2,4) then Catch] error name not in ErrorEquals: no retry, Catch 1 [] SUCCEEDED Charge>Refund [docs: no retry; Catch] interval 3, x2, max 3, MaxDelaySeconds 5 4 [3.0, 5.0, 5.0] SUCCEEDED Charge>Refund [docs: waits 3, 5, 5] interval 3, x2, max 3, no cap 4 [3.0, 6.0, 12.0] SUCCEEDED Charge>Refund [docs: waits 3, 6, 12] JitterStrategy FULL, interval 2, x2, max 3 (run 1) 4 [1.0, 3.0, 8.0] SUCCEEDED Charge>Refund [docs: waits random in [0,2], [0,4], [0,8]] JitterStrategy FULL, interval 2, x2, max 3 (run 2) 4 [2.0, 3.0, 1.0] SUCCEEDED Charge>Refund [docs: waits random in [0,2], [0,4], [0,8]] docs' 'retry all except States.Timeout': Timeout thrown 1 [] SUCCEEDED Charge>Refund [docs: States.Timeout never retried] same policy: another error thrown once 2 [1.0] SUCCEEDED Charge>Done [docs: retried once, success] --- cost9.py --- """What the order workflow costs on Step Functions Standard, from the pricing page read 2026-10-13 (fetch tool): $0.000025 per state transition in US East (N. Virginia), 4,000 free transitions a month, 'each retry will be charged as an additional state transition'. The page's Standard price table was missing from the text, so the rate comes from its worked examples. My count of transitions = states entered + retries; AWS's own count may differ by one (the page examples were not read in full). SQS, SNS and EventBridge: only the EventBridge custom-event price ($1.00 per million, Region not stated) was available; SQS and SNS request prices were NOT on the pages as fetched, so they are left out rather than guessed.""" import json RATE = 0.000025 FREE = 4000 rows = json.load(open("out/sfn_retry_rows.json")) paths = {} for label, tries, gaps, status, states in rows: paths[label] = (len(states), tries - 1) print(f"{'execution path (from the retry runs)':62s} {'states':6s} {'retries':7s} {'transitions':11s} {'per 1M executions'}") seen = set() for label, (states, retries) in paths.items(): key = (states, retries) if key in seen: continue seen.add(key) tr = states + retries print(f"{label:62s} {states:<6d} {retries:<7d} {tr:<11d} ${tr * 1_000_000 * RATE:,.2f}") print(f"\nfree tier: {FREE:,} transitions a month = {FREE / 4:,.0f} executions of a 4-transition path") print("one more retry on every execution of the 2-state path adds 50% to its transition count") print(f"EventBridge custom events: 1,000,000 events x $1.00/M = $1.00 (Region not stated on the page)") --- py cost9.py (saved output of the real run) --- execution path (from the retry runs) states retries transitions per 1M executions 2 transient failures, then success (1 s, x2, max 3) 2 2 4 $100.00 always fails: retries used up, then Catch 2 3 5 $125.00 error name not in ErrorEquals: no retry, Catch 2 0 2 $50.00 same policy: another error thrown once 2 1 3 $75.00 free tier: 4,000 transitions a month = 1,000 executions of a 4-transition path one more retry on every execution of the 2-state path adds 50% to its transition count EventBridge custom events: 1,000,000 events x $1.00/M = $1.00 (Region not stated on the page) --- asl_mutations.py --- """One-change mutations of a Step Functions definition. Who notices: Step Functions Local (CreateStateMachine) and cfn-lint (on a CloudFormation template that holds the definition as DefinitionString)? 'basis' says where my expectation comes from.""" import copy, json, os, re, subprocess, time from botocore.exceptions import ClientError from common import client sfn = client("stepfunctions", "http://127.0.0.1:8083") LINT = "C:/lwaws/c9/venv/Scripts/cfn-lint.exe" FN = "arn:aws:lambda:us-east-1:123456789012:function:validate" BASE = { "Comment": "order flow", "StartAt": "Validate", "States": { "Validate": {"Type": "Task", "Resource": "arn:aws:states:::lambda:invoke", "Parameters": {"FunctionName": FN, "Payload.$": "$"}, "ResultPath": "$.v", "Retry": [{"ErrorEquals": ["Lambda.ServiceException", "Lambda.TooManyRequestsException"], "IntervalSeconds": 2, "MaxAttempts": 3, "BackoffRate": 2}, {"ErrorEquals": ["States.ALL"], "MaxAttempts": 1}], "Catch": [{"ErrorEquals": ["States.ALL"], "Next": "Fail"}], "Next": "Route"}, "Route": {"Type": "Choice", "Choices": [{"Variable": "$.amount", "NumericGreaterThan": 1000, "Next": "Review"}], "Default": "Fulfil"}, "Review": {"Type": "Wait", "Seconds": 5, "Next": "Fulfil"}, "Fulfil": {"Type": "Parallel", "Branches": [{"StartAt": "A", "States": {"A": {"Type": "Pass", "End": True}}}, {"StartAt": "B", "States": {"B": {"Type": "Pass", "End": True}}}], "Next": "Each"}, "Each": {"Type": "Map", "ItemsPath": "$.items", "MaxConcurrency": 2, "Iterator": {"StartAt": "Item", "States": {"Item": {"Type": "Pass", "End": True}}}, "Next": "Done"}, "Done": {"Type": "Succeed"}, "Fail": {"Type": "Fail", "Error": "ValidationFailed", "Cause": "input rejected"}, }, } def S(d): return d["States"] def m(label, basis, expect, fn): return (label, basis, expect, fn) MUT = [ m("state Type 'Task' -> 'Taks'", "ASL spec (not read)", "invalid", lambda d: S(d)["Validate"].update(Type="Taks")), m("Next points to a state that does not exist", "ASL spec (not read)", "invalid", lambda d: S(d)["Review"].update(Next="Nowhere")), m("StartAt names a missing state", "ASL spec (not read)", "invalid", lambda d: d.update(StartAt="Nowhere")), m("StartAt removed", "ASL spec (not read)", "invalid", lambda d: d.pop("StartAt")), m("an unreachable extra state", "ASL spec (not read)", "invalid", lambda d: S(d).update(Orphan={"Type": "Pass", "End": True})), m("Choice rule without Next", "ASL spec (not read)", "invalid", lambda d: S(d)["Route"]["Choices"][0].pop("Next")), m("Choice without a Default", "error-handling page: a Choice with no matching rule is a runtime error", "valid", lambda d: S(d)["Route"].pop("Default")), m("Task without Resource", "ASL spec (not read)", "invalid", lambda d: S(d)["Validate"].pop("Resource")), m("Parallel without Branches", "ASL spec (not read)", "invalid", lambda d: S(d)["Fulfil"].pop("Branches")), m("a state with both Next and End", "ASL spec (not read)", "invalid", lambda d: S(d)["Review"].update(End=True)), m("Succeed state given a Next", "ASL spec (not read)", "invalid", lambda d: S(d)["Done"].update(Next="Fail")), m("Fail state given a Next", "ASL spec (not read)", "invalid", lambda d: S(d)["Fail"].update(Next="Done")), m("Wait with Seconds and Timestamp", "ASL spec (not read)", "invalid", lambda d: S(d)["Review"].update(Timestamp="2030-01-01T00:00:00Z")), m("Wait with neither Seconds nor Timestamp", "ASL spec (not read)", "invalid", lambda d: S(d)["Review"].pop("Seconds")), m("Map MaxConcurrency 'two'", "ASL spec (not read)", "invalid", lambda d: S(d)["Each"].update(MaxConcurrency="two")), m("Retry in a Pass state", "error-handling page: Retry is for Task, Parallel and Map", "invalid", lambda d: S(d)["Branches_dummy"] if False else S(d)["Fulfil"]["Branches"][0]["States"]["A"].update(Retry=[{"ErrorEquals": ["States.ALL"]}])), m("Retry MaxAttempts -1", "error-handling page: positive integer, 0 = never", "invalid", lambda d: S(d)["Validate"]["Retry"][0].update(MaxAttempts=-1)), m("Retry IntervalSeconds 0", "error-handling page: a positive integer", "invalid", lambda d: S(d)["Validate"]["Retry"][0].update(IntervalSeconds=0)), m("Retry BackoffRate 0.5", "page gives only the default (2.0)", "unknown", lambda d: S(d)["Validate"]["Retry"][0].update(BackoffRate=0.5)), m("Retry JitterStrategy 'HALF'", "error-handling page: FULL or NONE", "invalid", lambda d: S(d)["Validate"]["Retry"][0].update(JitterStrategy="HALF")), m("Retry MaxDelaySeconds 0", "error-handling page: greater than 0", "invalid", lambda d: S(d)["Validate"]["Retry"][0].update(MaxDelaySeconds=0)), m("States.ALL not last in Retry", "error-handling page: must be in the last retrier", "invalid", lambda d: S(d)["Validate"].update(Retry=[S(d)["Validate"]["Retry"][1], S(d)["Validate"]["Retry"][0]])), m("States.ALL together with another error name", "error-handling page: must appear alone", "invalid", lambda d: S(d)["Validate"]["Catch"][0].update(ErrorEquals=["States.ALL", "Custom.Error"])), m("Catch Next names a missing state", "ASL spec (not read)", "invalid", lambda d: S(d)["Validate"]["Catch"][0].update(Next="Nowhere")), m("custom error name starting with 'States.'", "error-handling page: custom names cannot begin with States.", "unknown", lambda d: S(d)["Validate"]["Retry"][0].update(ErrorEquals=["States.MadeUp"])), m("Fail state without Error", "ASL spec (not read)", "valid", lambda d: S(d)["Fail"].pop("Error")), ] def local_verdict(defn): name = "m" + str(int(time.time() * 1000))[-9:] try: sfn.create_state_machine(name=name, definition=json.dumps(defn), roleArn="arn:aws:iam::123456789012:role/DummyRole") return "accepted" except ClientError as e: return "rejected" def lint_verdict(defn, i): tpl = {"AWSTemplateFormatVersion": "2010-09-09", "Resources": {"SM": {"Type": "AWS::StepFunctions::StateMachine", "Properties": { "RoleArn": "arn:aws:iam::123456789012:role/R", "DefinitionString": json.dumps(defn)}}}} p = f"C:/lwaws/c9/_asl_{i}.json" open(p, "w").write(json.dumps(tpl)) r = subprocess.run([LINT, p], capture_output=True, text=True) os.remove(p) codes = sorted(set(re.findall(r"^([EW]\d{4})", r.stdout, re.M))) return ("rejected " + ",".join(codes)) if r.returncode in (2, 6) else ("warning " + ",".join(codes) if r.returncode == 4 else "accepted") print(f"{'change':52s} {'expect':8s} {'Step Functions Local':22s} {'cfn-lint 1.57.2'}") print(f"{'(control) the definition as written':52s} {'valid':8s} {local_verdict(BASE):22s} {lint_verdict(BASE, 0)}") rows, missed_both, wrong = [], [], [] for i, (label, basis, expect, fn) in enumerate(MUT, 1): d = copy.deepcopy(BASE) try: fn(d) except KeyError as e: print("mutation failed to apply:", label, e); continue a, b = local_verdict(d), lint_verdict(d, i) rows.append((label, basis, expect, a, b)) print(f"{label:52s} {expect:8s} {a:22s} {b}") inv = [r for r in rows if r[2] == "invalid"] print(f"\n{len(rows)} changes; {len(inv)} are invalid by the docs or the ASL spec as I read it.") print(f"of those {len(inv)}: rejected by Step Functions Local {sum(r[3] == 'rejected' for r in inv)}, by cfn-lint {sum(r[4].startswith('rejected') for r in inv)}, " f"by at least one {sum(r[3] == 'rejected' or r[4].startswith('rejected') for r in inv)}, by neither {sum(r[3] != 'rejected' and not r[4].startswith('rejected') for r in inv)}") print("accepted by both although invalid:", [r[0] for r in inv if r[3] != 'rejected' and not r[4].startswith('rejected')]) print("'valid' or 'unknown' cases rejected by a tool:", [(r[0], r[3], r[4]) for r in rows if r[2] != "invalid" and (r[3] == "rejected" or r[4].startswith("rejected"))]) json.dump(rows, open("out/asl_rows.json", "w")) --- py asl_mutations.py (saved output of the real run) --- change expect Step Functions Local cfn-lint 1.57.2 (control) the definition as written valid accepted accepted state Type 'Task' -> 'Taks' invalid rejected rejected E3601 Next points to a state that does not exist invalid rejected accepted StartAt names a missing state invalid rejected rejected E3601 StartAt removed invalid rejected rejected E3601 an unreachable extra state invalid rejected accepted Choice rule without Next invalid rejected rejected E1101 Choice without a Default valid accepted accepted Task without Resource invalid rejected rejected E3601 Parallel without Branches invalid rejected rejected E3601 a state with both Next and End invalid rejected rejected E3601 Succeed state given a Next invalid rejected rejected E3601 Fail state given a Next invalid rejected rejected E3601 Wait with Seconds and Timestamp invalid rejected accepted Wait with neither Seconds nor Timestamp invalid rejected accepted Map MaxConcurrency 'two' invalid rejected rejected E3601 Retry in a Pass state invalid rejected rejected E1101 Retry MaxAttempts -1 invalid rejected rejected E1101 Retry IntervalSeconds 0 invalid accepted accepted Retry BackoffRate 0.5 unknown rejected accepted Retry JitterStrategy 'HALF' invalid rejected accepted Retry MaxDelaySeconds 0 invalid rejected accepted States.ALL not last in Retry invalid rejected accepted States.ALL together with another error name invalid rejected accepted Catch Next names a missing state invalid rejected accepted custom error name starting with 'States.' unknown accepted accepted Fail state without Error valid accepted accepted 26 changes; 22 are invalid by the docs or the ASL spec as I read it. of those 22: rejected by Step Functions Local 21, by cfn-lint 12, by at least one 21, by neither 1 accepted by both although invalid: ['Retry IntervalSeconds 0'] 'valid' or 'unknown' cases rejected by a tool: [('Retry BackoffRate 0.5', 'rejected', 'accepted')] --- probe of DefinitionString plain versus Fn::Sub (inline script) (saved output of the real run) --- valid definition, plain string exit 0 - valid definition, wrapped in Fn::Sub exit 4 W1020 broken JSON (missing brace), plain string exit 0 - broken JSON (missing brace), wrapped in Fn::Sub exit 4 W1020 Next to a missing state, plain string exit 0 - Next to a missing state, wrapped in Fn::Sub exit 4 W1020 EXPRESS + .waitForTaskToken, plain string exit 0 - EXPRESS + .waitForTaskToken, wrapped in Fn::Sub exit 4 W1020 STANDARD + .waitForTaskToken, plain string (should be fine) exit 0 - --- stack.yaml --- AWSTemplateFormatVersion: "2010-09-09" Description: Chapter 9 messaging design - queue with dead-letter queue, topic fan-out, FIFO queue, event bus rule, workflow Parameters: FunctionTimeout: Type: Number Default: 30 Resources: OrdersDLQ: Type: AWS::SQS::Queue Properties: QueueName: orders-dlq MessageRetentionPeriod: 1209600 OrdersQueue: Type: AWS::SQS::Queue Properties: QueueName: orders VisibilityTimeout: 185 MessageRetentionPeriod: 345600 ReceiveMessageWaitTimeSeconds: 20 RedrivePolicy: deadLetterTargetArn: !GetAtt OrdersDLQ.Arn maxReceiveCount: 5 PaymentsQueue: Type: AWS::SQS::Queue Properties: QueueName: payments.fifo FifoQueue: true ContentBasedDeduplication: true VisibilityTimeout: 60 OrdersTopic: Type: AWS::SNS::Topic Properties: TopicName: order-events OrdersQueuePolicy: Type: AWS::SQS::QueuePolicy Properties: Queues: [!Ref OrdersQueue] PolicyDocument: Version: "2012-10-17" Statement: - Effect: Allow Principal: {Service: sns.amazonaws.com} Action: sqs:SendMessage Resource: !GetAtt OrdersQueue.Arn Condition: ArnEquals: {"aws:SourceArn": !Ref OrdersTopic} OrdersSubscription: Type: AWS::SNS::Subscription Properties: TopicArn: !Ref OrdersTopic Protocol: sqs Endpoint: !GetAtt OrdersQueue.Arn RawMessageDelivery: true FilterPolicyScope: MessageAttributes FilterPolicy: store: [example_corp] price_usd: - numeric: [">=", 100] Bus: Type: AWS::Events::EventBus Properties: Name: orders-bus AuditQueue: Type: AWS::SQS::Queue Properties: QueueName: audit AuditQueuePolicy: Type: AWS::SQS::QueuePolicy Properties: Queues: [!Ref AuditQueue] PolicyDocument: Version: "2012-10-17" Statement: - Effect: Allow Principal: {Service: events.amazonaws.com} Action: sqs:SendMessage Resource: !GetAtt AuditQueue.Arn Condition: ArnEquals: {"aws:SourceArn": !GetAtt AuditRule.Arn} AuditRule: Type: AWS::Events::Rule Properties: EventBusName: !Ref Bus EventPattern: source: [app.orders] detail-type: [Order Placed] detail: total: [{numeric: [">", 100]}] Targets: - Id: audit Arn: !GetAtt AuditQueue.Arn RetryPolicy: {MaximumRetryAttempts: 10, MaximumEventAgeInSeconds: 3600} DeadLetterConfig: {Arn: !GetAtt OrdersDLQ.Arn} ConsumerRole: 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 - arn:aws:iam::aws:policy/service-role/AWSLambdaSQSQueueExecutionRole Consumer: Type: AWS::Lambda::Function Properties: Runtime: python3.13 Handler: index.handler Role: !GetAtt ConsumerRole.Arn Timeout: !Ref FunctionTimeout Code: ZipFile: | def handler(event, context): return {"batchItemFailures": []} ConsumerMapping: Type: AWS::Lambda::EventSourceMapping Properties: FunctionName: !Ref Consumer EventSourceArn: !GetAtt OrdersQueue.Arn BatchSize: 10 MaximumBatchingWindowInSeconds: 5 FunctionResponseTypes: [ReportBatchItemFailures] ScalingConfig: {MaximumConcurrency: 5} WorkflowRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: "2012-10-17" Statement: - {Effect: Allow, Principal: {Service: states.amazonaws.com}, Action: "sts:AssumeRole"} Workflow: Type: AWS::StepFunctions::StateMachine Properties: StateMachineType: STANDARD RoleArn: !GetAtt WorkflowRole.Arn DefinitionString: !Sub | { "StartAt": "Charge", "States": { "Charge": { "Type": "Task", "Resource": "arn:aws:states:::lambda:invoke", "Parameters": {"FunctionName": "${Consumer.Arn}", "Payload.$": "$"}, "Retry": [{"ErrorEquals": ["Lambda.ServiceException", "Lambda.TooManyRequestsException"], "IntervalSeconds": 2, "MaxAttempts": 3, "BackoffRate": 2}], "Catch": [{"ErrorEquals": ["States.ALL"], "Next": "Refund"}], "Next": "Done" }, "Refund": {"Type": "Pass", "End": true}, "Done": {"Type": "Succeed"} } } --- mutate_stack.py --- """One-change mutations of stack.yaml, linted with cfn-lint 1.57.2. 'basis' says which docs page (read 2026-10-13) the expectation rests on; 'semantic' = the template is well-formed but breaks a rule that needs knowledge of two resources together.""" import os, re, subprocess LINT = "C:/lwaws/c9/venv/Scripts/cfn-lint.exe" SRC = open("C:/lwaws/c9/stack.yaml", encoding="utf-8").read() M = [ ("queue VisibilityTimeout 185 -> 43201", "SQS quotas: max 12 h (43,200 s)", "VisibilityTimeout: 185", "VisibilityTimeout: 43201"), ("queue MessageRetentionPeriod -> 59", "SQS quotas: min 60 s", "MessageRetentionPeriod: 345600", "MessageRetentionPeriod: 59"), ("queue MessageRetentionPeriod -> 1209601", "SQS quotas: max 14 days", "MessageRetentionPeriod: 1209600", "MessageRetentionPeriod: 1209601"), ("ReceiveMessageWaitTimeSeconds 20 -> 21", "SQS polling page: max long-poll wait 20 s", "ReceiveMessageWaitTimeSeconds: 20", "ReceiveMessageWaitTimeSeconds: 21"), ("queue MaximumMessageSize above 1 MiB", "SQS quotas: max 1,048,576 bytes", " QueueName: orders\n", " QueueName: orders\n MaximumMessageSize: 1048577\n"), ("queue DelaySeconds 901", "SQS quotas: max 15 minutes", " QueueName: audit\n", " QueueName: audit\n DelaySeconds: 901\n"), ("FIFO queue name without .fifo", "FIFO queues must be named with .fifo (SQS/SNS pages: not read, my understanding)", "QueueName: payments.fifo", "QueueName: payments"), ("ContentBasedDeduplication on a standard queue", "FIFO-only attribute (dedup page describes FIFO only)", " QueueName: audit\n", " QueueName: audit\n ContentBasedDeduplication: true\n"), ("maxReceiveCount 5 -> 'five'", "DLQ page: a number", "maxReceiveCount: 5", "maxReceiveCount: five"), ("DLQ target ARN is a queue that does not exist", "semantic", "deadLetterTargetArn: !GetAtt OrdersDLQ.Arn", "deadLetterTargetArn: !GetAtt NoSuchQueue.Arn"), ("DLQ target ARN is the queue's own ARN", "semantic (a queue cannot be its own DLQ; page not read)", "deadLetterTargetArn: !GetAtt OrdersDLQ.Arn", "deadLetterTargetArn: !GetAtt OrdersQueue.Arn"), ("event source mapping BatchSize 10 -> 10001", "Lambda SQS page: standard queue max 10,000", " BatchSize: 10\n", " BatchSize: 10001\n"), ("BatchSize 100 with no batching window", "Lambda SQS page: over 10 needs a window of at least 1 s", " BatchSize: 10\n MaximumBatchingWindowInSeconds: 5\n", " BatchSize: 100\n"), ("FIFO queue mapped with BatchSize 11", "Lambda SQS page: FIFO max 10", " EventSourceArn: !GetAtt OrdersQueue.Arn\n BatchSize: 10\n", " EventSourceArn: !GetAtt PaymentsQueue.Arn\n BatchSize: 11\n"), ("function Timeout 30 -> 600 (queue timeout stays 185)", "Lambda SQS page: function timeout must not exceed the queue's visibility timeout", " Default: 30", " Default: 600"), ("queue VisibilityTimeout 185 -> 60 (less than 6 x 30 s)", "Lambda SQS page: at least six times the function timeout", "VisibilityTimeout: 185", "VisibilityTimeout: 60"), ("subscription Endpoint is the queue URL, not its ARN", "semantic (SQS endpoint must be an ARN; my understanding)", "Endpoint: !GetAtt OrdersQueue.Arn", "Endpoint: !Ref OrdersQueue"), ("subscription RawMessageDelivery 'yes'", "SNS raw delivery page: attribute set to true", "RawMessageDelivery: true", "RawMessageDelivery: yes"), ("FilterPolicyScope 'Body'", "SNS filter pages: MessageAttributes or MessageBody", "FilterPolicyScope: MessageAttributes", "FilterPolicyScope: Body"), ("filter policy operator 'numerc'", "SNS example filter policies page: numeric", "- numeric: [\">=\", 100]", "- numerc: [\">=\", 100]"), ("queue policy removed (SNS may not send to the queue)", "semantic (my understanding; the fan-out page did not load)", " OrdersQueuePolicy:\n Type: AWS::SQS::QueuePolicy\n Properties:\n Queues: [!Ref OrdersQueue]\n PolicyDocument:\n Version: \"2012-10-17\"\n Statement:\n - Effect: Allow\n Principal: {Service: sns.amazonaws.com}\n Action: sqs:SendMessage\n Resource: !GetAtt OrdersQueue.Arn\n Condition:\n ArnEquals: {\"aws:SourceArn\": !Ref OrdersTopic}\n\n", ""), ("audit queue policy trusts the wrong service", "semantic (events.amazonaws.com is the sender)", "Principal: {Service: events.amazonaws.com}", "Principal: {Service: sns.amazonaws.com}"), ("event pattern with an unknown operator 'prefx'", "operators page: prefix", "total: [{numeric: [\">\", 100]}]", "total: [{prefx: \"1\"}]"), ("event pattern numeric with a string operand", "operators page: numeric takes numbers", "total: [{numeric: [\">\", 100]}]", "total: [{numeric: [\">\", high]}]"), ("rule target RetryPolicy MaximumRetryAttempts 'many'", "retry policy page: number of attempts", "MaximumRetryAttempts: 10", "MaximumRetryAttempts: many"), ("Express workflow that has a callback (.waitForTaskToken)", "workflow-type page: Express has no callback pattern", "StateMachineType: STANDARD", "StateMachineType: EXPRESS"), ("state machine definition is not valid JSON", "semantic", "\"Done\": {\"Type\": \"Succeed\"}", "\"Done\": {\"Type\": \"Succeed\""), ("state machine Next names a state that does not exist", "ASL: states must exist", "\"Next\": \"Done\"\n", "\"Next\": \"Nowhere\"\n"), ("Lambda runtime python3.7 (retired)", "linter rule seen in chapter 8", "Runtime: python3.13", "Runtime: python3.7"), ("consumer role trusted by the wrong service", "semantic", "Principal: {Service: lambda.amazonaws.com}", "Principal: {Service: events.amazonaws.com}"), ] # the callback case needs a callback resource: patch the Resource in that one mutation EXTRA = {"Express workflow that has a callback (.waitForTaskToken)": ("\"Resource\": \"arn:aws:states:::lambda:invoke\"", "\"Resource\": \"arn:aws:states:::lambda:invoke.waitForTaskToken\"")} def lint(text, tag): p = f"C:/lwaws/c9/_m{tag}.yaml" open(p, "w", encoding="utf-8").write(text) r = subprocess.run([LINT, p], capture_output=True, text=True) os.remove(p) rules = sorted(set(re.findall(r"^([EW]\d{4})", r.stdout, re.M))) return r.returncode, rules rc, rules = lint(SRC, 0) print(f"{'change':62s} {'cfn-lint 1.57.2'} basis") print(f"{'(control) template as written':62s} {'exit ' + str(rc)}") rows = [] for i, (label, basis, old, new) in enumerate(M, 1): assert SRC.count(old) == 1, (label, SRC.count(old)) t = SRC.replace(old, new) if label in EXTRA: o2, n2 = EXTRA[label]; assert t.count(o2) == 1; t = t.replace(o2, n2) rc, rules = lint(t, i) verdict = "caught " + ",".join(r for r in rules if r.startswith("E")) if rc in (2, 6) else ("warning " + ",".join(rules) if rc == 4 else "missed") rows.append((label, basis, verdict)) print(f"{label:62s} {verdict:22s} {basis}") caught = sum(r[2].startswith("caught") for r in rows) print(f"\n{len(rows)} one-change mutations: cfn-lint reported an error for {caught}; {len(rows) - caught} passed or only warned") print("missed:", [r[0] for r in rows if not r[2].startswith("caught")]) --- py mutate_stack.py (saved output of the real run) --- change cfn-lint 1.57.2 basis (control) template as written exit 0 queue VisibilityTimeout 185 -> 43201 caught E3034 SQS quotas: max 12 h (43,200 s) queue MessageRetentionPeriod -> 59 caught E3034 SQS quotas: min 60 s queue MessageRetentionPeriod -> 1209601 caught E3034 SQS quotas: max 14 days ReceiveMessageWaitTimeSeconds 20 -> 21 caught E3034 SQS polling page: max long-poll wait 20 s queue MaximumMessageSize above 1 MiB caught E3034 SQS quotas: max 1,048,576 bytes queue DelaySeconds 901 caught E3034 SQS quotas: max 15 minutes FIFO queue name without .fifo caught E3501 FIFO queues must be named with .fifo (SQS/SNS pages: not read, my understanding) ContentBasedDeduplication on a standard queue caught E3501 FIFO-only attribute (dedup page describes FIFO only) maxReceiveCount 5 -> 'five' caught E3012 DLQ page: a number DLQ target ARN is a queue that does not exist caught E1010 semantic DLQ target ARN is the queue's own ARN caught E3004 semantic (a queue cannot be its own DLQ; page not read) event source mapping BatchSize 10 -> 10001 caught E3034 Lambda SQS page: standard queue max 10,000 BatchSize 100 with no batching window missed Lambda SQS page: over 10 needs a window of at least 1 s FIFO queue mapped with BatchSize 11 caught E3705 Lambda SQS page: FIFO max 10 function Timeout 30 -> 600 (queue timeout stays 185) missed Lambda SQS page: function timeout must not exceed the queue's visibility timeout queue VisibilityTimeout 185 -> 60 (less than 6 x 30 s) missed Lambda SQS page: at least six times the function timeout subscription Endpoint is the queue URL, not its ARN missed semantic (SQS endpoint must be an ARN; my understanding) subscription RawMessageDelivery 'yes' missed SNS raw delivery page: attribute set to true FilterPolicyScope 'Body' missed SNS filter pages: MessageAttributes or MessageBody filter policy operator 'numerc' missed SNS example filter policies page: numeric queue policy removed (SNS may not send to the queue) missed semantic (my understanding; the fan-out page did not load) audit queue policy trusts the wrong service missed semantic (events.amazonaws.com is the sender) event pattern with an unknown operator 'prefx' missed operators page: prefix event pattern numeric with a string operand missed operators page: numeric takes numbers rule target RetryPolicy MaximumRetryAttempts 'many' caught E3012 retry policy page: number of attempts Express workflow that has a callback (.waitForTaskToken) missed workflow-type page: Express has no callback pattern state machine definition is not valid JSON missed semantic state machine Next names a state that does not exist missed ASL: states must exist Lambda runtime python3.7 (retired) caught E2531 linter rule seen in chapter 8 consumer role trusted by the wrong service missed semantic 30 one-change mutations: cfn-lint reported an error for 15; 15 passed or only warned missed: ['BatchSize 100 with no batching window', 'function Timeout 30 -> 600 (queue timeout stays 185)', 'queue VisibilityTimeout 185 -> 60 (less than 6 x 30 s)', 'subscription Endpoint is the queue URL, not its ARN', "subscription RawMessageDelivery 'yes'", "FilterPolicyScope 'Body'", "filter policy operator 'numerc'", 'queue policy removed (SNS may not send to the queue)', 'audit queue policy trusts the wrong service', "event pattern with an unknown operator 'prefx'", 'event pattern numeric with a string operand', 'Express workflow that has a callback (.waitForTaskToken)', 'state machine definition is not valid JSON', 'state machine Next names a state that does not exist', 'consumer role trusted by the wrong service'] --- crosscheck.py --- """Cross-resource checks that cfn-lint did not make, written from the Lambda SQS page (read 2026-10-13): - function timeout <= queue visibility timeout (Lambda validates this when the mapping is created) - visibility timeout >= 6 x function timeout + MaximumBatchingWindowInSeconds (recommended) - FIFO queue: batch size <= 10; standard queue: batch size <= 10,000; batch size > 10 needs a window of at least 1 s - maxReceiveCount >= 5 recommended on a queue read by Lambda - the DLQ's retention should be longer than the source queue's (SQS dead-letter-queue page) It reads the template, follows Ref/GetAtt between the resources, and prints findings. It is deliberately small and only handles this shape of template.""" import sys import yaml class Loader(yaml.SafeLoader): pass def tag(loader, suffix, node): if isinstance(node, yaml.ScalarNode): return {suffix: loader.construct_scalar(node)} if isinstance(node, yaml.SequenceNode): return {suffix: loader.construct_sequence(node)} return {suffix: loader.construct_mapping(node)} Loader.add_multi_constructor("!", tag) def ref_name(v): """the logical ID a Ref or GetAtt points at, else None""" if isinstance(v, dict): if "Ref" in v: return v["Ref"] if "GetAtt" in v: g = v["GetAtt"]; return (g.split(".")[0] if isinstance(g, str) else g[0]) return None def check(text): t = yaml.load(text, Loader=Loader) R, P = t["Resources"], t.get("Parameters", {}) out = [] def val(v, default=None): n = ref_name(v) if n in P: return P[n].get("Default", default) return v if v is not None else default for mid, m in R.items(): if m["Type"] != "AWS::Lambda::EventSourceMapping": continue pr = m["Properties"] q, fn = R.get(ref_name(pr["EventSourceArn"])), R.get(ref_name(pr["FunctionName"])) if not q or q["Type"] != "AWS::SQS::Queue" or not fn: continue qp, fp = q["Properties"], fn["Properties"] vt = int(qp.get("VisibilityTimeout", 30)) timeout = int(val(fp.get("Timeout"), 3)) window = int(pr.get("MaximumBatchingWindowInSeconds", 0)) batch = int(pr.get("BatchSize", 10)) fifo = bool(qp.get("FifoQueue")) if timeout > vt: out.append(("ERROR", mid, f"function timeout {timeout} s is longer than the queue's visibility timeout {vt} s (Lambda rejects this mapping)")) need = 6 * timeout + window if vt < need: out.append(("WARN", mid, f"visibility timeout {vt} s is below the recommended 6 x {timeout} s + {window} s window = {need} s")) if fifo and batch > 10: out.append(("ERROR", mid, f"batch size {batch} on a FIFO queue (max 10)")) if not fifo and batch > 10000: out.append(("ERROR", mid, f"batch size {batch} on a standard queue (max 10,000)")) if batch > 10 and window < 1: out.append(("ERROR", mid, f"batch size {batch} > 10 needs MaximumBatchingWindowInSeconds of at least 1")) rp = qp.get("RedrivePolicy") if rp: mrc = int(rp.get("maxReceiveCount", 0)) if mrc < 5: out.append(("WARN", mid, f"maxReceiveCount {mrc} is below the recommended 5 for a Lambda-fed queue")) d = R.get(ref_name(rp.get("deadLetterTargetArn"))) if d and int(d["Properties"].get("MessageRetentionPeriod", 345600)) <= int(qp.get("MessageRetentionPeriod", 345600)): out.append(("WARN", mid, "the dead-letter queue's retention is not longer than the source queue's")) return out if __name__ == "__main__": for path in sys.argv[1:]: res = check(open(path, encoding="utf-8").read()) print(path, "->", "no findings" if not res else "") for lvl, mid, msg in res: print(f" {lvl:5s} {mid}: {msg}") --- crosscheck_demo.py --- import crosscheck SRC = open("C:/lwaws/c9/stack.yaml", encoding="utf-8").read() CASES = [ ("the template as written", []), ("function Timeout 30 -> 600 (queue timeout stays 185)", [(" Default: 30", " Default: 600")]), ("queue VisibilityTimeout 185 -> 60", [("VisibilityTimeout: 185", "VisibilityTimeout: 60")]), ("queue VisibilityTimeout 180 (my first draft, with a 5 s window)", [("VisibilityTimeout: 185", "VisibilityTimeout: 180")]), ("BatchSize 100 with no batching window", [(" BatchSize: 10\n MaximumBatchingWindowInSeconds: 5\n", " BatchSize: 100\n")]), ("FIFO queue mapped with BatchSize 11", [(" EventSourceArn: !GetAtt OrdersQueue.Arn\n BatchSize: 10\n", " EventSourceArn: !GetAtt PaymentsQueue.Arn\n BatchSize: 11\n")]), ("maxReceiveCount 5 -> 3", [("maxReceiveCount: 5", "maxReceiveCount: 3")]), ("DLQ retention 1209600 -> 345600 (same as source)", [("MessageRetentionPeriod: 1209600", "MessageRetentionPeriod: 345600")]), ] for label, subs in CASES: t = SRC for a, b in subs: assert t.count(a) == 1, (label, a) t = t.replace(a, b) res = crosscheck.check(t) print(f"{label}\n" + (" no findings" if not res else "\n".join(f" {l:5s} {m}" for l, _, m in res))) --- py crosscheck_demo.py (saved output of the real run) --- the template as written no findings function Timeout 30 -> 600 (queue timeout stays 185) ERROR function timeout 600 s is longer than the queue's visibility timeout 185 s (Lambda rejects this mapping) WARN visibility timeout 185 s is below the recommended 6 x 600 s + 5 s window = 3605 s queue VisibilityTimeout 185 -> 60 WARN visibility timeout 60 s is below the recommended 6 x 30 s + 5 s window = 185 s queue VisibilityTimeout 180 (my first draft, with a 5 s window) WARN visibility timeout 180 s is below the recommended 6 x 30 s + 5 s window = 185 s BatchSize 100 with no batching window ERROR batch size 100 > 10 needs MaximumBatchingWindowInSeconds of at least 1 FIFO queue mapped with BatchSize 11 WARN visibility timeout 60 s is below the recommended 6 x 30 s + 5 s window = 185 s ERROR batch size 11 on a FIFO queue (max 10) maxReceiveCount 5 -> 3 WARN maxReceiveCount 3 is below the recommended 5 for a Lambda-fed queue DLQ retention 1209600 -> 345600 (same as source) WARN the dead-letter queue's retention is not longer than the source queue's --- lambda_batch/fn/app.py --- import json def handler(event, context): """Process each record on its own; report only the failures (ReportBatchItemFailures).""" failures = [] for record in event["Records"]: try: body = json.loads(record["body"]) if body.get("poison"): raise ValueError("cannot process " + body["order"]) print("processed", body["order"]) except Exception as exc: print("failed", record["messageId"], exc) failures.append({"itemIdentifier": record["messageId"]}) return {"batchItemFailures": failures} --- sam local invoke BatchFn -e ev_five.json (saved output of the real run) --- No current session found, using default AWS::AccountId START RequestId: c2e777de-713f-48da-a03b-0fcb7ceaf46a Version: $LATEST processed A-1 failed id2 cannot process A-2 processed A-3 failed id4 cannot process A-4 processed A-5 END RequestId: c2e777de-713f-48da-a03b-0fcb7ceaf46a REPORT RequestId: c2e777de-713f-48da-a03b-0fcb7ceaf46a Init Duration: 0.04 ms Duration: 216.66 ms Billed Duration: 217 ms Memory Size: 128 MB Max Memory Used: 128 MB {"batchItemFailures": [{"itemIdentifier": "id2"}, {"itemIdentifier": "id4"}]} -------------------------- Reading the result: retry waits on Step Functions Local matched the docs exactly (1,2 / 1,2,4 / 3,5,5 with MaxDelaySeconds 5 / 3,6,12 without); jitter stayed inside the documented ranges. Of 22 invalid definitions Step Functions Local rejected 21 and cfn-lint 12; both accepted IntervalSeconds 0. cfn-lint passed 15 of 30 template mutations, including everything that needs two resources: it did not check a definition's JSON or its Next targets even when the definition was a plain string. The cross-resource checker flagged the Lambda/SQS rules, including my own first-draft visibility timeout of 180 s (the docs' rule with a 5 s window and a 30 s function is 185 s). Mistakes made and fixed: (1) the first draft of stack.yaml had VisibilityTimeout 180, below the docs' recommended 6 x 30 + 5; I found it by rereading the Lambda page against the template, not through a tool, then wrote crosscheck.py and changed the template (the mutation script's labels were updated to 185). (2) I first thought Fn::Sub was hiding the definition from cfn-lint; the probe showed the same definitions pass as plain strings, so the cause is that this cfn-lint version does not check them. How to avoid both: reread the docs' numbers against your own template, and test one suspected cause at a time.