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 0b31fb10..1e8664c9 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 @@ -11,6 +11,7 @@ import ( "fmt" "os" "path/filepath" + "slices" "strings" "sync" "sync/atomic" @@ -24,6 +25,7 @@ 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/gopackageregistry" "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" @@ -600,6 +602,47 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7 b.awaitFP(t, "netcheck", third, isState(delivery.BuildDone)) b.awaitSaid(t, "kept: fingerprint "+third+" is numbered under it") reg.awaitFinding(t, failed, false) + // DLV-B4: the seats' refused and removed, heard on the builder's durable consumer, move a numbered version: a retention + // removal turns the entry removed with its reason, a refusal refused with the holder's reason and the bytes it refused. + reg.outcome(t, ociartifactregistry.SubjectRemoved, versions.Outcome{Name: "netcheck/tools-typescript", Version: "0.1.5", Kind: "bundle", Holder: "distribution", Reason: "retention"}) + reg.outcome(t, ociartifactregistry.SubjectRefused, versions.Outcome{Name: "netcheck/tools-go", Version: "0.1.6", Digest: digestOf("other"), Kind: "bundle", Holder: "distribution", Reason: "the cap"}) + reg.awaitOf(t, "netcheck/tools-typescript", func(es []versions.Entry) bool { + return slices.ContainsFunc(es, func(e versions.Entry) bool { + return e.Version == "0.1.5" && e.State == versions.Removed && e.Reason == "retention" + }) + }) + reg.awaitOf(t, "netcheck/tools-go", func(es []versions.Entry) bool { + return slices.ContainsFunc(es, func(e versions.Entry) bool { + return e.Version == "0.1.6" && e.State == versions.Refused && e.Reason == "the cap" && e.RefusedDigest == digestOf("other") + }) + }) + // DLV-B6: a branch's end (the forge bridge's merged) asks the seat to remove each version tagged for it, the reason + // tag-gone, and the seat's removed turns it removed. + tagged := versions.Entry{Name: "mesh-sdk/go", Kind: versions.KindPackage, Series: "0.1", Version: "0.1.15-feat-x.1", Tag: "feat-x", Module: "mesh-sdk", + Fingerprint: "src2:" + strings.Repeat("5", 64), Repository: "novox/mesh-sdk", Seat: gopackageregistry.Seat, State: versions.Published, At: time.Now().UTC()} + raw, _ := json.Marshal(tagged) + if _, err := reg.versions.Put(b.ctx, versions.Key(tagged.Name, tagged.Version), raw); err != nil { + t.Fatal(err) + } + if _, err := b.js.Publish(b.ctx, process.ModuleStreamSubject(delivery.ForgeBridgeModule, "forge"), + []byte(`{"kind":"`+delivery.ForgeMerged+`","repository":"novox/mesh-sdk","branch":"feat/x"}`)); err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(20 * time.Second); ; time.Sleep(50 * time.Millisecond) { + reg.mu.Lock() + asked := slices.ContainsFunc(reg.removes, func(a map[string]any) bool { + return a["name"] == tagged.Name && a["version"] == tagged.Version && a["reason"] == "tag-gone" + }) + reg.mu.Unlock() + if asked { + break + } + if time.Now().After(deadline) { + t.Fatalf("the seat's remove of %s@%s was not asked at the branch's end: %v", tagged.Name, tagged.Version, reg.removes) + } + } + reg.outcome(t, gopackageregistry.SubjectRemoved, versions.Outcome{Name: tagged.Name, Version: tagged.Version, Kind: versions.KindPackage, Holder: "gitea", Reason: "tag-gone"}) + reg.awaitOf(t, tagged.Name, func(es []versions.Entry) bool { return len(es) == 1 && es[0].State == versions.Removed }) } // TestAnArchiveOnlyModuleAndARootModuleAreBuilt: the indexer gives an archive no language, so modules.ca-trust has no @@ -1066,6 +1109,7 @@ type registry struct { versions, findings, reasks jetstream.KeyValue mu sync.Mutex numbered []versions.NumberedBody + removes []map[string]any } func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *registry { @@ -1080,14 +1124,33 @@ func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *regis } 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) + // EVENTS with two seats' outcomes, and the forge bridge's stream (a branch's end): each with the builder's durable consumer. + for stream, subjects := range map[string][]string{ + eventsStream: {delivery.ArtifactsGoneSubject, ociartifactregistry.SubjectPublished, ociartifactregistry.SubjectRefused, ociartifactregistry.SubjectRemoved, + gopackageregistry.SubjectPublished, gopackageregistry.SubjectRefused, gopackageregistry.SubjectRemoved}, + forgeStream: {process.ModuleStreamSubject(delivery.ForgeBridgeModule, ">")}, + } { + if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: stream, Subjects: subjects}); err != nil { + t.Fatal(err) + } + if _, err := b.js.CreateOrUpdateConsumer(b.ctx, stream, jetstream.ConsumerConfig{Durable: node + "_" + delivery.BuilderModule, + AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } } - if _, err := b.js.CreateOrUpdateConsumer(b.ctx, eventsStream, jetstream.ConsumerConfig{Durable: node + "_" + delivery.BuilderModule, - AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + // The Go seat's remove, as its holder answers it; each call is kept. + rm, err := b.js.Conn().Subscribe("mesh.seat."+gopackageregistry.Seat+".call.remove.>", func(m *nats.Msg) { + var args map[string]any + _ = json.Unmarshal(m.Data, &args) + r.mu.Lock() + r.removes = append(r.removes, args) + r.mu.Unlock() + _ = m.Respond([]byte(`{"result":{},"node":"ctl"}`)) + }) + if err != nil { t.Fatal(err) } + t.Cleanup(func() { _ = rm.Unsubscribe() }) 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) @@ -1160,12 +1223,18 @@ func (r *registry) awaitOf(t *testing.T, name string, pred func([]versions.Entry return es } -// published says the seat's published for name@version, as its holder does. +// published says the artifact 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) + r.outcome(t, ociartifactregistry.SubjectPublished, versions.Outcome{Name: name, Version: version, Digest: digest, Kind: "bundle", Holder: "distribution"}) +} + +// outcome says a seat's outcome on subject, as its holder does. +func (r *registry) outcome(t *testing.T, subject string, o versions.Outcome) { + t.Helper() + data, _ := json.Marshal(o) + msg := nats.NewMsg(subject) + msg.Header.Set("x-event-id", subject+"-"+o.Name+"-"+o.Version) msg.Data = data if _, err := r.b.js.PublishMsg(r.b.ctx, msg); err != nil { t.Fatal(err)