From d03ffa80388fad009df76be1c0d3acc81a4bfef3 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:22:24 +0100 Subject: [PATCH] G findings: mesh-config's and mesh-vault's owner views are followed by the reader (Follow), not adopted: they hear the owner's own views event Adopt subscribes a seat's event with further parts (..>), which never matches the plain views event of an owner; the reader's Follow hears it and asks again on a gap. Also the distribution whole test counts its own two events by subject, as the writer's are said beside them. Builds: design 47 G A-28, A-26, A-30 --- .../distribution/whole_integration_test.go | 9 ++-- .../mesh-config/cmd/mesh-config/ownerviews.go | 52 +++++++----------- .../cmd/mesh-config/ownerviews_test.go | 2 +- .../cmd/mesh-vault/assignmentsview.go | 53 +++++++------------ .../cmd/mesh-vault/assignmentsview_test.go | 5 +- modules/mesh-vault/cmd/mesh-vault/main.go | 9 +--- 6 files changed, 50 insertions(+), 80 deletions(-) diff --git a/modules/distribution/cmd/distribution/whole_integration_test.go b/modules/distribution/cmd/distribution/whole_integration_test.go index c60bb57bc..e0ee7b7da 100644 --- a/modules/distribution/cmd/distribution/whole_integration_test.go +++ b/modules/distribution/cmd/distribution/whole_integration_test.go @@ -210,9 +210,12 @@ func TestDistributionWhole(t *testing.T) { i, err := c.Info(ctx) return err == nil && i.NumPending == 0 && i.NumAckPending == 0 && i.Delivered.Consumer >= 2 }) - info, _ := s.Info(ctx) - if info.State.Msgs != 2 { - t.Errorf("distribution.events holds %d events, want old-name-closed and direct-writes-closed once each", info.State.Msgs) + // The stream holds the module's own events once each; the finding writer's (design 47 G A-30) are said beside them. + info, _ := s.Info(ctx, jetstream.WithSubjectFilter("mesh.stream.distribution.event.>")) + for _, ev := range []string{holder.OldNameClosedEvent, holder.DirectWritesClosedEvent} { + if n := info.State.Subjects["mesh.stream.distribution.event."+ev]; n != 1 { + t.Errorf("distribution.events holds %s %d times, want once", ev, n) + } } for _, ev := range []string{holder.OldNameClosedEvent, holder.DirectWritesClosedEvent} { if _, err := s.GetLastMsgForSubject(ctx, "mesh.stream.distribution.event."+ev); err != nil { diff --git a/modules/mesh-config/cmd/mesh-config/ownerviews.go b/modules/mesh-config/cmd/mesh-config/ownerviews.go index 2920b0c22..623739cda 100644 --- a/modules/mesh-config/cmd/mesh-config/ownerviews.go +++ b/modules/mesh-config/cmd/mesh-config/ownerviews.go @@ -18,7 +18,6 @@ import ( catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "sort" "strings" - "time" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/views" @@ -49,48 +48,37 @@ var followed = []struct{ Read, Owner, State, Tool string }{ } // ownerViews are the pictures, by the read name, followed through one views.Reader holding the module's writer (design 47 G -// A-28): each follower is adopted, so the reader raises and clears its findings and hears its owner's views event. +// A-28): the reader raises and clears the first views' findings and hears each owner's views event, and asks the first view +// again on a gap. type ownerViews struct { by map[string]*views.Follower reader *views.Reader } -// newOwnerViews makes a follower for each state and adopts it; changed is told when any picture changes. findings is the -// module's writer, which the reader writes the first views' findings through. +// newOwnerViews makes a follower for each state; changed is told when any picture changes. findings is the module's writer, +// which the reader writes the first views' findings through. func newOwnerViews(ask func(key string, body any) (json.RawMessage, error), changed func(), findings views.WireWriter) *ownerViews { - o := &ownerViews{by: map[string]*views.Follower{}, reader: &views.Reader{Module: Module, Findings: findings, Logf: logf}} + o := &ownerViews{by: map[string]*views.Follower{}} + o.reader = &views.Reader{Module: Module, Findings: findings, Logf: logf, + Changed: func(string, string, json.RawMessage, bool) { changed() }} for _, f := range followed { f := f - fol := &views.Follower{Owner: f.Owner, Tool: f.Owner + "." + f.Tool, State: f.State, - Ask: func(_ context.Context, after string) (views.Answer, error) { - body := map[string]any{"values": true} - if after != "" { - body["after"] = after - } - raw, err := ask(f.Owner+"."+f.Tool, body) - if err != nil { - return views.Answer{}, err - } - answer, output, ok := unwrapAnswer(raw) - if !ok { - return views.Answer{}, fmt.Errorf("%s refused: %s", f.Tool, output) - } - return views.DecodeAnswer(answer) - }, - Changed: func(string, json.RawMessage, bool, time.Time) { changed() }} - hear := func(ctx context.Context, body []byte) error { - ev, err := views.DecodeEvent(body) + fol := o.reader.Follow(f.Owner, f.State, f.Owner+"."+f.Tool) + // the owner's ordinary read tool answers wrapped as a tool's answer, or refused with its output + fol.Ask = func(_ context.Context, after string) (views.Answer, error) { + body := map[string]any{"values": true} + if after != "" { + body["after"] = after + } + raw, err := ask(f.Owner+"."+f.Tool, body) if err != nil { - return err + return views.Answer{}, err } - if gap, why := fol.Apply(ev); gap { - logf("[mesh-config] a gap in %s's %s (%s): its first view is asked once more", f.Owner, f.State, why) - go func() { _ = fol.Start(ctx) }() + answer, output, ok := unwrapAnswer(raw) + if !ok { + return views.Answer{}, fmt.Errorf("%s refused: %s", f.Tool, output) } - return nil - } - if err := o.reader.Adopt(fol, f.Owner, views.EventName(f.State), hear); err != nil { - logf("[mesh-config] %v", err) + return views.DecodeAnswer(answer) } o.by[f.Read] = fol } diff --git a/modules/mesh-config/cmd/mesh-config/ownerviews_test.go b/modules/mesh-config/cmd/mesh-config/ownerviews_test.go index e688d25bb..9b0579144 100644 --- a/modules/mesh-config/cmd/mesh-config/ownerviews_test.go +++ b/modules/mesh-config/cmd/mesh-config/ownerviews_test.go @@ -37,7 +37,7 @@ func TestAFollowedStateIsReadFromItsOwnersFirstViewAndEvents(t *testing.T) { t.Fatalf("before the first view: followed %v, %v", followed, err) } if !o.reader.Follows("mesh-delivery-assigner.wants") { - t.Error("the wants view is not one of the reader's (A-28: adopted)") + t.Error("the wants view is not one of the reader's (A-28)") } if _, followed, _ := o.get(BuildsRead, "x"); followed { t.Error("the builder's builds is followed, and it has no owner tool") diff --git a/modules/mesh-vault/cmd/mesh-vault/assignmentsview.go b/modules/mesh-vault/cmd/mesh-vault/assignmentsview.go index c9f177d1c..90e1094e4 100644 --- a/modules/mesh-vault/cmd/mesh-vault/assignmentsview.go +++ b/modules/mesh-vault/cmd/mesh-vault/assignmentsview.go @@ -12,7 +12,6 @@ import ( "fmt" "sort" "strings" - "time" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -36,20 +35,13 @@ var assignmentsView *views.Follower // newAssignmentsView makes the follower: ask calls the tool, gone is told of a deleted key, changed of every change after // it is held (a first view's keys too). -func newAssignmentsView(ask func(key string, body any) (json.RawMessage, error), gone func(key string), changed func()) *views.Follower { - return &views.Follower{Owner: assignerModule, Tool: AssignmentsTool, State: "assignments", Logf: say, - Ask: func(_ context.Context, after string) (views.Answer, error) { - body := map[string]any{"values": true} - if after != "" { - body["after"] = after - } - raw, err := ask(AssignmentsTool, body) - if err != nil { - return views.Answer{}, err - } - return views.DecodeAnswer(unwrapToolAnswer(raw)) - }, - Changed: func(key string, _ json.RawMessage, deleted bool, _ time.Time) { +func newAssignmentsView(ask func(key string, body any) (json.RawMessage, error), gone func(key string), changed func(), + findings views.WireWriter) (*views.Follower, *views.Reader) { + // A views.Reader holding the vault's writer follows it (design 47 G A-28), so the reader writes and clears its findings + // (past its bound the finding names the owner and the tool, and the answer clears it; no secret) and hears the owner's + // views event, asking again on a gap. + r := &views.Reader{Module: Seat, Findings: findings, Logf: say, + Changed: func(_, key string, _ json.RawMessage, deleted bool) { if deleted { gone(key) } @@ -57,6 +49,19 @@ func newAssignmentsView(ask func(key string, body any) (json.RawMessage, error), changed() } }} + f := r.Follow(assignerModule, "assignments", AssignmentsTool) + f.Ask = func(_ context.Context, after string) (views.Answer, error) { + body := map[string]any{"values": true} + if after != "" { + body["after"] = after + } + raw, err := ask(AssignmentsTool, body) + if err != nil { + return views.Answer{}, err + } + return views.DecodeAnswer(unwrapToolAnswer(raw)) + } + return f, r } // unwrapToolAnswer takes a tool's answer out of the protocol's content list when it comes in one. @@ -72,24 +77,6 @@ func unwrapToolAnswer(raw json.RawMessage) []byte { return raw } -// adoptAssignmentsView holds the follower in a views.Reader with the vault's writer (design 47 G A-28), so the reader writes -// and clears its findings (past its bound the finding names the owner and the tool, and the answer clears it; no secret). -func adoptAssignmentsView(f *views.Follower, findings views.WireWriter) (*views.Reader, error) { - r := &views.Reader{Module: Seat, Findings: findings, Logf: say} - hear := func(ctx context.Context, body []byte) error { - ev, err := views.DecodeEvent(body) - if err != nil { - return err - } - if gap, why := f.Apply(ev); gap { - say("a gap in the assigner's assignments (%s): its first view is asked once more", why) - go func() { _ = f.Start(ctx) }() - } - return nil - } - return r, r.Adopt(f, assignerModule, views.EventName("assignments"), hear) -} - // viewAssignments is the assignments reader over the picture: a key asked before the first view is answered is an error, so // the caller waits (the "not yet" of the vault's other reads), never a read of the owner's bucket. func viewAssignments(f func() *views.Follower) AssignerReader { diff --git a/modules/mesh-vault/cmd/mesh-vault/assignmentsview_test.go b/modules/mesh-vault/cmd/mesh-vault/assignmentsview_test.go index 415d09913..ae5118c49 100644 --- a/modules/mesh-vault/cmd/mesh-vault/assignmentsview_test.go +++ b/modules/mesh-vault/cmd/mesh-vault/assignmentsview_test.go @@ -29,10 +29,7 @@ func TestAnUnansweredFirstViewOfTheAssignmentsIsAFindingTheAnswerClears(t *testi Values: map[string]json.RawMessage{"g14.node": json.RawMessage(`{}`)}}) return raw, nil } - f := newAssignmentsView(ask, func(string) {}, nil) - if _, err := adoptAssignmentsView(f, heldFindings{&mu, held}); err != nil { - t.Fatal(err) - } + f, _ := newAssignmentsView(ask, func(string) {}, nil, heldFindings{&mu, held}) f.Bound, f.Wait = 30*time.Millisecond, 10*time.Millisecond reader := viewAssignments(func() *views.Follower { return f }) if _, err := reader.Keys(); err == nil { diff --git a/modules/mesh-vault/cmd/mesh-vault/main.go b/modules/mesh-vault/cmd/mesh-vault/main.go index 3a93bebf9..c113a05b9 100644 --- a/modules/mesh-vault/cmd/mesh-vault/main.go +++ b/modules/mesh-vault/cmd/mesh-vault/main.go @@ -132,16 +132,11 @@ func watchLeavers(env *Env) bool { if err != nil { return false } - f := newAssignmentsView(stdio.Ask, func(key string) { + f, r := newAssignmentsView(stdio.Ask, func(key string) { if node, ok := leaverKey(key); ok { env.Vault.NodeLeft(node) } - }, env.Vault.AssignmentsChanged) - r, err := adoptAssignmentsView(f, w) - if err != nil { - say("the assignments view is not adopted: %v", err) - return false - } + }, env.Vault.AssignmentsChanged, w) assignmentsView = f r.Start(context.Background()) }