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

package check

import (
	"context"
	"errors"
	"fmt"
	"slices"
	"strings"
	"sync"
	"time"

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

// DefaultBound is how long a check may run before it is ended failed naming the step it was stuck at (ADR 0484, issue 1678).
const DefaultBound = 30 * time.Minute

// NodesWords is in the Words of a failure that is the build nodes' (none qualifies, the toolchains disagree): such a verdict
// is judged again when the nodes change, by a watch and never a timer (ADR 0437 §8).
const NodesWords = "build node qualifies"

// UnheardWords end the Words of a failure because a part of the mesh the check reads was not heard (the forge, issue 1624):
// like a failure of the nodes, it is not the change's, and is judged again when the nodes change.
const UnheardWords = "not heard; judged again when the nodes change"

// The events the checker says on its stream checks (module.json): a preview or a status was written, so a part that reads
// the bucket, which is made when the checker is assigned and may not exist when that part starts, finds it (ADR 0437 §8: no
// timer). The body is {"key": ...}.
const (
	EventPreviewWritten = "preview.written"
	EventOwedWritten    = "owed.written"
)

// Announcer says an event of the checker's stream, with the key written.
type Announcer func(ctx context.Context, event, key string)

// Outcome is what judging a case gives: the verdict (state, failure, levels, times, composition; the checker adds the
// group, the heads and the trunk) and, for the line, what the check knows of the builds of the modules it touches.
type Outcome struct {
	Verdict delivery.Verdict
	Builds  []BuildNote
	// Slowest names the slowest tests of a level that went over its bound ("unit pkg.Test 1.2s").
	Slowest []string
	// Judged says, by pull request (pullKey), which program judged a change to a build module (ADR 0535 §5); a pull request
	// that changes none is not named.
	Judged map[string]string
}

// BuildNote is the state of a candidate's module build, by module: built, failed or waiting (ADR 0484 clause 7).
type BuildNote struct {
	Module string
	State  string
}

// Judge runs the check of a case: the merge, the levels. It writes what it is doing to c.Progress. It answers an error
// only when it could not judge at all; a failure of the change is a Verdict.
type Judge interface {
	Judge(ctx context.Context, c *Case) (Outcome, error)
}

// Checker judges the candidates the indexer writes, dark (ADR 0453 clause 4, ADR 0484): for each candidate it claims a
// verdict key, runs the check through its Judge, writes the verdict and, for each pull request judged, the owed status
// `mesh/check` the forge bridge performs. It holds no timer; a check starts from a candidate or from a change of the
// nodes, the trunk or the controller's plan, and ends when its judge does or its bound passes.
type Checker struct {
	Verdicts KV
	Owed     KV
	Judge    Judge
	Trunks   interface{ Heads() map[string]string }
	// Bound is how long one check may run; DefaultBound when 0.
	Bound time.Duration
	// Limit is how many checks run at once; two when 0. A freed place goes to the first waiting check by rank (ADR 0536 §4).
	Limit int
	// Hold says whether a delivery builds, so no new check starts (ADR 0484 clause 12); nil never holds.
	Hold func() bool
	// Link is the base of the address a status points at; the verdict key follows.
	Link string
	// Announce says a status was written (EventOwedWritten); nil says nothing.
	Announce Announcer
	Now      func() time.Time
	Logf     func(format string, args ...any)

	once      sync.Once
	limit     int
	running   int       // checks holding a place
	waiting   []*waiter // checks waiting for a place, in the order they asked
	seq       uint64
	wg        sync.WaitGroup
	mu        sync.Mutex
	known     map[string]delivery.Candidate // by candidates key
	inflight  map[string]bool               // by verdict key
	again     map[string]bool               // by verdict key: asked again while it ran
	tried     map[string]bool               // by verdict key: looked at since this process started
	epoch     uint64                        // changes of the nodes seen since this process started
	nodesFail map[string]uint64             // by verdict key: the epoch a failure of the nodes was written at
}

func (ch *Checker) init() {
	ch.once.Do(func() {
		limit := ch.Limit
		if limit <= 0 {
			limit = 2
		}
		ch.limit = limit
		ch.known, ch.inflight, ch.again, ch.tried = map[string]delivery.Candidate{}, map[string]bool{}, map[string]bool{}, map[string]bool{}
		ch.nodesFail = map[string]uint64{}
	})
}

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

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

func (ch *Checker) bound() time.Duration {
	if ch.Bound > 0 {
		return ch.Bound
	}
	return DefaultBound
}

// Wait returns when every check started so far has ended.
func (ch *Checker) Wait() { ch.wg.Wait() }

// Candidate takes one value of the indexer's candidates bucket in (the key is the group) and checks it if it is to be
// checked. A value that is not a candidate is said and left.
func (ch *Checker) Candidate(ctx context.Context, key string, value []byte) {
	ch.init()
	c, err := delivery.DecodeCandidate(value)
	if err != nil {
		ch.logf("candidates.%s holds what is not a candidate, left: %v", key, err)
		return
	}
	ch.mu.Lock()
	ch.known[key] = c
	ch.mu.Unlock()
	ch.consider(ctx, key, c)
}

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

// 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 check that failed because no node qualified is run again only after such a change.
func (ch *Checker) NodesChanged(ctx context.Context) {
	ch.init()
	ch.mu.Lock()
	ch.epoch++
	ch.mu.Unlock()
	ch.Reconsider(ctx)
}

// Reconsider looks at every candidate again: the trunk moved, the nodes changed or a delivery ended. A candidate whose
// verdict stands for the trunk it was judged against is left.
func (ch *Checker) Reconsider(ctx context.Context) {
	ch.init()
	ch.mu.Lock()
	known := make(map[string]delivery.Candidate, len(ch.known))
	for k, c := range ch.known {
		known[k] = c
	}
	ch.mu.Unlock()
	for k, c := range known {
		ch.consider(ctx, k, c)
	}
}

func (ch *Checker) heads() map[string]string {
	if ch.Trunks == nil {
		return nil
	}
	return ch.Trunks.Heads()
}

func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidate) {
	cs := CaseOf(c, ch.heads())
	if cs == nil {
		return // nothing to judge: hq's pull requests, or no trunk known
	}
	if ch.Hold != nil && ch.Hold() {
		ch.logf("%s: not started, a delivery builds", cs.Key)
		return // started when the plan bucket empties (Reconsider)
	}
	cs.Queued = ch.now()
	ch.mu.Lock()
	if ch.inflight[cs.Key] {
		ch.again[cs.Key] = true // looked at once more when the check ends, so a change of the trunk is never lost
		ch.mu.Unlock()
		return
	}
	ch.inflight[cs.Key] = true
	ch.mu.Unlock()
	ch.wg.Add(1)
	go func() {
		defer ch.wg.Done()
		for {
			ch.admit(cs, ch.kindOf(ctx, cs))
			ch.try(ctx, key, cs)
			ch.leave()
			ch.mu.Lock()
			more := ch.again[cs.Key]
			delete(ch.again, cs.Key)
			if !more || ctx.Err() != nil {
				delete(ch.inflight, cs.Key)
				ch.mu.Unlock()
				return
			}
			ch.mu.Unlock()
			if next := CaseOf(ch.candidate(key, c), ch.heads()); next != nil {
				cs = next
			}
			cs.Queued = ch.now() // asked again: it waits for a place by its rank like any other
		}
	}()
}

// The kinds of a check waiting for a place, by the rank they start in (ADR 0536 §4): a head never judged and a head asked
// again after a failure of the nodes first, then a head asked again only because its trunk moved.
const (
	KindNew   = "new"
	KindNodes = "nodes"
	KindTrunk = "trunk"
)

var kindRank = map[string]int{KindNew: 0, KindNodes: 0, KindTrunk: 1}

// waiter is a check waiting for a place.
type waiter struct {
	group, key, kind string
	queued           time.Time
	seq              uint64
	ready            chan struct{}
}

// Waiting names a check waiting for a place: its group, its verdict key, its kind and when it was asked.
type Waiting struct {
	Group  string    `json:"group"`
	Key    string    `json:"key"`
	Kind   string    `json:"kind"`
	Queued time.Time `json:"queued"`
}

// Waiting are the checks waiting for a place, in the order they would start.
func (ch *Checker) Waiting() []Waiting {
	ch.init()
	ch.mu.Lock()
	defer ch.mu.Unlock()
	ws := slices.Clone(ch.waiting)
	slices.SortStableFunc(ws, before)
	out := make([]Waiting, len(ws))
	for i, w := range ws {
		out[i] = Waiting{Group: w.group, Key: w.key, Kind: w.kind, Queued: w.queued}
	}
	return out
}

// before orders waiting checks: the lower rank, then the one asked first, then the one that asked for a place first.
func before(a, b *waiter) int {
	if ra, rb := kindRank[a.kind], kindRank[b.kind]; ra != rb {
		return ra - rb
	}
	if c := a.queued.Compare(b.queued); c != 0 {
		return c
	}
	return int(a.seq) - int(b.seq)
}

// kindOf is the kind of a case's check, read from its verdict key: none (or one left running) is a head never judged; a
// failure of the nodes is asked again after the nodes changed; a verdict on other trunk heads is a re-judge after a trunk
// move. A verdict that stands is ranked with the re-judges: its claim finds nothing to do.
func (ch *Checker) kindOf(ctx context.Context, cs *Case) string {
	value, _, found, err := ch.Verdicts.Get(ctx, cs.Key)
	if err != nil || !found {
		return KindNew
	}
	v, err := delivery.DecodeVerdict(value)
	switch {
	case err != nil || v.State == delivery.VerdictRunning:
		return KindNew
	case IsNodesFailure(v):
		return KindNodes
	}
	return KindTrunk
}

// admit returns when the check holds a place: at once when one is free and no check waits, else when a check's end hands its
// place to this one as the first by rank. No timer: a check's end wakes the next (leave).
func (ch *Checker) admit(cs *Case, kind string) {
	ch.mu.Lock()
	if ch.running < ch.limit && len(ch.waiting) == 0 {
		ch.running++
		ch.mu.Unlock()
		return
	}
	ch.seq++
	w := &waiter{group: cs.Group, key: cs.Key, kind: kind, queued: cs.Queued, seq: ch.seq, ready: make(chan struct{})}
	ch.waiting = append(ch.waiting, w)
	ch.handOn()
	ch.mu.Unlock()
	<-w.ready
}

// leave frees the check's place and hands it to the first waiting check by rank.
func (ch *Checker) leave() {
	ch.mu.Lock()
	ch.running--
	ch.handOn()
	ch.mu.Unlock()
}

// handOn gives each free place to the first waiting check by rank; it is called with mu held.
func (ch *Checker) handOn() {
	for ch.running < ch.limit && len(ch.waiting) > 0 {
		first := 0
		for i := range ch.waiting {
			if before(ch.waiting[i], ch.waiting[first]) < 0 {
				first = i
			}
		}
		w := ch.waiting[first]
		ch.waiting = slices.Delete(ch.waiting, first, first+1)
		ch.running++
		close(w.ready)
	}
}

func (ch *Checker) candidate(key string, fallback delivery.Candidate) delivery.Candidate {
	ch.mu.Lock()
	defer ch.mu.Unlock()
	if c, ok := ch.known[key]; ok {
		return c
	}
	return fallback
}

// claimed is a verdict key the checker holds: the revision the next write is made at.
type claimed struct{ rev uint64 }

// claim takes the check of the case, or answers nil when it is not this checker's to run now:
//   - no key: it is created running; a create that finds the key made meanwhile gets nothing;
//   - a key left running is taken over as it stands when this process looks at it for the first time (it was left by the
//     checker that stopped; there is one instance, as there is one builder per node, ADR 0478 clause 6), and left otherwise;
//   - a pass or a failure for the trunk heads the case has stands; one for other trunk heads (the trunk moved) is taken
//     again by compare-and-set on the same key, and so is a failure of the nodes after the nodes changed.
func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64) (*claimed, error) {
	running := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictRunning}
	data, err := delivery.EncodeVerdict(running)
	if err != nil {
		return nil, err
	}
	value, rev, found, err := ch.Verdicts.Get(ctx, cs.Key)
	if err != nil {
		return nil, fmt.Errorf("%s: %w", cs.Key, err)
	}
	if !found {
		rev, err := ch.Verdicts.Create(ctx, cs.Key, data)
		if errors.Is(err, ErrConflict) {
			return nil, nil // another checker made it first
		}
		if err != nil {
			return nil, fmt.Errorf("%s: %w", cs.Key, err)
		}
		return &claimed{rev}, nil
	}
	cur, err := delivery.DecodeVerdict(value)
	if err != nil {
		return nil, fmt.Errorf("%s holds what is not a verdict: %w", cs.Key, err)
	}
	retake := false
	switch {
	case cur.State == delivery.VerdictRunning:
		if first {
			return &claimed{rev}, nil
		}
		return nil, nil
	case !sameTrunk(cur.Trunk, cs.Trunk):
		retake = true
	case cur.State == delivery.VerdictFail && IsNodesFailure(cur):
		ch.mu.Lock()
		at, ok := ch.nodesFail[cs.Key]
		ch.mu.Unlock()
		retake = !ok || epoch > at
	}
	if !retake {
		return nil, nil
	}
	rev, err = ch.Verdicts.Update(ctx, cs.Key, data, rev)
	if errors.Is(err, ErrConflict) {
		return nil, nil
	}
	if err != nil {
		return nil, fmt.Errorf("%s: %w", cs.Key, err)
	}
	return &claimed{rev}, nil
}

// IsNodesFailure says whether a verdict failed because no build node could take the check.
func IsNodesFailure(v delivery.Verdict) bool {
	return v.State == delivery.VerdictFail && (strings.Contains(v.Failed.Words, NodesWords) || strings.Contains(v.Failed.Words, UnheardWords))
}

// try claims the check, judges it within its bound and writes the verdict and the owed statuses. A check that ends
// because ctx did is left running: the checker takes it again when it starts.
func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
	ch.mu.Lock()
	first := !ch.tried[cs.Key]
	ch.tried[cs.Key] = true
	epoch := ch.epoch
	ch.mu.Unlock()
	cl, err := ch.claim(ctx, cs, first, epoch)
	if err != nil {
		ch.logf("%s: %v", cs.Key, err)
		return
	}
	if cl == nil {
		return
	}
	began := ch.now()
	ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{})
	out := ch.judge(ctx, cs)
	if ctx.Err() != nil {
		return
	}
	v := out.Verdict
	v.Group, v.Heads, v.Trunk = cs.Group, cs.Heads, cs.Trunk
	if v.Took.Queue == 0 && !cs.Queued.IsZero() && began.After(cs.Queued) {
		v.Took.Queue = began.Sub(cs.Queued).Milliseconds()
	}
	data, err := delivery.EncodeVerdict(v)
	if err != nil {
		ch.logf("%s: the verdict cannot be written: %v", cs.Key, err)
		v = delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictFail,
			Failed: delivery.FailedFirst{Machine: "checker", Words: "the verdict could not be written: " + err.Error()}}
		if data, err = delivery.EncodeVerdict(v); err != nil {
			return
		}
	}
	if _, err := ch.Verdicts.Update(ctx, cs.Key, data, cl.rev); err != nil {
		ch.logf("%s: the verdict could not be written: %v", cs.Key, err)
		return
	}
	if IsNodesFailure(v) {
		ch.mu.Lock()
		ch.nodesFail[cs.Key] = epoch
		ch.mu.Unlock()
	}
	if ch.movedSince(cs) {
		// Judged on a trunk that moved while it ran: the verdict is kept as judged, and the status stays pending until the
		// check of the new trunk, which Reconsider starts, ends.
		ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{})
		ch.logf("%s: %s on a trunk that moved while it ran; the status stays pending", cs.Key, v.State)
		return
	}
	ch.writeOwed(ctx, cs, v, out)
	ch.logf("%s: %s", cs.Key, v.State)
}

// judge runs the case within the bound. A judge that fails to judge, and a check that ran past its bound, are verdicts of
// failure naming the machine and the step.
func (ch *Checker) judge(ctx context.Context, cs *Case) Outcome {
	out, _ := judgeWithin(ctx, ch.Judge, cs, ch.bound())
	return out
}

var errBound = errors.New("the check ran past its bound")

func failure(machine, words string) delivery.Verdict {
	return delivery.Verdict{State: delivery.VerdictFail, Failed: delivery.FailedFirst{Machine: machine, Words: words}}
}

// writeOwed writes, for each pull request judged, the status the forge bridge performs: pending while the check runs, success
// or failure at its end (ADR 0484 clause 2). A key already holding the same value is not written again, so the bridge
// performs one act for each change; an older head's key is never touched.
func (ch *Checker) writeOwed(ctx context.Context, cs *Case, v delivery.Verdict, out Outcome) {
	for _, p := range cs.Pulls {
		key, err := delivery.CheckerOwedKey(p.Repository, p.Number, p.Head)
		if err != nil {
			ch.logf("%s: %v", cs.Key, err)
			continue
		}
		state := map[string]string{delivery.VerdictRunning: "pending", delivery.VerdictPass: "success", delivery.VerdictFail: "failure"}[v.State]
		o := delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state,
			Target: strings.TrimSuffix(ch.Link, "/") + "/" + cs.Key, Text: Line(v, cs, p, out.Builds, out.Slowest, out.Judged)}
		data, err := delivery.EncodeCheckerOwed(o)
		if err != nil {
			ch.logf("%s: the status of %s cannot be written: %v", cs.Key, key, err)
			continue
		}
		wrote, err := putOnce(ctx, ch.Owed, key, data, func(cur []byte) bool {
			held, err := delivery.DecodeCheckerOwed(cur)
			o.V = delivery.Version
			return err == nil && delivery.SameOwed(held, o)
		})
		if err != nil {
			ch.logf("%s: the status of %s cannot be written: %v", cs.Key, key, err)
		} else if wrote && ch.Announce != nil {
			ch.Announce(ctx, EventOwedWritten, key)
		}
	}
}

// putOnce writes data under key by create, or by compare-and-set when the key exists, and writes nothing when same says the
// key holds the value already.
func putOnce(ctx context.Context, kv KV, key string, data []byte, same func(cur []byte) bool) (bool, error) {
	for try := 0; try < 4; try++ {
		cur, rev, found, err := kv.Get(ctx, key)
		if err != nil {
			return false, err
		}
		if found && same(cur) {
			return false, nil
		}
		if !found {
			_, err = kv.Create(ctx, key, data)
		} else {
			_, err = kv.Update(ctx, key, data, rev)
		}
		if errors.Is(err, ErrConflict) {
			continue
		}
		return err == nil, err
	}
	return false, fmt.Errorf("%s changed under every one of 4 tries", key)
}

// Sweep closes, at start, every verdict left running for a candidate that no longer holds those heads: nothing will check
// them again, and the key would stay running for ever (the builder's Sweep, issue 1660). Call it once the candidates have been
// read. A key of heads a candidate holds now is not touched: the check takes it over.
func (ch *Checker) Sweep(ctx context.Context) error {
	ch.init()
	keys, err := ch.Verdicts.Keys(ctx)
	if err != nil {
		return err
	}
	ch.mu.Lock()
	current := map[string]bool{}
	for _, c := range ch.known {
		if k, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash); err == nil {
			current[k] = true
		}
	}
	ch.mu.Unlock()
	for _, key := range keys {
		if current[key] {
			continue
		}
		value, rev, found, err := ch.Verdicts.Get(ctx, key)
		if err != nil || !found {
			continue
		}
		cur, err := delivery.DecodeVerdict(value)
		if err != nil || cur.State != delivery.VerdictRunning {
			continue
		}
		cur.State = delivery.VerdictFail
		cur.Failed = delivery.FailedFirst{Machine: "checker", Words: "closed at start: left running, and no candidate holds these heads any more, so nothing will check them"}
		data, err := delivery.EncodeVerdict(cur)
		if err != nil {
			return err
		}
		if _, err := ch.Verdicts.Update(ctx, key, data, rev); err != nil && !errors.Is(err, ErrConflict) {
			return fmt.Errorf("%s: %w", key, err)
		}
		ch.logf("%s: closed, left running for heads no candidate holds", key)
	}
	return nil
}

// TrunkMoved is Reconsider after a move of a trunk, which first owes mesh/check pending on every open pull request of a
// candidate whose verdict was judged on other trunk heads, every member's for a group (the lead's decision L4, the checker's
// part of issue 1302): the forge's success of the old trunk never stands while the check waits for a place or for a delivery
// that holds it, so the gate never reads it as the head's.
func (ch *Checker) TrunkMoved(ctx context.Context) {
	ch.init()
	ch.mu.Lock()
	known := make(map[string]delivery.Candidate, len(ch.known))
	for k, c := range ch.known {
		known[k] = c
	}
	ch.mu.Unlock()
	heads := ch.heads()
	for _, c := range known {
		cs := CaseOf(c, heads)
		if cs == nil {
			continue
		}
		value, _, found, err := ch.Verdicts.Get(ctx, cs.Key)
		if err != nil || !found {
			continue // never judged: no status to put out of date
		}
		if v, err := delivery.DecodeVerdict(value); err == nil && sameTrunk(v.Trunk, cs.Trunk) {
			continue
		}
		ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{})
	}
	ch.Reconsider(ctx)
}

// movedSince says whether a trunk the case was judged on is no longer the trunk's head.
func (ch *Checker) movedSince(cs *Case) bool {
	now := ch.heads()
	for repo, head := range cs.Trunk {
		if now[repo] != head {
			return true
		}
	}
	return false
}