Saga

Use a saga for multi-resource operations you cannot fit in one database transaction (reserve a seat, charge a card, issue a ticket). Each SagaStep has a forward Execute and optional reverse Compensate; on failure the runner compensates executed steps in reverse, or rolls forward with retries after a pivot step. A crash resumes from the persisted position, and a step never re-executes after its progress is persisted. Sagas ride the workflow battery; there is no separate saga config entry.

Quick start

WORKFLOW_ADAPTER=memory
WORKFLOW_NAMESPACE=bookings

main.go

package main

import (
  "context"
  "fmt"

  workflowmemory "github.com/zenta-dev/zever/adapters/workflow/memory"
  "github.com/zenta-dev/zever/config"
  "github.com/zenta-dev/zever/container"
  "github.com/zenta-dev/zever/core/workflow"
)

func main() {
  workflowmemory.Register()
  wf, _ := container.New(config.Default()).Workflow()

  wf.(workflow.SagaRegistrar).RegisterSaga("booking", []workflow.SagaStep{
    {Name: "reserve-seat", Execute: func(context.Context, any) (any, error) { return "seat-7", nil },
      Compensate: func(_ context.Context, result any) (any, error) { fmt.Println("released", result); return nil, nil }},
    {Name: "charge-card", Execute: func(context.Context, any) (any, error) { return "ch-1", nil },
      Compensate: func(_ context.Context, result any) (any, error) { fmt.Println("refunded", result); return nil, nil }},
  })

  ctx := context.Background()
  id, _ := wf.(workflow.SagaRunner).RunSaga(ctx, "booking", map[string]string{"booking": "b-1"}, "b-1")
  st, _ := wf.(workflow.SagaInspector).SagaStatus(ctx, id)
  fmt.Println(id, st.Status)
}

b-1 completed

If charge-card fails, the run prints only released seat-7 (the failed step itself is not compensated) and the status is failed. With Pivot: true on reserve-seat, a later failure rolls forward with retries and never compensates. Type-assert the resolved workflow to workflow.SagaRegistrar, SagaRunner, and SagaInspector; both memory and db implement all three, and engines without saga support are unaffected.

Understand steps, pivots, and replay

• SagaStep{Name, Execute, Compensate, Pivot}: forward action plus optional rollback (nil Compensate means none). Input and output must be JSON-marshalable.
• SagaRegistrar.RegisterSaga(name, steps): registers or replaces the definition. Definitions are in-memory (steps are funcs); only run progress is durable.
• SagaRunner.RunSaga(ctx, name, input, workflowID): starts or resumes. Empty workflowID generates one; a matching ID resumes that run (terminal runs return their ID unchanged, others continue from the persisted current_step).
• SagaInspector.SagaStatus(ctx, runID): returns Status, CurrentStep, FailedStep, and Err.

Statuses: pending, running, completed, compensating, failed, compensation_failed.

Pivot. When a step fails and an earlier executed step has Pivot=true, the saga commits to completing: it retries remaining steps forward with backoff and never compensates. Otherwise it compensates executed steps in reverse, skipping nil Compensate and (on db) steps already compensated before a crash. A failed compensation marks the run compensation_failed and may need manual intervention.

Replay. RunSaga with the same workflowID never re-executes a completed step. On db, the run row’s current_step and the run_id:step_index guard make replay safe once progress is persisted; on memory, the executed index does the same.

Declare a saga in .zen

A saga over existing service RPCs is schema, not wiring:

schema/app.zen

saga CheckoutSaga {
  step reserve {
    execute: InventoryService.Reserve
    compensate: InventoryService.Release
  }
  step charge {
    execute: PaymentService.Charge
    compensate: PaymentService.Refund
    pivot: true
  }
}

execute is required per step; compensate and pivot are optional (pivot defaults to false). Step names are unique within a saga, and at most one step may set pivot: true. Both execute and compensate reference <Service>.<RPC> and resolve against every service in the schema, so a saga may compose RPCs across modules.

For each module with a saga, the gogen backend emits <module>/saga.go with per-step typed input/output structs, a SagaCaller interface, and RegisterSagas:

app/saga.go

package app

import (
  "context"

  "github.com/zenta-dev/zever/core/workflow"
)

// CheckoutSagaReserveInput is the typed input for saga "CheckoutSaga" step "reserve".
type CheckoutSagaReserveInput struct {
  ItemId string `json:"item_id"`
}

// CheckoutSagaReserveOutput is the typed output for saga "CheckoutSaga" step "reserve".
type CheckoutSagaReserveOutput struct {
  Id string `json:"id"`
}

// SagaCaller invokes the saga steps declared in this module; wire it to
// shared/grpcclient or an in-process dispatcher. One method per step keeps
// every call typed: no untyped payloads cross this seam.
type SagaCaller interface {
  CheckoutSagaReserve(ctx context.Context, in CheckoutSagaReserveInput, out *CheckoutSagaReserveOutput) error
  CheckoutSagaReserveCompensate(ctx context.Context, in CheckoutSagaReserveCompensateInput, out *CheckoutSagaReserveCompensateOutput) error
  CheckoutSagaCharge(ctx context.Context, in CheckoutSagaChargeInput, out *CheckoutSagaChargeOutput) error
  CheckoutSagaChargeCompensate(ctx context.Context, in CheckoutSagaChargeCompensateInput, out *CheckoutSagaChargeCompensateOutput) error
}

// RegisterSagas registers every saga declared in this module.
func RegisterSagas(reg workflow.TypedSagaRegistrar, caller SagaCaller) {
  reg.RegisterTypedSaga("CheckoutSaga", []workflow.TypedStep{
    workflow.NewTypedStep("reserve", checkoutSagaReserveExecute(caller), checkoutSagaReserveCompensate(caller), false),
    workflow.NewTypedStep("charge", checkoutSagaChargeExecute(caller), checkoutSagaChargeCompensate(caller), true),
  })
}

// checkoutSagaReserveExecute adapts SagaCaller.CheckoutSagaReserve to a typed workflow step function.

Wire generated sagas at boot

SagaCaller is the one seam the host supplies; RegisterSagas is deliberately not called from RegisterModule, so wiring stays explicit. Implement one method per step with shared/grpcclient (dial, marshal in, invoke <Service>/<Method>, unmarshal into out) or an in-process dispatcher, which needs no gRPC because the generated file depends only on core/workflow. Resolve c.Workflow(), type-assert it to workflow.TypedSagaRegistrar (both memory and db implement it alongside SagaRegistrar), and call app.RegisterSagas(reg, caller) once at boot. Step inputs/outputs convert across step boundaries with encoding/json inside workflow.NewTypedStep, so keep every generated struct JSON-marshalable.

Recover crashed runs

Use db for anything durable: it persists current_step after each successful step in workflow_saga_runs, extends a lease while running, and creates workflow_saga_runs and workflow_saga_compensations when missing. Recovery is explicit and caller-driven, with no background goroutine. Call RecoverStuckSagas through a local interface to resume runs stuck in running or compensating whose lease expired:

recover.go

type sagaRecoverer interface {
  RecoverStuckSagas(ctx context.Context, olderThan time.Duration) (int, error)
}

rec, ok := wf.(sagaRecoverer)
if ok {
  n, err := rec.RecoverStuckSagas(ctx, time.Minute)
  // n runs resumed; runs whose definition is no longer registered are skipped
}

Adapter	Package	Notes
memory	adapters/workflow/memory	Default. In-process saga state, non-durable. A restart loses all runs.
db	adapters/workflow/db	Durable runs, lease-based crash recovery, RecoverStuckSagas.

Switching memory to db needs no call-site change. The db adapter shares the container pool for its DSN (first opener’s max_conns wins); use dedicated_pool: true with a separate database under load. Keys: workflow config reference.

Production checklist

• Make every step idempotent (locks plus idempotency keys for money-moving steps). Replicas on memory each see only their own runs.
• Compensation is best-effort: make it idempotent and retryable. db retries each with bounded backoff and records attempts in workflow_saga_compensations.
• Input and results become persisted state. Keep secrets and PII out, pass references, and never log raw option maps (use config.RedactedServices).
• Workflow.Close() error takes no context. The container joins close errors via errors.Join; panics become *ClosePanicError.

Troubleshooting

ErrUnknownSaga

The name was never registered on this adapter instance (definitions are in-memory). Register the same definitions on every replica at boot, or RecoverStuckSagas skips those runs.

A step re-runs after a crash

The step was not idempotent. Gate side-effecting steps behind lock singletons and idempotency keys, and persist progress before the side effect where the adapter allows it.

Runs stay in running after a crash

Nobody called RecoverStuckSagas. Call it at worker boot with a lease window longer than the worst-case step latency.

compensation_failed status

A compensation itself failed. Inspect workflow_saga_compensations for the attempt count and last error, fix the cause, and re-drive the compensation manually.

SagaStatus returns ErrSagaNotFound

Unknown run ID, or the memory adapter restarted. Use db for anything that must survive a restart.

Related

WorkflowsThe underlying run engine.

OutboxAtomic event publication alongside saga steps.

Locks and idempotencyMake steps idempotent.

ObservabilitySpans per saga step.