RabbitMQ Lab¶

rabbitmq-lab/
├── compose.yaml              single node (+ --profile monitoring)
├── compose.cluster.yaml      3 nodes behind HAProxy (+ --profile monitoring)
├── config/                   rabbitmq.conf, cluster peer discovery, haproxy, prometheus, grafana
├── py/                       one script per exercise; every script has --help
└── scripts/                  healthcheck.sh, triage.sh, perftest.sh

make env       # writes .env with random passwords and an Erlang cookie
make up        # docker compose up -d --wait: returns once the health check passes
make venv      # .venv with pika
make creds     # the generated login
source .venv/bin/activate && cd py

# config/rabbitmq.conf
## Lab settings, layered on top of the image's /etc/rabbitmq/conf.d/10-defaults.conf.
## Mounted as /etc/rabbitmq/conf.d/20-lab.conf.

## --- Resource alarms -------------------------------------------------------
## Publishers are blocked when either alarm fires. RabbitMQ reads the container's
## cgroup memory limit, so the relative watermark applies to mem_limit in compose.
vm_memory_high_watermark.relative = 0.6
## The built-in default (50MB) is far too low for anything real.
disk_free_limit.absolute = 1GB

## --- Logging ----------------------------------------------------------------
## Containers log to stdout; let the container runtime handle collection.
log.console = true
log.console.level = info
log.file = false

## --- Connections ------------------------------------------------------------
## Server-suggested heartbeat (seconds). Clients negotiate the lower of the two.
heartbeat = 60

## --- Consumers --------------------------------------------------------------
## Max time a delivery may stay unacknowledged (ms). As of 4.3 this is enforced
## by quorum queues only; classic queues and streams ignore it.
consumer_timeout = 1800000

## --- Prometheus ---------------------------------------------------------------
## Keep /metrics aggregated (cheap). Per-queue series come from /metrics/detailed,
## which Prometheus scrapes with an explicit family filter.
prometheus.return_per_object_metrics = false

docker compose exec rabbitmq rabbitmq-diagnostics status           # versions, config files loaded, resources
docker compose exec rabbitmq rabbitmq-diagnostics environment | grep -E 'vm_memory|disk_free'
docker compose exec rabbitmq rabbitmq-plugins list -e              # enabled plugins
docker compose exec rabbitmq rabbitmqctl list_feature_flags name state

# py/basics.py
def declare_topology(ch) -> None:
    """Declared identically by producer and consumer. Declarations are idempotent
    only if the arguments match: redeclaring a queue with different arguments
    fails with 406 PRECONDITION_FAILED and closes the channel."""
    ch.exchange_declare(exchange=EXCHANGE, exchange_type="topic", durable=True)
    ch.queue_declare(
        queue=QUEUE,
        durable=True,
        arguments={
            "x-queue-type": "quorum",
            "x-message-ttl": 24 * 60 * 60 * 1000,  # drop messages older than 24h
            "x-max-length": 10_000,
            "x-overflow": "reject-publish",  # when full, nack publishers instead of dropping old messages
        },
    )
    # "app.#" matches app.info, app.web.error, ... but not db.error.
    ch.queue_bind(queue=QUEUE, exchange=EXCHANGE, routing_key="app.#")

# py/basics.py
def produce(_args) -> None:
    conn = connect("basics-producer")
    ch = conn.channel()
    declare_topology(ch)
    ch.confirm_delivery()  # every basic_publish now waits for the broker's ack

    events = [
        ("app.info", "application started"),
        ("app.warning", "memory usage high"),
        ("app.error", "database connection failed"),
        ("app.web.info", "request served"),
        ("db.error", "this key matches no binding"),  # unroutable on purpose
    ]
    for routing_key, message in events:
        body = to_json(
            {
                "ts": datetime.now(timezone.utc).isoformat(),
                "key": routing_key,
                "message": message,
            }
        )
        try:
            # mandatory=True: if no queue is bound for this key, the broker returns
            # the message instead of silently dropping it.
            ch.basic_publish(EXCHANGE, routing_key, body, json_props(), mandatory=True)
            print(f" [x] confirmed  {routing_key:13} {message}")
        except pika.exceptions.UnroutableError:
            print(f" [!] unroutable {routing_key:13} (returned by broker)")
        except pika.exceptions.NackError:
            print(f" [!] nacked     {routing_key:13} (queue full or broker refused)")
    conn.close()

python basics.py consume      # T1
python basics.py produce      # T2

# py/basics.py
def consume(args) -> None:
    conn = connect("basics-consumer")
    ch = conn.channel()
    declare_topology(ch)
    ch.basic_qos(prefetch_count=args.prefetch)

    def on_message(ch, method, props, body):
        try:
            event = json.loads(body)
        except json.JSONDecodeError:
            # A message that can never be parsed must not be requeued, or it loops
            # forever. With a DLX configured it would be dead-lettered; here it's dropped.
            print(f" [!] unparseable message, rejecting: {body[:60]!r}")
            ch.basic_reject(method.delivery_tag, requeue=False)
            return
        print(f" [<] {method.routing_key:13} redelivered={method.redelivered!s:5} {event['message']}")
        time.sleep(0.2)  # simulated work
        ch.basic_ack(method.delivery_tag)

    ch.basic_consume(QUEUE, on_message, auto_ack=False)
    print(f" [*] consuming {QUEUE} (prefetch={args.prefetch}); Ctrl-C to exit")
    run_consumer(ch)

# py/work_queue.py
def worker(args) -> None:
    conn = connect(f"worker-{args.id}")
    ch = conn.channel()
    declare(ch)
    ch.basic_qos(prefetch_count=args.prefetch)

    def on_message(ch, method, _props, body):
        name, seconds = body.decode().split(":")
        print(f" [worker {args.id}] {name} ({seconds}s) ...", end="", flush=True)
        time.sleep(int(seconds))
        ch.basic_ack(method.delivery_tag)
        print(" done")

    ch.basic_consume(QUEUE, on_message)
    print(f" [worker {args.id}] waiting (prefetch={args.prefetch})")
    run_consumer(ch)

python work_queue.py worker 1     # T1
python work_queue.py worker 2     # T2
python work_queue.py send         # T3

# py/pubsub.py
def subscribe(args) -> None:
    conn = connect(f"pubsub-{args.name}")
    ch = conn.channel()
    ch.exchange_declare(exchange=EXCHANGE, exchange_type="fanout", durable=True)
    # queue="" lets the broker generate a name; exclusive=True ties the queue's
    # lifetime to this connection. Exclusive transient queues remain fully supported
    # (it's *non-exclusive* transient queues that 4.3 rejects by default).
    result = ch.queue_declare(queue="", exclusive=True)
    queue = result.method.queue
    ch.queue_bind(queue=queue, exchange=EXCHANGE)

    def on_message(_ch, _method, _props, body):
        print(f" [{args.name}] {body.decode()}")

    # auto_ack is acceptable here: notifications are fire-and-forget.
    ch.basic_consume(queue, on_message, auto_ack=True)
    print(f" [{args.name}] listening on {queue}")
    run_consumer(ch)

python pubsub.py subscribe email      # T1
python pubsub.py subscribe sms        # T2
python pubsub.py publish "maintenance in 10 minutes"   # T3

# py/slow_worker.py
def work(args) -> None:
    conn = connect("slow-worker", heartbeat=HEARTBEAT)
    ch = conn.channel()
    declare(ch)
    ch.basic_qos(prefetch_count=1)

    def ack(tag: int) -> None:
        if ch.is_open:
            ch.basic_ack(tag)
            print(f" [✓] acked delivery {tag}")

    def on_message(ch, method, props, body):
        seconds = int(body)
        headers = props.headers or {}
        print(f" [>] job {method.delivery_tag}: {seconds}s (delivery-count={headers.get('x-delivery-count', 0)})")
        if args.blocking:
            time.sleep(seconds)  # starves heartbeats: expect the connection to drop
            ch.basic_ack(method.delivery_tag)
            return

        def job() -> None:
            time.sleep(seconds)
            conn.add_callback_threadsafe(functools.partial(ack, method.delivery_tag))

        threading.Thread(target=job, daemon=True).start()

    consumer_args = {}
    if args.consumer_timeout:
        consumer_args["x-consumer-timeout"] = args.consumer_timeout
    ch.add_on_cancel_callback(
        lambda _frame: print(" [!] broker cancelled this consumer (consumer timeout); the job was requeued")
    )
    ch.basic_consume(QUEUE, on_message, arguments=consumer_args or None)
    mode = "blocking (broken)" if args.blocking else "threaded"
    print(f" [*] worker running, mode={mode}, heartbeat={HEARTBEAT}s")
    try:
        run_consumer(ch)
    except pika.exceptions.AMQPConnectionError as exc:
        print(f" [!] connection lost: {exc!r}")

python slow_worker.py send --seconds 20             # T2
python slow_worker.py work --blocking               # T1: the wrong way

python slow_worker.py work                          # T1: threaded

python slow_worker.py send --seconds 20
python slow_worker.py work --consumer-timeout 10000

# py/retry.py
def declare(ch) -> None:
    ch.exchange_declare(exchange=DLX, exchange_type="topic", durable=True)
    ch.queue_declare(
        queue=DLQ,
        durable=True,
        arguments={
            "x-queue-type": "quorum",
            # Quorum queues default to a delivery limit of 20; a DLQ you peek at
            # repeatedly should never dead-letter (or drop) its own contents.
            "x-delivery-limit": -1,
        },
    )
    ch.queue_bind(queue=DLQ, exchange=DLX, routing_key="#")

    base = {
        "x-queue-type": "quorum",
        "x-delivery-limit": DELIVERY_LIMIT,
        "x-dead-letter-exchange": DLX,
        # at-least-once requires reject-publish overflow; otherwise RabbitMQ
        # silently falls back to at-most-once dead-lettering.
        "x-dead-letter-strategy": "at-least-once",
        "x-overflow": "reject-publish",
    }
    ch.queue_declare(queue=QUEUE, durable=True, arguments=base)
    ch.queue_declare(
        queue=DELAYED_QUEUE,
        durable=True,
        arguments={
            **base,
            # RabbitMQ 4.3+: hold failed messages aside before redelivery.
            # delay = min(min * delivery_count, max) -> 2s, 4s, 6s ...
            "x-delayed-retry-type": "failed",
            "x-delayed-retry-min": 2000,
            "x-delayed-retry-max": 8000,
        },
    )

python retry.py produce
python retry.py consume          # Ctrl-C once the output settles
python retry.py inspect

python retry.py produce
python retry.py consume --nack

python retry.py produce --delayed
python retry.py consume --delayed

# py/delayed.py
def declare(ch) -> None:
    ch.exchange_declare(exchange=EXCHANGE, exchange_type="direct", durable=True)
    ch.queue_declare(queue=TARGET, durable=True, arguments={"x-queue-type": "quorum"})
    ch.queue_bind(queue=TARGET, exchange=EXCHANGE, routing_key="due")
    for name, ttl in TIERS.items():
        ch.queue_declare(
            queue=f"lab.delay.{name}",
            durable=True,
            arguments={
                "x-queue-type": "quorum",
                "x-message-ttl": ttl,
                "x-dead-letter-exchange": EXCHANGE,
                "x-dead-letter-routing-key": "due",
                "x-dead-letter-strategy": "at-least-once",
                "x-overflow": "reject-publish",
            },
        )

python delayed.py consume                                  # T1
python delayed.py schedule "stand-up in 5" --delay 5s      # T2
python delayed.py schedule "coffee" --delay 30s

# py/priority.py
def queue_for(ch, classic: bool) -> str:
    if classic:
        name = "lab.priority.classic"
        ch.queue_declare(queue=name, durable=True, arguments={"x-queue-type": "classic", "x-max-priority": 10})
    else:
        name = "lab.priority"
        ch.queue_declare(queue=name, durable=True, arguments={"x-queue-type": "quorum"})
    return name

python priority.py publish      # with nothing consuming
python priority.py consume

# py/rpc.py
class RpcClient:
    def __init__(self, timeout: float) -> None:
        self.timeout = timeout
        self.conn = connect("rpc-client")
        self.ch = self.conn.channel()
        self.pending: dict[str, bytes | None] = {}
        # Must consume from the pseudo-queue (auto_ack) BEFORE publishing requests.
        self.ch.basic_consume(REPLY_TO, self._on_reply, auto_ack=True)

    def _on_reply(self, _ch, _method, props, body) -> None:
        if props.correlation_id in self.pending:
            self.pending[props.correlation_id] = body

    def call(self, n: int) -> int:
        corr_id = str(uuid.uuid4())
        self.pending[corr_id] = None
        self.ch.basic_publish(
            "", QUEUE, str(n).encode(),
            pika.BasicProperties(
                reply_to=REPLY_TO,
                correlation_id=corr_id,
                expiration=str(int(self.timeout * 1000)),
            ),
        )
        deadline = time.monotonic() + self.timeout
        while self.pending[corr_id] is None:
            if time.monotonic() > deadline:
                del self.pending[corr_id]
                raise TimeoutError(f"no reply for fib({n}) within {self.timeout}s; is the server running?")
            self.conn.process_data_events(time_limit=0.1)
        return int(self.pending.pop(corr_id))

    def close(self) -> None:
        self.conn.close()

python rpc.py server            # T1
python rpc.py call 10 50 90     # T2

# py/streams.py
def consume(args) -> None:
    conn = connect("stream-consumer")
    ch = conn.channel()
    declare(ch)
    ch.basic_qos(prefetch_count=100)  # required for stream consumers
    received = 0

    def on_message(ch, method, props, body):
        nonlocal received
        offset = (props.headers or {}).get("x-stream-offset")
        print(f" [<] offset={offset} {json.loads(body)}")
        ch.basic_ack(method.delivery_tag)
        received += 1
        if args.limit and received >= args.limit:
            ch.stop_consuming()

    offset = parse_offset(args)
    ch.basic_consume(STREAM, on_message, arguments={"x-stream-offset": offset})
    print(f" [*] reading {STREAM} from offset={offset!s}")
    run_consumer(ch)

python streams.py publish --count 100
python streams.py consume --offset first --limit 5    # replay from the beginning
python streams.py consume --offset 50 --limit 5       # from offset 50
python streams.py consume --offset next               # T1: only new messages
python streams.py publish --count 10                  # T2
python streams.py consume --since 60 --limit 5        # from a point in time

python basics.py produce && python basics.py produce   # leave a backlog, no consumer
make triage

make down && make up-monitoring

# config/prometheus/single.yml
global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - /etc/prometheus/alerts.yml

scrape_configs:
  # Aggregated node/cluster metrics: cheap, safe to scrape often.
  - job_name: rabbitmq
    static_configs:
      - targets: ["rabbitmq:15692"]

  # Per-queue metrics, limited to the metric families the alerts actually use.
  # Scraping every per-object family on a broker with thousands of queues is expensive.
  - job_name: rabbitmq-detailed
    metrics_path: /metrics/detailed
    params:
      family: [queue_coarse_metrics, queue_consumer_count]
    static_configs:
      - targets: ["rabbitmq:15692"]

# config/prometheus/alerts.yml
groups:
  - name: rabbitmq-node
    rules:
      - alert: RabbitMQNodeDown
        expr: up{job="rabbitmq"} == 0
        for: 1m
        labels: {severity: critical}
        annotations:
          summary: "RabbitMQ node {{ $labels.instance }} is not reporting metrics"

      - alert: RabbitMQMemoryAlarm
        expr: rabbitmq_alarms_memory_used_watermark == 1
        for: 1m
        labels: {severity: critical}
        annotations:
          summary: "Memory alarm on {{ $labels.instance }}: publishers are blocked"

      - alert: RabbitMQDiskAlarm
        expr: rabbitmq_alarms_free_disk_space_watermark == 1
        for: 1m
        labels: {severity: critical}
        annotations:
          summary: "Disk alarm on {{ $labels.instance }}: publishers are blocked"

      - alert: RabbitMQFileDescriptorsHigh
        expr: rabbitmq_process_open_fds / rabbitmq_process_max_fds > 0.8
        for: 5m
        labels: {severity: warning}
        annotations:
          summary: "{{ $labels.instance }} is using {{ $value | humanizePercentage }} of its file descriptors"

      # Opening a connection per message is the most common RabbitMQ anti-pattern.
      - alert: RabbitMQConnectionChurn
        expr: rate(rabbitmq_connections_opened_total[5m]) > 10
        for: 10m
        labels: {severity: warning}
        annotations:
          summary: "{{ $labels.instance }} is opening {{ $value | humanize }} connections/s; clients should reuse connections"

      - alert: RabbitMQUnroutableMessages
        expr: rate(rabbitmq_global_messages_unroutable_dropped_total[5m]) > 0
        for: 5m
        labels: {severity: warning}
        annotations:
          summary: "Messages published to {{ $labels.instance }} match no binding and are being dropped"

  - name: rabbitmq-queues
    rules:
      - alert: RabbitMQQueueBacklog
        expr: rabbitmq_detailed_queue_messages_ready > 10000
        for: 10m
        labels: {severity: warning}
        annotations:
          summary: "Queue {{ $labels.vhost }}/{{ $labels.queue }} has {{ $value }} ready messages"

      - alert: RabbitMQQueueNoConsumers
        expr: |
          rabbitmq_detailed_queue_messages_ready > 0
          and on (instance, vhost, queue)
          rabbitmq_detailed_queue_consumers == 0
        for: 10m
        labels: {severity: warning}
        annotations:
          summary: "Queue {{ $labels.vhost }}/{{ $labels.queue }} has messages but no consumers"

      - alert: RabbitMQUnackedHigh
        expr: rabbitmq_detailed_queue_messages_unacked > 1000
        for: 10m
        labels: {severity: warning}
        annotations:
          summary: "Queue {{ $labels.vhost }}/{{ $labels.queue }} has {{ $value }} unacknowledged deliveries (slow or stuck consumers)"

make health
# WARNING: lab.app-logs (vhost /): 8 ready, no consumers

docker compose exec rabbitmq rabbitmqctl set_vm_memory_high_watermark 0.0001
python basics.py produce

docker compose exec rabbitmq rabbitmqctl set_vm_memory_high_watermark 0.6

# py/confirms.py
def run(ch, queue: str, count: int, body: bytes, label: str) -> None:
    ch.queue_purge(queue)
    props = pika.BasicProperties(delivery_mode=PERSISTENT)
    start = time.perf_counter()
    for _ in range(count):
        ch.basic_publish("", queue, body, props)
    elapsed = time.perf_counter() - start
    print(f"  {label:<26} {count / elapsed:>10,.0f} msg/s  ({elapsed:.2f}s)")

python confirms.py --count 5000
python confirms.py --count 5000 --queue-type classic

make perftest
./scripts/perftest.sh single -- --help     # every PerfTest option

./scripts/perftest.sh single -- --quorum-queue --queue perf.qq --producers 4 --consumers 4 \
  --size 4096 --confirm 200 --qos 200 --time 60

make down
make cluster-up
docker compose -f compose.cluster.yaml exec rmq1 rabbitmqctl cluster_status

# config/cluster/rabbitmq.conf
## Cluster-only settings, mounted as /etc/rabbitmq/conf.d/30-cluster.conf
## alongside the shared 20-lab.conf.

cluster_name = rmqlab

## Peer discovery: nodes find each other at boot, no manual join_cluster needed.
## Node names must match the container hostnames (rabbit@<hostname>).
cluster_formation.peer_discovery_backend = classic_config
cluster_formation.classic_config.nodes.1 = rabbit@rmq1
cluster_formation.classic_config.nodes.2 = rabbit@rmq2
cluster_formation.classic_config.nodes.3 = rabbit@rmq3

# config/haproxy.cfg
global
    log stdout format raw local0
    maxconn 4096

# Docker's embedded DNS, so backends resolve even if a node starts late or restarts
# with a new IP.
resolvers docker
    nameserver dns 127.0.0.11:53
    hold valid 10s

defaults
    log global
    mode tcp
    option tcplog
    timeout connect 5s
    # Client/server timeouts MUST exceed the AMQP heartbeat interval (60s here),
    # or HAProxy silently kills idle-but-healthy connections.
    timeout client 3h
    timeout server 3h
    timeout check 5s
    default-server check inter 5s rise 2 fall 3 resolvers docker init-addr last,libc,none

frontend amqp
    bind :5672
    default_backend rabbit_amqp

backend rabbit_amqp
    # Long-lived connections: leastconn spreads them better than roundrobin.
    balance leastconn
    # When a node is marked down, cut its sessions so clients reconnect elsewhere
    # immediately instead of waiting for heartbeats to time out.
    default-server on-marked-down shutdown-sessions
    server rmq1 rmq1:5672
    server rmq2 rmq2:5672
    server rmq3 rmq3:5672

frontend management
    mode http
    option httplog
    timeout client 60s
    bind :15672
    default_backend rabbit_management

backend rabbit_management
    mode http
    timeout server 60s
    balance roundrobin
    option httpchk
    http-check send meth GET uri /
    http-check expect status 200
    server rmq1 rmq1:15672
    server rmq2 rmq2:15672
    server rmq3 rmq3:15672

listen stats
    mode http
    timeout client 60s
    timeout server 60s
    bind :8404
    stats enable
    stats uri /
    stats refresh 5s

alias c1='docker compose -f compose.cluster.yaml exec rmq1'
c1 rabbitmq-queues quorum_status lab.ha
c1 rabbitmqctl list_queues name type leader members online

python failover.py consume            # T1
python failover.py publish            # T2
c1 rabbitmqctl list_queues name leader          # T3: which node leads lab.ha?
docker compose -f compose.cluster.yaml stop rmq2   # stop that node (rmq2 in this example)

c1 rabbitmq-queues check_if_node_is_quorum_critical     # safe to stop rmq1?
docker compose -f compose.cluster.yaml exec rmq2 rabbitmq-upgrade drain
c1 rabbitmqctl list_queues name leader
docker compose -f compose.cluster.yaml exec rmq2 rabbitmq-upgrade revive
c1 rabbitmq-queues rebalance quorum

docker compose -f compose.cluster.yaml stop rmq2 rmq3

order.created ──▶ [payment] ──▶ payment.completed ──▶ [inventory] ──▶ order.completed
                     │                                     │
                     └─▶ payment.failed            inventory.failed
                                                   payment.refund.requested ──▶ [payment] ──▶ payment.refunded

# py/orders.py
def declare(ch) -> None:
    ch.exchange_declare(exchange=EXCHANGE, exchange_type="topic", durable=True)
    ch.exchange_declare(exchange=DLX, exchange_type="fanout", durable=True)
    ch.queue_declare(queue=DEAD, durable=True, arguments={"x-queue-type": "quorum", "x-delivery-limit": -1})
    ch.queue_bind(queue=DEAD, exchange=DLX)
    for queue, keys in QUEUES.items():
        ch.queue_declare(
            queue=queue,
            durable=True,
            arguments={
                "x-queue-type": "quorum",
                "x-delivery-limit": 5,
                "x-dead-letter-exchange": DLX,
                "x-dead-letter-strategy": "at-least-once",
                "x-overflow": "reject-publish",
            },
        )
        for key in keys:
            ch.queue_bind(queue=queue, exchange=EXCHANGE, routing_key=key)

# py/orders.py
class Service:
    """One connection, a consuming channel and a separate confirmed publishing channel."""

    def __init__(self, name: str, queue: str, fail_rate: float) -> None:
        self.name, self.queue, self.fail_rate = name, queue, fail_rate
        self.conn = connect(f"orders-{name}")
        self.consume_ch = self.conn.channel()
        declare(self.consume_ch)
        self.consume_ch.basic_qos(prefetch_count=10)
        self.publish_ch = self.conn.channel()
        self.publish_ch.confirm_delivery()
        self.done: set[tuple[str, str]] = set()  # (routing key, order id) already handled

    def emit(self, routing_key: str, order: dict) -> None:
        order = {**order, "history": [*order.get("history", []), routing_key]}
        self.publish_ch.basic_publish(
            EXCHANGE, routing_key, to_json(order),
            json_props(message_id=str(uuid.uuid4()), correlation_id=order["order_id"]),
            mandatory=True,
        )

    def fails(self) -> bool:
        return random.random() < self.fail_rate

    def run(self, handler) -> None:
        def on_message(ch, method, _props, body):
            try:
                order = json.loads(body)
                key = (method.routing_key, order["order_id"])
            except (json.JSONDecodeError, KeyError):
                print(f" [{self.name}] malformed message, rejecting to the DLQ")
                ch.basic_reject(method.delivery_tag, requeue=False)
                return
            if key in self.done:  # redelivery of something already handled
                ch.basic_ack(method.delivery_tag)
                return
            time.sleep(0.2)  # simulated work
            try:
                handler(method.routing_key, order)
            except Exception as exc:  # unexpected bug or outage: retry, then dead-letter
                print(f" [{self.name}] error on {order['order_id']}: {exc!r}; rejecting for retry")
                ch.basic_reject(method.delivery_tag, requeue=True)
                return
            self.done.add(key)
            ch.basic_ack(method.delivery_tag)  # only after the outcome is confirmed

        self.consume_ch.basic_consume(self.queue, on_message)
        print(f" [{self.name}] consuming {self.queue}")
        run_consumer(self.consume_ch)

python orders.py payment          # T1
python orders.py inventory        # T2
python orders.py notify           # T3
python orders.py create --count 20

make cluster-down      # or: make down
make clean             # both stacks, including volumes