diff --git a/modules/mesh-conditions/cmd/mesh-conditions/events.go b/modules/mesh-conditions/cmd/mesh-conditions/events.go index 3fbb7fbfd..bba53987e 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/events.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/events.go @@ -7,8 +7,10 @@ package main import ( "encoding/json" + "errors" "fmt" "slices" + "sort" "strings" "time" @@ -21,6 +23,7 @@ import ( const ( EventFindingWritten = "finding-written" EventFindingCleared = "finding-cleared" + EventFindingsStated = "findings-stated" ) // emitterOf is the module an event's envelope key `.` names. @@ -37,6 +40,9 @@ func (k *Keeper) FindingEvent(e stdio.Envelope) error { if observer, ok := emitterOf(e.Key, EventFindingCleared); ok { return k.FindingCleared(observer, e.Node, e.Body) } + if observer, ok := emitterOf(e.Key, EventFindingsStated); ok { + return k.FindingsStated(observer, e.Node, e.Body) + } return nil } @@ -67,11 +73,32 @@ func (k *Keeper) FindingWritten(observer, node string, body []byte) error { f.Scope, f.ID, f.Subject.Scope, f.Subject.ID) return nil } + if !k.accepts(observer, node, f.At) { + k.Logf("%s's finding-written on %s is not its holder's, or is older than the holder's last statement: left out", observer, node) + return nil + } + return k.take(observer, node, f) +} + +// take holds one finding an observer says on a node, and raises or changes its condition. +func (k *Keeper) take(observer, node string, f seat.Finding) error { v := storedFinding(f) value, err := json.Marshal(v) if err != nil { return err } + k.mu.Lock() + k.init() + if k.keyNode[observer] == nil { + k.keyNode[observer] = map[string]string{} + } + k.keyNode[observer][v.Key] = node + named := k.view(observer, node).Pending[v.Key] + delete(k.view(observer, node).Pending, v.Key) + k.mu.Unlock() + if !named.IsZero() { + _ = k.Store.Delete(BoundsState, pendingKey(observer, node, v.Key)) + } return k.Finding(observer, v.Key, value, false) } @@ -86,6 +113,10 @@ func (k *Keeper) FindingCleared(observer, node string, body []byte) error { k.Logf("%s's finding-cleared on %s is not readable: %v", observer, node, err) return nil } + if !k.accepts(observer, node, c.At) { + k.Logf("%s's finding-cleared on %s is not its holder's, or is older than the holder's last statement: left out", observer, node) + return nil + } k.mu.Lock() k.init() key := cc.Key(c.Scope, c.ID, c.Kind) @@ -119,7 +150,7 @@ func storedFinding(f seat.Finding) seat.FindingsByKeyValue { // hear subscribes to the observers' events once the runtime serves, asked again until it takes each. func hear(k *Keeper) { - for _, event := range []string{EventFindingWritten, EventFindingCleared} { + for _, event := range []string{EventFindingWritten, EventFindingCleared, EventFindingsStated} { pattern := "*." + event go func() { <-stdio.Served() @@ -134,3 +165,190 @@ func hear(k *Keeper) { }() } } + +// pendingPrefix starts the keys of the state `bounds` that are a pending key's bound. +const pendingPrefix = "pending=" + +func pendingKey(observer, node, key string) string { + return pendingPrefix + observer + "=" + node + "=" + key +} + +// notJudged is why a pending key that nothing named within its bound is cleared. +const notJudged = "not judged within its bound" + +// sourceView is what the keeper heard of one observer on one node. +type sourceView struct { + // At is the last findings-stated's at; Pending are its pending keys with the end of their bound. + At time.Time + Pending map[string]time.Time +} + +// view is the observer's view on the node, made when there is none. Called with mu held. +func (k *Keeper) view(observer, node string) *sourceView { + v := k.views[observer+"."+node] + if v == nil { + v = &sourceView{Pending: map[string]time.Time{}} + k.views[observer+"."+node] = v + } + return v +} + +// holderOf says whether an observer's events are taken from node alone, and so from which (design 47 G §384). +func (k *Keeper) holderOf(observer string) (string, bool) { + if k.Holder == nil { + return "", false + } + return k.Holder(observer) +} + +// accepts says whether an event of the observer on the node, said at at, is taken: for an observer held once, only the +// holder's, and only with an at later than the holder's last statement, so an old holder's late event changes nothing. +func (k *Keeper) accepts(observer, node string, at time.Time) bool { + holder, once := k.holderOf(observer) + if !once { + return true + } + if node != holder { + return false + } + k.mu.Lock() + defer k.mu.Unlock() + k.init() + v := k.views[observer+"."+node] + return v == nil || at.After(v.At) +} + +// FindingsStated sets the keeper's view of an observer on a node to the set it states (design 47 G §384): what is new is +// raised, each pending key keeps its condition unchanged, and every other condition of the observer on the node (of every +// node, for an observer held once) is cleared. +func (k *Keeper) FindingsStated(observer, node string, body []byte) error { + if !k.isSource(observer) { + k.Logf("%s says findings-stated on %s, and is no source of findings: left out", observer, node) + return nil + } + var s seat.FindingsStated + if err := json.Unmarshal(body, &s); err != nil { + k.Logf("%s's findings-stated on %s is not readable: %v", observer, node, err) + return nil + } + for _, f := range s.Findings { + if f.Observer != observer || f.Scope != f.Subject.Scope || f.ID != f.Subject.ID { + k.Logf("%s's findings-stated on %s names the observer %q for %s.%s: refused", observer, node, f.Observer, f.Scope, f.ID) + return nil + } + } + holder, once := k.holderOf(observer) + if once && node != holder { + k.Logf("%s's findings-stated on %s is not its holder's (%s): left out", observer, node, holder) + return nil + } + now := k.Now().UTC() + k.mu.Lock() + k.init() + v := k.view(observer, node) + if !v.At.IsZero() && !s.At.After(v.At) { + k.mu.Unlock() + k.Logf("%s's findings-stated on %s at %s is not newer than the one at %s: left out", observer, node, s.At, v.At) + return nil + } + v.At = s.At + named := map[string]bool{} + for _, f := range s.Findings { + named[storedFinding(f).Key] = true + } + oldPending := v.Pending + v.Pending = map[string]time.Time{} + type pend struct { + key string + deadline time.Time + } + var pending []pend + for _, p := range s.Pending { + bound, err := time.ParseDuration(p.Bound) + if err != nil { + bound = 0 + } + named[p.Key] = true + v.Pending[p.Key] = s.At.Add(bound) + pending = append(pending, pend{p.Key, s.At.Add(bound)}) + } + var drop []string + for key := range k.findings[observer] { + if !named[key] && (once || k.keyNode[observer][key] == node) { + drop = append(drop, key) + } + } + k.mu.Unlock() + sort.Strings(drop) + for key := range oldPending { + if _, still := v.Pending[key]; !still { + _ = k.Store.Delete(BoundsState, pendingKey(observer, node, key)) + } + } + var problems []string + for _, f := range s.Findings { + if err := k.take(observer, node, f); err != nil { + problems = append(problems, err.Error()) + } + } + why := "not stated when asked" + if s.Asked == "" { + why = "not stated at " + observer + "'s start" + } + for _, key := range drop { + k.mu.Lock() + k.whys[key] = why + k.mu.Unlock() + if err := k.Finding(observer, key, nil, true); err != nil { + problems = append(problems, err.Error()) + } + k.mu.Lock() + delete(k.whys, key) + delete(k.keyNode[observer], key) + k.mu.Unlock() + } + for _, p := range pending { + _ = k.Store.Delete(BoundsState, pendingKey(observer, node, p.key)) + ttl := p.deadline.Sub(now) + if ttl < time.Second { + if err := k.PendingPassed(pendingKey(observer, node, p.key)); err != nil { + problems = append(problems, err.Error()) + } + continue + } + if err := k.Store.CreateTTL(BoundsState, pendingKey(observer, node, p.key), map[string]any{"observer": observer, "node": node, + "key": p.key, "until": p.deadline}, ttl); err != nil && !errors.Is(err, stdio.ErrChanged) { + problems = append(problems, err.Error()) + } + } + if len(problems) > 0 { + return errors.New(strings.Join(problems, "; ")) + } + return nil +} + +// PendingPassed is a pending key's bound passing: unless a finding or a later set named it since, its condition is cleared. +func (k *Keeper) PendingPassed(bound string) error { + rest := strings.TrimPrefix(bound, pendingPrefix) + observer, rest, ok1 := strings.Cut(rest, "=") + node, key, ok2 := strings.Cut(rest, "=") + if !ok1 || !ok2 { + return nil + } + k.mu.Lock() + k.init() + v := k.views[observer+"."+node] + if v == nil || v.Pending[key].IsZero() { + k.mu.Unlock() + return nil + } + delete(v.Pending, key) + k.whys[key] = notJudged + k.mu.Unlock() + err := k.Finding(observer, key, nil, true) + k.mu.Lock() + delete(k.whys, key) + delete(k.keyNode[observer], key) + k.mu.Unlock() + return err +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/hold.go b/modules/mesh-conditions/cmd/mesh-conditions/hold.go index f34e5655d..f37b7da31 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/hold.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/hold.go @@ -229,6 +229,9 @@ func (k *Keeper) Assignments(key string, value []byte, deleted bool) { // BoundPassed is a delivery's bound key expiring: what it holds is raised, saying the delivery did not end. func (k *Keeper) BoundPassed(key string) error { + if strings.HasPrefix(key, pendingPrefix) { + return k.PendingPassed(key) + } node, module := splitNodeModule(key) k.mu.Lock() k.init() diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go index 1a5341e32..2ca8086a0 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -144,6 +144,14 @@ type Keeper struct { outOfHold map[string]string cleared map[string]clearedMemo seq uint64 + // views are, by `.`, what the keeper heard of that observer on that node (events.go); keyNode is the + // node each finding of an observer came from; whys is the reason a key's next clearing says. + views map[string]*sourceView + keyNode map[string]map[string]string + whys map[string]string + // Holder says, for an observer whose catalog entry claims only seats held once for the mesh, the node the seat list + // names as its holder (design 47 G §384); nil, or once false, is an observer on every node. + Holder func(observer string) (node string, once bool) } func (k *Keeper) init() { @@ -154,6 +162,7 @@ func (k *Keeper) init() { map[string]string{}, map[string]held{}, map[string]clearedMemo{} k.passed, k.outOfHold, k.holders = map[string]string{}, map[string]string{}, map[string]string{} k.assignments = assignerclient.NewKeys() + k.views, k.keyNode, k.whys = map[string]*sourceView{}, map[string]map[string]string{}, map[string]string{} } if k.Now == nil { k.Now = time.Now @@ -341,7 +350,14 @@ func (k *Keeper) settleOnce(key string) error { if !open { return nil } - return k.clear(key, was, rev, now, "its finding went") + why := "its finding went" + k.mu.Lock() + if w, ok := k.whys[key]; ok { + why = w + delete(k.whys, key) + } + k.mu.Unlock() + return k.clear(key, was, rev, now, why) } if holding { return k.keepHeld(hold) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index cfa1e6f96..f0074bfc0 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -457,3 +457,86 @@ func TestTheKeeperReadsTheOldLayoutAndWritesTheSeatsLayout(t *testing.T) { t.Errorf("the event the seat hears: %+v", last) } } + +// Design 47 G §384: the keeper sets its view of an observer on a node to the set it states: what is new is raised, a +// pending key is kept unchanged, every other condition of the observer on that node is cleared with its reason, a pending key +// nothing named within its bound is cleared, and for an observer held once the holder's set replaces every node's. +func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) { + r := newRig() + holder := "ace" + r.k.Holder = func(o string) (string, bool) { return holder, o == selfCheck } + f := func(observer, id string) seat.Finding { + g := eventFinding(observer, r.now) + g.ID, g.Subject.ID = id, id + return g + } + stated := func(observer, node, asked string, at time.Time, pend []seat.FindingsStatedPendingItem, fs ...seat.Finding) error { + return r.k.FindingsStated(observer, node, mustJSON(t, seat.FindingsStated{At: at, Asked: asked, Findings: fs, Pending: pend})) + } + why := func() string { + defer func() { r.said = nil }() + if len(r.said) == 0 || r.said[len(r.said)-1].Change != cc.ChangeCleared { + return "" + } + return r.said[len(r.said)-1].Why + } + t0 := r.now + steps := []struct { + name string + do func() error + open []string // the keys open after + why string + }{ + {"a start states two findings: both raised", func() error { + return stated(watchman, "ace", "", t0, nil, f(watchman, "a"), f(watchman, "b")) + }, []string{"machine.a.silent", "machine.b.silent"}, ""}, + {"a later set without b clears it, with its reason", func() error { + return stated(watchman, "ace", "ask-1", t0.Add(time.Minute), nil, f(watchman, "a")) + }, []string{"machine.a.silent"}, "not stated when asked"}, + {"a restart that states nothing clears the rest", func() error { + return stated(watchman, "ace", "", t0.Add(2*time.Minute), nil) + }, nil, "not stated at " + watchman + "'s start"}, + {"a pending key's condition stays open and unsaid", func() error { + _ = r.k.FindingWritten(watchman, "ace", mustJSON(t, f(watchman, "c"))) + r.said = nil + return stated(watchman, "ace", "", t0.Add(3*time.Minute), []seat.FindingsStatedPendingItem{{Key: "machine.c.silent", Since: t0, Bound: "5m"}}) + }, []string{"machine.c.silent"}, ""}, + {"a pending key nothing named within its bound is cleared", func() error { + return r.k.PendingPassed(pendingKey(watchman, "ace", "machine.c.silent")) + }, nil, notJudged}, + {"a set of another node clears nothing of ace", func() error { + _ = r.k.FindingWritten(watchman, "ace", mustJSON(t, f(watchman, "d"))) + r.said = nil + return stated(watchman, "g14", "", t0.Add(4*time.Minute), nil) + }, []string{"machine.d.silent"}, ""}, + {"an older set is left out", func() error { + return stated(watchman, "ace", "", t0.Add(-time.Hour), nil) + }, []string{"machine.d.silent"}, ""}, + {"a held-once observer's old holder raises nothing after the new holder's set", func() error { + if err := stated(selfCheck, "ace", "", t0.Add(5*time.Minute), nil, f(selfCheck, "e")); err != nil { + return err + } + holder = "g14" + if err := stated(selfCheck, "g14", "", t0.Add(6*time.Minute), nil, f(selfCheck, "g")); err != nil { + return err + } + return r.k.FindingWritten(selfCheck, "ace", mustJSON(t, func() seat.Finding { x := f(selfCheck, "h"); x.At = t0.Add(7 * time.Minute); return x }())) + }, []string{"machine.d.silent", "machine.g.silent"}, "not stated at " + selfCheck + "'s start"}, + } + for _, s := range steps { + if err := s.do(); err != nil { + t.Fatalf("%s: %v", s.name, err) + } + var open []string + for _, c := range func() []Condition { c, _ := r.k.Open(); return c }() { + open = append(open, c.Key) + } + sort.Strings(open) + if !slices.Equal(open, s.open) { + t.Errorf("%s: open %v, want %v", s.name, open, s.open) + } + if got := why(); got != s.why { + t.Errorf("%s: cleared because %q, want %q", s.name, got, s.why) + } + } +}