From 076e1f7947db3277e572cee2425011121b8b2ca8 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 14:16:59 +0200 Subject: [PATCH] issue 1840/S2: let a process watch a state, current values first then each change, as a tools bundle can --- go/process/state.go | 98 ++++++++++++++++++++++- go/process/state_integration_test.go | 105 ++++++++++++++++++++++++ go/process/state_test.go | 115 +++++++++++++++++++++++++++ 3 files changed, 317 insertions(+), 1 deletion(-) diff --git a/go/process/state.go b/go/process/state.go index e6c0655..1e79302 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -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 +} diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index 01085b3..f0b1cfd 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -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) + } +} diff --git a/go/process/state_test.go b/go/process/state_test.go index ba02ad9..847abd2 100644 --- a/go/process/state_test.go +++ b/go/process/state_test.go @@ -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 +}