G findings: mesh-config's and mesh-vault's own followers go through a views.Reader holding the writer, adopted
Builds: design 47 G A-28, A-26, A-22, §687 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -21,7 +21,6 @@ import (
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
)
|
||||
|
||||
@@ -49,19 +48,20 @@ var followed = []struct{ Read, Owner, State, Tool string }{
|
||||
{WantsRead, assignerModule, "wants", AssignerWantsTool},
|
||||
}
|
||||
|
||||
// ownerViews are the pictures, by the read name.
|
||||
// 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.
|
||||
type ownerViews struct {
|
||||
by map[string]*views.Follower
|
||||
// findings keeps the findings a first view unanswered writes.
|
||||
findings views.FindingWriter
|
||||
by map[string]*views.Follower
|
||||
reader *views.Reader
|
||||
}
|
||||
|
||||
// newOwnerViews makes a follower for each state; changed is told when any picture changes.
|
||||
func newOwnerViews(ask func(key string, body any) (json.RawMessage, error), changed func(), findings views.FindingWriter) *ownerViews {
|
||||
o := &ownerViews{by: map[string]*views.Follower{}, findings: findings}
|
||||
// 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.
|
||||
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}}
|
||||
for _, f := range followed {
|
||||
f := f
|
||||
fol := &views.Follower{Owner: f.Owner, Tool: f.Owner + "." + f.Tool, State: f.State, Logf: logf,
|
||||
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 != "" {
|
||||
@@ -78,54 +78,27 @@ func newOwnerViews(ask func(key string, body any) (json.RawMessage, error), chan
|
||||
return views.DecodeAnswer(answer)
|
||||
},
|
||||
Changed: func(string, json.RawMessage, bool, time.Time) { changed() }}
|
||||
fol.Raise = func(s views.Status) {
|
||||
_ = findings.Put(views.UnansweredFinding(Module, s, time.Now()))
|
||||
hear := func(ctx context.Context, body []byte) error {
|
||||
ev, err := views.DecodeEvent(body)
|
||||
if err != nil {
|
||||
return 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) }()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
fol.Clear = func(s views.Status) {
|
||||
_ = findings.Delete(views.UnansweredFinding(Module, s, time.Now()))
|
||||
if err := o.reader.Adopt(fol, f.Owner, views.EventName(f.State), hear); err != nil {
|
||||
logf("[mesh-config] %v", err)
|
||||
}
|
||||
o.by[f.Read] = fol
|
||||
}
|
||||
return o
|
||||
}
|
||||
|
||||
// Start asks every first view until answered (each on its goroutine) and subscribes to the owners' views events.
|
||||
func (o *ownerViews) Start(ctx context.Context) {
|
||||
for _, fol := range o.by {
|
||||
go func(f *views.Follower) {
|
||||
if err := f.Start(ctx); err != nil {
|
||||
logf("[mesh-config] the first view of %s's %s ended unanswered: %v", f.Owner, f.State, err)
|
||||
}
|
||||
}(fol)
|
||||
}
|
||||
for _, f := range followed {
|
||||
f := f
|
||||
pattern := f.Owner + "." + views.EventName(f.State)
|
||||
go func() {
|
||||
<-stdio.Served()
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
err := stdio.Subscribe(pattern, func(e stdio.Envelope) error {
|
||||
ev, err := views.DecodeEvent(e.Body)
|
||||
if err != nil {
|
||||
logf("[mesh-config] %s is not a views event: %v", pattern, err)
|
||||
return nil
|
||||
}
|
||||
fol := o.by[f.Read]
|
||||
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) }()
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
logf("[mesh-config] does not hear %s yet (%v); asking again in %s", pattern, err, wait)
|
||||
time.Sleep(wait)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
// Start subscribes the owners' views events and asks every first view until answered, each on its goroutine.
|
||||
func (o *ownerViews) Start(ctx context.Context) { o.reader.Start(ctx) }
|
||||
|
||||
// read is the picture of a read name, and whether the name is one followed.
|
||||
func (o *ownerViews) read(state string) (*views.Follower, bool) {
|
||||
|
||||
@@ -36,6 +36,9 @@ func TestAFollowedStateIsReadFromItsOwnersFirstViewAndEvents(t *testing.T) {
|
||||
if _, followed, err := o.get(WantsRead, "want.g14.dunst"); !followed || err == nil || !strings.Contains(err.Error(), "not answered yet") {
|
||||
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)")
|
||||
}
|
||||
if _, followed, _ := o.get(BuildsRead, "x"); followed {
|
||||
t.Error("the builder's builds is followed, and it has no owner tool")
|
||||
}
|
||||
@@ -78,5 +81,6 @@ func TestAFollowedStateIsReadFromItsOwnersFirstViewAndEvents(t *testing.T) {
|
||||
// nopFindings takes a finding and keeps none.
|
||||
type nopFindings struct{}
|
||||
|
||||
func (nopFindings) Put(views.Finding) error { return nil }
|
||||
func (nopFindings) Delete(views.Finding) error { return nil }
|
||||
func (nopFindings) WriteWire(context.Context, map[string]any) error { return nil }
|
||||
func (nopFindings) Clear(_ context.Context, _, _, _, _ string) error { return nil }
|
||||
func (nopFindings) ModuleNode() (string, string) { return Module, "" }
|
||||
|
||||
@@ -14,8 +14,6 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
)
|
||||
|
||||
@@ -38,8 +36,7 @@ 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(),
|
||||
findings views.FindingWriter) *views.Follower {
|
||||
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}
|
||||
@@ -52,19 +49,6 @@ func newAssignmentsView(ask func(key string, body any) (json.RawMessage, error),
|
||||
}
|
||||
return views.DecodeAnswer(unwrapToolAnswer(raw))
|
||||
},
|
||||
// Past its bound the vault's finding names the owner and the tool, and the condition keeper raises it (the
|
||||
// answer deletes it); the vault keeps asking meanwhile.
|
||||
Raise: func(s views.Status) {
|
||||
say("the vault has no first view of the assigner's assignments: %s has not answered since %s (%s); it keeps asking", s.Tool,
|
||||
s.Asked.Format(time.RFC3339), s.Error)
|
||||
f := views.UnansweredFinding(Seat, s, time.Now())
|
||||
if err := findings.Put(f); err != nil {
|
||||
say("the finding %s is not written: %v", f.ID, err)
|
||||
}
|
||||
},
|
||||
Clear: func(s views.Status) {
|
||||
_ = findings.Delete(views.UnansweredFinding(Seat, s, time.Now()))
|
||||
},
|
||||
Changed: func(key string, _ json.RawMessage, deleted bool, _ time.Time) {
|
||||
if deleted {
|
||||
gone(key)
|
||||
@@ -88,35 +72,22 @@ func unwrapToolAnswer(raw json.RawMessage) []byte {
|
||||
return raw
|
||||
}
|
||||
|
||||
// beginAssignmentsView asks the first view until answered and hears the owner's events.
|
||||
func beginAssignmentsView(f *views.Follower) {
|
||||
go func() {
|
||||
if err := f.Start(context.Background()); err != nil {
|
||||
say("the first view of the assigner's assignments ended unanswered: %v", err)
|
||||
// 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
|
||||
}
|
||||
}()
|
||||
go func() {
|
||||
<-stdio.Served()
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
err := stdio.Subscribe(AssignmentsViewsEvent, func(e stdio.Envelope) error {
|
||||
ev, err := views.DecodeEvent(e.Body)
|
||||
if err != nil {
|
||||
say("%s is not a views event: %v", AssignmentsViewsEvent, err)
|
||||
return nil
|
||||
}
|
||||
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(context.Background()) }()
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err == nil {
|
||||
return
|
||||
}
|
||||
say("does not hear %s yet (%v); asking again in %s", AssignmentsViewsEvent, err, wait)
|
||||
time.Sleep(wait)
|
||||
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
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -28,7 +29,10 @@ func TestAnUnansweredFirstViewOfTheAssignmentsIsAFindingTheAnswerClears(t *testi
|
||||
Values: map[string]json.RawMessage{"g14.node": json.RawMessage(`{}`)}})
|
||||
return raw, nil
|
||||
}
|
||||
f := newAssignmentsView(ask, func(string) {}, nil, heldFindings{&mu, held})
|
||||
f := newAssignmentsView(ask, func(string) {}, nil)
|
||||
if _, err := adoptAssignmentsView(f, heldFindings{&mu, held}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f.Bound, f.Wait = 30*time.Millisecond, 10*time.Millisecond
|
||||
reader := viewAssignments(func() *views.Follower { return f })
|
||||
if _, err := reader.Keys(); err == nil {
|
||||
@@ -62,15 +66,16 @@ type heldFindings struct {
|
||||
held map[string]json.RawMessage
|
||||
}
|
||||
|
||||
func (h heldFindings) Put(f views.Finding) error {
|
||||
func (h heldFindings) WriteWire(_ context.Context, f map[string]any) error {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
h.held[f.ID+"."+f.Kind] = json.RawMessage(f.Summary)
|
||||
h.held[fmt.Sprint(f["id"], ".", f["kind"])] = nil
|
||||
return nil
|
||||
}
|
||||
func (h heldFindings) Delete(f views.Finding) error {
|
||||
func (h heldFindings) Clear(_ context.Context, _, id, kind, _ string) error {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
delete(h.held, f.ID+"."+f.Kind)
|
||||
delete(h.held, id+"."+kind)
|
||||
return nil
|
||||
}
|
||||
func (h heldFindings) ModuleNode() (string, string) { return Seat, "" }
|
||||
|
||||
@@ -132,13 +132,18 @@ func watchLeavers(env *Env) bool {
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
assignmentsView = newAssignmentsView(stdio.Ask, func(key string) {
|
||||
f := newAssignmentsView(stdio.Ask, func(key string) {
|
||||
if node, ok := leaverKey(key); ok {
|
||||
env.Vault.NodeLeft(node)
|
||||
}
|
||||
}, env.Vault.AssignmentsChanged,
|
||||
w)
|
||||
beginAssignmentsView(assignmentsView)
|
||||
}, env.Vault.AssignmentsChanged)
|
||||
r, err := adoptAssignmentsView(f, w)
|
||||
if err != nil {
|
||||
say("the assignments view is not adopted: %v", err)
|
||||
return false
|
||||
}
|
||||
assignmentsView = f
|
||||
r.Start(context.Background())
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user