From 6bfd14b14bb3b47a9615a2dd32f4981b8fd1d5a3 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:23:21 +0100 Subject: [PATCH] CAT-RD1..RD10 sdk: views.Reader, a module's pictures of the owners' states (A8's patch 08da702) Keys, Get, Values and Watch over the views followers, the first-view tools, the finding past the bound, and the re-read recorded through mesh-hand-acts before it is asked; its cases are subtests of the views table. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- go/views/reader.go | 341 +++++++++++++++++++++++++++++++++++++++++ go/views/views_test.go | 65 ++++++++ 2 files changed, 406 insertions(+) create mode 100644 go/views/reader.go diff --git a/go/views/reader.go b/go/views/reader.go new file mode 100644 index 0000000..5e5d30b --- /dev/null +++ b/go/views/reader.go @@ -0,0 +1,341 @@ +package views + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/handacts" +) + +// Reader is a module's pictures of the states of the indexer and the assigner it used to read (novox/hq ADR 0615 §1): it +// asks each owner's read tool for the first view, subscribes the owner's views family, and answers the module's own +// `Keys` and `Get` of a state from the picture, by the name the module's `reads` used to carry (`.`). +// A module in a stdio tools bundle declares `invokes` for the tools and `consumes` for the events, and names no `reads` +// of these owners any more. Pictures are kept in memory only: the owner's state is the truth, and a restart asks again. +type Reader struct { + // Module is the reading module, named in the findings it writes. + Module string + // Findings writes and clears this module's finding of an unanswered first view (its own state `findings`); nil + // says it in the log only. + Findings interface { + Put(key string, value []byte) error + Delete(key string) error + } + Logf func(string, ...any) + // Changed is told of each key a followed state changes, after the picture holds it (state is `.`). + Changed func(state, key string, value json.RawMessage, deleted bool) + // Acts records a re-read by hand through the seat mesh-hand-acts before it is asked (the zero Recorder records nothing + // and the re-read is refused). + Acts handacts.Recorder + // Now, Bound and Wait are for tests: zero are the clock, Bound and five seconds. + Now func() time.Time + Bound time.Duration + Wait time.Duration + + mu sync.Mutex + followers map[string]*Follower // by "." + order []string + byOwner map[string]map[string]*Follower // owner -> state -> follower + watchers map[string][]func(key string, value []byte, gone bool) +} + +func (r *Reader) logf(format string, args ...any) { + if r.Logf != nil { + r.Logf(format, args...) + } +} + +// Follow adds a state of an owner to follow, asked through the owner's read tool (`.`, as the reader's +// `invokes` names it), with `values` so the answer is the first view. Call it before Start. +func (r *Reader) Follow(owner, state, tool string) *Follower { + r.mu.Lock() + defer r.mu.Unlock() + if r.followers == nil { + r.followers, r.byOwner = map[string]*Follower{}, map[string]map[string]*Follower{} + } + name := owner + "." + state + f := &Follower{Owner: owner, Tool: tool, State: state, Logf: r.Logf, Now: r.Now, Bound: r.Bound, Wait: r.Wait} + f.Ask = func(_ context.Context, after string) (Answer, error) { + body := map[string]any{"values": true} + if after != "" { + body["after"] = after + } + raw, err := stdio.Ask(tool, body) + if err != nil { + return Answer{}, err + } + return DecodeAnswer(raw) + } + f.Raise = func(s Status) { r.raise(s) } + f.Clear = func(s Status) { r.clear(s) } + f.Changed = func(key string, value json.RawMessage, deleted bool, _ time.Time) { + r.mu.Lock() + ws := append([]func(string, []byte, bool){}, r.watchers[name]...) + r.mu.Unlock() + for _, w := range ws { + w(key, value, deleted) + } + if r.Changed != nil { + r.Changed(name, key, value, deleted) + } + } + r.followers[name] = f + r.order = append(r.order, name) + if r.byOwner[owner] == nil { + r.byOwner[owner] = map[string]*Follower{} + } + r.byOwner[owner][state] = f + return f +} + +func (r *Reader) findingKey(s Status) (string, []byte, error) { + return UnansweredFinding(r.Module, s, r.nowTime()) +} + +func (r *Reader) nowTime() time.Time { + if r.Now != nil { + return r.Now().UTC() + } + return time.Now().UTC() +} + +func (r *Reader) raise(s Status) { + r.logf("the first view %s of %s's %s is unanswered past its bound: %s", s.Tool, s.Owner, s.State, s.Error) + if r.Findings == nil { + return + } + key, value, err := r.findingKey(s) + if err == nil { + err = r.Findings.Put(key, value) + } + if err != nil { + r.logf("the finding of the unanswered first view %s could not be written: %v", s.Tool, err) + } +} + +func (r *Reader) clear(s Status) { + if r.Findings == nil { + return + } + key, _, err := r.findingKey(s) + if err == nil { + err = r.Findings.Delete(key) + } + if err != nil { + r.logf("the finding of the first view %s could not be cleared: %v", s.Tool, err) + } +} + +// Start subscribes each owner's views family (the module's `consumes`: `.views.`) and asks every first +// view until it is answered, each on a goroutine of its own, so a module starts at once and a reader whose owner is down +// waits, saying so (ADR 0615 §6). It returns when the subscriptions are taken. +func (r *Reader) Start(ctx context.Context) { + r.mu.Lock() + names := append([]string(nil), r.order...) + r.mu.Unlock() + for _, name := range names { + f := r.followers[name] + go func() { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := stdio.Subscribe(f.Owner+"."+EventName(f.State), r.hear(ctx, f)) + if err == nil { + break + } + r.logf("not hearing %s's %s events yet (%v); asking again in %s", f.Owner, f.State, err, wait) + select { + case <-ctx.Done(): + return + case <-time.After(wait): + } + } + if err := f.Start(ctx); err != nil { + r.logf("the first view %s was not answered: %v", f.Tool, err) + } + }() + } +} + +// hear is the handler of one followed state's events: HearEvent routes it, and a gap asks that first view once more as a +// gap, said in the log. +func (r *Reader) hear(ctx context.Context, f *Follower) func(stdio.Envelope) error { + return func(e stdio.Envelope) error { + if err := HearEvent(ctx, map[string]*Follower{f.State: f}, e.Body, r.logf); err != nil { + r.logf("views: %s is not a views event, dropped: %v", e.Key, err) + } + return nil + } +} + +// Ready is closed once every followed state had its first view. +func (r *Reader) Ready() <-chan struct{} { + r.mu.Lock() + fs := make([]*Follower, 0, len(r.followers)) + for _, name := range r.order { + fs = append(fs, r.followers[name]) + } + r.mu.Unlock() + done := make(chan struct{}) + go func() { + for _, f := range fs { + <-f.Ready() + } + close(done) + }() + return done +} + +// Keys is the keys of a followed state (`.`); a state not followed is an error, never a read of it. +func (r *Reader) Keys(state string) ([]string, error) { + f, err := r.picture(state) + if err != nil { + return nil, err + } + vals := f.Values() + keys := make([]string, 0, len(vals)) + for k := range vals { + keys = append(keys, k) + } + sort.Strings(keys) + return keys, nil +} + +// Get is one key of a followed state. +func (r *Reader) Get(state, key string) ([]byte, bool, error) { + f, err := r.picture(state) + if err != nil { + return nil, false, err + } + v, ok := f.Get(key) + return v, ok, nil +} + +// Values is every key of a followed state, by key. +func (r *Reader) Values(state string) (map[string][]byte, error) { + f, err := r.picture(state) + if err != nil { + return nil, err + } + out := map[string][]byte{} + for k, v := range f.Values() { + out[k] = v + } + return out, nil +} + +func (r *Reader) picture(state string) (*Follower, error) { + r.mu.Lock() + f := r.followers[state] + r.mu.Unlock() + if f == nil { + return nil, fmt.Errorf("%s is not a state this module follows", state) + } + select { + case <-f.Ready(): + return f, nil + default: + return nil, fmt.Errorf("no first view of %s yet: %s has not answered", state, f.Tool) + } +} + +// Watch is a followed state as a module's store watches one: it returns once the first view was answered and every +// current value was handed to handle, and then hands every change for as long as the process runs (gone is a key +// deleted). A change heard while the current values are handed over may be handed twice; a handler is idempotent. +func (r *Reader) Watch(state string, handle func(key string, value []byte, gone bool)) error { + r.mu.Lock() + f := r.followers[state] + if f == nil { + r.mu.Unlock() + return fmt.Errorf("%s is not a state this module follows", state) + } + r.mu.Unlock() + <-f.Ready() + r.mu.Lock() + if r.watchers == nil { + r.watchers = map[string][]func(string, []byte, bool){} + } + r.watchers[state] = append(r.watchers[state], handle) + r.mu.Unlock() + for k, v := range f.Values() { + handle(k, v, false) + } + return nil +} + +// Follows says whether the state is one this reader follows. +func (r *Reader) Follows(state string) bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.followers[state] != nil +} + +// Statuses is each followed state's status, in the order they were added. +func (r *Reader) Statuses() []Status { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]Status, 0, len(r.order)) + for _, name := range r.order { + out = append(out, r.followers[name].Status()) + } + return out +} + +// Tools are the reader's two tools (design 25 § 12.2): `_first_views` says where each first view stands and +// `_first_view_again` asks one again by hand (a hand act, counted in the status). +func (r *Reader) Tools(prefix string) []stdio.Tool { + return []stdio.Tool{ + {Name: prefix + "_first_views", + Description: "Where each first view this module keeps stands (novox/hq ADR 0615): the owner and its tool, the revision the picture holds, when it was asked and answered, how many asks (by hand among them) and whether its condition is raised. (r)", + Run: func(map[string]any) (any, error) { return map[string]any{"views": r.Statuses()}, nil }}, + {Name: prefix + "_first_view_again", + Description: "Ask a first view again by hand, for the state named (.), or every one with none: a hand act, for a picture the operator doubts. Answers where it stands after. (a)", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "state": map[string]any{"type": "string", "description": ".; none asks every first view"}}}, + Run: func(args map[string]any) (any, error) { + want, _ := args["state"].(string) + want = strings.TrimSpace(want) + r.mu.Lock() + var fs []*Follower + for _, name := range r.order { + if want == "" || want == name { + fs = append(fs, r.followers[name]) + } + } + r.mu.Unlock() + if len(fs) == 0 { + return nil, fmt.Errorf("%q is not a state this module follows", want) + } + // A hand act, recorded through the seat mesh-hand-acts before it is done; one that cannot be recorded is + // said and nothing is asked. + why := "the operator asked" + if _, err := r.Acts.Record(handacts.Request{What: "asked a first view again by hand (" + r.Module + ")", Why: why, + Cause: "first-view-again", Verb: "first-view-again"}); err != nil { + return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was asked again: %w", err) + } + for _, f := range fs { + if err := f.Again(context.Background(), why); err != nil { + return nil, err + } + } + return map[string]any{"views": r.Statuses()}, nil + }}, + } +} + +// OwnFindings writes a reader's findings into its own state `findings` through its runtime, where the condition keeper +// raises and clears conditions from them: the Findings of a Reader whose module keeps the state `findings`. +type OwnFindings struct{} + +// Put writes one finding. +func (OwnFindings) Put(key string, value []byte) error { + _, err := stdio.State("findings").Put(key, json.RawMessage(value)) + return err +} + +// Delete clears one finding. +func (OwnFindings) Delete(key string) error { return stdio.State("findings").Delete(key) } diff --git a/go/views/views_test.go b/go/views/views_test.go index b8867b5..ebeef12 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -10,6 +10,7 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/handacts" ) // mapSource is a state held in memory, as a Source. @@ -212,6 +213,53 @@ func TestViews(t *testing.T) { t.Error("data that is no views event was taken") } }) + t.Run("a module's reader answers Keys and Get from its pictures, refuses a state it does not follow or has no first view of, and writes and clears its finding", func(t *testing.T) { + acted := 0 + fs := &findingsMap{} + r := &Reader{Module: "reader", Findings: fs, Bound: time.Nanosecond, Wait: time.Millisecond, Acts: handacts.Recorder{Ask: func(string, any) (json.RawMessage, error) { + acted++ + return json.RawMessage(`{"log":"mesh-hand-acts"}`), nil + }}} + f := r.Follow("o", "modules", "o.indexer_read_modules") + down := 2 + f.Ask = func(ctx context.Context, after string) (Answer, error) { + if down > 0 { + down-- + return Answer{}, errors.New("no responders") + } + return answer(ctx, after) + } + if _, err := r.Keys("o.modules"); err == nil { + t.Errorf("a state with no first view is read") + } + if _, err := r.Keys("o.other"); err == nil { + t.Errorf("a state not followed is read") + } + if err := f.Start(context.Background()); err != nil { + t.Fatal(err) + } + keys, err := r.Keys("o.modules") + v, ok, _ := r.Get("o.modules", "b") + if err != nil || len(keys) != 2 || !ok || string(v) != `{"n":2}` || len(fs.m) != 0 || fs.puts != 1 || fs.deletes != 1 { + t.Errorf("keys %v, b %s, findings %+v", keys, v, fs) + } + handed := map[string]bool{} + if err := r.Watch("o.modules", func(k string, _ []byte, gone bool) { handed[k] = !gone }); err != nil || len(handed) != 2 || !r.Follows("o.modules") || r.Follows("o.x") { + t.Errorf("watch handed %v, %v", handed, err) + } + f.Apply(ev(10, "c", `{"n":3}`)) + f.Apply(ev(11, "a", "")) + if !handed["c"] || handed["a"] { + t.Errorf("a change was not handed to the watch: %v", handed) + } + if got := r.Tools("reader"); len(got) != 2 || got[0].Name != "reader_first_views" { + t.Errorf("tools = %v", got) + } + again, err := r.Tools("reader")[1].Run(map[string]any{"state": "o.modules"}) + if err != nil || len(again.(map[string]any)["views"].([]Status)) != 1 || f.Status().ByHand != 1 || acted != 1 { + t.Errorf("asked again by hand: %v, %v", again, err) + } + }) t.Run("an event names its state on its subject, and decodes as it was encoded", func(t *testing.T) { if s := StateOfEvent("mesh.stream.mesh-delivery-indexer.event.views.config-candidates"); s != "config-candidates" { t.Errorf("state of subject = %q", s) @@ -226,3 +274,20 @@ func TestViews(t *testing.T) { } }) } + +// findingsMap is a module's own findings state in memory. +type findingsMap struct { + m map[string][]byte + puts, deletes int +} + +func (f *findingsMap) Put(k string, v []byte) error { + if f.m == nil { + f.m = map[string][]byte{} + } + f.m[k] = v + f.puts++ + return nil +} + +func (f *findingsMap) Delete(k string) error { delete(f.m, k); f.deletes++; return nil }