Fix tests

Signed-off-by: Samuel Attwood <[email protected]>

Author: samuelattwood

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/mattn/go-sqlite3 v1.14.32
 	github.com/nats-io/jsm.go v0.3.0
 	github.com/nats-io/nats-server/v2 v2.12.0
-	github.com/nats-io/nats.go v1.46.1
+	github.com/nats-io/nats.go v1.47.0
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.23.2
 	github.com/shengdoushi/base58 v1.0.0

go.sum

@@ -127,8 +127,8 @@ github.com/nats-io/jwt/v2 v2.8.0 h1:K7uzyz50+yGZDO5o772eRE7atlcSEENpL7P+b74JV1g=
 github.com/nats-io/jwt/v2 v2.8.0/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA=
 github.com/nats-io/nats-server/v2 v2.12.0 h1:OIwe8jZUqJFrh+hhiyKu8snNib66qsx806OslqJuo74=
 github.com/nats-io/nats-server/v2 v2.12.0/go.mod h1:nr8dhzqkP5E/lDwmn+A2CvQPMd1yDKXQI7iGg3lAvww=
-github.com/nats-io/nats.go v1.46.1 h1:bqQ2ZcxVd2lpYI97xYASeRTY3I5boe/IVmuUDPitHfo=
-github.com/nats-io/nats.go v1.46.1/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g=
+github.com/nats-io/nats.go v1.47.0 h1:YQdADw6J/UfGUd2Oy6tn4Hq6YHxCaJrVKayxxFqYrgM=
+github.com/nats-io/nats.go v1.47.0/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g=
 github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0=
 github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=

pkg/drivers/nats/backend_test.go

@@ -3,6 +3,7 @@ package nats
 import (
 	"context"
 	"io"
+	"os"
 	"testing"
 	"time"
 
@@ -52,6 +53,7 @@ func TestBackend_Create(t *testing.T) {
 	ns, nc, b := setupBackend(t)
 	defer ns.Shutdown()
 	defer nc.Drain()
+	defer os.RemoveAll(ns.StoreDir())
 
 	ctx := context.Background()
 
@@ -108,6 +110,7 @@ func TestBackend_Get(t *testing.T) {
 	ns, nc, b := setupBackend(t)
 	defer ns.Shutdown()
 	defer nc.Drain()
+	defer os.RemoveAll(ns.StoreDir())
 
 	ctx := context.Background()
 
@@ -138,7 +141,7 @@ func TestBackend_Get(t *testing.T) {
 
 	// Get at later revision, does not exist.
 	_, _, err = b.Get(ctx, "/a", "", 0, 2, false)
-	expEqualErr(t, nil, err)
+	expEqualErr(t, kserver.ErrFutureRev, err)
 
 	// Create it again and update it.
 	rev, err := b.Create(ctx, "/a", []byte("c"), 0)
@@ -163,6 +166,7 @@ func TestBackend_Update(t *testing.T) {
 	ns, nc, b := setupBackend(t)
 	defer ns.Shutdown()
 	defer nc.Drain()
+	defer os.RemoveAll(ns.StoreDir())
 
 	ctx := context.Background()
 
@@ -199,6 +203,7 @@ func TestBackend_Delete(t *testing.T) {
 	ns, nc, b := setupBackend(t)
 	defer ns.Shutdown()
 	defer nc.Drain()
+	defer os.RemoveAll(ns.StoreDir())
 
 	ctx := context.Background()
 
@@ -237,6 +242,7 @@ func TestBackend_List(t *testing.T) {
 	ns, nc, b := setupBackend(t)
 	defer ns.Shutdown()
 	defer nc.Drain()
+	defer os.RemoveAll(ns.StoreDir())
 
 	ctx := context.Background()
 
@@ -310,19 +316,19 @@ func TestBackend_Watch(t *testing.T) {
 	ns, nc, b := setupBackend(t)
 	defer ns.Shutdown()
 	defer nc.Drain()
+	defer os.RemoveAll(ns.StoreDir())
 
 	ctx := context.Background()
 
 	cctx, cancel := context.WithCancel(ctx)
-	defer cancel()
 
 	rev1, _ := b.Create(ctx, "/a", nil, 0)
 	rev2, _ := b.Create(ctx, "/a/1", nil, 0)
 	rev1, _, _, _ = b.Update(ctx, "/a", nil, rev1, 0)
 	b.Delete(ctx, "/a", rev1)
 	b.Update(ctx, "/a/1", nil, rev2, 0)
 
-	wr := b.Watch(cctx, "/a", 0)
+	wr := b.Watch(cctx, "/", 0)
 	time.Sleep(20 * time.Millisecond)
 	cancel()
 
@@ -332,4 +338,17 @@ func TestBackend_Watch(t *testing.T) {
 	}
 
 	expEqual(t, 5, len(events))
+
+	cctx, cancel = context.WithCancel(ctx)
+	wr = b.Watch(cctx, "/a/", 0)
+	time.Sleep(20 * time.Millisecond)
+	cancel()
+
+	events = make([]*kserver.Event, 0)
+
+	for es := range wr.Events {
+		events = append(events, es...)
+	}
+
+	expEqual(t, 2, len(events))
 }

pkg/drivers/nats/kv.go

@@ -316,29 +316,34 @@ func (e *KeyValue) Watch(ctx context.Context, keys string, startRev int64) (KeyW
 	// Everything but the last token will be treated as a filter
 	// on the watcher. The last token will used as a deliver-time filter.
 
-	watchRange := true
-
 	filter := keys
+
+	// Must watch all keys in this case
+	if !strings.HasPrefix(keys, "/") {
+		filter = "/"
+	}
+
 	if !strings.HasSuffix(filter, "/") {
-		watchRange = false
 		idx := strings.LastIndexByte(filter, '/')
 		if idx > -1 {
 			filter = keys[:idx+1]
 		}
 	}
 
 	if filter != "" {
-		var p string
-		var err error
-		if watchRange {
-			p, err = e.kc.EncodeRange(filter)
-		} else {
-			p, err = e.kc.Encode(filter)
-		}
+		p, err := e.kc.EncodeRange(filter)
 		if err != nil {
 			return nil, err
 		}
 
+		// Special case, reduce watches on root key
+		if keys == compactRevAPI {
+			p, err = e.kc.Encode(compactRevAPI)
+			if err != nil {
+				return nil, err
+			}
+		}
+
 		filter = fmt.Sprintf("$KV.%s.%s", e.nkv.Bucket(), p)
 	}
 
@@ -352,7 +357,6 @@ func (e *KeyValue) Watch(ctx context.Context, keys string, startRev int64) (KeyW
 		if keys != "" {
 			dkey, err := e.kc.Decode(strings.TrimPrefix(key, "."))
 			if err != nil || (keys != "/" && !strings.HasPrefix(dkey, keys)) {
-				logrus.Errorf("watch: invalid key: %s, err=%v", key, err)
 				return
 			}
 		}
@@ -511,7 +515,7 @@ func (e *KeyValue) List(ctx context.Context, prefix, startKey string, limit, rev
 	go func(expired []*keySeq) {
 		for _, ex := range expired {
 			err = e.Delete(ctx, ex.key, jetstream.LastRevision(ex.seq))
-			if err != nil {
+			if err != nil && err != context.Canceled {
 				logrus.Errorf("list: error deleting expired key %s: %v", ex.key, err)
 			}
 		}