Chapter 20: Orchestrating Multi-Step Workflows with Process Managers

Order fulfillment in a real bookstore is a multi-step process: confirm the order, reserve inventory, arrange shipping. If shipping fails (invalid address), the inventory reservation must be reversed. This is a long-running workflow that spans multiple aggregates and must handle failures gracefully.

A domain service (Chapter 13) handles synchronous, single-step cross-aggregate validation: it checks the rule before one aggregate changes. A process manager handles asynchronous, multi-step workflows with compensation logic.

Domain Service vs. Process Manager

	Domain Service	Process Manager
When	Synchronous, before the one aggregate the command changes	Asynchronous, across multiple transactions
State	Stateless	Stateful, remembers what happened so far
Failure	Rejects the command, so nothing is saved	Issues compensating commands
Use case	"Check inventory before confirming"	"Confirm → reserve → ship → complete"

The Shipping Aggregate

First, we need a Shipping aggregate to represent shipments:

@domain.aggregate
class Shipment:
    order_id: Identifier(required=True)
    status: String(max_length=20, default="PENDING")
    tracking_number: String(max_length=50)

    def create_shipment(self, address: str):
        """Attempt to create a shipment. May fail if address is invalid."""
        if not address or address.strip() == "":
            self.status = "FAILED"
            self.raise_(
                ShipmentFailed(order_id=self.order_id, reason="Invalid address")
            )
        else:
            self.status = "CREATED"
            self.tracking_number = f"TRK-{self.order_id[:8]}"
            self.raise_(
                ShipmentCreated(
                    order_id=self.order_id, tracking_number=self.tracking_number
                )
            )

The create_shipment method validates the address and raises either ShipmentCreated or ShipmentFailed. This is important, the process manager reacts to these events to decide the next step.

The Process Manager

In Chapter 13, the reserve_inventory method on OrderEventHandler sent a ReserveInventory command for each title in OrderConfirmed. The process manager now sends those commands, so remove reserve_inventory from OrderEventHandler. Then one place starts the reservation: the process manager, which also releases the stock if a later step fails. If both stay, Inventory's command handler finds the order in reserved_order_ids and skips the second command, so the stock is still reserved once. OrderConfirmed and ReserveInventory stay as Chapter 13 defined them. Inventory.reserve() now also raises InventoryReserved, which tells the process manager a title is reserved, and records the quantity it took for the order in reserved_quantities. Inventory's command handler also skips an order in released_order_ids, so a reservation that arrives after compensation has released the order takes no stock.

The OrderFulfillmentPM coordinates the fulfillment workflow:

@domain.process_manager(stream_categories=["order", "inventory", "shipment"])
class OrderFulfillmentPM:
    """Coordinates the order fulfillment workflow across aggregates."""

    order_id: Identifier(required=True)
    items = List(content_type=Dict)
    awaiting_titles = List(content_type=String)

    @handle(OrderConfirmed, start=True, correlate="order_id")
    def on_order_confirmed(self, event: OrderConfirmed):
        """Step 1: Order confirmed. Reserve each title."""
        self.order_id = event.order_id
        self.items = event.items
        self.awaiting_titles = [item["book_title"] for item in event.items]
        for item in event.items:
            current_domain.process(
                ReserveInventory(
                    order_id=event.order_id,
                    book_title=item["book_title"],
                    quantity=item["quantity"],
                )
            )

    @handle(InventoryReserved, correlate="order_id")
    def on_inventory_reserved(self, event: InventoryReserved):
        """Step 2: Once every title is reserved, create the shipment."""
        if event.book_title not in self.awaiting_titles:
            return  # a repeated event for a title already counted
        self.awaiting_titles = [
            title for title in self.awaiting_titles if title != event.book_title
        ]
        if self.awaiting_titles:
            return
        order = current_domain.repository_for(Order).get(self.order_id)
        current_domain.process(
            CreateShipment(
                order_id=self.order_id,
                address=order.shipping_address or "",
            )
        )

    @handle(ShipmentCreated, correlate="order_id", end=True)
    def on_shipment_created(self, event: ShipmentCreated):
        """Step 3: Shipment created — complete the order."""
        current_domain.process(CompleteOrder(order_id=event.order_id))

    @handle(ShipmentFailed, correlate="order_id", end=True)
    def on_shipment_failed(self, event: ShipmentFailed):
        """Compensation: Shipment failed. Release the stock and cancel the order."""
        for item in self.items:
            current_domain.process(
                ReleaseInventory(order_id=event.order_id, book_title=item["book_title"])
            )
        current_domain.process(CancelOrder(order_id=event.order_id))

Key Concepts

stream_categories, The process manager subscribes to events from the order, inventory, and shipment streams. It sees all events from these three aggregates.

order_id, items and awaiting_titles fields, Process managers are stateful. These fields are saved between events. order_id says which order the PM is tracking. items keeps the titles and quantities, so the PM knows what to release if shipping fails. awaiting_titles lists the titles not yet reserved, so the PM creates the shipment only after the last one. An InventoryReserved for a title no longer in the list is a repeat, and the PM ignores it. Shipment's command handler also skips an order that already has a shipment.

start=True, The on_order_confirmed handler is marked as the start event. When an OrderConfirmed event arrives, Protean creates a new PM instance and stores the correlated order_id.

correlate, Each handler declares how to match incoming events to existing PM instances:

correlate="order_id": Protean reads the event's order_id value and routes the event to the PM instance started with that value. Every event in this workflow carries the order's id, which is why InventoryReserved has an order_id field.
When an event carries the order's id under another name, use a dictionary, such as correlate={"order_id": "order_ref"}. Protean reads event.order_ref and matches its value the same way.

end=True, Marks the PM instance as finished once the handler returns. Subsequent events for this order_id are skipped. Both the success handler and the compensation handler end the workflow.

How It Works

OrderConfirmed fires → PM starts (via start=True), issues one ReserveInventory command per title.
InventoryReserved fires for each title → once every title is reserved, PM issues CreateShipment command.
ShipmentCreated fires → PM issues CompleteOrder command and completes (end=True).
ShipmentFailed fires → PM issues ReleaseInventory and CancelOrder commands (compensation) and completes (end=True).

The Compensation Pattern

When ShipmentFailed fires, the PM must undo the reservation:

OrderConfirmed ──► ReserveInventory ──► CreateShipment
                                              │
                                       ShipmentFailed
                                              │
                                  ┌───────────┴───────────┐
                                  ▼                       ▼
                          ReleaseInventory          CancelOrder

The PM sends one ReleaseInventory per title. The command carries no quantity: Inventory returns the quantity it recorded for the order when it reserved.

This chapter compensates a failed shipment only. If a reservation fails because another order took the last copies, Inventory.reserve() raises ValidationError and the order stays CONFIRMED. Handling that takes a ReservationFailed event in place of the error, and a PM handler that releases the titles already reserved and cancels the order, as on_shipment_failed does.

This is the saga pattern. Each step has a compensating action that undoes it if a subsequent step fails. The process manager is the coordinator that decides when to compensate.

Design compensating actions to be idempotent. If ReleaseInventory is processed twice (e.g., due to a retry), the second invocation should be harmless. This is critical for reliability in asynchronous systems. Inventory.release() returns the quantity in reserved_quantities for the order and adds the order to released_order_ids. A second release finds the order there and changes nothing. A release that arrives before its reservation returns nothing, and the reservation that follows is skipped.

The full source below also has the command handlers that carry out each command: reserving and releasing stock, creating the shipment, and completing or cancelling the order.

Testing the Process Manager

# tests/test_process_managers.py (example)


def place_confirmed_order(address: str) -> Order:
    order = Order(
        customer_name="Alice",
        shipping_address=address,
        items=[
            OrderItem(book_title="Dune", quantity=2, unit_price=Money(amount=15.99))
        ],
    )
    repo = domain.repository_for(Order)
    repo.add(order)
    order.confirm()
    repo.add(order)  # with sync processing, the whole chain runs here
    return repo.get(order.id)


def test_fulfillment_happy_path():
    """Order confirmed → inventory reserved → shipment created → order completed."""
    with domain.domain_context():
        inventory = Inventory(book_id="book-1", title="Dune", quantity=10)
        domain.repository_for(Inventory).add(inventory)

        order = place_confirmed_order("123 Main St")

        assert order.status == "COMPLETED"
        assert domain.repository_for(Inventory).get(inventory.id).quantity == 8


def test_fulfillment_compensation():
    """Shipment failed → inventory released → order cancelled."""
    with domain.domain_context():
        inventory = Inventory(book_id="book-1", title="Dune", quantity=10)
        domain.repository_for(Inventory).add(inventory)

        order = place_confirmed_order("")  # an empty address fails the shipment

        assert order.status == "CANCELLED"
        assert domain.repository_for(Inventory).get(inventory.id).quantity == 10

Process managers are best tested through integration tests that verify the full event chain. In sync processing mode (domain.config["event_processing"] = "sync"), events are processed immediately. Saving the confirmed order runs the whole chain, and the test then checks the final state of the order and the stock.

What We Built

A Shipping aggregate representing shipments with success and failure outcomes.
An OrderFulfillmentPM process manager that coordinates fulfillment across Order, Inventory, and Shipping.
Event correlation using correlate to route events to the correct PM instance.
Compensation logic that reverses reservations when shipping fails.
The saga pattern for long-running, multi-aggregate workflows.

In the next chapter, we will explore advanced query patterns for building a rich storefront.

Full Source

from protean import Domain, handle
from protean.exceptions import ValidationError
from protean.fields import (
    Dict,
    Float,
    HasMany,
    Identifier,
    Integer,
    List,
    String,
    ValueObject,
)
from protean.utils.globals import current_domain

domain = Domain()
domain.config["command_processing"] = "sync"
domain.config["event_processing"] = "sync"


@domain.value_object
class Money:
    currency: String(max_length=3, default="USD")
    amount: Float(required=True)


@domain.aggregate
class Order:
    customer_name: String(max_length=150, required=True)
    status: String(max_length=20, default="PENDING")
    shipping_address: String(max_length=500)
    items = HasMany("OrderItem")

    def quantities_by_title(self) -> dict[str, int]:
        """Total quantity per book, adding up lines for the same title."""
        totals: dict[str, int] = {}
        for item in self.items:
            totals[item.book_title] = totals.get(item.book_title, 0) + item.quantity
        return totals

    def confirm(self):
        self.status = "CONFIRMED"
        self.raise_(
            OrderConfirmed(
                order_id=self.id,
                customer_name=self.customer_name,
                items=[
                    {"book_title": title, "quantity": quantity}
                    for title, quantity in self.quantities_by_title().items()
                ],
            )
        )

    def complete(self):
        self.status = "COMPLETED"

    def cancel(self):
        self.status = "CANCELLED"


@domain.entity(part_of=Order)
class OrderItem:
    book_title: String(max_length=200, required=True)
    quantity: Integer(required=True)
    unit_price = ValueObject(Money)


@domain.aggregate
class Inventory:
    book_id: Identifier(required=True)
    title: String(max_length=200, required=True)
    quantity: Integer(default=0)
    reserved_order_ids = List(content_type=String)
    reserved_quantities = Dict()  # order id -> copies held for that order
    released_order_ids = List(content_type=String)

    def reserve(self, order_id: str, amount: int):
        if self.quantity < amount:
            raise ValidationError(
                {
                    "quantity": [
                        f"Insufficient stock: {self.quantity} available, {amount} requested"
                    ]
                }
            )
        self.quantity -= amount
        self.reserved_order_ids = [*self.reserved_order_ids, order_id]
        self.reserved_quantities = {**self.reserved_quantities, order_id: amount}
        self.raise_(
            InventoryReserved(order_id=order_id, book_title=self.title, quantity=amount)
        )

    def release(self, order_id: str):
        if order_id in self.released_order_ids:
            return  # already released
        # Return what this order holds, whatever a command says. An order
        # released before its reservation arrives holds nothing, and the
        # release still counts, so that late reservation is skipped.
        self.quantity += self.reserved_quantities.get(order_id, 0)
        self.reserved_quantities = {
            held_for: amount
            for held_for, amount in self.reserved_quantities.items()
            if held_for != order_id
        }
        self.released_order_ids = [*self.released_order_ids, order_id]


@domain.aggregate
class Shipment:
    order_id: Identifier(required=True)
    status: String(max_length=20, default="PENDING")
    tracking_number: String(max_length=50)

    def create_shipment(self, address: str):
        """Attempt to create a shipment. May fail if address is invalid."""
        if not address or address.strip() == "":
            self.status = "FAILED"
            self.raise_(
                ShipmentFailed(order_id=self.order_id, reason="Invalid address")
            )
        else:
            self.status = "CREATED"
            self.tracking_number = f"TRK-{self.order_id[:8]}"
            self.raise_(
                ShipmentCreated(
                    order_id=self.order_id, tracking_number=self.tracking_number
                )
            )




# Events
@domain.event(part_of=Order)
class OrderConfirmed:
    order_id: Identifier(required=True)
    customer_name: String(max_length=150, required=True)
    items = List(content_type=Dict)


@domain.event(part_of=Inventory)
class InventoryReserved:
    order_id: Identifier(required=True)
    book_title: String(max_length=200, required=True)
    quantity: Integer(required=True)


@domain.event(part_of=Shipment)
class ShipmentCreated:
    order_id: Identifier(required=True)
    tracking_number: String(max_length=50)


@domain.event(part_of=Shipment)
class ShipmentFailed:
    order_id: Identifier(required=True)
    reason: String(max_length=500)


# Commands
@domain.command(part_of=Inventory)
class ReserveInventory:
    order_id: Identifier(required=True)
    book_title: String(max_length=200, required=True)
    quantity: Integer(required=True)


@domain.command(part_of=Shipment)
class CreateShipment:
    order_id: Identifier(required=True)
    address: String(max_length=500)


@domain.command(part_of=Order)
class CompleteOrder:
    order_id: Identifier(required=True)


@domain.command(part_of=Order)
class CancelOrder:
    order_id: Identifier(required=True)


@domain.command(part_of=Inventory)
class ReleaseInventory:
    order_id: Identifier(required=True)
    book_title: String(max_length=200, required=True)


@domain.process_manager(stream_categories=["order", "inventory", "shipment"])
class OrderFulfillmentPM:
    """Coordinates the order fulfillment workflow across aggregates."""

    order_id: Identifier(required=True)
    items = List(content_type=Dict)
    awaiting_titles = List(content_type=String)

    @handle(OrderConfirmed, start=True, correlate="order_id")
    def on_order_confirmed(self, event: OrderConfirmed):
        """Step 1: Order confirmed. Reserve each title."""
        self.order_id = event.order_id
        self.items = event.items
        self.awaiting_titles = [item["book_title"] for item in event.items]
        for item in event.items:
            current_domain.process(
                ReserveInventory(
                    order_id=event.order_id,
                    book_title=item["book_title"],
                    quantity=item["quantity"],
                )
            )

    @handle(InventoryReserved, correlate="order_id")
    def on_inventory_reserved(self, event: InventoryReserved):
        """Step 2: Once every title is reserved, create the shipment."""
        if event.book_title not in self.awaiting_titles:
            return  # a repeated event for a title already counted
        self.awaiting_titles = [
            title for title in self.awaiting_titles if title != event.book_title
        ]
        if self.awaiting_titles:
            return
        order = current_domain.repository_for(Order).get(self.order_id)
        current_domain.process(
            CreateShipment(
                order_id=self.order_id,
                address=order.shipping_address or "",
            )
        )

    @handle(ShipmentCreated, correlate="order_id", end=True)
    def on_shipment_created(self, event: ShipmentCreated):
        """Step 3: Shipment created — complete the order."""
        current_domain.process(CompleteOrder(order_id=event.order_id))

    @handle(ShipmentFailed, correlate="order_id", end=True)
    def on_shipment_failed(self, event: ShipmentFailed):
        """Compensation: Shipment failed. Release the stock and cancel the order."""
        for item in self.items:
            current_domain.process(
                ReleaseInventory(order_id=event.order_id, book_title=item["book_title"])
            )
        current_domain.process(CancelOrder(order_id=event.order_id))




@domain.command_handler(part_of=Inventory)
class InventoryCommandHandler:
    @handle(ReserveInventory)
    def reserve(self, command: ReserveInventory) -> None:
        repo = current_domain.repository_for(Inventory)
        inventory = repo.find_by(title=command.book_title)
        if (
            command.order_id in inventory.reserved_order_ids
            or command.order_id in inventory.released_order_ids
        ):
            return  # reserved before, or already released by compensation
        inventory.reserve(command.order_id, command.quantity)
        repo.add(inventory)

    @handle(ReleaseInventory)
    def release(self, command: ReleaseInventory) -> None:
        repo = current_domain.repository_for(Inventory)
        inventory = repo.find_by(title=command.book_title)
        inventory.release(command.order_id)
        repo.add(inventory)


@domain.command_handler(part_of=Shipment)
class ShipmentCommandHandler:
    @handle(CreateShipment)
    def create(self, command: CreateShipment) -> None:
        repo = current_domain.repository_for(Shipment)
        if repo.query.filter(order_id=command.order_id).all().items:
            return  # this order already has a shipment
        shipment = Shipment(order_id=command.order_id)
        shipment.create_shipment(command.address)
        repo.add(shipment)


@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(CompleteOrder)
    def complete(self, command: CompleteOrder) -> None:
        repo = current_domain.repository_for(Order)
        order = repo.get(command.order_id)
        order.complete()
        repo.add(order)

    @handle(CancelOrder)
    def cancel(self, command: CancelOrder) -> None:
        repo = current_domain.repository_for(Order)
        order = repo.get(command.order_id)
        order.cancel()
        repo.add(order)




domain.init(traverse=False)


# tests/test_process_managers.py (example)


def place_confirmed_order(address: str) -> Order:
    order = Order(
        customer_name="Alice",
        shipping_address=address,
        items=[
            OrderItem(book_title="Dune", quantity=2, unit_price=Money(amount=15.99))
        ],
    )
    repo = domain.repository_for(Order)
    repo.add(order)
    order.confirm()
    repo.add(order)  # with sync processing, the whole chain runs here
    return repo.get(order.id)


def test_fulfillment_happy_path():
    """Order confirmed → inventory reserved → shipment created → order completed."""
    with domain.domain_context():
        inventory = Inventory(book_id="book-1", title="Dune", quantity=10)
        domain.repository_for(Inventory).add(inventory)

        order = place_confirmed_order("123 Main St")

        assert order.status == "COMPLETED"
        assert domain.repository_for(Inventory).get(inventory.id).quantity == 8


def test_fulfillment_compensation():
    """Shipment failed → inventory released → order cancelled."""
    with domain.domain_context():
        inventory = Inventory(book_id="book-1", title="Dune", quantity=10)
        domain.repository_for(Inventory).add(inventory)

        order = place_confirmed_order("")  # an empty address fails the shipment

        assert order.status == "CANCELLED"
        assert domain.repository_for(Inventory).get(inventory.id).quantity == 10