Refuse a key-value write too large for its bucket before it is put (issue 774)

mesh-grants' sealed grants and the controller's merge record each met "message size exceeds maximum" (JetStream
10054) only after the put, and a writer that retries looped on a value that can never land; an update over the
limit was even answered as a revision mismatch, which every compare-and-set writer retries.

Quick fix: cap the two values in their modules. Thorough fix, chosen because the kind was met twice and every
module writes through the SDK: the package kvlimit checks a value against its bucket's largest message (read from
its stream and kept) before the put, naming the bucket, the key, the size and the limit, and turns the bus's own
refusal into the same *TooLarge (errors.Is kvlimit.ErrTooLarge), so no writer retries it. The SDK's own writers
use it: process.State (Put, Create, CreateTTL, Update), process.AwaitedBucket, the runtime state of a bundle
(stdio KeptState, against the module state's 256 KiB) and mesh-hand-compose. A value that can outgrow its bucket
is split over keys or kept in the object store with its name in the key.

go/cmd/mesh-hand-compose/main.go

@@ -32,6 +32,7 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/compose"
 	"git.novox.be/novox/mesh-sdk/go/declaration"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/kvlimit"
 )
 
 func main() {
@@ -190,7 +191,7 @@ func writeIt(url, creds, node, bucket, key string, encoded []byte, digest string
 	if err != nil {
 		return fmt.Errorf("the bucket %s: %w", bucket, err)
 	}
-	if _, err := composed.Put(ctx, key, encoded); err != nil {
+	if _, err := kvlimit.Put(ctx, composed, key, encoded); err != nil {
 		return fmt.Errorf("%s in %s: %w", key, bucket, err)
 	}
 	value, err := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: digest, Judgement: delivery.JudgedCannot,
@@ -202,7 +203,7 @@ func writeIt(url, creds, node, bucket, key string, encoded []byte, digest string
 	if err != nil {
 		return fmt.Errorf("the bucket %s: %w", delivery.ControllerComposedBucket, err)
 	}
-	if _, err := digests.Put(ctx, key, value); err != nil {
+	if _, err := kvlimit.Put(ctx, digests, key, value); err != nil {
 		return fmt.Errorf("%s in %s: %w", key, delivery.ControllerComposedBucket, err)
 	}
 	return nil

go/kvlimit/kvlimit.go

@@ -0,0 +1,175 @@
+// Package kvlimit refuses a key-value write whose value is larger than its bucket takes, before it is put, naming the
+// bucket, the key, the size and the limit (issue 774, sweep item 5). The bus refuses such a write too, but only after
+// the put, as "message size exceeds maximum" (JetStream error 10054); a writer that retries on errors then loops on a
+// value that can never land. A refusal from here, or the bus's own one passed through Refused, is a *TooLarge that
+// errors.Is ErrTooLarge: a writer stops on it and says it, it never writes the same value again.
+//
+// A value that may outgrow its bucket is shaped one of two ways, never retried:
+//   - split over keys: one key per part that changes on its own (per node, per module), each well under the limit;
+//   - or kept in the object store, with only its name and digest in the key.
+package kvlimit
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// ModuleStateValueBytes is the most a value in a module's declared state may hold: every module bucket is made with
+// it (buscompose.StateMaxValueBytes).
+const ModuleStateValueBytes = 256 * 1024
+
+// HeaderRoom is what a write's headers may take of the limit: the bus counts headers and value together, and a create,
+// an update or a key's time to live adds one header each. A value closer to the limit than this is refused.
+const HeaderRoom = 128
+
+// codeMessageExceedsMaximum is JetStream's error code for a message over the stream's largest (JSStreamMessageExceedsMaximumErr).
+const codeMessageExceedsMaximum jetstream.ErrorCode = 10054
+
+// ErrTooLarge is what every *TooLarge is: a value its bucket does not take, which no retry will change.
+var ErrTooLarge = errors.New("a value larger than its bucket takes")
+
+// TooLarge is one refused write. Limit is zero when the bus refused it and its limit was not known here.
+type TooLarge struct {
+	Bucket, Key string
+	Size, Limit int64
+}
+
+func (e *TooLarge) Error() string {
+	limit := "the most its bucket takes"
+	if e.Limit > 0 {
+		limit = fmt.Sprintf("the %d bytes the bucket takes (headers included)", e.Limit)
+	}
+	return fmt.Sprintf("the value for %s in %s is %d bytes, over %s: it is refused and written again it is refused again; "+
+		"split it over keys or keep it in the object store with its name in the key", e.Key, e.Bucket, e.Size, limit)
+}
+
+func (e *TooLarge) Unwrap() error { return ErrTooLarge }
+
+// Check refuses a value of size bytes for key in bucket when it does not fit limit with HeaderRoom to spare; a limit of
+// zero or less is no limit.
+func Check(bucket, key string, size int, limit int64) error {
+	if limit > 0 && int64(size)+HeaderRoom > limit {
+		return &TooLarge{Bucket: bucket, Key: key, Size: int64(size), Limit: limit}
+	}
+	return nil
+}
+
+// Refused is err as a *TooLarge when it is the bus's refusal of a value too large (the stream's 10054, the client's
+// maximum payload, or either said in words by a runtime that passed it on), and err unchanged otherwise.
+func Refused(bucket, key string, size int, limit int64, err error) error {
+	if err == nil || errors.Is(err, ErrTooLarge) {
+		return err
+	}
+	var api *jetstream.APIError
+	tooLarge := errors.As(err, &api) && api.ErrorCode == codeMessageExceedsMaximum
+	tooLarge = tooLarge || errors.Is(err, nats.ErrMaxPayload)
+	words := err.Error()
+	tooLarge = tooLarge || strings.Contains(words, "message size exceeds maximum") || strings.Contains(words, "maximum payload exceeded")
+	if !tooLarge {
+		return err
+	}
+	forget(bucket)
+	return fmt.Errorf("%w (the bus: %v)", &TooLarge{Bucket: bucket, Key: key, Size: int64(size), Limit: limit}, err)
+}
+
+// limits are the buckets' largest messages as read, by bucket name: a limit is kept once read, and only a limit (an
+// unlimited bucket is read again each time). A kept limit that refuses a value is read again before the refusal is
+// given, so a bucket made again with a larger limit is never refused by an old one; one made with a smaller limit is
+// refused by the bus, and Refused forgets the kept one.
+var limits sync.Map // bucket name → int64
+
+func forget(bucket string) { limits.Delete(bucket) }
+
+// LimitOf is the largest message kv's bucket takes, read from its stream and kept; zero when it has no limit or it
+// cannot be read (the write then goes, and Refused names the bus's refusal).
+func LimitOf(ctx context.Context, kv jetstream.KeyValue) int64 {
+	if l, ok := limits.Load(nameOf(kv)); ok {
+		return l.(int64)
+	}
+	return readLimit(ctx, kv)
+}
+
+func readLimit(ctx context.Context, kv jetstream.KeyValue) (limit int64) {
+	name := nameOf(kv)
+	if name == "" {
+		return 0
+	}
+	// A test's fake that implements only the writes it needs panics on Status; its limit is not known.
+	defer func() {
+		if recover() != nil {
+			limit = 0
+		}
+	}()
+	st, err := kv.Status(ctx)
+	if err != nil {
+		return 0
+	}
+	s, ok := st.(*jetstream.KeyValueBucketStatus)
+	if !ok || s.StreamInfo() == nil || s.StreamInfo().Config.MaxMsgSize <= 0 {
+		forget(name)
+		return 0
+	}
+	limit = int64(s.StreamInfo().Config.MaxMsgSize)
+	limits.Store(name, limit)
+	return limit
+}
+
+// fits is the limit of kv's bucket and nil when a value of size bytes fits it, or the *TooLarge refusing it, checked
+// against a freshly read limit.
+func fits(ctx context.Context, kv jetstream.KeyValue, key string, size int) (int64, error) {
+	limit := LimitOf(ctx, kv)
+	if Check(nameOf(kv), key, size, limit) == nil {
+		return limit, nil
+	}
+	limit = readLimit(ctx, kv)
+	return limit, Check(nameOf(kv), key, size, limit)
+}
+
+// nameOf is kv's bucket name, or "" for none or a test's fake that does not say it.
+func nameOf(kv jetstream.KeyValue) (name string) {
+	if kv == nil {
+		return ""
+	}
+	defer func() {
+		if recover() != nil {
+			name = ""
+		}
+	}()
+	return kv.Bucket()
+}
+
+// Put is kv.Put, refused before it is put when value does not fit the bucket.
+func Put(ctx context.Context, kv jetstream.KeyValue, key string, value []byte) (uint64, error) {
+	limit, err := fits(ctx, kv, key, len(value))
+	if err != nil {
+		return 0, err
+	}
+	rev, err := kv.Put(ctx, key, value)
+	return rev, Refused(nameOf(kv), key, len(value), limit, err)
+}
+
+// Create is kv.Create, refused before it is put when value does not fit the bucket.
+func Create(ctx context.Context, kv jetstream.KeyValue, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error) {
+	limit, err := fits(ctx, kv, key, len(value))
+	if err != nil {
+		return 0, err
+	}
+	rev, err := kv.Create(ctx, key, value, opts...)
+	return rev, Refused(nameOf(kv), key, len(value), limit, err)
+}
+
+// Update is kv.Update, refused before it is put when value does not fit the bucket.
+func Update(ctx context.Context, kv jetstream.KeyValue, key string, value []byte, revision uint64) (uint64, error) {
+	limit, err := fits(ctx, kv, key, len(value))
+	if err != nil {
+		return 0, err
+	}
+	rev, err := kv.Update(ctx, key, value, revision)
+	return rev, Refused(nameOf(kv), key, len(value), limit, err)
+}

go/process/buckets.go

@@ -10,6 +10,8 @@ import (
 
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/kvlimit"
 )
 
 // **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the delivery
@@ -167,7 +169,7 @@ func (b *AwaitedBucket) Put(ctx context.Context, key string, value []byte) (uint
 		kv = b.kv
 		b.mu.Unlock()
 	}
-	return kv.Put(ctx, key, value)
+	return kvlimit.Put(ctx, kv, key, value)
 }
 
 // bind tries one STREAM.INFO of the bucket and, when it is there, binds it; it answers the bucket or nil.
@@ -196,7 +198,7 @@ func (b *AwaitedBucket) bound(kv jetstream.KeyValue, appeared bool) jetstream.Ke
 	written, failed := 0, 0
 	for _, key := range b.order {
 		ctx, cancel := context.WithTimeout(b.ctx, bindWithin)
-		_, err := kv.Put(ctx, key, b.pending[key])
+		_, err := kvlimit.Put(ctx, kv, key, b.pending[key])
 		cancel()
 		if err != nil {
 			failed++

go/process/state.go

@@ -13,6 +13,7 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/kvlimit"
 )
 
 // stateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a denied publish is
@@ -103,7 +104,7 @@ func (s State) Create(key string, value any) (uint64, error) {
 	if err != nil {
 		return 0, err
 	}
-	rev, err := kv.Create(ctx, key, data)
+	rev, err := kvlimit.Create(ctx, kv, key, data)
 	if errors.Is(err, jetstream.ErrKeyExists) {
 		return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
 	}
@@ -134,7 +135,7 @@ func (s State) CreateTTL(key string, value any, ttl time.Duration) (uint64, erro
 	if err != nil {
 		return 0, err
 	}
-	rev, err := kv.Create(ctx, key, data, jetstream.KeyTTL(ttl))
+	rev, err := kvlimit.Create(ctx, kv, key, data, jetstream.KeyTTL(ttl))
 	if errors.Is(err, jetstream.ErrKeyExists) {
 		return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
 	}
@@ -171,7 +172,7 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) {
 	if err != nil {
 		return 0, err
 	}
-	rev, err := kv.Update(ctx, key, data, revision)
+	rev, err := kvlimit.Update(ctx, kv, key, data, revision)
 	if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
 		return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
 	}
@@ -193,7 +194,7 @@ func (s State) Put(key string, value any) (uint64, error) {
 	if err != nil {
 		return 0, err
 	}
-	rev, err := kv.Put(ctx, key, data)
+	rev, err := kvlimit.Put(ctx, kv, key, data)
 	if err != nil {
 		return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
 	}

go/process/state_integration_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/nats-io/nats.go/jetstream"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/kvlimit"
 )
 
 // 1188/AC1: given a process connection to a test bus granted the bucket mesh-mail_sends, when it creates a
@@ -148,7 +149,8 @@ func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) {
 func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) {
 	url, account := grantedSendProcess(t)
 	_, adminJetStream := adminJS(t, url)
-	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil {
+	admins, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends", MaxValueSize: 1024})
+	if err != nil {
 		t.Fatalf("the admin could not make the bucket: %v", err)
 	}
 	conn, err := Connect(account)
@@ -158,6 +160,27 @@ func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) {
 	defer conn.Close()
 	s := conn.State("sends")
 
+	// Issue 774 (sweep item 5): a value over the bucket's largest is refused before it is put, naming the bucket, the key,
+	// its size and the limit, as kvlimit.ErrTooLarge, so a writer stops on it; nothing reaches the bucket.
+	big := strings.Repeat("x", 2000)
+	for name, write := range map[string]func() (uint64, error){
+		"put":    func() (uint64, error) { return s.Put("big", big) },
+		"create": func() (uint64, error) { return s.Create("big", big) },
+		"update": func() (uint64, error) { return s.Update("big", big, 1) },
+	} {
+		_, err := write()
+		var tooLarge *kvlimit.TooLarge
+		if !errors.Is(err, kvlimit.ErrTooLarge) || !errors.As(err, &tooLarge) || tooLarge.Limit != 1024 ||
+			tooLarge.Bucket != "mesh-mail_sends" || tooLarge.Key != "big" || tooLarge.Size != 2002 || strings.Contains(err.Error(), "the bus:") {
+			t.Errorf("a %s of 2002 bytes into a bucket of 1024: %v", name, err)
+		}
+	}
+	// The bus's own refusal (JetStream 10054), as a writer that put past the check meets it, is the same refusal.
+	_, err = admins.Put(ctxFor(t), "big", []byte(big))
+	if err == nil || !errors.Is(kvlimit.Refused("mesh-mail_sends", "big", 2000, 1024, err), kvlimit.ErrTooLarge) {
+		t.Errorf("the bus's refusal is not ErrTooLarge: %v", err)
+	}
+
 	if keys, err := s.Keys(); err != nil || len(keys) != 0 {
 		t.Fatalf("Keys of an empty bucket: %v, %v", keys, err)
 	}

go/state.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"fmt"
 	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/kvlimit"
 )
 
 // State — the current value of something a module keeps on the bus, which every machine sees,
@@ -64,12 +66,30 @@ func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, err
 	if err := refusedUnannounced(s.name); err != nil {
 		return nil, err
 	}
+	// A write's value is measured before it is sent: one over the bucket's limit is refused here, by name, and a
+	// refusal the bus gives anyway comes back as the same *kvlimit.TooLarge, so no writer retries it (issue 774).
+	size := -1
+	key, _ := params["key"].(string)
+	if value, ok := params["value"]; ok {
+		data, err := json.Marshal(value)
+		if err != nil {
+			return nil, fmt.Errorf("the value for %s in the state %s: %w", key, s.name, err)
+		}
+		if err := kvlimit.Check("the state "+s.name, key, len(data), kvlimit.ModuleStateValueBytes); err != nil {
+			return nil, err
+		}
+		params["value"], size = json.RawMessage(data), len(data)
+	}
 	c, err := live()
 	if err != nil {
 		return nil, err
 	}
 	params["state"] = s.name
-	return c.ask("mesh/state."+verb, params, stateTimeout)
+	raw, err := c.ask("mesh/state."+verb, params, stateTimeout)
+	if size >= 0 {
+		err = kvlimit.Refused("the state "+s.name, key, size, kvlimit.ModuleStateValueBytes, err)
+	}
+	return raw, err
 }
 
 // Get is a key's current value, or nil when it has none.

go/state_test.go

@@ -122,6 +122,21 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) {
 	if r := next(); r["id"] != float64(2) || r["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"] != "9" {
 		t.Fatalf("the put answered %v", r)
 	}
+	// Issue 774 (sweep item 5): a value over a module state's largest is refused before the runtime is asked, naming the
+	// key, its size and the limit; one the runtime's bus refuses anyway comes back as the same refusal.
+	send(map[string]any{"jsonrpc": "2.0", "id": 30, "method": "tools/call", "params": map[string]any{"name": "remember",
+		"arguments": map[string]any{"key": "all.big", "value": strings.Repeat("x", 300*1024)}}})
+	if r := next(); r["id"] != float64(30) || !strings.Contains(fmt.Sprint(r), "all.big in the state servers is 307202 bytes, over the 262144") {
+		t.Fatalf("a value over the limit was not refused before the runtime was asked: %v", r)
+	}
+	send(map[string]any{"jsonrpc": "2.0", "id": 31, "method": "tools/call", "params": map[string]any{"name": "remember",
+		"arguments": map[string]any{"key": "all.near", "value": "x"}}})
+	near := next()
+	send(map[string]any{"jsonrpc": "2.0", "id": near["id"], "error": map[string]any{"code": -32000,
+		"message": "nats: API error: code=400 err_code=10054 description=message size exceeds maximum allowed"}})
+	if r := next(); r["id"] != float64(31) || !strings.Contains(fmt.Sprint(r), "written again it is refused again") {
+		t.Fatalf("the bus's refusal was not named as too large: %v", r)
+	}
 	// PutTTL and CreateTTL ask the runtime's put and create with the key's time to live in whole seconds; a create the
 	// runtime refuses as changed (-32010) is ErrChanged.
 	for i, create := range []bool{false, true} {