mesh-host/internal/bootstrap/bus_adoption.go

package bootstrap

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"encoding/pem"
	"errors"
	"fmt"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"

	"github.com/novox/mesh-host/internal/declaration"
)

// Adopting the bus is the mesh's planned step for replacing it (novox/hq ADR 0333 §3, ADR 0236).
//
// The adoption replaces the bus's image once: the module's built image takes the place of the upstream
// one the bundle raised, on the same streams, users and certificate. The controller refuses a push that
// would replace the bus — "a planned step: `bus upgrade --why …` snapshots its streams first and checks
// them after" — so genesis stopped at step 18 (issue 590). The installer takes that step: it snapshots the
// streams itself, because at genesis no backup holder exists to be asked, and hands the snapshot to
// `bus upgrade`, which sends the machine, opens `bus-maintenance`, and ends the step done only once the
// bus answers healthy again — every stream, every durable consumer, a round trip (H-bus). The installer
// waits for that end.

// adoptionWhy is what the step is recorded as, for the record of hand acts and the bus's step.
const adoptionWhy = "genesis adopts its bus as the nats module (ADR 0333)"

// Where the installer keeps the streams' snapshot before the adoption, and the credential the snapshot
// program reads while it runs. Beside genesis's own credentials; variables so a test writes elsewhere.
var (
	busSnapshotFile       = "/var/lib/mesh-genesis/bus-before-adoption.tar"
	busSnapshotCredential = "/var/lib/mesh-genesis/bus-snapshot.credential"
	busCertificateFile    = "/var/lib/mesh-broker-tls/tls.crt"
	busControllerSecret   = BusControllerFile
	// busStepWithin bounds the wait for the controller to say the step ended: its own bound (15 min,
	// mesh-controller busStepBound), and its doctor's next look after that.
	busStepWithin = 17 * time.Minute
)

// adoptTheBus sends the nats module to this machine through the bus's planned step. A bus already on the
// module's image is the module's: it is sent like any module, and nothing is replaced.
func adoptTheBus(ctx context.Context, o Options, control controlPlane, bus *declaration.Container,
	say func(string)) error {

	if running := runningImage(ctx, control, bus.Name); running != "" && running != bus.Image {
		say("  already adopted   " + bus.Name + " — it runs the module's image")
		_, err := pushNode(ctx, o, control, say)
		return err
	}
	image, err := theModulesBusImage(ctx, o, control)
	if err != nil {
		return err
	}
	if err := snapshotTheBus(ctx, o, control, image, say); err != nil {
		return err
	}
	say("  restarting        " + bus.Name + " — once, onto the " + BusModule + " module's image, as the bus's " +
		"planned step; its streams, users and certificate stay where they are (novox/hq ADR 0333 §3, ADR 0236)")
	if _, err := control.within(2*o.Wait).tell(ctx, "bus", "upgrade", "--why", adoptionWhy,
		"--reversible", "--snapshot-taken", busSnapshotFile); err != nil {
		return fmt.Errorf("the bus's planned step was refused, so the bus is not replaced: %w", err)
	}
	return nil
}

// meshBuiltImage is an image the mesh built, as it records it: `artifact-store://<module>/<artifact>@sha256:…`.
var meshBuiltImage = regexp.MustCompile(`artifact-store://([a-z0-9][a-z0-9-]*/[a-z0-9][a-z0-9-]*@sha256:[0-9a-f]{64})`)

// theModulesBusImage is the nats module's server image, as the build just made it, pulled from this
// mesh's registry: the image that carries the snapshot program.
func theModulesBusImage(ctx context.Context, o Options, control controlPlane) (string, error) {
	said, err := control.tell(ctx, "builds", BusModule, "-n", "1")
	if err != nil {
		return "", err
	}
	for _, m := range meshBuiltImage.FindAllStringSubmatch(said, -1) {
		if strings.HasPrefix(m[1], BusModule+"/") {
			return strings.TrimSuffix(o.Registry, "/") + "/" + m[1], nil
		}
	}
	return "", fmt.Errorf("the mesh says no image it built for the %s module, so its snapshot program cannot "+
		"be run before the bus is replaced:\n%s", BusModule, indent(strings.TrimSpace(said)))
}

// snapshotTheBus writes every stream of the running bus to busSnapshotFile with the module's own snapshot
// program (mesh-catalog modules/nats, ADR 0235), as the controller's account, pinned to the bus's
// certificate. The credential is written for the program's one run and removed after it.
func snapshotTheBus(ctx context.Context, o Options, control controlPlane, image string, say func(string)) error {
	password, err := readCredentialFile(busControllerSecret)
	if err != nil {
		return fmt.Errorf("the controller's bus credential is not at %s, so the streams cannot be snapshotted "+
			"before the bus is replaced: %w", busControllerSecret, err)
	}
	fingerprint, err := certificateFingerprint(busCertificateFile)
	if err != nil {
		return err
	}
	port := o.Ports.orDefaults().Bus
	server := "tls://127.0.0.1:" + strconv.Itoa(port)
	credential, err := json.Marshal(map[string]string{"url": server, "fingerprint": fingerprint,
		"user": busControllerUser, "password": password})
	if err != nil {
		return err
	}
	if err := os.WriteFile(busSnapshotCredential, credential, 0o600); err != nil {
		return fmt.Errorf("the snapshot's credential could not be written: %w", err)
	}
	defer os.Remove(busSnapshotCredential)

	script := "docker run --rm -i --network host --entrypoint /usr/local/bin/mesh-nats-snapshot " + image +
		" snapshot --server " + server + " --credential - < " + busSnapshotCredential + " > " + busSnapshotFile +
		".partial && mv " + busSnapshotFile + ".partial " + busSnapshotFile
	running, cancel := context.WithTimeout(ctx, busStepWithin)
	defer cancel()
	if out, err := control.run(running, "sh", "-c", script); err != nil {
		_, _ = control.run(ctx, "rm", "-f", busSnapshotFile+".partial")
		return fmt.Errorf("the bus's streams could not be snapshotted, so the bus is not replaced: %w\n%s",
			err, indent(strings.TrimSpace(out)))
	}
	say("  snapshotted       every stream of the bus, at " + busSnapshotFile + " (its module's snapshot program)")
	return nil
}

// certificateFingerprint is the pin of the bus's certificate: sha256 of its DER, as every machine pins it.
func certificateFingerprint(path string) (string, error) {
	raw, err := os.ReadFile(path)
	if err != nil {
		return "", fmt.Errorf("the bus's certificate is not at %s, so it cannot be pinned for the snapshot: %w", path, err)
	}
	block, _ := pem.Decode(raw)
	if block == nil {
		return "", fmt.Errorf("%s holds no PEM certificate", path)
	}
	sum := sha256.Sum256(block.Bytes)
	return "sha256:" + hex.EncodeToString(sum[:]), nil
}

// waitForTheBusStep waits until the controller says the bus's planned step ended: done once the bus
// answered healthy — every stream, every durable consumer, a round trip — or failed, with its snapshot as
// the way back. The step's own bound is the controller's; this waits a little past it.
func waitForTheBusStep(ctx context.Context, o Options, control controlPlane, say func(string)) error {
	deadline := time.Now().Add(busStepWithin)
	var last string
	for {
		said, err := control.tell(ctx, "bus")
		if err == nil {
			for _, line := range strings.Split(said, "\n") {
				if strings.HasPrefix(strings.TrimSpace(line), "last step ") {
					last = strings.TrimSpace(line)
				}
			}
			switch {
			case strings.Contains(last, ": done at "):
				say("  bus step done     " + last)
				return nil
			case strings.Contains(last, ": failed at "):
				return fmt.Errorf("the bus's planned step ended failed: %s\n%s", last, indent(strings.TrimSpace(said)))
			}
		}
		if !time.Now().Before(deadline) {
			break
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(answerEvery):
		}
	}
	if last == "" {
		return errors.New("the controller says no bus step after the adoption was sent; `mesh-controller bus` says how it stands")
	}
	return fmt.Errorf("the bus's planned step did not end within %s: %s", busStepWithin, last)
}