diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index d220de5d..38bac683 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -41,8 +41,10 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-checker/checkermsg" "git.novox.be/novox/mesh-delivery/sdk/buildstep" "git.novox.be/novox/mesh-sdk/go/announce" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views/viewstest" @@ -434,15 +436,30 @@ func startBed(t *testing.T) *bed { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: buildrun.DeadLettersStream, Subjects: []string{"mesh.events.dead.>"}}); err != nil { t.Fatal(err) } - // The checker's streams as its manifest declares them (ADR 0619, ADR 0620): checks, which the indexer and the forge - // bridge hear, and asks, with verdict and tag-wanted at the subjects the mesh-delivery-checker definition names. - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHECKS", Subjects: []string{meshdeliverychecker.HolderSubjectPreviewWritten(check.CheckerModule)}}); err != nil { + // The checker's streams as its manifest declares them (ADR 0619, ADR 0620), read from it so the bed holds every event + // it declares: checks, which the indexer and the forge bridge hear (preview, owed and a whole try's start and end, whose + // say would otherwise wait out its retries on a subject no stream holds), and asks, with verdict and tag-wanted. + raw, err := os.ReadFile("../../module.json") + if err != nil { t.Fatal(err) } - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASKS", Subjects: []string{meshdeliverychecker.HolderSubjectVerdict(check.CheckerModule), - meshdeliverychecker.HolderSubjectTagWanted(check.CheckerModule)}}); err != nil { + m, err := catalogue.Decode(raw) + if err != nil { t.Fatal(err) } + for _, s := range m.Streams { + var subjects []string + for _, e := range s.Events { + subjects = append(subjects, process.ModuleStreamSubject(check.CheckerModule, strings.ReplaceAll(e, "**", ">"))) + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: s.BusName(), Subjects: subjects}); err != nil { + t.Fatal(err) + } + } + if !slices.ContainsFunc(m.Streams, func(s catalogue.StreamDeclaration) bool { return s.Name == "checks" }) || + meshdeliverychecker.HolderSubjectVerdict(check.CheckerModule) != process.ModuleStreamSubject(check.CheckerModule, "verdict") { + t.Fatalf("the checker's manifest declares the streams %+v, want checks, and asks at the definition's subjects", m.Streams) + } // The shared events stream with the seats' and modules' events (the package registry seats' published among them), and the checker's durable consumer of it. if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil { t.Fatal(err) @@ -1032,6 +1049,22 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) { if b.fake.count("mesh-build-go.build_go_fetch") == 0 { t.Errorf("the try fetched nothing") } + // A whole try's start and end are said on the stream checks, which the indexer hears to write and delete the + // candidate of its group: said once each, never waited out on a subject no stream holds. + checks, err := b.js.Stream(b.ctx, "CHECKS") + if err != nil { + t.Fatal(err) + } + for _, event := range []string{checkermsg.Started, checkermsg.Ended} { // the end is said once the try's end is written + said, err := b.js.OrderedConsumer(b.ctx, checks.CachedInfo().Config.Name, jetstream.OrderedConsumerConfig{ + FilterSubjects: []string{process.ModuleStreamSubject(check.CheckerModule, event)}}) + if err != nil { + t.Fatal(err) + } + if m, err := said.Next(jetstream.FetchMaxWait(5 * time.Second)); err != nil || !strings.Contains(string(m.Data()), id) { + t.Errorf("the stream checks holds no %s of the try %s: %v", event, id, err) + } + } for _, name := range []string{check.CheckerOwedBucket, check.CheckerVerdictsBucket} { if keys, _ := b.buckets[name].Keys(b.ctx); len(keys) != 0 { t.Errorf("a try wrote %s: %v", name, keys) diff --git a/modules/mesh-delivery-checker/internal/check/try.go b/modules/mesh-delivery-checker/internal/check/try.go index 2f39fe33..862a7a39 100644 --- a/modules/mesh-delivery-checker/internal/check/try.go +++ b/modules/mesh-delivery-checker/internal/check/try.go @@ -711,7 +711,10 @@ func (tr *Tryer) run(ctx context.Context, key string) { return } tr.logf("%s: %s", key, t.State) - tr.sendEnd(ctx, key) + // The end just written comes back on the key's watch, and Put then stops this run's context (it stops judging a try + // ended by hand) while this run says the end, which Put's own say leaves to it: the end is said on a context that + // stop does not cancel, or it is never said and the indexer keeps the candidate of the try's group. + tr.sendEnd(context.WithoutCancel(ctx), key) } // sendEnd says the end of an ended try whose start was said (Told) with the state it ended in, then clears Told in the try's