omniworkflow

README

OmniWorkflow

A provider-neutral durable-orchestration abstraction: one Go interface with interchangeable backends (AWS Step Functions and Temporal), so you write pipeline-shaped orchestration once and run it on either engine by configuration.

OmniWorkflow follows the plexusone Omni* convention — provider-neutral capability libraries that *Forge products consume. ActionForge is the first consumer.

Installation

go get github.com/plexusone/omniworkflow

Model

You hand Start a Plan — a flat DAG of steps connected by DependsOn edges, with Gate steps that hold until an approval Signal — and register a StepExecutor callback that performs each step's work. Each provider ships a generic DAG interpreter plus an embedded worker, so you never author Temporal workflow code or Step Functions state-machine definitions.

Plan ──▶ WorkflowProvider.Start ──▶ [interpreter walks the DAG]
                                          │
                                          ▼
                              StepExecutor.ExecuteStep (your code)

The Plan targets pipeline-shaped work; it is deliberately not a general-purpose workflow language (no loops, dynamic fan-out, or arbitrary control flow).

Providers

Provider	Package	Status
In-memory (reference/testing)	github.com/plexusone/omniworkflow/inmemory	ready
Durable embedded (no cluster)	github.com/plexusone/omniworkflow/durable	ready
AWS Step Functions	github.com/plexusone/omniworkflow/stepfunctions	ready
Temporal	github.com/plexusone/omniworkflow/temporal	ready

A shared conformance suite (github.com/plexusone/omniworkflow/conformance) proves every provider behaves identically. The in-memory and durable providers run it hermetically on every go test; the Step Functions provider runs it against live AWS, and the Temporal provider against a dev or real server, when configured. The Temporal provider's workflow logic is additionally unit-tested hermetically in the Temporal test environment.

The durable provider survives process restarts with no cluster, broker, or AWS — it checkpoints a small snapshot (plan + the interpreter's outcome sets) to a pluggable StateStore (a zero-dependency file store by default) after each wave, and resumes non-terminal runs on startup. Steps are at-least-once, so step executors must be idempotent. It is the low-idle-cost self-hosted option; a production deployment can supply a Postgres- or bbolt-backed StateStore.

The in-memory provider is a non-durable reference for tests and local use.

License

MIT

Documentation

Overview

Package omniworkflow is a provider-neutral durable-orchestration abstraction: one Go interface (WorkflowProvider) with interchangeable backends (AWS Step Functions and Temporal), so a caller writes pipeline- shaped orchestration once and runs it on either engine by configuration.

It follows the plexusone Omni* convention: OmniX names are provider-neutral capability libraries, XForge names are the products that consume them. ActionForge (github.com/plexusone/actionforge) is the first consumer.

Model

Callers hand Start a Plan — a flat DAG of PlanStep values connected by DependsOn edges, with Gate steps that hold until an approval Signal — and register a StepExecutor callback that performs each step's work. Each provider ships a generic DAG interpreter plus an embedded worker (NewWorker), so callers never author Temporal workflow code or Step Functions state-machine definitions. The normalized RunStatus vocabulary hides each backend's native execution-status types from callers.

The Plan is deliberately not a general-purpose workflow language: loops, dynamic fan-out, and arbitrary control flow are out of scope. It targets pipeline-shaped work.

Index

• Constants
• Variables
• func ApproveSignal(stepID string) string
• type Plan
  • func (p Plan) Validate() error
• type PlanStep
• type Run
• type RunStatus
  • func (s RunStatus) IsTerminal() bool
• type Signal
• type StartRequest
• type StepExecutor
• type Worker
• type WorkflowProvider

Constants

const Version = "0.1.0"

Version is the module's semantic version.

Variables

var (
	ErrEmptyPlan         = errors.New("omniworkflow: plan has no steps")
	ErrEmptyStepID       = errors.New("omniworkflow: step has empty id")
	ErrDuplicateStepID   = errors.New("omniworkflow: duplicate step id")
	ErrUnknownDependency = errors.New("omniworkflow: step depends on unknown step id")
	ErrSelfDependency    = errors.New("omniworkflow: step depends on itself")
	ErrCyclicPlan        = errors.New("omniworkflow: plan contains a dependency cycle")
	ErrRunNotFound       = errors.New("omniworkflow: run not found")
	ErrDuplicateRun      = errors.New("omniworkflow: run id already exists")
)

Errors returned across the package. Providers wrap ErrRunNotFound and ErrDuplicateRun so callers can match with errors.Is.

Functions

func ApproveSignal

func ApproveSignal(stepID string) string

ApproveSignal returns the signal name that approves a Gate step. A Gate step holds until a Signal with this name is delivered, then executes.

Types

type Plan

type Plan struct {
	Steps []PlanStep `json:"steps"`
}

Plan is a flat DAG of steps. It is deliberately not a general-purpose workflow language: loops, dynamic fan-out, and arbitrary control flow are out of scope. It targets pipeline-shaped work.

The JSON tags are load-bearing: providers serialize the Plan across the orchestration boundary (Step Functions carries it as execution state), so the wire shape is a stable part of the contract.

func (Plan) Validate

func (p Plan) Validate() error

Validate checks the Plan's structure: it must be non-empty, every step ID must be unique and non-empty, every DependsOn must reference an existing step and never the step itself, and the dependency graph must be acyclic. Runtime execution semantics (readiness, gates, failure propagation) are the interpreter's concern, not Validate's.

type PlanStep

type PlanStep struct {
	// ID uniquely identifies the step within its Plan.
	ID string `json:"id"`
	// DependsOn lists step IDs that must succeed before this step is ready.
	// Empty means the step is ready immediately.
	DependsOn []string `json:"dependsOn,omitempty"`
	// Gate, when true, makes the step hold once ready until an approval
	// Signal (ApproveSignal(ID)) is delivered, then execute.
	Gate bool `json:"gate,omitempty"`
	// Input is opaque data passed to the StepExecutor verbatim.
	// JSON-serializable; after crossing the provider boundary it is decoded
	// as generic JSON (e.g. map[string]any), so executors should treat it as
	// such rather than relying on a concrete Go type identity.
	Input any `json:"input,omitempty"`
}

PlanStep is a single unit of work in a Plan.

type Run

type Run struct {
	ID       string
	Status   RunStatus
	Output   any               // populated on success; JSON-serializable
	Metadata map[string]string // provider-specific extras, opt-in (e.g. execution ARN)
}

Run is a provider-neutral view of a workflow run.

type RunStatus

type RunStatus string

RunStatus is the normalized, provider-independent lifecycle status of a run. Every provider maps its native execution-status vocabulary (Temporal's WorkflowExecutionStatus, Step Functions' ExecutionStatus, ...) into these values so callers never see backend-specific types.

const (
	StatusRunning   RunStatus = "RUNNING"
	StatusSucceeded RunStatus = "SUCCEEDED"
	StatusFailed    RunStatus = "FAILED"
	StatusCancelled RunStatus = "CANCELLED"
)

func (RunStatus) IsTerminal

func (s RunStatus) IsTerminal() bool

IsTerminal reports whether the status is a final state that will not change.

type Signal

type Signal struct {
	Name    string
	Payload any // JSON-serializable
}

Signal is a message delivered to a running run.

type StartRequest

type StartRequest struct {
	// ID is the caller-assigned run identity and idempotency key.
	ID string
	// Plan is the DAG to execute.
	Plan Plan
}

StartRequest is the input to WorkflowProvider.Start.

type StepExecutor

type StepExecutor interface {
	ExecuteStep(ctx context.Context, runID string, step PlanStep) (output any, err error)
}

StepExecutor performs the work of a single step. The caller registers one StepExecutor with a provider's worker; the interpreter invokes it for each ready step. runID is the enclosing run's ID (StartRequest.ID).

type Worker

type Worker interface {
	// Run starts the worker and blocks until ctx is cancelled or a fatal
	// error occurs. A clean shutdown via ctx cancellation returns nil.
	Run(ctx context.Context) error
}

Worker hosts a provider's embedded execution loop, dispatching ready steps to the registered StepExecutor. A caller obtains one from a provider's NewWorker and runs it inside its own process.

type WorkflowProvider

type WorkflowProvider interface {
	// Start begins executing req.Plan. The run is identified by req.ID, which
	// is also the idempotency key: starting again with the same ID must not
	// create a duplicate execution (it returns the existing run).
	Start(ctx context.Context, req StartRequest) (*Run, error)

	// Get returns the current state of the run with the given id.
	// It returns ErrRunNotFound if no such run exists.
	Get(ctx context.Context, id string) (*Run, error)

	// Signal delivers a signal to a running run — most importantly the
	// approval of a Gate step (see ApproveSignal). It returns ErrRunNotFound
	// if no such run exists.
	Signal(ctx context.Context, id string, signal Signal) error

	// Cancel requests cancellation of the run. After cancellation no further
	// steps execute and the run's terminal status is StatusCancelled.
	Cancel(ctx context.Context, id string) error
}

WorkflowProvider is the provider-neutral control-plane interface. A provider (Step Functions, Temporal, ...) implements it to start and manage runs of a Plan. Callers program against this interface alone and select a provider by configuration.

Values carried by any-typed fields (PlanStep.Input, Run.Output, Signal.Payload) cross the provider boundary and must be JSON-serializable — Step Functions passes state as JSON, and Temporal serializes payloads.