package selfcompose import ( "context" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "github.com/novox/mesh-host/internal/buckets" "github.com/novox/mesh-host/internal/buckets/bucketstest" engcomp "github.com/novox/mesh-host/internal/compositions" "slices" "strings" "sync" "testing" "time" "github.com/novox/mesh-host/internal/agent" ) // fakeComposed is a node's composition state: its keys, and every write and deletion made. type fakeComposed struct { mu sync.Mutex kv map[string]ComposedEntry rev uint64 puts []string deletes []string } func (f *fakeComposed) Entries(context.Context) (map[string]ComposedEntry, error) { f.mu.Lock() defer f.mu.Unlock() out := map[string]ComposedEntry{} for k, v := range f.kv { out[k] = v } return out, nil } func (f *fakeComposed) Put(_ context.Context, key string, value []byte) (uint64, error) { f.mu.Lock() defer f.mu.Unlock() f.rev++ f.kv[key] = ComposedEntry{Value: value, Revision: f.rev} f.puts = append(f.puts, key) return f.rev, nil } func (f *fakeComposed) Delete(_ context.Context, key string) error { f.mu.Lock() defer f.mu.Unlock() delete(f.kv, key) f.deletes = append(f.deletes, key) return nil } func (f *fakeComposed) written() []string { f.mu.Lock() defer f.mu.Unlock() return slices.Clone(f.puts) } // novox/hq ADR 0593 §6: the node publishes each composition it makes in its composition state, a record never applied: // the wanted module and the node's own at the first pass; nothing again for a pass that changes nothing; a done build // of a module placed here that no want names, at its trunk's build, written once as its own key (before any want moves); // the same build landing again writes nothing. And a module's keys are kept while among the five it wrote last or named // by a want, and deleted otherwise. func TestTheNodePublishesItsCompositions(t *testing.T) { t.Run("the verb compose is answered while a pass runs", composeAnsweredWhileAPassRuns) b := newFakeBus() fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64) ofp := meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64) digest := meshcatalogclient.ManifestDigest([]byte(lemurs)) oldie := `{"module":"oldie","version":"1","capabilities":["systemd"],"resources":[{"id":"service","type":"service","unit":"oldie.service","state":"running","boot":"enabled"}]}` odigest := meshcatalogclient.ManifestDigest([]byte(oldie)) brokenJSON := strings.ReplaceAll(oldie, "oldie", "broken") waitingJSON := strings.ReplaceAll(oldie, "oldie", "waiting") bdigest, wdigest := meshcatalogclient.ManifestDigest([]byte(brokenJSON)), meshcatalogclient.ManifestDigest([]byte(waitingJSON)) for d, json := range map[string]string{digest: lemurs, odigest: oldie, bdigest: brokenJSON, wdigest: waitingJSON} { man, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: map[string]string{digest: "lemurs", odigest: "oldie", bdigest: "broken", wdigest: "waiting"}[d], Manifest: json}) key, _ := meshcatalogclient.ManifestsKey(d) b.kv[buckets.Manifests+"/"+key] = man } state := &fakeComposed{kv: map[string]ComposedEntry{}} env := b.env() env.ComposedState = state // Design 47, *What each step says*: each module composition written says composed, with its fingerprint and digest. var saidMu sync.Mutex var saidComposed []string env.SayComposed = func(module, fingerprint, digest string) { saidMu.Lock() defer saidMu.Unlock() saidComposed = append(saidComposed, module+" "+fingerprint+" "+digest) } ctx, cancel := context.WithCancel(context.Background()) defer cancel() c := New(env) out, err := c.Watch(ctx, "n1.>") if err != nil { t.Fatal(err) } place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}, {Module: "broken"}, {Module: "waiting"}}}) have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}) omod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: ofp, Manifest: odigest, Repository: "novox/o", Commit: "c"}) want, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true}) nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true}) b.putAssigned(t, place) b.put(t, buckets.Have, "n1.node", have) b.put(t, buckets.Modules, "lemurs", mod) b.put(t, buckets.Modules, "oldie", omod) bfp := meshcatalogclient.SourcePrefix2 + strings.Repeat("d", 64) wfp := meshcatalogclient.SourcePrefix2 + strings.Repeat("e", 64) for m, f := range map[string]string{"broken": bfp, "waiting": wfp} { raw, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: f, Manifest: map[string]string{"broken": bdigest, "waiting": wdigest}[m], Repository: "novox/o", Commit: "c"}) b.put(t, buckets.Modules, m, raw) } b.put(t, buckets.Wants, "want.n1.lemurs", want) b.put(t, buckets.Wants, "want.n1.node", nodeWant) for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} { b.send(bucket, agent.Update{Caught: true}) } for u := next(t, out); !u.Caught; u = next(t, out) { } lemursKey, _ := engcomp.Key("n1", "lemurs", fp) oldieKey, _ := engcomp.Key("n1", "oldie", ofp) waitFor := func(what string, ok func([]string) bool) []string { t.Helper() deadline := time.Now().Add(5 * time.Second) for time.Now().Before(deadline) { if got := state.written(); ok(got) { return got } time.Sleep(10 * time.Millisecond) } b.mu.Lock() t.Logf("said: %q", b.said) b.mu.Unlock() t.Fatalf("%s: written %v", what, state.written()) return nil } first := waitFor("the first pass", func(got []string) bool { return len(got) >= 2 }) if !slices.Equal(first, []string{lemursKey, "n1.node"}) { t.Errorf("the first pass wrote %v, want the wanted module's key and the node's own", first) } if raw := state.kv[lemursKey].Value; raw != nil { comp, err := engcomp.Decode(raw) if err != nil || comp.Module != "lemurs" || comp.FP != fp || len(comp.Resources) == 0 { t.Errorf("the published composition of lemurs: %+v, %v", comp, err) } } if comp, err := engcomp.Decode(state.kv[lemursKey].Value); err == nil { dig, _ := engcomp.DigestOf(comp) saidMu.Lock() if len(saidComposed) != 1 || saidComposed[0] != "lemurs "+fp+" "+dig || dig == "" { t.Errorf("composed said %v, want once for lemurs at %s with its digest %q (the node's own says none)", saidComposed, fp, dig) } saidMu.Unlock() } // Design 47 G §199, §556 (test 56): what the nodes say they run is acted on once the first read of the stream ended; the // newest by `at` stands, an older one and the same again change nothing. at := time.Now() c.NoteRunning(nodeengine.Running{Node: "n2", Module: "lemurs", Fingerprint: "a", Manifest: "sha256:a", At: at}) if got := c.heardNow(); got != nil { t.Errorf("heard %v before the first read ended", got) } c.RunningRead() c.NoteRunning(nodeengine.Running{Node: "n2", Module: "lemurs", Fingerprint: "old", At: at.Add(-time.Hour)}) c.NoteRunning(nodeengine.Running{Node: "n3", Module: "lemurs", Stopped: true, At: at}) if got := c.heardNow(); len(got) != 2 || got["n2"]["lemurs"].Fingerprint != "a" || !got["n3"]["lemurs"].Stopped { t.Errorf("heard %v, want n2's lemurs at a (not the older) and n3's stopped", got) } c.RunningLost() if got := c.heardNow(); got != nil { t.Errorf("heard %v with the stream lost", got) } // A pass that changes nothing writes nothing. setFor(t, &askedWithin, 0) if _, err := c.ComposeNow(ctx); err != nil { t.Fatal(err) } time.Sleep(50 * time.Millisecond) if got := state.written(); len(got) != 2 { t.Errorf("a pass that changed nothing wrote %v", got[2:]) } // oldie's trunk build is done: its key at that build is written once, though no want names it. build, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: odigest}) bkey, _ := builds.BuilderBuildKey("oldie", ofp) b.mu.Lock() b.kv[builds.BuilderBuildsBucket+"/"+bkey] = build b.mu.Unlock() b.put(t, builds.BuilderBuildsBucket, bkey, build) got := waitFor("oldie's done build", func(got []string) bool { return len(got) >= 3 }) if got[2] != oldieKey { t.Errorf("the done build of oldie wrote %v, want %s", got[2:], oldieKey) } b.put(t, builds.BuilderBuildsBucket, bkey, build) time.Sleep(100 * time.Millisecond) if got := state.written(); len(got) != 3 { t.Errorf("the same build landing again wrote %v", got[3:]) } // A placed module whose newest build failed is published as a refusal at that build with the build's words (the // provisioner holds it and does not wait); one whose build is queued or running stays unpublished. for m, st := range map[string]builds.BuilderBuild{ "broken": {State: builds.BuildFailed, Error: `refused kind: package of language "go"`}, "waiting": {State: builds.BuildQueued}, } { f := map[string]string{"broken": bfp, "waiting": wfp}[m] raw, _ := builds.EncodeBuilderBuild(st) k, _ := builds.BuilderBuildKey(m, f) b.mu.Lock() b.kv[builds.BuilderBuildsBucket+"/"+k] = raw b.mu.Unlock() b.put(t, builds.BuilderBuildsBucket, k, raw) } brokenKey, _ := engcomp.Key("n1", "broken", bfp) got = waitFor("broken's failed build", func(got []string) bool { return len(got) >= 4 }) // The pass that wrote it says so ("1 key(s) written" was said once before, for oldie); then only brokenKey is new. for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) { n := 0 b.mu.Lock() for _, l := range b.said { if strings.Contains(l, "this node's composition state: 1 key(s) written") { n++ } } b.mu.Unlock() if n >= 2 { break } } if got := state.written(); len(got) != 4 || got[3] != brokenKey { t.Errorf("a failed build and a queued one wrote %v, want only %s", got[3:], brokenKey) } if comp, err := engcomp.Decode(state.kv[brokenKey].Value); err != nil || comp.Refused != `refused kind: package of language "go"` || len(comp.Resources) != 0 { t.Errorf("the failed build's key: %+v, %v", comp, err) } // What is kept: the five keys of a module written last, and one a want names; the rest is deleted. keep := New(b.env()) keep.pubKept = map[string]ComposedEntry{} var keys []string for i, ch := range "123456" { key, _ := engcomp.Key("n1", "lemurs", meshcatalogclient.SourcePrefix2+strings.Repeat(string(ch), 64)) keys = append(keys, key) keep.pubKept[key] = ComposedEntry{Revision: uint64(10 + i)} } keep.pubKept["n1.node"] = ComposedEntry{Revision: 1} if past := keep.pastKeeping(); !slices.Equal(past, []string{keys[0]}) { t.Errorf("past keeping: %v, want the oldest of six, %s", past, keys[0]) } keep.wants = map[string]meshwants.WantsByWantNodeModuleValue{"lemurs": {Fp: meshcatalogclient.SourcePrefix2 + strings.Repeat("1", 64)}} if past := keep.pastKeeping(); len(past) != 0 { t.Errorf("past keeping with a want naming the oldest: %v, want none", past) } }