From 66908d9bbc1f20a188ae9a08ccba00f8fbdeee03 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 21:28:38 +0200 Subject: [PATCH] The builder builds the fingerprints a candidate carries, once each, from the head merged onto the trunk (issue 1684) --- .../mesh-delivery-builder/integration_test.go | 72 ++++++++++-- .../cmd/mesh-delivery-builder/main.go | 19 +++- .../internal/build/candidates.go | 104 ++++++++++++++++++ .../internal/build/candidates_test.go | 96 ++++++++++++++++ .../internal/build/run.go | 40 ++++++- .../mesh-delivery-builder/manifest_test.go | 2 +- modules/mesh-delivery-builder/module.json | 1 + 7 files changed, 316 insertions(+), 18 deletions(-) create mode 100644 modules/mesh-delivery-builder/internal/build/candidates.go create mode 100644 modules/mesh-delivery-builder/internal/build/candidates_test.go diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index 305ed79..926260a 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 @@ -56,6 +56,7 @@ type fakeBuildModules struct { failBuild atomic.Bool mu sync.Mutex calls map[string]int // by `.` + fetches []map[string]any } func (f *fakeBuildModules) count(tool string) int { @@ -83,6 +84,11 @@ func (f *fakeBuildModules) serve(t *testing.T, nc *nats.Conn, module string) { 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() + } id := fmt.Sprintf("s%d", f.n.Add(1)) key, _ := delivery.StepKey(node, id) put := func(s delivery.Step) { @@ -119,16 +125,17 @@ func (f *fakeBuildModules) serve(t *testing.T, nc *nats.Conn, module string) { } type bed struct { - ctx context.Context - js jetstream.JetStream - modules jetstream.KeyValue - builds jetstream.KeyValue - tools map[string]jetstream.KeyValue - presence jetstream.KeyValue - have jetstream.KeyValue - fake *fakeBuildModules - srvURL string - dir string + ctx context.Context + js jetstream.JetStream + modules jetstream.KeyValue + candidates jetstream.KeyValue + builds jetstream.KeyValue + tools map[string]jetstream.KeyValue + presence jetstream.KeyValue + have jetstream.KeyValue + fake *fakeBuildModules + srvURL string + dir string } // startBed starts the bus, the buckets the controller makes from the builder's and the build modules' manifests, the @@ -193,6 +200,7 @@ func startBed(t *testing.T) *bed { return kv } b.modules = mk(delivery.ModulesBucket, delivery.ModulesHistory) + 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) @@ -522,3 +530,47 @@ func TestAKeyLeftBuildingForAModuleThatIsGoneIsClosedAtStart(t *testing.T) { t.Errorf("build = %+v, want it closed naming who left it", v) } } + +// TestBuilderBuildsACandidatesFingerprintFromTheHeadMergedOntoTheTrunk: the trunk's record of netcheck carries a fingerprint of its +// own; a candidate the indexer wrote gives netcheck the fingerprint fp at a pull request's head; the program writes +// builds.netcheck. done, its fetch given the head, the ref that carries it and the trunk commit (issue 1684). +func TestBuilderBuildsACandidatesFingerprintFromTheHeadMergedOntoTheTrunk(t *testing.T) { // N7/AC1 + b := startBed(t) + other := "src2:" + strings.Repeat("e", 64) + digest := b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck") + data, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: commit, FP: "src1:" + strings.Repeat("f", 64), FP2: other, + Manifest: digest, Path: "modules/netcheck", Toolchains: map[string]delivery.ToolchainUsed{"go": {Version: "go1.27.1", Recipe: "1"}, "typescript": {Version: "tsc5", Recipe: "1"}}}) + if err != nil { + t.Fatal(err) + } + if _, err := b.modules.Put(b.ctx, "netcheck", data); err != nil { + t.Fatal(err) + } + const head = "1111111111111111111111111111111111111111" + pulls := []delivery.CandidatePull{{Head: delivery.Head{Repository: "novox/mesh-catalog", Number: 7, Head: head}, Modules: []string{"netcheck"}, + Fps: map[string]delivery.CandidateModule{"netcheck": {FP2: fp, Manifest: digest}}}} + hash, _ := delivery.HeadsHash([]delivery.Head{pulls[0].Head}) + cand, err := delivery.EncodeCandidate(delivery.Candidate{V: delivery.Version, Group: "feat/x", HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)}) + if err != nil { + t.Fatal(err) + } + if _, err := b.candidates.Put(b.ctx, "feat%2Fx", cand); err != nil { + t.Fatal(err) + } + b.start(t, "n1", nil) + v := b.await(t, "netcheck", isState(delivery.BuildDone)) + if len(v.Artifacts) != 2 { + t.Fatalf("build = %+v", v) + } + b.fake.mu.Lock() + defer b.fake.mu.Unlock() + found := false + for _, f := range b.fake.fetches { + if f["commit"] == head && f["ref"] == "refs/pull/7/head" && f["base"] == commit { + found = true + } + } + if !found { + t.Errorf("fetches = %v, want one of the candidate's head with its ref and the trunk commit", b.fake.fetches) + } +} 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 b5d2d34..f2b61e6 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -141,6 +141,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + candidates, err := bucket(delivery.CandidatesBucket) + 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) @@ -256,9 +260,22 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := b.Sweep(ctx); err != nil { return err } + // The fingerprints the indexer gives the candidates' modules are built too (issue 1684), once the trunk's modules are known. + eachCandidate := func(e jetstream.KeyValueEntry) { + if opOf(e) { + b.Candidate(ctx, e.Key(), e.Value()) + } else { + b.ForgetCandidate(e.Key()) + } + } + cw, err := settle(ctx, candidates, eachCandidate) + if err != nil { + return err + } logger.Printf("started on %s", cred.Node) - errs := make(chan error, 1) + errs := make(chan error, 2) go func() { errs <- follow(ctx, w, each) }() + go func() { errs <- follow(ctx, cw, eachCandidate) }() select { case <-ctx.Done(): b.Wait() diff --git a/modules/mesh-delivery-builder/internal/build/candidates.go b/modules/mesh-delivery-builder/internal/build/candidates.go new file mode 100644 index 0000000..93e7183 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/candidates.go @@ -0,0 +1,104 @@ +package build + +import ( + "context" + "fmt" + "sort" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/delivery/buildrun" +) + +// How a candidate's fingerprint is built (novox/hq ADR 0484 clause 3, issue 1684): the indexer writes, for each module a +// pull request touches, the fingerprint of the new form it has at the head merged onto the trunk (candidates.), and +// the builder builds each, once, as a key of its own (builds..) that the checker's line reads. The +// build is the trunk module's: the same build modules, the same nodes and capacity; what differs is where the source comes +// from: the fetch step is given the pull request's head, the ref that carries it and the trunk commit to merge it onto, as +// the checker's fetch is, and builds the merge's module directory. + +// via is how a build's fetch step reaches its source when it is a candidate's: the ref that carries the head and the trunk +// commit the head is merged onto. +type via struct{ Ref, Base string } + +// Candidate takes one value of the indexer's candidates bucket in (the key is the group) and builds the fingerprints it +// carries. A value that is not a candidate is said and left. +func (b *Builder) Candidate(ctx context.Context, key string, value []byte) { + b.init() + c, err := delivery.DecodeCandidate(value) + if err != nil { + b.logf("candidates.%s holds what is not a candidate, left: %v", key, err) + return + } + b.mu.Lock() + b.cands[key] = c + b.mu.Unlock() + b.considerCandidate(ctx, c) +} + +// ForgetCandidate takes a candidate out (its key was deleted). +func (b *Builder) ForgetCandidate(key string) { + b.init() + b.mu.Lock() + delete(b.cands, key) + b.mu.Unlock() +} + +func (b *Builder) considerCandidate(ctx context.Context, c delivery.Candidate) { + for _, p := range c.Pulls { + names := make([]string, 0, len(p.Fps)) + for name := range p.Fps { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + cm := p.Fps[name] + if cm.FP2 == "" || cm.Manifest == "" { + continue // no fingerprint of the new form for it (a reason in none2) or no manifest to build by + } + b.mu.Lock() + trunk, known := b.known[name] + b.mu.Unlock() + if !known || trunk.Path == "" { + b.logf("candidates.%s: the module %s has no record of the trunk yet, so its fingerprint %s is not built", c.Group, name, cm.FP2) + continue + } + key, err := delivery.BuilderBuildKey(name, cm.FP2) + if err != nil { + b.logf("candidates.%s: %v", c.Group, err) + continue + } + m := delivery.Module{Repository: p.Repository, Commit: p.Head.Head, FP2: cm.FP2, Manifest: cm.Manifest, Path: trunk.Path, + Toolchains: b.agreedFor(ctx, cm.Manifest)} + b.mu.Lock() + if _, onTrunk := b.trunkFP[key]; !onTrunk { + b.via[key] = via{Ref: fmt.Sprintf("refs/pull/%d/head", p.Number), Base: trunk.Commit} + } + b.mu.Unlock() + b.consider(ctx, name, m) + } + } +} + +// agreedFor are the toolchains the build nodes agree on now, for the languages of the manifest's plan: the toolchains the +// candidate's fingerprint was computed with, as the indexer recomputes it whenever they change. A language with none agreed +// is left out, and the build then waits for it as the trunk's does. +func (b *Builder) agreedFor(ctx context.Context, manifest string) map[string]delivery.ToolchainUsed { + plan, err, _ := b.planOf(ctx, manifest) + if err != nil { + return nil + } + out := map[string]delivery.ToolchainUsed{} + for _, bm := range plan.Modules() { + if !plan.NeedsToolchain(bm) { + continue + } + language := buildrun.LanguageOf[bm] + if tc, none := delivery.AgreedToolchain(b.Toolchains.Entries(language)); none == "" { + out[language] = delivery.ToolchainUsed{Version: tc.Version, Recipe: tc.Recipe} + } + } + if len(out) == 0 { + return nil + } + return out +} diff --git a/modules/mesh-delivery-builder/internal/build/candidates_test.go b/modules/mesh-delivery-builder/internal/build/candidates_test.go new file mode 100644 index 0000000..acc5976 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/candidates_test.go @@ -0,0 +1,96 @@ +package build + +import ( + "context" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// trunkModule writes the trunk's record of the module m at a fingerprint of its own, so that the candidate's fingerprint (fp1) +// is one the trunk does not carry. +func trunkModule(t *testing.T, r *rig) { + t.Helper() + data, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: "7a74cee83f88e6585a5511c3cc53bdf0f5303d20", FP: "src1:" + hex64('f'), + FP2: "src2:" + hex64('e'), Manifest: mdigest('a'), Path: "modules/m", Toolchains: agreedToolchains}) + if err != nil { + t.Fatal(err) + } + r.b.Module(context.Background(), "m", data) + r.b.Wait() + r.bm.mu.Lock() + r.bm.calls = nil + r.bm.mu.Unlock() +} + +func candidateValue(t *testing.T, group string, number int, head string) []byte { + t.Helper() + pulls := []delivery.CandidatePull{{Head: delivery.Head{Repository: "novox/mesh-catalog", Number: number, Head: head}, Modules: []string{"m"}, + Fps: map[string]delivery.CandidateModule{"m": {FP2: fp1, Manifest: mdigest('a')}}}} + hash, err := delivery.HeadsHash([]delivery.Head{pulls[0].Head}) + if err != nil { + t.Fatal(err) + } + data, err := delivery.EncodeCandidate(delivery.Candidate{V: delivery.Version, Group: group, HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)}) + if err != nil { + t.Fatal(err) + } + return data +} + +const ( + headX = "1111111111111111111111111111111111111111" + headY = "2222222222222222222222222222222222222222" +) + +func TestACandidatesFingerprintIsBuiltOnceAsAKeyOfItsOwnFromTheHeadMergedOntoTheTrunk(t *testing.T) { // N7/AC1, N7/AC2 (issue 1684) + r := newRig(t, goBundle) + trunkModule(t, r) + r.b.Candidate(context.Background(), "feat%2Fx", candidateValue(t, "feat/x", 7, headX)) + r.b.Wait() + v := valueOf(t, r.st, r.key) + if v.State != delivery.BuildDone || len(v.Artifacts) != 1 { + t.Fatalf("build = %+v, want the candidate's fingerprint built", v) + } + fetches := r.bm.callsOf("build_go_fetch") + if len(fetches) != 2 { // one on each of two nodes + t.Fatalf("%d fetches, want one per node", len(fetches)) + } + f := fetches[0].body.(map[string]any) + if f["commit"] != headX || f["ref"] != "refs/pull/7/head" || f["base"] != "7a74cee83f88e6585a5511c3cc53bdf0f5303d20" || f["repository"] != "novox/mesh-catalog" { + t.Errorf("fetch args = %+v, want the head, its ref and the trunk commit to merge onto", f) + } + if b := r.bm.callsOf("build_go_build")[0].body.(map[string]any); b["module"] != "modules/m" { + t.Errorf("build args = %+v", b) + } + // Two heads carrying one fingerprint are one build. + r.b.Candidate(context.Background(), "feat%2Fy", candidateValue(t, "feat/y", 8, headY)) + r.b.Wait() + if n := len(r.bm.callsOf("build_go_fetch")); n != 2 { + t.Errorf("%d fetches after a second head with the same fingerprint, want no more", n) + } +} + +func TestACandidateWhoseModuleHasNoTrunkRecordIsNotBuilt(t *testing.T) { + r := newRig(t, goBundle) + r.b.Candidate(context.Background(), "feat%2Fx", candidateValue(t, "feat/x", 7, headX)) + r.b.Wait() + if _, ok := r.st.vals[r.key]; ok { + t.Error("a build key was written for a module the builder knows no directory of") + } +} + +func TestAFingerprintTheTrunkCarriesIsBuiltFromTheTrunksCommit(t *testing.T) { // the trunk's source wins + r := newRig(t, goBundle) + r.b.Candidate(context.Background(), "feat%2Fx", candidateValue(t, "feat/x", 7, headX)) + r.b.Module(context.Background(), "m", moduleValue(t, "modules/m")) // the trunk now carries fp1 + r.b.Wait() + for _, c := range r.bm.callsOf("build_go_fetch") { + if f := c.body.(map[string]any); f["ref"] != nil || f["base"] != nil { + t.Errorf("fetch args = %+v: a trunk fingerprint is built from the trunk's commit alone", f) + } + } + if v := valueOf(t, r.st, r.key); v.State != delivery.BuildDone { + t.Errorf("build = %+v", v) + } +} diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index 66877e3..495027f 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -50,11 +50,14 @@ type Builder struct { sem chan struct{} wg sync.WaitGroup mu sync.Mutex - known map[string]delivery.Module // by modules key - inflight map[string]bool // by build key - again map[string]delivery.Module // by build key: the newest value that arrived while the key was being built - plans map[string]planned // by manifest digest - tried map[string]bool // by build key: looked at since this process started + known map[string]delivery.Module // by modules key + inflight map[string]bool // by build key + again map[string]delivery.Module // by build key: the newest value that arrived while the key was being built + plans map[string]planned // by manifest digest + cands map[string]delivery.Candidate // by candidates key (candidates.go) + via map[string]via // by build key: a candidate's fetch (candidates.go) + trunkFP map[string]bool // by build key: a trunk module value carries this fingerprint + tried map[string]bool // by build key: looked at since this process started // epoch counts the changes of the nodes (online, healthy, a toolchain) seen since this process started; a done key // whose second-node build failed is held back until it moves (issue 1664). epoch uint64 @@ -73,6 +76,7 @@ func (b *Builder) init() { limit = 4 } 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.secondAt = map[string]uint64{} if b.Places == nil { @@ -101,6 +105,11 @@ func (b *Builder) Module(ctx context.Context, key string, value []byte) { } b.mu.Lock() b.known[key] = m + if bk, err := delivery.BuilderBuildKey(key, m.FP2); err == nil && m.FP2 != "" { + // A fingerprint the trunk carries is built from the trunk's commit, whoever asked for it first. + b.trunkFP[bk] = true + delete(b.via, bk) + } b.mu.Unlock() b.consider(ctx, key, m) } @@ -182,6 +191,16 @@ func (b *Builder) Reconsider(ctx context.Context) { for _, k := range sortedKeys(known) { b.consider(ctx, k, known[k]) } + b.mu.Lock() + cands := make([]delivery.Candidate, 0, len(b.cands)) + for _, c := range b.cands { + cands = append(cands, c) + } + b.mu.Unlock() + sort.Slice(cands, func(i, j int) bool { return cands[i].Group < cands[j].Group }) + for _, c := range cands { + b.considerCandidate(ctx, c) + } } func sortedKeys(m map[string]delivery.Module) []string { @@ -490,7 +509,16 @@ func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m del return } defer release() - raw, err := b.step(ctx, l.module, l.node, "fetch", map[string]any{"repository": repository, "commit": m.Commit}) + fetchArgs := map[string]any{"repository": repository, "commit": m.Commit} + if key, err := delivery.BuilderBuildKey(module, m.FP2); err == nil { + b.mu.Lock() + v, candidate := b.via[key] + b.mu.Unlock() + if candidate { + fetchArgs["ref"], fetchArgs["base"] = v.Ref, v.Base + } + } + raw, err := b.step(ctx, l.module, l.node, "fetch", fetchArgs) if err != nil { l.err = err return diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 66ac3cf..70f4ca4 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -87,7 +87,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) { + "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) { t.Errorf("reads = %v, want %v", m.Reads, want) } if want := []string{"mesh-build-go.build_go_build", "mesh-build-go.build_go_fetch", "mesh-build-node.build_node_build", "mesh-build-node.build_node_fetch"}; !reflect.DeepEqual(sorted(m.Invokes), want) { diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 292c9ae..cb976f3 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -14,6 +14,7 @@ "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", + "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence",