mesh-host/internal/bootstrap/bus_adoption_test.go

package bootstrap

import (
	"context"
	"encoding/json"
	"os"
	"strings"
	"testing"
	"time"
)

// Defends novox/hq issue 590 at its broker step (ADR 0333 §3, ADR 0236): the adoption replaces the bus's
// image, and the mesh replaces its bus only as a planned step. A plain push was refused — "sending anchor
// would replace the bus … a planned step: `bus upgrade --why …`" — and genesis stopped at step 18.

func TestTheBrokerStepAdoptsTheBusThroughItsPlannedStep(t *testing.T) {
	catalogue := aCatalogue(t, map[string]string{BusModule: aBusModule(theModulesLayout)})
	rewritten, err := Rewrite(theRealBundle(t), held)
	if err != nil {
		t.Fatal(err)
	}
	bundleImage := containerNamed(rewritten.Declaration, "mesh-broker-nats").Image
	upgraded := false
	var credential []byte
	runtime := &asked{}
	agrees := aMeshThatAgrees(nil)
	runtime.answer = func(name string, args []string) (string, error) {
		joined := strings.Join(args, " ")
		switch {
		case name == "sh" && strings.Contains(joined, "mesh-nats-snapshot"):
			credential, _ = os.ReadFile(busSnapshotCredential) // as the program reads it
		case strings.Contains(joined, "bus upgrade --why"):
			upgraded = true
		case strings.Contains(joined, "inspect --format {{.Config.Image}} mesh-broker-nats"):
			if upgraded {
				return theModulesImage + "\n", nil
			}
			return bundleImage + "\n", nil
		}
		return agrees(name, args)
	}
	_, fetch := aBusThatAdopts()
	o := installing(t, catalogue)
	o.CatalogSource = Source{Repository: "git://forge/mesh-catalog.git"}
	o.Wait = 5 * time.Second
	said, err := busStepWith(t, o, runtime, fetch)
	if err != nil {
		t.Fatal(err)
	}

	snapshot, upgrade := -1, -1
	for i, c := range runtime.commands {
		switch {
		case strings.HasPrefix(c, "sh -c") && strings.Contains(c, "mesh-nats-snapshot"):
			snapshot = i
			for _, want := range []string{"127.0.0.1:5000/nats/server@sha256:" + strings.Repeat("4", 64),
				"snapshot --server tls://127.0.0.1:5671 --credential - < " + busSnapshotCredential,
				"> " + busSnapshotFile + ".partial && mv " + busSnapshotFile + ".partial " + busSnapshotFile} {
				if !strings.Contains(c, want) {
					t.Errorf("the snapshot is not taken with %q: %s", want, c)
				}
			}
		case strings.Contains(c, "/mesh-controller bus upgrade"):
			upgrade = i
			if want := "bus upgrade --why " + adoptionWhy + " --reversible --snapshot-taken " + busSnapshotFile; !strings.HasSuffix(c, want) {
				t.Errorf("the planned step is asked as %q; want it to end %q", c, want)
			}
		case strings.HasSuffix(c, "/mesh-controller push anchor"):
			t.Errorf("the bus's module was sent by a plain push, which the mesh refuses for the bus: %v", runtime.commands)
		}
	}
	if snapshot < 0 || upgrade < 0 || snapshot > upgrade {
		t.Fatalf("the streams were not snapshotted before the bus's planned step (snapshot %d, upgrade %d): %v",
			snapshot, upgrade, runtime.commands)
	}
	var cred map[string]string
	if err := json.Unmarshal(credential, &cred); err != nil {
		t.Fatalf("the snapshot program was given no credential: %v", err)
	}
	fingerprint, _ := certificateFingerprint(busCertificateFile)
	if cred["user"] != "controller" || cred["password"] != "BUS-PW" || cred["fingerprint"] != fingerprint ||
		cred["url"] != "tls://127.0.0.1:5671" {
		t.Errorf("the snapshot's credential is %v", cred)
	}
	if _, err := os.Stat(busSnapshotCredential); !os.IsNotExist(err) {
		t.Error("the snapshot's credential was left on the machine")
	}
	text := strings.Join(said, "\n")
	for _, want := range []string{"snapshotted       every stream", "restarting        mesh-broker-nats — once",
		"bus answers       mesh-broker-nats", "bus step done     last step 1", "adopted           mesh-broker-nats"} {
		if !strings.Contains(text, want) {
			t.Errorf("the step did not say %q:\n%s", want, text)
		}
	}
}

// A planned step the controller ends failed is the broker step failing, with the controller's words.
func TestABusStepThatEndsFailedIsRefused(t *testing.T) {
	aGenesisBusOnDisk(t)
	runtime := &asked{answer: func(name string, args []string) (string, error) {
		return "last step 1: not known → 53e51864 on anchor by installer (x): failed at 2026-10-13 11:59; snapshot s\n" +
			"  stream KV_x has no messages\n", nil
	}}
	control := controlPlane{container: "mesh-controller", run: runtime.run, timeout: time.Second}
	err := waitForTheBusStep(context.Background(), Options{}, control, quietly)
	if err == nil || !strings.Contains(err.Error(), "failed at") || !strings.Contains(err.Error(), "KV_x") {
		t.Fatalf("a failed bus step was not refused with its words: %v", err)
	}
}

// The bundle's bus runs the release the module's server is, pinned by the same digest, so the adoption changes no
// server release and is reversible as the step says (ADR 0333 §3). The module takes the server as an upstream
// artifact, its image reference pinned by digest (issue 1641); it no longer builds an image on a NATS_BASE argument,
// which this test read until issue 1875.
func TestTheBundlesBusRunsTheReleaseTheModulesImageIsBuiltOn(t *testing.T) {
	var m struct {
		Build struct {
			Artifacts []struct{ Name, Kind, From string } `json:"artifacts"`
		} `json:"build"`
	}
	if err := json.Unmarshal(theCataloguesModule(t, BusModule), &m); err != nil {
		t.Fatal(err)
	}
	from := ""
	for _, a := range m.Build.Artifacts {
		if a.Name == "server" && a.Kind == "upstream" {
			from = a.From
		}
	}
	name, digest, pinned := strings.Cut(from, "@")
	if !pinned || !strings.HasPrefix(digest, "sha256:") {
		t.Fatalf("the nats module's server is no upstream image pinned by digest: %q", from)
	}
	repository, _, _ := strings.Cut(name, ":")
	rewritten, err := Rewrite(theRealBundle(t), held)
	if err != nil {
		t.Fatal(err)
	}
	if got, want := containerNamed(rewritten.Declaration, "mesh-broker-nats").Image, repository+"@"+digest; got != want {
		t.Errorf("the bundle raises the bus on %s; the nats module's server is %s (%s)", got, want, from)
	}
}