Optimizing Aggregate State Reconstruction with Snapshots

Hello! Welcome back to our module on State Management with Event Sourcing.

In our last lesson, we implemented event replay, the fundamental mechanism for reconstructing an aggregate's state from its history. We also identified its primary drawback: replaying a long stream of events for a long-lived aggregate can become a significant performance bottleneck. Your experience building high-throughput financial systems at Revolut has likely given you a practical appreciation for why such bottlenecks must be addressed proactively in system design.

Today, we will tackle this challenge head-on. This lesson is dedicated to the learning outcome: Implement snapshots for an aggregate to optimize state reconstruction performance for long-lived objects. We will explore what snapshots are, the strategies for using them, and how to implement them in our FXTrade example.

1. The Snapshotting Mechanism

A snapshot is a performance optimization. It's a serialized representation of an aggregate's state at a specific version. Instead of replaying events from the very beginning, we can start from a much more recent state, drastically reducing the number of events we need to process.

The process of rehydrating an aggregate using a snapshot is as follows:

1. Retrieve the most recent snapshot for the aggregate.
2. Deserialize the snapshot and apply its state to a newly instantiated aggregate object.
3. Query the event store for all events that occurred after the snapshot's version.
4. Replay this smaller, subsequent list of events to bring the aggregate to its true current state.

This video provides an excellent conceptual overview and demonstration of this mechanism.

As the video explains, the snapshot acts as a checkpoint. You restore the state from the last checkpoint and then process only the changes that have happened since. This is analogous to financial reconciliation, where you start with a summarized closing balance rather than re-calculating from every single transaction since the account was opened.

2. Snapshotting Strategies: When, Where, and How

While the concept is straightforward, implementing snapshots involves several important design decisions. Adding this optimization introduces complexity, so these choices must be made carefully.

When to Take a Snapshot?

The trigger for creating a snapshot is a critical tuning parameter. Creating them too frequently can degrade write performance, while creating them too infrequently diminishes their benefit.

To summarize the common strategies:

• Every N events: A simple and popular method. For example, create a snapshot every 100 events. This provides a predictable upper bound on the number of events that need replaying.
• On a specific event type: This is domain-driven. For example, in a financial context, you might snapshot an account at the end of each month when a MonthEndStatementGenerated event occurs.
• On a schedule: A background process creates snapshots for active aggregates periodically (e.g., nightly).
• After each event: This essentially turns your event store into a state store, often negating some benefits of event sourcing and potentially harming write performance. It's generally used only in specific scenarios.

Where to Store Snapshots?

Snapshots can be stored in various ways, each with its own trade-offs:

• In a separate event stream: As shown in the CodeOpinion video, you can have a stream like trade-123-snapshot that stores snapshot events. You can configure this stream to only keep the latest event ($maxCount: 1), which is a clean approach.
• In a separate database/cache: Storing snapshots in a key-value store (like Redis) or a document database (like MongoDB) is also common. This separates the optimization mechanism from the primary event store but adds another moving part to your infrastructure.
• In the same event stream: This is less common. It involves storing the snapshot as a special event within the main event stream. This can complicate the stream and make schema evolution harder, as old, obsolete snapshot events remain part of the history.

How to Update Snapshots?

This decision concerns whether the snapshot is updated as part of the command-processing (write) path or the query (read) path.

The key takeaways from this are:

• Updating on read is an anti-pattern: It complicates the read path, adds latency, and can cause unnecessary database writes.
• Synchronous update on write is often the simplest viable approach: When your application saves new events for an aggregate, it also checks if a snapshot should be created and saves it within the same operation or transaction. This keeps the snapshot perfectly in sync but can add latency to the command processing.
• Asynchronous update is the most scalable approach: The command handler only writes events. A separate, independent background process listens for new events and creates/updates snapshots. This decouples the systems but introduces eventual consistency for the snapshots and requires a reliable mechanism (like database subscriptions) to avoid missing events.

For our implementation, we will use a synchronous update on write strategy, as it's the most direct way to demonstrate the complete logic.

3. Implementing Snapshots in Python

Let's now modify our FXTrade system to incorporate snapshots. We will implement the following logic:

1. Define a Snapshot data structure.
2. Modify our repository's loading logic to use snapshots.
3. Modify our repository's saving logic to create snapshots based on a trigger.

For our example, we'll use a simple trigger: create a snapshot every 5 versions.

Step 1: Define the Snapshot and Repository Interfaces

First, let's define what a snapshot looks like and what our storage interfaces will be. We'll need a way to store both events and snapshots.

import uuid
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any

# Assume our Event classes (FXTradeInitiated, etc.) are defined as before

@dataclass
class Snapshot:
    """A serialized representation of an aggregate's state at a specific version."""
    aggregate_id: uuid.UUID
    state: Dict[str, Any]
    version: int

class IEventStore:
    """Interface for our event storage."""
    def get_events(self, aggregate_id: uuid.UUID, after_version: int = 0) -> List[Any]:
        raise NotImplementedError

    def save_events(self, aggregate_id: uuid.UUID, events: List[Any], expected_version: int):
        raise NotImplementedError

class ISnapshotStore:
    """Interface for our snapshot storage."""
    def get_snapshot(self, aggregate_id: uuid.UUID) -> Optional[Snapshot]:
        raise NotImplementedError
    
    def save_snapshot(self, snapshot: Snapshot):
        raise NotImplementedError

Step 2: Update the FXTrade Aggregate

The aggregate itself needs a way to be restored from a snapshot and to create a new one.

class FXTrade:
    # __init__, command methods (initiate, book, etc.), and _apply remain the same
    # ...

    def __init__(self):
        self.id = None
        self.status = None
        self.currency_pair = None
        self.dealt_amount = None
        self.version = 0
        self._uncommitted_events = []

    def _apply(self, event):
        # ... same as before
        if isinstance(event, FXTradeInitiated):
            self.id = event.trade_id
            self.status = 'INITIATED'
            # ...
        # ... other event types
        self.version += 1

    # NEW: Method to create a snapshot
    def to_snapshot(self) -> Snapshot:
        """Creates a snapshot of the current state."""
        state_dict = {
            "id": self.id,
            "status": self.status,
            "currency_pair": self.currency_pair,
            "dealt_amount": self.dealt_amount,
            # ... serialize all other state fields
        }
        return Snapshot(aggregate_id=self.id, state=state_dict, version=self.version)

    # NEW: Factory method to load from a snapshot
    @classmethod
    def from_snapshot(cls, snapshot: Snapshot):
        """Restores an aggregate from a snapshot."""
        trade = cls()
        
        # Apply the snapshot's state
        trade.id = snapshot.state.get("id")
        trade.status = snapshot.state.get("status")
        trade.currency_pair = snapshot.state.get("currency_pair")
        trade.dealt_amount = snapshot.state.get("dealt_amount")
        # ... restore all other fields
        
        # Set the version from the snapshot
        trade.version = snapshot.version
        
        return trade

Step 3: Update the Repository/Application Logic

The orchestration logic is where everything comes together. The repository will now interact with both the event store and the snapshot store.

class TradeRepository:
    SNAPSHOT_INTERVAL = 5 # Take a snapshot every 5 versions

    def __init__(self, event_store: IEventStore, snapshot_store: ISnapshotStore):
        self._event_store = event_store
        self._snapshot_store = snapshot_store

    def get(self, trade_id: uuid.UUID) -> FXTrade:
        """Loads an aggregate, using a snapshot if available."""
        
        # 1. Try to load from a snapshot
        snapshot = self._snapshot_store.get_snapshot(trade_id)
        
        if snapshot:
            # Restore from snapshot and get version
            trade = FXTrade.from_snapshot(snapshot)
            version_to_load_from = snapshot.version
        else:
            # No snapshot, start from scratch
            trade = FXTrade()
            version_to_load_from = 0

        # 2. Load subsequent events
        events = self._event_store.get_events(trade_id, after_version=version_to_load_from)
        
        # 3. Replay subsequent events
        for event in events:
            trade._apply(event)
            
        return trade

    def save(self, trade: FXTrade):
        """Saves uncommitted events and potentially a new snapshot."""
        if not trade._uncommitted_events:
            return

        # The expected version for optimistic concurrency control is the version
        # before any new events were applied.
        expected_version = trade.version - len(trade._uncommitted_events)
        
        self._event_store.save_events(
            trade.id,
            trade._uncommitted_events,
            expected_version
        )
        
        # Clear uncommitted events after saving
        trade._uncommitted_events = []

        # Check if we should create a snapshot
        # We check if the version has crossed a multiple of our interval
        if (expected_version // self.SNAPSHOT_INTERVAL) < (trade.version // self.SNAPSHOT_INTERVAL):
            snapshot = trade.to_snapshot()
            self._snapshot_store.save_snapshot(snapshot)
            print(f"Snapshot created for trade {trade.id} at version {trade.version}")

This implementation demonstrates the full synchronous-on-write cycle. The get method is optimized to use snapshots, and the save method is responsible for creating them when the version count crosses our defined interval.

The detailed TypeScript implementation in the Kurrent.io article provides further examples of these patterns, particularly around storing snapshots in separate streams.

Conclusion

Today we have addressed a critical performance issue in event sourcing by implementing snapshots. This is a classic trade-off in distributed systems: we've added complexity to our write path and storage in order to gain a significant optimization on our read path.

Key Takeaways:

• Snapshots are "checkpoints" of an aggregate's state that prevent the need to replay an entire event history.
• The rehydration process becomes: load the latest snapshot, then replay only the events that have occurred since.
• Implementing snapshots requires strategic decisions about the trigger (when to create one), storage (where to keep it), and update mechanism (synchronous vs. asynchronous).
• A synchronous-on-write approach is a direct way to implement snapshotting, where the repository orchestrates saving both events and snapshots.

Preview of the Next Lesson:
Snapshots are most useful for long-lived aggregates. However, long-lived aggregates are also the most likely to undergo changes in their structure and business logic over time. If we change the structure of our FXTrade aggregate, how do we handle old events (and old snapshots) that were created with the previous structure? This is the problem of schema evolution. In our next lesson, we will explore common strategies for managing these changes, such as upcasting.