diff --git a/modules/mesh-grants/internal/buscompose/create.go b/modules/mesh-grants/internal/buscompose/create.go index f4d76a68..5c534ed6 100644 --- a/modules/mesh-grants/internal/buscompose/create.go +++ b/modules/mesh-grants/internal/buscompose/create.go @@ -25,12 +25,16 @@ func BucketStreamConfig(b Bucket) jetstream.StreamConfig { if age > 0 && age < dupes { dupes = age } + maxBytes := int64(StateMaxBytes) + if b.MaxMiB > 0 { + maxBytes = int64(b.MaxMiB) << 20 + } cfg := jetstream.StreamConfig{ Name: "KV_" + b.Bucket(), Description: b.Why(), Subjects: []string{"$KV." + b.Bucket() + ".>"}, MaxMsgsPerSubject: int64(history), - MaxBytes: StateMaxBytes, + MaxBytes: maxBytes, MaxAge: age, MaxMsgSize: StateMaxValueBytes, Storage: jetstream.FileStorage, diff --git a/modules/mesh-grants/internal/buscompose/create_test.go b/modules/mesh-grants/internal/buscompose/create_test.go index 0df9580e..64f8bc47 100644 --- a/modules/mesh-grants/internal/buscompose/create_test.go +++ b/modules/mesh-grants/internal/buscompose/create_test.go @@ -10,6 +10,7 @@ import ( // Defends issue 1790 (ADR 0547 ยง5): the grants part creates a declared bucket and stream as the controller asserts them. func declaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) { plain := BucketStreamConfig(Bucket{Module: "mesh-vault", Name: "bus-accounts"}) + declared := BucketStreamConfig(Bucket{Module: "c", Name: "compositions", MaxMiB: 256}) keyed := BucketStreamConfig(Bucket{Module: "m", Name: "presence", History: 5, TTLSeconds: 60, KeyTTL: true}) queue := ModuleStreamConfig(ModuleStream{Module: "b", Name: "ASKS", Events: []string{"ask"}, MaxAgeDays: 7, MaxMiB: 32, WorkQueue: true, Duplicates: 1}) @@ -21,6 +22,7 @@ func declaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) { {"bucket subjects", plain.Subjects[0], "$KV.mesh-vault_bus-accounts.>"}, {"bucket history defaults to one", plain.MaxMsgsPerSubject, int64(1)}, {"bucket cap", plain.MaxBytes, int64(StateMaxBytes)}, + {"bucket cap as its state declares it (max-mib)", declared.MaxBytes, int64(256 << 20)}, {"bucket value cap", plain.MaxMsgSize, int32(StateMaxValueBytes)}, {"bucket duplicate window", plain.Duplicates, 2 * time.Minute}, {"bucket discards new", plain.Discard, jetstream.DiscardNew}, diff --git a/modules/mesh-grants/internal/buscompose/state.go b/modules/mesh-grants/internal/buscompose/state.go index 7a57694a..072b239e 100644 --- a/modules/mesh-grants/internal/buscompose/state.go +++ b/modules/mesh-grants/internal/buscompose/state.go @@ -36,6 +36,8 @@ type Bucket struct { History int // TTLSeconds is how long a value lives; zero is until replaced or deleted. TTLSeconds int + // MaxMiB is the bucket's cap in MiB as its state declares (max-mib); zero is StateMaxBytes. + MaxMiB int // PerMachine says each machine's instance writes only the key named for its machine (novox/hq ADR // 0260), and is granted that key and no other. PerMachine bool diff --git a/modules/mesh-grants/internal/grants/assertions.go b/modules/mesh-grants/internal/grants/assertions.go index 03f1c8bc..32b15ed3 100644 --- a/modules/mesh-grants/internal/grants/assertions.go +++ b/modules/mesh-grants/internal/grants/assertions.go @@ -277,9 +277,14 @@ func Composed(m catalogue.Manifest, nodes []string) []Item { if per < 1 { per = 1 } + // A bucket's size cap is composed only where its state declares one (max-mib); the default is the bus's own. + var maxBytes int64 + if b.MaxMiB > 0 { + maxBytes = buscompose.BucketStreamConfig(b).MaxBytes + } out = append(out, Item{Kind: "bucket", Stream: "KV_" + b.Bucket(), Composed: Limits{ Subjects: []string{"$KV." + b.Bucket() + ".>"}, Retention: string(buscompose.RetentionLimits), - MaxAgeHours: b.TTLSeconds / 3600, MaxPerSubject: per}}) + MaxAgeHours: b.TTLSeconds / 3600, MaxPerSubject: per, MaxBytes: maxBytes}}) } for _, l := range logsOf(m) { c := l.Config() diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index ae8842d7..b4a78958 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -269,7 +269,7 @@ func issuedTools(m catalogue.Manifest) []string { func bucketsOf(m catalogue.Manifest) []buscompose.Bucket { var out []buscompose.Bucket for _, s := range m.State { - out = append(out, buscompose.Bucket{Module: m.Module, Name: s.Name, History: s.History, TTLSeconds: s.TTLSeconds, + out = append(out, buscompose.Bucket{Module: m.Module, Name: s.Name, History: s.History, TTLSeconds: s.TTLSeconds, MaxMiB: s.MaxMiB, PerMachine: s.PerMachine, KeyTTL: s.KeyTTL, ReadOnly: s.ReadOnly}) } return out diff --git a/modules/mesh-grants/internal/grants/statecap_test.go b/modules/mesh-grants/internal/grants/statecap_test.go new file mode 100644 index 00000000..3289b718 --- /dev/null +++ b/modules/mesh-grants/internal/grants/statecap_test.go @@ -0,0 +1,40 @@ +package grants + +import ( + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// raiser is a Creator that holds one stream at a cap and records the cap it is raised to. +type raiser struct { + Creator + to int64 +} + +func (r *raiser) RaiseMaxBytes(_ string, to int64) (int64, bool, error) { + r.to = to + return 64 << 20, true, nil +} + +// A state's declared max-mib reaches the bucket (issue 1975 note 6): the composed stream carries it, and a bucket the bus +// holds at the default 64 MiB is raised to it by the assertion. +func TestAStatesDeclaredCapReachesItsBucket(t *testing.T) { + m := catalogue.Manifest{Module: "mesh-delivery-checker", State: []catalogue.StateDeclaration{{Name: "compositions", MaxMiB: 256}}} + var composed *Item + for _, it := range Composed(m, nil) { + if it.Stream == "KV_mesh-delivery-checker_compositions" { + composed = &it + } + } + if composed == nil || composed.Composed.MaxBytes != 256<<20 { + t.Fatalf("the composed bucket is %+v, want a cap of 256 MiB", composed) + } + a := Assertion{Manifest: "d", Items: []Item{*composed}} + a.Items[0].Held = &Limits{MaxBytes: 64 << 20} + r := &raiser{} + _, raises := createMissing(r, &creations{}, &a, m, nil, false) + if r.to != 256<<20 || len(raises) != 1 { + t.Errorf("the held bucket was raised to %d (%v), want 256 MiB", r.to, raises) + } +}