From 46c9f33d649b0bc2e2f91f634003a4767f1aecef Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 11:23:39 +0200 Subject: [PATCH] Build each new-form fingerprint through the build modules on two nodes and write its key done or failed (issue 1607, AC7, AC8) --- .../cmd/mesh-delivery-builder/main.go | 105 +++++- .../internal/build/claim_test.go | 8 + .../internal/build/run.go | 324 ++++++++++++++++++ .../internal/build/run_test.go | 295 ++++++++++++++++ .../internal/build/steps_test.go | 11 +- 5 files changed, 737 insertions(+), 6 deletions(-) create mode 100644 modules/mesh-delivery-builder/internal/build/run.go create mode 100644 modules/mesh-delivery-builder/internal/build/run_test.go 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 c023e2e..c85cfe4 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -19,11 +19,23 @@ import ( "os/signal" "strings" "syscall" + "time" "github.com/nats-io/nats.go" "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" +) + +const ( + // defaultStepBound is how long the builder waits for one step of a build module (a fetch downloads the Go modules, a + // build compiles). + defaultStepBound = 30 * time.Minute + // nodesPerFingerprint is how many build nodes build each fingerprint until release R4e (ADR 0478 clause 4). + nodesPerFingerprint = 2 ) func main() { @@ -109,10 +121,95 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - if _, err := js.KeyValue(ctx, delivery.BuilderBuildsBucket); err != nil { - return fmt.Errorf("the bucket %s: %w", delivery.BuilderBuildsBucket, err) + bucket := func(name string) (jetstream.KeyValue, error) { + kv, err := js.KeyValue(ctx, name) + if err != nil { + return nil, fmt.Errorf("the bucket %s: %w", name, err) + } + return kv, nil + } + builds, err := bucket(delivery.BuilderBuildsBucket) + if err != nil { + return err + } + modules, err := bucket(delivery.ModulesBucket) + if err != nil { + return err + } + manifests, err := bucket(delivery.ManifestsBucket) + 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) + } + } + tools := &build.ToolchainSet{} + 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, + Manifests: manifestReader{manifests}, + Nodes: nodesPerFingerprint, + Logf: logger.Printf, + } + // The toolchains first, so a build that starts from the first modules value finds the nodes. + watch := newToolchainWatch(ctx, js, tools, b.Reconsider, logger) + if err := watch.start(); err != nil { + return err + } + // A build module says each time it wrote its toolchains key: its stream, and the builder's durable consumer on it, are + // made by the controller. Where the stream is not there yet the builder says so and is started again with the build + // module. + for _, module := range []string{build.GoBuildModule, build.NodeBuildModule} { + stream := toolchainStreams[module] + cons, err := js.Consumer(ctx, stream, cred.Node+"_"+cred.Module) + if err != nil { + logger.Printf("the consumer %s_%s on %s does not exist yet, so %s's reports are not heard until the builder starts again: %v", cred.Node, cred.Module, stream, module, err) + continue + } + cc, err := cons.Consume(func(m jetstream.Msg) { watch.Reported(m) }, + jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + if err != nil { + return err + } + defer cc.Stop() + } + each := func(e jetstream.KeyValueEntry) { + if opOf(e) { + b.Module(ctx, e.Key(), e.Value()) + } else { + b.Forget(e.Key()) + } + } + w, err := settle(ctx, modules, each) + if err != nil { + return err } logger.Printf("started on %s", cred.Node) - <-ctx.Done() - return nil + errs := make(chan error, 1) + go func() { errs <- follow(ctx, w, each) }() + select { + case <-ctx.Done(): + b.Wait() + return nil + case err := <-errs: + return err + case err := <-watch.errs: + return err + } } diff --git a/modules/mesh-delivery-builder/internal/build/claim_test.go b/modules/mesh-delivery-builder/internal/build/claim_test.go index c6a8e36..7589524 100644 --- a/modules/mesh-delivery-builder/internal/build/claim_test.go +++ b/modules/mesh-delivery-builder/internal/build/claim_test.go @@ -3,6 +3,7 @@ package build import ( "context" "errors" + "sync" "testing" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -10,6 +11,7 @@ import ( // memStore is a bucket's create and compare-and-set in memory, with a revision per write. type memStore struct { + mu sync.Mutex vals map[string][]byte revs map[string]uint64 next uint64 @@ -21,6 +23,8 @@ type memStore struct { func newMem() *memStore { return &memStore{vals: map[string][]byte{}, revs: map[string]uint64{}} } func (m *memStore) Get(_ context.Context, key string) ([]byte, uint64, bool, error) { + m.mu.Lock() + defer m.mu.Unlock() v, ok := m.vals[key] return v, m.revs[key], ok, nil } @@ -40,6 +44,8 @@ func (m *memStore) put(key string, value []byte) uint64 { } func (m *memStore) Create(_ context.Context, key string, value []byte) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() m.hook() if _, ok := m.vals[key]; ok { return 0, ErrConflict @@ -48,6 +54,8 @@ func (m *memStore) Create(_ context.Context, key string, value []byte) (uint64, } func (m *memStore) Update(_ context.Context, key string, value []byte, rev uint64) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() m.hook() if _, ok := m.vals[key]; !ok || m.revs[key] != rev { return 0, ErrConflict diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go new file mode 100644 index 0000000..8e6c35a --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -0,0 +1,324 @@ +package build + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ToolchainEntries are the nodes' toolchain entries of a language (ToolchainSet). +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{GoBuildModule: "build_go_", NodeBuildModule: "build_node_"} + +// 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 +// change of the toolchains. +type Builder struct { + Claimer Claimer + Runner Runner + Toolchains ToolchainEntries + Manifests Manifests + // Nodes is how many build nodes build each fingerprint: two until release R4e (ADR 0478 clause 4). + Nodes int + // Limit is how many builds run at once; four when 0. + Limit int + Logf func(format string, args ...any) + + once sync.Once + sem chan struct{} + wg sync.WaitGroup + mu sync.Mutex + known map[string]delivery.Module // by modules key + inflight map[string]bool // by build key + running map[string]int // steps in flight, by `/` + plans map[string]planned // by manifest digest +} + +type planned struct { + plan Plan + err error +} + +func (b *Builder) init() { + b.once.Do(func() { + limit := b.Limit + if limit <= 0 { + limit = 4 + } + b.sem = make(chan struct{}, limit) + b.known, b.inflight, b.running, b.plans = map[string]delivery.Module{}, map[string]bool{}, map[string]int{}, map[string]planned{} + }) +} + +func (b *Builder) logf(format string, args ...any) { + if b.Logf != nil { + b.Logf(format, args...) + } +} + +// Wait returns when every build started so far has ended. +func (b *Builder) Wait() { b.wg.Wait() } + +// Module takes one value of the indexer's modules bucket in (the key is the module's name) and builds it if it is +// to be built. A value that is not a module is said and left. +func (b *Builder) Module(ctx context.Context, key string, value []byte) { + b.init() + m, err := delivery.DecodeModule(value) + if err != nil { + b.logf("modules.%s holds what is not a module, left: %v", key, err) + return + } + b.mu.Lock() + b.known[key] = m + b.mu.Unlock() + b.consider(ctx, key, m) +} + +// Forget takes a module out (its key was deleted). +func (b *Builder) Forget(key string) { + b.init() + b.mu.Lock() + delete(b.known, key) + b.mu.Unlock() +} + +// Reconsider looks at every module again: the toolchains changed, so a build that failed for them may build now. +func (b *Builder) Reconsider(ctx context.Context) { + b.init() + b.mu.Lock() + known := make(map[string]delivery.Module, len(b.known)) + for k, m := range b.known { + known[k] = m + } + b.mu.Unlock() + for _, k := range sortedKeys(known) { + b.consider(ctx, k, known[k]) + } +} + +func sortedKeys(m map[string]delivery.Module) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} + +// planOf is the plan of the manifest with this digest, read once: a manifest by digest never changes. A manifest that +// cannot be read is not kept, so it is read again when the module is. +func (b *Builder) planOf(ctx context.Context, digest string) (Plan, error, bool) { + b.mu.Lock() + p, ok := b.plans[digest] + b.mu.Unlock() + if ok { + return p.plan, p.err, true + } + m, err := b.Manifests.Manifest(ctx, digest) + if err != nil { + return Plan{}, fmt.Errorf("its manifest %s cannot be read: %w", digest, err), false + } + plan, err := PlanOf(m) + b.mu.Lock() + b.plans[digest] = planned{plan, err} + b.mu.Unlock() + return plan, err, true +} + +func (b *Builder) consider(ctx context.Context, module string, m delivery.Module) { + if m.FP2 == "" || m.Manifest == "" { + return + } + key, err := delivery.BuilderBuildKey(module, m.FP2) + if err != nil { + b.logf("modules.%s: %v", module, err) + return + } + b.mu.Lock() + if b.inflight[key] { + b.mu.Unlock() + return + } + b.inflight[key] = true + b.mu.Unlock() + b.wg.Add(1) + go func() { + defer b.wg.Done() + defer func() { b.mu.Lock(); delete(b.inflight, key); b.mu.Unlock() }() + b.sem <- struct{}{} + defer func() { <-b.sem }() + b.try(ctx, module, m) + }() +} + +// try claims the fingerprint and builds it, writing the key done or failed. A build that ends because ctx did is left +// building by this builder: it takes it again when it starts (ADR 0478 clause 5). +func (b *Builder) try(ctx context.Context, module string, m delivery.Module) { + plan, planErr, read := b.planOf(ctx, m.Manifest) + if read && planErr == nil && len(plan.Tasks) == 0 { + return // nothing in the manifest for a builder to build + } + cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest) + if err != nil { + b.logf("%s: %v", module, err) + return + } + if cl == nil { + return + } + result := b.build(ctx, module, m, plan, planErr) + if ctx.Err() != nil { + return + } + result.By, result.Manifest = b.Claimer.By, m.Manifest + if err := b.Claimer.Finish(ctx, cl, result); err != nil { + b.logf("%s: the build could not be written: %v", cl.Key, err) + return + } + if result.State == delivery.BuildFailed { + b.logf("%s: failed: %s", cl.Key, result.Error) + } +} + +func failed(format string, args ...any) delivery.BuilderBuild { + return delivery.BuilderBuild{State: delivery.BuildFailed, Error: fmt.Sprintf(format, args...)} +} + +// lane is the build of a plan's tasks of one build module on one node. +type lane struct { + module, node string + tasks []Task + digests map[string]string // artifact -> digest + err error +} + +func (b *Builder) build(ctx context.Context, module string, m delivery.Module, plan Plan, planErr error) delivery.BuilderBuild { + switch { + case planErr != nil: + return failed("%s", planErr) + case m.Path == "": + return failed("the module's key names no directory (path), so the build tools cannot be told where %s lives", module) + } + repository := delivery.RepositoryName(m.Repository) + var lanes []*lane + used := map[string]delivery.ToolchainUsed{} + var steps []string + for _, bm := range plan.Modules() { + language := LanguageOf[bm] + nodes, tc, err := ChooseNodes(language, b.Toolchains.Entries(language), b.runningOn(bm), b.Nodes) + if err != nil { + return failed("%s", err) + } + used[language] = delivery.ToolchainUsed{Version: tc.Version, Recipe: tc.Recipe} + if len(nodes) == 1 { + steps = append(steps, fmt.Sprintf("%s on one node, %s", bm, nodes[0])) + } else { + sorted := append([]string(nil), nodes...) + sort.Strings(sorted) + steps = append(steps, fmt.Sprintf("%s on %s", bm, strings.Join(sorted, " and "))) + } + 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{}}) + } + } + var wg sync.WaitGroup + for _, l := range lanes { + wg.Add(1) + go func() { defer wg.Done(); b.runLane(ctx, l, repository, m, module) }() + } + wg.Wait() + var errs []string + for _, l := range lanes { + if l.err != nil { + errs = append(errs, l.err.Error()) + } + } + if len(errs) > 0 { + return failed("%s", strings.Join(errs, "; ")) + } + var artifacts []delivery.BuiltArtifact + for _, t := range plan.Tasks { + byNode := map[string]string{} + for _, l := range lanes { + if d, ok := l.digests[t.Artifact]; ok && l.module == t.Module { + byNode[l.node] = d + } + } + d, err := Agree(t.Artifact, byNode) + if err != nil { + return failed("%s", err) + } + 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} +} + +func (b *Builder) runningOn(buildModule string) map[string]int { + b.mu.Lock() + defer b.mu.Unlock() + out := map[string]int{} + for k, n := range b.running { + if mod, node, _ := strings.Cut(k, "/"); mod == buildModule { + out[node] = n + } + } + return out +} + +// step runs one step of a build module, counting it as running on the node while it does. +func (b *Builder) step(ctx context.Context, bm, node, tool string, args map[string]any) (json.RawMessage, error) { + k := bm + "/" + node + b.mu.Lock() + b.running[k]++ + b.mu.Unlock() + defer func() { b.mu.Lock(); b.running[k]--; b.mu.Unlock() }() + 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. +func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m delivery.Module, module string) { + raw, err := b.step(ctx, l.module, l.node, "fetch", map[string]any{"repository": repository, "commit": m.Commit}) + if err != nil { + l.err = err + return + } + var fetched struct { + Workspace string `json:"workspace"` + } + 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 + } + for _, t := range l.tasks { + raw, err := b.step(ctx, l.module, l.node, "build", map[string]any{"workspace": fetched.Workspace, "module": m.Path, "artifact": t.Artifact}) + if err != nil { + l.err = err + return + } + var built struct { + Digest string `json:"digest"` + } + if err := json.Unmarshal(raw, &built); err != nil || built.Digest == "" { + l.err = fmt.Errorf("the build of %s of %s on %s answered no digest: %s", t.Artifact, module, l.node, raw) + return + } + l.digests[t.Artifact] = built.Digest + } +} diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go new file mode 100644 index 0000000..fb71751 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -0,0 +1,295 @@ +package build + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// buildModules simulates the two build modules on the nodes: a long tool answers a step and key at once and writes the +// step's end into the steps (as mesh-build-go and mesh-build-node do), with a digest by node and artifact, or a failure. +type buildModules struct { + mu sync.Mutex + n int + calls []call + steps *fakeSteps + digests map[string]string // `/` -> digest; the default is `digest` + fails map[string]string // `/` -> the reason the step fails with +} + +func newBuildModules() *buildModules { + return &buildModules{steps: &fakeSteps{steps: map[string]delivery.Step{}}, digests: map[string]string{}, fails: map[string]string{}} +} + +func (b *buildModules) CallOn(node, key string, body any) (json.RawMessage, error) { + b.mu.Lock() + defer b.mu.Unlock() + b.calls = append(b.calls, call{node, key, body}) + b.n++ + id := fmt.Sprintf("s%d", b.n) + module, tool, _ := strings.Cut(key, ".") + args, _ := body.(map[string]any) + step := delivery.Step{ID: id, Tool: tool, Node: node, State: delivery.StepPassed} + if reason, ok := b.fails[node+"/"+tool]; ok { + step.State, step.Reason = delivery.StepFailed, reason + } else if strings.HasSuffix(tool, "_fetch") { + step.Result = json.RawMessage(`{"workspace":"w-` + node + `"}`) + } else { + d := digest + if v, ok := b.digests[node+"/"+fmt.Sprint(args["artifact"])]; ok { + d = v + } + step.Result = json.RawMessage(`{"digest":"` + d + `"}`) + } + sk, _ := delivery.StepKey(node, id) + b.steps.mu.Lock() + b.steps.steps[module+"/"+sk] = step + b.steps.mu.Unlock() + return json.RawMessage(`{"step":"` + id + `","key":"` + sk + `","state":"running"}`), nil +} + +func (b *buildModules) callsOf(tool string) (out []call) { + b.mu.Lock() + defer b.mu.Unlock() + for _, c := range b.calls { + if strings.HasSuffix(c.key, "."+tool) { + out = append(out, c) + } + } + return out +} + +type fakeManifests struct { + byDigest map[string]catalogue.Manifest +} + +func (f *fakeManifests) Manifest(_ context.Context, d string) (catalogue.Manifest, error) { + m, ok := f.byDigest[d] + if !ok { + return catalogue.Manifest{}, fmt.Errorf("manifests.%s is not there", d) + } + return m, nil +} + +func mdigest(c byte) string { return hex64(c) } + +type rig struct { + b *Builder + st *memStore + bm *buildModules + tools *ToolchainSet + mf *fakeManifests + key string +} + +func newRig(t *testing.T, artifacts ...catalogue.Artifact) *rig { + t.Helper() + r := &rig{st: newMem(), bm: newBuildModules(), tools: &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(GoBuildModule, n+".go", tcValue(t, "go", "go1.27.1"), false) + r.tools.Put(NodeBuildModule, n+".typescript", tcValue(t, "typescript", "tsc5"), false) + } + r.b = &Builder{ + Claimer: Claimer{Store: r.st, By: "novox.builder"}, Runner: Runner{Caller: r.bm, Steps: r.bm.steps, Bound: time.Second}, + Toolchains: r.tools, Manifests: r.mf, Nodes: 2, Logf: t.Logf, + } + var err error + if r.key, err = delivery.BuilderBuildKey("m", fp1); err != nil { + t.Fatal(err) + } + return r +} + +func moduleValue(t *testing.T, path string) []byte { + t.Helper() + data, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: "7a74cee83f88e6585a5511c3cc53bdf0f5303d20", FP: "src1:" + hex64('f'), FP2: fp1, Manifest: mdigest('a'), Path: path}) + if err != nil { + t.Fatal(err) + } + return data +} + +func (r *rig) consider(t *testing.T, path string) delivery.BuilderBuild { + t.Helper() + r.b.Module(context.Background(), "m", moduleValue(t, path)) + r.b.Wait() + if _, ok := r.st.vals[r.key]; !ok { + t.Fatal("no build key was written") + } + return valueOf(t, r.st, r.key) +} + +var ( + goBundle = catalogue.Artifact{Name: "tool", Kind: "bundle", Language: "go"} + tsBundle = catalogue.Artifact{Name: "web", Kind: "bundle", Language: "typescript"} + archiveA = catalogue.Artifact{Name: "files", Kind: "archive"} + imageArt = catalogue.Artifact{Name: "img", Kind: "image"} + manifestD = mdigest('a') +) + +func TestABundleAndAnArchiveAreBuiltOnTwoNodesAndDoneWhenTheDigestsAgree(t *testing.T) { // 1607/AC7, ADR 0478 clauses 1, 4 + r := newRig(t, goBundle, archiveA, tsBundle) + v := r.consider(t, "modules/m") + if v.State != delivery.BuildDone || v.By != "novox.builder" || v.Manifest != manifestD || len(v.Artifacts) != 3 { + t.Fatalf("build = %+v, want done by novox.builder with three artifacts", v) + } + for i, want := range []string{"files", "tool", "web"} { + if a := v.Artifacts[i]; a.Name != want || a.Digest != digest || a.Reference != "" { + t.Errorf("artifact %d = %+v, want %s with the digest and no reference (ADR 0478 clause 6)", i, a, want) + } + } + if v.Artifacts[0].Kind != "archive" || v.Artifacts[1].Kind != "bundle" { + t.Errorf("kinds = %+v", v.Artifacts) + } + if got := v.Toolchains["go"]; got.Version != "go1.27.1" || got.Recipe != "1" { + t.Errorf("toolchains = %+v, want go and typescript named", v.Toolchains) + } + if got := v.Toolchains["typescript"]; got.Version != "tsc5" { + t.Errorf("toolchains = %+v", v.Toolchains) + } + // Two fetches per build module, one on each node; every build names the module's directory. + if n := len(r.bm.callsOf("build_go_fetch")); n != 2 { + t.Errorf("%d go fetches, want one per node", n) + } + if n := len(r.bm.callsOf("build_go_build")); n != 4 { // tool and files, each on two nodes + t.Errorf("%d go builds, want 4", n) + } + if n := len(r.bm.callsOf("build_node_build")); n != 2 { + t.Errorf("%d node builds, want 2", n) + } + c := r.bm.callsOf("build_go_build")[0] + if args := c.body.(map[string]any); args["module"] != "modules/m" || !strings.HasPrefix(fmt.Sprint(args["workspace"]), "w-") { + t.Errorf("build args = %+v, want the module's path and the fetch's workspace", args) + } + f := r.bm.callsOf("build_go_fetch")[0].body.(map[string]any) + if f["repository"] != "novox/mesh-catalog" || f["commit"] != "7a74cee83f88e6585a5511c3cc53bdf0f5303d20" { + t.Errorf("fetch args = %+v", f) + } +} + +func TestDifferingDigestsFailTheBuildNamingBothNodes(t *testing.T) { // 1607/AC4, ADR 0478 clause 4 + r := newRig(t, goBundle) + r.bm.digests["ace/tool"] = "sha256:" + hex64('9') + v := r.consider(t, "modules/m") + if v.State != delivery.BuildFailed || !strings.Contains(v.Error, "ace") || !strings.Contains(v.Error, "novox") || !strings.Contains(v.Error, digest) || !strings.Contains(v.Error, hex64('9')) { + t.Errorf("build = %+v, want failed naming both nodes and both digests", v) + } +} + +func TestOneQualifyingNodeBuildsOnceAndTheKeySaysSo(t *testing.T) { // 1607/AC4, ADR 0478 clause 4 + r := newRig(t, goBundle) + r.tools.Put(GoBuildModule, "ace.go", nil, true) + v := r.consider(t, "modules/m") + if v.State != delivery.BuildDone || !strings.Contains(v.Step, "one node") || !strings.Contains(v.Step, "novox") { + t.Errorf("build = %+v, want done with its step saying one node, novox", v) + } + if n := len(r.bm.callsOf("build_go_build")); n != 1 { + t.Errorf("%d builds, want one", n) + } +} + +func TestAFailingStepFailsTheBuildWithItsReasonAndNodeAndIsTriedAgainAsANewRevision(t *testing.T) { // 1607/AC7, ADR 0429 clause 18 + r := newRig(t, goBundle) + r.bm.fails["ace/build_go_build"] = "go build of tool: undefined: x" + v := r.consider(t, "modules/m") + if v.State != delivery.BuildFailed || !strings.Contains(v.Error, "ace") || !strings.Contains(v.Error, "undefined: x") { + t.Fatalf("build = %+v, want failed with the node and the reason", v) + } + rev := r.st.revs[r.key] + delete(r.bm.fails, "ace/build_go_build") + v = r.consider(t, "modules/m") + if v.State != delivery.BuildDone || r.st.revs[r.key] <= rev { + t.Errorf("build = %+v at revision %d (was %d), want done as a new revision of the same key", v, r.st.revs[r.key], rev) + } +} + +func TestAModuleWithAnImageGetsARefusedKeyNeverBuiltAgain(t *testing.T) { // 1607/AC2, ADR 0478 clause 2 + r := newRig(t, archiveA, imageArt) + v := r.consider(t, "modules/m") + if v.State != delivery.BuildFailed || !strings.HasPrefix(v.Error, RefusedPrefix) || !strings.Contains(v.Error, "image") || !strings.Contains(v.Error, "build-agent") { + t.Fatalf("build = %+v, want failed 'refused kind: image' naming build-agent", v) + } + if len(r.bm.calls) != 0 { + t.Errorf("%d calls, want none for a refused kind", len(r.bm.calls)) + } + writes := r.st.writes + r.consider(t, "modules/m") + if r.st.writes != writes { + t.Errorf("a refused key was written again (%d writes, was %d)", r.st.writes, writes) + } +} + +func TestAModuleKeyWithNoPathFailsNamingThatReason(t *testing.T) { // 1607/AC8, ADR 0478 clause 8 + r := newRig(t, goBundle) + v := r.consider(t, "") + if v.State != delivery.BuildFailed || !strings.Contains(v.Error, "path") || len(r.bm.calls) != 0 { + t.Errorf("build = %+v with %d calls, want failed naming the missing path and nothing called", v, len(r.bm.calls)) + } +} + +func TestNoQualifyingNodeFailsTheBuildNamingTheLanguage(t *testing.T) { // 1607/AC4 + r := newRig(t, tsBundle) + r.tools.Put(NodeBuildModule, "ace.typescript", nil, true) + r.tools.Put(NodeBuildModule, "novox.typescript", nil, true) + v := r.consider(t, "modules/m") + if v.State != delivery.BuildFailed || !strings.Contains(v.Error, "typescript") || !strings.Contains(v.Error, delivery.ToolchainNoReport) { + t.Errorf("build = %+v, want failed naming typescript and that no toolchain is reported", v) + } +} + +func TestAnUnreadableManifestFailsTheBuild(t *testing.T) { + r := newRig(t, goBundle) + delete(r.mf.byDigest, mdigest('a')) + v := r.consider(t, "modules/m") + if v.State != delivery.BuildFailed || !strings.Contains(v.Error, "manifest") { + t.Errorf("build = %+v, want failed naming the manifest", v) + } +} + +func TestAModuleWithNoFingerprintOrNoArtifactIsNotBuilt(t *testing.T) { // 1607/AC2 + r := newRig(t) + r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) + data, _ := delivery.EncodeModule(delivery.Module{Repository: "novox/r", Commit: "c", FP: "src1:" + hex64('f'), None2: "toolchain not yet reported"}) + r.b.Module(context.Background(), "n", data) + r.b.Wait() + if len(r.st.vals) != 0 { + t.Errorf("keys %v were written: a module with no new-form fingerprint or no artifact has nothing to build", r.st.vals) + } +} + +func TestAKeyTheBuilderLeftBuildingIsBuiltAtStartAndDoneKeysAreNotBuiltTwice(t *testing.T) { // 1607/AC3, AC7 + r := newRig(t, goBundle) + seed(t, r.st, delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.builder", Manifest: manifestD}) + if v := r.consider(t, "modules/m"); v.State != delivery.BuildDone { + t.Fatalf("build = %+v, want the builder's own building key built", v) + } + calls := len(r.bm.calls) + r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) + r.b.Reconsider(context.Background()) + r.b.Wait() + if len(r.bm.calls) != calls { + t.Errorf("a done key was built again (%d calls, was %d)", len(r.bm.calls), calls) + } +} + +func TestTwoInstancesBuildOneFingerprintOnce(t *testing.T) { // 1607/AC7 + r := newRig(t, goBundle) + other := &Builder{Claimer: Claimer{Store: r.st, By: "ace.builder"}, Runner: r.b.Runner, Toolchains: r.tools, Manifests: r.mf, Nodes: 2, Logf: t.Logf} + value := moduleValue(t, "modules/m") + var wg sync.WaitGroup + for _, b := range []*Builder{r.b, other} { + wg.Add(1) + 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) + } +} diff --git a/modules/mesh-delivery-builder/internal/build/steps_test.go b/modules/mesh-delivery-builder/internal/build/steps_test.go index f6963ac..4a39fb9 100644 --- a/modules/mesh-delivery-builder/internal/build/steps_test.go +++ b/modules/mesh-delivery-builder/internal/build/steps_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "strings" + "sync" "testing" "time" @@ -32,10 +33,16 @@ func (f *fakeCaller) CallOn(node, key string, body any) (json.RawMessage, error) } // fakeSteps holds the steps values by `/`; Wait on a key it does not hold blocks until the context ends. -type fakeSteps struct{ steps map[string]delivery.Step } +type fakeSteps struct { + mu sync.Mutex + steps map[string]delivery.Step +} func (f *fakeSteps) Wait(ctx context.Context, module, key string) (delivery.Step, error) { - if s, ok := f.steps[module+"/"+key]; ok { + f.mu.Lock() + s, ok := f.steps[module+"/"+key] + f.mu.Unlock() + if ok { return s, nil } <-ctx.Done()