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) } } }