Saga: compensating a partial failure

This recipe undoes exactly the work that completed, in reverse order, and nothing else.

The shape

reserve_inventory ──> charge_payment ──> book_shipment
                                              │ fails
                                              ▼
                                     failureWorkflow starts
                                              │
                     read the failed execution ──> refund_payment ──> release_inventory

The main workflow does not contain its own rollback branches. It declares a failureWorkflow, and Conductor starts that workflow when the main one fails after exhausting retries.

Why compensation has to read the failed execution

The naive compensation workflow undoes every step. That is wrong: if reserve_inventory failed, there is no reservation to release and no charge to refund, and blindly calling refund produces a support ticket.

Conductor hands the failure workflow five inputs, and the last one is what makes this tractable:

Input	What it gives you
reason	Why the workflow failed
workflowId	The failed execution's id
failureStatus	Its terminal status
failureTaskId	The id of the task that failed
failedWorkflow	The entire failed execution, including every task and its output

So compensation starts by asking the execution what actually happened:

{
  "name": "determine_what_completed",
  "taskReferenceName": "completed_steps",
  "type": "JSON_JQ_TRANSFORM",
  "inputParameters": {
    "failed": "${workflow.input.failedWorkflow}",
    "queryExpression": "((.failed.tasks // []) | map(select(.status == \"COMPLETED\")) | map(.referenceTaskName)) as $done | {done: $done, undoPayment: ($done | index(\"charge_payment\") != null), undoInventory: ($done | index(\"reserve_inventory\") != null)}"
  }
}

Each undo is then behind a SWITCH on that answer. Nothing gets undone that never happened.

Prerequisites

A running Conductor server. The recipe calls three HTTP endpoints; a stub is included so you can run it without wiring real services.

Save this as saga_stub_service.py and leave it running:

"""A tiny stand-in for the three services the saga calls.

Run it before starting the workflow:

    python3 saga_stub_service.py          # listens on http://localhost:8088

Endpoints
    POST /inventory/reserve   -> 200, returns a reservationId
    POST /payments/charge     -> 200, returns a chargeId
    POST /shipping/book       -> status taken from the ?status= query (default 200)
    POST /payments/refund     -> 200
    POST /inventory/release   -> 200
    GET  /calls               -> every call received, so you can prove what ran
    POST /reset               -> clear the call log

Each write endpoint is idempotent on the Idempotency-Key header: a repeat of a
key it has already seen is acknowledged without doing the work twice.
"""

import json
from http.server import BaseHTTPRequestHandler, HTTPServer

CALLS = []
SEEN_KEYS = {}


class Handler(BaseHTTPRequestHandler):
    def _send(self, code, payload):
        body = json.dumps(payload).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path.startswith("/calls"):
            self._send(200, {"calls": CALLS})
        else:
            self._send(404, {"error": "not found"})

    def do_POST(self):
        path = self.path.split("?")[0]
        length = int(self.headers.get("Content-Length") or 0)
        raw = self.rfile.read(length) if length else b"{}"
        try:
            body = json.loads(raw or b"{}")
        except ValueError:
            body = {"raw": raw.decode(errors="replace")}
        key = self.headers.get("Idempotency-Key")

        if path == "/reset":
            CALLS.clear()
            SEEN_KEYS.clear()
            self._send(200, {"reset": True})
            return

        # Idempotency: same key, same answer, no repeated work.
        if key and key in SEEN_KEYS:
            CALLS.append({"path": path, "key": key, "replayed": True})
            self._send(200, SEEN_KEYS[key])
            return

        if path == "/shipping/book":
            status = 200
            if "status=" in self.path:
                try:
                    status = int(self.path.split("status=")[1].split("&")[0])
                except ValueError:
                    status = 200
            CALLS.append({"path": path, "key": key, "status": status, "body": body})
            if status >= 400:
                self._send(status, {"error": "carrier unavailable"})
                return
            result = {"shipmentId": f"SHP-{len(CALLS)}"}
        elif path == "/inventory/reserve":
            result = {"reservationId": f"RES-{len(CALLS) + 1}"}
            CALLS.append({"path": path, "key": key, "body": body})
        elif path == "/payments/charge":
            result = {"chargeId": f"CHG-{len(CALLS) + 1}"}
            CALLS.append({"path": path, "key": key, "body": body})
        elif path in ("/payments/refund", "/inventory/release"):
            result = {"undone": True, "path": path}
            CALLS.append({"path": path, "key": key, "body": body})
        else:
            self._send(404, {"error": "not found"})
            return

        if key:
            SEEN_KEYS[key] = result
        self._send(200, result)

    def log_message(self, *args):
        pass


if __name__ == "__main__":
    print("saga stub listening on http://localhost:8088")
    HTTPServer(("127.0.0.1", 8088), Handler).serve_forever()

python3 saga_stub_service.py      # http://localhost:8088

It records every call at GET /calls, which is how you prove what the saga did.

The main workflow

Save this as saga-order-fulfillment.json:

{
  "name": "saga_order_fulfillment",
  "description": "Three-step order saga. Each step records what it did so compensation can undo it. On unrecoverable failure Conductor starts the compensation workflow with the full failed execution.",
  "version": 1,
  "schemaVersion": 2,
  "timeoutSeconds": 600,
  "timeoutPolicy": "TIME_OUT_WF",
  "failureWorkflow": "saga_order_compensation",
  "inputParameters": [
    "orderId",
    "amount",
    "shipmentStatus"
  ],
  "tasks": [
    {
      "name": "reserve_inventory",
      "taskReferenceName": "reserve_inventory",
      "type": "HTTP",
      "inputParameters": {
        "http_request": {
          "uri": "http://localhost:8088/inventory/reserve",
          "method": "POST",
          "headers": { "Idempotency-Key": "${workflow.input.orderId}-reserve" },
          "body": { "orderId": "${workflow.input.orderId}", "action": "reserve" },
          "connectionTimeOut": 10000,
          "readTimeOut": 40000
        }
      }
    },
    {
      "name": "charge_payment",
      "taskReferenceName": "charge_payment",
      "type": "HTTP",
      "inputParameters": {
        "http_request": {
          "uri": "http://localhost:8088/payments/charge",
          "method": "POST",
          "headers": { "Idempotency-Key": "${workflow.input.orderId}-charge" },
          "body": { "orderId": "${workflow.input.orderId}", "amount": "${workflow.input.amount}" },
          "connectionTimeOut": 10000,
          "readTimeOut": 40000
        }
      }
    },
    {
      "name": "book_shipment",
      "taskReferenceName": "book_shipment",
      "type": "HTTP",
      "inputParameters": {
        "http_request": {
          "uri": "http://localhost:8088/shipping/book?status=${workflow.input.shipmentStatus}",
          "method": "POST",
          "headers": { "Idempotency-Key": "${workflow.input.orderId}-ship" },
          "body": { "orderId": "${workflow.input.orderId}" },
          "connectionTimeOut": 10000,
          "readTimeOut": 40000
        }
      },
      "taskDefinition": {
        "name": "book_shipment",
        "retryCount": 1,
        "retryLogic": "EXPONENTIAL_BACKOFF",
        "retryDelaySeconds": 2,
        "responseTimeoutSeconds": 30,
        "timeoutSeconds": 60,
        "timeoutPolicy": "TIME_OUT_WF"
      }
    }
  ],
  "outputParameters": {
    "orderId": "${workflow.input.orderId}",
    "reservationId": "${reserve_inventory.output.response.body.reservationId}",
    "chargeId": "${charge_payment.output.response.body.chargeId}"
  }
}

The compensation workflow

Save this as saga-order-compensation.json:

{
  "name": "saga_order_compensation",
  "description": "Compensation workflow for saga_order_fulfillment. Reads the failed execution to learn which steps completed, then undoes only those, in reverse order, idempotently.",
  "version": 1,
  "schemaVersion": 2,
  "timeoutSeconds": 900,
  "timeoutPolicy": "TIME_OUT_WF",
  "inputParameters": [
    "reason",
    "workflowId",
    "failureStatus",
    "failureTaskId",
    "failedWorkflow"
  ],
  "tasks": [
    {
      "name": "determine_what_completed",
      "taskReferenceName": "completed_steps",
      "type": "JSON_JQ_TRANSFORM",
      "inputParameters": {
        "failed": "${workflow.input.failedWorkflow}",
        "queryExpression": "((.failed.tasks // []) | map(select(.status == \"COMPLETED\")) | map(.referenceTaskName)) as $done | {done: $done, orderId: ((.failed.input.orderId) // \"unknown\"), undoPayment: ($done | index(\"charge_payment\") != null), undoInventory: ($done | index(\"reserve_inventory\") != null)}"
      }
    },
    {
      "name": "route_payment_refund",
      "taskReferenceName": "route_refund",
      "type": "SWITCH",
      "evaluatorType": "value-param",
      "expression": "undoPayment",
      "inputParameters": {
        "undoPayment": "${completed_steps.output.result.undoPayment}"
      },
      "decisionCases": {
        "true": [
          {
            "name": "refund_payment",
            "taskReferenceName": "refund_payment",
            "type": "HTTP",
            "inputParameters": {
              "http_request": {
                "uri": "http://localhost:8088/payments/refund",
                "method": "POST",
                "headers": { "Idempotency-Key": "${completed_steps.output.result.orderId}-refund" },
                "body": {
                  "orderId": "${completed_steps.output.result.orderId}",
                  "reason": "${workflow.input.reason}"
                },
                "connectionTimeOut": 10000,
                "readTimeOut": 40000
              }
            },
            "taskDefinition": {
              "name": "refund_payment",
              "retryCount": 5,
              "retryLogic": "EXPONENTIAL_BACKOFF",
              "retryDelaySeconds": 5,
              "responseTimeoutSeconds": 30,
              "timeoutSeconds": 300,
              "timeoutPolicy": "TIME_OUT_WF"
            }
          }
        ]
      },
      "defaultCase": []
    },
    {
      "name": "route_inventory_release",
      "taskReferenceName": "route_release",
      "type": "SWITCH",
      "evaluatorType": "value-param",
      "expression": "undoInventory",
      "inputParameters": {
        "undoInventory": "${completed_steps.output.result.undoInventory}"
      },
      "decisionCases": {
        "true": [
          {
            "name": "release_inventory",
            "taskReferenceName": "release_inventory",
            "type": "HTTP",
            "inputParameters": {
              "http_request": {
                "uri": "http://localhost:8088/inventory/release",
                "method": "POST",
                "headers": { "Idempotency-Key": "${completed_steps.output.result.orderId}-release" },
                "body": { "orderId": "${completed_steps.output.result.orderId}" },
                "connectionTimeOut": 10000,
                "readTimeOut": 40000
              }
            },
            "taskDefinition": {
              "name": "release_inventory",
              "retryCount": 5,
              "retryLogic": "EXPONENTIAL_BACKOFF",
              "retryDelaySeconds": 5,
              "responseTimeoutSeconds": 30,
              "timeoutSeconds": 300,
              "timeoutPolicy": "TIME_OUT_WF"
            }
          }
        ]
      },
      "defaultCase": []
    }
  ],
  "outputParameters": {
    "compensatedOrder": "${completed_steps.output.result.orderId}",
    "stepsCompleted": "${completed_steps.output.result.done}",
    "paymentRefunded": "${completed_steps.output.result.undoPayment}",
    "inventoryReleased": "${completed_steps.output.result.undoInventory}",
    "originalFailure": "${workflow.input.reason}"
  }
}

Register and run

conductor workflow create saga-order-compensation.json
conductor workflow create saga-order-fulfillment.json

Happy path — the carrier accepts the shipment:

conductor workflow start -w saga_order_fulfillment --sync \
  -i '{"orderId":"ORD-1","amount":49.00,"shipmentStatus":"200"}'

Failure path — the carrier is down, after the card has already been charged:

conductor workflow start -w saga_order_fulfillment \
  -i '{"orderId":"ORD-2","amount":49.00,"shipmentStatus":"503"}'

Open Executions in the Conductor UI and select the new execution to review the task graph, and each task's inputs and outputs.

The failed workflow's output carries conductor.failure_workflow — the id of the compensation run. Open it and you will see:

completed_steps      JSON_JQ_TRANSFORM   COMPLETED
route_refund         SWITCH              COMPLETED
refund_payment       HTTP                COMPLETED
route_release        SWITCH              COMPLETED
release_inventory    HTTP                COMPLETED

with output:

{
  "stepsCompleted": ["reserve_inventory", "charge_payment"],
  "paymentRefunded": true,
  "inventoryReleased": true,
  "compensatedOrder": "ORD-2"
}

Ask the stub what it actually received:

curl -s http://localhost:8088/calls

1. /inventory/reserve   key=ORD-2-reserve
2. /payments/charge     key=ORD-2-charge
3. /shipping/book       key=ORD-2-ship
4. /shipping/book       key=ORD-2-ship
5. /shipping/book       key=ORD-2-ship
6. /shipping/book       key=ORD-2-ship
7. /payments/refund     key=ORD-2-refund
8. /inventory/release   key=ORD-2-release

Two things are worth staring at. The undo calls arrive in reverse order — refund before release. And /shipping/book was attempted four times before the workflow gave up, which is the whole argument for the next section.

Production notes

• Every write needs an idempotency key. A failing endpoint gets called repeatedly by task retries. The stub replays the stored answer for a repeated Idempotency-Key instead of doing the work twice; your services must do the same.
• Compensation must be idempotent too. The failure workflow can itself be retried. refund_payment carries ORD-2-refund so a second attempt is a no-op, not a second refund.
• Undo only what completed. Drive each undo from the failed execution's task statuses, never from the assumption that everything ran.
• Give compensation more retries than the forward path. Here the forward shipment call retries once; refund and release retry five times with backoff. Failing to undo is worse than failing to do.
• Compensation is not rollback. A refund is a new transaction with its own ledger entry. Design for "eventually consistent and explainable", not "as if it never happened".
• Alert when compensation fails. A saga that cannot undo needs a human. Give the compensation workflow its own failureWorkflow or a status listener.
• Keep the order id out of generated state. Both workflows derive keys from orderId supplied by the caller, so a restart produces the same keys.

Related

• Handling workflow errors — retry strategies, timeout policies, and status listeners
• Task timeouts and retries — tuning the forward path
• Microservice orchestration — the HTTP chain this builds on