From 4375ed8b2cf0148e4aa9eb0bc67fedd35b45e27e Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 19:18:11 +0200 Subject: [PATCH] Move the builder's shared build-run parts into go/delivery/buildrun, so the checker shares them and no module keeps a copy (issue 1671) --- go/delivery/buildrun/choose.go | 88 ++++++++++++ go/delivery/buildrun/choose_test.go | 99 ++++++++++++++ go/delivery/buildrun/doc.go | 5 + go/delivery/buildrun/have.go | 56 ++++++++ go/delivery/buildrun/modules.go | 7 + go/delivery/buildrun/places.go | 171 ++++++++++++++++++++++++ go/delivery/buildrun/places_test.go | 126 +++++++++++++++++ go/delivery/buildrun/presence.go | 53 ++++++++ go/delivery/buildrun/presence_test.go | 43 ++++++ go/delivery/buildrun/steps.go | 67 ++++++++++ go/delivery/buildrun/steps_test.go | 116 ++++++++++++++++ go/delivery/buildrun/toolchains.go | 64 +++++++++ go/delivery/buildrun/toolchains_test.go | 51 +++++++ 13 files changed, 946 insertions(+) create mode 100644 go/delivery/buildrun/choose.go create mode 100644 go/delivery/buildrun/choose_test.go create mode 100644 go/delivery/buildrun/doc.go create mode 100644 go/delivery/buildrun/have.go create mode 100644 go/delivery/buildrun/modules.go create mode 100644 go/delivery/buildrun/places.go create mode 100644 go/delivery/buildrun/places_test.go create mode 100644 go/delivery/buildrun/presence.go create mode 100644 go/delivery/buildrun/presence_test.go create mode 100644 go/delivery/buildrun/steps.go create mode 100644 go/delivery/buildrun/steps_test.go create mode 100644 go/delivery/buildrun/toolchains.go create mode 100644 go/delivery/buildrun/toolchains_test.go diff --git a/go/delivery/buildrun/choose.go b/go/delivery/buildrun/choose.go new file mode 100644 index 0000000..06aea83 --- /dev/null +++ b/go/delivery/buildrun/choose.go @@ -0,0 +1,88 @@ +package buildrun + +import ( + "fmt" + "sort" + "strings" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ChooseNodes picks up to want build nodes for a language (ADR 0478 clause 3): the nodes whose toolchains key reports the +// version the nodes agree on, the ones with room under their build capacity first (places), then the fewest lanes the +// part has running or chosen there, a tie broken by the node's name. A place is reserved on each node chosen, in the +// same step as the choice, so builds choosing at once do not choose the same full node while another has room; the +// caller takes the place (Places.Acquire) or gives it up (Places.Cancel). A node at its capacity is still chosen when +// the others are too (fewer than want have room): its lane waits for a place. A node qualifies only when qualifies is true +// of it (online and its build module healthy; nil means every node does). It answers the toolchain all the reporting nodes +// agree on, qualifying or not, as the indexer computes it. When nothing is reported, or the nodes report different +// versions (so the indexer agreed none), it fails naming the language and why; when no reporting node qualifies it fails +// naming the nodes. A node whose build module refuses on its toolchain refuses the step, and the build fails naming it. +func ChooseNodes(language string, entries []delivery.NodeToolchain, places *Places, qualifies func(node string) bool, want int) ([]string, delivery.Toolchain2, error) { + tc, none := delivery.AgreedToolchain(entries) + if none != "" { + return nil, delivery.Toolchain2{}, fmt.Errorf("no %s build node qualifies: %s", language, none) + } + nodes, err := choose(language, entries, places, qualifies, want) + return nodes, tc, err +} + +// ChooseAnyNodes is ChooseNodes for artifacts that need no toolchain, such as an archive (ADR 0478 clause 1): the +// build module's nodes agree on nothing, so any node of the build module that qualifies is chosen, by capacity and load only. +func ChooseAnyNodes(language string, entries []delivery.NodeToolchain, places *Places, qualifies func(node string) bool, want int) ([]string, error) { + return choose(language, entries, places, qualifies, want) +} + +func choose(language string, entries []delivery.NodeToolchain, places *Places, qualifies func(node string) bool, want int) ([]string, error) { + if places == nil { + places = &Places{} + } + nodes := make([]string, 0, len(entries)) + var reporting []string + seen := map[string]bool{} + for _, e := range entries { + if seen[e.Node] { + continue + } + seen[e.Node] = true + reporting = append(reporting, e.Node) + if places.CapacityOf(e.Node) > 0 && (qualifies == nil || qualifies(e.Node)) { // a capacity of 0 keeps the part off a node + nodes = append(nodes, e.Node) + } + } + if len(nodes) == 0 { + sort.Strings(reporting) + return nil, fmt.Errorf("no %s build node qualifies: %s report the toolchain, and none is online with a healthy build module and a build capacity", language, strings.Join(reporting, ", ")) + } + return places.choose(nodes, want), nil +} + +// DigestsDifferPrefix opens the error of a build whose nodes built an artifact to different digests (ADR 0478 clause 4): +// such a key is not built again as a new revision, since the same source on the same toolchains gave two answers and +// building again settles nothing; it waits for another fingerprint. +const DigestsDifferPrefix = "digests differ:" + +// Agree is the one digest every node built an artifact to (ADR 0478 clause 4), or an error naming each node and its +// digest when they differ. One node's digest agrees with itself. +func Agree(artifact string, byNode map[string]string) (string, error) { + nodes := make([]string, 0, len(byNode)) + for n := range byNode { + nodes = append(nodes, n) + } + sort.Strings(nodes) + var first string + for i, n := range nodes { + if i == 0 { + first = byNode[n] + continue + } + if byNode[n] != first { + msg := fmt.Sprintf("%s the nodes built %s to different digests:", DigestsDifferPrefix, artifact) + for _, n := range nodes { + msg += fmt.Sprintf(" %s %s;", n, byNode[n]) + } + return "", fmt.Errorf("%s", msg[:len(msg)-1]) + } + } + return first, nil +} diff --git a/go/delivery/buildrun/choose_test.go b/go/delivery/buildrun/choose_test.go new file mode 100644 index 0000000..6e9c2bb --- /dev/null +++ b/go/delivery/buildrun/choose_test.go @@ -0,0 +1,99 @@ +package buildrun + +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +const digest = "sha256:2222222222222222222222222222222222222222222222222222222222222222" + +func hex64(c byte) string { return strings.Repeat(string(c), 64) } + +func tcs(language, version string, nodes ...string) []delivery.NodeToolchain { + var out []delivery.NodeToolchain + for _, n := range nodes { + out = append(out, delivery.NodeToolchain{Node: n, Toolchain: delivery.Toolchain{Language: language, Version: version, Recipe: "1"}}) + } + return out +} + +func TestChooseTakesTheAgreedNodesWithFewestRunningStepsTiesByName(t *testing.T) { // 1607/AC4, ADR 0478 clause 3 + entries := tcs("go", "go1.27.1", "novox", "ace", "g14", "shanks") + nodes, tc, err := ChooseNodes("go", entries, placesRunning(map[string]int{"novox": 3, "ace": 1, "g14": 1, "shanks": 0}), nil, 2) + if err != nil { + t.Fatal(err) + } + if len(nodes) != 2 || nodes[0] != "shanks" || nodes[1] != "ace" { + t.Errorf("nodes = %v, want [shanks ace]: fewest steps first, a tie by name", nodes) + } + if tc.Version != "go1.27.1" || tc.Recipe != "1" || tc.Language != "go" { + t.Errorf("toolchain = %+v", tc) + } +} + +func TestChooseWithOneNodeAnswersOneNode(t *testing.T) { // 1607/AC4 + nodes, _, err := ChooseNodes("go", tcs("go", "go1.27.1", "novox"), nil, nil, 2) + if err != nil || len(nodes) != 1 || nodes[0] != "novox" { + t.Errorf("nodes = %v, err %v, want [novox]", nodes, err) + } +} + +func TestChooseFailsNamingTheLanguageAndTheVersionsWhenNoNodeQualifies(t *testing.T) { // 1607/AC4 + _, _, err := ChooseNodes("typescript", nil, nil, nil, 2) + if err == nil || !strings.Contains(err.Error(), "typescript") || !strings.Contains(err.Error(), delivery.ToolchainNoReport) { + t.Errorf("no report: error = %v, want it to name typescript and %q", err, delivery.ToolchainNoReport) + } + entries := append(tcs("go", "go1.27.1", "novox"), tcs("go", "go1.26.0", "ace")...) + _, _, err = ChooseNodes("go", entries, nil, nil, 2) + if err == nil || !strings.Contains(err.Error(), "go1.27.1") || !strings.Contains(err.Error(), "go1.26.0") || !strings.Contains(err.Error(), "go") { + t.Errorf("disagreeing nodes: error = %v, want both versions named", err) + } +} + +func TestAgreeNeedsEqualDigestsAndNamesBothNodesWhenTheyDiffer(t *testing.T) { // 1607/AC4, ADR 0478 clause 4 + d, err := Agree("tool", map[string]string{"ace": digest, "novox": digest}) + if err != nil || d != digest { + t.Errorf("Agree = %q, %v, want %q", d, err, digest) + } + _, err = Agree("tool", map[string]string{"ace": digest, "novox": "sha256:" + hex64('9')}) + if err == nil || !strings.Contains(err.Error(), "ace") || !strings.Contains(err.Error(), "novox") || !strings.Contains(err.Error(), digest) || !strings.Contains(err.Error(), hex64('9')) || !strings.Contains(err.Error(), "tool") { + t.Errorf("differing digests: error = %v, want the artifact, both nodes and both digests", err) + } + if d, err := Agree("tool", map[string]string{"ace": digest}); err != nil || d != digest { + t.Errorf("one node: %q, %v", d, err) + } +} + +func TestChooseSkipsNodesThatAreNotOnlineAndFailsNamingThemWhenNoneIs(t *testing.T) { // 1607 review 3, ADR 0478 clause 3 + entries := tcs("go", "go1.27.1", "ace", "g14", "novox") + online := func(n string) bool { return n != "g14" } // a sleeping laptop + nodes, tc, err := ChooseNodes("go", entries, placesRunning(map[string]int{"novox": 5}), online, 2) + if err != nil || len(nodes) != 2 || nodes[0] != "ace" || nodes[1] != "novox" || tc.Version != "go1.27.1" { + t.Errorf("nodes = %v, toolchain %+v, err %v, want [ace novox] with g14 left out although it runs nothing", nodes, tc, err) + } + _, tc, err = ChooseNodes("go", entries, nil, func(string) bool { return false }, 2) + if err == nil || !strings.Contains(err.Error(), "online") || !strings.Contains(err.Error(), "g14") || !strings.Contains(err.Error(), "go") { + t.Errorf("none online: error = %v, want it to name the nodes that report and that none is online", err) + } + if tc.Version != "go1.27.1" { + t.Errorf("none online: the agreed toolchain %+v is still answered", tc) + } +} + +// placesRunning is Places with these steps counted as running, for a test of the choice. +func placesRunning(running map[string]int) *Places { + return &Places{running: running, changed: make(chan struct{})} +} + +func TestChoosePrefersNodesBelowTheirCapacity(t *testing.T) { // round 3 + entries := tcs("go", "go1.27.1", "ace", "novox", "shanks") + // shanks has capacity 3 with 2 running: room; ace has the default 1 with 1 running: none; novox has none running. + p := placesRunning(map[string]int{"shanks": 2, "ace": 1}) + p.Capacity = map[string]int{"shanks": 3} + nodes, _, err := ChooseNodes("go", entries, p, nil, 2) + if err != nil || len(nodes) != 2 || nodes[0] != "novox" || nodes[1] != "shanks" { + t.Errorf("nodes = %v, err %v, want [novox shanks]: the nodes with room, ace at capacity left out", nodes, err) + } +} diff --git a/go/delivery/buildrun/doc.go b/go/delivery/buildrun/doc.go new file mode 100644 index 0000000..b4a943d --- /dev/null +++ b/go/delivery/buildrun/doc.go @@ -0,0 +1,5 @@ +// Package buildrun holds the parts every module that runs steps on the build modules shares: the caller, the steps +// watch and the runner of a long step (ADR 0469), the sets of presence, toolchains and healthy build modules, the +// places a node offers by capacity, and the choice of build nodes (ADR 0478). The builder and the checker of the +// delivery use them; they moved here from the builder's own package so neither keeps a copy (issue 1671, ADR 0484). +package buildrun diff --git a/go/delivery/buildrun/have.go b/go/delivery/buildrun/have.go new file mode 100644 index 0000000..f3c079c --- /dev/null +++ b/go/delivery/buildrun/have.go @@ -0,0 +1,56 @@ +package buildrun + +import ( + "sync" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// HaveSet holds what each node's node-engine says of each module on it (node-engine_have, key `.`): the +// builder reads it to know whether a node's build module is running and healthy, since a build module whose installed +// toolchain is not the one its recipe builds with is unhealthy (mesh-build-go build_go_health) and its steps would +// refuse (ADR 0478 clause 3). A module the node-engine has written no key of is not healthy. Safe for use from the watch +// and from a build at once. +type HaveSet struct { + mu sync.Mutex + healthy map[string]bool // by `.` +} + +// Put takes a change of the have bucket in and answers whether a module's healthy state changed. A node's own key +// (`.node`) and a value that is not a have are left out. +func (s *HaveSet) Put(key string, value []byte, deleted bool) bool { + node, module, isNode, err := delivery.ParseHaveKey(key) + if err != nil || isNode { + return false + } + ok := false + if !deleted { + h, err := delivery.DecodeHave(value) + if err != nil { + return false + } + ok = h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK + } + k := node + "." + module + s.mu.Lock() + defer s.mu.Unlock() + if s.healthy[k] == ok { + return false + } + if s.healthy == nil { + s.healthy = map[string]bool{} + } + if ok { + s.healthy[k] = true + } else { + delete(s.healthy, k) + } + return true +} + +// Healthy is whether the module on the node was last seen running and healthy. +func (s *HaveSet) Healthy(node, module string) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.healthy[node+"."+module] +} diff --git a/go/delivery/buildrun/modules.go b/go/delivery/buildrun/modules.go new file mode 100644 index 0000000..db15aed --- /dev/null +++ b/go/delivery/buildrun/modules.go @@ -0,0 +1,7 @@ +package buildrun + +// The build modules the mesh runs, by the language each builds (ADR 0469). +const ( + GoBuildModule = "mesh-build-go" + NodeBuildModule = "mesh-build-node" +) diff --git a/go/delivery/buildrun/places.go b/go/delivery/buildrun/places.go new file mode 100644 index 0000000..4b189ba --- /dev/null +++ b/go/delivery/buildrun/places.go @@ -0,0 +1,171 @@ +package buildrun + +import ( + "context" + "fmt" + "sort" + "strconv" + "strings" + "sync" +) + +// DefaultCapacity is the build capacity of a node that has no setting: one build step at a time. +const DefaultCapacity = 1 + +// ParseCapacity reads a capacity setting: `node=steps` pairs separated by commas, such as +// `shanks=3,ace=2`. A node without a pair has DefaultCapacity. A pair with no node, a count that is not a whole number of +// at least 0, and a node named twice are refused. A count of 0 keeps the part off the node: it is never chosen. +func ParseCapacity(s string) (map[string]int, error) { + out := map[string]int{} + for _, pair := range strings.Split(s, ",") { + pair = strings.TrimSpace(pair) + if pair == "" { + continue + } + node, count, ok := strings.Cut(pair, "=") + node, count = strings.TrimSpace(node), strings.TrimSpace(count) + n, err := strconv.Atoi(count) + if !ok || node == "" || err != nil || n < 0 { + return nil, fmt.Errorf("the capacity %q is not =", pair) + } + if _, dup := out[node]; dup { + return nil, fmt.Errorf("the capacity names %s twice", node) + } + out[node] = n + } + return out, nil +} + +// Places counts the build steps a part runs on each node and holds a lane back while its node is at its +// capacity (the operator's decision of 2026-10-21; ADR 0478 clause 3). A lane is the fetch and the builds of one build +// module on one node, run one step at a time, so a lane holds one place while it runs. A node is chosen for a lane by +// reserving a place on it at the moment of the choice (ChooseNodes), so that builds choosing at once see each other's +// choices and none waits on a full node while another has room. A lane waits for its place by watching the ends of the +// other lanes, not by a timer: ending a lane wakes the waiting ones. +type Places struct { + // Capacity is the number of steps at once by node; a node not in it has DefaultCapacity. + Capacity map[string]int + + mu sync.Mutex + running map[string]int // lanes holding a place + reserved map[string]int // lanes chosen for the node that do not hold a place yet + changed chan struct{} // closed, and replaced, each time a lane ends +} + +// CapacityOf is how many steps the part runs on the node at once: DefaultCapacity for a node the setting does not +// name, and 0 for one it names so, where the part never builds. +func (p *Places) CapacityOf(node string) int { + if n, ok := p.Capacity[node]; ok && n >= 0 { + return n + } + return DefaultCapacity +} + +// Running is the steps in flight now, by node. +func (p *Places) Running() map[string]int { + p.mu.Lock() + defer p.mu.Unlock() + out := make(map[string]int, len(p.running)) + for n, c := range p.running { + if c > 0 { + out[n] = c + } + } + return out +} + +// Reserved is the lanes chosen for a node that hold no place yet, by node. +func (p *Places) Reserved() map[string]int { + p.mu.Lock() + defer p.mu.Unlock() + out := make(map[string]int, len(p.reserved)) + for n, c := range p.reserved { + if c > 0 { + out[n] = c + } + } + return out +} + +func (p *Places) init() { + if p.running == nil { + p.running = map[string]int{} + } + if p.reserved == nil { + p.reserved = map[string]int{} + } + if p.changed == nil { + p.changed = make(chan struct{}) + } +} + +// Acquire takes a place on the node, waiting while the node is at its capacity, and answers the function that gives it +// back. A place reserved for the node by a choice is the one taken. It returns an error when ctx ends first, and then +// gives up the reservation too. +func (p *Places) Acquire(ctx context.Context, node string) (release func(), err error) { + for { + p.mu.Lock() + p.init() + if p.running[node] < p.CapacityOf(node) { + p.running[node]++ + if p.reserved[node] > 0 { + p.reserved[node]-- + } + p.mu.Unlock() + var once sync.Once + return func() { + once.Do(func() { + p.mu.Lock() + p.running[node]-- + close(p.changed) + p.changed = make(chan struct{}) + p.mu.Unlock() + }) + }, nil + } + wait := p.changed + p.mu.Unlock() + select { + case <-wait: + case <-ctx.Done(): + p.Cancel(node) + return nil, fmt.Errorf("waiting for a place on %s: %w", node, ctx.Err()) + } + } +} + +// Cancel gives up a place reserved for the node that will not be taken. +func (p *Places) Cancel(node string) { + p.mu.Lock() + defer p.mu.Unlock() + p.init() + if p.reserved[node] > 0 { + p.reserved[node]-- + } +} + +// choose orders the nodes (those with room first, counting the lanes running and the lanes reserved, then the fewest of +// them, then by name), keeps the first want and reserves a place on each, all at once. +func (p *Places) choose(nodes []string, want int) []string { + p.mu.Lock() + defer p.mu.Unlock() + p.init() + load := func(n string) int { return p.running[n] + p.reserved[n] } + sort.SliceStable(nodes, func(i, j int) bool { + ri, rj := load(nodes[i]) < p.CapacityOf(nodes[i]), load(nodes[j]) < p.CapacityOf(nodes[j]) + if ri != rj { + return ri + } + if load(nodes[i]) != load(nodes[j]) { + return load(nodes[i]) < load(nodes[j]) + } + return nodes[i] < nodes[j] + }) + if len(nodes) > want { + nodes = nodes[:want] + } + for _, n := range nodes { + p.reserved[n]++ + } + return nodes +} diff --git a/go/delivery/buildrun/places_test.go b/go/delivery/buildrun/places_test.go new file mode 100644 index 0000000..384c537 --- /dev/null +++ b/go/delivery/buildrun/places_test.go @@ -0,0 +1,126 @@ +package buildrun + +import ( + "context" + "strings" + "testing" + "time" +) + +func TestParseCapacityReadsNodeEqualsCountPairs(t *testing.T) { + c, err := ParseCapacity(" shanks=3, ace=2 ") + if err != nil || c["shanks"] != 3 || c["ace"] != 2 || len(c) != 2 { + t.Errorf("capacity = %v, %v", c, err) + } + if c, err := ParseCapacity(""); err != nil || len(c) != 0 { + t.Errorf("empty: %v, %v", c, err) + } + for _, bad := range []string{"shanks", "shanks=x", "=2", "a=1,a=2", "shanks=-1"} { + if _, err := ParseCapacity(bad); err == nil { + t.Errorf("%q accepted", bad) + } + } +} + +func acquireAsync(p *Places, node string) chan func() { + got := make(chan func(), 1) + go func() { + release, err := p.Acquire(context.Background(), node) + if err != nil { + close(got) + return + } + got <- release + }() + return got +} + +func TestThreeStepsOnANodeOfCapacityTwoRunTwoAtOnceAndTheThirdStartsWhenOneEnds(t *testing.T) { // round 3 + p := &Places{Capacity: map[string]int{"ace": 2}} + a, b := acquireAsync(p, "ace"), acquireAsync(p, "ace") + ra, rb := <-a, <-b + third := acquireAsync(p, "ace") + select { + case <-third: + t.Fatal("a third step started on a node of capacity 2 with two running") + case <-time.After(50 * time.Millisecond): + } + if p.Running()["ace"] != 2 { + t.Errorf("running = %v, want 2 on ace", p.Running()) + } + ra() + select { + case r := <-third: + r() + case <-time.After(2 * time.Second): + t.Fatal("the third step did not start when one ended") + } + rb() +} + +func TestANodeWithNoSettingRunsOneStepAtATime(t *testing.T) { // round 3 + p := &Places{Capacity: map[string]int{"shanks": 3}} + first := <-acquireAsync(p, "g14") + second := acquireAsync(p, "g14") + select { + case <-second: + t.Fatal("two steps on a node without a setting") + case <-time.After(50 * time.Millisecond): + } + first() + r := <-second + r() + if p.CapacityOf("g14") != 1 || p.CapacityOf("shanks") != 3 { + t.Errorf("capacities = %d, %d", p.CapacityOf("g14"), p.CapacityOf("shanks")) + } +} + +func TestAWaitingStepStopsWhenItsContextEnds(t *testing.T) { // round 3 + p := &Places{} + r := <-acquireAsync(p, "ace") + defer r() + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { _, err := p.Acquire(ctx, "ace"); done <- err }() + cancel() + select { + case err := <-done: + if err == nil { + t.Error("Acquire returned no error for a cancelled wait") + } + case <-time.After(2 * time.Second): + t.Fatal("a cancelled wait did not return") + } +} + +// The operator's decision of 2026-10-21 13:40: the builder builds on shanks and ace only; a capacity of 0 keeps it off a node. +func TestACapacityOfZeroKeepsThePartOffTheNodeAndANodeNotNamedKeepsOne(t *testing.T) { // issue 1661 + c, err := ParseCapacity("shanks=3,ace=2,g14=0,novox=0") + if err != nil || c["g14"] != 0 || c["novox"] != 0 || c["shanks"] != 3 || len(c) != 4 { + t.Fatalf("capacity = %v, %v", c, err) + } + p := &Places{Capacity: c} + if p.CapacityOf("g14") != 0 || p.CapacityOf("novox") != 0 || p.CapacityOf("ace") != 2 || p.CapacityOf("other") != DefaultCapacity { + t.Errorf("capacities = %d %d %d %d, want 0 0 2 and the default for a node not named", p.CapacityOf("g14"), p.CapacityOf("novox"), p.CapacityOf("ace"), p.CapacityOf("other")) + } + if _, err := ParseCapacity("shanks=-1"); err == nil { + t.Error("a negative capacity accepted") + } + entries := tcs("go", "go1.27.1", "ace", "g14", "novox", "shanks", "other") + for i := 0; i < 6; i++ { + nodes, _, err := ChooseNodes("go", entries, p, nil, 2) + if err != nil { + t.Fatal(err) + } + for _, n := range nodes { + if n == "g14" || n == "novox" { + t.Fatalf("choice %d: %v includes a node of capacity 0", i, nodes) + } + } + } + // Only nodes of capacity 0 report: nothing qualifies, and the build says why. + _, _, err = ChooseNodes("go", tcs("go", "go1.27.1", "g14", "novox"), p, nil, 2) + if err == nil || !strings.Contains(err.Error(), "capacity") { + t.Errorf("only capacity-0 nodes: error = %v, want it to name the capacity", err) + } +} diff --git a/go/delivery/buildrun/presence.go b/go/delivery/buildrun/presence.go new file mode 100644 index 0000000..63f8243 --- /dev/null +++ b/go/delivery/buildrun/presence.go @@ -0,0 +1,53 @@ +package buildrun + +import ( + "sync" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// PresenceSet holds which nodes are online, as the watchman writes it (mesh-delivery-watchman_presence, key ``, the +// node-engine's connection): a node asleep or off the bus is offline and its build module cannot answer. A node the +// watchman has no key of is not online. Safe for use from the watch and from a build at once. +type PresenceSet struct { + mu sync.Mutex + online map[string]bool +} + +// Put takes a change of the presence bucket in and answers whether a node's online state changed. A part's key and a +// value that is not a presence are left out. +func (s *PresenceSet) Put(key string, value []byte, deleted bool) bool { + node, part, _, err := delivery.ParsePresenceKey(key) + if err != nil || part != "" { + return false + } + on := false + if !deleted { + p, err := delivery.DecodePresence(value) + if err != nil { + return false + } + on = p.Online + } + s.mu.Lock() + defer s.mu.Unlock() + if s.online[node] == on { + return false + } + if s.online == nil { + s.online = map[string]bool{} + } + if on { + s.online[node] = true + } else { + delete(s.online, node) + } + return true +} + +// Online is whether the watchman last saw the node online. +func (s *PresenceSet) Online(node string) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.online[node] +} diff --git a/go/delivery/buildrun/presence_test.go b/go/delivery/buildrun/presence_test.go new file mode 100644 index 0000000..5157609 --- /dev/null +++ b/go/delivery/buildrun/presence_test.go @@ -0,0 +1,43 @@ +package buildrun + +import ( + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func presenceValue(t *testing.T, online bool) []byte { + t.Helper() + b, err := delivery.EncodePresence(delivery.Presence{Online: online}) + if err != nil { + t.Fatal(err) + } + return b +} + +func TestPresenceSetKnowsOnlyNodesTheWatchmanSawOnline(t *testing.T) { // 1607 review 3 + var s PresenceSet + if s.Online("ace") { + t.Error("a node the watchman never wrote is online") + } + if !s.Put("ace", presenceValue(t, true), false) || !s.Online("ace") { + t.Error("ace online was not taken in") + } + if s.Put("ace", presenceValue(t, true), false) { + t.Error("the same presence again counted as a change") + } + if !s.Put("ace", presenceValue(t, false), false) || s.Online("ace") { + t.Error("ace going offline was not taken in") + } + s.Put("g14", presenceValue(t, true), false) + if !s.Put("g14", nil, true) || s.Online("g14") { + t.Error("a deleted key is still online") + } + // A part's instance (part..) is not a node. + if s.Put("part.mesh-delivery-builder.novox", presenceValue(t, true), false) || s.Online("part") { + t.Error("a part's key was taken for a node") + } + if s.Put("novox", []byte("junk"), false) { + t.Error("junk counted") + } +} diff --git a/go/delivery/buildrun/steps.go b/go/delivery/buildrun/steps.go new file mode 100644 index 0000000..f08e002 --- /dev/null +++ b/go/delivery/buildrun/steps.go @@ -0,0 +1,67 @@ +package buildrun + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Caller calls a tool of a module on one named machine (mesh-sdk process.Conn.CallOn; ADR 0469 clause 3). +type Caller interface { + CallOn(node, key string, body any) (json.RawMessage, error) +} + +// Steps waits for the end of a build module's long step: the value of its key `.` in the module's steps state +// (ADR 0469 clause 2) once its state is not running. It is a watch of that key, not a poll; it returns when the step +// ends or ctx does. +type Steps interface { + Wait(ctx context.Context, module, key string) (delivery.Step, error) +} + +// Runner runs a build module's long steps: the call answers at once with the step's key, and the result is read from that +// key when the step ends (ADR 0469 clause 2). +type Runner struct { + Caller Caller + Steps Steps + // Bound is how long a step may run before the builder stops waiting for it. + Bound time.Duration +} + +// Run calls tool of the build module on node with args, waits for the step's end and answers its result. A step that +// failed or was refused is an error naming the node, the state and the reason. +func (r Runner) Run(ctx context.Context, module, node, tool string, args map[string]any) (json.RawMessage, error) { + raw, err := r.Caller.CallOn(node, module+"."+tool, args) + if err != nil { + return nil, fmt.Errorf("%s on %s: %w", tool, node, err) + } + var answer struct { + Step string `json:"step"` + Key string `json:"key"` + State string `json:"state"` + } + if err := json.Unmarshal(raw, &answer); err != nil { + return nil, fmt.Errorf("%s on %s answered what is not a step: %w", tool, node, err) + } + if answer.Key == "" { + return nil, fmt.Errorf("%s on %s answered no step key: %s", tool, node, raw) + } + if n, _, err := delivery.ParseStepKey(answer.Key); err != nil || n != node { + return nil, fmt.Errorf("%s on %s answered the step key %q, which is not a key of that node", tool, node, answer.Key) + } + wait, cancel := context.WithTimeoutCause(ctx, r.Bound, fmt.Errorf("the step %s did not end within %s", answer.Key, r.Bound)) + defer cancel() + step, err := r.Steps.Wait(wait, module, answer.Key) + if err != nil { + if cause := context.Cause(wait); cause != nil && ctx.Err() == nil { + err = cause + } + return nil, fmt.Errorf("%s on %s: %w", tool, node, err) + } + if step.State != delivery.StepPassed { + return nil, fmt.Errorf("%s on %s: the step %s %s: %s", tool, node, answer.Key, step.State, step.Reason) + } + return step.Result, nil +} diff --git a/go/delivery/buildrun/steps_test.go b/go/delivery/buildrun/steps_test.go new file mode 100644 index 0000000..9d6d024 --- /dev/null +++ b/go/delivery/buildrun/steps_test.go @@ -0,0 +1,116 @@ +package buildrun + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +type call struct { + node, key string + body any +} + +// fakeCaller answers a long tool as the build modules do: at once, with the step and its key. +type fakeCaller struct { + calls []call + answer string + err error +} + +func (f *fakeCaller) CallOn(node, key string, body any) (json.RawMessage, error) { + f.calls = append(f.calls, call{node, key, body}) + if f.err != nil { + return nil, f.err + } + return json.RawMessage(f.answer), nil +} + +// fakeSteps holds the steps values by `/`; Wait on a key it does not hold blocks until the context ends. +type fakeSteps struct { + mu sync.Mutex + steps map[string]delivery.Step + // during, when set, is called as a wait starts and the function it answers as the wait ends. + during func(key string) func() +} + +func (f *fakeSteps) Wait(ctx context.Context, module, key string) (delivery.Step, error) { + f.mu.Lock() + s, ok := f.steps[module+"/"+key] + during := f.during + f.mu.Unlock() + if ok { + if during != nil { + defer during(key)() + } + return s, nil + } + <-ctx.Done() + return delivery.Step{}, context.Cause(ctx) +} + +func stepAt(node, id, state, reason string, result string) delivery.Step { + s := delivery.Step{ID: id, Tool: "build_go_fetch", Node: node, State: state, Reason: reason} + if result != "" { + s.Result = json.RawMessage(result) + } + return s +} + +func TestRunCallsTheToolOnTheNodeAndReadsTheStepsKey(t *testing.T) { // 1607/AC5 + caller := &fakeCaller{answer: `{"step":"abc1","key":"ace.abc1","state":"running"}`} + steps := &fakeSteps{steps: map[string]delivery.Step{"mesh-build-go/ace.abc1": stepAt("ace", "abc1", delivery.StepPassed, "", `{"workspace":"w1","commit":"c"}`)}} + res, err := Runner{Caller: caller, Steps: steps, Bound: time.Second}.Run(context.Background(), "mesh-build-go", "ace", "build_go_fetch", map[string]any{"repository": "novox/r", "commit": "c"}) + if err != nil { + t.Fatal(err) + } + if len(caller.calls) != 1 || caller.calls[0].node != "ace" || caller.calls[0].key != "mesh-build-go.build_go_fetch" { + t.Errorf("calls = %+v, want one on ace for mesh-build-go.build_go_fetch", caller.calls) + } + var fr struct{ Workspace string } + if err := json.Unmarshal(res, &fr); err != nil || fr.Workspace != "w1" { + t.Errorf("result = %s (%v), want the step's result", res, err) + } +} + +func TestRunFailsOnAFailedOrRefusedStepNamingItsReason(t *testing.T) { // 1607/AC5 + for _, state := range []string{delivery.StepFailed, delivery.StepRefused} { + caller := &fakeCaller{answer: `{"step":"abc1","key":"ace.abc1","state":"running"}`} + steps := &fakeSteps{steps: map[string]delivery.Step{"mesh-build-go/ace.abc1": stepAt("ace", "abc1", state, "go is go1.26.0, the recipe builds with go1.27.1", "")}} + _, err := Runner{Caller: caller, Steps: steps, Bound: time.Second}.Run(context.Background(), "mesh-build-go", "ace", "build_go_build", nil) + if err == nil || !strings.Contains(err.Error(), "go1.27.1") || !strings.Contains(err.Error(), state) || !strings.Contains(err.Error(), "ace") { + t.Errorf("%s: error = %v, want the node, the state and the reason", state, err) + } + } +} + +func TestRunStopsWaitingAtTheBound(t *testing.T) { // 1607/AC5 + caller := &fakeCaller{answer: `{"step":"abc1","key":"ace.abc1","state":"running"}`} + _, err := Runner{Caller: caller, Steps: &fakeSteps{}, Bound: 20 * time.Millisecond}.Run(context.Background(), "mesh-build-go", "ace", "build_go_build", nil) + if err == nil || !strings.Contains(err.Error(), "ace.abc1") || !strings.Contains(err.Error(), "did not end") { + t.Errorf("error = %v, want the step key and that it did not end within its bound", err) + } +} + +func TestRunRefusesAnAnswerWithNoKeyAndAKeyOfAnotherNode(t *testing.T) { // 1607/AC5 + for _, answer := range []string{`{"state":"running"}`, `{"step":"abc1","key":"novox.abc1","state":"running"}`, `not json`} { + _, err := Runner{Caller: &fakeCaller{answer: answer}, Steps: &fakeSteps{}, Bound: time.Second}.Run(context.Background(), "mesh-build-go", "ace", "build_go_fetch", nil) + if err == nil { + t.Errorf("answer %s: no error", answer) + } + } +} + +func TestRunReturnsTheCallsOwnError(t *testing.T) { // 1607/AC5 + boom := errors.New("no responders") + _, err := Runner{Caller: &fakeCaller{err: boom}, Steps: &fakeSteps{}, Bound: time.Second}.Run(context.Background(), "mesh-build-go", "ace", "build_go_fetch", nil) + if !errors.Is(err, boom) || !strings.Contains(err.Error(), "ace") { + t.Errorf("error = %v, want the call's error naming the node", err) + } +} diff --git a/go/delivery/buildrun/toolchains.go b/go/delivery/buildrun/toolchains.go new file mode 100644 index 0000000..6170425 --- /dev/null +++ b/go/delivery/buildrun/toolchains.go @@ -0,0 +1,64 @@ +package buildrun + +import ( + "sync" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// LanguageOf is the language each build module's toolchains state holds (the indexer's ToolchainModules, reversed). +var LanguageOf = map[string]string{GoBuildModule: "go", NodeBuildModule: "typescript"} + +// ToolchainSet holds what the build modules' toolchains buckets hold: each node's entry of a language, as the watches +// put them in. Safe for use from the watches and from a build at once. +type ToolchainSet struct { + mu sync.Mutex + by map[string]map[string]delivery.Toolchain // language -> node -> entry +} + +// Put takes a change of a build module's toolchains bucket in: a put of a node's entry (key .) or its +// delete. It answers whether the language's entries changed (a node's version or recipe, or a node appearing or going). +// A key or a value that is not an entry of the module's language is left out. +func (s *ToolchainSet) Put(module, key string, value []byte, deleted bool) bool { + language, ok := LanguageOf[module] + if !ok { + return false + } + node, keyLanguage, err := delivery.ParseToolchainKey(key) + if err != nil || keyLanguage != language { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + if deleted { + if _, had := s.by[language][node]; !had { + return false + } + delete(s.by[language], node) + return true + } + tc, err := delivery.DecodeToolchain(value) + if err != nil || tc.Language != language { + return false + } + if s.by == nil { + s.by = map[string]map[string]delivery.Toolchain{} + } + if s.by[language] == nil { + s.by[language] = map[string]delivery.Toolchain{} + } + cur, had := s.by[language][node] + s.by[language][node] = tc + return !had || cur.Version != tc.Version || cur.Recipe != tc.Recipe +} + +// Entries are the nodes' entries of a language, in no particular order. +func (s *ToolchainSet) Entries(language string) []delivery.NodeToolchain { + s.mu.Lock() + defer s.mu.Unlock() + var out []delivery.NodeToolchain + for node, tc := range s.by[language] { + out = append(out, delivery.NodeToolchain{Node: node, Toolchain: tc}) + } + return out +} diff --git a/go/delivery/buildrun/toolchains_test.go b/go/delivery/buildrun/toolchains_test.go new file mode 100644 index 0000000..e434b8a --- /dev/null +++ b/go/delivery/buildrun/toolchains_test.go @@ -0,0 +1,51 @@ +package buildrun + +import ( + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func tcValue(t *testing.T, language, version string) []byte { + t.Helper() + b, err := delivery.EncodeToolchain(delivery.Toolchain{Language: language, Version: version, Recipe: "1", At: time.Now()}) + if err != nil { + t.Fatal(err) + } + return b +} + +func TestToolchainSetKeepsEachNodesEntryByLanguageAndSaysWhenOneChanged(t *testing.T) { // 1607/AC6 + var s ToolchainSet + if !s.Put(GoBuildModule, "novox.go", tcValue(t, "go", "go1.27.1"), false) { + t.Error("a first entry did not count as a change") + } + if s.Put(GoBuildModule, "novox.go", tcValue(t, "go", "go1.27.1"), false) { + t.Error("the same entry again counted as a change") + } + s.Put(GoBuildModule, "ace.go", tcValue(t, "go", "go1.27.1"), false) + s.Put(NodeBuildModule, "ace.typescript", tcValue(t, "typescript", "tsc5"), false) + if got := s.Entries("go"); len(got) != 2 { + t.Errorf("go entries = %+v, want two", got) + } + if got := s.Entries("typescript"); len(got) != 1 || got[0].Node != "ace" || got[0].Version != "tsc5" { + t.Errorf("typescript entries = %+v", got) + } + if !s.Put(GoBuildModule, "ace.go", nil, true) || len(s.Entries("go")) != 1 { + t.Error("a delete did not remove the node's entry") + } +} + +func TestToolchainSetLeavesOutWhatIsNotAToolchainOfTheModulesLanguage(t *testing.T) { // 1607/AC6 + var s ToolchainSet + if s.Put(GoBuildModule, "novox.typescript", tcValue(t, "typescript", "x"), false) { + t.Error("a typescript toolchain in the go build module's bucket counted") + } + if s.Put(GoBuildModule, "novox.go", []byte("not json"), false) || s.Put("other-module", "novox.go", tcValue(t, "go", "x"), false) { + t.Error("garbage or an unknown module counted") + } + if len(s.Entries("go")) != 0 { + t.Error("an entry was kept") + } +}