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"
	"fmt"
	"io"
	"log"
	"maps"
	"os"
	"os/signal"
	"slices"
	"strings"
	"sync"
	"syscall"
	"time"

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

	"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)
	}
}

// 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)
	cred, err := process.ReadBrokerAccount(getenv)
	if err != nil {
		return err
	}
	conn, js, err := cred.Dial()
	if err != nil {
		return err
	}
	defer conn.Close()
	builds, err := process.EssentialBucket(ctx, js, "builder", delivery.BuilderBuildsBucket, "claim or write a build key")
	if err != nil {
		return err
	}
	modules, err := process.EssentialBucket(ctx, js, "builder", delivery.ModulesBucket, "know any fingerprint to build")
	if err != nil {
		return err
	}
	manifests, err := process.EssentialBucket(ctx, js, "builder", delivery.ManifestsBucket, "read what a module builds")
	if err != nil {
		return err
	}
	// The candidates' fingerprints are built too (issue 1684): a builder that cannot read them builds the trunk's.
	candidates := process.OptionalBucket(ctx, js, delivery.CandidatesBucket, "no candidate's fingerprint is built until the builder starts again; the trunk's are", logger.Printf)
	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 asks the build seats gave up on (issue 734): heard from DEAD_LETTERS, each ends the wait for its step at once,
	// and the lane asks its part again (Builder.waitOnce) instead of waiting for the check bound.
	// Heard in the background, and again at a reconnect or an ask while it is not; whether it is heard is kept in the
	// state hearing, which builder_stuck says (dead_letters).
	dead := &buildrun.DeadLetters{}
	// The state may not be on the bus yet when a delivery started the builder (the grants part creates it, issue 1790): it
	// is bound when the bus says it was created, and the latest hearing is written then.
	hearingKV := process.AwaitBucket(ctx, js, conn, delivery.BuilderModule+"_"+buildrun.HearingState, "builder_stuck cannot say whether DEAD_LETTERS is heard", logger.Printf)
	hearing := buildrun.NewDeadLetterHearing(ctx, buildrun.JetStreamDeadLetters{JS: js}, delivery.AskLanguages, dead, logger.Printf,
		buildrun.SayHearing(ctx, hearingKV, logger.Printf))
	defer hearing.Stop()
	// Each ask's step is heard by its own watch and, at every step end of its build module and every reconnect, read
	// (buildrun.HearEnds): a watch that goes deaf loses no end (the lost check of 2026-10-28).
	var buildModules []string
	for _, l := range delivery.AskLanguages {
		if m, err := delivery.BuildModuleOf(l); err == nil {
			buildModules = append(buildModules, m)
		}
	}
	heard := buildrun.HearEnds(ctx, asks, asks, buildModules, logger.Printf)
	conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again(); heard.Nudge("") })
	// 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.
	calls, err := process.Connect(cred, 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: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound},
		Held:    heldOnTheBus{js},
		// An ask that can no longer be answered is asked again (issue 1945), read at the step bound: a context's bound.
		Gone:       heldOnTheBus{js},
		CheckAfter: bound,
		Toolchains: tools,
		Manifests:  manifestReader{manifests},
		Logf:       logger.Printf,
		Stored:     storedVia(calls.Call),
		Node:       cred.Node,
	}
	// A candidate's preview is merged onto the trunk head the indexer previewed it onto: source.main at the candidate's base
	// revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision
	// is not what a read grants): a revision heard after the candidate that names it builds that candidate then. Where it
	// cannot be read the trunk's keys are built and the candidates wait (issue 1945).
	errs := make(chan error, 6)
	trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: b.TrunkMoved, errs: errs, logf: logger.Printf}
	trunks.start()
	b.Trunk = trunks.head
	// The builder's log (ADR 0571 §1): each build's start and end in its module log `build-log`, read by the view, the
	// indexer and the assigner through their reads, and by mesh-catalog through the tool builder_builds_since.
	b.Log = func(e delivery.BuildLogEntry) {
		if err := appendLog(context.WithoutCancel(ctx), js, cred, e); err != nil {
			logger.Printf("%v", err)
		}
	}
	b.Said = func(e build.AskedAgain) {
		if err := emitAskedAgain(context.WithoutCancel(ctx), js, cred, e); err != nil {
			logger.Printf("%v", err)
		}
	}
	// **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 := process.OptionalBucket(ctx, js, instancesBucket, "the controller is not told that this builder writes references until it starts again", logger.Printf); {
	case kv == nil:
	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)
	// Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the
	// placement is settled first, so the toolchains settle through the gate already knowing it.
	gate := buildrun.NewToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf)
	watch.put = gate.Toolchain
	placementErrs, err := buildrun.FollowPlacement(ctx, js, gate, b.NodesChanged, logger.Printf)
	if err != nil {
		return err
	}
	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 := cred.EventConsumer(ctx, js, stream)
		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 := process.HearEvents(cons, func(m jetstream.Msg) { watch.Reported(m) }, logger.Printf)
		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 := cred.EventConsumer(ctx, js, eventsStream); {
	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 := process.HearEvents(cons, func(m jetstream.Msg) { heardGone(ctx, b, m, logger.Printf) }, logger.Printf)
		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())
		}
	}
	if candidates != nil {
		cw, err := settle(ctx, candidates, eachCandidate)
		if err != nil {
			process.NotWatched(logger.Printf, delivery.CandidatesBucket, err, "no candidate's fingerprint is built until the builder starts again")
		} else {
			go func() { errs <- follow(ctx, cw, eachCandidate) }()
		}
	}
	// A person asks a key's build again with builder_ask_again, which writes the reasks state (issue 1945): watched, as
	// every other input. The builder asks again by itself without it.
	// A state not on the bus yet at this start is watched once it appears (issue 1790).
	process.AwaitBucket(ctx, js, conn, build.ReAsksBucket, "builder_ask_again is not heard; an ask that can no longer be answered is still asked again by itself", logger.Printf).Ready(func(kv jetstream.KeyValue) {
		eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, kv, e, logger.Printf) }
		if rw, err := settle(ctx, kv, eachReAsk); err != nil {
			process.NotWatched(logger.Printf, build.ReAsksBucket, err, "builder_ask_again is not heard until the builder starts again")
		} else {
			go func() { errs <- follow(ctx, rw, eachReAsk) }()
		}
	})
	// A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the
	// builder alone may read the stream, so it answers there.
	process.AwaitBucket(ctx, js, conn, build.QueuesBucket, "builder_seat_queue is not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) {
		eachRead := func(e jetstream.KeyValueEntry) { queueRead(ctx, heldOnTheBus{js}, kv, e, logger.Printf) }
		if qw, err := settle(ctx, kv, eachRead); err != nil {
			process.NotWatched(logger.Printf, build.QueuesBucket, err, "builder_seat_queue is not answered until the builder starts again")
		} else {
			go func() { errs <- follow(ctx, qw, eachRead) }()
		}
	})
	logger.Printf("started on %s", cred.Node)
	go func() { errs <- follow(ctx, w, each) }()
	select {
	case <-ctx.Done():
		b.Wait()
		return nil
	case err := <-errs:
		return err
	case err := <-placementErrs:
		return err
	case err := <-watch.errs:
		return err
	}
}

// sourceHeads are the trunk heads of each revision of the gate's source.main the watch delivered, the newest
// delivery.SourceHistory of them: what a candidate's base revision names.
type sourceHeads struct {
	mu    sync.Mutex
	byRev map[uint64]map[string]string
}

// put takes one revision in and answers whether it is new.
func (s *sourceHeads) put(e jetstream.KeyValueEntry) bool {
	if !opOf(e) {
		return false
	}
	src, err := delivery.DecodeSource(e.Value())
	if err != nil {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.byRev == nil {
		s.byRev = map[uint64]map[string]string{}
	}
	if _, had := s.byRev[e.Revision()]; had {
		return false
	}
	s.byRev[e.Revision()] = src.Heads
	for len(s.byRev) > delivery.SourceHistory {
		delete(s.byRev, slices.Min(slices.Collect(maps.Keys(s.byRev))))
	}
	return true
}

// head is the head of repository in revision rev; an error when the revision is not heard (yet, or any more).
func (s *sourceHeads) head(_ context.Context, rev uint64, repository string) (string, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	heads, ok := s.byRev[rev]
	if !ok {
		return "", fmt.Errorf("source revision %d is not heard (it is built once it is)", rev)
	}
	for r, head := range heads {
		if delivery.SameRepository(r, repository) {
			return head, nil
		}
	}
	return "", nil
}