novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
Adopt subscribes a seat's event with further parts (<seat>.<event>.>), 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
This commit is contained in:
jschoubben
2026-11-11 22:22:24 +01:00
parent c81915b039
commit d03ffa8038
6 changed files with 50 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/distribution/cmd/distribution/whole_integration_test.go
+6 -3
View File
@@ -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 {
modules/mesh-config/cmd/mesh-config/ownerviews.go
+20 -32
View File
@@ -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
}
modules/mesh-config/cmd/mesh-config/ownerviews_test.go
+1 -1
View File
@@ -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")
modules/mesh-vault/cmd/mesh-vault/assignmentsview.go
+20 -33
View File
@@ -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 {
modules/mesh-vault/cmd/mesh-vault/assignmentsview_test.go
+1 -4
View File
@@ -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 {
modules/mesh-vault/cmd/mesh-vault/main.go
+2 -7
View File
@@ -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())
}