Task Queue и жизненный цикл Worker

Минимальное «Согласование» уже разделено на Client, Service и Worker. Теперь рассмотрим, как Task Queue связывает эти компоненты и что происходит, когда исполнитель отсутствует. Главный результат урока — перестать считать Worker владельцем единственной копии процесса. Он исполняет код, но длительное состояние и ожидающие задания не должны существовать только в его памяти.

Используем полные worker.ts и client.ts предыдущего урока, стадию 02-minimal. Автор не запускал Worker и не наблюдал остановку. Примеры сценариев ниже ожидаемые; они не являются выполненной проверкой отказоустойчивости проекта.

Имя очереди как договорённость

Константа TASK_QUEUE равна soglasovanie. Client передаёт её при start, Worker указывает то же имя при создании. Тем самым запуск связывается с исполнителем, который знает нужный Workflow type и Activities. Namespace тоже совпадает: default. Совпадение только имени очереди в разных namespaces не связывает их исполнения.

Task Queue не хранится в вашем TypeScript-массиве и не является Redis-ключом, который программа самостоятельно обрабатывает. Очередью управляет Temporal Service. Worker опрашивает Service и получает задания. Это позволяет Client принять запуск даже в ситуации, когда подходящий Worker временно отсутствует.

Если Client отправил workflow в очередь soglasovanie-new, а Worker слушает soglasovanie, запуск может существовать, но ожидаемая работа не выполняется. Повторные starts с новыми ID только создадут больше ожидающих исполнений. Сначала нужно сопоставить namespace, taskQueue и зарегистрированные типы, а не объявлять все заявки потерянными.

В учебной конфигурации один Worker содержит и Workflow, и Activities. Для более крупного приложения их можно разделять по ролям и очередям, однако тогда нужно явно настроить маршрутизацию Activity-вызовов. Простое копирование worker.ts с другим названием не создаёт предметно правильного распределения.

Workflow Task и Activity Task

Workflow Task нужен для продвижения детерминированной последовательности: обработать новые события, принять решения и сформировать команды. Activity Task нужен для внешнего действия, например HTTP POST gateway. Поэтому «Workflow ждёт публикацию» не означает, что один поток Node бесконечно занят выполнением всех частей процесса.

В минимальной стадии после start Worker вычисляет необходимость reserveSlot. Затем Activity выполняет HTTP-запрос и возвращает результат. История получает соответствующее завершение, а последующий Workflow Task решает вызвать publish. Такая последовательность помогает разделить ожидание внешней системы и воспроизведение предметных решений.

Остановка Worker между этими задачами не должна удалить факт завершённой брони из Service. Другой подходящий Worker сможет восстановить дальнейшие решения по истории. Однако если остановка произошла во время внешнего эффекта, нужна идемпотентность Activity: факт на стороне gateway мог уже измениться до получения подтверждения Service.

Ресурсы Worker также ограничены. maxConcurrentActivityTaskExecutions: 4 ограничивает число одновременно исполняемых Activities в нашем процессе. Это не предел общего числа Workflow в Service и не запрет хранить тысячи ожидающих таймеров. Нужно различать объём долговременного состояния и число активных внешних операций.

Корректная остановка исполнителя

В полном worker.ts обработчики сигналов вызывают worker.shutdown(). После выхода из worker.run() закрывается NativeConnection. Фрагмент:

const shutdown = () => worker.shutdown();
process.once('SIGINT', shutdown);
process.once('SIGTERM', shutdown);
try {
  await worker.run();
} finally {
  process.removeListener('SIGINT', shutdown);
  process.removeListener('SIGTERM', shutdown);
}

Это описание жизненного цикла процесса, не отмена предметной заявки. Shutdown Worker и cancel Workflow имеют разный смысл. Первый прекращает работу исполнителя, второй сообщает процессу, что бизнес-операцию нужно остановить и, возможно, выполнить cleanup. Подмена одного другим скрывает ответственность за незавершённые внешние действия.

При обычном завершении Worker получает возможность закончить активную работу согласно настройкам SDK. Принудительное уничтожение процесса такой возможности не даёт. В обоих случаях состояние в Service не должно зависеть от корректного выхода Node, но сроки повторного назначения Activities и обнаружения остановки зависят от timeout и heartbeat.

Не помещайте в finally Worker безусловное releaseSlot всех заявок. Процесс Worker может обслуживать много исполнений и не является владельцем их предметного завершения. Cleanup относится к конкретному Workflow и должен быть записан в его модели. Иначе обычный выпуск нового Worker случайно отменит действующие согласования.

Диагностика ожидающей работы

Рассмотрим будущий сценарий: читатель останавливает Worker, запускает новую заявку через Client и затем снова запускает Worker той же стадии. Ожидается, что запуск сохранён, а после появления исполнителя продолжается работа. Из этого не следует, что пользователь увидит мгновенный результат: нужно время на получение и выполнение задач.

Если result ждёт, это не обязательно сбой Client. Команда handle.result() ждёт завершения соответствующего Workflow. Другой клиентский процесс может читать тот же результат по ID. Поэтому исчезновение терминала, выполнившего start, не должно считаться удалением заявки. Источником истины остаётся Service.

Для диагностики пригодны состояние исполнения в локальной UI и журналы Worker. При доступном Service, но отсутствующем Worker история покажет, что нужные задачи ещё не завершены. При недоступном gateway Activity будет выдавать отдельные ошибки. Наблюдение очереди и внешнего I/O нужно сопоставить, иначе одинаковое пользовательское «ещё ждёт» скрывает разные причины.

Два Worker могут слушать одну очередь, если они совместимы с исполняемыми историями и знают нужные типы. Это не означает, что оба одновременно выполняют весь Workflow независимо. Service распределяет задачи. Но два разных несовместимых варианта approvePublication на одной очереди создадут риск nondeterminism при обработке историй.

В курсе переключение WORKFLOW_STAGE служит учебному сравнению. Для новой стадии обычной заявки используют новый submissionId, а не произвольно продолжают старое исполнение другим кодом. В production стратегия совместимости включает versioning и правила развёртывания. Эти вопросы подробнее рассмотрим ближе к завершению серии.

Официальная модель Worker описана в руководстве TypeScript Workers, а параметры — в API Worker. Практический результат этого этапа — известная связь queue и исполнителя, корректный shutdown и понимание, почему временно остановленный Worker не должен быть единственным носителем заявки.

В следующем уроке добавим устойчивый таймер. Он сделает различие между жизнью процесса Node и сроком Workflow особенно заметным.

Оглавление серии