From fb4fc8a602f73781dd394554145a36941b1efd1a Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 13:20:49 +0100 Subject: [PATCH] DLV-B10: design 51's round-2 builder cases are cases of the builder's whole-module test The bed holds the builder's versions, findings, floor-waits and reasks, EVENTS with the artifact seat's outcomes, and the floor's three sources served on the bus. Two merges close together number the second at once; builder_publish_again after published takes a new number; a step bound running out raises build-failed, and the fingerprint built again keeps its number, said with why, and clears it. Each case failed first with its behaviour off. The kept version is said in the process log until the build log takes build.said. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../mesh-delivery-builder/integration_test.go | 241 +++++++++++++++++- 1 file changed, 235 insertions(+), 6 deletions(-) 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 2b6b08b1..0b31fb10 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 @@ -24,9 +24,11 @@ import ( // The build runner the build modules pull and run asks with: mesh-build-go's own package (novox/hq ADR 0593 ยง1), // another repository's, by its published version, so the builder is tested against the real pulling code. runner "git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep" + "git.novox.be/novox/mesh-contracts/go/seats/ociartifactregistry" "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-builder/internal/build" @@ -72,6 +74,9 @@ type fakeBuildModules struct { // waiting in their streams). closed atomic.Bool pullers []*runner.Puller + // hang, while it holds a channel, keeps every build step running until the channel is closed: a step that never ends, + // so the builder's step bound runs out (design 51's round-2 case). + hang atomic.Pointer[chan struct{}] } // open lets the nodes pull again and wakes their pullers. @@ -110,6 +115,12 @@ func (f *fakeBuildModules) step(module, name string, args map[string]any) (any, f.builds = append(f.builds, args) } f.mu.Unlock() + if h := f.hang.Load(); h != nil && name == "build" { + select { + case <-*h: + case <-f.ctx.Done(): + } + } switch { case name == "fetch": merge := "" @@ -341,15 +352,24 @@ func (b *bed) report(t *testing.T, module, node, language, version, recipe strin // module writes modules. as the indexer does for a module with a new-form fingerprint, and its manifest by digest. func (b *bed) module(t *testing.T, name, fixture, path string) string { - return b.moduleWith(t, name, fixture, path, map[string]delivery.ToolchainUsed{"go": {Version: "go1.27.1", Recipe: "1"}, "typescript": {Version: "tsc5", Recipe: "1"}}) + return b.moduleWith(t, name, fixture, path, bothToolchains) } +// bothToolchains are the toolchains the bed's nodes report, as a module's fingerprint names them. +var bothToolchains = map[string]delivery.ToolchainUsed{"go": {Version: "go1.27.1", Recipe: "1"}, "typescript": {Version: "tsc5", Recipe: "1"}} + // moduleWith is module with the toolchains its fingerprint was computed with. func (b *bed) moduleWith(t *testing.T, name, fixture, path string, toolchains map[string]delivery.ToolchainUsed) string { return b.moduleCommit(t, name, fixture, path, toolchains, commit) } func (b *bed) moduleCommit(t *testing.T, name, fixture, path string, toolchains map[string]delivery.ToolchainUsed, at string) string { + t.Helper() + return b.moduleFP(t, name, fixture, path, toolchains, at, fp, 0) +} + +// moduleFP is moduleCommit with the new-form fingerprint fp2 and the source revision src of its merge (0: none). +func (b *bed) moduleFP(t *testing.T, name, fixture, path string, toolchains map[string]delivery.ToolchainUsed, at, fp2 string, src uint64) string { t.Helper() raw, err := os.ReadFile(filepath.Join("testdata", fixture)) if err != nil { @@ -367,8 +387,8 @@ func (b *bed) moduleCommit(t *testing.T, name, fixture, path string, toolchains if _, err := manifests.Put(b.ctx, digest, mv); err != nil { t.Fatal(err) } - data, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: at, FP: "src1:" + strings.Repeat("f", 64), FP2: fp, - Manifest: digest, Path: path, Toolchains: toolchains}) + data, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-catalog", Commit: at, FP: "src1:" + strings.Repeat("f", 64), FP2: fp2, + Manifest: digest, Path: path, Toolchains: toolchains, Src: src}) if err != nil { t.Fatal(err) } @@ -444,7 +464,13 @@ func (b *bed) awaitSaid(t *testing.T, want string) string { // await reads the build key until pred holds of it, within 20 s. func (b *bed) await(t *testing.T, module string, pred func(delivery.BuilderBuild) bool) delivery.BuilderBuild { t.Helper() - key, err := delivery.BuilderBuildKey(module, fp) + return b.awaitFP(t, module, fp, pred) +} + +// awaitFP is await of the build key of fingerprint at. +func (b *bed) awaitFP(t *testing.T, module, at string, pred func(delivery.BuilderBuild) bool) delivery.BuilderBuild { + t.Helper() + key, err := delivery.BuilderBuildKey(module, at) if err != nil { t.Fatal(err) } @@ -471,7 +497,14 @@ func isState(state string) func(delivery.BuilderBuild) bool { // the digests of both bundles, both build modules' toolchains, after a fetch and a build on one node (ADR 0478 clause 4). func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7 b := startBed(t) - digest := b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck") + // DLV-B10 (design 51, round 2): the builder's versions, findings and the floor's three sources are on the bus; the seat + // answers 0.1.4, so netcheck's bundles are numbered from 0.1.5. Two merges close together, the first replaced before + // its claim: the second is numbered at once, and the first never. + reg := b.registry(t, "n1", "0.1.4") + const netcheck = "catalogue-a67f0196-netcheck.module.json" + first := "src2:" + strings.Repeat("1", 64) + b.moduleFP(t, "netcheck", netcheck, "modules/netcheck", bothToolchains, commit, first, 1) + digest := b.moduleFP(t, "netcheck", netcheck, "modules/netcheck", bothToolchains, commit, fp, 2) // DLV-B6: a tag ask kept and never numbered (no check builds its head) is narrated once its bound ends on the bus, with // what it waits on, and its bound is written again; no timer runs in the builder. key, ask, err := versions.NewTagAsk("novox/mesh-sdk", "feat/x", "abc", "a group member's check", "me@shanks", time.Now()) @@ -481,8 +514,14 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7 if _, err := b.tagAsks.Put(b.ctx, key, ask); err != nil { t.Fatal(err) } - b.start(t, "n1", map[string]string{"MESH_BUILDER_TAG_ASK_BOUND": "1s"}) + b.start(t, "n1", map[string]string{"MESH_BUILDER_TAG_ASK_BOUND": "1s", "MESH_BUILDER_STEP_BOUND": "1s"}) v := b.await(t, "netcheck", isState(delivery.BuildDone)) + for _, a := range []string{"tools-go", "tools-typescript"} { + es := reg.of(t, "netcheck/"+a) + if len(es) != 1 || es[0].Version != "0.1.5" || es[0].Fingerprint != fp || es[0].State != versions.Numbered || es[0].Digest != digestOf(a) { + t.Errorf("versions of netcheck/%s = %+v, want the second merge numbered 0.1.5 at once with its digest, the first never", a, es) + } + } if v.By != "n1.mesh-delivery-builder" || v.Manifest != digest || len(v.Artifacts) != 2 { t.Fatalf("build = %+v", v) } @@ -526,6 +565,41 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7 t.Fatalf("the bound is not written again after its narration: %v", err) } } + if got := reg.numberedOf("netcheck/tools-go"); len(got) != 1 || got[0] != "0.1.5" { + t.Errorf("version.numbered of netcheck/tools-go = %v, want 0.1.5 once", got) + } + // DLV-B10: builder_publish_again after a published takes a new number, and says it numbered. + for _, a := range []string{"tools-go", "tools-typescript"} { + reg.published(t, "netcheck/"+a, "0.1.5", digestOf(a)) + } + reg.awaitOf(t, "netcheck/tools-go", func(es []versions.Entry) bool { return len(es) == 1 && es[0].State == versions.Published }) + key, again, err := build.NewPublishAgain("netcheck/tools-go", fp, "the holder lost it", "me@shanks", time.Now()) + if err != nil { + t.Fatal(err) + } + if a := reg.answered(t, key, again); a.Refused || a.Do != versions.AgainRenumber || a.Version != "0.1.6" { + t.Errorf("builder_publish_again = %+v, want a new number, 0.1.6", a) + } + if got := reg.numberedOf("netcheck/tools-go"); len(got) != 2 || got[1] != "0.1.6" { + t.Errorf("version.numbered of netcheck/tools-go = %v, want 0.1.6 said after 0.1.5", got) + } + // DLV-B10: a build module's bound running out fails the trunk build, which is the builder's finding build-failed; the + // fingerprint built again keeps its number, said with why, and its done build clears the finding. + release := make(chan struct{}) + b.fake.hang.Store(&release) + third := "src2:" + strings.Repeat("4", 64) + b.moduleFP(t, "netcheck", netcheck, "modules/netcheck", bothToolchains, commit, third, 3) + if f := b.awaitFP(t, "netcheck", third, isState(delivery.BuildFailed)); !strings.Contains(f.Error, "did not end within") { + t.Errorf("the failed build = %+v, want its bound run out", f) + } + failed := health.FindingKey(health.ScopeModule, "netcheck", BuildFailed) + reg.awaitFinding(t, failed, true) + b.fake.hang.Store(nil) + close(release) + b.moduleFP(t, "netcheck", netcheck, "modules/netcheck", bothToolchains, commit, third, 3) // the indexer writes it again + b.awaitFP(t, "netcheck", third, isState(delivery.BuildDone)) + b.awaitSaid(t, "kept: fingerprint "+third+" is numbered under it") + reg.awaitFinding(t, failed, false) } // TestAnArchiveOnlyModuleAndARootModuleAreBuilt: the indexer gives an archive no language, so modules.ca-trust has no @@ -982,3 +1056,158 @@ func TestARestartedBuilderPutsNoSecondAsk(t *testing.T) { t.Errorf("done = %+v", done) } } + +// registry is the part of the bed design 51's builder cases need: the builder's states versions, findings, floor-waits and +// reasks, EVENTS with the seat oci-artifact-registry's outcomes and the builder's durable consumer of it, the builder's +// own events kept, and the floor's three sources served on the bus as their holders answer through the tool runner +// ({result}): the seat's versions, the forge bridge's bridge_tags and the indexer's indexer_requirements. +type registry struct { + b *bed + versions, findings, reasks jetstream.KeyValue + mu sync.Mutex + numbered []versions.NumberedBody +} + +func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *registry { + t.Helper() + r := ®istry{b: b} + mk := func(name string) jetstream.KeyValue { + kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: name, History: 1}) + if err != nil { + t.Fatal(err) + } + return kv + } + r.versions, r.findings, r.reasks = mk(versions.Bucket), mk(findingsBucket), mk(build.ReAsksBucket) + mk(versions.FloorWaitsBucket) + subjects := []string{delivery.ArtifactsGoneSubject, ociartifactregistry.SubjectPublished, ociartifactregistry.SubjectRefused, ociartifactregistry.SubjectRemoved} + if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: subjects}); err != nil { + t.Fatal(err) + } + if _, err := b.js.CreateOrUpdateConsumer(b.ctx, eventsStream, jetstream.ConsumerConfig{Durable: node + "_" + delivery.BuilderModule, + AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + own := "mesh.mod." + delivery.BuilderModule + ".event.>" + if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "BUILDER_EVENTS", Subjects: []string{own}}); err != nil { + t.Fatal(err) + } + nc := b.js.Conn() + sub, err := nc.Subscribe("mesh.mod."+delivery.BuilderModule+".event."+versions.NumberedEvent, func(m *nats.Msg) { + var nb versions.NumberedBody + if json.Unmarshal(m.Data, &nb) == nil { + r.mu.Lock() + r.numbered = append(r.numbered, nb) + r.mu.Unlock() + } + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + var listed []map[string]string + for _, v := range seatVersions { + listed = append(listed, map[string]string{"version": v}) + } + for subject, answer := range map[string]any{ + "mesh.seat." + ociartifactregistry.Seat + ".call.versions.>": map[string]any{"versions": listed}, + "mesh.mod.mesh-delivery-forge-bridge.call.bridge_tags.>": map[string]any{"tags": []string{}}, + "mesh.mod." + delivery.IndexerModule + ".call.indexer_requirements.>": map[string]any{"requirements": []string{}}, + } { + reply, _ := json.Marshal(map[string]any{"result": answer, "node": "ctl"}) + s, err := nc.Subscribe(subject, func(m *nats.Msg) { _ = m.Respond(reply) }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = s.Unsubscribe() }) + } + return r +} + +// of is every entry versions holds of name. +func (r *registry) of(t *testing.T, name string) []versions.Entry { + t.Helper() + l, err := r.versions.ListKeysFiltered(r.b.ctx, versions.NameFilter(name)) + if err != nil { + t.Fatal(err) + } + defer func() { _ = l.Stop() }() + var out []versions.Entry + for k := range l.Keys() { + e, err := r.versions.Get(r.b.ctx, k) + if err != nil { + t.Fatal(err) + } + var v versions.Entry + if err := json.Unmarshal(e.Value(), &v); err != nil { + t.Fatal(err) + } + out = append(out, v) + } + return out +} + +// awaitOf reads name's entries until pred holds of them, within 20 s. +func (r *registry) awaitOf(t *testing.T, name string, pred func([]versions.Entry) bool) []versions.Entry { + t.Helper() + var es []versions.Entry + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if es = r.of(t, name); pred(es) { + return es + } + } + t.Fatalf("versions of %s = %+v within 20 s", name, es) + return es +} + +// published says the seat's published for name@version, as its holder does. +func (r *registry) published(t *testing.T, name, version, digest string) { + t.Helper() + data, _ := json.Marshal(versions.Outcome{Name: name, Version: version, Digest: digest, Kind: "bundle", Holder: "distribution"}) + msg := nats.NewMsg(ociartifactregistry.SubjectPublished) + msg.Header.Set("x-event-id", "published-"+name+"-"+version) + msg.Data = data + if _, err := r.b.js.PublishMsg(r.b.ctx, msg); err != nil { + t.Fatal(err) + } +} + +// answered writes a reasks value and reads it until the builder answered it, within 20 s. +func (r *registry) answered(t *testing.T, key string, value []byte) build.ReAsk { + t.Helper() + if _, err := r.reasks.Put(r.b.ctx, key, value); err != nil { + t.Fatal(err) + } + var a build.ReAsk + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err := r.reasks.Get(r.b.ctx, key); err == nil && json.Unmarshal(e.Value(), &a) == nil && !a.Answered.IsZero() { + return a + } + } + t.Fatalf("reasks.%s = %+v, not answered within 20 s", key, a) + return a +} + +// awaitFinding reads the builder's finding key until it is there (raised) or gone, within 20 s. +func (r *registry) awaitFinding(t *testing.T, key string, raised bool) { + t.Helper() + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if _, err := r.findings.Get(r.b.ctx, key); (err == nil) == raised { + return + } + } + t.Fatalf("the finding %s raised = %v within 20 s, want %v", key, !raised, raised) +} + +// numberedOf are the version.numbered the builder said for name, in order. +func (r *registry) numberedOf(name string) []string { + r.mu.Lock() + defer r.mu.Unlock() + var out []string + for _, nb := range r.numbered { + if nb.Name == name { + out = append(out, nb.Version) + } + } + return out +}