mesh-sdk/go/pernode/pernode_test.go

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