CQRS – Validating Commands with Required Queries

cqrsevent-sourcingeventual-consistency

I'm aware that this question has been asked several times, but I have some concerns regarding querying from the write side that I don't see addressed in the already existing questions, more especifically regarding eventual consistency in the command model.

I have a simple CQRS + ES architecture for an application. Customers can buy stuff from my site, but there is a hardcoded requirement: A customer cannot purchase more than 500$ of products from our store. If they try, the purchase should not be accepted.

So, this is how my command handler looks like (in python, and simplified from concerns like currencies, injection for the sake of simplicity):

class NewPurchaseCommand:
    customer_id: int
    product_ids: List[int]

class PurchasesCommandHandler:
    purchase_repository: PurchaseRepository
    product_repository: ProductRepository
    customer_query_service: CustomerQueryService

    def handle(self, cmd: NewPurchaseCommand):
        current_amount_purchased = self.customer_query_service.get_total(cmd.customer_id)

        purchase_amount = 0
        for product_id in cmd.product_ids:
            product = self.product_repository.get(product_id)
            purchase_amount += product.amount

        if current_amount_purchase + purchase_amount > 500:
             raise Exception('You cannot purchase over 500$')

        new_purchase = Purchase.create(cmd.customer_id, cmd.product_ids)
        self.purchase_repository.save(new_purchase)

        # Then, after the purchase is saved, a PurchaseCreated event is persisted, 
        # sent to a queue which will then update several read projections, which one 
        # of them is the underlying table that the customer_query_service uses.

The CustomerQueryService uses an underlying table to quickly retrieve the amount that the user has purchased at the time being, and this table is exclusively used by the write side, and updated eventually:

CustomerPurchasedAmount table
CustomerId | Amount
10         | 480

While my command handler works on simple scenarios, I want to know how to handle the possible edge cases that might happen:

  • This user 10, which is a malicious one, makes two purchases at the same time of 20$. But since the CustomerPurchasedAmount table is updated eventually, both requests will succeed (this is the case I'm most concerned)
  • There might exist the possibility that some product price might change while processing the request (unlikely, but then again, it can happen).

My questions are:

  • How can I protect the command from the concurrency case exposed before?
  • How should read models specifically tailored for the write side be updated? Synchronously? Asynchronously like I'm doing right now?
  • And in general, how should command validation happen if the information you are querying in order to validate might be stale?

Best Answer

How can I avoid and protect the command from the concurrency case exposed before?

The only way to "protect" yourself against concurrent changes is to hold a lock, which effectively means to have both of the changes be part of the same thing. Once you have decided to distribute the information, concurrency is unavoidable.

In some cases, you can mitigate by rethinking the model so that you are working with immutable values. For instance, instead of asking for the price "now", you ask for the price at a particular time, and you take steps to ensure that for any given time there is only one price (think quotes, or sales; "offer good until 2019-12-31).

How should read models specifically tailored for the write side be updated? Synchronously? Asynchronously like I'm doing right now?

Usually asynchronously, but largely "it depends". The "read model" being used by the write side is closer in form to a locally cached copy. This changes the line of thinking to something more like "what happens if we have a cache miss?"

Sometimes, the right answer is to fail, sometimes the right answer is to accept the offered change provisionally, and defer complete processing until the information is available.

how should command validation happen if the information you are querying in order to validate might be stale

What I've found is that you need to stop thinking of command processing as being a linear sequence of transitions along the happy path, and begin thinking instead about the processing being a state machine.

We receive an order, and therefore we need price checks on A, B, and C. The price for A is available, so we pass that in, and now we are in a state that needs prices for B and C. No other checks complete within the time limit, so we save the current work, schedule it to be resumed later, and return a response indicating that the order is in processing.

If you want the advantages of distributed autonomous services, you have to let go of the concept of centralized control.

Related Topic