novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

The builder builds the fingerprints a candidate carries, once each, from the head merged onto the trunk (issue 1684)

Browse Source
This commit is contained in:
jschoubben
2026-10-21 21:28:38 +02:00
parent 7433b0d047
commit 66908d9bbc
7 changed files with 316 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go
+62 -10
View File
@@ -56,6 +56,7 @@ type fakeBuildModules struct {
failBuild atomic.Bool
mu sync.Mutex
calls map[string]int // by `<module>.<tool>`
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.<fp> 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)
}
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go
+18 -1
View File
@@ -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()
modules/mesh-delivery-builder/internal/build/candidates.go
+104
View File
@@ -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.<group>), and
// the builder builds each, once, as a key of its own (builds.<module>.<fingerprint>) 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
}
modules/mesh-delivery-builder/internal/build/candidates_test.go
+96
View File
@@ -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)
}
}
modules/mesh-delivery-builder/internal/build/run.go
+34 -6
View File
@@ -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
modules/mesh-delivery-builder/manifest_test.go
+1 -1
View File
@@ -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) {
modules/mesh-delivery-builder/module.json
+1
View File
@@ -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",