Repository: nats-io/nats.go
Pull request #1040 (2022-08-10 12:54:43)

Remove TODO.md
Removing `TODO.md`, as it is outdated (last commit 7 years ago)


Patch:
diff --git a/TODO.md b/TODO.md
deleted file mode 100644
index 213aaec..0000000
--- a/TODO.md
+++ /dev/null
@@ -1,26 +0,0 @@
-
-- [ ] Better constructors, options handling
-- [ ] Functions for callback settings after connection created.
-- [ ] Better options for subscriptions. Slow Consumer state settable, Go routines vs Inline.
-- [ ] Move off of channels for subscribers, use syncPool linkedLists, etc with highwater.
-- [ ] Test for valid subjects on publish and subscribe?
-- [ ] SyncSubscriber and Next for EncodedConn
-- [ ] Fast Publisher?
-- [ ] pooling for structs used? leaky bucket?
-- [ ] Timeout 0 should work as no timeout
-- [x] Ping timer
-- [x] Name in Connect for gnatsd
-- [x] Asynchronous error handling
-- [x] Parser rewrite
-- [x] Reconnect
-- [x] Hide Lock
-- [x] Easier encoder interface
-- [x] QueueSubscribeSync
-- [x] Make nats specific errors prefixed with 'nats:'
-- [x] API test for closed connection
-- [x] TLS/SSL
-- [x] Stats collection
-- [x] Disconnect detection
-- [x] Optimized Publish (coalescing)
-- [x] Do Examples via Go style
-- [x] Standardized Errors
diff --git a/js.go b/js.go
index 7726585..bb827f9 100644
--- a/js.go
+++ b/js.go
@@ -524,7 +524,7 @@ func (js *js) PublishMsg(m *Msg, opts ...PubOpt) (*PubAck, error) {
 		return nil, ErrInvalidJSAck
 	}
 	if pa.Error != nil {
-		return nil, fmt.Errorf("nats: %s", pa.Error.Description)
+		return nil, pa.Error
 	}
 	if pa.PubAck == nil || pa.PubAck.Stream == _EMPTY_ {
 		return nil, ErrInvalidJSAck
@@ -728,7 +728,7 @@ func (js *js) handleAsyncReply(m *Msg) {
 		return
 	}
 	if pa.Error != nil {
-		doErr(fmt.Errorf("nats: %s", pa.Error.Description))
+		doErr(pa.Error)
 		return
 	}
 	if pa.PubAck == nil || pa.PubAck.Stream == _EMPTY_ {
@@ -1037,8 +1037,9 @@ type ConsumerConfig struct {
 	HeadersOnly     bool            `json:"headers_only,omitempty"`
 
 	// Pull based options.
-	MaxRequestBatch   int           `json:"max_batch,omitempty"`
-	MaxRequestExpires time.Duration `json:"max_expires,omitempty"`
+	MaxRequestBatch    int           `json:"max_batch,omitempty"`
+	MaxRequestExpires  time.Duration `json:"max_expires,omitempty"`
+	MaxRequestMaxBytes int           `json:"max_bytes,omitempty"`
 
 	// Push based consumers.
 	DeliverSubject string `json:"deliver_subject,omitempty"`
@@ -1084,9 +1085,10 @@ type SequencePair struct {
 
 // nextRequest is for getting next messages for pull based consumers from JetStream.
 type nextRequest struct {
-	Expires time.Duration `json:"expires,omitempty"`
-	Batch   int           `json:"batch,omitempty"`
-	NoWait  bool          `json:"no_wait,omitempty"`
+	Expires  time.Duration `json:"expires,omitempty"`
+	Batch    int           `json:"batch,omitempty"`
+	NoWait   bool          `json:"no_wait,omitempty"`
+	MaxBytes int           `json:"max_bytes,omitempty"`
 }
 
 // jsSub includes JetStream subscription info.
@@ -1228,7 +1230,10 @@ func (js *js) ChanQueueSubscribe(subj, queue string, ch chan *Msg, opts ...SubOp
 // See important note in Subscribe()
 func (js *js) PullSubscribe(subj, durable string, opts ...SubOpt) (*Subscription, error) {
 	mch := make(chan *Msg, js.nc.Opts.SubChanLen)
-	return js.subscribe(subj, _EMPTY_, nil, mch, true, true, append(opts, Durable(durable)))
+	if durable != "" {
+		opts = append(opts, Durable(durable))
+	}
+	return js.subscribe(subj, _EMPTY_, nil, mch, true, true, opts)
 }
 
 func processConsInfo(info *ConsumerInfo, userCfg *ConsumerConfig, isPullMode bool, subj, queue string) (string, error) {
@@ -1394,7 +1399,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		// If this is a queue subscription and no consumer nor durable name was specified,
 		// then we will use the queue name as a durable name.
 		if o.consumer == _EMPTY_ && o.cfg.Durable == _EMPTY_ {
-			if err := checkDurName(queue); err != nil {
+			if err := checkConsumerName(queue); err != nil {
 				return nil, err
 			}
 			o.cfg.Durable = queue
@@ -1651,8 +1656,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 				cleanUpSub()
 			}
 			if consumer != _EMPTY_ &&
-				(strings.Contains(cinfo.Error.Description, `consumer already exists`) ||
-					strings.Contains(cinfo.Error.Description, `consumer name already in use`)) {
+				(cinfo.Error.ErrorCode == JSErrCodeConsumerAlreadyExists || cinfo.Error.ErrorCode == JSErrCodeConsumerNameExists) {
 
 				info, err = js.ConsumerInfo(stream, consumer)
 				if err != nil {
@@ -1689,10 +1693,10 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 					hasHeartbeats = info.Config.Heartbeat > 0
 				}
 			} else {
-				if cinfo.Error.Code == 404 {
+				if errors.Is(cinfo.Error, ErrStreamNotFound) {
 					return nil, ErrStreamNotFound
 				}
-				return nil, fmt.Errorf("nats: %s", cinfo.Error.Description)
+				return nil, cinfo.Error
 			}
 		} else {
 			// Since the library created the JS consumer, it will delete it on Unsubscribe()/Drain()
@@ -1970,7 +1974,7 @@ func (sub *Subscription) resetOrderedConsumer(sseq uint64) {
 		}
 
 		if cinfo.Error != nil {
-			pushErr(fmt.Errorf("nats: %s", cinfo.Error.Description))
+			pushErr(cinfo.Error)
 			return
 		}
 
@@ -2185,17 +2189,8 @@ func Description(description string) SubOpt {
 	})
 }
 
-// Check that the durable name is valid, that is, that it does not contain
-// any ".", and if it does return ErrInvalidDurableName, otherwise nil.
-func checkDurName(dur string) error {
-	if strings.Contains(dur, ".") {
-		return ErrInvalidDurableName
-	}
-	return nil
-}
-
 // Durable defines the consumer name for JetStream durable subscribers.
-// This function will return ErrInvalidDurableName in the name contains
+// This function will return ErrInvalidConsumerName in the name contains
 // any dot ".".
 func Durable(consumer string) SubOpt {
 	return subOptFn(func(opts *subOpts) error {
@@ -2205,7 +2200,7 @@ func Durable(consumer string) SubOpt {
 		if opts.consumer != _EMPTY_ && opts.consumer != consumer {
 			return fmt.Errorf("nats: duplicate consumer names (%s and %s)", opts.consumer, consumer)
 		}
-		if err := checkDurName(consumer); err != nil {
+		if err := checkConsumerName(consumer); err != nil {
 			return err
 		}
 
@@ -2445,6 +2440,15 @@ func MaxRequestExpires(max time.Duration) SubOpt {
 	})
 }
 
+// MaxRequesMaxBytes sets the maximum pull consumer request bytes that a
+// Fetch() can receive.
+func MaxRequestMaxBytes(bytes int) SubOpt {
+	return subOptFn(func(opts *subOpts) error {
+		opts.cfg.MaxRequestMaxBytes = bytes
+		return nil
+	})
+}
+
 // InactiveThreshold indicates how long the server should keep an ephemeral
 // after detecting loss of interest.
 func InactiveThreshold(threshold time.Duration) SubOpt {
@@ -2485,8 +2489,9 @@ func (sub *Subscription) ConsumerInfo() (*ConsumerInfo, error) {
 }
 
 type pullOpts struct {
-	ttl time.Duration
-	ctx context.Context
+	maxBytes int
+	ttl      time.Duration
+	ctx      context.Context
 }
 
 // PullOpt are the options that can be passed when pulling a batch of messages.
@@ -2502,6 +2507,14 @@ func PullMaxWaiting(n int) SubOpt {
 	})
 }
 
+// PullMaxBytes defines the max bytes allowed for a fetch request.
+type PullMaxBytes int
+
+func (n PullMaxBytes) configurePull(opts *pullOpts) error {
+	opts.maxBytes = int(n)
+	return nil
+}
+
 var (
 	// errNoMessages is an error that a Fetch request using no_wait can receive to signal
 	// that there are no more messages available.
@@ -2677,6 +2690,7 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 		// For batch real size of 1, it does not make sense to set no_wait in
 		// the request.
 		noWait := batch-len(msgs) > 1
+
 		var nr nextRequest
 
 		sendReq := func() error {
@@ -2701,6 +2715,7 @@ func (sub *Subscription) Fetch(batch int, opts ...PullOpt) ([]*Msg, error) {
 			nr.Batch = batch - len(msgs)
 			nr.Expires = expires
 			nr.NoWait = noWait
+			nr.MaxBytes = o.maxBytes
 			req, _ := json.Marshal(nr)
 			return nc.PublishRequest(nms, rply, req)
 		}
@@ -2757,10 +2772,13 @@ func (js *js) getConsumerInfoContext(ctx context.Context, stream, consumer strin
 		return nil, err
 	}
 	if info.Error != nil {
-		if info.Error.Code == 404 {
+		if errors.Is(info.Error, ErrConsumerNotFound) {
 			return nil, ErrConsumerNotFound
 		}
-		return nil, fmt.Errorf("nats: %s", info.Error.Description)
+		if errors.Is(info.Error, ErrStreamNotFound) {
+			return nil, ErrStreamNotFound
+		}
+		return nil, info.Error
 	}
 	return info.ConsumerInfo, nil
 }
diff --git a/jserrors.go b/jserrors.go
new file mode 100644
index 0000000..a6f2dea
--- /dev/null
+++ b/jserrors.go
@@ -0,0 +1,185 @@
+// Copyright 2020-2022 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 nats
+
+import (
+	"errors"
+	"fmt"
+)
+
+var (
+	// API errors
+
+	// ErrJetStreamNotEnabled is an error returned when JetStream is not enabled for an account.
+	ErrJetStreamNotEnabled JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeJetStreamNotEnabled, Description: "jetstream not enabled", Code: 503}}
+
+	// ErrJetStreamNotEnabledForAccount is an error returned when JetStream is not enabled for an account.
+	ErrJetStreamNotEnabledForAccount JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeJetStreamNotEnabledForAccount, Description: "jetstream not enabled for account", Code: 503}}
+
+	// ErrStreamNotFound is an error returned when stream with given name does not exist.
+	ErrStreamNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeStreamNotFound, Description: "stream not found", Code: 404}}
+
+	// ErrStreamNameAlreadyInUse is returned when a stream with given name already exists and has a different configuration
+	ErrStreamNameAlreadyInUse JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeStreamNameInUse, Description: "stream name already in use", Code: 400}}
+
+	// ErrConsumerNotFound is an error returned when consumer with given name does not exist.
+	ErrConsumerNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerNotFound, Description: "consumer not found", Code: 404}}
+
+	// ErrMsgNotFound is returned when message with provided sequence number does npt exist.
+	ErrMsgNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeMessageNotFound, Description: "message not found", Code: 404}}
+
+	// ErrBadRequest is returned when invalid request is sent to JetStream API.
+	ErrBadRequest JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeBadRequest, Description: "bad request", Code: 400}}
+
+	// Client errors
+
+	// ErrConsumerNotFound is an error returned when consumer with given name does not exist.
+	ErrConsumerNameAlreadyInUse JetStreamError = &jsError{message: "consumer name already in use"}
+
+	// ErrConsumerNotActive is an error returned when consumer is not active.
+	ErrConsumerNotActive JetStreamError = &jsError{message: "consumer not active"}
+
+	// ErrInvalidJSAck is returned when JetStream ack from message publish is invalid.
+	ErrInvalidJSAck JetStreamError = &jsError{message: "invalid jetstream publish response"}
+
+	// ErrStreamConfigRequired is returned when empty stream configuration is supplied to add/update stream.
+	ErrStreamConfigRequired JetStreamError = &jsError{message: "stream configuration is required"}
+
+	// ErrStreamNameRequired is returned when the provided stream name is empty.
+	ErrStreamNameRequired JetStreamError = &jsError{message: "stream name is required"}
+
+	// ErrConsumerNameRequired is returned when the provided consumer durable name is empty,
+	ErrConsumerNameRequired JetStreamError = &jsError{message: "consumer name is required"}
+
+	// ErrConsumerConfigRequired is returned when empty consumer consuguration is supplied to add/update consumer.
+	ErrConsumerConfigRequired JetStreamError = &jsError{message: "consumer configuration is required"}
+
+	// ErrPullSubscribeToPushConsumer is returned when attempting to use PullSubscribe on push consumer.
+	ErrPullSubscribeToPushConsumer JetStreamError = &jsError{message: "cannot pull subscribe to push based consumer"}
+
+	// ErrPullSubscribeRequired is returned when attempting to use subscribe methods not suitable for pull consumers for pull consumers.
+	ErrPullSubscribeRequired JetStreamError = &jsError{message: "must use pull subscribe to bind to pull based consumer"}
+
+	// ErrMsgAlreadyAckd is returned when attempting to acknowledge message more than once.
+	ErrMsgAlreadyAckd JetStreamError = &jsError{message: "message was already acknowledged"}
+
+	// ErrNoStreamResponse is returned when there is no response from stream (e.g. no responders error).
+	ErrNoStreamResponse JetStreamError = &jsError{message: "no response from stream"}
+
+	// ErrNotJSMessage is returned when attempting to get metadata from non JetStream message .
+	ErrNotJSMessage JetStreamError = &jsError{message: "not a jetstream message"}
+
+	// ErrInvalidStreamName is returned when the provided stream name is invalid (contains '.').
+	ErrInvalidStreamName JetStreamError = &jsError{message: "invalid stream name"}
+
+	// ErrInvalidConsumerName is returned when the provided consumer name is invalid (contains '.').
+	ErrInvalidConsumerName JetStreamError = &jsError{message: "invalid consumer name"}
+
+	// ErrNoMatchingStream is returned when stream lookup by subject is unsuccessful.
+	ErrNoMatchingStream JetStreamError = &jsError{message: "no stream matches subject"}
+
+	// ErrSubjectMismatch is returned when the provided subject does not match consumer's filter subject.
+	ErrSubjectMismatch JetStreamError = &jsError{message: "subject does not match consumer"}
+
+	// ErrContextAndTimeout is returned when attempting to use both context and timeout.
+	ErrContextAndTimeout JetStreamError = &jsError{message: "context and timeout can not both be set"}
+
+	// ErrCantAckIfConsumerAckNone is returned when attempting to ack a message for consumer with AckNone policy set.
+	ErrCantAckIfConsumerAckNone JetStreamError = &jsError{message: "cannot acknowledge a message for a consumer with AckNone policy"}
+
+	// DEPRECATED: ErrInvalidDurableName is no longer returned and will be removed in future releases
+	// Use ErrInvalidConsumerName instead
+	ErrInvalidDurableName = errors.New("nats: invalid durable name")
+)
+
+// Error code represents JetStream error codes returned by the API
+type ErrorCode uint16
+
+const (
+	JSErrCodeJetStreamNotEnabledForAccount ErrorCode = 10039
+	JSErrCodeJetStreamNotEnabled           ErrorCode = 10076
+
+	JSErrCodeStreamNotFound  ErrorCode = 10059
+	JSErrCodeStreamNameInUse ErrorCode = 10058
+
+	JSErrCodeConsumerNotFound      ErrorCode = 10014
+	JSErrCodeConsumerNameExists    ErrorCode = 10013
+	JSErrCodeConsumerAlreadyExists ErrorCode = 10105
+
+	JSErrCodeMessageNotFound ErrorCode = 10037
+
+	JSErrCodeBadRequest ErrorCode = 10003
+)
+
+// APIError is included in all API responses if there was an error.
+type APIError struct {
+	Code        int       `json:"code"`
+	ErrorCode   ErrorCode `json:"err_code"`
+	Description string    `json:"description,omitempty"`
+}
+
+// Error prints the JetStream API error code and description
+func (e *APIError) Error() string {
+	return fmt.Sprintf("nats: API error %d: %s", e.ErrorCode, e.Description)
+}
+
+// APIError implements the JetStreamError interface.
+func (e *APIError) APIError() *APIError {
+	return e
+}
+
+// Is matches against an APIError.
+func (e *APIError) Is(err error) bool {
+	if e == nil {
+		return false
+	}
+	// Extract internal APIError to match against.
+	var aerr *APIError
+	ok := errors.As(err, &aerr)
+	if !ok {
+		return ok
+	}
+	return e.ErrorCode == aerr.ErrorCode
+}
+
+// JetStreamError is an error result that happens when using JetStream.
+// In case of client-side error, `APIError()` returns nil
+type JetStreamError interface {
+	APIError() *APIError
+	error
+}
+
+type jsError struct {
+	apiErr  *APIError
+	message string
+}
+
+func (err *jsError) APIError() *APIError {
+	return err.apiErr
+}
+
+func (err *jsError) Error() string {
+	if err.apiErr != nil && err.apiErr.Description != "" {
+		return err.apiErr.Error()
+	}
+	return fmt.Sprintf("nats: %s", err.message)
+}
+
+func (err *jsError) Unwrap() error {
+	// Allow matching to embedded APIError in case there is one.
+	if err.apiErr == nil {
+		return nil
+	}
+	return err.apiErr
+}
diff --git a/jsm.go b/jsm.go
index 8d85450..c043083 100644
--- a/jsm.go
+++ b/jsm.go
@@ -154,17 +154,10 @@ type ExternalStream struct {
 	DeliverPrefix string `json:"deliver"`
 }
 
-// apiError is included in all API responses if there was an error.
-type apiError struct {
-	Code        int    `json:"code"`
-	ErrorCode   int    `json:"err_code"`
-	Description string `json:"description,omitempty"`
-}
-
 // apiResponse is a standard response from the JetStream JSON API
 type apiResponse struct {
 	Type  string    `json:"type"`
-	Error *apiError `json:"error,omitempty"`
+	Error *APIError `json:"error,omitempty"`
 }
 
 // apiPaged includes variables used to create paged responses from the JSON API
@@ -244,13 +237,11 @@ func (js *js) AccountInfo(opts ...JSOpt) (*AccountInfo, error) {
 		return nil, err
 	}
 	if info.Error != nil {
-		var err error
-		if strings.Contains(info.Error.Description, "not enabled for") {
-			err = ErrJetStreamNotEnabled
-		} else {
-			err = errors.New(info.Error.Description)
+		// Internally checks based on error code instead of description match.
+		if errors.Is(info.Error, ErrJetStreamNotEnabledForAccount) {
+			return nil, ErrJetStreamNotEnabledForAccount
 		}
-		return nil, err
+		return nil, info.Error
 	}
 
 	return &info.AccountInfo, nil
@@ -268,6 +259,34 @@ type consumerResponse struct {
 
 // AddConsumer will add a JetStream consumer.
 func (js *js) AddConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error) {
+	if cfg != nil && cfg.Durable != _EMPTY_ {
+		consInfo, err := js.ConsumerInfo(stream, cfg.Durable)
+		if err != nil && !errors.Is(err, ErrConsumerNotFound) && !errors.Is(err, ErrStreamNotFound) {
+			return nil, err
+		}
+
+		if consInfo != nil {
+			sameConfig := checkConfig(&consInfo.Config, cfg)
+			if sameConfig != nil {
+				return nil, fmt.Errorf("%w: creating consumer %q on stream %q", ErrConsumerNameAlreadyInUse, cfg.Durable, stream)
+			}
+		}
+	}
+
+	return js.upsertConsumer(stream, cfg, opts...)
+}
+
+func (js *js) UpdateConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error) {
+	if cfg == nil {
+		return nil, ErrConsumerConfigRequired
+	}
+	if cfg.Durable == _EMPTY_ {
+		return nil, ErrConsumerNameRequired
+	}
+	return js.upsertConsumer(stream, cfg, opts...)
+}
+
+func (js *js) upsertConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error) {
 	if err := checkStreamName(stream); err != nil {
 		return nil, err
 	}
@@ -286,7 +305,7 @@ func (js *js) AddConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*C
 
 	var ccSubj string
 	if cfg != nil && cfg.Durable != _EMPTY_ {
-		if err := checkDurName(cfg.Durable); err != nil {
+		if err := checkConsumerName(cfg.Durable); err != nil {
 			return nil, err
 		}
 		ccSubj = fmt.Sprintf(apiDurableCreateT, stream, cfg.Durable)
@@ -307,30 +326,17 @@ func (js *js) AddConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*C
 		return nil, err
 	}
 	if info.Error != nil {
-		if info.Error.ErrorCode == 10059 {
+		if errors.Is(info.Error, ErrStreamNotFound) {
 			return nil, ErrStreamNotFound
 		}
-		if info.Error.Code == 404 {
+		if errors.Is(info.Error, ErrConsumerNotFound) {
 			return nil, ErrConsumerNotFound
 		}
-		return nil, errors.New(info.Error.Description)
+		return nil, info.Error
 	}
 	return info.ConsumerInfo, nil
 }
 
-func (js *js) UpdateConsumer(stream string, cfg *ConsumerConfig, opts ...JSOpt) (*ConsumerInfo, error) {
-	if err := checkStreamName(stream); err != nil {
-		return nil, err
-	}
-	if cfg == nil {
-		return nil, ErrConsumerConfigRequired
-	}
-	if cfg.Durable == _EMPTY_ {
-		return nil, ErrInvalidDurableName
-	}
-	return js.AddConsumer(stream, cfg, opts...)
-}
-
 // consumerDeleteResponse is the response for a Consumer delete request.
 type consumerDeleteResponse struct {
 	apiResponse
@@ -347,6 +353,8 @@ func checkStreamName(stream string) error {
 	return nil
 }
 
+// Check that the durable name exists and is valid, that is, that it does not contain any "."
+// Returns ErrConsumerNameRequired if consumer name is empty, ErrInvalidConsumerName is invalid, otherwise nil
 func checkConsumerName(consumer string) error {
 	if consumer == _EMPTY_ {
 		return ErrConsumerNameRequired
@@ -384,10 +392,10 @@ func (js *js) DeleteConsumer(stream, consumer string, opts ...JSOpt) error {
 	}
 
 	if resp.Error != nil {
-		if resp.Error.Code == 404 {
+		if errors.Is(resp.Error, ErrConsumerNotFound) {
 			return ErrConsumerNotFound
 		}
-		return errors.New(resp.Error.Description)
+		return resp.Error
 	}
 	return nil
 }
@@ -474,7 +482,7 @@ func (c *consumerLister) Next() bool {
 		return false
 	}
 	if resp.Error != nil {
-		c.err = errors.New(resp.Error.Description)
+		c.err = resp.Error
 		return false
 	}
 
@@ -571,7 +579,7 @@ func (c *consumerNamesLister) Next() bool {
 		return false
 	}
 	if resp.Error != nil {
-		c.err = errors.New(resp.Error.Description)
+		c.err = resp.Error
 		return false
 	}
 
@@ -655,10 +663,10 @@ func (js *js) AddStream(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error) {
 		return nil, err
 	}
 	if resp.Error != nil {
-		if resp.Error.ErrorCode == 10058 {
+		if errors.Is(resp.Error, ErrStreamNameAlreadyInUse) {
 			return nil, ErrStreamNameAlreadyInUse
 		}
-		return nil, errors.New(resp.Error.Description)
+		return nil, resp.Error
 	}
 
 	return resp.StreamInfo, nil
@@ -703,10 +711,10 @@ func (js *js) StreamInfo(stream string, opts ...JSOpt) (*StreamInfo, error) {
 		return nil, err
 	}
 	if resp.Error != nil {
-		if resp.Error.Code == 404 {
+		if errors.Is(resp.Error, ErrStreamNotFound) {
 			return nil, ErrStreamNotFound
 		}
-		return nil, fmt.Errorf("nats: %s", resp.Error.Description)
+		return nil, resp.Error
 	}
 
 	return resp.StreamInfo, nil
@@ -724,9 +732,11 @@ type StreamInfo struct {
 
 // StreamSourceInfo shows information about an upstream stream source.
 type StreamSourceInfo struct {
-	Name   string        `json:"name"`
-	Lag    uint64        `json:"lag"`
-	Active time.Duration `json:"active"`
+	Name     string          `json:"name"`
+	Lag      uint64          `json:"lag"`
+	Active   time.Duration   `json:"active"`
+	External *ExternalStream `json:"external"`
+	Error    *APIError       `json:"error"`
 }
 
 // StreamState is information about the given stream.
@@ -793,7 +803,10 @@ func (js *js) UpdateStream(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error
 		return nil, err
 	}
 	if resp.Error != nil {
-		return nil, errors.New(resp.Error.Description)
+		if errors.Is(resp.Error, ErrStreamNotFound) {
+			return nil, ErrStreamNotFound
+		}
+		return nil, resp.Error
 	}
 	return resp.StreamInfo, nil
 }
@@ -828,10 +841,10 @@ func (js *js) DeleteStream(name string, opts ...JSOpt) error {
 	}
 
 	if resp.Error != nil {
-		if resp.Error.Code == 404 {
+		if errors.Is(resp.Error, ErrStreamNotFound) {
 			return ErrStreamNotFound
 		}
-		return errors.New(resp.Error.Description)
+		return resp.Error
 	}
 	return nil
 }
@@ -932,10 +945,13 @@ func (js *js) getMsg(name string, mreq *apiMsgGetRequest, opts ...JSOpt) (*RawSt
 		return nil, err
 	}
 	if resp.Error != nil {
-		if resp.Error.Code == 404 && strings.Contains(resp.Error.Description, "message") {
+		if errors.Is(resp.Error, ErrMsgNotFound) {
 			return nil, ErrMsgNotFound
 		}
-		return nil, fmt.Errorf("nats: %s", resp.Error.Description)
+		if errors.Is(resp.Error, ErrStreamNotFound) {
+			return nil, ErrStreamNotFound
+		}
+		return nil, resp.Error
 	}
 
 	msg := resp.Message
@@ -998,9 +1014,16 @@ func convertDirectGetMsgResponseToMsg(name string, r *Msg) (*RawStreamMsg, error
 	if timeStr == _EMPTY_ {
 		return nil, fmt.Errorf("nats: missing timestamp header")
 	}
-	tm, err := time.Parse("2006-01-02 15:04:05.999999999 +0000 UTC", timeStr)
+	// Temporary code: the server in main branch is sending with format
+	// "2006-01-02 15:04:05.999999999 +0000 UTC", but will be changed
+	// to use format RFC3339Nano. Because of server test deps/cycle,
+	// support both until the server PR lands.
+	tm, err := time.Parse(time.RFC3339Nano, timeStr)
 	if err != nil {
-		return nil, fmt.Errorf("nats: invalid timestamp header '%s': %v", timeStr, err)
+		tm, err = time.Parse("2006-01-02 15:04:05.999999999 +0000 UTC", timeStr)
+		if err != nil {
+			return nil, fmt.Errorf("nats: invalid timestamp header '%s': %v", timeStr, err)
+		}
 	}
 	subj := r.Header.Get(JSSubject)
 	if subj == _EMPTY_ {
@@ -1073,7 +1096,7 @@ func (js *js) deleteMsg(ctx context.Context, stream string, req *msgDeleteReques
 		return err
 	}
 	if resp.Error != nil {
-		return errors.New(resp.Error.Description)
+		return resp.Error
 	}
 	return nil
 }
@@ -1136,10 +1159,10 @@ func (js *js) purgeStream(stream string, req *StreamPurgeRequest, opts ...JSOpt)
 		return err
 	}
 	if resp.Error != nil {
-		if resp.Error.Code == 400 {
+		if errors.Is(resp.Error, ErrBadRequest) {
 			return fmt.Errorf("%w: %s", ErrBadRequest, "invalid purge request body")
 		}
-		return errors.New(resp.Error.Description)
+		return resp.Error
 	}
 	return nil
 }
@@ -1206,7 +1229,7 @@ func (s *streamLister) Next() bool {
 		return false
 	}
 	if resp.Error != nil {
-		s.err = errors.New(resp.Error.Description)
+		s.err = resp.Error
 		return false
 	}
 
@@ -1290,7 +1313,7 @@ func (l *streamNamesLister) Next() bool {
 		return false
 	}
 	if resp.Error != nil {
-		l.err = errors.New(resp.Error.Description)
+		l.err = resp.Error
 		return false
 	}
 
diff --git a/nats.go b/nats.go
index bea0d3d..3143b9a 100644
--- a/nats.go
+++ b/nats.go
@@ -90,86 +90,55 @@ const (
 
 // Errors
 var (
-	ErrConnectionClosed             = errors.New("nats: connection closed")
-	ErrConnectionDraining           = errors.New("nats: connection draining")
-	ErrDrainTimeout                 = errors.New("nats: draining connection timed out")
-	ErrConnectionReconnecting       = errors.New("nats: connection reconnecting")
-	ErrSecureConnRequired           = errors.New("nats: secure connection required")
-	ErrSecureConnWanted             = errors.New("nats: secure connection not available")
-	ErrBadSubscription              = errors.New("nats: invalid subscription")
-	ErrTypeSubscription             = errors.New("nats: invalid subscription type")
-	ErrBadSubject                   = errors.New("nats: invalid subject")
-	ErrBadQueueName                 = errors.New("nats: invalid queue name")
-	ErrSlowConsumer                 = errors.New("nats: slow consumer, messages dropped")
-	ErrTimeout                      = errors.New("nats: timeout")
-	ErrBadTimeout                   = errors.New("nats: timeout invalid")
-	ErrAuthorization                = errors.New("nats: authorization violation")
-	ErrAuthExpired                  = errors.New("nats: authentication expired")
-	ErrAuthRevoked                  = errors.New("nats: authentication revoked")
-	ErrAccountAuthExpired           = errors.New("nats: account authentication expired")
-	ErrNoServers                    = errors.New("nats: no servers available for connection")
-	ErrJsonParse                    = errors.New("nats: connect message, json parse error")
-	ErrChanArg                      = errors.New("nats: argument needs to be a channel type")
-	ErrMaxPayload                   = errors.New("nats: maximum payload exceeded")
-	ErrMaxMessages                  = errors.New("nats: maximum messages delivered")
-	ErrSyncSubRequired              = errors.New("nats: illegal call on an async subscription")
-	ErrMultipleTLSConfigs           = errors.New("nats: multiple tls.Configs not allowed")
-	ErrNoInfoReceived               = errors.New("nats: protocol exception, INFO not received")
-	ErrReconnectBufExceeded         = errors.New("nats: outbound buffer limit exceeded")
-	ErrInvalidConnection            = errors.New("nats: invalid connection")
-	ErrInvalidMsg                   = errors.New("nats: invalid message or message nil")
-	ErrInvalidArg                   = errors.New("nats: invalid argument")
-	ErrInvalidContext               = errors.New("nats: invalid context")
-	ErrNoDeadlineContext            = errors.New("nats: context requires a deadline")
-	ErrNoEchoNotSupported           = errors.New("nats: no echo option not supported by this server")
-	ErrClientIDNotSupported         = errors.New("nats: client ID not supported by this server")
-	ErrUserButNoSigCB               = errors.New("nats: user callback defined without a signature handler")
-	ErrNkeyButNoSigCB               = errors.New("nats: nkey defined without a signature handler")
-	ErrNoUserCB                     = errors.New("nats: user callback not defined")
-	ErrNkeyAndUser                  = errors.New("nats: user callback and nkey defined")
-	ErrNkeysNotSupported            = errors.New("nats: nkeys not supported by the server")
-	ErrStaleConnection              = errors.New("nats: " + STALE_CONNECTION)
-	ErrTokenAlreadySet              = errors.New("nats: token and token handler both set")
-	ErrMsgNotBound                  = errors.New("nats: message is not bound to subscription/connection")
-	ErrMsgNoReply                   = errors.New("nats: message does not have a reply")
-	ErrClientIPNotSupported         = errors.New("nats: client IP not supported by this server")
-	ErrDisconnected                 = errors.New("nats: server is disconnected")
-	ErrHeadersNotSupported          = errors.New("nats: headers not supported by this server")
-	ErrBadHeaderMsg                 = errors.New("nats: message could not decode headers")
-	ErrNoResponders                 = errors.New("nats: no responders available for request")
-	ErrNoContextOrTimeout           = errors.New("nats: no context or timeout given")
-	ErrPullModeNotAllowed           = errors.New("nats: pull based not supported")
-	ErrJetStreamNotEnabled          = errors.New("nats: jetstream not enabled")
-	ErrJetStreamBadPre              = errors.New("nats: jetstream api prefix not valid")
-	ErrNoStreamResponse             = errors.New("nats: no response from stream")
-	ErrNotJSMessage                 = errors.New("nats: not a jetstream message")
-	ErrInvalidStreamName            = errors.New("nats: invalid stream name")
-	ErrInvalidDurableName           = errors.New("nats: invalid durable name")
-	ErrInvalidConsumerName          = errors.New("nats: invalid consumer name")
-	ErrNoMatchingStream             = errors.New("nats: no stream matches subject")
-	ErrSubjectMismatch              = errors.New("nats: subject does not match consumer")
-	ErrContextAndTimeout            = errors.New("nats: context and timeout can not both be set")
-	ErrInvalidJSAck                 = errors.New("nats: invalid jetstream publish response")
-	ErrMultiStreamUnsupported       = errors.New("nats: multiple streams are not supported")
-	ErrStreamConfigRequired         = errors.New("nats: stream configuration is required")
-	ErrStreamNameRequired           = errors.New("nats: stream name is required")
-	ErrStreamNotFound               = errors.New("nats: stream not found")
-	ErrConsumerNotFound             = errors.New("nats: consumer not found")
-	ErrConsumerNameRequired         = errors.New("nats: consumer name is required")
-	ErrConsumerConfigRequired       = errors.New("nats: consumer configuration is required")
-	ErrStreamSnapshotConfigRequired = errors.New("nats: stream snapshot configuration is required")
-	ErrDeliverSubjectRequired       = errors.New("nats: deliver subject is required")
-	ErrPullSubscribeToPushConsumer  = errors.New("nats: cannot pull subscribe to push based consumer")
-	ErrPullSubscribeRequired        = errors.New("nats: must use pull subscribe to bind to pull based consumer")
-	ErrConsumerNotActive            = errors.New("nats: consumer not active")
-	ErrMsgNotFound                  = errors.New("nats: message not found")
-	ErrMsgAlreadyAckd               = errors.New("nats: message was already acknowledged")
-	ErrCantAckIfConsumerAckNone     = errors.New("nats: cannot acknowledge a message for a consumer with AckNone policy")
-	ErrStreamInfoMaxSubjects        = errors.New("nats: subject details would exceed maximum allowed")
-	ErrStreamNameAlreadyInUse       = errors.New("nats: stream name already in use")
-	ErrMaxConnectionsExceeded       = errors.New("nats: server maximum connections exceeded")
-	ErrBadRequest                   = errors.New("nats: bad request")
-	ErrConnectionNotTLS             = errors.New("nats: connection is not tls")
+	ErrConnectionClosed       = errors.New("nats: connection closed")
+	ErrConnectionDraining     = errors.New("nats: connection draining")
+	ErrDrainTimeout           = errors.New("nats: draining connection timed out")
+	ErrConnectionReconnecting = errors.New("nats: connection reconnecting")
+	ErrSecureConnRequired     = errors.New("nats: secure connection required")
+	ErrSecureConnWanted       = errors.New("nats: secure connection not available")
+	ErrBadSubscription        = errors.New("nats: invalid subscription")
+	ErrTypeSubscription       = errors.New("nats: invalid subscription type")
+	ErrBadSubject             = errors.New("nats: invalid subject")
+	ErrBadQueueName           = errors.New("nats: invalid queue name")
+	ErrSlowConsumer           = errors.New("nats: slow consumer, messages dropped")
+	ErrTimeout                = errors.New("nats: timeout")
+	ErrBadTimeout             = errors.New("nats: timeout invalid")
+	ErrAuthorization          = errors.New("nats: authorization violation")
+	ErrAuthExpired            = errors.New("nats: authentication expired")
+	ErrAuthRevoked            = errors.New("nats: authentication revoked")
+	ErrAccountAuthExpired     = errors.New("nats: account authentication expired")
+	ErrNoServers              = errors.New("nats: no servers available for connection")
+	ErrJsonParse              = errors.New("nats: connect message, json parse error")
+	ErrChanArg                = errors.New("nats: argument needs to be a channel type")
+	ErrMaxPayload             = errors.New("nats: maximum payload exceeded")
+	ErrMaxMessages            = errors.New("nats: maximum messages delivered")
+	ErrSyncSubRequired        = errors.New("nats: illegal call on an async subscription")
+	ErrMultipleTLSConfigs     = errors.New("nats: multiple tls.Configs not allowed")
+	ErrNoInfoReceived         = errors.New("nats: protocol exception, INFO not received")
+	ErrReconnectBufExceeded   = errors.New("nats: outbound buffer limit exceeded")
+	ErrInvalidConnection      = errors.New("nats: invalid connection")
+	ErrInvalidMsg             = errors.New("nats: invalid message or message nil")
+	ErrInvalidArg             = errors.New("nats: invalid argument")
+	ErrInvalidContext         = errors.New("nats: invalid context")
+	ErrNoDeadlineContext      = errors.New("nats: context requires a deadline")
+	ErrNoEchoNotSupported     = errors.New("nats: no echo option not supported by this server")
+	ErrClientIDNotSupported   = errors.New("nats: client ID not supported by this server")
+	ErrUserButNoSigCB         = errors.New("nats: user callback defined without a signature handler")
+	ErrNkeyButNoSigCB         = errors.New("nats: nkey defined without a signature handler")
+	ErrNoUserCB               = errors.New("nats: user callback not defined")
+	ErrNkeyAndUser            = errors.New("nats: user callback and nkey defined")
+	ErrNkeysNotSupported      = errors.New("nats: nkeys not supported by the server")
+	ErrStaleConnection        = errors.New("nats: " + STALE_CONNECTION)
+	ErrTokenAlreadySet        = errors.New("nats: token and token handler both set")
+	ErrMsgNotBound            = errors.New("nats: message is not bound to subscription/connection")
+	ErrMsgNoReply             = errors.New("nats: message does not have a reply")
+	ErrClientIPNotSupported   = errors.New("nats: client IP not supported by this server")
+	ErrDisconnected           = errors.New("nats: server is disconnected")
+	ErrHeadersNotSupported    = errors.New("nats: headers not supported by this server")
+	ErrBadHeaderMsg           = errors.New("nats: message could not decode headers")
+	ErrNoResponders           = errors.New("nats: no responders available for request")
+	ErrMaxConnectionsExceeded = errors.New("nats: server maximum connections exceeded")
+	ErrConnectionNotTLS       = errors.New("nats: connection is not tls")
 )
 
 func init() {
@@ -634,22 +603,21 @@ type Subscription struct {
 // Msg represents a message delivered by NATS. This structure is used
 // by Subscribers and PublishMsg().
 //
-// Types of Acknowledgements
+// # Types of Acknowledgements
 //
 // In case using JetStream, there are multiple ways to ack a Msg:
 //
-//   // Acknowledgement that a message has been processed.
-//   msg.Ack()
+//	// Acknowledgement that a message has been processed.
+//	msg.Ack()
 //
-//   // Negatively acknowledges a message.
-//   msg.Nak()
+//	// Negatively acknowledges a message.
+//	msg.Nak()
 //
-//   // Terminate a message so that it is not redelivered further.
-//   msg.Term()
-//
-//   // Signal the server that the message is being worked on and reset redelivery timer.
-//   msg.InProgress()
+//	// Terminate a message so that it is not redelivered further.
+//	msg.Term()
 //
+//	// Signal the server that the message is being worked on and reset redelivery timer.
+//	msg.InProgress()
 type Msg struct {
 	Subject string
 	Reply   string
@@ -1083,6 +1051,28 @@ func UserCredentials(userOrChainedFile string, seedFiles ...string) Option {
 	return UserJWT(userCB, sigCB)
 }
 
+// UserJWTAndSeed is a convenience function that takes the JWT and seed
+// values as strings.
+func UserJWTAndSeed(jwt string, seed string) Option {
+	userCB := func() (string, error) {
+		return jwt, nil
+	}
+
+	sigCB := func(nonce []byte) ([]byte, error) {
+		kp, err := nkeys.FromSeed([]byte(seed))
+		if err != nil {
+			return nil, fmt.Errorf("unable to extract key pair from seed: %v", err)
+		}
+		// Wipe our key on exit.
+		defer kp.Wipe()
+
+		sig, _ := kp.Sign(nonce)
+		return sig, nil
+	}
+
+	return UserJWT(userCB, sigCB)
+}
+
 // UserJWT will set the callbacks to retrieve the user's JWT and
 // the signature callback to sign the server nonce. This an the Nkey
 // option are mutually exclusive.
diff --git a/object.go b/object.go
index d2124bb..11376fa 100644
--- a/object.go
+++ b/object.go
@@ -37,7 +37,7 @@ import (
 //
 // This functionality is EXPERIMENTAL and may be changed in later releases.
 type ObjectStoreManager interface {
-	// ObjectStore will lookup and bind to an existing object store instance.
+	// ObjectStore will look up and bind to an existing object store instance.
 	ObjectStore(bucket string) (ObjectStore, error)
 	// CreateObjectStore will create an object store.
 	CreateObjectStore(cfg *ObjectStoreConfig) (ObjectStore, error)
@@ -62,7 +62,7 @@ type ObjectStore interface {
 	// GetBytes is a convenience function to pull an object from this object store and return it as a byte slice.
 	GetBytes(name string, opts ...ObjectOpt) ([]byte, error)
 
-	// PutBytes is convenience function to put a string into this object store.
+	// PutString is convenience function to put a string into this object store.
 	PutString(name string, data string, opts ...ObjectOpt) (*ObjectInfo, error)
 	// GetString is a convenience function to pull an object from this object store and return it as a string.
 	GetString(name string, opts ...ObjectOpt) (string, error)
@@ -80,7 +80,7 @@ type ObjectStore interface {
 	// Delete will delete the named object.
 	Delete(name string) error
 
-	// AddLink will add a link to another object into this object store.
+	// AddLink will add a link to another object.
 	AddLink(name string, obj *ObjectInfo) (*ObjectInfo, error)
 
 	// AddBucketLink will add a link to another object store.
@@ -126,9 +126,11 @@ var (
 	ErrBadObjectMeta        = errors.New("nats: object-store meta information invalid")
 	ErrObjectNotFound       = errors.New("nats: object not found")
 	ErrInvalidStoreName     = errors.New("nats: invalid object-store name")
-	ErrInvalidObjectName    = errors.New("nats: invalid object name")
 	ErrDigestMismatch       = errors.New("nats: received a corrupt object, digests do not match")
 	ErrNoObjectsFound       = errors.New("nats: no objects found")
+	ErrObjectAlreadyExists  = errors.New("nats: an object already exists with that name")
+	ErrNameRequired         = errors.New("nats: name is required")
+	ErrNeeds262             = errors.New("nats: object-store requires at least server version 2.6.2")
 )
 
 // ObjectStoreConfig is the config for the object store.
@@ -206,15 +208,14 @@ type ObjectResult interface {
 }
 
 const (
-	objNameTmpl         = "OBJ_%s"
-	objSubjectsPre      = "$O."
-	objAllChunksPreTmpl = "$O.%s.C.>"
-	objAllMetaPreTmpl   = "$O.%s.M.>"
-	objChunksPreTmpl    = "$O.%s.C.%s"
-	objMetaPreTmpl      = "$O.%s.M.%s"
+	objNameTmpl         = "OBJ_%s"     // OBJ_<bucket> // stream name
+	objAllChunksPreTmpl = "$O.%s.C.>"  // $O.<bucket>.C.> // chunk stream subject
+	objAllMetaPreTmpl   = "$O.%s.M.>"  // $O.<bucket>.M.> // meta stream subject
+	objChunksPreTmpl    = "$O.%s.C.%s" // $O.<bucket>.C.<object-nuid> // chunk message subject
+	objMetaPreTmpl      = "$O.%s.M.%s" // $O.<bucket>.M.<name-encoded> // meta message subject
 	objNoPending        = "0"
 	objDefaultChunkSize = uint32(128 * 1024) // 128k
-	objDigestType       = "sha-256="
+	objDigestType       = "SHA-256="
 	objDigestTmpl       = objDigestType + "%s"
 )
 
@@ -227,7 +228,7 @@ type obs struct {
 // CreateObjectStore will create an object store.
 func (js *js) CreateObjectStore(cfg *ObjectStoreConfig) (ObjectStore, error) {
 	if !js.nc.serverMinVersion(2, 6, 2) {
-		return nil, errors.New("nats: object-store requires at least server version 2.6.2")
+		return nil, ErrNeeds262
 	}
 	if cfg == nil {
 		return nil, ErrObjectConfigRequired
@@ -240,17 +241,30 @@ func (js *js) CreateObjectStore(cfg *ObjectStoreConfig) (ObjectStore, error) {
 	chunks := fmt.Sprintf(objAllChunksPreTmpl, name)
 	meta := fmt.Sprintf(objAllMetaPreTmpl, name)
 
+	// 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.
+	// See kv
+	replicas := cfg.Replicas
+	if replicas == 0 {
+		replicas = 1
+	}
+	maxBytes := cfg.MaxBytes
+	if maxBytes == 0 {
+		maxBytes = -1
+	}
+
 	scfg := &StreamConfig{
 		Name:        fmt.Sprintf(objNameTmpl, name),
 		Description: cfg.Description,
 		Subjects:    []string{chunks, meta},
 		MaxAge:      cfg.TTL,
-		MaxBytes:    cfg.MaxBytes,
+		MaxBytes:    maxBytes,
 		Storage:     cfg.Storage,
-		Replicas:    cfg.Replicas,
+		Replicas:    replicas,
 		Placement:   cfg.Placement,
 		Discard:     DiscardNew,
 		AllowRollup: true,
+		AllowDirect: true,
 	}
 
 	// Create our stream.
@@ -262,13 +276,13 @@ func (js *js) CreateObjectStore(cfg *ObjectStoreConfig) (ObjectStore, error) {
 	return &obs{name: name, stream: scfg.Name, js: js}, nil
 }
 
-// ObjectStore will lookup and bind to an existing object store instance.
+// ObjectStore will look up and bind to an existing object store instance.
 func (js *js) ObjectStore(bucket string) (ObjectStore, error) {
 	if !validBucketRe.MatchString(bucket) {
 		return nil, ErrInvalidStoreName
 	}
 	if !js.nc.serverMinVersion(2, 6, 2) {
-		return nil, errors.New("nats: key-value requires at least server version 2.6.2")
+		return nil, ErrNeeds262
 	}
 
 	stream := fmt.Sprintf(objNameTmpl, bucket)
@@ -285,22 +299,16 @@ func (js *js) DeleteObjectStore(bucket string) error {
 	return js.DeleteStream(stream)
 }
 
-func sanitizeName(name string) string {
-	stream := strings.ReplaceAll(name, ".", "_")
-	return strings.ReplaceAll(stream, " ", "_")
+func encodeName(name string) string {
+	return base64.URLEncoding.EncodeToString([]byte(name))
 }
 
 // Put will place the contents from the reader into this object-store.
 func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectInfo, error) {
-	if meta == nil {
+	if meta == nil || meta.Name == "" {
 		return nil, ErrBadObjectMeta
 	}
 
-	obj := sanitizeName(meta.Name)
-	if !keyValid(obj) {
-		return nil, ErrInvalidObjectName
-	}
-
 	var o objOpts
 	for _, opt := range opts {
 		if opt != nil {
@@ -311,17 +319,19 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 	}
 	ctx := o.ctx
 
-	// Grab existing meta info.
-	einfo, err := obs.GetInfo(meta.Name)
+	// Create the new nuid so chunks go on a new subject if the name is re-used
+	newnuid := nuid.Next()
+
+	// These will be used in more than one place
+	chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, newnuid)
+
+	// Grab existing meta info (einfo). Ok to be found or not found, any other error is a problem
+	// Chunks on the old nuid can be cleaned up at the end
+	einfo, err := obs.GetInfo(meta.Name) // GetInfo will encode the name
 	if err != nil && err != ErrObjectNotFound {
 		return nil, err
 	}
 
-	// Create a random subject prefixed with the object stream name.
-	id := nuid.Next()
-	chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, id)
-	metaSubj := fmt.Sprintf(objMetaPreTmpl, obs.name, obj)
-
 	// For async error handling
 	var perr error
 	var mu sync.Mutex
@@ -351,7 +361,9 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 
 	m, h := NewMsg(chunkSubj), sha256.New()
 	chunk, sent, total := make([]byte, chunkSize), 0, uint64(0)
-	info := &ObjectInfo{Bucket: obs.name, NUID: id, ObjectMeta: *meta}
+
+	// set up the info object. The chunk upload sets the size and digest
+	info := &ObjectInfo{Bucket: obs.name, NUID: newnuid, ObjectMeta: *meta}
 
 	for r != nil {
 		if ctx != nil {
@@ -411,7 +423,8 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 		}
 	}
 
-	// Publish the metadata.
+	// Prepare the meta message
+	metaSubj := fmt.Sprintf(objMetaPreTmpl, obs.name, encodeName(meta.Name))
 	mm := NewMsg(metaSubj)
 	mm.Header.Set(MsgRollup, MsgRollupSubject)
 	mm.Data, err = json.Marshal(info)
@@ -421,7 +434,8 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 		}
 		return nil, err
 	}
-	// Send meta message.
+
+	// Publish the meta message.
 	_, err = js.PublishMsgAsync(mm)
 	if err != nil {
 		if r != nil {
@@ -434,20 +448,28 @@ func (obs *obs) Put(meta *ObjectMeta, r io.Reader, opts ...ObjectOpt) (*ObjectIn
 	select {
 	case <-js.PublishAsyncComplete():
 		if err := getErr(); err != nil {
-			purgePartial()
+			if r != nil {
+				purgePartial()
+			}
 			return nil, err
 		}
 	case <-time.After(obs.js.opts.wait):
 		return nil, ErrTimeout
 	}
-	info.ModTime = time.Now().UTC()
 
-	// Delete any original one.
+	info.ModTime = time.Now().UTC() // This time is not actually the correct time
+
+	// Delete any original chunks.
 	if einfo != nil && !einfo.Deleted {
-		chunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, einfo.NUID)
-		obs.js.purgeStream(obs.stream, &StreamPurgeRequest{Subject: chunkSubj})
+		echunkSubj := fmt.Sprintf(objChunksPreTmpl, obs.name, einfo.NUID)
+		obs.js.purgeStream(obs.stream, &StreamPurgeRequest{Subject: echunkSubj})
 	}
 
+	// TODO would it be okay to do this to return the info with the correct time?
+	// With the understanding that it is an extra call to the server.
+	// Otherwise the time the user gets back is the client time, not the server time.
+	// return obs.GetInfo(info.Name)
+
 	return info, nil
 }
 
@@ -475,12 +497,20 @@ func (obs *obs) Get(name string, opts ...ObjectOpt) (ObjectResult, error) {
 		return nil, ErrBadObjectMeta
 	}
 
-	// Check for object links.If single objects we do a pass through.
+	// Check for object links. If single objects we do a pass through.
 	if info.isLink() {
 		if info.ObjectMeta.Opts.Link.Name == _EMPTY_ {
-			return nil, errors.New("nats: link is a bucket")
+			return nil, errors.New("nats: object is a link to a bucket")
+		}
+
+		// is the link in the same bucket?
+		lbuck := info.ObjectMeta.Opts.Link.Bucket
+		if lbuck == obs.name {
+			return obs.Get(info.ObjectMeta.Opts.Link.Name)
 		}
-		lobs, err := obs.js.ObjectStore(info.ObjectMeta.Opts.Link.Bucket)
+
+		// different bucket
+		lobs, err := obs.js.ObjectStore(lbuck)
 		if err != nil {
 			return nil, err
 		}
@@ -588,11 +618,11 @@ func (obs *obs) Delete(name string) error {
 		return ErrBadObjectMeta
 	}
 
-	// Place a rollup delete marker.
+	// Place a rollup delete marker and publish the info
 	info.Deleted = true
 	info.Size, info.Chunks, info.Digest = 0, 0, _EMPTY_
 
-	metaSubj := fmt.Sprintf(objMetaPreTmpl, obs.name, sanitizeName(name))
+	metaSubj := fmt.Sprintf(objMetaPreTmpl, obs.name, encodeName(name))
 	mm := NewMsg(metaSubj)
 	mm.Data, err = json.Marshal(info)
 	if err != nil {
@@ -609,55 +639,77 @@ func (obs *obs) Delete(name string) error {
 	return obs.js.purgeStream(obs.stream, &StreamPurgeRequest{Subject: chunkSubj})
 }
 
-// AddLink will add a link to another object into this object store.
+// AddLink will add a link to another object if it's not deleted and not another link
+// name is the name of this link object
+// obj is what is being linked too
 func (obs *obs) AddLink(name string, obj *ObjectInfo) (*ObjectInfo, error) {
-	if obj == nil {
+	if name == "" {
+		return nil, ErrNameRequired
+	}
+	if obj == nil || obj.Name == "" {
 		return nil, errors.New("nats: object required")
 	}
 	if obj.Deleted {
-		return nil, errors.New("nats: object is deleted")
+		return nil, errors.New("nats: not allowed to link to a deleted object")
 	}
-	name = sanitizeName(name)
-	if !keyValid(name) {
-		return nil, ErrInvalidObjectName
+	if obj.isLink() {
+		return nil, errors.New("nats: not allowed to link to another link")
 	}
 
-	// Same object store.
-	if obj.Bucket == obs.name {
-		info := *obj
-		info.Name = name
-		if err := obs.UpdateMeta(obj.Name, &info.ObjectMeta); err != nil {
-			return nil, err
+	// If object with link's name is found, error.
+	// If link with link's name is found, that's okay to overwrite.
+	// If there was an error that was not ErrObjectNotFound, error.
+	einfo, err := obs.GetInfo(name)
+	if einfo != nil {
+		if !einfo.isLink() {
+			return nil, ErrObjectAlreadyExists
 		}
-		return obs.GetInfo(name)
+	} else if err != ErrObjectNotFound {
+		return nil, err
 	}
 
-	link := &ObjectLink{Bucket: obj.Bucket, Name: obj.Name}
+	// create the meta for the link
 	meta := &ObjectMeta{
 		Name: name,
-		Opts: &ObjectMetaOptions{Link: link},
+		Opts: &ObjectMetaOptions{Link: &ObjectLink{Bucket: obj.Bucket, Name: obj.Name}},
 	}
+
+	// put the link object
 	return obs.Put(meta, nil)
 }
 
 // AddBucketLink will add a link to another object store.
 func (ob *obs) AddBucketLink(name string, bucket ObjectStore) (*ObjectInfo, error) {
+	if name == "" {
+		return nil, ErrNameRequired
+	}
 	if bucket == nil {
 		return nil, errors.New("nats: bucket required")
 	}
-	name = sanitizeName(name)
-	if !keyValid(name) {
-		return nil, ErrInvalidObjectName
-	}
-
 	bos, ok := bucket.(*obs)
 	if !ok {
 		return nil, errors.New("nats: bucket malformed")
 	}
+
+	// If object with link's name is found, error.
+	// If link with link's name is found, that's okay to overwrite.
+	// If there was an error that was not ErrObjectNotFound, error.
+	einfo, err := ob.GetInfo(name)
+	if einfo != nil {
+		if !einfo.isLink() {
+			return nil, ErrObjectAlreadyExists
+		}
+	} else if err != ErrObjectNotFound {
+		return nil, err
+	}
+
+	// create the meta for the link
 	meta := &ObjectMeta{
 		Name: name,
 		Opts: &ObjectMetaOptions{Link: &ObjectLink{Bucket: bos.name}},
 	}
+
+	// put the link object
 	return ob.Put(meta, nil)
 }
 
@@ -681,7 +733,7 @@ func (obs *obs) GetBytes(name string, opts ...ObjectOpt) ([]byte, error) {
 	return b.Bytes(), nil
 }
 
-// PutBytes is convenience function to put a string into this object store.
+// PutString is convenience function to put a string into this object store.
 func (obs *obs) PutString(name string, data string, opts ...ObjectOpt) (*ObjectInfo, error) {
 	return obs.Put(&ObjectMeta{Name: name}, strings.NewReader(data), opts...)
 }
@@ -734,17 +786,15 @@ func (obs *obs) GetFile(name, file string, opts ...ObjectOpt) error {
 
 // GetInfo will retrieve the current information for the object.
 func (obs *obs) GetInfo(name string) (*ObjectInfo, error) {
-	// Lookup the stream to get the bound subject.
-	obj := sanitizeName(name)
-	if !keyValid(obj) {
-		return nil, ErrInvalidObjectName
+	// Grab last meta value we have.
+	if name == "" {
+		return nil, ErrNameRequired
 	}
 
-	// Grab last meta value we have.
-	meta := fmt.Sprintf(objMetaPreTmpl, obs.name, obj)
+	metaSubj := fmt.Sprintf(objMetaPreTmpl, obs.name, encodeName(name)) // used as data in a JS API call
 	stream := fmt.Sprintf(objNameTmpl, obs.name)
 
-	m, err := obs.js.GetLastMsg(stream, meta)
+	m, err := obs.js.GetLastMsg(stream, metaSubj)
 	if err != nil {
 		if err == ErrMsgNotFound {
 			err = ErrObjectNotFound
@@ -759,25 +809,64 @@ func (obs *obs) GetInfo(name string) (*ObjectInfo, error) {
 	return &info, nil
 }
 
-// UpdateMeta will update the meta data for the object.
+// UpdateMeta will update the meta for the object.
 func (obs *obs) UpdateMeta(name string, meta *ObjectMeta) error {
 	if meta == nil {
 		return ErrBadObjectMeta
 	}
-	// Grab meta info.
+
+	// Grab the current meta.
 	info, err := obs.GetInfo(name)
 	if err != nil {
 		return err
 	}
-	// Copy new meta
-	info.ObjectMeta = *meta
-	mm := NewMsg(fmt.Sprintf(objMetaPreTmpl, obs.name, sanitizeName(meta.Name)))
+
+	if info.Deleted {
+		return errors.New("nats: cannot update meta for a deleted object")
+	}
+
+	// If the new name is different from the old, and it exists, error
+	// If there was an error that was not ErrObjectNotFound, error.
+	// sff - Is there a better go way to do this?
+	if name != meta.Name {
+		_, err = obs.GetInfo(meta.Name)
+		if err != ErrObjectNotFound {
+			if err == nil {
+				return ErrObjectAlreadyExists
+			}
+			return err
+		}
+	}
+
+	// Update Meta prevents update of ObjectMetaOptions (Link, ChunkSize)
+	// These should only be updated internally when appropriate.
+	info.Name = meta.Name
+	info.Description = meta.Description
+	info.Headers = meta.Headers
+
+	// Prepare the meta message
+	metaSubj := fmt.Sprintf(objMetaPreTmpl, obs.name, encodeName(meta.Name))
+	mm := NewMsg(metaSubj)
+	mm.Header.Set(MsgRollup, MsgRollupSubject)
 	mm.Data, err = json.Marshal(info)
 	if err != nil {
 		return err
 	}
+
+	// Publish the meta message.
 	_, err = obs.js.PublishMsg(mm)
-	return err
+	if err != nil {
+		return err
+	}
+
+	// did the name of this object change? We just stored the meta under the new name
+	// so delete the meta from the old name via purge stream for subject
+	if name != meta.Name {
+		metaSubj := fmt.Sprintf(objMetaPreTmpl, obs.name, encodeName(name))
+		return obs.js.purgeStream(obs.stream, &StreamPurgeRequest{Subject: metaSubj})
+	}
+
+	return nil
 }
 
 // Seal will seal the object store, no further modifications will be allowed.


Test patch:
diff --git a/nats_test.go b/nats_test.go
index 1b5ce5e..176f6d0 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -1576,6 +1576,21 @@ func TestUserCredentialsChainedFile(t *testing.T) {
 	}
 }
 
+func TestUserJWTAndSeed(t *testing.T) {
+	if server.VERSION[0] == '1' {
+		t.Skip()
+	}
+	ts := runTrustServer()
+	defer ts.Shutdown()
+
+	url := fmt.Sprintf("nats://127.0.0.1:%d", TEST_PORT)
+	nc, err := Connect(url, UserJWTAndSeed(uJWT, string(uSeed)))
+	if err != nil {
+		t.Fatalf("Expected to connect, got %v", err)
+	}
+	nc.Close()
+}
+
 func TestExpiredAuthentication(t *testing.T) {
 	// The goal of these tests was to check how a client with an expiring JWT
 	// behaves. It should receive an async -ERR indicating that the auth
diff --git a/test/js_test.go b/test/js_test.go
index f644d64..46f6440 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -64,32 +64,169 @@ func TestJetStreamNotEnabled(t *testing.T) {
 	}
 }
 
-func TestJetStreamNotAccountEnabled(t *testing.T) {
-	conf := createConfFile(t, []byte(`
-		listen: 127.0.0.1:-1
-		no_auth_user: rip
-		jetstream: {max_mem_store: 64GB, max_file_store: 10TB}
-		accounts: {
-			JS: {
-				jetstream: enabled
-				users: [ {user: dlc, password: foo} ]
-			},
-			IU: {
-				users: [ {user: rip, password: bar} ]
-			},
+func TestJetStreamErrors(t *testing.T) {
+	t.Run("API error", func(t *testing.T) {
+		conf := createConfFile(t, []byte(`
+			listen: 127.0.0.1:-1
+			no_auth_user: rip
+			jetstream: {max_mem_store: 64GB, max_file_store: 10TB}
+			accounts: {
+				JS: {
+					jetstream: enabled
+					users: [ {user: dlc, password: foo} ]
+				},
+				IU: {
+					users: [ {user: rip, password: bar} ]
+				},
+			}
+		`))
+		defer os.Remove(conf)
+
+		s, _ := RunServerWithConfig(conf)
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		_, err := js.AccountInfo()
+		// check directly to var (backwards compatible)
+		if err != nats.ErrJetStreamNotEnabledForAccount {
+			t.Fatalf("Did not get the proper error, got %v", err)
 		}
-	`))
-	defer os.Remove(conf)
 
-	s, _ := RunServerWithConfig(conf)
-	defer shutdownJSServerAndRemoveStorage(t, s)
+		// matching via errors.Is
+		if ok := errors.Is(err, nats.ErrJetStreamNotEnabledForAccount); !ok {
+			t.Fatal("Expected ErrJetStreamNotEnabledForAccount")
+		}
 
-	nc, js := jsClient(t, s)
-	defer nc.Close()
+		// matching wrapped via error.Is
+		err2 := fmt.Errorf("custom error: %w", nats.ErrJetStreamNotEnabledForAccount)
+		if ok := errors.Is(err2, nats.ErrJetStreamNotEnabledForAccount); !ok {
+			t.Fatal("Expected wrapped ErrJetStreamNotEnabled")
+		}
+
+		// via classic type assertion.
+		jserr, ok := err.(nats.JetStreamError)
+		if !ok {
+			t.Fatal("Expected a JetStreamError")
+		}
+		expected := nats.JSErrCodeJetStreamNotEnabledForAccount
+		if jserr.APIError().ErrorCode != expected {
+			t.Fatalf("Expected: %v, got: %v", expected, jserr.APIError().ErrorCode)
+		}
+		if jserr.APIError() == nil {
+			t.Fatal("Expected APIError")
+		}
+
+		// matching to interface via errors.As(...)
+		var apierr nats.JetStreamError
+		ok = errors.As(err, &apierr)
+		if !ok {
+			t.Fatal("Expected a JetStreamError")
+		}
+		if apierr.APIError() == nil {
+			t.Fatal("Expected APIError")
+		}
+		if apierr.APIError().ErrorCode != expected {
+			t.Fatalf("Expected: %v, got: %v", expected, apierr.APIError().ErrorCode)
+		}
+		expectedMessage := "nats: API error 10039: jetstream not enabled for account"
+		if apierr.Error() != expectedMessage {
+			t.Fatalf("Expected: %v, got: %v", expectedMessage, apierr.Error())
+		}
+
+		// an APIError also implements the JetStreamError interface.
+		var _ nats.JetStreamError = &nats.APIError{}
+
+		// matching arbitrary custom error via errors.Is(...)
+		customErr := &nats.APIError{ErrorCode: expected}
+		if ok := errors.Is(customErr, nats.ErrJetStreamNotEnabledForAccount); !ok {
+			t.Fatal("Expected wrapped ErrJetStreamNotEnabledForAccount")
+		}
+		customErr = &nats.APIError{ErrorCode: 1}
+		if ok := errors.Is(customErr, nats.ErrJetStreamNotEnabledForAccount); ok {
+			t.Fatal("Expected to not match ErrJetStreamNotEnabled")
+		}
+		var cerr nats.JetStreamError
+		if ok := errors.As(customErr, &cerr); !ok {
+			t.Fatal("Expected custom error to be a JetStreamError")
+		}
+
+		// matching to concrete type via errors.As(...)
+		var aerr *nats.APIError
+		ok = errors.As(err, &aerr)
+		if !ok {
+			t.Fatal("Expected an APIError")
+		}
+		if aerr.ErrorCode != expected {
+			t.Fatalf("Expected: %v, got: %v", expected, aerr.ErrorCode)
+		}
+		expectedMessage = "nats: API error 10039: jetstream not enabled for account"
+		if aerr.Error() != expectedMessage {
+			t.Fatalf("Expected: %v, got: %v", expectedMessage, apierr.Error())
+		}
+	})
+
+	t.Run("test non-api error", func(t *testing.T) {
+		s := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, s)
+
+		nc, js := jsClient(t, s)
+		defer nc.Close()
+
+		// stream with empty name
+		_, err := js.AddStream(&nats.StreamConfig{})
+		if err == nil {
+			t.Fatalf("Expected error, got nil")
+		}
+
+		// check directly to var (backwards compatible)
+		if err != nats.ErrStreamNameRequired {
+			t.Fatalf("Expected: %v; got: %v", nats.ErrInvalidStreamName, err)
+		}
+
+		// matching via errors.Is
+		if ok := errors.Is(err, nats.ErrStreamNameRequired); !ok {
+			t.Fatalf("Expected: %v; got: %v", nats.ErrStreamNameRequired, err)
+		}
+
+		// matching wrapped via error.Is
+		err2 := fmt.Errorf("custom error: %w", nats.ErrStreamNameRequired)
+		if ok := errors.Is(err2, nats.ErrStreamNameRequired); !ok {
+			t.Fatal("Expected wrapped ErrStreamNameRequired")
+		}
+
+		// via classic type assertion.
+		jserr, ok := err.(nats.JetStreamError)
+		if !ok {
+			t.Fatal("Expected a JetStreamError")
+		}
+		if jserr.APIError() != nil {
+			t.Fatalf("Expected: empty APIError; got: %v", jserr.APIError())
+		}
+
+		// matching to interface via errors.As(...)
+		var jserr2 nats.JetStreamError
+		ok = errors.As(err, &jserr2)
+		if !ok {
+			t.Fatal("Expected a JetStreamError")
+		}
+		if jserr2.APIError() != nil {
+			t.Fatalf("Expected: empty APIError; got: %v", jserr2.APIError())
+		}
+		expectedMessage := "nats: stream name is required"
+		if jserr2.Error() != expectedMessage {
+			t.Fatalf("Expected: %v, got: %v", expectedMessage, jserr2.Error())
+		}
+
+		// matching to concrete type via errors.As(...)
+		var aerr *nats.APIError
+		ok = errors.As(err, &aerr)
+		if ok {
+			t.Fatal("Expected ErrStreamNameRequired not to map to APIError")
+		}
+	})
 
-	if _, err := js.AccountInfo(); err != nats.ErrJetStreamNotEnabled {
-		t.Fatalf("Did not get the proper error, got %v", err)
-	}
 }
 
 func TestJetStreamPublish(t *testing.T) {
@@ -336,7 +473,7 @@ func TestJetStreamSubscribe(t *testing.T) {
 	// Check that Queue subscribe without durable name requires queue name
 	// to not have "." in the name.
 	_, err = js.QueueSubscribeSync("foo", "bar.baz")
-	if err != nats.ErrInvalidDurableName {
+	if err != nats.ErrInvalidConsumerName {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
@@ -668,7 +805,7 @@ func TestJetStreamSubscribe(t *testing.T) {
 	cancel()
 
 	// Prevent invalid durable names
-	if _, err := js.SubscribeSync("baz", nats.Durable("test.durable")); err != nats.ErrInvalidDurableName {
+	if _, err := js.SubscribeSync("baz", nats.Durable("test.durable")); err != nats.ErrInvalidConsumerName {
 		t.Fatalf("Expected invalid durable name error")
 	}
 
@@ -1136,6 +1273,12 @@ func TestJetStreamManagement(t *testing.T) {
 		}
 	})
 
+	t.Run("stream with given name already exists", func(t *testing.T) {
+		if _, err := js.AddStream(&nats.StreamConfig{Name: "foo", Description: "desc"}); !errors.Is(err, nats.ErrStreamNameAlreadyInUse) {
+			t.Fatalf("Expected error: %v; got: %v", nats.ErrStreamNameAlreadyInUse, err)
+		}
+	})
+
 	for i := 0; i < 25; i++ {
 		js.Publish("foo", []byte("hi"))
 	}
@@ -1211,8 +1354,20 @@ func TestJetStreamManagement(t *testing.T) {
 			t.Fatalf("ConsumerInfo is not correct %+v", ci)
 		}
 
-		if _, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "test.durable"}); err != nats.ErrInvalidDurableName {
-			t.Fatalf("Expected invalid durable name error")
+		if _, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "test.durable"}); err != nats.ErrInvalidConsumerName {
+			t.Fatalf("Expected: %v; got: %v", nats.ErrInvalidConsumerName, err)
+		}
+	})
+
+	t.Run("consumer with given name already exists", func(t *testing.T) {
+		// configs do not match
+		if _, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckAllPolicy}); !errors.Is(err, nats.ErrConsumerNameAlreadyInUse) {
+			t.Fatalf("Expected error: %v; got: %v", nats.ErrConsumerNameAlreadyInUse, err)
+		}
+
+		// configs are the same
+		if _, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy}); err != nil {
+			t.Fatalf("Expected no error; got: %v", err)
 		}
 	})
 
@@ -1233,8 +1388,8 @@ func TestJetStreamManagement(t *testing.T) {
 			t.Fatalf("Expected %v, got: %v", nats.ErrInvalidStreamName, err)
 		}
 		_, err = js.AddConsumer("foo", &nats.ConsumerConfig{Durable: "bad.consumer.name", AckPolicy: nats.AckExplicitPolicy})
-		if err != nats.ErrInvalidDurableName {
-			t.Fatalf("Expected %v, got: %v", nats.ErrInvalidDurableName, err)
+		if err != nats.ErrInvalidConsumerName {
+			t.Fatalf("Expected %v, got: %v", nats.ErrInvalidConsumerName, err)
 		}
 	})
 
@@ -1365,13 +1520,13 @@ func TestJetStreamManagement(t *testing.T) {
 		// Check that durable name is required
 		expected.Durable = ""
 		_, err = js.UpdateConsumer("foo", &expected)
-		if err != nats.ErrInvalidDurableName {
+		if err != nats.ErrConsumerNameRequired {
 			t.Fatalf("Expected consumer name required error, got %v", err)
 		}
 		// Check that durable name is valid
 		expected.Durable = "bad.consumer.name"
 		_, err = js.UpdateConsumer("foo", &expected)
-		if err != nats.ErrInvalidDurableName {
+		if err != nats.ErrInvalidConsumerName {
 			t.Fatalf("Expected consumer name required error, got %v", err)
 		}
 		expected.Durable = "update_push_consumer"
@@ -1411,6 +1566,8 @@ func TestJetStreamManagement(t *testing.T) {
 		expected.HeadersOnly = true
 		expected.MaxRequestBatch = 10
 		expected.MaxRequestExpires = 2 * time.Second
+		expected.MaxRequestMaxBytes = 1024
+
 		ci, err = js.UpdateConsumer("foo", &expected)
 		if err != nil {
 			t.Fatalf("Error on update: %v", err)
@@ -1578,6 +1735,20 @@ func TestAccountInfo(t *testing.T) {
 			`,
 			withError: nats.ErrJetStreamNotEnabled,
 		},
+		{
+			name: "jetstream not enabled for account",
+			cfg: `
+			listen: 127.0.0.1:-1
+			no_auth_user: foo
+			jetstream: enabled
+			accounts: {
+				A: {
+					users: [ {user: foo} ]
+				},
+			}
+			`,
+			withError: nats.ErrJetStreamNotEnabledForAccount,
+		},
 	}
 
 	for _, test := range tests {
@@ -4790,7 +4961,7 @@ func testJetStreamMirror_Source(t *testing.T, nodes ...*jsServer) {
 			t.Fatal("Unexpected success")
 		}
 		if !errors.Is(err, nats.ErrStreamNotFound) {
-			t.Fatal("Expected stream not found error", err.Error())
+			t.Fatalf("Expected error: %v; got: %v", nats.ErrStreamNotFound, err)
 		}
 	})
 
@@ -4799,8 +4970,9 @@ func testJetStreamMirror_Source(t *testing.T, nodes ...*jsServer) {
 		if err == nil {
 			t.Fatal("Unexpected success")
 		}
-		if err.Error() != `nats: consumer filter subject is not a valid subset of the interest subjects` {
-			t.Fatal("Expected stream not found error")
+		apiErr := &nats.APIError{}
+		if !errors.As(err, &apiErr) || apiErr.ErrorCode != 10093 {
+			t.Fatalf("Expected API error 10093; got: %v", err)
 		}
 	})
 
@@ -5870,6 +6042,20 @@ func testJetStreamFetchOptions(t *testing.T, srvs ...*jsServer) {
 		}
 	})
 
+	t.Run("max request max bytes", func(t *testing.T) {
+		defer js.PurgeStream(subject)
+
+		sub, err := js.PullSubscribe(subject, "max-request-max-bytes", nats.MaxRequestMaxBytes(100))
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer sub.Unsubscribe()
+
+		if _, err := sub.Fetch(10, nats.PullMaxBytes(200)); err == nil || !strings.Contains(err.Error(), "MaxRequestMaxBytes of 100") {
+			t.Fatalf("Expected error about max request max bytes, got %v", err)
+		}
+	})
+
 	t.Run("max request expires", func(t *testing.T) {
 		defer js.PurgeStream(subject)
 
diff --git a/test/object_test.go b/test/object_test.go
index 4bed2c7..6d8f09c 100644
--- a/test/object_test.go
+++ b/test/object_test.go
@@ -21,7 +21,6 @@ import (
 	"path"
 	"path/filepath"
 	"reflect"
-	"runtime"
 	"testing"
 	"time"
 
@@ -83,10 +82,6 @@ func TestObjectBasics(t *testing.T) {
 		t.Fatalf("invalid description: '%s'", status.Description())
 	}
 
-	// Check simple errors.
-	_, err = obs.Get("FOO")
-	expectErr(t, err)
-
 	// Now get the object back.
 	result, err := obs.Get("BLOB")
 	expectOk(t, err)
@@ -111,6 +106,19 @@ func TestObjectBasics(t *testing.T) {
 	expectOk(t, err)
 	_, err = obs.Get("BLOB")
 	expectErr(t, err, nats.ErrStreamNotFound)
+
+	// Check simple errors.
+	_, err = obs.Get("FOO")
+	expectErr(t, err)
+
+	_, err = obs.Get("")
+	expectErr(t, err)
+
+	_, err = obs.Get("")
+	expectErr(t, err)
+
+	_, err = obs.PutBytes("", blob)
+	expectErr(t, err)
 }
 
 func TestDefaultObjectStatus(t *testing.T) {
@@ -142,13 +150,6 @@ func TestDefaultObjectStatus(t *testing.T) {
 }
 
 func TestObjectFileBasics(t *testing.T) {
-	// The filename is likely to be something like C:\xxx and we don't allow
-	// the '\' or ':' characters. We could translate the tmpFile.Name() from
-	// C:\foo\bar to C:/foo/bar (since Windows is OK with that), however,
-	// that would still fail on key validation because `:` is not allowed.
-	if runtime.GOOS == "windows" {
-		t.SkipNow()
-	}
 	s := RunBasicJetStreamServer()
 	defer shutdownJSServerAndRemoveStorage(t, s)
 
@@ -322,21 +323,19 @@ func TestObjectNames(t *testing.T) {
 	// Test filename like naming.
 	_, err = obs.PutString("BLOB.txt", "A")
 	expectOk(t, err)
+
 	// Spaces ok
 	_, err = obs.PutString("foo bar", "A")
 	expectOk(t, err)
 
+	// things that can be in a filename across multiple OSes
+	// dot, asterisk, lt, gt, colon, double-quote, fwd-slash, backslash, pipe, question-mark, ampersand
+	_, err = obs.PutString(".*<>:\"/\\|?&", "A")
+	expectOk(t, err)
+
 	// Errors
-	_, err = obs.PutString("*", "A")
-	expectErr(t, err)
-	_, err = obs.PutString(">", "A")
-	expectErr(t, err)
 	_, err = obs.PutString("", "A")
 	expectErr(t, err)
-	_, err = obs.PutString("", "\t")
-	expectErr(t, err)
-	_, err = obs.PutString("", "\n")
-	expectErr(t, err)
 }
 
 func TestObjectMetadata(t *testing.T) {
@@ -353,27 +352,52 @@ func TestObjectMetadata(t *testing.T) {
 	_, err = obs.PutString("A", "AAA")
 	expectOk(t, err)
 
-	meta := &nats.ObjectMeta{Name: "B"}
+	meta := &nats.ObjectMeta{Name: "A"}
+	meta.Description = "descA"
+	meta.Headers = make(nats.Header)
+	meta.Headers.Set("color", "blue")
 
+	// simple update that does not change the name, just adds data
 	err = obs.UpdateMeta("A", meta)
 	expectOk(t, err)
 
-	info, err := obs.GetInfo("B")
+	info, err := obs.GetInfo("A")
 	expectOk(t, err)
-	if info.Name != "B" {
+	if info.Name != "A" || info.Description != "descA" || info.Headers == nil || info.Headers.Get("color") != "blue" {
 		t.Fatalf("Update failed: %+v", info)
 	}
+
+	// update that changes the name and some data
+	meta = &nats.ObjectMeta{Name: "B"}
+	meta.Description = "descB"
 	meta.Headers = make(nats.Header)
 	meta.Headers.Set("color", "red")
 
-	err = obs.UpdateMeta("B", meta)
+	err = obs.UpdateMeta("A", meta)
 	expectOk(t, err)
 
+	_, err = obs.GetInfo("A")
+	if err == nil {
+		t.Fatal("Object meta for original name was not removed.")
+	}
+
 	info, err = obs.GetInfo("B")
 	expectOk(t, err)
-	if info.Headers == nil || info.Headers.Get("color") != "red" {
+
+	if info.Name != "B" || info.Description != "descB" || info.Headers == nil || info.Headers.Get("color") != "red" {
 		t.Fatalf("Update failed: %+v", info)
 	}
+
+	// delete the object to test updating against a deleted object
+	err = obs.Delete("B")
+	expectOk(t, err)
+	err = obs.UpdateMeta("B", meta)
+	expectErr(t, err)
+
+	err = obs.UpdateMeta("X", meta)
+	if err == nil {
+		t.Fatal("Expected an error when trying to update an object that does not exist.")
+	}
 }
 
 func TestObjectWatch(t *testing.T) {
@@ -448,12 +472,15 @@ func TestObjectWatch(t *testing.T) {
 	expectOk(t, err)
 
 	// Update Meta
-	info, err := obs.GetInfo("A")
+	deletedInfo, err := obs.GetInfo("A")
 	expectOk(t, err)
-	meta := &info.ObjectMeta
-	meta.Description = "A's are better than B's"
+	if !deletedInfo.Deleted {
+		t.Fatalf("Expected object to be deleted.")
+	}
+	meta := &deletedInfo.ObjectMeta
+	meta.Description = "Making a change."
 	err = obs.UpdateMeta("A", meta)
-	expectOk(t, err)
+	expectErr(t, err)
 }
 
 func TestObjectLinks(t *testing.T) {
@@ -471,12 +498,17 @@ func TestObjectLinks(t *testing.T) {
 	_, err = root.PutString("B", "BBB")
 	expectOk(t, err)
 
-	info, err := root.GetInfo("A")
+	infoA, err := root.GetInfo("A")
 	expectOk(t, err)
 
-	// Self link to individual object.
-	_, err = root.AddLink("LA", info)
+	// Link to individual object.
+	infoLA, err := root.AddLink("LA", infoA)
 	expectOk(t, err)
+	expectLinkIsCorrect(t, infoA, infoLA)
+
+	// link to a link
+	_, err = root.AddLink("LALA", infoLA)
+	expectErr(t, err)
 
 	dir, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "DIR"})
 	expectOk(t, err)
@@ -486,27 +518,99 @@ func TestObjectLinks(t *testing.T) {
 	_, err = dir.PutString("DIR/B", "DIR-BBB")
 	expectOk(t, err)
 
-	info, err = dir.GetInfo("DIR/B")
+	infoB, err := dir.GetInfo("DIR/B")
+	expectOk(t, err)
+
+	infoLB, err := root.AddLink("DBL", infoB)
 	expectOk(t, err)
+	expectLinkIsCorrect(t, infoB, infoLB)
 
-	binfo, err := root.AddLink("DBL", info)
+	// Now add whole other store as a link, like a directory.
+	infoBucketLink, err := root.AddBucketLink("dir", dir)
 	expectOk(t, err)
 
-	if binfo.Name != "DBL" || binfo.NUID == "" || binfo.ModTime.IsZero() {
-		t.Fatalf("Link info not what was expected: %+v", binfo)
+	expectLinkPartsAreCorrect(t, infoBucketLink, "DIR", "")
+
+	// Try to get a linked object, same bucket
+	getLA, err := root.GetString("LA")
+	expectOk(t, err)
+
+	if getLA != "AAA" {
+		t.Fatalf("Expected %q but got %q", "AAA", getLA)
 	}
 
-	// Now add whole other store as a link, like a directory.
-	_, err = root.AddBucketLink("dir", dir)
+	// Try to get a linked object, cross bucket
+	getDbl, err := root.GetString("DBL")
 	expectOk(t, err)
 
-	// Now try to get a linked object.
-	dbl, err := root.GetString("DBL")
+	if getDbl != "DIR-BBB" {
+		t.Fatalf("Expected %q but got %q", "DIR-BBB", getDbl)
+	}
+
+	// change a link
+	infoB, err = root.GetInfo("B")
 	expectOk(t, err)
 
-	if dbl != "DIR-BBB" {
-		t.Fatalf("Expected %q but got %q", "DIR-BBB", dbl)
+	infoLA, err = root.GetInfo("LA")
+	expectOk(t, err)
+	expectLinkIsCorrect(t, infoA, infoLA)
+
+	infoLA, err = root.AddLink("LA", infoB)
+	expectOk(t, err)
+	expectLinkIsCorrect(t, infoB, infoLA)
+
+	// change a bucket link
+	infoBucketLink, err = root.GetInfo("dir")
+	expectOk(t, err)
+	expectLinkPartsAreCorrect(t, infoBucketLink, "DIR", "")
+
+	infoBucketLink, err = root.AddBucketLink("dir", root)
+	expectOk(t, err)
+	expectLinkPartsAreCorrect(t, infoBucketLink, "ROOT", "")
+
+	// Check simple errors.
+	_, err = root.AddLink("", infoB)
+	expectErr(t, err)
+
+	_, err = root.AddLink("Nil Object", nil)
+	expectErr(t, err)
+
+	infoB.Name = ""
+	_, err = root.AddLink("Empty Info Name", infoB)
+	expectErr(t, err)
+
+	// Check Error Link to a Link
+	_, err = root.AddLink("Link To Link", infoLB)
+	expectErr(t, err)
+
+	// Check Error Link to a Link
+	_, err = root.AddBucketLink("", root)
+	expectErr(t, err)
+
+	_, err = root.AddBucketLink("Nil Bucket", nil)
+	expectErr(t, err)
+}
+
+func expectLinkIsCorrect(t *testing.T, originalObject *nats.ObjectInfo, linkObject *nats.ObjectInfo) {
+	if linkObject.Opts.Link != nil {
+		if expectLinkPartsAreCorrect(t, linkObject, originalObject.Bucket, originalObject.Name) {
+			return
+		}
+	}
+	t.Fatalf("Link info not what was expected:\nActual: %+v\nTarget: %+v", linkObject, originalObject)
+}
+
+func expectLinkPartsAreCorrect(t *testing.T, linkObject *nats.ObjectInfo, bucket, name string) bool {
+	if linkObject.Opts.Link.Bucket == bucket {
+		if linkObject.Opts.Link.Name == name {
+			if !linkObject.ModTime.IsZero() {
+				if linkObject.NUID != "" {
+					return true
+				}
+			}
+		}
 	}
+	return false
 }
 
 // Right now no history, just make sure we are cleaning up after ourselves.
@@ -520,12 +624,20 @@ func TestObjectHistory(t *testing.T) {
 	obs, err := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "OBJS"})
 	expectOk(t, err)
 
-	_, err = obs.PutBytes("A", bytes.Repeat([]byte("A"), 100))
+	info, err := obs.PutBytes("A", bytes.Repeat([]byte("A"), 10))
 	expectOk(t, err)
 
-	_, err = obs.PutBytes("A", bytes.Repeat([]byte("a"), 100))
+	if info.Size != 10 {
+		t.Fatalf("Invalid first put when testing history %+v", info)
+	}
+
+	info, err = obs.PutBytes("A", bytes.Repeat([]byte("a"), 20))
 	expectOk(t, err)
 
+	if info.Size != 20 {
+		t.Fatalf("Invalid second put when testing history %+v", info)
+	}
+
 	// Should only be 1 copy of 'A', so 1 data and 1 meta since history was not selected.
 	si, err := js.StreamInfo("OBJ_OBJS")
 	expectOk(t, err)