diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go index 6030353..a5d7c6c 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go @@ -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": "/"}, + "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-"}}}, + 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) } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main_test.go index 96d29e7..4bafd10 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/mesh-delivery-checker-tools b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/mesh-delivery-checker-tools new file mode 100755 index 0000000..49d7508 Binary files /dev/null and b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/mesh-delivery-checker-tools differ diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go index e3489c3..6415687 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/bus.go @@ -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 diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index cbf005b..a44507a 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -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) + } + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 1d3a461..3320df7 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -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 diff --git a/modules/mesh-delivery-checker/internal/check/case.go b/modules/mesh-delivery-checker/internal/check/case.go index a6ea453..4240ed4 100644 --- a/modules/mesh-delivery-checker/internal/check/case.go +++ b/modules/mesh-delivery-checker/internal/check/case.go @@ -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 } diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 7e36d5b..a53aa5f 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -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 } diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index 7395ee2..a241d56 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -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)) diff --git a/modules/mesh-delivery-checker/internal/check/try.go b/modules/mesh-delivery-checker/internal/check/try.go new file mode 100644 index 0000000..ac5c6db --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/try.go @@ -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-. + 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, /", 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-", 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 +} diff --git a/modules/mesh-delivery-checker/internal/check/try_test.go b/modules/mesh-delivery-checker/internal/check/try_test.go new file mode 100644 index 0000000..65b89b5 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/try_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index bf22da6..a4b5442 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -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"} { diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 8168ace..2c22576 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -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": [