aws2-9 Exercise 1: SQS Against the Documentation, Its Limits, and an At-Least-Once Consumer ======================================================================== 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. --- setup used for every script --- set AWS_ACCESS_KEY_ID=test, AWS_SECRET_ACCESS_KEY=test, AWS_DEFAULT_REGION=eu-west-1 (us-east-1 for Step Functions Local) py -m venv venv ; venv\Scripts\python.exe -m pip install "moto[server,sqs,sns,events,stepfunctions]" boto3 cfn-lint PyYAML jsonschema venv\Scripts\python.exe -m moto.server -p 5000 (leave running; stop it when done) --- common.py --- import boto3, os, time, json os.environ.setdefault("AWS_ACCESS_KEY_ID", "test"); os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "test") EP = "http://127.0.0.1:5000"; REGION = "eu-west-1" def client(name, ep=EP): return boto3.client(name, endpoint_url=ep, region_name=REGION) def uniq(p): return f"{p}-{int(time.time()*1000) % 10**9}" --- sqs_behaviour.py --- """SQS behaviour documented in the developer guide, checked against the moto emulator (moto 5.2.4, server mode). Each check prints what the docs say and what the emulator did. Nothing here talks to AWS.""" import threading, time from botocore.exceptions import ClientError from common import client, uniq sqs = client("sqs") results = [] def check(name, docs, got, ok): results.append(ok) print(f"{'ok ' if ok else 'DIFF'} {name}\n docs: {docs}\n emulator: {got}") def mkq(prefix, **attrs): return sqs.create_queue(QueueName=uniq(prefix), Attributes={k: str(v) for k, v in attrs.items()})["QueueUrl"] def recv(q, **kw): return sqs.receive_message(QueueUrl=q, MaxNumberOfMessages=10, AttributeNames=["All"], **kw).get("Messages", []) # 1. visibility timeout q = mkq("vt", VisibilityTimeout=2) sqs.send_message(QueueUrl=q, MessageBody="m1") m = recv(q) again = recv(q) time.sleep(2.4) m2 = recv(q) check("visibility timeout hides a received message, then it reappears", "received message is invisible until the timeout expires; then another receive can get it", f"1st receive {len(m)} message, immediate 2nd receive {len(again)}, after 2.4 s receive {len(m2)} (ReceiveCount {m2[0]['Attributes']['ApproximateReceiveCount'] if m2 else '-'})", len(m) == 1 and len(again) == 0 and len(m2) == 1) # 2. ChangeMessageVisibility to 0 ends the timeout at once q = mkq("cmv", VisibilityTimeout=30) sqs.send_message(QueueUrl=q, MessageBody="m") m = recv(q) sqs.change_message_visibility(QueueUrl=q, ReceiptHandle=m[0]["ReceiptHandle"], VisibilityTimeout=0) m2 = recv(q) check("ChangeMessageVisibility to 0 makes the message available at once", "setting VisibilityTimeout to 0 'immediately makes the message available for other consumers'", f"receive right after: {len(m2)} message", len(m2) == 1) # 3. dead-letter queue dlq = mkq("dlq") dlq_arn = sqs.get_queue_attributes(QueueUrl=dlq, AttributeNames=["QueueArn"])["Attributes"]["QueueArn"] import json q = mkq("src", VisibilityTimeout=1, RedrivePolicy=json.dumps({"deadLetterTargetArn": dlq_arn, "maxReceiveCount": 3})) sqs.send_message(QueueUrl=q, MessageBody="poison") counts = [] for i in range(6): got = recv(q) counts.append(len(got)) time.sleep(1.3) in_dlq = recv(dlq) check("after maxReceiveCount receives the message moves to the dead-letter queue", "maxReceiveCount = 'the number of times a consumer can receive a message from a source queue before it is moved to a dead-letter queue'", f"receives that returned the message (3 allowed): {counts}; message in DLQ afterwards: {[x['Body'] for x in in_dlq]}", sum(counts) == 3 and len(in_dlq) == 1) # 4. FIFO fq = sqs.create_queue(QueueName=uniq("f") + ".fifo", Attributes={"FifoQueue": "true"})["QueueUrl"] try: sqs.send_message(QueueUrl=fq, MessageBody="x", MessageDeduplicationId="d1") got = "accepted" except ClientError as e: got = e.response["Error"]["Code"] check("FIFO send without MessageGroupId fails", "'MessageGroupId is required for FIFO queues' (the action fails)", got, got != "accepted") sqs.send_message(QueueUrl=fq, MessageBody="a", MessageGroupId="g", MessageDeduplicationId="same") sqs.send_message(QueueUrl=fq, MessageBody="a-retry", MessageGroupId="g", MessageDeduplicationId="same") msgs = recv(fq) check("FIFO drops a resend with the same deduplication id inside the 5-minute interval", "'If you retry the SendMessage action within the 5-minute deduplication interval, Amazon SQS doesn't introduce any duplicates'", f"two sends with the same id, receive returned {[x['Body'] for x in msgs]}", [x["Body"] for x in msgs] == ["a"]) fq2 = sqs.create_queue(QueueName=uniq("fc") + ".fifo", Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true"})["QueueUrl"] sqs.send_message(QueueUrl=fq2, MessageBody="same body", MessageGroupId="g") sqs.send_message(QueueUrl=fq2, MessageBody="same body", MessageGroupId="g", MessageAttributes={"k": {"DataType": "String", "StringValue": "different"}}) msgs = recv(fq2) check("content-based deduplication hashes the body only, not attributes", "SHA-256 hash 'using the body of the message but not the attributes of the message'", f"same body with different attributes sent twice, receive returned {len(msgs)}", len(msgs) == 1) fq3 = sqs.create_queue(QueueName=uniq("fo") + ".fifo", Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true", "VisibilityTimeout": "30"})["QueueUrl"] for body in ("g1-1", "g1-2", "g2-1"): sqs.send_message(QueueUrl=fq3, MessageBody=body, MessageGroupId=body.split("-")[0]) batch = recv(fq3) print(f"info one receive asking for up to 10 messages returned {[x['Body'] for x in batch]} (the page I read does not say whether a single call may return several messages of one group, so this is not scored)") fq4 = sqs.create_queue(QueueName=uniq("fo1") + ".fifo", Attributes={"FifoQueue": "true", "ContentBasedDeduplication": "true", "VisibilityTimeout": "30"})["QueueUrl"] for body in ("g1-1", "g1-2", "g2-1"): sqs.send_message(QueueUrl=fq4, MessageBody=body, MessageGroupId=body.split("-")[0]) one = sqs.receive_message(QueueUrl=fq4, MaxNumberOfMessages=1).get("Messages", []) two = sqs.receive_message(QueueUrl=fq4, MaxNumberOfMessages=1).get("Messages", []) three = sqs.receive_message(QueueUrl=fq4, MaxNumberOfMessages=1).get("Messages", []) check("FIFO: while a group's message is in flight, later messages of that group are not delivered", "'subsequent messages in that group are not made available until the in-flight message is either deleted or the visibility timeout expires'", f"one message per receive, nothing deleted: {[[x['Body'] for x in r] for r in (one, two, three)]}", [x["Body"] for x in one] == ["g1-1"] and all(x["Body"] != "g1-2" for x in two + three)) sqs.delete_message(QueueUrl=fq4, ReceiptHandle=one[0]["ReceiptHandle"]) after = sqs.receive_message(QueueUrl=fq4, MaxNumberOfMessages=1).get("Messages", []) check("FIFO: after the in-flight message is deleted the next one of its group is delivered", "the next message of the group becomes available once the in-flight one is deleted", f"after deleting g1-1 a receive returned {[x['Body'] for x in after]}", [x["Body"] for x in after] == ["g1-2"]) # 5. long polling q = mkq("lp") t0 = time.time(); r = sqs.receive_message(QueueUrl=q, WaitTimeSeconds=3); dt = time.time() - t0 check("long polling waits for the wait time on an empty queue", "long polling returns an empty response only 'if the polling wait time expires'", f"WaitTimeSeconds=3 on an empty queue returned after {dt:.1f} s with {len(r.get('Messages', []))} messages", 2.5 <= dt <= 4.5) threading.Timer(1.0, lambda: sqs.send_message(QueueUrl=q, MessageBody="late")).start() t0 = time.time(); r = sqs.receive_message(QueueUrl=q, WaitTimeSeconds=10); dt = time.time() - t0 check("long polling returns as soon as a message arrives", "'Return messages as soon as they become available'", f"message sent after 1 s, receive with WaitTimeSeconds=10 returned after {dt:.1f} s with {len(r.get('Messages', []))} message", dt < 4 and len(r.get("Messages", [])) == 1) print(f"\n{sum(results)} of {len(results)} behaviour checks agree with the documentation") --- py sqs_behaviour.py (saved output of the real run) --- ok visibility timeout hides a received message, then it reappears docs: received message is invisible until the timeout expires; then another receive can get it emulator: 1st receive 1 message, immediate 2nd receive 0, after 2.4 s receive 1 (ReceiveCount 2) ok ChangeMessageVisibility to 0 makes the message available at once docs: setting VisibilityTimeout to 0 'immediately makes the message available for other consumers' emulator: receive right after: 1 message ok after maxReceiveCount receives the message moves to the dead-letter queue docs: maxReceiveCount = 'the number of times a consumer can receive a message from a source queue before it is moved to a dead-letter queue' emulator: receives that returned the message (3 allowed): [1, 1, 1, 0, 0, 0]; message in DLQ afterwards: ['poison'] ok FIFO send without MessageGroupId fails docs: 'MessageGroupId is required for FIFO queues' (the action fails) emulator: MissingParameter ok FIFO drops a resend with the same deduplication id inside the 5-minute interval docs: 'If you retry the SendMessage action within the 5-minute deduplication interval, Amazon SQS doesn't introduce any duplicates' emulator: two sends with the same id, receive returned ['a'] ok content-based deduplication hashes the body only, not attributes docs: SHA-256 hash 'using the body of the message but not the attributes of the message' emulator: same body with different attributes sent twice, receive returned 1 info one receive asking for up to 10 messages returned ['g1-1', 'g1-2', 'g2-1'] (the page I read does not say whether a single call may return several messages of one group, so this is not scored) ok FIFO: while a group's message is in flight, later messages of that group are not delivered docs: 'subsequent messages in that group are not made available until the in-flight message is either deleted or the visibility timeout expires' emulator: one message per receive, nothing deleted: [['g1-1'], ['g2-1'], []] ok FIFO: after the in-flight message is deleted the next one of its group is delivered docs: the next message of the group becomes available once the in-flight one is deleted emulator: after deleting g1-1 a receive returned ['g1-2'] ok long polling waits for the wait time on an empty queue docs: long polling returns an empty response only 'if the polling wait time expires' emulator: WaitTimeSeconds=3 on an empty queue returned after 3.0 s with 0 messages ok long polling returns as soon as a message arrives docs: 'Return messages as soon as they become available' emulator: message sent after 1 s, receive with WaitTimeSeconds=10 returned after 1.0 s with 1 message 10 of 10 behaviour checks agree with the documentation --- sqs_quotas.py --- """Does the moto emulator enforce the SQS limits the docs list (quotas-messages page, read 2026-10-13)? For each limit: one value just inside it and one just outside. Expected: inside accepted, outside rejected.""" from botocore.exceptions import ClientError from common import client, uniq sqs = client("sqs") Q = sqs.create_queue(QueueName=uniq("quota"))["QueueUrl"] def attempt(fn): try: fn() return "accepted" except ClientError as e: return "rejected (" + e.response["Error"]["Code"] + ")" except Exception as e: # client-side parameter validation in botocore return "rejected client-side (" + type(e).__name__ + ")" def mk(**attrs): return lambda: sqs.create_queue(QueueName=uniq("qq"), Attributes={k: str(v) for k, v in attrs.items()}) cases = [ # (limit as documented, inside label, inside call, outside label, outside call) ("message size max 1,048,576 bytes", "1,048,576 bytes", lambda: sqs.send_message(QueueUrl=Q, MessageBody="a" * 1048576), "1,048,577 bytes", lambda: sqs.send_message(QueueUrl=Q, MessageBody="a" * 1048577)), ("visibility timeout max 12 hours (43,200 s)", "43200", mk(VisibilityTimeout=43200), "43201", mk(VisibilityTimeout=43201)), ("retention min 60 s", "60", mk(MessageRetentionPeriod=60), "59", mk(MessageRetentionPeriod=59)), ("retention max 1,209,600 s (14 days)", "1209600", mk(MessageRetentionPeriod=1209600), "1209601", mk(MessageRetentionPeriod=1209601)), ("message timer (delay) max 15 minutes", "900", lambda: sqs.send_message(QueueUrl=Q, MessageBody="x", DelaySeconds=900), "901", lambda: sqs.send_message(QueueUrl=Q, MessageBody="x", DelaySeconds=901)), ("long polling wait max 20 s", "20 (empty queue, so this waits)", lambda: sqs.receive_message(QueueUrl=sqs.create_queue(QueueName=uniq("w"))["QueueUrl"], WaitTimeSeconds=1), "21", lambda: sqs.receive_message(QueueUrl=Q, WaitTimeSeconds=21)), ("batch max 10 messages", "10 entries", lambda: sqs.send_message_batch(QueueUrl=Q, Entries=[{"Id": f"i{i}", "MessageBody": "x"} for i in range(10)]), "11 entries", lambda: sqs.send_message_batch(QueueUrl=Q, Entries=[{"Id": f"i{i}", "MessageBody": "x"} for i in range(11)])), ("up to 10 message attributes", "10 attributes", lambda: sqs.send_message(QueueUrl=Q, MessageBody="x", MessageAttributes={f"a{i}": {"DataType": "String", "StringValue": "v"} for i in range(10)}), "11 attributes", lambda: sqs.send_message(QueueUrl=Q, MessageBody="x", MessageAttributes={f"a{i}": {"DataType": "String", "StringValue": "v"} for i in range(11)})), ("FIFO MessageGroupId max 128 characters", "128", lambda: sqs.send_message(QueueUrl=FQ, MessageBody="x", MessageGroupId="g" * 128, MessageDeduplicationId="d1"), "129", lambda: sqs.send_message(QueueUrl=FQ, MessageBody="x", MessageGroupId="g" * 129, MessageDeduplicationId="d2")), ("batched message ID max 80 characters, [A-Za-z0-9_-]", "80", lambda: sqs.send_message_batch(QueueUrl=Q, Entries=[{"Id": "i" * 80, "MessageBody": "x"}]), "81", lambda: sqs.send_message_batch(QueueUrl=Q, Entries=[{"Id": "i" * 81, "MessageBody": "x"}])), ("batched message ID only alphanumerics, - and _", "'a-b_1'", lambda: sqs.send_message_batch(QueueUrl=Q, Entries=[{"Id": "a-b_1", "MessageBody": "x"}]), "'a b!'", lambda: sqs.send_message_batch(QueueUrl=Q, Entries=[{"Id": "a b!", "MessageBody": "x"}])), ("message content: control character #x1 not allowed", "'plain text'", lambda: sqs.send_message(QueueUrl=Q, MessageBody="plain text"), "text with U+0001", lambda: sqs.send_message(QueueUrl=Q, MessageBody="bad\u0001char")), ] FQ = sqs.create_queue(QueueName=uniq("fq") + ".fifo", Attributes={"FifoQueue": "true"})["QueueUrl"] print(f"{'limit (docs)':56s} {'inside':34s} {'outside'}") ok = 0 for limit, il, ic, ol, oc in cases: a, b = attempt(ic), attempt(oc) good = a == "accepted" and b.startswith("rejected") ok += good print(f"{limit:56s} {il[:14] + ' ' + a:34s} {ol[:14]} {b}" + ("" if good else " <-- differs from the docs")) print(f"\n{ok} of {len(cases)} limits enforced exactly as documented") --- py sqs_quotas.py (saved output of the real run) --- limit (docs) inside outside message size max 1,048,576 bytes 1,048,576 byte accepted 1,048,577 byte rejected (InvalidParameterValue) visibility timeout max 12 hours (43,200 s) 43200 accepted 43201 accepted <-- differs from the docs retention min 60 s 60 accepted 59 accepted <-- differs from the docs retention max 1,209,600 s (14 days) 1209600 accepted 1209601 accepted <-- differs from the docs message timer (delay) max 15 minutes 900 accepted 901 rejected (InvalidParameterValue) long polling wait max 20 s 20 (empty queu accepted 21 rejected (InvalidParameterValue) batch max 10 messages 10 entries accepted 11 entries rejected (AWS.SimpleQueueService.TooManyEntriesInBatchRequest) up to 10 message attributes 10 attributes accepted 11 attributes accepted <-- differs from the docs FIFO MessageGroupId max 128 characters 128 accepted 129 accepted <-- differs from the docs batched message ID max 80 characters, [A-Za-z0-9_-] 80 accepted 81 rejected (AWS.SimpleQueueService.InvalidBatchEntryId) batched message ID only alphanumerics, - and _ 'a-b_1' accepted 'a b!' rejected (AWS.SimpleQueueService.InvalidBatchEntryId) message content: control character #x1 not allowed 'plain text' accepted text with U+00 accepted <-- differs from the docs 6 of 12 limits enforced exactly as documented --- consumer_demo.py --- """At-least-once delivery, shown on the moto emulator: a consumer that is slower than the queue's visibility timeout sees the same message twice. Then the same consumer made idempotent with a processed-ID set (an in-memory stand-in for a conditional write to DynamoDB, which is what you would use for real).""" import json, threading, time from common import client, uniq sqs = client("sqs") def make_queue(): return sqs.create_queue(QueueName=uniq("orders"), Attributes={"VisibilityTimeout": "2"})["QueueUrl"] def worker(q, name, process_seconds, handler, stop, log): while not stop.is_set(): r = sqs.receive_message(QueueUrl=q, MaxNumberOfMessages=1, WaitTimeSeconds=1, AttributeNames=["ApproximateReceiveCount"]).get("Messages", []) for m in r: body = json.loads(m["Body"]) done = handler(name, body, process_seconds, m) log.append((round(time.time() - T0, 1), name, body["order"], "receive#" + m["Attributes"]["ApproximateReceiveCount"], done)) try: sqs.delete_message(QueueUrl=q, ReceiptHandle=m["ReceiptHandle"]) except Exception as e: log.append((round(time.time() - T0, 1), name, body["order"], "delete failed", type(e).__name__)) EFFECTS = [] def naive(name, body, secs, m): time.sleep(secs) EFFECTS.append(body["order"]) # the side effect (charge the card) return "charged" PROCESSED, LOCK = set(), threading.Lock() def idempotent(name, body, secs, m): with LOCK: if body["order"] in PROCESSED: return "skipped (already processed)" PROCESSED.add(body["order"]) time.sleep(secs) EFFECTS.append(body["order"]) return "charged" def scenario(title, handler): global T0 EFFECTS.clear(); PROCESSED.clear() q = make_queue(); log = []; stop = threading.Event() sqs.send_message(QueueUrl=q, MessageBody=json.dumps({"order": "A-1001"})) T0 = time.time() ts = [threading.Thread(target=worker, args=(q, n, 3.5, handler, stop, log)) for n in ("worker-1", "worker-2")] for t in ts: t.start() time.sleep(9) stop.set() for t in ts: t.join() print(f"-- {title} (visibility timeout 2 s, processing takes 3.5 s, two workers)") for row in sorted(log): print(" t=%4.1fs %-9s order %s %-13s %s" % row) print(f" side effects performed: {len(EFFECTS)} -> {EFFECTS}") return EFFECTS[:] a = scenario("naive consumer", naive) b = scenario("idempotent consumer", idempotent) print(f"\nnaive consumer charged order A-1001 {len(a)} times; idempotent consumer {len(b)} time(s)") --- py consumer_demo.py (saved output of the real run) --- -- naive consumer (visibility timeout 2 s, processing takes 3.5 s, two workers) t= 3.5s worker-1 order A-1001 receive#1 charged t= 5.6s worker-2 order A-1001 receive#2 charged side effects performed: 2 -> ['A-1001', 'A-1001'] -- idempotent consumer (visibility timeout 2 s, processing takes 3.5 s, two workers) t= 2.1s worker-2 order A-1001 receive#2 skipped (already processed) t= 3.5s worker-1 order A-1001 receive#1 charged side effects performed: 1 -> ['A-1001'] naive consumer charged order A-1001 2 times; idempotent consumer 1 time(s) -------------------------- Reading the result: moto agreed with the documentation on 10 of 10 SQS behaviour checks but enforced only 6 of 12 documented limits (it accepted a visibility timeout of 43,201 s, retention of 59 and 1,209,601 s, 11 message attributes, a 129-character group ID and a control character in the body). Mistakes made and fixed: (1) my first FIFO ordering check asked for up to 10 messages in one call and the emulator returned the whole group at once; the page I read does not say whether the real service does that, so I kept it as an unscored information line and added a one-message-at-a-time check. (2) My first consumer demo did not request the ApproximateReceiveCount attribute, so each worker crashed after the side effect and before the delete, and the first numbers were invalid; I fixed the request and re-ran. How to avoid both: state what the page says before writing the check, and run a demo once with the output visible before trusting its totals.