Repository: nats-io/nats-server
Pull request #5101 (2024-02-17 06:30:56)

Track recovery from slow consumer events in routes
Adds extra logging for routes to notice when a route that has temporarily entered the Slow Consumer state has recovered.  
Also changes the counter for route slow consumer to only count each time the server detects that a route became a slow consumer.

Example logs:

```
[20] 2024/02/17 06:18:48.891218 [INF] 10.84.13.21:33226 - rid:3579 - Slow Consumer Detected: WriteDeadline of 10s exceeded with 160926 chunks of 349984270 total bytes.
[20] 2024/02/17 06:18:58.919430 [INF] 10.84.13.21:33226 - rid:3579 - Slow Consumer State: WriteDeadline of 10s exceeded with 237701 chunks of 515825015 total bytes.
[20] 2024/02/17 06:19:08.979502 [INF] 10.84.13.21:33226 - rid:3579 - Slow Consumer State: WriteDeadline of 10s exceeded with 329433 chunks of 715565941 total bytes.
[20] 2024/02/17 06:19:18.994486 [INF] 10.84.13.21:33226 - rid:3579 - Slow Consumer State: WriteDeadline of 10s exceeded with 533167 chunks of 1158645903 total bytes.
[20] 2024/02/17 06:19:29.025948 [INF] 10.84.13.21:33226 - rid:3579 - Slow Consumer State: WriteDeadline of 10s exceeded with 622193 chunks of 1352896878 total bytes.
[20] 2024/02/17 06:19:39.048491 [INF] 10.84.13.21:33226 - rid:3579 - Slow Consumer State: WriteDeadline of 10s exceeded with 624307 chunks of 1357420714 total bytes.
[20] 2024/02/17 06:19:49.164586 [INF] 10.84.13.21:33226 - rid:3579 - Slow Consumer State: WriteDeadline of 10s exceeded with 595656 chunks of 1294726213 total bytes.
[20] 2024/02/17 06:19:59.177052 [INF] 10.84.13.21:33226 - rid:3579 - Slow Consumer State: WriteDeadline of 10s exceeded with 495507 chunks of 1076217270 total bytes.
[20] 2024/02/17 06:20:09.188345 [INF] 10.84.13.21:33226 - rid:3579 - Slow Consumer State: WriteDeadline of 10s exceeded with 401938 chunks of 872752366 total bytes.
[20] 2024/02/17 06:20:15.430644 [INF] 10.84.13.21:33226 - rid:3579 - Slow Consumer Recovered: Flush took 6.239s with 218191 chunks of 473998204 total bytes.
```


Patch:
diff --git a/go.mod b/go.mod
index 7cc0b457..74c401ba 100644
--- a/go.mod
+++ b/go.mod
@@ -5,8 +5,8 @@ go 1.20
 require (
 	github.com/klauspost/compress v1.17.6
 	github.com/minio/highwayhash v1.0.2
-	github.com/nats-io/jwt/v2 v2.5.4
-	github.com/nats-io/nats.go v1.33.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 bcc0f52c..a90c64d7 100644
--- a/go.sum
+++ b/go.sum
@@ -3,10 +3,10 @@ github.com/klauspost/compress v1.17.6 h1:60eq2E/jlfwQXtvZEeBUYADs+BwKBWURIY+Gj2e
 github.com/klauspost/compress v1.17.6/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM=
 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.4 h1:Bz+drKl2GbE30fxTOtb0NYl1BQ5RwZ+Zcqkg3mR5bbI=
-github.com/nats-io/jwt/v2 v2.5.4/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
-github.com/nats-io/nats.go v1.33.0 h1:rRg0l2F29B30n6EPl0j50hl8eYp7rA2ecoJ74E62US8=
-github.com/nats-io/nats.go v1.33.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=
diff --git a/server/accounts.go b/server/accounts.go
index 754aa5bb..4600dcc3 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -96,7 +96,12 @@ type Account struct {
 	nameTag      string
 	lastLimErr   int64
 	routePoolIdx int
-	traceDest    string
+	// 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 (
@@ -263,11 +268,12 @@ func (a *Account) setTraceDest(dest string) {
 	a.mu.Unlock()
 }
 
-func (a *Account) getTraceDest() string {
+func (a *Account) getTraceDestAndSampling() (string, int) {
 	a.mu.RLock()
 	dest := a.traceDest
+	sampling := a.traceDestSampling
 	a.mu.RUnlock()
-	return dest
+	return dest, sampling
 }
 
 // Used to create shallow copies of accounts for transfer
@@ -278,7 +284,7 @@ func (a *Account) getTraceDest() string {
 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))
@@ -3239,12 +3245,18 @@ 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 TraceDest
-		a.traceDest = string(ac.Trace.Destination)
-	} else {
-		a.traceDest = _EMPTY_
+		// 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() {
diff --git a/server/client.go b/server/client.go
index 09baccea..28d10b76 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)
@@ -1651,9 +1652,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 +1694,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 +1724,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 +1736,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 +1748,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
 }
@@ -4356,8 +4374,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/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 94a27ace..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,13 +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 {
-			dest = acc.getTraceDest()
+			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,
@@ -472,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
@@ -637,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}
@@ -672,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 c9b147e9..98f90db2 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -2896,6 +2896,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 +3088,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{
diff --git a/server/stream.go b/server/stream.go
index dc3776f2..405ee958 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -4224,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


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index 0f8c403e..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.getTraceDest(), 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/leafnode_test.go b/server/leafnode_test.go
index 038e8674..1b508f5e 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -7398,6 +7398,64 @@ 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))
+
+	// 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
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index efbfa2e4..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},
diff --git a/server/msgtrace_test.go b/server/msgtrace_test.go
index 6550f7e1..a5d001e5 100644
--- a/server/msgtrace_test.go
+++ b/server/msgtrace_test.go
@@ -4898,3 +4898,216 @@ func TestMsgTraceStreamJWTUpdate(t *testing.T) {
 		})
 	}
 }
+
+func TestMsgTraceParseAccountDestWithSampling(t *testing.T) {
+	tmpl := `
+		port: -1
+		accounts {
+			A {
+				users: [{user: a, password: pwd}]
+				%s
+			}
+		}
+	`
+	for _, test := range []struct {
+		name        string
+		samplingStr string
+		want        int
+	}{
+		{"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) {
+			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/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())
+	}
+}