The node agent's applies run through a runner its options name, and one integration test proves an unwarranted bus want keeps the server and a warranted one replaces it (ADR 0594 §2-§4, §8)

2026-11-03 09:06:19 +01:00
parent 052004d1cf
commit a096abac92
3 changed files with 230 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,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)
+	}
+}