Repository: nats-io/nats-server
Pull request #3965 (2023-03-15 09:41:20)

Cherry-pick #3733 (Refactor outbound queues, remove dynamic sizing, add buffer reuse)
This brings #3733 forward from `dev` into `main`, to go into the next release.

Signed-off-by: Neil Twigg <neil@nats.io>


Patch:
diff --git a/.github/actions/nightly-release/action.yaml b/.github/actions/nightly-release/action.yaml
index 7fab7dd6..a2376354 100644
--- a/.github/actions/nightly-release/action.yaml
+++ b/.github/actions/nightly-release/action.yaml
@@ -10,18 +10,6 @@ inputs:
     description: The label to use for built images
     required: true
 
-  hub_username:
-    description: Docker hub username
-    required: true
-
-  hub_password:
-    description: Docker hub password
-    required: true
-
-  workdir:
-    description: The working directory for actions requiring it
-    required: true
-
 runs:
   using: composite
   steps:
@@ -32,26 +20,27 @@ runs:
 
     - name: goreleaser
       uses: goreleaser/goreleaser-action@v3
+      env:
+        LABEL:
       with:
-        workdir: "${{ inputs.workdir }}"
         version: latest
         args: release --snapshot --config .goreleaser-nightly.yml
 
     - name: images
-      shell: bash
-      run: docker images
-
-    - name: docker_login
-      shell: bash
-      run: docker login -u "${{ inputs.hub_username }}" -p "${{ inputs.hub_password }}"
-
-    - name: docker_push
       shell: bash
       run: |
-        NDATE=$(date +%Y%m%d)
-    
-        docker tag synadia/nats-server:nightly-${NDATE} synadia/nats-server:${{ inputs.label }}-${NDATE}
-        docker tag synadia/nats-server:nightly-${NDATE} synadia/nats-server:${{ inputs.label }}
-    
-        docker push synadia/nats-server:${{ inputs.label }}-${NDATE}
-        docker push synadia/nats-server:${{ inputs.label }}
+        docker images
+
+        - name: docker_login
+          run: |
+            docker login -u "${{ secrets.DOCKER_USERNAME }}" -p "${{ secrets.DOCKER_PASSWORD }}"
+
+        - name: docker_push
+          run: |
+            NDATE=$(date +%Y%m%d)
+        
+            docker tag synadia/nats-server:nightly-${NDATE} synadia/nats-server:${{ inputs.label }}-${NDATE}
+            docker tag synadia/nats-server:nightly-${NDATE} synadia/nats-server:${{ inputs.label }}
+        
+            docker push synadia/nats-server:${{ inputs.label }}-${NDATE}
+            docker push synadia/nats-server:${{ inputs.label }}
diff --git a/.github/workflows/nightly.yaml b/.github/workflows/nightly.yaml
index 19dbd475..8944880c 100644
--- a/.github/workflows/nightly.yaml
+++ b/.github/workflows/nightly.yaml
@@ -1,4 +1,4 @@
-name: "NATS Server Nightly: DEV"
+name: NATS Server Nightly
 on:
   workflow_dispatch: {}
 
@@ -9,17 +9,39 @@ on:
 jobs:
   nightly_release:
     runs-on: ubuntu-latest
+
+    env:
+      GOPATH: /home/runner/work/nats-server
+      GO111MODULE: "on"
+
     steps:
       - name: Checkout code
-        uses: actions/checkout@v3
+        uses: actions/checkout@v1
         with:
           path: src/github.com/nats-io/nats-server
           ref: dev
 
-      - uses: ./src/github.com/nats-io/nats-server/.github/actions/nightly-release
+      - name: Set up Go
+        uses: actions/setup-go@v3
+        with:
+          go-version: 1.19
+
+      - name: goreleaser
+        uses: goreleaser/goreleaser-action@v3
         with:
-          go: "1.19"
-          workdir: src/github.com/nats-io/nats-server
-          label: nightly
-          hub_username: "${{ secrets.DOCKER_USERNAME }}"
-          hub_password: "${{ secrets.DOCKER_PASSWORD }}"
\ No newline at end of file
+          version: latest
+          args: release --snapshot --config .goreleaser-nightly.yml
+
+      - name: images
+        run: |
+          docker images
+
+      - name: docker_login
+        run: |
+          docker login -u "${{ secrets.DOCKER_USERNAME }}" -p "${{ secrets.DOCKER_PASSWORD }}"
+
+      - name: docker_push
+        run: |
+          NDATE=$(date +%Y%m%d)
+          docker push synadia/nats-server:nightly-${NDATE}
+          docker push synadia/nats-server:nightly
diff --git a/.github/workflows/rc_nightly.yaml b/.github/workflows/rc_nightly.yaml
index 9dc68f7d..bb8a445a 100644
--- a/.github/workflows/rc_nightly.yaml
+++ b/.github/workflows/rc_nightly.yaml
@@ -1,4 +1,4 @@
-name: "NATS Server Nightly: MAIN"
+name: NATS Server Nightly MAIN
 on:
   workflow_dispatch: {}
 
@@ -16,10 +16,7 @@ jobs:
           path: src/github.com/nats-io/nats-server
           ref: main
 
-      - uses: ./src/github.com/nats-io/nats-server/.github/actions/nightly-release
+      - uses: src/github.com/nats-io/nats-server/.github/actions/nightly-release
         with:
           go: "1.19"
-          workdir: src/github.com/nats-io/nats-server
           label: nightly-main
-          hub_username: "${{ secrets.DOCKER_USERNAME }}"
-          hub_password: "${{ secrets.DOCKER_PASSWORD }}"
\ No newline at end of file
diff --git a/.goreleaser.yml b/.goreleaser.yml
index b1599b43..9b0e1aca 100644
--- a/.goreleaser.yml
+++ b/.goreleaser.yml
@@ -11,10 +11,8 @@ changelog:
   skip: true
 
 builds:
-- main: .
+- main: ./main.go
   binary: nats-server
-  flags:
-  - -trimpath
   ldflags:
     - -w -X github.com/nats-io/nats-server/v2/server.gitCommit={{.ShortCommit}}
   env:
diff --git a/main.go b/main.go
index 767719fc..53652c6e 100644
--- a/main.go
+++ b/main.go
@@ -134,6 +134,8 @@ func main() {
 		s.Warnf("Failed to set GOMAXPROCS: %v", err)
 	} else {
 		defer undo()
+		// Reset these from the snapshots from init for monitor.go
+		server.SnapshotMonitorInfo()
 	}
 
 	s.WaitForShutdown()
diff --git a/server/client.go b/server/client.go
index 2218fc6d..25d8f0be 100644
--- a/server/client.go
+++ b/server/client.go
@@ -293,13 +293,9 @@ type pinfo struct {
 
 // outbound holds pending data for a socket.
 type outbound struct {
-	p   []byte        // Primary write buffer
-	s   []byte        // Secondary for use post flush
-	nb  net.Buffers   // net.Buffers for writev IO
-	sz  int32         // limit size per []byte, uses variable BufSize constants, start, min, max.
-	sws int32         // Number of short writes, used for dynamic resizing.
+	nb  net.Buffers   // Pending buffers for send, each has fixed capacity as per nbPool below.
+	wnb net.Buffers   // Working copy of "nb", reused on each flushOutbound call, partial writes may leave entries here for next iteration.
 	pb  int64         // Total pending/queued bytes.
-	pm  int32         // Total pending/queued messages.
 	fsp int32         // Flush signals that are pending per producer from readLoop's pcd.
 	sg  *sync.Cond    // To signal writeLoop that there is data to flush.
 	wdl time.Duration // Snapshot of write deadline.
@@ -308,6 +304,37 @@ type outbound struct {
 	stc chan struct{} // Stall chan we create to slow down producers on overrun, e.g. fan-in.
 }
 
+const nbPoolSizeSmall = 4096  // Underlying array size of small buffer
+const nbPoolSizeLarge = 65536 // Underlying array size of large buffer
+
+var nbPoolSmall = &sync.Pool{
+	New: func() any {
+		b := [nbPoolSizeSmall]byte{}
+		return &b
+	},
+}
+
+var nbPoolLarge = &sync.Pool{
+	New: func() any {
+		b := [nbPoolSizeLarge]byte{}
+		return &b
+	},
+}
+
+func nbPoolPut(b []byte) {
+	switch cap(b) {
+	case nbPoolSizeSmall:
+		b := (*[nbPoolSizeSmall]byte)(b[0:nbPoolSizeSmall])
+		nbPoolSmall.Put(b)
+	case nbPoolSizeLarge:
+		b := (*[nbPoolSizeLarge]byte)(b[0:nbPoolSizeLarge])
+		nbPoolLarge.Put(b)
+	default:
+		// Ignore frames that are the wrong size, this might happen
+		// with WebSocket/MQTT messages as they are framed
+	}
+}
+
 type perm struct {
 	allow *Sublist
 	deny  *Sublist
@@ -584,7 +611,6 @@ func (c *client) initClient() {
 	c.cid = atomic.AddUint64(&s.gcid, 1)
 
 	// Outbound data structure setup
-	c.out.sz = startBufSize
 	c.out.sg = sync.NewCond(&(c.mu))
 	opts := s.getOpts()
 	// Snapshots to avoid mutex access in fast paths.
@@ -1344,11 +1370,6 @@ func (c *client) collapsePtoNB() (net.Buffers, int64) {
 	if c.isWebsocket() {
 		return c.wsCollapsePtoNB()
 	}
-	if c.out.p != nil {
-		p := c.out.p
-		c.out.p = nil
-		return append(c.out.nb, p), c.out.pb
-	}
 	return c.out.nb, c.out.pb
 }
 
@@ -1359,9 +1380,6 @@ func (c *client) handlePartialWrite(pnb net.Buffers) {
 		c.ws.frames = append(pnb, c.ws.frames...)
 		return
 	}
-	nb, _ := c.collapsePtoNB()
-	// The partial needs to be first, so append nb to pnb
-	c.out.nb = append(pnb, nb...)
 }
 
 // flushOutbound will flush outbound buffer to a client.
@@ -1385,26 +1403,37 @@ func (c *client) flushOutbound() bool {
 		return true // true because no need to queue a signal.
 	}
 
-	// Place primary on nb, assign primary to secondary, nil out nb and secondary.
-	nb, attempted := c.collapsePtoNB()
-	c.out.p, c.out.nb, c.out.s = c.out.s, nil, nil
-	if nb == nil {
-		return true
-	}
-
-	// For selecting primary replacement.
-	cnb := nb
-	var lfs int
-	if len(cnb) > 0 {
-		lfs = len(cnb[0])
-	}
+	// In the case of a normal socket connection, "collapsed" is just a ref
+	// to "nb". In the case of WebSockets, additional framing is added to
+	// anything that is waiting in "nb". Also keep a note of how many bytes
+	// were queued before we release the mutex.
+	collapsed, attempted := c.collapsePtoNB()
+
+	// Frustratingly, (net.Buffers).WriteTo() modifies the receiver so we
+	// can't work on "nb" directly — while the mutex is unlocked during IO,
+	// something else might call queueOutbound and modify it. So instead we
+	// need a working copy — we'll operate on "wnb" instead. Note that in
+	// the case of a partial write, "wnb" may have remaining data from the
+	// previous write, and in the case of WebSockets, that data may already
+	// be framed, so we are careful not to re-frame "wnb" here. Instead we
+	// will just frame up "nb" and append it onto whatever is left on "wnb".
+	// "nb" will be reset back to its starting position so it can be modified
+	// safely by queueOutbound calls.
+	c.out.wnb = append(c.out.wnb, collapsed...)
+	orig := append(net.Buffers{}, c.out.wnb...)
+	c.out.nb = c.out.nb[:0]
+
+	// Since WriteTo is lopping things off the beginning, we need to remember
+	// the start position of the underlying array so that we can get back to it.
+	// Otherwise we'll always "slide forward" and that will result in reallocs.
+	startOfWnb := c.out.wnb[0:]
 
 	// In case it goes away after releasing the lock.
 	nc := c.nc
-	apm := c.out.pm
 
 	// Capture this (we change the value in some tests)
 	wdl := c.out.wdl
+
 	// Do NOT hold lock during actual IO.
 	c.mu.Unlock()
 
@@ -1416,7 +1445,7 @@ func (c *client) flushOutbound() bool {
 	nc.SetWriteDeadline(start.Add(wdl))
 
 	// Actual write to the socket.
-	n, err := nb.WriteTo(nc)
+	n, err := c.out.wnb.WriteTo(nc)
 	nc.SetWriteDeadline(time.Time{})
 
 	lft := time.Since(start)
@@ -1424,11 +1453,35 @@ func (c *client) flushOutbound() bool {
 	// Re-acquire client lock.
 	c.mu.Lock()
 
+	// At this point, "wnb" has been mutated by WriteTo and any consumed
+	// buffers have been lopped off the beginning, so in order to return
+	// them to the pool, we need to look at the difference between "orig"
+	// and "wnb".
+	for i := 0; i < len(orig)-len(c.out.wnb); i++ {
+		nbPoolPut(orig[i])
+	}
+
+	// At this point it's possible that "nb" has been modified by another
+	// call to queueOutbound while the lock was released, so we'll leave
+	// those for the next iteration. Meanwhile it's possible that we only
+	// managed a partial write of "wnb", so we'll shift anything that
+	// remains up to the beginning of the array to prevent reallocating.
+	// Anything left in "wnb" has already been framed for WebSocket conns
+	// so leave them alone for the next call to flushOutbound.
+	c.out.wnb = append(startOfWnb[:0], c.out.wnb...)
+
+	// If we've written everything but the underlying array of our working
+	// buffer has grown excessively then free it — the GC will tidy it up
+	// and we can allocate a new one next time.
+	if len(c.out.wnb) == 0 && cap(c.out.wnb) > nbPoolSizeLarge*8 {
+		c.out.wnb = nil
+	}
+
 	// Ignore ErrShortWrite errors, they will be handled as partials.
 	if err != nil && err != io.ErrShortWrite {
 		// Handle timeout error (slow consumer) differently
 		if ne, ok := err.(net.Error); ok && ne.Timeout() {
-			if closed := c.handleWriteTimeout(n, attempted, len(cnb)); closed {
+			if closed := c.handleWriteTimeout(n, attempted, len(c.out.nb)); closed {
 				return true
 			}
 		} else {
@@ -1452,43 +1505,11 @@ func (c *client) flushOutbound() bool {
 	if c.isWebsocket() {
 		c.ws.fs -= n
 	}
-	c.out.pm -= apm // FIXME(dlc) - this will not be totally accurate on partials.
 
 	// Check for partial writes
 	// TODO(dlc) - zero write with no error will cause lost message and the writeloop to spin.
 	if n != attempted && n > 0 {
-		c.handlePartialWrite(nb)
-	} else if int32(n) >= c.out.sz {
-		c.out.sws = 0
-	}
-
-	// Adjust based on what we wrote plus any pending.
-	pt := n + c.out.pb
-
-	// Adjust sz as needed downward, keeping power of 2.
-	// We do this at a slower rate.
-	if pt < int64(c.out.sz) && c.out.sz > minBufSize {
-		c.out.sws++
-		if c.out.sws > shortsToShrink {
-			c.out.sz >>= 1
-		}
-	}
-	// Adjust sz as needed upward, keeping power of 2.
-	if pt > int64(c.out.sz) && c.out.sz < maxBufSize {
-		c.out.sz <<= 1
-	}
-
-	// Check to see if we can reuse buffers.
-	if lfs != 0 && n >= int64(lfs) {
-		oldp := cnb[0][:0]
-		if cap(oldp) >= int(c.out.sz) {
-			// Replace primary or secondary if they are nil, reusing same buffer.
-			if c.out.p == nil {
-				c.out.p = oldp
-			} else if c.out.s == nil || cap(c.out.s) < int(c.out.sz) {
-				c.out.s = oldp
-			}
-		}
+		c.handlePartialWrite(c.out.nb)
 	}
 
 	// Check that if there is still data to send and writeLoop is in wait,
@@ -1989,6 +2010,49 @@ func (c *client) queueOutbound(data []byte) {
 	// Add to pending bytes total.
 	c.out.pb += int64(len(data))
 
+	// Take a copy of the slice ref so that we can chop bits off the beginning
+	// without affecting the original "data" slice.
+	toBuffer := data
+
+	// All of the queued []byte have a fixed capacity, so if there's a []byte
+	// at the tail of the buffer list that isn't full yet, we should top that
+	// up first. This helps to ensure we aren't pulling more []bytes from the
+	// pool than we need to.
+	if len(c.out.nb) > 0 {
+		last := &c.out.nb[len(c.out.nb)-1]
+		if free := cap(*last) - len(*last); free > 0 {
+			if l := len(toBuffer); l < free {
+				free = l
+			}
+			*last = append(*last, toBuffer[:free]...)
+			toBuffer = toBuffer[free:]
+		}
+	}
+
+	// Now we can push the rest of the data into new []bytes from the pool
+	// in fixed size chunks. This ensures we don't go over the capacity of any
+	// of the buffers and end up reallocating.
+	for len(toBuffer) > 0 {
+		var new []byte
+		if len(c.out.nb) == 0 && len(toBuffer) <= nbPoolSizeSmall {
+			// If the buffer is empty, try to allocate a small buffer if the
+			// message will fit in it. This will help for cases like pings.
+			new = nbPoolSmall.Get().(*[nbPoolSizeSmall]byte)[:0]
+		} else {
+			// If "nb" isn't empty, default to large buffers in all cases as
+			// this means we are always coalescing future messages into
+			// larger buffers. Reduces the number of buffers into writev.
+			new = nbPoolLarge.Get().(*[nbPoolSizeLarge]byte)[:0]
+		}
+		l := len(toBuffer)
+		if c := cap(new); l > c {
+			l = c
+		}
+		new = append(new, toBuffer[:l]...)
+		c.out.nb = append(c.out.nb, new)
+		toBuffer = toBuffer[l:]
+	}
+
 	// Check for slow consumer via pending bytes limit.
 	// ok to return here, client is going away.
 	if c.kind == CLIENT && c.out.pb > c.out.mp {
@@ -2004,58 +2068,6 @@ func (c *client) queueOutbound(data []byte) {
 		return
 	}
 
-	if c.out.p == nil && len(data) < maxBufSize {
-		if c.out.sz == 0 {
-			c.out.sz = startBufSize
-		}
-		if c.out.s != nil && cap(c.out.s) >= int(c.out.sz) {
-			c.out.p = c.out.s
-			c.out.s = nil
-		} else {
-			// FIXME(dlc) - make power of 2 if less than maxBufSize?
-			c.out.p = make([]byte, 0, c.out.sz)
-		}
-	}
-	// Determine if we copy or reference
-	available := cap(c.out.p) - len(c.out.p)
-	if len(data) > available {
-		// We can't fit everything into existing primary, but message will
-		// fit in next one we allocate or utilize from the secondary.
-		// So copy what we can.
-		if available > 0 && len(data) < int(c.out.sz) {
-			c.out.p = append(c.out.p, data[:available]...)
-			data = data[available:]
-		}
-		// Put the primary on the nb if it has a payload
-		if len(c.out.p) > 0 {
-			c.out.nb = append(c.out.nb, c.out.p)
-			c.out.p = nil
-		}
-		// TODO: It was found with LeafNode and Websocket that referencing
-		// the data buffer when > maxBufSize would cause corruption
-		// (reproduced with small maxBufSize=10 and TestLeafNodeWSNoBufferCorruption).
-		// So always make a copy for now.
-
-		// We will copy to primary.
-		if c.out.p == nil {
-			// Grow here
-			if (c.out.sz << 1) <= maxBufSize {
-				c.out.sz <<= 1
-			}
-			if len(data) > int(c.out.sz) {
-				c.out.p = make([]byte, 0, len(data))
-			} else {
-				if c.out.s != nil && cap(c.out.s) >= int(c.out.sz) { // TODO(dlc) - Size mismatch?
-					c.out.p = c.out.s
-					c.out.s = nil
-				} else {
-					c.out.p = make([]byte, 0, c.out.sz)
-				}
-			}
-		}
-	}
-	c.out.p = append(c.out.p, data...)
-
 	// Check here if we should create a stall channel if we are falling behind.
 	// We do this here since if we wait for consumer's writeLoop it could be
 	// too late with large number of fan in producers.
@@ -3249,8 +3261,6 @@ func (c *client) deliverMsg(prodIsMQTT bool, sub *subscription, acc *Account, su
 		client.queueOutbound([]byte(CR_LF))
 	}
 
-	client.out.pm++
-
 	// If we are tracking dynamic publish permissions that track reply subjects,
 	// do that accounting here. We only look at client.replies which will be non-nil.
 	if client.replies != nil && len(reply) > 0 {
@@ -3265,7 +3275,7 @@ func (c *client) deliverMsg(prodIsMQTT bool, sub *subscription, acc *Account, su
 	// to intervene before this producer goes back to top of readloop. We are in the producer's
 	// readloop go routine at this point.
 	// FIXME(dlc) - We may call this alot, maybe suppress after first call?
-	if client.out.pm > 1 && client.out.pb > maxBufSize*2 {
+	if len(client.out.nb) != 0 {
 		client.flushSignal()
 	}
 
@@ -4672,7 +4682,10 @@ func (c *client) flushAndClose(minimalFlush bool) {
 		}
 		c.flushOutbound()
 	}
-	c.out.p, c.out.s = nil, nil
+	for i := range c.out.nb {
+		nbPoolPut(c.out.nb[i])
+	}
+	c.out.nb = nil
 
 	// Close the low level connection.
 	if c.nc != nil {
diff --git a/server/const.go b/server/const.go
index 26e634bb..52da7043 100644
--- a/server/const.go
+++ b/server/const.go
@@ -41,7 +41,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.9.16-RC.3"
+	VERSION = "2.9.16-beta"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index 23067214..9b892056 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2023 The NATS Authors
+// Copyright 2019-2022 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -2388,11 +2388,6 @@ func (o *consumer) sampleAck(sseq, dseq, dc uint64) {
 
 func (o *consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 	o.mu.Lock()
-	if o.closed {
-		o.mu.Unlock()
-		return
-	}
-
 	var sagap uint64
 	var needSignal bool
 
@@ -2500,25 +2495,20 @@ func (o *consumer) needAck(sseq uint64, subj string) bool {
 	var needAck bool
 	var asflr, osseq uint64
 	var pending map[uint64]*Pending
-
 	o.mu.RLock()
-	defer o.mu.RUnlock()
-
-	isFiltered := o.isFiltered()
-	if isFiltered && o.mset == nil {
-		return false
-	}
 
 	// Check if we are filtered, and if so check if this is even applicable to us.
-	if isFiltered {
+	if o.isFiltered() && o.mset != nil {
 		if subj == _EMPTY_ {
 			var svp StoreMsg
 			if _, err := o.mset.store.LoadMsg(sseq, &svp); err != nil {
+				o.mu.RUnlock()
 				return false
 			}
 			subj = svp.subj
 		}
 		if !o.isFilteredMatch(subj) {
+			o.mu.RUnlock()
 			return false
 		}
 	}
@@ -2528,12 +2518,15 @@ func (o *consumer) needAck(sseq uint64, subj string) bool {
 		pending = o.pending
 	} else {
 		if o.store == nil {
+			o.mu.RUnlock()
 			return false
 		}
 		state, err := o.store.BorrowState()
 		if err != nil || state == nil {
 			// Fall back to what we track internally for now.
-			return sseq > o.asflr && !o.isFiltered()
+			needAck := sseq > o.asflr && !o.isFiltered()
+			o.mu.RUnlock()
+			return needAck
 		}
 		// If loading state as here, the osseq is +1.
 		asflr, osseq, pending = state.AckFloor.Stream, state.Delivered.Stream+1, state.Pending
@@ -2552,6 +2545,7 @@ func (o *consumer) needAck(sseq uint64, subj string) bool {
 		}
 	}
 
+	o.mu.RUnlock()
 	return needAck
 }
 
@@ -3819,8 +3813,7 @@ func (o *consumer) checkPending() {
 	o.mu.RLock()
 	mset := o.mset
 	// On stop, mset and timer will be nil.
-	if o.closed || mset == nil || o.ptmr == nil {
-		stopAndClearTimer(&o.ptmr)
+	if mset == nil || o.ptmr == nil {
 		o.mu.RUnlock()
 		return
 	}
@@ -4357,15 +4350,14 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 		}
 
 		var rmseqs []uint64
-		mset.mu.Lock()
+		mset.mu.RLock()
 		for seq := start; seq <= stop; seq++ {
-			if mset.noInterest(seq, o) {
+			if !mset.checkInterest(seq, o) {
 				rmseqs = append(rmseqs, seq)
 			}
 		}
-		mset.mu.Unlock()
+		mset.mu.RUnlock()
 
-		// These can be removed.
 		for _, seq := range rmseqs {
 			mset.store.RemoveMsg(seq)
 		}
@@ -4377,7 +4369,7 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 			n.Delete()
 		} else {
 			// Try to install snapshot on clean exit
-			if o.store != nil && (o.retention != LimitsPolicy || n.NeedSnapshot()) {
+			if o.store != nil && n.NeedSnapshot() {
 				if snap, err := o.store.EncodedState(); err == nil {
 					n.InstallSnapshot(snap)
 				}
@@ -4563,44 +4555,3 @@ func (o *consumer) isMonitorRunning() bool {
 	defer o.mu.Unlock()
 	return o.inMonitor
 }
-
-// If we are a consumer of an interest or workqueue policy stream, process that state and make sure consistent.
-func (o *consumer) checkStateForInterestStream() {
-	o.mu.Lock()
-	// See if we need to process this update if our parent stream is not a limits policy stream.
-	mset := o.mset
-	shouldProcessState := mset != nil && o.retention != LimitsPolicy
-	if o.closed || !shouldProcessState {
-		o.mu.Unlock()
-		return
-	}
-	state, err := o.store.State()
-	o.mu.Unlock()
-
-	if err != nil {
-		return
-	}
-
-	// We should make sure to update the acks.
-	var ss StreamState
-	mset.store.FastState(&ss)
-
-	asflr := state.AckFloor.Stream
-	for seq := ss.FirstSeq; seq <= asflr; seq++ {
-		mset.ackMsg(o, seq)
-	}
-
-	o.mu.RLock()
-	// See if we need to process this update if our parent stream is not a limits policy stream.
-	state, _ = o.store.State()
-	o.mu.RUnlock()
-
-	// If we have pending, we will need to walk through to delivered in case we missed any of those acks as well.
-	if state != nil && len(state.Pending) > 0 {
-		for seq := state.AckFloor.Stream + 1; seq <= state.Delivered.Stream; seq++ {
-			if _, ok := state.Pending[seq]; !ok {
-				mset.ackMsg(o, seq)
-			}
-		}
-	}
-}
diff --git a/server/events.go b/server/events.go
index 0ff6a257..755cd0e6 100644
--- a/server/events.go
+++ b/server/events.go
@@ -22,7 +22,6 @@ import (
 	"fmt"
 	"math/rand"
 	"net/http"
-	"runtime"
 	"strconv"
 	"strings"
 	"sync"
@@ -604,7 +603,7 @@ func (s *Server) updateServerUsage(v *ServerStats) {
 	defer s.mu.Lock()
 	var vss int64
 	pse.ProcUsage(&v.CPU, &v.Mem, &vss)
-	v.Cores = runtime.NumCPU()
+	v.Cores = numCores
 }
 
 // Generate a route stat for our statz update.
diff --git a/server/filestore.go b/server/filestore.go
index bb09a230..d2067141 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -106,7 +106,6 @@ type psi struct {
 }
 
 type fileStore struct {
-	srv         *Server
 	mu          sync.RWMutex
 	state       StreamState
 	ld          *LostStreamData
@@ -312,11 +311,8 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	if err != nil {
 		return nil, fmt.Errorf("storage directory is not writable")
 	}
-
 	tmpfile.Close()
-	<-dios
 	os.Remove(tmpfile.Name())
-	dios <- struct{}{}
 
 	fs := &fileStore{
 		fcfg: fcfg,
@@ -377,12 +373,6 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	return fs, nil
 }
 
-func (fs *fileStore) registerServer(s *Server) {
-	fs.mu.Lock()
-	defer fs.mu.Unlock()
-	fs.srv = s
-}
-
 // Lock all existing message blocks.
 // Lock held on entry.
 func (fs *fileStore) lockAllMsgBlocks() {
@@ -775,7 +765,6 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint32) (*msgBlock, e
 	if ld, _ := mb.rebuildState(); ld != nil {
 		fs.addLostData(ld)
 	}
-
 	if mb.msgs > 0 && !mb.noTrack && fs.psim != nil {
 		fs.populateGlobalPerSubjectInfo(mb)
 		// Try to dump any state we needed on recovery.
@@ -1047,7 +1036,7 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, error) {
 		rl &^= hbit
 		dlen := int(rl) - msgHdrSize
 		// Do some quick sanity checks here.
-		if dlen < 0 || int(slen) > (dlen-8) || dlen > int(rl) || rl > rlBadThresh {
+		if dlen < 0 || int(slen) > dlen || dlen > int(rl) || rl > rlBadThresh {
 			truncate(index)
 			return gatherLost(lbuf - index), errBadMsg
 		}
@@ -1151,11 +1140,9 @@ func (fs *fileStore) recoverMsgs() error {
 
 	// Check for any left over purged messages.
 	pdir := filepath.Join(fs.fcfg.StoreDir, purgeDir)
-	<-dios
 	if _, err := os.Stat(pdir); err == nil {
 		os.RemoveAll(pdir)
 	}
-	dios <- struct{}{}
 
 	mdir := filepath.Join(fs.fcfg.StoreDir, msgDir)
 	fis, err := os.ReadDir(mdir)
@@ -2638,16 +2625,6 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
 	if secure && fs.prf != nil {
 		secure = false
 	}
-
-	if fs.state.Msgs == 0 {
-		var err = ErrStoreEOF
-		if seq <= fs.state.LastSeq {
-			err = ErrStoreMsgNotFound
-		}
-		fsUnlock()
-		return false, err
-	}
-
 	mb := fs.selectMsgBlock(seq)
 	if mb == nil {
 		var err = ErrStoreEOF
@@ -2660,8 +2637,8 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
 
 	mb.mu.Lock()
 
-	// See if we are closed or the sequence number is still relevant.
-	if mb.closed || seq < mb.first.seq {
+	// See if the sequence number is still relevant.
+	if seq < mb.first.seq {
 		mb.mu.Unlock()
 		fsUnlock()
 		return false, nil
@@ -2956,7 +2933,6 @@ func (mb *msgBlock) slotInfo(slot int) (uint32, uint32, bool, error) {
 	if mb.cache == nil || slot >= len(mb.cache.idx) {
 		return 0, 0, false, errPartialCache
 	}
-
 	bi := mb.cache.idx[slot]
 	ri, hashChecked := (bi &^ hbit), (bi&hbit) != 0
 
@@ -3935,12 +3911,11 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 		dlen := int(rl) - msgHdrSize
 
 		// Do some quick sanity checks here.
-		if dlen < 0 || int(slen) > dlen || dlen > int(rl) || index+rl > lbuf || rl > 32*1024*1024 {
+		if dlen < 0 || int(slen) > dlen || dlen > int(rl) || rl > 32*1024*1024 {
 			// This means something is off.
 			// TODO(dlc) - Add into bad list?
 			return errCorruptState
 		}
-
 		// Clear erase bit.
 		seq = seq &^ ebit
 		// Adjust if we guessed wrong.
@@ -4018,13 +3993,15 @@ func (mb *msgBlock) flushPendingMsgsLocked() (*LostStreamData, error) {
 
 	// Check if we need to encrypt.
 	if mb.bek != nil && lob > 0 {
-		// Need to leave original alone.
+		const rsz = 32 * 1024 // 32k
+		var rdst [rsz]byte
 		var dst []byte
-		if lob <= defaultLargeBlockSize {
-			dst = getMsgBlockBuf(lob)[:lob]
-		} else {
+		if lob > rsz {
 			dst = make([]byte, lob)
+		} else {
+			dst = rdst[:lob]
 		}
+		// Need to leave original alone.
 		mb.bek.XORKeyStream(dst, buf)
 		buf = dst
 	}
@@ -4781,9 +4758,6 @@ func (mb *msgBlock) writeIndexInfoLocked() error {
 		if err != nil {
 			return err
 		}
-		if fi, _ := ifd.Stat(); fi != nil {
-			mb.liwsz = fi.Size()
-		}
 		mb.ifd = ifd
 	}
 
@@ -5203,7 +5177,7 @@ func (fs *fileStore) purge(fseq uint64) (uint64, error) {
 // but not including the seq parameter.
 // Will return the number of purged messages.
 func (fs *fileStore) Compact(seq uint64) (uint64, error) {
-	if seq == 0 {
+	if seq == 0 || seq > fs.lastSeq() {
 		return fs.purge(seq)
 	}
 
@@ -5211,16 +5185,15 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 
 	// We have to delete interior messages.
 	fs.mu.Lock()
-	if lseq := fs.state.LastSeq; seq > lseq {
-		fs.mu.Unlock()
-		return fs.purge(seq)
-	}
-
 	smb := fs.selectMsgBlock(seq)
 	if smb == nil {
 		fs.mu.Unlock()
 		return 0, nil
 	}
+	if err := smb.loadMsgs(); err != nil {
+		fs.mu.Unlock()
+		return 0, err
+	}
 
 	// All msgblocks up to this one can be thrown away.
 	var deleted int
@@ -5247,12 +5220,7 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 	var isEmpty bool
 
 	smb.mu.Lock()
-	if smb.first.seq == seq {
-		isEmpty = smb.msgs == 0
-		goto SKIP
-	}
-
-	// Make sure we have the messages loaded.
+	// Since we loaded before we acquired our lock, double check here under lock that we have the messages loaded.
 	if smb.cacheNotLoaded() {
 		if err = smb.loadMsgsWithLock(); err != nil {
 			goto SKIP
@@ -5263,7 +5231,7 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 		if err == errDeletedMsg {
 			// Update dmap.
 			if len(smb.dmap) > 0 {
-				delete(smb.dmap, mseq)
+				delete(smb.dmap, seq)
 				if len(smb.dmap) == 0 {
 					smb.dmap = nil
 				}
@@ -5299,7 +5267,7 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 
 		// Check if we should reclaim the head space from this block.
 		// This will be optimistic only, so don't continue if we encounter any errors here.
-		if smb.rbytes > compactMinimum && smb.bytes*2 < smb.rbytes {
+		if smb.bytes*2 < smb.rbytes {
 			var moff uint32
 			moff, _, _, err = smb.slotInfo(int(smb.first.seq - smb.cache.fseq))
 			if err != nil || moff >= uint32(len(smb.cache.buf)) {
@@ -5333,13 +5301,13 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 	}
 
 SKIP:
+	smb.mu.Unlock()
+
 	if !isEmpty {
 		// Make sure to write out our index info.
-		smb.writeIndexInfoLocked()
+		smb.writeIndexInfo()
 	}
 
-	smb.mu.Unlock()
-
 	if deleted > 0 {
 		// Update block map.
 		if fs.bim != nil {
@@ -5363,7 +5331,7 @@ SKIP:
 	cb := fs.scb
 	fs.mu.Unlock()
 
-	if cb != nil && purged > 0 {
+	if cb != nil {
 		cb(-int64(purged), -int64(bytes), 0, _EMPTY_)
 	}
 
@@ -5372,6 +5340,7 @@ SKIP:
 
 // Will completely reset our store.
 func (fs *fileStore) reset() error {
+
 	fs.mu.Lock()
 	if fs.closed {
 		fs.mu.Unlock()
@@ -5385,12 +5354,14 @@ func (fs *fileStore) reset() error {
 	var purged, bytes uint64
 	cb := fs.scb
 
-	for _, mb := range fs.blks {
-		mb.mu.Lock()
-		purged += mb.msgs
-		bytes += mb.bytes
-		mb.dirtyCloseWithRemove(true)
-		mb.mu.Unlock()
+	if cb != nil {
+		for _, mb := range fs.blks {
+			mb.mu.Lock()
+			purged += mb.msgs
+			bytes += mb.bytes
+			mb.dirtyCloseWithRemove(true)
+			mb.mu.Unlock()
+		}
 	}
 
 	// Reset
@@ -6685,16 +6656,15 @@ func (o *consumerFileStore) UpdateAcks(dseq, sseq uint64) error {
 	if o.cfg.AckPolicy == AckNone {
 		return ErrNoAckPolicy
 	}
+	if len(o.state.Pending) == 0 || o.state.Pending[sseq] == nil {
+		return ErrStoreMsgNotFound
+	}
 
 	// On restarts the old leader may get a replay from the raft logs that are old.
 	if dseq <= o.state.AckFloor.Consumer {
 		return nil
 	}
 
-	if len(o.state.Pending) == 0 || o.state.Pending[sseq] == nil {
-		return ErrStoreMsgNotFound
-	}
-
 	// Check for AckAll here.
 	if o.cfg.AckPolicy == AckAll {
 		sgap := sseq - o.state.AckFloor.Stream
@@ -6717,15 +6687,21 @@ func (o *consumerFileStore) UpdateAcks(dseq, sseq uint64) error {
 		delete(o.state.Pending, sseq)
 		dseq = p.Sequence // Use the original.
 	}
+	// Now remove from redelivered.
+	if len(o.state.Redelivered) > 0 {
+		delete(o.state.Redelivered, sseq)
+	}
+
 	if len(o.state.Pending) == 0 {
 		o.state.AckFloor.Consumer = o.state.Delivered.Consumer
 		o.state.AckFloor.Stream = o.state.Delivered.Stream
 	} else if dseq == o.state.AckFloor.Consumer+1 {
+		first := o.state.AckFloor.Consumer == 0
 		o.state.AckFloor.Consumer = dseq
 		o.state.AckFloor.Stream = sseq
 
-		if o.state.Delivered.Consumer > dseq {
-			for ss := sseq + 1; ss <= o.state.Delivered.Stream; ss++ {
+		if !first && o.state.Delivered.Consumer > dseq {
+			for ss := sseq + 1; ss < o.state.Delivered.Stream; ss++ {
 				if p, ok := o.state.Pending[ss]; ok {
 					if p.Sequence > 0 {
 						o.state.AckFloor.Consumer = p.Sequence - 1
@@ -6736,8 +6712,6 @@ func (o *consumerFileStore) UpdateAcks(dseq, sseq uint64) error {
 			}
 		}
 	}
-	// We do these regardless.
-	delete(o.state.Redelivered, sseq)
 
 	o.kickFlusher()
 	return nil
@@ -6781,7 +6755,7 @@ func (o *consumerFileStore) Update(state *ConsumerState) error {
 	defer o.mu.Unlock()
 
 	// Check to see if this is an outdated update.
-	if state.Delivered.Consumer < o.state.Delivered.Consumer || state.AckFloor.Stream < o.state.AckFloor.Stream {
+	if state.Delivered.Consumer < o.state.Delivered.Consumer {
 		return nil
 	}
 
diff --git a/server/jetstream.go b/server/jetstream.go
index e5452e99..3c8c0623 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -578,9 +578,6 @@ func (s *Server) DisableJetStream() error {
 	// Normal shutdown.
 	s.shutdownJetStream()
 
-	// Shut down the RAFT groups.
-	s.shutdownRaftNodes()
-
 	return nil
 }
 
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index b54c2c10..c8822016 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -281,37 +281,6 @@ func (s *Server) JetStreamStepdownStream(account, stream string) error {
 	return nil
 }
 
-func (s *Server) JetStreamStepdownConsumer(account, stream, consumer string) error {
-	js, cc := s.getJetStreamCluster()
-	if js == nil {
-		return NewJSNotEnabledError()
-	}
-	if cc == nil {
-		return NewJSClusterNotActiveError()
-	}
-	// Grab account
-	acc, err := s.LookupAccount(account)
-	if err != nil {
-		return err
-	}
-	// Grab stream
-	mset, err := acc.lookupStream(stream)
-	if err != nil {
-		return err
-	}
-
-	o := mset.lookupConsumer(consumer)
-	if o == nil {
-		return NewJSConsumerNotFoundError()
-	}
-
-	if node := o.raftNode(); node != nil && node.Leader() {
-		node.StepDown()
-	}
-
-	return nil
-}
-
 func (s *Server) JetStreamSnapshotStream(account, stream string) error {
 	js, cc := s.getJetStreamCluster()
 	if js == nil {
@@ -351,7 +320,7 @@ func (s *Server) JetStreamClusterPeers() []string {
 	defer js.mu.RUnlock()
 
 	cc := js.cluster
-	if !cc.isLeader() || cc.meta == nil {
+	if !cc.isLeader() {
 		return nil
 	}
 	peers := cc.meta.Peers()
@@ -652,8 +621,6 @@ func (js *jetStream) setupMetaGroup() error {
 		s.Errorf("Error creating filestore: %v", err)
 		return err
 	}
-	// Register our server.
-	fs.registerServer(s)
 
 	cfg := &RaftConfig{Name: defaultMetaGroupName, Store: storeDir, Log: fs}
 
@@ -787,7 +754,7 @@ func (js *jetStream) isGroupLeaderless(rg *raftGroup) bool {
 	// If we don't have a leader.
 	if rg.node.GroupLeader() == _EMPTY_ {
 		// Threshold for jetstream startup.
-		const startupThreshold = 10 * time.Second
+		const startupThreshold = 5 * time.Second
 
 		if rg.node.HadPreviousLeader() {
 			// Make sure we have been running long enough to intelligently determine this.
@@ -1164,12 +1131,8 @@ func (js *jetStream) monitorCluster() {
 			}
 			aq.recycle(&ces)
 		case isLeader = <-lch:
-			// For meta layer synchronize everyone to our state on becoming leader.
-			if isLeader {
-				n.SendSnapshot(js.metaSnapshot())
-			}
-			// Process the change.
 			js.processLeaderChange(isLeader)
+
 			if isLeader {
 				s.sendInternalMsgLocked(serverStatsPingReqSubj, _EMPTY_, nil, nil)
 				// Install a snapshot as we become leader.
@@ -1487,9 +1450,6 @@ func (js *jetStream) processAddPeer(peer string) {
 	defer js.mu.Unlock()
 
 	s, cc := js.srv, js.cluster
-	if cc == nil || cc.meta == nil {
-		return
-	}
 	isLeader := cc.isLeader()
 
 	// Now check if we are meta-leader. We will check for any re-assignments.
@@ -1531,7 +1491,7 @@ func (js *jetStream) processAddPeer(peer string) {
 func (js *jetStream) processRemovePeer(peer string) {
 	js.mu.Lock()
 	s, cc := js.srv, js.cluster
-	if cc == nil || cc.meta == nil {
+	if cc.meta == nil {
 		js.mu.Unlock()
 		return
 	}
@@ -1595,9 +1555,6 @@ func (js *jetStream) removePeerFromStreamLocked(sa *streamAssignment, peer strin
 	}
 
 	s, cc, csa := js.srv, js.cluster, sa.copyGroup()
-	if cc == nil || cc.meta == nil {
-		return false
-	}
 	replaced := cc.remapStreamAssignment(csa, peer)
 	if !replaced {
 		s.Warnf("JetStream cluster could not replace peer for stream '%s > %s'", sa.Client.serviceAccount(), sa.Config.Name)
@@ -1839,8 +1796,6 @@ func (js *jetStream) createRaftGroup(accName string, rg *raftGroup, storage Stor
 			s.Errorf("Error creating filestore WAL: %v", err)
 			return err
 		}
-		// Register our server.
-		fs.registerServer(s)
 		store = fs
 	} else {
 		ms, err := newMemStore(&StreamConfig{Name: rg.Name, Storage: MemoryStorage})
@@ -1937,22 +1892,14 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		return
 	}
 
-	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
-	defer n.Stop()
-
-	// Make sure only one is running.
-	if mset != nil {
-		if mset.checkInMonitor() {
-			return
-		}
-		defer mset.clearMonitorRunning()
-	}
-
 	qch, lch, aq, uch, ourPeerId := n.QuitC(), n.LeadChangeC(), n.ApplyQ(), mset.updateC(), meta.ID()
 
 	s.Debugf("Starting stream monitor for '%s > %s' [%s]", sa.Client.serviceAccount(), sa.Config.Name, n.Group())
 	defer s.Debugf("Exiting stream monitor for '%s > %s' [%s]", sa.Client.serviceAccount(), sa.Config.Name, n.Group())
 
+	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
+	defer n.Stop()
+
 	// Make sure we do not leave the apply channel to fill up and block the raft layer.
 	defer func() {
 		if n.State() == Closed {
@@ -2063,31 +2010,6 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	}
 	defer stopDirectMonitoring()
 
-	// Check if we are interest based and if so and we have an active stream wait until we
-	// have the consumers attached. This can become important when a server has lots of assets
-	// since we process streams first then consumers as an asset class.
-	if mset != nil && mset.isInterestRetention() {
-		js.mu.RLock()
-		numExpectedConsumers := len(sa.consumers)
-		js.mu.RUnlock()
-		if mset.numConsumers() < numExpectedConsumers {
-			s.Debugf("Waiting for consumers for interest based stream '%s > %s'", accName, mset.name())
-			// Wait up to 10s
-			const maxWaitTime = 10 * time.Second
-			const sleepTime = 250 * time.Millisecond
-			timeout := time.Now().Add(maxWaitTime)
-			for time.Now().Before(timeout) {
-				if mset.numConsumers() >= numExpectedConsumers {
-					break
-				}
-				time.Sleep(sleepTime)
-			}
-			if actual := mset.numConsumers(); actual < numExpectedConsumers {
-				s.Warnf("All consumers not online for '%s > %s': expected %d but only have %d", accName, mset.name(), numExpectedConsumers, actual)
-			}
-		}
-	}
-
 	// This is triggered during a scale up from R1 to clustered mode. We need the new followers to catchup,
 	// similar to how we trigger the catchup mechanism post a backup/restore.
 	// We can arrive here NOT being the leader, so we send the snapshot only if we are, and in this case
@@ -2097,7 +2019,6 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		n.SendSnapshot(mset.stateSnapshot())
 		sendSnapshot = false
 	}
-
 	for {
 		select {
 		case <-s.quitCh:
@@ -2233,7 +2154,6 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 
 		case <-t.C:
 			doSnapshot()
-
 		case <-uch:
 			// keep stream assignment current
 			sa = mset.streamAssignment()
@@ -2572,11 +2492,6 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 				if lseq-clfs < last {
 					s.Debugf("Apply stream entries for '%s > %s' skipping message with sequence %d with last of %d",
 						mset.account(), mset.name(), lseq+1-clfs, last)
-					// Check for any preAcks in case we are interest based.
-					mset.mu.Lock()
-					seq := lseq + 1 - mset.clfs
-					mset.clearAllPreAcks(seq)
-					mset.mu.Unlock()
 					continue
 				}
 
@@ -2590,9 +2505,7 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 				// Messages to be skipped have no subject or timestamp or msg or hdr.
 				if subject == _EMPTY_ && ts == 0 && len(msg) == 0 && len(hdr) == 0 {
 					// Skip and update our lseq.
-					last := mset.store.SkipMsg()
-					mset.setLastSeq(last)
-					mset.clearAllPreAcks(last)
+					mset.setLastSeq(mset.store.SkipMsg())
 					continue
 				}
 
@@ -3532,12 +3445,7 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 			mset.monitorWg.Wait()
 			err = mset.stop(true, wasLeader)
 			stopped = true
-		} else if isMember {
-			s.Warnf("JetStream failed to lookup running stream while removing stream '%s > %s' from this server",
-				sa.Client.serviceAccount(), sa.Config.Name)
 		}
-	} else if isMember {
-		s.Warnf("JetStream failed to lookup account while removing stream '%s > %s' from this server", sa.Client.serviceAccount(), sa.Config.Name)
 	}
 
 	// Always delete the node if present.
@@ -3550,16 +3458,11 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 	// 2) node was nil (and couldn't be deleted)
 	if !stopped || node == nil {
 		if sacc := s.SystemAccount(); sacc != nil {
-			saccName := sacc.GetName()
-			os.RemoveAll(filepath.Join(js.config.StoreDir, saccName, defaultStoreDirName, sa.Group.Name))
+			os.RemoveAll(filepath.Join(js.config.StoreDir, sacc.GetName(), defaultStoreDirName, sa.Group.Name))
 			// cleanup dependent consumer groups
 			if !stopped {
 				for _, ca := range sa.consumers {
-					// Make sure we cleanup any possible running nodes for the consumers.
-					if isMember && ca.Group != nil && ca.Group.node != nil {
-						ca.Group.node.Delete()
-					}
-					os.RemoveAll(filepath.Join(js.config.StoreDir, saccName, defaultStoreDirName, ca.Group.Name))
+					os.RemoveAll(filepath.Join(js.config.StoreDir, sacc.GetName(), defaultStoreDirName, ca.Group.Name))
 				}
 			}
 		}
@@ -3814,7 +3717,6 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 		if ca.Config.MemoryStorage {
 			storage = MemoryStorage
 		}
-		// No-op if R1.
 		js.createRaftGroup(accName, rg, storage)
 	} else {
 		// If we are clustered update the known peers.
@@ -4177,9 +4079,6 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 		return
 	}
 
-	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
-	defer n.Stop()
-
 	// Make sure only one is running.
 	if o.checkInMonitor() {
 		return
@@ -4191,6 +4090,9 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 	s.Debugf("Starting consumer monitor for '%s > %s > %s' [%s]", o.acc.Name, ca.Stream, ca.Name, n.Group())
 	defer s.Debugf("Exiting consumer monitor for '%s > %s > %s' [%s]", o.acc.Name, ca.Stream, ca.Name, n.Group())
 
+	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
+	defer n.Stop()
+
 	const (
 		compactInterval = 2 * time.Minute
 		compactSizeMin  = 64 * 1024 // What is stored here is always small for consumers.
@@ -4211,9 +4113,9 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 	var lastSnap []byte
 	var lastSnapTime time.Time
 
-	doSnapshot := func(force bool) {
+	doSnapshot := func() {
 		// Bail if trying too fast and not in a forced situation.
-		if !force && time.Since(lastSnapTime) < minSnapDelta {
+		if time.Since(lastSnapTime) < minSnapDelta {
 			return
 		}
 
@@ -4221,7 +4123,7 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 		ne, nb := n.Size()
 		if !n.NeedSnapshot() {
 			// Check if we should compact etc. based on size of log.
-			if !force && ne < compactNumMin && nb < compactSizeMin {
+			if ne < compactNumMin && nb < compactSizeMin {
 				return
 			}
 		}
@@ -4279,17 +4181,15 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 				if ce == nil {
 					recovering = false
 					if n.NeedSnapshot() {
-						doSnapshot(true)
+						doSnapshot()
 					}
-					// Check our state if we are under an interest based stream.
-					o.checkStateForInterestStream()
 					continue
 				}
 				if err := js.applyConsumerEntries(o, ce, isLeader); err == nil {
 					ne, nb := n.Applied(ce.Index)
 					// If we have at least min entries to compact, go ahead and snapshot/compact.
 					if nb > 0 && ne >= compactNumMin || nb > compactSizeMin {
-						doSnapshot(false)
+						doSnapshot()
 					}
 				} else {
 					s.Warnf("Error applying consumer entries to '%s > %s'", ca.Client.serviceAccount(), ca.Name)
@@ -4300,10 +4200,8 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 			if recovering && !isLeader {
 				js.setConsumerAssignmentRecovering(ca)
 			}
-
-			// Process the change.
 			if err := js.processConsumerLeaderChange(o, isLeader); err == nil && isLeader {
-				doSnapshot(true)
+				doSnapshot()
 			}
 
 			// We may receive a leader change after the consumer assignment which would cancel us
@@ -4392,7 +4290,7 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 			}
 
 		case <-t.C:
-			doSnapshot(false)
+			doSnapshot()
 		}
 	}
 }
@@ -4410,7 +4308,6 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 				}
 				panic(err.Error())
 			}
-
 			if err = o.store.Update(state); err != nil {
 				o.mu.RLock()
 				s, acc, mset, name := o.srv, o.acc, o.mset, o.name
@@ -4418,10 +4315,7 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 				if s != nil && mset != nil {
 					s.Warnf("Consumer '%s > %s > %s' error on store update from snapshot entry: %v", acc, mset.name(), name, err)
 				}
-			} else {
-				o.checkStateForInterestStream()
 			}
-
 		} else if e.Type == EntryRemovePeer {
 			js.mu.RLock()
 			var ourID string
@@ -4516,20 +4410,13 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 
 func (o *consumer) processReplicatedAck(dseq, sseq uint64) {
 	o.mu.Lock()
-
-	mset := o.mset
-	if o.closed || mset == nil {
-		o.mu.Unlock()
-		return
-	}
-
 	// Update activity.
 	o.lat = time.Now()
-
 	// Do actual ack update to store.
 	o.store.UpdateAcks(dseq, sseq)
 
-	if o.retention == LimitsPolicy {
+	mset := o.mset
+	if mset == nil || o.retention == LimitsPolicy {
 		o.mu.Unlock()
 		return
 	}
@@ -4768,9 +4655,6 @@ func (js *jetStream) processStreamAssignmentResults(sub *subscription, c *client
 	defer js.mu.Unlock()
 
 	s, cc := js.srv, js.cluster
-	if cc == nil || cc.meta == nil {
-		return
-	}
 
 	// This should have been done already in processStreamAssignment, but in
 	// case we have a code path that gets here with no processStreamAssignment,
@@ -4844,9 +4728,6 @@ func (js *jetStream) processConsumerAssignmentResults(sub *subscription, c *clie
 	defer js.mu.Unlock()
 
 	s, cc := js.srv, js.cluster
-	if cc == nil || cc.meta == nil {
-		return
-	}
 
 	if sa := js.streamAssignment(result.Account, result.Stream); sa != nil && sa.consumers != nil {
 		if ca := sa.consumers[result.Consumer]; ca != nil && !ca.responded {
@@ -5616,10 +5497,6 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 	// Now process the request and proposal.
 	js.mu.Lock()
 	defer js.mu.Unlock()
-	meta := cc.meta
-	if meta == nil {
-		return
-	}
 
 	var resp = JSApiStreamUpdateResponse{ApiResponse: ApiResponse{Type: JSApiStreamUpdateResponseType}}
 
@@ -5883,12 +5760,13 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 	}
 
 	sa := &streamAssignment{Group: rg, Sync: osa.Sync, Created: osa.Created, Config: newCfg, Subject: subject, Reply: reply, Client: ci}
-	meta.Propose(encodeUpdateStreamAssignment(sa))
+	cc.meta.Propose(encodeUpdateStreamAssignment(sa))
 
 	// Process any staged consumers.
 	for _, ca := range consumers {
-		meta.Propose(encodeAddConsumerAssignment(ca))
+		cc.meta.Propose(encodeAddConsumerAssignment(ca))
 	}
+
 }
 
 func (s *Server) jsClusteredStreamDeleteRequest(ci *ClientInfo, acc *Account, stream, subject, reply string, rmsg []byte) {
@@ -5900,10 +5778,6 @@ func (s *Server) jsClusteredStreamDeleteRequest(ci *ClientInfo, acc *Account, st
 	js.mu.Lock()
 	defer js.mu.Unlock()
 
-	if cc.meta == nil {
-		return
-	}
-
 	osa := js.streamAssignment(acc.Name, stream)
 	if osa == nil {
 		var resp = JSApiStreamDeleteResponse{ApiResponse: ApiResponse{Type: JSApiStreamDeleteResponseType}}
@@ -5977,10 +5851,6 @@ func (s *Server) jsClusteredStreamRestoreRequest(
 	js.mu.Lock()
 	defer js.mu.Unlock()
 
-	if cc.meta == nil {
-		return
-	}
-
 	cfg := &req.Config
 	resp := JSApiStreamRestoreResponse{ApiResponse: ApiResponse{Type: JSApiStreamRestoreResponseType}}
 
@@ -6352,10 +6222,6 @@ func (s *Server) jsClusteredConsumerDeleteRequest(ci *ClientInfo, acc *Account,
 	js.mu.Lock()
 	defer js.mu.Unlock()
 
-	if cc.meta == nil {
-		return
-	}
-
 	var resp = JSApiConsumerDeleteResponse{ApiResponse: ApiResponse{Type: JSApiConsumerDeleteResponseType}}
 
 	sa := js.streamAssignment(acc.Name, stream)
@@ -6544,10 +6410,6 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 	js.mu.Lock()
 	defer js.mu.Unlock()
 
-	if cc.meta == nil {
-		return
-	}
-
 	// Lookup the stream assignment.
 	sa := js.streamAssignment(acc.Name, stream)
 	if sa == nil {
@@ -7206,18 +7068,15 @@ func (mset *stream) calculateSyncRequest(state *StreamState, snap *streamSnapsho
 // processSnapshotDeletes will update our current store based on the snapshot
 // but only processing deletes and new FirstSeq / purges.
 func (mset *stream) processSnapshotDeletes(snap *streamSnapshot) {
-	mset.mu.Lock()
 	var state StreamState
 	mset.store.FastState(&state)
+
 	// Always adjust if FirstSeq has moved beyond our state.
 	if snap.FirstSeq > state.FirstSeq {
 		mset.store.Compact(snap.FirstSeq)
 		mset.store.FastState(&state)
-		mset.lseq = state.LastSeq
-		mset.clearAllPreAcksBelowFloor(state.FirstSeq)
+		mset.setLastSeq(state.LastSeq)
 	}
-	mset.mu.Unlock()
-
 	// Range the deleted and delete if applicable.
 	for _, dseq := range snap.Deleted {
 		if dseq > state.FirstSeq && dseq <= state.LastSeq {
@@ -7402,40 +7261,25 @@ func (mset *stream) processSnapshot(snap *streamSnapshot) (e error) {
 	// we are synched for the next message sequence properly.
 	lastRequested := sreq.LastSeq
 	checkFinalState := func() {
-		// Bail if no stream.
-		if mset == nil {
-			return
-		}
-
-		mset.mu.Lock()
-		var state StreamState
-		mset.store.FastState(&state)
-		var didReset bool
-		firstExpected := lastRequested + 1
-		if state.FirstSeq != firstExpected {
-			// Reset our notion of first.
-			mset.store.Compact(firstExpected)
+		if mset != nil {
+			mset.mu.Lock()
+			var state StreamState
 			mset.store.FastState(&state)
-			// Make sure last is also correct in case this also moved.
-			mset.lseq = state.LastSeq
-			mset.clearAllPreAcksBelowFloor(state.FirstSeq)
-			didReset = true
-		}
-		mset.mu.Unlock()
-
-		if didReset {
-			s.Warnf("Catchup for stream '%s > %s' resetting first sequence: %d on catchup complete",
-				mset.account(), mset.name(), firstExpected)
-		}
-
-		mset.mu.RLock()
-		consumers := make([]*consumer, 0, len(mset.consumers))
-		for _, o := range mset.consumers {
-			consumers = append(consumers, o)
-		}
-		mset.mu.RUnlock()
-		for _, o := range consumers {
-			o.checkStateForInterestStream()
+			var didReset bool
+			firstExpected := lastRequested + 1
+			if state.FirstSeq != firstExpected {
+				// Reset our notion of first.
+				mset.store.Compact(firstExpected)
+				mset.store.FastState(&state)
+				// Make sure last is also correct in case this also moved.
+				mset.lseq = state.LastSeq
+				didReset = true
+			}
+			mset.mu.Unlock()
+			if didReset {
+				s.Warnf("Catchup for stream '%s > %s' resetting first sequence: %d on catchup complete",
+					mset.account(), mset.name(), firstExpected)
+			}
 		}
 	}
 
@@ -7627,17 +7471,11 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 		return 0, errCatchupBadMsg
 	}
 
-	mset.mu.Lock()
+	mset.mu.RLock()
 	st := mset.cfg.Storage
 	ddloaded := mset.ddloaded
 	tierName := mset.tier
-
-	if mset.hasAllPreAcks(seq, subj) {
-		mset.clearAllPreAcks(seq)
-		// Mark this to be skipped
-		subj, ts = _EMPTY_, 0
-	}
-	mset.mu.Unlock()
+	mset.mu.RUnlock()
 
 	if mset.js.limitsExceeded(st) {
 		return 0, NewJSInsufficientResourcesError()
diff --git a/server/monitor.go b/server/monitor.go
index 3b3488b0..6c2ead7c 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -36,6 +36,19 @@ import (
 	"github.com/nats-io/nats-server/v2/server/pse"
 )
 
+// Snapshot this
+var numCores int
+var maxProcs int
+
+func SnapshotMonitorInfo() {
+	numCores = runtime.NumCPU()
+	maxProcs = runtime.GOMAXPROCS(0)
+}
+
+func init() {
+	SnapshotMonitorInfo()
+}
+
 // Connz represents detailed information on current client connections.
 type Connz struct {
 	ID       string      `json:"server_id"`
@@ -1515,8 +1528,8 @@ func (s *Server) createVarz(pcpu float64, rss int64) *Varz {
 		},
 		Start:                 s.start,
 		MaxSubs:               opts.MaxSubs,
-		Cores:                 runtime.NumCPU(),
-		MaxProcs:              runtime.GOMAXPROCS(0),
+		Cores:                 numCores,
+		MaxProcs:              maxProcs,
 		Tags:                  opts.Tags,
 		TrustedOperatorsJwt:   opts.operatorJWT,
 		TrustedOperatorsClaim: opts.TrustedOperators,
@@ -3068,13 +3081,14 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
 
 	// Clustered JetStream
 	js.mu.RLock()
+	defer js.mu.RUnlock()
+
 	cc := js.cluster
-	js.mu.RUnlock()
 
 	const na = "unavailable"
 
 	// Currently single server we make sure the streams were recovered.
-	if cc == nil {
+	if cc == nil || cc.meta == nil {
 		sdir := js.config.StoreDir
 		// Whip through account folders and pull each stream name.
 		fis, _ := os.ReadDir(sdir)
@@ -3099,19 +3113,17 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
 	}
 
 	// If we are here we want to check for any assets assigned to us.
-	var meta RaftNode
-	js.mu.RLock()
-	meta = cc.meta
-	js.mu.RUnlock()
+	meta := cc.meta
+	ourID := meta.ID()
 
 	// If no meta leader.
-	if meta == nil || meta.GroupLeader() == _EMPTY_ {
+	if meta.GroupLeader() == _EMPTY_ {
 		health.Status = na
 		health.Error = "JetStream has not established contact with a meta leader"
 		return health
 	}
 	// If we are not current with the meta leader.
-	if !meta.Healthy() {
+	if !meta.Current() {
 		health.Status = na
 		health.Error = "JetStream is not current with the meta leader"
 		return health
@@ -3124,7 +3136,6 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
 
 	// Range across all accounts, the streams assigned to them, and the consumers.
 	// If they are assigned to this server check their status.
-	ourID := meta.ID()
 	for acc, asa := range cc.streams {
 		for stream, sa := range asa {
 			if sa.Group.isMember(ourID) {
diff --git a/server/raft.go b/server/raft.go
index 9bdbe5c3..6a4ed96b 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -25,7 +25,6 @@ import (
 	"net"
 	"os"
 	"path/filepath"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -486,12 +485,6 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig) (RaftNode, error
 
 	n.debug("Started")
 
-	// Check if we need to start in observer mode due to lame duck status.
-	if s.isLameDuckMode() {
-		n.debug("Will start in observer mode due to lame duck status")
-		n.SetObserver(true)
-	}
-
 	n.Lock()
 	n.resetElectionTimeout()
 	n.llqrt = time.Now()
@@ -618,9 +611,6 @@ func (s *Server) shutdownRaftNodes() {
 	}
 }
 
-// Used in lameduck mode to move off the leaders.
-// We also put all nodes in observer mode so new leaders
-// can not be placed on this server.
 func (s *Server) transferRaftLeaders() bool {
 	if s == nil {
 		return false
@@ -641,7 +631,6 @@ func (s *Server) transferRaftLeaders() bool {
 			node.StepDown()
 			didTransfer = true
 		}
-		node.SetObserver(true)
 	}
 	return didTransfer
 }
@@ -866,11 +855,6 @@ func (n *raft) Applied(index uint64) (entries uint64, bytes uint64) {
 	n.Lock()
 	defer n.Unlock()
 
-	// Ignore if not applicable. This can happen during a reset.
-	if index > n.commit {
-		return 0, 0
-	}
-
 	// Ignore if already applied.
 	if index > n.applied {
 		n.applied = index
@@ -955,7 +939,7 @@ func (n *raft) InstallSnapshot(data []byte) error {
 	var state StreamState
 	n.wal.FastState(&state)
 
-	if n.applied == 0 {
+	if n.applied == 0 || len(data) == 0 {
 		n.Unlock()
 		return errNoSnapAvailable
 	}
@@ -1213,9 +1197,9 @@ func (n *raft) isCurrent(includeForwardProgress bool) bool {
 	// forward progress.
 	if startDelta := n.commit - n.applied; startDelta > 0 {
 		for i := 0; i < 10; i++ { // 5ms, in 0.5ms increments
-			n.Unlock()
+			n.RUnlock()
 			time.Sleep(time.Millisecond / 2)
-			n.Lock()
+			n.RLock()
 			if n.commit-n.applied < startDelta {
 				// The gap is getting smaller, so we're making forward progress.
 				return true
@@ -1232,8 +1216,8 @@ func (n *raft) Current() bool {
 	if n == nil {
 		return false
 	}
-	n.Lock()
-	defer n.Unlock()
+	n.RLock()
+	defer n.RUnlock()
 	return n.isCurrent(false)
 }
 
@@ -1242,8 +1226,8 @@ func (n *raft) Healthy() bool {
 	if n == nil {
 		return false
 	}
-	n.Lock()
-	defer n.Unlock()
+	n.RLock()
+	defer n.RUnlock()
 	return n.isCurrent(true)
 }
 
@@ -1341,7 +1325,6 @@ func (n *raft) StepDown(preferred ...string) error {
 	}
 
 	stepdown := n.stepdown
-	prop := n.prop
 	n.Unlock()
 
 	if len(preferred) > 0 && maybeLeader == noLeader {
@@ -1351,13 +1334,11 @@ func (n *raft) StepDown(preferred ...string) error {
 	// If we have a new leader selected, transfer over to them.
 	if maybeLeader != noLeader {
 		n.debug("Selected %q for new leader", maybeLeader)
-		prop.push(&Entry{EntryLeaderTransfer, []byte(maybeLeader)})
-		time.AfterFunc(250*time.Millisecond, func() { stepdown.push(noLeader) })
-	} else {
-		// Force us to stepdown here.
-		n.debug("Stepping down")
-		stepdown.push(noLeader)
+		n.sendAppendEntry([]*Entry{{EntryLeaderTransfer, []byte(maybeLeader)}})
 	}
+	// Force us to stepdown here.
+	n.debug("Stepping down")
+	stepdown.push(noLeader)
 
 	return nil
 }
@@ -2074,7 +2055,6 @@ func (n *raft) runAsLeader() {
 		n.Unlock()
 	}()
 
-	// To send out our initial peer state.
 	n.sendPeerState()
 
 	hb := time.NewTicker(hbInterval)
@@ -2112,7 +2092,7 @@ func (n *raft) runAsLeader() {
 					continue
 				}
 				n.sendAppendEntry(entries)
-				// We need to re-create `entries` because there is a reference
+				// We need to re-craete `entries` because there is a reference
 				// to it in the node's pae map.
 				entries = nil
 			}
@@ -2432,11 +2412,9 @@ func (n *raft) applyCommit(index uint64) error {
 		if ae, err = n.loadEntry(index); err != nil {
 			if err != ErrStoreClosed && err != ErrStoreEOF {
 				if err == errBadMsg {
-					n.warn("Got an error loading %d index: %v - will reset", index, err)
-					n.resetWAL()
-				} else {
-					n.warn("Got an error loading %d index: %v", index, err)
+					n.setWriteErrLocked(err)
 				}
+				n.warn("Got an error loading %d index: %v", index, err)
 			}
 			n.commit = original
 			return errEntryLoadFailed
@@ -2511,13 +2489,6 @@ func (n *raft) applyCommit(index uint64) error {
 
 			// We pass these up as well.
 			committed = append(committed, e)
-
-		case EntryLeaderTransfer:
-			if n.state == Leader {
-				n.debug("Stepping down")
-				n.stepdown.push(noLeader)
-			}
-			// No-op
 		}
 	}
 	// Pass to the upper layers if we have normal entries.
@@ -2536,10 +2507,6 @@ func (n *raft) applyCommit(index uint64) error {
 // Used to track a success response and apply entries.
 func (n *raft) trackResponse(ar *appendEntryResponse) {
 	n.Lock()
-	if n.state == Closed {
-		n.Unlock()
-		return
-	}
 
 	// Update peer's last index.
 	if ps := n.peers[ar.peer]; ps != nil && ar.index > ps.li {
@@ -2565,8 +2532,8 @@ func (n *raft) trackResponse(ar *appendEntryResponse) {
 		if nr := len(results); nr >= n.qn {
 			// We have a quorum.
 			for index := n.commit + 1; index <= ar.index; index++ {
-				if err := n.applyCommit(index); err != nil && err != errNodeClosed {
-					n.error("Got an error applying commit for %d: %v", index, err)
+				if err := n.applyCommit(index); err != nil {
+					n.error("Got an error apply commit for %d: %v", index, err)
 					break
 				}
 			}
@@ -2881,6 +2848,15 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		}
 	}
 
+	// Ignore old terms.
+	if isNew && ae.term < n.term {
+		ar := &appendEntryResponse{n.term, n.pindex, n.id, false, _EMPTY_}
+		n.Unlock()
+		n.debug("AppendEntry ignoring old term")
+		n.sendRPC(ae.reply, _EMPTY_, ar.encode(arbuf))
+		return
+	}
+
 	// If we are catching up ignore old catchup subs.
 	// This could happen when we stall or cancel a catchup.
 	if !isNew && catchingUp && sub != n.catchup.sub {
@@ -2916,7 +2892,6 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 
 	// If this term is greater than ours.
 	if ae.term > n.term {
-		n.pterm = ae.pterm
 		n.term = ae.term
 		n.vote = noVote
 		if isNew {
@@ -2936,7 +2911,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		n.updateLeadChange(false)
 	}
 
-	if (isNew && ae.pterm != n.pterm) || ae.pindex != n.pindex {
+	if ae.pterm != n.pterm || ae.pindex != n.pindex {
 		// Check if this is a lower or equal index than what we were expecting.
 		if ae.pindex <= n.pindex {
 			n.debug("AppendEntry detected pindex less than ours: %d:%d vs %d:%d", ae.pterm, ae.pindex, n.pterm, n.pindex)
@@ -3025,7 +3000,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	}
 
 	// Save to our WAL if we have entries.
-	if ae.shouldStore() {
+	if len(ae.entries) > 0 {
 		// Only store if an original which will have sub != nil
 		if sub != nil {
 			if err := n.storeToWAL(ae); err != nil {
@@ -3050,26 +3025,26 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			n.pterm = ae.term
 			n.pindex = ae.pindex + 1
 		}
-	}
 
-	// Check to see if we have any related entries to process here.
-	for _, e := range ae.entries {
-		switch e.Type {
-		case EntryLeaderTransfer:
-			if isNew {
-				maybeLeader := string(e.Data)
-				if maybeLeader == n.id && !n.observer && !n.paused {
-					n.lxfer = true
-					n.xferCampaign()
+		// Check to see if we have any related entries to process here.
+		for _, e := range ae.entries {
+			switch e.Type {
+			case EntryLeaderTransfer:
+				if isNew {
+					maybeLeader := string(e.Data)
+					if maybeLeader == n.id && !n.observer && !n.paused {
+						n.lxfer = true
+						n.xferCampaign()
+					}
 				}
-			}
-		case EntryAddPeer:
-			if newPeer := string(e.Data); len(newPeer) == idLen {
-				// Track directly, but wait for commit to be official
-				if ps := n.peers[newPeer]; ps != nil {
-					ps.ts = time.Now().UnixNano()
-				} else {
-					n.peers[newPeer] = &lps{time.Now().UnixNano(), 0, false}
+			case EntryAddPeer:
+				if newPeer := string(e.Data); len(newPeer) == idLen {
+					// Track directly, but wait for commit to be official
+					if ps := n.peers[newPeer]; ps != nil {
+						ps.ts = time.Now().UnixNano()
+					} else {
+						n.peers[newPeer] = &lps{time.Now().UnixNano(), 0, false}
+					}
 				}
 			}
 		}
@@ -3129,14 +3104,12 @@ func (n *raft) processAppendEntryResponse(ar *appendEntryResponse) {
 		n.trackResponse(ar)
 	} else if ar.term > n.term {
 		// False here and they have a higher term.
-		n.Lock()
 		n.term = ar.term
 		n.vote = noVote
 		n.writeTermVote()
 		n.warn("Detected another leader with higher term, will stepdown and reset")
 		n.stepdown.push(noLeader)
 		n.resetWAL()
-		n.Unlock()
 	} else if ar.reply != _EMPTY_ {
 		n.catchupFollower(ar)
 	}
@@ -3154,11 +3127,6 @@ func (n *raft) buildAppendEntry(entries []*Entry) *appendEntry {
 	return &appendEntry{n.id, n.term, n.commit, n.pterm, n.pindex, entries, _EMPTY_, nil, nil}
 }
 
-// Determine if we should store an entry.
-func (ae *appendEntry) shouldStore() bool {
-	return ae != nil && len(ae.entries) > 0 && ae.entries[0].Type != EntryLeaderTransfer
-}
-
 // Store our append entry to our WAL.
 // lock should be held.
 func (n *raft) storeToWAL(ae *appendEntry) error {
@@ -3168,7 +3136,6 @@ func (n *raft) storeToWAL(ae *appendEntry) error {
 	if n.werr != nil {
 		return n.werr
 	}
-
 	seq, _, err := n.wal.StoreMsg(_EMPTY_, nil, ae.buf)
 	if err != nil {
 		n.setWriteErrLocked(err)
@@ -3215,7 +3182,7 @@ func (n *raft) sendAppendEntry(entries []*Entry) {
 	}
 
 	// If we have entries store this in our wal.
-	if ae.shouldStore() {
+	if len(entries) > 0 {
 		if err := n.storeToWAL(ae); err != nil {
 			return
 		}
@@ -3435,10 +3402,8 @@ func (n *raft) setWriteErrLocked(err error) {
 	}
 	n.werr = err
 
-	if isOutOfSpaceErr(err) {
-		// For now since this can be happening all under the covers, we will call up and disable JetStream.
-		go n.s.handleOutOfSpace(nil)
-	}
+	// For now since this can be happening all under the covers, we will call up and disable JetStream.
+	go n.s.handleOutOfSpace(nil)
 }
 
 // Capture our write error if any and hold.
@@ -3591,10 +3556,10 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 	// If this is a higher term go ahead and stepdown.
 	if vr.term > n.term {
 		if n.state != Follower {
-			n.debug("Stepping down from %s, detected higher term: %d vs %d", vr.term, n.term, strings.ToLower(n.state.String()))
+			n.debug("Stepping down from candidate, detected higher term: %d vs %d", vr.term, n.term)
 			n.stepdown.push(noLeader)
-			n.term = vr.term
 		}
+		n.term = vr.term
 		n.vote = noVote
 		n.writeTermVote()
 	}
@@ -3603,7 +3568,6 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 	voteOk := n.vote == noVote || n.vote == vr.candidate
 	if voteOk && (vr.lastTerm > n.pterm || vr.lastTerm == n.pterm && vr.lastIndex >= n.pindex) {
 		vresp.granted = true
-		n.term = vr.term
 		n.vote = vr.candidate
 		n.writeTermVote()
 	} else {
diff --git a/server/server.go b/server/server.go
index 71c1fbad..782372a3 100644
--- a/server/server.go
+++ b/server/server.go
@@ -2993,7 +2993,7 @@ func (s *Server) readyForConnections(d time.Duration) error {
 		chk["server"] = info{ok: s.listener != nil || opts.DontListen, err: s.listenerErr}
 		chk["route"] = info{ok: (opts.Cluster.Port == 0 || s.routeListener != nil), err: s.routeListenerErr}
 		chk["gateway"] = info{ok: (opts.Gateway.Name == _EMPTY_ || s.gatewayListener != nil), err: s.gatewayListenerErr}
-		chk["leafnode"] = info{ok: (opts.LeafNode.Port == 0 || s.leafNodeListener != nil), err: s.leafNodeListenerErr}
+		chk["leafNode"] = info{ok: (opts.LeafNode.Port == 0 || s.leafNodeListener != nil), err: s.leafNodeListenerErr}
 		chk["websocket"] = info{ok: (opts.Websocket.Port == 0 || s.websocket.listener != nil), err: s.websocket.listenerErr}
 		chk["mqtt"] = info{ok: (opts.MQTT.Port == 0 || s.mqtt.listener != nil), err: s.mqtt.listenerErr}
 		s.mu.RUnlock()
diff --git a/server/stream.go b/server/stream.go
index 89cbb807..1683e790 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -210,7 +210,6 @@ type stream struct {
 	qch       chan struct{}
 	active    bool
 	ddloaded  bool
-	closed    bool
 
 	// Mirror
 	mirror *sourceInfo
@@ -231,10 +230,6 @@ type stream struct {
 	sigq  *ipQueue[*cMsg]
 	csl   *Sublist
 
-	// For non limits policy streams when they process an ack before the actual msg.
-	// Can happen in stretch clusters, multi-cloud, or during catchup for a restarted server.
-	preAcks map[uint64]map[*consumer]struct{}
-
 	// TODO(dlc) - Hide everything below behind two pointers.
 	// Clustered mode.
 	sa         *streamAssignment
@@ -250,7 +245,6 @@ type stream struct {
 	catchups   map[string]uint64
 	uch        chan struct{}
 	compressOK bool
-	inMonitor  bool
 
 	// Direct get subscription.
 	directSub *subscription
@@ -1691,7 +1685,6 @@ func (mset *stream) deleteMsg(seq uint64) (bool, error) {
 		return false, fmt.Errorf("invalid stream")
 	}
 	mset.mu.RUnlock()
-
 	return mset.store.RemoveMsg(seq)
 }
 
@@ -3310,8 +3303,6 @@ func (mset *stream) setupStore(fsCfg *FileStoreConfig) error {
 			return err
 		}
 		mset.store = fs
-		// Register our server.
-		fs.registerServer(s)
 	}
 	mset.mu.Unlock()
 
@@ -3682,7 +3673,7 @@ var (
 func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte, lseq uint64, ts int64) error {
 	mset.mu.Lock()
 	c, s, store := mset.client, mset.srv, mset.store
-	if mset.closed || c == nil {
+	if c == nil {
 		mset.mu.Unlock()
 		return nil
 	}
@@ -4004,14 +3995,7 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	} else {
 		// Make sure to take into account any message assignments that we had to skip (clfs).
 		seq = lseq + 1 - clfs
-		// Check for preAcks and the need to skip vs store.
-
-		if mset.hasAllPreAcks(seq, subject) {
-			mset.clearAllPreAcks(seq)
-			store.SkipMsg()
-		} else {
-			err = store.StoreRawMsg(subject, hdr, msg, seq, ts)
-		}
+		err = store.StoreRawMsg(subject, hdr, msg, seq, ts)
 	}
 
 	if err != nil {
@@ -4453,7 +4437,6 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 
 	// Clean up consumers.
 	mset.mu.Lock()
-	mset.closed = true
 	var obs []*consumer
 	for _, o := range mset.consumers {
 		obs = append(obs, o)
@@ -4633,12 +4616,6 @@ func (mset *stream) getPublicConsumers() []*consumer {
 	return obs
 }
 
-func (mset *stream) isInterestRetention() bool {
-	mset.mu.RLock()
-	defer mset.mu.RUnlock()
-	return mset.cfg.Retention != LimitsPolicy
-}
-
 // NumConsumers reports on number of active consumers for this stream.
 func (mset *stream) numConsumers() int {
 	mset.mu.RLock()
@@ -4828,162 +4805,35 @@ func (mset *stream) potentialFilteredConsumers() bool {
 	return false
 }
 
-// Check if there is no interest in this sequence number across our consumers.
-// The consumer passed is optional if we are processing the ack for that consumer.
-// Write lock should be held.
-func (mset *stream) noInterest(seq uint64, obs *consumer) bool {
-	return !mset.checkForInterest(seq, obs)
-}
-
-// Check if there is no interest in this sequence number and subject across our consumers.
-// The consumer passed is optional if we are processing the ack for that consumer.
-// Write lock should be held.
-func (mset *stream) noInterestWithSubject(seq uint64, subj string, obs *consumer) bool {
-	return !mset.checkForInterestWithSubject(seq, subj, obs)
-}
-
-// Write lock should be held here for the stream to avoid race conditions on state.
-func (mset *stream) checkForInterest(seq uint64, obs *consumer) bool {
+func (mset *stream) checkInterest(seq uint64, obs *consumer) bool {
 	var subj string
 	if mset.potentialFilteredConsumers() {
 		pmsg := getJSPubMsgFromPool()
 		defer pmsg.returnToPool()
 		sm, err := mset.store.LoadMsg(seq, &pmsg.StoreMsg)
 		if err != nil {
-			if err == ErrStoreEOF {
-				// Register this as a preAck.
-				mset.registerPreAck(obs, seq)
-				return true
-			}
-			mset.clearAllPreAcks(seq)
 			return false
 		}
 		subj = sm.subj
 	}
-	return mset.checkForInterestWithSubject(seq, subj, obs)
-}
-
-// Checks for interest given a sequence and subject.
-func (mset *stream) checkForInterestWithSubject(seq uint64, subj string, obs *consumer) bool {
 	for _, o := range mset.consumers {
-		// If this is us or we have a registered preAck for this consumer continue inspecting.
-		if o == obs || mset.hasPreAck(o, seq) {
-			continue
-		}
-		// Check if we need an ack.
-		if o.needAck(seq, subj) {
+		if o != obs && o.needAck(seq, subj) {
 			return true
 		}
 	}
-	mset.clearAllPreAcks(seq)
 	return false
 }
 
-// Check if we have a pre-registered ack for this sequence.
-// Write lock should be held.
-func (mset *stream) hasPreAck(o *consumer, seq uint64) bool {
-	if o == nil || len(mset.preAcks) == 0 {
-		return false
-	}
-	consumers := mset.preAcks[seq]
-	if len(consumers) == 0 {
-		return false
-	}
-	_, found := consumers[o]
-	return found
-}
-
-// Check if we have all consumers pre-acked for this sequence and subject.
-// Write lock should be held.
-func (mset *stream) hasAllPreAcks(seq uint64, subj string) bool {
-	if len(mset.preAcks) == 0 || len(mset.preAcks[seq]) == 0 {
-		return false
-	}
-	// Since these can be filtered and mutually exclusive,
-	// if we have some preAcks we need to check all interest here.
-	return mset.noInterestWithSubject(seq, subj, nil)
-}
-
-// Check if we have all consumers pre-acked.
-// Write lock should be held.
-func (mset *stream) clearAllPreAcks(seq uint64) {
-	delete(mset.preAcks, seq)
-}
-
-// Clear all preAcks below floor.
-// Write lock should be held.
-func (mset *stream) clearAllPreAcksBelowFloor(floor uint64) {
-	for seq := range mset.preAcks {
-		if seq < floor {
-			delete(mset.preAcks, seq)
-		}
-	}
-}
-
-// This will register an ack for a consumer if it arrives before the actual message.
-func (mset *stream) registerPreAckLock(o *consumer, seq uint64) {
-	mset.mu.Lock()
-	defer mset.mu.Unlock()
-	mset.registerPreAck(o, seq)
-}
-
-// This will register an ack for a consumer if it arrives before
-// the actual message.
-// Write lock should be held.
-func (mset *stream) registerPreAck(o *consumer, seq uint64) {
-	if o == nil {
-		return
-	}
-	if mset.preAcks == nil {
-		mset.preAcks = make(map[uint64]map[*consumer]struct{})
-	}
-	if mset.preAcks[seq] == nil {
-		mset.preAcks[seq] = make(map[*consumer]struct{})
-	}
-	mset.preAcks[seq][o] = struct{}{}
-}
-
-// This will clear an ack for a consumer.
-// Write lock should be held.
-func (mset *stream) clearPreAck(o *consumer, seq uint64) {
-	if o == nil || len(mset.preAcks) == 0 {
-		return
-	}
-	if consumers := mset.preAcks[seq]; len(consumers) > 0 {
-		delete(consumers, o)
-		if len(consumers) == 0 {
-			delete(mset.preAcks, seq)
-		}
-	}
-}
-
 // ackMsg is called into from a consumer when we have a WorkQueue or Interest Retention Policy.
 func (mset *stream) ackMsg(o *consumer, seq uint64) {
-	if seq == 0 || mset.cfg.Retention == LimitsPolicy {
-		return
-	}
-
-	// Don't make this RLock(). We need to have only 1 running at a time to gauge interest across all consumers.
-	mset.mu.Lock()
-	if mset.closed || mset.store == nil {
-		mset.mu.Unlock()
+	if mset.cfg.Retention == LimitsPolicy {
 		return
 	}
 
+	// Make sure this sequence is not below our first sequence.
 	var state StreamState
 	mset.store.FastState(&state)
-
-	// Make sure this sequence is not below our first sequence.
 	if seq < state.FirstSeq {
-		mset.clearPreAck(o, seq)
-		mset.mu.Unlock()
-		return
-	}
-
-	// If this has arrived before we have processed the message itself.
-	if seq > state.LastSeq {
-		mset.registerPreAck(o, seq)
-		mset.mu.Unlock()
 		return
 	}
 
@@ -4992,21 +4842,24 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
 	case WorkQueuePolicy:
 		// Normally we just remove a message when its ack'd here but if we have direct consumers
 		// from sources and/or mirrors we need to make sure they have delivered the msg.
-		shouldRemove = mset.directs <= 0 || mset.noInterest(seq, o)
+		mset.mu.RLock()
+		shouldRemove = mset.directs <= 0 || !mset.checkInterest(seq, o)
+		mset.mu.RUnlock()
 	case InterestPolicy:
-		shouldRemove = mset.noInterest(seq, o)
-	}
-	mset.mu.Unlock()
-
-	// If nothing else to do.
-	if !shouldRemove {
-		return
+		mset.mu.RLock()
+		shouldRemove = !mset.checkInterest(seq, o)
+		mset.mu.RUnlock()
 	}
 
-	// If we are here we should attempt to remove.
-	if _, err := mset.store.RemoveMsg(seq); err == ErrStoreEOF {
-		// This should not happen, but being pedantic.
-		mset.registerPreAckLock(o, seq)
+	if shouldRemove {
+		if _, err := mset.store.RemoveMsg(seq); err == ErrStoreEOF {
+			// This should be rare but I have seen it.
+			// The ack reached us before the actual msg with AckNone and InterestPolicy.
+			if n := mset.raftNode(); n != nil {
+				md := streamMsgDelete{Seq: seq, NoErase: true, Stream: mset.cfg.Name}
+				n.ForwardProposal(encodeMsgDelete(&md))
+			}
+		}
 	}
 }
 
@@ -5205,17 +5058,37 @@ func (a *Account) RestoreStream(ncfg *StreamConfig, r io.Reader) (*stream, error
 	return mset, nil
 }
 
-// This is to check for dangling messages on interest retention streams.
+// This is to check for dangling messages.
 // Issue https://github.com/nats-io/nats-server/issues/3612
 func (mset *stream) checkForOrphanMsgs() {
+	// We need to grab the low water mark for all consumers.
+	var ackFloor uint64
 	mset.mu.RLock()
-	consumers := make([]*consumer, 0, len(mset.consumers))
 	for _, o := range mset.consumers {
-		consumers = append(consumers, o)
+		o.mu.RLock()
+		if o.store != nil {
+			if state, err := o.store.BorrowState(); err == nil {
+				if ackFloor == 0 || state.AckFloor.Stream < ackFloor {
+					ackFloor = state.AckFloor.Stream
+				}
+			}
+		}
+		o.mu.RUnlock()
 	}
+	// Grabs stream state.
+	var state StreamState
+	mset.store.FastState(&state)
+	s, acc := mset.srv, mset.acc
 	mset.mu.RUnlock()
-	for _, o := range consumers {
-		o.checkStateForInterestStream()
+
+	if ackFloor > state.FirstSeq {
+		req := &JSApiStreamPurgeRequest{Sequence: ackFloor + 1}
+		purged, err := mset.purge(req)
+		if err != nil {
+			s.Warnf("stream '%s > %s' could not auto purge orphaned messages: %v", acc, mset.name(), err)
+		} else {
+			s.Debugf("stream '%s > %s' auto purged %d messages", acc, mset.name(), purged)
+		}
 	}
 }
 
@@ -5241,22 +5114,3 @@ func (mset *stream) checkConsumerReplication() {
 		o.mu.RUnlock()
 	}
 }
-
-// Will check if we are running in the monitor already and if not set the appropriate flag.
-func (mset *stream) checkInMonitor() bool {
-	mset.mu.Lock()
-	defer mset.mu.Unlock()
-
-	if mset.inMonitor {
-		return true
-	}
-	mset.inMonitor = true
-	return false
-}
-
-// Clear us being in the monitor routine.
-func (mset *stream) clearMonitorRunning() {
-	mset.mu.Lock()
-	defer mset.mu.Unlock()
-	mset.inMonitor = false
-}
diff --git a/server/websocket.go b/server/websocket.go
index a884b49b..fd06c986 100644
--- a/server/websocket.go
+++ b/server/websocket.go
@@ -1266,13 +1266,7 @@ func (c *client) wsCollapsePtoNB() (net.Buffers, int64) {
 	if c.ws.browser {
 		mfs = wsFrameSizeForBrowsers
 	}
-	if len(c.out.p) > 0 {
-		p := c.out.p
-		c.out.p = nil
-		nb = append(c.out.nb, p)
-	} else if len(c.out.nb) > 0 {
-		nb = c.out.nb
-	}
+	nb = c.out.nb
 	mask := c.ws.maskwrite
 	// Start with possible already framed buffers (that we could have
 	// got from partials or control messages such as ws pings or pongs).


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index 623ba9c0..02b6eb04 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -2099,12 +2099,17 @@ func TestCrossAccountServiceResponseTypes(t *testing.T) {
 
 	cfoo.parseAsync(string(mReply))
 
-	var b [256]byte
-	n, err := crBar.Read(b[:])
-	if err != nil {
-		t.Fatalf("Error reading response: %v", err)
+	var buf []byte
+	for i := 0; i < 20; i++ {
+		b, err := crBar.ReadBytes('\n')
+		if err != nil {
+			t.Fatalf("Error reading response: %v", err)
+		}
+		buf = append(buf[:], b...)
+		if mraw = msgPat.FindAllStringSubmatch(string(buf), -1); len(mraw) == 10 {
+			break
+		}
 	}
-	mraw = msgPat.FindAllStringSubmatch(string(b[:n]), -1)
 	if len(mraw) != 10 {
 		t.Fatalf("Expected a response but got %d", len(mraw))
 	}
diff --git a/server/client_test.go b/server/client_test.go
index b578db81..bc1a9df9 100644
--- a/server/client_test.go
+++ b/server/client_test.go
@@ -31,7 +31,6 @@ import (
 	"testing"
 	"time"
 
-	"crypto/rand"
 	"crypto/tls"
 
 	"github.com/nats-io/jwt/v2"
@@ -1484,7 +1483,11 @@ func TestWildcardCharsInLiteralSubjectWorks(t *testing.T) {
 	}
 }
 
-func TestDynamicBuffers(t *testing.T) {
+// This test ensures that coalescing into the fixed-size output
+// queues works as expected. When bytes are queued up, they should
+// not overflow a buffer until the capacity is exceeded, at which
+// point a new buffer should be added.
+func TestClientOutboundQueueCoalesce(t *testing.T) {
 	opts := DefaultOptions()
 	s := RunServer(opts)
 	defer s.Shutdown()
@@ -1495,139 +1498,114 @@ func TestDynamicBuffers(t *testing.T) {
 	}
 	defer nc.Close()
 
-	// Grab the client from server.
-	s.mu.Lock()
-	lc := len(s.clients)
-	c := s.clients[s.gcid]
-	s.mu.Unlock()
+	clients := s.GlobalAccount().getClients()
+	if len(clients) != 1 {
+		t.Fatal("Expecting a client to exist")
+	}
+	client := clients[0]
+	client.mu.Lock()
+	defer client.mu.Unlock()
 
-	if lc != 1 {
-		t.Fatalf("Expected only 1 client but got %d\n", lc)
+	// First up, queue something small into the queue.
+	client.queueOutbound([]byte{1, 2, 3, 4, 5})
+
+	if len(client.out.nb) != 1 {
+		t.Fatal("Expecting a single queued buffer")
 	}
-	if c == nil {
-		t.Fatal("Expected to retrieve client\n")
+	if l := len(client.out.nb[0]); l != 5 {
+		t.Fatalf("Expecting only 5 bytes in the first queued buffer, found %d instead", l)
 	}
 
-	// Create some helper functions and data structures.
-	done := make(chan bool)            // Used to stop recording.
-	type maxv struct{ rsz, wsz int32 } // Used to hold max values.
-	results := make(chan maxv)
+	// Then queue up a few more bytes, but not enough
+	// to overflow into the next buffer.
+	client.queueOutbound([]byte{6, 7, 8, 9, 10})
 
-	// stopRecording stops the recording ticker and releases go routine.
-	stopRecording := func() maxv {
-		done <- true
-		return <-results
+	if len(client.out.nb) != 1 {
+		t.Fatal("Expecting a single queued buffer")
 	}
-	// max just grabs max values.
-	max := func(a, b int32) int32 {
-		if a > b {
-			return a
-		}
-		return b
+	if l := len(client.out.nb[0]); l != 10 {
+		t.Fatalf("Expecting 10 bytes in the first queued buffer, found %d instead", l)
 	}
-	// Returns current value of the buffer sizes.
-	getBufferSizes := func() (int32, int32) {
-		c.mu.Lock()
-		defer c.mu.Unlock()
-		return c.in.rsz, c.out.sz
+
+	// Finally, queue up something that is guaranteed
+	// to overflow.
+	b := nbPoolSmall.Get().(*[nbPoolSizeSmall]byte)[:]
+	b = b[:cap(b)]
+	client.queueOutbound(b)
+	if len(client.out.nb) != 2 {
+		t.Fatal("Expecting buffer to have overflowed")
 	}
-	// Record the max values seen.
-	recordMaxBufferSizes := func() {
-		ticker := time.NewTicker(10 * time.Microsecond)
-		defer ticker.Stop()
+	if l := len(client.out.nb[0]); l != cap(b) {
+		t.Fatalf("Expecting %d bytes in the first queued buffer, found %d instead", cap(b), l)
+	}
+	if l := len(client.out.nb[1]); l != 10 {
+		t.Fatalf("Expecting 10 bytes in the second queued buffer, found %d instead", l)
+	}
+}
 
-		var m maxv
+// This test ensures that outbound queues don't cause a run on
+// memory when sending something to lots of clients.
+func TestClientOutboundQueueMemory(t *testing.T) {
+	opts := DefaultOptions()
+	s := RunServer(opts)
+	defer s.Shutdown()
 
-		recordMax := func() {
-			rsz, wsz := getBufferSizes()
-			m.rsz = max(m.rsz, rsz)
-			m.wsz = max(m.wsz, wsz)
-		}
+	var before runtime.MemStats
+	var after runtime.MemStats
 
-		for {
-			select {
-			case <-done:
-				recordMax()
-				results <- m
-				return
-			case <-ticker.C:
-				recordMax()
-			}
-		}
-	}
-	// Check that the current value is what we expected.
-	checkBuffers := func(ers, ews int32) {
-		t.Helper()
-		rsz, wsz := getBufferSizes()
-		if rsz != ers {
-			t.Fatalf("Expected read buffer of %d, but got %d\n", ers, rsz)
-		}
-		if wsz != ews {
-			t.Fatalf("Expected write buffer of %d, but got %d\n", ews, wsz)
-		}
-	}
+	var err error
+	clients := make([]*nats.Conn, 50000)
+	wait := &sync.WaitGroup{}
+	wait.Add(len(clients))
 
-	// Check that the max was as expected.
-	checkResults := func(m maxv, rsz, wsz int32) {
-		t.Helper()
-		if rsz != m.rsz {
-			t.Fatalf("Expected read buffer of %d, but got %d\n", rsz, m.rsz)
-		}
-		if wsz != m.wsz {
-			t.Fatalf("Expected write buffer of %d, but got %d\n", wsz, m.wsz)
+	for i := 0; i < len(clients); i++ {
+		clients[i], err = nats.Connect(fmt.Sprintf("nats://%s:%d", opts.Host, opts.Port), nats.InProcessServer(s))
+		if err != nil {
+			t.Fatalf("Error on connect: %v", err)
 		}
-	}
-
-	// Here is where testing begins..
+		defer clients[i].Close()
 
-	// Should be at or below the startBufSize for both.
-	rsz, wsz := getBufferSizes()
-	if rsz > startBufSize {
-		t.Fatalf("Expected read buffer of <= %d, but got %d\n", startBufSize, rsz)
-	}
-	if wsz > startBufSize {
-		t.Fatalf("Expected write buffer of <= %d, but got %d\n", startBufSize, wsz)
+		clients[i].Subscribe("test", func(m *nats.Msg) {
+			wait.Done()
+		})
 	}
 
-	// Send some data.
-	data := make([]byte, 2048)
-	rand.Read(data)
+	runtime.GC()
+	runtime.ReadMemStats(&before)
 
-	go recordMaxBufferSizes()
-	for i := 0; i < 200; i++ {
-		nc.Publish("foo", data)
+	nc, err := nats.Connect(fmt.Sprintf("nats://%s:%d", opts.Host, opts.Port), nats.InProcessServer(s))
+	if err != nil {
+		t.Fatalf("Error on connect: %v", err)
 	}
-	nc.Flush()
-	m := stopRecording()
+	defer nc.Close()
 
-	if m.rsz != maxBufSize && m.rsz != maxBufSize/2 {
-		t.Fatalf("Expected read buffer of %d or %d, but got %d\n", maxBufSize, maxBufSize/2, m.rsz)
-	}
-	if m.wsz > startBufSize {
-		t.Fatalf("Expected write buffer of <= %d, but got %d\n", startBufSize, m.wsz)
+	var m [48000]byte
+	if err = nc.Publish("test", m[:]); err != nil {
+		t.Fatal(err)
 	}
 
-	// Create Subscription to test outbound buffer from server.
-	nc.Subscribe("foo", func(m *nats.Msg) {
-		// Just eat it..
-	})
-	go recordMaxBufferSizes()
+	wait.Wait()
 
-	for i := 0; i < 200; i++ {
-		nc.Publish("foo", data)
-	}
-	nc.Flush()
+	runtime.GC()
+	runtime.ReadMemStats(&after)
 
-	m = stopRecording()
-	checkResults(m, maxBufSize, maxBufSize)
+	hb, ha := float64(before.HeapAlloc), float64(after.HeapAlloc)
+	ms := float64(len(m))
+	diff := float64(ha) - float64(hb)
+	inc := (diff / float64(hb)) * 100
 
-	// Now test that we shrink correctly.
+	fmt.Printf("Message size:       %.1fKB\n", ms/1024)
+	fmt.Printf("Subscribed clients: %d\n", len(clients))
+	fmt.Printf("Heap allocs before: %.1fMB\n", hb/1024/1024)
+	fmt.Printf("Heap allocs after:  %.1fMB\n", ha/1024/1024)
+	fmt.Printf("Heap allocs delta:  %.1f%%\n", inc)
 
-	// Should go to minimum for both..
-	for i := 0; i < 20; i++ {
-		nc.Flush()
-	}
-	checkBuffers(minBufSize, minBufSize)
+	// TODO: What threshold makes sense here for a failure?
+	/*
+		if inc > 10 {
+			t.Fatalf("memory increase was %.1f%% (should be <= 10%%)", inc)
+		}
+	*/
 }
 
 func TestClientTraceRace(t *testing.T) {
@@ -2246,7 +2224,6 @@ func TestFlushOutboundNoSliceReuseIfPartial(t *testing.T) {
 		expected.Write(buf)
 		c.mu.Lock()
 		c.queueOutbound(buf)
-		c.out.sz = 10
 		c.flushOutbound()
 		fakeConn.partial = false
 		c.mu.Unlock()
diff --git a/server/filestore_test.go b/server/filestore_test.go
index e3050018..92ccfd2f 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -2776,8 +2776,8 @@ func TestFileStoreConsumerDeliveredAndAckUpdates(t *testing.T) {
 			}
 		}
 
-		testAck(1, 100, 1, 109)
-		testAck(3, 130, 1, 109)
+		testAck(1, 100, 1, 100)
+		testAck(3, 130, 1, 100)
 		testAck(2, 110, 3, 149) // We do not track explicit state on previous stream floors, so we take last known -1
 		testAck(5, 165, 3, 149)
 		testAck(4, 150, 5, 165)
@@ -3668,7 +3668,7 @@ func TestFileStoreFetchPerf(t *testing.T) {
 // https://github.com/nats-io/nats-server/issues/2936
 func TestFileStoreCompactReclaimHeadSpace(t *testing.T) {
 	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
-		fcfg.BlockSize = 4 * 1024 * 1024
+		fcfg.BlockSize = 1024 * 1024
 
 		fs, err := newFileStore(
 			fcfg,
@@ -3678,7 +3678,7 @@ func TestFileStoreCompactReclaimHeadSpace(t *testing.T) {
 		defer fs.Stop()
 
 		// Create random bytes for payload to test for corruption vs repeated.
-		msg := make([]byte, 64*1024)
+		msg := make([]byte, 16*1024)
 		crand.Read(msg)
 
 		// This gives us ~63 msgs in first and ~37 in second.
@@ -5372,3 +5372,126 @@ func TestFileStoreSubjectsTotals(t *testing.T) {
 		t.Fatalf("Expected %d subjects for %q, got %d", expected, "*.*", len(st))
 	}
 }
+
+func TestFileStoreNumPending(t *testing.T) {
+	// No need for all permutations here.
+	storeDir := t.TempDir()
+	fcfg := FileStoreConfig{
+		StoreDir:  storeDir,
+		BlockSize: 2 * 1024, // Create many blocks on purpose.
+	}
+	fs, err := newFileStore(fcfg, StreamConfig{Name: "zzz", Subjects: []string{"*.*.*.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	tokens := []string{"foo", "bar", "baz"}
+	genSubj := func() string {
+		return fmt.Sprintf("%s.%s.%s.%s",
+			tokens[rand.Intn(len(tokens))],
+			tokens[rand.Intn(len(tokens))],
+			tokens[rand.Intn(len(tokens))],
+			tokens[rand.Intn(len(tokens))],
+		)
+	}
+
+	for i := 0; i < 50_000; i++ {
+		subj := genSubj()
+		_, _, err := fs.StoreMsg(subj, nil, []byte("Hello World"))
+		require_NoError(t, err)
+	}
+
+	state := fs.State()
+
+	// Scan one by one for sanity check against other calculations.
+	sanityCheck := func(sseq uint64, filter string) SimpleState {
+		t.Helper()
+		var ss SimpleState
+		var smv StoreMsg
+		// For here we know 0 is invalid, set to 1.
+		if sseq == 0 {
+			sseq = 1
+		}
+		for seq := sseq; seq <= state.LastSeq; seq++ {
+			sm, err := fs.LoadMsg(seq, &smv)
+			if err != nil {
+				t.Logf("Encountered error %v loading sequence: %d", err, seq)
+				continue
+			}
+			if subjectIsSubsetMatch(sm.subj, filter) {
+				ss.Msgs++
+				ss.Last = seq
+				if ss.First == 0 || seq < ss.First {
+					ss.First = seq
+				}
+			}
+		}
+		return ss
+	}
+
+	check := func(sseq uint64, filter string) {
+		t.Helper()
+		np, lvs := fs.NumPending(sseq, filter, false)
+		ss := fs.FilteredState(sseq, filter)
+		sss := sanityCheck(sseq, filter)
+		if lvs != state.LastSeq {
+			t.Fatalf("Expected NumPending to return valid through last of %d but got %d", state.LastSeq, lvs)
+		}
+		if ss.Msgs != np {
+			t.Fatalf("NumPending of %d did not match ss.Msgs of %d", np, ss.Msgs)
+		}
+		if ss != sss {
+			t.Fatalf("Failed sanity check, expected %+v got %+v", sss, ss)
+		}
+	}
+
+	sanityCheckLastOnly := func(sseq uint64, filter string) SimpleState {
+		t.Helper()
+		var ss SimpleState
+		var smv StoreMsg
+		// For here we know 0 is invalid, set to 1.
+		if sseq == 0 {
+			sseq = 1
+		}
+		seen := make(map[string]bool)
+		for seq := state.LastSeq; seq >= sseq; seq-- {
+			sm, err := fs.LoadMsg(seq, &smv)
+			if err != nil {
+				t.Logf("Encountered error %v loading sequence: %d", err, seq)
+				continue
+			}
+			if !seen[sm.subj] && subjectIsSubsetMatch(sm.subj, filter) {
+				ss.Msgs++
+				if ss.Last == 0 {
+					ss.Last = seq
+				}
+				if ss.First == 0 || seq < ss.First {
+					ss.First = seq
+				}
+				seen[sm.subj] = true
+			}
+		}
+		return ss
+	}
+
+	checkLastOnly := func(sseq uint64, filter string) {
+		t.Helper()
+		np, lvs := fs.NumPending(sseq, filter, true)
+		ss := sanityCheckLastOnly(sseq, filter)
+		if lvs != state.LastSeq {
+			t.Fatalf("Expected NumPending to return valid through last of %d but got %d", state.LastSeq, lvs)
+		}
+		if ss.Msgs != np {
+			t.Fatalf("NumPending of %d did not match ss.Msgs of %d", np, ss.Msgs)
+		}
+	}
+
+	startSeqs := []uint64{0, 1, 2, 200, 444, 555, 2222, 8888, 12_345, 28_222, 33_456, 44_400, 49_999}
+	checkSubs := []string{"foo.>", "*.bar.>", "foo.bar.*.baz", "*.bar.>", "*.foo.bar.*", "foo.foo.bar.baz"}
+
+	for _, filter := range checkSubs {
+		for _, start := range startSeqs {
+			check(start, filter)
+			checkLastOnly(start, filter)
+		}
+	}
+}
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index 20566f01..b899f0e3 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -971,7 +971,6 @@ func TestJetStreamClusterRestoreSingleConsumer(t *testing.T) {
 	c.stopAll()
 	c.restartAll()
 	c.waitOnLeader()
-	c.waitOnStreamLeader("$G", "foo")
 
 	s = c.randomServer()
 	nc, js = jsClientConnect(t, s)
@@ -3720,7 +3719,6 @@ func TestJetStreamClusterAccountPurge(t *testing.T) {
 				resolver: {
 					type: full
 					dir: '%s/jwt'
-					timeout: "10ms"
 				}`, ojwt, syspub, storeDir)
 		})
 	defer c.shutdown()
@@ -3893,6 +3891,7 @@ func TestJetStreamClusterAccountPurge(t *testing.T) {
 		}
 		c.restartAll()
 		checkForDirs(t, 6, 4)
+		c.waitOnClusterReady() // unfortunately, this does not wait until leader is not catching up.
 		purge(t)
 		checkForDirs(t, 0, 0)
 		c.stopAll()
diff --git a/server/jetstream_cluster_2_test.go b/server/jetstream_cluster_2_test.go
index a13ada9c..bd530ca1 100644
--- a/server/jetstream_cluster_2_test.go
+++ b/server/jetstream_cluster_2_test.go
@@ -4322,7 +4322,7 @@ func TestJetStreamClusterStreamReplicaUpdates(t *testing.T) {
 		require_NoError(t, err)
 		c.waitOnStreamLeader("$G", "TEST")
 
-		checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
+		checkFor(t, 5*time.Second, 100*time.Millisecond, func() error {
 			si, err = js.StreamInfo("TEST")
 			require_NoError(t, err)
 			if len(si.Cluster.Replicas) != r-1 {
@@ -4936,11 +4936,11 @@ func TestJetStreamClusterDuplicateMsgIdsOnCatchupAndLeaderTakeover(t *testing.T)
 	// Now restart
 	sr = c.restartServer(sr)
 	c.waitOnStreamCurrent(sr, "$G", "TEST")
-	c.waitOnStreamLeader("$G", "TEST")
 
 	// Now make them the leader.
 	for sr != c.streamLeader("$G", "TEST") {
-		nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "TEST"), nil, time.Second)
+		_, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "TEST"), nil, time.Second)
+		require_NoError(t, err)
 		c.waitOnStreamLeader("$G", "TEST")
 	}
 
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 88f55b64..05c1a45e 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -1,4 +1,4 @@
-// Copyright 2022-2023 The NATS Authors
+// Copyright 2022 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -313,7 +313,7 @@ func TestJetStreamClusterDeleteConsumerWhileServerDown(t *testing.T) {
 
 	// Restart.
 	s = c.restartServer(s)
-	checkFor(t, 10*time.Second, 200*time.Millisecond, func() error {
+	checkFor(t, time.Second, 200*time.Millisecond, func() error {
 		hs := s.healthz(&HealthzOptions{
 			JSEnabledOnly: false,
 			JSServerOnly:  false,
@@ -1046,7 +1046,6 @@ func TestJetStreamClusterSourceWithOptStartTime(t *testing.T) {
 			sd := s.JetStreamConfig().StoreDir
 			s.Shutdown()
 			s = RunJetStreamServerOnPort(-1, sd)
-			defer s.Shutdown()
 		}
 
 		// Wait a bit before checking because sync'ing (even with the defect)
@@ -1591,7 +1590,7 @@ func TestJetStreamGhostEphemeralsAfterRestart(t *testing.T) {
 	defer nc.Close()
 
 	subj := fmt.Sprintf(JSApiConsumerListT, "TEST")
-	checkFor(t, 10*time.Second, 200*time.Millisecond, func() error {
+	checkFor(t, 5*time.Second, 200*time.Millisecond, func() error {
 		m, err := nc.Request(subj, nil, time.Second)
 		if err != nil {
 			return err
@@ -2729,16 +2728,13 @@ func TestJetStreamClusterInterestPolicyEphemeral(t *testing.T) {
 			}
 
 			const msgs = 5_000
-			done, count := make(chan bool, 1), 0
+			done, count := make(chan bool), 0
 
 			sub, err := js.Subscribe(_EMPTY_, func(msg *nats.Msg) {
 				require_NoError(t, msg.Ack())
 				count++
 				if count >= msgs {
-					select {
-					case done <- true:
-					default:
-					}
+					done <- true
 				}
 			}, nats.Bind(test.stream, name), nats.ManualAck())
 			require_NoError(t, err)
@@ -2962,7 +2958,6 @@ func TestJetStreamClusterWorkQueueConsumerReplicatedAfterScaleUp(t *testing.T) {
 
 	require_True(t, ci.Config.Replicas == 0 || ci.Config.Replicas == 3)
 
-	c.waitOnConsumerLeader(globalAccountName, "TEST", ci.Name)
 	s := c.consumerLeader(globalAccountName, "TEST", ci.Name)
 	require_NotNil(t, s)
 
@@ -3016,497 +3011,7 @@ func TestJetStreamClusterWorkQueueAfterScaleUp(t *testing.T) {
 	sendStreamMsg(t, nc, "WQ", "SOME WORK")
 	<-wch
 
-	checkFor(t, time.Second, 200*time.Millisecond, func() error {
-		si, err := js.StreamInfo("TEST")
-		require_NoError(t, err)
-		if si.State.Msgs == 0 {
-			return nil
-		}
-		return fmt.Errorf("Still have %d msgs left", si.State.Msgs)
-	})
-}
-
-func TestJetStreamClusterInterestBasedStreamAndConsumerSnapshots(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",
-		Replicas:  3,
-		Subjects:  []string{"foo"},
-		Retention: nats.InterestPolicy,
-	})
-	require_NoError(t, err)
-
-	sub, err := js.SubscribeSync("foo", nats.Durable("d22"))
-	require_NoError(t, err)
-
-	num := 200
-	for i := 0; i < num; i++ {
-		js.PublishAsync("foo", []byte("ok"))
-	}
-	select {
-	case <-js.PublishAsyncComplete():
-	case <-time.After(5 * time.Second):
-		t.Fatalf("Did not receive completion signal")
-	}
-
-	checkSubsPending(t, sub, num)
-
-	// Shutdown one server.
-	s := c.randomServer()
-	s.Shutdown()
-
-	c.waitOnStreamLeader(globalAccountName, "TEST")
-
-	nc, js = jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	// Now ack all messages while the other server is down.
-	for i := 0; i < num; i++ {
-		m, err := sub.NextMsg(time.Second)
-		require_NoError(t, err)
-		m.AckSync()
-	}
-
-	// Wait for all message acks to be processed and all messages to be removed.
-	checkFor(t, time.Second, 200*time.Millisecond, func() error {
-		si, err := js.StreamInfo("TEST")
-		require_NoError(t, err)
-		if si.State.Msgs == 0 {
-			return nil
-		}
-		return fmt.Errorf("Still have %d msgs left", si.State.Msgs)
-	})
-
-	// Force a snapshot on the consumer leader before restarting the downed server.
-	cl := c.consumerLeader(globalAccountName, "TEST", "d22")
-	require_NotNil(t, cl)
-
-	mset, err := cl.GlobalAccount().lookupStream("TEST")
-	require_NoError(t, err)
-
-	o := mset.lookupConsumer("d22")
-	require_NotNil(t, o)
-
-	snap, err := o.store.EncodedState()
-	require_NoError(t, err)
-
-	n := o.raftNode()
-	require_NotNil(t, n)
-	require_NoError(t, n.InstallSnapshot(snap))
-
-	// Now restart the downed server.
-	s = c.restartServer(s)
-
-	// Make the restarted server the eventual leader.
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnStreamLeader(globalAccountName, "TEST")
-		if sl := c.streamLeader(globalAccountName, "TEST"); sl != s {
-			sl.JetStreamStepdownStream(globalAccountName, "TEST")
-			return fmt.Errorf("Server %s is not leader yet", s)
-		}
-		return nil
-	})
-
 	si, err := js.StreamInfo("TEST")
 	require_NoError(t, err)
 	require_True(t, si.State.Msgs == 0)
 }
-
-func TestJetStreamClusterConsumerFollowerStoreStateAckFloorBug(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",
-		Replicas: 3,
-		Subjects: []string{"foo"},
-	})
-	require_NoError(t, err)
-
-	sub, err := js.PullSubscribe(_EMPTY_, "C", nats.BindStream("TEST"), nats.ManualAck())
-	require_NoError(t, err)
-
-	num := 100
-	for i := 0; i < num; i++ {
-		sendStreamMsg(t, nc, "foo", "data")
-	}
-
-	// This one prevents the state for pending from reaching 0 and resetting, which would not show the bug.
-	sendStreamMsg(t, nc, "foo", "data")
-
-	// Ack all but one and out of order and make sure all consumers have the same stored state.
-	msgs, err := sub.Fetch(num, nats.MaxWait(time.Second))
-	require_NoError(t, err)
-	require_True(t, len(msgs) == num)
-
-	_, err = sub.Fetch(1, nats.MaxWait(time.Second))
-	require_NoError(t, err)
-
-	rand.Shuffle(len(msgs), func(i, j int) { msgs[i], msgs[j] = msgs[j], msgs[i] })
-	for _, m := range msgs {
-		m.AckSync()
-	}
-
-	checkConsumerState := func(delivered, ackFloor nats.SequenceInfo, numAckPending int) {
-		expectedDelivered := uint64(num) + 1
-		if delivered.Stream != expectedDelivered || delivered.Consumer != expectedDelivered {
-			t.Fatalf("Wrong delivered, expected %d got %+v", expectedDelivered, delivered)
-		}
-		expectedAck := uint64(num)
-		if ackFloor.Stream != expectedAck || ackFloor.Consumer != expectedAck {
-			t.Fatalf("Wrong ackFloor, expected %d got %+v", expectedAck, ackFloor)
-		}
-		require_True(t, numAckPending == 1)
-	}
-
-	ci, err := js.ConsumerInfo("TEST", "C")
-	require_NoError(t, err)
-	checkConsumerState(ci.Delivered, ci.AckFloor, ci.NumAckPending)
-
-	// Check each consumer on each server for it's store state and make sure it matches as well.
-	for _, s := range c.servers {
-		mset, err := s.GlobalAccount().lookupStream("TEST")
-		require_NoError(t, err)
-		require_NotNil(t, mset)
-		o := mset.lookupConsumer("C")
-		require_NotNil(t, o)
-
-		state, err := o.store.State()
-		require_NoError(t, err)
-
-		delivered := nats.SequenceInfo{Stream: state.Delivered.Stream, Consumer: state.Delivered.Consumer}
-		ackFloor := nats.SequenceInfo{Stream: state.AckFloor.Stream, Consumer: state.AckFloor.Consumer}
-		checkConsumerState(delivered, ackFloor, len(state.Pending))
-	}
-
-	// Now stepdown the consumer and move its leader and check the state after transition.
-	// Make the restarted server the eventual leader.
-	seen := make(map[*Server]bool)
-	cl := c.consumerLeader(globalAccountName, "TEST", "C")
-	require_NotNil(t, cl)
-	seen[cl] = true
-
-	allSeen := func() bool {
-		for _, s := range c.servers {
-			if !seen[s] {
-				return false
-			}
-		}
-		return true
-	}
-
-	checkAllLeaders := func() {
-		t.Helper()
-		checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-			c.waitOnConsumerLeader(globalAccountName, "TEST", "C")
-			if allSeen() {
-				return nil
-			}
-			cl := c.consumerLeader(globalAccountName, "TEST", "C")
-			seen[cl] = true
-			ci, err := js.ConsumerInfo("TEST", "C")
-			require_NoError(t, err)
-			checkConsumerState(ci.Delivered, ci.AckFloor, ci.NumAckPending)
-			cl.JetStreamStepdownConsumer(globalAccountName, "TEST", "C")
-			return fmt.Errorf("Not all servers have been consumer leader yet")
-		})
-	}
-
-	checkAllLeaders()
-
-	// No restart all servers and check again.
-	c.stopAll()
-	c.restartAll()
-	c.waitOnLeader()
-
-	nc, js = jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	seen = make(map[*Server]bool)
-	checkAllLeaders()
-}
-
-func TestNoRaceJetStreamClusterDifferentRTTInterestBasedStreamPreAck(t *testing.T) {
-	tmpl := `
-	listen: 127.0.0.1:-1
-	server_name: %s
-	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
-
-	cluster {
-		name: "F3"
-		listen: 127.0.0.1:%d
-		routes = [%s]
-	}
-
-	accounts {
-		$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
-	}
-	`
-
-	//  Route Ports
-	//	"S1": 14622,
-	//	"S2": 15622,
-	//	"S3": 16622,
-
-	// S2 (stream leader) will have a slow path to S1 (via proxy) and S3 (consumer leader) will have a fast path.
-
-	// Do these in order, S1, S2 (proxy) then S3.
-	c := &cluster{t: t, servers: make([]*Server, 3), opts: make([]*Options, 3), name: "F3"}
-
-	// S1
-	conf := fmt.Sprintf(tmpl, "S1", t.TempDir(), 14622, "route://127.0.0.1:15622, route://127.0.0.1:16622")
-	c.servers[0], c.opts[0] = RunServerWithConfig(createConfFile(t, []byte(conf)))
-
-	// S2
-	// Create the proxy first. Connect this to S1. Make it slow, e.g. 5ms RTT.
-	np := createNetProxy(1*time.Millisecond, 1024*1024*1024, 1024*1024*1024, "route://127.0.0.1:14622", true)
-	routes := fmt.Sprintf("%s, route://127.0.0.1:16622", np.routeURL())
-	conf = fmt.Sprintf(tmpl, "S2", t.TempDir(), 15622, routes)
-	c.servers[1], c.opts[1] = RunServerWithConfig(createConfFile(t, []byte(conf)))
-
-	// S3
-	conf = fmt.Sprintf(tmpl, "S3", t.TempDir(), 16622, "route://127.0.0.1:14622, route://127.0.0.1:15622")
-	c.servers[2], c.opts[2] = RunServerWithConfig(createConfFile(t, []byte(conf)))
-
-	c.checkClusterFormed()
-	c.waitOnClusterReady()
-	defer c.shutdown()
-	defer np.stop()
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	// Now create the stream.
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:      "EVENTS",
-		Subjects:  []string{"EV.>"},
-		Replicas:  3,
-		Retention: nats.InterestPolicy,
-	})
-	require_NoError(t, err)
-
-	// Make sure it's leader is on S2.
-	sl := c.servers[1]
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnStreamLeader(globalAccountName, "EVENTS")
-		if s := c.streamLeader(globalAccountName, "EVENTS"); s != sl {
-			s.JetStreamStepdownStream(globalAccountName, "EVENTS")
-			return fmt.Errorf("Server %s is not stream leader yet", sl)
-		}
-		return nil
-	})
-
-	// Now create the consumer.
-	_, err = js.AddConsumer("EVENTS", &nats.ConsumerConfig{
-		Durable:        "C",
-		AckPolicy:      nats.AckExplicitPolicy,
-		DeliverSubject: "dx",
-	})
-	require_NoError(t, err)
-
-	// Make sure the consumer leader is on S3.
-	cl := c.servers[2]
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnConsumerLeader(globalAccountName, "EVENTS", "C")
-		if s := c.consumerLeader(globalAccountName, "EVENTS", "C"); s != cl {
-			s.JetStreamStepdownConsumer(globalAccountName, "EVENTS", "C")
-			return fmt.Errorf("Server %s is not consumer leader yet", sl)
-		}
-		return nil
-	})
-
-	// Create the real consumer on the consumer leader to make it efficient.
-	nc, js = jsClientConnect(t, cl)
-	defer nc.Close()
-
-	_, err = js.Subscribe(_EMPTY_, func(msg *nats.Msg) {
-		msg.Ack()
-	}, nats.BindStream("EVENTS"), nats.Durable("C"), nats.ManualAck())
-	require_NoError(t, err)
-
-	for i := 0; i < 1_000; i++ {
-		_, err := js.PublishAsync("EVENTS.PAID", []byte("ok"))
-		require_NoError(t, err)
-	}
-	select {
-	case <-js.PublishAsyncComplete():
-	case <-time.After(5 * time.Second):
-		t.Fatalf("Did not receive completion signal")
-	}
-
-	slow := c.servers[0]
-	mset, err := slow.GlobalAccount().lookupStream("EVENTS")
-	require_NoError(t, err)
-
-	// Make sure preAck is non-nil, so we know the logic has kicked in.
-	mset.mu.RLock()
-	preAcks := mset.preAcks
-	mset.mu.RUnlock()
-	require_NotNil(t, preAcks)
-
-	checkFor(t, 5*time.Second, 200*time.Millisecond, func() error {
-		state := mset.state()
-		if state.Msgs == 0 {
-			mset.mu.RLock()
-			lp := len(mset.preAcks)
-			mset.mu.RUnlock()
-			if lp == 0 {
-				return nil
-			} else {
-				t.Fatalf("Expected no preAcks with no msgs, but got %d", lp)
-			}
-		}
-		return fmt.Errorf("Still have %d msgs left", state.Msgs)
-	})
-
-}
-
-func TestJetStreamInterestLeakOnDisableJetStream(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
-	defer c.shutdown()
-
-	nc, js := jsClientConnect(t, c.leader())
-	defer nc.Close()
-
-	for i := 1; i <= 5; i++ {
-		_, err := js.AddStream(&nats.StreamConfig{
-			Name:     fmt.Sprintf("test_%d", i),
-			Subjects: []string{fmt.Sprintf("test_%d", i)},
-			Replicas: 3,
-		})
-		require_NoError(t, err)
-	}
-
-	c.waitOnAllCurrent()
-
-	server := c.randomNonLeader()
-	account := server.SystemAccount()
-
-	server.DisableJetStream()
-
-	var sublist []*subscription
-	account.sl.localSubs(&sublist, false)
-
-	var danglingJSC, danglingRaft int
-	for _, sub := range sublist {
-		if strings.HasPrefix(string(sub.subject), "$JSC.") {
-			danglingJSC++
-		} else if strings.HasPrefix(string(sub.subject), "$NRG.") {
-			danglingRaft++
-		}
-	}
-	if danglingJSC > 0 || danglingRaft > 0 {
-		t.Fatalf("unexpected dangling interests for JetStream assets after shutdown (%d $JSC, %d $NRG)", danglingJSC, danglingRaft)
-	}
-}
-
-func TestJetStreamNoLeadersDuringLameDuck(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
-	defer c.shutdown()
-
-	// Grab the first server and set lameduck option directly.
-	s := c.servers[0]
-	s.optsMu.Lock()
-	s.opts.LameDuckDuration = 5 * time.Second
-	s.opts.LameDuckGracePeriod = -5 * time.Second
-	s.optsMu.Unlock()
-
-	// Connect to the third server.
-	nc, js := jsClientConnect(t, c.servers[2])
-	defer nc.Close()
-
-	allServersHaveLeaders := func() bool {
-		haveLeader := make(map[*Server]bool)
-		for _, s := range c.servers {
-			s.rnMu.RLock()
-			for _, n := range s.raftNodes {
-				if n.Leader() {
-					haveLeader[s] = true
-					break
-				}
-			}
-			s.rnMu.RUnlock()
-		}
-		return len(haveLeader) == len(c.servers)
-	}
-
-	// Create streams until we have a leader on all the servers.
-	var index int
-	checkFor(t, 10*time.Second, time.Millisecond, func() error {
-		if allServersHaveLeaders() {
-			return nil
-		}
-		index++
-		_, err := js.AddStream(&nats.StreamConfig{
-			Name:     fmt.Sprintf("TEST_%d", index),
-			Subjects: []string{fmt.Sprintf("foo.%d", index)},
-			Replicas: 3,
-		})
-		require_NoError(t, err)
-		return fmt.Errorf("All servers do not have at least one leader")
-	})
-
-	// Put our server into lameduck mode.
-	// Need a client.
-	dummy, _ := jsClientConnect(t, s)
-	defer dummy.Close()
-	go s.lameDuckMode()
-
-	// Wait for all leaders to move off.
-	checkFor(t, 2*time.Second, 50*time.Millisecond, func() error {
-		s.rnMu.RLock()
-		defer s.rnMu.RUnlock()
-		for _, n := range s.raftNodes {
-			if n.Leader() {
-				return fmt.Errorf("Server still has a leader")
-			}
-		}
-		return nil
-	})
-
-	// All leader evacuated.
-
-	// Create a go routine that will create streams constantly.
-	qch := make(chan bool)
-	go func() {
-		var index int
-		for {
-			select {
-			case <-time.After(time.Millisecond):
-				index++
-				_, err := js.AddStream(&nats.StreamConfig{
-					Name:     fmt.Sprintf("NEW_TEST_%d", index),
-					Subjects: []string{fmt.Sprintf("bar.%d", index)},
-					Replicas: 3,
-				})
-				if err != nil {
-					return
-				}
-			case <-qch:
-				return
-			}
-		}
-	}()
-	defer close(qch)
-
-	// Make sure we do not have any leaders placed on the lameduck server.
-	for s.isRunning() {
-		var hasLeader bool
-		s.rnMu.RLock()
-		for _, n := range s.raftNodes {
-			hasLeader = hasLeader || n.Leader()
-		}
-		s.rnMu.RUnlock()
-		if hasLeader {
-			t.Fatalf("Server had a leader when it should not due to lameduck mode")
-		}
-	}
-}
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index 2dd0612e..9b9a8230 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020-2023 The NATS Authors
+// Copyright 2020-2022 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -37,21 +37,12 @@ import (
 func init() {
 	// Speed up raft for tests.
 	hbInterval = 50 * time.Millisecond
-	minElectionTimeout = 750 * time.Millisecond
-	maxElectionTimeout = 2500 * time.Millisecond
-	lostQuorumInterval = 500 * time.Millisecond
+	minElectionTimeout = 1 * time.Second
+	maxElectionTimeout = 3 * time.Second
+	lostQuorumInterval = time.Second
 	lostQuorumCheck = 4 * hbInterval
 }
 
-// Used to setup clusters of clusters for tests.
-type cluster struct {
-	servers  []*Server
-	opts     []*Options
-	name     string
-	t        testing.TB
-	nproxies []*netProxy
-}
-
 // Used to setup superclusters for tests.
 type supercluster struct {
 	t        *testing.T
@@ -360,9 +351,6 @@ type gwProxy struct {
 	down int
 }
 
-// For use in normal clusters.
-type clusterProxy = gwProxy
-
 // Maps cluster names to proxy settings.
 type gwProxyMap map[string]*gwProxy
 
@@ -509,7 +497,7 @@ func (sc *supercluster) leader() *Server {
 
 func (sc *supercluster) waitOnLeader() {
 	sc.t.Helper()
-	expires := time.Now().Add(30 * time.Second)
+	expires := time.Now().Add(10 * time.Second)
 	for time.Now().Before(expires) {
 		for _, c := range sc.clusters {
 			if leader := c.leader(); leader != nil {
@@ -548,7 +536,7 @@ func (sc *supercluster) waitOnPeerCount(n int) {
 	sc.t.Helper()
 	sc.waitOnLeader()
 	leader := sc.leader()
-	expires := time.Now().Add(30 * time.Second)
+	expires := time.Now().Add(20 * time.Second)
 	for time.Now().Before(expires) {
 		peers := leader.JetStreamClusterPeers()
 		if len(peers) == n {
@@ -704,19 +692,9 @@ func createJetStreamCluster(t testing.TB, tmpl string, clusterName, snPre string
 
 type modifyCb func(serverName, clusterName, storeDir, conf string) string
 
-func createJetStreamClusterAndModHook(t testing.TB, tmpl, cName, snPre string, numServers int, portStart int, waitOnReady bool, modify modifyCb) *cluster {
-	return createJetStreamClusterEx(t, tmpl, cName, snPre, numServers, portStart, waitOnReady, modify, nil)
-}
-
-func createJetStreamClusterWithNetProxy(t testing.TB, cName string, numServers int, cnp *clusterProxy) *cluster {
-	startPorts := []int{7_122, 9_122, 11_122, 15_122}
-	port := startPorts[rand.Intn(len(startPorts))]
-	return createJetStreamClusterEx(t, jsClusterTempl, cName, _EMPTY_, numServers, port, true, nil, cnp)
-}
-
-func createJetStreamClusterEx(t testing.TB, tmpl, cName, snPre string, numServers int, portStart int, wait bool, modify modifyCb, cnp *clusterProxy) *cluster {
+func createJetStreamClusterAndModHook(t testing.TB, tmpl string, clusterName, snPre string, numServers int, portStart int, waitOnReady bool, modify modifyCb) *cluster {
 	t.Helper()
-	if cName == _EMPTY_ || numServers < 1 {
+	if clusterName == _EMPTY_ || numServers < 1 {
 		t.Fatalf("Bad params")
 	}
 
@@ -737,32 +715,20 @@ func createJetStreamClusterEx(t testing.TB, tmpl, cName, snPre string, numServer
 
 	// Build out the routes that will be shared with all configs.
 	var routes []string
-	var nproxies []*netProxy
 	for cp := portStart; cp < portStart+numServers; cp++ {
-		routeURL := fmt.Sprintf("nats-route://127.0.0.1:%d", cp)
-		if cnp != nil {
-			np := createNetProxy(cnp.rtt, cnp.up, cnp.down, routeURL, false)
-			nproxies = append(nproxies, np)
-			routeURL = np.routeURL()
-		}
-		routes = append(routes, routeURL)
+		routes = append(routes, fmt.Sprintf("nats-route://127.0.0.1:%d", cp))
 	}
 	routeConfig := strings.Join(routes, ",")
 
 	// Go ahead and build configurations and start servers.
-	c := &cluster{servers: make([]*Server, 0, numServers), opts: make([]*Options, 0, numServers), name: cName, nproxies: nproxies}
-
-	// Start any proxies.
-	for _, np := range nproxies {
-		np.start()
-	}
+	c := &cluster{servers: make([]*Server, 0, numServers), opts: make([]*Options, 0, numServers), name: clusterName}
 
 	for cp := portStart; cp < portStart+numServers; cp++ {
 		storeDir := t.TempDir()
 		sn := fmt.Sprintf("%sS-%d", snPre, cp-portStart+1)
-		conf := fmt.Sprintf(tmpl, sn, storeDir, cName, cp, routeConfig)
+		conf := fmt.Sprintf(tmpl, sn, storeDir, clusterName, cp, routeConfig)
 		if modify != nil {
-			conf = modify(sn, cName, storeDir, conf)
+			conf = modify(sn, clusterName, storeDir, conf)
 		}
 		s, o := RunServerWithConfig(createConfFile(t, []byte(conf)))
 		c.servers = append(c.servers, s)
@@ -772,7 +738,7 @@ func createJetStreamClusterEx(t testing.TB, tmpl, cName, snPre string, numServer
 
 	// Wait til we are formed and have a leader.
 	c.checkClusterFormed()
-	if wait {
+	if waitOnReady {
 		c.waitOnClusterReady()
 	}
 
@@ -1237,7 +1203,7 @@ func (c *cluster) waitOnPeerCount(n int) {
 
 func (c *cluster) waitOnConsumerLeader(account, stream, consumer string) {
 	c.t.Helper()
-	expires := time.Now().Add(30 * time.Second)
+	expires := time.Now().Add(20 * time.Second)
 	for time.Now().Before(expires) {
 		if leader := c.consumerLeader(account, stream, consumer); leader != nil {
 			time.Sleep(200 * time.Millisecond)
@@ -1329,7 +1295,7 @@ func (c *cluster) waitOnServerHealthz(s *Server) {
 
 func (c *cluster) waitOnServerCurrent(s *Server) {
 	c.t.Helper()
-	expires := time.Now().Add(30 * time.Second)
+	expires := time.Now().Add(20 * time.Second)
 	for time.Now().Before(expires) {
 		time.Sleep(100 * time.Millisecond)
 		if !s.JetStreamEnabled() || s.JetStreamIsCurrent() {
@@ -1655,14 +1621,18 @@ func (np *netProxy) loop(rtt time.Duration, tbw int, r, w net.Conn) {
 
 	rl := rate.NewLimiter(rate.Limit(tbw), rbl)
 
-	for {
+	for fr := true; ; {
+		sr := time.Now()
 		n, err := r.Read(buf[:])
 		if err != nil {
 			return
 		}
 		// RTT delays
-		if delay > 0 {
-			time.Sleep(delay)
+		if fr || time.Since(sr) > 250*time.Millisecond {
+			fr = false
+			if delay > 0 {
+				time.Sleep(delay)
+			}
 		}
 		if err := rl.WaitN(ctx, n); err != nil {
 			return
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 9b381c11..0ca21d89 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -18842,7 +18842,6 @@ func TestJetStreamAccountPurge(t *testing.T) {
 	require_NoError(t, os.Remove(storeDir+"/jwt/"+accpub+".jwt"))
 
 	s, o = RunServerWithConfig(o.ConfigFile)
-	defer s.Shutdown()
 	inspectDirs(t, 1)
 	purge(t)
 	inspectDirs(t, 0)
diff --git a/server/norace_test.go b/server/norace_test.go
index 17b20f36..787a6e5f 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -26,7 +26,6 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"math"
 	"math/rand"
 	"net"
 	"net/http"
@@ -50,7 +49,6 @@ import (
 	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nkeys"
-	"github.com/nats-io/nuid"
 )
 
 // IMPORTANT: Tests in this file are not executed when running with the -race flag.
@@ -5267,7 +5265,7 @@ func TestNoRaceJetStreamClusterDirectAccessAllPeersSubs(t *testing.T) {
 		t.Fatalf("Expected to see messages increase, got %d", si.State.Msgs)
 	}
 
-	checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		// Make sure they are all the same from a state perspective.
 		// Leader will have the expected state.
 		lmset, err := c.streamLeader("$G", "TEST").GlobalAccount().lookupStream("TEST")
@@ -6122,15 +6120,12 @@ func TestNoRaceJetStreamClusterEnsureWALCompact(t *testing.T) {
 	err = node.InstallSnapshot(snap)
 	require_NoError(t, err)
 
-	received, done := 0, make(chan bool, 1)
+	received, done := 0, make(chan bool)
 
 	nc.Subscribe("zz", func(m *nats.Msg) {
 		received++
 		if received >= ns {
-			select {
-			case done <- true:
-			default:
-			}
+			done <- true
 		}
 		m.Ack()
 	})
@@ -6532,1158 +6527,3 @@ func TestNoRaceJetStreamClusterGhostConsumers(t *testing.T) {
 		return fmt.Errorf("Still have missing: %+v", missing)
 	})
 }
-
-// This is to test a publish slowdown and general instability experienced in a setup simular to this.
-// We have feeder streams that are all sourced to an aggregate stream. All streams are interest retention.
-// We want to monitor the avg publish time for the sync publishers to the feeder streams, the ingest rate to
-// the aggregate stream, and general health of the consumers on the aggregate stream.
-// Target publish rate is ~2k/s with publish time being ~40-60ms but remaining stable.
-// We can also simulate max redeliveries that create interior deletes in streams.
-func TestNoRaceJetStreamClusterF3Setup(t *testing.T) {
-	// Uncomment to run. Needs to be on a pretty big machine. Do not want as part of Travis tests atm.
-	skip(t)
-
-	// These and the settings below achieve ~60ms pub time on avg and ~2k msgs per sec inbound to the aggregate stream.
-	// On my machine though.
-	np := clusterProxy{
-		rtt:  2 * time.Millisecond,
-		up:   1 * 1024 * 1024 * 1024, // 1gbit
-		down: 1 * 1024 * 1024 * 1024, // 1gbit
-	}
-
-	// Test params.
-	numSourceStreams := 20
-	numConsumersPerSource := 1
-	numPullersPerConsumer := 50
-	numPublishers := 100
-	setHighStartSequence := false
-	simulateMaxRedeliveries := false
-	maxBadPubTimes := uint32(20)
-	badPubThresh := 5 * time.Second
-	testTime := 5 * time.Minute // make sure to do --timeout=65m
-
-	t.Logf("Starting Test: Total Test Time %v", testTime)
-
-	c := createJetStreamClusterWithNetProxy(t, "R3S", 3, &np)
-	defer c.shutdown()
-
-	// Do some quick sanity checking for latency stuff.
-	{
-		nc, js := jsClientConnect(t, c.randomServer())
-		defer nc.Close()
-
-		_, err := js.AddStream(&nats.StreamConfig{
-			Name:      "TEST",
-			Replicas:  3,
-			Subjects:  []string{"foo"},
-			Retention: nats.InterestPolicy,
-		})
-		require_NoError(t, err)
-		defer js.DeleteStream("TEST")
-
-		sl := c.streamLeader(globalAccountName, "TEST")
-		nc, js = jsClientConnect(t, sl)
-		defer nc.Close()
-		start := time.Now()
-		_, err = js.Publish("foo", []byte("hello"))
-		require_NoError(t, err)
-		// This is best case, and with client connection being close to free, this should be at least > rtt
-		if elapsed := time.Since(start); elapsed < np.rtt {
-			t.Fatalf("Expected publish time to be > %v, got %v", np.rtt, elapsed)
-		}
-
-		nl := c.randomNonStreamLeader(globalAccountName, "TEST")
-		nc, js = jsClientConnect(t, nl)
-		defer nc.Close()
-		start = time.Now()
-		_, err = js.Publish("foo", []byte("hello"))
-		require_NoError(t, err)
-		// This is worst case, meaning message has to travel to leader, then to fastest replica, then back.
-		// So should be at 3x rtt, so check at least > 2x rtt.
-		if elapsed := time.Since(start); elapsed < 2*np.rtt {
-			t.Fatalf("Expected publish time to be > %v, got %v", 2*np.rtt, elapsed)
-		}
-	}
-
-	// Setup source streams.
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	t.Logf("Creating %d Source Streams", numSourceStreams)
-
-	var sources []string
-	wg := sync.WaitGroup{}
-	for i := 0; i < numSourceStreams; i++ {
-		sname := fmt.Sprintf("EVENT-%s", nuid.Next())
-		wg.Add(1)
-		go func(stream string) {
-			defer wg.Done()
-			t.Logf("  %q", stream)
-			sources = append(sources, stream)
-			subj := fmt.Sprintf("%s.>", stream)
-			_, err := js.AddStream(&nats.StreamConfig{
-				Name:      stream,
-				Subjects:  []string{subj},
-				Replicas:  3,
-				Retention: nats.InterestPolicy,
-			})
-			require_NoError(t, err)
-			for j := 0; j < numConsumersPerSource; j++ {
-				consumer := fmt.Sprintf("C%d", j)
-				_, err := js.Subscribe(_EMPTY_, func(msg *nats.Msg) {
-					msg.Ack()
-				}, nats.BindStream(stream), nats.Durable(consumer), nats.ManualAck())
-				require_NoError(t, err)
-			}
-		}(sname)
-	}
-	wg.Wait()
-
-	var streamSources []*nats.StreamSource
-	for _, src := range sources {
-		streamSources = append(streamSources, &nats.StreamSource{Name: src})
-
-	}
-
-	t.Log("Creating Aggregate Stream")
-
-	// Now create the aggregate stream.
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:      "EVENTS",
-		Replicas:  3,
-		Retention: nats.InterestPolicy,
-		Sources:   streamSources,
-	})
-	require_NoError(t, err)
-
-	// Set first sequence to a high number.
-	if setHighStartSequence {
-		require_NoError(t, js.PurgeStream("EVENTS", &nats.StreamPurgeRequest{Sequence: 32_000_001}))
-	}
-
-	// Now create 2 pull consumers.
-	_, err = js.PullSubscribe(_EMPTY_, "C1",
-		nats.BindStream("EVENTS"),
-		nats.MaxDeliver(1),
-		nats.AckWait(10*time.Second),
-		nats.ManualAck(),
-	)
-	require_NoError(t, err)
-
-	_, err = js.PullSubscribe(_EMPTY_, "C2",
-		nats.BindStream("EVENTS"),
-		nats.MaxDeliver(1),
-		nats.AckWait(10*time.Second),
-		nats.ManualAck(),
-	)
-	require_NoError(t, err)
-
-	t.Logf("Creating %d x 2 Pull Subscribers", numPullersPerConsumer)
-
-	// Now create the pullers.
-	for _, subName := range []string{"C1", "C2"} {
-		for i := 0; i < numPullersPerConsumer; i++ {
-			go func(subName string) {
-				nc, js := jsClientConnect(t, c.randomServer())
-				defer nc.Close()
-
-				sub, err := js.PullSubscribe(_EMPTY_, subName,
-					nats.BindStream("EVENTS"),
-					nats.MaxDeliver(1),
-					nats.AckWait(10*time.Second),
-					nats.ManualAck(),
-				)
-				require_NoError(t, err)
-
-				for {
-					msgs, err := sub.Fetch(25, nats.MaxWait(2*time.Second))
-					if err != nil && err != nats.ErrTimeout {
-						t.Logf("Exiting pull subscriber %q: %v", subName, err)
-						return
-					}
-					// Shuffle
-					rand.Shuffle(len(msgs), func(i, j int) { msgs[i], msgs[j] = msgs[j], msgs[i] })
-
-					// Wait for a random interval up to 100ms.
-					time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
-
-					for _, m := range msgs {
-						// If we want to simulate max redeliveries being hit, since not acking
-						// once will cause it due to subscriber setup.
-						// 100_000 == 0.01%
-						if simulateMaxRedeliveries && rand.Intn(100_000) == 0 {
-							md, err := m.Metadata()
-							require_NoError(t, err)
-							t.Logf("** Skipping Ack: %d **", md.Sequence.Stream)
-						} else {
-							m.Ack()
-						}
-					}
-				}
-			}(subName)
-		}
-	}
-
-	// Now create feeder publishers.
-	eventTypes := []string{"PAYMENT", "SUBMISSION", "CANCEL"}
-
-	msg := make([]byte, 2*1024) // 2k payload
-	rand.Read(msg)
-
-	// For tracking pub times.
-	var pubs int
-	var totalPubTime time.Duration
-	var pmu sync.Mutex
-	last := time.Now()
-
-	updatePubStats := func(elapsed time.Duration) {
-		pmu.Lock()
-		defer pmu.Unlock()
-		// Reset every 5s
-		if time.Since(last) > 5*time.Second {
-			pubs = 0
-			totalPubTime = 0
-			last = time.Now()
-		}
-		pubs++
-		totalPubTime += elapsed
-	}
-	avgPubTime := func() time.Duration {
-		pmu.Lock()
-		np := pubs
-		tpt := totalPubTime
-		pmu.Unlock()
-		return tpt / time.Duration(np)
-	}
-
-	t.Logf("Creating %d Publishers", numPublishers)
-
-	var numLimitsExceeded atomic.Uint32
-	errCh := make(chan error, 100)
-
-	for i := 0; i < numPublishers; i++ {
-		go func() {
-			nc, js := jsClientConnect(t, c.randomServer())
-			defer nc.Close()
-
-			for {
-				// Grab a random source stream
-				stream := sources[rand.Intn(len(sources))]
-				// Grab random event type.
-				evt := eventTypes[rand.Intn(len(eventTypes))]
-				subj := fmt.Sprintf("%s.%s", stream, evt)
-				start := time.Now()
-				_, err := js.Publish(subj, msg)
-				if err != nil {
-					t.Logf("Exiting publisher: %v", err)
-					return
-				}
-				elapsed := time.Since(start)
-				if elapsed > badPubThresh {
-					t.Logf("Publish time took more than expected: %v", elapsed)
-					numLimitsExceeded.Add(1)
-					if ne := numLimitsExceeded.Load(); ne > maxBadPubTimes {
-						errCh <- fmt.Errorf("Too many exceeded times on publish: %d", ne)
-						return
-					}
-				}
-				updatePubStats(elapsed)
-			}
-		}()
-	}
-
-	t.Log("Creating Monitoring Routine - Data in ~10s")
-
-	// Create monitoring routine.
-	go func() {
-		nc, js := jsClientConnect(t, c.randomServer())
-		defer nc.Close()
-
-		fseq, lseq := uint64(0), uint64(0)
-		for {
-			// Grab consumers
-			var minAckFloor uint64 = math.MaxUint64
-			for _, consumer := range []string{"C1", "C2"} {
-				ci, err := js.ConsumerInfo("EVENTS", consumer)
-				if err != nil {
-					t.Logf("Exiting Monitor: %v", err)
-					return
-				}
-				if lseq > 0 {
-					t.Logf("%s:\n  Delivered:\t%d\n  AckFloor:\t%d\n  AckPending:\t%d\n  NumPending:\t%d",
-						consumer, ci.Delivered.Stream, ci.AckFloor.Stream, ci.NumAckPending, ci.NumPending)
-				}
-				if ci.AckFloor.Stream < minAckFloor {
-					minAckFloor = ci.AckFloor.Stream
-				}
-			}
-			// Now grab aggregate stream state.
-			si, err := js.StreamInfo("EVENTS")
-			if err != nil {
-				t.Logf("Exiting Monitor: %v", err)
-				return
-			}
-			state := si.State
-			if lseq != 0 {
-				t.Logf("Stream:\n  Msgs: \t%d\n  First:\t%d\n  Last: \t%d\n  Deletes:\t%d\n",
-					state.Msgs, state.FirstSeq, state.LastSeq, state.NumDeleted)
-				t.Logf("Publish Stats:\n  Msgs/s:\t%0.2f\n  Avg Pub:\t%v\n\n", float64(si.State.LastSeq-lseq)/5.0, avgPubTime())
-				if si.State.FirstSeq < minAckFloor && si.State.FirstSeq == fseq {
-					t.Log("Stream first seq < minimum ack floor")
-				}
-			}
-			fseq, lseq = si.State.FirstSeq, si.State.LastSeq
-			time.Sleep(5 * time.Second)
-		}
-
-	}()
-
-	select {
-	case e := <-errCh:
-		t.Fatal(e)
-	case <-time.After(testTime):
-		t.Fatalf("Did not receive completion signal")
-	}
-}
-
-// Unbalanced stretch cluster.
-// S2 (stream leader) will have a slow path to S1 (via proxy) and S3 (consumer leader) will have a fast path.
-//
-//	 Route Ports
-//		"S1": 14622
-//		"S2": 15622
-//		"S3": 16622
-func createStretchUnbalancedCluster(t testing.TB) (c *cluster, np *netProxy) {
-	t.Helper()
-
-	tmpl := `
-	listen: 127.0.0.1:-1
-	server_name: %s
-	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
-
-	cluster {
-		name: "F3"
-		listen: 127.0.0.1:%d
-		routes = [%s]
-	}
-
-	accounts {
-		$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
-	}
-	`
-	// Do these in order, S1, S2 (proxy) then S3.
-	c = &cluster{t: t, servers: make([]*Server, 3), opts: make([]*Options, 3), name: "F3"}
-
-	// S1
-	conf := fmt.Sprintf(tmpl, "S1", t.TempDir(), 14622, "route://127.0.0.1:15622, route://127.0.0.1:16622")
-	c.servers[0], c.opts[0] = RunServerWithConfig(createConfFile(t, []byte(conf)))
-
-	// S2
-	// Create the proxy first. Connect this to S1. Make it slow, e.g. 5ms RTT.
-	np = createNetProxy(1*time.Millisecond, 1024*1024*1024, 1024*1024*1024, "route://127.0.0.1:14622", true)
-	routes := fmt.Sprintf("%s, route://127.0.0.1:16622", np.routeURL())
-	conf = fmt.Sprintf(tmpl, "S2", t.TempDir(), 15622, routes)
-	c.servers[1], c.opts[1] = RunServerWithConfig(createConfFile(t, []byte(conf)))
-
-	// S3
-	conf = fmt.Sprintf(tmpl, "S3", t.TempDir(), 16622, "route://127.0.0.1:14622, route://127.0.0.1:15622")
-	c.servers[2], c.opts[2] = RunServerWithConfig(createConfFile(t, []byte(conf)))
-
-	c.checkClusterFormed()
-	c.waitOnClusterReady()
-
-	return c, np
-}
-
-// We test an interest based stream that has a cluster with a node with asymmetric paths from
-// the stream leader and the consumer leader such that the consumer leader path is fast and
-// replicated acks arrive sooner then the actual message. This path was considered, but also
-// categorized as very rare and was expensive as it tried to forward a new stream msg delete
-// proposal to the original stream leader. It now will deal with the issue locally and not
-// slow down the ingest rate to the stream's publishers.
-func TestNoRaceJetStreamClusterDifferentRTTInterestBasedStreamSetup(t *testing.T) {
-	// Uncomment to run. Do not want as part of Travis tests atm.
-	skip(t)
-
-	c, np := createStretchUnbalancedCluster(t)
-	defer c.shutdown()
-	defer np.stop()
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	// Now create the stream.
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:      "EVENTS",
-		Subjects:  []string{"EV.>"},
-		Replicas:  3,
-		Retention: nats.InterestPolicy,
-	})
-	require_NoError(t, err)
-
-	// Make sure it's leader is on S2.
-	sl := c.servers[1]
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnStreamLeader(globalAccountName, "EVENTS")
-		if s := c.streamLeader(globalAccountName, "EVENTS"); s != sl {
-			s.JetStreamStepdownStream(globalAccountName, "EVENTS")
-			return fmt.Errorf("Server %s is not stream leader yet", sl)
-		}
-		return nil
-	})
-
-	// Now create the consumer.
-	_, err = js.PullSubscribe(_EMPTY_, "C", nats.BindStream("EVENTS"), nats.ManualAck())
-	require_NoError(t, err)
-
-	// Make sure the consumer leader is on S3.
-	cl := c.servers[2]
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnConsumerLeader(globalAccountName, "EVENTS", "C")
-		if s := c.consumerLeader(globalAccountName, "EVENTS", "C"); s != cl {
-			s.JetStreamStepdownConsumer(globalAccountName, "EVENTS", "C")
-			return fmt.Errorf("Server %s is not consumer leader yet", cl)
-		}
-		return nil
-	})
-
-	go func(js nats.JetStream) {
-		sub, err := js.PullSubscribe(_EMPTY_, "C", nats.BindStream("EVENTS"), nats.ManualAck())
-		require_NoError(t, err)
-
-		for {
-			msgs, err := sub.Fetch(100, nats.MaxWait(2*time.Second))
-			if err != nil && err != nats.ErrTimeout {
-				return
-			}
-			// Shuffle
-			rand.Shuffle(len(msgs), func(i, j int) { msgs[i], msgs[j] = msgs[j], msgs[i] })
-			for _, m := range msgs {
-				m.Ack()
-			}
-		}
-	}(js)
-
-	numPublishers := 25
-	pubThresh := 2 * time.Second
-	var maxExceeded atomic.Int64
-	errCh := make(chan error, numPublishers)
-	wg := sync.WaitGroup{}
-
-	msg := make([]byte, 2*1024) // 2k payload
-	rand.Read(msg)
-
-	// Publishers.
-	for i := 0; i < numPublishers; i++ {
-		wg.Add(1)
-		go func(iter int) {
-			defer wg.Done()
-
-			// Connect to random, the slow ones will be connected to the slow node.
-			// But if you connect them all there it will pass.
-			s := c.randomServer()
-			nc, js := jsClientConnect(t, s)
-			defer nc.Close()
-
-			for i := 0; i < 1_000; i++ {
-				start := time.Now()
-				_, err := js.Publish("EV.PAID", msg)
-				if err != nil {
-					errCh <- fmt.Errorf("Publish error: %v", err)
-					return
-				}
-				if elapsed := time.Since(start); elapsed > pubThresh {
-					errCh <- fmt.Errorf("Publish time exceeded")
-					if int64(elapsed) > maxExceeded.Load() {
-						maxExceeded.Store(int64(elapsed))
-					}
-					return
-				}
-			}
-		}(i)
-	}
-
-	wg.Wait()
-
-	select {
-	case e := <-errCh:
-		t.Fatalf("%v: threshold is %v, maximum seen: %v", e, pubThresh, time.Duration(maxExceeded.Load()))
-	default:
-	}
-}
-
-func TestNoRaceJetStreamInterestStreamCheckInterestRaceBug(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,
-		Retention: nats.InterestPolicy,
-	})
-	require_NoError(t, err)
-
-	numConsumers := 10
-	for i := 0; i < numConsumers; i++ {
-		nc, js := jsClientConnect(t, c.randomServer())
-		defer nc.Close()
-
-		_, err = js.Subscribe("foo", func(m *nats.Msg) {
-			m.Ack()
-		}, nats.Durable(fmt.Sprintf("C%d", i)), nats.ManualAck())
-		require_NoError(t, err)
-	}
-
-	numToSend := 10_000
-	for i := 0; i < numToSend; i++ {
-		_, err := js.PublishAsync("foo", nil)
-		require_NoError(t, err)
-	}
-	select {
-	case <-js.PublishAsyncComplete():
-	case <-time.After(20 * time.Second):
-		t.Fatalf("Did not receive completion signal")
-	}
-
-	// Wait til ackfloor is correct for all consumers.
-	checkFor(t, 20*time.Second, 100*time.Millisecond, func() error {
-		for _, s := range c.servers {
-			mset, err := s.GlobalAccount().lookupStream("TEST")
-			require_NoError(t, err)
-
-			mset.mu.RLock()
-			defer mset.mu.RUnlock()
-
-			require_True(t, len(mset.consumers) == numConsumers)
-
-			for _, o := range mset.consumers {
-				state, err := o.store.State()
-				require_NoError(t, err)
-				if state.AckFloor.Stream != uint64(numToSend) {
-					return fmt.Errorf("Ackfloor not correct yet")
-				}
-			}
-		}
-		return nil
-	})
-
-	for _, s := range c.servers {
-		mset, err := s.GlobalAccount().lookupStream("TEST")
-		require_NoError(t, err)
-
-		mset.mu.RLock()
-		defer mset.mu.RUnlock()
-
-		state := mset.state()
-		require_True(t, state.Msgs == 0)
-		require_True(t, state.FirstSeq == uint64(numToSend+1))
-	}
-}
-
-func TestNoRaceJetStreamClusterInterestStreamConsistencyAfterRollingRestart(t *testing.T) {
-	// Uncomment to run. Needs to be on a big machine. Do not want as part of Travis tests atm.
-	skip(t)
-
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
-	defer c.shutdown()
-
-	numStreams := 200
-	numConsumersPer := 5
-	numPublishers := 10
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	qch := make(chan bool)
-
-	var mm sync.Mutex
-	ackMap := make(map[string]map[uint64][]string)
-
-	addAckTracking := func(seq uint64, stream, consumer string) {
-		mm.Lock()
-		defer mm.Unlock()
-		sam := ackMap[stream]
-		if sam == nil {
-			sam = make(map[uint64][]string)
-			ackMap[stream] = sam
-		}
-		sam[seq] = append(sam[seq], consumer)
-	}
-
-	doPullSubscriber := func(stream, consumer, filter string) {
-		nc, js := jsClientConnect(t, c.randomServer())
-		defer nc.Close()
-
-		var err error
-		var sub *nats.Subscription
-		timeout := time.Now().Add(5 * time.Second)
-		for time.Now().Before(timeout) {
-			sub, err = js.PullSubscribe(filter, consumer, nats.BindStream(stream), nats.ManualAck())
-			if err == nil {
-				break
-			}
-		}
-		if err != nil {
-			t.Logf("Error on pull subscriber: %v", err)
-			return
-		}
-
-		for {
-			select {
-			case <-time.After(500 * time.Millisecond):
-				msgs, err := sub.Fetch(100, nats.MaxWait(time.Second))
-				if err != nil {
-					continue
-				}
-				// Shuffle
-				rand.Shuffle(len(msgs), func(i, j int) { msgs[i], msgs[j] = msgs[j], msgs[i] })
-				for _, m := range msgs {
-					meta, err := m.Metadata()
-					require_NoError(t, err)
-					m.Ack()
-					addAckTracking(meta.Sequence.Stream, stream, consumer)
-					if meta.NumDelivered > 1 {
-						t.Logf("Got a msg redelivered %d for sequence %d on %q %q\n", meta.NumDelivered, meta.Sequence.Stream, stream, consumer)
-					}
-				}
-			case <-qch:
-				nc.Flush()
-				return
-			}
-		}
-	}
-
-	// Setup
-	wg := sync.WaitGroup{}
-	for i := 0; i < numStreams; i++ {
-		wg.Add(1)
-		go func(stream string) {
-			defer wg.Done()
-			subj := fmt.Sprintf("%s.>", stream)
-			_, err := js.AddStream(&nats.StreamConfig{
-				Name:      stream,
-				Subjects:  []string{subj},
-				Replicas:  3,
-				Retention: nats.InterestPolicy,
-			})
-			require_NoError(t, err)
-			for i := 0; i < numConsumersPer; i++ {
-				consumer := fmt.Sprintf("C%d", i)
-				filter := fmt.Sprintf("%s.%d", stream, i)
-				_, err = js.AddConsumer(stream, &nats.ConsumerConfig{
-					Durable:       consumer,
-					FilterSubject: filter,
-					AckPolicy:     nats.AckExplicitPolicy,
-					AckWait:       2 * time.Second,
-				})
-				require_NoError(t, err)
-				c.waitOnConsumerLeader(globalAccountName, stream, consumer)
-				go doPullSubscriber(stream, consumer, filter)
-			}
-		}(fmt.Sprintf("A-%d", i))
-	}
-	wg.Wait()
-
-	msg := make([]byte, 2*1024) // 2k payload
-	rand.Read(msg)
-
-	// Controls if publishing is on or off.
-	var pubActive atomic.Bool
-
-	doPublish := func() {
-		nc, js := jsClientConnect(t, c.randomServer())
-		defer nc.Close()
-
-		for {
-			select {
-			case <-time.After(100 * time.Millisecond):
-				if pubActive.Load() {
-					for i := 0; i < numStreams; i++ {
-						for j := 0; j < numConsumersPer; j++ {
-							subj := fmt.Sprintf("A-%d.%d", i, j)
-							// Don't care about errors here for this test.
-							js.Publish(subj, msg)
-						}
-					}
-				}
-			case <-qch:
-				return
-			}
-		}
-	}
-
-	pubActive.Store(true)
-
-	for i := 0; i < numPublishers; i++ {
-		go doPublish()
-	}
-
-	// Let run for a bit.
-	time.Sleep(20 * time.Second)
-
-	// Do a rolling restart.
-	for _, s := range c.servers {
-		t.Logf("Shutdown %v\n", s)
-		s.Shutdown()
-		s.WaitForShutdown()
-		time.Sleep(20 * time.Second)
-		t.Logf("Restarting %v\n", s)
-		s = c.restartServer(s)
-		c.waitOnServerHealthz(s)
-	}
-
-	// Let run for a bit longer.
-	time.Sleep(10 * time.Second)
-
-	// Stop pubs.
-	pubActive.Store(false)
-
-	// Let settle.
-	time.Sleep(10 * time.Second)
-	close(qch)
-	time.Sleep(20 * time.Second)
-
-	nc, js = jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	minAckFloor := func(stream string) (uint64, string) {
-		var maf uint64 = math.MaxUint64
-		var consumer string
-		for i := 0; i < numConsumersPer; i++ {
-			cname := fmt.Sprintf("C%d", i)
-			ci, err := js.ConsumerInfo(stream, cname)
-			require_NoError(t, err)
-			if ci.AckFloor.Stream < maf {
-				maf = ci.AckFloor.Stream
-				consumer = cname
-			}
-		}
-		return maf, consumer
-	}
-
-	checkStreamAcks := func(stream string) {
-		mm.Lock()
-		defer mm.Unlock()
-		if sam := ackMap[stream]; sam != nil {
-			for seq := 1; ; seq++ {
-				acks := sam[uint64(seq)]
-				if acks == nil {
-					if sam[uint64(seq+1)] != nil {
-						t.Logf("Missing an ack on stream %q for sequence %d\n", stream, seq)
-					} else {
-						break
-					}
-				}
-				if len(acks) > 1 {
-					t.Logf("Multiple acks for %d which is not expected: %+v", seq, acks)
-				}
-			}
-		}
-	}
-
-	// Now check all streams such that their first sequence is equal to the minimum of all consumers.
-	for i := 0; i < numStreams; i++ {
-		stream := fmt.Sprintf("A-%d", i)
-		si, err := js.StreamInfo(stream)
-		require_NoError(t, err)
-
-		if maf, consumer := minAckFloor(stream); maf > si.State.FirstSeq {
-			t.Logf("\nBAD STATE DETECTED FOR %q, CHECKING OTHER SERVERS! ACK %d vs %+v LEADER %v, CL FOR %q %v\n",
-				stream, maf, si.State, c.streamLeader(globalAccountName, stream), consumer, c.consumerLeader(globalAccountName, stream, consumer))
-
-			t.Logf("TEST ACKS %+v\n", ackMap)
-
-			checkStreamAcks(stream)
-
-			for _, s := range c.servers {
-				mset, err := s.GlobalAccount().lookupStream(stream)
-				require_NoError(t, err)
-				state := mset.state()
-				t.Logf("Server %v Stream STATE %+v\n", s, state)
-
-				var smv StoreMsg
-				if sm, err := mset.store.LoadMsg(state.FirstSeq, &smv); err == nil {
-					t.Logf("Subject for msg %d is %q", state.FirstSeq, sm.subj)
-				} else {
-					t.Logf("Could not retrieve msg for %d: %v", state.FirstSeq, err)
-				}
-
-				if len(mset.preAcks) > 0 {
-					t.Logf("%v preAcks %+v\n", s, mset.preAcks)
-				}
-
-				for _, o := range mset.consumers {
-					ostate, err := o.store.State()
-					require_NoError(t, err)
-					t.Logf("Consumer STATE for %q is %+v\n", o.name, ostate)
-				}
-			}
-			t.Fatalf("BAD STATE: ACKFLOOR > FIRST %d vs %d\n", maf, si.State.FirstSeq)
-		}
-	}
-}
-
-func TestNoRaceFileStoreNumPending(t *testing.T) {
-	// No need for all permutations here.
-	storeDir := t.TempDir()
-	fcfg := FileStoreConfig{
-		StoreDir:  storeDir,
-		BlockSize: 2 * 1024, // Create many blocks on purpose.
-	}
-	fs, err := newFileStore(fcfg, StreamConfig{Name: "zzz", Subjects: []string{"*.*.*.*"}, Storage: FileStorage})
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	tokens := []string{"foo", "bar", "baz"}
-	genSubj := func() string {
-		return fmt.Sprintf("%s.%s.%s.%s",
-			tokens[rand.Intn(len(tokens))],
-			tokens[rand.Intn(len(tokens))],
-			tokens[rand.Intn(len(tokens))],
-			tokens[rand.Intn(len(tokens))],
-		)
-	}
-
-	for i := 0; i < 50_000; i++ {
-		subj := genSubj()
-		_, _, err := fs.StoreMsg(subj, nil, []byte("Hello World"))
-		require_NoError(t, err)
-	}
-
-	state := fs.State()
-
-	// Scan one by one for sanity check against other calculations.
-	sanityCheck := func(sseq uint64, filter string) SimpleState {
-		t.Helper()
-		var ss SimpleState
-		var smv StoreMsg
-		// For here we know 0 is invalid, set to 1.
-		if sseq == 0 {
-			sseq = 1
-		}
-		for seq := sseq; seq <= state.LastSeq; seq++ {
-			sm, err := fs.LoadMsg(seq, &smv)
-			if err != nil {
-				t.Logf("Encountered error %v loading sequence: %d", err, seq)
-				continue
-			}
-			if subjectIsSubsetMatch(sm.subj, filter) {
-				ss.Msgs++
-				ss.Last = seq
-				if ss.First == 0 || seq < ss.First {
-					ss.First = seq
-				}
-			}
-		}
-		return ss
-	}
-
-	check := func(sseq uint64, filter string) {
-		t.Helper()
-		np, lvs := fs.NumPending(sseq, filter, false)
-		ss := fs.FilteredState(sseq, filter)
-		sss := sanityCheck(sseq, filter)
-		if lvs != state.LastSeq {
-			t.Fatalf("Expected NumPending to return valid through last of %d but got %d", state.LastSeq, lvs)
-		}
-		if ss.Msgs != np {
-			t.Fatalf("NumPending of %d did not match ss.Msgs of %d", np, ss.Msgs)
-		}
-		if ss != sss {
-			t.Fatalf("Failed sanity check, expected %+v got %+v", sss, ss)
-		}
-	}
-
-	sanityCheckLastOnly := func(sseq uint64, filter string) SimpleState {
-		t.Helper()
-		var ss SimpleState
-		var smv StoreMsg
-		// For here we know 0 is invalid, set to 1.
-		if sseq == 0 {
-			sseq = 1
-		}
-		seen := make(map[string]bool)
-		for seq := state.LastSeq; seq >= sseq; seq-- {
-			sm, err := fs.LoadMsg(seq, &smv)
-			if err != nil {
-				t.Logf("Encountered error %v loading sequence: %d", err, seq)
-				continue
-			}
-			if !seen[sm.subj] && subjectIsSubsetMatch(sm.subj, filter) {
-				ss.Msgs++
-				if ss.Last == 0 {
-					ss.Last = seq
-				}
-				if ss.First == 0 || seq < ss.First {
-					ss.First = seq
-				}
-				seen[sm.subj] = true
-			}
-		}
-		return ss
-	}
-
-	checkLastOnly := func(sseq uint64, filter string) {
-		t.Helper()
-		np, lvs := fs.NumPending(sseq, filter, true)
-		ss := sanityCheckLastOnly(sseq, filter)
-		if lvs != state.LastSeq {
-			t.Fatalf("Expected NumPending to return valid through last of %d but got %d", state.LastSeq, lvs)
-		}
-		if ss.Msgs != np {
-			t.Fatalf("NumPending of %d did not match ss.Msgs of %d", np, ss.Msgs)
-		}
-	}
-
-	startSeqs := []uint64{0, 1, 2, 200, 444, 555, 2222, 8888, 12_345, 28_222, 33_456, 44_400, 49_999}
-	checkSubs := []string{"foo.>", "*.bar.>", "foo.bar.*.baz", "*.bar.>", "*.foo.bar.*", "foo.foo.bar.baz"}
-
-	for _, filter := range checkSubs {
-		for _, start := range startSeqs {
-			check(start, filter)
-			checkLastOnly(start, filter)
-		}
-	}
-}
-
-func TestNoRaceJetStreamClusterUnbalancedInterestMultipleConsumers(t *testing.T) {
-	c, np := createStretchUnbalancedCluster(t)
-	defer c.shutdown()
-	defer np.stop()
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	// Now create the stream.
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:      "EVENTS",
-		Subjects:  []string{"EV.>"},
-		Replicas:  3,
-		Retention: nats.InterestPolicy,
-	})
-	require_NoError(t, err)
-
-	// Make sure it's leader is on S2.
-	sl := c.servers[1]
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnStreamLeader(globalAccountName, "EVENTS")
-		if s := c.streamLeader(globalAccountName, "EVENTS"); s != sl {
-			s.JetStreamStepdownStream(globalAccountName, "EVENTS")
-			return fmt.Errorf("Server %s is not stream leader yet", sl)
-		}
-		return nil
-	})
-
-	// Create a fast ack consumer.
-	_, err = js.Subscribe("EV.NEW", func(m *nats.Msg) {
-		m.Ack()
-	}, nats.Durable("C"), nats.ManualAck())
-	require_NoError(t, err)
-
-	// Make sure the consumer leader is on S3.
-	cl := c.servers[2]
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnConsumerLeader(globalAccountName, "EVENTS", "C")
-		if s := c.consumerLeader(globalAccountName, "EVENTS", "C"); s != cl {
-			s.JetStreamStepdownConsumer(globalAccountName, "EVENTS", "C")
-			return fmt.Errorf("Server %s is not consumer leader yet", cl)
-		}
-		return nil
-	})
-
-	// Connect a client directly to the stream leader.
-	nc, js = jsClientConnect(t, sl)
-	defer nc.Close()
-
-	// Now create a pull subscriber.
-	sub, err := js.PullSubscribe("EV.NEW", "D", nats.ManualAck())
-	require_NoError(t, err)
-
-	// Make sure this consumer leader is on S1.
-	cl = c.servers[0]
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnConsumerLeader(globalAccountName, "EVENTS", "D")
-		if s := c.consumerLeader(globalAccountName, "EVENTS", "D"); s != cl {
-			s.JetStreamStepdownConsumer(globalAccountName, "EVENTS", "D")
-			return fmt.Errorf("Server %s is not consumer leader yet", cl)
-		}
-		return nil
-	})
-
-	numToSend := 1000
-	for i := 0; i < numToSend; i++ {
-		_, err := js.PublishAsync("EV.NEW", nil)
-		require_NoError(t, err)
-	}
-	select {
-	case <-js.PublishAsyncComplete():
-	case <-time.After(20 * time.Second):
-		t.Fatalf("Did not receive completion signal")
-	}
-
-	// Now make sure we can pull messages since we have not acked.
-	// The bug is that the acks arrive on S1 faster then the messages but we want to
-	// make sure we do not remove prematurely.
-	msgs, err := sub.Fetch(100, nats.MaxWait(time.Second))
-	require_NoError(t, err)
-	require_True(t, len(msgs) == 100)
-	for _, m := range msgs {
-		m.AckSync()
-	}
-
-	ci, err := js.ConsumerInfo("EVENTS", "D")
-	require_NoError(t, err)
-	require_True(t, ci.NumPending == uint64(numToSend-100))
-	require_True(t, ci.NumAckPending == 0)
-	require_True(t, ci.Delivered.Stream == 100)
-	require_True(t, ci.AckFloor.Stream == 100)
-
-	// Check stream state on all servers.
-	for _, s := range c.servers {
-		mset, err := s.GlobalAccount().lookupStream("EVENTS")
-		require_NoError(t, err)
-		state := mset.state()
-		require_True(t, state.Msgs == 900)
-		require_True(t, state.FirstSeq == 101)
-		require_True(t, state.LastSeq == 1000)
-		require_True(t, state.Consumers == 2)
-	}
-
-	msgs, err = sub.Fetch(900, nats.MaxWait(time.Second))
-	require_NoError(t, err)
-	require_True(t, len(msgs) == 900)
-	for _, m := range msgs {
-		m.AckSync()
-	}
-
-	// Let acks propagate.
-	time.Sleep(250 * time.Millisecond)
-
-	// Check final stream state on all servers.
-	for _, s := range c.servers {
-		mset, err := s.GlobalAccount().lookupStream("EVENTS")
-		require_NoError(t, err)
-		state := mset.state()
-		require_True(t, state.Msgs == 0)
-		require_True(t, state.FirstSeq == 1001)
-		require_True(t, state.LastSeq == 1000)
-		require_True(t, state.Consumers == 2)
-		// Now check preAcks
-		mset.mu.RLock()
-		numPreAcks := len(mset.preAcks)
-		mset.mu.RUnlock()
-		require_True(t, numPreAcks == 0)
-	}
-}
-
-func TestNoRaceJetStreamClusterUnbalancedInterestMultipleFilteredConsumers(t *testing.T) {
-	c, np := createStretchUnbalancedCluster(t)
-	defer c.shutdown()
-	defer np.stop()
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	// Now create the stream.
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:      "EVENTS",
-		Subjects:  []string{"EV.>"},
-		Replicas:  3,
-		Retention: nats.InterestPolicy,
-	})
-	require_NoError(t, err)
-
-	// Make sure it's leader is on S2.
-	sl := c.servers[1]
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnStreamLeader(globalAccountName, "EVENTS")
-		if s := c.streamLeader(globalAccountName, "EVENTS"); s != sl {
-			s.JetStreamStepdownStream(globalAccountName, "EVENTS")
-			return fmt.Errorf("Server %s is not stream leader yet", sl)
-		}
-		return nil
-	})
-
-	// Create a fast ack consumer.
-	_, err = js.Subscribe("EV.NEW", func(m *nats.Msg) {
-		m.Ack()
-	}, nats.Durable("C"), nats.ManualAck())
-	require_NoError(t, err)
-
-	// Make sure the consumer leader is on S3.
-	cl := c.servers[2]
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnConsumerLeader(globalAccountName, "EVENTS", "C")
-		if s := c.consumerLeader(globalAccountName, "EVENTS", "C"); s != cl {
-			s.JetStreamStepdownConsumer(globalAccountName, "EVENTS", "C")
-			return fmt.Errorf("Server %s is not consumer leader yet", cl)
-		}
-		return nil
-	})
-
-	// Connect a client directly to the stream leader.
-	nc, js = jsClientConnect(t, sl)
-	defer nc.Close()
-
-	// Now create another fast ack consumer.
-	_, err = js.Subscribe("EV.UPDATED", func(m *nats.Msg) {
-		m.Ack()
-	}, nats.Durable("D"), nats.ManualAck())
-	require_NoError(t, err)
-
-	// Make sure this consumer leader is on S1.
-	cl = c.servers[0]
-	checkFor(t, 20*time.Second, 200*time.Millisecond, func() error {
-		c.waitOnConsumerLeader(globalAccountName, "EVENTS", "D")
-		if s := c.consumerLeader(globalAccountName, "EVENTS", "D"); s != cl {
-			s.JetStreamStepdownConsumer(globalAccountName, "EVENTS", "D")
-			return fmt.Errorf("Server %s is not consumer leader yet", cl)
-		}
-		return nil
-	})
-
-	numToSend := 500
-	for i := 0; i < numToSend; i++ {
-		_, err := js.PublishAsync("EV.NEW", nil)
-		require_NoError(t, err)
-		_, err = js.PublishAsync("EV.UPDATED", nil)
-		require_NoError(t, err)
-	}
-	select {
-	case <-js.PublishAsyncComplete():
-	case <-time.After(20 * time.Second):
-		t.Fatalf("Did not receive completion signal")
-	}
-
-	// Let acks propagate.
-	time.Sleep(250 * time.Millisecond)
-
-	ci, err := js.ConsumerInfo("EVENTS", "D")
-	require_NoError(t, err)
-	require_True(t, ci.NumPending == 0)
-	require_True(t, ci.NumAckPending == 0)
-	require_True(t, ci.Delivered.Consumer == 500)
-	require_True(t, ci.Delivered.Stream == 1000)
-	require_True(t, ci.AckFloor.Consumer == 500)
-	require_True(t, ci.AckFloor.Stream == 1000)
-
-	// Check final stream state on all servers.
-	for _, s := range c.servers {
-		mset, err := s.GlobalAccount().lookupStream("EVENTS")
-		require_NoError(t, err)
-		state := mset.state()
-		require_True(t, state.Msgs == 0)
-		require_True(t, state.FirstSeq == 1001)
-		require_True(t, state.LastSeq == 1000)
-		require_True(t, state.Consumers == 2)
-		// Now check preAcks
-		mset.mu.RLock()
-		numPreAcks := len(mset.preAcks)
-		mset.mu.RUnlock()
-		require_True(t, numPreAcks == 0)
-	}
-}
diff --git a/server/opts_test.go b/server/opts_test.go
index d52ae4c4..6764728e 100644
--- a/server/opts_test.go
+++ b/server/opts_test.go
@@ -2533,9 +2533,13 @@ func TestParsingLeafNodeRemotes(t *testing.T) {
 }
 
 func TestLargeMaxControlLine(t *testing.T) {
-	confFileName := createConfFile(t, []byte(`
-		max_control_line = 3000000000
-	`))
+	confFileName := "big_mcl.conf"
+	content := `
+    max_control_line = 3000000000
+    `
+	if err := os.WriteFile(confFileName, []byte(content), 0666); err != nil {
+		t.Fatalf("Error writing config file: %v", err)
+	}
 	if _, err := ProcessConfigFile(confFileName); err == nil {
 		t.Fatalf("Expected an error from too large of a max_control_line entry")
 	}
diff --git a/server/test_test.go b/server/test_test.go
index 64717f90..a853310f 100644
--- a/server/test_test.go
+++ b/server/test_test.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2023 The NATS Authors
+// Copyright 2019-2021 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
@@ -52,6 +52,14 @@ func RunRandClientPortServer() *Server {
 	return RunServer(&opts)
 }
 
+// Used to setup clusters of clusters for tests.
+type cluster struct {
+	servers []*Server
+	opts    []*Options
+	name    string
+	t       testing.TB
+}
+
 func require_True(t *testing.T, b bool) {
 	t.Helper()
 	if !b {
@@ -268,11 +276,6 @@ func (c *cluster) shutdown() {
 	if c == nil {
 		return
 	}
-	// Stop any proxies.
-	for _, np := range c.nproxies {
-		np.stop()
-	}
-	// Shutdown and cleanup servers.
 	for i, s := range c.servers {
 		sd := s.StoreDir()
 		s.Shutdown()
diff --git a/test/cluster_test.go b/test/cluster_test.go
index 86ba7c0c..f8354d46 100644
--- a/test/cluster_test.go
+++ b/test/cluster_test.go
@@ -240,6 +240,10 @@ func TestClusterQueueSubs(t *testing.T) {
 	sendB("PING\r\n")
 	expectB(pongRe)
 
+	// Give plenty of time for the messages to flush, so that we don't
+	// accidentally only read some of them.
+	time.Sleep(time.Millisecond * 250)
+
 	// Should receive 5.
 	matches = expectMsgsA(5)
 	checkForQueueSid(t, matches, qg1SidsA)
@@ -248,6 +252,10 @@ func TestClusterQueueSubs(t *testing.T) {
 	// Send to A
 	sendA("PUB foo 2\r\nok\r\n")
 
+	// Give plenty of time for the messages to flush, so that we don't
+	// accidentally only read some of them.
+	time.Sleep(time.Millisecond * 250)
+
 	// Should receive 5.
 	matches = expectMsgsA(5)
 	checkForQueueSid(t, matches, qg1SidsA)
@@ -270,6 +278,10 @@ func TestClusterQueueSubs(t *testing.T) {
 	// Send to B
 	sendB("PUB foo 2\r\nok\r\n")
 
+	// Give plenty of time for the messages to flush, so that we don't
+	// accidentally only read some of them.
+	time.Sleep(time.Millisecond * 250)
+
 	// Should receive 1 from B.
 	matches = expectMsgsB(1)
 	checkForQueueSid(t, matches, qg2SidsB)
@@ -308,6 +320,10 @@ func TestClusterQueueSubs(t *testing.T) {
 	// Send to A
 	sendA("PUB foo 2\r\nok\r\n")
 
+	// Give plenty of time for the messages to flush, so that we don't
+	// accidentally only read some of them.
+	time.Sleep(time.Millisecond * 250)
+
 	// Should receive 4 now.
 	matches = expectMsgsA(4)
 	checkForPubSids(t, matches, pSids)