package main import ( "context" "encoding/json" "fmt" "slices" "strings" "testing" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/compose" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/inventory" ) // fakeBuckets is the bus's buckets by name, for deleting. type fakeBuckets map[string]bool func (f fakeBuckets) DeleteKeyValue(_ context.Context, bucket string) error { if !f[bucket] { return jetstream.ErrBucketNotFound } delete(f, bucket) return nil } // 1945/AC14 (unit half): the first start deletes the merge window's, the silent sends' and the building plans' buckets // and says each once; a later start finds none and says nothing; a bucket of anything else is left alone. func TestTheFirstStartDeletesThePlansBucketsAndSaysEachOnce(t *testing.T) { bus := fakeBuckets{"mesh-controller_merge-window": true, "mesh-controller_rollout-silent": true, "mesh-controller_plans-building": true, "mesh-controller_grants": true} var said []string deleted, err := deletePlanStateBuckets(context.Background(), bus, func(b string) { said = append(said, b) }) if err != nil || len(deleted) != 3 || strings.Join(said, " ") != strings.Join(broker.PlanStateBuckets, " ") { t.Fatalf("first start: deleted %v, said %v, %v", deleted, said, err) } if !bus["mesh-controller_grants"] || len(bus) != 1 { t.Errorf("the bus holds %v after the first start", bus) } said = nil if deleted, err := deletePlanStateBuckets(context.Background(), bus, func(b string) { said = append(said, b) }); err != nil || len(deleted)+len(said) != 0 { t.Errorf("a later start deleted %v and said %v (%v)", deleted, said, err) } } // An earlier fingerprint is composed from its own build's manifest with today's inputs: the controller this one // replaced (1945/AC6), and every kept composition of a module whose config changed (ADR 0545: on 2026-10-25 a want back // to an earlier mesh-build-go on g14 applied the capacity of 0 its kept composition was made with). The composition // holds what that manifest declares and today's config, the kept keys are those of the touched modules at a // fingerprint other than the registered one, the build of a fingerprint is its newest that worked, and a write of // mesh-config's resolved values names the node and module it touches. func TestAnEarlierFingerprintIsComposedFromItsOwnManifest(t *testing.T) { fp := func(c string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(c, 64) } t.Run("from its own manifest, with today's inputs", func(t *testing.T) { manifest := func(version string) catalogue.Manifest { return catalogue.Manifest{Module: controllerModule, Version: version, Settings: map[string]catalogue.SettingDeclaration{ "capacity": {Kind: catalogue.KindPreference, Default: float64(0), Why: "how many builds at once"}}, Resources: []map[string]any{{"id": "settings", "type": "file", "path": "/etc/mesh-controller/settings", "content": version + " ${setting:capacity}\n"}}} } now, before := manifest("six"), manifest("five") in := compose.Input{Catalogue: map[string]catalogue.Manifest{controllerModule: now}, Nodes: []compose.Node{{Name: "anchor", Assigned: []string{controllerModule}, Stated: true, Settings: map[string][]catalogue.Layer{controllerModule: {{From: "mesh", Values: map[string]any{"capacity": float64(4)}}}}}}} for name, compose := range map[string]func() (sdkdelivery.Composed, error){ "the replaced controller": func() (sdkdelivery.Composed, error) { return replacedComposition(in, replacedController{node: "anchor", fp: fp("5"), manifest: before}) }, "a kept composition": func() (sdkdelivery.Composed, error) { return keptComposition(in, "anchor", controllerModule, fp("5"), before) }, } { c, err := compose() if err != nil { t.Fatalf("%s: %v", name, err) } if c.Module != controllerModule || c.FP != fp("5") || c.Node != "anchor" || len(c.Resources) != 1 { t.Fatalf("%s: composed %s at %s on %s with %d resources", name, c.Module, c.FP, c.Node, len(c.Resources)) } var f struct{ Content string } _ = json.Unmarshal(c.Resources[0], &f) if f.Content != "five 4\n" { t.Errorf("%s holds %q, want its own manifest's five with today's capacity 4", name, f.Content) } } if in.Catalogue[controllerModule].Version != "six" { t.Errorf("composing an earlier fingerprint changed the Input's catalogue") } }) t.Run("which build, which keys, which touch", func(t *testing.T) { m := []byte(`{"module":"mesh-controller","version":"5"}`) builds := []inventory.Build{ {ID: "c6", SourceFingerprint: fp("6"), Manifest: m}, {ID: "c5-failed", SourceFingerprint: fp("5"), Failed: "no", Manifest: m}, {ID: "c5", SourceFingerprint: fp("5"), Manifest: m}, {ID: "c4", SourceFingerprint: fp("4"), Manifest: m}, } if b, ok := replacedBuild(builds, fp("6")); !ok || b.ID != "c5" { t.Errorf("the replaced build is %+v (%v), want c5", b, ok) } if _, ok := replacedBuild(builds[:1], fp("6")); ok { t.Errorf("a controller with no earlier build replaced one") } if b, ok := buildAt(builds, fp("5")); !ok || b.ID != "c5" { t.Errorf("the build at 5 is %+v (%v), want c5, the one that worked", b, ok) } if _, ok := buildAt(builds, fp("3")); ok { t.Errorf("a fingerprint with no build has one") } key := func(node, module, f string) string { k, _ := sdkdelivery.ComposedKey(node, module, f); return k } keys := []string{key("g14", "mesh-build-go", fp("2")), key("g14", "mesh-build-go", fp("1")), key("g14", "mesh-build-go", fp("3")), key("g14", "hello-web", fp("1")), key("ace", "mesh-build-go", fp("1"))} registered := map[string]string{"mesh-build-go": fp("3"), "hello-web": fp("9")} got := keptToFollow(keys, "g14", map[string]bool{"mesh-build-go": true}, registered) if want := []keptKey{{"mesh-build-go", fp("1")}, {"mesh-build-go", fp("2")}}; !slices.Equal(got, want) { t.Errorf("kept to follow %v, want %v", got, want) } if got := keptToFollow(keys, "g14", nil, registered); len(got) != 3 { t.Errorf("every module's kept compositions: %v", got) } for k, want := range map[string][2]string{"g14.mesh-build-go.capacity": {"g14", "mesh-build-go"}, "g14.mesh-build-go.limits.cpu": {"g14", "mesh-build-go"}, "g14.mesh-build-go": {}, "": {}} { node, module, ok := configTouched(k) if [2]string{node, module} != want || ok != (want[0] != "") { t.Errorf("%q touches %s %s (%v), want %v", k, node, module, ok, want) } } // A value set before the consumer was made reaches the kept compositions: it reads each key once at creation. if c := configFollowConfig(); c.DeliverPolicy != jetstream.DeliverLastPerSubjectPolicy || c.AckPolicy != jetstream.AckExplicitPolicy { t.Errorf("the config consumer delivers %v, acknowledged %v; want each key's last value, acknowledged", c.DeliverPolicy, c.AckPolicy) } }) } // 1945/AC14 (unit half): the declarations' stream stays while one node's have names another node-engine than the one // registered, or none, and goes once every node's names it. func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) { n6 := sdkdelivery.SourcePrefix2 + strings.Repeat("6", 64) nodes := []string{"ace", "g14", "novox"} if declarationsDone(nodes, map[string]string{"ace": n6, "novox": n6}, n6) { t.Errorf("done while g14 says nothing (it sleeps)") } if declarationsDone(nodes, map[string]string{"ace": n6, "g14": "src2:old", "novox": n6}, n6) { t.Errorf("done while g14 runs the old node-engine") } if !declarationsDone(nodes, map[string]string{"ace": n6, "g14": n6, "novox": n6}, n6) { t.Errorf("not done with every node on the new node-engine") } if declarationsDone(nodes, map[string]string{"ace": "", "g14": "", "novox": ""}, "") || declarationsDone(nil, nil, n6) { t.Errorf("done with no fingerprint registered, or no node") } } // streamsGone is a bus that holds none of the streams asked for; every other part of it is unread. type streamsGone struct{ jetstream.JetStream } func (streamsGone) Stream(context.Context, string) (jetstream.Stream, error) { return nil, jetstream.ErrStreamNotFound } // Issue 1945 (the alarm of a stream the start deleted): once the declarations' stream is gone from the bus, the // declarations are retired whatever the nodes run now, so no raise, health check, probe or healer expects the stream // or a node's consumer on it again. A node-engine newer than the registered one, or a node asleep on an older have, // made the done test false after the stream went, and the bus's objects named it missing again (2026-10-25 14:52). func TestTheDeclarationsStayRetiredOnceTheirStreamIsGone(t *testing.T) { retired, err := func() (retired bool, err error) { defer func() { if p := recover(); p != nil { err = fmt.Errorf("it read what the nodes run although the stream is gone: %v", p) } }() return declarationsRetiredOn(t.Context(), nil, streamsGone{}) }() if err != nil || !retired { t.Fatalf("with no declarations' stream on the bus: retired %v (%v), want retired", retired, err) } }