diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go index 359b1e9..2f32c84 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go @@ -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(): 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 5a37fc5..5184160 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 @@ -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. done, its fetch given the head, the ref that carries it and the trunk commit (issue 1684). +// builds.netcheck. 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) } 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 b146faa..77cbc22 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -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 +} diff --git a/modules/mesh-delivery-builder/internal/build/candidates.go b/modules/mesh-delivery-builder/internal/build/candidates.go index d96df2c..108c149 100644 --- a/modules/mesh-delivery-builder/internal/build/candidates.go +++ b/modules/mesh-delivery-builder/internal/build/candidates.go @@ -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). 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() 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.consider(ctx, name, m) @@ -106,3 +114,18 @@ func (b *Builder) agreedFor(ctx context.Context, manifest string) map[string]del } 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 +} diff --git a/modules/mesh-delivery-builder/internal/build/candidates_test.go b/modules/mesh-delivery-builder/internal/build/candidates_test.go index 97ff67a..af8d616 100644 --- a/modules/mesh-delivery-builder/internal/build/candidates_test.go +++ b/modules/mesh-delivery-builder/internal/build/candidates_test.go @@ -40,8 +40,11 @@ func candidateValue(t *testing.T, group string, number int, head string) []byte } const ( - headX = "1111111111111111111111111111111111111111" - headY = "2222222222222222222222222222222222222222" + // trunkHead is the trunk head the rig's source revision names: newer than the module value's commit, which stays at the + // commit its fingerprint first appeared at. + trunkHead = "3333333333333333333333333333333333333333" + headX = "1111111111111111111111111111111111111111" + headY = "2222222222222222222222222222222222222222" ) 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)) } 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 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 head the candidate was previewed 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) diff --git a/modules/mesh-delivery-builder/internal/build/r4e_test.go b/modules/mesh-delivery-builder/internal/build/r4e_test.go index fab0854..a28bfda 100644 --- a/modules/mesh-delivery-builder/internal/build/r4e_test.go +++ b/modules/mesh-delivery-builder/internal/build/r4e_test.go @@ -379,8 +379,9 @@ func TestACandidatesFetchNamesTheNamedPathsTheIndexerGaveIt(t *testing.T) { // A t.Fatal("no fetch") } 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 == "" { - t.Errorf("fetch request = %+v, %v; want the candidate's base, the module's directory and its named paths", req, err) + // The base is the trunk head the candidate was previewed onto, never the module value's commit (an older one). + 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) } } diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index 1d140d7..492cae5 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -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 // 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) + // 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 sem chan struct{} @@ -269,6 +272,13 @@ func (b *Builder) Reconsider(ctx context.Context) { for _, k := range sortedKeys(known) { 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() cands := make([]delivery.Candidate, 0, len(b.cands)) for _, c := range b.cands { diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index f3e82b8..f4ad7b2 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -176,6 +176,7 @@ func newRig(t *testing.T, artifacts ...catalogue.Artifact) *rig { r.b = &Builder{ 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, + Trunk: func(context.Context, uint64, string) (string, error) { return trunkHead, nil }, } var err error if r.key, err = delivery.BuilderBuildKey("m", fp1); err != nil { diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index e5dd046..27e44e9 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -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 // 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", - "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) } if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) { diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 0333c9a..e15c633 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -21,6 +21,7 @@ "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments", + "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"