mesh-delivery/modules/mesh-delivery-checker/internal/check/judge.go

package check

import (
	"context"
	"encoding/json"
	"fmt"
	"path"
	"sort"
	"strings"
	"time"

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

// goModule is the build module the Go repositories are judged by.
const goModule = buildrun.GoBuildModule

// verdictPtr is a verdict of failure a step of the check ends with.
type verdictPtr = *delivery.Verdict

func failed(machine, words string) verdictPtr { return fail(machine, words) }

// Steps runs a build module's long step on a node and answers its result (buildrun.Runner).
type Steps interface {
	Run(ctx context.Context, module, node, tool string, args map[string]any) (json.RawMessage, error)
}

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

// BuildModules says which build modules judge a repository: the Go one always, the Node one when the repository holds a
// TypeScript module.
type BuildModules interface {
	BuildModules(repository string) []string
}

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

// Examiner judges a case through the build modules (ADR 0484 clauses 3, 4, 10, 13): for each pull request it fetches the head
// merged onto its trunk head on a build node, then runs the unit level and the integration level, taking the kept verdicts
// of unchanged packages, running the change's own tests first and stopping at the first failure.
type Examiner struct {
	Steps      Steps
	Toolchains ToolchainEntries
	// Presence says which nodes are online; Health whether a node's build module is running and healthy. A node that is not
	// is not chosen; nil means every node is.
	Presence interface{ Online(node string) bool }
	Health   interface {
		Healthy(node, module string) bool
	}
	// Places holds a step back while its node is at the checker's capacity (ADR 0484 clause 12).
	Places *buildrun.Places
	Kept   Kept
	// Previews is the previews bucket; nil writes none.
	Previews KV
	Modules  BuildModules
	// Runs says which commit the mesh runs of a module (the indexer's modules); nil knows none.
	Runs interface{ CommitOf(module string) string }
	// Forge reads the go.mod files of the Go members and says whether a mesh-sdk commit is on its trunk (ADR 0484 clause 9);
	// nil checks no pin.
	Forge Forge
	// Builds says how the build of a fingerprint the candidate carries stands (the builder's builds bucket); nil says nothing
	// of the builds.
	Builds BuildStates
	// Announce says a preview was written under the key (EventPreviewWritten); nil says nothing.
	Announce Announcer
	// Trunks are the trunk heads of every repository, for the siblings a repository's tests read.
	Trunks interface{ Heads() map[string]string }
	Now    func() time.Time
	Logf   func(format string, args ...any)
}

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

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

// target is a pull request on one build module: the node chosen, the workspace its fetch made and what it said.
type target struct {
	pull      Pull
	module    string
	node      string
	toolchain string
	fetched   Fetched
	// priority are the Go packages (import paths) or the Node module directories whose test files the merge changes.
	changed []string
	// beside says the workspace was given siblings: its tests may read what the digest does not cover, so nothing is kept for it.
	beside bool
	plans  map[string]*goPlan // by level
}

// goPlan is what a Go target has to run at one level: the digests, the packages without a kept verdict and the tests taken.
type goPlan struct {
	digests map[string]string
	todo    []string
	// all says the digest step gave nothing to key a kept verdict by, so the level runs whole and keeps nothing.
	all bool
}

// acc collects a level's counts across targets.
type acc struct {
	run, kept, skipped int
	elapsed, setup     float64
	over               bool
	slowest            []TestTime
}

// state is one judging.
type state struct {
	j       *Examiner
	c       *Case
	targets []*target
	levels  map[string]*acc
	nodes   map[string]bool
	clone   time.Duration
	slow    []string
}

// Judge runs the case (it is the Judge of a Checker). A failure of the change, of a merge or of a step is a verdict; an error is only a case that cannot
// be judged at all.
func (j *Examiner) Judge(ctx context.Context, c *Case) (Outcome, error) {
	st := &state{j: j, c: c, levels: map[string]*acc{delivery.LevelUnit: {}, delivery.LevelIntegration: {}}, nodes: map[string]bool{}}
	if v := st.fetchAll(ctx); v != nil {
		return st.finish(ctx, *v), nil
	}
	if v := st.checkPins(ctx); v != nil {
		return st.finish(ctx, *v), nil
	}
	for _, level := range []string{delivery.LevelUnit, delivery.LevelIntegration} {
		for _, priority := range []bool{true, false} {
			for _, t := range st.targets {
				if v := st.runTarget(ctx, t, level, priority); v != nil {
					return st.finish(ctx, *v), nil
				}
			}
		}
	}
	return st.finish(ctx, delivery.Verdict{State: delivery.VerdictPass}), nil
}

// finish is the outcome of the verdict with the state of the builds the candidate's modules carry.
func (st *state) finish(ctx context.Context, v delivery.Verdict) Outcome {
	out := st.outcome(v)
	out.Builds = st.j.buildNotes(ctx, st.c)
	return out
}

// outcome fills the verdict's levels, times, machines and composition from what ran.
func (st *state) outcome(v delivery.Verdict) Outcome {
	if v.State == delivery.VerdictPass || len(v.Levels) == 0 {
		v.Levels = map[string]delivery.LevelResult{}
		for level, a := range st.levels {
			v.Levels[level] = delivery.LevelResult{Run: a.run, Kept: a.kept, Skipped: a.skipped}
		}
	}
	ms := func(s float64) int64 { return int64(s * 1000) }
	u, i := st.levels[delivery.LevelUnit], st.levels[delivery.LevelIntegration]
	v.Took = delivery.Took{Clone: st.clone.Milliseconds(), Unit: ms(u.elapsed), IntegrationSetup: ms(i.setup), IntegrationRun: ms(i.elapsed)}
	for node := range st.nodes {
		v.Machines = append(v.Machines, node)
	}
	sort.Strings(v.Machines)
	for _, level := range []string{delivery.LevelUnit, delivery.LevelIntegration} {
		if st.levels[level].over {
			v.OverBound = append(v.OverBound, level)
		}
	}
	// The composition is the assigner's check (issue 1297): until it delivers a verdict says it was not run (ADR 0484 clause 6).
	v.Composition = &delivery.Composition{State: delivery.CompositionNotRun}
	var slow []string
	for _, level := range []string{delivery.LevelUnit, delivery.LevelIntegration} {
		a := st.levels[level]
		if !a.over {
			continue
		}
		sort.SliceStable(a.slowest, func(x, y int) bool { return a.slowest[x].Elapsed > a.slowest[y].Elapsed })
		for n, s := range a.slowest {
			if n == 3 {
				break
			}
			slow = append(slow, fmt.Sprintf("%s %s %.1fs", level, path.Base(s.Package)+"."+s.Test, s.Elapsed))
		}
	}
	return Outcome{Verdict: v, Slowest: slow}
}

// fail is a verdict of failure at a machine.
func fail(machine, words string) *delivery.Verdict {
	v := failure(machine, words)
	return &v
}

// step runs one step of a build module on the node the target is on, holding a place there while it runs.
func (st *state) step(ctx context.Context, t *target, tool, label string, args map[string]any) (json.RawMessage, error) {
	st.c.Progress.Set(label, t.node)
	release, err := st.j.Places.Acquire(ctx, t.node)
	if err != nil {
		return nil, err
	}
	defer release()
	return st.j.Steps.Run(ctx, t.module, t.node, toolPrefix[t.module]+tool, args)
}

// choose picks the build node for a pull request's repository on a build module.
func (st *state) choose(bm string) (node, toolchain string, err error) {
	language := buildrun.LanguageOf[bm]
	qualifies := func(node string) bool {
		return (st.j.Presence == nil || st.j.Presence.Online(node)) && (st.j.Health == nil || st.j.Health.Healthy(node, bm))
	}
	nodes, tc, err := buildrun.ChooseNodes(language, st.j.Toolchains.Entries(language), st.j.Places, qualifies, 1)
	if err != nil {
		return "", "", err
	}
	return nodes[0], tc.Version, nil
}

// fetchAll fetches every pull request's head merged onto its trunk on a build node of each build module the repository
// needs. A head that does not merge, or a step that fails, ends the check at once, running no test (ADR 0484 clause 13).
func (st *state) fetchAll(ctx context.Context) *delivery.Verdict {
	began := st.j.now()
	defer func() { st.clone = st.j.now().Sub(began) }()
	for _, p := range st.c.Pulls {
		var merged []Fetched
		for _, bm := range st.j.Modules.BuildModules(p.Repository) {
			node, tc, err := st.choose(bm)
			if err != nil {
				return fail(bm, err.Error())
			}
			t := &target{pull: p, module: bm, node: node, toolchain: tc, plans: map[string]*goPlan{}}
			st.nodes[node] = true
			args := map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number), "base": p.Trunk}
			if bm == buildrun.GoBuildModule {
				beside, err := st.j.besideFor(ctx, st.c, p)
				if err != nil {
					return fail(node, err.Error())
				}
				if len(beside) > 0 {
					args["beside"] = beside
					t.beside = true
				}
			}
			raw, err := st.step(ctx, t, "fetch", fmt.Sprintf("fetch %s#%d on %s", p.Repository, p.Number, node), args)
			if err != nil {
				return fail(node, fmt.Sprintf("fetch of %s#%d: %v%s", p.Repository, p.Number, err, pinHint(st.c, err.Error())))
			}
			if err := json.Unmarshal(raw, &t.fetched); err != nil || t.fetched.Workspace == "" {
				return fail(node, fmt.Sprintf("the fetch of %s#%d answered no workspace: %s", p.Repository, p.Number, raw))
			}
			if len(t.fetched.Conflicts) > 0 {
				st.j.writePreview(ctx, p, append(merged, t.fetched))
				return fail(node, fmt.Sprintf("%s#%d does not merge into the trunk: files in conflict: %s", p.Repository, p.Number, strings.Join(t.fetched.Conflicts, ", ")))
			}
			t.changed = t.fetched.Changed
			merged = append(merged, t.fetched)
			st.targets = append(st.targets, t)
		}
		st.j.writePreview(ctx, p, merged)
	}
	return nil
}

// isTestFile says whether a repository-relative path is a test file of the languages the checker judges: Go's _test.go and
// the TypeScript module's .test. and .spec. files and test directories.
func isTestFile(p string) bool {
	switch {
	case strings.HasSuffix(p, "_test.go"):
		return true
	case strings.Contains(p, ".test.") || strings.Contains(p, ".spec."):
		return true
	}
	for _, seg := range strings.Split(path.Dir(p), "/") {
		if seg == "__tests__" || seg == "test" || seg == "tests" {
			return true
		}
	}
	return false
}

// changesTestOf says whether the merge changes a test file of the Go package with the import path pkg: the directory of the
// file and the import path agree on their last segments, whichever holds the module's directory and path in front.
func changesTestOf(changed []string, pkg string) bool {
	for _, f := range changed {
		if !strings.HasSuffix(f, "_test.go") {
			continue
		}
		dir := path.Dir(f)
		if dir == pkg || strings.HasSuffix(pkg, "/"+dir) || strings.HasSuffix(dir, "/"+pkg) {
			return true
		}
	}
	return false
}

// changesTestIn says whether the merge changes a test file inside the directory of a Node module.
func changesTestIn(changed []string, dir string) bool {
	for _, f := range changed {
		if !isTestFile(f) || strings.HasSuffix(f, "_test.go") {
			continue
		}
		if dir == "." || dir == "" || strings.HasPrefix(f, strings.TrimSuffix(dir, "/")+"/") {
			return true
		}
	}
	return false
}

// failedName names a failure: its package and test.
func failedName(f Failure) string {
	if f.Test == "" {
		return f.Package
	}
	return f.Package + "." + f.Test
}

// firstLine is the first non-empty line of a failure's output, shortened.
func firstLine(s string) string {
	for _, l := range strings.Split(s, "\n") {
		if l = strings.TrimSpace(l); l != "" {
			if len(l) > 120 {
				l = l[:120]
			}
			return l
		}
	}
	return ""
}

// failure turns a level that did not pass into a verdict naming what failed first.
func levelFailure(t *target, level string, res Tested) *delivery.Verdict {
	v := delivery.Verdict{State: delivery.VerdictFail}
	switch {
	case len(res.Failed) > 0:
		f := res.Failed[0]
		name := failedName(f)
		if t.module == buildrun.NodeBuildModule && f.Package == "" {
			name = res.Module
		}
		v.Failed = delivery.FailedFirst{Test: level + ": " + name, Words: firstLine(f.Output)}
	default:
		why := res.Why
		if why == "" {
			why = "the " + level + " level did not pass"
		}
		v.Failed = delivery.FailedFirst{Test: level + ": " + t.pull.Repository, Words: why}
	}
	return &v
}

// runTarget runs one level of a target: first the packages the change's own tests are in (priority), then the others.
func (st *state) runTarget(ctx context.Context, t *target, level string, priority bool) *delivery.Verdict {
	if t.module == buildrun.NodeBuildModule {
		return st.runNode(ctx, t, level, priority)
	}
	return st.runGo(ctx, t, level, priority)
}

// A level's work is planned once for a target: the digest of each package, the kept verdicts taken, the packages left to run.
func (st *state) planGo(ctx context.Context, t *target, level string) *delivery.Verdict {
	if t.plans[level] != nil {
		return nil
	}
	plan := &goPlan{digests: map[string]string{}}
	t.plans[level] = plan
	if t.beside {
		// The digest covers the package's own files, the toolchain and the header, not the siblings placed beside the tree: a
		// package whose tests read one must never be kept, nor taken from a verdict kept for other inputs (ADR 0444).
		plan.all = true
		return nil
	}
	raw, err := st.step(ctx, t, "digest", fmt.Sprintf("digest the %s level of %s#%d on %s", level, t.pull.Repository, t.pull.Number, t.node),
		map[string]any{"workspace": t.fetched.Workspace, "level": level})
	var d Digested
	if err == nil {
		err = json.Unmarshal(raw, &d)
	}
	if ctx.Err() != nil {
		return fail(t.node, ctx.Err().Error())
	}
	if err != nil || len(d.Packages) == 0 {
		// The build module named no digest (a level it cannot digest here): the level runs whole and keeps nothing.
		if err != nil {
			st.j.logf("%s#%d: no digest of the %s level, so it runs whole: %v", t.pull.Repository, t.pull.Number, level, err)
		}
		plan.all = true
		return nil
	}
	plan.digests = d.Packages
	a := st.levels[level]
	pkgs := make([]string, 0, len(d.Packages))
	for p := range d.Packages {
		pkgs = append(pkgs, p)
	}
	sort.Strings(pkgs)
	for _, p := range pkgs {
		k, found, err := st.j.Kept.Lookup(ctx, level, d.Packages[p])
		if err != nil {
			st.j.logf("the kept verdict of %s at %s cannot be read, so it runs: %v", p, level, err)
		}
		if found {
			a.kept += k.Tests
			continue
		}
		plan.todo = append(plan.todo, p)
	}
	return nil
}

func (st *state) runGo(ctx context.Context, t *target, level string, priority bool) *delivery.Verdict {
	if v := st.planGo(ctx, t, level); v != nil {
		return v
	}
	plan := t.plans[level]
	var run []string
	switch {
	case plan.all:
		if priority {
			return nil // nothing to put first: the level runs whole after
		}
	default:
		for _, p := range plan.todo {
			if changesTestOf(t.changed, p) == priority {
				run = append(run, p)
			}
		}
		if len(run) == 0 {
			return nil
		}
	}
	args := map[string]any{"workspace": t.fetched.Workspace, "level": level}
	if run != nil {
		args["packages"] = run
	}
	raw, err := st.step(ctx, t, "test", fmt.Sprintf("test the %s level of %s#%d on %s", level, t.pull.Repository, t.pull.Number, t.node), args)
	if err != nil {
		return fail(t.node, fmt.Sprintf("the %s level of %s#%d: %v", level, t.pull.Repository, t.pull.Number, err))
	}
	var res Tested
	if err := json.Unmarshal(raw, &res); err != nil {
		return fail(t.node, fmt.Sprintf("the %s level of %s#%d answered what is not a result: %v", level, t.pull.Repository, t.pull.Number, err))
	}
	a := st.levels[level]
	a.run += res.Tests
	a.skipped += res.Skipped
	a.elapsed += res.Elapsed
	a.setup += res.Setup
	a.over = a.over || res.OverBound
	a.slowest = append(a.slowest, res.Slowest...)
	if len(plan.digests) > 0 {
		if _, err := st.j.Kept.KeepPasses(ctx, t.pull.Repository, level, t.toolchain, res, plan.digests, st.j.now()); err != nil {
			st.j.logf("%s#%d: a passing package could not be kept: %v", t.pull.Repository, t.pull.Number, err)
		}
	}
	if !res.Passed {
		return levelFailure(t, level, res)
	}
	return nil
}

func (st *state) runNode(ctx context.Context, t *target, level string, priority bool) *delivery.Verdict {
	for _, dir := range t.fetched.Modules {
		if changesTestIn(t.changed, dir) != priority {
			continue
		}
		raw, err := st.step(ctx, t, "test", fmt.Sprintf("test the %s level of %s#%d (%s) on %s", level, t.pull.Repository, t.pull.Number, dir, t.node),
			map[string]any{"workspace": t.fetched.Workspace, "module": dir, "level": level})
		if err != nil {
			return fail(t.node, fmt.Sprintf("the %s level of %s#%d (%s): %v", level, t.pull.Repository, t.pull.Number, dir, err))
		}
		var res Tested
		if err := json.Unmarshal(raw, &res); err != nil {
			return fail(t.node, fmt.Sprintf("the %s level of %s#%d (%s) answered what is not a result: %v", level, t.pull.Repository, t.pull.Number, dir, err))
		}
		if res.Module == "" {
			res.Module = dir
		}
		a := st.levels[level]
		if !res.NoTests {
			a.run++ // a Node package is counted whole: its build module counts no tests (ADR 0484 clause 10)
		}
		a.elapsed += res.Elapsed
		a.setup += res.Setup
		a.over = a.over || res.OverBound
		if !res.Passed {
			return levelFailure(t, level, res)
		}
	}
	return nil
}