Raise genesis's bus where the nats module keeps it, so adopting it keeps it (hq issue 590, ADR 0333)

The bundle raised the bus as mesh-broker on docker volumes; the catalogue's nats module keeps it as mesh-broker-nats on directories of the machine, so the broker step refused rather than raise a second bus, and genesis stopped at step 18 of 24. The bundle now raises mesh-broker-nats with its configuration and user list in /var/lib/nats-module/conf, its streams in /var/lib/mesh-broker-nats and its certificate in /var/lib/mesh-broker-tls, published on the node's bus port. The broker step holds that layout to the module's manifest field by field before registering anything, says the one restart onto the module's image, and waits until the bus runs that image and its monitor answers. The installer's bridge writes the composed users at the module's bus-users path and signals mesh-broker-nats. The bundle's bus configuration and user list are seeds, so a re-run after the adoption does not put the bundle's one user back over the module's.
2026-10-13 02:09:03 +02:00
parent 7a41e12063
commit bc2a448512
9 changed files with 612 additions and 69 deletions
@@ -4,7 +4,9 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net/http"
 	"strings"
+	"time"

 	"github.com/novox/mesh-host/internal/declaration"
 )
@@ -165,13 +167,21 @@ const BusModule = "nats"
 // (ADR 0331 §1, §5). The installer's own placing of that list (enrol.go placeTheBusUsers) was the
 // bridge to here and is not run again.
 //
-// **What is compared is the container's name, not its image.** The store's module pins the upstream
-// postgres the foundation runs; the bus's module builds its server image (it carries the mesh's
-// snapshot tool), so no pinned reference can be equal to the foundation's. A module whose server is
-// not the foundation's container by name would raise a second bus beside the first, and is refused
-// by name before anything is registered.
+// **The bundle raises the bus where the module keeps it** (novox/hq ADR 0333 §1): the same container
+// name, the same streams directory, configuration directory, user list and certificate. That is checked
+// here against the module's own manifest before anything is registered, and a difference is refused
+// naming each one — under another name or on other paths the module would raise a second bus beside the
+// first, or an empty one in its place.
+//
+// **The image is the one thing that changes, once** (ADR 0333 §3). The store's module pins the upstream
+// postgres the foundation runs; the bus's module builds its server image (it carries the mesh's snapshot
+// tool), so no pinned reference can be equal to the foundation's. The applier recreates the container on
+// the module's image, on the same streams, users and certificate. This says so before the push and waits
+// after it until the container runs the module's image and the bus answers, so nothing after this step
+// dials a bus that is restarting.
 func InstallBroker(ctx context.Context, o Options, control controlPlane,
-	foundation *declaration.Declaration, say func(string)) error {
+	foundation *declaration.Declaration, fetch func(context.Context, string) (int, string, error),
+	say func(string)) error {

 	const module = BusModule
 	manifest, err := readManifest(o.Catalogue, module)
@@ -182,12 +192,23 @@ func InstallBroker(ctx context.Context, o Options, control controlPlane,
 	if err != nil {
 		return err
 	}
-	if err := busServerIsTheFoundations(manifest, bus, module); err != nil {
+	if err := busLayoutIsTheModules(manifest, foundation, module); err != nil {
 		return err
 	}
 	say("  adopting          " + bus.Name + " — the bus the foundation raised, its streams kept")
 	if err := installProvider(ctx, o, control, module, manifest, "(the server, adopting the foundation's bus)",
-		nil, say); err != nil {
+		func() error {
+			if running := runningImage(ctx, control, bus.Name); running != "" && running != bus.Image {
+				say("  already restarted " + bus.Name + " — it runs the module's image")
+				return nil
+			}
+			say("  restarting        " + bus.Name + " — once, onto the " + module + " module's image; its " +
+				"streams, users and certificate stay where they are (novox/hq ADR 0333 §3)")
+			return nil
+		}, say); err != nil {
+		return err
+	}
+	if err := waitForTheAdoptedBus(ctx, o, control, bus, fetch, say); err != nil {
 		return err
 	}
 	say("  adopted           " + bus.Name + " — the foundation's bus is now the " + module +
@@ -195,41 +216,190 @@ func InstallBroker(ctx context.Context, o Options, control controlPlane,
 	return nil
 }

-// busServerIsTheFoundations checks that the bus module's server container is the one the
-// foundation raised, by name, so the applier takes it over rather than raising a second bus.
-func busServerIsTheFoundations(manifest []byte, bus *declaration.Container, module string) error {
+// busMonitor is where the bus says it is healthy: its monitoring port, published on the machine's own
+// loopback by the bundle and by the module alike (both "127.0.0.1:8222:8222"). /healthz answers ok only
+// once the server is up and JetStream has its streams back.
+const busMonitor = "http://127.0.0.1:8222/healthz"
+
+// runningImage is the image the named container was created from, or "" when there is none.
+func runningImage(ctx context.Context, control controlPlane, name string) string {
+	asking, cancel := context.WithTimeout(ctx, control.timeout)
+	defer cancel()
+	out, err := control.run(asking, "docker", "inspect", "--format", "{{.Config.Image}}", name)
+	if err != nil {
+		return ""
+	}
+	return strings.TrimSpace(out)
+}
+
+// waitForTheAdoptedBus waits until the bus container runs an image other than the one the bundle raised
+// it from — the module's, after the applier's one recreate — and the bus answers on its monitor.
+//
+// Bounded by --wait, and asked at least once. A bus still on the bundle's image when it runs out is
+// refused, naming what it runs: the push was sent and the node has not applied it, and every step after
+// this one reaches the mesh through that bus.
+func waitForTheAdoptedBus(ctx context.Context, o Options, control controlPlane, bus *declaration.Container,
+	fetch func(context.Context, string) (int, string, error), say func(string)) error {
+
+	deadline := time.Now().Add(o.Wait)
+	var running, said string
+	for {
+		running = runningImage(ctx, control, bus.Name)
+		if running != "" && running != bus.Image {
+			code, body, err := fetch(ctx, busMonitor)
+			if err == nil && code == http.StatusOK {
+				say("  bus answers       " + bus.Name + " on " + running)
+				return nil
+			}
+			said = strings.TrimSpace(body)
+			if err != nil {
+				said = err.Error()
+			}
+		}
+		if !time.Now().Before(deadline) {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(answerEvery):
+		}
+	}
+	switch {
+	case running == "":
+		return fmt.Errorf("the bus container %s is not there after the %s module was pushed, so nothing "+
+			"after this step could reach the mesh. The node-engine's journal says why it did not raise it",
+			bus.Name, BusModule)
+	case running == bus.Image:
+		return fmt.Errorf("%s still runs the bundle's image %s %s after the %s module was pushed: the node "+
+			"has not applied it. Its streams are untouched; run this installer again once the node-engine "+
+			"applies (its journal says why it did not)", bus.Name, bus.Image, o.Wait, BusModule)
+	}
+	return fmt.Errorf("%s runs the %s module's image %s and its monitor (%s) did not answer ok within %s: %s",
+		bus.Name, BusModule, running, busMonitor, o.Wait, orUnknown(said))
+}
+
+// BusUsersID is the bundle's resource for the bus's first user list.
+const BusUsersID = "bus-accounts"
+
+// busLayout is where a bus lives: its container, the machine's directories it mounts as its streams,
+// its configuration and its certificate, and the file its user list is.
+type busLayout struct {
+	Container, Streams, Conf, Certificate, Users string
+}
+
+// Where the server reads each, inside its container: the same in the bundle and the module, since both
+// run the same server with the same configuration.
+const (
+	busStreamsInside     = "/data"
+	busConfInside        = "/etc/nats"
+	busCertificateInside = "/tls"
+)
+
+// foundationBusLayout reads the layout of the bus the bundle raises.
+func foundationBusLayout(foundation *declaration.Declaration) (busLayout, error) {
+	bus, err := brokerIn(foundation)
+	if err != nil {
+		return busLayout{}, err
+	}
+	at := mountedFrom(bus.Volumes)
+	l := busLayout{Container: bus.Name, Streams: at[busStreamsInside], Conf: at[busConfInside],
+		Certificate: at[busCertificateInside]}
+	for _, r := range foundation.Resources {
+		if f, ok := r.(*declaration.File); ok && r.Identity() == BusUsersID {
+			l.Users = f.Path
+		}
+	}
+	return l, nil
+}
+
+// moduleBusLayout reads the layout of the bus the module declares: its server container, the mounts of
+// that container with every `${access:<id>}` resolved to the access's path, and its `bus-users`.
+func moduleBusLayout(manifest []byte, module string) (busLayout, error) {
 	var m struct {
+		BusUsers  string `json:"bus-users"`
 		Resources []struct {
-			Type string `json:"type"`
-			Name string `json:"name"`
+			ID      string   `json:"id"`
+			Type    string   `json:"type"`
+			Name    string   `json:"name"`
+			Volumes []string `json:"volumes"`
 		} `json:"resources"`
+		Accesses []struct {
+			ID   string `json:"id"`
+			Path string `json:"path"`
+		} `json:"accesses"`
 	}
 	if err := json.Unmarshal(manifest, &m); err != nil {
-		return fmt.Errorf("the %s module's manifest is not readable: %w", module, err)
+		return busLayout{}, fmt.Errorf("the %s module's manifest is not readable: %w", module, err)
 	}
-	var names []string
-	for _, r := range m.Resources {
-		if r.Type != "container" {
-			continue
+	var containers []int
+	for i, r := range m.Resources {
+		if r.Type == "container" {
+			containers = append(containers, i)
 		}
-		if r.Name == bus.Name {
-			return nil
+	}
+	if len(containers) != 1 {
+		var names []string
+		for _, i := range containers {
+			names = append(names, m.Resources[i].Name)
 		}
-		names = append(names, r.Name)
+		return busLayout{}, fmt.Errorf("the %s module declares %s; the bus it adopts is one server, so it "+
+			"declares one container", module, describeNames(names))
+	}
+	server := m.Resources[containers[0]]
+	resolved := make([]string, 0, len(server.Volumes))
+	for _, v := range server.Volumes {
+		for _, a := range m.Accesses {
+			v = strings.ReplaceAll(v, "${access:"+a.ID+"}", a.Path)
+		}
+		resolved = append(resolved, v)
+	}
+	at := mountedFrom(resolved)
+	return busLayout{Container: server.Name, Streams: at[busStreamsInside], Conf: at[busConfInside],
+		Certificate: at[busCertificateInside], Users: m.BusUsers}, nil
+}
+
+// busLayoutIsTheModules checks that the bus the bundle raised is where the module keeps it, field by
+// field, so the applier takes it over rather than raising a second bus beside it or an empty one in its
+// place (novox/hq ADR 0333 §1). Every difference is named, with both sides.
+func busLayoutIsTheModules(manifest []byte, foundation *declaration.Declaration, module string) error {
+	theirs, err := moduleBusLayout(manifest, module)
+	if err != nil {
+		return err
+	}
+	ours, err := foundationBusLayout(foundation)
+	if err != nil {
+		return err
+	}
+	var differ []string
+	for _, f := range []struct{ what, module, bundle string }{
+		{"the server container", theirs.Container, ours.Container},
+		{"its streams (" + busStreamsInside + ")", theirs.Streams, ours.Streams},
+		{"its configuration (" + busConfInside + ")", theirs.Conf, ours.Conf},
+		{"its certificate (" + busCertificateInside + ")", theirs.Certificate, ours.Certificate},
+		{"its user list", theirs.Users, ours.Users},
+	} {
+		if f.module == "" || f.module != f.bundle {
+			differ = append(differ, fmt.Sprintf("%s: the module's is %q, the bundle's %q", f.what,
+				f.module, f.bundle))
+		}
+	}
+	if len(differ) == 0 {
+		return nil
 	}
 	return fmt.Errorf(
-		"the %s module's server is %s, and the foundation's bus is the container %q.\n"+
-			"The module adopts the foundation's bus in place, so its server has to carry the name the "+
-			"foundation raised; under another name it would raise a second bus beside the first, and "+
-			"the mesh would run two (novox/hq ADR 0331 §1). Nothing has been registered",
-		module, describeNames(names), bus.Name)
+		"the bus the foundation raised is not where the %s module keeps it:\n  - %s\n"+
+			"The module adopts the foundation's bus in place, so the bundle raises it with the module's "+
+			"name and paths; otherwise adopting it would raise a second bus beside the first, or an empty "+
+			"one on other paths (novox/hq ADR 0333 §1). Nothing has been registered",
+		module, strings.Join(differ, "\n  - "))
 }

 func describeNames(names []string) string {
 	if len(names) == 0 {
 		return "no container at all"
 	}
-	return "the container " + strings.Join(names, ", ")
+	return "the containers " + strings.Join(names, ", ")
 }

 // InstallVault installs the vault as a foundation module (novox/hq ADR 0085, amended). Nothing to