Repository: nats-io/nats-server
Pull request #5076 (2024-02-13 23:31:45)

Bump Go version to v1.21.7


Patch:
diff --git a/.travis.yml b/.travis.yml
index 9a518edb..b687f3ad 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -8,7 +8,7 @@ language: go
 go:
   # This should be quoted or use .x, but should not be unquoted.
   # Remember that a YAML bare float drops trailing zeroes.
-  - "1.21.6"
+  - "1.21.7"
   - "1.20.13"
 
 go_import_path: github.com/nats-io/nats-server
diff --git a/go.mod b/go.mod
index 996548dc..b9a660c9 100644
--- a/go.mod
+++ b/go.mod
@@ -3,10 +3,10 @@ module github.com/nats-io/nats-server/v2
 go 1.20
 
 require (
-	github.com/klauspost/compress v1.17.6
+	github.com/klauspost/compress v1.17.7
 	github.com/minio/highwayhash v1.0.2
-	github.com/nats-io/jwt/v2 v2.5.3
-	github.com/nats-io/nats.go v1.32.0
+	github.com/nats-io/jwt/v2 v2.5.5
+	github.com/nats-io/nats.go v1.33.1
 	github.com/nats-io/nkeys v0.4.7
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/automaxprocs v1.5.3
diff --git a/go.sum b/go.sum
index 010a129d..cd1fcf47 100644
--- a/go.sum
+++ b/go.sum
@@ -1,14 +1,12 @@
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
-github.com/klauspost/compress v1.17.5 h1:d4vBd+7CHydUqpFBgUEKkSdtSugf9YFmSkvUYPquI5E=
-github.com/klauspost/compress v1.17.5/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
-github.com/klauspost/compress v1.17.6 h1:60eq2E/jlfwQXtvZEeBUYADs+BwKBWURIY+Gj2eRGjI=
-github.com/klauspost/compress v1.17.6/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
+github.com/klauspost/compress v1.17.7 h1:ehO88t2UGzQK66LMdE8tibEd1ErmzZjNEqWkjLAKQQg=
+github.com/klauspost/compress v1.17.7/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
-github.com/nats-io/jwt/v2 v2.5.3 h1:/9SWvzc6hTfamcgXJ3uYRpgj+QuY2aLNqRiqrKcrpEo=
-github.com/nats-io/jwt/v2 v2.5.3/go.mod h1:iysuPemFcc7p4IoYots3IuELSI4EDe9Y0bQMe+I3Bf4=
-github.com/nats-io/nats.go v1.32.0 h1:Bx9BZS+aXYlxW08k8Gd3yR2s73pV5XSoAQUyp1Kwvp0=
-github.com/nats-io/nats.go v1.32.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
+github.com/nats-io/jwt/v2 v2.5.5 h1:ROfXb50elFq5c9+1ztaUbdlrArNFl2+fQWP6B8HGEq4=
+github.com/nats-io/jwt/v2 v2.5.5/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
+github.com/nats-io/nats.go v1.33.1 h1:8TxLZZ/seeEfR97qV0/Bl939tpDnt2Z2fK3HkPypj70=
+github.com/nats-io/nats.go v1.33.1/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
 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/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
@@ -18,13 +16,9 @@ github.com/prashantv/gostub v1.1.0 h1:BTyx3RfQjRHnUWaGF9oQos79AlQ5k8WNktv7VGvVH4
 github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
 go.uber.org/automaxprocs v1.5.3 h1:kWazyxZUrS3Gs4qUpbwo5kEIMGe/DAvi5Z4tl2NW4j8=
 go.uber.org/automaxprocs v1.5.3/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0=
-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/crypto v0.19.0 h1:ENy+Az/9Y1vSrlrvBSyna3PITt4tiZLf7sgCjZBX7Wo=
 golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-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/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y=
 golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
diff --git a/server/accounts.go b/server/accounts.go
index 56795ae3..c8fa73fd 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -54,7 +54,6 @@ type Account struct {
 	Name         string
 	Nkey         string
 	Issuer       string
-	TraceDest    string
 	claimJWT     string
 	updated      time.Time
 	mu           sync.RWMutex
@@ -97,6 +96,12 @@ type Account struct {
 	nameTag      string
 	lastLimErr   int64
 	routePoolIdx int
+	// If the trace destination is specified and a message with a traceParentHdr
+	// is received, and has the least significant bit of the last token set to 1,
+	// then if traceDestSampling is > 0 and < 100, a random value will be selected
+	// and if it falls between 0 and that value, message tracing will be triggered.
+	traceDest         string
+	traceDestSampling int
 }
 
 const (
@@ -157,6 +162,7 @@ type serviceImport struct {
 	share       bool
 	tracking    bool
 	didDeliver  bool
+	atrc        bool        // allow trace (got from service export)
 	trackingHdr http.Header // header from request
 }
 
@@ -256,12 +262,29 @@ func (a *Account) String() string {
 	return a.Name
 }
 
+func (a *Account) setTraceDest(dest string) {
+	a.mu.Lock()
+	a.traceDest = dest
+	a.mu.Unlock()
+}
+
+func (a *Account) getTraceDestAndSampling() (string, int) {
+	a.mu.RLock()
+	dest := a.traceDest
+	sampling := a.traceDestSampling
+	a.mu.RUnlock()
+	return dest, sampling
+}
+
 // Used to create shallow copies of accounts for transfer
 // from opts to real accounts in server struct.
+// Account `na` write lock is expected to be held on entry
+// while account `a` is the one from the Options struct
+// being loaded/reloaded and do not need locking.
 func (a *Account) shallowCopy(na *Account) {
 	na.Nkey = a.Nkey
 	na.Issuer = a.Issuer
-	na.TraceDest = a.TraceDest
+	na.traceDest, na.traceDestSampling = a.traceDest, a.traceDestSampling
 
 	if a.imports.streams != nil {
 		na.imports.streams = make([]*streamImport, 0, len(a.imports.streams))
@@ -905,6 +928,13 @@ func (a *Account) registerLeafNodeCluster(cluster string) {
 	a.leafClusters[cluster]++
 }
 
+// Check to see if we already have this cluster registered.
+func (a *Account) hasLeafNodeCluster(cluster string) bool {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	return a.leafClusters[cluster] > 0
+}
+
 // Check to see if this cluster is isolated, meaning the only one.
 // Read Lock should be held.
 func (a *Account) isLeafNodeClusterIsolated(cluster string) bool {
@@ -1892,11 +1922,13 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 		return nil, ErrMissingAccount
 	}
 
+	var atrc bool
 	dest.mu.RLock()
 	se := dest.getServiceExport(to)
 	if se != nil {
 		rt = se.respType
 		lat = se.latency
+		atrc = se.atrc
 	}
 	dest.mu.RUnlock()
 
@@ -1940,7 +1972,7 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 	if claim != nil {
 		share = claim.Share
 	}
-	si := &serviceImport{dest, claim, se, nil, from, to, tr, 0, rt, lat, nil, nil, usePub, false, false, share, false, false, nil}
+	si := &serviceImport{dest, claim, se, nil, from, to, tr, 0, rt, lat, nil, nil, usePub, false, false, share, false, false, atrc, nil}
 	a.imports.services[from] = si
 	a.mu.Unlock()
 
@@ -2405,7 +2437,7 @@ func (a *Account) addRespServiceImport(dest *Account, to string, osi *serviceImp
 
 	// dest is the requestor's account. a is the service responder with the export.
 	// Marked as internal here, that is how we distinguish.
-	si := &serviceImport{dest, nil, osi.se, nil, nrr, to, nil, 0, rt, nil, nil, nil, false, true, false, osi.share, false, false, nil}
+	si := &serviceImport{dest, nil, osi.se, nil, nrr, to, nil, 0, rt, nil, nil, nil, false, true, false, osi.share, false, false, false, nil}
 
 	if a.exports.responses == nil {
 		a.exports.responses = make(map[string]*serviceImport)
@@ -2515,10 +2547,9 @@ func (a *Account) addMappedStreamImportWithClaim(account *Account, from, to stri
 		a.mu.Unlock()
 		return ErrStreamImportDuplicate
 	}
-	// TODO(ik): When AllowTrace is added to JWT, uncomment those lines:
-	// if imClaim != nil {
-	// 	allowTrace = imClaim.AllowTrace
-	// }
+	if imClaim != nil {
+		allowTrace = imClaim.AllowTrace
+	}
 	a.imports.streams = append(a.imports.streams, &streamImport{account, from, to, tr, nil, imClaim, usePub, false, allowTrace})
 	a.mu.Unlock()
 	return nil
@@ -2853,7 +2884,9 @@ func (a *Account) checkStreamImportsEqual(b *Account) bool {
 		bm[bim.acc.Name+bim.from+bim.to] = bim
 	}
 	for _, aim := range a.imports.streams {
-		if _, ok := bm[aim.acc.Name+aim.from+aim.to]; !ok {
+		if bim, ok := bm[aim.acc.Name+aim.from+aim.to]; !ok {
+			return false
+		} else if aim.atrc != bim.atrc {
 			return false
 		}
 	}
@@ -2939,6 +2972,9 @@ func isServiceExportEqual(a, b *serviceExport) bool {
 			return false
 		}
 	}
+	if a.atrc != b.atrc {
+		return false
+	}
 	return true
 }
 
@@ -3216,6 +3252,19 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 	a.nameTag = ac.Name
 	a.tags = ac.Tags
 
+	var td string
+	var tds int
+	if ac.Trace != nil {
+		// Update trace destination and sampling
+		td, tds = string(ac.Trace.Destination), ac.Trace.Sampling
+		if !IsValidPublishSubject(td) {
+			td, tds = _EMPTY_, 0
+		} else if tds <= 0 || tds > 100 {
+			tds = 100
+		}
+	}
+	a.traceDest, a.traceDestSampling = td, tds
+
 	// Check for external authorization.
 	if ac.HasExternalAuthorization() {
 		a.extAuth = &jwt.ExternalAuthorization{}
@@ -3344,6 +3393,9 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 					s.Debugf("Error adding service export response threshold for [%s]: %v", a.traceLabel(), err)
 				}
 			}
+			if err := a.SetServiceExportAllowTrace(sub, e.AllowTrace); err != nil {
+				s.Debugf("Error adding allow_trace for %q: %v", sub, err)
+			}
 		}
 
 		var revocationChanged *bool
@@ -3484,10 +3536,15 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 				if si != nil && si.acc.Name == a.Name {
 					// Check for if we are still authorized for an import.
 					si.invalid = !a.checkServiceImportAuthorized(acc, si.to, si.claim)
-					if si.latency != nil && !si.response {
-						// Make sure we should still be tracking latency.
+					// Make sure we should still be tracking latency and if we
+					// are allowed to trace.
+					if !si.response {
 						if se := a.getServiceExport(si.to); se != nil {
-							si.latency = se.latency
+							if si.latency != nil {
+								si.latency = se.latency
+							}
+							// Update allow trace.
+							si.atrc = se.atrc
 						}
 					}
 				}
diff --git a/server/client.go b/server/client.go
index 1938096e..a25ac6c2 100644
--- a/server/client.go
+++ b/server/client.go
@@ -143,6 +143,7 @@ const (
 	connectProcessFinished                        // Marks if this connection has finished the connect process.
 	compressionNegotiated                         // Marks if this connection has negotiated compression level with remote.
 	didTLSFirst                                   // Marks if this connection requested and was accepted doing the TLS handshake first (prior to INFO).
+	isSlowConsumer                                // Marks connection as a slow consumer.
 )
 
 // set the flag (would be equivalent to set the boolean to true)
@@ -797,8 +798,14 @@ func (c *client) registerWithAccount(acc *Account) error {
 	// Check if we have a max connections violation
 	if kind == CLIENT && acc.MaxTotalConnectionsReached() {
 		return ErrTooManyAccountConnections
-	} else if kind == LEAF && acc.MaxTotalLeafNodesReached() {
-		return ErrTooManyAccountConnections
+	} else if kind == LEAF {
+		// Check if we are already connected to this cluster.
+		if rc := c.remoteCluster(); rc != _EMPTY_ && acc.hasLeafNodeCluster(rc) {
+			return ErrLeafNodeLoop
+		}
+		if acc.MaxTotalLeafNodesReached() {
+			return ErrTooManyAccountConnections
+		}
 	}
 
 	// Add in new one.
@@ -1651,9 +1658,11 @@ func (c *client) flushOutbound() bool {
 	}
 
 	// Ignore ErrShortWrite errors, they will be handled as partials.
+	var gotWriteTimeout bool
 	if err != nil && err != io.ErrShortWrite {
 		// Handle timeout error (slow consumer) differently
 		if ne, ok := err.(net.Error); ok && ne.Timeout() {
+			gotWriteTimeout = true
 			if closed := c.handleWriteTimeout(n, attempted, len(orig)); closed {
 				return true
 			}
@@ -1691,6 +1700,11 @@ func (c *client) flushOutbound() bool {
 		close(c.out.stc)
 		c.out.stc = nil
 	}
+	// Check if the connection is recovering from being a slow consumer.
+	if !gotWriteTimeout && c.flags.isSet(isSlowConsumer) {
+		c.Noticef("Slow Consumer Recovered: Flush took %.3fs with %d chunks of %d total bytes.", time.Since(start).Seconds(), len(orig), attempted)
+		c.flags.clear(isSlowConsumer)
+	}
 
 	return true
 }
@@ -1716,6 +1730,11 @@ func (c *client) handleWriteTimeout(written, attempted int64, numChunks int) boo
 		c.markConnAsClosed(SlowConsumerWriteDeadline)
 		return true
 	}
+	alreadySC := c.flags.isSet(isSlowConsumer)
+	scState := "Detected"
+	if alreadySC {
+		scState = "State"
+	}
 
 	// Aggregate slow consumers.
 	atomic.AddInt64(&c.srv.slowConsumers, 1)
@@ -1723,7 +1742,10 @@ func (c *client) handleWriteTimeout(written, attempted int64, numChunks int) boo
 	case CLIENT:
 		c.srv.scStats.clients.Add(1)
 	case ROUTER:
-		c.srv.scStats.routes.Add(1)
+		// Only count each Slow Consumer event once.
+		if !alreadySC {
+			c.srv.scStats.routes.Add(1)
+		}
 	case GATEWAY:
 		c.srv.scStats.gateways.Add(1)
 	case LEAF:
@@ -1732,13 +1754,15 @@ func (c *client) handleWriteTimeout(written, attempted int64, numChunks int) boo
 	if c.acc != nil {
 		atomic.AddInt64(&c.acc.slowConsumers, 1)
 	}
-	c.Noticef("Slow Consumer Detected: WriteDeadline of %v exceeded with %d chunks of %d total bytes.",
-		c.out.wdl, numChunks, attempted)
+	c.Noticef("Slow Consumer %s: WriteDeadline of %v exceeded with %d chunks of %d total bytes.",
+		scState, c.out.wdl, numChunks, attempted)
 
 	// We always close CLIENT connections, or when nothing was written at all...
 	if c.kind == CLIENT || written == 0 {
 		c.markConnAsClosed(SlowConsumerWriteDeadline)
 		return true
+	} else {
+		c.flags.setIfNotSet(isSlowConsumer)
 	}
 	return false
 }
@@ -2007,10 +2031,26 @@ func (c *client) processConnect(arg []byte) error {
 		}
 	}
 
-	// If websocket client and JWT not in the CONNECT, use the cookie JWT (possibly empty).
-	if ws := c.ws; ws != nil && c.opts.JWT == "" {
-		c.opts.JWT = ws.cookieJwt
+	// if websocket client, maybe some options through cookies
+	if ws := c.ws; ws != nil {
+		// if JWT not in the CONNECT, use the cookie JWT (possibly empty).
+		if c.opts.JWT == _EMPTY_ {
+			c.opts.JWT = ws.cookieJwt
+		}
+		// if user not in the CONNECT, use the cookie user (possibly empty)
+		if c.opts.Username == _EMPTY_ {
+			c.opts.Username = ws.cookieUsername
+		}
+		// if pass not in the CONNECT, use the cookie password (possibly empty).
+		if c.opts.Password == _EMPTY_ {
+			c.opts.Password = ws.cookiePassword
+		}
+		// if token not in the CONNECT, use the cookie token (possibly empty).
+		if c.opts.Token == _EMPTY_ {
+			c.opts.Token = ws.cookieToken
+		}
 	}
+
 	// when not in operator mode, discard the jwt
 	if srv != nil && srv.trustedKeys == nil {
 		c.opts.JWT = _EMPTY_
@@ -4198,7 +4238,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 		}
 	}
 	siAcc := si.acc
-	allowTrace := si.se != nil && si.se.atrc
+	allowTrace := si.atrc
 	acc.mu.RUnlock()
 
 	// We have a special case where JetStream pulls in all service imports through one export.
@@ -4356,8 +4396,8 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 				// We also need to disable the message trace headers so that
 				// if the message is routed, it does not initialize tracing in the
 				// remote.
-				positions := mt.disableTraceHeaders(c, msg)
-				defer mt.enableTraceHeaders(c, msg, positions)
+				positions := disableTraceHeaders(c, msg)
+				defer enableTraceHeaders(c, msg, positions)
 			}
 		}
 	}
diff --git a/server/consumer.go b/server/consumer.go
index a9ad0ebc..563ae5e9 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -55,6 +55,8 @@ type ConsumerInfo struct {
 	NumPending     uint64          `json:"num_pending"`
 	Cluster        *ClusterInfo    `json:"cluster,omitempty"`
 	PushBound      bool            `json:"push_bound,omitempty"`
+	Paused         bool            `json:"paused,omitempty"`
+	PauseRemaining time.Duration   `json:"pause_remaining,omitempty"`
 	// TimeStamp indicates when the info was gathered
 	TimeStamp time.Time `json:"ts"`
 }
@@ -104,6 +106,9 @@ type ConsumerConfig struct {
 
 	// Metadata is additional metadata for the Consumer.
 	Metadata map[string]string `json:"metadata,omitempty"`
+
+	// PauseUntil is for suspending the consumer until the deadline.
+	PauseUntil *time.Time `json:"pause_until,omitempty"`
 }
 
 // SequenceInfo has both the consumer and the stream sequence and last activity.
@@ -352,11 +357,12 @@ type consumer struct {
 	active            bool
 	replay            bool
 	dtmr              *time.Timer
+	uptmr             *time.Timer // Unpause timer
 	gwdtmr            *time.Timer
 	dthresh           time.Duration
-	mch               chan struct{}
-	qch               chan struct{}
-	inch              chan bool
+	mch               chan struct{} // Message channel
+	qch               chan struct{} // Quit channel
+	inch              chan bool     // Interest change channel
 	sfreq             int32
 	ackEventT         string
 	nakEventT         string
@@ -1072,6 +1078,34 @@ func (o *consumer) updateInactiveThreshold(cfg *ConsumerConfig) {
 	}
 }
 
+// Updates the paused state. If we are the leader and the pause deadline
+// hasn't passed yet then we will start a timer to kick the consumer once
+// that deadline is reached. Lock should be held.
+func (o *consumer) updatePauseState(cfg *ConsumerConfig) {
+	if o.uptmr != nil {
+		stopAndClearTimer(&o.uptmr)
+	}
+	if !o.isLeader() {
+		// Only the leader will run the timer as only the leader will run
+		// loopAndGatherMsgs.
+		return
+	}
+	if cfg.PauseUntil == nil || cfg.PauseUntil.IsZero() || cfg.PauseUntil.Before(time.Now()) {
+		// Either the PauseUntil is unset (is effectively zero) or the
+		// deadline has already passed, in which case there is nothing
+		// to do.
+		return
+	}
+	o.uptmr = time.AfterFunc(time.Until(*cfg.PauseUntil), func() {
+		o.mu.Lock()
+		defer o.mu.Unlock()
+
+		stopAndClearTimer(&o.uptmr)
+		o.sendPauseAdvisoryLocked(&o.cfg)
+		o.signalNewMessages()
+	})
+}
+
 func (o *consumer) consumerAssignment() *consumerAssignment {
 	o.mu.RLock()
 	defer o.mu.RUnlock()
@@ -1265,6 +1299,9 @@ func (o *consumer) setLeader(isLeader bool) {
 			o.dtmr = time.AfterFunc(o.dthresh, o.deleteNotActive)
 		}
 
+		// Update the consumer pause tracking.
+		o.updatePauseState(&o.cfg)
+
 		// If we are not in ReplayInstant mode mark us as in replay state until resolved.
 		if o.cfg.ReplayPolicy != ReplayInstant {
 			o.replay = true
@@ -1332,7 +1369,8 @@ func (o *consumer) setLeader(isLeader bool) {
 		}
 		// Stop any inactivity timers. Should only be running on leaders.
 		stopAndClearTimer(&o.dtmr)
-
+		// Stop any unpause timers. Should only be running on leaders.
+		stopAndClearTimer(&o.uptmr)
 		// Make sure to clear out any re-deliver queues
 		stopAndClearTimer(&o.ptmr)
 		o.rdq = nil
@@ -1449,6 +1487,32 @@ func (o *consumer) sendCreateAdvisory() {
 	o.sendAdvisory(subj, j)
 }
 
+func (o *consumer) sendPauseAdvisoryLocked(cfg *ConsumerConfig) {
+	e := JSConsumerPauseAdvisory{
+		TypedEvent: TypedEvent{
+			Type: JSConsumerPauseAdvisoryType,
+			ID:   nuid.Next(),
+			Time: time.Now().UTC(),
+		},
+		Stream:   o.stream,
+		Consumer: o.name,
+		Domain:   o.srv.getOpts().JetStreamDomain,
+	}
+
+	if cfg.PauseUntil != nil {
+		e.PauseUntil = *cfg.PauseUntil
+		e.Paused = time.Now().Before(e.PauseUntil)
+	}
+
+	j, err := json.Marshal(e)
+	if err != nil {
+		return
+	}
+
+	subj := JSAdvisoryConsumerPausePre + "." + o.stream + "." + o.name
+	o.sendAdvisory(subj, j)
+}
+
 // Created returns created time.
 func (o *consumer) createdTime() time.Time {
 	o.mu.Lock()
@@ -1812,6 +1876,12 @@ func (o *consumer) updateConfig(cfg *ConsumerConfig) error {
 		return err
 	}
 
+	// Make sure we always store PauseUntil in UTC.
+	if cfg.PauseUntil != nil {
+		utc := (*cfg.PauseUntil).UTC()
+		cfg.PauseUntil = &utc
+	}
+
 	if o.store != nil {
 		// Update local state always.
 		if err := o.store.UpdateConfig(cfg); err != nil {
@@ -1860,6 +1930,22 @@ func (o *consumer) updateConfig(cfg *ConsumerConfig) error {
 			o.dtmr = time.AfterFunc(o.dthresh, o.deleteNotActive)
 		}
 	}
+	// Check whether the pause has changed
+	{
+		var old, new time.Time
+		if o.cfg.PauseUntil != nil {
+			old = *o.cfg.PauseUntil
+		}
+		if cfg.PauseUntil != nil {
+			new = *cfg.PauseUntil
+		}
+		if !old.Equal(new) {
+			o.updatePauseState(cfg)
+			if o.isLeader() {
+				o.sendPauseAdvisoryLocked(cfg)
+			}
+		}
+	}
 
 	// Check for Subject Filters update.
 	newSubjects := gatherSubjectFilters(cfg.FilterSubject, cfg.FilterSubjects)
@@ -2574,6 +2660,12 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
 		PushBound:      o.isPushMode() && o.active,
 		TimeStamp:      time.Now().UTC(),
 	}
+	if o.cfg.PauseUntil != nil {
+		p := *o.cfg.PauseUntil
+		if info.Paused = time.Now().Before(p); info.Paused {
+			info.PauseRemaining = time.Until(p)
+		}
+	}
 
 	// If we are replicated and we are not the leader we need to pull certain data from our store.
 	if rg != nil && rg.node != nil && !o.isLeader() && o.store != nil {
@@ -3100,7 +3192,7 @@ func (o *consumer) nextWaiting(sz int) *waitingRequest {
 				// Since we can't send that message to the requestor, we need to
 				// notify that we are closing the request.
 				const maxBytesT = "NATS/1.0 409 Message Size Exceeds MaxBytes\r\n%s: %d\r\n%s: %d\r\n\r\n"
-				hdr := []byte(fmt.Sprintf(maxBytesT, JSPullRequestPendingMsgs, wr.n, JSPullRequestPendingBytes, wr.b))
+				hdr := fmt.Appendf(nil, maxBytesT, JSPullRequestPendingMsgs, wr.n, JSPullRequestPendingBytes, wr.b)
 				o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 				// Remove the current one, no longer valid due to max bytes limit.
 				o.waiting.removeCurrent()
@@ -3123,7 +3215,7 @@ func (o *consumer) nextWaiting(sz int) *waitingRequest {
 			}
 		} else {
 			// We do check for expiration in `processWaiting`, but it is possible to hit the expiry here, and not there.
-			hdr := []byte(fmt.Sprintf("NATS/1.0 408 Request Timeout\r\n%s: %d\r\n%s: %d\r\n\r\n", JSPullRequestPendingMsgs, wr.n, JSPullRequestPendingBytes, wr.b))
+			hdr := fmt.Appendf(nil, "NATS/1.0 408 Request Timeout\r\n%s: %d\r\n%s: %d\r\n\r\n", JSPullRequestPendingMsgs, wr.n, JSPullRequestPendingBytes, wr.b)
 			o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 			o.waiting.removeCurrent()
 			if o.node != nil {
@@ -3135,7 +3227,7 @@ func (o *consumer) nextWaiting(sz int) *waitingRequest {
 		}
 		if wr.interest != wr.reply {
 			const intExpT = "NATS/1.0 408 Interest Expired\r\n%s: %d\r\n%s: %d\r\n\r\n"
-			hdr := []byte(fmt.Sprintf(intExpT, JSPullRequestPendingMsgs, wr.n, JSPullRequestPendingBytes, wr.b))
+			hdr := fmt.Appendf(nil, intExpT, JSPullRequestPendingMsgs, wr.n, JSPullRequestPendingBytes, wr.b)
 			o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 		}
 		// Remove the current one, no longer valid.
@@ -3208,7 +3300,7 @@ func (o *consumer) processNextMsgRequest(reply string, msg []byte) {
 	}
 
 	sendErr := func(status int, description string) {
-		hdr := []byte(fmt.Sprintf("NATS/1.0 %d %s\r\n\r\n", status, description))
+		hdr := fmt.Appendf(nil, "NATS/1.0 %d %s\r\n\r\n", status, description)
 		o.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 	}
 
@@ -3600,7 +3692,7 @@ func (o *consumer) processWaiting(eos bool) (int, int, int, time.Time) {
 		wr := wq.reqs[rp]
 		// Check expiration.
 		if (eos && wr.noWait && wr.d > 0) || (!wr.expires.IsZero() && now.After(wr.expires)) {
-			hdr := []byte(fmt.Sprintf("NATS/1.0 408 Request Timeout\r\n%s: %d\r\n%s: %d\r\n\r\n", JSPullRequestPendingMsgs, wr.n, JSPullRequestPendingBytes, wr.b))
+			hdr := fmt.Appendf(nil, "NATS/1.0 408 Request Timeout\r\n%s: %d\r\n%s: %d\r\n\r\n", JSPullRequestPendingMsgs, wr.n, JSPullRequestPendingBytes, wr.b)
 			o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 			remove(wr, rp)
 			i++
@@ -3841,6 +3933,8 @@ func (o *consumer) suppressDeletion() {
 	}
 }
 
+// loopAndGatherMsgs waits for messages for the consumer. qch is the quit channel,
+// upch is the unpause channel which fires when the PauseUntil deadline is reached.
 func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 	// On startup check to see if we are in a reply situation where replay policy is not instant.
 	var (
@@ -3907,6 +4001,13 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		// Clear last error.
 		err = nil
 
+		// If the consumer is paused then stop sending.
+		if o.cfg.PauseUntil != nil && !o.cfg.PauseUntil.IsZero() && time.Now().Before(*o.cfg.PauseUntil) {
+			// If the consumer is paused and we haven't reached the deadline yet then
+			// go back to waiting.
+			goto waitForMsgs
+		}
+
 		// If we are in push mode and not active or under flowcontrol let's stop sending.
 		if o.isPushMode() {
 			if !o.active || (o.maxpb > 0 && o.pbytes > o.maxpb) {
@@ -4033,7 +4134,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 
 		// If given request fulfilled batch size, but there are still pending bytes, send information about it.
 		if wrn <= 0 && wrb > 0 {
-			o.outq.send(newJSPubMsg(dsubj, _EMPTY_, _EMPTY_, []byte(fmt.Sprintf(JsPullRequestRemainingBytesT, JSPullRequestPendingMsgs, wrn, JSPullRequestPendingBytes, wrb)), nil, nil, 0))
+			o.outq.send(newJSPubMsg(dsubj, _EMPTY_, _EMPTY_, fmt.Appendf(nil, JsPullRequestRemainingBytesT, JSPullRequestPendingMsgs, wrn, JSPullRequestPendingBytes, wrb), nil, nil, 0))
 		}
 		// Reset our idle heartbeat timer if set.
 		if hb != nil {
@@ -4096,10 +4197,10 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 func (o *consumer) sendIdleHeartbeat(subj string) {
 	const t = "NATS/1.0 100 Idle Heartbeat\r\n%s: %d\r\n%s: %d\r\n\r\n"
 	sseq, dseq := o.sseq-1, o.dseq-1
-	hdr := []byte(fmt.Sprintf(t, JSLastConsumerSeq, dseq, JSLastStreamSeq, sseq))
+	hdr := fmt.Appendf(nil, t, JSLastConsumerSeq, dseq, JSLastStreamSeq, sseq)
 	if fcp := o.fcid; fcp != _EMPTY_ {
 		// Add in that we are stalled on flow control here.
-		addOn := []byte(fmt.Sprintf("%s: %s\r\n\r\n", JSConsumerStalled, fcp))
+		addOn := fmt.Appendf(nil, "%s: %s\r\n\r\n", JSConsumerStalled, fcp)
 		hdr = append(hdr[:len(hdr)-LEN_CR_LF], []byte(addOn)...)
 	}
 	o.outq.send(newJSPubMsg(subj, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
@@ -4783,6 +4884,23 @@ func (o *consumer) selectStartingSeqNo() {
 				// If we are here we are time based.
 				// TODO(dlc) - Once clustered can't rely on this.
 				o.sseq = o.mset.store.GetSeqFromTime(*o.cfg.OptStartTime)
+				// Here we want to see if we are filtered, and if so possibly close the gap
+				// to the nearest first given our starting sequence from time. This is so we do
+				// not force the system to do a linear walk between o.sseq and the real first.
+				if len(o.subjf) > 0 {
+					nseq := state.LastSeq
+					for _, filter := range o.subjf {
+						// Use first sequence since this is more optimized atm.
+						ss := o.mset.store.FilteredState(state.FirstSeq, filter.subject)
+						if ss.First > o.sseq && ss.First < nseq {
+							nseq = ss.First
+						}
+					}
+					// Skip ahead if possible.
+					if nseq > o.sseq && nseq < state.LastSeq {
+						o.sseq = nseq
+					}
+				}
 			} else {
 				// DeliverNew
 				o.sseq = state.LastSeq + 1
@@ -4901,17 +5019,26 @@ func (o *consumer) hasNoLocalInterest() bool {
 
 // This is when the underlying stream has been purged.
 // sseq is the new first seq for the stream after purge.
-// Lock should be held.
-func (o *consumer) purge(sseq uint64, slseq uint64) {
+// Lock should NOT be held.
+func (o *consumer) purge(sseq uint64, slseq uint64, isWider bool) {
 	// Do not update our state unless we know we are the leader.
 	if !o.isLeader() {
 		return
 	}
 	// Signals all have been purged for this consumer.
-	if sseq == 0 {
+	if sseq == 0 && !isWider {
 		sseq = slseq + 1
 	}
 
+	var store StreamStore
+	if isWider {
+		o.mu.RLock()
+		if o.mset != nil {
+			store = o.mset.store
+		}
+		o.mu.RUnlock()
+	}
+
 	o.mu.Lock()
 	// Do not go backwards
 	if o.sseq < sseq {
@@ -4920,7 +5047,6 @@ func (o *consumer) purge(sseq uint64, slseq uint64) {
 
 	if o.asflr < sseq {
 		o.asflr = sseq - 1
-
 		// We need to remove those no longer relevant from pending.
 		for seq, p := range o.pending {
 			if seq <= o.asflr {
@@ -4934,8 +5060,24 @@ func (o *consumer) purge(sseq uint64, slseq uint64) {
 				delete(o.rdc, seq)
 				// rdq handled below.
 			}
+			if isWider && store != nil {
+				// Our filtered subject, which could be all, is wider than the underlying purge.
+				// We need to check if the pending items left are still valid.
+				var smv StoreMsg
+				if _, err := store.LoadMsg(seq, &smv); err == errDeletedMsg || err == ErrStoreMsgNotFound {
+					if p.Sequence > o.adflr {
+						o.adflr = p.Sequence
+						if o.adflr > o.dseq {
+							o.dseq = o.adflr
+						}
+					}
+					delete(o.pending, seq)
+					delete(o.rdc, seq)
+				}
+			}
 		}
 	}
+
 	// This means we can reset everything at this point.
 	if len(o.pending) == 0 {
 		o.pending, o.rdc = nil, nil
@@ -5179,6 +5321,7 @@ func (o *consumer) switchToEphemeral() {
 	rr := o.acc.sl.Match(o.cfg.DeliverSubject)
 	// Setup dthresh.
 	o.updateInactiveThreshold(&o.cfg)
+	o.updatePauseState(&o.cfg)
 	o.mu.Unlock()
 
 	// Update interest
diff --git a/server/errors.go b/server/errors.go
index 0c744bca..c73c5a54 100644
--- a/server/errors.go
+++ b/server/errors.go
@@ -60,6 +60,9 @@ var (
 	// connections.
 	ErrTooManyAccountConnections = errors.New("maximum account active connections exceeded")
 
+	// ErrLeafNodeLoop signals a leafnode is trying to register for a cluster we already have registered.
+	ErrLeafNodeLoop = errors.New("leafnode loop detected")
+
 	// ErrTooManySubs signals a client that the maximum number of subscriptions per connection
 	// has been reached.
 	ErrTooManySubs = errors.New("maximum subscriptions exceeded")
diff --git a/server/filestore.go b/server/filestore.go
index 24e2242a..7f43d8fd 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1466,6 +1466,16 @@ func (fs *fileStore) warn(format string, args ...any) {
 	fs.srv.Warnf(fmt.Sprintf("Filestore [%s] %s", fs.cfg.Name, format), args...)
 }
 
+// For doing debug logging.
+// Lock should be held.
+func (fs *fileStore) debug(format string, args ...any) {
+	// No-op if no server configured.
+	if fs.srv == nil {
+		return
+	}
+	fs.srv.Debugf(fmt.Sprintf("Filestore [%s] %s", fs.cfg.Name, format), args...)
+}
+
 // Track local state but ignore timestamps here.
 func updateTrackingState(state *StreamState, mb *msgBlock) {
 	if state.FirstSeq == 0 {
@@ -2225,30 +2235,44 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 
 	fseq, isAll, subs := start, filter == _EMPTY_ || filter == fwcs, []string{filter}
 
-	if err := mb.ensurePerSubjectInfoLoaded(); err != nil {
-		return nil, false, err
+	var didLoad bool
+	if mb.fssNotLoaded() {
+		// Make sure we have fss loaded.
+		mb.loadMsgsWithLock()
+		didLoad = true
 	}
 
 	// If we only have 1 subject currently and it matches our filter we can also set isAll.
 	if !isAll && len(mb.fss) == 1 {
 		_, isAll = mb.fss[filter]
 	}
-	// Skip scan of mb.fss if number of messages in the block are less than
-	// 1/2 the number of subjects in mb.fss. Or we have a wc and lots of fss entries.
-	const linearScanMaxFSS = 32
 	// Make sure to start at mb.first.seq if fseq < mb.first.seq
 	if seq := atomic.LoadUint64(&mb.first.seq); seq > fseq {
 		fseq = seq
 	}
 	lseq := atomic.LoadUint64(&mb.last.seq)
-	doLinearScan := isAll || 2*int(lseq-fseq) < len(mb.fss) || (wc && len(mb.fss) > linearScanMaxFSS)
 
+	// Optionally build the isMatch for wildcard filters.
+	tsa := [32]string{}
+	fsa := [32]string{}
+	var fts []string
+	var isMatch func(subj string) bool
+	// Decide to build.
+	if wc {
+		fts = tokenizeSubjectIntoSlice(fsa[:0], filter)
+		isMatch = func(subj string) bool {
+			tts := tokenizeSubjectIntoSlice(tsa[:0], subj)
+			return isSubsetMatchTokenized(tts, fts)
+		}
+	}
+	// Only do linear scan if isAll or we are wildcarded and have to traverse more fss than actual messages.
+	doLinearScan := isAll || (wc && len(mb.fss) > int(lseq-fseq))
 	if !doLinearScan {
 		// If we have a wildcard match against all tracked subjects we know about.
 		if wc {
 			subs = subs[:0]
 			for subj := range mb.fss {
-				if subjectIsSubsetMatch(subj, filter) {
+				if isMatch(subj) {
 					subs = append(subs, subj)
 				}
 			}
@@ -2270,11 +2294,17 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 		}
 	}
 
+	// If we guess to not do a linear scan, but the above resulted in alot of subs that will
+	// need to be checked for every scanned message, revert.
+	// TODO(dlc) - we could memoize the subs across calls.
+	if len(subs) > int(lseq-fseq) {
+		doLinearScan = true
+	}
+
 	if fseq > lseq {
-		return nil, false, ErrStoreMsgNotFound
+		return nil, didLoad, ErrStoreMsgNotFound
 	}
 
-	var didLoad bool
 	// Need messages loaded from here on out.
 	if mb.cacheNotLoaded() {
 		if err := mb.loadMsgsWithLock(); err != nil {
@@ -2298,7 +2328,7 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 			return fsm, expireOk, nil
 		}
 		if doLinearScan {
-			if wc && subjectIsSubsetMatch(fsm.subj, filter) {
+			if wc && isMatch(sm.subj) {
 				return fsm, expireOk, nil
 			} else if !wc && fsm.subj == filter {
 				return fsm, expireOk, nil
@@ -2824,7 +2854,7 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 			} else {
 				// We need to adjust for all matches in this block.
 				// Make sure we have fss loaded. This loads whole block now.
-				if mb.cacheNotLoaded() {
+				if mb.fssNotLoaded() {
 					mb.loadMsgsWithLock()
 					shouldExpire = true
 				}
@@ -2877,7 +2907,10 @@ func (fs *fileStore) SubjectsTotals(filter string) map[string]uint64 {
 	if fs.psim.Size() == 0 {
 		return nil
 	}
-
+	// Match all if no filter given.
+	if filter == _EMPTY_ {
+		filter = fwcs
+	}
 	fst := make(map[string]uint64)
 	fs.psim.Match(stringToBytes(filter), func(subj []byte, psi *psi) {
 		fst[string(subj)] = psi.total
@@ -7537,7 +7570,7 @@ func (fs *fileStore) writeFullState() error {
 	}
 
 	if cap(buf) > sz {
-		fs.warn("WriteFullState reallocated from %d to %d", sz, cap(buf))
+		fs.debug("WriteFullState reallocated from %d to %d", sz, cap(buf))
 	}
 
 	// Write to a tmp file and rename.
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 03c55ea5..8dcf6bad 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -165,6 +165,11 @@ const (
 	JSApiConsumerDelete  = "$JS.API.CONSUMER.DELETE.*.*"
 	JSApiConsumerDeleteT = "$JS.API.CONSUMER.DELETE.%s.%s"
 
+	// JSApiConsumerPause is the endpoint to pause or unpause consumers.
+	// Will return JSON response.
+	JSApiConsumerPause  = "$JS.API.CONSUMER.PAUSE.*.*"
+	JSApiConsumerPauseT = "$JS.API.CONSUMER.PAUSE.%s.%s"
+
 	// JSApiRequestNextT is the prefix for the request next message(s) for a consumer in worker/pull mode.
 	JSApiRequestNextT = "$JS.API.CONSUMER.MSG.NEXT.%s.%s"
 
@@ -265,6 +270,9 @@ const (
 	// JSAdvisoryConsumerDeletedPre notification that a template deleted.
 	JSAdvisoryConsumerDeletedPre = "$JS.EVENT.ADVISORY.CONSUMER.DELETED"
 
+	// JSAdvisoryConsumerPausePre notification that a consumer paused/unpaused.
+	JSAdvisoryConsumerPausePre = "$JS.EVENT.ADVISORY.CONSUMER.PAUSE"
+
 	// JSAdvisoryStreamSnapshotCreatePre notification that a snapshot was created.
 	JSAdvisoryStreamSnapshotCreatePre = "$JS.EVENT.ADVISORY.STREAM.SNAPSHOT_CREATE"
 
@@ -667,6 +675,19 @@ type JSApiConsumerDeleteResponse struct {
 
 const JSApiConsumerDeleteResponseType = "io.nats.jetstream.api.v1.consumer_delete_response"
 
+type JSApiConsumerPauseRequest struct {
+	PauseUntil time.Time `json:"pause_until,omitempty"`
+}
+
+const JSApiConsumerPauseResponseType = "io.nats.jetstream.api.v1.consumer_pause_response"
+
+type JSApiConsumerPauseResponse struct {
+	ApiResponse
+	Paused         bool          `json:"paused"`
+	PauseUntil     time.Time     `json:"pause_until"`
+	PauseRemaining time.Duration `json:"pause_remaining,omitempty"`
+}
+
 type JSApiConsumerInfoResponse struct {
 	ApiResponse
 	*ConsumerInfo
@@ -928,6 +949,7 @@ func (s *Server) setJetStreamExportSubs() error {
 		{JSApiConsumerList, s.jsConsumerListRequest},
 		{JSApiConsumerInfo, s.jsConsumerInfoRequest},
 		{JSApiConsumerDelete, s.jsConsumerDeleteRequest},
+		{JSApiConsumerPause, s.jsConsumerPauseRequest},
 	}
 
 	js.mu.Lock()
@@ -3973,6 +3995,12 @@ func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, a *Accoun
 		return
 	}
 
+	// If the consumer already exists then don't allow updating the PauseUntil, just set
+	// it back to whatever the current configured value is.
+	if o := stream.lookupConsumer(consumerName); o != nil {
+		req.Config.PauseUntil = o.cfg.PauseUntil
+	}
+
 	o, err := stream.addConsumerWithAction(&req.Config, req.Action)
 
 	if err != nil {
@@ -3987,6 +4015,10 @@ func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, a *Accoun
 	}
 	resp.ConsumerInfo = o.initialInfo()
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
+
+	if o.cfg.PauseUntil != nil && !o.cfg.PauseUntil.IsZero() && time.Now().Before(*o.cfg.PauseUntil) {
+		o.sendPauseAdvisoryLocked(&o.cfg)
+	}
 }
 
 // Request for the list of all consumer names.
@@ -4455,6 +4487,123 @@ func (s *Server) jsConsumerDeleteRequest(sub *subscription, c *client, _ *Accoun
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
 }
 
+// Request to pause or unpause a Consumer.
+func (s *Server) jsConsumerPauseRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
+	if c == nil || !s.JetStreamEnabled() {
+		return
+	}
+	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	if err != nil {
+		s.Warnf(badAPIRequestT, msg)
+		return
+	}
+
+	var req JSApiConsumerPauseRequest
+	var resp = JSApiConsumerPauseResponse{ApiResponse: ApiResponse{Type: JSApiConsumerPauseResponseType}}
+
+	if !isEmptyRequest(msg) {
+		if err := json.Unmarshal(msg, &req); err != nil {
+			resp.Error = NewJSInvalidJSONError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+	}
+
+	// Determine if we should proceed here when we are in clustered mode.
+	isClustered := s.JetStreamIsClustered()
+	js, cc := s.getJetStreamCluster()
+	if isClustered {
+		if js == nil || cc == nil {
+			return
+		}
+		if js.isLeaderless() {
+			resp.Error = NewJSClusterNotAvailError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		// Make sure we are meta leader.
+		if !s.JetStreamIsLeader() {
+			return
+		}
+	}
+
+	if hasJS, doErr := acc.checkJetStream(); !hasJS {
+		if doErr {
+			resp.Error = NewJSNotEnabledForAccountError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		}
+		return
+	}
+
+	stream := streamNameFromSubject(subject)
+	consumer := consumerNameFromSubject(subject)
+
+	if isClustered {
+		sa := js.streamAssignment(acc.Name, stream)
+		if sa == nil {
+			resp.Error = NewJSStreamNotFoundError(Unless(err))
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+
+		ca, ok := sa.consumers[consumer]
+		if !ok || ca == nil {
+			resp.Error = NewJSConsumerNotFoundError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+
+		nca := *ca
+		pauseUTC := req.PauseUntil.UTC()
+		if !pauseUTC.IsZero() {
+			nca.Config.PauseUntil = &pauseUTC
+		}
+		eca := encodeAddConsumerAssignment(&nca)
+		cc.meta.Propose(eca)
+
+		resp.PauseUntil = pauseUTC
+		if resp.Paused = time.Now().Before(pauseUTC); resp.Paused {
+			resp.PauseRemaining = time.Until(pauseUTC)
+		}
+		s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
+		return
+	}
+
+	mset, err := acc.lookupStream(stream)
+	if err != nil {
+		resp.Error = NewJSStreamNotFoundError(Unless(err))
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	obs := mset.lookupConsumer(consumer)
+	if obs == nil {
+		resp.Error = NewJSConsumerNotFoundError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	ncfg := obs.cfg
+	pauseUTC := req.PauseUntil.UTC()
+	if !pauseUTC.IsZero() {
+		ncfg.PauseUntil = &pauseUTC
+	}
+
+	if err := obs.updateConfig(&ncfg); err != nil {
+		// The only type of error that should be returned here is from o.store,
+		// so use a store failed error type.
+		resp.Error = NewJSConsumerStoreFailedError(err)
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	resp.PauseUntil = pauseUTC
+	if resp.Paused = time.Now().Before(pauseUTC); resp.Paused {
+		resp.PauseRemaining = time.Until(pauseUTC)
+	}
+	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
+}
+
 // sendJetStreamAPIAuditAdvisor will send the audit event for a given event.
 func (s *Server) sendJetStreamAPIAuditAdvisory(ci *ClientInfo, acc *Account, subject, request, response string) {
 	s.publishAdvisory(acc, JSAuditAdvisory, JSAPIAudit{
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 4040023d..c775dbc6 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -5081,6 +5081,13 @@ func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) err
 		}
 	}
 
+	// Only send a pause advisory on consumer create if we're
+	// actually paused. The timer would have been kicked by now
+	// by the call to o.setLeader() above.
+	if isLeader && o.cfg.PauseUntil != nil && !o.cfg.PauseUntil.IsZero() && time.Now().Before(*o.cfg.PauseUntil) {
+		o.sendPauseAdvisoryLocked(&o.cfg)
+	}
+
 	return nil
 }
 
@@ -5550,6 +5557,7 @@ func (e *selectPeerError) accumulate(eAdd *selectPeerError) {
 
 // selectPeerGroup will select a group of peers to start a raft group.
 // when peers exist already the unique tag prefix check for the replaceFirstExisting will be skipped
+// js lock should be held.
 func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamConfig, existing []string, replaceFirstExisting int, ignore []string) ([]string, *selectPeerError) {
 	if cluster == _EMPTY_ || cfg == nil {
 		return nil, &selectPeerError{misc: true}
@@ -5571,6 +5579,7 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 		id    string
 		avail uint64
 		ha    int
+		ns    int
 	}
 
 	var nodes []wn
@@ -5635,6 +5644,22 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 		}
 	}
 
+	// Grab the number of streams and HA assets currently assigned to each peer.
+	// HAAssets under usage is async, so calculate here in realtime based on assignments.
+	peerStreams := make(map[string]int, len(peers))
+	peerHA := make(map[string]int, len(peers))
+	for _, asa := range cc.streams {
+		for _, sa := range asa {
+			isHA := len(sa.Group.Peers) > 1
+			for _, peer := range sa.Group.Peers {
+				peerStreams[peer]++
+				if isHA {
+					peerHA[peer]++
+				}
+			}
+		}
+	}
+
 	maxHaAssets := s.getOpts().JetStreamLimits.MaxHAAssets
 
 	// An error is a result of multiple individual placement decisions.
@@ -5697,7 +5722,6 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 		}
 
 		var available uint64
-		var ha int
 		if ni.stats != nil {
 			switch cfg.Storage {
 			case MemoryStorage:
@@ -5717,7 +5741,6 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 					available = uint64(ni.cfg.MaxStore) - used
 				}
 			}
-			ha = ni.stats.HAAssets
 		}
 
 		// Otherwise check if we have enough room if maxBytes set.
@@ -5749,7 +5772,7 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 			}
 		}
 		// Add to our list of potential nodes.
-		nodes = append(nodes, wn{p.ID, available, ha})
+		nodes = append(nodes, wn{p.ID, available, peerHA[p.ID], peerStreams[p.ID]})
 	}
 
 	// If we could not select enough peers, fail.
@@ -5761,10 +5784,14 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 		}
 		return nil, &err
 	}
-	// Sort based on available from most to least.
-	sort.Slice(nodes, func(i, j int) bool { return nodes[i].avail > nodes[j].avail })
-
-	// If we are placing a replicated stream, let's sort based in haAssets, as that is more important to balance.
+	// Sort based on available from most to least, breaking ties by number of total streams assigned to the peer.
+	sort.Slice(nodes, func(i, j int) bool {
+		if nodes[i].avail == nodes[j].avail {
+			return nodes[i].ns < nodes[j].ns
+		}
+		return nodes[i].avail > nodes[j].avail
+	})
+	// If we are placing a replicated stream, let's sort based on HAAssets, as that is more important to balance.
 	if cfg.Replicas > 1 {
 		sort.SliceStable(nodes, func(i, j int) bool { return nodes[i].ha < nodes[j].ha })
 	}
@@ -7171,6 +7198,10 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 			Created: time.Now().UTC(),
 		}
 	} else {
+		// If the consumer already exists then don't allow updating the PauseUntil, just set
+		// it back to whatever the current configured value is.
+		cfg.PauseUntil = ca.Config.PauseUntil
+
 		nca := ca.copyGroup()
 
 		rBefore := nca.Config.replicas(sa.Config)
diff --git a/server/jetstream_events.go b/server/jetstream_events.go
index 1852811b..b39e7cca 100644
--- a/server/jetstream_events.go
+++ b/server/jetstream_events.go
@@ -80,6 +80,18 @@ type JSConsumerActionAdvisory struct {
 
 const JSConsumerActionAdvisoryType = "io.nats.jetstream.advisory.v1.consumer_action"
 
+// JSConsumerPauseAdvisory indicates that a consumer was paused or unpaused
+type JSConsumerPauseAdvisory struct {
+	TypedEvent
+	Stream     string    `json:"stream"`
+	Consumer   string    `json:"consumer"`
+	Paused     bool      `json:"paused"`
+	PauseUntil time.Time `json:"pause_until,omitempty"`
+	Domain     string    `json:"domain,omitempty"`
+}
+
+const JSConsumerPauseAdvisoryType = "io.nats.jetstream.advisory.v1.consumer_pause"
+
 // JSConsumerAckMetric is a metric published when a user acknowledges a message, the
 // number of these that will be published is dependent on SampleFrequency
 type JSConsumerAckMetric struct {
diff --git a/server/leafnode.go b/server/leafnode.go
index 1be72160..b370ec41 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -2031,8 +2031,8 @@ func (s *Server) initLeafNodeSmapAndSendSubs(c *client) {
 		c.leaf.smap[oldGWReplyPrefix+"*.>"]++
 		c.leaf.smap[gwReplyPrefix+">"]++
 	}
-	// Detect loop by subscribing to a specific subject and checking
-	// if this is coming back to us.
+	// Detect loops by subscribing to a specific subject and checking
+	// if this sub is coming back to us.
 	c.leaf.smap[lds]++
 
 	// Check if we need to add an existing siReply to our map.
@@ -2096,7 +2096,7 @@ func (acc *Account) updateLeafNodes(sub *subscription, delta int32) {
 	isLDS := bytes.HasPrefix(sub.subject, []byte(leafNodeLoopDetectionSubjectPrefix))
 
 	// Capture the cluster even if its empty.
-	cluster := _EMPTY_
+	var cluster string
 	if sub.origin != nil {
 		cluster = bytesToString(sub.origin)
 	}
@@ -2125,11 +2125,11 @@ func (acc *Account) updateLeafNodes(sub *subscription, delta int32) {
 		}
 		// Check to make sure this sub does not have an origin cluster that matches the leafnode.
 		ln.mu.Lock()
-		skip := (cluster != _EMPTY_ && cluster == ln.remoteCluster()) || (delta > 0 && !ln.canSubscribe(subject))
 		// If skipped, make sure that we still let go the "$LDS." subscription that allows
-		// the detection of a loop.
-		if isLDS || !skip {
-			ln.updateSmap(sub, delta)
+		// the detection of loops as long as different cluster.
+		clusterDifferent := cluster != ln.remoteCluster()
+		if (isLDS && clusterDifferent) || ((cluster == _EMPTY_ || clusterDifferent) && (delta <= 0 || ln.canSubscribe(subject))) {
+			ln.updateSmap(sub, delta, isLDS)
 		}
 		ln.mu.Unlock()
 	}
@@ -2138,7 +2138,7 @@ func (acc *Account) updateLeafNodes(sub *subscription, delta int32) {
 // This will make an update to our internal smap and determine if we should send out
 // an interest update to the remote side.
 // Lock should be held.
-func (c *client) updateSmap(sub *subscription, delta int32) {
+func (c *client) updateSmap(sub *subscription, delta int32, isLDS bool) {
 	if c.leaf.smap == nil {
 		return
 	}
@@ -2146,7 +2146,7 @@ func (c *client) updateSmap(sub *subscription, delta int32) {
 	// If we are solicited make sure this is a local client or a non-solicited leaf node
 	skind := sub.client.kind
 	updateClient := skind == CLIENT || skind == SYSTEM || skind == JETSTREAM || skind == ACCOUNT
-	if c.isSpokeLeafNode() && !(updateClient || (skind == LEAF && !sub.client.isSpokeLeafNode())) {
+	if !isLDS && c.isSpokeLeafNode() && !(updateClient || (skind == LEAF && !sub.client.isSpokeLeafNode())) {
 		return
 	}
 
@@ -2328,6 +2328,7 @@ func (c *client) processLeafSub(argo []byte) (err error) {
 	acc := c.acc
 	// Check if we have a loop.
 	ldsPrefix := bytes.HasPrefix(sub.subject, []byte(leafNodeLoopDetectionSubjectPrefix))
+
 	if ldsPrefix && bytesToString(sub.subject) == acc.getLDSubject() {
 		c.mu.Unlock()
 		c.handleLeafNodeLoop(true)
@@ -2996,6 +2997,9 @@ func (s *Server) leafNodeFinishConnectProcess(c *client) {
 		if err == ErrTooManyAccountConnections {
 			c.maxAccountConnExceeded()
 			return
+		} else if err == ErrLeafNodeLoop {
+			c.handleLeafNodeLoop(true)
+			return
 		}
 		c.Errorf("Registering leaf with account %s resulted in error: %v", acc.Name, err)
 		c.closeConnection(ProtocolViolation)
diff --git a/server/memstore.go b/server/memstore.go
index 6abf0f00..87fa5b82 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2023 The NATS Authors
+// Copyright 2019-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
@@ -392,7 +392,7 @@ func (ms *memStore) filteredStateLocked(sseq uint64, filter string, lastPerSubje
 
 	// 1. See if we match any subs from fss.
 	// 2. If we match and the sseq is past ss.Last then we can use meta only.
-	// 3. If we match and we need to do a partial, break and clear any totals and do a full scan like num pending.
+	// 3. If we match we need to do a partial, break and clear any totals and do a full scan like num pending.
 
 	isMatch := func(subj string) bool {
 		if isAll {
@@ -429,7 +429,7 @@ func (ms *memStore) filteredStateLocked(sseq uint64, filter string, lastPerSubje
 		if sseq <= fss.First {
 			update(fss)
 		} else if sseq <= fss.Last {
-			// We matched but its a partial.
+			// We matched but it is a partial.
 			havePartial = true
 			// Don't break here, we will update to keep tracking last.
 			update(fss)
@@ -444,7 +444,7 @@ func (ms *memStore) filteredStateLocked(sseq uint64, filter string, lastPerSubje
 	// If we are here we need to scan the msgs.
 	// Capture first and last sequences for scan and then clear what we had.
 	first, last := ss.First, ss.Last
-	// To track if we decide to exclude and we need to calculate first.
+	// To track if we decide to exclude we need to calculate first.
 	var needScanFirst bool
 	if first < sseq {
 		first = sseq
@@ -792,10 +792,11 @@ func (ms *memStore) Compact(seq uint64) (uint64, error) {
 			ms.mu.Unlock()
 			return 0, ErrStoreMsgNotFound
 		}
+		fseq := ms.state.FirstSeq
 		ms.state.FirstSeq = seq
 		ms.state.FirstTime = time.Unix(0, sm.ts).UTC()
 
-		for seq := seq - 1; seq > 0; seq-- {
+		for seq := seq - 1; seq >= fseq; seq-- {
 			if sm := ms.msgs[seq]; sm != nil {
 				bytes += memStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 				purged++
@@ -1127,7 +1128,7 @@ func (ms *memStore) removeSeqPerSubject(subj string, seq uint64) {
 	}
 }
 
-// Will recalulate the first sequence for this subject in this block.
+// Will recalculate the first sequence for this subject in this block.
 // Lock should be held.
 func (ms *memStore) recalculateFirstForSubj(subj string, startSeq uint64, ss *SimpleState) {
 	for tseq := startSeq + 1; tseq <= ss.Last; tseq++ {
@@ -1140,7 +1141,7 @@ func (ms *memStore) recalculateFirstForSubj(subj string, startSeq uint64, ss *Si
 }
 
 // Removes the message referenced by seq.
-// Lock should he held.
+// Lock should be held.
 func (ms *memStore) removeMsg(seq uint64, secure bool) bool {
 	var ss uint64
 	sm, ok := ms.msgs[seq]
@@ -1479,7 +1480,7 @@ func (o *consumerMemStore) UpdateDelivered(dseq, sseq, dc uint64, ts int64) erro
 			if o.state.Redelivered == nil {
 				o.state.Redelivered = make(map[uint64]uint64)
 			}
-			// Only update if greater then what we already have.
+			// Only update if greater than what we already have.
 			if o.state.Redelivered[sseq] < dc-1 {
 				o.state.Redelivered[sseq] = dc - 1
 			}
diff --git a/server/mqtt.go b/server/mqtt.go
index 0e1ef71a..53e02230 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -2793,7 +2793,8 @@ func (as *mqttAccountSessionManager) loadRetainedMessages(subjects map[string]st
 // Composes a NATS message for a storeable mqttRetainedMsg.
 func mqttEncodeRetainedMessage(rm *mqttRetainedMsg) (natsMsg []byte, headerLen int) {
 	// No need to encode the subject, we can restore it from topic.
-	l := len(mqttNatsRetainedMessageTopic) + 1 + len(rm.Topic) + 2 // 1 byte for ':', 2 bytes for CRLF
+	l := len(hdrLine)
+	l += len(mqttNatsRetainedMessageTopic) + 1 + len(rm.Topic) + 2 // 1 byte for ':', 2 bytes for CRLF
 	if rm.Origin != _EMPTY_ {
 		l += len(mqttNatsRetainedMessageOrigin) + 1 + len(rm.Origin) + 2 // 1 byte for ':', 2 bytes for CRLF
 	}
@@ -2804,7 +2805,7 @@ func mqttEncodeRetainedMessage(rm *mqttRetainedMsg) (natsMsg []byte, headerLen i
 	l += 2                                             // 2 bytes for the extra CRLF after the header
 	l += len(rm.Msg)
 
-	buf := bytes.NewBuffer(make([]byte, l))
+	buf := bytes.NewBuffer(make([]byte, 0, l))
 
 	buf.WriteString(hdrLine)
 
diff --git a/server/msgtrace.go b/server/msgtrace.go
index 12358c2f..d5382129 100644
--- a/server/msgtrace.go
+++ b/server/msgtrace.go
@@ -17,6 +17,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"math/rand"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -435,15 +436,21 @@ func (c *client) initMsgTrace() *msgTrace {
 	// If external, we need to have the account's trace destination set,
 	// otherwise, we are not enabling tracing.
 	if external {
+		var sampling int
 		if acc != nil {
-			acc.mu.RLock()
-			dest = acc.TraceDest
-			acc.mu.RUnlock()
+			dest, sampling = acc.getTraceDestAndSampling()
 		}
 		if dest == _EMPTY_ {
 			// No account destination, no tracing for external trace headers.
 			return nil
 		}
+		// Check sampling, but only from origin server.
+		if c.kind == CLIENT && !sample(sampling) {
+			// Need to desactivate the traceParentHdr so that if the message
+			// is routed, it does possibly trigger a trace there.
+			disableTraceHeaders(c, hdr)
+			return nil
+		}
 	}
 	c.pa.trace = &msgTrace{
 		srv:  c.srv,
@@ -474,6 +481,15 @@ func (c *client) initMsgTrace() *msgTrace {
 	return c.pa.trace
 }
 
+func sample(sampling int) bool {
+	// Option parsing should ensure that sampling is [1..100], but consider
+	// any value outside of this range to be 100%.
+	if sampling <= 0 || sampling >= 100 {
+		return true
+	}
+	return rand.Int31n(100) <= int32(sampling)
+}
+
 // This function will return the header as a map (instead of http.Header because
 // we want to preserve the header names' case) and a boolean that indicates if
 // the headers have been lifted due to the presence of the external trace header
@@ -639,7 +655,7 @@ func (t *msgTrace) setHopHeader(c *client, msg []byte) []byte {
 // Note that if `msg` can be either the header alone or the full message
 // (header and payload). This function will use c.pa.hdr to limit the
 // search to the header section alone.
-func (t *msgTrace) disableTraceHeaders(c *client, msg []byte) []int {
+func disableTraceHeaders(c *client, msg []byte) []int {
 	// Code largely copied from getHeader(), except that we don't need the value
 	if c.pa.hdr <= 0 {
 		return []int{-1, -1}
@@ -674,7 +690,7 @@ func (t *msgTrace) disableTraceHeaders(c *client, msg []byte) []int {
 
 // Changes back the character at the given position `pos` in the `msg`
 // byte slice to the first character of the MsgTraceSendTo header.
-func (t *msgTrace) enableTraceHeaders(c *client, msg []byte, positions []int) {
+func enableTraceHeaders(c *client, msg []byte, positions []int) {
 	firstChar := [2]byte{MsgTraceDest[0], traceParentHdr[0]}
 	for i, pos := range positions {
 		if pos == -1 {
diff --git a/server/ocsp.go b/server/ocsp.go
index 4ecc3700..7f8fb8fa 100644
--- a/server/ocsp.go
+++ b/server/ocsp.go
@@ -14,12 +14,14 @@
 package server
 
 import (
+	"bytes"
 	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
 	"encoding/asn1"
 	"encoding/base64"
 	"encoding/pem"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -161,16 +163,43 @@ func (oc *OCSPMonitor) getRemoteStatus() ([]byte, *ocsp.Response, error) {
 	if config := opts.OCSPConfig; config != nil {
 		overrideURLs = config.OverrideURLs
 	}
-	getRequestBytes := func(u string, hc *http.Client) ([]byte, error) {
+	getRequestBytes := func(u string, reqDER []byte, hc *http.Client) ([]byte, error) {
+		reqEnc := base64.StdEncoding.EncodeToString(reqDER)
+		u = fmt.Sprintf("%s/%s", u, reqEnc)
+		start := time.Now()
 		resp, err := hc.Get(u)
 		if err != nil {
 			return nil, err
 		}
 		defer resp.Body.Close()
-		if resp.StatusCode != http.StatusOK {
-			return nil, fmt.Errorf("non-ok http status: %d", resp.StatusCode)
+
+		oc.srv.Debugf("Received OCSP response (method=GET, status=%v, url=%s, duration=%.3fs)",
+			resp.StatusCode, u, time.Since(start).Seconds())
+		if resp.StatusCode > 299 {
+			return nil, fmt.Errorf("non-ok http status on GET request (reqlen=%d): %d", len(reqEnc), resp.StatusCode)
 		}
+		return io.ReadAll(resp.Body)
+	}
+	postRequestBytes := func(u string, body []byte, hc *http.Client) ([]byte, error) {
+		hreq, err := http.NewRequest("POST", u, bytes.NewReader(body))
+		if err != nil {
+			return nil, err
+		}
+		hreq.Header.Add("Content-Type", "application/ocsp-request")
+		hreq.Header.Add("Accept", "application/ocsp-response")
 
+		start := time.Now()
+		resp, err := hc.Do(hreq)
+		if err != nil {
+			return nil, err
+		}
+		defer resp.Body.Close()
+
+		oc.srv.Debugf("Received OCSP response (method=POST, status=%v, url=%s, duration=%.3fs)",
+			resp.StatusCode, u, time.Since(start).Seconds())
+		if resp.StatusCode > 299 {
+			return nil, fmt.Errorf("non-ok http status on POST request (reqlen=%d): %d", len(body), resp.StatusCode)
+		}
 		return io.ReadAll(resp.Body)
 	}
 
@@ -182,8 +211,6 @@ func (oc *OCSPMonitor) getRemoteStatus() ([]byte, *ocsp.Response, error) {
 		return nil, nil, err
 	}
 
-	reqEnc := base64.StdEncoding.EncodeToString(reqDER)
-
 	responders := oc.Leaf.OCSPServer
 	if len(overrideURLs) > 0 {
 		responders = overrideURLs
@@ -195,18 +222,30 @@ func (oc *OCSPMonitor) getRemoteStatus() ([]byte, *ocsp.Response, error) {
 	oc.mu.Lock()
 	hc := oc.hc
 	oc.mu.Unlock()
+
 	var raw []byte
 	for _, u := range responders {
+		var postErr, getErr error
 		u = strings.TrimSuffix(u, "/")
-		raw, err = getRequestBytes(fmt.Sprintf("%s/%s", u, reqEnc), hc)
-		if err == nil {
+		// Prefer to make POST requests first.
+		raw, postErr = postRequestBytes(u, reqDER, hc)
+		if postErr == nil {
+			err = nil
 			break
+		} else {
+			// Fallback to use a GET request.
+			raw, getErr = getRequestBytes(u, reqDER, hc)
+			if getErr == nil {
+				err = nil
+				break
+			} else {
+				err = errors.Join(postErr, getErr)
+			}
 		}
 	}
 	if err != nil {
 		return nil, nil, fmt.Errorf("exhausted ocsp servers: %w", err)
 	}
-
 	resp, err := ocsp.ParseResponse(raw, oc.Issuer)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to get remote status: %w", err)
diff --git a/server/opts.go b/server/opts.go
index 22bd507c..a8361aee 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -1,4 +1,4 @@
-// Copyright 2012-2023 The NATS Authors
+// 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
@@ -434,6 +434,26 @@ type WebsocketOpts struct {
 	// "jwt" specified in the CONNECT options is missing or empty.
 	JWTCookie string
 
+	// Name of the cookie, which if present in WebSocket upgrade headers,
+	// will be treated as Username during CONNECT phase as long as
+	// "user" specified in the CONNECT options is missing or empty.
+	UsernameCookie string
+
+	// Name of the cookie, which if present in WebSocket upgrade headers,
+	// will be treated as Password during CONNECT phase as long as
+	// "pass" specified in the CONNECT options is missing or empty.
+	PasswordCookie string
+
+	// Name of the cookie, which if present in WebSocket upgrade headers,
+	// will be treated as Token during CONNECT phase as long as
+	// "auth_token" specified in the CONNECT options is missing or empty.
+	// Note that when this is useful for passing a JWT to an cuth callout
+	// when the server uses delegated authentication ("operator mode") or
+	// when using delegated authentication, but the auth callout validates some
+	// other JWT or string. Note that this does map to an actual server-wide
+	// "auth_token", note that using it for that purpose is greatly discouraged.
+	TokenCookie string
+
 	// Authentication section. If anything is configured in this section,
 	// it will override the authorization configuration of regular clients.
 	Username string
@@ -2293,7 +2313,7 @@ func parseLeafNodes(v interface{}, opts *Options, errors *[]error, warnings *[]e
 			}
 			opts.LeafNode.Remotes = remotes
 		case "reconnect", "reconnect_delay", "reconnect_interval":
-			opts.LeafNode.ReconnectInterval = time.Duration(int(mv.(int64))) * time.Second
+			opts.LeafNode.ReconnectInterval = parseDuration("reconnect", tk, mv, errors, warnings)
 		case "tls":
 			tc, err := parseTLS(tk, true)
 			if err != nil {
@@ -2896,6 +2916,69 @@ func parseAccountLimits(mv interface{}, acc *Account, errors *[]error, warnings
 	return nil
 }
 
+func parseAccountMsgTrace(mv any, topKey string, acc *Account) error {
+	processDest := func(tk token, k string, v any) error {
+		td, ok := v.(string)
+		if !ok {
+			return &configErr{tk, fmt.Sprintf("Field %q should be a string, got %T", k, v)}
+		}
+		if !IsValidPublishSubject(td) {
+			return &configErr{tk, fmt.Sprintf("Trace destination %q is not valid", td)}
+		}
+		acc.traceDest = td
+		return nil
+	}
+	processSampling := func(tk token, n int) error {
+		if n <= 0 || n > 100 {
+			return &configErr{tk, fmt.Sprintf("Ttrace destination sampling value %d is invalid, needs to be [1..100]", n)}
+		}
+		acc.traceDestSampling = n
+		return nil
+	}
+
+	var lt token
+	tk, v := unwrapValue(mv, &lt)
+	switch vv := v.(type) {
+	case string:
+		return processDest(tk, topKey, v)
+	case map[string]any:
+		for k, v := range vv {
+			tk, v := unwrapValue(v, &lt)
+			switch strings.ToLower(k) {
+			case "dest":
+				if err := processDest(tk, k, v); err != nil {
+					return err
+				}
+			case "sampling":
+				switch vv := v.(type) {
+				case int64:
+					if err := processSampling(tk, int(vv)); err != nil {
+						return err
+					}
+				case string:
+					s := strings.TrimSuffix(vv, "%")
+					n, err := strconv.Atoi(s)
+					if err != nil {
+						return &configErr{tk, fmt.Sprintf("Invalid trace destination sampling value %q", vv)}
+					}
+					if err := processSampling(tk, n); err != nil {
+						return err
+					}
+				default:
+					return &configErr{tk, fmt.Sprintf("Trace destination sampling field %q should be an integer or a percentage, got %T", k, v)}
+				}
+			default:
+				if !tk.IsUsedVariable() {
+					return &configErr{tk, fmt.Sprintf("Unknown field %q parsing account message trace map/struct %q", k, topKey)}
+				}
+			}
+		}
+	default:
+		return &configErr{tk, fmt.Sprintf("Expected account message trace %q to be a string or a map/struct, got %T", topKey, v)}
+	}
+	return nil
+}
+
 // parseAccounts will parse the different accounts syntax.
 func parseAccounts(v interface{}, opts *Options, errors *[]error, warnings *[]error) error {
 	var (
@@ -3025,14 +3108,23 @@ func parseAccounts(v interface{}, opts *Options, errors *[]error, warnings *[]er
 						*errors = append(*errors, err)
 						continue
 					}
-				case "trace_dest", "trace_destination", "trace_subject":
-					td := mv.(string)
-					if !IsValidPublishSubject(td) {
-						err := &configErr{tk, fmt.Sprintf("Trace destination %q is not valid", mv)}
+				case "msg_trace", "trace_dest":
+					if err := parseAccountMsgTrace(tk, k, acc); err != nil {
 						*errors = append(*errors, err)
 						continue
 					}
-					acc.TraceDest = td
+					// If trace destination is set but no sampling, set it to 100%.
+					if acc.traceDest != _EMPTY_ && acc.traceDestSampling == 0 {
+						acc.traceDestSampling = 100
+					} else if acc.traceDestSampling > 0 && acc.traceDest == _EMPTY_ {
+						// If no trace destination is provided, no trace would be
+						// triggered, so if the user set a sampling value expecting
+						// something to happen, want and set the value to 0 for good
+						// measure.
+						*warnings = append(*warnings,
+							&configErr{tk, "Trace destination sampling ignored since no destination was set"})
+						acc.traceDestSampling = 0
+					}
 				default:
 					if !tk.IsUsedVariable() {
 						err := &unknownConfigFieldErr{
@@ -4687,6 +4779,12 @@ func parseWebsocket(v interface{}, o *Options, errors *[]error, warnings *[]erro
 			o.Websocket.AuthTimeout = auth.timeout
 		case "jwt_cookie":
 			o.Websocket.JWTCookie = mv.(string)
+		case "user_cookie":
+			o.Websocket.UsernameCookie = mv.(string)
+		case "pass_cookie":
+			o.Websocket.PasswordCookie = mv.(string)
+		case "token_cookie":
+			o.Websocket.TokenCookie = mv.(string)
 		case "no_auth_user":
 			o.Websocket.NoAuthUser = mv.(string)
 		default:
diff --git a/server/raft.go b/server/raft.go
index f041fb08..f0be23ae 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -196,15 +196,14 @@ type raft struct {
 	hcommit   uint64 // The commit at the time that applies were paused
 	pobserver bool   // Whether we were an observer at the time that applies were paused
 
-	prop     *ipQueue[*Entry]               // Proposals
-	entry    *ipQueue[*appendEntry]         // Append entries
-	resp     *ipQueue[*appendEntryResponse] // Append entries responses
-	apply    *ipQueue[*CommittedEntry]      // Apply queue (committed entries to be passed to upper layer)
-	reqs     *ipQueue[*voteRequest]         // Vote requests
-	votes    *ipQueue[*voteResponse]        // Vote responses
-	stepdown *ipQueue[string]               // Stepdown requests
-	leadc    chan bool                      // Leader changes
-	quit     chan struct{}                  // Raft group shutdown
+	prop  *ipQueue[*Entry]               // Proposals
+	entry *ipQueue[*appendEntry]         // Append entries
+	resp  *ipQueue[*appendEntryResponse] // Append entries responses
+	apply *ipQueue[*CommittedEntry]      // Apply queue (committed entries to be passed to upper layer)
+	reqs  *ipQueue[*voteRequest]         // Vote requests
+	votes *ipQueue[*voteResponse]        // Vote responses
+	leadc chan bool                      // Leader changes
+	quit  chan struct{}                  // Raft group shutdown
 }
 
 // cacthupState structure that holds our subscription, and catchup term and index
@@ -388,7 +387,6 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 		entry:    newIPQueue[*appendEntry](s, qpfx+"appendEntry"),
 		resp:     newIPQueue[*appendEntryResponse](s, qpfx+"appendEntryResponse"),
 		apply:    newIPQueue[*CommittedEntry](s, qpfx+"committedEntry"),
-		stepdown: newIPQueue[string](s, qpfx+"stepdown"),
 		accName:  accName,
 		leadc:    make(chan bool, 1),
 		observer: cfg.Observer,
@@ -863,7 +861,7 @@ func (n *raft) PauseApply() error {
 
 	// If we are currently a candidate make sure we step down.
 	if n.State() == Candidate {
-		n.stepdown.push(noLeader)
+		n.stepdownLocked(noLeader)
 	}
 
 	n.debug("Pausing our apply channel")
@@ -1252,6 +1250,21 @@ func (n *raft) Leader() bool {
 	return n.State() == Leader
 }
 
+// stepdown immediately steps down the Raft node to the
+// follower state. This will take the lock itself.
+func (n *raft) stepdown(newLeader string) {
+	n.Lock()
+	defer n.Unlock()
+	n.stepdownLocked(newLeader)
+}
+
+// stepdownLocked immediately steps down the Raft node to the
+// follower state. This requires the lock is already held.
+func (n *raft) stepdownLocked(newLeader string) {
+	n.debug("Stepping down")
+	n.switchToFollowerLocked(newLeader)
+}
+
 // isCatchingUp returns true if a catchup is currently taking place.
 func (n *raft) isCatchingUp() bool {
 	n.RLock()
@@ -1459,7 +1472,6 @@ func (n *raft) StepDown(preferred ...string) error {
 	n.vote = noVote
 	n.writeTermVote()
 
-	stepdown := n.stepdown
 	prop := n.prop
 	n.Unlock()
 
@@ -1473,8 +1485,7 @@ func (n *raft) StepDown(preferred ...string) error {
 		prop.push(newEntry(EntryLeaderTransfer, []byte(maybeLeader)))
 	} else {
 		// Force us to stepdown here.
-		n.debug("Stepping down")
-		stepdown.push(noLeader)
+		n.stepdown(noLeader)
 	}
 
 	return nil
@@ -1651,7 +1662,7 @@ func (n *raft) shutdown(shouldDelete bool) {
 	// just will remove them from the central monitoring map
 	queues := []interface {
 		unregister()
-	}{n.reqs, n.votes, n.prop, n.entry, n.resp, n.apply, n.stepdown}
+	}{n.reqs, n.votes, n.prop, n.entry, n.resp, n.apply}
 	for _, q := range queues {
 		q.unregister()
 	}
@@ -1913,7 +1924,7 @@ func (n *raft) processAppendEntries() {
 // runAsFollower is called by run and will block for as long as the node is
 // running in the follower state.
 func (n *raft) runAsFollower() {
-	for {
+	for n.State() == Follower {
 		elect := n.electTimer()
 
 		select {
@@ -1964,13 +1975,6 @@ func (n *raft) runAsFollower() {
 			if voteReq, ok := n.reqs.popOne(); ok {
 				n.processVoteRequest(voteReq)
 			}
-		case <-n.stepdown.ch:
-			// We've received a stepdown request, start following the new leader if
-			// we can.
-			if newLeader, ok := n.stepdown.popOne(); ok {
-				n.switchToFollower(newLeader)
-				return
-			}
 		}
 	}
 }
@@ -2307,14 +2311,14 @@ func (n *raft) runAsLeader() {
 	fsub, err := n.subscribe(psubj, n.handleForwardedProposal)
 	if err != nil {
 		n.warn("Error subscribing to forwarded proposals: %v", err)
-		n.stepdown.push(noLeader)
+		n.stepdown(noLeader)
 		return
 	}
 	rpsub, err := n.subscribe(rpsubj, n.handleForwardedRemovePeerProposal)
 	if err != nil {
 		n.warn("Error subscribing to forwarded remove peer proposals: %v", err)
 		n.unsubscribe(fsub)
-		n.stepdown.push(noLeader)
+		n.stepdown(noLeader)
 		return
 	}
 
@@ -2369,7 +2373,7 @@ func (n *raft) runAsLeader() {
 				if b.Type == EntryLeaderTransfer {
 					n.prop.recycle(&es)
 					n.debug("Stepping down due to leadership transfer")
-					n.switchToFollower(noLeader)
+					n.stepdown(noLeader)
 					return
 				}
 				// We need to re-create `entries` because there is a reference
@@ -2384,7 +2388,7 @@ func (n *raft) runAsLeader() {
 			}
 		case <-lq.C:
 			if n.lostQuorum() {
-				n.switchToFollower(noLeader)
+				n.stepdown(noLeader)
 				return
 			}
 		case <-n.votes.ch:
@@ -2394,7 +2398,7 @@ func (n *raft) runAsLeader() {
 				continue
 			}
 			if vresp.term > n.Term() {
-				n.switchToFollower(noLeader)
+				n.stepdown(noLeader)
 				return
 			}
 			n.trackPeer(vresp.peer)
@@ -2403,11 +2407,6 @@ func (n *raft) runAsLeader() {
 			if voteReq, ok := n.reqs.popOne(); ok {
 				n.processVoteRequest(voteReq)
 			}
-		case <-n.stepdown.ch:
-			if newLeader, ok := n.stepdown.popOne(); ok {
-				n.switchToFollower(newLeader)
-				return
-			}
 		case <-n.entry.ch:
 			n.processAppendEntries()
 		}
@@ -2576,7 +2575,7 @@ func (n *raft) sendSnapshotToFollower(subject string) (uint64, error) {
 	snap, err := n.loadLastSnapshot()
 	if err != nil {
 		// We need to stepdown here when this happens.
-		n.stepdown.push(noLeader)
+		n.stepdownLocked(noLeader)
 		// We need to reset our state here as well.
 		n.resetWAL()
 		return 0, err
@@ -2642,7 +2641,7 @@ func (n *raft) catchupFollower(ar *appendEntryResponse) {
 		n.warn("Request from follower for entry at index [%d] errored for state %+v - %v", start, state, err)
 		if err == ErrStoreEOF {
 			// If we are here we are seeing a request for an item beyond our state, meaning we should stepdown.
-			n.stepdown.push(noLeader)
+			n.stepdownLocked(noLeader)
 			n.Unlock()
 			arPool.Put(ar)
 			return
@@ -2654,7 +2653,7 @@ func (n *raft) catchupFollower(ar *appendEntryResponse) {
 		// If we are here we are seeing a request for an item we do not have, meaning we should stepdown.
 		// This is possible on a reset of our WAL but the other side has a snapshot already.
 		// If we do not stepdown this can cycle.
-		n.stepdown.push(noLeader)
+		n.stepdownLocked(noLeader)
 		n.Unlock()
 		arPool.Put(ar)
 		return
@@ -2711,7 +2710,7 @@ func (n *raft) applyCommit(index uint64) error {
 			if err != ErrStoreClosed && err != ErrStoreEOF {
 				n.warn("Got an error loading %d index: %v - will reset", index, err)
 				if n.State() == Leader {
-					n.stepdown.push(n.selectNextLeader())
+					n.stepdownLocked(n.selectNextLeader())
 				}
 				// Reset and cancel any catchup.
 				n.resetWAL()
@@ -2789,7 +2788,7 @@ func (n *raft) applyCommit(index uint64) error {
 
 			// If this is us and we are the leader we should attempt to stepdown.
 			if peer == n.id && n.State() == Leader {
-				n.stepdown.push(n.selectNextLeader())
+				n.stepdown(n.selectNextLeader())
 			}
 
 			// Remove from string intern map.
@@ -2922,7 +2921,7 @@ func (n *raft) runAsCandidate() {
 	// We vote for ourselves.
 	votes := 1
 
-	for {
+	for n.State() == Candidate {
 		elect := n.electTimer()
 		select {
 		case <-n.entry.ch:
@@ -2965,8 +2964,8 @@ func (n *raft) runAsCandidate() {
 				n.term = vresp.term
 				n.vote = noVote
 				n.writeTermVote()
-				n.stepdown.push(noLeader)
 				n.lxfer = false
+				n.stepdownLocked(noLeader)
 				n.Unlock()
 			}
 		case <-n.reqs.ch:
@@ -2974,11 +2973,6 @@ func (n *raft) runAsCandidate() {
 			if voteReq, ok := n.reqs.popOne(); ok {
 				n.processVoteRequest(voteReq)
 			}
-		case <-n.stepdown.ch:
-			if newLeader, ok := n.stepdown.popOne(); ok {
-				n.switchToFollower(newLeader)
-				return
-			}
 		}
 	}
 }
@@ -3132,7 +3126,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			n.vote = noVote
 			n.writeTermVote()
 			n.debug("Received append entry from another leader, stepping down to %q", ae.leader)
-			n.stepdown.push(ae.leader)
+			n.stepdownLocked(ae.leader)
 		} else {
 			// Let them know we are the leader.
 			ar := newAppendEntryResponse(n.term, n.pindex, n.id, false)
@@ -3155,7 +3149,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			n.vote = noVote
 			n.writeTermVote()
 		}
-		n.stepdown.push(ae.leader)
+		n.stepdownLocked(ae.leader)
 	}
 
 	// Catching up state.
@@ -3217,7 +3211,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		}
 		if n.State() != Follower {
 			n.debug("Term higher than ours and we are not a follower: %v, stepping down to %q", n.State(), ae.leader)
-			n.stepdown.push(ae.leader)
+			n.stepdownLocked(ae.leader)
 		}
 	}
 
@@ -3458,7 +3452,7 @@ func (n *raft) processAppendEntryResponse(ar *appendEntryResponse) {
 		n.vote = noVote
 		n.writeTermVote()
 		n.warn("Detected another leader with higher term, will stepdown and reset")
-		n.stepdown.push(noLeader)
+		n.stepdownLocked(noLeader)
 		n.resetWAL()
 		n.Unlock()
 		arPool.Put(ar)
@@ -3506,7 +3500,7 @@ func (n *raft) storeToWAL(ae *appendEntry) error {
 	if index := ae.pindex + 1; index != seq {
 		n.warn("Wrong index, ae is %+v, index stored was %d, n.pindex is %d, will reset", ae, seq, n.pindex)
 		if n.State() == Leader {
-			n.stepdown.push(n.selectNextLeader())
+			n.stepdownLocked(n.selectNextLeader())
 		}
 		// Reset and cancel any catchup.
 		n.resetWAL()
@@ -3933,7 +3927,7 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 		if n.State() != Follower {
 			n.debug("Stepping down from %s, detected higher term: %d vs %d",
 				strings.ToLower(n.State().String()), vr.term, n.term)
-			n.stepdown.push(noLeader)
+			n.stepdownLocked(noLeader)
 			n.term = vr.term
 		}
 		n.vote = noVote
@@ -4064,13 +4058,17 @@ const (
 )
 
 func (n *raft) switchToFollower(leader string) {
+	n.Lock()
+	defer n.Unlock()
+
+	n.switchToFollowerLocked(leader)
+}
+
+func (n *raft) switchToFollowerLocked(leader string) {
 	if n.State() == Closed {
 		return
 	}
 
-	n.Lock()
-	defer n.Unlock()
-
 	n.debug("Switching to follower")
 
 	n.lxfer = false
diff --git a/server/server.go b/server/server.go
index 15c67c5b..8003a076 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1046,15 +1046,6 @@ func validateOptions(o *Options) error {
 		return fmt.Errorf("max_payload (%v) cannot be higher than max_pending (%v)",
 			o.MaxPayload, o.MaxPending)
 	}
-	// Check that account's trace_dest is a valid subject.
-	for _, acc := range o.Accounts {
-		if acc.TraceDest == _EMPTY_ {
-			continue
-		}
-		if !IsValidPublishSubject(acc.TraceDest) {
-			return fmt.Errorf("trace_dest %q of account %q is not valid", acc.TraceDest, acc.Name)
-		}
-	}
 	// Check that the trust configuration is correct.
 	if err := validateTrustedOperators(o); err != nil {
 		return err
@@ -1195,9 +1186,11 @@ func (s *Server) configureAccounts(reloading bool) (map[string]struct{}, error)
 		// If we have leafnodes they need to be updated.
 		if reloading && a == s.gacc {
 			a.mu.Lock()
-			var mappings []*mapping
+			mappings := make(map[string]*mapping)
 			if len(a.mappings) > 0 && a.nleafs > 0 {
-				mappings = append(mappings, a.mappings...)
+				for _, em := range a.mappings {
+					mappings[em.src] = em
+				}
 			}
 			a.mu.Unlock()
 			if len(mappings) > 0 || len(oldGMappings) > 0 {
@@ -1208,7 +1201,10 @@ func (s *Server) configureAccounts(reloading bool) (map[string]struct{}, error)
 					}
 					// Remove any old ones if needed.
 					for _, em := range oldGMappings {
-						lc.forceRemoveFromSmap(em.src)
+						// Only remove if not in the new ones.
+						if _, ok := mappings[em.src]; !ok {
+							lc.forceRemoveFromSmap(em.src)
+						}
 					}
 				}
 				a.lmu.RUnlock()
diff --git a/server/stream.go b/server/stream.go
index df27e4b1..405ee958 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -2003,12 +2003,13 @@ func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err err
 	// Purge consumers.
 	// Check for filtered purge.
 	if preq != nil && preq.Subject != _EMPTY_ {
-		ss := store.FilteredState(state.FirstSeq, preq.Subject)
+		ss := store.FilteredState(fseq, preq.Subject)
 		fseq = ss.First
 	}
 
 	mset.clsMu.RLock()
 	for _, o := range mset.cList {
+		start := fseq
 		o.mu.RLock()
 		// we update consumer sequences if:
 		// no subject was specified, we can purge all consumers sequences
@@ -2018,10 +2019,15 @@ func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err err
 			// or consumer filter subject is subset of purged subject,
 			// but not the other way around.
 			o.isEqualOrSubsetMatch(preq.Subject)
+		// Check if a consumer has a wider subject space then what we purged
+		var isWider bool
+		if !doPurge && preq != nil && o.isFilteredMatch(preq.Subject) {
+			doPurge, isWider = true, true
+			start = state.FirstSeq
+		}
 		o.mu.RUnlock()
 		if doPurge {
-			o.purge(fseq, lseq)
-
+			o.purge(start, lseq, isWider)
 		}
 	}
 	mset.clsMu.RUnlock()
@@ -4158,7 +4164,7 @@ func (mset *stream) getDirectRequest(req *JSApiMsgGetRequest, reply string) {
 
 		if isBatchRequest {
 			if len(hdr) == 0 {
-				hdr = []byte(fmt.Sprintf(dgb, name, sm.subj, sm.seq, ts.Format(time.RFC3339Nano), np, lseq))
+				hdr = fmt.Appendf(nil, dgb, name, sm.subj, sm.seq, ts.Format(time.RFC3339Nano), np, lseq)
 			} else {
 				hdr = copyBytes(hdr)
 				hdr = genHeader(hdr, JSStream, name)
@@ -4172,7 +4178,7 @@ func (mset *stream) getDirectRequest(req *JSApiMsgGetRequest, reply string) {
 			np--
 		} else {
 			if len(hdr) == 0 {
-				hdr = []byte(fmt.Sprintf(dg, name, sm.subj, sm.seq, ts.Format(time.RFC3339Nano)))
+				hdr = fmt.Appendf(nil, dg, name, sm.subj, sm.seq, ts.Format(time.RFC3339Nano))
 			} else {
 				hdr = copyBytes(hdr)
 				hdr = genHeader(hdr, JSStream, name)
@@ -4198,7 +4204,7 @@ func (mset *stream) getDirectRequest(req *JSApiMsgGetRequest, reply string) {
 		if mset.lastSeq() > validThrough {
 			np, _ = store.NumPending(seq, req.NextFor, false)
 		}
-		hdr := []byte(fmt.Sprintf("NATS/1.0 204 EOB\r\nNats-Num-Pending: %d\r\nNats-Last-Sequence: %d\r\n\r\n", np, lseq))
+		hdr := fmt.Appendf(nil, "NATS/1.0 204 EOB\r\nNats-Num-Pending: %d\r\nNats-Last-Sequence: %d\r\n\r\n", np, lseq)
 		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 	}
 }
@@ -4218,7 +4224,7 @@ func (mset *stream) processInboundJetStreamMsg(_ *subscription, c *client, _ *Ac
 		// to prevent a trace event to be generated when a stored message
 		// is delivered to a consumer and routed.
 		if !traceOnly {
-			mt.disableTraceHeaders(c, hdr)
+			disableTraceHeaders(c, hdr)
 		}
 		// This will add the jetstream event while in the client read loop.
 		// Since the event will be updated in a different go routine, the
@@ -4679,10 +4685,10 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 			const ht = "NATS/1.0\r\nNats-Stream: %s\r\nNats-Subject: %s\r\nNats-Sequence: %d\r\nNats-Time-Stamp: %s\r\nNats-Last-Sequence: %d\r\n\r\n"
 			const htho = "NATS/1.0\r\nNats-Stream: %s\r\nNats-Subject: %s\r\nNats-Sequence: %d\r\nNats-Time-Stamp: %s\r\nNats-Last-Sequence: %d\r\nNats-Msg-Size: %d\r\n\r\n"
 			if !thdrsOnly {
-				hdr = []byte(fmt.Sprintf(ht, name, subject, seq, tsStr, tlseq))
+				hdr = fmt.Appendf(nil, ht, name, subject, seq, tsStr, tlseq)
 				rpMsg = copyBytes(msg)
 			} else {
-				hdr = []byte(fmt.Sprintf(htho, name, subject, seq, tsStr, tlseq, len(msg)))
+				hdr = fmt.Appendf(nil, htho, name, subject, seq, tsStr, tlseq, len(msg))
 			}
 		} else {
 			// Slow path.
diff --git a/server/stree/node.go b/server/stree/node.go
index 0afecfab..6285ba3c 100644
--- a/server/stree/node.go
+++ b/server/stree/node.go
@@ -34,8 +34,7 @@ type node interface {
 
 // Maximum prefix len
 // We expect the most savings to come from long shared prefixes.
-// This also makes the meta base layer exactly 64 bytes, a normal L1 cache line.
-const maxPrefixLen = 60
+const maxPrefixLen = 24
 
 // 64 bytes total - an L1 cache line.
 type meta struct {
diff --git a/server/stree/parts.go b/server/stree/parts.go
index 71783ee1..8106ccb4 100644
--- a/server/stree/parts.go
+++ b/server/stree/parts.go
@@ -99,10 +99,9 @@ func matchParts(parts [][]byte, frag []byte) ([][]byte, bool) {
 		}
 		end := min(si+lp, len(frag))
 		// If part is bigger then the fragment, adjust to a portion on the part.
-		partialPart := lp > end
-		if partialPart {
+		if partialPart := lp > end; partialPart {
 			// Frag is smaller then part itself.
-			part = part[:end]
+			part = part[:end-si]
 		}
 		if !bytes.Equal(part, frag[si:end]) {
 			return parts, false
@@ -117,7 +116,7 @@ func matchParts(parts [][]byte, frag []byte) ([][]byte, bool) {
 		if end < lp {
 			if end >= lf {
 				parts = append([][]byte{}, parts...) // Create a copy before modifying.
-				parts[i] = parts[i][lf:]
+				parts[i] = parts[i][lf-si:]
 			} else {
 				i++
 			}
diff --git a/server/stree/stree.go b/server/stree/stree.go
index 452f5f35..ad75eb5f 100644
--- a/server/stree/stree.go
+++ b/server/stree/stree.go
@@ -278,7 +278,7 @@ func (t *SubjectTree[T]) match(n node, parts [][]byte, pre []byte, cb func(subje
 		}
 		// We have matched here. If we are a leaf and have exhausted all parts or he have a FWC fire callback.
 		if n.isLeaf() {
-			if len(nparts) == 0 || hasFWC {
+			if len(nparts) == 0 || (hasFWC && len(nparts) == 1) {
 				ln := n.(*leaf[T])
 				cb(append(pre, ln.suffix...), &ln.value)
 			}
diff --git a/server/sublist.go b/server/sublist.go
index fdd7cc5f..0d70cb64 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -1,4 +1,4 @@
-// Copyright 2016-2023 The NATS Authors
+// Copyright 2016-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
@@ -1052,7 +1052,16 @@ func visitLevel(l *level, depth int) int {
 
 // Determine if a subject has any wildcard tokens.
 func subjectHasWildcard(subject string) bool {
-	return !subjectIsLiteral(subject)
+	// This one exits earlier then !subjectIsLiteral(subject)
+	for i, c := range subject {
+		if c == pwc || c == fwc {
+			if (i == 0 || subject[i-1] == btsep) &&
+				(i+1 == len(subject) || subject[i+1] == btsep) {
+				return true
+			}
+		}
+	}
+	return false
 }
 
 // Determine if the subject has any wildcards. Fast version, does not check for
diff --git a/server/websocket.go b/server/websocket.go
index e026674d..7c9bb718 100644
--- a/server/websocket.go
+++ b/server/websocket.go
@@ -1,4 +1,4 @@
-// Copyright 2020-2023 The NATS Authors
+// Copyright 2020-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
@@ -106,18 +106,21 @@ var wsGUID = []byte("258EAFA5-E914-47DA-95CA-C5AB0DC85B11")
 var wsTestRejectNoMasking = false
 
 type websocket struct {
-	frames     net.Buffers
-	fs         int64
-	closeMsg   []byte
-	compress   bool
-	closeSent  bool
-	browser    bool
-	nocompfrag bool // No fragment for compressed frames
-	maskread   bool
-	maskwrite  bool
-	compressor *flate.Writer
-	cookieJwt  string
-	clientIP   string
+	frames         net.Buffers
+	fs             int64
+	closeMsg       []byte
+	compress       bool
+	closeSent      bool
+	browser        bool
+	nocompfrag     bool // No fragment for compressed frames
+	maskread       bool
+	maskwrite      bool
+	compressor     *flate.Writer
+	cookieJwt      string
+	cookieUsername string
+	cookiePassword string
+	cookieToken    string
+	clientIP       string
 }
 
 type srvWebsocket struct {
@@ -807,9 +810,19 @@ func (s *Server) wsUpgrade(w http.ResponseWriter, r *http.Request) (*wsUpgradeRe
 			// So make the combination of the two.
 			ws.nocompfrag = ws.compress && strings.Contains(ua, "Version/") && strings.Contains(ua, "Safari/")
 		}
-		if opts.Websocket.JWTCookie != _EMPTY_ {
-			if c, err := r.Cookie(opts.Websocket.JWTCookie); err == nil && c != nil {
-				ws.cookieJwt = c.Value
+
+		if cookies := r.Cookies(); len(cookies) > 0 {
+			ows := &opts.Websocket
+			for _, c := range cookies {
+				if ows.JWTCookie == c.Name {
+					ws.cookieJwt = c.Value
+				} else if ows.UsernameCookie == c.Name {
+					ws.cookieUsername = c.Value
+				} else if ows.PasswordCookie == c.Name {
+					ws.cookiePassword = c.Value
+				} else if ows.TokenCookie == c.Name {
+					ws.cookieToken = c.Value
+				}
 			}
 		}
 	}


Test patch:
diff --git a/.github/workflows/MQTT_test.yaml b/.github/workflows/MQTT_test.yaml
index bfbd9210..2bfd0bed 100644
--- a/.github/workflows/MQTT_test.yaml
+++ b/.github/workflows/MQTT_test.yaml
@@ -29,7 +29,7 @@ jobs:
         run: |
           wget https://github.com/hivemq/mqtt-cli/releases/download/v4.20.0/mqtt-cli-4.20.0.deb
           sudo apt install ./mqtt-cli-4.20.0.deb
-          go install github.com/ConnectEverything/mqtt-test@latest
+          go install github.com/ConnectEverything/mqtt-test@4dd571c31318dcfebe5443242f53f262403ceafb
 
       # - name: Download benchmark result for ${{ github.base_ref || github.ref_name }}
       #   uses: dawidd6/action-download-artifact@v2
diff --git a/server/accounts_test.go b/server/accounts_test.go
index f586af0f..eca0374a 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -510,6 +510,7 @@ func TestAccountSimpleConfig(t *testing.T) {
 
 func TestAccountParseConfig(t *testing.T) {
 	traceDest := "my.trace.dest"
+	traceSampling := 50
 	confFileName := createConfFile(t, []byte(fmt.Sprintf(`
     accounts {
       synadia {
@@ -519,14 +520,14 @@ func TestAccountParseConfig(t *testing.T) {
         ]
       }
       nats.io {
-		trace_dest: %q
+		msg_trace: {dest: %q, sampling: %d%%}
         users = [
           {user: derek, password: foo}
           {user: ivan, password: bar}
         ]
       }
     }
-    `, traceDest)))
+    `, traceDest, traceSampling)))
 	opts, err := ProcessConfigFile(confFileName)
 	if err != nil {
 		t.Fatalf("Received an error processing config file: %v", err)
@@ -550,7 +551,9 @@ func TestAccountParseConfig(t *testing.T) {
 	if natsAcc == nil {
 		t.Fatalf("Error retrieving account for 'nats.io'")
 	}
-	require_Equal[string](t, natsAcc.TraceDest, traceDest)
+	td, tds := natsAcc.getTraceDestAndSampling()
+	require_Equal[string](t, td, traceDest)
+	require_Equal[int](t, tds, traceSampling)
 
 	for _, u := range opts.Users {
 		if u.Username == "derek" {
diff --git a/server/config_check_test.go b/server/config_check_test.go
index 23d3b27e..aefebc74 100644
--- a/server/config_check_test.go
+++ b/server/config_check_test.go
@@ -920,7 +920,7 @@ func TestConfigCheck(t *testing.T) {
                   A { trace_dest: 123 }
                 }
 			`,
-			err:       errors.New(`interface conversion: interface {} is int64, not string`),
+			err:       errors.New(`Expected account message trace "trace_dest" to be a string or a map/struct, got int64`),
 			errorLine: 3,
 			errorPos:  23,
 		},
@@ -946,6 +946,94 @@ func TestConfigCheck(t *testing.T) {
 			errorLine: 3,
 			errorPos:  23,
 		},
+		{
+			name: "when account message trace dest is wrong type",
+			config: `
+                accounts {
+                  A { msg_trace: {dest: 123} }
+                }
+			`,
+			err:       errors.New(`Field "dest" should be a string, got int64`),
+			errorLine: 3,
+			errorPos:  35,
+		},
+		{
+			name: "when account message trace dest is invalid",
+			config: `
+                accounts {
+                  A { msg_trace: {dest: "invalid..dest"} }
+                }
+			`,
+			err:       errors.New(`Trace destination "invalid..dest" is not valid`),
+			errorLine: 3,
+			errorPos:  35,
+		},
+		{
+			name: "when account message trace sampling is wrong type",
+			config: `
+                accounts {
+                  A { msg_trace: {dest: "acc.dest", sampling: {wront: "type"}} }
+                }
+			`,
+			err:       errors.New(`Trace destination sampling field "sampling" should be an integer or a percentage, got map[string]interface {}`),
+			errorLine: 3,
+			errorPos:  53,
+		},
+		{
+			name: "when account message trace sampling is wrong string",
+			config: `
+                accounts {
+                  A { msg_trace: {dest: "acc.dest", sampling: abc%} }
+                }
+			`,
+			err:       errors.New(`Invalid trace destination sampling value "abc%"`),
+			errorLine: 3,
+			errorPos:  53,
+		},
+		{
+			name: "when account message trace sampling is negative",
+			config: `
+                accounts {
+                  A { msg_trace: {dest: "acc.dest", sampling: -1} }
+                }
+			`,
+			err:       errors.New(`Ttrace destination sampling value -1 is invalid, needs to be [1..100]`),
+			errorLine: 3,
+			errorPos:  53,
+		},
+		{
+			name: "when account message trace sampling is zero",
+			config: `
+                accounts {
+                  A { msg_trace: {dest: "acc.dest", sampling: 0} }
+                }
+			`,
+			err:       errors.New(`Ttrace destination sampling value 0 is invalid, needs to be [1..100]`),
+			errorLine: 3,
+			errorPos:  53,
+		},
+		{
+			name: "when account message trace sampling is more than 100",
+			config: `
+                accounts {
+                  A { msg_trace: {dest: "acc.dest", sampling: 101} }
+                }
+			`,
+			err:       errors.New(`Ttrace destination sampling value 101 is invalid, needs to be [1..100]`),
+			errorLine: 3,
+			errorPos:  53,
+		},
+		{
+			name: "when account message trace has unknown field",
+			config: `
+                accounts {
+                  A { msg_trace: {wrong: "field"} }
+                }
+			`,
+			err:       errors.New(`Unknown field "wrong" parsing account message trace map/struct "msg_trace"`),
+			errorLine: 3,
+			errorPos:  35,
+		},
 		{
 			name: "when user authorization config has both token and users",
 			config: `
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 25ac9b34..032e49cd 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -6519,3 +6519,446 @@ Setup:
 		}
 	}
 }
+
+// https://github.com/nats-io/nats-server/issues/5071
+func TestJetStreamClusterStreamPlacementDistribution(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 5)
+	defer c.shutdown()
+
+	s := c.randomNonLeader()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	for i := 1; i <= 10; i++ {
+		_, err := js.AddStream(&nats.StreamConfig{
+			Name:     fmt.Sprintf("TEST:%d", i),
+			Subjects: []string{fmt.Sprintf("foo.%d.*", i)},
+			Replicas: 3,
+		})
+		require_NoError(t, err)
+	}
+
+	// 10 streams, 3 replicas div 5 servers.
+	expectedStreams := 10 * 3 / 5
+	for _, s := range c.servers {
+		jsz, err := s.Jsz(nil)
+		require_NoError(t, err)
+		require_Equal(t, jsz.Streams, expectedStreams)
+	}
+}
+
+func TestJetStreamClusterConsumerPauseViaConfig(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+		Name:     "my_consumer",
+		Replicas: 3,
+	})
+
+	sub, err := js.PullSubscribe("foo", "", nats.Bind("TEST", "my_consumer"))
+	require_NoError(t, err)
+
+	stepdown := func() {
+		t.Helper()
+		_, err := nc.Request(fmt.Sprintf(JSApiConsumerLeaderStepDownT, "TEST", "my_consumer"), nil, time.Second)
+		require_NoError(t, err)
+		c.waitOnConsumerLeader(globalAccountName, "TEST", "my_consumer")
+	}
+
+	publish := func(wait time.Duration) {
+		t.Helper()
+		for i := 0; i < 5; i++ {
+			_, err = js.Publish("foo", []byte("OK"))
+			require_NoError(t, err)
+		}
+		msgs, err := sub.Fetch(5, nats.MaxWait(wait))
+		require_NoError(t, err)
+		require_Equal(t, len(msgs), 5)
+	}
+
+	// This should be fast as there's no deadline.
+	publish(time.Second)
+
+	// Now we're going to set the deadline.
+	deadline := jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", time.Now().Add(time.Second*3))
+
+	// It will now take longer than 3 seconds.
+	publish(time.Second * 5)
+	require_True(t, time.Now().After(deadline))
+
+	// The next set of publishes after the deadline should now be fast.
+	publish(time.Second)
+
+	// We'll kick the leader, but since we're after the deadline, this
+	// should still be fast.
+	stepdown()
+	publish(time.Second)
+
+	// Now we're going to do an update and then immediately kick the
+	// leader. The pause should still be in effect afterwards.
+	deadline = jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", time.Now().Add(time.Second*3))
+	publish(time.Second * 5)
+	require_True(t, time.Now().After(deadline))
+
+	// The next set of publishes after the deadline should now be fast.
+	publish(time.Second)
+}
+
+func TestJetStreamClusterConsumerPauseViaEndpoint(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"push", "pull"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	t.Run("PullConsumer", func(t *testing.T) {
+		_, err := js.AddConsumer("TEST", &nats.ConsumerConfig{
+			Name: "pull_consumer",
+		})
+		require_NoError(t, err)
+
+		sub, err := js.PullSubscribe("pull", "", nats.Bind("TEST", "pull_consumer"))
+		require_NoError(t, err)
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("pull", []byte("OK"))
+			require_NoError(t, err)
+		}
+		msgs, err := sub.Fetch(10, nats.MaxWait(time.Second))
+		require_NoError(t, err)
+		require_Equal(t, len(msgs), 10)
+
+		// Now we'll pause the consumer for 3 seconds.
+		deadline := time.Now().Add(time.Second * 3)
+		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "pull_consumer", deadline).Equal(deadline))
+
+		// This should fail as we'll wait for only half of the deadline.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("pull", []byte("OK"))
+			require_NoError(t, err)
+		}
+		_, err = sub.Fetch(10, nats.MaxWait(time.Until(deadline)/2))
+		require_Error(t, err, nats.ErrTimeout)
+
+		// This should succeed after a short wait, and when we're done,
+		// we should be after the deadline.
+		msgs, err = sub.Fetch(10)
+		require_NoError(t, err)
+		require_Equal(t, len(msgs), 10)
+		require_True(t, time.Now().After(deadline))
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("pull", []byte("OK"))
+			require_NoError(t, err)
+		}
+		msgs, err = sub.Fetch(10, nats.MaxWait(time.Second))
+		require_NoError(t, err)
+		require_Equal(t, len(msgs), 10)
+
+		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "pull_consumer", time.Time{}).Equal(time.Time{}))
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("pull", []byte("OK"))
+			require_NoError(t, err)
+		}
+		msgs, err = sub.Fetch(10, nats.MaxWait(time.Second))
+		require_NoError(t, err)
+		require_Equal(t, len(msgs), 10)
+	})
+
+	t.Run("PushConsumer", func(t *testing.T) {
+		ch := make(chan *nats.Msg, 100)
+		_, err = js.ChanSubscribe("push", ch, nats.BindStream("TEST"), nats.ConsumerName("push_consumer"))
+		require_NoError(t, err)
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("push", []byte("OK"))
+			require_NoError(t, err)
+		}
+		for i := 0; i < 10; i++ {
+			msg := require_ChanRead(t, ch, time.Second)
+			require_NotEqual(t, msg, nil)
+		}
+
+		// Now we'll pause the consumer for 3 seconds.
+		deadline := time.Now().Add(time.Second * 3)
+		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "push_consumer", deadline).Equal(deadline))
+
+		// This should succeed after a short wait, and when we're done,
+		// we should be after the deadline.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("push", []byte("OK"))
+			require_NoError(t, err)
+		}
+		for i := 0; i < 10; i++ {
+			msg := require_ChanRead(t, ch, time.Second*5)
+			require_NotEqual(t, msg, nil)
+			require_True(t, time.Now().After(deadline))
+		}
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("push", []byte("OK"))
+			require_NoError(t, err)
+		}
+		for i := 0; i < 10; i++ {
+			msg := require_ChanRead(t, ch, time.Second)
+			require_NotEqual(t, msg, nil)
+		}
+
+		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "push_consumer", time.Time{}).Equal(time.Time{}))
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("push", []byte("OK"))
+			require_NoError(t, err)
+		}
+		for i := 0; i < 10; i++ {
+			msg := require_ChanRead(t, ch, time.Second)
+			require_NotEqual(t, msg, nil)
+		}
+	})
+}
+
+func TestJetStreamClusterConsumerPauseTimerFollowsLeader(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	deadline := time.Now().Add(time.Hour)
+	jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+		Name:       "my_consumer",
+		PauseUntil: &deadline,
+		Replicas:   3,
+	})
+
+	for i := 0; i < 10; i++ {
+		c.waitOnConsumerLeader(globalAccountName, "TEST", "my_consumer")
+		c.waitOnAllCurrent()
+
+		for _, s := range c.servers {
+			stream, err := s.gacc.lookupStream("TEST")
+			require_NoError(t, err)
+
+			consumer := stream.lookupConsumer("my_consumer")
+			require_NotEqual(t, consumer, nil)
+
+			isLeader := s.JetStreamIsConsumerLeader(globalAccountName, "TEST", "my_consumer")
+
+			consumer.mu.RLock()
+			hasTimer := consumer.uptmr != nil
+			consumer.mu.RUnlock()
+
+			require_Equal(t, isLeader, hasTimer)
+		}
+
+		_, err = nc.Request(fmt.Sprintf(JSApiConsumerLeaderStepDownT, "TEST", "my_consumer"), nil, time.Second)
+		require_NoError(t, err)
+	}
+}
+
+func TestJetStreamClusterConsumerPauseHeartbeats(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	deadline := time.Now().Add(time.Hour)
+	dsubj := "deliver_subj"
+
+	ci := jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+		Name:           "my_consumer",
+		PauseUntil:     &deadline,
+		Heartbeat:      time.Millisecond * 100,
+		DeliverSubject: dsubj,
+	})
+	require_True(t, ci.Config.PauseUntil.Equal(deadline))
+
+	ch := make(chan *nats.Msg, 10)
+	_, err = nc.ChanSubscribe(dsubj, ch)
+	require_NoError(t, err)
+
+	for i := 0; i < 20; i++ {
+		msg := require_ChanRead(t, ch, time.Millisecond*200)
+		require_Equal(t, msg.Header.Get("Status"), "100")
+		require_Equal(t, msg.Header.Get("Description"), "Idle Heartbeat")
+	}
+}
+
+func TestJetStreamClusterConsumerPauseAdvisories(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	checkAdvisory := func(msg *nats.Msg, shouldBePaused bool, deadline time.Time) {
+		t.Helper()
+		var advisory JSConsumerPauseAdvisory
+		require_NoError(t, json.Unmarshal(msg.Data, &advisory))
+		require_Equal(t, advisory.Stream, "TEST")
+		require_Equal(t, advisory.Consumer, "my_consumer")
+		require_Equal(t, advisory.Paused, shouldBePaused)
+		require_True(t, advisory.PauseUntil.Equal(deadline))
+	}
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	ch := make(chan *nats.Msg, 10)
+	_, err = nc.ChanSubscribe(JSAdvisoryConsumerPausePre+".TEST.my_consumer", ch)
+	require_NoError(t, err)
+
+	deadline := time.Now().Add(time.Second)
+	jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+		Name:       "my_consumer",
+		PauseUntil: &deadline,
+		Replicas:   3,
+	})
+
+	// First advisory should tell us that the consumer was paused
+	// on creation.
+	msg := require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, true, deadline)
+	require_Len(t, len(ch), 0) // Should only receive one advisory.
+
+	// The second one for the unpause.
+	msg = require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, false, deadline)
+	require_Len(t, len(ch), 0) // Should only receive one advisory.
+
+	// Now we'll pause the consumer for a second using the API.
+	deadline = time.Now().Add(time.Second)
+	require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", deadline).Equal(deadline))
+
+	// Third advisory should tell us about the pause via the API.
+	msg = require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, true, deadline)
+	require_Len(t, len(ch), 0) // Should only receive one advisory.
+
+	// Finally that should unpause.
+	msg = require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, false, deadline)
+	require_Len(t, len(ch), 0) // Should only receive one advisory.
+
+	// Now we're going to set the deadline into the future so we can
+	// see what happens when we kick leaders or restart.
+	deadline = time.Now().Add(time.Hour)
+	require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", deadline).Equal(deadline))
+
+	// Setting the deadline should have generated an advisory.
+	msg = require_ChanRead(t, ch, time.Second)
+	checkAdvisory(msg, true, deadline)
+	require_Len(t, len(ch), 0) // Should only receive one advisory.
+
+	// Try to kick the consumer leader.
+	srv := c.consumerLeader(globalAccountName, "TEST", "my_consumer")
+	srv.JetStreamStepdownConsumer(globalAccountName, "TEST", "my_consumer")
+	c.waitOnConsumerLeader(globalAccountName, "TEST", "my_consumer")
+
+	// This shouldn't have generated an advisory.
+	require_NoChanRead(t, ch, time.Second)
+}
+
+func TestJetStreamClusterConsumerPauseSurvivesRestart(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	checkTimer := func(s *Server) {
+		stream, err := s.gacc.lookupStream("TEST")
+		require_NoError(t, err)
+
+		consumer := stream.lookupConsumer("my_consumer")
+		require_NotEqual(t, consumer, nil)
+
+		consumer.mu.RLock()
+		timer := consumer.uptmr
+		consumer.mu.RUnlock()
+		require_True(t, timer != nil)
+	}
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	deadline := time.Now().Add(time.Hour)
+	jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+		Name:       "my_consumer",
+		PauseUntil: &deadline,
+		Replicas:   3,
+	})
+
+	// First try with just restarting the consumer leader.
+	srv := c.consumerLeader(globalAccountName, "TEST", "my_consumer")
+	srv.Shutdown()
+	c.restartServer(srv)
+	c.waitOnAllCurrent()
+	c.waitOnConsumerLeader(globalAccountName, "TEST", "my_consumer")
+	leader := c.consumerLeader(globalAccountName, "TEST", "my_consumer")
+	require_True(t, leader != nil)
+	checkTimer(leader)
+
+	// Then try restarting the entire cluster.
+	c.stopAll()
+	c.restartAllSamePorts()
+	c.waitOnAllCurrent()
+	c.waitOnConsumerLeader(globalAccountName, "TEST", "my_consumer")
+	leader = c.consumerLeader(globalAccountName, "TEST", "my_consumer")
+	require_True(t, leader != nil)
+	checkTimer(leader)
+}
diff --git a/server/jetstream_super_cluster_test.go b/server/jetstream_super_cluster_test.go
index b7b7f69f..a000f959 100644
--- a/server/jetstream_super_cluster_test.go
+++ b/server/jetstream_super_cluster_test.go
@@ -4047,3 +4047,87 @@ func TestJetStreamSuperClusterR1StreamPeerRemove(t *testing.T) {
 	_, err = js.StreamInfo("TEST")
 	require_NoError(t, err)
 }
+
+func TestJetStreamSuperClusterConsumerPauseAdvisories(t *testing.T) {
+	sc := createJetStreamSuperCluster(t, 3, 3)
+	defer sc.shutdown()
+
+	nc, js := jsClientConnect(t, sc.randomServer())
+	defer nc.Close()
+
+	pauseReq := func(consumer string, deadline time.Time) time.Time {
+		j, err := json.Marshal(JSApiConsumerPauseRequest{
+			PauseUntil: deadline,
+		})
+		require_NoError(t, err)
+		msg, err := nc.Request(fmt.Sprintf(JSApiConsumerPauseT, "TEST", consumer), j, time.Second)
+		require_NoError(t, err)
+		var res JSApiConsumerPauseResponse
+		err = json.Unmarshal(msg.Data, &res)
+		require_NoError(t, err)
+		return res.PauseUntil
+	}
+
+	checkAdvisory := func(msg *nats.Msg, shouldBePaused bool, deadline time.Time) {
+		t.Helper()
+		var advisory JSConsumerPauseAdvisory
+		require_NoError(t, json.Unmarshal(msg.Data, &advisory))
+		require_Equal(t, advisory.Stream, "TEST")
+		require_Equal(t, advisory.Consumer, "my_consumer")
+		require_Equal(t, advisory.Paused, shouldBePaused)
+		require_True(t, advisory.PauseUntil.Equal(deadline))
+	}
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	ch := make(chan *nats.Msg, 10)
+	_, err = nc.ChanSubscribe(JSAdvisoryConsumerPausePre+".TEST.my_consumer", ch)
+	require_NoError(t, err)
+
+	deadline := time.Now().Add(time.Second)
+	jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+		Name:       "my_consumer",
+		PauseUntil: &deadline,
+		Replicas:   3,
+	})
+
+	// First advisory should tell us that the consumer was paused
+	// on creation.
+	msg := require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, true, deadline)
+	require_Len(t, len(ch), 0) // Should only receive one advisory.
+
+	// The second one for the unpause.
+	msg = require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, false, deadline)
+	require_Len(t, len(ch), 0) // Should only receive one advisory.
+
+	// Now we'll pause the consumer for a second using the API.
+	deadline = time.Now().Add(time.Second)
+	require_True(t, pauseReq("my_consumer", deadline).Equal(deadline))
+
+	// Third advisory should tell us about the pause via the API.
+	msg = require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, true, deadline)
+	require_Len(t, len(ch), 0) // Should only receive one advisory.
+
+	// Finally that should unpause.
+	msg = require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, false, deadline)
+	require_Len(t, len(ch), 0) // Should only receive one advisory.
+
+	// Now we're going to set the deadline into the future so we can
+	// see what happens when we kick leaders or restart.
+	deadline = time.Now().Add(time.Hour)
+	require_True(t, pauseReq("my_consumer", deadline).Equal(deadline))
+
+	// Setting the deadline should have generated an advisory.
+	msg = require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, true, deadline)
+	require_Len(t, len(ch), 0) // Should only receive one advisory.
+}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 98d234f7..89dbedec 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -11437,7 +11437,7 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 	}
 
 	// Faster timeout since we loop below checking for condition.
-	js2, err := nc.JetStream(nats.MaxWait(250 * time.Millisecond))
+	js2, err := nc.JetStream(nats.MaxWait(500 * time.Millisecond))
 	require_NoError(t, err)
 
 	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
@@ -11605,25 +11605,25 @@ func TestJetStreamMirrorBasics(t *testing.T) {
 			si, err := js2.StreamInfo(streamName, &nats.StreamInfoRequest{SubjectsFilter: subject})
 			require_NoError(t, err)
 			if ss, ok := si.State.Subjects[subject]; !ok {
-				t.Logf("Expected messages with the transformed subject %s", subject)
+				return fmt.Errorf("expected messages with the transformed subject %s", subject)
 			} else {
 				if ss != subjectNumMsgs {
-					t.Fatalf("Expected %d messages on the transformed subject %s but got %d", subjectNumMsgs, subject, ss)
+					return fmt.Errorf("expected %d messages on the transformed subject %s but got %d", subjectNumMsgs, subject, ss)
 				}
 			}
 			if si.State.Msgs != streamNumMsg {
-				return fmt.Errorf("Expected %d stream messages, got state: %+v", streamNumMsg, si.State)
+				return fmt.Errorf("expected %d stream messages, got state: %+v", streamNumMsg, si.State)
 			}
 			if si.State.FirstSeq != firstSeq || si.State.LastSeq != lastSeq {
-				return fmt.Errorf("Expected first sequence=%d and last sequence=%d, but got state: %+v", firstSeq, lastSeq, si.State)
+				return fmt.Errorf("expected first sequence=%d and last sequence=%d, but got state: %+v", firstSeq, lastSeq, si.State)
 			}
 			return nil
 		}
 	}
 
-	checkFor(t, 10*time.Second, 100*time.Millisecond, f("M5", "foo2", 100, 100, 251, 350))
-	checkFor(t, 10*time.Second, 100*time.Millisecond, f("M6", "bar2", 50, 150, 101, 250))
-	checkFor(t, 10*time.Second, 100*time.Millisecond, f("M6", "baz2", 100, 150, 101, 250))
+	checkFor(t, 10*time.Second, 500*time.Millisecond, f("M5", "foo2", 100, 100, 251, 350))
+	checkFor(t, 10*time.Second, 500*time.Millisecond, f("M6", "bar2", 50, 150, 101, 250))
+	checkFor(t, 10*time.Second, 500*time.Millisecond, f("M6", "baz2", 100, 150, 101, 250))
 
 }
 
@@ -13534,6 +13534,13 @@ func TestJetStreamPurgeAndFilteredConsumers(t *testing.T) {
 		t.Fatalf("Expected NumPending to be 10, got %d", ci.NumPending)
 	}
 
+	// Also check unfiltered with interleaving messages.
+	_, err = js.AddConsumer("S", &nats.ConsumerConfig{
+		Durable:   "all",
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+
 	// Now purge only adam.
 	jr, _ := json.Marshal(&JSApiStreamPurgeRequest{Subject: "FOO.adam"})
 	_, err = nc.Request(fmt.Sprintf(JSApiStreamPurgeT, "S"), jr, time.Second)
@@ -13559,6 +13566,12 @@ func TestJetStreamPurgeAndFilteredConsumers(t *testing.T) {
 	if ci.AckFloor.Stream != 20 {
 		t.Fatalf("Expected AckFloor for stream to be 20, got %d", ci.AckFloor.Stream)
 	}
+
+	ci, err = js.ConsumerInfo("S", "all")
+	require_NoError(t, err)
+	if ci.NumPending != 10 {
+		t.Fatalf("Expected NumPending to be 10, got %d", ci.NumPending)
+	}
 }
 
 // Issue #2662
@@ -22325,3 +22338,405 @@ func TestJetStreamConsumerNakThenAckFloorMove(t *testing.T) {
 	require_Equal(t, ci.AckFloor.Stream, 11)
 	require_Equal(t, ci.NumAckPending, 0)
 }
+
+func TestJetStreamSubjectFilteredPurgeClearsPendingAcks(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo", "bar"},
+	})
+	require_NoError(t, err)
+
+	for i := 0; i < 5; i++ {
+		js.Publish("foo", []byte("OK"))
+		js.Publish("bar", []byte("OK"))
+	}
+
+	// Note that there are no subject filters here, this is deliberate
+	// as previously the purge with filter code was checking for them.
+	// We want to prove that unfiltered consumers also get purged.
+	ci, err := js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Name:          "my_consumer",
+		AckPolicy:     nats.AckExplicitPolicy,
+		MaxAckPending: 10,
+	})
+	require_NoError(t, err)
+	require_Equal(t, ci.NumPending, 10)
+	require_Equal(t, ci.NumAckPending, 0)
+
+	sub, err := js.PullSubscribe(">", "", nats.Bind("TEST", "my_consumer"))
+	require_NoError(t, err)
+
+	msgs, err := sub.Fetch(10)
+	require_NoError(t, err)
+	require_Len(t, len(msgs), 10)
+
+	ci, err = js.ConsumerInfo("TEST", "my_consumer")
+	require_NoError(t, err)
+	require_Equal(t, ci.NumPending, 0)
+	require_Equal(t, ci.NumAckPending, 10)
+
+	require_NoError(t, js.PurgeStream("TEST", &nats.StreamPurgeRequest{
+		Subject: "foo",
+	}))
+
+	ci, err = js.ConsumerInfo("TEST", "my_consumer")
+	require_NoError(t, err)
+	require_Equal(t, ci.NumPending, 0)
+	require_Equal(t, ci.NumAckPending, 5)
+
+	for i := 0; i < 5; i++ {
+		js.Publish("foo", []byte("OK"))
+	}
+	msgs, err = sub.Fetch(5)
+	require_NoError(t, err)
+	require_Len(t, len(msgs), 5)
+
+	ci, err = js.ConsumerInfo("TEST", "my_consumer")
+	require_NoError(t, err)
+	require_Equal(t, ci.NumPending, 0)
+	require_Equal(t, ci.NumAckPending, 10)
+}
+
+// Helper function for TestJetStreamConsumerPause*, TestJetStreamClusterConsumerPause*, TestJetStreamSuperClusterConsumerPause*
+func jsTestPause_CreateOrUpdateConsumer(t *testing.T, nc *nats.Conn, action ConsumerAction, stream string, cc ConsumerConfig) *JSApiConsumerCreateResponse {
+	t.Helper()
+	j, err := json.Marshal(CreateConsumerRequest{
+		Stream: stream,
+		Config: cc,
+		Action: action,
+	})
+	require_NoError(t, err)
+	subj := fmt.Sprintf("$JS.API.CONSUMER.CREATE.%s.%s", stream, cc.Name)
+	m, err := nc.Request(subj, j, time.Second*3)
+	require_NoError(t, err)
+	var res JSApiConsumerCreateResponse
+	require_NoError(t, json.Unmarshal(m.Data, &res))
+	require_True(t, res.Config != nil)
+	return &res
+}
+
+// Helper function for TestJetStreamConsumerPause*, TestJetStreamClusterConsumerPause*, TestJetStreamSuperClusterConsumerPause*
+func jsTestPause_PauseConsumer(t *testing.T, nc *nats.Conn, stream, consumer string, deadline time.Time) time.Time {
+	t.Helper()
+	j, err := json.Marshal(JSApiConsumerPauseRequest{
+		PauseUntil: deadline,
+	})
+	require_NoError(t, err)
+	subj := fmt.Sprintf("$JS.API.CONSUMER.PAUSE.%s.%s", stream, consumer)
+	msg, err := nc.Request(subj, j, time.Second)
+	require_NoError(t, err)
+	var res JSApiConsumerPauseResponse
+	require_NoError(t, json.Unmarshal(msg.Data, &res))
+	return res.PauseUntil
+}
+
+func TestJetStreamConsumerPauseViaConfig(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	t.Run("CreateShouldSucceed", func(t *testing.T) {
+		deadline := time.Now().Add(time.Hour)
+		ci := jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+			Name:       "my_consumer_1",
+			PauseUntil: &deadline,
+		})
+		require_True(t, ci != nil)
+		require_True(t, ci.Config != nil)
+		require_True(t, ci.Config.PauseUntil != nil)
+		require_True(t, ci.Config.PauseUntil.Equal(deadline))
+	})
+
+	t.Run("UpdateShouldFail", func(t *testing.T) {
+		deadline := time.Now().Add(time.Hour)
+		ci := jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+			Name: "my_consumer_2",
+		})
+		require_True(t, ci != nil)
+		require_True(t, ci.Config != nil)
+		require_True(t, ci.Config.PauseUntil == nil || ci.Config.PauseUntil.IsZero())
+
+		var cc ConsumerConfig
+		j, err := json.Marshal(ci.Config)
+		require_NoError(t, err)
+		require_NoError(t, json.Unmarshal(j, &cc))
+
+		pauseUntil := time.Now().Add(time.Hour)
+		cc.PauseUntil = &pauseUntil
+		ci2 := jsTestPause_CreateOrUpdateConsumer(t, nc, ActionUpdate, "TEST", cc)
+		require_False(t, ci2.Config.PauseUntil != nil && ci2.Config.PauseUntil.Equal(deadline))
+		require_True(t, ci2.Config.PauseUntil == nil || ci2.Config.PauseUntil.Equal(time.Time{}))
+	})
+}
+
+func TestJetStreamConsumerPauseViaEndpoint(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"push", "pull"},
+	})
+	require_NoError(t, err)
+
+	t.Run("PullConsumer", func(t *testing.T) {
+		_, err := js.AddConsumer("TEST", &nats.ConsumerConfig{
+			Name: "pull_consumer",
+		})
+		require_NoError(t, err)
+
+		sub, err := js.PullSubscribe("pull", "", nats.Bind("TEST", "pull_consumer"))
+		require_NoError(t, err)
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("pull", []byte("OK"))
+			require_NoError(t, err)
+		}
+		msgs, err := sub.Fetch(10, nats.MaxWait(time.Second))
+		require_NoError(t, err)
+		require_Equal(t, len(msgs), 10)
+
+		// Now we'll pause the consumer for 3 seconds.
+		deadline := time.Now().Add(time.Second * 3)
+		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "pull_consumer", deadline).Equal(deadline))
+
+		// This should fail as we'll wait for only half of the deadline.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("pull", []byte("OK"))
+			require_NoError(t, err)
+		}
+		_, err = sub.Fetch(10, nats.MaxWait(time.Until(deadline)/2))
+		require_Error(t, err, nats.ErrTimeout)
+
+		// This should succeed after a short wait, and when we're done,
+		// we should be after the deadline.
+		msgs, err = sub.Fetch(10)
+		require_NoError(t, err)
+		require_Equal(t, len(msgs), 10)
+		require_True(t, time.Now().After(deadline))
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("pull", []byte("OK"))
+			require_NoError(t, err)
+		}
+		msgs, err = sub.Fetch(10, nats.MaxWait(time.Second))
+		require_NoError(t, err)
+		require_Equal(t, len(msgs), 10)
+
+		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "pull_consumer", time.Time{}).Equal(time.Time{}))
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("pull", []byte("OK"))
+			require_NoError(t, err)
+		}
+		msgs, err = sub.Fetch(10, nats.MaxWait(time.Second))
+		require_NoError(t, err)
+		require_Equal(t, len(msgs), 10)
+	})
+
+	t.Run("PushConsumer", func(t *testing.T) {
+		ch := make(chan *nats.Msg, 100)
+		_, err = js.ChanSubscribe("push", ch, nats.BindStream("TEST"), nats.ConsumerName("push_consumer"))
+		require_NoError(t, err)
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("push", []byte("OK"))
+			require_NoError(t, err)
+		}
+		for i := 0; i < 10; i++ {
+			msg := require_ChanRead(t, ch, time.Second)
+			require_NotEqual(t, msg, nil)
+		}
+
+		// Now we'll pause the consumer for 3 seconds.
+		deadline := time.Now().Add(time.Second * 3)
+		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "push_consumer", deadline).Equal(deadline))
+
+		// This should succeed after a short wait, and when we're done,
+		// we should be after the deadline.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("push", []byte("OK"))
+			require_NoError(t, err)
+		}
+		for i := 0; i < 10; i++ {
+			msg := require_ChanRead(t, ch, time.Second*5)
+			require_NotEqual(t, msg, nil)
+			require_True(t, time.Now().After(deadline))
+		}
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("push", []byte("OK"))
+			require_NoError(t, err)
+		}
+		for i := 0; i < 10; i++ {
+			msg := require_ChanRead(t, ch, time.Second)
+			require_NotEqual(t, msg, nil)
+		}
+
+		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "push_consumer", time.Time{}).Equal(time.Time{}))
+
+		// This should succeed as there's no pause, so it definitely
+		// shouldn't take more than a second.
+		for i := 0; i < 10; i++ {
+			_, err = js.Publish("push", []byte("OK"))
+			require_NoError(t, err)
+		}
+		for i := 0; i < 10; i++ {
+			msg := require_ChanRead(t, ch, time.Second)
+			require_NotEqual(t, msg, nil)
+		}
+	})
+}
+
+func TestJetStreamConsumerPauseHeartbeats(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	deadline := time.Now().Add(time.Hour)
+	dsubj := "deliver_subj"
+
+	ci := jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+		Name:           "my_consumer",
+		PauseUntil:     &deadline,
+		Heartbeat:      time.Millisecond * 100,
+		DeliverSubject: dsubj,
+	})
+	require_True(t, ci.Config.PauseUntil.Equal(deadline))
+
+	ch := make(chan *nats.Msg, 10)
+	_, err = nc.ChanSubscribe(dsubj, ch)
+	require_NoError(t, err)
+
+	for i := 0; i < 20; i++ {
+		msg := require_ChanRead(t, ch, time.Millisecond*200)
+		require_Equal(t, msg.Header.Get("Status"), "100")
+		require_Equal(t, msg.Header.Get("Description"), "Idle Heartbeat")
+	}
+}
+
+func TestJetStreamConsumerPauseAdvisories(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	checkAdvisory := func(msg *nats.Msg, shouldBePaused bool, deadline time.Time) {
+		t.Helper()
+		var advisory JSConsumerPauseAdvisory
+		require_NoError(t, json.Unmarshal(msg.Data, &advisory))
+		require_Equal(t, advisory.Stream, "TEST")
+		require_Equal(t, advisory.Consumer, "my_consumer")
+		require_Equal(t, advisory.Paused, shouldBePaused)
+		require_True(t, advisory.PauseUntil.Equal(deadline))
+	}
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	ch := make(chan *nats.Msg, 10)
+	_, err = nc.ChanSubscribe(JSAdvisoryConsumerPausePre+".TEST.my_consumer", ch)
+	require_NoError(t, err)
+
+	deadline := time.Now().Add(time.Second)
+	jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+		Name:       "my_consumer",
+		PauseUntil: &deadline,
+	})
+
+	// First advisory should tell us that the consumer was paused
+	// on creation.
+	msg := require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, true, deadline)
+
+	// The second one for the unpause.
+	msg = require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, false, deadline)
+
+	// Now we'll pause the consumer using the API.
+	deadline = time.Now().Add(time.Second)
+	require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", deadline).Equal(deadline))
+
+	// Third advisory should tell us about the pause via the API.
+	msg = require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, true, deadline)
+
+	// Finally that should unpause.
+	msg = require_ChanRead(t, ch, time.Second*2)
+	checkAdvisory(msg, false, deadline)
+}
+
+func TestJetStreamConsumerSurvivesRestart(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	deadline := time.Now().Add(time.Hour)
+	jsTestPause_CreateOrUpdateConsumer(t, nc, ActionCreate, "TEST", ConsumerConfig{
+		Name:       "my_consumer",
+		PauseUntil: &deadline,
+	})
+
+	sd := s.JetStreamConfig().StoreDir
+	s.Shutdown()
+	s = RunJetStreamServerOnPort(-1, sd)
+	defer s.Shutdown()
+
+	stream, err := s.gacc.lookupStream("TEST")
+	require_NoError(t, err)
+
+	consumer := stream.lookupConsumer("my_consumer")
+	require_NotEqual(t, consumer, nil)
+
+	consumer.mu.RLock()
+	timer := consumer.uptmr
+	consumer.mu.RUnlock()
+	require_True(t, timer != nil)
+}
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index 038e8674..6c8f9bb1 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -837,6 +837,7 @@ func TestLeafNodeLoop(t *testing.T) {
 		// This test requires that we set the port to known value because
 		// we want A point to B and B to A.
 		oa := DefaultOptions()
+		oa.ServerName = "A"
 		if !cluster {
 			oa.Cluster.Port = 0
 			oa.Cluster.Name = _EMPTY_
@@ -849,10 +850,11 @@ func TestLeafNodeLoop(t *testing.T) {
 		sa := RunServer(oa)
 		defer sa.Shutdown()
 
-		l := &loopDetectedLogger{ch: make(chan string, 1)}
-		sa.SetLogger(l, false, false)
+		la := &loopDetectedLogger{ch: make(chan string, 1)}
+		sa.SetLogger(la, false, false)
 
 		ob := DefaultOptions()
+		ob.ServerName = "B"
 		if !cluster {
 			ob.Cluster.Port = 0
 			ob.Cluster.Name = _EMPTY_
@@ -867,10 +869,15 @@ func TestLeafNodeLoop(t *testing.T) {
 		sb := RunServer(ob)
 		defer sb.Shutdown()
 
+		lb := &loopDetectedLogger{ch: make(chan string, 1)}
+		sb.SetLogger(lb, false, false)
+
 		select {
-		case <-l.ch:
+		case <-la.ch:
 			// OK!
-		case <-time.After(2 * time.Second):
+		case <-lb.ch:
+			// OK!
+		case <-time.After(5 * time.Second):
 			t.Fatalf("Did not get any error regarding loop")
 		}
 
@@ -892,6 +899,7 @@ func TestLeafNodeLoopFromDAG(t *testing.T) {
 	// We need to cancel clustering since now this will suppress on its own.
 	oa := DefaultOptions()
 	oa.ServerName = "A"
+	oa.LeafNode.connDelay = 50 * time.Millisecond
 	oa.LeafNode.ReconnectInterval = 10 * time.Millisecond
 	oa.LeafNode.Port = -1
 	oa.Cluster = ClusterOpts{}
@@ -903,6 +911,7 @@ func TestLeafNodeLoopFromDAG(t *testing.T) {
 	// B will point to A
 	ob := DefaultOptions()
 	ob.ServerName = "B"
+	ob.LeafNode.connDelay = 50 * time.Millisecond
 	ob.LeafNode.ReconnectInterval = 10 * time.Millisecond
 	ob.LeafNode.Port = -1
 	ob.LeafNode.Remotes = []*RemoteLeafOpts{{URLs: []*url.URL{ua}}}
@@ -918,6 +927,7 @@ func TestLeafNodeLoopFromDAG(t *testing.T) {
 	// C will point to A and B
 	oc := DefaultOptions()
 	oc.ServerName = "C"
+	oc.LeafNode.connDelay = 50 * time.Millisecond
 	oc.LeafNode.ReconnectInterval = 10 * time.Millisecond
 	oc.LeafNode.Remotes = []*RemoteLeafOpts{{URLs: []*url.URL{ua}}, {URLs: []*url.URL{ub}}}
 	oc.LeafNode.connDelay = 100 * time.Millisecond // Allow logger to be attached before connecting.
@@ -944,6 +954,7 @@ func TestLeafNodeLoopFromDAG(t *testing.T) {
 	// Shutdown C and restart without the loop.
 	sc.Shutdown()
 	oc.LeafNode.Remotes = []*RemoteLeafOpts{{URLs: []*url.URL{ub}}}
+
 	sc = RunServer(oc)
 	defer sc.Shutdown()
 
@@ -7398,6 +7409,63 @@ func TestLeafNodeServerReloadSubjectMappings(t *testing.T) {
 	checkSubNoInterest(t, l, globalAccountName, "source1", 2*time.Second)
 }
 
+// https://github.com/nats-io/nats-server/issues/5099
+func TestLeafNodeServerReloadSubjectMappingsWithSameSubject(t *testing.T) {
+	stmpl := `
+		listen: 127.0.0.1:-1
+		server_name: test-server
+		mappings = { "source": "target1" }
+		leaf { listen: 127.0.0.1:-1 }
+	`
+	conf := createConfFile(t, []byte(stmpl))
+	s, o := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	tmpl := `
+		listen: 127.0.0.1:-1
+		server_name: test-leaf
+		leaf {
+			remotes: [ { urls: [ nats-leaf://127.0.0.1:{LEAF_PORT} ] } ]
+		}
+	`
+	tmpl = strings.Replace(tmpl, "{LEAF_PORT}", fmt.Sprintf("%d", o.LeafNode.Port), 1)
+	lConf := createConfFile(t, []byte(tmpl))
+	l, _ := RunServerWithConfig(lConf)
+	defer l.Shutdown()
+
+	checkLeafNodeConnected(t, l)
+
+	// Create our subscriber.
+	nc := natsConnect(t, s.ClientURL())
+	defer nc.Close()
+	sub1 := natsSubSync(t, nc, "target1")
+	sub2 := natsSubSync(t, nc, "target2")
+	natsFlush(t, nc)
+
+	// Create our publisher.
+	ncl := natsConnect(t, l.ClientURL())
+	defer ncl.Close()
+	// Publish our message.
+	ncl.Publish("source", []byte("OK"))
+
+	// Make sure we receive it.
+	checkSubsPending(t, sub1, 1)
+	// Make sure the other does not.
+	checkSubsPending(t, sub2, 0)
+
+	// Now change mapping, but only the "to" subject, keeping same "from"
+	reloadUpdateConfig(t, s, conf, strings.Replace(stmpl, "target1", "target2", 1))
+	checkLeafNodeConnected(t, l)
+
+	// Publish our new message.
+	ncl.Publish("source", []byte("OK"))
+
+	// Make sure we receive it.
+	checkSubsPending(t, sub2, 1)
+	// Make sure the other does not.
+	checkSubsPending(t, sub1, 1)
+}
+
 func TestLeafNodeNkeyAuth(t *testing.T) {
 	conf := createConfFile(t, []byte(`
 		listen: 127.0.0.1:-1
@@ -7459,3 +7527,83 @@ func TestLeafNodeAccountNkeysAuth(t *testing.T) {
 
 	checkLeafNodeConnected(t, l)
 }
+
+// https://github.com/nats-io/nats-server/issues/5117
+func TestLeafNodeLoopDetectionOnActualLoop(t *testing.T) {
+	// Setup:  B --[leaf]--> A    C --[leaf]--> A    C --[leaf] --> B
+	accConf := `
+		accounts: {
+			APP: {
+				users: [ { user:u, password: u,
+					permissions: { publish = "u.>", subscribe = "u.>" }} ]
+			}
+			$SYS: { users = [ {user: "s", password: "s"} ] }
+		}`
+
+	confA := createConfFile(t, []byte(fmt.Sprintf(`
+		server_name: a1
+		port: -1
+		cluster: { name: A }
+		leafnodes {
+			port: 17422
+		}
+		%s`, accConf)))
+
+	confB := createConfFile(t, []byte(fmt.Sprintf(`
+		server_name: b1
+		port: -1
+		cluster: { name: B }
+		leafnodes {
+			port: 17432
+			remotes [
+				{ urls: ["nats-leaf://u:u@localhost:17422"], account: "APP" }
+			]
+			reconnect: "2s"
+		}
+		%s`, accConf)))
+
+	confC := createConfFile(t, []byte(fmt.Sprintf(`
+		server_name: c1
+		port: -1
+		cluster: { name: C }
+		leafnodes {
+			port: 17442
+			remotes [
+				{ urls: ["nats-leaf://u:u@localhost:17422"], account: "APP" }
+				# This one creates the loop
+				{ urls: ["nats-leaf://u:u@localhost:17432"], account: "APP" }
+			]
+			reconnect: "0.5s"
+		}
+		%s`, accConf)))
+
+	// Start order will be B -> C -> A
+	// We will force C to connect to A first before B using different reconnect intervals.
+	// If B connects first we detect loops fine. If C connects first we do not.
+
+	srvB, _ := RunServerWithConfig(confB)
+	defer srvB.Shutdown()
+	lb := &loopDetectedLogger{ch: make(chan string, 1)}
+	srvB.SetLogger(lb, false, false)
+
+	srvC, _ := RunServerWithConfig(confC)
+	defer srvC.Shutdown()
+	lc := &loopDetectedLogger{ch: make(chan string, 1)}
+	srvC.SetLogger(lc, false, false)
+
+	// C should connect to B
+	checkLeafNodeConnectedCount(t, srvC, 1)
+
+	srvA, _ := RunServerWithConfig(confA)
+	defer srvA.Shutdown()
+	la := &loopDetectedLogger{ch: make(chan string, 1)}
+	srvA.SetLogger(la, false, false)
+
+	select {
+	case <-la.ch:
+	case <-lb.ch:
+	case <-lc.ch:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not get any error regarding loop")
+	}
+}
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index 0a510450..4358826e 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -4262,6 +4262,14 @@ func TestMQTTPublishRetain(t *testing.T) {
 	s := testMQTTRunServer(t, o)
 	defer testMQTTShutdownServer(s)
 
+	buf := strings.Builder{}
+	for i := 0; i < 128; i++ { // 128Kb
+		for j := 0; j < 64; j++ { // 16 * 64 = 1Kb
+			buf.Write([]byte("0123456789abcdef"))
+		}
+	}
+	large := buf.String()
+
 	for _, test := range []struct {
 		name          string
 		retained      bool
@@ -4269,6 +4277,7 @@ func TestMQTTPublishRetain(t *testing.T) {
 		expectedValue string
 		subGetsIt     bool
 	}{
+		{"publish large retained", true, large, large, true},
 		{"publish retained", true, "retained", "retained", true},
 		{"publish not retained", false, "not retained", "retained", true},
 		{"remove retained", true, "", "", false},
@@ -7508,8 +7517,16 @@ func TestMQTTDecodeRetainedMessage(t *testing.T) {
 	defer mc.Close()
 	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
 	testMQTTSub(t, 1, mc, r, []*mqttFilter{{filter: "foo/+", qos: 0}}, []byte{0})
-	testMQTTCheckPubMsg(t, mc, r, "foo/1", mqttPubFlagRetain, []byte("msg1"))
-	testMQTTCheckPubMsg(t, mc, r, "foo/2", mqttPubFlagRetain, []byte("msg2"))
+	for i := 0; i < 2; i++ {
+		b, pl := testMQTTReadPacket(t, r)
+		if pt := b & mqttPacketMask; pt != mqttPacketPub {
+			t.Fatalf("Expected PUBLISH packet %x, got %x", mqttPacketPub, pt)
+		}
+		_, _, topic := testMQTTGetPubMsgExEx(t, mc, r, mqttPubFlagRetain, pl, "", nil)
+		if string(topic) != "foo/1" && string(topic) != "foo/2" {
+			t.Fatalf("Expected foo/1 or foo/2, got %q", topic)
+		}
+	}
 	testMQTTExpectNothing(t, r)
 	mc.Close()
 
diff --git a/server/msgtrace_test.go b/server/msgtrace_test.go
index 37d79dae..a5d001e5 100644
--- a/server/msgtrace_test.go
+++ b/server/msgtrace_test.go
@@ -26,7 +26,9 @@ import (
 	"time"
 
 	"github.com/klauspost/compress/s2"
+	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nkeys"
 )
 
 func init() {
@@ -1697,71 +1699,74 @@ func TestMsgTraceWithGatewayToOldServer(t *testing.T) {
 }
 
 func TestMsgTraceServiceImport(t *testing.T) {
-	for _, mainTest := range []struct {
-		name  string
-		allow bool
-	}{
-		{"allowed", true},
-		{"not allowed", false},
-	} {
-		t.Run(mainTest.name, func(t *testing.T) {
-			conf := createConfFile(t, []byte(fmt.Sprintf(`
-			listen: 127.0.0.1:-1
-			accounts {
-				A {
-					users: [{user: a, password: pwd}]
-					exports: [ { service: ">", allow_trace: %v} ]
-					mappings = {
-						bar: bozo
-					}
-				}
-				B {
-					users: [{user: b, password: pwd}]
-					imports: [ { service: {account: "A", subject:">"} } ]
-					exports: [ { service: ">", allow_trace: %v} ]
-				}
-				C {
-					users: [{user: c, password: pwd}]
-					exports: [ { service: ">", allow_trace: %v } ]
+	tmpl := `
+		listen: 127.0.0.1:-1
+		accounts {
+			A {
+				users: [{user: a, password: pwd}]
+				exports: [ { service: ">", allow_trace: %v} ]
+				mappings = {
+					bar: bozo
 				}
-				D {
-					users: [{user: d, password: pwd}]
-					imports: [
-						{ service: {account: "B", subject:"bar"}, to: baz }
-						{ service: {account: "C", subject:">"} }
-					]
-					mappings = {
-							bat: baz
-					}
+			}
+			B {
+				users: [{user: b, password: pwd}]
+				imports: [ { service: {account: "A", subject:">"} } ]
+				exports: [ { service: ">", allow_trace: %v} ]
+			}
+			C {
+				users: [{user: c, password: pwd}]
+				exports: [ { service: ">", allow_trace: %v } ]
+			}
+			D {
+				users: [{user: d, password: pwd}]
+				imports: [
+					{ service: {account: "B", subject:"bar"}, to: baz }
+					{ service: {account: "C", subject:">"} }
+				]
+				mappings = {
+						bat: baz
 				}
 			}
-		`, mainTest.allow, mainTest.allow, mainTest.allow)))
-			s, _ := RunServerWithConfig(conf)
-			defer s.Shutdown()
+		}
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, false, false, false)))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
 
-			nc := natsConnect(t, s.ClientURL(), nats.UserInfo("d", "pwd"), nats.Name("Requestor"))
-			defer nc.Close()
+	nc := natsConnect(t, s.ClientURL(), nats.UserInfo("d", "pwd"), nats.Name("Requestor"))
+	defer nc.Close()
 
-			traceSub := natsSubSync(t, nc, "my.trace.subj")
-			sub := natsSubSync(t, nc, "my.service.response.inbox")
+	traceSub := natsSubSync(t, nc, "my.trace.subj")
+	sub := natsSubSync(t, nc, "my.service.response.inbox")
 
-			nc2 := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("ServiceA"))
-			defer nc2.Close()
-			recv := int32(0)
-			natsQueueSub(t, nc2, "*", "my_queue", func(m *nats.Msg) {
-				atomic.AddInt32(&recv, 1)
-				m.Respond(m.Data)
-			})
-			natsFlush(t, nc2)
+	nc2 := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("ServiceA"))
+	defer nc2.Close()
+	recv := int32(0)
+	natsQueueSub(t, nc2, "*", "my_queue", func(m *nats.Msg) {
+		atomic.AddInt32(&recv, 1)
+		m.Respond(m.Data)
+	})
+	natsFlush(t, nc2)
 
-			nc3 := natsConnect(t, s.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("ServiceC"))
-			defer nc3.Close()
-			natsSub(t, nc3, "baz", func(m *nats.Msg) {
-				atomic.AddInt32(&recv, 1)
-				m.Respond(m.Data)
-			})
-			natsFlush(t, nc3)
+	nc3 := natsConnect(t, s.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("ServiceC"))
+	defer nc3.Close()
+	natsSub(t, nc3, "baz", func(m *nats.Msg) {
+		atomic.AddInt32(&recv, 1)
+		m.Respond(m.Data)
+	})
+	natsFlush(t, nc3)
 
+	for mainIter, mainTest := range []struct {
+		name  string
+		allow bool
+	}{
+		{"not allowed", false},
+		{"allowed", true},
+		{"not allowed again", false},
+	} {
+		atomic.StoreInt32(&recv, 0)
+		t.Run(mainTest.name, func(t *testing.T) {
 			for _, test := range []struct {
 				name       string
 				deliverMsg bool
@@ -1890,6 +1895,12 @@ func TestMsgTraceServiceImport(t *testing.T) {
 					}
 				})
 			}
+			switch mainIter {
+			case 0:
+				reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, true, true, true))
+			case 1:
+				reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, false, false, false))
+			}
 		})
 	}
 }
@@ -2596,50 +2607,52 @@ func TestMsgTraceServiceImportWithLeafNodeLeaf(t *testing.T) {
 }
 
 func TestMsgTraceStreamExport(t *testing.T) {
-	for _, mainTest := range []struct {
+	tmpl := `
+		listen: 127.0.0.1:-1
+		accounts {
+			A {
+				users: [{user: a, password: pwd}]
+				exports: [
+					{ stream: "info.*.*.>"}
+				]
+			}
+			B {
+				users: [{user: b, password: pwd}]
+				imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "B.info.$2.$1.>", allow_trace: %v } ]
+			}
+			C {
+				users: [{user: c, password: pwd}]
+				imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "C.info.$1.$2.>", allow_trace: %v } ]
+			}
+		}
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, false, false)))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("Tracer"))
+	defer nc.Close()
+	traceSub := natsSubSync(t, nc, "my.trace.subj")
+
+	nc2 := natsConnect(t, s.ClientURL(), nats.UserInfo("b", "pwd"), nats.Name("sub1"))
+	defer nc2.Close()
+	sub1 := natsSubSync(t, nc2, "B.info.*.*.>")
+	natsFlush(t, nc2)
+
+	nc3 := natsConnect(t, s.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("sub2"))
+	defer nc3.Close()
+	sub2 := natsQueueSubSync(t, nc3, "C.info.>", "my_queue")
+	natsFlush(t, nc3)
+
+	for mainIter, mainTest := range []struct {
 		name  string
 		allow bool
 	}{
-		{"allowed", true},
 		{"not allowed", false},
+		{"allowed", true},
+		{"not allowed again", false},
 	} {
 		t.Run(mainTest.name, func(t *testing.T) {
-			conf := createConfFile(t, []byte(fmt.Sprintf(`
-				listen: 127.0.0.1:-1
-				accounts {
-					A {
-						users: [{user: a, password: pwd}]
-						exports: [
-							{ stream: "info.*.*.>"}
-						]
-					}
-					B {
-						users: [{user: b, password: pwd}]
-						imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "B.info.$2.$1.>", allow_trace: %v } ]
-					}
-					C {
-						users: [{user: c, password: pwd}]
-						imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "C.info.$1.$2.>", allow_trace: %v } ]
-					}
-				}
-			`, mainTest.allow, mainTest.allow)))
-			s, _ := RunServerWithConfig(conf)
-			defer s.Shutdown()
-
-			nc := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("Tracer"))
-			defer nc.Close()
-			traceSub := natsSubSync(t, nc, "my.trace.subj")
-
-			nc2 := natsConnect(t, s.ClientURL(), nats.UserInfo("b", "pwd"), nats.Name("sub1"))
-			defer nc2.Close()
-			sub1 := natsSubSync(t, nc2, "B.info.*.*.>")
-			natsFlush(t, nc2)
-
-			nc3 := natsConnect(t, s.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("sub2"))
-			defer nc3.Close()
-			sub2 := natsQueueSubSync(t, nc3, "C.info.>", "my_queue")
-			natsFlush(t, nc3)
-
 			for _, test := range []struct {
 				name       string
 				deliverMsg bool
@@ -2720,6 +2733,12 @@ func TestMsgTraceStreamExport(t *testing.T) {
 					}
 				})
 			}
+			switch mainIter {
+			case 0:
+				reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, true, true))
+			case 1:
+				reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, false, false))
+			}
 		})
 	}
 }
@@ -3529,9 +3548,7 @@ func TestMsgTraceJetStreamWithSuperCluster(t *testing.T) {
 		for _, s := range cl.servers {
 			acc, err := s.LookupAccount(globalAccountName)
 			require_NoError(t, err)
-			acc.mu.Lock()
-			acc.TraceDest = traceDest
-			acc.mu.Unlock()
+			acc.setTraceDest(traceDest)
 		}
 	}
 
@@ -4598,32 +4615,499 @@ func TestMsgTraceTriggeredByExternalHeader(t *testing.T) {
 			}
 		})
 	}
+}
 
-	nc1.Close()
-	nc2.Close()
-	s2.Shutdown()
-	s1.Shutdown()
-	s1 = nil
+func TestMsgTraceAccountTraceDestJWTUpdate(t *testing.T) {
+	// create system account
+	sysKp, _ := nkeys.CreateAccount()
+	sysPub, _ := sysKp.PublicKey()
+	sysCreds := newUser(t, sysKp)
+	// create account A
+	akp, _ := nkeys.CreateAccount()
+	aPub, _ := akp.PublicKey()
+	claim := jwt.NewAccountClaims(aPub)
+	aJwt, err := claim.Encode(oKp)
+	require_NoError(t, err)
 
-	o1.Accounts = o1.Accounts[:1]
-	accName := o1.Accounts[0].Name
-	o1.Port, o1.Cluster.Port = -1, -1
+	dir := t.TempDir()
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: -1
+		operator: %s
+		resolver: {
+			type: full
+			dir: '%s'
+		}
+		system_account: %s
+    `, ojwt, dir, sysPub)))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+	updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
+
+	nc := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, akp))
+	defer nc.Close()
+
+	sub := natsSubSync(t, nc, "acc.trace.dest")
+	natsFlush(t, nc)
+
+	for i, test := range []struct {
+		name           string
+		traceTriggered bool
+	}{
+		{"no acc dest", false},
+		{"adding trace dest", true},
+		{"removing trace dest", false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			msg := nats.NewMsg("foo")
+			msg.Header.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
+			msg.Data = []byte("hello")
+			err = nc.PublishMsg(msg)
+			require_NoError(t, err)
 
+			if test.traceTriggered {
+				tm := natsNexMsg(t, sub, time.Second)
+				var e MsgTraceEvent
+				err = json.Unmarshal(tm.Data, &e)
+				require_NoError(t, err)
+				// Simple check
+				require_Equal[string](t, e.Server.Name, s.Name())
+			}
+			// No (more) trace message expected.
+			tm, err := sub.NextMsg(250 * time.Millisecond)
+			if err != nats.ErrTimeout {
+				t.Fatalf("Expected no trace message, got %s", tm.Data)
+			}
+			if i < 2 {
+				if i == 0 {
+					claim.Trace = &jwt.MsgTrace{Destination: "acc.trace.dest"}
+				} else {
+					claim.Trace = nil
+				}
+				aJwt, err = claim.Encode(oKp)
+				require_NoError(t, err)
+				updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
+			}
+		})
+	}
+}
+
+func TestMsgTraceServiceJWTUpdate(t *testing.T) {
+	// create system account
+	sysKp, _ := nkeys.CreateAccount()
+	sysPub, _ := sysKp.PublicKey()
+	sysCreds := newUser(t, sysKp)
+	// create account A
+	akp, _ := nkeys.CreateAccount()
+	aPub, _ := akp.PublicKey()
+	aClaim := jwt.NewAccountClaims(aPub)
+	serviceExport := &jwt.Export{Subject: "req", Type: jwt.Service}
+	aClaim.Exports.Add(serviceExport)
+	aJwt, err := aClaim.Encode(oKp)
+	require_NoError(t, err)
+	// create account B
+	bkp, _ := nkeys.CreateAccount()
+	bPub, _ := bkp.PublicKey()
+	bClaim := jwt.NewAccountClaims(bPub)
+	serviceImport := &jwt.Import{Account: aPub, Subject: "req", Type: jwt.Service}
+	bClaim.Imports.Add(serviceImport)
+	bJwt, err := bClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	dir := t.TempDir()
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: -1
+		operator: %s
+		resolver: {
+			type: full
+			dir: '%s'
+		}
+		system_account: %s
+	`, ojwt, dir, sysPub)))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+	updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
+	updateJwt(t, s.ClientURL(), sysCreds, bJwt, 1)
+
+	ncA := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, akp), nats.Name("Service"))
+	defer ncA.Close()
+
+	natsSub(t, ncA, "req", func(m *nats.Msg) {
+		m.Respond([]byte("resp"))
+	})
+	natsFlush(t, ncA)
+
+	ncB := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, bkp))
+	defer ncB.Close()
+
+	sub := natsSubSync(t, ncB, "trace.dest")
+	natsFlush(t, ncB)
+
+	for i, test := range []struct {
+		name       string
+		allowTrace bool
+	}{
+		{"trace not allowed", false},
+		{"trace allowed", true},
+		{"trace not allowed again", false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			msg := nats.NewMsg("req")
+			msg.Header.Set(MsgTraceDest, sub.Subject)
+			msg.Data = []byte("request")
+			reply, err := ncB.RequestMsg(msg, time.Second)
+			require_NoError(t, err)
+			require_Equal[string](t, string(reply.Data), "resp")
+
+			tm := natsNexMsg(t, sub, time.Second)
+			var e MsgTraceEvent
+			err = json.Unmarshal(tm.Data, &e)
+			require_NoError(t, err)
+			require_Equal[string](t, e.Server.Name, s.Name())
+			require_Equal[string](t, e.Ingress().Account, bPub)
+			sis := e.ServiceImports()
+			require_Equal[int](t, len(sis), 1)
+			si := sis[0]
+			require_Equal[string](t, si.Account, aPub)
+			egresses := e.Egresses()
+			if !test.allowTrace {
+				require_Equal[int](t, len(egresses), 0)
+			} else {
+				require_Equal[int](t, len(egresses), 1)
+				eg := egresses[0]
+				require_Equal[string](t, eg.Name, "Service")
+				require_Equal[string](t, eg.Account, aPub)
+				require_Equal[string](t, eg.Subscription, "req")
+			}
+			// No (more) trace message expected.
+			tm, err = sub.NextMsg(250 * time.Millisecond)
+			if err != nats.ErrTimeout {
+				t.Fatalf("Expected no trace message, got %s", tm.Data)
+			}
+			if i < 2 {
+				// Set AllowTrace to true at the first iteration, then
+				// false at the second.
+				aClaim.Exports[0].AllowTrace = (i == 0)
+				aJwt, err = aClaim.Encode(oKp)
+				require_NoError(t, err)
+				updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
+			}
+		})
+	}
+}
+
+func TestMsgTraceStreamJWTUpdate(t *testing.T) {
+	// create system account
+	sysKp, _ := nkeys.CreateAccount()
+	sysPub, _ := sysKp.PublicKey()
+	sysCreds := newUser(t, sysKp)
+	// create account A
+	akp, _ := nkeys.CreateAccount()
+	aPub, _ := akp.PublicKey()
+	aClaim := jwt.NewAccountClaims(aPub)
+	streamExport := &jwt.Export{Subject: "info", Type: jwt.Stream}
+	aClaim.Exports.Add(streamExport)
+	aJwt, err := aClaim.Encode(oKp)
+	require_NoError(t, err)
+	// create account B
+	bkp, _ := nkeys.CreateAccount()
+	bPub, _ := bkp.PublicKey()
+	bClaim := jwt.NewAccountClaims(bPub)
+	streamImport := &jwt.Import{Account: aPub, Subject: "info", To: "b", Type: jwt.Stream}
+	bClaim.Imports.Add(streamImport)
+	bJwt, err := bClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	dir := t.TempDir()
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: -1
+		operator: %s
+		resolver: {
+			type: full
+			dir: '%s'
+		}
+		system_account: %s
+	`, ojwt, dir, sysPub)))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+	updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
+	updateJwt(t, s.ClientURL(), sysCreds, bJwt, 1)
+
+	ncA := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, akp))
+	defer ncA.Close()
+
+	traceSub := natsSubSync(t, ncA, "trace.dest")
+	natsFlush(t, ncA)
+
+	ncB := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, bkp), nats.Name("BInfo"))
+	defer ncB.Close()
+
+	appSub := natsSubSync(t, ncB, "b.info")
+	natsFlush(t, ncB)
+
+	for i, test := range []struct {
+		name       string
+		allowTrace bool
+	}{
+		{"trace not allowed", false},
+		{"trace allowed", true},
+		{"trace not allowed again", false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			msg := nats.NewMsg("info")
+			msg.Header.Set(MsgTraceDest, traceSub.Subject)
+			msg.Data = []byte("some info")
+			err = ncA.PublishMsg(msg)
+			require_NoError(t, err)
+
+			appMsg := natsNexMsg(t, appSub, time.Second)
+			require_Equal[string](t, string(appMsg.Data), "some info")
+
+			tm := natsNexMsg(t, traceSub, time.Second)
+			var e MsgTraceEvent
+			err = json.Unmarshal(tm.Data, &e)
+			require_NoError(t, err)
+			require_Equal[string](t, e.Server.Name, s.Name())
+			ses := e.StreamExports()
+			require_Equal[int](t, len(ses), 1)
+			se := ses[0]
+			require_Equal[string](t, se.Account, bPub)
+			require_Equal[string](t, se.To, "b.info")
+			egresses := e.Egresses()
+			if !test.allowTrace {
+				require_Equal[int](t, len(egresses), 0)
+			} else {
+				require_Equal[int](t, len(egresses), 1)
+				eg := egresses[0]
+				require_Equal[string](t, eg.Name, "BInfo")
+				require_Equal[string](t, eg.Account, bPub)
+				require_Equal[string](t, eg.Subscription, "info")
+			}
+			// No (more) trace message expected.
+			tm, err = traceSub.NextMsg(250 * time.Millisecond)
+			if err != nats.ErrTimeout {
+				t.Fatalf("Expected no trace message, got %s", tm.Data)
+			}
+			if i < 2 {
+				// Set AllowTrace to true at the first iteration, then
+				// false at the second.
+				bClaim.Imports[0].AllowTrace = (i == 0)
+				bJwt, err = bClaim.Encode(oKp)
+				require_NoError(t, err)
+				updateJwt(t, s.ClientURL(), sysCreds, bJwt, 1)
+			}
+		})
+	}
+}
+
+func TestMsgTraceParseAccountDestWithSampling(t *testing.T) {
+	tmpl := `
+		port: -1
+		accounts {
+			A {
+				users: [{user: a, password: pwd}]
+				%s
+			}
+		}
+	`
 	for _, test := range []struct {
-		name      string
-		traceDest string
+		name        string
+		samplingStr string
+		want        int
 	}{
-		{"account trace invalid subject", "invalid..subject"},
-		{"account trace invalid publish subject", "invalid.publish.*.subject"},
+		{"trace sampling no dest", `msg_trace: {sampling: 50}`, 0},
+		{"trace dest only", `msg_trace: {dest: foo}`, 100},
+		{"trace dest with number only", `msg_trace: {dest: foo, sampling: 20}`, 20},
+		{"trace dest with percentage", `msg_trace: {dest: foo, sampling: 50%}`, 50},
 	} {
 		t.Run(test.name, func(t *testing.T) {
-			o1.Accounts[0].TraceDest = test.traceDest
-			s1, err := NewServer(o1)
-			if err == nil || !strings.Contains(err.Error(), fmt.Sprintf("trace_dest %q of account %q is not valid", test.traceDest, accName)) {
-				if s1 != nil {
-					s1.Shutdown()
+			conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, test.samplingStr)))
+			o := LoadConfig(conf)
+			_, sampling := o.Accounts[0].getTraceDestAndSampling()
+			require_Equal[int](t, test.want, sampling)
+		})
+	}
+}
+
+func TestMsgTraceAccountDestWithSampling(t *testing.T) {
+	tmpl := `
+		port: -1
+		server_name: %s
+		accounts {
+			A {
+				users: [{user: a, password:pwd}]
+				msg_trace: {dest: "acc.dest"%s}
+			}
+		}
+		cluster {
+			port: -1
+			%s
+		}
+	`
+	conf1 := createConfFile(t, []byte(fmt.Sprintf(tmpl, "A", _EMPTY_, _EMPTY_)))
+	s1, o1 := RunServerWithConfig(conf1)
+	defer s1.Shutdown()
+
+	routes := fmt.Sprintf("routes: [\"nats://127.0.0.1:%d\"]", o1.Cluster.Port)
+	conf2 := createConfFile(t, []byte(fmt.Sprintf(tmpl, "B", _EMPTY_, routes)))
+	s2, _ := RunServerWithConfig(conf2)
+	defer s2.Shutdown()
+
+	checkClusterFormed(t, s1, s2)
+
+	nc2 := natsConnect(t, s2.ClientURL(), nats.UserInfo("a", "pwd"))
+	defer nc2.Close()
+	natsSub(t, nc2, "foo", func(_ *nats.Msg) {})
+	natsFlush(t, nc2)
+
+	nc1 := natsConnect(t, s1.ClientURL(), nats.UserInfo("a", "pwd"))
+	defer nc1.Close()
+	sub := natsSubSync(t, nc1, "acc.dest")
+	natsFlush(t, nc1)
+
+	checkSubInterest(t, s1, "A", "foo", time.Second)
+	checkSubInterest(t, s2, "A", "acc.dest", time.Second)
+
+	for iter, test := range []struct {
+		name        string
+		samplingStr string
+		sampling    int
+	}{
+		// Sampling is considered 100% if not specified or <=0 or >= 100.
+		// To disable sampling, the account destination should not be specified.
+		{"no sampling specified", _EMPTY_, 100},
+		{"sampling specified", ", sampling: \"25%\"", 25},
+		{"no sampling again", _EMPTY_, 100},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			if iter > 0 {
+				reloadUpdateConfig(t, s1, conf1, fmt.Sprintf(tmpl, "A", test.samplingStr, _EMPTY_))
+				reloadUpdateConfig(t, s2, conf2, fmt.Sprintf(tmpl, "B", test.samplingStr, routes))
+			}
+			msg := nats.NewMsg("foo")
+			msg.Header.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
+			total := 400
+			for i := 0; i < total; i++ {
+				err := nc1.PublishMsg(msg)
+				require_NoError(t, err)
+			}
+			// Wait a bit to make sure that we received all traces that should
+			// have been received.
+			time.Sleep(500 * time.Millisecond)
+			n, _, err := sub.Pending()
+			require_NoError(t, err)
+			fromClient := 0
+			fromRoute := 0
+			for i := 0; i < n; i++ {
+				msg = natsNexMsg(t, sub, time.Second)
+				var e MsgTraceEvent
+				err = json.Unmarshal(msg.Data, &e)
+				require_NoError(t, err)
+				ingress := e.Ingress()
+				require_True(t, ingress != nil)
+				switch ingress.Kind {
+				case CLIENT:
+					fromClient++
+				case ROUTER:
+					fromRoute++
+				default:
+					t.Fatalf("Unexpected ingress: %+v", ingress)
 				}
 			}
+			// There should be as many messages coming from the origin server
+			// and the routed server. This checks that if sampling is not 100%
+			// then when a message is routed, the header is properly deactivated.
+			require_Equal[int](t, fromClient, fromRoute)
+			// Now check that if sampling was 100%, we have the total number
+			// of published messages.
+			if test.sampling == 100 {
+				require_Equal[int](t, fromClient, total)
+			} else {
+				// Otherwise, we should have no more (but let's be conservative)
+				// than the sampling number.
+				require_LessThan[int](t, fromClient, int(float64(test.sampling*total/100)*1.35))
+			}
+		})
+	}
+}
+
+func TestMsgTraceAccDestWithSamplingJWTUpdate(t *testing.T) {
+	// create system account
+	sysKp, _ := nkeys.CreateAccount()
+	sysPub, _ := sysKp.PublicKey()
+	sysCreds := newUser(t, sysKp)
+	// create account A
+	akp, _ := nkeys.CreateAccount()
+	aPub, _ := akp.PublicKey()
+	claim := jwt.NewAccountClaims(aPub)
+	claim.Trace = &jwt.MsgTrace{Destination: "acc.trace.dest"}
+	aJwt, err := claim.Encode(oKp)
+	require_NoError(t, err)
+
+	dir := t.TempDir()
+	conf := createConfFile(t, []byte(fmt.Sprintf(`
+			listen: -1
+			operator: %s
+			resolver: {
+				type: full
+				dir: '%s'
+			}
+			system_account: %s
+		`, ojwt, dir, sysPub)))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+	updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
+
+	nc := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, akp))
+	defer nc.Close()
+
+	sub := natsSubSync(t, nc, "acc.trace.dest")
+	natsFlush(t, nc)
+
+	for iter, test := range []struct {
+		name     string
+		sampling int
+	}{
+		{"no sampling specified", 100},
+		{"sampling", 25},
+		{"set back sampling to 0", 100},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			if iter > 0 {
+				claim.Trace = &jwt.MsgTrace{Destination: "acc.trace.dest", Sampling: test.sampling}
+				aJwt, err = claim.Encode(oKp)
+				require_NoError(t, err)
+				updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
+			}
+
+			msg := nats.NewMsg("foo")
+			msg.Header.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
+			msg.Data = []byte("hello")
+
+			total := 400
+			for i := 0; i < total; i++ {
+				err := nc.PublishMsg(msg)
+				require_NoError(t, err)
+			}
+			// Wait a bit to make sure that we received all traces that should
+			// have been received.
+			time.Sleep(500 * time.Millisecond)
+			n, _, err := sub.Pending()
+			require_NoError(t, err)
+			for i := 0; i < n; i++ {
+				msg = natsNexMsg(t, sub, time.Second)
+				var e MsgTraceEvent
+				err = json.Unmarshal(msg.Data, &e)
+				require_NoError(t, err)
+			}
+			// Now check that if sampling was 100%, we have the total number
+			// of published messages.
+			if test.sampling == 100 {
+				require_Equal[int](t, n, total)
+			} else {
+				// Otherwise, we should have no more (but let's be conservative)
+				// than the sampling number.
+				require_LessThan[int](t, n, int(float64(test.sampling*total/100)*1.35))
+			}
 		})
 	}
 }
diff --git a/server/norace_test.go b/server/norace_test.go
index de85203f..3770a68b 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -9599,3 +9599,52 @@ func TestNoRaceJetStreamKVReplaceWithServerRestart(t *testing.T) {
 		t.Fatalf("Encountered errors")
 	}
 }
+
+func TestNoRaceMemStoreCompactPerformance(t *testing.T) {
+	//Load MemStore so that it is full
+	subj, msg := "foo", make([]byte, 1000)
+	storedMsgSize := memStoreMsgSize(subj, nil, msg)
+
+	toStore := uint64(10_000)
+	toStoreOnTop := uint64(1_000)
+	setSeqNo := uint64(10_000_000_000)
+
+	expectedPurge := toStore - 1
+	maxBytes := storedMsgSize * toStore
+
+	ms, err := newMemStore(&StreamConfig{Storage: MemoryStorage, MaxBytes: int64(maxBytes)})
+	require_NoError(t, err)
+	defer ms.Stop()
+
+	for i := uint64(0); i < toStore; i++ {
+		ms.StoreMsg(subj, nil, msg)
+	}
+	state := ms.State()
+	require_Equal(t, toStore, state.Msgs)
+	require_Equal(t, state.Bytes, storedMsgSize*toStore)
+
+	//1st run: Load additional messages then compact
+	for i := uint64(0); i < toStoreOnTop; i++ {
+		ms.StoreMsg(subj, nil, msg)
+	}
+	startFirstRun := time.Now()
+	purgedFirstRun, _ := ms.Compact(toStore + toStoreOnTop)
+	elapsedFirstRun := time.Since(startFirstRun)
+	require_Equal(t, expectedPurge, purgedFirstRun)
+
+	//set the seq number to a very high value by compacting with a too high seq number
+	purgedFull, _ := ms.Compact(setSeqNo)
+	require_Equal(t, 1, purgedFull)
+
+	//2nd run: Compact again
+	for i := uint64(0); i < toStore; i++ {
+		ms.StoreMsg(subj, nil, msg)
+	}
+	startSecondRun := time.Now()
+	purgedSecondRun, _ := ms.Compact(setSeqNo + toStore - 1)
+	elapsedSecondRun := time.Since(startSecondRun)
+	require_Equal(t, expectedPurge, purgedSecondRun)
+
+	//Calculate delta between runs and fail if it is too high
+	require_LessThan(t, elapsedSecondRun-elapsedFirstRun, time.Duration(1)*time.Second)
+}
diff --git a/server/routes_test.go b/server/routes_test.go
index faf01600..fbbddea7 100644
--- a/server/routes_test.go
+++ b/server/routes_test.go
@@ -4160,6 +4160,129 @@ func TestRouteNoLeakOnSlowConsumer(t *testing.T) {
 	}
 }
 
+func TestRouteSlowConsumerRecover(t *testing.T) {
+	o1 := DefaultOptions()
+	o1.Cluster.PoolSize = -1
+	s1 := RunServer(o1)
+	defer s1.Shutdown()
+
+	rtt := 1500 * time.Nanosecond
+	upRate := 1024 * 1024
+	downRate := 128 * 1024
+	np := createNetProxy(rtt, upRate, downRate, fmt.Sprintf("nats://127.0.0.1:%d", o1.Cluster.Port), true)
+	defer np.stop()
+
+	o2 := DefaultOptions()
+	o2.Cluster.PoolSize = -1
+	o2.Routes = RoutesFromStr(np.routeURL())
+	s2 := RunServer(o2)
+	defer s2.Shutdown()
+
+	checkClusterFormed(t, s1, s2)
+
+	changeWriteDeadline := func(s *Server, duration time.Duration) {
+		s.mu.Lock()
+		for _, cl := range s.routes {
+			for _, c := range cl {
+				c.mu.Lock()
+				c.out.wdl = duration
+				c.mu.Unlock()
+			}
+		}
+		s.mu.Unlock()
+	}
+	hasSlowConsumerRoutes := func(s *Server) bool {
+		var sc bool
+		s.mu.Lock()
+	Loop:
+		for _, cl := range s.routes {
+			for _, c := range cl {
+				c.mu.Lock()
+				sc = c.flags.isSet(isSlowConsumer)
+				c.mu.Unlock()
+				if sc {
+					break Loop
+				}
+			}
+		}
+		s.mu.Unlock()
+		return sc
+	}
+
+	// Start with a shorter write deadline to cause errors
+	// then bump it again later to let it recover.
+	changeWriteDeadline(s1, 1*time.Second)
+
+	ncA, err := nats.Connect(s1.Addr().String())
+	require_NoError(t, err)
+
+	ncB, err := nats.Connect(s2.Addr().String())
+	require_NoError(t, err)
+
+	var wg sync.WaitGroup
+	ncB.Subscribe("test", func(*nats.Msg) {
+		ncB.Close()
+	})
+	ncB.Flush()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 800*time.Millisecond)
+	defer cancel()
+
+	go func() {
+		var total int
+		payload := fmt.Appendf(nil, strings.Repeat("A", 132*1024))
+		for range time.NewTicker(30 * time.Millisecond).C {
+			select {
+			case <-ctx.Done():
+				wg.Done()
+				return
+			default:
+			}
+			ncA.Publish("test", payload)
+			ncA.Flush()
+			total++
+		}
+	}()
+	wg.Add(1)
+
+	checkFor(t, 20*time.Second, 2*time.Millisecond, func() error {
+		if s1.NumRoutes() < 1 {
+			return fmt.Errorf("No routes connected")
+		}
+		if !hasSlowConsumerRoutes(s1) {
+			if s1.NumSlowConsumersRoutes() > 0 {
+				// In case it has recovered already.
+				return nil
+			}
+			return fmt.Errorf("Expected Slow Consumer routes")
+		}
+		return nil
+	})
+	cancel()
+	changeWriteDeadline(s1, 5*time.Second)
+	np.updateRTT(0)
+	checkFor(t, 20*time.Second, 10*time.Millisecond, func() error {
+		if s1.NumRoutes() < 1 {
+			return fmt.Errorf("No routes connected")
+		}
+		if hasSlowConsumerRoutes(s1) {
+			return fmt.Errorf("Expected Slow Consumer routes to recover")
+		}
+		return nil
+	})
+
+	checkFor(t, 20*time.Second, 100*time.Millisecond, func() error {
+		var got, expected int64
+		got = int64(s1.NumSlowConsumersRoutes())
+		expected = 1
+		if got != expected {
+			return fmt.Errorf("got: %d, expected: %d", got, expected)
+		}
+		return nil
+	})
+	wg.Wait()
+}
+
 func TestRouteNoLeakOnAuthTimeout(t *testing.T) {
 	opts := DefaultOptions()
 	opts.Cluster.Username = "foo"
diff --git a/server/stree/stree_test.go b/server/stree/stree_test.go
index 8911b933..46ff4d80 100644
--- a/server/stree/stree_test.go
+++ b/server/stree/stree_test.go
@@ -522,6 +522,18 @@ func TestSubjectTreeInsertSamePivotBug(t *testing.T) {
 	}
 }
 
+func TestSubjectTreeMatchTsepSecondThenPartialPartBug(t *testing.T) {
+	st := NewSubjectTree[int]()
+	st.Insert(b("foo.xxxxx.foo1234.zz"), 22)
+	st.Insert(b("foo.yyy.foo123.zz"), 22)
+	st.Insert(b("foo.yyybar789.zz"), 22)
+	st.Insert(b("foo.yyy.foo12345.zz"), 22)
+	st.Insert(b("foo.yyy.foo12345.yy"), 22)
+	st.Insert(b("foo.yyy.foo123456789.zz"), 22)
+	match(t, st, "foo.*.foo123456789.*", 1)
+	match(t, st, "foo.*.*.zzz.foo.>", 0)
+}
+
 func TestSubjectTreeRandomTrackEntries(t *testing.T) {
 	st := NewSubjectTree[int]()
 	smap := make(map[string]struct{}, 1000)
@@ -566,7 +578,7 @@ func TestSubjectTreeRandomTrackEntries(t *testing.T) {
 
 func TestSubjectTreeMetaSize(t *testing.T) {
 	var base meta
-	require_Equal(t, unsafe.Sizeof(base), 64)
+	require_Equal(t, unsafe.Sizeof(base), 28)
 }
 
 func b(s string) []byte {
diff --git a/server/test_test.go b/server/test_test.go
index 35f9c526..8f0d393a 100644
--- a/server/test_test.go
+++ b/server/test_test.go
@@ -151,6 +151,15 @@ func require_ChanRead[T any](t *testing.T, ch chan T, timeout time.Duration) T {
 	panic("this shouldn't be possible")
 }
 
+func require_NoChanRead[T any](t *testing.T, ch chan T, timeout time.Duration) {
+	t.Helper()
+	select {
+	case <-ch:
+		t.Fatalf("require no read from channel within %v but got something", timeout)
+	case <-time.After(timeout):
+	}
+}
+
 func checkNatsError(t *testing.T, e *ApiError, id ErrorIdentifier) {
 	t.Helper()
 	ae, ok := ApiErrors[id]
diff --git a/server/websocket_test.go b/server/websocket_test.go
index 5853731d..1eacf4b7 100644
--- a/server/websocket_test.go
+++ b/server/websocket_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020-2023 The NATS Authors
+// Copyright 2020-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
@@ -1956,14 +1956,20 @@ func setupAddCookie(o *Options) {
 	o.Websocket.JWTCookie = "jwt"
 }
 
-func testWSCreateClientGetInfo(t testing.TB, compress, web bool, host string, port int) (net.Conn, *bufio.Reader, []byte) {
+func testWSCreateClientGetInfo(t testing.TB, compress, web bool, host string, port int, cookies ...string) (net.Conn, *bufio.Reader, []byte) {
 	t.Helper()
-	return testNewWSClient(t, testWSClientOptions{
+	opts := testWSClientOptions{
 		compress: compress,
 		web:      web,
 		host:     host,
 		port:     port,
-	})
+	}
+
+	if len(cookies) > 0 {
+		opts.extraHeaders = map[string][]string{}
+		opts.extraHeaders["Cookie"] = cookies
+	}
+	return testNewWSClient(t, opts)
 }
 
 func testWSCreateClient(t testing.TB, compress, web bool, host string, port int) (net.Conn, *bufio.Reader) {
@@ -3140,11 +3146,12 @@ func TestWSCompressionFrameSizeLimit(t *testing.T) {
 
 func TestWSBasicAuth(t *testing.T) {
 	for _, test := range []struct {
-		name string
-		opts func() *Options
-		user string
-		pass string
-		err  string
+		name    string
+		opts    func() *Options
+		user    string
+		pass    string
+		err     string
+		cookies []string
 	}{
 		{
 			"top level auth, no override, wrong u/p",
@@ -3155,6 +3162,7 @@ func TestWSBasicAuth(t *testing.T) {
 				return o
 			},
 			"websocket", "client", "-ERR 'Authorization Violation'",
+			nil,
 		},
 		{
 			"top level auth, no override, correct u/p",
@@ -3165,6 +3173,7 @@ func TestWSBasicAuth(t *testing.T) {
 				return o
 			},
 			"normal", "client", "",
+			nil,
 		},
 		{
 			"no top level auth, ws auth, wrong u/p",
@@ -3175,6 +3184,7 @@ func TestWSBasicAuth(t *testing.T) {
 				return o
 			},
 			"normal", "client", "-ERR 'Authorization Violation'",
+			nil,
 		},
 		{
 			"no top level auth, ws auth, correct u/p",
@@ -3185,6 +3195,7 @@ func TestWSBasicAuth(t *testing.T) {
 				return o
 			},
 			"websocket", "client", "",
+			nil,
 		},
 		{
 			"top level auth, ws override, wrong u/p",
@@ -3197,6 +3208,7 @@ func TestWSBasicAuth(t *testing.T) {
 				return o
 			},
 			"normal", "client", "-ERR 'Authorization Violation'",
+			nil,
 		},
 		{
 			"top level auth, ws override, correct u/p",
@@ -3209,6 +3221,68 @@ func TestWSBasicAuth(t *testing.T) {
 				return o
 			},
 			"websocket", "client", "",
+			nil,
+		},
+		{
+			"username/password from cookies",
+			func() *Options {
+				o := testWSOptions()
+				o.Websocket.UsernameCookie = "un"
+				o.Websocket.PasswordCookie = "pw"
+				o.Username = "me"
+				o.Password = "s3cr3t!"
+				return o
+			},
+			"", "", "",
+			[]string{"un=me", "pw=s3cr3t!"},
+		},
+		{
+			"bad username/ good password from cookies",
+			func() *Options {
+				o := testWSOptions()
+				o.Websocket.UsernameCookie = "un"
+				o.Websocket.PasswordCookie = "pw"
+				o.Username = "me"
+				o.Password = "s3cr3t!"
+				return o
+			},
+			"", "", "-ERR 'Authorization Violation",
+			[]string{"un=m", "pw=s3cr3t!"},
+		},
+		{
+			"good username/ bad password from cookies",
+			func() *Options {
+				o := testWSOptions()
+				o.Websocket.UsernameCookie = "un"
+				o.Websocket.PasswordCookie = "pw"
+				o.Username = "me"
+				o.Password = "s3cr3t!"
+				return o
+			},
+			"", "", "-ERR 'Authorization Violation",
+			[]string{"un=me", "pw=hi!"},
+		},
+		{
+			"token from cookie",
+			func() *Options {
+				o := testWSOptions()
+				o.Websocket.TokenCookie = "tok"
+				o.Authorization = "l3tm31n!"
+				return o
+			},
+			"", "", "",
+			[]string{"tok=l3tm31n!"},
+		},
+		{
+			"bad token from cookie",
+			func() *Options {
+				o := testWSOptions()
+				o.Websocket.TokenCookie = "tok"
+				o.Authorization = "l3tm31n!"
+				return o
+			},
+			"", "", "-ERR 'Authorization Violation",
+			[]string{"tok=hello!"},
 		},
 	} {
 		t.Run(test.name, func(t *testing.T) {
@@ -3216,7 +3290,7 @@ func TestWSBasicAuth(t *testing.T) {
 			s := RunServer(o)
 			defer s.Shutdown()
 
-			wsc, br, _ := testWSCreateClientGetInfo(t, false, false, o.Websocket.Host, o.Websocket.Port)
+			wsc, br, _ := testWSCreateClientGetInfo(t, false, false, o.Websocket.Host, o.Websocket.Port, test.cookies...)
 			defer wsc.Close()
 
 			connectProto := fmt.Sprintf("CONNECT {\"verbose\":false,\"protocol\":1,\"user\":\"%s\",\"pass\":\"%s\"}\r\nPING\r\n",
@@ -3735,7 +3809,6 @@ func TestWSJWTWithAllowedConnectionTypes(t *testing.T) {
 }
 
 func TestWSJWTCookieUser(t *testing.T) {
-
 	nucSigFunc := func() *jwt.UserClaims { return newJWTTestUserClaims() }
 	nucBearerFunc := func() *jwt.UserClaims {
 		ret := newJWTTestUserClaims()
diff --git a/test/ocsp_peer_test.go b/test/ocsp_peer_test.go
index 6df120d7..ed25ef85 100644
--- a/test/ocsp_peer_test.go
+++ b/test/ocsp_peer_test.go
@@ -2821,7 +2821,7 @@ func TestOCSPPeerNextUpdateUnset(t *testing.T) {
 	respCertPEM := "configs/certs/ocsp_peer/mini-ca/ocsp1/ocsp1_bundle.pem"
 	respKeyPEM := "configs/certs/ocsp_peer/mini-ca/ocsp1/private/ocsp1_keypair.pem"
 	issuerCertPEM := "configs/certs/ocsp_peer/mini-ca/intermediate1/intermediate1_cert.pem"
-	intermediateCA1Responder := newOCSPResponderBase(t, issuerCertPEM, respCertPEM, respKeyPEM, true, "127.0.0.1:18888", 0)
+	intermediateCA1Responder := newOCSPResponderBase(t, issuerCertPEM, respCertPEM, respKeyPEM, true, "127.0.0.1:18888", 0, "")
 	intermediateCA1ResponderURL := fmt.Sprintf("http://%s", intermediateCA1Responder.Addr)
 	defer intermediateCA1Responder.Shutdown(ctx)
 	setOCSPStatus(t, intermediateCA1ResponderURL, "configs/certs/ocsp_peer/mini-ca/client1/UserA1_cert.pem", ocsp.Good)
diff --git a/test/ocsp_test.go b/test/ocsp_test.go
index 4f1ecfe0..56eef3fc 100644
--- a/test/ocsp_test.go
+++ b/test/ocsp_test.go
@@ -3091,20 +3091,25 @@ func TestOCSPCustomConfigReloadEnable(t *testing.T) {
 
 func newOCSPResponderCustomAddress(t *testing.T, issuerCertPEM, issuerKeyPEM string, addr string) *http.Server {
 	t.Helper()
-	return newOCSPResponderBase(t, issuerCertPEM, issuerCertPEM, issuerKeyPEM, false, addr, defaultResponseTTL)
+	return newOCSPResponderBase(t, issuerCertPEM, issuerCertPEM, issuerKeyPEM, false, addr, defaultResponseTTL, "")
 }
 
 func newOCSPResponder(t *testing.T, issuerCertPEM, issuerKeyPEM string) *http.Server {
 	t.Helper()
-	return newOCSPResponderBase(t, issuerCertPEM, issuerCertPEM, issuerKeyPEM, false, defaultAddress, defaultResponseTTL)
+	return newOCSPResponderBase(t, issuerCertPEM, issuerCertPEM, issuerKeyPEM, false, defaultAddress, defaultResponseTTL, "")
 }
 
 func newOCSPResponderDesignatedCustomAddress(t *testing.T, issuerCertPEM, respCertPEM, respKeyPEM string, addr string) *http.Server {
 	t.Helper()
-	return newOCSPResponderBase(t, issuerCertPEM, respCertPEM, respKeyPEM, true, addr, defaultResponseTTL)
+	return newOCSPResponderBase(t, issuerCertPEM, respCertPEM, respKeyPEM, true, addr, defaultResponseTTL, "")
 }
 
-func newOCSPResponderBase(t *testing.T, issuerCertPEM, respCertPEM, respKeyPEM string, embed bool, addr string, responseTTL time.Duration) *http.Server {
+func newOCSPResponderPreferringHTTPMethod(t *testing.T, issuerCertPEM, issuerKeyPEM, method string) *http.Server {
+	t.Helper()
+	return newOCSPResponderBase(t, issuerCertPEM, issuerCertPEM, issuerKeyPEM, false, defaultAddress, defaultResponseTTL, method)
+}
+
+func newOCSPResponderBase(t *testing.T, issuerCertPEM, respCertPEM, respKeyPEM string, embed bool, addr string, responseTTL time.Duration, method string) *http.Server {
 	t.Helper()
 	var mu sync.Mutex
 	status := make(map[string]int)
@@ -3157,12 +3162,26 @@ func newOCSPResponderBase(t *testing.T, issuerCertPEM, respCertPEM, respKeyPEM s
 	// OCSP status request and signs a response with a CA. Lightly based off:
 	// https://www.ietf.org/rfc/rfc2560.txt
 	mux.HandleFunc("/", func(rw http.ResponseWriter, r *http.Request) {
-		if r.Method != "GET" {
+		var reqData []byte
+		var err error
+
+		switch {
+		case r.Method == "GET":
+			if method != "" && r.Method != method {
+				http.Error(rw, "", http.StatusBadRequest)
+				return
+			}
+			reqData, err = base64.StdEncoding.DecodeString(r.URL.Path[1:])
+		case r.Method == "POST":
+			if method != "" && r.Method != method {
+				http.Error(rw, "", http.StatusBadRequest)
+				return
+			}
+			reqData, err = io.ReadAll(r.Body)
+		default:
 			http.Error(rw, "Method Not Allowed", http.StatusMethodNotAllowed)
 			return
 		}
-
-		reqData, err := base64.StdEncoding.DecodeString(r.URL.Path[1:])
 		if err != nil {
 			http.Error(rw, err.Error(), http.StatusBadRequest)
 			return
@@ -4198,3 +4217,151 @@ func TestMixedCAOCSPSuperCluster(t *testing.T) {
 		t.Errorf("Unexpected error: %v", lerr)
 	}
 }
+
+func TestOCSPResponderHTTPMethods(t *testing.T) {
+	t.Run("prefer get", func(t *testing.T) {
+		testOCSPResponderHTTPMethods(t, "GET")
+	})
+	t.Run("prefer post", func(t *testing.T) {
+		testOCSPResponderHTTPMethods(t, "POST")
+	})
+	t.Run("all methods failing", func(t *testing.T) {
+		testOCSPResponderFailing(t, "TEST")
+	})
+}
+
+func testOCSPResponderHTTPMethods(t *testing.T, method string) {
+	const (
+		caCert     = "configs/certs/ocsp/ca-cert.pem"
+		caKey      = "configs/certs/ocsp/ca-key.pem"
+		serverCert = "configs/certs/ocsp/server-cert.pem"
+		serverKey  = "configs/certs/ocsp/server-key.pem"
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ocspr := newOCSPResponderPreferringHTTPMethod(t, caCert, caKey, method)
+	defer ocspr.Shutdown(ctx)
+	addr := fmt.Sprintf("http://%s", ocspr.Addr)
+	setOCSPStatus(t, addr, serverCert, ocsp.Good)
+
+	// Add another responder that fails.
+	badaddr := "http://127.0.0.1:8889"
+	badocsp := newOCSPResponderCustomAddress(t, caCert, caKey, badaddr)
+	defer badocsp.Shutdown(ctx)
+
+	opts := server.Options{}
+	opts.Host = "127.0.0.1"
+	opts.NoLog = true
+	opts.NoSigs = true
+	opts.MaxControlLine = 4096
+	opts.Port = -1
+	opts.TLSCert = serverCert
+	opts.TLSKey = serverKey
+	opts.TLSCaCert = caCert
+	opts.TLSTimeout = 5
+	tcOpts := &server.TLSConfigOpts{
+		CertFile: opts.TLSCert,
+		KeyFile:  opts.TLSKey,
+		CaFile:   opts.TLSCaCert,
+		Timeout:  opts.TLSTimeout,
+	}
+
+	tlsConf, err := server.GenTLSConfig(tcOpts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	opts.TLSConfig = tlsConf
+
+	opts.OCSPConfig = &server.OCSPConfig{
+		Mode:         server.OCSPModeAlways,
+		OverrideURLs: []string{badaddr, addr},
+	}
+	srv := RunServer(&opts)
+	defer srv.Shutdown()
+
+	nc, err := nats.Connect(fmt.Sprintf("tls://localhost:%d", opts.Port),
+		nats.Secure(&tls.Config{
+			VerifyConnection: func(s tls.ConnectionState) error {
+				resp, err := getOCSPStatus(s)
+				if err != nil {
+					return err
+				}
+				if resp.Status != ocsp.Good {
+					return fmt.Errorf("invalid staple")
+				}
+				return nil
+			},
+		}),
+		nats.RootCAs(caCert),
+		nats.ErrorHandler(noOpErrHandler),
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	sub, err := nc.SubscribeSync("foo")
+	if err != nil {
+		t.Fatal(err)
+	}
+	nc.Publish("foo", []byte("hello world"))
+	nc.Flush()
+
+	_, err = sub.NextMsg(1 * time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	nc.Close()
+}
+
+func testOCSPResponderFailing(t *testing.T, method string) {
+	const (
+		caCert     = "configs/certs/ocsp/ca-cert.pem"
+		caKey      = "configs/certs/ocsp/ca-key.pem"
+		serverCert = "configs/certs/ocsp/server-cert.pem"
+		serverKey  = "configs/certs/ocsp/server-key.pem"
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ocspr := newOCSPResponderPreferringHTTPMethod(t, caCert, caKey, method)
+	defer ocspr.Shutdown(ctx)
+	addr := fmt.Sprintf("http://%s", ocspr.Addr)
+	setOCSPStatus(t, addr, serverCert, ocsp.Good)
+
+	opts := server.Options{}
+	opts.Host = "127.0.0.1"
+	opts.NoLog = true
+	opts.NoSigs = true
+	opts.MaxControlLine = 4096
+	opts.Port = -1
+	opts.TLSCert = serverCert
+	opts.TLSKey = serverKey
+	opts.TLSCaCert = caCert
+	opts.TLSTimeout = 5
+	tcOpts := &server.TLSConfigOpts{
+		CertFile: opts.TLSCert,
+		KeyFile:  opts.TLSKey,
+		CaFile:   opts.TLSCaCert,
+		Timeout:  opts.TLSTimeout,
+	}
+
+	tlsConf, err := server.GenTLSConfig(tcOpts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	opts.TLSConfig = tlsConf
+
+	opts.OCSPConfig = &server.OCSPConfig{
+		Mode:         server.OCSPModeAlways,
+		OverrideURLs: []string{addr},
+	}
+	expected := "bad OCSP status update for certificate at 'configs/certs/ocsp/server-cert.pem': "
+	expected += "exhausted ocsp servers: non-ok http status on POST request (reqlen=68): 400\nnon-ok http status on GET request (reqlen=92): 400"
+	_, err = server.NewServer(&opts)
+	if err == nil {
+		t.Error("Unexpected success setting up server")
+	} else if err.Error() != expected {
+		t.Errorf("Expected %q, got: %q", expected, err.Error())
+	}
+}