Documentation
¶
Overview ¶
Package engine processes commands: for each partition of the command stream it loads the execution state (cache, or snapshot + tail), decides the events with the pure fold, appends them with optimistic concurrency and acks the command. A conflicting append means another engine wrote first: the state is reloaded and the command decided again. Invalid commands are dead-lettered instead of redelivered forever.
Index ¶
- type Engine
- func (e *Engine) Archive(ctx context.Context, execID string) error
- func (e *Engine) Delete(ctx context.Context, execID string) error
- func (e *Engine) RunCron(ctx context.Context, ready func()) error
- func (e *Engine) RunPartition(ctx context.Context, p int, ready func()) error
- func (e *Engine) RunTrigger(ctx context.Context, def *flow.Flow, i int, tr flow.Trigger, ready func()) error
- type ScheduleSpec
- type TooRecentError
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Engine ¶
type Engine struct {
In *infra.Infra
Loader *loader.Loader
Cache *statecache.Cache
Metrics *metrics.M
Drain time.Duration
// Now is the clock (tests inject one); time enters the fold as data.
Now func() time.Time
// MaxEvents is the largest decision that can be appended (0 =
// eventlog.MaxEvents).
MaxEvents int
// HistoryLimit continues an execution as new once its live log holds
// that many events (0: never).
HistoryLimit int
// Redispatch lifts the quarantine of an execution (wired to the
// dispatcher's replay).
Redispatch func(ctx context.Context, execID string) error
// contains filtered or unexported fields
}
Engine processes commands.
func (*Engine) Archive ¶
Archive moves a terminal execution's events to the archive object store and purges it from the hot resources: event subject, snapshot, blobs, timers. Reads of an archived execution come from the archive (I-17). A flow with an archive retention has the archive deleted that long after.
func (*Engine) Delete ¶ added in v1.0.1
Delete removes a terminal or archived execution and everything derived from it: index entries, blobs, snapshot, pending timers, archived history segments, the archive object and the event log. Nothing is left: the execution reads as never having existed and its id is free again. A running execution, or one that does not exist, is left alone.
Message ids are derived from the execution id and the streams remember them for the deduplication window: a new execution under the same id would have its first jobs and commands dropped as duplicates. So an execution is only deleted once that window has passed since it finished; before, Delete returns a *TooRecentError.
The index goes first and the sources of truth (archive, log) last, so a delete interrupted half-way still finds the execution when it is retried.
func (*Engine) RunCron ¶
RunCron turns fired cron messages into start commands. The execution id is "<schedule>-<stream sequence of the firing>": unique per firing and stable across redelivery, so a firing starts exactly one execution. ready is called once it pulls.
func (*Engine) RunPartition ¶
RunPartition processes the commands of partition p until ctx is done.
func (*Engine) RunTrigger ¶
func (e *Engine) RunTrigger(ctx context.Context, def *flow.Flow, i int, tr flow.Trigger, ready func()) error
RunTrigger starts def whenever a message arrives on one of its triggers. The execution id is "<flow>-<Nats-Msg-Id>" (see triggerExecID), so a duplicate message starts nothing twice and every flow triggered by one message gets its own execution. With a stream the trigger is a durable consumer (at-least-once; without a usable Msg-Id the id is "<flow>-<stream>-<sequence>", stable across redelivery): a message that cannot be started after the redelivery budget is dead-lettered. Otherwise it is a core queue subscription (at-most-once). ready is called once it receives.
type ScheduleSpec ¶
type ScheduleSpec struct {
Flow string `json:"flow"`
Version string `json:"version,omitempty"`
Input json.RawMessage `json:"input,omitempty"`
}
ScheduleSpec is the body of a cron schedule message: what to start on each firing.
type TooRecentError ¶ added in v1.0.1
TooRecentError is returned by Delete for an execution that finished less than the deduplication window ago: it can be deleted from Until.
func (*TooRecentError) Error ¶ added in v1.0.1
func (e *TooRecentError) Error() string
Source Files