Files
mesh-delivery/modules/mesh-delivery-checker/internal/check/try.go
T
jschoubben d58b717243
mesh/delivery merged: source.main moved to novox/mesh-delivery@2d201f781ced
mesh/check being carried to d58b71724312 on 2d201f781ced | from 205a3b37501f
A try's end is said on a context its own stop does not cancel (issue 1921)
The end a run writes comes back on the key's watch, where Put stops the run's context and leaves the saying to the run, so the end was never said and the indexer kept the try's candidate. The group try's unit test now echoes the end on the watch while the run says it.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-06 10:33:49 +01:00

854 lines
33 KiB
Go

package check
import (
"context"
"encoding/json"
"errors"
"fmt"
"regexp"
"slices"
"sort"
"strconv"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// A try is a pushed head judged by the checker as a pull request's head is, before its pull request is opened (ADR 0501
// §5; issue 686's `mesh-controller.try`, which calls these tools from R4e): the head merged onto its trunk, the unit and the
// integration level through the build modules. It owes the forge nothing (no status, no preview) and writes no verdict; its
// state is the value of its key in the checker's state tries.
// The try's tools and state.
const (
TryTool = "checker_try"
TryReadTool = "checker_try_read"
TryEndTool = "checker_try_end"
// TriesTool lists the tries that wait or run (OpenTries).
TriesTool = "checker_tries"
// TriesState is the checker's state of tries: the bucket mesh-delivery-checker_tries, keys try-<unix nanoseconds>.
TriesState = "tries"
TriesBucket = CheckerModule + "_" + TriesState
// MaxTries and MaxTriesWaiting are checked when a try is asked, mesh-wide: a new try is refused while MaxTries tries run
// on a build node or MaxTriesWaiting wait for one (queued, or their asks not yet taken by a node). It is no bound on
// what runs later: tries let in while waiting may later all run at once, and each node's build capacity bounds what
// runs on it. A try waiting behind the seat's other asks holds no node, so it no longer blocks the next try (issue
// 1890: at 19:53 on 2026-10-26 four tries sat queued behind 44 checks for up to 47 minutes and no fifth could be asked).
MaxTries = 4
MaxTriesWaiting = 8
// TryScan is how many of the newest tries a new ask reads to find the ones that wait or run (issue 1827).
TryScan = 64
// TryKeep is how long an ended try's key is kept before the checker removes it.
TryKeep = 7 * 24 * time.Hour
)
// The states of a try: asked and waiting, running, and its end.
const (
TryQueued = "queued"
TryRunning = "running"
TryPass = "pass"
TryFail = "fail"
TryError = "error"
)
var (
tryHeadOK = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`)
tryTestsOK = regexp.MustCompile(`^[A-Za-z0-9_^$|().*+?/\[\]-]+$`)
tryRepoOK = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*$`)
tryTrunkOK = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]*$`)
)
// Try is the value of a tries key.
type Try struct {
V int `json:"v"`
ID string `json:"id"`
// Repository, Head, Trunk and Branch are the first member's (the try's headline, and all a try of one repository
// names); Members are every member of the try, by repository: each an exact commit merged onto a trunk commit (a
// group try, issue 1921). A try written before members read as one member.
Repository string `json:"repository"`
// Head is the commit tried; Trunk the trunk commit it is merged onto, and Branch the trunk's branch when it was asked by name.
Head string `json:"head"`
Trunk string `json:"trunk"`
Branch string `json:"branch,omitempty"`
Members []TryMember `json:"members,omitempty"`
// Tests is a go test -run pattern: only those tests run, and nothing is kept. Level is unit or integration: only that
// level runs. Empty is every test of both levels.
Tests string `json:"tests,omitempty"`
Level string `json:"level,omitempty"`
State string `json:"state"`
// Told is true from the moment a whole try's start is said on the checker's stream until its end is said there: kept in the
// try's own state, so an end that is written and not said (the checker stopped between) is said when the checker starts.
Told bool `json:"told,omitempty"`
// Waiting is what the running try waits for now (its group's candidate from the indexer, ADR 0639 §5), empty when it waits for nothing;
// try-read gives it, so a try that sits still says why.
Waiting string `json:"waiting,omitempty"`
// Said is the try's report in words: what failed first, the tests run, kept and skipped per level, the time of each part.
Said string `json:"said,omitempty"`
// Took is the time from its start to its end, as Go writes a duration.
Took string `json:"took,omitempty"`
// Failed is every test that failed, each with its first line, at most meshdeliverycheckerclient.MaxFailures and then how many more
// (issue 1921); or what else failed first (a conflict, a refused step).
Failed []string `json:"failed,omitempty"`
Levels map[string]meshdeliverychecker.VerdictsByGroupHeadsValueLevelsValue `json:"levels,omitempty"`
Machines []string `json:"machines,omitempty"`
// Composition is what the composition on every placed node came to (a whole try, not one narrowed to a level or to
// some tests), and Notes the words the check noted on its way (the manifests judged, the module graph, the kept
// verdicts taken): the same as a group's pull-request check reports.
Composition *meshdeliverychecker.VerdictsByGroupHeadsValueComposition `json:"composition,omitempty"`
Notes []string `json:"notes,omitempty"`
// Asks are the ids of the asks the try put on the build seats' streams, each ending on the step key <node>.<id> of its
// build module: the delivery view counts those steps as a try's, not a candidate's check (build_usage).
Asks []string `json:"asks,omitempty"`
// Steps say where each ask stands while the try runs and after: asked until a node takes it, then the node, its
// state, the phase its step says and when (issue 1890: a 31-minute try could not be told queued from running).
Steps []TryStep `json:"steps,omitempty"`
Asked time.Time `json:"asked"`
Started time.Time `json:"started,omitzero"`
Ended time.Time `json:"ended,omitzero"`
}
// TryMember is one repository of a try: the exact commit asked, never the branch's head now, and the trunk commit it is
// merged onto. Branch is the trunk's branch when the trunk was asked by name.
type TryMember struct {
Repository string `json:"repository"`
Head string `json:"head"`
Trunk string `json:"trunk"`
Branch string `json:"branch,omitempty"`
}
// MembersOf are the try's members: its Members, or the one its headline names.
func (t Try) MembersOf() []TryMember {
if len(t.Members) > 0 {
return t.Members
}
return []TryMember{{Repository: t.Repository, Head: t.Head, Trunk: t.Trunk, Branch: t.Branch}}
}
// TryAsked is a TryStep's state before a build node took its ask: it waits in the seat's stream.
const TryAsked = "asked"
// TryStep is one ask of a try as its step key last said it.
type TryStep struct {
Ask string `json:"ask"`
Node string `json:"node,omitempty"`
State string `json:"state"`
Phase string `json:"phase,omitempty"`
Started time.Time `json:"started,omitzero"`
PhaseAt time.Time `json:"phase-at,omitzero"`
Ended time.Time `json:"ended,omitzero"`
}
// tryWriter writes a running try's record again as its asks are put and their steps move: one writer per try, holding
// the revision it wrote last, so the try's end is written at that revision. A write refused (the try ended by hand)
// stops the writes; the end's own write then says it.
type tryWriter struct {
mu sync.Mutex
tr *Tryer
ctx context.Context
key string
t Try
rev uint64
stopped bool
}
// heard takes an ask put (step nil) or a value of its step key in, and writes the record when it changed.
func (w *tryWriter) heard(askID string, step *buildstep.Step) {
w.mu.Lock()
defer w.mu.Unlock()
if w.stopped {
return
}
changed := false
if !slices.Contains(w.t.Asks, askID) {
w.t.Asks, changed = append(w.t.Asks, askID), true
}
i := slices.IndexFunc(w.t.Steps, func(s TryStep) bool { return s.Ask == askID })
if i < 0 {
w.t.Steps, changed = append(w.t.Steps, TryStep{Ask: askID, State: TryAsked}), true
i = len(w.t.Steps) - 1
}
if step != nil {
now := TryStep{Ask: askID, Node: step.Node, State: step.State, Phase: step.Phase, Started: step.Started,
PhaseAt: step.PhaseAt, Ended: step.Ended}
changed = changed || now != w.t.Steps[i]
w.t.Steps[i] = now
}
if changed {
w.write()
}
}
// waiting writes what the try waits for now ("" when it waits for nothing).
func (w *tryWriter) waiting(why string) {
w.mu.Lock()
defer w.mu.Unlock()
if w.stopped || w.t.Waiting == why {
return
}
w.t.Waiting = why
w.write()
}
// write writes the record at the revision written last; a refusal stops the writes.
func (w *tryWriter) write() {
data, err := encodeTry(w.t)
if err != nil {
return
}
rev, err := w.tr.Tries.Update(w.ctx, w.key, data, w.rev)
if err != nil {
w.stopped = true
w.tr.logf("%s: its asks' progress is no longer written: %v", w.key, err)
return
}
w.rev = rev
}
// end stops the writes and answers the record and its revision, for the try's end.
func (w *tryWriter) end() (Try, uint64) {
w.mu.Lock()
defer w.mu.Unlock()
w.stopped = true
return w.t, w.rev
}
// HoldsANode says whether the try holds a build node now: it runs, and a step of its asks runs on a node, or it names
// no step (a checker that writes its asks only at the end, or one about to put its first).
func (t Try) HoldsANode() bool {
if t.State != TryRunning {
return false
}
if len(t.Steps) == 0 {
return true
}
return slices.ContainsFunc(t.Steps, func(s TryStep) bool { return s.State == buildstep.StepRunning })
}
// Over says whether the try came to its end.
func (t Try) Over() bool { return t.State == TryPass || t.State == TryFail || t.State == TryError }
// DecodeTry reads a tries value.
func DecodeTry(raw []byte) (Try, error) {
var t Try
if err := json.Unmarshal(raw, &t); err != nil {
return Try{}, fmt.Errorf("not a try: %w", err)
}
if !meshcatalogclient.IsTryGroup(t.ID) || t.Repository == "" || t.Head == "" || t.Trunk == "" {
return Try{}, errors.New("not a try: it names no id, repository, head or trunk")
}
t.Members = t.MembersOf()
return t, nil
}
func encodeTry(t Try) ([]byte, error) {
t.V = keys.Version
return json.Marshal(t)
}
// TryAsk is what checker_try is given.
type TryAsk struct {
Repository, Head, Trunk, Tests, Level string
// Members are the repositories of a group try, each an exact commit and, optionally, its trunk (a commit, or the
// trunk's branch for its head now); the single Repository, Head and Trunk above are one member. Both at once are refused.
Members []TryAskMember
}
// TryAskMember is one repository of a group try as it is asked.
type TryAskMember struct {
Repository, Head, Trunk string
}
// TryStore is the tries state as the tools see it.
type TryStore interface {
Keys() ([]string, error)
Get(key string) (json.RawMessage, error)
Create(key string, value []byte) error
}
// NotChecked is the error of a try of a repository the checker judges no level of: one whose trunk the gate's source does
// not name.
func NotChecked(repository string) error {
return fmt.Errorf("no level of %s is checked by the checker", repository)
}
// tryMember checks one member as it is asked and resolves its trunk: the gate's trunk head of its repository, a commit
// asked, or the trunk's branch (its head now). A repository with no trunk head is no repository the checker checks.
func tryMember(trunks map[string]string, a TryAskMember) (TryMember, error) {
repo := strings.ToLower(strings.TrimSuffix(strings.TrimSpace(a.Repository), ".git"))
head := strings.ToLower(strings.TrimSpace(a.Head))
switch {
case !tryRepoOK.MatchString(repo) || strings.Contains(repo, ".."):
return TryMember{}, fmt.Errorf("%q is not a repository, <owner>/<name>", a.Repository)
case !tryHeadOK.MatchString(head):
return TryMember{}, fmt.Errorf("the head %q of %s is not a full commit, 40 or 64 hex digits", a.Head, repo)
}
trunkHead := ""
for r, h := range trunks {
if strings.EqualFold(r, repo) {
trunkHead = h
}
}
if trunkHead == "" {
return TryMember{}, NotChecked(repo)
}
m := TryMember{Repository: repo, Head: head}
switch trunk := strings.TrimSpace(a.Trunk); {
case trunk == "":
m.Trunk = trunkHead
case tryHeadOK.MatchString(strings.ToLower(trunk)):
m.Trunk = strings.ToLower(trunk)
case tryTrunkOK.MatchString(trunk) && !strings.Contains(trunk, ".."):
// The checker judges pull requests into the trunk alone (ADR 0484 clause 11): a branch named is the trunk's head now.
m.Trunk, m.Branch = trunkHead, trunk
default:
return TryMember{}, fmt.Errorf("the trunk %q of %s is neither a full commit nor a branch", a.Trunk, repo)
}
return m, nil
}
// sameMembers says whether two tries ask the same commits onto the same trunk commits.
func sameMembers(a, b []TryMember) bool {
return slices.EqualFunc(a, b, func(x, y TryMember) bool {
return x.Repository == y.Repository && x.Head == y.Head && x.Trunk == y.Trunk
})
}
// AskTry writes a new try, queued, and answers its id: a try of the same commits that waits or runs is the answer to a second
// ask, and a new try is refused while MaxTries tries run on a build node or MaxTriesWaiting wait for one. A try is a group of
// one or more members (issue 1921), each an exact commit, never the branch's head now: a push after the ask changes nothing
// of what is tried. trunks are the gate's trunk heads of every repository (source.main): a repository with none is no
// repository the checker checks. A trunk asked by name is its trunk head now.
func AskTry(store TryStore, trunks map[string]string, a TryAsk, now time.Time) (string, error) {
asked := a.Members
if a.Repository != "" || a.Head != "" || a.Trunk != "" {
if len(asked) > 0 {
return "", errors.New("name the members, or one repository with its head, not both")
}
asked = []TryAskMember{{Repository: a.Repository, Head: a.Head, Trunk: a.Trunk}}
}
if len(asked) == 0 {
return "", errors.New("name what to try: members (repository and head each), or a repository and its head")
}
switch {
case a.Level != "" && a.Level != buildstep.LevelUnit && a.Level != buildstep.LevelIntegration:
return "", fmt.Errorf("the level %q is neither unit nor integration", a.Level)
case a.Tests != "" && !tryTestsOK.MatchString(a.Tests):
return "", fmt.Errorf("the tests %q are not a go test -run pattern", a.Tests)
}
var members []TryMember
for _, m := range asked {
tm, err := tryMember(trunks, m)
if err != nil {
return "", err
}
if slices.ContainsFunc(members, func(o TryMember) bool { return o.Repository == tm.Repository }) {
return "", fmt.Errorf("%s is named twice: a try holds one commit of a repository", tm.Repository)
}
members = append(members, tm)
}
slices.SortFunc(members, func(x, y TryMember) int { return strings.Compare(x.Repository, y.Repository) })
first := members[0]
t := Try{Repository: first.Repository, Head: first.Head, Trunk: first.Trunk, Branch: first.Branch, Members: members,
Tests: a.Tests, Level: a.Level, State: TryQueued, Asked: now.UTC()}
keys, err := store.Keys()
if err != nil {
return "", fmt.Errorf("the tries cannot be listed: %w", err)
}
// An ended try's key stays TryKeep, so the bucket holds thousands; one that waits or runs was asked lately. Only the
// newest TryScan keys are read, in parallel (a try id is the time in nanoseconds, so the newest sorts last).
sort.Slice(keys, func(i, j int) bool {
if len(keys[i]) != len(keys[j]) {
return len(keys[i]) > len(keys[j])
}
return keys[i] > keys[j]
})
if len(keys) > TryScan {
keys = keys[:TryScan]
}
tries := make([]Try, len(keys))
live := make([]bool, len(keys))
if err := parallel(len(keys), func(i int) error {
raw, err := store.Get(keys[i])
if err != nil || len(raw) == 0 {
return nil
}
cur, err := DecodeTry(raw)
tries[i], live[i] = cur, err == nil && !cur.Over()
return nil
}); err != nil {
return "", err
}
var holding, waiting []string
for i, cur := range tries {
if !live[i] {
continue
}
if sameMembers(cur.MembersOf(), t.Members) && cur.Tests == t.Tests && cur.Level == t.Level {
return cur.ID, nil
}
said := fmt.Sprintf("%s (%s, %s)", cur.ID, sayMembers(cur.MembersOf()), cur.State)
if cur.HoldsANode() {
holding = append(holding, said)
} else {
waiting = append(waiting, said)
}
}
// Checked at this ask only: a try queued behind the seat's checks holds no node; tries let in while waiting may later
// all run at once, and each node's build capacity bounds what runs on it.
if len(holding) >= MaxTries || len(waiting) >= MaxTriesWaiting {
return "", fmt.Errorf("a new try is refused while %d tries run on a build node or %d wait for one; now %d run and %d wait: running %s; waiting %s; ask again when one ends",
MaxTries, MaxTriesWaiting, len(holding), len(waiting), orNone(holding), orNone(waiting))
}
t.ID = "try-" + strconv.FormatInt(now.UnixNano(), 10)
data, err := encodeTry(t)
if err != nil {
return "", err
}
if err := store.Create(t.ID, data); err != nil {
return "", fmt.Errorf("the try %s cannot be written: %w", t.ID, err)
}
return t.ID, nil
}
// ReadTry is a try's value, as stored.
func ReadTry(store TryStore, id string) (json.RawMessage, error) {
if !meshcatalogclient.IsTryGroup(id) {
return nil, fmt.Errorf("%q is not a try's id, try-<digits>", id)
}
raw, err := store.Get(id)
if err != nil {
return nil, err
}
if len(raw) == 0 {
return nil, fmt.Errorf("no try %s: it was never asked, or it ended more than %s ago and was removed", id, TryKeep)
}
return raw, nil
}
// TryEnder is the tries state as the end tool sees it: one key with its revision, and a write at that revision.
type TryEnder interface {
Entry(key string) (json.RawMessage, uint64, error)
Update(key string, value []byte, revision uint64) error
}
// EndTry ends a try that waits or runs as error, saying why and that it was ended by hand (issue 734): its place is
// free at once, and the next checker_try of its head asks anew. A try that ended already is refused, naming its end.
func EndTry(store TryEnder, id, why string, now time.Time) (Try, error) {
if !meshcatalogclient.IsTryGroup(id) {
return Try{}, fmt.Errorf("%q is not a try's id, try-<digits>", id)
}
why = strings.TrimSpace(why)
if why == "" {
return Try{}, errors.New("say why the try is ended: the why is kept in what it says")
}
raw, rev, err := store.Entry(id)
if err != nil {
return Try{}, err
}
if len(raw) == 0 {
return Try{}, fmt.Errorf("no try %s: it was never asked, or it ended more than %s ago and was removed", id, TryKeep)
}
t, err := DecodeTry(raw)
if err != nil {
return Try{}, err
}
if t.Over() {
return Try{}, fmt.Errorf("%s ended already, %s at %s", id, t.State, t.Ended.Format(time.RFC3339))
}
t.State, t.Ended = TryError, now.UTC()
if !t.Started.IsZero() {
t.Took = t.Ended.Sub(t.Started).Round(time.Second).String()
}
t.Said = strings.TrimSpace("ended by hand (" + TryEndTool + "): " + why + " | " + t.Said)
t.Said = strings.TrimSuffix(t.Said, " |")
data, err := encodeTry(t)
if err != nil {
return Try{}, err
}
if err := store.Update(id, data, rev); err != nil {
return Try{}, fmt.Errorf("%s changed while it was ended, read it and end it again: %w", id, err)
}
return t, nil
}
// TryOptions narrow what a try runs; a pull request's check has none.
type TryOptions struct {
Level string
Run string
}
// Whole says the try runs everything a group's pull-request check runs (both levels of every member, the manifests judged by
// the manifest validator, the module graph, the composition on every placed node): it is narrowed to neither a level nor
// some tests. Such a try judges by the same code as the check, so the two cannot drift (issue 1921).
func (o *TryOptions) Whole() bool { return o != nil && o.Level == "" && o.Run == "" }
// TryCase is the case a try is judged as: one pull of the group per member at its exact commit, with no pull request number
// (no ref: the commit is fetched as it is), the group named by the try's key, and the heads hash its candidate is keyed by.
func TryCase(key string, t Try, watch func(string, *buildstep.Step)) *Case {
cs := &Case{Key: key, Group: key, Watch: watch, Trunk: map[string]string{}, Queued: t.Asked, Progress: &Progress{},
Try: &TryOptions{Level: t.Level, Run: t.Tests}}
var heads []meshdeliveryclient.Head
for _, m := range t.MembersOf() {
cs.Pulls = append(cs.Pulls, Pull{Repository: m.Repository, Head: m.Head, Trunk: m.Trunk})
cs.Trunk[m.Repository] = m.Trunk
heads = append(heads, meshdeliveryclient.Head{Repository: m.Repository, Head: m.Head})
}
cs.HeadsHash, _ = meshdeliveryclient.HeadsHash(heads)
cs.Heads = cs.HeadsHash
return cs
}
// compositionMade says whether a composition was made; notMadeWords what it says when it was not.
func compositionMade(c *meshdeliverychecker.VerdictsByGroupHeadsValueComposition) bool {
return c != nil && c.State != meshdeliverycheckerclient.CompositionNotRun
}
func notMadeWords(c *meshdeliverychecker.VerdictsByGroupHeadsValueComposition) string {
if c == nil || c.Words == "" {
return "no reason was given"
}
return c.Words
}
// Tryer runs the tries: it takes a queued try, judges it through the same Judge as a pull request's check (a check ask, ADR
// 0525 §1), and writes its end. It holds no timer: a try starts when its key is written.
type Tryer struct {
Tries KV
Judge Judge
Bound time.Duration
// NotTaken ends a try whose ask no build node took within it as error (issue 734); none when 0.
NotTaken time.Duration
Now func() time.Time
Logf func(format string, args ...any)
// Say publishes an event of the checker's stream (checkermsg.Started, checkermsg.Ended) with its body; nil says nothing. The
// indexer, which hears them, writes and deletes the candidate of a whole try's group.
// It answers the publish's error, nil only when the stream acknowledged it: an end not acknowledged is said again.
Say func(ctx context.Context, event string, body checkermsg.Event) error
mu sync.Mutex
ending map[string]bool // tries whose end is being said now
wg sync.WaitGroup
inflight map[string]context.CancelFunc
seen map[string]bool // looked at since this process started: a try left running is taken over the first time
}
func (tr *Tryer) now() time.Time {
if tr.Now != nil {
return tr.Now()
}
return time.Now()
}
func (tr *Tryer) logf(format string, args ...any) {
if tr.Logf != nil {
tr.Logf(format, args...)
}
}
// Wait returns when every try started so far has ended.
func (tr *Tryer) Wait() { tr.wg.Wait() }
// Put takes a value of the tries state in: a queued try is started, and a try left running is taken over the first time
// this process sees it (the checker that ran it stopped; there is one checker).
func (tr *Tryer) Put(ctx context.Context, key string, value []byte) {
t, err := DecodeTry(value)
if err != nil {
tr.logf("tries.%s holds what is not a try, left: %v", key, err)
return
}
tr.mu.Lock()
if tr.inflight == nil {
tr.inflight, tr.seen = map[string]context.CancelFunc{}, map[string]bool{}
}
if t.Over() {
// Ended while it ran (checker_try_end): its judging stops, and its asks are no longer waited for.
if stop := tr.inflight[key]; stop != nil {
stop()
}
tr.mu.Unlock()
tr.sendEnd(ctx, key) // ended by hand while it ran, or its end was written and not said: the indexer drops its candidate
return
}
if tr.inflight[key] != nil || (t.State == TryRunning && tr.seen[key]) {
tr.mu.Unlock()
return
}
run, stop := context.WithCancel(ctx)
tr.seen[key] = true
tr.inflight[key] = stop
tr.mu.Unlock()
tr.wg.Add(1)
go func() {
defer tr.wg.Done()
defer func() { tr.mu.Lock(); delete(tr.inflight, key); tr.mu.Unlock(); stop() }()
tr.run(run, key)
}()
}
// Reconsider looks at every try again (a delivery that held them ended), and removes the ended tries older than TryKeep.
func (tr *Tryer) Reconsider(ctx context.Context) {
keys, err := tr.Tries.Keys(ctx)
if err != nil {
tr.logf("the tries cannot be listed: %v", err)
return
}
for _, k := range keys {
raw, _, found, err := tr.Tries.Get(ctx, k)
if err != nil || !found {
continue
}
if t, err := DecodeTry(raw); err == nil && t.Over() && tr.now().Sub(t.Ended) > TryKeep {
if d, ok := tr.Tries.(interface {
Delete(ctx context.Context, key string) error
}); ok {
if err := d.Delete(ctx, k); err != nil {
tr.logf("%s ended %s ago and could not be removed: %v", k, tr.now().Sub(t.Ended).Round(time.Hour), err)
}
}
continue
}
tr.Put(ctx, k, raw)
}
}
// run claims the try by compare-and-set, judges it within its bound and writes its end.
func (tr *Tryer) run(ctx context.Context, key string) {
raw, rev, found, err := tr.Tries.Get(ctx, key)
if err != nil || !found {
return
}
t, err := DecodeTry(raw)
if err != nil || t.Over() {
return
}
t.State, t.Started = TryRunning, tr.now().UTC()
t.Told = (&TryOptions{Level: t.Level, Run: t.Tests}).Whole() // its start is said below; its end is owed from now on
data, err := encodeTry(t)
if err != nil {
return
}
if rev, err = tr.Tries.Update(ctx, key, data, rev); err != nil {
if !errors.Is(err, ErrConflict) {
tr.logf("%s: %v", key, err)
}
return
}
progress := &tryWriter{tr: tr, ctx: ctx, key: key, t: t, rev: rev}
if t.Told {
started := checkermsg.Event{ID: key}
for _, m := range t.MembersOf() {
started.Members = append(started.Members, checkermsg.Member{Repository: m.Repository, Head: m.Head, Trunk: m.Trunk})
}
tr.say(ctx, checkermsg.Started, started)
}
cs := TryCase(key, t, progress.heard)
cs.Wait = progress.waiting
judging := ctx
if tr.NotTaken > 0 {
judging = buildrun.NotTakenBound(ctx, tr.NotTaken)
}
out, judged := judgeWithin(judging, tr.Judge, cs, tr.Bound)
if ctx.Err() != nil {
return // left running: the checker takes it over when it starts; or ended by hand, which wrote its end
}
t, rev = progress.end()
t.Waiting = ""
v := out.Verdict
t.Ended = tr.now().UTC()
t.Took = t.Ended.Sub(t.Started).Round(time.Second).String()
t.Levels, t.Machines, t.Asks = v.Levels, v.Machines, cs.Asks()
t.Composition = v.Composition
t.Said = Report(v, cs, cs.Pulls[0], out.Builds, out.Slowest, out.Judged)
if len(cs.Pulls) > 1 {
t.Said = sayMembers(t.MembersOf()) + ": " + t.Said
}
t.Notes = cs.Notes()
for _, n := range t.Notes {
t.Said += " | " + n
}
switch {
case !judged || IsNodesFailure(v) || IsLostAsk(v) || IsNotTaken(v):
// Not the change's: its ask was given up or never taken (issue 734), or a part of the mesh was not heard.
t.State = TryError
if IsLostAsk(v) || IsNotTaken(v) {
t.Failed = []string{failedOf(v).Words}
}
case v.State == meshdeliverycheckerclient.VerdictPass && cs.Try.Whole() && compositionMade(v.Composition) && strings.Contains(v.Composition.Words, nodeengineclient.NoAnswer):
// A node that gave no answer passes a group's check with a warning; a try that is to say the group composes says it
// did not, naming the node (ADR 0637 §3).
t.State = TryError
t.Failed = []string{"the composition was not made on every placed node: " + v.Composition.Words}
case v.State == meshdeliverycheckerclient.VerdictPass && cs.Try.Whole() && !compositionMade(v.Composition):
// A whole try that judged the tests and not the manifests and the composition has not tried what its group's check
// judges: it is no pass, and it says why (failure must be loud). It is an error, not the change's failure.
t.State = TryError
t.Failed = []string{"the composition was not made: " + notMadeWords(v.Composition)}
case v.State == meshdeliverycheckerclient.VerdictPass:
t.State = TryPass
default:
t.State = TryFail
t.Failed = failedLines(v)
}
if t.Tests != "" {
t.Said += " | only the tests " + t.Tests + " (Go); nothing kept"
}
if data, err = encodeTry(t); err == nil {
_, err = tr.Tries.Update(ctx, key, data, rev)
}
if err != nil {
tr.logf("%s: its end could not be written: %v", key, err)
return
}
tr.logf("%s: %s", key, t.State)
// The end just written comes back on the key's watch, and Put then stops this run's context (it stops judging a try
// ended by hand) while this run says the end, which Put's own say leaves to it: the end is said on a context that
// stop does not cancel, or it is never said and the indexer keeps the candidate of the try's group.
tr.sendEnd(context.WithoutCancel(ctx), key)
}
// sendEnd says the end of an ended try whose start was said (Told) with the state it ended in, then clears Told in the try's
// own state; a checker that stopped between the two says it again at its start. Every end is said so: pass, fail, error and
// by hand.
func (tr *Tryer) sendEnd(ctx context.Context, key string) {
tr.mu.Lock()
if tr.ending[key] {
tr.mu.Unlock()
return
}
if tr.ending == nil {
tr.ending = map[string]bool{}
}
tr.ending[key] = true
tr.mu.Unlock()
defer func() { tr.mu.Lock(); delete(tr.ending, key); tr.mu.Unlock() }()
raw, rev, found, err := tr.Tries.Get(ctx, key)
if err != nil || !found {
return
}
t, err := DecodeTry(raw)
if err != nil || !t.Over() || !t.Told {
return
}
if tr.say(ctx, checkermsg.Ended, checkermsg.Event{ID: key, State: t.State}) != nil {
return // not acknowledged: Told stays, and the next Put or Reconsider says the end again
}
t.Told = false
data, err := encodeTry(t)
if err == nil {
_, err = tr.Tries.Update(ctx, key, data, rev)
}
if err != nil && !errors.Is(err, ErrConflict) {
tr.logf("%s: its end was said and Told could not be cleared: %v", key, err)
}
}
// say publishes a try event; a failure is said in the log (the indexer finds the open tries again at its start).
func (tr *Tryer) say(ctx context.Context, event string, body checkermsg.Event) error {
if tr.Say == nil {
return nil
}
err := tr.Say(ctx, event, body)
if err != nil {
tr.logf("%s of %s could not be said: %v", event, body.ID, err)
}
return err
}
// OpenTries are every try that waits or runs now: what checker_tries answers, so a part that missed an event (the indexer at
// its start) finds them without reading the checker's state. All the keys are read (an old open try is as open as a new one), in
// parallel.
func OpenTries(store TryStore) ([]checkermsg.Open, error) {
keys, err := store.Keys()
if err != nil {
return nil, fmt.Errorf("the tries cannot be listed: %w", err)
}
sort.Slice(keys, func(i, j int) bool {
if len(keys[i]) != len(keys[j]) {
return len(keys[i]) > len(keys[j])
}
return keys[i] > keys[j]
})
found := make([]*checkermsg.Open, len(keys))
if err := parallel(len(keys), func(i int) error {
raw, err := store.Get(keys[i])
if err != nil || len(raw) == 0 {
return nil
}
t, err := DecodeTry(raw)
if err != nil || t.Over() {
return nil
}
o := checkermsg.Open{ID: t.ID, State: t.State, Whole: (&TryOptions{Level: t.Level, Run: t.Tests}).Whole()}
for _, m := range t.MembersOf() {
o.Members = append(o.Members, checkermsg.Member{Repository: m.Repository, Head: m.Head, Trunk: m.Trunk})
}
found[i] = &o
return nil
}); err != nil {
return nil, err
}
out := []checkermsg.Open{} // none open is an empty list, never null: a reader tells it from no answer (issue 1921)
for _, o := range found {
if o != nil {
out = append(out, *o)
}
}
return out, nil
}
// judgeWithin runs a case within the bound (none when 0: the asks' steps are bounded from their start, ADR 0525 §6). A
// judge that fails to judge, and a check that ran past its bound, are verdicts of failure naming the machine and the step,
// and judged says false for them.
func judgeWithin(ctx context.Context, j Judge, cs *Case, bound time.Duration) (Outcome, bool) {
run, cancel := context.WithCancel(ctx)
if bound > 0 {
run, cancel = context.WithTimeoutCause(ctx, bound, errBound)
}
defer cancel()
out, err := j.Judge(run, cs)
stage, node := cs.Progress.Now()
if node == "" {
node = "checker"
}
switch {
case ctx.Err() != nil:
return Outcome{}, false
case context.Cause(run) == errBound:
if stage == "" {
stage = "starting"
}
return Outcome{Verdict: failure(node, fmt.Sprintf("the check ran past its bound of %s, stuck at: %s", bound, stage))}, false
case err != nil:
return Outcome{Verdict: failure(node, err.Error())}, false
}
return out, true
}
// sayMembers names a try's members, each repository at its short commit.
func sayMembers(ms []TryMember) string {
var out []string
for _, m := range ms {
out = append(out, fmt.Sprintf("%s@%.12s", m.Repository, m.Head))
}
return strings.Join(out, " + ")
}
// orNone joins the tries said, or says none.
func orNone(said []string) string {
if len(said) == 0 {
return "none"
}
return strings.Join(said, ", ")
}