700 lines
24 KiB
Go
700 lines
24 KiB
Go
package conditions
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/novox/mesh-controller/internal/outward"
|
|
)
|
|
|
|
// Backend is where the open conditions are kept: one value per key, written by compare-and-set.
|
|
type Backend interface {
|
|
// Get is one key's value and revision; false when it holds none.
|
|
Get(ctx context.Context, key string) (Entry, bool, error)
|
|
// Create writes a key that holds nothing, and fails with ErrMoved when it holds something.
|
|
Create(ctx context.Context, key string, value []byte) error
|
|
// Update writes a key at the revision it was read at, and fails with ErrMoved when it moved.
|
|
Update(ctx context.Context, key string, value []byte, revision uint64) error
|
|
// Delete removes a key at the revision it was read at, and fails with ErrMoved when it moved.
|
|
Delete(ctx context.Context, key string, revision uint64) error
|
|
// All is every key's value. An error is an error: never an empty store (ADR 0227 rule 4).
|
|
All(ctx context.Context) (map[string]Entry, error)
|
|
}
|
|
|
|
// Entry is one key's value, at a revision.
|
|
type Entry struct {
|
|
Value []byte
|
|
Revision uint64
|
|
}
|
|
|
|
// ErrMoved is a compare-and-set that lost: somebody wrote the key since it was read.
|
|
var ErrMoved = errors.New("the condition was written by somebody else since it was read")
|
|
|
|
// History keeps every transition (to-be 45 §2): appended, read back from a moment.
|
|
type History interface {
|
|
Append(ctx context.Context, e Event) error
|
|
// Since is every transition from a moment, oldest first.
|
|
Since(ctx context.Context, since time.Time) ([]Event, error)
|
|
}
|
|
|
|
// Teller says a transition on the bus, as the mesh-controller seat's event. The link's bus is one.
|
|
type Teller interface {
|
|
PublishSeatEvent(ctx context.Context, seat, event string, body []byte) error
|
|
}
|
|
|
|
// Seat is the role the events are said under (novox/hq ADR 0134): the control plane's.
|
|
const Seat = "mesh-controller"
|
|
|
|
// Keeper raises, observes, silences and clears conditions, and says each transition.
|
|
type Keeper struct {
|
|
store Backend
|
|
history History
|
|
teller Teller
|
|
now func() time.Time
|
|
say func(format string, args ...any)
|
|
changed func()
|
|
epoch func() (uint64, error)
|
|
|
|
mu sync.Mutex
|
|
// cleared is when each recently cleared condition cleared and how often it had been raised, so
|
|
// one raised again within ReopenWithin is the same one again.
|
|
cleared map[string]clearing
|
|
|
|
// out is the transitions still to be said and kept, in order: said by one goroutine, so a
|
|
// condition's events arrive in the order they happened, and offered again while the bus is away.
|
|
out chan Event
|
|
drained chan struct{}
|
|
closing sync.Once
|
|
// Unsaid counts the transitions given up on, for the self-check to say.
|
|
unsaid int
|
|
// reworded are the keys whose summary was said in words for the operator's channel, said once each.
|
|
reworded map[string]bool
|
|
}
|
|
|
|
type clearing struct {
|
|
at time.Time
|
|
count int
|
|
silenced *Silence
|
|
// tried is what healers tried before it cleared: a reopening is the same fault, and what was
|
|
// tried on it is still what was tried.
|
|
tried []Attempt
|
|
}
|
|
|
|
// Options are what a Keeper is made with.
|
|
type Options struct {
|
|
Store Backend
|
|
History History
|
|
// Teller says the transitions; nil says nothing (a test, or a command run with no bus to say on).
|
|
Teller Teller
|
|
Now func() time.Time
|
|
// Say is where a transition that could not be said or kept is said instead.
|
|
Say func(format string, args ...any)
|
|
// Changed is told of every transition, at once — for `status`, which leads with what is open.
|
|
Changed func()
|
|
// Epoch is the controller lease this keeper writes under (to-be 45 §6): every write carries its
|
|
// epoch, and none is made while it is not held. Nil writes with no epoch (a test, the memory store).
|
|
Epoch func() (uint64, error)
|
|
}
|
|
|
|
// TellFor is how long one transition is offered to the bus before it is said lost.
|
|
var TellFor = 10 * time.Minute
|
|
|
|
// NewKeeper is a keeper over a store. It reads what cleared lately from the history, so a condition
|
|
// that cleared just before this controller started and is raised again now is a reopening.
|
|
func NewKeeper(ctx context.Context, o Options) *Keeper {
|
|
k := &Keeper{store: o.Store, history: o.History, teller: o.Teller, now: o.Now, say: o.Say, changed: o.Changed,
|
|
epoch: o.Epoch,
|
|
cleared: map[string]clearing{}, reworded: map[string]bool{}, out: make(chan Event, 1024), drained: make(chan struct{})}
|
|
if k.now == nil {
|
|
k.now = time.Now
|
|
}
|
|
if k.say == nil {
|
|
k.say = func(string, ...any) {}
|
|
}
|
|
if k.history != nil {
|
|
if recent, err := k.history.Since(ctx, k.now().Add(-ReopenWithin)); err == nil {
|
|
for _, e := range recent {
|
|
if e.Change == ChangeCleared {
|
|
k.cleared[e.Key] = clearing{at: e.At, count: e.Condition.Count, silenced: e.Condition.Silenced,
|
|
tried: e.Condition.Tried}
|
|
}
|
|
}
|
|
} else {
|
|
k.say("what cleared lately could not be read from the condition history, so a condition "+
|
|
"raised again now is said as new rather than reopened: %v", err)
|
|
}
|
|
}
|
|
go k.telling()
|
|
return k
|
|
}
|
|
|
|
// Close says what is still to be said, waiting at most until ctx ends.
|
|
func (k *Keeper) Close(ctx context.Context) {
|
|
k.closing.Do(func() { close(k.out) })
|
|
select {
|
|
case <-k.drained:
|
|
case <-ctx.Done():
|
|
k.say("%d condition transition(s) were not yet said when this process ended", len(k.out))
|
|
}
|
|
}
|
|
|
|
// Unsaid is how many transitions were given up on since this keeper started.
|
|
func (k *Keeper) Unsaid() int {
|
|
k.mu.Lock()
|
|
defer k.mu.Unlock()
|
|
return k.unsaid
|
|
}
|
|
|
|
// stamp is the gate every write passes: the epoch it carries, set on the condition, or why this
|
|
// keeper may not write — a controller that lost the lease raises, observes and clears nothing.
|
|
func (k *Keeper) stamp(c *Condition) error {
|
|
if k.epoch == nil {
|
|
return nil
|
|
}
|
|
epoch, err := k.epoch()
|
|
if err != nil {
|
|
return fmt.Errorf("the condition %s is not written: %w", c.Key, err)
|
|
}
|
|
c.Epoch = epoch
|
|
return nil
|
|
}
|
|
|
|
// tries bounds one compare-and-set: two writers rarely race more than once.
|
|
const tries = 8
|
|
|
|
// Observe records one observation: raises the condition if it is not open, and otherwise adds the
|
|
// evidence. Says a raising, a reopening, and a change of severity or resolver; an observation that
|
|
// changes neither is written and said nowhere.
|
|
func (k *Keeper) Observe(ctx context.Context, o Observation) (Condition, error) {
|
|
if err := o.check(); err != nil {
|
|
return Condition{}, err
|
|
}
|
|
o = k.sayable(o)
|
|
o = plainly(o)
|
|
key := o.Key()
|
|
for i := 0; i < tries; i++ {
|
|
now := k.now().UTC()
|
|
said := o.Said
|
|
if said == "" {
|
|
said = o.Summary
|
|
}
|
|
entry, found, err := k.store.Get(ctx, key)
|
|
if err != nil {
|
|
return Condition{}, fmt.Errorf("reading the condition %s: %w", key, err)
|
|
}
|
|
if !found {
|
|
c := Condition{Key: key, Kind: o.Kind, Subject: Subject{Scope: o.Scope, ID: o.ID, Machine: o.Machine, Also: o.Also},
|
|
Severity: o.Severity, Summary: o.Summary, Headline: o.Headline, Explanation: o.Explanation,
|
|
Resolved: o.Resolved, Needs: o.Needs, Actions: o.Actions, Evidence: []Evidence{{At: now, Said: said}},
|
|
Source: o.Source, Raised: now, LastObserved: now, Observations: 1, Count: 1,
|
|
Resolver: orSelf(o.Resolver)}
|
|
change := ChangeRaised
|
|
k.mu.Lock()
|
|
if before, ok := k.cleared[key]; ok && now.Sub(before.at) <= ReopenWithin {
|
|
c.Count, change = before.count+1, ChangeReopened
|
|
// A silence a person gave the condition before it cleared still holds: they said
|
|
// they knew, and the same fault again ten minutes later is what they knew about.
|
|
if before.silenced != nil && now.Before(before.silenced.Until) {
|
|
c.Silenced = before.silenced
|
|
}
|
|
c.Tried = before.tried
|
|
}
|
|
k.mu.Unlock()
|
|
if err := k.stamp(&c); err != nil {
|
|
return Condition{}, err
|
|
}
|
|
body, err := json.Marshal(c)
|
|
if err != nil {
|
|
return Condition{}, err
|
|
}
|
|
if err := k.store.Create(ctx, key, body); errors.Is(err, ErrMoved) {
|
|
continue
|
|
} else if err != nil {
|
|
return Condition{}, fmt.Errorf("raising the condition %s: %w", key, err)
|
|
}
|
|
k.mu.Lock()
|
|
delete(k.cleared, key)
|
|
k.mu.Unlock()
|
|
k.tell(Event{Condition: c, At: now, Change: change})
|
|
return c, nil
|
|
}
|
|
var c Condition
|
|
if err := json.Unmarshal(entry.Value, &c); err != nil {
|
|
return Condition{}, fmt.Errorf("the condition %s on the bus cannot be read: %w", key, err)
|
|
}
|
|
var changes []Event
|
|
// A healer's budget spent made it urgent; the watchdog seeing it again does not undo that.
|
|
if o.Severity != c.Severity && !(c.Escalated() && o.Severity != Urgent) {
|
|
changes = append(changes, Event{Change: ChangeSeverity, Was: string(c.Severity)})
|
|
c.Severity = o.Severity
|
|
}
|
|
if r := orSelf(o.Resolver); o.Resolver != "" && r != c.Resolver {
|
|
changes = append(changes, Event{Change: ChangeResolver, Was: c.Resolver})
|
|
c.Resolver = r
|
|
}
|
|
// The kind as the source says it now: a source that gave the same key a kind of its own since
|
|
// (a probe's finding split out for a healer) is read by that kind from its next observation.
|
|
c.Kind, c.Summary, c.Source, c.LastObserved = o.Kind, o.Summary, o.Source, now
|
|
wasNeeds, wasActions := c.Needs, c.Actions
|
|
c.Headline, c.Explanation, c.Resolved, c.Needs, c.Actions = o.Headline, o.Explanation, o.Resolved, o.Needs, o.Actions
|
|
escalatedWords(&c)
|
|
if c.Needs != wasNeeds || !sameActions(c.Actions, wasActions) {
|
|
changes = append(changes, Event{Change: ChangeWords, Was: wasNeeds})
|
|
}
|
|
if o.Machine != "" {
|
|
c.Subject.Machine = o.Machine
|
|
}
|
|
if len(o.Also) > 0 {
|
|
c.Subject.Also = o.Also
|
|
}
|
|
c.Observations++
|
|
c.Evidence = append([]Evidence{{At: now, Said: said}}, c.Evidence...)
|
|
if len(c.Evidence) > KeptEvidence {
|
|
c.Evidence = c.Evidence[:KeptEvidence]
|
|
}
|
|
if err := k.stamp(&c); err != nil {
|
|
return Condition{}, err
|
|
}
|
|
body, err := json.Marshal(c)
|
|
if err != nil {
|
|
return Condition{}, err
|
|
}
|
|
if err := k.store.Update(ctx, key, body, entry.Revision); errors.Is(err, ErrMoved) {
|
|
continue
|
|
} else if err != nil {
|
|
return Condition{}, fmt.Errorf("observing the condition %s: %w", key, err)
|
|
}
|
|
for _, e := range changes {
|
|
e.At, e.Condition = now, c
|
|
k.tell(e)
|
|
}
|
|
return c, nil
|
|
}
|
|
return Condition{}, fmt.Errorf("the condition %s kept moving under this write; %d tries", key, tries)
|
|
}
|
|
|
|
// Clear removes a condition an observation says is resolved, and says so. False when none was open.
|
|
func (k *Keeper) Clear(ctx context.Context, key, why string) (bool, error) {
|
|
return k.ClearSaying(ctx, key, why, "")
|
|
}
|
|
|
|
// ClearSaying is Clear with the line the operator reads when it clears, in place of the condition's own
|
|
// resolved line: for a condition that ends because it became another (novox/hq ADR 0254: a module not
|
|
// working that now waits for a new login is not "working again"). Empty keeps the condition's own.
|
|
func (k *Keeper) ClearSaying(ctx context.Context, key, why, resolved string) (bool, error) {
|
|
for i := 0; i < tries; i++ {
|
|
entry, found, err := k.store.Get(ctx, key)
|
|
if err != nil {
|
|
return false, fmt.Errorf("reading the condition %s: %w", key, err)
|
|
}
|
|
if !found {
|
|
return false, nil
|
|
}
|
|
var c Condition
|
|
if err := json.Unmarshal(entry.Value, &c); err != nil {
|
|
// Unreadable is not resolved: kept, and said, rather than removed unread.
|
|
return false, fmt.Errorf("the condition %s on the bus cannot be read, so it is not cleared: %w", key, err)
|
|
}
|
|
if err := k.stamp(&c); err != nil {
|
|
return false, err
|
|
}
|
|
if resolved != "" {
|
|
c.Resolved = resolved
|
|
}
|
|
if err := k.store.Delete(ctx, key, entry.Revision); errors.Is(err, ErrMoved) {
|
|
continue
|
|
} else if err != nil {
|
|
return false, fmt.Errorf("clearing the condition %s: %w", key, err)
|
|
}
|
|
now := k.now().UTC()
|
|
k.mu.Lock()
|
|
k.cleared[key] = clearing{at: now, count: c.Count, silenced: c.Silenced, tried: c.Tried}
|
|
k.mu.Unlock()
|
|
k.tell(Event{Condition: c, At: now, Change: ChangeCleared, Why: why, Cleared: &now})
|
|
return true, nil
|
|
}
|
|
return false, fmt.Errorf("the condition %s kept moving under this clearing; %d tries", key, tries)
|
|
}
|
|
|
|
// Tried records a healer's attempt on an open condition (to-be 45 §7) and makes the healer its
|
|
// resolver: said as `condition-changed` when the resolver changes, kept in `tried` either way. **It
|
|
// never clears the condition**: a repair that worked is seen by the observation that raised it, which
|
|
// clears it — a healer marking its own work done would be a second opinion of the fact. False when no
|
|
// condition is open under the key: it cleared meanwhile, and there is nothing to record against.
|
|
func (k *Keeper) Tried(ctx context.Context, key string, a Attempt, resolver string) (Condition, bool, error) {
|
|
return k.amend(ctx, key, func(c *Condition, now time.Time) []Event {
|
|
c.Tried = appendAttempt(c.Tried, a, now)
|
|
var changes []Event
|
|
if resolver != "" && resolver != c.Resolver && !c.Escalated() {
|
|
changes = append(changes, Event{Change: ChangeResolver, Was: c.Resolver, Why: a.Outcome})
|
|
c.Resolver = resolver
|
|
}
|
|
return changes
|
|
})
|
|
}
|
|
|
|
// Escalate is a healer's budget spent, or a repair it may not make (to-be 45 §2, §7): the attempt that
|
|
// says so is kept in `tried`, the resolver becomes the operator and the severity urgent, each said as
|
|
// `condition-changed`. Observation still clears it when the fault goes; nothing else does.
|
|
func (k *Keeper) Escalate(ctx context.Context, key string, a Attempt) (Condition, bool, error) {
|
|
return k.amend(ctx, key, func(c *Condition, now time.Time) []Event {
|
|
c.Tried = appendAttempt(c.Tried, a, now)
|
|
var changes []Event
|
|
if c.Severity != Urgent {
|
|
changes = append(changes, Event{Change: ChangeSeverity, Was: string(c.Severity), Why: a.Outcome})
|
|
c.Severity = Urgent
|
|
}
|
|
if c.Resolver != ResolverOperator {
|
|
changes = append(changes, Event{Change: ChangeResolver, Was: c.Resolver, Why: a.Outcome})
|
|
c.Resolver = ResolverOperator
|
|
}
|
|
escalatedWords(c)
|
|
return changes
|
|
})
|
|
}
|
|
|
|
// appendAttempt adds an attempt, stamped when it has no time, keeping the newest KeptAttempts.
|
|
func appendAttempt(tried []Attempt, a Attempt, now time.Time) []Attempt {
|
|
if a.At.IsZero() {
|
|
a.At = now
|
|
}
|
|
tried = append(tried, a)
|
|
if len(tried) > KeptAttempts {
|
|
tried = tried[len(tried)-KeptAttempts:]
|
|
}
|
|
return tried
|
|
}
|
|
|
|
// amend changes one open condition by compare-and-set and says what changed. False when none is open.
|
|
func (k *Keeper) amend(ctx context.Context, key string, change func(*Condition, time.Time) []Event) (Condition, bool, error) {
|
|
for i := 0; i < tries; i++ {
|
|
entry, found, err := k.store.Get(ctx, key)
|
|
if err != nil {
|
|
return Condition{}, false, fmt.Errorf("reading the condition %s: %w", key, err)
|
|
}
|
|
if !found {
|
|
return Condition{}, false, nil
|
|
}
|
|
var c Condition
|
|
if err := json.Unmarshal(entry.Value, &c); err != nil {
|
|
return Condition{}, false, fmt.Errorf("the condition %s on the bus cannot be read: %w", key, err)
|
|
}
|
|
now := k.now().UTC()
|
|
changes := change(&c, now)
|
|
if err := k.stamp(&c); err != nil {
|
|
return Condition{}, false, err
|
|
}
|
|
body, err := json.Marshal(c)
|
|
if err != nil {
|
|
return Condition{}, false, err
|
|
}
|
|
if err := k.store.Update(ctx, key, body, entry.Revision); errors.Is(err, ErrMoved) {
|
|
continue
|
|
} else if err != nil {
|
|
return Condition{}, false, fmt.Errorf("writing the condition %s: %w", key, err)
|
|
}
|
|
for _, e := range changes {
|
|
e.At, e.Condition = now, c
|
|
k.tell(e)
|
|
}
|
|
return c, true, nil
|
|
}
|
|
return Condition{}, false, fmt.Errorf("the condition %s kept moving under this write; %d tries", key, tries)
|
|
}
|
|
|
|
// Reconcile is one source's whole observation: every condition it observes is observed, and every
|
|
// condition it raised before and no longer observes is cleared — the observation says it is
|
|
// resolved. A source that could not observe must not call this: an empty observation clears all it
|
|
// raised, which is exactly the fault of saying "none" for "I could not tell" (ADR 0227 rule 4).
|
|
func (k *Keeper) Reconcile(ctx context.Context, source string, observed []Observation) error {
|
|
all, err := k.Open(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
seen := map[string]bool{}
|
|
var problems []string
|
|
for _, o := range observed {
|
|
o.Source = source
|
|
seen[o.Key()] = true
|
|
if _, err := k.Observe(ctx, o); err != nil {
|
|
problems = append(problems, err.Error())
|
|
}
|
|
}
|
|
for _, c := range all {
|
|
if c.Source != source || seen[c.Key] {
|
|
continue
|
|
}
|
|
if _, err := k.Clear(ctx, c.Key, source+" no longer observes it"); err != nil {
|
|
problems = append(problems, err.Error())
|
|
}
|
|
}
|
|
if len(problems) > 0 {
|
|
return errors.New(strings.Join(problems, "; "))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Silence stops a condition's messages for a while, with a reason, by somebody (to-be 45 §2). The
|
|
// condition stays open and `status` still says it; recording the act in the hand-act log is the
|
|
// caller's, which knows who acted.
|
|
func (k *Keeper) Silence(ctx context.Context, key string, d time.Duration, by, why string) (Condition, error) {
|
|
if strings.TrimSpace(why) == "" {
|
|
return Condition{}, errors.New("a silence says why: --why <text>")
|
|
}
|
|
if d <= 0 || d > MaxSilence {
|
|
return Condition{}, fmt.Errorf("a condition is silenced for a while, at most %s — not %s", MaxSilence, d)
|
|
}
|
|
for i := 0; i < tries; i++ {
|
|
entry, found, err := k.store.Get(ctx, key)
|
|
if err != nil {
|
|
return Condition{}, fmt.Errorf("reading the condition %s: %w", key, err)
|
|
}
|
|
if !found {
|
|
return Condition{}, fmt.Errorf("no condition %s is open — `conditions` lists them", key)
|
|
}
|
|
var c Condition
|
|
if err := json.Unmarshal(entry.Value, &c); err != nil {
|
|
return Condition{}, fmt.Errorf("the condition %s on the bus cannot be read: %w", key, err)
|
|
}
|
|
now := k.now().UTC()
|
|
c.Silenced = &Silence{Until: now.Add(d), By: by, Why: strings.TrimSpace(why), Since: now}
|
|
if err := k.stamp(&c); err != nil {
|
|
return Condition{}, err
|
|
}
|
|
body, err := json.Marshal(c)
|
|
if err != nil {
|
|
return Condition{}, err
|
|
}
|
|
if err := k.store.Update(ctx, key, body, entry.Revision); errors.Is(err, ErrMoved) {
|
|
continue
|
|
} else if err != nil {
|
|
return Condition{}, fmt.Errorf("silencing the condition %s: %w", key, err)
|
|
}
|
|
k.tell(Event{Condition: c, At: now, Change: ChangeSilenced, Why: c.Silenced.Why})
|
|
return c, nil
|
|
}
|
|
return Condition{}, fmt.Errorf("the condition %s kept moving under this silence; %d tries", key, tries)
|
|
}
|
|
|
|
// EndSilences ends every silence that has run out, and says each: the condition is still open, and
|
|
// its messages start again.
|
|
func (k *Keeper) EndSilences(ctx context.Context) error {
|
|
all, err := k.Open(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
now := k.now().UTC()
|
|
for _, c := range all {
|
|
if c.Silenced == nil || now.Before(c.Silenced.Until) {
|
|
continue
|
|
}
|
|
for i := 0; i < tries; i++ {
|
|
entry, found, err := k.store.Get(ctx, c.Key)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !found {
|
|
break
|
|
}
|
|
var held Condition
|
|
if err := json.Unmarshal(entry.Value, &held); err != nil {
|
|
return fmt.Errorf("the condition %s on the bus cannot be read: %w", c.Key, err)
|
|
}
|
|
if held.Silenced == nil || now.Before(held.Silenced.Until) {
|
|
break
|
|
}
|
|
was := held.Silenced.Why
|
|
held.Silenced = nil
|
|
if err := k.stamp(&held); err != nil {
|
|
return err
|
|
}
|
|
body, err := json.Marshal(held)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := k.store.Update(ctx, c.Key, body, entry.Revision); errors.Is(err, ErrMoved) {
|
|
continue
|
|
} else if err != nil {
|
|
return err
|
|
}
|
|
k.tell(Event{Condition: held, At: now, Change: ChangeUnsilenced, Why: was})
|
|
break
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Open is every open condition, urgent first and then oldest first.
|
|
func (k *Keeper) Open(ctx context.Context) ([]Condition, error) {
|
|
return Read(ctx, k.store)
|
|
}
|
|
|
|
// Get is one open condition.
|
|
func (k *Keeper) Get(ctx context.Context, key string) (Condition, bool, error) {
|
|
return ReadOne(ctx, k.store, key)
|
|
}
|
|
|
|
// HistorySince is every transition from a moment, oldest first.
|
|
func (k *Keeper) HistorySince(ctx context.Context, since time.Time) ([]Event, error) {
|
|
if k.history == nil {
|
|
return nil, errors.New("this keeper has no history to read")
|
|
}
|
|
return k.history.Since(ctx, since)
|
|
}
|
|
|
|
// Read is every open condition in a store, in the order status says them. A value that cannot be
|
|
// read is an error naming its key, never a condition left out (ADR 0227 rule 4).
|
|
func Read(ctx context.Context, store Backend) ([]Condition, error) {
|
|
all, err := store.All(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the open conditions cannot be read: %w", err)
|
|
}
|
|
out := make([]Condition, 0, len(all))
|
|
for key, e := range all {
|
|
var c Condition
|
|
if err := json.Unmarshal(e.Value, &c); err != nil {
|
|
return nil, fmt.Errorf("the condition %s cannot be read: %w", key, err)
|
|
}
|
|
out = append(out, c)
|
|
}
|
|
Order(out)
|
|
return out, nil
|
|
}
|
|
|
|
// ReadOne is one open condition from a store.
|
|
func ReadOne(ctx context.Context, store Backend, key string) (Condition, bool, error) {
|
|
e, found, err := store.Get(ctx, key)
|
|
if err != nil || !found {
|
|
return Condition{}, found, err
|
|
}
|
|
var c Condition
|
|
if err := json.Unmarshal(e.Value, &c); err != nil {
|
|
return Condition{}, false, fmt.Errorf("the condition %s cannot be read: %w", key, err)
|
|
}
|
|
return c, true, nil
|
|
}
|
|
|
|
// tell queues a transition to be kept and said. Never blocks the caller for long: a queue that is
|
|
// full is a bus away for a long time, and the transition is said lost rather than holding a watchdog.
|
|
func (k *Keeper) tell(e Event) {
|
|
e.Event = eventFor(e.Change)
|
|
e.Show = e.Condition.Show()
|
|
if k.changed != nil {
|
|
k.changed()
|
|
}
|
|
defer func() {
|
|
// A keeper closed while a write was in flight: said, not a panic.
|
|
if recover() != nil {
|
|
k.lost(e, errors.New("the keeper was closed"))
|
|
}
|
|
}()
|
|
select {
|
|
case k.out <- e:
|
|
default:
|
|
k.lost(e, errors.New("too many transitions are waiting to be said"))
|
|
}
|
|
}
|
|
|
|
func (k *Keeper) lost(e Event, err error) {
|
|
k.mu.Lock()
|
|
k.unsaid++
|
|
k.mu.Unlock()
|
|
k.say("the condition %s was %s and that could NOT be said or kept: %v", e.Key, e.Change, err)
|
|
}
|
|
|
|
// telling keeps and says every transition in order, offering each again while the bus is away.
|
|
func (k *Keeper) telling() {
|
|
defer close(k.drained)
|
|
for e := range k.out {
|
|
body, err := json.Marshal(e)
|
|
if err != nil {
|
|
k.lost(e, err)
|
|
continue
|
|
}
|
|
deadline := time.Now().Add(TellFor)
|
|
wait := 200 * time.Millisecond
|
|
kept, said := k.history == nil, k.teller == nil
|
|
for {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
if !kept {
|
|
kept = k.history.Append(ctx, e) == nil
|
|
}
|
|
if !said {
|
|
said = k.teller.PublishSeatEvent(ctx, Seat, e.Event, body) == nil
|
|
}
|
|
cancel()
|
|
if kept && said {
|
|
break
|
|
}
|
|
if time.Now().After(deadline) {
|
|
what := "said"
|
|
if !kept {
|
|
what = "kept in the history"
|
|
}
|
|
k.lost(e, fmt.Errorf("not %s within %s", what, TellFor))
|
|
break
|
|
}
|
|
time.Sleep(wait)
|
|
wait = min(2*wait, 10*time.Second)
|
|
}
|
|
}
|
|
}
|
|
|
|
func orSelf(resolver string) string {
|
|
if resolver == "" {
|
|
return ResolverSelf
|
|
}
|
|
return resolver
|
|
}
|
|
|
|
// Unsayable is told of every observation whose words the operator's channel would withhold (novox/hq
|
|
// ADR 0234 §6): its summary or its key carries an address, a domain, a path or a secret's shape. The
|
|
// keeper says such a summary in words itself, and keeps what it said whole in the evidence; a test
|
|
// suite sets this to fail the producer, which is where the summary should have been said rightly.
|
|
var Unsayable func(o Observation, field string, r outward.Refusal)
|
|
|
|
// sayable is an observation whose summary may leave the mesh: **a summary says things in machine names
|
|
// and the mesh's words** (novox/hq issue 277). One that carries what the channel withholds — a raw error
|
|
// with an address in it, a path — is said in words here, as the last stand before the operator would
|
|
// read "its words are withheld" instead of the alert; what it carried goes to the evidence, which
|
|
// stays inside the mesh.
|
|
func (k *Keeper) sayable(o Observation) Observation {
|
|
machines := append([]string{o.Machine}, o.Also...)
|
|
if r, ok := outward.Check(o.Key(), machines...); !ok && Unsayable != nil {
|
|
Unsayable(o, "key", r)
|
|
}
|
|
r, ok := outward.Check(o.Summary, machines...)
|
|
if ok {
|
|
return o
|
|
}
|
|
if Unsayable != nil {
|
|
Unsayable(o, "summary", r)
|
|
}
|
|
whole := o.Summary
|
|
o.Summary = outward.Scrub(whole, fmt.Sprintf("a %s condition about %s %s: what it says is kept in its evidence "+
|
|
"(`conditions show`)", o.Kind, o.Scope, o.ID), machines...)
|
|
switch {
|
|
case o.Said == "":
|
|
o.Said = whole
|
|
case !strings.Contains(o.Said, whole):
|
|
o.Said += " — as raised: " + whole
|
|
}
|
|
k.mu.Lock()
|
|
if k.reworded == nil {
|
|
k.reworded = map[string]bool{}
|
|
}
|
|
first := !k.reworded[o.Key()]
|
|
k.reworded[o.Key()] = true
|
|
k.mu.Unlock()
|
|
if first {
|
|
k.say("the condition %s's summary carried %s, which the operator's channel withholds: said in words, "+
|
|
"and kept whole in its evidence — its source should say it so", o.Key(), r.What)
|
|
}
|
|
return o
|
|
}
|