verbs

package
v0.6.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 31, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Overview

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.

Index

Constants

This section is empty.

Variables

View Source 
var ErrSagaCancelled = errors.New("saga cancelled")

ErrSagaCancelled is returned by the cancel verb for a self-cancel. Advance transitions the run to cancelled (terminal) and stops the loop.

View Source 
var ErrSagaPaused = errors.New("saga paused")

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.

View Source 
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",
}

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".

Functions

func AggregateJoinResults added in v0.6.0 

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.<step_id>.branches" (join) or "_parallel.<step_id>.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 

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 added in v0.6.0 

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 

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 added in v0.6.0 

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 

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 

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.

Types

type ActionDispatchPublisher 

type ActionDispatchPublisher interface {
	PublishActionDispatch(ctx context.Context, routingKey string, payload []byte) error
}

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.

type ActionHTTPDispatcher added in v0.3.0 

type ActionHTTPDispatcher interface {
	DispatchHTTP(ctx context.Context, address string, payload []byte) error
}

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)

type ActionPayload 

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"` // "<service>.<action_name>"
	Inputs         map[string]any `json:"inputs"`
	DryRun         bool           `json:"dry_run,omitempty"`
}

ActionPayload is the body of a saga.advance → action dispatch message. Workers deserialise this to drive their handler.

type ActionRMQDispatcher added in v0.3.0 

type ActionRMQDispatcher interface {
	DispatchRMQQueue(ctx context.Context, queue string, payload []byte) error
}

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)

type ActionVerb 

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
}

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): "<service>.<action_name>". 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.

func (ActionVerb) DispatchCompensation added in v0.3.0 

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 "<service>.<action_name>" form.

func (ActionVerb) Execute 

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 

type AssertVerb struct{}

AssertVerb evaluates a CEL expression; on false returns an error. Inputs:

  • "expr" (required, string)
  • "code" (optional, string; default "assertion_failed")

func (AssertVerb) Execute 

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 

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
}

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).

func (CancelVerb) Execute 

func (v CancelVerb) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error)

type CollectInputVerb 

type CollectInputVerb struct {
	S     store.Store
	Clock clock.Clock
}

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)

func (CollectInputVerb) Execute 

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 added in v0.3.0 

type CompensationPayload struct {
	RunID  string         `json:"run_id"`
	StepID string         `json:"step_id"` // the step being compensated
	Action string         `json:"action"`  // "<service>.<action_name>"
	Inputs map[string]any `json:"inputs"`
	DryRun bool           `json:"dry_run,omitempty"`
}

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.

type DecisionVerb 

type DecisionVerb struct {
	S store.Store
}

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.

func (DecisionVerb) Execute 

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 added in v0.3.0 

type DefaultOption func(*defaultConfig)

DefaultOption configures optional verb dependencies (e.g. the http/rmq action dispatchers) without breaking Default's positional signature.

func WithHTTPDispatcher added in v0.3.0 

func WithHTTPDispatcher(d ActionHTTPDispatcher) DefaultOption

WithHTTPDispatcher wires the http action dispatcher for transport="http" action registrations. (issue #59)

func WithRMQDispatcher added in v0.3.0 

func WithRMQDispatcher(d ActionRMQDispatcher) DefaultOption

WithRMQDispatcher wires the rmq action dispatcher for transport="rmq" action registrations. (issue #59)

type EmitEventVerb 

type EmitEventVerb struct {
	Emitter EventEmitter
}

EmitEventVerb publishes an event via the configured EventEmitter. Inputs:

  • "topic" (required, string)
  • "headers" (optional, map[string]any -> stringified)
  • "payload" (optional, map[string]any)

func (EmitEventVerb) Execute 

func (v EmitEventVerb) Execute(ctx context.Context, _ domain.SagaRun, step domain.Step) (map[string]any, error)

Execute implements Handler.

type EmitSignalVerb 

type EmitSignalVerb struct {
	S         store.Store
	Publisher Publisher
}

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.

func (EmitSignalVerb) Execute 

func (v EmitSignalVerb) Execute(ctx context.Context, _ domain.SagaRun, step domain.Step) (map[string]any, error)

Execute implements Handler.

type EndVerb 

type EndVerb struct{}

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.

func (EndVerb) Execute 

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 

type ErrorVerb struct{}

ErrorVerb halts the saga with a non-retryable error. Inputs:

  • "code" (required, string)
  • "message" (optional, string)

func (ErrorVerb) Execute 

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 

type EventEmitter interface {
	EmitEvent(ctx context.Context, topic string, headers map[string]string, payload map[string]any) error
}

EventEmitter publishes an event other sagas/triggers can match. Real impls: an in-process matcher (embedded) or a RabbitMQ publisher (service mode).

type FilterVerb 

type FilterVerb struct{}

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.

func (FilterVerb) Execute 

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 

type ForeachVerb struct {
	S         store.Store
	Publisher Publisher
}

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.

func (ForeachVerb) Execute 

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 

type HTTPRequestVerb struct {
	Secrets secrets.Resolver
	Client  *http.Client // optional; nil → constructed per-call with timeout_s
}

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.

func (HTTPRequestVerb) Execute 

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 

type Handler interface {
	Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error)
}

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.

type HandlerFunc 

type HandlerFunc func(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error)

HandlerFunc adapts a plain function to the Handler interface so a custom verb can be a closure instead of a struct.

func (HandlerFunc) Execute 

func (f HandlerFunc) Execute(ctx context.Context, run domain.SagaRun, step domain.Step) (map[string]any, error)

Execute satisfies Handler.

type JoinVerb added in v0.6.0 

type JoinVerb struct {
	S store.Store
}

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.<step_id>.branches" as a list of {key, variables, state, _user_task?} entries, mirroring the "_parallel.<step_id>.branches" shape produced by the parallel verb.

func (JoinVerb) Execute added in v0.6.0 

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 

type LogVerb struct {
	S store.Store
}

LogVerb appends a saga_run_events row of type `log`. Inputs:

  • "message" (required, string)
  • "level" (optional, string: "info" | "warn" | "error"; default "info")

func (LogVerb) Execute 

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 

type ManualApprovalVerb struct {
	S     store.Store
	Clock clock.Clock
}

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.

func (ManualApprovalVerb) Execute 

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 

type MapVerb struct{}

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).

func (MapVerb) Execute 

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 

type MergeVerb struct{}

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).

func (MergeVerb) Execute 

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 

type MetricEmitVerb struct {
	S store.Store
}

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.

func (MetricEmitVerb) Execute 

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 

type NoopVerb struct{}

NoopVerb does nothing. Useful in tests and during authoring to hold a place in the workflow graph without side effects.

func (NoopVerb) Execute 

func (NoopVerb) Execute(_ context.Context, _ domain.SagaRun, _ domain.Step) (map[string]any, error)

Execute does nothing and returns an empty result.

type ParallelVerb 

type ParallelVerb struct {
	S         store.Store
	Publisher Publisher
}

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.

func (ParallelVerb) Execute 

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 

type ParentJoinChecker interface {
	CheckParentJoin(ctx context.Context, run domain.SagaRun)
}

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.

type Publisher 

type Publisher interface {
	PublishSagaAdvance(ctx context.Context, runID string) error
}

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.

type Registry 

type Registry map[domain.StepType]RegistryEntry

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.

func Default 

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 

type RegistryEntry struct {
	Handler      Handler
	LicenseGroup string // canonical group name, e.g. "external_io_advanced"
}

RegistryEntry is what each registered StepType resolves to. LicenseGroup lets the engine gate dispatch by the tenant's license features.

type SetVarVerb 

type SetVarVerb struct{}

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).

func (SetVarVerb) Execute 

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 

type SpawnSagaVerb struct {
	S         store.Store
	Publisher Publisher
}

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.

func (SpawnSagaVerb) Execute 

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 

type SubSagaVerb struct {
	S         store.Store
	Publisher Publisher
}

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).

func (SubSagaVerb) Execute 

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 

type SwitchVerb struct{}

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.

func (SwitchVerb) Execute 

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": <result>}.

type TransformVerb 

type TransformVerb struct{}

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).

func (TransformVerb) Execute 

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 

type TryCatchVerb struct {
	S store.Store
}

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.

func (TryCatchVerb) Execute 

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 

type WaitDurationVerb struct {
	S     store.Store
	Clock clock.Clock
}

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.

func (WaitDurationVerb) Execute 

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 

type WaitForEventVerb struct {
	S     store.Store
	Clock clock.Clock
}

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.

func (WaitForEventVerb) Execute 

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 

type WaitForSignalVerb struct {
	S     store.Store
	Clock clock.Clock
}

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.

func (WaitForSignalVerb) Execute 

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 

type WaitUntilVerb struct {
	S     store.Store
	Clock clock.Clock
}

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.

func (WaitUntilVerb) Execute 

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 

type WebhookEmitVerb struct {
	Secrets secrets.Resolver
	Client  *http.Client
}

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=<hex>" 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.

func (WebhookEmitVerb) Execute 

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 

type WhileVerb struct{}

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).

func (WhileVerb) Execute 

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.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.