Architecture Patterns event-sourcing snapshots cqrs Oleksandr Andrushchenko Published on — Modified on

Snapshotting and Projection Rebuilding in Event-Sourced Systems

0 0

Snapshotting and projection rebuilding solve two different performance and reliability problems in event-sourced systems. As an aggregate accumulates events, reconstructing its state by replaying the entire stream can become expensive. A snapshot stores a checkpoint of that aggregate's state so subsequent commands only need to replay newer events. Projection rebuilding addresses a different problem: restoring or changing a read model by replaying the authoritative event history.

Both operations depend on a fundamental Event Sourcing principle: stored events remain the source of truth, while snapshots and projections are derived data. Treating snapshots as authoritative, losing projection checkpoints, or replaying events while triggering external side effects can introduce serious consistency problems. This guide explains snapshot versions, replay algorithms, projection checkpoints, safe rebuilding strategies, and production trade-offs. It also includes a Python, FastAPI, PostgreSQL, SQLAlchemy, and pytest implementation that demonstrates aggregate snapshots and a transactional account-balance projection rebuild.

Table of Contents

What Are Snapshotting and Projection Rebuilding?

In Event Sourcing, an aggregate's state is reconstructed from a durable sequence of events. This enables historical queries and replay, but also creates practical questions about performance and maintenance. What happens when one account has hundreds of thousands of events? How do you repair a read model after discovering a calculation bug?

Aggregate snapshots and projection rebuilding address these problems at different boundaries. Snapshots optimize how the application loads its write-side business state. Projection rebuilding restores or replaces read-side representations derived from the event history.

Snapshots vs. Projections vs. Event Stores

These concepts are related but should not be treated as interchangeable.

Component Purpose Authoritative?
Event store Persist accepted business events in order Yes
Aggregate snapshot Accelerate reconstruction of one aggregate No
Read projection Provide query-optimized current state No
Projection checkpoint Track which events have been processed No

A snapshot might contain an account's balance and stream version. A read projection might contain account balances together with customer names, transaction counts, or reporting categories.

Why Event-Sourced Systems Need These Operations

Without snapshots, every balance-changing command might need to load and replay an account's entire event history. If the stream has 100,000 events, even a simple deposit may require unnecessary database reads and CPU work.

Without a reliable projection rebuild process, fixing a bug in a read model can be difficult. A team might attempt to patch individual rows manually without knowing whether every affected record was corrected.

Snapshotting improves aggregate-loading performance. Projection rebuilding provides a repeatable way to reconstruct derived state after changes, failures, or new requirements.

How Aggregate Snapshotting Works

A snapshot captures an aggregate's state at a specific event-stream version. When the aggregate is loaded again, the application can start from that snapshot and apply only subsequent events.

The essential requirement is that the snapshot and stream positions agree. A snapshot at version 250 represents state after applying events through version 250, not merely a timestamp near the 250th event.

Loading a Snapshot and Replaying Newer Events

Suppose an account has 1,050 events and a valid snapshot at version 1,000.

Instead of replaying all 1,050 events, the application loads the snapshot and applies versions 1,001 through 1,050.

# Conceptual aggregate reconstruction.
#
# load_snapshot and load_events_after are
# provided by the persistence adapter.

def restore_account(account_id: str):
    snapshot = load_snapshot(account_id)

    if snapshot is None:
        account = Account()
        version = 0
    else:
        account = Account.from_snapshot(
            snapshot.state
        )
        version = snapshot.version

    events = load_events_after(
        account_id,
        version,
    )

    for event in events:
        account.apply(event)

    return account

Replay remains necessary because new events can arrive after a snapshot is created.

Choosing Snapshot Frequency

There is no universal snapshot interval. Creating a snapshot after every event can increase database writes and storage without delivering meaningful performance benefits.

Common strategies include:

  • Fixed event interval: Snapshot every 100, 500, or 1,000 events.
  • Replay-time threshold: Snapshot when aggregate loading exceeds a measured latency target.
  • Aggregate-size threshold: Snapshot large or expensive aggregates more frequently.
  • Background snapshotting: Produce snapshots asynchronously using a consistent event-stream version.

For example, snapshotting every 500 events means a normally loaded aggregate needs to replay at most approximately 499 newer events, assuming snapshot creation keeps up and there are no failures or compatibility issues.

The right interval depends on event payload size, replay complexity, database throughput, and workload characteristics.

Snapshot Versions, Validation, and Recovery

A snapshot should include at least the aggregate identifier, aggregate version, snapshot schema version, and serialized state.

Before using one, the application should verify that it is compatible with the current snapshot decoder and that its version does not exceed the authoritative stream's current version.

For stronger integrity guarantees, a system may also store hashes or other consistency metadata.

If a snapshot is missing or incompatible, the application should be able to reconstruct the aggregate from the event store.

However, silently accepting corrupted snapshot state is dangerous. An invalid snapshot can lead to incorrect business decisions even when the underlying event history is correct.

A robust implementation should reject unsupported snapshot formats, report the error, and use a verified replay fallback where possible.

How Projection Rebuilding Works

Projections transform event history into representations optimized for queries. Common examples include account balances, order summaries, inventory availability views, and reporting dashboards.

A projection rebuild discards or replaces derived state and reprocesses the necessary events to generate a correct representation.

Projections and Read Models

Consider an AccountBalance projection containing an account identifier, balance, and last processed stream version.

Its update logic might apply events as follows:

from dataclasses import dataclass


@dataclass
class BalanceProjection:
    balance_cents: int = 0
    version: int = 0

    def apply(self, event):
        if event.version != self.version + 1:
            raise ValueError("Projection event gap")

        if event.event_type == "AccountOpened":
            self.balance_cents = 0

        elif event.event_type == "MoneyDeposited":
            self.balance_cents += event.amount_cents

        elif event.event_type == "MoneyWithdrawn":
            self.balance_cents -= event.amount_cents

        else:
            raise ValueError("Unknown event")

        self.version = event.version

This projection is intentionally simpler than the write-side aggregate. It does not decide whether a withdrawal is allowed; the aggregate performs command validation before accepting the event.

Projection Checkpoints and Idempotency

A projection checkpoint records the last successfully processed event position. It allows processing to resume after a crash without restarting the entire history.

A critical distinction is that a checkpoint must advance only after the corresponding projection update is durable.

If the read model and checkpoint are stored in the same database, updating both in one transaction is a convenient way to preserve this guarantee.

If they are stored separately, the system must provide idempotent projection updates and a recovery strategy for partial failures.

Checkpoints can use an aggregate-local sequence or a global event position, depending on the projection.

A projection combining events from several aggregates usually needs a reliable global cursor or another ordering strategy. An account-local version cannot be used as a global position across unrelated event streams.

In-Place vs. Blue-Green Rebuilding

Two common approaches are rebuilding an existing read model and building a replacement alongside it.

Strategy Advantages Trade-offs
In-place rebuild Simpler implementation, fewer tables May block writes or temporarily interrupt queries
Blue-green rebuild Old projection stays available during reconstruction Requires parallel storage, catch-up, validation, and switching

An in-place rebuild can be appropriate for a local development environment or small system with a controlled maintenance window.

In production, rebuilding a large projection into a new version is often safer.

A replacement projection can be generated and checked while readers continue using the old one. Once the replacement has caught up and passed validation, a controlled switch redirects readers to it.

Rebuilding While New Events Arrive

Suppose a new projection is built through global event position 1,000,000. While it is rebuilding, the application writes another 5,000 events.

Switching to the new projection immediately would expose stale state.

A safer online workflow is:

  1. Create a new projection generation without modifying the active one.
  2. Record a stable starting event position or consistent read boundary.
  3. Replay historical events into the new generation.
  4. Process additional events that arrived during the rebuild.
  5. Validate the new projection against expected results.
  6. Perform a controlled cutover once consistency requirements are met.
  7. Keep the previous generation available temporarily for rollback.

The precise cutover protocol depends on the event store and read database. Some systems temporarily pause relevant writers or acquire a short coordination lock. Others switch a projection pointer after the replacement has reached a verified checkpoint.

The example later in this article demonstrates a simpler transactional rebuild of one account projection. It does not claim to implement a zero-downtime, multi-million-event rebuild.

Production Design and Failure Scenarios

Snapshotting and projection rebuilding become especially useful when the event store has already been in production long enough for event streams and read models to accumulate significant history.

Event-Sourced Account Example

Consider an account application with three event types:

  • AccountOpened: Establishes an account with zero balance.
  • MoneyDeposited: Increases the balance.
  • MoneyWithdrawn: Decreases the balance.

The event stream is authoritative. An aggregate snapshot allows command handlers to restore the account efficiently, while a current-balance projection allows API queries without replaying the stream on every request.

If a projection implementation contains a bug, the application can rebuild the balance by processing historical events using corrected logic.

If a snapshot is incompatible with a new application release, the system can ignore that snapshot and reconstruct the aggregate from accepted events.

Corrupt Snapshots, Missing Events, and Projection Drift

Several failure conditions deserve explicit handling:

  • Corrupt snapshot: Reject the snapshot, alert, and reconstruct from events where possible.
  • Unsupported snapshot schema: Use a compatible decoder or fall back to full replay.
  • Missing stream event: Stop reconstruction rather than silently skipping a version.
  • Projection lag: Report the processed checkpoint and expected lag.
  • Projector crash: Resume from a durable checkpoint.
  • Duplicate delivery: Ignore already processed events without reapplying their effects.
  • Historical event incompatibility: Use versioned decoders or upcasters.
  • Rebuild failure: Keep the previous serving projection when using a generation-based strategy.

Instead, determine whether the difference comes from stale processing, invalid event interpretation, missing history, or a defect in the projection code.

Ready-to-Use Example with FastAPI and PostgreSQL

This example extends an event-sourced account application with aggregate snapshots and a rebuildable balance projection. It stores all three representations in PostgreSQL to demonstrate their relationship clearly.

The event store remains authoritative. A snapshot is saved every three events for demonstration purposes, and the projection can be rebuilt from the full event history through an explicit maintenance endpoint.

Project Structure and Dependencies

Create the following project structure, including empty __init__.py files in the Python package directories:

event-sourcing-recovery/
  app/
    __init__.py
    domain.py
    persistence.py
    service.py
    api.py
  tests/
    test_domain.py
  requirements.txt
  docker-compose.yml

Create requirements.txt:

fastapi>=0.115,<1
uvicorn>=0.30,<1
sqlalchemy>=2.0,<3
psycopg[binary]>=3.2,<4
pytest>=8,<10
httpx>=0.27,<1

For production, use tested, pinned dependency versions and versioned database migrations. The example below uses automatic table creation only for local development.

Domain Events and Aggregate Reconstruction

The domain model supports deterministic replay, restoring state from a snapshot, and producing new deposit or withdrawal events.

Create app/domain.py:

from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class AccountEvent:
    version: int
    event_type: str
    data: dict[str, Any]


@dataclass(frozen=True)
class AccountSnapshot:
    version: int
    balance_cents: int
    is_open: bool
    schema_version: int = 1


class Account:
    def __init__(
        self,
        balance_cents: int = 0,
        version: int = 0,
        is_open: bool = False,
    ):
        self.balance_cents = balance_cents
        self.version = version
        self.is_open = is_open

    @classmethod
    def from_snapshot(
        cls,
        snapshot: AccountSnapshot,
    ) -> "Account":
        if snapshot.schema_version != 1:
            raise ValueError("Unsupported snapshot schema")

        if snapshot.version < 1:
            raise ValueError("Invalid snapshot version")

        if not snapshot.is_open:
            raise ValueError("Invalid account snapshot")

        if snapshot.balance_cents < 0:
            raise ValueError("Invalid snapshot balance")

        return cls(
            balance_cents=snapshot.balance_cents,
            version=snapshot.version,
            is_open=snapshot.is_open,
        )

    def snapshot(self) -> AccountSnapshot:
        if not self.is_open:
            raise ValueError("Account not opened")

        return AccountSnapshot(
            version=self.version,
            balance_cents=self.balance_cents,
            is_open=True,
        )

    @staticmethod
    def validate_amount(amount: int) -> None:
        if type(amount) is not int or amount <= 0:
            raise ValueError("Amount must be positive")

    def apply(self, event: AccountEvent) -> None:
        if event.version != self.version + 1:
            raise ValueError("Event version gap")

        if event.event_type == "AccountOpened":
            if self.is_open:
                raise ValueError("Already opened")

            self.is_open = True

        elif event.event_type == "MoneyDeposited":
            if not self.is_open:
                raise ValueError("Account not opened")

            amount = event.data["amount_cents"]
            self.validate_amount(amount)
            self.balance_cents += amount

        elif event.event_type == "MoneyWithdrawn":
            if not self.is_open:
                raise ValueError("Account not opened")

            amount = event.data["amount_cents"]
            self.validate_amount(amount)

            if amount > self.balance_cents:
                raise ValueError("Invalid historical balance")

            self.balance_cents -= amount

        else:
            raise ValueError("Unknown event type")

        self.version = event.version

    def deposit(self, amount: int) -> AccountEvent:
        if not self.is_open:
            raise ValueError("Account not opened")

        self.validate_amount(amount)

        return AccountEvent(
            version=self.version + 1,
            event_type="MoneyDeposited",
            data={"amount_cents": amount},
        )

    def withdraw(self, amount: int) -> AccountEvent:
        if not self.is_open:
            raise ValueError("Account not opened")

        self.validate_amount(amount)

        if amount > self.balance_cents:
            raise ValueError("Insufficient funds")

        return AccountEvent(
            version=self.version + 1,
            event_type="MoneyWithdrawn",
            data={"amount_cents": amount},
        )


def replay(
    events: list[AccountEvent],
    snapshot: AccountSnapshot | None = None,
) -> Account:
    account = (
        Account.from_snapshot(snapshot)
        if snapshot is not None
        else Account()
    )

    for event in events:
        account.apply(event)

    return account

The replay function accepts an optional snapshot. When one is provided, callers must supply only events after that snapshot's version.

The apply() method rejects missing or nonconsecutive event versions. It also validates historical event types and monetary values rather than silently accepting malformed history.

Event Store, Snapshots, and Read Models

The persistence layer defines three tables: authoritative events, aggregate snapshots, and an account read projection.

Create app/persistence.py:

import os

from sqlalchemy import (
    Boolean,
    CheckConstraint,
    ForeignKey,
    Integer,
    JSON,
    String,
    UniqueConstraint,
    create_engine,
    select,
)
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    Session,
    mapped_column,
    sessionmaker,
)

from app.domain import AccountEvent, AccountSnapshot


DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql+psycopg://app:app@localhost:5432/accounts",
)

engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
)

SessionFactory = sessionmaker(
    bind=engine,
    expire_on_commit=False,
)


class Base(DeclarativeBase):
    pass


class AccountRow(Base):
    __tablename__ = "accounts"

    account_id: Mapped[str] = mapped_column(
        String(100),
        primary_key=True,
    )

    balance_cents: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
        default=0,
    )

    version: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
        default=0,
    )


class EventRow(Base):
    __tablename__ = "account_events"

    __table_args__ = (
        UniqueConstraint(
            "account_id",
            "version",
            name="uq_account_stream_version",
        ),
        CheckConstraint(
            "version >= 1",
            name="ck_event_positive_version",
        ),
    )

    event_id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
        autoincrement=True,
    )

    account_id: Mapped[str] = mapped_column(
        String(100),
        ForeignKey("accounts.account_id"),
        nullable=False,
        index=True,
    )

    version: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
    )

    event_type: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
    )

    data: Mapped[dict] = mapped_column(
        JSON,
        nullable=False,
    )


class SnapshotRow(Base):
    __tablename__ = "account_snapshots"

    account_id: Mapped[str] = mapped_column(
        String(100),
        ForeignKey("accounts.account_id"),
        primary_key=True,
    )

    version: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
    )

    balance_cents: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
    )

    is_open: Mapped[bool] = mapped_column(
        Boolean,
        nullable=False,
    )

    schema_version: Mapped[int] = mapped_column(
        Integer,
        nullable=False,
        default=1,
    )


def load_events(
    session: Session,
    account_id: str,
    after_version: int = 0,
) -> list[AccountEvent]:
    rows = session.scalars(
        select(EventRow)
        .where(
            EventRow.account_id == account_id,
            EventRow.version > after_version,
        )
        .order_by(EventRow.version)
    ).all()

    return [
        AccountEvent(
            version=row.version,
            event_type=row.event_type,
            data=row.data,
        )
        for row in rows
    ]


def load_snapshot(
    session: Session,
    account_id: str,
) -> AccountSnapshot | None:
    row = session.get(SnapshotRow, account_id)

    if row is None:
        return None

    return AccountSnapshot(
        version=row.version,
        balance_cents=row.balance_cents,
        is_open=row.is_open,
        schema_version=row.schema_version,
    )

The composite uniqueness constraint on account ID and stream version prevents two events from occupying the same position in one account's history.

The account table is a read projection and also provides a row that can be locked to serialize commands for the same account.

Commands, Snapshot Creation, and Rebuilding

The application service uses snapshots for normal command processing but deliberately ignores snapshots when rebuilding the read projection. That ensures the rebuild uses the authoritative event history.

Create app/service.py:

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from app.domain import (
    Account,
    AccountEvent,
    replay,
)
from app.persistence import (
    AccountRow,
    EventRow,
    SnapshotRow,
    SessionFactory,
    load_events,
    load_snapshot,
)


SNAPSHOT_INTERVAL = 3


class AccountService:
    def open_account(self, account_id: str) -> dict:
        if not account_id.strip():
            raise ValueError("Account ID required")

        try:
            with SessionFactory.begin() as session:
                session.add(
                    AccountRow(
                        account_id=account_id,
                        balance_cents=0,
                        version=1,
                    )
                )

                session.add(
                    EventRow(
                        account_id=account_id,
                        version=1,
                        event_type="AccountOpened",
                        data={},
                    )
                )

                # Flush detects conflicting IDs
                # before leaving the transaction.
                session.flush()

        except IntegrityError as exc:
            raise ValueError(
                "Account already exists"
            ) from exc

        return {
            "account_id": account_id,
            "balance_cents": 0,
            "version": 1,
        }

    def deposit(
        self,
        account_id: str,
        amount: int,
    ) -> dict:
        return self._change(
            account_id,
            amount,
            "deposit",
        )

    def withdraw(
        self,
        account_id: str,
        amount: int,
    ) -> dict:
        return self._change(
            account_id,
            amount,
            "withdraw",
        )

    def _change(
        self,
        account_id: str,
        amount: int,
        operation: str,
    ) -> dict:
        with SessionFactory.begin() as session:
            # Lock the account to serialize commands.
            row = session.scalar(
                select(AccountRow)
                .where(AccountRow.account_id == account_id)
                .with_for_update()
            )

            if row is None:
                raise LookupError("Account not found")

            snapshot = load_snapshot(
                session,
                account_id,
            )

            if snapshot is not None:
                if snapshot.version > row.version:
                    raise RuntimeError(
                        "Snapshot is ahead of event stream"
                    )

            start_version = (
                snapshot.version
                if snapshot is not None
                else 0
            )

            # Load only events newer than snapshot.
            events = load_events(
                session,
                account_id,
                after_version=start_version,
            )

            account = replay(events, snapshot)

            if account.version != row.version:
                raise RuntimeError(
                    "Aggregate and projection disagree"
                )

            if operation == "deposit":
                event = account.deposit(amount)
            elif operation == "withdraw":
                event = account.withdraw(amount)
            else:
                raise ValueError("Unknown operation")

            # Accepted event changes domain state.
            account.apply(event)

            session.add(
                EventRow(
                    account_id=account_id,
                    version=event.version,
                    event_type=event.event_type,
                    data=event.data,
                )
            )

            # Update projection atomically with event.
            row.balance_cents = account.balance_cents
            row.version = account.version

            if account.version % SNAPSHOT_INTERVAL == 0:
                state = account.snapshot()

                existing = session.get(
                    SnapshotRow,
                    account_id,
                )

                if existing is None:
                    session.add(
                        SnapshotRow(
                            account_id=account_id,
                            version=state.version,
                            balance_cents=state.balance_cents,
                            is_open=state.is_open,
                            schema_version=state.schema_version,
                        )
                    )
                else:
                    existing.version = state.version
                    existing.balance_cents = (
                        state.balance_cents
                    )
                    existing.is_open = state.is_open
                    existing.schema_version = (
                        state.schema_version
                    )

            result = {
                "account_id": account_id,
                "balance_cents": account.balance_cents,
                "version": account.version,
            }

        return result

    def get_account(self, account_id: str) -> dict:
        with SessionFactory() as session:
            row = session.get(AccountRow, account_id)

            if row is None:
                raise LookupError("Account not found")

            return {
                "account_id": row.account_id,
                "balance_cents": row.balance_cents,
                "version": row.version,
            }

    def rebuild_projection(self, account_id: str) -> dict:
        # Demonstration: one-account transactional
        # rebuild. Not a zero-downtime bulk rebuild.
        with SessionFactory.begin() as session:
            row = session.scalar(
                select(AccountRow)
                .where(AccountRow.account_id == account_id)
                .with_for_update()
            )

            if row is None:
                raise LookupError("Account not found")

            # Ignore snapshots: use authoritative
            # events from the beginning.
            events = load_events(
                session,
                account_id,
                after_version=0,
            )

            reconstructed = replay(events)

            if not reconstructed.is_open:
                raise RuntimeError(
                    "Invalid account event history"
                )

            # Rebuild projection in the same
            # transaction while holding the lock.
            row.balance_cents = (
                reconstructed.balance_cents
            )
            row.version = reconstructed.version

            result = {
                "account_id": account_id,
                "balance_cents": row.balance_cents,
                "version": row.version,
                "events_replayed": len(events),
            }

        return result

At versions divisible by three, it also creates or replaces the aggregate snapshot. This interval is intentionally small so the behavior is easy to observe.

The rebuild method takes a different path. It locks the account row, loads the entire event stream, reconstructs the balance, and overwrites the projection inside one transaction.

The account-level lock also prevents a concurrent command from committing a new event for that account during the rebuild. For very large streams, holding that lock while replaying everything is undesirable. Use a generation-based online rebuild with catch-up and controlled cutover instead.

FastAPI Endpoints

The HTTP adapter exposes account creation, balance changes, reads, and a demonstration maintenance operation for rebuilding one projection.

Create app/api.py:

from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from app.persistence import Base, engine
from app.service import AccountService


class OpenAccountRequest(BaseModel):
    account_id: str = Field(min_length=1)


class AmountRequest(BaseModel):
    amount_cents: int = Field(gt=0)


class AccountResponse(BaseModel):
    account_id: str
    balance_cents: int
    version: int


class RebuildResponse(AccountResponse):
    events_replayed: int


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Local demonstration only.
    # Use Alembic for production migrations.
    Base.metadata.create_all(bind=engine)
    yield
    engine.dispose()


app = FastAPI(
    title="Event Sourcing Snapshots and Projections",
    lifespan=lifespan,
)

service = AccountService()


@app.post(
    "/accounts",
    status_code=201,
    response_model=AccountResponse,
)
def open_account(request: OpenAccountRequest):
    try:
        return service.open_account(
            request.account_id
        )
    except ValueError as exc:
        raise HTTPException(
            status_code=409,
            detail=str(exc),
        ) from exc


@app.post(
    "/accounts/{account_id}/deposits",
    response_model=AccountResponse,
)
def deposit(
    account_id: str,
    request: AmountRequest,
):
    try:
        return service.deposit(
            account_id,
            request.amount_cents,
        )
    except LookupError as exc:
        raise HTTPException(
            status_code=404,
            detail=str(exc),
        ) from exc


@app.post(
    "/accounts/{account_id}/withdrawals",
    response_model=AccountResponse,
)
def withdraw(
    account_id: str,
    request: AmountRequest,
):
    try:
        return service.withdraw(
            account_id,
            request.amount_cents,
        )
    except LookupError as exc:
        raise HTTPException(
            status_code=404,
            detail=str(exc),
        ) from exc
    except ValueError as exc:
        raise HTTPException(
            status_code=409,
            detail=str(exc),
        ) from exc


@app.get(
    "/accounts/{account_id}",
    response_model=AccountResponse,
)
def get_account(account_id: str):
    try:
        return service.get_account(account_id)
        return service.get_account(account_id)
    except LookupError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc


@app.post("/maintenance/accounts/{account_id}/rebuild", response_model=RebuildResponse)
def rebuild_account(account_id: str):
    # Demo only: protect maintenance operations in production.
    try:
        return service.rebuild_projection(account_id)
    except LookupError as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc

The maintenance endpoint is intended for local demonstrations. Production rebuilds require authorization, auditing, operational controls, and background execution for large streams.

Unit Tests for Snapshot Replay

Verify that replaying the full history and restoring from a snapshot produce identical results. Also test invalid schemas, missing versions, and rejected withdrawals.

Create tests/test_domain.py:

import pytest
from app.domain import AccountEvent, AccountSnapshot, replay


def history() -> list[AccountEvent]:
    return [
        AccountEvent(1, "AccountOpened", {}),
        AccountEvent(2, "MoneyDeposited", {"amount_cents": 50000}),
        AccountEvent(3, "MoneyWithdrawn", {"amount_cents": 12500}),
        AccountEvent(4, "MoneyDeposited", {"amount_cents": 2500}),
        AccountEvent(5, "MoneyWithdrawn", {"amount_cents": 10000}),
    ]


def test_complete_replay():
    state = replay(history())
    assert (state.balance_cents, state.version) == (30000, 5)


def test_snapshot_replay_matches_full_history():
    events = history()
    snapshot = replay(events[:3]).snapshot()
    restored = replay(events[3:], snapshot)
    complete = replay(events)
    assert (restored.balance_cents, restored.version) == (
        complete.balance_cents, complete.version
    )


def test_snapshot_without_new_events():
    snapshot = AccountSnapshot(3, 37500, True)
    state = replay([], snapshot)
    assert (state.balance_cents, state.version) == (37500, 3)


def test_unsupported_snapshot_schema():
    with pytest.raises(ValueError):
        replay([], AccountSnapshot(3, 37500, True, schema_version=99))


def test_missing_event_version():
    with pytest.raises(ValueError):
        replay([
            AccountEvent(1, "AccountOpened", {}),
            AccountEvent(3, "MoneyDeposited", {"amount_cents": 1000}),
        ])


def test_withdrawal_rejects_insufficient_balance():
    with pytest.raises(ValueError):
        replay(history()).withdraw(30001)


def test_snapshot_state_is_not_mutated():
    snapshot = AccountSnapshot(3, 37500, True)
    state = replay([
        AccountEvent(4, "MoneyDeposited", {"amount_cents": 500})
    ], snapshot)
    assert state.balance_cents == 38000
    assert snapshot.balance_cents == 37500


def test_invalid_historical_withdrawal():
    with pytest.raises(ValueError):
        replay([
            AccountEvent(1, "AccountOpened", {}),
            AccountEvent(2, "MoneyWithdrawn", {"amount_cents": 100}),
        ])

These unit tests establish snapshot/replay equivalence. Additional integration tests should verify PostgreSQL locks, transaction rollback, snapshot persistence, concurrent commands, and historical event compatibility.

Run and Verify the Example

For local development, start PostgreSQL using this docker-compose.yml:

services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: accounts
      POSTGRES_USER: app
      POSTGRES_PASSWORD: app
    ports:
      - "127.0.0.1:5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U app -d accounts"]
      interval: 5s
      timeout: 3s
      retries: 10

Run the tests, launch the API, then create an account and several events:

# Start the local database and install dependencies.
docker compose up -d --wait
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
pytest -q

# Start API in a separate terminal after exporting DATABASE_URL.
export DATABASE_URL="postgresql+psycopg://app:app@localhost:5432/accounts"
uvicorn app.api:app --reload

# In another terminal, create an account and change its balance.
curl -X POST localhost:8000/accounts -H 'Content-Type: application/json' -d '{"account_id":"account-1001"}'
curl -X POST localhost:8000/accounts/account-1001/deposits -H 'Content-Type: application/json' -d '{"amount_cents":50000}'
curl -X POST localhost:8000/accounts/account-1001/withdrawals -H 'Content-Type: application/json' -d '{"amount_cents":12500}'
curl -X POST localhost:8000/accounts/account-1001/deposits -H 'Content-Type: application/json' -d '{"amount_cents":2500}'

# Inspect the snapshot and current projection.
docker compose exec postgres psql -U app -d accounts -c 'SELECT * FROM account_snapshots;'
docker compose exec postgres psql -U app -d accounts -c 'SELECT * FROM accounts;'

# Rebuild from events, then read the restored projection.
curl -X POST localhost:8000/maintenance/accounts/account-1001/rebuild
curl localhost:8000/accounts/account-1001

# Stop the local database when done.
docker compose down

On a fresh database the snapshot represents version 3 with a balance of 37,500 cents. The projection at version 4 should contain 40,000 cents. Rebuilding replays four events and restores the same state.

Production Trade-offs and Operational Practices

Snapshots exchange storage and writes for faster aggregate reconstruction. Projection rebuilding uses CPU, database I/O, and operational capacity to regenerate derived data. Neither should be enabled without a measurable need and a recovery plan.

Performance and Replay Costs

Monitor aggregate load latency, events replayed per command, snapshot age, projector throughput, checkpoint lag, rebuild duration, lock contention, and consistency failures.

A fast reducer may replay thousands of events cheaply, while a complex aggregate may require snapshots much earlier. Choose intervals based on measurements, not a universal fixed number.

Common Mistakes

Typical failures arise from treating derived data as authoritative or failing to coordinate concurrent updates.

  • Wrong snapshot version: Events are skipped or applied twice.
  • Unsupported snapshot schema: Invalid state enters command decisions.
  • Missing event-gap checks: Incomplete histories pass unnoticed.
  • Replay with side effects: Payments or notifications repeat.
  • Checkpoint before projection commit: Events are lost after crashes.
  • Uncoordinated live rebuild: Newer events disappear at cutover.

Production Checklist

Before deployment, confirm the following controls are present.

  • Authoritative events remain durable, ordered, and readable across schema versions.
  • Snapshots store stream and schema versions; invalid snapshots can be bypassed safely.
  • Projection writes and checkpoints are atomic or idempotently recoverable.
  • Live rebuilds catch up with new writes before switching readers.
  • Reconstruction performs no external side effects.
  • Rollback, replay, and concurrent-write scenarios have integration tests and metrics.

Frequently Asked Questions

Both techniques rely on replay, but snapshots optimize command-side aggregate loading while projection rebuilds regenerate query-side state.

Does a Snapshot Replace Historical Events?

No. Snapshots are derived checkpoints. The event store remains the authoritative source and should support reconstruction after a snapshot fails.

Should You Snapshot After Every Event?

Usually not. Frequent snapshots increase write amplification. Choose a frequency based on measured reconstruction latency and stream size.

Can a Projection Be Rebuilt Without Downtime?

Yes. Build a new projection generation, replay history, catch up with concurrent events, validate, and cut over. This article's small example instead locks one account during an in-place rebuild.

What Happens if a Projection Worker Crashes?

It resumes from a durable checkpoint. Applying events must be idempotent, and a checkpoint should not advance ahead of durable projection changes.

Does Snapshotting Require CQRS?

No. Aggregate snapshots can be used without separate read and write systems. CQRS and snapshotting solve different problems.

What Happens When Event Schemas Change?

Use version-aware readers or upcasters. Incompatible snapshots should be rejected and rebuilt using readable authoritative events.

Conclusion

Snapshots speed up state restoration by skipping historical events already represented by a verified checkpoint. Projection rebuilding restores or changes read models from authoritative history. Correct ordering, versioning, replay safety, and concurrent-write coordination are essential in both cases.

Key takeaway: Keep events authoritative, snapshots disposable, and projection rebuilding a tested operational process.

Comments (0)

Related articles