jschoubben
875c680d98
# Conflicts: # go/catalogue/state.go # go/catalogue/state_test.go # go/process/state.go
283 lines
8.5 KiB
Go
283 lines
8.5 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"
|
|
)
|
|
|
|
// 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 := kv.Create(ctx, 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
|
|
}
|
|
|
|
// 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 := kv.Update(ctx, 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 := kv.Put(ctx, 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
|
|
}
|