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.
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 {
 		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 {
 		select {
 		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 {
 	ctx        context.Context
 	js         jetstream.JetStream
 	modules    jetstream.KeyValue
 	candidates jetstream.KeyValue
+	source     jetstream.KeyValue
 	builds     jetstream.KeyValue
 	tools      map[string]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.candidates = mk(delivery.CandidatesBucket, delivery.CandidatesHistory)
+	b.source = mk(delivery.SourceBucket, delivery.SourceHistory)
 	mk(delivery.ManifestsBucket, delivery.ManifestsHistory)
 	b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
 	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
 // 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
 	b := startBed(t)
 	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"},
 		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)})
+	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 {
 		t.Fatal(err)
 	}
@@ -550,12 +571,12 @@ func TestBuilderBuildsACandidatesFingerprintFromTheHeadMergedOntoTheTrunk(t *tes
 	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 {
+		if f["commit"] == head && f["ref"] == "refs/pull/7/head" && f["base"] == trunkHead {
 			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)
+		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"},
 		Fps: map[string]delivery.CandidateModule{"imaged": {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)})
+	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 {
 		t.Fatal(err)
 	}
@@ -16,9 +16,12 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
 	"io"
 	"log"
+	"maps"
 	"os"
 	"os/signal"
+	"slices"
 	"strings"
+	"sync"
 	"syscall"
 	"time"

@@ -174,6 +177,27 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		Logf:       logger.Printf,
 		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
 	// 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
@@ -267,9 +291,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	}
 	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, cw, eachCandidate) }()
+	if sw != nil {
+		go func() {
+			errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) {
+				if trunks.put(e) {
+					b.TrunkMoved(ctx)
+				}
+			})
+		}()
+	}
 	select {
 	case <-ctx.Done():
 		b.Wait()
@@ -282,3 +315,50 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		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
+}