From 78b12be7528c964af6d17438403d39f5bfbcae53 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 04:59:19 +0100 Subject: [PATCH] NATS5: the bus module reads the bus's storage limits, refuses a plan whose caps pass the account's and shows the sum in list MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §547, §495, §661 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- modules/nats/cmd/nats-tools/broker.go | 18 +++++- modules/nats/cmd/nats-tools/broker_test.go | 18 +++++- modules/nats/cmd/nats-tools/keep.go | 2 +- modules/nats/cmd/nats-tools/main.go | 2 +- modules/nats/cmd/nats-tools/monitor.go | 21 +++++++ modules/nats/internal/keeper/fill.go | 5 ++ modules/nats/internal/keeper/jetstream.go | 21 +++++++ modules/nats/internal/keeper/keeper.go | 68 +++++++++++++++++++++ modules/nats/internal/keeper/keeper_test.go | 27 ++++++++ 9 files changed, 177 insertions(+), 5 deletions(-) diff --git a/modules/nats/cmd/nats-tools/broker.go b/modules/nats/cmd/nats-tools/broker.go index 6bddf5e5d..9647be55b 100644 --- a/modules/nats/cmd/nats-tools/broker.go +++ b/modules/nats/cmd/nats-tools/broker.go @@ -48,6 +48,16 @@ func HeldInState() (map[string]keeper.Asked, error) { return out, nil } +// capsOf is the sum of the bus's caps against its limits (design 47 G §547, §661), in MiB; a limit the keeper has not read, +// or that is none, is absent. The sum is over every module, whatever list is narrowed to. +func capsOf(all map[string]keeper.Asked, limits func() (keeper.Limits, error)) *meshbroker.ListAnswerCaps { + c := &meshbroker.ListAnswerCaps{SumMiB: keeper.CapSum(all, "", "") / keeper.MiB} + if l, err := limits(); err == nil { + c.AccountLimitMiB, c.ServerLimitMiB = l.AccountBytes/keeper.MiB, l.ServerBytes/keeper.MiB + } + return c +} + // busStream is one stream as the monitoring endpoint answers it. type busStream struct { Config jetstream.StreamConfig `json:"config"` @@ -157,7 +167,7 @@ func anyOf(m map[string]json.RawMessage) map[string]any { } // brokerTools are the four reading verbs and nats_history. -func brokerTools(read heldEntries, jsz func() ([]byte, error), history func(module, local string, limit int) ([]keeper.Event, error)) []stdio.Tool { +func brokerTools(read heldEntries, limits func() (keeper.Limits, error), jsz func() ([]byte, error), history func(module, local string, limit int) ([]keeper.Event, error)) []stdio.Tool { both := func() (map[string]keeper.Asked, map[string]busStream, error) { entries, err := read() if err != nil { @@ -194,7 +204,11 @@ func brokerTools(read heldEntries, jsz func() ([]byte, error), history func(modu if m := arg(a, "module"); m != "" { entries = map[string]keeper.Asked{m: entries[m]} } - return meshbroker.ListAnswer{Entries: each(entries, held, func(keeper.Made) bool { return true })}, nil + all, err := read() + if err != nil { + return nil, err + } + return meshbroker.ListAnswer{Caps: capsOf(all, limits), Entries: each(entries, held, func(keeper.Made) bool { return true })}, nil }}, {Name: "mesh-broker.show", Description: "One entry of a module's kept data on the bus: its bus name, template and options as asked, the settings the bus holds of each of its objects, and its fill against its cap. Whether it is refused, waiting, drifted, full or orphaned is what nats_history says of it.", diff --git a/modules/nats/cmd/nats-tools/broker_test.go b/modules/nats/cmd/nats-tools/broker_test.go index 6042b3af3..b3fb30360 100644 --- a/modules/nats/cmd/nats-tools/broker_test.go +++ b/modules/nats/cmd/nats-tools/broker_test.go @@ -1,6 +1,7 @@ package main import ( + "context" "encoding/json" "fmt" "reflect" @@ -41,6 +42,9 @@ func TestTheSeatsReadingVerbsAnswerFromTheKeeperAndTheBus(t *testing.T) { {"name":"m_log","config":{"name":"m_log","retention":"limits","discard":"new","max_bytes":33554432,"max_msg_size":262144},"state":{"messages":9,"bytes":9000}}]}]}` tools := map[string]stdio.Tool{} for _, tl := range brokerTools(func() (map[string]keeper.Asked, error) { return held, nil }, + func() (keeper.Limits, error) { + return keeper.Limits{AccountBytes: 2048 << 20, ServerBytes: 4096 << 20}, nil + }, func() ([]byte, error) { return []byte(jsz), nil }, func(module, local string, limit int) ([]keeper.Event, error) { return []keeper.Event{{Module: module, Local: local, Why: "first"}, {Module: module, Local: local, Why: "second"}}[:limit], nil @@ -54,10 +58,12 @@ func TestTheSeatsReadingVerbsAnswerFromTheKeeperAndTheBus(t *testing.T) { err string }{ // Each answers the seat's contract (mesh-contracts meshbroker ListAnswer, Held, CheckAnswer, OrphansAnswer). - {"mesh-broker.list", nil, []string{`{"entries":[`, `"bus-name":"m_drafts"`, `"local-name":"drafts"`, `"storage-type":"key-value"`, + {"mesh-broker.list", nil, []string{`"entries":[`, `"bus-name":"m_drafts"`, `"local-name":"drafts"`, `"storage-type":"key-value"`, `"KV_m_drafts":{"max-bytes":67108864`, `"fill":{"bytes":1048576,"cap":83886080,"count":3}`, `"options":{"max-mib":16}`, // The binding of an entry the bus holds is the one its created carries (design 25 bindings-first-view §2). `"binding":{"bucket":"m_drafts","local":"drafts","objects":"m_drafts-objects","storage":"key-value","template":"state"}`}, ""}, + // The sum of the caps against both limits (design 47 G §547): drafts 64+16, log 16, old 1024 (orphaned, still held); later is not made. + {"mesh-broker.list", nil, []string{`"caps":{"account-limit-mib":2048,"server-limit-mib":4096,"sum-mib":1120`}, ""}, {"mesh-broker.show", map[string]any{"module": "m", "local-name": "log"}, []string{`"bus-name":"m_log"`, `"template":"log"`, `"m_log":{"max-bytes":33554432`}, ""}, {"mesh-broker.show", map[string]any{"module": "m", "local-name": "later"}, []string{`"local-name":"later"`}, ""}, @@ -115,3 +121,13 @@ func TestTheSeatsReadingVerbsAnswerFromTheKeeperAndTheBus(t *testing.T) { }) } } + +// The server's storage limit is read from the monitoring document's config (design 47 G §547); none is 0. +func TestTheServersStorageLimitIsReadFromItsMonitor(t *testing.T) { + for doc, want := range map[string]int64{`{"config":{"max_storage":5368709120}}`: 5368709120, `{"config":{"max_storage":-1}}`: 0, `{}`: 0} { + got, err := serverStorageLimit(func(context.Context, string) ([]byte, error) { return []byte(doc), nil })(context.Background()) + if err != nil || got != want { + t.Errorf("%s: %d (%v), want %d", doc, got, err, want) + } + } +} diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index 2547bf5b1..fa2f82cbd 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/modules/nats/cmd/nats-tools/keep.go @@ -267,7 +267,7 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer defer writer.Close() k := &keeper.Keeper{ Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR")), NC: nc, - Dir: getenv("MESH_NATS_SNAPSHOTS_DIR"), Log: logger.Printf}, + Dir: getenv("MESH_NATS_SNAPSHOTS_DIR"), Log: logger.Printf, Server: serverStorageLimit(LiveMonitor())}, Store: processStore{conn}, Emit: streamEmitter(js, account), Say: sayer(js), View: firstView(conn), Act: func(what, why, cause string) error { _, err := recorder.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause}) diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index b20eefcba..c915c3f55 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -192,7 +192,7 @@ func allTools() []stdio.Tool { } out := append(tools(monitor, ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime, HeldInState)...) out = append(out, expiryTools(LiveBus(), SaidByTheRuntime, RecordedBySeat)...) - out = append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...) + out = append(out, brokerTools(HeldInState, func() (keeper.Limits, error) { s, err := StatusInState(); return s.Limits, err }, jsz, StreamHistory(LiveBus()))...) out = append(out, keeperHealthTool(StatusInState, HeldInState, jsz)) out = append(out, readTools(jsz, LiveMsgGet(LiveBus()))...) rs := RequestsInState() diff --git a/modules/nats/cmd/nats-tools/monitor.go b/modules/nats/cmd/nats-tools/monitor.go index 191e7b16c..961fcdbf7 100644 --- a/modules/nats/cmd/nats-tools/monitor.go +++ b/modules/nats/cmd/nats-tools/monitor.go @@ -11,6 +11,7 @@ package main import ( "context" + "encoding/json" "fmt" "io" "net/http" @@ -86,3 +87,23 @@ func ParseWritten(s string) (time.Time, error) { } return t, nil } + +// serverStorageLimit reads the server's own JetStream storage limit in bytes from its monitoring endpoint (`config`'s +// max_storage; 0 or less is none), shown beside the account's (design 47 G §547). +func serverStorageLimit(monitor Monitor) func(ctx context.Context) (int64, error) { + return func(ctx context.Context) (int64, error) { + raw, err := monitor(ctx, "/jsz?config=1") + if err != nil { + return 0, err + } + var doc struct { + Config struct { + MaxStorage int64 `json:"max_storage"` + } `json:"config"` + } + if err := json.Unmarshal(raw, &doc); err != nil { + return 0, err + } + return max(doc.Config.MaxStorage, 0), nil + } +} diff --git a/modules/nats/internal/keeper/fill.go b/modules/nats/internal/keeper/fill.go index ae062a478..87a9de1e7 100644 --- a/modules/nats/internal/keeper/fill.go +++ b/modules/nats/internal/keeper/fill.go @@ -21,6 +21,11 @@ func (k *Keeper) FillPass() { } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() + if lim, err := k.Bus.Limits(ctx); err == nil { + k.setStatus(func(s *Status) { s.Limits = lim }) + } else { + k.logf("the bus's storage limits cannot be read: %v", err) + } for _, module := range keys { held, _, found := k.asked(module) if !found { diff --git a/modules/nats/internal/keeper/jetstream.go b/modules/nats/internal/keeper/jetstream.go index c3e20079f..c5bd4b628 100644 --- a/modules/nats/internal/keeper/jetstream.go +++ b/modules/nats/internal/keeper/jetstream.go @@ -23,6 +23,27 @@ type JetStream struct { NC *nats.Conn Dir string Log func(string, ...any) + // Server reads the server's own storage limit in bytes (0 none), from its monitoring endpoint. + Server func(ctx context.Context) (int64, error) +} + +// Limits reads the account's storage limit (-1 is none) and the server's (design 47 G §547). +func (b JetStream) Limits(ctx context.Context) (Limits, error) { + info, err := b.JS.AccountInfo(ctx) + if err != nil { + return Limits{}, err + } + var l Limits + if info.Limits.MaxStore > 0 { + l.AccountBytes = info.Limits.MaxStore + } + if b.Server != nil { + // The server's limit is only shown, never held against (§547): one that cannot be read is shown as none. + if n, err := b.Server(ctx); err == nil { + l.ServerBytes = n + } + } + return l, nil } // Held reads the object's stream. diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go index a3e6c4a20..a25ba9e08 100644 --- a/modules/nats/internal/keeper/keeper.go +++ b/modules/nats/internal/keeper/keeper.go @@ -127,6 +127,8 @@ type Status struct { Started time.Time `json:"started"` // Pass is when the last fill pass ended. Pass time.Time `json:"pass"` + // Limits are the bus's storage limits as last read (design 47 G §547): what `list` shows against the sum of the caps. + Limits Limits `json:"limits,omitempty"` // FirstView is the indexer's first view: when asked, when answered and at which revision, or why not. FirstView FirstViewStatus `json:"first-view"` // Heard is when the last manifest-accepted was heard, and Unanswered since when one has waited unhandled. @@ -193,6 +195,54 @@ type Bus interface { // keeper's own partial stream of a restore that did not end). Restore(ctx context.Context, path string) error DeleteStream(ctx context.Context, name string) error + // Limits reads the bus's storage limits: the keeper's account's and the server's (design 47 G §547). + Limits(ctx context.Context) (Limits, error) +} + +// Limits are the bus's storage limits in bytes; 0 is no limit (an unlimited account, or a server that sets none). +type Limits struct { + AccountBytes int64 `json:"account-bytes,omitempty"` + ServerBytes int64 `json:"server-bytes,omitempty"` +} + +// CapSum is the sum, in bytes, of the caps of every object the entries made hold, but for the entry of skipModule and +// skipLocal (design 47 G §547): what the bus's limits are read against. An entry never made, or refused, holds none. +func CapSum(held map[string]Asked, skipModule, skipLocal string) int64 { + var sum int64 + for module, a := range held { + for local, m := range a.Made { + if (module == skipModule && local == skipLocal) || !(m.Created || m.Orphaned) { + continue + } + for _, o := range m.Plan.Objects { + sum += o.Settings.MaxBytes + } + } + } + return sum +} + +// capRoom refuses a plan that adds or raises a cap when the sum of the bus's caps would pass the account's limit, +// naming both (design 47 G §547); caps already made stay as they are. With no account limit nothing is refused. +func (k *Keeper) capRoom(ctx context.Context, held *Asked, m *Made) error { + lim, err := k.Bus.Limits(ctx) + if err != nil { + return fmt.Errorf("%w: the bus's storage limits cannot be read: %v", ErrRetry, err) + } + k.setStatus(func(s *Status) { s.Limits = lim }) + if lim.AccountBytes <= 0 { + return nil + } + all := k.Held() + all[held.Module] = *held + sum := CapSum(all, held.Module, m.Plan.Local) + for _, o := range m.Plan.Objects { + sum += o.Settings.MaxBytes + } + if sum > lim.AccountBytes { + return fmt.Errorf("the plan takes the sum of the bus's caps to %d MiB, past the account's storage limit of %d MiB", sum/MiB, lim.AccountBytes/MiB) + } + return nil } // Fill is how much one object holds: its bytes and messages. @@ -499,6 +549,18 @@ func (k *Keeper) ensure(held *Asked, m *Made) { return } if !found { + if err := k.capRoom(ctx, held, m); err != nil { + if errors.Is(err, ErrRetry) { + m.Waiting = err.Error() + return + } + m.Refused, m.Waiting = fmt.Sprintf("%s: %v", o.Stream(), err), "" + k.emit(EventRefused, m.Plan, held.Revision, m.Refused, asked, nil) + k.raise(conditionRefused, held.Module, m.Plan.Local, occ.SeverityWarning, "The bus module refused to create kept data", + "a smaller cap, or more storage for the bus's account", + fmt.Sprintf("%s of %s was not created: %s. The module gets no binding for it and fails at its start, naming it", m.Plan.Local, held.Module, m.Refused)) + return + } if err := k.Bus.Create(ctx, o); err != nil { if errors.Is(err, ErrRetry) { m.Waiting = fmt.Sprintf("%s cannot be created yet: %v", o.Stream(), err) @@ -520,6 +582,12 @@ func (k *Keeper) ensure(held *Asked, m *Made) { continue } if Grows(o.Settings, s) { + if o.Settings.MaxBytes > s.MaxBytes { + if err := k.capRoom(ctx, held, m); err != nil { + m.Waiting = err.Error() + return + } + } if err := k.Bus.Update(ctx, o); err != nil { m.Waiting = fmt.Sprintf("%s's larger settings cannot be applied yet: %v", o.Stream(), err) return diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index 31a5105b5..30c07cb47 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -38,6 +38,8 @@ type fakeBus struct { restored []string gate chan struct{} deletedStreams []string + // limits are the bus's storage limits the fake answers. + limits Limits } func (b *fakeBus) SnapshotStreams(path string) ([]string, error) { @@ -73,6 +75,12 @@ func (b *fakeBus) DeleteStream(_ context.Context, name string) error { return nil } +func (b *fakeBus) Limits(context.Context) (Limits, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.limits, nil +} + func newFakeBus() *fakeBus { return &fakeBus{held: map[string]Settings{}, fill: map[string]uint64{}, refuse: map[string]error{}} } @@ -548,6 +556,25 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { t.Errorf("said %v %v: cleared once asked as held", out.events, out.conds) } }}, + {"a plan whose caps pass the account's storage limit is refused, naming the sum and the limit, and the caps made stay (§547)", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) { + // drafts: 64 MiB state + 16 MiB objects = 80 MiB; log: 1024 MiB. + bus.limits = Limits{AccountBytes: 100 * MiB} + k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts}}) + k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{drafts, log}}) + m := k.Held()["m"].Made + if fmt.Sprint(bus.created) != "[KV_m_drafts OBJ_m_drafts-objects]" || !m["drafts"].Created || m["log"].Created || + !strings.Contains(m["log"].Refused, "1104 MiB, past the account's storage limit of 100 MiB") || fmt.Sprint(out.events) != "[created m drafts refused m log]" { + t.Errorf("created %v, said %v, log %+v: the caps made stay, the plan that passes the limit refused naming both", bus.created, out.events, m["log"]) + } + if CapSum(k.Held(), "", "") != 80*MiB { + t.Errorf("the sum is %d, want 80 MiB: a refused entry holds none", CapSum(k.Held(), "", "")) + } + bus.limits = Limits{} + k.CheckAll() + if !k.Held()["m"].Made["log"].Created { + t.Errorf("with no account limit nothing is refused: %+v", k.Held()["m"].Made["log"]) + } + }}, {"a state made before its objects store existed gets the store and its binding said again (§58)", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) { p, _ := PlanOf("m", drafts) bus.held["KV_m_drafts"] = p.Objects[0].Settings