issue 1840/S2: let a process watch a state, current values first then each change, as a tools bundle can

2026-10-22 14:16:59 +02:00
parent 26517ac00b
commit 076e1f7947
3 changed files with 317 additions and 1 deletions
@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"sort"
+	"strings"
 	"time"

 	"github.com/nats-io/nats.go"
@@ -28,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.
@@ -39,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) {
 		js, err := jetstream.New(c.nc)
 		if err != nil {
 			return nil, err
@@ -167,3 +170,96 @@ func (s State) Keys() ([]string, error) {
 	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
+}
@@ -6,6 +6,7 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"

 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
@@ -129,3 +130,107 @@ func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) {
 		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))
+	}
+
+	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 {
+				t.Fatalf("narrowed watch got %+v, want %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)
+	}
+}
@@ -7,6 +7,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"

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

@@ -34,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{}} }
@@ -92,6 +96,28 @@ func (f *fakeKV) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, err
 	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 }}
 }
@@ -162,3 +188,92 @@ func TestStateKeysAreSortedAndEmptyWhenNone(t *testing.T) {
 		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
+}