Documentation
¶
Overview ¶
Package dispatch is the projection that turns events into side effects. It consumes each partition of the events stream in order, keeps the folded state of every execution it sees, and for each event:
NodeScheduled publishes a job to <ns>.work.<kind> (or a cached result)
TimerScheduled installs a JetStream schedule that fires a timer command
ChildStarted starts the child execution
terminal events notifies the parent, cancels children, purges timers and
schedules the archive
NodeCompleted stores cacheable results
and updates the visibility index. Effects are never written together with the events (they live in other streams): every publish is idempotent instead (deduplication ids, rollup subjects, stale checks in the fold), so redelivering an event after a crash is harmless.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func CacheKey ¶
CacheKey derives the result-cache key of a task: the flow version, the node and everything the worker can observe of the state (not visit counters, which change on every loop). Keys are hashes, so they are always valid KV keys and never collide across prefixes.
func Deterministic ¶
Deterministic reports whether err can never be fixed by retrying the same event: an undecodable or unfoldable event, an unknown flow, malformed data, a request the server rejects as invalid, a missing blob.
Types ¶
type Dispatcher ¶
type Dispatcher struct {
In *infra.Infra
Loader *loader.Loader
Cache *statecache.Cache
Metrics *metrics.M
Drain time.Duration
// contains filtered or unexported fields
}
Dispatcher projects events into effects.
func (*Dispatcher) Replay ¶
func (d *Dispatcher) Replay(ctx context.Context, execID string) error
Replay lifts the quarantine of execID: it runs the effects of every event from the one that failed (which must now succeed), then releases the mark on every dispatcher and catches up on events skipped meanwhile. On error the execution stays quarantined.
func (*Dispatcher) Run ¶
func (d *Dispatcher) Run(ctx context.Context, p int, ready func()) error
Run consumes partition p until ctx is done. ready is called once it pulls. The process must be following control notices (WatchControl) first.
func (*Dispatcher) WatchControl ¶
func (d *Dispatcher) WatchControl() (stop func(), err error)
WatchControl follows quarantine/release broadcasts, so every dispatcher process skips (or resumes) an execution at once, until stop is called. Call it before Run: the subscriptions then precede each partition's read of the quarantine list, so no broadcast falls between the two.
Source Files