Files
mesh-delivery/modules/mesh-delivery-checker/internal/check/try_test.go
T
jschoubben f9578cb7b4
mesh/delivery merged: source.main moved to novox/mesh-delivery@abe9ea3796ea
mesh/check being carried to f9578cb7b418 on 8c67adaf9136 | from abe9ea3796ea
A try's end is said again until the stream acknowledges it; a waiting try says why; a hand recovery never overlaps a try event (issue 1921, ADR 0639)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 19:24:45 +01:00

482 lines
22 KiB
Go

package check
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/sdk/tryevent"
"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
}
// writtenTries is a memKV that keeps every running try it was given to write, in order.
type writtenTries struct {
*memKV
running []Try
}
func (w *writtenTries) Update(ctx context.Context, key string, value []byte, rev uint64) (uint64, error) {
if t, err := DecodeTry(value); err == nil && !t.Over() {
w.running = append(w.running, t)
}
return w.memKV.Update(ctx, key, value, rev)
}
var tryTrunks = map[string]string{repoA: trunkA, repoSDK: trunkS, hqRepo: 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)
}
// Issue 1890: tries on a build node (a step of theirs running, or none named yet) and tries waiting for one (queued, or
// their asks not taken) are bounded apart, MaxTries and MaxTriesWaiting; the refusal names both counts.
// A group try is the same try whatever the order of its members; the commits asked are the commits tried.
group := TryAsk{Members: []TryAskMember{{Repository: repoSDK, Head: head2}, {Repository: repoA, Head: head1}}}
gid, err := AskTry(store, tryTrunks, group, at.Add(time.Second))
if err != nil || gid == id {
t.Fatalf("the group try = %q, %v, want a try of its own", gid, err)
}
group.Members[0], group.Members[1] = group.Members[1], group.Members[0]
if again, err := AskTry(store, tryTrunks, group, at.Add(time.Minute)); err != nil || again != gid {
t.Errorf("the group asked in another order answered %q, %v, want %q", again, err, gid)
}
if g := readTry(t, kv, gid); len(g.Members) != 2 || g.Members[0].Repository != repoA || g.Members[0].Head != head1 || g.Members[1].Trunk != trunkS {
t.Errorf("group try = %+v, want its members by repository at the commits asked", g.Members)
}
if open, err := OpenTries(store); err != nil || len(open) != 2 || !open[0].Whole || len(open[1].Members) != 2 && len(open[0].Members) != 2 {
t.Errorf("open tries = %+v, %v, want the two queued", open, err)
}
if !IsTryGroup(gid) || IsTryGroup("feat/x") || IsTryGroup("fix/1921-try-1") {
t.Errorf("which candidates keys are a try's group: %s is, a branch is not", gid)
}
onNode := func(i int) Try {
return Try{ID: fmt.Sprintf("try-%d", i), Repository: repoA, Head: fmt.Sprintf("%040x", i), Trunk: trunkA, State: TryRunning,
Steps: []TryStep{{Ask: "a", State: delivery.StepRunning}}}
}
waitingFor := func(i int) Try {
return Try{ID: fmt.Sprintf("try-%d", i), Repository: repoA, Head: fmt.Sprintf("%040x", i), Trunk: trunkA, State: TryRunning,
Steps: []TryStep{{Ask: "a", State: TryAsked}}}
}
for _, c := range []struct {
what string
holding, wait int
refused string
}{
{"four queued behind the checks, none on a node: a fifth is taken", 0, 4, ""},
{"four on a node", MaxTries, 0, "now 4 run and 0 wait"},
{"three on a node, eight waiting", 3, MaxTriesWaiting, "a new try is refused while 4 tries run on a build node or 8 wait for one; now 3 run and 8 wait"},
} {
kv := newMemKV()
for i := range c.holding + c.wait {
tr := waitingFor(i + 1)
if i < c.holding {
tr = onNode(i + 1)
}
data, _ := encodeTry(tr)
_, _ = kv.Create(context.Background(), tr.ID, data)
}
_, err := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoSDK, Head: head1}, at.Add(time.Minute))
if (c.refused == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), c.refused)) {
t.Errorf("%s: answered %v, want refused %q", c.what, err, c.refused)
}
}
}
func TestATryOfARepositoryWithNoCheckedLevelIsRefused(t *testing.T) { // the agreed error of the try
store := memTries{newMemKV()}
for _, repo := range []string{"novox/no-trunk", "novox/mesh-lab"} { // hq is checked since R4g (ADR 0527): see tryTrunks
_, 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"},
// A group try (issue 1921): members or one repository, never both or none; one commit per repository; every member checked.
{},
{Repository: repoA, Head: head1, Members: []TryAskMember{{Repository: repoSDK, Head: head2}}},
{Members: []TryAskMember{{Repository: repoA, Head: head1}, {Repository: repoA, Head: head2}}},
{Members: []TryAskMember{{Repository: repoA, Head: head1}, {Repository: repoSDK, Head: "main"}}},
{Members: []TryAskMember{{Repository: repoA, Head: head1}, {Repository: "novox/no-trunk", Head: head2}}}} {
if _, err := AskTry(store, tryTrunks, bad, time.Now()); err == nil {
t.Errorf("%+v was taken", bad)
}
}
}
// fakeTryCandidates answers the indexer's candidate of a try's group from memory, recording the keys asked.
type fakeTryCandidates struct {
cand delivery.Candidate
err error
asked []string
// during is called while the try waits for the candidate.
during func()
}
func (f *fakeTryCandidates) TryCandidate(_ context.Context, group, heads string, _ time.Duration) (delivery.Candidate, error) {
if f.during != nil {
f.during()
}
f.asked = append(f.asked, group+"."+heads)
return f.cand, f.err
}
// wholeBench makes the bench judge a whole try as the checker does: the composition on every placed node and the
// indexer's candidate of the try's group.
func wholeBench(b *bench, cand delivery.Candidate) (*fakeCompositions, *fakeTryCandidates) {
comps := &fakeCompositions{v: delivery.CompositionVerdict{State: delivery.CompositionPass, SDK: "v0.1.14-x", Machines: []string{"g14", "shanks"}, Took: 40}}
cands := &fakeTryCandidates{cand: cand}
b.ex.Compositions, b.ex.Candidates = comps, cands
return comps, cands
}
// A group try (issue 1921): several repositories each at the exact commit asked, judged by the code of a group's pull-request
// check; the two cannot drift, and a later check of the same heads keeps what the try passed.
func TestATryIsJudgedAsAGroupsCheckAndOwesTheForgeNothing(t *testing.T) { // ADR 0501 §5, issue 1921
ctx := context.Background()
kv := newMemKV()
id, err := AskTry(memTries{kv}, tryTrunks, TryAsk{Members: []TryAskMember{{Repository: repoSDK, Head: head2, Trunk: trunkS}, {Repository: repoA, Head: head1, Trunk: trunkA}}}, time.Now())
if err != nil {
t.Fatal(err)
}
b := newBench(t)
comps, cands := wholeBench(b, delivery.Candidate{FPSet: "fps-of-the-try"})
written := &writtenTries{memKV: kv}
var waiting string
cands.during = func() { waiting = readTry(t, kv, id).Waiting } // ADR 0639 §5: the stored try says what it waits for
var said []string
tr := &Tryer{Tries: written, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) error {
said = append(said, fmt.Sprintf("%s %s %d %s", event, e.ID, len(e.Members), e.State))
return nil
}}
raw, _, _, _ := kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
// The indexer hears the start with the members at their commits, and the end (it writes and deletes the group's candidate).
if len(said) != 2 || said[0] != "try.started "+id+" 2 " || said[1] != "try.ended "+id+" 0 "+TryPass {
t.Errorf("events = %v, want the start with 2 members and the end", said)
}
// What the indexer reads: tryevent.Decode accepts the events as said and refuses a body that is no try's.
for _, row := range []struct {
name, event, body string
ok bool
}{
{"a start with its members", tryevent.Started, `{"id":"` + id + `","members":[{"repository":"` + repoA + `","head":"` + head1 + `","trunk":"` + trunkA + `"}]}`, true},
{"an end with its state", tryevent.Ended, `{"id":"` + id + `","state":"pass"}`, true},
{"an id that is no try's", tryevent.Ended, `{"id":"pull-7","state":"pass"}`, false},
{"a start with no member", tryevent.Started, `{"id":"` + id + `"}`, false},
{"a body that is no JSON", tryevent.Ended, `try-7`, false},
} {
if e, err := tryevent.Decode(row.event, []byte(row.body)); (err == nil) != row.ok || (row.ok && e.ID != id) {
t.Errorf("Decode of %s = %+v, %v, want ok=%v", row.name, e, err, row.ok)
}
}
got := readTry(t, kv, id)
if !strings.Contains(waiting, "candidate") || got.Waiting != "" {
t.Errorf("waiting while the candidate was awaited = %q, after the end %q: want the reason, then none", waiting, got.Waiting)
}
if got.Told { // the end was said: nothing is owed
t.Errorf("the ended try still owes its end: %+v", got)
}
if open, err := OpenTries(memTries{kv}); err != nil || len(open) != 0 {
t.Errorf("open tries after the end = %v, %v", open, err)
}
if got.State != TryPass || got.Took == "" || len(got.Members) != 2 || got.Members[0].Repository != repoA || got.Members[1].Head != head2 ||
got.Said == "" || got.Levels[delivery.LevelUnit].Run+got.Levels[delivery.LevelUnit].Kept != 4 { // the bench gives both members the same digests
t.Fatalf("try = %+v", got)
}
if got.Composition == nil || got.Composition.State != delivery.CompositionPass || len(got.Machines) == 0 {
t.Errorf("the try's composition = %+v, want the composition on every placed node, as a check's", got.Composition)
}
// The same key the indexer's candidate of the group has: the try's id and the heads of its members.
if len(comps.asked) != 1 || comps.asked[0] != id || len(cands.asked) != 1 || !strings.HasPrefix(cands.asked[0], id+".") {
t.Errorf("composed %v, candidate asked %v, want the try's key %s", comps.asked, cands.asked, id)
}
// Issue 1890: while it runs, the try names its asks as soon as they are put, and then where each step stands.
if len(written.running) < 3 || len(written.running[1].Asks) != 1 || written.running[1].Steps[0].State != TryAsked {
t.Errorf("the running try's writes = %+v, want its ask named while it ran", written.running)
}
if len(got.Steps) != 2 || got.Steps[0].Node != "shanks" || got.Steps[0].State != delivery.StepPassed || len(got.Asks) != 2 {
t.Errorf("the ended try's steps = %+v, asks %v", got.Steps, got.Asks)
}
for _, a := range got.Asks { // the delivery view counts their steps as a try's (build_usage)
if _, err := delivery.AskStepKey("shanks", a); err != nil {
t.Errorf("the try names %q, which is no ask id: %v", a, err)
}
}
// Each member is fetched at the exact commit asked, merged onto its own trunk, with no pull request's ref.
fetch := b.steps.tool("fetch")
if len(fetch) != 2 {
t.Fatalf("fetch = %+v, want one per member", fetch)
}
for _, f := range fetch {
want := map[string][2]string{repoA: {head1, trunkA}, repoSDK: {head2, trunkS}}[f.args["repository"].(string)]
if f.args["commit"] != want[0] || f.args["base"] != want[1] || f.args["ref"] != "" {
t.Errorf("fetch = %+v, want the exact commit %s merged onto %s", f.args, want[0], want[1])
}
}
// A whole try writes the merge previews the indexer makes its candidate from; it writes no status and asks no Approve.
if keys, _ := b.prev.Keys(ctx); len(keys) != 2 {
t.Errorf("a whole try wrote the previews %v, want one per member", 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 != 2 {
t.Errorf("an ended try was fetched again: %d fetches", n)
}
// The kept verdicts: what the try passed is kept under the digests of the merged trees, so the pull-request check of
// the same head on the same trunk runs none of it again (and a push of other code runs only what changed).
before := len(b.steps.tool("test"))
out := b.judge(t, caseOf(t, pullOf(repoA, 7, head1)))
if out.Verdict.State != delivery.VerdictPass || out.Verdict.Levels[delivery.LevelUnit].Kept == 0 || out.Verdict.Levels[delivery.LevelUnit].Run != 0 ||
out.Verdict.Levels[delivery.LevelIntegration].Run != 0 {
t.Errorf("the check of the tried head = %+v, want everything the try passed kept and nothing run", out.Verdict.Levels)
}
if after := len(b.steps.tool("test")); after != before {
t.Errorf("the check of the tried head ran %d test steps, want none: the try's passes are kept", after-before)
}
}
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, Say: func(context.Context, string, tryevent.Event) error {
t.Error("a try of one level and some tests said an event: it has no candidate")
return nil
}}
raw, _, _, _ := kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
if open, _ := OpenTries(memTries{kv}); len(open) != 0 {
t.Errorf("open = %v", open)
}
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)
}
t.Run("a try of one level keeps only that level", func(t *testing.T) {
kv := newMemKV()
id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1, Level: delivery.LevelUnit}, time.Now())
b := newBench(t)
tr := &Tryer{Tries: kv, Judge: b.ex}
raw, _, _, _ := kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
keys, _ := b.kept.Keys(ctx)
if got := readTry(t, kv, id); got.State != TryPass || len(keys) == 0 {
t.Fatalf("try = %+v, kept %v", got, keys)
}
for _, k := range keys {
if !strings.HasPrefix(k, delivery.LevelUnit+".") {
t.Errorf("a try of the unit level kept %s", k)
}
}
})
}
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) != 2 || !strings.Contains(got.Failed[0], "unit: fixture/failing/bad.TestThreeIsFour") ||
!strings.Contains(got.Failed[1], "integration: fixture/failing/bad.TestThreeIsFour") { // both levels ran (the operator, 2026-11-02)
t.Errorf("try = %+v", got)
}
// A whole try that could not judge the manifests and the composition is no pass; a failed composition is named beside
// the failed tests (the group's check does as much), and a candidate of other heads is not the try's.
for _, row := range []struct {
name string
cand *fakeTryCandidates
comp delivery.CompositionVerdict
state, said string
}{
{name: "the indexer wrote no candidate", cand: &fakeTryCandidates{err: errors.New("the indexer wrote no candidate within 5m0s")}, state: TryError, said: "the indexer wrote no candidate within 5m0s"},
{name: "a node that gave no composition answer", cand: &fakeTryCandidates{}, comp: delivery.CompositionVerdict{State: delivery.CompositionPass, SDK: "x", Warnings: []string{"g14: " + delivery.ComposeNoAnswer}},
state: TryError, said: "g14: " + delivery.ComposeNoAnswer},
{name: "a composition refused on a node", cand: &fakeTryCandidates{}, comp: delivery.CompositionVerdict{State: delivery.CompositionFail, Machine: "g14", Words: "not composed on g14: web: needs a state mail does not keep"},
state: TryFail, said: "not composed on g14: web: needs a state mail does not keep"},
} {
kv := newMemKV()
b := newBench(t)
comps, _ := wholeBench(b, delivery.Candidate{})
b.ex.Candidates, comps.v = row.cand, row.comp
if row.comp.State == "" {
comps.v.State = delivery.CompositionPass
}
id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
var ends []string
tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) error {
if event == tryevent.Ended {
ends = append(ends, e.State)
}
return nil
}}
raw, _, _, _ := kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
if len(ends) != 1 || ends[0] != row.state { // every end says its state: error as well as fail
t.Errorf("%s: ends said = %v, want once %s", row.name, ends, row.state)
}
if got := readTry(t, kv, id); got.State != row.state || !strings.Contains(strings.Join(got.Failed, " | ")+got.Said, row.said) {
t.Errorf("%s: try = %s, failed %v, said %q, want %s naming %q", row.name, got.State, got.Failed, got.Said, row.state, row.said)
}
}
// No build node is no error since R4g: the try's ask waits in the check stream until a node pulls it (ADR 0525 §4).
}
func TestATryLeftRunningIsTakenOver(t *testing.T) {
ctx := context.Background()
// Whether the start was said is in the try's own state (Told), so a checker that starts later says the end of a try it did
// not start, whatever way the try ended.
for _, row := range []struct {
name string
told bool
failFirst bool // the stream does not acknowledge the first end said
after func(t *testing.T, kv *memKV, id string)
ends []string
}{
{name: "the first publish of the end is not acknowledged", told: true, failFirst: true, ends: []string{TryPass}},
{name: "left running after its start was said", told: true, ends: []string{TryPass}},
{name: "left running with no start said", told: false, ends: []string{TryPass}},
{name: "ended by hand after the start was said", told: true, ends: []string{TryError}, after: func(t *testing.T, kv *memKV, id string) {
if _, err := EndTry(endable{kv}, id, "stuck", time.Now()); err != nil {
t.Fatal(err)
}
}},
} {
t.Run(row.name, func(t *testing.T) {
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, cur.Told = TryRunning, row.told
data, _ := encodeTry(cur)
if _, err := kv.Update(ctx, id, data, rev); err != nil {
t.Fatal(err)
}
if row.after != nil {
row.after(t, kv, id)
}
b := newBench(t)
wholeBench(b, delivery.Candidate{})
var ends []string
failed := false
tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) error {
if event == tryevent.Ended {
if row.failFirst && !failed {
failed = true
return errors.New("the stream did not acknowledge")
}
ends = append(ends, e.State)
}
return nil
}}
tr.Reconsider(ctx)
tr.Wait()
if row.failFirst && !readTry(t, kv, id).Told {
t.Errorf("an end the stream did not acknowledge cleared Told: it is never said again")
}
tr.Reconsider(ctx) // said once: the second look finds nothing owed
tr.Wait()
if row.after == nil {
if got := readTry(t, kv, id); got.State != TryPass {
t.Errorf("the try left running = %+v, want taken over and passed", got)
}
}
if !slices.Equal(ends, row.ends) || readTry(t, kv, id).Told {
t.Errorf("ends said = %v, want %v, and nothing owed after", ends, row.ends)
}
})
}
// An ended try is kept TryKeep and removed by the next Reconsider after.
t.Run("an ended try is kept seven days", func(t *testing.T) {
kv := newMemKV()
id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
raw, rev, _, _ := kv.Get(ctx, id)
cur, _ := DecodeTry(raw)
ended := time.Now()
cur.State, cur.Ended = TryPass, ended
data, _ := encodeTry(cur)
if _, err := kv.Update(ctx, id, data, rev); err != nil {
t.Fatal(err)
}
now := ended.Add(TryKeep - time.Hour)
tr := &Tryer{Tries: kv, Now: func() time.Time { return now }}
tr.Reconsider(ctx)
if _, _, found, _ := kv.Get(ctx, id); !found {
t.Fatal("the try was removed before TryKeep")
}
now = ended.Add(TryKeep + time.Hour)
tr.Reconsider(ctx)
if _, _, found, _ := kv.Get(ctx, id); found {
t.Error("the try was kept past TryKeep")
}
})
}