Documentation
¶
Overview ¶
Package runtime is the packtrail execution engine: it walks flow graphs, invokes nodes through a pluggable Invoker, and drives fanout/fanin, choice and signal nodes. All progress is durable — every transition is a CAS write to the executions KV and each step is triggered by a durable work message, so a crashed instance's work is picked up by another that acquires the ownership lease.
Index ¶
- type Config
- type Engine
- func (e *Engine) CompleteActivity(ctx context.Context, execID, node string, attempt int, res invoker.Result) error
- func (e *Engine) OnReconcile(fn func(context.Context) error)
- func (e *Engine) Resume(ctx context.Context, execID string) error
- func (e *Engine) Run(ctx context.Context) error
- func (e *Engine) ScheduleFlow(ctx context.Context, name, flowName, cronExpr string, payload json.RawMessage) error
- func (e *Engine) ScheduleReconcile(ctx context.Context, cronExpr string) error
- func (e *Engine) Signal(ctx context.Context, execID, name string, payload json.RawMessage) error
- func (e *Engine) Start(ctx context.Context, flowName string, payload json.RawMessage) (string, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
OwnerID string // unique per instance; defaults to a random id
LeaseTTL time.Duration // ownership lease TTL (default 30s)
AckWait time.Duration // work consumer ack wait (default 60s)
RetryBaseDelay time.Duration // base backoff for task retries (default 1s)
RetryMaxDelay time.Duration // cap on backoff (default 60s)
MaxConcurrency int // max work items processed at once (default 64)
DefaultTimeout time.Duration // task timeout when a node omits one (default 30s)
}
Config tunes engine behaviour. Zero values fall back to sensible defaults.
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine processes executions for a set of flows.
func New ¶
func New( inv invoker.Invoker, st *store.Store, sched *scheduler.Scheduler, flows map[string]*dsl.Flow, cfg Config, ) (*Engine, error)
New builds an engine and precompiles every choice expression in flows. flows maps flow name -> definition. inv executes task/branch nodes; it is typically an *invoker.Registry (optionally wrapped in an *invoker.Cache for idempotency).
func (*Engine) CompleteActivity ¶
func (e *Engine) CompleteActivity(ctx context.Context, execID, node string, attempt int, res invoker.Result) error
CompleteActivity settles an asynchronous activity that an Invoker previously reported as StatusPending. node and attempt identify the dispatched work; res is its outcome (StatusOK to advance, StatusError to fail, StatusRetry/transient to retry per the node policy). It is idempotent and stale-safe: a duplicate or out-of-date completion (the execution already moved on, or a different attempt) is a no-op. It settles either the execution's current task node or a pending fanout branch.
func (*Engine) OnReconcile ¶
OnReconcile registers a callback invoked when a "reconcile" schedule fires. The visibility indexer's Reconcile is the intended hook. It is optional; if unset, fired reconcile schedules are ignored.
func (*Engine) Resume ¶
Resume revives a failed execution, re-running its current node with a fresh retry budget. The durable payload is preserved, so the flow continues from the node that failed (useful when the failure was transient). Only failed executions can be resumed; anything else returns an error. It enqueues durable work, so the engine need not be the same instance — any running engine picks it up (and if none is running yet, it runs when one starts).
func (*Engine) ScheduleFlow ¶
func (e *Engine) ScheduleFlow(ctx context.Context, name, flowName, cronExpr string, payload json.RawMessage) error
ScheduleFlow installs a recurring schedule that starts a new execution of flowName on the given 6-field cron expression ("sec min hour dom mon dow"). name uniquely identifies the schedule; reusing it replaces the schedule.
func (*Engine) ScheduleReconcile ¶
ScheduleReconcile installs a recurring schedule that fires reconciliation on the given 6-field cron expression ("sec min hour dom mon dow"), e.g. "0 */5 * * * *" for every five minutes. Pair it with OnReconcile.
Source Files