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

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

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) {

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

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)

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, ", "))
 	}

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, ", "))

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