From 2940df27b11909cb3db3f9184878ccbb326ec9ba Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:03:48 +0100 Subject: [PATCH] VW5: the view follows the catalog half per watched branch through meshcatalogclient's branch follower MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The process and the tools bundle read the catalog half through Branches (FollowBranchesProcess, FollowBranches) and Branches.Entry instead of one view of a branch named in code; the module's value carries the entry's manifest-digest bare and computes none. The manifest consumes registration-changed and source-moved and invokes registrations, and revision, entry and source in the long form with the fields relied on. The built binary committed at the module's root goes. Builds: design 47 G §532, §614, §558, §506, §578, A-103 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-view-tools/catalogread.go | 100 +++++----- .../cmd/mesh-delivery-view-tools/main.go | 19 +- .../mesh-delivery-view-tools/owners_test.go | 80 +++++++- .../mesh-delivery-view/integration_test.go | 36 +++- .../cmd/mesh-delivery-view/main.go | 174 +++++++++++++++--- .../internal/view/catalog.go | 12 +- modules/mesh-delivery-view/manifest_test.go | 59 ++++-- modules/mesh-delivery-view/module.json | 7 +- 8 files changed, 361 insertions(+), 126 deletions(-) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/catalogread.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/catalogread.go index 2ec90638..cdddb592 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/catalogread.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/catalogread.go @@ -1,65 +1,55 @@ package main import ( - "context" "encoding/json" "fmt" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "sort" - "sync" - "time" "mesh-delivery-view/internal/view" ) -// catalogWithin is how long a call waits for the seat mesh-catalog's first view before it says there is none. -var catalogWithin = 30 * time.Second - -// catalogRead is the modules the view's tools answer, read from the catalog half through the seat mesh-catalog's verb -// `revision` (design 47, Two graphs): the trunk's newest revision, asked once per call and in pages, each entry in the -// form of the value the tools always answered. With no view the read says so; it never answers no modules. -func catalogRead(ask func(key string, body any) (json.RawMessage, error)) view.StateRead { - var ( - once sync.Once - values map[string]json.RawMessage - rev uint64 - failed error - ) - load := func() { - ctx, cancel := context.WithTimeout(context.Background(), catalogWithin) - defer cancel() - c, err := meshcatalogclient.NewCatalog(ask, "main") - if err != nil { - failed = err - return +// catalogRead is the modules the view's tools answer, read from the catalog half per watched branch through +// meshcatalogclient's branch follower (design 47 G §532, §614): every module a followed branch's view lists, each the +// entry Branches.Entry reads, the first branch of the module's own registration (A-103), in the form of the value the +// tools always answered. Until every watched branch has its first view the read says what it waits for; it never answers +// no modules. +func catalogRead(b *meshcatalogclient.Branches) view.StateRead { + read := func() (map[string]json.RawMessage, error) { + if err := b.Ready(); err != nil { + return nil, fmt.Errorf("%w (%s)", err, meshcatalogclient.RevisionVerb) } - if err := c.Start(ctx); err != nil { - failed = fmt.Errorf("the catalog half has no first view yet (%s): %w", meshcatalogclient.RevisionVerb, err) - return - } - half, ok := c.View() - if !ok { - failed = fmt.Errorf("the catalog half has no first view yet (%s)", meshcatalogclient.RevisionVerb) - return - } - rev, values = uint64(half.Revision.Revision), map[string]json.RawMessage{} - for module, e := range half.Entries { - raw, _ := json.Marshal(e) - v, err := view.ModuleValue(raw) - if err != nil { - failed = fmt.Errorf("the catalog entry of %s: %w", module, err) - return + values := map[string]json.RawMessage{} + for _, branch := range b.Followed() { + c, ok := b.Catalog(branch) + if !ok { + continue + } + half, _ := c.View() + for module := range half.Entries { + if _, done := values[module]; done { + continue + } + e, ok, err := b.Entry(module) + if err != nil { + return nil, err + } + if !ok { + continue + } + raw, _ := json.Marshal(e) + v, err := view.ModuleValue(raw) + if err != nil { + return nil, fmt.Errorf("the catalog entry of %s: %w", module, err) + } + values[module] = v } - values[module] = v } - } - read := func() (map[string]json.RawMessage, uint64, error) { - once.Do(load) - return values, rev, failed + return values, nil } return view.StateRead{ Keys: func() ([]string, error) { - v, _, err := read() + v, err := read() keys := make([]string, 0, len(v)) for k := range v { keys = append(keys, k) @@ -68,16 +58,16 @@ func catalogRead(ask func(key string, body any) (json.RawMessage, error)) view.S return keys, err }, Get: func(key string) (json.RawMessage, error) { - v, _, err := read() - return v[key], err - }, - GetEntry: func(key string) (json.RawMessage, uint64, error) { - v, r, err := read() - return v[key], r, err - }, - All: func() (map[string]json.RawMessage, error) { - v, _, err := read() - return v, err + if err := b.Ready(); err != nil { + return nil, fmt.Errorf("%w (%s)", err, meshcatalogclient.RevisionVerb) + } + e, ok, err := b.Entry(key) + if err != nil || !ok { + return nil, err + } + raw, _ := json.Marshal(e) + return view.ModuleValue(raw) }, + All: read, } } diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index 5aaf5331..5c2e08c1 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -4,6 +4,7 @@ package main import ( + "context" "encoding/json" "fmt" "mesh-delivery-view/internal/view" @@ -16,6 +17,7 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" @@ -395,11 +397,11 @@ func read(state stdio.KeptState) view.StateRead { // ownersRead is the view's reading of the indexer's and the assigner's states: each through its owner's read tool, asked // by the bundle on each call (ADR 0615 §1, design 25 § 12.2: the view keeps its tools and answers them by asking the // owners), never their buckets. -func ownersRead(ask func(key string, body any) (json.RawMessage, error)) view.Reader { +func ownersRead(ask func(key string, body any) (json.RawMessage, error), catalog *meshcatalogclient.Branches) view.Reader { // Each owner's read tool as the view's invokes names it. indexer := func(tool string) view.StateRead { return view.OwnerRead(ask, view.IndexerModule+"."+tool) } assigner := func(tool string) view.StateRead { return view.OwnerRead(ask, view.AssignerModule+"."+tool) } - return view.Reader{State: catalogRead(ask), CandidatesState: indexer("indexer_read_candidates"), + return view.Reader{State: catalogRead(catalog), CandidatesState: indexer("indexer_read_candidates"), WantsState: assigner("assigner_read_wants"), DifferencesState: assigner("assigner_read_differences"), HeardState: indexer("indexer_read_heard"), PredefinedState: indexer("indexer_read_predefined"), @@ -418,7 +420,18 @@ func holderOf(ask func(key string, body any) (json.RawMessage, error)) func(seat } func main() { - r := ownersRead(stdio.Ask) + logf := func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-delivery-view-tools] "+f+"\n", a...) } + // The catalog half per watched branch, followed by the bundle itself through meshcatalogclient's branch follower on + // its own views reader (design 47 G §532, §614, A-103), as other bundles follow the views they read. + cat := &views.Reader{Module: viewstate.ViewModule, Logf: logf, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}} + branches, err := meshcatalogclient.FollowBranches(cat, stdio.Ask) + if err != nil { + logf("%v", err) + os.Exit(1) + } + branches.Logf = logf + cat.Start(context.Background()) + r := ownersRead(stdio.Ask, branches) r.Builds, r.HavesState = read(stdio.State("mesh-delivery-builder.builds")), read(stdio.State("node-engine.have")) r.ReachState = read(stdio.State("reach")) r.SignersState = read(stdio.State("mesh-vault.signers")) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go index b5affab6..1ace4595 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go @@ -1,6 +1,7 @@ package main import ( + "context" "encoding/json" "fmt" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" @@ -89,9 +90,9 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { } asked := map[string]int{} ask := func(key string, body any) (json.RawMessage, error) { - if key == meshcatalogclient.RevisionVerb { + if a, ok := catalogSeats(key, body); ok { asked[key]++ - return json.Marshal(catalogAnswer) + return a() } name, ok := tools[key] if !ok { @@ -117,7 +118,7 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { } return json.Marshal(a) } - owned := ownersRead(ask) + owned := ownersRead(ask, startedBranches(t, context.Background(), ask)) for _, row := range []struct { tool string args map[string]any @@ -147,7 +148,7 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { // the form of the value the tool always answered, the modules without a fingerprint say why, and the module's history // is not a thing the catalog half keeps. a, err := run(t, owned, view.ModuleReadTool, map[string]any{"module": "a"}) - if raw, _ := json.Marshal(a); err != nil || !strings.Contains(string(raw), `"fp2":"`+fp2A+`"`) || !strings.Contains(string(raw), `"commit":"c1"`) || !strings.Contains(string(raw), `"src":7`) { + if raw, _ := json.Marshal(a); err != nil || !strings.Contains(string(raw), `"fp2":"`+fp2A+`"`) || !strings.Contains(string(raw), `"commit":"c1"`) || !strings.Contains(string(raw), `"src":7`) || !strings.Contains(string(raw), `"manifest":"`+manifestDigestA+`"`) { t.Errorf("indexer_module a = %s, %v", raw, err) } if _, err := run(t, owned, view.ModuleReadTool, map[string]any{"module": "gone"}); err == nil { @@ -169,9 +170,11 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { t.Errorf("the want's history through the assigner's read tool = %+v, %v", revs, err) } // With no view of the catalog half the modules are not read: the tools say so, never that no module is held. - catalogWithin = 20 * time.Millisecond - none := ownersRead(func(key string, body any) (json.RawMessage, error) { return nil, fmt.Errorf("the holder is away") }) - if out, err := run(t, none, view.ModulesReadTool, map[string]any{}); err == nil || !strings.Contains(err.Error(), "no first view") { + away := func(key string, body any) (json.RawMessage, error) { return nil, fmt.Errorf("the holder is away") } + gone, cancel := context.WithCancel(context.Background()) + cancel() + none := ownersRead(away, startedBranches(t, gone, away)) + if out, err := run(t, none, view.ModulesReadTool, map[string]any{}); err == nil || !strings.Contains(err.Error(), "no watched branch") { t.Errorf("indexer_modules with no view = %v, %v, want an error naming it", out, err) } // Each owner state the tools read was asked of its owner's read tool. @@ -180,21 +183,78 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { t.Errorf("%s was never asked", tool) } } - if asked[meshcatalogclient.RevisionVerb] == 0 { - t.Errorf("%s was never asked", meshcatalogclient.RevisionVerb) + for _, verb := range []string{meshcatalogclient.RevisionVerb, meshcatalogclient.RegistrationsVerb, meshcatalogclient.SourceVerb} { + if asked[verb] == 0 { + t.Errorf("%s was never asked", verb) + } } } +// catalogBranch is the one watched branch of the catalog half in these tests: the registrations name it, the gate's source +// keeps it, and the seat's revision answers it alone, so a reader that names a branch of its own reads nothing. +const catalogBranch = "release" + +// catalogSeats answers the seats the branch follower asks (design 47 G §614): the registrations, each module on +// catalogBranch, the gate's source, and the newest trunk revision of catalogBranch; ok is false for any other key. +func catalogSeats(key string, body any) (func() (json.RawMessage, error), bool) { + switch key { + case meshcatalogclient.RegistrationsVerb: + return func() (json.RawMessage, error) { + return json.Marshal(meshcatalog.RegistrationsAnswer{Registrations: []meshcatalog.Registration{ + {Module: "a", Version: 1, Branches: []string{catalogBranch}}, {Module: "b", Version: 1, Branches: []string{catalogBranch}}}}) + }, true + case meshcatalogclient.SourceVerb: + return func() (json.RawMessage, error) { return json.Marshal(map[string][]string{"branches": {catalogBranch}}) }, true + case meshcatalogclient.RevisionVerb: + return func() (json.RawMessage, error) { + if in, _ := body.(meshcatalog.RevisionInput); in.Branch != catalogBranch { + return nil, fmt.Errorf("no watched branch %q", in.Branch) + } + return json.Marshal(catalogAnswer) + }, true + } + return nil, false +} + +// startWiring is the bundle's reader for a test: each view takes its first view as it is adopted, within ctx. +type startWiring struct { + t *testing.T + ctx context.Context +} + +func (w startWiring) Adopt(f *views.Follower, _, _ string, _ func(context.Context, []byte) error) error { + if err := f.Start(w.ctx); err != nil && w.ctx.Err() == nil { + w.t.Errorf("the first view of %s.%s: %v", f.Owner, f.State, err) + } + return nil +} + +func (startWiring) Drop(string, string) error { return nil } + +// startedBranches is the branch follower of meshcatalogclient asking ask, its views started at once. +func startedBranches(t *testing.T, ctx context.Context, ask func(string, any) (json.RawMessage, error)) *meshcatalogclient.Branches { + t.Helper() + b, err := meshcatalogclient.NewBranches(startWiring{t, ctx}, ask) + if err != nil { + t.Fatal(err) + } + return b +} + // catalogAnswer is the seat mesh-catalog's newest trunk revision: a with a fingerprint and the source revision it moved at, // b with none and the reason. var catalogAnswer = meshcatalog.RevisionAnswer{ Revision: &meshcatalog.CatalogRevision{Revision: 4, Kind: meshcatalogclient.KindTrunk, Source: 7}, Entries: []meshcatalog.CatalogEntry{ - {Module: "a", Repository: "novox/r", Commit: "c1", Path: ".", Manifest: map[string]any{"module": "a"}, Fp: fp2A, Inputs: &meshcatalog.EntryInputs{Src: 7}}, + {Module: "a", Repository: "novox/r", Commit: "c1", Path: ".", Manifest: map[string]any{"module": "a"}, ManifestDigest: "sha256:" + manifestDigestA, Fp: fp2A, Inputs: &meshcatalog.EntryInputs{Src: 7}}, {Module: "b", Repository: "novox/r", Commit: "c2", Path: "modules/b", Manifest: map[string]any{"module": "b"}, None: "builder's fingerprint not yet known"}, }, } +// manifestDigestA is the digest the indexer read of a's manifest bytes, which the module's value carries bare (design 47 G +// §506, §578), never one computed again from the manifest whole. +const manifestDigestA = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + // view_read_answers answers the view's own answers as every owner's read tool answers its state (design 47 G A-178): one // key's value, and every key when none is named. func TestTheViewReadsItsAnswersAsAnOwner(t *testing.T) { diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index 5f2cad89..e26e754e 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -86,16 +86,38 @@ func serveCatalog(t testing.TB, ctx context.Context, js jetstream.JetStream, nc raw, _ := json.Marshal(map[string]any{"result": a, "node": "n0"}) _ = m.Respond(raw) } - for _, subject := range []string{"mesh.seat.mesh-catalog.call.revision.>", "mesh.seat.mesh-catalog.call.revision"} { - sub, err := nc.Subscribe(subject, answer) - if err != nil { - t.Fatal(err) + // The branch follower's other two first views (design 47 G §614): the registrations, each module on catalogBranch, and the + // gate's source, which keeps catalogBranch. + registrations := func(m *nats.Msg) { + h.mu.Lock() + a := meshcatalog.RegistrationsAnswer{Registrations: []meshcatalog.Registration{}} + for module := range h.entries { + a.Registrations = append(a.Registrations, meshcatalog.Registration{Module: module, Version: 1, Branches: []string{catalogBranch}}) + } + h.mu.Unlock() + raw, _ := json.Marshal(map[string]any{"result": a, "node": "n0"}) + _ = m.Respond(raw) + } + source := func(m *nats.Msg) { + raw, _ := json.Marshal(map[string]any{"result": map[string]any{"branches": []string{catalogBranch}}, "node": "n0"}) + _ = m.Respond(raw) + } + for subject, handle := range map[string]nats.MsgHandler{"mesh.seat.mesh-catalog.call.revision": answer, + "mesh.seat.mesh-catalog.call.registrations": registrations, "mesh.seat.mesh-delivery.call.source": source} { + for _, s := range []string{subject + ".>", subject} { + sub, err := nc.Subscribe(s, handle) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) } - t.Cleanup(func() { _ = sub.Unsubscribe() }) } return h } +// catalogBranch is the one watched branch of these tests: no branch is named in the view's code (design 47 G §532). +const catalogBranch = "release" + // put is the holder's next trunk revision with the entry of one module changed, said as catalog-revision. func (h *catalogHolder) put(e meshcatalog.CatalogEntry) { h.t.Helper() @@ -104,8 +126,8 @@ func (h *catalogHolder) put(e meshcatalog.CatalogEntry) { h.entries[e.Module] = e rev := h.rev h.mu.Unlock() - body, _ := json.Marshal(meshcatalog.CatalogRevisionEvent{Kind: meshcatalogclient.KindTrunk, Branch: "main", Revision: rev, Changed: []string{e.Module}}) - if _, err := h.js.Publish(h.ctx, "mesh.seat.mesh-catalog.event."+meshcatalog.EventCatalogRevision, body); err != nil { + body, _ := json.Marshal(meshcatalog.CatalogRevisionEvent{Kind: meshcatalogclient.KindTrunk, Branch: catalogBranch, Revision: rev, Changed: []string{e.Module}}) + if _, err := h.js.Publish(h.ctx, "mesh.seat.mesh-catalog.event."+meshcatalog.EventCatalogRevision+"."+catalogBranch, body); err != nil { h.t.Fatal(err) } } diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index 3d02f714..a4262295 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -14,15 +14,19 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "io" "log" + "maps" "mesh-delivery-view/internal/view" "net/http" "os" "os/signal" + "slices" "strings" + "sync" "syscall" "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate" "git.novox.be/novox/mesh-sdk/go/process" @@ -145,14 +149,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // catalog half the view takes, in its order (below). go follow(ctx, open, view.SourceBucket, true, model, logger) statusBucket := process.AwaitBucket(ctx, js, conn, firstViewsBucket, "view_first_views reads nothing and view_first_view_again is not heard", logger.Printf) - // The modules are the catalog half's, adopted whole by the seat's client: its first view, its status, its hand asks and - // its event catalog-revision (on the module's durable consumer) are the reader's. - catalog, _, err := meshcatalogclient.WireCatalogProcess(reader, calls.Call, "main", catalogChanged(model, logger)) + // The modules are the catalog half's, per watched branch, through meshcatalogclient's branch follower adopted on the + // reader (design 47 G §532, §614): the registrations' branches, before the first registrations the gate's source's; + // each branch's first view, its status, its hand asks and its event catalog-revision are the reader's. A module's entry + // is the one Branches.Entry reads, the first branch of its own registration (A-103). + branches, err := meshcatalogclient.FollowBranchesProcess(reader, calls.Call) if err != nil { return err } + branches.Logf = logger.Printf + modules := &catalogModules{ctx: ctx, branches: branches, model: model, logger: logger, applied: map[string]bool{}} + branches.Changed = modules.changed reader.Start(ctx, js, cred, statusBucket) - catalogHeld(ctx, catalog, model) + go func() { + select { + case <-branches.Registrations.Follower.Ready(): + modules.settle() + case <-ctx.Done(): + } + }() var said error select { case <-ctx.Done(): @@ -319,34 +334,135 @@ func emitEvent(ctx context.Context, js jetstream.JetStream, cred process.Account return err } -// catalogChanged hands one module of a revision the view took to the model, in the form of its value; the modules' -// history is these changes in the order of the revisions. -func catalogChanged(model *view.Model, logger *log.Logger) func(string, json.RawMessage, bool, time.Time) { - return func(module string, entry json.RawMessage, gone bool, at time.Time) { - e := view.Entry{Bucket: view.ModulesBucket, Key: module, At: at, Deleted: gone} - if !gone { - v, err := view.ModuleValue(entry) - if err != nil { - logger.Printf("the catalog entry of %s is not taken: %v", module, err) - return - } - e.Value = v - } - model.Apply(e) - e.History = true - model.Apply(e) - } +// catalogModules hands the model the catalog half's modules, one value per module: the entry Branches.Entry reads. Until +// every watched branch has its first view (Branches.Ready) it hands nothing; once they all have, it hands every module +// whole and tells the model the modules are read (Initial), then each module a branch's change names. A branch adopted +// later without its first view stops the handing until it has one, when every module is handed whole again. +type catalogModules struct { + ctx context.Context + branches *meshcatalogclient.Branches + model *view.Model + logger *log.Logger + + mu sync.Mutex + held bool // every module was handed whole since the last time a branch had no first view + waiting bool // a settle waits on the branches' first views + applied map[string]bool // the modules the model holds } -// catalogHeld makes the model's pictures of the modules initial once the catalog half's first view is held: the process -// starts at once and waits for the seat's holder, saying so. -func catalogHeld(ctx context.Context, cat *meshcatalogclient.Catalog, model *view.Model) { +// changed is the branch follower's Changed: a module of one branch changed. +func (c *catalogModules) changed(_, module string, _ json.RawMessage, _ bool, at time.Time) { + c.mu.Lock() + held := c.held + c.mu.Unlock() + if !held { + c.settle() + return + } + e, ok, err := c.branches.Entry(module) + if err != nil { + c.mu.Lock() + c.held = false + c.mu.Unlock() + c.settle() + return + } + c.apply(module, e, ok, at) +} + +// settle waits, on its own goroutine, for the first view of every followed branch, then hands every module whole. It +// waits on the views' Ready, never on a timer; with no branch followed it ends, and the next change starts it again. +func (c *catalogModules) settle() { + c.mu.Lock() + if c.waiting || c.held { + c.mu.Unlock() + return + } + c.waiting = true + c.mu.Unlock() go func() { - select { - case <-cat.Follower.Ready(): - model.Initial(view.ModulesBucket) - model.Initial(view.ReachHistory(view.ModulesBucket)) - case <-ctx.Done(): + defer func() { + c.mu.Lock() + c.waiting = false + c.mu.Unlock() + }() + for c.ctx.Err() == nil { + followed := c.branches.Followed() + if len(followed) == 0 { + return + } + for _, branch := range followed { + if cat, ok := c.branches.Catalog(branch); ok { + select { + case <-cat.Ready(): + case <-c.ctx.Done(): + return + } + } + } + if c.branches.Ready() == nil { + c.whole() + return + } } }() } + +// whole hands every module the followed branches list, removes from the model each it holds that none lists any more, +// and tells the model the modules are read. +func (c *catalogModules) whole() { + at := time.Now() + listed := map[string]bool{} + for _, branch := range c.branches.Followed() { + if cat, ok := c.branches.Catalog(branch); ok { + half, _ := cat.View() + for module := range half.Entries { + listed[module] = true + } + } + } + c.mu.Lock() + for module := range c.applied { + if !listed[module] { + listed[module] = false + } + } + c.mu.Unlock() + for _, module := range slices.Sorted(maps.Keys(listed)) { + e, ok, err := c.branches.Entry(module) + if err != nil { + c.logger.Printf("the catalog half is handed again at its next change: %v", err) + return + } + c.apply(module, e, ok, at) + } + c.mu.Lock() + c.held = true + c.mu.Unlock() + c.model.Initial(view.ModulesBucket) + c.model.Initial(view.ReachHistory(view.ModulesBucket)) +} + +// apply hands the model one module's entry, or its removal when no branch lists it. +func (c *catalogModules) apply(module string, entry meshcatalog.CatalogEntry, listed bool, at time.Time) { + e := view.Entry{Bucket: view.ModulesBucket, Key: module, At: at, Deleted: !listed} + if listed { + raw, _ := json.Marshal(entry) + v, err := view.ModuleValue(raw) + if err != nil { + c.logger.Printf("the catalog entry of %s is not taken: %v", module, err) + return + } + e.Value = v + } + c.mu.Lock() + if listed { + c.applied[module] = true + } else { + delete(c.applied, module) + } + c.mu.Unlock() + c.model.Apply(e) + e.History = true + c.model.Apply(e) +} diff --git a/modules/mesh-delivery-view/internal/view/catalog.go b/modules/mesh-delivery-view/internal/view/catalog.go index 8d6889ae..42f5dc4a 100644 --- a/modules/mesh-delivery-view/internal/view/catalog.go +++ b/modules/mesh-delivery-view/internal/view/catalog.go @@ -8,6 +8,7 @@ package view import ( "encoding/json" "fmt" + "strings" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" @@ -22,7 +23,8 @@ const CatalogSeat = meshcatalog.Seat const ModulesBucket = CatalogSeat + "_modules" // ModuleValue is the value of one module of the model, as the model decodes it, made from the catalog entry (its JSON): -// the fingerprint is the entry's `fp` (the only form there is), the source revisions are the entry's inputs'. A manifest +// the fingerprint is the entry's `fp` (the only form there is), the manifest's digest the entry's `manifest-digest`, the +// source revisions are the entry's inputs'. A manifest // held for unknown fields has no verdict here: the model reads the fingerprint and the commit. func ModuleValue(entry json.RawMessage) (json.RawMessage, error) { var e meshcatalog.CatalogEntry @@ -30,11 +32,9 @@ func ModuleValue(entry json.RawMessage) (json.RawMessage, error) { return nil, fmt.Errorf("a catalog entry does not read: %v", err) } v := meshcatalog.ModulesByModuleValue{Commit: e.Commit, Fp2: e.Fp, None2: e.None, Path: e.Path, Repository: e.Repository, Packages: e.Packages} - if len(e.Manifest) > 0 { - // The entry holds its manifest whole; the model keeps the digest of its bytes, as the value it always kept does. - raw, _ := json.Marshal(e.Manifest) - v.Manifest = meshcatalogclient.ManifestDigest(raw) - } + // The entry carries the digest of its manifest's bytes as the indexer read them, `sha256:` and the 64 hex digits; the + // model keeps the bare digits, as the value it always kept, and computes none (design 47 G §506, §578). + v.Manifest = strings.TrimPrefix(e.ManifestDigest, "sha256:") if e.Verdict == meshcatalogclient.VerdictAccepted || e.Verdict == meshcatalogclient.VerdictRefused { v.Verdict = e.Verdict } diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index def20938..aa549fe3 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -67,7 +67,7 @@ type manifest struct { Slug string `json:"slug"` Reads []string `json:"reads"` Consumes []string `json:"consumes"` - Invokes []string `json:"invokes"` + Invokes []invoke `json:"invokes"` Uses []string `json:"uses"` Tools []string `json:"tools"` Emits []string `json:"emits"` @@ -82,6 +82,32 @@ type manifest struct { } `json:"build"` } +// invoke is an entry of `invokes`: a verb, or in its long form {verb, fields}, the fields of the verb's answer the view +// relies on (design 47 G §558). +type invoke struct { + Verb string `json:"verb"` + Fields []string `json:"fields,omitempty"` +} + +func (i *invoke) UnmarshalJSON(raw []byte) error { + if err := json.Unmarshal(raw, &i.Verb); err == nil { + return nil + } + type long invoke + dec := json.NewDecoder(bytes.NewReader(raw)) + dec.DisallowUnknownFields() + return dec.Decode((*long)(i)) +} + +// verbs are the verbs of the invokes, whichever their form. +func verbs(in []invoke) []string { + out := make([]string, 0, len(in)) + for _, i := range in { + out = append(out, i.Verb) + } + return out +} + func load(t *testing.T) (manifest, map[string]json.RawMessage) { t.Helper() raw, err := os.ReadFile("module.json") @@ -119,16 +145,19 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 } // The indexer's and the assigner's states are asked of their owners' read tools (design 25 § 12.2), the six the process // follows then heard on their views families (ADR 0615 §1, §5; DLV-M4). - if want := []string{"mesh-delivery-assigner.assigner_read_asks", "mesh-delivery-assigner.assigner_read_differences", "mesh-delivery-assigner.assigner_read_assignments", - "mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_config_candidates", - "mesh-delivery-indexer.indexer_read_graph", "mesh-delivery-indexer.indexer_read_heard", "mesh-delivery-indexer.indexer_pending", - "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_read_predefined", "mesh-delivery-indexer.indexer_read_taken", - "mesh-delivery-teller.teller_read_answers", // A-175: the teller's answers by their views family - "seat:mesh-catalog.revision", // ADR 0663: the modules are the catalog half's, read through the seat's verb revision, never the indexer's - "seat:mesh-catalog.seat-definitions", // design 47 G A-82, §204: the build seats are read from the seat list - "seat:mesh-hand-acts.hand-act", // the join: its first view asked again by hand is a hand act (views.StatusTools) - "seat:mesh-module-assigner.assignments", // A-82: the holder of each build seat - "seat:node-engine.manifest-reads"}; !reflect.DeepEqual(m.Invokes, want) { // ADR 0618 §3: node_manifest_reads asks the node-engine seat's verb of every node; 1585/T24: view_tiers asks indexer_pending, a verb edge (ADR 0614 §2) + if want := []invoke{{Verb: "mesh-delivery-assigner.assigner_read_asks"}, {Verb: "mesh-delivery-assigner.assigner_read_differences"}, {Verb: "mesh-delivery-assigner.assigner_read_assignments"}, + {Verb: "mesh-delivery-assigner.assigner_read_wants"}, {Verb: "mesh-delivery-indexer.indexer_read_candidates"}, {Verb: "mesh-delivery-indexer.indexer_read_config_candidates"}, + {Verb: "mesh-delivery-indexer.indexer_read_graph"}, {Verb: "mesh-delivery-indexer.indexer_read_heard"}, {Verb: "mesh-delivery-indexer.indexer_pending"}, + {Verb: "mesh-delivery-indexer.indexer_read_packages"}, {Verb: "mesh-delivery-indexer.indexer_read_predefined"}, {Verb: "mesh-delivery-indexer.indexer_read_taken"}, + {Verb: "mesh-delivery-teller.teller_read_answers"}, // A-175: the teller's answers by their views family + {Verb: "seat:mesh-catalog.revision", Fields: []string{"none-yet"}}, + {Verb: "seat:mesh-catalog.registrations"}, // §614: the watched branches of the registrations, then registration-changed + {Verb: "seat:mesh-catalog.entry", Fields: []string{"manifest-digest"}}, // §558: the module's value is the entry's manifest-digest (§506) + {Verb: "seat:mesh-delivery.source", Fields: []string{"branches"}}, // §507, §614: before the first registrations, the gate's source's branches // ADR 0663: the modules are the catalog half's, read through the seat's verb revision, never the indexer's + {Verb: "seat:mesh-catalog.seat-definitions"}, // design 47 G A-82, §204: the build seats are read from the seat list + {Verb: "seat:mesh-hand-acts.hand-act"}, // the join: its first view asked again by hand is a hand act (views.StatusTools) + {Verb: "seat:mesh-module-assigner.assignments"}, // A-82: the holder of each build seat + {Verb: "seat:node-engine.manifest-reads"}}; !reflect.DeepEqual(m.Invokes, want) { // ADR 0618 §3: node_manifest_reads asks the node-engine seat's verb of every node; 1585/T24: view_tiers asks indexer_pending, a verb edge (ADR 0614 §2) t.Errorf("invokes = %v, want %v", m.Invokes, want) } // design 47 G §387: the keeper's ask for the findings is heard, the finding stream declared (§383), and §368: version is MAJOR.MINOR. @@ -142,7 +171,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 t.Errorf("version = %q, want 0.1", m.Version) } if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks", "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants", - "mesh-catalog.catalog-revision", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.taken", "mesh-delivery-teller.views.answers"}; !reflect.DeepEqual(m.Consumes, want) { + "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.taken", "mesh-delivery-teller.views.answers"}; !reflect.DeepEqual(m.Consumes, want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } // The one event the view says: a source revision is delivered (1906/D9, ADR 0539 §17), derived by it from the wants and the haves. @@ -201,8 +230,8 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 t.Errorf("the view follows %s, its manifest consumes %v", family, m.Consumes) } } - for _, tool := range []string{meshcatalogclient.RevisionVerb, view.IndexerModule + ".indexer_read_candidates", view.IndexerModule + ".indexer_pending", view.AssignerModule + ".assigner_read_wants", view.TellerModule + ".teller_read_answers"} { - if !slices.Contains(m.Invokes, tool) { + for _, tool := range []string{meshcatalogclient.RevisionVerb, meshcatalogclient.RegistrationsVerb, meshcatalogclient.SourceVerb, meshcatalogclient.EntryVerb, view.IndexerModule + ".indexer_read_candidates", view.IndexerModule + ".indexer_pending", view.AssignerModule + ".assigner_read_wants", view.TellerModule + ".teller_read_answers"} { + if !slices.Contains(verbs(m.Invokes), tool) { t.Errorf("the view asks %s, its manifest invokes %v", tool, m.Invokes) } } @@ -217,7 +246,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if !slices.Contains(m.Consumes, view.CatalogSeat+"."+meshcatalog.EventCatalogRevision) { t.Errorf("the view hears the seat's %s, its manifest consumes %v", meshcatalog.EventCatalogRevision, m.Consumes) } - for _, name := range append(slices.Clone(m.Invokes), m.Consumes...) { + for _, name := range append(verbs(m.Invokes), m.Consumes...) { if strings.Contains(name, "indexer_read_modules") || strings.Contains(name, "views.modules") { t.Errorf("the manifest names %s: the modules are the catalog half's (ADR 0663)", name) } diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 19914b85..7b42f0ef 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -64,6 +64,8 @@ "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants", "mesh-catalog.catalog-revision", + "mesh-catalog.registration-changed", + "mesh-delivery.source-moved", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.candidates", @@ -84,7 +86,10 @@ "mesh-delivery-indexer.indexer_read_predefined", "mesh-delivery-indexer.indexer_read_taken", "mesh-delivery-teller.teller_read_answers", - "seat:mesh-catalog.revision", + {"verb": "seat:mesh-catalog.revision", "fields": ["none-yet"]}, + "seat:mesh-catalog.registrations", + {"verb": "seat:mesh-catalog.entry", "fields": ["manifest-digest"]}, + {"verb": "seat:mesh-delivery.source", "fields": ["branches"]}, "seat:mesh-catalog.seat-definitions", "seat:mesh-hand-acts.hand-act", "seat:mesh-module-assigner.assignments",