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

W16.6: the tool runner reads which modules exist and their manifests from the catalog half through meshcatalogclient's branch follower on its own consumer (milestone 47/G)

Browse Source 
firstview adopts the shared branch follower over a wiring of its own consumer (views held with the assigner's, findings through its
held writer, events routed to each adopted view); the console's Source gives Modules from the catalog entries, the manifest an entry
carries whole; the indexer's modules and manifests first views and their consumes are removed, the registrations and the gate's
source-moved are consumed.

Builds: design 47 G §196, §594, §614; lead answers A-5, A-27, A-103

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-11 23:40:23 +01:00
parent 4bfae38a73
commit 3501164f4b
9 changed files with 333 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node-tools/internal/firstview/firstview.go
+204 -26
View File
@@ -1,5 +1,7 @@
// Package firstview is the tool runner's picture of the four states it reads of the indexer and the assigner: the assigner's
// assignments and seat-holders, the indexer's manifests and modules (novox/hq ADR 0615 §1, §3; ADR 0621 §2; issue 1966 TLS-5).
// Package firstview is the tool runner's picture of what it reads of the assigner and the catalog half: the assigner's
// assignments and seat-holders (novox/hq ADR 0615 §1, §3; ADR 0621 §2; issue 1966 TLS-5), and which modules exist with the
// manifest each entry carries, followed per watched branch through meshcatalogclient's branch follower on this process's own
// consumer (design 47 G §196, §594, §614; lead answers A-5, A-27, A-103), never the indexer's state `modules` or `manifests`.
//
// **It never reads the owners' state.** At start it asks each owner's first-view tool until answered (the owner answers in a
// runtime of its own, never through this one: ADR 0621 §1), then follows the owner's views family, one event per changed key;
@@ -12,12 +14,15 @@ import (
"context"
"encoding/json"
"fmt"
"slices"
"sort"
"strings"
"sync"
"time"
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/keys"
@@ -26,40 +31,37 @@ import (
"github.com/novox/mesh-tools/node-tools/internal/bus"
)
// The owners and their first-view tools (the tools' addresses are `<owner>.<tool>`).
const (
Indexer = "mesh-delivery-indexer"
Assigner = "mesh-delivery-assigner"
)
// The owner and its first-view tools (the tools' addresses are `<owner>.<tool>`).
const Assigner = "mesh-delivery-assigner"
// The names the console reads the four states by (`<owner>.<state>`), as the manifest's `consumes` names their views
// The names the console reads the assigner's two states by (`<owner>.<state>`), as the manifest's `consumes` names their views
// families (no seat's client names a holder, novox/hq ADR 0625 §4; TestTheFirstViewsAreTheManifestsViews).
const (
AssignmentsRead = Assigner + ".assignments"
SeatHoldersRead = Assigner + ".seat-holders"
ModulesRead = Indexer + ".modules"
ManifestsRead = Indexer + ".manifests"
)
// Spec is one first view the runner holds.
type Spec struct{ Owner, Tool, State string }
// Specs are the four.
// Specs are the two owner states asked of an owner's tool; the catalog half's views are adopted by the branch follower.
var Specs = []Spec{
{Assigner, "assignments", "assignments"},
{Assigner, "seat-holders", "seat-holders"},
{Indexer, "manifests", "manifests"},
{Indexer, "modules", "modules"},
}
// Name is the name a state is read by, `<owner>.<state>`: AssignmentsRead, SeatHoldersRead and the indexer's.
// Name is the name a state is read by, `<owner>.<state>`: AssignmentsRead and SeatHoldersRead.
func (s Spec) Name() string { return s.Owner + "." + s.State }
// Views is the four followers.
// Views is the first views: the assigner's two, and the catalog half's per watched branch.
type Views struct {
logf func(string, ...any)
followers map[string]*views.Follower // by Spec.Name()
byState map[string]*views.Follower // by the state an event names
mu sync.Mutex
followers map[string]*views.Follower // by `<owner>.<state>`
byState map[string]*views.Follower // by the state an event names, the owners' views families
adopted map[string]*adoption // by `<owner>.<state>`, the views the branch follower adopted
hearers map[string][]*adoption // event key `<seat>.<event>` -> the adopted views that hear it
branches *meshcatalogclient.Branches
ctx context.Context
cancel context.CancelFunc
wg sync.WaitGroup
@@ -94,7 +96,7 @@ type FindingWriter interface {
Clear(ctx context.Context, scope, id, kind, why string) error
}
// Start asks the four first views, each until answered, and follows the owners' views families on the runner's own
// Start asks the first views, each until answered, and follows the owners' views families on the runner's own
// consumers. It returns at once: the picture fills as the owners answer.
func Start(parent context.Context, conn *bus.Conn, logf func(string, ...any)) (*Views, func()) {
return StartWith(parent, DepsOf(conn, logf), logf)
@@ -159,14 +161,19 @@ func (s findingsState) Keys() ([]string, error) { return s.conn.StateKeys(s.modu
// StartWith is Start over deps.
func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*Views, func()) {
ctx, cancel := context.WithCancel(parent)
v := &Views{logf: logf, ctx: ctx, cancel: cancel, followers: map[string]*views.Follower{}, byState: map[string]*views.Follower{}, deps: d}
v := &Views{logf: logf, ctx: ctx, cancel: cancel, followers: map[string]*views.Follower{}, byState: map[string]*views.Follower{},
adopted: map[string]*adoption{}, hearers: map[string][]*adoption{}, deps: d}
for _, s := range Specs {
s := s
call := func(tool string, body any) (json.RawMessage, error) { return d.Ask(s.Owner+"."+tool, body) }
f := &views.Follower{Owner: s.Owner, Tool: s.Tool, State: s.State, Ask: views.AskVia(call, s.Tool), Logf: logf,
Bound: d.Bound, Wait: d.Wait,
Raise: func(st views.Status) { v.finding(st, true) },
Clear: func(st views.Status) { v.finding(st, false) }}
Raise: func(st views.Status) {
v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, true)
},
Clear: func(st views.Status) {
v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false)
}}
v.followers[s.Name()] = f
v.byState[s.State] = f
v.wg.Add(1)
@@ -180,6 +187,15 @@ func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*View
stopFindings = stop
}
}
// The catalog half is followed by the shared branch follower on this process's own wiring (design 47 G §614, A-103): the
// registrations, the gate's source and a view per watched branch, each heard on this consumer.
branches, err := meshcatalogclient.NewBranches(wiring{v}, func(key string, body any) (json.RawMessage, error) { return d.Ask(key, body) })
if err != nil {
logf("[mesh-tools] the catalog half is not followed (%v): the console cannot say which modules exist", err)
} else {
branches.Logf = func(format string, args ...any) { logf("[mesh-tools] "+format, args...) }
v.branches = branches
}
stop, err := d.Consume(v.hear)
if err != nil {
logf("[mesh-tools] the owners' views are not followed (%v): the pictures are the first views alone until the runner starts again", err)
@@ -190,6 +206,17 @@ func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*View
// hear applies one views event; a gap asks the first view once more.
func (v *Views) hear(env bus.Envelope) error {
v.mu.Lock()
hs := append([]*adoption(nil), v.hearers[env.Key]...)
v.mu.Unlock()
if len(hs) > 0 { // an event of the catalog half or the gate's source, heard by the views the branch follower adopted
for _, a := range hs {
if err := a.hear(v.ctx, env.Body); err != nil {
v.logf("[mesh-tools] %v", err)
}
}
return nil
}
ev, err := views.DecodeEvent(env.Body)
if err != nil {
return nil // not a views event: nothing to hold, nothing to retry
@@ -206,8 +233,7 @@ func (v *Views) hear(env bus.Envelope) error {
}
// finding writes or clears the runner's finding of a first view unanswered past its bound, through its writer (design 47 G A-28).
func (v *Views) finding(st views.Status, raised bool) {
f := unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now())
func (v *Views) finding(f condseat.Finding, tool string, raised bool) {
if v.deps.Findings == nil {
v.logf("[mesh-tools] %s", f.Summary)
return
@@ -219,14 +245,16 @@ func (v *Views) finding(st views.Status, raised bool) {
err = v.deps.Findings.Clear(context.Background(), f.Scope, f.ID, f.Kind, "the owner answered")
}
if err != nil {
v.logf("[mesh-tools] the finding of the first view %s is not kept: %v", st.Tool, err)
v.logf("[mesh-tools] the finding of the first view %s is not kept: %v", tool, err)
}
}
// State is the picture of the state spec names (`<owner>.<state>`), every current value by key; an error says the owner has
// not answered yet, so nothing is claimed of it.
func (v *Views) State(name string) (map[string][]byte, error) {
v.mu.Lock()
f := v.followers[name]
v.mu.Unlock()
if f == nil {
return nil, fmt.Errorf("the tool runner holds no first view of %s", name)
}
@@ -254,17 +282,28 @@ func (v *Views) Entry(name, key string) ([]byte, error) {
// Status is each first view as it stands, by `<owner>.<state>`, sorted.
func (v *Views) Status() []views.Status {
var out []views.Status
for _, f := range v.followers {
for _, f := range v.all() {
out = append(out, f.Status())
}
sort.Slice(out, func(i, j int) bool { return out[i].Owner+out[i].State < out[j].Owner+out[j].State })
return out
}
// all is every first view held now: the owners', then the adopted ones.
func (v *Views) all() []*views.Follower {
v.mu.Lock()
defer v.mu.Unlock()
out := make([]*views.Follower, 0, len(v.followers))
for _, f := range v.followers {
out = append(out, f)
}
return out
}
// Again asks every first view once more, by hand, for why.
func (v *Views) Again(ctx context.Context, why string) []error {
var errs []error
for _, f := range v.followers {
for _, f := range v.all() {
if err := f.Again(ctx, why); err != nil {
errs = append(errs, err)
}
@@ -285,3 +324,142 @@ func unansweredFinding(module, node string, st views.Status, at time.Time) conds
// scopeModule is the scope of a finding about a module.
const scopeModule = "module"
// viewWaitsFinding is the finding of a view waited for past views.WaitsBound, in the contract's form: about the runner's module,
// a warning, its id the view's name escaped as a key's name is.
func viewWaitsFinding(module, node string, st views.Status, at time.Time) condseat.Finding {
return condseat.Finding{Kind: views.ViewWaitsKind, Scope: scopeModule, ID: keys.EscapeName(module + "." + st.Owner + "." + st.State),
Severity: condseat.FindingSeverityWarning, Source: "views", Observer: module,
Subject: condseat.Subject{Scope: scopeModule, ID: module, Machine: node},
Summary: fmt.Sprintf("%s waits for the view %s.%s at revision %s since %s and acts on nothing new until it holds it; it holds %s",
module, st.Owner, st.State, st.WaitsFor, st.WaitingSince.UTC().Format(time.RFC3339), orNone(st.View)),
Headline: fmt.Sprintf("%s waits for a view of %s", module, st.Owner), At: at.UTC()}
}
func orNone(s string) string {
if s == "" {
return "none"
}
return s
}
// adoption is a view the branch follower put on this process's consumer: its follower, the event it hears and its handler.
type adoption struct {
name, key string
f *views.Follower
hear func(context.Context, []byte) error
cancel context.CancelFunc
}
// wiring is the branch follower's wiring over this process's own consumer (meshcatalogclient.Wiring; design 47 G §614, A-103):
// an adopted view is held with the owners', takes its first view at once, hears its seat's event on the runner's consumer,
// writes its findings through the runner's held writer, and a dropped one clears them.
type wiring struct{ v *Views }
func (w wiring) Adopt(f *views.Follower, seat, event string, hear func(context.Context, []byte) error) error {
v := w.v
if f == nil || f.Owner == "" || f.State == "" || seat == "" || event == "" || hear == nil {
return fmt.Errorf("the tool runner adopts a view that names its owner, state, seat's event and handler")
}
name := f.Owner + "." + f.State
key := seat + "." + event
v.mu.Lock()
if v.followers[name] != nil {
v.mu.Unlock()
return fmt.Errorf("%s is followed already", name)
}
if f.Logf == nil {
f.Logf = v.logf
}
if f.Bound == 0 {
f.Bound = v.deps.Bound
}
if f.Wait == 0 {
f.Wait = v.deps.Wait
}
held := func() bool { v.mu.Lock(); defer v.mu.Unlock(); return v.followers[name] == f }
f.Raise = func(st views.Status) {
if held() {
v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, true)
}
}
f.Clear = func(st views.Status) {
v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false)
}
f.RaiseWaits = func(st views.Status) {
if held() {
v.finding(viewWaitsFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, true)
}
}
f.ClearWaits = func(st views.Status) {
v.finding(viewWaitsFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false)
}
ctx, cancel := context.WithCancel(v.ctx)
a := &adoption{name: name, key: key, f: f, hear: hear, cancel: cancel}
v.followers[name] = f
v.adopted[name] = a
v.hearers[key] = append(v.hearers[key], a)
v.mu.Unlock()
v.wg.Add(1)
go func() { defer v.wg.Done(); _ = f.Start(ctx) }()
return nil
}
func (w wiring) Drop(owner, state string) error {
v := w.v
name := owner + "." + state
v.mu.Lock()
a := v.adopted[name]
if a == nil {
v.mu.Unlock()
return fmt.Errorf("%s is not a view this tool runner adopted", name)
}
delete(v.followers, name)
delete(v.adopted, name)
v.hearers[a.key] = slices.DeleteFunc(v.hearers[a.key], func(o *adoption) bool { return o == a })
v.mu.Unlock()
a.cancel()
st := a.f.Status()
if st.Raised {
v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false)
}
if st.WaitsRaised {
v.finding(viewWaitsFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false)
}
return nil
}
// Modules is every module the catalog half holds, by name, with its entry: the entries of the watched branches' newest trunk
// revisions, the branch followed first naming a module more than one hold (design 47 G §196, §594). An error says the half has
// no first view yet, never that no module exists.
func (v *Views) Modules() (map[string]catalogseat.CatalogEntry, error) {
if v.branches == nil {
return nil, fmt.Errorf("the tool runner does not follow the catalog half")
}
followed := v.branches.Followed()
if len(followed) == 0 {
return nil, fmt.Errorf("the catalog half names no watched branch yet (the registrations of %s are asked until they answer)", catalogseat.Seat)
}
out := map[string]catalogseat.CatalogEntry{}
var waiting []string
for _, branch := range followed {
c, ok := v.branches.Catalog(branch)
if !ok {
continue
}
half, held := c.View()
if !held {
waiting = append(waiting, branch)
continue
}
for m, e := range half.Entries {
if _, taken := out[m]; !taken {
out[m] = e
}
}
}
if len(waiting) > 0 {
return nil, fmt.Errorf("the catalog half has no first view of %s yet", strings.Join(waiting, ", "))
}
return out, nil
}