Как мы внедрили откаты по схеме Saga в Cloudflare Workflows

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

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

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

1. Списание со счета в банке A

2. Зачисление на счет в банке B

3. Отправка подтверждения по электронной почте обоим владельцам счетов

Что произойдет, если шаг 2 (зачисление на счет в банке B) завершится сбоем? После успешного списания в банке A транзакция зафиксирована, и деньги покинули его систему. Как координатор транзакции, вы не можете просто "отменить" операцию в системе банка A. Вместо этого деньги должны быть возвращены на счет в банке A с помощью новой операции, которая семантически обращает первую.

Такое сочетание операции и ее компенсирующей логики называется паттерном Saga.

До сегодняшнего дня разработчикам приходилось реализовывать собственную компенсирующую логику для отслеживания успешных операций, сбоев и действий, которые необходимо предпринять при сбое, вне прямых определений шагов. Теперь вы можете определить компенсирующую логику для каждого step.do() в качестве аргумента внутри самих шагов, сохраняя надежность вашего рабочего процесса и для отката.

// отслеживаем, что было выполнено, чтобы знать, что отменить
let debitA;
let creditB;
try {
  debitA = await step.do("debit-bank-a", () => bankA.debit(from, amount));
  creditB = await step.do("credit-bank-b", () => bankB.credit(to, amount));
  await step.do("notify", () => notifyBoth(from, to, amount));
} catch (error) {
  // откат в обратном порядке. каждая отмена — отдельный надежный шаг,
  // должен быть идемпотентным и продолжаться, если один из них не удался.
  if (creditB) {
    try {
      await step.do("reverse-credit-b", () => bankB.debit(to, amount, creditB.id));
    } catch (e) {
      await alertOnCall("reverse-credit-b failed", e);
    }
  }
  if (debitA) {
    try {
      await step.do("refund-debit-a", () => bankA.credit(from, amount, debitA.id));
    } catch (e) {
      await alertOnCall("refund-debit-a failed", e);
    }
  }
  throw error;
}

^{Без откатов}

// каждый шаг снабжается собственной отменой. добавьте шаг —
// добавьте его откат прямо здесь. никакого растущего блока catch,
// ручного упорядочивания или логики воспроизведения.
await step.do("debit-bank-a", () => bankA.debit(from, amount), {
  rollback: async ({ output }) => bankA.credit(from, amount, output.id),
});
await step.do("credit-bank-b", () => bankB.credit(to, amount), {
  rollback: async ({ output }) => bankB.debit(to, amount, output.id),
});
await step.do("notify", () => notifyBoth(from, to, amount));

^{С откатами}

Чтобы использовать откаты, просто передайте объект параметров, содержащий функцию rollback, в качестве последнего аргумента в step.do().

const debit = await step.do(
  "debit-account-a",
  async () => {
    return await bankA.debit({
      accountId: fromAccountId,
      amount,
      idempotencyKey: `${transferId}:debit-account-a`,
    });
  },
  {
    rollback: async () => {
      await bankA.credit({
        accountId: fromAccountId,
        amount,
        idempotencyKey: `${transferId}:rollback-debit-account-a`,
      });
    },
  }
);

// Ключи идемпотентности делают как прямые операции, так и откаты безопасными для повторения без дублирования перевода

const credit = await step.do(
  "credit-account-b",
  async () => {
    return await bankB.credit({
      accountId: toAccountId,
      amount,
      idempotencyKey: `${transferId}:credit-account-b`,
    });
  },
  {
    rollback: async ({ output }) => {
      if (output === undefined) {
        return;
      }

      await bankB.debit({
        accountId: toAccountId,
        amount,
        idempotencyKey: `${transferId}:rollback-credit-account-b`,
      });
    },
  }
);


// Если здесь произойдет сбой, мы, возможно, захотим отменить все предыдущие платежи. Пользователям не придётся оборачивать код в сложную логику try-catch только для того, чтобы отменить два мелких платежа (см. ниже)

await step.do("send-confirmation", async () => {
  await sendTransferConfirmation({ ... });
});

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

Если какой-либо шаг завершается сбоем, обработчики отката будут выполняться в обратном порядке step-start. Это звучит просто: запускать шаги отмены при сбое. На практике есть несколько деталей, которые делают API и модель выполнения важными.

1. Даже для отказавшего шага может потребоваться откат. Неудавшийся step.do() все еще может быть пригоден для отката, если он зарегистрировал обработчик отката.

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

Почему? Шаг мог частично взаимодействовать с внешней системой до сбоя. Например, платежный провайдер может списать средства, но шаг может завершиться ошибкой до того, как вернет chargeId в Workflows. Именно поэтому обработчики отката получают output, но должны обрабатывать случай output === undefined.

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

Когда откат запускается, Workflows находит подходящие вызовы step.do(), запускает их обработчики отката, а затем фиксирует окончательный сбой Workflow.

3. Порядок должен быть предсказуемым. Для последовательных Workflows порядок отката кажется очевидным:

1. Зарезервировать товар на складе.

2. Списать средства с карты.

3. Создать отправление.

4. Если отправление не удалось, вернуть деньги за карту и освободить товар.

Параллельные шаги делают это более сложным. Порядок завершения может отличаться от порядка запуска, поэтому Workflows использует обратный порядок запуска шагов (step-start), а не обратный порядок завершения.

Практические правила таковы:

1. Все запущенные или завершенные шаги с обработчиками отката являются подходящими.

2. Неудавшийся step.do() также подходит, если он зарегистрировал обработчик отката.

3. Обработчики запускаются в обратном порядке запуска шагов, а не порядке их завершения.

После того как мы определились с ожидаемым поведением, нам нужно было добавить этот новый паттерн в API Workflows. Откаты прошли через несколько итераций, прежде чем мы остановились на rollback options.

Первым подходом была плавная форма: step.do(...).rollback(...). Она хорошо читается. Прямое действие и компенсация находятся рядом, а место вызова выглядит как обычная цепочка вызовов JavaScript.

Проблема в том, что step.do() уже имеет важное значение: он запускает надежный шаг и возвращает Promise для результата шага. В Workers значения, похожие на Promise, особенно важны, потому что Workers RPC поддерживает конвейеризацию промисов (promise pipelining) — паттерн, унаследованный от таких систем, как Cap'n Proto.

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

const session = api.authenticate(apiKey);
const name = await session.whoami();

Здесь session — это еще не реальный объект сессии. Это скорее дескриптор сессии, которая скоро появится. Когда вы вызываете session.whoami(), Workers может отправить этот вызов удаленной стороне заранее со словами: «как только аутентификация создаст сессию, вызови на ней whoami()».

Это экономит один сетевой обход. Вызывающему коду не нужно ждать полного завершения authenticate(), прежде чем запрашивать whoami().

Мы рассматривали fluent API:

step.do("charge-card", chargeCard).rollback(refundCharge);

Для читателя это может выглядеть как «вызвать .rollback() на результате charge-card». Но откат не является частью вывода шага. Это часть параметров step.do(), регистрируемая до начала шага, чтобы Workflows знал, как компенсировать шаг в случае сбоя последующего шага.

Fluent API также усложняет анализ временных характеристик шагов. Сегодня step.do() запускает шаг при вызове, поэтому разработчики могут запустить шаг, заняться другой работой, а затем дождаться завершения первого шага:

const first = step.do("first", () => serviceA.call());

await step.do("second", () => serviceB.call());

await first;

В текущей модели выполнения first запускается немедленно, до second. Fluent API усложнил бы это. Воркфлоу пришлось бы ждать и проверять, добавляется ли .rollback(), прежде чем станет известно полное определение шага. Это могло бы задержать отправку шага в движок.

В предыдущем примере first мог бы начаться в момент await first вместо step.do("first", ...), уже после завершения second.

Это делает параллельные воркфлоу сложнее для понимания: время выполнения шага зависело бы не только от места вызова step.do(), но и от момента потребления возвращённого Promise.

Мы также рассматривали API в стиле строителя:

const charge = await step
	.saga("charge")
	.do(() => chargeCard())
	.rollback(() => refundCharge())
	.run();

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

Но он добавляет излишнюю формальность. Каждый шаг требует финального .run(), забыть .run() легко и трудно заметить без инструментов, а простые случаи с одним шагом начинают выглядеть как цепочки конфигурации. Также вводится новый строитель step.saga(), что нарушает существующий паттерн step.<action>. Самое главное — он заставляет step.do() выглядеть как устаревшее API, а не как основной примитив воркфлоу. Целью отката было расширить step.do(), а не заменить его.

step.do(..., { rollback })

В итоге мы выбрали явную форму, где откат является метаданными шага.

Таким образом, каждый откат определяется внутри самого прямого шага. Каждый обработчик получает ошибку, вызвавшую начало отката, контекст шага и вывод — либо сохранённое значение, возвращённое прямым шагом (может быть undefined), либо undefined, если шаг не выполнился до сохранения значения.

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

Важно, что исходный сбой воркфлоу остаётся отдельным: откат — это то, что воркфлоу делает после сбоя, а не причина сбоя.

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

{
  rollback: async ({ output }) => {
    await bankA.credit({ accountId: fromAccountId, amount, transferId: `${transferId}-reversal` });
  },
  rollbackConfig: {
    retries: { limit: 10, delay: '30 seconds', backoff: 'exponential' },
    timeout: '2 minutes',
  },
}

Это соответствует желаемой ментальной модели событий жизненного цикла. step.do() уже описывает долговечную единицу работы, которую воркфлоу записывает, повторяет и позже показывает в логах. Откат — это ещё одно поведение жизненного цикла для той же единицы работы. Оно должно находиться рядом с определением шага, а не в отдельной обёртке или строителе.

• Шаг по-прежнему запускается, когда step.do() запускается обычно.

• Возвращённый промис по-прежнему представляет результат шага.

• Параллельный код воркфлоу сохраняет ту же модель выполнения.

• Опции повторных попыток и тайм-аута для отката находятся рядом с обработчиком отката.

• Существующие вызовы step.do() продолжают работать точно так же, как и сейчас.

Эта форма немного более явная, чем fluent API, но эта явность полезна. Операция и её компенсация по-прежнему находятся в одном месте, и API не вводит новый строитель шага или новый тип промиса. Разработчики, уже понимающие step.do(), должны изучить лишь один дополнительный объект options.

Это менее магично, но проще в освоении и понятнее.

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

Обычный step.do() уже имеет долговечную запись. Воркфлоу записывает, что шаг начался, завершился ли он, что вернул и нужно ли его пропустить вместо повторения при возобновлении воркфлоу позже.

Откаты добавляют к этой записи ещё одну вещь: зарегистрировала ли логика компенсации.

Это означает, что воркфлоу имеет две части информации, которые нужно сопоставить в случае сбоя.

Первая — это долговечная история шагов. Движок воркфлоу хранит данные, чтобы знать, что выполнялось, что завершилось, какой вывод был сохранён и был ли зарегистрирован откат.

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

В Workers RPC такая вызываемая ссылка называется заглушкой (stub). Заглушка позволяет одной части системы вызывать код, выполняющийся в другом месте. Заглушки также имеют время жизни — они могут быть освобождены, когда вызов или контекст выполнения завершается. Если нужно сохранить заглушку после этого момента, Workers RPC предоставляет метод dup(), который создаёт ещё один дескриптор того же целевого объекта.

Для отката эта модель полезна. Долговечная история шагов записывает, что требует компенсации. Заглушка отката даёт воркфлоу способ вызвать код компенсации. И поскольку обработчики отката могут пережить непосредственный вызов step.do(), который их зарегистрировал, воркфлоу хранит собственную вызываемую ссылку на обработчик для фазы отката.

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

Это становится более тонким, когда воркфлоу необходимо восстановиться после перезапуска.

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

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

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

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

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

• Какие шаги запускались?

• Какие шаги завершились?

• Какой неудачный шаг может всё ещё требовать очистки?

• Какие шаги зарегистрировали обработчики отката?

• Какие выходные данные должен получить каждый обработчик отката?

• В каком порядке должна выполняться компенсация?

Затем воркфлоу вызывает каждую заглушку отката с контекстом отката: исходная ошибка, контекст шага и вывод шага, если он был сохранён.

Деталь порядка важна. В обычном JavaScript, особенно с Promise.all(), порядок завершения не всегда совпадает с порядком запуска. Если шаг A запускается первым, а шаг B вторым, шаг B может завершиться первым. Для отката воркфлоу использует сохранённый порядок запуска как стабильный источник истины, а затем разворачивает его в обратном порядке.

Обработчики отката также выполняются через обычный механизм шагов воркфлоу. Это означает, что компенсация получает те же операционные свойства, которые вы ожидаете от воркфлоу: повторные попытки, тайм-ауты, события жизненного цикла, логи и итоговый зафиксированный результат. Если обработчик отката продолжает давать сбой после настроенных повторных попыток, воркфлоу записывает результат отката как «неудачный», прекращает выполнение оставшихся обработчиков отката, и экземпляр воркфлоу в итоге переходит в состояние Errored (с ошибкой).

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

Именно поэтому API помещает откат непосредственно на step.do(). Откат — это не отдельный глобальный обработчик ошибок, а метаданные, прикреплённые к долговечной единице работы, которую Workflows уже понимает.

Наша первая итерация откатов включает: 

• Явные обработчики отката для каждого шага step.do()

• Последовательное выполнение отката

• Настройка повторных попыток и тайм-аута для компенсации

Далее мы хотим изучить:

• Поддержку отката для waitForEvent

• Поддержку параллельного выполнения отката

• Поддержку отката для Python Workflows

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