diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index 25a157c0..d581eacc 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/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) } diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 08acdb3f..2b6d620f 100644 --- a/modules/mesh-grants/grants_test.go +++ b/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) {