Documentation
¶
Overview ¶
Package seq turns a compiled rule into a sequence of steps and walks it.
A rule is a recipe: its steps run in order, and a step is submitted only once the step before it has finished. Nothing here owns a goroutine. Step N+1 is submitted from step N's completion callback, which runs on the pool worker that just finished step N, so a scooter with a hundred half-finished sequences pays for a hundred small structs and nothing else. A step with an after delay is the one exception to "submitted right away": it parks on a scheduler timer between the step before it finishing and its own submit, holding no worker and no goroutine of its own while it waits.
Index ¶
- Constants
- type Commander
- type CompiledStep
- type Hasher
- type Logger
- type Pending
- type PendingStore
- type RuleStats
- type Runner
- func (rn *Runner) Active() int
- func (rn *Runner) CancelMatching(topic string) int
- func (rn *Runner) Fire(s *Sequence, e eventbus.Event)
- func (rn *Runner) RecordError(name string)
- func (rn *Runner) Refused() uint64
- func (rn *Runner) Replay(seqs []*Sequence, window time.Duration) int
- func (rn *Runner) RuleStats(name string) RuleStats
- func (rn *Runner) Stop()
- type Sequence
Constants ¶
const PendingHash = "extensions:pending"
PendingHash is the datastore hash durable steps are recorded in. One field per waiting step, keyed by the run's id.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Commander ¶
type Commander interface {
HSet(key, field string, value any) error
HGetAll(key string) (map[string]string, error)
Do(cmd string, args ...any) (any, error)
}
Commander is what NewClientHasher needs from a datastore client: the two hash operations the client has methods for, plus the raw command interface, because it has no HDel of its own.
type CompiledStep ¶
type CompiledStep struct {
Action action.Action
When *vm.Program
After time.Duration
Durable bool
Fingerprint string
}
CompiledStep is one step ready to submit. When is nil for a step with no condition of its own. After is zero for a step that runs as soon as it is reached; a step with After greater than zero parks on a timer for that long before it is submitted. Durable, which only a step with After can be, puts that pending fire in the datastore so a restart during the wait can pick it up again. Fingerprint identifies what the step was configured to do, so a record written for it can be told apart from a record written for whatever now sits at the same index.
type Hasher ¶
type Hasher interface {
HSet(key, field string, value any) error
HGetAll(key string) (map[string]string, error)
HDel(key string, fields ...string) error
}
Hasher is the datastore surface the pending store needs. Depending on three operations rather than a whole client keeps the store testable, and keeps the package free of the datastore library.
func NewClientHasher ¶
NewClientHasher adapts a datastore client to Hasher.
type Pending ¶
type Pending struct {
ID string `json:"id"`
Rule string `json:"rule"`
Source string `json:"source"`
Step int `json:"step"`
Iter int `json:"iter"`
FireAt int64 `json:"fire-at"` // unix milliseconds
Fingerprint string `json:"fingerprint"`
Event eventbus.Event `json:"event"`
}
Pending is one step waiting out its after delay, written down so a service restart during the wait does not lose it. Rule and Step are what replay needs to find the step again; Iter is the repeat pass the run was on, so a replayed run finishes the passes it had left. Event is the trigger, because a step's when is evaluated against whatever fired its own run.
FireAt is when the step is due, not when it was recorded: replay works out the remaining delay from it, and drops a record too far past due to be worth running.
Fingerprint is what the step was configured to do when the record was written. A user is free to edit their rule file while the service is down, and Step is only an index: without the fingerprint, moving a step or rewriting it would have the record fire whatever ended up at that index instead.
type PendingStore ¶
type PendingStore struct {
// contains filtered or unexported fields
}
PendingStore reads and writes the pending records. There is no sweep and no expiry: a record is written when a durable step is scheduled and removed when it resolves, so a scooter with nothing running writes nothing at all.
func NewPendingStore ¶
func NewPendingStore(c Hasher, log Logger) *PendingStore
NewPendingStore returns a store over the standard hash.
func (*PendingStore) Drop ¶
func (s *PendingStore) Drop(id string) error
Drop removes a record. Removing one that is not there is not an error: the step that fires and the cancel that arrives at the same moment both aim at the same field, and only one of them can be first.
func (*PendingStore) Load ¶
func (s *PendingStore) Load() ([]Pending, error)
Load reads every record. A value that will not decode is logged and removed rather than returned or fatal: whatever wrote it is not going to write it better on the next boot, and leaving it in place turns one bad record into one log line per start forever.
func (*PendingStore) Put ¶
func (s *PendingStore) Put(p Pending) error
Put writes a record. It overwrites whatever the run's id held before, which is what a run reaching its second durable step wants: one run has at most one step waiting at a time.
type RuleStats ¶
RuleStats counts actual run starts (including replay) and evaluation/action errors. It does not count triggers suppressed by cooldown or concurrency.
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner walks sequences. In-flight runs are registered under their rule name, which is the handle anything acting on a whole rule needs.
func NewRunner ¶
NewRunner returns a Runner that submits steps to pool and parks a step's after delay on sch. store records the steps that are waiting, so a restart mid-wait can pick them up again; a nil store turns that off and the runner keeps everything in memory.
func (*Runner) Active ¶
Active is how many runs are part-way through their steps, including a run currently parked on a timer waiting out an after or the gap between two repeat passes. A trigger sitting in a queue-policy rule's backlog is not active: it has not started.
func (*Runner) CancelMatching ¶
CancelMatching drops every live run whose rule names topic in cancel-on, together with anything that rule is holding queued behind them, and returns how many runs it cancelled. Queued triggers are not runs and are not in the count; keeping them would defeat the cancel, since the backlog is usually more of exactly what is being cancelled.
What a cancel stops is everything the run has not handed over yet: a pending timer is dropped and no further step is submitted. A step already given to the pool still runs, whether a worker has picked it up or it is waiting its turn in the queue behind other work. The pool owns an accepted job either way and there is no way to take one back, so "cancel" here means the tail, not the step already on its way.
func (*Runner) Fire ¶
Fire starts a run of s triggered by e, or does whatever the rule's concurrency policy says to do about a run that is already live. It returns as soon as the first step is queued, so the caller, which is the goroutine reading the bus, is never held for the length of a sequence.
func (*Runner) RecordError ¶
func (*Runner) Refused ¶
Refused counts the triggers turned away by a queue-policy rule's bound since start. It is cumulative, and a number that climbs is the visible end of a rule whose sequence cannot keep up with what triggers it.
func (*Runner) Replay ¶
Replay picks up the steps that were waiting when the service last went down and returns how many it resumed. Call it once, after the rules are compiled and before the bus subscription starts: a replayed step must not race a live re-fire of the same rule.
A record is dropped, with a line saying why, when the rule it names is no longer loaded, when the rule no longer has the step it indexes, when the step at that index is not the step the record was written for any more, when it is more than window past due, or when it is dated further ahead than the step it names could ever wait. The middle two are one situation: a user is free to edit their rule files while the service is down, and a record whose step moved or was rewritten must not fire whatever took its place. The last two are the limits on the clock in both directions. Anything left is resumed on the run it belonged to, keeping its id, so the record still covers it until it fires.
func (*Runner) Stop ¶
func (rn *Runner) Stop()
Stop refuses new runs, throws away anything queued behind them, and abandons the in-flight ones, cancelling any pending tail along the way: a run parked on a timer is still active, and letting that timer fire after the runner considers itself stopped would submit a step into a pool that may already be gone. A step already on a worker still runs to the end; its callback then finds the run ended and stops there rather than queueing more work into a pool that is going away.
Ending goes through endLocked like every other cancel, so there is one definition of what ending a run means rather than two that can drift.
The pending records endLocked hands back are deliberately dropped on the floor rather than removed from the datastore. Every other caller ends a run because it is over; Stop ends one because this process is. A step still waiting out its delay has not run, and leaving its record is what lets the next start run it. Removing them here would break the one case durability exists for.
type Sequence ¶
type Sequence struct {
Rule *rules.Rule
Steps []CompiledStep
}
Sequence is a rule with every step built into a runnable action.
Source Files