Sagas & Compensation

In Ironflow, every recorded execution (durable workflow) is a potential Saga. By using step.compensate(), you can implement the Saga Pattern (specifically orchestration-based sagas) to ensure eventual consistency across distributed systems.

What is a Saga?

A Saga is a design pattern used in distributed systems to manage complex, long-running transactions. Unlike a traditional database transaction that follows the ACID properties (Atomicity, Consistency, Isolation, Durability), a Saga ensures eventual consistency by breaking a single transaction into a sequence of smaller, independent steps.

If one step fails, the Saga executes compensating transactions (undo actions) for all previously completed steps to return the system to its initial state. This is essential when working with distributed services (like Stripe, Twilio, or external databases) that cannot be wrapped in a single database transaction.

Workflows as Sagas

Unlike other platforms that require a separate “Saga” primitive or complex state machines, Ironflow treats Sagas as a first-class capability of its standard workflow engine. A workflow acts like a saga simply by registering undo actions for its side effects.

• Orchestration-based: The workflow function acts as an in-process orchestrator coordinating steps and their undos.
• Unified Identity: There is no difference between a “background job,” a “recorded execution,” and a “Saga” in Ironflow — they are all built on the same engine, and every step is a recorded fact.
• Continuous History Integration: Because Ironflow records every step as a permanent fact in the continuous history, the “undo” path is just as durable and observable as the “happy” path. Compensation steps are recorded facts too.

How It Works

1. After a step succeeds, register a compensation handler for it.
2. If a later step throws a non-retryable error, the SDK runs all registered compensations in reverse order.
3. Each compensation is a durable step — each completed compensation is memoized, so if the run is retried (e.g., after a crash), it won’t re-run compensations that already completed. Make undo actions idempotent: one that was running at the crash can run again.

step 1: charge-payment    ✓  → register refund-payment
step 2: reserve-inventory ✓  → register release-inventory
step 3: ship-order        ✗  ← non-retryable failure

Compensations run in reverse:
  release-inventory  (compensates step 2)
  refund-payment     (compensates step 1)

Compensations only run on terminal (non-retryable) failures. Transient errors that trigger automatic retries do not trigger compensation.

This is decided per attempt: the SDK runs compensations only when the attempt’s failure ends the run. A non-retryable error always does. A retryable error does only on the last attempt of a run whose onExhausted policy is fail, because the engine sends the attempt number, the budget and the policy with every attempt and the SDK can tell that this failure is final. Under the default pause policy the last failed attempt pauses the run instead of ending it, so nothing is compensated until it is resumed and finishes, or is cancelled. The engine cannot ask the SDK process for compensations after the fact. Mark the failure non-retryable at the point you want the undo path to run: NonRetryableError in TypeScript, ironflow.NewNonRetryableError(msg) or ironflow.WrapNonRetryable(err) in Go. See What counts as retryable for the rule.

Cancelling a run mid-saga does not execute compensations — the run terminates cleanly without running saga compensate steps. Compensation runs only on terminal (non-retryable) failure, as described above. Automatic compensation on cancel is not supported; the proposal to add it (#571) was closed as not planned. (Push-mode compensation is structurally impossible regardless — compensation closures live inside the live SDK process, so there is no re-entry point after a cancel signal.)

---

step.compensate(stepName, fn) — Register a compensation

TypeScript

import { ironflow } from "@ironflow/node";
import { NonRetryableError } from "@ironflow/node";

export const processOrder = ironflow.createFunction(
  {
    id: "process-order",
    triggers: [{ event: "order.placed" }],
  },
  async ({ event, step }) => {
    // Step 1: charge the customer
    const payment = await step.run("charge-payment", async () => {
      return await stripe.charges.create({
        amount: event.data.total,
        customerId: event.data.customerId,
      });
    });

    // Register undo for step 1
    step.compensate("charge-payment", async () => {
      await stripe.refunds.create({ charge: payment.id });
    });

    // Step 2: reserve inventory
    const reservation = await step.run("reserve-inventory", async () => {
      return await inventory.reserve(event.data.items);
    });

    // Register undo for step 2
    step.compensate("reserve-inventory", async () => {
      await inventory.release(reservation.id);
    });

    // Step 3: ship the order — if this fails terminally,
    // release-inventory and refund-payment run automatically
    await step.run("ship-order", async () => {
      const result = await shipping.createLabel(event.data.address);
      if (!result.ok) {
        throw new NonRetryableError("Shipping unavailable for this address");
      }
      return result;
    });
  },
);

Go

import "github.com/sahina/ironflow-go/ironflow"

var ProcessOrder = ironflow.CreateFunction(ironflow.FunctionConfig{
    ID:       "process-order",
    Triggers: []ironflow.Trigger{{Event: "order.placed"}},
}, func(ctx ironflow.Context) (any, error) {
    var data struct {
        Total      int    `json:"total"`
        CustomerID string `json:"customerId"`
        Address    string `json:"address"`
        Items      []any  `json:"items"`
    }
    if err := ctx.Event.Data(&data); err != nil {
        return nil, err
    }

    // Step 1: charge the customer
    payment, err := ironflow.Run(ctx, "charge-payment", func() (Payment, error) {
        return stripe.CreateCharge(data.Total, data.CustomerID)
    })
    if err != nil {
        return nil, err
    }

    // Register undo for step 1
    ironflow.Compensate(ctx, "charge-payment", func() error {
        return stripe.CreateRefund(payment.ID)
    })

    // Step 2: reserve inventory
    reservation, err := ironflow.Run(ctx, "reserve-inventory", func() (Reservation, error) {
        return inventory.Reserve(data.Items)
    })
    if err != nil {
        return nil, err
    }

    // Register undo for step 2
    ironflow.Compensate(ctx, "reserve-inventory", func() error {
        return inventory.Release(reservation.ID)
    })

    // Step 3: ship the order — if this fails terminally,
    // release-inventory and refund-payment run automatically
    _, err = ironflow.Run(ctx, "ship-order", func() (ShipResult, error) {
        result, err := shipping.CreateLabel(data.Address)
        if err != nil {
            // Non-retryable: the run stops and the compensations run.
            return ShipResult{}, ironflow.NewError(err.Error(), "SHIP_FAILED", false)
        }
        return result, nil
    })
    return nil, err
})

Python

from ironflow.worker import Context, NonRetryableError, function


@function(id="process-order", triggers=[{"event": "order.placed"}])
async def process_order(ctx: Context):
    data = ctx.event.data

    # Step 1: charge the customer
    payment = await ctx.step.run(
        "charge-payment", lambda: stripe.create_charge(data["total"], data["customerId"])
    )

    # Register undo for step 1
    ctx.step.compensate("charge-payment", lambda: stripe.create_refund(payment["id"]))

    # Step 2: reserve inventory
    reservation = await ctx.step.run("reserve-inventory", lambda: inventory.reserve(data["items"]))

    # Register undo for step 2
    ctx.step.compensate("reserve-inventory", lambda: inventory.release(reservation["id"]))

    # Step 3: ship the order — if this fails terminally,
    # release-inventory and refund-payment run automatically
    def ship():
        try:
            return shipping.create_label(data["address"])
        except ShippingError as err:
            raise NonRetryableError(str(err), code="SHIP_FAILED") from err

    return await ctx.step.run("ship-order", ship)

step.compensate is not async: it only registers the undo.

---

Compensation Ordering

Compensations always run in reverse registration order, regardless of which step failed:

step.compensate("step-a", undoA);  // registered 1st → runs last
step.compensate("step-b", undoB);  // registered 2nd → runs first

ctx.step.compensate("step-a", undo_a)  # registered 1st → runs last
ctx.step.compensate("step-b", undo_b)  # registered 2nd → runs first

ironflow.Compensate(ctx, "step-a", undoA) // registered 1st → runs last
ironflow.Compensate(ctx, "step-b", undoB) // registered 2nd → runs first

This ensures later side effects are undone before earlier ones, consistent with the saga pattern.

---

Compensation Durability

Each compensation is executed as a named durable step with the prefix compensate::

• compensate:charge-payment
• compensate:reserve-inventory

If a run is retried after a crash mid-compensation, already-completed compensations are skipped via memoization. A compensation is recorded only after it finishes. A crash while one runs can run it again, so make every undo action idempotent.

---

Compensation Failures

If a compensation handler throws, Ironflow:

1. Records the failed compensation step in the run’s step history (status: failed)
2. Logs the compensation failure with step ID and error details (TypeScript only — the Go and Python SDKs record the failed step without logging it)
3. Continues running the remaining compensations

A failed compensation does not stop other compensations from running. Monitor failed compensation steps by querying the run’s step history for steps with compensate: prefixed IDs that have a failed status.

---

Using Compensations in Parallel Branches

In TypeScript, step.compensate works the same inside step.parallel branches since each branch receives a scoped step client. In Go, use ironflow.CompensateInBranch:

TypeScript

await step.parallel("provision-resources", [
  async (s) => {
    const db = await s.run("create-db", async () => createDatabase());
    s.compensate("create-db", async () => deleteDatabase(db.id));
    return db;
  },
  async (s) => {
    const cache = await s.run("create-cache", async () => createCache());
    s.compensate("create-cache", async () => deleteCache(cache.id));
    return cache;
  },
]);

Go

ironflow.Parallel(ctx, "provision-resources",
    []func(*ironflow.BranchContext) (any, error){
        func(b *ironflow.BranchContext) (any, error) {
            db, err := ironflow.RunWithBranch(b, "create-db", func() (DB, error) {
                return createDatabase()
            })
            if err != nil {
                return nil, err
            }
            ironflow.CompensateInBranch(b, "create-db", func() error {
                return deleteDatabase(db.ID)
            })
            return db, nil
        },
    },
)

Python

async def create_db(b):
    db = await b.run("create-db", create_database)
    b.compensate("create-db", lambda: delete_database(db["id"]))
    return db

await ctx.step.parallel("provision-resources", [create_db])

---

When to Use Sagas

Scenario	Use Sagas?
Multi-service transaction (payment + inventory + shipping)	Yes
Single database write	No — use a transaction
Idempotent operations with no side effects	No
Operations that can’t be reversed (e.g., sending an email)	Optional — record the fact, alert ops

---

What’s Next?

• Error Handling — NonRetryableError and retries
• Step Primitives — All available step types
• Debugging — Inspect compensation steps in the run detail view

Domain-Driven Design

Ironflow’s saga pattern maps directly to the DDD Saga and Process Manager patterns for maintaining consistency across aggregate boundaries. See Sagas & Process Managers for the full DDD perspective.