From d8599855e14d8dd825408a2fdd9c10f9406b61aa Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:25:22 +0200 Subject: [PATCH] Compose the machines a change touches, from the trunk's and the change's manifests and the controller's assignments on the bus, and write the verdict under the checker's key; what needs the facts snapshot is said as not judged (issue 1297, subtasks B3a-B3c; ADR 0446 clauses 1-3, ADR 0528 clause 5, ADR 0482 clause 3 as R4g delivers it) --- .../integration_test.go | 70 ++++++ .../cmd/mesh-delivery-assigner/main.go | 95 ++++++- .../cmd/mesh-delivery-assigner/pass.go | 81 ++++++ .../cmd/mesh-delivery-assigner/world.go | 231 ++++++++++++++++++ .../cmd/mesh-delivery-assigner/world_test.go | 90 +++++++ .../internal/assign/compose.go | 196 +++++++++++++++ .../internal/assign/compose_test.go | 116 +++++++++ 7 files changed, 877 insertions(+), 2 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go create mode 100644 modules/mesh-delivery-assigner/internal/assign/compose.go create mode 100644 modules/mesh-delivery-assigner/internal/assign/compose_test.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 739b315..d45b86a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -17,7 +17,9 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/announce" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/provider" ) // bed is an embedded bus with JetStream, every bucket the assigner reads and keeps, and the stream the mesh's events @@ -142,3 +144,71 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) { } } } + +// putJSON writes a value into one of the bed's buckets. +func (b *bed) putJSON(t *testing.T, bucket, key string, value []byte) { + t.Helper() + if _, err := b.buckets[bucket].Put(b.ctx, key, value); err != nil { + t.Fatalf("%s %s: %v", bucket, key, err) + } +} + +// manifest keeps a manifest in the indexer's bucket by its digest and answers the digest. +func (b *bed) manifest(t *testing.T, module, body string) string { + t.Helper() + digest := delivery.ManifestDigest([]byte(body)) + key, _ := delivery.ManifestsKey(digest) + value, err := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: body}) + if err != nil { + t.Fatal(err) + } + b.putJSON(t, delivery.ManifestsBucket, key, value) + return digest +} + +// 1297/AC1: against the real bus, a fixture change touching one module on one machine is composed and answered within +// seconds under the checker's key of the check, naming the machine composed and the mesh-sdk it composed with. +func TestAChangeIsComposedOnTheBus(t *testing.T) { + b := startBed(t, "") + trunk := b.manifest(t, "web", `{"module":"web","version":"1"}`) + changed := b.manifest(t, "web", `{"module":"web","version":"2"}`) + mod, err := delivery.EncodeModule(delivery.Module{FP2: delivery.SourcePrefix2 + strings.Repeat("a", 64), Manifest: trunk, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath}) + if err != nil { + t.Fatal(err) + } + b.putJSON(t, delivery.ModulesBucket, "web", mod) + b.putJSON(t, provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"build-1"}`)) + b.putJSON(t, provider.AssignmentsBucket, "ace.other", []byte(`{"node":"ace","module":"other","build":"build-2"}`)) + heads := []delivery.Head{{Repository: "novox/web", Number: 7, Head: strings.Repeat("2", 40)}} + hash, _ := delivery.HeadsHash(heads) + cand, err := delivery.EncodeCandidate(delivery.Candidate{Group: "feat/x", HeadsHash: hash, Pulls: []delivery.CandidatePull{{Head: heads[0], + Modules: []string{"web"}, Fps: map[string]delivery.CandidateModule{"web": {FP2: delivery.SourcePrefix2 + strings.Repeat("b", 64), Manifest: changed}}}}}) + if err != nil { + t.Fatal(err) + } + ckey, _ := delivery.CandidatesKey("feat/x") + b.putJSON(t, delivery.CandidatesBucket, ckey, cand) + began := time.Now() + b.start(t) + key, _ := delivery.CompositionKey("feat/x", hash) + var v delivery.CompositionVerdict + for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(50 * time.Millisecond) { + if time.Now().After(deadline) { + t.Fatalf("no composition under %s after 10 s", key) + } + e, err := b.buckets[delivery.AssignerCompositionsBucket].Get(b.ctx, key) + if err != nil { + continue + } + if v, err = delivery.DecodeCompositionVerdict(e.Value()); err != nil { + t.Fatal(err) + } + break + } + if v.State != delivery.CompositionPass || len(v.Machines) != 1 || v.Machines[0] != "g14" || v.SDK == "" { + t.Fatalf("composition = %+v, want a pass on g14 alone naming its mesh-sdk", v) + } + if took := time.Since(began); took > 5*time.Second { + t.Errorf("the composition took %s", took) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 4ac7045..9bfae45 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -122,6 +122,97 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro logger.Printf("what the assigner keeps could not be said on the bus, so its readers are not granted until it is: %v", err) } logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs)) - <-ctx.Done() - return nil + return follow(ctx, kvs, logger) +} + +// watched are the buckets whose every value a pass reads; the manifests are read by digest when a pass needs one, and +// watched only for keys, so a manifest that arrives wakes a composition that waits for it. +var watched = []string{ + provider.AssignmentsBucket, + delivery.ControllerGrantsBucket, + delivery.BuilderBuildsBucket, + delivery.CandidatesBucket, + delivery.ModulesBucket, + delivery.PresenceBucket, + delivery.HaveBucket, + delivery.AssignerCompositionsBucket, + delivery.AssignerDifferencesBucket, + delivery.AssignerWantsBucket, +} + +// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after +// each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends. +func follow(ctx context.Context, kvs buckets, logger *log.Logger) error { + w := newWorld() + dirty := make(chan struct{}, 1) + wake := func() { + select { + case dirty <- struct{}{}: + default: + } + } + errs := make(chan error, len(watched)+1) + settled := make(chan struct{}, len(watched)+1) + watch := func(name string, opts ...jetstream.WatchOpt) error { + kw, err := kvs[name].WatchAll(ctx, opts...) + if err != nil { + return fmt.Errorf("the bucket %s cannot be watched: %w", name, err) + } + go func() { + defer kw.Stop() + for { + select { + case <-ctx.Done(): + errs <- nil + return + case e, ok := <-kw.Updates(): + if !ok { + errs <- fmt.Errorf("the watch of %s ended", name) + return + } + if e == nil { + settled <- struct{}{} + continue + } + var value []byte + if e.Operation() == jetstream.KeyValuePut { + value = e.Value() + } + w.put(name, e.Key(), value) + wake() + } + } + }() + return nil + } + for _, name := range watched { + if err := watch(name); err != nil { + return err + } + } + if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil { + return err + } + for range len(watched) + 1 { + select { + case <-ctx.Done(): + return nil + case err := <-errs: + return err + case <-settled: + } + } + p := &passer{kvs: kvs, w: w, logf: logger.Printf} + wake() + for { + select { + case <-ctx.Done(): + return nil + case err := <-errs: + return err + case <-settled: + case <-dirty: + p.pass(ctx) + } + } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go new file mode 100644 index 0000000..f5c55fc --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -0,0 +1,81 @@ +package main + +import ( + "context" + "errors" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/sdkversion" + + "mesh-delivery-assigner/internal/assign" +) + +// passer makes the assigner's passes: from what the world holds now, it writes each key whose value differs from the +// one written, and only those (the Same functions). +type passer struct { + kvs buckets + w *world + logf func(string, ...any) + sdk string +} + +func (p *passer) pass(ctx context.Context) { + p.compositions(ctx) +} + +// compositions writes the composition check's verdict of every group the indexer holds a candidate for, under the +// checker's key of that check: waits, with why, while the indexer holds what it composes from only in part. +func (p *passer) compositions(ctx context.Context) { + if p.sdk == "" { + p.sdk = sdkversion.Version() + } + written := p.w.read(delivery.AssignerCompositionsBucket) + trunk := p.w.trunk() + var mesh *assign.Mesh + for key, value := range p.w.read(delivery.CandidatesBucket) { + c, err := delivery.DecodeCandidate(value) + if err != nil { + p.logf("the candidate %s cannot be read, so it is not composed: %v", key, err) + continue + } + out, err := delivery.CompositionKey(c.Group, c.HeadsHash) + if err != nil { + p.logf("the candidate %s names no composition key: %v", key, err) + continue + } + began := time.Now() + ch, waits := p.w.change(ctx, kvGetter(p.kvs[delivery.ManifestsBucket]), c, trunk) + var v delivery.CompositionVerdict + if waits != "" { + v = delivery.CompositionVerdict{Group: c.Group, Heads: c.HeadsHash, State: delivery.CompositionWaits, Words: waits} + } else { + if mesh == nil { + m := p.w.mesh(ctx, kvGetter(p.kvs[delivery.ManifestsBucket])) + mesh = &m + } + v = assign.Compose(*mesh, ch, p.sdk, time.Since(began)) + } + if before, err := delivery.DecodeCompositionVerdict(written[out]); err == nil { + before.Took, before.V = v.Took, 0 + if delivery.SameCompositionVerdict(before, v) { + continue + } + } + p.put(ctx, delivery.AssignerCompositionsBucket, out, func() ([]byte, error) { return delivery.EncodeCompositionVerdict(v) }) + } +} + +// put writes one key of the assigner's own bucket; a write that fails is said and made again at the next pass. +func (p *passer) put(ctx context.Context, bucket, key string, encode func() ([]byte, error)) { + value, err := encode() + if err != nil { + p.logf("%s %s is not written: %v", bucket, key, err) + return + } + if _, err := p.kvs[bucket].Put(ctx, key, value); err != nil && !errors.Is(err, context.Canceled) { + p.logf("%s %s could not be written: %v", bucket, key, err) + return + } + p.w.put(bucket, key, value) +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go new file mode 100644 index 0000000..e96c070 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -0,0 +1,231 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/provider" + + "mesh-delivery-assigner/internal/assign" +) + +// noModuleWords is the indexer's reason for a module of a candidate whose directory the merge preview holds no manifest +// for (mesh-delivery-indexer internal/index/previews.go NonePreviewNoModule): the change removes the module. +const noModuleWords = "the merge preview names no manifest for the module's directory" + +// assignment holds only the fields of the controller's Assignment the assigner reads (the view's own, mesh-delivery-view +// internal/view/view.go): an assignment stands while it has not ended and has no state. +type assignment struct { + Node string `json:"node"` + Module string `json:"module"` + Build string `json:"build"` + Ended time.Time `json:"ended"` + State string `json:"state"` +} + +func (a assignment) open() bool { return a.Ended.IsZero() && a.State == "" } + +// world is what the assigner watched, as it stands now: every value by bucket and key, and the manifests read by +// digest. The watches write it; a pass reads it. +type world struct { + mu sync.Mutex + values map[string]map[string][]byte + manifests map[string]parsed +} + +// parsed is a manifest read by digest: its module and its manifest, or why it cannot be read. +type parsed struct { + module string + manifest catalogue.Manifest + err string +} + +func newWorld() *world { + return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}} +} + +// put takes a change of a watched bucket in: a value, or nil for a key deleted or purged. +func (w *world) put(bucket, key string, value []byte) { + w.mu.Lock() + defer w.mu.Unlock() + if w.values[bucket] == nil { + w.values[bucket] = map[string][]byte{} + } + if value == nil { + delete(w.values[bucket], key) + return + } + w.values[bucket][key] = value +} + +// read is a copy of what a bucket holds now. +func (w *world) read(bucket string) map[string][]byte { + w.mu.Lock() + defer w.mu.Unlock() + out := make(map[string][]byte, len(w.values[bucket])) + for k, v := range w.values[bucket] { + out[k] = v + } + return out +} + +// getter reads one key's value; errNotThere when the key does not exist. +type getter func(ctx context.Context, key string) ([]byte, error) + +var errNotThere = errors.New("no such key") + +// kvGetter reads keys of a bucket. +func kvGetter(kv jetstream.KeyValue) getter { + return func(ctx context.Context, key string) ([]byte, error) { + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, errNotThere + } + if err != nil { + return nil, err + } + return e.Value(), nil + } +} + +// manifest reads the manifest of a digest from the indexer's bucket once and keeps it: a digest's manifest never changes. +func (w *world) manifest(ctx context.Context, get getter, digest string) parsed { + w.mu.Lock() + p, ok := w.manifests[digest] + w.mu.Unlock() + if ok { + return p + } + key, err := delivery.ManifestsKey(digest) + if err != nil { + return parsed{err: err.Error()} + } + value, err := get(ctx, key) + if errors.Is(err, errNotThere) { + return parsed{err: "the indexer holds no manifest of digest " + digest + " yet"} + } + if err != nil { + return parsed{err: err.Error()} + } + m, err := delivery.DecodeManifest(value) + if err != nil { + p = parsed{err: err.Error()} + } else if man, err := catalogue.ParseManifest([]byte(m.JSON)); err != nil { + p = parsed{module: m.Module, err: err.Error()} + } else { + p = parsed{module: m.Module, manifest: man} + } + w.mu.Lock() + w.manifests[digest] = p + w.mu.Unlock() + return p +} + +// assigned is each node's modules with an open assignment, read from the controller's assignments, in name order. +func (w *world) assigned() (map[string][]string, map[string]assignment) { + out, all := map[string][]string{}, map[string]assignment{} + for _, value := range w.read(provider.AssignmentsBucket) { + var a assignment + if json.Unmarshal(value, &a) != nil || a.Node == "" || a.Module == "" || !a.open() { + continue + } + out[a.Node] = append(out[a.Node], a.Module) + all[a.Node+"."+a.Module] = a + } + for _, mods := range out { + sort.Strings(mods) + } + return out, all +} + +// trunk is every module the indexer holds at the trunk, with its value. +func (w *world) trunk() map[string]delivery.Module { + out := map[string]delivery.Module{} + for key, value := range w.read(delivery.ModulesBucket) { + name, err := delivery.ParseModulesKey(key) + if err != nil { + continue + } + if m, err := delivery.DecodeModule(value); err == nil { + out[name] = m + } + } + return out +} + +// mesh is the mesh as the trunk holds it, for the composition check. +func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh { + m := assign.Mesh{Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}} + m.Assigned, _ = w.assigned() + for name, mod := range w.trunk() { + if mod.Manifest == "" { + continue + } + p := w.manifest(ctx, manifests, mod.Manifest) + if p.err != "" { + m.Unreadable[name] = p.err + continue + } + m.Manifests[name] = p.manifest + } + return m +} + +// change is a candidate's change for the composition check, or why the composition waits for it. +func (w *world) change(ctx context.Context, manifests getter, c delivery.Candidate, trunk map[string]delivery.Module) (assign.Change, string) { + ch := assign.Change{Group: c.Group, Heads: c.HeadsHash, Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}, Twice: map[string][]string{}} + definedIn := map[string]map[string]bool{} + for _, p := range c.Pulls { + if p.Repository == assign.SDKRepository { + ch.SDKHead = p.Head.Head + } + for _, name := range p.Modules { + fp, ok := p.Fps[name] + if !ok || fp.Manifest == "" { + if ok && fp.None2 == noModuleWords { + ch.Removed = append(ch.Removed, name) + continue + } + return ch, fmt.Sprintf("the indexer holds no manifest of %s for %s#%d yet", name, p.Repository, p.Number) + } + if fp.FP2 != "" && fp.FP2 != trunk[name].FP2 { + ch.Rebuilt++ + } + got := w.manifest(ctx, manifests, fp.Manifest) + if strings.HasPrefix(got.err, "the indexer holds no manifest") { + return ch, got.err + } + if got.err != "" { + ch.Unreadable[name] = got.err + continue + } + ch.Manifests[got.module] = got.manifest + if definedIn[got.module] == nil { + definedIn[got.module] = map[string]bool{} + } + definedIn[got.module][p.Repository] = true + if t, ok := trunk[got.module]; ok && t.Repository != "" && t.Repository != p.Repository { + definedIn[got.module][t.Repository] = true + } + } + } + for name, repos := range definedIn { + if len(repos) > 1 { + for r := range repos { + ch.Twice[name] = append(ch.Twice[name], r) + } + sort.Strings(ch.Twice[name]) + } + } + return ch, "" +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go new file mode 100644 index 0000000..ee3a8e3 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go @@ -0,0 +1,90 @@ +package main + +import ( + "context" + "slices" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-assigner/internal/assign" +) + +// fakeManifests is the indexer's manifests bucket in memory. +type fakeManifests map[string][]byte + +func (f fakeManifests) add(t *testing.T, module, body string) string { + t.Helper() + digest := delivery.ManifestDigest([]byte(body)) + value, err := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: body}) + if err != nil { + t.Fatal(err) + } + f[digest] = value + return digest +} + +func (f fakeManifests) get(_ context.Context, key string) ([]byte, error) { + if v, ok := f[key]; ok { + return v, nil + } + return nil, errNotThere +} + +func fp(c byte) string { return delivery.SourcePrefix2 + strings.Repeat(string(c), 64) } + +func pull(repo string, number int, modules map[string]delivery.CandidateModule) delivery.CandidatePull { + p := delivery.CandidatePull{Head: delivery.Head{Repository: repo, Number: number, Head: strings.Repeat("3", 40)}, Fps: modules} + for m := range modules { + p.Modules = append(p.Modules, m) + } + slices.Sort(p.Modules) + return p +} + +// 1297/B3b: a candidate's change reads each module's manifest by its digest, counts the modules rebuilt, takes a +// module with no manifest in the preview as removed, a mesh-sdk member's head, and a module the trunk holds from +// another repository as defined twice. +func TestTheChangeOfACandidate(t *testing.T) { + f := fakeManifests{} + web := f.add(t, "web", `{"module":"web","version":"2"}`) + mail := f.add(t, "mail", `{"module":"mail","version":"1"}`) + trunk := map[string]delivery.Module{"web": {FP2: fp('a'), Repository: "novox/web"}, "mail": {FP2: fp('c'), Repository: "novox/mail"}} + c := delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", Pulls: []delivery.CandidatePull{ + pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: web}, "old": {None2: noModuleWords}}), + pull("novox/other", 2, map[string]delivery.CandidateModule{"mail": {FP2: fp('c'), Manifest: mail}}), + pull(assign.SDKRepository, 3, nil), + }} + ch, waits := newWorld().change(context.Background(), f.get, c, trunk) + if waits != "" { + t.Fatalf("the change waits: %s", waits) + } + if ch.Manifests["web"].Version != "2" || ch.Rebuilt != 1 || !slices.Equal(ch.Removed, []string{"old"}) || ch.SDKHead != strings.Repeat("3", 40) { + t.Errorf("change = %+v", ch) + } + if !slices.Equal(ch.Twice["mail"], []string{"novox/mail", "novox/other"}) || len(ch.Twice) != 1 { + t.Errorf("twice = %v, want mail in novox/mail and novox/other", ch.Twice) + } +} + +// 1297/B3b: a candidate whose module has no manifest digest yet, or whose manifest the indexer does not hold yet, waits, +// saying for what; an unreadable manifest is the change's, not a wait. +func TestAChangeWaitsForItsManifests(t *testing.T) { + f := fakeManifests{} + bad := f.add(t, "web", `{"module":"web","version":"2","scope":"x"}`) + for _, c := range []delivery.CandidatePull{ + pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {None2: "no preview yet"}}), + pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: strings.Repeat("9", 64)}}), + } { + _, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", Pulls: []delivery.CandidatePull{c}}, nil) + if waits == "" { + t.Errorf("a change without its manifest %+v does not wait", c.Fps) + } + } + ch, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", + Pulls: []delivery.CandidatePull{pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: bad}})}}, nil) + if waits != "" || ch.Unreadable["web"] == "" { + t.Errorf("an unreadable manifest: waits %q, change %+v", waits, ch) + } +} diff --git a/modules/mesh-delivery-assigner/internal/assign/compose.go b/modules/mesh-delivery-assigner/internal/assign/compose.go new file mode 100644 index 0000000..6de7006 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/compose.go @@ -0,0 +1,196 @@ +// Package assign is the assigner's judgement, as functions over what it read from the bus: the composition check of a +// change (novox/hq ADR 0446 clauses 1 to 5, ADR 0528 clause 5, issue 1297), the six rules of what each node should +// run (ADR 0429 ยง6, ADR 0528 clause 3) and the comparison with what the controller sends (ADR 0528 clause 4). It reads +// and writes nothing itself: the program hands it what it watched and writes what it answers. +package assign + +import ( + "fmt" + "slices" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// SDKRepository is the repository of mesh-sdk: a group with a member there changes the composition itself. +const SDKRepository = "novox/mesh-sdk" + +// NotJudged names, in a composition's warnings, the failures of ADR 0446 clause 1 the assigner cannot judge from the +// bus, because what they need is only in the controller's facts snapshot, which is not on the bus yet (issue 1461; +// the design limit of R4g, issue 1297): said on every verdict, never passed silently. +const NotJudged = "not judged until the facts snapshot is on the bus (issue 1461): a consumer left out of a grant, a " + + "credential left bound elsewhere, a stored setting the changed definition cannot keep, and a new module the " + + "node-engine's validator would refuse" + +// Mesh is the mesh as the trunk holds it: every module's manifest by name, the manifests that cannot be read with +// why, and each node's open assignments. +type Mesh struct { + Manifests map[string]catalogue.Manifest + Unreadable map[string]string + Assigned map[string][]string +} + +// Change is a group's change as the indexer's candidate holds it: the manifest of every module it moves or adds, those +// it cannot read and why, the modules it removes, every module defined twice with where, the heads of a mesh-sdk +// member (empty when none), and how many modules it rebuilds. +type Change struct { + Group, Heads string + Manifests map[string]catalogue.Manifest + Unreadable map[string]string + Removed []string + Twice map[string][]string + SDKHead string + Rebuilt int +} + +// Touched is the machines a change touches (ADR 0446 clause 3): those running a module the change moves, adds or +// removes, those running a consumer of a provision such a module provides (before or after), and those running a +// provider of a provision such a module requires after the change. A module the change adds that is assigned nowhere +// touches no machine. +func Touched(m Mesh, c Change) []string { + moved := map[string]bool{} + for name := range c.Manifests { + moved[name] = true + } + for name := range c.Unreadable { + moved[name] = true + } + for _, name := range c.Removed { + moved[name] = true + } + provided, required := map[string]bool{}, map[string]bool{} + for name := range moved { + for _, man := range []catalogue.Manifest{m.Manifests[name], c.Manifests[name]} { + for _, o := range man.Provides { + provided[o.Name] = true + } + } + for _, r := range c.Manifests[name].Requires { + required[r] = true + } + } + var out []string + for node, modules := range m.Assigned { + for _, mod := range modules { + man := m.Manifests[mod] + if moved[mod] || slices.ContainsFunc(man.Requires, func(r string) bool { return provided[r] }) || + slices.ContainsFunc(man.Provides, func(o catalogue.Offer) bool { return required[o.Name] }) { + out = append(out, node) + break + } + } + } + sort.Strings(out) + return out +} + +// input is the composition's input for the mesh's catalogue and assignments: the nodes by name with what is assigned +// to them, and every seat the assigned modules claim as held there. What the bus does not carry (places on the +// private network, settings, credentials, licences) is left empty, so a trunk composition and a change's are made +// with the same gaps and compared (ADR 0446 clause 1: what was already broken is said and fails nothing). +func input(catalog map[string]catalogue.Manifest, assigned map[string][]string) compose.Input { + in := compose.Input{Catalogue: catalog} + nodes := make([]string, 0, len(assigned)) + for n := range assigned { + nodes = append(nodes, n) + } + sort.Strings(nodes) + for _, n := range nodes { + in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: slices.Clone(assigned[n])}) + for _, mod := range assigned[n] { + for _, cl := range catalog[mod].Claims { + scope := cl.Scope + if seat, ok := catalogue.SeatNamed(cl.Name); ok && scope == "" { + scope = seat.Scope + } + in.Holdings = append(in.Holdings, catalogue.Held{Claim: cl.Name, Scope: scope, Node: n, Module: mod}) + } + } + } + return in +} + +// Compose is the composition check of change c over the mesh m, composed with the mesh-sdk version sdk: every touched +// machine resolved as the trunk holds it and as the change leaves it. It fails on a manifest of the change that cannot +// be read, a module a machine runs that the change removes, a module defined twice, and a machine that resolves on the +// trunk and not with the change; a machine that does not resolve on either is said and fails nothing. It warns past +// CompositionRebuildWarning modules rebuilt, says what it does not judge (NotJudged), and for a group with a mesh-sdk +// member says it composed with its own mesh-sdk, not the member's head (issue 1621). took is how long it took. +func Compose(m Mesh, c Change, sdk string, took time.Duration) delivery.CompositionVerdict { + v := delivery.CompositionVerdict{Group: c.Group, Heads: c.Heads, SDK: sdk, Took: took.Milliseconds()} + var fails []string + failAt := func(machine, words string) { + if v.Machine == "" && machine != "" { + v.Machine = machine + } + fails = append(fails, words) + } + for _, name := range sortedKeys(c.Unreadable) { + failAt("", fmt.Sprintf("the manifest of %s cannot be read: %s", name, c.Unreadable[name])) + } + for _, name := range sortedKeys(c.Twice) { + failAt("", fmt.Sprintf("%s is defined twice, in %s", name, strings.Join(c.Twice[name], " and "))) + } + machines := Touched(m, c) + v.Machines = machines + for _, node := range machines { + for _, name := range c.Removed { + if slices.Contains(m.Assigned[node], name) { + failAt(node, fmt.Sprintf("%s runs %s, which the change removes from its source", node, name)) + } + } + } + trunk := input(m.Manifests, m.Assigned) + changed := map[string]catalogue.Manifest{} + for name, man := range m.Manifests { + changed[name] = man + } + for name, man := range c.Manifests { + changed[name] = man + } + for _, name := range c.Removed { + delete(changed, name) + } + after := input(changed, m.Assigned) + for _, node := range machines { + _, before := compose.Resolve(trunk, node) + _, now := compose.Resolve(after, node) + switch { + case now == nil: + case before != nil: + v.Warnings = append(v.Warnings, fmt.Sprintf("%s does not compose on the trunk either, so the change is not blamed: %v", node, before)) + default: + failAt(node, fmt.Sprintf("%s composes on the trunk and not with the change: %v", node, now)) + } + } + if c.Rebuilt > delivery.CompositionRebuildWarning { + v.Warnings = append(v.Warnings, fmt.Sprintf("the merge rebuilds %d modules, more than %d", c.Rebuilt, delivery.CompositionRebuildWarning)) + } + if c.SDKHead != "" { + v.Warnings = append(v.Warnings, fmt.Sprintf("composed with the assigner's own mesh-sdk %s, not the group's mesh-sdk head %s, "+ + "and a field the running controller cannot read is not judged (issues 1621, 1622, 1623)", sdk, c.SDKHead)) + } + v.Warnings = append(v.Warnings, NotJudged) + if len(machines) == 0 { + v.Warnings = append(v.Warnings, "the change touches no machine: no module it moves is assigned anywhere") + } + if len(fails) > 0 { + v.State, v.Words = delivery.CompositionFail, strings.Join(fails, "; ") + return v + } + v.State = delivery.CompositionPass + return v +} + +func sortedKeys[V any](m map[string]V) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} diff --git a/modules/mesh-delivery-assigner/internal/assign/compose_test.go b/modules/mesh-delivery-assigner/internal/assign/compose_test.go new file mode 100644 index 0000000..889988a --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/compose_test.go @@ -0,0 +1,116 @@ +package assign + +import ( + "slices" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func meshOf() Mesh { + return Mesh{ + Manifests: map[string]catalogue.Manifest{ + "mail": {Module: "mail", Version: "1", Requires: []string{"database"}}, + "store": {Module: "store", Version: "1", Provides: []catalogue.Offer{{Name: "database"}}}, + "web": {Module: "web", Version: "1"}, + }, + Assigned: map[string][]string{"ace": {"mail", "store"}, "g14": {"web"}}, + } +} + +func hasWarning(v delivery.CompositionVerdict, part string) bool { + return slices.ContainsFunc(v.Warnings, func(w string) bool { return strings.Contains(w, part) }) +} + +// 1297/AC1 (its unit half): a change touching one module on one machine composes that machine alone, passes, names +// the machine and the mesh-sdk it composed with, and says what it did not judge. +func TestAChangeOfOneModuleComposesItsMachine(t *testing.T) { + c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"web": {Module: "web", Version: "2"}}} + v := Compose(meshOf(), c, "v0.1.14-x", 0) + if v.State != delivery.CompositionPass || !slices.Equal(v.Machines, []string{"g14"}) || v.SDK != "v0.1.14-x" { + t.Fatalf("composition = %+v, want a pass on g14 alone", v) + } + if !hasWarning(v, "issue 1461") { + t.Errorf("the verdict does not say what it did not judge: %v", v.Warnings) + } + if _, err := delivery.EncodeCompositionVerdict(v); err != nil { + t.Errorf("the verdict cannot be written: %v", err) + } +} + +// 1297/AC2: a change that removes a module a machine runs fails, naming the machine and the module. +func TestRemovingAModuleAMachineRunsFails(t *testing.T) { + c := Change{Group: "g", Heads: "0123456789ab", Removed: []string{"web"}} + v := Compose(meshOf(), c, "sdk", 0) + if v.State != delivery.CompositionFail || v.Machine != "g14" || !strings.Contains(v.Words, "g14 runs web") { + t.Fatalf("composition = %+v, want a fail naming g14 and web", v) + } +} + +// 1297/B3c: a machine that composes on the trunk and not with the change fails it, naming the machine and why; the +// consumer's machine is touched when its provider changes. +func TestAMachineTheChangeBreaksFails(t *testing.T) { + c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"store": {Module: "store", Version: "2"}}} + v := Compose(meshOf(), c, "sdk", 0) + if v.State != delivery.CompositionFail || v.Machine != "ace" || !strings.Contains(v.Words, "database") { + t.Fatalf("composition = %+v, want a fail on ace naming the provision nothing provides", v) + } +} + +// 1297/B3c: what was already broken on the trunk is said and fails nothing. +func TestWhatWasBrokenBeforeFailsNothing(t *testing.T) { + m := meshOf() + m.Manifests["store"] = catalogue.Manifest{Module: "store", Version: "1"} + c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"mail": {Module: "mail", Version: "2", Requires: []string{"database"}}}} + v := Compose(m, c, "sdk", 0) + if v.State != delivery.CompositionPass || !hasWarning(v, "on the trunk either") { + t.Fatalf("composition = %+v, want a pass saying ace was already broken", v) + } +} + +// 1297/B3c: an unreadable manifest and a module defined twice fail; more than twelve rebuilt modules warn. +func TestUnreadableTwiceAndWide(t *testing.T) { + c := Change{Group: "g", Heads: "0123456789ab", Unreadable: map[string]string{"web": "unknown field \"scope\""}, Rebuilt: 13} + v := Compose(meshOf(), c, "sdk", 0) + if v.State != delivery.CompositionFail || !strings.Contains(v.Words, "web cannot be read") || !hasWarning(v, "rebuilds 13 modules") { + t.Fatalf("composition = %+v", v) + } + c = Change{Group: "g", Heads: "0123456789ab", Twice: map[string][]string{"web": {"novox/a", "novox/b"}}, Rebuilt: 12} + v = Compose(meshOf(), c, "sdk", 0) + if v.State != delivery.CompositionFail || !strings.Contains(v.Words, "web is defined twice") || hasWarning(v, "rebuilds") { + t.Fatalf("composition = %+v", v) + } +} + +// 1622's criterion as R4g delivers it: a group with a mesh-sdk member is composed with the assigner's own mesh-sdk, and +// the verdict says so, naming the member's head it did not use; a group without one says nothing of it. +func TestAGroupWithAnSDKMemberSaysWhichSDKComposedIt(t *testing.T) { + c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"web": {Module: "web", Version: "2"}}, SDKHead: "f625d98713e4"} + v := Compose(meshOf(), c, "v0.1.14-own", 0) + if !hasWarning(v, "not the group's mesh-sdk head f625d98713e4") || !hasWarning(v, "v0.1.14-own") || !hasWarning(v, "1623") { + t.Errorf("the verdict does not say it composed with its own mesh-sdk: %v", v.Warnings) + } + c.SDKHead = "" + if v := Compose(meshOf(), c, "v0.1.14-own", 0); hasWarning(v, "mesh-sdk head") { + t.Errorf("a group without a mesh-sdk member says it has one: %v", v.Warnings) + } +} + +// 1297/B3a: the machines touched are those running a moved module or a consumer of what it provides; an added module +// assigned nowhere touches none, and the verdict says so. +func TestTouched(t *testing.T) { + m := meshOf() + if got := Touched(m, Change{Manifests: map[string]catalogue.Manifest{"store": m.Manifests["store"]}}); !slices.Equal(got, []string{"ace"}) { + t.Errorf("touched by store = %v", got) + } + m.Assigned["g14"] = append(m.Assigned["g14"], "mail") + if got := Touched(m, Change{Manifests: map[string]catalogue.Manifest{"store": m.Manifests["store"]}}); !slices.Equal(got, []string{"ace", "g14"}) { + t.Errorf("touched by store with a consumer on g14 = %v", got) + } + c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"new": {Module: "new", Version: "1"}}} + if v := Compose(m, c, "sdk", 0); len(v.Machines) != 0 || !hasWarning(v, "touches no machine") || v.State != delivery.CompositionPass { + t.Errorf("an added module assigned nowhere: %+v", v) + } +}