From 1b262125ee8f55779a550196150dcb3e0ad27f25 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 21:32:04 +0100 Subject: [PATCH] Open tries test folded; the consumer update granted for every composed consumer; a standing refusal of an addition said once (issue 1921) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-indexer/named_test.go | 40 +++++++++---------- modules/mesh-grants/grants_test.go | 29 ++++++++++++++ .../internal/buscompose/buscompose_test.go | 5 +++ .../mesh-grants/internal/buscompose/nats.go | 7 ++-- .../mesh-grants/internal/grants/assertions.go | 29 +++++++++++++- .../mesh-grants/internal/grants/consumers.go | 14 ++++++- modules/mesh-grants/internal/grants/part.go | 2 + 7 files changed, 100 insertions(+), 26 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go index 6f80ef0c..4d36c450 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go @@ -89,6 +89,25 @@ func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3 if _, err := ownBuildVia(none, "n1")(context.Background()); err == nil { t.Error("an engine that says nothing runs gave a build") } + // Issue 1921: the checker's list of open tries; none open is an answer, whether [] or null. + for _, row := range []struct { + name, answer string + want int + refused bool + }{ + {"one open", `{"tries":[{"id":"try-1","state":"running","whole":true,"members":[{"repository":"novox/a","head":"h","trunk":"t"}]}]}`, 1, false}, + {"an empty list", `{"tries":[]}`, 0, false}, + {"null, as an older checker answers none", `{"tries":null}`, 0, false}, + {"no tries key", `{}`, 0, true}, + {"not json", `nope`, 0, true}, + {"tries of another shape", `{"tries":{"a":1}}`, 0, true}, + } { + call := func(string, any) (json.RawMessage, error) { return json.RawMessage(row.answer), nil } + got, err := openTriesVia(call)(context.Background()) + if (err != nil) != row.refused || len(got) != row.want { + t.Errorf("%s: %d tries, error %v; want %d, refused %v", row.name, len(got), err, row.want, row.refused) + } + } } func TestReadViaReadsManifestsAndLocks(t *testing.T) { @@ -120,24 +139,3 @@ func TestReadViaReadsManifestsAndLocks(t *testing.T) { t.Fatalf("read = %+v", got) } } - -func TestTheOpenTriesAnswerTakesNoneOpenAsAnAnswer(t *testing.T) { // issue 1921: {"tries": null} refused the recovery for hours - for _, row := range []struct { - name, answer string - want int - refused bool - }{ - {"one open", `{"tries":[{"id":"try-1","state":"running","whole":true,"members":[{"repository":"novox/a","head":"h","trunk":"t"}]}]}`, 1, false}, - {"an empty list", `{"tries":[]}`, 0, false}, - {"null, as an older checker answers none", `{"tries":null}`, 0, false}, - {"no tries key", `{}`, 0, true}, - {"not json", `nope`, 0, true}, - {"tries of another shape", `{"tries":{"a":1}}`, 0, true}, - } { - call := func(string, any) (json.RawMessage, error) { return json.RawMessage(row.answer), nil } - got, err := openTriesVia(call)(context.Background()) - if (err != nil) != row.refused || len(got) != row.want { - t.Errorf("%s: %d tries, error %v; want %d, refused %v", row.name, len(got), err, row.want, row.refused) - } - } -} diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 8126ca37..08acdb3f 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -55,6 +55,7 @@ type fakeBus struct { creates []string // subjectAdds and filterAdds are the additions the part asked of a held stream and a held consumer. subjectAdds, filterAdds []string + refuseAdd string createdAge map[string]time.Duration refuseCreate bool // refuseUsers refuses the bus module's nats_users: the loaded list cannot be read. @@ -109,6 +110,9 @@ func (b *fakeBus) AddSubjects(stream string, add []string) error { b.mu.Lock() defer b.mu.Unlock() b.subjectAdds = append(b.subjectAdds, stream+" +"+strings.Join(add, ",")) + if b.refuseAdd != "" { + return errors.New(b.refuseAdd) + } for _, s := range b.streams { if s["name"] == stream { have, _ := s["subjects"].([]string) @@ -129,6 +133,9 @@ func (b *fakeBus) AddFilters(stream, name string, add []string) error { b.mu.Lock() defer b.mu.Unlock() b.filterAdds = append(b.filterAdds, stream+"."+name+" +"+strings.Join(add, ",")) + if b.refuseAdd != "" { + return errors.New(b.refuseAdd) + } for _, c := range b.consumers[stream] { if c["name"] == name { have, _ := c["filters"].([]string) @@ -1895,6 +1902,20 @@ func TestGrants(t *testing.T) { if again := p.Compose(); len(b.subjectAdds) != 1 || len(again.Asserted) != 0 { t.Errorf("a stream that holds every subject was added to again: %v, %v", b.subjectAdds, again.Asserted) } + // A refusal of the bus (the subject overlaps another stream's) is said at the first pass and tried at the next, said once. + b.mu.Lock() + for _, s := range b.streams { + if s["name"] == "THINGS" { + s["subjects"] = []string{"dropped.subject"} + } + } + b.refuseAdd = "subjects overlap" + b.mu.Unlock() + first := strings.Join(p.Compose().NotAsserted, " ") + second := strings.Join(p.Compose().NotAsserted, " ") + if !strings.Contains(first, "THINGS: subjects") || !strings.Contains(first, "subjects overlap") || second != "" || len(b.subjectAdds) != 3 { + t.Errorf("a standing refusal: first %q, second %q, asked %v: want it said once and asked at every pass", first, second, b.subjectAdds) + } }) t.Run("account consumers: a held consumer gains the filters of a new event, never loses one", func(t *testing.T) { @@ -1916,6 +1937,14 @@ func TestGrants(t *testing.T) { if len(b.filterAdds) != 1 { t.Errorf("a consumer that holds every filter was added to again: %v", b.filterAdds) } + // One filter growing to two, refused by the bus: said once. + b.consumers["EVENTS"][0]["filters"] = []string{"mesh.mod.mod-y.event.done"} + b.refuseAdd = "filters overlap" + first := p.Compose().Consumers.Matched + second := p.Compose().Consumers.Matched + if len(first) != 1 || !strings.Contains(first[0], "filters overlap") || len(second) != 0 || len(b.filterAdds) != 3 { + t.Errorf("a standing refusal of a filter: first %v, second %v, asked %v", first, second, b.filterAdds) + } }) t.Run("assertions: the node-engine's declared state is created though no node is assigned it", func(t *testing.T) { diff --git a/modules/mesh-grants/internal/buscompose/buscompose_test.go b/modules/mesh-grants/internal/buscompose/buscompose_test.go index c57ae568..d5203b3b 100644 --- a/modules/mesh-grants/internal/buscompose/buscompose_test.go +++ b/modules/mesh-grants/internal/buscompose/buscompose_test.go @@ -587,6 +587,11 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) { if err != nil { t.Fatal(err) } + // A consumer composed with one filter is granted the same create-or-update, the form that grows its filters to two. + one := CreateConsumerGrants([]ModuleConsumer{{Consumer: Consumer{Stream: "CHECKS", Name: "a_idx", Filters: []string{"x.preview.written"}}}}) + if !slices.Contains(one, "$JS.API.CONSUMER.CREATE.CHECKS.a_idx") || len(one) != 2 { + t.Errorf("a one-filter consumer's grants = %v: its update is the no-filter create form", one) + } for _, u := range users { if u.Module != GrantsModule || u.Kind != KindModule { continue diff --git a/modules/mesh-grants/internal/buscompose/nats.go b/modules/mesh-grants/internal/buscompose/nats.go index 64cbf73e..90477125 100644 --- a/modules/mesh-grants/internal/buscompose/nats.go +++ b/modules/mesh-grants/internal/buscompose/nats.go @@ -1459,13 +1459,14 @@ func CreateStreamGrants(streams []string) []string { } // CreateConsumerGrants are the subjects that create each consumer by its stream and name, and read its info back: the -// create of the client library's CreateConsumer for a consumer with no filter or several, never one filter's form (a -// wildcard in a filter would widen the grant), an update or a delete. +// create-or-update of the client library's CreateConsumer and UpdateConsumer for a consumer with no filter or several, which is +// the form that adds a filter to a consumer that holds one (grants.ConsumerCreator.AddFilters), granted for every composed +// consumer whatever its filters now are; never one filter's form (a wildcard in a filter would widen the grant), or a delete. func CreateConsumerGrants(consumers []ModuleConsumer) []string { var out []string for _, mc := range consumers { c := mc.Consumer - if !safeSubject.MatchString(c.Stream) || !safeSubject.MatchString(c.Name) || len(c.Filters) == 1 { + if !safeSubject.MatchString(c.Stream) || !safeSubject.MatchString(c.Name) { continue } out = append(out, "$JS.API.CONSUMER.CREATE."+c.Stream+"."+c.Name, "$JS.API.CONSUMER.INFO."+c.Stream+"."+c.Name) diff --git a/modules/mesh-grants/internal/grants/assertions.go b/modules/mesh-grants/internal/grants/assertions.go index 2a55d1d8..33f2085d 100644 --- a/modules/mesh-grants/internal/grants/assertions.go +++ b/modules/mesh-grants/internal/grants/assertions.go @@ -103,6 +103,27 @@ func Configs(m catalogue.Manifest, nodes []string) map[string]jetstream.StreamCo type creations struct { mu sync.Mutex made map[string]string + // refused is, by stream, the refusal of the last addition of subjects (an overlap with another stream, say), so a + // refusal that stands is said once and not at every pass; a success or a different refusal says again. + refused map[string]string +} + +// refusedAgain says whether this refusal of an addition to the stream was the last one said, and keeps it. +func (c *creations) refusedAgain(stream, why string) bool { + c.mu.Lock() + defer c.mu.Unlock() + if c.refused == nil { + c.refused = map[string]string{} + } + again := c.refused[stream] == why + c.refused[stream] = why + return again +} + +func (c *creations) cleared(stream string) { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.refused, stream) } func (c *creations) madeAt(stream string) string { @@ -169,9 +190,15 @@ func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifes continue } if err := c.AddSubjects(it.Stream, add); err != nil { - it.Created = ItemNotCreated + "subjects " + strings.Join(add, ", ") + " not added: " + err.Error() + " (tried again at the next event)" + why := "subjects " + strings.Join(add, ", ") + " not added: " + err.Error() + if done.refusedAgain(it.Stream, why) { + it.Created = why + " (said before; tried again at the next event)" // not ItemNotCreated: said once, not at every pass + } else { + it.Created = ItemNotCreated + why + " (tried again at the next event)" + } continue } + done.cleared(it.Stream) it.Created = "subjects added: " + strings.Join(add, ", ") made = append(made, "stream "+it.Stream+" gained "+strings.Join(add, ", ")) } diff --git a/modules/mesh-grants/internal/grants/consumers.go b/modules/mesh-grants/internal/grants/consumers.go index 268db150..4596eaf9 100644 --- a/modules/mesh-grants/internal/grants/consumers.go +++ b/modules/mesh-grants/internal/grants/consumers.go @@ -146,8 +146,20 @@ func (p *Part) consumers(res *Result, recs buscompose.Records, match string) { // a filter it holds that is no longer composed stays, and nothing else about it changes. if add := filtersToAdd(c, h); len(add) > 0 && st.On && p.Consumers != nil { if err := p.Consumers.AddFilters(stream, c.Name, add); err != nil { - st.Matched = append(st.Matched, key+": filters "+strings.Join(add, ", ")+" not added: "+err.Error()) + // A refusal that stands (an overlap with another filter, say) is said once, not at every pass. + why := key + ": filters " + strings.Join(add, ", ") + " not added: " + err.Error() + if p.filterRefused[key] != why { + st.Matched = append(st.Matched, why) + if p.Logf != nil { + p.Logf("%s", why) + } + } + if p.filterRefused == nil { + p.filterRefused = map[string]string{} + } + p.filterRefused[key] = why } else { + delete(p.filterRefused, key) st.Matched = append(st.Matched, key+": filters added: "+strings.Join(add, ", ")) if p.Logf != nil { p.Logf("consumer %s gained the filters %s", key, strings.Join(add, ", ")) diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index d7be1513..a2e3c634 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -183,6 +183,8 @@ type Part struct { behindSaid string behindKnown bool behindSeen map[string]bool + // filterRefused is, by consumer, the refusal of the last addition of filters, so a standing refusal is said once. + filterRefused map[string]string // fullSaid is the history streams and object buckets past their mark last said as a condition, once fullKnown (full.go). fullSaid string fullKnown bool