Модуль 12 ⏱ ~34 мин чтения 📐 1 Mermaid диаграмм

9. Event sourcing и базы

Введение: Состояние как последовательность фактов

Представьте гроссбух классического венецианского купца эпохи Возрождения. Главный бухгалтер никогда не берет нож, чтобы соскоблить чернила со страницы, и не заклеивает ошибочные записи. Если при расчете поставки шелка допущена ошибка на десять дукатов, бухгалтер не переписывает баланс — он аккуратно заносит на следующей строчке корректирующую проводку: «Списано 10 дукатов ввиду ошибки пересчета». Текущее финансовое состояние торгового дома — это не число, выбитое на мраморной плите, а сумма всех записей в книге от первой страницы до последней. Вы всегда можете перелистнуть страницы назад и увидеть, как выглядел баланс купца ровно три года назад во вторник в полдень.

Event Sourcing (ES) — это архитектурный паттерн хранения данных, при котором состояние приложения представляется не как текущий снимок (как в классических CRUD-системах), а как неизменяемая последовательность событий (фактов), которые привели систему в это состояние. Вместо UPDATE users SET balance = 100 вы записываете BalanceDeposited { amount: 50, transaction_id: "abc" }.

Для инженера уровня Senior/Lead ES — это не просто модная альтернатива таблицам. Это строгий контракт на неизменяемость, полный аудиторский трейл из коробки, возможность воспроизвести состояние на любой момент времени и естественная интеграция с асинхронной архитектурой. Однако паттерн привносит фундаментальные изменения в работу с хранилищем, требования к памяти и стратегии миграции.

В этой статье мы разберем:

  • Архитектуру Event Store: паттерн Append-Only, последовательные журналы и отличие от WAL.
  • Механику ввода-вывода под капотом: как ОС и диск обрабатывают последовательную запись, роль fsync, O_APPEND и page cache.
  • Идиоматичную реализацию хранилища в Go: загрузка состояния, применение снапшотов, обработка дубликатов.
  • Эволюцию схемы событий: upcasters, версионирование и толерантное чтение.
  • Влияние на рантайм Go: аллокации при реплее миллионов событий, давление на GC и оптимизации через sync.Pool.
  • Типичные ловушки, вопросы с хардовых собеседований и сравнение с другими экосистемами.

1. Архитектура хранилища событий и паттерн Append-Only

Классическая СУБД оптимизирована под актуальное состояние. Event Store оптимизирован под хронологию изменений.

flowchart TD
    classDef process  fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success  fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error    fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning  fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry    fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    classDef aux      fill:#334155,stroke:#64748b,stroke-width:1px,color:#f8fafc
    classDef system   fill:#312e81,stroke:#a78bfa,stroke-width:1px,color:#f8fafc
    classDef data     fill:#132d29,stroke:#2dd4bf,stroke-width:1px,color:#f8fafc
    classDef network  fill:#881337,stroke:#fb7185,stroke-width:1px,color:#f8fafc
    classDef accent   fill:#1e1b4b,stroke:#818cf8,stroke-width:1px,color:#f8fafc
    Cmd["Команда: Создать перевод"]:::process --> Val["Валидация бизнес правил"]
    Val --> Ev["Генерация события: MoneyTransferred"]
    Ev --> Store["Event Store<br>Последовательный журнал"]
    Store -->|"1 - Append + fsync"| Disk["Блочное устройство / SSD"]
    Store -->|"2 - Publish"| Proj["Projection Service"]
    Proj -->|"Применение к модели"| State["Текущее состояние агрегата"]
    State -->|"Ответ клиенту"| Cmd

Основные компоненты

  1. Event Stream: Логический контейнер событий для одного агрегата (например, order-123). Имеет монотонно возрастающую версию (позицию).
  2. Global Sequence: Глобальный счетчик всех событий в системе. Используется для шардирования, репликации и проекций.
  3. Snapshots: Периодические снимки состояния агрегата, позволяющие не реплеить историю с самого начала.

2. Под капотом: Механика последовательной записи и IO

Почему Event Sourcing часто быстрее классических INSERT/UPDATE на высоких нагрузках? Потому что он эксплуатирует физику дисков и ОС.

Последовательная запись и Page Cache

При вызове INSERT в B-Tree СУБД выполняет:

  • Поиск страницы в памяти (Buffer Pool).
  • Модификацию и пометку страницы как dirty.
  • При необходимости: чтение других страниц с диска для балансировки дерева (сложность O(log N)).

При Append-Only записи:

  • Данные добавляются в конец файла или журнала.
  • ОС (через Page Cache в Linux) буферизует запись последовательно.
  • fsync сбрасывает данные на диск одним непрерывным блоком.
  • Сложность записи O(1), независимо от размера базы.

Файловые дескрипторы и O_APPEND

В Go при открытии файла для логов событий используется флаг O_APPEND. Это атомарная операция на уровне ядра: каждое write() автоматически позиционирует указатель в конец файла перед записью. Это гарантирует, что даже при параллельных горутинах, пишущих в один файл (что не рекомендуется для высоконагруженных систем), данные не перетрутся. Однако для продакшена предпочтительнее использовать специализированные Event Store СУБД или реляционные БД с INSERT ... RETURNING position, так как файловые журналы требуют ручного управления ротацией, блокировками и репликацией.

3. Идиоматичная реализация в Go

Реализация Event Store в Go требует аккуратной работы с контекстами, транзакциями и конкурентностью. Ниже показан паттерн загрузки состояния агрегата с учетом снапшотов и идемпотентности.

Go 
package eventstore

import (
    "context"
    "database/sql"
    "encoding/json"
    "errors"
    "fmt"
    "time"
)

// AggregateState представляет восстановленное состояние агрегата
type AggregateState struct {
    ID      string
    Balance int64
}

// StoredEvent представляет запись в хранилище
type StoredEvent struct {
    ID        string
    StreamID  string
    Version   int64
    Type      string
    Payload   []byte
    CreatedAt time.Time
}

// LoadAggregate загружает состояние, начиная со снапшота или начала истории
func LoadAggregate(ctx context.Context, db *sql.DB, streamID string, snapshotLoader func(context.Context, string) (*AggregateState, int64, error), applier func(*AggregateState, StoredEvent) error) (*AggregateState, error) {
    // 1. Проверяем снапшот
    state, baseVersion, err := snapshotLoader(ctx, streamID)
    if err != nil && !errors.Is(err, sql.ErrNoRows) {
        return nil, fmt.Errorf("load snapshot: %w", err)
    }
    if state == nil {
        state = &AggregateState{} // Начальное состояние
        baseVersion = 0
    }

    // 2. Загружаем события после снапшота
    query := `
        SELECT id, stream_id, version, type, payload, created_at 
        FROM events 
        WHERE stream_id = $1 AND version > $2 
        ORDER BY version ASC
    `
    rows, err := db.QueryContext(ctx, query, streamID, baseVersion)
    if err != nil {
        return nil, fmt.Errorf("query events: %w", err)
    }
    defer rows.Close()

    for rows.Next() {
        var ev StoredEvent
        if err := rows.Scan(&ev.ID, &ev.StreamID, &ev.Version, &ev.Type, &ev.Payload, &ev.CreatedAt); err != nil {
            return nil, fmt.Errorf("scan event: %w", err)
        }
        if err := applier(state, ev); err != nil {
            return nil, fmt.Errorf("apply event v%d: %w", ev.Version, err)
        }
    }
    if err := rows.Err(); err != nil {
        return nil, fmt.Errorf("iterate rows: %w", err)
    }
    return state, nil
}

Сохранение с оптимистичной блокировкой (Version Check)

Конкурентность в ES решается через проверку версии потока. Это аналог SELECT ... FOR UPDATE, но быстрее и без блокировок на чтение.

Go 
func SaveEvents(ctx context.Context, db *sql.DB, streamID string, expectedVersion int64, events []StoredEvent) error {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // Проверка версии одной транзакцией
    var currentVersion int64
    err = tx.QueryRowContext(ctx, 
        "SELECT COALESCE(MAX(version), 0) FROM events WHERE stream_id = $1 FOR UPDATE", 
        streamID,
    ).Scan(&currentVersion)
    if err != nil {
        return err
    }

    if currentVersion != expectedVersion {
        return errors.New("concurrency conflict: stream version mismatch")
    }

    stmt, _ := tx.PrepareContext(ctx, `
        INSERT INTO events (id, stream_id, version, type, payload, created_at)
        VALUES ($1, $2, $3, $4, $5, $6)
    `)
    defer stmt.Close()

    for i, ev := range events {
        ev.Version = expectedVersion + int64(i) + 1
        if _, err := stmt.ExecContext(ctx, ev.ID, streamID, ev.Version, ev.Type, ev.Payload, time.Now()); err != nil {
            return err
        }
    }
    return tx.Commit()
}

4. Эволюция схемы событий и Upcasters

События неизменяемы, но бизнес-логика меняется. Как добавить поле currency в событие 3-летней давности? Никогда не меняйте оригинал. Используйте Upcasters.

Механика Upcasting

При чтении событий из хранилища применяется слой трансформации, который конвертирует старые версии структуры в текущую схему агрегата.

Go 
type OrderCreatedEvent struct {
    ID       string `json:"id"`
    Amount   int64  `json:"amount"`
    Currency string `json:"currency"`
}

func UpcastEvent(ev StoredEvent) (interface{}, error) {
    switch ev.Type {
    case "OrderCreated":
        // v1 не имел поля Currency
        if !strings.Contains(string(ev.Payload), "currency") {
            // Встраиваем дефолтное значение в JSON до десериализации
            payload := strings.Replace(string(ev.Payload), "}", `,"currency":"USD"}`, 1)
            var target OrderCreatedEvent
            err := json.Unmarshal([]byte(payload), &target)
            return target, err
        }
        var target OrderCreatedEvent
        err := json.Unmarshal(ev.Payload, &target)
        return target, err
    default:
        return nil, fmt.Errorf("unknown event type: %s", ev.Type)
    }
}

5. Производительность, память и GC при реплее

Восстановление состояния (replay) — самая ресурсоемкая операция в ES. При загрузке агрегата с 10 000 событий вы десериализуете 10 000 структур, применяете их к объекту и создаете тысячи временных аллокаций.

Давление на Garbage Collector

Каждое json.Unmarshal или proto.Unmarshal выделяет память в куче. При холодном старте сервиса, который реплеит сотни агрегатов, GC может потреблять до 20-30% CPU времени.

Оптимизации в Go:

  1. sync.Pool для буферов: Переиспользуйте []byte буферы для чтения из rows.Scan.
  2. Компактная сериализация: Используйте Protobuf или Avro вместо JSON. Они создают меньше аллокаций и имеют предсказуемый размер.
  3. Пакетное чтение: Вместо SELECT ... ORDER BY version для каждого агрегата, используйте курсоры или pgx.CopyFrom-подобные механизмы для пакетной загрузки событий нескольких агрегатов, затем группируйте их в памяти.
  4. Снапшоты как барьер для GC: Настраивайте создание снапшотов каждые N событий или T времени. Это сокращает длину цепочки реплея и, соответственно, количество объектов, живущих в памяти одновременно.

6. Ловушки, антипаттерны и вопросы с собеседований

  1. Отсутствие стратегии снапшотов

    • Проблема: Агрегат растет годами. Время загрузки с 10 мс доходит до 2 секунд.
    • Решение: Автоматический триггер снапшота при превышении лимита событий. Снапшот должен храниться в том же хранилище с привязкой к версии потока.

  2. Попытка использовать ES для аналитики

    • Проблема: Запросы COUNT, SUM, GROUP BY по событиям выполняются через полный скан лога.
    • Решение: ES — источник истины для состояния, не для отчетов. Всегда проецируйте события в оптимизированные Read Model (ClickHouse, Elasticsearch, материализованные виды).

  3. Поздние события (Late Events)

    • Проблема: Событие с таймстампом вчера приходит сегодня из-за сетевых задержек. Проекция применяет его в конец, нарушая хронологию.
    • Решение: Проекции должны быть идемпотентными и поддерживать версионирование по времени. При получении позднего события пересчитывайте сегмент состояния или используйте компенсирующие транзакции.

  4. Сравнение с C#/PHP

    • C# (.NET): Фреймворки вроде EventStore.Client или Akka.NET берут на себя управление соединениями, снапшотами и проекциями. Много "магии", но быстрая разработка.
    • PHP: Stateless модель запросов делает ES крайне неудобным. Каждое восстановление состояния требует полной загрузки из БД, что убивает производительность. В PHP ES используют редко, преимущественно через внешние шины (Kafka) и легковесные проекторы.
    • Go: Требует явной реализации реплея, снапшотов и конкурентности. Это больше кода, но дает полный контроль над памятью, пулами соединений и поведением при сбоях. Идеально для высоконагруженных микросервисов.

7. Сравнение подходов и выбор стека

Характеристика Реляционная БД (PG/MySQL) EventStoreDB / Kafka Файловые журналы
Производительность записи Высокая (с батчингом) Очень высокая (аппенд-лог) Максимальная (прямой диск)
Запросы по событиям Полнотекстовые, сложные Ограниченные, по потоку Только последовательный скан
Транзакционность ACID, надежная Eventual или строгая (зависит от конфига) Отсутствует (требует ручной реализации)
Интеграция с Go database/sql, pgx Клиентские SDK, gRPC os.OpenFile, io.Reader
Рекомендация Стартапы, малые/средние нагрузки, простота поддержки Высоконагруженные системы, строгий аудит, микросервисы Встраиваемые системы, офлайн-режим, максимальный контроль

Итог

Event Sourcing — это мощный архитектурный выбор, который превращает данные из пассивного хранилища состояния в активный журнал бизнес-фактов. В экосистеме Go он реализуется через явные паттерны загрузки, оптимистичную блокировку версий, снапшоты и идемпотентные проекции. Ключевые принципы для уровня Senior/Lead:

  • События неизменяемы. Ошибки исправляются новыми событиями, а не правкой истории.
  • Используйте оптимистичную блокировку (version check) для гарантии порядка.
  • Управляйте памятью и GC через снапшоты, компактную сериализацию и переиспользование буферов.
  • Проецируйте состояние в специализированные Read Model, никогда не используйте Event Store для аналитики.
  • Проектируйте upcasters для эволюции схем без нарушения исторических данных.

Освоив Event Sourcing, вы получаете систему с полным аудитом, возможностью отката во времени и естественной асинхронностью. Но когда данные распределены между несколькими микросервисами и каждый из них поддерживает свою локальную консистентность, как гарантировать, что бизнес-процесс в целом не нарушит инварианты? В следующей статье мы разберем стратегии обеспечения согласованности данных в распределенных системах: 10. Data consistency в микросервисах.