Merge branch 'feat/47-G-build' into feat/47-G-build-bus

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 `<module>.<event>` 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
+}

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()

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 `<observer>.<node>`, 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)

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)
+		}
+	}
+}