Rows for ADR 0640 §5: additions change only subjects and filters, on a real bus too (issue 1921)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i

modules/mesh-grants/cmd/mesh-grants/integration_test.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -194,6 +195,62 @@ func TestTheGrantsPartWhole(t *testing.T) {
 			t.Errorf("the bucket %s: %v, want none", bucket, err)
 		}
 	}
+	// ADR 0640 §5: the real additions change the subjects and the filters and nothing else, and recreate nothing.
+	t.Run("adding subjects and filters changes only those", func(t *testing.T) {
+		stream, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ADDS", Subjects: []string{"a.>"}, MaxAge: time.Hour, MaxMsgs: 100,
+			Retention: jetstream.LimitsPolicy, Discard: jetstream.DiscardNew})
+		if err != nil {
+			t.Fatal(err)
+		}
+		cons, err := stream.CreateConsumer(ctx, jetstream.ConsumerConfig{Durable: "idx", AckPolicy: jetstream.AckExplicitPolicy, FilterSubject: "a.one", MaxDeliver: 7})
+		if err != nil {
+			t.Fatal(err)
+		}
+		for range 3 {
+			if _, err := js.Publish(ctx, "a.one", []byte("x")); err != nil {
+				t.Fatal(err)
+			}
+		}
+		got, err := cons.Fetch(2)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for m := range got.Messages() {
+			_ = m.Ack()
+		}
+		before, _ := stream.Info(ctx)
+		cbefore, _ := cons.Info(ctx)
+		if cbefore.Delivered.Stream != 2 {
+			t.Fatalf("the consumer delivered %d, want 2", cbefore.Delivered.Stream)
+		}
+		c := creating{ctx: ctx, js: js}
+		if err := c.AddSubjects("ADDS", []string{"b.>"}); err != nil {
+			t.Fatal(err)
+		}
+		if err := c.AddFilters("ADDS", "idx", []string{"a.two"}); err != nil {
+			t.Fatal(err)
+		}
+		after, _ := stream.Info(ctx)
+		cafter, _ := cons.Info(ctx)
+		b, a := before.Config, after.Config
+		if !slices.Equal(a.Subjects, []string{"a.>", "b.>"}) || a.MaxAge != b.MaxAge || a.MaxMsgs != b.MaxMsgs || a.Retention != b.Retention ||
+			a.Discard != b.Discard || !after.Created.Equal(before.Created) || after.State.Msgs != before.State.Msgs || after.State.LastSeq != before.State.LastSeq {
+			t.Errorf("the stream after: %+v created %v, before %+v created %v", a, after.Created, b, before.Created)
+		}
+		cb, ca := cbefore.Config, cafter.Config
+		if !slices.Equal(ca.FilterSubjects, []string{"a.one", "a.two"}) || ca.FilterSubject != "" || ca.MaxDeliver != cb.MaxDeliver || ca.AckPolicy != cb.AckPolicy ||
+			cafter.Delivered.Stream != cbefore.Delivered.Stream || cafter.AckFloor.Stream != cbefore.AckFloor.Stream || !cafter.Created.Equal(cbefore.Created) {
+			t.Errorf("the consumer after: %+v delivered %d created %v, before %+v delivered %d created %v", ca, cafter.Delivered.Stream, cafter.Created, cb, cbefore.Delivered.Stream, cbefore.Created)
+		}
+		// Nothing is removed: a second addition of the same and of a subset keeps what is there.
+		if err := c.AddSubjects("ADDS", []string{"a.>"}); err != nil {
+			t.Fatal(err)
+		}
+		if again, _ := stream.Info(ctx); !slices.Equal(again.Config.Subjects, []string{"a.>", "b.>"}) {
+			t.Errorf("subjects after adding a held one: %v", again.Config.Subjects)
+		}
+	})
+
 	// And what the part composes holds on a server that loads it (one integration test per module, the test budget).
 	t.Run("the composed grants hold on a server that loads them", theBusNodeFollowsTheSealedListUnderTheComposedGrants)
 }

modules/mesh-grants/grants_test.go

@@ -793,7 +793,15 @@ func TestGrants(t *testing.T) {
 		b.assign("a", "reader")
 		b.put(grants.ReadResolved, "a.mesh-mail.grants", map[string]any{"node": "a", "module": "mesh-mail", "key": "grants", "value": map[string]any{"box": []string{"reader"}}})
 		p := partOn(b)
+		p.Consumers = b
+		b.consumers = map[string][]map[string]any{"EVENTS": {{"name": "a_reader", "filters": []string{"old"}, "max_deliver": 5}}, "MAIL": {{"name": "a_reader", "filters": []string{"old"}, "max_deliver": 5}}}
 		res := p.Compose()
+		// A restricted stream's consumer gains only the filters its grant allows: the hearing is withheld, so none of the stream's.
+		for _, add := range b.filterAdds {
+			if strings.Contains(add, "mesh-mail") {
+				t.Errorf("a withheld hearing's filter was added: %v", b.filterAdds)
+			}
+		}
 		const key = "reader.mail.a"
 		var h grants.WithheldHearing
 		if res.Withheld == nil || !slices.Equal(res.Withheld.Written, []string{key}) || json.Unmarshal(b.own[grants.StateWithheld][key], &h) != nil ||
@@ -808,6 +816,11 @@ func TestGrants(t *testing.T) {
 		if heard := p.Compose(); heard.Withheld == nil || !slices.Equal(heard.Withheld.Deleted, []string{key}) || b.own[grants.StateWithheld][key] != nil {
 			t.Errorf("once reader runs as its own account the pass did %+v (errors %v), want %s deleted", heard.Withheld, heard.Errors, key)
 		}
+		for _, add := range b.filterAdds { // and once granted, still only what the records compose for the account's consumer
+			if strings.Contains(add, "mesh-mail") || !strings.Contains(add, "+mesh.") {
+				t.Errorf("a filter the records do not compose was added: %v", b.filterAdds)
+			}
+		}
 	})
 
 	// ADR 0601 clause 3: a held module's seats, hearings and reads are composed from the manifest its running build was judged
@@ -1881,7 +1894,7 @@ func TestGrants(t *testing.T) {
 	t.Run("assertions: a held stream gains the subjects its manifest names, and loses, changes nothing else", func(t *testing.T) {
 		b := fixture()
 		b.loaded = shadowOf(t, b)
-		b.streams = []map[string]any{{"name": "THINGS", "subjects": []string{"dropped.subject"}, "retention": "limits", "max_age_hours": 48}}
+		b.streams = []map[string]any{{"name": "THINGS", "subjects": []string{"dropped.subject"}, "retention": "limits", "max_age_hours": 24}}
 		p := partOn(b)
 		p.Create = b
 		var logged []string
@@ -1894,7 +1907,7 @@ func TestGrants(t *testing.T) {
 		}
 		for _, s := range b.streams {
 			if s["name"] == "THINGS" {
-				if got := s["subjects"].([]string); !slices.Contains(got, "dropped.subject") || len(got) < 2 || s["max_age_hours"] != 48 {
+				if got := s["subjects"].([]string); !slices.Contains(got, "dropped.subject") || len(got) < 2 || s["max_age_hours"] != 24 {
 					t.Errorf("THINGS after the addition = %v: a subject the manifest dropped must stay, the limits as they were", s)
 				}
 			}
@@ -1916,6 +1929,23 @@ func TestGrants(t *testing.T) {
 		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)
 		}
+		// The bus refusing for another reason is said again.
+		b.mu.Lock()
+		b.refuseAdd = "permissions violation"
+		b.mu.Unlock()
+		if other := strings.Join(p.Compose().NotAsserted, " "); !strings.Contains(other, "permissions violation") {
+			t.Errorf("a refusal for another reason was not said: %q", other)
+		}
+		// A limit the manifest changed (the held max age is not the manifest's) is not touched; only subjects are ever added.
+		b.mu.Lock()
+		b.refuseAdd = ""
+		b.mu.Unlock()
+		p.Compose()
+		for _, s := range b.streams {
+			if s["name"] == "THINGS" && (s["max_age_hours"] != 24 || s["retention"] != "limits") {
+				t.Errorf("a limit or retention changed with the subjects: %v", s)
+			}
+		}
 	})
 
 	t.Run("account consumers: a held consumer gains the filters of a new event, never loses one", func(t *testing.T) {