From bfc64ac65ea1af14f3feddf4e3b7dad85219d909 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 13:15:28 +0200 Subject: [PATCH] Build an archive with no toolchain, mark digests that differ and a one-node key so each is handled, reserve a place when a node is chosen, choose only nodes whose build module is healthy, and take over any key found building (issue 1607, review round 4) --- .../mesh-delivery-builder/integration_test.go | 131 ++++++-- .../cmd/mesh-delivery-builder/main.go | 25 ++ .../internal/build/choose.go | 46 ++- .../internal/build/claim.go | 45 ++- .../internal/build/claim_test.go | 27 +- .../internal/build/health.go | 56 ++++ .../internal/build/places.go | 86 ++++- .../internal/build/plan.go | 11 + .../internal/build/round4_test.go | 293 ++++++++++++++++++ .../internal/build/run.go | 117 +++++-- .../internal/build/run_test.go | 12 +- .../mesh-delivery-builder/manifest_test.go | 2 +- modules/mesh-delivery-builder/module.json | 3 +- 13 files changed, 749 insertions(+), 105 deletions(-) create mode 100644 modules/mesh-delivery-builder/internal/build/health.go create mode 100644 modules/mesh-delivery-builder/internal/build/round4_test.go diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index 2131046..780387e 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -119,14 +119,16 @@ func (f *fakeBuildModules) serve(t *testing.T, nc *nats.Conn, module string) { } type bed struct { - ctx context.Context - js jetstream.JetStream - modules jetstream.KeyValue - builds jetstream.KeyValue - tools map[string]jetstream.KeyValue - fake *fakeBuildModules - srvURL string - dir string + ctx context.Context + js jetstream.JetStream + modules jetstream.KeyValue + builds jetstream.KeyValue + tools map[string]jetstream.KeyValue + presence jetstream.KeyValue + have jetstream.KeyValue + fake *fakeBuildModules + srvURL string + dir string } // startBed starts the bus, the buckets the controller makes from the builder's and the build modules' manifests, the @@ -194,10 +196,11 @@ func startBed(t *testing.T) *bed { mk(delivery.ManifestsBucket, delivery.ManifestsHistory) b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory) presence := mk(delivery.PresenceBucket, delivery.PresenceHistory) + b.presence, b.have = presence, mk(delivery.HaveBucket, delivery.HaveHistory) for _, node := range []string{"n1", "n2"} { - data, _ := delivery.EncodePresence(delivery.Presence{Online: true}) - if _, err := presence.Put(ctx, node, data); err != nil { - t.Fatal(err) + b.setOnline(t, node, true) + for _, module := range []string{"mesh-build-go", "mesh-build-node"} { + b.setHealth(t, node, module, delivery.HealthOK) } } b.fake = &fakeBuildModules{js: js, ctx: ctx, calls: map[string]int{}} @@ -210,6 +213,28 @@ func startBed(t *testing.T) *bed { return b } +// setOnline writes the watchman's presence of a node. +func (b *bed) setOnline(t *testing.T, node string, online bool) { + t.Helper() + data, _ := delivery.EncodePresence(delivery.Presence{Online: online}) + if _, err := b.presence.Put(b.ctx, node, data); err != nil { + t.Fatal(err) + } +} + +// setHealth writes the node-engine's have of a build module on a node, running with this health. +func (b *bed) setHealth(t *testing.T, node, module string, health delivery.Health) { + t.Helper() + data, err := delivery.EncodeHave(delivery.Have{Phase: delivery.PhaseRunning, Health: health}) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.HaveKey(node, module) + if _, err := b.have.Put(b.ctx, key, data); err != nil { + t.Fatal(err) + } +} + // report writes a node's toolchains key of a build module and says so on the module's stream, as the module does. func (b *bed) report(t *testing.T, module, node, language, version, recipe string) { t.Helper() @@ -265,6 +290,13 @@ func (b *bed) moduleCommit(t *testing.T, name, fixture, path string, toolchains return digest } +// archive writes modules. as the real indexer does for a module of an archive only: its fingerprint names no +// toolchain (mesh-delivery-indexer internal/index inputs.go languagesOf gives an archive no language), so Toolchains is empty. +func (b *bed) archive(t *testing.T, name, path, at string) string { + t.Helper() + return b.moduleCommit(t, name, "catalogue-a67f0196-ca-trust.module.json", path, nil, at) +} + // start runs the program as the builder of a node. func (b *bed) start(t *testing.T, node string, env map[string]string) { t.Helper() @@ -334,19 +366,69 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7 } } -// TestASecondBuilderInstanceDoesNotBuildTheSameSourceAgain: two builders on two nodes start together on one module key; the -// fingerprint is built once. -func TestASecondBuilderInstanceDoesNotBuildTheSameSourceAgain(t *testing.T) { // 1607/AC7 +// TestTwoBuilderInstancesStartingTogetherEndWithOneDoneKey: two builders on two nodes start together on one module key of an +// archive only (no toolchain named, as the indexer writes it); the second takes over the key the first holds building +// (ADR 0478 clause 6), and the key ends done, written once. +func TestTwoBuilderInstancesStartingTogetherEndWithOneDoneKey(t *testing.T) { // 1607/AC7 b := startBed(t) - b.module(t, "ca-trust", "catalogue-a67f0196-ca-trust.module.json", "modules/ca-trust") + b.archive(t, "ca-trust", "modules/ca-trust", commit) b.start(t, "n1", nil) b.start(t, "n2", nil) v := b.await(t, "ca-trust", isState(delivery.BuildDone)) if len(v.Artifacts) != 1 || v.Artifacts[0].Kind != "archive" || v.Artifacts[0].Digest != digestOf("files") { t.Fatalf("build = %+v, want the archive's digest", v) } - if n := b.fake.count("mesh-build-go.build_go_build"); n != 2 { // one instance built it, on two nodes - t.Errorf("build_go_build was called %d times, want 2", n) + if n := b.fake.count("mesh-build-go.build_go_build"); n < 2 || n%2 != 0 { // each instance that built, on two nodes + t.Errorf("build_go_build was called %d times, want two for each instance that built", n) + } +} + +// TestAnArchiveOnlyModuleNamingNoToolchainIsBuilt: the indexer gives an archive no language, so modules.ca-trust has no +// toolchains; the builder builds it on the build module's nodes all the same (review round 4). +func TestAnArchiveOnlyModuleNamingNoToolchainIsBuilt(t *testing.T) { + b := startBed(t) + b.archive(t, "ca-trust", "modules/ca-trust", commit) + b.start(t, "n1", nil) + v := b.await(t, "ca-trust", isState(delivery.BuildDone)) + if len(v.Toolchains) != 0 || len(v.Artifacts) != 1 || strings.HasPrefix(v.Step, "one node (") { + t.Errorf("build = %+v, want done on two nodes naming no toolchain", v) + } +} + +// TestAKeyBuiltOnOneNodeIsBuiltAgainWhenASecondNodeComesOnline: n2 is offline, so the key is done on n1 alone and marked; +// n2 coming online makes the builder build it again on both nodes, as a new revision (ADR 0478 clause 4). +func TestAKeyBuiltOnOneNodeIsBuiltAgainWhenASecondNodeComesOnline(t *testing.T) { + b := startBed(t) + b.setOnline(t, "n2", false) + b.archive(t, "ca-trust", "modules/ca-trust", commit) + b.start(t, "n1", nil) + one := b.await(t, "ca-trust", isState(delivery.BuildDone)) + if !strings.HasPrefix(one.Step, "one node (") { + t.Fatalf("build = %+v, want it marked as built on one node", one) + } + key := mustKey(t, "ca-trust") + e, err := b.builds.Get(b.ctx, key) + if err != nil { + t.Fatal(err) + } + b.setOnline(t, "n2", true) + two := b.await(t, "ca-trust", func(v delivery.BuilderBuild) bool { + return v.State == delivery.BuildDone && !strings.HasPrefix(v.Step, "one node (") + }) + if e2, _ := b.builds.Get(b.ctx, key); e2.Revision() <= e.Revision() || !strings.Contains(two.Step, "n1 and n2") { + t.Errorf("build = %+v at revision %d (was %d), want a new revision built on n1 and n2", two, e2.Revision(), e.Revision()) + } +} + +// TestANodeWhoseBuildModuleIsUnhealthyIsNotUsed: the node-engine's have of mesh-build-go on n2 says failed, so only n1 builds. +func TestANodeWhoseBuildModuleIsUnhealthyIsNotUsed(t *testing.T) { + b := startBed(t) + b.setHealth(t, "n2", "mesh-build-go", delivery.HealthFailed) + b.archive(t, "ca-trust", "modules/ca-trust", commit) + b.start(t, "n1", nil) + v := b.await(t, "ca-trust", isState(delivery.BuildDone)) + if !strings.Contains(v.Step, "on one node, n1") { + t.Errorf("build = %+v, want it built on n1 alone", v) } } @@ -356,7 +438,7 @@ func TestASecondBuilderInstanceDoesNotBuildTheSameSourceAgain(t *testing.T) { // func TestAFailedKeyIsBuiltAgainAsANewRevisionFromANewModulesValueWithTheSameFingerprint(t *testing.T) { // 1607/AC6, AC7 b := startBed(t) b.fake.failBuild.Store(true) - b.module(t, "ca-trust", "catalogue-a67f0196-ca-trust.module.json", "modules/ca-trust") + b.archive(t, "ca-trust", "modules/ca-trust", commit) b.start(t, "n1", nil) v := b.await(t, "ca-trust", isState(delivery.BuildFailed)) if !strings.Contains(v.Error, "n1") || !strings.Contains(v.Error, "boom") { @@ -369,7 +451,7 @@ func TestAFailedKeyIsBuiltAgainAsANewRevisionFromANewModulesValueWithTheSameFing } b.fake.failBuild.Store(false) // The indexer writes the module again (here at a later commit), with the same fingerprint. - b.moduleAt(t, "ca-trust", "catalogue-a67f0196-ca-trust.module.json", "modules/ca-trust", "go1.27.1", "89abcdef0123456789abcdef0123456789abcdef") + b.archive(t, "ca-trust", "modules/ca-trust", "89abcdef0123456789abcdef0123456789abcdef") b.await(t, "ca-trust", isState(delivery.BuildDone)) done, err := b.builds.Get(b.ctx, key) if err != nil { @@ -387,12 +469,13 @@ func TestAToolchainChangeDoesNotBuildTheOldFingerprint(t *testing.T) { // 1615/A b := startBed(t) b.report(t, "mesh-build-go", "n1", "go", "go1.27.2", "1") b.report(t, "mesh-build-go", "n2", "go", "go1.27.2", "1") - b.module(t, "ca-trust", "catalogue-a67f0196-ca-trust.module.json", "modules/ca-trust") // names go1.27.1 + b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck") // names go1.27.1 b.start(t, "n1", nil) // Another module, whose fingerprint names the new toolchain, is built; by then the old one has been looked at. - b.moduleAt(t, "ca-trust2", "catalogue-a67f0196-ca-trust.module.json", "modules/ca-trust", "go1.27.2", commit) - b.await(t, "ca-trust2", isState(delivery.BuildDone)) - if _, err := b.builds.Get(b.ctx, mustKey(t, "ca-trust")); err == nil { + b.moduleCommit(t, "netcheck2", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck", + map[string]delivery.ToolchainUsed{"go": {Version: "go1.27.2", Recipe: "1"}, "typescript": {Version: "tsc5", Recipe: "1"}}, commit) + b.await(t, "netcheck2", isState(delivery.BuildDone)) + if _, err := b.builds.Get(b.ctx, mustKey(t, "netcheck")); err == nil { t.Error("the old fingerprint, naming go1.27.1, was built with go1.27.2") } if n := b.fake.count("mesh-build-go.build_go_build"); n != 2 { @@ -403,7 +486,7 @@ func TestAToolchainChangeDoesNotBuildTheOldFingerprint(t *testing.T) { // 1615/A // TestARootModuleIsBuilt: a module whose path is "." is given to the build tools as it is. func TestARootModuleIsBuilt(t *testing.T) { // 1615/AC4 b := startBed(t) - b.module(t, "ca-trust", "catalogue-a67f0196-ca-trust.module.json", delivery.RootPath) + b.archive(t, "ca-trust", delivery.RootPath, commit) b.start(t, "n1", nil) b.await(t, "ca-trust", isState(delivery.BuildDone)) } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 70521de..f93a061 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -164,11 +164,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } tools := &build.ToolchainSet{} presence := &build.PresenceSet{} + have := &build.HaveSet{} b := &build.Builder{ Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module}, Runner: build.Runner{Caller: caller, Steps: stepsWaiter{js}, Bound: bound}, Toolchains: tools, Presence: presence, + Health: have, Places: &build.Places{Capacity: capacity}, Manifests: manifestReader{manifests}, Nodes: nodesPerFingerprint, @@ -217,6 +219,27 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro }) }() } + // The node-engine's have says whether a node's build module runs and is healthy: a node whose build module is not is not + // chosen (ADR 0478 clause 3). Where its bucket is not there yet no build module counts as healthy, so nothing builds. + haveErrs := make(chan error, 1) + switch kv, err := js.KeyValue(ctx, delivery.HaveBucket); { + case errors.Is(err, jetstream.ErrBucketNotFound): + logger.Printf("the bucket %s does not exist yet: no build module counts as healthy until the builder starts again", delivery.HaveBucket) + case err != nil: + return fmt.Errorf("the bucket %s: %w", delivery.HaveBucket, err) + default: + hw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { have.Put(e.Key(), e.Value(), !opOf(e)) }) + if err != nil { + return err + } + go func() { + haveErrs <- follow(ctx, hw, func(e jetstream.KeyValueEntry) { + if have.Put(e.Key(), e.Value(), !opOf(e)) { + b.Reconsider(ctx) // a build module became healthy or stopped being: a build that waited may go on + } + }) + }() + } each := func(e jetstream.KeyValueEntry) { if opOf(e) { b.Module(ctx, e.Key(), e.Value()) @@ -241,5 +264,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err case err := <-presenceErrs: return err + case err := <-haveErrs: + return err } } diff --git a/modules/mesh-delivery-builder/internal/build/choose.go b/modules/mesh-delivery-builder/internal/build/choose.go index 656684a..a71c61c 100644 --- a/modules/mesh-delivery-builder/internal/build/choose.go +++ b/modules/mesh-delivery-builder/internal/build/choose.go @@ -9,18 +9,31 @@ import ( ) // 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 steps the -// builder has running there, a tie broken by the node's name. A node at its capacity is still chosen when the others are -// too (fewer than want have room): its steps wait for a place (Places.Acquire). A node qualifies only when online is true of it (the watchman's presence of the node; nil means every -// node is). It answers the toolchain all the reporting nodes agree on, online 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 is online 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, online func(node string) bool, want int) ([]string, delivery.Toolchain2, error) { +// version the nodes agree on, the ones with room under their build capacity first (places), then the fewest lanes the +// builder 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) { nodes := make([]string, 0, len(entries)) var reporting []string seen := map[string]bool{} @@ -30,24 +43,25 @@ func ChooseNodes(language string, entries []delivery.NodeToolchain, places *Plac } seen[e.Node] = true reporting = append(reporting, e.Node) - if online == nil || online(e.Node) { + if qualifies == nil || qualifies(e.Node) { nodes = append(nodes, e.Node) } } if len(nodes) == 0 { sort.Strings(reporting) - return nil, tc, fmt.Errorf("no %s build node qualifies: %s report the toolchain, and none is online", language, strings.Join(reporting, ", ")) + return nil, fmt.Errorf("no %s build node qualifies: %s report the toolchain, and none is online with a healthy build module", language, strings.Join(reporting, ", ")) } if places == nil { places = &Places{} } - places.order(nodes) // nodes with room first, then the fewest steps running, then by name - if len(nodes) > want { - nodes = nodes[:want] - } - return nodes, tc, nil + 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) { @@ -63,7 +77,7 @@ func Agree(artifact string, byNode map[string]string) (string, error) { continue } if byNode[n] != first { - msg := fmt.Sprintf("the nodes built %s to different digests:", artifact) + 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]) } diff --git a/modules/mesh-delivery-builder/internal/build/claim.go b/modules/mesh-delivery-builder/internal/build/claim.go index 0411c4d..2875d57 100644 --- a/modules/mesh-delivery-builder/internal/build/claim.go +++ b/modules/mesh-delivery-builder/internal/build/claim.go @@ -34,14 +34,44 @@ type Claim struct { Build delivery.BuilderBuild } +// OneNodePrefix opens the Step of a done build that a build module built on one node only, because no second node +// qualified (ADR 0478 clause 4); the build modules that did so follow in parentheses, then ": ". Claim reads it to build +// the key again, as a new revision, once a second node qualifies. +const OneNodePrefix = "one node (" + +// OneNodeModules are the build modules a done build built on one node only, or none when it is not such a build. +func OneNodeModules(b delivery.BuilderBuild) []string { + if b.State != delivery.BuildDone || !strings.HasPrefix(b.Step, OneNodePrefix) { + return nil + } + list, _, ok := strings.Cut(strings.TrimPrefix(b.Step, OneNodePrefix), "): ") + if !ok || list == "" { + return nil + } + return strings.Split(list, ", ") +} + +// ClaimOptions say what the builder knows about the moment of a claim. +type ClaimOptions struct { + // TakeOver takes a key found building whoever holds it: the builder has just started and looks at the key for the first + // time, and a build left building by a builder that is gone, on this node or another, is nobody's (ADR 0478 clause 6). + TakeOver bool + // MoreNodes says whether one of these build modules, which built a done key on one node only, has a second node that + // qualifies now. Nil means none does. + MoreNodes func(modules []string) bool +} + // Claim takes the build of module at fingerprint, or answers nil when it is not this builder's to build now: // - no key: it is created building, by this builder; a create that finds the key made meanwhile gets nothing; -// - a done key is never taken again (the same source is never built twice); +// - a done key is not taken again (the same source is never built twice), except one built on one node only (OneNodePrefix) +// when a second node qualifies now (ClaimOptions.MoreNodes): that is taken as it stands, with the done value, and its +// builder writes the new revision when the second build ends (ADR 0478 clause 4); // - a failed key is taken again as a new revision of the same key (ADR 0429 clause 18), except one failed for a kind no -// builder builds yet (RefusedPrefix, ADR 0478 clause 2), which waits for another fingerprint; +// builder builds yet (RefusedPrefix, ADR 0478 clause 2) or for digests that differ (DigestsDifferPrefix, clause 4), +// which wait for another fingerprint; // - a key left queued or building by this builder is taken as it stands (ADR 0478 clause 5: a stopped builder takes its -// own again when it starts); one held by another builder is left, since nothing reclaims without presence. -func (c Claimer) Claim(ctx context.Context, module, fingerprint, manifest string) (*Claim, error) { +// own again when it starts), and so is one left by another builder when ClaimOptions.TakeOver (clause 6). +func (c Claimer) Claim(ctx context.Context, module, fingerprint, manifest string, opt ClaimOptions) (*Claim, error) { key, err := delivery.BuilderBuildKey(module, fingerprint) if err != nil { return nil, err @@ -60,12 +90,15 @@ func (c Claimer) Claim(ctx context.Context, module, fingerprint, manifest string } switch { case cur.State == delivery.BuildDone: + if modules := OneNodeModules(cur); len(modules) > 0 && opt.MoreNodes != nil && opt.MoreNodes(modules) { + return &Claim{Key: key, Rev: rev, Build: cur}, nil + } return nil, nil - case cur.State == delivery.BuildFailed && strings.HasPrefix(cur.Error, RefusedPrefix): + case cur.State == delivery.BuildFailed && (strings.HasPrefix(cur.Error, RefusedPrefix) || strings.HasPrefix(cur.Error, DigestsDifferPrefix)): return nil, nil case cur.State == delivery.BuildFailed: return c.write(ctx, key, building, func(data []byte) (uint64, error) { return c.Store.Update(ctx, key, data, rev) }) - case cur.By == c.By: + case cur.By == c.By || opt.TakeOver: return &Claim{Key: key, Rev: rev, Build: cur}, nil } return nil, nil diff --git a/modules/mesh-delivery-builder/internal/build/claim_test.go b/modules/mesh-delivery-builder/internal/build/claim_test.go index 7589524..644b586 100644 --- a/modules/mesh-delivery-builder/internal/build/claim_test.go +++ b/modules/mesh-delivery-builder/internal/build/claim_test.go @@ -94,14 +94,14 @@ func valueOf(t *testing.T, st *memStore, key string) delivery.BuilderBuild { func TestClaimCreatesTheKeyBuildingAndASecondBuilderGetsNothing(t *testing.T) { // 1607/AC3 st := newMem() a, b := Claimer{Store: st, By: "novox.a"}, Claimer{Store: st, By: "novox.b"} - got, err := a.Claim(context.Background(), "m", fp1, hex64('a')) + got, err := a.Claim(context.Background(), "m", fp1, hex64('a'), ClaimOptions{}) if err != nil || got == nil { t.Fatalf("first claim = %v, %v, want a claim", got, err) } if v := valueOf(t, st, got.Key); v.State != delivery.BuildBuilding || v.By != "novox.a" || v.Manifest != hex64('a') { t.Errorf("the key holds %+v, want building by novox.a with the manifest digest", v) } - if second, err := b.Claim(context.Background(), "m", fp1, hex64('a')); err != nil || second != nil { + if second, err := b.Claim(context.Background(), "m", fp1, hex64('a'), ClaimOptions{}); err != nil || second != nil { t.Errorf("second builder's claim = %v, %v, want none: the same source is never built twice", second, err) } } @@ -109,7 +109,7 @@ func TestClaimCreatesTheKeyBuildingAndASecondBuilderGetsNothing(t *testing.T) { func TestClaimLosingTheCreateRaceGetsNothing(t *testing.T) { // 1607/AC3 st := newMem() st.before = func() { seed(t, st, delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.b"}) } - got, err := Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, "") + got, err := Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, "", ClaimOptions{}) if err != nil || got != nil { t.Errorf("claim = %v, %v, want none: the create failed because the key exists", got, err) } @@ -119,7 +119,7 @@ func TestADoneKeyIsNeverTakenAgain(t *testing.T) { // 1607/AC3 st := newMem() seed(t, st, delivery.BuilderBuild{State: delivery.BuildDone, By: "novox.a", Manifest: hex64('a'), Artifacts: []delivery.BuiltArtifact{{Name: "x", Digest: digest}}}) writes := st.writes - got, err := Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, hex64('a')) + got, err := Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, hex64('a'), ClaimOptions{}) if err != nil || got != nil || st.writes != writes { t.Errorf("claim = %v, %v, writes %d (was %d), want none and no write", got, err, st.writes, writes) } @@ -129,7 +129,7 @@ func TestAFailedKeyIsTakenAgainAsANewRevisionExceptARefusedKind(t *testing.T) { st := newMem() key := seed(t, st, delivery.BuilderBuild{State: delivery.BuildFailed, By: "novox.a", Error: "the step failed"}) rev := st.revs[key] - got, err := Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, hex64('a')) + got, err := Claimer{Store: st, By: "novox.a"}.Claim(context.Background(), "m", fp1, hex64('a'), ClaimOptions{}) if err != nil || got == nil || st.revs[key] <= rev { t.Fatalf("claim = %v, %v, revision %d (was %d), want a claim as a new revision of the same key", got, err, st.revs[key], rev) } @@ -140,26 +140,15 @@ func TestAFailedKeyIsTakenAgainAsANewRevisionExceptARefusedKind(t *testing.T) { st = newMem() key = seed(t, st, delivery.BuilderBuild{State: delivery.BuildFailed, By: "novox.a", Error: RefusedPrefix + " image (artifact x): build-agent builds it"}) writes := st.writes - if got, err := (Claimer{Store: st, By: "novox.a"}).Claim(context.Background(), "m", fp1, ""); err != nil || got != nil || st.writes != writes { + if got, err := (Claimer{Store: st, By: "novox.a"}).Claim(context.Background(), "m", fp1, "", ClaimOptions{}); err != nil || got != nil || st.writes != writes { t.Errorf("a refused kind: claim = %v, %v, writes %d (was %d), want none and no write", got, err, st.writes, writes) } } -func TestABuildLeftBuildingBySameBuilderIsResumedAndAnotherBuildersIsLeft(t *testing.T) { // 1607/AC3, ADR 0478 clause 5 - st := newMem() - seed(t, st, delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.a"}) - if got, err := (Claimer{Store: st, By: "novox.a"}).Claim(context.Background(), "m", fp1, ""); err != nil || got == nil { - t.Errorf("own building key: claim = %v, %v, want it taken again", got, err) - } - if got, err := (Claimer{Store: st, By: "novox.b"}).Claim(context.Background(), "m", fp1, ""); err != nil || got != nil { - t.Errorf("another builder's building key: claim = %v, %v, want none: nothing reclaims without presence", got, err) - } -} - func TestFinishMovesTheKeyByCompareAndSetAndWritesOnlyADifferentValue(t *testing.T) { // 1607/AC3 st := newMem() c := Claimer{Store: st, By: "novox.a"} - got, _ := c.Claim(context.Background(), "m", fp1, hex64('a')) + got, _ := c.Claim(context.Background(), "m", fp1, hex64('a'), ClaimOptions{}) done := delivery.BuilderBuild{State: delivery.BuildDone, By: "novox.a", Manifest: hex64('a'), Artifacts: []delivery.BuiltArtifact{{Name: "x", Kind: "archive", Digest: digest}}} if err := c.Finish(context.Background(), got, done); err != nil { t.Fatal(err) @@ -179,7 +168,7 @@ func TestFinishMovesTheKeyByCompareAndSetAndWritesOnlyADifferentValue(t *testing func TestFinishAfterTheKeyChangedIsAConflictNotAnOverwrite(t *testing.T) { // 1607/AC3 st := newMem() c := Claimer{Store: st, By: "novox.a"} - got, _ := c.Claim(context.Background(), "m", fp1, "") + got, _ := c.Claim(context.Background(), "m", fp1, "", ClaimOptions{}) st.put(got.Key, []byte(`{"v":1,"state":"failed","error":"someone else"}`)) err := c.Finish(context.Background(), got, delivery.BuilderBuild{State: delivery.BuildFailed, Error: "mine"}) if !errors.Is(err, ErrConflict) { diff --git a/modules/mesh-delivery-builder/internal/build/health.go b/modules/mesh-delivery-builder/internal/build/health.go new file mode 100644 index 0000000..d75e40b --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/health.go @@ -0,0 +1,56 @@ +package build + +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/modules/mesh-delivery-builder/internal/build/places.go b/modules/mesh-delivery-builder/internal/build/places.go index 3ff6701..18e2c30 100644 --- a/modules/mesh-delivery-builder/internal/build/places.go +++ b/modules/mesh-delivery-builder/internal/build/places.go @@ -36,16 +36,20 @@ func ParseCapacity(s string) (map[string]int, error) { return out, nil } -// Places counts the build steps the builder runs on each node and holds a step back while its node is at its -// capacity (the operator's decision of 2026-10-21; ADR 0478 clause 3). A step waits for a place by watching the ends of -// the other steps, not by a timer: ending a step wakes the waiting ones. +// Places counts the build steps the builder 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 (choose), 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 - changed chan struct{} // closed, and replaced, each time a step ends + 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 builder runs on the node at once. @@ -69,17 +73,43 @@ func (p *Places) Running() map[string]int { 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. It returns an error when ctx ends first. +// 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() - if p.running == nil { - p.running = map[string]int{} - p.changed = make(chan struct{}) - } + 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() { @@ -97,22 +127,44 @@ func (p *Places) Acquire(ctx context.Context, node string) (release func(), err select { case <-wait: case <-ctx.Done(): + p.Cancel(node) return nil, fmt.Errorf("waiting for a place on %s: %w", node, ctx.Err()) } } } -// order orders the nodes for a choice: those under their capacity first, then fewest running, then by name. -func (p *Places) order(nodes []string) { - running := p.Running() +// 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 := running[nodes[i]] < p.CapacityOf(nodes[i]), running[nodes[j]] < p.CapacityOf(nodes[j]) + ri, rj := load(nodes[i]) < p.CapacityOf(nodes[i]), load(nodes[j]) < p.CapacityOf(nodes[j]) if ri != rj { return ri } - if running[nodes[i]] != running[nodes[j]] { - return running[nodes[i]] < running[nodes[j]] + 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/modules/mesh-delivery-builder/internal/build/plan.go b/modules/mesh-delivery-builder/internal/build/plan.go index 78aff09..d765825 100644 --- a/modules/mesh-delivery-builder/internal/build/plan.go +++ b/modules/mesh-delivery-builder/internal/build/plan.go @@ -46,6 +46,17 @@ func (p Plan) Modules() []string { return out } +// NeedsToolchain is whether any task the build module builds is built with a toolchain: a bundle is, an archive is not +// (the indexer gives an archive no language, so a fingerprint names no toolchain for it; ADR 0478 clause 1). +func (p Plan) NeedsToolchain(module string) bool { + for _, t := range p.Tasks { + if t.Module == module && t.Kind != catalogue.ArtifactArchive { + return true + } + } + return false +} + // Refused is the error of a plan the builder does not build: an artifact of a kind (or of a language) it has no build // module for. type Refused struct{ Artifact, What string } diff --git a/modules/mesh-delivery-builder/internal/build/round4_test.go b/modules/mesh-delivery-builder/internal/build/round4_test.go new file mode 100644 index 0000000..fc7f638 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/round4_test.go @@ -0,0 +1,293 @@ +package build + +import ( + "context" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func haveValue(t *testing.T, phase delivery.Phase, health delivery.Health) []byte { + t.Helper() + b, err := delivery.EncodeHave(delivery.Have{Phase: phase, Health: health}) + if err != nil { + t.Fatal(err) + } + return b +} + +// 1. A module of an archive only: the real indexer gives an archive no language (inputs.go languagesOf), so its +// fingerprint names no toolchain, and the builder builds it on the build module's nodes with no toolchain agreed. +func TestAnArchiveOnlyModuleIsBuiltAlthoughItsFingerprintNamesNoToolchain(t *testing.T) { // review round 4, 1 + r := newRig(t, archiveA) + v := func() delivery.BuilderBuild { + r.b.Module(context.Background(), "m", moduleValueWith(t, "modules/m", nil)) + r.b.Wait() + if _, ok := r.st.vals[r.key]; !ok { + t.Fatal("no build key was written for the archive-only module") + } + return valueOf(t, r.st, r.key) + }() + if v.State != delivery.BuildDone || len(v.Artifacts) != 1 || v.Artifacts[0].Kind != "archive" || len(v.Toolchains) != 0 { + t.Fatalf("build = %+v, want done with the archive and no toolchain named", v) + } + if n := len(r.bm.callsOf("build_go_build")); n != 2 { + t.Errorf("%d go builds, want one on each of two nodes", n) + } +} + +func TestAnArchiveOnlyModuleNeedsNoAgreedToolchainButNeedsNodesThatQualify(t *testing.T) { // review round 4, 1 + r := newRig(t, archiveA) + r.tools.Put(GoBuildModule, "ace.go", tcValue(t, "go", "go1.26.0"), false) // the nodes disagree: a bundle would wait + r.b.Health = &HaveSet{} // no build module is healthy anywhere + r.b.Module(context.Background(), "m", moduleValueWith(t, "modules/m", nil)) + r.b.Wait() + if v := valueOf(t, r.st, r.key); v.State != delivery.BuildFailed || !strings.Contains(v.Error, "healthy") { + t.Errorf("build = %+v, want failed: no node's build module is healthy", v) + } + have := &HaveSet{} + have.Put("novox.mesh-build-go", haveValue(t, delivery.PhaseRunning, delivery.HealthOK), false) + r.b.Health = have + r.b.Reconsider(context.Background()) + r.b.Wait() + v := valueOf(t, r.st, r.key) + if v.State != delivery.BuildDone || !strings.HasPrefix(v.Step, OneNodePrefix) { + t.Errorf("build = %+v, want done on novox, the one node that qualifies, marked as one node", v) + } +} + +func TestABundleAndAnArchiveOfOneBuildModuleStillNeedTheAgreedToolchain(t *testing.T) { // review round 4, 1 + p, err := PlanOf(catalogue.Manifest{Module: "m", Build: &catalogue.Build{Artifacts: []catalogue.Artifact{goBundle, archiveA}}}) + if err != nil || !p.NeedsToolchain(GoBuildModule) { + t.Errorf("a plan with a Go bundle: %v, %v", p.NeedsToolchain(GoBuildModule), err) + } + p, _ = PlanOf(catalogue.Manifest{Module: "m", Build: &catalogue.Build{Artifacts: []catalogue.Artifact{archiveA}}}) + if p.NeedsToolchain(GoBuildModule) { + t.Error("a plan of an archive needs a toolchain") + } +} + +// 2. The digests-differ failure opens with its prefix, and nothing builds it again. +func TestDifferingDigestsFailTheKeyWithThePrefixAndNothingBuildsItAgain(t *testing.T) { // review round 4, 2 + r := newRig(t, goBundle) + r.bm.digests["ace/tool"] = "sha256:" + hex64('9') + v := r.consider(t, "modules/m") + if v.State != delivery.BuildFailed || !strings.HasPrefix(v.Error, "digests differ:") { + t.Fatalf("build = %+v, want a failure opening with exactly %q", v, "digests differ:") + } + calls, writes := len(r.bm.calls), r.st.writes + r.bm.digests = map[string]string{} // the nodes would now agree + r.b.Reconsider(context.Background()) + r.b.Wait() + r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) + r.b.Wait() + if len(r.bm.calls) != calls || r.st.writes != writes { + t.Errorf("a failed-for-digests key was built again: %d calls (was %d), %d writes (was %d)", len(r.bm.calls), calls, r.st.writes, writes) + } +} + +func TestClaimLeavesAKeyFailedForDifferingDigestsAlone(t *testing.T) { // review round 4, 2 + st := newMem() + seed(t, st, delivery.BuilderBuild{State: delivery.BuildFailed, Error: DigestsDifferPrefix + " the nodes built tool to different digests: ace x; novox y"}) + writes := st.writes + got, err := (Claimer{Store: st, By: "novox.a"}).Claim(context.Background(), "m", fp1, "", ClaimOptions{TakeOver: true}) + if err != nil || got != nil || st.writes != writes { + t.Errorf("claim = %v, %v, %d writes (was %d), want none and no write", got, err, st.writes, writes) + } +} + +// 3. A done key built on one node is built again on a second node once one qualifies. +func TestADoneKeyOfOneNodeIsBuiltAgainOnASecondNodeAndTheDigestsCompared(t *testing.T) { // review round 4, 3 + r := newRig(t, goBundle) + r.pres.Put("ace", presenceValue(t, false), false) // only novox qualifies + v := r.consider(t, "modules/m") + if v.State != delivery.BuildDone || !strings.HasPrefix(v.Step, OneNodePrefix+GoBuildModule+"): ") || !strings.Contains(v.Step, "on one node, novox") { + t.Fatalf("build = %+v, want done on one node, marked", v) + } + if got := OneNodeModules(v); len(got) != 1 || got[0] != GoBuildModule { + t.Errorf("OneNodeModules = %v", got) + } + rev, calls := r.st.revs[r.key], len(r.bm.callsOf("build_go_build")) + // Nothing changes while one node qualifies: looking again builds nothing. + r.b.Reconsider(context.Background()) + r.b.Wait() + if r.st.revs[r.key] != rev || len(r.bm.callsOf("build_go_build")) != calls { + t.Fatal("a one-node key was built again with no second node") + } + // ace comes online: the key is built on ace and novox, as a new revision. + r.pres.Put("ace", presenceValue(t, true), false) + r.b.Reconsider(context.Background()) + r.b.Wait() + v = valueOf(t, r.st, r.key) + if v.State != delivery.BuildDone || strings.HasPrefix(v.Step, OneNodePrefix) || !strings.Contains(v.Step, "ace and novox") { + t.Errorf("build = %+v, want done on ace and novox", v) + } + if r.st.revs[r.key] <= rev { + t.Errorf("revision %d, was %d: not a new revision", r.st.revs[r.key], rev) + } + builds := r.bm.callsOf("build_go_build") + if len(builds) != calls+2 || builds[len(builds)-2].node == builds[len(builds)-1].node { + t.Errorf("%d builds (was %d), want one on each of two nodes", len(builds), calls) + } +} + +func TestASecondNodeThatDisagreesFailsTheDoneKeyWithTheDigestsPrefix(t *testing.T) { // review round 4, 3 + r := newRig(t, goBundle) + r.pres.Put("ace", presenceValue(t, false), false) + r.consider(t, "modules/m") + r.bm.digests["ace/tool"] = "sha256:" + hex64('9') + r.pres.Put("ace", presenceValue(t, true), false) + r.b.Reconsider(context.Background()) + r.b.Wait() + if v := valueOf(t, r.st, r.key); v.State != delivery.BuildFailed || !strings.HasPrefix(v.Error, DigestsDifferPrefix) { + t.Errorf("build = %+v, want failed with digests differ", v) + } +} + +func TestASecondBuildThatFailsForAnotherReasonLeavesTheDoneKeyStanding(t *testing.T) { // review round 4, 3 + r := newRig(t, goBundle) + r.pres.Put("ace", presenceValue(t, false), false) + first := r.consider(t, "modules/m") + r.bm.fails["ace/build_go_fetch"] = "ace is out of disk" + r.pres.Put("ace", presenceValue(t, true), false) + r.b.Reconsider(context.Background()) + r.b.Wait() + if v := valueOf(t, r.st, r.key); !delivery.SameBuilderBuild(v, first) { + t.Errorf("build = %+v, want the first done build to stand: %+v", v, first) + } +} + +// 4. A step never waits on a full node while another qualifying node has room. +func TestChoosesMadeAtOnceSeeEachOthersReservationsSoNoLaneWaitsWhileANodeHasRoom(t *testing.T) { // review round 4, 4 + p := &Places{Capacity: map[string]int{"ace": 2, "g14": 1, "novox": 1, "shanks": 3}} + entries := tcs("go", "go1.27.1", "ace", "g14", "novox", "shanks") + total := 2 + 1 + 1 + 3 + var waiting int + for i := 1; i <= 4; i++ { // four modules choose before any lane has started + room := 0 + for _, n := range []string{"ace", "g14", "novox", "shanks"} { + room += p.CapacityOf(n) - p.Running()[n] - p.Reserved()[n] + } + nodes, _, err := ChooseNodes("go", entries, p, nil, 2) + if err != nil || len(nodes) != 2 { + t.Fatalf("choice %d: %v, %v", i, nodes, err) + } + if room >= 2 { + for _, n := range nodes { + if p.Reserved()[n] > p.CapacityOf(n) { + t.Errorf("choice %d: %v put a lane on full %s while %d places were free", i, nodes, n, room) + } + } + } else { + waiting += 2 - room + } + } + if want := 4*2 - total; waiting != want { + t.Errorf("%d lanes wait, want %d: the 8 lanes less the %d places", waiting, want, total) + } + loads := p.Reserved() + for n, c := range loads { + if c > p.CapacityOf(n) && n != "shanks" && n != "ace" && n != "g14" && n != "novox" { + t.Errorf("unknown node %s", n) + } + } + if loads["shanks"] != 3 || loads["ace"] != 2 || loads["g14"]+loads["novox"] != 3 { + t.Errorf("reservations = %v, want shanks 3, ace 2 and the single-place nodes taking the rest", loads) + } +} + +func TestACancelledReservationFreesTheChoiceAndAnAcquireTakesItsReservation(t *testing.T) { // review round 4, 4 + p := &Places{} + nodes, _, _ := ChooseNodes("go", tcs("go", "go1.27.1", "ace"), p, nil, 1) + if len(nodes) != 1 || p.Reserved()["ace"] != 1 { + t.Fatalf("nodes %v, reserved %v", nodes, p.Reserved()) + } + release, err := p.Acquire(context.Background(), "ace") + if err != nil || p.Reserved()["ace"] != 0 || p.Running()["ace"] != 1 { + t.Fatalf("acquire: %v, reserved %v, running %v", err, p.Reserved(), p.Running()) + } + release() + ChooseNodes("go", tcs("go", "go1.27.1", "ace"), p, nil, 1) + p.Cancel("ace") + if len(p.Reserved()) != 0 { + t.Errorf("reserved after Cancel = %v", p.Reserved()) + } +} + +// 5. A node whose build module is not healthy is not chosen. +func TestHaveSetKnowsOnlyBuildModulesRunningAndHealthy(t *testing.T) { // review round 4, 5 + var s HaveSet + if s.Healthy("ace", GoBuildModule) { + t.Error("a module with no have is healthy") + } + if !s.Put("ace.mesh-build-go", haveValue(t, delivery.PhaseRunning, delivery.HealthOK), false) || !s.Healthy("ace", GoBuildModule) { + t.Error("a running, healthy module is not") + } + if s.Put("ace.mesh-build-go", haveValue(t, delivery.PhaseRunning, delivery.HealthOK), false) { + t.Error("the same state counted as a change") + } + if !s.Put("ace.mesh-build-go", haveValue(t, delivery.PhaseRunning, delivery.HealthFailed), false) || s.Healthy("ace", GoBuildModule) { + t.Error("a module whose health failed is healthy") + } + s.Put("ace.mesh-build-go", haveValue(t, delivery.PhaseRunning, delivery.HealthOK), false) + if !s.Put("ace.mesh-build-go", haveValue(t, delivery.PhaseFailed, delivery.HealthOK), false) || s.Healthy("ace", GoBuildModule) { + t.Error("a module whose phase failed is healthy") + } + s.Put("ace.mesh-build-go", haveValue(t, delivery.PhaseRunning, delivery.HealthOK), false) + if !s.Put("ace.mesh-build-go", nil, true) || s.Healthy("ace", GoBuildModule) { + t.Error("a deleted have is healthy") + } + if s.Put("ace.node", []byte(`{}`), false) || s.Put("ace.mesh-build-go", []byte(`not json`), false) { + t.Error("a node's key or a non-have counted") + } +} + +func TestANodeWhoseBuildModuleIsNotHealthyIsNotChosen(t *testing.T) { // review round 4, 5 + r := newRig(t, goBundle) + have := &HaveSet{} + have.Put("ace.mesh-build-go", haveValue(t, delivery.PhaseRunning, delivery.HealthFailed), false) // go is not the recipe's + have.Put("novox.mesh-build-go", haveValue(t, delivery.PhaseRunning, delivery.HealthOK), false) + r.b.Health = have + v := r.consider(t, "modules/m") + if v.State != delivery.BuildDone || !strings.Contains(v.Step, "on one node, novox") { + t.Fatalf("build = %+v, want done on novox alone", v) + } + for _, c := range r.bm.calls { + if c.node == "ace" { + t.Errorf("a step was called on ace, whose build module is unhealthy: %+v", c) + } + } + have.Put("novox.mesh-build-go", haveValue(t, delivery.PhaseRunning, delivery.HealthFailed), false) + r.st.vals = map[string][]byte{} // a fresh key + r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) + r.b.Wait() + if v := valueOf(t, r.st, r.key); v.State != delivery.BuildFailed || !strings.Contains(v.Error, "healthy") { + t.Errorf("build = %+v, want failed: no node has a healthy build module", v) + } +} + +// 6. A key found building is taken over, whoever holds it. +func TestABuildLeftBuildingByAnyBuilderIsTakenOverAtStart(t *testing.T) { // review round 4, 6; 1607/AC3, ADR 0478 clauses 5, 6 + st := newMem() + seed(t, st, delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.a"}) + if got, err := (Claimer{Store: st, By: "novox.a"}).Claim(context.Background(), "m", fp1, "", ClaimOptions{}); err != nil || got == nil { + t.Errorf("own building key: claim = %v, %v, want it taken again", got, err) + } + if got, err := (Claimer{Store: st, By: "ace.b"}).Claim(context.Background(), "m", fp1, "", ClaimOptions{TakeOver: true}); err != nil || got == nil || got.Build.By != "novox.a" { + t.Errorf("another node's building key at start: claim = %v, %v, want it taken over", got, err) + } + if got, _ := (Claimer{Store: st, By: "ace.b"}).Claim(context.Background(), "m", fp1, "", ClaimOptions{}); got != nil { + t.Errorf("another builder's building key after the start was taken: a running builder's key is its own") + } +} + +func TestABuilderFindingAKeyBuildingByAnotherNodeBuildsItOnStart(t *testing.T) { // review round 4, 6 + r := newRig(t, goBundle) + seed(t, r.st, delivery.BuilderBuild{State: delivery.BuildBuilding, By: "ace.builder", Manifest: manifestD}) + v := r.consider(t, "modules/m") + if v.State != delivery.BuildDone || v.By != "novox.builder" { + t.Errorf("build = %+v, want it taken over and done by novox.builder", v) + } +} diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index cd25c70..f7ade56 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -31,6 +31,11 @@ type Builder struct { // Presence says which nodes are online (the watchman's); a node that is not is not chosen. Presence interface{ Online(node string) bool } Manifests Manifests + // Health says whether a node's build module is running and healthy (the node-engine's have); a node whose build module + // is not is not chosen. Nil means every node's is. + Health interface { + Healthy(node, module string) bool + } // Places holds a step back while its node is at its build capacity; a Builder with none runs one step at a time on each node. Places *Places // Nodes is how many build nodes build each fingerprint: two until release R4e (ADR 0478 clause 4). @@ -47,6 +52,7 @@ type Builder struct { inflight map[string]bool // by build key again map[string]delivery.Module // by build key: the newest value that arrived while the key was being built plans map[string]planned // by manifest digest + tried map[string]bool // by build key: looked at since this process started } type planned struct { @@ -61,7 +67,7 @@ func (b *Builder) init() { limit = 4 } b.sem = make(chan struct{}, limit) - b.known, b.inflight, b.again, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]planned{} + b.known, b.inflight, b.again, b.plans, b.tried = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]planned{}, map[string]bool{} if b.Places == nil { b.Places = &Places{} } @@ -196,7 +202,12 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) { return } } - cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest) + key, _ := delivery.BuilderBuildKey(module, m.FP2) + b.mu.Lock() + first := !b.tried[key] + b.tried[key] = true + b.mu.Unlock() + cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, MoreNodes: func(modules []string) bool { return b.moreNodes(modules) }}) if err != nil { b.logf("%s: %v", module, err) return @@ -213,6 +224,12 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) { return } result.By, result.Manifest = b.Claimer.By, m.Manifest + if cl.Build.State == delivery.BuildDone && result.State == delivery.BuildFailed && !strings.HasPrefix(result.Error, DigestsDifferPrefix) { + // A done key built again on a second node keeps its done value when the new build fails for another reason than + // the digests: the first build stands, and a later change of the nodes looks again. + b.logf("%s: built again on a second node and failed, the done build stands: %s", cl.Key, result.Error) + return + } if err := b.Claimer.Finish(ctx, cl, result); err != nil { b.logf("%s: the build could not be written: %v", cl.Key, err) return @@ -226,6 +243,9 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) { // has agreed (novox/hq ADR 0465) must be the one the fingerprint was computed with (Module.Toolchains). func (b *Builder) stale(m delivery.Module, plan Plan) string { for _, bm := range plan.Modules() { + if !plan.NeedsToolchain(bm) { + continue // an archive is packed with no toolchain + } language := LanguageOf[bm] tc, none := delivery.AgreedToolchain(b.Toolchains.Entries(language)) if none != "" { @@ -262,18 +282,47 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p repository := delivery.RepositoryName(m.Repository) var lanes []*lane used := map[string]delivery.ToolchainUsed{} - var steps []string + var steps, oneNode []string + // Every node chosen holds a reservation until its lane takes the place; a build that ends before its lanes start gives them up. + giveUp := func(nodes ...string) { + for _, n := range nodes { + b.Places.Cancel(n) + } + } + giveUpLanes := func() { + for _, l := range lanes { + b.Places.Cancel(l.node) + } + } for _, bm := range plan.Modules() { language := LanguageOf[bm] - nodes, tc, err := ChooseNodes(language, b.Toolchains.Entries(language), b.Places, b.online, b.Nodes) - if err != nil { - return failed("%s", err) + entries := b.Toolchains.Entries(language) + qualifies := func(node string) bool { return b.qualifies(node, bm) } + var nodes []string + if plan.NeedsToolchain(bm) { + ns, tc, err := ChooseNodes(language, entries, b.Places, qualifies, b.Nodes) + if err != nil { + giveUpLanes() + return failed("%s", err) + } + if want := m.Toolchains[language]; want.Version != tc.Version || want.Recipe != tc.Recipe { + giveUp(ns...) + giveUpLanes() + return delivery.BuilderBuild{} // the agreed toolchain moved since try looked: nothing is written + } + used[language] = delivery.ToolchainUsed{Version: tc.Version, Recipe: tc.Recipe} + nodes = ns + } else { + // An archive is packed with no toolchain (the fingerprint names none): any node of the build module that qualifies. + ns, err := ChooseAnyNodes(language, entries, b.Places, qualifies, b.Nodes) + if err != nil { + giveUpLanes() + return failed("%s", err) + } + nodes = ns } - if want := m.Toolchains[language]; want.Version != tc.Version || want.Recipe != tc.Recipe { - return delivery.BuilderBuild{} // the agreed toolchain moved since try looked: nothing is written - } - used[language] = delivery.ToolchainUsed{Version: tc.Version, Recipe: tc.Recipe} if len(nodes) == 1 { + oneNode = append(oneNode, bm) steps = append(steps, fmt.Sprintf("%s on one node, %s", bm, nodes[0])) } else { sorted := append([]string(nil), nodes...) @@ -319,23 +368,55 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p } artifacts = append(artifacts, delivery.BuiltArtifact{Name: t.Artifact, Kind: t.Kind, Digest: d}) } - return delivery.BuilderBuild{State: delivery.BuildDone, Step: "built " + strings.Join(steps, "; "), Artifacts: artifacts, Toolchains: used} + step := "built " + strings.Join(steps, "; ") + if len(oneNode) > 0 { + // Marked so that Claim builds the key again once a second node qualifies (ADR 0478 clause 4). + step = OneNodePrefix + strings.Join(oneNode, ", ") + "): " + step + } + if len(used) == 0 { + used = nil + } + return delivery.BuilderBuild{State: delivery.BuildDone, Step: step, Artifacts: artifacts, Toolchains: used} } func (b *Builder) online(node string) bool { return b.Presence == nil || b.Presence.Online(node) } -// step runs one step of a build module, once its node has a place for it. -func (b *Builder) step(ctx context.Context, bm, node, tool string, args map[string]any) (json.RawMessage, error) { - release, err := b.Places.Acquire(ctx, node) - if err != nil { - return nil, err +// qualifies is whether the node can build with the build module now: it is online and the build module on it is healthy. +func (b *Builder) qualifies(node, bm string) bool { + return b.online(node) && (b.Health == nil || b.Health.Healthy(node, bm)) +} + +// moreNodes is whether any of these build modules has two nodes that qualify now. +func (b *Builder) moreNodes(modules []string) bool { + for _, bm := range modules { + seen := map[string]bool{} + n := 0 + for _, e := range b.Toolchains.Entries(LanguageOf[bm]) { + if !seen[e.Node] && b.qualifies(e.Node, bm) { + seen[e.Node] = true + n++ + } + } + if n >= 2 { + return true + } } - defer release() + return false +} + +// step runs one step of a build module; the lane that runs it holds its node's place. +func (b *Builder) step(ctx context.Context, bm, node, tool string, args map[string]any) (json.RawMessage, error) { return b.Runner.Run(ctx, bm, node, toolPrefix[bm]+tool, args) } -// runLane fetches the commit on the node and builds each of the lane's artifacts there. +// runLane takes a place on the node, fetches the commit there and builds each of the lane's artifacts, one step at a time. func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m delivery.Module, module string) { + release, err := b.Places.Acquire(ctx, l.node) + if err != nil { + l.err = err + return + } + defer release() raw, err := b.step(ctx, l.module, l.node, "fetch", map[string]any{"repository": repository, "commit": m.Commit}) if err != nil { l.err = err diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index 3a00ba8..34c89a6 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -286,7 +286,10 @@ func TestAKeyTheBuilderLeftBuildingIsBuiltAtStartAndDoneKeysAreNotBuiltTwice(t * } } -func TestTwoInstancesBuildOneFingerprintOnce(t *testing.T) { // 1607/AC7 +// Two instances that start together on one key: the second finds the key building and takes it over (ADR 0478 clause 6: +// a key found building at start is taken over, whoever holds it), so both may build; the key is written once, by the one +// whose compare-and-set comes first, and the other's write is refused, never overwriting it. +func TestTwoInstancesStartingTogetherEndWithOneDoneKey(t *testing.T) { // 1607/AC7, review round 4 r := newRig(t, goBundle) other := &Builder{Claimer: Claimer{Store: r.st, By: "ace.builder"}, Runner: r.b.Runner, Toolchains: r.tools, Presence: r.pres, Manifests: r.mf, Nodes: 2, Logf: t.Logf} value := moduleValue(t, "modules/m") @@ -296,8 +299,11 @@ func TestTwoInstancesBuildOneFingerprintOnce(t *testing.T) { // 1607/AC7 go func() { defer wg.Done(); b.Module(context.Background(), "m", value); b.Wait() }() } wg.Wait() - if n := len(r.bm.callsOf("build_go_build")); n != 2 { // one instance, two nodes - t.Errorf("%d builds, want 2: one instance built it", n) + if v := valueOf(t, r.st, r.key); v.State != delivery.BuildDone || len(v.Artifacts) != 1 { + t.Errorf("build = %+v, want one done key", v) + } + if n := len(r.bm.callsOf("build_go_build")); n < 2 || n%2 != 0 { + t.Errorf("%d builds, want two for each instance that built", n) } } diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 87781bc..374764b 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -81,7 +81,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence"}; !reflect.DeepEqual(sorted(m.Reads), want) { + "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } if want := []string{"mesh-build-go.build_go_build", "mesh-build-go.build_go_fetch", "mesh-build-node.build_node_build", "mesh-build-node.build_node_fetch"}; !reflect.DeepEqual(sorted(m.Invokes), want) { diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index a886fbe..115de62 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -16,7 +16,8 @@ "mesh-build-node.toolchains", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", - "mesh-delivery-watchman.presence" + "mesh-delivery-watchman.presence", + "node-engine.have" ], "invokes": [ "mesh-build-go.build_go_build",