Merge remote-tracking branch 'origin/main' into feat/r4e-sdk

# Conflicts: # go/catalogue/state.go # go/catalogue/state_test.go # go/process/state.go
2026-10-22 20:37:49 +02:00
parent d1af87a83a cdfc63cf10
commit 875c680d98
7 changed files with 727 additions and 8 deletions
@@ -5,6 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sort"
+	"strings"
 	"time"

 	"github.com/nats-io/nats.go"
@@ -27,6 +29,8 @@ type State struct {
 	// 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.
@@ -38,7 +42,7 @@ func (s State) bucket() string { return s.Module + "_" + s.Name }
 // 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, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
+	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
 		}
@@ -131,3 +135,148 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) {
 	}
 	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
+}
@@ -6,6 +6,7 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"

 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -94,3 +95,182 @@ func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) {
 		t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error())
 	}
 }
+
+// 1840/S1: given a process connection granted its bucket, when it puts keys whole and lists them, then Put
+// returns a new revision each time and Keys lists them sorted (none for a bucket with none).
+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 {
+		t.Fatalf("the admin could not make the bucket: %v", err)
+	}
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	s := conn.State("sends")
+
+	if keys, err := s.Keys(); err != nil || len(keys) != 0 {
+		t.Fatalf("Keys of an empty bucket: %v, %v", keys, err)
+	}
+	rev, err := s.Put("b", map[string]any{"v": 1})
+	if err != nil || rev == 0 {
+		t.Fatalf("Put on a granted bucket: %d, %v", rev, err)
+	}
+	rev2, err := s.Put("b", map[string]any{"v": 2})
+	if err != nil || rev2 <= rev {
+		t.Fatalf("Put over a key: %d, %v", rev2, err)
+	}
+	if _, err := s.Put("a", 1); err != nil {
+		t.Fatal(err)
+	}
+	keys, err := s.Keys()
+	if err != nil || strings.Join(keys, ",") != "a,b" {
+		t.Fatalf("Keys: %v, %v, want a,b", keys, err)
+	}
+}
+
+// 1840/S2: given a process connection granted its bucket, which already holds two keys and a deleted one,
+// when it watches, then it gets the two current values marked Current, and then a later put and a delete in
+// order; a narrowed watch gets only the keys it names.
+func TestStateWatchOnAGrantedBucket(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	_, adminJetStream := adminJS(t, url)
+	kv, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"})
+	if err != nil {
+		t.Fatalf("the admin could not make the bucket: %v", err)
+	}
+	ctx := ctxFor(t)
+	for k, v := range map[string]string{"a.x": `1`, "a.y": `2`, "b.z": `3`, "gone": `4`} {
+		if _, err := kv.Put(ctx, k, []byte(v)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := kv.Delete(ctx, "gone"); err != nil {
+		t.Fatal(err)
+	}
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	s := conn.State("sends")
+
+	all := make(chan stdio.StateChange, 16)
+	if err := s.Watch("", func(c stdio.StateChange) error { all <- c; return nil }); err != nil {
+		t.Fatalf("Watch on a granted bucket: %v", err)
+	}
+	current := map[string]bool{}
+	for len(all) > 0 {
+		c := <-all
+		if !c.Current || c.Op != "put" {
+			t.Fatalf("a current value arrived as %+v", c)
+		}
+		current[c.Key] = true
+	}
+	if len(current) != 3 || current["gone"] {
+		t.Fatalf("current values %v, want a.x, a.y, b.z and not gone", current)
+	}
+
+	narrow := make(chan stdio.StateChange, 16)
+	if err := s.Watch("a.*", func(c stdio.StateChange) error { narrow <- c; return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if len(narrow) != 2 {
+		t.Fatalf("a narrowed watch handed %d current values, want 2", len(narrow))
+	}
+	// the current values come in the order of their last revision, not of their keys: taken as a set.
+	narrowCurrent := map[string]bool{}
+	for len(narrow) > 0 {
+		c := <-narrow
+		if !c.Current || c.Op != "put" {
+			t.Fatalf("a narrowed watch's current value arrived as %+v", c)
+		}
+		narrowCurrent[c.Key] = true
+	}
+	if !narrowCurrent["a.x"] || !narrowCurrent["a.y"] {
+		t.Fatalf("a narrowed watch's current values %v, want a.x and a.y", narrowCurrent)
+	}
+
+	if _, err := s.Put("a.x", 5); err != nil {
+		t.Fatal(err)
+	}
+	if err := kv.Delete(ctx, "a.y"); err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []struct{ key, op string }{{"a.x", "put"}, {"a.y", "delete"}} {
+		select {
+		case c := <-all:
+			if c.Key != want.key || c.Op != want.op || c.Current {
+				t.Fatalf("got %+v, want %s %s, not current", c, want.op, want.key)
+			}
+		case <-time.After(10 * time.Second):
+			t.Fatalf("no %s of %s", want.op, want.key)
+		}
+	}
+	if _, err := s.Put("b.q", 6); err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []string{"a.x", "a.y"} {
+		select {
+		case c := <-narrow:
+			if c.Key != want || c.Current {
+				t.Fatalf("narrowed watch got %+v, want the change of %s", c, want)
+			}
+		case <-time.After(10 * time.Second):
+			t.Fatalf("narrowed watch missed %s", want)
+		}
+	}
+	select {
+	case c := <-narrow:
+		t.Fatalf("narrowed watch got %+v for a key it did not name", c)
+	case <-time.After(500 * time.Millisecond):
+	}
+}
+
+// 1840/AC8: given an account not granted a bucket, when it watches it, then the refusal names the bucket.
+func TestStateWatchRefusesAnUngrantedBucketNamingIt(t *testing.T) {
+	url, account := grantedSendProcess(t)
+	_, adminJetStream := adminJS(t, url)
+	if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_unrelated"}); err != nil {
+		t.Fatalf("the admin could not make the bucket: %v", err)
+	}
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	err = conn.State("unrelated").Watch("", func(stdio.StateChange) error { return nil })
+	if err == nil || !errors.Is(err, nats.ErrPermissionViolation) || !strings.Contains(err.Error(), "mesh-mail_unrelated") {
+		t.Fatalf("Watch on an ungranted bucket: %v, want a refusal naming mesh-mail_unrelated", err)
+	}
+}
+
+// 1840/S1: given a process connection granted its bucket, when it deletes a key, then the key is gone and
+// deleting it again is no error.
+func TestStateDeleteOnAGrantedBucket(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 {
+		t.Fatalf("the admin could not make the bucket: %v", err)
+	}
+	conn, err := Connect(account)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	s := conn.State("sends")
+	if _, err := s.Put("k", 1); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Delete("k"); err != nil {
+		t.Fatalf("Delete on a granted bucket: %v", err)
+	}
+	if e, err := s.Get("k"); err != nil || e != nil {
+		t.Fatalf("Get after Delete: %+v, %v", e, err)
+	}
+	if err := s.Delete("k"); err != nil {
+		t.Fatalf("Delete of an absent key: %v", err)
+	}
+}
@@ -3,8 +3,11 @@ package process
 import (
 	"context"
 	"errors"
+	"sort"
+	"strings"
 	"sync"
 	"testing"
+	"time"

 	"github.com/nats-io/nats.go/jetstream"

@@ -32,6 +35,9 @@ type fakeKV struct {
 	values map[string][]byte
 	revs   map[string]uint64
 	next   uint64
+	// initial is what a Watch hands first, before its nil marker; watcher is the last one made.
+	initial []jetstream.KeyValueEntry
+	watcher *fakeWatcher
 }

 func newFakeKV() *fakeKV { return &fakeKV{values: map[string][]byte{}, revs: map[string]uint64{}} }
@@ -68,6 +74,58 @@ func (f *fakeKV) Update(_ context.Context, key string, value []byte, revision ui
 	return f.next, nil
 }

+func (f *fakeKV) Put(_ context.Context, key string, value []byte) (uint64, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.next++
+	f.values[key], f.revs[key] = value, f.next
+	return f.next, nil
+}
+
+func (f *fakeKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	delete(f.values, key)
+	delete(f.revs, key)
+	return nil
+}
+
+func (f *fakeKV) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if len(f.values) == 0 {
+		return nil, jetstream.ErrNoKeysFound
+	}
+	keys := make([]string, 0, len(f.values))
+	for k := range f.values {
+		keys = append(keys, k)
+	}
+	sort.Sort(sort.Reverse(sort.StringSlice(keys))) // as a bucket lists them: in no order the caller may rely on
+	return keys, nil
+}
+
+// fakeWatcher is a jetstream.KeyWatcher fed by a test.
+type fakeWatcher struct {
+	jetstream.KeyWatcher
+	updates chan jetstream.KeyValueEntry
+	pattern string
+}
+
+func (w *fakeWatcher) Updates() <-chan jetstream.KeyValueEntry { return w.updates }
+func (w *fakeWatcher) Stop() error                             { return nil }
+
+func (f *fakeKV) Watch(_ context.Context, pattern string, _ ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	w := &fakeWatcher{updates: make(chan jetstream.KeyValueEntry, 16), pattern: pattern}
+	f.watcher = w
+	for _, e := range f.initial {
+		w.updates <- e
+	}
+	w.updates <- nil
+	return w, nil
+}
+
 func stateOn(kv jetstream.KeyValue) State {
 	return State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { return kv, nil }}
 }
@@ -103,3 +161,144 @@ func TestStateSatisfiesAskerStore(t *testing.T) {
 		t.Fatal("Update did not advance the revision")
 	}
 }
+
+// 1840/S1: given a state, when a key is put twice, then each Put returns a new revision and the key reads
+// back the last value, whether or not it had one.
+func TestStatePutWritesAKeyWhole(t *testing.T) {
+	s := stateOn(newFakeKV())
+	rev, err := s.Put("k", map[string]any{"a": 1})
+	if err != nil || rev == 0 {
+		t.Fatalf("Put of a new key: %d, %v", rev, err)
+	}
+	rev2, err := s.Put("k", map[string]any{"a": 2})
+	if err != nil || rev2 <= rev {
+		t.Fatalf("Put over a key: %d, %v, want a revision after %d", rev2, err, rev)
+	}
+	e, err := s.Get("k")
+	if err != nil || e == nil || e.Revision != rev2 || string(e.Value) != `{"a":2}` {
+		t.Fatalf("Get after Put: %+v, %v", e, err)
+	}
+}
+
+// 1840/S1: given a state, when its keys are listed, then they come sorted, and none is an empty list, not an error.
+func TestStateKeysAreSortedAndEmptyWhenNone(t *testing.T) {
+	s := stateOn(newFakeKV())
+	if keys, err := s.Keys(); err != nil || len(keys) != 0 {
+		t.Fatalf("Keys of an empty state: %v, %v", keys, err)
+	}
+	for _, k := range []string{"b", "c", "a"} {
+		if _, err := s.Put(k, 1); err != nil {
+			t.Fatal(err)
+		}
+	}
+	keys, err := s.Keys()
+	if err != nil || strings.Join(keys, ",") != "a,b,c" {
+		t.Fatalf("Keys: %v, %v, want a,b,c", keys, err)
+	}
+}
+
+type watchEntry struct {
+	fakeEntry
+	op jetstream.KeyValueOp
+}
+
+func (e watchEntry) Operation() jetstream.KeyValueOp { return e.op }
+
+// 1840/S2: given a state with current values (one of them deleted), when it is watched, then the handler gets
+// each current value marked Current and not the deleted one, Watch returns, and a later put and delete arrive
+// in order unmarked.
+func TestStateWatchHandsCurrentValuesThenChanges(t *testing.T) {
+	kv := newFakeKV()
+	kv.initial = []jetstream.KeyValueEntry{
+		watchEntry{fakeEntry{key: "a", value: []byte(`1`), rev: 1}, jetstream.KeyValuePut},
+		watchEntry{fakeEntry{key: "gone", rev: 2}, jetstream.KeyValueDelete},
+		watchEntry{fakeEntry{key: "b", value: []byte(`2`), rev: 3}, jetstream.KeyValuePut},
+	}
+	got := make(chan stdio.StateChange, 8)
+	s := stateOn(kv)
+	if err := s.Watch("", func(c stdio.StateChange) error { got <- c; return nil }); err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []stdio.StateChange{
+		{State: "sends", Key: "a", Op: "put", Value: []byte(`1`), Revision: 1, Current: true},
+		{State: "sends", Key: "b", Op: "put", Value: []byte(`2`), Revision: 3, Current: true},
+	} {
+		select {
+		case c := <-got:
+			if c.Key != want.Key || c.Op != want.Op || string(c.Value) != string(want.Value) || c.Revision != want.Revision || c.Current != want.Current || c.State != want.State {
+				t.Fatalf("got %+v, want %+v", c, want)
+			}
+		default:
+			t.Fatalf("Watch returned before handling %+v", want)
+		}
+	}
+	if len(got) != 0 {
+		t.Fatalf("a deleted key was handed: %+v", <-got)
+	}
+	kv.watcher.updates <- watchEntry{fakeEntry{key: "a", value: []byte(`3`), rev: 4}, jetstream.KeyValuePut}
+	kv.watcher.updates <- watchEntry{fakeEntry{key: "b", rev: 5}, jetstream.KeyValueDelete}
+	for _, want := range []stdio.StateChange{
+		{Key: "a", Op: "put", Revision: 4}, {Key: "b", Op: "delete", Revision: 5},
+	} {
+		select {
+		case c := <-got:
+			if c.Key != want.Key || c.Op != want.Op || c.Revision != want.Revision || c.Current {
+				t.Fatalf("got %+v, want %+v", c, want)
+			}
+		case <-time.After(5 * time.Second):
+			t.Fatalf("no change %+v", want)
+		}
+	}
+}
+
+// 1840/S2: a key narrowing reaches the bus as a subject pattern: `*` one name, `**` the rest, empty every key.
+func TestStateWatchNarrowing(t *testing.T) {
+	for key, want := range map[string]string{"": ">", "a.b": "a.b", "a.*": "a.*", "a.**": "a.>", "**": ">"} {
+		if got := watchPattern(key); got != want {
+			t.Errorf("watchPattern(%q) = %q, want %q", key, got, want)
+		}
+	}
+}
+
+// 1840/S2: given a handler that refuses a current value, then Watch returns its error.
+func TestStateWatchReturnsAHandlersErrorOnCurrentValues(t *testing.T) {
+	kv := newFakeKV()
+	kv.initial = []jetstream.KeyValueEntry{watchEntry{fakeEntry{key: "a", value: []byte(`1`), rev: 1}, jetstream.KeyValuePut}}
+	boom := errors.New("boom")
+	if err := stateOn(kv).Watch("", func(stdio.StateChange) error { return boom }); !errors.Is(err, boom) {
+		t.Fatalf("Watch: %v, want the handler's error", err)
+	}
+}
+
+// 1840/S2: given a connection that closes before the current values arrive, then Watch returns an error.
+func TestStateWatchEndsWithAnErrorWhenTheConnectionCloses(t *testing.T) {
+	s := State{Module: "mesh-mail", Name: "sends", closed: func() bool { return true },
+		kv: func(context.Context) (jetstream.KeyValue, error) { return silentKV{}, nil }}
+	if err := s.Watch("", func(stdio.StateChange) error { return nil }); err == nil {
+		t.Fatal("Watch on a closed connection returned no error")
+	}
+}
+
+// silentKV watches but never delivers.
+type silentKV struct{ jetstream.KeyValue }
+
+func (silentKV) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) {
+	return &fakeWatcher{updates: make(chan jetstream.KeyValueEntry)}, nil
+}
+
+// 1840/S1: given a key, when it is deleted, then it reads as absent; deleting a key that was not there is no error.
+func TestStateDeleteRemovesAKeyAndAnAbsentOneIsNoError(t *testing.T) {
+	s := stateOn(newFakeKV())
+	if _, err := s.Put("k", 1); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Delete("k"); err != nil {
+		t.Fatalf("Delete of a key: %v", err)
+	}
+	if e, err := s.Get("k"); err != nil || e != nil {
+		t.Fatalf("Get after Delete: %+v, %v, want absent", e, err)
+	}
+	if err := s.Delete("never"); err != nil {
+		t.Fatalf("Delete of an absent key: %v", err)
+	}
+}