From 960c8e104a88d27e9f3702bc14b3c0b2f876fd32 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 23:43:35 +0100 Subject: [PATCH] =?UTF-8?q?=C2=A7416,=20=C2=A725,=20A-104:=20a=20step=20as?= =?UTF-8?q?k=20carries=20the=20kept=20results=20of=20its=20package=20and?= =?UTF-8?q?=20the=20toolchain=20of=20its=20language?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The asking half reads, in one pass for all a pipeline's asks, the kept entries whose package is the artifact's package from the catalog entry or lies beneath it, and the runner's own toolchain per language, kept in the new state toolchains as heard from toolchain-agreed. Builds: design 47 G §25, §416, A-104 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-pipeline/main.go | 10 ++- .../cmd/mesh-delivery-pipeline/steps.go | 69 +++++++++++++++++++ .../internal/pipeline/place.go | 9 ++- .../internal/pipeline/place_test.go | 52 ++++++++++++++ .../internal/pipeline/start.go | 52 ++++++++++++++ .../internal/pipeline/toolchain.go | 28 ++++++++ .../mesh-delivery-pipeline/manifest_test.go | 2 +- modules/mesh-delivery-pipeline/module.json | 4 ++ 8 files changed, 223 insertions(+), 3 deletions(-) create mode 100644 modules/mesh-delivery-pipeline/internal/pipeline/toolchain.go 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,