diff --git a/modules/mesh-build-go/buildstep/askrun.go b/modules/mesh-build-go/buildstep/askrun.go index 6400ec15d..050a0b046 100644 --- a/modules/mesh-build-go/buildstep/askrun.go +++ b/modules/mesh-build-go/buildstep/askrun.go @@ -47,6 +47,13 @@ type AskSteps struct { // Has says whether this node has a need a package's tests declare (sdkstep.AskNeeds, issue 890); a package needing // what it lacks is not tested here but named in the level's Unmet, for the asker to place. Nil has nothing. Has func(need string) bool + // Lint is the module's lint body for a pipeline step lint (novox/hq ADR 0666 §7): it takes the workspace and the + // module's directory and answers a LintAnswer. Nil ends every lint step as nothing to run. + Lint StepFunc + // Node is the node this holder runs on, which a step's end names; SayStep says a pipeline step's end as the seat's + // event step-ended on that node's subject (sdkstep.StepEndedSubject), under the ask's id. Nil runs no step ask. + Node string + SayStep func(end sdkstep.StepEnded) error } // RunAsk runs ask's job with steps and answers a sdkstep.BuildAnswer or a sdkstep.CheckAnswer. @@ -56,6 +63,9 @@ func RunAsk(ctx context.Context, steps AskSteps, ask sdkstep.StepAsk) (any, erro job.Fetch = withPackages(job.Fetch, ask.Packages) return steps.build(ctx, job) } + if ask.Step != nil { + return steps.runStep(ctx, ask) + } if ask.Check != nil { job := *ask.Check job.Fetch = *withPackages(&job.Fetch, ask.Packages) diff --git a/modules/mesh-build-go/buildstep/askrun_test.go b/modules/mesh-build-go/buildstep/askrun_test.go index d9a93e195..693afbb83 100644 --- a/modules/mesh-build-go/buildstep/askrun_test.go +++ b/modules/mesh-build-go/buildstep/askrun_test.go @@ -369,3 +369,103 @@ func TestASeatsAskTypeRunsAsTheNeutralAsk(t *testing.T) { t.Errorf("the ask as another seat's type: %+v, %v", back, err) } } + +// 086 A-C2 (ADR 0666 §7, §8, §21; design 47 *Asked on a work queue, ended by an event*, test 65): a pipeline step ask +// runs its one level as a check of it, carries the packages the runner kept at their digest, keeps what it passed, says +// its counts, and says its end once on its own node; a level with no tests ends nothing to run, a fault of the source +// could not run, and a transport fault or an end that could not be said is handed back unsaid, to be given again. +func TestRunAskRunsAPipelineStepAndSaysItsEnd(t *testing.T) { + type said struct{ ends []sdkstep.StepEnded } + stepOf := func(step string, kept ...sdkstep.KeptResult) sdkstep.StepAsk { + return sdkstep.StepAsk{ID: sdkstep.NewID(), Kind: sdkstep.AskStep, Language: sdkstep.AskGo, Asker: "mesh-delivery-pipeline", At: time.Now(), + Step: &sdkstep.StepRun{Pipeline: "main.12.m", Step: step, Module: "m", Artifact: "m:program", Commit: runCommit, Fingerprint: "fp-1", + Fetch: sdkstep.Fetch{Repository: "novox/x", Commit: runCommit, Dir: "modules/m"}, Kept: kept, + Names: sdkstep.StepNames{Modules: []string{"m"}, Repositories: []string{"novox/x"}, Branches: []string{"main"}}}} + } + for _, row := range []struct { + name string + ask sdkstep.StepAsk + digests map[string]string + tests map[string]bool + lint StepFunc + fetchErr error + sayErr error + outcome string // "" when nothing is said + counts int + carried []string + kept []string + passing bool // the error is a transport fault + noTests bool // the level holds no test, as the test body says it + }{ + {name: "unit: the kept package carried, the other tested and kept", ask: stepOf(sdkstep.StepUnit, sdkstep.KeptResult{Package: "x/a", Digest: "d-a", Level: "unit"}), + digests: map[string]string{"x/a": "d-a", "x/b": "d-b"}, tests: map[string]bool{"x/b": true}, outcome: sdkstep.OutcomePassed, counts: 1, carried: []string{"x/a"}, kept: []string{"x/b"}}, + {name: "unit failed: said with why, nothing kept", ask: stepOf(sdkstep.StepUnit), digests: map[string]string{"x/b": "d-b"}, tests: map[string]bool{"x/b": false}, + outcome: sdkstep.OutcomeFailed, counts: 1}, + {name: "integration with no tests: nothing to run", ask: stepOf(sdkstep.StepIntegration), digests: map[string]string{}, noTests: true, outcome: sdkstep.OutcomeNothingToRun}, + {name: "lint failed", ask: stepOf(sdkstep.StepLint), lint: func(context.Context, map[string]any) (any, error) { + return LintAnswer{Failed: []sdkstep.Failure{{Package: "modules/m", Test: "gofmt", Cause: "not as gofmt writes them"}}}, nil + }, outcome: sdkstep.OutcomeFailed}, + {name: "lint passed", ask: stepOf(sdkstep.StepLint), lint: func(context.Context, map[string]any) (any, error) { return LintAnswer{}, nil }, outcome: sdkstep.OutcomePassed}, + {name: "no lint body: nothing to run", ask: stepOf(sdkstep.StepLint), outcome: sdkstep.OutcomeNothingToRun}, + {name: "the source cannot be fetched: could not run", ask: stepOf(sdkstep.StepUnit), fetchErr: Refuse("no commit %s", runCommit), outcome: sdkstep.OutcomeCouldNotRun}, + {name: "a transport fault: unsaid, given again", ask: stepOf(sdkstep.StepUnit), fetchErr: &sdkstep.Passing{Err: errors.New("no responders")}, passing: true}, + {name: "the end cannot be said: given again", ask: stepOf(sdkstep.StepLint), sayErr: errors.New("bus down"), passing: true}, + } { + r := &recorder{fetch: map[string]any{"workspace": "/w"}, tests: row.tests, digests: row.digests} + s := r.steps() + if row.fetchErr != nil { + s.Fetch = func(context.Context, map[string]any) (any, error) { return nil, row.fetchErr } + } + s.Lint, s.Node = row.lint, "ace" + if row.noTests { + s.Test = func(context.Context, map[string]any) (any, error) { + return map[string]any{"passed": true, "tests": 0, "no_tests": true}, nil + } + } + var got said + s.SayStep = func(e sdkstep.StepEnded) error { + if row.sayErr != nil { + return row.sayErr + } + got.ends = append(got.ends, e) + return nil + } + out, err := RunAsk(context.Background(), s, row.ask) + if row.passing { + if !sdkstep.IsPassing(err) || len(got.ends) != 0 { + t.Errorf("%s: %v, said %+v; want a transport fault, nothing said", row.name, err, got.ends) + } + continue + } + if err != nil || len(got.ends) != 1 { + t.Fatalf("%s: %v, said %+v; want one end", row.name, err, got.ends) + } + e := got.ends[0] + if e.Outcome != row.outcome || e.Node != "ace" || e.ID != row.ask.ID || e.Fingerprint != "fp-1" || out.(StepAnswer).Ended.Outcome != row.outcome { + t.Errorf("%s: said %+v, want %s on ace for the ask", row.name, e, row.outcome) + } + if (row.outcome == sdkstep.OutcomeFailed || row.outcome == sdkstep.OutcomeCouldNotRun) && e.Said == "" { + t.Errorf("%s: %s and not why", row.name, row.outcome) + } + if n := 0; e.Tests != nil { + n = e.Tests.Unit + e.Tests.Integration + if n != row.counts { + t.Errorf("%s: counts %+v, want %d", row.name, e.Tests, row.counts) + } + } else if row.counts != 0 { + t.Errorf("%s: no counts, want %d", row.name, row.counts) + } + pkgs := func(ks []sdkstep.KeptResult) (out []string) { + for _, k := range ks { + out = append(out, k.Package) + } + return out + } + if !slices.Equal(pkgs(e.Carried), row.carried) || !slices.Equal(pkgs(e.Kept), row.kept) { + t.Errorf("%s: carried %v kept %v, want %v and %v", row.name, pkgs(e.Carried), pkgs(e.Kept), row.carried, row.kept) + } + if _, err := sdkstep.EncodeStepEnded(e); err != nil { + t.Errorf("%s: the end does not encode: %v", row.name, err) + } + } +} diff --git a/modules/mesh-build-go/buildstep/steprun.go b/modules/mesh-build-go/buildstep/steprun.go new file mode 100644 index 000000000..3549c170b --- /dev/null +++ b/modules/mesh-build-go/buildstep/steprun.go @@ -0,0 +1,252 @@ +package buildstep + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + + sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" +) + +// **A pipeline step, run and said** (novox/hq ADR 0666 §7, §8, §17, §21; design 47 *Pipeline steps and the policy*, +// *Asked on a work queue, ended by an event*, *Carried, not run*). The runner asks one lint or test step of an artifact +// as the accepted kind step; the holder fetches the change's heads, runs the step and says its end as the seat's event +// step-ended on its own node's subject before the ask is acknowledged, so a holder restarted mid-step gets the ask again +// from its work queue and finishes it (ADR 0664 §1), its end said under the ask's id. A test step runs only the packages +// with no kept result for their digest at its level, says the others carried, and keeps what it passed; a step with +// nothing to do ends `nothing to run`, never passed; one that cannot run ends `could not run`, saying why. + +// StepAnswer is what a step ask answers on its step key: the end it said. +type StepAnswer struct { + Ended sdkstep.StepEnded `json:"ended"` +} + +// LintAnswer is what a module's lint body answers: the failures of its language's lint, none when it passed, and +// Nothing when the tree holds nothing its lint reads. +type LintAnswer struct { + Failed []sdkstep.Failure `json:"failed,omitempty"` + Nothing bool `json:"nothing,omitempty"` +} + +// runStep runs ask's pipeline step and says its end. A transport fault, a decline or the process stopping is handed back +// unsaid, for the bus to give the ask again, here or on another node. +func (s AskSteps) runStep(ctx context.Context, ask sdkstep.StepAsk) (any, error) { + if s.SayStep == nil { + return nil, Refuse("this build module says no step's end") + } + end, err := sdkstep.EndOf(ask, s.Node, sdkstep.OutcomePassed) + if err != nil { + return nil, Refuse("%v", err) + } + if err := s.stepOutcome(ctx, ask, &end); err != nil { + return nil, err + } + if err := s.SayStep(end); err != nil { + return nil, &sdkstep.Passing{Err: fmt.Errorf("the end of the step %s could not be said: %w", ask.ID, err)} + } + return StepAnswer{Ended: end}, nil +} + +// stepOutcome runs the step into end; the error it answers is one that keeps the end unsaid. +func (s AskSteps) stepOutcome(ctx context.Context, ask sdkstep.StepAsk, end *sdkstep.StepEnded) error { + var err error + if ask.Step.Step == sdkstep.StepLint { + err = s.lintStep(ctx, *ask.Step, end) + } else { + err = s.testStep(ctx, ask, end) + } + var dec *Declining + switch { + case err == nil: + return nil + case sdkstep.IsPassing(err), errors.As(err, &dec), ctx.Err() != nil: + return err + default: + end.Outcome, end.Said = sdkstep.OutcomeCouldNotRun, err.Error() + end.Tests, end.Carried, end.Kept = nil, nil, nil + return nil + } +} + +// lintStep fetches the heads and runs the module's lint body over the artifact's tree. +func (s AskSteps) lintStep(ctx context.Context, run sdkstep.StepRun, end *sdkstep.StepEnded) error { + if s.Lint == nil { + end.Outcome = sdkstep.OutcomeNothingToRun + return nil + } + if _, err := s.onToolchain(ctx, run.Toolchain); err != nil { + return err + } + fetch := run.Fetch + _, f, err := s.fetch(ctx, &fetch) + if err != nil { + return err + } + defer s.release(f.Workspace) + if len(f.Conflicts) > 0 { + return fmt.Errorf("the heads do not merge: %s", strings.Join(f.Conflicts, ", ")) + } + Phase(ctx, "lint") + out, err := s.Lint(ctx, map[string]any{"workspace": f.Workspace, "dir": fetch.Dir}) + if err != nil { + return err + } + r, err := raw(out) + if err != nil { + return err + } + var la LintAnswer + if err := json.Unmarshal(r, &la); err != nil { + return fmt.Errorf("the lint answered what is not a lint answer: %s: %w", r, err) + } + switch { + case len(la.Failed) > 0: + end.Outcome, end.Said = sdkstep.OutcomeFailed, failedSaid(la.Failed) + case la.Nothing: + end.Outcome = sdkstep.OutcomeNothingToRun + default: + end.Outcome = sdkstep.OutcomePassed + } + return nil +} + +// testStep runs one test level as a check of that level alone, taking the step's kept results in the checker's place. +func (s AskSteps) testStep(ctx context.Context, ask sdkstep.StepAsk, end *sdkstep.StepEnded) error { + run := *ask.Step + level := run.Step + s.Kept = func(_ context.Context, lvl, digest string) (int, bool, error) { + for _, k := range run.Kept { + if k.Level == lvl && k.Digest == digest { + return 0, true, nil + } + } + return 0, false, nil + } + job := sdkstep.CheckJob{Fetch: *withPackages(&run.Fetch, ask.Packages), Levels: []string{level}, Kept: true, Toolchain: run.Toolchain} + out, err := s.check(ctx, job) + if err != nil { + return err + } + answer, _ := out.(sdkstep.CheckAnswer) + if len(answer.Levels) == 0 { + var conflicts struct { + Conflicts []string `json:"conflicts"` + } + _ = json.Unmarshal(mustRaw(answer.Fetch), &conflicts) + return fmt.Errorf("the heads do not merge: %s", strings.Join(conflicts.Conflicts, ", ")) + } + la := answer.Levels[0] + if len(la.Unmet) > 0 { + var what []string + for _, u := range la.Unmet { + what = append(what, u.Package+" ("+strings.Join(u.Needs, ", ")+")") + } + return fmt.Errorf("the %s tests of %s need what this node lacks", level, strings.Join(what, ", ")) + } + tests, passed, ran := 0, true, false + var why []string + for _, t := range la.Tests { + var tr struct { + Passed bool `json:"passed"` + Tests int `json:"tests"` + NoTests bool `json:"no_tests"` + Why string `json:"why"` + } + if err := json.Unmarshal(t, &tr); err != nil { + return fmt.Errorf("the %s level answered what is no test result: %w", level, err) + } + tests += tr.Tests + ran = ran || (!tr.NoTests && tr.Tests > 0) || !tr.Passed + if !tr.Passed { + passed = false + if tr.Why != "" { + why = append(why, tr.Why) + } + } + } + for _, k := range la.Kept { + end.Carried = append(end.Carried, keptOf(run, k.Package, k.Digest)) + } + switch { + case !passed: + end.Outcome, end.Said = sdkstep.OutcomeFailed, strings.Join(why, "; ") + if end.Said == "" { + end.Said = "the " + level + " tests failed" + } + case !ran && len(end.Carried) == 0: + end.Outcome = sdkstep.OutcomeNothingToRun + default: + end.Outcome = sdkstep.OutcomePassed + if la.Digest != nil { + for _, p := range la.Digest.Packages { + if !carried(end.Carried, p.Package) { + end.Kept = append(end.Kept, keptOf(run, p.Package, p.Digest)) + } + } + } + } + if end.Outcome != sdkstep.OutcomeNothingToRun { + end.Tests = &sdkstep.TestCounts{} + if level == sdkstep.StepUnit { + end.Tests.Unit = tests + } else { + end.Tests.Integration = tests + } + } + return nil +} + +func keptOf(run sdkstep.StepRun, pkg, digest string) sdkstep.KeptResult { + return sdkstep.KeptResult{Package: pkg, Digest: digest, Level: run.Step, Toolchain: run.Toolchain, Pipeline: run.Pipeline} +} + +func carried(kept []sdkstep.KeptResult, pkg string) bool { + for _, k := range kept { + if k.Package == pkg { + return true + } + } + return false +} + +func mustRaw(v any) json.RawMessage { + r, _ := raw(v) + return r +} + +// failedSaid names each failure in one line, the first ten. +func failedSaid(failed []sdkstep.Failure) string { + var names []string + for i, f := range failed { + if i == 10 { + names = append(names, fmt.Sprintf("and %d more", len(failed)-10)) + break + } + names = append(names, strings.TrimSpace(f.Package+" "+f.Test+": "+firstOf(f.Cause))) + } + return fmt.Sprintf("%d failed: %s", len(failed), strings.Join(names, "; ")) +} + +func firstOf(s string) string { + l, _, _ := strings.Cut(strings.TrimSpace(s), "\n") + return l +} + +// SayStepOn is a holder's SayStep: the end, encoded and checked by the domain SDK (sdkstep.EncodeStepEnded), published +// on the language's build seat's step-ended subject of node with publish (stdio.SeatPublish), de-duplicated by the +// ask's id, so a step run again after a restart is said once. +func SayStepOn(language, node string, publish func(subject string, body any, id string) error) func(sdkstep.StepEnded) error { + return func(end sdkstep.StepEnded) error { + subject, err := sdkstep.StepEndedSubject(language, node) + if err != nil { + return err + } + body, err := sdkstep.EncodeStepEnded(end) + if err != nil { + return err + } + return publish(subject, json.RawMessage(body), "step-ended."+end.ID) + } +} diff --git a/modules/mesh-build-go/cmd/mesh-build-go/asks.go b/modules/mesh-build-go/cmd/mesh-build-go/asks.go index a1fcf324a..fa751589a 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/asks.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/asks.go @@ -56,6 +56,8 @@ func (m *Module) askSteps(steps []StepTool) buildstep.AskSteps { Has: m.Has, Release: func(ws string) error { return m.spaces().Release(ws) }, Logf: m.Say, + Node: m.Node, + SayStep: m.sayStep(), } } @@ -151,3 +153,12 @@ func (m *Module) endLeftRunning(ctx context.Context, scan buildrun.Scan, booted func (m *Module) stepScan() buildrun.Scan { return buildrun.ScanOf(stdio.State(seat.StateSteps)) } + +// sayStep says a pipeline step's end on this node's subject of the seat (design 47 *Asked on a work queue, ended by an +// event*, lines 584-586 at b229f5c4): SayStep when a test sets it, else the runtime's seat publish. +func (m *Module) sayStep() func(sdkstep.StepEnded) error { + if m.SayStep != nil { + return m.SayStep + } + return buildstep.SayStepOn(sdkstep.AskGo, m.Node, stdio.SeatPublish) +} diff --git a/modules/mesh-build-go/cmd/mesh-build-go/module.go b/modules/mesh-build-go/cmd/mesh-build-go/module.go index 1c57a3d28..b5c99cecf 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/module.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/module.go @@ -102,6 +102,8 @@ type Module struct { Settings string Capacity func() int ReadKept func(key string) (json.RawMessage, error) + // SayStep says a pipeline step's end (buildstep.AskSteps.SayStep); the seat's publish on this node when nil. + SayStep func(sdkstep.StepEnded) error // ReadManifest asks the delivery indexer's read tool (ManifestsTool) for one manifest by its key, which a compose ask's // request names its manifests by (compose.go); it reads none of the indexer's state (ADR 0596); GoModCache, when set, is the Go module cache in place of Cache's (a test). ReadManifest func(key string) (json.RawMessage, error)