Optimistic Concurrency with Aggregate Versions

Hello! Welcome back to our course on distributed systems architecture.

In our last lesson, we defined the API contract for a generic event store, establishing the append_events and get_events methods as its core functions. We specifically highlighted the expected_version parameter in append_events as the key to ensuring data integrity in the face of concurrent operations.

Today, we will bring that concept to life. This lesson directly addresses the learning outcome: Implement optimistic concurrency control in an event store service by using aggregate version numbers to detect write conflicts. We will explore the "optimistic" philosophy, detail the mechanics of using version numbers, and then walk through a concrete Python implementation that prevents the kind of race conditions that can corrupt data in high-throughput systems.

1. The Philosophy: Optimistic vs. Pessimistic Concurrency

In systems where multiple processes might try to change the same piece of data simultaneously—like two users trying to book the last seat on a flight, or two automated processes trading the same financial instrument—we need a strategy to prevent conflicts.

There are two main philosophies for this:

• Pessimistic Concurrency Control: This approach is "pessimistic" because it assumes conflicts are likely. It locks a resource (like a database row or table) when a process begins to work with it, preventing any other process from accessing it until the first one is finished. This is safe but can create performance bottlenecks, as processes spend time waiting for locks to be released. A database's SELECT FOR UPDATE is a classic example of pessimistic locking.

• Optimistic Concurrency Control (OCC): This approach is "optimistic" because it assumes conflicts are rare. It allows multiple processes to read and work with the same data concurrently without any initial locking. Only at the moment of writing the changes back to the database does it check if the data has been modified by another process in the meantime. If a conflict is detected, the write operation fails, and it's up to the application to handle the failure, typically by retrying the entire operation.

For many distributed systems, especially those requiring high throughput, OCC is preferred because it avoids the overhead of managing locks when conflicts are infrequent.

Let's watch a segment of a talk that illustrates this concept with a simple bank account example.

As the video explains, the core idea is to check if the state has changed since you last read it. If it has, your command was based on stale data, and the system rejects it to maintain consistency.

This diagram illustrates the conflict scenario described in the video:

2. The Mechanism: Aggregate Versioning

To implement OCC, we need a way to track changes to an aggregate. We do this by giving each aggregate a version number.

• An aggregate's version is the sequence number of the last event that was applied to it.
• When we load an aggregate from the event store, we replay its events to get its current state and also note its current version.
• When we execute a command that changes the aggregate's state, we generate one or more new events. The aggregate's internal version number is incremented for each new event.
• When we call append_events, we pass the version of the aggregate before the new events were applied as the expected_version.

This raises a design question: where should the version number be managed? Is it a domain concern or an infrastructure concern? The book Cosmic Python argues that while it feels like an infrastructure detail, including it in the domain model is often the cleanest solution.

By placing version_number on the aggregate itself and incrementing it within the business method (like allocate), we make the versioning explicit and part of the aggregate's state change logic.

3. Implementing OCC in the Event Store

Now we get to the core of the lesson: implementing the append_events method from our IEventStore contract to enforce OCC. The implementation requires a two-level defense: a check in our application code and a constraint in our database.

Let's look at a complete implementation that uses a PostgreSQL database.

Let's break down the implementation from the article into its key components.

3.1. The Database-Level Guarantee

The foundation of our OCC implementation is a database constraint.

CREATE TABLE events (
    id SERIAL PRIMARY KEY,
    aggregate_id UUID NOT NULL,
    version INT NOT NULL,
    -- ... other columns
    UNIQUE (aggregate_id, version)
);

The UNIQUE (aggregate_id, version) constraint is our ultimate safety net. It makes it physically impossible for the database to store two events for the same aggregate with the same version number. If two concurrent transactions try to commit an event for aggregate_id='A' with version=5, the database will allow the first one to succeed and reject the second with an integrity violation error. This prevents data corruption even if there's a bug in our application-level check.

3.2. The Application-Level Check

Relying solely on the database constraint is not ideal because it provides a generic error. We want to provide a clear, specific ConcurrencyException to the caller. This is done with an explicit check inside our save_events (or append_events) method.

Here's the pseudocode for the logic within a database transaction:

# Inside EventStore.append_events(stream_id, events, expected_version)

# 1. Start a database transaction.
#    (In the resource's example, this is managed by the try/except/finally block with conn.commit/rollback)

# 2. Get the current version of the aggregate from the database.
db_cursor.execute(
    "SELECT version FROM events WHERE aggregate_id = ? ORDER BY version DESC LIMIT 1",
    (stream_id,)
)
current_db_version = db_cursor.fetchone() or 0

# 3. The Optimistic Concurrency Check.
if current_db_version != expected_version:
    # Another process has saved events since we read the aggregate.
    raise ConcurrencyException(f"Conflict: Expected version {expected_version}, but found {current_db_version}.")

# 4. If the check passes, insert the new events.
#    The new events will have versions: expected_version + 1, expected_version + 2, ...
for event in events:
    db_cursor.execute(
        "INSERT INTO events (aggregate_id, version, ...)",
        (stream_id, event.version, ...)
    )

# 5. Commit the transaction.
#    If a concurrent transaction committed between steps 2 and 5,
#    the UNIQUE constraint will trigger an IntegrityError here,
#    causing a rollback.

This implementation provides two layers of protection:

1. The explicit if check: Catches most conflicts and raises our specific ConcurrencyException.
2. The UNIQUE constraint: Catches the rare race condition where another transaction commits after our SELECT but before our INSERT, ensuring correctness.

When a caller receives a ConcurrencyException, the standard response is to retry the entire business operation: reload the aggregate (which will now have the latest version), re-run the command logic, and attempt to save the new events again.

4. Testing for Concurrency

How can we be confident that our implementation works? We need to write an integration test that reliably simulates a race condition.

The typical way to do this is to use threading. The test can:

1. Set up an initial state for an aggregate in the database.
2. Start two separate threads.
3. Have both threads attempt to perform an operation on the same aggregate. To ensure they interleave in a way that causes a conflict, you can add a small time.sleep() in the first thread's transaction after it reads but before it writes.
4. Assert that one of the threads succeeded (the aggregate's version was incremented) and the other failed with a ConcurrencyException.

The Cosmic Python book provides a great example of such a test. You don't need to implement it now, but it's valuable to see the pattern.

Conclusion

In this lesson, we have moved from the abstract concept of the expected_version parameter to a concrete and robust implementation of optimistic concurrency control.

Key Takeaways:

• Optimistic Concurrency Control (OCC) is a strategy that assumes conflicts are rare, avoiding upfront locking in favor of a check-at-commit time.
• The mechanism relies on versioning aggregates. The aggregate's version number is passed as the expected_version when saving new events.
• A robust implementation uses a two-level defense: an application-level check that compares the expected_version with the current database version, and a database-level UNIQUE constraint on (aggregate_id, version) as a final guarantee of consistency.
• When a ConcurrencyException is raised, the client is responsible for retrying the operation.

Preview of the Next Lesson:
We have now designed a generic event store API and implemented its most critical feature for ensuring data consistency. Our implementation used a relational database (PostgreSQL). However, relational databases are not the only option. In the next lesson, we will compare the trade-offs between different persistence technologies for implementing an event store, looking at relational databases, document databases (like MongoDB), and specialized event store databases (like EventStoreDB). We will analyze how the choice of technology impacts implementation, performance, and operational complexity.