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

package build

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

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

// ToolchainEntries are the nodes' toolchain entries of a language (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{GoBuildModule: "build_go_", NodeBuildModule: "build_node_"}

// 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     Runner
	Toolchains ToolchainEntries
	Manifests  Manifests
	// 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
	running  map[string]int             // steps in flight, by `<build module>/<node>`
	plans    map[string]planned         // by manifest digest
}

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.known, b.inflight, b.running, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]int{}, map[string]planned{}
	})
}

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
	b.mu.Unlock()
	b.consider(ctx, key, m)
}

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

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.mu.Unlock()
		return
	}
	b.inflight[key] = true
	b.mu.Unlock()
	b.wg.Add(1)
	go func() {
		defer b.wg.Done()
		defer func() { b.mu.Lock(); delete(b.inflight, key); b.mu.Unlock() }()
		b.sem <- struct{}{}
		defer func() { <-b.sem }()
		b.try(ctx, module, m)
	}()
}

// 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)
	if read && planErr == nil && len(plan.Tasks) == 0 {
		return // nothing in the manifest for a builder to build
	}
	cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest)
	if err != nil {
		b.logf("%s: %v", module, err)
		return
	}
	if cl == nil {
		return
	}
	result := b.build(ctx, module, m, plan, planErr)
	if ctx.Err() != nil {
		return
	}
	result.By, result.Manifest = b.Claimer.By, m.Manifest
	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)
	}
}

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
	err          error
}

func (b *Builder) build(ctx context.Context, module string, m delivery.Module, plan Plan, planErr error) delivery.BuilderBuild {
	switch {
	case planErr != nil:
		return failed("%s", planErr)
	case m.Path == "":
		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 []string
	for _, bm := range plan.Modules() {
		language := LanguageOf[bm]
		nodes, tc, err := ChooseNodes(language, b.Toolchains.Entries(language), b.runningOn(bm), b.Nodes)
		if err != nil {
			return failed("%s", err)
		}
		used[language] = delivery.ToolchainUsed{Version: tc.Version, Recipe: tc.Recipe}
		if len(nodes) == 1 {
			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{}})
		}
	}
	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 {
		return failed("%s", strings.Join(errs, "; "))
	}
	var artifacts []delivery.BuiltArtifact
	for _, t := range plan.Tasks {
		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 := Agree(t.Artifact, byNode)
		if err != nil {
			return failed("%s", err)
		}
		artifacts = append(artifacts, delivery.BuiltArtifact{Name: t.Artifact, Kind: t.Kind, Digest: d})
	}
	return delivery.BuilderBuild{State: delivery.BuildDone, Step: "built " + strings.Join(steps, "; "), Artifacts: artifacts, Toolchains: used}
}

func (b *Builder) runningOn(buildModule string) map[string]int {
	b.mu.Lock()
	defer b.mu.Unlock()
	out := map[string]int{}
	for k, n := range b.running {
		if mod, node, _ := strings.Cut(k, "/"); mod == buildModule {
			out[node] = n
		}
	}
	return out
}

// step runs one step of a build module, counting it as running on the node while it does.
func (b *Builder) step(ctx context.Context, bm, node, tool string, args map[string]any) (json.RawMessage, error) {
	k := bm + "/" + node
	b.mu.Lock()
	b.running[k]++
	b.mu.Unlock()
	defer func() { b.mu.Lock(); b.running[k]--; b.mu.Unlock() }()
	return b.Runner.Run(ctx, bm, node, toolPrefix[bm]+tool, args)
}

// runLane fetches the commit on the node and builds each of the lane's artifacts there.
func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m delivery.Module, module string) {
	raw, err := b.step(ctx, l.module, l.node, "fetch", map[string]any{"repository": repository, "commit": m.Commit})
	if err != nil {
		l.err = err
		return
	}
	var fetched struct {
		Workspace string `json:"workspace"`
	}
	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
	}
	for _, t := range l.tasks {
		raw, err := b.step(ctx, l.module, l.node, "build", map[string]any{"workspace": fetched.Workspace, "module": m.Path, "artifact": t.Artifact})
		if err != nil {
			l.err = err
			return
		}
		var built struct {
			Digest string `json:"digest"`
		}
		if err := json.Unmarshal(raw, &built); err != nil || built.Digest == "" {
			l.err = fmt.Errorf("the build of %s of %s on %s answered no digest: %s", t.Artifact, module, l.node, raw)
			return
		}
		l.digests[t.Artifact] = built.Digest
	}
}