mesh-delivery/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go

// Command mesh-delivery-builder is delivery's builder (novox/hq ADR 0453 clause 4, ADR 0478, issue 1607): it builds every
// new-form fingerprint the indexer writes, dark, into its own build keys (builds.<module>.<fingerprint>), through the build
// modules' fetch and build tools. Nothing reads its keys to deliver until release R4e. It serves no request subject and
// holds no timer: every reaction is a watch or a bus event (ADR 0437 §8).
package main

import (
	"context"
	"crypto/sha256"
	"crypto/tls"
	"crypto/x509"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
	"io"
	"log"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/process"

	"mesh-delivery-builder/internal/build"
)

const (
	// defaultStepBound is how long one ask's step may run once a build node took it (a fetch downloads the Go modules, a
	// build compiles): from its start, never from its put, so an ask waiting in its stream for a free slot is not failed
	// (buildrun.Asker.Bound; ADR 0525 §3, §6).
	defaultStepBound = 30 * time.Minute
)

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	if err := run(ctx, os.Getenv, os.Stderr); err != nil {
		fmt.Fprintln(os.Stderr, "mesh-delivery-builder:", err)
		os.Exit(1)
	}
}

// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
// fingerprint to pin, the principal and its password (as the indexer reads it; issue 1242 moves this into the SDK).
type credential struct {
	URL         string `json:"url"`
	Fingerprint string `json:"fingerprint"`
	Node        string `json:"node"`
	Module      string `json:"module"`
	User        string `json:"user"`
	Password    string `json:"password"`
}

func readCredential(path string) (credential, error) {
	raw, err := os.ReadFile(path)
	if err != nil {
		return credential{}, err
	}
	var c credential
	if err := json.Unmarshal(raw, &c); err != nil {
		return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
	}
	if c.URL == "" || c.User == "" {
		return credential{}, errors.New("the broker credential names no bus address or user")
	}
	return c, nil
}

// pinned accepts exactly the certificate with this SHA-256 and no other.
func pinned(want string) *tls.Config {
	return &tls.Config{
		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
		MinVersion:         tls.VersionTLS12,
		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
			if len(rawCerts) == 0 {
				return errors.New("the bus presented no certificate")
			}
			sum := sha256.Sum256(rawCerts[0])
			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
				return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
			}
			return nil
		},
	}
}

// run is the program: it returns when ctx ends or the bus cannot be used.
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
	logger := log.New(logOut, "mesh-delivery-builder: ", log.LstdFlags)
	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
	if path == "" {
		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to hear the bus with")
	}
	cred, err := readCredential(path)
	if err != nil {
		return err
	}
	opts := []nats.Option{
		nats.Name(cred.Node + "." + cred.Module),
		nats.UserInfo(cred.User, cred.Password),
		nats.CustomInboxPrefix("_INBOX." + cred.User),
		nats.MaxReconnects(-1),
		nats.RetryOnFailedConnect(true),
	}
	if strings.TrimSpace(cred.Fingerprint) != "" {
		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
	}
	conn, err := nats.Connect(cred.URL, opts...)
	if err != nil {
		return fmt.Errorf("connecting to the bus: %w", err)
	}
	defer conn.Close()
	js, err := jetstream.New(conn)
	if err != nil {
		return err
	}
	bucket := func(name string) (jetstream.KeyValue, error) {
		kv, err := js.KeyValue(ctx, name)
		if err != nil {
			return nil, fmt.Errorf("the bucket %s: %w", name, err)
		}
		return kv, nil
	}
	builds, err := bucket(delivery.BuilderBuildsBucket)
	if err != nil {
		return err
	}
	modules, err := bucket(delivery.ModulesBucket)
	if err != nil {
		return err
	}
	manifests, err := bucket(delivery.ManifestsBucket)
	if err != nil {
		return err
	}
	candidates, err := bucket(delivery.CandidatesBucket)
	if err != nil {
		return err
	}
	bound := defaultStepBound
	if v := strings.TrimSpace(getenv("MESH_BUILDER_STEP_BOUND")); v != "" {
		if bound, err = time.ParseDuration(v); err != nil {
			return fmt.Errorf("MESH_BUILDER_STEP_BOUND: %w", err)
		}
	}
	tools := &buildrun.ToolchainSet{}
	// Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot
	// (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity.
	asks := buildrun.JetStreamAsks{JS: js}
	// The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs
	// (the seat verb held, novox/hq issue 1937) before it leaves the key done.
	account, err := process.ReadAccount(path)
	if err != nil {
		return err
	}
	calls, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
	if err != nil {
		return err
	}
	defer calls.Close()
	b := &build.Builder{
		Claimer:    build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
		Asks:       buildrun.Asker{Publisher: asks, Waiter: asks, Bound: bound},
		Held:       heldOnTheBus{js},
		Toolchains: tools,
		Manifests:  manifestReader{manifests},
		Logf:       logger.Printf,
		Stored:     storedVia(calls.Call),
	}
	// **Said once every build module answered a reference, taken away when stopped** (issue 1886, ADR 0522): the
	// controller delivers from this builder's keys only while an instance says it writes references. A bucket not there
	// yet (the manifest's state not made) is said, and the controller's plan waits for the keys, named (no build-agent
	// is asked since R4g, ADR 0524 §7).
	switch kv, err := js.KeyValue(ctx, instancesBucket); {
	case errors.Is(err, jetstream.ErrBucketNotFound):
		logger.Printf("the bucket %s does not exist yet: the controller is not told that this builder writes references until it starts again", instancesBucket)
	case err != nil:
		return fmt.Errorf("the bucket %s: %w", instancesBucket, err)
	default:
		says := newSayer(ctx, kv, cred.Node, []string{buildrun.GoBuildModule, buildrun.NodeBuildModule, build.ImageBuildModule}, logger.Printf)
		b.Answered = says.Answered
		defer func() {
			stopping, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
			defer cancel()
			if err := says.unsay(stopping); err != nil {
				logger.Printf("%v", err)
			}
		}()
	}
	// The toolchains first, so a build that starts from the first modules value finds the nodes.
	watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger)
	if err := watch.start(); err != nil {
		return err
	}
	// A build module says each time it wrote its toolchains key: its stream, and the builder's durable consumer on it, are
	// made by the controller. Where the stream is not there yet the builder says so and is started again with the build
	// module.
	for _, module := range []string{buildrun.GoBuildModule, buildrun.NodeBuildModule} {
		stream := toolchainStreams[module]
		cons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module)
		if err != nil {
			logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the builder starts again: %v", cred.Node, cred.Module, stream, module, err)
			continue
		}
		cc, err := cons.Consume(func(m jetstream.Msg) { watch.Reported(m) },
			jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
		if err != nil {
			return err
		}
		defer cc.Stop()
	}
	// The controller says artifacts-gone when its take-in found a key's blob gone (issue 1937): the builder's durable
	// consumer on EVENTS, made by the controller from the manifest's consumes. Where it is not there yet the builder says
	// so; it still asks held at its start and at every module value.
	switch cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module); {
	case err != nil:
		logger.Printf("the consumer %s_%s on %s does not exist yet, so %s is not heard until the builder starts again: %v",
			cred.Node, cred.Module, eventsStream, delivery.ArtifactsGoneEvent, err)
	default:
		cc, err := cons.Consume(func(m jetstream.Msg) { heardGone(ctx, b, m, logger.Printf) },
			jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
		if err != nil {
			return err
		}
		defer cc.Stop()
	}
	each := func(e jetstream.KeyValueEntry) {
		if opOf(e) {
			b.Module(ctx, e.Key(), e.Value())
		} else {
			b.Forget(e.Key())
		}
	}
	w, err := settle(ctx, modules, each)
	if err != nil {
		return err
	}
	// A key left building for a fingerprint its module no longer has is closed (issue 1660); the modules are all known now.
	if err := b.Sweep(ctx); err != nil {
		return err
	}
	// The fingerprints the indexer gives the candidates' modules are built too (issue 1684), once the trunk's modules are known.
	eachCandidate := func(e jetstream.KeyValueEntry) {
		if opOf(e) {
			b.Candidate(ctx, e.Key(), e.Value())
		} else {
			b.ForgetCandidate(e.Key())
		}
	}
	cw, err := settle(ctx, candidates, eachCandidate)
	if err != nil {
		return err
	}
	logger.Printf("started on %s", cred.Node)
	errs := make(chan error, 2)
	go func() { errs <- follow(ctx, w, each) }()
	go func() { errs <- follow(ctx, cw, eachCandidate) }()
	select {
	case <-ctx.Done():
		b.Wait()
		return nil
	case err := <-errs:
		return err
	case err := <-watch.errs:
		return err
	}
}