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. A process manager handles asynchronous, multi-step workflows with compensation logic.

Domain Service vs. Process Manager

	Domain Service	Process Manager
When	Synchronous, same transaction	Asynchronous, across multiple transactions
State	Stateless	Stateful, remembers what happened so far
Failure	Rolls back the transaction	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

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)

    @handle(OrderConfirmed, start=True, correlate="order_id")
    def on_order_confirmed(self, event: OrderConfirmed):
        """Step 1: Order confirmed — reserve inventory."""
        current_domain.process(
            ReserveInventory(
                order_id=event.order_id,
                book_id="placeholder",  # In a real system, load order items
                quantity=1,
            )
        )

    @handle(InventoryReserved, correlate={"order_id": "book_id"})
    def on_inventory_reserved(self, event: InventoryReserved):
        """Step 2: Inventory reserved — create shipment."""
        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")
    def on_shipment_created(self, event: ShipmentCreated):
        """Step 3: Shipment created — complete the order."""
        current_domain.process(CompleteOrder(order_id=event.order_id))
        self.mark_as_complete()

    @handle(ShipmentFailed, correlate="order_id")
    def on_shipment_failed(self, event: ShipmentFailed):
        """Compensation: Shipment failed — release inventory and cancel order."""
        current_domain.process(ReleaseInventory(book_id="placeholder", quantity=1))
        current_domain.process(CancelOrder(order_id=event.order_id))
        self.mark_as_complete()

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 field, Process managers are stateful. The order_id field is persisted between events, so the PM remembers which order it is tracking across the entire workflow.

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": The event's order_id field matches the PM's order_id field directly.
• correlate={"order_id": "book_id"}: Maps the PM's order_id to the event's book_id field (for events that don't carry order_id).

mark_as_complete(), Marks the PM instance as finished. Subsequent events for this order_id are skipped.

How It Works

1. OrderConfirmed fires → PM starts (via start=True), issues ReserveInventory command.
2. InventoryReserved fires → PM issues CreateShipment command.
3. ShipmentCreated fires → PM issues CompleteOrder command, marks itself complete.
4. ShipmentFailed fires → PM issues ReleaseInventory and CancelOrder commands (compensation).

The Compensation Pattern

When ShipmentFailed fires, the PM must undo the reservation:

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

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.

Testing the Process Manager

# tests/test_process_managers.py (example)


def test_fulfillment_happy_path():
    """Order confirmed → inventory reserved → shipment created → order completed."""
    # This test would use sync processing to verify the full chain


def test_fulfillment_compensation():
    """Shipment failed → inventory released → order cancelled."""
    # This test would verify compensation logic

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, making it straightforward to assert the final state after triggering the initial event.

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.fields import Float, Identifier, Integer, String
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)

    def confirm(self):
        self.status = "CONFIRMED"
        self.raise_(OrderConfirmed(order_id=self.id, customer_name=self.customer_name))

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

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


@domain.aggregate
class Inventory:
    book_id: Identifier(required=True)
    title: String(max_length=200, required=True)
    quantity: Integer(default=0)

    def reserve(self, amount: int):
        self.quantity -= amount
        self.raise_(InventoryReserved(book_id=self.book_id, quantity=amount))

    def release(self, amount: int):
        self.quantity += amount


@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)


@domain.event(part_of=Inventory)
class InventoryReserved:
    book_id: Identifier(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_id: Identifier(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:
    book_id: Identifier(required=True)
    quantity: Integer(required=True)


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

    order_id: Identifier(required=True)

    @handle(OrderConfirmed, start=True, correlate="order_id")
    def on_order_confirmed(self, event: OrderConfirmed):
        """Step 1: Order confirmed — reserve inventory."""
        current_domain.process(
            ReserveInventory(
                order_id=event.order_id,
                book_id="placeholder",  # In a real system, load order items
                quantity=1,
            )
        )

    @handle(InventoryReserved, correlate={"order_id": "book_id"})
    def on_inventory_reserved(self, event: InventoryReserved):
        """Step 2: Inventory reserved — create shipment."""
        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")
    def on_shipment_created(self, event: ShipmentCreated):
        """Step 3: Shipment created — complete the order."""
        current_domain.process(CompleteOrder(order_id=event.order_id))
        self.mark_as_complete()

    @handle(ShipmentFailed, correlate="order_id")
    def on_shipment_failed(self, event: ShipmentFailed):
        """Compensation: Shipment failed — release inventory and cancel order."""
        current_domain.process(ReleaseInventory(book_id="placeholder", quantity=1))
        current_domain.process(CancelOrder(order_id=event.order_id))
        self.mark_as_complete()




domain.init(traverse=False)


# tests/test_process_managers.py (example)


def test_fulfillment_happy_path():
    """Order confirmed → inventory reserved → shipment created → order completed."""
    # This test would use sync processing to verify the full chain


def test_fulfillment_compensation():
    """Shipment failed → inventory released → order cancelled."""
    # This test would verify compensation logic

Next

Chapter 21: Advanced Query Patterns →