mesh-delivery/modules/mesh-delivery-builder/internal/build/run.go

package build

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"

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

// ToolchainEntries are the nodes' toolchain entries of a language (buildrun.ToolchainSet).
type ToolchainEntries interface {
	Entries(language string) []delivery.NodeToolchain
}

// toolPrefix is the start of each build module's tool names: <prefix>fetch and <prefix>build.
var toolPrefix = map[string]string{buildrun.GoBuildModule: "build_go_", buildrun.NodeBuildModule: "build_node_", ImageBuildModule: "build_image_"}

// Builder builds, dark, every new-form fingerprint the indexer writes (novox/hq ADR 0453 clause 4, ADR 0478): it is
// handed each modules value as the watch of the indexer's bucket sees it, and for each module with a fingerprint and an
// artifact to build claims the fingerprint's build key, builds it through the build modules on the nodes
// the toolchains choose, and writes the key done or failed. It holds no timer; a build starts from a modules value or a
// change of the toolchains.
type Builder struct {
	Claimer    Claimer
	Runner     buildrun.Runner
	Toolchains ToolchainEntries
	// Presence says which nodes are online (the watchman's); a node that is not is not chosen.
	Presence  interface{ Online(node string) bool }
	Manifests Manifests
	// Health says whether a node's build module is running and healthy (the node-engine's have); a node whose build module
	// is not is not chosen. Nil means every node's is.
	Health interface {
		Healthy(node, module string) bool
	}
	// buildrun.Places holds a step back while its node is at its build capacity; a Builder with none runs one step at a time on each node.
	Places *buildrun.Places
	// Nodes is how many build nodes build each fingerprint: two until release R4e (ADR 0478 clause 4).
	Nodes int
	// Limit is how many builds run at once; four when 0.
	Limit int
	Logf  func(format string, args ...any)

	once     sync.Once
	sem      chan struct{}
	wg       sync.WaitGroup
	mu       sync.Mutex
	known    map[string]delivery.Module    // by modules key
	inflight map[string]bool               // by build key
	again    map[string]delivery.Module    // by build key: the newest value that arrived while the key was being built
	plans    map[string]planned            // by manifest digest
	cands    map[string]delivery.Candidate // by candidates key (candidates.go)
	via      map[string]via                // by build key: a candidate's fetch (candidates.go)
	trunkFP  map[string]bool               // by build key: a trunk module value carries this fingerprint
	tried    map[string]bool               // by build key: looked at since this process started
	// epoch counts the changes of the nodes (online, healthy, a toolchain) seen since this process started; a done key
	// whose second-node build failed is held back until it moves (issue 1664).
	epoch    uint64
	secondAt map[string]uint64 // by build key: the epoch a second-node build of a one-node key failed at
	// published are the package versions `<name>@<version>` a done key of this bucket names as a package's reference: a key
	// waiting for one is built once it is here (ADR 0500 clause 6).
	published map[string]bool
}

type planned struct {
	plan Plan
	err  error
}

func (b *Builder) init() {
	b.once.Do(func() {
		limit := b.Limit
		if limit <= 0 {
			limit = 4
		}
		b.sem = make(chan struct{}, limit)
		b.cands, b.via, b.trunkFP = map[string]delivery.Candidate{}, map[string]via{}, map[string]bool{}
		b.known, b.inflight, b.again, b.plans, b.tried = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]planned{}, map[string]bool{}
		b.secondAt, b.published = map[string]uint64{}, map[string]bool{}
		if b.Places == nil {
			b.Places = &buildrun.Places{}
		}
	})
}

func (b *Builder) logf(format string, args ...any) {
	if b.Logf != nil {
		b.Logf(format, args...)
	}
}

// Wait returns when every build started so far has ended.
func (b *Builder) Wait() { b.wg.Wait() }

// Module takes one value of the indexer's modules bucket in (the key is the module's name) and builds it if it is
// to be built. A value that is not a module is said and left.
func (b *Builder) Module(ctx context.Context, key string, value []byte) {
	b.init()
	m, err := delivery.DecodeModule(value)
	if err != nil {
		b.logf("modules.%s holds what is not a module, left: %v", key, err)
		return
	}
	b.mu.Lock()
	b.known[key] = m
	if bk, err := delivery.BuilderBuildKey(key, m.FP2); err == nil && m.FP2 != "" {
		// A fingerprint the trunk carries is built from the trunk's commit, whoever asked for it first.
		b.trunkFP[bk] = true
		delete(b.via, bk)
	}
	b.mu.Unlock()
	b.consider(ctx, key, m)
}

// NodesChanged is Reconsider after a change of the nodes: one came online or went, a build module became healthy or
// stopped being, or a toolchain moved. A done key built on one node whose second-node build failed is tried again only
// after such a change, not at every modules value (issue 1664).
func (b *Builder) NodesChanged(ctx context.Context) {
	b.init()
	b.mu.Lock()
	b.epoch++
	b.mu.Unlock()
	b.Reconsider(ctx)
}

// Sweep reads, at start, every key of the bucket: it keeps the package versions done keys name as published (a key
// waiting for one is then looked at again, ADR 0500 clause 6), and closes every key left building for a fingerprint its
// module no longer has (issue 1660, ADR 0478 clause 6): the key can never be built, since nothing asks for it, and would
// stay building for ever. It fails the key naming who left it. A key of the fingerprint the module has now is not
// touched: the build takes it over. Call it once the modules bucket has been read, so every module is known.
func (b *Builder) Sweep(ctx context.Context) error {
	b.init()
	keys, err := b.Claimer.Store.Keys(ctx)
	if err != nil {
		return err
	}
	b.mu.Lock()
	current := map[string]bool{}
	for name, m := range b.known {
		if m.FP2 != "" {
			if k, err := delivery.BuilderBuildKey(name, m.FP2); err == nil {
				current[k] = true
			}
		}
	}
	b.mu.Unlock()
	newly := false
	for _, key := range keys {
		value, rev, found, err := b.Claimer.Store.Get(ctx, key)
		if err != nil || !found {
			continue
		}
		cur, err := delivery.DecodeBuilderBuild(value)
		if err != nil {
			continue
		}
		newly = b.notePublished(cur) || newly
		if current[key] || cur.State != delivery.BuildBuilding {
			continue
		}
		closed := delivery.BuilderBuild{State: delivery.BuildFailed, By: b.Claimer.By, Manifest: cur.Manifest, Step: cur.Step,
			Error: fmt.Sprintf("closed at start: left building by %s, and the module no longer has this fingerprint, so nothing will build it", cur.By)}
		data, err := delivery.EncodeBuilderBuild(closed)
		if err != nil {
			return err
		}
		if _, err := b.Claimer.Store.Update(ctx, key, data, rev); err != nil && !errors.Is(err, ErrConflict) {
			return fmt.Errorf("%s: %w", key, err)
		}
		b.logf("%s: closed, left building by %s for a fingerprint the module no longer has", key, cur.By)
	}
	if newly {
		// A key read before the scan may wait for a package the scan found published (ADR 0500 clause 6).
		b.Reconsider(ctx)
	}
	return nil
}

// notePublished keeps the package versions a done build names as references, and says whether one was new.
func (b *Builder) notePublished(v delivery.BuilderBuild) bool {
	if v.State != delivery.BuildDone {
		return false
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	newly := false
	for _, a := range v.Artifacts {
		if a.Kind == catalogue.ArtifactPackage && a.Reference != "" && !b.published[a.Reference] {
			b.published[a.Reference] = true
			newly = true
		}
	}
	return newly
}

func (b *Builder) isPublished(ref string) bool {
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.published[ref]
}

// Forget takes a module out (its key was deleted).
func (b *Builder) Forget(key string) {
	b.init()
	b.mu.Lock()
	delete(b.known, key)
	b.mu.Unlock()
}

// Reconsider looks at every module again: the toolchains changed, so a build that failed for them may build now.
func (b *Builder) Reconsider(ctx context.Context) {
	b.init()
	b.mu.Lock()
	known := make(map[string]delivery.Module, len(b.known))
	for k, m := range b.known {
		known[k] = m
	}
	b.mu.Unlock()
	for _, k := range sortedKeys(known) {
		b.consider(ctx, k, known[k])
	}
	b.mu.Lock()
	cands := make([]delivery.Candidate, 0, len(b.cands))
	for _, c := range b.cands {
		cands = append(cands, c)
	}
	b.mu.Unlock()
	sort.Slice(cands, func(i, j int) bool { return cands[i].Group < cands[j].Group })
	for _, c := range cands {
		b.considerCandidate(ctx, c)
	}
}

func sortedKeys(m map[string]delivery.Module) []string {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys
}

// planOf is the plan of the manifest with this digest, read once: a manifest by digest never changes. A manifest that
// cannot be read is not kept, so it is read again when the module is.
func (b *Builder) planOf(ctx context.Context, digest string) (Plan, error, bool) {
	b.mu.Lock()
	p, ok := b.plans[digest]
	b.mu.Unlock()
	if ok {
		return p.plan, p.err, true
	}
	m, err := b.Manifests.Manifest(ctx, digest)
	if err != nil {
		return Plan{}, fmt.Errorf("its manifest %s cannot be read: %w", digest, err), false
	}
	plan, err := PlanOf(m)
	b.mu.Lock()
	b.plans[digest] = planned{plan, err}
	b.mu.Unlock()
	return plan, err, true
}

func (b *Builder) consider(ctx context.Context, module string, m delivery.Module) {
	if m.FP2 == "" || m.Manifest == "" {
		return
	}
	key, err := delivery.BuilderBuildKey(module, m.FP2)
	if err != nil {
		b.logf("modules.%s: %v", module, err)
		return
	}
	b.mu.Lock()
	if b.inflight[key] {
		b.again[key] = m // looked at once more when the build ends, so a change is never lost
		b.mu.Unlock()
		return
	}
	b.inflight[key] = true
	b.mu.Unlock()
	b.wg.Add(1)
	go func() {
		defer b.wg.Done()
		b.sem <- struct{}{}
		defer func() { <-b.sem }()
		for {
			b.try(ctx, module, m)
			b.mu.Lock()
			next, more := b.again[key]
			delete(b.again, key)
			if !more || ctx.Err() != nil {
				delete(b.inflight, key)
				b.mu.Unlock()
				return
			}
			b.mu.Unlock()
			m = next
		}
	}()
}

// try claims the fingerprint and builds it, writing the key done or failed. A build that ends because ctx did is left
// building by this builder: it takes it again when it starts (ADR 0478 clause 5).
func (b *Builder) try(ctx context.Context, module string, m delivery.Module) {
	plan, planErr, read := b.planOf(ctx, m.Manifest)
	nothing := read && planErr == nil && len(plan.Tasks) == 0
	if planErr == nil && !nothing {
		if why := b.stale(m, plan); why != "" {
			// The fingerprint names a toolchain the build nodes no longer agree on: a new fingerprint is coming, and a build
			// now would be of this source with another toolchain (ADR 0477 clause 2). Nothing is written.
			b.logf("%s at %s is not built: %s", module, m.FP2, why)
			return
		}
	}
	key, _ := delivery.BuilderBuildKey(module, m.FP2)
	b.mu.Lock()
	trunk := b.trunkFP[key]
	first := !b.tried[key]
	b.tried[key] = true
	epoch := b.epoch
	failedAt, failedBefore := b.secondAt[key]
	b.mu.Unlock()
	moreNodes := func(modules []string) bool {
		// A second build that failed is not tried again until the nodes change (issue 1664).
		return (!failedBefore || epoch > failedAt) && b.moreNodes(modules)
	}
	cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, MoreNodes: moreNodes, ImageBuilder: b.imageBuilderRuns, Trunk: trunk, Published: b.isPublished})
	if err != nil {
		b.logf("%s: %v", module, err)
		return
	}
	if cl == nil {
		return
	}
	// A done key taken again builds only what is missing: the build modules that ran on one node, and the image when it was
	// built as a candidate's; the other artifacts stay as the done key has them (ADR 0487 clauses 5 and 8).
	var prior *delivery.BuilderBuild
	var only map[string]bool
	publishing := false
	if cl.Build.State == delivery.BuildDone {
		prior, only = &cl.Build, map[string]bool{}
		for _, bm := range OneNodeModules(cl.Build) {
			only[bm] = true
		}
		if trunk {
			for _, bm := range publishable(cl.Build, b.imageBuilderRuns) {
				only[bm] = true
				publishing = true
			}
		}
	}
	var result delivery.BuilderBuild
	if nothing {
		// A manifest with no artifact is built by building nothing: its key is done with no artifact, claimed and written as
		// any other, so the controller has a done key to register it from (issue 1613, ADR 0478 clause 1, ADR 0501 clause 1).
		result = delivery.BuilderBuild{State: delivery.BuildDone, Step: "nothing to build: the manifest names no artifact"}
	} else {
		result = b.build(ctx, module, m, plan, planErr, prior, only)
	}
	if ctx.Err() != nil {
		return
	}
	if result.State == "" {
		b.logf("%s: left building, the toolchains moved under it", cl.Key)
		return
	}
	result.By, result.Manifest = b.Claimer.By, m.Manifest
	if publishing && result.State == delivery.BuildFailed {
		// The trunk's image or package was built only as a candidate's and its push or publish failed: the key says so, so the builds view, the
		// checker and the failed-key retake see it, and a look does not build the image again for nothing.
		result.Error = keyError([]string{"publishing what the merged commit built failed: " + result.Error})
	} else if cl.Build.State == delivery.BuildDone && result.State == delivery.BuildFailed && !strings.HasPrefix(result.Error, buildrun.DigestsDifferPrefix) {
		// A done key built again on a second node keeps its done value when the new build fails for another reason than
		// the digests: the first build stands, and a later change of the nodes looks again.
		b.logf("%s: built again on a second node and failed, the done build stands: %s", cl.Key, result.Error)
		b.mu.Lock()
		b.secondAt[cl.Key] = epoch
		b.mu.Unlock()
		return
	}
	if err := b.Claimer.Finish(ctx, cl, result); err != nil {
		b.logf("%s: the build could not be written: %v", cl.Key, err)
		return
	}
	if result.State == delivery.BuildFailed {
		b.logf("%s: failed: %s", cl.Key, result.Error)
	}
	if result.State == delivery.BuildQueued {
		b.logf("%s: waits for the package %s to be published", cl.Key, result.Waits)
	}
	if b.notePublished(result) {
		// A key may wait for the package this build published (ADR 0500 clause 6).
		b.Reconsider(ctx)
	}
}

// stale is why the module's fingerprint is not built now, or empty when it is: the toolchain each language of the plan
// has agreed (novox/hq ADR 0465) must be the one the fingerprint was computed with (Module.Toolchains).
func (b *Builder) stale(m delivery.Module, plan Plan) string {
	for _, bm := range plan.Modules() {
		if !plan.NeedsToolchain(bm) {
			continue // an archive is packed with no toolchain
		}
		language := buildrun.LanguageOf[bm]
		tc, none := delivery.AgreedToolchain(b.Toolchains.Entries(language))
		if none != "" {
			return fmt.Sprintf("the %s toolchain is not agreed: %s", language, none)
		}
		want, ok := m.Toolchains[language]
		if !ok || want.Version != tc.Version || want.Recipe != tc.Recipe {
			return fmt.Sprintf("the fingerprint names %s %s (recipe %s) and the nodes agree on %s (recipe %s)", language, want.Version, want.Recipe, tc.Version, tc.Recipe)
		}
	}
	return ""
}

func failed(format string, args ...any) delivery.BuilderBuild {
	return delivery.BuilderBuild{State: delivery.BuildFailed, Error: fmt.Sprintf(format, args...)}
}

// lane is the build of a plan's tasks of one build module on one node.
type lane struct {
	module, node string
	tasks        []Task
	digests      map[string]string // artifact -> digest
	references   map[string]string // artifact -> the reference an image was pushed to (a trunk build only)
	err          error
}

func (b *Builder) build(ctx context.Context, module string, m delivery.Module, plan Plan, planErr error, prior *delivery.BuilderBuild, only map[string]bool) delivery.BuilderBuild {
	switch {
	case planErr != nil:
		return failed("%s", planErr)
	case m.Path == "":
		// Empty is not known (a root module's path is "."): the indexer writes it again (issue 1615).
		return failed("the module's key names no directory (path), so the build tools cannot be told where %s lives", module)
	}
	repository := delivery.RepositoryName(m.Repository)
	var lanes []*lane
	used := map[string]delivery.ToolchainUsed{}
	var steps, oneNode []string
	// Every node chosen holds a reservation until its lane takes the place; a build that ends before its lanes start gives them up.
	giveUp := func(nodes ...string) {
		for _, n := range nodes {
			b.Places.Cancel(n)
		}
	}
	giveUpLanes := func() {
		for _, l := range lanes {
			b.Places.Cancel(l.node)
		}
	}
	var kept []string // build modules whose artifacts stay as the prior done key has them
	for _, bm := range plan.Modules() {
		if prior != nil && len(only) > 0 && !only[bm] {
			kept = append(kept, bm)
			continue
		}
		language := buildrun.LanguageOf[bm]
		entries := b.Toolchains.Entries(language)
		qualifies := func(node string) bool { return b.qualifies(node, bm) }
		var nodes []string
		if bm == ImageBuildModule {
			// An image is built on one node, chosen from the nodes whose image build module runs healthy (ADR 0487 clause 8):
			// the module has no toolchain language, so no toolchain entry names its nodes, and a second build would only
			// push the image again.
			ns, err := buildrun.ChooseAnyNodes("image", b.imageEntries(), b.Places, qualifies, 1)
			if err != nil {
				giveUpLanes()
				return failed("%s no node has %s, so the image cannot be built: the provision rootless-image-build is held by the module buildah; assign buildah and %s to a node (%v)", NoImageBuilderPrefix, ImageBuildModule, ImageBuildModule, err)
			}
			nodes = ns
		} else if plan.NeedsToolchain(bm) {
			ns, tc, err := buildrun.ChooseNodes(language, entries, b.Places, qualifies, b.Nodes)
			if err != nil {
				giveUpLanes()
				return failed("%s", err)
			}
			if want := m.Toolchains[language]; want.Version != tc.Version || want.Recipe != tc.Recipe {
				giveUp(ns...)
				giveUpLanes()
				return delivery.BuilderBuild{} // the agreed toolchain moved since try looked: nothing is written
			}
			used[language] = delivery.ToolchainUsed{Version: tc.Version, Recipe: tc.Recipe}
			nodes = ns
		} else {
			// An archive is packed with no toolchain (the fingerprint names none): any node of the build module that qualifies.
			ns, err := buildrun.ChooseAnyNodes(language, entries, b.Places, qualifies, b.Nodes)
			if err != nil {
				giveUpLanes()
				return failed("%s", err)
			}
			nodes = ns
		}
		if bm == ImageBuildModule {
			steps = append(steps, fmt.Sprintf("%s on %s", bm, nodes[0]))
		} else if len(nodes) == 1 {
			oneNode = append(oneNode, bm)
			steps = append(steps, fmt.Sprintf("%s on one node, %s", bm, nodes[0]))
		} else {
			sorted := append([]string(nil), nodes...)
			sort.Strings(sorted)
			steps = append(steps, fmt.Sprintf("%s on %s", bm, strings.Join(sorted, " and ")))
		}
		var tasks []Task
		for _, t := range plan.Tasks {
			if t.Module == bm {
				tasks = append(tasks, t)
			}
		}
		for _, n := range nodes {
			lanes = append(lanes, &lane{module: bm, node: n, tasks: tasks, digests: map[string]string{}, references: map[string]string{}})
		}
	}
	var wg sync.WaitGroup
	for _, l := range lanes {
		wg.Add(1)
		go func() { defer wg.Done(); b.runLane(ctx, l, repository, m, module) }()
	}
	wg.Wait()
	var errs []string
	for _, l := range lanes {
		if l.err != nil {
			errs = append(errs, l.err.Error())
		}
	}
	if len(errs) > 0 {
		if ref := waitsOf(errs); ref != "" {
			// The module's lock names a version of a package the registry does not hold yet: the key waits for it, queued, and
			// is not failed (ADR 0500 clause 6).
			return delivery.BuilderBuild{State: delivery.BuildQueued, Waits: ref, Step: "waits for the package " + ref + " to be published"}
		}
		return failed("%s", keyError(errs))
	}
	var artifacts []delivery.BuiltArtifact
	keptModule := map[string]bool{}
	for _, bm := range kept {
		keptModule[bm] = true
	}
	for _, t := range plan.Tasks {
		if keptModule[t.Module] {
			found := false
			for _, a := range prior.Artifacts {
				if a.Name == t.Artifact {
					artifacts, found = append(artifacts, a), true
				}
			}
			if !found {
				return failed("the done build holds no artifact %s to keep", t.Artifact)
			}
			continue
		}
		byNode := map[string]string{}
		for _, l := range lanes {
			if d, ok := l.digests[t.Artifact]; ok && l.module == t.Module {
				byNode[l.node] = d
			}
		}
		d, err := buildrun.Agree(t.Artifact, byNode)
		if err != nil {
			return failed("%s", err)
		}
		built := delivery.BuiltArtifact{Name: t.Artifact, Kind: t.Kind, Digest: d}
		for _, l := range lanes {
			if l.module != t.Module {
				continue
			}
			r := l.references[t.Artifact]
			if r == "" && stored(t.Kind) {
				// A bundle, an archive or an upstream copy is done only when its push was answered with where it is (ADR 0498
				// clause 1); a candidate's included (clause 3).
				return failed("%s%s on %s: the build answered no reference, so its push to the artifact store was not answered", delivery.NotStored, t.Artifact, l.node)
			}
			if r != "" {
				built.Reference = r
			}
		}
		artifacts = append(artifacts, built)
	}
	for _, bm := range kept {
		steps = append(steps, bm+" kept")
		for lang, u := range prior.Toolchains {
			if _, ok := used[lang]; !ok && buildrun.LanguageOf[bm] == lang {
				used[lang] = u
			}
		}
	}
	step := "built " + strings.Join(steps, "; ")
	if len(oneNode) > 0 {
		// Marked so that Claim builds the key again once a second node qualifies (ADR 0478 clause 4).
		step = OneNodePrefix + strings.Join(oneNode, ", ") + "): " + step
	}
	if len(used) == 0 {
		used = nil
	}
	done := delivery.BuilderBuild{State: delivery.BuildDone, By: b.Claimer.By, Manifest: m.Manifest, Step: step, Artifacts: artifacts, Toolchains: used}
	if _, err := delivery.EncodeBuilderBuild(done); err != nil {
		// A reference the build module answered that does not name the artifact and its digest is no reference.
		return failed("%s%v", delivery.NotStored, err)
	}
	return done
}

// stored is whether an artifact of this kind is in the artifact store once built, a candidate's too, so its done key
// names its reference (ADR 0498 clauses 1, 3 and 4); an image or a package names one only when built for the trunk.
func stored(kind string) bool {
	return kind == catalogue.ArtifactBundle || kind == catalogue.ArtifactArchive || kind == catalogue.ArtifactUpstream
}

// waitsFor opens the reason of a fetch step whose module's lock names a package version the registry does not hold, then
// `<name>@<version>` (the Node build module's, ADR 0500 clause 6).
const waitsFor = delivery.WaitsFor

// waitsOf is the package version a lane's error says the module waits for, or "" when none does or what it names is not
// `<name>@<version>`.
func waitsOf(errs []string) string {
	for _, e := range errs {
		at := strings.Index(e, waitsFor)
		if at < 0 {
			continue
		}
		ref, _, _ := strings.Cut(e[at+len(waitsFor):], " ")
		ref = strings.TrimRight(ref, ":;,")
		if _, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildQueued, Waits: ref}); err == nil {
			return ref
		}
	}
	return ""
}

// keyPrefixes are the errors a build module's step may open its reason with that the key's error must open with too, so
// the claim reads them (delivery.FailedUntilChanged), first the ones not built again.
var keyPrefixes = []string{delivery.NotPinned, delivery.VersionTaken, delivery.RefusedKind, delivery.NotStored}

// keyError is the error of a key from its lanes' errors: when one names a prefix of keyPrefixes inside the step's words
// (`<tool> on <node>: the step <key> failed: not stored: ...`), the key's error opens with that prefix, so a reader of the
// key need not search for it.
func keyError(errs []string) string {
	for _, prefix := range keyPrefixes {
		for i, e := range errs {
			if strings.HasPrefix(e, prefix) {
				return strings.Join(errs, "; ")
			}
			if at := strings.Index(e, prefix); at >= 0 {
				rest := append([]string{prefix + e[:at] + e[at+len(prefix):]}, append(append([]string(nil), errs[:i]...), errs[i+1:]...)...)
				return strings.Join(rest, "; ")
			}
		}
	}
	return strings.Join(errs, "; ")
}

func (b *Builder) online(node string) bool { return b.Presence == nil || b.Presence.Online(node) }

// qualifies is whether the node can build with the build module now: it is online and the build module on it is healthy.
func (b *Builder) qualifies(node, bm string) bool {
	return b.Places.CapacityOf(node) > 0 && b.online(node) && (b.Health == nil || b.Health.Healthy(node, bm))
}

// imageEntries are the nodes an image may be built on, as toolchain entries for the chooser: every node the builder has a
// capacity for or a build module toolchain from. Whether the image build module runs there is qualifies', from the have.
func (b *Builder) imageEntries() []delivery.NodeToolchain {
	seen := map[string]bool{}
	var entries []delivery.NodeToolchain
	add := func(n string) {
		if !seen[n] {
			seen[n] = true
			entries = append(entries, delivery.NodeToolchain{Node: n})
		}
	}
	for n := range b.Places.Capacity {
		add(n)
	}
	for _, language := range []string{"go", "typescript"} {
		for _, e := range b.Toolchains.Entries(language) {
			add(e.Node)
		}
	}
	sort.Slice(entries, func(i, j int) bool { return entries[i].Node < entries[j].Node })
	return entries
}

// imageBuilderRuns is whether a node can build an image now.
func (b *Builder) imageBuilderRuns() bool {
	for _, e := range b.imageEntries() {
		if b.qualifies(e.Node, ImageBuildModule) {
			return true
		}
	}
	return false
}

// moreNodes is whether any of these build modules has two nodes that qualify now.
func (b *Builder) moreNodes(modules []string) bool {
	for _, bm := range modules {
		seen := map[string]bool{}
		n := 0
		for _, e := range b.Toolchains.Entries(buildrun.LanguageOf[bm]) {
			if !seen[e.Node] && b.qualifies(e.Node, bm) {
				seen[e.Node] = true
				n++
			}
		}
		if n >= 2 {
			return true
		}
	}
	return false
}

// step runs one step of a build module; the lane that runs it holds its node's place.
func (b *Builder) step(ctx context.Context, bm, node, tool string, args map[string]any) (json.RawMessage, error) {
	return b.Runner.Run(ctx, bm, node, toolPrefix[bm]+tool, args)
}

// fetchResult and buildResult are the results of a build module's fetch and build steps as the builder reads them.
type fetchResult struct {
	Workspace string   `json:"workspace"`
	Merge     string   `json:"merge"`
	Conflicts []string `json:"conflicts"`
}

type buildResult struct {
	Digest    string `json:"digest"`
	Reference string `json:"reference"`
}

// runLane takes a place on the node, fetches the commit there and builds each of the lane's artifacts, one step at a time.
// An upstream artifact is copied by its pinned reference with no fetch (ADR 0498 clause 4); a lane of copies alone fetches
// nothing.
func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m delivery.Module, module string) {
	release, err := b.Places.Acquire(ctx, l.node)
	if err != nil {
		l.err = err
		return
	}
	defer release()
	req := buildstep.FetchRequest{Repository: repository, Commit: m.Commit}
	if key, err := delivery.BuilderBuildKey(module, m.FP2); err == nil {
		b.mu.Lock()
		v, candidate := b.via[key]
		b.mu.Unlock()
		if candidate {
			req.Ref, req.Base = v.Ref, v.Base
		}
	}
	if m.Inputs2 != nil && len(m.Inputs2.Paths) > 0 {
		// A module whose every artifact names its paths is built from those alone (ADR 0499 clause 4): the paths come from
		// the indexer's kept inputs, never from a manifest (the lead's decision L8 of R4e).
		req.Dir = m.Path
		for p := range m.Inputs2.Paths {
			req.Paths = append(req.Paths, p)
		}
		sort.Strings(req.Paths)
	}
	fetchArgs := req.Args()
	candidate := req.Base != ""
	var fetched fetchResult
	if needsFetch(l.tasks) {
		raw, err := b.step(ctx, l.module, l.node, "fetch", fetchArgs)
		if err != nil {
			l.err = err
			return
		}
		if err := json.Unmarshal(raw, &fetched); err != nil || fetched.Workspace == "" {
			l.err = fmt.Errorf("the fetch on %s answered no workspace: %s", l.node, raw)
			return
		}
		// A candidate is built from its head merged onto the trunk. A head that does not merge ends the fetch passed with the tree at
		// the trunk, and building that would store the trunk's build under the candidate's fingerprint.
		if candidate {
			if len(fetched.Conflicts) > 0 {
				l.err = fmt.Errorf("the head does not merge into the trunk on %s: files in conflict: %s", l.node, strings.Join(fetched.Conflicts, ", "))
				return
			}
			if fetched.Merge == "" {
				l.err = fmt.Errorf("the fetch with a base on %s answered no merge: %s", l.node, raw)
				return
			}
		}
	}
	for _, t := range l.tasks {
		tool, args := "build", map[string]any{"workspace": fetched.Workspace, "module": m.Path, "artifact": t.Artifact}
		if t.Kind == catalogue.ArtifactUpstream {
			// The copy runs no code of a change: the module's name (the reference's path), the artifact and its pin.
			tool, args = "copy", map[string]any{"module": module, "artifact": t.Artifact, "from": t.From}
		} else if !candidate {
			// Only a trunk key's build pushes an image or publishes a package; a candidate's never does (ADR 0487 clause 12,
			// ADR 0498 clauses 5 and 6). A bundle or an archive is pushed either way (clause 3).
			args["publish"] = true
		}
		raw, err := b.step(ctx, l.module, l.node, tool, args)
		if err != nil {
			l.err = err
			return
		}
		var built buildResult
		if err := json.Unmarshal(raw, &built); err != nil || built.Digest == "" {
			l.err = fmt.Errorf("the %s of %s of %s on %s answered no digest: %s", tool, t.Artifact, module, l.node, raw)
			return
		}
		l.digests[t.Artifact] = built.Digest
		if built.Reference != "" && (stored(t.Kind) || !candidate) {
			l.references[t.Artifact] = built.Reference
		}
	}
}

// needsFetch is whether any of the tasks builds from the module's source: every one but an upstream copy.
func needsFetch(tasks []Task) bool {
	for _, t := range tasks {
		if t.Kind != catalogue.ArtifactUpstream {
			return true
		}
	}
	return false
}