Temporal

Представьте, что вы разрабатываете интернет-магазин. Когда покупатель оформляет заказ, системе нужно последовательно выполнить несколько шагов:

1. Списать деньги с карты
2. Зарезервировать товар на складе
3. Оформить отправку заказа
4. Дождаться доставки посылки покупателю
5. Отправить письмо с просьбой оставить отзыв

Если пока не думать о сбоях, хочется написать примерно такой код:

function processOrder(order):
    chargeCustomer(order)
    reserveInventory(order)
    requestShipment(order)
 
    while not isDelivered(order):
        sleep(5 days)
 
    sendReviewEmail(order)

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

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

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

Именно для таких задач и создан Temporal.

Проектируем надежное исполнение с нуля

При первом знакомстве с Temporal такие понятия, как workflow, activity, воспроизведение истории и детерминированное выполнение, могут показаться непривычными. Самый простой способ понять, откуда они взялись и зачем нужны - попробовать спроектировать систему надежного исполнения самостоятельно.

Устойчивость к падению сервера

Начнем с самого базового требования: функция processOrder должна переживать падение сервера.

Первая сложность заключается в том, что сразу несколько шагов этой функции обращаются к внешним системам за пределами нашего процесса. Вызов chargeCustomer идет в платежную систему, reserveInventory обновляет запись в базе данных, а requestShipment отправляет запрос в сервис доставки.

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

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

Например, после списания денег и резервирования товара на складе история выполнения будет выглядеть так:

ChargeCustomerCompleted {
    result: paymentId_123
}

ReserveInventoryCompleted {
    result: reservationId_456
}

Теперь допустим, что сервер падает до того, как мы успели запросить отправку заказа.

Процесс, в котором выполнялась функция processOrder завершился, но результаты уже проделанной работы остались в надежном хранилище.

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

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

chargeCustomer(order)
→ ChargeCustomerCompleted уже есть в истории
→ вернуть paymentId_123

reserveInventory(order)
→ ReserveInventoryCompleted уже есть в истории
→ вернуть reservationId_456

requestShipment(order)
→ сохраненного результата нет
→ выполнить операцию
→ записать результат

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

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

Этот механизм называется воспроизведением (replay).

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

Так в нашем фреймворке естественным образом выделяются два типа кода. Код оркестрации можно безопасно воспроизводить заново, тогда как операции, которые взаимодействуют с внешним миром, должны выполняться отдельно, а их результаты - сохраняться в истории.

В Temporal воспроизводимый код оркестрации называется workflow, а внешние операции - activity.

Предсказуемость воспроизведения

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

Представим, что наш интернет-магазин отправляет заказы в тот же день, если они оформлены до 17:00. Добавим в функцию такое условие:

function processOrder(order):
    chargeCustomer(order)
    reserveInventory(order)
 
    if currentTime() < 17:00:
        requestSameDayPickup(order)
    else:
        requestNextDayPickup(order)
 
    waitUntilDelivered(order)
    sendReviewEmail(order)

Допустим, покупатель оформляет заказ в 16:59.

Workflow списывает деньги, резервирует товар на складе, видит, что на часах еще нет 17:00, и заказывает отгрузку на сегодня. В истории сохраняются такие события:

ChargeCustomerCompleted
ReserveInventoryCompleted
RequestSameDayPickupCompleted

После этого сервер падает.

Через несколько минут другой воркер начинает воспроизводить processOrder с самого начала. Шаги со списанием средств и резервированием товара сразу возвращают сохраненные результаты.

Затем выполнение снова доходит до проверки времени:

if currentTime() < 17:00:

Но на часах уже 17:03.

При первом запуске функция пошла по ветке requestSameDayPickup, а при воспроизведении выбирает ветку requestNextDayPickup.

Это серьезная проблема. В сохраненной истории записано, что следующим событием было RequestSameDayPickupCompleted, а код при воспроизведении требует выполнить совершенно другой шаг - requestNextDayPickup.

Воспроизведение работает только тогда, когда по одной и той же истории workflow каждый раз проходит по одному и тому же пути. Иными словами, код workflow должен быть детерминированным.

Чтобы решить проблему со временем, получать текущее время тоже нужно через фреймворк. Вместо того чтобы напрямую читать системные часы сервера, workflow запрашивает текущее время у фреймворка.

При первом выполнении фреймворк вернет 16:59 и сохранит в истории информацию, которая позволит повторить это значение позже. Во время воспроизведения он вернет то же самое сохраненное время, а не текущее показание часов на новом воркере.

Тогда workflow пойдет по нужной ветке, и ход его выполнения не разойдется с историей.

Temporal накладывает на код workflow точно такое же ограничение. Изнутри workflow нельзя напрямую обращаться к внешним системам или читать недетерминированные значения, такие как системное время машины. Вместо этого SDK Temporal предоставляет специальные безопасные операции, в том числе для чтения текущего времени, чтобы при воспроизведении код всегда вел себя одинаково.

Долгое ожидание в workflow

До сих пор наш workflow делал только те шаги, которые выполняются быстро. Но в нашем процессе обработки заказа был еще и такой вызов:

waitUntilDelivered(order)

Доставка может занять несколько дней.

Прежде чем разбираться, как именно мы узнаем о вручении посылки, решим более простую задачу. Допустим, сервис доставки сообщает, что посылка придет через пять дней, и нам нужно, чтобы workflow просто подождал пять дней и затем продолжил работу.

В коде это хочется записать так:

function processOrder(order):
    chargeCustomer(order)
    reserveInventory(order)
    requestShipment(order)
 
    sleep(5 days)
 
    sendReviewEmail(order)

Понятно, что вызов sleep(5 days) не должен буквально занимать процесс воркера на пять суток. Интернет-магазин может одновременно обрабатывать миллионы заказов, и почти все они в каждый момент времени просто ждут доставки.

Значит, само ожидание тоже нужно сделать надежным.

Когда выполнение workflow доходит до sleep(5 days), фреймворк вычисляет время срабатывания таймера и сохраняет его в надежное хранилище:

TimerStarted {
    fireAt: 11 ноября, 10:00
}

На этом этапе текущей работы у workflow больше нет. Мы можем полностью остановить его выполнение и освободить воркер для других задач.

Фреймворку нужен планировщик, который отслеживает таймеры с наступившим сроком срабатывания. Если хранить таймеры в базе данных отсортированными по времени срабатывания, планировщику не придется проверять каждый ожидающий workflow по отдельности. Через пять дней планировщик дойдет до нашего таймера и добавит в историю новое событие:

TimerFired

При следующем запуске workflow воспроизведение снова дойдет до шага с таймером:

sleep(5 days)
→ событие TimerStarted уже есть
→ событие TimerFired есть
→ продолжить выполнение

Если бы таймер еще не сработал, workflow снова остановился бы на этой строке.

Благодаря этому со стороны все выглядит так, будто workflow спит пять дней, пять месяцев или еще дольше, хотя все это время ни один процесс в памяти его не держит.

В Temporal этот механизм реализован в виде надежных таймеров (durable timers). Сервис истории сохраняет задачу таймера со сроком срабатывания. Фоновый обработчик таймеров вычитывает такие задачи, когда подходит срок, записывает в историю событие TimerFired и ставит workflow в очередь на выполнение на свободном воркере. Ниже мы подробно разберем, как эти внутренние сервисы взаимодействуют друг с другом.

Один обработчик таймеров обслуживает сразу множество workflow, поэтому каждому спящему процессу не нужен собственный поток или воркер. Если экземпляр сервиса истории упадет, его работу подхватит другой экземпляр и восстановит активные таймеры из хранилища. Сбой может немного задержать срабатывание таймера, но сам таймер при этом не потеряется.

Пробуждение workflow по внешнему событию

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

Посылка может приехать уже завтра, а может задержаться на неделю. На самом деле нам нужно, чтобы workflow ждал до тех пор, пока сервис доставки сам не сообщит о вручении заказа.

Вернемся к исходному шагу:

waitUntilDelivered(order)

Когда workflow доходит до этого места, делать ему пока нечего, поэтому его выполнение можно приостановить.

Проходит три дня, и сервис доставки присылает вебхук с уведомлением о том, что посылка доставлена.

Нашему фреймворку нужен способ передать эту информацию в уже запущенный workflow. Проще всего дописать новое событие в его историю:

ChargeCustomerCompleted
ReserveInventoryCompleted
RequestShipmentCompleted
PackageDelivered

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

waitUntilDelivered(order)
→ событие PackageDelivered есть
→ продолжить выполнение

sendReviewEmail(order)
→ выполнить операцию
→ записать результат

В результате workflow может днями или неделями находиться в режиме ожидания и возобновить работу, когда поступят новые данные.

В Temporal такие внешние сообщения называются сигналами (signals). Сигнал позволяет внешней системе передать данные в выполняющийся workflow. Temporal сохраняет полученные данные в истории workflow и будит его, чтобы код мог отреагировать на новое событие.

В нашем примере с заказом обработчик вебхука от сервиса доставки отправляет в workflow сигнал о вручении посылки. После этого вызов waitUntilDelivered(order) завершается, и система отправляет покупателю письмо с просьбой оставить отзыв.

Что такое Temporal?

Мы начали с обычной функции и шаг за шагом, от одной проблемы к другой, собрали почти всю модель исполнения Temporal.

Temporal - это система оркестрации workflow, построенная вокруг надежного исполнения. В workflow описывается воспроизводимая логика выполнения, а операции, которые обращаются к внешнему миру, выносятся в activity. Temporal сохраняет результаты их работы в истории выполнения, за счет чего после любого сбоя может заново воспроизвести workflow и восстановить его состояние.

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

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

workflow processOrder(order):
    chargeCustomer(order)
    reserveInventory(order)
    requestShipment(order)

    waitUntilDelivered(order)

    sendReviewEmail(order)

Теперь разберемся в самом интересном: как именно Temporal реализует все это внутри.

Как Temporal устроен внутри

В общих чертах архитектура Temporal выглядит так:

• Frontend Service - это точка входа в кластер Temporal, без собственного состояния. К нему подключаются клиентские приложения и воркеры, а он маршрутизирует их запросы к нужным внутренним сервисам.
• History Service отвечает за надежное хранение состояния выполняющихся workflow. Он обрабатывает завершение activity, входящие сигналы, таймеры и решения, которые принимает код workflow.
• Matching Service управляет очередями задач, которые связывают кластер Temporal с вашими воркерами.
• Workers - это процессы вашего приложения, в которых выполняется код. Temporal выступает в роли оркестратора и не запускает внутри себя ни код workflow, ни код activity. Воркеры опрашивают Temporal в ожидании новых задач, выполняют нужный код и отправляют результат обратно.

Как хранится состояние workflow

Кластер Temporal может одновременно управлять миллионами запущенных workflow, поэтому одна машина сервиса истории не может отвечать за все сразу.

Вместо этого все выполняющиеся workflow распределяются по фиксированному числу шардов. Каждый экземпляр History Service владеет своим набором шардов и отвечает за обработку запросов и таймеров для тех workflow, которые находятся внутри этих шардов. Если экземпляры History Service добавляются, удаляются или падают, владение шардами перераспределяется между другими экземплярами.

Когда с одним из наших workflow обработки заказа что-то происходит, сервис истории переводит его в новое состояние.

Допустим, операция chargeCustomer успешно завершилась.

На концептуальном уровне сервис истории делает следующее:

Помимо уже знакомой нам истории событий Event History, Temporal сохраняет в базе структуру под названием изменяемое состояние (mutable state). По сути это материализованное представление текущего состояния workflow на стороне сервера: список выполняющихся activity, активных таймеров и дочерних workflow.

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

Кроме того, сервису истории нужно гарантировать, что сбой сервера не разорвет обновление состояния workflow и планирование следующего шага.

Представим, что activity завершилась и теперь нужно снова запустить сам workflow. Вместе с обновленным состоянием Temporal сохраняет в базу внутреннюю задачу передачи (Transfer Task). Фоновый обработчик читает эту задачу и пересылает ее в Matching Service, который делает соответствующую задачу workflow доступной для воркера.

По сути это паттерн Transactional Outbox. Даже если сервис истории упадет сразу после фиксации состояния workflow в базе, запись о задаче сохранится, и система будет знать, что эту работу нужно отправить на выполнение.

Таймеры работают по тому же общему принципу. Таймер сохраняется как внутренняя задача Timer Task. Когда наступает время срабатывания, Temporal записывает событие о срабатывании таймера и создает задачу для очередного запуска workflow.

При самостоятельном развертывании в качестве надежного хранилища можно использовать PostgreSQL, MySQL или Cassandra. Схему базы данных для истории событий, изменяемого состояния и внутренних задач предоставляет Temporal.

Как Temporal избегает постоянного воспроизведения с нуля

Воспроизведение дает Temporal надежный механизм восстановления после сбоев, но если бы длинный workflow приходилось заново прогонять по всей истории событий при каждом новом шаге, это обходилось бы слишком дорого.

Поэтому воркеры кэшируют состояние workflow в оперативной памяти.

Когда воркер впервые получает задачу на выполнение workflow, он восстанавливает состояние workflow по истории событий и оставляет его в памяти в кэше workflow.

После этого Temporal старается направлять следующие задачи для этого workflow на тот же самый воркер. Такая оптимизация называется привязкой выполнения (sticky execution).

Под капотом воркер опрашивает сразу две очереди: общую очередь задач и свою индивидуальную закрепленную очередь (sticky queue). Как только воркер сохранил workflow у себя в кэше, следующие задачи для этого workflow можно направлять через закрепленную очередь, чтобы другому воркеру не приходилось заново восстанавливать состояние с нуля.

При этом кэш остается исключительно оптимизацией. Если воркер упадет или workflow будет вытеснен из кэша, любой другой воркер сможет восстановить его по полной истории событий. В этом случае Temporal отправит следующую задачу workflow через общую очередь задач, откуда ее подхватит свободный воркер.

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

Воркеры и очереди задач

Как именно Temporal находит воркер для выполнения следующего фрагмента кода? В приложении каждой задаче назначается имя очереди задач (task queue), например payments, а воркеры, способные выполнять эту работу, настраиваются на опрос очереди с тем же именем.

Очередь задач - это именованный адрес назначения для работы. Управляет этими очередями и раздает задачи опрашивающим их воркерам сервис сопоставления Matching Service. Когда сервису истории нужно запустить код, его фоновые обработчики отправляют задачу в Matching Service для соответствующей очереди, а тот передает ее одному из воркеров, ожидающих работу на этой очереди.

В системе есть два основных типа задач:

• Задача workflow (workflow task) говорит воркеру запустить код самого workflow и определить, какие шаги нужно выполнить дальше.
• Задача activity (activity task) говорит воркеру выполнить конкретную операцию, например обратиться к внешнему API или отправить письмо.

Эти задачи не обязаны выполняться на одних и тех же машинах.

Например, маршрутизацию можно выстроить так:

Благодаря этому разные виды activity могут работать на совершенно разных пулах воркеров. Вычислительно тяжелые задачи можно вынести на одни машины, задачи для видеокарт - на другой пул серверов, а чувствительные платежные операции запускать в отдельном изолированном сервисе. Если же такое разделение не нужно, один и тот же процесс воркера может выполнять и код workflow, и код activity.

Сервису Matching Service не нужно ничего знать о внутренней реализации этих воркеров. Он просто передает задачи тем воркерам, которые опрашивают нужную очередь.

Эти очереди реализованы внутри самого Temporal. Если воркер в данный момент уже опрашивает очередь, Matching Service может отдать ему задачу напрямую. В противном случае задача сохраняется в базу данных и ждет, пока освободится воркер.

Сами очереди задач также можно разделить на несколько разделов между экземплярами Matching Service, чтобы повысить пропускную способность. За счет этого работа с нагруженной очередью не упирается в возможности одной машины.

Как масштабируется кластер

Одна из главных причин, по которой инженеры выбирают Temporal, состоит в том, что количество отслеживаемых процессов может расти без необходимости держать по воркеру на каждый из них. В нашем интернет-магазине миллионы заказов могут одновременно ждать доставки, тогда как реальную работу прямо сейчас нужно выполнять лишь для малой их доли.

Ожидающие заказы занимают место в базе данных, но не удерживают воркеры. Если в какой-то момент придет волна уведомлений о доставке, мы сможем добавить воркеров, чтобы разослать письма с просьбой об отзыве. Объем вычислительных ресурсов для воркеров зависит от количества работы, готовой к выполнению прямо сейчас, а не от общего числа открытых заказов.

Сам кластер Temporal масштабируется отдельно и отвечает за координацию. Дополнительные экземпляры Frontend Service принимают растущий трафик API, новые экземпляры History Service делят между собой шарды истории, а дополнительные экземпляры Matching Service обслуживают разделы очередей задач. Если узким местом становится обновление состояния workflow, добавление воркеров приложения проблему не решит - потребуется нарастить мощность сервиса истории или его базы данных.

Такое разделение позволяет добавлять ресурсы именно там, где растет нагрузка. Если скопилась очередь неотправленных писем - добавляем почтовые воркеры. Если вырос поток обновлений workflow - масштабируем сервисы Temporal и уровень хранения данных.

Когда стоит использовать Temporal

Temporal хорошо подходит для многошаговых процессов, которые должны переживать сбои и долгие периоды ожидания.

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

Присмотреться к Temporal стоит в следующих случаях:

• Процесс состоит из нескольких зависимых шагов, которые должны надежно выполниться. Конвейер постобработки видео на YouTube нарезает видеоролик на фрагменты, параллельно перекодирует их и собирает результаты в готовые файлы.
• Процесс может долго оставаться открытым в ожидании события. Подбор машины в Uber ждет ответа от водителя или истечения таймаута, прежде чем решить, предлагать ли поездку следующему кандидату.
• Нужно восстанавливаться после частичного выполнения. Если воркер обработки видео на YouTube упадет на середине направленного ациклического графа задач, Temporal восстановит состояние workflow и продолжит планировать только незавершенную работу.
• Повторные попытки и обработка сбоев становятся заметной частью логики приложения. В какой-то момент вы фактически начинаете писать собственный движок оркестрации workflow.

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

Примеры из наших разборов задач

Мы предлагаем использовать Temporal в трех наших разборах задач по System Design:

• Проектирование Uber использует workflow, чтобы предложить поездку одному водителю, подождать 10 секунд и перейти к следующему кандидату в случае отказа или таймаута.
• Проектирование YouTube использует Temporal для оркестрации направленного ациклического графа задач, который разбивает загруженное видео на части, параллельно обрабатывает фрагменты и собирает итоговый результат.
• Проектирование Распродажи использует надежный таймер Temporal как один из способов автоматически снимать резерв товара.

В каждой из этих задач оркестрацию можно было бы собрать самостоятельно из таблиц состояния в базе данных, очередей, таймеров и воркеров. Temporal позволяет вместо этого выразить всю эту координацию в коде workflow.

Ограничения Temporal

Temporal нужен далеко не везде, где встречается асинхронная обработка.

• Простые фоновые задачи. В нашем разборе системы уведомлений мы сознательно используем обычную очередь вместо Temporal, поскольку сама суть этой задачи на собеседовании заключается в проектировании механизма доставки и повторных попыток. Для одиночной независимой задачи обычная очередь, такая как SQS, зачастую оказывается намного проще.
• Потоковая обработка данных. Temporal не заменяет Kafka или другие платформы потоковой обработки событий. Если задача состоит прежде всего в непрерывной обработке большого потока событий, выбирайте систему, созданную специально для этого.
• Внешним операциям по-прежнему нужна идемпотентность. Temporal может повторять выполнение activity, а значит вызовы к системам, таким как Yookassa или ваша база данных, могут отправляться больше одного раза. Вам все равно нужно позаботиться о том, чтобы повторное выполнение таких операций было безопасным.
• Ограничения на код workflow. Поскольку Temporal восстанавливает workflow с помощью детерминированного воспроизведения, писать код workflow совсем как обычный прикладной код нельзя. Операции, результат которых может измениться при повторном запуске, например запросы к базе данных или обращения к внешним API, должны выполняться внутри activity.

Заключение

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

Вместе с этим появляются и дополнительные требования. Код workflow должен вести себя предсказуемо при воспроизведении, а внешние операции должны быть безопасны при повторном вызове. Temporal сохраняет историю выполнения и отвечает за восстановление после сбоев, но ваше приложение по-прежнему само решает, что делать, если не прошел платеж, закончился товар на складе или покупатель отменил заказ.

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