про Event Sourcing

[Adelf]

@Вурдалак вот тут про ES говоришь, что юзаешь плотно. есть "пара" вопросов. если лень отвечать, кинь ссылки на что-то полезное. Меня тема интересует, но пока в теории.
Какие системы хранения обычно юзают для Event store?
Тоже про read data store. это RDBMS? или некие noSQL? как выглядят системы, которые накатывают события на read data store. эвенты закидываются в какой-нибудь rabbitMQ и по правильному порядку обрабатываются... или как?
Такие системы в итоге всегда Eventual консистентны? как проблемы в ним решают в UI? Т.е. юзер что-то создал, он ведь желает это увидеть сразу.

Да и я нередко теперь практикую ES, а там каждое изменение влечёт за собой иногда по несколько тысяч событий (имеется в виду, при накатывании прошлых), но работает для write model всё равно быстро. То есть, необходимости в lazy loading я не встречал.

Имеются ввиду изменения разных внутренностей агрегата? А то я думал, что обычно сущности меняются мало и это дает довольно производительно делать ES. Просто мне трудно представить перебор тысячи эвентов для создания еще одного я слышал еще, про некие "слепки", когда состояние агрегата кешируется на какой-то момент, чтобы не перебирать эвенты до него.
Как достигается блокировка? Оптимистическая с Event.Version? нормально ли работает? А то вон там у микрософтов пример с продажей билетов и моя первая мысль, что там наверняка возможны конфликты изза билетов на одно и то же событие.
Есть ли проблемы с изменением структуры эвента... когда в базе куча старых? как решаются?

ну и последнее. когда все-таки нужен или полезен ES? часто говорят про перфоманс... это легко поверить если в системе действительно много write операций на разные агрегаты, но это весьма специфично.

 

[Вурдалак]

Какие системы хранения обычно юзают для Event store?

«Обычно» – не знаю, у нас MySQL.

Тоже про read data store. это RDBMS?

У нас это таблички в базе, да.

как выглядят системы, которые накатывают события на read data store. эвенты закидываются в какой-нибудь rabbitMQ и по правильному порядку обрабатываются... или как?

Всё гораздо проще. Прямо в репозитории в конце (после успешного закрытия транзакции для aggregate root) есть такой код:

try {
    $this->fooProjector->project($events, ...);
} catch (\Throwable $e) {
    $this->logger->error(...);
}

Исключение не перебрасывается специально, потому что по факту aggregate root уже успешно сохранился и мы не имеем права падать с ошибкой.

$events тут — это не совсем те $events, что в aggregate root, точнее они в обёртке: AggregateRootEvent($eventId, $createdAt, ..., $event).

В табличку с read model я записываю последний eventId.
В background крутится скрипт, который проверяет расхождение последнего eventId в сущности и в read model.
При необходимости (если есть расхождение), Projector заново пытается «накатить» события в хронологическом порядке, начиная с первого утерянного.
И так до победного конца.
Это и есть eventual consistency.

Такие системы в итоге всегда Eventual консистентны? как проблемы в ним решают в UI? Т.е. юзер что-то создал, он ведь желает это увидеть сразу.

Ну нет, в 99.99% случаев всё будет сразу: ведь я еще в репозитории пытаюсь read model обновить до последней версии.
Но если что-то пошло не так, то read models обновится просто попозже. Но это обычно нештатная ситуация (например, сервер с read models лежал полчаса).

Имеются ввиду изменения разных внутренностей агрегата? А то я думал, что обычно сущности меняются мало и это дает довольно производительно делать ES. Просто мне трудно представить перебор тысячи эвентов для создания еще одного я слышал еще, про некие "слепки", когда состояние агрегата кешируется на какой-то момент, чтобы не перебирать эвенты до него.

Ну ведь по сути это просто тысячи вызовов callback'ов. По современным меркам, это не так долго. И это изменение.
Snapshots же пока не приходилось делать, это обычный кеш, и по возможности лучше обновиться без него.

Как достигается блокировка? Оптимистическая с Event.Version?

Да просто там таблица с событиями вида aggregate_root_id | event_id (event_id у нас последовательный: 1, 2, 3, ...).
И стоит UNIQUE KEY (aggregate_root_id, event_id).
Включается транзакция, делается INSERT IGNORE и если affected rows === 0, то кидаем исключение о конкурентном изменении и пытаемся повторить весь процесс заново (считаваем события, получаем aggregate root, вызываем изменение aggregate root, сохраняем).

нормально ли работает?

Ну там всё так тупо, то хз что там может не работать )

А то вон там у микрософтов пример с продажей билетов и моя первая мысль, что там наверняка возможны конфликты изза билетов на одно и то же событие.
Есть ли проблемы с изменением структуры эвента... когда в базе куча старых? как решаются?

Если ты добавляешь новое обязательное поле в событие, то можно использовать т.н. event upcasting: прямо в рантайме при отсутствии нужных полей еще до десериализации, мы подставляем дефолтные.
Если устарело само событие, что ему добавляем phpdoc @deprecated, всех листеренов подписываем на какое-то новое, но для старого события по-прежнему навечно остаются методы-мутаторы aggregate root для корректного восстановления из событий.

ну и последнее. когда все-таки нужен или полезен ES? часто говорят про перфоманс... это легко поверить если в системе действительно много write операций на разные агрегаты, но это весьма специфично.

Я не понял последнюю мысль, performance тут наоборот будет чувствоваться при большом количестве read'ов. write-то медленные.

Мне он был нужен по принципу «а почему бы и нет».
У нас есть очень важные для сайта aggregate roots, логи изменения которых так или иначе должны выводиться в админке.
А если нужны логи, то почему бы не использовать сразу ES?
Плюс, у нас есть понятие Context-объекта, это типа такая штука, которая собирает по пути через CommandBus/EventBus кучу всякой инфы.
Это позволяет в «логах» (по сути — чтение event storage) видеть помимо событий IP юзера, номер билда, userId, $request_id от nginx, хост машины и т.д.
Я могу глядя в логи практически сразу сказать в чём может быть проблема или найти виновника проблемы.

Единственное, события мы храним на шардах.

 

[Вурдалак]

Хотя если более серьёзно, то плюсы ещё такие:

1. Лог никогда не будет врать.
2. Нет необходимости заниматься мэппингом полей на какую-то таблицу. Поля вообще можно пост-фактум какие угодно добавлять.
3. Собирать статистику ретроспективно по событиям.
4. Повышение устойчивости по аналогии с read models: можно понимать где мы упали и с какого события нужно начинать повтор, если мы будем запоминать id последнего события.

 

[Adelf]

event_id у нас последовательный: 1, 2, 3, ...

event_id к агрегатам хранится в редис каком-нибудь?

Я не понял последнюю мысль, performance тут наоборот будет чувствоваться при большом количестве read'ов. write-то медленные.

с точки зрения read store - ничего ж от ES не поменялось. идут такие же операции write. и такие же read. поэтому я не вижу никаких вещей, которые повысят продуктивность read операций. А про перфоманс write - кто-то что-то когда то мне говорил и мне показалось, что если юзать оптимизированное хранилище для event store - которое знает что не будет никаких update и delete - например как clickhouse, но только с транзакциями, то write часть будет оптимальнее работать.

Единственное, события мы храним на шардах.

не знаю почему это важно, но шардится наверно по aggregate_root_id?

 

[Вурдалак]

event_id к агрегатам хранится в редис каком-нибудь?

Ты считываешь все события AR, просто берешь event id последнего из считанных.

с точки зрения read store - ничего ж от ES не поменялось.

Ну, на практике read-то можно покрыть кешем, а по событиям точечно его инвалидировать.
Можно иметь две таблицы read models (одна чисто для чтения, другая обновляется с каждым событием) и атомарно их менять.
То есть, появляются какие-то возможности.

А про перфоманс write - кто-то что-то когда то мне говорил и мне показалось, что если юзать оптимизированное хранилище для event store - которое знает что не будет никаких update и delete - например как clickhouse, но только с транзакциями, то write часть будет оптимальнее работать.

К сожалению, не могу прокомментировать, не знаю, не слышал.

не знаю почему это важно, но шардится наверно по aggregate_root_id?

По user_id, но это не принципиально, я думаю.

 

[Вурдалак]

Какие системы хранения обычно юзают для Event store?

Просто упомяну на всякий случай проект Greg Young'а (автора понятия ES): https://eventstore.org
Мне просто не хотелось зоопарк разводить, MySQL тоже вполне подходит + у нас совсем немного aggregate roots, которые реально хранятся в виде событий. Большинство — это псевдо-ES, когда сама модель меняется через события, но мы храним состояние (примерно тоже самое, если бы мы делали snapshot после каждого сохранения).

 

[Вурдалак]

А, ну и минусы попробую сформулировать:

1. Данных реально может быть так много, что лучше сразу с шардинга начинать.
2. Всех в команде придётся контролировать и обучать, они сначала ничего не понимают.
3. Нельзя допускать попадания события нового типа в продакшен через какие-нибудь staging-сервера. Продакшен такое событие увидит и упадёт с иключением об неизвестном типа события. Поэтому желательно сначала разложить хотя бы событие + пустой мутатор в AR, а уже потом уже остальное (чем-то очень напоминает процесс добавления новой колонки: сначала добавляешь nullable-поле, только потом раскладываешься и уже потом nullable может убрать).
4. Слишком сложно для бедных на бизнес-логику сущностей.

Зато потом кайф.
К тебе приходит тестер и говорит, что-то не так.
Ты спрашиваешь userId и отвечаешь за минуту-две.
Больше нет такого говна как «блин, он заблокирован, но в логах модератора пусто. Что такое?!». И ты такой смотришь: «ну, заблокирован... в логах пусто... может что-то вручную поменял в базе? или логи модераторов сломали? Или заблокирован системой?». Здесь же в табличке событий есть после context с JSON'ом, который содержит кучу инфы, включая название консольного скрипта, который мог это сделать, заканчивая id модератора, который это сделал через интерфейс).

Не бойтесь, ES, короче. Очень мощная штука для важных для бизнеса aggregate root'в.

 

[Adelf]

3. Нельзя допускать попадания события нового типа в продакшен через какие-нибудь staging-сервера. Продакшен такое событие увидит и упадёт с иключением об неизвестном типа события. Поэтому желательно сначала разложить хотя бы событие + пустой мутатор в AR, а уже потом уже остальное (чем-то очень напоминает процесс добавления новой колонки: сначала добавляешь nullable-поле, только потом раскладываешься и уже потом nullable может убрать).

ну это стандартная проблема со многими вещами, если у тебя не один веб-инстанс. сначала АПИ задеплоить, а потом только юз этого АПИ с фронтэнда. С кешом тоже.

Кстати события разных типов AR хранятся в разных таблицах же? А то я вижу некоторые C# реализации.. у них там Guid и видимо они все в одном хранят. Хотя в том же eventstore.org, есть понятие streamname, которое обычно typeof(AR class).

А вообще, спасибо. Ты мне многое упростил в понимании необязательна какая-то специальная база для event-store. необязательно строить сложные проекции с эвентов на read model(хотя такой вот подход - выглядит канонично). Можно довольно дешево попробовать

 

[Вурдалак]

Кстати события разных типов AR хранятся в разных таблицах же?

В разных, да. Хотя структура одинаковая. Ну, вместо «aggregate_root_id» там что-то более конкретное, конечно.

Но здесь надо признать, что такой подход (с разными таблицами) тоже имеет свой недостаток: труднее вычитывать поток событий без учёта типа. Это нужно для сложных read models, которые зависят от событий разных AR . Почитать все события из одной, а потом из другой — это не то. На время создания тоже не особо завяжешься, только если с микросекундами. Иначе может быть недетерминированная проекция (read model получается разной).

Добавлю, что ты ничего не теряешь от ES-подхода безвозвратно (кроме времени, возможно), потому что откатиться от ES до state persistence можно всегда. Обратно — уже нельзя (точнее, можно, но с потерей всей истории).

Если будешь двигаться в сторону псевдо-ES моделей (внутри всё как в ES, но в базу пишется состояние), и активно заниматься логгированием, тебе понадобятся разные read models, то ты быстро прийдёшь к тому, что ES тебе и нужен.

Я реально даже не изучал особо eventstore, нет необходимости. У нас все прекрасно знакомы с MySQL и некоторыми другими технологиями, поэтому с моей стороны это было бы неразумно. Перейти на eventstore всегда можно, на самом деле. Только нужны весомые плюсы. Но если бы я писал свой уютный проектик для себя, то почему бы и нет.

 

[Вурдалак]

Но здесь надо признать, что такой подход (с разными таблицами) тоже имеет свой недостаток: труднее вычитывать поток событий без учёта типа. Это нужно для сложных read models, которые зависят от событий разных AR

Хотя можно добавить ещё поле global_event_id, тоже sequential и тогда считывать, склеивая таблицы на ходу, будет вполне реально.

У нас просто пока такого не было.

 

[Вурдалак]

А, кстати, мой подходит ещё может смутить тем, что у меня прямо в repository находится projector, ведь их может быть несколько.
Но в большинстве случаев, если все read models для AR находится на одном сервере, то можно иметь всего лишь один projector и в одной транзакции обновлять сразу несколько таблиц c read models.

Можно сделать техническое событие (к domain никакого отношения оно иметь не будет, конечно) типа FooAggregateRooPtroducedEvents(..., $events), на которое будут подписываться разные projector'ы.

Но у нас пока такой необходимости не было.

вот подход - выглядит канонично

У меня агнозия на JS-код, сорри. Врачи говорят, что это неизлечимо.

 

[Adelf]

Это я ссылки перепутал. https://eventstore.org/blog/20130212/projections-1-theory/ тут про функциональный подход с проекциями. Я тоже с яваскриптом не очень, но смысл там не в нем.

Посмотрел сегодня некоторые примеры и опять проснулось страшное желание писать на C#. https://github.com/gregoryyoung/m-r/blob/master/SimpleCQRS/Domain.cs

public abstract class AggregateRoot
    {
        private readonly List<Event> _changes = new List<Event>();

        public abstract Guid Id { get; }
        public int Version { get; internal set; }
...
        public void LoadsFromHistory(IEnumerable<Event> history)
        {
            foreach (var e in history) ApplyChange(e, false);
        }

        // push atomic aggregate changes to local history for further processing (EventStore.SaveEvents)
        private void ApplyChange(Event @event, bool isNew)
        {
            this.AsDynamic().Apply(@event);
            if(isNew) _changes.Add(@event);
        }
    }

    public class InventoryItem : AggregateRoot
    {
        private bool _activated;
        private Guid _id;

        private void Apply(InventoryItemCreated e)
        {
            _id = e.Id;
            _activated = true;
        }

        private void Apply(InventoryItemDeactivated e)
        {
            _activated = false;
        }
...
    }

this.AsDynamic().Apply(@event); - это вот конечно динамика, но уж очень красивая. Оно просит его динамически найти нужный перегруженный метод.

 

[Вурдалак]

Хммм, а что тут необычного-то? Мы так и пишем -> $this->apply(new InventoryItemDeactivated($this->id)).

Правда вместо отдельных методов я пришёл к тому, что лучше иметь один:

private function mutate(Event $event)
{
    triggerForEvent(
        $event, [
            function (InventoryItemDeactivated $event) {
                $this->items[$event->itemId]->deactivated = true;
            },

            function (IntentoryWasFuckedUp $event) {
                $this->items = [];
            }
        ]
    );
}

Там получается компактнее.

 

[Adelf]

ну у нас все равно где-то должен быть map: event type => mutator(кстати где оно у тебя? тоже как-то хитрочерез рефлексию или что?). А там оно прям средствами языка.

 

[Вурдалак]

ну у нас все равно где-то должен быть map: event type => mutator(кстати где оно у тебя? тоже как-то хитрочерез рефлексию или что?). А там оно прям средствами языка.

Через рефлексию можно понять какой аргумент принимает замыкание. Сделать это можно единожды, закешировав map EventClassName => closure. Ну, тут просто как тебе удобнее это будет оформить, я лишь псевдокод привёл.

У нас через рефлексию, короче.

 

[Adelf]

А optimistic locking достигается каким-нибудь уникальным индексом на id агрегата и version? или есть что интереснее?

 

[Вурдалак]

А optimistic locking достигается каким-нибудь уникальным индексом на id агрегата и version?

Да.

 

[Adelf]

Посмотрел пару реализаций ES... Видимо, там нормально можно использовать только примитивы(хотя я сумел сериализовать в эвент VO с публичными полями).
Если в домене активно используются VO - их разворачивать в примитивы для записи в эвент и сворачивать обратно в VO. Это нормально? Или это можно как-то решать.

 

[fixxxer]

А что мешает использовать symfony/serializer? Производительность?

 

[grigori]

не совсем про ES, но о том же https://habr.com/post/413817/

кто-нибудь уже попробовал в работе потоки в redis?