Distributed Ball of Mud, and Thinking in Nouns

Before we get into that, let’s talk about the alternatives. We regularly talk to engineers who are trying to build out a Microservices architecture. Often they are migrating from an existing application, and their first instinct is to split their system up into nouns.

What nouns have we introduced so far in our system? Well, we have batches of stock, we have orders, products, and customers. So a naive attempt at breaking up the system might have looked like Figure 11-2 (notice that we’ve named our system after a noun, batches, instead of allocation).

Figure 11-2. Context diagram for the “batches” service

[plantuml, batches_context_diagram, config=plantuml.cfg]
@startuml Batches Context Diagram
!includeurl https://raw.githubusercontent.com/RicardoNiepel/C4-PlantUML/master/C4_Context.puml

skinparam roundCorner 20

System(batches, "Batches", "Knows about available stock")
Person(customer, "Customer", "Wants to buy furniture")
System(orders, "Orders", "Knows about customer orders")
System(warehouse, "Warehouse", "Knows about shipping instructions")

Rel_R(customer, orders, "Places order with")
Rel_D(orders, batches, "Reserves stock with")
Rel_D(batches, warehouse, "Sends instructions to")

@enduml

Each “thing” in our system has an associated service, which exposes an HTTP API.

Let’s work through an example happy-path flow in Figure 11-3: our users visit a website and can choose from products that are in stock. When they add an item to their basket, we will reserve some stock for them. When an order is complete, we confirm the reservation, which causes us to send dispatch instructions to the warehouse. Let’s also say, if this is the customer’s third order, we want to update the customer record to flag them as a VIP.

Figure 11-3. Command Flow 1

[plantuml, command_flow_diagram_1, config=plantuml.cfg]
@startuml

actor Customer
entity Orders
entity Batches
entity Warehouse
database CRM


== Reservation ==

  Customer -> Orders: Add product to basket
  Orders -> Batches: Reserve stock

== Purchase ==

  Customer -> Orders: Place order
  activate Orders
  Orders -> Batches: Confirm reservation
  Batches -> Warehouse: Dispatch goods
  Orders -> CRM: Update customer record
  deactivate Orders


@enduml

We can think of each of these steps as a Command in our system: ReserveStock, ConfirmReservation, DispatchGoods, MakeCustomerVIP…

This style of architecture, where we create a microservice per database table, and treat our HTTP APIs as CRUD interfaces to anemic models is the most common initial way for people to approach service-oriented design.

This works fine for systems that are very simple, but quickly degrades into a Distributed Ball of Mud.

To see why, let’s consider another case. Sometimes, when stock arrives at the warehouse, we discover that items have been water damaged during transit. We can’t sell water damaged sofas, so we have to throw them away, and request more stock from our partners. We also need to update our stock model, and that might mean we need to re-allocate a customer’s order.

Where does this logic go?

Well, the warehouse system knows that the stock has been damaged, so maybe it should own this process, as shown in Figure 11-4?

Figure 11-4. Command Flow 2

[plantuml, command_flow_diagram_2, config=plantuml.cfg]
@startuml

actor w as "Warehouse worker"
entity Warehouse
entity Batches
entity Orders
database CRM


  w -> Warehouse: Report stock damage
  activate Warehouse
  Warehouse -> Batches: Decrease available stock
  Batches -> Batches: Reallocate orders
  Batches -> Orders: Update order status
  Orders -> CRM: Update order history
  deactivate Warehouse

@enduml

This sort of works, too, but now our dependency graph is a mess. In order to allocate stock, the orders service drives the batches system, which drives warehouse; but in order to handle problems at the warehouse, our warehouse system drives batches, which drives orders.

Multiply this by all the other workflows we need to provide, and you can see how services very quickly get tangled up together.

Error Handling in Distributed Systems

“Things break” is a universal law of software engineering. What happens in our system when one of our requests fails? Let’s say that a network error happens right after we take a user’s order for 3 MISBEGOTTEN-RUG, Figure 11-5.

Figure 11-5. Command Flow with Error

[plantuml, command_flow_diagram_with_error, config=plantuml.cfg]
@startuml

actor Customer
entity Orders
entity Batches

Customer -> Orders: Place order
Orders -[#red]x Batches: Confirm reservation
hnote right: network error
Orders --> Customer: ???

@enduml

We have two options here: we can place the order anyway, and leave it unallocated, or we can refuse to take the order since the allocation can’t be guaranteed. The failure state of our batches service has bubbled up and is affecting the reliability of our order service.

When two things have to be changed together, we say that they are coupled. We can think of this failure-cascade as a kind of temporal coupling - every part of the system has to work at the same time for any part of it to work. As the system gets bigger, there is an exponentially increasing probability that some part is degraded.

The Alternative: Temporal Decoupling using Asynchronous Messaging

We’ve already seen part of the answer, which is that we should think in terms of verbs, not nouns. Our domain model is about modelling a business process; it’s not a static data model about a thing, it’s a model of a verb.

So instead of thinking about a system for orders and a system for batches, we think about a system for ordering and a system for allocating, and so on.

When we separate things this way, it’s a little easier to see which system should be responsible for what. When thinking about ordering, really we want to make sure that when we place an order, the order is placed. Everything else can happen later so long as it happens.

Like aggregates, microservices should be consistency boundaries. Between two services, we can accept eventual consistency, and that means we don’t need to rely on synchronous calls. Each service accepts commands from the outside world and raises events to record the result. Other services can listen to those events to trigger the next steps in the workflow.

To avoid the “Distributed BOM” antipattern, instead of temporally coupled HTTP API calls, we want to use asynchronous messaging to integrate our systems. We want our “batch quantity changed” messages to come in as external messages from upstream systems, and we want our system to publish “allocated” events for downstream systems to listen to.

Why is this better? Firstly, because things can fail independently, it’s easier to handle degraded behavior: we can still take orders if the allocations system is having a bad day.

Secondly, we’re reducing the strength of coupling between our systems. If we need to change the order of operations, or to introduce new steps in the process we can do that locally.

Using a Redis Pubsub Channel for Integration

Let’s see how it will all work concretely. We’ll need some way of getting events out of one system and into another, like our message bus, but for services. This piece of infrastructure is often called a Message Broker. The role of a message broker is to take messages from publishers and deliver them to subscribers.

At MADE.com we use Event Store; Kafka or RabbitMQ are valid alternatives. A lightweight solution based on Redis pubsub channels can also work just fine, and since Redis is much more generally familiar to people, we thought we’d use it for this book.

Our new flow will look like this:

Figure 11-6. Sequence diagram for reallocation flow

[plantuml, reallocation_sequence_diagram_with_redis, config=plantuml.cfg]

@startuml

Redis -> MessageBus : BatchQuantityChanged event

group BatchQuantityChanged Handler + Unit of Work 1
    MessageBus -> Domain_Model : change batch quantity
    Domain_Model -> MessageBus : emit Allocate command(s)
end


group Allocate Handler + Unit of Work 2 (or more)
    MessageBus -> Domain_Model : allocate
    Domain_Model -> MessageBus : emit Allocated event(s)
end

MessageBus -> Redis : publish to line_allocated channel
@enduml

Test-Driving It All Using an End-To-End Test

Here’s how we might start with an end-to-end test. We can use our existing API to create batches, and then we’ll test both inbound and outbound messages:

def test_change_batch_quantity_leading_to_reallocation():
    # start with two batches and an order allocated to one of them  
    orderid, sku = random_orderid(), random_sku()
    earlier_batch, later_batch = random_batchref('old'), random_batchref('newer')
    api_client.post_to_add_batch(earlier_batch, sku, qty=10, eta='2011-01-02')  
    api_client.post_to_add_batch(later_batch, sku, qty=10, eta='2011-01-02')  
    response = api_client.post_to_allocate(orderid, sku, 10)  
    assert response.json()['batchref'] == earlier_batch

    subscription = redis_client.subscribe_to('line_allocated')  

    # change quantity on allocated batch so it's less than our order  
    redis_client.publish_message('change_batch_quantity', {  
        'batchref': earlier_batch, 'qty': 5
    })

    # wait until we see a message saying the order has been reallocated  
    messages = []
    for attempt in Retrying(stop=stop_after_delay(3), reraise=True):  
        with attempt:
            message = subscription.get_message(timeout=1)
            if message:
                messages.append(message)
                print(messages)
            data = json.loads(messages[-1]['data'])
            assert data['orderid'] == orderid
            assert data['batchref'] == later_batch

You can read the story of what’s going on in this test from the comments: we want to send an event into the system that causes an order line to be reallocated, and we see that reallocation come out as an event in redis too.

api_client is a little helper that we refactored out to share between our two test types, it wraps our calls to requests.post

redis_client is another test little test helper, the details of which don’t really matter; its job is to be able to send and receive messages from various Redis channels. We’ll use a channel called change_batch_quantity to send in our request to change the quantity for a batch, and we’ll listen to another channel called line_allocated to look out for the expected reallocation.

Because of the asynchronous nature of the system under test, we need to use the tenacity library again to add a retry loop. Firstly because it may take some time for our new line_allocated message to arrive, but also because it won’t be the only message on that channel.

r = redis.Redis(**config.get_redis_host_and_port())


def main():
    orm.start_mappers()
    pubsub = r.pubsub(ignore_subscribe_messages=True)
    pubsub.subscribe('change_batch_quantity')  

    for m in pubsub.listen():
        handle_change_batch_quantity(m)


def handle_change_batch_quantity(m):
    logging.debug('handling %s', m)
    data = json.loads(m['data'])  
    cmd = commands.ChangeBatchQuantity(ref=data['batchref'], qty=data['qty'])  
    messagebus.handle(cmd, uow=unit_of_work.SqlAlchemyUnitOfWork())

r = redis.Redis(**config.get_redis_host_and_port())


def publish(channel, event: events.Event):  
    logging.debug('publishing: channel=%s, event=%s', channel, event)
    r.publish(channel, json.dumps(asdict(event)))

@dataclass
class Allocated(Event):
    orderid: str
    sku: str
    qty: int
    batchref: str

class Product:
    ...
    def allocate(self, line: OrderLine) -> str:
        ...

            batch.allocate(line)
            self.version_number += 1
            self.events.append(events.Allocated(
                orderid=line.orderid, sku=line.sku, qty=line.qty,
                batchref=batch.reference,
            ))
            return batch.reference

HANDLERS = {
    events.Allocated: [handlers.publish_allocated_event],
    events.OutOfStock: [handlers.send_out_of_stock_notification],
}  # type: Dict[Type[events.Event], List[Callable]]

def publish_allocated_event(
        event: events.Allocated, uow: unit_of_work.AbstractUnitOfWork,
):
    redis_eventpublisher.publish('line_allocated', event)

Wrap-Up

Events can come from the outside, but they can also be published externally—our publish handler converts an event to a message on a Redis channel. We use events to talk to the outside world. This kind of temporal decoupling buys us a lot of flexibility in our application integrations, but, as always, it comes at a cost.

Event notification is nice because it implies a low level of coupling, and is pretty simple to set up. It can become problematic, however, if there is a logical flow that runs over various event notifications. […] It can be hard to see as it’s not explicit in any program text; this can make it hard to debug and modify.

Martin Fowler, What do you mean by “Event-Driven”? (martinfowler.com)

If you’re moving from a model of synchronous messaging to an async one, you also open up a whole host of problems to do with message reliability and eventual consistency. Read on to the “Footguns” section in the epilogue.