A held stream's size cap is raised to its manifest's when larger, never lowered (issue 1966)
mesh/delivery merged: source.main moved to novox/mesh-delivery@58520d343537
mesh/check checking 159183188622 on 58520d343537

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
This commit is contained in:
2026-11-09 10:10:53 +01:00
parent f8b8be1600
commit 1591831886
8 changed files with 157 additions and 25 deletions
@@ -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)
@@ -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)
+3 -2
View File
@@ -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 " +
+50 -8
View File
@@ -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)
@@ -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 {
@@ -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 `<stream>: <why>`.
+2 -2
View File
@@ -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)
+7 -6
View File
@@ -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