Docs
Kiwi Fabric
Version Current

How-to guide

Run a distributed transaction

Define a saga that writes to several systems, watch it execute, and see it roll back on failure.

Fabric Current 6 min read Updated

Prerequisites

  • Write-enabled capabilities with compensation declared for each table the saga touches
  • A user or API key with permission to write to those tables

A saga is a JSON definition of ordered writes across any number of registered sources. Kiwi Fabric runs independent steps in parallel, passes values between dependent steps, and, if any step fails, compensates every completed step in reverse order.

Define a saga#

This example inserts a category in SQL Server, then uses the key it generated to create an inventory record in MongoDB, while a review is written in parallel:

json
{
  "steps": [
    {
      "id": "new-category",
      "table": "Categories",
      "operation": "insert",
      "payload": { "CategoryName": "New Beverages", "Description": "Small-batch and imported drinks" }
    },
    {
      "id": "init-inventory",
      "table": "inventory",
      "operation": "insert",
      "dependsOn": ["new-category"],
      "payload": {
        "productId": { "$ref": "new-category.Category_id" },
        "warehouse": "WH-EAST",
        "quantity": 0
      }
    },
    {
      "id": "seed-review",
      "table": "reviews",
      "operation": "insert",
      "payload": { "reviewId": 90101, "productId": 1, "rating": 5, "title": "Great selection" }
    }
  ]
}
Field Meaning
id Unique name of the step, used by dependsOn and $ref
table The registered capability to write to
operation insert, update, delete or rawsql (see the step kinds below)
dependsOn Step ids that must finish first. Steps without dependencies run together in the first wave
payload Column values. { "$ref": "stepId.keyName" } takes a value produced by an earlier step, such as a generated identity

Each step can also carry an idempotency key, which makes a retry after a timeout safe.

Step kinds#

Kind What it does How it is undone
Insert Inserts a row Deletes by the generated identity captured at write time
Upsert Inserts or updates by identity column Restores the pre-upsert row
Update Updates rows matching a filter Restores the affected rows from a snapshot
Delete Deletes matching rows Re-inserts the deleted rows from a snapshot
Raw SQL Runs arbitrary SQL on a named source You must supply compensation SQL
REST call Calls an external HTTP endpoint You must supply a compensation URL
Execute stored procedure Runs a registered write procedure The compensation procedure registered with the capability
Query Runs a federated SELECT and passes the result to later steps Not needed (read-only)

Updates and deletes use a filter tree with eq, neq, gt, gte, lt, lte, like, in, between, isNull, and, or and not. A numeric column can be changed relatively, for example subtracting 5 from a stock count, instead of overwritten.

Run it#

bash
kiwi fabric saga-run --file my-saga.json

The output shows the saga id, success or failure, the duration, a table of steps (table, mode, success, milliseconds, error) and, if applicable, a compensation gap report.

Over HTTP use POST /api/saga for one JSON response, or POST /api/saga/stream for Server-Sent Events (Planned, WaveStarted, StepCompleted, Compensating, CompensationStepCompleted and Result frames):

bash
curl -N -X POST http://localhost:8080/api/saga/stream \
  -H "Content-Type: application/json" -H "Authorization: Bearer <your-api-key>" \
  -d @my-saga.json

In the Web UI, the Saga Runner shows a live step-by-step view: which steps run concurrently, which have finished, and, on failure, the compensation steps firing in reverse order.

The Saga Runner page with a Generate Saga with AI box and a visual saga definition showing an insert step with its step id, table, operation and payload fields
The Saga Runner: describe a saga for AI to draft, or build it step by step.

Switch between the Visual builder and the JSON editor at any time; Execute Saga runs the definition you are viewing.

The JSON view of a saga definition with a steps array, and Parameters, Save as Template, Visual and JSON buttons above it
The same saga definition as JSON.

Check the result later#

bash
kiwi fabric saga-status --id <saga-id>
kiwi fabric saga-gaps   --id <saga-id>

Save a saga as a template#

Recurring transaction patterns can be stored once and run by name with runtime values. In the Web UI, use Save as Template in the Saga Runner, or the Saga Templates page, and mark parameters with {{paramName}} in the saga JSON:

The Saga Templates page with a template name, description, a parameters section and a visual editor for the template's steps
Defining a parameterised saga template.

From the CLI:

bash
kiwi fabric template-add --file place-order.json
kiwi fabric template-run --name place-order --param orderId=42,customerId=7

Build it with the SDK#

The .NET, Java and Node.js SDKs have a fluent saga builder. The Python SDK takes typed request models, or the saga JSON itself. Pick your language:

.NET
using Kiwify.Kiwi.Fabric.Client.Builders;

var request = new SagaRequestBuilder("saga-place-order")
    .Insert("create-order", "sample_orders", new()
    {
        ["customer_id"] = 1, ["product"] = "Laptop", ["quantity"] = 1,
        ["total"] = 999.99, ["status"] = "pending"
    })
    .Insert("queue-fulfilment", "fulfilment", new()
    {
        ["order_id"]  = SagaRef.To("create-order", "id"),
        ["warehouse"] = "WH-EAST"
    }, dependsOn: ["create-order"])
    .Update("deduct-credit", "sample_customers",
        SetClause.RelativeSubtract("credit_limit", 999.99),
        Filter.Eq("id", 1),
        dependsOn: ["create-order"])
    .Build();

var result = await client.ExecuteSagaAsync(request);
Java
import com.kiwify.fabric.client.builder.Filter;
import com.kiwify.fabric.client.builder.SagaBuilder;
import com.kiwify.fabric.client.builder.SetClause;
import com.kiwify.fabric.client.model.saga.SagaRef;
import java.util.List;
import java.util.Map;

var request = new SagaBuilder("saga-place-order")
    .insert("create-order", "sample_orders", Map.of(
        "customer_id", 1, "product", "Laptop", "quantity", 1,
        "total", 999.99, "status", "pending"))
    .insert("queue-fulfilment", "fulfilment", Map.of(
        "order_id", SagaRef.to("create-order", "id"), "warehouse", "WH-EAST"),
        List.of("create-order"))
    .update("deduct-credit", "sample_customers",
        SetClause.relativeSubtract("credit_limit", 999.99),
        Filter.eq("id", 1),
        List.of("create-order"))
    .build();

var result = client.sagas.execute(request);
Node.js
import { SagaBuilder, Filter, SetClause, SagaRef } from "kiwifabric";

const request = new SagaBuilder("saga-place-order")
  .insert("create-order", "sample_orders", {
    customer_id: 1, product: "Laptop", quantity: 1, total: 999.99, status: "pending",
  })
  .insert("queue-fulfilment", "fulfilment", {
    order_id: SagaRef.to("create-order", "id"), warehouse: "WH-EAST",
  }, ["create-order"])
  .update("deduct-credit", "sample_customers",
    SetClause.relativeSubtract("credit_limit", 999.99),
    Filter.eq("id", 1),
    ["create-order"])
  .build();

const result = await client.sagas.execute(request);
Python
from kiwifabric.models.saga import (
    FilterRequest, SagaRequest, SagaStepRequest, SetClauseRequest, saga_ref,
)

request = SagaRequest(
    id="saga-place-order",
    steps=[
        SagaStepRequest(
            id="create-order", table="sample_orders", operation="insert",
            payload={"customer_id": 1, "product": "Laptop", "quantity": 1,
                     "total": 999.99, "status": "pending"},
        ),
        SagaStepRequest(
            id="queue-fulfilment", table="fulfilment", operation="insert",
            depends_on=["create-order"],
            payload={"order_id": saga_ref("create-order", "id"), "warehouse": "WH-EAST"},
        ),
        SagaStepRequest(
            id="deduct-credit", table="sample_customers", operation="update",
            depends_on=["create-order"],
            set_clauses=[SetClauseRequest(type="relative", column="credit_limit",
                                          operation="subtract", delta=999.99)],
            filter=FilterRequest(type="eq", column="id", value=1),
        ),
    ],
)

result = client.sagas.run(request)

SagaRef.to(step, key) (saga_ref in Python) passes a key generated by an earlier step, such as an auto-increment id, into a later one. If any step fails, the completed steps are compensated in reverse order.

Note

Describe the transaction in plain English and let an AI assistant draft the JSON for you. See Generate SQL and sagas with AI.

Expected result

  • On success, every step completes and generated keys flow between steps
  • On failure, completed steps are undone in reverse order