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) Cancel(ctx context.Context, execID, reason string) error
- func (e *Engine) CompleteActivity(ctx context.Context, execID, node string, attempt int, res invoker.Result) (err error)
- func (e *Engine) CompleteActivityWithGeneration(ctx context.Context, execID, node string, generation uint64, attempt int, ...) (err error)
- func (e *Engine) OnReconcileActive(fn func(context.Context) error)
- func (e *Engine) OnReconcileFull(fn func(context.Context) error)
- func (e *Engine) ReclaimFiredSchedules(ctx context.Context) (uint64, error)
- func (e *Engine) RedriveStalled(ctx context.Context, execID string, olderThan time.Duration) (bool, error)
- func (e *Engine) Results(ctx context.Context, execID string) (json.RawMessage, 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) ScheduleReconcileActive(ctx context.Context, cronExpr string) error
- func (e *Engine) ScheduleReconcileFull(ctx context.Context, cronExpr string) error
- func (e *Engine) Signal(ctx context.Context, execID, name string, payload json.RawMessage) error
- func (e *Engine) SignalWithID(ctx context.Context, execID, name, idempotencyKey string, ...) error
- func (e *Engine) Start(ctx context.Context, flowName string, payload json.RawMessage) (string, error)
- func (e *Engine) StartWithID(ctx context.Context, execID, 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)
MaxDeliver int // max deliveries of a work item before dead-lettering (default 10)
DrainTimeout time.Duration // max time a graceful shutdown waits for in-flight work (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, signals *signal.Signals, 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). New performs no NATS I/O: sched and signals must already have their streams ensured by the caller (scheduler.EnsureStream / signal.Signals.EnsureStream), which keeps all stream creation in the caller's single-threaded, context- carrying setup phase.
func (*Engine) Cancel ¶ added in v0.1.0
Cancel transitions a running or waiting execution to the terminal cancelled state with the given reason. It is idempotent and stale-safe: cancelling an already-terminal execution (completed/failed/cancelled) — or one that no longer exists — is a no-op. In-flight work is abandoned rather than interrupted: any later work item or async CompleteActivity for this execution finds it non-active and no-ops (process and CompleteActivity both drop non-active executions), so pending retries, fanin evaluations and signal timeouts settle harmlessly. Unlike Resume, a cancelled execution is terminal and cannot be revived.
func (*Engine) CompleteActivity ¶
func (e *Engine) CompleteActivity( ctx context.Context, execID, node string, attempt int, res invoker.Result, ) (err error)
CompleteActivity settles an asynchronous activity using the legacy attempt-only identity. Prefer CompleteActivityWithGeneration when Request.Generation is available so stale completions from earlier node visits cannot settle a later legal cycle or resume.
func (*Engine) CompleteActivityWithGeneration ¶ added in v0.1.0
func (e *Engine) CompleteActivityWithGeneration( ctx context.Context, execID, node string, generation uint64, attempt int, res invoker.Result, ) (err error)
CompleteActivityWithGeneration settles an asynchronous activity only if the completion matches the node visit generation that dispatched it. A zero generation preserves the legacy attempt-only API.
func (*Engine) OnReconcileActive ¶ added in v0.1.0
OnReconcileActive registers the callback fired by the active-set reconcile schedule (the cheap, frequent pass over in-flight executions). Optional; if unset, fired active schedules are ignored.
func (*Engine) OnReconcileFull ¶ added in v0.1.0
OnReconcileFull registers the callback fired by the full reconcile schedule (the authoritative deep scan). Optional; if unset, fired full schedules are ignored.
func (*Engine) ReclaimFiredSchedules ¶ added in v0.1.0
ReclaimFiredSchedules purges already-processed fired-schedule messages from the schedule stream (see scheduler.ReclaimFired), bounding the growth of consumed fire.* messages. It is safe to run on the full-reconcile cadence. Returns how many messages were purged.
func (*Engine) RedriveStalled ¶ added in v0.1.0
func (e *Engine) RedriveStalled(ctx context.Context, execID string, olderThan time.Duration) (bool, error)
RedriveStalled re-drives one execution if it looks stranded: still active, quiet for longer than olderThan (non-positive means the default of 5×AckWait), not inside a scheduled retry backoff, and with no live ownership lease (a held lease means an instance is processing it right now). A stalled running execution gets an advance; a stalled fanin wait gets a join re-evaluation. Signal waits are excluded (their timeout owns them) and so are async task waits (CompleteActivity owns them, and it may legitimately take arbitrarily long).
Every transition is guarded, so a false-positive re-drive is state-safe — at worst it duplicates an invocation within the documented at-least-once contract. It returns whether a work item was enqueued.
This is the operational backstop for the transactional outbox: an execution whose committed follow-on messages were never flushed (crash between the CAS write and the publish) self-heals within one watchdog pass instead of waiting for a manual Resume; the blind re-drives below additionally cover anything with an empty outbox that still looks stranded.
func (*Engine) Results ¶ added in v0.1.0
Results assembles the execution's data-plane view — the same context document invokers and choice rules see. ErrNotFound if the execution does not exist.
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) ScheduleReconcileActive ¶ added in v0.1.0
ScheduleReconcileActive installs the recurring active-set reconcile schedule on the given 6-field cron expression ("sec min hour dom mon dow"), e.g. "0 */5 * * * *". Pair it with OnReconcileActive.
func (*Engine) ScheduleReconcileFull ¶ added in v0.1.0
ScheduleReconcileFull installs the recurring full reconcile schedule on the given 6-field cron expression, e.g. "0 0 * * * *" for hourly. Run it less often than the active schedule; pair it with OnReconcileFull.
func (*Engine) SignalWithID ¶ added in v0.1.0
func (e *Engine) SignalWithID( ctx context.Context, execID, name, idempotencyKey string, payload json.RawMessage, ) error
SignalWithID publishes an external signal with a caller-supplied idempotency key for safe retry after ambiguous publish failures.
func (*Engine) Start ¶
func (e *Engine) Start(ctx context.Context, flowName string, payload json.RawMessage) (string, error)
Start creates a new execution of flowName with the given initial payload, minting a fresh execution id, and enqueues the first step. It returns the id. The payload must be a JSON object (or empty, defaulted to {}); it becomes the `input` field of every invocation context and choice expression. nil is treated as an empty object.
func (*Engine) StartWithID ¶ added in v0.1.0
func (e *Engine) StartWithID(ctx context.Context, execID, flowName string, payload json.RawMessage) (string, error)
StartWithID is an idempotent Start keyed by a caller-supplied execution id (an idempotency key). The first call creates and enqueues the execution; any later call with the same id and the same arguments is a no-op that returns the id unchanged — so a timed-out-then-retried Start produces exactly one execution. First-write wins, and reuse is checked: the id is bound to the first call's flow and byte-identical payload; a repeat naming a different flow or carrying a different payload returns an error rather than silently reporting the existing execution as its own. (The arguments are also validated before the existence check, so a retry that supplies an unknown flow or a non-object payload returns that validation error — retries must replay the same arguments.) The id must match [A-Za-z0-9_-]{1,128} (it becomes a NATS subject token and KV key); supply a stable key such as your domain id (e.g. "order-12345"). Like Start, the payload must be a JSON object (or empty).
Source Files