engine
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.
const FeatureCronTriggers = "wf.cron_triggers"
TimerAdvisoryLockID is the Postgres advisory lock the timer dispatcher holds while it's the leader.
const TimerAdvisoryLockID = int64(0xBA70C420)
func Backoff
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
func DefaultRetryPolicy() domain.RetryPolicy
DefaultRetryPolicy returns the spec default per § 3.4.
func InjectStartupVariables
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
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
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
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
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.
type Coordinator struct {
// contains filtered or unexported fields
}
func NewCoordinator
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
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
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
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
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.
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
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.
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.
type EventSubscriber struct {
S store.Store
Publisher TimerPublisher // same interface as Timer
Dispatcher *TriggerDispatcher // optional; nil = wake-only mode
}
func (*EventSubscriber) Deliver
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
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.
type LicenseGateError struct {
StepID string `json:"step_id"`
Group string `json:"group"`
Feature string `json:"feature"`
}
func (LicenseGateError) Error
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.
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.
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).
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
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.
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.
type TriggerDispatcher struct {
S store.Store
Publisher TimerPublisher // reuse — same PublishSagaAdvance method
StartupProviders []StartupVariableProvider
}
func (*TriggerDispatcher) Dispatch
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