From 15918318862217fbe14f6a40f8e15fb40711aab7 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 10:10:53 +0100 Subject: [PATCH] A held stream's size cap is raised to its manifest's when larger, never lowered (issue 1966) The operator, 2026-11-09 09:07: raise only. The grants part's assert raises it, says it in its asserted event and the journal with the old and new cap, and grants_assert asks it by hand. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- modules/mesh-grants/cmd/mesh-grants/create.go | 22 +++++++ .../cmd/mesh-grants/integration_test.go | 17 +++++- modules/mesh-grants/cmd/mesh-grants/tools.go | 5 +- modules/mesh-grants/grants_test.go | 58 ++++++++++++++++--- .../mesh-grants/internal/buscompose/nats.go | 5 +- .../mesh-grants/internal/grants/assertions.go | 58 +++++++++++++++++-- modules/mesh-grants/internal/grants/part.go | 4 +- modules/mesh-grants/internal/grants/tools.go | 13 +++-- 8 files changed, 157 insertions(+), 25 deletions(-) diff --git a/modules/mesh-grants/cmd/mesh-grants/create.go b/modules/mesh-grants/cmd/mesh-grants/create.go index c4271f70..23621ec8 100644 --- a/modules/mesh-grants/cmd/mesh-grants/create.go +++ b/modules/mesh-grants/cmd/mesh-grants/create.go @@ -88,6 +88,28 @@ func (c creating) AddSubjects(stream string, add []string) error { return err } +// RaiseMaxBytes raises a held stream's size cap to `to` when that is larger than the cap it holds, and changes nothing +// else: it reads the stream's configuration and updates with the rest as it was. An unbounded stream (-1) and a cap that +// is larger or equal are left as they are. +func (c creating) RaiseMaxBytes(stream string, to int64) (int64, bool, error) { + ctx, cancel := context.WithTimeout(c.ctx, requestWithin) + defer cancel() + s, err := c.js.Stream(ctx, stream) + if err != nil { + return 0, false, err + } + cfg := s.CachedInfo().Config + if cfg.MaxBytes <= 0 || cfg.MaxBytes >= to { + return cfg.MaxBytes, false, nil + } + from := cfg.MaxBytes + cfg.MaxBytes = to + if _, err := c.js.UpdateStream(ctx, cfg); err != nil { + return from, false, err + } + return from, true, nil +} + // 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) diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index 0bb0d6fa..efbf6f98 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -324,7 +324,7 @@ func TestTheGrantsPartWhole(t *testing.T) { // 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}) + Retention: jetstream.LimitsPolicy, Discard: jetstream.DiscardNew, MaxBytes: 1 << 20}) if err != nil { t.Fatal(err) } @@ -368,6 +368,21 @@ func TestTheGrantsPartWhole(t *testing.T) { 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) } + // A size cap is only ever raised, and nothing else changes with it (the operator, 2026-11-09). + for _, row := range []struct { + to, from, want int64 + raised bool + }{{4 << 20, 1 << 20, 4 << 20, true}, {2 << 20, 4 << 20, 4 << 20, false}} { + from, raised, err := c.RaiseMaxBytes("ADDS", row.to) + if err != nil { + t.Fatal(err) + } + now, _ := stream.Info(ctx) + if from != row.from || raised != row.raised || now.Config.MaxBytes != row.want || now.Config.MaxAge != b.MaxAge || + !slices.Equal(now.Config.Subjects, []string{"a.>", "b.>"}) || !now.Created.Equal(before.Created) || now.State.Msgs != before.State.Msgs { + t.Errorf("raise to %d: from %d raised %v, config %+v", row.to, from, raised, now.Config) + } + } // 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) diff --git a/modules/mesh-grants/cmd/mesh-grants/tools.go b/modules/mesh-grants/cmd/mesh-grants/tools.go index 1282ff02..53268412 100644 --- a/modules/mesh-grants/cmd/mesh-grants/tools.go +++ b/modules/mesh-grants/cmd/mesh-grants/tools.go @@ -192,8 +192,9 @@ func toolsOf(p *grants.Part) []stdio.Tool { "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. 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.", + "subjects its manifest adds (and a consumer the filters its module's consumes adds) and has its size cap RAISED to the " + + "manifest's when that is larger (said as an asserted event and in the journal, with the old and new cap), only ever adding or raising: it never removes " + + "a subject, lowers a cap or any other limit, changes 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 " + diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 99d97ff2..746fe606 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -72,9 +72,11 @@ 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 + // raises are the size caps the part raised on a held stream, as "stream from to". + raises []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). @@ -145,6 +147,26 @@ func (b *fakeBus) AddSubjects(stream string, add []string) error { return errors.New("stream not found") } +// RaiseMaxBytes is the bus's raise of a held stream's size cap: only to a larger cap, an unbounded stream and every other +// setting kept. +func (b *fakeBus) RaiseMaxBytes(stream string, to int64) (int64, bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + for _, s := range b.streams { + if s["name"] != stream { + continue + } + have, _ := s["max_bytes"].(int64) + if have <= 0 || have >= to { + return have, false, nil + } + s["max_bytes"] = to + b.raises = append(b.raises, fmt.Sprintf("%s %d %d", stream, have, to)) + return have, true, nil + } + return 0, false, 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() @@ -583,9 +605,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 two updates of a stream: adding subjects to a held one (the operator, 2026-11-05) and raising its size cap + // (2026-11-09), both in create.go. + if n := strings.Count(string(src), "UpdateStream("); (path == createOnly && n != 2) || (path != createOnly && n != 0) { + t.Errorf("%s holds UpdateStream %d times: only create.go updates a stream, twice, to add subjects and to raise a size cap", 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 { @@ -2044,7 +2067,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": 24}} + b.streams = []map[string]any{{"name": "THINGS", "subjects": []string{"dropped.subject"}, "retention": "limits", "max_age_hours": 24, "max_bytes": int64(1 << 20)}} p := partOn(b) p.Create = b var logged []string @@ -2062,6 +2085,11 @@ func TestGrants(t *testing.T) { } } } + // Its size cap, smaller than the manifest's 5 MiB, is raised to it, said as an event with the old and new cap and in the journal. + if len(b.raises) != 1 || b.raises[0] != fmt.Sprintf("THINGS %d %d", 1<<20, 5<<20) || + !strings.Contains(strings.Join(res.Asserted, " "), "raised from 1048576 to 5242880") || !strings.Contains(strings.Join(logged, " "), "raised from 1048576 to 5242880") { + t.Errorf("the cap was not raised or not said: %v, %v, %v", b.raises, res.Asserted, logged) + } 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) } @@ -2086,11 +2114,25 @@ func TestGrants(t *testing.T) { 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. + // A limit the manifest changed (the held max age is not the manifest's) is not touched; only subjects are ever added + // and a size cap raised: a held cap larger than the manifest's is never lowered, at a hand assert too. b.mu.Lock() b.refuseAdd = "" + for _, s := range b.streams { + if s["name"] == "THINGS" { + s["max_bytes"] = int64(9 << 20) + } + } b.mu.Unlock() + if _, err := partOn(b).AssertOne("mod-x"); err != nil { + t.Fatal(err) + } p.Compose() + for _, s := range b.streams { + if s["name"] == "THINGS" && (s["max_bytes"] != int64(9<<20) || len(b.raises) != 1) { + t.Errorf("a larger held cap was lowered: %v, %v", s, b.raises) + } + } 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) diff --git a/modules/mesh-grants/internal/buscompose/nats.go b/modules/mesh-grants/internal/buscompose/nats.go index dbee661c..58b04956 100644 --- a/modules/mesh-grants/internal/buscompose/nats.go +++ b/modules/mesh-grants/internal/buscompose/nats.go @@ -1506,8 +1506,9 @@ 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. 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. +// acts the part does on a held stream: adding the subjects a manifest's new events need (grants.Creator.AddSubjects) and +// raising its size cap to the manifest's when larger (grants.Creator.RaiseMaxBytes; the operator, 2026-11-09); it never +// removes a subject, lowers a limit or recreates the stream, and never deletes or purges one. func CreateStreamGrants(streams []string) []string { var out []string for _, s := range streams { diff --git a/modules/mesh-grants/internal/grants/assertions.go b/modules/mesh-grants/internal/grants/assertions.go index 33f2085d..03f1c8bc 100644 --- a/modules/mesh-grants/internal/grants/assertions.go +++ b/modules/mesh-grants/internal/grants/assertions.go @@ -54,7 +54,8 @@ 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; 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" + "manifest; a stream the bus holds only ever gains the subjects its manifest adds and has its size cap raised to the manifest's when that is larger " + + "(never lowered), 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 @@ -64,6 +65,17 @@ type Creator interface { // 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 + // RaiseMaxBytes raises the size cap of a stream the bus holds to `to` when `to` is larger than the cap it holds, and + // changes nothing else (the operator, 2026-11-09 09:07: "raise only"). It answers the cap it held and whether it + // raised; a cap that is larger or equal, or an unbounded stream, is left as it is. It never lowers any limit. + RaiseMaxBytes(stream string, to int64) (from int64, raised bool, err error) +} + +// Raise is one held stream's size cap raised to the manifest's, said in the event and the journal. +type Raise struct { + Stream string `json:"stream"` + From int64 `json:"from"` + To int64 `json:"to"` } // ErrHeld is a create refused because the bus holds the stream already (made meanwhile): left as it is. @@ -103,6 +115,9 @@ func Configs(m catalogue.Manifest, nodes []string) map[string]jetstream.StreamCo type creations struct { mu sync.Mutex made map[string]string + // raised is, by stream, the manifest digest at which its size cap was last compared with the manifest's, so the bus is + // asked for it once per manifest (a hand assert asks again). + raised 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 @@ -132,6 +147,21 @@ func (c *creations) madeAt(stream string) string { return c.made[stream] } +func (c *creations) raisedAt(stream string) string { + c.mu.Lock() + defer c.mu.Unlock() + return c.raised[stream] +} + +func (c *creations) keepRaised(stream, digest string) { + c.mu.Lock() + defer c.mu.Unlock() + if c.raised == nil { + c.raised = map[string]string{} + } + c.raised[stream] = digest +} + func (c *creations) keep(stream, digest string) { c.mu.Lock() defer c.mu.Unlock() @@ -143,8 +173,8 @@ func (c *creations) keep(stream, digest string) { // createMissing creates each item of a the bus lacks, with the manifest's limits, and says what became of it. It never // touches an item the bus holds, and does nothing when what the bus holds was not read. force asks again a stream -// created at this manifest and gone since (a hand assert). It answers the streams created. -func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifest, nodes []string, force bool) []string { +// created at this manifest and gone since (a hand assert). It answers the streams created and the size caps raised. +func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifest, nodes []string, force bool) ([]string, []Raise) { cfgs := Configs(m, nodes) var made []string a.Says = CreatesMissing @@ -202,10 +232,30 @@ func createMissing(c Creator, done *creations, a *Assertion, m catalogue.Manifes it.Created = "subjects added: " + strings.Join(add, ", ") made = append(made, "stream "+it.Stream+" gained "+strings.Join(add, ", ")) } + // A held stream, of any kind, whose manifest declares a larger size cap gets that cap, and no other limit changes: a + // cap the bus holds that is larger or equal stays (the operator, 2026-11-09 09:07). Compared once per manifest. + var raises []Raise + for i := range a.Items { + it := &a.Items[i] + if it.Held == nil || it.Composed.MaxBytes <= 0 || (!force && done.raisedAt(it.Stream) == a.Manifest) { + continue + } + from, raised, err := c.RaiseMaxBytes(it.Stream, it.Composed.MaxBytes) + if err != nil { + it.Created = ItemNotCreated + "size cap not raised: " + err.Error() + " (tried again at the next event)" + continue + } + done.keepRaised(it.Stream, a.Manifest) + if raised { + it.Created = fmt.Sprintf("size cap raised from %d to %d bytes", from, it.Composed.MaxBytes) + raises = append(raises, Raise{Stream: it.Stream, From: from, To: it.Composed.MaxBytes}) + made = append(made, fmt.Sprintf("%s %s size cap raised from %d to %d bytes", it.Kind, it.Stream, from, it.Composed.MaxBytes)) + } + } if len(made) > 0 { a.Asserted = true } - return made + return made, raises } // notCreated is each item of a creation refused, with why, as `: `. diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index 5e4ea403..a4c1ac8b 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -503,12 +503,12 @@ func (p *Part) Compose() Result { a := Assert(module, in.Digests[module], in.Manifests[module], in.Nodes, held, heldErr) // Created only where the bus was read and lacks it, never changed where it holds it (issue 1790). if p.Create != nil && heldErr == nil { - if made := createMissing(p.Create, &p.made, &a, in.Manifests[module], in.Nodes, forced == module); len(made) > 0 { + if made, raises := 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}) + p.say(&res, EventAsserted, map[string]any{"module": module, "manifest": a.Manifest, "created": made, "raised": raises}) } for _, why := range notCreated(a) { res.NotAsserted = append(res.NotAsserted, module+" "+why) diff --git a/modules/mesh-grants/internal/grants/tools.go b/modules/mesh-grants/internal/grants/tools.go index 445a9761..18439364 100644 --- a/modules/mesh-grants/internal/grants/tools.go +++ b/modules/mesh-grants/internal/grants/tools.go @@ -258,8 +258,8 @@ func (p *Part) UserList() (any, error) { // AssertOne answers the tool assert: one module's buckets, logs and streams as composed from its manifest at the // indexer's trunk, beside what the bus holds, and asks the part's process (which alone holds the grant to create) to -// create what the bus lacks now, even one it created at this manifest and that is gone since (issue 1790). It changes no -// stream the bus holds. Here, in the tools, it creates nothing itself: the answer names the ask. +// create what the bus lacks now, even one it created at this manifest and that is gone since (issue 1790), and to raise a +// held stream's size cap to the manifest's where that is larger (the operator, 2026-11-09). It lowers no limit. Here, in the tools, it creates nothing itself: the answer names the ask. func (p *Part) AssertOne(module string) (any, error) { if module == "" { return nil, errors.New("assert names a module") @@ -292,16 +292,17 @@ func (p *Part) AssertOne(module string) (any, error) { missing++ } } - if missing == 0 { - out["says"] = "the bus holds every bucket, log and stream the manifest declares: nothing to create; a held stream only ever gains the subjects a manifest adds, any other difference is only said" - return out, nil - } ask, _ := json.Marshal(Ask{By: "grants_assert", At: p.now(), Assert: module}) rev, err := p.Store.Put(StateUserList, KeyAsked, ask) if err != nil { return nil, fmt.Errorf("the ask to create could not be written: %w", err) } out["asked"] = rev + if missing == 0 { + out["says"] = "the bus holds every bucket, log and stream the manifest declares: nothing to create; asked the part's process to raise " + + "a held stream's size cap to the manifest's where that is larger (never lower), and to add the subjects a manifest adds; any other difference is only said" + return out, nil + } out["says"] = fmt.Sprintf("%d the bus lacks: the part's process creates them at this ask if %s is placed (its grant names only placed "+ "modules' streams) and says asserted; grants_assert again shows what the bus holds then", missing, module) return out, nil