Repository: nats-io/nats-server
Pull request #4912 (2023-12-26 17:31:10)

Cherry-pick PRs into v2.10.8 release branch
Includes:

https://github.com/nats-io/nats-server/pull/4901
https://github.com/nats-io/nats-server/pull/4903
https://github.com/nats-io/nats-server/pull/4905
https://github.com/nats-io/nats-server/pull/4909
https://github.com/nats-io/nats-server/pull/4910



Patch:
diff --git a/internal/fastrand/LICENSE b/internal/fastrand/LICENSE
new file mode 100644
index 00000000..c12aa07b
--- /dev/null
+++ b/internal/fastrand/LICENSE
@@ -0,0 +1,27 @@
+Copyright (c) 2011 The LevelDB-Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+   * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+   * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
\ No newline at end of file
diff --git a/internal/fastrand/fastrand.go b/internal/fastrand/fastrand.go
new file mode 100644
index 00000000..47953812
--- /dev/null
+++ b/internal/fastrand/fastrand.go
@@ -0,0 +1,23 @@
+// Copyright 2020-2023 The LevelDB-Go, Pebble and NATS Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package fastrand
+
+import _ "unsafe" // required by go:linkname
+
+// Uint32 returns a lock free uint32 value.
+//
+//go:linkname Uint32 runtime.fastrand
+func Uint32() uint32
+
+// Uint32n returns a lock free uint32 value in the interval [0, n).
+//
+//go:linkname Uint32n runtime.fastrandn
+func Uint32n(n uint32) uint32
+
+// Uint32 returns a lock free uint64 value.
+func Uint64() uint64 {
+	v := uint64(Uint32())
+	return v<<32 | uint64(Uint32())
+}
diff --git a/server/accounts.go b/server/accounts.go
index 6d301ebd..a9bf057e 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -33,6 +33,7 @@ import (
 	"time"
 
 	"github.com/nats-io/jwt/v2"
+	"github.com/nats-io/nats-server/v2/internal/fastrand"
 	"github.com/nats-io/nkeys"
 	"github.com/nats-io/nuid"
 )
@@ -85,10 +86,9 @@ type Account struct {
 	incomplete   bool
 	signingKeys  map[string]jwt.Scope
 	extAuth      *jwt.ExternalAuthorization
-	srv          *Server    // server this account is registered with (possibly nil)
-	lds          string     // loop detection subject for leaf nodes
-	siReply      []byte     // service reply prefix, will form wildcard subscription.
-	prand        *rand.Rand // NOT threadsafe, must have WRITE lock on Account
+	srv          *Server // server this account is registered with (possibly nil)
+	lds          string  // loop detection subject for leaf nodes
+	siReply      []byte  // service reply prefix, will form wildcard subscription.
 	eventIds     *nuid.NUID
 	eventIdsMu   sync.Mutex
 	defaultPerms *Permissions
@@ -238,7 +238,6 @@ func NewAccount(name string) *Account {
 		Name:     name,
 		limits:   limits{-1, -1, -1, -1, false},
 		eventIds: nuid.New(),
-		prand:    rand.New(rand.NewSource(time.Now().UnixNano())),
 	}
 	return a
 }
@@ -290,7 +289,6 @@ func (a *Account) shallowCopy(na *Account) {
 		}
 	}
 	na.mappings = a.mappings
-	na.prand = rand.New(rand.NewSource(time.Now().UnixNano()))
 	na.hasMapped.Store(len(na.mappings) > 0)
 
 	// JetStream
@@ -802,7 +800,7 @@ func (a *Account) selectMappedSubject(dest string) (string, bool) {
 	if len(dests) == 1 && dests[0].weight == 100 {
 		d = dests[0]
 	} else {
-		w := uint8(a.prand.Int31n(100))
+		w := uint8(fastrand.Uint32n(100))
 		for _, rm := range dests {
 			if w < rm.weight {
 				d = rm
@@ -2186,7 +2184,7 @@ func (a *Account) processServiceImportResponse(sub *subscription, c *client, _ *
 // Lock should be held.
 func (a *Account) createRespWildcard() {
 	var b = [baseServerLen]byte{'_', 'R', '_', '.'}
-	rn := a.prand.Uint64()
+	rn := fastrand.Uint64()
 	for i, l := replyPrefixLen, rn; i < len(b); i++ {
 		b[i] = digits[l%base]
 		l /= base
@@ -2205,7 +2203,7 @@ func isTrackedReply(reply []byte) bool {
 func (a *Account) newServiceReply(tracking bool) []byte {
 	a.mu.Lock()
 	s := a.srv
-	rn := a.prand.Uint64()
+	rn := fastrand.Uint64()
 
 	// Check if we need to create the reply here.
 	var createdSiReply bool
diff --git a/server/client.go b/server/client.go
index 5fabe77c..4ef424cb 100644
--- a/server/client.go
+++ b/server/client.go
@@ -35,6 +35,7 @@ import (
 
 	"github.com/klauspost/compress/s2"
 	"github.com/nats-io/jwt/v2"
+	"github.com/nats-io/nats-server/v2/internal/fastrand"
 )
 
 // Type of client connection.
@@ -442,8 +443,6 @@ type readCache struct {
 	// to make sure to only send one message and properly scope to queues as needed.
 	rts []routeTarget
 
-	prand *rand.Rand
-
 	// These are all temporary totals for an invocation of a read in readloop.
 	msgs  int32
 	bytes int32
@@ -4505,12 +4504,6 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 		goto sendToRoutesOrLeafs
 	}
 
-	// Check to see if we have our own rand yet. Global rand
-	// has contention with lots of clients, etc.
-	if c.in.prand == nil {
-		c.in.prand = rand.New(rand.NewSource(time.Now().UnixNano()))
-	}
-
 	// Process queue subs
 	for i := 0; i < len(r.qsubs); i++ {
 		qsubs := r.qsubs[i]
@@ -4558,7 +4551,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 		sindex := 0
 		lqs := len(qsubs)
 		if lqs > 1 {
-			sindex = c.in.prand.Int() % lqs
+			sindex = int(fastrand.Uint32()) % lqs
 		}
 
 		// Find a subscription that is able to deliver this message starting at a random index.
diff --git a/server/filestore.go b/server/filestore.go
index 50361a5e..e4ba028b 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -795,7 +795,7 @@ var blkPoolSmall sync.Pool  // 2MB
 
 // Get a new msg block based on sz estimate.
 func getMsgBlockBuf(sz int) (buf []byte) {
-	var pb interface{}
+	var pb any
 	if sz <= defaultSmallBlockSize {
 		pb = blkPoolSmall.Get()
 	} else if sz <= defaultMediumBlockSize {
@@ -6132,21 +6132,23 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 	for i := 0; i < len(fs.blks); i++ {
 		mb := fs.blks[i]
 		mb.mu.Lock()
-		if err := mb.ensurePerSubjectInfoLoaded(); err != nil {
-			mb.mu.Unlock()
-			continue
+		var shouldExpire bool
+		if mb.cacheNotLoaded() {
+			mb.loadMsgsWithLock()
+			shouldExpire = true
 		}
+
 		t, f, l := mb.filteredPendingLocked(subject, wc, atomic.LoadUint64(&mb.first.seq))
 		if t == 0 {
+			// Expire if we were responsible for loading.
+			if shouldExpire {
+				// Expire this cache before moving on.
+				mb.tryForceExpireCacheLocked()
+			}
 			mb.mu.Unlock()
 			continue
 		}
 
-		var shouldExpire bool
-		if mb.cacheNotLoaded() {
-			mb.loadMsgsWithLock()
-			shouldExpire = true
-		}
 		if sequence > 1 && sequence <= l {
 			l = sequence - 1
 		}
diff --git a/server/jetstream.go b/server/jetstream.go
index d146186e..6b333878 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -71,11 +71,13 @@ type JetStreamAccountLimits struct {
 }
 
 type JetStreamTier struct {
-	Memory    uint64                 `json:"memory"`
-	Store     uint64                 `json:"storage"`
-	Streams   int                    `json:"streams"`
-	Consumers int                    `json:"consumers"`
-	Limits    JetStreamAccountLimits `json:"limits"`
+	Memory         uint64                 `json:"memory"`
+	Store          uint64                 `json:"storage"`
+	ReservedMemory uint64                 `json:"reserved_memory"`
+	ReservedStore  uint64                 `json:"reserved_storage"`
+	Streams        int                    `json:"streams"`
+	Consumers      int                    `json:"consumers"`
+	Limits         JetStreamAccountLimits `json:"limits"`
 }
 
 // JetStreamAccountStats returns current statistics about the account's JetStream usage.
@@ -1580,6 +1582,40 @@ func diffCheckedLimits(a, b map[string]JetStreamAccountLimits) map[string]JetStr
 	return diff
 }
 
+// Return reserved bytes for memory and store for this account on this server.
+// Lock should be held.
+func (jsa *jsAccount) reservedStorage(tier string) (mem, store uint64) {
+	for _, mset := range jsa.streams {
+		cfg := &mset.cfg
+		if tier == _EMPTY_ || tier == tierName(cfg) && cfg.MaxBytes > 0 {
+			switch cfg.Storage {
+			case FileStorage:
+				store += uint64(cfg.MaxBytes)
+			case MemoryStorage:
+				mem += uint64(cfg.MaxBytes)
+			}
+		}
+	}
+	return mem, store
+}
+
+// Return reserved bytes for memory and store for this account in clustered mode.
+// js lock should be held.
+func reservedStorage(sas map[string]*streamAssignment, tier string) (mem, store uint64) {
+	for _, sa := range sas {
+		cfg := sa.Config
+		if tier == _EMPTY_ || tier == tierName(cfg) && cfg.MaxBytes > 0 {
+			switch cfg.Storage {
+			case FileStorage:
+				store += uint64(cfg.MaxBytes)
+			case MemoryStorage:
+				mem += uint64(cfg.MaxBytes)
+			}
+		}
+	}
+	return mem, store
+}
+
 // JetStreamUsage reports on JetStream usage and limits for an account.
 func (a *Account) JetStreamUsage() JetStreamAccountStats {
 	a.mu.RLock()
@@ -1591,6 +1627,8 @@ func (a *Account) JetStreamUsage() JetStreamAccountStats {
 	if jsa != nil {
 		js := jsa.js
 		js.mu.RLock()
+		cc := js.cluster
+		singleServer := cc == nil
 		jsa.mu.RLock()
 		jsa.usageMu.RLock()
 		stats.Memory, stats.Store = jsa.storageTotals()
@@ -1599,6 +1637,11 @@ func (a *Account) JetStreamUsage() JetStreamAccountStats {
 			Total:  jsa.apiTotal,
 			Errors: jsa.apiErrors,
 		}
+		if singleServer {
+			stats.ReservedMemory, stats.ReservedStore = jsa.reservedStorage(_EMPTY_)
+		} else {
+			stats.ReservedMemory, stats.ReservedStore = reservedStorage(cc.streams[aname], _EMPTY_)
+		}
 		l, defaultTier := jsa.limits[_EMPTY_]
 		if defaultTier {
 			stats.Limits = l
@@ -1611,27 +1654,42 @@ func (a *Account) JetStreamUsage() JetStreamAccountStats {
 					// In case this shows an empty stream, that tier will be added when iterating over streams
 					skipped++
 				} else {
-					stats.Tiers[t] = JetStreamTier{
+					tier := JetStreamTier{
 						Memory: uint64(total.total.mem),
 						Store:  uint64(total.total.store),
 						Limits: jsa.limits[t],
 					}
+					if singleServer {
+						tier.ReservedMemory, tier.ReservedStore = jsa.reservedStorage(t)
+					} else {
+						tier.ReservedMemory, tier.ReservedStore = reservedStorage(cc.streams[aname], t)
+					}
+					stats.Tiers[t] = tier
 				}
 			}
 			if len(accJsLimits) != len(jsa.usage)-skipped {
 				// insert unused limits
 				for t, lim := range accJsLimits {
 					if _, ok := stats.Tiers[t]; !ok {
-						stats.Tiers[t] = JetStreamTier{Limits: lim}
+						tier := JetStreamTier{Limits: lim}
+						if singleServer {
+							tier.ReservedMemory, tier.ReservedStore = jsa.reservedStorage(t)
+						} else {
+							tier.ReservedMemory, tier.ReservedStore = reservedStorage(cc.streams[aname], t)
+						}
+						stats.Tiers[t] = tier
 					}
 				}
 			}
 		}
 		jsa.usageMu.RUnlock()
-		if cc := jsa.js.cluster; cc != nil {
+
+		// Clustered
+		if cc := js.cluster; cc != nil {
 			sas := cc.streams[aname]
 			if defaultTier {
 				stats.Streams = len(sas)
+				stats.ReservedMemory, stats.ReservedStore = reservedStorage(sas, _EMPTY_)
 			}
 			for _, sa := range sas {
 				stats.Consumers += len(sa.consumers)
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 96ec1f62..6c282c26 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -218,6 +218,9 @@ const (
 	JSApiServerStreamCancelMove  = "$JS.API.ACCOUNT.STREAM.CANCEL_MOVE.*.*"
 	JSApiServerStreamCancelMoveT = "$JS.API.ACCOUNT.STREAM.CANCEL_MOVE.%s.%s"
 
+	// The prefix for system level account API.
+	jsAPIAccountPre = "$JS.API.ACCOUNT."
+
 	// jsAckT is the template for the ack message stream coming back from a consumer
 	// when they ACK/NAK, etc a message.
 	jsAckT      = "$JS.ACK.%s.%s"
@@ -347,6 +350,8 @@ type ApiResponse struct {
 	Error *ApiError `json:"error,omitempty"`
 }
 
+const JSApiSystemResponseType = "io.nats.jetstream.api.v1.system_response"
+
 // When passing back to the clients generalize store failures.
 var (
 	errStreamStoreFailed   = errors.New("error creating store for stream")
@@ -739,26 +744,59 @@ type jsAPIRoutedReq struct {
 }
 
 func (js *jetStream) apiDispatch(sub *subscription, c *client, acc *Account, subject, reply string, rmsg []byte) {
+	// Ignore system level directives meta stepdown and peer remove requests here.
+	if subject == JSApiLeaderStepDown ||
+		subject == JSApiRemoveServer ||
+		strings.HasPrefix(subject, jsAPIAccountPre) {
+		return
+	}
 	// No lock needed, those are immutable.
 	s, rr := js.srv, js.apiSubs.Match(subject)
 
-	hdr, _ := c.msgParts(rmsg)
+	hdr, msg := c.msgParts(rmsg)
 	if len(getHeader(ClientInfoHdr, hdr)) == 0 {
 		// Check if this is the system account. We will let these through for the account info only.
-		if s.SystemAccount() != acc || subject != JSApiAccountInfo {
+		sacc := s.SystemAccount()
+		if sacc != acc {
+			return
+		}
+		if subject != JSApiAccountInfo {
+			// Only respond from the initial server entry to the NATS system.
+			if c.kind == CLIENT || c.kind == LEAF {
+				var resp = ApiResponse{
+					Type:  JSApiSystemResponseType,
+					Error: NewJSNotEnabledForAccountError(),
+				}
+				s.sendAPIErrResponse(nil, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			}
 			return
 		}
 	}
 
 	// Short circuit for no interest.
 	if len(rr.psubs)+len(rr.qsubs) == 0 {
+		if (c.kind == CLIENT || c.kind == LEAF) && acc != s.SystemAccount() {
+			ci, acc, _, _, _ := s.getRequestInfo(c, rmsg)
+			var resp = ApiResponse{
+				Type:  JSApiSystemResponseType,
+				Error: NewJSBadRequestError(),
+			}
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		}
 		return
 	}
 
 	// We should only have psubs and only 1 per result.
-	// FIXME(dlc) - Should we respond here with NoResponders or error?
 	if len(rr.psubs) != 1 {
 		s.Warnf("Malformed JetStream API Request: [%s] %q", subject, rmsg)
+		if c.kind == CLIENT || c.kind == LEAF {
+			ci, acc, _, _, _ := s.getRequestInfo(c, rmsg)
+			var resp = ApiResponse{
+				Type:  JSApiSystemResponseType,
+				Error: NewJSBadRequestError(),
+			}
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		}
 		return
 	}
 	jsub := rr.psubs[0]
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 6168a2db..bc1b5e28 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -5694,25 +5694,18 @@ func groupName(prefix string, peers []string, storage StorageType) string {
 // returns stream count for this tier as well as applicable reservation size (not including reservations for cfg)
 // jetStream read lock should be held
 func tieredStreamAndReservationCount(asa map[string]*streamAssignment, tier string, cfg *StreamConfig) (int, int64) {
-	numStreams := len(asa)
-	reservation := int64(0)
-	if tier == _EMPTY_ {
-		for _, sa := range asa {
-			if sa.Config.MaxBytes > 0 && sa.Config.Name != cfg.Name {
-				if sa.Config.Storage == cfg.Storage {
-					reservation += (int64(sa.Config.Replicas) * sa.Config.MaxBytes)
-				}
-			}
-		}
-	} else {
-		numStreams = 0
-		for _, sa := range asa {
-			if isSameTier(sa.Config, cfg) {
-				numStreams++
-				if sa.Config.MaxBytes > 0 {
-					if sa.Config.Storage == cfg.Storage && sa.Config.Name != cfg.Name {
-						reservation += (int64(sa.Config.Replicas) * sa.Config.MaxBytes)
-					}
+	var numStreams int
+	var reservation int64
+	for _, sa := range asa {
+		if tier == _EMPTY_ || isSameTier(sa.Config, cfg) {
+			numStreams++
+			if sa.Config.MaxBytes > 0 && sa.Config.Storage == cfg.Storage && sa.Config.Name != cfg.Name {
+				// If tier is empty, all storage is flat and we should adjust for replicas.
+				// Otherwise if tiered, storage replication already taken into consideration.
+				if tier == _EMPTY_ && cfg.Replicas > 1 {
+					reservation += sa.Config.MaxBytes * int64(cfg.Replicas)
+				} else {
+					reservation += sa.Config.MaxBytes
 				}
 			}
 		}
diff --git a/server/mqtt.go b/server/mqtt.go
index 783af64b..730d1404 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -217,7 +217,7 @@ var (
 	errMQTTEmptyUsername            = errors.New("empty user name not allowed")
 	errMQTTTopicIsEmpty             = errors.New("topic cannot be empty")
 	errMQTTPacketIdentifierIsZero   = errors.New("packet identifier cannot be 0")
-	errMQTTUnsupportedCharacters    = errors.New("characters ' ' and '.' not supported for MQTT topics")
+	errMQTTUnsupportedCharacters    = errors.New("character ' ' not supported for MQTT topics")
 	errMQTTInvalidSession           = errors.New("invalid MQTT session")
 )
 
diff --git a/server/raft.go b/server/raft.go
index afd30e3b..1180979a 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -31,6 +31,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/nats-io/nats-server/v2/internal/fastrand"
+
 	"github.com/minio/highwayhash"
 )
 
@@ -202,8 +204,6 @@ type raft struct {
 	stepdown *ipQueue[string]               // Stepdown requests
 	leadc    chan bool                      // Leader changes
 	quit     chan struct{}                  // Raft group shutdown
-
-	prand *rand.Rand // Random generator, used to generate inboxes for instance
 }
 
 // cacthupState structure that holds our subscription, and catchup term and index
@@ -349,7 +349,6 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 	sq := s.sys.sq
 	sacc := s.sys.account
 	hash := s.sys.shash
-	pub := s.info.ID
 	s.mu.RUnlock()
 
 	// Do this here to process error quicker.
@@ -362,12 +361,6 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 	}
 
 	qpfx := fmt.Sprintf("[ACC:%s] RAFT '%s' ", accName, cfg.Name)
-	rsrc := time.Now().UnixNano()
-	if len(pub) >= 32 {
-		if h, _ := highwayhash.New64([]byte(pub[:32])); h != nil {
-			rsrc += int64(h.Sum64())
-		}
-	}
 	n := &raft{
 		created:  time.Now(),
 		id:       hash[:idLen],
@@ -399,7 +392,6 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 		leadc:    make(chan bool, 1),
 		observer: cfg.Observer,
 		extSt:    ps.domainExt,
-		prand:    rand.New(rand.NewSource(rsrc)),
 	}
 	n.c.registerWithAccount(sacc)
 
@@ -1685,7 +1677,7 @@ const (
 // Lock should be held (due to use of random generator)
 func (n *raft) newCatchupInbox() string {
 	var b [replySuffixLen]byte
-	rn := n.prand.Int63()
+	rn := fastrand.Uint64()
 	for i, l := 0, rn; i < len(b); i++ {
 		b[i] = digits[l%base]
 		l /= base
@@ -1695,7 +1687,7 @@ func (n *raft) newCatchupInbox() string {
 
 func (n *raft) newInbox() string {
 	var b [replySuffixLen]byte
-	rn := n.prand.Int63()
+	rn := fastrand.Uint64()
 	for i, l := 0, rn; i < len(b); i++ {
 		b[i] = digits[l%base]
 		l /= base
diff --git a/server/stream.go b/server/stream.go
index 7317837b..bf0139b2 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -369,9 +369,7 @@ type ddentry struct {
 }
 
 // Replicas Range
-const (
-	StreamMaxReplicas = 5
-)
+const StreamMaxReplicas = 5
 
 // AddStream adds a stream for the given account.
 func (a *Account) addStream(config *StreamConfig) (*stream, error) {
@@ -1255,6 +1253,7 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		}
 		return exists, cfg
 	}
+
 	hasStream := func(streamName string) (bool, int32, []string) {
 		exists, cfg := getStream(streamName)
 		return exists, cfg.MaxMsgSize, cfg.Subjects
@@ -1633,13 +1632,7 @@ func (jsa *jsAccount) configUpdateCheck(old, new *StreamConfig, s *Server) (*Str
 
 	// Save the user configured MaxBytes.
 	newMaxBytes := cfg.MaxBytes
-
 	maxBytesOffset := int64(0)
-	if old.MaxBytes > 0 {
-		if excessRep := cfg.Replicas - old.Replicas; excessRep > 0 {
-			maxBytesOffset = old.MaxBytes * int64(excessRep)
-		}
-	}
 
 	// We temporarily set cfg.MaxBytes to maxBytesDiff because checkAllLimits
 	// adds cfg.MaxBytes to the current reserved limit and checks if we've gone
@@ -1671,7 +1664,11 @@ func (jsa *jsAccount) configUpdateCheck(old, new *StreamConfig, s *Server) (*Str
 		_, reserved = tieredStreamAndReservationCount(js.cluster.streams[acc.Name], tier, &cfg)
 	}
 	// reservation does not account for this stream, hence add the old value
-	reserved += int64(old.Replicas) * old.MaxBytes
+	if tier == _EMPTY_ && old.Replicas > 1 {
+		reserved += old.MaxBytes * int64(old.Replicas)
+	} else {
+		reserved += old.MaxBytes
+	}
 	if err := js.checkAllLimits(&selected, &cfg, reserved, maxBytesOffset); err != nil {
 		return nil, err
 	}


Test patch:
diff --git a/internal/fastrand/fastrand_test.go b/internal/fastrand/fastrand_test.go
new file mode 100644
index 00000000..9a828655
--- /dev/null
+++ b/internal/fastrand/fastrand_test.go
@@ -0,0 +1,72 @@
+// Copyright 2020-23 The LevelDB-Go, Pebble and NATS Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package fastrand
+
+import (
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+)
+
+type defaultRand struct {
+	mu  sync.Mutex
+	src rand.Source64
+}
+
+func newDefaultRand() *defaultRand {
+	r := &defaultRand{
+		src: rand.New(rand.NewSource(time.Now().UnixNano())),
+	}
+	return r
+}
+
+func (r *defaultRand) Uint32() uint32 {
+	r.mu.Lock()
+	i := uint32(r.src.Uint64())
+	r.mu.Unlock()
+	return i
+}
+
+func (r *defaultRand) Uint64() uint64 {
+	r.mu.Lock()
+	i := uint64(r.src.Uint64())
+	r.mu.Unlock()
+	return i
+}
+
+func BenchmarkFastRand32(b *testing.B) {
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			Uint32()
+		}
+	})
+}
+
+func BenchmarkFastRand64(b *testing.B) {
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			Uint64()
+		}
+	})
+}
+
+func BenchmarkDefaultRand32(b *testing.B) {
+	r := newDefaultRand()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			r.Uint32()
+		}
+	})
+}
+
+func BenchmarkDefaultRand64(b *testing.B) {
+	r := newDefaultRand()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			r.Uint64()
+		}
+	})
+}
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 7856562b..427e137c 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -6286,6 +6286,38 @@ func TestFileStoreCorruptPSIMOnDisk(t *testing.T) {
 	require_True(t, bytes.Equal(sm.msg, []byte("XYZ")))
 }
 
+func TestFileStorePurgeExBufPool(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd, BlockSize: 1024},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := bytes.Repeat([]byte("ABC"), 33) // ~100bytes
+	for i := 0; i < 1000; i++ {
+		fs.StoreMsg("foo.foo", nil, msg)
+		fs.StoreMsg("foo.bar", nil, msg)
+	}
+
+	p, err := fs.PurgeEx("foo.bar", 1, 0)
+	require_NoError(t, err)
+	require_Equal(t, p, 1000)
+
+	// Now make sure we do not have all of the msg blocks cache's loaded.
+	var loaded int
+	fs.mu.RLock()
+	for _, mb := range fs.blks {
+		mb.mu.RLock()
+		if mb.cacheAlreadyLoaded() {
+			loaded++
+		}
+		mb.mu.RUnlock()
+	}
+	fs.mu.RUnlock()
+	require_Equal(t, loaded, 1)
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/jetstream_cluster_2_test.go b/server/jetstream_cluster_2_test.go
index bfcd7e39..7a6f60e6 100644
--- a/server/jetstream_cluster_2_test.go
+++ b/server/jetstream_cluster_2_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020-2022 The NATS Authors
+// Copyright 2020-2023 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -23,6 +23,7 @@ import (
 	"encoding/binary"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math/rand"
 	"os"
@@ -3642,14 +3643,15 @@ func TestJetStreamClusterAccountReservations(t *testing.T) {
 	defer nc.Close()
 	accMax := 3
 
+	errResources := errors.New("nats: insufficient storage resources available")
+
 	test := func(t *testing.T, replica int) {
 		mb := int64((1+accMax)-replica) * 1024 * 1024 * 1024 // GB, corrected for replication factor
 		_, err := js.AddStream(&nats.StreamConfig{Name: "S1", Subjects: []string{"s1"}, MaxBytes: mb, Replicas: replica})
 		require_NoError(t, err)
 
 		_, err = js.AddStream(&nats.StreamConfig{Name: "S2", Subjects: []string{"s2"}, MaxBytes: 1024, Replicas: replica})
-		require_Error(t, err)
-		require_Equal(t, err.Error(), "nats: insufficient storage resources available")
+		require_Error(t, err, errResources)
 
 		_, err = js.UpdateStream(&nats.StreamConfig{Name: "S1", Subjects: []string{"s1"}, MaxBytes: mb / 2, Replicas: replica})
 		require_NoError(t, err)
@@ -3658,12 +3660,10 @@ func TestJetStreamClusterAccountReservations(t *testing.T) {
 		require_NoError(t, err)
 
 		_, err = js.AddStream(&nats.StreamConfig{Name: "S3", Subjects: []string{"s3"}, MaxBytes: 1024, Replicas: replica})
-		require_Error(t, err)
-		require_Equal(t, err.Error(), "nats: insufficient storage resources available")
+		require_Error(t, err, errResources)
 
 		_, err = js.UpdateStream(&nats.StreamConfig{Name: "S2", Subjects: []string{"s2"}, MaxBytes: mb/2 + 1, Replicas: replica})
-		require_Error(t, err)
-		require_Equal(t, err.Error(), "nats: insufficient storage resources available")
+		require_Error(t, err, errResources)
 
 		require_NoError(t, js.DeleteStream("S1"))
 		require_NoError(t, js.DeleteStream("S2"))
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index f5a15d72..5662832c 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -5893,3 +5893,31 @@ func TestJetStreamClusterStreamLimitsOnScaleUpAndMove(t *testing.T) {
 	})
 	require_Error(t, err, errors.New("insufficient storage resources"))
 }
+
+func TestJetStreamClusterAPIAccessViaSystemAccount(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	// Connect to system account.
+	nc, js := jsClientConnect(t, c.randomServer(), nats.UserInfo("admin", "s3cr3t!"))
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "TEST"})
+	require_Error(t, err, NewJSNotEnabledForAccountError())
+
+	// Make sure same behavior swith single server.
+	tmpl := `
+		listen: 127.0.0.1:-1
+		jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+		accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, t.TempDir())))
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc, js = jsClientConnect(t, s, nats.UserInfo("admin", "s3cr3t!"))
+	defer nc.Close()
+
+	_, err = js.AddStream(&nats.StreamConfig{Name: "TEST"})
+	require_Error(t, err, NewJSNotEnabledForAccountError())
+}
diff --git a/server/jetstream_jwt_test.go b/server/jetstream_jwt_test.go
index aee469f9..e487053b 100644
--- a/server/jetstream_jwt_test.go
+++ b/server/jetstream_jwt_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020-2022 The NATS Authors
+// Copyright 2020-2023 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -576,7 +576,7 @@ func TestJetStreamJWTClusteredTiersChange(t *testing.T) {
 	accClaim.Limits.JetStreamTieredLimits["R1"] = jwt.JetStreamLimits{
 		DiskStorage: 1000, MemoryStorage: 0, Consumer: 1, Streams: 1}
 	accClaim.Limits.JetStreamTieredLimits["R3"] = jwt.JetStreamLimits{
-		DiskStorage: 1500, MemoryStorage: 0, Consumer: 1, Streams: 1}
+		DiskStorage: 500, MemoryStorage: 0, Consumer: 1, Streams: 1}
 	accJwt1 := encodeClaim(t, accClaim, aExpPub)
 	accCreds := newUser(t, accKp)
 	start := time.Now()
@@ -622,12 +622,11 @@ func TestJetStreamJWTClusteredTiersChange(t *testing.T) {
 
 	cfg.Replicas = 3
 	_, err = js.UpdateStream(cfg)
-	require_Error(t, err)
-	require_Equal(t, err.Error(), "nats: insufficient storage resources available")
+	require_Error(t, err, errors.New("nats: insufficient storage resources available"))
 
 	time.Sleep(time.Second - time.Since(start)) // make sure the time stamp changes
 	accClaim.Limits.JetStreamTieredLimits["R3"] = jwt.JetStreamLimits{
-		DiskStorage: 3000, MemoryStorage: 0, Consumer: 1, Streams: 1}
+		DiskStorage: 1000, MemoryStorage: 0, Consumer: 1, Streams: 1}
 	accJwt2 := encodeClaim(t, accClaim, aExpPub)
 
 	updateJwt(t, c.randomServer().ClientURL(), sysCreds, accJwt2, 3)
@@ -1362,3 +1361,102 @@ func TestJetStreamJWTHAStorageLimitsAndAccounting(t *testing.T) {
 	delta = maxMemStorage - int64(si.State.Bytes)
 	require_True(t, int(delta) < len(msg))
 }
+
+func TestJetStreamJWTHAStorageLimitsOnScaleAndUpdate(t *testing.T) {
+	sysKp, syspub := createKey(t)
+	sysJwt := encodeClaim(t, jwt.NewAccountClaims(syspub), syspub)
+	newUser(t, sysKp)
+
+	maxFileStorage := int64(5 * 1024 * 1024)
+	maxMemStorage := int64(1 * 1024 * 1024)
+
+	accKp, aExpPub := createKey(t)
+	accClaim := jwt.NewAccountClaims(aExpPub)
+	accClaim.Name = "acc"
+	accClaim.Limits.JetStreamTieredLimits["R3"] = jwt.JetStreamLimits{DiskStorage: maxFileStorage, MemoryStorage: maxMemStorage}
+	accClaim.Limits.JetStreamTieredLimits["R1"] = jwt.JetStreamLimits{DiskStorage: maxFileStorage, MemoryStorage: maxMemStorage}
+
+	accJwt := encodeClaim(t, accClaim, aExpPub)
+	accCreds := newUser(t, accKp)
+	tmlp := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+		leaf { listen: 127.0.0.1:-1 }
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+	` + fmt.Sprintf(`
+		operator: %s
+		system_account: %s
+		resolver = MEMORY
+		resolver_preload = {
+			%s : %s
+			%s : %s
+		}
+	`, ojwt, syspub, syspub, sysJwt, aExpPub, accJwt)
+
+	c := createJetStreamClusterWithTemplate(t, tmlp, "cluster", 3)
+	defer c.shutdown()
+
+	nc := natsConnect(t, c.randomServer().ClientURL(), nats.UserCredentials(accCreds))
+	defer nc.Close()
+
+	js, err := nc.JetStream()
+	require_NoError(t, err)
+
+	// Test max bytes first.
+	_, err = js.AddStream(&nats.StreamConfig{Name: "TEST", Replicas: 3, MaxBytes: maxFileStorage, Subjects: []string{"foo"}})
+	require_NoError(t, err)
+	// Now delete
+	require_NoError(t, js.DeleteStream("TEST"))
+	// Now do 5 1MB streams.
+	for i := 1; i <= 5; i++ {
+		sname := fmt.Sprintf("TEST%d", i)
+		_, err = js.AddStream(&nats.StreamConfig{Name: sname, Replicas: 3, MaxBytes: 1 * 1024 * 1024})
+		require_NoError(t, err)
+	}
+	// Should fail.
+	_, err = js.AddStream(&nats.StreamConfig{Name: "TEST6", Replicas: 3, MaxBytes: 1 * 1024 * 1024})
+	require_Error(t, err, errors.New("insufficient storage resources"))
+
+	// Update Test1 and Test2 to smaller reservations.
+	_, err = js.UpdateStream(&nats.StreamConfig{Name: "TEST1", Replicas: 3, MaxBytes: 512 * 1024})
+	require_NoError(t, err)
+	_, err = js.UpdateStream(&nats.StreamConfig{Name: "TEST2", Replicas: 3, MaxBytes: 512 * 1024})
+	require_NoError(t, err)
+	// Now make sure TEST6 succeeds.
+	_, err = js.AddStream(&nats.StreamConfig{Name: "TEST6", Replicas: 3, MaxBytes: 1 * 1024 * 1024})
+	require_NoError(t, err)
+	// Now delete the R3 version.
+	require_NoError(t, js.DeleteStream("TEST6"))
+	// Now do R1 version and then we will scale up.
+	_, err = js.AddStream(&nats.StreamConfig{Name: "TEST6", Replicas: 1, MaxBytes: 1 * 1024 * 1024})
+	require_NoError(t, err)
+	// Now make sure scale up works.
+	_, err = js.UpdateStream(&nats.StreamConfig{Name: "TEST6", Replicas: 3, MaxBytes: 1 * 1024 * 1024})
+	require_NoError(t, err)
+	// Add in a few more streams to check reserved reporting in account info.
+	_, err = js.AddStream(&nats.StreamConfig{Name: "TEST7", Replicas: 1, MaxBytes: 2 * 1024 * 1024})
+	require_NoError(t, err)
+	_, err = js.AddStream(&nats.StreamConfig{Name: "TEST8", Replicas: 1, MaxBytes: 256 * 1024, Storage: nats.MemoryStorage})
+	require_NoError(t, err)
+	_, err = js.AddStream(&nats.StreamConfig{Name: "TEST9", Replicas: 3, MaxBytes: 22 * 1024, Storage: nats.MemoryStorage})
+	require_NoError(t, err)
+
+	// Now make sure we report reserved correctly.
+	// Do this direct to server since client does not support it yet.
+	var info JSApiAccountInfoResponse
+	resp, err := nc.Request("$JS.API.INFO", nil, time.Second)
+	require_NoError(t, err)
+	require_NoError(t, json.Unmarshal(resp.Data, &info))
+	stats := info.JetStreamAccountStats
+	r1, r3 := stats.Tiers["R1"], stats.Tiers["R3"]
+
+	require_Equal(t, r1.ReservedMemory, 256*1024)   // TEST8
+	require_Equal(t, r1.ReservedStore, 2*1024*1024) // TEST7
+	require_Equal(t, r3.ReservedMemory, 22*1024)    // TEST9
+	require_Equal(t, r3.ReservedStore, 5*1024*1024) // TEST1-TEST6
+}