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 5c2e08c1..27d69d37 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 @@ -17,8 +17,8 @@ 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-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate" @@ -437,7 +437,7 @@ func main() { r.SignersState = read(stdio.State("mesh-vault.signers")) r.ConditionsState, r.AttemptsState = read(stdio.State(view.StatusConditionsState)), read(stdio.State(view.StatusAttemptsState)) r.PresenceState, r.StuckState = read(stdio.State(view.StatusPresenceState)), read(stdio.State(view.StatusStuckState)) - r.SourceState = read(stdio.State(view.GateModule + ".source")) + r.Sources = view.SourcesOf(stdio.Ask, branches.Followed) r.ProvisionsState = read(stdio.State(view.ProvisionsRead)) r.BuildLog = stdio.Log(builds.BuildLogRead).Since r.Pending = pendingOf diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/tiers_rows_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/tiers_rows_test.go index 615b516a..5d0e6dba 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/tiers_rows_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/tiers_rows_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "fmt" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" "mesh-delivery-view/internal/view" "reflect" "strings" @@ -137,10 +138,22 @@ func tiersToolSaysEachModulesWait(t *testing.T) { t.Errorf("view_tiers with no answer of indexer_pending = %+v, %v", out, err) } r.Pending = nil - // No revision named: the gate's newest, its source key's revision. - r.SourceState = view.StateRead{GetEntry: func(string) (json.RawMessage, uint64, error) { return json.RawMessage(`{}`), 9, nil }} - if out, err := run(t, r, view.TiersTool, map[string]any{}); err != nil || out.(view.Tiers).Revision != 9 || out.(view.Tiers).Modules != 3 { - t.Errorf("view_tiers of the gate's newest = %+v, %v", out, err) + // No revision named: per watched branch, the newest source revision the gate's verb source answers for it (design 47 G + // A-173); a branch with no source revision yet is left out. + asked := map[string]bool{} + call := func(key string, body any) (json.RawMessage, error) { + in, _ := body.(meshdelivery.SourceInput) + asked[key+" "+in.Branch] = true + rev := map[string]int64{"release": 9}[in.Branch] + return json.Marshal(meshdelivery.SourceAnswer{Branch: in.Branch, Revision: rev}) + } + r.Sources = view.SourcesOf(call, func() []string { return []string{"release", "beta"} }) + if out, err := run(t, r, view.TiersTool, map[string]any{}); err != nil || len(out.(view.BranchTiers).Branches) != 1 || + out.(view.BranchTiers).Branches[0].Branch != "release" || out.(view.BranchTiers).Branches[0].Revision != 9 || out.(view.BranchTiers).Branches[0].Modules != 3 { + t.Errorf("view_tiers of each branch's newest = %+v, %v", out, err) + } + if !asked["seat:mesh-delivery.source release"] || !asked["seat:mesh-delivery.source beta"] { + t.Errorf("the gate's verb source was asked %v, want once per watched branch", asked) } // ADR 0614 §2-§5: a want held on a verb, a reader capability or bus state is named by its edge; bus state is no // provisioner key (none is ever made for it), so a module asking it shows no provision wait of it; and each verb edge 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 e26e754e..b88736d9 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 @@ -59,7 +59,9 @@ type catalogHolder struct { // rev is the revision's number; entries are its entries by module. rev int64 entries map[string]meshcatalog.CatalogEntry - ctx context.Context + // src is the gate's source revision of catalogBranch, as its verb source answers it. + src meshdelivery.SourceAnswer + ctx context.Context } func serveCatalog(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) *catalogHolder { @@ -98,8 +100,17 @@ 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) } + // The gate's verb source: given no branch the branches it keeps, given catalogBranch its source revision (A-173). source := func(m *nats.Msg) { - raw, _ := json.Marshal(map[string]any{"result": map[string]any{"branches": []string{catalogBranch}}, "node": "n0"}) + var in meshdelivery.SourceInput + _ = json.Unmarshal(m.Data, &in) + var a any = map[string]any{"branches": []string{catalogBranch}} + if in.Branch != "" { + h.mu.Lock() + a = meshdelivery.SourceAnswer{Branch: in.Branch, Revision: h.src.Revision, Group: h.src.Group, Repositories: h.src.Repositories} + h.mu.Unlock() + } + raw, _ := json.Marshal(map[string]any{"result": a, "node": "n0"}) _ = m.Respond(raw) } for subject, handle := range map[string]nats.MsgHandler{"mesh.seat.mesh-catalog.call.revision": answer, @@ -118,6 +129,21 @@ func serveCatalog(t testing.TB, ctx context.Context, js jetstream.JetStream, nc // 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" +// moveSource is the gate's next source revision of catalogBranch, said as source-moved (design 47 G §507, A-173). +func (h *catalogHolder) moveSource(rev int64, group string, heads map[string]string) { + h.t.Helper() + h.mu.Lock() + h.src = meshdelivery.SourceAnswer{Revision: rev, Group: group} + for repository, head := range heads { + h.src.Repositories = append(h.src.Repositories, meshdelivery.SourceHead{Repository: repository, Head: head, Trunk: catalogBranch}) + } + h.mu.Unlock() + body, _ := json.Marshal(meshdelivery.SourceMovedEvent{Branch: catalogBranch, Revision: rev, Group: group}) + if _, err := h.js.Publish(h.ctx, meshdelivery.SubjectSourceMoved(catalogBranch), body); err != nil { + h.t.Fatal(err) + } +} + // 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() @@ -762,7 +788,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { t.Fatal(err) } kvs := map[string]jetstream.KeyValue{} - for _, name := range []string{view.SourceBucket, view.AssignerWantsBucket, view.HaveBucket, viewstate.ReachBucket} { + for _, name := range []string{view.AssignerWantsBucket, view.HaveBucket, viewstate.ReachBucket} { kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5}) if err != nil { t.Fatal(err) @@ -791,7 +817,6 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { want, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fp, Composed: "sha256:" + strings.Repeat("c", 64)}) failed, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed}) good, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) - src, _ := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "feat/x", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}}) catalog.put(meshcatalog.CatalogEntry{Module: "web", Fp: fp, Manifest: map[string]any{"module": "a"}, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) if _, err := kvs[view.AssignerWantsBucket].Put(ctx, "want.ace.web", want); err != nil { t.Fatal(err) @@ -799,10 +824,9 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { if _, err := kvs[view.HaveBucket].Put(ctx, "ace.web", failed); err != nil { t.Fatal(err) } - rev, err := kvs[view.SourceBucket].Put(ctx, meshdeliveryclient.SourceKey("main"), src) - if err != nil { - t.Fatal(err) - } + // The gate's source of the watched branch moves: the view asks the verb source for it and reads no gate state (A-173). + const rev = 41 + catalog.moveSource(rev, "feat/x", map[string]string{"novox/web": strings.Repeat("1", 40)}) time.Sleep(time.Second) if info, err := events.Info(ctx, jetstream.WithSubjectFilter("mesh.mod.mesh-delivery-view.event."+DeliveredEventType)); err != nil || len(info.State.Subjects) != 0 { t.Fatalf("an event was said before the revision was delivered: %v %+v", err, info.State) @@ -818,8 +842,8 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { t.Fatal("the delivery of the source revision was never said") } var e view.DeliveredEvent - if err := json.Unmarshal(msg.Data, &e); err != nil || e.Revision != rev || e.Group != "feat/x" || e.Heads["novox/web"] != strings.Repeat("1", 40) { - t.Errorf("the event = %+v, %v; want revision %d, group feat/x and the heads", e, err, rev) + if err := json.Unmarshal(msg.Data, &e); err != nil || e.Branch != catalogBranch || e.Revision != rev || e.Group != "feat/x" || e.Heads["novox/web"] != strings.Repeat("1", 40) { + t.Errorf("the event = %+v, %v; want the branch %s, revision %d, group feat/x and the heads", e, err, catalogBranch, rev) } if msg.Header.Get("x-source") != "mesh-delivery-view" || msg.Header.Get("x-event-id") == "" { t.Errorf("the headers = %v", msg.Header) 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 a4262295..181196c1 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -142,12 +142,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } } open := buckets{js: js, reader: reader} - for _, bucket := range []string{view.HaveBucket, view.AssignmentsBucket, view.PresenceBucket, view.StuckBucket, view.CandidatesBucket, view.AnswersBucket, builds.BuilderBuildsBucket, view.AssignerWantsBucket, view.SourceBucket, view.AssignerAsksBucket, viewstate.ViewAnswersBucket, view.TakenBucket} { + for _, bucket := range []string{view.HaveBucket, view.AssignmentsBucket, view.PresenceBucket, view.StuckBucket, view.CandidatesBucket, view.AnswersBucket, builds.BuilderBuildsBucket, view.AssignerWantsBucket, view.AssignerAsksBucket, viewstate.ViewAnswersBucket, view.TakenBucket, viewstate.ReachBucket} { go follow(ctx, open, bucket, false, model, logger) } - // The source's history gives each head's source revision (ADR 0556); the modules' history is each revision of the - // catalog half the view takes, in its order (below). - go follow(ctx, open, view.SourceBucket, true, model, logger) + // The view's own state reach, as it stood at the start, is what the derivation lays its history over (design 47 G A-225); + // the modules' history is each revision of the catalog half the view takes, in its order (below), and each watched + // branch's source is the seat mesh-delivery's verb source per branch, followed by its source-moved (A-173). 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, 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; @@ -158,7 +158,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } branches.Logf = logger.Printf - modules := &catalogModules{ctx: ctx, branches: branches, model: model, logger: logger, applied: map[string]bool{}} + sources := &branchSources{ctx: ctx, reader: reader, call: calls.Call, model: model, logger: logger} + modules := &catalogModules{ctx: ctx, branches: branches, sources: sources, model: model, logger: logger, applied: map[string]bool{}} branches.Changed = modules.changed reader.Start(ctx, js, cred, statusBucket) go func() { @@ -341,6 +342,7 @@ func emitEvent(ctx context.Context, js jetstream.JetStream, cred process.Account type catalogModules struct { ctx context.Context branches *meshcatalogclient.Branches + sources *branchSources // the source of each branch followed (A-173) model *view.Model logger *log.Logger @@ -352,6 +354,7 @@ type catalogModules struct { // 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.follow() c.mu.Lock() held := c.held c.mu.Unlock() @@ -411,6 +414,7 @@ func (c *catalogModules) settle() { // 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() { + c.follow() at := time.Now() listed := map[string]bool{} for _, branch := range c.branches.Followed() { @@ -443,9 +447,35 @@ func (c *catalogModules) whole() { c.model.Initial(view.ReachHistory(view.ModulesBucket)) } +// follow has the source of every followed branch followed, and no other (A-173). +func (c *catalogModules) follow() { + if c.sources != nil { + c.sources.ensure(c.branches.Followed()) + } +} + +// branchOf is the watched branch a module's entry is read from: the first branch of its registration, before the first +// registrations the first followed branch that lists it (design 47 G A-103, A-173); "" when none. +func (c *catalogModules) branchOf(module string) string { + if reg, ok := c.branches.Registrations.Registration(module); ok && len(reg.Branches) > 0 { + return reg.Branches[0] + } + for _, branch := range c.branches.Followed() { + if cat, ok := c.branches.Catalog(branch); ok { + if _, ok := cat.Entry(module); ok { + return branch + } + } + } + return "" +} + // 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 { + e.Branch = c.branchOf(module) + } if listed { raw, _ := json.Marshal(entry) v, err := view.ModuleValue(raw) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/sources.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/sources.go new file mode 100644 index 00000000..0721deaa --- /dev/null +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/sources.go @@ -0,0 +1,153 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "mesh-delivery-view/internal/view" + "sync" + "time" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/sourcebranches" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// sourceState is the state of the view of one watched branch's source in the view's reader: `mesh-delivery.source.`. +const sourceState = "source" + +// adopter is how branchSources puts a branch's view on the view's one reader and takes it off (views.ProcessReader). +type adopter interface { + Adopt(f *views.Follower, seat, event string, hear func(context.Context, []byte) error) (*views.Picture, error) + Drop(owner, state string) error +} + +// branchSources follows the source of each watched branch on its own (design 47 G §507, §512, A-173): per branch the +// first view of the seat mesh-delivery's verb source given that branch, then each source-moved of that branch, asked +// again until the revision heard is held; each answer goes to the model as the branch's SourceBucket entry. The branches +// are the ones the branch follower follows (ensure); none is named here. +type branchSources struct { + ctx context.Context + reader adopter + call func(key string, body any) (json.RawMessage, error) + model *view.Model + logger *log.Logger + + mu sync.Mutex + branches map[string]*views.Follower // by escaped branch +} + +// ensure follows the source of every branch named and drops the view of each followed branch no longer named. +func (s *branchSources) ensure(followed []string) { + s.mu.Lock() + defer s.mu.Unlock() + if s.branches == nil { + s.branches = map[string]*views.Follower{} + } + named := map[string]bool{} + for _, branch := range followed { + esc := keys.EscapeName(branch) + named[esc] = true + if s.branches[esc] != nil { + continue + } + f := s.follower(branch) + if _, err := s.reader.Adopt(f, meshdelivery.Seat, meshdelivery.EventSourceMoved, s.hear(branch, f)); err != nil { + s.logger.Printf("the source of %s is not followed, and is adopted again at the next change of the catalog half: %v", branch, err) + continue + } + s.branches[esc] = f + } + for esc := range s.branches { + if named[esc] { + continue + } + if err := s.reader.Drop(meshdelivery.Seat, views.BranchState(sourceState, mustUnescape(esc))); err != nil { + s.logger.Printf("the source of %s is not dropped: %v", esc, err) + continue + } + delete(s.branches, esc) + s.model.Apply(view.Entry{Bucket: view.SourceBucket, Key: esc, Deleted: true, At: time.Now()}) + } +} + +func mustUnescape(esc string) string { + name, err := keys.UnescapeName(esc) + if err != nil { + return esc + } + return name +} + +// follower is the view of one branch's source: its one value, keyed by the escaped branch, the verb's answer. +func (s *branchSources) follower(branch string) *views.Follower { + esc := keys.EscapeName(branch) + f := &views.Follower{Owner: meshdelivery.Seat, Tool: view.SourceVerb, State: views.BranchState(sourceState, branch)} + f.Ask = func(context.Context, string) (views.Answer, error) { + a, err := view.SourceOf(s.call, branch) + if err != nil { + return views.Answer{}, err + } + raw, err := json.Marshal(a) + if err != nil { + return views.Answer{}, err + } + return views.Answer{V: 1, Owner: meshdelivery.Seat, State: f.State, Revision: uint64(a.Revision), Created: sourcebranches.Created, + Values: map[string]json.RawMessage{esc: raw}, At: time.Now()}, nil + } + f.Changed = func(_ string, value json.RawMessage, deleted bool, at time.Time) { + e := view.Entry{Bucket: view.SourceBucket, Key: esc, Deleted: deleted, At: at} + if !deleted { + var a meshdelivery.SourceAnswer + if err := json.Unmarshal(value, &a); err != nil { + s.logger.Printf("the source of %s is not taken: %v", branch, err) + return + } + se, err := view.SourceEntry(branch, a) + if err != nil { + s.logger.Printf("the source of %s is not taken: %v", branch, err) + return + } + se.At = at + e = se + } + s.model.Apply(e) + } + return f +} + +// revisionOf is the source revision f holds of the branch, 0 when none. +func revisionOf(f *views.Follower, esc string) int64 { + raw, ok := f.Get(esc) + if !ok { + return 0 + } + var a meshdelivery.SourceAnswer + if json.Unmarshal(raw, &a) != nil { + return 0 + } + return a.Revision +} + +// hear is the handler of source-moved for one branch's view: another branch's event is not its own; a revision it does not +// hold yet asks the first view again until it does, on its own goroutine, so the consumer is never held. +func (s *branchSources) hear(branch string, f *views.Follower) func(context.Context, []byte) error { + esc := keys.EscapeName(branch) + return func(ctx context.Context, data []byte) error { + var ev meshdelivery.SourceMovedEvent + if err := json.Unmarshal(data, &ev); err != nil || ev.Branch == "" { + return fmt.Errorf("not a %s event: %s", meshdelivery.EventSourceMoved, sourcebranches.ClipBytes(data)) + } + if ev.Branch != branch && ev.Branch != esc { + return nil + } + held := func() bool { return revisionOf(f, esc) >= ev.Revision } + if held() { + return nil + } + go sourcebranches.CatchUp(s.ctx, f, held, fmt.Sprintf("source of %s moved to %d", branch, ev.Revision)) + return nil + } +} diff --git a/modules/mesh-delivery-view/internal/view/names.go b/modules/mesh-delivery-view/internal/view/names.go index 826e2f36..c20c2408 100644 --- a/modules/mesh-delivery-view/internal/view/names.go +++ b/modules/mesh-delivery-view/internal/view/names.go @@ -3,7 +3,6 @@ package view import ( "fmt" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" "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-sdk/go/keys" @@ -40,8 +39,9 @@ const ( AssignerWantsBucket = AssignerModule + "_wants" AssignerAsksBucket = AssignerModule + "_asks" - // The gate's source, and the teller's state answers as the view follows it by its views family (design 47 G §431, A-175): - // its picture `_`. + // SourceBucket names, in the model's entries, each watched branch's source as the seat mesh-delivery's verb source answers + // it, keyed by the escaped branch (design 47 G §507, A-173): never a bucket the view reads. The teller's state answers as + // the view follows it by its views family (design 47 G §431, A-175): its picture `_`. SourceBucket = GateModule + "_source" AnswersState = "answers" AnswersBucket = TellerModule + "_" + AnswersState @@ -88,7 +88,3 @@ func PackagesKey(module string) (string, error) { } return meshcatalog.PackagesByModuleKey(keys.EscapeName(module)), nil } - -// trunkSourceKey is the key of the gate's state source for the trunk, the watched branch `main` (design 47 G §305: the source -// is kept per watched branch, meshdeliveryclient.SourceKey; the view's reads are of the trunk's). -var trunkSourceKey = meshdeliveryclient.SourceKey("main") diff --git a/modules/mesh-delivery-view/internal/view/reach.go b/modules/mesh-delivery-view/internal/view/reach.go index cf21c245..92c8b245 100644 --- a/modules/mesh-delivery-view/internal/view/reach.go +++ b/modules/mesh-delivery-view/internal/view/reach.go @@ -9,7 +9,6 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" "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" @@ -22,14 +21,15 @@ import ( // head as its event delivered, per repository. Nothing here records a merge as delivered (ADR 0429 §1): a merge at or before // the reached head is on the machines, derived by whoever reads it. -// ReachHistory names the watch of a bucket with its history, as Initial is told it: the modules' and the source's history -// give each fingerprint's first commit and each head's source revision. +// ReachHistory names the watch of a bucket with its history, as Initial is told it: the modules' history gives each +// fingerprint's first commit. Each head's source revision is from the branch's sources heard since the start (design 47 G +// A-173, A-225), laid over the reach the view's own state kept. func ReachHistory(bucket string) string { return bucket + "#history" } // reachBuckets are the watches the derivation waits for before it writes or says anything, so a start does not write a reach -// derived from half its inputs. -var reachBuckets = []string{ModulesBucket, AssignerWantsBucket, HaveBucket, SourceBucket, - TakenBucket, ReachHistory(ModulesBucket), ReachHistory(SourceBucket)} +// derived from half its inputs; the view's own reach first, which a restart lays the derivation over (A-225). +var reachBuckets = []string{ModulesBucket, AssignerWantsBucket, HaveBucket, TakenBucket, ReachHistory(ModulesBucket), + viewstate.ReachBucket} // ReachWrite is a reach value to write under its key in the view's state reach. type ReachWrite struct { @@ -45,8 +45,7 @@ func (m *Model) SetReachWriter(put func(ReachWrite) error) { m.reachPut = put } -// applyHistory takes one revision of a watch with history: of the indexer's modules (each module's values in order) and of -// the gate's source (each revision's heads). +// applyHistory takes one revision of a watch with history: of the catalog half's modules (each module's values in order). func (m *Model) applyHistory(e Entry) { switch e.Bucket { case ModulesBucket: @@ -70,15 +69,6 @@ func (m *Model) applyHistory(e Entry) { h = h[len(h)-viewstate.MostChanges:] } m.modHist[e.Key] = h - case SourceBucket: - if e.Key != trunkSourceKey || e.Deleted || e.Revision == 0 { - return - } - src, err := meshdeliveryclient.DecodeSource(e.Value) - if err != nil { - return - } - m.srcHist[e.Revision] = src.Heads } } @@ -88,11 +78,13 @@ type headAt struct { rev uint64 } -// headsOf are the heads of a repository in the source's history, oldest first, each with the first revision naming it. -func (m *Model) headsOf(repository string) []headAt { +// headsOf are the heads of a repository in the history of one branch's source (escaped), oldest first, each with the first +// revision naming it. +func (m *Model) headsOf(branch, repository string) []headAt { var out []headAt - for _, rev := range slices.Sorted(maps.Keys(m.srcHist)) { - for r, head := range m.srcHist[rev] { + hist := m.srcHist[branch] + for _, rev := range slices.Sorted(maps.Keys(hist)) { + for r, head := range hist[rev] { if !trunk.SameRepository(r, repository) || head == "" { continue } @@ -136,10 +128,11 @@ func (m *Model) wantKeys() map[string][]string { return out } -// deriveReach is the reach of one repository from the model as it stands; wants are the wants' keys by module (wantKeys). -func (m *Model) deriveReach(repository string, wants map[string][]string) viewstate.Reach { +// deriveReach is the reach of one repository from the source of its branch (escaped) as the model holds it; wants are the +// wants' keys by module (wantKeys). +func (m *Model) deriveReach(branch, repository string, wants map[string][]string) viewstate.Reach { r := viewstate.Reach{Repository: trunk.RepositoryName(repository), Modules: map[string]viewstate.ModuleReach{}} - heads := m.headsOf(repository) + heads := m.headsOf(branch, repository) key, _ := trunk.RepositoryKey(repository) if t, ok := m.taken[key]; ok { r.Indexed = t.Merge @@ -178,6 +171,11 @@ func (m *Model) deriveReach(repository string, wants map[string][]string) viewst r.Reached, r.ReachedSrc = h.commit, h.rev } } + // The history is the revisions heard since the start: the reach the view's own state kept at the start stands until one + // derived from them is newer (A-225). + if k, ok := m.kept[key]; ok && k.Reached != "" && (r.Reached == "" || r.ReachedSrc < k.ReachedSrc) { + r.Reached, r.ReachedSrc = k.Reached, k.ReachedSrc + } if n := len(heads); n > 0 && !sameCommit(heads[n-1].commit, r.Reached) { if indexed != 0 && indexed < heads[n-1].rev { holds = append(holds, fmt.Sprintf("the indexer has not yet taken in %.12s", heads[n-1].commit)) @@ -387,11 +385,30 @@ func (m *Model) reachReady() bool { return false } } - return m.source != nil && m.source.err == "" + return true } -// reachNow derives the reach of every repository the source names, and answers the values that changed and the reached -// heads not yet said. It is called with the lock held. +// branchOf is the escaped branch a repository's reach is derived from: the first branch of its modules' registration (by module +// name, the first that has one), before the first registrations the first branch whose source names the repository (design 47 +// G A-173). "" when none. +func (m *Model) branchOf(repository string) string { + for _, name := range slices.Sorted(maps.Keys(m.modules)) { + if me := m.modules[name]; me.err == "" && trunk.SameRepository(me.value.Repository, repository) && m.modBranch[name] != "" { + return m.modBranch[name] + } + } + for _, b := range slices.Sorted(maps.Keys(m.sources)) { + for r := range m.sources[b].heads { + if trunk.SameRepository(r, repository) { + return b + } + } + } + return "" +} + +// reachNow derives the reach of every repository a watched branch's source names, from the source of the repository's own +// branch (branchOf), and answers the values that changed and the reached heads not yet said. It is called with the lock held. func (m *Model) reachNow() ([]ReachWrite, []DeliveredEvent) { if (m.reachPut == nil && m.deliveredEvent == nil) || !m.reachReady() { return nil, nil @@ -399,12 +416,25 @@ func (m *Model) reachNow() ([]ReachWrite, []DeliveredEvent) { var writes []ReachWrite var events []DeliveredEvent wants := m.wantKeys() - for _, repository := range slices.Sorted(maps.Keys(m.source.heads)) { + type repoAt struct{ branch, repository string } + var repos []repoAt + for _, b := range slices.Sorted(maps.Keys(m.sources)) { + if m.sources[b].err != "" { + continue + } + for _, repository := range slices.Sorted(maps.Keys(m.sources[b].heads)) { + if m.branchOf(repository) == b { + repos = append(repos, repoAt{b, repository}) + } + } + } + for _, ra := range repos { + repository := ra.repository key, err := trunk.RepositoryKey(repository) if err != nil { continue } - r := m.deriveReach(repository, wants) + r := m.deriveReach(ra.branch, repository, wants) if prev, ok := m.reach[key]; !ok || !viewstate.SameReach(prev, r) { if data, err := viewstate.EncodeReach(r); err == nil { m.reach[key] = r @@ -415,7 +445,7 @@ func (m *Model) reachNow() ([]ReachWrite, []DeliveredEvent) { said := strings.Join(notYet, ",") if r.Reached != "" && (r.ReachedSrc > m.reachSaid[key] || said != m.reachNotYet[key]) { m.reachSaid[key], m.reachNotYet[key] = r.ReachedSrc, said - events = append(events, DeliveredEvent{Revision: r.ReachedSrc, Heads: map[string]string{r.Repository: r.Reached}, + events = append(events, DeliveredEvent{Branch: m.sources[ra.branch].branch, Revision: r.ReachedSrc, Heads: map[string]string{r.Repository: r.Reached}, Repository: r.Repository, NotYet: notYet}) } } diff --git a/modules/mesh-delivery-view/internal/view/reach_test.go b/modules/mesh-delivery-view/internal/view/reach_test.go index 06c58aa2..69ebee73 100644 --- a/modules/mesh-delivery-view/internal/view/reach_test.go +++ b/modules/mesh-delivery-view/internal/view/reach_test.go @@ -16,6 +16,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" "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/keys" ) // 1280 (ADR 0556 clauses 2 and 3): the view derives, per repository, how far each module reached and the reached head, writes @@ -30,14 +31,14 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { if err != nil { t.Fatal(err) } - return Entry{Bucket: ModulesBucket, Key: module, Value: v, At: t0, History: history} + return Entry{Bucket: ModulesBucket, Key: module, Value: v, At: t0, History: history, Branch: testBranch} } modIn := func(fp, commit string, src uint64, in *meshcatalog.ModulesByModuleValueInputs2, history bool) Entry { v, err := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: strings.Repeat("a", 64), Repository: web, Commit: commit, Src: int64(src), Path: "web", Inputs2: in}) if err != nil { t.Fatal(err) } - return Entry{Bucket: ModulesBucket, Key: "web", Value: v, At: t0, History: history} + return Entry{Bucket: ModulesBucket, Key: "web", Value: v, At: t0, History: history, Branch: testBranch} } in := func(tree string, pkgs ...string) *meshcatalog.ModulesByModuleValueInputs2 { return &meshcatalog.ModulesByModuleValueInputs2{Tree: tree, Packages: pkgs} @@ -48,12 +49,23 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { later := func(commit string, to *meshcatalog.ModulesByModuleValueInputs2, extra ...Entry) []Entry { return append([]Entry{modIn(fpLater, commit, 12, to, true), modIn(fpLater, commit, 12, to, false)}, extra...) } - src := func(rev uint64, webHead string) Entry { + // Each watched branch's source as its verb source answers it, one revision heard after another (design 47 G A-173). + srcOn := func(branch string, rev uint64, webHead string) Entry { v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "g", Heads: map[string]string{web: webHead, sdk: commitS}}) if err != nil { t.Fatal(err) } - return Entry{Bucket: SourceBucket, Key: trunkSourceKey, Value: v, At: t0, Revision: rev, History: true} + return Entry{Bucket: SourceBucket, Key: keys.EscapeName(branch), Value: v, At: t0, Revision: rev} + } + src := func(rev uint64, webHead string) Entry { return srcOn(testBranch, rev, webHead) } + // The reach the view's own state held at its start (A-225). + kept := func(reached string, rev uint64) Entry { + key, _ := trunk.RepositoryKey(web) + v, err := viewstate.EncodeReach(viewstate.Reach{Repository: web, Reached: reached, ReachedSrc: rev}) + if err != nil { + t.Fatal(err) + } + return Entry{Bucket: viewstate.ReachBucket, Key: key, Value: v, At: t0} } taken := func(repository, merge string) Entry { key, _ := trunk.RepositoryKey(repository) @@ -70,17 +82,18 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { return e } // web changed at A (fpOld) and at C (fpNew); mail first appeared at A. The source named A, B, then C as web's head. - world := func(t *testing.T) []Entry { - cur, _ := meshdeliveryclient.DecodeSource(src(12, commitC).Value) - current := Entry{Bucket: SourceBucket, Key: trunkSourceKey, Revision: 12, At: t0} - current.Value, _ = meshdeliveryclient.EncodeSource(cur) - return []Entry{src(10, commitA), src(11, commitB), src(12, commitC), current, + world := func(t *testing.T, restarted bool) []Entry { + heard := []Entry{src(10, commitA), src(11, commitB), src(12, commitC)} + if restarted { // a start after which only the newest revision is heard, over the reach the state kept (A-225) + heard = []Entry{kept(commitB, 11), src(12, commitC)} + } + return append(heard, mod("web", fpOld, commitA, 10, true), modIn(fpNew, commitC, 12, in("t1", "mesh-sdk/go"), true), mod("mail", fpMail, commitA, 0, true), modIn(fpNew, commitC, 12, in("t1", "mesh-sdk/go"), false), mod("mail", fpMail, commitA, 0, false), taken(web, commitC), taken(sdk, commitS), wantE(t, "ace", "web", meshwantsclient.Want{Fp: fpNew}), haveFP(t, "ace", "web", fpNew, nodeengineclient.HealthOK), wantE(t, "g14", "web", meshwantsclient.Want{Fp: fpNew}), haveFP(t, "g14", "web", fpNew, nodeengineclient.HealthOK), - wantE(t, "ace", "mail", meshwantsclient.Want{Fp: fpMail}), haveFP(t, "ace", "mail", fpMail, nodeengineclient.HealthOK)} + wantE(t, "ace", "mail", meshwantsclient.Want{Fp: fpMail}), haveFP(t, "ace", "mail", fpMail, nodeengineclient.HealthOK)) } for _, tc := range []struct { name string @@ -90,8 +103,17 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { waiting string // words web's module web waits with; "" none notYet []string // the merges the reached head's event says are not yet on the machines refused bool // the bus refuses every write and event until the watches handed over (a start, issue 1126) + // restarted: the view started again, heard only the newest source revision, and its state reach kept commitB (A-225) + restarted bool }{ {name: "every node runs the newest build", reached: commitC}, + // design 47 G A-173: a repository's reach is from the source of its modules' branch; another branch's source moves it not. + {name: "another branch's source naming the repository moves not its reach", reached: commitC, + then: []Entry{srcOn("beta", 13, commitX)}}, + // A-225: after a restart the history is the revisions heard since; the kept reach stands until a newer one is derived. + {name: "after a restart the reach kept stands while the newest head lags", reached: commitB, holds: "web on g14", restarted: true, + then: []Entry{haveFP(t, "g14", "web", fpOld, nodeengineclient.HealthOK)}, waiting: "not yet the"}, + {name: "after a restart a newer head reached replaces the kept one", reached: commitC, restarted: true}, {name: "a node lagging stops the head before the change it lacks", reached: commitB, holds: "web on g14", then: []Entry{haveFP(t, "g14", "web", fpOld, nodeengineclient.HealthOK)}, waiting: "not yet the"}, // ADR 0603 §7: a node away holds no head back, is named, and holds it again once it returns still behind. @@ -134,12 +156,10 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { {name: "a merge that removed the last module is reached at the merge the indexer took in", reached: commitD, then: func() []Entry { head := src(13, commitD) - current := head - current.History = false gone := func(module string, history bool) Entry { return Entry{Bucket: ModulesBucket, Key: module, Deleted: true, At: t0, History: history} } - return []Entry{head, current, gone("web", true), gone("mail", true), gone("web", false), gone("mail", false), taken(web, commitD)} + return []Entry{head, gone("web", true), gone("mail", true), gone("web", false), gone("mail", false), taken(web, commitD)} }()}, // 1126: the view started while the bus did not answer its writes; its reach and its event are made at the next change it // takes in, though the derivation is the same, so the state reach never keeps an older have than node_haves says. @@ -168,7 +188,7 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) { } return nil }) - fill(m, world(t)...) + fill(m, world(t, tc.restarted)...) if len(written) != 0 { t.Fatalf("written before every watch handed over: %v", written) } diff --git a/modules/mesh-delivery-view/internal/view/read.go b/modules/mesh-delivery-view/internal/view/read.go index df828161..b3209763 100644 --- a/modules/mesh-delivery-view/internal/view/read.go +++ b/modules/mesh-delivery-view/internal/view/read.go @@ -66,9 +66,12 @@ type Reader struct { PredefinedState, ConfigCandidatesState StateRead // PackagesState is the indexer's published packages (PackagesBucket, ADR 0593 §2); unset reads none. PackagesState StateRead - // GraphState is the indexer's module graph (graph.Bucket, ADR 0603 §2), SourceState the gate's source and - // ProvisionsState the provisioner's keys: what view_tiers reads (tiers.go). Unset reads none, said in the answer. - GraphState, SourceState, ProvisionsState StateRead + // GraphState is the indexer's module graph (graph.Bucket, ADR 0603 §2) and ProvisionsState the provisioner's keys: what + // view_tiers reads (tiers.go). Unset reads none, said in the answer. + GraphState, ProvisionsState StateRead + // Sources is the newest source revision of each watched branch, by branch name, as the seat mesh-delivery's verb source + // answers it (design 47 G §507, A-173): what view_tiers given no revision answers per branch. Unset, it must be named. + Sources func() (map[string]uint64, error) // Pending reads the indexer's waits through its tool indexer_pending (ADR 0633 §5); unset names no module held back. Pending func() (takein.PendingAnswer, error) // BuildLog reads the builder's log from a time (stdio.Log(builds.BuildLogRead).Since); unset reads none. diff --git a/modules/mesh-delivery-view/internal/view/serve.go b/modules/mesh-delivery-view/internal/view/serve.go index b3294ff3..c3464576 100644 --- a/modules/mesh-delivery-view/internal/view/serve.go +++ b/modules/mesh-delivery-view/internal/view/serve.go @@ -251,13 +251,15 @@ func wantText(s Snapshot) [][]string { return out } -// sourceText is the source revision's derived state in one line, then one line for each module on a node it waits for. +// sourceText is each watched branch's source revision's derived state in one line, then one line for each module on a node it +// waits for. func sourceText(s Snapshot) []string { - if s.Source == nil { - return nil + var out []string + for _, src := range s.Sources { + out = append(out, fmt.Sprintf("%s: source revision %d: %s", src.Branch, src.Revision, src.State)) + out = append(out, src.Waiting...) } - out := []string{fmt.Sprintf("source revision %d: %s", s.Source.Revision, s.Source.State)} - return append(out, s.Source.Waiting...) + return out } // backlogText is one line per language and kind: its waiting asks, the asks held by a node, the oldest waiting ask's age. diff --git a/modules/mesh-delivery-view/internal/view/source.go b/modules/mesh-delivery-view/internal/view/source.go new file mode 100644 index 00000000..d0c05680 --- /dev/null +++ b/modules/mesh-delivery-view/internal/view/source.go @@ -0,0 +1,71 @@ +package view + +import ( + "encoding/json" + "fmt" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// SourceVerb is the seat mesh-delivery's verb source as the view's invokes names it: each watched branch's source revision, +// asked of the seat and never read from the gate's state (design 47 G §507, §512, A-173). +const SourceVerb = "seat:" + meshdelivery.Seat + "." + meshdelivery.VerbSource + +// SourceOf asks the seat, through call, for one watched branch's source revision (branch as named) and reads its declared +// answer; an answer that carries an error is that error. +func SourceOf(call func(key string, body any) (json.RawMessage, error), branch string) (meshdelivery.SourceAnswer, error) { + raw, err := call(SourceVerb, meshdelivery.SourceInput{Branch: keys.EscapeName(branch)}) + if err != nil { + return meshdelivery.SourceAnswer{}, fmt.Errorf("%s of %s: %w", SourceVerb, branch, err) + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return meshdelivery.SourceAnswer{}, fmt.Errorf("%s of %s answered: %s", SourceVerb, branch, said) + } + var a meshdelivery.SourceAnswer + if err := json.Unmarshal(answer, &a); err != nil { + return a, fmt.Errorf("the answer of %s of %s is not a source revision: %w", SourceVerb, branch, err) + } + if a.Error != "" { + return a, fmt.Errorf("%s of %s: %s", SourceVerb, branch, a.Error) + } + return a, nil +} + +// SourceEntry is a branch's source answer as the model's entry of SourceBucket: keyed by the escaped branch, its revision the +// source revision, its value the heads by repository; a branch the gate holds no source revision of yet is a removal. +func SourceEntry(branch string, a meshdelivery.SourceAnswer) (Entry, error) { + heads := map[string]string{} + for _, h := range a.Repositories { + heads[h.Repository] = h.Head + } + e := Entry{Bucket: SourceBucket, Key: keys.EscapeName(branch), Revision: uint64(a.Revision)} + if a.Revision == 0 && len(heads) == 0 { // the gate holds no source revision of the branch yet + e.Deleted = true + return e, nil + } + v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: a.Group, Heads: heads}) + if err != nil { + return Entry{}, err + } + e.Value = v + return e, nil +} + +// SourcesOf is Reader.Sources over the watched branches followed now: each one's newest source revision through call. +func SourcesOf(call func(key string, body any) (json.RawMessage, error), followed func() []string) func() (map[string]uint64, error) { + return func() (map[string]uint64, error) { + out := map[string]uint64{} + for _, branch := range followed() { + a, err := SourceOf(call, branch) + if err != nil { + return nil, err + } + out[branch] = uint64(a.Revision) + } + return out, nil + } +} diff --git a/modules/mesh-delivery-view/internal/view/tiers.go b/modules/mesh-delivery-view/internal/view/tiers.go index 03aa92cf..11d5726c 100644 --- a/modules/mesh-delivery-view/internal/view/tiers.go +++ b/modules/mesh-delivery-view/internal/view/tiers.go @@ -41,6 +41,8 @@ const TiersMost = 500 // could not be read (then every module is in the first tier), Cycle a cycle among them (its modules form the last // tier), and Away the nodes left out. type Tiers struct { + // Branch is the watched branch whose newest source revision it is, when view_tiers was given no revision (A-173). + Branch string `json:"branch,omitempty"` Revision uint64 `json:"revision"` Modules int `json:"modules"` Tiers [][]TierModule `json:"tiers"` @@ -76,20 +78,43 @@ type TierWait struct { Since time.Time `json:"since,omitzero"` } -// Tiers works out the tiers of a source revision, the gate's newest when revision is 0. +// BranchTiers is view_tiers' answer given no revision: the tiers of each watched branch's newest source revision, in branch +// order (design 47 G A-173). +type BranchTiers struct { + Branches []Tiers `json:"branches"` +} + +// Newest works out the tiers of each watched branch's newest source revision, as the gate's verb source answers it. +func (r Reader) Newest() (BranchTiers, error) { + if r.Sources == nil { + return BranchTiers{}, fmt.Errorf("the gate's source cannot be read here: name the source revision") + } + revs, err := r.Sources() + if err != nil { + return BranchTiers{}, fmt.Errorf("the gate's source: %w", err) + } + out := BranchTiers{Branches: []Tiers{}} + for _, branch := range slices.Sorted(maps.Keys(revs)) { + if revs[branch] == 0 { + continue + } + t, err := r.Tiers(revs[branch]) + if err != nil { + return BranchTiers{}, fmt.Errorf("%s: %w", branch, err) + } + t.Branch = branch + out.Branches = append(out.Branches, t) + } + if len(out.Branches) == 0 { + return BranchTiers{}, fmt.Errorf("the gate holds no source revision of any watched branch yet") + } + return out, nil +} + +// Tiers works out the tiers of one source revision (not 0). func (r Reader) Tiers(revision uint64) (Tiers, error) { if revision == 0 { - if r.SourceState.GetEntry == nil { - return Tiers{}, fmt.Errorf("the gate's source cannot be read here: name the source revision") - } - _, rev, err := r.SourceState.GetEntry(trunkSourceKey) - if err != nil { - return Tiers{}, fmt.Errorf("the gate's source: %w", err) - } - if rev == 0 { - return Tiers{}, fmt.Errorf("the gate holds no source revision yet") - } - revision = rev + return Tiers{}, fmt.Errorf("name the source revision") } modules, err := values(r.State) if err != nil { @@ -453,5 +478,8 @@ func (r Reader) TiersOf(args map[string]any) (any, error) { if err != nil { return nil, err } + if revision == 0 { + return r.Newest() + } return r.Tiers(revision) } diff --git a/modules/mesh-delivery-view/internal/view/view.go b/modules/mesh-delivery-view/internal/view/view.go index 279086db..cfc4a2e9 100644 --- a/modules/mesh-delivery-view/internal/view/view.go +++ b/modules/mesh-delivery-view/internal/view/view.go @@ -26,6 +26,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildrecord" "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/keys" "git.novox.be/novox/mesh-sdk/go/motion" ) @@ -36,8 +37,12 @@ type Entry struct { Value []byte At time.Time Deleted bool - // Revision is the bucket revision of the change, when the watch gives one (the source revision is the gate's source key's). + // Revision is the bucket revision of the change, when the watch gives one; for a branch's source (SourceBucket, keyed by the + // escaped branch) it is the source revision the gate's verb source answers. Revision uint64 + // Branch is, for a module's entry (ModulesBucket), the watched branch it is read from: the first branch of the module's + // registration, before the first registrations the first branch the gate's source answers that lists it (design 47 G A-173). + Branch string // History says the change comes from a watch with history (ReachHistory): it feeds only the history the reach reads. History bool } @@ -87,9 +92,9 @@ type Snapshot struct { Asks []AskRow `json:"asks"` // Wants is every want of the assigner beside the have of its node and where it stands (novox/hq ADR 0539 §8, issue 1906 D7). Wants []WantRow `json:"wants"` - // Source is the state of the mesh's source revision, derived from the wants and the haves (ADR 0429 §1): recorded by no part. - // Nil when the gate holds no source yet. - Source *SourceRow `json:"source,omitempty"` + // Sources is the state of each watched branch's source revision, derived from the wants and the haves (ADR 0429 §1; + // design 47 G A-173): recorded by no part, by branch name. Empty while the gate's verb source answers no branch's. + Sources []SourceRow `json:"sources,omitempty"` // Backlog is, for each language and kind of build ask with one waiting or held, the asks waiting for a node, the asks a node // holds and the oldest waiting ask's age (issue 1887 A8a). Backlog []BacklogRow `json:"backlog"` @@ -129,10 +134,12 @@ const ( SourceNotDelivered = "not delivered" ) -// SourceRow is the mesh's source revision (the revision of the gate's source key) and whether it is delivered: every want equals its -// module's fingerprint at the indexer and every have equals its want, running and healthy, and every node a module is assigned to -// has a want. Waiting names each module on a node that is not there yet, and why. +// SourceRow is one watched branch's source revision (as the gate's verb source answers it) and whether it is delivered: every +// want of a module whose branch it is equals its module's fingerprint in the catalog half and every have equals its want, running +// and healthy, and every node such a module is assigned to has a want. Waiting names each module on a node that is not there +// yet, and why. type SourceRow struct { + Branch string `json:"branch"` Revision uint64 `json:"revision"` State string `json:"state"` Waiting []string `json:"waiting,omitempty"` @@ -235,12 +242,13 @@ type Model struct { modules map[string]moduleEntry // the indexer's modules, by module keys map[string]keyEntry // the builder's build keys, by . wants map[string]wantEntry // the assigner's wants, by . - source *sourceEntry // the gate's source key + sources map[string]*sourceEntry // each watched branch's source, by escaped branch (design 47 G A-173) + modBranch map[string]string // the escaped branch each module's entry is read from, by module deliveredEvent func(DeliveredEvent) error motionEvent func(MotionEvent) error told map[string]bool // the motion events said, by their ids initial map[string]bool // the buckets whose watch handed over what stood at its start - said uint64 // the source revision whose delivery was said + said map[string]uint64 // the source revision whose delivery was said, by escaped branch aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key chosen map[string]viewstate.ViewAnswer // the answers the view wrote, by ask key writer Writer @@ -256,7 +264,8 @@ type Model struct { stuck map[string]stuckEntry // how far each repository reached (reach.go, ADR 0556) modHist map[string][]meshcatalog.ModulesByModuleValue // the indexer's values of each module, oldest first - srcHist map[uint64]map[string]string // the source's heads, by revision + srcHist map[string]map[uint64]map[string]string // each branch's source heads heard since the start, by escaped branch and revision + kept map[string]viewstate.Reach // the reach the view's own state held at its start, by repository key (A-225) taken map[string]takein.Taken // the indexer's taken, by repository key reach map[string]viewstate.Reach // the reach derived last, by repository key reachSaid map[string]uint64 // the reached head's source revision said last, by repository key @@ -326,6 +335,7 @@ type wantEntry struct { } type sourceEntry struct { + branch string // the branch's name revision uint64 group string heads map[string]string @@ -359,7 +369,8 @@ func New() *Model { return &Model{placed: meshmoduleassignerclient.NewPlaced(), builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, wants: map[string]wantEntry{}, told: map[string]bool{}, initial: map[string]bool{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]viewstate.ViewAnswer{}, answers: map[string]string{}, answerOf: map[string]meshdeliveryclient.Answer{}, holds: map[string]heldEntry{}, newest: map[string]string{}, haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{}, presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}, - modHist: map[string][]meshcatalog.ModulesByModuleValue{}, srcHist: map[uint64]map[string]string{}, taken: map[string]takein.Taken{}, + modHist: map[string][]meshcatalog.ModulesByModuleValue{}, srcHist: map[string]map[uint64]map[string]string{}, kept: map[string]viewstate.Reach{}, + sources: map[string]*sourceEntry{}, modBranch: map[string]string{}, said: map[string]uint64{}, taken: map[string]takein.Taken{}, reach: map[string]viewstate.Reach{}, reachSaid: map[string]uint64{}, reachNotYet: map[string]string{}} } @@ -367,21 +378,23 @@ func New() *Model { // the source revision delivered, the event of it is said once (SetDeliveredEvent), after the model's lock is released. func (m *Model) Apply(e Entry) { m.mu.Lock() - var event *DeliveredEvent + var events []DeliveredEvent var motion []MotionEvent if e.History { m.applyHistory(e) } else { m.apply(e) - event = m.deliveredNow() + events = m.deliveredNow() motion = m.motionNow(e) } writes, reached := m.reachNow() say, put := m.deliveredEvent, m.reachPut sayMotion := m.motionEvent m.mu.Unlock() - if event != nil && say != nil && say(*event) != nil { - m.unsay(*event) + for _, event := range events { + if say != nil && say(event) != nil { + m.unsay(event) + } } m.sendReach(put, say, writes, reached) m.sayMotion(sayMotion, motion) @@ -449,8 +462,8 @@ const unsaidMark = "\x00not said" func (m *Model) unsay(ev DeliveredEvent) { m.mu.Lock() defer m.mu.Unlock() - if m.said == ev.Revision { - m.said = 0 + if b := keys.EscapeName(ev.Branch); m.said[b] == ev.Revision { + delete(m.said, b) } } @@ -462,6 +475,8 @@ func (m *Model) unsay(ev DeliveredEvent) { // is the source revision that named that head and Group is empty; a consumer takes every earlier merge of that repository as on // the machines. type DeliveredEvent struct { + // Branch is the watched branch whose source revision it is (design 47 G A-173). + Branch string `json:"branch"` Revision uint64 `json:"revision"` Group string `json:"group"` Heads map[string]string `json:"heads"` @@ -478,18 +493,26 @@ func (m *Model) SetDeliveredEvent(say func(DeliveredEvent) error) { m.deliveredEvent = say } -// deliveredNow is the event of the source revision when it is delivered now and has not been said, or nil. A revision that went -// bad after it was delivered and recovered is not said again. -func (m *Model) deliveredNow() *DeliveredEvent { - if m.deliveredEvent == nil || m.source == nil || m.source.err != "" || m.said == m.source.revision { +// deliveredNow is the event of each watched branch's source revision that is delivered now and has not been said, in branch +// order. A revision that went bad after it was delivered and recovered is not said again. +func (m *Model) deliveredNow() []DeliveredEvent { + if m.deliveredEvent == nil { return nil } - row := m.sourceRow() - if row == nil || row.State != SourceDelivered || row.Wants == 0 { - return nil + var out []DeliveredEvent + for _, b := range slices.Sorted(maps.Keys(m.sources)) { + src := m.sources[b] + if src.err != "" || m.said[b] == src.revision { + continue + } + row := m.sourceRow(b) + if row == nil || row.State != SourceDelivered || row.Wants == 0 { + continue + } + m.said[b] = src.revision + out = append(out, DeliveredEvent{Branch: src.branch, Revision: src.revision, Group: src.group, Heads: src.heads}) } - m.said = m.source.revision - return &DeliveredEvent{Revision: m.source.revision, Group: m.source.group, Heads: m.source.heads} + return out } func (m *Model) apply(e Entry) { @@ -542,10 +565,16 @@ func (m *Model) apply(e Entry) { case ModulesBucket: if e.Deleted { delete(m.modules, e.Key) + delete(m.modBranch, e.Key) return } mod, err := meshcatalogclient.DecodeModule(e.Value) m.modules[e.Key] = moduleEntry{value: mod, err: errText(err)} + if e.Branch != "" { + m.modBranch[e.Key] = keys.EscapeName(e.Branch) + } else { + delete(m.modBranch, e.Key) + } case AssignerAsksBucket: if e.Deleted { delete(m.aasks, e.Key) @@ -593,16 +622,31 @@ func (m *Model) apply(e Entry) { } else { delete(m.taken, e.Key) } - case SourceBucket: - if e.Key != trunkSourceKey { + case SourceBucket: // one watched branch's source, keyed by the escaped branch (design 47 G A-173) + if e.Deleted { + delete(m.sources, e.Key) + delete(m.said, e.Key) return } - if e.Deleted { - m.source = nil + name, err := keys.UnescapeName(e.Key) + if err != nil { return } src, err := meshdeliveryclient.DecodeSource(e.Value) - m.source = &sourceEntry{revision: e.Revision, group: src.Group, heads: src.Heads, err: errText(err)} + m.sources[e.Key] = &sourceEntry{branch: name, revision: e.Revision, group: src.Group, heads: src.Heads, err: errText(err)} + if err == nil && e.Revision != 0 { // the history the reach reads: each revision heard since the start (A-225) + if m.srcHist[e.Key] == nil { + m.srcHist[e.Key] = map[uint64]map[string]string{} + } + m.srcHist[e.Key][e.Revision] = src.Heads + } + case viewstate.ReachBucket: // the view's own state reach, as it stood at the start (A-225) + if m.initial[viewstate.ReachBucket] { + return + } + if r, err := viewstate.DecodeReach(e.Value); err == nil && !e.Deleted { + m.kept[e.Key] = r + } case AnswersBucket: if e.Deleted { delete(m.answers, e.Key) @@ -791,7 +835,7 @@ func (m *Model) snapshot() Snapshot { return unreadable[i].Bucket+"."+unreadable[i].Key < unreadable[j].Bucket+"."+unreadable[j].Key }) return Snapshot{Rows: rows, Nodes: nodes, Offline: offline, Stuck: stuck, Unreadable: unreadable, - Groups: m.groupRows(), Fingerprints: m.fingerprintRows(), Asks: m.askRows(), Wants: m.wantRows(), Source: m.sourceRow()} + Groups: m.groupRows(), Fingerprints: m.fingerprintRows(), Asks: m.askRows(), Wants: m.wantRows(), Sources: m.sourceRows()} } // groupRows are the readable candidate groups, ordered by group, their pull requests by repository and number. @@ -1033,12 +1077,25 @@ func (m *Model) wantRows() []WantRow { return out } -// sourceRow derives the state of the source revision the gate holds (ADR 0429 §1; recorded by no part): delivered when every want -// equals its module's fingerprint at the indexer and every have equals its want, running and healthy, and every node a module is -// assigned to has a want, compared only on the nodes present: a node the watchman says is offline is named away and left out -// (ADR 0603 §7, changing ADR 0429 §1). It is nil when the gate holds no readable source. -func (m *Model) sourceRow() *SourceRow { - if m.source == nil || m.source.err != "" { +// sourceRows are the rows of every watched branch's readable source, in branch order (design 47 G A-173). +func (m *Model) sourceRows() []SourceRow { + var out []SourceRow + for _, b := range slices.Sorted(maps.Keys(m.sources)) { + if row := m.sourceRow(b); row != nil { + out = append(out, *row) + } + } + return out +} + +// sourceRow derives the state of one watched branch's source revision (ADR 0429 §1; recorded by no part), branch escaped: +// delivered when every want of a module whose branch it is equals its module's fingerprint and every have equals its want, +// running and healthy, and every node such a module is assigned to has a want, compared only on the nodes present: a node the +// watchman says is offline is named away and left out (ADR 0603 §7, changing ADR 0429 §1). It is nil when the gate's verb +// answers no readable source of the branch. +func (m *Model) sourceRow(branch string) *SourceRow { + src := m.sources[branch] + if src == nil || src.err != "" { return nil } waiting := []string{} @@ -1057,7 +1114,7 @@ func (m *Model) sourceRow() *SourceRow { continue } mod, ok := m.modules[module] - if w.err != "" || w.value.Removed || !ok || mod.err != "" || mod.value.Fp2 == "" { + if w.err != "" || w.value.Removed || !ok || mod.err != "" || mod.value.Fp2 == "" || m.modBranch[module] != branch { continue } looked++ @@ -1081,12 +1138,12 @@ func (m *Model) sourceRow() *SourceRow { if _, ok := m.wants[wk]; ok { continue } - if mod, ok := m.modules[a.Module]; ok && mod.err == "" && mod.value.Fp2 != "" { + if mod, ok := m.modules[a.Module]; ok && mod.err == "" && mod.value.Fp2 != "" && m.modBranch[a.Module] == branch { waiting = append(waiting, fmt.Sprintf("%s on %s has no want yet", a.Module, a.Node)) } } sort.Strings(waiting) - row := &SourceRow{Revision: m.source.revision, State: SourceDelivered, Wants: looked, Away: slices.Sorted(maps.Keys(away))} + row := &SourceRow{Branch: src.branch, Revision: src.revision, State: SourceDelivered, Wants: looked, Away: slices.Sorted(maps.Keys(away))} if len(waiting) > 0 { row.State, row.Waiting = SourceNotDelivered, waiting } diff --git a/modules/mesh-delivery-view/internal/view/wants_test.go b/modules/mesh-delivery-view/internal/view/wants_test.go index d4c3d6b7..c6e2f0e6 100644 --- a/modules/mesh-delivery-view/internal/view/wants_test.go +++ b/modules/mesh-delivery-view/internal/view/wants_test.go @@ -14,6 +14,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-sdk/go/keys" ) const ( @@ -56,16 +57,32 @@ func moduleE(t *testing.T, module, fp string) Entry { if err != nil { t.Fatal(err) } - return Entry{Bucket: ModulesBucket, Key: module, Value: v, At: t0} + return Entry{Bucket: ModulesBucket, Key: module, Value: v, At: t0, Branch: testBranch} } -func sourceE(t *testing.T, revision uint64) Entry { +// testBranch is the watched branch of these tests: no branch is named in the view's code (design 47 G §532, A-173). +const testBranch = "release" + +func sourceE(t *testing.T, revision uint64) Entry { return sourceOn(t, testBranch, revision) } + +// sourceOn is one watched branch's source as the gate's verb source answers it (A-173). +func sourceOn(t *testing.T, branch string, revision uint64) Entry { t.Helper() v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "g", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}}) if err != nil { t.Fatal(err) } - return Entry{Bucket: SourceBucket, Key: trunkSourceKey, Value: v, At: t0, Revision: revision} + return Entry{Bucket: SourceBucket, Key: keys.EscapeName(branch), Value: v, At: t0, Revision: revision} +} + +// sourceRowOf is the snapshot's source row of one branch, nil when it has none. +func sourceRowOf(m *Model, branch string) *SourceRow { + for _, r := range m.Snapshot().Sources { + if r.Branch == branch { + return &r + } + } + return nil } // 1906/D7 AC7: every want is a row beside the have of its node, with where it stands. @@ -107,44 +124,66 @@ func TestEveryWantIsARowAgainstItsHave(t *testing.T) { // want, healthy; one failed have shows it not delivered and names the module. func TestASourceRevisionIsDerivedFromWantsAndHaves(t *testing.T) { m := New() + var said []DeliveredEvent + m.SetDeliveredEvent(func(e DeliveredEvent) error { + if e.Repository == "" { + said = append(said, e) + } + return nil + }) fill(m, sourceE(t, 14355), moduleE(t, "web", fpNew), moduleE(t, "mail", fpNew), place(t, "ace", "web"), place(t, "g14", "web"), wantE(t, "ace", "web", meshwantsclient.Want{Fp: fpNew}), haveFP(t, "ace", "web", fpNew, nodeengineclient.HealthOK), wantE(t, "g14", "web", meshwantsclient.Want{Fp: fpNew}), haveFP(t, "g14", "web", fpNew, nodeengineclient.HealthOK), wantE(t, "ace", "mail", meshwantsclient.Want{Fp: fpNew}), haveFP(t, "ace", "mail", fpNew, nodeengineclient.HealthOK)) - s := m.Snapshot().Source + s := sourceRowOf(m, testBranch) if s == nil || s.Revision != 14355 || s.State != SourceDelivered || len(s.Waiting) != 0 { t.Fatalf("source = %+v, want revision 14355 delivered", s) } + // design 47 G A-173: each watched branch's source has its own row and its own delivered, over the modules whose branch it + // is: a module of another branch not yet run holds that branch's revision alone. + cms := moduleE(t, "cms", fpNew) + cms.Branch = "beta" + fill(m, sourceOn(t, "beta", 14360), cms, wantE(t, "ace", "cms", meshwantsclient.Want{Fp: fpNew})) + if b := sourceRowOf(m, "beta"); b == nil || b.Revision != 14360 || b.State != SourceNotDelivered || b.Wants != 1 { + t.Errorf("the branch beta's source = %+v, want revision 14360 not delivered over its one want", b) + } + if s := sourceRowOf(m, testBranch); s == nil || s.State != SourceDelivered || s.Wants != 3 { + t.Errorf("the branch %s's source = %+v, want delivered over its own three wants", testBranch, s) + } + fill(m, haveFP(t, "ace", "cms", fpNew, nodeengineclient.HealthOK)) + if len(said) != 2 || said[0].Branch != testBranch || said[0].Revision != 14355 || said[1].Branch != "beta" || said[1].Revision != 14360 { + t.Errorf("delivered said = %+v, want %s at 14355, then beta at 14360", said, testBranch) + } fill(m, haveFP(t, "g14", "web", fpNew, nodeengineclient.HealthFailed)) - s = m.Snapshot().Source + s = sourceRowOf(m, testBranch) if s.State != SourceNotDelivered || len(s.Waiting) != 1 || !strings.Contains(s.Waiting[0], "web") || !strings.Contains(s.Waiting[0], "g14") { t.Errorf("a failed have: %+v", s) } fill(m, haveFP(t, "g14", "web", fpNew, nodeengineclient.HealthOK), wantE(t, "ace", "mail", meshwantsclient.Want{Fp: fpOld})) - s = m.Snapshot().Source + s = sourceRowOf(m, testBranch) if s.State != SourceNotDelivered || !strings.Contains(strings.Join(s.Waiting, ";"), "mail") { t.Errorf("a want that is not the module's fingerprint: %+v", s) } fill(m, wantE(t, "ace", "mail", meshwantsclient.Want{Fp: fpNew}), place(t, "novox", "web")) - s = m.Snapshot().Source + s = sourceRowOf(m, testBranch) if s.State != SourceNotDelivered || !strings.Contains(strings.Join(s.Waiting, ";"), "novox") { t.Errorf("a node assigned the module with no want: %+v", s) } // ADR 0603 §7: a node away holds nothing back and is named; it counts once it returns and runs its wants. fill(m, presenceE(t, "novox", false, t0)) - s = m.Snapshot().Source + s = sourceRowOf(m, testBranch) if s.State != SourceDelivered || len(s.Away) != 1 || s.Away[0] != "novox" { t.Errorf("a node away: %+v", s) } fill(m, presenceE(t, "novox", true, t0)) - if s = m.Snapshot().Source; s.State != SourceNotDelivered || len(s.Away) != 0 { + if s = sourceRowOf(m, testBranch); s.State != SourceNotDelivered || len(s.Away) != 0 { t.Errorf("the node returned and runs nothing yet: %+v", s) } } func TestNoSourceRevisionHeldMeansNoStateIsDerived(t *testing.T) { - if s := New().Snapshot().Source; s != nil { + if s := New().Snapshot().Sources; len(s) != 0 { t.Errorf("source = %+v", s) } } diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index aa549fe3..d5793658 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -132,7 +132,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if m.Module != "mesh-delivery-view" { t.Errorf("module = %q", m.Module) } - want := []string{"mesh-conditions.conditions", "mesh-config.resolved", "mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.source", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "mesh-healers.attempts", "mesh-provisioner.provisions", "mesh-vault.signers", "node-build-go.pauses", "node-build-go.steps", "node-build-image.pauses", "node-build-image.steps", "node-build-node.pauses", "node-build-node.steps", "node-engine.have"} // the build seats' states by seat, never a build module (design 47 G A-82); // no reads of the indexer's or the assigner's states (ADR 0615 §1, DLV-M4); view_tiers reads the module graph and the provisioner's keys (ADR 0603 §6); view_status reads the keeper's conditions and the healers' attempts (1952/V1); build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 + want := []string{"mesh-conditions.conditions", "mesh-config.resolved", "mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "mesh-healers.attempts", "mesh-provisioner.provisions", "mesh-vault.signers", "node-build-go.pauses", "node-build-go.steps", "node-build-image.pauses", "node-build-image.steps", "node-build-node.pauses", "node-build-node.steps", "node-engine.have"} // the build seats' states by seat, never a build module (design 47 G A-82); no read of the gate's source: each watched branch's is its verb source's (§512, A-173); // no reads of the indexer's or the assigner's states (ADR 0615 §1, DLV-M4); view_tiers reads the module graph and the provisioner's keys (ADR 0603 §6); view_status reads the keeper's conditions and the healers' attempts (1952/V1); build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 got := slices.Clone(m.Reads) slices.Sort(got) if !reflect.DeepEqual(got, want) { @@ -151,13 +151,13 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 {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) + {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", "group", "repositories", "revision"}}, // §507, §614, A-173: the branches before the first registrations, and each watched branch's source (§558: fields G adds) // 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. @@ -202,8 +202,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 // D-11: the holders and buckets the view names are its manifest's (no contract names a holder, novox/hq ADR 0634 §1): a // state it reads directly is in `reads`, one it follows is a family it consumes, and the owners' tools it asks are invoked. t.Run("the names are the manifest's", func(t *testing.T) { - direct := map[string]string{view.SourceBucket: view.GateModule + ".source", - view.PresenceBucket: view.WatchmanModule + ".presence", view.StuckBucket: view.WatchmanModule + ".stuck", view.HaveBucket: view.EngineModule + ".have"} + direct := map[string]string{view.PresenceBucket: view.WatchmanModule + ".presence", view.StuckBucket: view.WatchmanModule + ".stuck", view.HaveBucket: view.EngineModule + ".have"} for bucket, read := range direct { if strings.ReplaceAll(read, ".", "_") != bucket { t.Errorf("the bucket %s is not the state %s", bucket, read) diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 7b42f0ef..19107792 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -89,7 +89,7 @@ {"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"]}, + {"verb": "seat:mesh-delivery.source", "fields": ["branches", "group", "repositories", "revision"]}, "seat:mesh-catalog.seat-definitions", "seat:mesh-hand-acts.hand-act", "seat:mesh-module-assigner.assignments", @@ -106,7 +106,6 @@ "mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", - "mesh-delivery-gate.source", "mesh-conditions.conditions", "mesh-healers.attempts", "mesh-provisioner.provisions",