Cloudflare Workflows のために saga ロールバックをどのように構築したか

Cloudflare Workflows を使うと、組み込みのリトライと長時間実行プロセスにまたがる状態永続化を備えた、耐久性のあるマルチステップアプリケーションを構築できます。Workflow が実行されると、各ステップは外部システムを呼び出し、失敗をリトライし、再起動にまたがって状態を永続化できます。しかし、あるステップが失敗すると、完了した前のステップの作業が不整合な部分的状態で残る可能性があります。今日、Workflows に saga ロールバックを導入しました。これにより、失敗時に各ステップ内でロールバックロジックを宣言できるようになります。

例えば、異なる 2 つの銀行間で資金を移動するワークフローを考えてみます。

• Bank A の口座から引き落とす（Debit）
• Bank B の口座に入金する（Credit）
• 両口座所有者に確認メールを送る

ステップ 2（Bank B への入金）が失敗したらどうなりますか？

Bank A のデビットが成功するとトランザクションはコミットされ、そのシステムからお金が出ます。オーケストレータとして、Bank A のシステム内の操作を単純に "元に戻す" ことはできません。代わりに、最初の操作を意味的に逆にする新しい操作で Bank A の口座にお金を戻す必要があります。この操作とその補償ロジックの組み合わせを saga pattern と呼びます。

以前は、開発者は何が成功し何が失敗したか、失敗時にどのアクションを取るべきかをステップ定義の外で追跡し、自前で補償ロジックを実装する必要がありました。今では、各 step.do() の引数として補償ロジックを定義でき、ロールバックに対してもワークフローの耐久性を維持できます。

// track what completed so we know what to undo
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) {
  // unwind in reverse. each undo is its own durable step,
  // must be idempotent, and must keep going if one fails.
  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;
}

このような手動の try/catch によるアンワインドは冗長でエラーが入りやすいです。そこで、各ステップにロールバックを直接定義できるようにしました。

ロールバックを使うと、ステップの最後の引数として rollback 関数を含むオプションオブジェクトを渡すだけです。

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));

次に実際の使用例を示します。

Try it out

ロールバックを使うには、最後の引数として 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`,
      });
    },
  }
);

// The idempotency keys make both the forward operations and rollback operations safe to retry without duplicating the transfer
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`,
      });
    },
  }
);

// If we fail here, we may want to revert all previous payments. Users should not have to wrap their code in complex try-catch logic just to revert two small payments (see below)
await step.do("send-confirmation", async () => {
  await sendTransferConfirmation({ ... });
});

ロールバック関数は通常の Workflow ステップと同様に冪等であるべきです。課金を払い戻す場合は支払いプロバイダの idempotency key を使用し、在庫を解放する場合は複数回呼んでも安全な解放操作にしてください。

任意のステップが失敗した場合、ロールバックハンドラはステップの開始順序の逆で実行されます。一見単純ですが、API と実行モデルを考えると注意すべき細かい点がいくつかあります。

1. 失敗したステップ自身もロールバックの対象になり得る

   失敗した step.do() が rollback ハンドラを登録していれば、そのステップもロールバック対象になり得ます。ユーザーコードがエラーをキャッチして Workflow が継続する場合はロールバックは開始しませんが、あるステップのエラーがキャッチされたあとで別の理由により Workflow が最終的に失敗した場合、以前に登録されたハンドラはロールバックの対象になりうるため、rollback は実行されます。なぜなら、ステップが失敗する前に外部システムと部分的にやり取りしている可能性があるからです。例えば、支払いプロバイダが課金をキャプチャしたが、ステップが chargeId を Workflows に返す前に失敗することがあります。だからこそ rollback ハンドラは output を受け取りますが、output === undefined に対応する必要があります。

2. ロールバックは Workflow が失敗したときにのみ始まる

   ロールバックハンドラを追加しても、すべてのステップエラーがロールバックを引き起こすわけではありません。ユーザーコードがエラーをキャッチして処理を続行すれば Workflow は継続します。ロールバックは Workflow 自体が致命的に失敗しようとしているときに始まります。ロールバックが始まると、Workflows は対象となる step.do() 呼び出しを見つけ、その rollback ハンドラを実行してから最終的な Workflow の失敗を記録します。

3. 順序は予測可能である必要がある

   逐次的な Workflow では、ロールバック順序は直感的です：在庫を確保し、カードに請求し、配送を作成する。配送が失敗したらカードを返金して在庫を解放します。並列ステップではこれが微妙になります。完了順序は開始順序と異なる場合があるため、Workflows は完了順序の逆ではなく、ステップ開始順の逆を使います。

   実用的なルールは以下の通りです：

   • rollback ハンドラを持つ開始済みまたは完了済みの任意のステップは対象になる。
   • 失敗した step.do() も rollback ハンドラを登録していれば対象になる。
   • ハンドラは完了順ではなく、ステップ開始順の逆で実行される。

How we designed the API

期待される振る舞いを定めたら、この新しいパターンを Workflows API にどのように追加するかを考える必要がありました。ロールバックは複数の反復を経て、最終的に rollback options という形に落ち着きました。

なぜ fluent や builder スタイルにしなかったのか？

最初のアプローチは fluent 形式でした：

step.do(...).rollback(...)

見た目は読みやすく、順方向のアクションと補償が隣り合います。ですが、step.do() には重要な意味があります：それは耐久性のあるステップを開始し、そのステップ出力の Promise を返します。Workers では promise-like な値が特に重要で、Workers RPC は promise pipelining をサポートしています。これは Cap'n Proto のようなシステムから受け継がれたパターンです。Promise pipelining により、値が完全に返る前にその将来の値に対してメソッドを呼べます。例えば：

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

ここで session はまだ本物のセッションオブジェクトではありません。将来存在するセッションへのハンドルのようなものです。session.whoami() を呼ぶと、Workers は authenticate がセッションを作成したらすぐに whoami() を呼び出すようにリモート側に伝えられます。これによりラウンドトリップが省けます。

fluent API を使うと、読者には "charge-card の結果に対して .rollback() を呼んでいる" ように見えますが、rollback はステップの出力の一部ではなく、ステップが開始される前に登録されるべきメタデータです。Workflows は後続ステップが失敗した場合に補償できるよう、ステップが始まる前に rollback が登録されていることを知っている必要があります。

さらに fluent API はステップのタイミングの推論を難しくします。現在、step.do() は呼ばれたときにステップを開始するため、開発者はステップを開始して他の作業をしてから最初のステップを await できます。

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

現行モデルでは first は second の前にすぐ開始されます。fluent API では .rollback() がいつ添付されるかを見るまで step.do() の開始を遅らせる必要があり、これが step の送出タイミングを遅らせ得ます。そうすると並行 Workflow の挙動が分かりにくくなります。

次に builder スタイルを検討しました：

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

ビルダー API は Promise の曖昧さを避け、将来のステップレベルのオプションの置き場所を与えます。しかし、冗長さが増えます。各ステップに最後の .run() が必要になり、.run() を忘れるのは見落としやすくなりますし、シンプルな 1 ステップのケースでも設定チェーンのように見えてしまいます。さらに新しい step.saga() ビルダーを導入すると既存の step.<action> パターンと乖離します。最も重要なのは、rollback の目標が step.do() を置き換えることではなく拡張することだった点です。

rollback をステップのメタデータとして扱う

最終的に、rollback をステップのメタデータとして渡す明示的な形にしました。こうすることで、各ロールバックは順方向のステップ定義の中に定義されます。各ハンドラはロールバックを開始した原因のエラー、step コンテキスト、そして output を受け取ります。output は順方向ステップが永続化した値（undefined になり得る）か、ステップが値を永続化する前に失敗した場合は undefined になります。

ロールバックはライフサイクルイベントを発行するので、補償が開始されたか、どのロールバックハンドラが失敗したか、ロールバックが成功裏に完了したかを把握できます。重要な点は、元の Workflow の失敗は別物として残ることです：ロールバックは Workflows が失敗したあとに行う処理であり、それ自体が Workflow の失敗原因ではありません。

WorkflowStepConfig でカスタムのリトライやタイムアウトを定義できるのと同様に、rollback 固有の値は 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() はすでに Workflows が記録しリトライしログに表示する耐久的な単位を表しています。ロールバックは同じ単位の別のライフサイクル挙動です。ロールバックはステップ定義とともに存在すべきであり、別のラッパやビルダに置くべきではありません。

• step.do() は従来通りにステップを開始する。
• 返される Promise は引き続きステップ出力を表す。
• 並行する Workflow のコードは同じ実行モデルを維持する。
• rollback のリトライやタイムアウトオプションは rollback ハンドラの横に置かれる。
• 既存の step.do() 呼び出しはそのまま動作する。

この形は fluent API よりやや明示的ですが、その明示性は有益です。操作と補償は同じ場所にあり、API は新しいステップビルダや新しい種類の Promise を導入しません。既に step.do() を理解している開発者はすぐに使えます。