Repository: nats-io/nats.go
Pull request #1380 (2023-08-26 14:59:23)

[IMPROVED] Fixed flaky tests
This PR fixes 2 flaky tests:
- `TestPullSubscribeFetchBatch` somethimes failed on "cancel context during fetch" step because of too short wait before cancellation (messages were not fetched in time)
- `TestListKeyValueStores` failed when 1025 kv stores were not created before context timed out


Patch:
diff --git a/jetstream/README.md b/jetstream/README.md
index cd791b0..bd1be57 100644
--- a/jetstream/README.md
+++ b/jetstream/README.md
@@ -178,7 +178,7 @@ for s := range streams.Info() {
     fmt.Println(s.Config.Name)
 }
 if streams.Err() != nil {
-    fmt.Println("Unexpected error ocurred")
+    fmt.Println("Unexpected error occurred")
 }
 
 // list stream names
@@ -187,7 +187,7 @@ for name := range names.Name() {
     fmt.Println(name)
 }
 if names.Err() != nil {
-    fmt.Println("Unexpected error ocurred")
+    fmt.Println("Unexpected error occurred")
 }
 ```
 
@@ -314,7 +314,7 @@ for cons := range consumers.Info() {
     fmt.Println(cons.Name)
 }
 if consumers.Err() != nil {
-    fmt.Println("Unexpected error ocurred")
+    fmt.Println("Unexpected error occurred")
 }
 
 // list consumer names
@@ -323,7 +323,7 @@ for name := range names.Name() {
     fmt.Println(name)
 }
 if names.Err() != nil {
-    fmt.Println("Unexpected error ocurred")
+    fmt.Println("Unexpected error occurred")
 }
 ```
 
@@ -579,7 +579,7 @@ case err := <-ackF.Err():
     fmt.Println(err)
 }
 
-// similarly to syncronous publish, there is a helper method accepting subject and data
+// similarly to synchronous publish, there is a helper method accepting subject and data
 ackF, err = js.PublishAsync("ORDERS.new", []byte("hello"))
 ```
 
diff --git a/jetstream/consumer.go b/jetstream/consumer.go
index ecd75d3..05e0d6c 100644
--- a/jetstream/consumer.go
+++ b/jetstream/consumer.go
@@ -28,7 +28,7 @@ type (
 	// Consumer contains methods for fetching/processing messages from a stream, as well as fetching consumer info
 	Consumer interface {
 		// Fetch is used to retrieve up to a provided number of messages from a stream.
-		// This method will always send a single request and wait until either all messages are retreived
+		// This method will always send a single request and wait until either all messages are retrieved
 		// or request times out.
 		Fetch(batch int, opts ...FetchOpt) (MessageBatch, error)
 		// FetchBytes is used to retrieve up to a provided bytes from the stream.
diff --git a/jetstream/kv.go b/jetstream/kv.go
index 9d7eae7..3546f01 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -856,12 +856,18 @@ func (kv *kvs) Watch(ctx context.Context, keys string, opts ...WatchOpt) (KeyWat
 	sub.SetClosedHandler(func(_ string) {
 		close(w.updates)
 	})
-	initialPending, err := sub.InitialConsumerPending()
 	// If there were no pending messages at the time of the creation
 	// of the consumer, send the marker.
-	if err == nil && initialPending == 0 {
+	// Skip if UpdatesOnly() is set, since there will never be updates initially.
+	if !o.updatesOnly {
+		initialPending, err := sub.InitialConsumerPending()
+		if err == nil && initialPending == 0 {
+			w.initDone = true
+			w.updates <- nil
+		}
+	} else {
+		// if UpdatesOnly was used, mark initialization as complete
 		w.initDone = true
-		w.updates <- nil
 	}
 	w.sub = sub
 	return w, nil
diff --git a/jetstream/ordered.go b/jetstream/ordered.go
index 0f5b704..231b413 100644
--- a/jetstream/ordered.go
+++ b/jetstream/ordered.go
@@ -230,7 +230,7 @@ func (s *orderedSubscription) Stop() {
 }
 
 // Fetch is used to retrieve up to a provided number of messages from a stream.
-// This method will always send a single request and wait until either all messages are retreived
+// This method will always send a single request and wait until either all messages are retrieved
 // or context reaches its deadline.
 func (c *orderedConsumer) Fetch(batch int, opts ...FetchOpt) (MessageBatch, error) {
 	if c.consumerType == consumerTypeConsume {
diff --git a/js.go b/js.go
index 8836bbb..05637d5 100644
--- a/js.go
+++ b/js.go
@@ -396,7 +396,7 @@ func DirectGetNext(subject string) JSOpt {
 }
 
 // StreamListFilter is an option that can be used to configure `StreamsInfo()` and `StreamNames()` requests.
-// It allows filtering the retured streams by subject associated with each stream.
+// It allows filtering the returned streams by subject associated with each stream.
 // Wildcards can be used. For example, `StreamListFilter(FOO.*.A) will return
 // all streams which have at least one subject matching the provided pattern (e.g. FOO.TEST.A).
 func StreamListFilter(subject string) JSOpt {
@@ -3362,7 +3362,7 @@ func (p AckPolicy) MarshalJSON() ([]byte, error) {
 	case AckExplicitPolicy:
 		return json.Marshal("explicit")
 	default:
-		return nil, fmt.Errorf("nats: unknown acknowlegement policy %v", p)
+		return nil, fmt.Errorf("nats: unknown acknowledgement policy %v", p)
 	}
 }
 
diff --git a/kv.go b/kv.go
index 7d2ab15..17c9e2f 100644
--- a/kv.go
+++ b/kv.go
@@ -941,6 +941,7 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 			w.updates <- entry
 		}
 		// Check if done and initial values.
+		// Skip if UpdatesOnly() is set, since there will never be updates initially.
 		if !w.initDone {
 			w.received++
 			// We set this on the first trip through..
@@ -980,9 +981,15 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 	sub.mu.Lock()
 	// If there were no pending messages at the time of the creation
 	// of the consumer, send the marker.
-	if sub.jsi != nil && sub.jsi.pending == 0 {
+	// Skip if UpdatesOnly() is set, since there will never be updates initially.
+	if !o.updatesOnly {
+		if sub.jsi != nil && sub.jsi.pending == 0 {
+			w.initDone = true
+			w.updates <- nil
+		}
+	} else {
+		// if UpdatesOnly was used, mark initialization as complete
 		w.initDone = true
-		w.updates <- nil
 	}
 	// Set us up to close when the waitForMessages func returns.
 	sub.pDone = func(_ string) {
diff --git a/object.go b/object.go
index b12af5c..b025072 100644
--- a/object.go
+++ b/object.go
@@ -150,13 +150,13 @@ var (
 
 // ObjectStoreConfig is the config for the object store.
 type ObjectStoreConfig struct {
-	Bucket      string
-	Description string
-	TTL         time.Duration
-	MaxBytes    int64
-	Storage     StorageType
-	Replicas    int
-	Placement   *Placement
+	Bucket      string        `json:"bucket"`
+	Description string        `json:"description,omitempty"`
+	TTL         time.Duration `json:"max_age,omitempty"`
+	MaxBytes    int64         `json:"max_bytes,omitempty"`
+	Storage     StorageType   `json:"storage,omitempty"`
+	Replicas    int           `json:"num_replicas,omitempty"`
+	Placement   *Placement    `json:"placement,omitempty"`
 }
 
 type ObjectStoreStatus interface {
@@ -1056,6 +1056,8 @@ func (obs *obs) Watch(opts ...WatchOpt) (ObjectWatcher, error) {
 			w.updates <- &info
 		}
 
+		// if UpdatesOnly is set, no not send nil to the channel
+		// as it would always be triggered after initializing the watcher
 		if !initDoneMarker && meta.NumPending == 0 {
 			initDoneMarker = true
 			w.updates <- nil
@@ -1064,9 +1066,17 @@ func (obs *obs) Watch(opts ...WatchOpt) (ObjectWatcher, error) {
 
 	allMeta := fmt.Sprintf(objAllMetaPreTmpl, obs.name)
 	_, err := obs.js.GetLastMsg(obs.stream, allMeta)
-	if err == ErrMsgNotFound {
+	// if there are no messages on the stream and we are not watching
+	// updates only, send nil to the channel to indicate that the initial
+	// watch is done
+	if !o.updatesOnly {
+		if errors.Is(err, ErrMsgNotFound) {
+			initDoneMarker = true
+			w.updates <- nil
+		}
+	} else {
+		// if UpdatesOnly was used, mark initialization as complete
 		initDoneMarker = true
-		w.updates <- nil
 	}
 
 	// Used ordered consumer to deliver results.
@@ -1074,6 +1084,9 @@ func (obs *obs) Watch(opts ...WatchOpt) (ObjectWatcher, error) {
 	if !o.includeHistory {
 		subOpts = append(subOpts, DeliverLastPerSubject())
 	}
+	if o.updatesOnly {
+		subOpts = append(subOpts, DeliverNew())
+	}
 	sub, err := obs.js.Subscribe(allMeta, update, subOpts...)
 	if err != nil {
 		return nil, err


Test patch:
diff --git a/jetstream/test/kv_test.go b/jetstream/test/kv_test.go
index 764182d..788a318 100644
--- a/jetstream/test/kv_test.go
+++ b/jetstream/test/kv_test.go
@@ -365,11 +365,8 @@ func TestKeyValueWatch(t *testing.T) {
 		watcher, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
 		expectOk(t, err)
 		defer watcher.Stop()
-		expectInitDone := expectInitDoneF(t, watcher)
 		expectUpdate := expectUpdateF(t, watcher)
 		expectDelete := expectDeleteF(t, watcher)
-		// make sure we do not get any values here
-		expectInitDone()
 
 		// now update some keys and expect updates
 		_, err = kv.Put(ctx, "name", []byte("pp"))
@@ -397,11 +394,8 @@ func TestKeyValueWatch(t *testing.T) {
 		watcher, err = kv.Watch(ctx, "t.*", jetstream.UpdatesOnly())
 		expectOk(t, err)
 		defer watcher.Stop()
-		expectInitDone = expectInitDoneF(t, watcher)
 		expectUpdate = expectUpdateF(t, watcher)
 
-		expectInitDone()
-
 		// update some keys and expect updates
 		_, err = kv.Put(ctx, "t.name", []byte("pp"))
 		expectOk(t, err)
@@ -985,7 +979,7 @@ func TestListKeyValueStores(t *testing.T) {
 
 			nc, js := jsClient(t, s)
 			defer nc.Close()
-			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 			defer cancel()
 
 			// create stream without the chunk subject, but with KV_ prefix
diff --git a/jetstream/test/message_test.go b/jetstream/test/message_test.go
index c150ba2..b463927 100644
--- a/jetstream/test/message_test.go
+++ b/jetstream/test/message_test.go
@@ -69,7 +69,7 @@ func TestMessageDetails(t *testing.T) {
 		t.Fatalf("unexpected error during fetch: %v", err)
 	}
 	if string(msg.Data()) != "msg" {
-		t.Fatalf("Invalid messge body; want: 'msg'; got: %q", string(msg.Data()))
+		t.Fatalf("Invalid message body; want: 'msg'; got: %q", string(msg.Data()))
 	}
 	metadata, err := msg.Metadata()
 	if err != nil {
diff --git a/jetstream/test/publish_test.go b/jetstream/test/publish_test.go
index c8712b5..80ddc71 100644
--- a/jetstream/test/publish_test.go
+++ b/jetstream/test/publish_test.go
@@ -1149,7 +1149,7 @@ func TestPublishMsgAsync(t *testing.T) {
 						t.Fatalf("Invalid headers on message; want: %v; got: %v", pub.expectedHeaders, pub.msg.Header)
 					}
 					if string(msg.Data) != string(pub.msg.Data) {
-						t.Fatalf("Invaid message in ack; want: %q; got: %q", string(pub.msg.Data), string(msg.Data))
+						t.Fatalf("Invalid message in ack; want: %q; got: %q", string(pub.msg.Data), string(msg.Data))
 					}
 				case err := <-ackFuture.Err():
 					if pub.withAckError == nil {
diff --git a/nats_test.go b/nats_test.go
index 5cee524..69cfb28 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -2854,7 +2854,7 @@ func TestCustomInboxPrefix(t *testing.T) {
 	for _, p := range []string{"$BOB.", "$BOB.*", "$BOB.>", ">", ".", "", "BOB.*.X", "BOB.>.X"} {
 		err := CustomInboxPrefix(p)(opts)
 		if err == nil {
-			t.Fatalf("Expeted error for %q", p)
+			t.Fatalf("Expected error for %q", p)
 		}
 	}
 
diff --git a/test/compat_test.go b/test/compat_test.go
new file mode 100644
index 0000000..5a0f0ba
--- /dev/null
+++ b/test/compat_test.go
@@ -0,0 +1,370 @@
+//go:build compat
+// +build compat
+
+package test
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"encoding/json"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+type objectStepConfig[T any] struct {
+	Suite   string `json:"suite"`
+	Test    string `json:"test"`
+	Command string `json:"command"`
+	URL     string `json:"url"`
+	Bucket  string `json:"bucket"`
+	Object  string `json:"object"`
+	Config  T      `json:"config"`
+}
+
+func TestCompatibilityObjectStoreDefaultBucket(t *testing.T) {
+	t.Parallel()
+	nc, js := connect(t)
+	defer nc.Close()
+
+	// setup subscription on which tester will be sending requests
+	sub, err := nc.SubscribeSync("tests.object-store.default-bucket.>")
+	if err != nil {
+		t.Fatalf("Error subscribing to test subject: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	// 1. Create default bucket
+	msg, err := sub.NextMsg(1 * time.Hour)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+
+	_, err = js.CreateObjectStore(&nats.ObjectStoreConfig{
+		Bucket: "test",
+	})
+	if err != nil {
+		t.Fatalf("Error creating object store: %v", err)
+	}
+	// send empty response to indicate client is done
+	if err := msg.Respond(nil); err != nil {
+		t.Fatalf("Error responding to message: %v", err)
+	}
+	validateTestResult(t, sub)
+}
+
+func TestCompatibilityObjectStoreCustomBucket(t *testing.T) {
+	t.Parallel()
+	nc, js := connect(t)
+	defer nc.Close()
+
+	// setup subscription on which tester will be sending requests
+	sub, err := nc.SubscribeSync("tests.object-store.custom-bucket.>")
+	if err != nil {
+		t.Fatalf("Error subscribing to test subject: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	// 1. Create custom bucket
+	msg, err := sub.NextMsg(1 * time.Hour)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+	var cfg objectStepConfig[*nats.ObjectStoreConfig]
+	if err := json.Unmarshal(msg.Data, &cfg); err != nil {
+		t.Fatalf("Error unmarshalling message: %v", err)
+	}
+
+	_, err = js.CreateObjectStore(cfg.Config)
+	if err != nil {
+		t.Fatalf("Error creating object store: %v", err)
+	}
+	// send empty response to indicate client is done
+	if err := msg.Respond(nil); err != nil {
+		t.Fatalf("Error responding to message: %v", err)
+	}
+
+	validateTestResult(t, sub)
+}
+
+func TestCompatibilityObjectStoreGetObject(t *testing.T) {
+	t.Parallel()
+	type config struct {
+		Bucket string `json:"bucket"`
+		Object string `json:"object"`
+	}
+
+	nc, js := connect(t)
+	defer nc.Close()
+
+	// setup subscription on which tester will be sending requests
+	sub, err := nc.SubscribeSync("tests.object-store.get-object.>")
+	if err != nil {
+		t.Fatalf("Error subscribing to test subject: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	msg, err := sub.NextMsg(1 * time.Hour)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+	var cfg config
+	if err := json.Unmarshal(msg.Data, &cfg); err != nil {
+		t.Fatalf("Error unmarshalling message: %v", err)
+	}
+	// Get object
+	os, err := js.ObjectStore(cfg.Bucket)
+	if err != nil {
+		t.Fatalf("Error getting object store: %v", err)
+	}
+	obj, err := os.Get(cfg.Object)
+	if err != nil {
+		t.Fatalf("Error creating object store: %v", err)
+	}
+	data, err := io.ReadAll(obj)
+	if err != nil {
+		t.Fatalf("Error reading object: %v", err)
+	}
+
+	// calculate sha256 of the object
+	h := sha256.New()
+	h.Write(data)
+	sha := h.Sum(nil)
+
+	// send response to indicate client is done
+	if err := msg.Respond(sha); err != nil {
+		t.Fatalf("Error responding to message: %v", err)
+	}
+	validateTestResult(t, sub)
+}
+
+func TestCompatibilityObjectStorePutObject(t *testing.T) {
+	t.Parallel()
+
+	nc, js := connect(t)
+	defer nc.Close()
+
+	// setup subscription on which tester will be sending requests
+	sub, err := nc.SubscribeSync("tests.object-store.put-object.>")
+	if err != nil {
+		t.Fatalf("Error subscribing to test subject: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	msg, err := sub.NextMsg(1 * time.Hour)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+	// Put object
+	var putObjectCfg objectStepConfig[*nats.ObjectMeta]
+	if err := json.Unmarshal(msg.Data, &putObjectCfg); err != nil {
+		t.Fatalf("Error unmarshalling message: %v", err)
+	}
+	os, err := js.ObjectStore(putObjectCfg.Bucket)
+	if err != nil {
+		t.Fatalf("Error getting object store: %v", err)
+	}
+	client := http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{DisableKeepAlives: true}}
+	resp, err := client.Get(putObjectCfg.URL)
+	if err != nil {
+		t.Fatalf("Error getting content: %v", err)
+	}
+	data, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("Error reading content: %v", err)
+	}
+	defer resp.Body.Close()
+	if _, err := os.Put(putObjectCfg.Config, bytes.NewBuffer(data)); err != nil {
+		t.Fatalf("Error putting object: %v", err)
+	}
+	if err := msg.Respond(nil); err != nil {
+		t.Fatalf("Error responding to message: %v", err)
+	}
+	validateTestResult(t, sub)
+}
+
+func TestCompatibilityObjectStoreUpdateMetadata(t *testing.T) {
+	t.Parallel()
+
+	nc, js := connect(t)
+	defer nc.Close()
+
+	// setup subscription on which tester will be sending requests
+	sub, err := nc.SubscribeSync("tests.object-store.update-metadata.>")
+	if err != nil {
+		t.Fatalf("Error subscribing to test subject: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	msg, err := sub.NextMsg(1 * time.Hour)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+	// Update object metadata
+	var putObjectCfg objectStepConfig[*nats.ObjectMeta]
+	if err := json.Unmarshal(msg.Data, &putObjectCfg); err != nil {
+		t.Fatalf("Error unmarshalling message: %v", err)
+	}
+	os, err := js.ObjectStore(putObjectCfg.Bucket)
+	if err != nil {
+		t.Fatalf("Error getting object store: %v", err)
+	}
+	if err := os.UpdateMeta(putObjectCfg.Object, putObjectCfg.Config); err != nil {
+		t.Fatalf("Error putting object: %v", err)
+	}
+	if err := msg.Respond(nil); err != nil {
+		t.Fatalf("Error responding to message: %v", err)
+	}
+	validateTestResult(t, sub)
+}
+
+func TestCompatibilityObjectStoreWatch(t *testing.T) {
+	t.Parallel()
+
+	type config struct {
+		Bucket string `json:"bucket"`
+		Object string `json:"object"`
+	}
+
+	nc, js := connect(t)
+	defer nc.Close()
+
+	// setup subscription on which tester will be sending requests
+	sub, err := nc.SubscribeSync("tests.object-store.watch.>")
+	if err != nil {
+		t.Fatalf("Error subscribing to test subject: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	msg, err := sub.NextMsg(1 * time.Hour)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+	// Watch object
+	var cfg config
+	if err := json.Unmarshal(msg.Data, &cfg); err != nil {
+		t.Fatalf("Error unmarshalling message: %v", err)
+	}
+	os, err := js.ObjectStore(cfg.Bucket)
+	if err != nil {
+		t.Fatalf("Error getting object store: %v", err)
+	}
+	watcher, err := os.Watch()
+	if err != nil {
+		t.Fatalf("Error getting watcher: %v", err)
+	}
+	var digests []string
+	var info *nats.ObjectInfo
+
+	// get the initial value
+	select {
+	case info = <-watcher.Updates():
+		digests = append(digests, info.Digest)
+	case <-time.After(30 * time.Second):
+		t.Fatalf("Timeout waiting for object update")
+	}
+
+	// init done, should receive nil
+	select {
+	case info = <-watcher.Updates():
+		if info != nil {
+			t.Fatalf("Expected nil, got: %v", info)
+		}
+	case <-time.After(30 * time.Second):
+		t.Fatalf("Timeout waiting for object update")
+	}
+
+	// get the updated value
+	select {
+	case info = <-watcher.Updates():
+		digests = append(digests, info.Digest)
+	case <-time.After(30 * time.Second):
+		t.Fatalf("Timeout waiting for object update")
+	}
+
+	if err := msg.Respond([]byte(strings.Join(digests, ","))); err != nil {
+		t.Fatalf("Error responding to message: %v", err)
+	}
+	validateTestResult(t, sub)
+}
+
+func TestCompatibilityObjectStoreWatchUpdates(t *testing.T) {
+	t.Parallel()
+
+	type config struct {
+		Bucket string `json:"bucket"`
+		Object string `json:"object"`
+	}
+
+	nc, js := connect(t)
+	defer nc.Close()
+
+	// setup subscription on which tester will be sending requests
+	sub, err := nc.SubscribeSync("tests.object-store.watch-updates.>")
+	if err != nil {
+		t.Fatalf("Error subscribing to test subject: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	msg, err := sub.NextMsg(1 * time.Hour)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+	// Watch object
+	var cfg config
+	if err := json.Unmarshal(msg.Data, &cfg); err != nil {
+		t.Fatalf("Error unmarshalling message: %v", err)
+	}
+	os, err := js.ObjectStore(cfg.Bucket)
+	if err != nil {
+		t.Fatalf("Error getting object store: %v", err)
+	}
+	watcher, err := os.Watch(nats.UpdatesOnly())
+	if err != nil {
+		t.Fatalf("Error getting watcher: %v", err)
+	}
+	var info *nats.ObjectInfo
+	select {
+	case info = <-watcher.Updates():
+	case <-time.After(30 * time.Second):
+		t.Fatalf("Timeout waiting for object update")
+	}
+
+	if err := msg.Respond([]byte(info.Digest)); err != nil {
+		t.Fatalf("Error responding to message: %v", err)
+	}
+	validateTestResult(t, sub)
+}
+
+func validateTestResult(t *testing.T, sub *nats.Subscription) {
+	t.Helper()
+	stepEnd, err := sub.NextMsg(5 * time.Second)
+	if err != nil {
+		t.Fatalf("Error getting message: %v", err)
+	}
+	if strings.Contains(string(stepEnd.Subject), "fail") {
+		t.Fatalf("Test step failed: %v", string(stepEnd.Subject))
+	}
+}
+
+func connect(t *testing.T) (*nats.Conn, nats.JetStreamContext) {
+	t.Helper()
+	natsURL := os.Getenv("NATS_URL")
+	if natsURL == "" {
+		natsURL = nats.DefaultURL
+	}
+	nc, err := nats.Connect(natsURL, nats.Timeout(1*time.Hour))
+	if err != nil {
+		t.Fatalf("Error connecting to NATS: %v", err)
+	}
+	js, err := nc.JetStream()
+	if err != nil {
+		t.Fatalf("Error getting JetStream context: %v", err)
+	}
+	return nc, js
+}
diff --git a/test/configs/docker/Dockerfile b/test/configs/docker/Dockerfile
new file mode 100644
index 0000000..450aced
--- /dev/null
+++ b/test/configs/docker/Dockerfile
@@ -0,0 +1,7 @@
+FROM golang:1.20
+WORKDIR /usr/src/nats.go
+COPY . /usr/src/nats.go
+RUN go mod tidy -modfile go_test.mod
+RUN go test -run TestNone -modfile go_test.mod -tags compat ./test/...
+ENV NATS_URL=localhost:4222
+CMD go test -v -run TestCompatibility -modfile go_test.mod -tags compat ./test/... -count 1 -parallel 10
diff --git a/test/conn_test.go b/test/conn_test.go
index 5c0d6fe..777cb6f 100644
--- a/test/conn_test.go
+++ b/test/conn_test.go
@@ -2279,7 +2279,7 @@ func TestReceiveInfoWithEmptyConnectURLs(t *testing.T) {
 				// The processing of PONG in sendConnect() should leave the
 				// rest for the readLoop to process.
 				c.Write([]byte("PONG\r\nINFO {\"server_id\":\"server1\",\"connect_urls\":[\"127.0.0.1:4222\", \"127.0.0.1:4223\", \"127.0.0.1:4224\"]}\r\n"))
-				// Wait for the notication
+				// Wait for the notification
 				<-ch
 				// Close the connection in our side and go back into accept
 				c.Close()
diff --git a/test/drain_test.go b/test/drain_test.go
index cf8f279..4db4bca 100644
--- a/test/drain_test.go
+++ b/test/drain_test.go
@@ -257,7 +257,7 @@ func TestDrainConnection(t *testing.T) {
 	expected := int32(50)
 	sleep := 10 * time.Millisecond
 
-	// Create the listner for responses on "bar"
+	// Create the listener for responses on "bar"
 	_, err = nc2.Subscribe("bar", func(_ *nats.Msg) {
 		r := atomic.AddInt32(&responses, 1)
 		if r == expected {
diff --git a/test/js_test.go b/test/js_test.go
index 421fa0b..d9ad8ca 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -1424,7 +1424,7 @@ func TestPullSubscribeFetchBatch(t *testing.T) {
 			t.Fatalf("Unexpected error: %s", err)
 		}
 		go func() {
-			time.Sleep(10 * time.Millisecond)
+			time.Sleep(200 * time.Millisecond)
 			cancel()
 		}()
 		msgs := make([]*nats.Msg, 0)
@@ -2727,7 +2727,7 @@ func TestAccountInfo(t *testing.T) {
 			}
 
 			if !reflect.DeepEqual(test.expected, info) {
-				t.Fatalf("Accoount info does not match; expected: %v; got: %v", test.expected, info)
+				t.Fatalf("Account info does not match; expected: %v; got: %v", test.expected, info)
 			}
 			_, err = js.AddStream(&nats.StreamConfig{Name: "FOO", MaxBytes: 1024})
 			if err != nil {
@@ -2749,7 +2749,7 @@ func TestAccountInfo(t *testing.T) {
 			}
 
 			if !reflect.DeepEqual(test.expected, info) {
-				t.Fatalf("Accoount info does not match; expected: %v; got: %v", test.expected, info)
+				t.Fatalf("Account info does not match; expected: %v; got: %v", test.expected, info)
 			}
 		})
 	}
@@ -4109,7 +4109,7 @@ func TestJetStreamSubscribe_DeliverPolicy(t *testing.T) {
 		t.Fatalf("Error on next msg: %v", err)
 	}
 	if string(msg.Data) != "bar msg 2" {
-		t.Fatalf("Unexepcted last message: %q", msg.Data)
+		t.Fatalf("Unexpected last message: %q", msg.Data)
 	}
 }
 
@@ -7869,7 +7869,7 @@ func TestJetStreamDomain(t *testing.T) {
 	}
 }
 
-// Test that we properly enfore per subject msg limits.
+// Test that we properly enforce per subject msg limits.
 func TestJetStreamMaxMsgsPerSubject(t *testing.T) {
 	const subjectMax = 5
 	msc := nats.StreamConfig{
diff --git a/test/kv_test.go b/test/kv_test.go
index 56e547a..026891a 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -330,11 +330,8 @@ func TestKeyValueWatch(t *testing.T) {
 		watcher, err := kv.WatchAll(nats.UpdatesOnly())
 		expectOk(t, err)
 		defer watcher.Stop()
-		expectInitDone := expectInitDoneF(t, watcher)
 		expectUpdate := expectUpdateF(t, watcher)
 		expectDelete := expectDeleteF(t, watcher)
-		// make sure we do not get any values here
-		expectInitDone()
 
 		// now update some keys and expect updates
 		kv.Put("name", []byte("pp"))
@@ -356,11 +353,8 @@ func TestKeyValueWatch(t *testing.T) {
 		watcher, err = kv.Watch("t.*", nats.UpdatesOnly())
 		expectOk(t, err)
 		defer watcher.Stop()
-		expectInitDone = expectInitDoneF(t, watcher)
 		expectUpdate = expectUpdateF(t, watcher)
 
-		expectInitDone()
-
 		// update some keys and expect updates
 		kv.Put("t.name", []byte("pp"))
 		expectUpdate("t.name", "pp", 11)
diff --git a/test/object_test.go b/test/object_test.go
index 550015c..93e345e 100644
--- a/test/object_test.go
+++ b/test/object_test.go
@@ -79,7 +79,7 @@ func TestObjectBasics(t *testing.T) {
 	status, err := obs.Status()
 	expectOk(t, err)
 	if !status.Sealed() {
-		t.Fatalf("exected sealed status")
+		t.Fatalf("expected sealed status")
 	}
 	if status.Size() == 0 {
 		t.Fatalf("size is 0")
@@ -474,86 +474,136 @@ func TestObjectMetadata(t *testing.T) {
 }
 
 func TestObjectWatch(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer shutdownJSServerAndRemoveStorage(t, s)
-
-	nc, js := jsClient(t, s)
-	defer nc.Close()
-
-	obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "WATCH-TEST"})
-	expectOk(t, err)
-
-	watcher, err := obs.Watch()
-	expectOk(t, err)
-	defer watcher.Stop()
-
-	expectUpdate := func(name string) {
-		t.Helper()
-		select {
-		case info := <-watcher.Updates():
-			if false && info.Name != name { // TODO what is supposed to happen here?
-				t.Fatalf("Expected update for %q, but got %+v", name, info)
+	expectUpdateF := func(t *testing.T, watcher nats.ObjectWatcher) func(name string) {
+		return func(name string) {
+			t.Helper()
+			select {
+			case info := <-watcher.Updates():
+				if false && info.Name != name { // TODO what is supposed to happen here?
+					t.Fatalf("Expected update for %q, but got %+v", name, info)
+				}
+			case <-time.After(time.Second):
+				t.Fatalf("Did not receive an update like expected")
 			}
-		case <-time.After(time.Second):
-			t.Fatalf("Did not receive an update like expected")
 		}
 	}
 
-	expectNoMoreUpdates := func() {
-		t.Helper()
-		select {
-		case info := <-watcher.Updates():
-			t.Fatalf("Got an unexpected update: %+v", info)
-		case <-time.After(100 * time.Millisecond):
+	expectNoMoreUpdatesF := func(t *testing.T, watcher nats.ObjectWatcher) func() {
+		return func() {
+			t.Helper()
+			select {
+			case info := <-watcher.Updates():
+				t.Fatalf("Got an unexpected update: %+v", info)
+			case <-time.After(100 * time.Millisecond):
+			}
 		}
 	}
 
-	expectInitDone := func() {
-		t.Helper()
-		select {
-		case info := <-watcher.Updates():
-			if info != nil {
-				t.Fatalf("Did not get expected: %+v", info)
+	expectInitDoneF := func(t *testing.T, watcher nats.ObjectWatcher) func() {
+		return func() {
+			t.Helper()
+			select {
+			case info := <-watcher.Updates():
+				if info != nil {
+					t.Fatalf("Did not get expected: %+v", info)
+				}
+			case <-time.After(time.Second):
+				t.Fatalf("Did not receive a init done like expected")
 			}
-		case <-time.After(time.Second):
-			t.Fatalf("Did not receive a init done like expected")
 		}
 	}
 
-	// We should get a marker that is nil when all initital values are delivered.
-	expectInitDone()
+	t.Run("default watcher", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
 
-	_, err = obs.PutString("A", "AAA")
-	expectOk(t, err)
-	_, err = obs.PutString("B", "BBB")
-	expectOk(t, err)
+		nc, js := jsClient(t, s)
+		defer nc.Close()
 
-	// Initial Values.
-	expectUpdate("A")
-	expectUpdate("B")
-	expectNoMoreUpdates()
+		obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "WATCH-TEST"})
+		expectOk(t, err)
 
-	// Delete
-	err = obs.Delete("A")
-	expectOk(t, err)
+		watcher, err := obs.Watch()
+		expectOk(t, err)
+		defer watcher.Stop()
 
-	expectUpdate("A")
-	expectNoMoreUpdates()
+		expectUpdate := expectUpdateF(t, watcher)
+		expectNoMoreUpdates := expectNoMoreUpdatesF(t, watcher)
+		expectInitDone := expectInitDoneF(t, watcher)
 
-	// New
-	_, err = obs.PutString("C", "CCC")
-	expectOk(t, err)
+		// We should get a marker that is nil when all initial values are delivered.
+		expectInitDone()
 
-	// Update Meta
-	deletedInfo, err := obs.GetInfo("A", nats.GetObjectInfoShowDeleted())
-	expectOk(t, err)
-	if !deletedInfo.Deleted {
-		t.Fatalf("Expected object to be deleted.")
-	}
-	meta := &deletedInfo.ObjectMeta
-	meta.Description = "Making a change."
-	err = obs.UpdateMeta("A", meta)
-	expectErr(t, err, nats.ErrUpdateMetaDeleted)
+		_, err = obs.PutString("A", "AAA")
+		expectOk(t, err)
+		_, err = obs.PutString("B", "BBB")
+		expectOk(t, err)
+
+		// Initial Values.
+		expectUpdate("A")
+		expectUpdate("B")
+		expectNoMoreUpdates()
+
+		// Delete
+		err = obs.Delete("A")
+		expectOk(t, err)
+
+		expectUpdate("A")
+		expectNoMoreUpdates()
+
+		// New
+		_, err = obs.PutString("C", "CCC")
+		expectOk(t, err)
+
+		// Update Meta
+		deletedInfo, err := obs.GetInfo("A", nats.GetObjectInfoShowDeleted())
+		expectOk(t, err)
+		if !deletedInfo.Deleted {
+			t.Fatalf("Expected object to be deleted.")
+		}
+		meta := &deletedInfo.ObjectMeta
+		meta.Description = "Making a change."
+		err = obs.UpdateMeta("A", meta)
+		expectErr(t, err, nats.ErrUpdateMetaDeleted)
+	})
+
+	t.Run("watcher with update", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "WATCH-TEST"})
+		expectOk(t, err)
+
+		_, err = obs.PutString("A", "AAA")
+		expectOk(t, err)
+		_, err = obs.PutString("B", "BBB")
+		expectOk(t, err)
+
+		watcher, err := obs.Watch(nats.UpdatesOnly())
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectUpdate := expectUpdateF(t, watcher)
+		expectNoMoreUpdates := expectNoMoreUpdatesF(t, watcher)
+
+		// when listening for updates only, we should not receive anything when watcher is started
+		expectNoMoreUpdates()
+
+		// Delete
+		err = obs.Delete("A")
+		expectOk(t, err)
+
+		expectUpdate("A")
+		expectNoMoreUpdates()
+
+		// New
+		_, err = obs.PutString("C", "CCC")
+		expectOk(t, err)
+		expectUpdate("C")
+	})
 }
 
 func TestObjectLinks(t *testing.T) {