BatSchool · Путь письма: workflow

В прошлом уроке письмо надёжно ложилось в книги и в очередь: окошко не теряло его и не писало дважды. Очередь из урока 106 выдаёт задание воркеру. Дальше начинается путь, и путь длинный: перегон до T-1, ночь, перегон до T-2, ночь, и так до адресата. Воркер, который держит этот путь в памяти, держит его ровно до первого перезапуска.

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

Перегон, который помнят

Начнём с одного перегона. Башня T-2 принимает письмо через сервис Hops и иногда отказывает:

export class HopFailed extends Schema.TaggedError<HopFailed>()('HopFailed', {
  tower: Schema.String,
}) {}

export class Hops extends Context.Service<
  Hops,
  { readonly relay: (letter: number, tower: string) => Effect.Effect<void, HopFailed> }
>()('liniya/Hops') {}

В обычном коде перегон это вызов hops.relay(letter, 'T-2'). Запустили, получили итог, забыли. Если воркер упал после перегона, но до следующего шага, при перезапуске перегон пойдёт снова: письмо придёт на T-2 второй раз.

Активность это перегон, который движок помнит:

yield* Activity.make({
  name: `hop ${tower}`,
  error: HopFailed,
  execute: hops.relay(letter, tower).pipe(Effect.retry({ times: 2, schedule: Schedule.spaced('1 second') })),
});

name это имя шага в записи пути. По нему движок узнаёт, что шаг уже был. Поэтому имя должно быть одним и тем же при каждом прогоне: hop T-2, а не hop ${Date.now()}.
error это схема беды. Итог активности движок пишет в журнал, а значит, кодирует: и успех, и ошибку.
execute это сам перегон. Повторы из урока 73 живут внутри: два повтора раз в секунду, и только если все три попытки провалились, перегон считается проваленным.

Записанный итог переживает перезапуск. Активность, которая один раз закончилась, второй раз не начнётся.

Весь путь

Путь письма собирается из перегонов:

export const CarryLetter = Workflow.make('CarryLetter', {
  payload: { letter: Schema.Int, path: Schema.Array(Schema.String) },
  success: Schema.String,
  error: HopFailed,
  idempotencyKey: ({ letter }) => `letter-${letter}`,
});

export const Arrived = DurableDeferred.make('arrived', { success: Schema.String });

export const CarryLetterLayer = CarryLetter.toLayer(
  Effect.fn(function* ({ letter, path }) {
    const hops = yield* Hops;
    for (const tower of path) {
      yield* Activity.make({
        name: `hop ${tower}`,
        error: HopFailed,
        execute: hops.relay(letter, tower).pipe(Effect.retry({ times: 2, schedule: Schedule.spaced('1 second') })),
      });
    }
    const by = yield* DurableDeferred.await(Arrived);
    return `письмо ${letter} принял ${by}`;
  }),
);

Разберём, что тут нового.

Workflow.make описывает путь, как Rpc.make описывал вызов в уроке 93: имя, схема входа, схема успеха, схема беды. Всё это схемы, потому что движок хранит вход и итог.
idempotencyKey это ключ из прошлого урока, только для пути: одно письмо, один путь. Второй запуск с тем же номером письма не начинает новый путь, а присоединяется к старому. Если старый уже закончился, execute сразу отдаёт его итог.
toLayer даёт телу пути сервисы и регистрирует путь в движке. Тело это обычный Effect.fn, а for по башням обычный for.
DurableDeferred это Deferred из урока 66, который переживает перезапуск. Путь ждёт расписки адресата. Ждать он может неделю: тело пути при этом не держит файбер, движок разбудит его по расписке. Ждёт только тот, кто вызвал execute и хочет итог.

Расписку присылает адресат, снаружи пути:

export const confirmArrival = (letter: number, path: ReadonlyArray<string>, by: string) =>
  Effect.flatMap(
    DurableDeferred.tokenFromPayload(Arrived, { workflow: CarryLetter, payload: { letter, path } }),
    (token) => DurableDeferred.succeed(Arrived, { token, value: by }),
  );

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

Ночь между перегонами

Ночёвку на башне даёт DurableClock.sleep:

yield* DurableClock.sleep({ name: 'ночь на T-2', duration: '8 hours' });

Это не Effect.sleep. Обычный сон держит файбер восемь часов, и перезапуск обрывает его вместе с файбером. Долгий сон движок записывает в журнал как будильник: «разбудить этот путь в шесть ноль-ноль». Перезапустили машину в три, и будильник никуда не делся. Тикка права: часы, которые заводят, помнят время.

Под капотом

Пондер Стиб · единственный, кто понимает Гекс

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

Отсюда правило. Всё, что зависит от мира, живёт внутри активности: запрос, запись, Math.random, текущее время. Тело пути должно при повторе принимать те же решения. Если тело выберет следующую башню по Math.random(), повтор выберет другую, и журнал перестанет совпадать с телом.

Где тут дыра

const carry = (letter: number, path: ReadonlyArray<string>) =>
  Effect.forEach(path, (tower) =>
    Effect.andThen(hops.relay(letter, tower), Effect.sleep('8 hours')),
  );

На его примере письмо идёт по четырём башням, и всё хорошо. Соседний пример: в три ночи Гекс перезапускает машину. Цикл живёт в памяти воркера, и вместе с воркером пропадает знание, что две башни уже пройдены. Очередь отдаёт задание снова, цикл начинает с T-1, и первые две башни получают письмо второй раз. Если перезапуски случаются чаще, чем письмо проходит путь, оно не дойдёт никогда. В workflow тот же перезапуск ничего не повторит: два перегона записаны, а будильник ждёт в журнале.

Путь в тесте

Движок в памяти это WorkflowEngine.layerMemory. Перезапуски он не переживает, зато ведёт себя как настоящий: журнал, активности, ключи, будильники.

const tick = (seconds: number) =>
  Effect.gen(function* () {
    for (let second = 0; second < seconds; second++) {
      for (let spin = 0; spin < 5; spin++) yield* Effect.yieldNow;
      yield* TestClock.adjust('1 second');
    }
  });

Повторы перегонов ждут по часам, поэтому тест двигает TestClock. Двигать надо шагами и давать файберам уснуть (yieldNow): сдвиг часов разбудит только того, кто уже заснул. Сдвиг на час одним махом, пока путь ещё не дошёл до сна, ничего не разбудит.

Нестабильно. Модули effect/workflow (Workflow, Activity, DurableClock, DurableDeferred, WorkflowEngine) в 4.1 помечены @stability unstable: имена и опции могут поменяться в минорной версии. Урок написан на 4.1.0.

Линия, этап 6

Линия получает пути писем. Новый модуль src/route.ts: перегон как активность с повторами, путь как workflow с ключом по номеру письма, расписка адресата как DurableDeferred. Тесты проверяют порядок перегонов, то, что повторный запуск ничего не шлёт заново, и то, что капризная башня получает письмо после двух отказов.

EF-111-1★★

Путь письма 5 прошёл T-1 и T-3 и ждёт расписки. Кто-то вызвал CarryLetter.execute({ letter: 5, path: ['T-1', 'T-4'] }): тот же номер, другой путь. Что сделает движок? Что придёт на T-4? Чем это опасно и как это исправить в idempotencyKey?

EF-111-2★★

Ринцвальд хочет выбирать следующую башню по погоде: const next = weather() === 'буря' ? 'T-5' : 'T-2' прямо в теле пути, до активности. Погода к утру меняется. Что увидит движок, когда после будильника прогонит тело с начала? Куда перенести вызов weather()?

Практика

Глубже в архиве

Долгие процессы, саги и машины состояний без привязки к Effect разобраны в уроке про процессы и формальные методы раздела про бэкенд.

Дальше

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

Заказ ждёт в следующем уроке.