A-C2: a build module runs the pipeline step kind step and says its end on its node (issue 1966)
The runner asks a lint or test step of an artifact as the build seat's accepted kind step, and hears its end as the seat's step-ended on the node that ran it (ADR 0666 sections 7, 8, 21; design 47 *The steps and who serves them* line 571 and *Asked on a work queue, ended by an event* lines 583-590 at hq b229f5c4). The build modules' shared package runs a test step as a check of its one level, carries the packages whose digest the ask names kept, keeps what it passed and says the counts; a level with no tests ends nothing to run, a source fault could not run, and a transport fault or an end that could not be said is handed back unsaid, so the work queue gives the step again and it ends once under its ask's id. The neutral forms are the domain SDK's (sdk/buildstep pipelinestep.go, mesh-delivery dc0df1b5); the Go module says the end through the runtime's seat publish. Milestone 47/G, research 086 area A. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ (Joined from 72fee3c6.) Builds: design 47 G §22, §24, §86, §116, §140 (BLD3, BLD4) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user