AWS,
service by service.

Snippets below in Go and Python — switch with the control in the bar, or press g / p.

// config.go — built once at start-up, shared by every client in the process.
cfg, err := config.LoadDefaultConfig(ctx,
    config.WithRegion("eu-west-1"),
    config.WithRetryer(func() aws.Retryer {
        // Standard: 3 attempts, exponential backoff with full jitter, and a
        // token bucket so a storm of retries cannot amplify an outage.
        // retry.NewAdaptiveMode wraps it with a client-side rate limiter that
        // slows you down when the service starts throttling.
        return retry.NewStandard(func(o *retry.StandardOptions) {
            o.MaxAttempts = 5
            o.MaxBackoff = 3 * time.Second
        })
    }),
    config.WithHTTPClient(awshttp.NewBuildableClient().
        WithTimeout(10*time.Second).                // PER ATTEMPT, not per call
        WithTransportOptions(func(t *http.Transport) {
            t.MaxIdleConnsPerHost = 100             // the SDK client defaults to 10
        })),
)
if err != nil {
    return fmt.Errorf("aws config: %w", err)        // fail the process, not the request
}

// One client per service. They are goroutine-safe and hold the pool.
app := &App{
    ddb: dynamodb.NewFromConfig(cfg),
    s3:  s3.NewFromConfig(cfg, func(o *s3.Options) { o.UsePathStyle = false }),
    sqs: sqs.NewFromConfig(cfg),
}

# aws.py — imported once; module-level clients live for the whole process.
import boto3
from botocore.config import Config

CFG = Config(
    region_name="eu-west-1",
    retries={"max_attempts": 5, "mode": "adaptive"},  # legacy | standard | adaptive
    connect_timeout=2,             # PER ATTEMPT
    read_timeout=10,               # PER ATTEMPT — total wall time can be 5x this
    max_pool_connections=50,       # default 10, shared by every thread
    tcp_keepalive=True,
    user_agent_appid="orders",     # lands in CloudTrail; worth its weight in triage
)

_SESSION = boto3.session.Session()
DDB = _SESSION.client("dynamodb", config=CFG)
S3 = _SESSION.client("s3", config=CFG)
SQS = _SESSION.client("sqs", config=CFG)

# Clients are thread-safe. Resources (boto3.resource) are NOT — if you use
# them, build one per thread, or stay on clients as everything here does.

Errors: the code matters more than the message

Every AWS error carries a code, an HTTP status and a request ID. Branch on the code — never on the message text, which AWS reserves the right to reword. The request ID is what support asks for, so log it.

// Typed errors: the modelled ones have Go types, the rest satisfy smithy.APIError.
_, err := app.ddb.PutItem(ctx, &dynamodb.PutItemInput{
    TableName:           aws.String("orders"),
    Item:                item,
    ConditionExpression: aws.String("attribute_not_exists(pk)"),
})

var dup *types.ConditionalCheckFailedException
var thr *types.ProvisionedThroughputExceededException
var api smithy.APIError

switch {
case errors.As(err, &dup):
    return ErrAlreadyExists                       // not an error: someone won the race
case errors.As(err, &thr):
    return backoffAndRetryLater(ctx)              // the SDK already tried; the table is hot
case errors.As(err, &api):
    // RequestID is on the response metadata, not on the error itself
    var re *awshttp.ResponseError
    id := ""
    if errors.As(err, &re) {
        id = re.ServiceRequestID()
    }
    log.Error("aws call failed", "code", api.ErrorCode(),
        "fault", api.ErrorFault().String(),       // client | server | unknown
        "request_id", id)
    return err
case err != nil:
    return err                                    // context cancelled, DNS, TLS…
}

# botocore raises ClientError for anything the service returned; the shape is
# always the same, and every client exposes its modelled errors as attributes.
from botocore.exceptions import ClientError, EndpointConnectionError

try:
    DDB.put_item(
        TableName="orders",
        Item=item,
        ConditionExpression="attribute_not_exists(pk)",
    )
except DDB.exceptions.ConditionalCheckFailedException:
    raise AlreadyExists                    # not an error: someone won the race
except DDB.exceptions.ProvisionedThroughputExceededException:
    raise Retryable("table is hot")        # the SDK already retried; still throttled
except ClientError as e:
    err = e.response["Error"]
    meta = e.response["ResponseMetadata"]
    log.error(
        "aws call failed",
        extra={
            "code": err["Code"],                          # branch on THIS
            "status": meta["HTTPStatusCode"],
            "request_id": meta["RequestId"],              # what support will ask for
            "attempts": meta.get("RetryAttempts"),        # how many the SDK burned
        },
    )
    raise
except EndpointConnectionError:            # never reached the service at all:
    raise Unreachable                      # DNS, route, security group, endpoint

Pagination and waiters, without the loop you always get wrong

Every list-shaped AWS API is paginated, and the page you get back is capped by size as well as count — a Query can return zero items and still hand you a continuation token because it scanned 1 MB of non-matching rows. Never treat an empty page as the end. Both SDKs ship paginators that get this right; use them.

p := s3.NewListObjectsV2Paginator(app.s3, &s3.ListObjectsV2Input{
    Bucket: aws.String("orders-archive"),
    Prefix: aws.String("2026/08/"),
})
for p.HasMorePages() {
    page, err := p.NextPage(ctx)          // ctx cancels mid-listing
    if err != nil {
        return err
    }
    for _, o := range page.Contents {
        handle(aws.ToString(o.Key), aws.ToInt64(o.Size))
    }
}

// Waiters poll for you, with the service’s own recommended interval —
// and, crucially, with a maximum wait, so a stuck resource fails the deploy.
w := rds.NewDBInstanceAvailableWaiter(app.rds)
err = w.Wait(ctx, &rds.DescribeDBInstancesInput{
    DBInstanceIdentifier: aws.String("orders-prod"),
}, 20*time.Minute)

for page in S3.get_paginator("list_objects_v2").paginate(
    Bucket="orders-archive",
    Prefix="2026/08/",
    PaginationConfig={"PageSize": 1000},          # per request, not a total cap
):
    for obj in page.get("Contents", []):          # .get: an empty page has no key
        handle(obj["Key"], obj["Size"])

# Server-side filtering beats client-side: JMESPath runs in botocore, but
# search() still downloads every page — use Prefix/StartAfter where you can.
keys = S3.get_paginator("list_objects_v2").paginate(
    Bucket="orders-archive", Prefix="2026/08/"
).search("Contents[?Size > `0`].Key")

# Waiters: the service’s own polling loop, with a bounded maximum.
RDS.get_waiter("db_instance_available").wait(
    DBInstanceIdentifier="orders-prod",
    WaiterConfig={"Delay": 30, "MaxAttempts": 40},   # 20 minutes, then raise
)

Assuming a role from code — and refreshing it

The one detail people miss: AssumeRole returns credentials with a fixed expiry. If your process outlives it and you did not wrap the call in something that refreshes, you get a clean, confusing ExpiredToken an hour after deploy.

// Cross-account: assume the data account’s reader role. NewCredentialsCache
// is what makes this safe for a long-lived process — it refreshes shortly
// before expiry and serialises concurrent refreshes into one STS call.
base, err := config.LoadDefaultConfig(ctx, config.WithRegion("eu-west-1"))
if err != nil {
    return err
}

prov := stscreds.NewAssumeRoleProvider(sts.NewFromConfig(base),
    "arn:aws:iam::222233334444:role/orders-reader",
    func(o *stscreds.AssumeRoleOptions) {
        o.RoleSessionName = "orders-" + hostname       // shows up in CloudTrail
        o.Duration = 30 * time.Minute                  // short: blast radius of a leak
        o.ExternalID = aws.String(externalID)          // third-party: confused deputy
        o.Tags = []types.Tag{                          // session tags can drive ABAC
            {Key: aws.String("service"), Value: aws.String("orders")},
        }
    })

cross := base.Copy()
cross.Credentials = aws.NewCredentialsCache(prov)
reader := dynamodb.NewFromConfig(cross)

// Who am I, really? The first line of any IAM debugging session.
who, err := sts.NewFromConfig(cross).GetCallerIdentity(ctx, nil)
log.Info("identity", "arn", aws.ToString(who.Arn))     // .../orders-reader/orders-h7

# boto3’s assume_role returns FIXED credentials — fine for a script, a trap
# for a server. For anything long-lived, hand botocore a refreshable provider
# so it re-assumes before expiry instead of dying with ExpiredToken.
from botocore.credentials import DeferredRefreshableCredentials
from botocore.session import get_session

STS = boto3.client("sts")
ROLE = "arn:aws:iam::222233334444:role/orders-reader"


def _fetch():
    r = STS.assume_role(
        RoleArn=ROLE,
        RoleSessionName=f"orders-{HOSTNAME}",     # shows up in CloudTrail
        DurationSeconds=1800,                     # short: blast radius of a leak
        ExternalId=EXTERNAL_ID,                   # third-party: confused deputy
        Tags=[{"Key": "service", "Value": "orders"}],   # session tags drive ABAC
    )["Credentials"]
    return {
        "access_key": r["AccessKeyId"],
        "secret_key": r["SecretAccessKey"],
        "token": r["SessionToken"],
        "expiry_time": r["Expiration"].isoformat(),
    }


_bs = get_session()
_bs._credentials = DeferredRefreshableCredentials(_fetch, "assume-role")
CROSS = boto3.Session(botocore_session=_bs)
reader = CROSS.client("dynamodb", config=CFG)

# Who am I, really? First line of any IAM debugging session.
print(CROSS.client("sts").get_caller_identity()["Arn"])   # .../orders-reader/orders-h7

The policy you should actually write

A policy with "Resource": "*" is a policy you will never be able to audit. Scope by resource ARN, then narrow further with conditions — the two below are the highest-value conditions on the platform, and neither costs you anything.

// The same JSON, whichever language calls it. Two things make it real:
// a resource ARN with a prefix, and conditions that survive a leaked key.
policy := `{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ReadOwnTenantObjects",
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::orders-archive/tenant/${aws:PrincipalTag/tenant}/*",
      "Condition": {
        "Bool":        {"aws:SecureTransport": "true"},
        "StringEquals":{"aws:SourceVpce": "vpce-0abc123"}
      }
    },
    {
      "Sid": "DenyEverythingOutsideTheOrg",
      "Effect": "Deny",
      "Action": "*",
      "Resource": "*",
      "Condition": {"StringNotEquals": {"aws:PrincipalOrgID": "o-exampleorgid"}}
    }
  ]
}`

// Pass it as a session policy to shrink a broad role for one job.
o.Policy = aws.String(policy)   // inside AssumeRoleOptions: intersects, never grants

# The same JSON, whichever language calls it. Two things make it real:
# a resource ARN with a prefix, and conditions that survive a leaked key.
POLICY = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ReadOwnTenantObjects",
            "Effect": "Allow",
            "Action": ["s3:GetObject"],
            "Resource": "arn:aws:s3:::orders-archive/tenant/${aws:PrincipalTag/tenant}/*",
            "Condition": {
                "Bool": {"aws:SecureTransport": "true"},
                "StringEquals": {"aws:SourceVpce": "vpce-0abc123"},
            },
        },
        {
            "Sid": "DenyEverythingOutsideTheOrg",
            "Effect": "Deny",
            "Action": "*",
            "Resource": "*",
            "Condition": {"StringNotEquals": {"aws:PrincipalOrgID": "o-exampleorgid"}},
        },
    ],
}

# Pass it as a session policy to shrink a broad role for one job.
STS.assume_role(RoleArn=ROLE, RoleSessionName="job",
                Policy=json.dumps(POLICY))    # intersects, never grants

Make the network failure legible at boot

The worst version of this bug is a service that starts healthy, passes its health check, and only fails when the first real request needs S3. A ten-line probe at start-up turns a 3 a.m. mystery into a log line.

// reachable dials the service endpoint and separates the failure modes:
// name resolution (DNS / private DNS) from the TCP handshake (route / SG).
func reachable(ctx context.Context, host string) error {
    ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
    defer cancel()

    ips, err := net.DefaultResolver.LookupHost(ctx, host)
    if err != nil {
        // private DNS off on the interface endpoint, or DNS support off on the VPC
        return fmt.Errorf("dns %s: %w", host, err)
    }

    var d net.Dialer
    conn, err := d.DialContext(ctx, "tcp", net.JoinHostPort(host, "443"))
    if err != nil {
        // resolved but unreachable: no route, no egress rule, or no endpoint
        return fmt.Errorf("dial %s (%v): %w", host, ips[0], err)
    }
    _ = conn.Close()
    return nil
}

// At start-up, before the health check goes green:
for _, h := range []string{
    "sts.eu-west-1.amazonaws.com",              // credentials
    "dynamodb.eu-west-1.amazonaws.com",         // gateway endpoint or NAT
    "secretsmanager.eu-west-1.amazonaws.com",   // interface endpoint only
} {
    if err := reachable(ctx, h); err != nil {
        return fmt.Errorf("startup connectivity: %w", err)   // crash loudly, now
    }
}

# reachable() dials the service endpoint and separates the failure modes:
# name resolution (DNS / private DNS) from the TCP handshake (route / SG).
import socket


def reachable(host: str, port: int = 443, timeout: float = 3.0) -> None:
    try:
        infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
    except socket.gaierror as e:
        # private DNS off on the interface endpoint, or DNS support off on the VPC
        raise Unreachable(f"dns {host}: {e}") from e

    family, kind, proto, _, addr = infos[0]
    with socket.socket(family, kind, proto) as s:
        s.settimeout(timeout)
        try:
            s.connect(addr)
        except OSError as e:
            # resolved but unreachable: no route, no egress rule, or no endpoint
            raise Unreachable(f"dial {host} ({addr[0]}): {e}") from e


# At start-up, before the health check goes green:
for host in (
    "sts.eu-west-1.amazonaws.com",              # credentials
    "dynamodb.eu-west-1.amazonaws.com",         # gateway endpoint or NAT
    "secretsmanager.eu-west-1.amazonaws.com",   # interface endpoint only
):
    reachable(host)                             # crash loudly, now, not at 03:00

The parameters worth setting on day one

A pre-deploy guard for the database

Three RDS facts should block a release: pending maintenance that will reboot you mid-deploy, a certificate authority about to expire, and storage close to the autoscaling ceiling. All three are one API call each.

// preflight fails the pipeline if the database is about to move under us.
func preflight(ctx context.Context, c *rds.Client, id string) error {
    d, err := c.DescribeDBInstances(ctx, &rds.DescribeDBInstancesInput{
        DBInstanceIdentifier: aws.String(id),
    })
    if err != nil {
        return err
    }
    in := d.DBInstances[0]

    // 1. a modification already queued for the next window (class, storage, params)
    if p := in.PendingModifiedValues; p != nil && p.DBInstanceClass != nil {
        return fmt.Errorf("pending class change to %s", aws.ToString(p.DBInstanceClass))
    }

    // 2. static parameters saved but not yet live — the reboot is still owed
    for _, g := range in.DBParameterGroups {
        if aws.ToString(g.ParameterApplyStatus) == "pending-reboot" {
            return fmt.Errorf("parameter group %s awaits a reboot",
                aws.ToString(g.DBParameterGroupName))
        }
    }

    // 3. maintenance AWS will apply for us, whether or not we are deploying
    m, err := c.DescribePendingMaintenanceActions(ctx,
        &rds.DescribePendingMaintenanceActionsInput{
            ResourceIdentifier: in.DBInstanceArn,
        })
    if err != nil {
        return err
    }
    for _, r := range m.PendingMaintenanceActions {
        for _, a := range r.PendingMaintenanceActionDetails {
            log.Warn("pending maintenance",
                "action", aws.ToString(a.Action),                 // system-update, ca-certificate-rotation…
                "auto_applied_after", a.AutoAppliedAfterDate)
        }
    }
    log.Info("db ok", "ca", aws.ToString(in.CACertificateIdentifier),
        "engine", aws.ToString(in.EngineVersion),
        "multi_az", aws.ToBool(in.MultiAZ))
    return nil
}

# preflight() fails the pipeline if the database is about to move under us.
RDS = boto3.client("rds", config=CFG)


def preflight(db_id: str) -> None:
    inst = RDS.describe_db_instances(DBInstanceIdentifier=db_id)["DBInstances"][0]

    # 1. a modification already queued for the next window (class, storage, params)
    pending = inst.get("PendingModifiedValues") or {}
    if "DBInstanceClass" in pending:
        raise Blocked(f"pending class change to {pending['DBInstanceClass']}")

    # 2. static parameters saved but not yet live — the reboot is still owed
    for group in inst["DBParameterGroups"]:
        if group["ParameterApplyStatus"] == "pending-reboot":
            raise Blocked(f"{group['DBParameterGroupName']} awaits a reboot")

    # 3. maintenance AWS will apply for us, whether or not we are deploying
    actions = RDS.describe_pending_maintenance_actions(
        ResourceIdentifier=inst["DBInstanceArn"]
    )["PendingMaintenanceActions"]
    for res in actions:
        for a in res["PendingMaintenanceActionDetails"]:
            log.warning(
                "pending maintenance",
                extra={"action": a["Action"],            # ca-certificate-rotation…
                       "after": a.get("AutoAppliedAfterDate")},
            )

    log.info("db ok", extra={"ca": inst["CACertificateIdentifier"],
                             "engine": inst["EngineVersion"],
                             "multi_az": inst["MultiAZ"]})

The recovery you should have rehearsed

“Someone dropped a table at 14:07” is not a backup restore — it is a point-in-time restore into a new instance, from which you copy the one table back. Practise it once and the real thing is forty minutes instead of a day.

// Restore to one second before the mistake. This never touches production:
// it creates a second instance you then read from.
target := "orders-prod-recovery-1407"
_, err := c.RestoreDBInstanceToPointInTime(ctx, &rds.RestoreDBInstanceToPointInTimeInput{
    SourceDBInstanceIdentifier: aws.String("orders-prod"),
    TargetDBInstanceIdentifier: aws.String(target),
    RestoreTime:                aws.Time(dropTime.Add(-1 * time.Second)),
    DBSubnetGroupName:          aws.String("private"),        // same VPC, no public IP
    VpcSecurityGroupIds:        []string{"sg-db"},
    DBInstanceClass:            aws.String("db.r7g.xlarge"),  // size for the copy job
    MultiAZ:                    aws.Bool(false),              // it is temporary
    DeletionProtection:         aws.Bool(false),
})
if err != nil {
    return err
}

// Restores take tens of minutes on a large volume; the waiter has a cap so a
// stuck restore fails the runbook instead of hanging it.
err = rds.NewDBInstanceAvailableWaiter(c).Wait(ctx,
    &rds.DescribeDBInstancesInput{DBInstanceIdentifier: aws.String(target)},
    45*time.Minute)

# Restore to one second before the mistake. This never touches production:
# it creates a second instance you then read from.
target = "orders-prod-recovery-1407"
RDS.restore_db_instance_to_point_in_time(
    SourceDBInstanceIdentifier="orders-prod",
    TargetDBInstanceIdentifier=target,
    RestoreTime=drop_time - timedelta(seconds=1),
    DBSubnetGroupName="private",          # same VPC, no public IP
    VpcSecurityGroupIds=["sg-db"],
    DBInstanceClass="db.r7g.xlarge",      # size for the copy job, not for prod
    MultiAZ=False,                        # it is temporary
    DeletionProtection=False,
)

# Restores take tens of minutes on a large volume; the waiter has a cap so a
# stuck restore fails the runbook instead of hanging it.
RDS.get_waiter("db_instance_available").wait(
    DBInstanceIdentifier=target,
    WaiterConfig={"Delay": 30, "MaxAttempts": 90},    # 45 minutes
)

The Data API: SQL over HTTPS, no pool at all

Aurora — provisioned and Serverless v2 alike — exposes an HTTP endpoint that runs statements with IAM auth and credentials from Secrets Manager. There is no connection to manage, which makes it the right answer for Lambda — and the wrong answer for a chatty request path, because every statement is a signed HTTPS round trip.

// Data API: no driver, no pool, no VPC attachment. Each call is signed HTTPS.
rd := rdsdata.NewFromConfig(cfg)

const clusterARN = "arn:aws:rds:eu-west-1:1234:cluster:orders"
const secretARN = "arn:aws:secretsmanager:eu-west-1:1234:secret:orders-db"

out, err := rd.ExecuteStatement(ctx, &rdsdata.ExecuteStatementInput{
    ResourceArn: aws.String(clusterARN),
    SecretArn:   aws.String(secretARN),
    Database:    aws.String("orders"),
    Sql:         aws.String(`INSERT INTO orders (id, email, total_cents)
                             VALUES (:id, :email, :total)
                             ON CONFLICT (id) DO NOTHING
                             RETURNING created_at`),
    Parameters: []types.SqlParameter{                     // never string-concatenate SQL
        {Name: aws.String("id"), Value: &types.FieldMemberStringValue{Value: id}},
        {Name: aws.String("email"), Value: &types.FieldMemberStringValue{Value: email}},
        {Name: aws.String("total"), Value: &types.FieldMemberLongValue{Value: cents}},
    },
    FormatRecordsAs: types.RecordsFormatTypeJson,          // JSON rows, not column soup
})
if err != nil {
    return err
}
if out.NumberOfRecordsUpdated == 0 {
    return ErrAlreadyExists                                // the ON CONFLICT branch
}

// Multi-statement work needs an explicit transaction: BeginTransaction returns
// an id you pass to each call, then Commit or Rollback. An abandoned
// transaction is rolled back after about three minutes idle — do not rely on it.
tx, err := rd.BeginTransaction(ctx, &rdsdata.BeginTransactionInput{
    ResourceArn: aws.String(clusterARN), SecretArn: aws.String(secretARN),
    Database: aws.String("orders"),
})

# Data API: no driver, no pool, no VPC attachment. Each call is signed HTTPS.
DATA = boto3.client("rds-data", config=CFG)

CLUSTER = "arn:aws:rds:eu-west-1:1234:cluster:orders"
SECRET = "arn:aws:secretsmanager:eu-west-1:1234:secret:orders-db"

res = DATA.execute_statement(
    resourceArn=CLUSTER,
    secretArn=SECRET,
    database="orders",
    sql="""INSERT INTO orders (id, email, total_cents)
           VALUES (:id, :email, :total)
           ON CONFLICT (id) DO NOTHING
           RETURNING created_at""",
    parameters=[                                   # never string-format SQL
        {"name": "id", "value": {"stringValue": order_id}},
        {"name": "email", "value": {"stringValue": email}},
        {"name": "total", "value": {"longValue": cents}},
    ],
    formatRecordsAs="JSON",                        # rows as JSON, not column soup
)
if res["numberOfRecordsUpdated"] == 0:
    raise AlreadyExists                            # the ON CONFLICT branch

# Multi-statement work needs an explicit transaction: begin_transaction returns
# an id you pass to each call, then commit or rollback. An abandoned
# transaction is rolled back after about three minutes idle — do not rely on it.
tx = DATA.begin_transaction(resourceArn=CLUSTER, secretArn=SECRET, database="orders")
try:
    DATA.execute_statement(resourceArn=CLUSTER, secretArn=SECRET,
                           transactionId=tx["transactionId"], sql="…")
    DATA.commit_transaction(resourceArn=CLUSTER, secretArn=SECRET,
                            transactionId=tx["transactionId"])
except Exception:
    DATA.rollback_transaction(resourceArn=CLUSTER, secretArn=SECRET,
                              transactionId=tx["transactionId"])
    raise

Rehearse the failover, in a script

A failover you have never triggered deliberately is a failover you do not know the cost of. Run this against staging monthly with traffic flowing, and watch your own error rate rather than the console.

// Game day: promote a specific reader and time what the application sees.
start := time.Now()
_, err := c.FailoverDBCluster(ctx, &rds.FailoverDBClusterInput{
    DBClusterIdentifier:        aws.String("orders"),
    TargetDBInstanceIdentifier: aws.String("orders-reader-1"),   // pick, do not guess
})
if err != nil {
    return err
}

// The cluster reports "available" almost at once; what matters is when a
// write through the cluster endpoint succeeds again. Measure THAT.
for {
    if _, err := pool.Exec(ctx, "SELECT pg_is_in_recovery()"); err == nil {
        var readOnly bool
        _ = pool.QueryRow(ctx, "SELECT pg_is_in_recovery()").Scan(&readOnly)
        if !readOnly {
            log.Info("writable again", "after", time.Since(start))
            break
        }
    }
    select {
    case <-ctx.Done():
        return ctx.Err()
    case <-time.After(500 * time.Millisecond):
    }
}

# Game day: promote a specific reader and time what the application sees.
start = time.monotonic()
RDS.failover_db_cluster(
    DBClusterIdentifier="orders",
    TargetDBInstanceIdentifier="orders-reader-1",     # pick, do not guess
)

# The cluster reports "available" almost at once; what matters is when a
# write through the cluster endpoint succeeds again. Measure THAT.
while True:
    try:
        with engine.connect() as conn:
            read_only = conn.exec_driver_sql("SELECT pg_is_in_recovery()").scalar()
        if not read_only:
            log.info("writable again", extra={"after_s": time.monotonic() - start})
            break
    except OperationalError:
        pass                       # expected: the old writer is gone
    time.sleep(0.5)

// pgxpool, sized deliberately, with IAM auth and a failover-aware health check.
cfgp, err := pgxpool.ParseConfig(
    "host=orders.cluster-abc.eu-west-1.rds.amazonaws.com port=5432 " +
        "user=orders_app dbname=orders sslmode=verify-full " +
        "sslrootcert=/etc/ssl/certs/rds-global-bundle.pem")
if err != nil {
    return err
}

// Sizing: (max_connections * 0.8) / number of tasks, then leave headroom for a
// deploy running twice the tasks. Bigger is not better — it just queues later.
cfgp.MaxConns = 10
cfgp.MinConns = 2
cfgp.MaxConnLifetime = 30 * time.Minute      // forces rebalancing after a failover
cfgp.MaxConnIdleTime = 5 * time.Minute
cfgp.HealthCheckPeriod = 30 * time.Second

// IAM auth: the password is a signed, 15-minute token, generated per connection.
// No secret to rotate, and every connection is attributable in CloudTrail.
cfgp.BeforeConnect = func(ctx context.Context, c *pgx.ConnConfig) error {
    tok, err := auth.BuildAuthToken(ctx, c.Host+":5432", "eu-west-1",
        c.User, cfg.Credentials)
    if err != nil {
        return err
    }
    c.Password = tok
    return nil
}

// The four lines that turn a 30-minute outage into a 30-second one: refuse to
// hand out a connection to an instance that is no longer the writer.
cfgp.BeforeAcquire = func(ctx context.Context, c *pgx.Conn) bool {
    var inRecovery bool
    if err := c.QueryRow(ctx, "SELECT pg_is_in_recovery()").Scan(&inRecovery); err != nil {
        return false                          // broken: destroy it
    }
    return !inRecovery                        // demoted: destroy it, reconnect via DNS
}

pool, err := pgxpool.NewWithConfig(ctx, cfgp)

# SQLAlchemy, sized deliberately, with IAM auth and a failover-aware check.
from sqlalchemy import create_engine, event, text
from sqlalchemy.exc import DisconnectionError

HOST = "orders.cluster-abc.eu-west-1.rds.amazonaws.com"

engine = create_engine(
    f"postgresql+psycopg://orders_app@{HOST}:5432/orders",
    # Sizing: (max_connections * 0.8) / workers, with headroom for a deploy
    # running twice the tasks. pool_size + max_overflow is the real ceiling.
    pool_size=8,
    max_overflow=2,
    pool_timeout=3,            # fail fast; do not queue behind a saturated pool
    pool_recycle=1800,         # forces rebalancing after a failover
    pool_pre_ping=True,        # cheap SELECT 1 before handing a connection out
    connect_args={"sslmode": "verify-full",
                  "sslrootcert": "/etc/ssl/certs/rds-global-bundle.pem"},
)


@event.listens_for(engine, "do_connect")
def _iam_token(dialect, conn_rec, cargs, cparams):
    # IAM auth: the password is a signed, 15-minute token, per connection.
    # No secret to rotate, and every connection is attributable in CloudTrail.
    cparams["password"] = RDS.generate_db_auth_token(
        DBHostname=HOST, Port=5432, DBUsername="orders_app", Region="eu-west-1"
    )


@event.listens_for(engine, "checkout")
def _reject_demoted(dbapi_conn, conn_rec, conn_proxy):
    # The check that turns a 30-minute outage into a 30-second one: never hand
    # out a connection to an instance that is no longer the writer.
    with dbapi_conn.cursor() as cur:
        cur.execute("SELECT pg_is_in_recovery()")
        if cur.fetchone()[0]:
            raise DisconnectionError("demoted to reader")   # pool discards + retries

Write like the retry is guaranteed

Two idioms cover most correctness problems: a condition expression that makes a create idempotent, and a version attribute that makes an update safe against concurrent writers. Both are free — a failed condition costs capacity but never corrupts.

// Optimistic locking: succeed only if the item is still at the version we read.
// A lost race returns ConditionalCheckFailed, which is a normal outcome here.
upd := expression.Set(expression.Name("status"), expression.Value("shipped")).
    Set(expression.Name("version"), expression.Value(cur.Version+1)).
    Set(expression.Name("updated_at"), expression.Value(time.Now().UTC().Format(time.RFC3339)))

cond := expression.Name("version").Equal(expression.Value(cur.Version))

expr, err := expression.NewBuilder().WithUpdate(upd).WithCondition(cond).Build()
if err != nil {
    return err
}

_, err = app.ddb.UpdateItem(ctx, &dynamodb.UpdateItemInput{
    TableName: aws.String("orders"),
    Key: map[string]types.AttributeValue{
        "pk": &types.AttributeValueMemberS{Value: "ORDER#" + id},
        "sk": &types.AttributeValueMemberS{Value: "META"},
    },
    UpdateExpression:          expr.Update(),
    ConditionExpression:       expr.Condition(),
    ExpressionAttributeNames:  expr.Names(),      // handles reserved words for you
    ExpressionAttributeValues: expr.Values(),
    ReturnValues:              types.ReturnValueAllNew,
})

var stale *types.ConditionalCheckFailedException
if errors.As(err, &stale) {
    return ErrConcurrentUpdate                     // re-read and decide, do not blind-retry
}

// Idempotent create: the same request twice leaves one item and no error.
_, err = app.ddb.PutItem(ctx, &dynamodb.PutItemInput{
    TableName:           aws.String("orders"),
    Item:                item,
    ConditionExpression: aws.String("attribute_not_exists(pk)"),
})

# Optimistic locking: succeed only if the item is still at the version we read.
# A lost race raises ConditionalCheckFailed, which is a normal outcome here.
try:
    DDB.update_item(
        TableName="orders",
        Key={"pk": {"S": f"ORDER#{order_id}"}, "sk": {"S": "META"}},
        UpdateExpression=(
            "SET #s = :new, version = :next, updated_at = :now"
        ),
        ConditionExpression="version = :cur",
        # "status" is a reserved word — every table hits this eventually
        ExpressionAttributeNames={"#s": "status"},
        ExpressionAttributeValues={
            ":new": {"S": "shipped"},
            ":cur": {"N": str(current_version)},
            ":next": {"N": str(current_version + 1)},
            ":now": {"S": datetime.now(timezone.utc).isoformat()},
        },
        ReturnValues="ALL_NEW",
    )
except DDB.exceptions.ConditionalCheckFailedException:
    raise ConcurrentUpdate      # re-read and decide; do not blind-retry

# Idempotent create: the same request twice leaves one item and no error.
try:
    DDB.put_item(
        TableName="orders", Item=item,
        ConditionExpression="attribute_not_exists(pk)",
    )
except DDB.exceptions.ConditionalCheckFailedException:
    pass                        # already created — that is success

Read the page, not the table

A Query stops at 1 MB of scanned data and hands back a cursor. A FilterExpression runs after that limit, so filtering never saves you capacity — it only saves you bandwidth. And a batch write is not a transaction: it can partially succeed, and the items it refused come back for you to send again.

// Paginated query over one partition, newest first, only the fields we need.
kc := expression.Key("pk").Equal(expression.Value("ORDER#"+id)).
    And(expression.Key("sk").BeginsWith("EVENT#"))
expr, _ := expression.NewBuilder().
    WithKeyCondition(kc).
    WithProjection(expression.NamesList(
        expression.Name("sk"), expression.Name("kind"))).
    Build()

p := dynamodb.NewQueryPaginator(app.ddb, &dynamodb.QueryInput{
    TableName:                 aws.String("orders"),
    KeyConditionExpression:    expr.KeyCondition(),
    ProjectionExpression:      expr.Projection(),
    ExpressionAttributeNames:  expr.Names(),
    ExpressionAttributeValues: expr.Values(),
    ScanIndexForward:          aws.Bool(false),      // descending by sort key
    Limit:                     aws.Int32(100),       // items per page, not a total
})
for p.HasMorePages() {
    page, err := p.NextPage(ctx)
    if err != nil {
        return err
    }
    // An empty page with a cursor is normal: 1 MB scanned, nothing matched.
    var events []Event
    if err := attributevalue.UnmarshalListOfMaps(page.Items, &events); err != nil {
        return err
    }
    out = append(out, events...)
}

// BatchWriteItem is NOT atomic. Unprocessed items are the normal case under
// load, and dropping them silently loses data.
req := map[string][]types.WriteRequest{"orders": writes}
for attempt := 0; len(req) > 0 && attempt < 5; attempt++ {
    out, err := app.ddb.BatchWriteItem(ctx, &dynamodb.BatchWriteItemInput{
        RequestItems: req,
    })
    if err != nil {
        return err
    }
    req = out.UnprocessedItems
    if len(req) > 0 {
        time.Sleep(time.Duration(1<<attempt) * 50 * time.Millisecond)   // back off
    }
}
if len(req) > 0 {
    return fmt.Errorf("gave up with %d unwritten items", len(req))
}

# Paginated query over one partition, newest first, only the fields we need.
pages = DDB.get_paginator("query").paginate(
    TableName="orders",
    KeyConditionExpression="pk = :pk AND begins_with(sk, :prefix)",
    ExpressionAttributeValues={
        ":pk": {"S": f"ORDER#{order_id}"},
        ":prefix": {"S": "EVENT#"},
    },
    ProjectionExpression="sk, kind",
    ScanIndexForward=False,                # descending by sort key
    PaginationConfig={"PageSize": 100},    # items per request, not a total
)
events = []
for page in pages:
    # An empty page with a cursor is normal: 1 MB scanned, nothing matched.
    events.extend(page.get("Items", []))

# BatchWriteItem is NOT atomic. Unprocessed items are the normal case under
# load, and dropping them silently loses data.
pending = {"orders": writes}
for attempt in range(5):
    res = DDB.batch_write_item(RequestItems=pending)
    pending = res.get("UnprocessedItems") or {}
    if not pending:
        break
    time.sleep(0.05 * 2 ** attempt)        # back off before resending
else:
    raise IncompleteWrite(f"{len(pending.get('orders', []))} items unwritten")

Presign the upload, and never proxy the bytes

The default architecture for user uploads is wrong: bytes flow through your service, consuming memory, request time and NAT charges. A presigned URL moves them directly to S3 with a signature that constrains the method, the key and the expiry — and the conditional PUT below makes the write idempotent as well.

// A presigned PUT: the browser uploads straight to S3, we never see the bytes.
ps := s3.NewPresignClient(app.s3)

req, err := ps.PresignPutObject(ctx, &s3.PutObjectInput{
    Bucket:      aws.String("orders-uploads"),
    Key:         aws.String("tenant/" + tenantID + "/" + uuid.NewString()),
    ContentType: aws.String("application/pdf"),
    // Signed headers are constraints: a client that sends a different
    // content type or length gets a 403, not a surprise object. Note that
    // this pins an EXACT length; a MAXIMUM needs a presigned POST, below.
    ContentLength:        aws.Int64(size),
    ServerSideEncryption: types.ServerSideEncryptionAwsKms,
    SSEKMSKeyId:          aws.String(kmsKeyARN),
}, s3.WithPresignExpires(10*time.Minute))       // bounded by the caller’s own session
if err != nil {
    return err
}
return json.NewEncoder(w).Encode(map[string]string{"url": req.URL})

// Write-once semantics without a lock: If-None-Match fails if the key exists,
// so two workers racing to produce the same artefact cannot overwrite it.
_, err = app.s3.PutObject(ctx, &s3.PutObjectInput{
    Bucket:      aws.String("orders-archive"),
    Key:         aws.String(key),
    Body:        bytes.NewReader(payload),
    IfNoneMatch: aws.String("*"),
    ChecksumAlgorithm: types.ChecksumAlgorithmCrc32c,   // verified end to end
})
// S3 does not model 412 as a typed error in the Go SDK, so match the code.
var api smithy.APIError
if errors.As(err, &api) && api.ErrorCode() == "PreconditionFailed" {
    return ErrAlreadyWritten        // somebody else got there; that is fine
}

# A presigned PUT: the browser uploads straight to S3, we never see the bytes.
url = S3.generate_presigned_url(
    "put_object",
    Params={
        "Bucket": "orders-uploads",
        "Key": f"tenant/{tenant_id}/{uuid4()}",
        "ContentType": "application/pdf",
        # Signed parameters are constraints: a client that sends a different
        # content type gets a 403, not a surprise object.
        "ServerSideEncryption": "aws:kms",
        "SSEKMSKeyId": KMS_KEY_ARN,
    },
    ExpiresIn=600,           # bounded by the caller’s own session length
)

# For browser form uploads, presigned POST is better: it can enforce a
# maximum size, which a presigned PUT cannot.
post = S3.generate_presigned_post(
    Bucket="orders-uploads", Key=f"tenant/{tenant_id}/${{filename}}",
    Conditions=[["content-length-range", 1, 20 * 1024 * 1024],
                {"Content-Type": "application/pdf"}],
    ExpiresIn=600,
)

# Write-once semantics without a lock: If-None-Match fails if the key exists,
# so two workers racing to produce the same artefact cannot overwrite it.
try:
    S3.put_object(
        Bucket="orders-archive", Key=key, Body=payload,
        IfNoneMatch="*",
        ChecksumAlgorithm="CRC32C",      # verified end to end
    )
except ClientError as e:
    if e.response["Error"]["Code"] != "PreconditionFailed":
        raise
    # somebody else got there; that is fine

Big objects: let the transfer manager do it

// The uploader splits, parallelises and retries individual parts. It also
// aborts the multipart upload on failure — the one thing hand-rolled code
// forgets, and the reason the abort lifecycle rule exists.
up := manager.NewUploader(app.s3, func(u *manager.Uploader) {
    u.PartSize = 16 * 1024 * 1024      // 16 MiB parts; minimum is 5 MiB
    u.Concurrency = 8
})
_, err := up.Upload(ctx, &s3.PutObjectInput{
    Bucket:       aws.String("orders-archive"),
    Key:          aws.String(key),
    Body:         file,                // any io.Reader — no need to know the size
    StorageClass: types.StorageClassIntelligentTiering,
})

// Downloading a slice of a large object beats downloading all of it:
// a range GET is billed on bytes transferred, not on object size.
obj, err := app.s3.GetObject(ctx, &s3.GetObjectInput{
    Bucket: aws.String("orders-archive"),
    Key:    aws.String(key),
    Range:  aws.String("bytes=0-8191"),      // just the header of the file
})

# upload_fileobj splits, parallelises and retries individual parts. It also
# aborts the multipart upload on failure — the one thing hand-rolled code
# forgets, and the reason the abort lifecycle rule exists.
from boto3.s3.transfer import TransferConfig

TRANSFER = TransferConfig(
    multipart_threshold=16 * 1024 * 1024,   # switch to multipart above this
    multipart_chunksize=16 * 1024 * 1024,   # minimum part size is 5 MiB
    max_concurrency=8,
    use_threads=True,
)

with open(path, "rb") as fh:
    S3.upload_fileobj(
        fh, "orders-archive", key,
        ExtraArgs={"StorageClass": "INTELLIGENT_TIERING"},
        Config=TRANSFER,
    )

# Downloading a slice of a large object beats downloading all of it:
# a range GET is billed on bytes transferred, not on object size.
head = S3.get_object(
    Bucket="orders-archive", Key=key, Range="bytes=0-8191"
)["Body"].read()                            # just the header of the file

// Cache-aside with single flight and fail-open. golang.org/x/sync/singleflight
// collapses concurrent misses of the same key into one origin call.
type Cache struct {
    rdb *redis.Client
    sf  singleflight.Group
}

func (c *Cache) Order(ctx context.Context, id string) (*Order, error) {
    key := "order:" + id

    // A cache read gets a much tighter deadline than the request itself.
    rctx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
    defer cancel()

    if b, err := c.rdb.Get(rctx, key).Bytes(); err == nil {
        var o Order
        if json.Unmarshal(b, &o) == nil {
            return &o, nil
        }
    } else if !errors.Is(err, redis.Nil) {
        cacheErrors.Inc()          // FAIL OPEN: log it, fall through to origin
    }

    v, err, _ := c.sf.Do(key, func() (any, error) {
        o, err := loadFromDB(ctx, id)
        if err != nil {
            return nil, err
        }
        b, _ := json.Marshal(o)
        // Jitter: 10 minutes +/- 10%, so a batch cached together does not
        // expire together and rebuild the stampede you just prevented.
        ttl := 10*time.Minute + time.Duration(rand.Int63n(int64(2*time.Minute))) - time.Minute
        _ = c.rdb.Set(ctx, key, b, ttl).Err()
        return o, nil
    })
    if err != nil {
        return nil, err
    }
    return v.(*Order), nil
}

# Cache-aside with single flight and fail-open. One asyncio task per key does
# the work; everyone else awaits the same future.
import asyncio, json, random

_inflight: dict[str, asyncio.Future] = {}


async def get_order(order_id: str) -> Order:
    key = f"order:{order_id}"

    # A cache read gets a much tighter deadline than the request itself.
    try:
        raw = await asyncio.wait_for(redis.get(key), timeout=0.05)
        if raw is not None:
            return Order(**json.loads(raw))
    except (asyncio.TimeoutError, RedisError):
        CACHE_ERRORS.inc()      # FAIL OPEN: count it, fall through to origin

    # Single flight: first caller creates the future, the rest await it.
    fut = _inflight.get(key)
    if fut is not None:
        return await asyncio.shield(fut)

    fut = asyncio.get_running_loop().create_future()
    _inflight[key] = fut
    try:
        order = await load_from_db(order_id)
        # Jitter: 10 minutes +/- 10%, so a batch cached together does not
        # expire together and rebuild the stampede you just prevented.
        ttl = int(600 * random.uniform(0.9, 1.1))
        await redis.set(key, json.dumps(order.dict()), ex=ttl)
        fut.set_result(order)
        return order
    except Exception as exc:
        fut.set_exception(exc)
        raise
    finally:
        _inflight.pop(key, None)

The handler shape that gets everything right

Clients built at module scope, per-invocation deadlines derived from the remaining time, and partial batch failure reporting so one poison message does not replay the other nine.

// main.go — Go runs on provided.al2023 as a custom runtime; the binary is the
// runtime. Everything above main() happens once, during INIT, with full CPU.
var (
    cfg = mustConfig()                          // credential chain resolved once
    ddb = dynamodb.NewFromConfig(cfg)           // pool survives across invocations
    log = slog.New(slog.NewJSONHandler(os.Stdout, nil))
)

func handle(ctx context.Context, ev events.SQSEvent) (events.SQSEventResponse, error) {
    // Leave headroom: finish, flush and return before Lambda kills us.
    if dl, ok := ctx.Deadline(); ok {
        var cancel context.CancelFunc
        ctx, cancel = context.WithDeadline(ctx, dl.Add(-500*time.Millisecond))
        defer cancel()
    }

    // Partial batch response: without this, ONE bad message redelivers all ten,
    // which is how a single poison record becomes a permanent replay loop.
    var failures []events.SQSBatchItemFailure

    for _, m := range ev.Records {
        if err := process(ctx, m); err != nil {
            log.Error("record failed", "message_id", m.MessageId, "err", err)
            failures = append(failures, events.SQSBatchItemFailure{
                ItemIdentifier: m.MessageId,
            })
        }
    }
    // Return nil error: the batch itself succeeded. Only the listed ids return
    // to the queue. Requires ReportBatchItemFailures on the event source mapping.
    return events.SQSEventResponse{BatchItemFailures: failures}, nil
}

func main() { lambda.Start(handle) }

# handler.py — everything at module scope runs once, during INIT, with full
# CPU. Imports and client construction belong here, not in the handler.
import json, os, time
import boto3
from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.utilities.batch import (
    BatchProcessor, EventType, process_partial_response,
)

logger, tracer, metrics = Logger(), Tracer(), Metrics()
DDB = boto3.client("dynamodb")          # connection pool survives invocations
TABLE = os.environ["TABLE_NAME"]

processor = BatchProcessor(event_type=EventType.SQS)


@tracer.capture_method
def record_handler(record):
    body = json.loads(record.body)
    DDB.put_item(
        TableName=TABLE,
        Item={"pk": {"S": body["id"]}, "payload": {"S": record.body}},
        ConditionExpression="attribute_not_exists(pk)",   # at-least-once delivery
    )


@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics                    # flushes EMF metrics BEFORE the freeze
def handler(event, context):
    # Partial batch response: without this, ONE bad message redelivers all ten,
    # which is how a single poison record becomes a permanent replay loop.
    # Requires ReportBatchItemFailures on the event source mapping.
    return process_partial_response(
        event=event, record_handler=record_handler,
        processor=processor, context=context,
    )

Shut down properly, or drop requests forever

// The four-stage shutdown ECS expects. Nothing here is Go-specific except the
// syntax — the sequence is what matters.
srv := &http.Server{Addr: ":8080", Handler: mux}

go func() {
    if err := srv.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
        log.Error("listen", "err", err)
        os.Exit(1)
    }
}()

sig := make(chan os.Signal, 1)
signal.Notify(sig, syscall.SIGTERM, syscall.SIGINT)
<-sig

// 1. Fail readiness first, so the load balancer stops sending new requests
//    even before ECS deregisters us. A few seconds of overlap is deliberate.
ready.Store(false)
time.Sleep(5 * time.Second)

// 2. Stop accepting, finish what is in flight. Shutdown does exactly this.
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
if err := srv.Shutdown(ctx); err != nil {
    log.Warn("graceful shutdown timed out", "err", err)   // stopTimeout must exceed 25s
}

// 3. Then, and only then, release everything downstream.
pool.Close()
_ = tracerProvider.Shutdown(ctx)

# The four-stage shutdown ECS expects. Uvicorn already handles SIGTERM by
# draining, but the readiness flip in front of it is yours to write.
import signal, asyncio, contextlib, uvicorn

READY = True


@app.get("/readyz")
async def readyz():
    # 1. Fail readiness first, so the load balancer stops sending new requests
    #    even before ECS deregisters us. A few seconds of overlap is deliberate.
    return Response(status_code=200 if READY else 503)


# Run uvicorn programmatically, so the signal handler can hand the signal
# back to it. The graceful timeout must be SHORTER than stopTimeout.
server = uvicorn.Server(uvicorn.Config(
    app, host="0.0.0.0", port=8080,
    timeout_keep_alive=75,             # above the ALB idle timeout
    timeout_graceful_shutdown=20,      # below stopTimeout, which we set to 60
))


@contextlib.asynccontextmanager
async def lifespan(app):
    loop = asyncio.get_running_loop()

    def _term():
        global READY
        READY = False
        # Installing this handler REPLACED uvicorn’s own, so we have to pass
        # the signal on — otherwise nothing drains and ECS just SIGKILLs us.
        loop.call_later(5.0, server.handle_exit, signal.SIGTERM, None)

    loop.add_signal_handler(signal.SIGTERM, _term)
    yield
    # 2. By the time teardown runs, uvicorn has stopped accepting and has
    #    already waited out the in-flight requests.
    # 3. Then, and only then, release everything downstream.
    await engine.dispose()
    await tracer_provider.shutdown()

Deploy from code, and fail the pipeline honestly

// UpdateService returns as soon as the deployment is accepted — not when it
// is safe. The waiter is what makes a pipeline meaningful.
_, err := ecsc.UpdateService(ctx, &ecs.UpdateServiceInput{
    Cluster:            aws.String("prod"),
    Service:            aws.String("orders-api"),
    TaskDefinition:     aws.String(newTaskDefARN),
    ForceNewDeployment: aws.Bool(true),
    DeploymentConfiguration: &types.DeploymentConfiguration{
        MinimumHealthyPercent: aws.Int32(100),
        MaximumPercent:        aws.Int32(200),
        DeploymentCircuitBreaker: &types.DeploymentCircuitBreaker{
            Enable:   true,
            Rollback: true,       // revert to the last good task definition
        },
    },
})
if err != nil {
    return err
}

// "Stable" means: desired count running, one deployment, all targets healthy.
if err := ecs.NewServicesStableWaiter(ecsc).Wait(ctx, &ecs.DescribeServicesInput{
    Cluster: aws.String("prod"), Services: []string{"orders-api"},
}, 15*time.Minute); err != nil {
    // The circuit breaker may have rolled us back to the previous revision —
    // which is a successful rollback and a FAILED deploy. Say so.
    return fmt.Errorf("deployment did not stabilise: %w", err)
}

# update_service returns as soon as the deployment is accepted — not when it
# is safe. The waiter is what makes a pipeline meaningful.
ECS = boto3.client("ecs", config=CFG)

ECS.update_service(
    cluster="prod",
    service="orders-api",
    taskDefinition=new_task_def_arn,
    forceNewDeployment=True,
    deploymentConfiguration={
        "minimumHealthyPercent": 100,
        "maximumPercent": 200,
        "deploymentCircuitBreaker": {
            "enable": True,
            "rollback": True,       # revert to the last good task definition
        },
    },
)

# "Stable" means: desired count running, one deployment, all targets healthy.
try:
    ECS.get_waiter("services_stable").wait(
        cluster="prod", services=["orders-api"],
        WaiterConfig={"Delay": 15, "MaxAttempts": 60},   # 15 minutes
    )
except WaiterError as e:
    # The circuit breaker may have rolled us back to the previous revision —
    # which is a successful rollback and a FAILED deploy. Say so.
    events = ECS.describe_services(cluster="prod", services=["orders-api"])
    for ev in events["services"][0]["events"][:5]:
        log.error("ecs event", extra={"message": ev["message"]})
    raise DeployFailed("deployment did not stabilise") from e

// Work out the crossover instead of arguing about it. Prices are illustrative
// US-East-1 list rates — substitute your own before quoting anyone.
const (
    lambdaPerGBSecond = 0.0000166667
    lambdaPerRequest  = 0.0000002
    fargateVCPUHour   = 0.04048       // x86; Graviton and Spot are cheaper
    fargateGBHour     = 0.004445
)

// One Lambda invocation: memory in GB, duration rounded up to the millisecond.
func lambdaMonthly(reqsPerMonth float64, memGB, durSec float64) float64 {
    return reqsPerMonth*(lambdaPerRequest+memGB*durSec*lambdaPerGBSecond)
}

// A Fargate task charges whether or not a request arrives.
func fargateMonthly(tasks, vcpu, memGB float64) float64 {
    return tasks * 730 * (vcpu*fargateVCPUHour + memGB*fargateGBHour)
}

// 40 ms at 512 MB, against two 0.5 vCPU / 1 GB tasks:
//   5M requests/month   Lambda $2.67   Fargate $36.04   → Lambda, easily
//   500M requests/month Lambda $267    Fargate $36.04   → containers, easily
// The crossover is near 68M requests — but only if the tasks can actually
// serve them. Capacity check first: reqs/sec x duration must fit in
// tasks x concurrency_per_task, or you are comparing against a fantasy.
fmt.Printf("%.2f vs %.2f\n",
    lambdaMonthly(5e6, 0.5, 0.04), fargateMonthly(2, 0.5, 1))

# Work out the crossover instead of arguing about it. Prices are illustrative
# US-East-1 list rates — substitute your own before quoting anyone.
LAMBDA_GB_SECOND = 0.0000166667
LAMBDA_REQUEST = 0.0000002
FARGATE_VCPU_HOUR = 0.04048        # x86; Graviton and Spot are cheaper
FARGATE_GB_HOUR = 0.004445


def lambda_monthly(reqs: float, mem_gb: float, dur_s: float) -> float:
    # One invocation: memory in GB, duration rounded up to the millisecond.
    return reqs * (LAMBDA_REQUEST + mem_gb * dur_s * LAMBDA_GB_SECOND)


def fargate_monthly(tasks: float, vcpu: float, mem_gb: float) -> float:
    # A Fargate task charges whether or not a request arrives.
    return tasks * 730 * (vcpu * FARGATE_VCPU_HOUR + mem_gb * FARGATE_GB_HOUR)


# 40 ms at 512 MB, against two 0.5 vCPU / 1 GB tasks:
#   5M requests/month   Lambda $2.67   Fargate $36.04   → Lambda, easily
#   500M requests/month Lambda $267    Fargate $36.04   → containers, easily
# The crossover is near 68M requests — but only if the tasks can actually
# serve them. Capacity check first: reqs/sec * duration must fit in
# tasks * concurrency_per_task, or you are comparing against a fantasy.
print(f"{lambda_monthly(5e6, 0.5, 0.04):.2f} vs {fargate_monthly(2, 0.5, 1):.2f}")

// A consumer loop that does the three things people leave out: long polling,
// a visibility heartbeat, and a batch delete of exactly what succeeded.
for {
    out, err := app.sqs.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
        QueueUrl:              aws.String(queueURL),
        MaxNumberOfMessages:   10,
        WaitTimeSeconds:       20,          // long poll: cheaper AND lower latency
        VisibilityTimeout:     60,
        MessageAttributeNames: []string{"All"},
        // ApproximateReceiveCount tells you this is a retry — log it, and treat
        // a high count as a signal rather than a surprise.
        MessageSystemAttributeNames: []types.MessageSystemAttributeName{"ApproximateReceiveCount"},
    })
    if err != nil {
        if errors.Is(err, context.Canceled) {
            return nil                       // shutting down
        }
        log.Error("receive", "err", err)
        time.Sleep(time.Second)
        continue
    }

    var done []types.DeleteMessageBatchRequestEntry
    for _, m := range out.Messages {
        hb, stop := context.WithCancel(ctx)
        go heartbeat(hb, app.sqs, queueURL, *m.ReceiptHandle, 60)   // extend while busy

        err := handle(ctx, m)
        stop()
        if err != nil {
            log.Error("handler", "id", aws.ToString(m.MessageId), "err", err)
            continue                          // leave it: it becomes visible again
        }
        done = append(done, types.DeleteMessageBatchRequestEntry{
            Id: m.MessageId, ReceiptHandle: m.ReceiptHandle,
        })
    }

    if len(done) > 0 {
        // A batch delete can partially fail. Anything in .Failed stays in the
        // queue and WILL be delivered again — which is why handlers are idempotent.
        res, err := app.sqs.DeleteMessageBatch(ctx, &sqs.DeleteMessageBatchInput{
            QueueUrl: aws.String(queueURL), Entries: done,
        })
        if err == nil && len(res.Failed) > 0 {
            log.Warn("delete partially failed", "count", len(res.Failed))
        }
    }
}

# A consumer loop that does the three things people leave out: long polling,
# a visibility heartbeat, and a batch delete of exactly what succeeded.
import threading

QUEUE = "https://sqs.eu-west-1.amazonaws.com/1234/orders"


def heartbeat(receipt: str, every: float, timeout: int, stop: threading.Event):
    # Extend the loan while we are still working on it.
    while not stop.wait(every):
        SQS.change_message_visibility(
            QueueUrl=QUEUE, ReceiptHandle=receipt, VisibilityTimeout=timeout
        )


while running:
    res = SQS.receive_message(
        QueueUrl=QUEUE,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=20,              # long poll: cheaper AND lower latency
        VisibilityTimeout=60,
        MessageAttributeNames=["All"],
        # ApproximateReceiveCount tells you this is a retry — log it, and treat
        # a high count as a signal rather than a surprise.
        MessageSystemAttributeNames=["ApproximateReceiveCount"],
    )

    done = []
    for msg in res.get("Messages", []):
        stop = threading.Event()
        t = threading.Thread(
            target=heartbeat, args=(msg["ReceiptHandle"], 20.0, 60, stop), daemon=True
        )
        t.start()
        try:
            handle(msg)
            done.append({"Id": msg["MessageId"], "ReceiptHandle": msg["ReceiptHandle"]})
        except Exception:
            log.exception("handler failed", extra={"id": msg["MessageId"]})
            # leave it: visibility expires and it comes back
        finally:
            stop.set()

    if done:
        # A batch delete can partially fail. Anything in Failed stays in the
        # queue and WILL be delivered again — hence idempotent handlers.
        out = SQS.delete_message_batch(QueueUrl=QUEUE, Entries=done)
        if out.get("Failed"):
            log.warning("delete partially failed", extra={"n": len(out["Failed"])})

// EventBridge: the envelope matters more than the payload. Source, DetailType
// and the fields inside Detail are what every consumer’s rule will match on,
// so they are a public interface — version them.
type OrderPlaced struct {
    Version    string `json:"version"`      // "1" — additive changes only
    OrderID    string `json:"orderId"`
    TenantID   string `json:"tenantId"`
    TotalCents int64  `json:"totalCents"`
    Currency   string `json:"currency"`
    PlacedAt   string `json:"placedAt"`
}

detail, _ := json.Marshal(OrderPlaced{Version: "1", OrderID: id, TenantID: tenant,
    TotalCents: cents, Currency: "GBP", PlacedAt: now.Format(time.RFC3339)})

out, err := app.eb.PutEvents(ctx, &eventbridge.PutEventsInput{
    Entries: []types.PutEventsRequestEntry{{
        EventBusName: aws.String("orders"),
        Source:       aws.String("com.example.orders"),   // reverse-DNS, stable
        DetailType:   aws.String("OrderPlaced"),          // the event name
        Detail:       aws.String(string(detail)),
        Resources:    []string{"arn:aws:acme:orders::" + id},
        TraceHeader:  aws.String(traceHeader),            // keeps X-Ray joined up
    }},
})
// PutEvents returns 200 even when individual entries fail. This is the single
// most-missed error path in EventBridge code.
if err != nil {
    return err
}
if out.FailedEntryCount > 0 {                   // a plain int32 here, not a pointer
    for _, e := range out.Entries {
        if e.ErrorCode != nil {
            return fmt.Errorf("putevents: %s %s",
                aws.ToString(e.ErrorCode), aws.ToString(e.ErrorMessage))
        }
    }
}

# EventBridge: the envelope matters more than the payload. Source, DetailType
# and the fields inside Detail are what every consumer’s rule will match on,
# so they are a public interface — version them.
EB = boto3.client("events", config=CFG)

detail = {
    "version": "1",                 # additive changes only
    "orderId": order_id,
    "tenantId": tenant_id,
    "totalCents": cents,
    "currency": "GBP",
    "placedAt": now.isoformat(),
}

res = EB.put_events(
    Entries=[{
        "EventBusName": "orders",
        "Source": "com.example.orders",      # reverse-DNS, stable
        "DetailType": "OrderPlaced",         # the event name
        "Detail": json.dumps(detail),
        "Resources": [f"arn:aws:acme:orders::{order_id}"],
        "TraceHeader": trace_header,         # keeps X-Ray joined up
    }]
)

# put_events returns HTTP 200 even when individual entries fail. This is the
# single most-missed error path in EventBridge code.
if res["FailedEntryCount"]:
    bad = [e for e in res["Entries"] if "ErrorCode" in e]
    raise PublishFailed(bad[0]["ErrorCode"], bad[0]["ErrorMessage"])

The rule that receives it — and the filter that saves you money

An EventBridge pattern matches the structure of the event, not a string. Anything you omit matches everything; anything you include must match exactly, unless you use one of the comparison operators. Filtering at the bus is free; filtering inside a Lambda costs an invocation.

// A rule pattern is JSON that mirrors the event. Values are always ARRAYS —
// "any of these" — which is the syntax people get wrong first.
pattern := `{
  "source": ["com.example.orders"],
  "detail-type": ["OrderPlaced"],
  "detail": {
    "currency": ["GBP", "EUR"],
    "totalCents": [{"numeric": [">=", 50000]}],
    "tenantId": [{"anything-but": ["internal-test"]}]
  }
}`

_, err := app.eb.PutRule(ctx, &eventbridge.PutRuleInput{
    Name:         aws.String("large-orders-to-review"),
    EventBusName: aws.String("orders"),
    EventPattern: aws.String(pattern),
    State:        types.RuleStateEnabled,
})

// Every target needs its own retry policy and dead letter queue: a target
// that fails without one loses the event after 24 hours of attempts.
_, err = app.eb.PutTargets(ctx, &eventbridge.PutTargetsInput{
    Rule:         aws.String("large-orders-to-review"),
    EventBusName: aws.String("orders"),
    Targets: []types.Target{{
        Id:  aws.String("review-queue"),
        Arn: aws.String(reviewQueueARN),
        RetryPolicy: &types.RetryPolicy{
            MaximumRetryAttempts:     aws.Int32(4),
            MaximumEventAgeInSeconds: aws.Int32(3600),
        },
        DeadLetterConfig: &types.DeadLetterConfig{Arn: aws.String(dlqARN)},
    }},
})

# A rule pattern is JSON that mirrors the event. Values are always LISTS —
# "any of these" — which is the syntax people get wrong first.
PATTERN = {
    "source": ["com.example.orders"],
    "detail-type": ["OrderPlaced"],
    "detail": {
        "currency": ["GBP", "EUR"],
        "totalCents": [{"numeric": [">=", 50000]}],
        "tenantId": [{"anything-but": ["internal-test"]}],
    },
}

EB.put_rule(
    Name="large-orders-to-review",
    EventBusName="orders",
    EventPattern=json.dumps(PATTERN),
    State="ENABLED",
)

# Every target needs its own retry policy and dead letter queue: a target
# that fails without one loses the event after 24 hours of attempts.
EB.put_targets(
    Rule="large-orders-to-review",
    EventBusName="orders",
    Targets=[{
        "Id": "review-queue",
        "Arn": REVIEW_QUEUE_ARN,
        "RetryPolicy": {"MaximumRetryAttempts": 4,
                        "MaximumEventAgeInSeconds": 3600},
        "DeadLetterConfig": {"Arn": DLQ_ARN},
    }],
)

# The SNS equivalent of the same idea — a subscription filter policy, which
# can read message attributes or, with FilterPolicyScope, the body itself.
SNS.set_subscription_attributes(
    SubscriptionArn=sub_arn,
    AttributeName="FilterPolicy",
    AttributeValue=json.dumps({"currency": ["GBP", "EUR"]}),
)

// PutRecords is a batch of up to 500 records — and it returns 200 with
// individual failures inside. Resending the WHOLE batch duplicates the
// records that succeeded, so resend only the failed indexes.
func putAll(ctx context.Context, k *kinesis.Client, stream string,
    recs []types.PutRecordsRequestEntry) error {

    for attempt := 0; attempt < 5 && len(recs) > 0; attempt++ {
        out, err := k.PutRecords(ctx, &kinesis.PutRecordsInput{
            StreamName: aws.String(stream),
            Records:    recs,
        })
        if err != nil {
            return err
        }
        if aws.ToInt32(out.FailedRecordCount) == 0 {
            return nil
        }

        var retry []types.PutRecordsRequestEntry
        for i, r := range out.Records {
            if r.ErrorCode != nil {
                // ProvisionedThroughputExceededException — this shard is hot
                retry = append(retry, recs[i])
            }
        }
        recs = retry
        time.Sleep(time.Duration(1<<attempt) * 100 * time.Millisecond)
    }
    return fmt.Errorf("%d records never accepted", len(recs))
}

// The partition key is the ordering AND the sharding decision. orderID keeps
// each order’s events in sequence and spreads load; "orders" would put the
// entire stream on one shard.
entry := types.PutRecordsRequestEntry{
    PartitionKey: aws.String(orderID),
    Data:         payload,
}

# put_records is a batch of up to 500 records — and it returns HTTP 200 with
# individual failures inside. Resending the WHOLE batch duplicates the records
# that succeeded, so resend only the failed indexes.
KIN = boto3.client("kinesis", config=CFG)


def put_all(stream: str, records: list[dict]) -> None:
    pending = records
    for attempt in range(5):
        res = KIN.put_records(StreamName=stream, Records=pending)
        if not res["FailedRecordCount"]:
            return
        # ProvisionedThroughputExceededException — that shard is hot
        pending = [
            rec for rec, out in zip(pending, res["Records"]) if "ErrorCode" in out
        ]
        time.sleep(0.1 * 2 ** attempt)
    raise PublishFailed(f"{len(pending)} records never accepted")


# The partition key is the ordering AND the sharding decision. order_id keeps
# each order’s events in sequence and spreads load; "orders" would put the
# entire stream on one shard.
put_all("order-events", [
    {"PartitionKey": order_id, "Data": json.dumps(event).encode()},
])

// An HTTP server configured for life behind an ALB. Every timeout here exists
// to fail in a way you can reason about.
srv := &http.Server{
    Addr:              ":8080",
    Handler:           mux,
    ReadHeaderTimeout: 5 * time.Second,     // slow-header attacks, and stuck clients
    ReadTimeout:       15 * time.Second,
    WriteTimeout:      30 * time.Second,
    // THE important one: strictly greater than the ALB idle timeout (60s),
    // so the balancer always closes idle connections, never the application.
    IdleTimeout: 75 * time.Second,
}

// Two endpoints, two meanings. Conflating them is how a slow database becomes
// a fleet-wide outage.
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) {
    // Liveness: am I this process, and is my event loop turning? Nothing else.
    w.WriteHeader(http.StatusOK)
})

mux.HandleFunc("GET /readyz", func(w http.ResponseWriter, r *http.Request) {
    // Readiness: should the load balancer send me traffic right now?
    // Flipped to false by the SIGTERM handler before the drain begins.
    if !ready.Load() {
        w.WriteHeader(http.StatusServiceUnavailable)
        return
    }
    w.WriteHeader(http.StatusOK)
})

// The client’s real address is in X-Forwarded-For, appended by the ALB. Trust
// only the LAST entry — anything earlier is client-supplied and forgeable.
func clientIP(r *http.Request) string {
    xff := r.Header.Values("X-Forwarded-For")
    if len(xff) == 0 {
        return r.RemoteAddr
    }
    parts := strings.Split(xff[len(xff)-1], ",")
    return strings.TrimSpace(parts[len(parts)-1])
}

# A uvicorn server configured for life behind an ALB. Every timeout here
# exists to fail in a way you can reason about.
#
#   uvicorn app:app --host 0.0.0.0 --port 8080 \
#     --timeout-keep-alive 75 \      THE important one: strictly greater than
#                                    the ALB idle timeout (60s), so the
#                                    balancer always closes idle connections
#     --timeout-graceful-shutdown 20 \
#     --workers 4 --proxy-headers --forwarded-allow-ips='*'

# Two endpoints, two meanings. Conflating them is how a slow database becomes
# a fleet-wide outage.


@app.get("/healthz")
async def healthz():
    # Liveness: am I this process, and is my event loop turning? Nothing else.
    return Response(status_code=200)


@app.get("/readyz")
async def readyz():
    # Readiness: should the load balancer send me traffic right now?
    # Flipped to False by the SIGTERM handler before the drain begins.
    return Response(status_code=200 if READY else 503)


def client_ip(request: Request) -> str:
    # The client’s real address is in X-Forwarded-For, appended by the ALB.
    # Trust only the LAST entry — anything earlier is forgeable by the client.
    xff = request.headers.get("x-forwarded-for")
    if not xff:
        return request.client.host
    return xff.split(",")[-1].strip()

// An HTTP API (payload format 2.0) handler. The event shape is different from
// the REST API’s v1 — routeKey, http.method, and a flattened single-value
// header map are the differences that bite when you migrate.
func handle(ctx context.Context, req events.APIGatewayV2HTTPRequest) (
    events.APIGatewayV2HTTPResponse, error) {

    id := req.PathParameters["id"]                       // from route GET /orders/{id}

    // The authorizer’s output arrives here — already verified, so do not
    // re-parse the token, and never trust a claim you did not configure.
    tenant, _ := req.RequestContext.Authorizer.JWT.Claims["tenant_id"]

    order, err := app.load(ctx, tenant, id)
    switch {
    case errors.Is(err, ErrNotFound):
        return jsonErr(404, "not_found", "no such order"), nil
    case err != nil:
        // Returning an error gives the caller an opaque 500 and no body.
        // Prefer an explicit response you control, and log the detail.
        log.Error("load order", "id", id, "err", err)
        return jsonErr(500, "internal", "try again"), nil
    }

    body, _ := json.Marshal(order)
    return events.APIGatewayV2HTTPResponse{
        StatusCode: 200,
        Headers: map[string]string{
            "content-type":  "application/json",
            "cache-control": "no-store",
        },
        Body: string(body),
    }, nil
}

# An HTTP API (payload format 2.0) handler. The event shape differs from the
# REST API’s v1 — routeKey, requestContext.http.method, and single-value
# headers are the differences that bite when you migrate.
from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
from aws_lambda_powertools.event_handler.exceptions import NotFoundError

app_router = APIGatewayHttpResolver()


@app_router.get("/orders/<order_id>")
def get_order(order_id: str):
    # The authorizer’s output arrives already verified — do not re-parse the
    # token, and never trust a claim you did not configure the authorizer to check.
    claims = app_router.current_event.request_context.authorizer.jwt_claim
    tenant = claims["tenant_id"]

    order = load(tenant, order_id)
    if order is None:
        raise NotFoundError                     # becomes a clean 404 JSON body

    return Response(
        status_code=200,
        content_type="application/json",
        headers={"cache-control": "no-store"},
        body=json.dumps(order),
    )


@logger.inject_lambda_context
@tracer.capture_lambda_handler
def handler(event, context):
    # The resolver turns unhandled exceptions into 500s with a request id,
    # instead of the opaque gateway error you get from raising here.
    return app_router.resolve(event, context)

// EMF: a metric and its context in one stdout line. CloudWatch parses the
// _aws block and creates the metrics; every other field stays queryable in
// Logs Insights. No API call, no added latency, no throttling.
type metricDef struct {
    Name string `json:"Name"`
    Unit string `json:"Unit"`
}

type directive struct {
    Namespace  string      `json:"Namespace"`
    Dimensions [][]string  `json:"Dimensions"`
    Metrics    []metricDef `json:"Metrics"`
}

func emit(op string, d time.Duration, tenant, order, trace string) {
    rec := map[string]any{
        "_aws": map[string]any{
            "Timestamp": time.Now().UnixMilli(),
            "CloudWatchMetrics": []directive{{
                Namespace: "Orders",
                // TWO dimensions, not five: every unique combination is a
                // separate billed metric series, for ever.
                Dimensions: [][]string{{"service", "operation"}},
                Metrics:    []metricDef{{Name: "latency", Unit: "Milliseconds"}},
            }},
        },
        // Dimensions must also appear as top-level fields.
        "service":   "orders-api",
        "operation": op,
        "latency":   d.Milliseconds(),
        // High-cardinality context: searchable in Logs Insights, but NOT a
        // dimension, so not billed as thousands of metric series.
        "tenantId": tenant,
        "orderId":  order,
        "traceId":  trace,
    }
    _ = json.NewEncoder(os.Stdout).Encode(rec)
}

# EMF: a metric and its context in one stdout line. CloudWatch parses it and
# creates the metrics; every other field stays queryable in Logs Insights.
# No API call, no added latency, no throttling. Powertools writes the format.
from aws_lambda_powertools import Metrics
from aws_lambda_powertools.metrics import MetricUnit

metrics = Metrics(namespace="Orders", service="orders-api")


def place_order(tenant_id: str, order_id: str, elapsed_ms: float):
    # Dimensions: the two or three you would actually alarm on. Every unique
    # combination is a separate billed metric series, for ever.
    metrics.add_dimension(name="operation", value="place_order")
    metrics.add_metric(name="latency", unit=MetricUnit.Milliseconds, value=elapsed_ms)

    # High-cardinality context goes in metadata: searchable in Logs Insights,
    # attached to the same record, and NOT a billed metric series.
    metrics.add_metadata(key="tenantId", value=tenant_id)
    metrics.add_metadata(key="orderId", value=order_id)


# Correlate logs with traces so one click goes from alarm to request:
logger = Logger(service="orders-api")
logger.info("order placed", extra={"orderId": order_id})
# Powertools injects xray_trace_id automatically when tracing is enabled.

# The Logs Insights query that turns those lines into an answer:
#   fields @timestamp, tenantId, latency
#   | filter operation = "place_order" and latency > 1000
#   | stats count(*) as slow, pct(latency, 99) as p99 by bin(5m), tenantId
#   | sort slow desc | limit 20

// Cache with a TTL, and refresh on demand when authentication fails.
// Both halves matter: the cache saves money, the invalidation survives rotation.
type SecretCache struct {
    mu    sync.RWMutex
    sm    *secretsmanager.Client
    ttl   time.Duration
    value string
    at    time.Time
}

func (c *SecretCache) Get(ctx context.Context, force bool) (string, error) {
    c.mu.RLock()
    if !force && c.value != "" && time.Since(c.at) < c.ttl {
        v := c.value
        c.mu.RUnlock()
        return v, nil
    }
    c.mu.RUnlock()

    c.mu.Lock()
    defer c.mu.Unlock()
    if !force && c.value != "" && time.Since(c.at) < c.ttl {
        return c.value, nil                    // another goroutine refreshed it
    }

    out, err := c.sm.GetSecretValue(ctx, &secretsmanager.GetSecretValueInput{
        SecretId:     aws.String("orders/db"),
        VersionStage: aws.String("AWSCURRENT"),   // never pin a version id
    })
    if err != nil {
        return "", err
    }
    c.value, c.at = aws.ToString(out.SecretString), time.Now()
    return c.value, nil
}

// At the call site: one retry with force=true turns a rotation from an outage
// into a single failed connection.
pw, _ := cache.Get(ctx, false)
if err := connect(ctx, pw); errors.Is(err, ErrAuth) {
    pw, _ = cache.Get(ctx, true)
    err = connect(ctx, pw)
}

# Cache with a TTL, and refresh on demand when authentication fails.
# Both halves matter: the cache saves money, the invalidation survives rotation.
import threading, time

SM = boto3.client("secretsmanager", config=CFG)


class SecretCache:
    def __init__(self, secret_id: str, ttl: float = 900.0):
        self._id, self._ttl = secret_id, ttl
        self._lock = threading.Lock()
        self._value: str | None = None
        self._at = 0.0

    def get(self, force: bool = False) -> str:
        fresh = self._value is not None and time.monotonic() - self._at < self._ttl
        if fresh and not force:
            return self._value
        with self._lock:
            # re-check: another thread may have refreshed while we waited
            fresh = self._value is not None and time.monotonic() - self._at < self._ttl
            if fresh and not force:
                return self._value
            res = SM.get_secret_value(
                SecretId=self._id,
                VersionStage="AWSCURRENT",     # never pin a version id
            )
            self._value, self._at = res["SecretString"], time.monotonic()
            return self._value


# At the call site: one retry with force=True turns a rotation from an outage
# into a single failed connection.
cache = SecretCache("orders/db")
try:
    connect(cache.get())
except AuthError:
    connect(cache.get(force=True))

// Month over month, by service, biggest increase first. Ten seconds of API
// calls that answers "why is the bill up" better than any dashboard.
ce := costexplorer.NewFromConfig(cfg)

get := func(start, end string) (map[string]float64, error) {
    out, err := ce.GetCostAndUsage(ctx, &costexplorer.GetCostAndUsageInput{
        TimePeriod:  &types.DateInterval{Start: aws.String(start), End: aws.String(end)},
        Granularity: types.GranularityMonthly,
        Metrics:     []string{"UnblendedCost"},
        GroupBy: []types.GroupDefinition{
            {Type: types.GroupDefinitionTypeDimension, Key: aws.String("SERVICE")},
        },
    })
    if err != nil {
        return nil, err
    }
    m := map[string]float64{}
    for _, r := range out.ResultsByTime {
        for _, g := range r.Groups {
            v, _ := strconv.ParseFloat(aws.ToString(g.Metrics["UnblendedCost"].Amount), 64)
            m[g.Keys[0]] += v
        }
    }
    return m, nil
}

prev, _ := get("2026-07-01", "2026-08-01")
curr, _ := get("2026-08-01", "2026-09-01")
type row struct {
    svc           string
    amount, delta float64
}
rows := make([]row, 0, len(curr))
for svc, amount := range curr {
    rows = append(rows, row{svc, amount, amount - prev[svc]})
}
// a map has no order, so sort — otherwise "biggest increase first" is a lie
sort.Slice(rows, func(i, j int) bool { return rows[i].delta > rows[j].delta })

for _, r := range rows {
    if r.delta > 50 {                        // only what is worth a conversation
        fmt.Printf("%-42s %8.2f  (%+.2f)\n", r.svc, r.amount, r.delta)
    }
}

# Month over month, by service, biggest increase first. Ten seconds of API
# calls that answers "why is the bill up" better than any dashboard.
CE = boto3.client("ce", region_name="us-east-1")   # Cost Explorer is us-east-1


def by_service(start: str, end: str) -> dict[str, float]:
    res = CE.get_cost_and_usage(
        TimePeriod={"Start": start, "End": end},
        Granularity="MONTHLY",
        Metrics=["UnblendedCost"],
        GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
    )
    out: dict[str, float] = {}
    for period in res["ResultsByTime"]:
        for g in period["Groups"]:
            out[g["Keys"][0]] = out.get(g["Keys"][0], 0.0) + float(
                g["Metrics"]["UnblendedCost"]["Amount"]
            )
    return out


prev = by_service("2026-07-01", "2026-08-01")
curr = by_service("2026-08-01", "2026-09-01")

rows = [(svc, amt, amt - prev.get(svc, 0.0)) for svc, amt in curr.items()]
for svc, amt, delta in sorted(rows, key=lambda r: -r[2]):
    if delta > 50:                     # only what is worth a conversation
        print(f"{svc:<42} {amt:8.2f}  ({delta:+.2f})")

// POST /orders — the handler, with every failure mode from this page in it.
func (a *App) PlaceOrder(w http.ResponseWriter, r *http.Request) {
    ctx, cancel := context.WithTimeout(r.Context(), 8*time.Second)
    defer cancel()

    key := r.Header.Get("Idempotency-Key")
    if key == "" {
        writeErr(w, 400, "idempotency_key_required")
        return
    }

    var req PlaceOrderRequest
    if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&req); err != nil {
        writeErr(w, 400, "bad_json")
        return
    }

    // 1. Claim the key. Conditional put: whoever wins does the work, and the
    //    TTL means abandoned claims clean themselves up.
    claimed, prior, err := a.claim(ctx, key, 24*time.Hour)
    switch {
    case err != nil:
        writeErr(w, 503, "try_again")
        return
    case !claimed && prior != nil:
        w.Header().Set("Idempotent-Replay", "true")
        _, _ = w.Write(prior)                    // byte-identical to the first reply
        return
    case !claimed:
        writeErr(w, 409, "in_progress")          // claimed, not finished: retry later
        return
    }

    // 2. Order and outbox row commit together, or neither exists.
    tx, err := a.pool.Begin(ctx)
    if err != nil {
        writeErr(w, 503, "try_again")
        return
    }
    defer tx.Rollback(ctx)                        // no-op after a successful commit

    order, err := insertOrder(ctx, tx, req)
    if err != nil {
        writeErr(w, 500, "internal")
        return
    }
    event, _ := json.Marshal(OrderPlaced{Version: "1", OrderID: order.ID})
    if err := insertOutbox(ctx, tx, order.ID, "OrderPlaced", event); err != nil {
        writeErr(w, 500, "internal")
        return
    }
    if err := tx.Commit(ctx); err != nil {
        // A commit that fails after a failover may still have applied. The
        // idempotency key is what makes the client’s retry safe here.
        writeErr(w, 503, "try_again")
        return
    }

    // 3. Store the reply against the key, so the retry gets THIS body.
    body, _ := json.Marshal(order)
    _ = a.completeClaim(ctx, key, body)
    w.WriteHeader(201)
    _, _ = w.Write(body)
}

# POST /orders — the handler, with every failure mode from this page in it.


@app.post("/orders", status_code=201)
async def place_order(req: PlaceOrderRequest, response: Response,
                      idempotency_key: str = Header(alias="Idempotency-Key")):

    # 1. Claim the key. Conditional put: whoever wins does the work, and the
    #    TTL means abandoned claims clean themselves up.
    claimed, prior = await claim(idempotency_key, ttl=timedelta(hours=24))
    if not claimed and prior is not None:
        response.headers["Idempotent-Replay"] = "true"
        return Response(content=prior, media_type="application/json")
    if not claimed:
        raise HTTPException(409, "in_progress")   # claimed, not finished

    # 2. Order and outbox row commit together, or neither exists.
    try:
        async with engine.begin() as conn:        # commits on exit, rolls back on raise
            order = await insert_order(conn, req)
            await conn.execute(
                text("""INSERT INTO outbox (aggregate_id, type, payload)
                        VALUES (:id, :type, CAST(:payload AS jsonb))"""),
                {"id": order.id, "type": "OrderPlaced",
                 "payload": json.dumps({"version": "1", "orderId": order.id})},
            )
    except OperationalError:
        # A commit that fails after a failover may still have applied. The
        # idempotency key is what makes the client’s retry safe here.
        raise HTTPException(503, "try_again")

    # 3. Store the reply against the key, so the retry gets THIS body.
    body = order.model_dump_json()
    await complete_claim(idempotency_key, body)
    return Response(content=body, status_code=201, media_type="application/json")

The claim, in DynamoDB

One conditional write does all of it: reserve the key if nobody has it, return the stored response if the work is finished, and allow a re-attempt if a previous holder died before completing.

// claim reserves the idempotency key, or hands back the finished response.
func (a *App) claim(ctx context.Context, key string, ttl time.Duration) (bool, []byte, error) {
    now := time.Now()
    _, err := a.ddb.PutItem(ctx, &dynamodb.PutItemInput{
        TableName: aws.String("idempotency"),
        Item: map[string]types.AttributeValue{
            "pk":       &types.AttributeValueMemberS{Value: key},
            "state":    &types.AttributeValueMemberS{Value: "in_progress"},
            "expires":  &types.AttributeValueMemberN{Value: fmt.Sprint(now.Add(ttl).Unix())},
            "lockedAt": &types.AttributeValueMemberN{Value: fmt.Sprint(now.Unix())},
        },
        // Win if the key is unknown, OR if a previous holder crashed mid-flight
        // and its lock is stale. A FINISHED claim is never retaken — without the
        // state check, a retry 30s later re-runs the order instead of replaying it.
        // ("state" is a DynamoDB reserved word, hence the #st alias.)
        ConditionExpression: aws.String(
            "attribute_not_exists(pk) OR (#st = :inprog AND lockedAt < :stale)"),
        ExpressionAttributeNames: map[string]string{"#st": "state"},
        ExpressionAttributeValues: map[string]types.AttributeValue{
            ":inprog": &types.AttributeValueMemberS{Value: "in_progress"},
            ":stale": &types.AttributeValueMemberN{
                Value: fmt.Sprint(now.Add(-30 * time.Second).Unix())},
        },
        ReturnValuesOnConditionCheckFailure: types.ReturnValuesOnConditionCheckFailureAllOld,
    })
    if err == nil {
        return true, nil, nil
    }

    var failed *types.ConditionalCheckFailedException
    if !errors.As(err, &failed) {
        return false, nil, err
    }
    // We lost. The existing item comes back with the failure, so a completed
    // request needs no second read.
    if resp, ok := failed.Item["response"].(*types.AttributeValueMemberB); ok {
        return false, resp.Value, nil
    }
    return false, nil, nil                 // still in progress elsewhere
}

# claim() reserves the idempotency key, or hands back the finished response.
async def claim(key: str, ttl: timedelta) -> tuple[bool, bytes | None]:
    now = int(time.time())
    try:
        DDB.put_item(
            TableName="idempotency",
            Item={
                "pk": {"S": key},
                "state": {"S": "in_progress"},
                "expires": {"N": str(now + int(ttl.total_seconds()))},
                "lockedAt": {"N": str(now)},
            },
            # Win if the key is unknown, OR if a previous holder crashed mid-flight
            # and its lock is stale. A FINISHED claim is never retaken — without the
            # state check, a retry 30s later re-runs the order instead of replaying it.
            # ("state" is a DynamoDB reserved word, hence the #st alias.)
            ConditionExpression=(
                "attribute_not_exists(pk) OR (#st = :inprog AND lockedAt < :stale)"
            ),
            ExpressionAttributeNames={"#st": "state"},
            ExpressionAttributeValues={":inprog": {"S": "in_progress"},
                                       ":stale": {"N": str(now - 30)}},
            ReturnValuesOnConditionCheckFailure="ALL_OLD",
        )
        return True, None
    except DDB.exceptions.ConditionalCheckFailedException as e:
        # We lost. The existing item comes back with the failure, so a completed
        # request needs no second read.
        item = e.response.get("Item") or {}
        stored = item.get("response", {}).get("B")
        return False, stored          # stored is None while still in progress

The infrastructure, in about forty lines

Written once, reviewed like code, and identical in both languages because it is neither. The parts worth pointing at are the circuit breaker, the DLQ, and the two timers that must be ordered correctly.

# The pieces that this page argued for, expressed as configuration.

resource "aws_ecs_service" "orders" {
  name            = "orders-api"
  desired_count   = 4
  launch_type     = "FARGATE"

  deployment_circuit_breaker {
    enable   = true          # stop a bad rollout
    rollback = true          # and put the last good revision back
  }
  deployment_minimum_healthy_percent = 100
  deployment_maximum_percent         = 200      # doubles connections for 90s
  health_check_grace_period_seconds  = 60       # let the app warm up first
}

resource "aws_lb_target_group" "orders" {
  deregistration_delay = 30                     # > longest request, < stopTimeout
  health_check {
    path                = "/readyz"             # NOT a check that touches the DB
    interval            = 10
    healthy_threshold   = 2
    unhealthy_threshold = 2
  }
}

resource "aws_sqs_queue" "billing" {
  visibility_timeout_seconds = 180              # 6x the consumer’s 30s timeout
  receive_wait_time_seconds  = 20               # long polling, always
  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.billing_dlq.arn
    maxReceiveCount     = 5
  })
}

resource "aws_s3_bucket_lifecycle_configuration" "archive" {
  rule {
    id     = "abort-incomplete-multipart"       # the free money rule
    status = "Enabled"
    abort_incomplete_multipart_upload { days_after_initiation = 7 }
  }
}

resource "aws_cloudwatch_log_group" "orders" {
  retention_in_days = 30                        # the default is FOREVER
}

// preflight runs before the readiness probe goes green. Everything it checks
// is something that has silently broken a production deploy at least once.
func preflight(ctx context.Context, a *App) error {
    ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
    defer cancel()

    // 1. Am I who the deployment thinks I am?
    who, err := a.sts.GetCallerIdentity(ctx, nil)
    if err != nil {
        return fmt.Errorf("no usable credentials: %w", err)
    }
    if !strings.Contains(aws.ToString(who.Arn), expectedRole) {
        return fmt.Errorf("wrong role: %s", aws.ToString(who.Arn))
    }

    // 2. Does the table exist, in this region, with a TTL as designed?
    tt, err := a.ddb.DescribeTimeToLive(ctx, &dynamodb.DescribeTimeToLiveInput{
        TableName: aws.String("idempotency"),
    })
    if err != nil {
        return fmt.Errorf("idempotency table: %w", err)
    }
    if tt.TimeToLiveDescription.TimeToLiveStatus != types.TimeToLiveStatusEnabled {
        return errors.New("idempotency table has no TTL: it will grow for ever")
    }

    // 3. Does every queue we consume have somewhere for poison to go?
    for _, q := range a.queues {
        at, err := a.sqs.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{
            QueueUrl:       aws.String(q),
            AttributeNames: []sqstypes.QueueAttributeName{"RedrivePolicy"},
        })
        if err != nil {
            return err
        }
        if at.Attributes["RedrivePolicy"] == "" {
            return fmt.Errorf("queue %s has no dead letter queue", q)
        }
    }

    // 4. Can we actually write, with the timeout we promised?
    return a.pool.QueryRow(ctx, "SELECT 1").Scan(new(int))
}

# preflight() runs before the readiness probe goes green. Everything it checks
# is something that has silently broken a production deploy at least once.
def preflight() -> None:
    # 1. Am I who the deployment thinks I am?
    arn = boto3.client("sts", config=CFG).get_caller_identity()["Arn"]
    if EXPECTED_ROLE not in arn:
        raise Misconfigured(f"wrong role: {arn}")

    # 2. Does the table exist, in this region, with a TTL as designed?
    ttl = DDB.describe_time_to_live(TableName="idempotency")
    if ttl["TimeToLiveDescription"]["TimeToLiveStatus"] != "ENABLED":
        raise Misconfigured("idempotency table has no TTL: it will grow for ever")

    # 3. Does every queue we consume have somewhere for poison to go?
    for url in QUEUES:
        attrs = SQS.get_queue_attributes(
            QueueUrl=url, AttributeNames=["RedrivePolicy"]
        )["Attributes"]
        if "RedrivePolicy" not in attrs:
            raise Misconfigured(f"queue {url} has no dead letter queue")

    # 4. Can we actually write, with the timeout we promised?
    with engine.connect() as conn:
        conn.exec_driver_sql("SELECT 1")


# Called from the lifespan handler, before READY flips to True. A crash loop
# here is a failed deploy — which is exactly what it should be.