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

Builds: design 47 G §547, §495, §661

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

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.",

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)
+		}
+	}
+}

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})

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()

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
+	}
+}

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 {

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.

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

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