From a096abac92653621faa28cc39077c67330187c3d Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 09:06:19 +0100 Subject: [PATCH] =?UTF-8?q?The=20node=20agent's=20applies=20run=20through?= =?UTF-8?q?=20a=20runner=20its=20options=20name,=20and=20one=20integration?= =?UTF-8?q?=20test=20proves=20an=20unwarranted=20bus=20want=20keeps=20the?= =?UTF-8?q?=20server=20and=20a=20warranted=20one=20replaces=20it=20(ADR=20?= =?UTF-8?q?0594=20=C2=A72-=C2=A74,=20=C2=A78)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-host/main.go | 15 +- cmd/mesh-host/nodeagent.go | 2 +- cmd/mesh-host/nodeagent_integration_test.go | 216 ++++++++++++++++++++ 3 files changed, 230 insertions(+), 3 deletions(-) diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index b352d48..bff6c87 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -156,6 +156,17 @@ type options struct { file string // out is where what a command says goes. Stdout, and a buffer under test. out io.Writer + // runner is what the node agent's applies run commands through (issue 774, ADR 0594): nil is the machine's own, + // apply.ExecRunner; a test gives one that records what an apply asked of the container runtime. + runner apply.Runner +} + +// run is the runner the node agent's applies use: the options' own, else the machine's. +func (o options) run() apply.Runner { + if o.runner != nil { + return o.runner + } + return apply.ExecRunner } // parseArgs takes the subcommand first, then its flags. @@ -1637,11 +1648,11 @@ func applyModuleHeld(ctx context.Context, opts options, declared *declaration.De func(known store.State) (apply.Report, store.State, error) { if scope == nil { return apply.ApplyMindingWindows(ctx, built, declared, known, store.OriginDeclared, - apply.ExecRunner, announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)), + opts.run(), announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)), apply.WindowsIn(filepath.Dir(opts.state))) } return apply.ApplyModule(ctx, built, declared, *scope, known, store.OriginDeclared, - apply.ExecRunner, announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)), + opts.run(), announceOr(say), sealOpener(opts.state), apply.KeepIn(filepath.Dir(opts.state)), apply.WindowsIn(filepath.Dir(opts.state))) }, announceOr(say)) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 08cbecb..09e9b31 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -508,7 +508,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus // What this engine applied, kept on the machine and read at the start, so what runs is known before the node // composes its own (issue 774). Applied: appliedStoreFor(opts.state), - Inactive: inactiveIn(&liveness.Exec{Run: apply.ExecRunner, System: builtFor}, heldBy(apply.WindowsIn(filepath.Dir(opts.state)))), + Inactive: inactiveIn(&liveness.Exec{Run: liveness.Runner(opts.run()), System: builtFor}, heldBy(apply.WindowsIn(filepath.Dir(opts.state)))), Revisions: func(ctx context.Context, key string) ([]agent.Update, error) { got, err := session.KVHistory(ctx, composedBucket, key) if err != nil { diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 0c9f677..a611c86 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -6,6 +6,7 @@ import ( "context" "crypto/ed25519" "encoding/json" + "errors" "os" "path/filepath" "strings" @@ -14,13 +15,17 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/declaration" "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/link" "github.com/novox/mesh-host/internal/nodegrant" "github.com/novox/mesh-host/internal/selfcompose" + "github.com/novox/mesh-host/internal/store" + "github.com/novox/mesh-host/internal/system" ) // The node agent whole, against a real bus with JetStream started in the test: the wants bucket, the node's composition @@ -619,3 +624,214 @@ func mustComposedKey(t *testing.T, module, fp string) string { } return key } + +// recordingRuntime is a container runtime that keeps its containers in memory and records every command an apply asks +// of it (issue 774): what the bus server's container met is read from it, never from a real runtime. +type recordingRuntime struct { + mu sync.Mutex + containers map[string]string // name: the spec label it was made with + asked []string +} + +func (r *recordingRuntime) run(_ context.Context, name string, args ...string) (string, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.asked = append(r.asked, name+" "+strings.Join(args, " ")) + if name != "docker" || len(args) == 0 { + return "", nil + } + switch args[0] { + case "info": + return "27.0\n", nil + case "container": + if len(args) > 1 && args[1] == "inspect" { + spec, ok := r.containers[args[len(args)-1]] + if !ok { + return "", errors.New("no such container") + } + return "true\t" + spec + "\n", nil + } + case "image": + if len(args) > 1 && args[1] == "inspect" { + return "2026-01-01T00:00:00Z\n", nil + } + case "rm": + delete(r.containers, args[len(args)-1]) + case "run", "create": + var made, spec string + for i, a := range args { + if a == "--name" && i+1 < len(args) { + made = args[i+1] + } + if a == "--label" && i+1 < len(args) && strings.HasPrefix(args[i+1], "mesh-host.spec=") { + spec = strings.TrimPrefix(args[i+1], "mesh-host.spec=") + } + } + r.containers[made] = spec + return "made\n", nil + } + return "", nil +} + +// moved is each command that removed, made, stopped or restarted the container named. +func (r *recordingRuntime) moved(container string) []string { + r.mu.Lock() + defer r.mu.Unlock() + var out []string + for _, a := range r.asked { + for _, verb := range []string{"docker rm ", "docker run ", "docker create ", "docker stop ", "docker restart ", "docker kill ", + "docker rename ", "docker start ", "docker container rm ", "docker container rename ", "docker container stop ", + "docker container restart ", "docker container start ", "docker container kill "} { + if strings.HasPrefix(a, verb) && strings.Contains(a, container) { + out = append(out, a) + } + } + } + return out +} + +// Issue 774, ADR 0594 §2-§4: the bus server the node's own composition made while the bus module was folded is taken +// over by the bus module's first want, at a new-form fingerprint and without the warrant's mark, and left as it runs: +// the node agent started whole against a real bus composes the bus module, writes its file and records the server as +// the bus module's, asks the container runtime to remove, make, stop or restart nothing, and its have names the +// fingerprint applied beside the old-form build the server still runs; its own key says it keeps the bus server. The +// operator's warranted want at the same fingerprint then replaces the server once, and the have says it runs that +// fingerprint. +func TestNatsTheBusServerIsKeptUntilAWarrantedWantReplacesIt(t *testing.T) { + b := newItBed(t) + runtime := &recordingRuntime{containers: map[string]string{}} + b.opts.runner = runtime.run + const server = "mesh-broker-it" + conf := filepath.Join(b.dir, "nats.conf") + image := "registry.example/nats@sha256:" + strings.Repeat("a", 64) + + // What the node's own composition made of the folded bus: the server and its file, recorded as the node's own. + built, err := system.For(builtFor) + if err != nil { + t.Fatal(err) + } + folded, err := declaration.Parse([]byte(`{"declaration":1,"resources":[` + + `{"id":"nats.server-conf","type":"file","path":"` + conf + `","content":"listen: old\n"},` + + `{"id":"nats.server","type":"container","name":"` + server + `","image":"` + image + `","env":{"FOLDED":"1"}}]}`)) + if err != nil { + t.Fatal(err) + } + _, known, err := apply.ApplyModule(context.Background(), built, folded, apply.Scope{Module: store.NodeModule}, store.State{}, + store.OriginDeclared, runtime.run, nil, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + if err := store.Save(b.opts.state, known); err != nil { + t.Fatal(err) + } + if moved := runtime.moved(server); len(moved) != 1 { + t.Fatalf("the folded server was not made once: %v", moved) + } + made := runtime.containers[server] + runtime.asked = nil + + kvOf := func(bucket string) nats.KeyValue { + kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 5}) + if err != nil { + t.Fatal(err) + } + return kv + } + placement, modules, manifests := kvOf(delivery.PlacementBucket), kvOf(delivery.ModulesBucket), kvOf(delivery.ManifestsBucket) + kvOf(selfcompose.ResolvedBucket) + kvOf(delivery.BuilderBuildsBucket) + kvOf(delivery.PlacementCandidatesBucket) + kvOf(link.PackagesBucket) + man := `{"module":"nats","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}],"resources":[` + + `{"id":"server-conf","type":"file","path":"` + conf + `","mode":"0644","content":"listen: new\n"},` + + `{"id":"server","type":"container","name":"` + server + `","image":"` + image + `"}]}` + digest := delivery.ManifestDigest([]byte(man)) + mraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "nats", JSON: man}) + mkey, _ := delivery.ManifestsKey(digest) + fp := fpIt("e") + modv, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/mesh-catalog", Commit: strings.Repeat("3", 40)}) + rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: itNode, + Capabilities: []string{"service-manager", "package-manager", "container-runtime"}}}) + hkey, _ := delivery.HaveNodeKey(itNode) + places, _ := delivery.PlacementValues(delivery.Placement{Node: itNode, Modules: []delivery.PlacedModule{{Module: "nats"}}}) + // The have the engine before the keeper wrote of the folded bus: the old-form build, no fingerprint. + const took = "build-1792674595610176744" + natsHave, _ := delivery.HaveKey(itNode, "nats") + oldHave, _ := delivery.EncodeHave(delivery.Have{Build: took, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + for _, w := range []struct { + kv nats.KeyValue + key string + value []byte + }{{manifests, mkey, mraw}, {modules, "nats", modv}, {b.haves, hkey, rec}, {b.haves, natsHave, oldHave}} { + if _, err := w.kv.Put(w.key, w.value); err != nil { + t.Fatal(err) + } + } + busWant := func(warrant string) { + raw, err := delivery.EncodeWant(delivery.Want{FP: fp, Self: true, Warrant: warrant}) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.WantKey(itNode, "nats") + if _, err := b.wants.Put(key, raw); err != nil { + t.Fatal(err) + } + } + busWant("") + stop := b.start() + defer stop() + for key, value := range places { + if _, err := placement.Put(key, value); err != nil { + t.Fatal(err) + } + } + + within(t, 10*time.Second, "the bus module's want was not applied", func() bool { + h := b.have("nats") + got, _ := os.ReadFile(conf) + return h.FP == fp && h.Phase == delivery.PhaseRunning && string(got) == "listen: new\n" + }) + if h := b.have("nats"); h.Server != took { + t.Errorf("the bus's have says its server runs %q, not the build it took over", h.Server) + } + if moved := runtime.moved(server); len(moved) != 0 || runtime.containers[server] != made { + t.Fatalf("an unwarranted want moved the bus server: %v", moved) + } + state, err := store.Load(b.opts.state) + if err != nil { + t.Fatal(err) + } + if rec, ok := state.Find("nats.server"); !ok || rec.Module != "nats" { + t.Errorf("the server is recorded %+v, not as the bus module's", rec) + } + // Said with the node's facts, as every start of its agent says them. + b.session.SayFacts(context.Background(), link.Report{Profile: map[string]any{"capabilities": []any{ + map[string]any{"name": "container-runtime", "present": true}}}}) + within(t, 5*time.Second, "the node's key does not say it keeps the bus server", func() bool { + e, err := b.haves.Get(hkey) + if err != nil { + return false + } + n, err := delivery.DecodeHaveNode(e.Value()) + return err == nil && len(n.Keeps) == 1 && n.Keeps[0] == delivery.KeepsBusServer + }) + + // The operator's warrant at the fingerprint applied: the server is replaced once. + busWant(delivery.WarrantOperator) + within(t, 10*time.Second, "the warranted want did not replace the bus server", func() bool { + return b.have("nats").Server == fp + }) + var runs int + for _, a := range runtime.moved(server) { + if strings.HasPrefix(a, "docker run ") && strings.Contains(a, " --name "+server+" ") { + runs++ + } + } + if runs != 1 || runtime.containers[server] == made { + t.Errorf("the warranted want made the server %d times, its spec still the folded one %v: %v", runs, + runtime.containers[server] == made, runtime.moved(server)) + } + if h := b.have("nats"); h.FP != fp || h.Phase != delivery.PhaseRunning { + t.Errorf("the bus's have after the warrant: %+v", h) + } +}