workflow

package
v0.10.0 Latest Latest
Warning

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

 Go to latest Published: Aug 11, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

Documentation

Overview

Package workflow provides ShiftLock's workflow foundation: definitions, step/compensate graphs, durable checkpoints, dry-run, and Runtime-soft integration with guard/audit/lockdown without importing the root module.

Index

Constants

View Source 
const DefaultMaxInstances = 256

DefaultMaxInstances bounds in-memory/file store cardinality.

View Source 
const DefaultMaxParallel = 8

DefaultMaxParallel bounds concurrent step execution within a parallel group.

Variables

View Source 
var (
	ErrInvalidDefinition      = errors.New("workflow: invalid definition")
	ErrUnknownStep            = errors.New("workflow: unknown step")
	ErrCycle                  = errors.New("workflow: step dependency cycle")
	ErrInvalidState           = errors.New("workflow: invalid state transition")
	ErrNotFound               = errors.New("workflow: instance not found")
	ErrClosed                 = errors.New("workflow: engine closed")
	ErrLockdown               = errors.New("workflow: lockdown blocks mutation")
	ErrCapability             = errors.New("workflow: resource capability validation failed")
	ErrAmbiguous              = errors.New("workflow: ambiguous step outcome")
	ErrRequiresReconciliation = errors.New("workflow: requires reconciliation")
	ErrNotRetryable           = errors.New("workflow: step is not retryable")
	ErrCompensationFailed     = errors.New("workflow: compensation failed")
	ErrStaleEpoch             = errors.New("workflow: resource epoch newer than compensation target")
	ErrCancelled              = errors.New("workflow: cancelled")
	ErrBoundExceeded          = errors.New("workflow: bound exceeded")
	ErrInvalidArgument        = errors.New("workflow: invalid argument")
)

Functions

func CanTransition 

func CanTransition(from, to State) bool

CanTransition reports whether an engine may move from -> to. This is a conservative allow-list for validation/fuzzing; the engine may use a subset of these edges.

Types

type ActionFunc 

type ActionFunc func(ctx context.Context, exec *ExecContext) (Result, error)

ActionFunc executes a step.

type Auditor 

type Auditor interface {
	Audit(actor, action, resource, result, operationID string)
}

Auditor is a soft dependency on Runtime audit.

type Builder 

type Builder struct {
	// contains filtered or unexported fields
}

Builder constructs a Definition.

func Define 

func Define(name string) *Builder

Define starts a workflow definition.

func (*Builder) Build 

func (b *Builder) Build() (*Definition, error)

Build validates and returns an immutable Definition.

func (*Builder) Compensate 

func (b *Builder) Compensate(step string, fn CompensateFunc) *Builder

Compensate attaches a compensating action to a step.

func (*Builder) Depend 

func (b *Builder) Depend(step string, deps ...string) *Builder

Depend declares that step depends on deps (deps run first).

func (*Builder) Idempotency 

func (b *Builder) Idempotency(step string, mode IdempotencyMode) *Builder

Idempotency sets retry mode for a step.

func (*Builder) Mutating 

func (b *Builder) Mutating(step string, mutates bool) *Builder

Mutating marks a step as a protected resource mutation.

func (*Builder) ParallelGroup 

func (b *Builder) ParallelGroup(group string, steps ...string) *Builder

ParallelGroup assigns steps to a named parallel group (same group may run concurrently).

func (*Builder) RequireCaps 

func (b *Builder) RequireCaps(step string, id resource.ResourceID, caps resource.ResourceCapabilities) *Builder

RequireCaps requires resource capabilities when a ResourceID is set on the step.

func (*Builder) Step 

func (b *Builder) Step(name string, action ActionFunc) *Builder

Step adds or replaces a step.

type Checkpoint 

type Checkpoint struct {
	InstanceID string               `json:"instance_id"`
	Workflow   string               `json:"workflow"`
	State      State                `json:"state"`
	Steps      map[string]StepState `json:"steps"`
	UpdatedAt  time.Time            `json:"updated_at"`
	DryRun     bool                 `json:"dry_run,omitempty"`
	Attrs      map[string]string    `json:"attrs,omitempty"`
	// CompletedSteps lists successfully completed steps in order (for compensate).
	CompletedSteps []string `json:"completed_steps,omitempty"`
	// EpochAtStep records resource epochs observed at step completion.
	EpochAtStep map[string]uint64 `json:"epoch_at_step,omitempty"`
}

Checkpoint is durable workflow progress.

type CompensateFunc 

type CompensateFunc func(ctx context.Context, exec *ExecContext) (Result, error)

CompensateFunc undoes a completed step.

type Definition 

type Definition struct {
	Name string
	// contains filtered or unexported fields
}

Definition is an immutable workflow graph after Validate.

func (*Definition) Step 

func (d *Definition) Step(name string) (*StepDef, bool)

Step returns a step definition.

func (*Definition) Steps 

func (d *Definition) Steps() []string

Steps returns step names in deterministic dependency order.

type Engine 

type Engine struct {
	// contains filtered or unexported fields
}

Engine executes workflow definitions with checkpoints and soft Runtime hooks.

func NewEngine 

func NewEngine(cfg EngineConfig) *Engine

NewEngine constructs an engine. Store defaults to MemoryStore.

func (*Engine) Close 

func (e *Engine) Close()

Close prevents new runs.

func (*Engine) Get 

func (e *Engine) Get(instanceID string) (*Instance, error)

Get returns a live instance snapshot.

func (*Engine) ListDefinitions 

func (e *Engine) ListDefinitions() []string

ListDefinitions returns registered workflow names.

func (*Engine) ListInstances 

func (e *Engine) ListInstances() []Instance

ListInstances returns snapshots of tracked instances.

func (*Engine) Register 

func (e *Engine) Register(def *Definition) error

Register adds a validated definition.

func (*Engine) Run 

func (e *Engine) Run(ctx context.Context, workflowName string, opts RunOptions) (*Instance, error)

Run executes (or resumes) a workflow.

func (*Engine) SetHooks 

func (e *Engine) SetHooks(h Hooks)

SetHooks updates soft Runtime integrations.

type EngineConfig 

type EngineConfig struct {
	Store       Store
	Hooks       Hooks
	Clock       func() time.Time
	MaxRuns     int // bound concurrent tracked instances; 0 → DefaultMaxInstances
	MaxParallel int // bound parallel group concurrency; 0 → DefaultMaxParallel
}

EngineConfig configures the workflow engine.

type Error 

type Error struct {
	Op       string
	Workflow string
	Step     string
	Err      error
	Message  string
}

Error is a typed workflow error.

func (*Error) Error 

func (e *Error) Error() string

func (*Error) Unwrap 

func (e *Error) Unwrap() error

type Evidence 

type Evidence struct {
	Time    time.Time         `json:"time"`
	Event   string            `json:"event"`
	Summary string            `json:"summary,omitempty"`
	Attrs   map[string]string `json:"attrs,omitempty"`
}

Evidence is a size-bounded observation (aligned with resource.Evidence).

type ExecContext 

type ExecContext struct {
	Workflow    string
	InstanceID  string
	Step        string
	Attempt     int
	DryRun      bool
	OperationID string
	Attrs       map[string]string
	// ResourceEpoch is the epoch observed at step start (for compensation fencing).
	ResourceEpoch resource.ResourceEpoch
	ResourceID    resource.ResourceID
}

ExecContext is passed to actions.

type FileStore 

type FileStore struct {
	// contains filtered or unexported fields
}

FileStore persists checkpoints as a single JSON map with fsync-ish flush (write temp → Sync → rename). Suitable for local-first single-process use.

func NewFileStore 

func NewFileStore(path string, max int) *FileStore

NewFileStore creates a file-backed store. Parent directories are created on first Save.

func (*FileStore) Delete 

func (s *FileStore) Delete(instanceID string) error

func (*FileStore) List 

func (s *FileStore) List() ([]Checkpoint, error)

func (*FileStore) Load 

func (s *FileStore) Load(instanceID string) (Checkpoint, error)

func (*FileStore) Save 

func (s *FileStore) Save(cp Checkpoint) error

type Hooks 

type Hooks struct {
	Lockdown  LockdownGate
	Audit     Auditor
	Resources ResourceLookup
}

Hooks wires soft Runtime integrations without import cycles.

type IdempotencyMode 

type IdempotencyMode string

IdempotencyMode controls retry semantics. 

const (
	// Idempotent — safe to retry on transient failure.
	Idempotent IdempotencyMode = "idempotent"
	// RequiresOperationID — retries only with a stable operation id.
	RequiresOperationID IdempotencyMode = "requires-operation-id"
	// NotRetryable — never retry; fail the step.
	NotRetryable IdempotencyMode = "not-retryable"
	// RequiresReconciliation — ambiguous outcomes must not be blindly retried.
	RequiresReconciliation IdempotencyMode = "requires-reconciliation"
)

type Instance 

type Instance struct {
	ID         string
	Workflow   string
	State      State
	DryRun     bool
	Checkpoint Checkpoint
}

Instance is a running or paused workflow.

type JournalStore 

type JournalStore struct {
	// contains filtered or unexported fields
}

JournalStore appends NDJSON checkpoint snapshots and recovers the latest state per instance on open. Deletes are recorded as tombstones.

func NewJournalStore 

func NewJournalStore(path string, max int) (*JournalStore, error)

NewJournalStore opens (or creates) a journal-backed store and replays it.

func (*JournalStore) Delete 

func (s *JournalStore) Delete(instanceID string) error

func (*JournalStore) List 

func (s *JournalStore) List() ([]Checkpoint, error)

func (*JournalStore) Load 

func (s *JournalStore) Load(instanceID string) (Checkpoint, error)

func (*JournalStore) Save 

func (s *JournalStore) Save(cp Checkpoint) error

type LockdownGate 

type LockdownGate interface {
	BlocksMutations() bool
}

LockdownGate is a soft dependency on Runtime lockdown.

type MemoryStore 

type MemoryStore struct {
	// contains filtered or unexported fields
}

MemoryStore is an in-process durable store for tests and local-first mode.

func NewMemoryStore 

func NewMemoryStore(max int) *MemoryStore

NewMemoryStore creates a MemoryStore.

func (*MemoryStore) Delete 

func (s *MemoryStore) Delete(instanceID string) error

func (*MemoryStore) List 

func (s *MemoryStore) List() ([]Checkpoint, error)

func (*MemoryStore) Load 

func (s *MemoryStore) Load(instanceID string) (Checkpoint, error)

func (*MemoryStore) Save 

func (s *MemoryStore) Save(cp Checkpoint) error

type ResourceLookup 

type ResourceLookup interface {
	Get(id resource.ResourceID) (*resource.Entry, error)
}

ResourceLookup resolves registered resources for capability/epoch checks.

type Result 

type Result struct {
	Evidence Evidence
	// Ambiguous marks an unknown outcome — engine enters requires-reconciliation
	// instead of retrying when mode demands it.
	Ambiguous bool
}

Result is the outcome of an action or compensation.

type RunOptions 

type RunOptions struct {
	InstanceID  string
	DryRun      bool
	OperationID string
	Attrs       map[string]string
	// Resume loads an existing checkpoint by InstanceID instead of starting fresh.
	Resume bool
}

RunOptions configures a single run.

type State 

type State string

State is the lifecycle state of a workflow instance. 

const (
	StateCreated                State = "created"
	StateValidating             State = "validating"
	StateWaiting                State = "waiting"
	StateRunning                State = "running"
	StatePaused                 State = "paused"
	StateCompensating           State = "compensating"
	StateCompleted              State = "completed"
	StateFailed                 State = "failed"
	StateCancelled              State = "cancelled"
	StateBlocked                State = "blocked"
	StateLockedDown             State = "locked-down"
	StateRequiresIntervention   State = "requires-intervention"
	StateRequiresReconciliation State = "requires-reconciliation"
)

func (State) Terminal 

func (s State) Terminal() bool

Terminal reports whether s is a terminal state.

type StepDef 

type StepDef struct {
	Name          string
	Action        ActionFunc
	Compensate    CompensateFunc
	DependsOn     []string
	ParallelGroup string
	Idempotency   IdempotencyMode
	Mutates       bool // protected mutation — blocked under lockdown
	RequiredCaps  resource.ResourceCapabilities
	ResourceID    resource.ResourceID // optional capability/epoch target
}

StepDef describes one step in a definition.

type StepState 

type StepState struct {
	Name       string     `json:"name"`
	Status     StepStatus `json:"status"`
	Attempt    int        `json:"attempt"`
	StartedAt  time.Time  `json:"started_at,omitempty"`
	FinishedAt time.Time  `json:"finished_at,omitempty"`
	Error      string     `json:"error,omitempty"`
	Evidence   []Evidence `json:"evidence,omitempty"`
}

StepState is runtime progress for one step.

type StepStatus 

type StepStatus string

StepStatus is the status of one step within an instance. 

const (
	StepPending      StepStatus = "pending"
	StepRunning      StepStatus = "running"
	StepCompleted    StepStatus = "completed"
	StepFailed       StepStatus = "failed"
	StepCompensating StepStatus = "compensating"
	StepCompensated  StepStatus = "compensated"
	StepSkipped      StepStatus = "skipped"
	StepReconcile    StepStatus = "requires-reconciliation"
)

type Store 

type Store interface {
	Save(cp Checkpoint) error
	Load(instanceID string) (Checkpoint, error)
	Delete(instanceID string) error
	List() ([]Checkpoint, error)
}

Store persists checkpoints.

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.