diff --git a/go/pernode/pernode.go b/go/pernode/pernode.go new file mode 100644 index 0000000..26d4dc0 --- /dev/null +++ b/go/pernode/pernode.go @@ -0,0 +1,345 @@ +// Package pernode reads a state its owner keeps one bucket per node as one state (novox/hq ADR 0593 §6: each node's +// engine writes its own `composed-`, and the provisioner, the assigner and the view read them all), over any +// connection that satisfies Bus: a process's own (process.Conn) and the runtime's (node-tools) alike, so the rules are +// written once. +// +// A reader names the state as its manifest's `reads` does (`.`, such as `mesh-host.composed`) with the +// prefix of its buckets (`composed-`): each bucket `` holds keys whose first part is that node, so a key +// names its bucket. Get reads one key from its node's bucket; Keys and Watch go over every bucket of the family the bus +// holds, and Watch takes in a node's bucket created after it began, as the bus says it (its creation advisory), so a +// node added later is heard with no timer. It reads only: the family has no write, and the owner's nodes write as their +// grants allow. +package pernode + +import ( + "context" + "errors" + "fmt" + "regexp" + "sort" + "strings" + "sync" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// What a reader of a per-node state is granted beside each bucket's read (the grants part composes them): the listing +// of stream names (ListSubject) and the creation of any stream (CreatedAdvisory), which says a name only. A subject's +// wildcard is a whole token, so `KV_composed-*` cannot be granted: every creation is heard and the family's picked. +const ( + ListSubject = "$JS.API.STREAM.NAMES" + CreatedAdvisory = "$JS.EVENT.ADVISORY.STREAM.CREATED.*" + // createdPrefix starts the subject of a key-value bucket's creation advisory, the bucket's name after it. + createdPrefix = "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_" +) + +// Bus is what a connection gives the family: the names of the key-value buckets the bus holds, a bucket opened by name +// (jetstream.ErrBucketNotFound when it is not there), and the creation advisories (CreatedAdvisory) heard, each by its +// subject, until stop. +type Bus interface { + BucketNames(ctx context.Context) ([]string, error) + OpenBucket(ctx context.Context, name string) (jetstream.KeyValue, error) + SubscribeCreated(heard func(subject string)) (stop func(), err error) +} + +// Family is a per-node state: its read name, its buckets' prefix, and the connection it is read over. +type Family struct { + Read string + Prefix string + Bus Bus +} + +// Change is one change a watch hands over: the node whose bucket it is, the entry (its operation says put, delete or +// purge), and whether it was there when that bucket's watch began. +type Change struct { + Node string + Entry jetstream.KeyValueEntry + Current bool +} + +// nodeName is a node a bucket and a subject token accept. +var nodeName = regexp.MustCompile(`^[A-Za-z0-9_-]+$`) + +// Check refuses a family with no read name `.`, or a prefix a bucket name cannot start with or that +// does not end in `-` or `_`, since a prefix without its separator would read a look-alike family (`composed` reading +// `composedx-ace`). +func (f Family) Check() error { + module, name, ok := strings.Cut(f.Read, ".") + if !ok || module == "" || name == "" || strings.Contains(name, ".") { + return fmt.Errorf("%q is not a state another module keeps, .", f.Read) + } + if !nodeName.MatchString(strings.TrimRight(f.Prefix, "-_")) || !strings.HasSuffix(f.Prefix, "-") && !strings.HasSuffix(f.Prefix, "_") { + return fmt.Errorf("the state %s: %q is not a prefix of buckets, letters, digits, '-' and '_' ending in '-' or '_'", f.Read, f.Prefix) + } + if f.Bus == nil { + return fmt.Errorf("the state %s is read over no connection", f.Read) + } + return nil +} + +// Bucket is the bucket of a node. +func (f Family) Bucket(node string) (string, error) { + if err := f.Check(); err != nil { + return "", err + } + if !nodeName.MatchString(node) { + return "", fmt.Errorf("the state %s: %q is not a node a bucket is named for", f.Read, node) + } + return f.Prefix + node, nil +} + +// NodeOf is the node a key of the family names: its first part. +func NodeOf(key string) (string, error) { + node, _, ok := strings.Cut(key, ".") + if !ok || !nodeName.MatchString(node) { + return "", fmt.Errorf("%q names no node: a key of a per-node state starts .", key) + } + return node, nil +} + +// nodeOfBucket is the node of a bucket name of the family, and whether it is one. +func (f Family) nodeOfBucket(name string) (string, bool) { + node, ok := strings.CutPrefix(name, f.Prefix) + return node, ok && nodeName.MatchString(node) +} + +// Nodes are the nodes whose bucket of the family the bus holds now, sorted. +func (f Family) Nodes(ctx context.Context) ([]string, error) { + if err := f.Check(); err != nil { + return nil, err + } + names, err := f.Bus.BucketNames(ctx) + if err != nil { + return nil, fmt.Errorf("the state %s: listing the buckets %s*: %w", f.Read, f.Prefix, err) + } + var nodes []string + for _, name := range names { + if node, ok := f.nodeOfBucket(name); ok { + nodes = append(nodes, node) + } + } + sort.Strings(nodes) + return nodes, nil +} + +// Open is the bucket of a node, opened; nil, nil when the bus holds none. +func (f Family) Open(ctx context.Context, node string) (jetstream.KeyValue, error) { + bucket, err := f.Bucket(node) + if err != nil { + return nil, err + } + kv, err := f.Bus.OpenBucket(ctx, bucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("the state %s (bucket %s): %w", f.Read, bucket, err) + } + return kv, nil +} + +// Get is a key's current entry, read from the bucket of the node the key names, or nil when it has none or that node +// has no bucket. +func (f Family) Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) { + node, err := NodeOf(key) + if err != nil { + return nil, err + } + kv, err := f.Open(ctx, node) + if err != nil || kv == nil { + return nil, err + } + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("the state %s, key %s: %w", f.Read, key, err) + } + return e, nil +} + +// Keys is every key with a current value in every bucket of the family, sorted. +func (f Family) Keys(ctx context.Context) ([]string, error) { + nodes, err := f.Nodes(ctx) + if err != nil { + return nil, err + } + keys := []string{} + for _, node := range nodes { + kv, err := f.Open(ctx, node) + if err != nil { + return nil, err + } + if kv == nil { + continue + } + k, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + continue + } + if err != nil { + return nil, fmt.Errorf("the state %s on %s, keys: %w", f.Read, node, err) + } + keys = append(keys, k...) + } + sort.Strings(keys) + return keys, nil +} + +// Pattern is a key narrowing as the bus names it: `*` stays one name, `**` is the rest, empty is every key. +func Pattern(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, one call at a time, the current values of every bucket of the family the bus holds (each +// Current, deletes left out), then every change, and the current values and changes of a bucket the bus creates later. +// It returns once the current values of the buckets there now are handled; an error before that is returned, and an +// error from handle after it ends that bucket's watch. key narrows it within each bucket (Pattern); a key whose first +// part names a node watches only that node's bucket. ctx ends every watch and the hearing of new buckets. +func (f Family) Watch(ctx context.Context, key string, handle func(Change) error) error { + if err := f.Check(); err != nil { + return err + } + var mu sync.Mutex + one := func(c Change) error { + mu.Lock() + defer mu.Unlock() + return handle(c) + } + var wmu sync.Mutex + watched := map[string]bool{} + watch := func(node string) error { + wmu.Lock() + if watched[node] { + wmu.Unlock() + return nil + } + watched[node] = true + wmu.Unlock() + kv, err := f.Open(ctx, node) + if err != nil || kv == nil { + return err + } + return f.watchBucket(ctx, kv, node, key, one) + } + if first, _, _ := strings.Cut(key, "."); key != "" && first != "*" && first != "**" { + return watch(first) + } + // Heard before the buckets are listed, so a bucket created between the two is not missed. + stop, err := f.Bus.SubscribeCreated(func(subject string) { + bucket, isKV := strings.CutPrefix(subject, createdPrefix) + if node, ok := f.nodeOfBucket(bucket); isKV && ok { + go func() { _ = watch(node) }() + } + }) + if err != nil { + return fmt.Errorf("the state %s: hearing new buckets %s*: %w", f.Read, f.Prefix, err) + } + go func() { <-ctx.Done(); stop() }() + nodes, err := f.Nodes(ctx) + if err != nil { + stop() + return err + } + for _, node := range nodes { + if err := watch(node); err != nil { + stop() + return err + } + } + return nil +} + +// watchBucket hands one bucket's current values, then its changes from a goroutine, until ctx ends or handle errs. +func (f Family) watchBucket(ctx context.Context, kv jetstream.KeyValue, node, key string, handle func(Change) error) error { + w, err := kv.Watch(ctx, Pattern(key)) + if err != nil { + return fmt.Errorf("the state %s on %s, watching %q: %w", f.Read, node, key, err) + } + for { + select { + case <-ctx.Done(): + _ = w.Stop() + return ctx.Err() + case e, open := <-w.Updates(): + if !open { + return fmt.Errorf("the state %s on %s: the watch ended", f.Read, node) + } + if e == nil { + go func() { + defer func() { _ = w.Stop() }() + for { + select { + case <-ctx.Done(): + return + case e, open := <-w.Updates(): + if !open || e == nil { + return + } + if handle(Change{Node: node, Entry: e}) != nil { + return + } + } + } + }() + return nil + } + if e.Operation() != jetstream.KeyValuePut { + continue + } + if err := handle(Change{Node: node, Entry: e, Current: true}); err != nil { + _ = w.Stop() + return err + } + } + } +} + +// OverNATS is the Bus of a plain connection: its JetStream's bucket names and buckets, and the creation advisories +// subscribed on it. A connection that words a refused open its own way (process.Conn) gives its own OpenBucket. +func OverNATS(nc *nats.Conn) Bus { return natsBus{nc: nc} } + +type natsBus struct{ nc *nats.Conn } + +func (b natsBus) BucketNames(ctx context.Context) ([]string, error) { + js, err := jetstream.New(b.nc) + if err != nil { + return nil, err + } + lister := js.KeyValueStoreNames(ctx) + var names []string + for name := range lister.Name() { + names = append(names, name) + } + if err := lister.Error(); err != nil && !errors.Is(err, jetstream.ErrEndOfData) { + return nil, err + } + return names, nil +} + +func (b natsBus) OpenBucket(ctx context.Context, name string) (jetstream.KeyValue, error) { + js, err := jetstream.New(b.nc) + if err != nil { + return nil, err + } + return js.KeyValue(ctx, name) +} + +func (b natsBus) SubscribeCreated(heard func(subject string)) (func(), error) { + sub, err := b.nc.Subscribe(CreatedAdvisory, func(m *nats.Msg) { heard(m.Subject) }) + if err != nil { + return nil, err + } + return func() { _ = sub.Unsubscribe() }, nil +} diff --git a/go/pernode/pernode_test.go b/go/pernode/pernode_test.go new file mode 100644 index 0000000..bfb7403 --- /dev/null +++ b/go/pernode/pernode_test.go @@ -0,0 +1,158 @@ +package pernode + +import ( + "context" + "slices" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +type fakeEntry struct { + jetstream.KeyValueEntry + key, value string +} + +func (e fakeEntry) Key() string { return e.key } +func (e fakeEntry) Value() []byte { return []byte(e.value) } +func (e fakeEntry) Revision() uint64 { return 1 } +func (e fakeEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut } + +type fakeWatcher struct{ ch chan jetstream.KeyValueEntry } + +func (w fakeWatcher) Updates() <-chan jetstream.KeyValueEntry { return w.ch } +func (w fakeWatcher) Stop() error { return nil } + +type fakeKV struct { + jetstream.KeyValue + values map[string]string + watcher fakeWatcher +} + +func (k *fakeKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { + v, ok := k.values[key] + if !ok { + return nil, jetstream.ErrKeyNotFound + } + return fakeEntry{key: key, value: v}, nil +} + +func (k *fakeKV) Keys(context.Context, ...jetstream.WatchOpt) ([]string, error) { + var keys []string + for key := range k.values { + keys = append(keys, key) + } + return keys, nil +} + +func (k *fakeKV) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + k.watcher = fakeWatcher{ch: make(chan jetstream.KeyValueEntry, 16)} + for key, v := range k.values { + k.watcher.ch <- fakeEntry{key: key, value: v} + } + k.watcher.ch <- nil + return k.watcher, nil +} + +type fakeBus struct { + mu sync.Mutex + buckets map[string]*fakeKV + heard func(string) + opened []string +} + +func (b *fakeBus) BucketNames(context.Context) ([]string, error) { + b.mu.Lock() + defer b.mu.Unlock() + var names []string + for n := range b.buckets { + names = append(names, n) + } + return names, nil +} + +func (b *fakeBus) OpenBucket(_ context.Context, name string) (jetstream.KeyValue, error) { + b.mu.Lock() + defer b.mu.Unlock() + b.opened = append(b.opened, name) + kv, ok := b.buckets[name] + if !ok { + return nil, jetstream.ErrBucketNotFound + } + return kv, nil +} + +func (b *fakeBus) SubscribeCreated(heard func(string)) (func(), error) { + b.heard = heard + return func() {}, nil +} + +// A per-node state over a fake connection (ADR 0593 §6): a key is read from its node's bucket, the keys of every +// node's bucket are listed together, a look-alike bucket is never opened, a bucket created during a watch is heard, +// and a family whose prefix would read a look-alike is refused. +func TestAFamilyReadsEveryNodesBucketOverAnyConnection(t *testing.T) { + bus := &fakeBus{buckets: map[string]*fakeKV{ + "composed-ace": {values: map[string]string{"ace.netcheck.aa": "1"}}, + "composed-g14": {values: map[string]string{"g14.netcheck.bb": "2"}}, + "composedx-ace": {values: map[string]string{"ace.other.cc": "3"}}, + }} + f := Family{Read: "mesh-host.composed", Prefix: "composed-", Bus: bus} + ctx := context.Background() + if nodes, err := f.Nodes(ctx); err != nil || !slices.Equal(nodes, []string{"ace", "g14"}) { + t.Fatalf("nodes = %v, %v", nodes, err) + } + if e, err := f.Get(ctx, "g14.netcheck.bb"); err != nil || e == nil || string(e.Value()) != "2" { + t.Fatalf("get: %v, %v", e, err) + } + if e, err := f.Get(ctx, "laptop.netcheck.bb"); err != nil || e != nil { + t.Fatalf("a node with no bucket: %v, %v", e, err) + } + if keys, err := f.Keys(ctx); err != nil || !slices.Equal(keys, []string{"ace.netcheck.aa", "g14.netcheck.bb"}) { + t.Fatalf("keys = %v, %v", keys, err) + } + var mu sync.Mutex + var heard []string + wctx, cancel := context.WithCancel(ctx) + defer cancel() + if err := f.Watch(wctx, "", func(c Change) error { + mu.Lock() + defer mu.Unlock() + heard = append(heard, c.Node+":"+c.Entry.Key()) + return nil + }); err != nil { + t.Fatal(err) + } + bus.mu.Lock() + bus.buckets["composed-shanks"] = &fakeKV{values: map[string]string{"shanks.netcheck.dd": "4"}} + bus.mu.Unlock() + bus.heard("$JS.EVENT.ADVISORY.STREAM.CREATED.KV_composed-shanks") + bus.heard("$JS.EVENT.ADVISORY.STREAM.CREATED.KV_composedx-g14") + want := []string{"ace:ace.netcheck.aa", "g14:g14.netcheck.bb", "shanks:shanks.netcheck.dd"} + for deadline := time.Now().Add(2 * time.Second); ; { + mu.Lock() + got := slices.Sorted(slices.Values(heard)) + mu.Unlock() + if slices.Equal(got, want) { + break + } + if time.Now().After(deadline) { + t.Fatalf("heard %v, want %v", got, want) + } + time.Sleep(10 * time.Millisecond) + } + bus.mu.Lock() + for _, o := range bus.opened { + if o == "composedx-ace" || o == "composedx-g14" { + t.Fatalf("a look-alike bucket was opened: %v", bus.opened) + } + } + bus.mu.Unlock() + for _, bad := range []Family{{Read: "mesh-host.composed", Prefix: "composed", Bus: bus}, {Read: "composed", Prefix: "composed-", Bus: bus}, + {Read: "mesh-host.composed", Prefix: "composed-"}} { + if _, err := bad.Nodes(ctx); err == nil { + t.Errorf("%+v was read", bad) + } + } +} diff --git a/go/process/pernode.go b/go/process/pernode.go index 9494dd6..5d844a8 100644 --- a/go/process/pernode.go +++ b/go/process/pernode.go @@ -2,34 +2,26 @@ package process import ( "context" - "errors" - "fmt" - "sort" + "encoding/json" "strings" - "sync" - "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/pernode" ) -// **A state its owner keeps one bucket per node is read as one state** (novox/hq ADR 0593 §6: each node's engine -// writes its own `composed-`, and the provisioner, the assigner and the view read them all). A reader names it as -// its manifest's `reads` does (`.`, such as `mesh-host.composed`) with the prefix of its buckets -// (`composed-`): each bucket `` holds keys whose first part is that node, so a key names its bucket. -// Get reads one key from its node's bucket; Keys and Watch go over every bucket of the family the bus holds, and Watch -// takes in a node's bucket created after it began, as the bus says it (its creation advisory), so a node added later is -// heard with no timer. Reads only: the owner's nodes write, as their grants allow. +// A state its owner keeps one bucket per node (novox/hq ADR 0593 §6: `composed-`), read as one state on this +// connection: the rules are pernode's, this connection only adapts to its Bus, opening a bucket as State does (a bucket +// not granted refused promptly, naming it). -// What a reader of a per-node state is granted beside each bucket's read (the grants part composes them): the listing -// of bucket names (FamilyListSubject) and the creation of any stream (FamilyCreatedAdvisory), which says a name only. +// What a reader of a per-node state is granted beside each bucket's read (pernode.ListSubject, CreatedAdvisory). const ( - FamilyListSubject = "$JS.API.STREAM.NAMES" - FamilyCreatedAdvisory = "$JS.EVENT.ADVISORY.STREAM.CREATED.*" + FamilyListSubject = pernode.ListSubject + FamilyCreatedAdvisory = pernode.CreatedAdvisory ) -// NodeFamily is a per-node state read across every node's bucket. +// NodeFamily is a per-node state read across every node's bucket on this connection. type NodeFamily struct { Read string Prefix string @@ -42,149 +34,82 @@ func (c *Conn) ReadPerNode(read, prefix string) NodeFamily { return NodeFamily{Read: read, Prefix: prefix, c: c} } -// checked refuses a family with no read name or prefix. -func (f NodeFamily) checked() error { - module, name, ok := strings.Cut(f.Read, ".") - if !ok || module == "" || name == "" || strings.Contains(name, ".") { - return fmt.Errorf("%q is not a state another module keeps, .", f.Read) - } - if f.Prefix == "" || strings.ContainsAny(f.Prefix, ".*> ") { - return fmt.Errorf("the state %s names no prefix of its buckets, or one a bucket name cannot start with: %q", f.Read, f.Prefix) - } - return nil +// connBus is pernode.Bus over this connection. +type connBus struct { + pernode.Bus + f NodeFamily } -// Node is the state of one node of the family: its bucket ``, bound as Read binds a bucket. -func (f NodeFamily) Node(node string) State { - module, name, _ := strings.Cut(f.Read, ".") - return f.c.bound(module, name, f.Prefix+node, func() error { - if err := f.checked(); err != nil { - return err - } - if node == "" || strings.ContainsAny(node, ".*> ") { - return fmt.Errorf("the state %s: %q is not a node a bucket is named for", f.Read, node) - } - return nil - }) +func (b connBus) OpenBucket(ctx context.Context, name string) (jetstream.KeyValue, error) { + module, state, _ := cutRead(b.f.Read) + return b.f.c.bound(module, state, name, func() error { return nil }).kv(ctx) } -// nodeOf is the node a key of the family names: its first part. -func nodeOf(key string) (string, error) { - node, _, ok := strings.Cut(key, ".") - if !ok || node == "" { - return "", fmt.Errorf("%q names no node: a key of a per-node state starts .", key) - } - return node, nil +func cutRead(read string) (string, string, bool) { return strings.Cut(read, ".") } + +func (f NodeFamily) family() pernode.Family { + return pernode.Family{Read: f.Read, Prefix: f.Prefix, Bus: connBus{Bus: pernode.OverNATS(f.c.nc), f: f}} } -// Nodes are the nodes whose bucket of the family the bus holds now, sorted: the names of the key-value buckets that -// start with the prefix, the prefix taken off. +// Nodes are the nodes whose bucket of the family the bus holds now, sorted. func (f NodeFamily) Nodes() ([]string, error) { - if err := f.checked(); err != nil { - return nil, err - } ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) defer cancel() - js, err := jetstream.New(f.c.nc) - if err != nil { - return nil, err - } - names := js.KeyValueStoreNames(ctx) - var nodes []string - for name := range names.Name() { - if node, ok := strings.CutPrefix(name, f.Prefix); ok && node != "" { - nodes = append(nodes, node) - } - } - if err := names.Error(); err != nil && !errors.Is(err, jetstream.ErrEndOfData) { - return nil, fmt.Errorf("the state %s: listing the buckets %s*: %w", f.Read, f.Prefix, err) - } - sort.Strings(nodes) - return nodes, nil + return f.family().Nodes(ctx) } -// Get is a key's current value, read from the bucket of the node the key names, or nil when it has none or that -// node has no bucket. +// Node is the state of one node of the family, its bucket ``, bound as Read binds a bucket. +func (f NodeFamily) Node(node string) State { + module, name, _ := cutRead(f.Read) + bucket, err := f.family().Bucket(node) + return f.c.bound(module, name, bucket, func() error { return err }) +} + +// Get is a key's current value, read from the bucket of the node the key names, or nil when it has none or that node +// has no bucket. func (f NodeFamily) Get(key string) (*stdio.StateEntry, error) { - node, err := nodeOf(key) - if err != nil { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + e, err := f.family().Get(ctx, key) + if err != nil || e == nil { return nil, err } - e, err := f.Node(node).Get(key) - if errors.Is(err, jetstream.ErrBucketNotFound) { - return nil, nil - } - return e, err + return &stdio.StateEntry{Key: e.Key(), Value: json.RawMessage(e.Value()), Revision: e.Revision()}, nil } // Keys is every key with a current value in every bucket of the family, sorted. func (f NodeFamily) Keys() ([]string, error) { - nodes, err := f.Nodes() - if err != nil { - return nil, err - } - keys := []string{} - for _, node := range nodes { - k, err := f.Node(node).Keys() - if err != nil { - return nil, err - } - keys = append(keys, k...) - } - sort.Strings(keys) - return keys, nil + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + return f.family().Keys(ctx) } -// Watch is State.Watch over every bucket of the family: handle hears the current values of every bucket the bus holds, -// then every change, and the current values and changes of a bucket the bus creates later. Calls to handle are one at -// a time. key narrows it as State.Watch's does, within each bucket; a key whose first part names a node watches only -// that node's bucket, but is still told of no bucket that appears later for another node. +// Watch is pernode.Family.Watch on this connection, each change handed as a State change: every bucket's current values, +// then its changes, a bucket created later included, one call at a time; the connection closing ends it. func (f NodeFamily) Watch(key string, handle func(stdio.StateChange) error) error { - if err := f.checked(); err != nil { - return err - } - var mu sync.Mutex - one := func(c stdio.StateChange) error { - mu.Lock() - defer mu.Unlock() - return handle(c) - } - watched := map[string]bool{} - var wmu sync.Mutex - watch := func(node string) error { - wmu.Lock() - if watched[node] { - wmu.Unlock() - return nil + ctx, cancel := context.WithCancel(context.Background()) + go func() { + for !f.c.nc.IsClosed() { + select { + case <-ctx.Done(): + return + case <-f.c.nc.StatusChanged(): + } } - watched[node] = true - wmu.Unlock() - return f.Node(node).Watch(key, one) - } - if first, _, _ := strings.Cut(key, "."); key != "" && first != "*" && first != "**" { - return watch(first) - } - // Heard before the buckets are listed, so a bucket created between the two is not missed. A subject's wildcard is a - // whole token, so every stream's creation is heard (FamilyCreatedAdvisory) and the family's picked by name. - sub, err := f.c.nc.Subscribe(FamilyCreatedAdvisory, func(m *nats.Msg) { - bucket, isKV := strings.CutPrefix(m.Subject, strings.TrimSuffix(FamilyCreatedAdvisory, "*")+"KV_") - if node, ok := strings.CutPrefix(bucket, f.Prefix); isKV && ok && node != "" { - go func() { _ = watch(node) }() + cancel() + }() + _, name, _ := cutRead(f.Read) + err := f.family().Watch(ctx, key, func(c pernode.Change) error { + sc := stdio.StateChange{State: name, Key: c.Entry.Key(), Op: stdio.OpPut, Revision: c.Entry.Revision(), Current: c.Current} + if c.Entry.Operation() == jetstream.KeyValuePut { + sc.Value = json.RawMessage(c.Entry.Value()) + } else { + sc.Op = stdio.OpDelete } + return handle(sc) }) if err != nil { - return fmt.Errorf("the state %s: hearing new buckets %s*: %w", f.Read, f.Prefix, err) + cancel() } - nodes, err := f.Nodes() - if err != nil { - _ = sub.Unsubscribe() - return err - } - for _, node := range nodes { - if err := watch(node); err != nil { - _ = sub.Unsubscribe() - return err - } - } - return nil + return err }