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.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)
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.
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)
}
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)
}
}
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)
},
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 |
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.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 | plug-in | ✓ | PostgreSQL | ✗ |
| Generics (typed state) | ✓ | ✗ | ✗ | ✗ |
| Parallel steps | ✓ | ✓ | ✓ | ✗ |
| Per-step retry + backoff | ✓ | ✓ | ✓ | ✗ |
| Per-step timeout | ✓ | ✓ | ✗ | ✗ |
| 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.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
}
}
Parallel steps ¶
Use Parallel to run independent steps concurrently:
kata.Parallel("notify",
kata.Step("email", sendEmail),
kata.Step("sms", sendSMS).Compensate(cancelSMS),
)
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) { ... },
}))
Index ¶
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.
func (*CompensationError) Error ¶
func (e *CompensationError) Error() string
func (*CompensationError) Unwrap ¶
func (e *CompensationError) Unwrap() error
type CompensationFailure ¶
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)
// 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.
Use flow.Parallel() to create one.
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 Exponential ¶
func Exponential(base time.Duration) RetryPolicy
Exponential doubles the wait time on each attempt, starting from base. e.g. base=100ms -> 100ms, 200ms, 400ms, 800ms...
func Fixed ¶
func Fixed(d time.Duration) RetryPolicy
Fixed waits the same duration between every attempt.
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 := flow.New(
flow.Step("charge", chargeCard).Compensate(refundCard).Retry(3, flow.Exponential(100*time.Millisecond)),
flow.Step("reserve", reserveStock).Compensate(releaseStock),
flow.Parallel("notify",
flow.Step("email", sendEmail),
flow.Step("sms", sendSMS),
),
)
func (*Runner[T]) Run ¶
Run executes all steps in order against the given state.
Returns:
- nil on success
- *StepError if a step failed and all compensations ran successfully
- *CompensationError if a step failed AND some compensations also failed
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 := flow.New(step1, step2).WithOptions(flow.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 flow.Step(), then chain builder methods to configure it.
func Step ¶
Step creates a new step definition with the given name and function.
flow.Step("charge-card", chargeCard).
Compensate(refundCard).
Retry(3, flow.Exponential(100*time.Millisecond)).
Timeout(10*time.Second)
func (*StepDef[T]) Compensate ¶
Compensate sets the rollback function for this step. It is called automatically (in reverse order) if a later step fails.
Source Files
Directories