diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go index b456fbd..359b1e9 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go @@ -102,43 +102,6 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key } } -// stepsWaiter waits for a build module's step by watching its key in the module's steps bucket (ADR 0469 clause 2): the -// watch hands over the value the key holds now, so a step that ended before the watch began is read at once, and every -// change after. No timer: the caller bounds the wait with its context. -type stepsWaiter struct{ js bucketOpener } - -func (w stepsWaiter) Wait(ctx context.Context, module, key string) (delivery.Step, error) { - kv, err := w.js.KeyValue(ctx, delivery.StepsBucket(module)) - if err != nil { - return delivery.Step{}, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) - } - watch, err := kv.Watch(ctx, key) - if err != nil { - return delivery.Step{}, err - } - defer watch.Stop() - for { - select { - case <-ctx.Done(): - return delivery.Step{}, ctx.Err() - case e, ok := <-watch.Updates(): - if !ok { - return delivery.Step{}, errors.New("the watch of the step ended") - } - if e == nil || !opOf(e) { - continue - } - step, err := delivery.DecodeStep(e.Value()) - if err != nil { - return delivery.Step{}, fmt.Errorf("the step %s is not a step: %w", key, err) - } - if step.State != delivery.StepRunning { - return step, nil - } - } - } -} - // manifestReader reads manifests. of the indexer and parses the manifest it holds. type manifestReader struct{ kv jetstream.KeyValue } 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 182885b..d7d0faa 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 @@ -22,6 +22,7 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -47,13 +48,12 @@ const ( fp = "src2:3333333333333333333333333333333333333333333333333333333333333333" ) -// fakeBuildModules serve the tools of mesh-build-go and mesh-build-node as the real ones do (ADR 0469 clause 2): a call on -// mesh.mod..call... answers at once with the step and its key, and the step's end is written to -// the key . of the module's steps bucket, some milliseconds later. +// fakeBuildModules are the build modules on the nodes as the real ones run since R4g (ADR 0525; issue 1887 A4): each pulls +// its build seat's asks on its nodes (buildstep.Puller over the seat's workers) and runs an ask's job with step bodies that +// answer as the real tools do, ending the step on . of its steps bucket. type fakeBuildModules struct { js jetstream.JetStream ctx context.Context - n atomic.Int64 failBuild atomic.Bool // refusePush makes every bundle and archive build fail as a push the artifact store refused (ADR 0498 clause 2). refusePush atomic.Bool @@ -75,76 +75,68 @@ func digestOf(artifact string) string { return "sha256:" + hex.EncodeToString(sum[:]) } -func (f *fakeBuildModules) serve(t *testing.T, nc *nats.Conn, module string) { - steps, err := f.js.CreateKeyValue(f.ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket(module), History: 1}) +var prefixOf = map[string]string{"mesh-build-go": "build_go_", "mesh-build-node": "build_node_", "mesh-build-image": "build_image_"} + +// step is one tool's answer, as the real module's. +func (f *fakeBuildModules) step(module, name string, args map[string]any) (any, error) { + tool := prefixOf[module] + name + f.mu.Lock() + f.calls[module+"."+tool]++ + if name == "fetch" { + f.fetches = append(f.fetches, args) + } else { + f.builds = append(f.builds, args) + } + f.mu.Unlock() + switch { + case name == "fetch": + merge := "" + if args["base"] != nil { // a fetch with a base answers the merge commit (mesh-build-go FetchResult) + merge = `,"merge":"` + strings.Repeat("9", 40) + `"` + } + return json.RawMessage(`{"workspace":"w` + buildstep.NewID() + `","commit":"` + commit + `"` + merge + `}`), nil + case f.failBuild.Load() && tool != "build_image_copy": // the build fails wherever it runs + return nil, fmt.Errorf("go build of %v: boom", args["artifact"]) + case tool == "build_image_copy": // the upstream copy answers the digest and the registry's reference (ADR 0498 clause 4) + d := digestOf(fmt.Sprint(args["artifact"])) + return json.RawMessage(`{"artifact":"` + fmt.Sprint(args["artifact"]) + `","digest":"` + d + `","reference":"registry.example/` + fmt.Sprint(args["module"]) + `/` + fmt.Sprint(args["artifact"]) + `@` + d + `","pushed":true}`), nil + case tool != "build_image_build" && f.refusePush.Load(): + return nil, fmt.Errorf("%s%v: the artifact store answered 507 Insufficient Storage", delivery.NotStored, args["artifact"]) + case tool != "build_image_build": // a bundle or an archive is pushed and its reference answered (ADR 0498 clause 1) + d := digestOf(fmt.Sprint(args["artifact"])) + return json.RawMessage(`{"digest":"` + d + `","reference":"artifact-store://` + filepath.Base(fmt.Sprint(args["module"])) + `/` + fmt.Sprint(args["artifact"]) + `/blobs/` + d + `"}`), nil + case tool == "build_image_build" && args["publish"] == true: // a trunk image is pushed: the digest and the reference + d := digestOf(fmt.Sprint(args["artifact"])) + return json.RawMessage(`{"digest":"` + d + `","reference":"registry.example/imaged/server@` + d + `"}`), nil + default: + return json.RawMessage(`{"digest":"` + digestOf(fmt.Sprint(args["artifact"])) + `"}`), nil + } +} + +// serve makes module's steps bucket and its build seat's stream and workers, and pulls its asks on nodes. +func (f *fakeBuildModules) serve(t *testing.T, module string, nodes ...string) { + kv, err := f.js.CreateKeyValue(f.ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket(module), History: 1}) if err != nil { t.Fatal(err) } - sub, err := nc.Subscribe("mesh.mod."+module+".call.>", func(m *nats.Msg) { - tokens := strings.Split(m.Subject, ".") // mesh mod call - tool, node := tokens[4], tokens[5] - f.mu.Lock() - f.calls[module+"."+tool]++ - f.mu.Unlock() - var args map[string]any - _ = json.Unmarshal(m.Data, &args) - if strings.HasSuffix(tool, "_fetch") { - f.mu.Lock() - f.fetches = append(f.fetches, args) - f.mu.Unlock() - } else { - f.mu.Lock() - f.builds = append(f.builds, args) - f.mu.Unlock() - } - id := fmt.Sprintf("s%d", f.n.Add(1)) - key, _ := delivery.StepKey(node, id) - put := func(s delivery.Step) { - data, err := delivery.EncodeStep(s) - if err == nil { - _, err = steps.Put(f.ctx, key, data) - } - if err != nil { - t.Errorf("writing step %s: %v", key, err) - } - } - step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepRunning, Started: time.Now().UTC()} - put(step) - answer, _ := json.Marshal(map[string]any{"result": map[string]any{"step": id, "key": key, "state": "running"}, "error": "", "node": node}) - _ = m.Respond(answer) - go func() { - time.Sleep(30 * time.Millisecond) - step.Ended = time.Now().UTC() - switch { - case strings.HasSuffix(tool, "_fetch"): - merge := "" - if args["base"] != nil { // a fetch with a base answers the merge commit (mesh-build-go FetchResult) - merge = `,"merge":"` + strings.Repeat("9", 40) + `"` - } - step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"workspace":"w`+id+`","commit":"`+commit+`"`+merge+`}`) - case f.failBuild.Load() && tool != "build_image_copy": // every node's build fails: one node builds a fingerprint - step.State, step.Reason = delivery.StepFailed, fmt.Sprintf("go build of %v: boom", args["artifact"]) - case tool == "build_image_copy": // the upstream copy answers the digest and the registry's reference (ADR 0498 clause 4) - d := digestOf(fmt.Sprint(args["artifact"])) - step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"artifact":"`+fmt.Sprint(args["artifact"])+`","digest":"`+d+`","reference":"registry.example/`+fmt.Sprint(args["module"])+`/`+fmt.Sprint(args["artifact"])+`@`+d+`","pushed":true}`) - case tool != "build_image_build" && f.refusePush.Load(): - step.State, step.Reason = delivery.StepFailed, fmt.Sprintf("%s%v: the artifact store answered 507 Insufficient Storage", delivery.NotStored, args["artifact"]) - case tool != "build_image_build": // a bundle or an archive is pushed and its reference answered (ADR 0498 clause 1) - d := digestOf(fmt.Sprint(args["artifact"])) - step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"digest":"`+d+`","reference":"artifact-store://`+filepath.Base(fmt.Sprint(args["module"]))+`/`+fmt.Sprint(args["artifact"])+`/blobs/`+d+`"}`) - case tool == "build_image_build" && args["publish"] == true: // a trunk image is pushed: the digest and the reference - d := digestOf(fmt.Sprint(args["artifact"])) - step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"digest":"`+d+`","reference":"registry.example/imaged/server@`+d+`"}`) - default: - step.State, step.Result = delivery.StepPassed, json.RawMessage(`{"digest":"`+digestOf(fmt.Sprint(args["artifact"]))+`"}`) - } - put(step) - }() - }) - if err != nil { + language := strings.TrimPrefix(module, "mesh-build-") + if err := buildstep.DeclareAskWorkers(f.ctx, f.js, language, 30*time.Second); err != nil { t.Fatal(err) } - t.Cleanup(func() { _ = sub.Unsubscribe() }) + put := func(key string, value any) (uint64, error) { + raw, _ := value.(json.RawMessage) + return kv.Put(f.ctx, key, raw) + } + call := func(name string) buildstep.StepFunc { + return func(ctx context.Context, args map[string]any) (any, error) { return f.step(module, name, args) } + } + steps := buildstep.AskSteps{Fetch: call("fetch"), Build: call("build"), Copy: call("copy")} + for _, node := range nodes { + p := &buildstep.Puller{Source: &buildstep.JetStreamAsks{JS: f.js, Language: language}, Language: language, Node: node, Put: put, + Capacity: func() int { return 2 }, PullWait: 200 * time.Millisecond, + Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return buildstep.RunAsk(ctx, steps, a) }} + go func() { _ = p.Pull(f.ctx) }() + } } type bed struct { @@ -154,8 +146,6 @@ type bed struct { candidates jetstream.KeyValue builds jetstream.KeyValue tools map[string]jetstream.KeyValue - presence jetstream.KeyValue - have jetstream.KeyValue instances jetstream.KeyValue stop func() // stops the builder started last fake *fakeBuildModules @@ -228,20 +218,11 @@ func startBed(t *testing.T) *bed { b.candidates = mk(delivery.CandidatesBucket, delivery.CandidatesHistory) 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) b.instances = mk(instancesBucket, 1) - for _, node := range []string{"n1", "n2"} { - 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{}} - b.fake.serve(t, ctl, "mesh-build-go") - b.fake.serve(t, ctl, "mesh-build-node") - b.fake.serve(t, ctl, "mesh-build-image") - b.setHealth(t, "n1", "mesh-build-image", delivery.HealthOK) // the image build module runs on n1 only + b.fake.serve(t, "mesh-build-go", "n1", "n2") + b.fake.serve(t, "mesh-build-node", "n1", "n2") + b.fake.serve(t, "mesh-build-image", "n1") // the image build module runs on n1 only b.report(t, "mesh-build-go", "n1", "go", "go1.27.1", "1") b.report(t, "mesh-build-go", "n2", "go", "go1.27.1", "1") b.report(t, "mesh-build-node", "n1", "typescript", "tsc5", "1") @@ -249,28 +230,6 @@ 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() @@ -433,20 +392,8 @@ func TestAnArchiveOnlyModuleNamingNoToolchainIsBuilt(t *testing.T) { } } -// 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 n1") { - t.Errorf("build = %+v, want it built on n1 alone", v) - } -} - -// TestAFailedKeyIsBuiltAgainAsANewRevisionFromANewModulesValueWithTheSameFingerprint: a build step fails on one node, so the -// key is failed naming it; the indexer writing modules. again with the same fingerprint wakes the builder, which builds +// TestAFailedKeyIsBuiltAgainAsANewRevisionFromANewModulesValueWithTheSameFingerprint: a build step fails, so the key is +// failed naming it; the indexer writing modules. again with the same fingerprint wakes the builder, which builds // the key again as a new revision of the same key, now done. func TestAFailedKeyIsBuiltAgainAsANewRevisionFromANewModulesValueWithTheSameFingerprint(t *testing.T) { // 1607/AC6, AC7 b := startBed(t) @@ -742,3 +689,32 @@ func TestABuilderWhoseBuildModulesAnsweredReferencesSaysItAndStoppedSaysNothing( t.Fatalf("the stopped builder still says: %v", err) } } + +// 1887/AC1, AC2 on the real bus: the builder puts the trunk fingerprint's build as asks on the build seats' delivery +// streams and calls no node's tool; the build modules that pulled them wrote the steps, and the key is done naming the +// references. +func TestTheBuilderAsksTheBuildSeatsAndCallsNoNode(t *testing.T) { + b := startBed(t) + b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck") + sub, err := nats.Connect(b.srvURL, nats.UserInfo("ctl", "pw")) + if err != nil { + t.Fatal(err) + } + defer sub.Close() + var calls atomic.Int64 + s, err := sub.Subscribe("mesh.mod.*.call.>", func(*nats.Msg) { calls.Add(1) }) + if err != nil { + t.Fatal(err) + } + defer s.Unsubscribe() + b.start(t, "n1", nil) + v := b.await(t, "netcheck", isState(delivery.BuildDone)) + for _, a := range v.Artifacts { + if a.Reference == "" { + t.Errorf("artifact %s names no reference", a.Name) + } + } + if n := calls.Load(); n != 0 { + t.Errorf("%d tool calls reached a build module", n) + } +} 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 822b325..6d597a3 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -26,7 +26,6 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-builder/internal/build" ) @@ -143,38 +142,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - // Calls to the build modules go out as this module's own account, naming the machine (mesh-sdk process.CallOn); the - // buckets are read and written on the connection above. - account, err := process.ReadAccount(path) - if err != nil { - return err - } - caller, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) - if err != nil { - return err - } - defer caller.Close() - bound := defaultStepBound if v := strings.TrimSpace(getenv("MESH_BUILDER_STEP_BOUND")); v != "" { if bound, err = time.ParseDuration(v); err != nil { return fmt.Errorf("MESH_BUILDER_STEP_BOUND: %w", err) } } - capacity, err := capacityFrom(strings.TrimSpace(getenv("MESH_BUILDER_SETTINGS")), getenv("MESH_BUILDER_CAPACITY")) - if err != nil { - return fmt.Errorf("the build capacity setting: %w", err) - } tools := &buildrun.ToolchainSet{} - presence := &buildrun.PresenceSet{} - have := &buildrun.HaveSet{} + // Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot + // (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity. + asks := buildrun.JetStreamAsks{JS: js} b := &build.Builder{ Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module}, - Runner: buildrun.Runner{Caller: caller, Steps: stepsWaiter{js}, Bound: bound}, + Asks: buildrun.Asker{Publisher: asks, Waiter: asks, Bound: bound}, Toolchains: tools, - Presence: presence, - Health: have, - Places: &buildrun.Places{Capacity: capacity}, Manifests: manifestReader{manifests}, Logf: logger.Printf, } @@ -219,48 +200,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer cc.Stop() } - // The watchman's presence says which nodes are asleep. Where its bucket is not there yet no node is online, so nothing - // builds; the builder says so and is started again with the watchman. - presenceErrs := make(chan error, 1) - switch kv, err := js.KeyValue(ctx, delivery.PresenceBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: no node counts as online until the builder starts again", delivery.PresenceBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", delivery.PresenceBucket, err) - default: - pw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { presence.Put(e.Key(), e.Value(), !opOf(e)) }) - if err != nil { - return err - } - go func() { - presenceErrs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) { - if presence.Put(e.Key(), e.Value(), !opOf(e)) { - b.NodesChanged(ctx) // a node came online or went: a build that waited for one may go on - } - }) - }() - } - // 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.NodesChanged(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()) @@ -300,9 +239,5 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err case err := <-watch.errs: return err - case err := <-presenceErrs: - return err - case err := <-haveErrs: - return err } } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/settings.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/settings.go deleted file mode 100644 index e0255f1..0000000 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/settings.go +++ /dev/null @@ -1,28 +0,0 @@ -package main - -import ( - "encoding/json" - "fmt" - "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" - "os" -) - -// capacityFrom is the build capacity of each node (ADR 0478 clause 3): the module's setting `capacity`, `=` pairs separated by commas, laid into the settings file the controller composes (issue 1661); empty is one step at a -// time on every node. With no settings file named, fallback is read instead, for a run outside the module's launch. -func capacityFrom(path, fallback string) (map[string]int, error) { - if path == "" { - return buildrun.ParseCapacity(fallback) - } - raw, err := os.ReadFile(path) - if err != nil { - return nil, fmt.Errorf("the settings file: %w", err) - } - var s struct { - Capacity string `json:"capacity"` - } - if err := json.Unmarshal(raw, &s); err != nil { - return nil, fmt.Errorf("the settings file %s is not JSON: %w", path, err) - } - return buildrun.ParseCapacity(s.Capacity) -} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/settings_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/settings_test.go deleted file mode 100644 index 31b1acf..0000000 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/settings_test.go +++ /dev/null @@ -1,67 +0,0 @@ -package main - -import ( - "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" - "os" - "path/filepath" - "testing" -) - -func settingsFile(t *testing.T, body string) string { - t.Helper() - p := filepath.Join(t.TempDir(), "settings.json") - if err := os.WriteFile(p, []byte(body), 0o600); err != nil { - t.Fatal(err) - } - return p -} - -// 1661: the build capacity is the module's setting `capacity`, laid into the settings file: empty is one step at a time everywhere. -func TestCapacityIsReadFromTheSettingsFile(t *testing.T) { - for body, want := range map[string]map[string]int{ - `{"capacity": "shanks=3,ace=2"}`: {"shanks": 3, "ace": 2}, - `{"capacity": ""}`: {}, - `{}`: {}, - } { - got, err := capacityFrom(settingsFile(t, body), "") - if err != nil || len(got) != len(want) || got["shanks"] != want["shanks"] || got["ace"] != want["ace"] { - t.Errorf("%s: capacity %v, %v, want %v", body, got, err, want) - } - } - p := &buildrun.Places{Capacity: mustCapacity(t, settingsFile(t, `{"capacity": ""}`))} - if p.CapacityOf("shanks") != 1 || p.CapacityOf("ace") != 1 { - t.Errorf("an empty setting gives %d and %d, want 1 everywhere", p.CapacityOf("shanks"), p.CapacityOf("ace")) - } -} - -func mustCapacity(t *testing.T, path string) map[string]int { - t.Helper() - c, err := capacityFrom(path, "") - if err != nil { - t.Fatal(err) - } - return c -} - -func TestACapacitySettingThatIsNotOneIsRefusedNamingIt(t *testing.T) { - if _, err := capacityFrom(settingsFile(t, `{"capacity": "shanks=-1"}`), ""); err == nil { - t.Error("shanks=-1 accepted") - } - if c, err := capacityFrom(settingsFile(t, `{"capacity": "shanks=3,ace=2,g14=0,novox=0"}`), ""); err != nil || c["g14"] != 0 || len(c) != 4 { - t.Errorf("the starting value: %v, %v", c, err) - } - if _, err := capacityFrom(settingsFile(t, `not json`), ""); err == nil { - t.Error("a file that is not JSON accepted") - } - if _, err := capacityFrom(filepath.Join(t.TempDir(), "none"), ""); err == nil { - t.Error("a settings file that is not there accepted") - } -} - -// With no settings file named (a run outside the module's launch, as the tests do), the capacity is the environment's. -func TestWithNoSettingsFileTheCapacityIsTheEnvironments(t *testing.T) { - got, err := capacityFrom("", "ace=2") - if err != nil || got["ace"] != 2 { - t.Errorf("capacity %v, %v", got, err) - } -} diff --git a/modules/mesh-delivery-builder/files/settings.json b/modules/mesh-delivery-builder/files/settings.json deleted file mode 100644 index 3175fe3..0000000 --- a/modules/mesh-delivery-builder/files/settings.json +++ /dev/null @@ -1,3 +0,0 @@ -{ - "capacity": "shanks=3,ace=2,g14=0,novox=0" -} diff --git a/modules/mesh-delivery-builder/internal/build/fixtures_test.go b/modules/mesh-delivery-builder/internal/build/fixtures_test.go index b5a6566..4ecc3e0 100644 --- a/modules/mesh-delivery-builder/internal/build/fixtures_test.go +++ b/modules/mesh-delivery-builder/internal/build/fixtures_test.go @@ -1,10 +1,14 @@ package build import ( + "context" "encoding/json" "os" + "strings" "testing" + "time" + "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -25,17 +29,31 @@ func captured(t *testing.T, name string) json.RawMessage { return step.Result } +// The captured image steps, run as a pulled ask runs them (buildstep.RunAsk), answer what the builder reads: the trunk's +// build its digest and reference, a candidate's its digest and no reference (issue 1887 A4a). func TestCapturedImageStepsDecodeWithTheBuildersOwnTypes(t *testing.T) { // review r2, mesh-catalog 2; ADR 0487 clause 1 - var f fetchResult - if err := json.Unmarshal(captured(t, "fetch.json"), &f); err != nil || f.Workspace == "" { - t.Errorf("fetch = %+v, %v, want a workspace (the field the builder reads)", f, err) - } - var trunk buildResult - if err := json.Unmarshal(captured(t, "build-trunk.json"), &trunk); err != nil || trunk.Digest == "" || trunk.Reference == "" { - t.Errorf("trunk build = %+v, %v, want a digest and a reference", trunk, err) - } - var cand buildResult - if err := json.Unmarshal(captured(t, "build-candidate.json"), &cand); err != nil || cand.Digest == "" || cand.Reference != "" { - t.Errorf("candidate build = %+v, %v, want a digest and no reference", cand, err) + for _, tc := range []struct { + build string + publish bool + ref bool + }{{"build-trunk.json", true, true}, {"build-candidate.json", false, false}} { + steps := buildstep.AskSteps{ + Fetch: func(context.Context, map[string]any) (any, error) { + return json.RawMessage(captured(t, "fetch.json")), nil + }, + Build: func(context.Context, map[string]any) (any, error) { return json.RawMessage(captured(t, tc.build)), nil }, + } + ask := delivery.StepAsk{ID: buildstep.NewID(), Kind: delivery.AskDelivery, Language: delivery.AskImage, Asker: delivery.BuilderModule, + At: time.Now(), Build: &delivery.BuildJob{Module: "m", Path: "modules/m", Publish: tc.publish, + Fetch: map[string]any{"repository": "novox/mesh-catalog", "commit": strings.Repeat("a", 40)}, + Artifacts: []delivery.AskArtifact{{Artifact: "server"}}}} + out, err := buildstep.RunAsk(context.Background(), steps, ask) + if err != nil { + t.Fatalf("%s: %v", tc.build, err) + } + a := out.(delivery.BuildAnswer) + if len(a.Artifacts) != 1 || a.Artifacts[0].Digest == "" || (a.Artifacts[0].Reference != "") != tc.ref { + t.Errorf("%s answered %+v, want a digest and a reference only for the trunk", tc.build, a) + } } } diff --git a/modules/mesh-delivery-builder/internal/build/r4e_test.go b/modules/mesh-delivery-builder/internal/build/r4e_test.go index d3d851c..fab0854 100644 --- a/modules/mesh-delivery-builder/internal/build/r4e_test.go +++ b/modules/mesh-delivery-builder/internal/build/r4e_test.go @@ -111,7 +111,7 @@ func TestADoneKeyWrittenBeforeR4eIsBuiltAgainOnceAsANewRevision(t *testing.T) { func TestAPinnedUpstreamIsCopiedOnceOnOneImageNodeWithNoFetchAndItsKeyNamesTheReference(t *testing.T) { // B4, ADR 0498 clause 4 up := catalogue.Artifact{Name: "up", Kind: "upstream", From: "nats:2.11.17-alpine@" + digest} r := newRig(t, up) - r.b.Health = imageHave(t, "ace", "novox") + r.puller = "ace" v := r.consider(t, "modules/m") if v.State != delivery.BuildDone || len(v.Artifacts) != 1 { t.Fatalf("build = %+v, want done with the copy", v) @@ -133,7 +133,7 @@ func TestAPinnedUpstreamIsCopiedOnceOnOneImageNodeWithNoFetchAndItsKeyNamesTheRe func TestAnUpstreamBesideAnImageIsCopiedAndTheImageFetchedAndBuiltOnTheSameNode(t *testing.T) { // B4 r := newRig(t, imageArt, catalogue.Artifact{Name: "up", Kind: "upstream", From: "nats:2.11.17-alpine@" + digest}) - r.b.Health = imageHave(t, "ace") + r.puller = "ace" v := r.consider(t, "modules/m") if ok, missing := v.Referenced(); !ok { t.Fatalf("build = %+v, missing references %v", v, missing) @@ -145,7 +145,7 @@ func TestAnUpstreamBesideAnImageIsCopiedAndTheImageFetchedAndBuiltOnTheSameNode( func TestACopyThatFailsNotStoredFailsTheKeyAndIsBuiltAgain(t *testing.T) { // B4, ADR 0498 clause 2 r := newRig(t, catalogue.Artifact{Name: "up", Kind: "upstream", From: "nats:2.11.17-alpine@" + digest}) - r.b.Health = imageHave(t, "ace") + r.puller = "ace" r.bm.fails["ace/build_image_copy"] = delivery.NotStored + "up: the registry answered 500" v := r.consider(t, "modules/m") if v.State != delivery.BuildFailed || !strings.HasPrefix(v.Error, delivery.NotStored) || delivery.FailedUntilChanged(v.Error) { diff --git a/modules/mesh-delivery-builder/internal/build/round4_test.go b/modules/mesh-delivery-builder/internal/build/round4_test.go index f2bcc51..e396831 100644 --- a/modules/mesh-delivery-builder/internal/build/round4_test.go +++ b/modules/mesh-delivery-builder/internal/build/round4_test.go @@ -3,7 +3,6 @@ package build import ( "context" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" - "strings" "testing" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -39,26 +38,6 @@ func TestAnArchiveOnlyModuleIsBuiltAlthoughItsFingerprintNamesNoToolchain(t *tes } } -func TestAnArchiveOnlyModuleNeedsNoAgreedToolchainButNeedsNodesThatQualify(t *testing.T) { // review round 4, 1 - r := newRig(t, archiveA) - r.tools.Put(buildrun.GoBuildModule, "ace.go", tcValue(t, "go", "go1.26.0"), false) // the nodes disagree: a bundle would wait - r.b.Health = &buildrun.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 := &buildrun.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.Contains(v.Step, "on novox") { - t.Errorf("build = %+v, want done on novox, the one node that qualifies", 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(buildrun.GoBuildModule) { @@ -80,115 +59,6 @@ func TestClaimLeavesAKeyFailedForDifferingDigestsAlone(t *testing.T) { // review } } -// 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 := &buildrun.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 := buildrun.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 := &buildrun.Places{} - nodes, _, _ := buildrun.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() - buildrun.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 buildrun.HaveSet - if s.Healthy("ace", buildrun.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", buildrun.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", buildrun.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", buildrun.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", buildrun.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 := &buildrun.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 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() diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index 3021dd0..64e3bcd 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -5,9 +5,11 @@ import ( "encoding/json" "errors" "fmt" + "maps" "sort" "strings" "sync" + "time" "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -20,29 +22,23 @@ type ToolchainEntries interface { Entries(language string) []delivery.NodeToolchain } -// toolPrefix is the start of each build module's tool names: fetch and build. -var toolPrefix = map[string]string{buildrun.GoBuildModule: "build_go_", buildrun.NodeBuildModule: "build_node_", ImageBuildModule: "build_image_"} - // Builder builds, dark, every new-form fingerprint the indexer writes (novox/hq ADR 0453 clause 4, ADR 0478): it is // handed each modules value as the watch of the indexer's bucket sees it, and for each module with a fingerprint and an -// artifact to build claims the fingerprint's build key, builds it through the build modules on the nodes -// the toolchains choose, and writes the key done or failed. It holds no timer; a build starts from a modules value or a +// artifact to build claims the fingerprint's build key, builds it by asking the build seats (one ask per build module, +// pulled by whichever build node has a free slot: ADR 0525), and writes the key done or failed. It holds no timer; a build starts from a modules value or a // change of the toolchains. type Builder struct { - Claimer Claimer - Runner buildrun.Runner - Toolchains ToolchainEntries - // 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 + Claimer Claimer + // Asks puts each build module's part of a build as one ask on its build seat's stream and waits for its step (novox/hq + // ADR 0525 §1, §3, §6; issue 1887 A4): the builder chooses no node, whichever build node pulls the ask builds it. + Asks interface { + Put(ctx context.Context, a delivery.StepAsk) error + Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) } - // buildrun.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 *buildrun.Places - // Limit is how many builds run at once; four when 0. + Toolchains ToolchainEntries + Manifests Manifests + // Limit is how many builds the builder has asked at once; their asks wait in the build seats' streams, which are the + // backlog (ADR 0525 §9), not here. 256 when 0. Limit int Logf func(format string, args ...any) // Answered is told the build module whose step answered a reference (issue 1886): what the builder says it writes @@ -75,15 +71,12 @@ func (b *Builder) init() { b.once.Do(func() { limit := b.Limit if limit <= 0 { - limit = 4 + limit = 256 } b.sem = make(chan struct{}, limit) b.cands, b.via, b.trunkFP = map[string]delivery.Candidate{}, map[string]via{}, map[string]bool{} 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{} b.published = map[string]bool{} - if b.Places == nil { - b.Places = &buildrun.Places{} - } }) } @@ -116,8 +109,7 @@ func (b *Builder) Module(ctx context.Context, key string, value []byte) { b.consider(ctx, key, m) } -// NodesChanged is Reconsider after a change of the nodes: one came online or went, a build module became healthy or -// stopped being, or a toolchain moved, so a build that waited for a node may go on. +// NodesChanged is Reconsider after a toolchain moved, so a build that waited for the nodes to agree on one may go on. func (b *Builder) NodesChanged(ctx context.Context) { b.Reconsider(ctx) } // Sweep reads, at start, every key of the bucket: it keeps the package versions done keys name as published (a key @@ -314,7 +306,7 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) { 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, ImageBuilder: b.imageBuilderRuns, Trunk: trunk, Published: b.isPublished}) + cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished}) if err != nil { b.logf("%s: %v", module, err) return @@ -330,7 +322,7 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) { if cl.Build.State == delivery.BuildDone { prior, only = &cl.Build, map[string]bool{} if trunk { - for _, bm := range publishable(cl.Build, b.imageBuilderRuns) { + for _, bm := range publishable(cl.Build, imageSeat) { only[bm] = true publishing = true } @@ -342,7 +334,7 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) { // any other, so the controller has a done key to register it from (issue 1613, ADR 0478 clause 1, ADR 0501 clause 1). result = delivery.BuilderBuild{State: delivery.BuildDone, Step: "nothing to build: the manifest names no artifact"} } else { - result = b.build(ctx, module, m, plan, planErr, prior, only) + result = b.build(ctx, module, m, plan, planErr, prior, only, b.notes(cl)) } if ctx.Err() != nil { return @@ -399,14 +391,15 @@ func failed(format string, args ...any) delivery.BuilderBuild { // lane is the build of a plan's tasks of one build module on one node. type lane struct { - module, node string + module, node string // node is the one that pulled the lane's ask tasks []Task + toolchain *delivery.ToolchainUsed digests map[string]string // artifact -> digest references map[string]string // artifact -> the reference an image was pushed to (a trunk build only) err error } -func (b *Builder) build(ctx context.Context, module string, m delivery.Module, plan Plan, planErr error, prior *delivery.BuilderBuild, only map[string]bool) delivery.BuilderBuild { +func (b *Builder) build(ctx context.Context, module string, m delivery.Module, plan Plan, planErr error, prior *delivery.BuilderBuild, only map[string]bool, asked *askNotes) delivery.BuilderBuild { switch { case planErr != nil: return failed("%s", planErr) @@ -417,76 +410,39 @@ 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 - // 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) - } - } var kept []string // build modules whose artifacts stay as the prior done key has them for _, bm := range plan.Modules() { if prior != nil && len(only) > 0 && !only[bm] { kept = append(kept, bm) continue } - language := buildrun.LanguageOf[bm] - entries := b.Toolchains.Entries(language) - qualifies := func(node string) bool { return b.qualifies(node, bm) } - var nodes []string - if bm == ImageBuildModule { - // An image is built on one node, chosen from the nodes whose image build module runs healthy (ADR 0487 clause 8): - // the module has no toolchain language, so no toolchain entry names its nodes, and a second build would only - // push the image again. - ns, err := buildrun.ChooseAnyNodes("image", b.imageEntries(), b.Places, qualifies, 1) - if err != nil { - giveUpLanes() - return failed("%s no node has %s, so the image cannot be built: the provision rootless-image-build is held by the module buildah; assign buildah and %s to a node (%v)", NoImageBuilderPrefix, ImageBuildModule, ImageBuildModule, err) - } - nodes = ns - } else if plan.NeedsToolchain(bm) { - ns, tc, err := buildrun.ChooseNodes(language, entries, b.Places, qualifies, 1) - if err != nil { - giveUpLanes() - return failed("%s", err) + var toolchain *delivery.ToolchainUsed + if bm != ImageBuildModule && plan.NeedsToolchain(bm) { + // The ask names the toolchain the fingerprint was computed with; a node building with another refuses it. + language := buildrun.LanguageOf[bm] + tc, none := delivery.AgreedToolchain(b.Toolchains.Entries(language)) + if none != "" { + return failed("the %s toolchain is not agreed: %s", language, none) } 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 := buildrun.ChooseAnyNodes(language, entries, b.Places, qualifies, 1) - if err != nil { - giveUpLanes() - return failed("%s", err) - } - nodes = ns + u := delivery.ToolchainUsed{Version: tc.Version, Recipe: tc.Recipe} + used[language], toolchain = u, &u } - // One node builds each fingerprint (ADR 0478 clause 4, from R4e). - steps = append(steps, fmt.Sprintf("%s on %s", bm, nodes[0])) var tasks []Task for _, t := range plan.Tasks { if t.Module == bm { tasks = append(tasks, t) } } - for _, n := range nodes { - lanes = append(lanes, &lane{module: bm, node: n, tasks: tasks, digests: map[string]string{}, references: map[string]string{}}) - } + // One ask builds each fingerprint's part of a build module, on whichever node pulls it (ADR 0525 §3). + lanes = append(lanes, &lane{module: bm, tasks: tasks, toolchain: toolchain, digests: map[string]string{}, references: map[string]string{}}) } var wg sync.WaitGroup for _, l := range lanes { wg.Add(1) - go func() { defer wg.Done(); b.runLane(ctx, l, repository, m, module) }() + go func() { defer wg.Done(); b.runLane(ctx, l, repository, m, module, asked) }() } wg.Wait() var errs []string @@ -548,6 +504,10 @@ func (b *Builder) build(ctx context.Context, module string, m delivery.Module, p } artifacts = append(artifacts, built) } + var steps []string + for _, l := range lanes { + steps = append(steps, fmt.Sprintf("%s on %s", l.module, l.node)) + } for _, bm := range kept { steps = append(steps, bm+" kept") for lang, u := range prior.Toolchains { @@ -617,73 +577,56 @@ func keyError(errs []string) string { return strings.Join(errs, "; ") } -func (b *Builder) online(node string) bool { return b.Presence == nil || b.Presence.Online(node) } +// imageSeat says an image build is asked of the image build seat, which takes it when a node pulls it: a key failed for +// want of an image builder before R4g is taken again (ADR 0525 §1). +func imageSeat() bool { return true } -// 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.Places.CapacityOf(node) > 0 && b.online(node) && (b.Health == nil || b.Health.Healthy(node, bm)) +// askLanguage is the build seat language of each build module (delivery.AskSubject). +var askLanguage = map[string]string{buildrun.GoBuildModule: delivery.AskGo, buildrun.NodeBuildModule: delivery.AskNode, ImageBuildModule: delivery.AskImage} + +// askNotes writes a claimed key's asks into it, building, before each is put (ADR 0525 §1; issue 1887 A4b), and holds +// the asks a key taken over at start already names, so they are waited for and never put again. +type askNotes struct { + mu sync.Mutex + b *Builder + cl *Claim + resumed map[string]string } -// imageEntries are the nodes an image may be built on, as toolchain entries for the chooser: every node the builder has a -// capacity for or a build module toolchain from. Whether the image build module runs there is qualifies', from the have. -func (b *Builder) imageEntries() []delivery.NodeToolchain { - seen := map[string]bool{} - var entries []delivery.NodeToolchain - add := func(n string) { - if !seen[n] { - seen[n] = true - entries = append(entries, delivery.NodeToolchain{Node: n}) - } +func (b *Builder) notes(cl *Claim) *askNotes { + n := &askNotes{b: b, cl: cl} + if cl.Build.State == delivery.BuildBuilding && len(cl.Build.Asks) > 0 { + n.resumed = maps.Clone(cl.Build.Asks) } - for n := range b.Places.Capacity { - add(n) + return n +} + +// resume is the id of the ask a taken-over key names for language, or "". +func (n *askNotes) resume(language string) string { + n.mu.Lock() + defer n.mu.Unlock() + return n.resumed[language] +} + +// note writes the ask's id into the key, building, by compare-and-set at the claim's revision. +func (n *askNotes) note(ctx context.Context, language, id, step string) error { + n.mu.Lock() + defer n.mu.Unlock() + v := n.cl.Build + v.State, v.By, v.Step, v.Error, v.Artifacts = delivery.BuildBuilding, n.b.Claimer.By, step, "", nil + v.Asks = maps.Clone(v.Asks) + if v.Asks == nil { + v.Asks = map[string]string{} } - for _, language := range []string{"go", "typescript"} { - for _, e := range b.Toolchains.Entries(language) { - add(e.Node) - } - } - sort.Slice(entries, func(i, j int) bool { return entries[i].Node < entries[j].Node }) - return entries + v.Asks[language] = id + return n.b.Claimer.Finish(ctx, n.cl, v) } -// imageBuilderRuns is whether a node can build an image now. -func (b *Builder) imageBuilderRuns() bool { - for _, e := range b.imageEntries() { - if b.qualifies(e.Node, ImageBuildModule) { - return true - } - } - 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) -} - -// fetchResult and buildResult are the results of a build module's fetch and build steps as the builder reads them. -type fetchResult struct { - Workspace string `json:"workspace"` - Merge string `json:"merge"` - Conflicts []string `json:"conflicts"` -} - -type buildResult struct { - Digest string `json:"digest"` - Reference string `json:"reference"` -} - -// runLane takes a place on the node, fetches the commit there and builds each of the lane's artifacts, one step at a time. -// An upstream artifact is copied by its pinned reference with no fetch (ADR 0498 clause 4); a lane of copies alone fetches -// nothing. -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() +// runLane puts the lane's part of the build as one ask on its build seat's stream (the delivery stream for a trunk key, the +// check stream for a candidate's) and reads what the node that pulled it built (ADR 0525 §1, §3, §6). An upstream artifact +// is copied by its pinned reference with no fetch (ADR 0498 clause 4); a lane of copies alone fetches nothing. A lane of a +// key taken over at start waits for the ask the key names and puts none. +func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m delivery.Module, module string, asked *askNotes) { req := buildstep.FetchRequest{Repository: repository, Commit: m.Commit} if key, err := delivery.BuilderBuildKey(module, m.FP2); err == nil { b.mu.Lock() @@ -702,57 +645,69 @@ func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m del } sort.Strings(req.Paths) } - fetchArgs := req.Args() candidate := req.Base != "" - var fetched fetchResult + language := askLanguage[l.module] + kind := delivery.AskDelivery + if candidate { + kind = delivery.AskCheck + } + job := &delivery.BuildJob{Module: module, Path: m.Path, FP: m.FP2, Toolchain: l.toolchain, + // Only a trunk key's build pushes an image or publishes a package; a candidate's never does (ADR 0487 clause 12, + // ADR 0498 clauses 5 and 6). A bundle or an archive is pushed either way (clause 3). + Publish: !candidate} + for _, t := range l.tasks { + a := delivery.AskArtifact{Artifact: t.Artifact} + if t.Kind == catalogue.ArtifactUpstream { + a.From = t.From // the copy runs no code of a change: the module's name, the artifact and its pin + } + job.Artifacts = append(job.Artifacts, a) + } if needsFetch(l.tasks) { - raw, err := b.step(ctx, l.module, l.node, "fetch", fetchArgs) - if err != nil { + job.Fetch = req.Args() + } + ask := delivery.StepAsk{ID: buildstep.NewID(), Kind: kind, Language: language, Asker: delivery.BuilderModule, At: time.Now(), Build: job} + if id := asked.resume(language); id != "" { + ask.ID = id // put before this builder started again: the stream holds it, or a node already ran it + } else { + if err := asked.note(ctx, language, ask.ID, "asked "+l.module); err != nil { + l.err = fmt.Errorf("the ask of %s could not be written to the key: %w", l.module, err) + return + } + if err := b.Asks.Put(ctx, ask); err != nil { l.err = err return } - if err := json.Unmarshal(raw, &fetched); err != nil || fetched.Workspace == "" { - l.err = fmt.Errorf("the fetch on %s answered no workspace: %s", l.node, raw) - return - } - // A candidate is built from its head merged onto the trunk. A head that does not merge ends the fetch passed with the tree at - // the trunk, and building that would store the trunk's build under the candidate's fingerprint. - if candidate { - if len(fetched.Conflicts) > 0 { - l.err = fmt.Errorf("the head does not merge into the trunk on %s: files in conflict: %s", l.node, strings.Join(fetched.Conflicts, ", ")) - return - } - if fetched.Merge == "" { - l.err = fmt.Errorf("the fetch with a base on %s answered no merge: %s", l.node, raw) - return - } - } + } + step, err := b.Asks.Wait(ctx, ask) + if err != nil { + l.err = fmt.Errorf("%s: %w", l.module, err) + return + } + l.node = step.Node + if step.State != delivery.StepPassed { + l.err = fmt.Errorf("%s on %s: the ask %s %s: %s", l.module, step.Node, ask.ID, step.State, step.Reason) + return + } + var answer delivery.BuildAnswer + if err := json.Unmarshal(step.Result, &answer); err != nil { + l.err = fmt.Errorf("the ask of %s on %s answered what is not a build: %s", l.module, step.Node, step.Result) + return + } + built := map[string]delivery.AskBuilt{} + for _, a := range answer.Artifacts { + built[a.Artifact] = a } for _, t := range l.tasks { - tool, args := "build", map[string]any{"workspace": fetched.Workspace, "module": m.Path, "artifact": t.Artifact} - if t.Kind == catalogue.ArtifactUpstream { - // The copy runs no code of a change: the module's name (the reference's path), the artifact and its pin. - tool, args = "copy", map[string]any{"module": module, "artifact": t.Artifact, "from": t.From} - } else if !candidate { - // Only a trunk key's build pushes an image or publishes a package; a candidate's never does (ADR 0487 clause 12, - // ADR 0498 clauses 5 and 6). A bundle or an archive is pushed either way (clause 3). - args["publish"] = true - } - raw, err := b.step(ctx, l.module, l.node, tool, args) - if err != nil { - l.err = err + a, ok := built[t.Artifact] + if !ok || a.Digest == "" { + l.err = fmt.Errorf("the build of %s of %s on %s answered no digest: %s", t.Artifact, module, step.Node, step.Result) return } - var built buildResult - if err := json.Unmarshal(raw, &built); err != nil || built.Digest == "" { - l.err = fmt.Errorf("the %s of %s of %s on %s answered no digest: %s", tool, t.Artifact, module, l.node, raw) - return + l.digests[t.Artifact] = a.Digest + if a.Reference != "" && (stored(t.Kind) || !candidate) { + l.references[t.Artifact] = a.Reference } - l.digests[t.Artifact] = built.Digest - if built.Reference != "" && (stored(t.Kind) || !candidate) { - l.references[t.Artifact] = built.Reference - } - if built.Reference != "" && b.Answered != nil { + if a.Reference != "" && b.Answered != nil { b.Answered(l.module) } } diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index 4a8524c..ba94b0e 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -10,6 +10,7 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -124,22 +125,56 @@ type rig struct { st *memStore bm *buildModules tools *buildrun.ToolchainSet - pres *buildrun.PresenceSet mf *fakeManifests key string + // asks are the build seats' streams; puller is the node that pulls every ask ("ace" when empty). + asks *buildrun.FakeAsks + puller string +} + +// toolPrefix is the start of each build module's tool names, as the fake build modules answer them. +var toolPrefix = map[string]string{buildrun.GoBuildModule: "build_go_", buildrun.NodeBuildModule: "build_node_", ImageBuildModule: "build_image_"} + +// answer runs an ask as the build module that pulls it does (buildstep.RunAsk), with the fake build modules' tools on the +// rig's pulling node, and answers the step that node writes. +func (r *rig) answer(a delivery.StepAsk) delivery.Step { + node := r.puller + if node == "" { + node = "ace" + } + module, _ := delivery.BuildModuleOf(a.Language) + runner := buildrun.Runner{Caller: r.bm, Steps: r.bm.steps, Bound: time.Second} + tool := func(name string) buildstep.StepFunc { + return func(ctx context.Context, args map[string]any) (any, error) { + raw, err := runner.Run(ctx, module, node, toolPrefix[module]+name, args) + if err != nil { + return nil, err + } + return raw, nil + } + } + steps := buildstep.AskSteps{Fetch: tool("fetch"), Build: tool("build"), Copy: tool("copy")} + out, err := buildstep.RunAsk(context.Background(), steps, a) + step := delivery.Step{ID: a.ID, Tool: a.Kind + " ask", Node: node, State: delivery.StepPassed} + if err != nil { + step.State, step.Reason = delivery.StepFailed, err.Error() + return step + } + step.Result, _ = json.Marshal(out) + return step } func newRig(t *testing.T, artifacts ...catalogue.Artifact) *rig { t.Helper() - r := &rig{st: newMem(), bm: newBuildModules(), tools: &buildrun.ToolchainSet{}, pres: &buildrun.PresenceSet{}, mf: &fakeManifests{byDigest: map[string]catalogue.Manifest{mdigest('a'): {Module: "m", Build: &catalogue.Build{Artifacts: artifacts}}}}} + r := &rig{st: newMem(), bm: newBuildModules(), tools: &buildrun.ToolchainSet{}, mf: &fakeManifests{byDigest: map[string]catalogue.Manifest{mdigest('a'): {Module: "m", Build: &catalogue.Build{Artifacts: artifacts}}}}} for _, n := range []string{"ace", "novox"} { r.tools.Put(buildrun.GoBuildModule, n+".go", tcValue(t, "go", "go1.27.1"), false) r.tools.Put(buildrun.NodeBuildModule, n+".typescript", tcValue(t, "typescript", "tsc5"), false) - r.pres.Put(n, presenceValue(t, true), false) } + r.asks = &buildrun.FakeAsks{Answer: r.answer} r.b = &Builder{ - Claimer: Claimer{Store: r.st, By: "novox.builder"}, Runner: buildrun.Runner{Caller: r.bm, Steps: r.bm.steps, Bound: time.Second}, - Toolchains: r.tools, Presence: r.pres, Manifests: r.mf, Logf: t.Logf, + Claimer: Claimer{Store: r.st, By: "novox.builder"}, Asks: buildrun.Asker{Publisher: r.asks, Waiter: r.asks, Bound: 2 * time.Second}, + Toolchains: r.tools, Manifests: r.mf, Logf: t.Logf, } var err error if r.key, err = delivery.BuilderBuildKey("m", fp1); err != nil { @@ -220,15 +255,53 @@ func TestABundleAndAnArchiveAreBuiltOnOneNodeAndDone(t *testing.T) { // 1607/AC7 } } -func TestOneQualifyingNodeBuildsOnce(t *testing.T) { // 1607/AC4, ADR 0478 clause 4 +// 1887/AC1, AC2: a trunk fingerprint with no key is one ask on the Go seat's delivery stream, named in the building key +// before it is put, and the builder calls no node; the step a node wrote for it makes the key done naming its reference. +func TestATrunkFingerprintIsOneAskOnTheDeliveryStream(t *testing.T) { r := newRig(t, goBundle) - r.tools.Put(buildrun.GoBuildModule, "ace.go", nil, true) - v := r.consider(t, "modules/m") - if v.State != delivery.BuildDone || !strings.Contains(v.Step, "on novox") { - t.Errorf("build = %+v, want done with its step saying novox", v) + r.asks.Answer = nil // no node has pulled it yet + r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) + var asks []delivery.StepAsk + for deadline := time.Now().Add(2 * time.Second); len(asks) == 0; time.Sleep(time.Millisecond) { + if time.Now().After(deadline) { + t.Fatal("no ask was put") + } + asks = r.asks.Asks() } - if n := len(r.bm.callsOf("build_go_build")); n != 1 { - t.Errorf("%d builds, want one", n) + a := asks[0] + if len(asks) != 1 || a.Kind != delivery.AskDelivery || a.Language != delivery.AskGo || a.Build == nil || a.Build.FP != fp1 { + t.Fatalf("asks = %+v, want one Go delivery ask of the fingerprint", asks) + } + if v := valueOf(t, r.st, r.key); v.State != delivery.BuildBuilding || v.Asks[delivery.AskGo] != a.ID { + t.Fatalf("the key = %+v, want building naming the ask %s", v, a.ID) + } + if n := len(r.bm.callsOf("build_go_build")) + len(r.bm.callsOf("build_go_fetch")); n != 0 { + t.Fatalf("the builder called a node %d times", n) + } + r.asks.End(r.answer(a)) // a node pulled it and ran it + r.b.Wait() + if v := valueOf(t, r.st, r.key); v.State != delivery.BuildDone || !strings.Contains(v.Step, "on ace") || v.Artifacts[0].Reference == "" || len(v.Asks) != 0 { + t.Fatalf("the key = %+v, want done on ace naming the reference", v) + } +} + +// 1887/AC3: a builder that starts again while its ask waits puts no second ask: it waits for the step of the ask the +// building key names. +func TestARestartedBuilderWaitsForTheAskTheKeyNames(t *testing.T) { + r := newRig(t, goBundle) + const id = "mvl0frv18d1957" + data, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.builder", Manifest: manifestD, + Step: "asked mesh-build-go", Asks: map[string]string{delivery.AskGo: id}}) + r.st.vals[r.key] = data + r.asks.Answer = nil + r.asks.End(delivery.Step{ID: id, Node: "shanks", State: delivery.StepPassed, + Result: []byte(`{"artifacts":[{"artifact":"tool","digest":"` + digest + `","reference":"artifact-store://m/tool/blobs/` + digest + `"}]}`)}) + v := r.consider(t, "modules/m") + if n := len(r.asks.Asks()); n != 0 { + t.Fatalf("%d asks put again", n) + } + if v.State != delivery.BuildDone || !strings.Contains(v.Step, "on shanks") { + t.Fatalf("the key = %+v, want done from the ask's step on shanks", v) } } @@ -247,20 +320,9 @@ func TestAFailingStepFailsTheBuildWithItsReasonAndNodeAndIsTriedAgainAsANewRevis } } -// imageHave marks mesh-build-image running and healthy on these nodes. -func imageHave(t *testing.T, nodes ...string) *buildrun.HaveSet { - t.Helper() - have := &buildrun.HaveSet{} - for _, n := range nodes { - have.Put(n+".mesh-build-go", haveValue(t, delivery.PhaseRunning, delivery.HealthOK), false) - have.Put(n+"."+ImageBuildModule, haveValue(t, delivery.PhaseRunning, delivery.HealthOK), false) - } - return have -} - func TestAnImageGoesToOneNodeAndGetsOneDoneKeyWithItsReference(t *testing.T) { // 1705a, ADR 0487 clause 5 r := newRig(t, imageArt) - r.b.Health = imageHave(t, "ace", "novox") // two nodes qualify; an image still builds on one + r.puller = "ace" // two nodes qualify; an image still builds on one v := r.consider(t, "modules/m") if v.State != delivery.BuildDone || len(v.Artifacts) != 1 || v.Artifacts[0].Kind != "image" || v.Artifacts[0].Digest != digest { t.Fatalf("build = %+v, want done with one image artifact and its digest", v) @@ -278,19 +340,9 @@ func TestAnImageGoesToOneNodeAndGetsOneDoneKeyWithItsReference(t *testing.T) { / } } -func TestAnImageIsBuiltOnlyWhereMeshBuildImageIsHealthy(t *testing.T) { // 1705a - r := newRig(t, imageArt) - r.b.Health = imageHave(t, "novox") - r.consider(t, "modules/m") - c := r.bm.callsOf("build_image_build") - if len(c) != 1 || c[0].node != "novox" { - t.Errorf("calls = %+v, want one on novox", c) - } -} - func TestABundleAndAnImageOfOneModuleBuildOnOneNodeEach(t *testing.T) { // 1705a r := newRig(t, goBundle, imageArt) - r.b.Health = imageHave(t, "ace", "novox") + r.puller = "ace" v := r.consider(t, "modules/m") if v.State != delivery.BuildDone || len(v.Artifacts) != 2 { t.Fatalf("build = %+v", v) @@ -303,33 +355,9 @@ func TestABundleAndAnImageOfOneModuleBuildOnOneNodeEach(t *testing.T) { // 1705a } } -func TestWithNoImageNodeTheKeyFailsNamingTheProvisionAndIsNotRetakenUntilANodeReportsTheModule(t *testing.T) { // 1705b, lead's decision - r := newRig(t, imageArt) - r.b.Health = imageHave(t) // mesh-build-image runs nowhere - v := r.consider(t, "modules/m") - if v.State != delivery.BuildFailed || !strings.HasPrefix(v.Error, NoImageBuilderPrefix) || !strings.Contains(v.Error, "rootless-image-build") || !strings.Contains(v.Error, "buildah") || !strings.Contains(v.Error, ImageBuildModule) { - t.Fatalf("build = %+v, want failed naming rootless-image-build, buildah and mesh-build-image", v) - } - if len(r.bm.calls) != 0 { - t.Errorf("%d calls, want none", len(r.bm.calls)) - } - writes := r.st.writes - r.consider(t, "modules/m") - r.consider(t, "modules/m") - if r.st.writes != writes { - t.Errorf("the failed key was written again (%d writes, was %d)", r.st.writes, writes) - } - // A node reports the module: the key is taken again. - r.b.Health = imageHave(t, "ace") - v = r.consider(t, "modules/m") - if v.State != delivery.BuildDone { - t.Errorf("build = %+v, want done once a node reports mesh-build-image", v) - } -} - func TestACandidatesImageCallCarriesNoPublishAndItsKeyNoReference(t *testing.T) { // 1705b, lead's decision r := newRig(t, imageArt) - r.b.Health = imageHave(t, "ace") + r.puller = "ace" trunkModule(t, r) r.b.Candidate(context.Background(), "feat%2Fx", candidateValue(t, "feat/x", 7, headX)) r.b.Wait() @@ -348,7 +376,7 @@ func TestACandidatesImageCallCarriesNoPublishAndItsKeyNoReference(t *testing.T) func TestATrunkImageCallCarriesPublish(t *testing.T) { // 1705b r := newRig(t, imageArt) - r.b.Health = imageHave(t, "ace") + r.puller = "ace" r.consider(t, "modules/m") calls := r.bm.callsOf("build_image_build") if len(calls) != 1 || calls[0].body.(map[string]any)["publish"] != true { @@ -439,7 +467,7 @@ func TestAKeyTheBuilderLeftBuildingIsBuiltAtStartAndDoneKeysAreNotBuiltTwice(t * // 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, Logf: t.Logf} + other := &Builder{Claimer: Claimer{Store: r.st, By: "ace.builder"}, Asks: r.b.Asks, Toolchains: r.tools, Manifests: r.mf, Logf: t.Logf} value := moduleValue(t, "modules/m") var wg sync.WaitGroup for _, b := range []*Builder{r.b, other} { @@ -522,36 +550,9 @@ func TestARootModulesPathIsDotAndIsGivenToTheBuildTools(t *testing.T) { // 1615/ } } -func TestANodeThatIsNotOnlineIsNotChosenAndComingBackWakesTheBuild(t *testing.T) { // 1607 review 3 - r := newRig(t, goBundle) - r.tools.Put(buildrun.GoBuildModule, "g14.go", tcValue(t, "go", "go1.27.1"), false) // a laptop that reports its toolchain and sleeps - v := r.consider(t, "modules/m") - if v.State != delivery.BuildDone { - t.Fatalf("build = %+v", v) - } - for _, c := range r.bm.calls { - if c.node == "g14" { - t.Errorf("a call went to g14, which is not online: %+v", c) - } - } - // With no node online the key fails naming why; a node coming online builds it as a new revision. - r2 := newRig(t, goBundle) - r2.pres.Put("ace", nil, true) - r2.pres.Put("novox", nil, true) - if v := r2.consider(t, "modules/m"); v.State != delivery.BuildFailed || !strings.Contains(v.Error, "online") { - t.Fatalf("build = %+v, want failed because no node is online", v) - } - r2.pres.Put("ace", presenceValue(t, true), false) - r2.b.Reconsider(context.Background()) - r2.b.Wait() - if v := valueOf(t, r2.st, r2.key); v.State != delivery.BuildDone { - t.Errorf("build = %+v, want done once a node is online", v) - } -} - func TestACandidatesImageIsPublishedOnceWhenTheTrunkNamesItsFingerprint(t *testing.T) { // review r1, mesh-delivery 1; ADR 0487 clause 5 r := newRig(t, imageArt) - r.b.Health = imageHave(t, "ace") + r.puller = "ace" trunkModule(t, r) // the trunk carries another fingerprint; fp1 is the candidate's r.b.Candidate(context.Background(), "feat%2Fx", candidateValue(t, "feat/x", 7, headX)) r.b.Wait() @@ -590,7 +591,7 @@ func TestACandidatesImageIsPublishedOnceWhenTheTrunkNamesItsFingerprint(t *testi func TestAFailedPublishOfACandidatesImageIsWrittenToTheKey(t *testing.T) { // review r2, mesh-delivery 1; ADR 0487 clause 5 r := newRig(t, imageArt) - r.b.Health = imageHave(t, "ace") + r.puller = "ace" trunkModule(t, r) r.b.Candidate(context.Background(), "feat%2Fx", candidateValue(t, "feat/x", 7, headX)) r.b.Wait() diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 6f13144..930b0d8 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -38,6 +38,7 @@ type manifest struct { Consumes []string `json:"consumes"` State []state `json:"state"` Reads []string `json:"reads"` + Uses []string `json:"uses"` Invokes []string `json:"invokes"` Config map[string]struct { Type string `json:"type"` @@ -86,12 +87,17 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if want := []state{{Name: "builds", History: 5}, {Name: "instances", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory t.Errorf("state = %+v, want %+v", m.State, want) } + // 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and + // holds no capacity; it puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5). if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) { + "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"}; !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-image.build_image_build", "mesh-build-image.build_image_copy", "mesh-build-image.build_image_fetch", "mesh-build-node.build_node_build", "mesh-build-node.build_node_fetch"}; !reflect.DeepEqual(sorted(m.Invokes), want) { - t.Errorf("invokes = %v, want only the build modules' fetch and build (go, image, node) and the image module's copy (ADR 0498 clause 4)", m.Invokes) + if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) { + t.Errorf("uses = %v, want the three build seats", m.Uses) + } + if len(m.Invokes) != 0 || len(m.Config) != 0 { + t.Errorf("invokes = %v, config = %v, want none: no tool call from the builder and no capacity", m.Invokes, m.Config) } // The builder is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it starts. if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported"}; !reflect.DeepEqual(sorted(m.Consumes), want) { @@ -103,7 +109,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if m.SecretsOwner != "mesh-delivery-builder" { t.Errorf("secrets-owner = %q", m.SecretsOwner) } - for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats", "bus"} { + for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "tools", "seat", "seats", "bus", "invokes", "config"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the builder has none", k) } @@ -129,35 +135,19 @@ func TestManifestProcess(t *testing.T) { // 1607/AC1 t.Errorf("process = %v", p) } env, _ := p["env"].(map[string]any) - // The build capacity of each node is the module's setting `capacity` (issue 1661, ADR 0478 clause 3), laid into the - // settings file the program reads; empty is one step at a time everywhere. The environment holds no capacity. - if len(env) != 2 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" || env["MESH_BUILDER_SETTINGS"] != "${dir:state}/settings.json" { - t.Errorf("env = %v", p["env"]) + if len(env) != 1 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" { + t.Errorf("env = %v, want the broker file alone", p["env"]) } - if !reflect.DeepEqual(p["restart-on"], []any{"settings"}) { - t.Errorf("restart-on = %v, want the settings file, so a changed capacity reaches the running program", p["restart-on"]) - } - c, ok := m.Config["capacity"] - if !ok || len(m.Config) != 1 || c.Type != "string" || c.Default != "shanks=3,ace=2,g14=0,novox=0" || c.Description == "" { - t.Errorf("config = %+v, want one string setting, capacity, shanks=3,ace=2,g14=0,novox=0 to start", m.Config) - } - var settings map[string]any for _, r := range m.Resources { if r["id"] == "settings" { - settings = r + t.Errorf("a settings resource %v: the builder holds no setting since the capacity is the build modules' (ADR 0525 §5)", r) } } - if settings == nil || settings["type"] != "file" || settings["merge"] != "json" || !reflect.DeepEqual(settings["takes-config"], []any{"capacity"}) || settings["path"] != "${dir:state}/settings.json" { - t.Errorf("settings resource = %v", settings) - } - if raw, err := os.ReadFile("files/settings.json"); err != nil || !strings.Contains(string(raw), `"capacity": "shanks=3,ace=2,g14=0,novox=0"`) { - t.Errorf("files/settings.json = %q, %v, want the setting's starting value", raw, err) - } if _, bus := p["bus"]; bus { t.Errorf("the process declares a bus: it runs on its own broker credential") } - if len(m.Build.Artifacts) != 2 || m.Build.Artifacts[1].Name != "files" || m.Build.Artifacts[1].Kind != "archive" || m.Build.Artifacts[1].From != "files" { - t.Fatalf("artifacts = %+v, want the program and the files", m.Build.Artifacts) + if len(m.Build.Artifacts) != 1 { + t.Fatalf("artifacts = %+v, want the program alone", m.Build.Artifacts) } a := m.Build.Artifacts[0] if st, err := os.Stat(a.From); err != nil || !st.IsDir() || a.Binary != "mesh-delivery-builder" || !strings.HasSuffix(a.From, a.Binary) || a.Kind != "bundle" || a.Language != "go" { diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index f18e708..798cb79 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -18,27 +18,14 @@ "mesh-build-node.toolchains", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", - "mesh-delivery-indexer.modules", - "mesh-delivery-watchman.presence", - "node-engine.have" + "mesh-delivery-indexer.modules" ], - "invokes": [ - "mesh-build-go.build_go_build", - "mesh-build-go.build_go_fetch", - "mesh-build-image.build_image_build", - "mesh-build-image.build_image_copy", - "mesh-build-image.build_image_fetch", - "mesh-build-node.build_node_build", - "mesh-build-node.build_node_fetch" + "uses": [ + "node-build-go", + "node-build-image", + "node-build-node" ], - "config": { - "capacity": { - "type": "string", - "description": "How many build steps the builder runs at once on each node, as node=steps pairs separated by commas, such as shanks=3,ace=2 (ADR 0478 clause 3); a node set to 0 never builds, a node not named runs one at a time, and empty is one everywhere. The builder builds on shanks and ace only (the operator, 2026-10-21).", - "pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*", - "default": "shanks=3,ace=2,g14=0,novox=0" - } - }, + "own-secrets": { "broker": "${dir:state}/broker" }, @@ -59,17 +46,6 @@ "place": ".", "unknown": "program" }, - { - "id": "settings", - "trusted": false, - "type": "file", - "path": "${dir:state}/settings.json", - "mode": "0600", - "owner": "mesh-delivery-builder", - "takes-config": ["capacity"], - "merge": "json", - "template": "files/settings.json" - }, { "id": "builder", "type": "process", @@ -79,12 +55,8 @@ "./mesh-delivery-builder" ], "env": { - "MESH_BROKER_FILE": "${dir:state}/broker", - "MESH_BUILDER_SETTINGS": "${dir:state}/settings.json" + "MESH_BROKER_FILE": "${dir:state}/broker" }, - "restart-on": [ - "settings" - ], "user": "mesh-delivery-builder", "health": { "kind": "unit" @@ -100,11 +72,6 @@ "system": "arch", "from": "cmd/mesh-delivery-builder", "binary": "mesh-delivery-builder" - }, - { - "name": "files", - "kind": "archive", - "from": "files" } ] }