diff --git a/go/cmd/mesh-hand-compose/main.go b/go/cmd/mesh-hand-compose/main.go index cf59e98..624b0b2 100644 --- a/go/cmd/mesh-hand-compose/main.go +++ b/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 diff --git a/go/kvlimit/kvlimit.go b/go/kvlimit/kvlimit.go new file mode 100644 index 0000000..95d51ee --- /dev/null +++ b/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) +} diff --git a/go/process/buckets.go b/go/process/buckets.go index bc2421c..a84d406 100644 --- a/go/process/buckets.go +++ b/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++ diff --git a/go/process/state.go b/go/process/state.go index 90523b4..d2a462f 100644 --- a/go/process/state.go +++ b/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) } diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index 7d86e70..588959d 100644 --- a/go/process/state_integration_test.go +++ b/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) } diff --git a/go/state.go b/go/state.go index ac3d5fd..ad49de2 100644 --- a/go/state.go +++ b/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. diff --git a/go/state_test.go b/go/state_test.go index 7b5f109..dc80765 100644 --- a/go/state_test.go +++ b/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} {