Documentation
¶
Overview ¶
Package durable is a durable embedded WorkflowProvider. It survives process restarts without a cluster, an external broker, or AWS — proving OmniWorkflow's self-hostable, low-idle-cost thesis end-to-end.
Durability here is deliberately narrow: the provider checkpoints a small Snapshot (the plan plus three string-sets — succeeded, failed, approved — and the status) after each wave, and reloads it to resume on restart. There is no event-sourcing or workflow-code replay to get wrong, because the "workflow logic" is the pure interpreter core (interp.Advance) over that explicit state. Steps are at-least-once: a step in flight when the process crashes re-runs on resume, exactly as Temporal activities and Step Functions tasks are — so step executors must be idempotent.
The default FileStore is a zero-dependency reference. A production deployment can supply a Postgres- or bbolt-backed StateStore via the same interface.
Index ¶
- Variables
- type FileStore
- type MemStore
- type Provider
- func (p *Provider) Cancel(_ context.Context, id string) error
- func (p *Provider) Get(ctx context.Context, id string) (*omniworkflow.Run, error)
- func (p *Provider) NewWorker(executor omniworkflow.StepExecutor) omniworkflow.Worker
- func (p *Provider) Signal(_ context.Context, id string, signal omniworkflow.Signal) error
- func (p *Provider) Start(ctx context.Context, req omniworkflow.StartRequest) (*omniworkflow.Run, error)
- type Snapshot
- type StateStore
Constants ¶
This section is empty.
Variables ¶
var ErrNoExecutor = errors.New("omniworkflow/durable: no executor registered; call NewWorker first")
ErrNoExecutor is returned by Start when no worker/executor has been registered yet (call NewWorker first).
var ErrNoPendingApproval = errors.New("omniworkflow/durable: no gate awaiting this approval")
ErrNoPendingApproval is returned by Signal when no gate is currently awaiting the given approval.
Functions ¶
This section is empty.
Types ¶
type FileStore ¶
type FileStore struct {
// contains filtered or unexported fields
}
FileStore is a zero-dependency StateStore that keeps one JSON file per run in a directory. Writes are atomic (temp file + rename) so a crash mid-write never corrupts a snapshot. It is the reference durable store; a production deployment may prefer a Postgres- or bbolt-backed StateStore.
func NewFileStore ¶
NewFileStore returns a FileStore rooted at dir, creating the directory if needed.
type MemStore ¶
type MemStore struct {
// contains filtered or unexported fields
}
MemStore is an in-memory StateStore. It is durable only within a single process lifetime — useful for the store's own tests and for exercising the resume path without disk. For real durability use FileStore.
type Provider ¶
type Provider struct {
// contains filtered or unexported fields
}
Provider is the durable embedded WorkflowProvider. It checkpoints run state to a StateStore after every wave and resumes non-terminal runs on startup.
func New ¶
func New(store StateStore) *Provider
New returns a durable Provider backed by store. Register an executor and start the lifecycle (including resuming persisted runs) by obtaining a Worker with NewWorker and running it.
func (*Provider) Get ¶
Get returns the current state of the run, consulting the store if it is not in memory (e.g. a terminal run from a previous process).
func (*Provider) NewWorker ¶
func (p *Provider) NewWorker(executor omniworkflow.StepExecutor) omniworkflow.Worker
NewWorker registers executor as the provider's step executor and returns a Worker. Running the worker resumes any non-terminal runs found in the store.
func (*Provider) Signal ¶
Signal approves a waiting gate. The signal name must be ApproveSignal(gateID).
func (*Provider) Start ¶
func (p *Provider) Start(ctx context.Context, req omniworkflow.StartRequest) (*omniworkflow.Run, error)
Start begins executing req.Plan, persisting an initial snapshot before the driver runs. The ID is the idempotency key: an existing run (in memory or in the store) is returned without relaunching.
type Snapshot ¶
type Snapshot struct {
RunID string `json:"runId"`
Plan omniworkflow.Plan `json:"plan"`
Succeeded map[string]bool `json:"succeeded"`
Failed map[string]bool `json:"failed"`
Approved map[string]bool `json:"approved"`
Status omniworkflow.RunStatus `json:"status"`
}
Snapshot is the complete durable state of one run. It is small by design: the plan plus the interpreter's three outcome sets and the status.
type StateStore ¶
type StateStore interface {
// Save writes (or overwrites) a run's snapshot durably.
Save(ctx context.Context, snap Snapshot) error
// Load returns a run's snapshot, or ok=false if none is stored.
Load(ctx context.Context, runID string) (snap Snapshot, ok bool, err error)
// List returns every stored snapshot, for resume on startup.
List(ctx context.Context) ([]Snapshot, error)
}
StateStore persists run snapshots. Implementations must be safe for concurrent use across distinct run IDs; the provider serializes writes per run (one driver goroutine owns each run's snapshot).
Source Files