Комментарии 25
интересно, а как защититься от ситуации когда в процессе работы
BuyItemsSagaпроизойдет, к примеру, нехватка памяти и процесс упадет?
кто обеспечит возврат денег или предметов?
Она упадет, потом сделает ретрай. Там пока в БД не будет записано что обработчик выполнился и выполнено действие оно будет повторяться. Либо до достижения лимита ретраев.
Тут надо в ней поля добавить для количества предметов и суммы денег что она себе забрала.
Дальше записываем данные в эти поля и потом уже делаем запрос на возвращение денег и предметов обратно. Да, консистентность со временем.
Для важных процессов ретраев делается много и с большим нарастающим промежутком между ними. Кроме того, в логи пишется если все ретраи выполнены и процесс не удалось завершить (Fault) и уже такие случаи можно индивидуально рассмотреть. Вообще через Jaeger это все трейситься обычно.
Тут для простоты фронт ждет ответа. Обычно записывают в БД Id таска при постановке задачи и дальше при событии оповещающем о том что сага завершила свою работу выставляют данные этом Таске в БД и еще в SignalR пушим что таск завершился. Фронт может и сам тоже проверить сделав запрос завершился ли Таск и какой у него результат.
public class DelayedTask
{
public Guid Id { get; set; }
public bool IsCompleted { get; set; }
public string ResultJson { get; set; }
}И в MassTransit и в NServiceBus в реализации Saga есть перзистентность и возможность повторов выполнения т.е. они умеют после падения востанавливаться и продолжать свою работу там где они остановились.
А как масштабируется эта персистентность?
Как она реализуется?
Сколько IOPS нужно на каждый шаг саги?
Можно ли организовать компенсацию не отдельного шага, а всего сценария?
И чем это лучше того же temporal.io?
1, 2, 3 - Зависит от реализации. MassTransit старается от этого абстрагироваться.
4) Суть саги в том что это набор мелких транзакций. Так или иначе компенсаций это тоже будет несколько мелких транзакций. Можно в том числе в стаье от Avito про это почитать.
5) Чем HttpClient лучше temporal.io? Как Camunda сделана с помошью Zeebe так и вы можете сделать свой temporal.io с помошью MassTransit и еще кучи других библиотек.
1) А какие существуют реализации, как они обеспечивают гарантии в распределенной среде? Это, собственно, самое интересное же в движках саг.
4) Обычно в реальной жизни не нужен поток компенсаций отдельных шагов, нужна компенсация всей бизнес-транзакции (паттерн "Сага" сам по себе не очень пригоден для бизнес-транзакций, да и создавался для совсем других целей и технологий). Поэтому в большинстве движков оркестрации есть возможность компенсации всей транзакции, а не набора шагов
5) Пока не понятно, как поможет MassTransit для решения задачи temporal.io (который вообще не про саги в классическом виде). Проще с нуля написать (ну, или взять уже написанный).
Ну, там нет ничего про собственно гарантии. Сам MassTransit берет на себя только rerty, а вот все прочее (в том числе гарантии транзакционности, гарантии сохранения данных, реализацию нормального outbox и тому подобное) - уже приходится делать руками.
А в .Net есть решения для саг не настолько низкого уровня?
Фактически, саги в MT - не столько саги, сколько просто акторные движки с возможностью подключить персистанс, но не гарантирующие при этом ничего сверху. На персистентных акторах можно делать распределенные бизнес-транзакции, я даже на HL про это рассказывал когда-то, но это не самый удобный и не самый современный подход. Да и в реализации от MT слишком про много вещей придется думать, если нужны хоть какие-то гарантии.
Вообще есть и корреляция и комутативность и идемпотентность. Ну и конечно сохранение данных. Стей саги в БД храниться обычно. Хотя есть возможность просто InMemory ну это больше для тестов. Транзакционность между своими состояними он тоже обеспеспечивает если вы Postgres в качестве хранилища используете. Транзакционность всей оперции - ну это такое. Суть паттерна Saga в том что она это обеспечивает набором компенсирующих действий. OutBox в MT вообще есть. В примере кода есть строчка где он подключается. У меня уже складывается ощуние что статью вы не читали. Это вообще учитывая что она вводная и послее нее хорошо бы пойти почитать документацию к MT.
А этот доклад доступен на ютубе?
Тут в принцыпе "пример ради примера" и не стал расписывать компенсирующие действия потому что стала бы сага тогда совсем огромной. Можно повесить Таймаут на сагу и если через 30 минут например не удалось провести операцию то начать пытаться до посинения откатить изменения в системе. Вернуть денги и предметы обрабтно или только деньги. Таймауты тоже настраиваются как перзистентные и тоже настраиваются с ретраями
Если простое действие требует кучу кода - то может быть нужно искать другие решения?
Простые вещи должны быть простыми, а бизнес-операция из двух шагов - простая штука.
Безнес операция в распределенной системе это совсем другое. Вообще про распределенные системы для начала почитайте прежде чем такие заявления делать. В целом MassTransit как библиотека для распределенных систем и так довольно жирный слой абстракции дает и многое на себя берет позволяя на его DSL писать более менее простой воркфлоу. Мне интересно как вы себе преставляет простое рещение когда у вас надо изменить данные в БД микросервиса А и микросервиса Б согласованно чтобы соблюдался хотяб ACD из ACID? Притом что один из них или оба вообще не принадлежат ващей компании. Сервисы Google и Facebook например. Пример кода привидите.
Если устраивает eventual consistency, то в том же temporal.io это пишется примерно в три строчки, остальная магия как раз внутри движка (сохранение стейта после каждого шага, ретраи, восстановление после падения сервисов, запуск компенсации с теми же гарантиями и так далее). Примеры кода - да на том же https://docs.temporal.io/docs/temporal-explained/introduction
Собственно, для этого temporal.io и делался, чтобы простые вещи можно было писать достаточно компактно (впрочем, там тоже хватает своих проблем, разумеется).
(У меня сейчас свой движок для распределенных бизнес-транзакций, там все еще компактнее получается, но это не open source)
Хм, точно так же пишеться все как и в MassTransit судя по примерам что я посмотрел.
Судя по ващим коментариям вы вообще не понимаете о чем идет речь. Так даже и нет примера для случая когда у вас есть внешнее API от Google и API от Facebook и вам нужно сделать запрос в одно потом в другое и если провалилось первое то сделать компенсирующее запросы для второго. В реальных приложениях не все сервисы к которым вы делаете запросы вы контролируете. В какую-то пратежную систему, сделать запрос, потом в API службы доставкий какой-то. Да, можно используя MassTransit сделать еще слой асбстракции и выставить наружу API как будто одним действием это делается однако под капотом оно не так.
Э, какая разница, в контролируемые или нет сервисы делаются запросы? Или сервисы поддерживают компенсацию по шагам и тогда сага еще применима. Или, что чаще, не поддерживают и тогда нужно говорить о компенсации всей бизнес-транзакции целиком. И как раз такие сценарии на temporal реализуются легко (с разными логиками повторов, ожиданий, обработки ошибок и тому подобное), в рамках линейного кода и с понятными гарантиями выполнения сценария даже при падении выполняющего сервера.
В MassTransit, судя по примерам, требуется гораздо больше кода, да и с гарантиями все не очень понятно (ну, из статьи).
И, думаю, я все-таки хорошо понимаю, о чем говорю, тех же самых платежных систем сделал достаточно.
Вообще, конечно, если делать нормальную библиотеку работы с бизнес-транзакциями, то вызов двух API в ней должен выглядеть примерно так (тут вызов трех APIшных вызовов в двух разных сервисах и компенсация - и это все в 10 строчках и это реальный код)
suspend fun onboard(data: OnboardingData) : ClientId = play {
val identityId = step { identityService.nextId () }
step { identityService.create (identityId, data.credentials) }
val clientId = step { clientService.create (identityId, data.clientInfo) }
return@play clientId
}.onError {
step { if (clientId!=null) clientService.remove(clientId) }
step { if (identityId!=null) identityService.remove(identityId) }
}.async()
Отличный пример - тут так же в MassTransit будет за исключением что не будет одного onError а будет два шага отката. Потому что две транзакции это будет. Первая откатит первый запрос, а вторая второй. Тут под капотом чтобы реализовать такое все равно надо делать чезе две транзации и нужная более низкоуровневая библиотека вроде MassTransit для этого. В первой транзакции step { if (clientId!=null) clientService.remove(clientId) } выполняем и сохраняем в БД что эту операцию откатили. Во второй транзакции сохраняем step { if (identityId!=null) identityService.remove(identityId) }. Если упали то MassTransit перезапуститься и посмотрит из БД что первый шаг он уже откатил и сразу пойдет откатывать второй шаг.
1) Вот как раз откат по одной транзакции - в среднем плохая идея, так как чаще всего нет возможности откатить конкретный шаг (нельзя отменить отправку sms-ки или однофазовый платеж, например).
2) После падения откатывать - тоже плохая идея, нужно стараться дожимать сценарий (т.е. продолжать его с того шага, на котором упали). Это гораздо лучше для пользователя.
3) Не понятно, зачем там нужна низкоуровневая библиотека, что она дает? При этом у меня 10 строк на все, а в примере на MassTransit кода в разы больше - и это без компенсаций вообще.
Собственно, подоход temporal (или как в приведенном выше куске), с повтором бизнес-запросов, а не с откатом на каждый чих - гораздо удобнее и полезнее в реальной жизни. Ну и кода получается в разы меньше (хотя это уже зависит от конкретной реализации и MassTransit явно не очень удачная, да и кролик для описанной задачи только лишний)
1) Можно послать компенсирующу SMS с информацией что предыдущая была ошибочной.
2) Там любые действия идут с ретраями. В том числе и выполение позитивного действия. MS Saga это стейт машина. Если вы перешли в стейт из которого дальше нужно откатывать то она будет после востановления пытаться откатывать. Если в стейте на котором нужно выполнить действие то после востановления будет пытаться выполнить действие. В общем стейт машина будет пытаться продолжать с того момента на котором остановилась.
3) Вообще есть в OutBox в MassTransit. Ваш пример вообще не понятно как работает в реальном приложении. Где там идемпотентность? Где коммутативность? Где корреляция по одной бизнес операции? Что произойдет если в
}.onError { step { if (clientId!=null) clientService.remove(clientId) } step { if (identityId!=null) identityService.remove(identityId) }}.async()
на втором шаге произойдет падение? что произойдет если не remove будет у вас а какой нибудь запрос например на увеличение счетчика т. е. не идемпотентный? Если запрос бросит ошибку из-за прервывания сети на моменте ответа уже увеличив какой-то счетчик? Я вижу какой-то DSL. Возможно за ним какая-то магия, только непонятно как она работает. Почему вы не отделяете ошибочные ответы просто от таймаутов? Вообще можно в MT 1 в 1 сделать такой же пример как у вас. Будет только + код для обьявления контрактов и стейтов дополнительный. Позже пример вашей операции на MT сделаю после работы. MT очень гибкий иструменты который позволяет всякими такими ньюансами управлять когда надо. Да очень простой OutBox. Если нужен сильный OutBox то он есть у NServiceBus (брат близнец MT) - Outbox • NServiceBus • Particular Docs
Подробнее о отложенных сообщениях для Таймаутов можно почитать тут MassTransit Schedule
Вот пример того как сделать Таймаут для саги
https://masstransit-project.com/usage/sagas/automatonymous.html#schedule
public interface OrderCompletionTimeoutExpired
{
Guid OrderId { get; }
}
public class OrderState :
SagaStateMachineInstance
{
public Guid CorrelationId { get; set; }
public string CurrentState { get; set; }
public Guid? OrderCompletionTimeoutTokenId { get; set; }
}
public class OrderStateMachine :
MassTransitStateMachine<OrderState>
{
public OrderStateMachine()
{
Schedule(() => OrderCompletionTimeout, instance => instance.OrderCompletionTimeoutTokenId, s =>
{
s.Delay = TimeSpan.FromDays(30);
s.Received = r => r.CorrelateById(context => context.Message.OrderId);
});
During(Accepted,
When(OrderCompletionTimeout.Received)
.PublishAsync(context => context.Init<OrderCompleted>(new { OrderId = context.Saga.CorrelationId }))
.Finalize());
}
public Schedule<OrderState, OrderCompletionTimeoutExpired> OrderCompletionTimeout { get; private set; }
}
Спасибо за интересную статью и подробные комментарии в коде!
Активно использую в своих проектах саги из MT.
В теоретическую часть я бы добавил обобщение саги с позиции микросервисов, и что они бывают хореографическими и оркестрируемыми. MT как раз используют оркестрацию.
От паттернов Requests отказались по соображением производительности, используем синхронный RPC через http/rest.
Плюс роутинг MTский по большей части руками настраиваем из-за соображений гетерогенных систем, когда разные микросервисы могут иметь разные языки реализации.
Публикации
Пример микросервисной архитектуры с Saga на MassTransit