# go-saga-orchestration > A standalone, solution-agnostic saga orchestrator and synchronous CEL rule evaluator. Embed it as a Go library or run it as two service binaries. --- # Introduction Source: https://bugs5382.github.io/go-saga-orchestration/ # go-saga-orchestration A standalone, solution-agnostic **saga orchestrator + synchronous CEL rule evaluator** you can embed as a Go library or run as a two-binary service. - **31 saga step types** — transforms, HTTP/webhooks, timers, signals, events, parallel fan-out, foreach, loops, try/catch, human tasks, sub-sagas, and more. - **Embed or deploy** — run in-process with zero infrastructure, or deploy two Docker-friendly binaries backed by Postgres + RabbitMQ. - **CEL expressions** for conditions, transforms, filters, and routing. - **Scheduled & event-driven starts**, durable timers, and a license-gated feature model. ## Where to go next - **[Getting started](getting-started)** — build a working saga from empty. - **[Architecture](architecture)** — engine internals, coordinator, MQ topology, stores. - **[Deployment](deployment)** — container images and the Helm chart. - **[Verbs reference](verbs)** — every step type. --- # Getting started Source: https://bugs5382.github.io/go-saga-orchestration/docs/getting-started # Getting started This tutorial builds a realistic multi-step saga from empty, in embedded mode (no infrastructure). ## 1. Install ```bash go get github.com/Bugs5382/go-saga-orchestration ``` ## 2. A minimal saga ```go package main import ( "context" "fmt" "github.com/Bugs5382/go-saga-orchestration/saga" "github.com/Bugs5382/go-saga-orchestration/domain" "github.com/Bugs5382/go-saga-orchestration/engine/verbs" ) func main() { sc := saga.InMemory() // in-memory store + in-process advance sc.RegisterVerb("charge_card", "common", verbs.HandlerFunc(func(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) { return map[string]any{"ok": true}, nil })) sc.Register(domain.WorkflowDefinition{ ID: "checkout", Version: 1, Start: "charge", Published: true, Steps: []domain.Step{ {ID: "charge", Type: "charge_card", Next: "done"}, {ID: "done", Type: domain.StepTypeEnd}, }, }) runID, _ := sc.Start(context.Background(), "checkout", map[string]any{"total": 4200}) run, _ := sc.Get(context.Background(), runID) fmt.Println(run.State) // succeeded } ``` ## 3. Recovering from errors with `try_catch` When a step returns an error, the run normally transitions to `failed`. To recover instead, wrap the risky steps in a `try_catch` frame: if any step inside the protected region errors, the saga jumps to your `catch` step rather than failing, and the error is written to `Variables._error`. ```go sc := saga.InMemory() // A step that fails. sc.RegisterVerb("charge_card", "common", verbs.HandlerFunc(func(_ context.Context, _ domain.SagaRun, _ domain.Step) (map[string]any, error) { return nil, fmt.Errorf("gateway declined") })) // The catch handler — it can read the error context off Variables._error. sc.RegisterVerb("notify_ops", "common", verbs.HandlerFunc(func(_ context.Context, run domain.SagaRun, _ domain.Step) (map[string]any, error) { errInfo, _ := run.Variables["_error"].(map[string]any) return map[string]any{"recovered": true, "failed_step": errInfo["step_id"]}, nil })) sc.Register(domain.WorkflowDefinition{ ID: "checkout", Version: 1, Start: "protect", Published: true, Steps: []domain.Step{ // The frame: protect "charge", jump to "recover" on any error. {ID: "protect", Type: "try_catch", Inputs: map[string]any{"try": []string{"charge"}, "catch": "recover"}, Next: "charge"}, {ID: "charge", Type: "charge_card", Next: "done"}, {ID: "recover", Type: "notify_ops", Next: "done"}, {ID: "done", Type: domain.StepTypeEnd}, }, }) runID, _ := sc.Start(context.Background(), "checkout", nil) run, _ := sc.Get(context.Background(), runID) fmt.Println(run.State) // succeeded — the error was caught fmt.Println(run.Variables["recovered"]) // true ``` The `try_catch` step's `Next` points at the **first step inside** the protected region; the protected step's `Next` points past it. See the [`try_catch` verb](verbs#try_catch) for the nesting rules, and [Testing](testing) for asserting both the caught and uncaught (`failed`) paths. ### Step-level retry and compensation `try_catch` is one of three recovery mechanisms, and they compose: - **`Step.Retry`** — set a `RetryPolicy` on a step and the engine re-runs it on error, up to `MaxAttempts`, with exponential backoff (`InitialBackoffMS`, `MaxBackoffMS`, `Multiplier`, optional `Jitter`). Retries are exhausted before a `try_catch` frame catches the error or the run fails. - **`Step.Compensation`** — give a completed step a `Compensation` action and, when the run fails with no catching `try_catch` frame, the engine rolls back: it transitions the run to `RunStateCompensating`, then dispatches each already-completed compensable step's compensation action in **reverse order** before the run settles to `RunStateFailed`. A completed step with no `Compensation` is skipped. --- # Embedding Guide Source: https://bugs5382.github.io/go-saga-orchestration/docs/embedding # 🧩 Embedding Guide This guide walks you through adding `go-saga-orchestration` as an in-process library to your Go service — from a thirty-second hello world all the way to production wiring. --- ## 🚀 Quickstart The fastest path is `saga.InMemory()`: an in-process engine backed by a thread-safe in-memory store. No database, no message broker, no external processes required. ```go import ( "context" "fmt" "github.com/Bugs5382/go-saga-orchestration/saga" "github.com/Bugs5382/go-saga-orchestration/domain" ) sc := saga.InMemory() sc.Register(domain.WorkflowDefinition{ ID: "hello", Version: 1, Start: "greet", Published: true, Steps: []domain.Step{ {ID: "greet", Type: "noop", Next: "done"}, {ID: "done", Type: domain.StepTypeEnd}, }, }) runID, err := sc.Start(context.Background(), "hello", map[string]any{"name": "world"}) if err != nil { panic(err) } run, _ := sc.Get(context.Background(), runID) fmt.Println(run.State) // succeeded ``` `sc.Start` creates the run **and** advances it synchronously to the first pause or terminal state, so for an all-synchronous workflow the run is already complete by the time `Start` returns. > 💡 See [`examples/basic`](https://github.com/Bugs5382/go-saga-orchestration/tree/main/examples/basic) for a runnable standalone example. --- ## 🧩 Custom verbs Register your own step type with a closure. The return value is a `map[string]any` that gets **merged into** `run.Variables`. ```go import "github.com/Bugs5382/go-saga-orchestration/engine/verbs" sc.RegisterVerb( "charge_card", // step type name used in workflow JSON/Go "common", // license group — "common" means no gate verbs.HandlerFunc(func(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) { total, _ := run.Variables["total"].(float64) if total <= 0 { return nil, fmt.Errorf("charge_card: invalid total") } // ... call your payment service ... return map[string]any{"charge_id": "ch_abc123", "charged": total}, nil }), ) ``` The keys returned (`charge_id`, `charged`) land directly in `Variables` and are visible to every subsequent step. Use `set_var` or `transform` steps after the custom verb to rename or reshape them if needed. --- ## 🔌 Custom actions (worker round-trip) For steps that need to run in a **separate process** (e.g. a microservice that owns its own business logic), use `type: "action"` in your workflow definition: ```json { "id": "charge", "type": "action", "action": "payments.charge_card", "inputs": {"total": 4200}, "next": "confirm" } ``` The engine dispatches `payments.charge_card` over the configured publisher (RabbitMQ in service mode, in-process in embedded mode). A worker process built with the [Go worker SDK](https://github.com/Bugs5382/go-saga-orchestration/tree/main/clients/go/worker) connects over the gRPC `ExecuteStep` stream, registers a handler for `payments.charge_card`, and returns a result map that is merged into `Variables`. > ⚠️ Pure `saga.InMemory()` with no worker goroutine will leave an `action` step paused indefinitely — the action verb pauses the saga and waits for a worker reply. You need either a worker process (service mode) or a registered `RegisterVerb` handler with the same step type to handle it in-process. See [`docs/grpc.md`](grpc.md) for the worker protocol and [`clients/go/worker`](https://github.com/Bugs5382/go-saga-orchestration/tree/main/clients/go/worker) for the SDK. --- ## 🪢 Data flow between steps Each step operates on exactly one verb, and all data flows through `run.Variables`. CEL verbs (e.g. `transform`, `filter`, `switch`) read from `Variables` — **not** from `step.Inputs` directly. **Pattern:** use `set_var` to seed a variable from a literal or a previous step's output, then reference it in downstream CEL expressions. ``` start → set_var (out_var: "items", value: [...]) → filter → transform → end ``` **Two worked scenarios in [`examples/workflows/`](https://github.com/Bugs5382/go-saga-orchestration/tree/main/examples/workflows):** - [`scenario_action_to_setvar.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/scenario_action_to_setvar.json) — an `action` step returns a result, then a `set_var` step reads the worker's output key and assigns it to a clean variable name for downstream steps. - [`scenario_parallel_setvars.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/scenario_parallel_setvars.json) — parallel branches each write to distinct variables, which are available after the join. > 💡 Prefer descriptive `out_var` names. The `http_request` and `webhook_emit` verbs default to `http_result` / `webhook_result` — override `out_var` to avoid collisions when you call multiple endpoints in the same workflow. --- ## 🚪 Entry points / call tree Every `WorkflowDefinition` has a default entry point (`Start` field). You can define **named entry points** with the `Entrypoints` map: ```go domain.WorkflowDefinition{ ID: "order", Version: 1, Start: "charge", Entrypoints: map[string]string{ "refund": "start_refund", "cancel": "start_cancel", }, Steps: []domain.Step{ /* ... */ }, } ``` Start a run at a named entry point with `StartAt`: ```go runID, err := sc.StartAt(ctx, "order", "refund", map[string]any{"order_id": "ord_99"}) ``` `sub_saga` and `spawn_saga` steps also accept an `entrypoint` input so a parent workflow can invoke a specific slice of a child workflow without a separate definition. REST triggers (service mode) accept an `entrypoint` field in the trigger configuration — see [`docs/api.md`](api.md). --- ## 🏭 Production wiring Replace `InMemory()` with `saga.New(opts)` and provide your own store and infrastructure: ```go import ( "github.com/Bugs5382/go-saga-orchestration/saga" "github.com/Bugs5382/go-saga-orchestration/store/postgres" ) pgStore, err := postgres.New(ctx, databaseDSN) if err != nil { log.Fatal(err) } sc, err := saga.New(saga.Options{ Store: pgStore, // durable Postgres store (see store/postgres) Licensing: myLicenseResolver, // licensing.Resolver — controls feature groups Secrets: mySecretsResolver, // secrets.Resolver — for http_request/webhook_emit Publisher: rabbitPublisher, // engine.Publisher — RabbitMQ-backed Logger: &logger, // *zerolog.Logger Context: appCtx, // base context for background advances }) ``` Key option notes: - **`Store`** is the only required field. All others have in-process defaults. - **`Licensing`**: omit (or pass `nil`) for `StubAllowAll` (all groups permitted). Provide your own `licensing.Resolver` to gate feature groups in production. - **`Secrets`**: omit for an in-memory store seeded from a map. Provide a Vault-backed (or similar) resolver for production. - **`Publisher`**: omit for in-process fan-out. Provide a RabbitMQ publisher to enable multi-process workers and the `action` round-trip. - See [`store/postgres`](https://github.com/Bugs5382/go-saga-orchestration/tree/main/store/postgres) for the Postgres store implementation and SQL migrations. --- ## ⛔ Cancelling a run To abort an in-flight run from outside it — for example, an approval policy that re-submits or withdraws while a run is paused at a `manual_approval` — call `Cancel`: ```go err := sc.Cancel(ctx, runID, "approval withdrawn") ``` This transitions the run to terminal `cancelled`, closes its open user tasks (so none linger `pending` in an approver's inbox), and clears any awaited signal/event or pending wakeup so a stray advance can't resurrect it. `reason` is recorded on the run's `last_error`. `Cancel` is idempotent — a no-op once the run is already terminal. A run that ends in `failed` likewise records the failing step's error on `last_error`, so a terminal run is self-describing without diffing its event log: ```go run, _ := sc.Get(ctx, runID) if run.State == domain.RunStateFailed { log.Warn("run failed", "run", runID, "err", *run.LastError) } ``` --- ## ♻️ Lifecycle When your application shuts down, call `Shutdown` to drain in-flight background advances: ```go shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() if err := sc.Shutdown(shutdownCtx); err != nil { log.Warn("saga shutdown timed out", "err", err) } ``` `Shutdown` cancels the internal background context (pausing new work between steps) and waits for all in-flight goroutines (from `parallel`, `foreach`, and `spawn_saga` child advances) to drain. If they don't finish before `shutdownCtx` expires, `ctx.Err()` is returned. > ⚠️ After `Shutdown` returns, the `Saga` instance should not be reused. --- ## 🛰️ Service mode For multi-process deployments, the repo ships two reference binaries: - **`cmd/api`** — REST API on `:8080`. Handles workflow publishing, run lifecycle, signals, user tasks, and triggers. See [`docs/api.md`](api.md). - **`cmd/engine`** — Saga coordinator + gRPC worker server on `:9090`. Reads from the `saga.advance` RabbitMQ queue and drives runs. Hosts the `ExecuteStep` gRPC stream that workers connect to. See [`docs/grpc.md`](grpc.md). Both binaries require Postgres (`DATABASE_DSN`) and RabbitMQ (`RABBITMQ_URL`). The overall architecture is documented in [`docs/architecture.md`](architecture.md). ```bash go run ./cmd/api # REST API on :8080 go run ./cmd/engine # coordinator + gRPC on :9090 ``` --- # Testing Sagas Source: https://bugs5382.github.io/go-saga-orchestration/docs/testing # 🧪 Testing Sagas The in-memory store makes workflows trivial to unit-test: `saga.InMemory()` runs the coordinator **in-process and synchronously**, with no Postgres, RabbitMQ, or worker processes. By the time `Start` returns, an embedded run has advanced to a terminal state, so a test is just *register → publish → start → assert*. `InMemory()` wires `StubAllowAll` licensing by default, so every verb — including license-gated ones — is available in tests. (To exercise real gating, pass your own resolver via `saga.New(saga.Options{Licensing: ...})`.) --- ## A first unit test This test registers a custom verb, publishes a one-step workflow, starts it, and asserts both the terminal state and the variables the verb produced. ```go package checkout_test import ( "context" "testing" "github.com/Bugs5382/go-saga-orchestration/domain" "github.com/Bugs5382/go-saga-orchestration/engine/verbs" "github.com/Bugs5382/go-saga-orchestration/saga" ) func TestCheckout_ChargesCard(t *testing.T) { ctx := context.Background() sc := saga.InMemory() // Register the custom verb under test. sc.RegisterVerb("charge_card", "common", verbs.HandlerFunc(func(_ context.Context, run domain.SagaRun, _ domain.Step) (map[string]any, error) { total, _ := run.Variables["total"].(float64) return map[string]any{"charge_id": "ch_123", "charged": total}, nil })) // Define and publish the workflow. if err := sc.Register(domain.WorkflowDefinition{ ID: "checkout", Version: 1, Start: "charge", Published: true, Steps: []domain.Step{ {ID: "charge", Type: "charge_card", Next: "done"}, {ID: "done", Type: domain.StepTypeEnd}, }, }); err != nil { t.Fatalf("register workflow: %v", err) } // Start it and read the finished run back. runID, err := sc.Start(ctx, "checkout", map[string]any{"total": 4200.0}) if err != nil { t.Fatalf("start: %v", err) } run, err := sc.Get(ctx, runID) if err != nil { t.Fatalf("get: %v", err) } if run.State != domain.RunStateSucceeded { t.Fatalf("state = %q, want succeeded", run.State) } if got := run.Variables["charge_id"]; got != "ch_123" { t.Fatalf("charge_id = %v, want ch_123", got) } } ``` Run it with the rest of the suite: ```bash task test # go test ./... # or, for just this package: go test ./checkout/... ``` :::tip Numeric literals Inputs flow through the engine as `map[string]any`, and JSON-style numbers are `float64`. Pass `4200.0` (not `4200`) so the `run.Variables["total"].(float64)` type assertion in the verb succeeds. ::: --- ## Asserting failure and compensation A handler that returns an error fails the step. With no `Step.Retry`, `try`/`catch`, or `Step.Compensation` wired, the run lands in `RunStateFailed` — assert that directly: ```go sc.RegisterVerb("charge_card", "common", verbs.HandlerFunc(func(_ context.Context, _ domain.SagaRun, _ domain.Step) (map[string]any, error) { return nil, fmt.Errorf("gateway declined") })) // ... Start + Get ... if run.State != domain.RunStateFailed { t.Fatalf("state = %q, want failed", run.State) } ``` To test **retry**, give a step a `Step.Retry` policy and a handler that fails a fixed number of times before succeeding; drive the backoff waits with a fake clock and assert the handler ran the expected number of times and the run reached `RunStateSucceeded` (retry-then-success) or `RunStateFailed` (exhausted attempts). To test **compensation**, give the completed steps a `Step.Compensation` action and fail a later step with no catching `try_catch` frame. The engine transitions the run through `RunStateCompensating`, dispatches each completed compensable step's compensation action in reverse order, then settles to `RunStateFailed`. Assert the reverse-order dispatch and the terminal `RunStateFailed`. To assert the **caught happy path** instead, wrap the risky steps in a `try`/`catch` branch (see the [verb reference](./verbs.md)) and assert the run reaches `RunStateSucceeded` after the catch branch runs. --- ## Testing `action` steps in-process An `action` step normally pauses the run and waits for an external worker to reply over gRPC — so in a pure `InMemory()` test it would hang indefinitely. To test a workflow that contains an `action` without standing up a worker, register an in-process verb **with the same step type** to stand in for the worker: ```go // Stub the worker: handle the action's type in-process so the run completes. sc.RegisterVerb("action", "common", verbs.HandlerFunc(func(_ context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) { // Inspect step.Inputs / run.Variables and return the result the real // worker would have produced. return map[string]any{"ok": true}, nil })) ``` See [Custom actions](./embedding.md#-custom-actions-worker-round-trip) for how action dispatch works in service mode, and the [gRPC worker protocol](./grpc.md) for testing a real worker against the engine. --- ## Time-dependent verbs Verbs that wait (`wait_for`, `wait_for_event`, timers) read the clock through the injected `clock.Clock`. Inject a `clock.NewFakeClock(...)` via `saga.New(saga.Options{Clock: fc, Store: memory.New()})` and advance it in the test to drive timeouts deterministically, instead of sleeping. The engine's own follow-up tests under `test/e2e` use exactly this pattern. --- # Feature Gating & Licensing Source: https://bugs5382.github.io/go-saga-orchestration/docs/licensing # 🔑 Feature Gating & Licensing go-saga-orchestration ships under the permissive **MIT license** — you can embed it in anything, commercial or free, with no royalty or feature restriction. That software license is separate from the engine's **feature-gating layer** described on this page. The feature-gating layer is a single authorization hook the engine calls before running gated work. It is deliberately generic: you can use it to enforce **paid licensing tiers**, to implement **RBAC** (role-based access control), or both — the engine doesn't care what the answer means, only whether a `(tenant, feature)` pair is allowed. --- ## The resolver hook Every gate goes through one interface: ```go // licensing.Resolver type Resolver interface { IsFeatureEnabled( ctx context.Context, tenantID *uuid.UUID, // the principal: a tenant, org, or user feature string, // the capability being requested overrides map[string]bool, // per-request allow/deny, wins over the resolver ) (bool, error) } ``` Return `true` to allow, `false` to deny. The `tenantID` is your principal and `feature` is your permission — how you map them is entirely up to your implementation. That is what makes the same hook serve licensing and RBAC. --- ## What is gated, and when Each verb belongs to a **license group**. A group maps to a **feature flag**; the engine asks the resolver whether that flag is enabled. The `common` group is never gated. | License group | Feature flag | Example verbs | |---|---|---| | `common` | _(never gated)_ | `set_var`, `transform`, `noop`, `log` | | `waits` | `wf.timers` | `wait_duration`, `wait_until` | | `events_and_signals` | `wf.event_driven` | `emit_event`, `wait_for_event`, `emit_signal` | | `parallel_control` | `wf.parallel` | `parallel`, `foreach` | | `loops_and_recovery` | `wf.loops_recovery` | `while`, `try_catch`, `cancel` | | `human_interaction` | `wf.user_tasks` | `manual_approval`, `collect_input` | | `compositions` | `wf.compositions` | `sub_saga`, `spawn_saga` | | `external_io_advanced` | `wf.external_io` | authenticated `http_request`, `webhook_emit` | | `observability` | `wf.observability` | `metric_emit` | | _(cron triggers)_ | `wf.cron_triggers` | scheduled trigger starts | The check fires in three places: - **At publish** — `ValidateDefinition` rejects a workflow whose steps reference a feature the tenant lacks, so unlicensed workflows never go live. - **At runtime** — each step re-checks its feature before executing (entitlements can change between publish and run). - **At trigger time** — cron-scheduled starts check `wf.cron_triggers` both when the REST trigger is created and when the dispatcher fires it. A denied check fails the publish or the step with a `license_gate` error naming the step, group, and required feature. --- ## Use it for licensing (paid tiers) Map a tenant's purchased plan to the feature flags it unlocks. A `free` tenant publishing a workflow that uses `parallel` is rejected; a `premium` tenant is allowed. ```go type PlanResolver struct { planOf func(uuid.UUID) string // tenant -> "free" | "premium" unlocks map[string]map[string]bool // plan -> feature -> enabled } func (r PlanResolver) IsFeatureEnabled(_ context.Context, tenant *uuid.UUID, feature string, overrides map[string]bool) (bool, error) { if v, ok := overrides[feature]; ok { return v, nil // per-request override wins (e.g. a trial grant) } if tenant == nil { return false, nil } return r.unlocks[r.planOf(*tenant)][feature], nil } ``` ## Use it for RBAC The exact same hook is a per-principal permission check. Treat `feature` as a permission name and `tenantID` as the role-bearing principal, and deny verbs a role isn't allowed to run — independent of whether anyone paid. ```go // Allow only roles that hold the permission for the requested feature. func (r RoleResolver) IsFeatureEnabled(_ context.Context, principal *uuid.UUID, feature string, _ map[string]bool) (bool, error) { if principal == nil { return false, nil } role := r.roleOf(*principal) // e.g. "operator", "viewer" return r.permits(role, feature), nil // RBAC policy lookup } ``` Because licensing and RBAC share one interface, you can also compose them — wrap a plan check and a role check and require both to pass. --- ## Built-in resolvers | Resolver | Use | |---|---| | `licensing.StubAllowAll{}` | Allows everything. The default for `saga.InMemory()` and tests. | | `licensing.NewCached(inner, ttl)` | Wraps any resolver with a per-`(tenant, feature)` cache and applies per-request `overrides`. Wrap your real resolver with this in production. | | `licensing.HTTPFeatureResolver{BaseURL: ...}` | Resolves flags by `GET`-ing a remote service that returns `{"features":[...]}` for a tenant. Wrap it with `Cached`. | --- ## Wiring it in Pass your resolver to `saga.New`; omit it (or use `saga.InMemory()`) to allow everything during development. ```go sc, err := saga.New(saga.Options{ Store: store, Licensing: licensing.NewCached(PlanResolver{ /* ... */ }, 5*time.Minute), }) ``` See [Embedding](embedding#-production-wiring) for the rest of the production options and [Testing](testing) for asserting gated behavior with `StubAllowAll`. --- # REST API Guide Source: https://bugs5382.github.io/go-saga-orchestration/docs/api # REST API Guide This is the narrative companion to [`api/openapi.yaml`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/api/openapi.yaml). It describes how to drive the go-saga-orchestration engine over HTTP: the base URL and ports, the saga lifecycle, the live stream format, the error conventions, and a runnable `curl` example per endpoint group. All schemas in this guide are derived directly from the Go source (`internal/api/handler_*.go`, `internal/api/response.go`, `internal/domain/*.go`, `internal/store/store.go`, `internal/rules/rules.go`) and the handler tests. ## Base URL and ports The REST API is served by the `api` binary (`cmd/api`). It listens on `:8080` by default. The port is configurable via the `WORKFLOW_API_PORT` environment variable. ``` http://localhost:8080 ``` (The engine binary, `cmd/engine`, exposes a separate gRPC port — default `9090`, env `WORKFLOW_ENGINE_GRPC_PORT` — and is not part of this REST surface.) There is **no authentication** wired today. The stream endpoint explicitly notes that auth middleware will land in a later batch. ## Endpoint overview | Method | Path | Purpose | | --- | --- | --- | | GET | `/health/live` | Liveness probe | | GET | `/health/ready` | Readiness probe | | GET | `/api/v1/sagas` | List/filter saga runs (paginated) | | POST | `/api/v1/sagas/start` | Start a saga run | | GET | `/api/v1/sagas/{id}` | Get one saga run | | POST | `/api/v1/sagas/{run_id}/signal/{name}` | Deliver an external signal | | POST | `/api/v1/sagas/{run_id}/user_task/{task_id}/submit` | Submit a user task result | | POST | `/api/v1/sagas/{run_id}/actions/{step_id}/result` | Report an action result (http/rmq workers) | | GET | `/api/v1/sagas/{run_id}/stream` | Live run inspector (WebSocket) | | POST | `/api/v1/registry/register` | Register a service's actions | | GET | `/api/v1/registry/actions` | List registered actions | | POST | `/api/v1/rules/{rule_id}/evaluate` | Evaluate a rule | | POST | `/api/v1/triggers` | Create a trigger | | GET | `/api/v1/triggers` | List triggers | | GET | `/api/v1/triggers/{id}` | Get one trigger | | DELETE | `/api/v1/triggers/{id}` | Delete a trigger | | GET | `/api/v1/workflows/{wf_id}/stats` | Aggregate workflow stats | ## The saga lifecycle A saga is one running instance of a workflow definition. The typical lifecycle: 1. **Start** — `POST /api/v1/sagas/start` with a `workflow_id` and `inputs`. The engine resolves the published definition, creates a run in `pending` state, publishes a `saga.advance` message, and returns `202` with `{ "saga_run_id": "" }`. The run then progresses through states: `pending → running → (paused) → succeeded | failed | cancelled` (with `compensating` during rollback). See the `RunState` enum. 2. **Wait points** — when a workflow reaches a `wait_for_signal`, `manual_approval`, or `collect_input` step, the run moves to `paused` and records what it is awaiting (`awaited_signal`, etc.). 3. **Resume** — there are two ways to wake a paused run: - **Signal**: `POST /api/v1/sagas/{run_id}/signal/{name}`. The signal is always recorded. If the run was paused awaiting exactly this signal name, the server consumes it and publishes `saga.advance`, returning `202`. If the run was not paused-and-awaiting this name, it returns `409` (the signal is still recorded, but nothing advances). - **User task submit**: `POST /api/v1/sagas/{run_id}/user_task/{task_id}/submit`. This persists the task result, then internally appends a signal named `user_task.{task_id}.submitted` carrying the result as its payload, and advances the saga the same way a signal would. Always returns `202` on success. 4. **Observe** — `GET /api/v1/sagas/{run_id}/stream` (WebSocket) tails the run live; `GET /api/v1/sagas/{id}` fetches the current snapshot; and `GET /api/v1/sagas` lists/filters runs. ### The saga run object `GET /api/v1/sagas/{id}` returns a `SagaRun` (the full Go struct, JSON-tagged): ```json { "id": "f1e2d3c4-0000-0000-0000-000000000000", "workflow_id": "example_workflow_v1", "definition_id": "a1b2c3d4-0000-0000-0000-000000000000", "tenant_id": null, "state": "paused", "current_step": "await_approval", "inputs": { "order_id": "ORD-123" }, "variables": {}, "started_at": "2026-05-29T12:00:00Z", "last_event_at": "2026-05-29T12:00:05Z", "requires_manual_review": false, "awaited_signal": "approval.decided", "current_attempt": 0 } ``` Optional fields (`terminal_at`, `trigger_id`, `parent_run_id`, `wakeup_at`, `feature_overrides`, `dry_run`, etc.) are omitted when empty. On a terminal `failed` or `cancelled` run, `last_error` carries the failing step's error message (or the cancel reason), so the run is self-describing without replaying its event log; pair it with the `has_error` list filter to surface failures. ### Listing runs `GET /api/v1/sagas` is paginated (`limit` 1–500, default 50; `offset` >= 0, default 0) and supports filters: `workflow_id`, `state`, `trigger_type`, `since` (RFC3339), `has_error` (bool), `requires_review` (bool). The response always includes a non-null `sagas` array plus `total`, `limit`, `offset`. ### `X-Feature-Override` header (start only) `POST /api/v1/sagas/start` accepts an optional `X-Feature-Override` header to override license feature flags on a per-request basis. This is valid in any environment — standalone, on-prem, dev, or QA — not a QA-only facility. Format: comma-separated `feature=value` pairs, e.g. `wf.parallel=on,wf.timers=off`. Values `on`/`true`/`1` => true; `off`/`false`/`0` => false; anything else is silently ignored. ## The stream event format > **Note:** Despite being listed as an HTTP endpoint, `stream` upgrades to a > **WebSocket** using `gorilla/websocket`. The documentation below reflects > the actual handler implementation. `GET /api/v1/sagas/{run_id}/stream` validates the run exists (returning a JSON `400` for a bad UUID or `404` for an unknown run **before** upgrading, using the standard `{"error": code, "message": msg}` envelope), then upgrades the HTTP connection to a WebSocket. Messages are JSON text frames of the shape: ```json { "type": "run", "data": { /* SagaRun snapshot */ } } { "type": "event", "data": { /* SagaRunEvent */ } } ``` On connect the server sends, in order: 1. one `run` frame — the current `SagaRun` snapshot, 2. one `event` frame per existing audit event for the run, 3. then a live `event` frame for each new event as it is recorded (tailed via Postgres LISTEN/NOTIFY on a per-run channel). A `SagaRunEvent` looks like: ```json { "id": "...", "run_id": "...", "step_id": "await_approval", "attempt": 0, "event_type": "step.paused", "actor": "engine", "recorded_at": "2026-05-29T12:00:05Z" } ``` `event_type` is one of: `saga.started`, `step.dispatched`, `step.started`, `step.succeeded`, `step.failed`, `step.skipped`, `step.paused`, `run.succeeded`, `run.failed`, `run.cancelled`, `compensation.started`, `log`, `metric`, `rule.evaluated`, `license.gate.rejected`. ## Error conventions All API errors use a single structured JSON envelope: ```json { "error": "saga_not_found", "message": "f1e2d3c4-..." } ``` `error` is a stable, machine-readable code; `message` is human-readable detail. On `5xx` responses the real error is logged server-side and only a generic `"internal error"` message is returned to the client — never raw internal detail. On `4xx` the message may include safe client-input context (e.g. the offending field name). Machine-readable codes include: `bad_request`, `not_found`, `internal`, `invalid_config`, `publish_failed`, `unprocessable`, `conflict`, `workflow_not_found`, `saga_not_found`, `trigger_not_found`. ## curl examples by group ### Health ```bash curl -s http://localhost:8080/health/live # {"status":"live"} curl -s http://localhost:8080/health/ready # {"status":"ready"} ``` ### Sagas ```bash # Start a saga curl -s -X POST http://localhost:8080/api/v1/sagas/start \ -H 'Content-Type: application/json' \ -H 'X-Feature-Override: wf.parallel=on' \ -d '{"workflow_id":"example_workflow_v1","version":"latest","inputs":{"order_id":"ORD-123"}}' # 202 {"saga_run_id":""} # Get one run curl -s http://localhost:8080/api/v1/sagas/ # List/filter runs curl -s 'http://localhost:8080/api/v1/sagas?workflow_id=example_workflow_v1&state=paused&limit=20' # Deliver a signal (202 if it advanced a paused run, 409 otherwise) curl -s -X POST http://localhost:8080/api/v1/sagas//signal/approval.decided \ -H 'Content-Type: application/json' \ -d '{"payload":{"approved":true}}' # Submit a user task curl -s -X POST http://localhost:8080/api/v1/sagas//user_task//submit \ -H 'Content-Type: application/json' \ -d '{"submitted_by":"alice@example.com","result":{"decision":"approve"}}' # Stream (WebSocket) — use a WS client, e.g. websocat websocat ws://localhost:8080/api/v1/sagas//stream ``` ### Registry ```bash # Register actions (idempotent; call on service startup) # An action may carry an optional dispatch descriptor — "transport" # (grpc | http | rmq) and "address" (callback URL for http, queue name for # rmq; required only for http/rmq). Omit it for the gRPC default. curl -s -X POST http://localhost:8080/api/v1/registry/register \ -H 'Content-Type: application/json' \ -d '{ "service":"example", "service_version":"0.18.2", "actions":[ {"action_name":"set_state","version":1,"category":"record_lifecycle","compensable":true,"input_schema":{},"output_schema":{}}, {"action_name":"send_email","version":1,"input_schema":{},"output_schema":{},"transport":"http","address":"https://worker.example.com/actions/send_email"} ] }' # 200 {"service":"example","service_version":"0.18.2","registered":2} # List actions (descriptor is echoed back) curl -s 'http://localhost:8080/api/v1/registry/actions?service=example&category=record_lifecycle' # {"actions":[ ... ]} ``` ### Action result callback (http / rmq workers) `POST /api/v1/sagas/{run_id}/actions/{step_id}/result` gRPC workers reply over the `ExecuteStep` stream. Workers reached over the `http` or `rmq` transport have no return stream, so they report their result asynchronously here. The endpoint applies the same `CompleteAction` / `FailAction` semantics as the gRPC path: success merges the result into the run's variables and resumes the saga; failure transitions the run to `failed`. Attempt handling and idempotency are preserved — a stale `attempt` is a no-op. Send **exactly one** of `result` or `error`. `attempt` is optional; omitted, it defaults to the run's current attempt (the only in-flight dispatch). ```bash # Success — completes the action and advances the saga. curl -s -X POST http://localhost:8080/api/v1/sagas//actions//result \ -H 'Content-Type: application/json' \ -d '{"result":{"ticket_number":"INC-999"}}' # 202 # Failure — transitions the run to failed. curl -s -X POST http://localhost:8080/api/v1/sagas//actions//result \ -H 'Content-Type: application/json' \ -d '{"error":{"code":"ERR_WORKER_CRASH","message":"worker panicked","retryable":false}}' # 202 ``` ### Rules ```bash curl -s -X POST http://localhost:8080/api/v1/rules/triage/evaluate \ -H 'Content-Type: application/json' \ -d '{"inputs":{"priority":"p1"}}' # 200 {"output":{"branch":"high"},"audit":[{"index":0,"when":"priority == 'p1'","matched":true}]} ``` ### Triggers ```bash # Create curl -s -X POST http://localhost:8080/api/v1/triggers \ -H 'Content-Type: application/json' \ -d '{ "trigger_type":"record_transition", "workflow_id":"example_workflow_v1", "version":1, "config":{"record_type":"order","from_state":"created","to_state":"pending_review"}, "enabled":true, "created_by":"admin" }' # 201 — body uses PascalCase field names (see note below) # List (optional ?type= and ?enabled=true|false|1|0) curl -s 'http://localhost:8080/api/v1/triggers?type=record_transition&enabled=true' # {"triggers":[ ... ]} # Get one curl -s http://localhost:8080/api/v1/triggers/ # Delete (204 on success, 404 if missing) curl -s -X DELETE http://localhost:8080/api/v1/triggers/ ``` ### Workflows ```bash curl -s http://localhost:8080/api/v1/workflows/example_workflow_v1/stats # {"workflow_id":"example_workflow_v1","success_rate_24h":0.83,"last_run_at":"2026-05-29T12:00:00Z","in_flight":2} ``` ## Assumptions and ambiguities resolved - **Stream is WebSocket, not SSE.** The brief said SSE; the code (`handler_stream.go`) uses `gorilla/websocket` and emits `{type, data}` JSON frames over a WebSocket. Documented as WebSocket. The OpenAPI spec models the endpoint with a `101 Switching Protocols` response and documents the `StreamFrame` schema, since OpenAPI 3.0/3.1 cannot natively describe WebSocket message streams. - **`SagaTrigger` is serialized with PascalCase keys.** The `domain.SagaTrigger` struct has **no JSON tags**, so Go's `encoding/json` emits the exact Go field names: `ID`, `TriggerType`, `WorkflowID`, `Version`, `Config`, `Enabled`, `TenantID`, `CreatedAt`, `CreatedBy`. The handler tests confirm this by round-tripping responses into `domain.SagaTrigger`. The **request** body (`TriggerCreateRequest`), by contrast, is a separate struct with snake_case JSON tags. Both shapes are documented faithfully and differ on purpose. - **Single JSON error contract.** All handlers use `WriteError` producing `{"error": code, "message": msg}`. The `PlainTextError` OpenAPI component and the legacy `http.Error` call sites have been removed. - **`config` map values for triggers and rule `inputs`/`output` / action `input_schema`/`output_schema` are free-form JSON objects** (Go `map[string]any`), so they are modeled as objects with `additionalProperties: true`. For `trigger_type: record_transition` the server additionally requires `config.record_type`, `config.from_state`, and `config.to_state` to be non-empty strings (validated server-side; returns `422` `invalid_config` on failure). - **`tenant_id` typing differs by endpoint.** On `POST /sagas/start` it is a UUID (`*uuid.UUID`). On `POST /triggers` and `POST /rules/.../evaluate` it is a string in the request body that the server attempts to parse as a UUID (invalid values are silently dropped rather than rejected). Modeled per the actual struct types. - **`success_rate_24h` and `last_run_at` are nullable** in `WorkflowStats` (pointers in Go): `success_rate_24h` is null when there were no runs in the last 24h; `last_run_at` is null when there have been no runs at all. - **Signal/user-task success responses have empty bodies** (the handlers only call `w.WriteHeader`), so no response schema is defined for their 2xx/4xx status codes beyond the status itself. --- # Architecture & Concepts Source: https://bugs5382.github.io/go-saga-orchestration/docs/architecture # Architecture & Concepts Module: `github.com/Bugs5382/go-saga-orchestration` This document explains how the go-saga-orchestration engine works end to end. It is written to be read by a human or an AI assistant who needs to understand the system before changing it. Everything below is derived from the actual source; where the code's intent is ambiguous, the observable behavior is described rather than guessed. --- ## 1. Overview & the two binaries The system is a **saga / workflow orchestration engine**. Workflows are declarative graphs of *steps* (each step has a *verb*). Running instances are *saga runs*. The engine walks each run forward one step at a time, persisting state to Postgres and coordinating asynchronous work over RabbitMQ. Long-running or external work is delegated to *worker* services that connect over gRPC. There are two process binaries; both load the same `config.Config` (`internal/config/config.go`) from environment variables and both run Postgres migrations on boot. ### `cmd/api` — REST surface (`:8080`) Source: `cmd/api/main.go`. Boot sequence: 1. `postgres.Open` + `postgres.Migrate` (embedded migrations applied on every boot, idempotent). 2. Connect to RabbitMQ, declare topology (`mq.DeclareTopology`), open a `mq.Publisher`. 3. Construct the chi router (`api/router.go`) wiring the handlers below. 4. Serve HTTP on `cfg.API.Port` (default `8080`, env `WORKFLOW_API_PORT`), with graceful shutdown on SIGINT/SIGTERM. Routes (`api/router.go`): - `GET /health/live`, `GET /health/ready` - `GET /api/v1/sagas` — list/filter runs (paginated) - `POST /api/v1/sagas/start` — start a run (returns `202` + `saga_run_id`) - `GET /api/v1/sagas/{id}` — fetch one run - `POST /api/v1/sagas/{run_id}/signal/{name}` — deliver an external signal - `POST /api/v1/sagas/{run_id}/user_task/{task_id}/submit` — submit a user task - `GET /api/v1/sagas/{run_id}/stream` — WebSocket run inspector (tails audit events via Postgres LISTEN/NOTIFY) - `POST /api/v1/registry/register`, `GET /api/v1/registry/actions` — worker action registry - `POST /api/v1/rules/{rule_id}/evaluate` — evaluate a stored rule - `POST|GET|GET|DELETE /api/v1/triggers...` — trigger CRUD - `GET /api/v1/workflows/{wf_id}/stats` — per-workflow aggregate stats The API process does **not** advance sagas itself — it only persists rows and publishes `saga.advance` messages. ### `cmd/engine` — coordinator + gRPC (`:9090`) Source: `cmd/engine/main.go`. This is the worker process that actually runs sagas. Boot sequence: 1. Postgres open + migrate. 2. RabbitMQ connect + `mq.Publisher`. 3. Construct the `engine.Coordinator` with a `SystemClock`, an in-memory secrets resolver, and `licensing.StubAllowAll{}` (dev/test license resolver — approves everything). 4. Start the **timer dispatcher** goroutine (`engine.Timer`) — polls for due wakeups every second. The **cron dispatcher** goroutine (`engine.CronDispatcher`) starts here too, but only when `WORKFLOW_CRON_DISPATCHER` is enabled (default on). Disabling it lets cron run on a single dedicated engine pod while the saga-processing replicas skip the loop — see the cron topology note below. 5. Construct (but, in the committed code, leave un-started) the `TriggerDispatcher` + `EventSubscriber`. The comment notes `RunRMQ` wiring is deferred until the prod env has RMQ; `_ = sub` keeps it referenced. 6. Start the gRPC server on `cfg.Engine.GRPCPort` (default `9090`, env `WORKFLOW_ENGINE_GRPC_PORT`) so workers can open `ExecuteStep` streams. 7. Block in `mq.ConsumeSagaAdvance`, dispatching each `saga.advance` message to `coord.HandleAdvance`. --- ## 2. Core concepts: sagas, runs, definitions, coordinator ### Definitions (`domain/definition.go`) A `WorkflowDefinition` is one version of one workflow: an `ID` (stable workflow id), `Version`, optional `TenantID`, a `Start` step id, an optional `Entrypoints` map (see below), and a list of `Step`s. A `Step` has an `ID`, a `Type` (the verb / `StepType`), an optional `Next` (the default outgoing edge), an optional `Action` string, an `Inputs` map (verb-specific config), optional `Compensation`, optional `Retry` policy, and optional `Branches` (named outgoing edges used by `decision`/`while`/`parallel`). `StepByID` looks up a step within a definition. ### Entrypoints / call tree `WorkflowDefinition.Entrypoints` is an optional `map[string]string` that maps entry names to step IDs. The empty string `""` and the name `"default"` always resolve to `Start` regardless of the map. Any other name must appear in `Entrypoints`; an unknown name is a runtime error (returned by `ResolveEntry`). The `ValidateDefinition` function checks that every step ID referenced in `Entrypoints` actually exists in the definition. `saga.Start` is unchanged and is equivalent to `saga.StartAt(ctx, workflowID, "", inputs)`. `saga.StartAt(ctx, workflowID, entrypoint, inputs)` resolves the entry point before creating the run, setting `run.CurrentStep` to the resolved step ID instead of `def.Start`. Triggers honor the entry point via `config.entrypoint` in the trigger's config map — `TriggerDispatcher.Dispatch` calls `def.ResolveEntry(trigEntrypoint)` to determine the starting step. The `sub_saga` and `spawn_saga` verbs accept an `"entrypoint"` input key that is resolved against the child definition the same way. `RuleDefinition` (`domain/rule.go`) is a separately versioned, published artifact used by the `decision` verb (see §4). ### Runs (`domain/run.go`) A `SagaRun` is one executing instance. Key fields: - `State` (`RunState`): `pending`, `running`, `paused`, `compensating`, `succeeded`, `failed`, `cancelled`. `IsTerminal()` is true for `succeeded`/`failed`/`cancelled`. Runs that reach `failed`/`cancelled` also carry `LastError` — the failing step's error message or the cancel reason. An external caller can terminate a paused/in-flight run with `Coordinator.Cancel(runID, reason)` (or `saga.Cancel` when embedding), which closes its open user tasks and clears its await/wakeup markers. - `CurrentStep`: the step id the run is at. - `Inputs` (immutable start inputs) and `Variables` (mutable working state that verbs read and write). - Pause markers: `WakeupAt`, `AwaitedSignal`, `AwaitedEventTopic`, `AwaitedEventHeaders`, `AwaitedActionDispatch`, `CurrentAttempt`. - Composition links: `ParentRunID`, `ParentStepID`, `ParentBranchID` (set on child runs spawned by `parallel`/`foreach`/`sub_saga`/`spawn_saga`). - `TryCatchStack` (`[]TryCatchFrame`), `DryRun`, `FeatureOverrides` (per-run feature overrides), `RequiresManualReview`, `TriggerID`. ### The coordinator (`engine/coordinator.go`) `Coordinator` owns a `store.Store`, a `Publisher` (re-enqueues `saga.advance`), the verb `Registry`, a clock, a secrets resolver, and a license resolver. `NewCoordinator` builds the verb registry via `verbs.Default(...)`. `HandleAdvance` is the consumer callback; it simply calls `Advance(runID)` and wraps any error (a non-nil return NACKs the RabbitMQ message → requeue). ### How a run advances (`engine/advance.go`) `Advance` is a synchronous loop over one run. Each iteration **re-reads the run from the store** so it sees variable updates the previous step wrote. Per iteration: 1. **Terminal?** If `run.State.IsTerminal()`, return (ACK). 2. **Paused?** If state is `paused`, decide whether a wakeup condition holds: - *Time-based*: `WakeupAt != nil && WakeupAt <= now` (from `wait_duration`/`wait_until`, or `WakeFromExternal` which sets `wakeup_at=now`). - *External*: no pending await markers (`AwaitedSignal`/`AwaitedEventTopic` nil) **and** `WakeupAt == nil` (a parallel/sub-saga join wake). If a wakeup condition holds, it clears pause (`ClearPause`), emits `step.succeeded` for the paused step, transitions to `step.Next`, and continues the loop. The paused step is treated as already-succeeded because wait verbs persist their pause marker and return `ErrSagaPaused` *after* succeeding. If neither wakeup condition holds, the run is still legitimately paused → return (ACK; the message arrived prematurely). 3. **Resolve the step.** Load the definition, pick `run.CurrentStep` (or `def.Start` on first entry, emitting `saga.started`). Emit `step.dispatched` and set state `running`. 4. **`end` step** short-circuits to `completeRun` (emits `step.succeeded` + `run.succeeded`, sets state `succeeded`, then checks any parent join). `end` is intentionally NOT in the verb registry. 5. **License gate.** Look up the verb's `LicenseGroup` (with `LicenseGroupForStep` dynamic override), map to a feature flag (`GroupToFeature`), and call `licensing.IsFeatureEnabled`. On rejection: emit `license.gate.rejected`, set state `failed`, return error. (In the engine binary the resolver is `StubAllowAll`, so this never rejects in dev.) 6. **Execute the verb.** `entry.Handler.Execute(ctx, run, step)` returns `(result map, error)`. - If `error == ErrSagaPaused`: emit `step.paused`, return (ACK). The verb has already persisted its pause marker; a timer/signal/event/join wake will republish `saga.advance`. - If another error: **try_catch handling** — `PopTryCatch`; if a frame was popped, write `_error` (`{step_id, message, verb}`) into Variables, emit `step.failed` (actor `engine-caught`), set `CurrentStep` to the frame's `CatchStep`, and continue the loop. Otherwise emit `step.failed`, set state `failed` via `MarkRunFailed` (which stamps `terminal_at` and persists the failing step's error in the run's `last_error`), run `checkParentJoin`, return error. - On success: if `result` is non-empty, merge it into Variables (`UpdateRunVariables` — supports dotted keys for nested writes). Emit `step.succeeded`. 7. **Pick the next step.** Default is `step.Next`. If `result["branch"]` is a non-empty string and `step.Branches[branch]` exists, follow that branch instead. For `decision`/`while`, a missing branch is an error. A run with no `next` that is not `end` is an error. Set state `running` + new `CurrentStep`; loop. ### Joins (`checkParentJoin`, `aggregateChildResults`) After any terminal write of a child run, `checkParentJoin` runs. It only proceeds if the parent is still `paused` on the exact step that spawned the child (`parent.CurrentStep == *run.ParentStepID`) — this guards fire-and-forget `spawn_saga` children from prematurely waking a parent. Join strategy comes from the parent step's `inputs.join_strategy`: - `"all"` (default): wake only when **every** sibling is terminal. - `"quorum"`: wake when `quorum_n` siblings reach `succeeded` (`quorum_n` may be a literal int or a CEL string evaluated against the parent's variables; non-numeric/invalid falls back to `"all"`). Remaining branches keep running but no longer gate the parent. On wake it aggregates child results into `Variables._parallel..branches` (per child: `key`, `variables`, terminal `state`, and the first submitted `_user_task` if any), calls `WakeFromExternal(parent)` (clears await markers, sets `wakeup_at=now`), and publishes `saga.advance` for the parent. ### Retry (`engine/retry.go`) `DefaultRetryPolicy` is max 3 attempts, 1s initial backoff, 60s cap, ×2 multiplier, jitter on. `Backoff(policy, attempt, jitter)` computes exponential backoff with optional ±25% jitter. NOTE: these helpers exist and are tested, but the `Advance` loop in the committed code does not itself re-dispatch failed verbs through them — step-level retry is the worker/action concern (RabbitMQ redelivery + worker idempotency), and `current_attempt` is bumped by the `action` verb. This is a notable gap between the helper and its wiring. --- ## 3. The verb catalog All 31 step types live in `engine/verbs/` (30 in the registry plus `end`, which is handled inline by the coordinator). They implement `Handler.Execute(ctx, run, step) (map[string]any, error)`. The returned map is merged into `run.Variables`; returning `ErrSagaPaused` suspends the run; returning `ErrSagaCancelled` transitions the run to `cancelled` (terminal); any other error fails the step (subject to try_catch). The registry (`verbs.Default`, `registry.go`) maps each `StepType` to a handler plus a license group. | Verb | Purpose | Key fields / notes | |------|---------|--------------------| | `action` | Dispatch a registered action to a worker over its declared transport, then pause awaiting the worker's result. | `step.Action` = `"."` (must contain a dot); `step.Inputs` forwarded verbatim. Bumps `current_attempt`, computes a SHA-256 idempotency key from (run, step, attempt), calls `MarkAwaitingAction`, then dispatches the `ActionPayload` by the action's registered transport (see **Action dispatch routing** below), returns `ErrSagaPaused`. License group `common`. Resumed by `CompleteAction`/`FailAction` — over the gRPC stream for gRPC workers, or via the result-callback REST endpoint for http/rmq workers. | | `cancel` | Cancel a run. | `run_id` (optional string). No `run_id` (or `run_id` equal to the current run) → self-cancel: returns `ErrSagaCancelled` and the engine sets state `cancelled` (terminal); the run ends immediately. With a different `run_id` → cancels that target run (`UpdateRunState` → `cancelled`, appends `run.cancelled` event) and the current run continues to `Next`. `reason` (optional string) is available for logging. Group `loops_and_recovery`. | | `assert` | Fail the saga if a CEL expression is not true. | `expr` (required), `code` (optional, default `assertion_failed`). Evaluates against `run.Variables`; non-true → error `": is false"`. Group `common`. | | `collect_input` | Create a user task that **requires** a form and pause until submitted. | `assignee` (required), `form_schema` (required, non-empty), `due_in` (optional Go duration). Creates a `UserTask`, sets paused awaiting signal `user_task..submitted`, returns `ErrSagaPaused`. Group `human_interaction`. | | `decision` | Evaluate a stored rule and branch on its output. | `rule_id` (required), `inputs_map` (optional map ruleKey→variableName narrowing the inputs; otherwise all of `run.Variables`). Calls `rules.Evaluate`, emits `rule.evaluated` with the audit trail, returns the rule output (engine reads `result["branch"]` to pick `step.Branches`). Group `common`. | | `emit_event` | Publish an event via the configured `EventEmitter`. | `topic` (required string), `headers` (optional map — values stringified), `payload` (optional map). Delegates to `verbs.EventEmitter.EmitEvent`; fails if no emitter is configured. See below for in-process vs. service-mode behavior. Group `events_and_signals`. | | `emit_signal` | Send a signal to a target run (the send-side complement of `wait_for_signal`). | `run_id` (required string UUID), `name` (required string), `payload` (optional map). Appends a `SagaSignal` row via `AppendSignal`, then calls `TryConsumeAwaitedSignal`; if the target was paused awaiting that signal, clears its markers and publishes `saga.advance` to resume it immediately. Group `events_and_signals`. | | `end` | Terminate the saga successfully. | No inputs. Dispatched inline by the coordinator (`completeRun`), NOT via the registry — sets run state `succeeded`. | | `error` | Halt the saga with a non-retryable error. | `code` (required), `message` (optional). Always returns an error. Group `common`. | | `filter` | Keep list elements where a CEL predicate is truthy. | `list` (CEL → `[]any`), `expr` (predicate; current element bound as `_`), `out_var` (destination). All required. Group `common`. | | `foreach` | Fan out one child run per list element (parallel only in v1). | `list` (CEL → list), `body` (`[]any` of step objects), `start` (first step id in body), `parallel` (optional, default true; `false` is rejected — "use `while`"). Each child gets `_foreach_item` + `_foreach_index` in inputs. Empty list → advance to `Next` without spawning. Pauses parent; woken by the join hook. Group `parallel_control`. | | `http_request` | Synchronous outbound HTTP request; merge response into Variables. | `method` (default GET), `url` (required), `headers`, `body` (JSON), `timeout_s` (default 30), `secret_ref` (→ `Authorization` header via secrets resolver), `out_var` (default `http_result`). Writes ``, `_status`, `_headers`. Group dynamically `common` for GET-no-auth, else `external_io_advanced` (see `LicenseGroupForStep`). | | `log` | Append a `log` audit event. | `message` (required), `level` (optional, default `info`). No variable output. Group `common`. | | `manual_approval` | Create a user task (form optional) and pause until submitted. | `assignee` (required), `due_in` (optional duration), `form_schema` (optional). Same await-signal mechanism as `collect_input`. Group `human_interaction`. | | `map` | Transform each list element via a CEL expression. | `list`, `expr` (element bound as `_`), `out_var` — all required. Writes the mapped list to `out_var`. Group `common`. | | `merge` | Deep-merge a CEL-evaluated map into a target variable. | `from` (CEL → map), `into` (target var, dotted). Flattens nested maps into dotted keys rooted at `into` (last-write-wins). Group `common`. | | `metric_emit` | Append a `metric` audit event. | `name` (required), `value` (required), `labels` (optional map). Prometheus side-channel deferred; the event is the record. Group `observability`. | | `noop` | Do nothing (placeholder / authoring aid). | No inputs. Returns empty result. Group `common`. | | `parallel` | Fan out N branch child-runs and pause the parent until the join is satisfied. | `branches` (`[]any` of `{start, steps}` objects, short-form `{type, inputs}` normalized, or a CEL string → list); `join_strategy` `all` (default) or `quorum`; `quorum_n` required for quorum (literal or CEL, must be ≤ branch count). Each branch becomes a synthetic published definition + child run. Pauses parent; woken by `checkParentJoin`. Group `parallel_control`. | | `set_var` | Write a literal or CEL-evaluated value to a variable. | `out_var` (required, dotted ok); exactly one of `value` (literal) or `expr` (CEL); `expr` wins if both present. Group `common`. | | `spawn_saga` | Start a named workflow as a **fire-and-forget** child; parent continues immediately. | `workflow_id` (required), `inputs` (optional), `entrypoint` (optional — same semantics as `sub_saga`). Resolves the published workflow for the parent's tenant, `SpawnChildRun` (carries `ParentRunID` for audit), publishes child advance, returns empty result (no pause). The parent never pauses, so the join guard never wakes it. Group `compositions`. | | `sub_saga` | Start a named workflow as a child and **pause** until the child terminates. | `workflow_id` (required), `inputs` (optional), `entrypoint` (optional — name of a named entry point in the child definition; `""` / `"default"` → child's `Start`). Same spawn mechanism as `parallel` with a single branch; pauses parent (`ErrSagaPaused`), woken by `checkParentJoin`. Group `compositions`. | | `switch` | Evaluate a CEL expression to a string key and branch on it. | `expr` (required CEL expression). The expression must evaluate to a `string`; returns `{"branch": }`. The engine routes the run to `step.Branches[key].Next`. A missing branch key is a runtime error (same behavior as `decision`/`while`). Group `common`. | | `transform` | Evaluate a CEL expression and write the result to a variable. | `expr` (required), `out_var` (required, dotted ok). Group `common`. | | `try_catch` | Push a try/catch frame so an error inside the try body jumps to a catch step. | `try` (`[]any` of step ids — used by the publish-time validator to forbid parallel-in-try; not consumed at runtime), `catch` (required step id). Author wires `step.Next` to the first try step. Frame stays on the stack until the saga terminates. On error inside the block, the coordinator pops the frame, writes `_error`, and jumps to the catch step. Group `loops_and_recovery`. | | `wait_duration` | Pause for a fixed relative duration. | `duration` (required Go duration; negative rejected). Sets `wakeup_at = now + d` via `SetPausedWithWakeup`, returns `ErrSagaPaused`. Timer dispatcher republishes advance when due. Group `waits`. | | `wait_for_event` | Pause until a matching RabbitMQ event arrives. | `topic` (required routing key), `headers` (optional subset that the incoming event must match, values stringified). `SetPausedAwaitingEvent`, returns `ErrSagaPaused`. Woken by `EventSubscriber`. Group `events_and_signals`. | | `wait_for_signal` | Pause until a named external signal arrives via REST. | `name` (required), `timeout_s` (optional; sets a deadline, else waits indefinitely). `SetPausedAwaitingSignal`, returns `ErrSagaPaused`. The signal handler's `TryConsumeAwaitedSignal` clears markers + sets `wakeup_at=now`. Group `events_and_signals`. | | `wait_until` | Pause until an absolute wall-clock instant. | `timestamp` (required RFC3339). Past timestamps clamp to `now` (wake next tick). `SetPausedWithWakeup`, returns `ErrSagaPaused`. Group `waits`. | | `webhook_emit` | POST a payload to an external URL (optionally async / HMAC-signed). | `url` (required), `body` (required, JSON), `secret_ref` (optional → `X-Webhook-Sig: sha256=`), `timeout_s` (default 15), `headers`, `async` (default false — fire in a goroutine, ignore result), `out_var` (default `webhook_result`; sync writes `_status`, async writes `_async`). Group `external_io_advanced`. | | `while` | Evaluate a CEL condition and branch `continue`/`exit`, with a loop cap. | `condition` (required CEL), `max_iterations` (optional, default 100, hard cap 10000). Returns `{branch, _while..iter}`. Author wires `branches.continue` → loop body and `branches.exit` → after-loop; the body's last step loops back to this step. Group `loops_and_recovery`. | ### Action dispatch routing (`domain.ActionRegistration` dispatch descriptor) An `ActionRegistration` may carry an optional **dispatch descriptor** — `Transport` (`grpc` | `http` | `rmq`) and `Address` — telling the coordinator *how* to reach the worker that runs the action. `Transport` empty or `grpc` is the zero-config default; `http`/`rmq` require a non-empty `Address` (a callback URL for `http`, a queue name for `rmq`). The fields persist in `definitions.action_registry` (columns `transport`, `address`, migration `009_action_dispatch_descriptor`) and flow through the registry REST surface unchanged. When the `action` verb runs it resolves the registration for `step.Action` (parsing `.`, taking the **latest registered version**) and routes by transport: - **grpc** (default, or no descriptor, or an unregistered action) — publishes `ActionPayload` to the `action.direct` exchange with routing key = the action. The worker is connected over the gRPC `ExecuteStep` stream. - **http** — POSTs the `ActionPayload` (JSON) to `Address` via `internal/dispatch.HTTPDispatcher`. A 2xx is "accepted", not "completed". - **rmq** — publishes the `ActionPayload` to the queue named by `Address` (default exchange, routing key = queue name) via `mq.Publisher.DispatchRMQQueue`. The http/rmq dispatchers are wired into the engine via `verbs.WithHTTPDispatcher` / `verbs.WithRMQDispatcher` options on `NewCoordinator`; omitting them leaves the gRPC-only default and makes an `http`/`rmq` action a hard error at dispatch time. **Result return.** gRPC workers reply over the `ExecuteStep` stream (`internal/grpc/server.go`). http/rmq workers have no return stream, so they report their result asynchronously over a transport-agnostic **result-callback REST endpoint**: `POST /api/v1/sagas/{run_id}/actions/{step_id}/result`. Its handler (`api/handler_action_result.go`) applies the same `CompleteAction` (success) / `FailAction` (failure) store hooks the gRPC server uses, preserving attempt handling and idempotency (a stale `attempt` is a no-op). License groups → feature flags are defined in `engine/verbs/license_groups.go` (`GroupToFeature`): `common`→(none), `observability`→`wf.observability`, `external_io_advanced`→`wf.external_io`, `waits`→`wf.timers`, `events_and_signals`→`wf.event_driven`, `human_interaction`→`wf.user_tasks`, `parallel_control`→`wf.parallel`, `loops_and_recovery`→`wf.loops_recovery`, `compositions`→`wf.compositions`. ### EventEmitter (`verbs.EventEmitter`, `saga/event_emitter.go`, `cmd/engine`) `emit_event` delegates to a `verbs.EventEmitter` interface injected at coordinator construction. In **embedded mode** (`saga.InMemory()` / `saga.New(...)`), the `saga` package wires an `InProcessEventEmitter` that calls `store.FindRunsByAwaitedEvent`, applies the same header-subset match as `EventSubscriber`, calls `WakeFromExternal` on each match, and publishes `saga.advance` in-process — no broker needed. In **service mode** (`cmd/engine`), the coordinator is given an `mqEventEmitter` that calls `mq.Publisher.PublishEvent`, putting the event on the `workflow.events` exchange so that all engine pods receive it via their `EventSubscriber` goroutine. Note: starting *new* runs from event-triggers fired via `emit_event` is a follow-up; in the current implementation only already-paused runs are woken. ### Timeout / escalation branch Any wait step (`wait_for_signal`, `wait_for_event`, or any step that sets a deadline via `timeout_s`) can optionally define a `"timeout"` key in `step.Branches`. When the timer dispatcher fires the deadline (`WakeupAt <= now`) and the run's await markers (`AwaitedSignal` / `AwaitedEventTopic`) are still set — meaning a real signal or event did not arrive first — the coordinator routes the run to `step.Branches["timeout"].Next` instead of `step.Next`. If no `"timeout"` branch is defined, `step.Next` is used as normal (backward-compatible). A real signal/event always clears the await markers before the `saga.advance` message arrives, so `timedOut` is false in that case. --- ## 4. CEL rules: expression and rule evaluation ### CEL (`internal/cel`) `internal/cel/cel.go` wraps `google/cel-go`. `NewEnv(varNames...)` builds an environment where every supplied variable name is declared as `dyn` (any JSON-shaped value), then applies the v1 subset. `Compile(expr)` parses + type-checks once into a reusable `Program`; `Eval(vars)` runs it and **deep-converts** the result back to Go-native types (`[]any`, `map[string]any`) via `refValueToNative` — this is why verbs like `parallel`/`filter`/`map` can rely on getting `[]any` back. Map keys that are not strings cause an error. `internal/cel/subset.go` defines the allow-list: only the CEL **stdlib** (arithmetic, string ops, equality, `&&`, `||`, `in`) and the **strings extension** are enabled. Native Go host functions, and file/time/network functions, are deliberately NOT exposed — this is the single chokepoint for widening the surface later. Verbs that take CEL (`assert`, `transform`, `set_var.expr`, `filter`, `map`, `merge.from`, `while.condition`, `parallel.branches`/`quorum_n`, `foreach.list`) build the env from the keys of `run.Variables`. List-iterating verbs additionally bind the current element as `_`. ### Rules (`internal/rules` + `domain/rule.go`) A `RuleDefinition` (type `decision_table`, hit policy `first` — the only supported variants in v1) holds an ordered list of `DecisionTableRow{When (CEL), Then (output map)}` plus an optional `DefaultOutput`. `rules.Evaluate(def, inputs)` builds a CEL env from the input keys, evaluates each row's `When` in order, and returns the first matched row's `Then` (with an audit trail of `{index, when, matched}`). If nothing matches and there is a `DefaultOutput`, that is returned; otherwise it errors `no_decision_row_matched`. The `decision` verb calls this and records the audit in a `rule.evaluated` event. --- ## 5. Triggers, signals, and user tasks These are the three ways external activity drives runs. ### Triggers (`domain/trigger.go`, `engine/trigger_dispatcher.go`) A `SagaTrigger` binds an external event to a workflow. v1 supports one `TriggerType`: `record_transition`. The `TriggerDispatcher.Dispatch` inspects a RabbitMQ delivery: it only acts on routing keys shaped `*.record.transitioned.*`, decodes the body for `record_type`/`from_state`/`to_state`, lists enabled `record_transition` triggers, and for each whose config matches all three it: 1. builds saga inputs from the body via the trigger's `input_mapping` (v1 supports only top-level `$.field` references; unmapped values pass through as literals; empty mapping → the body itself), 2. resolves tenant (trigger's tenant wins, else body `tenant_id`, else nil), 3. resolves + upserts the published workflow definition, 4. creates a `SagaRun`, injects startup variables, and publishes `saga.advance`. Per-trigger failures are logged and skipped; the first store/publish error is returned. CRUD is exposed via `/api/v1/triggers`; some first-party triggers are seeded by migrations. ### Cron triggers (`domain/trigger.go`, `engine/cron_dispatcher.go`) A cron trigger is a `SagaTrigger` with `trigger_type: cron`. Its `config` map carries exactly one of: - `schedule` — a standard five-field cron expression (`* * * * *`) or a `@`-descriptor (`@hourly`, `@daily`, etc.). Granularity is one minute. - `interval` — a Go duration string (e.g. `"30s"`, `"5m"`) enabling sub-minute cadences. `next_fire_at` is advanced by the interval on each claim. Exactly one of `schedule` or `interval` must be present; supplying both or neither is rejected at create time with HTTP 400. The config also accepts: - `entrypoint` (optional) — a named entry point in the target workflow definition; resolves the same way as `config.entrypoint` on event triggers. - `input` (optional) — a JSON-compatible map injected as the run's start inputs. The `CronDispatcher` polls on a ~1-second tick. On each tick it calls `ListDueCronTriggers` (triggers whose `next_fire_at ≤ now`) and attempts to claim each one via `ClaimCronFire` — a compare-and-swap on `next_fire_at` guarded by the row's current value. Exactly one engine pod wins the claim per window; pods that lose the race silently skip. On a successful claim the dispatcher creates a `SagaRun` and publishes `saga.advance`. The `next_fire_at` is advanced to the next schedule slot and `last_fired_at` is recorded. **Missed-fire behavior.** If all engine pods are down when a scheduled window elapses, the trigger fires once on the next tick after they come back. There is no backfill of missed windows. **License gate.** Cron triggers are gated by the `wf.cron_triggers` feature flag, checked at both create time (REST) and fire time (dispatcher). The `wf.cron_triggers` feature is not part of the verb `GroupToFeature` map; it is checked directly against the license resolver. **Management.** Cron triggers are created, listed, and deleted through the existing `/api/v1/triggers` REST endpoints. `POST /api/v1/triggers` with `trigger_type: cron` and a valid `config.schedule` initializes `next_fire_at` to the next schedule tick after the current time (so the first run fires on schedule, not immediately) and enables the trigger. **Deployment topology.** Because `ClaimCronFire` makes firing exactly-once across pods, running the cron dispatcher on every engine pod is safe and is the default. For operational isolation — predictable scheduling, resource isolation, and one obvious writer — the recommended production topology is **many engine pods for saga processing with the cron dispatcher disabled, plus a single dedicated cron pod with it enabled**. `cmd/engine` gates the dispatcher goroutine on the `WORKFLOW_CRON_DISPATCHER` env flag (`config.EngineConfig.EnableCronDispatcher`), so the same engine image serves both roles. The Helm chart exposes this as a single switch, `cron.dedicated`: when true a single-replica `-cron` Deployment runs with the flag on (`Recreate` strategy so two cron pods never overlap during a rollout) and the dispatcher is turned off on every main engine pod; when false (default) no dedicated pod is created and every engine pod runs cron (CAS-safe). There is no separate per-engine cron value — the chart derives the env flag for both Deployments from `cron.dedicated`. See the [deployment topology diagram](./deployment.md#cron-dispatcher-topology) for the rendered Kubernetes layout. Raising `cron.replicas` above 1 stays correct on the CAS, though true leader-elected single-fire (the `TimerAdvisoryLockID` advisory-lock stub in `engine/timer.go`) is not yet wired. ### Signals (`domain/signal.go`, `api/handler_signals.go`) A `SagaSignal` is an external message addressed to a specific run by name. `POST /api/v1/sagas/{run_id}/signal/{name}` appends the signal row, then calls `TryConsumeAwaitedSignal(run, name)`. If the run was paused awaiting exactly that signal, it consumes it (clearing markers, setting `wakeup_at=now`) and publishes `saga.advance` → `202`. If the run was not awaiting it → `409` (recorded but not matched). Signals wake `wait_for_signal`, and (indirectly) `manual_approval`/`collect_input`, which await the synthetic signal `user_task..submitted`. ### User tasks (`domain/user_task.go`, `api/handler_user_tasks.go`) A `UserTask` is created by `manual_approval` or `collect_input` and pauses the run awaiting its submission. `POST /api/v1/sagas/{run_id}/user_task/{task_id}/submit` (1) records `submitted_at`/`submitted_by`/`result` on the task, (2) appends a signal named `user_task..submitted` carrying the result as payload, (3) tries to consume the awaited signal and, on match, publishes `saga.advance` → `202`. This means user-task completion reuses the signal machinery exactly. --- ## 6. Package layout and stores ### Public importable packages The engine is structured as an embeddable library. The top-level packages form the importable surface for consuming applications: | Package | Purpose | |---------|---------| | `saga` | Facade: `saga.InMemory()` and `saga.New(saga.Options{...})` return `*saga.Saga` — the entry point for embedding the engine. | | `domain` | Core types: `WorkflowDefinition`, `SagaRun`, `Step`, `RuleDefinition`, `SagaSignal`, `UserTask`, `SagaTrigger`, etc. | | `engine` | `Coordinator`, `Timer`, `Advance` — the saga execution engine. | | `engine/verbs` | The 31 built-in step implementations (30 in the registry + `end`) plus `verbs.HandlerFunc` for custom verbs. | | `store` | The `Store` interface (see below) and `ErrNotFound`. | | `store/memory` | In-memory `Store` implementation (tests and embedded in-process use). | | `store/postgres` | Production Postgres `Store` + embedded SQL migrations. | | `licensing` | `Resolver` interface + `StubAllowAll` for dev/test. | | `secrets` | Secrets resolver interface used by HTTP/webhook verbs. | | `clock` | `Clock` interface (`SystemClock` + test stub). | | `api` | REST handlers, router (`api/router.go`), and the OpenAPI spec (`api/openapi.yaml`). | Infrastructure that backs the public interfaces but is not part of the importable surface lives under `internal/`: `internal/mq` (RabbitMQ topology, publisher, consumer), `internal/cel` (CEL evaluator), `internal/rules` (decision-table evaluation), `internal/grpc` (gRPC worker liveness server), `internal/config` (environment-variable config), `internal/logging`. Consumers should not import these directly. ### Embedding the engine An application can run the saga engine entirely in-process without a separate service or message broker. Import `saga` and call `saga.InMemory()` to get a `*saga.Saga` backed by an in-memory store; register `domain.WorkflowDefinition` values with `Register`, add custom verb handlers with `RegisterVerb`, and drive runs with `Start`, `Signal`, and `Get`. For production use, pass a Postgres store and other options to `saga.New(saga.Options{Store: pgStore, ...})`. The `cmd/api` and `cmd/engine` binaries are reference apps for service-mode deployment (Postgres + RabbitMQ), not a prerequisite for library use. **Lifecycle.** Workflows that fan out (`parallel`/`foreach`/`spawn_saga`) advance their child runs on background goroutines via an in-process publisher. These run on a context derived from `Options.Context` (default `context.Background()`) and are tracked so they can be drained. Call `sc.Shutdown(ctx)` to cancel that context (the engine's advance loop stops between steps) and wait for in-flight background advances to finish, bounded by the passed `ctx` (it returns `ctx.Err()` if the drain exceeds the deadline). Linear workflows advance synchronously inside `Start` and need no draining. ### The `Store` interface `store/store.go` defines the `Store` interface that both the engine and API depend on (neither depends on a concrete implementation). It covers: workflow + rule definitions, run CRUD + listing + stats, audit events, run-variable merges (dotted-key aware), pause/resume helpers (`SetPausedWithWakeup`, `SetPausedAwaitingSignal`, `SetPausedAwaitingEvent`, `ClearPause`, `WakeFromExternal`, `FindRunsByDueWakeup`, `FindRunsByAwaitedEvent`, `TryConsumeAwaitedSignal`, `AppendSignal`), child runs + try/catch stack (`SpawnChildRun`, `ListChildrenByParent`, `PushTryCatch`/`PopTryCatch`), user tasks, the action registry, saga triggers, and action-dispatch tracking (`MarkAwaitingAction`/`CompleteAction`/`FailAction`, which are no-ops if `attempt` doesn't match `current_attempt`, handling late deliveries). `ErrNotFound` is the standard not-found error. ### `store/memory` In-memory implementation (`store.go`, plus `triggers.go`). Used by unit/e2e tests and by `saga.InMemory()`. Note: in the memory store, `UpsertWorkflowDefinition` may mint a new UUID per call (acceptable because runs only need *some* `definition_id` pointer). ### `store/postgres` Production implementation, split by concern: `pool.go` (pgx pool / `Open`), `definitions.go`, `runs.go`, `events.go`, `rules.go`, `registry.go`, `triggers.go`, `user_tasks.go`, and `migrate.go`. `Open` connects; `Migrate(dsn)` applies embedded migrations on every boot via golang-migrate's pgx5 driver (idempotent, no-ops at head). The postgres `UpsertWorkflowDefinition` keeps a stable id per `(workflow_id, version)`. ### Migrations (`store/postgres/migrations`) Numbered `NNN_*.up.sql` / `.down.sql`, embedded in the binary. They establish three schemas — `definitions`, `runtime`, `audit` — and evolve them: - `001_init` — `workflow_definitions`, `rule_definitions`, `action_registry`; `saga_runs`, `saga_dlq_items`, `saga_trigger_fires`, `saga_signals`, `saga_user_tasks`; `saga_run_events` (audit, unique on `(run_id, step_id, attempt, event_type)`). - `002_add_wait_columns` — `wakeup_at`, `awaited_signal`, `awaited_event_topic`, `awaited_event_headers` on `saga_runs` + partial indexes for due-wakeup and awaited-topic lookups. - `003_child_runs_and_try_catch` — `parent_step_id`, `parent_branch_id`, `try_catch_stack` + parent index. - `004_license_groups` — license-group support. - `005_action_dispatch` — `awaited_action_dispatch`, `current_attempt` + partial index. - `006_saga_triggers` — trigger persistence. - `007_saga_event_notify` — a Postgres trigger on `audit.saga_run_events` that `pg_notify`s on channel `saga_event_`; this powers the WebSocket run inspector (`api/handler_stream.go`) via LISTEN/NOTIFY. What persists: workflow/rule/action definitions (`definitions.*`); live run state, inputs, variables, pause/await markers, parent links, try/catch stack (`runtime.saga_runs`); signals, user tasks, trigger-fire records, DLQ items (`runtime.*`); and the full append-only event log (`audit.saga_run_events`). --- ## 7. Messaging: RabbitMQ topology (`internal/mq`) `mq.DeclareTopology` (declared by the go-saga-orchestration processes, idempotent) sets up: - **Exchange `action.direct`** (direct, durable) — step dispatch. The `action` verb publishes `ActionPayload` here with routing key `.`. Workers declare their own per-service queue (`.actions`) bound with `.*` and consume it. - **Exchange `workflow.events`** (topic, durable) — inbound events. `EventSubscriber.RunRMQ` binds a per-pod queue with `#` to consume all events (auto-ack, fire-and-forget); each delivery feeds both the `EventSubscriber` (wake paused sagas) and the `TriggerDispatcher` (start new sagas). - **Queues `saga.advance`, `saga.dlq`, `action.dlq`** (durable). `saga.advance` is the core work queue. Publishing (`mq/publisher.go`): `PublishSagaAdvance(runID)` sends `{"saga_run_id": ...}` JSON to `saga.advance` via the default exchange (persistent). `PublishActionDispatch(routingKey, payload)` publishes to `action.direct`. A `Publisher` owns one channel (channels are not concurrency-safe). Consuming (`mq/consumer.go`): `ConsumeSagaAdvance` is a competing consumer on `saga.advance` with `prefetch=1` (fairness), manual ack. Per delivery: malformed JSON → `Reject(false)` (→ DLQ); handler error → `Nack(requeue=true)`; success → `Ack`. --- ## 8. Request-flow narrative (API → engine → store/MQ → completion; workers) **Start.** A client calls `POST /api/v1/sagas/start` (`api/handler_sagas.go`). The API resolves the published `WorkflowDefinition` for the workflow id + tenant, upserts it to obtain a `definition_id`, creates a `pending` `SagaRun` (carrying `dry_run` and any `X-Feature-Override` flags), injects startup variables via any registered `StartupVariableProvider`s (none ship by default), publishes `saga.advance`, and returns `202` with the `saga_run_id`. (Triggers reach the same state via `TriggerDispatcher`.) **Advance.** The engine's `saga.advance` consumer hands the message to `Coordinator.HandleAdvance` → `Advance`. The loop (see §2) re-reads the run, dispatches the current step's verb through the license gate, merges the verb result into Variables, emits audit events at each transition, and moves to the next step (or branch). It keeps looping in-process for as long as steps complete synchronously. **Pause / async work.** When a verb suspends the run it persists a pause marker and returns `ErrSagaPaused`; the coordinator emits `step.paused` and ACKs. The run resumes when something republishes `saga.advance`: - **Timer** (`engine.Timer`, running in `cmd/engine`) polls `FindRunsByDueWakeup` every tick and republishes advance for `wait_duration`/`wait_until` (and any `wakeup_at=now` set by `WakeFromExternal`). - **Signals / user tasks** — the REST handlers consume the awaited signal and publish advance. - **Events** — `EventSubscriber` matches `workflow.events` deliveries to runs awaiting a topic + header subset and publishes advance. - **Joins** — `checkParentJoin` wakes a parent after its children's join condition is met. **Workers via gRPC + `clients/go/worker`.** The `action` verb publishes an `ActionPayload` to `action.direct` and pauses the run awaiting the action. A worker built with `clients/go/worker`: 1. registers its actions over REST (`/api/v1/registry/register`), 2. declares + binds its `.actions` queue to `action.direct` and consumes it (`prefetch=1`, manual ack), 3. holds a long-lived gRPC client to the engine. On each delivery it deserializes the `ActionPayload`, resolves the handler by action name, and **drives the `ExecuteStep` bidi stream** (`internal/grpc/server.go`, proto `proto/liveness.proto`): worker sends `StartJob{run_id, step_id, attempt}` → engine replies `Acknowledged` → worker may stream `Heartbeat`s → worker sends `Complete{result_json}` or `Error{code, message, retryable}`. The engine bridges `Complete`→`store.CompleteAction` (merge result, then publish `saga.advance` to resume) and `Error`→`store.FailAction` (transition the run to failed; no advance). `CompleteAction`/`FailAction` are no-ops when `attempt` doesn't match the run's `current_attempt`, so late/duplicate deliveries are safe; combined with the worker's idempotency key and RabbitMQ redelivery, this gives at-least-once delivery with idempotent completion. **Completion.** When the loop reaches an `end` step (or a verb pushes the run terminal), `completeRun` emits `step.succeeded` + `run.succeeded`, sets state `succeeded`, and runs `checkParentJoin` so any waiting parent advances. A failed step with no try/catch frame sets state `failed` and likewise notifies the parent. Throughout, every transition appends an immutable `audit.saga_run_events` row, which (via the `010` NOTIFY trigger) streams live to any connected run-inspector WebSocket. --- ## Notable findings & ambiguities - **Retry is defined but not wired into `Advance`.** `engine/retry.go` provides a full backoff policy + default, but the coordinator's verb-error path does not re-attempt steps using it; on error it either jumps to a try/catch catch step or fails the run. Step retry is effectively delegated to workers (RabbitMQ redelivery + idempotency) and the `current_attempt` counter the `action` verb maintains. - **Engine does not start the event/trigger consumers in the committed code.** `cmd/engine/main.go` constructs `EventSubscriber`/`TriggerDispatcher` but leaves `sub.RunRMQ` commented out (`_ = sub`), with a note that prod-RMQ wiring is deferred. So in this build, `wait_for_event` and `record_transition` triggers will not fire until that goroutine is started. The timer dispatcher and the `saga.advance` consumer **are** started. - **Timer leader election is a no-op.** `Timer.AcquireLeaderLock` is documented as a stub; every engine pod runs the timer. With 2+ replicas this could double-publish `saga.advance` — harmless because `Advance` is idempotent (a premature advance on a still-paused run just ACKs), but worth knowing. - **`ValidateDefinition` is not called at publish time.** `engine/validate.go` implements structural checks (forbids `parallel` inside `try_catch`) and license-gate validation, but its own TODO notes the publish handler does not yet invoke it. Runtime is still gated by the per-step license check in `Advance`. - **`foreach` is parallel-only.** Sequential mode is explicitly rejected with guidance to use `while` + a counter. - **`license.StubAllowAll` in the engine binary** means license gates never reject in this deployment; the gate logic is fully present and exercised by tests, just disabled by the resolver `cmd/engine` wires. - **`try_catch` frames are never popped on success** — they remain on the stack until the saga terminates (documented as acceptable given the max-nesting-depth-3 rule the validator would enforce). --- # Benchmarks Source: https://bugs5382.github.io/go-saga-orchestration/docs/benchmarks # Benchmarks Benchmarks for the saga coordinator's hot path: step advancement (`Coordinator.Advance`) and verb dispatch. They exist to (1) establish a baseline and (2) guard the allocation-reduction work that follows. This page records the **baseline** captured before any tuning. This is issue #19. The baseline is below; the [After — PR2](#after--pr2-cel-program-cache) section records the tuning deltas. ## What is measured All benchmarks run against the in-memory store (`store/memory`) with a `SystemClock`, so they isolate the engine's own CPU and allocation cost. In service mode the dominant cost is store and message-queue I/O (Postgres, RabbitMQ); that latency is deliberately **out of scope** here — these numbers measure engine overhead, not deployed throughput. | Area | Benchmark | Package | |------|-----------|---------| | Step advancement (serial) | `BenchmarkAdvance` | `engine` | | Step advancement (concurrent) | `BenchmarkAdvanceParallel` | `engine` | | Registry lookup | `BenchmarkRegistryLookup` | `engine/verbs` | | Verb dispatch (per verb) | `BenchmarkVerbExecute` | `engine/verbs` | | Verb dispatch (concurrent) | `BenchmarkVerbExecuteParallel` | `engine/verbs` | | CEL env / compile / eval | `BenchmarkNewEnv`, `BenchmarkCompile`, `BenchmarkEval`, `BenchmarkNewEnvCompileEval`, `BenchmarkNewEnvParallel` | `internal/cel` | | Audit event creation | `BenchmarkNewEvent` | `domain` | ## How to run ```sh # All hot-path benchmarks with allocation stats. go test -run='^$' -bench=. -benchmem ./engine/... ./internal/cel/... ./domain/... # A single area. go test -run='^$' -bench=BenchmarkAdvance -benchmem ./engine/... ``` `allocs/op` and `B/op` are deterministic and are the primary signal for this work. `ns/op` varies with host load (these were captured on a shared machine); treat the timings as indicative, not absolute, and always compare old vs new on the same host in one sitting. ## Baseline Captured with `-benchtime=200ms`, `GOMAXPROCS=4`, Go 1.26, Intel Xeon Gold 6426Y, Linux/amd64. Timings are indicative; allocation columns are the stable baseline. ### `engine` — step advancement A single `Advance` call drives every synchronous step of a run, so `multi_step_N` is the cost of an N-step linear saga end to end. | Benchmark | ns/op | B/op | allocs/op | |-----------|------:|-----:|----------:| | `Advance/trivial` | 3,804 | 1,314 | 7 | | `Advance/single_verb` | 12,061 | 3,316 | 15 | | `Advance/multi_step_10` | 31,212 | 15,550 | 94 | | `Advance/multi_step_100` | 303,304 | 142,082 | 1,003 | | `AdvanceParallel/trivial` | 6,099 | 1,312 | 7 | | `AdvanceParallel/single_verb` | 8,798 | 3,328 | 15 | | `AdvanceParallel/multi_step_10` | 34,995 | 15,601 | 94 | | `AdvanceParallel/multi_step_100` | 301,307 | 142,090 | 1,003 | Per step the loop costs roughly **~9 allocs** (the delta between consecutive `multi_step` sizes is ~10 allocs/step), driven by audit-event creation and per-step state/variable writes through the store. ### `engine/verbs` — dispatch | Benchmark | ns/op | B/op | allocs/op | |-----------|------:|-----:|----------:| | `RegistryLookup` | 7.2 | 0 | 0 | | `VerbExecute/noop` | 42 | 48 | 1 | | `VerbExecute/set_var_literal` | 200 | 336 | 2 | | `VerbExecute/set_var_cel` | 113,260 | 68,444 | 1,021 | | `VerbExecute/transform` | 111,778 | 68,442 | 1,021 | | `VerbExecute/map_10` | 185,184 | 117,267 | 1,545 | | `VerbExecute/filter_10` | 205,035 | 120,935 | 1,610 | | `VerbExecute/map_100` | 218,094 | 153,130 | 1,728 | | `VerbExecute/filter_100` | 235,462 | 156,798 | 1,793 | | `VerbExecute/decision` | 103,523 | 59,399 | 807 | | `VerbExecute/parallel_2` | 8,734 | 2,854 | 54 | | `VerbExecute/parallel_4` | 18,403 | 5,735 | 108 | | `VerbExecuteParallel/set_var_cel` | 65,044 | 68,447 | 1,021 | | `VerbExecuteParallel/transform` | 66,860 | 68,445 | 1,021 | | `VerbExecuteParallel/map_100` | 146,907 | 153,132 | 1,728 | The registry lookup and literal `set_var` are effectively free. **Every CEL-bearing verb is ~500x more expensive** — `set_var_cel` and `transform` each cost ~1,021 allocs/op even though they evaluate a trivial expression. ### `internal/cel` — expression primitives | Benchmark | ns/op | B/op | allocs/op | |-----------|------:|-----:|----------:| | `NewEnv/vars_0` | 24,038 | 20,837 | 261 | | `NewEnv/vars_5` | 25,002 | 21,518 | 272 | | `NewEnv/vars_20` | 24,447 | 23,615 | 305 | | `Compile` | 92,978 | 51,720 | 1,020 | | `Eval` | 245 | 0 | 0 | | `NewEnvCompileEval` | 133,653 | 82,146 | 1,331 | | `NewEnvParallel` | 14,659 | 21,518 | 272 | This isolates the headline finding: a compiled program **evaluates in 245 ns with zero allocations**, but the verbs rebuild the environment and recompile the expression on every single dispatch (`NewEnvCompileEval`: ~1,331 allocs/op). `NewEnv` and `Compile` together account for essentially all of the allocation cost seen in the CEL verbs above. ### `domain` — audit events | Benchmark | ns/op | B/op | allocs/op | |-----------|------:|-----:|----------:| | `NewEvent` | 490 | 16 | 1 | `NewEvent`'s cost is a `uuid.New()` plus a `time.Now()`; the hot loop emits two to three per step. The UUID and timestamp are audit-critical, so this is a documented floor rather than a tuning target. ## After — PR2 (CEL program cache) The tuning adds `cel.CompiledProgram`, a thread-safe cache that memoises the compiled CEL program for a given (declared variable set, expression) pair, so the verbs no longer rebuild and recompile the environment on every dispatch. The map/filter verbs additionally reuse a single activation map across elements instead of cloning `run.Variables` per element. Allocation reductions on the CEL-bearing verbs, measured with `benchstat old.txt new.txt` over `-count=10 -benchtime=100ms` runs taken back to back on the same host: | Benchmark | allocs/op (before → after) | B/op (before → after) | |-----------|:--------------------------:|:---------------------:| | `VerbExecute/set_var_cel` | 1,021 → 6 (−99.4%) | 68,444 → 456 (−99.3%) | | `VerbExecute/transform` | 1,021 → 6 (−99.4%) | 68,444 → 456 (−99.3%) | | `VerbExecute/map_10` | 1,545 → 21 (−98.6%) | 114.5Ki → 1.58Ki (−98.6%) | | `VerbExecute/map_100` | 1,728 → 24 (−98.6%) | 149.5Ki → 7.05Ki (−95.3%) | | `VerbExecute/filter_10` | 1,610 → 21 (−98.7%) | 118.1Ki → 1.58Ki (−98.7%) | | `VerbExecute/filter_100` | 1,793 → 24 (−98.7%) | 153.1Ki → 7.05Ki (−95.4%) | | `VerbExecute/decision` | 807 → 9 (−98.9%) | 57.9Ki → 1.34Ki (−97.7%) | | `VerbExecuteParallel/set_var_cel` | 1,021 → 6 (−99.4%) | 68,445 → 456 (−99.3%) | | `VerbExecuteParallel/transform` | 1,021 → 6 (−99.4%) | 68,442 → 456 (−99.3%) | | `VerbExecuteParallel/map_100` | 1,728 → 24 (−98.6%) | 149.5Ki → 7.05Ki (−95.3%) | | **`engine/verbs` geomean** | **323 → 12 (−96.4%)** | **23.2Ki → 1.19Ki (−94.9%)** | Wall-clock falls in step with the allocations once the cache is warm (`set_var_cel` ~136µs → ~0.8µs, the `engine/verbs` sec/op geomean −94%), but timings vary with host load — the allocation columns are the authoritative result. What is intentionally unchanged: - **`Advance/*`** uses literal `set_var` (no CEL), so its allocs/op are identical before and after; the small sec/op wobble is host noise. - **`VerbExecute/parallel_*`** passes literal branch lists (not a CEL string), so its path is untouched (54 / 108 allocs/op unchanged). - **`internal/cel` `NewEnv` / `Compile` / `NewEnvCompileEval`** call the raw primitives directly and remain the reference cost of an *uncached* build — they show what the cache now avoids. - **`NewEvent`** is unchanged: its UUID + timestamp are audit-critical, so it stays a documented floor rather than a tuning target. ## Comparing runs (benchstat) ```sh go test -run='^$' -bench=. -benchmem -count=10 ./engine/... ./internal/cel/... ./domain/... > old.txt # ...make a change... go test -run='^$' -bench=. -benchmem -count=10 ./engine/... ./internal/cel/... ./domain/... > new.txt benchstat old.txt new.txt ``` Run `old.txt` and `new.txt` back to back on the same idle host so the `ns/op` comparison is meaningful; the `allocs/op` delta is reliable regardless. --- # Caveats & Gotchas Source: https://bugs5382.github.io/go-saga-orchestration/docs/caveats # ⚠️ Caveats & Gotchas A friendly list of things that trip people up. Each item has a one-line workaround where relevant. --- - **One verb per step, no inline composition.** A step executes exactly one verb. You cannot chain a CEL transform and an HTTP call in the same step. _Workaround: add a second step._ - **`action` has no `out_var` — the worker controls output keys.** Whatever keys the worker returns are merged directly into `Variables`. If two `action` steps return overlapping keys, the later step's values overwrite the earlier ones. _Workaround: prefix output keys in your worker, or use a `set_var`/`transform` step immediately after to rename them._ - **CEL verbs read `Variables`, not `step.Inputs`.** `transform`, `filter`, `map`, `switch`, `while`, `assert`, and similar verbs compile their `expr` against `run.Variables`. If the value you need is in the run's initial inputs and not yet in `Variables`, it is not automatically visible. _Workaround: add a `set_var` step at the start of your workflow to promote input values into named variables._ - **Embedded `action` steps need a worker or service mode.** `saga.InMemory()` dispatches `action` steps to its in-process publisher, which pauses the saga. Without a registered worker goroutine to reply, the run will pause indefinitely. _Workaround: use `RegisterVerb` for in-process handlers, or run `cmd/engine` + a gRPC worker for true worker round-trips._ - 💡 **Embedded `emit_event` matches in-process (no broker).** The in-process `EventEmitter` both wakes runs awaiting the topic (header-subset match) **and** runs the trigger dispatcher, so matching `record_transition` triggers start new runs — parity with service mode. (Payload-CEL trigger matching beyond `record_transition` is not implemented in either mode yet.) - 💡 **Waits support `timeout_s`.** Both `wait_for_signal` and `wait_for_event` accept an optional `timeout_s`; on timeout the run routes to the step's `timeout` branch if defined, else to `next`. (`wait_duration`/`wait_until` are scheduled waits — their firing *is* the intended path, so no timeout branch.) - **Cancelling a parallel child re-checks the parent join.** `cancel` on a child that is a branch of a `parallel` join immediately re-evaluates the parent's join and wakes the parent if it is now satisfied (e.g. the cancelled child was the last non-terminal branch, or a `quorum` is met). Under `join_strategy: "all"` with other branches still running, the parent correctly stays paused until they finish. _For partial completion, use `join_strategy: "quorum"`._ - **`ValidateDefinition` is not auto-called by the REST publish path.** The engine provides `engine.ValidateDefinition(def)` to catch structural problems (missing steps, circular references, `parallel`-inside-`try_catch`, excessive nesting depth) before a workflow goes live. The REST `PUT /api/v1/workflows` endpoint does not call it automatically. _Workaround: call `ValidateDefinition` in your CI pipeline or publishing tooling before pushing definitions to production._ - **The module path is internal.** The Go module is hosted at `github.com/Bugs5382/go-saga-orchestration`. It is not published to the public Go module proxy. _Workaround: add a `GONOSUMCHECK` / `GONOSUMDB` / `GOFLAGS` directive in your environment, or vendor the module, per your organisation's private module setup._ --- # Deployment Source: https://bugs5382.github.io/go-saga-orchestration/docs/deployment # Deployment go-saga-orchestration ships two stateless services — the HTTP **api** and the gRPC **engine** — as container images, plus a Helm chart attached to each release. ## Container images Published to GHCR on every release (multi-arch amd64/arm64): - `ghcr.io/bugs5382/go-saga-orchestration/api` - `ghcr.io/bugs5382/go-saga-orchestration/engine` Tags: `vX.Y.Z`, `latest`, and the commit SHA. ## Prerequisites Both services require, at runtime: - A reachable **RabbitMQ** (`RABBITMQ_URL`) — both the api and the engine connect on startup and exit if it is unavailable. - A **store**: `postgres` (default), `redis`/`valkey`, or `memory` (single-process, dev only). postgres needs `DATABASE_DSN`; redis/valkey needs `REDIS_URL`. Supply the connection strings through a Secret and point the chart at it: ```bash kubectl create secret generic gosaga-conn \ --from-literal=rabbitmq-url='amqp://user:pass@rabbitmq:5672/' \ --from-literal=database-dsn='postgres://user:pass@postgres:5432/saga?sslmode=require' ``` ## Install the Helm chart The chart is attached to each GitHub Release as `go-saga-orchestration-.tgz`. Replace `` below with the release you want (see the [Releases](https://github.com/Bugs5382/go-saga-orchestration/releases) page): ```bash helm install go-saga \ https://github.com/Bugs5382/go-saga-orchestration/releases/download/v/go-saga-orchestration-.tgz \ --set store.type=postgres \ --set connectionSecret=gosaga-conn ``` ## Configuration | Value | Default | Description | |-------|---------|-------------| | `store.type` | `postgres` | `postgres` / `redis` / `valkey` / `memory` (→ `STORE_TYPE`) | | `connectionSecret` | `""` | Secret with `rabbitmq-url` (required) + `database-dsn` or `redis-url` | | `api.replicas` / `engine.replicas` | `1` | Replica counts (both stateless) | | `api.port` | `8080` | API HTTP port (`WORKFLOW_API_PORT`) | | `engine.grpcPort` | `9090` | Engine gRPC port (`WORKFLOW_ENGINE_GRPC_PORT`) | | `cron.dedicated` | `false` | Run the cron dispatcher on a dedicated pod and off the main engines (see below) | | `cron.replicas` | `1` | Replicas for the dedicated cron pod (only when `cron.dedicated`) | | `ingress.enabled` | `false` | Expose the api via Ingress | See `deployments/helm/values.yaml` for the full set, including probes, resources, and security context. ## Cron dispatcher topology The engine's cron dispatcher polls for due cron triggers and fires them. Firing is **exactly-once across every engine pod** — each fire is claimed with a compare-and-swap on the trigger's `next_fire_at` (`ClaimCronFire`), so even if many pods run the loop at once, only one wins each window. Running cron on every engine pod is therefore always safe. A single Helm switch, `cron.dedicated`, picks the topology: - **`cron.dedicated: false` (default)** — no extra pod. Every main engine pod runs the cron dispatcher (`WORKFLOW_CRON_DISPATCHER=true`); the CAS keeps firing single-shot. Simplest single-deployment setup. - **`cron.dedicated: true`** — one dedicated single-replica cron pod runs the dispatcher, and it is turned **off** on every (scaled) main engine pod. This gives predictable scheduling, resource isolation, and one obvious cron writer while the main engines scale out purely for saga processing. The cron pod reuses the engine image, ConfigMap, and connection Secret. Enabling the dedicated pod is the only switch — there is no separate per-engine cron flag; the chart derives `WORKFLOW_CRON_DISPATCHER` for both deployments from it. ```mermaid flowchart TB client([Clients]) --> api subgraph k8s["Kubernetes (cron.dedicated: true)"] api["api Deployment\nN replicas\nHTTP :8080"] engine["engine Deployment\nN replicas\ncron OFF\n(saga processing)"] cron["cron Deployment\n1 replica\ncron ON\n(WORKFLOW_CRON_DISPATCHER=true)"] end store[("Store\n(Postgres / Redis)")] rmq[["RabbitMQ"]] api --> store api --> rmq engine --> store engine --> rmq cron --> store cron --> rmq cron -. "fires cron triggers\n(single writer)" .-> rmq ``` Even with `cron.dedicated: true` you may raise `cron.replicas` above 1 for availability — the `ClaimCronFire` CAS still guarantees each window fires once, so duplicate cron pods do not double-fire. --- # gRPC Worker Protocol Source: https://bugs5382.github.io/go-saga-orchestration/docs/grpc # gRPC Worker Protocol External workers report progress on dispatched saga actions to the engine over a gRPC stream. This document describes that wire protocol for SDK authors and integrators. The schema source of truth is [`proto/liveness.proto`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/proto/liveness.proto) (proto package `saga.v1`, Go package `livenesspb`). This doc explains it; the `.proto` file remains authoritative for exact wire layout. ## Endpoint The engine (`cmd/engine`) serves gRPC on **`:9090`** by default (`WORKFLOW_ENGINE_GRPC_PORT`, see `internal/config/config.go`). The transport is plaintext (`insecure` credentials) — TLS is expected to be terminated by the service mesh / ingress, not by the engine itself. A worker's gRPC address is configured via `BootstrapConfig.GrpcURL`, e.g. `go-saga-orchestration-engine.platform.svc.cluster.local:9090`. ## Service: `WorkerLiveness` ```proto service WorkerLiveness { rpc ExecuteStep(stream WorkerEvent) returns (stream EngineEvent); } ``` `ExecuteStep` is a **bidirectional stream**. The worker is the client: it opens one stream per action execution, sends `WorkerEvent` messages, and receives `EngineEvent` messages. The full method name is `/saga.v1.WorkerLiveness/ExecuteStep`. ## Messages ### `WorkerEvent` (worker → engine) A `oneof event` carrying exactly one of: | Variant | Type | Meaning | |---|---|---| | `start` | `StartJob` | Opens the job. Must be the first message. | | `heartbeat` | `Heartbeat` | Optional liveness/progress signal while executing. | | `complete` | `Complete` | Terminal success. | | `error` | `Error` | Terminal failure. | **`StartJob`** - `run_id` (string) — saga run UUID. - `step_id` (string) — the step/action being executed. - `attempt` (int32) — attempt number (matches the dispatched `ActionPayload.attempt`). **`Heartbeat`** - `progress_pct` (int32) — 0–100 progress hint. - `note` (string) — free-text status. Heartbeats cause no state change today; the engine logs them at debug level. They exist as a hook for long-action timeout extension. **`Complete`** - `result_json` (bytes) — JSON object merged into the saga's variables on success. - `would_change_json` (bytes) — if non-empty, marks this as a dry-run preview (structured "what would change" rather than an applied side effect). **`Error`** - `code` (string) — stable error code. - `message` (string) — human-readable detail. - `retryable` (bool) — whether the engine should allow a retry. ### `EngineEvent` (engine → worker) A `oneof event` carrying exactly one of: | Variant | Type | Meaning | |---|---|---| | `ack` | `Acknowledged` | Engine accepted the `StartJob`. | | `cancel` | `CancelRequested` | Engine asks the worker to abandon the job. | **`Acknowledged`** — empty message. **`CancelRequested`** — `reason` (string). (Defined in the schema; the current engine implementation does not yet emit it.) ## Lifecycle / handshake One `ExecuteStep` stream maps to one action execution. The frame protocol (enforced server-side in `internal/grpc/server.go`): 1. **Worker → `StartJob`** with `run_id`, `step_id`, `attempt`. A second `StartJob` on the same stream is rejected (`duplicate StartJob`). 2. **Engine → `Acknowledged`**. 3. **Worker → `Heartbeat`** (zero or more, optional) while the handler runs. 4. **Worker → `Complete`** (success) **or `Error`** (failure). This is terminal. Sending `Complete`/`Error` before `StartJob` is rejected (`complete/error without start`). 5. **Engine** processes the terminal message and ends the stream (returns from the RPC). The worker then closes its send side. On the terminal message the engine: - `Complete` → parses `run_id` (UUID), JSON-decodes `result_json` (a non-JSON body is preserved under `_raw_result`), calls `store.CompleteAction(run, attempt, result)`, then publishes `saga.advance` via the `AdvancePublisher` to wake the paused saga. - `Error` → calls `store.FailAction(run, attempt, code, message, retryable)`, which transitions the run to failed. No advance is published. ### Failure / disconnect semantics Errors at the gRPC layer itself (network drop, decode failures, EOF before a terminal message) leave the saga in its awaiting-action state. They are **not** treated as action failures. Recovery comes from RabbitMQ redelivery of the dispatch plus the worker's idempotency wrapper — not from the gRPC stream. A handler error is reported as an `Error` message (engine fails the action), which is distinct from a transport error (stream just breaks and the delivery is retried). ## How the Go worker runtime drives it The reference client lives in `clients/go/worker`. Workers don't talk to the proto directly; they register `Handler`s and the runtime drives the stream. Flow (`runtime.go`): 1. `Bootstrap` registers actions over REST, declares a per-service RabbitMQ queue (`.actions`), and opens one long-lived gRPC client (`pb.NewWorkerLivenessClient`) to `GrpcURL`. 2. For each RabbitMQ delivery, `processDelivery` decodes an `ActionPayload` (`run_id`, `step_id`, `attempt`, `action`, `inputs`, `dry_run`), resolves the handler by the action-name suffix, and calls `driveStream`. 3. `driveStream` opens an `ExecuteStep` stream and performs the handshake: sends `StartJob`, waits for the `Acknowledged`, runs `Handler.Execute`, then sends `Complete{result_json}` on success or `Error{code, message, retryable}` on failure (codes come from errors implementing the `coded` interface, e.g. `worker.Errorf`). It always `CloseSend`s the stream. 4. RabbitMQ ack policy: handler/engine success → `Ack`; a transport-level stream failure → `Nack` with requeue (retry via redelivery); an undecodable payload or unknown action → `Nack` to the DLQ. A handler `Error` is reported on the stream but is *not* a transport failure, so the delivery is acked. In v1, the runtime sends no `Heartbeat`s and does not yet act on `CancelRequested`; those parts of the schema are forward-looking. --- # api Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/api # api ```go import "github.com/Bugs5382/go-saga-orchestration/api" ``` Package api — WebSocket stream handler for the run inspector. Auth is intentionally not enforced here; real authentication middleware should be added before production deployment. See the route comment in router.go. Package api wires the go\-saga\-orchestration REST API. Routes: health probes \+ /api/v1/sagas. ## Constants Error codes used in the JSON error envelope \(the \`error\` field written by WriteError\). Stable, machine\-readable identifiers shared across handlers. ```go const ( CodeBadRequest = "bad_request" CodeNotFound = "not_found" CodeForbidden = "forbidden" CodeInternal = "internal" CodeInvalidConfig = "invalid_config" CodePublishFailed = "publish_failed" CodeUnprocessable = "unprocessable" CodeConflict = "conflict" ) ``` ## func [HealthLive]() ```go func HealthLive(w http.ResponseWriter, _ *http.Request) ``` HealthLive returns 200 if the process is running. ## func [HealthReady]() ```go func HealthReady(w http.ResponseWriter, _ *http.Request) ``` HealthReady returns 200 once the service is ready to serve. v1 simple: always ready after process start. Future: gate on Postgres ping \+ a "registry warm" signal. ## func [NewRouter]() ```go func NewRouter(_ store.Store, sagas *SagaHandler, signals *SignalHandler, userTasks *UserTaskHandler, registryHandler *RegistryHandler, rulesHandler *RulesHandler, triggersHandler *TriggerHandler, streamHandler *SagaStreamHandler, workflows *WorkflowHandler, actionResults *ActionResultHandler) *chi.Mux ``` NewRouter builds the chi router. Saga routes attach via SagaHandler; registry routes attach via RegistryHandler; rule evaluation via RulesHandler; trigger CRUD via TriggerHandler; live run inspector via SagaStreamHandler; workflow stats via WorkflowHandler. ## func [WriteError]() ```go func WriteError(w http.ResponseWriter, status int, code, message string) ``` WriteError sends a structured error envelope. ## func [WriteJSON]() ```go func WriteJSON(w http.ResponseWriter, status int, v any) ``` WriteJSON serialises v as JSON with the given status. Errors during encoding are written to the response but cannot change the already\-sent header. ## type [ActionResultHandler]() ActionResultHandler accepts a transport\-agnostic, asynchronous result callback for an action step. gRPC workers reply over the ExecuteStep stream; http and rmq workers have no return stream, so they report their result here. Both paths converge on the same CompleteAction / FailAction store hooks the gRPC server uses, preserving attempt handling and idempotency. \(issue \#59\) ```go type ActionResultHandler struct { S store.Store Publisher AdvancePublisher } ``` ### func [NewActionResultHandler]() ```go func NewActionResultHandler(s store.Store, p AdvancePublisher) *ActionResultHandler ``` NewActionResultHandler returns an ActionResultHandler backed by the given store and advance publisher. ### func \(\*ActionResultHandler\) [Post]() ```go func (h *ActionResultHandler) Post(w http.ResponseWriter, r *http.Request) ``` Post handles POST /api/v1/sagas/\{run\_id\}/actions/\{step\_id\}/result. Body \(exactly one of\): ``` { "result": { ... } } -> CompleteAction { "error": { "code", "message", "retryable" } } -> FailAction ``` An optional "attempt" pins the report to a specific dispatch attempt; omitted, it uses the run's current attempt. Mirrors the gRPC Complete/Error semantics \(attempt handling \+ idempotency are enforced by the store hooks: a stale attempt is a no\-op\). Responses: ``` 202 — result recorded (success completed, or failure transitioned). 400 — bad run_id or malformed body. 500 — internal error. ``` ## type [AdvancePublisher]() AdvancePublisher abstracts the RabbitMQ publisher so handlers can be unit\-tested without a broker. ```go type AdvancePublisher interface { PublishSagaAdvance(ctx context.Context, runID string) error } ``` ## type [Canceller]() Canceller cancels an in\-flight run from outside the run. It is satisfied by \*engine.Coordinator; handlers depend on the interface so they can be unit\-tested with a fake. See engine.Coordinator.Cancel. ```go type Canceller interface { Cancel(ctx context.Context, runID uuid.UUID, reason string) error } ``` ## type [RegistryHandler]() RegistryHandler serves the action\-registry REST surface. ``` POST /api/v1/registry/register — services call this on startup to register or refresh their action declarations. GET /api/v1/registry/actions — read-only browser feed for admin/builder UIs. ``` ```go type RegistryHandler struct { S store.Store } ``` ### func [NewRegistryHandler]() ```go func NewRegistryHandler(s store.Store) *RegistryHandler ``` NewRegistryHandler returns a RegistryHandler backed by the given store. ### func \(\*RegistryHandler\) [List]() ```go func (h *RegistryHandler) List(w http.ResponseWriter, r *http.Request) ``` List returns the registered actions filtered by service/category/search. ``` GET /api/v1/registry/actions?service=&category=&search= ``` ### func \(\*RegistryHandler\) [Register]() ```go func (h *RegistryHandler) Register(w http.ResponseWriter, r *http.Request) ``` Register accepts a service's startup payload. Upserts each action by \(service, action\_name, version\). Idempotent — services may resend on every restart. ## type [RulesHandler]() RulesHandler exposes the rules\-package evaluator as a sync REST API. ```go type RulesHandler struct { S store.Store } ``` ### func [NewRulesHandler]() ```go func NewRulesHandler(s store.Store) *RulesHandler ``` NewRulesHandler returns a RulesHandler backed by the given store. ### func \(\*RulesHandler\) [Evaluate]() ```go func (h *RulesHandler) Evaluate(w http.ResponseWriter, r *http.Request) ``` Evaluate handles POST /api/v1/rules/\{rule\_id\}/evaluate. Responses: ``` 200 — rule found + evaluated; body {output, audit}. 404 — rule_id not found. 400 — bad request body. 422 — rule evaluation failed (CEL compile/eval error, no_decision_row_matched). 500 — internal error. ``` ## type [SagaHandler]() SagaHandler owns the /api/v1/sagas/\* routes. ```go type SagaHandler struct { // contains filtered or unexported fields } ``` ### func [NewSagaHandler]() ```go func NewSagaHandler(s store.Store, p AdvancePublisher, providers ...engine.StartupVariableProvider) *SagaHandler ``` NewSagaHandler constructs the handler. Optional StartupVariableProviders are invoked at saga start to inject per\-tenant "magic" variables. ### func \(\*SagaHandler\) [Cancel]() ```go func (h *SagaHandler) Cancel(w http.ResponseWriter, r *http.Request) ``` Cancel handles POST /api/v1/sagas/\{id\}/cancel. It routes to the coordinator's idempotent run\-level cancel: a terminal run is a no\-op and a cancelled child re\-evaluates its parent's join. Returns 202 on success, 400 on a bad id, 404 when the run does not exist, and 501 when no canceller is wired. ### func \(\*SagaHandler\) [Get]() ```go func (h *SagaHandler) Get(w http.ResponseWriter, r *http.Request) ``` Get handles GET /api/v1/sagas/\{id\}. ### func \(\*SagaHandler\) [List]() ```go func (h *SagaHandler) List(w http.ResponseWriter, r *http.Request) ``` List handles GET /api/v1/sagas. Parses optional filter query params, calls store.ListRuns \+ store.CountRuns, returns paginated JSON. ### func \(\*SagaHandler\) [Start]() ```go func (h *SagaHandler) Start(w http.ResponseWriter, r *http.Request) ``` Start handles POST /api/v1/sagas/start. Resolves the published definition, inserts a saga\_runs row, publishes saga.advance, returns 202 with the run ID. ### func \(\*SagaHandler\) [WithCanceller]() ```go func (h *SagaHandler) WithCanceller(c Canceller) *SagaHandler ``` WithCanceller attaches the Canceller used by the Cancel handler and returns the receiver for chaining. When unset, POST /sagas/\{id\}/cancel returns 501. ## type [SagaStreamHandler]() SagaStreamHandler upgrades GET /api/v1/sagas/\{run\_id\}/stream to a WebSocket. On connect: sends the current SagaRun snapshot, then tails audit.saga\_run\_events via Postgres LISTEN/NOTIFY \(channel "saga\_event\_\"\). One LISTEN per connection — the per\-connection pgx Conn is acquired from a dedicated channel pool the handler holds. Auth is intentionally not enforced today; wire real auth middleware before exposing this endpoint in production. ```go type SagaStreamHandler struct { S store.Store Pool *pgxpool.Pool // for LISTEN — acquire a dedicated conn per stream Upgrade websocket.Upgrader } ``` ### func [NewSagaStreamHandler]() ```go func NewSagaStreamHandler(s store.Store, pool *pgxpool.Pool) *SagaStreamHandler ``` NewSagaStreamHandler constructs the handler. ### func \(\*SagaStreamHandler\) [Stream]() ```go func (h *SagaStreamHandler) Stream(w http.ResponseWriter, r *http.Request) ``` Stream handles GET /api/v1/sagas/\{run\_id\}/stream. ## type [SignalHandler]() SignalHandler accepts external signals delivered to a saga. ```go type SignalHandler struct { S store.Store Publisher AdvancePublisher } ``` ### func [NewSignalHandler]() ```go func NewSignalHandler(s store.Store, p AdvancePublisher) *SignalHandler ``` NewSignalHandler returns a SignalHandler backed by the given store and advance publisher. ### func \(\*SignalHandler\) [Post]() ```go func (h *SignalHandler) Post(w http.ResponseWriter, r *http.Request) ``` Post handles POST /api/v1/sagas/\{run\_id\}/signal/\{name\}. Responses: ``` 202 — signal recorded AND matched a paused saga awaiting it (advance published). 409 — signal recorded but the run wasn't paused-and-awaiting this name. 400 — bad run_id. 404 — run not found (only when AppendSignal returns a not-found error). 500 — internal error. ``` ## type [TriggerHandler]() TriggerHandler serves the saga\-trigger REST surface. ``` POST /api/v1/triggers — create GET /api/v1/triggers — list (optional ?type= ?enabled=) GET /api/v1/triggers/{id} — get one DELETE /api/v1/triggers/{id} — remove ``` ```go type TriggerHandler struct { S store.Store Licensing licensing.Resolver Clock clock.Clock } ``` ### func [NewTriggerHandler]() ```go func NewTriggerHandler(s store.Store, lr licensing.Resolver, clk clock.Clock) *TriggerHandler ``` NewTriggerHandler constructs the handler. ### func \(\*TriggerHandler\) [Create]() ```go func (h *TriggerHandler) Create(w http.ResponseWriter, r *http.Request) ``` Create handles POST /api/v1/triggers. ### func \(\*TriggerHandler\) [Delete]() ```go func (h *TriggerHandler) Delete(w http.ResponseWriter, r *http.Request) ``` Delete handles DELETE /api/v1/triggers/\{id\}. Returns 204 on success, 404 if no row exists \(mirrors the non\-idempotent pattern — store.DeleteTrigger returns ErrNotFound for missing rows\). ### func \(\*TriggerHandler\) [Get]() ```go func (h *TriggerHandler) Get(w http.ResponseWriter, r *http.Request) ``` Get handles GET /api/v1/triggers/\{id\}. ### func \(\*TriggerHandler\) [List]() ```go func (h *TriggerHandler) List(w http.ResponseWriter, r *http.Request) ``` List handles GET /api/v1/triggers. Optional query params: ?type= ?enabled=true|false ## type [UserTaskHandler]() UserTaskHandler accepts task submissions from assignees. On submit: 1. Persist the task's submitted\_at / submitted\_by / result. 2. Append a saga\_signal of name \`user\_task.\{task\_id\}.submitted\` to the run, carrying the result as the signal payload. 3. Try to consume the awaited signal; if it matches, publish saga.advance. ```go type UserTaskHandler struct { S store.Store Publisher AdvancePublisher } ``` ### func [NewUserTaskHandler]() ```go func NewUserTaskHandler(s store.Store, p AdvancePublisher) *UserTaskHandler ``` NewUserTaskHandler constructs the handler. ### func \(\*UserTaskHandler\) [Submit]() ```go func (h *UserTaskHandler) Submit(w http.ResponseWriter, r *http.Request) ``` Submit handles POST /api/v1/sagas/\{run\_id\}/user\_task/\{task\_id\}/submit. ## type [WorkflowHandler]() WorkflowHandler owns workflow\-level aggregate routes. ```go type WorkflowHandler struct { S store.Store } ``` ### func [NewWorkflowHandler]() ```go func NewWorkflowHandler(s store.Store) *WorkflowHandler ``` NewWorkflowHandler constructs a WorkflowHandler. ### func \(\*WorkflowHandler\) [Stats]() ```go func (h *WorkflowHandler) Stats(w http.ResponseWriter, r *http.Request) ``` Stats handles GET /api/v1/workflows/\{wf\_id\}/stats. Returns aggregate metrics: success\_rate\_24h, last\_run\_at, in\_flight. Generated by [gomarkdoc]() --- # clock Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/clock # clock ```go import "github.com/Bugs5382/go-saga-orchestration/clock" ``` Package clock abstracts time so the engine \+ verbs can be tested without real wall\-clock delays. Production uses SystemClock; tests use FakeClock which advances on demand. ## type [Clock]() Clock is the abstraction injected into the coordinator \+ wait verbs. ```go type Clock interface { Now() time.Time // After returns a channel that receives the current time after d // elapses. FakeClock receives only when Advance(d) is called. After(d time.Duration) <-chan time.Time } ``` ## type [FakeClock]() FakeClock holds a virtual clock that advances on demand. ```go type FakeClock struct { // contains filtered or unexported fields } ``` ### func [NewFakeClock]() ```go func NewFakeClock(start time.Time) *FakeClock ``` NewFakeClock starts at the given instant \(use UTC\). ### func \(\*FakeClock\) [Advance]() ```go func (f *FakeClock) Advance(d time.Duration) ``` Advance moves the clock forward and fires any waiters whose deadline is at or before the new time. ### func \(\*FakeClock\) [After]() ```go func (f *FakeClock) After(d time.Duration) <-chan time.Time ``` After registers a waiter that fires only when Advance moves the virtual clock to at or past now\+d. ### func \(\*FakeClock\) [Now]() ```go func (f *FakeClock) Now() time.Time ``` Now returns the current virtual time. ## type [SystemClock]() SystemClock delegates to the stdlib time package. ```go type SystemClock struct{} ``` ### func \(SystemClock\) [After]() ```go func (SystemClock) After(d time.Duration) <-chan time.Time ``` After returns a channel that fires after d elapses, delegating to time.After. ### func \(SystemClock\) [Now]() ```go func (SystemClock) Now() time.Time ``` Now returns the current UTC wall\-clock time. Generated by [gomarkdoc]() --- # domain Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/domain # domain ```go import "github.com/Bugs5382/go-saga-orchestration/domain" ``` Package domain contains the in\-process types go\-saga\-orchestration exchanges with its HTTP layer, store, and engine. Definitions and runs are the core concepts. Package domain contains the core domain types for go\-saga\-orchestration. ## Constants Dispatch transport values for ActionRegistration.Transport. ```go const ( // TransportGRPC is the zero-config default: the worker is reachable over // the gRPC ExecuteStep stream. Equivalent to an empty Transport. TransportGRPC = "grpc" // TransportHTTP dispatches the action by POSTing the payload to Address. TransportHTTP = "http" // TransportRMQ dispatches the action by publishing the payload to the // RabbitMQ queue named by Address. TransportRMQ = "rmq" ) ``` ## type [ActionRegistration]() ActionRegistration describes one action a service exposes. Stored in definitions.action\_registry. ```go type ActionRegistration struct { ID uuid.UUID `json:"id"` Service string `json:"service"` ActionName string `json:"action_name"` Version int `json:"version"` Description string `json:"description,omitempty"` Category string `json:"category,omitempty"` Compensable bool `json:"compensable"` InputSchema map[string]any `json:"input_schema"` OutputSchema map[string]any `json:"output_schema"` ErrorCodes []string `json:"error_codes,omitempty"` DefaultRetry *RetryPolicy `json:"default_retry,omitempty"` DefaultTimeoutMS int `json:"default_timeout_ms,omitempty"` Deprecated bool `json:"deprecated"` RegisteredAt time.Time `json:"registered_at"` ServiceVersion string `json:"service_version,omitempty"` DryRunSupported bool `json:"dry_run_supported"` LicenseGroup string `json:"license_group,omitempty"` // Transport is the optional dispatch descriptor: how the coordinator // reaches the worker that runs this action. One of "grpc", "http" or // "rmq". Empty means grpc — the zero-config default where the worker // is connected over the gRPC ExecuteStep stream. (issue #59) Transport string `json:"transport,omitempty"` // Address is the dispatch target for non-gRPC transports: a callback // URL for "http" or a queue name for "rmq". Required only when // Transport is "http" or "rmq"; ignored for "grpc". (issue #59) Address string `json:"address,omitempty"` } ``` ## type [Branch]() Branch is one outgoing edge of a decision/parallel step. ```go type Branch struct { Next string `json:"next"` } ``` ## type [Compensation]() Compensation describes how to undo a completed step. Nil means "non\-compensable; log a warning when rolling back." ```go type Compensation struct { Action string `json:"action"` Inputs map[string]any `json:"inputs,omitempty"` } ``` ## type [DecisionTableRow]() DecisionTableRow is one row in a decision\_table. The When expression is CEL; on truthy result the Then map becomes the rule's output. ```go type DecisionTableRow struct { When string `json:"when"` Then map[string]any `json:"then"` } ``` ## type [EventType]() EventType identifies what changed. ```go type EventType string ``` The EventType constants enumerate the audit events recorded for a run. ```go const ( EventSagaStarted EventType = "saga.started" EventStepDispatched EventType = "step.dispatched" EventStepStarted EventType = "step.started" EventStepSucceeded EventType = "step.succeeded" EventStepFailed EventType = "step.failed" EventStepSkipped EventType = "step.skipped" EventStepPaused EventType = "step.paused" EventRunSucceeded EventType = "run.succeeded" EventRunFailed EventType = "run.failed" EventRunCancelled EventType = "run.cancelled" EventCompensationStarted EventType = "compensation.started" EventLog EventType = "log" EventMetric EventType = "metric" EventRuleEvaluated EventType = "rule.evaluated" EventLicenseGateRejected EventType = "license.gate.rejected" ) ``` ## type [HitPolicy]() HitPolicy controls how rows in a decision\_table are evaluated. v1 supports only HitPolicyFirst \(return the first matched row's output\). ```go type HitPolicy string ``` HitPolicyFirst returns the first matched row's output; the only policy in v1. ```go const ( HitPolicyFirst HitPolicy = "first" ) ``` ## type [RetryPolicy]() RetryPolicy bounds step\-level retry. Defaults applied at engine time if a step omits the field. ```go type RetryPolicy struct { MaxAttempts int `json:"max_attempts"` InitialBackoffMS int `json:"initial_backoff_ms"` MaxBackoffMS int `json:"max_backoff_ms,omitempty"` Multiplier float64 `json:"multiplier,omitempty"` Jitter bool `json:"jitter,omitempty"` } ``` ## type [RuleDefinition]() RuleDefinition is one version of one rule. Stored in definitions.rule\_definitions. ```go type RuleDefinition struct { ID uuid.UUID `json:"id"` RuleID string `json:"rule_id"` Version int `json:"version"` TenantID *string `json:"tenant_id,omitempty"` Name string `json:"name"` RuleType RuleType `json:"rule_type"` Spec RuleSpec `json:"spec"` Published bool `json:"published"` CreatedAt time.Time `json:"created_at,omitempty"` CreatedBy string `json:"created_by,omitempty"` } ``` ### func [NewRuleDefinition]() ```go func NewRuleDefinition(ruleID string, version int, name string, ruleType RuleType, spec RuleSpec, createdBy string) RuleDefinition ``` NewRuleDefinition returns a fresh RuleDefinition with a generated ID and current timestamp. Caller fills in the spec. ## type [RuleSpec]() RuleSpec is the union of rule\-type\-specific bodies. v1 only fills in DecisionTable; later rule types add their own fields. ```go type RuleSpec struct { HitPolicy HitPolicy `json:"hit_policy"` Rows []DecisionTableRow `json:"rows"` DefaultOutput map[string]any `json:"default_output,omitempty"` } ``` ## type [RuleType]() RuleType is the discriminant for the union of rule shapes. v1 ships only \`decision\_table\`; v1.5\+ adds \`script\` and \`expression\_tree\`. ```go type RuleType string ``` RuleTypeDecisionTable is the only rule type shipped in v1. ```go const ( RuleTypeDecisionTable RuleType = "decision_table" ) ``` ## type [RunState]() RunState is the saga\-level state. ```go type RunState string ``` The RunState constants enumerate the saga\-level lifecycle states. ```go const ( RunStatePending RunState = "pending" RunStateRunning RunState = "running" RunStatePaused RunState = "paused" RunStateCompensating RunState = "compensating" RunStateSucceeded RunState = "succeeded" RunStateFailed RunState = "failed" RunStateCancelled RunState = "cancelled" ) ``` ### func \(RunState\) [IsTerminal]() ```go func (s RunState) IsTerminal() bool ``` IsTerminal reports whether the state is final \(no further transitions\). ## type [SagaRun]() SagaRun is one running instance of a workflow. Stored in runtime.saga\_runs. ```go type SagaRun struct { ID uuid.UUID `json:"id"` WorkflowID string `json:"workflow_id"` DefinitionID uuid.UUID `json:"definition_id"` TenantID *uuid.UUID `json:"tenant_id,omitempty"` State RunState `json:"state"` CurrentStep string `json:"current_step,omitempty"` Inputs map[string]any `json:"inputs"` Variables map[string]any `json:"variables"` StartedAt time.Time `json:"started_at"` LastEventAt time.Time `json:"last_event_at"` TerminalAt *time.Time `json:"terminal_at,omitempty"` RequiresManualReview bool `json:"requires_manual_review"` TriggerID *uuid.UUID `json:"trigger_id,omitempty"` ParentRunID *uuid.UUID `json:"parent_run_id,omitempty"` ParentStepID *string `json:"parent_step_id,omitempty"` ParentBranchID *string `json:"parent_branch_id,omitempty"` TryCatchStack []TryCatchFrame `json:"try_catch_stack,omitempty"` DryRun bool `json:"dry_run,omitempty"` WakeupAt *time.Time `json:"wakeup_at,omitempty"` AwaitedSignal *string `json:"awaited_signal,omitempty"` AwaitedEventTopic *string `json:"awaited_event_topic,omitempty"` AwaitedEventHeaders map[string]string `json:"awaited_event_headers,omitempty"` FeatureOverrides map[string]bool `json:"feature_overrides,omitempty"` AwaitedActionDispatch *string `json:"awaited_action_dispatch,omitempty"` CurrentAttempt int `json:"current_attempt,omitempty"` // LastError records why a run reached a terminal failed/cancelled // state — the failing step's error message, or the cancel reason — so a // run is self-describing without diffing its event log. nil while the // run is non-terminal or terminated cleanly (succeeded). See issue #80. LastError *string `json:"last_error,omitempty"` } ``` ### func [NewSagaRun]() ```go func NewSagaRun(workflowID string, definitionID uuid.UUID, tenantID *uuid.UUID, inputs map[string]any) SagaRun ``` NewSagaRun constructs a fresh run row in pending state. The caller supplies the resolved definition ID \+ inputs; the engine fills in CurrentStep when it picks the run up. ## type [SagaRunEvent]() SagaRunEvent is one audit row. Stored in audit.saga\_run\_events. ```go type SagaRunEvent struct { ID uuid.UUID `json:"id"` RunID uuid.UUID `json:"run_id"` StepID string `json:"step_id,omitempty"` Attempt int `json:"attempt"` EventType EventType `json:"event_type"` FromState string `json:"from_state,omitempty"` ToState string `json:"to_state,omitempty"` Actor string `json:"actor"` Metadata map[string]any `json:"metadata,omitempty"` RecordedAt time.Time `json:"recorded_at"` } ``` ### func [NewEvent]() ```go func NewEvent(runID uuid.UUID, stepID string, attempt int, eventType EventType, actor string) SagaRunEvent ``` NewEvent constructs an audit event with ID \+ timestamp filled in. ## type [SagaSignal]() SagaSignal is one row in runtime.saga\_signals. External code writes these via POST /sagas/\{id\}/signal/\{name\}; the engine consumes them to wake \`wait\_for\_signal\` steps. ```go type SagaSignal struct { ID uuid.UUID `json:"id"` RunID uuid.UUID `json:"run_id"` SignalName string `json:"signal_name"` Payload map[string]any `json:"payload,omitempty"` ReceivedAt time.Time `json:"received_at"` ConsumedAt *time.Time `json:"consumed_at,omitempty"` } ``` ## type [SagaTrigger]() SagaTrigger persists the binding between an external signal and a workflow definition. Created via REST or seeded via migration. ```go type SagaTrigger struct { ID uuid.UUID TriggerType TriggerType WorkflowID string Version int // Config is shape-by-trigger-type. For TriggerRecordTransition: // { "record_type": "order", "from_state": "created", // "to_state": "pending_review", "input_mapping": {...} } Config map[string]any Enabled bool TenantID *uuid.UUID CreatedAt time.Time CreatedBy string // Cron scheduling bookkeeping (cron triggers only). NextFireAt *time.Time `json:"next_fire_at,omitempty"` LastFiredAt *time.Time `json:"last_fired_at,omitempty"` } ``` ## type [Step]() Step is a single node in the workflow graph. ```go type Step struct { ID string `json:"id"` Type StepType `json:"type"` Next string `json:"next,omitempty"` Action string `json:"action,omitempty"` Inputs map[string]any `json:"inputs,omitempty"` Compensation *Compensation `json:"compensation,omitempty"` Retry *RetryPolicy `json:"retry,omitempty"` Branches map[string]Branch `json:"branches,omitempty"` Extra map[string]any `json:"-"` } ``` ## type [StepType]() StepType is the discriminant for the union of step shapes. ```go type StepType string ``` The StepType constants enumerate every verb the engine can dispatch. ```go const ( StepTypeEnd StepType = "end" StepTypeAction StepType = "action" StepTypeDecision StepType = "decision" StepTypeError StepType = "error" StepTypeNoop StepType = "noop" StepTypeSetVar StepType = "set_var" StepTypeTransform StepType = "transform" StepTypeMerge StepType = "merge" StepTypeFilter StepType = "filter" StepTypeMap StepType = "map" StepTypeLog StepType = "log" StepTypeMetricEmit StepType = "metric_emit" StepTypeAssert StepType = "assert" StepTypeHTTPRequest StepType = "http_request" StepTypeWebhookEmit StepType = "webhook_emit" StepTypeWaitDuration StepType = "wait_duration" StepTypeWaitUntil StepType = "wait_until" StepTypeWaitForSignal StepType = "wait_for_signal" StepTypeWaitForEvent StepType = "wait_for_event" StepTypeParallel StepType = "parallel" StepTypeJoin StepType = "join" StepTypeForeach StepType = "foreach" StepTypeWhile StepType = "while" StepTypeTryCatch StepType = "try_catch" StepTypeSubSaga StepType = "sub_saga" StepTypeSpawnSaga StepType = "spawn_saga" StepTypeManualApproval StepType = "manual_approval" StepTypeCollectInput StepType = "collect_input" StepTypeSwitch StepType = "switch" StepTypeEmitSignal StepType = "emit_signal" StepTypeCancel StepType = "cancel" StepTypeEmitEvent StepType = "emit_event" ) ``` ## type [TriggerFireRow]() TriggerFireRow is a single row in runtime.saga\_trigger\_fires. ```go type TriggerFireRow struct { ID uuid.UUID `json:"id"` TriggerID uuid.UUID `json:"trigger_id"` WorkflowID string `json:"workflow_id"` FiredAt time.Time `json:"fired_at"` ResultingRunID *uuid.UUID `json:"resulting_run_id,omitempty"` Error string `json:"error,omitempty"` } ``` ## type [TriggerType]() TriggerType is the dispatch key for matching incoming signals \(events, cron ticks, etc.\) to workflows. ```go type TriggerType string ``` The TriggerType constants enumerate the supported trigger dispatch keys. ```go const ( // TriggerRecordTransition fires when a record state-transition event // arrives matching the trigger's config.record_type, config.from_state, // config.to_state. TriggerRecordTransition TriggerType = "record_transition" // TriggerCron fires on a schedule defined by a cron expression in // config.schedule (standard 5-field cron syntax, UTC). TriggerCron TriggerType = "cron" ) ``` ## type [TryCatchFrame]() TryCatchFrame is one entry in a saga's try\_catch stack. When a step inside the try block errors, the coordinator pops the top frame and jumps to its CatchStep instead of failing the saga. ```go type TryCatchFrame struct { StepID string `json:"step_id"` // the try_catch step itself CatchStep string `json:"catch_step"` // step ID to jump to on error } ``` ## type [UserTask]() UserTask is one row in runtime.saga\_user\_tasks. Created by the manual\_approval and collect\_input verbs; submitted via POST /api/v1/sagas/\{run\_id\}/user\_task/\{task\_id\}/submit. ```go type UserTask struct { ID uuid.UUID `json:"id"` RunID uuid.UUID `json:"run_id"` StepID string `json:"step_id"` Assignee string `json:"assignee"` DueAt *time.Time `json:"due_at,omitempty"` FormSchema map[string]any `json:"form_schema,omitempty"` SubmittedAt *time.Time `json:"submitted_at,omitempty"` SubmittedBy string `json:"submitted_by,omitempty"` Result map[string]any `json:"result,omitempty"` } ``` ## type [WorkflowDefinition]() WorkflowDefinition is one version of one workflow. Stored in definitions.workflow\_definitions. ```go type WorkflowDefinition struct { ID string `json:"id"` Version int `json:"version"` TenantID *string `json:"tenant_id,omitempty"` Name string `json:"name"` Description string `json:"description,omitempty"` Start string `json:"start"` Entrypoints map[string]string `json:"entrypoints,omitempty"` // entry name -> step id; "" / "default" => Start Steps []Step `json:"steps"` Published bool `json:"published"` CreatedAt time.Time `json:"created_at,omitempty"` CreatedBy string `json:"created_by,omitempty"` } ``` ### func \(WorkflowDefinition\) [ResolveEntry]() ```go func (d WorkflowDefinition) ResolveEntry(entrypoint string) (string, error) ``` ResolveEntry returns the step id a run should begin at for the named entry point. "" and "default" resolve to Start. An unknown name returns an error. ### func \(\*WorkflowDefinition\) [StepByID]() ```go func (d *WorkflowDefinition) StepByID(id string) (Step, bool) ``` StepByID returns the step with the given ID and whether it was found. Generated by [gomarkdoc]() --- # engine Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/engine # engine ```go import "github.com/Bugs5382/go-saga-orchestration/engine" ``` Package engine contains the saga coordinator and the built\-in verb dispatch table. Package engine — workflow\-definition validators run before persisting. Reject structural errors \(forbidden compositions, missing references\) at publish/create time so runtime never sees them. ## Constants FeatureCronTriggers is the licensing feature flag that gates cron\-triggered run starts. Tenants without this feature enabled are skipped. ```go const FeatureCronTriggers = "wf.cron_triggers" ``` TimerAdvisoryLockID is the Postgres advisory lock the timer dispatcher holds while it's the leader. ```go const TimerAdvisoryLockID = int64(0xBA70C420) ``` ## func [Backoff]() ```go func Backoff(p domain.RetryPolicy, attempt int, jitter bool) time.Duration ``` Backoff returns the wait duration for \`attempt\` \(zero\-indexed\) under policy p. Caps at MaxBackoffMS. If jitter is true, applies ±25% noise. ## func [DefaultRetryPolicy]() ```go func DefaultRetryPolicy() domain.RetryPolicy ``` DefaultRetryPolicy returns the spec default per § 3.4. ## func [InjectStartupVariables]() ```go func InjectStartupVariables(ctx context.Context, s store.Store, runID uuid.UUID, tenantID *uuid.UUID, log zerolog.Logger, providers ...StartupVariableProvider) ``` InjectStartupVariables runs each provider for the given tenant, merges the returned maps, and writes them onto the run via UpdateRunVariables. With no providers it is a no\-op \(no write\). Provider errors are logged and skipped. ## func [ParseSchedule]() ```go func ParseSchedule(expr string) (cron.Schedule, error) ``` ParseSchedule parses a standard 5\-field cron expression \(and @descriptors\). Granularity is one minute. For sub\-minute cadences use config.interval on the cron trigger instead of config.schedule. ## func [ResolveEntry]() ```go func ResolveEntry(def domain.WorkflowDefinition, entrypoint string) (string, error) ``` ResolveEntry resolves a workflow entry point to its start step id. Delegates to domain.WorkflowDefinition.ResolveEntry. ## func [ValidateDefinition]() ```go func ValidateDefinition(def domain.WorkflowDefinition) error ``` ValidateDefinition runs structural checks on a workflow definition. Returns nil if valid. Currently checks: - try\_catch cannot contain a parallel step \(per v1 spec §12.3\). Future checks: license\-group gates, max nesting depth, missing\-step\-reference detection. TODO: call ValidateDefinition from the POST /workflows publish handler so invalid definitions are rejected at the API boundary. ## func [ValidateDefinitionWithLicense]() ```go func ValidateDefinitionWithLicense(def domain.WorkflowDefinition, registry verbs.Registry, resolver licensing.Resolver, tenantID *uuid.UUID, overrides map[string]bool) error ``` ValidateDefinitionWithLicense runs the existing structural checks \(try\-contains\-parallel etc\) AND walks every step to verify the tenant's license includes the verb's feature flag. First license\-gate violation is returned; structural errors take precedence. Pass a Resolver from licensing. For non\-licensed environments \(dev/test\) pass licensing.StubAllowAll\{\}, which approves everything. overrides: per\-saga feature overrides \(X\-Feature\-Override header\). Pass nil for none. ## type [Coordinator]() Coordinator wires the saga.advance consumer to the per\-run advance loop. It owns no goroutines beyond the consumer; Advance does its work synchronously per message. ```go type Coordinator struct { // contains filtered or unexported fields } ``` ### func [NewCoordinator]() ```go func NewCoordinator(s store.Store, pub Publisher, clk clock.Clock, sec secrets.Resolver, lr licensing.Resolver, actionPub verbs.ActionDispatchPublisher, emitter verbs.EventEmitter, opts ...verbs.DefaultOption) *Coordinator ``` NewCoordinator constructs a Coordinator. pub is the Publisher used to re\-enqueue saga.advance \(multi\-step sagas advance one queue message at a time\) and to start child runs spawned by the parallel verb. pub may be nil in tests that only exercise synchronous / non\-spawning verbs. actionPub is the ActionDispatchPublisher for the action verb. Pass nil in tests that do not exercise action steps. lr is the license resolver; pass licensing.StubAllowAll\{\} for dev/test. A nil lr is normalised to StubAllowAll so callers need not guard. emitter is the EventEmitter used by emit\_event steps. Pass nil in tests that do not exercise emit\_event — EmitEventVerb checks for nil. opts wire optional verb dependencies, e.g. the http/rmq action dispatchers for the dispatch\-descriptor feature \(verbs.WithHTTPDispatcher / verbs.WithRMQDispatcher\); omit them for the gRPC\-only default. \(issue \#59\) ### func \(\*Coordinator\) [Advance]() ```go func (c *Coordinator) Advance(ctx context.Context, runIDStr string) error ``` Advance walks the saga forward until it either reaches a terminal state or a step that needs external I/O \(future async verbs\). Each loop iteration re\-reads the run from the store so it sees any variable updates written by the previous step. ### func \(\*Coordinator\) [Cancel]() ```go func (c *Coordinator) Cancel(ctx context.Context, runID uuid.UUID, reason string) error ``` Cancel terminates an in\-flight run from outside the run — the run\-level counterpart to the in\-step cancel verb. An external caller \(e.g. an approval policy withdrawing or re\-submitting while a run is paused at a manual\_approval\) uses this to abort the prior run so its pending user tasks leave the approver's inbox and a fresh run can start, instead of reaching around the API into the store. The store transition is atomic: terminal cancelled \+ terminal\_at, reason recorded in last\_error, open user tasks closed, awaited signal/event and pending wakeup cleared. Idempotent — a no\-op when the run is already terminal. If the cancelled run is a child, its parent's join is re\-evaluated so the parent is not left waiting. See issue \#80. ### func \(\*Coordinator\) [CheckParentJoin]() ```go func (c *Coordinator) CheckParentJoin(ctx context.Context, run domain.SagaRun) ``` CheckParentJoin re\-evaluates the parent join of run's parent \(if any\), waking the parent when the join is satisfied. Exported for verbs \(e.g. cancel\) that terminate a run outside the normal Advance flow. ### func \(\*Coordinator\) [RegisterVerb]() ```go func (c *Coordinator) RegisterVerb(stepType domain.StepType, handler verbs.Handler, licenseGroup string) ``` RegisterVerb adds or replaces a verb handler in the coordinator's registry, extending the engine with custom step types without rebuilding it. ## type [CronDispatcher]() CronDispatcher polls the store for due cron triggers and starts a new saga run for each one it claims. It uses a compare\-and\-swap on next\_fire\_at to ensure exactly one pod fires per trigger tick. Run it as a goroutine from cmd/engine alongside the Timer. ```go type CronDispatcher struct { S store.Store Publisher TimerPublisher Clock clock.Clock Tick time.Duration // default 1s Licensing licensing.Resolver Providers []StartupVariableProvider Logger zerolog.Logger } ``` ### func \(\*CronDispatcher\) [Run]() ```go func (d *CronDispatcher) Run(ctx context.Context) error ``` Run loops until ctx is cancelled. Each tick it calls fireDue to process all cron triggers whose next\_fire\_at is in the past. ## type [EventDelivery]() EventDelivery is the minimum shape EventSubscriber needs from a RabbitMQ delivery. Production wires from amqp.Delivery; tests synthesise it directly. ```go type EventDelivery struct { Topic string // RabbitMQ routing key Headers map[string]string // string-keyed headers (RMQ amqp.Table values stringified) Body []byte } ``` ## type [EventSubscriber]() EventSubscriber matches incoming RabbitMQ events to paused sagas awaiting matching topic \+ header subset. On match: clear pause \+ publish saga.advance. ```go type EventSubscriber struct { S store.Store Publisher TimerPublisher // same interface as Timer Dispatcher *TriggerDispatcher // optional; nil = wake-only mode } ``` ### func \(\*EventSubscriber\) [Deliver]() ```go func (e *EventSubscriber) Deliver(ctx context.Context, d EventDelivery) error ``` Deliver processes one event delivery. Returns nil if no matching run; returns the first error encountered while updating store / publishing. ### func \(\*EventSubscriber\) [RunRMQ]() ```go func (e *EventSubscriber) RunRMQ(ctx context.Context, conn *amqp.Connection, queueName string) error ``` RunRMQ binds a per\-pod queue to the workflow.events exchange and consumes deliveries indefinitely. ACKs every delivery \(events are fire\-and\-forget; redelivery semantics are not useful here\). ## type [LicenseGateError]() LicenseGateError is returned by ValidateDefinitionWithLicense when a step references a verb in a license\-group whose feature flag isn't enabled for the tenant. The API publish endpoint returns this as a 422 with the structured body. ```go type LicenseGateError struct { StepID string `json:"step_id"` Group string `json:"group"` Feature string `json:"feature"` } ``` ### func \(LicenseGateError\) [Error]() ```go func (e LicenseGateError) Error() string ``` Error formats the gate failure naming the step, group, and required feature. ## type [Publisher]() Publisher is the minimum surface the Coordinator needs from mq.Publisher: enqueue a saga.advance message so the next step runs. Using an interface \(instead of \*mq.Publisher directly\) lets in\-process tests supply a fake. \*mq.Publisher satisfies this interface. ```go type Publisher interface { PublishSagaAdvance(ctx context.Context, runID string) error } ``` ## type [StartupVariableProvider]() StartupVariableProvider supplies per\-saga "magic" variables \(leading\-underscore names by convention\) that are merged into a run's Variables at start. The engine ships no built\-in providers; integrators register their own. Best\-effort: a provider error is logged and skipped so a config glitch never strands a start. ```go type StartupVariableProvider interface { StartupVariables(ctx context.Context, tenantID *uuid.UUID) (map[string]any, error) } ``` ## type [Timer]() Timer polls saga\_runs for due wakeups and republishes saga.advance. Run it as a goroutine from cmd/engine; one leader\-elected instance per cluster at a time \(advisory\-lock guarded by the caller via AcquireLeaderLock\). ```go type Timer struct { S store.Store Publisher TimerPublisher Clock clock.Clock Tick time.Duration // default 1s; tests use 10ms with FakeClock BatchSize int // default 100 } ``` ### func \(\*Timer\) [Run]() ```go func (t *Timer) Run(ctx context.Context) error ``` Run loops until ctx is cancelled. Each tick: query due\-wakeup runs, publish saga.advance for each. ## type [TimerPublisher]() TimerPublisher is the surface Timer needs from mq.Publisher. Allows in\-process tests to supply a fake. ```go type TimerPublisher interface { PublishSagaAdvance(ctx context.Context, runID string) error } ``` ## type [TriggerDispatcher]() TriggerDispatcher matches incoming events to saga\_triggers rows and starts a new saga for each match. Peer of EventSubscriber: the same RabbitMQ delivery feeds both — EventSubscriber wakes paused sagas, TriggerDispatcher starts new ones. ```go type TriggerDispatcher struct { S store.Store Publisher TimerPublisher // reuse — same PublishSagaAdvance method StartupProviders []StartupVariableProvider } ``` ### func \(\*TriggerDispatcher\) [Dispatch]() ```go func (d *TriggerDispatcher) Dispatch(ctx context.Context, evt EventDelivery) error ``` Dispatch inspects one delivery. For each enabled record\_transition trigger whose config matches the event's record\_type, from\_state, and to\_state, it creates a new saga run and publishes saga.advance. Returns the first error encountered \(from store/publish failures\); logs and continues on per\-trigger failures so one bad trigger doesn't block the others. Generated by [gomarkdoc]() --- # verbs Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/engine/verbs # verbs ```go import "github.com/Bugs5382/go-saga-orchestration/engine/verbs" ``` Package verbs holds built\-in verb handlers. The \`end\` verb is dispatched inline by the coordinator; the other verbs are dispatched via the registry that maps StepType → Handler. This file exists to anchor the package and the interface shape that registry will use. ## Variables ErrSagaCancelled is returned by the cancel verb for a self\-cancel. Advance transitions the run to cancelled \(terminal\) and stops the loop. ```go var ErrSagaCancelled = errors.New("saga cancelled") ``` ErrSagaPaused signals "the verb successfully suspended the saga; do not advance, do not fail." Coordinator recognises this and ACKs the RabbitMQ message without transitioning state to failed. The verb is responsible for persisting the pause state \(wakeup\_at / awaited\_signal / awaited\_event\_\*\) via the store BEFORE returning this sentinel. ```go var ErrSagaPaused = errors.New("saga paused") ``` GroupToFeature maps each license\-group name to its feature flag. "common" maps to the empty string — that's the sentinel for "no gate, always allowed". ```go var GroupToFeature = map[string]string{ "common": "", "observability": "wf.observability", "external_io_advanced": "wf.external_io", "waits": "wf.timers", "events_and_signals": "wf.event_driven", "human_interaction": "wf.user_tasks", "parallel_control": "wf.parallel", "loops_and_recovery": "wf.loops_recovery", "compositions": "wf.compositions", } ``` ## func [AggregateJoinResults]() ```go func AggregateJoinResults(ctx context.Context, s store.Store, children []domain.SagaRun) []any ``` AggregateJoinResults builds the per\-child result list written into a run's Variables under "\_join.\.branches" \(join\) or "\_parallel.\.branches" \(parallel\). For each child: - key: the child's ParentBranchID \(or "b\{index\}" fallback\) - variables: the child's final Variables map - state: the child's terminal state \("succeeded" or "failed"\) - \_user\_task: the first submitted user\_task owned by the child \(if any\), as \{id, result, submitted\_by, submitted\_at\}. First by ID order wins. ## func [EvalQuorumNCEL]() ```go func EvalQuorumNCEL(expr string, vars map[string]any) (any, error) ``` EvalQuorumNCEL evaluates expr as a CEL expression against vars and returns the result as any \(expected to be numeric — int64 from CEL\). Mirrors evalBranchesCEL but expects a scalar numeric result. Exported for use by the engine's checkParentJoin path in advance.go. ## func [JoinConditionMet]() ```go func JoinConditionMet(inputs map[string]any, vars map[string]any, group []domain.SagaRun) bool ``` JoinConditionMet reports whether the join/parallel barrier described by inputs is satisfied by the given group of runs. It reads "join\_strategy" \("all" default, or "quorum"\) and "quorum\_n" \(int or CEL string\) exactly as the parallel and join verbs configure them, so the parallel child\-join hook and the join\-barrier hook in the engine share one implementation. - "all": every run in the group must be terminal. - "quorum": at least quorum\_n runs must be in RunStateSucceeded. A missing/invalid quorum\_n falls back to "all" \(matching the historical child\-join behaviour\), logged. An empty group is treated as not\-met. ## func [LicenseGroupForStep]() ```go func LicenseGroupForStep(step domain.Step, regGroup string) string ``` LicenseGroupForStep returns the effective license\-group for the given step. Most verbs have a static group \(set in registry.Default via RegistryEntry.LicenseGroup\). One exception: http\_request has a dynamic group depending on its inputs — GET with no secret\_ref is \`common\`; everything else is \`external\_io\_advanced\`. This function applies that dynamic override. ## func [ResolveJoinStreams]() ```go func ResolveJoinStreams(raw any, vars map[string]any) ([]string, error) ``` ResolveJoinStreams normalises the join verb's "streams" input into a list of upstream step IDs. Accepts a literal \[\]any of strings, a \[\]string, or a CEL string that evaluates against vars to a list of strings. The list must be non\-empty. Exported so the engine's join\-barrier hook resolves the same stream set the join verb watched. ## func [ToInt]() ```go func ToInt(v any) (int, bool) ``` ToInt coerces v to an int. Accepts int, int64, and float64 \(JSON\-decoded numbers\). Returns \(0, false\) for any other type. Exported so the engine package can reuse it when reading quorum\_n from a step's Inputs. ## func [ToIntFromAny]() ```go func ToIntFromAny(v any) (int, bool) ``` ToIntFromAny coerces v to int, accepting the same types as ToInt plus int32 and uint64 edge cases from CEL numeric coercions. Exported for use by the engine's checkParentJoin path in advance.go. ## type [ActionDispatchPublisher]() ActionDispatchPublisher is a separate interface for publishing action dispatch messages to RabbitMQ. Kept separate so existing test pubs that only implement PublishSagaAdvance do not need to change. ```go type ActionDispatchPublisher interface { PublishActionDispatch(ctx context.Context, routingKey string, payload []byte) error } ``` ## type [ActionHTTPDispatcher]() ActionHTTPDispatcher delivers an action dispatch payload to a worker over an HTTP callback. Used when an ActionRegistration declares transport="http". address is the callback URL from the registration; payload is the marshalled ActionPayload. The worker reports its result asynchronously via the result\-callback REST endpoint. \(issue \#59\) ```go type ActionHTTPDispatcher interface { DispatchHTTP(ctx context.Context, address string, payload []byte) error } ``` ## type [ActionPayload]() ActionPayload is the body of a saga.advance → action dispatch message. Workers deserialise this to drive their handler. ```go type ActionPayload struct { RunID string `json:"run_id"` StepID string `json:"step_id"` Attempt int `json:"attempt"` IdempotencyKey string `json:"idempotency_key"` Action string `json:"action"` // "." Inputs map[string]any `json:"inputs"` DryRun bool `json:"dry_run,omitempty"` } ``` ## type [ActionRMQDispatcher]() ActionRMQDispatcher delivers an action dispatch payload to a worker by publishing it to a named RabbitMQ queue. Used when an ActionRegistration declares transport="rmq"; address is the queue name from the registration. The worker reports its result asynchronously via the result\-callback REST endpoint. \(issue \#59\) ```go type ActionRMQDispatcher interface { DispatchRMQQueue(ctx context.Context, queue string, payload []byte) error } ``` ## type [ActionVerb]() ActionVerb dispatches a registered action to a worker over its declared transport. The transport comes from the action's ActionRegistration dispatch descriptor \(issue \#59\): - "" / "grpc": the zero\-config default. Publishes to ExchangeAction with routing key = step.Action; the worker is connected over the gRPC ExecuteStep stream. - "http": POSTs the ActionPayload to the registration's Address \(a callback URL\). - "rmq": publishes the ActionPayload to the RabbitMQ queue named by the registration's Address. Inputs: - step.Action \(required, string\): "\.\". Must contain a dot. - step.Inputs \(any\): forwarded verbatim to the worker. The verb: 1. Bumps current\_attempt \+ persists the awaiting state via MarkAwaitingAction. 2. Resolves the action's dispatch descriptor \(latest registered version\) and dispatches over the declared transport. 3. Returns ErrSagaPaused. gRPC workers reply via the ExecuteStep stream \(Complete/Error\). http and rmq workers have no return stream; they report their result asynchronously via the result\-callback REST endpoint \(POST /api/v1/sagas/\{run\_id\}/actions/\{step\_id\}/result\). Both paths land on the same CompleteAction / FailAction store hooks that resume or fail the saga. ```go type ActionVerb struct { S store.Store Publisher ActionDispatchPublisher // HTTPDispatcher delivers the payload for transport="http" registrations. // nil disables http dispatch (resolution falls back to an error). HTTPDispatcher ActionHTTPDispatcher // RMQDispatcher delivers the payload for transport="rmq" registrations. // nil disables rmq dispatch. RMQDispatcher ActionRMQDispatcher } ``` ### func \(ActionVerb\) [DispatchCompensation]() ```go func (v ActionVerb) DispatchCompensation(ctx context.Context, runID, stepID, action string, inputs map[string]any, dryRun bool) error ``` DispatchCompensation sends a step's compensation action to its worker over the action's declared transport, reusing the same routing the action verb uses. It does not mark the run awaiting or pause it: compensation runs are dispatched best\-effort while the run settles to failed. The action must be in "\.\" form. ### func \(ActionVerb\) [Execute]() ```go func (v ActionVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute validates the action name, persists the awaiting\-action state with a bumped attempt, publishes the dispatch message to the worker, and returns ErrSagaPaused so the saga waits for the worker's reply. ## type [AssertVerb]() AssertVerb evaluates a CEL expression; on false returns an error. Inputs: - "expr" \(required, string\) - "code" \(optional, string; default "assertion\_failed"\) ```go type AssertVerb struct{} ``` ### func \(AssertVerb\) [Execute]() ```go func (AssertVerb) Execute(_ context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute compiles and evaluates the CEL expr against run.Variables and returns an error tagged with code when the result is not true. ## type [CancelVerb]() CancelVerb cancels a run. With no run\_id \(or run\_id == the current run\) it self\-cancels: returns ErrSagaCancelled and the engine sets state=cancelled. With a different run\_id it cancels that target run and the current run continues to Next. Inputs: "run\_id" \(optional, string\), "reason" \(optional\). ```go type CancelVerb struct { S store.Store // JoinChecker, when set, re-evaluates a cancelled target's parent join so a // parent paused on a parallel/sub_saga join is woken. Nil-safe. JoinChecker ParentJoinChecker } ``` ### func \(CancelVerb\) [Execute]() ```go func (v CancelVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` ## type [CollectInputVerb]() CollectInputVerb is the same shape as ManualApprovalVerb but \`form\_schema\` is REQUIRED. Use this verb when the workflow needs structured data from the user \(e.g., remediation plan, additional context\) — vs. manual\_approval which is typically approve/reject. Inputs: - "assignee" \(required, string\) - "form\_schema" \(required, map\[string\]any\) - "due\_in" \(optional, string Go duration\) ```go type CollectInputVerb struct { S store.Store Clock clock.Clock } ``` ### func \(CollectInputVerb\) [Execute]() ```go func (v CollectInputVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute creates a user task carrying the required form schema, then pauses the saga awaiting the task's submitted signal, returning ErrSagaPaused. ## type [CompensationPayload]() CompensationPayload is the body of a compensation action dispatch. It mirrors ActionPayload but carries no attempt/idempotency machinery: compensation is a fire\-and\-forget rollback dispatch, not an awaited step. ```go type CompensationPayload struct { RunID string `json:"run_id"` StepID string `json:"step_id"` // the step being compensated Action string `json:"action"` // "." Inputs map[string]any `json:"inputs"` DryRun bool `json:"dry_run,omitempty"` } ``` ## type [DecisionVerb]() DecisionVerb evaluates a stored rule and returns its output map. The engine reads result\["branch"\] to pick step.Branches\[...\].Next. Inputs: - "rule\_id" \(required, string\) - "inputs\_map" \(optional, map\[string\]string\): narrow the inputs passed to the rule by mapping rule\-input\-name → variable\-name. If omitted, run.Variables is passed directly. ```go type DecisionVerb struct { S store.Store } ``` ### func \(DecisionVerb\) [Execute]() ```go func (v DecisionVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute loads the published rule, evaluates it against the \(optionally remapped\) inputs, records a rule\-evaluated event, and returns the rule's output map. ## type [DefaultOption]() DefaultOption configures optional verb dependencies \(e.g. the http/rmq action dispatchers\) without breaking Default's positional signature. ```go type DefaultOption func(*defaultConfig) ``` ### func [WithHTTPDispatcher]() ```go func WithHTTPDispatcher(d ActionHTTPDispatcher) DefaultOption ``` WithHTTPDispatcher wires the http action dispatcher for transport="http" action registrations. \(issue \#59\) ### func [WithRMQDispatcher]() ```go func WithRMQDispatcher(d ActionRMQDispatcher) DefaultOption ``` WithRMQDispatcher wires the rmq action dispatcher for transport="rmq" action registrations. \(issue \#59\) ## type [EmitEventVerb]() EmitEventVerb publishes an event via the configured EventEmitter. Inputs: - "topic" \(required, string\) - "headers" \(optional, map\[string\]any \-\> stringified\) - "payload" \(optional, map\[string\]any\) ```go type EmitEventVerb struct { Emitter EventEmitter } ``` ### func \(EmitEventVerb\) [Execute]() ```go func (v EmitEventVerb) Execute(ctx context.Context, _ domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute implements Handler. ## type [EmitSignalVerb]() EmitSignalVerb sends a signal to a target run \(the send\-side of wait\_for\_signal\). If the target run is currently paused awaiting that signal, it is consumed and a saga.advance message is published so the engine resumes the target immediately. Inputs: - "run\_id" \(required, string\): target run UUID. - "name" \(required, string\): signal name. - "payload" \(optional, map\[string\]any\): arbitrary signal payload. ```go type EmitSignalVerb struct { S store.Store Publisher Publisher } ``` ### func \(EmitSignalVerb\) [Execute]() ```go func (v EmitSignalVerb) Execute(ctx context.Context, _ domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute implements Handler. ## type [EndVerb]() EndVerb terminates the saga successfully. The coordinator dispatches this inline today; the type is here so it can move into a registry without breaking the contract. ```go type EndVerb struct{} ``` ### func \(EndVerb\) [Execute]() ```go func (EndVerb) Execute(_ context.Context, _ domain.SagaRun, _ domain.Step) (map[string]any, error) ``` Execute returns an empty result; the coordinator handles the run\-level state transition. ## type [ErrorVerb]() ErrorVerb halts the saga with a non\-retryable error. Inputs: - "code" \(required, string\) - "message" \(optional, string\) ```go type ErrorVerb struct{} ``` ### func \(ErrorVerb\) [Execute]() ```go func (ErrorVerb) Execute(_ context.Context, _ domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute always returns a non\-retryable error built from the required code and optional message. ## type [EventEmitter]() EventEmitter publishes an event other sagas/triggers can match. Real impls: an in\-process matcher \(embedded\) or a RabbitMQ publisher \(service mode\). ```go type EventEmitter interface { EmitEvent(ctx context.Context, topic string, headers map[string]string, payload map[string]any) error } ``` ## type [FilterVerb]() FilterVerb keeps list elements where expr is truthy. Inputs: - "list" \(required, string\): CEL expression that must evaluate to a list. - "expr" \(required, string\): CEL predicate; element bound as \`\_\`. - "out\_var" \(required, string\): variable to write the filtered list to. ```go type FilterVerb struct{} ``` ### func \(FilterVerb\) [Execute]() ```go func (FilterVerb) Execute(_ context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute evaluates the list expression, applies the predicate to each element \(bound as \`\_\`\), and writes the retained elements to out\_var. ## type [ForeachVerb]() ForeachVerb fans out one child run per element of a CEL\-evaluated list. PARALLEL mode runs one branch per list element. Sequential mode is intentionally deferred — use \`while\` with an explicit counter for now. TODO\(future\-batch\): add sequential mode \(parallel: false\) — state\-machine back\-edge loop where each iteration executes the body steps one at a time, advancing via a back\-edge to re\-enter the foreach step after the body completes. For now, use \`while\` with an index counter for sequential loops. Inputs: - "list" \(required, string\): CEL expression evaluating to a list. - "body" \(required, \[\]any\): step objects forming the loop body. Each child run gets the list element bound as Variables\["\_foreach\_item"\]. - "start" \(required, string\): ID of the first step inside body. - "parallel" \(optional, bool, default true\): the only supported mode in v1 is parallel; passing false returns an error noting the deferred sequential mode. ```go type ForeachVerb struct { S store.Store Publisher Publisher } ``` ### func \(ForeachVerb\) [Execute]() ```go func (v ForeachVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute evaluates the list expression and spawns one child run per element \(each with the element bound as \_foreach\_item\), then pauses the parent awaiting the children and returns ErrSagaPaused. An empty list advances without spawning; sequential mode is rejected. ## type [HTTPRequestVerb]() HTTPRequestVerb issues a synchronous outbound HTTP request and merges the response into Variables. Inputs: - "method" \(optional, string; default "GET"\) - "url" \(required, string\) - "headers" \(optional, map\[string\]any → stringified into request headers\) - "body" \(optional, any; JSON\-marshalled into request body\) - "timeout\_s" \(optional, number; default 30s\) - "secret\_ref" \(optional, string; resolves to a value set as Authorization header\) - "out\_var" \(optional, string; default "http\_result"\). Result keys: \{out\_var\} = parsed JSON body if application/json, else raw string. \{out\_var\}\_status = int64 status code. \{out\_var\}\_headers = map\[string\]string of response headers. ```go type HTTPRequestVerb struct { Secrets secrets.Resolver Client *http.Client // optional; nil → constructed per-call with timeout_s } ``` ### func \(HTTPRequestVerb\) [Execute]() ```go func (v HTTPRequestVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute issues the configured HTTP request \(resolving secret\_ref into an Authorization header when set\) and returns the parsed body, status code, and response headers keyed off out\_var. ## type [Handler]() Handler executes one verb. Returns the result map \(becomes the step's output for subsequent step inputs\) and an error. Built\-in verbs are pure functions of \(run, step, ctx\) — no external I/O. ```go type Handler interface { Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) } ``` ## type [HandlerFunc]() HandlerFunc adapts a plain function to the Handler interface so a custom verb can be a closure instead of a struct. ```go type HandlerFunc func(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` ### func \(HandlerFunc\) [Execute]() ```go func (f HandlerFunc) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute satisfies Handler. ## type [JoinVerb]() JoinVerb is a barrier that reconvenes independently\-spawned upstream streams before the run continues. Unlike parallel \(which spawns its own branches and immediately pauses\), join watches children that earlier steps spawned in the same run \- the natural producer is spawn\_saga, whose fire\-and\-forget children the parent did not wait on. join lets a later step gather those streams back together. Inputs: - "streams" \(required, \[\]any of strings, or a CEL string that evaluates to a non\-empty list of strings\): the IDs of upstream steps in THIS run whose spawned children the join waits on. Each named step must have spawned at least one child \(via spawn\_saga, parallel, foreach, or sub\_saga\); join collects the children of every named step by calling ListChildrenByParent\(run.ID, streamStepID\) and treats the union as the watched group. - "join\_strategy" \(optional, string, default "all"\): "all" waits for every watched child to reach a terminal state; "quorum" resolves once quorum\_n watched children have succeeded. - "quorum\_n" \(required when join\_strategy=="quorum", int or CEL string\): positive integer, must be \<= the number of watched children. Resolution: - If the barrier is already satisfied when the verb runs \(the watched children finished before control reached the join\), Execute aggregates their outputs and returns them so the run advances to step.Next without pausing. - Otherwise Execute pauses the run \(ErrSagaPaused\). The coordinator's checkJoinBarriers hook \(engine/advance.go\) re\-evaluates every join step whenever a watched child terminates and wakes the run once the strategy is satisfied, aggregating the outputs at wake time. Aggregated outputs land in Variables under "\_join.\.branches" as a list of \{key, variables, state, \_user\_task?\} entries, mirroring the "\_parallel.\.branches" shape produced by the parallel verb. ```go type JoinVerb struct { S store.Store } ``` ### func \(JoinVerb\) [Execute]() ```go func (v JoinVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute resolves the watched streams, validates the join strategy, and either aggregates\-and\-continues \(barrier already met\) or pauses the run \(ErrSagaPaused\) until checkJoinBarriers wakes it. ## type [LogVerb]() LogVerb appends a saga\_run\_events row of type \`log\`. Inputs: - "message" \(required, string\) - "level" \(optional, string: "info" | "warn" | "error"; default "info"\) ```go type LogVerb struct { S store.Store } ``` ### func \(LogVerb\) [Execute]() ```go func (v LogVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute appends a log event carrying the message and level to the run's event stream. ## type [ManualApprovalVerb]() ManualApprovalVerb creates a user task and pauses the saga awaiting its submission via POST /api/v1/sagas/\{run\_id\}/user\_task/\{task\_id\}/submit. The submit handler appends a signal of name \`user\_task.\{task\_id\}.submitted\` which wakes this saga. Inputs: - "assignee" \(required, string\): user ID or role expected to submit. - "due\_in" \(optional, string, Go duration\): sets due\_at = clock.Now\(\) \+ due\_in. - "form\_schema" \(optional, map\[string\]any\): rendered to the assignee in the UI \(admin panel\). For manual\_approval the form is typically a simple \{approve|reject\} radio; the schema is optional. ```go type ManualApprovalVerb struct { S store.Store Clock clock.Clock } ``` ### func \(ManualApprovalVerb\) [Execute]() ```go func (v ManualApprovalVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute creates a user task for the assignee and pauses the saga awaiting its submitted signal, returning ErrSagaPaused. ## type [MapVerb]() MapVerb transforms each element of a list via a CEL expression where the element is bound as \`\_\`. Inputs: list, expr, out\_var \(all required strings\). ```go type MapVerb struct{} ``` ### func \(MapVerb\) [Execute]() ```go func (MapVerb) Execute(_ context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute evaluates the list expression and applies the map expression to each element \(bound as \`\_\`\), writing the resulting list to out\_var. ## type [MergeVerb]() MergeVerb deep\-merges a CEL\-evaluated map into a target variable. Inputs: - "from" \(required, string\): CEL expression that must evaluate to a map. - "into" \(required, string\): target variable name \(dotted ok\). The target's existing value is merged with the from value \(last\-write\-wins per key, recursively for nested maps\). ```go type MergeVerb struct{} ``` ### func \(MergeVerb\) [Execute]() ```go func (MergeVerb) Execute(_ context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute evaluates the from expression to a map and returns dotted\-key entries rooted at into, deep\-merging nested maps. ## type [MetricEmitVerb]() MetricEmitVerb appends a saga\_run\_events row of type \`metric\`. Inputs: - "name" \(required, string\) - "value" \(required, number\) - "labels" \(optional, map\[string\]string\) Prometheus side\-channel wiring is future work; for now the event is enough — admin UI surfaces metric events in the run inspector. ```go type MetricEmitVerb struct { S store.Store } ``` ### func \(MetricEmitVerb\) [Execute]() ```go func (v MetricEmitVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute appends a metric event carrying the name, value, and labels to the run's event stream. ## type [NoopVerb]() NoopVerb does nothing. Useful in tests and during authoring to hold a place in the workflow graph without side effects. ```go type NoopVerb struct{} ``` ### func \(NoopVerb\) [Execute]() ```go func (NoopVerb) Execute(_ context.Context, _ domain.SagaRun, _ domain.Step) (map[string]any, error) ``` Execute does nothing and returns an empty result. ## type [ParallelVerb]() ParallelVerb fans out N child runs and pauses the parent until the join strategy is satisfied. Inputs: - "branches" \(required, \[\]any or CEL string\): each element is a workflow\-fragment object \{"start": "step\_id", "steps": \[...step objects...\]\}. Short\-form \{"type": "...", "inputs": \{...\}\} is also accepted and normalised on the fly. When a string is supplied it is evaluated as a CEL expression against run.Variables and the result must be a non\-empty list. Each branch becomes a child run. - "join\_strategy" \(optional, string, default "all"\): "all" waits for every branch to reach a terminal state. "quorum" wakes the parent once quorum\_n branches have succeeded \(remaining branches keep running but no longer gate the parent\). "first\_terminal" and other values are rejected. - "quorum\_n" \(required when join\_strategy=="quorum"\): positive integer, must be ≤ len\(branches\). The coordinator's child\-terminal hook \(engine/advance.go\) wakes the parent when all children are terminal. The parent then advances to step.Next. ```go type ParallelVerb struct { S store.Store Publisher Publisher } ``` ### func \(ParallelVerb\) [Execute]() ```go func (v ParallelVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute resolves the branches \(literal list or CEL expression\), validates the join strategy, spawns one child run per branch, then pauses the parent awaiting the join and returns ErrSagaPaused. ## type [ParentJoinChecker]() ParentJoinChecker re\-evaluates a parent's parallel/sub\_saga join after a child reaches a terminal state, waking the parent if the join is now satisfied. The Coordinator implements it; CancelVerb uses it so cancelling a target child does not leave a parent paused on a join. ```go type ParentJoinChecker interface { CheckParentJoin(ctx context.Context, run domain.SagaRun) } ``` ## type [Publisher]() Publisher is the minimum surface verbs need to enqueue saga.advance for child runs. Real impl: mq.Publisher. Tests: a fake. Structurally compatible with engine.Publisher so \*mq.Publisher satisfies both. ```go type Publisher interface { PublishSagaAdvance(ctx context.Context, runID string) error } ``` ## type [Registry]() Registry maps step.Type → entry. The engine's Advance loop looks up the entry, applies the license gate \(if non\-common\), then runs Handler.Execute. ```go type Registry map[domain.StepType]RegistryEntry ``` ### func [Default]() ```go func Default(s store.Store, clk clock.Clock, sec secrets.Resolver, pub Publisher, actionPub ActionDispatchPublisher, emitter EventEmitter, opts ...DefaultOption) Registry ``` Default builds the verb registry with license groups attached. The store/clock/secrets/publisher deps are threaded to verbs that need them. The \`end\` step is intentionally NOT in the registry — Advance short\-circuits it. actionPub is the publisher for action dispatch messages. Pass nil in tests that do not exercise action steps — ActionVerb checks for nil. emitter is the EventEmitter used by emit\_event steps. Pass nil in tests that do not exercise emit\_event — EmitEventVerb checks for nil. ## type [RegistryEntry]() RegistryEntry is what each registered StepType resolves to. LicenseGroup lets the engine gate dispatch by the tenant's license features. ```go type RegistryEntry struct { Handler Handler LicenseGroup string // canonical group name, e.g. "external_io_advanced" } ``` ## type [SetVarVerb]() SetVarVerb writes a value to a variable. Inputs: - "out\_var" \(required, string\): the destination variable name \(dotted keys allowed for nested scope writes\). - "value" \(optional\): a literal — written through unchanged. - "expr" \(optional\): a CEL expression evaluated against the current run.Variables; the result is written. Exactly one of "value" or "expr" must be set. If both are set, "expr" wins \(so workflow authors can swap a literal for an expression without renaming the key\). ```go type SetVarVerb struct{} ``` ### func \(SetVarVerb\) [Execute]() ```go func (SetVarVerb) Execute(_ context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute writes either the evaluated expr or the literal value to out\_var, with expr taking precedence when both are present. ## type [SpawnSagaVerb]() SpawnSagaVerb starts a named workflow as a fire\-and\-forget child. The parent continues immediately to step.Next without pausing. The child runs independently; its outcome doesn't block the parent. Implementation note: SpawnChildRun is reused \(same as sub\_saga\) so the child carries a ParentRunID for audit purposes. The "fire\-and\-forget" property is achieved by NOT calling UpdateRunState\(paused\) and NOT returning ErrSagaPaused — the parent advances normally. The coordinator's checkParentJoin guard \(engine/advance.go\) checks that the parent is still paused on the spawning step before waking it, so a fire\-and\-forget child terminating never prematurely wakes a parent that is paused on a later step. Inputs: - "workflow\_id" \(required, string\): the child workflow's stable ID. - "inputs" \(optional, map\[string\]any\): inputs passed to the child. ```go type SpawnSagaVerb struct { S store.Store Publisher Publisher } ``` ### func \(SpawnSagaVerb\) [Execute]() ```go func (v SpawnSagaVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute resolves and spawns the named workflow as a fire\-and\-forget child, then returns an empty result so the parent advances without pausing. ## type [SubSagaVerb]() SubSagaVerb starts a named workflow as a child saga and pauses the parent until the child reaches a terminal state. The coordinator's child\-terminal hook \(engine/advance.go:checkParentJoin\) wakes the parent when all children of this step terminate. Inputs: - "workflow\_id" \(required, string\): the child workflow's stable ID. - "inputs" \(optional, map\[string\]any\): inputs passed to the child. Note: sub\_saga reuses the same child\-run \+ WakeFromExternal mechanism as \`parallel\` — by definition there's exactly one "branch" \(the child\). ```go type SubSagaVerb struct { S store.Store Publisher Publisher } ``` ### func \(SubSagaVerb\) [Execute]() ```go func (v SubSagaVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute resolves and spawns the named workflow as a child saga, then pauses the parent awaiting the child's terminal state and returns ErrSagaPaused. ## type [SwitchVerb]() SwitchVerb evaluates a CEL expression over run.Variables to a string branch key and returns \{"branch": key\}; the engine routes that to step.Branches\[key\].Next. Inputs: - "expr" \(required, CEL string\) must evaluate to a string. ```go type SwitchVerb struct{} ``` ### func \(SwitchVerb\) [Execute]() ```go func (SwitchVerb) Execute(_ context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute compiles and evaluates the CEL expr against run.Variables, asserts the result is a string, and returns \{"branch": \\}. ## type [TransformVerb]() TransformVerb evaluates a CEL expression against current Variables and writes the result to out\_var. Inputs: - "expr" \(required, string\): CEL expression. - "out\_var" \(required, string\): variable name to write to \(dotted ok\). ```go type TransformVerb struct{} ``` ### func \(TransformVerb\) [Execute]() ```go func (TransformVerb) Execute(_ context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute evaluates the CEL expr against run.Variables and writes the result to out\_var. ## type [TryCatchVerb]() TryCatchVerb pushes a try\_catch frame onto the saga's stack. When a step inside the try body errors, the coordinator pops the top frame and advances to the catch step \(with the error context written to Variables.\_error\). On success, the body's last step's \`next\` should point to whatever comes after the try block — the frame stays on the stack until the saga terminates \(acceptable for v1 since max nesting depth is 3 per the publish\-time validator\). Inputs: - "try" \(required, \[\]any of step IDs\): metadata used by the validator \(engine.ValidateDefinition\) to reject parallel\-in\-try. Not consumed by the runtime — author wires step.Next to the first try step. - "catch" \(required, string\): step ID to jump to on error. ```go type TryCatchVerb struct { S store.Store } ``` ### func \(TryCatchVerb\) [Execute]() ```go func (v TryCatchVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute pushes a try\_catch frame \(recording the catch step\) onto the run's stack and returns an empty result so the engine routes to the first try step. ## type [WaitDurationVerb]() WaitDurationVerb pauses the saga for a fixed duration. Inputs: - "duration" \(required, string\): Go duration syntax e.g. "5s", "1h30m". Persists wakeup\_at on the saga run and returns ErrSagaPaused; the coordinator catches the sentinel, appends EventStepPaused, and ACKs the queue message. The timer dispatcher polls for due wakeups and republishes saga.advance. ```go type WaitDurationVerb struct { S store.Store Clock clock.Clock } ``` ### func \(WaitDurationVerb\) [Execute]() ```go func (v WaitDurationVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute parses the duration, persists wakeup\_at = now \+ duration on the run, and returns ErrSagaPaused for the timer dispatcher to resume. ## type [WaitForEventVerb]() WaitForEventVerb pauses the saga until a RabbitMQ event with a matching topic \+ header subset arrives. v1 only does string\-equality header filtering \(CEL on payload deferred to a later batch\). Inputs: - "topic" \(required, string\): RabbitMQ routing key the saga awaits. - "headers" \(optional, map\[string\]any\): header→value pairs the incoming event's headers must all match \(values stringified\). - "timeout\_s" \(optional, number\): max seconds to wait. On timeout the engine routes to the step's "timeout" branch if defined, else to Next. Omitted = wait indefinitely. ```go type WaitForEventVerb struct { S store.Store Clock clock.Clock } ``` ### func \(WaitForEventVerb\) [Execute]() ```go func (v WaitForEventVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute persists the awaited topic, header filter, and optional timeout deadline on the run and returns ErrSagaPaused until a matching event arrives or the deadline elapses. ## type [WaitForSignalVerb]() WaitForSignalVerb pauses the saga until a named external signal arrives via POST /api/v1/sagas/\{run\_id\}/signal/\{name\}. Inputs: - "name" \(required, string\): signal name to await. - "timeout\_s" \(optional, float64\): max seconds to wait before the timer dispatcher wakes the saga regardless. If omitted the saga waits indefinitely \(no wakeup\_at set by the verb itself; the signal handler sets wakeup\_at=now\(\) when it arrives\). On signal arrival: TryConsumeAwaitedSignal clears the await markers and sets wakeup\_at=now\(\). The signal REST handler publishes saga.advance; Advance sees paused\+due\-wakeup and resumes from the next step uniformly. ```go type WaitForSignalVerb struct { S store.Store Clock clock.Clock } ``` ### func \(WaitForSignalVerb\) [Execute]() ```go func (v WaitForSignalVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute persists the awaited signal name \(and optional timeout deadline\) on the run and returns ErrSagaPaused until the signal arrives or times out. ## type [WaitUntilVerb]() WaitUntilVerb pauses the saga until a wall\-clock instant. Inputs: - "timestamp" \(required, string\): RFC3339 timestamp \(e.g. "2026\-12\-31T23:59:00Z"\). If the timestamp is in the past relative to the engine's clock, the verb sets wakeup\_at to clock.Now\(\) so the next timer tick wakes the saga immediately. This mirrors \`wait\_duration\` but with an absolute rather than relative deadline. ```go type WaitUntilVerb struct { S store.Store Clock clock.Clock } ``` ### func \(WaitUntilVerb\) [Execute]() ```go func (v WaitUntilVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute parses the RFC3339 timestamp, persists it as wakeup\_at \(clamped to now if in the past\), and returns ErrSagaPaused for the timer to resume. ## type [WebhookEmitVerb]() WebhookEmitVerb POSTs a payload to an external URL. Inputs: - "url" \(required, string\) - "body" \(required, any; JSON\-marshalled into the request body\) - "secret\_ref" \(optional, string; resolves to an HMAC\-SHA256 signing key. When present, X\-Webhook\-Sig header = "sha256=\" of the body using the key.\) - "timeout\_s" \(optional, number; default 15s\) - "headers" \(optional, map\[string\]any → string headers\) - "async" \(optional, bool; default false. When true: fire request in a goroutine, return immediately without awaiting response. Failures are logged but not surfaced. Use for fire\-and\-forget notifications.\) - "out\_var" \(optional, string; default "webhook\_result"\). Only populated on synchronous \(non\-async\) success: \{out\_var\}\_status = int64 code. ```go type WebhookEmitVerb struct { Secrets secrets.Resolver Client *http.Client } ``` ### func \(WebhookEmitVerb\) [Execute]() ```go func (v WebhookEmitVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute POSTs the JSON\-marshalled body to the URL \(optionally HMAC\-signed\). In async mode it fires the request in a goroutine and returns immediately; otherwise it awaits the response and returns the status code under out\_var. ## type [WhileVerb]() WhileVerb evaluates a CEL condition and chooses a branch. Inputs: - "condition" \(required, string\): CEL expression evaluated against Variables. - "max\_iterations" \(optional, number, default 100, hard cap 10000\): if the per\-step iteration counter reaches this, the verb returns an error. Cap prevents runaway loops. The verb returns a \{"branch": "continue"|"exit"\} output map. The workflow author wires step.Branches: - "continue" → next: body's first step. - "exit" → next: step after the loop. The body's last step should point its next back at this while step so the loop closes. The iteration counter persists at Variables.\_while.\{step.ID\}.iter \(int64\). ```go type WhileVerb struct{} ``` ### func \(WhileVerb\) [Execute]() ```go func (WhileVerb) Execute(_ context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error) ``` Execute evaluates the condition, increments the persisted iteration counter, and returns branch "continue" \(condition true\) or "exit" \(false\). It errors once the iteration count reaches max\_iterations. Generated by [gomarkdoc]() --- # saga Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/saga # saga ```go import "github.com/Bugs5382/go-saga-orchestration/saga" ``` Package saga is the embedding entrypoint: construct an in\-process saga engine, register workflows and custom verbs, and drive runs — without running the engine binaries. ## type [InProcessEventEmitter]() InProcessEventEmitter delivers an emitted event in\-process, with no broker, mirroring service mode where the same event feeds both the event subscriber \(which wakes awaiting runs\) and the trigger dispatcher \(which starts new runs\): - wakes runs awaiting the topic, applying the same header\-subset match as engine.EventSubscriber, then - runs the trigger dispatcher so matching triggers start new runs. ```go type InProcessEventEmitter struct { // contains filtered or unexported fields } ``` ### func \(\*InProcessEventEmitter\) [EmitEvent]() ```go func (e *InProcessEventEmitter) EmitEvent(ctx context.Context, topic string, headers map[string]string, payload map[string]any) error ``` EmitEvent wakes paused runs awaiting topic \(header\-subset match\) and then runs the trigger dispatcher against the event so matching triggers start new runs. ## type [InProcessPublisher]() InProcessPublisher satisfies engine.Publisher \(and verbs.ActionDispatchPublisher\) for embedded use with no message broker. PublishSagaAdvance runs the coordinator's Advance in a background goroutine bound to the Saga's context and tracked on its WaitGroup, so Saga.Shutdown can cancel and drain in\-flight work. Action dispatch is unsupported in\-process \(use a worker / the service mode\). ```go type InProcessPublisher struct { // contains filtered or unexported fields } ``` ### func \(\*InProcessPublisher\) [PublishActionDispatch]() ```go func (p *InProcessPublisher) PublishActionDispatch(_ context.Context, _ string, _ []byte) error ``` ### func \(\*InProcessPublisher\) [PublishSagaAdvance]() ```go func (p *InProcessPublisher) PublishSagaAdvance(_ context.Context, runID string) error ``` PublishSagaAdvance advances the run in a background goroutine. Semantics for embedders: this is only reached by workflows that spawn child runs \(parallel / foreach / spawn\_saga\); linear workflows advance synchronously inside Start and never use the publisher. The advance runs on the Saga's context \(derived from Options.Context\) and is registered on the Saga's WaitGroup, so Saga.Shutdown cancels it \(Advance stops between steps\) and waits for it to drain. Errors cannot be returned to the caller; set Options.Logger to observe them. ## type [Options]() Options configures a Saga engine instance. ```go type Options struct { Store store.Store Clock clock.Clock Licensing licensing.Resolver Secrets secrets.Resolver Publisher engine.Publisher StartupProviders []engine.StartupVariableProvider Logger *zerolog.Logger // optional; nil = no logging // Context is the base context for background work (parallel/foreach/spawn // child advances run on a cancellable context derived from it). Defaults to // context.Background(). Shutdown cancels the derived context. Context context.Context } ``` ## type [Saga]() Saga is the embedding facade around the coordinator and store. ```go type Saga struct { // contains filtered or unexported fields } ``` ### func [InMemory]() ```go func InMemory() *Saga ``` InMemory returns a Saga backed by an in\-memory store with all defaults. Convenient for tests and examples. ### func [New]() ```go func New(opts Options) (*Saga, error) ``` New constructs a Saga from opts. opts.Store is required; all other fields have sensible defaults \(SystemClock, in\-memory secrets, StubAllowAll licensing, in\-process publisher\). ### func \(\*Saga\) [Cancel]() ```go func (s *Saga) Cancel(ctx context.Context, runID uuid.UUID, reason string) error ``` Cancel terminates an in\-flight run from outside the run — e.g. an approval policy withdrawing or re\-submitting while a run is paused at a manual\_approval. The run transitions to terminal cancelled, its open user tasks are closed \(so none linger pending\), and any awaited signal/event or pending wakeup is cleared so a stray advance cannot resurrect it. reason is recorded on the run's last\_error. Idempotent: a no\-op when the run is already terminal. See issue \#80. ### func \(\*Saga\) [Coordinator]() ```go func (s *Saga) Coordinator() *engine.Coordinator ``` Coordinator returns the underlying engine.Coordinator for advanced use. ### func \(\*Saga\) [Get]() ```go func (s *Saga) Get(ctx context.Context, runID uuid.UUID) (domain.SagaRun, error) ``` Get returns the current state of a run by ID. ### func \(\*Saga\) [Register]() ```go func (s *Saga) Register(def domain.WorkflowDefinition) error ``` Register upserts a workflow definition into the store so it can be started. ### func \(\*Saga\) [RegisterVerb]() ```go func (s *Saga) RegisterVerb(stepType string, licenseGroup string, h verbs.Handler) ``` RegisterVerb adds or replaces a verb handler identified by stepType in the coordinator's registry. ### func \(\*Saga\) [Shutdown]() ```go func (s *Saga) Shutdown(ctx context.Context) error ``` Shutdown cancels the Saga's background context \(so in\-flight background advances stop between steps\) and waits for them to drain, bounded by ctx. Returns ctx.Err\(\) if the drain does not complete before ctx is done. After Shutdown the Saga should not be reused. ### func \(\*Saga\) [Signal]() ```go func (s *Saga) Signal(ctx context.Context, runID uuid.UUID, name string, payload map[string]any) error ``` Signal delivers an external signal to a run. If the run was paused awaiting exactly this signal name, it is consumed and the run advances. ### func \(\*Saga\) [Start]() ```go func (s *Saga) Start(ctx context.Context, workflowID string, inputs map[string]any) (uuid.UUID, error) ``` ### func \(\*Saga\) [StartAt]() ```go func (s *Saga) StartAt(ctx context.Context, workflowID, entrypoint string, inputs map[string]any) (uuid.UUID, error) ``` StartAt creates a run beginning at the named entry point \("" =\> default/Start\) and advances it once \(synchronously to the first pause or terminal state\). ### func \(\*Saga\) [Store]() ```go func (s *Saga) Store() store.Store ``` Store returns the underlying store.Store for direct access. Generated by [gomarkdoc]() --- # secrets Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/secrets # secrets ```go import "github.com/Bugs5382/go-saga-orchestration/secrets" ``` Package secrets resolves a secret\_ref string \(e.g. "vault://path/to/key"\) into a value at runtime. It ships an in\-memory map\-backed stub; real Vault/SealedSecrets integration is a platform\-services concern. ## type [Memory]() Memory is a test\-time / dev\-time resolver. Production wires a real impl. ```go type Memory struct { M map[string]string } ``` ### func [NewMemory]() ```go func NewMemory(m map[string]string) *Memory ``` NewMemory returns a Memory resolver backed by the given ref\-to\-value map. ### func \(\*Memory\) [Get]() ```go func (r *Memory) Get(ref string) (string, error) ``` Get returns the value for ref, or an error if the ref is not in the map. ## type [Resolver]() Resolver looks up a secret by reference. ```go type Resolver interface { Get(ref string) (string, error) } ``` Generated by [gomarkdoc]() --- # store Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/store # store ```go import "github.com/Bugs5382/go-saga-orchestration/store" ``` Package store defines the persistence interface go\-saga\-orchestration uses for definitions, runs, events, and registry rows. Two implementations: - memory/ — in\-memory, test\-only. - postgres/ — production. Engine \+ API both depend on the interface, never on a concrete impl. ## type [ActionFilter]() ActionFilter narrows ListActions queries. All fields optional. ```go type ActionFilter struct { Service string Search string // substring match against action name (case-sensitive) Category string } ``` ## type [ErrNotFound]() ErrNotFound is returned by Get\* methods when no row matches. ```go type ErrNotFound struct { Entity string ID string } ``` ### func \(ErrNotFound\) [Error]() ```go func (e ErrNotFound) Error() string ``` Error reports the missing entity and ID. ## type [RunFilter]() RunFilter narrows ListRuns / CountRuns queries. All fields optional. ```go type RunFilter struct { WorkflowID string // empty = any State string // empty = any; values per domain.RunState TriggerType string // empty = any; checks saga_triggers.trigger_type via Run.TriggerID Since *time.Time // started_at >= Since when set HasError *bool // nil = any; true = state==failed; v1: state==failed is sufficient (noted) RequiresReview *bool // nil = any; checks SagaRun.RequiresManualReview Limit int // server caps at 500; 0 → 50 default Offset int } ``` ## type [Store]() Store is the persistence surface. ```go type Store interface { // Definitions GetWorkflowDefinition(ctx context.Context, id uuid.UUID) (domain.WorkflowDefinition, error) GetPublishedWorkflowByID(ctx context.Context, workflowID string, tenantID *uuid.UUID) (domain.WorkflowDefinition, error) UpsertWorkflowDefinition(ctx context.Context, def domain.WorkflowDefinition) (uuid.UUID, error) // Runs CreateRun(ctx context.Context, run domain.SagaRun) error GetRun(ctx context.Context, id uuid.UUID) (domain.SagaRun, error) UpdateRunState(ctx context.Context, id uuid.UUID, state domain.RunState, currentStep string) error // Cancel terminates an in-flight run from outside the run (e.g. an // approval policy withdrawing a paused manual_approval). It atomically: // transitions the run to the terminal "cancelled" state and stamps // terminal_at; records reason in last_error; closes the run's open user // tasks (so none linger pending in an approver's inbox); and clears any // awaited signal / event / pending wakeup so a stray Advance cannot // resurrect it. Idempotent: a no-op (returns nil, emits no event) when // the run is already terminal. See issue #80. Cancel(ctx context.Context, runID uuid.UUID, reason string) error // MarkRunFailed transitions a run to the terminal "failed" state, // stamps terminal_at, and persists the failing step's error in // last_error so a failed run is self-describing. Idempotent: a no-op // when the run is already terminal. See issue #80. MarkRunFailed(ctx context.Context, runID uuid.UUID, currentStep, lastError string) error // ListRuns returns saga runs matching the optional filter. Sorted by // started_at DESC (newest first). Limit + Offset are for pagination; // caller must validate (zero Limit → 50 default; hard max 500 enforced // server-side before this call). ListRuns(ctx context.Context, filter RunFilter) ([]domain.SagaRun, error) // CountRuns returns the total count matching filter (ignoring Limit/Offset) // so callers can render total + page nav. CountRuns(ctx context.Context, filter RunFilter) (int, error) // StatsForWorkflow returns aggregate metrics for a workflow. StatsForWorkflow(ctx context.Context, workflowID string) (WorkflowStats, error) // Events AppendEvent(ctx context.Context, evt domain.SagaRunEvent) error ListEventsByRun(ctx context.Context, runID uuid.UUID) ([]domain.SagaRunEvent, error) // GetEventByID returns a single audit event by its UUID, or ErrNotFound. GetEventByID(ctx context.Context, id uuid.UUID) (domain.SagaRunEvent, error) // Rule definitions. UpsertRuleDefinition(ctx context.Context, def domain.RuleDefinition) (uuid.UUID, error) GetPublishedRuleByID(ctx context.Context, ruleID string, tenantID *uuid.UUID) (domain.RuleDefinition, error) // Run variable mutation (merge a verb result map into saga_runs.variables JSONB). UpdateRunVariables(ctx context.Context, runID uuid.UUID, merge map[string]any) error // Pause / resume helpers. SetPausedWithWakeup(ctx context.Context, runID uuid.UUID, wakeupAt time.Time) error SetPausedAwaitingSignal(ctx context.Context, runID uuid.UUID, signalName string, deadline *time.Time) error SetPausedAwaitingEvent(ctx context.Context, runID uuid.UUID, topic string, headers map[string]string) error // SetPausedAwaitingEventWithDeadline is SetPausedAwaitingEvent but also sets a // wakeup deadline (nil = no deadline). On deadline elapse with the event still // unmatched, the engine routes to the step's "timeout" branch if defined. SetPausedAwaitingEventWithDeadline(ctx context.Context, runID uuid.UUID, topic string, headers map[string]string, deadline *time.Time) error ClearPause(ctx context.Context, runID uuid.UUID) error FindRunsByDueWakeup(ctx context.Context, now time.Time, limit int) ([]uuid.UUID, error) FindRunsByAwaitedEvent(ctx context.Context, topic string) ([]domain.SagaRun, error) TryConsumeAwaitedSignal(ctx context.Context, runID uuid.UUID, signalName string) (ok bool, err error) AppendSignal(ctx context.Context, sig domain.SagaSignal) error // WakeFromExternal clears all await-markers (awaited_signal, // awaited_event_topic, awaited_event_headers) and sets wakeup_at=now() // while leaving state=paused. The Advance loop sees paused+due-wakeup // and resumes the saga uniformly, regardless of whether a signal or // event triggered the wake. WakeFromExternal(ctx context.Context, runID uuid.UUID) error // Child runs / try_catch stack. SpawnChildRun(ctx context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any) (uuid.UUID, error) // SpawnChildRunAt is SpawnChildRun but the child begins at startStep // ("" => the child definition's Start). SpawnChildRun delegates with "". SpawnChildRunAt(ctx context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any, startStep string) (uuid.UUID, error) ListChildrenByParent(ctx context.Context, parentID uuid.UUID, parentStepID string) ([]domain.SagaRun, error) PushTryCatch(ctx context.Context, runID uuid.UUID, frame domain.TryCatchFrame) error PopTryCatch(ctx context.Context, runID uuid.UUID) (domain.TryCatchFrame, bool, error) // User tasks. CreateUserTask(ctx context.Context, task domain.UserTask) error GetUserTask(ctx context.Context, taskID uuid.UUID) (domain.UserTask, error) SubmitUserTask(ctx context.Context, taskID uuid.UUID, submittedBy string, result map[string]any) error // ListUserTasksByRun returns all user tasks created during runID's // execution (any step). Returned in creation order (by ID, since // domain.UserTask has no CreatedAt field). Empty list if none. ListUserTasksByRun(ctx context.Context, runID uuid.UUID) ([]domain.UserTask, error) // Action registry. UpsertActionRegistration(ctx context.Context, reg domain.ActionRegistration) error ListActions(ctx context.Context, filter ActionFilter) ([]domain.ActionRegistration, error) GetAction(ctx context.Context, service, name string, version int) (domain.ActionRegistration, error) // Saga triggers. UpsertTrigger(ctx context.Context, trigger domain.SagaTrigger) (uuid.UUID, error) GetTrigger(ctx context.Context, id uuid.UUID) (domain.SagaTrigger, error) ListTriggers(ctx context.Context, filter TriggerFilter) ([]domain.SagaTrigger, error) DeleteTrigger(ctx context.Context, id uuid.UUID) error // ListDueCronTriggers returns enabled cron triggers whose next_fire_at is at // or before now, oldest first, capped at limit. ListDueCronTriggers(ctx context.Context, now time.Time, limit int) ([]domain.SagaTrigger, error) // ClaimCronFire atomically advances a cron trigger's next_fire_at from // expectedNextFire to newNextFire (and stamps last_fired_at). Returns true // iff this caller won the row — the single-fire guarantee across pods. ClaimCronFire(ctx context.Context, id uuid.UUID, expectedNextFire, newNextFire time.Time) (bool, error) // RecordTriggerFire inserts a row into runtime.saga_trigger_fires. Best-effort: // callers must log errors but must not abort the triggering run on failure. // runID is nil when the fire failed before a run was created. fireErr is empty // on success. RecordTriggerFire(ctx context.Context, triggerID uuid.UUID, workflowID string, runID *uuid.UUID, fireErr string) error // Action dispatch tracking. // MarkAwaitingAction sets state=paused, records the dispatch key and attempt. // Idempotent on (runID, attempt) — same args = no-op. MarkAwaitingAction(ctx context.Context, runID uuid.UUID, dispatch string, attempt int) error // CompleteAction clears the await marker and merges result into variables. // If attempt does not match current_attempt the call is a no-op (late delivery). CompleteAction(ctx context.Context, runID uuid.UUID, attempt int, result map[string]any) error // FailAction transitions the run to failed and appends an audit event. // If attempt does not match current_attempt the call is a no-op (late delivery). FailAction(ctx context.Context, runID uuid.UUID, attempt int, code, message string, retryable bool) error } ``` ## type [TriggerFilter]() TriggerFilter narrows ListTriggers queries. All fields optional. ```go type TriggerFilter struct { Type domain.TriggerType Enabled *bool // nil = any TenantID *uuid.UUID // nil = any } ``` ## type [WorkflowStats]() WorkflowStats holds aggregate metrics for a single workflow. ```go type WorkflowStats struct { WorkflowID string `json:"workflow_id"` SuccessRate24h *float64 `json:"success_rate_24h"` // null when no runs in last 24h LastRunAt *time.Time `json:"last_run_at"` // null when no runs at all InFlight int `json:"in_flight"` } ``` Generated by [gomarkdoc]() --- # memory Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/store/memory # memory ```go import "github.com/Bugs5382/go-saga-orchestration/store/memory" ``` Package memory is an in\-process store used by unit tests. Production code uses store/postgres. ## type [Store]() Store is the in\-memory store. Safe for concurrent use. ```go type Store struct { // contains filtered or unexported fields } ``` ### func [New]() ```go func New() *Store ``` New returns an empty Store. Compile\-time check confirms it satisfies the interface. ### func \(\*Store\) [AppendEvent]() ```go func (s *Store) AppendEvent(_ context.Context, evt domain.SagaRunEvent) error ``` AppendEvent appends evt to the event list for evt.RunID. ### func \(\*Store\) [AppendSignal]() ```go func (s *Store) AppendSignal(_ context.Context, sig domain.SagaSignal) error ``` AppendSignal appends sig to the signal list for sig.RunID. ### func \(\*Store\) [Cancel]() ```go func (s *Store) Cancel(ctx context.Context, runID uuid.UUID, reason string) error ``` Cancel terminates an in\-flight run: terminal cancelled \+ terminal\_at, reason in last\_error, open user tasks closed, await markers / wakeup cleared. Idempotent — no\-op \(and no event\) once the run is terminal. See issue \#80. ### func \(\*Store\) [ClaimCronFire]() ```go func (s *Store) ClaimCronFire(_ context.Context, id uuid.UUID, expectedNextFire, newNextFire time.Time) (bool, error) ``` ClaimCronFire atomically advances next\_fire\_at from expectedNextFire to newNextFire and stamps last\_fired\_at. Returns true iff this caller won \(the CAS matched\). Returns false without error when the expected value no longer matches — the caller lost the race. ### func \(\*Store\) [ClearPause]() ```go func (s *Store) ClearPause(_ context.Context, runID uuid.UUID) error ``` ClearPause returns the run to the running state and clears all wakeup and await markers, or returns ErrNotFound. ### func \(\*Store\) [CompleteAction]() ```go func (s *Store) CompleteAction(ctx context.Context, runID uuid.UUID, attempt int, result map[string]any) error ``` CompleteAction merges result into Variables and sets wakeup\_at=now\(\) so the Advance paused\-handling loop resumes the saga. If attempt \!= current\_attempt it is a late/duplicate delivery and is silently ignored. ### func \(\*Store\) [CountRuns]() ```go func (s *Store) CountRuns(_ context.Context, filter store.RunFilter) (int, error) ``` CountRuns returns the total count matching filter \(ignoring Limit/Offset\). ### func \(\*Store\) [CreateRun]() ```go func (s *Store) CreateRun(_ context.Context, run domain.SagaRun) error ``` CreateRun stores run keyed by its ID. ### func \(\*Store\) [CreateUserTask]() ```go func (s *Store) CreateUserTask(_ context.Context, task domain.UserTask) error ``` CreateUserTask stores a new UserTask keyed by its ID. ### func \(\*Store\) [DeleteTrigger]() ```go func (s *Store) DeleteTrigger(_ context.Context, id uuid.UUID) error ``` DeleteTrigger removes the trigger from the store. Returns ErrNotFound if it does not exist. ### func \(\*Store\) [FailAction]() ```go func (s *Store) FailAction(ctx context.Context, runID uuid.UUID, attempt int, code, message string, retryable bool) error ``` FailAction transitions the run to failed and appends an audit event. If attempt \!= current\_attempt it is a late delivery and is silently ignored. ### func \(\*Store\) [FindRunsByAwaitedEvent]() ```go func (s *Store) FindRunsByAwaitedEvent(_ context.Context, topic string) ([]domain.SagaRun, error) ``` FindRunsByAwaitedEvent returns paused runs awaiting an event on topic. ### func \(\*Store\) [FindRunsByDueWakeup]() ```go func (s *Store) FindRunsByDueWakeup(_ context.Context, now time.Time, limit int) ([]uuid.UUID, error) ``` FindRunsByDueWakeup returns up to limit IDs of paused runs whose wakeup\_at is at or before now. ### func \(\*Store\) [GetAction]() ```go func (s *Store) GetAction(_ context.Context, service, name string, version int) (domain.ActionRegistration, error) ``` GetAction returns the registration for service\+name\+version, or ErrNotFound. ### func \(\*Store\) [GetEventByID]() ```go func (s *Store) GetEventByID(_ context.Context, id uuid.UUID) (domain.SagaRunEvent, error) ``` GetEventByID returns the first event whose ID matches, or ErrNotFound. ### func \(\*Store\) [GetPublishedRuleByID]() ```go func (s *Store) GetPublishedRuleByID(_ context.Context, ruleID string, _ *uuid.UUID) (domain.RuleDefinition, error) ``` GetPublishedRuleByID returns the newest published version of ruleID, falling back to the most recent version if none is published, or ErrNotFound. ### func \(\*Store\) [GetPublishedWorkflowByID]() ```go func (s *Store) GetPublishedWorkflowByID(_ context.Context, workflowID string, _ *uuid.UUID) (domain.WorkflowDefinition, error) ``` GetPublishedWorkflowByID returns the newest published version of workflowID, falling back to the most recent version if none is published, or ErrNotFound. ### func \(\*Store\) [GetRun]() ```go func (s *Store) GetRun(_ context.Context, id uuid.UUID) (domain.SagaRun, error) ``` GetRun returns the run with the given ID, or ErrNotFound. ### func \(\*Store\) [GetTrigger]() ```go func (s *Store) GetTrigger(_ context.Context, id uuid.UUID) (domain.SagaTrigger, error) ``` GetTrigger returns the SagaTrigger for id, or ErrNotFound. ### func \(\*Store\) [GetUserTask]() ```go func (s *Store) GetUserTask(_ context.Context, taskID uuid.UUID) (domain.UserTask, error) ``` GetUserTask returns the task or ErrNotFound. ### func \(\*Store\) [GetWorkflowDefinition]() ```go func (s *Store) GetWorkflowDefinition(_ context.Context, id uuid.UUID) (domain.WorkflowDefinition, error) ``` GetWorkflowDefinition returns the definition with the given storage ID, or ErrNotFound. ### func \(\*Store\) [ListActions]() ```go func (s *Store) ListActions(_ context.Context, filter store.ActionFilter) ([]domain.ActionRegistration, error) ``` ListActions returns all registrations matching the optional filter fields. ### func \(\*Store\) [ListChildrenByParent]() ```go func (s *Store) ListChildrenByParent(_ context.Context, parentID uuid.UUID, parentStepID string) ([]domain.SagaRun, error) ``` ListChildrenByParent returns all runs whose ParentRunID == parentID and ParentStepID == parentStepID. ### func \(\*Store\) [ListDueCronTriggers]() ```go func (s *Store) ListDueCronTriggers(_ context.Context, now time.Time, limit int) ([]domain.SagaTrigger, error) ``` ListDueCronTriggers returns enabled cron triggers whose next\_fire\_at is at or before now, sorted oldest\-first, capped at limit. ### func \(\*Store\) [ListEventsByRun]() ```go func (s *Store) ListEventsByRun(_ context.Context, runID uuid.UUID) ([]domain.SagaRunEvent, error) ``` ListEventsByRun returns a copy of the events recorded for runID. ### func \(\*Store\) [ListRuns]() ```go func (s *Store) ListRuns(_ context.Context, filter store.RunFilter) ([]domain.SagaRun, error) ``` ListRuns returns saga runs matching filter, sorted by StartedAt DESC. TriggerType filter: iterates s.triggers to build an id→type map, then checks each run's TriggerID. ### func \(\*Store\) [ListTriggers]() ```go func (s *Store) ListTriggers(_ context.Context, filter store.TriggerFilter) ([]domain.SagaTrigger, error) ``` ListTriggers returns triggers matching the optional filter. ### func \(\*Store\) [ListUserTasksByRun]() ```go func (s *Store) ListUserTasksByRun(_ context.Context, runID uuid.UUID) ([]domain.UserTask, error) ``` ListUserTasksByRun returns all user tasks whose RunID matches runID. Returned in insertion order by task.ID \(domain.UserTask has no CreatedAt field, so ID order is used as a stable proxy for creation order\). ### func \(\*Store\) [MarkAwaitingAction]() ```go func (s *Store) MarkAwaitingAction(_ context.Context, runID uuid.UUID, dispatch string, attempt int) error ``` MarkAwaitingAction sets state=paused and records the dispatch key \+ attempt. Idempotent on \(runID, attempt\): if the current\_attempt already equals attempt and the dispatch key is the same, the call is a no\-op. ### func \(\*Store\) [MarkRunFailed]() ```go func (s *Store) MarkRunFailed(_ context.Context, runID uuid.UUID, currentStep, lastError string) error ``` MarkRunFailed transitions a run to terminal failed, stamps terminal\_at, and persists lastError on the run. Idempotent on terminal runs. See issue \#80. ### func \(\*Store\) [PopTryCatch]() ```go func (s *Store) PopTryCatch(_ context.Context, runID uuid.UUID) (domain.TryCatchFrame, bool, error) ``` PopTryCatch removes and returns the top TryCatchFrame. Returns \(zero, false, nil\) when the stack is empty. ### func \(\*Store\) [PushTryCatch]() ```go func (s *Store) PushTryCatch(_ context.Context, runID uuid.UUID, frame domain.TryCatchFrame) error ``` PushTryCatch appends frame to the run's TryCatchStack. Returns an error if the stack is already at maximum depth \(3\). ### func \(\*Store\) [RecordTriggerFire]() ```go func (s *Store) RecordTriggerFire(_ context.Context, triggerID uuid.UUID, workflowID string, runID *uuid.UUID, fireErr string) error ``` RecordTriggerFire appends a TriggerFireRow to the in\-memory audit log. ### func \(\*Store\) [SetPausedAwaitingEvent]() ```go func (s *Store) SetPausedAwaitingEvent(ctx context.Context, runID uuid.UUID, topic string, headers map[string]string) error ``` SetPausedAwaitingEvent marks the run paused awaiting an event matching topic and the given header filter, or returns ErrNotFound. ### func \(\*Store\) [SetPausedAwaitingEventWithDeadline]() ```go func (s *Store) SetPausedAwaitingEventWithDeadline(_ context.Context, runID uuid.UUID, topic string, headers map[string]string, deadline *time.Time) error ``` SetPausedAwaitingEventWithDeadline is SetPausedAwaitingEvent plus an optional wakeup deadline \(nil = wait indefinitely\). ### func \(\*Store\) [SetPausedAwaitingSignal]() ```go func (s *Store) SetPausedAwaitingSignal(_ context.Context, runID uuid.UUID, signalName string, deadline *time.Time) error ``` SetPausedAwaitingSignal marks the run paused awaiting signalName, setting an optional wakeup deadline, or returns ErrNotFound. ### func \(\*Store\) [SetPausedWithWakeup]() ```go func (s *Store) SetPausedWithWakeup(_ context.Context, runID uuid.UUID, wakeupAt time.Time) error ``` SetPausedWithWakeup marks the run paused and sets its wakeup\_at, or returns ErrNotFound. ### func \(\*Store\) [SpawnChildRun]() ```go func (s *Store) SpawnChildRun(ctx context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any) (uuid.UUID, error) ``` SpawnChildRun creates a child run linked to parentID / parentStepID / branchKey, beginning at the child definition's default Start step. ### func \(\*Store\) [SpawnChildRunAt]() ```go func (s *Store) SpawnChildRunAt(_ context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any, startStep string) (uuid.UUID, error) ``` SpawnChildRunAt creates a child run linked to parentID / parentStepID / branchKey, beginning at startStep \(empty string means the child definition's Start field\). ### func \(\*Store\) [StatsForWorkflow]() ```go func (s *Store) StatsForWorkflow(_ context.Context, workflowID string) (store.WorkflowStats, error) ``` StatsForWorkflow computes aggregate metrics for workflowID by iterating s.runs. ### func \(\*Store\) [SubmitUserTask]() ```go func (s *Store) SubmitUserTask(_ context.Context, taskID uuid.UUID, submittedBy string, result map[string]any) error ``` SubmitUserTask marks the task as submitted. Idempotent \(re\-writes on repeated calls\). Returns ErrNotFound if the task does not exist. ### func \(\*Store\) [TriggerFires]() ```go func (s *Store) TriggerFires() []domain.TriggerFireRow ``` TriggerFires returns a copy of all recorded trigger\-fire rows. Intended for test assertions only. ### func \(\*Store\) [TryConsumeAwaitedSignal]() ```go func (s *Store) TryConsumeAwaitedSignal(_ context.Context, runID uuid.UUID, signalName string) (bool, error) ``` TryConsumeAwaitedSignal clears the await markers and marks any matching unconsumed signal consumed when the run is paused awaiting signalName, reporting whether it did so. ### func \(\*Store\) [UpdateRunState]() ```go func (s *Store) UpdateRunState(_ context.Context, id uuid.UUID, state domain.RunState, currentStep string) error ``` UpdateRunState sets the run's state and current step, or returns ErrNotFound. ### func \(\*Store\) [UpdateRunVariables]() ```go func (s *Store) UpdateRunVariables(_ context.Context, runID uuid.UUID, merge map[string]any) error ``` UpdateRunVariables merges the entries of merge into the run's variables, honouring dotted keys for nested writes, or returns ErrNotFound. ### func \(\*Store\) [UpsertActionRegistration]() ```go func (s *Store) UpsertActionRegistration(_ context.Context, reg domain.ActionRegistration) error ``` UpsertActionRegistration stores or replaces the registration keyed by service\+name\+version. ### func \(\*Store\) [UpsertRuleDefinition]() ```go func (s *Store) UpsertRuleDefinition(_ context.Context, def domain.RuleDefinition) (uuid.UUID, error) ``` UpsertRuleDefinition stores def, preserving a caller\-supplied ID \(or generating one\) and replacing any prior entry with the same ID. ### func \(\*Store\) [UpsertTrigger]() ```go func (s *Store) UpsertTrigger(_ context.Context, trigger domain.SagaTrigger) (uuid.UUID, error) ``` UpsertTrigger inserts or replaces a SagaTrigger. If trigger.ID == uuid.Nil a new ID is generated. If trigger.ID is set and a row already exists it is replaced in full. ### func \(\*Store\) [UpsertWorkflowDefinition]() ```go func (s *Store) UpsertWorkflowDefinition(_ context.Context, def domain.WorkflowDefinition) (uuid.UUID, error) ``` UpsertWorkflowDefinition stores def under a fresh ID and records it under the workflow ID's version list, returning the new storage ID. ### func \(\*Store\) [WakeFromExternal]() ```go func (s *Store) WakeFromExternal(_ context.Context, runID uuid.UUID) error ``` WakeFromExternal clears all await markers and wakeup\_at while leaving the run paused, so the Advance loop resumes it, or returns ErrNotFound. Generated by [gomarkdoc]() --- # postgres Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/store/postgres # postgres ```go import "github.com/Bugs5382/go-saga-orchestration/store/postgres" ``` Package postgres is the production Store implementation. It uses pgx for connection pooling and golang\-migrate to apply schema migrations. Both cmd/api and cmd/engine call postgres.Migrate\(dsn\) immediately after postgres.Open succeeds to apply any pending migrations at boot. ## func [AcquireAdvisoryLock]() ```go func AcquireAdvisoryLock(ctx context.Context, pool *pgxpool.Pool, lockID int64) (release func(), err error) ``` AcquireAdvisoryLock takes a session\-level Postgres advisory lock on a dedicated pooled connection and holds it until the returned release func is called. It is the single\-leader election primitive used to elect one engine replica to run the timer / cron dispatcher, so multi\-replica deployments do not fire duplicate wakeups. The call blocks until the lock is acquired \(pg\_advisory\_lock waits rather than failing when another session holds it\). release runs pg\_advisory\_unlock on a fresh short\-timeout context and then returns the connection to the pool; it is safe to call once, typically via defer on the goroutine that owns the leader role. ## func [Migrate]() ```go func Migrate(dsn string) error ``` Migrate applies every up\-migration in store/postgres/migrations that has not yet been recorded in the schema\_migrations table. Safe to call on every boot — no\-ops if the schema is already at Head. The migrations are embedded in the binary at build time so deploys do not need a sidecar migration job or an out\-of\-band run. cmd/api and cmd/engine each call this immediately after postgres.Open succeeds. DSN must use the pgx5 form, e.g. "postgres://user:pass@host:5432/db?sslmode=disable". ## type [Store]() Store wraps a pgxpool.Pool and implements store.Store. ```go type Store struct { // contains filtered or unexported fields } ``` ### func [Open]() ```go func Open(ctx context.Context, dsn string) (*Store, error) ``` Open dials Postgres using the supplied DSN. Caller MUST defer Close\(\). ### func \(\*Store\) [AppendEvent]() ```go func (s *Store) AppendEvent(ctx context.Context, evt domain.SagaRunEvent) error ``` AppendEvent inserts the audit event, ignoring duplicates on \(run\_id, step\_id, attempt, event\_type\). ### func \(\*Store\) [AppendSignal]() ```go func (s *Store) AppendSignal(ctx context.Context, sig domain.SagaSignal) error ``` AppendSignal inserts a received signal row for the run. ### func \(\*Store\) [Cancel]() ```go func (s *Store) Cancel(ctx context.Context, runID uuid.UUID, reason string) error ``` Cancel terminates an in\-flight run: terminal cancelled \+ terminal\_at, reason in last\_error, open user tasks closed, await markers / wakeup cleared — all in one transaction. Idempotent: when the run is already terminal the guard updates no row, so user tasks are untouched and no event is emitted. See issue \#80. ### func \(\*Store\) [ClaimCronFire]() ```go func (s *Store) ClaimCronFire(ctx context.Context, id uuid.UUID, expectedNextFire, newNextFire time.Time) (bool, error) ``` ClaimCronFire atomically advances next\_fire\_at from expectedNextFire to newNextFire and stamps last\_fired\_at=now\(\). Returns true iff this caller won the compare\-and\-swap \(false means another pod already claimed this fire\). ### func \(\*Store\) [ClearPause]() ```go func (s *Store) ClearPause(ctx context.Context, runID uuid.UUID) error ``` ClearPause returns the run to the running state and clears all wakeup and await markers. ### func \(\*Store\) [Close]() ```go func (s *Store) Close() ``` Close releases the pool. Safe to call once. ### func \(\*Store\) [CompleteAction]() ```go func (s *Store) CompleteAction(ctx context.Context, runID uuid.UUID, attempt int, result map[string]any) error ``` CompleteAction clears the await marker, merges result into variables, and sets wakeup\_at=now\(\) so the Advance paused\-handling loop resumes the saga. If attempt does not match current\_attempt the row is unchanged \(late delivery\). ### func \(\*Store\) [CountRuns]() ```go func (s *Store) CountRuns(ctx context.Context, filter store.RunFilter) (int, error) ``` CountRuns returns the total count of runs matching filter \(ignoring Limit/Offset\). ### func \(\*Store\) [CreateRun]() ```go func (s *Store) CreateRun(ctx context.Context, run domain.SagaRun) error ``` CreateRun inserts a new saga run row. ### func \(\*Store\) [CreateUserTask]() ```go func (s *Store) CreateUserTask(ctx context.Context, task domain.UserTask) error ``` CreateUserTask inserts a new user task row into runtime.saga\_user\_tasks. ### func \(\*Store\) [DeleteTrigger]() ```go func (s *Store) DeleteTrigger(ctx context.Context, id uuid.UUID) error ``` DeleteTrigger removes the trigger by id. Returns ErrNotFound if absent. ### func \(\*Store\) [FailAction]() ```go func (s *Store) FailAction(ctx context.Context, runID uuid.UUID, attempt int, code, message string, retryable bool) error ``` FailAction transitions the run to failed and appends an audit event. If attempt does not match current\_attempt the call is a no\-op \(late delivery\). ### func \(\*Store\) [FindRunsByAwaitedEvent]() ```go func (s *Store) FindRunsByAwaitedEvent(ctx context.Context, topic string) ([]domain.SagaRun, error) ``` FindRunsByAwaitedEvent returns paused runs awaiting an event on topic. ### func \(\*Store\) [FindRunsByDueWakeup]() ```go func (s *Store) FindRunsByDueWakeup(ctx context.Context, now time.Time, limit int) ([]uuid.UUID, error) ``` FindRunsByDueWakeup returns up to limit IDs of paused runs whose wakeup\_at is at or before now, ordered by wakeup\_at. ### func \(\*Store\) [GetAction]() ```go func (s *Store) GetAction(ctx context.Context, service, name string, version int) (domain.ActionRegistration, error) ``` GetAction returns the action registration for the given service/name/version. ### func \(\*Store\) [GetEventByID]() ```go func (s *Store) GetEventByID(ctx context.Context, id uuid.UUID) (domain.SagaRunEvent, error) ``` GetEventByID returns a single audit event by its UUID, or ErrNotFound. ### func \(\*Store\) [GetPublishedRuleByID]() ```go func (s *Store) GetPublishedRuleByID(ctx context.Context, ruleID string, tenantID *uuid.UUID) (domain.RuleDefinition, error) ``` GetPublishedRuleByID returns the most recent published version of a rule. ### func \(\*Store\) [GetPublishedWorkflowByID]() ```go func (s *Store) GetPublishedWorkflowByID(ctx context.Context, workflowID string, tenantID *uuid.UUID) (domain.WorkflowDefinition, error) ``` GetPublishedWorkflowByID returns the highest\-version published definition for workflowID scoped to tenantID \(nil = platform\), or ErrNotFound. ### func \(\*Store\) [GetRun]() ```go func (s *Store) GetRun(ctx context.Context, id uuid.UUID) (domain.SagaRun, error) ``` GetRun loads the run with the given ID, or returns ErrNotFound. ### func \(\*Store\) [GetTrigger]() ```go func (s *Store) GetTrigger(ctx context.Context, id uuid.UUID) (domain.SagaTrigger, error) ``` GetTrigger returns the SagaTrigger for id, or ErrNotFound. ### func \(\*Store\) [GetUserTask]() ```go func (s *Store) GetUserTask(ctx context.Context, taskID uuid.UUID) (domain.UserTask, error) ``` GetUserTask returns the user task by ID, or ErrNotFound. ### func \(\*Store\) [GetWorkflowDefinition]() ```go func (s *Store) GetWorkflowDefinition(ctx context.Context, id uuid.UUID) (domain.WorkflowDefinition, error) ``` GetWorkflowDefinition returns the definition with the given storage ID, or ErrNotFound. ### func \(\*Store\) [ListActions]() ```go func (s *Store) ListActions(ctx context.Context, filter store.ActionFilter) ([]domain.ActionRegistration, error) ``` ListActions returns all action registrations matching the optional filter. ### func \(\*Store\) [ListChildrenByParent]() ```go func (s *Store) ListChildrenByParent(ctx context.Context, parentID uuid.UUID, parentStepID string) ([]domain.SagaRun, error) ``` ListChildrenByParent returns all runs with parent\_run\_id=$1 AND parent\_step\_id=$2. ### func \(\*Store\) [ListDueCronTriggers]() ```go func (s *Store) ListDueCronTriggers(ctx context.Context, now time.Time, limit int) ([]domain.SagaTrigger, error) ``` ListDueCronTriggers returns enabled cron triggers whose next\_fire\_at is at or before now, ordered oldest\-first, capped at limit. ### func \(\*Store\) [ListEventsByRun]() ```go func (s *Store) ListEventsByRun(ctx context.Context, runID uuid.UUID) ([]domain.SagaRunEvent, error) ``` ListEventsByRun returns the events recorded for runID ordered by recorded\_at. ### func \(\*Store\) [ListRuns]() ```go func (s *Store) ListRuns(ctx context.Context, filter store.RunFilter) ([]domain.SagaRun, error) ``` ListRuns returns saga runs matching filter, sorted by started\_at DESC. For TriggerType filtering a LEFT JOIN on saga\_triggers is added. Limit defaults to 50 when 0; hard max of 500 is enforced by the handler before this is called. ### func \(\*Store\) [ListTriggers]() ```go func (s *Store) ListTriggers(ctx context.Context, filter store.TriggerFilter) ([]domain.SagaTrigger, error) ``` ListTriggers returns triggers matching the optional filter. ### func \(\*Store\) [ListUserTasksByRun]() ```go func (s *Store) ListUserTasksByRun(ctx context.Context, runID uuid.UUID) ([]domain.UserTask, error) ``` ListUserTasksByRun returns all user tasks for a given run, ordered by creation time \(id order is used as a stable proxy since saga\_user\_tasks has no separate created\_at column beyond the implicit id ordering\). ### func \(\*Store\) [MarkAwaitingAction]() ```go func (s *Store) MarkAwaitingAction(ctx context.Context, runID uuid.UUID, dispatch string, attempt int) error ``` MarkAwaitingAction sets state=paused, awaited\_action\_dispatch, and current\_attempt for the given run. Idempotent on \(runID, attempt, dispatch\). ### func \(\*Store\) [MarkRunFailed]() ```go func (s *Store) MarkRunFailed(ctx context.Context, runID uuid.UUID, currentStep, lastError string) error ``` MarkRunFailed transitions a run to terminal failed, stamps terminal\_at, and persists lastError on the run. Idempotent: the terminal guard means an already\-terminal run is left untouched. See issue \#80. ### func \(\*Store\) [Pool]() ```go func (s *Store) Pool() *pgxpool.Pool ``` Pool exposes the underlying pgxpool for migrations \+ tests. ### func \(\*Store\) [PopTryCatch]() ```go func (s *Store) PopTryCatch(ctx context.Context, runID uuid.UUID) (domain.TryCatchFrame, bool, error) ``` PopTryCatch removes and returns the top TryCatchFrame atomically. Returns \(zero, false, nil\) if the stack is empty. ### func \(\*Store\) [PushTryCatch]() ```go func (s *Store) PushTryCatch(ctx context.Context, runID uuid.UUID, frame domain.TryCatchFrame) error ``` PushTryCatch reads the try\_catch\_stack JSONB, appends frame \(enforcing max depth 3\), and writes it back within a transaction. ### func \(\*Store\) [RecordTriggerFire]() ```go func (s *Store) RecordTriggerFire(ctx context.Context, triggerID uuid.UUID, workflowID string, runID *uuid.UUID, fireErr string) error ``` RecordTriggerFire inserts a row into runtime.saga\_trigger\_fires. Best\-effort: callers log errors but must not abort the triggering run on failure. ### func \(\*Store\) [SetPausedAwaitingEvent]() ```go func (s *Store) SetPausedAwaitingEvent(ctx context.Context, runID uuid.UUID, topic string, headers map[string]string) error ``` SetPausedAwaitingEvent marks the run paused awaiting an event on topic with the given header filter. ### func \(\*Store\) [SetPausedAwaitingEventWithDeadline]() ```go func (s *Store) SetPausedAwaitingEventWithDeadline(ctx context.Context, runID uuid.UUID, topic string, headers map[string]string, deadline *time.Time) error ``` SetPausedAwaitingEventWithDeadline is SetPausedAwaitingEvent plus an optional wakeup deadline \(nil = wait indefinitely\). The deadline is stored as wakeup\_at so the timer dispatcher wakes the run if no matching event arrives in time. ### func \(\*Store\) [SetPausedAwaitingSignal]() ```go func (s *Store) SetPausedAwaitingSignal(ctx context.Context, runID uuid.UUID, signalName string, deadline *time.Time) error ``` SetPausedAwaitingSignal marks the run paused awaiting signalName, with an optional wakeup deadline. ### func \(\*Store\) [SetPausedWithWakeup]() ```go func (s *Store) SetPausedWithWakeup(ctx context.Context, runID uuid.UUID, wakeupAt time.Time) error ``` SetPausedWithWakeup marks the run paused and records wakeup\_at. ### func \(\*Store\) [SpawnChildRun]() ```go func (s *Store) SpawnChildRun(ctx context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any) (uuid.UUID, error) ``` SpawnChildRun looks up the definition\_id for def, constructs a child SagaRun with the parent fields set, and inserts it via CreateRun. The child begins at the definition's default Start step. ### func \(\*Store\) [SpawnChildRunAt]() ```go func (s *Store) SpawnChildRunAt(ctx context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any, startStep string) (uuid.UUID, error) ``` SpawnChildRunAt looks up the definition\_id for def, constructs a child SagaRun with the parent fields set, overrides CurrentStep with startStep when non\-empty, and inserts it via CreateRun. ### func \(\*Store\) [StatsForWorkflow]() ```go func (s *Store) StatsForWorkflow(ctx context.Context, workflowID string) (store.WorkflowStats, error) ``` StatsForWorkflow computes aggregate metrics for a single workflow using two aggregate queries: one for success\_rate\_24h \+ last\_run\_at, one for in\_flight. ### func \(\*Store\) [SubmitUserTask]() ```go func (s *Store) SubmitUserTask(ctx context.Context, taskID uuid.UUID, submittedBy string, result map[string]any) error ``` SubmitUserTask marks the task submitted with the given actor and result. Idempotent: re\-submitting overwrites the previous submission fields. Returns ErrNotFound if the task does not exist. ### func \(\*Store\) [TryConsumeAwaitedSignal]() ```go func (s *Store) TryConsumeAwaitedSignal(ctx context.Context, runID uuid.UUID, signalName string) (bool, error) ``` TryConsumeAwaitedSignal atomically clears the await markers and marks matching unconsumed signal rows consumed when the run is paused awaiting signalName, reporting whether it did so. ### func \(\*Store\) [UpdateRunState]() ```go func (s *Store) UpdateRunState(ctx context.Context, id uuid.UUID, state domain.RunState, currentStep string) error ``` UpdateRunState sets the run's state and current step, stamping terminal\_at when the new state is terminal. ### func \(\*Store\) [UpdateRunVariables]() ```go func (s *Store) UpdateRunVariables(ctx context.Context, runID uuid.UUID, merge map[string]any) error ``` UpdateRunVariables merges merge into saga\_runs.variables using jsonb\_set per top\-level key. Dotted\-key writes are flattened into jsonb\_set path expressions; nested merges go through the JSONB operator '||' for shallow object combine, then jsonb\_set for the dotted writes. ### func \(\*Store\) [UpsertActionRegistration]() ```go func (s *Store) UpsertActionRegistration(ctx context.Context, reg domain.ActionRegistration) error ``` UpsertActionRegistration inserts or updates an action\_registry row keyed by \(service, action\_name, version\). ### func \(\*Store\) [UpsertRuleDefinition]() ```go func (s *Store) UpsertRuleDefinition(ctx context.Context, def domain.RuleDefinition) (uuid.UUID, error) ``` UpsertRuleDefinition inserts \(or upserts on \(rule\_id, version\)\) a rule. ### func \(\*Store\) [UpsertTrigger]() ```go func (s *Store) UpsertTrigger(ctx context.Context, trigger domain.SagaTrigger) (uuid.UUID, error) ``` UpsertTrigger inserts or replaces a row in runtime.saga\_triggers keyed by id. If trigger.ID == uuid.Nil a new ID is generated. If the ID is set and a row already exists, all mutable columns are replaced \(upsert on conflict\). ### func \(\*Store\) [UpsertWorkflowDefinition]() ```go func (s *Store) UpsertWorkflowDefinition(ctx context.Context, def domain.WorkflowDefinition) (uuid.UUID, error) ``` UpsertWorkflowDefinition inserts def, or updates the existing row on a \(workflow\_id, version\) conflict, returning the row's storage ID. ### func \(\*Store\) [WakeFromExternal]() ```go func (s *Store) WakeFromExternal(ctx context.Context, runID uuid.UUID) error ``` WakeFromExternal clears all await markers and wakeup\_at while leaving the run paused, so the Advance loop resumes it. Generated by [gomarkdoc]() --- # redis Source: https://bugs5382.github.io/go-saga-orchestration/docs/reference/store/redis # redis ```go import "github.com/Bugs5382/go-saga-orchestration/store/redis" ``` Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. Package redis is a Redis/Valkey\-backed store.Store implementation. ## type [Option]() Option configures a Store. ```go type Option func(*Store) ``` ### func [WithPrefix]() ```go func WithPrefix(p string) Option ``` WithPrefix overrides the default "saga:" key prefix. ### func [WithRunTTL]() ```go func WithRunTTL(d time.Duration) Option ``` WithRunTTL sets the terminal\-run expiry \(0 disables\). ## type [Store]() Store is a Redis/Valkey\-backed store.Store. Safe for concurrent use. ```go type Store struct { // contains filtered or unexported fields } ``` ### func [Open]() ```go func Open(ctx context.Context, url string, opts ...Option) (*Store, error) ``` Open dials url \(redis:// or rediss://; Redis or Valkey\) and verifies it. ### func \(\*Store\) [AppendEvent]() ```go func (s *Store) AppendEvent(ctx context.Context, evt domain.SagaRunEvent) error ``` AppendEvent appends evt to the per\-run event list and stores it under its own key for direct lookup by ID. ### func \(\*Store\) [AppendSignal]() ```go func (s *Store) AppendSignal(ctx context.Context, sig domain.SagaSignal) error ``` AppendSignal appends sig to the signals:\{runID\} list. ### func \(\*Store\) [Cancel]() ```go func (s *Store) Cancel(ctx context.Context, runID uuid.UUID, reason string) error ``` Cancel terminates an in\-flight run: terminal cancelled \+ terminal\_at, reason in last\_error, open user tasks closed, await markers / wakeup cleared \(and the run dropped from the active indexes\). Idempotent — a no\-op \(and no event\) once the run is terminal. See issue \#80. ### func \(\*Store\) [ClaimCronFire]() ```go func (s *Store) ClaimCronFire(ctx context.Context, id uuid.UUID, expectedNextFire, newNextFire time.Time) (bool, error) ``` ClaimCronFire atomically advances next\_fire\_at from expectedNextFire to newNextFire and stamps last\_fired\_at=now\(\) using a Redis WATCH/MULTI transaction. Returns true iff this caller won the compare\-and\-swap. ### func \(\*Store\) [ClearPause]() ```go func (s *Store) ClearPause(ctx context.Context, runID uuid.UUID) error ``` ClearPause transitions the run to running, clears all await markers and wakeup\_at, and removes the run from idx:wakeup and idx:awaitevent:\{topic\}. ### func \(\*Store\) [Close]() ```go func (s *Store) Close() error ``` Close releases the client. ### func \(\*Store\) [CompleteAction]() ```go func (s *Store) CompleteAction(ctx context.Context, runID uuid.UUID, attempt int, result map[string]any) error ``` CompleteAction clears the dispatch marker, sets WakeupAt=now, and merges result into Variables. No\-op when attempt \!= CurrentAttempt \(late delivery\). ### func \(\*Store\) [CountRuns]() ```go func (s *Store) CountRuns(ctx context.Context, filter store.RunFilter) (int, error) ``` CountRuns returns the total count matching filter \(ignoring Limit/Offset\). ### func \(\*Store\) [CreateRun]() ```go func (s *Store) CreateRun(ctx context.Context, run domain.SagaRun) error ``` CreateRun stores the run blob and adds it to the two index structures. ### func \(\*Store\) [CreateUserTask]() ```go func (s *Store) CreateUserTask(ctx context.Context, task domain.UserTask) error ``` CreateUserTask stores a new UserTask and adds it to the per\-run index. ### func \(\*Store\) [DeleteTrigger]() ```go func (s *Store) DeleteTrigger(ctx context.Context, id uuid.UUID) error ``` DeleteTrigger removes the trigger. Returns ErrNotFound if absent. ### func \(\*Store\) [FailAction]() ```go func (s *Store) FailAction(ctx context.Context, runID uuid.UUID, attempt int, code, message string, retryable bool) error ``` FailAction transitions the run to failed and appends an audit event AFTER the tx. No\-op when attempt \!= CurrentAttempt \(late delivery\). ### func \(\*Store\) [FindRunsByAwaitedEvent]() ```go func (s *Store) FindRunsByAwaitedEvent(ctx context.Context, topic string) ([]domain.SagaRun, error) ``` FindRunsByAwaitedEvent returns all paused runs awaiting an event on topic by loading members of idx:awaitevent:\{topic\} and returning those whose run blob is still paused and awaiting that topic \(defensive filter\). ### func \(\*Store\) [FindRunsByDueWakeup]() ```go func (s *Store) FindRunsByDueWakeup(ctx context.Context, now time.Time, limit int) ([]uuid.UUID, error) ``` FindRunsByDueWakeup returns up to limit run IDs of paused runs whose wakeup\_at is at or before now. It queries idx:wakeup via ZRANGEBYSCORE then loads the candidate run blobs and retains only those with State==RunStatePaused, matching the memory\-store oracle \(store/memory/store.go\). The limit bounds the result count, not the ZSET scan range. ### func \(\*Store\) [GetAction]() ```go func (s *Store) GetAction(ctx context.Context, service, name string, version int) (domain.ActionRegistration, error) ``` GetAction returns the registration for service\+name\+version, or ErrNotFound. ### func \(\*Store\) [GetEventByID]() ```go func (s *Store) GetEventByID(ctx context.Context, id uuid.UUID) (domain.SagaRunEvent, error) ``` GetEventByID returns the event with the given ID, or ErrNotFound. ### func \(\*Store\) [GetPublishedRuleByID]() ```go func (s *Store) GetPublishedRuleByID(ctx context.Context, ruleID string, _ *uuid.UUID) (domain.RuleDefinition, error) ``` GetPublishedRuleByID returns the newest published version of ruleID, falling back to the most recent version when none is published. Returns ErrNotFound when ruleID has never been upserted. ### func \(\*Store\) [GetPublishedWorkflowByID]() ```go func (s *Store) GetPublishedWorkflowByID(ctx context.Context, workflowID string, _ *uuid.UUID) (domain.WorkflowDefinition, error) ``` GetPublishedWorkflowByID returns the newest published version of workflowID, falling back to the most recent version when none is published. Returns ErrNotFound when the workflow ID has never been upserted. ### func \(\*Store\) [GetRun]() ```go func (s *Store) GetRun(ctx context.Context, id uuid.UUID) (domain.SagaRun, error) ``` GetRun returns the run with the given ID, or ErrNotFound. ### func \(\*Store\) [GetTrigger]() ```go func (s *Store) GetTrigger(ctx context.Context, id uuid.UUID) (domain.SagaTrigger, error) ``` GetTrigger returns the SagaTrigger for id, or ErrNotFound. ### func \(\*Store\) [GetUserTask]() ```go func (s *Store) GetUserTask(ctx context.Context, taskID uuid.UUID) (domain.UserTask, error) ``` GetUserTask returns the UserTask or ErrNotFound. ### func \(\*Store\) [GetWorkflowDefinition]() ```go func (s *Store) GetWorkflowDefinition(ctx context.Context, id uuid.UUID) (domain.WorkflowDefinition, error) ``` GetWorkflowDefinition returns the definition stored at the given storage ID, or ErrNotFound. ### func \(\*Store\) [ListActions]() ```go func (s *Store) ListActions(ctx context.Context, filter store.ActionFilter) ([]domain.ActionRegistration, error) ``` ListActions returns all registrations matching the optional filter fields. ### func \(\*Store\) [ListChildrenByParent]() ```go func (s *Store) ListChildrenByParent(ctx context.Context, parentID uuid.UUID, parentStepID string) ([]domain.SagaRun, error) ``` ListChildrenByParent returns all child runs for parentID/parentStepID. ### func \(\*Store\) [ListDueCronTriggers]() ```go func (s *Store) ListDueCronTriggers(ctx context.Context, now time.Time, limit int) ([]domain.SagaTrigger, error) ``` ListDueCronTriggers returns enabled cron triggers whose next\_fire\_at is at or before now, sorted oldest\-first, capped at limit. ### func \(\*Store\) [ListEventsByRun]() ```go func (s *Store) ListEventsByRun(ctx context.Context, runID uuid.UUID) ([]domain.SagaRunEvent, error) ``` ListEventsByRun returns all events for runID in append order. An empty slice \(not an error\) is returned when no events have been appended. ### func \(\*Store\) [ListRuns]() ```go func (s *Store) ListRuns(ctx context.Context, filter store.RunFilter) ([]domain.SagaRun, error) ``` ListRuns returns saga runs matching filter, sorted StartedAt DESC, paginated. ### func \(\*Store\) [ListTriggers]() ```go func (s *Store) ListTriggers(ctx context.Context, filter store.TriggerFilter) ([]domain.SagaTrigger, error) ``` ListTriggers returns triggers matching the optional filter fields. ### func \(\*Store\) [ListUserTasksByRun]() ```go func (s *Store) ListUserTasksByRun(ctx context.Context, runID uuid.UUID) ([]domain.UserTask, error) ``` ListUserTasksByRun returns all user tasks for runID sorted by ID bytes. This matches the memory store's uuidLess ordering. ### func \(\*Store\) [MarkAwaitingAction]() ```go func (s *Store) MarkAwaitingAction(ctx context.Context, runID uuid.UUID, dispatch string, attempt int) error ``` MarkAwaitingAction sets state=paused and records the dispatch key \+ attempt. Idempotent on \(attempt, dispatch\): same pair returns without writing. ### func \(\*Store\) [MarkRunFailed]() ```go func (s *Store) MarkRunFailed(ctx context.Context, runID uuid.UUID, currentStep, lastError string) error ``` MarkRunFailed transitions a run to terminal failed, stamps terminal\_at, and persists lastError on the run. Idempotent on terminal runs. See issue \#80. ### func \(\*Store\) [PopTryCatch]() ```go func (s *Store) PopTryCatch(ctx context.Context, runID uuid.UUID) (domain.TryCatchFrame, bool, error) ``` PopTryCatch removes and returns the top TryCatchFrame. Returns \(zero, false, nil\) when the stack is empty. Uses a closure variable to surface the popped value. ### func \(\*Store\) [PushTryCatch]() ```go func (s *Store) PushTryCatch(ctx context.Context, runID uuid.UUID, frame domain.TryCatchFrame) error ``` PushTryCatch appends frame to the run's TryCatchStack. Returns an error if the stack is already at maximum depth \(3\). ### func \(\*Store\) [RecordTriggerFire]() ```go func (s *Store) RecordTriggerFire(_ context.Context, _ uuid.UUID, _ string, _ *uuid.UUID, _ string) error ``` RecordTriggerFire is a no\-op: Redis has no persistent audit table. ### func \(\*Store\) [SetPausedAwaitingEvent]() ```go func (s *Store) SetPausedAwaitingEvent(ctx context.Context, runID uuid.UUID, topic string, headers map[string]string) error ``` SetPausedAwaitingEvent marks the run paused awaiting an event on topic with the given header filter. The run is added to idx:awaitevent:\{topic\}. ### func \(\*Store\) [SetPausedAwaitingEventWithDeadline]() ```go func (s *Store) SetPausedAwaitingEventWithDeadline(ctx context.Context, runID uuid.UUID, topic string, headers map[string]string, deadline *time.Time) error ``` SetPausedAwaitingEventWithDeadline is SetPausedAwaitingEvent plus an optional wakeup deadline which is also recorded in idx:wakeup. ### func \(\*Store\) [SetPausedAwaitingSignal]() ```go func (s *Store) SetPausedAwaitingSignal(ctx context.Context, runID uuid.UUID, signalName string, deadline *time.Time) error ``` SetPausedAwaitingSignal marks the run paused awaiting signalName, with an optional deadline added to idx:wakeup. ### func \(\*Store\) [SetPausedWithWakeup]() ```go func (s *Store) SetPausedWithWakeup(ctx context.Context, runID uuid.UUID, wakeupAt time.Time) error ``` SetPausedWithWakeup marks the run paused with a wakeup time and registers it in the idx:wakeup ZSET. ### func \(\*Store\) [SpawnChildRun]() ```go func (s *Store) SpawnChildRun(ctx context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any) (uuid.UUID, error) ``` SpawnChildRun creates a child run beginning at the child definition's default start step. ### func \(\*Store\) [SpawnChildRunAt]() ```go func (s *Store) SpawnChildRunAt(ctx context.Context, parentID uuid.UUID, parentStepID, branchKey string, def domain.WorkflowDefinition, inputs map[string]any, startStep string) (uuid.UUID, error) ``` SpawnChildRunAt creates a child run linked to parentID/parentStepID/branchKey, beginning at startStep \(empty string means the child definition's default start\). The whole operation — def upsert and run/index writes — is performed inside a single WATCH/MULTI/EXEC optimistic\-transaction retry loop so it is atomic. ### func \(\*Store\) [StatsForWorkflow]() ```go func (s *Store) StatsForWorkflow(ctx context.Context, workflowID string) (store.WorkflowStats, error) ``` StatsForWorkflow computes aggregate metrics for workflowID. ### func \(\*Store\) [SubmitUserTask]() ```go func (s *Store) SubmitUserTask(ctx context.Context, taskID uuid.UUID, submittedBy string, result map[string]any) error ``` SubmitUserTask marks the task submitted. Returns ErrNotFound if it does not exist. ### func \(\*Store\) [TryConsumeAwaitedSignal]() ```go func (s *Store) TryConsumeAwaitedSignal(ctx context.Context, runID uuid.UUID, signalName string) (bool, error) ``` TryConsumeAwaitedSignal attempts to consume the awaited signal on the run. Returns \(false, nil\) when the run is missing, not paused, or the awaited signal name does not match. On match it clears all await markers and wakeup\_at, marks the first unconsumed matching signal as consumed, and returns \(true, nil\). ### func \(\*Store\) [UpdateRunState]() ```go func (s *Store) UpdateRunState(ctx context.Context, id uuid.UUID, state domain.RunState, currentStep string) error ``` UpdateRunState sets the run's state and current step via a WATCH/MULTI transaction. When the new state is terminal, the run is removed from the active idx:wakeup and idx:awaitevent indexes. ### func \(\*Store\) [UpdateRunVariables]() ```go func (s *Store) UpdateRunVariables(ctx context.Context, id uuid.UUID, merge map[string]any) error ``` UpdateRunVariables merges entries of merge into the run's Variables using dotted\-key path semantics, via a WATCH/MULTI transaction. ### func \(\*Store\) [UpsertActionRegistration]() ```go func (s *Store) UpsertActionRegistration(ctx context.Context, reg domain.ActionRegistration) error ``` UpsertActionRegistration stores or replaces a registration and adds it to idx:actions. ### func \(\*Store\) [UpsertRuleDefinition]() ```go func (s *Store) UpsertRuleDefinition(ctx context.Context, def domain.RuleDefinition) (uuid.UUID, error) ``` UpsertRuleDefinition stores def atomically using a WATCH/MULTI optimistic transaction retry loop. If def.ID is the zero UUID a new one is generated. When re\-upserting an existing storage ID the old version\-list entry is removed inside the same transaction to prevent duplicates under concurrency. ### func \(\*Store\) [UpsertTrigger]() ```go func (s *Store) UpsertTrigger(ctx context.Context, trigger domain.SagaTrigger) (uuid.UUID, error) ``` UpsertTrigger inserts or replaces a SagaTrigger. Generates a new ID when trigger.ID == uuid.Nil. Defaults CreatedAt to now when zero. ### func \(\*Store\) [UpsertWorkflowDefinition]() ```go func (s *Store) UpsertWorkflowDefinition(ctx context.Context, def domain.WorkflowDefinition) (uuid.UUID, error) ``` UpsertWorkflowDefinition stores def under a fresh storage ID and appends that ID to the def:byname:\{workflowID\} list \(oldest→newest order\). ### func \(\*Store\) [WakeFromExternal]() ```go func (s *Store) WakeFromExternal(ctx context.Context, runID uuid.UUID) error ``` WakeFromExternal clears all await markers and wakeup\_at while leaving the run state as paused, then removes it from idx:wakeup / idx:awaitevent. Generated by [gomarkdoc]() --- # Store Backends Source: https://bugs5382.github.io/go-saga-orchestration/docs/stores # Store Backends go-saga-orchestration persists saga state to a pluggable `store.Store`. The backend is selected at startup via the `STORE_TYPE` environment variable and cannot be changed at runtime. --- ## Backend selection Set `STORE_TYPE` to one of the values below. If the variable is absent the engine defaults to `postgres`. | `STORE_TYPE` | Backend | Notes | |---|---|---| | `postgres` _(default)_ | PostgreSQL | Fully durable; LISTEN/NOTIFY powers the live-stream endpoint. Requires `DATABASE_DSN`. | | `redis` | Redis / Valkey | Durable within Redis AOF/RDB persistence settings. `redis` and `valkey` are wire-compatible aliases — the same code path handles both. Requires `REDIS_URL`. | | `valkey` | Valkey / Redis | Alias for `redis` above. | | `memory` | In-process map | No persistence; all state is lost on restart. For tests and local development only. | ### Environment variables | Variable | Backends | Default | Purpose | |---|---|---|---| | `DATABASE_DSN` | `postgres` | _(empty)_ | Postgres connection string (`postgres://user:pass@host/db`). | | `REDIS_URL` | `redis`, `valkey` | _(empty)_ | Redis/Valkey connection URL (`redis://host:6379/0` or `rediss://` for TLS). Required when the store type is `redis` or `valkey`. | | `REDIS_RUN_TTL` | `redis`, `valkey` | `0s` (disabled) | Go duration string (e.g. `168h`, `72h`). When non-zero, the engine calls `EXPIRE` on all keys belonging to a saga run once it reaches a terminal state (succeeded, failed, or cancelled). Keys affected: run blob, event list, signals list, user-task index. Default `0s` keeps terminal runs forever. | --- ## Redis / Valkey durability Redis and Valkey use the same RESP protocol and no proprietary modules. The go-saga-orchestration redis backend is pure-RESP; either server can be used interchangeably behind `REDIS_URL`. ### Persistence modes and data-loss windows Redis and Valkey support three common persistence configurations: | Mode | How it works | Worst-case data-loss window | |---|---|---| | **AOF `appendfsync everysec`** (default) | Writes are flushed to the AOF file once per second. | Up to ~1 second of committed writes. | | **AOF `appendfsync always`** | Every write command is fsynced before the client gets a reply. | Near-zero (bounded by disk flush latency). | | **RDB snapshots only** (no AOF) | Point-in-time snapshots at a configured interval. | Up to the snapshot interval (often minutes). | For production use with saga state, AOF `everysec` is the standard trade-off between throughput and durability. Use `always` if the ~1 s loss window is unacceptable for your use case. RDB-only is not recommended for saga workloads where every state transition matters. ### Memory-bound storage Redis and Valkey keep all data in RAM (with optional disk persistence). The size of the saga keyspace grows with the number of runs, events, signals, and user tasks stored. Monitor memory usage and configure `maxmemory` and an eviction policy appropriate for your workload. Note that `noeviction` will cause write errors rather than silent data loss if memory is exhausted; `allkeys-lru` will silently discard saga keys. ### Run retention with `REDIS_RUN_TTL` By default, terminal-run keys are kept forever. Set `REDIS_RUN_TTL` to a Go duration (e.g. `168h` for one week) to have the engine auto-expire all keys for a run once it finishes. This bounds keyspace growth at the cost of losing historical run data after the TTL expires. Note that `REDIS_RUN_TTL` expires the run blob, event list, signals list, and user-task index keys, but the run-membership indexes (`idx:runs`, `idx:runs:byworkflow`, `idx:children`) are not expired and are pruned only by completion-path cleanup; with TTL enabled those index sets retain dangling member IDs over time (harmless: reads use MGET and skip missing keys) but will grow without bound if runs are not manually cleaned up. --- ## Limitation: live saga-stream requires postgres The `GET /api/v1/sagas/{run_id}/stream` endpoint tails audit events in real time using PostgreSQL `LISTEN/NOTIFY`. Under the `redis`, `valkey`, or `memory` backends, the handler returns **HTTP 501 Not Implemented** because the required Postgres connection pool is not available. All other API endpoints work with any backend. --- # Verb Reference Source: https://bugs5382.github.io/go-saga-orchestration/docs/verbs # 📖 Verb Reference This page documents all 31 saga step types ("verbs") supported by the engine. ## Quick-reference table | Verb | License group | One-liner | |---|---|---| | `action` | common | Dispatch a step to an external worker and pause until it replies. | | `decision` | common | Evaluate a stored rule table and branch on its output. | | `switch` | common | Evaluate a CEL expression and branch on the string result. | | `error` | common | Immediately fail the saga with a code and message. | | `noop` | common | Placeholder — do nothing and advance. | | `end` | _(terminal)_ | End the run as `succeeded`. | | `set_var` | common | Write a literal or CEL-computed value into a variable. | | `transform` | common | Evaluate a CEL expression and store the result. | | `merge` | common | Merge a CEL-evaluated map into an existing variable. | | `filter` | common | Keep list elements that satisfy a CEL predicate. | | `map` | common | Transform each element of a list with a CEL expression. | | `assert` | common | Fail the saga if a CEL expression is not truthy. | | `log` | common | Emit a structured log line at a chosen level. | | `metric_emit` | observability | Append a named metric event to the run's audit stream. | | `http_request` | common / external_io_advanced¹ | Issue a synchronous outbound HTTP request. | | `webhook_emit` | external_io_advanced | POST a JSON payload to an external URL. | | `wait_duration` | waits | Pause for a Go duration (e.g. `"5s"`, `"1h30m"`). | | `wait_until` | waits | Pause until an RFC3339 absolute timestamp. | | `wait_for_signal` | events_and_signals | Pause until a named external signal arrives (with optional timeout). | | `wait_for_event` | events_and_signals | Pause until a matching event topic + header subset arrives. | | `emit_signal` | events_and_signals | Send a signal to another (or the same) run. | | `emit_event` | events_and_signals | Publish an event via the configured EventEmitter. | | `while` | loops_and_recovery | Loop while a CEL condition holds; exit via branching. | | `try_catch` | loops_and_recovery | Push an error-handler frame; jump to `catch` on any step error. | | `cancel` | loops_and_recovery | Cancel a run (self or a target). | | `parallel` | parallel_control | Fan out N branches and join when a strategy is satisfied. | | `join` | parallel_control | Barrier: reconvene independently-spawned upstream streams before continuing. | | `foreach` | parallel_control | Fan out one branch per element of a CEL-evaluated list. | | `sub_saga` | compositions | Start a child workflow and pause the parent until it finishes. | | `spawn_saga` | compositions | Fire-and-forget: start a child workflow and continue immediately. | | `manual_approval` | human_interaction | Create a user task; pause until an assignee submits it. | | `collect_input` | human_interaction | Like `manual_approval` but `form_schema` is required. | > ¹ `http_request` is `common` when method is `GET` with no `secret_ref`; all other configurations require the `external_io_advanced` group. --- > 💡 **License groups** gate verbs at publish time and at runtime. In development the `StubAllowAll` licensing resolver (used by `saga.InMemory()` and `saga.New` with no `Licensing` set) permits every group. In production, override per-request with the `X-Feature-Override` header or the `FeatureOverrides` field on the run start request. --- ## Core / Control ### `action` Dispatches the step to a named worker process and **pauses the saga** until the worker replies. | Input | Required | Notes | |---|---|---| | `step.Action` | ✅ | `"service.name"` — must contain a dot. Identifies the worker queue. | | `step.Inputs` | optional | Forwarded verbatim to the worker as `inputs`. | **Output:** The worker's result map is merged directly into `Variables`. There is no `out_var` — whatever keys the worker returns become top-level variables. > ⚠️ In embedded mode (`saga.InMemory()`) the action verb publishes to an in-process publisher. You still need a worker goroutine (or service mode) to actually handle the dispatch; a plain in-memory saga with no registered worker handler will leave the run paused indefinitely. **Example:** [`examples/workflows/action.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/action.json) --- ### `decision` Evaluates a stored decision-table rule and returns its output map. The engine reads `result["branch"]` to pick a route from `step.Branches`. | Input | Required | Notes | |---|---|---| | `rule_id` | ✅ | Stable ID of a published `RuleDefinition`. | | `inputs_map` | optional | `map[string]string` — maps rule input keys to variable names. Omit to pass `Variables` directly. | **Output:** The rule's full output map (including `branch`) is merged into `Variables`. **Example:** [`examples/workflows/decision.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/decision.json) --- ### `switch` Evaluates a CEL expression to a string and routes via `step.Branches`. Simpler than `decision` when no rule table is needed. | Input | Required | Notes | |---|---|---| | `expr` | ✅ | CEL expression over `Variables`; must produce a string. | **Output:** `{"branch": ""}` — the engine picks `step.Branches[].Next`. An unknown branch value is a runtime error. **Example:** [`examples/workflows/switch.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/switch.json) --- ### `error` Immediately fails the saga as a non-retryable error. | Input | Required | Notes | |---|---|---| | `code` | ✅ | Error code string surfaced in the run's error record. | | `message` | optional | Human-readable description. | **Output:** None — the run terminates as `failed`. **Example:** [`examples/workflows/error.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/error.json) --- ### `noop` Does nothing. Advances to `step.Next`. Useful as a placeholder during development or as a join point for multiple branches. **Inputs:** none. **Output:** empty map. **Example:** [`examples/workflows/noop.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/noop.json) --- ### `end` Marks the run as `succeeded`. Every workflow needs at least one `end` step. It is dispatched inline by the engine (not via a queue) so no license group applies. **Inputs:** none. **Output:** none — the saga terminates. --- ## Data ### `set_var` Writes a value to a variable. Use it to seed variables before CEL verbs read them. | Input | Required | Notes | |---|---|---| | `out_var` | ✅ | Destination variable name. Dotted keys (`a.b.c`) write into nested maps. | | `value` | one of `value`/`expr` | Literal value — passed through unchanged. | | `expr` | one of `value`/`expr` | CEL expression over `Variables`; result is written. `expr` wins when both are present. | **Output:** `{out_var: }`. **Example:** [`examples/workflows/set_var.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/set_var.json) --- ### `transform` Evaluates a CEL expression and writes the result to a named variable. Equivalent to `set_var` with `expr`. | Input | Required | Notes | |---|---|---| | `expr` | ✅ | CEL expression over `Variables`. | | `out_var` | ✅ | Destination variable name. | **Output:** `{out_var: }`. **Example:** [`examples/workflows/transform.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/transform.json) --- ### `merge` Evaluates a CEL expression that must produce a map, then **deep-merges** it into an existing variable. | Input | Required | Notes | |---|---|---| | `from` | ✅ | CEL expression → `map`. | | `into` | ✅ | Name of the destination variable. Dotted paths are supported. | **Output:** The merged variable value under its existing key. **Example:** [`examples/workflows/merge.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/merge.json) --- ### `filter` Keeps list elements where a CEL predicate is truthy. | Input | Required | Notes | |---|---|---| | `list` | ✅ | CEL expression → list. | | `expr` | ✅ | CEL predicate; the current element is bound as `_`. | | `out_var` | ✅ | Variable to write the filtered list to. | **Output:** `{out_var: [filtered list]}`. **Example:** [`examples/workflows/filter.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/filter.json) --- ### `map` Transforms every element of a list with a CEL expression. | Input | Required | Notes | |---|---|---| | `list` | ✅ | CEL expression → list. | | `expr` | ✅ | CEL transform; element bound as `_`. | | `out_var` | ✅ | Variable to write the mapped list to. | **Output:** `{out_var: [mapped list]}`. **Example:** [`examples/workflows/map.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/map.json) --- ### `assert` Fails the saga if a CEL expression is not truthy. Use it for invariant checks mid-workflow. | Input | Required | Notes | |---|---|---| | `expr` | ✅ | CEL boolean expression. | | `code` | optional | Error code emitted on failure (default `"assertion_failed"`). | **Output:** Empty map on success; non-retryable error on failure. **Example:** [`examples/workflows/assert.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/assert.json) --- ## Observability ### `log` Emits a structured log line via the engine's logger. | Input | Required | Notes | |---|---|---| | `message` | ✅ | Log message string. | | `level` | optional | `"info"` (default) \| `"warn"` \| `"error"`. | **Output:** Empty map. **Example:** [`examples/workflows/log.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/log.json) --- ### `metric_emit` Appends a named metric event to the run's audit stream. _(Prometheus side-channel wiring is planned for a future release.)_ | Input | Required | Notes | |---|---|---| | `name` | ✅ | Metric name string. | | `value` | ✅ | Numeric value. | | `labels` | optional | `map[string]string` of label key/value pairs. | **Output:** Empty map. **Example:** [`examples/workflows/metric_emit.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/metric_emit.json) --- ## I/O ### `http_request` Issues a synchronous outbound HTTP request and merges the response into `Variables`. | Input | Required | Notes | |---|---|---| | `url` | ✅ | Target URL. | | `method` | optional | HTTP method (default `"GET"`). | | `headers` | optional | `map[string]any` → request headers. | | `body` | optional | Any value; JSON-marshalled into the request body. | | `timeout_s` | optional | Request timeout in seconds (default `30`). | | `secret_ref` | optional | Secret key resolved via the Secrets resolver → set as `Authorization` header. | | `out_var` | optional | Output prefix (default `"http_result"`). | **Output keys** (with `out_var = "http_result"`): - `http_result` — parsed JSON body (or raw string if non-JSON). - `http_result_status` — `int64` HTTP status code. - `http_result_headers` — `map[string]string` of response headers. > ⚠️ License group is `common` only for `GET` with no `secret_ref`. Any other method or authenticated request requires the `external_io_advanced` group. **Example:** [`examples/workflows/http_request.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/http_request.json) --- ### `webhook_emit` POSTs a JSON payload to an external URL, with optional HMAC-SHA256 request signing. | Input | Required | Notes | |---|---|---| | `url` | ✅ | Target URL. | | `body` | ✅ | Any value; JSON-marshalled. | | `secret_ref` | optional | Secret key → `X-Webhook-Sig: sha256=` header. | | `timeout_s` | optional | Timeout in seconds (default `15`). | | `headers` | optional | Additional request headers. | | `async` | optional | `bool`; default `false`. When `true`, fires the request in a goroutine and returns immediately (failures are logged only). | | `out_var` | optional | Output prefix (default `"webhook_result"`). | **Output (sync mode):** `{out_var}_status` — `int64` HTTP status code. **Output (async mode):** `{out_var}_async: true`. **Example:** [`examples/workflows/webhook_emit.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/webhook_emit.json) --- ## Timing ### `wait_duration` Pauses the saga for a duration expressed as a Go duration string. | Input | Required | Notes | |---|---|---| | `duration` | ✅ | Go duration string e.g. `"5s"`, `"1h30m"`, `"72h"`. | The engine's timer dispatcher wakes the run when the deadline passes. **Example:** [`examples/workflows/wait_duration.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/wait_duration.json) --- ### `wait_until` Pauses the saga until an absolute point in time. | Input | Required | Notes | |---|---|---| | `timestamp` | ✅ | RFC3339 datetime string, e.g. `"2026-01-01T09:00:00Z"`. | **Example:** [`examples/workflows/wait_until.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/wait_until.json) --- ## Events & Signals ### `wait_for_signal` Pauses the saga until a named external signal arrives via `POST /api/v1/sagas/{run_id}/signal/{name}`. | Input | Required | Notes | |---|---|---| | `name` | ✅ | Signal name to await. | | `timeout_s` | optional | Max seconds to wait. Omit to wait indefinitely. | **Timeout routing:** when the deadline fires while the signal has not yet arrived, the engine checks `step.Branches["timeout"].Next` first. If that branch exists, the run routes there instead of `step.Next` — handy for escalation paths. > 💡 Wire a `timeout` branch to an escalation step to handle missed approvals or SLA breaches without any extra polling. **Example:** [`examples/workflows/wait_for_signal.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/wait_for_signal.json) --- ### `wait_for_event` Pauses the saga until an event with a matching topic (and optional header subset) arrives via the event bus. | Input | Required | Notes | |---|---|---| | `topic` | ✅ | Event topic / RabbitMQ routing key to await. | | `headers` | optional | `map[string]any` — incoming event headers must contain all of these key/value pairs (string equality). | | `timeout_s` | optional | Number of seconds to wait. On timeout the run routes to the step's `timeout` branch if defined, else to `next`. Omitted = wait indefinitely. | > 💡 Add a `"timeout"` entry to the step's `branches` to escalate when no matching event arrives before `timeout_s` (same pattern as `wait_for_signal`). **Example:** [`examples/workflows/wait_for_event.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/wait_for_event.json) --- ### `emit_signal` Sends a signal to another run (the send-side complement of `wait_for_signal`). If the target is currently paused awaiting that signal, it is consumed and the target advances immediately. | Input | Required | Notes | |---|---|---| | `run_id` | ✅ | UUID of the target run. | | `name` | ✅ | Signal name. | | `payload` | optional | `map[string]any` carried with the signal. | **Output:** Empty map. **Example:** [`examples/workflows/emit_signal.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/emit_signal.json) --- ### `emit_event` Publishes an event via the configured `EventEmitter` (in-process when embedded; RabbitMQ in service mode). | Input | Required | Notes | |---|---|---| | `topic` | ✅ | Event topic / routing key. | | `headers` | optional | `map[string]any` → `map[string]string`. | | `payload` | optional | `map[string]any` event payload. | **Output:** Empty map. > 💡 In embedded mode the in-process emitter both wakes runs awaiting the topic **and** runs the trigger dispatcher, so matching `record_transition` triggers start new runs — parity with service mode (no broker needed). **Example:** [`examples/workflows/emit_event.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/emit_event.json) --- ## Loops & Recovery ### `while` Loops while a CEL condition evaluates to `true`. | Input | Required | Notes | |---|---|---| | `condition` | ✅ | CEL boolean expression over `Variables`. | | `max_iterations` | optional | Default `100`; hard cap `10000`. Prevents runaway loops. | **Output:** `{"branch": "continue" | "exit"}`. **Wiring pattern:** - `step.Branches.continue → next` — first step of the loop body. - `step.Branches.exit → next` — first step after the loop. - The body's last step sets `next` back to the `while` step. An iteration counter is maintained at `Variables._while..iter`. **Example:** [`examples/workflows/while.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/while.json) --- ### `try_catch` Pushes an error-handler frame. If any step inside the protected region errors, the saga jumps to the `catch` step instead of failing. The error context is written to `Variables._error`. | Input | Required | Notes | |---|---|---| | `try` | ✅ | `[]string` of step IDs in the protected region. Used by `ValidateDefinition` to reject disallowed nesting (e.g. `parallel` inside `try`). | | `catch` | ✅ | Step ID to jump to on error. | **Wiring:** set `step.Next` to the first step inside the `try` body. The body's last step sets `next` to whatever comes after the protected region. **Example:** [`examples/workflows/try_catch.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/try_catch.json) --- ### `cancel` Cancels a run. | Input | Required | Notes | |---|---|---| | `run_id` | optional | UUID of the target run. Omit (or set to the current run's ID) to self-cancel — the current run ends as `cancelled`. | | `reason` | optional | Human-readable reason string. | **Self-cancel:** returns `ErrSagaCancelled`; the run ends immediately as `cancelled`. **Target-cancel:** cancels the target run and the current run **continues** to `step.Next`. > ⚠️ When you cancel a run that is a child of a `parallel` join, the join may be left waiting if the join strategy expects all children. The cancelled child is counted as terminal, so with `join_strategy: "all"` the parent will eventually time out or remain paused unless all other children also complete. **Example:** [`examples/workflows/cancel.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/cancel.json) --- ## Parallelism ### `parallel` Fans out N branches as child runs and pauses the parent until the join strategy is satisfied. | Input | Required | Notes | |---|---|---| | `branches` | ✅ | `[]any` of branch objects **or** a CEL string → list. Each branch: long-form `{"start": "step_id", "steps": [...]}` or short-form `{"type": "...", "inputs": {...}}`. | | `join_strategy` | optional | `"all"` (default) — wait for every branch. `"quorum"` — wake after `quorum_n` successes. | | `quorum_n` | required when `quorum` | Positive integer ≤ branch count. Can also be a CEL string evaluated at runtime. | **Output:** Each branch's variables are aggregated into `Variables._parallel..branches` when the parent wakes. > ⚠️ Remaining branches continue to run after a quorum wake — they are not cancelled. **Example:** [`examples/workflows/parallel.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/parallel.json) --- ### `join` A **barrier** that reconvenes streams an earlier step spawned independently, before the run continues. Where `parallel` spawns its own branches and pauses on the same step, `join` waits on children that *previous* steps spawned in the same run — the natural producer is [`spawn_saga`](#spawn_saga), whose fire-and-forget children the run did not wait on. `join` lets a later step gather those streams back together. | Input | Required | Notes | |---|---|---| | `streams` | ✅ | `[]string` of upstream **step IDs** in this run whose spawned children the join waits on (or a CEL string → list of step IDs). Each named step must have spawned at least one child (`spawn_saga`, `parallel`, `foreach`, or `sub_saga`); the join watches the union of their children. | | `join_strategy` | optional | `"all"` (default) — wait for every watched child to reach a terminal state. `"quorum"` — resolve once `quorum_n` watched children have succeeded. | | `quorum_n` | required when `quorum` | Positive integer ≤ the watched-child count. Can also be a CEL string evaluated at runtime. | **Resolution:** If every/quorum watched child is already terminal when the join runs, it aggregates and continues without pausing. Otherwise it pauses; the engine re-evaluates the barrier whenever a watched child terminates and wakes the run once the strategy is met. **Output:** Each watched child's variables are aggregated into `Variables._join..branches` (same `{key, variables, state, _user_task?}` shape as `parallel`). > ⚠️ A misconfigured barrier fails fast: `streams` naming a step that spawned no children, or `quorum_n` exceeding the watched-child count, errors at the step rather than pausing forever. Remaining children continue to run after a quorum wake — they are not cancelled. **Example:** [`examples/workflows/join.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/join.json) --- ### `foreach` Fans out one child run per element of a CEL-evaluated list (parallel mode only in v1). | Input | Required | Notes | |---|---|---| | `list` | ✅ | CEL expression → list. | | `body` | ✅ | `[]any` of step objects forming the loop body. | | `start` | ✅ | ID of the first step inside `body`. | | `parallel` | optional | `bool`; default `true`. Sequential mode is not yet supported — use `while` with an index counter for sequential loops. | Each child run receives `Variables._foreach_item` (the element) and `Variables._foreach_index` (zero-based index). An empty list advances without spawning. **Example:** [`examples/workflows/foreach.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/foreach.json) --- ## Composition (Call Tree) ### `sub_saga` Starts a named workflow as a child saga and **pauses the parent** until the child reaches a terminal state. | Input | Required | Notes | |---|---|---| | `workflow_id` | ✅ | Stable ID of the child `WorkflowDefinition`. | | `inputs` | optional | `map[string]any` passed as the child's initial inputs. | | `entrypoint` | optional | Named entry point on the child definition (see `Entrypoints`). Defaults to `Start`. | **Output:** Empty map on parent resume (child variables are not automatically merged; wire a `set_var`/`transform` after if needed). **Example:** [`examples/workflows/sub_saga.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/sub_saga.json) --- ### `spawn_saga` Starts a named workflow as a **fire-and-forget** child. The parent continues immediately to `step.Next` without waiting. | Input | Required | Notes | |---|---|---| | `workflow_id` | ✅ | Stable ID of the child workflow. | | `inputs` | optional | `map[string]any` passed as the child's initial inputs. | | `entrypoint` | optional | Named entry point on the child. Defaults to `Start`. | **Output:** Empty map; parent is not paused. **Example:** [`examples/workflows/spawn_saga.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/spawn_saga.json) --- ## Human Interaction ### `manual_approval` Creates a user task and pauses the saga until the assignee submits it via `POST /api/v1/sagas/{run_id}/user_task/{task_id}/submit`. | Input | Required | Notes | |---|---|---| | `assignee` | ✅ | User ID or role expected to submit. | | `due_in` | optional | Go duration string — sets `due_at = now + due_in`. | | `form_schema` | optional | `map[string]any` rendered in the admin UI. | **Output:** None on pause; the submitted form data is available after the run resumes. **Example:** [`examples/workflows/manual_approval.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/manual_approval.json) --- ### `collect_input` Like `manual_approval` but `form_schema` is **required**. Use this when the workflow needs structured data from the user (remediation notes, parameters, etc.) rather than a simple approve/reject. | Input | Required | Notes | |---|---|---| | `assignee` | ✅ | User ID or role. | | `form_schema` | ✅ | `map[string]any` schema — must be non-empty. | | `due_in` | optional | Go duration deadline. | **Output:** None on pause. **Example:** [`examples/workflows/collect_input.json`](https://github.com/Bugs5382/go-saga-orchestration/blob/main/examples/workflows/collect_input.json)