From 23cbf9294a3bc1ef09a776cfb38de5980b4db270 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 11:07:16 +0100 Subject: [PATCH] =?UTF-8?q?The=20node=20agent's=20applies=20run=20through?= =?UTF-8?q?=20a=20runner=20its=20options=20name,=20so=20one=20integration?= =?UTF-8?q?=20test=20proves=20the=20bus=20module's=20first=20want=20takes?= =?UTF-8?q?=20the=20bus=20server=20over=20from=20the=20node's=20own=20comp?= =?UTF-8?q?osition=20without=20moving=20it,=20and=20a=20build=20that=20cha?= =?UTF-8?q?nges=20it=20replaces=20it=20once=20(ADR=200599=20=C2=A74,=20?= =?UTF-8?q?=C2=A76)?= 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 | 221 ++++++++++++++++++++ 3 files changed, 235 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 8001ee3..eb9659a 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..4243082 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,219 @@ 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 "rename": + if spec, ok := r.containers[args[1]]; ok { + delete(r.containers, args[1]) + r.containers[args[2]] = spec + } + 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, renamed 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{"rm ", "run ", "create ", "stop ", "restart ", "kill ", "rename ", "start "} { + for _, prefix := range []string{"docker ", "docker container "} { + if strings.HasPrefix(a, prefix+verb) && strings.Contains(a, container) { + out = append(out, a) + } + } + } + } + return out +} + +// ADR 0599 §6 and §4 (issue 774): 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, the node agent started whole against a real bus: it composes the bus module, writes +// its changed file and records the server as the bus module's, and asks the container runtime to remove, make, stop, +// rename or restart nothing, since the build does not change the server. A later want whose build changes the server +// replaces it once, as any module's, once the want carries the yes (§2: the assigner gives it from the release's Approve). +func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(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 + `","mode":"0644","content":"listen: old\n"},` + + `{"id":"nats.server","type":"container","name":"` + server + `","image":"` + image + `"}]}`)) + 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.mu.Lock() + runtime.asked = nil + runtime.mu.Unlock() + + 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) + builds := kvOf(delivery.BuilderBuildsBucket) + kvOf(delivery.PlacementCandidatesBucket) + kvOf(link.PackagesBucket) + // The bus module at a build, done: its manifest written and the indexer's module naming it at fp. + publish := func(fp, env string) { + 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 + `"` + env + `}]}` + digest := delivery.ManifestDigest([]byte(man)) + mraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "nats", JSON: man}) + mkey, _ := delivery.ManifestsKey(digest) + modv, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/mesh-catalog", Commit: strings.Repeat("3", 40)}) + if _, err := manifests.Put(mkey, mraw); err != nil { + t.Fatal(err) + } + bkey, _ := delivery.BuilderBuildKey("nats", fp) + done, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: digest}) + if _, err := builds.Put(bkey, done); err != nil { + t.Fatal(err) + } + if _, err := modules.Put("nats", modv); err != nil { + t.Fatal(err) + } + } + 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 wrote of the folded bus: the old-form build, no fingerprint. + natsHave, _ := delivery.HaveKey(itNode, "nats") + oldHave, _ := delivery.EncodeHave(delivery.Have{Build: "build-1792674595610176744", Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + for key, value := range map[string][]byte{hkey: rec, natsHave: oldHave} { + if _, err := b.haves.Put(key, value); err != nil { + t.Fatal(err) + } + } + busWant := func(w delivery.Want) { + raw, err := delivery.EncodeWant(w) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.WantKey(itNode, "nats") + if _, err := b.wants.Put(key, raw); err != nil { + t.Fatal(err) + } + } + first := fpIt("e") + publish(first, "") + busWant(delivery.Want{FP: first, Self: true}) + 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 first want was not applied", func() bool { + h := b.have("nats") + got, _ := os.ReadFile(conf) + return h.FP == first && h.Phase == delivery.PhaseRunning && string(got) == "listen: new\n" + }) + if moved := runtime.moved(server); len(moved) != 0 || runtime.containers[server] != made { + t.Fatalf("the first want moved the bus server its build does not change: %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) + } + + // A build that changes the server: replaced once, as any module's, with the yes the assigner writes from the release's + // Approve for a move the node judged to recreate it (ADR 0599 §2, §4). + next := fpIt("f") + publish(next, `,"env":{"CHANGED":"1"}`) + busWant(delivery.Want{FP: next, Self: true}) + within(t, 10*time.Second, "the build that changes the server was not applied", func() bool { + h := b.have("nats") + if h.Phase == delivery.PhaseWaitingYes && h.Waits != "" { + busWant(delivery.Want{FP: next, Self: true, Yes: h.Waits}) + } + return h.FP == next && h.Phase == delivery.PhaseRunning + }) + 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 build that changes the server made it %d times, its spec still the folded one %v: %v", runs, + runtime.containers[server] == made, runtime.moved(server)) + } +}