アプリケーション開発

Workflows の使用時に API 呼び出しを正確に 1 回だけ行う方法

2024年5月14日

Google Cloud Japan Team

※この投稿は米国時間 2024 年 5 月 4 日に、Google Cloud blog に投稿されたものの抄訳です。

はじめに

Workflows などの分散システムで課題となるのは、あるサービスから別のサービスに送信されるリクエストが、必要なときに正確に 1 回だけ処理されるようにすることです。たとえば、顧客の注文を配送キューに入れるとき、銀行口座から資金を引き出すとき、支払いを処理するときなどがあります。

このブログ投稿では、ウェブサイトが Workflows を呼び出し、Workflows が Cloud Functions を呼び出す例を紹介し、Workflows と Cloud Functions の両方のロジックが 1 回だけ実行されるようにする方法について説明します。また、HTTP コールバック、Pub/Sub メッセージ、Cloud Tasks を使用する場合に、Workflows を正確に 1 回だけ呼び出す方法についても説明します。

Workflows を正確に 1 回だけ呼び出す

オンラインショップを例として考えてみましょう。Workflows を使用して新規注文を作成し、Firestore に保存し、Cloud Functions を呼び出して支払いを処理するとします。

https://storage.googleapis.com/gweb-cloudblog-publish/images/image1_hVpeh5J.max-1500x1500.png

顧客から新規の注文が入り、ウェブサイトが Workflows に API 呼び出しを行ったもののエラーが発生しました。考えられるシナリオは 2 つあります。

（1）リクエストが失われ、ワークフローは呼び出されなかった

https://storage.googleapis.com/gweb-cloudblog-publish/images/image2_1VaigE8.max-500x500.png

（2）ワークフローは呼び出されて正常に実行されたが、レスポンスが失われた

https://storage.googleapis.com/gweb-cloudblog-publish/images/image3_Lfq82pR.max-600x600.png

ワークフローを正確に 1 回だけ実行するにはどうすればよいでしょうか？

これを解決するために、ウェブサイトは同じリクエストを再試行します。簡単なソリューションの 1 つは、Firestore にドキュメントがすでに存在しているかどうかをチェックすることです。

読み込んでいます...

processPayment ステップは、ドキュメントが正常に作成された場合にのみ実行されます。これは実質的に 1 ビットのステートマシンで、べき等であり、有効なソリューションです。ただし、このソリューションの欠点は、拡張性がないことです。状態を変更する前に、このハンドラで追加の作業を完了したり、システム内の状態の数を拡張したりすることが必要な場合があるかもしれません。次に、同じ問題に対するより高度なソリューションを考えてみましょう。

Workflows から Cloud Functions を 1 回だけ呼び出す

このワークフローが Cloud Functions を使用して支払いを処理する場合を見てみましょう。Cloud Functions を呼び出すには、次のような手順が必要になります。

読み込んでいます...

デフォルトでは、Workflows は HTTP リクエストで最大 1 回の配信（再試行なし）を提供します。通常、99.9% 以上の確率で呼び出しは成功し、レスポンスを受信するので問題は発生しません。

まれに失敗した場合は、ConnectionError が発生することがあります。前述したウェブサイトからワークフローへの状況と同様に、このワークフローはどのシナリオが発生したかを把握できません。その一方で、同様に、再試行を追加することもできます。

これを処理するために、デフォルトの再試行ポリシーを追加してみましょう。

読み込んでいます...

リクエストは Cloud Functions によって受信されるが、レスポンスが失われるという 2 つ目の配信シナリオについて考えてみましょう。再試行を追加することで、Workflows は Cloud Functions を複数回呼び出すことになります。このような場合、Cloud Functions のコードの実行を 1 回だけにするにはどうすればよいでしょうか？

Firestore の支払い状態をチェックして更新するには、Cloud Functions に追加のロジックを追加する必要があります。

https://storage.googleapis.com/gweb-cloudblog-publish/images/image4_HtrYAfU.max-1500x1500.png

さらに、Firestore でワークフローの EXECUTION_ID を追跡し、以下の order_state 列挙型を使用して支払い処理に柔軟性を高めるとします。

読み込んでいます...

前のワークフローを拡張し、Cloud Functions を呼び出して支払いを処理できます。

読み込んでいます...

main:
 params: []
 steps:
   - init:
       assign:
         - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
         - order_id: "12345"  # 実際には注文 ID をワークフロー パラメータとして渡す（例: ${params[0]}）
         - firestore_collection: "orders"
         - URL: https://us-central1-<your_project_id>.cloudfunctions.net/processpayment
   - create_document:
       try:
         call: googleapis.firestore.v1.projects.databases.documents.createDocument
         args:
           collectionId: ${firestore_collection}
           parent: ${"projects/" + project_id + "/databases/(default)/documents"}
           query:
             documentId: ${order_id}
           body:
             fields:
               order_state:  # 初期状態を設定
                 stringValue: "payment_not_processed"
               workflow_id:  # さらに、このワークフローの実行 ID も追跡
                 stringValue: ${sys.get_env("GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID")}
       except:
         as: e
         steps:
           - endEarly:
               return: ${e}  # ドキュメントがすでに存在する場合は例外が発生（例: ${e.code == 409}）
   - processPayment:
       try:
         call: http.post
         args:
             url: ${URL}  # 呼び出しが複数発生する可能性あり
             auth:
               type: OIDC
             body:
               order_id: ${order_id}
         result: r
       retry: ${http.default_retry}
   - returnStep:
       return: ${r}

以下は、支払いを処理する Cloud Functions（Node.js v20）です。

読み込んでいます...

const functions = require('@google-cloud/functions-framework');
const firestore = require('@google-cloud/firestore');

functions.http('helloHttp', (req, res) => {
   const fs = new firestore.Firestore();
   try{
// Firestore から現在の状態を読み取り、同じトランザクション内で更新して、このハンドラをべき等にする。トランザクションを使用することが重要。注: 複数回実行できるが、コミットされるのは 1 回のみ。
       return fs.runTransaction(t => {
           const docRef = fs.doc("orders/" + req.body.order_id);
           return t.get(docRef).then(doc => {
               console.log(doc, '=>', doc);
               var state = doc.data().order_state
               // Only process the order if we haven't already
               if (state == "payment_not_processed") {
                   // 支払い関連の処理を行う（例: 別の Firestore ドキュメントからの口座引き落としなど）
                   // ...
                   //
                   state = "payment_successful"
                   t.update(docRef, {order_state: state})
                   res.status(200).send(state);
                   return
               }
               res.status(200).send("request ignored, state already: " + state);
           });
       }).then(result => {
           console.log('Transaction result: ', result);
       });
   } catch (e) {
       console.log('Transaction failure:', e);
   }   
});

package.json

読み込んでいます...

重要なポイントは、すべての支払い処理作業が 1 つのトランザクション内で発生し、すべてのアクションがべき等になるということです。このトランザクション内のコードは、ワークフローの再試行によって複数回実行される可能性がありますが、コミットされるのは 1 回のみです。

HTTP コールバック、Pub/Sub、Cloud Tasks の場合

これまで、ウェブサイトからワークフローへのリクエスト、および Workflows から Cloud Functions へのリクエストを 1 回だけ行う方法について説明してきました。Workflows の呼び出しや再開をする方法は、他にも HTTP コールバック、Pub/Sub メッセージ、Cloud Tasks などがあります。これらのリクエストを正確に 1 回だけ行うにはどうすればよいでしょうか？次は、これについて見ていきましょう。

コールバック

幸いなことに、Workflows の HTTP コールバックはデフォルトで完全にべき等です。コールバックが失敗した場合でも安全に再試行できます。以下にその例を示します。

読み込んでいます...

最初のコールバックが外部の呼び出し元にエラーを返したと仮定します。呼び出し元は、このエラーでは、ワークフローのコールバックを受信したかどうかを把握できない可能性があるため、コールバックを再試行する必要があります。2 回目のコールバックで、呼び出し元は以下の HTTP ステータスコードのいずれかを受け取ります。

429: 最初のコールバックが正常に受信されたことを示します。2 回目のコールバックはワークフローによって拒否されます。
200: 2 回目のコールバックが正常に受信されたことを示します。最初のコールバックは受信されなかったか、受信されて正常に処理されています。後者の場合、await_callback は 1 回しか呼び出されないため、2 回目のコールバックは処理されません。2 回目のコールバックはワークフローの終了時に破棄されます。
404: コールバックが利用できないことを示します。最初のコールバックが受信されて処理されたのちワークフローが完了したか、あるいはワークフローが実行されていないか（そして、失敗したなど）のいずれかです。これを確認するには、API リクエストを送信してワークフローの実行状態をクエリする必要があります。

詳細については、コールバックを使用してワークフローをただ 1 回だけ呼び出すをご覧ください。

Pub/Sub メッセージ

Pub/Sub を使用して新しいワークフローの実行をトリガーする場合、Pub/Sub は Workflows で1 回以上の配信を使用し、配信が失敗した場合は再試行します。Pub/Sub メッセージは自動的に重複除去されます。その時間枠内（24 時間）での重複配信を心配する必要はありません。

Cloud Tasks

Cloud Tasks は、ワークフローの実行をバッファリングするためによく使用されます。1 回以上の配信を提供しますが、メッセージの重複除去は行いません。ワークフローハンドラはべき等である必要があります。

まとめ

リクエスト処理を正確に 1 回だけ行うことは難しい問題です。このブログ投稿では、Workflows の使用時に、正確に 1 回だけのリクエスト処理が必要になるシナリオをいくつか紹介しました。また、それを実装するためのアイデアもいくつか紹介しました。どのソリューションが適切かは、実際のユースケースや関連するサービスによって異なります。

-Workflows チームリーダー Randy Spruyt

-Cloud デベロッパー アドボケイト Mete Atamel

投稿先