Repository: nats-io/nats.go
Pull request #1711 (2024-09-04 14:02:03)

feat(kv): implement KeysWithFilters and ListKeysWithFilters with tests
- Added `KeysWithFilters` to filter and return a list of matching keys.
- Added `ListKeysWithFilters` to filter keys and return them as an iterable lister.
- Included test cases for both methods to validate filtering logic and key retrieval.

Resolves: #1655 

somratdutta: Thanks, @piotrpio for the detailed feedback on the PR.
I now get the approach and it makes a lot of sense.
I would love to still work on it as I have previously worked on many NATS-related PRs.
FYI : I am currently working on adding Jetstream support to Nats-Flink connector 😄 
piotrpio: Hello @somratdutta! We've merged the PR for `WatchFiltered` (https://github.com/nats-io/nats.go/pull/1739) so you can continue with this PR now :)
piotrpio: @somratdutta could you let us know if you'll be working on that? If you don't have time for that, I can continue your work :)
somratdutta: @piotrpio Thanks for the ping, I was busy with the `connector` work and the festive seasons.shenanigans :)
I will work on it tomorrow.

somratdutta: @piotrpio I have added the changes, let me know if this is fine.
Thanks. 

Patch:
diff --git a/README.md b/README.md
index 7980ecc..dbe1daf 100644
--- a/README.md
+++ b/README.md
@@ -23,7 +23,7 @@ A [Go](http://golang.org) client for the [NATS messaging system](https://nats.io
 go get github.com/nats-io/nats.go@latest
 
 # To get a specific version:
-go get github.com/nats-io/nats.go@v1.37.0
+go get github.com/nats-io/nats.go@v1.38.0
 
 # Note that the latest major version for NATS Server is v2:
 go get github.com/nats-io/nats-server/v2@latest
diff --git a/context.go b/context.go
index c19673c..382335e 100644
--- a/context.go
+++ b/context.go
@@ -88,7 +88,7 @@ func (nc *Conn) oldRequestWithContext(ctx context.Context, subj string, hdr, dat
 	inbox := nc.NewInbox()
 	ch := make(chan *Msg, RequestChanLen)
 
-	s, err := nc.subscribe(inbox, _EMPTY_, nil, ch, true, nil)
+	s, err := nc.subscribe(inbox, _EMPTY_, nil, ch, nil, true, nil)
 	if err != nil {
 		return nil, err
 	}
diff --git a/enc.go b/enc.go
index 78bcc21..34a3fae 100644
--- a/enc.go
+++ b/enc.go
@@ -258,7 +258,7 @@ func (c *EncodedConn) subscribe(subject, queue string, cb Handler) (*Subscriptio
 		cbValue.Call(oV)
 	}
 
-	return c.Conn.subscribe(subject, queue, natsCB, nil, false, nil)
+	return c.Conn.subscribe(subject, queue, natsCB, nil, nil, false, nil)
 }
 
 // FlushTimeout allows a Flush operation to have an associated timeout.
diff --git a/go.mod b/go.mod
index 76fd1ca..e8ec804 100644
--- a/go.mod
+++ b/go.mod
@@ -2,14 +2,16 @@ module github.com/nats-io/nats.go
 
 go 1.20
 
+toolchain go1.23.2
+
 require (
-	github.com/klauspost/compress v1.17.2
-	github.com/nats-io/nkeys v0.4.7
+	github.com/klauspost/compress v1.17.9
+	github.com/nats-io/nkeys v0.4.9
 	github.com/nats-io/nuid v1.0.1
-	golang.org/x/text v0.14.0
+	golang.org/x/text v0.21.0
 )
 
 require (
-	golang.org/x/crypto v0.18.0 // indirect
-	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/crypto v0.31.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
 )
diff --git a/go.sum b/go.sum
index ecf8e32..5432d9c 100644
--- a/go.sum
+++ b/go.sum
@@ -1,18 +1,15 @@
-github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
-github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
-github.com/nats-io/nkeys v0.4.6 h1:IzVe95ru2CT6ta874rt9saQRkWfe2nFj1NtvYSLqMzY=
-github.com/nats-io/nkeys v0.4.6/go.mod h1:4DxZNzenSVd1cYQoAa8948QY3QDjrHfcfVADymtkpts=
-github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
-github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
+github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
+github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
+github.com/klauspost/compress v1.17.10 h1:oXAz+Vh0PMUvJczoi+flxpnBEPxoER1IaAnU/NMPtT0=
+github.com/klauspost/compress v1.17.10/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
+github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
+github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
+github.com/nats-io/nkeys v0.4.9 h1:qe9Faq2Gxwi6RZnZMXfmGMZkg3afLLOtrU+gDZJ35b0=
+github.com/nats-io/nkeys v0.4.9/go.mod h1:jcMqs+FLG+W5YO36OX6wFIFcmpdAns+w1Wm6D3I/evE=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
-golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
-golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
-golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
-golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
-golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
-golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
-golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
-golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
+golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
diff --git a/jetstream/consumer_config.go b/jetstream/consumer_config.go
index 4e2e3d6..65f581d 100644
--- a/jetstream/consumer_config.go
+++ b/jetstream/consumer_config.go
@@ -263,6 +263,11 @@ type (
 		// Maximum number of attempts for the consumer to be recreated in a
 		// single recreation cycle. Defaults to unlimited.
 		MaxResetAttempts int
+
+		// Metadata is a set of application-defined key-value pairs for
+		// associating metadata on the consumer. This feature requires
+		// nats-server v2.10.0 or later.
+		Metadata map[string]string `json:"metadata,omitempty"`
 	}
 
 	// DeliverPolicy determines from which point to start delivering messages.
diff --git a/jetstream/kv.go b/jetstream/kv.go
index 38acbdc..152c7a4 100644
--- a/jetstream/kv.go
+++ b/jetstream/kv.go
@@ -178,6 +178,9 @@ type (
 		// the key value store in a streaming fashion (on a channel).
 		ListKeys(ctx context.Context, opts ...WatchOpt) (KeyLister, error)
 
+		// ListKeysFiltered ListKeysWithFilters returns a KeyLister for filtered keys in the bucket.
+		ListKeysFiltered(ctx context.Context, filters ...string) (KeyLister, error)
+
 		// History will return all historical values for the key (up to
 		// KeyValueMaxHistory).
 		History(ctx context.Context, key string, opts ...WatchOpt) ([]KeyValueEntry, error)
@@ -1271,6 +1274,36 @@ func (kv *kvs) ListKeys(ctx context.Context, opts ...WatchOpt) (KeyLister, error
 	return kl, nil
 }
 
+// ListKeysWithFilters returns a channel of keys matching the provided filters using WatchFiltered.
+func (kv *kvs) ListKeysFiltered(ctx context.Context, filters ...string) (KeyLister, error) {
+	watcher, err := kv.WatchFiltered(ctx, filters)
+	if err != nil {
+		return nil, err
+	}
+
+	// Reuse the existing keyLister implementation
+	kl := &keyLister{watcher: watcher, keys: make(chan string, 256)}
+
+	go func() {
+		defer close(kl.keys)
+		defer watcher.Stop()
+
+		for {
+			select {
+			case entry := <-watcher.Updates():
+				if entry == nil { // Indicates all initial values are received
+					return
+				}
+				kl.keys <- entry.Key()
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return kl, nil
+}
+
 func (kl *keyLister) Keys() <-chan string {
 	return kl.keys
 }
diff --git a/jetstream/ordered.go b/jetstream/ordered.go
index 5fe656e..055b5ef 100644
--- a/jetstream/ordered.go
+++ b/jetstream/ordered.go
@@ -393,26 +393,26 @@ func (s *orderedSubscription) Closed() <-chan struct{} {
 // reset the consumer for each subsequent Fetch call.
 // Consider using [Consumer.Consume] or [Consumer.Messages] instead.
 func (c *orderedConsumer) Fetch(batch int, opts ...FetchOpt) (MessageBatch, error) {
+	c.Lock()
 	if c.consumerType == consumerTypeConsume {
+		c.Unlock()
 		return nil, ErrOrderConsumerUsedAsConsume
 	}
-	c.currentConsumer.Lock()
 	if c.runningFetch != nil {
-		if !c.runningFetch.done {
-			c.currentConsumer.Unlock()
+		if !c.runningFetch.closed() {
 			return nil, ErrOrderedConsumerConcurrentRequests
 		}
 		if c.runningFetch.sseq != 0 {
 			c.cursor.streamSeq = c.runningFetch.sseq
 		}
 	}
-	c.currentConsumer.Unlock()
 	c.consumerType = consumerTypeFetch
 	sub := orderedSubscription{
 		consumer: c,
 		done:     make(chan struct{}),
 	}
 	c.subscription = &sub
+	c.Unlock()
 	err := c.reset()
 	if err != nil {
 		return nil, err
@@ -433,11 +433,13 @@ func (c *orderedConsumer) Fetch(batch int, opts ...FetchOpt) (MessageBatch, erro
 // reset the consumer for each subsequent Fetch call.
 // Consider using [Consumer.Consume] or [Consumer.Messages] instead.
 func (c *orderedConsumer) FetchBytes(maxBytes int, opts ...FetchOpt) (MessageBatch, error) {
+	c.Lock()
 	if c.consumerType == consumerTypeConsume {
+		c.Unlock()
 		return nil, ErrOrderConsumerUsedAsConsume
 	}
 	if c.runningFetch != nil {
-		if !c.runningFetch.done {
+		if !c.runningFetch.closed() {
 			return nil, ErrOrderedConsumerConcurrentRequests
 		}
 		if c.runningFetch.sseq != 0 {
@@ -450,6 +452,7 @@ func (c *orderedConsumer) FetchBytes(maxBytes int, opts ...FetchOpt) (MessageBat
 		done:     make(chan struct{}),
 	}
 	c.subscription = &sub
+	c.Unlock()
 	err := c.reset()
 	if err != nil {
 		return nil, err
@@ -608,6 +611,7 @@ func (c *orderedConsumer) getConsumerConfig() *ConsumerConfig {
 		Replicas:          1,
 		HeadersOnly:       c.cfg.HeadersOnly,
 		MemoryStorage:     true,
+		Metadata:          c.cfg.Metadata,
 	}
 	if len(c.cfg.FilterSubjects) == 1 {
 		cfg.FilterSubject = c.cfg.FilterSubjects[0]
diff --git a/jetstream/pull.go b/jetstream/pull.go
index 3869681..764bf2a 100644
--- a/jetstream/pull.go
+++ b/jetstream/pull.go
@@ -144,6 +144,7 @@ type (
 	}
 
 	fetchResult struct {
+		sync.Mutex
 		msgs chan Msg
 		err  error
 		done bool
@@ -780,7 +781,7 @@ func (p *pullConsumer) fetch(req *pullRequest) (MessageBatch, error) {
 		for {
 			select {
 			case msg := <-msgs:
-				p.Lock()
+				res.Lock()
 				if hbTimer != nil {
 					hbTimer.Reset(2 * req.Heartbeat)
 				}
@@ -791,11 +792,11 @@ func (p *pullConsumer) fetch(req *pullRequest) (MessageBatch, error) {
 						res.err = err
 					}
 					res.done = true
-					p.Unlock()
+					res.Unlock()
 					return
 				}
 				if !userMsg {
-					p.Unlock()
+					res.Unlock()
 					continue
 				}
 				res.msgs <- p.jetStream.toJSMsg(msg)
@@ -810,16 +811,20 @@ func (p *pullConsumer) fetch(req *pullRequest) (MessageBatch, error) {
 				}
 				if receivedMsgs == req.Batch || (req.MaxBytes != 0 && receivedBytes >= req.MaxBytes) {
 					res.done = true
-					p.Unlock()
+					res.Unlock()
 					return
 				}
-				p.Unlock()
+				res.Unlock()
 			case err := <-sub.errs:
+				res.Lock()
 				res.err = err
 				res.done = true
+				res.Unlock()
 				return
 			case <-time.After(req.Expires + 1*time.Second):
+				res.Lock()
 				res.done = true
+				res.Unlock()
 				return
 			}
 		}
@@ -828,13 +833,23 @@ func (p *pullConsumer) fetch(req *pullRequest) (MessageBatch, error) {
 }
 
 func (fr *fetchResult) Messages() <-chan Msg {
+	fr.Lock()
+	defer fr.Unlock()
 	return fr.msgs
 }
 
 func (fr *fetchResult) Error() error {
+	fr.Lock()
+	defer fr.Unlock()
 	return fr.err
 }
 
+func (fr *fetchResult) closed() bool {
+	fr.Lock()
+	defer fr.Unlock()
+	return fr.done
+}
+
 // Next is used to retrieve the next message from the stream. This
 // method will block until the message is retrieved or timeout is
 // reached.
diff --git a/js.go b/js.go
index e024fae..9467538 100644
--- a/js.go
+++ b/js.go
@@ -1527,7 +1527,7 @@ func processConsInfo(info *ConsumerInfo, userCfg *ConsumerConfig, isPullMode boo
 
 func checkConfig(s, u *ConsumerConfig) error {
 	makeErr := func(fieldName string, usrVal, srvVal any) error {
-		return fmt.Errorf("configuration requests %s to be %v, but consumer's value is %v", fieldName, usrVal, srvVal)
+		return fmt.Errorf("nats: configuration requests %s to be %v, but consumer's value is %v", fieldName, usrVal, srvVal)
 	}
 
 	if u.Durable != _EMPTY_ && u.Durable != s.Durable {
@@ -1839,7 +1839,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		ocb := cb
 		cb = func(m *Msg) { ocb(m); m.Ack() }
 	}
-	sub, err := nc.subscribe(deliver, queue, cb, ch, isSync, jsi)
+	sub, err := nc.subscribe(deliver, queue, cb, ch, nil, isSync, jsi)
 	if err != nil {
 		return nil, err
 	}
@@ -1910,7 +1910,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 				jsi.hbi = info.Config.Heartbeat
 
 				// Recreate the subscription here.
-				sub, err = nc.subscribe(jsi.deliver, queue, cb, ch, isSync, jsi)
+				sub, err = nc.subscribe(jsi.deliver, queue, cb, ch, nil, isSync, jsi)
 				if err != nil {
 					return nil, err
 				}
@@ -3114,20 +3114,27 @@ type MessageBatch interface {
 }
 
 type messageBatch struct {
+	sync.Mutex
 	msgs chan *Msg
 	err  error
 	done chan struct{}
 }
 
 func (mb *messageBatch) Messages() <-chan *Msg {
+	mb.Lock()
+	defer mb.Unlock()
 	return mb.msgs
 }
 
 func (mb *messageBatch) Error() error {
+	mb.Lock()
+	defer mb.Unlock()
 	return mb.err
 }
 
 func (mb *messageBatch) Done() <-chan struct{} {
+	mb.Lock()
+	defer mb.Unlock()
 	return mb.done
 }
 
@@ -3302,12 +3309,11 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 	}
 	var hbTimer *time.Timer
 	var hbErr error
-	hbLock := sync.Mutex{}
 	if o.hb > 0 {
 		hbTimer = time.AfterFunc(2*o.hb, func() {
-			hbLock.Lock()
+			result.Lock()
 			hbErr = ErrNoHeartbeat
-			hbLock.Unlock()
+			result.Unlock()
 			cancel()
 		})
 	}
@@ -3338,21 +3344,25 @@ func (sub *Subscription) FetchBatch(batch int, opts ...PullOpt) (MessageBatch, e
 				break
 			}
 			if usrMsg {
+				result.Lock()
 				result.msgs <- msg
+				result.Unlock()
 				requestMsgs++
 			}
 		}
 		if err != nil {
-			hbLock.Lock()
+			result.Lock()
 			if hbErr != nil {
 				result.err = hbErr
 			} else {
 				result.err = o.checkCtxErr(err)
 			}
-			hbLock.Unlock()
+			result.Unlock()
 		}
 		close(result.msgs)
+		result.Lock()
 		result.done <- struct{}{}
+		result.Unlock()
 	}()
 	return result, nil
 }
diff --git a/nats.go b/nats.go
index 3f12d61..0240fc5 100644
--- a/nats.go
+++ b/nats.go
@@ -47,7 +47,7 @@ import (
 
 // Default Constants
 const (
-	Version                   = "1.37.0"
+	Version                   = "1.38.0"
 	DefaultURL                = "nats://127.0.0.1:4222"
 	DefaultPort               = 4222
 	DefaultMaxReconnect       = 60
@@ -109,6 +109,7 @@ var (
 	ErrAuthorization               = errors.New("nats: authorization violation")
 	ErrAuthExpired                 = errors.New("nats: authentication expired")
 	ErrAuthRevoked                 = errors.New("nats: authentication revoked")
+	ErrPermissionViolation         = errors.New("nats: permissions violation")
 	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")
@@ -510,6 +511,11 @@ type Options struct {
 
 	// SkipHostLookup skips the DNS lookup for the server hostname.
 	SkipHostLookup bool
+
+	// PermissionErrOnSubscribe - if set to true, the client will return ErrPermissionViolation
+	// from SubscribeSync if the server returns a permissions error for a subscription.
+	// Defaults to false.
+	PermissionErrOnSubscribe bool
 }
 
 const (
@@ -618,17 +624,19 @@ type Subscription struct {
 	// For holding information about a JetStream consumer.
 	jsi *jsSub
 
-	delivered     uint64
-	max           uint64
-	conn          *Conn
-	mcb           MsgHandler
-	mch           chan *Msg
-	closed        bool
-	sc            bool
-	connClosed    bool
-	draining      bool
-	status        SubStatus
-	statListeners map[chan SubStatus][]SubStatus
+	delivered      uint64
+	max            uint64
+	conn           *Conn
+	mcb            MsgHandler
+	mch            chan *Msg
+	errCh          chan (error)
+	closed         bool
+	sc             bool
+	connClosed     bool
+	draining       bool
+	status         SubStatus
+	statListeners  map[chan SubStatus][]SubStatus
+	permissionsErr error
 
 	// Type of Subscription
 	typ SubscriptionType
@@ -1401,6 +1409,13 @@ func SkipHostLookup() Option {
 	}
 }
 
+func PermissionErrOnSubscribe(enabled bool) Option {
+	return func(o *Options) error {
+		o.PermissionErrOnSubscribe = enabled
+		return nil
+	}
+}
+
 // TLSHandshakeFirst is an Option to perform the TLS handshake first, that is
 // before receiving the INFO protocol. This requires the server to also be
 // configured with such option, otherwise the connection will fail.
@@ -3435,6 +3450,9 @@ slowConsumer:
 	}
 }
 
+var permissionsRe = regexp.MustCompile(`Subscription to "(\S+)"`)
+var permissionsQueueRe = regexp.MustCompile(`using queue "(\S+)"`)
+
 // processTransientError is called when the server signals a non terminal error
 // which does not close the connection or trigger a reconnect.
 // This will trigger the async error callback if set.
@@ -3444,6 +3462,27 @@ slowConsumer:
 func (nc *Conn) processTransientError(err error) {
 	nc.mu.Lock()
 	nc.err = err
+	if errors.Is(err, ErrPermissionViolation) {
+		matches := permissionsRe.FindStringSubmatch(err.Error())
+		if len(matches) >= 2 {
+			queueMatches := permissionsQueueRe.FindStringSubmatch(err.Error())
+			var q string
+			if len(queueMatches) >= 2 {
+				q = queueMatches[1]
+			}
+			subject := matches[1]
+			for _, sub := range nc.subs {
+				if sub.Subject == subject && sub.Queue == q && sub.permissionsErr == nil {
+					sub.mu.Lock()
+					if sub.errCh != nil {
+						sub.errCh <- err
+					}
+					sub.permissionsErr = err
+					sub.mu.Unlock()
+				}
+			}
+		}
+	}
 	if nc.Opts.AsyncErrorCB != nil {
 		nc.ach.push(func() { nc.Opts.AsyncErrorCB(nc, nil, err) })
 	}
@@ -3685,7 +3724,7 @@ func (nc *Conn) processErr(ie string) {
 	} else if e == MAX_CONNECTIONS_ERR {
 		close = nc.processOpErr(ErrMaxConnectionsExceeded)
 	} else if strings.HasPrefix(e, PERMISSIONS_ERR) {
-		nc.processTransientError(fmt.Errorf("nats: %s", ne))
+		nc.processTransientError(fmt.Errorf("%w: %s", ErrPermissionViolation, ne))
 	} else if strings.HasPrefix(e, MAX_SUBSCRIPTIONS_ERR) {
 		nc.processTransientError(ErrMaxSubscriptionsExceeded)
 	} else if authErr := checkAuthError(e); authErr != nil {
@@ -4028,10 +4067,6 @@ func (nc *Conn) respHandler(m *Msg) {
 // Helper to setup and send new request style requests. Return the chan to receive the response.
 func (nc *Conn) createNewRequestAndSend(subj string, hdr, data []byte) (chan *Msg, string, error) {
 	nc.mu.Lock()
-	// Do setup for the new style if needed.
-	if nc.respMap == nil {
-		nc.initNewResp()
-	}
 	// Create new literal Inbox and map to a chan msg.
 	mch := make(chan *Msg, RequestChanLen)
 	respInbox := nc.newRespInbox()
@@ -4042,7 +4077,7 @@ func (nc *Conn) createNewRequestAndSend(subj string, hdr, data []byte) (chan *Ms
 		// Create the response subscription we will use for all new style responses.
 		// This will be on an _INBOX with an additional terminal token. The subscription
 		// will be on a wildcard.
-		s, err := nc.subscribeLocked(nc.respSub, _EMPTY_, nc.respHandler, nil, false, nil)
+		s, err := nc.subscribeLocked(nc.respSub, _EMPTY_, nc.respHandler, nil, nil, false, nil)
 		if err != nil {
 			nc.mu.Unlock()
 			return nil, token, err
@@ -4140,7 +4175,7 @@ func (nc *Conn) oldRequest(subj string, hdr, data []byte, timeout time.Duration)
 	inbox := nc.NewInbox()
 	ch := make(chan *Msg, RequestChanLen)
 
-	s, err := nc.subscribe(inbox, _EMPTY_, nil, ch, true, nil)
+	s, err := nc.subscribe(inbox, _EMPTY_, nil, ch, nil, true, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -4246,14 +4281,14 @@ func (nc *Conn) respToken(respInbox string) string {
 // since it can't match more than one token.
 // Messages will be delivered to the associated MsgHandler.
 func (nc *Conn) Subscribe(subj string, cb MsgHandler) (*Subscription, error) {
-	return nc.subscribe(subj, _EMPTY_, cb, nil, false, nil)
+	return nc.subscribe(subj, _EMPTY_, cb, nil, nil, false, nil)
 }
 
 // ChanSubscribe will express interest in the given subject and place
 // all messages received on the channel.
 // You should not close the channel until sub.Unsubscribe() has been called.
 func (nc *Conn) ChanSubscribe(subj string, ch chan *Msg) (*Subscription, error) {
-	return nc.subscribe(subj, _EMPTY_, nil, ch, false, nil)
+	return nc.subscribe(subj, _EMPTY_, nil, ch, nil, false, nil)
 }
 
 // ChanQueueSubscribe will express interest in the given subject.
@@ -4263,7 +4298,7 @@ func (nc *Conn) ChanSubscribe(subj string, ch chan *Msg) (*Subscription, error)
 // You should not close the channel until sub.Unsubscribe() has been called.
 // Note: This is the same than QueueSubscribeSyncWithChan.
 func (nc *Conn) ChanQueueSubscribe(subj, group string, ch chan *Msg) (*Subscription, error) {
-	return nc.subscribe(subj, group, nil, ch, false, nil)
+	return nc.subscribe(subj, group, nil, ch, nil, false, nil)
 }
 
 // SubscribeSync will express interest on the given subject. Messages will
@@ -4273,7 +4308,11 @@ func (nc *Conn) SubscribeSync(subj string) (*Subscription, error) {
 		return nil, ErrInvalidConnection
 	}
 	mch := make(chan *Msg, nc.Opts.SubChanLen)
-	return nc.subscribe(subj, _EMPTY_, nil, mch, true, nil)
+	var errCh chan error
+	if nc.Opts.PermissionErrOnSubscribe {
+		errCh = make(chan error, 100)
+	}
+	return nc.subscribe(subj, _EMPTY_, nil, mch, errCh, true, nil)
 }
 
 // QueueSubscribe creates an asynchronous queue subscriber on the given subject.
@@ -4281,7 +4320,7 @@ func (nc *Conn) SubscribeSync(subj string) (*Subscription, error) {
 // only one member of the group will be selected to receive any given
 // message asynchronously.
 func (nc *Conn) QueueSubscribe(subj, queue string, cb MsgHandler) (*Subscription, error) {
-	return nc.subscribe(subj, queue, cb, nil, false, nil)
+	return nc.subscribe(subj, queue, cb, nil, nil, false, nil)
 }
 
 // QueueSubscribeSync creates a synchronous queue subscriber on the given
@@ -4290,7 +4329,11 @@ func (nc *Conn) QueueSubscribe(subj, queue string, cb MsgHandler) (*Subscription
 // given message synchronously using Subscription.NextMsg().
 func (nc *Conn) QueueSubscribeSync(subj, queue string) (*Subscription, error) {
 	mch := make(chan *Msg, nc.Opts.SubChanLen)
-	return nc.subscribe(subj, queue, nil, mch, true, nil)
+	var errCh chan error
+	if nc.Opts.PermissionErrOnSubscribe {
+		errCh = make(chan error, 100)
+	}
+	return nc.subscribe(subj, queue, nil, mch, errCh, true, nil)
 }
 
 // QueueSubscribeSyncWithChan will express interest in the given subject.
@@ -4300,7 +4343,7 @@ func (nc *Conn) QueueSubscribeSync(subj, queue string) (*Subscription, error) {
 // You should not close the channel until sub.Unsubscribe() has been called.
 // Note: This is the same than ChanQueueSubscribe.
 func (nc *Conn) QueueSubscribeSyncWithChan(subj, queue string, ch chan *Msg) (*Subscription, error) {
-	return nc.subscribe(subj, queue, nil, ch, false, nil)
+	return nc.subscribe(subj, queue, nil, ch, nil, false, nil)
 }
 
 // badSubject will do quick test on whether a subject is acceptable.
@@ -4324,16 +4367,16 @@ func badQueue(qname string) bool {
 }
 
 // subscribe is the internal subscribe function that indicates interest in a subject.
-func (nc *Conn) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync bool, js *jsSub) (*Subscription, error) {
+func (nc *Conn) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, errCh chan (error), isSync bool, js *jsSub) (*Subscription, error) {
 	if nc == nil {
 		return nil, ErrInvalidConnection
 	}
 	nc.mu.Lock()
 	defer nc.mu.Unlock()
-	return nc.subscribeLocked(subj, queue, cb, ch, isSync, js)
+	return nc.subscribeLocked(subj, queue, cb, ch, errCh, isSync, js)
 }
 
-func (nc *Conn) subscribeLocked(subj, queue string, cb MsgHandler, ch chan *Msg, isSync bool, js *jsSub) (*Subscription, error) {
+func (nc *Conn) subscribeLocked(subj, queue string, cb MsgHandler, ch chan *Msg, errCh chan (error), isSync bool, js *jsSub) (*Subscription, error) {
 	if nc == nil {
 		return nil, ErrInvalidConnection
 	}
@@ -4384,6 +4427,7 @@ func (nc *Conn) subscribeLocked(subj, queue string, cb MsgHandler, ch chan *Msg,
 	} else { // Sync Subscription
 		sub.typ = SyncSubscription
 		sub.mch = ch
+		sub.errCh = errCh
 	}
 
 	nc.subsMu.Lock()
@@ -4828,16 +4872,92 @@ func (s *Subscription) NextMsg(timeout time.Duration) (*Msg, error) {
 	t := globalTimerPool.Get(timeout)
 	defer globalTimerPool.Put(t)
 
+	if s.errCh != nil {
+		select {
+		case msg, ok = <-mch:
+			if !ok {
+				return nil, s.getNextMsgErr()
+			}
+			if err := s.processNextMsgDelivered(msg); err != nil {
+				return nil, err
+			}
+		case err := <-s.errCh:
+			return nil, err
+		case <-t.C:
+			return nil, ErrTimeout
+		}
+	} else {
+		select {
+		case msg, ok = <-mch:
+			if !ok {
+				return nil, s.getNextMsgErr()
+			}
+			if err := s.processNextMsgDelivered(msg); err != nil {
+				return nil, err
+			}
+		case <-t.C:
+			return nil, ErrTimeout
+		}
+	}
+
+	return msg, nil
+}
+
+// nextMsgNoTimeout works similarly to Subscription.NextMsg() but will not
+// time out. It is only used internally for non-timeout subscription iterator.
+func (s *Subscription) nextMsgNoTimeout() (*Msg, error) {
+	if s == nil {
+		return nil, ErrBadSubscription
+	}
+
+	s.mu.Lock()
+	err := s.validateNextMsgState(false)
+	if err != nil {
+		s.mu.Unlock()
+		return nil, err
+	}
+
+	// snapshot
+	mch := s.mch
+	s.mu.Unlock()
+
+	var ok bool
+	var msg *Msg
+
+	// If something is available right away, let's optimize that case.
 	select {
 	case msg, ok = <-mch:
+		if !ok {
+			return nil, s.getNextMsgErr()
+		}
+		if err := s.processNextMsgDelivered(msg); err != nil {
+			return nil, err
+		} else {
+			return msg, nil
+		}
+	default:
+	}
+
+	if s.errCh != nil {
+		select {
+		case msg, ok = <-mch:
+			if !ok {
+				return nil, s.getNextMsgErr()
+			}
+			if err := s.processNextMsgDelivered(msg); err != nil {
+				return nil, err
+			}
+		case err := <-s.errCh:
+			return nil, err
+		}
+	} else {
+		msg, ok = <-mch
 		if !ok {
 			return nil, s.getNextMsgErr()
 		}
 		if err := s.processNextMsgDelivered(msg); err != nil {
 			return nil, err
 		}
-	case <-t.C:
-		return nil, ErrTimeout
 	}
 
 	return msg, nil
@@ -4860,6 +4980,12 @@ func (s *Subscription) validateNextMsgState(pullSubInternal bool) error {
 	if s.mcb != nil {
 		return ErrSyncSubRequired
 	}
+	// if this subscription previously had a permissions error
+	// and no reconnect has been attempted, return the permissions error
+	// since the subscription does not exist on the server
+	if s.conn.Opts.PermissionErrOnSubscribe && s.permissionsErr != nil {
+		return s.permissionsErr
+	}
 	if s.sc {
 		s.changeSubStatus(SubscriptionActive)
 		s.sc = false
@@ -5235,6 +5361,9 @@ func (nc *Conn) resendSubscriptions() {
 	for _, s := range subs {
 		adjustedMax := uint64(0)
 		s.mu.Lock()
+		// when resending subscriptions, the permissions error should be cleared
+		// since the user may have fixed the permissions issue
+		s.permissionsErr = nil
 		if s.max > 0 {
 			if s.delivered < s.max {
 				adjustedMax = s.max - s.delivered
@@ -5273,9 +5402,6 @@ func (nc *Conn) clearPendingFlushCalls() {
 // This will clear any pending Request calls.
 // Lock is assumed to be held by the caller.
 func (nc *Conn) clearPendingRequestCalls() {
-	if nc.respMap == nil {
-		return
-	}
 	for key, ch := range nc.respMap {
 		if ch != nil {
 			close(ch)
diff --git a/nats_iter.go b/nats_iter.go
new file mode 100644
index 0000000..0f4d1e2
--- /dev/null
+++ b/nats_iter.go
@@ -0,0 +1,73 @@
+// Copyright 2012-2024 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.
+
+//go:build go1.23
+// +build go1.23
+
+package nats
+
+import (
+	"errors"
+	"iter"
+	"time"
+)
+
+// Msgs returns an iter.Seq2[*Msg, error] that can be used to iterate over
+// messages. It can only be used with a subscription that has been created with
+// SubscribeSync or QueueSubscribeSync, otherwise it will return an error on the
+// first iteration.
+//
+// The iterator will block until a message is available. The
+// subscription will not be closed when the iterator is done.
+func (sub *Subscription) Msgs() iter.Seq2[*Msg, error] {
+	return func(yield func(*Msg, error) bool) {
+		for {
+			msg, err := sub.nextMsgNoTimeout()
+			if err != nil {
+				yield(nil, err)
+				return
+			}
+			if !yield(msg, nil) {
+				return
+			}
+
+		}
+	}
+}
+
+// MsgsTimeout returns an iter.Seq2[*Msg, error] that can be used to iterate
+// over messages. It can only be used with a subscription that has been created
+// with SubscribeSync or QueueSubscribeSync, otherwise it will return an error
+// on the first iteration.
+//
+// The iterator will block until a message is available or the timeout is
+// reached. If the timeout is reached, the iterator will return nats.ErrTimeout
+// but it will not be closed.
+func (sub *Subscription) MsgsTimeout(timeout time.Duration) iter.Seq2[*Msg, error] {
+	return func(yield func(*Msg, error) bool) {
+		for {
+			msg, err := sub.NextMsg(timeout)
+			if err != nil {
+				if !yield(nil, err) {
+					return
+				}
+				if !errors.Is(err, ErrTimeout) {
+					return
+				}
+			}
+			if !yield(msg, nil) {
+				return
+			}
+		}
+	}
+}
diff --git a/netchan.go b/netchan.go
index 3722d9f..35d9214 100644
--- a/netchan.go
+++ b/netchan.go
@@ -113,5 +113,5 @@ func (c *EncodedConn) bindRecvChan(subject, queue string, channel any) (*Subscri
 		chVal.Send(oPtr)
 	}
 
-	return c.Conn.subscribe(subject, queue, cb, nil, false, nil)
+	return c.Conn.subscribe(subject, queue, cb, nil, nil, false, nil)
 }
diff --git a/rand.go b/rand.go
deleted file mode 100644
index 0cdee0a..0000000
--- a/rand.go
+++ /dev/null
@@ -1,29 +0,0 @@
-// Copyright 2023 The NATS Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//go:build !go1.20
-// +build !go1.20
-
-// A Go client for the NATS messaging system (https://nats.io).
-package nats
-
-import (
-	"math/rand"
-	"time"
-)
-
-func init() {
-	// This is not needed since Go 1.20 because now rand.Seed always happens
-	// by default (uses runtime.fastrand64 instead as source).
-	rand.Seed(time.Now().UnixNano())
-}


Test patch:
diff --git a/go_test.mod b/go_test.mod
index 20e1ab7..cb9f286 100644
--- a/go_test.mod
+++ b/go_test.mod
@@ -9,17 +9,17 @@ require (
 	github.com/klauspost/compress v1.17.9
 	github.com/nats-io/jwt v1.2.2
 	github.com/nats-io/nats-server/v2 v2.10.17
-	github.com/nats-io/nkeys v0.4.7
+	github.com/nats-io/nkeys v0.4.9
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/goleak v1.3.0
-	golang.org/x/text v0.16.0
+	golang.org/x/text v0.21.0
 	google.golang.org/protobuf v1.23.0
 )
 
 require (
 	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/nats-io/jwt/v2 v2.5.7 // indirect
-	golang.org/x/crypto v0.24.0 // indirect
-	golang.org/x/sys v0.21.0 // indirect
+	golang.org/x/crypto v0.31.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 )
diff --git a/go_test.sum b/go_test.sum
index df0ef6d..d0516d8 100644
--- a/go_test.sum
+++ b/go_test.sum
@@ -22,8 +22,8 @@ github.com/nats-io/jwt/v2 v2.5.7/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wT
 github.com/nats-io/nats-server/v2 v2.10.17 h1:PTVObNBD3TZSNUDgzFb1qQsQX4mOgFmOuG9vhT+KBUY=
 github.com/nats-io/nats-server/v2 v2.10.17/go.mod h1:5OUyc4zg42s/p2i92zbbqXvUNsbF0ivdTLKshVMn2YQ=
 github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
-github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
-github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
+github.com/nats-io/nkeys v0.4.9 h1:qe9Faq2Gxwi6RZnZMXfmGMZkg3afLLOtrU+gDZJ35b0=
+github.com/nats-io/nkeys v0.4.9/go.mod h1:jcMqs+FLG+W5YO36OX6wFIFcmpdAns+w1Wm6D3I/evE=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -34,17 +34,17 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.24.0 h1:mnl8DM0o513X8fdIkmyFE/5hTYxbwYOjDS/+rK6qpRI=
-golang.org/x/crypto v0.24.0/go.mod h1:Z1PMYSOR5nyMcyAVAIQSKCDwalqy85Aqn1x3Ws4L5DM=
+golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
+golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws=
-golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
-golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
 golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
diff --git a/jetstream/test/kv_test.go b/jetstream/test/kv_test.go
index 42c0d28..97194a1 100644
--- a/jetstream/test/kv_test.go
+++ b/jetstream/test/kv_test.go
@@ -1037,6 +1037,68 @@ func TestKeyValueListKeys(t *testing.T) {
 	}
 }
 
+func TestListKeysFiltered(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	// Create Key-Value store.
+	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "KVS", History: 2})
+	expectOk(t, err)
+
+	// Helper function to add key-value pairs.
+	putKeys := func(data map[string]string) {
+		for key, value := range data {
+			t.Helper()
+			_, err := kv.Put(ctx, key, []byte(value))
+			expectOk(t, err)
+		}
+	}
+
+	// Add key-value pairs.
+	putKeys(map[string]string{
+		"apple":  "fruit",
+		"banana": "fruit",
+		"carrot": "vegetable",
+	})
+
+	// Use filters to list keys matching "apple".
+	filters := []string{"apple"}
+	keyLister, err := kv.ListKeysFiltered(ctx, filters...)
+	expectOk(t, err)
+
+	// Collect filtered keys from KeyLister
+	var filteredKeys []string
+	for key := range keyLister.Keys() {
+		filteredKeys = append(filteredKeys, key)
+	}
+
+	// Validate expected keys.
+	expectedKeys := []string{"apple"}
+	if len(filteredKeys) != len(expectedKeys) {
+		t.Fatalf("Expected %d filtered key(s), got %d", len(expectedKeys), len(filteredKeys))
+	}
+
+	for _, key := range expectedKeys {
+		if !contains(filteredKeys, key) {
+			t.Fatalf("Expected key %s in filtered keys, but not found", key)
+		}
+	}
+}
+
+func contains(slice []string, key string) bool {
+	for _, k := range slice {
+		if k == key {
+			return true
+		}
+	}
+	return false
+}
+
 func TestKeyValueCrossAccounts(t *testing.T) {
 	conf := createConfFile(t, []byte(`
 		listen: 127.0.0.1:-1
diff --git a/jetstream/test/ordered_test.go b/jetstream/test/ordered_test.go
index 5a6231b..9712567 100644
--- a/jetstream/test/ordered_test.go
+++ b/jetstream/test/ordered_test.go
@@ -580,131 +580,123 @@ func TestOrderedConsumerConsume(t *testing.T) {
 	})
 
 	t.Run("wait for closed after drain", func(t *testing.T) {
-		for i := 0; i < 10; i++ {
-			t.Run(fmt.Sprintf("run %d", i), func(t *testing.T) {
-				srv := RunBasicJetStreamServer()
-				defer shutdownJSServerAndRemoveStorage(t, srv)
-				nc, err := nats.Connect(srv.ClientURL())
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 
-				js, err := jetstream.New(nc)
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
-				defer nc.Close()
-				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-				defer cancel()
-				s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
-				c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
-				msgs := make([]jetstream.Msg, 0)
-				lock := sync.Mutex{}
-				publishTestMsgs(t, js)
-				cc, err := c.Consume(func(msg jetstream.Msg) {
-					time.Sleep(50 * time.Millisecond)
-					msg.Ack()
-					lock.Lock()
-					msgs = append(msgs, msg)
-					lock.Unlock()
-				})
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
-				closed := cc.Closed()
-				time.Sleep(100 * time.Millisecond)
-				if err := s.DeleteConsumer(context.Background(), c.CachedInfo().Name); err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
-				publishTestMsgs(t, js)
-
-				// wait for the consumer to be recreated before calling drain
-				for i := 0; i < 5; i++ {
-					_, err = c.Info(ctx)
-					if err != nil {
-						if errors.Is(err, jetstream.ErrConsumerNotFound) {
-							time.Sleep(100 * time.Millisecond)
-							continue
-						}
-						t.Fatalf("Unexpected error: %v", err)
-					}
-					break
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		msgs := make([]jetstream.Msg, 0)
+		lock := sync.Mutex{}
+		publishTestMsgs(t, js)
+		cc, err := c.Consume(func(msg jetstream.Msg) {
+			time.Sleep(50 * time.Millisecond)
+			msg.Ack()
+			lock.Lock()
+			msgs = append(msgs, msg)
+			lock.Unlock()
+		})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		closed := cc.Closed()
+		time.Sleep(100 * time.Millisecond)
+		if err := s.DeleteConsumer(context.Background(), c.CachedInfo().Name); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		publishTestMsgs(t, js)
+
+		// wait for the consumer to be recreated before calling drain
+		for i := 0; i < 5; i++ {
+			_, err = c.Info(ctx)
+			if err != nil {
+				if errors.Is(err, jetstream.ErrConsumerNotFound) {
+					time.Sleep(100 * time.Millisecond)
+					continue
 				}
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			break
+		}
 
-				cc.Drain()
+		cc.Drain()
 
-				select {
-				case <-closed:
-				case <-time.After(5 * time.Second):
-					t.Fatalf("Timeout waiting for consume to be closed")
-				}
+		select {
+		case <-closed:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Timeout waiting for consume to be closed")
+		}
 
-				if len(msgs) != 2*len(testMsgs) {
-					t.Fatalf("Unexpected received message count after consume closed; want %d; got %d", 2*len(testMsgs), len(msgs))
-				}
-			})
+		if len(msgs) != 2*len(testMsgs) {
+			t.Fatalf("Unexpected received message count after consume closed; want %d; got %d", 2*len(testMsgs), len(msgs))
 		}
 	})
 
 	t.Run("wait for closed on already closed consume", func(t *testing.T) {
-		for i := 0; i < 10; i++ {
-			t.Run(fmt.Sprintf("run %d", i), func(t *testing.T) {
-				srv := RunBasicJetStreamServer()
-				defer shutdownJSServerAndRemoveStorage(t, srv)
-				nc, err := nats.Connect(srv.ClientURL())
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
+		srv := RunBasicJetStreamServer()
+		defer shutdownJSServerAndRemoveStorage(t, srv)
+		nc, err := nats.Connect(srv.ClientURL())
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 
-				js, err := jetstream.New(nc)
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
-				defer nc.Close()
-				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-				defer cancel()
-				s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
-				c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
-				msgs := make([]jetstream.Msg, 0)
-				lock := sync.Mutex{}
-				publishTestMsgs(t, js)
-				cc, err := c.Consume(func(msg jetstream.Msg) {
-					time.Sleep(50 * time.Millisecond)
-					msg.Ack()
-					lock.Lock()
-					msgs = append(msgs, msg)
-					lock.Unlock()
-				})
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
-				time.Sleep(100 * time.Millisecond)
-				if err := s.DeleteConsumer(context.Background(), c.CachedInfo().Name); err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
+		js, err := jetstream.New(nc)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		defer nc.Close()
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c, err := s.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		msgs := make([]jetstream.Msg, 0)
+		lock := sync.Mutex{}
+		publishTestMsgs(t, js)
+		cc, err := c.Consume(func(msg jetstream.Msg) {
+			time.Sleep(50 * time.Millisecond)
+			msg.Ack()
+			lock.Lock()
+			msgs = append(msgs, msg)
+			lock.Unlock()
+		})
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		time.Sleep(100 * time.Millisecond)
+		if err := s.DeleteConsumer(context.Background(), c.CachedInfo().Name); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 
-				cc.Stop()
+		cc.Stop()
 
-				time.Sleep(100 * time.Millisecond)
+		time.Sleep(100 * time.Millisecond)
 
-				select {
-				case <-cc.Closed():
-				case <-time.After(5 * time.Second):
-					t.Fatalf("Timeout waiting for consume to be closed")
-				}
-			})
+		select {
+		case <-cc.Closed():
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Timeout waiting for consume to be closed")
 		}
 	})
 }
@@ -2014,6 +2006,7 @@ func TestOrderedConsumerConfig(t *testing.T) {
 				ReplayPolicy:      jetstream.ReplayOriginalPolicy,
 				InactiveThreshold: 10 * time.Second,
 				HeadersOnly:       true,
+				Metadata:          map[string]string{"foo": "a"},
 			},
 			expected: jetstream.ConsumerConfig{
 				FilterSubjects:    []string{"foo.a", "foo.b"},
@@ -2026,6 +2019,7 @@ func TestOrderedConsumerConfig(t *testing.T) {
 				Replicas:          1,
 				MemoryStorage:     true,
 				HeadersOnly:       true,
+				Metadata:          map[string]string{"foo": "a"},
 			},
 		},
 		{
@@ -2037,6 +2031,7 @@ func TestOrderedConsumerConfig(t *testing.T) {
 				ReplayPolicy:      jetstream.ReplayOriginalPolicy,
 				InactiveThreshold: 10 * time.Second,
 				HeadersOnly:       true,
+				Metadata:          map[string]string{"foo": "a"},
 			},
 			expected: jetstream.ConsumerConfig{
 				FilterSubjects:    []string{"foo.a", "foo.b"},
@@ -2049,6 +2044,7 @@ func TestOrderedConsumerConfig(t *testing.T) {
 				Replicas:          1,
 				MemoryStorage:     true,
 				HeadersOnly:       true,
+				Metadata:          map[string]string{"foo": "a"},
 			},
 		},
 		{
diff --git a/jetstream/test/pull_test.go b/jetstream/test/pull_test.go
index 4042e52..dcca6d1 100644
--- a/jetstream/test/pull_test.go
+++ b/jetstream/test/pull_test.go
@@ -477,6 +477,91 @@ func TestPullConsumerFetch(t *testing.T) {
 	})
 }
 
+func TestPullConsumerFetchRace(t *testing.T) {
+	srv := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, srv)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer nc.Close()
+
+	s, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	c, err := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{AckPolicy: jetstream.AckExplicitPolicy})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	for i := 0; i < 3; i++ {
+		if _, err := js.Publish(context.Background(), "FOO.123", []byte(fmt.Sprintf("msg-%d", i))); err != nil {
+			t.Fatalf("Unexpected error during publish: %s", err)
+		}
+	}
+	msgs, err := c.Fetch(5)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	errCh := make(chan error)
+	go func() {
+		for {
+			err := msgs.Error()
+			if err != nil {
+				errCh <- err
+				return
+			}
+		}
+	}()
+	deleteErrCh := make(chan error, 1)
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		if err := s.DeleteConsumer(ctx, c.CachedInfo().Name); err != nil {
+			deleteErrCh <- err
+		}
+		close(deleteErrCh)
+	}()
+
+	var i int
+	for msg := range msgs.Messages() {
+		if string(msg.Data()) != fmt.Sprintf("msg-%d", i) {
+			t.Fatalf("Invalid msg on index %d; expected: %s; got: %s", i, fmt.Sprintf("msg-%d", i), string(msg.Data()))
+		}
+		i++
+	}
+	if i != 3 {
+		t.Fatalf("Invalid number of messages received; want: %d; got: %d", 5, i)
+	}
+	select {
+	case err := <-errCh:
+		if !errors.Is(err, jetstream.ErrConsumerDeleted) {
+			t.Fatalf("Expected error: %v; got: %v", jetstream.ErrConsumerDeleted, err)
+		}
+	case <-time.After(1 * time.Second):
+		t.Fatalf("Expected error: %v; got: %v", jetstream.ErrConsumerDeleted, nil)
+	}
+
+	// wait until the consumer is deleted, otherwise we may close the connection
+	// before the consumer delete response is received
+	select {
+	case ert, ok := <-deleteErrCh:
+		if !ok {
+			break
+		}
+		t.Fatalf("Error deleting consumer: %s", ert)
+	case <-time.After(1 * time.Second):
+		t.Fatalf("Expected done to be closed")
+	}
+}
+
 func TestPullConsumerFetchBytes(t *testing.T) {
 	testSubject := "FOO.123"
 	msg := [10]byte{}
diff --git a/test/js_test.go b/test/js_test.go
index db791eb..91a259b 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -10788,3 +10788,83 @@ func TestJetStreamTransform(t *testing.T) {
 	}
 
 }
+
+func TestPullConsumerFetchRace(t *testing.T) {
+	srv := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, srv)
+
+	nc, js := jsClient(t, srv)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "foo", Subjects: []string{"FOO.*"}})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	for i := 0; i < 3; i++ {
+		if _, err := js.Publish("FOO.123", []byte(fmt.Sprintf("msg-%d", i))); err != nil {
+			t.Fatalf("Unexpected error during publish: %s", err)
+		}
+	}
+	sub, err := js.PullSubscribe("FOO.123", "")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	cons, err := sub.ConsumerInfo()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	msgs, err := sub.FetchBatch(5)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	errCh := make(chan error)
+	go func() {
+		for {
+			err := msgs.Error()
+			if err != nil {
+				errCh <- err
+				return
+			}
+		}
+	}()
+	deleteErrCh := make(chan error, 1)
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		if err := js.DeleteConsumer("foo", cons.Name); err != nil {
+			deleteErrCh <- err
+		}
+		close(deleteErrCh)
+	}()
+
+	var i int
+	for msg := range msgs.Messages() {
+		if string(msg.Data) != fmt.Sprintf("msg-%d", i) {
+			t.Fatalf("Invalid msg on index %d; expected: %s; got: %s", i, fmt.Sprintf("msg-%d", i), string(msg.Data))
+		}
+		i++
+	}
+	if i != 3 {
+		t.Fatalf("Invalid number of messages received; want: %d; got: %d", 5, i)
+	}
+	select {
+	case err := <-errCh:
+		if !errors.Is(err, nats.ErrConsumerDeleted) {
+			t.Fatalf("Expected error: %v; got: %v", nats.ErrConsumerDeleted, err)
+		}
+	case <-time.After(1 * time.Second):
+		t.Fatalf("Expected error: %v; got: %v", nats.ErrConsumerDeleted, nil)
+	}
+
+	// wait until the consumer is deleted, otherwise we may close the connection
+	// before the consumer delete response is received
+	select {
+	case ert, ok := <-deleteErrCh:
+		if !ok {
+			break
+		}
+		t.Fatalf("Error deleting consumer: %s", ert)
+	case <-time.After(1 * time.Second):
+		t.Fatalf("Expected done to be closed")
+	}
+}
diff --git a/test/nats_iter_test.go b/test/nats_iter_test.go
new file mode 100644
index 0000000..a762aec
--- /dev/null
+++ b/test/nats_iter_test.go
@@ -0,0 +1,337 @@
+// Copyright 2012-2024 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.
+
+//go:build go1.23
+// +build go1.23
+
+package test
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+func TestSubscribeIterator(t *testing.T) {
+	t.Run("with timeout", func(t *testing.T) {
+		s := RunServerOnPort(-1)
+		defer s.Shutdown()
+
+		nc, err := nats.Connect(s.ClientURL(), nats.PermissionErrOnSubscribe(true))
+		if err != nil {
+			t.Fatalf("Error on connect: %v", err)
+		}
+		defer nc.Close()
+
+		sub, err := nc.SubscribeSync("foo")
+		if err != nil {
+			t.Fatal("Failed to subscribe: ", err)
+		}
+		defer sub.Unsubscribe()
+
+		total := 100
+		for i := 0; i < total/2; i++ {
+			if err := nc.Publish("foo", []byte("Hello")); err != nil {
+				t.Fatalf("Error on publish: %v", err)
+			}
+		}
+
+		// publish some more messages asynchronously
+		errCh := make(chan error, 1)
+		go func() {
+			for i := 0; i < total/2; i++ {
+				if err := nc.Publish("foo", []byte("Hello")); err != nil {
+					errCh <- err
+					return
+				}
+				time.Sleep(10 * time.Millisecond)
+			}
+			close(errCh)
+		}()
+
+		received := 0
+		for _, err := range sub.MsgsTimeout(100 * time.Millisecond) {
+			if err != nil {
+				if !errors.Is(err, nats.ErrTimeout) {
+					t.Fatalf("Error on subscribe: %v", err)
+				}
+				break
+			} else {
+				received++
+			}
+		}
+		if received != total {
+			t.Fatalf("Expected %d messages, got %d", total, received)
+		}
+	})
+
+	t.Run("no timeout", func(t *testing.T) {
+		s := RunServerOnPort(-1)
+		defer s.Shutdown()
+
+		nc, err := nats.Connect(s.ClientURL(), nats.PermissionErrOnSubscribe(true))
+		if err != nil {
+			t.Fatalf("Error on connect: %v", err)
+		}
+		defer nc.Close()
+
+		sub, err := nc.SubscribeSync("foo")
+		if err != nil {
+			t.Fatal("Failed to subscribe: ", err)
+		}
+		defer sub.Unsubscribe()
+
+		// Send some messages to ourselves.
+		total := 100
+		for i := 0; i < total/2; i++ {
+			if err := nc.Publish("foo", []byte("Hello")); err != nil {
+				t.Fatalf("Error on publish: %v", err)
+			}
+		}
+
+		received := 0
+
+		// publish some more messages asynchronously
+		errCh := make(chan error, 1)
+		go func() {
+			for i := 0; i < total/2; i++ {
+				if err := nc.Publish("foo", []byte("Hello")); err != nil {
+					errCh <- err
+					return
+				}
+				time.Sleep(10 * time.Millisecond)
+			}
+			close(errCh)
+		}()
+
+		for _, err := range sub.Msgs() {
+			if err != nil {
+				t.Fatalf("Error getting msg: %v", err)
+			}
+			received++
+			if received >= total {
+				break
+			}
+		}
+		err = <-errCh
+		if err != nil {
+			t.Fatalf("Error on publish: %v", err)
+		}
+		_, err = sub.NextMsg(100 * time.Millisecond)
+		if !errors.Is(err, nats.ErrTimeout) {
+			t.Fatalf("Expected timeout waiting for next message, got %v", err)
+		}
+	})
+
+	t.Run("permissions violation", func(t *testing.T) {
+		conf := createConfFile(t, []byte(`
+			listen: 127.0.0.1:-1
+			authorization: {
+				users = [
+					{
+						user: test
+						password: test
+						permissions: {
+							subscribe: {
+								deny: "foo"
+							}
+						}
+					}
+				]
+			}
+		`))
+		defer os.Remove(conf)
+
+		s, _ := RunServerWithConfig(conf)
+		defer s.Shutdown()
+
+		nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("test", "test"), nats.PermissionErrOnSubscribe(true))
+		if err != nil {
+			t.Fatalf("Error on connect: %v", err)
+		}
+		defer nc.Close()
+
+		sub, err := nc.SubscribeSync("foo")
+		if err != nil {
+			t.Fatalf("Error on subscribe: %v", err)
+		}
+		defer sub.Unsubscribe()
+
+		errs := make(chan error)
+		go func() {
+			var err error
+			for _, err = range sub.Msgs() {
+				break
+			}
+			errs <- err
+		}()
+
+		select {
+		case e := <-errs:
+			if !errors.Is(e, nats.ErrPermissionViolation) {
+				t.Fatalf("Expected permissions error, got %v", e)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatalf("Did not get the permission error")
+		}
+
+		_, err = sub.NextMsg(100 * time.Millisecond)
+		if !errors.Is(err, nats.ErrPermissionViolation) {
+			t.Fatalf("Expected permissions violation error, got %v", err)
+		}
+	})
+
+	t.Run("attempt iterator on async sub", func(t *testing.T) {
+		s := RunServerOnPort(-1)
+		defer s.Shutdown()
+
+		nc, err := nats.Connect(s.ClientURL(), nats.PermissionErrOnSubscribe(true))
+		if err != nil {
+			t.Fatalf("Error on connect: %v", err)
+		}
+		defer nc.Close()
+
+		sub, err := nc.Subscribe("foo", func(msg *nats.Msg) {})
+		if err != nil {
+			t.Fatal("Failed to subscribe: ", err)
+		}
+		defer sub.Unsubscribe()
+
+		for _, err := range sub.MsgsTimeout(100 * time.Millisecond) {
+			if !errors.Is(err, nats.ErrSyncSubRequired) {
+				t.Fatalf("Error on subscribe: %v", err)
+			}
+		}
+		for _, err := range sub.Msgs() {
+			if !errors.Is(err, nats.ErrSyncSubRequired) {
+				t.Fatalf("Error on subscribe: %v", err)
+			}
+		}
+	})
+}
+
+func TestQueueSubscribeIterator(t *testing.T) {
+	t.Run("basic", func(t *testing.T) {
+		s := RunServerOnPort(-1)
+		defer s.Shutdown()
+
+		nc, err := nats.Connect(s.ClientURL())
+		if err != nil {
+			t.Fatalf("Error on connect: %v", err)
+		}
+		defer nc.Close()
+
+		subs := make([]*nats.Subscription, 4)
+		for i := 0; i < 4; i++ {
+			sub, err := nc.QueueSubscribeSync("foo", "q")
+			if err != nil {
+				t.Fatal("Failed to subscribe: ", err)
+			}
+			subs[i] = sub
+			defer sub.Unsubscribe()
+		}
+
+		// Send some messages to ourselves.
+		total := 100
+		for i := 0; i < total; i++ {
+			if err := nc.Publish("foo", []byte(fmt.Sprintf("%d", i))); err != nil {
+				t.Fatalf("Error on publish: %v", err)
+			}
+		}
+
+		wg := sync.WaitGroup{}
+		wg.Add(100)
+		startWg := sync.WaitGroup{}
+		startWg.Add(4)
+
+		for i := range subs {
+			go func(i int) {
+				startWg.Done()
+				for _, err := range subs[i].MsgsTimeout(100 * time.Millisecond) {
+					if err != nil {
+						break
+					}
+					wg.Done()
+				}
+			}(i)
+		}
+
+		startWg.Wait()
+
+		wg.Wait()
+
+		for _, sub := range subs {
+			if _, err = sub.NextMsg(100 * time.Millisecond); !errors.Is(err, nats.ErrTimeout) {
+				t.Fatalf("Expected timeout waiting for next message, got %v", err)
+			}
+		}
+	})
+
+	t.Run("permissions violation", func(t *testing.T) {
+		conf := createConfFile(t, []byte(`
+			listen: 127.0.0.1:-1
+			authorization: {
+				users = [
+					{
+						user: test
+						password: test
+						permissions: {
+							subscribe: {
+								deny: "foo"
+							}
+						}
+					}
+				]
+			}
+		`))
+		defer os.Remove(conf)
+
+		s, _ := RunServerWithConfig(conf)
+		defer s.Shutdown()
+
+		nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("test", "test"), nats.PermissionErrOnSubscribe(true))
+		if err != nil {
+			t.Fatalf("Error on connect: %v", err)
+		}
+		defer nc.Close()
+
+		sub, err := nc.QueueSubscribeSync("foo", "q")
+		if err != nil {
+			t.Fatalf("Error on subscribe: %v", err)
+		}
+		defer sub.Unsubscribe()
+
+		errs := make(chan error)
+		go func() {
+			var err error
+			for _, err = range sub.MsgsTimeout(2 * time.Second) {
+				break
+			}
+			errs <- err
+		}()
+
+		select {
+		case e := <-errs:
+			if !errors.Is(e, nats.ErrPermissionViolation) {
+				t.Fatalf("Expected permissions error, got %v", e)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatalf("Did not get the permission error")
+		}
+	})
+}
diff --git a/test/sub_test.go b/test/sub_test.go
index 559efc5..01b78d0 100644
--- a/test/sub_test.go
+++ b/test/sub_test.go
@@ -1770,3 +1770,92 @@ func TestMaxSubscriptionsExceeded(t *testing.T) {
 	// wait for the server to process the SUBs
 	time.Sleep(100 * time.Millisecond)
 }
+
+func TestSubscribeSyncPermissionError(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+	listen: 127.0.0.1:-1
+	authorization: {
+		users = [
+			{
+				user: test
+				password: test
+				permissions: {
+					subscribe: {
+						deny: "foo"
+					}
+				}
+			}
+		]
+	}
+`))
+	defer os.Remove(conf)
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	t.Run("PermissionErrOnSubscribe enabled", func(t *testing.T) {
+
+		nc, err := nats.Connect(s.ClientURL(),
+			nats.UserInfo("test", "test"),
+			nats.PermissionErrOnSubscribe(true),
+			nats.ErrorHandler(func(*nats.Conn, *nats.Subscription, error) {}))
+		if err != nil {
+			t.Fatalf("Error on connect: %v", err)
+		}
+		defer nc.Close()
+
+		subs := make([]*nats.Subscription, 0, 100)
+		for i := 0; i < 10; i++ {
+			var subject string
+			if i%2 == 0 {
+				subject = "foo"
+			} else {
+				subject = "bar"
+			}
+			sub, err := nc.SubscribeSync(subject)
+			if err != nil {
+				t.Fatalf("Error on subscribe: %v", err)
+			}
+			defer sub.Unsubscribe()
+			subs = append(subs, sub)
+		}
+
+		for _, sub := range subs {
+			_, err = sub.NextMsg(100 * time.Millisecond)
+			if sub.Subject == "foo" {
+				if !errors.Is(err, nats.ErrPermissionViolation) {
+					t.Fatalf("Expected permissions violation error, got %v", err)
+				}
+				// subsequent calls should return the same error
+				_, err = sub.NextMsg(100 * time.Millisecond)
+				if !errors.Is(err, nats.ErrPermissionViolation) {
+					t.Fatalf("Expected permissions violation error, got %v", err)
+				}
+			} else {
+				if !errors.Is(err, nats.ErrTimeout) {
+					t.Fatalf("Expected timeout error, got %v", err)
+				}
+			}
+		}
+	})
+
+	t.Run("PermissionErrOnSubscribe disabled", func(t *testing.T) {
+		nc, err := nats.Connect(s.ClientURL(), nats.UserInfo("test", "test"))
+		if err != nil {
+			t.Fatalf("Error on connect: %v", err)
+		}
+		defer nc.Close()
+
+		// Cause a subscribe error
+		sub, err := nc.SubscribeSync("foo")
+		if err != nil {
+			t.Fatalf("Error on subscribe: %v", err)
+		}
+		defer sub.Unsubscribe()
+
+		_, err = sub.NextMsg(100 * time.Millisecond)
+		if !errors.Is(err, nats.ErrTimeout) {
+			t.Fatalf("Expected timeout error, got %v", err)
+		}
+	})
+}