Repository: nats-io/nats-server
Pull request #4016 (2023-04-03 14:41:04)

Alternative fix for issue 4014, where we always send pings on ROUTER connections
Alternative fix for issue https://github.com/nats-io/nats-server/issues/4014: where we always send pings for ROUTER, GATEWAY and LEAF spoke connections.

This is my own original work that I license to the project

[ x] Link to issue, e.g. Resolves #NNN
 Documentation added (if applicable)
 Tests added
[ x] Branch rebased on top of current main (git pull --rebase origin main)
[ x] Changes squashed to a single commit (described [here](http://gitready.com/advanced/2009/02/10/squashing-commits-with-rebase.html))
[x] Build is green in Travis CI
[ x] You have certified that the contribution is your original work and that you license the work to the project under the [Apache 2 license](https://github.com/nats-io/nats-server/blob/main/LICENSE)
Resolves #

Changes proposed in this pull request:
- in processPingTimer, add explicit test for ROUTER connection, and only hold off sending ping based on receive ping

sandykellagher: I've made a change so that the c.ping.last timestamp is updated for receive client data as well as pings, and squashed this commit with the previous one. I've also marked the other PR as closed.

Patch:
diff --git a/.github/workflows/nightly.yaml b/.github/workflows/nightly.yaml
index 61433fe2..19dbd475 100644
--- a/.github/workflows/nightly.yaml
+++ b/.github/workflows/nightly.yaml
@@ -1,15 +1,11 @@
 name: "NATS Server Nightly: DEV"
 on:
-  workflow_dispatch:
-    inputs:
-      target:
-        description: "Override image branch (optional)"
-        type: string
-        required: false
+  workflow_dispatch: {}
 
   schedule:
     - cron: "40 4 * * *"
 
+
 jobs:
   nightly_release:
     runs-on: ubuntu-latest
@@ -18,7 +14,7 @@ jobs:
         uses: actions/checkout@v3
         with:
           path: src/github.com/nats-io/nats-server
-          ref: ${{ inputs.target || 'dev' }}
+          ref: dev
 
       - uses: ./src/github.com/nats-io/nats-server/.github/actions/nightly-release
         with:
@@ -26,4 +22,4 @@ jobs:
           workdir: src/github.com/nats-io/nats-server
           label: nightly
           hub_username: "${{ secrets.DOCKER_USERNAME }}"
-          hub_password: "${{ secrets.DOCKER_PASSWORD }}"
+          hub_password: "${{ secrets.DOCKER_PASSWORD }}"
\ No newline at end of file
diff --git a/.github/workflows/rc_nightly.yaml b/.github/workflows/rc_nightly.yaml
index 2228e226..9dc68f7d 100644
--- a/.github/workflows/rc_nightly.yaml
+++ b/.github/workflows/rc_nightly.yaml
@@ -1,15 +1,11 @@
 name: "NATS Server Nightly: MAIN"
 on:
-  workflow_dispatch:
-    inputs:
-      target:
-        description: "Override image branch (optional)"
-        type: string
-        required: false
+  workflow_dispatch: {}
 
   schedule:
     - cron: "40 4 * * *"
 
+
 jobs:
   nightly_main_release:
     runs-on: ubuntu-latest
@@ -18,7 +14,7 @@ jobs:
         uses: actions/checkout@v3
         with:
           path: src/github.com/nats-io/nats-server
-          ref: ${{ inputs.target || 'main' }}
+          ref: main
 
       - uses: ./src/github.com/nats-io/nats-server/.github/actions/nightly-release
         with:
@@ -26,4 +22,4 @@ jobs:
           workdir: src/github.com/nats-io/nats-server
           label: nightly-main
           hub_username: "${{ secrets.DOCKER_USERNAME }}"
-          hub_password: "${{ secrets.DOCKER_PASSWORD }}"
+          hub_password: "${{ secrets.DOCKER_PASSWORD }}"
\ No newline at end of file
diff --git a/server/client.go b/server/client.go
index 25d8f0be..ccae5068 100644
--- a/server/client.go
+++ b/server/client.go
@@ -253,7 +253,9 @@ type client struct {
 	ping       pinfo
 	msgb       [msgScratchSize]byte
 	last       time.Time
-	headers    bool
+	lastIn     time.Time
+
+	headers bool
 
 	rtt      time.Duration
 	rttStart time.Time
@@ -286,9 +288,8 @@ type rrTracking struct {
 
 // Struct for PING initiation from the server.
 type pinfo struct {
-	tmr  *time.Timer
-	last time.Time
-	out  int
+	tmr *time.Timer
+	out int
 }
 
 // outbound holds pending data for a socket.
@@ -1307,8 +1308,10 @@ func (c *client) readLoop(pre []byte) {
 		c.mu.Lock()
 
 		// Activity based on interest changes or data/msgs.
+		// Also update last receive activity for ping sender
 		if c.in.msgs > 0 || c.in.subs > 0 {
 			c.last = last
+			c.lastIn = last
 		}
 
 		if n >= cap(b) {
@@ -2196,7 +2199,7 @@ func (c *client) processPing() {
 
 	// Record this to suppress us sending one if this
 	// is within a given time interval for activity.
-	c.ping.last = time.Now()
+	c.lastIn = time.Now()
 
 	// If not a CLIENT, we are done. Also the CONNECT should
 	// have been received, but make sure it is so before proceeding
@@ -4564,17 +4567,14 @@ func (c *client) processPingTimer() {
 	now := time.Now()
 	needRTT := c.rtt == 0 || now.Sub(c.rttStart) > DEFAULT_RTT_MEASUREMENT_INTERVAL
 
-	// Do not delay PINGs for GATEWAY or spoke LEAF connections.
-	if c.kind == GATEWAY || c.isSpokeLeafNode() {
+	// Do not delay PINGs for ROUTER, GATEWAY or spoke LEAF connections.
+	if c.kind == ROUTER || c.kind == GATEWAY || c.isSpokeLeafNode() {
 		sendPing = true
 	} else {
-		// If we have had activity within the PingInterval then
-		// there is no need to send a ping. This can be client data
-		// or if we received a ping from the other side.
-		if delta := now.Sub(c.last); delta < pingInterval && !needRTT {
-			c.Debugf("Delaying PING due to client activity %v ago", delta.Round(time.Second))
-		} else if delta := now.Sub(c.ping.last); delta < pingInterval && !needRTT {
-			c.Debugf("Delaying PING due to remote ping %v ago", delta.Round(time.Second))
+		// If we received client data or a ping from the other side within the PingInterval,
+		// then there is no need to send a ping.
+		if delta := now.Sub(c.lastIn); delta < pingInterval && !needRTT {
+			c.Debugf("Delaying PING due to remote client data or ping %v ago", delta.Round(time.Second))
 		} else {
 			sendPing = true
 		}
diff --git a/server/const.go b/server/const.go
index 5e7816e6..8eaf86a1 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.6"
+	VERSION = "2.9.16-RC.4"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index f3ac3b04..3f4f9103 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -254,7 +254,6 @@ type consumer struct {
 	ackReplyT         string
 	ackSubj           string
 	nextMsgSubj       string
-	nextMsgReqs       *ipQueue[*nextMsgReq]
 	maxp              int
 	pblimit           int
 	maxpb             int
@@ -751,8 +750,6 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	// Create our request waiting queue.
 	if o.isPullMode() {
 		o.waiting = newWaitQueue(config.MaxWaiting)
-		// Create our internal queue for next msg requests.
-		o.nextMsgReqs = newIPQueue[*nextMsgReq](s, fmt.Sprintf("[ACC:%s] consumer '%s' on stream '%s' pull requests", accName, o.name, mset.cfg.Name))
 	}
 
 	// Check if we have  filtered subject that is a wildcard.
@@ -1102,7 +1099,6 @@ func (o *consumer) setLeader(isLeader bool) {
 		if node != nil && o.pch == nil {
 			o.pch = make(chan struct{}, 1)
 		}
-		pullMode := o.isPullMode()
 		o.mu.Unlock()
 
 		// Snapshot initial info.
@@ -1114,12 +1110,6 @@ func (o *consumer) setLeader(isLeader bool) {
 		// Now start up Go routine to process acks.
 		go o.processInboundAcks(qch)
 
-		if pullMode {
-			// Now start up Go routine to process inbound next message requests.
-			go o.processInboundNextMsgReqs(qch)
-
-		}
-
 		// If we are R>1 spin up our proposal loop.
 		if node != nil {
 			// Determine if we can send pending requests info to the group.
@@ -1155,7 +1145,6 @@ func (o *consumer) setLeader(isLeader bool) {
 			if !o.isDurable() {
 				stopAndClearTimer(&o.dtmr)
 			}
-			o.nextMsgReqs.drain()
 		} else if o.srv.gateway.enabled {
 			stopAndClearTimer(&o.gwdtmr)
 		}
@@ -2844,37 +2833,6 @@ func (o *consumer) nextWaiting(sz int) *waitingRequest {
 	return nil
 }
 
-// Next message request.
-type nextMsgReq struct {
-	reply string
-	msg   []byte
-}
-
-var nextMsgReqPool sync.Pool
-
-func newNextMsgReq(reply string, msg []byte) *nextMsgReq {
-	var nmr *nextMsgReq
-	m := nextMsgReqPool.Get()
-	if m != nil {
-		nmr = m.(*nextMsgReq)
-	} else {
-		nmr = &nextMsgReq{}
-	}
-	// When getting something from a pool it is critical that all fields are
-	// initialized. Doing this way guarantees that if someone adds a field to
-	// the structure, the compiler will fail the build if this line is not updated.
-	(*nmr) = nextMsgReq{reply, msg}
-	return nmr
-}
-
-func (nmr *nextMsgReq) returnToPool() {
-	if nmr == nil {
-		return
-	}
-	nmr.reply, nmr.msg = _EMPTY_, nil
-	nextMsgReqPool.Put(nmr)
-}
-
 // processNextMsgReq will process a request for the next message available. A nil message payload means deliver
 // a single message. If the payload is a formal request or a number parseable with Atoi(), then we will send a
 // batch of messages without requiring another request to this endpoint, or an ACK.
@@ -2882,16 +2840,21 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 	if reply == _EMPTY_ {
 		return
 	}
+	_, msg = c.msgParts(msg)
 
-	// Short circuit error here.
-	if o.nextMsgReqs == nil {
-		hdr := []byte("NATS/1.0 409 Consumer is push based\r\n\r\n")
-		o.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
-		return
+	inlineOk := c.kind != ROUTER && c.kind != GATEWAY && c.kind != LEAF
+	if !inlineOk {
+		// Check how long we have been away from the readloop for the route or gateway or leafnode.
+		// If too long move to a separate go routine.
+		if elapsed := time.Since(c.in.start); elapsed < noBlockThresh {
+			inlineOk = true
+		}
+	}
+	if inlineOk {
+		o.processNextMsgRequest(reply, msg)
+	} else {
+		go o.processNextMsgRequest(reply, copyBytes(msg))
 	}
-
-	_, msg = c.msgParts(msg)
-	o.nextMsgReqs.push(newNextMsgReq(reply, copyBytes(msg)))
 }
 
 func (o *consumer) processNextMsgRequest(reply string, msg []byte) {
@@ -3261,30 +3224,6 @@ func (o *consumer) processInboundAcks(qch chan struct{}) {
 	}
 }
 
-// Process inbound next message requests.
-func (o *consumer) processInboundNextMsgReqs(qch chan struct{}) {
-	// Grab the server lock to watch for server quit.
-	o.mu.RLock()
-	s := o.srv
-	o.mu.RUnlock()
-
-	for {
-		select {
-		case <-o.nextMsgReqs.ch:
-			reqs := o.nextMsgReqs.pop()
-			for _, req := range reqs {
-				o.processNextMsgRequest(req.reply, req.msg)
-				req.returnToPool()
-			}
-			o.nextMsgReqs.recycle(&reqs)
-		case <-qch:
-			return
-		case <-s.quitCh:
-			return
-		}
-	}
-}
-
 // Suppress auto cleanup on ack activity of any kind.
 func (o *consumer) suppressDeletion() {
 	o.mu.Lock()
@@ -4371,9 +4310,6 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 	n := o.node
 	qgroup := o.cfg.DeliverGroup
 	o.ackMsgs.unregister()
-	if o.nextMsgReqs != nil {
-		o.nextMsgReqs.unregister()
-	}
 
 	// For cleaning up the node assignment.
 	var ca *consumerAssignment
diff --git a/server/events.go b/server/events.go
index e1fd581f..0ff6a257 100644
--- a/server/events.go
+++ b/server/events.go
@@ -1,4 +1,4 @@
-// Copyright 2018-2023 The NATS Authors
+// Copyright 2018-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
@@ -83,20 +83,6 @@ const (
 // FIXME(dlc) - make configurable.
 var eventsHBInterval = 30 * time.Second
 
-type sysMsgHandler func(sub *subscription, client *client, acc *Account, subject, reply string, hdr, msg []byte)
-
-// Used if we have to queue things internally to avoid the route/gw path.
-type inSysMsg struct {
-	sub  *subscription
-	c    *client
-	acc  *Account
-	subj string
-	rply string
-	hdr  []byte
-	msg  []byte
-	cb   sysMsgHandler
-}
-
 // Used to send and receive messages from inside the server.
 type internal struct {
 	account  *Account
@@ -108,7 +94,6 @@ type internal struct {
 	stmr     *time.Timer
 	replies  map[string]msgHandler
 	sendq    *ipQueue[*pubMsg]
-	recvq    *ipQueue[*inSysMsg]
 	resetCh  chan struct{}
 	wg       sync.WaitGroup
 	sq       *sendq
@@ -315,33 +300,6 @@ type TypedEvent struct {
 	Time time.Time `json:"timestamp"`
 }
 
-// internalReceiveLoop will be responsible for dispatching all messages that
-// a server receives and needs to internally process, e.g. internal subs.
-func (s *Server) internalReceiveLoop() {
-	s.mu.RLock()
-	if s.sys == nil || s.sys.recvq == nil {
-		s.mu.RUnlock()
-		return
-	}
-	recvq := s.sys.recvq
-	s.mu.RUnlock()
-
-	for s.eventsRunning() {
-		select {
-		case <-recvq.ch:
-			msgs := recvq.pop()
-			for _, m := range msgs {
-				if m.cb != nil {
-					m.cb(m.sub, m.c, m.acc, m.subj, m.rply, m.hdr, m.msg)
-				}
-			}
-			recvq.recycle(&msgs)
-		case <-s.quitCh:
-			return
-		}
-	}
-}
-
 // internalSendLoop will be responsible for serializing all messages that
 // a server wants to send.
 func (s *Server) internalSendLoop(wg *sync.WaitGroup) {
@@ -883,26 +841,26 @@ func (s *Server) initEventTracking() {
 	s.sys.inboxPre = subject
 	// This is for remote updates for connection accounting.
 	subject = fmt.Sprintf(accConnsEventSubjOld, "*")
-	if _, err := s.sysSubscribe(subject, s.noInlineCallback(s.remoteConnsUpdate)); err != nil {
+	if _, err := s.sysSubscribe(subject, s.remoteConnsUpdate); err != nil {
 		s.Errorf("Error setting up internal tracking for %s: %v", subject, err)
 	}
 	// This will be for responses for account info that we send out.
 	subject = fmt.Sprintf(connsRespSubj, s.info.ID)
-	if _, err := s.sysSubscribe(subject, s.noInlineCallback(s.remoteConnsUpdate)); err != nil {
+	if _, err := s.sysSubscribe(subject, s.remoteConnsUpdate); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
 	}
 	// Listen for broad requests to respond with number of subscriptions for a given subject.
-	if _, err := s.sysSubscribe(accNumSubsReqSubj, s.noInlineCallback(s.nsubsRequest)); err != nil {
+	if _, err := s.sysSubscribe(accNumSubsReqSubj, s.nsubsRequest); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
 	}
 	// Listen for statsz from others.
 	subject = fmt.Sprintf(serverStatsSubj, "*")
-	if _, err := s.sysSubscribe(subject, s.noInlineCallback(s.remoteServerUpdate)); err != nil {
+	if _, err := s.sysSubscribe(subject, s.remoteServerUpdate); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
 	}
 	// Listen for all server shutdowns.
 	subject = fmt.Sprintf(shutdownEventSubj, "*")
-	if _, err := s.sysSubscribe(subject, s.noInlineCallback(s.remoteServerShutdown)); err != nil {
+	if _, err := s.sysSubscribe(subject, s.remoteServerShutdown); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
 	}
 	// Listen for account claims updates.
@@ -912,62 +870,62 @@ func (s *Server) initEventTracking() {
 	}
 	if subscribeToUpdate {
 		for _, sub := range []string{accUpdateEventSubjOld, accUpdateEventSubjNew} {
-			if _, err := s.sysSubscribe(fmt.Sprintf(sub, "*"), s.noInlineCallback(s.accountClaimUpdate)); err != nil {
+			if _, err := s.sysSubscribe(fmt.Sprintf(sub, "*"), s.accountClaimUpdate); err != nil {
 				s.Errorf("Error setting up internal tracking: %v", err)
 			}
 		}
 	}
 	// Listen for ping messages that will be sent to all servers for statsz.
 	// This subscription is kept for backwards compatibility. Got replaced by ...PING.STATZ from below
-	if _, err := s.sysSubscribe(serverStatsPingReqSubj, s.noInlineCallback(s.statszReq)); err != nil {
+	if _, err := s.sysSubscribe(serverStatsPingReqSubj, s.statszReq); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
 	}
-	monSrvc := map[string]sysMsgHandler{
+	monSrvc := map[string]msgHandler{
 		"STATSZ": s.statszReq,
-		"VARZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"VARZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &VarzEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Varz(&optz.VarzOptions) })
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Varz(&optz.VarzOptions) })
 		},
-		"SUBSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"SUBSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &SubszEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Subsz(&optz.SubszOptions) })
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Subsz(&optz.SubszOptions) })
 		},
-		"CONNZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"CONNZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &ConnzEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Connz(&optz.ConnzOptions) })
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Connz(&optz.ConnzOptions) })
 		},
-		"ROUTEZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"ROUTEZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &RoutezEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Routez(&optz.RoutezOptions) })
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Routez(&optz.RoutezOptions) })
 		},
-		"GATEWAYZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"GATEWAYZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &GatewayzEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Gatewayz(&optz.GatewayzOptions) })
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Gatewayz(&optz.GatewayzOptions) })
 		},
-		"LEAFZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"LEAFZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &LeafzEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Leafz(&optz.LeafzOptions) })
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Leafz(&optz.LeafzOptions) })
 		},
-		"ACCOUNTZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"ACCOUNTZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &AccountzEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Accountz(&optz.AccountzOptions) })
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Accountz(&optz.AccountzOptions) })
 		},
-		"JSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"JSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &JszEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Jsz(&optz.JSzOptions) })
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Jsz(&optz.JSzOptions) })
 		},
-		"HEALTHZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"HEALTHZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &HealthzEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.healthz(&optz.HealthzOptions), nil })
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.healthz(&optz.HealthzOptions), nil })
 		},
 	}
 	for name, req := range monSrvc {
 		subject = fmt.Sprintf(serverDirectReqSubj, s.info.ID, name)
-		if _, err := s.sysSubscribe(subject, s.noInlineCallback(req)); err != nil {
+		if _, err := s.sysSubscribe(subject, req); err != nil {
 			s.Errorf("Error setting up internal tracking: %v", err)
 		}
 		subject = fmt.Sprintf(serverPingReqSubj, name)
-		if _, err := s.sysSubscribe(subject, s.noInlineCallback(req)); err != nil {
+		if _, err := s.sysSubscribe(subject, req); err != nil {
 			s.Errorf("Error setting up internal tracking: %v", err)
 		}
 	}
@@ -978,10 +936,10 @@ func (s *Server) initEventTracking() {
 			return tk[accReqAccIndex], nil
 		}
 	}
-	monAccSrvc := map[string]sysMsgHandler{
-		"SUBSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+	monAccSrvc := map[string]msgHandler{
+		"SUBSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &SubszEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(c, subject, msg); err != nil {
 					return nil, err
 				} else {
@@ -991,9 +949,9 @@ func (s *Server) initEventTracking() {
 				}
 			})
 		},
-		"CONNZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"CONNZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &ConnzEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(c, subject, msg); err != nil {
 					return nil, err
 				} else {
@@ -1002,9 +960,9 @@ func (s *Server) initEventTracking() {
 				}
 			})
 		},
-		"LEAFZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"LEAFZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &LeafzEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(c, subject, msg); err != nil {
 					return nil, err
 				} else {
@@ -1013,9 +971,9 @@ func (s *Server) initEventTracking() {
 				}
 			})
 		},
-		"JSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"JSZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &JszEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(c, subject, msg); err != nil {
 					return nil, err
 				} else {
@@ -1024,9 +982,9 @@ func (s *Server) initEventTracking() {
 				}
 			})
 		},
-		"INFO": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"INFO": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &AccInfoEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(c, subject, msg); err != nil {
 					return nil, err
 				} else {
@@ -1037,9 +995,9 @@ func (s *Server) initEventTracking() {
 		// STATZ is essentially a duplicate of CONNS with an envelope identical to the others.
 		// For historical reasons CONNS is the odd one out.
 		// STATZ is also less heavy weight than INFO
-		"STATZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		"STATZ": func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &AccountStatzEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if acc, err := extractAccount(c, subject, msg); err != nil {
 					return nil, err
 				} else if acc == "PING" { // Filter PING subject. Happens for server as well. But wildcards are not used
@@ -1059,16 +1017,16 @@ func (s *Server) initEventTracking() {
 		"CONNS": s.connsRequest,
 	}
 	for name, req := range monAccSrvc {
-		if _, err := s.sysSubscribe(fmt.Sprintf(accDirectReqSubj, "*", name), s.noInlineCallback(req)); err != nil {
+		if _, err := s.sysSubscribe(fmt.Sprintf(accDirectReqSubj, "*", name), req); err != nil {
 			s.Errorf("Error setting up internal tracking: %v", err)
 		}
 	}
 
 	// For now only the STATZ subject has an account specific ping equivalent.
 	if _, err := s.sysSubscribe(fmt.Sprintf(accPingReqSubj, "STATZ"),
-		s.noInlineCallback(func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+		func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 			optz := &AccountStatzEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
+			s.zReq(c, reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
 				if stz, err := s.AccountStatz(&optz.AccountStatzOptions); err != nil {
 					return nil, err
 				} else if len(stz.Accounts) == 0 && !optz.IncludeUnused {
@@ -1077,23 +1035,23 @@ func (s *Server) initEventTracking() {
 					return stz, nil
 				}
 			})
-		})); err != nil {
+		}); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
 	}
 
 	// Listen for updates when leaf nodes connect for a given account. This will
 	// force any gateway connections to move to `modeInterestOnly`
 	subject = fmt.Sprintf(leafNodeConnectEventSubj, "*")
-	if _, err := s.sysSubscribe(subject, s.noInlineCallback(s.leafNodeConnected)); err != nil {
+	if _, err := s.sysSubscribe(subject, s.leafNodeConnected); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
 	}
 	// For tracking remote latency measurements.
 	subject = fmt.Sprintf(remoteLatencyEventSubj, s.sys.shash)
-	if _, err := s.sysSubscribe(subject, s.noInlineCallback(s.remoteLatencyUpdate)); err != nil {
+	if _, err := s.sysSubscribe(subject, s.remoteLatencyUpdate); err != nil {
 		s.Errorf("Error setting up internal latency tracking: %v", err)
 	}
 	// This is for simple debugging of number of subscribers that exist in the system.
-	if _, err := s.sysSubscribeInternal(accSubsSubj, s.noInlineCallback(s.debugSubscribers)); err != nil {
+	if _, err := s.sysSubscribeInternal(accSubsSubj, s.debugSubscribers); err != nil {
 		s.Errorf("Error setting up internal debug service for subscribers: %v", err)
 	}
 }
@@ -1161,7 +1119,7 @@ func (s *Server) addSystemAccountExports(sacc *Account) {
 }
 
 // accountClaimUpdate will receive claim updates for accounts.
-func (s *Server) accountClaimUpdate(sub *subscription, c *client, _ *Account, subject, resp string, hdr, msg []byte) {
+func (s *Server) accountClaimUpdate(sub *subscription, c *client, _ *Account, subject, resp string, rmsg []byte) {
 	if !s.EventsEnabled() {
 		return
 	}
@@ -1175,7 +1133,7 @@ func (s *Server) accountClaimUpdate(sub *subscription, c *client, _ *Account, su
 		s.Debugf("Received account claims update on bad subject %q", subject)
 		return
 	}
-	if len(msg) == 0 {
+	if _, msg := c.msgParts(rmsg); len(msg) == 0 {
 		err := errors.New("request body is empty")
 		respondToUpdate(s, resp, pubKey, "jwt update error", err)
 	} else if claim, err := jwt.DecodeAccountClaims(string(msg)); err != nil {
@@ -1218,7 +1176,7 @@ func (s *Server) sameDomain(domain string) bool {
 }
 
 // remoteServerShutdownEvent is called when we get an event from another server shutting down.
-func (s *Server) remoteServerShutdown(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+func (s *Server) remoteServerShutdown(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if !s.eventsEnabled() {
@@ -1230,6 +1188,7 @@ func (s *Server) remoteServerShutdown(sub *subscription, c *client, _ *Account,
 		return
 	}
 
+	_, msg := c.msgParts(rmsg)
 	if len(msg) == 0 {
 		s.Errorf("Remote server sent invalid (empty) shutdown message to %q", subject)
 		return
@@ -1257,9 +1216,9 @@ func (s *Server) remoteServerShutdown(sub *subscription, c *client, _ *Account,
 }
 
 // remoteServerUpdate listens for statsz updates from other servers.
-func (s *Server) remoteServerUpdate(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+func (s *Server) remoteServerUpdate(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	var ssm ServerStatsMsg
-	if len(msg) == 0 {
+	if _, msg := c.msgParts(rmsg); len(msg) == 0 {
 		s.Debugf("Received empty server info for remote server update")
 		return
 	} else if err := json.Unmarshal(msg, &ssm); err != nil {
@@ -1389,7 +1348,7 @@ func (s *Server) shutdownEventing() {
 }
 
 // Request for our local connection count.
-func (s *Server) connsRequest(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+func (s *Server) connsRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if !s.eventsRunning() {
 		return
 	}
@@ -1400,7 +1359,7 @@ func (s *Server) connsRequest(sub *subscription, c *client, _ *Account, subject,
 	}
 	a := tk[accReqAccIndex]
 	m := accNumConnsReq{Account: a}
-	if len(msg) > 0 {
+	if _, msg := c.msgParts(rmsg); len(msg) > 0 {
 		if err := json.Unmarshal(msg, &m); err != nil {
 			s.sys.client.Errorf("Error unmarshalling account connections request message: %v", err)
 			return
@@ -1428,7 +1387,7 @@ func (s *Server) connsRequest(sub *subscription, c *client, _ *Account, subject,
 }
 
 // leafNodeConnected is an event we will receive when a leaf node for a given account connects.
-func (s *Server) leafNodeConnected(sub *subscription, _ *client, _ *Account, subject, reply string, hdr, msg []byte) {
+func (s *Server) leafNodeConnected(sub *subscription, _ *client, _ *Account, subject, reply string, msg []byte) {
 	m := accNumConnsReq{}
 	if err := json.Unmarshal(msg, &m); err != nil {
 		s.sys.client.Errorf("Error unmarshalling account connections request message: %v", err)
@@ -1587,7 +1546,7 @@ type ServerAPIConnzResponse struct {
 }
 
 // statszReq is a request for us to respond with current statsz.
-func (s *Server) statszReq(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+func (s *Server) statszReq(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if !s.EventsEnabled() {
 		return
 	}
@@ -1598,7 +1557,7 @@ func (s *Server) statszReq(sub *subscription, c *client, _ *Account, subject, re
 	}
 
 	opts := StatszEventOptions{}
-	if len(msg) != 0 {
+	if _, msg := c.msgParts(rmsg); len(msg) != 0 {
 		if err := json.Unmarshal(msg, &opts); err != nil {
 			response := &ServerAPIResponse{
 				Server: &ServerInfo{},
@@ -1637,13 +1596,14 @@ func getAcceptEncoding(hdr []byte) compressionType {
 	return unsupportedCompression
 }
 
-func (s *Server) zReq(c *client, reply string, hdr, msg []byte, fOpts *EventFilterOptions, optz interface{}, respf func() (interface{}, error)) {
+func (s *Server) zReq(c *client, reply string, rmsg []byte, fOpts *EventFilterOptions, optz interface{}, respf func() (interface{}, error)) {
 	if !s.EventsEnabled() || reply == _EMPTY_ {
 		return
 	}
 	response := &ServerAPIResponse{Server: &ServerInfo{}}
 	var err error
 	status := 0
+	hdr, msg := c.msgParts(rmsg)
 	if len(msg) != 0 {
 		if err = json.Unmarshal(msg, optz); err != nil {
 			status = http.StatusBadRequest // status is only included on error, so record how far execution got
@@ -1668,12 +1628,12 @@ func (s *Server) zReq(c *client, reply string, hdr, msg []byte, fOpts *EventFilt
 }
 
 // remoteConnsUpdate gets called when we receive a remote update from another server.
-func (s *Server) remoteConnsUpdate(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+func (s *Server) remoteConnsUpdate(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if !s.eventsRunning() {
 		return
 	}
 	var m AccountNumConns
-	if len(msg) == 0 {
+	if _, msg := c.msgParts(rmsg); len(msg) == 0 {
 		s.sys.client.Errorf("No message body provided")
 		return
 	} else if err := json.Unmarshal(msg, &m); err != nil {
@@ -2028,25 +1988,6 @@ func (s *Server) sendAuthErrorEvent(c *client) {
 // rmsg contains header and the message. use client.msgParts(rmsg) to split them apart
 type msgHandler func(sub *subscription, client *client, acc *Account, subject, reply string, rmsg []byte)
 
-// Create a wrapped callback handler for the subscription that will move it to an
-// internal recvQ for processing not inline with routes etc.
-func (s *Server) noInlineCallback(cb sysMsgHandler) msgHandler {
-	s.mu.RLock()
-	if !s.eventsEnabled() {
-		s.mu.RUnlock()
-		return nil
-	}
-	// Capture here for direct reference to avoid any unnecessary blocking inline with routes, gateways etc.
-	recvq := s.sys.recvq
-	s.mu.RUnlock()
-
-	return func(sub *subscription, c *client, acc *Account, subj, rply string, rmsg []byte) {
-		// Need to copy and split here.
-		hdr, msg := c.msgParts(rmsg)
-		recvq.push(&inSysMsg{sub, c, acc, subj, rply, copyBytes(hdr), copyBytes(msg), cb})
-	}
-}
-
 // Create an internal subscription. sysSubscribeQ for queue groups.
 func (s *Server) sysSubscribe(subject string, cb msgHandler) (*subscription, error) {
 	return s.systemSubscribe(subject, _EMPTY_, false, nil, cb)
@@ -2086,10 +2027,9 @@ func (s *Server) systemSubscribe(subject, queue string, internalOnly bool, c *cl
 	}
 
 	var q []byte
-	if queue != _EMPTY_ {
+	if queue != "" {
 		q = []byte(queue)
 	}
-
 	// Now create the subscription
 	return c.processSub([]byte(subject), q, []byte(sid), cb, internalOnly)
 }
@@ -2122,7 +2062,7 @@ func remoteLatencySubjectForResponse(subject []byte) string {
 }
 
 // remoteLatencyUpdate is used to track remote latency measurements for tracking on exported services.
-func (s *Server) remoteLatencyUpdate(sub *subscription, _ *client, _ *Account, subject, _ string, hdr, msg []byte) {
+func (s *Server) remoteLatencyUpdate(sub *subscription, _ *client, _ *Account, subject, _ string, msg []byte) {
 	if !s.eventsRunning() {
 		return
 	}
@@ -2258,7 +2198,7 @@ func totalSubs(rr *SublistResult, qg []byte) (nsubs int32) {
 
 // Allows users of large systems to debug active subscribers for a given subject.
 // Payload should be the subject of interest.
-func (s *Server) debugSubscribers(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+func (s *Server) debugSubscribers(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	// Even though this is an internal only subscription, meaning interest was not forwarded, we could
 	// get one here from a GW in optimistic mode. Ignore for now.
 	// FIXME(dlc) - Should we send no interest here back to the GW?
@@ -2266,26 +2206,8 @@ func (s *Server) debugSubscribers(sub *subscription, c *client, _ *Account, subj
 		return
 	}
 
-	var ci ClientInfo
-	if len(hdr) > 0 {
-		if err := json.Unmarshal(getHeader(ClientInfoHdr, hdr), &ci); err != nil {
-			return
-		}
-	}
-
-	var acc *Account
-	if ci.Service != _EMPTY_ {
-		acc, _ = s.LookupAccount(ci.Service)
-	} else if ci.Account != _EMPTY_ {
-		acc, _ = s.LookupAccount(ci.Account)
-	} else {
-		// Direct $SYS access.
-		acc = c.acc
-		if acc == nil {
-			acc = s.SystemAccount()
-		}
-	}
-	if acc == nil {
+	_, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	if err != nil {
 		return
 	}
 
@@ -2386,12 +2308,12 @@ func (s *Server) debugSubscribers(sub *subscription, c *client, _ *Account, subj
 
 // Request for our local subscription count. This will come from a remote origin server
 // that received the initial request.
-func (s *Server) nsubsRequest(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
+func (s *Server) nsubsRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	if !s.eventsRunning() {
 		return
 	}
 	m := accNumSubsReq{}
-	if len(msg) == 0 {
+	if _, msg := c.msgParts(rmsg); len(msg) == 0 {
 		s.sys.client.Errorf("request requires a body")
 		return
 	} else if err := json.Unmarshal(msg, &m); err != nil {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 924cf4f2..9c794c80 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -436,9 +436,11 @@ func (cc *jetStreamCluster) isStreamCurrent(account, stream string) bool {
 
 // isStreamHealthy will determine if the stream is up to date or very close.
 // For R1 it will make sure the stream is present on this server.
-// Read lock should be held.
 func (js *jetStream) isStreamHealthy(account, stream string) bool {
+	js.mu.RLock()
+	defer js.mu.RUnlock()
 	cc := js.cluster
+
 	if cc == nil {
 		// Non-clustered mode
 		return true
@@ -478,9 +480,11 @@ func (js *jetStream) isStreamHealthy(account, stream string) bool {
 
 // isConsumerCurrent will determine if the consumer is up to date.
 // For R1 it will make sure the consunmer is present on this server.
-// Read lock should be held.
 func (js *jetStream) isConsumerCurrent(account, stream, consumer string) bool {
+	js.mu.RLock()
+	defer js.mu.RUnlock()
 	cc := js.cluster
+
 	if cc == nil {
 		// Non-clustered mode
 		return true
@@ -1939,6 +1943,9 @@ 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() {
@@ -1947,11 +1954,6 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		defer mset.clearMonitorRunning()
 	}
 
-	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
-	// This should be below the checkInMonitor call though to avoid stopping it out
-	// from underneath the one that is running since it will be the same raft node.
-	defer n.Stop()
-
 	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())
@@ -4185,17 +4187,15 @@ 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
 	}
 	defer o.clearMonitorRunning()
 
-	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
-	// This should be below the checkInMonitor call though to avoid stopping it out
-	// from underneath the one that is running since it will be the same raft node.
-	defer n.Stop()
-
 	qch, lch, aq, uch, ourPeerId := n.QuitC(), n.LeadChangeC(), n.ApplyQ(), o.updateC(), cc.meta.ID()
 
 	s.Debugf("Starting consumer monitor for '%s > %s > %s' [%s]", o.acc.Name, ca.Stream, ca.Name, n.Group())
@@ -4293,16 +4293,16 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 					}
 					// Check our state if we are under an interest based stream.
 					o.checkStateForInterestStream()
-				} else if !recovering {
-					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)
-						}
-					} else {
-						s.Warnf("Error applying consumer entries to '%s > %s'", ca.Client.serviceAccount(), ca.Name)
+					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)
 					}
+				} else {
+					s.Warnf("Error applying consumer entries to '%s > %s'", ca.Client.serviceAccount(), ca.Name)
 				}
 			}
 			aq.recycle(&ces)
@@ -5573,20 +5573,19 @@ var (
 
 // blocking utility call to perform requests on the system account
 // returns (synchronized) v or error
-func sysRequest[T any](s *Server, subjFormat string, args ...interface{}) (*T, error) {
+func (s *Server) sysRequest(v interface{}, subjFormat string, args ...interface{}) (interface{}, error) {
 	isubj := fmt.Sprintf(subjFormat, args...)
-
 	s.mu.Lock()
 	inbox := s.newRespInbox()
-	results := make(chan *T, 1)
-	s.sys.replies[inbox] = func(_ *subscription, _ *client, _ *Account, _, _ string, msg []byte) {
-		var v T
-		if err := json.Unmarshal(msg, &v); err != nil {
+	results := make(chan interface{}, 1)
+	// Store our handler.
+	s.sys.replies[inbox] = func(sub *subscription, _ *client, _ *Account, subject, _ string, msg []byte) {
+		if err := json.Unmarshal(msg, v); err != nil {
 			s.Warnf("Error unmarshalling response for request '%s':%v", isubj, err)
 			return
 		}
 		select {
-		case results <- &v:
+		case results <- v:
 		default:
 			s.Warnf("Failed placing request response on internal channel")
 		}
@@ -5595,22 +5594,27 @@ func sysRequest[T any](s *Server, subjFormat string, args ...interface{}) (*T, e
 
 	s.sendInternalMsgLocked(isubj, inbox, nil, nil)
 
-	defer func() {
-		s.mu.Lock()
-		defer s.mu.Unlock()
-		if s.sys != nil && s.sys.replies != nil {
-			delete(s.sys.replies, inbox)
-		}
-	}()
+	const timeout = 2 * time.Second
+	notActive := time.NewTimer(timeout)
+	defer notActive.Stop()
+
+	var err error
+	var data interface{}
 
 	select {
 	case <-s.quitCh:
-		return nil, errReqSrvExit
-	case <-time.After(2 * time.Second):
-		return nil, errReqTimeout
-	case data := <-results:
-		return data, nil
+		err = errReqSrvExit
+	case <-notActive.C:
+		err = errReqTimeout
+	case data = <-results:
 	}
+	// Clean up here.
+	s.mu.Lock()
+	if s.sys != nil && s.sys.replies != nil {
+		delete(s.sys.replies, inbox)
+	}
+	s.mu.Unlock()
+	return data, err
 }
 
 func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, cfg *StreamConfig, peerSet []string) {
@@ -5704,9 +5708,9 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 		} else {
 			// Need to release js lock.
 			js.mu.Unlock()
-			if si, err := sysRequest[StreamInfo](s, clusterStreamInfoT, ci.serviceAccount(), cfg.Name); err != nil {
+			if si, err := s.sysRequest(&StreamInfo{}, clusterStreamInfoT, ci.serviceAccount(), cfg.Name); err != nil {
 				msg = fmt.Sprintf("error retrieving info: %s", err.Error())
-			} else if si != nil {
+			} else if si := si.(*StreamInfo); si != nil {
 				currentCount := 0
 				if si.Cluster.Leader != _EMPTY_ {
 					currentCount++
@@ -5761,12 +5765,10 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 			if !s.allPeersOffline(rg) {
 				// Need to release js lock.
 				js.mu.Unlock()
-				if si, err := sysRequest[StreamInfo](s, clusterStreamInfoT, ci.serviceAccount(), cfg.Name); err != nil {
+				if si, err := s.sysRequest(&StreamInfo{}, clusterStreamInfoT, ci.serviceAccount(), cfg.Name); err != nil {
 					s.Warnf("Did not receive stream info results for '%s > %s' due to: %s", acc, cfg.Name, err)
-				} else if si != nil {
-					if cl := si.Cluster; cl != nil && cl.Leader != _EMPTY_ {
-						curLeader = getHash(cl.Leader)
-					}
+				} else if cl := si.(*StreamInfo).Cluster; cl != nil && cl.Leader != _EMPTY_ {
+					curLeader = getHash(cl.Leader)
 				}
 				// Re-acquire here.
 				js.mu.Lock()
@@ -6707,12 +6709,10 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 			if !s.allPeersOffline(ca.Group) {
 				// Need to release js lock.
 				js.mu.Unlock()
-				if ci, err := sysRequest[ConsumerInfo](s, clusterConsumerInfoT, ci.serviceAccount(), sa.Config.Name, cfg.Durable); err != nil {
+				if ci, err := s.sysRequest(&ConsumerInfo{}, clusterConsumerInfoT, ci.serviceAccount(), sa.Config.Name, cfg.Durable); err != nil {
 					s.Warnf("Did not receive consumer info results for '%s > %s > %s' due to: %s", acc, sa.Config.Name, cfg.Durable, err)
-				} else if ci != nil {
-					if cl := ci.Cluster; cl != nil {
-						curLeader = getHash(cl.Leader)
-					}
+				} else if cl := ci.(*ConsumerInfo).Cluster; cl != nil {
+					curLeader = getHash(cl.Leader)
 				}
 				// Re-acquire here.
 				js.mu.Lock()
diff --git a/server/monitor.go b/server/monitor.go
index 4adbffb8..624091ce 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -3125,11 +3125,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()
-
-	// TODO(dlc) - Might be better here to not hold the lock the whole time.
-	js.mu.RLock()
-	defer js.mu.RUnlock()
-
 	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 b098ebe7..9bdbe5c3 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -2511,6 +2511,13 @@ 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.
@@ -3049,7 +3056,6 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	for _, e := range ae.entries {
 		switch e.Type {
 		case EntryLeaderTransfer:
-			// Only process these if they are new, so no replays or catchups.
 			if isNew {
 				maybeLeader := string(e.Data)
 				if maybeLeader == n.id && !n.observer && !n.paused {
@@ -3150,7 +3156,7 @@ func (n *raft) buildAppendEntry(entries []*Entry) *appendEntry {
 
 // Determine if we should store an entry.
 func (ae *appendEntry) shouldStore() bool {
-	return ae != nil && len(ae.entries) > 0
+	return ae != nil && len(ae.entries) > 0 && ae.entries[0].Type != EntryLeaderTransfer
 }
 
 // Store our append entry to our WAL.
diff --git a/server/server.go b/server/server.go
index 2e62a68e..71c1fbad 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1260,13 +1260,13 @@ func (s *Server) setSystemAccount(acc *Account) error {
 		servers: make(map[string]*serverUpdate),
 		replies: make(map[string]msgHandler),
 		sendq:   newIPQueue[*pubMsg](s, "System sendQ"),
-		recvq:   newIPQueue[*inSysMsg](s, "System recvQ"),
 		resetCh: make(chan struct{}),
 		sq:      s.newSendQ(),
 		statsz:  eventsHBInterval,
 		orphMax: 5 * eventsHBInterval,
 		chkOrph: 3 * eventsHBInterval,
 	}
+
 	s.sys.wg.Add(1)
 	s.mu.Unlock()
 
@@ -1279,9 +1279,6 @@ func (s *Server) setSystemAccount(acc *Account) error {
 	// We do our own wg here since we will stop first during shutdown.
 	go s.internalSendLoop(&s.sys.wg)
 
-	// Start the internal loop for inbound messages.
-	go s.internalReceiveLoop()
-
 	// Start up our general subscriptions
 	s.initEventTracking()
 
diff --git a/server/stream.go b/server/stream.go
index 5aaffff4..f9db14d1 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -3670,7 +3670,21 @@ func (mset *stream) getDirectRequest(req *JSApiMsgGetRequest, reply string) {
 // processInboundJetStreamMsg handles processing messages bound for a stream.
 func (mset *stream) processInboundJetStreamMsg(_ *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
 	hdr, msg := c.msgParts(rmsg)
-	mset.queueInboundMsg(subject, reply, hdr, msg)
+
+	// If we are not receiving directly from a client we should move this to another Go routine.
+	// Make sure to grab no stream or js locks.
+	if c.kind != CLIENT {
+		mset.queueInboundMsg(subject, reply, hdr, msg)
+		return
+	}
+
+	// This is directly from a client so process inline.
+	// If we are clustered we need to propose this message to the underlying raft group.
+	if mset.IsClustered() {
+		mset.processClusteredInboundMsg(subject, reply, hdr, msg)
+	} else {
+		mset.processJetStreamMsg(subject, reply, hdr, msg, 0, 0)
+	}
 }
 
 var (
@@ -4223,7 +4237,7 @@ func newJSPubMsg(dsubj, subj, reply string, hdr, msg []byte, o *consumer, seq ui
 	} else {
 		m = new(jsPubMsg)
 	}
-	// When getting something from a pool it is critical that all fields are
+	// When getting something from a pool it is criticical that all fields are
 	// initialized. Doing this way guarantees that if someone adds a field to
 	// the structure, the compiler will fail the build if this line is not updated.
 	(*m) = jsPubMsg{dsubj, reply, StoreMsg{subj, hdr, msg, buf, seq, 0}, o}


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index 28bc4c9c..70745e3f 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1,4 +1,4 @@
-// Copyright 2018-2023 The NATS Authors
+// Copyright 2018-2020 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
@@ -1220,7 +1220,6 @@ func TestAccountClaimsUpdates(t *testing.T) {
 		claimUpdateSubj := fmt.Sprintf(subj, pub)
 		nc.Publish(claimUpdateSubj, []byte(ajwt))
 		nc.Flush()
-		time.Sleep(200 * time.Millisecond)
 
 		acc, _ = s.LookupAccount(pub)
 		if acc.MaxActiveConnections() != 8 {
@@ -1341,9 +1340,6 @@ func TestAccountReqMonitoring(t *testing.T) {
 	require_NoError(t, nc.PublishRequest(pStatz, ib, nil))
 	resp, err = rSub.NextMsg(time.Second)
 	require_NoError(t, err)
-	// Since we now have processed our own message, msgs will be 1.
-	respContentAcc = []string{`"conns":1,`, `"total_conns":1`, `"slow_consumers":0`, `"sent":{"msgs":0,"bytes":0}`,
-		`"received":{"msgs":1,"bytes":0}`, fmt.Sprintf(`"acc":"%s"`, acc.Name)}
 	require_Contains(t, string(resp.Data), respContentAcc...)
 	_, err = rSub.NextMsg(200 * time.Millisecond)
 	require_Error(t, err)
@@ -1645,7 +1641,7 @@ func TestSystemAccountWithBadRemoteLatencyUpdate(t *testing.T) {
 		ReqId:   "_INBOX.22",
 	}
 	b, _ := json.Marshal(&rl)
-	s.remoteLatencyUpdate(nil, nil, nil, "foo", _EMPTY_, nil, b)
+	s.remoteLatencyUpdate(nil, nil, nil, "foo", _EMPTY_, b)
 }
 
 func TestSystemAccountWithGateways(t *testing.T) {
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 52e43d63..88f55b64 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -2242,15 +2242,14 @@ func TestJetStreamClusterMemLeaderRestart(t *testing.T) {
 
 	// Make sure that we have a META leader (there can always be a re-election)
 	c.waitOnLeader()
-	c.waitOnStreamLeader(globalAccountName, "foo")
 
 	// Should still have quorum and a new leader
-	checkFor(t, 5*time.Second, 200*time.Millisecond, func() error {
+	checkFor(t, time.Second, 200*time.Millisecond, func() error {
 		osi, err = jsc.StreamInfo("foo")
 		if err != nil {
 			return fmt.Errorf("expected healthy stream asset, got %s", err.Error())
 		}
-		if osi.Cluster.Leader == _EMPTY_ {
+		if osi.Cluster.Leader == "" {
 			return fmt.Errorf("expected healthy stream asset with new leader")
 		}
 		if osi.State.Msgs != uint64(toSend) {
diff --git a/server/norace_test.go b/server/norace_test.go
index deb5812c..06da28f8 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -6559,7 +6559,7 @@ func TestNoRaceJetStreamClusterF3Setup(t *testing.T) {
 	setHighStartSequence := false
 	simulateMaxRedeliveries := false
 	maxBadPubTimes := uint32(20)
-	badPubThresh := 500 * time.Millisecond
+	badPubThresh := 5 * time.Second
 	testTime := 5 * time.Minute // make sure to do --timeout=65m
 
 	t.Logf("Starting Test: Total Test Time %v", testTime)