Build a RabbitMQ Order Pipeline

Introduction

30 Second Summary

A checkout can confirm an order in seconds while payment or fulfillment takes much longer. A worker failure during that wait can leave a customer charged twice.

In this project, you will build a local order pipeline with RabbitMQ that separates checkout from payment and fulfillment. You will crash a payment worker on purpose to expose duplicate processing before fixing it with idempotency.

What You'll Build

You'll watch one checkout order move across three terminal windows from queued payment to completed fulfillment, even after the payment worker crashes.

By the end of this project, you'll have:

• A checkout producer that returns a visible QUEUED result while the order waits safely for payment.
• A two-stage order pipeline that routes each order from payment to fulfillment. Queue counts show the message moving between independent workers.
• A failure-tested recovery flow that reproduces message redelivery after a worker crash. SQLite-backed duplicate suppression keeps payment and fulfillment side effects to one each.
• Secret Mission: Route rejected payments into a dead-letter queue and inspect the quarantined order.

Are there any prerequisites?

You'll need Docker Desktop, Python 3.7 or newer, and Visual Studio Code installed. Everything runs locally with no incremental cost.

Before We Start

Before any hands-on work, lock in the reliability question that guides every later decision. You are building an asynchronous e-commerce checkout pipeline where duplicate-safe recovery protects an unacknowledged order that RabbitMQ redelivers after a worker crash.

Set Up the Local Broker and Python Environment

Package conflicts can blur the delivery behavior you are trying to study. An inconsistent RabbitMQ broker can do the same.

In this step, you'll isolate Pika in a Python virtual environment. A pinned RabbitMQ container in Docker Desktop gives every later reliability test the same starting point.

In this step, get ready to:

• Prepare an isolated Python workspace in Visual Studio Code.
• Install Pika 1.4.4 from requirements.txt.
• Run RabbitMQ 4.3.5 in a local container.

Prepare the isolated workspace

A virtual environment keeps this project's Python packages inside .venv. That boundary protects the setup from packages installed elsewhere on your Mac.

• Press Cmd+Space to open Spotlight.
• Type Terminal into Spotlight.
• Press Return to open Terminal.

You'll see a command prompt where you can create the project in a known location.

• Create the ecommerce-order-pipeline folder on your Desktop by running these commands:

cd ~/Desktop
mkdir ecommerce-order-pipeline
cd ecommerce-order-pipeline
pwd

What do these commands do?

• The first command moves your terminal to the Desktop.
• The second command creates the ecommerce-order-pipeline folder.
• The third command moves your terminal into the new folder.
• The final command prints the folder's location so you can confirm where you are.

The final output should end with /Desktop/ecommerce-order-pipeline.

Folder already exists?

If Terminal says the folder already exists, check your Desktop for an earlier ecommerce-order-pipeline folder. Remove that folder only if you are sure it contains no work you need.

Still stuck? Help me create the project folder on my Desktop.

• Press Cmd+Space to return to Spotlight.
• Type Visual Studio Code into Spotlight.
• Press Return to open Visual Studio Code.

Visual Studio Code opens in its own window.

• Click File in the menu bar.
• Select Open Folder.
• Select the ecommerce-order-pipeline folder on your Desktop.
• Click Open.

Your project now has a clear home on the Desktop. You should see ecommerce-order-pipeline at the top of the Explorer sidebar.

• Click View in the menu bar.
• Select Terminal.

The integrated terminal opens inside ecommerce-order-pipeline.

• Create the virtual environment inside the project folder by running these commands:

python3 -m venv .venv
source .venv/bin/activate

Why use a virtual environment?

The first command creates an isolated Python environment in .venv. The second command activates that environment for the current terminal.

Every package you install next stays attached to this project.

• Confirm the terminal prompt now includes .venv.
• Confirm the .venv folder appears in the Explorer sidebar.

Virtual environment not active?

Make sure the integrated terminal is inside the ecommerce-order-pipeline folder. A different location creates .venv in the wrong place.

Need help? Help me activate my macOS virtual environment.

Install the pinned Pika dependency

A dependency pin gives the project one known client-library version. Your requirements.txt file records that version so the environment can be recreated later.

• Right-click the ecommerce-order-pipeline folder in the Explorer sidebar.
• Select New File.
• Type requirements.txt as the file name.
• Press Return to create the file.
• Add the pinned dependency to requirements.txt by entering this line:

pika==1.4.4

What does this dependency pin do?

The double equals sign tells the package installer to use exactly Pika 1.4.4. This keeps the client behavior consistent throughout the project.

• Save requirements.txt.
• Confirm the file contains exactly one line with the Pika pin.

Dependency file looks different?

Check that the file is named requirements.txt. Remove any spaces around the double equals sign.

Need a second look? Help me check my requirements file.

✔️ Awesome, I've got everything!

Great. Save requirements.txt before installing the dependency.

ⓧ I'd like to double check the full code

pika==1.4.4

• Install the pinned dependency inside the active virtual environment by running these commands:

python -m pip install -r requirements.txt
python -m pip show pika

What do these commands prove?

The first command installs every package listed in requirements.txt into the active environment. The second command displays the installed Pika package details.

A successful result confirms that your Python version satisfies Pika's requirement of 3.7 or newer.

The package details should report Pika version 1.4.4. That confirms the dependency pin reached the active virtual environment.

Pika did not install?

• Confirm the terminal prompt includes .venv.
• Check that requirements.txt contains pika==1.4.4.
• Install the current stable macOS release from the official Python downloads page if the installer reports an unsupported Python version.
• Recreate .venv with the commands from the previous substep after updating Python.

Still blocked? Help me troubleshoot my Pika installation.

Start the local RabbitMQ broker

Docker Desktop supplies the local container engine. The pinned RabbitMQ image gives your producer and workers a broker at localhost:5672.

• Press Cmd+Space to open Spotlight.
• Type Docker Desktop into Spotlight.
• Press Return to start Docker Desktop.

The first launch can take several minutes while macOS performs security checks. This startup time is expected.

• Complete any first-run agreement or permission prompt shown by Docker Desktop.
• Switch back to the integrated terminal in Visual Studio Code.
• Confirm that the Docker engine is available by running this command:

docker version

What does this check?

This command asks both the Docker client and the local engine for their version details. Seeing both sections confirms that Docker Desktop is ready to run containers.

You should see version information for both the client and server components.

Docker engine unavailable?

Docker Desktop may still be starting. Wait until startup finishes before trying the version check again.

Need help? Help me connect the Docker CLI to Docker Desktop.

The broker runs inside a named container so later commands can inspect it consistently. Docker may download the RabbitMQ image before starting the container.

• Start the pinned RabbitMQ container by running this command:

docker run -d --hostname rabbitmq --name rabbitmq -p 5672:5672 rabbitmq:4.3.5

What does this command create?

• Detached mode keeps RabbitMQ running after the command finishes.
• The container name rabbitmq gives later health checks a stable target.
• The port mapping exposes the broker at localhost:5672.
• The image tag pins the broker to RabbitMQ 4.3.5.

The terminal should print a container ID. That output confirms Docker accepted the request to start the broker.

Container did not start?

A naming conflict can happen if an older container already uses rabbitmq. Make sure you are not relying on an earlier container before removing anything.

Need help? Help me troubleshoot the RabbitMQ container startup.

Before you run the final check, do you think RabbitMQ has finished booting inside the new container?

• Verify the broker health and installed Pika version by running these commands:

docker exec rabbitmq rabbitmq-diagnostics -q ping && echo "RabbitMQ ready"
python -m pip show pika

What does the final check prove?

The broker check runs RabbitMQ's diagnostic command inside the named container. The success message prints after the broker answers.

The package query confirms that the active virtual environment contains the pinned Pika release.

You should see RabbitMQ ready in the terminal. The package details should report Pika version 1.4.4.

Broker ping not ready?

• Wait a few seconds for RabbitMQ to finish booting.
• Run the final check again.
• Confirm Docker Desktop remains running if the container cannot be reached.

Still no response? Help me diagnose the RabbitMQ health check.

That's the local foundation ready: your broker answers health checks while Pika runs inside the project environment. Next, you'll queue the first checkout order without waiting for downstream work.

Queue the First Checkout Order

Your local RabbitMQ broker is healthy. It does not have an order route or a checkout message yet.

Checkout must accept an order without waiting for slow payment work. In this step, a producer publishes the order into a durable payment queue so RabbitMQ owns the backlog immediately.

In this step, get ready to:

• Define the order topology in broker.py.
• Build a checkout producer that publishes persistent JSON orders.
• Confirm order-1001 is waiting in the payment queue.

Define the broker connection and topology

A direct exchange sends each message to a queue whose binding matches the message's routing key. Durable quorum queues give the payment stage and fulfillment stage separate backlogs.

• Use the new-file control beside ecommerce-order-pipeline in Visual Studio Code's left sidebar to create broker.py.
• Add the broker foundation by pasting this code into broker.py:

import pika

ORDERS_EXCHANGE = "orders"
PAYMENT_QUEUE = "payment_orders"
FULFILLMENT_QUEUE = "fulfillment_orders"
ORDER_CREATED = "order.created"
PAYMENT_COMPLETED = "payment.completed"


def connect():
    return pika.BlockingConnection(
        pika.ConnectionParameters(host="localhost")
    )


def persistent_properties():
    return pika.BasicProperties(
        delivery_mode=pika.DeliveryMode.Persistent
    )

What does this code do?

• The Pika import provides the Python client used to communicate with RabbitMQ.
• The five constants give the exchange, queues, and routing keys one consistent set of names.
• The connect() function creates a connection to the broker at localhost.
• The persistent_properties() function marks published messages for persistent storage.

• Save broker.py.
• Confirm broker.py appears beside requirements.txt in the left sidebar.

Is the broker file missing?

Check that you created broker.py inside ecommerce-order-pipeline. A file created outside that folder cannot be imported by the checkout script.

Still stuck? Help me check my broker.py foundation.

• In broker.py, add the topology function below persistent_properties() by pasting this code:

def declare_topology(channel):
    channel.exchange_declare(
        exchange=ORDERS_EXCHANGE,
        exchange_type="direct",
        durable=True,
    )

    queue_arguments = {"x-queue-type": "quorum"}

    channel.queue_declare(
        queue=PAYMENT_QUEUE,
        durable=True,
        arguments=queue_arguments,
    )
    channel.queue_bind(
        queue=PAYMENT_QUEUE,
        exchange=ORDERS_EXCHANGE,
        routing_key=ORDER_CREATED,
    )

    channel.queue_declare(
        queue=FULFILLMENT_QUEUE,
        durable=True,
        arguments=queue_arguments,
    )
    channel.queue_bind(
        queue=FULFILLMENT_QUEUE,
        exchange=ORDERS_EXCHANGE,
        routing_key=PAYMENT_COMPLETED,
    )

How does the topology work?

• The orders exchange uses direct routing so each routing key selects its matching queue.
• The x-queue-type argument creates both queues as quorum queues.
• The order.created binding sends new orders to payment_orders.
• The payment.completed binding sends paid orders to fulfillment_orders.

• Save broker.py.
• Compare broker.py with the complete reference below.

The final binding should connect FULFILLMENT_QUEUE to PAYMENT_COMPLETED.

Does the topology look different?

Check that both queue declarations use durable=True. Check that both declarations receive queue_arguments.

Need another pair of eyes? Help me compare my RabbitMQ topology with the target code.

✔️ Awesome, I've got everything!

Great. Save broker.py before moving on.

ⓧ I'd like to double check the full code

import pika

ORDERS_EXCHANGE = "orders"
PAYMENT_QUEUE = "payment_orders"
FULFILLMENT_QUEUE = "fulfillment_orders"
ORDER_CREATED = "order.created"
PAYMENT_COMPLETED = "payment.completed"


def connect():
    return pika.BlockingConnection(
        pika.ConnectionParameters(host="localhost")
    )


def persistent_properties():
    return pika.BasicProperties(
        delivery_mode=pika.DeliveryMode.Persistent
    )


def declare_topology(channel):
    channel.exchange_declare(
        exchange=ORDERS_EXCHANGE,
        exchange_type="direct",
        durable=True,
    )

    queue_arguments = {"x-queue-type": "quorum"}

    channel.queue_declare(
        queue=PAYMENT_QUEUE,
        durable=True,
        arguments=queue_arguments,
    )
    channel.queue_bind(
        queue=PAYMENT_QUEUE,
        exchange=ORDERS_EXCHANGE,
        routing_key=ORDER_CREATED,
    )

    channel.queue_declare(
        queue=FULFILLMENT_QUEUE,
        durable=True,
        arguments=queue_arguments,
    )
    channel.queue_bind(
        queue=FULFILLMENT_QUEUE,
        exchange=ORDERS_EXCHANGE,
        routing_key=PAYMENT_COMPLETED,
    )

Build the checkout producer

Publisher confirms let the producer wait for broker acceptance before reporting success. The checkout script also marks the message as persistent so the queued order survives a broker restart.

• Use the new-file control beside ecommerce-order-pipeline in Visual Studio Code's left sidebar to create checkout.py.
• Add the checkout imports by pasting this code into checkout.py:

import json
import sys

from broker import (
    ORDER_CREATED,
    ORDERS_EXCHANGE,
    connect,
    declare_topology,
    persistent_properties,
)

Why these imports?

• The json module converts each order into a message body.
• The sys module gives the script access to the order details entered in the terminal.
• The broker imports reuse the exchange, routing key, connection, topology, and persistence settings from broker.py.

• Save checkout.py.
• Confirm checkout.py appears beside broker.py in the left sidebar.

Seeing an import warning?

Check that broker.py and checkout.py are in the same ecommerce-order-pipeline folder.

Need help? Help me troubleshoot the imports in checkout.py.

• Below the import block in checkout.py, add the input validation and order payload by pasting this code:

def main():
    if len(sys.argv) not in (3, 4):
        print("Usage: python checkout.py ORDER_ID AMOUNT_CENTS [crash]")
        return

    order_id = sys.argv[1]

    try:
        amount_cents = int(sys.argv[2])
    except ValueError:
        print("AMOUNT_CENTS must be an integer")
        return

    if len(sys.argv) == 4 and sys.argv[3] != "crash":
        print("The optional fourth argument must be crash")
        return

    order = {
        "order_id": order_id,
        "amount_cents": amount_cents,
        "simulate_crash": len(sys.argv) == 4,
    }

How is the order built?

• The argument check accepts an order ID, an amount, and an optional crash flag.
• The integer conversion prevents a non-numeric amount from entering the pipeline.
• The order dictionary stores the values that become the message body.
• The simulate_crash field records whether the optional crash path was requested.

• Save checkout.py.
• Confirm the order dictionary remains indented inside main().

Does the order block look misaligned?

Keep each line from the argument check through the order dictionary indented beneath main().

Still unsure? Help me check the indentation in checkout.py.

• In checkout.py, find "simulate_crash": len(sys.argv) == 4,.
• Add the publishing logic directly below the order dictionary by pasting this code:

    connection = connect()
    channel = connection.channel()
    declare_topology(channel)
    channel.confirm_delivery()
    channel.basic_publish(
        exchange=ORDERS_EXCHANGE,
        routing_key=ORDER_CREATED,
        body=json.dumps(order).encode("utf-8"),
        properties=persistent_properties(),
        mandatory=True,
    )
    connection.close()

    print(
        f"QUEUED {order_id} amount_cents={amount_cents} "
        f"crash={order['simulate_crash']}"
    )


if __name__ == "__main__":
    main()

How does publishing work?

• The connection creates a channel that can declare the exchange, queues, and bindings.
• The confirm_delivery() call enables publisher confirms before the message is sent.
• The basic_publish() call sends the encoded order with the order.created routing key.
• The mandatory=True setting requires the order to reach a matching queue.
• The final print reports success only after publishing completes.

• Save checkout.py.
• Confirm the if __name__ == "__main__": entry point sits at the left margin beneath main().

Seeing indentation warnings?

Keep the connection, publishing, and success output inside main(). Keep the final entry point outside the function.

Need help? Help me check the publishing section in checkout.py.

✔️ Awesome, I've got everything!

Your checkout producer is ready. Save checkout.py before publishing the first order.

ⓧ I'd like to double check the full code

import json
import sys

from broker import (
    ORDER_CREATED,
    ORDERS_EXCHANGE,
    connect,
    declare_topology,
    persistent_properties,
)


def main():
    if len(sys.argv) not in (3, 4):
        print("Usage: python checkout.py ORDER_ID AMOUNT_CENTS [crash]")
        return

    order_id = sys.argv[1]

    try:
        amount_cents = int(sys.argv[2])
    except ValueError:
        print("AMOUNT_CENTS must be an integer")
        return

    if len(sys.argv) == 4 and sys.argv[3] != "crash":
        print("The optional fourth argument must be crash")
        return

    order = {
        "order_id": order_id,
        "amount_cents": amount_cents,
        "simulate_crash": len(sys.argv) == 4,
    }

    connection = connect()
    channel = connection.channel()
    declare_topology(channel)
    channel.confirm_delivery()
    channel.basic_publish(
        exchange=ORDERS_EXCHANGE,
        routing_key=ORDER_CREATED,
        body=json.dumps(order).encode("utf-8"),
        properties=persistent_properties(),
        mandatory=True,
    )
    connection.close()

    print(
        f"QUEUED {order_id} amount_cents={amount_cents} "
        f"crash={order['simulate_crash']}"
    )


if __name__ == "__main__":
    main()

Queue and inspect the order

The broker-owned backlog separates checkout acceptance from payment processing. The order can wait safely until an independent payment worker is ready.

Before you run this, consider this question: which queue will hold the order.

• Publish order-1001 and inspect the queue backlog by running these commands:

python checkout.py order-1001 2500
docker exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged

What do these checks prove?

• The first command submits an order with an amount of 2500 cents.
• The checkout result proves the broker accepted the persistent message.
• The second command reports the ready and unacknowledged message counts for each queue.

Your terminal first prints QUEUED order-1001 amount_cents=2500 crash=False.

The payment_orders queue shows 1 under messages_ready. The fulfillment_orders queue shows 0.

That is your first asynchronous handoff working. Checkout has returned while RabbitMQ holds the payment work.

Is the order missing from the queue?

• Confirm the rabbitmq container from the previous step is still running.
• Check that checkout.py publishes with ORDER_CREATED.
• Check that broker.py binds PAYMENT_QUEUE with ORDER_CREATED.

Still missing the order? Help me troubleshoot why payment_orders has no ready message.

Your first order is waiting for payment. Next, you will build an independent worker that charges it and routes the result to fulfillment.

Route the Payment Event

Your checkout producer has placed order-1001 in payment_orders. The order is safely waiting in RabbitMQ.

A separate Pika consumer can process payment without holding checkout open. It publishes a confirmed payment.completed event before acknowledging the original order.

In this step, get ready to:

• Create a payment callback with rejection and controlled-crash paths.
• Publish a persistent payment.completed event before acknowledging the original order.
• Prove the payment-to-fulfillment handoff using queue counts.

Build the payment callback

A message callback decides what happens to each delivery. Manual acknowledgements keep the order owned by the broker until the callback completes its handoff.

• In the Visual Studio Code file sidebar, select the new-file icon.
• Name the file naive_payment_worker.py.
• Add the imports and first callback paths by pasting this code:

import json
import os

from broker import (
    ORDERS_EXCHANGE,
    PAYMENT_COMPLETED,
    PAYMENT_QUEUE,
    connect,
    declare_topology,
    persistent_properties,
)


def handle_payment(channel, method, _properties, body):
    order = json.loads(body)

    if order["amount_cents"] <= 0:
        print(f"REJECTED {order['order_id']}", flush=True)
        channel.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=False,
        )
        return

    print(
        f"CHARGED {order['order_id']} "
        f"redelivered={method.redelivered}",
        flush=True,
    )

What does this code do?

• The imports reuse the exchange names and connection helpers from broker.py.
• The JSON decoder turns the message body into an order dictionary.
• The non-positive amount path sends a negative acknowledgement without requeueing the rejected order.
• The redelivered flag records whether RabbitMQ has delivered this message before.

• Save naive_payment_worker.py.
• Confirm naive_payment_worker.py appears beside broker.py in the file sidebar.
• Confirm handle_payment() contains a rejected-order path.
• Confirm the callback also prints the charge result.

Seeing an indentation or import error?

Check that the imported names match broker.py exactly. Keep every line inside handle_payment() indented by four spaces.

Need a second pair of eyes? Help me check the first part of my payment callback.

The next part performs the stage handoff. The worker publishes the paid order before acknowledging the original delivery.

• Insert this continuation one blank line below the current closing parenthesis inside handle_payment() by pasting:

    if order["simulate_crash"] and not method.redelivered:
        print("CRASHING before acknowledgement", flush=True)
        os._exit(1)

    paid_order = {**order, "status": "paid"}
    channel.basic_publish(
        exchange=ORDERS_EXCHANGE,
        routing_key=PAYMENT_COMPLETED,
        body=json.dumps(paid_order).encode("utf-8"),
        properties=persistent_properties(),
        mandatory=True,
    )
    channel.basic_ack(delivery_tag=method.delivery_tag)
    print(f"PAYMENT_EVENT_PUBLISHED {order['order_id']}", flush=True)

How does the handoff stay reliable?

• The controlled-crash path exits before acknowledgement on the first delivery of a crash-marked order.
• The paid order keeps the original fields.
• The status field records that payment completed.
• The publish uses the payment.completed routing key to reach fulfillment_orders.
• The acknowledgement happens after the publish call succeeds.

• Save naive_payment_worker.py.
• Confirm handle_payment() ends with the PAYMENT_EVENT_PUBLISHED result.

Does the continuation sit outside the callback?

Make sure the entire continuation begins with four spaces. The final publish result stays inside handle_payment().

Still stuck? Help me place the publish and acknowledgement code inside handle_payment().

Start the payment worker

The startup function turns the callback into a long-running worker. Publisher confirms protect the outgoing event while prefetch_count=1 limits the worker to one unacknowledged delivery.

• Add the worker startup two blank lines below the callback by pasting this code:

def main():
    connection = connect()
    channel = connection.channel()
    declare_topology(channel)
    channel.confirm_delivery()
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(
        queue=PAYMENT_QUEUE,
        on_message_callback=handle_payment,
        auto_ack=False,
    )
    print("WAITING_FOR_PAYMENT_ORDERS", flush=True)
    channel.start_consuming()


if __name__ == "__main__":
    main()

What starts the worker?

• The worker opens a channel using the connection helper from broker.py.
• The confirm_delivery() call enables broker confirmation for the outgoing payment event.
• The basic_qos() call keeps a second order away from this worker until the current order is acknowledged.
• The consumer reads from payment_orders with automatic acknowledgements disabled.

• Save naive_payment_worker.py.
• Confirm the file ends with the call to main().

Does the worker startup look incomplete?

Check that main() sits outside handle_payment() with no leading spaces. Keep the final call to main() indented beneath its condition.

Need help comparing the structure? Help me check the function boundaries in naive_payment_worker.py.

✔️ Awesome, I've got everything!

Your worker file now contains the payment callback and its consumer loop.

ⓧ I'd like to double check the full code

import json
import os

from broker import (
    ORDERS_EXCHANGE,
    PAYMENT_COMPLETED,
    PAYMENT_QUEUE,
    connect,
    declare_topology,
    persistent_properties,
)


def handle_payment(channel, method, _properties, body):
    order = json.loads(body)

    if order["amount_cents"] <= 0:
        print(f"REJECTED {order['order_id']}", flush=True)
        channel.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=False,
        )
        return

    print(
        f"CHARGED {order['order_id']} "
        f"redelivered={method.redelivered}",
        flush=True,
    )

    if order["simulate_crash"] and not method.redelivered:
        print("CRASHING before acknowledgement", flush=True)
        os._exit(1)

    paid_order = {**order, "status": "paid"}
    channel.basic_publish(
        exchange=ORDERS_EXCHANGE,
        routing_key=PAYMENT_COMPLETED,
        body=json.dumps(paid_order).encode("utf-8"),
        properties=persistent_properties(),
        mandatory=True,
    )
    channel.basic_ack(delivery_tag=method.delivery_tag)
    print(f"PAYMENT_EVENT_PUBLISHED {order['order_id']}", flush=True)


def main():
    connection = connect()
    channel = connection.channel()
    declare_topology(channel)
    channel.confirm_delivery()
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(
        queue=PAYMENT_QUEUE,
        on_message_callback=handle_payment,
        auto_ack=False,
    )
    print("WAITING_FOR_PAYMENT_ORDERS", flush=True)
    channel.start_consuming()


if __name__ == "__main__":
    main()

How to read this reference

This full-file reference combines the callback and worker startup in their final order. It shows the exact indentation needed for the acknowledgement to remain inside handle_payment().

The worker keeps the integrated terminal occupied while it waits for orders. The queued order-1001 gives it work immediately.

Before you run the worker, predict whether the order stays in payment_orders or moves to fulfillment_orders.

• Start the payment worker by running this command:

python naive_payment_worker.py

What should you see?

• The worker first prints WAITING_FOR_PAYMENT_ORDERS.
• The payment callback prints CHARGED order-1001 redelivered=False.
• The confirmed handoff prints PAYMENT_EVENT_PUBLISHED order-1001.

• Stop the waiting worker after the payment event is published by pressing Control-C.

Is the worker still waiting?

If you only see WAITING_FOR_PAYMENT_ORDERS, confirm that order-1001 was queued in the previous step. The worker cannot print a charge until payment_orders contains a ready message.

If the connection fails, confirm the rabbitmq container from earlier is available. Help me diagnose why my payment worker cannot consume order-1001.

Verify the routed event

Queue counts provide broker-owned evidence of the handoff. The original payment queue should be empty after the paid event reaches the fulfillment queue.

Before you inspect the queues, predict the ready-message count for each queue.

• Inspect both queue backlogs by running this command:

docker exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged

What do the queue counts prove?

• The payment_orders queue shows 0 ready messages because the original order was acknowledged.
• The fulfillment_orders queue shows 1 ready message because the paid event was routed successfully.
• The movement between queues proves that payment runs as an independent pipeline stage.

You've completed the first full stage transition. RabbitMQ now holds the paid order for fulfillment.

Your payment worker now completes a reliable handoff. Next, you'll test what happens when that worker dies before acknowledging its order.

Reproduce a Duplicate Charge

Your first order moved from checkout to payment. Its payment event now waits for fulfillment in RabbitMQ.

That successful run leaves a reliability question unanswered. A worker can die after payment but before acknowledgement.

In this step, get ready to:

• Submit a checkout order configured for the controlled crash.
• Run the naive payment worker until it exits.
• Restart the worker to reveal how the broker handles the unacknowledged delivery.

Queue the controlled crash order

A crash-marked order follows the same persistent publishing path as the first order. Its extra flag tells the worker to exit during payment.

• Queue order-1002 for 4900 cents by running this command:

python checkout.py order-1002 4900 crash

What does this command do?

• The order-1002 value gives this failure test a unique order ID.
• The 4900 value stores the payment amount in cents.
• The crash argument sets simulate_crash to True.

You'll see QUEUED order-1002 amount_cents=4900 crash=True. The order is now waiting in payment_orders.

Order not queued?

• Confirm your terminal prompt starts with (.venv).
• Confirm the RabbitMQ container is running in Docker Desktop.
• Check that the command contains the crash argument after 4900.

Still stuck? Help me troubleshoot why order-1002 is not being queued.

Crash the first worker process

The order is ready for the existing naive worker. This run reaches the simulated charge before the controlled crash ends the process.

Before you run the worker, which line do you think appears last before the process exits?

• Process the crash-marked order by running this command:

python naive_payment_worker.py

The crash is intentional

The first charge line is CHARGED order-1002 redelivered=False.

The worker then prints CRASHING before acknowledgement. Its process exits immediately.

The connection closes before channel.basic_ack runs. RabbitMQ requeues the delivery.

Worker did not reach the crash?

• Confirm the checkout result ended with crash=True.
• Confirm you ran naive_payment_worker.py from the ecommerce-order-pipeline folder.
• Check the terminal for an earlier Python error that stopped the worker before it received the order.

Need a hand? Help me find why the controlled payment-worker crash did not run.

Run the redelivered order

RabbitMQ now holds the unacknowledged delivery again. A fresh worker receives that same order.

Before you restart the worker, do you expect the delivery marker to stay False or change?

• Process the requeued order by running the worker again:

python naive_payment_worker.py

What proves redelivery?

• The line CHARGED order-1002 redelivered=True shows that the same order returned after the crash.
• The line PAYMENT_EVENT_PUBLISHED order-1002 confirms that the worker published the paid event.
• The worker acknowledges the original order after publishing succeeds.

• Press Control-C after PAYMENT_EVENT_PUBLISHED order-1002 appears.

Second charge not appearing?

• Confirm the first worker process ended after CRASHING before acknowledgement.
• Confirm the second worker runs in the same activated .venv.
• Check that the second worker remains open long enough to receive the requeued order.

Still waiting? Help me troubleshoot why RabbitMQ is not redelivering order-1002.

Your terminal now shows CHARGED order-1002 twice. The delivery marker changes from False to True.

What did this failure prove?

At-least-once delivery preserved the order after the worker failed. RabbitMQ gave another worker a chance to finish it.

The naive consumer repeated the payment side effect because it had no idempotency check. Broker-level delivery safety does not guarantee one business action per order.

That's the hidden reliability gap exposed. Next, you'll protect payment and fulfillment from repeated deliveries.

Make Processing Idempotent

Your last test proved that RabbitMQ preserves an order when a worker dies. It also proved that at-least-once delivery can repeat a business side effect.

An idempotent consumer records whether an operation has already happened. You will use SQLite to make repeated deliveries safe before acknowledging them.

In this step, get ready to:

• Record each payment under its unique order ID before acknowledging the delivery.
• Suppress repeated payment or fulfillment side effects with SQLite records.
• Crash the payment worker again to prove that redelivery stays duplicate-safe.

Build the idempotent payment worker

The payment record must survive the controlled crash. A primary key on order_id turns the database into a durable record of completed charges.

• Use the file sidebar in Visual Studio Code to create payment_worker.py inside the ecommerce-order-pipeline folder.
• Add the imports and database name by pasting this code into payment_worker.py:

import json
import os
import sqlite3

from broker import (
    ORDERS_EXCHANGE,
    PAYMENT_COMPLETED,
    PAYMENT_QUEUE,
    connect,
    declare_topology,
    persistent_properties,
)

DATABASE = "orders.db"

What does this setup provide?

• The standard-library modules decode orders, trigger the controlled exit, and store payment records.
• The broker imports reuse the existing exchange, queue, connection, topology, and persistent-message helpers.
• The DATABASE value keeps both workers pointed at orders.db.

• Save payment_worker.py.
• Confirm payment_worker.py appears inside the ecommerce-order-pipeline folder in the file sidebar.

You now have a separate worker file for the duplicate-safe payment path.

Cannot find the new payment file?

Check that the file sits beside broker.py. A nested copy will not resolve the existing broker import from the expected location.

Still stuck? Help me place payment_worker.py in the correct project folder.

• Add the payment-recording function below DATABASE = "orders.db" by pasting this code:

def remember_payment(order):
    database = sqlite3.connect(DATABASE)
    try:
        database.execute(
            """
            CREATE TABLE IF NOT EXISTS payments (
                order_id TEXT PRIMARY KEY,
                amount_cents INTEGER NOT NULL
            )
            """
        )
        result = database.execute(
            """
            INSERT OR IGNORE INTO payments(order_id, amount_cents)
            VALUES (?, ?)
            """,
            (order["order_id"], order["amount_cents"]),
        )
        database.commit()
        return result.rowcount == 1
    finally:
        database.close()

How does the payment record stop duplicates?

• The payments table uses order_id as its primary key.
• The INSERT OR IGNORE statement inserts the first payment record while ignoring later attempts with the same order ID.
• The returned row count becomes True only when this delivery created a new record.
• The commit makes the payment record durable before the worker continues.

• Save payment_worker.py.
• Inspect the editor diagnostics for remember_payment().

You should see no syntax problems in the database function.

Seeing a problem in remember_payment()?

Check that both SQL strings keep their opening and closing triple quotes. Confirm that the parameter tuple contains order_id plus amount_cents.

Need another pair of eyes? Help me debug the payment-recording function.

• Add the first part of handle_payment() below remember_payment() by pasting this code:

def handle_payment(channel, method, _properties, body):
    order = json.loads(body)

    if order["amount_cents"] <= 0:
        print(f"REJECTED {order['order_id']}", flush=True)
        channel.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=False,
        )
        return

    is_new_payment = remember_payment(order)

    if is_new_payment:
        print(f"CHARGED {order['order_id']}", flush=True)
    else:
        print(
            f"DUPLICATE_PAYMENT_SKIPPED {order['order_id']}",
            flush=True,
        )

    if (
        is_new_payment
        and order["simulate_crash"]
        and not method.redelivered
    ):
        print("CRASHING after durable payment record", flush=True)
        os._exit(1)

What changes in the payment callback?

• Invalid amounts are rejected without requeueing.
• A new database row authorizes the visible CHARGED side effect.
• An existing row produces DUPLICATE_PAYMENT_SKIPPED instead.
• The controlled crash happens after the committed payment record on the first delivery.

• Save payment_worker.py.
• Check the indentation inside handle_payment().

The rejection branch, duplicate check, and crash condition should all remain inside the callback.

Callback indentation looks wrong?

Keep each statement after the function definition indented by four spaces. The lines inside each conditional need one additional indentation level.

Need help tracing the blocks? Help me fix the indentation in handle_payment().

• Place the publishing block inside handle_payment() immediately after os._exit(1) by pasting this code:

    paid_order = {**order, "status": "paid"}
    channel.basic_publish(
        exchange=ORDERS_EXCHANGE,
        routing_key=PAYMENT_COMPLETED,
        body=json.dumps(paid_order).encode("utf-8"),
        properties=persistent_properties(),
        mandatory=True,
    )
    channel.basic_ack(delivery_tag=method.delivery_tag)
    print(f"PAYMENT_EVENT_PUBLISHED {order['order_id']}", flush=True)

Why publish before acknowledging?

The worker publishes the paid order before acknowledging the original delivery. This ordering keeps the input message recoverable if publishing fails.

The persistent properties protect the outgoing event inside the durable broker topology.

• Save payment_worker.py.

• Add the worker startup code below handle_payment() by pasting this code:

def main():
    connection = connect()
    channel = connection.channel()
    declare_topology(channel)
    channel.confirm_delivery()
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(
        queue=PAYMENT_QUEUE,
        on_message_callback=handle_payment,
        auto_ack=False,
    )
    print("WAITING_FOR_PAYMENT_ORDERS", flush=True)
    channel.start_consuming()


if __name__ == "__main__":
    main()

How does the worker consume safely?

• Publisher confirms verify that the broker accepts the payment event.
• A prefetch count of 1 limits the worker to one unacknowledged delivery at a time.
• Manual acknowledgements keep RabbitMQ responsible for deliveries until the callback finishes.
• The waiting message gives you a visible startup signal.

• Save payment_worker.py.

Before you test the worker, what startup message do you expect it to print while the payment queue is empty?

• Start the payment worker from the active integrated terminal by running:

python payment_worker.py

What should I see?

You should see WAITING_FOR_PAYMENT_ORDERS. The running process is now listening to payment_orders.

• Stop the payment worker with Control-C after you see the waiting message.

Good progress. Your replacement payment worker now starts with durable duplicate detection in place.

Payment worker does not start?

Confirm the active terminal is inside ecommerce-order-pipeline. Check that broker.py sits beside payment_worker.py.

If the broker connection fails, confirm the existing rabbitmq container is running.

Still blocked? Help me debug why payment_worker.py will not start.

✔️ Awesome, I've got everything!

Your payment worker is saved. It records payments before the controlled crash and suppresses repeated charges.

ⓧ I'd like to double check the full code

import json
import os
import sqlite3

from broker import (
    ORDERS_EXCHANGE,
    PAYMENT_COMPLETED,
    PAYMENT_QUEUE,
    connect,
    declare_topology,
    persistent_properties,
)

DATABASE = "orders.db"


def remember_payment(order):
    database = sqlite3.connect(DATABASE)
    try:
        database.execute(
            """
            CREATE TABLE IF NOT EXISTS payments (
                order_id TEXT PRIMARY KEY,
                amount_cents INTEGER NOT NULL
            )
            """
        )
        result = database.execute(
            """
            INSERT OR IGNORE INTO payments(order_id, amount_cents)
            VALUES (?, ?)
            """,
            (order["order_id"], order["amount_cents"]),
        )
        database.commit()
        return result.rowcount == 1
    finally:
        database.close()


def handle_payment(channel, method, _properties, body):
    order = json.loads(body)

    if order["amount_cents"] <= 0:
        print(f"REJECTED {order['order_id']}", flush=True)
        channel.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=False,
        )
        return

    is_new_payment = remember_payment(order)

    if is_new_payment:
        print(f"CHARGED {order['order_id']}", flush=True)
    else:
        print(
            f"DUPLICATE_PAYMENT_SKIPPED {order['order_id']}",
            flush=True,
        )

    if (
        is_new_payment
        and order["simulate_crash"]
        and not method.redelivered
    ):
        print("CRASHING after durable payment record", flush=True)
        os._exit(1)

    paid_order = {**order, "status": "paid"}
    channel.basic_publish(
        exchange=ORDERS_EXCHANGE,
        routing_key=PAYMENT_COMPLETED,
        body=json.dumps(paid_order).encode("utf-8"),
        properties=persistent_properties(),
        mandatory=True,
    )
    channel.basic_ack(delivery_tag=method.delivery_tag)
    print(f"PAYMENT_EVENT_PUBLISHED {order['order_id']}", flush=True)


def main():
    connection = connect()
    channel = connection.channel()
    declare_topology(channel)
    channel.confirm_delivery()
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(
        queue=PAYMENT_QUEUE,
        on_message_callback=handle_payment,
        auto_ack=False,
    )
    print("WAITING_FOR_PAYMENT_ORDERS", flush=True)
    channel.start_consuming()


if __name__ == "__main__":
    main()

How to use this reference

Compare this complete file with your saved payment_worker.py. Pay close attention to indentation inside handle_payment().

Build the idempotent fulfillment worker

Payment now produces one confirmed event after recovery. Fulfillment needs its own record because any consumer can receive a message more than once.

• Use the file sidebar in Visual Studio Code to create fulfillment_worker.py inside the ecommerce-order-pipeline folder.
• Add the database-backed fulfillment memory by pasting this code into fulfillment_worker.py:

import json
import sqlite3

from broker import FULFILLMENT_QUEUE, connect, declare_topology

DATABASE = "orders.db"


def remember_fulfillment(order):
    database = sqlite3.connect(DATABASE)
    try:
        database.execute(
            """
            CREATE TABLE IF NOT EXISTS fulfillments (
                order_id TEXT PRIMARY KEY
            )
            """
        )
        result = database.execute(
            """
            INSERT OR IGNORE INTO fulfillments(order_id)
            VALUES (?)
            """,
            (order["order_id"],),
        )
        database.commit()
        return result.rowcount == 1
    finally:
        database.close()

How does fulfillment remember an order?

• The worker shares orders.db with the payment worker.
• The fulfillments table keeps one row for each order ID.
• A successful insert marks the first fulfillment attempt as new.
• An ignored insert identifies a repeated fulfillment delivery.

• Save fulfillment_worker.py.
• Confirm fulfillment_worker.py appears beside payment_worker.py.

Both workers now point to the same local database file.

Fulfillment file shows a database problem?

Check the comma in (order["order_id"],). That comma keeps the value in the parameter tuple expected by the SQL statement.

Need help with the function? Help me debug remember_fulfillment().

• Add handle_fulfillment() below remember_fulfillment() by pasting this code:

def handle_fulfillment(channel, method, _properties, body):
    order = json.loads(body)
    is_new_fulfillment = remember_fulfillment(order)

    if is_new_fulfillment:
        print(f"FULFILLED {order['order_id']}", flush=True)
    else:
        print(
            f"DUPLICATE_FULFILLMENT_SKIPPED {order['order_id']}",
            flush=True,
        )

    channel.basic_ack(delivery_tag=method.delivery_tag)

What does the fulfillment callback protect?

The first database insert permits the FULFILLED side effect. A repeated delivery produces DUPLICATE_FULFILLMENT_SKIPPED.

The worker acknowledges either outcome because both leave the order in a safe final state.

• Save fulfillment_worker.py.
• Inspect the editor diagnostics for handle_fulfillment().

You should see no syntax problems in the callback.

Fulfillment callback looks incomplete?

Confirm that channel.basic_ack sits after the conditional block. It must run for both new deliveries and duplicates.

Still unsure? Help me check the fulfillment callback structure.

• Add the fulfillment worker startup code below handle_fulfillment() by pasting this code:

def main():
    connection = connect()
    channel = connection.channel()
    declare_topology(channel)
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(
        queue=FULFILLMENT_QUEUE,
        on_message_callback=handle_fulfillment,
        auto_ack=False,
    )
    print("WAITING_FOR_FULFILLMENT_ORDERS", flush=True)
    channel.start_consuming()


if __name__ == "__main__":
    main()

How does fulfillment receive work?

• The worker consumes from fulfillment_orders with manual acknowledgements.
• The prefetch setting holds the worker to one outstanding delivery.
• The waiting message confirms that the consumer loop has started.

• Save fulfillment_worker.py.

✔️ Awesome, I've got everything!

Your fulfillment worker is saved. It records each completed order before acknowledging the delivery.

ⓧ I'd like to double check the full code

import json
import sqlite3

from broker import FULFILLMENT_QUEUE, connect, declare_topology

DATABASE = "orders.db"


def remember_fulfillment(order):
    database = sqlite3.connect(DATABASE)
    try:
        database.execute(
            """
            CREATE TABLE IF NOT EXISTS fulfillments (
                order_id TEXT PRIMARY KEY
            )
            """
        )
        result = database.execute(
            """
            INSERT OR IGNORE INTO fulfillments(order_id)
            VALUES (?)
            """,
            (order["order_id"],),
        )
        database.commit()
        return result.rowcount == 1
    finally:
        database.close()


def handle_fulfillment(channel, method, _properties, body):
    order = json.loads(body)
    is_new_fulfillment = remember_fulfillment(order)

    if is_new_fulfillment:
        print(f"FULFILLED {order['order_id']}", flush=True)
    else:
        print(
            f"DUPLICATE_FULFILLMENT_SKIPPED {order['order_id']}",
            flush=True,
        )

    channel.basic_ack(delivery_tag=method.delivery_tag)


def main():
    connection = connect()
    channel = connection.channel()
    declare_topology(channel)
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(
        queue=FULFILLMENT_QUEUE,
        on_message_callback=handle_fulfillment,
        auto_ack=False,
    )
    print("WAITING_FOR_FULFILLMENT_ORDERS", flush=True)
    channel.start_consuming()


if __name__ == "__main__":
    main()

How to use this reference

Compare this complete file with your saved fulfillment_worker.py. Confirm that the acknowledgement remains outside the duplicate-check conditional.

Before you start fulfillment, what message do you expect to see before it begins draining the existing fulfillment backlog?

• Start the fulfillment worker from the active integrated terminal by running:

python fulfillment_worker.py

What should I see?

You should first see WAITING_FOR_FULFILLMENT_ORDERS. The worker may then print fulfillment results for payment events already waiting in the queue.

Leave this terminal running for the final recovery test.

Fulfillment worker does not wait?

Confirm that the terminal still has the project virtual environment active. Check that the local broker container remains available.

Need help with the startup failure? Help me debug fulfillment_worker.py.

Prove duplicate-safe crash recovery

The same crash that charged order-1002 twice now becomes a recovery test. The first process records the payment before exiting.

The restarted process sees the durable record and suppresses the repeated charge. It can then publish the payment event safely.

• Create two additional integrated terminal sessions using the terminal controls in Visual Studio Code.
• Activate .venv in each new terminal by running this command:

source .venv/bin/activate

Why activate each new terminal?

Each terminal session has its own shell state. Activation makes the installed Pika dependency available to the worker or producer launched from that session.

• Start the payment worker in the second terminal by running:

python payment_worker.py

What should the payment terminal show?

You should see WAITING_FOR_PAYMENT_ORDERS. The worker is ready for the controlled crash order.

Before you submit the order, which payment message do you think appears before the worker exits?

• Submit the controlled crash order from the third terminal by running:

python checkout.py order-1003 7900 crash

What happened before the crash?

The checkout terminal prints QUEUED order-1003 amount_cents=7900 crash=True.

The payment terminal prints CHARGED order-1003 followed by CRASHING after durable payment record. The process exits before acknowledging the delivery.

Before you restart the payment worker, do you expect another charge or a duplicate-suppression message?

• Restart the payment worker in its terminal by running:

python payment_worker.py

What proves payment is idempotent?

The restarted worker prints DUPLICATE_PAYMENT_SKIPPED order-1003 followed by PAYMENT_EVENT_PUBLISHED order-1003.

The fulfillment terminal prints FULFILLED order-1003 exactly once. The database now contains one payment record plus one fulfillment record for the order.

Seeing CHARGED twice again?

Confirm that you restarted payment_worker.py instead of naive_payment_worker.py. Check that both payment attempts use the exact order ID order-1003.

If the second worker cannot read the record, confirm both workers use DATABASE = "orders.db" from the same project folder.

Still seeing duplicates? Help me trace why the SQLite idempotency record is not being reused.

Before you inspect the broker, do you expect either queue to retain a ready or unacknowledged message?

• Inspect the final queue counts from the third terminal by running:

docker exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged

What confirms the pipeline drained?

You should see payment_orders with 0 ready messages plus 0 unacknowledged messages.

You should also see fulfillment_orders with 0 ready messages plus 0 unacknowledged messages.

A queue count is not zero?

Keep both workers running long enough to consume their pending messages. Repeat the queue inspection after their terminals print the processing results.

If a message stays unacknowledged, inspect the corresponding worker terminal for a callback failure.

Need help reading the queue state? Help me diagnose why my RabbitMQ queues did not drain.

That is the reliability gap closed. RabbitMQ can redeliver the order without repeating its payment or fulfillment side effects.

Secret mission

Quarantine Rejected Payments

A permanently invalid order can keep returning to a worker unless the broker has somewhere safe to move it. Build a dead-letter path that quarantines rejected payments for operator inspection.

Clean Up Your Resources

Clean Up Your Resources

Everything in this project runs on your computer, so there are no ongoing cloud charges. Choose whether to keep the pipeline available, pause it, or delete its local state.

Resources you used:

• Local Docker container named rabbitmq running RabbitMQ 4.3.5. It contains the exchanges, bindings, quorum queues, messages, and payment-dlx policy.
• Local ecommerce-order-pipeline folder. It contains .venv, Python source files, requirements.txt, and the SQLite database orders.db.

Keep everything running

No action needed. This option suits another demonstration of the order pipeline or dead-letter path.

• Your rabbitmq container stays available at localhost:5672.
• Your active worker terminals remain ready to process more orders.
• Your ecommerce-order-pipeline folder keeps its source files, virtual environment, and idempotency records.

Pause - I'll come back to this later

A paused setup frees local memory while preserving the broker state and project files. You can restart the same pipeline later.

• Press Control-C in each worker terminal to stop the running worker.
• Stop the rabbitmq container by running this command:

docker stop rabbitmq

What Does This Do?

This stops the local broker process. The container keeps its queues, exchanges, policy, and stored messages.

• Resume the broker later by running this command:

docker start rabbitmq

What Does This Do?

This restarts the existing container. Your RabbitMQ topology becomes available again.

• Reactivate .venv in the Visual Studio Code terminal by running this command:

source .venv/bin/activate

What Does This Do?

This reconnects the terminal session to the existing isolated Python environment. Pika 1.4.4 is ready for the pipeline scripts.

Delete - I don't want to use this again

Deleting these resources is permanent. The first command removes the local broker container. The second removes the project folder.

• Press Control-C in each worker terminal to stop the running worker.
• Remove the rabbitmq container and its broker state by running this command:

docker rm -f rabbitmq

What Gets Removed?

This removes the container plus its exchanges, bindings, quorum queues, messages, and payment-dlx policy. The downloaded RabbitMQ image remains cached with no running cost.

• Remove the ecommerce-order-pipeline folder from your Desktop by running these commands:

cd ~/Desktop
rm -rf ecommerce-order-pipeline

What Gets Removed?

This removes .venv, the Python source files, requirements.txt, and orders.db. Your local pipeline files and idempotency records are now cleared.

Nice Work!

Nice Work!

You did it. Your resilient e-commerce order pipeline uses RabbitMQ to keep checkout responsive through independent payment and fulfillment stages.

You've learned how to:

• Build a checkout producer that publishes persistent orders with publisher confirms. The visible QUEUED result proves checkout can return before downstream work finishes.
• Connect separate payment and fulfillment workers through a direct exchange. Exact routing keys move each event into a durable quorum queue.
• Force a worker crash to observe at-least-once delivery. The redelivered message exposes a duplicate charge. SQLite-backed idempotency keyed by order_id now suppresses duplicate side effects.
• Secret Mission: Built a policy-driven dead-letter path from payment_orders to failed_payments. Inspected order-bad after RabbitMQ quarantined the rejected payment.

Ready to quiz yourself?