{"id":9730,"date":"2024-05-21T13:42:16","date_gmt":"2024-05-21T13:42:16","guid":{"rendered":"https:\/\/www.ntspl.co.in\/blog\/?p=9730"},"modified":"2024-08-20T11:54:37","modified_gmt":"2024-08-20T11:54:37","slug":"making-api-calls-exactly-once-when-using-workflows","status":"publish","type":"post","link":"https:\/\/www.ntspl.co.in\/blog\/making-api-calls-exactly-once-when-using-workflows\/","title":{"rendered":"Making API calls exactly once when using Workflows"},"content":{"rendered":"
\n

Introduction<\/strong><\/h2>\n

One challenge with any distributed system, including Workflows<\/a>, is ensuring that requests sent from one service to another are processed exactly once, when needed; for example, when placing a customer order in a shipping queue, withdrawing funds from a bank account, or processing a payment.<\/p>\n

In this blog post, we\u2019ll provide an example of a website invoking Workflows, and Workflows in turn invoking a Cloud Function<\/a>. We\u2019ll show how to make sure both Workflows and the Cloud Function logic only runs once. We\u2019ll also talk about how to invoke Workflows exactly once when using HTTP callbacks<\/a>, Pub\/Sub<\/a> messages, or Cloud Tasks<\/a>.<\/p>\n

Invoke Workflows exactly once<\/strong><\/h2>\n

Imagine you have an online store and you\u2019re using Workflows to create new orders, save to Firestore, and process payments by calling a Cloud Function:<\/p>\n<\/div>\n

\n
\n
\n
\"image1\"<\/figure>\n<\/div>\n<\/div>\n<\/div>\n
\n

A new customer order comes in, the website makes an API call to Workflows but receives an error. Two possible scenarios are:<\/p>\n

(1) The request is lost and the workflow is never invoked:<\/p>\n<\/div>\n

\n
\n
\n
\"image2\"<\/figure>\n<\/div>\n<\/div>\n<\/div>\n
\n

(2) The workflow is invoked and executes successfully, however the response is lost:<\/p>\n<\/div>\n

\n
\n
\n
\"image3\"<\/figure>\n<\/div>\n<\/div>\n<\/div>\n
\n

How can you make sure the workflow executes once?<\/p>\n

To solve this, the website retries the same request. One easy solution is to check if a document already exists in Firestore:<\/p>\n<\/div>\n

\n
\n
code_block<\/dt>\n
<ListValue: [StructValue([(‘code’, ‘main:rn params: []rn steps:rn – init:rn assign:rn – project_id: ${sys.get_env(“GOOGLE_CLOUD_PROJECT_ID”)}rn – order_id: “12345” # In practice we would pass in the order ID as a workflow parameter, e.g. ${params[0]}rn – firestore_collection: “orders”rn – URL: https:\/\/us-central1-<your_project_id>.cloudfunctions.net\/processpaymentrn – create_document:rn try:rn call: googleapis.firestore.v1.projects.databases.documents.createDocumentrn args:rn collectionId: ${firestore_collection}rn parent: ${“projects\/” + project_id + “\/databases\/(default)\/documents”}rn query:rn documentId: ${order_id}rn except:rn as: ern steps:rn – endEarly:rn return: ${e} # Exception is raised, e.g. ${e.code == 409} if doc already existsrn – processPayment:rn …’), (‘language’, ”), (‘caption’, <wagtail.rich_text.RichText object at 0x3e81cd479e50>)])]><\/dd>\n<\/dl>\n<\/div>\n
\n

The processPayment <\/code>step will execute only if a document is successfully created. This is effectively a 1-bit state machine, idempotent, and a valid solution. The downside of this solution is that it\u2019s not extensible. We might want to complete additional work in this handler before changing states, or expand the number of states within the system. Next, let\u2019s continue with a more advanced solution for the same problem.<\/p>\n

Invoke Cloud Functions from Workflows exactly once<\/strong><\/h2>\n

Let\u2019s see what happens when the workflow uses a Cloud Function to process the payment. You might have the following step to call Cloud Functions:<\/p>\n<\/div>\n

\n
\n
code_block<\/dt>\n
<ListValue: [StructValue([(‘code’, ‘- processPayment:rn call: http.postrn args:rn url: https:\/\/us-central1-<your_project_id>.cloudfunctions.net\/processpaymentrn auth:rn type: OIDC’), (‘language’, ”), (‘caption’, <wagtail.rich_text.RichText object at 0x3e81cd479c70>)])]><\/dd>\n<\/dl>\n<\/div>\n
\n

By default, Workflows offers at-most-once<\/strong> delivery (no retries) with HTTP requests. That\u2019s usually OK because 99.9+% of the time, the call is successful, and a response is received.<\/p>\n

In the rare case of failure, a ConnectionError<\/code><\/a> might be raised. As in the website-to-workflow situation discussed previously, the workflow can\u2019t tell which scenario occurred. Similarly, you can add retries.<\/p>\n

Let\u2019s add a default retry policy<\/a> to handle this:<\/p>\n<\/div>\n

\n
\n
code_block<\/dt>\n
<ListValue: [StructValue([(‘code’, “- processPayment:rn try:rn call: http.postrn args:rn url: https:\/\/us-central1-<your_project_id>.cloudfunctions.net\/processpaymentrn auth:rn type: OIDCrn retry: ${http.default_retry} # Retries up to 5 times, includes ‘ConnectionError'”), (‘language’, ”), (‘caption’, <wagtail.rich_text.RichText object at 0x3e81cd85cbb0>)])]><\/dd>\n<\/dl>\n<\/div>\n
\n

Let’s say the second delivery scenario occurs where the request is received by the Cloud Function but the response is lost. By adding retries, Workflows will likely invoke the Cloud Function multiple times. When this happens, how do you ensure that the code in the Cloud Function only runs once?<\/p>\n

You\u2019ll need to add extra logic to the Cloud Function to check and update the payment state in Firestore:<\/p>\n<\/div>\n

\n
\n
\n
\"image4\"<\/figure>\n<\/div>\n<\/div>\n<\/div>\n
\n

Let\u2019s also assume you want to track the workflow EXECUTION_ID<\/code> in Firestore and use the following order_state <\/code>enum to allow for additional flexibility in payment processing:<\/p>\n<\/div>\n

\n
\n
code_block<\/dt>\n
<ListValue: [StructValue([(‘code’, ‘payment_not_processed \/\/ Initial state when an order is createdrnpayment_declined \/\/ Payment was not successfulrnpayment_successful \/\/ Payment processed successfullyrn…’), (‘language’, ”), (‘caption’, <wagtail.rich_text.RichText object at 0x3e81cd9cbbb0>)])]><\/dd>\n<\/dl>\n<\/div>\n
\n

You can expand on the previous workflow and call a Cloud Function to process the payment:<\/p>\n<\/div>\n

\n
\n
code_block<\/dt>\n
<ListValue: [StructValue([(‘code’, ‘main:rn params: []rn steps:rn – init:rn assign:rn – project_id: ${sys.get_env(“GOOGLE_CLOUD_PROJECT_ID”)}rn – order_id: “12345” # In practice we would pass in the order ID as a workflow parameter, e.g. ${params[0]}rn – firestore_collection: “orders”rn – URL: https:\/\/us-central1-<your_project_id>.cloudfunctions.net\/processpaymentrn – create_document:rn try:rn call: googleapis.firestore.v1.projects.databases.documents.createDocumentrn args:rn collectionId: ${firestore_collection}rn parent: ${“projects\/” + project_id + “\/databases\/(default)\/documents”}rn query:rn documentId: ${order_id}rn body:rn fields:rn order_state: # We set an initial statern stringValue: “payment_not_processed”rn workflow_id: # And also track this workflow execution IDrn stringValue: ${sys.get_env(“GOOGLE_CLOUD_WORKFLOW_EXECUTION_ID”)}rn except:rn as: ern steps:rn – endEarly:rn return: ${e} # Exception is raised, e.g. ${e.code == 409} if doc already existsrn – processPayment:rn try:rn call: http.postrn args:rn url: ${URL} # Might get called multiple times!rn auth:rn type: OIDCrn body:rn order_id: ${order_id}rn result: rrn retry: ${http.default_retry}rn – returnStep:rn return: ${r}’), (‘language’, ”), (‘caption’, <wagtail.rich_text.RichText object at 0x3e81cd47faf0>)])]><\/dd>\n<\/dl>\n<\/div>\n
\n

Here\u2019s the Cloud Function (Node.js v20) that processes the payment:<\/p>\n<\/div>\n

\n
\n
code_block<\/dt>\n
<ListValue: [StructValue([(‘code’, ‘const functions = require(‘@google-cloud\/functions-framework’);rnconst firestore = require(‘@google-cloud\/firestore’);rnrnrnfunctions.http(‘helloHttp’, (req, res) => {rn const fs = new firestore.Firestore();rn try{rn\/\/ Reads the current state from Firestore and updates it within the same transaction to make this handler idempotent. Using a transaction is important. Note: It could be run multiple times but will only be committed once.rn return fs.runTransaction(t => {rn const docRef = fs.doc(“orders\/” + req.body.order_id);rn return t.get(docRef).then(doc => {rn console.log(doc, ‘=>’, doc);rn var state = doc.data().order_statern \/\/ Only process the order if we haven’t alreadyrn if (state == “payment_not_processed”) {rn \/\/ Do payment stuff, e.g. debit account from another Firestore documentrn \/\/ …rn \/\/rn state = “payment_successful”rn t.update(docRef, {order_state: state})rn res.status(200).send(state);rn returnrn }rn res.status(200).send(“request ignored, state already: ” + state);rn });rn }).then(result => {rn console.log(‘Transaction result: ‘, result);rn });rn } catch (e) {rn console.log(‘Transaction failure:’, e);rn } rn});’), (‘language’, ”), (‘caption’, <wagtail.rich_text.RichText object at 0x3e81cd47f550>)])]><\/dd>\n<\/dl>\n<\/div>\n
\n

package.json<\/code><\/p>\n<\/div>\n

\n
\n
code_block<\/dt>\n
<ListValue: [StructValue([(‘code’, ‘{rn “dependencies”: {rn “@google-cloud\/functions-framework”: “^3.3.0”,rn “@google-cloud\/firestore”: “^7.6.0″rn }rn}’), (‘language’, ”), (‘caption’, <wagtail.rich_text.RichText object at 0x3e81cd514fa0>)])]><\/dd>\n<\/dl>\n<\/div>\n
\n

The key takeaway is that all payment processing work occurs within a transaction<\/strong>, making all actions idempotent. The code within the transaction might run multiple times due to Workflows retries, but it\u2019s only committed once.<\/p>\n

What about HTTP callbacks, Pub\/Sub, Cloud Tasks?<\/strong><\/h2>\n

So far, we\u2019ve talked about how to make website-to-workflow and Workflows to Cloud Functions requests, exactly once. There are other ways of invoking or resuming Workflows such as HTTP callbacks<\/a>, Pub\/Sub messages or Cloud Tasks. How do you make those requests exactly once? Let\u2019s take a look.<\/p>\n

Callbacks<\/strong><\/h3>\n

The good news is that Workflows HTTP callbacks are fully idempotent by default. It\u2019s safe to retry a callback if it fails. For example:<\/p>\n<\/div>\n

\n
\n
code_block<\/dt>\n
<ListValue: [StructValue([(‘code’, ‘- createCallbackStep:rn call: events.create_callback_endpointrn args:rn http_callback_method: “POST”rn result: callback_detailsrn- sendOutURL:rn call: http.postrn args:rn url: “https:\/\/your-endpoint.com\/foo”rn body:rn callback_to_use: ${callback_details.url}rn…rn- callbackWaitStep:rn call: events.await_callbackrn args:rn callback: ${callback_details}’), (‘language’, ”), (‘caption’, <wagtail.rich_text.RichText object at 0x3e81cd48a100>)])]><\/dd>\n<\/dl>\n<\/div>\n
\n

Let\u2019s assume that the first callback returns an error to the external caller. Based on the error, the caller might not know if the workflow callback was received, and should retry the callback. On the second callback, the caller will receive one of the following HTTP status codes:<\/p>\n