Repository: nats-io/nats.go
Pull request #1371 (2023-08-21 09:02:52)

[FIXED] Nil pointer when sub is nil in service API err handler


Patch:
diff --git a/.github/ISSUE_TEMPLATE/blank_issue.md b/.github/ISSUE_TEMPLATE/blank_issue.md
deleted file mode 100644
index 38ba946..0000000
--- a/.github/ISSUE_TEMPLATE/blank_issue.md
+++ /dev/null
@@ -1,4 +0,0 @@
----
-name: Blank Issue
-about: Create an issue with a blank template.
----
diff --git a/.github/ISSUE_TEMPLATE/config.yml b/.github/ISSUE_TEMPLATE/config.yml
index 989f721..9502575 100644
--- a/.github/ISSUE_TEMPLATE/config.yml
+++ b/.github/ISSUE_TEMPLATE/config.yml
@@ -1,5 +1,8 @@
-blank_issues_enabled: true
+blank_issues_enabled: false
 contact_links:
-  - name: NATS Slack
+  - name: Discussion
+    url: https://github.com/nats-io/nats.go/discussions
+    about: Ideal for ideas, feedback, or longer form questions.
+  - name: Chat
     url: https://slack.nats.io
-    about: Please ask and answer questions in our Slack server here.
+    about: Ideal for short, one-off questions, general conversation, and meeting other NATS users!
diff --git a/.github/ISSUE_TEMPLATE/defect.md b/.github/ISSUE_TEMPLATE/defect.md
deleted file mode 100644
index be4ead9..0000000
--- a/.github/ISSUE_TEMPLATE/defect.md
+++ /dev/null
@@ -1,23 +0,0 @@
----
-name: Bug Report
-about: Report a bug found in the NATS Go library
-labels: bug
----
-
-## Defect
-
-Make sure that these boxes are checked before submitting your issue -- thank you!
-
- - [ ] Included nats.go version
- - [ ] Included a [Minimal, Complete, and Verifiable example] (https://stackoverflow.com/help/mcve)
-
-#### Versions of `nats.go` and the `nats-server` if one was involved:
-
-#### OS/Container environment:
-
-#### Steps or code to reproduce the issue:
-
-#### Expected result:
-
-#### Actual result:
-
diff --git a/.github/ISSUE_TEMPLATE/defect.yml b/.github/ISSUE_TEMPLATE/defect.yml
new file mode 100644
index 0000000..99b4800
--- /dev/null
+++ b/.github/ISSUE_TEMPLATE/defect.yml
@@ -0,0 +1,41 @@
+---
+name: Defect
+description: Report a defect, such as a bug or regression.
+labels:
+  - defect
+body:
+  - type: textarea
+    id: versions
+    attributes:
+      label: What version were you using?
+      description: Include the server version (`nats-server --version`) and any client versions when observing the issue.
+    validations:
+      required: true
+  - type: textarea
+    id: environment
+    attributes:
+      label: What environment was the server running in?
+      description: This pertains to the operating system, CPU architecture, and/or Docker image that was used.
+    validations:
+      required: true
+  - type: textarea
+    id: steps
+    attributes:
+      label: Is this defect reproducible?
+      description: Provide best-effort steps to showcase the defect.
+    validations:
+      required: true
+  - type: textarea
+    id: expected
+    attributes:
+      label: Given the capability you are leveraging, describe your expectation?
+      description: This may be the expected behavior or performance characteristics.
+    validations:
+      required: true
+  - type: textarea
+    id: actual
+    attributes:
+      label: Given the expectation, what is the defect you are observing?
+      description: This may be an unexpected behavior or regression in performance.
+    validations:
+      required: true
diff --git a/.github/ISSUE_TEMPLATE/feature_request.md b/.github/ISSUE_TEMPLATE/feature_request.md
deleted file mode 100644
index 10de66e..0000000
--- a/.github/ISSUE_TEMPLATE/feature_request.md
+++ /dev/null
@@ -1,16 +0,0 @@
----
-name: Feature Request
-about: Request a feature for the NATS Go library
-labels: feature
----
-
-## Feature Request
-
-#### Use Case:
-
-#### Proposed Change:
-
-#### Who Benefits From The Change(s)?
-
-#### Alternative Approaches
-
diff --git a/.github/ISSUE_TEMPLATE/proposal.yml b/.github/ISSUE_TEMPLATE/proposal.yml
new file mode 100644
index 0000000..d7da0ca
--- /dev/null
+++ b/.github/ISSUE_TEMPLATE/proposal.yml
@@ -0,0 +1,34 @@
+---
+name: Proposal
+description: Propose an enhancement or new feature.
+labels:
+  - proposal
+body:
+  - type: textarea
+    id: usecase
+    attributes:
+      label: What motivated this proposal?
+      description: Describe the use case justifying this request.
+    validations:
+      required: true
+  - type: textarea
+    id: change
+    attributes:
+      label: What is the proposed change?
+      description: This could be a behavior change, enhanced API, or a branch new feature.
+    validations:
+      required: true
+  - type: textarea
+    id: benefits
+    attributes:
+      label: Who benefits from this change?
+      description: Describe how this not only benefits you.
+    validations:
+      required: false
+  - type: textarea
+    id: alternates
+    attributes:
+      label: What alternatives have you evaluated?
+      description: This could be using existing features or relying on an external dependency.
+    validations:
+      required: false
diff --git a/.golangci.yaml b/.golangci.yaml
index 5bbb3e7..be66189 100644
--- a/.golangci.yaml
+++ b/.golangci.yaml
@@ -8,3 +8,6 @@ issues:
     - linters:
       - errcheck
       text: "msg.Ack"
+    - linters:
+      - errcheck
+      text: "watcher.Stop"
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/errors.go b/jetstream/errors.go
index ebfdebc..ffefea5 100644
--- a/jetstream/errors.go
+++ b/jetstream/errors.go
@@ -57,6 +57,8 @@ const (
 	JSErrCodeMessageNotFound ErrorCode = 10037
 
 	JSErrCodeBadRequest ErrorCode = 10003
+
+	JSErrCodeStreamWrongLastSequence ErrorCode = 10071
 )
 
 var (
@@ -178,6 +180,38 @@ var (
 	// ErrOrderedConsumerNotCreated is returned when trying to get consumer info of an
 	// ordered consumer which was not yet created.
 	ErrOrderedConsumerNotCreated = &jsError{message: "consumer instance not yet created"}
+
+	// KeyValue Errors
+
+	// ErrKeyExists is returned when attempting to create a key that already exists.
+	ErrKeyExists JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeStreamWrongLastSequence, Code: 400}, message: "key exists"}
+
+	// ErrKeyValueConfigRequired is returned when attempting to create a bucket without a config.
+	ErrKeyValueConfigRequired = &jsError{message: "config required"}
+
+	// ErrInvalidBucketName is returned when attempting to create a bucket with an invalid name.
+	ErrInvalidBucketName = &jsError{message: "invalid bucket name"}
+
+	// ErrInvalidKey is returned when attempting to create a key with an invalid name.
+	ErrInvalidKey = &jsError{message: "invalid key"}
+
+	// ErrBucketNotFound is returned when attempting to access a bucket that does not exist.
+	ErrBucketNotFound = &jsError{message: "bucket not found"}
+
+	// ErrBadBucket is returned when attempting to access a bucket that is not a key-value store.
+	ErrBadBucket = &jsError{message: "bucket not valid key-value store"}
+
+	// ErrKeyNotFound is returned when attempting to access a key that does not exist.
+	ErrKeyNotFound = &jsError{message: "key not found"}
+
+	// ErrKeyDeleted is returned when attempting to access a key that was deleted.
+	ErrKeyDeleted = &jsError{message: "key was deleted"}
+
+	// ErrHistoryToLarge is returned when provided history limit is larger than 64.
+	ErrHistoryTooLarge = &jsError{message: "history limited to a max of 64"}
+
+	// ErrNoKeysFound is returned when no keys are found.
+	ErrNoKeysFound = &jsError{message: "no keys found"}
 )
 
 // Error prints the JetStream API error code and description
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index cf1d211..1ff86c3 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -42,6 +42,7 @@ type (
 		StreamConsumerManager
 		StreamManager
 		Publisher
+		KeyValueManager
 	}
 
 	Publisher interface {
@@ -720,9 +721,6 @@ func (s *streamLister) streamInfos(ctx context.Context, streamsReq streamsReques
 	if err != nil {
 		return nil, err
 	}
-	if err != nil {
-		return nil, err
-	}
 
 	slSubj := apiSubj(s.js.apiPrefix, apiStreamListT)
 	var resp streamListResponse
diff --git a/jetstream/kv.go b/jetstream/kv.go
new file mode 100644
index 0000000..3546f01
--- /dev/null
+++ b/jetstream/kv.go
@@ -0,0 +1,1025 @@
+// Copyright 2023 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package jetstream
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/internal/parser"
+)
+
+// Public interfaces and structs
+type (
+	// KeyValueManager is used to manage KeyValue stores.
+	KeyValueManager interface {
+		// KeyValue will lookup and bind to an existing KeyValue store.
+		KeyValue(ctx context.Context, bucket string) (KeyValue, error)
+		// CreateKeyValue will create a KeyValue store with the following configuration.
+		CreateKeyValue(ctx context.Context, cfg KeyValueConfig) (KeyValue, error)
+		// DeleteKeyValue will delete this KeyValue store (JetStream stream).
+		DeleteKeyValue(ctx context.Context, bucket string) error
+		// KeyValueStoreNames is used to retrieve a list of key value store names
+		KeyValueStoreNames(ctx context.Context) KeyValueNamesLister
+		// KeyValueStores is used to retrieve a list of key value store statuses
+		KeyValueStores(ctx context.Context) KeyValueLister
+	}
+
+	// KeyValue contains methods to operate on a KeyValue store.
+	KeyValue interface {
+		// Get returns the latest value for the key.
+		Get(ctx context.Context, key string) (KeyValueEntry, error)
+		// GetRevision returns a specific revision value for the key.
+		GetRevision(ctx context.Context, key string, revision uint64) (KeyValueEntry, error)
+		// Put will place the new value for the key into the store.
+		Put(ctx context.Context, key string, value []byte) (uint64, error)
+		// PutString will place the string for the key into the store.
+		PutString(ctx context.Context, key string, value string) (uint64, error)
+		// Create will add the key/value pair if it does not exist.
+		Create(ctx context.Context, key string, value []byte) (uint64, error)
+		// Update will update the value if the latest revision matches.
+		Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error)
+		// Delete will place a delete marker and leave all revisions.
+		Delete(ctx context.Context, key string, opts ...KVDeleteOpt) error
+		// Purge will place a delete marker and remove all previous revisions.
+		Purge(ctx context.Context, key string, opts ...KVDeleteOpt) error
+		// Watch for any updates to keys that match the keys argument which could include wildcards.
+		// Watch will send a nil entry when it has received all initial values.
+		Watch(ctx context.Context, keys string, opts ...WatchOpt) (KeyWatcher, error)
+		// WatchAll will invoke the callback for all updates.
+		WatchAll(ctx context.Context, opts ...WatchOpt) (KeyWatcher, error)
+		// Keys will return all keys.
+		Keys(ctx context.Context, opts ...WatchOpt) ([]string, error)
+		// History will return all historical values for the key.
+		History(ctx context.Context, key string, opts ...WatchOpt) ([]KeyValueEntry, error)
+		// Bucket returns the current bucket name.
+		Bucket() string
+		// PurgeDeletes will remove all current delete markers.
+		PurgeDeletes(ctx context.Context, opts ...KVPurgeOpt) error
+		// Status retrieves the status and configuration of a bucket
+		Status(ctx context.Context) (KeyValueStatus, error)
+	}
+
+	// KeyValueConfig is for configuring a KeyValue store.
+	KeyValueConfig struct {
+		Bucket       string
+		Description  string
+		MaxValueSize int32
+		History      uint8
+		TTL          time.Duration
+		MaxBytes     int64
+		Storage      StorageType
+		Replicas     int
+		Placement    *Placement
+		RePublish    *RePublish
+		Mirror       *StreamSource
+		Sources      []*StreamSource
+	}
+
+	KeyValueLister interface {
+		Status() <-chan KeyValueStatus
+		Error() error
+	}
+
+	KeyValueNamesLister interface {
+		Name() <-chan string
+		Error() error
+	}
+
+	// KeyValueStatus is run-time status about a Key-Value bucket
+	KeyValueStatus interface {
+		// Bucket the name of the bucket
+		Bucket() string
+
+		// Values is how many messages are in the bucket, including historical values
+		Values() uint64
+
+		// History returns the configured history kept per key
+		History() int64
+
+		// TTL is how long the bucket keeps values for
+		TTL() time.Duration
+
+		// BackingStore indicates what technology is used for storage of the bucket
+		BackingStore() string
+
+		// Bytes returns the size in bytes of the bucket
+		Bytes() uint64
+	}
+
+	// KeyWatcher is what is returned when doing a watch.
+	KeyWatcher interface {
+		// Updates returns a channel to read any updates to entries.
+		Updates() <-chan KeyValueEntry
+		// Stop will stop this watcher.
+		Stop() error
+	}
+
+	// KeyValueEntry is a retrieved entry for Get or List or Watch.
+	KeyValueEntry interface {
+		// Bucket is the bucket the data was loaded from.
+		Bucket() string
+		// Key is the key that was retrieved.
+		Key() string
+		// Value is the retrieved value.
+		Value() []byte
+		// Revision is a unique sequence for this value.
+		Revision() uint64
+		// Created is the time the data was put in the bucket.
+		Created() time.Time
+		// Delta is distance from the latest value.
+		Delta() uint64
+		// Operation returns Put or Delete or Purge.
+		Operation() KeyValueOp
+	}
+)
+
+// Option types
+
+type (
+	WatchOpt interface {
+		configureWatcher(opts *watchOpts) error
+	}
+
+	watchOpts struct {
+		// Do not send delete markers to the update channel.
+		ignoreDeletes bool
+		// Include all history per subject, not just last one.
+		includeHistory bool
+		// Include only updates for keys.
+		updatesOnly bool
+		// retrieve only the meta data of the entry
+		metaOnly bool
+	}
+
+	KVDeleteOpt interface {
+		configureDelete(opts *deleteOpts) error
+	}
+
+	deleteOpts struct {
+		// Remove all previous revisions.
+		purge bool
+
+		// Delete only if the latest revision matches.
+		revision uint64
+	}
+
+	KVPurgeOpt interface {
+		configurePurge(opts *purgeOpts) error
+	}
+
+	purgeOpts struct {
+		dmthr time.Duration // Delete markers threshold
+	}
+)
+
+// kvs is the implementation of KeyValue
+type kvs struct {
+	name       string
+	streamName string
+	pre        string
+	putPre     string
+	pushJS     nats.JetStreamContext
+	js         *jetStream
+	stream     Stream
+	// If true, it means that APIPrefix/Domain was set in the context
+	// and we need to add something to some of our high level protocols
+	// (such as Put, etc..)
+	useJSPfx bool
+	// To know if we can use the stream direct get API
+	useDirect bool
+}
+
+// KeyValueOp represents the type of KV operation (Put, Delete, Purge)
+// Returned as part of watcher entry.
+type KeyValueOp uint8
+
+const (
+	KeyValuePut KeyValueOp = iota
+	KeyValueDelete
+	KeyValuePurge
+)
+
+func (op KeyValueOp) String() string {
+	switch op {
+	case KeyValuePut:
+		return "KeyValuePutOp"
+	case KeyValueDelete:
+		return "KeyValueDeleteOp"
+	case KeyValuePurge:
+		return "KeyValuePurgeOp"
+	default:
+		return "Unknown Operation"
+	}
+}
+
+const (
+	kvBucketNamePre         = "KV_"
+	kvBucketNameTmpl        = "KV_%s"
+	kvSubjectsTmpl          = "$KV.%s.>"
+	kvSubjectsPreTmpl       = "$KV.%s."
+	kvSubjectsPreDomainTmpl = "%s.$KV.%s."
+	kvNoPending             = "0"
+)
+
+const (
+	KeyValueMaxHistory = 64
+	AllKeys            = ">"
+	kvLatestRevision   = 0
+	kvop               = "KV-Operation"
+	kvdel              = "DEL"
+	kvpurge            = "PURGE"
+)
+
+// Regex for valid keys and buckets.
+var (
+	validBucketRe = regexp.MustCompile(`\A[a-zA-Z0-9_-]+\z`)
+	validKeyRe    = regexp.MustCompile(`\A[-/_=\.a-zA-Z0-9]+\z`)
+)
+
+func (js *jetStream) KeyValue(ctx context.Context, bucket string) (KeyValue, error) {
+	if !validBucketRe.MatchString(bucket) {
+		return nil, ErrInvalidBucketName
+	}
+	streamName := fmt.Sprintf(kvBucketNameTmpl, bucket)
+	stream, err := js.Stream(ctx, streamName)
+	if err != nil {
+		if err == ErrStreamNotFound {
+			err = ErrBucketNotFound
+		}
+		return nil, err
+	}
+	// Do some quick sanity checks that this is a correctly formed stream for KV.
+	// Max msgs per subject should be > 0.
+	if stream.CachedInfo().Config.MaxMsgsPerSubject < 1 {
+		return nil, ErrBadBucket
+	}
+	pushJS, err := js.legacyJetStream()
+	if err != nil {
+		return nil, err
+	}
+
+	return mapStreamToKVS(js, pushJS, stream), nil
+}
+
+// CreateKeyValue will create a KeyValue store with the following configuration.
+func (js *jetStream) CreateKeyValue(ctx context.Context, cfg KeyValueConfig) (KeyValue, error) {
+	if !validBucketRe.MatchString(cfg.Bucket) {
+		return nil, ErrInvalidBucketName
+	}
+	if _, err := js.AccountInfo(ctx); err != nil {
+		return nil, err
+	}
+
+	// Default to 1 for history. Max is 64 for now.
+	history := int64(1)
+	if cfg.History > 0 {
+		if cfg.History > KeyValueMaxHistory {
+			return nil, ErrHistoryTooLarge
+		}
+		history = int64(cfg.History)
+	}
+
+	replicas := cfg.Replicas
+	if replicas == 0 {
+		replicas = 1
+	}
+
+	// We will set explicitly some values so that we can do comparison
+	// if we get an "already in use" error and need to check if it is same.
+	maxBytes := cfg.MaxBytes
+	if maxBytes == 0 {
+		maxBytes = -1
+	}
+	maxMsgSize := cfg.MaxValueSize
+	if maxMsgSize == 0 {
+		maxMsgSize = -1
+	}
+	// When stream's MaxAge is not set, server uses 2 minutes as the default
+	// for the duplicate window. If MaxAge is set, and lower than 2 minutes,
+	// then the duplicate window will be set to that. If MaxAge is greater,
+	// we will cap the duplicate window to 2 minutes (to be consistent with
+	// previous behavior).
+	duplicateWindow := 2 * time.Minute
+	if cfg.TTL > 0 && cfg.TTL < duplicateWindow {
+		duplicateWindow = cfg.TTL
+	}
+	scfg := StreamConfig{
+		Name:              fmt.Sprintf(kvBucketNameTmpl, cfg.Bucket),
+		Description:       cfg.Description,
+		MaxMsgsPerSubject: history,
+		MaxBytes:          maxBytes,
+		MaxAge:            cfg.TTL,
+		MaxMsgSize:        maxMsgSize,
+		Storage:           cfg.Storage,
+		Replicas:          replicas,
+		Placement:         cfg.Placement,
+		AllowRollup:       true,
+		DenyDelete:        true,
+		Duplicates:        duplicateWindow,
+		MaxMsgs:           -1,
+		MaxConsumers:      -1,
+		AllowDirect:       true,
+		RePublish:         cfg.RePublish,
+	}
+	if cfg.Mirror != nil {
+		// Copy in case we need to make changes so we do not change caller's version.
+		m := cfg.Mirror.copy()
+		if !strings.HasPrefix(m.Name, kvBucketNamePre) {
+			m.Name = fmt.Sprintf(kvBucketNameTmpl, m.Name)
+		}
+		scfg.Mirror = m
+		scfg.MirrorDirect = true
+	} else if len(cfg.Sources) > 0 {
+		// For now we do not allow direct subjects for sources. If that is desired a user could use stream API directly.
+		for _, ss := range cfg.Sources {
+			if !strings.HasPrefix(ss.Name, kvBucketNamePre) {
+				ss = ss.copy()
+				ss.Name = fmt.Sprintf(kvBucketNameTmpl, ss.Name)
+			}
+			scfg.Sources = append(scfg.Sources, ss)
+		}
+	} else {
+		scfg.Subjects = []string{fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)}
+	}
+
+	stream, err := js.CreateStream(ctx, scfg)
+	if err != nil {
+		return nil, err
+	}
+	pushJS, err := js.legacyJetStream()
+	if err != nil {
+		return nil, err
+	}
+
+	return mapStreamToKVS(js, pushJS, stream), nil
+}
+
+// DeleteKeyValue will delete this KeyValue store (JetStream stream).
+func (js *jetStream) DeleteKeyValue(ctx context.Context, bucket string) error {
+	if !validBucketRe.MatchString(bucket) {
+		return ErrInvalidBucketName
+	}
+	stream := fmt.Sprintf(kvBucketNameTmpl, bucket)
+	return js.DeleteStream(ctx, stream)
+}
+
+// KeyValueStoreNames is used to retrieve a list of key value store names
+func (js *jetStream) KeyValueStoreNames(ctx context.Context) KeyValueNamesLister {
+	res := &kvLister{
+		kvNames: make(chan string),
+	}
+	l := &streamLister{js: js}
+	streamsReq := streamsRequest{
+		Subject: fmt.Sprintf(kvSubjectsTmpl, "*"),
+	}
+	go func() {
+		defer close(res.kvNames)
+		for {
+			page, err := l.streamNames(ctx, streamsReq)
+			if err != nil && !errors.Is(err, ErrEndOfData) {
+				res.err = err
+				return
+			}
+			for _, name := range page {
+				if !strings.HasPrefix(name, kvBucketNamePre) {
+					continue
+				}
+				res.kvNames <- name
+			}
+			if errors.Is(err, ErrEndOfData) {
+				return
+			}
+		}
+	}()
+	return res
+}
+
+// KeyValueStores is used to retrieve a list of key value store statuses
+func (js *jetStream) KeyValueStores(ctx context.Context) KeyValueLister {
+	res := &kvLister{
+		kvs: make(chan KeyValueStatus),
+	}
+	l := &streamLister{js: js}
+	streamsReq := streamsRequest{
+		Subject: fmt.Sprintf(kvSubjectsTmpl, "*"),
+	}
+	go func() {
+		defer close(res.kvs)
+		for {
+			page, err := l.streamInfos(ctx, streamsReq)
+			if err != nil && !errors.Is(err, ErrEndOfData) {
+				res.err = err
+				return
+			}
+			for _, info := range page {
+				if !strings.HasPrefix(info.Config.Name, kvBucketNamePre) {
+					continue
+				}
+				res.kvs <- &KeyValueBucketStatus{nfo: info, bucket: strings.TrimPrefix(info.Config.Name, kvBucketNamePre)}
+			}
+			if errors.Is(err, ErrEndOfData) {
+				return
+			}
+		}
+	}()
+	return res
+}
+
+// KeyValueBucketStatus represents status of a Bucket, implements KeyValueStatus
+type KeyValueBucketStatus struct {
+	nfo    *StreamInfo
+	bucket string
+}
+
+// Bucket the name of the bucket
+func (s *KeyValueBucketStatus) Bucket() string { return s.bucket }
+
+// Values is how many messages are in the bucket, including historical values
+func (s *KeyValueBucketStatus) Values() uint64 { return s.nfo.State.Msgs }
+
+// History returns the configured history kept per key
+func (s *KeyValueBucketStatus) History() int64 { return s.nfo.Config.MaxMsgsPerSubject }
+
+// TTL is how long the bucket keeps values for
+func (s *KeyValueBucketStatus) TTL() time.Duration { return s.nfo.Config.MaxAge }
+
+// BackingStore indicates what technology is used for storage of the bucket
+func (s *KeyValueBucketStatus) BackingStore() string { return "JetStream" }
+
+// StreamInfo is the stream info retrieved to create the status
+func (s *KeyValueBucketStatus) StreamInfo() *StreamInfo { return s.nfo }
+
+// Bytes is the size of the stream
+func (s *KeyValueBucketStatus) Bytes() uint64 { return s.nfo.State.Bytes }
+
+type kvLister struct {
+	kvs     chan KeyValueStatus
+	kvNames chan string
+	err     error
+}
+
+func (kl *kvLister) Status() <-chan KeyValueStatus {
+	return kl.kvs
+}
+
+func (kl *kvLister) Name() <-chan string {
+	return kl.kvNames
+}
+
+func (kl *kvLister) Error() error {
+	return kl.err
+}
+
+func (js *jetStream) legacyJetStream() (nats.JetStreamContext, error) {
+	opts := make([]nats.JSOpt, 0)
+	if js.apiPrefix != "" {
+		opts = append(opts, nats.APIPrefix(js.apiPrefix))
+	}
+	if js.clientTrace != nil {
+		opts = append(opts, nats.ClientTrace{
+			RequestSent:      js.clientTrace.RequestSent,
+			ResponseReceived: js.clientTrace.ResponseReceived,
+		})
+	}
+	return js.conn.JetStream(opts...)
+}
+
+func keyValid(key string) bool {
+	if len(key) == 0 || key[0] == '.' || key[len(key)-1] == '.' {
+		return false
+	}
+	return validKeyRe.MatchString(key)
+}
+
+func (kv *kvs) get(ctx context.Context, key string, revision uint64) (KeyValueEntry, error) {
+	if !keyValid(key) {
+		return nil, ErrInvalidKey
+	}
+
+	var b strings.Builder
+	b.WriteString(kv.pre)
+	b.WriteString(key)
+
+	var m *RawStreamMsg
+	var err error
+
+	if revision == kvLatestRevision {
+		m, err = kv.stream.GetLastMsgForSubject(ctx, b.String())
+	} else {
+		m, err = kv.stream.GetMsg(ctx, revision)
+		// If a sequence was provided, just make sure that the retrieved
+		// message subject matches the request.
+		if err == nil && m.Subject != b.String() {
+			return nil, ErrKeyNotFound
+		}
+	}
+	if err != nil {
+		if err == ErrMsgNotFound {
+			err = ErrKeyNotFound
+		}
+		return nil, err
+	}
+
+	entry := &kve{
+		bucket:   kv.name,
+		key:      key,
+		value:    m.Data,
+		revision: m.Sequence,
+		created:  m.Time,
+	}
+
+	// Double check here that this is not a DEL Operation marker.
+	if len(m.Header) > 0 {
+		switch m.Header.Get(kvop) {
+		case kvdel:
+			entry.op = KeyValueDelete
+			return entry, ErrKeyDeleted
+		case kvpurge:
+			entry.op = KeyValuePurge
+			return entry, ErrKeyDeleted
+		}
+	}
+
+	return entry, nil
+}
+
+// kve is the implementation of KeyValueEntry
+type kve struct {
+	bucket   string
+	key      string
+	value    []byte
+	revision uint64
+	delta    uint64
+	created  time.Time
+	op       KeyValueOp
+}
+
+func (e *kve) Bucket() string        { return e.bucket }
+func (e *kve) Key() string           { return e.key }
+func (e *kve) Value() []byte         { return e.value }
+func (e *kve) Revision() uint64      { return e.revision }
+func (e *kve) Created() time.Time    { return e.created }
+func (e *kve) Delta() uint64         { return e.delta }
+func (e *kve) Operation() KeyValueOp { return e.op }
+
+// Get returns the latest value for the key.
+func (kv *kvs) Get(ctx context.Context, key string) (KeyValueEntry, error) {
+	e, err := kv.get(ctx, key, kvLatestRevision)
+	if err != nil {
+		if err == ErrKeyDeleted {
+			return nil, ErrKeyNotFound
+		}
+		return nil, err
+	}
+
+	return e, nil
+}
+
+// GetRevision returns a specific revision value for the key.
+func (kv *kvs) GetRevision(ctx context.Context, key string, revision uint64) (KeyValueEntry, error) {
+	e, err := kv.get(ctx, key, revision)
+	if err != nil {
+		if err == ErrKeyDeleted {
+			return nil, ErrKeyNotFound
+		}
+		return nil, err
+	}
+
+	return e, nil
+}
+
+// Put will place the new value for the key into the store.
+func (kv *kvs) Put(ctx context.Context, key string, value []byte) (uint64, error) {
+	if !keyValid(key) {
+		return 0, ErrInvalidKey
+	}
+
+	var b strings.Builder
+	if kv.useJSPfx {
+		b.WriteString(kv.js.apiPrefix)
+	}
+	if kv.putPre != "" {
+		b.WriteString(kv.putPre)
+	} else {
+		b.WriteString(kv.pre)
+	}
+	b.WriteString(key)
+
+	pa, err := kv.js.Publish(ctx, b.String(), value)
+	if err != nil {
+		return 0, err
+	}
+	return pa.Sequence, err
+}
+
+// PutString will place the string for the key into the store.
+func (kv *kvs) PutString(ctx context.Context, key string, value string) (uint64, error) {
+	return kv.Put(ctx, key, []byte(value))
+}
+
+// Create will add the key/value pair iff it does not exist.
+func (kv *kvs) Create(ctx context.Context, key string, value []byte) (revision uint64, err error) {
+	v, err := kv.Update(ctx, key, value, 0)
+	if err == nil {
+		return v, nil
+	}
+
+	if e, err := kv.get(ctx, key, kvLatestRevision); err == ErrKeyDeleted {
+		return kv.Update(ctx, key, value, e.Revision())
+	}
+
+	// Check if the expected last subject sequence is not zero which implies
+	// the key already exists.
+	if errors.Is(err, ErrKeyExists) {
+		jserr := ErrKeyExists.(*jsError)
+		return 0, fmt.Errorf("%w: %s", err, jserr.message)
+	}
+
+	return 0, err
+}
+
+// Update will update the value if the latest revision matches.
+func (kv *kvs) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) {
+	if !keyValid(key) {
+		return 0, ErrInvalidKey
+	}
+
+	var b strings.Builder
+	if kv.useJSPfx {
+		b.WriteString(kv.js.apiPrefix)
+	}
+	b.WriteString(kv.pre)
+	b.WriteString(key)
+
+	m := nats.Msg{Subject: b.String(), Header: nats.Header{}, Data: value}
+	m.Header.Set(ExpectedLastSubjSeqHeader, strconv.FormatUint(revision, 10))
+
+	pa, err := kv.js.PublishMsg(ctx, &m)
+	if err != nil {
+		return 0, err
+	}
+	return pa.Sequence, err
+}
+
+// Delete will place a delete marker and leave all revisions.
+func (kv *kvs) Delete(ctx context.Context, key string, opts ...KVDeleteOpt) error {
+	if !keyValid(key) {
+		return ErrInvalidKey
+	}
+
+	var b strings.Builder
+	if kv.useJSPfx {
+		b.WriteString(kv.js.apiPrefix)
+	}
+	if kv.putPre != "" {
+		b.WriteString(kv.putPre)
+	} else {
+		b.WriteString(kv.pre)
+	}
+	b.WriteString(key)
+
+	// DEL op marker. For watch functionality.
+	m := nats.NewMsg(b.String())
+
+	var o deleteOpts
+	for _, opt := range opts {
+		if opt != nil {
+			if err := opt.configureDelete(&o); err != nil {
+				return err
+			}
+		}
+	}
+
+	if o.purge {
+		m.Header.Set(kvop, kvpurge)
+		m.Header.Set(MsgRollup, MsgRollupSubject)
+	} else {
+		m.Header.Set(kvop, kvdel)
+	}
+
+	if o.revision != 0 {
+		m.Header.Set(ExpectedLastSubjSeqHeader, strconv.FormatUint(o.revision, 10))
+	}
+
+	_, err := kv.js.PublishMsg(ctx, m)
+	return err
+}
+
+// Purge will place a delete marker and remove all previous revisions.
+func (kv *kvs) Purge(ctx context.Context, key string, opts ...KVDeleteOpt) error {
+	return kv.Delete(ctx, key, append(opts, purge())...)
+}
+
+// purge removes all previous revisions.
+func purge() KVDeleteOpt {
+	return deleteOptFn(func(opts *deleteOpts) error {
+		opts.purge = true
+		return nil
+	})
+}
+
+// Implementation for Watch
+type watcher struct {
+	mu          sync.Mutex
+	updates     chan KeyValueEntry
+	sub         *nats.Subscription
+	initDone    bool
+	initPending uint64
+	received    uint64
+}
+
+// Updates returns the interior channel.
+func (w *watcher) Updates() <-chan KeyValueEntry {
+	if w == nil {
+		return nil
+	}
+	return w.updates
+}
+
+// Stop will unsubscribe from the watcher.
+func (w *watcher) Stop() error {
+	if w == nil {
+		return nil
+	}
+	return w.sub.Unsubscribe()
+}
+
+// Watch for any updates to keys that match the keys argument which could include wildcards.
+// Watch will send a nil entry when it has received all initial values.
+func (kv *kvs) Watch(ctx context.Context, keys string, opts ...WatchOpt) (KeyWatcher, error) {
+	var o watchOpts
+	for _, opt := range opts {
+		if opt != nil {
+			if err := opt.configureWatcher(&o); err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	// Could be a pattern so don't check for validity as we normally do.
+	var b strings.Builder
+	b.WriteString(kv.pre)
+	b.WriteString(keys)
+	keys = b.String()
+
+	// We will block below on placing items on the chan. That is by design.
+	w := &watcher{updates: make(chan KeyValueEntry, 256)}
+
+	update := func(m *nats.Msg) {
+		tokens, err := parser.GetMetadataFields(m.Reply)
+		if err != nil {
+			return
+		}
+		if len(m.Subject) <= len(kv.pre) {
+			return
+		}
+		subj := m.Subject[len(kv.pre):]
+
+		var op KeyValueOp
+		if len(m.Header) > 0 {
+			switch m.Header.Get(kvop) {
+			case kvdel:
+				op = KeyValueDelete
+			case kvpurge:
+				op = KeyValuePurge
+			}
+		}
+		delta := parser.ParseNum(tokens[parser.AckNumPendingTokenPos])
+		w.mu.Lock()
+		defer w.mu.Unlock()
+		if !o.ignoreDeletes || (op != KeyValueDelete && op != KeyValuePurge) {
+			entry := &kve{
+				bucket:   kv.name,
+				key:      subj,
+				value:    m.Data,
+				revision: parser.ParseNum(tokens[parser.AckStreamSeqTokenPos]),
+				created:  time.Unix(0, int64(parser.ParseNum(tokens[parser.AckTimestampSeqTokenPos]))),
+				delta:    delta,
+				op:       op,
+			}
+			w.updates <- entry
+		}
+		// Check if done and initial values.
+		if !w.initDone {
+			w.received++
+			// We set this on the first trip through..
+			if w.initPending == 0 {
+				w.initPending = delta
+			}
+			if w.received > w.initPending || delta == 0 {
+				w.initDone = true
+				w.updates <- nil
+			}
+		}
+	}
+
+	// Used ordered consumer to deliver results.
+	subOpts := []nats.SubOpt{nats.BindStream(kv.streamName), nats.OrderedConsumer()}
+	if !o.includeHistory {
+		subOpts = append(subOpts, nats.DeliverLastPerSubject())
+	}
+	if o.updatesOnly {
+		subOpts = append(subOpts, nats.DeliverNew())
+	}
+	if o.metaOnly {
+		subOpts = append(subOpts, nats.HeadersOnly())
+	}
+	subOpts = append(subOpts, nats.Context(ctx))
+	// Create the sub and rest of initialization under the lock.
+	// We want to prevent the race between this code and the
+	// update() callback.
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	sub, err := kv.pushJS.Subscribe(keys, update, subOpts...)
+	if err != nil {
+		return nil, err
+	}
+	sub.SetClosedHandler(func(_ string) {
+		close(w.updates)
+	})
+	// If there were no pending messages at the time of the creation
+	// of the consumer, send the marker.
+	// 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.sub = sub
+	return w, nil
+}
+
+// WatchAll will invoke the callback for all updates.
+func (kv *kvs) WatchAll(ctx context.Context, opts ...WatchOpt) (KeyWatcher, error) {
+	return kv.Watch(ctx, AllKeys, opts...)
+}
+
+// Keys will return all keys.
+func (kv *kvs) Keys(ctx context.Context, opts ...WatchOpt) ([]string, error) {
+	opts = append(opts, IgnoreDeletes(), MetaOnly())
+	watcher, err := kv.WatchAll(ctx, opts...)
+	if err != nil {
+		return nil, err
+	}
+	defer watcher.Stop()
+
+	var keys []string
+	for entry := range watcher.Updates() {
+		if entry == nil {
+			break
+		}
+		keys = append(keys, entry.Key())
+	}
+	if len(keys) == 0 {
+		return nil, ErrNoKeysFound
+	}
+	return keys, nil
+}
+
+// History will return all historical values for the key.
+func (kv *kvs) History(ctx context.Context, key string, opts ...WatchOpt) ([]KeyValueEntry, error) {
+	opts = append(opts, IncludeHistory())
+	watcher, err := kv.Watch(ctx, key, opts...)
+	if err != nil {
+		return nil, err
+	}
+	defer watcher.Stop()
+
+	var entries []KeyValueEntry
+	for entry := range watcher.Updates() {
+		if entry == nil {
+			break
+		}
+		entries = append(entries, entry)
+	}
+	if len(entries) == 0 {
+		return nil, ErrKeyNotFound
+	}
+	return entries, nil
+}
+
+// Bucket returns the current bucket name.
+func (kv *kvs) Bucket() string {
+	return kv.name
+}
+
+const kvDefaultPurgeDeletesMarkerThreshold = 30 * time.Minute
+
+// PurgeDeletes will remove all current delete markers.
+func (kv *kvs) PurgeDeletes(ctx context.Context, opts ...KVPurgeOpt) error {
+	var o purgeOpts
+	for _, opt := range opts {
+		if opt != nil {
+			if err := opt.configurePurge(&o); err != nil {
+				return err
+			}
+		}
+	}
+	watcher, err := kv.WatchAll(ctx)
+	if err != nil {
+		return err
+	}
+	defer watcher.Stop()
+
+	var limit time.Time
+	olderThan := o.dmthr
+	// Negative value is used to instruct to always remove markers, regardless
+	// of age. If set to 0 (or not set), use our default value.
+	if olderThan == 0 {
+		olderThan = kvDefaultPurgeDeletesMarkerThreshold
+	}
+	if olderThan > 0 {
+		limit = time.Now().Add(-olderThan)
+	}
+
+	var deleteMarkers []KeyValueEntry
+	for entry := range watcher.Updates() {
+		if entry == nil {
+			break
+		}
+		if op := entry.Operation(); op == KeyValueDelete || op == KeyValuePurge {
+			deleteMarkers = append(deleteMarkers, entry)
+		}
+	}
+
+	var b strings.Builder
+	// Do actual purges here.
+	for _, entry := range deleteMarkers {
+		b.WriteString(kv.pre)
+		b.WriteString(entry.Key())
+		purgeOpts := []StreamPurgeOpt{WithPurgeSubject(b.String())}
+		if olderThan > 0 && entry.Created().After(limit) {
+			purgeOpts = append(purgeOpts, WithPurgeKeep(1))
+		}
+		if err := kv.stream.Purge(ctx, purgeOpts...); err != nil {
+			return err
+		}
+		b.Reset()
+	}
+	return nil
+}
+
+// Status retrieves the status and configuration of a bucket
+func (kv *kvs) Status(ctx context.Context) (KeyValueStatus, error) {
+	nfo, err := kv.stream.Info(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &KeyValueBucketStatus{nfo: nfo, bucket: kv.name}, nil
+}
+
+func mapStreamToKVS(js *jetStream, pushJS nats.JetStreamContext, stream Stream) *kvs {
+	info := stream.CachedInfo()
+	bucket := strings.TrimPrefix(info.Config.Name, kvBucketNamePre)
+	kv := &kvs{
+		name:       bucket,
+		streamName: info.Config.Name,
+		pre:        fmt.Sprintf(kvSubjectsPreTmpl, bucket),
+		js:         js,
+		pushJS:     pushJS,
+		stream:     stream,
+		// Determine if we need to use the JS prefix in front of Put and Delete operations
+		useJSPfx:  js.apiPrefix != DefaultAPIPrefix,
+		useDirect: info.Config.AllowDirect,
+	}
+
+	// If we are mirroring, we will have mirror direct on, so just use the mirror name
+	// and override use
+	if m := info.Config.Mirror; m != nil {
+		bucket := strings.TrimPrefix(m.Name, kvBucketNamePre)
+		if m.External != nil && m.External.APIPrefix != "" {
+			kv.useJSPfx = false
+			kv.pre = fmt.Sprintf(kvSubjectsPreTmpl, bucket)
+			kv.putPre = fmt.Sprintf(kvSubjectsPreDomainTmpl, m.External.APIPrefix, bucket)
+		} else {
+			kv.putPre = fmt.Sprintf(kvSubjectsPreTmpl, bucket)
+		}
+	}
+
+	return kv
+}
diff --git a/jetstream/options.go b/jetstream/options.go
index ce79fc5..e1b4b5b 100644
--- a/jetstream/options.go
+++ b/jetstream/options.go
@@ -316,3 +316,76 @@ func WithStallWait(ttl time.Duration) PublishOpt {
 		return nil
 	}
 }
+
+// KV Options
+
+type watchOptFn func(opts *watchOpts) error
+
+func (opt watchOptFn) configureWatcher(opts *watchOpts) error {
+	return opt(opts)
+}
+
+// IncludeHistory instructs the key watcher to include historical values as well.
+func IncludeHistory() WatchOpt {
+	return watchOptFn(func(opts *watchOpts) error {
+		if opts.updatesOnly {
+			return fmt.Errorf("%w: include history can not be used with updates only", ErrInvalidOption)
+		}
+		opts.includeHistory = true
+		return nil
+	})
+}
+
+// UpdatesOnly instructs the key watcher to only include updates on values (without latest values when started).
+func UpdatesOnly() WatchOpt {
+	return watchOptFn(func(opts *watchOpts) error {
+		if opts.includeHistory {
+			return fmt.Errorf("%w: updates only can not be used with include history", ErrInvalidOption)
+		}
+		opts.updatesOnly = true
+		return nil
+	})
+}
+
+// IgnoreDeletes will have the key watcher not pass any deleted keys.
+func IgnoreDeletes() WatchOpt {
+	return watchOptFn(func(opts *watchOpts) error {
+		opts.ignoreDeletes = true
+		return nil
+	})
+}
+
+// MetaOnly instructs the key watcher to retrieve only the entry meta data, not the entry value.
+func MetaOnly() WatchOpt {
+	return watchOptFn(func(opts *watchOpts) error {
+		opts.metaOnly = true
+		return nil
+	})
+}
+
+// DeleteMarkersOlderThan indicates that delete or purge markers older than that
+// will be deleted as part of PurgeDeletes() operation, otherwise, only the data
+// will be removed but markers that are recent will be kept.
+// Note that if no option is specified, the default is 30 minutes. You can set
+// this option to a negative value to instruct to always remove the markers,
+// regardless of their age.
+type DeleteMarkersOlderThan time.Duration
+
+func (ttl DeleteMarkersOlderThan) configurePurge(opts *purgeOpts) error {
+	opts.dmthr = time.Duration(ttl)
+	return nil
+}
+
+type deleteOptFn func(opts *deleteOpts) error
+
+func (opt deleteOptFn) configureDelete(opts *deleteOpts) error {
+	return opt(opts)
+}
+
+// LastRevision deletes if the latest revision matches.
+func LastRevision(revision uint64) KVDeleteOpt {
+	return deleteOptFn(func(opts *deleteOpts) error {
+		opts.revision = revision
+		return 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/jetstream/publish.go b/jetstream/publish.go
index 1215687..30371cd 100644
--- a/jetstream/publish.go
+++ b/jetstream/publish.go
@@ -402,6 +402,10 @@ func (js *jetStream) resetPendingAcksOnReconnect() {
 			paf.err = nats.ErrDisconnected
 		}
 		js.publisher.acks = nil
+		if js.publisher.doneCh != nil {
+			close(js.publisher.doneCh)
+			js.publisher.doneCh = nil
+		}
 		js.publisher.Unlock()
 	}
 }
diff --git a/js.go b/js.go
index bd8cb8b..a8deb98 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 {
@@ -698,6 +698,10 @@ func (js *js) resetPendingAcksOnReconnect() {
 			paf.err = ErrDisconnected
 		}
 		js.pafs = nil
+		if js.dch != nil {
+			close(js.dch)
+			js.dch = nil
+		}
 		js.mu.Unlock()
 	}
 }
@@ -1822,6 +1826,17 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 	return sub, nil
 }
 
+// InitialConsumerPending returns the number of messages pending to be
+// delivered to the consumer when the subscription was created.
+func (sub *Subscription) InitialConsumerPending() (uint64, error) {
+	sub.mu.Lock()
+	defer sub.mu.Unlock()
+	if sub.jsi == nil || sub.jsi.consumer == _EMPTY_ {
+		return 0, fmt.Errorf("%w: not a JetStream subscription", ErrTypeSubscription)
+	}
+	return sub.jsi.pending, nil
+}
+
 // This long-lived routine is used per ChanSubscription to check
 // on the number of delivered messages and check for flow control response.
 func (sub *Subscription) chanSubcheckForFlowControlResponse() {
@@ -1915,7 +1930,7 @@ func (sub *Subscription) checkOrderedMsgs(m *Msg) bool {
 	if err != nil {
 		return false
 	}
-	sseq, dseq := parser.ParseNum(tokens[ackStreamSeqTokenPos]), parser.ParseNum(tokens[ackConsumerSeqTokenPos])
+	sseq, dseq := parser.ParseNum(tokens[parser.AckStreamSeqTokenPos]), parser.ParseNum(tokens[parser.AckConsumerSeqTokenPos])
 
 	jsi := sub.jsi
 	if dseq != jsi.dseq {
@@ -2157,7 +2172,7 @@ func (nc *Conn) checkForSequenceMismatch(msg *Msg, s *Subscription, jsi *jsSub)
 
 	// Consumer sequence.
 	var ldseq string
-	dseq := tokens[ackConsumerSeqTokenPos]
+	dseq := tokens[parser.AckConsumerSeqTokenPos]
 	hdr := msg.Header[lastConsumerSeqHdr]
 	if len(hdr) == 1 {
 		ldseq = hdr[0]
@@ -2168,7 +2183,7 @@ func (nc *Conn) checkForSequenceMismatch(msg *Msg, s *Subscription, jsi *jsSub)
 	if ldseq != dseq {
 		// Dispatch async error including details such as
 		// from where the consumer could be restarted.
-		sseq := parser.ParseNum(tokens[ackStreamSeqTokenPos])
+		sseq := parser.ParseNum(tokens[parser.AckStreamSeqTokenPos])
 		if ordered {
 			s.mu.Lock()
 			s.resetOrderedConsumer(jsi.sseq + 1)
@@ -3280,18 +3295,6 @@ type MsgMetadata struct {
 	Domain       string
 }
 
-const (
-	ackDomainTokenPos       = 2
-	ackAccHashTokenPos      = 3
-	ackStreamTokenPos       = 4
-	ackConsumerTokenPos     = 5
-	ackNumDeliveredTokenPos = 6
-	ackStreamSeqTokenPos    = 7
-	ackConsumerSeqTokenPos  = 8
-	ackTimestampSeqTokenPos = 9
-	ackNumPendingTokenPos   = 10
-)
-
 // Metadata retrieves the metadata from a JetStream message. This method will
 // return an error for non-JetStream Msgs.
 func (m *Msg) Metadata() (*MsgMetadata, error) {
@@ -3305,15 +3308,15 @@ func (m *Msg) Metadata() (*MsgMetadata, error) {
 	}
 
 	meta := &MsgMetadata{
-		Domain:       tokens[ackDomainTokenPos],
-		NumDelivered: parser.ParseNum(tokens[ackNumDeliveredTokenPos]),
-		NumPending:   parser.ParseNum(tokens[ackNumPendingTokenPos]),
-		Timestamp:    time.Unix(0, int64(parser.ParseNum(tokens[ackTimestampSeqTokenPos]))),
-		Stream:       tokens[ackStreamTokenPos],
-		Consumer:     tokens[ackConsumerTokenPos],
-	}
-	meta.Sequence.Stream = parser.ParseNum(tokens[ackStreamSeqTokenPos])
-	meta.Sequence.Consumer = parser.ParseNum(tokens[ackConsumerSeqTokenPos])
+		Domain:       tokens[parser.AckDomainTokenPos],
+		NumDelivered: parser.ParseNum(tokens[parser.AckNumDeliveredTokenPos]),
+		NumPending:   parser.ParseNum(tokens[parser.AckNumPendingTokenPos]),
+		Timestamp:    time.Unix(0, int64(parser.ParseNum(tokens[parser.AckTimestampSeqTokenPos]))),
+		Stream:       tokens[parser.AckStreamTokenPos],
+		Consumer:     tokens[parser.AckConsumerTokenPos],
+	}
+	meta.Sequence.Stream = parser.ParseNum(tokens[parser.AckStreamSeqTokenPos])
+	meta.Sequence.Consumer = parser.ParseNum(tokens[parser.AckConsumerSeqTokenPos])
 	return meta, nil
 }
 
@@ -3363,7 +3366,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 9ad1b22..17c9e2f 100644
--- a/kv.go
+++ b/kv.go
@@ -123,6 +123,8 @@ type watchOpts struct {
 	ignoreDeletes bool
 	// Include all history per subject, not just last one.
 	includeHistory bool
+	// Include only updates for keys.
+	updatesOnly bool
 	// retrieve only the meta data of the entry
 	metaOnly bool
 }
@@ -136,11 +138,25 @@ func (opt watchOptFn) configureWatcher(opts *watchOpts) error {
 // IncludeHistory instructs the key watcher to include historical values as well.
 func IncludeHistory() WatchOpt {
 	return watchOptFn(func(opts *watchOpts) error {
+		if opts.updatesOnly {
+			return errors.New("nats: include history can not be used with updates only")
+		}
 		opts.includeHistory = true
 		return nil
 	})
 }
 
+// UpdatesOnly instructs the key watcher to only include updates on values (without latest values when started).
+func UpdatesOnly() WatchOpt {
+	return watchOptFn(func(opts *watchOpts) error {
+		if opts.includeHistory {
+			return errors.New("nats: updates only can not be used with include history")
+		}
+		opts.updatesOnly = true
+		return nil
+	})
+}
+
 // IgnoreDeletes will have the key watcher not pass any deleted keys.
 func IgnoreDeletes() WatchOpt {
 	return watchOptFn(func(opts *watchOpts) error {
@@ -622,7 +638,7 @@ func (kv *kvs) PutString(key string, value string) (revision uint64, err error)
 	return kv.Put(key, []byte(value))
 }
 
-// Create will add the key/value pair iff it does not exist.
+// Create will add the key/value pair if it does not exist.
 func (kv *kvs) Create(key string, value []byte) (revision uint64, err error) {
 	v, err := kv.Update(key, value, 0)
 	if err == nil {
@@ -645,7 +661,7 @@ func (kv *kvs) Create(key string, value []byte) (revision uint64, err error) {
 	return 0, err
 }
 
-// Update will update the value iff the latest revision matches.
+// Update will update the value if the latest revision matches.
 func (kv *kvs) Update(key string, value []byte, revision uint64) (uint64, error) {
 	if !keyValid(key) {
 		return 0, ErrInvalidKey
@@ -909,7 +925,7 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 				op = KeyValuePurge
 			}
 		}
-		delta := parser.ParseNum(tokens[ackNumPendingTokenPos])
+		delta := parser.ParseNum(tokens[parser.AckNumPendingTokenPos])
 		w.mu.Lock()
 		defer w.mu.Unlock()
 		if !o.ignoreDeletes || (op != KeyValueDelete && op != KeyValuePurge) {
@@ -917,14 +933,15 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 				bucket:   kv.name,
 				key:      subj,
 				value:    m.Data,
-				revision: parser.ParseNum(tokens[ackStreamSeqTokenPos]),
-				created:  time.Unix(0, int64(parser.ParseNum(tokens[ackTimestampSeqTokenPos]))),
+				revision: parser.ParseNum(tokens[parser.AckStreamSeqTokenPos]),
+				created:  time.Unix(0, int64(parser.ParseNum(tokens[parser.AckTimestampSeqTokenPos]))),
 				delta:    delta,
 				op:       op,
 			}
 			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..
@@ -943,6 +960,9 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 	if !o.includeHistory {
 		subOpts = append(subOpts, DeliverLastPerSubject())
 	}
+	if o.updatesOnly {
+		subOpts = append(subOpts, DeliverNew())
+	}
 	if o.metaOnly {
 		subOpts = append(subOpts, HeadersOnly())
 	}
@@ -961,12 +981,18 @@ 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() {
+	sub.pDone = func(_ string) {
 		close(w.updates)
 	}
 	sub.mu.Unlock()
@@ -1020,16 +1046,16 @@ func (kv *kvs) Status() (KeyValueStatus, error) {
 // KeyValueStoreNames is used to retrieve a list of key value store names
 func (js *js) KeyValueStoreNames() <-chan string {
 	ch := make(chan string)
-	l := &streamLister{js: js}
+	l := &streamNamesLister{js: js}
 	l.js.opts.streamListSubject = fmt.Sprintf(kvSubjectsTmpl, "*")
 	go func() {
 		defer close(ch)
 		for l.Next() {
-			for _, info := range l.Page() {
-				if !strings.HasPrefix(info.Config.Name, kvBucketNamePre) {
+			for _, name := range l.Page() {
+				if !strings.HasPrefix(name, kvBucketNamePre) {
 					continue
 				}
-				ch <- info.Config.Name
+				ch <- name
 			}
 		}
 	}()
diff --git a/micro/service.go b/micro/service.go
index cee4ef3..5d2dd36 100644
--- a/micro/service.go
+++ b/micro/service.go
@@ -474,6 +474,10 @@ func (s *service) wrapConnectionEventCallbacks() {
 	s.natsHandlers.asyncErr = s.nc.ErrorHandler()
 	if s.natsHandlers.asyncErr != nil {
 		s.nc.SetErrorHandler(func(c *nats.Conn, sub *nats.Subscription, err error) {
+			if sub == nil {
+				s.natsHandlers.asyncErr(c, sub, err)
+				return
+			}
 			endpoint, match := s.matchSubscriptionSubject(sub.Subject)
 			if !match {
 				s.natsHandlers.asyncErr(c, sub, err)
@@ -496,6 +500,9 @@ func (s *service) wrapConnectionEventCallbacks() {
 		})
 	} else {
 		s.nc.SetErrorHandler(func(c *nats.Conn, sub *nats.Subscription, err error) {
+			if sub == nil {
+				return
+			}
 			endpoint, match := s.matchSubscriptionSubject(sub.Subject)
 			if !match {
 				return
diff --git a/nats.go b/nats.go
index 350f230..ba262b4 100644
--- a/nats.go
+++ b/nats.go
@@ -613,7 +613,7 @@ type Subscription struct {
 	pHead *Msg
 	pTail *Msg
 	pCond *sync.Cond
-	pDone func()
+	pDone func(subject string)
 
 	// Pending stats, async subscriptions, high-speed etc.
 	pMsgs       int
@@ -946,6 +946,7 @@ func ReconnectWait(t time.Duration) Option {
 }
 
 // MaxReconnects is an Option to set the maximum number of reconnect attempts.
+// If negative, it will never stop trying to reconnect.
 // Defaults to 60.
 func MaxReconnects(max int) Option {
 	return func(o *Options) error {
@@ -2500,6 +2501,9 @@ func (nc *Conn) sendConnect() error {
 	// Construct the CONNECT protocol string
 	cProto, err := nc.connectProto()
 	if err != nil {
+		if !nc.initc && nc.Opts.AsyncErrorCB != nil {
+			nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, nil, err) })
+		}
 		return err
 	}
 
@@ -3025,7 +3029,7 @@ func (nc *Conn) waitForMsgs(s *Subscription) {
 	s.mu.Unlock()
 
 	if done != nil {
-		done()
+		done(s.Subject)
 	}
 }
 
@@ -4450,6 +4454,14 @@ func (s *Subscription) AutoUnsubscribe(max int) error {
 	return conn.unsubscribe(s, max, false)
 }
 
+// SetClosedHandler will set the closed handler for when a subscription
+// is closed (either unsubscribed or drained).
+func (s *Subscription) SetClosedHandler(handler func(subject string)) {
+	s.mu.Lock()
+	s.pDone = handler
+	s.mu.Unlock()
+}
+
 // unsubscribe performs the low level unsubscribe to the server.
 // Use Subscription.Unsubscribe()
 func (nc *Conn) unsubscribe(sub *Subscription, max int, drainMode bool) error {
diff --git a/object.go b/object.go
index 9d558cc..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 {
@@ -658,7 +658,7 @@ func (obs *obs) Get(name string, opts ...GetObjectOpt) (ObjectResult, error) {
 		result.digest.Write(m.Data)
 
 		// Check if we are done.
-		if tokens[ackNumPendingTokenPos] == objNoPending {
+		if tokens[parser.AckNumPendingTokenPos] == objNoPending {
 			pw.Close()
 			m.Sub.Unsubscribe()
 		}
@@ -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
new file mode 100644
index 0000000..788a318
--- /dev/null
+++ b/jetstream/test/kv_test.go
@@ -0,0 +1,1358 @@
+// Copyright 2023 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"reflect"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+func TestKeyValueBasics(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx := context.Background()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "TEST", History: 5, TTL: time.Hour})
+	expectOk(t, err)
+
+	if kv.Bucket() != "TEST" {
+		t.Fatalf("Expected bucket name to be %q, got %q", "TEST", kv.Bucket())
+	}
+
+	// Simple Put
+	r, err := kv.Put(ctx, "name", []byte("derek"))
+	expectOk(t, err)
+	if r != 1 {
+		t.Fatalf("Expected 1 for the revision, got %d", r)
+	}
+	// Simple Get
+	e, err := kv.Get(ctx, "name")
+	expectOk(t, err)
+	if string(e.Value()) != "derek" {
+		t.Fatalf("Got wrong value: %q vs %q", e.Value(), "derek")
+	}
+	if e.Revision() != 1 {
+		t.Fatalf("Expected 1 for the revision, got %d", e.Revision())
+	}
+
+	// Delete
+	err = kv.Delete(ctx, "name")
+	expectOk(t, err)
+	_, err = kv.Get(ctx, "name")
+	expectErr(t, err, jetstream.ErrKeyNotFound)
+	r, err = kv.Create(ctx, "name", []byte("derek"))
+	expectOk(t, err)
+	if r != 3 {
+		t.Fatalf("Expected 3 for the revision, got %d", r)
+	}
+	err = kv.Delete(ctx, "name", jetstream.LastRevision(4))
+	expectErr(t, err)
+	err = kv.Delete(ctx, "name", jetstream.LastRevision(3))
+	expectOk(t, err)
+
+	// Conditional Updates.
+	r, err = kv.Update(ctx, "name", []byte("rip"), 4)
+	expectOk(t, err)
+	_, err = kv.Update(ctx, "name", []byte("ik"), 3)
+	expectErr(t, err)
+	_, err = kv.Update(ctx, "name", []byte("ik"), r)
+	expectOk(t, err)
+	r, err = kv.Create(ctx, "age", []byte("22"))
+	expectOk(t, err)
+	_, err = kv.Update(ctx, "age", []byte("33"), r)
+	expectOk(t, err)
+
+	// Status
+	status, err := kv.Status(ctx)
+	expectOk(t, err)
+	if status.History() != 5 {
+		t.Fatalf("expected history of 5 got %d", status.History())
+	}
+	if status.Bucket() != "TEST" {
+		t.Fatalf("expected bucket TEST got %v", status.Bucket())
+	}
+	if status.TTL() != time.Hour {
+		t.Fatalf("expected 1 hour TTL got %v", status.TTL())
+	}
+	if status.Values() != 7 {
+		t.Fatalf("expected 7 values got %d", status.Values())
+	}
+	if status.BackingStore() != "JetStream" {
+		t.Fatalf("invalid backing store kind %s", status.BackingStore())
+	}
+
+	kvs := status.(*jetstream.KeyValueBucketStatus)
+	si := kvs.StreamInfo()
+	if si == nil {
+		t.Fatalf("StreamInfo not received")
+	}
+}
+
+func TestKeyValueHistory(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "LIST", History: 10})
+	expectOk(t, err)
+
+	for i := 0; i < 50; i++ {
+		age := strconv.FormatUint(uint64(i+22), 10)
+		_, err := kv.Put(ctx, "age", []byte(age))
+		expectOk(t, err)
+	}
+
+	vl, err := kv.History(ctx, "age")
+	expectOk(t, err)
+
+	if len(vl) != 10 {
+		t.Fatalf("Expected %d values, got %d", 10, len(vl))
+	}
+	for i, v := range vl {
+		if v.Key() != "age" {
+			t.Fatalf("Expected key of %q, got %q", "age", v.Key())
+		}
+		if v.Revision() != uint64(i+41) {
+			// History of 10, sent 50..
+			t.Fatalf("Expected revision of %d, got %d", i+41, v.Revision())
+		}
+		age, err := strconv.Atoi(string(v.Value()))
+		expectOk(t, err)
+		if age != i+62 {
+			t.Fatalf("Expected data value of %d, got %d", i+22, age)
+		}
+	}
+}
+
+func TestKeyValueWatch(t *testing.T) {
+	expectUpdateF := func(t *testing.T, watcher jetstream.KeyWatcher) func(key, value string, revision uint64) {
+		return func(key, value string, revision uint64) {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v.Key() != key || string(v.Value()) != value || v.Revision() != revision {
+					t.Fatalf("Did not get expected: %q %q %d vs %q %q %d", v.Key(), string(v.Value()), v.Revision(), key, value, revision)
+				}
+			case <-time.After(time.Second):
+				t.Fatalf("Did not receive an update like expected")
+			}
+		}
+	}
+	expectDeleteF := func(t *testing.T, watcher jetstream.KeyWatcher) func(key string, revision uint64) {
+		return func(key string, revision uint64) {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v.Operation() != jetstream.KeyValueDelete {
+					t.Fatalf("Expected a delete operation but got %+v", v)
+				}
+				if v.Revision() != revision {
+					t.Fatalf("Did not get expected revision: %d vs %d", revision, v.Revision())
+				}
+			case <-time.After(time.Second):
+				t.Fatalf("Did not receive an update like expected")
+			}
+		}
+	}
+	expectInitDoneF := func(t *testing.T, watcher jetstream.KeyWatcher) func() {
+		return func() {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v != nil {
+					t.Fatalf("Did not get expected: %+v", v)
+				}
+			case <-time.After(time.Second):
+				t.Fatalf("Did not receive a init done like expected")
+			}
+		}
+	}
+
+	t.Run("default watcher", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH"})
+		expectOk(t, err)
+
+		watcher, err := kv.WatchAll(ctx)
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectInitDone := expectInitDoneF(t, watcher)
+		expectUpdate := expectUpdateF(t, watcher)
+		expectDelete := expectDeleteF(t, watcher)
+		// Make sure we already got an initial value marker.
+		expectInitDone()
+
+		_, err = kv.Create(ctx, "name", []byte("derek"))
+		expectOk(t, err)
+		expectUpdate("name", "derek", 1)
+		_, err = kv.Put(ctx, "name", []byte("rip"))
+		expectOk(t, err)
+		expectUpdate("name", "rip", 2)
+		_, err = kv.Put(ctx, "name", []byte("ik"))
+		expectOk(t, err)
+		expectUpdate("name", "ik", 3)
+		_, err = kv.Put(ctx, "age", []byte("22"))
+		expectOk(t, err)
+		expectUpdate("age", "22", 4)
+		_, err = kv.Put(ctx, "age", []byte("33"))
+		expectOk(t, err)
+		expectUpdate("age", "33", 5)
+		expectOk(t, kv.Delete(ctx, "age"))
+		expectDelete("age", 6)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		// Now try wildcard matching and make sure we only get last value when starting.
+		_, err = kv.Put(ctx, "t.name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.name", []byte("ik"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("22"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("44"))
+		expectOk(t, err)
+
+		watcher, err = kv.Watch(ctx, "t.*")
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectInitDone = expectInitDoneF(t, watcher)
+		expectUpdate = expectUpdateF(t, watcher)
+		expectUpdate("t.name", "ik", 8)
+		expectUpdate("t.age", "44", 10)
+		expectInitDone()
+	})
+
+	t.Run("watcher with history included", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH", History: 64})
+		expectOk(t, err)
+
+		_, err = kv.Create(ctx, "name", []byte("derek"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "name", []byte("ik"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "age", []byte("22"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "age", []byte("33"))
+		expectOk(t, err)
+		expectOk(t, kv.Delete(ctx, "age"))
+
+		// when using IncludeHistory(), UpdatesOnly() is not allowed
+		if _, err := kv.WatchAll(ctx, jetstream.IncludeHistory(), jetstream.UpdatesOnly()); !errors.Is(err, jetstream.ErrInvalidOption) {
+			t.Fatalf("Expected %v, got %v", jetstream.ErrInvalidOption, err)
+		}
+
+		watcher, err := kv.WatchAll(ctx, jetstream.IncludeHistory())
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectInitDone := expectInitDoneF(t, watcher)
+		expectUpdate := expectUpdateF(t, watcher)
+		expectDelete := expectDeleteF(t, watcher)
+		expectUpdate("name", "derek", 1)
+		expectUpdate("name", "rip", 2)
+		expectUpdate("name", "ik", 3)
+		expectUpdate("age", "22", 4)
+		expectUpdate("age", "33", 5)
+		expectDelete("age", 6)
+		expectInitDone()
+		_, err = kv.Put(ctx, "name", []byte("pp"))
+		expectOk(t, err)
+		expectUpdate("name", "pp", 7)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		_, err = kv.Put(ctx, "t.name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.name", []byte("ik"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("22"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("44"))
+		expectOk(t, err)
+
+		// try wildcard watcher and make sure we get all historical values
+		watcher, err = kv.Watch(ctx, "t.*", jetstream.IncludeHistory())
+		expectOk(t, err)
+		defer watcher.Stop()
+		expectInitDone = expectInitDoneF(t, watcher)
+		expectUpdate = expectUpdateF(t, watcher)
+
+		expectUpdate("t.name", "rip", 8)
+		expectUpdate("t.name", "ik", 9)
+		expectUpdate("t.age", "22", 10)
+		expectUpdate("t.age", "44", 11)
+		expectInitDone()
+
+		_, err = kv.Put(ctx, "t.name", []byte("pp"))
+		expectOk(t, err)
+		expectUpdate("t.name", "pp", 12)
+	})
+
+	t.Run("watcher with updates only", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH", History: 64})
+		expectOk(t, err)
+
+		_, err = kv.Create(ctx, "name", []byte("derek"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "age", []byte("22"))
+		expectOk(t, err)
+
+		// when using UpdatesOnly(), IncludeHistory() is not allowed
+		if _, err := kv.WatchAll(ctx, jetstream.UpdatesOnly(), jetstream.IncludeHistory()); !errors.Is(err, jetstream.ErrInvalidOption) {
+			t.Fatalf("Expected %v, got %v", jetstream.ErrInvalidOption, err)
+		}
+
+		watcher, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
+		expectOk(t, err)
+		defer watcher.Stop()
+		expectUpdate := expectUpdateF(t, watcher)
+		expectDelete := expectDeleteF(t, watcher)
+
+		// now update some keys and expect updates
+		_, err = kv.Put(ctx, "name", []byte("pp"))
+		expectOk(t, err)
+		expectUpdate("name", "pp", 4)
+		_, err = kv.Put(ctx, "age", []byte("44"))
+		expectOk(t, err)
+		expectUpdate("age", "44", 5)
+		expectOk(t, kv.Delete(ctx, "age"))
+		expectDelete("age", 6)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		_, err = kv.Put(ctx, "t.name", []byte("rip"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.name", []byte("ik"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("22"))
+		expectOk(t, err)
+		_, err = kv.Put(ctx, "t.age", []byte("44"))
+		expectOk(t, err)
+
+		// try wildcard watcher and make sure we do not get any values initially
+		watcher, err = kv.Watch(ctx, "t.*", jetstream.UpdatesOnly())
+		expectOk(t, err)
+		defer watcher.Stop()
+		expectUpdate = expectUpdateF(t, watcher)
+
+		// update some keys and expect updates
+		_, err = kv.Put(ctx, "t.name", []byte("pp"))
+		expectOk(t, err)
+		expectUpdate("t.name", "pp", 11)
+		_, err = kv.Put(ctx, "t.age", []byte("66"))
+		expectOk(t, err)
+		expectUpdate("t.age", "66", 12)
+	})
+}
+
+func TestKeyValueWatchContext(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCHCTX"})
+	expectOk(t, err)
+
+	watcher, err := kv.WatchAll(ctx)
+	expectOk(t, err)
+	defer watcher.Stop()
+
+	// Trigger unsubscribe internally.
+	cancel()
+
+	// Wait for a bit for unsubscribe to be done.
+	time.Sleep(500 * time.Millisecond)
+
+	// Stopping watch that is already stopped via cancellation propagation is an error.
+	err = watcher.Stop()
+	if err == nil || err != nats.ErrBadSubscription {
+		t.Errorf("Expected invalid subscription, got: %v", err)
+	}
+}
+
+func TestKeyValueWatchContextUpdates(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCHCTX"})
+	expectOk(t, err)
+
+	watcher, err := kv.WatchAll(ctx)
+	expectOk(t, err)
+	defer watcher.Stop()
+
+	// Pull the initial state done marker which is nil.
+	select {
+	case v := <-watcher.Updates():
+		if v != nil {
+			t.Fatalf("Expected nil marker, got %+v", v)
+		}
+	case <-time.After(time.Second):
+		t.Fatalf("Did not receive nil marker like expected")
+	}
+
+	// Fire a timer and cancel the context after 250ms.
+	time.AfterFunc(250*time.Millisecond, cancel)
+
+	// Make sure canceling will break us out here.
+	select {
+	case <-watcher.Updates():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not break out like expected")
+	}
+}
+
+func TestKeyValueBindStore(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	_, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH"})
+	expectOk(t, err)
+
+	// Now bind to it..
+	_, err = js.KeyValue(ctx, "WATCH")
+	expectOk(t, err)
+
+	// Make sure we can't bind to a non-kv style stream.
+	// We have some protection with stream name prefix.
+	_, err = js.CreateStream(ctx, jetstream.StreamConfig{
+		Name:     "KV_TEST",
+		Subjects: []string{"foo"},
+	})
+	expectOk(t, err)
+
+	_, err = js.KeyValue(ctx, "TEST")
+	expectErr(t, err)
+	if err != jetstream.ErrBadBucket {
+		t.Fatalf("Expected %v but got %v", jetstream.ErrBadBucket, err)
+	}
+}
+
+func TestKeyValueDeleteStore(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	_, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "WATCH"})
+	expectOk(t, err)
+
+	err = js.DeleteKeyValue(ctx, "WATCH")
+	expectOk(t, err)
+
+	_, err = js.KeyValue(ctx, "WATCH")
+	expectErr(t, err)
+}
+
+func TestKeyValueDeleteVsPurge(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", History: 10})
+	expectOk(t, err)
+
+	put := func(key, value string) {
+		t.Helper()
+		_, err := kv.Put(ctx, key, []byte(value))
+		expectOk(t, err)
+	}
+
+	// Put in a few names and ages.
+	put("name", "derek")
+	put("age", "22")
+	put("name", "ivan")
+	put("age", "33")
+	put("name", "rip")
+	put("age", "44")
+
+	expectOk(t, kv.Delete(ctx, "age"))
+	entries, err := kv.History(ctx, "age")
+	expectOk(t, err)
+	// Expect three entries and delete marker.
+	if len(entries) != 4 {
+		t.Fatalf("Expected 4 entries for age after delete, got %d", len(entries))
+	}
+	err = kv.Purge(ctx, "name", jetstream.LastRevision(4))
+	expectErr(t, err)
+	err = kv.Purge(ctx, "name", jetstream.LastRevision(5))
+	expectOk(t, err)
+	// Check marker
+	e, err := kv.Get(ctx, "name")
+	expectErr(t, err, jetstream.ErrKeyNotFound)
+	if e != nil {
+		t.Fatalf("Expected a nil entry but got %v", e)
+	}
+	entries, err = kv.History(ctx, "name")
+	expectOk(t, err)
+	if len(entries) != 1 {
+		t.Fatalf("Expected only 1 entry for age after delete, got %d", len(entries))
+	}
+	// Make sure history also reports the purge operation.
+	if e := entries[0]; e.Operation() != jetstream.KeyValuePurge {
+		t.Fatalf("Expected a purge operation but got %v", e.Operation())
+	}
+}
+
+func TestKeyValueDeleteTombstones(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", History: 10})
+	expectOk(t, err)
+
+	put := func(key, value string) {
+		t.Helper()
+		_, err := kv.Put(ctx, key, []byte(value))
+		expectOk(t, err)
+	}
+
+	v := strings.Repeat("ABC", 33)
+	for i := 1; i <= 100; i++ {
+		put(fmt.Sprintf("key-%d", i), v)
+	}
+	// Now delete them.
+	for i := 1; i <= 100; i++ {
+		err := kv.Delete(ctx, fmt.Sprintf("key-%d", i))
+		expectOk(t, err)
+	}
+
+	// Now cleanup.
+	err = kv.PurgeDeletes(ctx, jetstream.DeleteMarkersOlderThan(-1))
+	expectOk(t, err)
+
+	si, err := js.Stream(ctx, "KV_KVS")
+	expectOk(t, err)
+	if si.CachedInfo().State.Msgs != 0 {
+		t.Fatalf("Expected no stream msgs to be left, got %d", si.CachedInfo().State.Msgs)
+	}
+
+	// Try with context
+	ctx, cancel = context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+	err = kv.PurgeDeletes(nats.Context(ctx))
+	expectOk(t, err)
+}
+
+func TestKeyValuePurgeDeletesMarkerThreshold(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", History: 10})
+	expectOk(t, err)
+
+	put := func(key, value string) {
+		t.Helper()
+		_, err := kv.Put(ctx, key, []byte(value))
+		expectOk(t, err)
+	}
+
+	put("foo", "foo1")
+	put("bar", "bar1")
+	put("foo", "foo2")
+	err = kv.Delete(ctx, "foo")
+	expectOk(t, err)
+
+	time.Sleep(200 * time.Millisecond)
+
+	err = kv.Delete(ctx, "bar")
+	expectOk(t, err)
+
+	err = kv.PurgeDeletes(ctx, jetstream.DeleteMarkersOlderThan(100*time.Millisecond))
+	expectOk(t, err)
+
+	// The key foo should have been completely cleared of the data
+	// and the delete marker.
+	fooEntries, err := kv.History(ctx, "foo")
+	if err != jetstream.ErrKeyNotFound {
+		t.Fatalf("Expected all entries for key foo to be gone, got err=%v entries=%v", err, fooEntries)
+	}
+	barEntries, err := kv.History(ctx, "bar")
+	expectOk(t, err)
+	if len(barEntries) != 1 {
+		t.Fatalf("Expected 1 entry, got %v", barEntries)
+	}
+	if e := barEntries[0]; e.Operation() != jetstream.KeyValueDelete {
+		t.Fatalf("Unexpected entry: %+v", e)
+	}
+}
+
+func TestKeyValueKeys(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", History: 2})
+	expectOk(t, err)
+
+	put := func(key, value string) {
+		t.Helper()
+		_, err := kv.Put(ctx, key, []byte(value))
+		expectOk(t, err)
+	}
+
+	_, err = kv.Keys(ctx)
+	expectErr(t, err, jetstream.ErrNoKeysFound)
+
+	// Put in a few names and ages.
+	put("name", "derek")
+	put("age", "22")
+	put("country", "US")
+	put("name", "ivan")
+	put("age", "33")
+	put("country", "US")
+	put("name", "rip")
+	put("age", "44")
+	put("country", "MT")
+
+	keys, err := kv.Keys(ctx)
+	expectOk(t, err)
+
+	kmap := make(map[string]struct{})
+	for _, key := range keys {
+		if _, ok := kmap[key]; ok {
+			t.Fatalf("Already saw %q", key)
+		}
+		kmap[key] = struct{}{}
+	}
+	if len(kmap) != 3 {
+		t.Fatalf("Expected 3 total keys, got %d", len(kmap))
+	}
+	expected := map[string]struct{}{
+		"name":    struct{}{},
+		"age":     struct{}{},
+		"country": struct{}{},
+	}
+	if !reflect.DeepEqual(kmap, expected) {
+		t.Fatalf("Expected %+v but got %+v", expected, kmap)
+	}
+	// Make sure delete and purge do the right thing and not return the keys.
+	err = kv.Delete(ctx, "name")
+	expectOk(t, err)
+	err = kv.Purge(ctx, "country")
+	expectOk(t, err)
+
+	keys, err = kv.Keys(ctx)
+	expectOk(t, err)
+
+	kmap = make(map[string]struct{})
+	for _, key := range keys {
+		if _, ok := kmap[key]; ok {
+			t.Fatalf("Already saw %q", key)
+		}
+		kmap[key] = struct{}{}
+	}
+	if len(kmap) != 1 {
+		t.Fatalf("Expected 1 total key, got %d", len(kmap))
+	}
+	if _, ok := kmap["age"]; !ok {
+		t.Fatalf("Expected %q to be only key present", "age")
+	}
+}
+
+func TestKeyValueCrossAccounts(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+        jetstream: enabled
+        accounts: {
+           A: {
+               users: [ {user: a, password: a} ]
+               jetstream: enabled
+               exports: [
+                   {service: '$JS.API.>' }
+                   {service: '$KV.>'}
+                   {stream: 'accI.>'}
+               ]
+           },
+           I: {
+               users: [ {user: i, password: i} ]
+               imports: [
+                   {service: {account: A, subject: '$JS.API.>'}, to: 'fromA.>' }
+                   {service: {account: A, subject: '$KV.>'}, to: 'fromA.$KV.>' }
+                   {stream: {subject: 'accI.>', account: A}}
+               ]
+           }
+		}`))
+	defer os.Remove(conf)
+	s, _ := RunServerWithConfig(conf)
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	watchNext := func(w jetstream.KeyWatcher) jetstream.KeyValueEntry {
+		t.Helper()
+		select {
+		case e := <-w.Updates():
+			return e
+		case <-time.After(time.Second):
+			t.Fatal("Fail to get the next update")
+		}
+		return nil
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	nc1, js1 := jsClient(t, s, nats.UserInfo("a", "a"))
+	defer nc1.Close()
+
+	kv1, err := js1.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "Map", History: 10})
+	if err != nil {
+		t.Fatalf("Error creating kv store: %v", err)
+	}
+
+	w1, err := kv1.Watch(ctx, "map")
+	if err != nil {
+		t.Fatalf("Error creating watcher: %v", err)
+	}
+	if e := watchNext(w1); e != nil {
+		t.Fatalf("Expected nil entry, got %+v", e)
+	}
+
+	nc2, err := nats.Connect(s.ClientURL(), nats.UserInfo("i", "i"), nats.CustomInboxPrefix("accI"))
+	if err != nil {
+		t.Fatalf("Error on connect: %v", err)
+	}
+	defer nc2.Close()
+	js2, err := jetstream.NewWithAPIPrefix(nc2, "fromA")
+	if err != nil {
+		t.Fatalf("Error getting jetstream context: %v", err)
+	}
+
+	kv2, err := js2.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "Map", History: 10})
+	if err != nil {
+		t.Fatalf("Error creating kv store: %v", err)
+	}
+
+	w2, err := kv2.Watch(ctx, "map")
+	if err != nil {
+		t.Fatalf("Error creating watcher: %v", err)
+	}
+	if e := watchNext(w2); e != nil {
+		t.Fatalf("Expected nil entry, got %+v", e)
+	}
+
+	// Do a Put from kv2
+	rev, err := kv2.Put(ctx, "map", []byte("value"))
+	if err != nil {
+		t.Fatalf("Error on put: %v", err)
+	}
+
+	// Get from kv1
+	e, err := kv1.Get(ctx, "map")
+	if err != nil {
+		t.Fatalf("Error on get: %v", err)
+	}
+	if e.Key() != "map" || string(e.Value()) != "value" {
+		t.Fatalf("Unexpected entry: +%v", e)
+	}
+
+	// Get from kv2
+	e, err = kv2.Get(ctx, "map")
+	if err != nil {
+		t.Fatalf("Error on get: %v", err)
+	}
+	if e.Key() != "map" || string(e.Value()) != "value" {
+		t.Fatalf("Unexpected entry: +%v", e)
+	}
+
+	// Watcher 1
+	if e := watchNext(w1); e == nil || e.Key() != "map" || string(e.Value()) != "value" {
+		t.Fatalf("Unexpected entry: %+v", e)
+	}
+
+	// Watcher 2
+	if e := watchNext(w2); e == nil || e.Key() != "map" || string(e.Value()) != "value" {
+		t.Fatalf("Unexpected entry: %+v", e)
+	}
+
+	// Try an update form kv2
+	if _, err := kv2.Update(ctx, "map", []byte("updated"), rev); err != nil {
+		t.Fatalf("Failed to update: %v", err)
+	}
+
+	// Get from kv1
+	e, err = kv1.Get(ctx, "map")
+	if err != nil {
+		t.Fatalf("Error on get: %v", err)
+	}
+	if e.Key() != "map" || string(e.Value()) != "updated" {
+		t.Fatalf("Unexpected entry: +%v", e)
+	}
+
+	// Get from kv2
+	e, err = kv2.Get(ctx, "map")
+	if err != nil {
+		t.Fatalf("Error on get: %v", err)
+	}
+	if e.Key() != "map" || string(e.Value()) != "updated" {
+		t.Fatalf("Unexpected entry: +%v", e)
+	}
+
+	// Watcher 1
+	if e := watchNext(w1); e == nil || e.Key() != "map" || string(e.Value()) != "updated" {
+		t.Fatalf("Unexpected entry: %+v", e)
+	}
+
+	// Watcher 2
+	if e := watchNext(w2); e == nil || e.Key() != "map" || string(e.Value()) != "updated" {
+		t.Fatalf("Unexpected entry: %+v", e)
+	}
+
+	// Purge from kv2
+	if err := kv2.Purge(ctx, "map"); err != nil {
+		t.Fatalf("Error on purge: %v", err)
+	}
+
+	// Check purge ok from w1
+	if e := watchNext(w1); e == nil || e.Operation() != jetstream.KeyValuePurge {
+		t.Fatalf("Unexpected entry: %+v", e)
+	}
+
+	// Check purge ok from w2
+	if e := watchNext(w2); e == nil || e.Operation() != jetstream.KeyValuePurge {
+		t.Fatalf("Unexpected entry: %+v", e)
+	}
+
+	// Delete purge records from kv2
+	if err := kv2.PurgeDeletes(ctx, jetstream.DeleteMarkersOlderThan(-1)); err != nil {
+		t.Fatalf("Error on purge deletes: %v", err)
+	}
+
+	// Check all gone from js1
+	if si, err := js1.Stream(ctx, "KV_Map"); err != nil || si == nil || si.CachedInfo().State.Msgs != 0 {
+		t.Fatalf("Error getting stream info: err=%v si=%+v", err, si)
+	}
+
+	// Delete key from kv2
+	if err := kv2.Delete(ctx, "map"); err != nil {
+		t.Fatalf("Error on delete: %v", err)
+	}
+
+	// Check key gone from kv1
+	if e, err := kv1.Get(ctx, "map"); err != jetstream.ErrKeyNotFound || e != nil {
+		t.Fatalf("Expected key not found, got err=%v e=%+v", err, e)
+	}
+}
+
+func TestKeyValueDuplicatesWindow(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	checkWindow := func(ttl, expectedDuplicates time.Duration) {
+		t.Helper()
+
+		_, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "TEST", History: 5, TTL: ttl})
+		expectOk(t, err)
+		defer func() { expectOk(t, js.DeleteKeyValue(ctx, "TEST")) }()
+
+		si, err := js.Stream(ctx, "KV_TEST")
+		if err != nil {
+			t.Fatalf("StreamInfo error: %v", err)
+		}
+		if si.CachedInfo().Config.Duplicates != expectedDuplicates {
+			t.Fatalf("Expected duplicates to be %v, got %v", expectedDuplicates, si.CachedInfo().Config.Duplicates)
+		}
+	}
+
+	checkWindow(0, 2*time.Minute)
+	checkWindow(time.Hour, 2*time.Minute)
+	checkWindow(5*time.Second, 5*time.Second)
+}
+
+func TestListKeyValueStores(t *testing.T) {
+	tests := []struct {
+		name       string
+		bucketsNum int
+	}{
+		{
+			name:       "single page",
+			bucketsNum: 5,
+		},
+		{
+			name:       "multi page",
+			bucketsNum: 1025,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			s := RunBasicJetStreamServer()
+			defer shutdownJSServerAndRemoveStorage(t, s)
+
+			nc, js := jsClient(t, s)
+			defer nc.Close()
+			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+			defer cancel()
+
+			// create stream without the chunk subject, but with KV_ prefix
+			_, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "KV_FOO", Subjects: []string{"FOO.*"}})
+			expectOk(t, err)
+			// create stream with chunk subject, but without "KV_" prefix
+			_, err = js.CreateStream(ctx, jetstream.StreamConfig{Name: "FOO", Subjects: []string{"$KV.ABC.>"}})
+			expectOk(t, err)
+			for i := 0; i < test.bucketsNum; i++ {
+				_, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: fmt.Sprintf("KVS_%d", i), MaxBytes: 1024})
+				expectOk(t, err)
+			}
+			names := make([]string, 0)
+			kvNames := js.KeyValueStoreNames(ctx)
+			for name := range kvNames.Name() {
+				names = append(names, name)
+			}
+			if kvNames.Error() != nil {
+				t.Fatalf("Unexpected error: %v", kvNames.Error())
+			}
+			if len(names) != test.bucketsNum {
+				t.Fatalf("Invalid number of stream names; want: %d; got: %d", test.bucketsNum, len(names))
+			}
+			infos := make([]nats.KeyValueStatus, 0)
+			kvInfos := js.KeyValueStores(ctx)
+			for info := range kvInfos.Status() {
+				infos = append(infos, info)
+			}
+			if kvInfos.Error() != nil {
+				t.Fatalf("Unexpected error: %v", kvNames.Error())
+			}
+			if len(infos) != test.bucketsNum {
+				t.Fatalf("Invalid number of streams; want: %d; got: %d", test.bucketsNum, len(infos))
+			}
+		})
+	}
+}
+
+func TestKeyValueMirrorCrossDomains(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		server_name: HUB
+		listen: 127.0.0.1:-1
+		jetstream: { domain: HUB }
+		leafnodes { listen: 127.0.0.1:7422 }
+	}`))
+	defer os.Remove(conf)
+	s, _ := RunServerWithConfig(conf)
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	lconf := createConfFile(t, []byte(`
+		server_name: LEAF
+		listen: 127.0.0.1:-1
+ 		jetstream: { domain:LEAF }
+ 		leafnodes {
+ 		 	remotes = [ { url: "leaf://127.0.0.1" } ]
+ 		}
+	}`))
+	defer os.Remove(lconf)
+	ln, _ := RunServerWithConfig(lconf)
+	defer shutdownJSServerAndRemoveStorage(t, ln)
+
+	// Create main KV on HUB
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "TEST"})
+	expectOk(t, err)
+
+	_, err = kv.PutString(ctx, "name", "derek")
+	expectOk(t, err)
+	_, err = kv.PutString(ctx, "age", "22")
+	expectOk(t, err)
+	_, err = kv.PutString(ctx, "v", "v")
+	expectOk(t, err)
+	err = kv.Delete(ctx, "v")
+	expectOk(t, err)
+
+	lnc, ljs := jsClient(t, ln)
+	defer lnc.Close()
+
+	// Capture cfg so we can make sure it does not change.
+	// NOTE: We use different name to test all possibilities, etc, but in practice for truly nomadic applications
+	// this should be named the same, e.g. TEST.
+	cfg := jetstream.KeyValueConfig{
+		Bucket: "MIRROR",
+		Mirror: &jetstream.StreamSource{
+			Name:   "TEST",
+			Domain: "HUB",
+		},
+	}
+	ccfg := cfg
+
+	_, err = ljs.CreateKeyValue(ctx, cfg)
+	expectOk(t, err)
+
+	if !reflect.DeepEqual(cfg, ccfg) {
+		t.Fatalf("Did not expect config to be altered: %+v vs %+v", cfg, ccfg)
+	}
+
+	si, err := ljs.Stream(ctx, "KV_MIRROR")
+	expectOk(t, err)
+
+	// Make sure mirror direct set.
+	if !si.CachedInfo().Config.MirrorDirect {
+		t.Fatalf("Expected mirror direct to be set")
+	}
+
+	// Make sure we sync.
+	checkFor(t, 2*time.Second, 15*time.Millisecond, func() error {
+		si, err := ljs.Stream(ctx, "KV_MIRROR")
+		expectOk(t, err)
+		if si.CachedInfo().State.Msgs == 3 {
+			return nil
+		}
+		return fmt.Errorf("Did not get synched messages: %d", si.CachedInfo().State.Msgs)
+	})
+
+	// Bind locally from leafnode and make sure both get and put work.
+	mkv, err := ljs.KeyValue(ctx, "MIRROR")
+	expectOk(t, err)
+
+	_, err = mkv.PutString(ctx, "name", "rip")
+	expectOk(t, err)
+
+	_, err = mkv.PutString(ctx, "v", "vv")
+	expectOk(t, err)
+	e, err := mkv.Get(ctx, "v")
+	expectOk(t, err)
+	if e.Operation() != jetstream.KeyValuePut {
+		t.Fatalf("Got wrong value: %q vs %q", e.Operation(), nats.KeyValuePut)
+	}
+	err = mkv.Delete(ctx, "v")
+	expectOk(t, err)
+	_, err = mkv.Get(ctx, "v")
+	expectErr(t, err, jetstream.ErrKeyNotFound)
+
+	e, err = mkv.Get(ctx, "name")
+	expectOk(t, err)
+	if string(e.Value()) != "rip" {
+		t.Fatalf("Got wrong value: %q vs %q", e.Value(), "rip")
+	}
+
+	// Also make sure we can create a watcher on the mirror KV.
+	watcher, err := mkv.WatchAll(ctx)
+	expectOk(t, err)
+	defer watcher.Stop()
+
+	// Bind through leafnode connection but to origin KV.
+	rjs, err := jetstream.NewWithDomain(nc, "HUB")
+	expectOk(t, err)
+
+	rkv, err := rjs.KeyValue(ctx, "TEST")
+	expectOk(t, err)
+
+	_, err = rkv.PutString(ctx, "name", "ivan")
+	expectOk(t, err)
+
+	e, err = rkv.Get(ctx, "name")
+	expectOk(t, err)
+	if string(e.Value()) != "ivan" {
+		t.Fatalf("Got wrong value: %q vs %q", e.Value(), "ivan")
+	}
+	_, err = rkv.PutString(ctx, "v", "vv")
+	expectOk(t, err)
+	e, err = mkv.Get(ctx, "v")
+	expectOk(t, err)
+	if e.Operation() != jetstream.KeyValuePut {
+		t.Fatalf("Got wrong value: %q vs %q", e.Operation(), nats.KeyValuePut)
+	}
+	err = rkv.Delete(ctx, "v")
+	expectOk(t, err)
+	_, err = rkv.Get(ctx, "v")
+	expectErr(t, err, jetstream.ErrKeyNotFound)
+
+	// Shutdown cluster and test get still work.
+	shutdownJSServerAndRemoveStorage(t, s)
+
+	e, err = mkv.Get(ctx, "name")
+	expectOk(t, err)
+	if string(e.Value()) != "ivan" {
+		t.Fatalf("Got wrong value: %q vs %q", e.Value(), "ivan")
+	}
+}
+
+func TestKeyValueRePublish(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
+		Bucket: "TEST_UPDATE",
+	}); err != nil {
+		t.Fatalf("Error creating store: %v", err)
+	}
+	// This is expected to fail since server does not support as of now
+	// the update of RePublish.
+	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
+		Bucket:    "TEST_UPDATE",
+		RePublish: &jetstream.RePublish{Source: ">", Destination: "bar.>"},
+	}); err == nil {
+		t.Fatal("Expected failure, did not get one")
+	}
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
+		Bucket:    "TEST",
+		RePublish: &jetstream.RePublish{Source: ">", Destination: "bar.>"},
+	})
+	if err != nil {
+		t.Fatalf("Error creating store: %v", err)
+	}
+	si, err := js.Stream(ctx, "KV_TEST")
+	if err != nil {
+		t.Fatalf("Error getting stream info: %v", err)
+	}
+	if si.CachedInfo().Config.RePublish == nil {
+		t.Fatal("Expected republish to be set, it was not")
+	}
+
+	sub, err := nc.SubscribeSync("bar.>")
+	if err != nil {
+		t.Fatalf("Error on sub: %v", err)
+	}
+	if _, err := kv.Put(ctx, "foo", []byte("value")); err != nil {
+		t.Fatalf("Error on put: %v", err)
+	}
+	msg, err := sub.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("Error on next: %v", err)
+	}
+	if v := string(msg.Data); v != "value" {
+		t.Fatalf("Unexpected value: %s", v)
+	}
+	// The message should also have a header with the actual subject
+	expected := "$KV.TEST.foo"
+	if v := msg.Header.Get(jetstream.SubjectHeader); v != expected {
+		t.Fatalf("Expected subject header %q, got %q", expected, v)
+	}
+}
+
+func TestKeyValueMirrorDirectGet(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "TEST"})
+	if err != nil {
+		t.Fatalf("Error creating kv: %v", err)
+	}
+	_, err = js.CreateStream(ctx, jetstream.StreamConfig{
+		Name:         "MIRROR",
+		Mirror:       &jetstream.StreamSource{Name: "KV_TEST"},
+		MirrorDirect: true,
+	})
+	if err != nil {
+		t.Fatalf("Error creating mirror: %v", err)
+	}
+
+	for i := 0; i < 100; i++ {
+		key := fmt.Sprintf("KEY.%d", i)
+		if _, err := kv.PutString(ctx, key, "42"); err != nil {
+			t.Fatalf("Error adding key: %v", err)
+		}
+	}
+
+	// Make sure all gets work.
+	for i := 0; i < 100; i++ {
+		if _, err := kv.Get(ctx, "KEY.22"); err != nil {
+			t.Fatalf("Got error getting key: %v", err)
+		}
+	}
+}
+
+func TestKeyValueCreate(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "TEST"})
+	if err != nil {
+		t.Fatalf("Error creating kv: %v", err)
+	}
+
+	_, err = kv.Create(ctx, "key", []byte("1"))
+	if err != nil {
+		t.Fatalf("Error creating key: %v", err)
+	}
+
+	_, err = kv.Create(ctx, "key", []byte("1"))
+	expected := "wrong last sequence: 1: key exists"
+	if !strings.Contains(err.Error(), expected) {
+		t.Fatalf("Expected %q, got: %v", expected, err)
+	}
+	if !errors.Is(err, jetstream.ErrKeyExists) {
+		t.Fatalf("Expected ErrKeyExists, got: %v", err)
+	}
+	aerr := &jetstream.APIError{}
+	if !errors.As(err, &aerr) {
+		t.Fatalf("Expected APIError, got: %v", err)
+	}
+	if aerr.Description != "wrong last sequence: 1" {
+		t.Fatalf("Unexpected APIError message, got: %v", aerr.Description)
+	}
+	if aerr.ErrorCode != 10071 {
+		t.Fatalf("Unexpected error code, got: %v", aerr.ErrorCode)
+	}
+	if aerr.Code != jetstream.ErrKeyExists.APIError().Code {
+		t.Fatalf("Unexpected error code, got: %v", aerr.Code)
+	}
+	var kerr jetstream.JetStreamError
+	if !errors.As(err, &kerr) {
+		t.Fatalf("Expected KeyValueError, got: %v", err)
+	}
+	if kerr.APIError().ErrorCode != 10071 {
+		t.Fatalf("Unexpected error code, got: %v", kerr.APIError().ErrorCode)
+	}
+}
+
+// Helpers
+
+func client(t *testing.T, s *server.Server, opts ...nats.Option) *nats.Conn {
+	t.Helper()
+	nc, err := nats.Connect(s.ClientURL(), opts...)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	return nc
+}
+
+func jsClient(t *testing.T, s *server.Server, opts ...nats.Option) (*nats.Conn, jetstream.JetStream) {
+	t.Helper()
+	nc := client(t, s, opts...)
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatalf("Unexpected error getting JetStream context: %v", err)
+	}
+	return nc, js
+}
+
+func expectOk(t *testing.T, err error) {
+	t.Helper()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+}
+
+func expectErr(t *testing.T, err error, expected ...error) {
+	t.Helper()
+	if err == nil {
+		t.Fatalf("Expected error but got none")
+	}
+	if len(expected) == 0 {
+		return
+	}
+	for _, e := range expected {
+		if err == e || strings.Contains(e.Error(), err.Error()) {
+			return
+		}
+	}
+	t.Fatalf("Expected one of %+v, got '%v'", expected, err)
+}
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..ac7a34c 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 {
@@ -1233,6 +1233,43 @@ func TestPublishMsgAsyncWithPendingMsgs(t *testing.T) {
 			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrTooManyStalledMsgs, err)
 		}
 	})
+
+	t.Run("with server restart", func(t *testing.T) {
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+
+		_, err = js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+
+		go func() {
+			for i := 0; i < 50; i++ {
+				_, _ = js.PublishAsync("FOO.1", []byte("msg"))
+			}
+		}()
+		srv = restartBasicJSServer(t, srv)
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+
+		select {
+		case <-js.PublishAsyncComplete():
+		case <-time.After(10 * time.Second):
+			t.Fatalf("Did not receive completion signal")
+		}
+	})
 }
 
 func TestPublishAsyncResetPendingOnReconnect(t *testing.T) {
diff --git a/nats_test.go b/nats_test.go
index 5cee524..b70a190 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -1575,6 +1575,38 @@ func TestUserCredentialsChainedFile(t *testing.T) {
 	}
 }
 
+func TestReconnectMissingCredentials(t *testing.T) {
+	ts := runTrustServer()
+	defer ts.Shutdown()
+
+	chainedFile := createTmpFile(t, []byte(chained))
+	defer os.Remove(chainedFile)
+
+	url := fmt.Sprintf("nats://127.0.0.1:%d", TEST_PORT)
+	errs := make(chan error, 1)
+	nc, err := Connect(url, UserCredentials(chainedFile), ErrorHandler(func(_ *Conn, _ *Subscription, err error) {
+		errs <- err
+	}))
+	if err != nil {
+		t.Fatalf("Expected to connect, got %v", err)
+	}
+	defer nc.Close()
+	os.Remove(chainedFile)
+	ts.Shutdown()
+
+	ts = runTrustServer()
+	defer ts.Shutdown()
+
+	select {
+	case err := <-errs:
+		if !strings.Contains(err.Error(), "no such file or directory") {
+			t.Fatalf("Expected error about missing creds file, got %q", err)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("Did not get error about missing creds file")
+	}
+}
+
 func TestUserJWTAndSeed(t *testing.T) {
 	if server.VERSION[0] == '1' {
 		t.Skip()
@@ -2854,7 +2886,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..dd77021
--- /dev/null
+++ b/test/compat_test.go
@@ -0,0 +1,470 @@
+//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 TestCompatibilityObjectStoreGetLink(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-link.>")
+	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)
+	}
+	obj, err := os.Get(cfg.Object)
+	if err != nil {
+		t.Fatalf("Error getting object: %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)
+
+	if err := msg.Respond(sha); err != nil {
+		t.Fatalf("Error responding to message: %v", err)
+	}
+
+	validateTestResult(t, sub)
+}
+
+func TestCompatibilityObjectStorePutLink(t *testing.T) {
+	t.Parallel()
+
+	type config struct {
+		Bucket   string `json:"bucket"`
+		Object   string `json:"object"`
+		LinkName string `json:"link_name"`
+	}
+
+	nc, js := connect(t)
+	defer nc.Close()
+
+	// setup subscription on which tester will be sending requests
+	sub, err := nc.SubscribeSync("tests.object-store.put-link.>")
+	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)
+	}
+	sourceObj, err := os.GetInfo(cfg.Object)
+	if err != nil {
+		t.Fatalf("Error getting object: %v", err)
+	}
+	_, err = os.AddLink(cfg.LinkName, sourceObj)
+	if err != nil {
+		t.Fatalf("Error adding link: %v", err)
+	}
+
+	if err := msg.Respond(nil); 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 849de3c..d9ad8ca 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -476,6 +476,13 @@ func TestJetStreamSubscribe(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
+	initialPending, err := sub.InitialConsumerPending()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if initialPending != 0 {
+		t.Fatalf("Expected no initial pending, got %d", initialPending)
+	}
 	sub.Unsubscribe()
 
 	// Check that Queue subscribe with HB or FC fails.
@@ -563,12 +570,20 @@ func TestJetStreamSubscribe(t *testing.T) {
 			done <- true
 		}
 	})
+
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 	expectConsumers(t, 3)
 	defer sub3.Unsubscribe()
 
+	initialPending, err = sub3.InitialConsumerPending()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if initialPending != 10 {
+		t.Fatalf("Expected initial pending of 10, got %d", initialPending)
+	}
 	select {
 	case <-done:
 	case <-time.After(5 * time.Second):
@@ -1409,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)
@@ -2712,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 {
@@ -2734,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)
 			}
 		})
 	}
@@ -4094,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)
 	}
 }
 
@@ -4155,12 +4170,11 @@ func TestJetStreamSubscribe_AckPolicy(t *testing.T) {
 			}
 
 			<-ctx.Done()
-			sub.Drain()
-
 			if got != totalMsgs {
 				t.Fatalf("Expected %d, got %d", totalMsgs, got)
 			}
 
+			// check if consumer is configured properly
 			ci, err := js.ConsumerInfo("TEST", test.name)
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
@@ -4168,6 +4182,9 @@ func TestJetStreamSubscribe_AckPolicy(t *testing.T) {
 			if ci.Config.AckPolicy != test.expected {
 				t.Fatalf("Expected %v, got %v", test.expected, ci.Config.AckPolicy)
 			}
+
+			// drain the subscription. This will remove the consumer
+			sub.Drain()
 		})
 	}
 
@@ -7852,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 c668e6f..026891a 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -149,88 +149,218 @@ func TestKeyValueHistory(t *testing.T) {
 }
 
 func TestKeyValueWatch(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer shutdownJSServerAndRemoveStorage(t, s)
-
-	nc, js := jsClient(t, s)
-	defer nc.Close()
-
-	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "WATCH"})
-	expectOk(t, err)
-
-	watcher, err := kv.WatchAll()
-	expectOk(t, err)
-	defer watcher.Stop()
-
-	expectUpdate := func(key, value string, revision uint64) {
-		t.Helper()
-		select {
-		case v := <-watcher.Updates():
-			if v.Key() != key || string(v.Value()) != value || v.Revision() != revision {
-				t.Fatalf("Did not get expected: %+v vs %q %q %d", v, key, value, revision)
+	expectUpdateF := func(t *testing.T, watcher nats.KeyWatcher) func(key, value string, revision uint64) {
+		return func(key, value string, revision uint64) {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v.Key() != key || string(v.Value()) != value || v.Revision() != revision {
+					t.Fatalf("Did not get expected: %q %q %d vs %q %q %d", v.Key(), string(v.Value()), v.Revision(), key, value, revision)
+				}
+			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")
 		}
 	}
-	expectDelete := func(key string, revision uint64) {
-		t.Helper()
-		select {
-		case v := <-watcher.Updates():
-			if v.Operation() != nats.KeyValueDelete {
-				t.Fatalf("Expected a delete operation but got %+v", v)
+	expectDeleteF := func(t *testing.T, watcher nats.KeyWatcher) func(key string, revision uint64) {
+		return func(key string, revision uint64) {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v.Operation() != nats.KeyValueDelete {
+					t.Fatalf("Expected a delete operation but got %+v", v)
+				}
+				if v.Revision() != revision {
+					t.Fatalf("Did not get expected revision: %d vs %d", revision, v.Revision())
+				}
+			case <-time.After(time.Second):
+				t.Fatalf("Did not receive an update like expected")
 			}
-			if v.Revision() != revision {
-				t.Fatalf("Did not get expected revision: %d vs %d", revision, v.Revision())
-			}
-		case <-time.After(time.Second):
-			t.Fatalf("Did not receive an update like expected")
 		}
 	}
-	expectInitDone := func() {
-		t.Helper()
-		select {
-		case v := <-watcher.Updates():
-			if v != nil {
-				t.Fatalf("Did not get expected: %+v", v)
+	expectInitDoneF := func(t *testing.T, watcher nats.KeyWatcher) func() {
+		return func() {
+			t.Helper()
+			select {
+			case v := <-watcher.Updates():
+				if v != nil {
+					t.Fatalf("Did not get expected: %+v", v)
+				}
+			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")
 		}
 	}
 
-	// Make sure we already got an initial value marker.
-	expectInitDone()
+	t.Run("default watcher", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
 
-	kv.Create("name", []byte("derek"))
-	expectUpdate("name", "derek", 1)
-	kv.Put("name", []byte("rip"))
-	expectUpdate("name", "rip", 2)
-	kv.Put("name", []byte("ik"))
-	expectUpdate("name", "ik", 3)
-	kv.Put("age", []byte("22"))
-	expectUpdate("age", "22", 4)
-	kv.Put("age", []byte("33"))
-	expectUpdate("age", "33", 5)
-	kv.Delete("age")
-	expectDelete("age", 6)
+		nc, js := jsClient(t, s)
+		defer nc.Close()
 
-	// Stop first watcher.
-	watcher.Stop()
+		kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "WATCH"})
+		expectOk(t, err)
 
-	// Now try wildcard matching and make sure we only get last value when starting.
-	kv.Put("t.name", []byte("rip"))
-	kv.Put("t.name", []byte("ik"))
-	kv.Put("t.age", []byte("22"))
-	kv.Put("t.age", []byte("44"))
+		watcher, err := kv.WatchAll()
+		expectOk(t, err)
+		defer watcher.Stop()
+
+		expectInitDone := expectInitDoneF(t, watcher)
+		expectUpdate := expectUpdateF(t, watcher)
+		expectDelete := expectDeleteF(t, watcher)
+		// Make sure we already got an initial value marker.
+		expectInitDone()
+
+		kv.Create("name", []byte("derek"))
+		expectUpdate("name", "derek", 1)
+		kv.Put("name", []byte("rip"))
+		expectUpdate("name", "rip", 2)
+		kv.Put("name", []byte("ik"))
+		expectUpdate("name", "ik", 3)
+		kv.Put("age", []byte("22"))
+		expectUpdate("age", "22", 4)
+		kv.Put("age", []byte("33"))
+		expectUpdate("age", "33", 5)
+		kv.Delete("age")
+		expectDelete("age", 6)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		// Now try wildcard matching and make sure we only get last value when starting.
+		kv.Put("t.name", []byte("rip"))
+		kv.Put("t.name", []byte("ik"))
+		kv.Put("t.age", []byte("22"))
+		kv.Put("t.age", []byte("44"))
+
+		watcher, err = kv.Watch("t.*")
+		expectOk(t, err)
+		defer watcher.Stop()
 
-	watcher, err = kv.Watch("t.*")
-	expectOk(t, err)
-	defer watcher.Stop()
+		expectInitDone = expectInitDoneF(t, watcher)
+		expectUpdate = expectUpdateF(t, watcher)
+		expectUpdate("t.name", "ik", 8)
+		expectUpdate("t.age", "44", 10)
+		expectInitDone()
+	})
+
+	t.Run("watcher with history included", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "WATCH", History: 64})
+		expectOk(t, err)
+
+		kv.Create("name", []byte("derek"))
+		kv.Put("name", []byte("rip"))
+		kv.Put("name", []byte("ik"))
+		kv.Put("age", []byte("22"))
+		kv.Put("age", []byte("33"))
+		kv.Delete("age")
+
+		// when using UpdatesOnly(), IncludeHistory() is not allowed
+		if _, err := kv.WatchAll(nats.IncludeHistory(), nats.UpdatesOnly()); !strings.Contains(err.Error(), "updates only can not be used with include history") {
+			t.Fatalf("Expected error to contain %q, got %q", "updates only can not be used with include history", err)
+		}
+
+		watcher, err := kv.WatchAll(nats.IncludeHistory())
+		expectOk(t, err)
+		defer watcher.Stop()
+		expectInitDone := expectInitDoneF(t, watcher)
+		expectUpdate := expectUpdateF(t, watcher)
+		expectDelete := expectDeleteF(t, watcher)
+		expectUpdate("name", "derek", 1)
+		expectUpdate("name", "rip", 2)
+		expectUpdate("name", "ik", 3)
+		expectUpdate("age", "22", 4)
+		expectUpdate("age", "33", 5)
+		expectDelete("age", 6)
+		expectInitDone()
+		kv.Put("name", []byte("pp"))
+		expectUpdate("name", "pp", 7)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		kv.Put("t.name", []byte("rip"))
+		kv.Put("t.name", []byte("ik"))
+		kv.Put("t.age", []byte("22"))
+		kv.Put("t.age", []byte("44"))
+
+		// try wildcard watcher and make sure we get all historical values
+		watcher, err = kv.Watch("t.*", nats.IncludeHistory())
+		expectOk(t, err)
+		defer watcher.Stop()
+		expectInitDone = expectInitDoneF(t, watcher)
+		expectUpdate = expectUpdateF(t, watcher)
+
+		expectUpdate("t.name", "rip", 8)
+		expectUpdate("t.name", "ik", 9)
+		expectUpdate("t.age", "22", 10)
+		expectUpdate("t.age", "44", 11)
+		expectInitDone()
+
+		kv.Put("t.name", []byte("pp"))
+		expectUpdate("t.name", "pp", 12)
+	})
+
+	t.Run("watcher with updates only", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "WATCH", History: 64})
+		expectOk(t, err)
 
-	expectUpdate("t.name", "ik", 8)
-	expectUpdate("t.age", "44", 10)
-	expectInitDone()
+		kv.Create("name", []byte("derek"))
+		kv.Put("name", []byte("rip"))
+		kv.Put("age", []byte("22"))
+
+		// when using UpdatesOnly(), IncludeHistory() is not allowed
+		if _, err := kv.WatchAll(nats.UpdatesOnly(), nats.IncludeHistory()); !strings.Contains(err.Error(), "include history can not be used with updates only") {
+			t.Fatalf("Expected error to contain %q, got %q", "include history can not be used with updates only", err)
+		}
+
+		watcher, err := kv.WatchAll(nats.UpdatesOnly())
+		expectOk(t, err)
+		defer watcher.Stop()
+		expectUpdate := expectUpdateF(t, watcher)
+		expectDelete := expectDeleteF(t, watcher)
+
+		// now update some keys and expect updates
+		kv.Put("name", []byte("pp"))
+		expectUpdate("name", "pp", 4)
+		kv.Put("age", []byte("44"))
+		expectUpdate("age", "44", 5)
+		kv.Delete("age")
+		expectDelete("age", 6)
+
+		// Stop first watcher.
+		watcher.Stop()
+
+		kv.Put("t.name", []byte("rip"))
+		kv.Put("t.name", []byte("ik"))
+		kv.Put("t.age", []byte("22"))
+		kv.Put("t.age", []byte("44"))
+
+		// try wildcard watcher and make sure we do not get any values initially
+		watcher, err = kv.Watch("t.*", nats.UpdatesOnly())
+		expectOk(t, err)
+		defer watcher.Stop()
+		expectUpdate = expectUpdateF(t, watcher)
+
+		// update some keys and expect updates
+		kv.Put("t.name", []byte("pp"))
+		expectUpdate("t.name", "pp", 11)
+		kv.Put("t.age", []byte("66"))
+		expectUpdate("t.age", "66", 12)
+	})
 }
 
 func TestKeyValueWatchContext(t *testing.T) {
@@ -952,6 +1082,8 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 
 	_, err = mkv.PutString("v", "vv")
 	expectOk(t, err)
+	// wait for the key to be propagated to the mirror
+	time.Sleep(10 * time.Millisecond)
 	e, err := mkv.Get("v")
 	expectOk(t, err)
 	if e.Operation() != nats.KeyValuePut {
@@ -959,6 +1091,7 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 	}
 	err = mkv.Delete("v")
 	expectOk(t, err)
+	time.Sleep(10 * time.Millisecond)
 	_, err = mkv.Get("v")
 	expectErr(t, err, nats.ErrKeyNotFound)
 
@@ -983,6 +1116,7 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 	_, err = rkv.PutString("name", "ivan")
 	expectOk(t, err)
 
+	time.Sleep(10 * time.Millisecond)
 	e, err = rkv.Get("name")
 	expectOk(t, err)
 	if string(e.Value()) != "ivan" {
@@ -990,6 +1124,7 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 	}
 	_, err = rkv.PutString("v", "vv")
 	expectOk(t, err)
+	time.Sleep(10 * time.Millisecond)
 	e, err = mkv.Get("v")
 	expectOk(t, err)
 	if e.Operation() != nats.KeyValuePut {
@@ -997,6 +1132,7 @@ func TestKeyValueMirrorCrossDomains(t *testing.T) {
 	}
 	err = rkv.Delete("v")
 	expectOk(t, err)
+	time.Sleep(10 * time.Millisecond)
 	_, err = rkv.Get("v")
 	expectErr(t, err, nats.ErrKeyNotFound)
 
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) {
diff --git a/test/sub_test.go b/test/sub_test.go
index 253e00f..cb177c7 100644
--- a/test/sub_test.go
+++ b/test/sub_test.go
@@ -1571,3 +1571,46 @@ func TestAutoUnsubOnSyncSubCanStillRespond(t *testing.T) {
 		t.Fatalf("Error responding: %v", err)
 	}
 }
+
+func TestSubscribe_ClosedHandler(t *testing.T) {
+	s := RunDefaultServer()
+	defer s.Shutdown()
+
+	nc := NewDefaultConnection(t)
+	defer nc.Close()
+
+	ch := make(chan string, 1)
+	sub, err := nc.Subscribe("foo", func(_ *nats.Msg) {})
+	if err != nil {
+		t.Fatalf("Error subscribing: %v", err)
+	}
+	sub.SetClosedHandler(func(subj string) {
+		ch <- subj
+	})
+	sub.Unsubscribe()
+	select {
+	case subj := <-ch:
+		if subj != "foo" {
+			t.Fatalf("Expected 'foo', got '%v'", subj)
+		}
+	case <-time.After(1 * time.Second):
+		t.Fatal("Did not receive closed callback")
+	}
+
+	sub, err = nc.Subscribe("bar", func(_ *nats.Msg) {})
+	if err != nil {
+		t.Fatalf("Error subscribing: %v", err)
+	}
+	sub.SetClosedHandler(func(subj string) {
+		ch <- subj
+	})
+	sub.Drain()
+	select {
+	case subj := <-ch:
+		if subj != "bar" {
+			t.Fatalf("Expected 'bar', got '%v'", subj)
+		}
+	case <-time.After(1 * time.Second):
+		t.Fatal("Did not receive closed callback")
+	}
+}