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 +}