Merge pull request 'A group try hears its events: the grants part adds a manifest's new event types, and no open tries is an answer (issue 1921)' (#157) from fix/1921-try-live into main
This commit was merged in pull request #157.
This commit is contained in:
@@ -795,7 +795,7 @@ func OpenTries(store TryStore) ([]tryevent.Open, error) {
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []tryevent.Open
|
||||
out := []tryevent.Open{} // none open is an empty list, never null: a reader tells it from no answer (issue 1921)
|
||||
for _, o := range found {
|
||||
if o != nil {
|
||||
out = append(out, *o)
|
||||
|
||||
@@ -223,6 +223,8 @@ func TestATryIsJudgedAsAGroupsCheckAndOwesTheForgeNothing(t *testing.T) { // ADR
|
||||
}
|
||||
if open, err := OpenTries(memTries{kv}); err != nil || len(open) != 0 {
|
||||
t.Errorf("open tries after the end = %v, %v", open, err)
|
||||
} else if raw, _ := json.Marshal(open); string(raw) != "[]" { // issue 1921: none open is [] on the wire, never null
|
||||
t.Errorf("none open is answered %s, want []", raw)
|
||||
}
|
||||
if got.State != TryPass || got.Took == "" || len(got.Members) != 2 || got.Members[0].Repository != repoA || got.Members[1].Head != head2 ||
|
||||
got.Said == "" || got.Levels[delivery.LevelUnit].Run+got.Levels[delivery.LevelUnit].Kept != 4 { // the bench gives both members the same digests
|
||||
|
||||
@@ -306,12 +306,20 @@ func openTriesVia(call toolCaller) func(ctx context.Context) ([]tryevent.Open, e
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var o struct {
|
||||
Tries *[]tryevent.Open `json:"tries"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &o); err != nil || o.Tries == nil {
|
||||
// No try open is an answer: the checker's list is empty, or null in an older checker (issue 1921). An answer without the
|
||||
// key is none.
|
||||
var o map[string]json.RawMessage
|
||||
if err := json.Unmarshal(raw, &o); err != nil {
|
||||
return nil, fmt.Errorf("the answer of %s is not {tries}: %v", openTriesTool, err)
|
||||
}
|
||||
return *o.Tries, nil
|
||||
list, ok := o["tries"]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("the answer of %s is not {tries}: no tries key", openTriesTool)
|
||||
}
|
||||
var tries []tryevent.Open
|
||||
if err := json.Unmarshal(list, &tries); err != nil {
|
||||
return nil, fmt.Errorf("the answer of %s is not {tries}: %v", openTriesTool, err)
|
||||
}
|
||||
return tries, nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
@@ -67,3 +68,45 @@ func (c creating) CreateDeclared(cfg jetstream.StreamConfig) error {
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// AddSubjects adds subjects to a held stream and changes nothing else: it reads the stream's configuration, sets its
|
||||
// subjects to the union and updates with the rest as it was.
|
||||
func (c creating) AddSubjects(stream string, add []string) error {
|
||||
ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
|
||||
defer cancel()
|
||||
s, err := c.js.Stream(ctx, stream)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cfg := s.CachedInfo().Config
|
||||
for _, a := range add {
|
||||
if !slices.Contains(cfg.Subjects, a) {
|
||||
cfg.Subjects = append(cfg.Subjects, a)
|
||||
}
|
||||
}
|
||||
_, err = c.js.UpdateStream(ctx, cfg)
|
||||
return err
|
||||
}
|
||||
|
||||
// AddFilters adds filter subjects to a held consumer and changes nothing else.
|
||||
func (c creating) AddFilters(stream, name string, add []string) error {
|
||||
ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
|
||||
defer cancel()
|
||||
held, err := c.js.Consumer(ctx, stream, name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cfg := held.CachedInfo().Config
|
||||
filters := slices.Clone(cfg.FilterSubjects)
|
||||
if cfg.FilterSubject != "" {
|
||||
filters = append(filters, cfg.FilterSubject)
|
||||
}
|
||||
for _, a := range add {
|
||||
if !slices.Contains(filters, a) {
|
||||
filters = append(filters, a)
|
||||
}
|
||||
}
|
||||
cfg.FilterSubject, cfg.FilterSubjects = "", filters
|
||||
_, err = c.js.UpdateConsumer(ctx, stream, cfg)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
// that brought the manifest; one the bus holds is never changed. **The user list is written sealed to the bus node**
|
||||
// (ADR 0547 clause 7) as an object of the part's own bucket named in `user-list.sealed`, which the bus node follows
|
||||
// alone since the controller ended by its unassignment (novox/hq ADR 0590, issue 1126). **The grants** the assigner
|
||||
// reads are written, and **the account consumers** made; a stream the bus holds is never changed, a difference only said.
|
||||
// reads are written, and **the account consumers** made; a stream the bus holds gains only the subjects its manifest adds (a consumer the filters), never loses one or changes a limit; any other difference is only said.
|
||||
//
|
||||
// It holds no timer, and never exits for a refused read, write or publish: a refusal is said and tried again at the next
|
||||
// event.
|
||||
|
||||
@@ -102,8 +102,9 @@ func toolsOf(p *grants.Part) []stdio.Tool {
|
||||
{Name: grants.ToolAssert, Description: "One module's buckets, logs and streams (ADR 0547 clause 6), composed from its manifest at " +
|
||||
"the indexer's trunk with their limits, beside what the bus holds (through the bus module's nats_streams). When the " +
|
||||
"bus lacks one, asks the part's process (which alone holds the grant) to create it now with the manifest's limits, " +
|
||||
"as it does by itself when a manifest is registered (issue 1790), and says asserted; one the bus holds is never " +
|
||||
"changed, a difference is only said. Answers the comparison and the ask; call again to see what the bus holds.",
|
||||
"as it does by itself when a manifest is registered (issue 1790), and says asserted. A stream the bus holds gains the " +
|
||||
"subjects its manifest adds (and a consumer the filters its module's consumes adds), only ever adding: it never removes " +
|
||||
"a subject, changes a limit or retention, deletes or recreates one; any other difference is only said. Answers the comparison and the ask; call again to see what the bus holds.",
|
||||
Input: schema(map[string]string{"module": "the module"}, "module"),
|
||||
Run: func(a map[string]any) (any, error) { return p.AssertOne(grants.Answer(a, "module")) }},
|
||||
{Name: grants.ToolAccounts, Description: "The bus accounts the mesh wants (novox/hq ADR 0561 §2, §9): every account the vault " +
|
||||
|
||||
@@ -52,9 +52,12 @@ type fakeBus struct {
|
||||
// hold, when set, keeps every nats_users call waiting until it is closed: a pass waiting on the bus.
|
||||
hold chan struct{}
|
||||
// creates are the streams the part asked to create, refused when refuseCreate.
|
||||
creates []string
|
||||
createdAge map[string]time.Duration
|
||||
refuseCreate bool
|
||||
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.
|
||||
refuseUsers bool
|
||||
// rootFree answers a machine's engine's root-free verb: nil is no answerer at all (no responders).
|
||||
@@ -102,6 +105,47 @@ func (b *fakeBus) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) er
|
||||
return nil
|
||||
}
|
||||
|
||||
// AddSubjects is the bus's addition of subjects to a held stream: the union, every other setting kept.
|
||||
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)
|
||||
for _, a := range add {
|
||||
if !slices.Contains(have, a) {
|
||||
have = append(have, a)
|
||||
}
|
||||
}
|
||||
s["subjects"] = have
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return errors.New("stream not found")
|
||||
}
|
||||
|
||||
// AddFilters is the bus's addition of filters to a held consumer: the union, max deliver and the rest kept.
|
||||
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)
|
||||
c["filters"] = append(slices.Clone(have), add...)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return errors.New("consumer not found")
|
||||
}
|
||||
|
||||
func (b *fakeBus) PublishMembership(node, module string, body []byte) error {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
@@ -474,7 +518,7 @@ func TestGrants(t *testing.T) {
|
||||
|
||||
t.Run("no timer, no update or delete, a publish only of conditions, a create only of declared streams", func(t *testing.T) {
|
||||
forbidden := []string{"time.Tick", "time.NewTicker", "time.Sleep", "time.After", "time.NewTimer", "time.AfterFunc",
|
||||
"PublishMsg", "CreateKeyValue", "CreateOrUpdate", "UpdateStream", "UpdateKeyValue", "DeleteStream", "PurgeStream",
|
||||
"PublishMsg", "CreateKeyValue", "CreateOrUpdate", "UpdateKeyValue", "DeleteStream", "PurgeStream",
|
||||
"DeleteKeyValue", "CreateObjectStore", "EnsureStream", "RaiseBuckets", "RaiseLogs", ".Create("}
|
||||
// The one publish: a condition, on the seat operator-channel's subject, in asking.go; a membership is written to
|
||||
// the part's state alone, never said on its subject (issue 1126). The one create: a declared stream the bus lacks,
|
||||
@@ -509,6 +553,10 @@ func TestGrants(t *testing.T) {
|
||||
if n := strings.Count(string(src), "CreateStream"); (path == createOnly && n != 1) || (path != createOnly && n != 0) {
|
||||
t.Errorf("%s holds CreateStream %d times: only create.go creates, once", path, n)
|
||||
}
|
||||
// The one update of a stream: adding subjects to a held one, in create.go (the operator, 2026-11-05).
|
||||
if n := strings.Count(string(src), "UpdateStream("); (path == createOnly && n != 1) || (path != createOnly && n != 0) {
|
||||
t.Errorf("%s holds UpdateStream %d times: only create.go updates a stream, once, to add subjects", path, n)
|
||||
}
|
||||
// The one object bucket: the part's own, of the user list sealed to the bus node, in sealed.go (ADR 0547 clause 7).
|
||||
if path == sealedOnly {
|
||||
if strings.Count(string(src), "CreateObjectStore(") != 1 || !strings.Contains(string(src), "delivery.GrantsUserListObjects") {
|
||||
@@ -745,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 ||
|
||||
@@ -760,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
|
||||
@@ -1830,6 +1891,92 @@ func TestGrants(t *testing.T) {
|
||||
|
||||
// 774 (ADR 0563 §2, §3): the node-engine's manifest (mesh-host) is placed on every node though no node is assigned it,
|
||||
// so its declared state acts is created as a placed module's, and composed for the user list's create right.
|
||||
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": 24}}
|
||||
p := partOn(b)
|
||||
p.Create = b
|
||||
var logged []string
|
||||
p.Logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) }
|
||||
res := p.Compose()
|
||||
if len(b.subjectAdds) != 1 || !strings.HasPrefix(b.subjectAdds[0], "THINGS +") || strings.Contains(b.subjectAdds[0], "dropped") ||
|
||||
!slices.Contains(b.creates, "LOG_mod-x_history") || slices.Contains(b.creates, "THINGS") || !slices.Contains(b.events, grants.EventAsserted) ||
|
||||
!strings.Contains(strings.Join(res.Asserted, " "), "gained") || !strings.Contains(strings.Join(logged, " "), "THINGS gained") {
|
||||
t.Fatalf("the pass added %v, created %v, said %v, journal %v", b.subjectAdds, b.creates, res.Asserted, logged)
|
||||
}
|
||||
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"] != 24 {
|
||||
t.Errorf("THINGS after the addition = %v: a subject the manifest dropped must stay, the limits as they were", s)
|
||||
}
|
||||
}
|
||||
}
|
||||
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)
|
||||
}
|
||||
// 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) {
|
||||
const cManifest = `{"module":"mod-c","version":"1","own-secrets":{"broker":"/var/lib/c/broker"},"consumes":["mod-y.done","mod-y.more"]}`
|
||||
b := fixture()
|
||||
b.manifestOf("mod-c", cManifest)
|
||||
b.assign("b", "mod-c")
|
||||
b.loaded = shadowOf(t, b)
|
||||
p := partOn(b)
|
||||
p.Consumers = b
|
||||
b.consumers = map[string][]map[string]any{"EVENTS": {{"name": "b_mod-c", "filters": []string{"old.filter", "mesh.mod.mod-y.event.done"}, "max_deliver": 9}}}
|
||||
res := p.Compose()
|
||||
if len(b.filterAdds) != 1 || !strings.Contains(b.filterAdds[0], "mesh.mod.mod-y.event.more") || strings.Contains(b.filterAdds[0], "done") ||
|
||||
!slices.Contains(b.consumers["EVENTS"][0]["filters"].([]string), "old.filter") || b.consumers["EVENTS"][0]["max_deliver"] != 9 ||
|
||||
len(res.Consumers.Matched) != 1 {
|
||||
t.Fatalf("filters added %v, now %v, said %+v", b.filterAdds, b.consumers["EVENTS"], res.Consumers)
|
||||
}
|
||||
p.Compose()
|
||||
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) {
|
||||
b := fixture()
|
||||
b.manifestOf(buscompose.EngineModule, `{"module":"mesh-host","version":"1","state":[{"name":"acts"}]}`)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
// The streams a manifest declares, as the server is asked to create them (novox/hq ADR 0547 §5, issue 1790): the
|
||||
// grants part creates each one the bus does not hold yet, with what the controller asserts at its start, so the two
|
||||
// make the same stream. Pure. Only a create is ever asked with these: a stream the bus holds is never updated by the
|
||||
// grants part (CreateStreamGrants grants no update).
|
||||
// grants part (CreateStreamGrants grants its update only to add subjects).
|
||||
|
||||
// BucketStreamConfig is a module's bucket as the stream the controller's CreateOrUpdateKeyValue makes of its options
|
||||
// (mesh-controller internal/broker BucketConfig, with no marker age held): the client library's key-value shape
|
||||
|
||||
@@ -65,7 +65,8 @@ func theGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
|
||||
{GrantsModule, "$JS.API.STREAM.CREATE.LOG_mesh-delivery-builder_builds", true},
|
||||
{GrantsModule, "$JS.API.STREAM.CREATE.BUILDASKS", true},
|
||||
{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-grants_assertions", true},
|
||||
{GrantsModule, "$JS.API.STREAM.UPDATE.KV_mesh-delivery-builder_steps", false},
|
||||
{GrantsModule, "$JS.API.STREAM.UPDATE.KV_mesh-delivery-builder_steps", true},
|
||||
{GrantsModule, "$JS.API.STREAM.UPDATE.KV_mesh-delivery-builder_undeclared", false},
|
||||
{GrantsModule, "$JS.API.STREAM.DELETE.KV_mesh-delivery-builder_steps", false},
|
||||
{GrantsModule, "$JS.API.STREAM.PURGE.KV_mesh-delivery-builder_steps", false},
|
||||
{GrantsModule, "$JS.API.STREAM.CREATE.KV_mesh-delivery-builder_undeclared", false},
|
||||
@@ -132,7 +133,7 @@ func theGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
|
||||
}
|
||||
}
|
||||
// A name that is no plain token is never granted, and a module other than the grants part is never granted one.
|
||||
if got := CreateStreamGrants([]string{"*", ">", "KV_a.b", "KV_a_b"}); len(got) != 2 {
|
||||
if got := CreateStreamGrants([]string{"*", ">", "KV_a.b", "KV_a_b"}); len(got) != 3 {
|
||||
t.Errorf("grants for wildcard names: %v", got)
|
||||
}
|
||||
other, err := PermissionsFor(Principal{Kind: KindModule, Node: "novox", Module: "shop", CreatesStreams: []string{"KV_shop_x"}})
|
||||
@@ -142,12 +143,15 @@ func theGrantsPartCreatesOnlyDeclaredStreams(t *testing.T) {
|
||||
if MayPublish(other, "$JS.API.STREAM.CREATE.KV_shop_x") {
|
||||
t.Error("a module other than the grants part was granted a create")
|
||||
}
|
||||
// The writers table refuses the grants part an update, and a create of a stream it was not given.
|
||||
// The writers table refuses the grants part an update of a stream not declared, a delete, and a create of a stream it was not given.
|
||||
g := Principal{Kind: KindModule, Node: "novox", Module: GrantsModule, CreatesStreams: []string{"KV_a_b"}}
|
||||
if err := CheckWriters(g, []string{"$JS.API.STREAM.CREATE.KV_a_b"}); err != nil {
|
||||
t.Errorf("the declared create refused: %v", err)
|
||||
}
|
||||
for _, refused := range []string{"$JS.API.STREAM.UPDATE.KV_a_b", "$JS.API.STREAM.CREATE.KV_a_c", "$JS.API.STREAM.CREATE.*"} {
|
||||
if err := CheckWriters(g, []string{"$JS.API.STREAM.UPDATE.KV_a_b"}); err != nil { // the update that adds subjects
|
||||
t.Errorf("the declared update refused: %v", err)
|
||||
}
|
||||
for _, refused := range []string{"$JS.API.STREAM.UPDATE.KV_a_c", "$JS.API.STREAM.UPDATE.*", "$JS.API.STREAM.DELETE.KV_a_b", "$JS.API.STREAM.CREATE.KV_a_c", "$JS.API.STREAM.CREATE.*"} {
|
||||
if CheckWriters(g, []string{refused}) == nil {
|
||||
t.Errorf("%s was not refused", refused)
|
||||
}
|
||||
|
||||
@@ -1444,26 +1444,29 @@ const WatchmanModule = "mesh-delivery-watchman"
|
||||
|
||||
// CreateStreamGrants is what creating each named stream needs, and nothing wider (novox/hq ADR 0547 §5; the operator,
|
||||
// 2026-10-26 08:54): its create, and its information to read back what it holds. Each by its own name: a name that
|
||||
// is not one plain subject token is left out, since a wildcard there would be every stream.
|
||||
// is not one plain subject token is left out, since a wildcard there would be every stream. Its update is granted for the one
|
||||
// act the part does on a held stream: adding the subjects a manifest's new events need (grants.Creator.AddSubjects); it
|
||||
// never removes a subject, changes a limit or recreates the stream, and never deletes or purges one.
|
||||
func CreateStreamGrants(streams []string) []string {
|
||||
var out []string
|
||||
for _, s := range streams {
|
||||
if !safeSubject.MatchString(s) {
|
||||
continue
|
||||
}
|
||||
out = append(out, "$JS.API.STREAM.CREATE."+s, "$JS.API.STREAM.INFO."+s)
|
||||
out = append(out, "$JS.API.STREAM.CREATE."+s, "$JS.API.STREAM.INFO."+s, "$JS.API.STREAM.UPDATE."+s)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// 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)
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -36,14 +36,19 @@ type WriterRow struct {
|
||||
|
||||
// createsDeclared, and the other writers' tests, are who a row's writer is as a principal.
|
||||
//
|
||||
// createsDeclared is the grants part creating one stream a manifest declares, by its own name: the create alone, never an
|
||||
// update or a delete (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54); or the bus module's tools deleting one of
|
||||
// createsDeclared is the grants part creating one stream a manifest declares, by its own name: the create, and the update
|
||||
// that only adds subjects, never a delete (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54, 2026-11-05); or the bus module's tools deleting one of
|
||||
// the ended controller's leftover streams (ADR 0590 clause 6).
|
||||
func createsDeclared(p Principal, pattern string) bool {
|
||||
if removesLeftover(p, pattern) {
|
||||
return true
|
||||
}
|
||||
name, ok := strings.CutPrefix(pattern, "$JS.API.STREAM.CREATE.")
|
||||
if upd, isUpdate := strings.CutPrefix(pattern, "$JS.API.STREAM.UPDATE."); isUpdate {
|
||||
// The update of a stream a manifest declares, by its name, to add the subjects a new event needs (the operator,
|
||||
// 2026-11-05); the part's process adds and never removes a subject or changes a limit.
|
||||
return p.Kind == KindModule && p.Module == GrantsModule && slices.Contains(p.CreatesStreams, upd) && safeSubject.MatchString(upd)
|
||||
}
|
||||
if ok && p.Kind == KindModule && p.Module == GrantsModule && name == GrantsUserListObjectStream {
|
||||
return true
|
||||
}
|
||||
@@ -151,7 +156,7 @@ var WritersTable = []WriterRow{
|
||||
Subjects: []string{"$JS.API.STREAM.CREATE.>", "$JS.API.STREAM.UPDATE.>", "$JS.API.STREAM.DELETE.>",
|
||||
"$JS.API.CONSUMER.DURABLE.CREATE.>"},
|
||||
Writes: createsDeclared,
|
||||
Shared: "the grants part creates a stream a manifest declares, by its name, and never updates or deletes one (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54); the bus module's tools account deletes the ended controller's leftover streams, each by its name, and no other (novox/hq ADR 0590 clause 6)"},
|
||||
Shared: "the grants part creates a stream a manifest declares, by its name, adds to a held one the subjects its manifest names, and never removes a subject or deletes a stream (novox/hq ADR 0547 §5; the operator, 2026-10-26 08:54, 2026-11-05); the bus module's tools account deletes the ended controller's leftover streams, each by its name, and no other (novox/hq ADR 0590 clause 6)"},
|
||||
{State: "builds and their outcomes", Writer: "the build seat's holder", KeptIn: "its own state",
|
||||
Others: "the builder asks",
|
||||
Subjects: []string{"mesh.seat.mesh-build-machine.event.built"},
|
||||
|
||||
@@ -40,7 +40,7 @@ type Limits struct {
|
||||
}
|
||||
|
||||
// Assertion is one module's declared buckets, logs and streams, composed and compared. Asserted says this pass created
|
||||
// one the bus lacked (the operator, 2026-10-26 08:54); one the bus holds is never changed.
|
||||
// one the bus lacked (the operator, 2026-10-26 08:54); one the bus holds only ever gains subjects.
|
||||
type Assertion struct {
|
||||
Module string `json:"module"`
|
||||
Manifest string `json:"manifest"`
|
||||
@@ -54,13 +54,16 @@ const ShadowAsserts = "in the tools: composed and compared, nothing asserted (th
|
||||
|
||||
// CreatesMissing is what an assertion of the part's process says (the operator, 2026-10-26 08:54; issue 1790).
|
||||
const CreatesMissing = "a bucket, log or stream the manifest declares and the bus lacks is created with the manifest's limits, once per " +
|
||||
"manifest; one the bus holds is never changed or deleted, a difference is only said"
|
||||
"manifest; a stream the bus holds only ever gains the subjects its manifest adds, and nothing is removed, deleted or otherwise changed, a difference is only said"
|
||||
|
||||
// Creator creates one stream the bus does not hold, with the configuration given, and never updates or deletes one: the
|
||||
// part's process alone, whose account is granted the create of each declared stream by name (the operator, 2026-10-26
|
||||
// 08:54). A stream the bus holds already answers ErrHeld.
|
||||
type Creator interface {
|
||||
CreateDeclared(cfg jetstream.StreamConfig) error
|
||||
// AddSubjects adds subjects to a stream the bus holds and nothing else: the stream's subjects become the union of what
|
||||
// it holds and add; no subject is removed, no limit, retention or other setting changed, the stream not recreated.
|
||||
AddSubjects(stream string, add []string) error
|
||||
}
|
||||
|
||||
// ErrHeld is a create refused because the bus holds the stream already (made meanwhile): left as it is.
|
||||
@@ -100,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 {
|
||||
@@ -149,6 +173,35 @@ func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifes
|
||||
it.Created = ItemNotCreated + err.Error() + " (tried again at the next event)"
|
||||
}
|
||||
}
|
||||
// A stream the bus holds whose manifest names events it has no subject for gains those subjects, and only ever those: a
|
||||
// subject the manifest dropped stays, limits and retention are left as they are (the operator, 2026-11-05).
|
||||
for i := range a.Items {
|
||||
it := &a.Items[i]
|
||||
if it.Held == nil || it.Kind != "stream" {
|
||||
continue
|
||||
}
|
||||
var add []string
|
||||
for _, s := range it.Composed.Subjects {
|
||||
if !slices.Contains(it.Held.Subjects, s) {
|
||||
add = append(add, s)
|
||||
}
|
||||
}
|
||||
if len(add) == 0 {
|
||||
continue
|
||||
}
|
||||
if err := c.AddSubjects(it.Stream, add); err != nil {
|
||||
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, ", "))
|
||||
}
|
||||
if len(made) > 0 {
|
||||
a.Asserted = true
|
||||
}
|
||||
|
||||
@@ -17,7 +17,7 @@ import (
|
||||
// **The durable consumers an account needs** (novox/hq ADR 0561 §6, issue 774): a module's consumer on EVENTS and each
|
||||
// process's on its work queue. Since the controller ended by its unassignment (ADR 0590, issue 1126) the grants part
|
||||
// alone creates them, with the configuration the bus composer derives (buscompose.AccountConsumers), only where the bus
|
||||
// lacks one; one the bus holds is never changed or deleted by a pass, a difference is only said.
|
||||
// lacks one; one the bus holds only ever gains the filters its module's consumes adds (never loses one or changes another setting, never deleted by a pass), any other difference is only said.
|
||||
|
||||
// ConsumerCreator creates one durable consumer the bus does not hold: the part's process alone, whose account is granted
|
||||
// the create of each account's consumer by stream and name. One the bus holds answers ErrHeld.
|
||||
@@ -26,6 +26,9 @@ type ConsumerCreator interface {
|
||||
// MatchConsumer brings a held consumer to cfg, keeping where it delivers from: only on a person's ask
|
||||
// (grants_consumer_match), never by a pass.
|
||||
MatchConsumer(stream string, cfg jetstream.ConsumerConfig) error
|
||||
// AddFilters adds filter subjects to a held consumer and nothing else: its filters become the union of what it holds and
|
||||
// add, every other setting and its delivery position kept.
|
||||
AddFilters(stream, name string, add []string) error
|
||||
}
|
||||
|
||||
// ConsumersState is what one pass found and did of the account consumers.
|
||||
@@ -85,6 +88,21 @@ func heldConsumers(c Caller, stream string) (map[string]heldConsumer, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// filtersToAdd are the composed filters of a consumer the bus's one lacks, in the composed order; none when the consumer
|
||||
// filters the whole stream (no filters) or holds all.
|
||||
func filtersToAdd(c buscompose.Consumer, h heldConsumer) []string {
|
||||
if len(h.Filters) == 0 {
|
||||
return nil
|
||||
}
|
||||
var add []string
|
||||
for _, f := range c.Filters {
|
||||
if !slices.Contains(h.Filters, f) && !slices.Contains(add, f) {
|
||||
add = append(add, f)
|
||||
}
|
||||
}
|
||||
return add
|
||||
}
|
||||
|
||||
// ConsumersOn is why the part creates and matches the account consumers: the controller ended by its unassignment, so
|
||||
// the part is their only maker (novox/hq ADR 0590, issue 1126).
|
||||
const ConsumersOn = "the grants part makes and matches the account consumers: the controller ended by its unassignment (issue 1126)"
|
||||
@@ -124,6 +142,32 @@ func (p *Part) consumers(res *Result, recs buscompose.Records, match string) {
|
||||
}
|
||||
seen[key] = true
|
||||
}
|
||||
// A consumer whose module now consumes more events of the stream gains those filters by itself, only ever added:
|
||||
// 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 {
|
||||
// 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, ", "))
|
||||
}
|
||||
p.say(res, EventAsserted, map[string]any{"consumer": c.Name, "stream": stream, "filters-added": add})
|
||||
h.Filters = append(slices.Clone(h.Filters), add...)
|
||||
}
|
||||
}
|
||||
why := consumerDiffers(c, h)
|
||||
if why != "" && key == match && st.On && p.Consumers != nil {
|
||||
if err := p.Consumers.MatchConsumer(stream, buscompose.ConsumerConfig(c)); err != nil {
|
||||
|
||||
@@ -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
|
||||
@@ -399,6 +401,9 @@ func (p *Part) Compose() Result {
|
||||
if p.Create != nil && heldErr == nil {
|
||||
if made := createMissing(p.Create, &p.made, &a, in.Manifests[module], in.Nodes, forced == module); len(made) > 0 {
|
||||
res.Asserted = append(res.Asserted, fmt.Sprintf("%s at %s: created %s", module, a.Manifest, strings.Join(made, ", ")))
|
||||
if p.Logf != nil {
|
||||
p.Logf("%s at %s: asserted on the bus: %s", module, a.Manifest, strings.Join(made, ", "))
|
||||
}
|
||||
p.say(&res, EventAsserted, map[string]any{"module": module, "manifest": a.Manifest, "created": made})
|
||||
}
|
||||
for _, why := range notCreated(a) {
|
||||
|
||||
Reference in New Issue
Block a user