diff --git a/modules/mesh-conditions/cmd/mesh-conditions/asks.go b/modules/mesh-conditions/cmd/mesh-conditions/asks.go new file mode 100644 index 000000000..a380abd87 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/asks.go @@ -0,0 +1,194 @@ +package main + +// The keeper's ask and what follows it (design 47 G §390): `findings-asked` on its seat asks the observers to state their +// findings again; one that states nothing within five minutes, its machine not away, is the keeper's finding +// `findings-unstated`, cleared by its next `findings-stated`. + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + cc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// AskedWithin is how long an observer has to state its findings after an ask. +const AskedWithin = 5 * time.Minute + +// The kind of the keeper's finding of an observer that stated nothing, and the prefix of its ask's key in `bounds`. +const ( + KindFindingsUnstated = "findings-unstated" + askedPrefix = "asked=" + // FindingsViewsTool reads what the keeper holds of each observer and node. + FindingsViewsTool = "conditions_findings_views" +) + +func askedKey(observer, node, ask string) string { + return askedPrefix + observer + "=" + node + "=" + ask +} + +func sayAsk(body any) error { + return stdio.SeatPublish(seat.SubjectFindingsAsked, body, "") +} + +func recordHandAct(what, why, verb string) error { + _, err := meshhandactsclient.Recorder{Ask: stdio.Ask}.Record(meshhandactsclient.Request{What: what, Why: why, Verb: verb}) + return err +} + +// machineAway says the watchman holds the machine away. +func (k *Keeper) machineAway(node string) bool { + k.mu.Lock() + defer k.mu.Unlock() + return k.away[node] +} + +// Ask says `findings-asked` for module alone, or for every observer when module is empty, and starts the bound of each +// observer on a machine not away. It answers the ask's id. +func (k *Keeper) Ask(by, module string) (string, error) { + now := k.Now().UTC() + k.mu.Lock() + k.init() + k.seq++ + ask := fmt.Sprintf("%d-%d", now.UnixNano(), k.seq) + var wait [][2]string + for _, v := range k.views { + if v.observer == "" || module != "" && v.observer != module || k.away[v.node] { + continue + } + v.Asks[ask] = now.Add(AskedWithin) + wait = append(wait, [2]string{v.observer, v.node}) + } + k.mu.Unlock() + body := map[string]any{"ask": ask, "at": now, "by": by} + if module != "" { + body["module"] = module + } + publish := k.Publish + if publish == nil { + publish = func(_ string, body any) error { return sayAsk(body) } + } + if err := publish(seat.EventFindingsAsked, body); err != nil { + return "", fmt.Errorf("asking the observers to state their findings: %w", err) + } + var problems []string + for _, w := range wait { + if err := k.Store.CreateTTL(BoundsState, askedKey(w[0], w[1], ask), map[string]any{"observer": w[0], "node": w[1], "ask": ask, + "until": now.Add(AskedWithin)}, AskedWithin); err != nil && !errors.Is(err, stdio.ErrChanged) { + problems = append(problems, err.Error()) + } + } + if len(problems) > 0 { + return ask, errors.New(strings.Join(problems, "; ")) + } + return ask, nil +} + +// ResyncVerb is `resync`: it writes the call in the hand-act log, asks the observers (or the one named) to state their +// findings, and settles every key the keeper holds. +func (k *Keeper) ResyncVerb(by, module string) (map[string]any, error) { + what := "asked every observer to state its findings again" + if module != "" { + what = "asked " + module + " to state its findings again" + } + record := k.Record + if record == nil { + record = recordHandAct + } + if err := record(what, "resync, asked by "+by, seat.Seat+".resync"); err != nil { + return nil, fmt.Errorf("the hand act was not recorded, so nothing was asked: %w", err) + } + ask, err := k.Ask(by, module) + if err != nil { + return nil, err + } + n, err := k.Resync() + return map[string]any{"ask": ask, "settled": n}, err +} + +// AskPassed is an ask's five minutes ending: an observer that stated nothing since, its machine not away, is the finding +// findings-unstated. +func (k *Keeper) AskPassed(bound string) error { + observer, rest, ok1 := strings.Cut(strings.TrimPrefix(bound, askedPrefix), "=") + node, ask, ok2 := strings.Cut(rest, "=") + if !ok1 || !ok2 { + return nil + } + k.mu.Lock() + k.init() + v := k.views[observer+"."+node] + if v == nil { + k.mu.Unlock() + return nil + } + _, open := v.Asks[ask] + delete(v.Asks, ask) + away := k.away[node] + k.mu.Unlock() + if !open || away { + return nil + } + now := k.Now().UTC() + f := seat.FindingsByKeyValue{V: cc.FindingVersion, Key: unstatedKey(observer, node), Kind: KindFindingsUnstated, + Subject: seat.Subject{Scope: cc.ScopeModule, ID: observer + "." + node, Machine: node}, + Severity: cc.Warning, Summary: fmt.Sprintf("%s on %s stated no findings within %s of the ask", observer, node, AskedWithin), + Headline: observer + " on " + node + " did not state its findings", + Explanation: "The keeper asked " + observer + " on " + node + " to state its findings and heard nothing; its conditions stand as they were.", + Source: "findings-asked", Observer: Module, At: now, Evidence: []seat.Evidence{{At: now, Said: "ask " + ask + " unanswered"}}} + raw, err := json.Marshal(f) + if err != nil { + return err + } + return k.Finding(Module, f.Key, raw, false) +} + +func unstatedKey(observer, node string) string { + return cc.Key(cc.ScopeModule, observer+"."+node, KindFindingsUnstated) +} + +// clearUnstated clears the finding findings-unstated of the observer on the node: it stated. +func (k *Keeper) clearUnstated(observer, node string) { + k.mu.Lock() + k.init() + _, held := k.findings[Module][unstatedKey(observer, node)] + k.mu.Unlock() + if held { + if err := k.Finding(Module, unstatedKey(observer, node), nil, true); err != nil { + k.Logf("%v", err) + } + } +} + +// FindingsViews answers each observer and node: its last stated set with its at, its pending keys and its open asks. +func (k *Keeper) FindingsViews() []map[string]any { + k.mu.Lock() + defer k.mu.Unlock() + k.init() + var out []map[string]any + for _, v := range k.views { + if v.observer == "" { + continue + } + pending, asks := []string{}, []string{} + for key := range v.Pending { + pending = append(pending, key) + } + for ask := range v.Asks { + asks = append(asks, ask) + } + sort.Strings(pending) + sort.Strings(asks) + out = append(out, map[string]any{"observer": v.observer, "node": v.node, "stated": map[string]any{"at": v.At, "findings": v.Findings}, + "pending": pending, "asks": asks}) + } + sort.Slice(out, func(i, j int) bool { + return fmt.Sprint(out[i]["observer"], out[i]["node"]) < fmt.Sprint(out[j]["observer"], out[j]["node"]) + }) + return out +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/events.go b/modules/mesh-conditions/cmd/mesh-conditions/events.go index bba53987e..830c8be39 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/events.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/events.go @@ -178,16 +178,20 @@ const notJudged = "not judged within its bound" // sourceView is what the keeper heard of one observer on one node. type sourceView struct { + observer, node string // 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 + At time.Time + Pending map[string]time.Time + Findings []seat.Finding + // Asks are the asks it has not answered, with the end of their bound (asks.go). + Asks 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{}} + v = &sourceView{observer: observer, node: node, Pending: map[string]time.Time{}, Asks: map[string]time.Time{}} k.views[observer+"."+node] = v } return v @@ -252,6 +256,9 @@ func (k *Keeper) FindingsStated(observer, node string, body []byte) error { return nil } v.At = s.At + v.Findings = s.Findings + answered := v.Asks + v.Asks = map[string]time.Time{} named := map[string]bool{} for _, f := range s.Findings { named[storedFinding(f).Key] = true @@ -279,6 +286,10 @@ func (k *Keeper) FindingsStated(observer, node string, body []byte) error { } } k.mu.Unlock() + for ask := range answered { + _ = k.Store.Delete(BoundsState, askedKey(observer, node, ask)) + } + k.clearUnstated(observer, node) sort.Strings(drop) for key := range oldPending { if _, still := v.Pending[key]; !still { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/hold.go b/modules/mesh-conditions/cmd/mesh-conditions/hold.go index f37b7da31..8c9247581 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/hold.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/hold.go @@ -232,6 +232,9 @@ func (k *Keeper) BoundPassed(key string) error { if strings.HasPrefix(key, pendingPrefix) { return k.PendingPassed(key) } + if strings.HasPrefix(key, askedPrefix) { + return k.AskPassed(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 2ca8086a0..ae523a835 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -152,6 +152,12 @@ type Keeper struct { // 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) + // Publish says the body on the subject of the seat's event (findings-asked); Record writes a hand act in the hand-act + // log before it is done (design 47 G §390). + Publish func(event string, body any) error + Record func(what, why, verb string) error + // away are the machines the watchman says are away (design 47 G §389). + away map[string]bool } func (k *Keeper) init() { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index f0074bfc0..4105d581f 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -481,6 +481,12 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) { return r.said[len(r.said)-1].Why } t0 := r.now + var asked []map[string]any + var acts []string + r.k.Publish = func(_ string, b any) error { asked = append(asked, b.(map[string]any)); return nil } + r.k.Record = func(what, why, verb string) error { acts = append(acts, verb+": "+what); return nil } + unstated := cc.Key(cc.ScopeModule, watchman+".ace", KindFindingsUnstated) + var ask string steps := []struct { name string do func() error @@ -522,6 +528,33 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) { } 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"}, + // Design 47 G §390: the ask, findings-unstated and the hand act. + {"an ask for every observer waits five minutes on each", func() (err error) { + ask, err = r.k.Ask("jochen", "") + return err + }, []string{"machine.d.silent", "machine.g.silent"}, ""}, + {"an observer silent past the ask is findings-unstated", func() error { + return r.k.AskPassed(askedKey(watchman, "ace", ask)) + }, []string{"machine.d.silent", "machine.g.silent", unstated}, ""}, + {"its next findings-stated clears it", func() error { + return stated(watchman, "ace", ask, t0.Add(10*time.Minute), nil, f(watchman, "d")) + }, []string{"machine.d.silent", "machine.g.silent"}, "its finding went"}, + {"a machine away is not named", func() error { + if _, err := r.k.Ask("jochen", ""); err != nil { + return err + } + r.k.away = map[string]bool{"g14": true} + return r.k.AskPassed(askedKey(selfCheck, "g14", asked[len(asked)-1]["ask"].(string))) + }, []string{"machine.d.silent", "machine.g.silent"}, ""}, + {"resync names one module: its hand act and its ask", func() error { + if _, err := r.k.ResyncVerb("person.jochen", selfCheck); err != nil { + return err + } + if len(acts) != 1 || asked[len(asked)-1]["module"] != selfCheck || len(r.k.FindingsViews()) == 0 { + t.Errorf("hand acts %v, last ask %v", acts, asked[len(asked)-1]) + } + return nil + }, []string{"machine.d.silent", "machine.g.silent"}, ""}, } for _, s := range steps { if err := s.do(); err != nil { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go index fb4a6fe85..98ef5b04b 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -63,7 +63,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { } // Clause 12: each state with the tool that reads it, each act with the tool that starts it. readBy := map[string]string{ConditionsState: Seat + ".conditions", HistoryState: HistoryTool, - HeldState: HeldTool, BoundsState: HeldTool, "a resync": Seat + ".resync", "what it reads": InputsTool} + HeldState: HeldTool, BoundsState: HeldTool, "a resync": Seat + ".resync", "what it holds of observers": FindingsViewsTool, "what it reads": InputsTool} tools := map[string]bool{} for _, tool := range Tools(&Keeper{}) { tools[tool.Name] = true diff --git a/modules/mesh-conditions/cmd/mesh-conditions/tools.go b/modules/mesh-conditions/cmd/mesh-conditions/tools.go index 9e0d80044..621fe6f46 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/tools.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/tools.go @@ -72,11 +72,15 @@ func Tools(k *Keeper) []stdio.Tool { c, err := k.Silence(strArg(args, "key"), d, stdio.CallerOf(args), strArg(args, "why")) return c.Stored(), err }}, - {Name: Seat + ".resync", Description: "Read every finding again and raise and clear what differs.", - Run: func(map[string]any) (any, error) { - n, err := k.Resync() - return map[string]any{"settled": n}, err + {Name: Seat + ".resync", Description: "Ask the observers to state their findings again, every one or the one named, and raise " + + "and clear what differs from the open conditions; the call is written in the hand-act log.", + Input: obj(map[string]any{"module": str("ask this observer alone; none asks all")}), + Run: func(args map[string]any) (any, error) { + return k.ResyncVerb(stdio.CallerOf(args), strArg(args, "module")) }}, + {Name: FindingsViewsTool, Description: "What the keeper holds of each observer and node: its last stated set with its at, " + + "its pending keys and its open asks (design 47 G §390).", + Run: func(map[string]any) (any, error) { return k.FindingsViews(), nil }}, {Name: HistoryTool, Description: "The keeper's history: every transition it said, newest first, at most limit.", Input: obj(map[string]any{"limit": str("how many (default 50)"), "key": str("only this condition")}), Run: func(args map[string]any) (any, error) {