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

package build

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"slices"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
	"git.novox.be/novox/mesh-delivery/sdk/forms"
	"git.novox.be/novox/mesh-sdk/go/buildstep"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/delivery"

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

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

// 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 by asking the build seats (one ask per build module,
// pulled by whichever build node has a free slot: ADR 0525), 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
	// Asks puts each build module's part of a build as one ask on its build seat's stream and waits for its step (novox/hq
	// ADR 0525 §1, §3, §6; issue 1887 A4): the builder chooses no node, whichever build node pulls the ask builds it.
	Asks interface {
		Put(ctx context.Context, a delivery.StepAsk) error
		Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error)
	}
	// Held says whether an ask a key names is on its stream or a build node wrote a step for it (issue 1910): a key taken
	// over at start whose ask is neither was left by a builder that stopped between naming the ask and putting it, and
	// the ask is put then, under the same id. Nil takes every named ask as held.
	Held interface {
		Held(ctx context.Context, a delivery.StepAsk) (bool, error)
	}
	Toolchains ToolchainEntries
	Manifests  Manifests
	// Limit is how many builds the builder has asked at once; their asks wait in the build seats' streams, which are the
	// backlog (ADR 0525 §9), not here. 256 when 0.
	Limit int
	Logf  func(format string, args ...any)
	// Answered is told the build module whose step answered a reference (issue 1886): what the builder says it writes
	// rests on it. Nil tells nobody.
	Answered func(buildModule string)
	// Stored answers which references the artifact store no longer holds (the seat mesh-artifact-sweep's verb held; issue 1937): a done
	// key of the trunk's fingerprint is asked before it is left done (ClaimOptions.Stored). Nil asks nothing.
	Stored func(ctx context.Context, references []string) (gone []string, err error)
	// Trunk answers the head of a repository in the gate's source revision rev (source.main at that revision): the trunk a
	// candidate's preview is merged onto (Candidate.Base). Nil builds no candidate's fingerprint, and says why.
	Trunk func(ctx context.Context, rev uint64, repository string) (string, error)
	// Gone says why an ask with no ended step can no longer be answered, or "" when it can (issue 1945): asked once the
	// ask has waited CheckAfter, and again each CheckAfter it goes on waiting. An ask that cannot be answered is asked
	// again under a new id, at most Askings times. Nil, or CheckAfter 0, never asks it.
	Gone interface {
		Gone(ctx context.Context, a delivery.StepAsk) (string, error)
	}
	CheckAfter time.Duration
	// Askings is how many times one build module's part of a build is asked, the first included, before its key is failed
	// with why; DefaultAskings when 0.
	Askings int
	// DeclineBound and DeclinePause bound and pace the asks build nodes off the agreed toolchain declined (ADR 0584 clause 2):
	// they are no askings. DefaultDeclineBound and DefaultDeclinePause when 0; a negative DeclinePause puts it again at once.
	DeclineBound, DeclinePause time.Duration
	// Said is told each ask asked again (the builder's event build.asked-again). Nil tells nobody.
	Said func(AskedAgain)
	// Log is told each build's start and end, the entries of the builder's log (novox/hq ADR 0571 §1). Nil tells nobody.
	Log func(delivery.BuildLogEntry)
	// Now is the builder's clock: of the log, and of the decline bound (DeclineBound runs from the first decline); nil is
	// time.Now. Node is the node the builder runs on, which asked the build.
	Now  func() time.Time
	Node string
	// Numbers numbers each trunk build at its claim and records what it was built with (design 51 *The versions*); nil
	// numbers nothing.
	Numbers Numbers
	// Tags answers a check's build's tag from the kept tag asks; nil numbers no check's build.
	Tags TagOf
	// Forms answers what each node's engine states it reads (ADR 0618 §2, ADR 0624 §1): present, the nodes whose engine
	// answered when last asked, and counted, every node the placement names, present or away, a node with no statement
	// stating nothing (the baseline's first forms). A done key names each artifact in the forms these keep, the oldest in
	// its single reference, and its needs. Nil: no node states anything, and the first forms are kept.
	Forms func() (present, counted map[string][]string)

	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
	// 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
	// held are the done keys' references the store answered held, by `<key> <references>`, asked once per process: a
	// blob collected later is found by the take-in, whose artifacts-gone forgets the key here (issue 1937).
	held map[string]bool
	// waiting are the lanes that wait for an ask's step, by build key and language: AskAgain asks through them.
	waiting map[string]map[string]chan string
	// hands are the requests to build a key again by hand that wait for the claim and its asks (BuildAgain), by build key.
	hands map[string]*handWatch
}

type planned struct {
	plan Plan
	err  error
}

func (b *Builder) init() {
	b.once.Do(func() {
		limit := b.Limit
		if limit <= 0 {
			limit = 256
		}
		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.published, b.held, b.waiting = map[string]bool{}, map[string]bool{}, map[string]map[string]chan string{}
		b.hands = map[string]*handWatch{}
	})
}

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 toolchain moved, so a build that waited for the nodes to agree on one may go on.
func (b *Builder) NodesChanged(ctx context.Context) { 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()
}

// storedOnce is Stored for one key, asked once per process while the store answers every reference held.
func (b *Builder) storedOnce(key string) func(ctx context.Context, references []string) ([]string, error) {
	if b.Stored == nil {
		return nil
	}
	return func(ctx context.Context, references []string) ([]string, error) {
		id := key + " " + strings.Join(references, " ")
		b.mu.Lock()
		known := b.held[id]
		b.mu.Unlock()
		if known {
			return nil, nil
		}
		gone, err := b.Stored(ctx, references)
		if err == nil && len(gone) == 0 {
			b.mu.Lock()
			b.held[id] = true
			b.mu.Unlock()
		}
		return gone, err
	}
}

// ArtifactsGone takes the seat mesh-artifact-sweep's event artifacts-gone in (issue 1937, ADR 0571 §2): the sweep found
// that the store does not hold what the module's key names. The module is looked at again, which asks the store (ClaimOptions.Stored) and
// builds the key again; an event for a fingerprint the module no longer has is said and left. The event only wakes the
// builder: what is built rests on the store's answer, never on the event alone.
func (b *Builder) ArtifactsGone(ctx context.Context, e delivery.ArtifactsGone) {
	b.init()
	b.mu.Lock()
	m, ok := b.known[e.Module]
	if key, err := delivery.BuilderBuildKey(e.Module, e.Fingerprint); err == nil {
		for id := range b.held {
			if strings.HasPrefix(id, key+" ") {
				delete(b.held, id)
			}
		}
	}
	b.mu.Unlock()
	if !ok || m.FP2 != e.Fingerprint {
		b.logf("%s: heard %s for %s, a fingerprint the module does not have now: nothing to build", e.Module,
			delivery.ArtifactsGoneEvent, e.Fingerprint)
		return
	}
	b.logf("%s: heard %s (%s): asking the artifact store and building the key again", e.Module,
		delivery.ArtifactsGoneEvent, strings.Join(e.Gone, ", "))
	b.consider(ctx, e.Module, m)
}

// 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.TrunkMoved(ctx)
}

// TrunkMoved considers every candidate again once a revision of the gate's source is heard: a candidate heard before the
// revision it was previewed onto had no trunk head to be merged onto.
func (b *Builder) TrunkMoved(ctx context.Context) {
	b.init()
	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
		}
	}()
}

// startOnce looks at the module for a key the caller has set inflight already (BuildAgain): the by-hand look runs once, and
// a look the builder queued for the key meanwhile (again) still runs after it, as consider's loop does.
func (b *Builder) startOnce(ctx context.Context, key, module string, m delivery.Module) {
	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 // a look the builder queued meanwhile is not lost (consider's rule); it is not a second by-hand build
		}
	}()
}

// 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) {
	key, _ := delivery.BuilderBuildKey(module, m.FP2)
	defer b.handEnded(key)
	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 0584 clause 1, which changes ADR 0477 clause 2). Nothing is written.
			b.logf("%s at %s is not built: %s", module, m.FP2, why)
			return
		}
	}
	b.mu.Lock()
	trunk := b.trunkFP[key]
	first := !b.tried[key]
	b.tried[key] = true
	b.mu.Unlock()
	var numbering versions.Numbering
	tag := ""
	if !trunk && b.Tags != nil && b.refOf(module, m.FP2) != "" {
		tag, _ = b.Tags(ctx, delivery.RepositoryName(m.Repository), m.Commit)
	}
	if (trunk || tag != "") && planErr == nil && !nothing {
		var goOn bool
		if numbering, goOn = b.number(ctx, module, m, plan, tag); !goOn {
			return
		}
	}
	cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished,
		Stored: b.storedOnce(key)})
	if err != nil {
		b.logf("%s: %v", module, err)
		return
	}
	if cl == nil {
		return
	}
	b.handClaimed(key, cl.Rev)
	b.logBuild(LogStarted(module, m, b.refOf(module, m.FP2), b.Node, b.now()))
	if cl.Unasked != nil {
		b.logf("%v", cl.Unasked)
	}
	if len(cl.Gone) > 0 {
		// Said aloud (issue 1937): the key was done, and the store's collector deleted what it names.
		b.logf("%s: built again: the artifact store no longer holds %s", cl.Key, strings.Join(cl.Gone, ", "))
	}
	// A done key taken again builds only what is missing: the image or the package built unpublished as a candidate's; the
	// other artifacts stay as the done key has them (ADR 0487 clauses 5 and 8, ADR 0498 clause 6).
	var prior *delivery.BuilderBuild
	var only map[string]bool
	publishing := false
	if cl.Build.State == delivery.BuildDone {
		prior, only = &cl.Build, map[string]bool{}
		if trunk {
			for _, bm := range publishable(cl.Build, imageSeat) {
				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, b.notes(cl))
	}
	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 len(cl.Gone) > 0 && result.State == delivery.BuildDone {
		result.Step += " (built again: the artifact store no longer held " + strings.Join(cl.Gone, ", ") + ")"
	}
	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})
	}
	if err := b.Claimer.Finish(ctx, cl, result); err != nil {
		b.logf("%s: the build could not be written: %v", cl.Key, err)
		return
	}
	b.logBuild(LogEnded(module, m, b.refOf(module, m.FP2), b.Node, b.now(), result))
	if result.State == delivery.BuildDone && b.Numbers != nil && len(numbering.Versions) > 0 {
		if err := b.Numbers.Built(ctx, module, m, plan, numbering, result.Artifacts); err != nil {
			b.logf("%s: %v", cl.Key, err)
		}
	}
	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]
		ag := delivery.ToolchainAgreement(b.Toolchains.Entries(language))
		if ag.Wait != "" {
			return fmt.Sprintf("the %s toolchain is not agreed: %s", language, ag.Wait)
		}
		tc := ag.Toolchain
		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 // node is the one that pulled the lane's ask
	tasks        []Task
	toolchain    *delivery.ToolchainUsed
	digests      map[string]string // artifact -> digest
	references   map[string]string // artifact -> the reference an image was pushed to (a trunk build only)
	imports      []string          // the mesh packages the lane's programs import, as its build module answered them
	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, asked *askNotes) 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 sitting []string // the nodes that sit out of a language's build, said in the key's step (ADR 0584 clause 2)
	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
		}
		var toolchain *delivery.ToolchainUsed
		if bm != ImageBuildModule && plan.NeedsToolchain(bm) {
			// The ask names the toolchain the fingerprint was computed with; a node building with another refuses it.
			language := buildrun.LanguageOf[bm]
			ag := delivery.ToolchainAgreement(b.Toolchains.Entries(language))
			if ag.Wait != "" {
				return failed("the %s toolchain is not agreed: %s", language, ag.Wait)
			}
			tc := ag.Toolchain
			sitting = append(sitting, SittingOutWords(language, ag)...)
			if want := m.Toolchains[language]; want.Version != tc.Version || want.Recipe != tc.Recipe {
				return delivery.BuilderBuild{} // the agreed toolchain moved since try looked: nothing is written
			}
			u := delivery.ToolchainUsed{Version: tc.Version, Recipe: tc.Recipe}
			used[language], toolchain = u, &u
		}
		var tasks []Task
		for _, t := range plan.Tasks {
			if t.Module == bm {
				tasks = append(tasks, t)
			}
		}
		// One ask builds each fingerprint's part of a build module, on whichever node pulls it (ADR 0525 §3).
		lanes = append(lanes, &lane{module: bm, tasks: tasks, toolchain: toolchain, digests: map[string]string{}, references: map[string]string{}})
	}
	if key, err := delivery.BuilderBuildKey(module, m.FP2); err == nil {
		b.handLanes(key, len(lanes))
	}
	var wg sync.WaitGroup
	for _, l := range lanes {
		wg.Add(1)
		go func() { defer wg.Done(); b.runLane(ctx, l, repository, m, module, asked) }()
	}
	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
	var present, counted map[string][]string
	if b.Forms != nil {
		present, counted = b.Forms()
	}
	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}
		var answered []string
		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 != "" {
				answered = append(answered, r)
			}
		}
		// The forms a counted node's engine reads are kept, the oldest in the single reference (ADR 0624 §1, §3).
		artifacts = append(artifacts, forms.Current.Choose(built, answered, present, counted))
	}
	var steps []string
	for _, l := range lanes {
		steps = append(steps, fmt.Sprintf("%s on %s", l.module, l.node))
	}
	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(sitting) > 0 {
		step += " (" + strings.Join(sitting, "; ") + ")"
	}
	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,
		Imports: importsOf(lanes), Needs: forms.Current.Needs(artifacts)} // the forms it is written in (ADR 0618 §1)
	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, "; ")
}

// imageSeat says an image build is asked of the image build seat, which takes it when a node pulls it: a key failed for
// want of an image builder before R4g is taken again (ADR 0525 §1).
func imageSeat() bool { return true }

// askLanguage is the build seat language of each build module (delivery.AskSubject).
var askLanguage = map[string]string{buildrun.GoBuildModule: delivery.AskGo, buildrun.NodeBuildModule: delivery.AskNode, ImageBuildModule: delivery.AskImage}

// askNotes writes a claimed key's asks into it, building, before each is put (ADR 0525 §1; issue 1887 A4b), and holds
// the asks a key taken over at start already names, so they are waited for and never put again.
type askNotes struct {
	mu      sync.Mutex
	b       *Builder
	cl      *Claim
	resumed map[string]string
	// step is the step words of a key taken over at start: an ask asked again says its asking there (askingOf).
	step string
}

func (b *Builder) notes(cl *Claim) *askNotes {
	n := &askNotes{b: b, cl: cl}
	if cl.Build.State == delivery.BuildBuilding && len(cl.Build.Asks) > 0 {
		n.resumed, n.step = maps.Clone(cl.Build.Asks), cl.Build.Step
	}
	return n
}

// resume is the id of the ask a taken-over key names for language, or "".
func (n *askNotes) resume(language string) string {
	n.mu.Lock()
	defer n.mu.Unlock()
	return n.resumed[language]
}

// note writes the ask's id into the key, building, by compare-and-set at the claim's revision.
func (n *askNotes) note(ctx context.Context, language, id, step string) error {
	n.mu.Lock()
	defer n.mu.Unlock()
	v := n.cl.Build
	v.State, v.By, v.Step, v.Error, v.Artifacts, v.Needs = delivery.BuildBuilding, n.b.Claimer.By, step, "", nil, nil
	v.Asks = maps.Clone(v.Asks)
	if v.Asks == nil {
		v.Asks = map[string]string{}
	}
	v.Asks[language] = id
	if err := n.b.Claimer.Finish(ctx, n.cl, v); err != nil {
		return err
	}
	n.b.handAsk(n.cl.Key, language, id)
	return nil
}

// runLane puts the lane's part of the build as one ask on its build seat's stream (the delivery stream for a trunk key, the
// check stream for a candidate's) and reads what the node that pulled it built (ADR 0525 §1, §3, §6). An upstream artifact
// is copied by its pinned reference with no fetch (ADR 0498 clause 4); a lane of copies alone fetches nothing. A lane of a
// key taken over at start waits for the ask the key names and puts none.
func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m delivery.Module, module string, asked *askNotes) {
	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)
	}
	candidate := req.Base != ""
	language := askLanguage[l.module]
	kind := delivery.AskDelivery
	if candidate {
		kind = delivery.AskCheck
	}
	job := &delivery.BuildJob{Module: module, Path: m.Path, FP: m.FP2, Toolchain: l.toolchain,
		// 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).
		Publish: !candidate}
	for _, t := range l.tasks {
		a := delivery.AskArtifact{Artifact: t.Artifact}
		if t.Kind == catalogue.ArtifactUpstream {
			a.From = t.From // the copy runs no code of a change: the module's name, the artifact and its pin
		}
		job.Artifacts = append(job.Artifacts, a)
	}
	if needsFetch(l.tasks) {
		job.Fetch = req.Args()
	}
	ask := delivery.StepAsk{ID: buildstep.NewID(), Kind: kind, Language: language, Asker: delivery.BuilderModule, At: time.Now(), Build: job}
	if language == delivery.AskGo && job.Fetch != nil && len(m.Packages) > 0 {
		// The versions of the mesh packages the fingerprint was computed with (novox/hq ADR 0573 §2): the build module writes
		// them into the workspace, whatever the repository's go.mod names.
		ask.Packages = maps.Clone(m.Packages)
	}
	if id := asked.resume(language); id != "" {
		ask.ID = id // put before this builder started again: the stream holds it, or a node already ran it
		if b.Held != nil {
			held, err := b.Held.Held(ctx, ask)
			if err != nil {
				// Said, and the key is not failed: the ask may well be on its stream and its build finish (issue 1910).
				b.logf("whether the ask %s of %s the key names is on its stream cannot be read, so it is waited for as put: %v", ask.ID, l.module, err)
				held = true
			}
			if !held {
				// The builder that named it stopped before it put it (issue 1910): put it now, under the same id.
				b.logf("the ask %s of %s that the key names was never put: it is put now", ask.ID, l.module)
				if err := b.Asks.Put(ctx, ask); err != nil {
					l.err = err
					return
				}
			}
		}
	} else {
		if err := asked.note(ctx, language, ask.ID, "asked "+l.module); err != nil {
			l.err = fmt.Errorf("the ask of %s could not be written to the key: %w", l.module, err)
			return
		}
		if err := b.Asks.Put(ctx, ask); err != nil {
			l.err = err
			return
		}
	}
	step, ask, err := b.answer(ctx, l, ask, asked, module, m.FP2)
	if err != nil {
		l.err = fmt.Errorf("%s: %w", l.module, err)
		return
	}
	l.node = step.Node
	if step.State != delivery.StepPassed {
		l.err = fmt.Errorf("%s on %s: the ask %s %s: %s", l.module, step.Node, ask.ID, step.State, step.Reason)
		return
	}
	var answer delivery.BuildAnswer
	if err := json.Unmarshal(step.Result, &answer); err != nil {
		l.err = fmt.Errorf("the ask of %s on %s answered what is not a build: %s", l.module, step.Node, step.Result)
		return
	}
	l.imports = answer.Imports
	built := map[string]delivery.AskBuilt{}
	for _, a := range answer.Artifacts {
		built[a.Artifact] = a
	}
	for _, t := range l.tasks {
		a, ok := built[t.Artifact]
		if !ok || a.Digest == "" {
			l.err = fmt.Errorf("the build of %s of %s on %s answered no digest: %s", t.Artifact, module, step.Node, step.Result)
			return
		}
		l.digests[t.Artifact] = a.Digest
		if a.Reference != "" && (stored(t.Kind) || !candidate) {
			l.references[t.Artifact] = a.Reference
		}
		if a.Reference != "" && b.Answered != nil {
			b.Answered(l.module)
		}
	}
}

// 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
}

func (b *Builder) now() time.Time {
	if b.Now != nil {
		return b.Now()
	}
	return time.Now()
}

// logBuild tells the log an entry, said in the journal when it cannot be one.
func (b *Builder) logBuild(e delivery.BuildLogEntry) {
	if b.Log == nil {
		return
	}
	if _, err := delivery.EncodeBuildLogEntry(e); err != nil {
		b.logf("%s at %s: no log entry: %v", e.Module, e.FP, err)
		return
	}
	b.Log(e)
}

// refOf is the ref a candidate's build of module's fingerprint is built for, "" for the trunk's.
func (b *Builder) refOf(module, fp string) string {
	key, err := delivery.BuilderBuildKey(module, fp)
	if err != nil {
		return ""
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.via[key].Ref
}

// LogStarted is the log's entry of a build of module's fingerprint starting (ADR 0571 §1): the module, the fingerprint, the
// source revision, where it is built from (repository, directory, commit, the candidate's ref), the node of the builder
// that asked it and the time.
func LogStarted(module string, m delivery.Module, ref, by string, at time.Time) delivery.BuildLogEntry {
	dir := m.Path
	if dir == "" {
		dir = delivery.RootPath
	}
	return delivery.BuildLogEntry{Event: delivery.BuildStartedEvent, Module: module, FP: m.FP2, Source: strconv.FormatUint(m.Src, 10),
		Repository: m.Repository, Directory: dir, Commit: m.Commit, Ref: ref, Node: by, At: at.UTC()}
}

// LogEnded is the log's entry of that build ending: the outcome (done, failed or queued), the manifest built, the bases it
// was built against and each artifact made, as the done key has them: what the operator's command nox builds shows of a build.
func LogEnded(module string, m delivery.Module, ref, by string, at time.Time, result delivery.BuilderBuild) delivery.BuildLogEntry {
	e := LogStarted(module, m, ref, by, at)
	e.Event, e.Outcome, e.Manifest = delivery.BuildEndedEvent, result.State, m.Manifest
	if result.Manifest != "" {
		e.Manifest = result.Manifest
	}
	if m.Inputs2 != nil {
		e.Against = append([]string(nil), m.Inputs2.Bases...)
	}
	e.Made = append([]delivery.BuiltArtifact(nil), result.Artifacts...)
	e.Imports = append([]string(nil), result.Imports...) // the indexer keeps them on the module (novox/hq ADR 0593 §2)
	return e
}

// importsOf are the mesh packages the lanes' programs import, as their build modules answered them (novox/hq ADR 0593
// §2): sorted and each once; nil when none answered any. A lane kept from a prior build answered none, so a build that
// kept one reports only what it built.
func importsOf(lanes []*lane) []string {
	seen := map[string]bool{}
	for _, l := range lanes {
		for _, p := range l.imports {
			seen[p] = true
		}
	}
	if len(seen) == 0 {
		return nil
	}
	return slices.Sorted(maps.Keys(seen))
}

// SittingOutWords says, per node that is not on the agreed toolchain of a language, that it sits out and which version
// it has against the agreed one: "ace sits out: typescript 26.12.1, agreed 26.13.0" (novox/hq ADR 0584 clause 2).
func SittingOutWords(language string, a delivery.Agreement) []string {
	var out []string
	for _, s := range a.SittingOut {
		out = append(out, fmt.Sprintf("%s sits out: %s %s, agreed %s", s.Node, language, s.Version, a.Toolchain.Version))
	}
	return out
}