Trade-offs in AWS SQS

brew tap tensor9ine/tensor9
brew install tensor9
tensor9 --version

tensor9 explain \
  -origin aws::1.0.0::sqs::classic \
  -target google::1.0.0::pubsub \
  -fmt Human

tensor9 adapt svc run \
  --origin aws::1.0.0::sqs::classic \
  --backend google::1.0.0::pubsub

export AWS_ENDPOINT_URL_SQS="<endpoint printed by tensor9>"
export AWS_PROFILE="adapter-test"
export AWS_REGION="us-east-1"
aws sqs create-queue \
  --queue-name tradeoffs-jobs \
  --attributes VisibilityTimeout=10,MessageRetentionPeriod=600 \
  --tags 't9:backend=google::1.0.0::pubsub' \
  --query QueueUrl --output text

aws sqs get-queue-attributes \
  --queue-url "<QueueUrl returned by create-queue>" \
  --attribute-names VisibilityTimeout MessageRetentionPeriod

python3 -m venv .venv-sqs-tradeoffs
source .venv-sqs-tradeoffs/bin/activate
python -m pip install boto3

import json
import os
import sqlite3
import time
import uuid
import boto3

sqs = boto3.client(
    "sqs",
    endpoint_url="<endpoint printed by tensor9>",
    region_name=os.environ["AWS_REGION"],
)
queue = sqs.get_queue_url(QueueName="tradeoffs-jobs")["QueueUrl"]
job_id = str(uuid.uuid4())
sqs.send_message(QueueUrl=queue, MessageBody=json.dumps({"jobId": job_id}))

def receive_job():
    deadline = time.monotonic() + 90
    while time.monotonic() < deadline:
        result = sqs.receive_message(
            QueueUrl=queue, MaxNumberOfMessages=1,
            WaitTimeSeconds=10, VisibilityTimeout=10,
        )
        for message in result.get("Messages", []):
            if json.loads(message["Body"])["jobId"] != job_id:
                raise RuntimeError("Unexpected job; inspect this disposable queue")
            return message
    raise TimeoutError("No delivery observed within the trial window")

first = receive_job()
# Local stand-in for one transactional application effect.
ledger = sqlite3.connect(":memory:")
ledger.execute("CREATE TABLE completed (job_id TEXT PRIMARY KEY)")

def handle(message):
    with ledger:
        ledger.execute(
            "INSERT OR IGNORE INTO completed VALUES (?)",
            (json.loads(message["Body"])["jobId"],),
        )

handle(first)
# Simulate processing completed, but DeleteMessage never happened.
time.sleep(12)
second = receive_job()
handle(second)
assert ledger.execute("SELECT COUNT(*) FROM completed").fetchone()[0] == 1
sqs.delete_message(QueueUrl=queue, ReceiptHandle=second["ReceiptHandle"])
ledger.close()
print("Observed redelivery; two handler calls recorded one local effect.")

python sqs-redelivery.py

aws sqs delete-queue --queue-url "<QueueUrl returned by create-queue>"
unset AWS_ENDPOINT_URL_SQS
deactivate