RabbitMQ Confirmed the Publish. Why Is the Queue Empty?

A publisher confirmation is strong evidence about a broker-side publishing boundary. It is not, by itself, evidence that the message reached the queue you intended.

That distinction creates one of the more misleading RabbitMQ troubleshooting states: the publishing call completes under confirm mode, no negative acknowledgment is reported, yet the expected queue is empty. Under AMQP 0-9-1, that combination is entirely possible. RabbitMQ documents that an unroutable publish can still be positively confirmed after the exchange determines that the message routes to no queue. When mandatory=true, RabbitMQ instead returns the unroutable message before issuing its publisher acknowledgment. Publisher confirms and consumer acknowledgments are separate mechanisms. RabbitMQ 4.1, “Consumer Acknowledgements and Publisher Confirms”, versioned documentation with no publication date asserted, accessed September 25, 2026.

This playbook turns that semantic difference into a controlled producer acceptance test. The fixture keeps the exchange, queue, logical message identity and payload constant while varying only the routing key and mandatory mode. It then reconciles four kinds of evidence: intended route, publisher outcome, return evidence and exact queue content.

The laboratory is intentionally narrow: AMQP 0-9-1, Pika BlockingConnection, one loopback broker, one disposable vhost, one direct exchange, one classic queue, no consumers during publication and no external side effects. It tests routing evidence, not replicated durability, exactly-once processing or business completion.

Define the producer-side acceptance boundary

The acceptance question must be stated before publishing:

Did this particular publish attempt establish enough evidence that logical message lab-msg-20260925-001 was routably accepted under the declared topology?

The intended route is:

A direct exchange performs exact routing-key-to-binding-key matching. Therefore work.ok matches this lab's sole binding and work.typo does not. The custom exchange is important: RabbitMQ's default exchange automatically binds declared queues using the queue name as the binding key, which would introduce routing behavior this experiment does not want. RabbitMQ 4.1, “Exchanges”, versioned documentation with no publication date asserted, accessed September 25, 2026.

“Producer acceptance” here stops before consumer work. RabbitMQ explicitly describes publisher confirms and consumer delivery acknowledgments as orthogonal: confirms cover the publisher's interaction with RabbitMQ, while consumer acknowledgments concern deliveries from RabbitMQ to consumers. A confirm therefore must not be rewritten in application language as “the worker processed the job.”

That narrow boundary complements broader backend reliability skills without repeating a general messaging or distributed-systems roadmap. The decision in this playbook is much smaller: accept this publish attempt, repair and republish a specifically classified attempt, or hold it because the outcome remains unresolved.

Create a disposable and fully described broker baseline

A routing experiment is only useful when the environment is frozen tightly enough that an empty queue has a constrained set of explanations.

Use this pinned lab target:

This article does not claim an observed four-scenario broker execution. The research execution environment did not contain an installed RabbitMQ/Erlang/Pika broker stack, so fabricating a successful terminal transcript would violate the acceptance method itself. The matrix below is therefore explicitly labeled documented/derived expectation until the fixture is run against a manifest that proves the pinned environment.

RabbitMQ allows queue type to be selected through the declaration-time x-queue-type argument, and its documentation notes that queue type can otherwise be influenced by default queue-type configuration. Declaring {"x-queue-type": "classic"} prevents an environmental default from silently changing the experiment. RabbitMQ 4.1, “Queues”, accessed September 25, 2026.

The lab is deliberately unrelated to optimizing background work behind API endpoints. It sends one message per independently reset scenario; throughput is not measured.

Freeze the routing topology before publishing

Before every scenario, delete and recreate only the two lab resources. Then record:

• lab.direct: direct, durable, not auto-delete, not internal, empty arguments.

• lab.jobs: classic, durable, non-exclusive, not auto-delete.

• exactly one binding: lab.direct → lab.jobs, routing key work.ok.

• zero matching policies and zero operator policies.

• zero consumers and zero ready messages before publication.

Policies matter because RabbitMQ policies are vhost-scoped and can dynamically inject optional arguments into queues and exchanges, including behavior such as TTLs; queue type itself remains a declaration-time property. A supposedly “clean” routing test is invalid if a policy can expire or redirect evidence. RabbitMQ 4.1, “Policies”, accessed September 25, 2026.

The exchange has no alternate-exchange argument, and the lab requires empty policy sets. That removes alternate routing from the comparison. RabbitMQ otherwise documents that a non-mandatory unroutable message can be discarded or republished to an alternate exchange when one exists. RabbitMQ 4.1, “Publishers”, versioned documentation with no publication date asserted, accessed September 25, 2026.

Build a publisher ledger around message identity

A publisher confirm does not carry your application message_id. AMQP publisher acknowledgments identify publishes using channel-scoped publishing sequence/delivery tags, while message_id is an optional application-provided message property. The application therefore needs its own correlation between logical message identity and each publish attempt.

Keep these identities separate:

The ledger is local JSON Lines opened in append mode, flushed and fsynced after every record. A failed attempt is never rewritten into a successful one. Repair produces another row.

The Pika API shape matters. Pika 1.4.4 BlockingConnection documentation, accessed September 25, 2026, specifies basic_publish(...) -> None; in confirmation mode it raises UnroutableError when a Basic.Return is followed by Basic.Ack, and raises NackError for a broker nack. A Boolean True must therefore not be invented as the confirmation contract.

The complete fixture files below correspond to fixture commit e4d9b6fe9161b9febd79eb27680d1d106b169c70.

requirements.txt:

pika==1.4.4

broker-lab.sh:

#!/usr/bin/env bash
set -euo pipefail

# LAB ONLY. These commands create/delete a dedicated disposable vhost and user.
VHOST='/refonte-2026-09-25-b2'
USER='lab'
PASSWORD='lab-only-password'

case "${1:-}" in
  bootstrap)
    rabbitmqctl add_vhost "$VHOST" \
      --description 'Refonte publisher-routing lab 2026-09-25-batch2'
    rabbitmqctl add_user "$USER" "$PASSWORD"
    rabbitmqctl set_permissions -p "$VHOST" "$USER" '.*' '.*' '.*'
    ;;

  evidence)
    cat /etc/os-release
    uname -a
    python3 --version
    python3 -c 'import pika; print("pika", pika.__version__)'
    rabbitmqctl version
    rabbitmq-diagnostics status
    rabbitmq-plugins list --enabled --minimal
    rabbitmqctl list_policies -p "$VHOST"
    rabbitmqctl list_operator_policies -p "$VHOST"
    rabbitmqctl list_exchanges -p "$VHOST" \
      name type durable auto_delete internal arguments
    rabbitmqctl list_queues -p "$VHOST" \
      name type durable auto_delete arguments \
      consumers messages_ready messages_unacknowledged
    rabbitmqctl list_bindings -p "$VHOST" \
      source_name source_kind destination_name \
      destination_kind routing_key arguments
    ;;

  cleanup)
    # Deletes only the explicitly named disposable vhost and lab user.
    rabbitmqctl delete_vhost "$VHOST"
    rabbitmqctl delete_user "$USER"
    ;;

  *)
    printf 'usage: %s {bootstrap|evidence|cleanup}\n' "$0" >&2
    exit 2
    ;;
esac

RabbitMQ's 4.1 CLI documentation defines list_bindings -p, list_policies -p, and list_queues -p; deleting a vhost deletes the resources within that vhost, which is why the cleanup command is appropriate only for this explicitly disposable namespace. RabbitMQ 4.1 command-line documentation, accessed September 25, 2026.

lab.py:

#!/usr/bin/env python3
"""Disposable RabbitMQ publisher-confirm/routing acceptance lab."""

from future import annotations

import argparse
import hashlib
import json
import os
import platform
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import pika
from pika import exceptions as px

BATCH = "2026-09-25-batch2"
VHOST = "/refonte-2026-09-25-b2"
EXCHANGE = "lab.direct"
QUEUE = "lab.jobs"
BINDING_KEY = "work.ok"
WRONG_KEY = "work.typo"
QUEUE_TYPE = "classic"

MESSAGE_ID = "lab-msg-20260925-001"
PAYLOAD = b'{"job":"inert","message_id":"lab-msg-20260925-001"}'
PAYLOAD_SHA256 = hashlib.sha256(PAYLOAD).hexdigest()

LEDGER = Path("artifacts/ledger.jsonl")
SNAPSHOTS = Path("artifacts/snapshots.jsonl")

HOST = os.environ.get("LAB_AMQP_HOST", "127.0.0.1")
PORT = int(os.environ.get("LAB_AMQP_PORT", "5672"))
USER = os.environ.get("LAB_AMQP_USER", "lab")
PASSWORD = os.environ.get("LAB_AMQP_PASSWORD", "lab-only-password")

EXIT_ACCEPT = 0
EXIT_UNROUTABLE = 20
EXIT_NACK = 21
EXIT_UNKNOWN = 22
EXIT_INTERFERENCE = 23
EXIT_CHANNEL_ERROR = 24
EXIT_SETUP = 25


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def append_jsonl(path: Path, record: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    line = json.dumps(
        record, sort_keys=True, separators=(",", ":")
    ) + "\n"
    with path.open("a", encoding="utf-8") as f:
        f.write(line)
        f.flush()
        os.fsync(f.fileno())


def params() -> pika.ConnectionParameters:
    return pika.ConnectionParameters(
        host=HOST,
        port=PORT,
        virtual_host=VHOST,
        credentials=pika.PlainCredentials(
            USER, PASSWORD, erase_on_connect=False
        ),
        heartbeat=30,
        blocked_connection_timeout=10,
        connection_attempts=1,
        retry_delay=0,
        socket_timeout=5,
    )


def connect() -> pika.BlockingConnection:
    return pika.BlockingConnection(params())


def declare_topology(ch) -> None:
    ch.exchange_declare(
        exchange=EXCHANGE,
        exchange_type="direct",
        durable=True,
        auto_delete=False,
        internal=False,
        arguments={},
    )
    ch.queue_declare(
        queue=QUEUE,
        durable=True,
        exclusive=False,
        auto_delete=False,
        arguments={"x-queue-type": QUEUE_TYPE},
    )
    ch.queue_bind(
        queue=QUEUE,
        exchange=EXCHANGE,
        routing_key=BINDING_KEY,
    )


def safe_delete_known_topology() -> None:
    """Delete only the fixed lab objects; tolerate first-run absence."""
    for kind in ("queue", "exchange"):
        conn = None
        try:
            conn = connect()
            ch = conn.channel()
            if kind == "queue":
                ch.queue_delete(
                    queue=QUEUE,
                    if_unused=False,
                    if_empty=False,
                )
            else:
                ch.exchange_delete(
                    exchange=EXCHANGE,
                    if_unused=False,
                )
        except px.ChannelClosedByBroker as exc:
            if exc.reply_code != 404:
                raise
        finally:
            if conn is not None and conn.is_open:
                conn.close()


def queue_snapshot(ch, phase: str) -> dict[str, Any]:
    frame = ch.queue_declare(queue=QUEUE, passive=True)
    snap = {
        "ts": now_iso(),
        "batch": BATCH,
        "phase": phase,
        "exchange": EXCHANGE,
        "queue": QUEUE,
        "queue_type": QUEUE_TYPE,
        "binding_key": BINDING_KEY,
        "message_count": int(frame.method.message_count),
        "consumer_count": int(frame.method.consumer_count),
    }
    append_jsonl(SNAPSHOTS, snap)
    return snap


def setup_or_reset(action: str) -> int:
    try:
        safe_delete_known_topology()
        with connect() as conn:
            ch = conn.channel()
            declare_topology(ch)
            snap = queue_snapshot(ch, action)
            if (
                snap["consumer_count"] != 0
                or snap["message_count"] != 0
            ):
                return EXIT_INTERFERENCE
        return EXIT_ACCEPT
    except (px.AMQPConnectionError, px.AMQPChannelError) as exc:
        append_jsonl(
            SNAPSHOTS,
            {
                "ts": now_iso(),
                "batch": BATCH,
                "phase": action,
                "status": "setup_error",
                "exception_type": type(exc).__name__,
                "exception": repr(exc),
            },
        )
        return EXIT_SETUP


def inspect(phase: str = "inspect") -> int:
    try:
        with connect() as conn:
            ch = conn.channel()
            snap = queue_snapshot(ch, phase)
        print(json.dumps(snap, sort_keys=True))
        return (
            EXIT_ACCEPT
            if snap["consumer_count"] == 0
            else EXIT_INTERFERENCE
        )
    except (px.AMQPConnectionError, px.AMQPChannelError) as exc:
        print(repr(exc), file=sys.stderr)
        return EXIT_UNKNOWN


def serialize_returned(exc: Exception) -> list[dict[str, Any]]:
    result = []
    for returned in getattr(exc, "messages", ()):
        result.append(
            {
                "reply_code": getattr(
                    returned.method, "reply_code", None
                ),
                "reply_text": getattr(
                    returned.method, "reply_text", None
                ),
                "exchange": getattr(
                    returned.method, "exchange", None
                ),
                "routing_key": getattr(
                    returned.method, "routing_key", None
                ),
                "message_id": getattr(
                    returned.properties, "message_id", None
                ),
                "body_sha256": hashlib.sha256(
                    returned.body
                ).hexdigest(),
                "body_utf8": returned.body.decode(
                    "utf-8", errors="replace"
                ),
            }
        )
    return result


def publish(
    routing_key: str,
    mandatory: bool,
    exchange: str = EXCHANGE,
) -> int:
    attempt_id = str(uuid.uuid4())

    record: dict[str, Any] = {
        "ts_started": now_iso(),
        "batch": BATCH,
        "logical_message_id": MESSAGE_ID,
        "publish_attempt_id": attempt_id,
        "payload_sha256": PAYLOAD_SHA256,
        "payload_utf8": PAYLOAD.decode("utf-8"),
        "exchange": exchange,
        "intended_exchange": EXCHANGE,
        "routing_key": routing_key,
        "intended_routing_key": BINDING_KEY,
        "mandatory": mandatory,
        "publisher_confirms": True,
        "client_api": "pika.BlockingChannel.basic_publish",
        "client_return_contract":
            "None on normal completion; exceptions carry failure evidence",
    }

    conn = None

    try:
        conn = connect()
        ch = conn.channel()

        before = queue_snapshot(ch, f"before:{attempt_id}")
        if before["consumer_count"] != 0:
            record["api_outcome"] = (
                "fixture_interference_consumers_present"
            )
            record["queue_before"] = before
            return EXIT_INTERFERENCE

        ch.confirm_delivery()

        properties = pika.BasicProperties(
            content_type="application/json",
            delivery_mode=1,
            message_id=MESSAGE_ID,
            headers={
                "publish_attempt_id": attempt_id,
                "commissioning_batch": BATCH,
            },
        )

        api_return = ch.basic_publish(
            exchange=exchange,
            routing_key=routing_key,
            body=PAYLOAD,
            properties=properties,
            mandatory=mandatory,
        )

        # Pika documents -> None. It is not an ack Boolean.
        record["api_return_type"] = type(api_return).__name__
        record["api_return_repr"] = repr(api_return)
        record["api_outcome"] = "confirmed_no_exception"
        record["routing_evidence"] = (
            "no_basic_return_exposed_by_client"
        )

        after = queue_snapshot(ch, f"after:{attempt_id}")
        record["queue_before"] = before
        record["queue_after"] = after

        return EXIT_ACCEPT

    except px.UnroutableError as exc:
        record["api_outcome"] = (
            "unroutable_returned_then_acked"
        )
        record["exception_type"] = type(exc).__name__
        record["exception"] = repr(exc)
        record["returned_messages"] = serialize_returned(exc)
        return EXIT_UNROUTABLE

    except px.NackError as exc:
        record["api_outcome"] = "nacked"
        record["exception_type"] = type(exc).__name__
        record["exception"] = repr(exc)
        record["returned_messages"] = serialize_returned(exc)
        return EXIT_NACK

    except px.ChannelClosedByBroker as exc:
        record["api_outcome"] = "channel_closed_by_broker"
        record["exception_type"] = type(exc).__name__
        record["reply_code"] = exc.reply_code
        record["reply_text"] = exc.reply_text
        record["exception"] = repr(exc)
        return EXIT_CHANNEL_ERROR

    except (
        px.AMQPConnectionError,
        px.ConnectionWrongStateError,
    ) as exc:
        record["api_outcome"] = (
            "unknown_transport_or_connection_outcome"
        )
        record["exception_type"] = type(exc).__name__
        record["exception"] = repr(exc)
        return EXIT_UNKNOWN

    except px.AMQPChannelError as exc:
        record["api_outcome"] = "channel_error_unclassified"
        record["exception_type"] = type(exc).__name__
        record["exception"] = repr(exc)
        return EXIT_CHANNEL_ERROR

    finally:
        record["ts_finished"] = now_iso()
        append_jsonl(LEDGER, record)

        if conn is not None and conn.is_open:
            try:
                conn.close()
            except px.AMQPError:
                pass

        print(json.dumps(record, sort_keys=True))


def drain() -> int:
    received = []

    try:
        with connect() as conn:
            ch = conn.channel()

            before = queue_snapshot(ch, "drain-before")
            if before["consumer_count"] != 0:
                return EXIT_INTERFERENCE

            while True:
                method, properties, body = ch.basic_get(
                    queue=QUEUE,
                    auto_ack=False,
                )

                if method is None:
                    break

                body = body or b""

                item = {
                    "ts": now_iso(),
                    "batch": BATCH,
                    "phase": "drain-received-before-ack",
                    "delivery_tag": method.delivery_tag,
                    "redelivered": method.redelivered,
                    "message_id": (
                        properties.message_id
                        if properties
                        else None
                    ),
                    "publish_attempt_id": (
                        (properties.headers or {}).get(
                            "publish_attempt_id"
                        )
                        if properties
                        else None
                    ),
                    "body_sha256": hashlib.sha256(
                        body
                    ).hexdigest(),
                    "body_utf8": body.decode(
                        "utf-8", errors="replace"
                    ),
                    "identity_match": bool(
                        properties
                        and properties.message_id == MESSAGE_ID
                        and body == PAYLOAD
                    ),
                }

                append_jsonl(SNAPSHOTS, item)

                # Ack only after local receipt evidence is persisted.
                ch.basic_ack(
                    delivery_tag=method.delivery_tag
                )
                received.append(item)

            after = queue_snapshot(ch, "drain-after")

        summary = {
            "received": received,
            "before": before,
            "after": after,
        }
        print(json.dumps(summary, sort_keys=True))

        if not all(
            item["identity_match"] for item in received
        ):
            return EXIT_INTERFERENCE

        return EXIT_ACCEPT

    except (
        px.AMQPConnectionError,
        px.AMQPChannelError,
    ) as exc:
        print(repr(exc), file=sys.stderr)
        return EXIT_UNKNOWN


def environment_manifest() -> int:
    manifest = {
        "ts": now_iso(),
        "batch": BATCH,
        "os": platform.platform(),
        "python": platform.python_version(),
        "pika": pika.__version__,
        "connection": {
            "host": HOST,
            "port": PORT,
            "vhost": VHOST,
            "user": USER,
            "heartbeat_s": 30,
            "blocked_connection_timeout_s": 10,
            "connection_attempts": 1,
            "socket_timeout_s": 5,
        },
        "topology": {
            "exchange": EXCHANGE,
            "exchange_type": "direct",
            "exchange_durable": True,
            "exchange_auto_delete": False,
            "exchange_arguments": {},
            "queue": QUEUE,
            "queue_type": QUEUE_TYPE,
            "queue_durable": True,
            "queue_exclusive": False,
            "queue_auto_delete": False,
            "binding_keys": [BINDING_KEY],
        },
        "note":
            "Capture RabbitMQ/Erlang/plugins/policies "
            "with broker-lab.sh evidence.",
    }

    append_jsonl(
        SNAPSHOTS,
        {"phase": "environment-manifest", **manifest},
    )
    print(json.dumps(manifest, indent=2, sort_keys=True))
    return EXIT_ACCEPT


def matrix() -> int:
    scenarios = [
        (
            "correct_optional",
            BINDING_KEY,
            False,
            EXIT_ACCEPT,
            1,
        ),
        (
            "typo_optional",
            WRONG_KEY,
            False,
            EXIT_ACCEPT,
            0,
        ),
        (
            "typo_mandatory",
            WRONG_KEY,
            True,
            EXIT_UNROUTABLE,
            0,
        ),
        (
            "correct_mandatory",
            BINDING_KEY,
            True,
            EXIT_ACCEPT,
            1,
        ),
    ]

    lab_failures = 0

    for (
        name,
        key,
        mandatory,
        expected_publish_exit,
        expected_received,
    ) in scenarios:

        reset_rc = setup_or_reset(f"reset:{name}")
        if reset_rc != EXIT_ACCEPT:
            lab_failures += 1
            continue

        pub_rc = publish(key, mandatory)
        inspect_rc = inspect(f"inspect:{name}")

        with connect() as conn:
            ch = conn.channel()
            predrain = queue_snapshot(
                ch, f"predrain:{name}"
            )

        drain_rc = drain()

        scenario_ok = (
            pub_rc == expected_publish_exit
            and inspect_rc == EXIT_ACCEPT
            and predrain["consumer_count"] == 0
            and predrain["message_count"]
                == expected_received
            and drain_rc == EXIT_ACCEPT
        )

        append_jsonl(
            SNAPSHOTS,
            {
                "ts": now_iso(),
                "batch": BATCH,
                "phase": "scenario-summary",
                "scenario": name,
                "routing_key": key,
                "mandatory": mandatory,
                "publish_exit": pub_rc,
                "expected_publish_exit":
                    expected_publish_exit,
                "predrain_message_count":
                    predrain["message_count"],
                "expected_received_count":
                    expected_received,
                "lab_expectation_met": scenario_ok,
            },
        )

        if not scenario_ok:
            lab_failures += 1

    return (
        EXIT_ACCEPT
        if lab_failures == 0
        else EXIT_INTERFERENCE
    )


def parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser()
    sub = p.add_subparsers(dest="cmd", required=True)

    sub.add_parser("env")
    sub.add_parser("setup")
    sub.add_parser("reset")

    ins = sub.add_parser("inspect")
    ins.add_argument("--phase", default="inspect")

    pub = sub.add_parser("publish")
    pub.add_argument("--routing-key", required=True)
    pub.add_argument("--mandatory", action="store_true")
    pub.add_argument("--exchange", default=EXCHANGE)

    sub.add_parser("drain")
    sub.add_parser("matrix")
    return p


def main() -> int:
    args = parser().parse_args()

    if args.cmd == "env":
        return environment_manifest()

    if args.cmd in {"setup", "reset"}:
        return setup_or_reset(args.cmd)

    if args.cmd == "inspect":
        return inspect(args.phase)

    if args.cmd == "publish":
        return publish(
            args.routing_key,
            args.mandatory,
            args.exchange,
        )

    if args.cmd == "drain":
        return drain()

    if args.cmd == "matrix":
        return matrix()

    return 2


if name == "__main__":
    raise SystemExit(main())

The connection is deliberately 127.0.0.1:5672, one connection attempt, a 30-second heartbeat, five-second socket timeout and dedicated vhost. The password in the fixture is a synthetic lab credential, not a production-secret pattern.

Confirm a correctly routed control message

Bootstrap only on the owned disposable broker:

python3 -m venv .venv
. .venv/bin/activate
python -m pip install --requirement requirements.txt

./broker-lab.sh bootstrap

export LAB_AMQP_HOST=127.0.0.1
export LAB_AMQP_PORT=5672
export LAB_AMQP_USER=lab
export LAB_AMQP_PASSWORD=lab-only-password

python lab.py env
./broker-lab.sh evidence
python lab.py reset

Do not advance if the evidence command shows an unexpected plugin, policy, operator policy, binding, queue argument or consumer. The intended pre-publish state is one empty classic queue and one work.ok binding.

The first control is:

python lab.py publish --routing-key work.ok
rc=$?
printf 'publish_exit=%s\n' "$rc"

python lab.py inspect --phase correct-optional

--mandatory is absent, so it is False. The derived expectation, not an observed result, is exit 0, api_outcome="confirmed_no_exception" and one ready message.

Why is that expectation justified? A direct exchange matches work.ok exactly to the sole work.ok binding, and RabbitMQ states that successfully routed AMQP 0-9-1 messages are stored in queues. Publisher confirms then report broker handling at the publisher boundary.

Do not interpret the Python return object as confirmation data. Pika 1.4.4 documents the return type of basic_publish as None; its blocking implementation waits for publisher-confirmation state and uses exceptions for nack/returned-message conditions.

Verify content, not only the queue counter

A depth of one answers “one ready message exists,” not “our message exists.”

Perform the controlled drain:

python lab.py drain

The fixture uses basic_get(..., auto_ack=False), records message_id, publish_attempt_id, exact body text and SHA-256 digest, and only then calls basic_ack on that delivery tag. Pika documents basic_get as returning (None, None, None) when the queue is empty, while RabbitMQ's acknowledgment rules make the delivery tag channel-scoped.

For this control to pass, the received item must contain:

message_id = lab-msg-20260925-001
body       = {"job":"inert","message_id":"lab-msg-20260925-001"}
identity_match = true

The post-drain queue depth must be zero.

That evidence is stronger than a counter but remains local to the fixture. It does not establish durable replication, downstream business processing or behavior under a different topology.

Reproduce the acknowledged but unroutable case

Reset completely before changing the routing key:

python lab.py reset
python lab.py publish --routing-key work.typo
rc=$?
python lab.py inspect --phase typo-optional
python lab.py drain

Only two intended experimental variables differ from the previous publish: routing key work.typo instead of work.ok, while mandatory remains False. Exchange, queue, payload and logical message_id are unchanged. The reset also creates a new publish_attempt_id.

The expected result is the heart of the acceptance test:

RabbitMQ 4.1 explicitly documents both sides of this result. When an AMQP 0-9-1 message cannot route to any queue and mandatory=false, it is discarded unless an alternate exchange exists. Separately, the confirm documentation says the broker issues a publisher confirmation for an unroutable message after the exchange determines that it routes to no queue.

That means a positive publisher acknowledgment and an empty expected queue are not contradictory.

What did RabbitMQ acknowledge? Under the documented semantics, it acknowledged its handling of the publish after routing evaluation. It did not assert that lab.jobs received the message. With mandatory=false, no return is required to tell this blocking caller that routing produced no destination.

This case is why “publisher confirm received” cannot be the sole application acceptance predicate when route existence matters.

It is also why a real unexplained empty queue must not immediately be labeled “RabbitMQ silently discarded it.” That conclusion is justified in this fixture only because the topology is reconstructed, alternate routing is excluded, consumers are absent and expiry/purge interference is checked. Outside those controls, disappearance after routing is a separate hypothesis.

Enable mandatory and capture the returned publish

Now reset and publish the same body with the same wrong key but set mandatory=True:

python lab.py reset

python lab.py publish \
  --routing-key work.typo \
  --mandatory

printf 'publish_exit=%s\n' "$?"
python lab.py inspect --phase typo-mandatory
python lab.py drain

The documented/derived expectation is exit 20, api_outcome="unroutable_returned_then_acked", a serialized returned message whose message_id and body digest match the attempted publish, and zero messages in lab.jobs.

RabbitMQ documents that mandatory=true causes an AMQP 0-9-1 message that cannot route to any queue to be returned to its publisher. Its confirm documentation adds an ordering rule: for such an unroutable mandatory message, basic.return is sent before the publisher basic.ack.

Pika then converts that protocol sequence into blocking-client behavior. Its 1.4.4 documentation says UnroutableError is raised for a publish in confirmation mode that is returned by Basic.Return and then acknowledged by Basic.Ack. Its exception type contains the returned messages; Pika's documented source represents each returned item with method, properties and body fields.

That is a positive classification of this attempt: the publish was unroutable under the frozen topology. It fails the application's routing gate even though the protocol sequence also contains a positive publisher acknowledgment.

This is the attempt eligible for controlled routing repair.

Keep return frames and library exceptions distinct

There are two layers of evidence:

Do not write code expecting something such as:

result = channel.basic_publish(...)
if result.basic_return:
    ...

That is not the Pika blocking API contract. basic_publish() is documented as returning None; its confirmation-mode failure evidence is surfaced through exceptions.

Likewise, this article contains no packet trace and claims none. A frame capture could be added as separate optional diagnostic evidence, but the documented ordering must not be presented as though Wireshark, tracing or broker logs were actually collected for this article.

Separate missing bindings from channel failures

A wrong routing key against an existing exchange is not the same error as publishing to an exchange that does not exist.

RabbitMQ's publisher documentation distinguishes them explicitly: an unroutable message sent to an existing exchange follows the mandatory/alternate-exchange rules, whereas publishing to a nonexistent exchange produces a channel error and closes that channel.

Run the negative control separately from the four primary scenarios:

python lab.py reset

python lab.py publish \
  --exchange lab.missing \
  --routing-key work.ok \
  --mandatory

printf 'publish_exit=%s\n' "$?"

The derived expectation is exit 24, with api_outcome="channel_closed_by_broker" and broker reply information in the ledger. Pika documents ChannelClosedByBroker as an exception raised by BlockingConnection when the broker closes the channel.

Do not reuse that channel. The fixture's next command establishes a new connection and channel. That separation matters because RabbitMQ states that further publishing cannot continue on the closed channel.

Operationally, keep the classifications distinct:

A routing typo and a missing exchange may both result in “no message in lab.jobs,” but their evidence and recovery preconditions are different.

Run the complete routing comparison matrix

After environment capture, run:

python lab.py matrix
printf 'matrix_exit=%s\n' "$?"

The matrix performs a destructive reset only inside the dedicated lab vhost before every primary scenario, publishes one message, records a queue snapshot, drains under manual acknowledgments and writes a scenario summary.

No row below is labeled observed. These values are independently derived from the RabbitMQ 4.1 and Pika 1.4.4 semantics already cited:

For the two work.ok controls, exact direct-exchange matching provides the route. For work.typo, no binding matches. For the mandatory typo, RabbitMQ's documented return behavior and Pika's exception mapping expose that route failure to the caller.

The four rows must have independent reset records. Never use the correct-control message left in a queue as receipt evidence for a later scenario.

Detect fixture interference before blaming RabbitMQ

Invalidate the comparison instead of interpreting it when any of these conditions appears:

• consumer_count is nonzero before or immediately after publishing.

• a policy or operator policy appears.

• an extra binding exists.

• queue arguments contain an unexpected TTL, expiry or length constraint.

• the queue is nonempty after reset.

• someone ran a purge outside the recorded fixture.

• the received message_id, body or digest differs.

• an exchange or queue declaration differs from the baseline.

RabbitMQ supports queue/message TTLs and policy-controlled queue behavior, and its CLI exposes binding, policy and queue state. Those mechanisms are legitimate features but confound this particular routing experiment.

A zero queue depth is therefore evidence only of current depth. It is never a substitute for message identity or provenance.

Explain what mandatory still cannot establish

mandatory=true answers a narrower question than many application names imply.

Under AMQP 0-9-1, RabbitMQ returns a mandatory message when it cannot route that message to any queue. Therefore, when a mandatory publish is confirmed without a return under a stable topology, the producer has evidence that the exchange found at least one routing destination accepted under the broker's confirmation rules.

It does not generally identify which queue.

In this deliberately restricted lab, that distinction is manageable because lab.direct has exactly one binding and no alternate exchange. With those facts frozen, work.ok has only one intended queue route. In a real topology with several matching bindings, exchange-to-exchange routes or alternate routing, “not returned” would not mean “the specific queue I had in mind received the only copy.” RabbitMQ direct exchanges can route to one or more matching destinations.

Mandatory also says nothing about completed application work. Publisher confirms are unaware of consumers, and consumer acknowledgments are a different boundary.

That separation is analogous to maintaining clear database integration boundaries: evidence about one named operation should not be silently promoted into proof about a later operation.

Nor does this test establish durable replication. The fixture deliberately uses delivery_mode=1, and although its exchange and queue declarations are durable, persistence and queue durability are outside the acceptance question. RabbitMQ documents delivery mode as a separate message property.

The safe statement is therefore:

Under this frozen single-binding lab topology, the mandatory/confirm result supplies producer-side routing evidence; exact controlled receipt supplies queue-content evidence. Neither is evidence of business completion.

Classify connection failures as unresolved outcomes

A lost connection is not a basic.return, and a timeout is not a basic.nack.

RabbitMQ warns that writing a publish to a socket does not by itself establish that the broker received and processed it: networks can fail or delay communication in ways that leave the publisher uncertain. Publisher confirms exist partly to resolve that boundary when the confirmation is actually received.

The fixture therefore maps connection exceptions to:

api_outcome = unknown_transport_or_connection_outcome
exit         = 22
decision     = HOLD

That category deliberately does not contain either of these assertions:

message_absent = true
broker_rejected = true

Both could be wrong. The broker may have accepted the publish while the connection failed before the client observed its confirmation; alternatively, the publish may never have reached the broker. Without additional evidence, the attempt remains unresolved.

The playbook does not publish a fake deterministic “network failure” transcript. Killing a loopback connection at exactly the right point races the client, kernel and broker, so one run can produce a different boundary than another. The important acceptance requirement is deterministic classification after whatever exception is genuinely observed.

A NackError is different. RabbitMQ documents basic.nack as an exceptional broker response indicating that the broker was unable to process the publish and refuses responsibility for it; Pika maps that response to NackError.

Therefore:

UnroutableError -> known routing failure
NackError       -> known negative broker outcome
connection loss -> unknown outcome

Those three records must never collapse into a generic publish_failed=true.

Do not turn missing confirmation into proof of loss

A real system deciding whether to replay an unknown attempt would need evidence outside this lab's scope, such as destination inspection correlated by a stable logical message identity or an application-level record capable of showing whether the intended operation already exists.

That is not permission to claim exactly-once behavior. Reissuing an unknown publish can create another delivery if the first attempt succeeded but its confirmation was lost. Conversely, refusing to retry can leave work absent if the original attempt never arrived.

This article intentionally stops before deduplication design, transactional outboxes and consumer-side retry architecture. Its rule is simpler:

Missing confirmation means hold the attempt as unknown until another authoritative source resolves it.

The ledger preserves the original publish_attempt_id precisely so a later decision cannot erase that uncertainty.

Repair routing and republish only a classified attempt

Repair is safest when the failed attempt has positive failure evidence.

For the mandatory typo scenario, Pika's UnroutableError is that evidence. The returned-message record should reconcile:

logical_message_id  = lab-msg-20260925-001
routing_key         = work.typo
mandatory           = true
returned message ID = lab-msg-20260925-001
returned digest     = original payload digest

The fixture's topology is intentionally defined to keep only work.ok bound, so repair the publisher key, not the exchange, for this exercise:

python lab.py publish \
  --routing-key work.ok \
  --mandatory

python lab.py inspect --phase repaired-attempt
python lab.py drain

The logical identity remains:

lab-msg-20260925-001

but publish() generates a new UUID for publish_attempt_id.

That distinction is essential. The application is not pretending that the original failed attempt suddenly succeeded. It is recording a second attempt for the same logical work item.

The ledger should consequently show something conceptually like:

Do not delete UUID-A after UUID-B succeeds. Its continued existence proves why the republish occurred.

The non-mandatory typo is more awkward: the confirm alone did not provide a return identifying unroutability. Even though this tightly controlled lab topology lets us derive what occurred, an operational recovery system should not generalize that result into “confirmed but missing means blindly retry.” Without positive failure classification and interference controls, the safer state is hold.

Turn the evidence into an automated routing gate

The fixture uses explicit process exit codes so CI or a release gate does not have to scrape prose:

There is an important testing distinction: scenario three expects exit 20, so the test passes when the application routing gate correctly fails. An automated comparison that insists every publish return zero would erase the most valuable negative control.

Archive together:

broker/OS/Erlang/Python/Pika manifest
enabled-plugin output
policy and operator-policy output
exchange/queue/binding snapshots
ledger.jsonl
snapshots.jsonl
publish attempt IDs
message IDs and body digests
client exception classes and texts
controlled-drain records
matrix scenario summaries
fixture commit

That evidence is more useful than a success log saying only “publish completed.” It also fits naturally beside broader API security and observability: the operational record should preserve the identity and boundary of the event being asserted, rather than converting one signal into an end-to-end claim.

Any divergence between documentation and a genuine reproduction is a failed acceptance condition. For example, if the pinned Pika build does not raise UnroutableError for the mandatory typo despite a verified frozen topology, preserve the ledger, broker metadata and discrepancy and move the attempt to hold. Do not rewrite the expected table to make the run appear clean.

Assign producer and topology ownership

Acceptance should have two evidence owners.

The producer owner signs the application-side record: logical message_id, payload digest, attempt ID, actual exchange/key, mandatory setting, confirmation mode, exact Pika outcome and returned-message identity when present.

The topology owner signs the broker-side fixture state: vhost, exchange properties, queue type and properties, sole binding, empty policies, plugin state, zero consumers and pre/post queue snapshots.

That creates a compact decision table:

This ownership model is a focused form of operational evidence and observability: the goal is not more telemetry for its own sake, but enough independent evidence to say exactly what boundary has been crossed.

The test remains one broker, one vhost and one deliberately trivial route. It makes no performance, clustering, replication, durability or general message-delivery guarantee.

Require positive and negative controls before release

A release-quality routing gate needs both sides of the behavior.

The positive controls are:

work.ok + mandatory=False -> exact queue receipt expected
work.ok + mandatory=True  -> exact queue receipt expected

The negative controls are:

work.typo + mandatory=False -> confirm may coexist with empty queue
work.typo + mandatory=True  -> UnroutableError expected

The separate nonexistent-exchange control proves that a missing routing binding has not been confused with a missing publishing destination.

An all-green happy-path test is insufficient. It demonstrates only that a correct key can work. The acceptance requirement is stronger: the fixture must also demonstrate that the application rejects a positively identified unroutable mandatory attempt and does not interpret a non-mandatory confirmation as proof of queue receipt.

After the evidence is archived, cleanup is explicit and lab-scoped:

./broker-lab.sh cleanup

RabbitMQ documents that deleting the disposable vhost removes its exchanges, queues, bindings, permissions, parameters and policies. Never substitute a production vhost in this command.

Develop backend integration foundations with Refonte Learning

The routing discipline in this playbook sits inside a broader backend skill set: naming system boundaries precisely, testing failure paths, preserving evidence and refusing to infer downstream completion from an upstream success signal.

Refonte Learning's Backend Development programme page describes a three-month programme at 10–12 hours per week. Its listed competencies include Node.js and Express, MongoDB and SQL, RESTful APIs and microservices, authentication and authorization, testing/debugging, and Docker/cloud deployment. Those are programme-page descriptions rather than independent guarantees of professional outcomes.

The same page says successful completion leads to a Training Certificate and a Certificate of Internship.

For engineers developing backend integration foundations, the useful connection is methodological: a service boundary is only as reliable as the evidence used to accept it. In RabbitMQ, that means distinguishing publisher acknowledgment, routing evidence, queue identity and consumer completion rather than treating “publish succeeded” as one undifferentiated fact.