Repository: nats-io/nats.go
Pull request #904 (2022-02-10 18:14:56)

Expose nats.Context option for nats.KeyWatcher
Signed-off-by: Matthew DeVenny <matt@boxboat.com>

Currently if you wish to extend the `nats.KeyValue` interface, when implementing `Watch` you have no means to get at the `nats.Context` option if it has been set on the `nats.KeyWatcher`. This PR proposes a method to the nats.KeyWatcher to access that context.



coveralls: 
[![Coverage Status](https://coveralls.io/builds/46427251/badge)](https://coveralls.io/builds/46427251)

Coverage decreased (-0.05%) to 85.122% when pulling **65759d5aa0ff769ddbdbd1f38d9606c0d64aed2a on boxboatmatt:feature/kv-interface-options** into **3ead80931bdf9055a83eba39c1bb0ae4c46de0cb on nats-io:main**.


Patch:
diff --git a/js.go b/js.go
index 4003a18..d526978 100644
--- a/js.go
+++ b/js.go
@@ -181,6 +181,12 @@ const (
 	// routine. Without this, the subscription would possibly stall until
 	// a new message or heartbeat/fc are received.
 	chanSubFCCheckInterval = 250 * time.Millisecond
+
+	// Default time wait between retries on Publish iff err is NoResponders.
+	DefaultPubRetryWait = 250 * time.Millisecond
+
+	// Default number of retries
+	DefaultPubRetryAttempts = 2
 )
 
 // Types of control messages, so far heartbeat and flow control
@@ -340,6 +346,11 @@ type pubOpts struct {
 	str string // Expected stream name
 	seq uint64 // Expected last sequence
 	lss uint64 // Expected last sequence per subject
+
+	// Publish retries for NoResponders err.
+	rwait time.Duration // Retry wait between attempts
+	rnum  int           // Retry attempts
+
 }
 
 // pubAckResponse is the ack response from the JetStream API when publishing a message.
@@ -377,7 +388,7 @@ const (
 
 // PublishMsg publishes a Msg to a stream from JetStream.
 func (js *js) PublishMsg(m *Msg, opts ...PubOpt) (*PubAck, error) {
-	var o pubOpts
+	var o = pubOpts{rwait: DefaultPubRetryWait, rnum: DefaultPubRetryAttempts}
 	if len(opts) > 0 {
 		if m.Header == nil {
 			m.Header = Header{}
@@ -422,11 +433,35 @@ func (js *js) PublishMsg(m *Msg, opts ...PubOpt) (*PubAck, error) {
 	}
 
 	if err != nil {
-		if err == ErrNoResponders {
-			err = ErrNoStreamResponse
+		for r, ttl := 0, o.ttl; err == ErrNoResponders && (r < o.rnum || o.rnum < 0); r++ {
+			// To protect against small blips in leadership changes etc, if we get a no responders here retry.
+			if o.ctx != nil {
+				select {
+				case <-o.ctx.Done():
+				case <-time.After(o.rwait):
+				}
+			} else {
+				time.Sleep(o.rwait)
+			}
+			if o.ttl > 0 {
+				ttl -= o.rwait
+				if ttl <= 0 {
+					err = ErrTimeout
+					break
+				}
+				resp, err = js.nc.RequestMsg(m, time.Duration(ttl))
+			} else {
+				resp, err = js.nc.RequestMsgWithContext(o.ctx, m)
+			}
+		}
+		if err != nil {
+			if err == ErrNoResponders {
+				err = ErrNoStreamResponse
+			}
+			return nil, err
 		}
-		return nil, err
 	}
+
 	var pa pubAckResponse
 	if err := json.Unmarshal(resp.Data, &pa); err != nil {
 		return nil, ErrInvalidJSAck
@@ -803,6 +838,22 @@ func ExpectLastMsgId(id string) PubOpt {
 	})
 }
 
+// RetryWait sets the retry wait time when ErrNoResponders is encountered.
+func RetryWait(dur time.Duration) PubOpt {
+	return pubOptFn(func(opts *pubOpts) error {
+		opts.rwait = dur
+		return nil
+	})
+}
+
+// RetryAttempts sets the retry number of attemopts when ErrNoResponders is encountered.
+func RetryAttempts(num int) PubOpt {
+	return pubOptFn(func(opts *pubOpts) error {
+		opts.rnum = num
+		return nil
+	})
+}
+
 type ackOpts struct {
 	ttl      time.Duration
 	ctx      context.Context
@@ -919,9 +970,6 @@ type ConsumerConfig struct {
 
 	// Ephemeral inactivity threshold.
 	InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
-
-	// Internal Use
-	Direct bool `json:"direct,omitempty"`
 }
 
 // ConsumerInfo is the info from a JetStream consumer.
@@ -1284,6 +1332,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		nms           string
 		hbi           time.Duration
 		ccreq         *createConsumerRequest // In case we need to hold onto it for ordered consumers.
+		maxap         int
 	)
 
 	// Do some quick checks here for ordered consumers. We do these here instead of spread out
@@ -1322,7 +1371,6 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		o.cfg.AckPolicy = AckNonePolicy
 		o.cfg.MaxDeliver = 1
 		o.cfg.AckWait = 22 * time.Hour // Just set to something known, not utilized.
-		o.cfg.Direct = true
 		if !hasHeartbeats {
 			o.cfg.Heartbeat = orderedHeartbeatsInterval
 		}
@@ -1365,6 +1413,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		icfg := &info.Config
 		hasFC, hbi = icfg.FlowControl, icfg.Heartbeat
 		hasHeartbeats = hbi > 0
+		maxap = icfg.MaxAckPending
 	case (err != nil && !notFoundErr) || (notFoundErr && consumerBound):
 		// If the consumer is being bound and we got an error on pull subscribe then allow the error.
 		if !(isPullMode && lookupErr && consumerBound) {
@@ -1402,16 +1451,9 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		}
 
 		// If we have acks at all and the MaxAckPending is not set go ahead
-		// and set to the internal max.
-		// TODO(dlc) - We should be able to update this if client updates PendingLimits.
-		if cfg.MaxAckPending == 0 && cfg.AckPolicy != AckNonePolicy {
-			if !isPullMode && cb != nil && hasFC {
-				cfg.MaxAckPending = DefaultSubPendingMsgsLimit * 16
-			} else if ch != nil {
-				cfg.MaxAckPending = cap(ch)
-			} else {
-				cfg.MaxAckPending = DefaultSubPendingMsgsLimit
-			}
+		// and set to the internal max for channel based consumers
+		if cfg.MaxAckPending == 0 && ch != nil && cfg.AckPolicy != AckNonePolicy {
+			cfg.MaxAckPending = cap(ch)
 		}
 		// Create request here.
 		ccreq = &createConsumerRequest{
@@ -1458,12 +1500,6 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		return nil, err
 	}
 
-	// With flow control enabled async subscriptions we will disable msgs
-	// limits, and set a larger pending bytes limit by default.
-	if !isPullMode && cb != nil && hasFC {
-		sub.SetPendingLimits(DefaultSubPendingMsgsLimit*16, DefaultSubPendingBytesLimit)
-	}
-
 	// If we fail and we had the sub we need to cleanup, but can't just do a straight Unsubscribe or Drain.
 	// We need to clear the jsi so we do not remove any durables etc.
 	cleanUpSub := func() {
@@ -1574,6 +1610,19 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 			}
 			sub.mu.Unlock()
 		}
+		// Capture max ack pending from the info response here which covers both
+		// success and failure followed by consumer lookup.
+		maxap = info.Config.MaxAckPending
+	}
+
+	// If maxap is greater than the default sub's pending limit, use that.
+	if maxap > DefaultSubPendingMsgsLimit {
+		// For bytes limit, use the min of maxp*1MB or DefaultSubPendingBytesLimit
+		bl := maxap * 1024 * 1024
+		if bl < DefaultSubPendingBytesLimit {
+			bl = DefaultSubPendingBytesLimit
+		}
+		sub.SetPendingLimits(maxap, bl)
 	}
 
 	// Do heartbeats last if needed.
diff --git a/jsm.go b/jsm.go
index 87ab37c..11545f3 100644
--- a/jsm.go
+++ b/jsm.go
@@ -587,6 +587,9 @@ func (js *js) AddStream(cfg *StreamConfig, opts ...JSOpt) (*StreamInfo, error) {
 		return nil, err
 	}
 	if resp.Error != nil {
+		if resp.Error.ErrorCode == 10058 {
+			return nil, ErrStreamNameAlreadyInUse
+		}
 		return nil, errors.New(resp.Error.Description)
 	}
 
@@ -621,7 +624,7 @@ func (js *js) StreamInfo(stream string, opts ...JSOpt) (*StreamInfo, error) {
 		if resp.Error.Code == 404 {
 			return nil, ErrStreamNotFound
 		}
-		return nil, errors.New(resp.Error.Description)
+		return nil, fmt.Errorf("nats: %s", resp.Error.Description)
 	}
 
 	return resp.StreamInfo, nil
diff --git a/kv.go b/kv.go
index cdbb4e3..5ba1160 100644
--- a/kv.go
+++ b/kv.go
@@ -17,6 +17,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"reflect"
 	"regexp"
 	"strconv"
 	"strings"
@@ -68,7 +69,7 @@ type KeyValue interface {
 	// Bucket returns the current bucket name.
 	Bucket() string
 	// PurgeDeletes will remove all current delete markers.
-	PurgeDeletes(opts ...WatchOpt) error
+	PurgeDeletes(opts ...PurgeOpt) error
 	// Status retrieves the status and configuration of a bucket
 	Status() (KeyValueStatus, error)
 }
@@ -93,6 +94,8 @@ type KeyValueStatus interface {
 
 // KeyWatcher is what is returned when doing a watch.
 type KeyWatcher interface {
+	// Context returns watcher context optionally provided by nats.Context option.
+	Context() context.Context
 	// Updates returns a channel to read any updates to entries.
 	Updates() <-chan KeyValueEntry
 	// Stop will stop this watcher.
@@ -149,6 +152,34 @@ func MetaOnly() WatchOpt {
 	})
 }
 
+type PurgeOpt interface {
+	configurePurge(opts *purgeOpts) error
+}
+
+type purgeOpts struct {
+	dmthr time.Duration // Delete markers threshold
+	ctx   context.Context
+}
+
+// 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
+}
+
+// For nats.Context() support.
+func (ctx ContextOpt) configurePurge(opts *purgeOpts) error {
+	opts.ctx = ctx
+	return nil
+}
+
 // KeyValueConfig is for configuring a KeyValue store.
 type KeyValueConfig struct {
 	Bucket       string
@@ -159,6 +190,7 @@ type KeyValueConfig struct {
 	MaxBytes     int64
 	Storage      StorageType
 	Replicas     int
+	Placement    *Placement
 }
 
 // Used to watch all keys.
@@ -263,6 +295,8 @@ func (js *js) KeyValue(bucket string) (KeyValue, error) {
 		stream: stream,
 		pre:    fmt.Sprintf(kvSubjectsPreTmpl, bucket),
 		js:     js,
+		// Determine if we need to use the JS prefix in front of Put and Delete operations
+		useJSPfx: js.opts.pre != defaultAPIPrefix,
 	}
 	return kv, nil
 }
@@ -296,18 +330,41 @@ func (js *js) CreateKeyValue(cfg *KeyValueConfig) (KeyValue, error) {
 		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,
 		Subjects:          []string{fmt.Sprintf(kvSubjectsTmpl, cfg.Bucket)},
 		MaxMsgsPerSubject: history,
-		MaxBytes:          cfg.MaxBytes,
+		MaxBytes:          maxBytes,
 		MaxAge:            cfg.TTL,
-		MaxMsgSize:        cfg.MaxValueSize,
+		MaxMsgSize:        maxMsgSize,
 		Storage:           cfg.Storage,
 		Replicas:          replicas,
+		Placement:         cfg.Placement,
 		AllowRollup:       true,
 		DenyDelete:        true,
+		Duplicates:        duplicateWindow,
+		MaxMsgs:           -1,
+		MaxConsumers:      -1,
 	}
 
 	// If we are at server version 2.7.2 or above use DiscardNew. We can not use DiscardNew for 2.7.1 or below.
@@ -316,7 +373,24 @@ func (js *js) CreateKeyValue(cfg *KeyValueConfig) (KeyValue, error) {
 	}
 
 	if _, err := js.AddStream(scfg); err != nil {
-		return nil, err
+		// If we have a failure to add, it could be because we have
+		// a config change if the KV was created against a pre 2.7.2
+		// and we are now moving to a v2.7.2+. If that is the case
+		// and the only difference is the discard policy, then update
+		// the stream.
+		if err == ErrStreamNameAlreadyInUse {
+			if si, _ := js.StreamInfo(scfg.Name); si != nil {
+				// To compare, make the server's stream info discard
+				// policy same than ours.
+				si.Config.Discard = scfg.Discard
+				if reflect.DeepEqual(&si.Config, scfg) {
+					_, err = js.UpdateStream(scfg)
+				}
+			}
+		}
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	kv := &kvs{
@@ -324,6 +398,8 @@ func (js *js) CreateKeyValue(cfg *KeyValueConfig) (KeyValue, error) {
 		stream: scfg.Name,
 		pre:    fmt.Sprintf(kvSubjectsPreTmpl, cfg.Bucket),
 		js:     js,
+		// Determine if we need to use the JS prefix in front of Put and Delete operations
+		useJSPfx: js.opts.pre != defaultAPIPrefix,
 	}
 	return kv, nil
 }
@@ -342,6 +418,10 @@ type kvs struct {
 	stream string
 	pre    string
 	js     *js
+	// 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
 }
 
 // Underlying entry.
@@ -453,6 +533,9 @@ func (kv *kvs) Put(key string, value []byte) (revision uint64, err error) {
 	}
 
 	var b strings.Builder
+	if kv.useJSPfx {
+		b.WriteString(kv.js.opts.pre)
+	}
 	b.WriteString(kv.pre)
 	b.WriteString(key)
 
@@ -491,6 +574,9 @@ func (kv *kvs) Update(key string, value []byte, revision uint64) (uint64, error)
 	}
 
 	var b strings.Builder
+	if kv.useJSPfx {
+		b.WriteString(kv.js.opts.pre)
+	}
 	b.WriteString(kv.pre)
 	b.WriteString(key)
 
@@ -520,6 +606,9 @@ func (kv *kvs) delete(key string, purge bool) error {
 	}
 
 	var b strings.Builder
+	if kv.useJSPfx {
+		b.WriteString(kv.js.opts.pre)
+	}
 	b.WriteString(kv.pre)
 	b.WriteString(key)
 
@@ -536,15 +625,43 @@ func (kv *kvs) delete(key string, purge bool) error {
 	return err
 }
 
+const kvDefaultPurgeDeletesMarkerThreshold = 30 * time.Minute
+
 // PurgeDeletes will remove all current delete markers.
 // This is a maintenance option if there is a larger buildup of delete markers.
-func (kv *kvs) PurgeDeletes(opts ...WatchOpt) error {
-	watcher, err := kv.WatchAll(opts...)
+// See DeleteMarkersOlderThan() option for more information.
+func (kv *kvs) PurgeDeletes(opts ...PurgeOpt) error {
+	var o purgeOpts
+	for _, opt := range opts {
+		if opt != nil {
+			if err := opt.configurePurge(&o); err != nil {
+				return err
+			}
+		}
+	}
+	// Transfer possible context purge option to the watcher. This is the
+	// only option that matters for the PurgeDeletes() feature.
+	var wopts []WatchOpt
+	if o.ctx != nil {
+		wopts = append(wopts, Context(o.ctx))
+	}
+	watcher, err := kv.WatchAll(wopts...)
 	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 {
@@ -564,8 +681,11 @@ func (kv *kvs) PurgeDeletes(opts ...WatchOpt) error {
 		b.WriteString(kv.pre)
 		b.WriteString(entry.Key())
 		pr.Subject = b.String()
-		err := kv.js.purgeStream(kv.stream, &pr)
-		if err != nil {
+		pr.Keep = 0
+		if olderThan > 0 && entry.Created().After(limit) {
+			pr.Keep = 1
+		}
+		if err := kv.js.purgeStream(kv.stream, &pr); err != nil {
 			return err
 		}
 		b.Reset()
@@ -625,6 +745,15 @@ type watcher struct {
 	initDone    bool
 	initPending uint64
 	received    uint64
+	ctx         context.Context
+}
+
+// Context returns the context for the watcher if set.
+func (w *watcher) Context() context.Context {
+	if w == nil {
+		return nil
+	}
+	return w.ctx
 }
 
 // Updates returns the interior channel.
@@ -667,7 +796,7 @@ func (kv *kvs) Watch(keys string, opts ...WatchOpt) (KeyWatcher, error) {
 	keys = b.String()
 
 	// We will block below on placing items on the chan. That is by design.
-	w := &watcher{updates: make(chan KeyValueEntry, 256)}
+	w := &watcher{updates: make(chan KeyValueEntry, 256), ctx: o.ctx}
 
 	update := func(m *Msg) {
 		tokens, err := getMetadataFields(m.Reply)
diff --git a/nats.go b/nats.go
index 3bf5b4d..566fced 100644
--- a/nats.go
+++ b/nats.go
@@ -160,6 +160,7 @@ var (
 	ErrMsgNotFound                  = errors.New("nats: message not found")
 	ErrMsgAlreadyAckd               = errors.New("nats: message was already acknowledged")
 	ErrStreamInfoMaxSubjects        = errors.New("nats: subject details would exceed maximum allowed")
+	ErrStreamNameAlreadyInUse       = errors.New("nats: stream name already in use")
 )
 
 func init() {
@@ -1435,7 +1436,7 @@ func (nc *Conn) setupServerPool() error {
 
 	// Check for Scheme hint to move to TLS mode.
 	for _, srv := range nc.srvPool {
-		if srv.url.Scheme == tlsScheme {
+		if srv.url.Scheme == tlsScheme || srv.url.Scheme == wsSchemeTLS {
 			// FIXME(dlc), this is for all in the pool, should be case by case.
 			nc.Opts.Secure = true
 			if nc.Opts.TLSConfig == nil {
@@ -1649,6 +1650,17 @@ func (w *natsWriter) doneWithPending() {
 	w.pending = nil
 }
 
+// Notify the reader that we are done with the connect, where "read" operations
+// happen synchronously and under the connection lock. After this point, "read"
+// will be happening from the read loop, without the connection lock.
+//
+// Note: this runs under the connection lock.
+func (r *natsReader) doneWithConnect() {
+	if wsr, ok := r.r.(*websocketReader); ok {
+		wsr.doneWithConnect()
+	}
+}
+
 func (r *natsReader) Read() ([]byte, error) {
 	if r.off >= 0 {
 		off := r.off
@@ -1803,6 +1815,21 @@ func (nc *Conn) ConnectedUrl() string {
 	return nc.current.url.String()
 }
 
+// ConnectedUrlRedacted reports the connected server's URL with passwords redacted
+func (nc *Conn) ConnectedUrlRedacted() string {
+	if nc == nil {
+		return _EMPTY_
+	}
+
+	nc.mu.RLock()
+	defer nc.mu.RUnlock()
+
+	if nc.status != CONNECTED {
+		return _EMPTY_
+	}
+	return nc.current.url.Redacted()
+}
+
 // ConnectedAddr returns the connected server's IP
 func (nc *Conn) ConnectedAddr() string {
 	if nc == nil {
@@ -1962,6 +1989,10 @@ func (nc *Conn) processConnectInit() error {
 	go nc.readLoop()
 	go nc.flusher()
 
+	// Notify the reader that we are done with the connect handshake, where
+	// reads were done synchronously and under the connection lock.
+	nc.br.doneWithConnect()
+
 	return nil
 }
 
@@ -2138,7 +2169,7 @@ func (nc *Conn) connectProto() (string, error) {
 		}
 		sigraw, err := o.SignatureCB([]byte(nc.info.Nonce))
 		if err != nil {
-			return _EMPTY_, err
+			return _EMPTY_, fmt.Errorf("error signing nonce: %v", err)
 		}
 		sig = base64.RawURLEncoding.EncodeToString(sigraw)
 	}
@@ -5237,7 +5268,7 @@ func nkeyPairFromSeedFile(seedFile string) (nkeys.KeyPair, error) {
 func sigHandler(nonce []byte, seedFile string) ([]byte, error) {
 	kp, err := nkeyPairFromSeedFile(seedFile)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("unable to extract key pair from file %q: %v", seedFile, err)
 	}
 	// Wipe our key on exit.
 	defer kp.Wipe()
diff --git a/object.go b/object.go
index 4de788b..94d9a4e 100644
--- a/object.go
+++ b/object.go
@@ -1,4 +1,4 @@
-// Copyright 2021 The NATS Authors
+// Copyright 2021-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
@@ -138,6 +138,7 @@ type ObjectStoreConfig struct {
 	TTL         time.Duration
 	Storage     StorageType
 	Replicas    int
+	Placement   *Placement
 }
 
 type ObjectStoreStatus interface {
@@ -245,6 +246,7 @@ func (js *js) CreateObjectStore(cfg *ObjectStoreConfig) (ObjectStore, error) {
 		MaxAge:      cfg.TTL,
 		Storage:     cfg.Storage,
 		Replicas:    cfg.Replicas,
+		Placement:   cfg.Placement,
 		Discard:     DiscardNew,
 		AllowRollup: true,
 	}
diff --git a/ws.go b/ws.go
index 2ef3f7f..4d24454 100644
--- a/ws.go
+++ b/ws.go
@@ -81,6 +81,7 @@ type websocketReader struct {
 	ib      []byte
 	ff      bool
 	fc      bool
+	nl      bool
 	dc      *wsDecompressor
 	nc      *Conn
 }
@@ -180,6 +181,15 @@ func wsNewReader(r io.Reader) *websocketReader {
 	return &websocketReader{r: r, ff: true}
 }
 
+// From now on, reads will be from the readLoop and we will need to
+// acquire the connection lock should we have to send/write a control
+// message from handleControlFrame.
+//
+// Note: this runs under the connection lock.
+func (r *websocketReader) doneWithConnect() {
+	r.nl = true
+}
+
 func (r *websocketReader) Read(p []byte) (int, error) {
 	var err error
 	var buf []byte
@@ -402,12 +412,12 @@ func (r *websocketReader) handleControlFrame(frameType wsOpCode, buf []byte, pos
 				}
 			}
 		}
-		r.nc.wsEnqueueCloseMsg(status, body)
+		r.nc.wsEnqueueCloseMsg(r.nl, status, body)
 		// Return io.EOF so that readLoop will close the connection as client closed
 		// after processing pending buffers.
 		return pos, io.EOF
 	case wsPingMessage:
-		r.nc.wsEnqueueControlMsg(wsPongMessage, payload)
+		r.nc.wsEnqueueControlMsg(r.nl, wsPongMessage, payload)
 	case wsPongMessage:
 		// Nothing to do..
 	}
@@ -644,14 +654,16 @@ func (nc *Conn) wsClose() {
 	nc.wsEnqueueCloseMsgLocked(wsCloseStatusNormalClosure, _EMPTY_)
 }
 
-func (nc *Conn) wsEnqueueCloseMsg(status int, payload string) {
+func (nc *Conn) wsEnqueueCloseMsg(needsLock bool, status int, payload string) {
 	// In some low-level unit tests it will happen...
 	if nc == nil {
 		return
 	}
-	nc.mu.Lock()
+	if needsLock {
+		nc.mu.Lock()
+		defer nc.mu.Unlock()
+	}
 	nc.wsEnqueueCloseMsgLocked(status, payload)
-	nc.mu.Unlock()
 }
 
 func (nc *Conn) wsEnqueueCloseMsgLocked(status int, payload string) {
@@ -675,25 +687,26 @@ func (nc *Conn) wsEnqueueCloseMsgLocked(status int, payload string) {
 	nc.bw.flush()
 }
 
-func (nc *Conn) wsEnqueueControlMsg(frameType wsOpCode, payload []byte) {
+func (nc *Conn) wsEnqueueControlMsg(needsLock bool, frameType wsOpCode, payload []byte) {
 	// In some low-level unit tests it will happen...
 	if nc == nil {
 		return
 	}
-	fh, key := wsCreateFrameHeader(false, frameType, len(payload))
-	nc.mu.Lock()
+	if needsLock {
+		nc.mu.Lock()
+		defer nc.mu.Unlock()
+	}
 	wr, ok := nc.bw.w.(*websocketWriter)
 	if !ok {
-		nc.mu.Unlock()
 		return
 	}
+	fh, key := wsCreateFrameHeader(false, frameType, len(payload))
 	wr.ctrlFrames = append(wr.ctrlFrames, fh)
 	if len(payload) > 0 {
 		wsMaskBuf(key, payload)
 		wr.ctrlFrames = append(wr.ctrlFrames, payload)
 	}
 	nc.bw.flush()
-	nc.mu.Unlock()
 }
 
 func wsPMCExtensionSupport(header http.Header) (bool, bool) {


Test patch:
diff --git a/kv_test.go b/kv_test.go
new file mode 100644
index 0000000..a7cd03e
--- /dev/null
+++ b/kv_test.go
@@ -0,0 +1,69 @@
+// Copyright 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 (
+	"testing"
+)
+
+func TestKeyValueDiscardOldToDiscardNew(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	checkDiscard := func(expected DiscardPolicy) KeyValue {
+		t.Helper()
+		kv, err := js.CreateKeyValue(&KeyValueConfig{Bucket: "TEST", History: 1})
+		if err != nil {
+			t.Fatalf("Error creating store: %v", err)
+		}
+		si, err := js.StreamInfo("KV_TEST")
+		if err != nil {
+			t.Fatalf("Error getting stream info: %v", err)
+		}
+		if si.Config.Discard != expected {
+			t.Fatalf("Expected discard policy %v, got %+v", expected, si)
+		}
+		return kv
+	}
+
+	// We are going to go from 2.7.1->2.7.2->2.7.1 and 2.7.2 again.
+	for i := 0; i < 2; i++ {
+		// Change the server version in the connection to
+		// create as-if we were connecting to a v2.7.1 server.
+		nc.mu.Lock()
+		nc.info.Version = "2.7.1"
+		nc.mu.Unlock()
+
+		kv := checkDiscard(DiscardOld)
+		if i == 0 {
+			if _, err := kv.PutString("foo", "value"); err != nil {
+				t.Fatalf("Error adding key: %v", err)
+			}
+		}
+
+		// Now change version to 2.7.2
+		nc.mu.Lock()
+		nc.info.Version = "2.7.2"
+		nc.mu.Unlock()
+
+		kv = checkDiscard(DiscardNew)
+		// Make sure the key still exists
+		if e, err := kv.Get("foo"); err != nil || string(e.Value()) != "value" {
+			t.Fatalf("Error getting key: err=%v e=%+v", err, e)
+		}
+	}
+}
diff --git a/nats_test.go b/nats_test.go
index d689d44..123df98 100644
--- a/nats_test.go
+++ b/nats_test.go
@@ -1483,6 +1483,17 @@ func TestUserCredentialsChainedFile(t *testing.T) {
 		t.Fatalf("Expected to connect, got %v", err)
 	}
 	nc.Close()
+
+	chainedFile = createTmpFile(t, []byte("invalid content"))
+	defer os.Remove(chainedFile)
+	nc, err = Connect(url, UserCredentials(chainedFile))
+	if err == nil || !strings.Contains(err.Error(),
+		"error signing nonce: unable to extract key pair from file") {
+		if nc != nil {
+			nc.Close()
+		}
+		t.Fatalf("Expected error about invalid creds file, got %q", err)
+	}
 }
 
 func TestExpiredAuthentication(t *testing.T) {
diff --git a/test/js_test.go b/test/js_test.go
index edbc239..7bb874e 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"io/ioutil"
+	mrand "math/rand"
 	"net"
 	"os"
 	"reflect"
@@ -6266,3 +6267,92 @@ func TestJetStreamSubscribeContextCancel(t *testing.T) {
 		})
 	})
 }
+
+func TestJetStreamClusterStreamLeaderChangeClientErr(t *testing.T) {
+	cfg := &nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	}
+
+	withJSClusterAndStream(t, "R3S", 3, cfg, func(t *testing.T, stream string, servers ...*jsServer) {
+		// We want to make sure the worse thing seen by the lower levels during a leadership change is NoResponders.
+		// We will have three concurrent contexts going on.
+		// 1. Leadership Changes every 500ms.
+		// 2. Publishing messages to the stream every 10ms.
+		// 3. StreamInfo calls every 15ms.
+		expires := time.Now().Add(5 * time.Second)
+		var wg sync.WaitGroup
+		wg.Add(3)
+
+		randServer := func() *server.Server {
+			return servers[mrand.Intn(len(servers))].Server
+		}
+
+		// Leadership changes.
+		go func() {
+			defer wg.Done()
+			nc, js := jsClient(t, randServer())
+			defer nc.Close()
+
+			sds := fmt.Sprintf(server.JSApiStreamLeaderStepDownT, "TEST")
+			for time.Now().Before(expires) {
+				time.Sleep(500 * time.Millisecond)
+				si, err := js.StreamInfo("TEST")
+				expectOk(t, err)
+				_, err = nc.Request(sds, nil, time.Second)
+				expectOk(t, err)
+
+				// Wait on new leader.
+				checkFor(t, 5*time.Second, 50*time.Millisecond, func() error {
+					si, err = js.StreamInfo("TEST")
+					if err != nil {
+						return err
+					}
+					if si.Cluster.Leader == "" {
+						return fmt.Errorf("No leader yet")
+					}
+					return nil
+				})
+			}
+		}()
+
+		// Published every 10ms
+		toc := 0
+		go func() {
+			defer wg.Done()
+			nc, js := jsClient(t, randServer())
+			defer nc.Close()
+
+			for time.Now().Before(expires) {
+				time.Sleep(10 * time.Millisecond)
+				_, err := js.Publish("foo", []byte("OK"))
+				if err == nats.ErrTimeout {
+					toc++
+					continue
+				}
+				expectOk(t, err)
+			}
+		}()
+
+		// StreamInfo calls.
+		go func() {
+			defer wg.Done()
+			nc, js := jsClient(t, randServer())
+			defer nc.Close()
+
+			for time.Now().Before(expires) {
+				time.Sleep(15 * time.Millisecond)
+				_, err := js.StreamInfo("TEST")
+				expectOk(t, err)
+			}
+		}()
+
+		wg.Wait()
+
+		// An occasional timeout can occur, but should be 0 or maybe 1 with ~10 leadership changes per test run.
+		if toc > 1 {
+			t.Fatalf("Got too many timeout errors from publish: %d", toc)
+		}
+	})
+}
diff --git a/test/kv_test.go b/test/kv_test.go
index 69c7da1..e6dd2c6 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -16,8 +16,8 @@ package test
 import (
 	"context"
 	"fmt"
+	"os"
 	"reflect"
-	"regexp"
 	"strconv"
 	"strings"
 	"testing"
@@ -420,7 +420,7 @@ func TestKeyValueDeleteTombstones(t *testing.T) {
 	}
 
 	// Now cleanup.
-	err = kv.PurgeDeletes()
+	err = kv.PurgeDeletes(nats.DeleteMarkersOlderThan(-1))
 	expectOk(t, err)
 
 	si, err := js.StreamInfo("KV_KVS")
@@ -428,6 +428,58 @@ func TestKeyValueDeleteTombstones(t *testing.T) {
 	if si.State.Msgs != 0 {
 		t.Fatalf("Expected no stream msgs to be left, got %d", si.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()
+
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "KVS", History: 10})
+	expectOk(t, err)
+
+	put := func(key, value string) {
+		t.Helper()
+		_, err := kv.Put(key, []byte(value))
+		expectOk(t, err)
+	}
+
+	put("foo", "foo1")
+	put("bar", "bar1")
+	put("foo", "foo2")
+	err = kv.Delete("foo")
+	expectOk(t, err)
+
+	time.Sleep(200 * time.Millisecond)
+
+	err = kv.Delete("bar")
+	expectOk(t, err)
+
+	err = kv.PurgeDeletes(nats.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("foo")
+	if err != nats.ErrKeyNotFound {
+		t.Fatalf("Expected all entries for key foo to be gone, got err=%v entries=%v", err, fooEntries)
+	}
+	barEntries, err := kv.History("bar")
+	expectOk(t, err)
+	if len(barEntries) != 1 {
+		t.Fatalf("Expected 1 entry, got %v", barEntries)
+	}
+	if e := barEntries[0]; e.Operation() != nats.KeyValueDelete {
+		t.Fatalf("Unexpected entry: %+v", e)
+	}
 }
 
 func TestKeyValueKeys(t *testing.T) {
@@ -505,46 +557,211 @@ func TestKeyValueKeys(t *testing.T) {
 	}
 }
 
-func TestKeyValueDiscardNew(t *testing.T) {
+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 nats.KeyWatcher) nats.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
+	}
+
+	nc1, js1 := jsClient(t, s, nats.UserInfo("a", "a"))
+	defer nc1.Close()
+
+	kv1, err := js1.CreateKeyValue(&nats.KeyValueConfig{Bucket: "Map", History: 10})
+	if err != nil {
+		t.Fatalf("Error creating kv store: %v", err)
+	}
+
+	w1, err := kv1.Watch("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 := nc2.JetStream(nats.APIPrefix("fromA"))
+	if err != nil {
+		t.Fatalf("Error getting jetstream context: %v", err)
+	}
+
+	kv2, err := js2.CreateKeyValue(&nats.KeyValueConfig{Bucket: "Map", History: 10})
+	if err != nil {
+		t.Fatalf("Error creating kv store: %v", err)
+	}
+
+	w2, err := kv2.Watch("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("map", []byte("value"))
+	if err != nil {
+		t.Fatalf("Error on put: %v", err)
+	}
+
+	// Get from kv1
+	e, err := kv1.Get("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("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("map", []byte("updated"), rev); err != nil {
+		t.Fatalf("Failed to update: %v", err)
+	}
+
+	// Get from kv1
+	e, err = kv1.Get("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("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("map"); err != nil {
+		t.Fatalf("Error on purge: %v", err)
+	}
+
+	// Check purge ok from w1
+	if e := watchNext(w1); e == nil || e.Operation() != nats.KeyValuePurge {
+		t.Fatalf("Unexpected entry: %+v", e)
+	}
+
+	// Check purge ok from w2
+	if e := watchNext(w2); e == nil || e.Operation() != nats.KeyValuePurge {
+		t.Fatalf("Unexpected entry: %+v", e)
+	}
+
+	// Delete purge records from kv2
+	if err := kv2.PurgeDeletes(nats.DeleteMarkersOlderThan(-1)); err != nil {
+		t.Fatalf("Error on purge deletes: %v", err)
+	}
+
+	// Check all gone from js1
+	if si, err := js1.StreamInfo("KV_Map"); err != nil || si == nil || si.State.Msgs != 0 {
+		t.Fatalf("Error getting stream info: err=%v si=%+v", err, si)
+	}
+
+	// Delete key from kv2
+	if err := kv2.Delete("map"); err != nil {
+		t.Fatalf("Error on delete: %v", err)
+	}
+
+	// Check key gone from kv1
+	if e, err := kv1.Get("map"); err != nats.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()
 
-	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "TEST", History: 1, MaxBytes: 256})
-	expectOk(t, err)
+	checkWindow := func(ttl, expectedDuplicates time.Duration) {
+		t.Helper()
 
-	vc := func() (major, minor, patch int) {
-		semVerRe := regexp.MustCompile(`\Av?([0-9]+)\.?([0-9]+)?\.?([0-9]+)?`)
-		m := semVerRe.FindStringSubmatch(nc.ConnectedServerVersion())
-		expectOk(t, err)
-		major, err = strconv.Atoi(m[1])
-		expectOk(t, err)
-		minor, err = strconv.Atoi(m[2])
-		expectOk(t, err)
-		patch, err = strconv.Atoi(m[3])
+		_, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "TEST", History: 5, TTL: ttl})
 		expectOk(t, err)
-		return major, minor, patch
-	}
+		defer js.DeleteKeyValue("TEST")
 
-	major, minor, patch := vc()
-	status, err := kv.Status()
-	expectOk(t, err)
-	kvs := status.(*nats.KeyValueBucketStatus)
-	si := kvs.StreamInfo()
-
-	// If we are 2.7.1 or below DiscardOld should be used.
-	// If 2.7.2 or above should be DiscardNew
-	if major <= 2 && minor <= 7 && patch <= 1 {
-		if si.Config.Discard != nats.DiscardOld {
-			t.Fatalf("Expected Discard Old for server version %d.%d.%d", major, minor, patch)
+		si, err := js.StreamInfo("KV_TEST")
+		if err != nil {
+			t.Fatalf("StreamInfo error: %v", err)
 		}
-	} else {
-		if si.Config.Discard != nats.DiscardNew {
-			t.Fatalf("Expected Discard New for server version %d.%d.%d", major, minor, patch)
+		if si.Config.Duplicates != expectedDuplicates {
+			t.Fatalf("Expected duplicates to be %v, got %v", expectedDuplicates, si.Config.Duplicates)
 		}
 	}
+
+	checkWindow(0, 2*time.Minute)
+	checkWindow(time.Hour, 2*time.Minute)
+	checkWindow(5*time.Second, 5*time.Second)
 }
 
 // Helpers
diff --git a/ws_test.go b/ws_test.go
index 0462cc7..84229b7 100644
--- a/ws_test.go
+++ b/ws_test.go
@@ -22,6 +22,7 @@ import (
 	"io"
 	"math/rand"
 	"reflect"
+	"runtime"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -599,7 +600,7 @@ func TestWSControlFrames(t *testing.T) {
 	defer nc.Close()
 
 	// Enqueue a PING and make sure that we don't break
-	nc.wsEnqueueControlMsg(wsPingMessage, []byte("this is a ping payload"))
+	nc.wsEnqueueControlMsg(true, wsPingMessage, []byte("this is a ping payload"))
 	select {
 	case e := <-dch:
 		t.Fatal(e)
@@ -860,6 +861,36 @@ func TestWSWithTLS(t *testing.T) {
 	}
 }
 
+func TestWSTlsNoConfig(t *testing.T) {
+	opts := GetDefaultOptions()
+	opts.Servers = []string{"wss://localhost:443"}
+
+	nc := &Conn{Opts: opts}
+	if err := nc.setupServerPool(); err != nil {
+		t.Fatalf("Error setting up pool: %v", err)
+	}
+	// Verify that this has set Secure/TLSConfig
+	nc.mu.Lock()
+	ok := nc.Opts.Secure && nc.Opts.TLSConfig != nil
+	nc.mu.Unlock()
+	if !ok {
+		t.Fatal("Secure and TLSConfig were not set")
+	}
+	// Now try to add a bare host:ip to the pool and verify
+	// that the wss:// scheme is added.
+	if err := nc.addURLToPool("1.2.3.4:443", true, false); err != nil {
+		t.Fatalf("Error adding to pool: %v", err)
+	}
+	nc.mu.Lock()
+	for _, srv := range nc.srvPool {
+		if srv.url.Scheme != wsSchemeTLS {
+			nc.mu.Unlock()
+			t.Fatalf("Expected scheme to be %q, got url: %s", wsSchemeTLS, srv.url)
+		}
+	}
+	nc.mu.Unlock()
+}
+
 func TestWSGossipAndReconnect(t *testing.T) {
 	o1 := testWSGetDefaultOptions(t, false)
 	o1.ServerName = "A"
@@ -1057,3 +1088,23 @@ func TestWSStress(t *testing.T) {
 		})
 	}
 }
+
+func TestWSNoDeadlockOnAuthFailure(t *testing.T) {
+	o := testWSGetDefaultOptions(t, false)
+	o.Username = "user"
+	o.Password = "pwd"
+	s := RunServerWithOptions(o)
+	defer s.Shutdown()
+
+	tm := time.AfterFunc(3*time.Second, func() {
+		buf := make([]byte, 1000000)
+		n := runtime.Stack(buf, true)
+		panic(fmt.Sprintf("Test has probably deadlocked!\n%s\n", buf[:n]))
+	})
+
+	if _, err := Connect(fmt.Sprintf("ws://127.0.0.1:%d", o.Websocket.Port)); err == nil {
+		t.Fatal("Expected auth error, did not get any error")
+	}
+
+	tm.Stop()
+}