Documentation
¶
Overview ¶
Package stepfunctions is the AWS Step Functions provider for OmniWorkflow. It runs a Plan on a generic interpreter state machine: a coordinator activity computes readiness/gates/terminal via the shared interpreter core (github.com/plexusone/omniworkflow/interp), and each wave of ready steps and awaiting gates is dispatched to a step activity that an embedded worker polls. Callers never author Amazon States Language.
Index ¶
- Constants
- Variables
- func RenderInterpreterASL(coordinatorActivityARN, stepActivityARN string) (string, error)
- type MemoryTokenStore
- type Options
- type Provider
- func (p *Provider) Cancel(ctx 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(ctx context.Context, id string, signal omniworkflow.Signal) error
- func (p *Provider) Start(ctx context.Context, req omniworkflow.StartRequest) (*omniworkflow.Run, error)
- type ProvisionConfig
- type Provisioned
- type TokenStore
Constants ¶
const DefaultNamePrefix = "omniworkflow"
DefaultNamePrefix is used when ProvisionConfig.NamePrefix is empty. Set a distinct prefix per deployment or tenant so their interpreter state machines and activities do not collide.
Variables ¶
var ErrEmptyActivityARN = errors.New("stepfunctions: activity ARN is empty")
ErrEmptyActivityARN is returned when rendering with a missing activity ARN.
var ErrNoPendingApproval = errors.New("stepfunctions: no gate awaiting this approval")
ErrNoPendingApproval is returned by Signal when no gate is currently awaiting the given approval (the gate has not reached its wait-task yet, or was already approved).
var ErrNoRoleARN = errors.New("stepfunctions: ProvisionConfig.RoleARN is required")
ErrNoRoleARN is returned when ProvisionConfig.RoleARN is empty.
Functions ¶
func RenderInterpreterASL ¶
RenderInterpreterASL substitutes the coordinator and step activity ARNs into the embedded interpreter template and returns the resulting Amazon States Language definition. It validates that the result is well-formed JSON so a malformed template can never reach CreateStateMachine.
Types ¶
type MemoryTokenStore ¶
type MemoryTokenStore struct {
// contains filtered or unexported fields
}
MemoryTokenStore is an in-memory, concurrency-safe TokenStore.
func NewMemoryTokenStore ¶
func NewMemoryTokenStore() *MemoryTokenStore
NewMemoryTokenStore returns an empty in-memory TokenStore.
func (*MemoryTokenStore) Delete ¶
func (m *MemoryTokenStore) Delete(_ context.Context, runID, gateID string) error
Delete removes a gate's task token once it has been released.
type Options ¶
type Options struct {
// TokenStore holds pending gate-approval tokens shared between the
// Provider (Signal) and its Worker. Defaults to an in-process
// MemoryTokenStore.
TokenStore TokenStore
// WorkerName is an optional identifier reported when polling activities.
WorkerName string
}
Options configures a Provider beyond its AWS resources.
type Provider ¶
type Provider struct {
// contains filtered or unexported fields
}
Provider is the AWS Step Functions implementation of omniworkflow.WorkflowProvider. Construct it with New (or NewFromConfig, which provisions first), then obtain a Worker with NewWorker and run it in-process.
func New ¶
func New(client *sfnutil.Client, resources Provisioned, opts Options) *Provider
New builds a Provider over already-provisioned resources (see Provision).
func NewFromConfig ¶
NewFromConfig provisions the interpreter resources (idempotently) and returns a ready Provider.
func (*Provider) NewWorker ¶
func (p *Provider) NewWorker(executor omniworkflow.StepExecutor) omniworkflow.Worker
NewWorker returns a Worker that drives this provider's interpreter: it polls the coordinator activity (advancing the DAG) and the step activity (executing steps and holding gate tokens), dispatching step work to executor. Run it in the caller's process; it shares the provider's TokenStore so Signal can release gates the worker is holding.
func (*Provider) Signal ¶
Signal delivers an approval to a waiting gate. The signal name must be ApproveSignal(gateID); the gate's task token is looked up and released via SendTaskSuccess.
func (*Provider) Start ¶
func (p *Provider) Start(ctx context.Context, req omniworkflow.StartRequest) (*omniworkflow.Run, error)
Start begins executing req.Plan as a Step Functions execution named req.ID. The execution name is the idempotency key: re-starting with the same ID returns the existing run rather than creating a duplicate.
type ProvisionConfig ¶
type ProvisionConfig struct {
// Region is the AWS region; empty uses the SDK's default resolution.
Region string
// RoleARN is the IAM role the state machine assumes to run its Map/Task
// states. Required.
RoleARN string
// NamePrefix prefixes the created resource names. Empty uses
// DefaultNamePrefix.
NamePrefix string
}
ProvisionConfig configures the one-time creation of the interpreter state machine and its activities.
type Provisioned ¶
type Provisioned struct {
StateMachineARN string
CoordinatorActivityARN string
StepActivityARN string
}
Provisioned holds the ARNs of the resources ensured by Provision.
func Provision ¶
func Provision(ctx context.Context, cfg ProvisionConfig) (*Provisioned, error)
Provision ensures the interpreter state machine and its two activities exist, creating them if absent and updating the state machine definition if present. It is idempotent: activities are create-idempotent in SFN, and the state machine is matched by name and updated in place. It returns the resource ARNs the provider's worker and control-plane methods need.
func ProvisionWithClient ¶
func ProvisionWithClient(ctx context.Context, client *sfnutil.Client, cfg ProvisionConfig) (*Provisioned, error)
ProvisionWithClient is Provision with a caller-supplied sfnutil client, for reuse or testing against a local Step Functions endpoint.
type TokenStore ¶
type TokenStore interface {
Put(ctx context.Context, runID, gateID, token string) error
Get(ctx context.Context, runID, gateID string) (token string, ok bool, err error)
Delete(ctx context.Context, runID, gateID string) error
}
TokenStore holds the Step Functions task tokens of gate steps that are waiting for approval, keyed by (runID, gateID). The worker records a token when it picks up a gate wait-task; Signal looks it up to release the gate via SendTaskSuccess.
The default MemoryTokenStore is process-local, which suits a single-process deployment where the worker, control plane, and signal handler share one instance (ActionForge v1). A multi-process or crash-durable deployment should supply a shared implementation (e.g. DynamoDB- or Postgres-backed); the interface is the seam for that.
Source Files