From 4242fb5b0bc4017c0c5ebdefeedaefaad84e1fba Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 21:17:09 +0200 Subject: [PATCH] The indexer gives a candidate's touched module the fingerprint its merge preview gives it, and wakes on the checker's preview.written without a restart; pin mesh-sdk to the R4c-5a head (issues 1445, 1681) --- modules/mesh-delivery-builder/go.mod | 2 +- modules/mesh-delivery-builder/go.sum | 4 +- .../cmd/mesh-delivery-checker/judge.go | 3 +- .../cmd/mesh-delivery-checker/main.go | 10 + .../internal/check/examiner_test.go | 12 + .../internal/check/judge.go | 2 + .../internal/check/preview.go | 3 + .../mesh-delivery-checker/manifest_test.go | 12 +- modules/mesh-delivery-checker/module.json | 3 + modules/mesh-delivery-forge-bridge/go.mod | 2 +- modules/mesh-delivery-forge-bridge/go.sum | 4 +- .../mesh-delivery-indexer/integration_test.go | 89 +++++++ .../cmd/mesh-delivery-indexer/main.go | 19 ++ .../cmd/mesh-delivery-indexer/previews.go | 88 +++++++ modules/mesh-delivery-indexer/go.mod | 2 +- modules/mesh-delivery-indexer/go.sum | 4 +- .../internal/index/candidates.go | 4 +- .../internal/index/previews.go | 247 ++++++++++++++++++ .../internal/index/previews_test.go | 181 +++++++++++++ .../internal/index/source.go | 5 +- .../internal/index/toolchains.go | 8 + .../mesh-delivery-indexer/manifest_test.go | 4 +- modules/mesh-delivery-indexer/module.json | 2 + 23 files changed, 694 insertions(+), 16 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go create mode 100644 modules/mesh-delivery-indexer/internal/index/previews.go create mode 100644 modules/mesh-delivery-indexer/internal/index/previews_test.go diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index 4431597..b8b9bd8 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-builder go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021100209-cb54756c3865 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021185947-afbe371003a5 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-builder/go.sum b/modules/mesh-delivery-builder/go.sum index 639aa71..7c963a0 100644 --- a/modules/mesh-delivery-builder/go.sum +++ b/modules/mesh-delivery-builder/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021100209-cb54756c3865 h1:VLir9NZbyi1LdH1YwWDIVRuL+lDC+AYlm6FZPlvmlm4= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021100209-cb54756c3865/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021185947-afbe371003a5 h1:1BiPKRPTYpgex8rSHWT9az0BteOafGWRDzd/Ba6GAcA= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021185947-afbe371003a5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go index b80e7b0..0d55bda 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go @@ -23,12 +23,13 @@ type judgeDeps struct { kept check.KV previews check.KV repos *repoModules + announce func(ctx context.Context, key string) logf func(format string, args ...any) } func newJudge(d judgeDeps) check.Judge { return &check.Examiner{Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places, - Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Trunks: d.trunks, Logf: d.logf} + Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Announce: d.announce, Trunks: d.trunks, Logf: d.logf} } // manifestKV reads a manifest by its digest. diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 9fc4703..537680e 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -32,6 +32,10 @@ import ( "mesh-delivery-checker/internal/check" ) +// previewWrittenSubject is the subject of the checker's preview.written event, on its stream previews (the bus names the +// stream PREVIEWS); the indexer's watch of the previews bucket is woken by it. +const previewWrittenSubject = "mesh.stream." + delivery.CheckerModule + ".event.preview.written" + const ( // defaultStepBound is how long the checker waits for one step of a build module (a fetch downloads the Go modules, a // test level runs a repository's tests). @@ -197,6 +201,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro runner: runner, tools: tools, presence: presence, have: have, places: places, trunks: trunks, kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]}, repos: repos, logf: logger.Printf, + announce: func(ctx context.Context, key string) { + body, _ := json.Marshal(map[string]string{"key": key}) + if _, err := js.Publish(ctx, previewWrittenSubject, body); err != nil { + logger.Printf("previews.%s was written and the indexer could not be told: %v", key, err) + } + }, }) // The toolchains first, so a check that starts from the first candidate finds the nodes. diff --git a/modules/mesh-delivery-checker/internal/check/examiner_test.go b/modules/mesh-delivery-checker/internal/check/examiner_test.go index c7aa9a1..270a504 100644 --- a/modules/mesh-delivery-checker/internal/check/examiner_test.go +++ b/modules/mesh-delivery-checker/internal/check/examiner_test.go @@ -504,3 +504,15 @@ func TestAPreviewIsWrittenOnceAndAnExistingKeyStands(t *testing.T) { t.Error("the second merge commit replaced the first") } } + +func TestTheIndexerIsToldOfAPreviewOnceWhenItIsCreated(t *testing.T) { // 1445/AC4 (the checker's half) + b := newBench(t) + var told []string + b.ex.Announce = func(_ context.Context, key string) { told = append(told, key) } + b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) + b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) + key, _ := delivery.PreviewKey(repoA, head1, trunkA) + if len(told) != 1 || told[0] != key { + t.Errorf("told %v, want once of %s", told, key) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index 6f90d74..c29508d 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -50,6 +50,8 @@ type Examiner struct { // Previews is the previews bucket; nil writes none. Previews KV Modules BuildModules + // Announce says a preview was written under the key (the checker's preview.written event); nil says nothing. + Announce func(ctx context.Context, key string) // Trunks are the trunk heads of every repository, for the siblings a repository's tests read. Trunks interface{ Heads() map[string]string } Now func() time.Time diff --git a/modules/mesh-delivery-checker/internal/check/preview.go b/modules/mesh-delivery-checker/internal/check/preview.go index 6884a1e..e44a9d5 100644 --- a/modules/mesh-delivery-checker/internal/check/preview.go +++ b/modules/mesh-delivery-checker/internal/check/preview.go @@ -62,6 +62,9 @@ func (j *Examiner) writePreview(ctx context.Context, p Pull, fetched []Fetched) if errors.Is(err, ErrConflict) { return } + if err == nil && j.Announce != nil { + j.Announce(ctx, key) + } } } if err != nil { diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 892ecbe..1517d59 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -31,7 +31,13 @@ type artifact struct { Binary string `json:"binary"` } +type stream struct { + Name string `json:"name"` + Events []string `json:"events"` +} + type manifest struct { + Streams []stream `json:"streams"` Module string `json:"module"` Version string `json:"version"` Slug string `json:"slug"` @@ -105,7 +111,11 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 if m.SecretsOwner != "mesh-delivery-checker" { t.Errorf("secrets-owner = %q", m.SecretsOwner) } - for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats", "bus"} { + // The checker says each time it wrote a preview, so the indexer's watch finds the bucket (ADR 0484 clause 3). + if want := []stream{{Name: "previews", Events: []string{"preview.**"}}}; !reflect.DeepEqual(m.Streams, want) { + t.Errorf("streams = %+v, want %+v", m.Streams, want) + } + for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats", "bus"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the checker has none", k) } diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 8044314..eef0cc1 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -6,6 +6,9 @@ "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported" ], + "streams": [ + { "name": "previews", "events": ["preview.**"] } + ], "state": [ { "name": "kept", "history": 1 }, { "name": "owed", "history": 5 }, diff --git a/modules/mesh-delivery-forge-bridge/go.mod b/modules/mesh-delivery-forge-bridge/go.mod index 25936a8..d073e5f 100644 --- a/modules/mesh-delivery-forge-bridge/go.mod +++ b/modules/mesh-delivery-forge-bridge/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261020182814-4024909a2960 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021185947-afbe371003a5 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-forge-bridge/go.sum b/modules/mesh-delivery-forge-bridge/go.sum index 0e67d23..7c963a0 100644 --- a/modules/mesh-delivery-forge-bridge/go.sum +++ b/modules/mesh-delivery-forge-bridge/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261020182814-4024909a2960 h1:FCId3IuRScG03X7Gx9vfKfGNZH9jWnfx9RChfJ3eWfo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261020182814-4024909a2960/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021185947-afbe371003a5 h1:1BiPKRPTYpgex8rSHWT9az0BteOafGWRDzd/Ba6GAcA= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021185947-afbe371003a5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 11d9623..a885d9d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -351,3 +351,92 @@ func TestIndexerWritesAGoModuleAgainWhenTheBuildModuleReportsItsToolchain(t *tes t.Errorf("none2 = %q within 5 s of the toolchain key, want the builder's fingerprint not yet known", got) } } + +// TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBucket: the indexer starts with the stream PREVIEWS +// and its durable consumer (made by the controller from the checker's declaration and the indexer's consumes) and no bucket +// mesh-delivery-checker_previews; a pull request of mesh-catalog touching avahi makes the candidate with the reason +// 'no merge preview is computed'; the checker then makes its bucket, writes the preview of that head onto the trunk head +// the source names, and says so on its stream; the candidate's avahi gets, within 5 s and with no restart, the reason +// computed from the preview (the toolchain is not reported here, so 'toolchain not yet reported'). +func TestIndexerGivesACandidateItsFingerprintWhenTheCheckerCreatesThePreviewsBucket(t *testing.T) { // 1445/AC4 + manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json") + if err != nil { + t.Fatal(err) + } + const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381" + const inputs = `{"Module":"avahi","Tree":"oldtree","Bases":null,"Contexts":null,"Toolchains":{"code":"go sha256:ccc 0123456789abcdef"},"Unpinned":""}` + const trunk, head = "5555555555555555555555555555555555555555", "0123456789abcdef0123456789abcdef01234567" + js, ctx, candidates, _ := startIndexer(t, func(ctx context.Context, builds jetstream.KeyValue) { + rec := `{"id":"build-1","repository":"novox/mesh-catalog","ref":"main","module":"avahi","commit":"7a74cee83f88e6585a5511c3cc53bdf0f5303d20","path":"modules/avahi","at":"2026-10-19T09:38:00Z","source-inputs":` + inputs + `}` + if _, err := builds.Put(ctx, delivery.BuildKey("avahi", "build-1"), []byte(rec)); err != nil { + t.Fatal(err) + } + if _, err := builds.Put(ctx, delivery.NewestKey("avahi"), []byte("build-1")); err != nil { + t.Fatal(err) + } + }, func(ctx context.Context, js jetstream.JetStream) { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: PreviewsStream, Subjects: []string{previewWrittenSubject}}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateOrUpdateConsumer(ctx, PreviewsStream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-indexer", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + source, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.SourceBucket, History: 4}) + if err != nil { + t.Fatal(err) + } + data, err := delivery.EncodeSource(delivery.Source{Group: "g", Heads: map[string]string{"novox/mesh-catalog": trunk}}) + if err != nil { + t.Fatal(err) + } + if _, err := source.Put(ctx, delivery.SourceMainKey, data); err != nil { + t.Fatal(err) + } + }) + pull, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgePull, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 301, Head: head, Action: delivery.PullUpdated, + Dirs: []string{"modules/avahi"}, DirsKnown: true, Manifests: map[string]string{"modules/avahi": string(manifest)}}) + if err != nil { + t.Fatal(err) + } + if _, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-catalog.pull", pull); err != nil { + t.Fatal(err) + } + none2 := func(want string) string { + var got string + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if e, err := candidates.Get(ctx, "feat/x"); err == nil { + if c, err := delivery.DecodeCandidate(e.Value()); err == nil && len(c.Pulls) == 1 { + got = c.Pulls[0].Fps["avahi"].None2 + if got == want { + break + } + } + } + } + return got + } + if got := none2("no merge preview is computed"); got != "no merge preview is computed" { + t.Fatalf("none2 = %q within 5 s of the pull request, want 'no merge preview is computed'", got) + } + // The checker's bucket is made only now, as when the checker is assigned after the indexer started. + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.PreviewsBucket, History: delivery.PreviewsHistory}) + if err != nil { + t.Fatal(err) + } + pv := delivery.Preview{Repository: "novox/mesh-catalog", Head: head, Trunk: trunk, Merge: "6666666666666666666666666666666666666666", + Modules: map[string]delivery.PreviewModule{"modules/avahi": {Tree: "newtree", Manifest: delivery.ManifestDigest(manifest), Locks: map[string]string{"go.sum": goSum}}}} + key, _ := delivery.PreviewKey(pv.Repository, pv.Head, pv.Trunk) + data, err := delivery.EncodePreview(pv) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Create(ctx, key, data); err != nil { + t.Fatal(err) + } + if _, err := js.Publish(ctx, previewWrittenSubject, []byte(`{"key":"`+key+`"}`)); err != nil { + t.Fatal(err) + } + if got := none2(delivery.ToolchainNoReport); got != delivery.ToolchainNoReport { + t.Errorf("none2 = %q within 5 s of the preview, want %q (computed from the preview)", got, delivery.ToolchainNoReport) + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 846879f..54df569 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -309,6 +309,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := tools.start(); err != nil { return err } + // The checker's merge previews give the candidates their fingerprints: the bucket is watched where it exists, and + // otherwise when the checker says it wrote a preview. + previews := newPreviewWatch(ctx, js, ix, logger) + if err := previews.ensure(); err != nil { + return err + } // Only now, with the toolchains read, are the builds heard to compute from (a build before it would write every Go // module as having no toolchain reported, and back); the builds heard until now are computed once. started.Store(ix) @@ -332,6 +338,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer tc.Stop() } + switch pcons, err := js.Consumer(ctx, PreviewsStream, cred.Node+"_"+cred.Module); { + case err != nil: + logger.Printf("the consumer %s_%s on %s does not exist yet, so the checker's previews are not heard until the indexer starts again: %v", cred.Node, cred.Module, PreviewsStream, err) + default: + pc, err := pcons.Consume(func(m jetstream.Msg) { previews.Written(m) }, + jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + if err != nil { + return err + } + defer pc.Stop() + } cons, err := js.Consumer(ctx, forgeStream, cred.Node+"_"+cred.Module) if err != nil { return fmt.Errorf("the consumer %s_%s on %s: %w", cred.Node, cred.Module, forgeStream, err) @@ -351,5 +368,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err case err := <-tools.errs: return err + case err := <-previews.errs: + return err } } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go new file mode 100644 index 0000000..fcf6a31 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go @@ -0,0 +1,88 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "strings" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-indexer/internal/index" +) + +// PreviewsStream is the stream the checker carries its preview.written event on, as the bus names it (declared by the +// checker as the stream previews, heard by this module through consumes; novox/hq ADR 0484 clause 3). +const PreviewsStream = "PREVIEWS" + +// previewWrittenSubject is the subject of the checker's preview.written event. +const previewWrittenSubject = "mesh.stream." + delivery.CheckerModule + ".event.preview.written" + +// previewWatch reads the merge previews the checker writes (mesh-delivery-checker_previews) and follows them: every change +// writes the candidates' fingerprints again (index.Indexer.ApplyPreview). The checker's bucket is made when it is assigned, +// after the indexer started, and the indexer holds no timer (ADR 0437 ยง8): it starts with the bucket absent, and the +// checker says each time it wrote a preview (preview.written, handled by Written); the bucket is then watched and settled +// as the others, with no restart. +type previewWatch struct { + ctx context.Context + js bucketOpener + ix *index.Indexer + logger *log.Logger + errs chan error + mu sync.Mutex + watched bool +} + +func newPreviewWatch(ctx context.Context, js bucketOpener, ix *index.Indexer, logger *log.Logger) *previewWatch { + return &previewWatch{ctx: ctx, js: js, ix: ix, logger: logger, errs: make(chan error, 1)} +} + +// ensure watches the previews bucket if it is not watched and exists, settling what it holds, then writes the candidates' +// fingerprints from it. +func (w *previewWatch) ensure() error { + w.mu.Lock() + defer w.mu.Unlock() + if w.watched { + return nil + } + kv, err := w.js.KeyValue(w.ctx, delivery.PreviewsBucket) + switch { + case errors.Is(err, jetstream.ErrBucketNotFound): + w.logger.Printf("the bucket %s does not exist yet: it is watched when the checker says it wrote a preview", delivery.PreviewsBucket) + return nil + case err != nil: + return fmt.Errorf("the bucket %s: %w", delivery.PreviewsBucket, err) + } + watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.ix.PutPreview(e.Key(), e.Value(), !opOf(e)) }) + if err != nil { + return err + } + w.watched = true + go func() { + w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) { + if err := w.ix.ApplyPreview(w.ctx, e.Key(), e.Value(), !opOf(e)); err != nil { + w.logger.Printf("%s.%s could not be taken into the candidates: %v", delivery.PreviewsBucket, e.Key(), err) + } + }) + }() + return w.ix.RecomputeCandidates(w.ctx) +} + +// Written takes one preview.written event of the checker in: its bucket is watched if it is not. A subject that is not the +// checker's is terminated; a failure stays unacknowledged, and the server delivers it again. +func (w *previewWatch) Written(m delivered) { + if !strings.HasPrefix(m.Subject(), previewWrittenSubject) { + w.logger.Printf("%s: not the checker's preview.written, dropped", m.Subject()) + _ = m.Term() + return + } + if err := w.ensure(); err != nil { + w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err) + return + } + _ = m.Ack() +} diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index c7b22bb..94177a4 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021100209-cb54756c3865 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021185947-afbe371003a5 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index 639aa71..7c963a0 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021100209-cb54756c3865 h1:VLir9NZbyi1LdH1YwWDIVRuL+lDC+AYlm6FZPlvmlm4= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021100209-cb54756c3865/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021185947-afbe371003a5 h1:1BiPKRPTYpgex8rSHWT9az0BteOafGWRDzd/Ba6GAcA= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021185947-afbe371003a5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 80ab566..2f1bf70 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -48,8 +48,9 @@ type Indexer struct { Builds *Builds Logf func(format string, args ...any) - mu sync.Mutex // guards source and toolchains, which the watches and the forge messages all reach + mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach source sourceNow + previews map[string]delivery.Preview // by previews key (previews.go) toolchains map[string]map[string]delivery.Toolchain // language, then node (toolchains.go) recompute sync.Mutex // one Recompute at a time } @@ -102,6 +103,7 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { } next := ix.next(cur, f, digests) if next != nil { + ix.withPreviews(ctx, next) if err := ix.refuseDisagreement(*next); err != nil { return nil, false, err } diff --git a/modules/mesh-delivery-indexer/internal/index/previews.go b/modules/mesh-delivery-indexer/internal/index/previews.go new file mode 100644 index 0000000..2aedb7d --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/previews.go @@ -0,0 +1,247 @@ +package index + +import ( + "context" + "fmt" + "maps" + "slices" + "strings" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The merge previews the checker writes (mesh-delivery-checker_previews; novox/hq ADR 0484 clause 3, issue 1445): a pull +// request's head merged onto its trunk's head, with the tree, the manifest digest and the lock hashes of each module +// directory. From one the indexer gives the candidate's touched module its new-form fingerprint, by the rules a module of +// the trunk is given its own (Recompute), or says why it has none. + +// The reasons a candidate's touched module has no fingerprint of the new form while a preview exists or is missing. +const ( + // NonePreviewStale is for a module whose preview was made onto a trunk head that is not the trunk's head now. + NonePreviewStale = "the merge preview is out of date: the trunk moved since it was made" + // NonePreviewNoModule is for a module whose directory the preview names no manifest for. + NonePreviewNoModule = "the merge preview names no manifest for the module's directory" + // NonePreviewNoManifest is for a module whose manifest at the merge the indexer does not hold. + NonePreviewNoManifest = "the manifest of the merge is not kept" + // NoneNoDirectory is for a module whose directory in its repository the builds do not tell. + NoneNoDirectory = "the module's directory in its repository is not known" +) + +// PutPreview takes a change of the previews bucket in without writing any candidate: a put of a preview (key +// delivery.PreviewKey) or its delete. It answers whether the previews held changed. A key or a value that is not a +// preview is said in the log and left out. +func (ix *Indexer) PutPreview(key string, value []byte, deleted bool) bool { + if _, _, _, err := delivery.ParsePreviewKey(key); err != nil { + ix.logf("previews.%s is not a preview key, left out: %v", key, err) + return false + } + ix.mu.Lock() + defer ix.mu.Unlock() + if deleted { + if _, had := ix.previews[key]; !had { + return false + } + delete(ix.previews, key) + return true + } + pv, err := delivery.DecodePreview(value) + if err != nil { + ix.logf("previews.%s is not a preview, left out: %v", key, err) + return false + } + if ix.previews == nil { + ix.previews = map[string]delivery.Preview{} + } + if cur, had := ix.previews[key]; had && delivery.SamePreview(cur, pv) { + return false + } + ix.previews[key] = pv + return true +} + +// ApplyPreview takes one change of the previews in (PutPreview) and, when the previews held changed, writes the +// candidates' fingerprints again (RecomputeCandidates). +func (ix *Indexer) ApplyPreview(ctx context.Context, key string, value []byte, deleted bool) error { + if !ix.PutPreview(key, value, deleted) { + return nil + } + return ix.RecomputeCandidates(ctx) +} + +// RecomputeCandidates writes again every candidate whose pull requests' fingerprints, from the previews held, are no longer +// what it holds. A candidate is written only when it differs, by compare-and-set on the revision read. +func (ix *Indexer) RecomputeCandidates(ctx context.Context) error { + if ix.Store == nil { + return nil + } + ix.mu.Lock() + none := len(ix.previews) == 0 + ix.mu.Unlock() + if none { + return nil // no preview held: every candidate has what the pull messages gave it + } + keys, err := ix.Store.Keys(ctx) + if err != nil { + return fmt.Errorf("listing the candidates: %w", err) + } + for _, key := range keys { + err := write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) { + if !found { + return nil, false, nil + } + c, err := delivery.DecodeCandidate(value) + if err != nil { + ix.logf("candidates.%s holds what is not a candidate, left: %v", key, err) + return nil, false, nil + } + if !ix.withPreviews(ctx, &c) { + return nil, false, nil + } + data, err := delivery.EncodeCandidate(c) + return data, false, err + }) + if err != nil { + return fmt.Errorf("candidates.%s: %w", key, err) + } + } + return nil +} + +// previewOf is the preview of a head merged onto the trunk head the source names now: found, or stale when only a preview +// onto another trunk head exists, or neither (found false, stale false) when the source is not known or none exists. +func (ix *Indexer) previewOf(repository, head string) (pv delivery.Preview, found, stale bool) { + ix.mu.Lock() + defer ix.mu.Unlock() + trunk := ix.source.heads[repository] + if trunk == "" { + return delivery.Preview{}, false, false + } + if key, err := delivery.PreviewKey(repository, head, trunk); err == nil { + if pv, ok := ix.previews[key]; ok { + return pv, true, false + } + } + for _, pv := range ix.previews { + if pv.Repository == repository && pv.Head == head { + return delivery.Preview{}, false, true + } + } + return delivery.Preview{}, false, false +} + +// withPreviews gives each touched module of the candidate's pull requests the fingerprint its preview gives it, or the +// reason it has none, and the candidate its fpset. It answers whether the candidate changed. A module of a pull request with +// no preview keeps what it had. +func (ix *Indexer) withPreviews(ctx context.Context, c *delivery.Candidate) bool { + changed := false + for i := range c.Pulls { + p := &c.Pulls[i] + pv, found, stale := ix.previewOf(p.Repository, p.Head.Head) + if !found && !stale { + continue + } + dirs := map[string]string{} + for _, k := range ix.Builds.Modules(p.Repository) { + dirs[k.Module] = k.Path + } + fps := maps.Clone(p.Fps) + for _, m := range p.Modules { + next := delivery.CandidateModule{Manifest: p.Fps[m].Manifest} + switch { + case stale: + next.None2 = NonePreviewStale + default: + dir, known := dirs[m] + if !known { + next.None2 = NoneNoDirectory + break + } + next = ix.candidateModule(ctx, m, dir, pv) + } + if fps == nil { + fps = map[string]delivery.CandidateModule{} + } + if fps[m] != next { + fps[m] = next + changed = true + } + } + if changed || !maps.Equal(fps, p.Fps) { + p.Fps = fps + } + } + if changed { + c.FPSet = delivery.FingerprintSet(c.Pulls) + } + return changed +} + +// candidateModule is the fingerprint of module m, whose directory is dir, at the preview: the module's inputs from the +// merge (tree, manifest, locks), its bases' and the builder's fingerprints as the trunk's modules hold them, and the toolchains +// the build modules agree on, by the rules Recompute applies to a module of the trunk. +func (ix *Indexer) candidateModule(ctx context.Context, m, dir string, pv delivery.Preview) delivery.CandidateModule { + pm, ok := pv.Modules[dir] + if !ok && dir == "" { + pm, ok = pv.Modules[delivery.RootPath] + } + if !ok { + return delivery.CandidateModule{None2: NonePreviewNoModule} + } + out := delivery.CandidateModule{Manifest: pm.Manifest} + if ix.Manifests == nil { + out.None2 = NonePreviewNoManifest + return out + } + value, _, found, err := ix.Manifests.Get(ctx, pm.Manifest) + if err != nil || !found { + out.None2 = NonePreviewNoManifest + return out + } + kept, err := delivery.DecodeManifest(value) + if err != nil { + out.None2 = NonePreviewNoManifest + return out + } + in := inputsOf(m, kept.JSON, pm.Tree, pm.Locks, ix.baseFP(ctx)) + if in.Kept == nil { + out.None2 = in.None2 + return out + } + if in.None2 != "" && in.None2 != delivery.ToolchainNoReport && in.None2 != None2Builder { + out.None2 = in.None2 // a context, or a base with no fingerprint + return out + } + tools := ix.agreed() + toolchains := map[string]delivery.Toolchain2{} + for _, artifact := range slices.Sorted(maps.Keys(in.Languages)) { + t := tools[in.Languages[artifact]] + if t.none != "" { + out.None2 = t.none + return out + } + toolchains[artifact] = t.tc + } + inputs := *in.Kept + if m == BuilderModule { + src, none := ix.builderInput() + if src == "" { + out.None2 = none + return out + } + inputs.Builder = src + } else { + fp, none := ix.baseFP(ctx)(BuilderModule) + if fp == "" { + out.None2 = strings.TrimSuffix(None2Builder+": "+none, ": ") + return out + } + inputs.Builder = fp + } + fp, none := delivery.SourceFingerprint2(inputs.With(m, toolchains)) + if fp == "" { + out.None2 = none + return out + } + out.FP2 = fp + return out +} diff --git a/modules/mesh-delivery-indexer/internal/index/previews_test.go b/modules/mesh-delivery-indexer/internal/index/previews_test.go new file mode 100644 index 0000000..f5484e6 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/previews_test.go @@ -0,0 +1,181 @@ +package index + +import ( + "context" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +const ( + trunkNow = "5555555555555555555555555555555555555555" + trunkOld = "4444444444444444444444444444444444444444" + mergeSha = "6666666666666666666666666666666666666666" +) + +// A trunk of the catalogue with avahi (and the builder, a base and others) merged, the toolchain agreed, and the gate's source +// naming the catalogue's trunk head: avahi has a fingerprint of the new form at tree-avahi. +func previewBed(t *testing.T, report bool) (*Indexer, *memStore, *memStore) { + t.Helper() + ix, mods := newFormIndexer(t, builderModules(t)...) + if report { + reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1") + } + if err := ix.ApplySource(context.Background(), 5, delivery.Source{Group: "g", Heads: map[string]string{catalogueRepo: trunkNow}}); err != nil { + t.Fatal(err) + } + return ix, mods, ix.Store.(*memStore) +} + +// avahiOnTools is avahi's manifest as builderModules has it: standing on mesh-tools. +func avahiOnTools(t *testing.T) string { + return standingOn(t, catalogueManifest(t, "avahi"), "mesh-tools") +} + +// avahiPull is the pull message of the catalogue's pull request 301 touching avahi, with its manifest. +func avahiPull(t *testing.T, head string) delivery.Forge { + t.Helper() + f := manifestPull(t, 301, head, avahiOnTools(t)) + f.Manifests = map[string]string{"modules/avahi": avahiOnTools(t)} + return f +} + +func avahiPreview(t *testing.T, trunk, head string) delivery.Preview { + return delivery.Preview{Repository: catalogueRepo, Head: head, Trunk: trunk, Merge: mergeSha, Modules: map[string]delivery.PreviewModule{ + "modules/avahi": {Tree: "tree-avahi", Manifest: delivery.ManifestDigest([]byte(avahiOnTools(t))), Locks: map[string]string{"go.sum": avahiGoSum}}}} +} + +func applyPreview(t *testing.T, ix *Indexer, pv delivery.Preview) { + t.Helper() + key, err := delivery.PreviewKey(pv.Repository, pv.Head, pv.Trunk) + if err != nil { + t.Fatal(err) + } + data, err := delivery.EncodePreview(pv) + if err != nil { + t.Fatal(err) + } + if err := ix.ApplyPreview(context.Background(), key, data, false); err != nil { + t.Fatal(err) + } +} + +func fpOf(t *testing.T, st *memStore) delivery.CandidateModule { + t.Helper() + c := candidateOf(t, st) + if len(c.Pulls) != 1 { + t.Fatalf("candidate = %+v", c) + } + return c.Pulls[0].Fps["avahi"] +} + +func TestACandidatesModuleGetsTheFingerprintATrunkModuleOfTheSameTreeHas(t *testing.T) { // 1445/AC2 + ix, mods, cands := previewBed(t, true) + trunk, _ := moduleOf(t, mods, "avahi") + if trunk.FP2 == "" { + t.Fatalf("the trunk's avahi has no fingerprint of the new form: %+v", trunk) + } + if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil { + t.Fatal(err) + } + if got := fpOf(t, cands); got.None2 != NoneNoPreview { + t.Fatalf("before the preview: %+v", got) + } + applyPreview(t, ix, avahiPreview(t, trunkNow, h1)) + got := fpOf(t, cands) + if got.FP2 != trunk.FP2 || got.None2 != "" || got.Manifest != delivery.ManifestDigest([]byte(avahiOnTools(t))) { + t.Fatalf("candidate avahi = %+v, want the trunk's fingerprint %s", got, trunk.FP2) + } + if c := candidateOf(t, cands); c.FPSet != delivery.FingerprintSet(c.Pulls) || c.FPSet == "" { + t.Errorf("fpset = %q", c.FPSet) + } + // Another tree gives another fingerprint. + pv := avahiPreview(t, trunkNow, h1) + m := pv.Modules["modules/avahi"] + m.Tree = "tree-changed" + pv.Modules["modules/avahi"] = m + pv.Merge = "7777777777777777777777777777777777777777" + applyPreview(t, ix, pv) + if other := fpOf(t, cands); other.FP2 == "" || other.FP2 == trunk.FP2 { + t.Errorf("a changed tree: %+v", other) + } +} + +func TestAnUnknownToolchainLeavesTheCandidatesModuleNamingIt(t *testing.T) { // 1445/AC2 + ix, _, cands := previewBed(t, false) + if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil { + t.Fatal(err) + } + applyPreview(t, ix, avahiPreview(t, trunkNow, h1)) + if got := fpOf(t, cands); got.FP2 != "" || !strings.Contains(got.None2, delivery.ToolchainNoReport) { + t.Errorf("candidate avahi = %+v, want none2 naming %q", got, delivery.ToolchainNoReport) + } +} + +func TestAPreviewMadeOntoAnOldTrunkIsOutOfDate(t *testing.T) { // 1445/AC3 + ix, _, cands := previewBed(t, true) + if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil { + t.Fatal(err) + } + applyPreview(t, ix, avahiPreview(t, trunkOld, h1)) + if got := fpOf(t, cands); got.FP2 != "" || got.None2 != NonePreviewStale { + t.Errorf("candidate avahi = %+v, want it out of date", got) + } + // A preview onto the trunk head now makes it good again, and the trunk moving again makes it stale. + applyPreview(t, ix, avahiPreview(t, trunkNow, h1)) + if got := fpOf(t, cands); got.FP2 == "" { + t.Fatalf("after the right preview: %+v", got) + } + if err := ix.ApplySource(context.Background(), 6, delivery.Source{Group: "g", Heads: map[string]string{catalogueRepo: "8888888888888888888888888888888888888888"}}); err != nil { + t.Fatal(err) + } + if got := fpOf(t, cands); got.None2 != NonePreviewStale { + t.Errorf("after the trunk moved: %+v", got) + } +} + +func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsWritten(t *testing.T) { // 1445/AC2 + ix, mods, cands := previewBed(t, true) + applyPreview(t, ix, avahiPreview(t, trunkNow, h1)) + if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil { + t.Fatal(err) + } + trunk, _ := moduleOf(t, mods, "avahi") + if got := fpOf(t, cands); got.FP2 != trunk.FP2 { + t.Errorf("candidate avahi = %+v, want %s", got, trunk.FP2) + } + // The same pull request at a new head has no preview of it yet. + if err := ix.Apply(context.Background(), avahiPull(t, h2)); err != nil { + t.Fatal(err) + } + if got := fpOf(t, cands); got.FP2 != "" || got.None2 != NoneNoPreview { + t.Errorf("a new head: %+v, want no preview", got) + } +} + +func TestAPreviewNamingNoManifestOfTheModulesDirectoryGivesAReason(t *testing.T) { + ix, _, cands := previewBed(t, true) + if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil { + t.Fatal(err) + } + pv := avahiPreview(t, trunkNow, h1) + pv.Modules = map[string]delivery.PreviewModule{"modules/other": pv.Modules["modules/avahi"]} + applyPreview(t, ix, pv) + if got := fpOf(t, cands); got.None2 != NonePreviewNoModule || !strings.Contains(NonePreviewNoModule, "manifest") { + t.Errorf("candidate avahi = %+v", got) + } +} + +func TestTheSamePreviewAppliedAgainWritesNothing(t *testing.T) { + ix, _, cands := previewBed(t, true) + if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil { + t.Fatal(err) + } + applyPreview(t, ix, avahiPreview(t, trunkNow, h1)) + w := cands.writes + applyPreview(t, ix, avahiPreview(t, trunkNow, h1)) + if cands.writes != w { + t.Errorf("candidates written %d times for the same preview, want none", cands.writes-w) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/source.go b/modules/mesh-delivery-indexer/internal/index/source.go index 19a5b7b..2235218 100644 --- a/modules/mesh-delivery-indexer/internal/index/source.go +++ b/modules/mesh-delivery-indexer/internal/index/source.go @@ -44,7 +44,7 @@ func (ix *Indexer) ApplySource(ctx context.Context, rev uint64, s delivery.Sourc ix.source = sourceNow{rev: rev, heads: maps.Clone(s.Heads)} ix.mu.Unlock() if ix.Modules == nil { - return nil + return ix.RecomputeCandidates(ctx) } repositories := slices.Sorted(maps.Keys(s.Heads)) for _, repository := range repositories { @@ -76,5 +76,6 @@ func (ix *Indexer) ApplySource(ctx context.Context, rev uint64, s delivery.Sourc } } } - return nil + // A trunk that moved makes the previews made onto the old one out of date. + return ix.RecomputeCandidates(ctx) } diff --git a/modules/mesh-delivery-indexer/internal/index/toolchains.go b/modules/mesh-delivery-indexer/internal/index/toolchains.go index 9b4dc04..ab29381 100644 --- a/modules/mesh-delivery-indexer/internal/index/toolchains.go +++ b/modules/mesh-delivery-indexer/internal/index/toolchains.go @@ -134,6 +134,14 @@ func (ix *Indexer) Recompute(ctx context.Context) error { return ix.recomputeWit // of what the bucket holds, are computed with the others, and are written once, with their final fingerprint or reason // (issue 1444: a merge writes each module once). A record of the overlay that has no kept inputs is written as it is. func (ix *Indexer) recomputeWith(ctx context.Context, overlay map[string]delivery.Module) error { + if err := ix.recomputeModules(ctx, overlay); err != nil { + return err + } + // What the modules moved to reaches the candidates' fingerprints too (a base, the builder, a toolchain). + return ix.RecomputeCandidates(ctx) +} + +func (ix *Indexer) recomputeModules(ctx context.Context, overlay map[string]delivery.Module) error { if ix.Modules == nil { return nil } diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 90d96c3..055782f 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -79,12 +79,12 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } - if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.builds", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { + if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } // "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control // internal/broker/state.go:62-93). - if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-forge-bridge.forge.**"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-delivery-checker.preview.written", "mesh-delivery-forge-bridge.forge.**"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index fcfe983..cb58b8b 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -5,6 +5,7 @@ "consumes": [ "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", + "mesh-delivery-checker.preview.written", "mesh-delivery-forge-bridge.forge.**" ], "state": [ @@ -16,6 +17,7 @@ "mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.builds", + "mesh-delivery-checker.previews", "mesh-delivery-gate.source" ], "own-secrets": {