The node agent's applies run through a runner its options name, so one integration test proves the bus module's first want takes the bus server over from the node's own composition without moving it, and a build that changes it replaces it once (ADR 0599 §4, §6)

2026-11-03 13:28:44 +01:00
parent 03b7fcd7bb
commit 23cbf9294a
3 changed files with 235 additions and 3 deletions
@@ -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))

@@ -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 {
@@ -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))
+	}
+}