From 3501164f4b40c9350545283a95bc1f6697534092 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 23:40:23 +0100 Subject: [PATCH] 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) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- node-tools/internal/console/address.go | 76 ++---- node-tools/internal/console/firstviews.go | 4 +- .../internal/console/manifest_waits_test.go | 14 +- node-tools/internal/console/source_test.go | 21 ++ node-tools/internal/firstview/firstview.go | 230 ++++++++++++++++-- .../firstview/firstview_integration_test.go | 14 +- .../internal/firstview/firstview_test.go | 69 +++++- node-tools/internal/meshtest/meshtest.go | 32 +-- node-tools/module.json | 4 +- 9 files changed, 333 insertions(+), 131 deletions(-) diff --git a/node-tools/internal/console/address.go b/node-tools/internal/console/address.go index cf7ed7e..b8860a7 100644 --- a/node-tools/internal/console/address.go +++ b/node-tools/internal/console/address.go @@ -19,7 +19,7 @@ package console import ( assigner "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" - meshcatalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-sdk/go/catalogue" "context" @@ -95,8 +95,7 @@ func discovery() []map[string]any { "bus, and when it was last heard — and every runtime or machine expected that was not heard. Read-only; " + "for when an address is said to be missing or on another machine."}, {"name": verbFirstViews, "inputSchema": obj(map[string]any{}), - "description": "The tool runner's first views 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): for each, the owner, the tool " + + "description": "The tool runner's first views: the assigner's assignments and seat-holders, and the catalog half's registrations, the gate's source and the view of each watched branch (which modules exist, each with its manifest): for each, the owner, the tool " + "asked, the revision held and when it was made, how many keys, when it was asked and answered, how often " + "it was asked and how often by hand, whether its condition is raised, and why it was last asked. Read-only."}, {"name": verbFirstAgain, "inputSchema": obj(map[string]any{ @@ -359,14 +358,14 @@ func indexOn(conn *bus.Conn, src Source, held announce.Known, before func(announ return x, nil } -// Source is where the console reads the four states of the indexer and the assigner it indexes the mesh by: the assigner's -// assignments and seat-holders, the indexer's modules and manifests, by `.` (firstview.AssignmentsRead and the -// rest). The tool runner's is its first views (firstview.Views), never the owners' state (novox/hq ADR 0615 §1). +// Source is where the console reads what it indexes the mesh by: the assigner's assignments and seat-holders, by +// `.` (firstview.AssignmentsRead and SeatHoldersRead), and the modules of the catalog half with their entries. +// The tool runner's is its first views (firstview.Views), never the owners' state (novox/hq ADR 0615 §1; design 47 G §196, §594). type Source interface { // State is every current value of a state by key; an error says why it cannot be told yet. State(name string) (map[string][]byte, error) - // Entry is one value of a state, nil when it holds none. - Entry(name, key string) ([]byte, error) + // Modules is every module the catalog half holds, by name, with its entry; an error says why it cannot be told yet. + Modules() (map[string]meshcatalog.CatalogEntry, error) } // seatHoldersOn reads the state seat-holders whole (novox/hq ADR 0571 clause 4), as firstview.SeatHoldersRead. @@ -374,22 +373,12 @@ func seatHoldersOn(src Source) (map[string][]byte, error) { return src.State(firstview.SeatHoldersRead) } -// The indexer's states the console reads (novox/hq ADR 0571 clauses 4 and 5): every module, and each manifest by its -// digest, by the names a reader's manifest declares. -var ( - modulesRead = firstview.ModulesRead - manifestsRead = firstview.ManifestsRead -) - -// manifestFacts is what the console takes of a manifest: the tools it declares and what each replaces. A manifest's -// digest never names another, so what is read once is kept. +// manifestFacts is what the console takes of a manifest: the tools it declares and what each replaces. type manifestFacts struct { Tools []string `json:"tools"` Replaces map[string][]string `json:"replaces"` } -var manifestsKept sync.Map // digest → manifestFacts - // recordsOn is every module the indexer holds, with where the assigner seat places it and whether it declares tools, // and every machine the assignments name (moduleNames, assigner.MachineNames). What cannot be read is said, one line // per state, for the listing's not answering. @@ -411,18 +400,19 @@ func recordsOn(src Source) (recorded []recordedModule, machines []string, why [] on[m.Module] = append(on[m.Module], node) } } - modules, err := src.State(modulesRead) + entries, err := src.Modules() if err != nil { - return nil, machines, append(why, modulesRead+" (which modules there are could not be read: "+err.Error()+")") + return nil, machines, append(why, "the catalog half's modules (which modules there are could not be read: "+err.Error()+")") } - names, err := moduleNames(modules) - if err != nil { - return nil, machines, append(why, modulesRead+" ("+err.Error()+")") + names := make([]string, 0, len(entries)) + for name := range entries { + names = append(names, name) } + sort.Strings(names) for _, name := range names { r := recordedModule{Module: name, On: on[name]} - if m, err := meshcatalogclient.DecodeModule(modules[name]); err == nil && m.Manifest != "" { - facts, err := manifestOf(src, m.Manifest) + if m := entries[name].Manifest; len(m) > 0 { + facts, err := manifestOf(m) if err != nil { why = append(why, name+" (its manifest could not be read, so whether it answers tools is not known: "+err.Error()+")") } else { @@ -434,42 +424,16 @@ func recordsOn(src Source) (recorded []recordedModule, machines []string, why [] return recorded, machines, why } -// moduleNames is every module the indexer holds, sorted, from its state `modules` as a reader holds its keys (key to -// value; a deleted key absent). A key that is not a module's is refused, naming it. -func moduleNames(keys map[string][]byte) ([]string, error) { - out := make([]string, 0, len(keys)) - for k := range keys { - module, err := meshcatalogclient.ParseModulesKey(k) - if err != nil { - return nil, err - } - out = append(out, module) - } - sort.Strings(out) - return out, nil -} - -// manifestOf is what the console takes of the manifest of a digest, read from the indexer's manifests once. -func manifestOf(src Source, digest string) (manifestFacts, error) { - if kept, ok := manifestsKept.Load(digest); ok { - return kept.(manifestFacts), nil - } - entry, err := src.Entry(manifestsRead, digest) - if err != nil { - return manifestFacts{}, err - } - if entry == nil { - return manifestFacts{}, fmt.Errorf("the indexer holds no manifest %s", digest) - } - m, err := meshcatalogclient.DecodeManifest(entry) +// manifestOf is what the console takes of the manifest an entry of the catalog half carries whole (design 47 G §443). +func manifestOf(manifest map[string]any) (manifestFacts, error) { + raw, err := json.Marshal(manifest) if err != nil { return manifestFacts{}, err } var facts manifestFacts - if err := json.Unmarshal([]byte(m.Manifest), &facts); err != nil { + if err := json.Unmarshal(raw, &facts); err != nil { return manifestFacts{}, err } - manifestsKept.Store(digest, facts) return facts, nil } diff --git a/node-tools/internal/console/firstviews.go b/node-tools/internal/console/firstviews.go index 0bb995c..0a7b05d 100644 --- a/node-tools/internal/console/firstviews.go +++ b/node-tools/internal/console/firstviews.go @@ -32,8 +32,8 @@ func (s *Surface) firstViewsAnswer(ctx context.Context, name string, args map[st } // A hand act, recorded through the seat mesh-hand-acts (novox/hq ADR 0615 §6, ADR 0565 §10) before it is done, and in the // runner's log. One that cannot be recorded is said, and the ask is not made. - log.Printf("[mesh-tools] hand act: %s asked the four first views again: %s", s.who, why) - if err := s.recordAct(map[string]any{"what": "asked the tool runner's four first views again", "why": why, + log.Printf("[mesh-tools] hand act: %s asked the first views again: %s", s.who, why) + if err := s.recordAct(map[string]any{"what": "asked the tool runner's first views again", "why": why, "cause": "first-view-again", "verb": verbFirstAgain}); err != nil { return failure("the hand act was not recorded through mesh-hand-acts, so the first views were not asked again: " + err.Error()) } diff --git a/node-tools/internal/console/manifest_waits_test.go b/node-tools/internal/console/manifest_waits_test.go index bce0e7f..f80ec5f 100644 --- a/node-tools/internal/console/manifest_waits_test.go +++ b/node-tools/internal/console/manifest_waits_test.go @@ -43,13 +43,13 @@ func TestTheManifestNamesTheFirstFamiliesOneByOne(t *testing.T) { t.Errorf("a family named by a wildcard: %s", w) } } - // A wait is no consumer (ADR 0367 §2): what it consumes is the four views families of its first views (ADR 0615 §5, issue - // 1966 TLS-5) and the two events of the seat that serves the seat list (ADR 0636 §4), and nothing else. - // Design 47 G §196, §387: the catalog half by `catalog-revision` (never the indexer's `views.modules`), and the keeper's - // `findings-asked`; the finding stream below. `views.modules` goes with the first view that follows it (T-2). - wantConsumes := []string{"mesh-catalog.catalog-revision", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", - "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.seat-holders", - "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules"} + // A wait is no consumer (ADR 0367 §2): what it consumes is the assigner's two views families (ADR 0615 §5, issue 1966 + // TLS-5), the events of the catalog half's branch follower (design 47 G §196, §614: `catalog-revision`, + // `registration-changed`, and the gate's `source-moved`; never the indexer's `views.modules` or `views.manifests`), the two + // events of the seat that serves the seat list (ADR 0636 §4) and the keeper's `findings-asked`, and nothing else. + wantConsumes := []string{"mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", + "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", + "mesh-delivery-assigner.views.seat-holders", "mesh-delivery.source-moved"} if !reflect.DeepEqual(m.Consumes, wantConsumes) { t.Errorf("the tool runner consumes %v, want %v", m.Consumes, wantConsumes) } diff --git a/node-tools/internal/console/source_test.go b/node-tools/internal/console/source_test.go index a3fed70..027117f 100644 --- a/node-tools/internal/console/source_test.go +++ b/node-tools/internal/console/source_test.go @@ -1,9 +1,13 @@ package console import ( + "encoding/json" "sync" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) // busSource is a Source over a bus connection's own states, which a test's fixture mesh fills: the console's tests stand a @@ -37,3 +41,20 @@ func (b busSource) Entry(name, key string) ([]byte, error) { } return entry.Value, nil } + +// Modules reads the test bed's entries of the catalog half (meshtest.ModulesRead), as the tool runner's first views hold them. +func (b busSource) Modules() (map[string]meshcatalog.CatalogEntry, error) { + raw, err := b.State(mt.ModulesRead) + if err != nil { + return nil, err + } + out := map[string]meshcatalog.CatalogEntry{} + for module, value := range raw { + var e meshcatalog.CatalogEntry + if err := json.Unmarshal(value, &e); err != nil { + return nil, err + } + out[module] = e + } + return out, nil +} diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go index 0ba4efb..77b7427 100644 --- a/node-tools/internal/firstview/firstview.go +++ b/node-tools/internal/firstview/firstview.go @@ -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 `.`). -const ( - Indexer = "mesh-delivery-indexer" - Assigner = "mesh-delivery-assigner" -) +// The owner and its first-view tools (the tools' addresses are `.`). +const Assigner = "mesh-delivery-assigner" -// The names the console reads the four states by (`.`), as the manifest's `consumes` names their views +// The names the console reads the assigner's two states by (`.`), 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, `.`: AssignmentsRead, SeatHoldersRead and the indexer's. +// Name is the name a state is read by, `.`: 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 `.` + byState map[string]*views.Follower // by the state an event names, the owners' views families + adopted map[string]*adoption // by `.`, the views the branch follower adopted + hearers map[string][]*adoption // event key `.` -> 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 (`.`), 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 `.`, 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 +} diff --git a/node-tools/internal/firstview/firstview_integration_test.go b/node-tools/internal/firstview/firstview_integration_test.go index 08db388..c9fadcd 100644 --- a/node-tools/internal/firstview/firstview_integration_test.go +++ b/node-tools/internal/firstview/firstview_integration_test.go @@ -41,12 +41,12 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen if err := json.Unmarshal(raw, &manifest); err != nil { t.Fatal(err) } - streamOf := map[string]string{Assigner: catalogue.StreamBusName("assigner-views"), Indexer: catalogue.StreamBusName("indexer-views")} + streamOf := map[string]string{Assigner: catalogue.StreamBusName("assigner-views")} filters := map[string][]string{} families := 0 for _, c := range manifest.Consumes { emitter, event, isView := strings.Cut(c, ".views.") - if !isView { // the seat mesh-catalog's events: the seat list's, on EVENTS (cmd/node-tools/seatlist.go) + if !isView { // the seats' events (the seat list's, the catalog half's), on EVENTS (cmd/node-tools/seatlist.go) continue } stream := streamOf[emitter] @@ -60,7 +60,7 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen t.Fatalf("the manifest consumes %d families, the runner holds %d first views", families, len(Specs)) } runner := mt.MembershipOf("node-tools", "anchor", false, nil) - for _, owner := range []string{Assigner, Indexer} { + for _, owner := range []string{Assigner} { stream := streamOf[owner] mesh.ModuleStream(t, owner, stream) mesh.ConsumerOn(t, stream, "anchor", "node-tools", filters[stream], 2*time.Second) @@ -99,7 +99,7 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen mt.Until(t, func() error { mu.Lock() defer mu.Unlock() - if len(findings) != len(Specs) { + if len(findings) != len(Specs)+catalogViews { return errFindings(len(findings)) } return nil @@ -130,7 +130,7 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen mt.Until(t, func() error { mu.Lock() defer mu.Unlock() - if len(findings) != 0 { + if len(findings) != catalogViews { // the seats' verbs are not served in this test: the catalog half stays unanswered return errFindings(len(findings)) } return nil @@ -154,6 +154,10 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen } } +// catalogViews is the first views the branch follower adopts before any branch is known: the registrations and the gate's source +// (design 47 G §614), each a finding while the seats' verbs go unanswered. +const catalogViews = 2 + type errFindings int func (e errFindings) Error() string { return "not yet" } diff --git a/node-tools/internal/firstview/firstview_test.go b/node-tools/internal/firstview/firstview_test.go index e3f5375..997203b 100644 --- a/node-tools/internal/firstview/firstview_test.go +++ b/node-tools/internal/firstview/firstview_test.go @@ -12,6 +12,9 @@ import ( "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" + delivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-sdk/go/views" "github.com/novox/mesh-tools/node-tools/internal/bus" @@ -26,12 +29,15 @@ type owners struct { asked map[string]int deliver func(bus.Envelope) error finds map[string]bool + // catalog is the revision of the watched branch main the seat mesh-catalog answers, and modules its entries + catalog int64 + modules []string } func newOwners() *owners { - return &owners{state: map[string]map[string]string{"assignments": {"anchor": `{"p":1}`}, "seat-holders": {"seat": `{"h":1}`}, - "manifests": {"d1": `{"m":1}`}, "modules": {"web": `{"w":1}`}}, - rev: map[string]uint64{"assignments": 4, "seat-holders": 2, "manifests": 7, "modules": 3}, asked: map[string]int{}, finds: map[string]bool{}} + return &owners{state: map[string]map[string]string{"assignments": {"anchor": `{"p":1}`}, "seat-holders": {"seat": `{"h":1}`}}, + rev: map[string]uint64{"assignments": 4, "seat-holders": 2}, asked: map[string]int{}, finds: map[string]bool{}, + catalog: 3, modules: []string{"web"}} } // Write and Clear are the runner's finding writer in a test: the findings written and not cleared, by their key. @@ -57,6 +63,18 @@ func (o *owners) ask(address string, _ any) (json.RawMessage, error) { if o.down { return nil, errors.New("503 no responders") } + switch address { + case "seat:mesh-delivery.source": + return json.Marshal(delivery.SourceAnswer{Branches: []string{"main"}}) + case "seat:mesh-catalog.registrations": + return json.Marshal(catalogseat.RegistrationsAnswer{Registrations: []catalogseat.Registration{}}) + case "seat:mesh-catalog.revision": + a := catalogseat.RevisionAnswer{Revision: &catalogseat.CatalogRevision{Revision: o.catalog, Kind: meshcatalogclient.KindTrunk, Branch: "main"}} + for _, m := range o.modules { + a.Entries = append(a.Entries, catalogseat.CatalogEntry{Module: m, Manifest: map[string]any{"module": m, "tools": []string{m + "_check"}}}) + } + return json.Marshal(a) + } vals := map[string]json.RawMessage{} for k, v := range o.state[tool] { vals[k] = json.RawMessage(v) @@ -118,15 +136,29 @@ func TestTheRunnerHoldsTheFourStatesFromTheOwnersFirstViewsAndTheirEvents(t *tes o.mu.Lock() o.down = false o.mu.Unlock() - eventually(t, "not answered", func() bool { _, err := v.State("mesh-delivery-indexer.manifests"); return err == nil }) + eventually(t, "not answered", func() bool { _, err := v.State("mesh-delivery-assigner.seat-holders"); return err == nil }) eventually(t, "findings not cleared", func() bool { o.mu.Lock(); defer o.mu.Unlock(); return len(o.finds) == 0 }) - if got, _ := v.Entry("mesh-delivery-indexer.modules", "web"); string(got) != `{"w":1}` { - t.Errorf("the modules picture holds %q", got) - } - if got, _ := v.Entry("mesh-delivery-indexer.modules", "none"); got != nil { + if got, _ := v.Entry("mesh-delivery-assigner.seat-holders", "none"); got != nil { t.Errorf("an absent key is %q", got) } + // The catalog half is followed by the shared branch follower on the runner's own consumer: the gate's source names main, + // main's view is adopted, and its entries are the modules, each with its manifest (design 47 G §196, §614). + eventually(t, "the catalog half holds no modules", func() bool { m, err := v.Modules(); return err == nil && len(m) == 1 }) + m, _ := v.Modules() + if m["web"].Manifest["module"] != "web" { + t.Errorf("the module web is %+v", m["web"]) + } + // A catalog-revision of the branch is heard on the same consumer and asks the view once more. + o.mu.Lock() + o.catalog, o.modules = 4, []string{"web", "db"} + o.mu.Unlock() + ev, _ := json.Marshal(catalogseat.CatalogRevisionEvent{V: 1, Branch: "main", Kind: meshcatalogclient.KindTrunk, Revision: 4}) + if err := o.deliver(bus.Envelope{Key: catalogseat.Seat + "." + catalogseat.EventCatalogRevision, Body: ev}); err != nil { + t.Fatal(err) + } + eventually(t, "the catalog-revision was not taken", func() bool { m, err := v.Modules(); return err == nil && len(m) == 2 }) + // Events: the next revision is applied; an old one passed over; a deletion removes; none of it asks again. asked := func(tool string) int { o.mu.Lock(); defer o.mu.Unlock(); return o.asked[tool] } before := asked("assignments") @@ -155,15 +187,15 @@ func TestTheRunnerHoldsTheFourStatesFromTheOwnersFirstViewsAndTheirEvents(t *tes if assignments.Owner != "mesh-delivery-assigner" || assignments.Tool != "assignments" || assignments.Asks < 2 || assignments.Answered.IsZero() { t.Errorf("the status of assignments is %+v", assignments) } - manifests := asked("manifests") + manifests := asked("seat-holders") if errs := v.Again(context.Background(), "doubted"); len(errs) != 0 { t.Fatalf("the hand ask failed: %v", errs) } - if asked("manifests") != manifests+1 { - t.Errorf("the hand ask did not ask the manifests again") + if asked("seat-holders") != manifests+1 { + t.Errorf("the hand ask did not ask the seat-holders again") } for _, st := range v.Status() { - if st.State == "manifests" && st.ByHand != 1 { + if st.State == "seat-holders" && st.ByHand != 1 { t.Errorf("the hand ask is not counted: %+v", st) } } @@ -186,7 +218,7 @@ func TestTheFirstViewsAreTheManifestsViews(t *testing.T) { if err := json.Unmarshal(raw, &m); err != nil { t.Fatal(err) } - want := map[string]bool{AssignmentsRead: true, SeatHoldersRead: true, ModulesRead: true, ManifestsRead: true} + want := map[string]bool{AssignmentsRead: true, SeatHoldersRead: true} for _, s := range Specs { if !want[s.Name()] { t.Errorf("the first view %s is read by no name here", s.Name()) @@ -199,4 +231,15 @@ func TestTheFirstViewsAreTheManifestsViews(t *testing.T) { if len(want) > 0 { t.Errorf("no first view reads %v", want) } + // the catalog half's follower hears these events and asks these verbs, and the manifest says so (design 47 G §614) + for _, ev := range []string{"mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved"} { + if !slices.Contains(m.Consumes, ev) { + t.Errorf("the catalog half's event %s is not in the manifest's consumes %v", ev, m.Consumes) + } + } + for _, ev := range []string{"mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.manifests"} { + if slices.Contains(m.Consumes, ev) { + t.Errorf("the manifest still consumes the indexer's %s", ev) + } + } } diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index 2f021e1..aafeef3 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -23,7 +23,6 @@ import ( assigner "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" seatassigner "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" - meshcatalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" meshcatalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-sdk/go/membership" @@ -153,25 +152,27 @@ func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership { return out } -// The indexer's two states the console reads, and their buckets (the test bed's own names: the indexer's client names +// The states the console reads, and their buckets (the test bed's own names: the indexer's client names // no holder). const ( - indexer = "mesh-delivery-indexer" + catalogHolder = "mesh-catalog" assignerHolder = "mesh-delivery-assigner" // AssignmentsBucket and SeatHoldersBucket are the assigner's states assignments and seat-holders. AssignmentsBucket = assignerHolder + "_assignments" SeatHoldersBucket = assignerHolder + "_seat-holders" - ModulesBucket = indexer + "_modules" - ManifestsBucket = indexer + "_manifests" + // ModulesBucket holds the catalog half's entries by module, the test bed's own stand-in for the runner's first views of it. + ModulesBucket = catalogHolder + "_modules" ) +// ModulesRead names the test bed's state of the catalog half's entries. +const ModulesRead = catalogHolder + ".modules" + // ConsoleReads are the states a console's runtime reads to say what the mesh holds (novox/hq ADR 0571 clauses 4, 5): -// assignments, the indexer's modules and manifests, and the seat holders. +// assignments, the catalog half's entries, and the seat holders. func ConsoleReads() []bus.StateIssued { return []bus.StateIssued{ {Name: assignerHolder + ".assignments", Bucket: AssignmentsBucket}, - {Name: indexer + ".modules", Bucket: ModulesBucket}, - {Name: indexer + ".manifests", Bucket: ManifestsBucket}, + {Name: ModulesRead, Bucket: ModulesBucket}, {Name: assignerHolder + ".seat-holders", Bucket: SeatHoldersBucket}, } } @@ -188,20 +189,11 @@ func (m *Mesh) Placed(t *testing.T, tools map[string][]string, on map[string][]s } return b } - modules, manifests, assigned := kv(ModulesBucket), kv(ManifestsBucket), kv(AssignmentsBucket) + modules, assigned := kv(ModulesBucket), kv(AssignmentsBucket) kv(SeatHoldersBucket) for module, declared := range tools { - manifest, _ := json.Marshal(map[string]any{"module": module, "tools": declared}) - digest := meshcatalogclient.ManifestDigest(manifest) - raw, err := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: module, Manifest: string(manifest), V: 1}) - if err != nil { - t.Fatal(err) - } - if _, err := manifests.Put(digest, raw); err != nil { - t.Fatal(err) - } - raw, err = meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{None: "not built here", Manifest: digest, Repository: "novox/" + module, - Commit: strings.Repeat("a", 40), V: 1}) + manifest := map[string]any{"module": module, "tools": declared} + raw, err := json.Marshal(meshcatalog.CatalogEntry{Module: module, Manifest: manifest, Commit: strings.Repeat("a", 40)}) if err != nil { t.Fatal(err) } diff --git a/node-tools/module.json b/node-tools/module.json index a202049..f206a4d 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -72,13 +72,13 @@ ], "consumes": [ "mesh-catalog.catalog-revision", + "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.seat-holders", - "mesh-delivery-indexer.views.manifests", - "mesh-delivery-indexer.views.modules" + "mesh-delivery.source-moved" ], "streams": [ {