Add Async Orders with RabbitMQ

Introduction

30 Second Summary

Online checkout feels instant until one service goes offline. When order handling waits directly on a stock check, a brief Inventory outage can block every purchase.

In this project, you will replace OrderFlow's synchronous checkout with asynchronous messaging through RabbitMQ. You will watch accepted orders survive an Inventory outage before a restarted worker drains the queue without reducing stock twice.

What You'll Build

You will submit an order while Inventory is offline before watching the queued work complete safely when its worker returns.

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

• An accepted checkout that returns HTTP 202 while Inventory is offline.
• A visible queue state that shows one ready message before the worker starts. It returns to zero after the worker finishes.
• Proven crash-safe stock updates that leave ASYNC-001 reduced once after an unacknowledged event is redelivered.
• Secret Mission: Scale two competing Inventory workers to share queued orders without processing an event twice.

Are there any prerequisites?

You need the completed OrderFlow Projects 1-4 workspace before starting. Step 1 verifies the required Linux tools before you change the inherited workspace.

Before We Start

Project 5 starts from the completed persistent-checkout workspace built in OrderFlow Projects 1-4. That inherited stack gives you the synchronous checkout behavior you will test next.

This readiness check confirms Docker Engine can reach its local runtime. It also confirms Docker Compose can manage the inherited stack.

You will verify Visual Studio Code before opening the existing workspace. Project 5 remains untouched during this step.

In this step, get ready to:

• Verify Docker Engine on Linux.
• Verify Docker Compose on Linux.
• Open the completed persistent-checkout workspace in Visual Studio Code.

Verify the Linux toolchain

A successful installation must be available from your terminal. Docker Engine also needs to respond through its local runtime.

• Check the required tools by running these commands:

docker version
docker compose version
code --version

What do these checks confirm?

• The docker version command checks that Docker can report its Client information plus its Server information.
• The docker compose version command checks that the Compose plugin is available.
• The code --version command checks that the Visual Studio Code command is available.

✔️ All three commands work

Good work. Your local toolchain is ready for the inherited OrderFlow stack.

ⓧ One or more commands are missing

• Install Docker Engine from the official Linux guide if docker version is unavailable.
• Install the Docker Compose plugin from the official Linux guide if docker compose version is unavailable.
• Install Visual Studio Code from the official Linux guide if code --version is unavailable.
• Repeat the three checks above after each installation finishes.

Installation still failing?

Describe the failed command plus what your terminal displays. Help me diagnose a missing Docker Engine, Docker Compose, or Visual Studio Code command on Linux.

ⓧ Docker cannot reach its runtime

The Docker command is installed. Its local runtime still needs to be available before containers can run.

• Use the official Docker Engine guide for your Linux distribution to start the runtime.
• Repeat the Docker Engine check above.

Docker still unavailable?

Your Docker service may be stopped. Your Linux account may also lack access to the local runtime.

Share the output without exposing private machine details. Help me troubleshoot why Docker Engine cannot respond on Linux.

Open the inherited workspace

The persistent-checkout workspace contains the service boundary established in Projects 1-4. Opening that exact folder keeps this project connected to the existing OrderFlow application.

• Start Visual Studio Code by running this command:

code

What does this command do?

This uses the editor command you just verified to start Visual Studio Code. You can now select the inherited workspace from its folder picker.

Your next path depends on whether the completed workspace from Projects 1-4 is available.

✔️ My workspace is complete

• Use the Visual Studio Code folder picker to select the completed persistent-checkout folder.
• Confirm persistent-checkout is shown as the open workspace name.

ⓧ I have not completed Projects 1-4

This project extends the application built throughout OrderFlow Projects 1-4. Those projects provide the synchronous checkout flow that you will convert.

• Complete OrderFlow Projects 1-4 before continuing with Project 5.
• Return to this section once the persistent-checkout workspace is complete.

Before you check the starting point, predict which workspace name Visual Studio Code should show.

• Check the open workspace name in Visual Studio Code.

You should see persistent-checkout as the open workspace. No Project 5 code or RabbitMQ resources have been added.

Your Linux toolchain is ready around the inherited workspace. Next, you will reproduce the checkout failure that motivates asynchronous messaging.

Reproduce the Synchronous Failure

OrderFlow's inherited checkout sends Inventory work during the same request. That synchronous HTTP dependency means checkout can only succeed while Inventory is available.

Before RabbitMQ enters the design, you need to experience that coupling from the client's side. The HTTP 503 response becomes the baseline that asynchronous processing must improve.

In this step, get ready to:

• Validate the inherited Compose configuration.
• Start the existing services as a healthy baseline.
• Prove that synchronous checkout depends on Inventory availability.

Establish the healthy baseline

A valid Docker Compose model proves the inherited services still fit together before you test an outage. Starting the unchanged stack gives you a known-good comparison.

• Open the integrated terminal in the existing Visual Studio Code workspace.
• Confirm that the terminal is scoped to the open persistent-checkout workspace.
• List the inherited services while validating the Compose configuration by running this command:

docker compose config --services

What Does This Check Prove?

The command parses compose.yaml before listing its services. A service list confirms that Docker Compose can read the inherited configuration.

You should see order-db, inventory-db, order-service, and inventory-service in the output.

Missing a Baseline Service?

Check that the terminal is scoped to the persistent-checkout workspace. A different folder can point Docker Compose at the wrong project.

If the configuration cannot be parsed, use the reported file location to inspect the inherited compose.yaml file.

Help me diagnose my inherited Compose configuration.

The first build can pause while Docker prepares images. Quiet output during that work does not mean the build has stopped.

• Build the inherited images and start the baseline services by running this command:

docker compose up --build --wait

What Does This Command Do?

• The --build flag refreshes the service images from the inherited source files.
• The --wait flag keeps the command active until the services are running or healthy.
• The command starts the existing containers without adding any Project 5 services.

The terminal returns control after the inherited services reach their running or healthy state.

Your healthy baseline is live. You now know the inherited checkout works before you remove one of its dependencies.

Baseline Failing to Start?

Read the final service named in the build output. A database health failure points to the corresponding inherited database service.

Confirm that no other local process is already using the ports required by the Order or Inventory API.

Help me diagnose the baseline startup failure.

Expose the synchronous dependency

Stopping only Inventory isolates one dependency. Order remains reachable, so the failed checkout points directly to the synchronous service call.

• Stop only the Inventory API by running this command:

docker compose stop inventory-service

What Does This Command Do?

The command stops the existing inventory-service container without removing it. The other baseline services remain available for the test.

Docker Compose reports that inventory-service has stopped. The Order API remains available.

Inventory Still Running?

Confirm that the command named inventory-service exactly. Compose service names must match the names listed by the configuration check.

Help me stop only the Inventory service.

The Order service is still reachable through FastAPI's interactive docs. This gives you a client-side view of the dependency failure.

• Open http://127.0.0.1:8000/docs in your browser.
• Expand POST /orders in the interactive docs.
• Enable request-body editing for the operation.
• Replace the example request body with this JSON:

{
  "client_reference": "async-baseline-001",
  "sku": "keyboard",
  "quantity": 1
}

What Does This Request Test?

• The client_reference value identifies this baseline checkout attempt.
• The sku value targets the inherited keyboard stock item.
• The quantity value requests one unit from Inventory.

Before you submit the request, do you think Order can complete checkout while Inventory is stopped?

• Submit the request from the interactive docs.

You'll see HTTP 503 in the response. Order cannot complete the synchronous checkout while Inventory is unavailable.

That failure is the result you wanted. You have now experienced the availability coupling that this project needs to remove.

Seeing a Different Result?

If the request succeeds, confirm that inventory-service is still stopped. If the Order docs are unreachable, confirm that order-service stayed running.

Help me reproduce the expected checkout failure.

Restore the inherited baseline

The failed request proves that Inventory availability controls checkout availability. Restoring Inventory returns the inherited workspace to its healthy starting state.

Before you run this command, do you expect it to finish before or after the Inventory startup checks pass?

• Restore Inventory and wait for its startup checks by running this command:

docker compose start --wait inventory-service

What Does This Command Do?

The command starts the existing inventory-service container. The --wait flag delays completion until the service is running.

The command completes successfully and returns control to the terminal. Inventory is available again for the next step.

Inventory Not Starting?

Confirm that the inherited inventory-db service is still running. Inventory needs its existing database connection during startup.

Help me restore the Inventory baseline.

What Changed in the Workspace?

Only the runtime state changed during this demonstration. The inherited compose.yaml configuration remains unchanged.

The inherited order-service/main.py implementation still uses synchronous checkout. The inherited inventory-service/main.py implementation also remains unchanged.

No RabbitMQ service or Inventory worker exists yet. The original baseline services are running again.

You have now felt the failure caused by a synchronous service boundary. Next, you'll add the RabbitMQ path that lets Order accept work while Inventory is unavailable.

Queue an Accepted Order

The synchronous checkout failed with HTTP 503 as soon as Inventory stopped. Order could only accept work while Inventory was available.

This step moves that boundary to asynchronous messaging through RabbitMQ. Order stores the checkout before publishing an event that can wait for an Inventory worker.

In this step, get ready to:

• Add the RabbitMQ broker and messaging dependency to the inherited stack.
• Refactor Order to publish durable events after accepting a checkout.
• Submit an order with no worker running and inspect its queue counters.

Add RabbitMQ to the stack

Pika connects each Python service to RabbitMQ. Both services receive the same 1.4.2 dependency because Order publishes events now while Inventory consumes them in the next step.

Why RabbitMQ for this boundary?

A RabbitMQ work queue holds background work outside the HTTP request path. Order can finish accepting a checkout while the event waits for a consumer.

This also keeps event transport separate from the Catalog cache introduced elsewhere in OrderFlow. The broker has one clear responsibility.

• In the Explorer sidebar in Visual Studio Code, select order-service/requirements.txt.
• Replace the file contents with the dependency pins below:

fastapi==0.141.1
uvicorn==0.52.1
psycopg[binary]==3.3.4
pika==1.4.2

What does this dependency file do?

• FastAPI and Uvicorn continue serving the Order API.
• Psycopg continues connecting Order to its PostgreSQL database.
• Pika gives Order the connection and publishing APIs used for RabbitMQ.

• Save order-service/requirements.txt.
• Select inventory-service/requirements.txt in the Explorer sidebar.
• Replace the file contents with the matching dependency pins below:

fastapi==0.141.1
uvicorn==0.52.1
psycopg[binary]==3.3.4
pika==1.4.2

Why update Inventory now?

The Inventory worker arrives in the next step. Adding Pika to the shared Inventory image now prepares that service for its consumer process.

• Save inventory-service/requirements.txt.
• Search the workspace for pika==1.4.2.

You should see the dependency once in each service requirements file.

Docker Compose now needs a broker that Order can reach by the stable hostname rabbitmq. A named volume keeps broker data available when its container stops.

• Select compose.yaml in the Explorer sidebar.
• Add the following service inside services after the inventory-db service:

  rabbitmq:
    image: rabbitmq:4.3.5
    hostname: rabbitmq
    environment:
      RABBITMQ_DEFAULT_USER: orderflow
      RABBITMQ_DEFAULT_PASS: orderflow-local
    volumes:
      - rabbitmq-data:/var/lib/rabbitmq
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
      interval: 5s
      timeout: 5s
      retries: 12
      start_period: 20s
    networks:
      - checkout-api

What does the broker configuration do?

• The rabbitmq:4.3.5 image pins the broker used by this project.
• The dedicated local credentials let clients connect from other containers.
• The health check confirms that the RabbitMQ runtime can answer its diagnostic command.
• The checkout-api network gives Order a private route to the broker.

• Find the volumes section near the bottom of compose.yaml.
• Replace the existing volume and network declarations with this block:

volumes:
  order-data:
  inventory-data:
  rabbitmq-data:
    name: orderflow-rabbitmq-data

networks:
  checkout-api:
  order-data:
    internal: true
  inventory-data:
    internal: true

Why use a named broker volume?

RabbitMQ stores durable broker data under /var/lib/rabbitmq. The external name orderflow-rabbitmq-data makes that local volume easy to identify during cleanup.

• Save compose.yaml.
• Check the broker and volume structure by running this command in the terminal from earlier:

docker compose config

What does this check prove?

Docker Compose parses the updated file before creating anything. A rendered configuration with no validation error proves the new service and volume references fit the existing stack.

You should see the rendered Compose configuration with a rabbitmq service and an orderflow-rabbitmq-data volume.

Compose configuration not valid?

Check that rabbitmq is indented at the same level as the existing services. Confirm that rabbitmq-data sits under the top-level volumes section.

Compare the spacing around checkout-api because YAML indentation controls which section owns each value.

Help me diagnose my Compose configuration error.

• Find the existing order-service block in compose.yaml.
• Replace that service block with the configuration below:

  order-service:
    build: ./order-service
    ports:
      - "127.0.0.1:8000:8000"
    environment:
      PGHOST: order-db
      PGPORT: "5432"
      PGDATABASE: orders
      PGUSER: order_user
      PGPASSWORD: local_order_password
      RABBITMQ_DEFAULT_USER: orderflow
      RABBITMQ_DEFAULT_PASS: orderflow-local
    depends_on:
      order-db:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
    networks:
      - order-data
      - checkout-api

How does Order reach RabbitMQ?

• The RabbitMQ environment values supply the local credentials used by Pika.
• The healthy dependency keeps Order from starting before the broker responds.
• The two network attachments preserve Order's database boundary while giving it access to the broker.

• Save compose.yaml.
• Validate the completed broker configuration by running:

docker compose config

What should the rendered service show?

The rendered order-service configuration should include the RabbitMQ credentials and healthy broker dependency. It should also list both Order networks.

✔️ Awesome, I've got everything!

Your dependency files and Compose configuration are ready. Save all three files before continuing.

ⓧ I'd like to double check the full code

Check the Order dependencies

fastapi==0.141.1
uvicorn==0.52.1
psycopg[binary]==3.3.4
pika==1.4.2

What should match?

Your Order dependency file should contain these four pins in this order.

Check the Inventory dependencies

fastapi==0.141.1
uvicorn==0.52.1
psycopg[binary]==3.3.4
pika==1.4.2

What should match?

Your Inventory dependency file should match the Order dependency file at this stage.

Check the Compose configuration

services:
  order-db:
    image: postgres:18.6
    environment:
      POSTGRES_DB: orders
      POSTGRES_USER: order_user
      POSTGRES_PASSWORD: local_order_password
    volumes:
      - order-data:/var/lib/postgresql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
      interval: 5s
      timeout: 5s
      retries: 5
    networks:
      - order-data

  inventory-db:
    image: postgres:18.6
    environment:
      POSTGRES_DB: inventory
      POSTGRES_USER: inventory_user
      POSTGRES_PASSWORD: local_inventory_password
    volumes:
      - inventory-data:/var/lib/postgresql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
      interval: 5s
      timeout: 5s
      retries: 5
    networks:
      - inventory-data

  rabbitmq:
    image: rabbitmq:4.3.5
    hostname: rabbitmq
    environment:
      RABBITMQ_DEFAULT_USER: orderflow
      RABBITMQ_DEFAULT_PASS: orderflow-local
    volumes:
      - rabbitmq-data:/var/lib/rabbitmq
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
      interval: 5s
      timeout: 5s
      retries: 12
      start_period: 20s
    networks:
      - checkout-api

  inventory-service:
    build: ./inventory-service
    ports:
      - "127.0.0.1:8001:8000"
    environment:
      PGHOST: inventory-db
      PGPORT: "5432"
      PGDATABASE: inventory
      PGUSER: inventory_user
      PGPASSWORD: local_inventory_password
    depends_on:
      inventory-db:
        condition: service_healthy
    networks:
      - inventory-data
      - checkout-api

  order-service:
    build: ./order-service
    ports:
      - "127.0.0.1:8000:8000"
    environment:
      PGHOST: order-db
      PGPORT: "5432"
      PGDATABASE: orders
      PGUSER: order_user
      PGPASSWORD: local_order_password
      RABBITMQ_DEFAULT_USER: orderflow
      RABBITMQ_DEFAULT_PASS: orderflow-local
    depends_on:
      order-db:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
    networks:
      - order-data
      - checkout-api

volumes:
  order-data:
  inventory-data:
  rabbitmq-data:
    name: orderflow-rabbitmq-data

networks:
  checkout-api:
  order-data:
    internal: true
  inventory-data:
    internal: true

What is absent at this stage?

This file contains the RabbitMQ broker and updated Order service. The Inventory worker belongs to the next step.

Refactor Order to publish events

The Order API needs two durable actions. It stores a queued order in PostgreSQL before publishing an order.created event with the same order details.

A direct exchange sends the event to a queue whose routing key matches order.created. Publisher confirmation tells Order whether RabbitMQ accepted the routed message.

• Select order-service/main.py in the Explorer sidebar.
• Replace the inherited synchronous implementation by pasting the following chunks into the file in order:

from contextlib import asynccontextmanager
from typing import Annotated
import json
import os

import pika
import psycopg
from fastapi import Body, FastAPI, HTTPException
from psycopg.rows import dict_row

EXCHANGE_NAME = "orderflow.events"
QUEUE_NAME = "inventory.order-created"
ROUTING_KEY = "order.created"

What does the foundation define?

• The JSON module serializes each event before publication.
• The operating system module reads the broker credentials supplied by Compose.
• The three constants keep the exchange, queue, and routing key consistent across the producer.

def initialize_database() -> None:
    with psycopg.connect() as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS orders (
                id BIGSERIAL PRIMARY KEY,
                sku TEXT NOT NULL,
                quantity INTEGER NOT NULL CHECK (quantity > 0),
                status TEXT NOT NULL,
                client_reference TEXT
            )
            """
        )
        connection.execute(
            "ALTER TABLE orders ADD COLUMN IF NOT EXISTS client_reference TEXT"
        )
        connection.execute(
            """
            CREATE UNIQUE INDEX IF NOT EXISTS orders_client_reference_key
            ON orders (client_reference)
            """
        )

How does Order prepare its database?

The initializer preserves the existing orders table while ensuring client_reference exists. Its unique index lets a client safely retry the same reference.

@asynccontextmanager
async def lifespan(_app: FastAPI):
    initialize_database()
    yield


app = FastAPI(title="Order Service", lifespan=lifespan)


def rabbitmq_parameters():
    credentials = pika.PlainCredentials(
        os.environ["RABBITMQ_DEFAULT_USER"],
        os.environ["RABBITMQ_DEFAULT_PASS"],
    )
    return pika.ConnectionParameters(
        host="rabbitmq",
        credentials=credentials,
        heartbeat=30,
        blocked_connection_timeout=5,
        connection_attempts=10,
        retry_delay=1,
    )

How is the broker connection prepared?

The API initializes its table during startup. The connection helper combines the local credentials with the broker hostname supplied by Docker Compose service discovery.

The retry and timeout settings give RabbitMQ a short window to become available without leaving the API blocked indefinitely.

def declare_order_topology(channel):
    channel.exchange_declare(
        exchange=EXCHANGE_NAME,
        exchange_type="direct",
        durable=True,
    )
    channel.queue_declare(
        queue=QUEUE_NAME,
        durable=True,
    )
    channel.queue_bind(
        queue=QUEUE_NAME,
        exchange=EXCHANGE_NAME,
        routing_key=ROUTING_KEY,
    )

What topology does Order declare?

The function creates a durable direct exchange and durable queue. The binding sends order.created messages into inventory.order-created.

def publish_order_created(order: dict, simulate_consumer_failure: bool) -> None:
    event = {
        "event_id": f"order-created-{order['id']}",
        "event_type": ROUTING_KEY,
        "order_id": order["id"],
        "client_reference": order["client_reference"],
        "sku": order["sku"],
        "quantity": order["quantity"],
        "simulate_consumer_failure": simulate_consumer_failure,
    }

    with pika.BlockingConnection(rabbitmq_parameters()) as connection:
        channel = connection.channel()
        declare_order_topology(channel)
        channel.confirm_delivery()
        channel.basic_publish(
            exchange=EXCHANGE_NAME,
            routing_key=ROUTING_KEY,
            body=json.dumps(event).encode("utf-8"),
            properties=pika.BasicProperties(
                content_type="application/json",
                delivery_mode=pika.DeliveryMode.Persistent,
                message_id=event["event_id"],
                type=event["event_type"],
            ),
            mandatory=True,
        )

How is the event published reliably?

• The event carries an ID derived from the stored order ID.
• Persistent delivery works with the durable topology to retain the message on the broker's durable node.
• Mandatory routing surfaces a publication that cannot reach a queue.
• Publisher confirms make the request wait for the broker's publication result.

@app.get("/health")
def health() -> dict:
    with psycopg.connect() as connection:
        connection.execute("SELECT 1")
    return {"status": "ok"}


@app.get("/orders")
def list_orders() -> list[dict]:
    with psycopg.connect() as connection:
        with connection.cursor(row_factory=dict_row) as cursor:
            return cursor.execute(
                """
                SELECT id, sku, quantity, status, client_reference
                FROM orders
                ORDER BY id
                """
            ).fetchall()

What can you inspect through Order?

The health route still checks Order's private database. The orders route now returns the status and client reference needed to inspect queued checkouts.

@app.post("/orders", status_code=202)
def create_order(payload: Annotated[dict, Body()]) -> dict:
    client_reference = str(payload.get("client_reference", "")).strip()
    sku = str(payload.get("sku", "")).strip()
    quantity = payload.get("quantity")
    simulate_consumer_failure = payload.get(
        "simulate_consumer_failure", False
    )

    if not client_reference or not sku:
        raise HTTPException(
            status_code=400,
            detail="client_reference and sku must be non-empty",
        )
    if not isinstance(quantity, int) or isinstance(quantity, bool) or quantity < 1:
        raise HTTPException(
            status_code=400,
            detail="quantity must be a positive integer",
        )
    if not isinstance(simulate_consumer_failure, bool):
        raise HTTPException(
            status_code=400,
            detail="simulate_consumer_failure must be a boolean",
        )

What does the request validation protect?

The endpoint requires a client reference and SKU with visible text. It also rejects invalid quantities and non-Boolean failure controls before storing an order.

    with psycopg.connect() as connection:
        with connection.cursor(row_factory=dict_row) as cursor:
            order = cursor.execute(
                """
                INSERT INTO orders (sku, quantity, status, client_reference)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (client_reference) DO NOTHING
                RETURNING id, sku, quantity, status, client_reference
                """,
                (sku, quantity, "queued", client_reference),
            ).fetchone()

            if order is None:
                order = cursor.execute(
                    """
                    SELECT id, sku, quantity, status, client_reference
                    FROM orders
                    WHERE client_reference = %s
                    """,
                    (client_reference,),
                ).fetchone()

How does the endpoint handle retries?

A new client reference creates one order with the status queued. A repeated reference loads the existing row instead of inserting a second order.

    if order["sku"] != sku or order["quantity"] != quantity:
        raise HTTPException(
            status_code=409,
            detail="client_reference was already used for a different order",
        )

    try:
        publish_order_created(order, simulate_consumer_failure)
    except pika.exceptions.AMQPError as error:
        raise HTTPException(
            status_code=503,
            detail="Order was stored but its event was not confirmed; retry the same client_reference",
        ) from error

    return order

When does Order return success?

A reused reference must still match the original SKU and quantity. Order publishes only after that check passes.

The endpoint returns HTTP 202 after RabbitMQ confirms publication. Broker failures return HTTP 503 with guidance to retry the same reference.

• Save order-service/main.py.
• Compare the complete file before rebuilding the service.

✔️ Awesome, I've got everything!

Your Order API now stores queued orders and publishes confirmed RabbitMQ events. Make sure order-service/main.py is saved.

ⓧ I'd like to double check the full code

from contextlib import asynccontextmanager
from typing import Annotated
import json
import os

import pika
import psycopg
from fastapi import Body, FastAPI, HTTPException
from psycopg.rows import dict_row

EXCHANGE_NAME = "orderflow.events"
QUEUE_NAME = "inventory.order-created"
ROUTING_KEY = "order.created"


def initialize_database() -> None:
    with psycopg.connect() as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS orders (
                id BIGSERIAL PRIMARY KEY,
                sku TEXT NOT NULL,
                quantity INTEGER NOT NULL CHECK (quantity > 0),
                status TEXT NOT NULL,
                client_reference TEXT
            )
            """
        )
        connection.execute(
            "ALTER TABLE orders ADD COLUMN IF NOT EXISTS client_reference TEXT"
        )
        connection.execute(
            """
            CREATE UNIQUE INDEX IF NOT EXISTS orders_client_reference_key
            ON orders (client_reference)
            """
        )


@asynccontextmanager
async def lifespan(_app: FastAPI):
    initialize_database()
    yield


app = FastAPI(title="Order Service", lifespan=lifespan)


def rabbitmq_parameters():
    credentials = pika.PlainCredentials(
        os.environ["RABBITMQ_DEFAULT_USER"],
        os.environ["RABBITMQ_DEFAULT_PASS"],
    )
    return pika.ConnectionParameters(
        host="rabbitmq",
        credentials=credentials,
        heartbeat=30,
        blocked_connection_timeout=5,
        connection_attempts=10,
        retry_delay=1,
    )


def declare_order_topology(channel):
    channel.exchange_declare(
        exchange=EXCHANGE_NAME,
        exchange_type="direct",
        durable=True,
    )
    channel.queue_declare(
        queue=QUEUE_NAME,
        durable=True,
    )
    channel.queue_bind(
        queue=QUEUE_NAME,
        exchange=EXCHANGE_NAME,
        routing_key=ROUTING_KEY,
    )


def publish_order_created(order: dict, simulate_consumer_failure: bool) -> None:
    event = {
        "event_id": f"order-created-{order['id']}",
        "event_type": ROUTING_KEY,
        "order_id": order["id"],
        "client_reference": order["client_reference"],
        "sku": order["sku"],
        "quantity": order["quantity"],
        "simulate_consumer_failure": simulate_consumer_failure,
    }

    with pika.BlockingConnection(rabbitmq_parameters()) as connection:
        channel = connection.channel()
        declare_order_topology(channel)
        channel.confirm_delivery()
        channel.basic_publish(
            exchange=EXCHANGE_NAME,
            routing_key=ROUTING_KEY,
            body=json.dumps(event).encode("utf-8"),
            properties=pika.BasicProperties(
                content_type="application/json",
                delivery_mode=pika.DeliveryMode.Persistent,
                message_id=event["event_id"],
                type=event["event_type"],
            ),
            mandatory=True,
        )


@app.get("/health")
def health() -> dict:
    with psycopg.connect() as connection:
        connection.execute("SELECT 1")
    return {"status": "ok"}


@app.get("/orders")
def list_orders() -> list[dict]:
    with psycopg.connect() as connection:
        with connection.cursor(row_factory=dict_row) as cursor:
            return cursor.execute(
                """
                SELECT id, sku, quantity, status, client_reference
                FROM orders
                ORDER BY id
                """
            ).fetchall()


@app.post("/orders", status_code=202)
def create_order(payload: Annotated[dict, Body()]) -> dict:
    client_reference = str(payload.get("client_reference", "")).strip()
    sku = str(payload.get("sku", "")).strip()
    quantity = payload.get("quantity")
    simulate_consumer_failure = payload.get(
        "simulate_consumer_failure", False
    )

    if not client_reference or not sku:
        raise HTTPException(
            status_code=400,
            detail="client_reference and sku must be non-empty",
        )
    if not isinstance(quantity, int) or isinstance(quantity, bool) or quantity < 1:
        raise HTTPException(
            status_code=400,
            detail="quantity must be a positive integer",
        )
    if not isinstance(simulate_consumer_failure, bool):
        raise HTTPException(
            status_code=400,
            detail="simulate_consumer_failure must be a boolean",
        )

    with psycopg.connect() as connection:
        with connection.cursor(row_factory=dict_row) as cursor:
            order = cursor.execute(
                """
                INSERT INTO orders (sku, quantity, status, client_reference)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (client_reference) DO NOTHING
                RETURNING id, sku, quantity, status, client_reference
                """,
                (sku, quantity, "queued", client_reference),
            ).fetchone()

            if order is None:
                order = cursor.execute(
                    """
                    SELECT id, sku, quantity, status, client_reference
                    FROM orders
                    WHERE client_reference = %s
                    """,
                    (client_reference,),
                ).fetchone()

    if order["sku"] != sku or order["quantity"] != quantity:
        raise HTTPException(
            status_code=409,
            detail="client_reference was already used for a different order",
        )

    try:
        publish_order_created(order, simulate_consumer_failure)
    except pika.exceptions.AMQPError as error:
        raise HTTPException(
            status_code=503,
            detail="Order was stored but its event was not confirmed; retry the same client_reference",
        ) from error

    return order

What should you compare?

Check the constants and helper names first. Then compare the database logic and all three routes before continuing.

Queue and inspect an accepted order

The broker and producer are ready to prove the new boundary. No Inventory worker exists yet, so a confirmed event remains ready in the durable queue.

• Build the updated image and start RabbitMQ with Order by running:

docker compose up --build --wait rabbitmq order-service

What does this startup command do?

Docker Compose rebuilds Order with Pika installed. It starts RabbitMQ before waiting for the requested services to run or become healthy.

The first build can take a few minutes while Docker downloads the new broker image and installs the updated Python dependencies.

You should see the command finish with RabbitMQ healthy and Order running.

Broker or Order did not start?

Confirm Docker can download the pinned broker image. Check that the RabbitMQ credentials in compose.yaml match the environment values on order-service.

A broker volume created with different credentials can also block this local setup. Remove only the stale broker volume before recreating RabbitMQ.

Help me troubleshoot the RabbitMQ and Order startup.

Repeat runs can leave an older ready message in the queue. Clear that backlog before submitting a replacement client reference.

• Purge stale ready messages by running this command only when repeating the project:

docker compose exec rabbitmq rabbitmqctl purge_queue inventory.order-created

What does the purge remove?

The command removes ready messages from inventory.order-created. It gives a repeat run an empty queue before the new event arrives.

The FastAPI docs provide an interactive request editor for the updated endpoint. This request uses the dedicated asynchronous demo SKU.

• Return to http://127.0.0.1:8000/docs in your browser.
• Expand POST /orders.
• Enable the endpoint's interactive request editor.
• Replace the request body with this JSON:

{"client_reference":"async-order-001","sku":"ASYNC-001","quantity":2}

What does this request represent?

• The client reference identifies this checkout across retries.
• The SKU targets stock prepared for the asynchronous processing demo.
• The quantity asks the future Inventory worker to reserve two units.

• Submit the request through the interactive endpoint.

You should receive HTTP 202. The response should show ASYNC-001 with quantity 2 and status queued.

Before you inspect the broker, predict which queue counter now holds the accepted event.

• Inspect the queue after publication by running:

docker compose exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers

What do the queue counters reveal?

• The name field identifies inventory.order-created.
• The messages_ready field counts events waiting for delivery.
• The messages_unacknowledged field counts delivered events awaiting acknowledgement.
• The consumers field counts workers attached to the queue.

You should see inventory.order-created with one ready message. You should also see zero unacknowledged messages and zero consumers.

Queue counters do not match?

If the queue has more than one ready message, purge the stale backlog and submit a new unique client reference. If the queue is missing, confirm the request returned HTTP 202 after the rebuilt Order service started.

If a consumer appears, confirm that no Inventory worker service has been added or started during this step.

Help me diagnose the unexpected RabbitMQ counters.

That is the decoupling working: Order accepted the checkout while no Inventory worker was available. Next, you will build the consumer that drains this queued event and reserves the stock.

Process Orders in an Inventory Worker

Your Order Service now accepts a checkout while Inventory is unavailable. RabbitMQ holds the accepted event until a consumer is ready.

That queued event is still unfinished work. In this step, you will build an independent Inventory worker that reserves stock before acknowledging the delivery.

In this step, get ready to:

• Move Inventory database initialization into a shared module.
• Build a worker that processes each queued event inside one database transaction.
• Start the worker and prove the queued order reduces stock exactly once.

Centralize Inventory database setup

The Inventory API and worker are separate processes that share one PostgreSQL database. A shared initialization module lets either process safely create the required tables before using them.

• Select the `inventory-service` folder in the VS Code Explorer.
• Create `database.py` inside the `inventory-service` folder.
• Add the Inventory table setup and demo stock by pasting this first section:

import psycopg


def initialize_database() -> None:
    with psycopg.connect() as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS inventory (
                sku TEXT PRIMARY KEY,
                quantity INTEGER NOT NULL CHECK (quantity >= 0)
            )
            """
        )
        connection.execute(
            """
            INSERT INTO inventory (sku, quantity)
            VALUES (%s, %s), (%s, %s), (%s, %s), (%s, %s)
            ON CONFLICT (sku) DO NOTHING
            """,
            (
                "keyboard", 5,
                "mouse", 10,
                "ASYNC-001", 5,
                "SCALE-001", 8,
            ),
        )

What does this setup do?

• `initialize_database()` opens a database transaction through Psycopg.
• `CREATE TABLE IF NOT EXISTS` makes repeated initialization safe.
• `ON CONFLICT DO NOTHING` preserves stock values that already exist.

• Save `database.py`.
• Confirm that `database.py` appears beneath `inventory-service` in the VS Code Explorer.

Seeing a Python syntax problem?

Check that `initialize_database()` is aligned beneath the import. Keep every `connection.execute()` call indented inside the connection block.

Ask for help with the file structure here: Help me fix the indentation in my Inventory database initialization code.

• Add the reservation table and processed-event table directly below the existing seed statement:

        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS reservations (
                client_reference TEXT PRIMARY KEY,
                sku TEXT NOT NULL,
                quantity INTEGER NOT NULL CHECK (quantity > 0)
            )
            """
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS processed_order_events (
                event_id TEXT PRIMARY KEY,
                order_id BIGINT NOT NULL,
                client_reference TEXT NOT NULL,
                sku TEXT NOT NULL,
                quantity INTEGER NOT NULL CHECK (quantity > 0),
                outcome TEXT NOT NULL,
                remaining INTEGER
            )
            """
        )

Why store processed events?

`processed_order_events` gives every event a durable claim keyed by `event_id`. The worker can recognize a redelivered event before changing stock a second time.

The outcome sits in Inventory's database beside the stock update. One transaction makes the claim and inventory change succeed together.

• Save `database.py`.
• Confirm that the VS Code Problems panel shows no Python indentation errors for `database.py`.

Is the second table outside the function?

Both new `connection.execute()` calls need eight leading spaces. They belong inside the same connection block as the Inventory table setup.

Use this prompt for targeted help: Help me place the reservations and processed_order_events statements inside initialize_database().

The Inventory API must use the same initialization path as the worker. It also needs an endpoint that exposes each durable processing result.

• Switch back to `inventory-service/main.py`.
• Replace its import and startup section with this shared initialization setup:

from contextlib import asynccontextmanager
from typing import Annotated

import psycopg
from database import initialize_database
from fastapi import Body, FastAPI, HTTPException
from psycopg.rows import dict_row


@asynccontextmanager
async def lifespan(_app: FastAPI):
    initialize_database()
    yield


app = FastAPI(title="Inventory Service", lifespan=lifespan)

What changes at startup?

The FastAPI lifespan now calls the shared `initialize_database()` function. The API no longer owns a separate copy of the schema setup.

• Keep the existing `health()`, `list_inventory()`, and `reserve_inventory()` functions in `inventory-service/main.py`.
• Add the processed-order endpoint below `list_inventory()`:

@app.get("/processed-orders")
def list_processed_orders() -> list[dict]:
    with psycopg.connect() as connection:
        with connection.cursor(row_factory=dict_row) as cursor:
            return cursor.execute(
                """
                SELECT event_id, order_id, client_reference, sku,
                       quantity, outcome, remaining
                FROM processed_order_events
                ORDER BY order_id
                """
            ).fetchall()

What does this endpoint expose?

`GET /processed-orders` returns the event identifier and processing outcome stored by the worker. Ordering by `order_id` keeps the result predictable during later recovery checks.

• Save `inventory-service/main.py`.
• Confirm that `list_processed_orders()` appears in the VS Code Outline for `main.py`.

Is the endpoint missing from the Outline?

Check that `@app.get("/processed-orders")` starts at the left edge. Make sure the function sits outside `list_inventory()`.

Get help with the endpoint placement here: Help me place list_processed_orders() correctly in inventory-service/main.py.

The Inventory image now needs all three Python modules. Copying them together gives the API and worker the same application code.

• Switch to `inventory-service/Dockerfile`.
• Replace its contents with this image definition:

FROM python:3.13.15-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY database.py main.py worker.py ./

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Why copy three modules?

The image contains the shared database module plus both process entry points. Compose can run the API's default command or override it with the worker command.

• Save `inventory-service/Dockerfile`.
• Confirm that its copy instruction names `database.py`, `main.py`, and `worker.py`.

Does the image build fail on worker.py?

The Dockerfile expects `worker.py` inside `inventory-service`. You will create that file before rebuilding the image.

Ask for help if the file layout differs: Help me check the files copied by my Inventory Dockerfile.

Use these references to compare the completed shared database setup before building the worker.

✔️ Awesome, I've got everything!

Your shared database module and Inventory API changes are ready. Continue to the worker.

ⓧ I'd like to double check the full code

Compare `inventory-service/database.py` with this complete reference.

import psycopg


def initialize_database() -> None:
    with psycopg.connect() as connection:
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS inventory (
                sku TEXT PRIMARY KEY,
                quantity INTEGER NOT NULL CHECK (quantity >= 0)
            )
            """
        )
        connection.execute(
            """
            INSERT INTO inventory (sku, quantity)
            VALUES (%s, %s), (%s, %s), (%s, %s), (%s, %s)
            ON CONFLICT (sku) DO NOTHING
            """,
            (
                "keyboard", 5,
                "mouse", 10,
                "ASYNC-001", 5,
                "SCALE-001", 8,
            ),
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS reservations (
                client_reference TEXT PRIMARY KEY,
                sku TEXT NOT NULL,
                quantity INTEGER NOT NULL CHECK (quantity > 0)
            )
            """
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS processed_order_events (
                event_id TEXT PRIMARY KEY,
                order_id BIGINT NOT NULL,
                client_reference TEXT NOT NULL,
                sku TEXT NOT NULL,
                quantity INTEGER NOT NULL CHECK (quantity > 0),
                outcome TEXT NOT NULL,
                remaining INTEGER
            )
            """
        )

Database comparison

This complete file creates all three Inventory-owned tables. It also seeds the four demo stock rows without overwriting existing quantities.

Compare `inventory-service/main.py` with this complete reference.

from contextlib import asynccontextmanager
from typing import Annotated

import psycopg
from database import initialize_database
from fastapi import Body, FastAPI, HTTPException
from psycopg.rows import dict_row


@asynccontextmanager
async def lifespan(_app: FastAPI):
    initialize_database()
    yield


app = FastAPI(title="Inventory Service", lifespan=lifespan)


@app.get("/health")
def health() -> dict:
    with psycopg.connect() as connection:
        connection.execute("SELECT 1")
    return {"status": "ok"}


@app.get("/inventory")
def list_inventory() -> list[dict]:
    with psycopg.connect() as connection:
        with connection.cursor(row_factory=dict_row) as cursor:
            return cursor.execute(
                "SELECT sku, quantity FROM inventory ORDER BY sku"
            ).fetchall()


@app.get("/processed-orders")
def list_processed_orders() -> list[dict]:
    with psycopg.connect() as connection:
        with connection.cursor(row_factory=dict_row) as cursor:
            return cursor.execute(
                """
                SELECT event_id, order_id, client_reference, sku,
                       quantity, outcome, remaining
                FROM processed_order_events
                ORDER BY order_id
                """
            ).fetchall()


@app.post("/reservations")
def reserve_inventory(payload: Annotated[dict, Body()]) -> dict:
    client_reference = str(payload.get("client_reference", "")).strip()
    sku = str(payload.get("sku", "")).strip()
    quantity = payload.get("quantity")

    if not client_reference or not sku:
        raise HTTPException(
            status_code=400,
            detail="client_reference and sku must be non-empty",
        )
    if not isinstance(quantity, int) or isinstance(quantity, bool) or quantity < 1:
        raise HTTPException(
            status_code=400,
            detail="quantity must be a positive integer",
        )

    with psycopg.connect() as connection:
        with connection.cursor(row_factory=dict_row) as cursor:
            claimed = cursor.execute(
                """
                INSERT INTO reservations (client_reference, sku, quantity)
                VALUES (%s, %s, %s)
                ON CONFLICT (client_reference) DO NOTHING
                RETURNING client_reference
                """,
                (client_reference, sku, quantity),
            ).fetchone()

            if claimed is None:
                existing = cursor.execute(
                    """
                    SELECT client_reference, sku, quantity
                    FROM reservations
                    WHERE client_reference = %s
                    """,
                    (client_reference,),
                ).fetchone()
                if existing["sku"] != sku or existing["quantity"] != quantity:
                    raise HTTPException(
                        status_code=409,
                        detail="client_reference was already used for a different reservation",
                    )
                return {
                    "reserved": quantity,
                    "sku": sku,
                    "duplicate": True,
                }

            item = cursor.execute(
                """
                UPDATE inventory
                SET quantity = quantity - %s
                WHERE sku = %s AND quantity >= %s
                RETURNING sku, quantity AS remaining
                """,
                (quantity, sku, quantity),
            ).fetchone()

            if item is None:
                raise HTTPException(
                    status_code=409,
                    detail="Insufficient inventory",
                )

    return {"reserved": quantity, **item, "duplicate": False}

API comparison

This reference keeps the existing idempotent reservation route. It adds shared initialization plus the processed-order query.

Compare `inventory-service/Dockerfile` with this complete reference.

FROM python:3.13.15-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY database.py main.py worker.py ./

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Image comparison

The final Inventory image includes the API entry point and worker entry point. The default command still starts the Inventory API.

Build the Inventory worker

The worker uses Pika to consume `order.created` events. It claims each `event_id` before changing stock so a repeated delivery becomes harmless.

• Create `worker.py` inside the `inventory-service` folder.
• Add the imports and RabbitMQ connection settings with this first section:

import json
import os
import socket

import pika
import psycopg
from database import initialize_database
from psycopg.rows import dict_row

EXCHANGE_NAME = "orderflow.events"
QUEUE_NAME = "inventory.order-created"
ROUTING_KEY = "order.created"
WORKER_NAME = socket.gethostname()


def rabbitmq_parameters():
    credentials = pika.PlainCredentials(
        os.environ["RABBITMQ_DEFAULT_USER"],
        os.environ["RABBITMQ_DEFAULT_PASS"],
    )
    return pika.ConnectionParameters(
        host="rabbitmq",
        credentials=credentials,
        heartbeat=30,
        blocked_connection_timeout=5,
        connection_attempts=10,
        retry_delay=1,
    )

What does the connection setup hold?

• The exchange, queue, and routing key match the Order Service publisher.
• `WORKER_NAME` records the container hostname so later logs identify the worker replica.
• The credentials come from the worker container environment.

• Save `worker.py`.
• Confirm that VS Code recognizes `rabbitmq_parameters()` as a function.

Are the imports unresolved?

Confirm that `pika==1.4.2` remains in `inventory-service/requirements.txt`. The package becomes available inside the rebuilt Inventory image.

Ask for help with the imports here: Help me check the imports in inventory-service/worker.py.

• Add the durable exchange and queue declarations below `rabbitmq_parameters()`:

def declare_order_topology(channel):
    channel.exchange_declare(
        exchange=EXCHANGE_NAME,
        exchange_type="direct",
        durable=True,
    )
    channel.queue_declare(
        queue=QUEUE_NAME,
        durable=True,
    )
    channel.queue_bind(
        queue=QUEUE_NAME,
        exchange=EXCHANGE_NAME,
        routing_key=ROUTING_KEY,
    )

Why declare topology in the worker?

The consumer confirms that the same durable topology exists before it starts. Declaring matching properties is safe when the producer already created these resources.

• Add strict event-field checks below `declare_order_topology()`:

def validate_event(event: dict) -> None:
    if event.get("event_type") != ROUTING_KEY:
        raise ValueError("unexpected event_type")
    if not str(event.get("event_id", "")).strip():
        raise ValueError("event_id is required")
    order_id = event.get("order_id")
    if not isinstance(order_id, int) or isinstance(order_id, bool):
        raise ValueError("order_id must be an integer")
    if not str(event.get("client_reference", "")).strip():
        raise ValueError("client_reference is required")
    if not str(event.get("sku", "")).strip():
        raise ValueError("sku is required")
    quantity = event.get("quantity")
    if not isinstance(quantity, int) or isinstance(quantity, bool) or quantity < 1:
        raise ValueError("quantity must be a positive integer")
    if not isinstance(event.get("simulate_consumer_failure", False), bool):
        raise ValueError("simulate_consumer_failure must be a boolean")

Why validate before opening the transaction?

The worker rejects malformed fields before they reach a database statement. The separate JSON-object check will also stop arrays and primitive values from reaching `validate_event()`.

• Save `worker.py`.
• Confirm that VS Code shows both `declare_order_topology()` and `validate_event()` in the Outline.

Is event validation nested incorrectly?

Make sure `validate_event()` begins at the left edge. It must remain separate from `declare_order_topology()`.

Use this prompt for help: Help me separate my RabbitMQ topology and event validation functions.

The processing transaction begins by inserting the event claim. A duplicate primary key means this event already reached a durable outcome.

• Add the event-claiming portion of `process_event()` below `validate_event()`:

def process_event(event: dict):
    with psycopg.connect() as connection:
        with connection.cursor(row_factory=dict_row) as cursor:
            claimed = cursor.execute(
                """
                INSERT INTO processed_order_events (
                    event_id, order_id, client_reference, sku,
                    quantity, outcome, remaining
                )
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (event_id) DO NOTHING
                RETURNING event_id
                """,
                (
                    event["event_id"],
                    event["order_id"],
                    event["client_reference"],
                    event["sku"],
                    event["quantity"],
                    "processing",
                    None,
                ),
            ).fetchone()

How does the claim work?

The insert uses `event_id` as the primary key. `ON CONFLICT DO NOTHING` turns a repeated event into an empty claim instead of an exception.

• Add the duplicate lookup directly below the `claimed` assignment:

            if claimed is None:
                existing = cursor.execute(
                    """
                    SELECT outcome, remaining
                    FROM processed_order_events
                    WHERE event_id = %s
                    """,
                    (event["event_id"],),
                ).fetchone()
                return False, existing

What happens to a duplicate?

A duplicate reads the outcome already stored for that event. Returning `False` tells the callback that no new stock update occurred.

• Save `worker.py`.
• Confirm that VS Code shows no indentation errors inside `process_event()`.

Is the duplicate branch unreachable?

Place `if claimed is None:` at the same indentation level as the `claimed` assignment. Keep the lookup nested inside that condition.

Ask for help with this transaction branch: Help me fix the duplicate event branch in process_event().

A new claim can now reserve stock. The stock update and final outcome remain inside the same transaction as the event claim.

• Add the stock update and outcome recording below the duplicate branch:

            item = cursor.execute(
                """
                UPDATE inventory
                SET quantity = quantity - %s
                WHERE sku = %s AND quantity >= %s
                RETURNING quantity AS remaining
                """,
                (event["quantity"], event["sku"], event["quantity"]),
            ).fetchone()

            outcome = "reserved" if item is not None else "rejected"
            remaining = item["remaining"] if item is not None else None
            cursor.execute(
                """
                UPDATE processed_order_events
                SET outcome = %s, remaining = %s
                WHERE event_id = %s
                """,
                (outcome, remaining, event["event_id"]),
            )

    return True, {"outcome": outcome, "remaining": remaining}

Why use one transaction?

The Psycopg connection context commits after the event claim and stock update complete successfully. An exception rolls back both changes.

A missing SKU or insufficient quantity records `rejected`. A successful update records `reserved` plus the remaining stock.

• Save `worker.py`.
• Confirm that `process_event()` ends with a result containing `outcome` and `remaining`.

Does process_event return no result?

Keep the final return outside both connection context blocks. It should remain indented once beneath `def process_event(event: dict):`.

Get help with the transaction boundary here: Help me fix the return position in process_event().

The delivery callback coordinates decoding, processing, logging, and acknowledgement. The acknowledgement happens only after `process_event()` returns from its committed transaction.

• Add the decoding and processing portion of `on_message()` below `process_event()`:

def on_message(channel, method, _properties, body):
    try:
        event = json.loads(body)
        if not isinstance(event, dict):
            raise ValueError("event body must be a JSON object")
        validate_event(event)
        processed, result = process_event(event)

        action = "processed" if processed else "duplicate"
        print(
            f"[{WORKER_NAME}] {action} {event['event_id']} "
            f"outcome={result['outcome']} redelivered={method.redelivered}",
            flush=True,
        )

What does the callback log?

The callback accepts only decoded JSON objects. Its log identifies whether the event was newly processed or recognized as a duplicate.

The `redelivered` value comes from RabbitMQ's delivery metadata. You will use it during the crash recovery step.

• Complete `on_message()` by adding the failure simulation and acknowledgement paths:

        if event.get("simulate_consumer_failure") and processed:
            print(
                f"[{WORKER_NAME}] simulating crash before acknowledgement",
                flush=True,
            )
            os._exit(1)

        channel.basic_ack(delivery_tag=method.delivery_tag)
    except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
        print(f"[{WORKER_NAME}] rejected malformed event: {error}", flush=True)
        channel.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=False,
        )

When does the worker acknowledge?

`basic_ack` runs after database processing completes. A malformed event receives a negative acknowledgement without requeueing.

The deliberate exit sits between commit and acknowledgement. That narrow gap creates the safe redelivery demonstration in the next step.

• Save `worker.py`.
• Confirm that `basic_ack` appears after the simulated-failure branch.

Is the message acknowledged too early?

Keep `channel.basic_ack()` below the simulated exit condition. It must remain inside the `try` block.

Ask for help with the acknowledgement order: Help me verify that my worker acknowledges only after database processing.

The worker entry point initializes the database before connecting to the queue. One-message prefetch keeps each worker focused on a single unacknowledged delivery.

• Add the worker entry point at the bottom of `worker.py`:

def main() -> None:
    initialize_database()
    with pika.BlockingConnection(rabbitmq_parameters()) as connection:
        channel = connection.channel()
        declare_order_topology(channel)
        channel.basic_qos(prefetch_count=1)
        channel.basic_consume(
            queue=QUEUE_NAME,
            on_message_callback=on_message,
            auto_ack=False,
        )
        print(f"[{WORKER_NAME}] waiting for {ROUTING_KEY}", flush=True)
        channel.start_consuming()


if __name__ == "__main__":
    main()

How does consumption begin?

• `initialize_database()` removes the startup race between the API and worker.
• `prefetch_count=1` limits the worker to one unacknowledged message at a time.
• `auto_ack=False` keeps acknowledgement under the callback's control.

• Save `worker.py`.
• Confirm that the VS Code Outline lists `main()` beneath `on_message()`.

Does the worker exit immediately?

Check that the final `main()` call remains inside the `if __name__ == "__main__":` block. Confirm that `channel.start_consuming()` stays inside the RabbitMQ connection context.

Use this prompt for help with the entry point: Help me verify the Inventory worker startup flow.

Your worker now contains the complete consume, transact, and acknowledge flow. Compare the complete file before attaching it to Compose.

✔️ Awesome, I've got everything!

Your worker is ready to run inside the Inventory image.

ⓧ I'd like to double check the full code

import json
import os
import socket

import pika
import psycopg
from database import initialize_database
from psycopg.rows import dict_row

EXCHANGE_NAME = "orderflow.events"
QUEUE_NAME = "inventory.order-created"
ROUTING_KEY = "order.created"
WORKER_NAME = socket.gethostname()


def rabbitmq_parameters():
    credentials = pika.PlainCredentials(
        os.environ["RABBITMQ_DEFAULT_USER"],
        os.environ["RABBITMQ_DEFAULT_PASS"],
    )
    return pika.ConnectionParameters(
        host="rabbitmq",
        credentials=credentials,
        heartbeat=30,
        blocked_connection_timeout=5,
        connection_attempts=10,
        retry_delay=1,
    )


def declare_order_topology(channel):
    channel.exchange_declare(
        exchange=EXCHANGE_NAME,
        exchange_type="direct",
        durable=True,
    )
    channel.queue_declare(
        queue=QUEUE_NAME,
        durable=True,
    )
    channel.queue_bind(
        queue=QUEUE_NAME,
        exchange=EXCHANGE_NAME,
        routing_key=ROUTING_KEY,
    )


def validate_event(event: dict) -> None:
    if event.get("event_type") != ROUTING_KEY:
        raise ValueError("unexpected event_type")
    if not str(event.get("event_id", "")).strip():
        raise ValueError("event_id is required")
    order_id = event.get("order_id")
    if not isinstance(order_id, int) or isinstance(order_id, bool):
        raise ValueError("order_id must be an integer")
    if not str(event.get("client_reference", "")).strip():
        raise ValueError("client_reference is required")
    if not str(event.get("sku", "")).strip():
        raise ValueError("sku is required")
    quantity = event.get("quantity")
    if not isinstance(quantity, int) or isinstance(quantity, bool) or quantity < 1:
        raise ValueError("quantity must be a positive integer")
    if not isinstance(event.get("simulate_consumer_failure", False), bool):
        raise ValueError("simulate_consumer_failure must be a boolean")


def process_event(event: dict):
    with psycopg.connect() as connection:
        with connection.cursor(row_factory=dict_row) as cursor:
            claimed = cursor.execute(
                """
                INSERT INTO processed_order_events (
                    event_id, order_id, client_reference, sku,
                    quantity, outcome, remaining
                )
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (event_id) DO NOTHING
                RETURNING event_id
                """,
                (
                    event["event_id"],
                    event["order_id"],
                    event["client_reference"],
                    event["sku"],
                    event["quantity"],
                    "processing",
                    None,
                ),
            ).fetchone()

            if claimed is None:
                existing = cursor.execute(
                    """
                    SELECT outcome, remaining
                    FROM processed_order_events
                    WHERE event_id = %s
                    """,
                    (event["event_id"],),
                ).fetchone()
                return False, existing

            item = cursor.execute(
                """
                UPDATE inventory
                SET quantity = quantity - %s
                WHERE sku = %s AND quantity >= %s
                RETURNING quantity AS remaining
                """,
                (event["quantity"], event["sku"], event["quantity"]),
            ).fetchone()

            outcome = "reserved" if item is not None else "rejected"
            remaining = item["remaining"] if item is not None else None
            cursor.execute(
                """
                UPDATE processed_order_events
                SET outcome = %s, remaining = %s
                WHERE event_id = %s
                """,
                (outcome, remaining, event["event_id"]),
            )

    return True, {"outcome": outcome, "remaining": remaining}


def on_message(channel, method, _properties, body):
    try:
        event = json.loads(body)
        if not isinstance(event, dict):
            raise ValueError("event body must be a JSON object")
        validate_event(event)
        processed, result = process_event(event)

        action = "processed" if processed else "duplicate"
        print(
            f"[{WORKER_NAME}] {action} {event['event_id']} "
            f"outcome={result['outcome']} redelivered={method.redelivered}",
            flush=True,
        )

        if event.get("simulate_consumer_failure") and processed:
            print(
                f"[{WORKER_NAME}] simulating crash before acknowledgement",
                flush=True,
            )
            os._exit(1)

        channel.basic_ack(delivery_tag=method.delivery_tag)
    except (json.JSONDecodeError, KeyError, TypeError, ValueError) as error:
        print(f"[{WORKER_NAME}] rejected malformed event: {error}", flush=True)
        channel.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=False,
        )


def main() -> None:
    initialize_database()
    with pika.BlockingConnection(rabbitmq_parameters()) as connection:
        channel = connection.channel()
        declare_order_topology(channel)
        channel.basic_qos(prefetch_count=1)
        channel.basic_consume(
            queue=QUEUE_NAME,
            on_message_callback=on_message,
            auto_ack=False,
        )
        print(f"[{WORKER_NAME}] waiting for {ROUTING_KEY}", flush=True)
        channel.start_consuming()


if __name__ == "__main__":
    main()

Worker comparison

This complete file validates each event before opening the processing transaction. It acknowledges a valid delivery only after the durable outcome commits.

Start the worker and drain the queue

Compose now needs a second process built from the Inventory image. The worker connects to the private Inventory database network plus the shared checkout network.

• Switch to `compose.yaml`.
• Replace the existing `inventory-service` definition with this network-aware service definition:

  inventory-service:
    build: ./inventory-service
    ports:
      - "127.0.0.1:8001:8000"
    environment:
      PGHOST: inventory-db
      PGPORT: "5432"
      PGDATABASE: inventory
      PGUSER: inventory_user
      PGPASSWORD: local_inventory_password
    depends_on:
      inventory-db:
        condition: service_healthy
    networks:
      - inventory-data
      - checkout-api

Why use two networks?

`inventory-data` gives the API access to its private database. `checkout-api` preserves the existing service boundary used by the application stack.

• Add the `inventory-worker` service directly below `inventory-service`:

  inventory-worker:
    build: ./inventory-service
    command: ["python", "worker.py"]
    environment:
      PGHOST: inventory-db
      PGPORT: "5432"
      PGDATABASE: inventory
      PGUSER: inventory_user
      PGPASSWORD: local_inventory_password
      RABBITMQ_DEFAULT_USER: orderflow
      RABBITMQ_DEFAULT_PASS: orderflow-local
    depends_on:
      inventory-db:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
    restart: on-failure
    networks:
      - inventory-data
      - checkout-api

How is the worker isolated?

The worker can reach `inventory-db` through `inventory-data`. It can reach RabbitMQ through `checkout-api`.

The command override starts `worker.py` instead of Uvicorn. `restart: on-failure` prepares the process for the deliberate crash in the next step.

• Save `compose.yaml`.
• Validate the updated Compose configuration by running:

docker compose config

What does this check prove?

Compose parses the updated service definitions and resolves their references. A successful check returns the rendered configuration without a validation error.

You should see the rendered `inventory-worker` service in the output. Its networks should include `inventory-data` plus `checkout-api`.

Does Compose reject the file?

Check that `inventory-worker` lines up with `inventory-service` beneath `services`. Keep its `environment`, `depends_on`, `restart`, and `networks` keys indented beneath the service.

Get help with the YAML structure here: Help me fix the inventory-worker service in compose.yaml.

Use the complete Compose file below if you want to confirm every service boundary before starting the worker.

✔️ Awesome, I've got everything!

Your Compose configuration includes the Inventory API and worker as separate processes.

ⓧ I'd like to double check the full code

services:
  order-db:
    image: postgres:18.6
    environment:
      POSTGRES_DB: orders
      POSTGRES_USER: order_user
      POSTGRES_PASSWORD: local_order_password
    volumes:
      - order-data:/var/lib/postgresql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
      interval: 5s
      timeout: 5s
      retries: 5
    networks:
      - order-data

  inventory-db:
    image: postgres:18.6
    environment:
      POSTGRES_DB: inventory
      POSTGRES_USER: inventory_user
      POSTGRES_PASSWORD: local_inventory_password
    volumes:
      - inventory-data:/var/lib/postgresql
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"]
      interval: 5s
      timeout: 5s
      retries: 5
    networks:
      - inventory-data

  rabbitmq:
    image: rabbitmq:4.3.5
    hostname: rabbitmq
    environment:
      RABBITMQ_DEFAULT_USER: orderflow
      RABBITMQ_DEFAULT_PASS: orderflow-local
    volumes:
      - rabbitmq-data:/var/lib/rabbitmq
    healthcheck:
      test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
      interval: 5s
      timeout: 5s
      retries: 12
      start_period: 20s
    networks:
      - checkout-api

  inventory-service:
    build: ./inventory-service
    ports:
      - "127.0.0.1:8001:8000"
    environment:
      PGHOST: inventory-db
      PGPORT: "5432"
      PGDATABASE: inventory
      PGUSER: inventory_user
      PGPASSWORD: local_inventory_password
    depends_on:
      inventory-db:
        condition: service_healthy
    networks:
      - inventory-data
      - checkout-api

  inventory-worker:
    build: ./inventory-service
    command: ["python", "worker.py"]
    environment:
      PGHOST: inventory-db
      PGPORT: "5432"
      PGDATABASE: inventory
      PGUSER: inventory_user
      PGPASSWORD: local_inventory_password
      RABBITMQ_DEFAULT_USER: orderflow
      RABBITMQ_DEFAULT_PASS: orderflow-local
    depends_on:
      inventory-db:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
    restart: on-failure
    networks:
      - inventory-data
      - checkout-api

  order-service:
    build: ./order-service
    ports:
      - "127.0.0.1:8000:8000"
    environment:
      PGHOST: order-db
      PGPORT: "5432"
      PGDATABASE: orders
      PGUSER: order_user
      PGPASSWORD: local_order_password
      RABBITMQ_DEFAULT_USER: orderflow
      RABBITMQ_DEFAULT_PASS: orderflow-local
    depends_on:
      order-db:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy
    networks:
      - order-data
      - checkout-api

volumes:
  order-data:
  inventory-data:
  rabbitmq-data:
    name: orderflow-rabbitmq-data

networks:
  checkout-api:
  order-data:
    internal: true
  inventory-data:
    internal: true

Compose comparison

This configuration keeps each PostgreSQL database on its private network. RabbitMQ and the application processes communicate through `checkout-api`.

The queued order expects `ASYNC-001` to begin at `5`. Resetting that single demo row makes the result deterministic if you have run earlier OrderFlow exercises.

• Reset `ASYNC-001` to `5` by running:

docker compose exec -e PGPASSWORD=local_inventory_password inventory-db psql -h 127.0.0.1 -U inventory_user -d inventory -c "UPDATE inventory SET quantity = 5 WHERE sku = 'ASYNC-001';"

What does the reset change?

The command updates only the `ASYNC-001` demo row inside Inventory's own PostgreSQL container. Existing orders and RabbitMQ messages remain untouched.

You should see PostgreSQL report that one row was updated.

Does the reset fail?

Confirm that `inventory-db` is healthy. Check that the command runs from the open `persistent-checkout` workspace.

Ask for help without sharing credentials: Help me troubleshoot the ASYNC-001 reset command.

Before you start the worker, what do you expect to happen to the one ready message and the `ASYNC-001` quantity?

• Build and start the Inventory worker by running:

docker compose up --build --wait inventory-worker

What happens during startup?

Compose rebuilds the shared Inventory image before starting the worker. It waits for the database and RabbitMQ health checks before launching `worker.py`.

The worker initializes its tables before consuming the queued event. It commits the reservation before sending the manual acknowledgement.

The backlog now drains. Your `async-order-001` event has one durable `reserved` outcome.

Does the worker fail to become healthy?

Check the worker output for a Python import or indentation problem. Confirm that `database.py`, `main.py`, and `worker.py` are all inside `inventory-service`.

Use this prompt for environment-specific help: Help me troubleshoot why inventory-worker is failing to start.

• Return to the Inventory API docs at `http://127.0.0.1:8001/docs`.
• Run `GET /inventory` from the Inventory API docs.

You should see `ASYNC-001` with quantity `3`. The queued order reserved two units from the reset quantity of `5`.

• Run `GET /processed-orders` from the Inventory API docs.

You should see one record for `async-order-001`. Its outcome should be `reserved` with `3` remaining.

Before you inspect the broker, do you expect either message counter to remain above zero?

• Inspect the final queue counters by running:

docker compose exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers

What do the counters prove?

• `messages_ready` at `0` proves no queued event is waiting for delivery.
• `messages_unacknowledged` at `0` proves the worker acknowledged its completed delivery.
• A consumer count of `1` proves the Inventory worker remains attached to the queue.

That's the worker running: the queue is empty, the processing result is durable, and `ASYNC-001` has fallen from `5` to `3` only after consumption.

Is the message still ready?

Confirm that the worker service remains running. Check that its queue name is exactly `inventory.order-created`.

If the message is unacknowledged, inspect the worker output for a database error before `basic_ack`. Help me diagnose why the Inventory queue did not drain.

Your asynchronous checkout now completes through an independent consumer. Next, you will crash that consumer after its commit and prove that redelivery cannot reduce stock twice.

Crash, Redeliver, and Recover

Your Docker Compose stack now drains queued orders through an independent Inventory worker. The first queued event reduced ASYNC-001 from 5 to 3.

The worker sends a manual acknowledgement only after its PostgreSQL transaction commits. This protects the durable processing outcome.

A crash in that gap triggers RabbitMQ redelivery. Idempotency must keep the second delivery from reducing stock twice.

In this step, get ready to:

• Trigger an intentional worker crash after its database transaction commits.
• Observe the restarted worker identify the redelivered event as a duplicate.
• Prove the stock changed once before confirming the queue is empty.

Trigger the post-commit crash

The simulate_consumer_failure field creates a precise failure window after the transaction commits. The worker exits before it acknowledges the delivery.

• Start following the Inventory worker logs from the persistent-checkout workspace by running this command:

docker compose logs --follow inventory-worker

What does this command do?

This keeps the Inventory worker output attached to the terminal. New log entries appear as the container processes the event before exiting.

• Open http://127.0.0.1:8000/docs in your browser.
• Expand POST /orders in the Order API docs.
• Replace the sample request body by copying this payload:

{"client_reference":"async-crash-001","sku":"ASYNC-001","quantity":1,"simulate_consumer_failure":true}

What does this request test?

This JSON payload asks the worker to reserve one unit of ASYNC-001. The failure flag makes the first processing attempt exit after the database commit.

The unique client_reference keeps this crash test separate from the order processed in the previous step.

• Change the numeric suffix in async-crash-001 if you have already submitted this crash test.

Before you submit the order, do you think the crash will leave stock at 2 or cause another reduction when the event returns?

• Submit the request from the Order API docs.

You'll receive an HTTP 202 response. Order has accepted the checkout without waiting for Inventory to complete it.

In the worker terminal, you'll first see a processed entry. The next entry reports simulating crash before acknowledgement.

This crash is intentional

The stock transaction has already committed when the process exits. RabbitMQ still considers the delivery unfinished because basic_ack has not run.

This is the failure window that creates at-least-once delivery. The same event can reach the worker again.

Worker did not crash?

• Confirm the request body sets simulate_consumer_failure to true.
• Change the suffix in async-crash-001 if that client reference already has a processed event.

Ask for help with the crash test logs.

Verify redelivery and idempotency

The worker uses restart: on-failure after its deliberate exit. RabbitMQ returns the unacknowledged delivery to the restarted consumer.

• Give the worker a few seconds to restart.
• Return to the worker log terminal from the previous substep.
• Locate the restarted worker entry containing redelivered=True.
• Locate duplicate in the same processing entry.

The restarted worker received the same event. Its existing event_id claim turned the second delivery into a harmless duplicate.

Why does the stock stay safe?

The first delivery inserts its event_id into processed_order_events inside the same transaction as the stock update. The committed row remains after the worker exits.

The redelivered event cannot claim that primary key again. The worker reads the existing outcome before acknowledging the duplicate.

• Open http://127.0.0.1:8001/docs in your browser.
• Send a GET /processed-orders request from the Inventory API docs.

You'll see one row whose client_reference is async-crash-001. Its outcome is reserved with remaining set to 2.

• Send a GET /inventory request from the Inventory API docs.
• Find the inventory row for ASYNC-001.

You'll see quantity set to 2. The redelivery did not apply a second stock reduction.

Stock changed more than once?

• Check the worker logs for one processed entry followed by one duplicate entry for the same event.
• Check GET /processed-orders for one row tied to async-crash-001.

Ask for help tracing the duplicate stock update.

That's the difficult failure handled. Your worker recovered from a post-commit crash without applying the order twice.

Confirm the recovered queue is empty

The database results prove the duplicate was harmless. The broker counters now confirm that the restarted worker acknowledged the recovered delivery.

Before you run the final queue check, do you expect RabbitMQ to show any message still waiting for work?

• Inspect the recovered queue counters by running this command:

docker compose exec rabbitmq rabbitmqctl list_queues name messages_ready messages_unacknowledged consumers

What do these counters prove?

• The messages_ready value counts deliveries waiting for a consumer.
• The messages_unacknowledged value counts deliveries held by consumers before acknowledgement.
• The consumers value shows how many workers are attached to the queue.

You'll see inventory.order-created with zero ready messages. You'll also see zero unacknowledged messages with one consumer attached.

Queue counters are not zero?

• Check the worker log terminal for a pending restart or processing error.
• Wait for the restarted worker to finish handling the redelivered event.
• Run the queue inspection again after the duplicate entry is acknowledged.

Ask for help interpreting the remaining queue counters.

You proved the full recovery path. The order survived a consumer crash while its durable outcome remained singular and its stock reduction happened once.

Secret mission

Scale Competing Inventory Workers

One worker can recover safely. Now add a second Inventory worker to increase throughput. Prove that both workers can share four queued orders while each event changes stock exactly once.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep the local stack active, pause it for later, or remove the Project 5 runtime resources. Your Docker Compose stack runs only on your Linux device, so it creates no cloud costs.

Resources you used:

• One local RabbitMQ broker container running version 4.3.5.
• One named broker volume called orderflow-rabbitmq-data.
• One local inventory-worker container built from the Inventory service image.

Your updated Order and Inventory images remain on disk. The inherited PostgreSQL volumes and source changes are also preserved.

Keep everything running

No action is needed while you continue testing the asynchronous checkout flow.

• The RabbitMQ broker remains ready to accept new order.created events.
• The inventory-worker service remains available with one replica.
• Your source changes and inherited database volumes stay available for the next OrderFlow project.
• The local stack uses no paid cloud services. It incurs no ongoing service charges while idle.

Pause - I'll come back to this later

Pausing releases the resources used by the running containers. Your containers and stored data remain available for a later restart.

• In the VS Code terminal for the persistent-checkout workspace, stop the local stack by running:

docker compose stop

What Gets Paused?

This command stops every container in the current Compose stack. It keeps the containers available for a later restart.

• Restore the existing containers when you return by running:

docker compose start --wait

What Gets Restored?

This command starts the existing containers. It waits until the services are running or healthy.

Delete - I don't want to use this again

Deleting these runtime resources removes only the Project 5 broker and worker containers. Your source changes and inherited database volumes stay intact.

• In the VS Code terminal for the persistent-checkout workspace, remove the worker and broker containers by running:

docker compose rm -f -s inventory-worker rabbitmq

What Gets Removed?

This command stops both named services before removing their containers. It leaves their images and Compose configuration untouched.

• Remove the RabbitMQ data volume by running:

docker volume rm orderflow-rabbitmq-data

Why Remove the Volume?

Removing orderflow-rabbitmq-data clears the broker's local state and queued messages. Your inherited order-data and inventory-data volumes remain.

The asynchronous flow remains unavailable until the RabbitMQ service and its volume are recreated from compose.yaml.

Nice Work!

Nice Work!

You did it! You completed OrderFlow Project 5 with a resilient asynchronous checkout flow powered by RabbitMQ.

What you learned:

• Built asynchronous order acceptance by replacing the synchronous Inventory request with a durable order.created event. The Order API now returns HTTP 202 while Inventory is unavailable.
• Created an idempotent Inventory worker that claims each event_id in PostgreSQL. The worker updates stock in the same transaction as the durable processing outcome. It acknowledges each message only after the transaction commits.
• Demonstrated crash-safe recovery by exiting the worker after its database commit. RabbitMQ redelivered the unacknowledged event. The duplicate-safe handler kept ASYNC-001 from losing stock twice.
• Secret Mission: Scaled two competing Inventory workers to share four queued orders. Each event changed SCALE-001 stock exactly once. The queue returned to zero ready messages.

Ready to quiz yourself?