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.
234 lines
8.4 KiB
Go
234 lines
8.4 KiB
Go
package stdio
|
|
|
|
import (
|
|
"encoding/json"
|
|
"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,
|
|
// including one that joins later (novox/hq ADR 0201). Not history: that is an event. Never a secret,
|
|
// sealed or not. A module declares what it keeps (`state`) and what it reads (`reads`, as
|
|
// `<module>.<name>`) and names each here as it named it there; the runtime refuses, with the reason,
|
|
// what it did not declare.
|
|
|
|
// StateEntry is one key's current value.
|
|
type StateEntry struct {
|
|
Key string `json:"key"`
|
|
Value json.RawMessage `json:"value"`
|
|
Revision uint64 `json:"revision"`
|
|
}
|
|
|
|
// The operations a watch hands over (StateChange.Op).
|
|
const (
|
|
// OpPut is a key written.
|
|
OpPut = "put"
|
|
// OpDelete is a key deleted or purged.
|
|
OpDelete = "delete"
|
|
// OpExpire is a key whose own time to live ended (CreateTTL, PutTTL): the runtime says the bus's marker of it as
|
|
// "expire", so a watcher tells an expiry, which is the event a pulse or a bound waits for, from a delete.
|
|
OpExpire = "expire"
|
|
)
|
|
|
|
// StateChange is one change a watch hands over: a key put (OpPut), deleted (OpDelete) or expired (OpExpire).
|
|
type StateChange struct {
|
|
State string `json:"state"`
|
|
Key string `json:"key"`
|
|
Op string `json:"op"`
|
|
Value json.RawMessage `json:"value,omitempty"`
|
|
Revision uint64 `json:"revision"`
|
|
// Current is true for a value that was there when the watch began, false for a change since.
|
|
Current bool `json:"current"`
|
|
}
|
|
|
|
// StateRevision is one value a bucket still keeps for a key: what was written, at which revision, by which
|
|
// operation ("put" or "delete", whose value is null) and when (novox/hq ADR 0307).
|
|
type StateRevision struct {
|
|
Key string `json:"key"`
|
|
Value json.RawMessage `json:"value"`
|
|
Revision uint64 `json:"revision"`
|
|
Op string `json:"op"`
|
|
At time.Time `json:"at"`
|
|
}
|
|
|
|
// KeptState is one state this module keeps, or reads.
|
|
type KeptState struct{ name string }
|
|
|
|
// State is a state this module keeps (by its local name) or reads (`<module>.<name>`).
|
|
func State(name string) KeptState { return KeptState{name: name} }
|
|
|
|
const stateTimeout = 30 * time.Second
|
|
|
|
func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, error) {
|
|
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
|
|
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.
|
|
func (s KeptState) Get(key string) (*StateEntry, error) {
|
|
raw, err := s.ask("get", map[string]any{"key": key})
|
|
if err != nil || len(raw) == 0 || string(raw) == "null" {
|
|
return nil, err
|
|
}
|
|
var e StateEntry
|
|
return &e, json.Unmarshal(raw, &e)
|
|
}
|
|
|
|
// History is every value the bucket still keeps for key, oldest first — as many as the state declared it
|
|
// keeps, deletes included — and empty for a key it holds nothing for (novox/hq ADR 0307). A read, like Get:
|
|
// the owner's instances and its readers may ask it, and it writes nothing.
|
|
func (s KeptState) History(key string) ([]StateRevision, error) {
|
|
raw, err := s.ask("history", map[string]any{"key": key})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
revisions := []StateRevision{}
|
|
if len(raw) == 0 || string(raw) == "null" {
|
|
return revisions, nil
|
|
}
|
|
return revisions, json.Unmarshal(raw, &revisions)
|
|
}
|
|
|
|
// Put writes a key — the owner's instances only — and returns the revision written.
|
|
func (s KeptState) Put(key string, value any) (uint64, error) {
|
|
raw, err := s.ask("put", map[string]any{"key": key, "value": value})
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
var put struct {
|
|
Revision uint64 `json:"revision"`
|
|
}
|
|
return put.Revision, json.Unmarshal(raw, &put)
|
|
}
|
|
|
|
// PutTTL writes a key whole with its own time to live — the owner's instances only — and returns the revision written:
|
|
// the bus removes it when ttl passes and a watch hands that over as OpExpire. The state is declared `key-ttl`.
|
|
func (s KeptState) PutTTL(key string, value any, ttl time.Duration) (uint64, error) {
|
|
raw, err := s.ask("put", map[string]any{"key": key, "value": value, "ttl_seconds": ttlSeconds(ttl)})
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
var put struct {
|
|
Revision uint64 `json:"revision"`
|
|
}
|
|
return put.Revision, json.Unmarshal(raw, &put)
|
|
}
|
|
|
|
// ttlSeconds is a time to live as the runtime takes it, whole seconds, at least one.
|
|
func ttlSeconds(ttl time.Duration) int64 {
|
|
if s := int64(ttl / time.Second); s > 1 {
|
|
return s
|
|
}
|
|
return 1
|
|
}
|
|
|
|
// Delete removes a key — the owner's instances only. A key that was not there is not an error.
|
|
func (s KeptState) Delete(key string) error {
|
|
_, err := s.ask("delete", map[string]any{"key": key})
|
|
return err
|
|
}
|
|
|
|
// Increment adds one to a counter kept at key and returns the number it now holds: one for a key with no
|
|
// value. Read, then written only if the key is still at the revision read (Create when it had none, Update
|
|
// when it had one), so of any number of writers each number is handed out once and no increment is lost.
|
|
// A write another writer won is read again and tried again, at most attempts times in all; past that the
|
|
// error is ErrChanged and nothing was counted. Any other error is returned at once, as is a value that is
|
|
// not a whole number (null included): nothing is ever reported counted that was not written. The reverse
|
|
// does not hold: an error that is not ErrChanged — a timeout above all — may come after the write landed, so
|
|
// that number may be spent and never handed out. A caller tolerates gaps, never a number handed out twice.
|
|
// Deleting the key starts the count again at one; a key whose numbers must stay unique is never deleted.
|
|
func (s KeptState) Increment(key string, attempts int) (uint64, error) {
|
|
if attempts < 1 {
|
|
return 0, fmt.Errorf("an increment of %s is tried at least once: %d attempts", key, attempts)
|
|
}
|
|
for i := 0; i < attempts; i++ {
|
|
e, err := s.Get(key)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
var next uint64 = 1
|
|
if e == nil {
|
|
_, err = s.Create(key, next)
|
|
} else {
|
|
var n uint64
|
|
if string(e.Value) == "null" {
|
|
return 0, fmt.Errorf("%s holds null, not a counter", key)
|
|
}
|
|
if err := json.Unmarshal(e.Value, &n); err != nil {
|
|
return 0, fmt.Errorf("%s holds %s, not a counter: %w", key, e.Value, err)
|
|
}
|
|
if n == ^uint64(0) {
|
|
return 0, fmt.Errorf("%s is at the largest number a counter holds", key)
|
|
}
|
|
next = n + 1
|
|
_, err = s.Update(key, next, e.Revision)
|
|
}
|
|
if err == nil {
|
|
return next, nil
|
|
}
|
|
if !errors.Is(err, ErrChanged) {
|
|
return 0, err
|
|
}
|
|
}
|
|
return 0, fmt.Errorf("%s was written by another writer first %d times: %w", key, attempts, ErrChanged)
|
|
}
|
|
|
|
// Keys is every key with a current value, sorted.
|
|
func (s KeptState) Keys() ([]string, error) {
|
|
raw, err := s.ask("keys", map[string]any{})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var keys []string
|
|
return keys, json.Unmarshal(raw, &keys)
|
|
}
|
|
|
|
// Watch hands handle every current value — none that is deleted — 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 watch lasts as long as the process.
|
|
func (s KeptState) Watch(key string, handle func(StateChange) error) error {
|
|
c, err := live()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
id := c.newID()
|
|
c.mu.Lock()
|
|
c.watches[id] = handle
|
|
c.mu.Unlock()
|
|
if _, err := c.askAs(id, "mesh/state.watch", map[string]any{"state": s.name, "key": key}, stateTimeout); err != nil {
|
|
c.mu.Lock()
|
|
delete(c.watches, id)
|
|
c.mu.Unlock()
|
|
return err
|
|
}
|
|
return nil
|
|
}
|