jschoubben
7abe230701
mesh-controller's try asked build-agent, which leaves the checking at R4e. A try is now a key in the checker's state tries: the checker judges the head merged onto its trunk as it judges a pull request's head, through the same judge, places and holds, writes its end in the key and owes the forge nothing (no status, no preview, no verdict). One try per head, at most four at once; hq and a repository whose trunk the gate does not name are refused with "no level of <repo> is checked by the checker".
500 lines
17 KiB
Go
500 lines
17 KiB
Go
package check
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"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"
|
|
|
|
// 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
|
|
}
|
|
|
|
// 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.
|
|
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
|
|
sem chan struct{}
|
|
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.sem = make(chan struct{}, 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()
|
|
ch.sem <- struct{}{}
|
|
defer func() { <-ch.sem }()
|
|
for {
|
|
ch.try(ctx, key, cs)
|
|
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
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
// 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}, nil, nil)
|
|
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}, nil, nil)
|
|
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.Builds, out.Slowest)
|
|
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, builds []BuildNote, slowest []string) {
|
|
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, builds, slowest)}
|
|
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}, nil, nil)
|
|
}
|
|
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
|
|
}
|