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 2fef5a6..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{ @@ -59,12 +126,69 @@ func Tools(r check.Reader) []stdio.Tool { kept, err := r.Kept(level, digest) return map[string]any{"kept": kept}, err }}, + {Name: check.PreviewsReadTool, + Description: "The merge previews the checker holds, each as stored under previews...: the merge commit, per module directory " + + "its tree, manifest digest and lock hashes, or the files in conflict. With a repository, that repository's; with a head (7 or more characters) " + + "only that head's. Keys that are no preview key are listed as unreadable. Read-only.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "repository": map[string]any{"type": "string", "description": "/"}, + "head": map[string]any{"type": "string", "description": "a pull request head commit, or a prefix of at least 7 characters"}}}, + Run: func(args map[string]any) (any, error) { + repository, _ := args["repository"].(string) + head, _ := args["head"].(string) + if head != "" && len(head) < 7 { + return nil, fmt.Errorf("the head %q is too short: give at least 7 characters", head) + } + ps, bad, err := r.Previews(repository, head) + return map[string]any{"previews": ps, "unreadable": bad}, err + }}, + {Name: check.StatusReadTool, + Description: "What holds the checker back and what runs: the capacity per node (the setting, the default one step per node), whether the controller's " + + "delivery plans in their build phase hold a key (no new check starts while one does), the steps in flight on the build modules by node, and the " + + "candidates without a verdict for their heads. The steps the checker holds waiting for a place are in its process, not in a bucket, and the " + + "answer says so. Read-only.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { return r.Status() }}, } } +func capacity() (string, error) { + path := os.Getenv("MESH_CHECKER_SETTINGS") + if path == "" { + return os.Getenv("MESH_CHECKER_CAPACITY"), nil + } + raw, err := os.ReadFile(path) + if err != nil { + return "", err + } + var s struct { + Capacity string `json:"capacity"` + } + return s.Capacity, json.Unmarshal(raw, &s) +} + func main() { - r := check.Reader{VerdictState: read(stdio.State("verdicts")), KeptState: read(stdio.State("kept")), CandidateState: read(stdio.State("mesh-delivery-indexer.candidates"))} - if err := stdio.Serve("", Tools(r)); err != nil { + r := check.Reader{VerdictState: read(stdio.State("verdicts")), KeptState: read(stdio.State("kept")), CandidateState: read(stdio.State("mesh-delivery-indexer.candidates")), + 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} + 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 5fec12b..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,12 +1,17 @@ package main import ( + "encoding/json" + "errors" + "slices" + "strings" "testing" + "time" "mesh-delivery-checker/internal/check" ) -func TestTheToolsServedAreTheThreeReadTools(t *testing.T) { +func TestTheToolsServedAreTheFiveReadTools(t *testing.T) { var names []string for _, tl := range Tools(check.Reader{}) { names = append(names, tl.Name) @@ -14,14 +19,63 @@ func TestTheToolsServedAreTheThreeReadTools(t *testing.T) { t.Errorf("%s has no description or no run", tl.Name) } } - want := []string{check.VerdictsReadTool, check.PullReadTool, check.KeptReadTool} - if len(names) != 3 || names[0] != want[0] || names[1] != want[1] || names[2] != want[2] { + want := []string{check.VerdictsReadTool, check.PullReadTool, check.KeptReadTool, check.PreviewsReadTool, check.StatusReadTool} + if !slices.Equal(names, want) { t.Errorf("tools = %v, want %v", names, want) } // With nothing to read from a tool says so and changes nothing. for _, tl := range Tools(check.Reader{}) { + if tl.Name == check.StatusReadTool { + continue // it answers from what it can read, and says what it could not + } if _, err := tl.Run(map[string]any{"repository": "novox/x", "number": float64(1)}); err == nil { t.Errorf("%s answered with no state to read", tl.Name) } } } + +// 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/forge.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/forge.go index 31702bf..b3aabfa 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/forge.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/forge.go @@ -3,10 +3,13 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "regexp" "strings" + "github.com/nats-io/nats.go" + "mesh-delivery-checker/internal/check" ) @@ -27,6 +30,16 @@ type forgeReads struct { call func(key string, body any) (json.RawMessage, error) } +// ask calls a forge tool; a call that reached nobody (no responders: the forge's tool runner re-serving, or the forge +// down) is marked Unheard, so the check fails at the forge and is judged again when the nodes change (issue 1624). +func (f forgeReads) ask(key string, body any) (json.RawMessage, error) { + raw, err := f.call(key, body) + if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrTimeout) || errors.Is(err, context.DeadlineExceeded) { + return nil, &check.Unheard{Err: err} + } + return raw, err +} + func ownerRepo(repository string) (string, string) { owner, repo, _ := strings.Cut(repository, "/") return owner, repo @@ -35,7 +48,7 @@ func ownerRepo(repository string) (string, string) { // File is the text of a file of a repository at a ref. func (f forgeReads) File(_ context.Context, repository, path, ref string) (string, error) { owner, repo := ownerRepo(repository) - raw, err := f.call(getFileTool, map[string]string{"owner": owner, "repo": repo, "path": path, "ref": ref}) + raw, err := f.ask(getFileTool, map[string]string{"owner": owner, "repo": repo, "path": path, "ref": ref}) if err != nil { if notFoundStatus.MatchString(err.Error()) { // The forge answers 404 for a file the ref does not hold (issue 1845). The status is matched in its own place, @@ -59,7 +72,7 @@ func (f forgeReads) File(_ context.Context, repository, path, ref string) (strin // not. func (f forgeReads) OnTrunk(_ context.Context, repository, commit string) (bool, error) { owner, repo := ownerRepo(repository) - raw, err := f.call(compareTool, map[string]string{"owner": owner, "repo": repo, "base": "main", "head": commit}) + raw, err := f.ask(compareTool, map[string]string{"owner": owner, "repo": repo, "base": "main", "head": commit}) if err != nil { return false, err } @@ -75,7 +88,7 @@ func (f forgeReads) OnTrunk(_ context.Context, repository, commit string) (bool, // FullSHA is the whole sha of a commit named by its first digits: the newest commit of the history at that ref. func (f forgeReads) FullSHA(_ context.Context, repository, commit string) (string, error) { owner, repo := ownerRepo(repository) - raw, err := f.call(commitsTool, map[string]any{"owner": owner, "repo": repo, "ref": commit, "limit": 1}) + raw, err := f.ask(commitsTool, map[string]any{"owner": owner, "repo": repo, "ref": commit, "limit": 1}) if err != nil { return "", err } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/forge_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/forge_test.go index e7ecc85..d6487d6 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/forge_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/forge_test.go @@ -9,6 +9,8 @@ import ( "strings" "testing" + "github.com/nats-io/nats.go" + "mesh-delivery-checker/internal/check" ) @@ -86,3 +88,18 @@ func TestAFileTheRefDoesNotHoldIsNotFoundAndOtherFailuresAreNot(t *testing.T) { t.Errorf("a 500 is %v, want a failure that is not ErrNotFound", err) } } + +func TestAForgeReadThatReachesNobodyIsUnheard(t *testing.T) { // 1624 C7: "nats: no responders" on mesh-catalog #293 + f := forgeReads{call: func(string, any) (json.RawMessage, error) { return nil, nats.ErrNoResponders }} + _, err := f.File(context.Background(), "novox/mesh-catalog", "checks/agenthome/go.mod", "950e37a46e8d") + var u *check.Unheard + if !errors.As(err, &u) { + t.Errorf("err = %v, want it marked as reaching nobody", err) + } + if _, err := f.OnTrunk(context.Background(), "novox/mesh-sdk", "abc"); !errors.As(err, &u) { + t.Errorf("compare: %v", err) + } + if _, err := f.FullSHA(context.Background(), "novox/mesh-sdk", "abc"); !errors.As(err, &u) { + t.Errorf("commits: %v", err) + } +} 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/judge.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go index 6dc5381..ed8ff96 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go @@ -3,6 +3,7 @@ package main import ( "context" "sort" + "strings" "sync" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -31,7 +32,7 @@ type judgeDeps struct { func newJudge(d judgeDeps) check.Judge { return &check.Examiner{Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places, - Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf} + Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Packages: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf} } // manifestKV reads a manifest by its digest. @@ -47,10 +48,13 @@ type repoModules struct { logf func(format string, args ...any) mu sync.Mutex - langs map[string]map[string]bool // by manifest digest: the languages of its artifacts - repoOf map[string]string // by module: its repository - commit map[string]string // by module: the commit its record was computed at (the commit the mesh runs) - typed map[string]string // by module: its manifest digest + langs map[string]map[string]bool // by manifest digest: the languages of its artifacts + tsPkgs map[string][]string // by manifest digest: the names of its TypeScript package artifacts + path map[string]string // by module: its directory in its repository + named map[string]map[string][]string // by module: each artifact's named paths, as the indexer keeps them + repoOf map[string]string // by module: its repository + commit map[string]string // by module: the commit its record was computed at (the commit the mesh runs) + typed map[string]string // by module: its manifest digest } // Put takes a change of the indexer's modules bucket in: a module's repository and the languages its manifest names. @@ -58,11 +62,14 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted r.mu.Lock() if r.repoOf == nil { r.repoOf, r.typed, r.langs, r.commit = map[string]string{}, map[string]string{}, map[string]map[string]bool{}, map[string]string{} + r.tsPkgs, r.path, r.named = map[string][]string{}, map[string]string{}, map[string]map[string][]string{} } if deleted { delete(r.repoOf, key) delete(r.typed, key) delete(r.commit, key) + delete(r.path, key) + delete(r.named, key) r.mu.Unlock() return } @@ -75,12 +82,15 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted r.repoOf[key] = m.Repository r.typed[key] = m.Manifest r.commit[key] = m.Commit + r.path[key] = m.Path + r.named[key] = m.ArtifactPaths _, known := r.langs[m.Manifest] r.mu.Unlock() if m.Manifest == "" || known { return } set := map[string]bool{} + var pkgs []string if mk, err := delivery.ManifestsKey(m.Manifest); err == nil { if raw, _, found, err := r.manifests.Get(ctx, mk); err == nil && found { if kept, err := delivery.DecodeManifest(raw); err == nil { @@ -89,6 +99,9 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted if a.Language != "" { set[a.Language] = true } + if a.Kind == "package" && a.Language == "typescript" { + pkgs = append(pkgs, a.Name) + } } } else if err != nil && r.logf != nil { r.logf("the manifest %s cannot be read, so its module counts as no TypeScript: %v", m.Manifest, err) @@ -98,9 +111,37 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted } r.mu.Lock() r.langs[m.Manifest] = set + r.tsPkgs[m.Manifest] = pkgs r.mu.Unlock() } +// PackagePaths are the paths, relative to the repository, that the TypeScript package artifacts of a repository's modules +// are built from, sorted: the named paths the indexer keeps in each module's value (delivery.Module.ArtifactPaths; only the +// indexer reads them from a manifest, the lead's decision L8), under the module's directory. None for a module whose value +// holds no paths. +func (r *repoModules) PackagePaths(repository string) []string { + r.mu.Lock() + defer r.mu.Unlock() + var out []string + for module, repo := range r.repoOf { + if repo != repository { + continue + } + dir := strings.Trim(r.path[module], "/") + for _, name := range r.tsPkgs[r.typed[module]] { + for _, p := range r.named[module][name] { + p = strings.Trim(strings.TrimPrefix(p, "./"), "/") + if dir != "" && dir != delivery.RootPath { + p = dir + "/" + p + } + out = append(out, p) + } + } + } + sort.Strings(out) + return out +} + // BuildModules are the build modules that judge a repository, sorted. func (r *repoModules) BuildModules(repository string) []string { out := []string{buildrun.GoBuildModule} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge_test.go index 28329bf..8afa3c1 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge_test.go @@ -73,3 +73,40 @@ func TestARepositoryWithATypeScriptModuleIsJudgedByTheNodeBuildModuleToo(t *test t.Errorf("a repository with no module: %v, want the Go build module", got) } } + +// sdkManifest is mesh-sdk's own manifest (its module.json at feat/r4e-sdk): one TypeScript package naming its paths. +const sdkManifest = `{"module":"mesh-sdk","version":"1","slug":"sdk","build":{"artifacts":[{"name":"lib","kind":"package","language":"typescript",` + + `"paths":["src","test","package.json","package-lock.json","tsconfig.json"]}]},"resources":[]}` + +func TestTheTypeScriptPackagesPathsComeFromTheIndexersModulesNotTheManifest(t *testing.T) { // ADR 0500 §5 with lead decision L8 + store := memManifests{} + data, err := delivery.EncodeManifest(delivery.Manifest{Module: "mesh-sdk", JSON: sdkManifest}) + if err != nil { + t.Fatal(err) + } + store[delivery.ManifestDigest([]byte(sdkManifest))] = data + r := &repoModules{manifests: store} + put := func(path string, paths map[string][]string) { + v, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-sdk", Commit: strings.Repeat("a", 40), FP: "src1:" + strings.Repeat("a", 64), + Manifest: delivery.ManifestDigest([]byte(sdkManifest)), Path: path, ArtifactPaths: paths}) + if err != nil { + t.Fatal(err) + } + r.Put(context.Background(), "mesh-sdk", v, false) + } + put(".", map[string][]string{"lib": {"src", "package.json"}}) // the indexer's word, not the manifest's five paths + if got := strings.Join(r.PackagePaths("novox/mesh-sdk"), ","); got != "package.json,src" { + t.Errorf("paths = %q, want the indexer's, at the repository's root", got) + } + put("ts", map[string][]string{"lib": {"src"}}) + if got := strings.Join(r.PackagePaths("novox/mesh-sdk"), ","); got != "ts/src" { + t.Errorf("paths = %q, want them under the module's directory", got) + } + put(".", nil) // a value written before the indexer kept paths: nothing to serve by + if got := r.PackagePaths("novox/mesh-sdk"); len(got) != 0 { + t.Errorf("paths = %v, want none", got) + } + if got := r.PackagePaths("novox/mesh-delivery"); len(got) != 0 { + t.Errorf("another repository: %v", got) + } +} 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 40ed0d7..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) @@ -308,7 +312,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go func() { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { if e.Key() == delivery.SourceMainKey && trunks.Put(e.Value(), !opOf(e)) { - ch.Reconsider(ctx) // the trunk moved: a verdict for the old trunk is judged again + ch.TrunkMoved(ctx) // the trunk moved: its statuses go pending, and a verdict for the old trunk is judged again } }) }() @@ -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/go.mod b/modules/mesh-delivery-checker/go.mod index 9c911a8..8f008d1 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index 3515fb8..bb52af2 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/modules/mesh-delivery-checker/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc h1:ZJScJSnF6LgFl55/VJ9RT/lNroVWun469sXtx0nlVYw= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7 h1:wK9O1CgNk7mo3IdJqW/CCx/hebg+1F0rkvW8CvJfnRM= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-checker/internal/check/case.go b/modules/mesh-delivery-checker/internal/check/case.go index 24cb19e..4240ed4 100644 --- a/modules/mesh-delivery-checker/internal/check/case.go +++ b/modules/mesh-delivery-checker/internal/check/case.go @@ -4,6 +4,7 @@ import ( "fmt" "maps" "sort" + "strings" "sync" "time" @@ -35,13 +36,17 @@ type Case struct { Key string Group string HeadsHash string - Pulls []Pull + // Heads is the heads hash the gate matches an Approve's members by (GateHeads): a verdict carries it. + Heads string + Pulls []Pull // Trunk is the trunk head of each repository of Pulls. Trunk map[string]string // Left says why pull requests of the candidate are not judged (hq's, a repository with no known trunk); empty when all are. 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 } @@ -75,7 +80,7 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case { if err != nil { return nil } - cs := &Case{Key: key, Group: c.Group, HeadsHash: c.HeadsHash, Trunk: map[string]string{}, Progress: &Progress{}} + cs := &Case{Key: key, Group: c.Group, HeadsHash: c.HeadsHash, Heads: GateHeads(c), Trunk: map[string]string{}, Progress: &Progress{}} for _, p := range c.Pulls { switch head, known := trunk[p.Repository]; { case p.Repository == HQRepository: @@ -99,6 +104,22 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case { return cs } +// GateHeads is the hash the gate matches the members an Approve covers by (mesh-catalog mesh-delivery-gate onestatus.go +// headsOf): delivery.HeadsHash over every pull request of the candidate, hq's included, as +// `#=`. A lone pull request is a group of one. Empty for a candidate +// with no pull request. +func GateHeads(c delivery.Candidate) string { + heads := make([]delivery.Head, 0, len(c.Pulls)) + for _, p := range c.Pulls { + heads = append(heads, delivery.Head{Repository: strings.ToLower(p.Repository), Number: p.Number, Head: strings.ToLower(p.Head.Head)}) + } + hash, err := delivery.HeadsHash(heads) + if err != nil { + return "" + } + return hash +} + // sameTrunk says whether two verdicts' trunk heads are the same. func sameTrunk(a, b map[string]string) bool { return maps.Equal(a, b) } diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 84753ed..1236af4 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -18,6 +18,10 @@ const DefaultBound = 30 * time.Minute // is judged again when the nodes change, by a watch and never a timer (ADR 0437 §8). const NodesWords = "build node qualifies" +// UnheardWords end the Words of a failure because a part of the mesh the check reads was not heard (the forge, issue 1624): +// like a failure of the nodes, it is not the change's, and is judged again when the nodes change. +const UnheardWords = "not heard; judged again when the nodes change" + // The events the checker says on its stream checks (module.json): a preview or a status was written, so a part that reads // the bucket, which is made when the checker is assigned and may not exist when that part starts, finds it (ADR 0437 §8: no // timer). The body is {"key": ...}. @@ -234,7 +238,7 @@ type claimed struct{ rev uint64 } // - a pass or a failure for the trunk heads the case has stands; one for other trunk heads (the trunk moved) is taken // again by compare-and-set on the same key, and so is a failure of the nodes after the nodes changed. func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64) (*claimed, error) { - running := delivery.Verdict{Group: cs.Group, Heads: cs.HeadsHash, Trunk: cs.Trunk, State: delivery.VerdictRunning} + running := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictRunning} data, err := delivery.EncodeVerdict(running) if err != nil { return nil, err @@ -287,7 +291,7 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64 // IsNodesFailure says whether a verdict failed because no build node could take the check. func IsNodesFailure(v delivery.Verdict) bool { - return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, NodesWords) + return v.State == delivery.VerdictFail && (strings.Contains(v.Failed.Words, NodesWords) || strings.Contains(v.Failed.Words, UnheardWords)) } // try claims the check, judges it within its bound and writes the verdict and the owed statuses. A check that ends @@ -313,14 +317,14 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) { return } v := out.Verdict - v.Group, v.Heads, v.Trunk = cs.Group, cs.HeadsHash, cs.Trunk + v.Group, v.Heads, v.Trunk = cs.Group, cs.Heads, cs.Trunk if v.Took.Queue == 0 && !cs.Queued.IsZero() && began.After(cs.Queued) { v.Took.Queue = began.Sub(cs.Queued).Milliseconds() } data, err := delivery.EncodeVerdict(v) if err != nil { ch.logf("%s: the verdict cannot be written: %v", cs.Key, err) - v = delivery.Verdict{Group: cs.Group, Heads: cs.HeadsHash, Trunk: cs.Trunk, State: delivery.VerdictFail, + v = delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictFail, Failed: delivery.FailedFirst{Machine: "checker", Words: "the verdict could not be written: " + err.Error()}} if data, err = delivery.EncodeVerdict(v); err != nil { return @@ -335,6 +339,13 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) { ch.nodesFail[cs.Key] = epoch ch.mu.Unlock() } + if ch.movedSince(cs) { + // Judged on a trunk that moved while it ran: the verdict is kept as judged, and the status stays pending until the + // check of the new trunk, which Reconsider starts, ends. + ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, nil, nil) + ch.logf("%s: %s on a trunk that moved while it ran; the status stays pending", cs.Key, v.State) + return + } ch.writeOwed(ctx, cs, v, out.Builds, out.Slowest) ch.logf("%s: %s", cs.Key, v.State) } @@ -342,25 +353,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 } @@ -467,3 +460,44 @@ func (ch *Checker) Sweep(ctx context.Context) error { } return nil } + +// TrunkMoved is Reconsider after a move of a trunk, which first owes mesh/check pending on every open pull request of a +// candidate whose verdict was judged on other trunk heads, every member's for a group (the lead's decision L4, the checker's +// part of issue 1302): the forge's success of the old trunk never stands while the check waits for a place or for a delivery +// that holds it, so the gate never reads it as the head's. +func (ch *Checker) TrunkMoved(ctx context.Context) { + ch.init() + ch.mu.Lock() + known := make(map[string]delivery.Candidate, len(ch.known)) + for k, c := range ch.known { + known[k] = c + } + ch.mu.Unlock() + heads := ch.heads() + for _, c := range known { + cs := CaseOf(c, heads) + if cs == nil { + continue + } + value, _, found, err := ch.Verdicts.Get(ctx, cs.Key) + if err != nil || !found { + continue // never judged: no status to put out of date + } + if v, err := delivery.DecodeVerdict(value); err == nil && sameTrunk(v.Trunk, cs.Trunk) { + continue + } + ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, nil, nil) + } + ch.Reconsider(ctx) +} + +// movedSince says whether a trunk the case was judged on is no longer the trunk's head. +func (ch *Checker) movedSince(cs *Case) bool { + now := ch.heads() + for repo, head := range cs.Trunk { + if now[repo] != head { + return true + } + } + return false +} diff --git a/modules/mesh-delivery-checker/internal/check/checker_test.go b/modules/mesh-delivery-checker/internal/check/checker_test.go index f57ed71..295a9ff 100644 --- a/modules/mesh-delivery-checker/internal/check/checker_test.go +++ b/modules/mesh-delivery-checker/internal/check/checker_test.go @@ -19,6 +19,7 @@ const ( trunkA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" trunkA2 = "abababababababababababababababababababab" trunkS = "5555555555555555555555555555555555555555" + trunkS2 = "5656565656565656565656565656565656565656" ) // candidateOf is a candidate of a group with the given pull requests, as the indexer writes one. diff --git a/modules/mesh-delivery-checker/internal/check/examiner_test.go b/modules/mesh-delivery-checker/internal/check/examiner_test.go index 66edd34..1864f55 100644 --- a/modules/mesh-delivery-checker/internal/check/examiner_test.go +++ b/modules/mesh-delivery-checker/internal/check/examiner_test.go @@ -2,10 +2,14 @@ package check import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "fmt" "os" "path/filepath" + "reflect" + "sort" "strings" "sync" "testing" @@ -565,7 +569,7 @@ func trunksOf(t *testing.T, heads map[string]string) *Trunks { return tr } -func TestNothingIsKeptForAWorkspaceGivenSiblingsBeside(t *testing.T) { // review D1, ADR 0444, issue 1287 +func TestNothingIsKeptForAWorkspaceGivenSiblingsWhenTheDigestDoesNotSayItCoversThem(t *testing.T) { // review D1, ADR 0444, issue 1287 b := newBench(t) b.ex.Trunks = trunksOf(t, map[string]string{repoSDK: trunkS, "novox/mesh-catalog": trunkA}) b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) @@ -573,7 +577,7 @@ func TestNothingIsKeptForAWorkspaceGivenSiblingsBeside(t *testing.T) { // review t.Fatalf("the fetch was not given siblings: %+v", b.steps.tool("fetch")) } if keys, _ := b.kept.Keys(context.Background()); len(keys) != 0 { - t.Errorf("kept %v for a package whose tests may read a sibling the digest does not cover", keys) + t.Errorf("kept %v for a package whose digest does not say it covers the siblings", keys) } // And a kept verdict of the same digest from elsewhere is not taken for it: everything runs. if _, err := (Kept{Store: b.kept}).Keep(context.Background(), "unit", digA, keptValue("fixture/slow", 2)); err != nil { @@ -585,3 +589,137 @@ func TestNothingIsKeptForAWorkspaceGivenSiblingsBeside(t *testing.T) { // review t.Errorf("%d test steps, want both levels run whole", n) } } + +// besideDigest stands for the Go build module's digest over a workspace with siblings (DigestVersion 3, lead decision L5): +// each package's digest folds the siblings' commits in, and the answer echoes the lines it folded. +func besideDigest(b *bench) func(c stepCall) (any, error) { + return func(c stepCall) (any, error) { + level := c.args["level"].(string) + var lines []string + for _, f := range b.steps.tool("fetch") { + refs, _ := f.args["beside"].([]BesideRef) + for _, r := range refs { + lines = append(lines, r.Repository+" "+r.Commit) + } + } + sort.Strings(lines) + sum := sha256.Sum256([]byte(level + "\n" + strings.Join(lines, "\n"))) + return Digested{Level: level, Packages: map[string]string{"fixture/slow": hex.EncodeToString(sum[:])}, Beside: lines}, nil + } +} + +func TestACoreRepositorysPassingPackageIsKeptUnderADigestCoveringItsSiblings(t *testing.T) { // 1287 AC1-AC3, AC7 with L5 + b := newBench(t) + b.steps.handlers["build_go_digest"] = besideDigest(b) + b.ex.Trunks = trunksOf(t, map[string]string{repoSDK: trunkS, "novox/mesh-catalog": trunkA}) + b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) + keys, _ := b.kept.Keys(context.Background()) + if len(keys) != 2 { + t.Fatalf("kept %v, want the passing package at both levels", keys) + } + // The same siblings again: everything is taken, nothing runs. + b.steps.calls = nil + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict + if n := len(b.steps.tool("test")); n != 0 || v.Levels[delivery.LevelUnit].Run != 0 || v.Levels[delivery.LevelUnit].Kept == 0 { + t.Fatalf("%d test steps, unit %+v, want none run and the kept verdict taken", n, v.Levels[delivery.LevelUnit]) + } + // A sibling's commit moves: a new digest, so the package runs again and is kept under it. + b.ex.Trunks = trunksOf(t, map[string]string{repoSDK: trunkS2, "novox/mesh-catalog": trunkA}) + b.steps.calls = nil + b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) + if n := len(b.steps.tool("test")); n != 2 { + t.Errorf("%d test steps after a sibling moved, want both levels run", n) + } + if keys, _ := b.kept.Keys(context.Background()); len(keys) != 4 { + t.Errorf("kept %v, want a second key per level under the new digest", keys) + } +} + +func TestADigestThatEchoesOtherSiblingsKeepsNothing(t *testing.T) { // 1287 with L5: an echo that differs is not trusted + b := newBench(t) + inner := besideDigest(b) + b.steps.handlers["build_go_digest"] = func(c stepCall) (any, error) { + v, err := inner(c) + d := v.(Digested) + d.Beside = d.Beside[:len(d.Beside)-1] + return d, err + } + b.ex.Trunks = trunksOf(t, map[string]string{repoSDK: trunkS, "novox/mesh-catalog": trunkA}) + b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) + if keys, _ := b.kept.Keys(context.Background()); len(keys) != 0 { + t.Errorf("kept %v under a digest that does not cover every sibling", keys) + } +} + +type pkgPaths map[string][]string + +func (p pkgPaths) PackagePaths(repository string) []string { return p[repository] } + +// groupPackageBench is a group of a TypeScript member (mesh-delivery) and the mesh-sdk member, whose merge changes sdkChanged. +func groupPackageBench(t *testing.T, sdkChanged []string) *bench { + b := newBench(t) + b.ex.Modules = langs{repoA: {buildrun.GoBuildModule, buildrun.NodeBuildModule}, repoSDK: {buildrun.GoBuildModule, buildrun.NodeBuildModule}} + b.ex.Packages = pkgPaths{repoSDK: {"src", "test", "package.json", "package-lock.json", "tsconfig.json"}} + fetch := func(c stepCall) (any, error) { + f := b.fetch + if c.args["repository"] == repoSDK { + f.Changed = sdkChanged + } + return f, nil + } + b.steps.handlers["build_go_fetch"], b.steps.handlers["build_node_fetch"] = fetch, fetch + return b +} + +func nodeFetchOf(b *bench, repo string) stepCall { + for _, c := range b.steps.tool("fetch") { + if c.module == buildrun.NodeBuildModule && c.args["repository"] == repo { + return c + } + } + return stepCall{} +} + +func TestATypeScriptMemberTakesTheGroupsPackageWhenTheSDKMemberChangesItsPaths(t *testing.T) { // ADR 0500 §5, C11b + b := groupPackageBench(t, []string{"go/delivery/x.go", "src/state.ts"}) + b.judge(t, caseOf(t, pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2))) + want := map[string]any{"repository": repoSDK, "commit": head2, "ref": "refs/pull/12/head"} + if got := nodeFetchOf(b, repoA).args["package"]; !reflect.DeepEqual(got, want) { + t.Errorf("mesh-delivery's Node fetch was given the package %v, want %v", got, want) + } + if got := nodeFetchOf(b, repoSDK).args["package"]; got != nil { + t.Errorf("the mesh-sdk member's own fetch was given %v: it builds its package itself", got) + } +} + +func TestATypeScriptMemberResolvesTheRegistryWhenTheSDKMemberLeavesItsPaths(t *testing.T) { // ADR 0500 §5: the manifest alone, or Go alone + for _, changed := range [][]string{{"go/delivery/x.go"}, {"module.json"}, {"srcx/a.ts"}} { + b := groupPackageBench(t, changed) + b.judge(t, caseOf(t, pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2))) + if got := nodeFetchOf(b, repoA).args["package"]; got != nil { + t.Errorf("%v: the Node fetch was given %v, want the registry", changed, got) + } + } + // A lone TypeScript pull request: no group package either. + b := groupPackageBench(t, []string{"src/state.ts"}) + b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) + if got := nodeFetchOf(b, repoA).args["package"]; got != nil { + t.Errorf("a lone pull request's Node fetch was given %v", got) + } +} + +func TestNoStepOfACheckCarriesPublish(t *testing.T) { // ADR 0500 §4, 1309 (the checker's half): only the builder's trunk build publishes + b := groupPackageBench(t, []string{"src/state.ts"}) + b.judge(t, caseOf(t, pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2))) + if len(b.steps.calls) == 0 { + t.Fatal("no step ran") + } + for _, c := range b.steps.calls { + if _, ok := c.args["publish"]; ok { + t.Errorf("%s was given publish: %v", c.tool, c.args) + } + if strings.HasSuffix(c.tool, "_build") || strings.HasSuffix(c.tool, "_pack") { + t.Errorf("the checker asked %s: a check builds no artifact of its own", c.tool) + } + } +} diff --git a/modules/mesh-delivery-checker/internal/check/gate_test.go b/modules/mesh-delivery-checker/internal/check/gate_test.go new file mode 100644 index 0000000..6c709e0 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/gate_test.go @@ -0,0 +1,106 @@ +package check + +import ( + "context" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// What the gate reads of the checker (R4e, lead's interface with dev-r4e-gate): the verdict's heads are the gate's hash over +// `#=`, a lone pull request is a group of one, and a move of a trunk marks +// mesh/check pending on every open pull request into it, every member's for a group, before any check starts. + +func owedState(t testing.TB, r *rig, repo string, number int, head string) string { + t.Helper() + key, err := delivery.CheckerOwedKey(repo, number, head) + if err != nil { + t.Fatal(err) + } + raw, _, found, _ := r.owed.Get(context.Background(), key) + if !found { + return "" + } + o, err := delivery.DecodeCheckerOwed(raw) + if err != nil { + t.Fatal(err) + } + return o.State +} + +func TestTheVerdictsHeadsAreTheGatesHashOverLowercaseHeads(t *testing.T) { // gate (b), (c) + ctx := context.Background() + r := newRig(t, passJudge()) + upper := strings.ToUpper(head1[:20]) + head1[20:] + r.setTrunks(t, map[string]string{"Novox/Mesh-Delivery": trunkA, repoA: trunkA, repoSDK: trunkS}) + cand := candidateOf(t, "feat/x", pullOf("Novox/Mesh-Delivery", 7, upper)) + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + want, _ := delivery.HeadsHash([]delivery.Head{{Repository: repoA, Number: 7, Head: head1}}) + if v := r.verdict(t, keyOf(t, "feat/x", cand)); v.Heads != want { + t.Errorf("heads = %q, want the gate's hash of the lone pull request, a group of one, %q", v.Heads, want) + } +} + +func TestAGroupsVerdictHeadsCoverEveryMember(t *testing.T) { // gate (b): hq's member too, as the gate's ask covers it + ctx := context.Background() + r := newRig(t, passJudge()) + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2), pullOf(HQRepository, 3, head2)) + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + want, _ := delivery.HeadsHash([]delivery.Head{{Repository: repoA, Number: 7, Head: head1}, {Repository: repoSDK, Number: 12, Head: head2}, + {Repository: HQRepository, Number: 3, Head: head2}}) + if v := r.verdict(t, keyOf(t, "feat/x", cand)); v.Heads != want { + t.Errorf("heads = %q, want %q", v.Heads, want) + } +} + +func TestATrunkMoveMarksEveryOpenPullRequestIntoItPendingAtOnce(t *testing.T) { // gate (a), lead decision L4 (1302's checker part) + ctx := context.Background() + r := newRig(t, passJudge()) + r.ch.Candidate(ctx, "feat%2Fg", candidateOf(t, "feat/g", pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2))) + r.ch.Candidate(ctx, "feat%2Fs", candidateOf(t, "feat/s", pullOf(repoSDK, 13, head1))) + r.ch.Wait() + for _, p := range []struct { + repo string + n int + head string + }{{repoA, 7, head1}, {repoSDK, 12, head2}, {repoSDK, 13, head1}} { + if s := owedState(t, r, p.repo, p.n, p.head); s != "success" { + t.Fatalf("%s#%d is %q before the move", p.repo, p.n, s) + } + } + // A delivery builds, so no check starts; mesh-delivery's trunk moves. + r.ch.Hold = func() bool { return true } + r.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS}) + r.ch.TrunkMoved(ctx) + r.ch.Wait() + if s := owedState(t, r, repoA, 7, head1); s != "pending" { + t.Errorf("mesh-delivery#7, into the moved trunk, is %q, want pending", s) + } + if s := owedState(t, r, repoSDK, 12, head2); s != "pending" { + t.Errorf("mesh-sdk#12, a member of the same group, is %q, want pending", s) + } + if s := owedState(t, r, repoSDK, 13, head1); s != "success" { + t.Errorf("mesh-sdk#13, alone into a trunk that did not move, is %q, want its success to stand", s) + } +} + +func TestACheckThatEndsAfterItsTrunkMovedOwesPendingNotItsVerdict(t *testing.T) { // gate (a): a pass on an old trunk is never shown + ctx := context.Background() + var r *rig + j := &judgeFunc{f: func(ctx context.Context, c *Case, call int) (Outcome, error) { + if call == 1 { + r.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS}) // the trunk moves while it runs + r.ch.Hold = func() bool { return true } // and a delivery holds the next check + } + return Outcome{Verdict: passing()}, nil + }} + r = newRig(t, j) + r.ch.Candidate(ctx, "feat%2Fx", candidateOf(t, "feat/x", pullOf(repoA, 7, head1))) + r.ch.Wait() + if s := owedState(t, r, repoA, 7, head1); s != "pending" { + t.Errorf("the status is %q, want pending: the pass was judged on a trunk that moved", s) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index df003a4..5593b58 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "path" + "slices" "sort" "strings" "time" @@ -68,6 +69,11 @@ type Examiner struct { Builds BuildStates // Announce says a preview was written under the key (EventPreviewWritten); nil says nothing. Announce Announcer + // Packages names, of a repository, the paths (repository-relative) its TypeScript package artifacts are built from, from + // the indexer's modules (delivery.Module.ArtifactPaths, lead decision L8); nil serves no group package. + Packages interface { + PackagePaths(repository string) []string + } // Trunks are the trunk heads of every repository, for the siblings a repository's tests read. Trunks interface{ Heads() map[string]string } Now func() time.Time @@ -87,6 +93,16 @@ func (j *Examiner) logf(format string, args ...any) { } } +// besideLines are the siblings as the digest step echoes them: ` `, sorted. +func besideLines(refs []BesideRef) []string { + out := make([]string, 0, len(refs)) + for _, r := range refs { + out = append(out, r.Repository+" "+r.Commit) + } + sort.Strings(out) + return out +} + // target is a pull request on one build module: the node chosen, the workspace its fetch made and what it said. type target struct { pull Pull @@ -98,8 +114,9 @@ type target struct { fetchArgs map[string]any // priority are the Go packages (import paths) or the Node module directories whose test files the merge changes. changed []string - // beside says the workspace was given siblings: its tests may read what the digest does not cover, so nothing is kept for it. - beside bool + // beside are the lines ` ` of the siblings the workspace was given, sorted: its tests may read them, + // so a verdict is kept or taken only under a digest that says it folded exactly these in (lead decision L5). + beside []string plans map[string]*goPlan // by level } @@ -142,6 +159,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 { @@ -254,7 +274,20 @@ func (st *state) choose(bm string) (node, toolchain string, err error) { func (st *state) fetchAll(ctx context.Context) *delivery.Verdict { began := st.j.now() defer func() { st.clone = st.j.now().Sub(began) }() - for _, p := range st.c.Pulls { + // The mesh-sdk member first: whether its merge changes the TypeScript package's paths decides whether the other members' + // Node fetches take the group's package (ADR 0500 §5). + pulls := slices.Clone(st.c.Pulls) + slices.SortStableFunc(pulls, func(a, b Pull) int { + if (a.Repository == SDKRepository) != (b.Repository == SDKRepository) { + if a.Repository == SDKRepository { + return -1 + } + return 1 + } + return 0 + }) + var groupPackage map[string]any + for _, p := range pulls { var merged []Fetched for _, bm := range st.j.Modules.BuildModules(p.Repository) { node, tc, err := st.choose(bm) @@ -265,14 +298,17 @@ func (st *state) fetchAll(ctx context.Context) *delivery.Verdict { st.nodes[node] = true args := map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number), "base": p.Trunk} t.fetchArgs = args + if bm == buildrun.NodeBuildModule && groupPackage != nil && p.Repository != SDKRepository { + args["package"] = groupPackage + } if bm == buildrun.GoBuildModule { beside, err := st.j.besideFor(ctx, st.c, p) if err != nil { - return fail(node, err.Error()) + return fail(forgeMachine(err, node), err.Error()+unheardWords(err)) } if len(beside) > 0 { args["beside"] = beside - t.beside = true + t.beside = besideLines(beside) } } raw, err := st.step(ctx, t, "fetch", fmt.Sprintf("fetch %s#%d on %s", p.Repository, p.Number, node), args) @@ -284,18 +320,45 @@ 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) + } + if p.Repository == SDKRepository && len(st.c.Pulls) > 1 && st.j.changesPackage(p, merged) { + groupPackage = map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number)} + } } return nil } +// changesPackage says whether the merge of a pull request changes a file inside the named paths of one of its repository's +// TypeScript packages: a change of the manifest alone, or of files outside those paths, does not (ADR 0500 §5). +func (j *Examiner) changesPackage(p Pull, merged []Fetched) bool { + if j.Packages == nil { + return false + } + paths := j.Packages.PackagePaths(p.Repository) + for _, f := range merged { + for _, changed := range f.Changed { + for _, named := range paths { + named = strings.TrimSuffix(strings.TrimPrefix(named, "./"), "/") + if changed == named || strings.HasPrefix(changed, named+"/") { + return true + } + } + } + } + return false +} + // isTestFile says whether a repository-relative path is a test file of the languages the checker judges: Go's _test.go and // the TypeScript module's .test. and .spec. files and test directories. func isTestFile(p string) bool { @@ -398,9 +461,8 @@ func (st *state) planGo(ctx context.Context, t *target, level string) *delivery. } plan := &goPlan{digests: map[string]string{}} t.plans[level] = plan - if t.beside { - // The digest covers the package's own files, the toolchain and the header, not the siblings placed beside the tree: a - // package whose tests read one must never be kept, nor taken from a verdict kept for other inputs (ADR 0444). + 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 } @@ -421,6 +483,13 @@ func (st *state) planGo(ctx context.Context, t *target, level string) *delivery. plan.all = true return nil } + if !slices.Equal(d.Beside, t.beside) { + // The siblings the tests read are not all in the digest (a build module before DigestVersion 3, or another set): a + // package whose tests read one is never kept, nor taken from a verdict kept for other inputs (ADR 0444, L5). + st.j.logf("%s#%d: the %s level's digest covers the siblings %v, not %v, so it runs whole and keeps nothing", t.pull.Repository, t.pull.Number, level, d.Beside, t.beside) + plan.all = true + return nil + } plan.digests = d.Packages a := st.levels[level] pkgs := make([]string, 0, len(d.Packages)) @@ -467,6 +536,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/line.go b/modules/mesh-delivery-checker/internal/check/line.go index 0882cf1..cedef15 100644 --- a/modules/mesh-delivery-checker/internal/check/line.go +++ b/modules/mesh-delivery-checker/internal/check/line.go @@ -2,6 +2,7 @@ package check import ( "fmt" + "slices" "strings" "unicode/utf8" @@ -50,7 +51,7 @@ func firstFailed(v delivery.Verdict) string { // pieces are the parts of a line, from the most to the least needed. type pieces struct { - head, commits, counts, shortCounts, times, shortTimes, over, builds, composition, group string + head, commits, counts, shortCounts, times, shortTimes, over, builds, composition, shortComposition, group string } func (p pieces) join(drop int) string { @@ -70,7 +71,10 @@ func (p pieces) join(drop int) string { if drop < 2 && p.group != "" { parts = append(parts, p.group) } - if p.composition != "" { + switch { + case drop >= 3 && p.shortComposition != "": + parts = append(parts, p.shortComposition) + case p.composition != "": parts = append(parts, p.composition) } out := make([]string, 0, len(parts)) @@ -90,14 +94,37 @@ func levelWords(name string, r delivery.LevelResult) string { return s } -func compositionWords(c *delivery.Composition) string { - switch { - case c == nil, c.State == delivery.CompositionNotRun: +// compositionWords says the composition: not yet run, or once the assigner's check reports it (issue 1297) passed or +// failed, with the time it took and the machines it composed (issue 1300). +func compositionWords(c *delivery.Composition, took int64) string { return composition(c, took, false) } + +// composition is compositionWords; short counts the machines and leaves the SDK commit to the report. +func composition(c *delivery.Composition, took int64, short bool) string { + if c == nil || c.State == delivery.CompositionNotRun { return "composition not yet run" - case c.State == delivery.CompositionPass: - return "composition passed with " + short12(c.SDK) } - return "composition failed: " + c.Words + in := "" + if took > 0 { + in = " in " + seconds(took) + } + on := "" + if len(c.Machines) > 0 { + machines := slices.Sorted(slices.Values(c.Machines)) + on = " on " + strings.Join(machines, ", ") + switch { + case len(machines) > 1 && short: + on = fmt.Sprintf(" on %d machines", len(machines)) + case len(machines) > 1: + on = fmt.Sprintf(" on %d machines: %s", len(machines), strings.Join(machines, ", ")) + } + } + switch { + case c.State == delivery.CompositionPass && short: + return "composition passed" + in + on // the SDK it composed with is in the report + case c.State == delivery.CompositionPass: + return "composition passed" + in + " with " + short12(c.SDK) + on + } + return "composition failed" + in + on + ": " + c.Words } func buildWords(builds []BuildNote) string { @@ -149,7 +176,7 @@ func parts(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote, slowest []s for _, x := range []struct { name string ms int64 - }{{"queue", t.Queue}, {"clone", t.Clone}, {"unit", t.Unit}, {"int setup", t.IntegrationSetup}, {"int run", t.IntegrationRun}, {"composition", t.Composition}} { + }{{"queue", t.Queue}, {"clone", t.Clone}, {"unit", t.Unit}, {"int setup", t.IntegrationSetup}, {"int run", t.IntegrationRun}} { if x.ms > 0 { times = append(times, x.name+" "+seconds(x.ms)) } @@ -181,7 +208,8 @@ func parts(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote, slowest []s } pc.builds = buildWords(builds) if v.State != delivery.VerdictRunning { - pc.composition = compositionWords(v.Composition) + pc.composition = compositionWords(v.Composition, v.Took.Composition) + pc.shortComposition = composition(v.Composition, v.Took.Composition, true) } if cs != nil && len(cs.Pulls) > 1 { pc.group = fmt.Sprintf("group %s of %d pull requests", cs.Group, len(cs.Pulls)) diff --git a/modules/mesh-delivery-checker/internal/check/line_test.go b/modules/mesh-delivery-checker/internal/check/line_test.go index 6ea2499..662db1d 100644 --- a/modules/mesh-delivery-checker/internal/check/line_test.go +++ b/modules/mesh-delivery-checker/internal/check/line_test.go @@ -133,3 +133,24 @@ func TestThePassedCompositionIsSaidOnlyWhenItWasRun(t *testing.T) { // ADR 0484 t.Errorf("a verdict with no composition: %q", r) } } + +func TestTheCompositionsTimeAndMachinesAreSaidOnceTheAssignerReportsThem(t *testing.T) { // 1300 (C10), the assigner's verdict (issue 1297) + cs, p := lineCase(t, pullOf(repoA, 7, head1)) + v := passVerdict() + v.Took.Composition = 2400 + v.Composition = &delivery.Composition{State: delivery.CompositionPass, SDK: "afbe371003a570f088be4a394d83aa7a9458f7ab", Machines: []string{"g14", "ace", "novox", "shanks"}} + r := Report(v, cs, p, nil, nil) + for _, want := range []string{"composition passed in 2.4s with afbe371003a5 on 4 machines: ace, g14, novox, shanks"} { + if !strings.Contains(r, want) { + t.Errorf("report %q lacks %q", r, want) + } + } + if l := Line(v, cs, p, nil, nil); !strings.Contains(l, "composition passed in 2.4s") || !strings.Contains(l, "on 4 machines") || utf8.RuneCountInString(l) > MaxLine { + t.Errorf("line = %q", l) + } + v.State, v.Failed = delivery.VerdictFail, delivery.FailedFirst{Machine: "novox", Words: "composition refused"} + v.Composition = &delivery.Composition{State: delivery.CompositionFail, Words: "novox: two seats claim dns", Machines: []string{"novox"}} + if r := Report(v, cs, p, nil, nil); !strings.Contains(r, "composition failed in 2.4s on novox: novox: two seats claim dns") { + t.Errorf("failed composition report = %q", r) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/pins.go b/modules/mesh-delivery-checker/internal/check/pins.go index 326d606..f1e2937 100644 --- a/modules/mesh-delivery-checker/internal/check/pins.go +++ b/modules/mesh-delivery-checker/internal/check/pins.go @@ -77,7 +77,7 @@ func (st *state) checkPins(ctx context.Context) verdictPtr { text, err = st.j.Forge.File(ctx, t.pull.Repository, file, t.pull.Trunk) } if err != nil { - return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: %s cannot be read to check its mesh-sdk pin: %v", t.pull.Repository, t.pull.Number, file, err)) + return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: %s cannot be read to check its mesh-sdk pin: %v%s", t.pull.Repository, t.pull.Number, file, err, unheardWords(err))) } pin, ok := SDKPin(text) if !ok { @@ -88,7 +88,7 @@ func (st *state) checkPins(ctx context.Context) verdictPtr { } onTrunk, err := st.j.Forge.OnTrunk(ctx, SDKRepository, pin) if err != nil { - return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: whether mesh-sdk %s is on its trunk cannot be told: %v", t.pull.Repository, t.pull.Number, pin, err)) + return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: whether mesh-sdk %s is on its trunk cannot be told: %v%s", t.pull.Repository, t.pull.Number, pin, err, unheardWords(err))) } if onTrunk { continue @@ -115,6 +115,14 @@ func pinHint(c *Case, fetchError string) string { return " (a pin to the head of the group's mesh-sdk pull request names a commit only that head carries, which go mod download cannot resolve; " + Issue1148 + ")" } +// unheardWords ends the words of a failed read of the forge that reached nobody: the forge was not heard (UnheardWords). +func unheardWords(err error) string { + if isUnheard(err) { + return " (the forge was " + UnheardWords + ")" + } + return "" +} + // forgeMachine is the machine a failed read of the forge is put at: the forge when nobody answered it (no build node is to // blame, issue 1820), the build node whose workspace was read otherwise. func forgeMachine(err error, node string) string { diff --git a/modules/mesh-delivery-checker/internal/check/pins_test.go b/modules/mesh-delivery-checker/internal/check/pins_test.go index 9198b17..703a11b 100644 --- a/modules/mesh-delivery-checker/internal/check/pins_test.go +++ b/modules/mesh-delivery-checker/internal/check/pins_test.go @@ -253,3 +253,15 @@ func TestThePinOfTheControllerThatCannotBeReadFailsTheCheck(t *testing.T) { // r t.Errorf("verdict = %+v", v) } } + +func TestAForgeThatIsNotHeardFailsAtTheForgeAndIsJudgedAgainWhenTheNodesChange(t *testing.T) { // 1624 C7 + b := newBench(t) + b.ex.Forge = &fakeForge{err: &Unheard{Err: fmt.Errorf("nats: no responders available for request")}} + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict + if v.State != delivery.VerdictFail || v.Failed.Machine != "forge" { + t.Fatalf("verdict = %+v, want failed at the forge", v) + } + if !IsNodesFailure(v) { + t.Errorf("a forge that was not heard is not the change's failure: %+v is not judged again when the nodes change", v.Failed) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/read.go b/modules/mesh-delivery-checker/internal/check/read.go index 874e99b..e4eebd3 100644 --- a/modules/mesh-delivery-checker/internal/check/read.go +++ b/modules/mesh-delivery-checker/internal/check/read.go @@ -7,6 +7,7 @@ import ( "strings" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" ) // The read tools (novox/hq ADR 0484 clauses 8 and 13, issue 1686): what the checker stored, as stored, with the line of the @@ -17,6 +18,9 @@ const ( VerdictsReadTool = "checker_verdicts" PullReadTool = "checker_pull" KeptReadTool = "checker_kept" + // PreviewsReadTool reads the previews state (issue 1680); StatusReadTool reads what the checker is held by and what it runs (issue 1685). + PreviewsReadTool = "checker_previews" + StatusReadTool = "checker_status" ) // StateRead reads one state: its keys and a key's stored value (nil for a key with none). @@ -26,7 +30,14 @@ type StateRead struct { } // Reader answers the read tools from the checker's verdicts and kept states and the indexer's candidates. -type Reader struct{ VerdictState, KeptState, CandidateState StateRead } +type Reader struct { + VerdictState, KeptState, CandidateState StateRead + + PreviewState StateRead // the checker's merge previews + PlansState StateRead // the controller's delivery plans in their build phase; zero where the bucket is absent + StepStates map[string]StateRead // the build modules' steps states, by module + Capacity func() (string, error) // the capacity setting as written, `node=steps,...` +} // VerdictRead is a verdict as stored with the full report of the status of each pull request it judged (Report), when the // candidate still holds those heads; without it the report is that of the verdict alone. @@ -57,7 +68,7 @@ func (r Reader) read(key string, raw json.RawMessage) (VerdictRead, error) { return out, fmt.Errorf("%s holds what is not a verdict: %w", key, err) } c, ok := r.candidate(v.Group) - if !ok || c.HeadsHash != v.Heads { + if !ok || (c.HeadsHash != v.Heads && GateHeads(c) != v.Heads) { out.Note = "the candidate no longer holds these heads" return out, nil } @@ -191,3 +202,153 @@ func (r Reader) Kept(level, digest string) (map[string]json.RawMessage, error) { } return out, nil } + +// PreviewRead is one preview as stored, with the parts of its key. +type PreviewRead struct { + Key string `json:"key"` + Repository string `json:"repository"` + Head string `json:"head"` + Trunk string `json:"trunk"` + Preview json.RawMessage `json:"preview"` +} + +// Previews are the previews the checker holds, each as stored. With a repository only that repository's; with a head (a +// prefix of at least 7 characters is enough) only previews of that head. A key that is no preview key is listed as unreadable. +func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, error) { + if r.PreviewState.Keys == nil || r.PreviewState.Get == nil { + return nil, nil, fmt.Errorf("the previews state cannot be read here") + } + keys, err := r.PreviewState.Keys() + if err != nil { + return nil, nil, err + } + sort.Strings(keys) + out := []PreviewRead{} + var bad []string + for _, k := range keys { + repo, h, trunk, err := delivery.ParsePreviewKey(k) + if err != nil { + bad = append(bad, k) + continue + } + if (repository != "" && repo != repository) || (head != "" && !strings.HasPrefix(h, head)) { + continue + } + raw, err := r.PreviewState.Get(k) + if err != nil { + return nil, nil, err + } + if len(raw) > 0 { + out = append(out, PreviewRead{Key: k, Repository: repo, Head: h, Trunk: trunk, Preview: raw}) + } + } + return out, bad, nil +} + +// RunningStep is a step in flight on a build module. +type RunningStep struct { + Module string `json:"module"` + Node string `json:"node"` + ID string `json:"id"` + Tool string `json:"tool"` + Since string `json:"since,omitempty"` +} + +// Status is what holds the checker back and what runs now, read from the buckets: the checker's held and running steps live in +// its process, so they are shown as what the buckets say (the steps in flight on the build modules, the candidates without a +// verdict for their heads, and the delivery plans in their build phase), not as the process's own counters. +type Status struct { + Capacity map[string]int `json:"capacity"` + CapacityNote string `json:"capacity_note,omitempty"` + DefaultPerNode int `json:"default_per_node"` + PlansBuilding struct { + Readable bool `json:"readable"` + Keys []string `json:"keys"` + Holds bool `json:"holds"` + } `json:"plans_building"` + Running []RunningStep `json:"running_steps"` + RunningByNode map[string]int `json:"running_by_node"` + Unjudged []string `json:"unjudged_candidates"` + Notes []string `json:"notes,omitempty"` +} + +// Status reads the capacity, the plan bucket, the build modules' steps and the candidates against their verdicts. +func (r Reader) Status() (Status, error) { + st := Status{DefaultPerNode: buildrun.DefaultCapacity, Capacity: map[string]int{}, RunningByNode: map[string]int{}} + if r.Capacity == nil { + st.CapacityNote = "the capacity setting cannot be read from here" + } else if setting, err := r.Capacity(); err != nil { + st.CapacityNote = "the capacity setting cannot be read: " + err.Error() + } else if st.Capacity, err = buildrun.ParseCapacity(setting); err != nil { + st.Capacity, st.CapacityNote = map[string]int{}, "the capacity setting is not valid: "+err.Error() + } + if r.PlansState.Keys == nil { + st.Notes = append(st.Notes, PlansBuildingBucket+" is not readable here (absent), so it is read as empty: no delivery holds the checker") + } else if keys, err := r.PlansState.Keys(); err != nil { + st.Notes = append(st.Notes, PlansBuildingBucket+" cannot be listed, so it is read as empty: "+err.Error()) + } else { + sort.Strings(keys) + st.PlansBuilding.Readable, st.PlansBuilding.Keys, st.PlansBuilding.Holds = true, keys, len(keys) > 0 + } + mods := make([]string, 0, len(r.StepStates)) + for m := range r.StepStates { + mods = append(mods, m) + } + sort.Strings(mods) + st.Running = []RunningStep{} + for _, m := range mods { + sr := r.StepStates[m] + if sr.Keys == nil || sr.Get == nil { + continue + } + keys, err := sr.Keys() + if err != nil { + st.Notes = append(st.Notes, "the steps of "+m+" cannot be listed: "+err.Error()) + continue + } + sort.Strings(keys) + for _, k := range keys { + raw, err := sr.Get(k) + if err != nil || len(raw) == 0 { + continue + } + step, err := delivery.DecodeStep(raw) + if err != nil || step.State != delivery.StepRunning { + continue + } + rs := RunningStep{Module: m, Node: step.Node, ID: step.ID, Tool: step.Tool} + if !step.Started.IsZero() { + rs.Since = step.Started.UTC().Format("2006-01-02T15:04:05Z") + } + st.Running = append(st.Running, rs) + st.RunningByNode[step.Node]++ + } + } + st.Unjudged = []string{} + if r.CandidateState.Keys != nil && r.CandidateState.Get != nil && r.VerdictState.Get != nil { + keys, err := r.CandidateState.Keys() + if err != nil { + return st, err + } + sort.Strings(keys) + for _, k := range keys { + raw, err := r.CandidateState.Get(k) + if err != nil || len(raw) == 0 { + continue + } + c, err := delivery.DecodeCandidate(raw) + if err != nil { + continue + } + vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash) + if err != nil { + continue + } + if v, err := r.VerdictState.Get(vkey); err == nil && len(v) == 0 { + st.Unjudged = append(st.Unjudged, c.Group) + } + } + } + st.Notes = append(st.Notes, "the steps the checker holds waiting for a place are in its process and in no bucket: a candidate without a verdict while a plan key exists, or while its node runs as many steps as its capacity, is held") + return st, nil +} diff --git a/modules/mesh-delivery-checker/internal/check/read_test.go b/modules/mesh-delivery-checker/internal/check/read_test.go index 1d1c8fb..de12542 100644 --- a/modules/mesh-delivery-checker/internal/check/read_test.go +++ b/modules/mesh-delivery-checker/internal/check/read_test.go @@ -3,8 +3,10 @@ package check import ( "context" "encoding/json" + "os" "strings" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -102,3 +104,64 @@ func TestTheKeptToolAnswersTheKeysAsStoredByLevelAndByDigest(t *testing.T) { // t.Error("a digest nothing kept was answered") } } + +// 1680 live acceptance: checker_previews shows the preview key of an open pull request, as stored, narrowed by repository and head. +func TestThePreviewsToolShowsTheStoredPreviewOfAPullRequest(t *testing.T) { + pv := delivery.Preview{Repository: repoA, Head: head1, Trunk: head2, Merge: "m1", Modules: map[string]delivery.PreviewModule{}} + data, err := delivery.EncodePreview(pv) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.PreviewKey(repoA, head1, head2) + rd := Reader{PreviewState: fakeState{key: data, "junk": []byte("{}")}.read()} + got, bad, err := rd.Previews(repoA, head1[:8]) + if err != nil || len(got) != 1 || got[0].Key != key || got[0].Head != head1 || got[0].Trunk != head2 || string(got[0].Preview) != string(data) { + t.Fatalf("previews = %+v, %v", got, err) + } + if len(bad) != 1 || bad[0] != "junk" { + t.Errorf("unreadable = %v", bad) + } + if got, _, _ := rd.Previews("novox/other", ""); len(got) != 0 { + t.Errorf("another repository's = %+v", got) + } + if _, _, err := (Reader{}).Previews("", ""); err == nil { + t.Error("answered with no state to read") + } +} + +// 1685 live acceptance: checker_status reads the capacity, whether a plan key holds the checker, the steps in flight by node and +// the candidates without a verdict. +func TestTheStatusToolReadsCapacityHoldsAndRunningSteps(t *testing.T) { + running, _ := delivery.EncodeStep(delivery.Step{ID: "abc1", Tool: "build_go_test", Node: "shanks", State: delivery.StepRunning, Started: time.Date(2026, 10, 22, 15, 0, 0, 0, time.UTC)}) + passed, _ := delivery.EncodeStep(delivery.Step{ID: "abc2", Tool: "build_go_test", Node: "ace", State: delivery.StepPassed}) + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + ckey, _ := delivery.CandidatesKey("feat/x") + rd := Reader{ + Capacity: func() (string, error) { return "shanks=2,ace=0", nil }, + PlansState: fakeState{"p1": []byte("{}")}.read(), + StepStates: map[string]StateRead{"mesh-build-go": fakeState{"shanks.abc1": running, "ace.abc2": passed}.read()}, + CandidateState: fakeState{ckey: cand}.read(), + VerdictState: fakeState{}.read(), + } + st, err := rd.Status() + if err != nil { + t.Fatal(err) + } + if st.Capacity["shanks"] != 2 || st.Capacity["ace"] != 0 || st.DefaultPerNode != 1 { + t.Errorf("capacity = %v default %d", st.Capacity, st.DefaultPerNode) + } + if !st.PlansBuilding.Holds || len(st.PlansBuilding.Keys) != 1 { + t.Errorf("plans = %+v", st.PlansBuilding) + } + if len(st.Running) != 1 || st.Running[0].Node != "shanks" || st.RunningByNode["shanks"] != 1 || st.RunningByNode["ace"] != 0 { + t.Errorf("running = %+v %v", st.Running, st.RunningByNode) + } + if len(st.Unjudged) != 1 || st.Unjudged[0] != "feat/x" { + t.Errorf("unjudged = %v", st.Unjudged) + } + // An absent plan bucket is read as empty and said; an unreadable capacity is said, not guessed. + st, err = Reader{Capacity: func() (string, error) { return "", os.ErrNotExist }}.Status() + if err != nil || st.PlansBuilding.Holds || st.CapacityNote == "" || len(st.Notes) == 0 { + t.Errorf("empty status = %+v, %v", st, err) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/results.go b/modules/mesh-delivery-checker/internal/check/results.go index 5583592..355185d 100644 --- a/modules/mesh-delivery-checker/internal/check/results.go +++ b/modules/mesh-delivery-checker/internal/check/results.go @@ -89,6 +89,9 @@ type Tested struct { // Digested is the answer of the Go build module's digest step: one digest per package that has tests at the level, by import // path. type Digested struct { + // Beside are the lines ` ` of the siblings the digest folded in (DigestVersion 3, lead decision L5), + // sorted; absent when the workspace has none or the build module folds none in. + Beside []string `json:"beside,omitempty"` Level string `json:"level"` Packages map[string]string `json:"packages"` WithoutTests []string `json:"without_tests,omitempty"` 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 c3d4592..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_pull", "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 87c1005..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": [ @@ -31,7 +32,11 @@ ], "tools": [ "checker_kept", + "checker_previews", "checker_pull", + "checker_status", + "checker_try", + "checker_try_read", "checker_verdicts" ], "invokes": [ diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go index 2a7b64b..92982c4 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go @@ -229,7 +229,7 @@ func TestTheCheckersStatusIsSetOncePerRevisionAndAKeyOfTheGateIsNotItsKey(t *tes key, v := checkerOwed(t, 12, "success", "pass aaaaaaaaaaaa on bbbbbbbbbbbb") p.checkOwed(context.Background(), key, 3, v) p.checkOwed(context.Background(), key, 3, v) - if len(f.calls) != 1 || f.calls[0].key != "gitea.gitea_commit_status" || + if len(f.calls) != 1 || f.calls[0].key != "gitea.gitea_check_status" || f.calls[0].body != `{"context":"mesh/check","description":"pass aaaaaaaaaaaa on bbbbbbbbbbbb","owner":"novox","repo":"mesh-delivery","sha":"`+headA+`","state":"success","target_url":"https://view.example/checks/g.0123456789ab"}` { t.Fatalf("calls = %+v", f.calls) } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go index 679a8ae..64d90c1 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go @@ -95,6 +95,10 @@ accounts { "mesh.mod.gitea.call.gitea_commit_status.n1~mesh-delivery-forge-bridge" "mesh.mod.gitea.tool.gitea_commit_status" "mesh.mod.gitea.tool.gitea_commit_status.*" + "mesh.mod.gitea.call.gitea_check_status.*.n1~mesh-delivery-forge-bridge" + "mesh.mod.gitea.call.gitea_check_status.n1~mesh-delivery-forge-bridge" + "mesh.mod.gitea.tool.gitea_check_status" + "mesh.mod.gitea.tool.gitea_check_status.*" "mesh.mod.gitea.call.gitea_delivery_view.*.n1~mesh-delivery-forge-bridge" "mesh.mod.gitea.call.gitea_delivery_view.n1~mesh-delivery-forge-bridge" "mesh.mod.gitea.tool.gitea_delivery_view" @@ -735,7 +739,7 @@ func sha256Sum(b []byte) []byte { s := sha256.Sum256(b); return s[:] } func TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) { // 1683/AC5, ADR 0484 §2 b := startBed(t, `{"result":{"pulls":[]}}`, nil) calls := make(chan string, 8) - sub, err := b.ctl.Subscribe("mesh.mod.gitea.tool.gitea_commit_status", func(m *nats.Msg) { + sub, err := b.ctl.Subscribe("mesh.mod.gitea.tool.gitea_check_status", func(m *nats.Msg) { calls <- string(m.Data) _ = m.Respond([]byte(`{"result":{},"node":"n0"}`)) }) diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/acts.go b/modules/mesh-delivery-forge-bridge/internal/bridge/acts.go index f844717..6073dc3 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/acts.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/acts.go @@ -31,6 +31,9 @@ type Act struct { // number, body), merge.ts gitea_merge_pull_request (owner, repo, number, head, approved_heads). const ( commitStatusTool = "gitea.gitea_commit_status" + // checkStatusTool is the one tool that writes mesh/check (R4e, issue 1301's option A): gitea_commit_status refuses that + // context, and only this bridge invokes it. Its arguments: owner, repo, sha, state, description, target_url, context. + checkStatusTool = "gitea.gitea_check_status" deliveryViewTool = "gitea.gitea_delivery_view" noteAppendTool = "gitea.gitea_note_append" addCommentTool = "gitea.gitea_add_comment" @@ -77,7 +80,11 @@ func OwedAct(key string, value []byte) (Act, error) { if o.Context != context { return Act{}, fmt.Errorf("the owed status %s says the context %q, not the %q its key names", key, o.Context, context) } - a.Call = Call{Tool: commitStatusTool, Args: map[string]any{"owner": owner, "repo": repo, "sha": head, "context": o.Context, + tool := commitStatusTool + if o.Context == delivery.StatusContextCheck { + tool = checkStatusTool + } + a.Call = Call{Tool: tool, Args: map[string]any{"owner": owner, "repo": repo, "sha": head, "context": o.Context, "state": o.State, "description": o.Text, "target_url": o.Target}} case delivery.OwedView: if err := needsPull(); err != nil { diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go index 5b555b8..d3a1d90 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/acts_test.go @@ -43,6 +43,26 @@ func TestOwedStatusIsACommitStatusCall(t *testing.T) { // 1341/AC1 } } +func TestOwedMeshCheckGoesThroughTheCheckStatusToolAlone(t *testing.T) { // R4e 1301-S4 option A: only gitea_check_status writes mesh/check + key, err := delivery.CheckerOwedKey("novox/mesh-delivery", 12, headA) + if err != nil { + t.Fatal(err) + } + v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"}) + if err != nil { + t.Fatal(err) + } + a, err := OwedAct(key, v) + if err != nil { + t.Fatal(err) + } + want := Call{Tool: "gitea.gitea_check_status", Args: map[string]any{"owner": "novox", "repo": "mesh-delivery", "sha": headA, + "context": "mesh/check", "state": "success", "description": "pass", "target_url": "https://view.example/checks/g.1"}} + if !reflect.DeepEqual(a.Call, want) { + t.Errorf("call = %+v, want %+v", a.Call, want) + } +} + func TestOwedViewNoteAndCommentCalls(t *testing.T) { // 1341/AC2 cases := []struct { act string diff --git a/modules/mesh-delivery-forge-bridge/manifest_test.go b/modules/mesh-delivery-forge-bridge/manifest_test.go index 40fb8a8..2653931 100644 --- a/modules/mesh-delivery-forge-bridge/manifest_test.go +++ b/modules/mesh-delivery-forge-bridge/manifest_test.go @@ -92,7 +92,7 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1 if want := []string{"mesh-delivery-checker.owed", "mesh-delivery-gate.merges", "mesh-delivery-gate.owed"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } - if want := []string{"gitea.gitea_add_comment", "gitea.gitea_commit_status", "gitea.gitea_commit_trees", "gitea.gitea_delivery_view", + if want := []string{"gitea.gitea_add_comment", "gitea.gitea_check_status", "gitea.gitea_commit_status", "gitea.gitea_commit_trees", "gitea.gitea_delivery_view", "gitea.gitea_get_file", "gitea.gitea_merge_pull_request", "gitea.gitea_note_append", "gitea.gitea_open_pulls"}; !reflect.DeepEqual(sorted(m.Invokes), want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } diff --git a/modules/mesh-delivery-forge-bridge/module.json b/modules/mesh-delivery-forge-bridge/module.json index d411e5f..895944a 100644 --- a/modules/mesh-delivery-forge-bridge/module.json +++ b/modules/mesh-delivery-forge-bridge/module.json @@ -23,6 +23,7 @@ "invokes": [ "gitea.gitea_merge_pull_request", "gitea.gitea_commit_status", + "gitea.gitea_check_status", "gitea.gitea_delivery_view", "gitea.gitea_note_append", "gitea.gitea_add_comment", diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index b6079de..645c726 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -35,18 +35,39 @@ func Tools(r view.Reader) []stdio.Tool { mods, err := r.Modules(only) return map[string]any{"modules": mods}, err }}, + {Name: view.CandidateReadTool, + Description: "What the indexer stored for one candidate (the open pull requests of a branch) under candidates., as stored: " + + "the pull requests and heads, the base, and per pull request the touched modules with their new-form fingerprint (fp2) or why " + + "they have none (none2). Read-only.", + Input: map[string]any{"type": "object", "required": []string{"group"}, "properties": map[string]any{ + "group": map[string]any{"type": "string", "description": "the candidate's group, the branch name its pull requests share"}}}, + Run: func(args map[string]any) (any, error) { + group, _ := args["group"].(string) + return r.Candidate(group) + }}, + {Name: view.CandidatesReadTool, + Description: "Every candidate the indexer holds, by group, each as stored under candidates.; with module, only those with a " + + "pull request touching that module. Keys that cannot be read are listed as unreadable. Read-only.", + Input: map[string]any{"type": "object", "properties": map[string]any{ + "module": map[string]any{"type": "string", "description": "only candidates touching this module"}}}, + Run: func(args map[string]any) (any, error) { + module, _ := args["module"].(string) + c, bad, err := r.Candidates(module) + return map[string]any{"candidates": c, "unreadable": bad}, err + }}, {Name: view.BuilderBuildsReadTool, Description: "The builder's build keys, each as stored in mesh-delivery-builder_builds: key, module, fingerprint (src2) and build " + "(state, by, step with the nodes it was built on, manifest, artifacts with their digests, toolchains, error). Narrow by module " + - "or state (queued, building, done, failed). Read-only.", + "or state (queued, building, done, failed); at most limit builds (default 50), newest revision first for a module, and the answer says when it was cut and lists keys that are no build key (unreadable). Without a module, give one to get its newest. Read-only.", Input: map[string]any{"type": "object", "properties": map[string]any{ "module": map[string]any{"type": "string", "description": "only this module's builds"}, - "state": map[string]any{"type": "string", "description": "only builds in this state"}}}, + "state": map[string]any{"type": "string", "description": "only builds in this state"}, + "limit": map[string]any{"type": "integer", "description": "at most this many builds (default 50), newest revision first"}}}, Run: func(args map[string]any) (any, error) { module, _ := args["module"].(string) state, _ := args["state"].(string) - rows, err := r.BuilderBuilds(module, state) - return map[string]any{"builds": rows}, err + limit, _ := args["limit"].(float64) + return r.BuilderBuilds(module, state, int(limit)) }}, {Name: view.BuilderBuildReadTool, Description: "One build of the builder as stored. Give the module and its fingerprint (fp2, src2:...), or the module alone for the build " + @@ -65,7 +86,13 @@ func Tools(r view.Reader) []stdio.Tool { } func read(state stdio.KeptState) view.StateRead { - return view.StateRead{Keys: state.Keys, Get: func(key string) (json.RawMessage, error) { + return view.StateRead{Keys: state.Keys, GetEntry: func(key string) (json.RawMessage, uint64, error) { + e, err := state.Get(key) + if err != nil || e == nil { + return nil, 0, err + } + return e.Value, e.Revision, nil + }, Get: func(key string) (json.RawMessage, error) { e, err := state.Get(key) if err != nil || e == nil { return nil, err @@ -75,7 +102,7 @@ func read(state stdio.KeptState) view.StateRead { } func main() { - r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds"))} + r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds")), CandidatesState: read(stdio.State("mesh-delivery-indexer.candidates"))} if err := stdio.Serve("", Tools(r)); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-view-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go index a5e2193..0c13bdf 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "fmt" "strings" "testing" @@ -117,13 +118,16 @@ func TestBuilderBuildsListsKeysAsStored(t *testing.T) { if err != nil { t.Fatal(err) } - rows := out.(map[string]any)["builds"].([]view.BuilderRow) + rows := out.(view.BuildsAnswer).Builds + if u := out.(view.BuildsAnswer).Unreadable; len(u) != 1 || u[0] != "junk" { + t.Errorf("unreadable = %v, want [junk] (issue 1818)", u) + } if len(rows) != 2 || rows[0].Key != keyA || rows[0].Module != "netcheck" || rows[0].Fingerprint != fp2A || string(rows[0].Build) != doneBuild || string(rows[1].Build) != failBuild { t.Errorf("rows = %+v", rows) } for args, want := range map[string]string{"state": keyB, "module": keyA} { out, _ := run(t, r, view.BuilderBuildsReadTool, map[string]any{args: map[string]string{keyB: "failed", keyA: "netcheck"}[want]}) - got := out.(map[string]any)["builds"].([]view.BuilderRow) + got := out.(view.BuildsAnswer).Builds if len(got) != 1 || got[0].Key != want { t.Errorf("%s filter = %+v, want %s", args, got, want) } @@ -156,3 +160,107 @@ func TestBuilderBuildJoinsTheIndexersFingerprint(t *testing.T) { t.Errorf("an unbuilt fingerprint: %v", err) } } + +// bigBuilds is a bucket of n keys over 100 modules that counts the values read. +type bigBuilds struct { + fake fake + reads *int +} + +func newBig(n int) bigBuilds { + f := fake{} + for i := 0; i < n; i++ { + f[fmt.Sprintf("mod%02d.%064x", i%100, i)] = doneBuild + } + return bigBuilds{fake: f, reads: new(int)} +} + +func (b bigBuilds) reader() view.Reader { + r := b.fake.reader() + get := r.State.Get + r.Builds = view.StateRead{Keys: r.State.Keys, Get: func(k string) (json.RawMessage, error) { *b.reads++; return get(k) }, + GetEntry: func(k string) (json.RawMessage, uint64, error) { + *b.reads++ + raw, err := get(k) + var rev uint64 + fmt.Sscanf(k[strings.Index(k, ".")+1:], "%x", &rev) // the fake's revision is the key's number + return raw, rev + 1, err + }} + return r +} + +// 1827/AC1: with 2,000 keys, a module's builds are found by key prefix, so only that module's values are read (20, not 2,000), +// newest revision first, within the limit, and the answer says it was cut. +func TestBuilderBuildsFiltersByPrefixBeforeAnyGet(t *testing.T) { + b := newBig(2000) + out, err := run(t, b.reader(), view.BuilderBuildsReadTool, map[string]any{"module": "mod07", "limit": float64(5)}) + if err != nil { + t.Fatal(err) + } + a := out.(view.BuildsAnswer) + if *b.reads != 20 { + t.Errorf("read %d values for a module of 20 keys, want 20", *b.reads) + } + if len(a.Builds) != 5 || !a.Cut || !strings.Contains(a.Note, "5") { + t.Errorf("answer = %d rows, cut %v, note %q", len(a.Builds), a.Cut, a.Note) + } + for i := 1; i < len(a.Builds); i++ { + if a.Builds[i-1].Revision < a.Builds[i].Revision { + t.Errorf("not newest first: %d before %d", a.Builds[i-1].Revision, a.Builds[i].Revision) + } + } + if a.Builds[0].Revision != 1908 { + t.Errorf("newest revision = %d, want 1908", a.Builds[0].Revision) + } +} + +// 1827/AC2: with no module the answer reads at most the limit's worth in key order (default 50), never the 2,000. +func TestBuilderBuildsWithoutAModuleIsBounded(t *testing.T) { + b := newBig(2000) + out, err := run(t, b.reader(), view.BuilderBuildsReadTool, map[string]any{}) + if err != nil { + t.Fatal(err) + } + a := out.(view.BuildsAnswer) + if len(a.Builds) != view.DefaultBuildsLimit || !a.Cut || *b.reads > view.DefaultBuildsLimit { + t.Errorf("%d rows, cut %v, %d reads", len(a.Builds), a.Cut, *b.reads) + } + b = newBig(2000) + out, _ = run(t, b.reader(), view.BuilderBuildsReadTool, map[string]any{"state": "failed"}) + if a := out.(view.BuildsAnswer); *b.reads != view.MaxBuildReads || !a.Cut || len(a.Builds) != 0 { + t.Errorf("a state matching nothing: %d reads (want the bound %d), cut %v", *b.reads, view.MaxBuildReads, a.Cut) + } +} + +// 1818: fp2 together with commit is refused, naming both. +func TestBuilderBuildRefusesFP2WithCommit(t *testing.T) { + r := both(fake{keyA: doneBuild}, fake{"netcheck": `{"v":3,"fp2":"` + fp2A + `","commit":"c1","path":"."}`}) + _, err := run(t, r, view.BuilderBuildReadTool, map[string]any{"module": "netcheck", "fp2": fp2A, "commit": "c1"}) + if err == nil || !strings.Contains(err.Error(), "fp2") || !strings.Contains(err.Error(), "commit") || !strings.Contains(err.Error(), "not both") { + t.Errorf("err = %v", err) + } +} + +// 1681: indexer_candidates and indexer_candidate answer the stored value; a pull request's fp2 per touched module shows. +func TestCandidateToolsAnswerTheStoredValue(t *testing.T) { + cand := `{"v":1,"group":"feat/x","base":3,"heads":"h","pulls":[{"modules":["zsh"],"fps":{"zsh":{"fp2":"` + fp2B + `","manifest":"m"}}}]}` + other := `{"v":1,"group":"fix","base":3,"heads":"h","pulls":[{"modules":["netcheck"]}]}` + r := fake{"feat/x": cand, "fix": other, "bad": "{"}.reader() + r.CandidatesState = r.State + out, err := run(t, r, view.CandidateReadTool, map[string]any{"group": "feat/x"}) + if err != nil || string(out.(json.RawMessage)) != cand || !strings.Contains(string(out.(json.RawMessage)), fp2B) { + t.Fatalf("candidate = %v, %v", out, err) + } + if _, err := run(t, r, view.CandidateReadTool, map[string]any{"group": "gone"}); err == nil || !strings.Contains(err.Error(), "no candidate gone") { + t.Errorf("err = %v", err) + } + out, err = run(t, r, view.CandidatesReadTool, map[string]any{}) + m := out.(map[string]any) + if err != nil || len(m["candidates"].(map[string]json.RawMessage)) != 2 || len(m["unreadable"].([]string)) != 1 { + t.Errorf("all = %v, %v", m, err) + } + out, _ = run(t, r, view.CandidatesReadTool, map[string]any{"module": "zsh"}) + if c := out.(map[string]any)["candidates"].(map[string]json.RawMessage); len(c) != 1 || c["feat/x"] == nil { + t.Errorf("by module = %v", c) + } +} diff --git a/modules/mesh-delivery-view/internal/view/read.go b/modules/mesh-delivery-view/internal/view/read.go index e99a327..8f1546a 100644 --- a/modules/mesh-delivery-view/internal/view/read.go +++ b/modules/mesh-delivery-view/internal/view/read.go @@ -7,6 +7,7 @@ package view import ( "encoding/json" "fmt" + "slices" "sort" "strings" @@ -23,12 +24,16 @@ const ( type StateRead struct { Keys func() ([]string, error) Get func(key string) (json.RawMessage, error) + // GetEntry, when set, also gives the revision of the value; builder_builds orders by it. + GetEntry func(key string) (json.RawMessage, uint64, error) } // Reader answers the read tools from a modules state. type Reader struct { State StateRead // the indexer's modules Builds StateRead // the builder's builds + + CandidatesState StateRead // the indexer's candidates } // Module is the stored value of modules., exactly as the indexer wrote it. @@ -84,6 +89,67 @@ func (r Reader) Modules(none2 bool) (map[string]json.RawMessage, error) { return out, nil } +// The read tools of the indexer's candidates (issue 1681): what the indexer stored under candidates., as stored, so the +// new-form fingerprint of each touched module of an open pull request (pulls[].fps) can be read live. +const ( + CandidateReadTool = "indexer_candidate" + CandidatesReadTool = "indexer_candidates" +) + +// Candidate is the stored value of the candidates key of a group, exactly as the indexer wrote it. +func (r Reader) Candidate(group string) (json.RawMessage, error) { + key, err := delivery.CandidatesKey(strings.TrimSpace(group)) + if err != nil { + return nil, err + } + if r.CandidatesState.Get == nil { + return nil, fmt.Errorf("the candidates state cannot be read here") + } + raw, err := r.CandidatesState.Get(key) + if err != nil { + return nil, err + } + if len(raw) == 0 { + return nil, fmt.Errorf("no candidate %s in the candidates state", group) + } + return raw, nil +} + +// Candidates is every candidate, by group, each as stored; with a module only the candidates with a pull request touching it. +// A value that is not readable is listed under unreadable with its key, not skipped. +func (r Reader) Candidates(module string) (map[string]json.RawMessage, []string, error) { + if r.CandidatesState.Keys == nil || r.CandidatesState.Get == nil { + return nil, nil, fmt.Errorf("the candidates state cannot be read here") + } + keys, err := r.CandidatesState.Keys() + if err != nil { + return nil, nil, err + } + sort.Strings(keys) + out := map[string]json.RawMessage{} + var unreadable []string + for _, k := range keys { + raw, err := r.CandidatesState.Get(k) + if err != nil { + return nil, nil, err + } + if len(raw) == 0 { + continue + } + group, perr := delivery.ParseCandidatesKey(k) + var c delivery.Candidate + if perr != nil || json.Unmarshal(raw, &c) != nil { + unreadable = append(unreadable, k) + continue + } + if module != "" && !slices.ContainsFunc(c.Pulls, func(p delivery.CandidatePull) bool { return slices.Contains(p.Modules, module) }) { + continue + } + out[group] = raw + } + return out, unreadable, nil +} + // The read tools of the builder's builds (novox/hq issue 1723): what the builder stored under `.` // in mesh-delivery-builder_builds, answered as stored, so its states, nodes (in the step) and digests can be checked live. const ( @@ -91,48 +157,100 @@ const ( BuilderBuildsReadTool = "builder_builds" ) -// BuilderRow is one key of the builder's builds: the key, the module and the fingerprint it parses to, and the stored value -// as the builder wrote it (state, by, step, manifest, artifacts with digests, toolchains, error). +// BuilderRow is one key of the builder's builds: the key, the module and the fingerprint it parses to, the revision of its +// stored value (when the read gives one) and the stored value as the builder wrote it. type BuilderRow struct { Key string `json:"key"` Module string `json:"module"` Fingerprint string `json:"fingerprint"` + Revision uint64 `json:"revision,omitempty"` Build json.RawMessage `json:"build"` } +// The bounds of builder_builds (issue 1827): the rows it returns when no limit is given, and the values it reads in one call. +// Reading a value is a bus round trip, so the answer must not read the whole bucket (it timed out at 30 s on the live one). +const ( + DefaultBuildsLimit = 50 + MaxBuildReads = 200 +) + +// BuildsAnswer is the answer of builder_builds: the rows, newest revision first, whether the answer was cut and why, and +// the keys that are no build key (Unreadable), which are listed rather than silently skipped (issue 1818). +type BuildsAnswer struct { + Builds []BuilderRow `json:"builds"` + Cut bool `json:"cut,omitempty"` + Note string `json:"note,omitempty"` + Unreadable []string `json:"unreadable,omitempty"` +} + +func (r Reader) get(key string) (json.RawMessage, uint64, error) { + if r.Builds.GetEntry != nil { + return r.Builds.GetEntry(key) + } + raw, err := r.Builds.Get(key) + return raw, 0, err +} + func (r Reader) builderRow(key string) (BuilderRow, bool, error) { module, fp, err := delivery.ParseBuilderBuildKey(key) if err != nil { return BuilderRow{}, false, nil // a key that is no build key is not a build } - raw, err := r.Builds.Get(key) + raw, rev, err := r.get(key) if err != nil { return BuilderRow{}, false, err } if len(raw) == 0 { return BuilderRow{}, false, nil } - return BuilderRow{Key: key, Module: module, Fingerprint: fp, Build: raw}, true, nil + return BuilderRow{Key: key, Module: module, Fingerprint: fp, Revision: rev, Build: raw}, true, nil } -// BuilderBuilds lists the builder's keys, each as stored, sorted by key; module and state, when given, keep only the builds -// of that module or in that state. -func (r Reader) BuilderBuilds(module, state string) ([]BuilderRow, error) { +// BuilderBuilds lists the builder's keys, each as stored, newest revision first. With a module it considers only the keys +// that begin `.`, before any value is read; with a state it keeps the builds in that state. It returns at most limit +// rows (DefaultBuildsLimit when limit is not above zero) and reads at most MaxBuildReads values, and says when it cut the +// answer. Keys that are no build key are listed as unreadable. Without a module the keys are walked in key order, so the +// answer is the first rows found, not the newest of the bucket: give a module to get its newest. +func (r Reader) BuilderBuilds(module, state string, limit int) (BuildsAnswer, error) { if r.Builds.Keys == nil || r.Builds.Get == nil { - return nil, fmt.Errorf("the builder's builds cannot be read here") + return BuildsAnswer{}, fmt.Errorf("the builder's builds cannot be read here") + } + if limit <= 0 { + limit = DefaultBuildsLimit } keys, err := r.Builds.Keys() if err != nil { - return nil, err + return BuildsAnswer{}, err } sort.Strings(keys) - rows := []BuilderRow{} + prefix := "" + if module = strings.TrimSpace(module); module != "" { + prefix = module + "." + } + out := BuildsAnswer{Builds: []BuilderRow{}} + var candidates []string for _, k := range keys { + if prefix != "" && !strings.HasPrefix(k, prefix) { + continue + } + if _, _, err := delivery.ParseBuilderBuildKey(k); err != nil { + out.Unreadable = append(out.Unreadable, k) + continue + } + candidates = append(candidates, k) + } + reads := 0 + for _, k := range candidates { + if reads >= MaxBuildReads { + out.Cut, out.Note = true, fmt.Sprintf("read %d values and stopped; give a module (and a state) to narrow the answer", MaxBuildReads) + break + } + reads++ row, ok, err := r.builderRow(k) if err != nil { - return nil, err + return BuildsAnswer{}, err } - if !ok || (module != "" && row.Module != module) { + if !ok { continue } if state != "" { @@ -143,9 +261,26 @@ func (r Reader) BuilderBuilds(module, state string) ([]BuilderRow, error) { continue } } - rows = append(rows, row) + out.Builds = append(out.Builds, row) + // Without a module the walk is in key order and stops at the limit; with one, all its (few) keys are read to order them. + if prefix == "" && len(out.Builds) >= limit { + if reads < len(candidates) { + out.Cut, out.Note = true, fmt.Sprintf("stopped at the limit of %d builds in key order, not newest first; give a module for its newest", limit) + } + break + } } - return rows, nil + sort.SliceStable(out.Builds, func(i, j int) bool { + if out.Builds[i].Revision != out.Builds[j].Revision { + return out.Builds[i].Revision > out.Builds[j].Revision + } + return out.Builds[i].Key < out.Builds[j].Key + }) + if len(out.Builds) > limit { + out.Builds = out.Builds[:limit] + out.Cut, out.Note = true, fmt.Sprintf("cut to the newest %d builds of %s", limit, module) + } + return out, nil } // BuilderBuild is the build of module at the new-form fingerprint fp2; with an empty fp2 it is the build at the fingerprint @@ -156,6 +291,9 @@ func (r Reader) BuilderBuild(module, fp2, commit string) (BuilderRow, error) { if r.Builds.Get == nil { return BuilderRow{}, fmt.Errorf("the builder's builds cannot be read here") } + if fp2 != "" && commit != "" { + return BuilderRow{}, fmt.Errorf("give fp2 or commit, not both: fp2 %s names a build directly, commit %s asks for the build at the indexer's fingerprint of that commit", fp2, commit) + } if fp2 == "" { raw, err := r.Module(module) if err != nil { diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index 0e185a2..b3c0e1c 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -95,7 +95,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 } // The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer // itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads. - if want := []string{"indexer_module", "indexer_modules", "builder_build", "builder_builds"}; !reflect.DeepEqual(m.Tools, want) { + if want := []string{"indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } for _, a := range m.Build.Artifacts { diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 89bf62c..09cd55b 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -6,6 +6,8 @@ "tools": [ "indexer_module", "indexer_modules", + "indexer_candidate", + "indexer_candidates", "builder_build", "builder_builds" ],