Event Sourcing и CQRS: журнал событий, проекции, снапшоты и когда это не нужно

Коротко

Две идеи часто называют вместе, но они независимы. CQRS без Event Sourcing — обычная практика: пишем в Postgres, читаем из Elasticsearch. Event Sourcing без CQRS возможен, но неудобен: из журнала событий плохо делать запросы вроде «все счета с балансом больше 500».

Проблема: база помнит только последнее состояние

Классическая таблица accounts(id, balance) отвечает на вопрос «сколько денег сейчас». На вопросы «как баланс стал таким», «каким он был 1 марта», «кто и когда изменил лимит» она не отвечает. UPDATE стирает прошлое.

Обычно это лечат таблицей аудита рядом с основной. Но тогда есть два источника правды, и рано или поздно они разойдутся: кто-то поправил баланс скриптом в обход аудита.

Event Sourcing переворачивает схему. Источник правды — журнал фактов, которые уже случились:

acc-1  v1  AccountOpened   {}
acc-1  v2  MoneyDeposited  {"amount": 1000}
acc-1  v3  MoneyWithdrawn  {"amount": 300}

Текущий баланс 700 — производная величина, его можно вычислить в любой момент. Так работает бухгалтерская книга: проводки не исправляют, ошибку исправляют новой сторнирующей проводкой. Так же устроены журнал транзакций в базе данных и история коммитов в git.

Журнал событий

Событие — факт в прошедшем времени: OrderPlaced, MoneyWithdrawn, а не PlaceOrder. Его нельзя отклонить, оно уже произошло. Отклонить можно команду, которая к нему ведёт.

События группируются в потоки (streams) — по одному на агрегат: счёт, заказ, корзину. Внутри потока у событий строгий порядок и номер версии. Хранилищу нужны всего две операции:

• прочитать поток с версии N;
• дописать события в конец потока, если его текущая версия равна ожидаемой.

Второе условие — оптимистичная блокировка, и это главный механизм согласованности. Минимальная реализация:

from dataclasses import dataclass


@dataclass(frozen=True)
class Event:
    stream_id: str
    version: int  # номер события внутри потока
    type: str
    data: dict


class ConcurrencyError(Exception):
    pass


class EventStore:
    def __init__(self):
        self.streams: dict[str, list[Event]] = {}
        self.log: list[Event] = []  # общий порядок для проекций

    def read(self, stream_id, from_version=0):
        return [e for e in self.streams.get(stream_id, []) if e.version > from_version]

    def append(self, stream_id, expected_version, new_events):
        stream = self.streams.setdefault(stream_id, [])
        if len(stream) != expected_version:
            raise ConcurrencyError(
                f'{stream_id}: ожидали версию {expected_version}, а там {len(stream)}'
            )
        for i, (type_, data) in enumerate(new_events, start=expected_version + 1):
            event = Event(stream_id, i, type_, data)
            stream.append(event)
            self.log.append(event)

В реляционной базе это таблица events(stream_id, version, type, data, created_at) с уникальным индексом по (stream_id, version). Два параллельных INSERT с одной версией — и второй упадёт на индексе. Есть и специализированные хранилища, например EventStoreDB, но Postgres для старта хватает.

Агрегат: состояние как свёртка

Агрегат загружается проигрыванием своих событий, проверяет инварианты при выполнении команды и порождает новые события:

class Account:
    def __init__(self, account_id):
        self.id = account_id
        self.version = 0
        self.balance = 0
        self.opened = False
        self.pending = []  # новые события, ещё не сохранённые

    # Состояние = свёртка событий
    def apply(self, type_, data):
        if type_ == 'AccountOpened':
            self.opened = True
        elif type_ == 'MoneyDeposited':
            self.balance += data['amount']
        elif type_ == 'MoneyWithdrawn':
            self.balance -= data['amount']

    @classmethod
    def load(cls, store, account_id):
        acc = cls(account_id)
        for e in store.read(account_id):
            acc.apply(e.type, e.data)
            acc.version = e.version
        return acc

    # Команды проверяют инварианты и порождают события
    def _emit(self, type_, data):
        self.apply(type_, data)
        self.pending.append((type_, data))

    def open(self):
        if self.opened:
            raise ValueError('счёт уже открыт')
        self._emit('AccountOpened', {})

    def deposit(self, amount):
        self._emit('MoneyDeposited', {'amount': amount})

    def withdraw(self, amount):
        if amount > self.balance:
            raise ValueError('недостаточно средств')
        self._emit('MoneyWithdrawn', {'amount': amount})

    def save(self, store):
        store.append(self.id, self.version, self.pending)
        self.version += len(self.pending)
        self.pending = []


store = EventStore()
acc = Account('acc-1')
acc.open()
acc.deposit(1000)
acc.withdraw(300)
acc.save(store)

acc = Account.load(store, 'acc-1')
print(acc.balance, acc.version)  # 700 3

Обратите внимание на разделение: apply только меняет состояние и никогда не бросает исключений — событие уже случилось, спорить с ним поздно. Все проверки — в командах withdraw, open.

Теперь гонка. Два запроса одновременно снимают по 500 со счёта, на котором 700:

a = Account.load(store, 'acc-1')
b = Account.load(store, 'acc-1')
a.withdraw(500)
b.withdraw(500)
a.save(store)
try:
    b.save(store)
except ConcurrencyError as e:
    print(e)  # acc-1: ожидали версию 3, а там 4
print(Account.load(store, 'acc-1').balance)  # 200

Оба прошли проверку баланса в памяти, но сохранился только первый. Второй получает конфликт, перечитывает поток и повторяет команду — теперь withdraw(500) при балансе 200 честно падает с «недостаточно средств». Блокировок на чтение нет, конфликт ловится в момент записи.

Отсюда важное следствие для дизайна: инвариант гарантируется только внутри одного потока. Правило «у пользователя не больше трёх счетов» затрагивает несколько агрегатов, и оптимистичная блокировка одного потока его не защитит. Такие правила решают отдельным агрегатом, уникальным индексом в read-модели или сагой (см. паттерн Saga).

CQRS и проекции

Из журнала легко получить состояние одного счёта, но невозможно эффективно ответить на «топ-10 клиентов по балансу». Для этого строят проекции — read-модели, которые подписаны на поток событий и поддерживают удобное для запроса представление.

Это и есть CQRS (Command Query Responsibility Segregation): команды идут в агрегаты и журнал, запросы — в проекции. Модели можно хранить где угодно: баланс в Redis, поиск в Elasticsearch, отчёты в ClickHouse.

class BalancesProjection:
    """Read-модель: таблица балансов для быстрых запросов."""

    def __init__(self):
        self.balances = {}
        self.position = 0  # сколько событий общего лога уже обработано

    def handle(self, e):
        if e.type == 'AccountOpened':
            self.balances[e.stream_id] = 0
        elif e.type == 'MoneyDeposited':
            self.balances[e.stream_id] += e.data['amount']
        elif e.type == 'MoneyWithdrawn':
            self.balances[e.stream_id] -= e.data['amount']

    def catch_up(self, store):
        for e in store.log[self.position:]:
            self.handle(e)
            self.position += 1


class RichClientsProjection:
    """Другая read-модель из тех же событий: счета с балансом от 500."""

    def __init__(self):
        self.balances = {}
        self.position = 0

    def catch_up(self, store):
        for e in store.log[self.position:]:
            sign = {'MoneyDeposited': 1, 'MoneyWithdrawn': -1}.get(e.type, 0)
            if sign:
                b = self.balances.get(e.stream_id, 0) + sign * e.data['amount']
                self.balances[e.stream_id] = b
            self.position += 1

    def rich(self):
        return sorted(k for k, v in self.balances.items() if v >= 500)


acc2 = Account('acc-2')
acc2.open()
acc2.deposit(900)
acc2.save(store)

balances = BalancesProjection()
rich = RichClientsProjection()
balances.catch_up(store)
rich.catch_up(store)
print(balances.balances, rich.rich())  # {'acc-1': 200, 'acc-2': 900} ['acc-2']

Что здесь важно:

• Проекций сколько угодно, и их можно добавлять задним числом. Понадобился новый отчёт через год — пишем проекцию и прогоняем через неё весь журнал с нуля. В обычной базе данных за прошлое для такого отчёта просто нет.
• Проекция одноразовая. Нашли баг в обработчике — удаляем read-модель и перестраиваем. Источник правды не пострадал.
• Позиция хранится атомарно с данными. Если проекция обновила таблицу, а позицию сохранить не успела, после рестарта она применит событие повторно. Решения: писать позицию в той же транзакции, что и read-модель, или делать обработчики идемпотентными по (stream_id, version). Подробнее о повторах — в статье про идемпотентность.

Eventual consistency

Проекции обновляются асинхронно, поэтому read-модель отстаёт от журнала:

acc2 = Account.load(store, 'acc-2')
acc2.withdraw(800)
acc2.save(store)
print(balances.balances['acc-2'])  # 900 — проекция ещё не догнала
balances.catch_up(store)
print(balances.balances['acc-2'])  # 100

Между записью и обновлением проекции пользователь может увидеть старые данные: нажал «перевести», открыл историю — перевода нет. В реальной системе задержка — миллисекунды или секунды, но при отставании консьюмера бывает и минуты.

Как с этим жить:

• Read-your-writes. Команда возвращает новую версию потока, клиент передаёт её в запросе, и read-модель ждёт, пока догонит эту версию, или честно отвечает «данные обновляются».
• Оптимистичный UI. Клиент сразу показывает результат своей команды, не дожидаясь проекции.
• Читать агрегат напрямую. Для экрана «мой счёт» можно загрузить один поток из журнала — это строго согласованно и быстро, если поток короткий.
• Мониторить отставание. Разница между последней позицией журнала и позицией проекции — ключевая метрика, на неё ставят алерт.

Это та же развилка, что и в теореме CAP: мы платим согласованностью чтения за независимость и масштабируемость моделей.

Снапшоты

Если в потоке миллион событий, загружать агрегат проигрыванием дорого. Снапшот — сохранённое состояние на версии N. Загрузка = снапшот + хвост событий после N:

snapshots = {}  # stream_id -> (version, state)


def save_snapshot(acc):
    snapshots[acc.id] = (acc.version, {'balance': acc.balance, 'opened': acc.opened})


def load_with_snapshot(store, account_id):
    acc = Account(account_id)
    if account_id in snapshots:
        acc.version, state = snapshots[account_id]
        acc.balance, acc.opened = state['balance'], state['opened']
    tail = store.read(account_id, from_version=acc.version)
    for e in tail:
        acc.apply(e.type, e.data)
        acc.version = e.version
    return acc, len(tail)


acc = Account.load(store, 'acc-1')
for _ in range(1000):
    acc.deposit(1)
acc.save(store)
save_snapshot(acc)
acc.deposit(50)
acc.save(store)

acc, replayed = load_with_snapshot(store, 'acc-1')
print(acc.balance, acc.version, replayed)  # 1250 1005 1

Вместо 1005 событий проиграно одно. Нюансы:

• Снапшот — кэш, а не источник правды. Его можно удалить и пересчитать. Поэтому снапшоты делают асинхронно, например каждые 100–500 событий, и не в транзакции с командой.
• Схема снапшота версионируется. Поменяли структуру агрегата — старые снапшоты выбрасываем и строим заново из событий.
• Частая потребность в снапшотах — сигнал. Если поток растёт бесконечно, возможно, агрегат выбран неудачно. Счёт с ежедневными операциями лучше разбить на периоды: «закрытие месяца» как событие, новый поток со стартовым балансом.

Версионирование событий

События неизменяемы и хранятся годами, а код меняется. Через год MoneyDeposited понадобилось поле currency, которого нет в старых событиях. Варианты:

• Апкастинг. При чтении старое событие преобразуется в новую версию: нет currency — подставляем RUB. Журнал не меняется.
• Новый тип события. MoneyDepositedV2 рядом со старым, агрегат умеет применять оба.
• Только добавлять поля. Никогда не переименовывать и не менять смысл существующих.

Переписывать журнал задним числом можно, но это крайняя мера — операция над всей историей, легко ошибиться.

Отдельная боль — удаление персональных данных по требованию закона, например GDPR. Неизменяемый журнал с этим конфликтует. Типовое решение — crypto-shredding: персональные данные в событиях шифруются ключом пользователя, при запросе на удаление ключ уничтожается.

Когда Event Sourcing не нужен

Честный ответ на собеседовании ценится выше энтузиазма. Event Sourcing — дорогой паттерн:

• Простой CRUD. Справочник товаров, профиль пользователя, настройки. Истории изменений никто не попросит, а сложность вырастет в разы.
• Нужна строгая согласованность чтения везде. Если каждый экран должен показывать данные сразу после записи, проекции будут мешать.
• Команда без опыта. Ошибки в границах агрегатов и версионировании событий исправлять дороже, чем ошибки в схеме таблиц.
• Аудит можно получить дешевле. CDC из журнала базы, триггеры или таблица истории решают задачу «кто что менял» без смены модели данных.

Где он оправдан:

• Финансы, бухгалтерия, биллинг — история операций и есть предметная область.
• Системы, где важно «как мы сюда пришли»: логистика, статусы заказов, страхование, медицинские карты.
• Много разных read-моделей из одних данных и необходимость добавлять новые задним числом.
• Интеграция через события — сервисы и так обмениваются фактами, журнал становится естественным источником для очереди сообщений.

Важно не путать журнал событий агрегата с Kafka-топиком. Kafka хороша для доставки событий проекциям и другим сервисам, но как основное хранилище агрегатов неудобна: нет эффективного чтения одного потока и нет append с проверкой ожидаемой версии.

Вопросы на собеседовании

• Чем Event Sourcing отличается от лога аудита? В ES журнал — единственный источник правды, состояние из него выводится. Аудит — копия рядом с состоянием, и он может разойтись с ним.
• Как обеспечить, что два запроса не испортят агрегат? Оптимистичная блокировка: append с ожидаемой версией, уникальный индекс (stream_id, version), при конфликте — перечитать и повторить команду.
• Как атомарно сохранить событие и опубликовать его в брокер? Журнал событий сам работает как outbox: релей читает новые события и публикует. Двойной записи нет.
• Что если проекция сломалась? Исправить код, удалить read-модель, перестроить из журнала. Пока перестраивается — либо держать старую версию, либо строить новую рядом и переключить трафик.
• Как долго хранить события? Обычно всегда — в этом смысл. Холодные потоки переносят в дешёвое хранилище, закрытые агрегаты архивируют.
• CQRS обязательно с двумя базами? Нет. Начать можно с двух моделей в одной базе: таблица событий и таблица-проекция, обновляемая в той же транзакции. Тогда чтение даже строго согласованно. Разносить по разным хранилищам стоит, когда нагрузки на чтение и запись реально различаются, — подробнее про масштабирование чтения в статье про шардирование и репликацию.
• Как удалить персональные данные? Crypto-shredding или хранение PII вне событий со ссылкой по идентификатору.

Частые ошибки

• События-команды. UpdateAccount {balance: 700} — это не событие, а снимок состояния. Теряется смысл: был это депозит, исправление ошибки или начисление процентов.
• Проверки в apply. Если apply бросает исключение, старые события перестают проигрываться после изменения правил.
• Внешние вызовы при проигрывании. Обработчик агрегата отправил письмо — и отправит его снова при каждой загрузке. Побочные эффекты — только в отдельных подписчиках.
• Слишком крупный агрегат. Один поток на всего пользователя со всеми заказами — постоянные конфликты версий и бесконечный поток.
• Чтение из проекции для проверки инварианта. Проекция отстаёт, проверка пропустит конфликт. Инварианты проверяет агрегат по своему потоку.
• Event Sourcing «потому что модно». Для системы из десяти CRUD-экранов это гарантированный перерасход времени.