diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index bf067462..59c6b4c9 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -58,6 +58,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + // The toolchain of each language as heard agreed, for the steps' asks (§416, A-104). + toolchains, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, "toolchains"), "name a step's toolchain") + if err != nil { + return err + } // The registrations and each watched branch's catalog half, first views then events (§21, §195, §200): until a // branch's view is held the runner acts on nothing new of it, and says so. calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) @@ -93,7 +98,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // for its step. starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say, Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module, - Arm: armBound(ctx, bounds)} + Arm: armBound(ctx, bounds), Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains)} resume := func(branch string) { half, ok := rv.Catalog.View(branch) if !ok { @@ -110,6 +115,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hearStepEnded(hub, hearer, logger.Printf); err != nil { return err } + if err := hearToolchain(ctx, hub, toolchains, logger.Printf); err != nil { + return err + } if err := hearAskTaken(ctx, hub, bounds, logger.Printf); err != nil { return err } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go index 0e026400..8e6c7ae4 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -9,6 +9,7 @@ import ( "github.com/nats-io/nats.go/jetstream" + builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -199,3 +200,71 @@ func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hea }() return nil } + +// keptReader reads the kept results of the packages named (§25, §416): the entries of the state kept whose package is a +// name or lies beneath it, in one pass over the bucket. +func keptReader(ctx context.Context, kept jetstream.KeyValue) func([]string) (map[string][]nodebuildgo.KeptResult, error) { + return func(pkgs []string) (map[string][]nodebuildgo.KeptResult, error) { + w, err := kept.WatchAll(ctx, jetstream.IgnoreDeletes()) + if err != nil { + return nil, err + } + defer w.Stop() + out := map[string][]nodebuildgo.KeptResult{} + for { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case e, ok := <-w.Updates(): + if !ok { + return out, nil + } + if e == nil { // the end of what the bucket holds now + return out, nil + } + var k nodebuildgo.KeptResult + if err := json.Unmarshal(e.Value(), &k); err != nil { + continue + } + for _, name := range pkgs { + if pipeline.InPackage(k.Package, name) { + out[name] = append(out[name], k) + } + } + } + } + } +} + +// toolchainReader reads the toolchain of a language as the runner heard it agreed, from its state toolchains (A-104); nil +// when none was heard. +func toolchainReader(ctx context.Context, toolchains jetstream.KeyValue) func(string) (*nodebuildgo.ToolchainUsed, error) { + return func(language string) (*nodebuildgo.ToolchainUsed, error) { + e, err := toolchains.Get(ctx, keys.EscapeName(language)) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return pipeline.ToolchainOf(e.Value()) + } +} + +// hearToolchain keeps each toolchain-agreed heard as `` in the state toolchains (A-104): the newest of a language. +func hearToolchain(ctx context.Context, hub *process.EventHub, toolchains jetstream.KeyValue, logf func(string, ...any)) error { + return hub.Handle(builds.SubjectToolchainAgreed, func(m jetstream.Msg) { + t, err := pipeline.ToolchainOf(m.Data()) + if err != nil || t == nil { + logf("%s is no toolchain-agreed: %v", m.Subject(), err) + _ = m.Term() + return + } + if _, err := toolchains.Put(ctx, keys.EscapeName(t.Language), m.Data()); err != nil { + logf("the toolchain of %s is not kept, heard again: %v", t.Language, err) + _ = m.Nak() + return + } + _ = m.Ack() + }) +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/place.go b/modules/mesh-delivery-pipeline/internal/pipeline/place.go index 885a44d4..6530c35e 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/place.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/place.go @@ -96,7 +96,11 @@ func Place(p seat.Pipeline, reg catalog.Registration, entries map[string]catalog s.Seat, s.Outcome, s.Started = bs, seat.StepResultOutcomeRunning, &now s.Ask = AskID(p.Pipeline, a.Ref, step) steps = append(steps, s) - asks = append(asks, Asked{Seat: bs, ID: s.Ask, Ask: stepAsk(p, entry, a, step, s.Ask, asker, now)}) + ask := Asked{Seat: bs, ID: s.Ask, Ask: stepAsk(p, entry, a, step, s.Ask, asker, now)} + if a.Package != nil { + ask.Package = a.Package.Name + } + asks = append(asks, ask) } } for _, step := range moduleSteps { @@ -126,6 +130,9 @@ type Asked struct { Seat string ID string Ask any + // Package is the name of the artifact's package, from the catalog entry, for a step on a build seat; its kept results + // are the ask's (§25, §416). + Package string } func stepAsk(p seat.Pipeline, e catalog.CatalogEntry, a catalog.Artifact, step, id, asker string, now time.Time) nodebuildgo.StepAsk { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/place_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/place_test.go index 92ee45de..6ae3bc11 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/place_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/place_test.go @@ -214,3 +214,55 @@ func TestResumeAsksTheStepsPlacedAndNotEnded(t *testing.T) { } } } + +// A step ask carries the kept results of its artifact's package and the toolchain of its language (design 47 G §25, §416, +// A-104): the results whose package is the artifact's package or lies beneath it, read once for all the asks; a failed +// read asks the steps with none. +func TestAStepAskCarriesTheKeptResultsAndTheToolchain(t *testing.T) { + reg, entries := placeFixture() + e := entries["app"] + e.Artifacts[0].Package = &catalog.ArtifactPackage{Ecosystem: "go", Name: "git.example/o/app"} + entries["app"] = e + store, asks := created{}, &stepAsks{} + reads := 0 + s := &Starter{Pipelines: store, Say: func(string, any) error { return nil }, Steps: asks, Name: "the-runner", + Registration: func(string) (catalog.Registration, bool) { return reg, true }, + Kept: func(pkgs []string) (map[string][]nodebuildgo.KeptResult, error) { + reads++ + if len(pkgs) != 1 || pkgs[0] != "git.example/o/app" { + t.Errorf("kept read for %v, want the one package of the entry", pkgs) + } + return map[string][]nodebuildgo.KeptResult{pkgs[0]: {{Digest: "sha256:aa", Level: "unit", Package: "git.example/o/app/sub"}}}, nil + }, + Toolchain: func(l string) (*nodebuildgo.ToolchainUsed, error) { + return &nodebuildgo.ToolchainUsed{Language: l, Version: "1.26"}, nil + }} + if _, err := s.Start(Begin{Branch: "main", Source: 4, Module: "app", Commit: "c1", Fp: "src2:aa", Repository: "o/app", Entries: entries}); err != nil { + t.Fatal(err) + } + var program, tools *nodebuildgo.StepRun + for _, a := range asks.asks { + if ask, ok := a.(nodebuildgo.StepAsk); ok { + switch { + case ask.Step.Artifact == "app:program" && ask.Step.Step == "lint": + program = ask.Step + case ask.Step.Artifact == "app:tools" && ask.Step.Step == "unit": + tools = ask.Step + } + } + } + if reads != 1 || program == nil || len(program.Kept) != 1 || program.Kept[0].Digest != "sha256:aa" || program.Toolchain == nil || program.Toolchain.Version != "1.26" { + t.Errorf("the program's ask = %+v after %d reads", program, reads) + } + if tools == nil || len(tools.Kept) != 0 || tools.Toolchain == nil { + t.Errorf("an artifact with no package carries no kept result and its language's toolchain: %+v", tools) + } + for pkg, want := range map[string]bool{"a/b": true, "a/b/c": true, "a/bc": false, "a": false} { + if InPackage(pkg, "a/b") != want { + t.Errorf("InPackage(%s, a/b) = %v, want %v", pkg, !want, want) + } + } + if tc, err := ToolchainOf([]byte(`{"language":"go","version":"1.26.1","digests":[]}`)); err != nil || tc.Language != "go" || tc.Version != "1.26.1" { + t.Errorf("ToolchainOf = %+v, %v", tc, err) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index e5366bb9..b31097da 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -11,6 +11,7 @@ import ( builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" "git.novox.be/novox/mesh-sdk/go/keys" ) @@ -60,6 +61,13 @@ type Starter struct { // asker. Steps StepAsker Name string + // Kept reads the kept results of each named package, by the package name an artifact's catalog entry carries (§25, §416): the + // results whose package is that name or one beneath it. A step ask carries them, so a build seat runs only the packages + // with no result for their digest. + Kept func(packages []string) (map[string][]nodebuildgo.KeptResult, error) + // Toolchain is the toolchain of a language as the runner heard it agreed (§416, A-104): its language and version; nil when + // none was heard, which asks the step with none. + Toolchain func(language string) (*nodebuildgo.ToolchainUsed, error) // Arm arms the bound of an ask put on a build seat's work queue (§535, §647); nil arms none. Arm Arm } @@ -106,6 +114,7 @@ func (s *Starter) Start(b Begin) (bool, error) { s.logf("pipeline-started of %s is not said: %v", key, err) } if s.Steps != nil && len(asks) > 0 { + s.carry(key, asks) if err := askAll(s.Steps, s.Arm, key, asks); err != nil { return true, err } @@ -163,6 +172,7 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) { if len(due) == 0 { continue } + s.carry(key, due) if err := askAll(s.Steps, s.Arm, key, due); err != nil && first == nil { first = err } @@ -171,6 +181,48 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) { return asked, first } +// carry puts the kept results of each ask's package and the toolchain of its language on its step ask (§25, §416, A-104). A +// read that fails asks the steps with none, so every package runs: slower, never wrong; it is said. +func (s *Starter) carry(key string, asks []Asked) { + var pkgs []string + seen := map[string]bool{} + for _, a := range asks { + if a.Package != "" && !seen[a.Package] { + seen[a.Package] = true + pkgs = append(pkgs, a.Package) + } + } + var kept map[string][]nodebuildgo.KeptResult + if s.Kept != nil && len(pkgs) > 0 { + var err error + if kept, err = s.Kept(pkgs); err != nil { + s.logf("the kept results for the steps of %s are not read, so each package runs: %v", key, err) + } + } + chains := map[string]*nodebuildgo.ToolchainUsed{} + for _, a := range asks { + ask, ok := a.Ask.(nodebuildgo.StepAsk) + if !ok || ask.Step == nil { + continue + } + if a.Package != "" { + ask.Step.Kept = kept[a.Package] + } + if s.Toolchain == nil || ask.Language == "" { + continue + } + t, done := chains[ask.Language] + if !done { + var err error + if t, err = s.Toolchain(ask.Language); err != nil { + s.logf("the toolchain of %s for the steps of %s is not read: %v", ask.Language, key, err) + } + chains[ask.Language] = t + } + ask.Step.Toolchain = t + } +} + func (s *Starter) logf(format string, args ...any) { if s.Logf != nil { s.Logf(format, args...) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/toolchain.go b/modules/mesh-delivery-pipeline/internal/pipeline/toolchain.go new file mode 100644 index 00000000..dc4607fe --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/toolchain.go @@ -0,0 +1,28 @@ +package pipeline + +import ( + "encoding/json" + "strings" + + builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" + nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" +) + +// ToolchainOf reads a toolchain-agreed as the toolchain a step ask names: its language and version (design 47 G §416, A-104). +// The recipe and the building node's digest are the seat's to report, not the runner's to know. +func ToolchainOf(raw []byte) (*nodebuildgo.ToolchainUsed, error) { + var t builds.ToolchainAgreed + if err := json.Unmarshal(raw, &t); err != nil { + return nil, err + } + if t.Language == "" { + return nil, nil + } + return &nodebuildgo.ToolchainUsed{Language: t.Language, Version: t.Version}, nil +} + +// InPackage says whether a kept result's package is the artifact's package name or lies beneath it (A-104: the kept entries +// whose package is the artifact's package, a Go module's packages being the module path and the paths beneath it). +func InPackage(pkg, name string) bool { + return name != "" && (pkg == name || strings.HasPrefix(pkg, name+"/")) +} diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index 813ecbd6..36894974 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -50,7 +50,7 @@ func TestManifest(t *testing.T) { {"claims the seat serving every verb of its definition", len(m.Claims) == 1 && m.Claims[0].Name == seat.Seat && equalSets(m.Claims[0].Serves, seat.Verbs())}, // §27, §85, §330, §355, §77: the old state sections. - {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "findings", "first-views"})}, + {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views"})}, // §535, §647, A-77, A-104: an ask no holder takes is bound by a key with its own time to live. {"bounds is key-ttl", keyTTL["bounds"]}, // §27, §85: the step log on an old stream, its outputs in an old objects section. diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index a9e6e8bd..044ee89a 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -71,6 +71,10 @@ "name": "aged", "history": 1 }, + { + "name": "toolchains", + "history": 1 + }, { "name": "bounds", "history": 1,