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:///@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) }