Judge a pushed head as a try on the checker, through checker_try and checker_try_read (ADR 0501 §5)

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".
This commit is contained in:
2026-10-22 18:03:28 +02:00
parent 2855ae7ebf
commit 7abe230701
13 changed files with 830 additions and 27 deletions
@@ -7,8 +7,10 @@ import (
"encoding/json"
"fmt"
"os"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-checker/internal/check"
)
@@ -23,6 +25,71 @@ func read(s stdio.KeptState) check.StateRead {
}}
}
// stateStore is a state the tools write: the tries.
type stateStore struct {
check.StateRead
create func(key string, value []byte) error
}
func (s stateStore) Keys() ([]string, error) { return s.StateRead.Keys() }
func (s stateStore) Get(key string) (json.RawMessage, error) { return s.StateRead.Get(key) }
func (s stateStore) Create(key string, value []byte) error { return s.create(key, value) }
// Trying is what the try's tools work on: the tries, and the gate's trunk heads.
type Trying struct {
Tries check.TryStore
Trunks func() (map[string]string, error)
Now func() time.Time
}
// TryTools are the try's two tools (ADR 0501 §5): ask one, read one. mesh-controller's try calls them.
func TryTools(tr Trying) []stdio.Tool {
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s }
return []stdio.Tool{
{Name: check.TryTool,
Description: "Try a pushed head before its pull request is opened: the checker judges it as it judges a pull request's head, merged onto its trunk's head, " +
"its unit and integration levels through the build modules, and owes the forge nothing. Answers the try's id at once; " + check.TryReadTool + " reads " +
"where it stands. A try of the same head that waits or runs is the answer to a second ask; at most 4 wait or run at once, mesh-wide. A repository " +
"the checker checks no level of (hq, one whose trunk the gate does not name) is refused.",
Input: map[string]any{"type": "object", "required": []string{"repository", "head"}, "properties": map[string]any{
"repository": map[string]any{"type": "string", "description": "<owner>/<name>"},
"head": map[string]any{"type": "string", "description": "the full commit to try, 40 or 64 hex digits"},
"trunk": map[string]any{"type": "string", "description": "the trunk commit to merge it onto, or the trunk's branch for its head now; the trunk's head when absent"},
"tests": map[string]any{"type": "string", "description": "a go test -run pattern: only those tests run, and nothing is kept"},
"level": map[string]any{"type": "string", "enum": []string{"unit", "integration"}, "description": "only this level; both when absent"}}},
Run: func(args map[string]any) (any, error) {
if tr.Tries == nil || tr.Trunks == nil {
return nil, fmt.Errorf("the tries state cannot be written here")
}
trunks, err := tr.Trunks()
if err != nil {
return nil, fmt.Errorf("the trunk heads cannot be read: %w", err)
}
now := time.Now
if tr.Now != nil {
now = tr.Now
}
id, err := check.AskTry(tr.Tries, trunks, check.TryAsk{Repository: str(args, "repository"), Head: str(args, "head"),
Trunk: str(args, "trunk"), Tests: str(args, "tests"), Level: str(args, "level")}, now())
if err != nil {
return nil, err
}
return map[string]any{"id": id}, nil
}},
{Name: check.TryReadTool,
Description: "Where a try stands, as stored: queued, running, or its end (pass, fail, or error when it could not run), with what was said of it, " +
"the head and the trunk commit it was merged onto, the time it took and what failed first. Read-only.",
Input: map[string]any{"type": "object", "required": []string{"id"}, "properties": map[string]any{
"id": map[string]any{"type": "string", "description": "the try's id, try-<digits>"}}},
Run: func(args map[string]any) (any, error) {
if tr.Tries == nil {
return nil, fmt.Errorf("the tries state cannot be read here")
}
return check.ReadTry(tr.Tries, str(args, "id"))
}},
}
}
// Tools are the read tools over the checker's states.
func Tools(r check.Reader) []stdio.Tool {
return []stdio.Tool{
@@ -105,7 +172,23 @@ func main() {
PreviewState: read(stdio.State("previews")), PlansState: read(stdio.State("mesh-controller.plans-building")),
StepStates: map[string]check.StateRead{"mesh-build-go": read(stdio.State("mesh-build-go.steps")), "mesh-build-node": read(stdio.State("mesh-build-node.steps"))},
Capacity: capacity}
if err := stdio.Serve("", Tools(r)); err != nil {
tries := stdio.State(check.TriesState)
source := stdio.State("mesh-delivery-gate.source")
trying := Trying{
Tries: stateStore{StateRead: read(tries), create: func(key string, value []byte) error {
_, err := tries.Create(key, json.RawMessage(value))
return err
}},
Trunks: func() (map[string]string, error) {
e, err := source.Get(delivery.SourceMainKey)
if err != nil || e == nil {
return nil, err
}
s, err := delivery.DecodeSource(e.Value)
return s.Heads, err
},
}
if err := stdio.Serve("", append(Tools(r), TryTools(trying)...)); err != nil {
fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err)
os.Exit(1)
}
@@ -1,8 +1,12 @@
package main
import (
"encoding/json"
"errors"
"slices"
"strings"
"testing"
"time"
"mesh-delivery-checker/internal/check"
)
@@ -29,3 +33,49 @@ func TestTheToolsServedAreTheFiveReadTools(t *testing.T) {
}
}
}
// memStore is a tries state in memory.
type memStore struct{ vals map[string]json.RawMessage }
func (m *memStore) Keys() ([]string, error) {
var out []string
for k := range m.vals {
out = append(out, k)
}
return out, nil
}
func (m *memStore) Get(key string) (json.RawMessage, error) { return m.vals[key], nil }
func (m *memStore) Create(key string, value []byte) error {
if _, ok := m.vals[key]; ok {
return errors.New("exists")
}
m.vals[key] = value
return nil
}
func TestTheTryToolsAskAndReadATry(t *testing.T) { // ADR 0501 §5, the agreed interface with mesh-controller's try
store := &memStore{vals: map[string]json.RawMessage{}}
tools := TryTools(Trying{Tries: store, Now: func() time.Time { return time.Unix(1, 5) },
Trunks: func() (map[string]string, error) {
return map[string]string{"novox/mesh-delivery": strings.Repeat("a", 40)}, nil
}})
if len(tools) != 2 || tools[0].Name != "checker_try" || tools[1].Name != "checker_try_read" {
t.Fatalf("tools = %+v", tools)
}
out, err := tools[0].Run(map[string]any{"repository": "novox/mesh-delivery", "head": strings.Repeat("1", 40), "trunk": "main"})
if err != nil {
t.Fatal(err)
}
id := out.(map[string]any)["id"].(string)
if id != "try-1000000005" {
t.Errorf("id = %q", id)
}
raw, err := tools[1].Run(map[string]any{"id": id})
if err != nil || !strings.Contains(string(raw.(json.RawMessage)), `"state":"queued"`) {
t.Errorf("read = %s, %v", raw, err)
}
if _, err := tools[0].Run(map[string]any{"repository": "novox/hq", "head": strings.Repeat("1", 40)}); err == nil ||
err.Error() != "no level of novox/hq is checked by the checker" {
t.Errorf("hq: %v", err)
}
}
@@ -57,6 +57,9 @@ func (s kvStore) Keys(ctx context.Context) ([]string, error) {
return keys, err
}
// Delete removes a key (an ended try past its keeping).
func (s kvStore) Delete(ctx context.Context, key string) error { return s.kv.Delete(ctx, key) }
func opOf(e jetstream.KeyValueEntry) bool { return e.Operation() == jetstream.KeyValuePut }
// settle reads the entries a watcher holds now, up to its mark of the last values, and returns the watcher for the
@@ -20,6 +20,8 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-checker/internal/check"
)
// busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
@@ -267,7 +269,7 @@ func startBed(t *testing.T) *bed {
for name, history := range map[string]int{delivery.CheckerVerdictsBucket: delivery.CheckerVerdictsHistory, delivery.CheckerKeptBucket: delivery.CheckerKeptHistory,
delivery.CheckerOwedBucket: delivery.CheckerOwedHistory, delivery.PreviewsBucket: delivery.PreviewsHistory, delivery.CandidatesBucket: delivery.CandidatesHistory,
delivery.ModulesBucket: delivery.ModulesHistory, delivery.ManifestsBucket: delivery.ManifestsHistory, delivery.SourceBucket: delivery.SourceHistory,
delivery.PresenceBucket: delivery.PresenceHistory, delivery.HaveBucket: delivery.HaveHistory} {
delivery.PresenceBucket: delivery.PresenceHistory, delivery.HaveBucket: delivery.HaveHistory, check.TriesBucket: 1} {
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: uint8(history)})
if err != nil {
t.Fatal(err)
@@ -590,3 +592,62 @@ func TestAMemberNamingAMeshSDKCommitOffTheTrunkFailsNamingTheCommit(t *testing.T
t.Errorf("%d test steps ran after a stale pin", n)
}
}
// triesStore is the tries bucket as the tools write it.
type triesStore struct {
kv jetstream.KeyValue
ctx context.Context
}
func (s triesStore) Keys() ([]string, error) {
keys, err := s.kv.Keys(s.ctx)
if err == jetstream.ErrNoKeysFound {
return nil, nil
}
return keys, err
}
func (s triesStore) Get(key string) (json.RawMessage, error) {
e, err := s.kv.Get(s.ctx, key)
if err == jetstream.ErrKeyNotFound {
return nil, nil
}
if err != nil {
return nil, err
}
return e.Value(), nil
}
func (s triesStore) Create(key string, value []byte) error {
_, err := s.kv.Create(s.ctx, key, value)
return err
}
// TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed: a try written as checker_try writes it is taken by the running checker,
// judged through the build modules as a pull request's head merged onto its trunk, and ended pass in its key, with no
// verdict and no status owed (ADR 0501 §5).
func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) {
b := startBed(t)
b.start(t, nil)
store := triesStore{b.buckets[check.TriesBucket], b.ctx}
id, err := check.AskTry(store, map[string]string{repo: trunk}, check.TryAsk{Repository: repo, Head: head1}, time.Now())
if err != nil {
t.Fatal(err)
}
var last check.Try
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
raw, _ := store.Get(id)
if last, err = check.DecodeTry(raw); err == nil && last.Over() {
break
}
}
if last.State != check.TryPass || last.Trunk != trunk || last.Took == "" {
t.Fatalf("try = %+v, want passed on the trunk", last)
}
if b.fake.count("mesh-build-go.build_go_fetch") == 0 {
t.Errorf("the try fetched nothing")
}
for _, name := range []string{delivery.CheckerOwedBucket, delivery.CheckerVerdictsBucket, delivery.PreviewsBucket} {
if keys, _ := b.buckets[name].Keys(b.ctx); len(keys) != 0 {
t.Errorf("a try wrote %s: %v", name, keys)
}
}
}
@@ -151,7 +151,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
buckets := map[string]jetstream.KeyValue{}
for _, name := range []string{delivery.CheckerVerdictsBucket, delivery.CheckerKeptBucket, delivery.CheckerOwedBucket, delivery.PreviewsBucket,
delivery.CandidatesBucket, delivery.ModulesBucket, delivery.ManifestsBucket} {
delivery.CandidatesBucket, delivery.ModulesBucket, delivery.ManifestsBucket, check.TriesBucket} {
if buckets[name], err = bucket(name); err != nil {
return err
}
@@ -209,8 +209,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
default:
builds = buildReader{kvStore{kv}}
}
holds := newHoldWatch(ctx, js, ch.Reconsider, logger)
// The tries (ADR 0501 §5) are judged by the same judge, held by the same deliveries.
tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, Logf: logger.Printf}
holds := newHoldWatch(ctx, js, func(ctx context.Context) { ch.Reconsider(ctx); tryer.Reconsider(ctx) }, logger)
ch.Hold = holds.Held
tryer.Hold = holds.Held
announce := func(ctx context.Context, event, key string) {
body, _ := json.Marshal(map[string]string{"key": key})
if _, err := js.Publish(ctx, eventSubject(event), body); err != nil {
@@ -225,6 +228,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
announce: announce,
})
ch.Announce = announce
tryer.Judge = ch.Judge
// The toolchains first, so a check that starts from the first candidate finds the nodes.
watch := newToolchainWatch(ctx, js, tools, ch.NodesChanged, logger)
@@ -336,11 +340,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err := ch.Sweep(ctx); err != nil {
return err
}
// The tries: a queued one starts, one left running is taken over, an ended one past its keeping is removed.
tw, err := settle(ctx, buckets[check.TriesBucket], func(jetstream.KeyValueEntry) {})
if err != nil {
return err
}
tryer.Reconsider(ctx)
go func() {
errs <- follow(ctx, tw, func(e jetstream.KeyValueEntry) {
if opOf(e) {
tryer.Put(ctx, e.Key(), e.Value())
}
})
}()
logger.Printf("started on %s", cred.Node)
go func() { errs <- follow(ctx, w, each) }()
select {
case <-ctx.Done():
ch.Wait()
tryer.Wait()
return nil
case err := <-errs:
return err
@@ -45,6 +45,8 @@ type Case struct {
Left []string
// Queued is when the checker took the case up, for the time of its queue.
Queued time.Time
// Try narrows a try's check (its level, its tests); nil for a pull request's.
Try *TryOptions
// Progress says what the check is doing, so a check that ran past its bound names the step it was stuck at.
Progress *Progress
}
@@ -349,25 +349,7 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
// 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 {
bound := ch.bound()
run, cancel := context.WithTimeoutCause(ctx, bound, errBound)
defer cancel()
out, err := ch.Judge.Judge(run, cs)
stage, node := cs.Progress.Now()
if node == "" {
node = "checker"
}
switch {
case ctx.Err() != nil:
return Outcome{}
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))}
case err != nil:
return Outcome{Verdict: failure(node, err.Error())}
}
out, _ := judgeWithin(ctx, ch.Judge, cs, ch.bound())
return out
}
@@ -154,6 +154,9 @@ func (j *Examiner) Judge(ctx context.Context, c *Case) (Outcome, error) {
return st.finish(ctx, *v), nil
}
for _, level := range []string{delivery.LevelUnit, delivery.LevelIntegration} {
if c.Try != nil && c.Try.Level != "" && c.Try.Level != level {
continue // a try of one level runs that level alone
}
for _, priority := range []bool{true, false} {
for _, t := range st.targets {
if v := st.runTarget(ctx, t, level, priority); v != nil {
@@ -296,14 +299,18 @@ func (st *state) fetchAll(ctx context.Context) *delivery.Verdict {
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))
if st.c.Try == nil {
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)
if st.c.Try == nil { // a try owes the indexer no preview: no pull request holds its head
st.j.writePreview(ctx, p, merged)
}
}
return nil
}
@@ -410,6 +417,11 @@ func (st *state) planGo(ctx context.Context, t *target, level string) *delivery.
}
plan := &goPlan{digests: map[string]string{}}
t.plans[level] = plan
if st.c.Try != nil && st.c.Try.Run != "" {
// A try of some tests runs them alone: a partial run proves no package, so nothing is taken or kept.
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
@@ -480,6 +492,9 @@ func (st *state) runGo(ctx context.Context, t *target, level string, priority bo
if run != nil {
args["packages"] = run
}
if st.c.Try != nil && st.c.Try.Run != "" {
args["run"] = st.c.Try.Run // build_go_test's go test -run pattern
}
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))
@@ -0,0 +1,389 @@
package check
import (
"context"
"encoding/json"
"errors"
"fmt"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// 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"
// TriesState is the checker's state of tries: the bucket mesh-delivery-checker_tries, keys try-<unix nanoseconds>.
TriesState = "tries"
TriesBucket = delivery.CheckerModule + "_" + TriesState
// MaxTries is how many tries may wait or run at once, mesh-wide: each holds a build node for minutes.
MaxTries = 4
// 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 (
tryKeyOK = regexp.MustCompile(`^try-[0-9]{1,20}$`)
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 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"`
// 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"`
// 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 []string `json:"failed,omitempty"`
Levels map[string]delivery.LevelResult `json:"levels,omitempty"`
Machines []string `json:"machines,omitempty"`
Asked time.Time `json:"asked"`
Started time.Time `json:"started,omitzero"`
Ended time.Time `json:"ended,omitzero"`
}
// 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 !tryKeyOK.MatchString(t.ID) || t.Repository == "" || t.Head == "" || t.Trunk == "" {
return Try{}, errors.New("not a try: it names no id, repository, head or trunk")
}
return t, nil
}
func encodeTry(t Try) ([]byte, error) {
t.V = delivery.Version
return json.Marshal(t)
}
// TryAsk is what checker_try is given.
type TryAsk struct {
Repository, Head, Trunk, Tests, Level 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: hq, judged by merge-check.sh (ADR 0484
// clause 8), and a repository 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)
}
// AskTry writes a new try, queued, and answers its id: a try of the same head that waits or runs is the answer to a second
// ask, and at most MaxTries wait or run at once. trunks are the gate's trunk heads of every repository (source.main): a
// repository with none, and hq, are 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) {
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 "", fmt.Errorf("%q is not a repository, <owner>/<name>", a.Repository)
case !tryHeadOK.MatchString(head):
return "", fmt.Errorf("the head %q is not a full commit, 40 or 64 hex digits", a.Head)
case a.Level != "" && a.Level != delivery.LevelUnit && a.Level != delivery.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)
}
trunkHead := ""
for r, h := range trunks {
if strings.EqualFold(r, repo) {
trunkHead = h
}
}
if repo == HQRepository || trunkHead == "" {
return "", NotChecked(repo)
}
t := Try{Repository: repo, Head: head, Tests: a.Tests, Level: a.Level, State: TryQueued, Asked: now.UTC()}
switch trunk := strings.TrimSpace(a.Trunk); {
case trunk == "":
t.Trunk = trunkHead
case tryHeadOK.MatchString(strings.ToLower(trunk)):
t.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.
t.Trunk, t.Branch = trunkHead, trunk
default:
return "", fmt.Errorf("the trunk %q is neither a full commit nor a branch", a.Trunk)
}
keys, err := store.Keys()
if err != nil {
return "", fmt.Errorf("the tries cannot be listed: %w", err)
}
sort.Strings(keys)
var open []string
for _, k := range keys {
raw, err := store.Get(k)
if err != nil || len(raw) == 0 {
continue
}
cur, err := DecodeTry(raw)
if err != nil || cur.Over() {
continue
}
if cur.Repository == t.Repository && cur.Head == t.Head && cur.Trunk == t.Trunk && cur.Tests == t.Tests && cur.Level == t.Level {
return cur.ID, nil
}
open = append(open, fmt.Sprintf("%s (%s@%.12s, %s)", cur.ID, cur.Repository, cur.Head, cur.State))
}
if len(open) >= MaxTries {
return "", fmt.Errorf("%d tries wait or run, the most at once: %s; ask again when one ends", len(open), strings.Join(open, ", "))
}
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 !tryKeyOK.MatchString(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
}
// TryOptions narrow what a try runs; a pull request's check has none.
type TryOptions struct {
Level string
Run string
}
// Tryer runs the tries: it takes a queued try, judges it through the same Judge as a pull request's check, and writes its
// end. It holds no timer: a try starts when its key is written, or when a delivery that held it ends (Reconsider).
type Tryer struct {
Tries KV
Judge Judge
// Hold says whether a delivery builds, so no try starts (ADR 0484 clause 12); nil never holds.
Hold func() bool
Bound time.Duration
Now func() time.Time
Logf func(format string, args ...any)
mu sync.Mutex
wg sync.WaitGroup
inflight map[string]bool
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
}
if t.Over() {
return
}
held := tr.Hold != nil && tr.Hold()
tr.mu.Lock()
if tr.inflight == nil {
tr.inflight, tr.seen = map[string]bool{}, map[string]bool{}
}
if tr.inflight[key] || (t.State == TryRunning && tr.seen[key]) {
tr.mu.Unlock()
return
}
if held {
tr.mu.Unlock() // not seen yet: a try left running is still taken over when the hold ends
tr.logf("%s: not started, a delivery builds", key)
return
}
tr.seen[key] = true
tr.inflight[key] = true
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() }()
tr.run(ctx, 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()
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
}
cs := &Case{Key: key, Group: key,
Pulls: []Pull{{Repository: t.Repository, Head: t.Head, Trunk: t.Trunk}},
Trunk: map[string]string{t.Repository: t.Trunk},
Queued: t.Asked,
Progress: &Progress{},
Try: &TryOptions{Level: t.Level, Run: t.Tests}}
out, judged := judgeWithin(ctx, tr.Judge, cs, tr.Bound)
if ctx.Err() != nil {
return // left running: the checker takes it over when it starts
}
v := out.Verdict
t.Ended = tr.now().UTC()
t.Took = t.Ended.Sub(t.Started).Round(time.Second).String()
t.Levels, t.Machines = v.Levels, v.Machines
t.Said = Report(v, cs, cs.Pulls[0], out.Builds, out.Slowest)
switch {
case !judged || IsNodesFailure(v):
t.State = TryError
case v.State == delivery.VerdictPass:
t.State = TryPass
default:
t.State = TryFail
if f := firstFailed(v); f != "" {
t.Failed = []string{f}
}
}
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)
}
// judgeWithin runs a case within the bound (DefaultBound when 0). 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) {
if bound <= 0 {
bound = DefaultBound
}
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
}
@@ -0,0 +1,196 @@
package check
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// memTries is the tries state as the tools see it, over a memKV.
type memTries struct{ kv *memKV }
func (m memTries) Keys() ([]string, error) { return m.kv.Keys(context.Background()) }
func (m memTries) Get(key string) (json.RawMessage, error) {
v, _, _, err := m.kv.Get(context.Background(), key)
return v, err
}
func (m memTries) Create(key string, value []byte) error {
_, err := m.kv.Create(context.Background(), key, value)
return err
}
var tryTrunks = map[string]string{repoA: trunkA, repoSDK: trunkS, HQRepository: trunkA}
func readTry(t testing.TB, kv *memKV, id string) Try {
t.Helper()
raw, _, found, _ := kv.Get(context.Background(), id)
if !found {
t.Fatalf("no try %s", id)
}
tr, err := DecodeTry(raw)
if err != nil {
t.Fatal(err)
}
return tr
}
func TestATryIsAskedQueuedOncePerHeadAndAtMostFourAtOnce(t *testing.T) { // ADR 0501 §5
kv := newMemKV()
store := memTries{kv}
at := time.Date(2026, 10, 22, 18, 0, 0, 0, time.UTC)
id, err := AskTry(store, tryTrunks, TryAsk{Repository: repoA, Head: head1, Trunk: "main"}, at)
if err != nil || !strings.HasPrefix(id, "try-") {
t.Fatalf("id %q, %v", id, err)
}
tr := readTry(t, kv, id)
if tr.State != TryQueued || tr.Trunk != trunkA || tr.Branch != "main" || tr.Head != head1 || tr.Repository != repoA {
t.Errorf("try = %+v, want queued on the trunk's head", tr)
}
again, err := AskTry(store, tryTrunks, TryAsk{Repository: repoA, Head: head1, Trunk: "main"}, at.Add(time.Second))
if err != nil || again != id {
t.Errorf("a second ask of the same head answered %q, %v, want %q", again, err, id)
}
heads := []string{head2, trunkA2, trunkS2}
for i, h := range heads {
if _, err := AskTry(store, tryTrunks, TryAsk{Repository: repoA, Head: h, Trunk: trunkA}, at.Add(time.Duration(i+2)*time.Second)); err != nil {
t.Fatalf("try %d: %v", i+2, err)
}
}
if _, err := AskTry(store, tryTrunks, TryAsk{Repository: repoSDK, Head: head1}, at.Add(time.Minute)); err == nil || !strings.Contains(err.Error(), "4 tries wait or run") {
t.Errorf("a fifth try answered %v, want refused naming the four", err)
}
}
func TestATryOfARepositoryWithNoCheckedLevelIsRefused(t *testing.T) { // the agreed error of the try
store := memTries{newMemKV()}
for _, repo := range []string{HQRepository, "novox/mesh-lab"} {
_, err := AskTry(store, tryTrunks, TryAsk{Repository: repo, Head: head1, Trunk: "main"}, time.Now())
if err == nil || err.Error() != "no level of "+repo+" is checked by the checker" {
t.Errorf("%s: %v", repo, err)
}
}
for _, bad := range []TryAsk{{Repository: repoA, Head: "abc"}, {Repository: repoA, Head: head1, Level: "e2e"},
{Repository: repoA, Head: head1, Tests: "a b"}, {Repository: "x", Head: head1}, {Repository: repoA, Head: head1, Trunk: "-x"}} {
if _, err := AskTry(store, tryTrunks, bad, time.Now()); err == nil {
t.Errorf("%+v was taken", bad)
}
}
}
func TestATryIsJudgedAsAPullRequestsHeadAndOwesTheForgeNothing(t *testing.T) { // ADR 0501 §5
ctx := context.Background()
kv := newMemKV()
id, err := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1, Trunk: trunkA}, time.Now())
if err != nil {
t.Fatal(err)
}
b := newBench(t)
tr := &Tryer{Tries: kv, Judge: b.ex}
raw, _, _, _ := kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
got := readTry(t, kv, id)
if got.State != TryPass || got.Took == "" || got.Head != head1 || got.Trunk != trunkA || got.Said == "" || got.Levels[delivery.LevelUnit].Run != 2 {
t.Fatalf("try = %+v", got)
}
fetch := b.steps.tool("fetch")
if len(fetch) != 1 || fetch[0].args["commit"] != head1 || fetch[0].args["base"] != trunkA || fetch[0].args["ref"] != "" {
t.Errorf("fetch = %+v, want the head merged onto the trunk, with no pull request's ref", fetch)
}
if keys, _ := b.prev.Keys(ctx); len(keys) != 0 {
t.Errorf("a try wrote the previews %v", keys)
}
// An ended try is not run again.
raw, _, _, _ = kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
if n := len(b.steps.tool("fetch")); n != 1 {
t.Errorf("an ended try was fetched again: %d fetches", n)
}
}
func TestATryOfOneLevelAndSomeTestsRunsThemAloneAndKeepsNothing(t *testing.T) {
ctx := context.Background()
kv := newMemKV()
id, err := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1, Level: delivery.LevelIntegration, Tests: "TestBus|TestStore"}, time.Now())
if err != nil {
t.Fatal(err)
}
b := newBench(t)
tr := &Tryer{Tries: kv, Judge: b.ex}
raw, _, _, _ := kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
tests := b.steps.tool("test")
if len(tests) != 1 || tests[0].args["level"] != delivery.LevelIntegration || tests[0].args["run"] != "TestBus|TestStore" {
t.Fatalf("tests = %+v, want the integration level alone with the run pattern", tests)
}
if n := len(b.steps.tool("digest")); n != 0 {
t.Errorf("%d digests for a try of some tests", n)
}
if keys, _ := b.kept.Keys(ctx); len(keys) != 0 {
t.Errorf("kept %v from a try of some tests", keys)
}
if got := readTry(t, kv, id); got.State != TryPass || !strings.Contains(got.Said, "only the tests TestBus|TestStore") {
t.Errorf("try = %+v", got)
}
}
func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T) {
ctx := context.Background()
kv := newMemKV()
b := newBench(t)
b.test = func(c stepCall) (any, error) { return captured[Tested](t, "go-test-failing"), nil }
id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
tr := &Tryer{Tries: kv, Judge: b.ex}
raw, _, _, _ := kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
if got := readTry(t, kv, id); got.State != TryFail || len(got.Failed) != 1 || !strings.Contains(got.Failed[0], "TestThreeIsFour") {
t.Errorf("try = %+v", got)
}
// No build node: the try could not run, which is an error, not a failure of the change.
kv2 := newMemKV()
b2 := newBench(t)
b2.ex.Toolchains = entries{}
id2, _ := AskTry(memTries{kv2}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
tr2 := &Tryer{Tries: kv2, Judge: b2.ex}
raw, _, _, _ = kv2.Get(ctx, id2)
tr2.Put(ctx, id2, raw)
tr2.Wait()
if got := readTry(t, kv2, id2); got.State != TryError {
t.Errorf("try with no node = %+v, want error", got)
}
}
func TestATryWaitsWhileADeliveryBuildsAndATryLeftRunningIsTakenOver(t *testing.T) {
ctx := context.Background()
kv := newMemKV()
id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
// Left running by a checker that stopped.
raw, rev, _, _ := kv.Get(ctx, id)
cur, _ := DecodeTry(raw)
cur.State = TryRunning
data, _ := encodeTry(cur)
if _, err := kv.Update(ctx, id, data, rev); err != nil {
t.Fatal(err)
}
b := newBench(t)
held := true
tr := &Tryer{Tries: kv, Judge: b.ex, Hold: func() bool { return held }}
tr.Reconsider(ctx)
tr.Wait()
if n := len(b.steps.tool("fetch")); n != 0 {
t.Fatalf("a try ran while a delivery builds")
}
held = false
tr.Reconsider(ctx)
tr.Wait()
if got := readTry(t, kv, id); got.State != TryPass {
t.Errorf("the try left running = %+v, want taken over and passed", got)
}
}
@@ -92,7 +92,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
t.Errorf("module = %q", m.Module)
}
// delivery.CheckerKeptHistory, CheckerOwedHistory, PreviewsHistory, CheckerVerdictsHistory; the names follow the SDK's buckets.
if want := []state{{Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) {
if want := []state{{Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) {
t.Errorf("state = %+v, want %+v", m.State, want)
}
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains",
@@ -118,7 +118,8 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
t.Errorf("streams = %+v, want %+v", m.Streams, want)
}
// The read tools (issue 1686) are served by the tools bundle, not by the checker's process.
if want := []string{"checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) {
// The try's two tools (ADR 0501 §5) beside them: mesh-controller's try calls them.
if want := []string{"checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_try", "checker_try_read", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) {
t.Errorf("tools = %v, want %v", m.Tools, want)
}
for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
@@ -13,6 +13,7 @@
{ "name": "kept", "history": 1 },
{ "name": "owed", "history": 5 },
{ "name": "previews", "history": 5 },
{ "name": "tries", "history": 1 },
{ "name": "verdicts", "history": 5 }
],
"reads": [
@@ -34,6 +35,8 @@
"checker_previews",
"checker_pull",
"checker_status",
"checker_try",
"checker_try_read",
"checker_verdicts"
],
"invokes": [