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 new file mode 100644 index 0000000..2682b37 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -0,0 +1,276 @@ +//go:build integration + +package main + +import ( + "context" + "encoding/json" + "io" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "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 +// are kept in, as the controller's broker raises them. +type bed struct { + srv *server.Server + js jetstream.JetStream + ctx context.Context + dir string + buckets map[string]jetstream.KeyValue +} + +func startBed(t *testing.T, leaveOut string) *bed { + t.Helper() + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "assigner") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: filepath.Join(dir, "js"), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + nc, err := nats.Connect(srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + t.Cleanup(cancel) + b := &bed{srv: srv, js: js, ctx: ctx, dir: dir, buckets: map[string]jetstream.KeyValue{}} + for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) { + if name == leaveOut { + continue + } + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5}) + if err != nil { + t.Fatal(err) + } + b.buckets[name] = kv + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil { + t.Fatal(err) + } + return b +} + +// start runs the program as the node does, through run with its environment and a broker file; it answers the +// channel run's error arrives on. +func (b *bed) start(t *testing.T) <-chan error { + t.Helper() + broker := filepath.Join(b.dir, "broker") + cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-assigner", User: "n1.mesh-delivery-assigner", Password: "pw"}) + if err := os.WriteFile(broker, cred, 0o600); err != nil { + t.Fatal(err) + } + env := map[string]string{"MESH_BROKER_FILE": broker} + ctx, stop := context.WithCancel(b.ctx) + done := make(chan error, 1) + go func() { done <- run(ctx, func(k string) string { return env[k] }, io.Discard) }() + t.Cleanup(func() { + stop() + select { + case <-done: + case <-time.After(10 * time.Second): + } + }) + return done +} + +// 1297/B2b: started whole on its credential, the assigner says on the bus the three states it keeps, each readable, +// having opened every bucket it names. +func TestTheAssignerStartsAndSaysWhatItKeeps(t *testing.T) { + b := startBed(t, "") + done := b.start(t) + events, err := b.js.Stream(b.ctx, "EVENTS") + if err != nil { + t.Fatal(err) + } + var said announce.States + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + select { + case err := <-done: + t.Fatalf("the assigner stopped: %v", err) + default: + } + m, err := events.GetLastMsgForSubject(b.ctx, announce.Subject("mesh-delivery-assigner")) + if err != nil { + continue + } + if err := json.Unmarshal(m.Data, &said); err != nil { + t.Fatal(err) + } + break + } + if said.Module != "mesh-delivery-assigner" || said.Node != "n1" || len(said.States) != 3 || + !said.Readable("wants") || !said.Readable("differences") || !said.Readable("compositions") { + t.Fatalf("the assigner said %+v, want its three states, each readable", said) + } +} + +// 1297/B2b: a bucket it reads that the bus does not hold stops the start, naming the bucket. +func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) { + for _, missing := range readBuckets { + b := startBed(t, missing) + select { + case err := <-b.start(t): + if err == nil || !strings.Contains(err.Error(), missing) { + t.Errorf("without %s the assigner stopped with %v, want the bucket named", missing, err) + } + case <-time.After(15 * time.Second): + t.Errorf("without %s the assigner started", missing) + } + } +} + +// 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) + } +} + +// get waits for a key of one of the bed's buckets and answers its value. +func (b *bed) get(t *testing.T, bucket, key string) []byte { + t.Helper() + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err := b.buckets[bucket].Get(b.ctx, key); err == nil { + return e.Value() + } + } + t.Fatalf("no %s %s after 10 s", bucket, key) + return nil +} + +// 1297/B4, B5 on the bus: the controller sent F to g14 while the canary ace does not run F healthy yet. Started whole, +// the assigner wants F on the canary, writes the difference on g14 explained by rule 3, and on ace one unexplained (the +// controller sends ace the older fingerprint while ace wants F). +func TestWantsAndTheirDifferencesOnTheBus(t *testing.T) { + b := startBed(t, "") + fNew, fOld := delivery.SourcePrefix2+strings.Repeat("b", 64), delivery.SourcePrefix2+strings.Repeat("a", 64) + digest := b.manifest(t, "web", `{"module":"web","version":"1"}`) + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fNew, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath}) + b.putJSON(t, delivery.ModulesBucket, "web", mod) + built, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest}) + if err != nil { + t.Fatal(err) + } + bkey, _ := delivery.BuilderBuildKey("web", fNew) + b.putJSON(t, delivery.BuilderBuildsBucket, bkey, built) + gkey, _ := delivery.ControllerGrantKey("web", fNew) + grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{}) + b.putJSON(t, delivery.ControllerGrantsBucket, gkey, grant) + for _, node := range []string{"ace", "g14"} { + on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) + b.putJSON(t, delivery.PresenceBucket, node, on) + have, _ := delivery.EncodeHave(delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + b.putJSON(t, delivery.HaveBucket, node+".web", have) + } + b.putJSON(t, provider.AssignmentsBucket, "ace.web", []byte(`{"node":"ace","module":"web","build":"b-8"}`)) + b.putJSON(t, provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"b-9"}`)) + b.putJSON(t, delivery.BuildsBucket, delivery.BuildKey("web", "b-8"), []byte(`{"id":"b-8","repository":"novox/web","source-fingerprint":"`+fOld+`"}`)) + b.putJSON(t, delivery.BuildsBucket, delivery.BuildKey("web", "b-9"), []byte(`{"id":"b-9","repository":"novox/web","source-fingerprint":"`+fNew+`"}`)) + b.start(t) + + wkey, _ := delivery.WantKey("ace", "web") + want, err := delivery.DecodeWant(b.get(t, delivery.AssignerWantsBucket, wkey)) + if err != nil || want.FP != fNew { + t.Fatalf("the canary ace's want = %+v, %v", want, err) + } + dkey, _ := delivery.DifferenceKey("g14", "web") + d, err := delivery.DecodeDifference(b.get(t, delivery.AssignerDifferencesBucket, dkey)) + if err != nil || d.Rule != delivery.RuleCanary || d.Sent != fNew { + t.Fatalf("g14's difference = %+v, %v; want rule 3", d, err) + } + akey, _ := delivery.DifferenceKey("ace", "web") + a, err := delivery.DecodeDifference(b.get(t, delivery.AssignerDifferencesBucket, akey)) + if err != nil || !a.Unexplained || a.Want != fNew || a.Sent != fOld { + t.Fatalf("ace's difference = %+v, %v; want it unexplained", a, err) + } + if e, err := b.buckets[delivery.AssignerWantsBucket].Get(b.ctx, "want.g14.web"); err == nil { + t.Errorf("g14, held back by the canary with no want before, was given one: %s", e.Value()) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go new file mode 100644 index 0000000..dc34b09 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -0,0 +1,222 @@ +// Command mesh-delivery-assigner is delivery's assigner (novox/hq ADR 0429 §1 and §6, ADR 0446, ADR 0528, issue 1297): a +// process on its own bus credential that watches what the indexer, the builder, the controller, the node-engines and +// the watchman write, and writes what each node should run (`want..`, by the six rules), the +// fingerprints it holds, every difference between a want and what the controller sends, and the composition check's +// verdict of each group the checker judges. In release 5 the wants are only compared: the controller still sends what +// each node runs (ADR 0429 §7). It holds no timer: every pass is started by a change it watched. +package main + +import ( + "context" + "errors" + "fmt" + "io" + "log" + "os" + "os/signal" + "strings" + "syscall" + + "github.com/nats-io/nats.go" + "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" +) + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + if err := run(ctx, os.Getenv, os.Stderr); err != nil { + fmt.Fprintln(os.Stderr, "mesh-delivery-assigner:", err) + os.Exit(1) + } +} + +// readBuckets are the buckets of other parts the assigner reads, each named in its manifest's reads (the manifest test +// holds the two lists equal), so the controller's broker grants exactly these. +var readBuckets = []string{ + provider.AssignmentsBucket, + delivery.BuildsBucket, + delivery.ControllerGrantsBucket, + delivery.BuilderBuildsBucket, + delivery.CandidatesBucket, + delivery.ManifestsBucket, + delivery.ModulesBucket, + delivery.PresenceBucket, + delivery.HaveBucket, +} + +// ownBuckets are the assigner's own states, as its manifest declares them. +var ownBuckets = []string{delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerWantsBucket} + +// keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view +// (and the node agent from release 6), its differences by the view, its compositions by the checker. +var keptStates = []announce.State{ + {Name: "compositions", Readable: true}, + {Name: "differences", Readable: true}, + {Name: "wants", Readable: true}, +} + +// buckets is every bucket the assigner opened, by name. +type buckets map[string]jetstream.KeyValue + +// open opens every bucket the assigner reads and writes; a bucket it cannot open stops the start, named, since a pass +// without it would write wants from half the mesh. +func open(ctx context.Context, js jetstream.JetStream) (buckets, error) { + out := buckets{} + for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) { + kv, err := js.KeyValue(ctx, name) + if err != nil { + return nil, fmt.Errorf("the bucket %s cannot be opened: %w", name, err) + } + out[name] = kv + } + return out, nil +} + +// run is the program: it returns when ctx ends or the bus cannot be used. +func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { + logger := log.New(logOut, "mesh-delivery-assigner: ", log.LstdFlags) + path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) + if path == "" { + return errors.New("MESH_BROKER_FILE is not set, so there is no credential to read the bus with") + } + account, err := process.ReadAccount(path) + if err != nil { + return err + } + opts := []nats.Option{ + nats.Name(account.Node + "." + account.Module), + nats.UserInfo(account.User, account.Password), + nats.CustomInboxPrefix(account.InboxPrefix()), + nats.MaxReconnects(-1), + nats.RetryOnFailedConnect(true), + } + if strings.TrimSpace(account.Fingerprint) != "" { + opts = append(opts, nats.Secure(process.PinnedToFingerprint(account.Fingerprint))) + } + conn, err := nats.Connect(account.URL, opts...) + if err != nil { + return fmt.Errorf("connecting to the bus: %w", err) + } + defer conn.Close() + js, err := jetstream.New(conn) + if err != nil { + return err + } + kvs, err := open(ctx, js) + if err != nil { + return err + } + // What the assigner keeps, said on the bus (ADR 0495 §2-§3): a reader is granted from it. Not fatal when it cannot + // be said now: process.Conn says it again at every reconnect. + said, err := process.Connect(account, nats.Name(account.Node+"."+account.Module+".keeps"), nats.RetryOnFailedConnect(true)) + if err != nil { + return err + } + defer said.Close() + if err := said.Keeps(keptStates...); err != nil { + 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)) + 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 { + var opts []jetstream.WatchOpt + if name == delivery.HaveBucket { + opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back + } + if err := watch(name, opts...); 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/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go new file mode 100644 index 0000000..f52e1fb --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -0,0 +1,59 @@ +package main + +import ( + "context" + "io" + "os" + "path/filepath" + "reflect" + "slices" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// 1297/B2a: the buckets the program opens are exactly the manifest's reads and states, as the broker names them +// (`.` read as the bucket `_`), so it is granted what it reads and reads nothing else. +func TestOpensWhatItsManifestNames(t *testing.T) { + raw, err := os.ReadFile(filepath.Join("..", "..", "module.json")) + if err != nil { + t.Fatal(err) + } + m, err := catalogue.ParseManifest(raw) + if err != nil { + t.Fatal(err) + } + var reads []string + for _, r := range m.Reads { + reads = append(reads, strings.Replace(r, ".", "_", 1)) + } + var states []string + for _, s := range m.State { + states = append(states, m.Module+"_"+s.Name) + } + for _, pair := range []struct { + what string + code, want []string + }{{"reads", readBuckets, reads}, {"states", ownBuckets, states}} { + got, want := slices.Sorted(slices.Values(pair.code)), slices.Sorted(slices.Values(pair.want)) + if !reflect.DeepEqual(got, want) { + t.Errorf("the program's %s are %v, the manifest's %v", pair.what, got, want) + } + } + var kept []string + for _, s := range keptStates { + kept = append(kept, m.Module+"_"+s.Name) + } + if got, want := slices.Sorted(slices.Values(kept)), slices.Sorted(slices.Values(ownBuckets)); !reflect.DeepEqual(got, want) { + t.Errorf("the program announces %v, keeps %v", got, want) + } +} + +// 1297/B2a: without a credential the program does not start, and says why. +func TestNoCredential(t *testing.T) { + err := run(context.Background(), func(string) string { return "" }, io.Discard) + if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") { + t.Errorf("run without a credential = %v", err) + } +} 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..754a812 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -0,0 +1,207 @@ +package main + +import ( + "context" + "errors" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "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) + rules := p.wants(ctx) + p.differences(ctx, rules) +} + +// differences writes, for each node and module the assigner wants or the controller sends, the difference between the +// two with the rule that explains it or unexplained (ADR 0528 clause 4), and deletes the key of a difference that +// ended; its history keeps what it was. +func (p *passer) differences(ctx context.Context, rules map[string]map[string]assign.Holding) { + _, assignments := p.w.assigned() + wanted := p.w.wantsOf() + trunk := p.w.trunk() + pairs := map[[2]string]bool{} + for _, a := range assignments { + pairs[[2]string{a.Node, a.Module}] = true + } + for module, byNode := range wanted { + for node := range byNode { + pairs[[2]string{node, module}] = true + } + } + written := p.w.read(delivery.AssignerDifferencesBucket) + keep := map[string]bool{} + for pair := range pairs { + node, module := pair[0], pair[1] + key, err := delivery.DifferenceKey(node, module) + if err != nil { + continue + } + var want *delivery.Want + if w, ok := wanted[module][node]; ok { + want = &w + } + var sent assign.Sent + if a, ok := assignments[node+"."+module]; ok { + sent = assign.Sent{Sends: true, Build: a.Build, FP: p.w.sentFP(ctx, kvGetter(p.kvs[delivery.BuildsBucket]), module, a.Build)} + } + d, ok := assign.Compare(want, sent, trunk[module].FP2, rules[module][node]) + if !ok { + continue + } + keep[key] = true + if before, err := delivery.DecodeDifference(written[key]); err == nil { + before.V = 0 + if delivery.SameDifference(before, d) { + continue + } + } + p.put(ctx, delivery.AssignerDifferencesBucket, key, func() ([]byte, error) { return delivery.EncodeDifference(d) }) + } + for key := range written { + if keep[key] { + continue + } + if err := p.kvs[delivery.AssignerDifferencesBucket].Delete(ctx, key); err != nil { + p.logf("the difference %s that ended could not be deleted: %v", key, err) + continue + } + p.w.put(delivery.AssignerDifferencesBucket, key, nil) + } +} + +// wants writes, for every module the trunk holds a fingerprint of, each node's want by the six rules (ADR 0429 §6), and +// the hold of a fingerprint whose health check failed. It returns, by module then node, the rule that holds a node +// back, for the comparison. +func (p *passer) wants(ctx context.Context) map[string]map[string]assign.Holding { + get := kvGetter(p.kvs[delivery.ManifestsBucket]) + assigned, _ := p.w.assigned() + trunk := p.w.trunk() + manifests := map[string]catalogue.Manifest{} + for name, mod := range trunk { + if mod.Manifest == "" { + continue + } + if got := p.w.manifest(ctx, get, mod.Manifest); got.err == "" { + manifests[name] = got.manifest + } + } + nodes := p.w.nodes(manifests, assigned) + nodesOf := map[string][]string{} + for node, modules := range assigned { + for _, m := range modules { + nodesOf[m] = append(nodesOf[m], node) + } + } + wanted := p.w.wantsOf() + rules := map[string]map[string]assign.Holding{} + names := map[string]bool{} + for name := range trunk { + names[name] = true + } + for name := range wanted { + names[name] = true + } + for name := range names { + fp := trunk[name].FP2 + m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name]} + if fp != "" { + held, _ := delivery.HeldKey(name, fp) + grant, _ := delivery.ControllerGrantKey(name, fp) + m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(delivery.AssignerWantsBucket, held), p.w.has(delivery.ControllerGrantsBucket, grant) + } + last := map[string]string{} + for _, node := range m.Nodes { + last[node] = p.w.lastHealthy(node, name, fp) + } + out := assign.Wants(m, nodes, wanted[name], last) + rules[name] = out.Rules + if out.Held != nil { + key, err := delivery.HeldKey(name, fp) + if err == nil { + held := *out.Held + p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeHeld(held) }) + } + } + for node, want := range out.Wants { + key, err := delivery.WantKey(node, name) + if err != nil { + p.logf("no want of %s on %s is written: %v", name, node, err) + continue + } + want := want + p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeWant(want) }) + } + } + return rules +} + +// 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..fcd91d8 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -0,0 +1,370 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "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-sdk delivery.CandidateNoneNoModule, which the indexer writes): the change removes the module. +const noModuleWords = delivery.CandidateNoneNoModule + +// 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 + // healthy is, by have key, every fingerprint the node ran healthy, oldest first, each once: what a set-back goes to. + healthy map[string][]string + // sent is, by build id, the fingerprint of the new form the controller's build record names, once it names one. + sent map[string]string +} + +// 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{}, healthy: map[string][]string{}, sent: map[string]string{}} +} + +// 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 + if bucket == delivery.HaveBucket { + if h, err := delivery.DecodeHave(value); err == nil && h.FP != "" && h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK { + list := slices.DeleteFunc(w.healthy[key], func(fp string) bool { return fp == h.FP }) + w.healthy[key] = append(list, h.FP) + } + } +} + +// lastHealthy is the newest fingerprint other than fp the node ran the module healthy at, or empty. +func (w *world) lastHealthy(node, module, fp string) string { + key, err := delivery.HaveKey(node, module) + if err != nil { + return "" + } + w.mu.Lock() + defer w.mu.Unlock() + list := w.healthy[key] + for i := len(list) - 1; i >= 0; i-- { + if list[i] != fp { + return list[i] + } + } + return "" +} + +// 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, "" +} + +// wantsOf is the wants the assigner wrote, by module then node. +func (w *world) wantsOf() map[string]map[string]delivery.Want { + out := map[string]map[string]delivery.Want{} + for key, value := range w.read(delivery.AssignerWantsBucket) { + node, module, err := delivery.ParseWantKey(key) + if err != nil { + continue + } + want, err := delivery.DecodeWant(value) + if err != nil { + continue + } + if out[module] == nil { + out[module] = map[string]delivery.Want{} + } + out[module][node] = want + } + return out +} + +// nodes is what the six rules read of the nodes: presence, haves, and every provision's providers with their nodes. +func (w *world) nodes(manifests map[string]catalogue.Manifest, assigned map[string][]string) assign.Nodes { + n := assign.Nodes{Online: map[string]bool{}, Haves: map[string]map[string]delivery.Have{}, Providers: map[string]map[string][]string{}} + for key, value := range w.read(delivery.PresenceBucket) { + node, _, _, err := delivery.ParsePresenceKey(key) + if err != nil || node == "" { + continue + } + if p, err := delivery.DecodePresence(value); err == nil { + n.Online[node] = p.Online + } + } + for key, value := range w.read(delivery.HaveBucket) { + node, module, isNode, err := delivery.ParseHaveKey(key) + if err != nil || isNode { + continue + } + if h, err := delivery.DecodeHave(value); err == nil { + if n.Haves[node] == nil { + n.Haves[node] = map[string]delivery.Have{} + } + n.Haves[node][module] = h + } + } + nodesOf := map[string][]string{} + for node, modules := range assigned { + for _, m := range modules { + nodesOf[m] = append(nodesOf[m], node) + } + } + for name, man := range manifests { + for _, o := range man.Provides { + if n.Providers[o.Name] == nil { + n.Providers[o.Name] = map[string][]string{} + } + n.Providers[o.Name][name] = nodesOf[name] + } + } + return n +} + +// has says whether a bucket holds a key now. +func (w *world) has(bucket, key string) bool { + w.mu.Lock() + defer w.mu.Unlock() + _, ok := w.values[bucket][key] + return ok +} + +// built says whether the builder's build of a module at a fingerprint is done. +func (w *world) built(module, fp string) bool { + key, err := delivery.BuilderBuildKey(module, fp) + if err != nil { + return false + } + w.mu.Lock() + value, ok := w.values[delivery.BuilderBuildsBucket][key] + w.mu.Unlock() + if !ok { + return false + } + b, err := delivery.DecodeBuilderBuild(value) + return err == nil && b.State == delivery.BuildDone +} + +// sentFP is the fingerprint of the new form of the build the controller's assignment names, read from its build record +// (mesh-controller_builds `.build.`) once and kept; empty when the record names none. +func (w *world) sentFP(ctx context.Context, get getter, module, build string) string { + if build == "" { + return "" + } + w.mu.Lock() + fp, ok := w.sent[build] + w.mu.Unlock() + if ok { + return fp + } + value, err := get(ctx, delivery.BuildKey(module, build)) + if err != nil { + return "" + } + var rec delivery.BuildRecord + if json.Unmarshal(value, &rec) != nil || !strings.HasPrefix(rec.SourceFingerprint, delivery.SourcePrefix2) { + return "" + } + w.mu.Lock() + w.sent[build] = rec.SourceFingerprint + w.mu.Unlock() + return rec.SourceFingerprint +} 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/go.mod b/modules/mesh-delivery-assigner/go.mod new file mode 100644 index 0000000..dbf6452 --- /dev/null +++ b/modules/mesh-delivery-assigner/go.mod @@ -0,0 +1,9 @@ +module mesh-delivery-assigner + +go 1.26.0 + +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023134155-3e5d4c30ffa7 + github.com/nats-io/nats-server/v2 v2.11.17 + github.com/nats-io/nats.go v1.54.0 +) diff --git a/modules/mesh-delivery-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum new file mode 100644 index 0000000..5c1a857 --- /dev/null +++ b/modules/mesh-delivery-assigner/go.sum @@ -0,0 +1,27 @@ +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4 h1:yYDxFRJoH59kiCoEulD+FBwctyf9CrtKIxlEAsVL39Q= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E= +github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= +github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= +github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ= +github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= +github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= +github.com/nats-io/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc= +github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc= +github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= +github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= diff --git a/modules/mesh-delivery-assigner/internal/assign/compare.go b/modules/mesh-delivery-assigner/internal/assign/compare.go new file mode 100644 index 0000000..952acd3 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/compare.go @@ -0,0 +1,52 @@ +package assign + +import ( + "fmt" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Sent is what the controller sends a node of a module: whether it sends it at all (an open assignment), the build its +// assignment names, and that build's fingerprint of the new form (empty when its record names none). +type Sent struct { + Sends bool + Build string + FP string +} + +// Compare is the difference between a node's want for a module and what the controller sends it (ADR 0528 clause 4), +// and whether there is one. want is nil when the assigner wrote none; newest is the module's newest fingerprint and +// hold the rule that holds the node back from it (Rule 0 when none does). A difference is explained by a rule when the +// controller sent the newest fingerprint while that rule holds the node's want back; every other is unexplained, said +// in words, for acceptance test 17 of design 47 to read. +func Compare(want *delivery.Want, sent Sent, newest string, hold Holding) (delivery.Difference, bool) { + var d delivery.Difference + removed := want != nil && want.Removed + if want != nil && !removed { + d.Want = want.FP + } + d.WantRemoved = removed + if !sent.Sends { + if want == nil || removed { + return delivery.Difference{}, false + } + d.Unexplained, d.Words = true, "the node wants "+short(want.FP)+" and the controller sends it nothing" + return d, true + } + d.Sent, d.SentBuild = sent.FP, sent.Build + switch { + case sent.FP == "": + d.Unexplained, d.Words = true, fmt.Sprintf("the build %s the controller sends names no fingerprint of the new form", sent.Build) + case removed: + d.Unexplained, d.Words = true, fmt.Sprintf("the want is removed and the controller still sends %s", short(sent.FP)) + case sent.FP == d.Want: + return delivery.Difference{}, false + case hold.Rule != 0 && sent.FP == newest: + d.Rule, d.Words = hold.Rule, fmt.Sprintf("the controller sent %s while %s", short(sent.FP), hold.Words) + case d.Want == "": + d.Unexplained, d.Words = true, fmt.Sprintf("the controller sends %s and the node has no want", short(sent.FP)) + default: + d.Unexplained, d.Words = true, fmt.Sprintf("the controller sends %s and the node wants %s", short(sent.FP), short(d.Want)) + } + return d, true +} diff --git a/modules/mesh-delivery-assigner/internal/assign/compare_test.go b/modules/mesh-delivery-assigner/internal/assign/compare_test.go new file mode 100644 index 0000000..da4a14f --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/compare_test.go @@ -0,0 +1,59 @@ +package assign + +import ( + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 1297/B5: given the controller sent F to a node while the canary is not yet healthy on F, the difference names rule 3. +func TestADifferenceExplainedByTheCanary(t *testing.T) { + m, n := ready() + n.Haves["ace"]["mail"] = delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK} + out := Wants(m, n, wantedOld(), nil) + want := delivery.Want{FP: fOld, Newest: fNew, Rule: delivery.RuleCanary} + d, ok := Compare(&want, Sent{Sends: true, Build: "b-9", FP: fNew}, fNew, out.Rules["g14"]) + if !ok || d.Rule != delivery.RuleCanary || d.Unexplained || d.Want != fOld || d.Sent != fNew { + t.Fatalf("difference = %+v, %v; want rule 3", d, ok) + } + if _, err := delivery.EncodeDifference(d); err != nil { + t.Errorf("the difference cannot be written: %v", err) + } +} + +// 1297/B5: a want equal to what is sent is no difference; nothing sent and no want is none; every other is written +// unexplained, with words, and is writable. +func TestCompare(t *testing.T) { + w := delivery.Want{FP: fNew} + if _, ok := Compare(&w, Sent{Sends: true, Build: "b", FP: fNew}, fNew, Holding{}); ok { + t.Errorf("an equal want and send is a difference") + } + if _, ok := Compare(nil, Sent{}, fNew, Holding{}); ok { + t.Errorf("no want and nothing sent is a difference") + } + removed := delivery.Want{Removed: true} + if _, ok := Compare(&removed, Sent{}, fNew, Holding{}); ok { + t.Errorf("a removed want and nothing sent is a difference") + } + old := delivery.Want{FP: fOld} + for name, c := range map[string]struct { + want *delivery.Want + sent Sent + }{ + "sends an older one": {&w, Sent{Sends: true, Build: "b", FP: fOld}}, + "names no fingerprint": {&w, Sent{Sends: true, Build: "b"}}, + "sends a removed module": {&removed, Sent{Sends: true, Build: "b", FP: fNew}}, + "sends nothing": {&w, Sent{}}, + "no want": {nil, Sent{Sends: true, Build: "b", FP: fNew}}, + "no rule holds it back": {&old, Sent{Sends: true, Build: "b", FP: fNew}}, + } { + d, ok := Compare(c.want, c.sent, fNew, Holding{}) + if !ok || !d.Unexplained || d.Words == "" { + t.Errorf("%s: difference = %+v, %v; want it unexplained, with words", name, d, ok) + continue + } + if _, err := delivery.EncodeDifference(d); err != nil { + t.Errorf("%s: the difference cannot be written: %v", name, err) + } + } +} 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) + } +} diff --git a/modules/mesh-delivery-assigner/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go new file mode 100644 index 0000000..524f823 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -0,0 +1,205 @@ +package assign + +import ( + "fmt" + "slices" + "sort" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Module is what the six rules read of one module (ADR 0429 §6, ADR 0528 clause 3): its newest fingerprint F at the +// trunk (the indexer's modules), whether the builder's build of F is done, whether F is held, whether the controller +// issued its credential for F, its manifest at F, and the nodes it is assigned to. +type Module struct { + Name string + FP string + Built bool + Held bool + Granted bool + Manifest catalogue.Manifest + Nodes []string +} + +// Node is what the rules read of the nodes: whether each is online (the watchman's presence) and each module's have. +type Nodes struct { + Online map[string]bool + Haves map[string]map[string]delivery.Have + // Providers is, by provision, the modules that provide it, with the nodes each is assigned to. + Providers map[string]map[string][]string +} + +// healthy says whether a have runs fp healthy. +func healthy(h delivery.Have, fp string) bool { + return h.FP == fp && h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK +} + +// Canary is the module's canary: the first of its nodes, by name, that is online (ADR 0435 §3; one ring, ordered by +// name, ADR 0528 clause 3). Empty when none is online. +func Canary(nodes []string, online map[string]bool) string { + sorted := slices.Sorted(slices.Values(nodes)) + for _, n := range sorted { + if online[n] { + return n + } + } + return "" +} + +// Holds is whether every rule holds for moving node's want for m to m.FP, and if not, the first rule that does not, +// with why (ADR 0429 §6). Rule 4, at most `step` others ahead, always holds: the mesh keeps no step, and step 0 is no +// limit (ADR 0528 clause 3). +func Holds(m Module, node string, n Nodes) (bool, int, string) { + switch { + case !m.Built: + return false, delivery.RuleBuildDone, fmt.Sprintf("the build of %s at %s is not done", m.Name, short(m.FP)) + case m.Held: + return false, delivery.RuleNotHeld, fmt.Sprintf("%s is held at %s", m.Name, short(m.FP)) + } + canary := Canary(m.Nodes, n.Online) + switch { + case canary == "": + return false, delivery.RuleCanary, fmt.Sprintf("none of the nodes of %s is online, so it has no canary", m.Name) + case node != canary && !healthy(n.Haves[canary][m.Name], m.FP): + return false, delivery.RuleCanary, fmt.Sprintf("the canary %s does not run %s healthy yet", canary, short(m.FP)) + } + for _, provision := range m.Manifest.Requires { + if ok, why := providedHealthy(provision, node, n); !ok { + return false, delivery.RuleProviders, why + } + } + if !m.Granted { + return false, delivery.RuleGrant, fmt.Sprintf("the controller has not issued the credential of %s for %s", m.Name, short(m.FP)) + } + return true, 0, "" +} + +// providedHealthy is rule 5 for one provision: some module providing it runs healthy on the node, or, assigned +// elsewhere only, at its canary. A provision no module provides is not a module's to judge (the mesh or the node +// offers it) and holds. +func providedHealthy(provision, node string, n Nodes) (bool, string) { + providers := n.Providers[provision] + if len(providers) == 0 { + return true, "" + } + names := make([]string, 0, len(providers)) + for p := range providers { + names = append(names, p) + } + sort.Strings(names) + for _, p := range names { + nodes := providers[p] + at := node + if !slices.Contains(nodes, node) { + at = Canary(nodes, n.Online) + } + if h, ok := n.Haves[at][p]; ok && at != "" && h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK { + return true, "" + } + } + return false, fmt.Sprintf("no provider of %s runs healthy on %s or at its canary", provision, node) +} + +// keepsData says whether a module holds data a set-back would lose: an own item of a class that keeps (irreplaceable, +// valuable or rebuildable; cache and none do not). +func keepsData(m catalogue.Manifest) bool { + for _, it := range m.DataItems() { + if it.Class == catalogue.ClassIrreplaceable || it.Class == catalogue.ClassValuable || it.Class == catalogue.ClassRebuildable { + return true + } + } + return false +} + +// Outcome is what one module's pass decides: each node's want to write (only those that change), and a hold to write. +type Outcome struct { + Wants map[string]delivery.Want + // Held is the hold of m.FP to write, when a node's health check failed on it and it is not held yet. + Held *delivery.Held + // Rules is, by node, the rule holding the node back from m.FP and why, for the comparison. + Rules map[string]Holding +} + +// Holding is a rule holding a node back, and why. +type Holding struct { + Rule int + Words string +} + +// Wants is the pass of one module m: for each of its nodes, the want by the six rules, starting from the want written +// before (wants, by node); lastHealthy is, by node, the newest fingerprint other than m.FP the node ran healthy. A failed +// health check of m.FP on a node holds it: every node wanting m.FP is set back to its last healthy fingerprint, but a +// module holding data is not set back (in release 5 it is asked about from release 6 on: wants are only compared). A +// node that wants m but is no longer among its nodes gets its want written removed, never deleted (ADR 0408 §4). +func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[string]string) Outcome { + out := Outcome{Wants: map[string]delivery.Want{}, Rules: map[string]Holding{}} + held := m.Held + if !held && m.FP != "" { + for _, node := range slices.Sorted(slices.Values(m.Nodes)) { + if h, ok := n.Haves[node][m.Name]; ok && h.FP == m.FP && h.Health == delivery.HealthFailed { + out.Held = &delivery.Held{Node: node, Words: h.Detail} + held = true + break + } + } + } + mm := m + mm.Held = held + for _, node := range m.Nodes { + if m.FP == "" { + break + } + before, had := wants[node] + var next delivery.Want + switch ok, rule, words := Holds(mm, node, n); { + case ok: + next = delivery.Want{FP: m.FP} + case held && had && before.FP == m.FP: + out.Rules[node] = Holding{Rule: rule, Words: words} + last := lastHealthy[node] + switch { + case keepsData(m.Manifest): + next = before + next.Words = fmt.Sprintf("%s holds data, so it is not set back from %s: the operator is asked from release 6", m.Name, short(m.FP)) + case last == "": + next = before + next.Words = fmt.Sprintf("%s is held at %s and %s ran no other fingerprint healthy to set it back to", m.Name, short(m.FP), node) + default: + next = delivery.Want{FP: last, Newest: m.FP, Rule: rule, Words: words} + } + default: + out.Rules[node] = Holding{Rule: rule, Words: words} + if !had || before.Removed { + continue + } + next = delivery.Want{FP: before.FP, Newest: m.FP, Rule: rule, Words: words} + if before.FP == m.FP { + next = delivery.Want{FP: m.FP} + } + } + if !had || !delivery.SameWant(normal(before), normal(next)) { + out.Wants[node] = next + } + } + for node, before := range wants { + if !slices.Contains(m.Nodes, node) && !before.Removed { + out.Wants[node] = delivery.Want{Removed: true, Words: fmt.Sprintf("%s is no longer assigned to %s", m.Name, node)} + } + } + return out +} + +// normal is a want without its version, as a decoded and a composed one are compared. +func normal(w delivery.Want) delivery.Want { + w.V = 0 + return w +} + +func short(fp string) string { + part, err := delivery.FingerprintKeyPart(fp) + if err != nil || len(part) < 12 { + return fp + } + return part[:12] +} diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go new file mode 100644 index 0000000..c0ff7df --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -0,0 +1,139 @@ +package assign + +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func fpOf(c byte) string { return delivery.SourcePrefix2 + strings.Repeat(string(c), 64) } + +var ( + fNew = fpOf('b') + fOld = fpOf('a') +) + +// ready is a module mail at fNew on ace and g14 for which every rule holds: built, not held, granted, its provider +// store healthy on ace and the canary ace running fNew healthy. +func ready() (Module, Nodes) { + m := Module{Name: "mail", FP: fNew, Built: true, Granted: true, Nodes: []string{"g14", "ace"}, + Manifest: catalogue.Manifest{Module: "mail", Requires: []string{"database"}}} + n := Nodes{ + Online: map[string]bool{"ace": true, "g14": true}, + Haves: map[string]map[string]delivery.Have{ + "ace": {"mail": {FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}, "store": {FP: fpOf('s'), Phase: delivery.PhaseRunning, Health: delivery.HealthOK}}, + "g14": {"mail": {FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}}, + }, + Providers: map[string]map[string][]string{"database": {"store": {"ace"}}}, + } + return m, n +} + +func wantedOld() map[string]delivery.Want { + return map[string]delivery.Want{"ace": {FP: fOld}, "g14": {FP: fOld}} +} + +// 1297/B4: with every rule holding, both nodes' wants move to the new fingerprint (the canary ace runs it healthy, so +// g14 follows at once: step 0, ADR 0528 clause 3). +func TestEveryRuleHoldsSoTheWantsMove(t *testing.T) { + m, n := ready() + out := Wants(m, n, wantedOld(), nil) + for _, node := range []string{"ace", "g14"} { + if out.Wants[node].FP != fNew || out.Wants[node].Rule != 0 { + t.Errorf("%s's want = %+v, want %s", node, out.Wants[node], fNew) + } + } +} + +// 1297/B4, one test per rule: given every rule but N holds, no want moves to F, and the rule N is named. +func TestEachRuleHoldsTheWantBack(t *testing.T) { + for rule, breakIt := range map[int]func(*Module, *Nodes){ + delivery.RuleBuildDone: func(m *Module, _ *Nodes) { m.Built = false }, + delivery.RuleNotHeld: func(m *Module, _ *Nodes) { m.Held = true }, + delivery.RuleCanary: func(_ *Module, n *Nodes) { + n.Haves["ace"]["mail"] = delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK} + }, + delivery.RuleProviders: func(_ *Module, n *Nodes) { + n.Haves["ace"]["store"] = delivery.Have{FP: fpOf('s'), Phase: delivery.PhaseRunning, Health: delivery.HealthWaits} + }, + delivery.RuleGrant: func(m *Module, _ *Nodes) { m.Granted = false }, + } { + m, n := ready() + breakIt(&m, &n) + out := Wants(m, n, wantedOld(), nil) + if w, ok := out.Wants["g14"]; ok && w.FP == fNew { + t.Errorf("rule %d broken: g14's want moved to F", rule) + } + if got := out.Rules["g14"].Rule; got != rule { + t.Errorf("rule %d broken: g14 is held back by rule %d (%s)", rule, got, out.Rules["g14"].Words) + } + if w := out.Wants["g14"]; w.FP != "" && (w.FP != fOld || w.Newest != fNew || w.Rule != rule) { + t.Errorf("rule %d broken: g14's want is %+v", rule, w) + } + } +} + +// 1297/B4, ADR 0435 §3: a canary that is offline is passed over: the next node by name is the canary, and moves first. +func TestAnOfflineCanaryIsPassedOver(t *testing.T) { + m, n := ready() + n.Online["ace"] = false + n.Haves["ace"]["mail"] = delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK} + n.Providers["database"]["store"] = []string{"ace", "g14"} + n.Haves["g14"]["store"] = delivery.Have{FP: fpOf('s'), Phase: delivery.PhaseRunning, Health: delivery.HealthOK} + if c := Canary(m.Nodes, n.Online); c != "g14" { + t.Fatalf("canary = %q, want g14", c) + } + out := Wants(m, n, wantedOld(), nil) + if out.Wants["g14"].FP != fNew { + t.Errorf("the canary g14's want = %+v", out.Wants["g14"]) + } + if out.Rules["ace"].Rule != delivery.RuleCanary { + t.Errorf("ace, behind the canary g14, is held back by %+v", out.Rules["ace"]) + } +} + +// 1297/B4: a failed health check of F holds F and sets the nodes that want it back to their last healthy fingerprint. +func TestAFailedHealthHoldsAndSetsBack(t *testing.T) { + m, n := ready() + n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"} + out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fNew}, "g14": {FP: fOld}}, map[string]string{"ace": fOld}) + if out.Held == nil || out.Held.Node != "ace" || out.Held.Words != "connection refused" { + t.Fatalf("held = %+v", out.Held) + } + if w := out.Wants["ace"]; w.FP != fOld || w.Newest != fNew || w.Rule != delivery.RuleNotHeld { + t.Errorf("ace's want = %+v, want set back to the last healthy fingerprint", w) + } + if w, ok := out.Wants["g14"]; ok && w.FP == fNew { + t.Errorf("g14 moved to a held fingerprint: %+v", w) + } +} + +// 1297/B4: a module holding data is not set back; its want stays and says why. +func TestADataModuleIsNotSetBack(t *testing.T) { + m, n := ready() + m.Manifest.Data = &catalogue.Data{Own: []catalogue.DataItem{{ID: "mail", Class: catalogue.ClassIrreplaceable}}} + n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed} + out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fNew}}, map[string]string{"ace": fOld}) + if out.Held == nil { + t.Fatal("F was not held") + } + if w := out.Wants["ace"]; w.FP != fNew || !strings.Contains(w.Words, "holds data") { + t.Errorf("ace's want = %+v, want it left at F, saying it holds data", w) + } +} + +// 1297/B4, ADR 0408 §4: a node that leaves the module's nodes gets its want written removed, never deleted; a want that +// is already as the rules give it is not written again. +func TestALeavingNodeIsRemovedAndNothingIsWrittenTwice(t *testing.T) { + m, n := ready() + m.Nodes = []string{"ace"} + out := Wants(m, n, map[string]delivery.Want{"ace": {V: 1, FP: fNew}, "g14": {FP: fOld}}, nil) + if w := out.Wants["g14"]; !w.Removed { + t.Errorf("g14's want = %+v, want removed", w) + } + if _, ok := out.Wants["ace"]; ok { + t.Errorf("ace's want, unchanged, was written again: %+v", out.Wants["ace"]) + } +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go new file mode 100644 index 0000000..417ac68 --- /dev/null +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -0,0 +1,57 @@ +package manifest + +import ( + "os" + "reflect" + "slices" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +func load(t *testing.T) catalogue.Manifest { + t.Helper() + raw, err := os.ReadFile("module.json") + if err != nil { + t.Fatal(err) + } + m, err := catalogue.ParseManifest(raw) + if err != nil { + t.Fatalf("mesh-sdk's manifest reader refuses the manifest: %v", err) + } + return m +} + +// 1297/B2a: the controller's reader (mesh-sdk catalogue, which the controller reads manifests with) takes the +// manifest; it reads exactly the buckets the assigner opens (the program's list, held equal in its own test), keeps +// its three states, and holds a credential of its own. +func TestManifest(t *testing.T) { + m := load(t) + if m.Module != "mesh-delivery-assigner" { + t.Errorf("module = %q", m.Module) + } + reads := slices.Clone(m.Reads) + slices.Sort(reads) + want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.grants", + "mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", + "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"} + if !reflect.DeepEqual(reads, want) { + t.Errorf("reads = %v, want exactly %v", reads, want) + } + states := map[string]int{} + for _, s := range m.State { + states[s.Name] = s.History + } + if want := map[string]int{"compositions": 5, "differences": 64, "wants": 20}; !reflect.DeepEqual(states, want) { + t.Errorf("state = %v, want %v", states, want) + } + if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { + t.Errorf("own-secrets = %v, want exactly broker (or no credential is issued)", m.OwnSecrets) + } + if m.SecretsOwner != "mesh-delivery-assigner" { + t.Errorf("secrets-owner = %q", m.SecretsOwner) + } + if len(m.Claims) != 0 || len(m.Tools) != 0 || len(m.Invokes) != 0 { + t.Errorf("the assigner claims, serves or calls something: claims %v, tools %v, invokes %v", m.Claims, m.Tools, m.Invokes) + } +} diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json new file mode 100644 index 0000000..599c3f6 --- /dev/null +++ b/modules/mesh-delivery-assigner/module.json @@ -0,0 +1,70 @@ +{ + "module": "mesh-delivery-assigner", + "version": "1", + "slug": "assigner", + "reads": [ + "mesh-controller.assignments", + "mesh-controller.builds", + "mesh-controller.grants", + "mesh-delivery-builder.builds", + "mesh-delivery-indexer.candidates", + "mesh-delivery-indexer.manifests", + "mesh-delivery-indexer.modules", + "mesh-delivery-watchman.presence", + "node-engine.have" + ], + "state": [ + { "name": "compositions", "history": 5 }, + { "name": "differences", "history": 64 }, + { "name": "wants", "history": 20 } + ], + "own-secrets": { + "broker": "${dir:state}/broker" + }, + "secrets-owner": "mesh-delivery-assigner", + "resources": [ + { + "id": "account", + "type": "user", + "name": "mesh-delivery-assigner", + "shell": "/usr/bin/nologin", + "home": "/var/lib/mesh-delivery-assigner" + }, + { + "id": "state", + "type": "directory", + "mode": "0700", + "owner": "mesh-delivery-assigner", + "place": ".", + "unknown": "program" + }, + { + "id": "assigner", + "type": "process", + "name": "mesh-delivery-assigner", + "artifact": "program", + "run": [ + "./mesh-delivery-assigner" + ], + "env": { + "MESH_BROKER_FILE": "${dir:state}/broker" + }, + "user": "mesh-delivery-assigner", + "health": { + "kind": "unit" + } + } + ], + "build": { + "artifacts": [ + { + "name": "program", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/mesh-delivery-assigner", + "binary": "mesh-delivery-assigner" + } + ] + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go new file mode 100644 index 0000000..5770315 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go @@ -0,0 +1,99 @@ +package main + +import ( + "context" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-checker/internal/check" +) + +// compositionWatch reads the assigner's composition of a check (ADR 0528 clause 5) by watching its key and the +// assigner's presence: no poll and no timer. With no assigner online it answers check.ErrAssignerNotRunning at once. +type compositionWatch struct { + kv jetstream.KeyValue + // presence is the watchman's presence bucket; nil (not there) counts as no assigner online. + presence jetstream.KeyValue +} + +// Await answers the composition under key once the assigner wrote one that does not wait (check.AwaitComposition). +func (c compositionWatch) Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + comps := make(chan check.CompositionEvent) + presence := make(chan check.PresenceEvent) + cw, err := c.kv.Watch(ctx, key) + if err != nil { + return delivery.CompositionVerdict{}, err + } + defer cw.Stop() + go func() { + for e := range cw.Updates() { + ev := check.CompositionEvent{Settled: e == nil} + if e != nil && e.Operation() == jetstream.KeyValuePut { + ev.Value = e.Value() + } + select { + case comps <- ev: + case <-ctx.Done(): + return + } + } + }() + if c.presence == nil { + go func() { + select { + case presence <- check.PresenceEvent{Settled: true}: + case <-ctx.Done(): + } + }() + } else { + pw, err := c.presence.Watch(ctx, "part."+delivery.AssignerModule+".*") + if err != nil { + return delivery.CompositionVerdict{}, err + } + defer pw.Stop() + go func() { + for e := range pw.Updates() { + ev := check.PresenceEvent{Settled: e == nil} + if e != nil { + ev.Key = e.Key() + if e.Operation() == jetstream.KeyValuePut { + if p, err := delivery.DecodePresence(e.Value()); err == nil { + ev.Online = p.Online + } + } + } + select { + case presence <- ev: + case <-ctx.Done(): + return + } + } + }() + } + return check.AwaitComposition(ctx, presence, comps) +} + +// followCompositions hands the checker each composition the assigner writes from now on, so a verdict that said its +// composition was not judged takes it in when the assigner starts (event-driven). +func followCompositions(ctx context.Context, kv jetstream.KeyValue, ch *check.Checker) error { + w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + return err + } + go func() { + defer w.Stop() + for e := range w.Updates() { + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + if comp, err := delivery.DecodeCompositionVerdict(e.Value()); err == nil { + ch.Composed(ctx, comp) + } + } + }() + return nil +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go new file mode 100644 index 0000000..0325a35 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go @@ -0,0 +1,106 @@ +//go:build integration + +package main + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-checker/internal/check" +) + +// 1297/AC3 (its bus half): on the real bus, the checker's read of the assigner's composition waits while the key is +// absent and while the assigner says it waits, and answers the verdict the assigner then writes under the check's key. +func TestTheCheckerAwaitsTheAssignersComposition(t *testing.T) { + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "comp") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: filepath.Join(dir, "js"), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + nc, err := nats.Connect(srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerCompositionsBucket, History: delivery.AssignerCompositionsHistory}) + if err != nil { + t.Fatal(err) + } + pkv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.PresenceBucket, History: 5}) + if err != nil { + t.Fatal(err) + } + pkey, _ := delivery.PresencePartKey(delivery.AssignerModule, "novox") + on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) + if _, err := pkv.Put(ctx, pkey, on); err != nil { + t.Fatal(err) + } + // With no assigner online the wait answers at once (a second key, before the assigner of this test is online). + off, _ := delivery.EncodePresence(delivery.Presence{Online: false}) + if _, err := pkv.Put(ctx, pkey, off); err != nil { + t.Fatal(err) + } + other, _ := delivery.CompositionKey("feat/y", "0123456789ab") + if _, err := (compositionWatch{kv: kv, presence: pkv}).Await(ctx, other); !errors.Is(err, check.ErrAssignerNotRunning) { + t.Fatalf("with no assigner online the wait answered %v", err) + } + if _, err := pkv.Put(ctx, pkey, on); err != nil { + t.Fatal(err) + } + key, _ := delivery.CompositionKey("feat/x", "0123456789ab") + got := make(chan delivery.CompositionVerdict, 1) + go func() { + v, err := compositionWatch{kv: kv, presence: pkv}.Await(ctx, key) + if err != nil { + t.Error(err) + } + got <- v + }() + put := func(v delivery.CompositionVerdict) { + data, err := delivery.EncodeCompositionVerdict(v) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, key, data); err != nil { + t.Fatal(err) + } + } + time.Sleep(200 * time.Millisecond) + put(delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionWaits, Words: "the indexer holds no candidates yet"}) + select { + case v := <-got: + t.Fatalf("a waiting composition was answered: %+v", v) + case <-time.After(300 * time.Millisecond): + } + put(delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "x", Machines: []string{"g14"}}) + select { + case v := <-got: + if v.State != delivery.CompositionPass || v.Machines[0] != "g14" { + t.Errorf("answered %+v", v) + } + case <-time.After(5 * time.Second): + t.Fatal("the composition written was not answered") + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go index ed8ff96..ecc4e6b 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go @@ -27,11 +27,13 @@ type judgeDeps struct { forge check.Forge announce check.Announcer builds check.BuildStates - logf func(format string, args ...any) + // compositions is the assigner's bucket of compositions; nil leaves every composition not run. + compositions check.CompositionReader + logf func(format string, args ...any) } func newJudge(d judgeDeps) check.Judge { - return &check.Examiner{Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places, + return &check.Examiner{Compositions: d.compositions, Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places, Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Packages: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf} } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 41377e8..08b3c0b 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -222,6 +222,24 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro default: builds = buildReader{kvStore{kv}} } + // The assigner's compositions (ADR 0528 clause 5, issue 1297); where its bucket is not there yet a verdict says the + // composition was not run, until the checker starts again. + var compositions check.CompositionReader + switch kv, err := js.KeyValue(ctx, delivery.AssignerCompositionsBucket); { + case errors.Is(err, jetstream.ErrBucketNotFound): + logger.Printf("the bucket %s does not exist yet: a verdict says the composition was not run until the checker starts again", delivery.AssignerCompositionsBucket) + case err != nil: + return fmt.Errorf("the bucket %s: %w", delivery.AssignerCompositionsBucket, err) + default: + cw := compositionWatch{kv: kv} + if pkv, err := js.KeyValue(ctx, delivery.PresenceBucket); err == nil { + cw.presence = pkv + } + compositions = cw + if err := followCompositions(ctx, kv, ch); err != nil { + return fmt.Errorf("the bucket %s cannot be watched: %w", delivery.AssignerCompositionsBucket, err) + } + } // The tries (ADR 0501 §5) are judged by the same judge, held by the same deliveries. tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, Logf: logger.Printf} holds := newHoldWatch(ctx, js, func(ctx context.Context) { ch.Reconsider(ctx); tryer.Reconsider(ctx) }, logger) @@ -238,7 +256,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro runner: runner, tools: tools, presence: presence, have: have, places: places, trunks: trunks, kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]}, repos: repos, builds: builds, forge: forgeReads{call: caller.Call}, logf: logger.Printf, - announce: announce, + announce: announce, compositions: compositions, }) ch.Announce = announce tryer.Judge = ch.Judge diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 0de4a55..53b499f 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023134155-3e5d4c30ffa7 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/internal/check/awaiting.go b/modules/mesh-delivery-checker/internal/check/awaiting.go new file mode 100644 index 0000000..1fbae2c --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/awaiting.go @@ -0,0 +1,141 @@ +package check + +import ( + "context" + "errors" + "fmt" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ErrAssignerNotRunning is what a wait for a composition answers when no instance of the assigner is online: nobody is +// making it, so the check does not wait for it (issue 1297). The verdict says the composition is not judged, and when the +// assigner starts and writes it, the checker takes it in then (Checker.Composed). +var ErrAssignerNotRunning = errors.New(delivery.AssignerNotRunning) + +// PresenceEvent is a change of the watchman's presence bucket, or Settled once what stands was handed. +type PresenceEvent struct { + Key string + Online bool + Settled bool +} + +// CompositionEvent is a change of the composition's key (its value, nil for a delete), or Settled once what stands was +// handed. +type CompositionEvent struct { + Value []byte + Settled bool +} + +// AwaitComposition answers the assigner's composition once one is written that does not wait, while an instance of the +// assigner is online by the watchman's presence (`part.mesh-delivery-assigner.`). Once both watches handed +// what stands, no instance online is ErrAssignerNotRunning at once, and so is the last instance going offline during the +// wait. No timer: every answer follows an event; ctx ending is ctx's error. +func AwaitComposition(ctx context.Context, presence <-chan PresenceEvent, comps <-chan CompositionEvent) (delivery.CompositionVerdict, error) { + online := map[string]bool{} + presenceSettled, compSettled := false, false + for { + select { + case <-ctx.Done(): + return delivery.CompositionVerdict{}, ctx.Err() + case e, ok := <-presence: + if !ok { + return delivery.CompositionVerdict{}, errors.New("the watch of the assigner's presence ended") + } + if e.Settled { + presenceSettled = true + break + } + if _, part, instance, err := delivery.ParsePresenceKey(e.Key); err == nil && part == delivery.AssignerModule { + if e.Online { + online[instance] = true + } else { + delete(online, instance) + } + } + case e, ok := <-comps: + if !ok { + return delivery.CompositionVerdict{}, errors.New("the watch of the composition ended") + } + if e.Settled { + compSettled = true + break + } + if e.Value == nil { + break + } + v, err := delivery.DecodeCompositionVerdict(e.Value) + if err != nil { + return delivery.CompositionVerdict{}, fmt.Errorf("the composition cannot be read: %w", err) + } + if v.State != delivery.CompositionWaits { + return v, nil + } + } + if presenceSettled && compSettled && len(online) == 0 { + return delivery.CompositionVerdict{}, ErrAssignerNotRunning + } + } +} + +// withComposition is v with the assigner's composition taken in: a failed one fails the check, naming its machine. +func withComposition(v *delivery.Verdict, comp delivery.CompositionVerdict) { + v.Composition = comp.Composition() + v.Took.Composition = comp.Took + if comp.State == delivery.CompositionFail { + machine := comp.Machine + if machine == "" { + machine = "the composition" + } + v.State, v.Failed = delivery.VerdictFail, delivery.FailedFirst{Machine: machine, Words: comp.Words} + } +} + +// notJudged says whether a verdict's composition was not judged because the assigner was not running. +func notJudged(v delivery.Verdict) bool { + return v.Composition != nil && v.Composition.State == delivery.CompositionNotRun && v.Composition.Words == delivery.AssignerNotRunning +} + +// Composed takes in a composition the assigner wrote after the check's verdict said it was not judged because the +// assigner was not running (the assigner started since): the verdict and its status are written again with it. A +// composition of heads no candidate holds now, or of a verdict that judged its composition, is left. +func (ch *Checker) Composed(ctx context.Context, comp delivery.CompositionVerdict) { + ch.init() + if comp.State == delivery.CompositionWaits { + return + } + ckey, err := delivery.CandidatesKey(comp.Group) + if err != nil { + return + } + ch.mu.Lock() + c, ok := ch.known[ckey] + ch.mu.Unlock() + if !ok { + return + } + cs := CaseOf(c, ch.heads()) + if cs == nil || cs.HeadsHash != comp.Heads { + return + } + cur, rev, found, err := ch.Verdicts.Get(ctx, cs.Key) + if err != nil || !found { + return + } + v, err := delivery.DecodeVerdict(cur) + if err != nil || v.State != delivery.VerdictPass || !notJudged(v) { + return + } + withComposition(&v, comp) + data, err := delivery.EncodeVerdict(v) + if err != nil { + ch.logf("%s: the verdict with its composition cannot be written: %v", cs.Key, err) + return + } + if _, err := ch.Verdicts.Update(ctx, cs.Key, data, rev); err != nil { + ch.logf("%s: the verdict with its composition could not be written: %v", cs.Key, err) + return + } + ch.writeOwed(ctx, cs, v, Outcome{Verdict: v}) + ch.logf("%s: the composition came after the verdict: %s", cs.Key, comp.State) +} diff --git a/modules/mesh-delivery-checker/internal/check/awaiting_test.go b/modules/mesh-delivery-checker/internal/check/awaiting_test.go new file mode 100644 index 0000000..c775c49 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/awaiting_test.go @@ -0,0 +1,149 @@ +package check + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func assignerOn(instance string, online bool) PresenceEvent { + key, _ := delivery.PresencePartKey(delivery.AssignerModule, instance) + return PresenceEvent{Key: key, Online: online} +} + +func compEvent(t *testing.T, v delivery.CompositionVerdict) CompositionEvent { + t.Helper() + data, err := delivery.EncodeCompositionVerdict(v) + if err != nil { + t.Fatal(err) + } + return CompositionEvent{Value: data} +} + +var passed = delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "x", Machines: []string{"g14"}} + +// 1297/B3d: with no assigner online, the wait answers ErrAssignerNotRunning as soon as both watches handed what stands. +func TestNoAssignerAnswersAtOnce(t *testing.T) { + presence, comps := make(chan PresenceEvent, 4), make(chan CompositionEvent, 4) + presence <- assignerOn("novox", false) + presence <- PresenceEvent{Settled: true} + comps <- CompositionEvent{Settled: true} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if _, err := AwaitComposition(ctx, presence, comps); !errors.Is(err, ErrAssignerNotRunning) { + t.Fatalf("answered %v, want the assigner not running", err) + } +} + +// 1297/B3d: with the assigner online it waits for the key, past a composition that waits, and answers the verdict. +func TestAnOnlineAssignerIsWaitedFor(t *testing.T) { + presence, comps := make(chan PresenceEvent, 4), make(chan CompositionEvent, 4) + presence <- assignerOn("novox", true) + presence <- PresenceEvent{Settled: true} + comps <- CompositionEvent{Settled: true} + done := make(chan error, 1) + var got delivery.CompositionVerdict + go func() { + var err error + got, err = AwaitComposition(context.Background(), presence, comps) + done <- err + }() + comps <- compEvent(t, delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionWaits, Words: "w"}) + select { + case err := <-done: + t.Fatalf("answered before the composition was written: %v", err) + case <-time.After(50 * time.Millisecond): + } + comps <- compEvent(t, passed) + if err := <-done; err != nil || got.State != delivery.CompositionPass { + t.Fatalf("answered %+v, %v", got, err) + } + // And the last instance going offline during a wait ends it as not running. + presence2, comps2 := make(chan PresenceEvent, 4), make(chan CompositionEvent, 4) + presence2 <- assignerOn("novox", true) + presence2 <- PresenceEvent{Settled: true} + comps2 <- CompositionEvent{Settled: true} + presence2 <- assignerOn("novox", false) + if _, err := AwaitComposition(context.Background(), presence2, comps2); !errors.Is(err, ErrAssignerNotRunning) { + t.Errorf("the assigner going offline answered %v", err) + } +} + +// 1297/B3d: a passing check with no assigner says at once, loud, that its composition is not judged; never a pass of it. +func TestAPassingCheckWithNoAssignerSaysTheCompositionIsNotJudged(t *testing.T) { + b := newBench(t) + b.ex.Compositions = &fakeCompositions{err: ErrAssignerNotRunning} + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict + if v.Composition == nil || v.Composition.State != delivery.CompositionNotRun || v.Composition.Words != delivery.AssignerNotRunning { + t.Fatalf("composition = %+v", v.Composition) + } + if words := compositionWords(v.Composition, 0); words != "composition not judged: the assigner is not running" { + t.Errorf("the line says %q", words) + } +} + +// 1297/B3d: the assigner starting writes the composition later, and the checker takes it into the verdict that said it +// was not judged, writing the status again; a failed one fails the check. +func TestTheAssignerStartingBringsTheComposition(t *testing.T) { + for _, comp := range []delivery.CompositionVerdict{passed, + {Group: "feat/x", Heads: "", State: delivery.CompositionFail, SDK: "x", Machine: "g14", Words: "g14 runs web, which the change removes"}} { + ch, cs := checkerWithNotJudged(t) + comp.Heads = cs.HeadsHash + ch.Composed(context.Background(), comp) + raw, _, _, _ := ch.Verdicts.Get(context.Background(), cs.Key) + v, err := delivery.DecodeVerdict(raw) + if err != nil || v.Composition == nil || v.Composition.State != comp.State { + t.Fatalf("after the composition came, the verdict is %+v, %v", v, err) + } + if comp.State == delivery.CompositionFail && (v.State != delivery.VerdictFail || v.Failed.Machine != "g14") { + t.Errorf("a failed composition left the verdict %+v", v) + } + owed := ch.Owed.(*memKV) + keys, _ := owed.Keys(context.Background()) + if len(keys) == 0 { + t.Fatal("no status was written again") + } + data, _, _, _ := owed.Get(context.Background(), keys[0]) + o, _ := delivery.DecodeCheckerOwed(data) + if !strings.Contains(o.Text, "composition passed") && !strings.Contains(o.Text, "g14 runs web") { + t.Errorf("the status's line says %q", o.Text) + } + } +} + +// checkerWithNotJudged is a checker holding a candidate whose passing verdict said its composition was not judged. +func checkerWithNotJudged(t *testing.T) (*Checker, *Case) { + t.Helper() + cs := caseOf(t, pullOf(repoA, 7, head1)) + raw := candidateOf(t, cs.Group, pullOf(repoA, 7, head1)) + ch := &Checker{Verdicts: newMemKV(), Owed: newMemKV(), Trunks: fixedTrunks{cs.Trunk}} + ckey, _ := delivery.CandidatesKey(cs.Group) + ch.init() + c, err := delivery.DecodeCandidate(raw) + if err != nil { + t.Fatal(err) + } + ch.known[ckey] = c + if got := CaseOf(c, ch.heads()); got == nil || got.Key != cs.Key { + t.Fatalf("the case is %+v, want %s", got, cs.Key) + } + v := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, State: delivery.VerdictPass, + Levels: map[string]delivery.LevelResult{delivery.LevelUnit: {Run: 1}, delivery.LevelIntegration: {Run: 1}}, + Composition: &delivery.Composition{State: delivery.CompositionNotRun, Words: delivery.AssignerNotRunning}} + data, err := delivery.EncodeVerdict(v) + if err != nil { + t.Fatal(err) + } + if _, err := ch.Verdicts.Create(context.Background(), cs.Key, data); err != nil { + t.Fatal(err) + } + return ch, cs +} + +type fixedTrunks struct{ heads map[string]string } + +func (f fixedTrunks) Heads() map[string]string { return f.heads } diff --git a/modules/mesh-delivery-checker/internal/check/composition_test.go b/modules/mesh-delivery-checker/internal/check/composition_test.go new file mode 100644 index 0000000..19bb6b6 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/composition_test.go @@ -0,0 +1,80 @@ +package check + +import ( + "context" + "errors" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// fakeCompositions answers the assigner's composition of a check from memory, recording the keys asked. +type fakeCompositions struct { + v delivery.CompositionVerdict + err error + asked []string +} + +func (f *fakeCompositions) Await(_ context.Context, key string) (delivery.CompositionVerdict, error) { + f.asked = append(f.asked, key) + return f.v, f.err +} + +// 1297/AC3 (its unit half): a check whose tests pass takes the assigner's composition of it, by the check's own key, +// into its verdict: a pass says it passed with the mesh-sdk and machines it composed, and is no longer "not run". +func TestAPassingCheckTakesTheAssignersComposition(t *testing.T) { + b := newBench(t) + f := &fakeCompositions{v: delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "v0.1.14-x", Machines: []string{"g14"}, Took: 40}} + b.ex.Compositions = f + cs := caseOf(t, pullOf(repoA, 7, head1)) + v := b.judge(t, cs).Verdict + if v.State != delivery.VerdictPass || v.Composition == nil || v.Composition.State != delivery.CompositionPass || v.Composition.SDK != "v0.1.14-x" || v.Took.Composition != 40 { + t.Fatalf("verdict = %+v, composition %+v", v, v.Composition) + } + if len(f.asked) != 1 || f.asked[0] != cs.Key { + t.Errorf("asked %v, want the check's key %s", f.asked, cs.Key) + } + if !compositionMentions(v, "passed") { + t.Errorf("the line says %q", compositionWords(v.Composition, v.Took.Composition)) + } +} + +func compositionMentions(v delivery.Verdict, word string) bool { + return strings.Contains(compositionWords(v.Composition, v.Took.Composition), word) +} + +// 1297/AC2 at the checker: a failed composition fails the check, naming the machine it failed at and its words. +func TestAFailedCompositionFailsTheCheck(t *testing.T) { + b := newBench(t) + b.ex.Compositions = &fakeCompositions{v: delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionFail, SDK: "x", + Machine: "g14", Words: "g14 runs web, which the change removes from its source"}} + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict + if v.State != delivery.VerdictFail || v.Failed.Machine != "g14" || v.Composition == nil || v.Composition.State != delivery.CompositionFail { + t.Fatalf("verdict = %+v", v) + } + v.Group, v.Heads = "feat/x", "0123456789ab" // the checker names them when it writes the verdict + if _, err := delivery.EncodeVerdict(v); err != nil { + t.Errorf("the verdict cannot be written: %v", err) + } +} + +// 1297/B3d: a composition that cannot be read leaves it not run; a try is not composed. +func TestACompositionNotHadIsNotRun(t *testing.T) { + b := newBench(t) + f := &fakeCompositions{err: errors.New("the check's bound passed")} + b.ex.Compositions = f + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict + if v.State != delivery.VerdictPass || v.Composition == nil || v.Composition.State != delivery.CompositionNotRun { + t.Fatalf("verdict = %+v", v) + } + b2 := newBench(t) + f2 := &fakeCompositions{v: delivery.CompositionVerdict{State: delivery.CompositionFail, Words: "w", SDK: "x"}} + b2.ex.Compositions = f2 + cs := caseOf(t, pullOf(repoA, 7, head1)) + cs.Try = &TryOptions{} + b2.judge(t, cs) + if len(f2.asked) != 0 { + t.Errorf("a try asked for a composition: %v", f2.asked) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index dcd07f0..d7e420c 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -3,6 +3,7 @@ package check import ( "context" "encoding/json" + "errors" "fmt" "path" "slices" @@ -105,8 +106,18 @@ type Examiner struct { } // Trunks are the trunk heads of every repository, for the siblings a repository's tests read. Trunks interface{ Heads() map[string]string } - Now func() time.Time - Logf func(format string, args ...any) + // Compositions answers the assigner's composition check of a check by the check's key (delivery.CompositionKey), + // waiting until the assigner wrote a verdict that does not wait, or ctx ends (ADR 0528 clause 5, issue 1297); nil + // leaves the composition not run. + Compositions CompositionReader + Now func() time.Time + Logf func(format string, args ...any) +} + +// CompositionReader answers the assigner's composition of a check by the check's key, waiting until the assigner wrote +// a verdict that does not wait, or ctx ends. +type CompositionReader interface { + Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) } func (j *Examiner) now() time.Time { @@ -230,13 +241,36 @@ func (j *Examiner) Judge(ctx context.Context, c *Case) (Outcome, error) { return st.finish(ctx, delivery.Verdict{State: delivery.VerdictPass}), nil } -// finish is the outcome of the verdict with the state of the builds the candidate's modules carry. +// finish is the outcome of the verdict with the state of the builds the candidate's modules carry, and, for a check +// whose tests passed, the assigner's composition of it. func (st *state) finish(ctx context.Context, v delivery.Verdict) Outcome { out := st.outcome(v) + if out.Verdict.State == delivery.VerdictPass { + st.composed(ctx, &out.Verdict) + } out.Builds = st.j.buildNotes(ctx, st.c) return out } +// composed reads the assigner's composition of the check into its verdict (ADR 0528 clause 5): a failed composition +// fails the check, naming the machine it failed at. A try is not composed (it has no candidate); a composition that +// cannot be had leaves it not run, said; with no assigner running it is not judged, said at once. +func (st *state) composed(ctx context.Context, v *delivery.Verdict) { + if st.j.Compositions == nil || st.c.Try != nil { + return + } + comp, err := st.j.Compositions.Await(ctx, st.c.Key) + switch { + case errors.Is(err, ErrAssignerNotRunning): + v.Composition = &delivery.Composition{State: delivery.CompositionNotRun, Words: delivery.AssignerNotRunning} + return + case err != nil: + st.j.logf("%s: the composition could not be read, so it is said as not run: %v", st.c.Key, err) + return + } + withComposition(v, comp) +} + // outcome fills the verdict's levels, times, machines and composition from what ran. func (st *state) outcome(v delivery.Verdict) Outcome { if v.State == delivery.VerdictPass || len(v.Levels) == 0 { diff --git a/modules/mesh-delivery-checker/internal/check/line.go b/modules/mesh-delivery-checker/internal/check/line.go index 85e6830..1030f52 100644 --- a/modules/mesh-delivery-checker/internal/check/line.go +++ b/modules/mesh-delivery-checker/internal/check/line.go @@ -99,6 +99,9 @@ func compositionWords(c *delivery.Composition, took int64) string { return compo // composition is compositionWords; short counts the machines and leaves the SDK commit to the report. func composition(c *delivery.Composition, took int64, short bool) string { + if c != nil && c.State == delivery.CompositionNotRun && c.Words != "" { + return "composition not judged: " + c.Words + } if c == nil || c.State == delivery.CompositionNotRun { return "composition not yet run" } diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index a4b5442..0fc6074 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -96,7 +96,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-controller.plans-building", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", + "mesh-controller.plans-building", "mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 2c22576..1f15b6c 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -22,6 +22,7 @@ "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.plans-building", + "mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 6390b2e..47ca71f 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023134155-3e5d4c30ffa7 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/internal/index/previews.go b/modules/mesh-delivery-indexer/internal/index/previews.go index 5e8b454..d22a762 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews.go +++ b/modules/mesh-delivery-indexer/internal/index/previews.go @@ -23,8 +23,9 @@ const ( // NonePreviewConflict opens the reason of a module whose pull request's head does not merge into the trunk; the files // in conflict follow. NonePreviewConflict = "does not merge: the head conflicts with the trunk in " - // NonePreviewNoModule is for a module whose directory the preview names no manifest for. - NonePreviewNoModule = "the merge preview names no manifest for the module's directory" + // NonePreviewNoModule is for a module whose directory the preview names no manifest for: mesh-sdk's words, which the + // assigner reads as the change removing the module (issue 1297). + NonePreviewNoModule = delivery.CandidateNoneNoModule // NonePreviewNoManifest is for a module whose manifest at the merge the indexer does not hold. NonePreviewNoManifest = "the manifest of the merge is not kept" // NoneNoDirectory is for a module whose directory in its repository neither the modules bucket nor the preview tells.