Merge pull request 'Grants carry a state's declared cap; the checker's compositions get room and a drop tool; the assigner trusts no stale reads (issue 1975)' (#206) from fix/engine-reads-and-compositions into main

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go

@@ -552,13 +552,10 @@ func sayVia(ctx context.Context, js jetstream.JetStream) func(oc.Condition) erro
 	}
 }
 
-// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable
-// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of
-// indexerViews, and starts once both are registered.
-func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, live func(), logf func(string, ...any)) {
-	account := process.Account{Node: thisNode, Module: assignment.Module}
-	w := notHeardWriter(findings, logf)
-	handle := func(m jetstream.Msg) {
+// readsChangedHandler is what the assigner does with each event manifest-reads-changed: acknowledges it, keeps what it says
+// of the node (a statement older than it does not stand) and asks the node again.
+func readsChangedHandler(reads *enginereads.Keeper, logf func(string, ...any)) func(jetstream.Msg) {
+	return func(m jetstream.Msg) {
 		if err := m.Ack(); err != nil {
 			logf("acknowledging %s: %v", m.Subject(), err)
 		}
@@ -569,9 +566,22 @@ func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.
 			logf("%s holds what is not %s, and %s is asked again all the same: %v", m.Subject(), enginereads.Event, node, err)
 			reads.AskNode(node, enginereads.Event+" that does not read")
 		default:
+			// What the event says is kept first: a statement older than it does not stand while the node is asked again.
+			if e, derr := enginereads.DecodeChanged(m.Data()); derr == nil {
+				reads.HeardChanged(e)
+			}
 			reads.AskNode(node, enginereads.Event)
 		}
 	}
+}
+
+// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable
+// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of
+// indexerViews, and starts once both are registered.
+func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, live func(), logf func(string, ...any)) {
+	account := process.Account{Node: thisNode, Module: assignment.Module}
+	w := notHeardWriter(findings, logf)
+	handle := readsChangedHandler(reads, logf)
 	err := hub.Handle(enginereads.SubjectPrefix+"*", handle)
 	// At each bind every node the assigner holds a statement of is asked again: an event said before its consumer existed
 	// is not delivered to it (issue 1966 note 155: ace's bootstrapped statement stood; ADR 0668 §4).

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go

@@ -317,6 +317,23 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) {
 		if asked["ace"] != 2 {
 			t.Errorf("ace, back online, was asked %d times, want 2", asked["ace"])
 		}
+		// Issue 1975: ace's engine answered its bootstrapped seats, then stated the full list and said so; an answer older
+		// than that event does not stand: the rules read what the event says, and the node is asked until it is as new.
+		// The engine's verb keeps answering the bootstrapped statement (its stated-at is older than the event); the event is
+		// handed to the assigner's own handler (readsChangedHandler, what hearReadsChanged registers) as the bus delivers it.
+		bootstrapped := func(node, _ string, _ any) (json.RawMessage, error) {
+			return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"seat:a"}, StatedAt: "2026-11-12T01:14:36.498Z"})
+		}
+		rk := &enginereads.Keeper{Call: bootstrapped, Sync: true}
+		rk.AskNode("ace", "the assigner started")
+		body, _ := json.Marshal(nodeengine.ManifestReadsChangedEvent{V: 1, Node: "ace", Why: enginereads.ReadsStarted, At: "2026-11-12T01:14:39.002Z", ManifestReads: []string{"seat:a", "seat:b"}})
+		readsChangedHandler(rk, t.Logf)(readsMsg{subject: enginereads.SubjectPrefix + "ace", data: body})
+		rp := &passer{w: w, reads: rk, online: map[string]bool{"ace": true}}
+		rn := w.nodes(nil, nil)
+		rp.engineReads(rn)
+		if !slices.Equal(rn.Reads["ace"], []string{"seat:a", "seat:b"}) {
+			t.Errorf("the rules read the statement older than the event heard: %v", rn.Reads["ace"])
+		}
 		// ADR 0668 §3 (review round 1): a node whose engine answers nothing while its machine is heard is asked again at each
 		// pass once its ask ended, and past five minutes is the finding call-unanswered, cleared once it answers; none is
 		// written while its machine is away.
@@ -1491,3 +1508,14 @@ func readAskID(t *testing.T, p *passer, key string) string {
 	}
 	return d.Ask.ID
 }
+
+// readsMsg is a message of the events stream as the assigner's handler reads it.
+type readsMsg struct {
+	jetstream.Msg
+	subject string
+	data    []byte
+}
+
+func (m readsMsg) Ack() error      { return nil }
+func (m readsMsg) Subject() string { return m.subject }
+func (m readsMsg) Data() []byte    { return m.data }

modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go

@@ -378,8 +378,8 @@ func RequestTools(r check.StateRead) []stdio.Tool {
 		Description: "The composition request a check sent to a node: the composer's whole input for that node (the node-engine's composition input, the case's " +
 			"manifests and verdicts, the seat list), kept exactly as sent so a session reproduces the check's composition (the composer run by hand over the same " +
 			"request gives the check's answer). Kept in the checker's state compositions, per check key (group.heads, the 12-digit heads hash) and node, " +
-			"0.7 to 0.8 MB each (kept compressed, in pieces that each fit the state's 256 KiB value), for 3 days (the state's ttl-seconds). With a check and a node, that request whole; with a check, " +
-			"the nodes it kept one for; with neither, every kept request, each with its size, digest and when it was kept. Read-only.",
+			"0.7 to 0.8 MB each (kept compressed, in pieces that each fit the state's 256 KiB value), for 3 days (the state's ttl-seconds), the state holding at most 256 MiB (max-mib, which mesh-grants raises a held bucket to). With a check and a node, that request whole; with a check, " +
+			"the nodes it kept one for; with neither, every kept request, each with its size, digest and when it was kept. Read-only; " + check.RequestsDropTool + " drops a check's.",
 		Input: map[string]any{"type": "object", "properties": map[string]any{
 			"check": map[string]any{"type": "string", "description": "the check's key, <group>.<heads hash>, as the verdict and composition name it"},
 			"node":  map[string]any{"type": "string", "description": "the node the request was sent for, with a check"}}},
@@ -390,6 +390,26 @@ func RequestTools(r check.StateRead) []stdio.Tool {
 		}}}
 }
 
+// RequestDropTools is the tool that drops a check's kept composition requests by hand, which frees the state when it is
+// full (tools-are-how-a-module-is-spoken-to): it records the hand act through the seat mesh-hand-acts first.
+func RequestDropTools(s check.StateDrop, acts views.HandActs) []stdio.Tool {
+	return []stdio.Tool{{Name: check.RequestsDropTool,
+		Description: "Drop the composition requests the checker kept for one check (" + check.RequestsReadTool + " lists them): every node's, and the pieces " +
+			"no other kept request names. The hand act is recorded through the seat mesh-hand-acts first, and a drop whose act could not be recorded is not made. " +
+			"Answers the nodes and pieces dropped. Use it when a check's words say its request could not be kept in the state compositions because it is full; " +
+			"the state itself drops a request 3 days after it was kept.",
+		Input: map[string]any{"type": "object", "required": []string{"check", "why"}, "properties": map[string]any{
+			"check": map[string]any{"type": "string", "description": "the check's key, <group>.<heads hash>, whose kept requests are dropped"},
+			"why":   map[string]any{"type": "string", "description": "why they are dropped, in words; recorded as the hand act"}}},
+		Run: func(args map[string]any) (any, error) {
+			key, _ := args["check"].(string)
+			why, _ := args["why"].(string)
+			return check.DropRequests(s, key, why, func(what string) error {
+				return acts.RecordHandAct(views.HandAct{What: what, Why: why, Verb: check.RequestsDropTool})
+			})
+		}}}
+}
+
 // RecoveryTools are the tools that steer a check by hand (tools-are-how-a-module-is-spoken-to): ask a check again whose
 // wait judged nothing (the lost check of 2026-10-28). They write the checker's verdicts through w.
 func RecoveryTools(r check.Reader, w check.VerdictEntry) []stdio.Tool {
@@ -469,6 +489,8 @@ func main() {
 	tools = append(tools, views.StatusTools(check.CheckerModule, "checker_first_views", "checker_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...)
 	tools = append(tools, SeatInputTools(r)...)
 	tools = append(tools, RequestTools(read(stdio.State(check.RequestsStateName)))...)
+	requests := stdio.State(check.RequestsStateName)
+	tools = append(tools, RequestDropTools(check.StateDrop{StateRead: read(requests), Delete: requests.Delete}, meshhandactsclient.Recorder{Ask: stdio.Ask})...)
 	if err := stdio.Serve("", logged(tools)); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err)
 		os.Exit(1)

modules/mesh-delivery-checker/internal/check/engines_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"mesh-delivery-checker/rules"
 	"strings"
 	"sync"
@@ -235,8 +236,34 @@ func TestTheCompositionIsMadeFromTheEnginesAnswers(t *testing.T) {
 		}
 		store.err = errors.New("the state is full")
 		raw, _ = ec.Compose(context.Background(), host)
-		if v := rules.JudgeEngines(raw); v.State != meshdeliverycheckerclient.CompositionFail || !strings.Contains(v.Words, "could not be kept") || !strings.Contains(v.Words, "the state is full") {
-			t.Errorf("a request that cannot be kept gave %+v", v)
+		if v := rules.JudgeEngines(raw); v.State != meshdeliverycheckerclient.CompositionFail || !strings.Contains(v.Words, "could not be kept") || !strings.Contains(v.Words, "the state is full") || !strings.Contains(v.Words, RequestsDropTool) {
+			t.Errorf("a request that cannot be kept gave %+v, which must name %s", v, RequestsDropTool)
+		}
+		// The drop tool: records the hand act first (a drop whose act is not recorded is not made), then drops that check's
+		// requests and the pieces no other kept request names; another check's stay.
+		store.err = nil
+		other := RequestKeeper{Store: store}
+		if err := other.Keep(context.Background(), "g.other", "anchor", []byte("another check's request")); err != nil {
+			t.Fatal(err)
+		}
+		drop := StateDrop{StateRead: store.read(), Delete: func(k string) error { store.mu.Lock(); delete(store.vals, k); store.mu.Unlock(); return nil }}
+		before := len(store.vals)
+		if _, err := DropRequests(drop, keyOfHost, "the state is full", func(string) error { return errors.New("no holder") }); err == nil || len(store.vals) != before {
+			t.Errorf("a drop whose hand act was not recorded was made: %v, %d of %d kept", err, len(store.vals), before)
+		}
+		var acts []string
+		answer, err := DropRequests(drop, keyOfHost, "the state is full", func(what string) error { acts = append(acts, what); return nil })
+		if b, _ := json.Marshal(answer); err != nil || len(acts) != 1 || !strings.Contains(string(b), `"dropped_requests":["anchor"]`) {
+			t.Errorf("the drop answered %s, %v, acts %v", b, err, acts)
+		}
+		if _, still := store.vals[RequestKey(keyOfHost, "anchor")]; still {
+			t.Errorf("the dropped check's request is kept still")
+		}
+		if got, err := ReadRequests(store.read(), "g.other", "anchor"); err != nil || !strings.Contains(fmt.Sprint(got), "another check's request") {
+			t.Errorf("another check's request after the drop: %v, %v", got, err)
+		}
+		if _, err := DropRequests(drop, keyOfHost, "again", func(string) error { return nil }); err == nil {
+			t.Errorf("a check with no kept request was dropped")
 		}
 		ec.Requests = nil
 		// Issue 1921: the composer is built at the group's heads, also a member that publishes no package yet.

modules/mesh-delivery-checker/internal/check/requests.go

@@ -27,7 +27,8 @@ import (
 // a value of its own under RequestChunkKey, of at most RequestChunkBytes. The kept data is the checker's, a foundation
 // part's, so it stays in the old section `state` (ADR 0619 §2) and is read through the tool runner's state read, which
 // serves no object store of the old sections. The manifest's ttl-seconds is the lifetime (3 days) of the pointer and of
-// every piece: the bus drops a value past it, so nothing waits on a timer here. A check kept again overwrites the key;
+// every piece and its max-mib the most the state holds (256): the bus drops a value past the one and refuses a put past the
+// other, so nothing waits on a timer here; a state that is full says so in the check's words, which name RequestsDropTool. A check kept again overwrites the key;
 // a request whose pieces cannot all be put is refused naming the size (never dropped, never half kept: the pointer is
 // put after its pieces).
 const (
@@ -36,6 +37,8 @@ const (
 	RequestsBucket    = CheckerModule + "_" + RequestsStateName
 	// RequestsReadTool is the tool that reads a kept request.
 	RequestsReadTool = "checker_composition_requests"
+	// RequestsDropTool is the tool that drops a check's kept requests by hand (a state that is full is freed with it).
+	RequestsDropTool = "checker_composition_requests_drop"
 	// RequestChunkBytes is the most compressed bytes one piece holds: base64 of it, in its document, stays well under
 	// the state's 256 KiB value.
 	RequestChunkBytes = 128 << 10
@@ -110,8 +113,8 @@ func (k RequestKeeper) Keep(ctx context.Context, key, node string, body []byte)
 	digest := hex.EncodeToString(sum[:])
 	pieces := (packed.Len() + RequestChunkBytes - 1) / RequestChunkBytes
 	fail := func(err error) error {
-		return fmt.Errorf("the composition request of check %s to %s (%d bytes, %d compressed in %d pieces) could not be kept in the state %s: %w",
-			key, node, len(body), packed.Len(), pieces, RequestsStateName, err)
+		return fmt.Errorf("the composition request of check %s to %s (%d bytes, %d compressed in %d pieces) could not be kept in the state %s: %w (when the state is full, free it by dropping the kept requests of a check no longer read, with the tool %s)",
+			key, node, len(body), packed.Len(), pieces, RequestsStateName, err, RequestsDropTool)
 	}
 	for i := 0; i < pieces; i++ {
 		end := min((i+1)*RequestChunkBytes, packed.Len())
@@ -221,3 +224,77 @@ func unpackRequest(s StateRead, raw []byte) (KeptRequest, []byte, error) {
 	}
 	return kept, body, nil
 }
+
+// StateDrop is the state of kept requests as the drop tool needs it: its keys, a value, and a key removed.
+type StateDrop struct {
+	StateRead
+	Delete func(key string) error
+}
+
+// DropRequests drops the requests kept for a check, after record (the hand act, recorded first: a drop that is not
+// recorded is not made), and answers what it dropped: the nodes, and the pieces. A piece another kept request still names
+// (the same request kept for another check has the same digest) stays. A check naming no kept request is refused.
+func DropRequests(s StateDrop, key, why string, record func(what string) error) (any, error) {
+	if strings.TrimSpace(key) == "" || strings.TrimSpace(why) == "" {
+		return nil, fmt.Errorf("drop takes check (its key, group.heads) and why")
+	}
+	if s.Keys == nil || s.Get == nil || s.Delete == nil {
+		return nil, fmt.Errorf("the state %s cannot be changed here", RequestsStateName)
+	}
+	keys, err := s.Keys()
+	if err != nil {
+		return nil, err
+	}
+	sort.Strings(keys)
+	var drop []string
+	kept := map[string]bool{} // digests the requests that stay name
+	dropping := map[string]bool{}
+	for _, k := range keys {
+		if isPieceKey(k) {
+			continue
+		}
+		raw, err := s.Get(k)
+		if err != nil {
+			return nil, err
+		}
+		var r KeptRequest
+		if len(raw) == 0 || json.Unmarshal(raw, &r) != nil {
+			continue
+		}
+		if strings.HasPrefix(k, key+"/") {
+			drop = append(drop, k)
+			dropping[r.Digest] = true
+			continue
+		}
+		kept[r.Digest] = true
+	}
+	if len(drop) == 0 {
+		return nil, fmt.Errorf("no composition request of check %s is kept (none was sent, or it is past its lifetime): nothing was dropped", key)
+	}
+	if err := record(fmt.Sprintf("dropped the %d kept composition requests of check %s (%s)", len(drop), key, RequestsStateName)); err != nil {
+		return nil, fmt.Errorf("the hand act was not recorded, so nothing was dropped: %w", err)
+	}
+	var pieces int
+	for _, k := range drop {
+		if err := s.Delete(k); err != nil {
+			return nil, fmt.Errorf("dropping %s: %w", k, err)
+		}
+	}
+	for _, k := range keys {
+		if !isPieceKey(k) {
+			continue
+		}
+		digest := strings.Split(strings.TrimPrefix(k, "pieces/"), "/")[0]
+		if dropping[digest] && !kept[digest] {
+			if err := s.Delete(k); err != nil {
+				return nil, fmt.Errorf("dropping %s: %w", k, err)
+			}
+			pieces++
+		}
+	}
+	nodes := make([]string, len(drop))
+	for i, k := range drop {
+		nodes[i] = strings.TrimPrefix(k, key+"/")
+	}
+	return map[string]any{"check": key, "dropped_requests": nodes, "dropped_pieces": pieces, "why": why}, nil
+}

modules/mesh-delivery-checker/manifest_test.go

@@ -23,6 +23,7 @@ type state struct {
 	Name    string `json:"name"`
 	History int    `json:"history,omitempty"`
 	TTL     int    `json:"ttl-seconds,omitempty"`
+	MaxMiB  int    `json:"max-mib,omitempty"`
 }
 
 type artifact struct {
@@ -108,7 +109,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
 		t.Errorf("module = %q", m.Module)
 	}
 	// The histories of the checker's four seat states (checked against internal/check names below); the names follow the seat mesh-delivery-checker.
-	if want := []state{{Name: "asks", History: 1}, {Name: "hearing", History: 1}, {Name: "inputs", TTL: 86400}, {Name: "compositions", TTL: 259200}, {Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}, {Name: "first-views", History: 1}, {Name: "findings", History: 1}}; !reflect.DeepEqual(m.State, want) {
+	if want := []state{{Name: "asks", History: 1}, {Name: "hearing", History: 1}, {Name: "inputs", TTL: 86400}, {Name: "compositions", TTL: 259200, MaxMiB: 256}, {Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}, {Name: "first-views", History: 1}, {Name: "findings", History: 1}}; !reflect.DeepEqual(m.State, want) {
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains",
@@ -186,7 +187,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
 	// The read tools (issue 1686) are served by the tools bundle, not by the checker's process.
 	// The try's tools (ADR 0501 §5; the end by hand, issue 734) beside them: mesh-controller's try calls them. A lost check
 	// is asked again by hand (checker_ask_again).
-	if want := []string{"checker_ask_again", "checker_composition_requests", "checker_first_view_again", "checker_first_views", "checker_inputs", "checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_tries", "checker_try", "checker_try_end", "checker_try_read", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) {
+	if want := []string{"checker_ask_again", "checker_composition_requests", "checker_composition_requests_drop", "checker_first_view_again", "checker_first_views", "checker_inputs", "checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_tries", "checker_try", "checker_try_end", "checker_try_read", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) {
 		t.Errorf("tools = %v, want %v", m.Tools, want)
 	}
 	// The try is the checker's own seat's verbs (ADR 0547 §10, issue 774), served beside the tools by the same code; so is

modules/mesh-delivery-checker/module.json

@@ -1,6 +1,6 @@
 {
   "module": "mesh-delivery-checker",
-  "version": "1",
+  "version": "2",
   "slug": "checker",
   "claims": [
     {
@@ -37,7 +37,7 @@
     { "name": "asks", "history": 1 },
     { "name": "hearing", "history": 1 },
     { "name": "inputs", "ttl-seconds": 86400 },
-    { "name": "compositions", "ttl-seconds": 259200 },
+    { "name": "compositions", "ttl-seconds": 259200, "max-mib": 256 },
     { "name": "kept", "history": 1 },
     { "name": "owed", "history": 5 },
     { "name": "previews", "history": 5 },
@@ -64,6 +64,7 @@
   "tools": [
     "checker_ask_again",
     "checker_composition_requests",
+    "checker_composition_requests_drop",
     "checker_inputs",
     "checker_kept",
     "checker_previews",

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,

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

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

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

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

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