Merge a candidate's preview onto the trunk head it was previewed onto, not the module value's commit

The builder took a candidate build's merge base from the module's value commit. Now that a value keeps the commit its
fingerprint first appeared at, that commit can be many merges behind the trunk, so a preview would be built onto a stale
trunk (review round 1). The base is now the head of the pull request's repository in the gate's source revision the
candidate names (Candidate.Base): the builder reads mesh-delivery-gate.source, watched with its history (a direct get of
one revision is not what a read grants), and a candidate heard before its revision is built once the revision is heard.
This commit is contained in:
2026-10-25 21:13:59 +01:00
parent eaf27f36af
commit 9afef800c0
10 changed files with 159 additions and 14 deletions
@@ -66,6 +66,11 @@ func settle(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyV
if err != nil { if err != nil {
return nil, err return nil, err
} }
return settled(ctx, w, each)
}
// settled hands each entry the watch delivers before its end-of-initial marker to each, and answers the watch.
func settled(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.KeyValueEntry)) (jetstream.KeyWatcher, error) {
for { for {
select { select {
case <-ctx.Done(): case <-ctx.Done():
@@ -162,11 +162,29 @@ func (f *fakeBuildModules) serve(t *testing.T, module string, nodes ...string) {
} }
} }
// trunkHead is the trunk head of novox/mesh-catalog the bed's source revision names: newer than the module values' commit.
const trunkHead = "3333333333333333333333333333333333333333"
// trunk writes the gate's source.main naming head as novox/mesh-catalog's trunk head, and answers its revision.
func (b *bed) trunk(t *testing.T, head string) uint64 {
t.Helper()
data, err := delivery.EncodeSource(delivery.Source{V: delivery.Version, Group: "g", Heads: map[string]string{"novox/mesh-catalog": head}})
if err != nil {
t.Fatal(err)
}
rev, err := b.source.Put(b.ctx, delivery.SourceMainKey, data)
if err != nil {
t.Fatal(err)
}
return rev
}
type bed struct { type bed struct {
ctx context.Context ctx context.Context
js jetstream.JetStream js jetstream.JetStream
modules jetstream.KeyValue modules jetstream.KeyValue
candidates jetstream.KeyValue candidates jetstream.KeyValue
source jetstream.KeyValue
builds jetstream.KeyValue builds jetstream.KeyValue
tools map[string]jetstream.KeyValue tools map[string]jetstream.KeyValue
instances jetstream.KeyValue instances jetstream.KeyValue
@@ -243,6 +261,7 @@ func startBedWith(t *testing.T, closed bool) *bed {
} }
b.modules = mk(delivery.ModulesBucket, delivery.ModulesHistory) b.modules = mk(delivery.ModulesBucket, delivery.ModulesHistory)
b.candidates = mk(delivery.CandidatesBucket, delivery.CandidatesHistory) b.candidates = mk(delivery.CandidatesBucket, delivery.CandidatesHistory)
b.source = mk(delivery.SourceBucket, delivery.SourceHistory)
mk(delivery.ManifestsBucket, delivery.ManifestsHistory) mk(delivery.ManifestsBucket, delivery.ManifestsHistory)
b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory) b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
b.instances = mk(instancesBucket, 1) b.instances = mk(instancesBucket, 1)
@@ -513,7 +532,9 @@ func TestAKeyLeftBuildingForAModuleThatIsGoneIsClosedAtStart(t *testing.T) {
// TestBuilderBuildsACandidatesFingerprintFromTheHeadMergedOntoTheTrunk: the trunk's record of netcheck carries a fingerprint of its // 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 // 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). // builds.netcheck.<fp> done, its fetch given the head, the ref that carries it and the trunk head of the candidate's source
// revision, newer than the module value's commit (issue 1684; the operator 2026-10-25: a value names the commit its
// fingerprint first appeared at).
func TestBuilderBuildsACandidatesFingerprintFromTheHeadMergedOntoTheTrunk(t *testing.T) { // N7/AC1 func TestBuilderBuildsACandidatesFingerprintFromTheHeadMergedOntoTheTrunk(t *testing.T) { // N7/AC1
b := startBed(t) b := startBed(t)
other := "src2:" + strings.Repeat("e", 64) other := "src2:" + strings.Repeat("e", 64)
@@ -530,7 +551,7 @@ func TestBuilderBuildsACandidatesFingerprintFromTheHeadMergedOntoTheTrunk(t *tes
pulls := []delivery.CandidatePull{{Head: delivery.Head{Repository: "novox/mesh-catalog", Number: 7, Head: head}, Modules: []string{"netcheck"}, 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}}}} Fps: map[string]delivery.CandidateModule{"netcheck": {FP2: fp, Manifest: digest}}}}
hash, _ := delivery.HeadsHash([]delivery.Head{pulls[0].Head}) 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)}) cand, err := delivery.EncodeCandidate(delivery.Candidate{V: delivery.Version, Group: "feat/x", Base: b.trunk(t, trunkHead), HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -550,12 +571,12 @@ func TestBuilderBuildsACandidatesFingerprintFromTheHeadMergedOntoTheTrunk(t *tes
defer b.fake.mu.Unlock() defer b.fake.mu.Unlock()
found := false found := false
for _, f := range b.fake.fetches { for _, f := range b.fake.fetches {
if f["commit"] == head && f["ref"] == "refs/pull/7/head" && f["base"] == commit { if f["commit"] == head && f["ref"] == "refs/pull/7/head" && f["base"] == trunkHead {
found = true found = true
} }
} }
if !found { if !found {
t.Errorf("fetches = %v, want one of the candidate's head with its ref and the trunk commit", b.fake.fetches) t.Errorf("fetches = %v, want one of the candidate's head with its ref and the trunk head it was previewed onto, not the module value's commit", b.fake.fetches)
} }
} }
@@ -603,7 +624,7 @@ func TestAnImageIsBuiltOnOneNodeAndOnlyATrunkKeyPublishes(t *testing.T) { // 170
pulls := []delivery.CandidatePull{{Head: delivery.Head{Repository: "novox/mesh-catalog", Number: 7, Head: head}, Modules: []string{"imaged"}, pulls := []delivery.CandidatePull{{Head: delivery.Head{Repository: "novox/mesh-catalog", Number: 7, Head: head}, Modules: []string{"imaged"},
Fps: map[string]delivery.CandidateModule{"imaged": {FP2: fp, Manifest: digest}}}} Fps: map[string]delivery.CandidateModule{"imaged": {FP2: fp, Manifest: digest}}}}
hash, _ := delivery.HeadsHash([]delivery.Head{pulls[0].Head}) 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)}) cand, err := delivery.EncodeCandidate(delivery.Candidate{V: delivery.Version, Group: "feat/x", Base: b.trunk(t, trunkHead), HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
@@ -16,9 +16,12 @@ import (
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
"io" "io"
"log" "log"
"maps"
"os" "os"
"os/signal" "os/signal"
"slices"
"strings" "strings"
"sync"
"syscall" "syscall"
"time" "time"
@@ -174,6 +177,27 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
Logf: logger.Printf, Logf: logger.Printf,
Stored: storedVia(calls.Call), Stored: storedVia(calls.Call),
} }
// A candidate's preview is merged onto the trunk head the indexer previewed it onto: source.main at the candidate's base
// revision (the gate's source state). Where the bucket is not there yet no candidate's fingerprint is built, and said.
// The source's revisions are watched with their history (a direct get of one revision is not what a read grants): a
// revision heard after the candidate that names it builds that candidate then.
trunks := &sourceHeads{}
var sw jetstream.KeyWatcher
switch kv, err := js.KeyValue(ctx, delivery.SourceBucket); {
case errors.Is(err, jetstream.ErrBucketNotFound):
logger.Printf("the bucket %s does not exist yet: no candidate's fingerprint is built until the builder starts again", delivery.SourceBucket)
case err != nil:
return fmt.Errorf("the bucket %s: %w", delivery.SourceBucket, err)
default:
w, err := kv.Watch(ctx, delivery.SourceMainKey, jetstream.IncludeHistory())
if err != nil {
return fmt.Errorf("the bucket %s cannot be watched: %w", delivery.SourceBucket, err)
}
if sw, err = settled(ctx, w, func(e jetstream.KeyValueEntry) { trunks.put(e) }); err != nil {
return err
}
b.Trunk = trunks.head
}
// **Said once every build module answered a reference, taken away when stopped** (issue 1886, ADR 0522): the // **Said once every build module answered a reference, taken away when stopped** (issue 1886, ADR 0522): the
// controller delivers from this builder's keys only while an instance says it writes references. A bucket not there // controller delivers from this builder's keys only while an instance says it writes references. A bucket not there
// yet (the manifest's state not made) is said, and the controller's plan waits for the keys, named (no build-agent // yet (the manifest's state not made) is said, and the controller's plan waits for the keys, named (no build-agent
@@ -267,9 +291,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err return err
} }
logger.Printf("started on %s", cred.Node) logger.Printf("started on %s", cred.Node)
errs := make(chan error, 2) errs := make(chan error, 3)
go func() { errs <- follow(ctx, w, each) }() go func() { errs <- follow(ctx, w, each) }()
go func() { errs <- follow(ctx, cw, eachCandidate) }() go func() { errs <- follow(ctx, cw, eachCandidate) }()
if sw != nil {
go func() {
errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) {
if trunks.put(e) {
b.TrunkMoved(ctx)
}
})
}()
}
select { select {
case <-ctx.Done(): case <-ctx.Done():
b.Wait() b.Wait()
@@ -282,3 +315,50 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err return err
} }
} }
// sourceHeads are the trunk heads of each revision of the gate's source.main the watch delivered, the newest
// delivery.SourceHistory of them: what a candidate's base revision names.
type sourceHeads struct {
mu sync.Mutex
byRev map[uint64]map[string]string
}
// put takes one revision in and answers whether it is new.
func (s *sourceHeads) put(e jetstream.KeyValueEntry) bool {
if !opOf(e) {
return false
}
src, err := delivery.DecodeSource(e.Value())
if err != nil {
return false
}
s.mu.Lock()
defer s.mu.Unlock()
if s.byRev == nil {
s.byRev = map[uint64]map[string]string{}
}
if _, had := s.byRev[e.Revision()]; had {
return false
}
s.byRev[e.Revision()] = src.Heads
for len(s.byRev) > delivery.SourceHistory {
delete(s.byRev, slices.Min(slices.Collect(maps.Keys(s.byRev))))
}
return true
}
// head is the head of repository in revision rev; an error when the revision is not heard (yet, or any more).
func (s *sourceHeads) head(_ context.Context, rev uint64, repository string) (string, error) {
s.mu.Lock()
defer s.mu.Unlock()
heads, ok := s.byRev[rev]
if !ok {
return "", fmt.Errorf("source revision %d is not heard (it is built once it is)", rev)
}
for r, head := range heads {
if delivery.SameRepository(r, repository) {
return head, nil
}
}
return "", nil
}
@@ -73,9 +73,17 @@ func (b *Builder) considerCandidate(ctx context.Context, c delivery.Candidate) {
// The named paths' objects the preview gave it: the fetch names the paths (runLane, ADR 0499 clause 4). // The named paths' objects the preview gave it: the fetch names the paths (runLane, ADR 0499 clause 4).
m.Inputs2 = &delivery.KeptInputs2{Paths: cm.Paths} m.Inputs2 = &delivery.KeptInputs2{Paths: cm.Paths}
} }
// The preview is merged onto the trunk head the candidate was previewed onto: the head of the pull request's
// repository in the gate's source revision c.Base (ADR 0484 §3). Never the module's value's commit: that is the
// commit its fingerprint first appeared at, and may be many merges behind the trunk.
base, err := b.trunkHead(ctx, c.Base, p.Repository)
if err != nil {
b.logf("candidates.%s: the module %s's fingerprint %s is not built: %v", c.Group, name, cm.FP2, err)
continue
}
b.mu.Lock() b.mu.Lock()
if _, onTrunk := b.trunkFP[key]; !onTrunk { if _, onTrunk := b.trunkFP[key]; !onTrunk {
b.via[key] = via{Ref: fmt.Sprintf("refs/pull/%d/head", p.Number), Base: trunk.Commit} b.via[key] = via{Ref: fmt.Sprintf("refs/pull/%d/head", p.Number), Base: base}
} }
b.mu.Unlock() b.mu.Unlock()
b.consider(ctx, name, m) b.consider(ctx, name, m)
@@ -106,3 +114,18 @@ func (b *Builder) agreedFor(ctx context.Context, manifest string) map[string]del
} }
return out return out
} }
// trunkHead is the head of repository in the gate's source revision rev (Builder.Trunk), or why it cannot be said.
func (b *Builder) trunkHead(ctx context.Context, rev uint64, repository string) (string, error) {
if b.Trunk == nil {
return "", fmt.Errorf("the gate's source cannot be read here, so the trunk head %s was previewed onto is not known", repository)
}
head, err := b.Trunk(ctx, rev, repository)
if err != nil {
return "", fmt.Errorf("the trunk head of %s at source revision %d cannot be read: %w", repository, rev, err)
}
if head == "" {
return "", fmt.Errorf("source revision %d names no trunk head of %s", rev, repository)
}
return head, nil
}
@@ -40,8 +40,11 @@ func candidateValue(t *testing.T, group string, number int, head string) []byte
} }
const ( const (
headX = "1111111111111111111111111111111111111111" // trunkHead is the trunk head the rig's source revision names: newer than the module value's commit, which stays at the
headY = "2222222222222222222222222222222222222222" // commit its fingerprint first appeared at.
trunkHead = "3333333333333333333333333333333333333333"
headX = "1111111111111111111111111111111111111111"
headY = "2222222222222222222222222222222222222222"
) )
func TestACandidatesFingerprintIsBuiltOnceAsAKeyOfItsOwnFromTheHeadMergedOntoTheTrunk(t *testing.T) { // N7/AC1, N7/AC2 (issue 1684) func TestACandidatesFingerprintIsBuiltOnceAsAKeyOfItsOwnFromTheHeadMergedOntoTheTrunk(t *testing.T) { // N7/AC1, N7/AC2 (issue 1684)
@@ -58,8 +61,8 @@ func TestACandidatesFingerprintIsBuiltOnceAsAKeyOfItsOwnFromTheHeadMergedOntoThe
t.Fatalf("%d fetches, want one", len(fetches)) t.Fatalf("%d fetches, want one", len(fetches))
} }
f := fetches[0].body.(map[string]any) 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" { if f["commit"] != headX || f["ref"] != "refs/pull/7/head" || f["base"] != trunkHead || f["repository"] != "novox/mesh-catalog" {
t.Errorf("fetch args = %+v, want the head, its ref and the trunk commit to merge onto", f) t.Errorf("fetch args = %+v, want the head, its ref and the trunk head the candidate was previewed onto", f)
} }
if b := r.bm.callsOf("build_go_build")[0].body.(map[string]any); b["module"] != "modules/m" { if b := r.bm.callsOf("build_go_build")[0].body.(map[string]any); b["module"] != "modules/m" {
t.Errorf("build args = %+v", b) t.Errorf("build args = %+v", b)
@@ -379,8 +379,9 @@ func TestACandidatesFetchNamesTheNamedPathsTheIndexerGaveIt(t *testing.T) { // A
t.Fatal("no fetch") t.Fatal("no fetch")
} }
req, err := buildstep.ParseFetchRequest(f[0].body.(map[string]any)) req, err := buildstep.ParseFetchRequest(f[0].body.(map[string]any))
if err != nil || req.Dir != "modules/m" || strings.Join(req.Paths, ",") != "package.json,src" || req.Base == "" { // The base is the trunk head the candidate was previewed onto, never the module value's commit (an older one).
t.Errorf("fetch request = %+v, %v; want the candidate's base, the module's directory and its named paths", req, err) if err != nil || req.Dir != "modules/m" || strings.Join(req.Paths, ",") != "package.json,src" || req.Base != trunkHead {
t.Errorf("fetch request = %+v, %v; want the trunk head %.8s as the base, the module's directory and its named paths", req, err, trunkHead)
} }
} }
@@ -53,6 +53,9 @@ type Builder struct {
// Stored answers which references the artifact store no longer holds (the controller's verb held; issue 1937): a done // Stored answers which references the artifact store no longer holds (the controller's verb held; issue 1937): a done
// key of the trunk's fingerprint is asked before it is left done (ClaimOptions.Stored). Nil asks nothing. // key of the trunk's fingerprint is asked before it is left done (ClaimOptions.Stored). Nil asks nothing.
Stored func(ctx context.Context, references []string) (gone []string, err error) Stored func(ctx context.Context, references []string) (gone []string, err error)
// Trunk answers the head of a repository in the gate's source revision rev (source.main at that revision): the trunk a
// candidate's preview is merged onto (Candidate.Base). Nil builds no candidate's fingerprint, and says why.
Trunk func(ctx context.Context, rev uint64, repository string) (string, error)
once sync.Once once sync.Once
sem chan struct{} sem chan struct{}
@@ -269,6 +272,13 @@ func (b *Builder) Reconsider(ctx context.Context) {
for _, k := range sortedKeys(known) { for _, k := range sortedKeys(known) {
b.consider(ctx, k, known[k]) b.consider(ctx, k, known[k])
} }
b.TrunkMoved(ctx)
}
// TrunkMoved considers every candidate again once a revision of the gate's source is heard: a candidate heard before the
// revision it was previewed onto had no trunk head to be merged onto.
func (b *Builder) TrunkMoved(ctx context.Context) {
b.init()
b.mu.Lock() b.mu.Lock()
cands := make([]delivery.Candidate, 0, len(b.cands)) cands := make([]delivery.Candidate, 0, len(b.cands))
for _, c := range b.cands { for _, c := range b.cands {
@@ -176,6 +176,7 @@ func newRig(t *testing.T, artifacts ...catalogue.Artifact) *rig {
r.b = &Builder{ r.b = &Builder{
Claimer: Claimer{Store: r.st, By: "novox.builder"}, Asks: buildrun.Asker{Publisher: r.asks, Waiter: r.asks, Bound: 2 * time.Second}, 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, Toolchains: r.tools, Manifests: r.mf, Logf: t.Logf,
Trunk: func(context.Context, uint64, string) (string, error) { return trunkHead, nil },
} }
var err error var err error
if r.key, err = delivery.BuilderBuildKey("m", fp1); err != nil { if r.key, err = delivery.BuilderBuildKey("m", fp1); err != nil {
@@ -90,7 +90,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and // 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). // 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-controller.assignments", if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments",
"mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"}; !reflect.DeepEqual(sorted(m.Reads), want) { "mesh-delivery-gate.source", "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) t.Errorf("reads = %v, want %v", m.Reads, want)
} }
if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) { if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
@@ -21,6 +21,7 @@
"mesh-build-node.steps", "mesh-build-node.steps",
"mesh-build-node.toolchains", "mesh-build-node.toolchains",
"mesh-controller.assignments", "mesh-controller.assignments",
"mesh-delivery-gate.source",
"mesh-delivery-indexer.candidates", "mesh-delivery-indexer.candidates",
"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules" "mesh-delivery-indexer.modules"