grants: a state's declared max-mib reaches its bucket, and a held bucket is raised to it (issue 1975 note 6)
Builds: issue 1975 note 6 (the lead's review round 1; no design line) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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},
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user