README
¶
kata
In martial arts, a kata is a precise sequence of movements - executed with full commitment, or not at all. If you break the form, you return to the beginning.
kata is an embedded Go library for orchestrating multi-step operations with automatic compensation on failure. No external services, no databases, no brokers - just import and use.
runner := kata.New(
kata.Step("charge-card", chargeCard).Compensate(refundCard).Retry(3, kata.Jitter(kata.Exponential(100*time.Millisecond))),
kata.Step("reserve-stock", reserveStock).Compensate(releaseStock),
kata.Step("create-shipment", createShipment),
)
if err := runner.Run(ctx, &OrderState{CardToken: "tok_123", Amount: 9900}); err != nil {
// all compensations already ran automatically
}
If create-shipment fails, kata automatically calls releaseStock then refundCard - in reverse order, with the full state available.
Why kata?
Every non-trivial service has operations that span multiple steps: charge a card, reserve inventory, create a shipment. When step 3 fails, you need to undo steps 1 and 2. Most teams write this rollback logic by hand - scattered defer calls, nested if err != nil blocks, easy to get wrong.
The alternatives are either too heavy (Temporal, Cadence require a dedicated server cluster) or too primitive (existing Go saga libraries have no generics, no retry, no parallel execution).
kata sits in the middle: zero dependencies, idiomatic Go, production-ready features.
Installation
go get github.com/kerlenton/kata
Requires Go 1.22+.
Core concepts
Steps
A Step is a named operation that reads from and writes to your shared state. Each step can optionally define a compensation (rollback) function.
kata.Step("charge-card", func(ctx context.Context, s *OrderState) error {
id, err := stripe.Charge(s.CardToken, s.Amount)
if err != nil {
return err
}
s.ChargeID = id // store result for later steps (and compensation)
return nil
}).Compensate(func(ctx context.Context, s *OrderState) error {
return stripe.Refund(s.ChargeID)
})
Retry
Steps can be retried with configurable backoff:
kata.Step("call-flaky-api", callAPI).
Retry(3, kata.Exponential(100*time.Millisecond))
// attempts: immediate -> 100ms -> 200ms -> 400ms
kata.Step("call-another", callOther).
Retry(5, kata.Fixed(1*time.Second))
kata.Step("call-fast", callFast).
Retry(2, kata.NoDelay)
Retry policies are composable - wrap them to add jitter or cap the delay:
// Add ±25% random jitter to prevent thundering herd
kata.Jitter(kata.Exponential(100*time.Millisecond))
// Cap maximum delay at 30 seconds
kata.Cap(kata.Exponential(100*time.Millisecond), 30*time.Second)
// Combine both
kata.Cap(kata.Jitter(kata.Exponential(100*time.Millisecond)), 30*time.Second)
Exponential has a built-in ceiling of 5 minutes to prevent overflow at high retry counts.
Timeout
kata.Step("slow-step", doWork).
Timeout(5 * time.Second)
If the step exceeds the timeout, the context is cancelled and the step fails with context.DeadlineExceeded. Compensations are triggered normally.
Parallel steps
Run multiple steps concurrently within a group. If any step in the group fails, the others are cancelled and the successful ones are compensated.
kata.Parallel("notify-customer",
kata.Step("send-email", sendEmail),
kata.Step("send-sms", sendSMS).Compensate(cancelSMS),
kata.Step("send-push", sendPush),
)
If a later sequential step fails after the parallel group succeeds, all steps in the group are compensated in reverse order.
Parallel groups can be nested - useful when you have logically distinct groups that should run concurrently with each other:
kata.Parallel("all-notifications",
kata.Parallel("customer",
kata.Step("email", sendEmail),
kata.Step("sms", sendSMS),
),
kata.Parallel("internal",
kata.Step("slack", notifySlack),
kata.Step("analytics", trackEvent),
),
)
Thread safety: all steps in a parallel group share state T concurrently. Either assign disjoint fields to each step, or protect shared fields with a sync.Mutex in your state struct.
Runner
New creates a reusable runner - define it once, call Run per request:
// define once (e.g. at startup or in a constructor)
var orderRunner = kata.New(
kata.Step("charge", chargeCard).Compensate(refundCard),
kata.Step("reserve", reserveStock).Compensate(releaseStock),
kata.Parallel("notify",
kata.Step("email", sendEmail),
kata.Step("sms", sendSMS),
),
)
// call per request
func (s *OrderService) PlaceOrder(ctx context.Context, req *PlaceOrderRequest) error {
state := &OrderState{CardToken: req.CardToken, ItemID: req.ItemID}
return orderRunner.Run(ctx, state)
}
The runner checks the context between steps - if the context is cancelled (e.g. SIGTERM, request timeout), it stops immediately and compensates all completed steps. Compensation always runs with context.Background() to guarantee rollback completes regardless of the caller's context.
Error handling
kata distinguishes between two failure modes:
err := runner.Run(ctx, state)
var stepErr *kata.StepError
var compErr *kata.CompensationError
switch {
case err == nil:
// all steps succeeded
case errors.As(err, &stepErr):
// a step failed, all compensations ran successfully
// stepErr.StepName - which step failed
// stepErr.Cause - the original error
log.Printf("rolled back cleanly after %q: %v", stepErr.StepName, stepErr.Cause)
case errors.As(err, &compErr):
// a step failed AND one or more compensations also failed
// the system may be in a partially inconsistent state
// manual intervention may be required
log.Printf("ALERT: step %q failed, compensations also failed:", compErr.StepName)
for _, f := range compErr.Failed {
log.Printf(" - %q: %v", f.StepName, f.Err)
}
}
Both error types implement Unwrap(), so errors.Is works against the original cause.
Observability
Attach hooks for logging, metrics, or tracing - no changes to step code required:
runner := kata.New(steps...).WithOptions(
kata.WithHooks(kata.Hooks{
OnStepStart: func(ctx context.Context, name string) {
metrics.Inc("kata.step.started", name)
},
OnStepDone: func(ctx context.Context, name string, d time.Duration) {
metrics.Histogram("kata.step.duration", d, name)
},
OnStepFailed: func(ctx context.Context, name string, err error) {
log.Errorf("step %q failed: %v", name, err)
},
OnRetry: func(ctx context.Context, name string, attempt int, err error) {
log.Warnf("retrying %q (attempt %d): %v", name, attempt, err)
},
OnCompensationStart: func(ctx context.Context, name string) {
log.Warnf("compensating %q", name)
},
OnCompensationFailed: func(ctx context.Context, name string, err error) {
alerts.Fire("compensation_failed", name, err)
},
}),
)
Available hooks:
| Hook | When |
|---|---|
OnStepStart |
Before a step begins |
OnStepDone |
After a step succeeds |
OnStepFailed |
After a step exhausts all retries and fails |
OnRetry |
Before each retry attempt (with attempt number and previous error) |
OnCompensationStart |
Before a compensation begins |
OnCompensationDone |
After a compensation succeeds |
OnCompensationFailed |
After a compensation fails |
Full example
type OrderState struct {
// inputs
CardToken string
ItemID string
UserEmail string
Amount int64
// filled in by steps
ChargeID string
ReservationID string
}
var orderRunner = kata.New(
kata.Step("charge-card", func(ctx context.Context, s *OrderState) error {
id, err := payments.Charge(ctx, s.CardToken, s.Amount)
s.ChargeID = id
return err
}).Compensate(func(ctx context.Context, s *OrderState) error {
return payments.Refund(ctx, s.ChargeID)
}).Retry(3, kata.Jitter(kata.Exponential(100*time.Millisecond))).Timeout(10*time.Second),
kata.Step("reserve-stock", func(ctx context.Context, s *OrderState) error {
id, err := warehouse.Reserve(ctx, s.ItemID)
s.ReservationID = id
return err
}).Compensate(func(ctx context.Context, s *OrderState) error {
return warehouse.Release(ctx, s.ReservationID)
}),
kata.Step("create-shipment", func(ctx context.Context, s *OrderState) error {
return shipping.Create(ctx, s.ReservationID)
}),
kata.Parallel("notify",
kata.Step("email", func(ctx context.Context, s *OrderState) error {
return mailer.Send(ctx, s.UserEmail, "Your order is confirmed!")
}),
kata.Step("analytics", func(ctx context.Context, s *OrderState) error {
return analytics.Track(ctx, "order_placed", s.ItemID)
}),
),
)
func PlaceOrder(ctx context.Context, req *Request) error {
state := &OrderState{
CardToken: req.CardToken,
ItemID: req.ItemID,
UserEmail: req.UserEmail,
Amount: req.Amount,
}
err := orderRunner.Run(ctx, state)
if err != nil {
var compErr *kata.CompensationError
if errors.As(err, &compErr) {
// compensation failed - alert on-call
pagerduty.Fire(compErr)
}
return err
}
return nil
}
Comparison
| kata | Temporal/Cadence | floxy | go-saga | |
|---|---|---|---|---|
| External service required | ✗ | ✓ (server cluster) | ✗ | ✗ |
| Persistent state | ✗ (in-memory) | ✓ | PostgreSQL | ✗ |
| Generics (typed state) | ✓ | ✗ | ✗ | ✗ |
| Parallel steps | ✓ | ✓ | ✓ | ✗ |
| Nested parallel groups | ✓ | ✓ | ✗ | ✗ |
| Per-step retry + backoff | ✓ | ✓ | ✓ | ✗ |
| Per-step timeout | ✓ | ✓ | ✗ | ✗ |
| Composable retry policies | ✓ | ✗ | ✗ | ✗ |
| Observability hooks | ✓ | ✓ | ✗ | ✗ |
| Zero dependencies | ✓ | ✗ | ✗ | ✓ |
License
MIT
Documentation
¶
Overview ¶
Package kata provides an embedded saga orchestrator for Go.
kata executes a sequence of named steps against shared state, and automatically compensates (rolls back) completed steps in reverse order when any step fails. No external services or databases required.
Basic usage ¶
type OrderState struct {
CardToken string
ChargeID string
}
runner := kata.New(
kata.Step("charge", chargeCard).
Compensate(refundCard).
Retry(3, kata.Jitter(kata.Exponential(100*time.Millisecond))),
kata.Step("reserve", reserveStock).
Compensate(releaseStock),
kata.Step("ship", createShipment),
)
if err := runner.Run(ctx, &OrderState{CardToken: "tok_123"}); err != nil {
var stepErr *kata.StepError
var compErr *kata.CompensationError
switch {
case errors.As(err, &compErr):
// step failed AND some compensations failed - needs manual fix
case errors.As(err, &stepErr):
// step failed, all compensations ran cleanly
}
}
Retry policies ¶
Retry policies are composable via wrappers:
kata.Exponential(100*time.Millisecond) // 100ms, 200ms, 400ms, ... kata.Jitter(kata.Exponential(100*time.Millisecond)) // same with ±25% randomness kata.Cap(kata.Exponential(100*time.Millisecond), 30*time.Second) // capped at 30s kata.Cap(kata.Jitter(kata.Exponential(100*time.Millisecond)), 30*time.Second) // both
Parallel steps ¶
Use Parallel to run independent steps concurrently:
kata.Parallel("notify",
kata.Step("email", sendEmail),
kata.Step("sms", sendSMS).Compensate(cancelSMS),
)
All steps in a parallel group share state T concurrently. See ParallelDef for thread safety guidance.
Observability ¶
Attach hooks for logging and metrics without changing step code:
runner.WithOptions(kata.WithHooks(kata.Hooks{
OnStepStart: func(ctx context.Context, name string) { ... },
OnStepFailed: func(ctx context.Context, name string, err error) { ... },
OnRetry: func(ctx context.Context, name string, attempt int, err error) { ... },
}))
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type CompensationError ¶
type CompensationError struct {
// StepName is the step that originally failed and triggered compensation.
StepName string
// StepCause is the original error that triggered compensation.
StepCause error
// Failed contains the names and errors of compensations that failed.
Failed []CompensationFailure
}
CompensationError is returned when a step fails AND one or more compensations also fail. The saga is in a partially inconsistent state - requires manual intervention.
Example ¶
package main
import (
"context"
"errors"
"fmt"
"github.com/kerlenton/kata"
)
func main() {
type state struct{}
runner := kata.New(
kata.Step("step-1", func(_ context.Context, _ *state) error {
return nil
}).Compensate(func(_ context.Context, _ *state) error {
return fmt.Errorf("compensate failed")
}),
kata.Step("step-2", func(_ context.Context, _ *state) error {
return fmt.Errorf("boom")
}),
)
err := runner.Run(context.Background(), &state{})
var compErr *kata.CompensationError
if errors.As(err, &compErr) {
fmt.Println("trigger:", compErr.StepName)
for _, f := range compErr.Failed {
fmt.Printf("compensation %q failed: %v\n", f.StepName, f.Err)
}
}
}
Output: trigger: step-2 compensation "step-1" failed: compensate failed
func (*CompensationError) Error ¶
func (e *CompensationError) Error() string
func (*CompensationError) Unwrap ¶
func (e *CompensationError) Unwrap() error
type CompensationFailure ¶
type CompensationFailure struct {
// StepName is the name of the step whose compensation failed.
StepName string
// Err is the error returned by the compensation function.
Err error
}
CompensationFailure holds the name and error for a single failed compensation.
type Hooks ¶
type Hooks struct {
// OnStepStart is called before each step begins executing.
OnStepStart func(ctx context.Context, name string)
// OnStepDone is called after a step completes successfully.
OnStepDone func(ctx context.Context, name string, duration time.Duration)
// OnStepFailed is called when a step fails (after all retries are exhausted).
OnStepFailed func(ctx context.Context, name string, err error)
// OnRetry is called before each retry attempt.
// attempt starts at 1 (first retry), err is the error from the previous attempt.
OnRetry func(ctx context.Context, name string, attempt int, err error)
// OnCompensationStart is called before a compensation function begins.
OnCompensationStart func(ctx context.Context, name string)
// OnCompensationDone is called after a compensation completes successfully.
OnCompensationDone func(ctx context.Context, name string)
// OnCompensationFailed is called when a compensation function fails.
OnCompensationFailed func(ctx context.Context, name string, err error)
}
Hooks provides lifecycle callbacks for observability (logging, metrics, tracing).
type ParallelDef ¶
type ParallelDef[T any] struct { // contains filtered or unexported fields }
ParallelDef runs a group of steps concurrently.
Behaviour:
- All steps start at the same time and share the same state.
- If any step fails, the group cancels all remaining steps and compensates the ones that already succeeded (in reverse order).
- If the whole group succeeds and a later sequential step fails, all steps in the group are compensated in reverse order.
Thread Safety ¶
All steps in a parallel group run concurrently and share the same state T. The caller is responsible for synchronizing concurrent access to shared fields. Common approaches:
- Use a sync.Mutex in your state struct for fields written by parallel steps.
- Assign disjoint fields to each step (e.g. step "email" writes EmailSentAt, step "sms" writes SmsSentAt) so no synchronization is needed.
Nesting ¶
Parallel groups can contain other parallel groups:
kata.Parallel("all-notifications",
kata.Parallel("customer",
kata.Step("email", sendEmail),
kata.Step("sms", sendSMS),
),
kata.Parallel("internal",
kata.Step("slack", notifySlack),
kata.Step("analytics", trackEvent),
),
)
Use kata.Parallel() to create one.
func Parallel ¶
func Parallel[T any](name string, steps ...stepper[T]) *ParallelDef[T]
Parallel creates a group of steps that execute concurrently.
Accepts both Step and nested Parallel groups. All steps share state T concurrently - see ParallelDef for thread safety notes.
kata.Parallel("notifications",
kata.Step("email", sendEmail),
kata.Step("sms", sendSMS).Compensate(cancelSMS),
kata.Step("push", sendPush),
)
Example ¶
package main
import (
"context"
"fmt"
"sort"
"sync"
"github.com/kerlenton/kata"
)
func main() {
type state struct {
mu sync.Mutex
Log []string
}
runner := kata.New(
kata.Parallel("notify",
kata.Step("email", func(_ context.Context, s *state) error {
s.mu.Lock()
s.Log = append(s.Log, "email sent")
s.mu.Unlock()
return nil
}),
kata.Step("sms", func(_ context.Context, s *state) error {
s.mu.Lock()
s.Log = append(s.Log, "sms sent")
s.mu.Unlock()
return nil
}),
),
)
s := &state{}
err := runner.Run(context.Background(), s)
fmt.Println("err:", err)
sort.Strings(s.Log)
for _, entry := range s.Log {
fmt.Println(entry)
}
}
Output: err: <nil> email sent sms sent
type RetryPolicy ¶
RetryPolicy determines the wait duration between retry attempts.
var NoDelay RetryPolicy = func(_ int) time.Duration {
return 0
}
NoDelay retries immediately with no wait between attempts.
func Cap ¶ added in v1.0.0
func Cap(policy RetryPolicy, max time.Duration) RetryPolicy
Cap wraps a retry policy capping the delay at max.
kata.Cap(kata.Exponential(100*time.Millisecond), 30*time.Second)
func Exponential ¶
func Exponential(base time.Duration) RetryPolicy
Exponential doubles the wait time on each attempt, starting from base. The result is capped at 5 minutes to prevent overflow.
Exponential(100*time.Millisecond) // attempt 0: 100ms, 1: 200ms, 2: 400ms, 3: 800ms, ...
func Fixed ¶
func Fixed(d time.Duration) RetryPolicy
Fixed waits the same duration between every attempt.
func Jitter ¶ added in v1.0.0
func Jitter(policy RetryPolicy) RetryPolicy
Jitter wraps a retry policy adding ±25% random jitter to each delay. This prevents thundering-herd effects when many callers retry simultaneously.
kata.Jitter(kata.Exponential(100*time.Millisecond))
type Runner ¶
type Runner[T any] struct { // contains filtered or unexported fields }
Runner orchestrates a sequence of steps with automatic compensation on failure. It is safe to reuse a Runner across multiple Run calls (e.g. per-request).
func New ¶
New creates a reusable Runner from a sequence of steps and parallel groups.
Steps execute in order. On failure, completed steps are compensated in reverse.
runner := kata.New(
kata.Step("charge", chargeCard).Compensate(refundCard).Retry(3, kata.Exponential(100*time.Millisecond)),
kata.Step("reserve", reserveStock).Compensate(releaseStock),
kata.Parallel("notify",
kata.Step("email", sendEmail),
kata.Step("sms", sendSMS),
),
)
func (*Runner[T]) Run ¶
Run executes all steps in order against the given state.
If the context is cancelled between steps, the runner stops and compensates all completed steps. Compensation always runs with context.Background() to guarantee rollback completes even after cancellation (e.g. SIGTERM).
Returns:
- nil on success
- *StepError if a step failed and all compensations ran successfully
- *CompensationError if a step failed AND some compensations also failed
Example ¶
package main
import (
"context"
"errors"
"fmt"
"github.com/kerlenton/kata"
)
func main() {
type state struct {
Log []string
}
runner := kata.New(
kata.Step("step-1", func(_ context.Context, s *state) error {
s.Log = append(s.Log, "step-1: done")
return nil
}).Compensate(func(_ context.Context, s *state) error {
s.Log = append(s.Log, "step-1: compensated")
return nil
}),
kata.Step("step-2", func(_ context.Context, s *state) error {
return fmt.Errorf("boom")
}),
)
s := &state{}
err := runner.Run(context.Background(), s)
var stepErr *kata.StepError
if errors.As(err, &stepErr) {
fmt.Println("failed step:", stepErr.StepName)
}
for _, entry := range s.Log {
fmt.Println(entry)
}
}
Output: failed step: step-2 step-1: done step-1: compensated
func (*Runner[T]) WithOptions ¶
func (r *Runner[T]) WithOptions(opts ...RunnerOption) *Runner[T]
WithOptions returns a new Runner with the given options applied. Useful when you want to add hooks without changing the step definitions.
runner := kata.New(step1, step2).WithOptions(kata.WithHooks(myHooks))
type RunnerOption ¶
type RunnerOption func(*runnerConfig)
RunnerOption is a functional option for configuring a Runner.
func WithHooks ¶
func WithHooks(h Hooks) RunnerOption
WithHooks is an option to set observability hooks on a Runner.
type StepDef ¶
type StepDef[T any] struct { // contains filtered or unexported fields }
StepDef defines a single sequential step with its configuration. Create one with kata.Step(), then chain builder methods to configure it.
func Step ¶
Step creates a new step definition with the given name and function.
kata.Step("charge-card", chargeCard).
Compensate(refundCard).
Retry(3, kata.Exponential(100*time.Millisecond)).
Timeout(10*time.Second)
Example ¶
package main
import (
"context"
"fmt"
"github.com/kerlenton/kata"
)
func main() {
type state struct{ Charged bool }
runner := kata.New(
kata.Step("charge", func(_ context.Context, s *state) error {
s.Charged = true
return nil
}).Compensate(func(_ context.Context, s *state) error {
s.Charged = false
return nil
}),
)
s := &state{}
err := runner.Run(context.Background(), s)
fmt.Println("err:", err)
fmt.Println("charged:", s.Charged)
}
Output: err: <nil> charged: true
Example (Retry) ¶
package main
import (
"context"
"fmt"
"sync/atomic"
"github.com/kerlenton/kata"
)
func main() {
type state struct{}
var attempts atomic.Int32
runner := kata.New(
kata.Step("flaky", func(_ context.Context, _ *state) error {
n := attempts.Add(1)
if n < 3 {
return fmt.Errorf("transient error")
}
return nil
}).Retry(3, kata.NoDelay),
)
err := runner.Run(context.Background(), &state{})
fmt.Println("err:", err)
fmt.Println("attempts:", attempts.Load())
}
Output: err: <nil> attempts: 3
func (*StepDef[T]) Compensate ¶
Compensate sets the rollback function for this step. It is called automatically (in reverse order) if a later step fails.
type StepError ¶
type StepError struct {
// StepName is the name of the step that failed.
StepName string
// Cause is the original error returned by the step function.
Cause error
}
StepError is returned when a step fails and all compensations ran successfully. This means the saga was rolled back cleanly.
Source Files
Directories