jschoubben
7b02f63daf
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.
329 lines
10 KiB
Go
329 lines
10 KiB
Go
package process
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"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
|
|
// caught at once by Conn's own ErrorHandler (conn.go withRefusal/publishRefused), never waited out, so
|
|
// this stays the ordinary request bound and is not exposed for a test to shrink.
|
|
const stateTimeout = 30 * time.Second
|
|
|
|
// State is one of this process's module's states, read and written on this connection (novox/hq ADR 0436
|
|
// clause 4), in the bucket <module>_<name> (mesh-controller internal/broker/state.go:49) — only as the
|
|
// account's grants allow. It satisfies go/asker's Store.
|
|
type State struct {
|
|
Module string
|
|
Name string
|
|
// kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a
|
|
// unit test, a fake that never reaches a bus.
|
|
kv func(ctx context.Context) (jetstream.KeyValue, error)
|
|
// closed says the connection is gone, so a watch ends; nil in a unit test.
|
|
closed func() bool
|
|
}
|
|
|
|
// bucket is this state's name on the bus.
|
|
func (s State) bucket() string { return s.Module + "_" + s.Name }
|
|
|
|
// State is the named state this connection's module keeps, bound lazily to its bucket on each call so a
|
|
// bucket the account is not granted fails at the call that needed it, naming the bucket. The bind itself
|
|
// (js.KeyValue) is what asks $JS.API.STREAM.INFO.KV_<bucket>, so it is what withRefusal watches: a bucket
|
|
// not granted is refused there, promptly, as nats.ErrPermissionViolation.
|
|
func (c *Conn) State(name string) State {
|
|
bucket := c.account.Module + "_" + name
|
|
return State{Module: c.account.Module, Name: name, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
|
|
if err := c.refusedUnannounced(name); err != nil {
|
|
return nil, err
|
|
}
|
|
js, err := jetstream.New(c.nc)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
subject := "$JS.API.STREAM.INFO.KV_" + bucket
|
|
ctx, stop := c.withRefusal(ctx, subject)
|
|
defer stop()
|
|
kv, err := js.KeyValue(ctx, bucket)
|
|
if err != nil {
|
|
if errors.Is(context.Cause(ctx), errPermissionRefused) {
|
|
return nil, fmt.Errorf("%w: publishing %s", nats.ErrPermissionViolation, subject)
|
|
}
|
|
return nil, err
|
|
}
|
|
return kv, nil
|
|
}}
|
|
}
|
|
|
|
func (s State) store(ctx context.Context) (jetstream.KeyValue, error) {
|
|
kv, err := s.kv(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%s's state %s (bucket %s): %w", s.Module, s.Name, s.bucket(), err)
|
|
}
|
|
return kv, nil
|
|
}
|
|
|
|
// Get is a key's current value, or nil when it has none.
|
|
func (s State) Get(key string) (*stdio.StateEntry, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
|
defer cancel()
|
|
kv, err := s.store(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
entry, err := kv.Get(ctx, key)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
|
|
}
|
|
return &stdio.StateEntry{Key: key, Value: json.RawMessage(entry.Value()), Revision: entry.Revision()}, nil
|
|
}
|
|
|
|
// Create writes a key only when it has none, and returns the revision; a key that has one is refused as
|
|
// stdio.ErrChanged.
|
|
func (s State) Create(key string, value any) (uint64, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
|
defer cancel()
|
|
kv, err := s.store(ctx)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
data, err := json.Marshal(value)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
rev, err := kvlimit.Create(ctx, kv, key, data)
|
|
if errors.Is(err, jetstream.ErrKeyExists) {
|
|
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
|
}
|
|
if err != nil {
|
|
return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
|
|
}
|
|
return rev, nil
|
|
}
|
|
|
|
// MinTTL is the shortest time to live a key may be given: the bus counts a key's age in whole seconds.
|
|
const MinTTL = time.Second
|
|
|
|
// CreateTTL writes a key only when it has none, with its own time to live (novox/hq ADR 0564 clauses 4 and 7, issue
|
|
// 1952 S6): the bus removes it when ttl passes and its watchers hear the removal, so an expiry is an event and no timer
|
|
// runs. A key that has a value is refused as stdio.ErrChanged: an act begun once cannot be begun again. The bucket must
|
|
// take keys with their own time to live (a state declared `key-ttl`).
|
|
func (s State) CreateTTL(key string, value any, ttl time.Duration) (uint64, error) {
|
|
if ttl < MinTTL {
|
|
return 0, fmt.Errorf("%s's state %s, key %s: a time to live of %s is shorter than the bus keeps (%s)", s.Module, s.Name, key, ttl, MinTTL)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
|
defer cancel()
|
|
kv, err := s.store(ctx)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
data, err := json.Marshal(value)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
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)
|
|
}
|
|
if err != nil {
|
|
return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
|
|
}
|
|
return rev, nil
|
|
}
|
|
|
|
// RenewTTL writes a key whole with a fresh time to live, whatever it held: a pulse its signal renews (ADR 0564 clause
|
|
// 4). A time to live is set only when a key is created, so a live key is deleted first; its watchers hear the delete and
|
|
// then the new value.
|
|
func (s State) RenewTTL(key string, value any, ttl time.Duration) (uint64, error) {
|
|
rev, err := s.CreateTTL(key, value, ttl)
|
|
if !errors.Is(err, stdio.ErrChanged) {
|
|
return rev, err
|
|
}
|
|
if err := s.Delete(key); err != nil {
|
|
return 0, err
|
|
}
|
|
return s.CreateTTL(key, value, ttl)
|
|
}
|
|
|
|
// Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile
|
|
// is refused as stdio.ErrChanged.
|
|
func (s State) Update(key string, value any, revision uint64) (uint64, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
|
defer cancel()
|
|
kv, err := s.store(ctx)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
data, err := json.Marshal(value)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
rev, err := kvlimit.Update(ctx, kv, key, data, revision)
|
|
if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
|
|
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
|
}
|
|
if err != nil {
|
|
return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
|
|
}
|
|
return rev, nil
|
|
}
|
|
|
|
// Put writes a key whole, whatever it held, and returns the revision written.
|
|
func (s State) Put(key string, value any) (uint64, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
|
defer cancel()
|
|
kv, err := s.store(ctx)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
data, err := json.Marshal(value)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
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)
|
|
}
|
|
return rev, nil
|
|
}
|
|
|
|
// Keys is every key with a current value, sorted; none is an empty list.
|
|
func (s State) Keys() ([]string, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
|
defer cancel()
|
|
kv, err := s.store(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
keys, err := kv.Keys(ctx)
|
|
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
|
return []string{}, nil
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%s's state %s, keys: %w", s.Module, s.Name, err)
|
|
}
|
|
sort.Strings(keys)
|
|
return keys, nil
|
|
}
|
|
|
|
// watchPattern is a key narrowing as the bus names it: `*` stays one name, `**` is the rest, empty is every key.
|
|
func watchPattern(key string) string {
|
|
if key == "" {
|
|
return jetstream.AllKeys
|
|
}
|
|
tokens := strings.Split(key, ".")
|
|
for i, t := range tokens {
|
|
if t == "**" {
|
|
tokens[i] = ">"
|
|
}
|
|
}
|
|
return strings.Join(tokens, ".")
|
|
}
|
|
|
|
// Watch hands handle every current value — none that is deleted, each with Current set — and then every
|
|
// change in order, and returns once the current values have all been handled. key narrows it: a key, with
|
|
// `*` for one name and `**` for the rest; empty is every key. The changes go on for the life of the
|
|
// connection; an error from handle, or the connection closing, ends them. Before the current values are
|
|
// handled, either is returned as the error.
|
|
func (s State) Watch(key string, handle func(stdio.StateChange) error) error {
|
|
// the watch outlives this call, so its context is not the call's timeout: the connection ends it.
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
bind, stopBind := context.WithTimeout(ctx, stateTimeout)
|
|
kv, err := s.store(bind)
|
|
stopBind()
|
|
if err != nil {
|
|
cancel()
|
|
return err
|
|
}
|
|
w, err := kv.Watch(ctx, watchPattern(key))
|
|
if err != nil {
|
|
cancel()
|
|
return fmt.Errorf("%s's state %s, watching %q: %w", s.Module, s.Name, key, err)
|
|
}
|
|
end := func() { w.Stop(); cancel() }
|
|
next := func() (jetstream.KeyValueEntry, bool, error) {
|
|
tick := time.NewTicker(250 * time.Millisecond)
|
|
defer tick.Stop()
|
|
for {
|
|
select {
|
|
case e, open := <-w.Updates():
|
|
if !open {
|
|
return nil, false, fmt.Errorf("%s's state %s: the watch ended", s.Module, s.Name)
|
|
}
|
|
return e, true, nil
|
|
case <-tick.C:
|
|
if s.closed != nil && s.closed() {
|
|
return nil, false, fmt.Errorf("%s's state %s: the connection closed", s.Module, s.Name)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
deliver := func(e jetstream.KeyValueEntry, current bool) error {
|
|
c := stdio.StateChange{State: s.Name, Key: e.Key(), Op: "put", Revision: e.Revision(), Current: current}
|
|
if e.Operation() == jetstream.KeyValuePut {
|
|
c.Value = json.RawMessage(e.Value())
|
|
} else {
|
|
c.Op = "delete"
|
|
}
|
|
return handle(c)
|
|
}
|
|
for {
|
|
e, _, err := next()
|
|
if err != nil {
|
|
end()
|
|
return err
|
|
}
|
|
if e == nil {
|
|
break
|
|
}
|
|
if e.Operation() != jetstream.KeyValuePut {
|
|
continue
|
|
}
|
|
if err := deliver(e, true); err != nil {
|
|
end()
|
|
return err
|
|
}
|
|
}
|
|
go func() {
|
|
defer end()
|
|
for {
|
|
e, _, err := next()
|
|
if err != nil || e == nil {
|
|
return
|
|
}
|
|
if deliver(e, false) != nil {
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
return nil
|
|
}
|
|
|
|
// Delete removes a key. A key that was not there is not an error.
|
|
func (s State) Delete(key string) error {
|
|
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
|
defer cancel()
|
|
kv, err := s.store(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
|
|
}
|
|
return nil
|
|
}
|