Repository: nats-io/nats-server
Pull request #2775 (2022-01-10 22:13:58)

Replace use of go channels with queues


Patch:
diff --git a/server/consumer.go b/server/consumer.go
index f1675248..ecca76e4 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1421,7 +1421,9 @@ func (o *consumer) loopAndForwardProposals(qch chan struct{}) {
 			sz += len(proposal.data)
 			if sz > maxBatch {
 				node.ProposeDirect(entries)
-				sz, entries = 0, entries[:0]
+				// We need to re-craete `entries` because there is a reference
+				// to it in the node's pae map.
+				sz, entries = 0, nil
 			}
 		}
 		if len(entries) > 0 {
@@ -2110,7 +2112,7 @@ func (o *consumer) nextWaiting() *waitingRequest {
 			}
 		}
 		hdr := []byte("NATS/1.0 408 Request Timeout\r\n\r\n")
-		o.outq.send(&jsPubMsg{wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0, nil})
+		o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 		// Remove the current one, no longer valid.
 		o.waiting.removeCurrent()
 		wr.recycle()
@@ -2137,7 +2139,7 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 
 	sendErr := func(status int, description string) {
 		hdr := []byte(fmt.Sprintf("NATS/1.0 %d %s\r\n\r\n", status, description))
-		o.outq.send(&jsPubMsg{reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0, nil})
+		o.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 	}
 
 	if o.isPushMode() {
@@ -2325,7 +2327,7 @@ func (o *consumer) forceExpireFirstWaiting() {
 	if rr := wr.acc.sl.Match(wr.interest); len(rr.psubs)+len(rr.qsubs) > 0 && o.mset != nil {
 		// We still appear to have interest, so send alert as courtesy.
 		hdr := []byte("NATS/1.0 408 Request Timeout\r\n\r\n")
-		o.outq.send(&jsPubMsg{wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0, nil})
+		o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 	}
 	o.waiting.removeCurrent()
 	wr.recycle()
@@ -2343,7 +2345,7 @@ func (o *consumer) expireWaiting() int {
 			} else {
 				hdr = []byte("NATS/1.0 408 Request Timeout\r\n\r\n")
 			}
-			o.outq.send(&jsPubMsg{wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0, nil})
+			o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 			o.waiting.removeCurrent()
 			wr.recycle()
 			expired++
@@ -2543,7 +2545,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 					addOn := []byte(fmt.Sprintf("%s: %s\r\n\r\n", JSConsumerStalled, fcp))
 					hdr = append(hdr[:len(hdr)-LEN_CR_LF], []byte(addOn)...)
 				}
-				outq.send(&jsPubMsg{odsubj, _EMPTY_, _EMPTY_, hdr, nil, nil, 0, nil})
+				outq.send(newJSPubMsg(odsubj, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 			}
 			// Reset our idle heartbeat timer.
 			hb.Reset(hbd)
@@ -2616,7 +2618,7 @@ func (o *consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint6
 		msg = nil
 	}
 
-	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, dseq, dc, ts, o.adjustedPending()), hdr, msg, o, seq, nil}
+	pmsg := newJSPubMsg(dsubj, subj, o.ackReply(seq, dseq, dc, ts, o.adjustedPending()), hdr, msg, o, seq)
 	if o.maxpb > 0 {
 		o.pbytes += pmsg.size()
 	}
@@ -2724,7 +2726,7 @@ func (o *consumer) sendFlowControl() {
 	subj, rply := o.cfg.DeliverSubject, o.fcReply()
 	o.fcsz, o.fcid = o.pbytes, rply
 	hdr := []byte("NATS/1.0 100 FlowControl Request\r\n\r\n")
-	o.outq.send(&jsPubMsg{subj, _EMPTY_, rply, hdr, nil, nil, 0, nil})
+	o.outq.send(newJSPubMsg(subj, _EMPTY_, rply, hdr, nil, nil, 0))
 }
 
 // Tracks our outstanding pending acks. Only applicable to AckExplicit mode.
diff --git a/server/events.go b/server/events.go
index 8454a0b4..d097c47a 100644
--- a/server/events.go
+++ b/server/events.go
@@ -91,7 +91,7 @@ type internal struct {
 	sweeper  *time.Timer
 	stmr     *time.Timer
 	replies  map[string]msgHandler
-	sendq    chan *pubMsg
+	sendq    *ipQueue // of *pubMsg
 	resetCh  chan struct{}
 	wg       sync.WaitGroup
 	sq       *sendq
@@ -248,6 +248,33 @@ type pubMsg struct {
 	last bool
 }
 
+var pubMsgPool sync.Pool
+
+func newPubMsg(c *client, sub, rply string, si *ServerInfo, hdr map[string]string,
+	msg interface{}, oct compressionType, echo, last bool) *pubMsg {
+
+	var m *pubMsg
+	pm := pubMsgPool.Get()
+	if pm != nil {
+		m = pm.(*pubMsg)
+	} else {
+		m = &pubMsg{}
+	}
+	// 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) = pubMsg{c, sub, rply, si, hdr, msg, oct, echo, last}
+	return m
+}
+
+func (pm *pubMsg) returnToPool() {
+	if pm == nil {
+		return
+	}
+	pm.c, pm.sub, pm.rply, pm.si, pm.hdr, pm.msg = nil, _EMPTY_, _EMPTY_, nil, nil, nil
+	pubMsgPool.Put(pm)
+}
+
 // Used to track server updates.
 type serverUpdate struct {
 	seq   uint64
@@ -288,130 +315,125 @@ RESET:
 	}
 	s.mu.Unlock()
 
-	// Warn when internal send queue is backed up past 75%
-	warnThresh := 3 * internalSendQLen / 4
-	warnFreq := time.Second
-	last := time.Now().Add(-warnFreq)
-
 	for s.eventsRunning() {
-		// Setup information for next message
-		if len(sendq) > warnThresh && time.Since(last) >= warnFreq {
-			s.Warnf("Internal system send queue > 75%%")
-			last = time.Now()
-		}
-
 		select {
-		case pm := <-sendq:
-			if pm.si != nil {
-				pm.si.Name = servername
-				pm.si.Domain = domain
-				pm.si.Host = host
-				pm.si.Cluster = cluster
-				pm.si.ID = id
-				pm.si.Seq = atomic.AddUint64(seqp, 1)
-				pm.si.Version = VERSION
-				pm.si.Time = time.Now().UTC()
-				pm.si.JetStream = js
-			}
-			var b []byte
-			if pm.msg != nil {
-				switch v := pm.msg.(type) {
-				case string:
-					b = []byte(v)
-				case []byte:
-					b = v
-				default:
-					b, _ = json.Marshal(pm.msg)
+		case <-sendq.ch:
+			msgs := sendq.pop()
+			for _, pmi := range msgs {
+				pm := pmi.(*pubMsg)
+				if pm.si != nil {
+					pm.si.Name = servername
+					pm.si.Domain = domain
+					pm.si.Host = host
+					pm.si.Cluster = cluster
+					pm.si.ID = id
+					pm.si.Seq = atomic.AddUint64(seqp, 1)
+					pm.si.Version = VERSION
+					pm.si.Time = time.Now().UTC()
+					pm.si.JetStream = js
+				}
+				var b []byte
+				if pm.msg != nil {
+					switch v := pm.msg.(type) {
+					case string:
+						b = []byte(v)
+					case []byte:
+						b = v
+					default:
+						b, _ = json.Marshal(pm.msg)
+					}
 				}
-			}
 
-			// Setup our client. If the user wants to use a non-system account use our internal
-			// account scoped here so that we are not changing out accounts for the system client.
-			var c *client
-			if pm.c != nil {
-				c = pm.c
-			} else {
-				c = sysc
-			}
+				// Setup our client. If the user wants to use a non-system account use our internal
+				// account scoped here so that we are not changing out accounts for the system client.
+				var c *client
+				if pm.c != nil {
+					c = pm.c
+				} else {
+					c = sysc
+				}
 
-			// Grab client lock.
-			c.mu.Lock()
+				// Grab client lock.
+				c.mu.Lock()
 
-			// Prep internal structures needed to send message.
-			c.pa.subject, c.pa.reply = []byte(pm.sub), []byte(pm.rply)
-			c.pa.size, c.pa.szb = len(b), []byte(strconv.FormatInt(int64(len(b)), 10))
-			c.pa.hdr, c.pa.hdb = -1, nil
-			trace := c.trace
-
-			// Now check for optional compression.
-			var contentHeader string
-			var bb bytes.Buffer
-
-			if len(b) > 0 {
-				switch pm.oct {
-				case gzipCompression:
-					zw := gzip.NewWriter(&bb)
-					zw.Write(b)
-					zw.Close()
-					b = bb.Bytes()
-					contentHeader = "gzip"
-				case snappyCompression:
-					sw := s2.NewWriter(&bb, s2.WriterSnappyCompat())
-					sw.Write(b)
-					sw.Close()
-					b = bb.Bytes()
-					contentHeader = "snappy"
-				case unsupportedCompression:
-					contentHeader = "identity"
+				// Prep internal structures needed to send message.
+				c.pa.subject, c.pa.reply = []byte(pm.sub), []byte(pm.rply)
+				c.pa.size, c.pa.szb = len(b), []byte(strconv.FormatInt(int64(len(b)), 10))
+				c.pa.hdr, c.pa.hdb = -1, nil
+				trace := c.trace
+
+				// Now check for optional compression.
+				var contentHeader string
+				var bb bytes.Buffer
+
+				if len(b) > 0 {
+					switch pm.oct {
+					case gzipCompression:
+						zw := gzip.NewWriter(&bb)
+						zw.Write(b)
+						zw.Close()
+						b = bb.Bytes()
+						contentHeader = "gzip"
+					case snappyCompression:
+						sw := s2.NewWriter(&bb, s2.WriterSnappyCompat())
+						sw.Write(b)
+						sw.Close()
+						b = bb.Bytes()
+						contentHeader = "snappy"
+					case unsupportedCompression:
+						contentHeader = "identity"
+					}
 				}
-			}
-			// Optional Echo
-			replaceEcho := c.echo != pm.echo
-			if replaceEcho {
-				c.echo = !c.echo
-			}
-			c.mu.Unlock()
+				// Optional Echo
+				replaceEcho := c.echo != pm.echo
+				if replaceEcho {
+					c.echo = !c.echo
+				}
+				c.mu.Unlock()
 
-			// Add in NL
-			b = append(b, _CRLF_...)
+				// Add in NL
+				b = append(b, _CRLF_...)
 
-			// Check if we should set content-encoding
-			if contentHeader != _EMPTY_ {
-				b = c.setHeader(contentEncodingHeader, contentHeader, b)
-			}
+				// Check if we should set content-encoding
+				if contentHeader != _EMPTY_ {
+					b = c.setHeader(contentEncodingHeader, contentHeader, b)
+				}
 
-			// Optional header processing.
-			if pm.hdr != nil {
-				for k, v := range pm.hdr {
-					b = c.setHeader(k, v, b)
+				// Optional header processing.
+				if pm.hdr != nil {
+					for k, v := range pm.hdr {
+						b = c.setHeader(k, v, b)
+					}
+				}
+				// Tracing
+				if trace {
+					c.traceInOp(fmt.Sprintf("PUB %s %s %d", c.pa.subject, c.pa.reply, c.pa.size), nil)
+					c.traceMsg(b)
 				}
-			}
-			// Tracing
-			if trace {
-				c.traceInOp(fmt.Sprintf("PUB %s %s %d", c.pa.subject, c.pa.reply, c.pa.size), nil)
-				c.traceMsg(b)
-			}
 
-			// Process like a normal inbound msg.
-			c.processInboundClientMsg(b)
+				// Process like a normal inbound msg.
+				c.processInboundClientMsg(b)
 
-			// Put echo back if needed.
-			if replaceEcho {
-				c.mu.Lock()
-				c.echo = !c.echo
-				c.mu.Unlock()
-			}
+				// Put echo back if needed.
+				if replaceEcho {
+					c.mu.Lock()
+					c.echo = !c.echo
+					c.mu.Unlock()
+				}
 
-			// See if we are doing graceful shutdown.
-			if !pm.last {
-				c.flushClients(0) // Never spend time in place.
-			} else {
-				// For the Shutdown event, we need to send in place otherwise
-				// there is a chance that the process will exit before the
-				// writeLoop has a chance to send it.
-				c.flushClients(time.Second)
-				return
+				// See if we are doing graceful shutdown.
+				if !pm.last {
+					c.flushClients(0) // Never spend time in place.
+				} else {
+					// For the Shutdown event, we need to send in place otherwise
+					// there is a chance that the process will exit before the
+					// writeLoop has a chance to send it.
+					c.flushClients(time.Second)
+					return
+				}
+				pm.returnToPool()
 			}
+			sendq.recycle(&msgs)
 		case <-resetCh:
 			goto RESET
 		case <-s.quitCh:
@@ -433,10 +455,10 @@ func (s *Server) sendShutdownEvent() {
 	s.sys.sendq = nil
 	// Unhook all msgHandlers. Normal client cleanup will deal with subs, etc.
 	s.sys.replies = nil
-	s.mu.Unlock()
 	// Send to the internal queue and mark as last.
 	si := &ServerInfo{}
-	sendq <- &pubMsg{nil, subj, _EMPTY_, si, nil, si, noCompression, false, true}
+	sendq.push(newPubMsg(nil, subj, _EMPTY_, si, nil, si, noCompression, false, true))
+	s.mu.Unlock()
 }
 
 // Used to send an internal message to an arbitrary account.
@@ -451,19 +473,15 @@ func (s *Server) sendInternalAccountMsgWithReply(a *Account, subject, reply stri
 		s.mu.Unlock()
 		return ErrNoSysAccount
 	}
-	sendq := s.sys.sendq
-	// Don't hold lock while placing on the channel.
 	c := s.sys.client
-	s.mu.Unlock()
-
 	// Replace our client with the account's internal client.
 	if a != nil {
 		a.mu.Lock()
 		c = a.internalClient()
 		a.mu.Unlock()
 	}
-
-	sendq <- &pubMsg{c, subject, reply, nil, hdr, msg, noCompression, echo, false}
+	s.sys.sendq.push(newPubMsg(c, subject, reply, nil, hdr, msg, noCompression, echo, false))
+	s.mu.Unlock()
 	return nil
 }
 
@@ -481,11 +499,7 @@ func (s *Server) sendInternalMsg(subj, rply string, si *ServerInfo, msg interfac
 	if s.sys == nil || s.sys.sendq == nil {
 		return
 	}
-	sendq := s.sys.sendq
-	// Don't hold lock while placing on the channel.
-	s.mu.Unlock()
-	sendq <- &pubMsg{nil, subj, rply, si, nil, msg, noCompression, false, false}
-	s.mu.Lock()
+	s.sys.sendq.push(newPubMsg(nil, subj, rply, si, nil, msg, noCompression, false, false))
 }
 
 // Will send an api response.
@@ -495,10 +509,8 @@ func (s *Server) sendInternalResponse(subj string, response *ServerAPIResponse)
 		s.mu.Unlock()
 		return
 	}
-	sendq := s.sys.sendq
-	// Don't hold lock while placing on the channel.
+	s.sys.sendq.push(newPubMsg(nil, subj, _EMPTY_, response.Server, nil, response, response.compress, false, false))
 	s.mu.Unlock()
-	sendq <- &pubMsg{nil, subj, _EMPTY_, response.Server, nil, response, response.compress, false, false}
 }
 
 // Used to send internal messages from other system clients to avoid no echo issues.
@@ -512,13 +524,11 @@ func (c *client) sendInternalMsg(subj, rply string, si *ServerInfo, msg interfac
 	}
 	s.mu.Lock()
 	if s.sys == nil || s.sys.sendq == nil {
+		s.mu.Unlock()
 		return
 	}
-	sendq := s.sys.sendq
-	// Don't hold lock while placing on the channel.
+	s.sys.sendq.push(newPubMsg(c, subj, rply, si, nil, msg, noCompression, false, false))
 	s.mu.Unlock()
-
-	sendq <- &pubMsg{c, subj, rply, si, nil, msg, noCompression, false, false}
 }
 
 // Locked version of checking if events system running. Also checks server.
@@ -1622,7 +1632,6 @@ func (s *Server) sendAccConnsUpdate(a *Account, subj ...string) {
 	// Build event with account name and number of local clients and leafnodes.
 	eid := s.nextEventID()
 	a.mu.Lock()
-	s.mu.Unlock()
 	localConns := a.numLocalConnections()
 	m := &AccountNumConns{
 		TypedEvent: TypedEvent{
@@ -1647,17 +1656,10 @@ func (s *Server) sendAccConnsUpdate(a *Account, subj ...string) {
 		}
 	}
 	for _, sub := range subj {
-		msg := &pubMsg{nil, sub, _EMPTY_, &m.Server, nil, &m, noCompression, false, false}
-		select {
-		case sendQ <- msg:
-		default:
-			a.mu.Unlock()
-			sendQ <- msg
-			a.mu.Lock()
-		}
+		msg := newPubMsg(nil, sub, _EMPTY_, &m.Server, nil, &m, noCompression, false, false)
+		sendQ.push(msg)
 	}
 	a.mu.Unlock()
-	s.mu.Lock()
 }
 
 // accConnsUpdate is called whenever there is a change to the account's
diff --git a/server/ipqueue.go b/server/ipqueue.go
new file mode 100644
index 00000000..f496e24d
--- /dev/null
+++ b/server/ipqueue.go
@@ -0,0 +1,214 @@
+// Copyright 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
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"sync"
+)
+
+const ipQueueDefaultMaxRecycleSize = 4 * 1024
+const ipQueueDefaultWarnThreshold = 32 * 1024
+
+type ipQueueLogger interface {
+	// The ipQueue will invoke this function with the queue's name and the number
+	// of pending elements. This call CANNOT block. It is ok to drop the logging
+	// if desired, but not block.
+	log(name string, pending int)
+}
+
+// This is a generic intra-process queue.
+type ipQueue struct {
+	sync.RWMutex
+	ch     chan struct{}
+	elts   []interface{}
+	pos    int
+	pool   *sync.Pool
+	mrs    int
+	name   string
+	logger ipQueueLogger
+	lt     int
+}
+
+type ipQueueOpts struct {
+	maxRecycleSize int
+	name           string
+	logger         ipQueueLogger
+}
+
+type ipQueueOpt func(*ipQueueOpts)
+
+// This option allows to set the maximum recycle size when attempting
+// to put back a slice to the pool.
+func ipQueue_MaxRecycleSize(max int) ipQueueOpt {
+	return func(o *ipQueueOpts) {
+		o.maxRecycleSize = max
+	}
+}
+
+// This option provides the logger to be used by this queue to log
+// when the number of pending elements reaches a certain threshold.
+func ipQueue_Logger(name string, l ipQueueLogger) ipQueueOpt {
+	return func(o *ipQueueOpts) {
+		o.name, o.logger = name, l
+	}
+}
+
+func newIPQueue(opts ...ipQueueOpt) *ipQueue {
+	qo := ipQueueOpts{maxRecycleSize: ipQueueDefaultMaxRecycleSize}
+	for _, o := range opts {
+		o(&qo)
+	}
+	q := &ipQueue{
+		ch:     make(chan struct{}, 1),
+		mrs:    qo.maxRecycleSize,
+		pool:   &sync.Pool{},
+		name:   qo.name,
+		logger: qo.logger,
+		lt:     ipQueueDefaultWarnThreshold,
+	}
+	return q
+}
+
+// Add the element `e` to the queue, notifying the queue channel's `ch` if the
+// entry is the first to be added, and returns the length of the queue after
+// this element is added.
+func (q *ipQueue) push(e interface{}) int {
+	var signal bool
+	q.Lock()
+	l := len(q.elts) - q.pos
+	if l == 0 {
+		signal = true
+		eltsi := q.pool.Get()
+		if eltsi != nil {
+			// Reason we use pointer to slice instead of slice is explained
+			// here: https://staticcheck.io/docs/checks#SA6002
+			q.elts = (*(eltsi.(*[]interface{})))[:0]
+		}
+		if cap(q.elts) == 0 {
+			q.elts = make([]interface{}, 0, 32)
+		}
+	}
+	q.elts = append(q.elts, e)
+	l++
+	if l >= q.lt && q.logger != nil {
+		q.logger.log(q.name, l)
+	}
+	q.Unlock()
+	if signal {
+		select {
+		case q.ch <- struct{}{}:
+		default:
+		}
+	}
+	return l
+}
+
+// Returns the whole list of elements currently present in the queue,
+// emptying the queue. This should be called after receiving a notification
+// from the queue's `ch` notification channel that indicates that there
+// is something in the queue.
+// However, in cases where `drain()` may be called from another go
+// routine, it is possible that a routine is notified that there is
+// something, but by the time it calls `pop()`, the drain() would have
+// emptied the queue. So the caller should never assume that pop() will
+// return a slice of 1 or more, it could return `nil`.
+func (q *ipQueue) pop() []interface{} {
+	var elts []interface{}
+	q.Lock()
+	if q.pos == 0 {
+		elts = q.elts
+	} else {
+		elts = q.elts[q.pos:]
+	}
+	q.elts, q.pos = nil, 0
+	q.Unlock()
+	return elts
+}
+
+func (q *ipQueue) resetAndReturnToPool(elts *[]interface{}) {
+	for i, l := 0, len(*elts); i < l; i++ {
+		(*elts)[i] = nil
+	}
+	q.pool.Put(elts)
+}
+
+// Returns the first element from the queue, if any. See comment above
+// regarding calling after being notified that there is something and
+// the use of drain(). In short, the caller should always expect that
+// pop() or popOne() may return `nil`.
+func (q *ipQueue) popOne() interface{} {
+	q.Lock()
+	l := len(q.elts) - q.pos
+	if l < 1 {
+		q.Unlock()
+		return nil
+	}
+	e := q.elts[q.pos]
+	q.pos++
+	l--
+	if l > 0 {
+		// We need to re-signal
+		select {
+		case q.ch <- struct{}{}:
+		default:
+		}
+	} else {
+		// We have just emptied the queue, so we can recycle now.
+		q.resetAndReturnToPool(&q.elts)
+		q.elts, q.pos = nil, 0
+	}
+	q.Unlock()
+	return e
+}
+
+// After a pop(), the slice can be recycled for the next push() when
+// a first element is added to the queue.
+// Reason we use pointer to slice instead of slice is explained
+// here: https://staticcheck.io/docs/checks#SA6002
+func (q *ipQueue) recycle(elts *[]interface{}) {
+	// If invoked with an nil list, don't recyle.
+	// We also don't want to recycle huge slices, so check against the max.
+	// q.mrs is normally immutable but can be changed, in a safe way, in some tests.
+	if elts == nil || *elts == nil || cap(*elts) > q.mrs {
+		return
+	}
+	q.resetAndReturnToPool(elts)
+}
+
+// Returns the current length of the queue.
+func (q *ipQueue) len() int {
+	q.RLock()
+	l := len(q.elts) - q.pos
+	q.RUnlock()
+	return l
+}
+
+// Empty the queue and consumes the notification signal if present.
+// Note that this could cause a reader go routine that has been
+// notified that there is something in the queue (reading from queue's `ch`)
+// may then get nothing if `drain()` is invoked before the `pop()` or `popOne()`.
+func (q *ipQueue) drain() {
+	q.Lock()
+	if q.elts != nil {
+		q.resetAndReturnToPool(&q.elts)
+		q.elts, q.pos = nil, 0
+	}
+	// Consume the signal if it was present to reduce the chance of a reader
+	// routine to be think that there is something in the queue...
+	select {
+	case <-q.ch:
+	default:
+	}
+	q.Unlock()
+}
diff --git a/server/jetstream.go b/server/jetstream.go
index 6b62a5f1..ce032c38 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -131,7 +131,7 @@ type jsAccount struct {
 	updatesPub string
 	updatesSub *subscription
 	// From server
-	sendq   chan *pubMsg
+	sendq   *ipQueue // of *pubMsg
 	lupdate time.Time
 	utimer  *time.Timer
 }
@@ -1568,7 +1568,7 @@ func (jsa *jsAccount) sendClusterUsageUpdate() {
 	le.PutUint64(b[16:], uint64(jsa.usage.api))
 	le.PutUint64(b[24:], uint64(jsa.usage.err))
 
-	jsa.sendq <- &pubMsg{nil, jsa.updatesPub, _EMPTY_, nil, nil, b, noCompression, false, false}
+	jsa.sendq.push(newPubMsg(nil, jsa.updatesPub, _EMPTY_, nil, nil, b, noCompression, false, false))
 }
 
 func (js *jetStream) wouldExceedLimits(storeType StorageType, sz int) bool {
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 6d8e7c05..78e17fa5 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -2724,7 +2724,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 
 	// For signaling to upper layers.
 	resultCh := make(chan result, 1)
-	activeCh := make(chan int, 32)
+	activeQ := newIPQueue() // of int
 
 	var total int
 
@@ -2776,7 +2776,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 			return
 		}
 
-		activeCh <- len(msg)
+		activeQ.push(len(msg))
 
 		s.sendInternalAccountMsg(acc, reply, nil)
 	}
@@ -2862,7 +2862,8 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 				// Signal to the upper layers.
 				doneCh <- err
 				return
-			case n := <-activeCh:
+			case <-activeQ.ch:
+				n := activeQ.popOne().(int)
 				total += n
 				notActive.Reset(activityInterval)
 			case <-notActive.C:
@@ -3056,7 +3057,7 @@ func (s *Server) streamSnapshot(ci *ClientInfo, acc *Account, mset *stream, sr *
 		chunk = chunk[:n]
 		if err != nil {
 			if n > 0 {
-				mset.outq.send(&jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, chunk, nil, 0, nil})
+				mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, chunk, nil, 0))
 			}
 			break
 		}
@@ -3072,13 +3073,13 @@ func (s *Server) streamSnapshot(ci *ClientInfo, acc *Account, mset *stream, sr *
 			}
 		}
 		ackReply := fmt.Sprintf("%s.%d.%d", ackSubj, len(chunk), index)
-		mset.outq.send(&jsPubMsg{reply, _EMPTY_, ackReply, nil, chunk, nil, 0, nil})
+		mset.outq.send(newJSPubMsg(reply, _EMPTY_, ackReply, nil, chunk, nil, 0))
 		atomic.AddInt32(&out, int32(len(chunk)))
 	}
 done:
 	// Send last EOF
 	// TODO(dlc) - place hash in header
-	mset.outq.send(&jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, nil, nil, 0, nil})
+	mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, nil, nil, 0))
 }
 
 // Request to create a durable consumer.
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 1616d1fb..bb345d39 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -770,7 +770,7 @@ func (cc *jetStreamCluster) isConsumerLeader(account, stream, consumer string) b
 
 func (js *jetStream) monitorCluster() {
 	s, n := js.server(), js.getMetaGroup()
-	qch, lch, ach := n.QuitC(), n.LeadChangeC(), n.ApplyC()
+	qch, lch, aq := n.QuitC(), n.LeadChangeC(), n.ApplyQ()
 
 	defer s.grWG.Done()
 
@@ -817,24 +817,29 @@ func (js *jetStream) monitorCluster() {
 			return
 		case <-qch:
 			return
-		case ce := <-ach:
-			if ce == nil {
-				// Signals we have replayed all of our metadata.
-				isRecovering = false
-				s.Debugf("Recovered JetStream cluster metadata")
-				continue
-			}
-			// FIXME(dlc) - Deal with errors.
-			if didSnap, didRemoval, err := js.applyMetaEntries(ce.Entries, isRecovering); err == nil {
-				_, nb := n.Applied(ce.Index)
-				if js.hasPeerEntries(ce.Entries) || didSnap || (didRemoval && time.Since(lastSnapTime) > 2*time.Second) {
-					// Since we received one make sure we have our own since we do not store
-					// our meta state outside of raft.
-					doSnapshot()
-				} else if lls := len(lastSnap); nb > uint64(lls*8) && lls > 0 {
-					doSnapshot()
+		case <-aq.ch:
+			ces := aq.pop()
+			for _, cei := range ces {
+				if cei == nil {
+					// Signals we have replayed all of our metadata.
+					isRecovering = false
+					s.Debugf("Recovered JetStream cluster metadata")
+					continue
+				}
+				ce := cei.(*CommittedEntry)
+				// FIXME(dlc) - Deal with errors.
+				if didSnap, didRemoval, err := js.applyMetaEntries(ce.Entries, isRecovering); err == nil {
+					_, nb := n.Applied(ce.Index)
+					if js.hasPeerEntries(ce.Entries) || didSnap || (didRemoval && time.Since(lastSnapTime) > 2*time.Second) {
+						// Since we received one make sure we have our own since we do not store
+						// our meta state outside of raft.
+						doSnapshot()
+					} else if lls := len(lastSnap); nb > uint64(lls*8) && lls > 0 {
+						doSnapshot()
+					}
 				}
 			}
+			aq.recycle(&ces)
 		case isLeader = <-lch:
 			// We want to make sure we are updated on statsz so ping the extended cluster.
 			if isLeader {
@@ -1461,7 +1466,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 		return
 	}
 
-	qch, lch, ach := n.QuitC(), n.LeadChangeC(), n.ApplyC()
+	qch, lch, aq := n.QuitC(), n.LeadChangeC(), n.ApplyQ()
 
 	s.Debugf("Starting stream monitor for '%s > %s'", sa.Client.serviceAccount(), sa.Config.Name)
 	defer s.Debugf("Exiting stream monitor for '%s > %s'", sa.Client.serviceAccount(), sa.Config.Name)
@@ -1474,14 +1479,8 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 		if n.Leader() {
 			n.StepDown()
 		}
-		// Drain the commit channel..
-		for len(ach) > 0 {
-			select {
-			case <-ach:
-			default:
-				return
-			}
-		}
+		// Drain the commit queue...
+		aq.drain()
 	}()
 
 	const (
@@ -1529,39 +1528,45 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 			return
 		case <-qch:
 			return
-		case ce := <-ach:
-			// No special processing needed for when we are caught up on restart.
-			if ce == nil {
-				isRecovering = false
-				// Check on startup if we should snapshot/compact.
-				if _, b := n.Size(); b > compactSizeMin || n.NeedSnapshot() {
-					doSnapshot()
-				}
-				continue
-			}
-			// Apply our entries.
-			if err := js.applyStreamEntries(mset, ce, isRecovering); err == nil {
-				ne, nb := n.Applied(ce.Index)
-				// If we have at least min entries to compact, go ahead and snapshot/compact.
-				if ne >= compactNumMin || nb > compactSizeMin {
-					doSnapshot()
+		case <-aq.ch:
+			ces := aq.pop()
+			for _, cei := range ces {
+				// No special processing needed for when we are caught up on restart.
+				if cei == nil {
+					isRecovering = false
+					// Check on startup if we should snapshot/compact.
+					if _, b := n.Size(); b > compactSizeMin || n.NeedSnapshot() {
+						doSnapshot()
+					}
+					continue
 				}
-			} else {
-				s.Warnf("Error applying entries to '%s > %s': %v", sa.Client.serviceAccount(), sa.Config.Name, err)
-				if isClusterResetErr(err) {
-					if mset.isMirror() && mset.IsLeader() {
-						mset.retryMirrorConsumer()
-						continue
+				ce := cei.(*CommittedEntry)
+				// Apply our entries.
+				if err := js.applyStreamEntries(mset, ce, isRecovering); err == nil {
+					ne, nb := n.Applied(ce.Index)
+					// If we have at least min entries to compact, go ahead and snapshot/compact.
+					if ne >= compactNumMin || nb > compactSizeMin {
+						doSnapshot()
 					}
-					// We will attempt to reset our cluster state.
-					if mset.resetClusteredState(err) {
-						return
+				} else {
+					s.Warnf("Error applying entries to '%s > %s': %v", sa.Client.serviceAccount(), sa.Config.Name, err)
+					if isClusterResetErr(err) {
+						if mset.isMirror() && mset.IsLeader() {
+							mset.retryMirrorConsumer()
+							continue
+						}
+						// We will attempt to reset our cluster state.
+						if mset.resetClusteredState(err) {
+							aq.recycle(&ces)
+							return
+						}
+					} else if isOutOfSpaceErr(err) {
+						// If applicable this will tear all of this down, but don't assume so and return.
+						s.handleOutOfSpace(mset)
 					}
-				} else if isOutOfSpaceErr(err) {
-					// If applicable this will tear all of this down, but don't assume so and return.
-					s.handleOutOfSpace(mset)
 				}
 			}
+			aq.recycle(&ces)
 		case isLeader = <-lch:
 			if isLeader {
 				if isRestore {
@@ -2819,7 +2824,7 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 		} else if err == errNoInterest {
 			// This is a stranded ephemeral, let's clean this one up.
 			subject := fmt.Sprintf(JSApiConsumerDeleteT, ca.Stream, ca.Name)
-			mset.outq.send(&jsPubMsg{subject, _EMPTY_, _EMPTY_, nil, nil, nil, 0, nil})
+			mset.outq.send(newJSPubMsg(subject, _EMPTY_, _EMPTY_, nil, nil, nil, 0))
 		}
 		js.mu.Unlock()
 
@@ -2994,7 +2999,7 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 		return
 	}
 
-	qch, lch, ach := n.QuitC(), n.LeadChangeC(), n.ApplyC()
+	qch, lch, aq := n.QuitC(), n.LeadChangeC(), n.ApplyQ()
 
 	s.Debugf("Starting consumer monitor for '%s > %s > %s", o.acc.Name, ca.Stream, ca.Name)
 	defer s.Debugf("Exiting consumer monitor for '%s > %s > %s'", o.acc.Name, ca.Stream, ca.Name)
@@ -3036,23 +3041,28 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 			return
 		case <-qch:
 			return
-		case ce := <-ach:
-			// No special processing needed for when we are caught up on restart.
-			if ce == nil {
-				if n.NeedSnapshot() {
-					doSnapshot()
+		case <-aq.ch:
+			ces := aq.pop()
+			for _, cei := range ces {
+				// No special processing needed for when we are caught up on restart.
+				if cei == nil {
+					if n.NeedSnapshot() {
+						doSnapshot()
+					}
+					continue
 				}
-				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()
+				ce := cei.(*CommittedEntry)
+				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()
+					}
+				} else {
+					s.Warnf("Error applying consumer entries to '%s > %s'", ca.Client.serviceAccount(), ca.Name)
 				}
-			} else {
-				s.Warnf("Error applying consumer entries to '%s > %s'", ca.Client.serviceAccount(), ca.Name)
 			}
+			aq.recycle(&ces)
 		case isLeader = <-lch:
 			if !isLeader && n.GroupLeader() != noLeader {
 				js.setConsumerAssignmentRecovering(ca)
@@ -4770,7 +4780,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 		s.resourcesExeededError()
 		if canRespond {
 			b, _ := json.Marshal(&JSPubAckResponse{PubAck: &PubAck{Stream: name}, Error: NewJSInsufficientResourcesError()})
-			outq.send(&jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0, nil})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, b, nil, 0))
 		}
 		// Stepdown regardless.
 		if node := mset.raftNode(); node != nil {
@@ -4803,7 +4813,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
 			resp.Error = NewJSAccountResourcesExceededError()
 			response, _ = json.Marshal(resp)
-			outq.send(&jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, response, nil, 0, nil})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, response, nil, 0))
 		}
 		return err
 	}
@@ -4816,7 +4826,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
 			resp.Error = NewJSStreamMessageExceedsMaximumError()
 			response, _ = json.Marshal(resp)
-			outq.send(&jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, response, nil, 0, nil})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, response, nil, 0))
 		}
 		return err
 	}
@@ -4830,7 +4840,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
 			resp.Error = NewJSStreamHeaderExceedsMaximumError()
 			response, _ = json.Marshal(resp)
-			outq.send(&jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, response, nil, 0, nil})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, response, nil, 0))
 		}
 		return err
 	}
@@ -4860,7 +4870,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 			resp.Error = &ApiError{Code: 503, Description: err.Error()}
 			response, _ = json.Marshal(resp)
 			// If we errored out respond here.
-			outq.send(&jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, response, nil, 0, nil})
+			outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, nil, response, nil, 0))
 		}
 	}
 
@@ -4990,6 +5000,7 @@ func (mset *stream) processSnapshot(snap *streamSnapshot) error {
 	mset.store.FastState(&state)
 	sreq := mset.calculateSyncRequest(&state, snap)
 	s, js, subject, n := mset.srv, mset.js, mset.sa.Sync, mset.node
+	qname := fmt.Sprintf("Stream %q snapshot", mset.cfg.Name)
 	mset.mu.Unlock()
 
 	// Make sure our state's first sequence is <= the leader's snapshot.
@@ -5064,21 +5075,14 @@ RETRY:
 		reply string
 	}
 
-	sz := int(sreq.LastSeq-sreq.FirstSeq) + 1
-	msgsC := make(chan *im, sz)
+	msgsQ := newIPQueue(ipQueue_Logger(qname, s.ipqLog)) // of *im
 
 	// Send our catchup request here.
 	reply := syncReplySubject()
 	sub, err = s.sysSubscribe(reply, func(_ *subscription, _ *client, _ *Account, _, reply string, msg []byte) {
 		// Make copies
 		// TODO(dlc) - Since we are using a buffer from the inbound client/route.
-		select {
-		case msgsC <- &im{copyBytes(msg), reply}:
-		default:
-			s.Warnf("Failed to place catchup message onto internal channel: %d pending", len(msgsC))
-			return
-		}
-
+		msgsQ.push(&im{copyBytes(msg), reply})
 	})
 	if err != nil {
 		s.Errorf("Could not subscribe to stream catchup: %v", err)
@@ -5095,34 +5099,40 @@ RETRY:
 	// Run our own select loop here.
 	for qch, lch := n.QuitC(), n.LeadChangeC(); ; {
 		select {
-		case mrec := <-msgsC:
+		case <-msgsQ.ch:
 			notActive.Reset(activityInterval)
-			msg := mrec.msg
 
-			// Check for eof signaling.
-			if len(msg) == 0 {
-				return nil
-			}
-			if lseq, err := mset.processCatchupMsg(msg); err == nil {
-				if lseq >= last {
+			mrecs := msgsQ.pop()
+			for _, mreci := range mrecs {
+				mrec := mreci.(*im)
+				msg := mrec.msg
+
+				// Check for eof signaling.
+				if len(msg) == 0 {
 					return nil
 				}
-			} else if isOutOfSpaceErr(err) {
-				return err
-			} else if err == NewJSInsufficientResourcesError() {
-				if mset.js.limitsExceeded(mset.cfg.Storage) {
-					s.resourcesExeededError()
+				if lseq, err := mset.processCatchupMsg(msg); err == nil {
+					if lseq >= last {
+						return nil
+					}
+				} else if isOutOfSpaceErr(err) {
+					return err
+				} else if err == NewJSInsufficientResourcesError() {
+					if mset.js.limitsExceeded(mset.cfg.Storage) {
+						s.resourcesExeededError()
+					} else {
+						s.Warnf("Catchup for stream '%s > %s' errored, account resources exceeded: %v", mset.account(), mset.name(), err)
+					}
+					return err
 				} else {
-					s.Warnf("Catchup for stream '%s > %s' errored, account resources exceeded: %v", mset.account(), mset.name(), err)
+					s.Warnf("Catchup for stream '%s > %s' errored, will retry: %v", mset.account(), mset.name(), err)
+					goto RETRY
+				}
+				if mrec.reply != _EMPTY_ {
+					s.sendInternalMsgLocked(mrec.reply, _EMPTY_, nil, nil)
 				}
-				return err
-			} else {
-				s.Warnf("Catchup for stream '%s > %s' errored, will retry: %v", mset.account(), mset.name(), err)
-				goto RETRY
-			}
-			if mrec.reply != _EMPTY_ {
-				s.sendInternalMsgLocked(mrec.reply, _EMPTY_, nil, nil)
 			}
+			msgsQ.recycle(&mrecs)
 		case <-notActive.C:
 			s.Warnf("Catchup for stream '%s > %s' stalled", mset.account(), mset.name())
 			notActive.Reset(activityInterval)
diff --git a/server/mqtt.go b/server/mqtt.go
index 63118893..ce1cec9a 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -229,27 +229,15 @@ type mqttAccountSessionManager struct {
 	replicas   int
 	rrmLastSeq uint64        // Restore retained messages expected last sequence
 	rrmDoneCh  chan struct{} // To notify the caller that all retained messages have been loaded
-	sp         sessPersist   // Used for cluster-wide processing of session records being persisted
+	sp         *ipQueue      // of uint64. Used for cluster-wide processing of session records being persisted
 	domainTk   string        // Domain (with trailing "."), or possibly empty. This is added to session subject.
 }
 
-type sessPersist struct {
-	mu   sync.Mutex
-	ch   chan struct{}
-	head *sessPersistRecord
-	tail *sessPersistRecord
-}
-
-type sessPersistRecord struct {
-	seq  uint64
-	next *sessPersistRecord
-}
-
 type mqttJSA struct {
 	mu        sync.Mutex
 	id        string
 	c         *client
-	sendq     chan *mqttJSPubMsg
+	sendq     *ipQueue // of *mqttJSPubMsg
 	rplyr     string
 	replies   sync.Map
 	nuid      *nuid.NUID
@@ -933,6 +921,7 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 
 	id := string(getHash(s.Name()))
 	replicas := s.mqttDetermineReplicas()
+	qname := fmt.Sprintf("MQTT account %q send", accName)
 	as := &mqttAccountSessionManager{
 		sessions:   make(map[string]*mqttSession),
 		sessByHash: make(map[string]*mqttSession),
@@ -943,13 +932,11 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 			id:     id,
 			c:      c,
 			rplyr:  mqttJSARepliesPrefix + id + ".",
-			sendq:  make(chan *mqttJSPubMsg, 8192),
+			sendq:  newIPQueue(ipQueue_Logger(qname, s.ipqLog)), // of *mqttJSPubMsg
 			nuid:   nuid.New(),
 			quitCh: quitCh,
 		},
-		sp: sessPersist{
-			ch: make(chan struct{}, 1),
-		},
+		sp: newIPQueue(), // of uint64
 	}
 	// TODO record domain name in as here
 
@@ -1213,12 +1200,12 @@ func (jsa *mqttJSA) newRequestEx(kind, subject string, hdr int, msg []byte, time
 	jsa.replies.Store(reply, ch)
 
 	subject = jsa.prefixDomain(subject)
-	jsa.sendq <- &mqttJSPubMsg{
+	jsa.sendq.push(&mqttJSPubMsg{
 		subj:  subject,
 		reply: reply,
 		hdr:   hdr,
 		msg:   msg,
-	}
+	})
 
 	var i interface{}
 	// We don't want to use time.After() which causes memory growth because the timer
@@ -1343,10 +1330,10 @@ func (jsa *mqttJSA) deleteMsg(stream string, seq uint64, wait bool) error {
 	req, _ := json.Marshal(dreq)
 	subj := jsa.prefixDomain(fmt.Sprintf(JSApiMsgDeleteT, stream))
 	if !wait {
-		jsa.sendq <- &mqttJSPubMsg{
+		jsa.sendq.push(&mqttJSPubMsg{
 			subj: subj,
 			msg:  req,
-		}
+		})
 		return nil
 	}
 	dmi, err := jsa.newRequest(mqttJSAMsgDelete, subj, 0, req)
@@ -1516,27 +1503,11 @@ func (as *mqttAccountSessionManager) processSessionPersist(_ *subscription, pc *
 	// We would need to lookup the message that that is a request/reply
 	// that we can do in place here. So move that to a long-running routine
 	// that will process the session persist record.
-	as.mu.RLock()
-	sp := &as.sp
-	as.mu.RUnlock()
-
-	spr := &sessPersistRecord{seq: par.Sequence}
-	sp.mu.Lock()
-	if sp.tail != nil {
-		sp.tail.next = spr
-	} else {
-		sp.head = spr
-		select {
-		case sp.ch <- struct{}{}:
-		default:
-		}
-	}
-	sp.tail = spr
-	sp.mu.Unlock()
+	as.sp.push(par.Sequence)
 }
 
-func (as *mqttAccountSessionManager) processSessPersistRecord(spr *sessPersistRecord) {
-	smsg, err := as.jsa.loadMsg(mqttSessStreamName, spr.seq)
+func (as *mqttAccountSessionManager) processSessPersistRecord(seq uint64) {
+	smsg, err := as.jsa.loadMsg(mqttSessStreamName, seq)
 	if err != nil {
 		return
 	}
@@ -1551,7 +1522,7 @@ func (as *mqttAccountSessionManager) processSessPersistRecord(spr *sessPersistRe
 	// If our current session's stream sequence is higher, it means that this
 	// update is stale, so we don't do anything here.
 	sess.mu.Lock()
-	ignore := spr.seq < sess.seq
+	ignore := seq < sess.seq
 	sess.mu.Unlock()
 	if ignore {
 		return
@@ -1573,22 +1544,18 @@ func (as *mqttAccountSessionManager) processSessPersistRecord(spr *sessPersistRe
 
 func (as *mqttAccountSessionManager) sessPersistProcessing(closeCh chan struct{}) {
 	as.mu.RLock()
-	sp := &as.sp
+	sp := as.sp
 	quitCh := as.jsa.quitCh
 	as.mu.RUnlock()
 
 	for {
 		select {
 		case <-sp.ch:
-			sp.mu.Lock()
-			l := sp.head
-			sp.head, sp.tail = nil, nil
-			sp.mu.Unlock()
-
-			for spr := l; spr != nil; spr = l.next {
-				l = spr
-				as.processSessPersistRecord(spr)
+			seqs := sp.pop()
+			for _, seq := range seqs {
+				as.processSessPersistRecord(seq.(uint64))
 			}
+			sp.recycle(&seqs)
 		case <-closeCh:
 			return
 		case <-quitCh:
@@ -1683,44 +1650,49 @@ func (as *mqttAccountSessionManager) sendJSAPIrequests(s *Server, c *client, acc
 
 	for {
 		select {
-		case r := <-sendq:
-
-			var nsize int
-
-			msg := r.msg
-			// If r.hdr is set to -1, it means that there is no need for any header.
-			if r.hdr != -1 {
-				bb := bytes.Buffer{}
-				if r.hdr > 0 {
-					// This means that the header has been set by the caller and is
-					// already part of `msg`, so simply set c.pa.hdr to the given value.
-					c.pa.hdr = r.hdr
+		case <-sendq.ch:
+			pmis := sendq.pop()
+			for _, pmi := range pmis {
+				r := pmi.(*mqttJSPubMsg)
+
+				var nsize int
+
+				msg := r.msg
+				// If r.hdr is set to -1, it means that there is no need for any header.
+				if r.hdr != -1 {
+					bb := bytes.Buffer{}
+					if r.hdr > 0 {
+						// This means that the header has been set by the caller and is
+						// already part of `msg`, so simply set c.pa.hdr to the given value.
+						c.pa.hdr = r.hdr
+						nsize = len(msg)
+						msg = append(msg, _CRLF_...)
+					} else {
+						// We need the ClientInfo header, so add it here.
+						bb.Write(hdrb)
+						c.pa.hdr = bb.Len()
+						bb.Write(r.msg)
+						nsize = bb.Len()
+						bb.WriteString(_CRLF_)
+						msg = bb.Bytes()
+					}
+					c.pa.hdb = []byte(strconv.Itoa(c.pa.hdr))
+				} else {
+					c.pa.hdr = -1
+					c.pa.hdb = nil
 					nsize = len(msg)
 					msg = append(msg, _CRLF_...)
-				} else {
-					// We need the ClientInfo header, so add it here.
-					bb.Write(hdrb)
-					c.pa.hdr = bb.Len()
-					bb.Write(r.msg)
-					nsize = bb.Len()
-					bb.WriteString(_CRLF_)
-					msg = bb.Bytes()
 				}
-				c.pa.hdb = []byte(strconv.Itoa(c.pa.hdr))
-			} else {
-				c.pa.hdr = -1
-				c.pa.hdb = nil
-				nsize = len(msg)
-				msg = append(msg, _CRLF_...)
-			}
 
-			c.pa.subject = []byte(r.subj)
-			c.pa.reply = []byte(r.reply)
-			c.pa.size = nsize
-			c.pa.szb = []byte(strconv.Itoa(nsize))
+				c.pa.subject = []byte(r.subj)
+				c.pa.reply = []byte(r.reply)
+				c.pa.size = nsize
+				c.pa.szb = []byte(strconv.Itoa(nsize))
 
-			c.processInboundClientMsg(msg)
-			c.flushClients(0)
+				c.processInboundClientMsg(msg)
+				c.flushClients(0)
+			}
+			sendq.recycle(&pmis)
 
 		case <-closeCh:
 			return
@@ -2125,10 +2097,10 @@ func (as *mqttAccountSessionManager) notifyRetainedMsgDeleted(subject string, se
 	}
 	b, _ := json.Marshal(&req)
 	jsa := &as.jsa
-	jsa.sendq <- &mqttJSPubMsg{
+	jsa.sendq.push(&mqttJSPubMsg{
 		subj: jsa.rplyr + mqttJSARetainedMsgDel,
 		msg:  b,
-	}
+	})
 }
 
 func (as *mqttAccountSessionManager) transferUniqueSessStreamsToMuxed(log *Server) {
@@ -2267,11 +2239,10 @@ func (sess *mqttSession) clear() error {
 		}
 	}
 	sess.subs, sess.pending, sess.cpending, sess.seq, sess.tmaxack = nil, nil, nil, 0, 0
-	sess.mu.Unlock()
-
 	for _, dur := range durs {
-		sess.jsa.sendq <- &mqttJSPubMsg{subj: sess.jsa.prefixDomain(fmt.Sprintf(JSApiConsumerDeleteT, mqttStreamName, dur))}
+		sess.jsa.sendq.push(&mqttJSPubMsg{subj: sess.jsa.prefixDomain(fmt.Sprintf(JSApiConsumerDeleteT, mqttStreamName, dur))})
 	}
+	sess.mu.Unlock()
 	if seq > 0 {
 		if err := sess.jsa.deleteMsg(mqttSessStreamName, seq, true); err != nil {
 			return fmt.Errorf("unable to delete session %q record at sequence %v", id, seq)
@@ -2422,8 +2393,8 @@ func (sess *mqttSession) createConsumer(consConfig *ConsumerConfig) error {
 func (sess *mqttSession) deleteConsumer(cc *ConsumerConfig) {
 	sess.mu.Lock()
 	sess.tmaxack -= cc.MaxAckPending
+	sess.jsa.sendq.push(&mqttJSPubMsg{subj: sess.jsa.prefixDomain(fmt.Sprintf(JSApiConsumerDeleteT, mqttStreamName, cc.Durable))})
 	sess.mu.Unlock()
-	sess.jsa.sendq <- &mqttJSPubMsg{subj: sess.jsa.prefixDomain(fmt.Sprintf(JSApiConsumerDeleteT, mqttStreamName, cc.Durable))}
 }
 
 //////////////////////////////////////////////////////////////////////////////
@@ -3168,14 +3139,13 @@ func (c *client) mqttProcessPubAck(pi uint16) {
 		}
 		ackSubject = ack.subject
 	}
-	sess.mu.Unlock()
-
-	// Send the ack if applicable, this is done outside of the session lock.
+	// Send the ack if applicable.
 	if ackSubject != _EMPTY_ {
 		// We pass -1 for the hdr so that the send loop does not need to
 		// add the "client info" header. This is not a JS API request per se.
-		sess.jsa.sendq <- &mqttJSPubMsg{subj: ackSubject, hdr: -1}
+		sess.jsa.sendq.push(&mqttJSPubMsg{subj: ackSubject, hdr: -1})
 	}
+	sess.mu.Unlock()
 }
 
 // Return the QoS from the given PUBLISH protocol's flags
diff --git a/server/raft.go b/server/raft.go
index fb7687f9..2953c32c 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -61,7 +61,7 @@ type RaftNode interface {
 	AdjustClusterSize(csz int) error
 	AdjustBootClusterSize(csz int) error
 	ClusterSize() int
-	ApplyC() <-chan *CommittedEntry
+	ApplyQ() *ipQueue // of *CommittedEntry
 	PauseApply()
 	ResumeApply()
 	LeadChangeC() <-chan bool
@@ -181,22 +181,22 @@ type raft struct {
 	catchup *catchupState
 
 	// For leader or server catching up a follower.
-	progress map[string]chan uint64
+	progress map[string]*ipQueue // of uint64
 
 	// For when we have paused our applyC.
 	paused  bool
 	hcommit uint64
 
-	// Channels
-	propc    chan *Entry
-	entryc   chan *appendEntry
-	respc    chan *appendEntryResponse
-	applyc   chan *CommittedEntry
-	quit     chan struct{}
-	reqs     chan *voteRequest
-	votes    chan *voteResponse
+	// Queues and Channels
+	prop     *ipQueue // of *Entry
+	entry    *ipQueue // of *appendEntry
+	resp     *ipQueue // of *appendEntryResponse
+	apply    *ipQueue // of *CommittedEntry
+	reqs     *ipQueue // of *voteRequest
+	votes    *ipQueue // of *voteResponse
+	stepdown *ipQueue // of string
 	leadc    chan bool
-	stepdown chan string
+	quit     chan struct{}
 }
 
 // cacthupState structure that holds our subscription, and catchup term and index
@@ -243,12 +243,10 @@ type RaftConfig struct {
 }
 
 var (
-	errProposalFailed    = errors.New("raft: proposal failed")
 	errNotLeader         = errors.New("raft: not leader")
 	errAlreadyLeader     = errors.New("raft: already leader")
 	errNilCfg            = errors.New("raft: no config given")
 	errCorruptPeers      = errors.New("raft: corrupt peer state")
-	errStepdownFailed    = errors.New("raft: stepdown failed")
 	errEntryLoadFailed   = errors.New("raft: could not load entry from WAL")
 	errEntryStoreFailed  = errors.New("raft: could not storeentry to WAL")
 	errNodeClosed        = errors.New("raft: node is closed")
@@ -263,7 +261,6 @@ var (
 	errLeaderLen         = fmt.Errorf("raft: leader should be exactly %d bytes", idLen)
 	errTooManyEntries    = errors.New("raft: append entry can contain a max of 64k entries")
 	errBadAppendEntry    = errors.New("raft: append entry corrupt")
-	errChannelFull       = errors.New("raft: entry failed to be placed on internal channel")
 )
 
 // This will bootstrap a raftNode by writing its config into the store directory.
@@ -353,6 +350,7 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 		return nil, errNoPeerState
 	}
 
+	qpfx := fmt.Sprintf("RAFT [%s - %s] ", hash[:idLen], cfg.Name)
 	n := &raft{
 		created:  time.Now(),
 		id:       hash[:idLen],
@@ -375,14 +373,14 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 		quit:     make(chan struct{}),
 		wtvch:    make(chan struct{}, 1),
 		wpsch:    make(chan struct{}, 1),
-		reqs:     make(chan *voteRequest, 8),
-		votes:    make(chan *voteResponse, 32),
-		propc:    make(chan *Entry, 8192),
-		entryc:   make(chan *appendEntry, 32768),
-		respc:    make(chan *appendEntryResponse, 32768),
-		applyc:   make(chan *CommittedEntry, 32768),
-		leadc:    make(chan bool, 8),
-		stepdown: make(chan string, 8),
+		reqs:     newIPQueue(),                                                     // of *voteRequest
+		votes:    newIPQueue(),                                                     // of *voteResponse
+		prop:     newIPQueue(ipQueue_Logger(qpfx+"Entry", s.ipqLog)),               // of *Entry
+		entry:    newIPQueue(ipQueue_Logger(qpfx+"AppendEntry", s.ipqLog)),         // of *appendEntry
+		resp:     newIPQueue(ipQueue_Logger(qpfx+"AppendEntryResponse", s.ipqLog)), // of *appendEntryResponse
+		apply:    newIPQueue(ipQueue_Logger(qpfx+"CommittedEntry", s.ipqLog)),      // of *CommittedEntry
+		stepdown: newIPQueue(),                                                     // of string
+		leadc:    make(chan bool, 1),
 		observer: cfg.Observer,
 		extSt:    ps.domainExt,
 	}
@@ -423,13 +421,6 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 			}
 		}
 
-		// Replay the log.
-		// Since doing this in place we need to make sure we have enough room on the applyc.
-		needed := state.Msgs + 1 // 1 is for nil to mark end of replay.
-		if uint64(cap(n.applyc)) < needed {
-			n.applyc = make(chan *CommittedEntry, needed)
-		}
-
 		for index := state.FirstSeq; index <= state.LastSeq; index++ {
 			ae, err := n.loadEntry(index)
 			if err != nil {
@@ -451,7 +442,7 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 	}
 
 	// Send nil entry to signal the upper layers we are done doing replay/restore.
-	n.applyc <- nil
+	n.apply.push(nil)
 
 	// Setup our internal subscriptions.
 	if err := n.createInternalSubs(); err != nil {
@@ -613,12 +604,10 @@ func (n *raft) Propose(data []byte) error {
 		n.RUnlock()
 		return werr
 	}
-	propc := n.propc
+	prop := n.prop
 	n.RUnlock()
 
-	// For entering and exiting the system, proposals and apply
-	// we will block.
-	propc <- &Entry{EntryNormal, data}
+	prop.push(&Entry{EntryNormal, data})
 	return nil
 }
 
@@ -667,21 +656,17 @@ func (n *raft) ProposeAddPeer(peer string) error {
 		n.RUnlock()
 		return werr
 	}
-	propc := n.propc
+	prop := n.prop
 	n.RUnlock()
 
-	select {
-	case propc <- &Entry{EntryAddPeer, []byte(peer)}:
-	default:
-		return errProposalFailed
-	}
+	prop.push(&Entry{EntryAddPeer, []byte(peer)})
 	return nil
 }
 
 // ProposeRemovePeer is called to remove a peer from the group.
 func (n *raft) ProposeRemovePeer(peer string) error {
 	n.RLock()
-	propc, subj := n.propc, n.rpsubj
+	prop, subj := n.prop, n.rpsubj
 	isLeader := n.state == Leader
 	werr := n.werr
 	n.RUnlock()
@@ -692,11 +677,7 @@ func (n *raft) ProposeRemovePeer(peer string) error {
 	}
 
 	if isLeader {
-		select {
-		case propc <- &Entry{EntryRemovePeer, []byte(peer)}:
-		default:
-			return errProposalFailed
-		}
+		prop.push(&Entry{EntryRemovePeer, []byte(peer)})
 		return nil
 	}
 
@@ -1025,7 +1006,7 @@ func (n *raft) setupLastSnapshot() {
 		n.pterm = snap.lastTerm
 		n.commit = snap.lastIndex
 		n.applied = snap.lastIndex
-		n.applyc <- &CommittedEntry{n.commit, []*Entry{{EntrySnapshot, snap.data}}}
+		n.apply.push(&CommittedEntry{n.commit, []*Entry{{EntrySnapshot, snap.data}}})
 		if _, err := n.wal.Compact(snap.lastIndex + 1); err != nil {
 			n.setWriteErrLocked(err)
 		}
@@ -1207,11 +1188,7 @@ func (n *raft) StepDown(preferred ...string) error {
 		n.sendAppendEntry([]*Entry{{EntryLeaderTransfer, []byte(maybeLeader)}})
 	}
 	// Force us to stepdown here.
-	select {
-	case stepdown <- noLeader:
-	default:
-		return errStepdownFailed
-	}
+	stepdown.push(noLeader)
 	return nil
 }
 
@@ -1298,9 +1275,9 @@ func (n *raft) Peers() []*Peer {
 	return peers
 }
 
-func (n *raft) ApplyC() <-chan *CommittedEntry { return n.applyc }
-func (n *raft) LeadChangeC() <-chan bool       { return n.leadc }
-func (n *raft) QuitC() <-chan struct{}         { return n.quit }
+func (n *raft) ApplyQ() *ipQueue         { return n.apply } // queue of *CommittedEntry
+func (n *raft) LeadChangeC() <-chan bool { return n.leadc }
+func (n *raft) QuitC() <-chan struct{}   { return n.quit }
 
 func (n *raft) Created() time.Time {
 	n.RLock()
@@ -1501,12 +1478,36 @@ func (n *raft) setObserver(isObserver bool, extSt extensionState) {
 	n.extSt = extSt
 }
 
+// Invoked when being notified that there is something in the entryc's queue
+func (n *raft) processAppendEntries() {
+	aes := n.entry.pop()
+	for _, aei := range aes {
+		ae := aei.(*appendEntry)
+		n.processAppendEntry(ae, ae.sub)
+	}
+	n.entry.recycle(&aes)
+}
+
+func convertVoteRequest(i interface{}) *voteRequest {
+	if i == nil {
+		return nil
+	}
+	return i.(*voteRequest)
+}
+
+func convertVoteResponse(i interface{}) *voteResponse {
+	if i == nil {
+		return nil
+	}
+	return i.(*voteResponse)
+}
+
 func (n *raft) runAsFollower() {
 	for {
 		elect := n.electTimer()
 		select {
-		case ae := <-n.entryc:
-			n.processAppendEntry(ae, ae.sub)
+		case <-n.entry.ch:
+			n.processAppendEntries()
 		case <-n.s.quitCh:
 			n.shutdown(false)
 			return
@@ -1524,13 +1525,17 @@ func (n *raft) runAsFollower() {
 				n.switchToCandidate()
 				return
 			}
-		case <-n.respc:
-			// Ignore
-		case <-n.votes:
+		case <-n.votes.ch:
 			n.debug("Ignoring old vote response, we have stepped down")
-		case vreq := <-n.reqs:
-			n.processVoteRequest(vreq)
-		case newLeader := <-n.stepdown:
+			n.votes.popOne()
+		case <-n.resp.ch:
+			// Ignore
+			n.resp.popOne()
+		case <-n.reqs.ch:
+			// Because of drain() it is possible that we get nil from popOne().
+			n.processVoteRequest(convertVoteRequest(n.reqs.popOne()))
+		case <-n.stepdown.ch:
+			newLeader := n.stepdown.popOne().(string)
 			n.switchToFollower(newLeader)
 			return
 		}
@@ -1739,7 +1744,7 @@ func (n *raft) handleForwardedRemovePeerProposal(sub *subscription, c *client, _
 	peer := string(copyBytes(msg))
 
 	n.RLock()
-	propc, werr := n.propc, n.werr
+	prop, werr := n.prop, n.werr
 	n.RUnlock()
 
 	// Ignore if we have had a write error previous.
@@ -1747,11 +1752,7 @@ func (n *raft) handleForwardedRemovePeerProposal(sub *subscription, c *client, _
 		return
 	}
 
-	select {
-	case propc <- &Entry{EntryRemovePeer, []byte(peer)}:
-	default:
-		n.warn("Failed to place peer removal proposal onto propose channel")
-	}
+	prop.push(&Entry{EntryRemovePeer, []byte(peer)})
 }
 
 // Called when a peer has forwarded a proposal.
@@ -1764,7 +1765,7 @@ func (n *raft) handleForwardedProposal(sub *subscription, c *client, _ *Account,
 	msg = copyBytes(msg)
 
 	n.RLock()
-	propc, werr := n.propc, n.werr
+	prop, werr := n.prop, n.werr
 	n.RUnlock()
 
 	// Ignore if we have had a write error previous.
@@ -1772,11 +1773,7 @@ func (n *raft) handleForwardedProposal(sub *subscription, c *client, _ *Account,
 		return
 	}
 
-	select {
-	case propc <- &Entry{EntryNormal, msg}:
-	default:
-		n.warn("Failed to place forwarded proposal onto propose channel")
-	}
+	prop.push(&Entry{EntryNormal, msg})
 }
 
 func (n *raft) runAsLeader() {
@@ -1820,24 +1817,32 @@ func (n *raft) runAsLeader() {
 			return
 		case <-n.quit:
 			return
-		case ar := <-n.respc:
-			n.processAppendEntryResponse(ar)
-		case b := <-n.propc:
-			entries := []*Entry{b}
-			if b.Type == EntryNormal {
-				const maxBatch = 256 * 1024
-			gather:
-				for sz := 0; sz < maxBatch && len(entries) < math.MaxUint16; {
-					select {
-					case e := <-n.propc:
-						entries = append(entries, e)
-						sz += len(e.Data) + 1
-					default:
-						break gather
-					}
+		case <-n.resp.ch:
+			ars := n.resp.pop()
+			for _, ari := range ars {
+				ar := ari.(*appendEntryResponse)
+				n.processAppendEntryResponse(ar)
+			}
+			n.resp.recycle(&ars)
+		case <-n.prop.ch:
+			const maxBatch = 256 * 1024
+			var entries []*Entry
+
+			es := n.prop.pop()
+			sz := 0
+			for i, bi := range es {
+				b := bi.(*Entry)
+				entries = append(entries, b)
+				sz += len(b.Data) + 1
+				if i != len(es)-1 && sz < maxBatch && len(entries) < math.MaxUint16 {
+					continue
 				}
+				n.sendAppendEntry(entries)
+				// We need to re-craete `entries` because there is a reference
+				// to it in the node's pae map.
+				entries = nil
 			}
-			n.sendAppendEntry(entries)
+			n.prop.recycle(&es)
 		case <-hb.C:
 			if n.notActive() {
 				n.sendHeartbeat()
@@ -1846,19 +1851,26 @@ func (n *raft) runAsLeader() {
 				n.switchToFollower(noLeader)
 				return
 			}
-		case vresp := <-n.votes:
+		case <-n.votes.ch:
+			// Because of drain() it is possible that we get nil from popOne().
+			vresp := convertVoteResponse(n.votes.popOne())
+			if vresp == nil {
+				continue
+			}
 			if vresp.term > n.currentTerm() {
 				n.switchToFollower(noLeader)
 				return
 			}
 			n.trackPeer(vresp.peer)
-		case vreq := <-n.reqs:
-			n.processVoteRequest(vreq)
-		case newLeader := <-n.stepdown:
+		case <-n.reqs.ch:
+			// Because of drain() it is possible that we get nil from popOne().
+			n.processVoteRequest(convertVoteRequest(n.reqs.popOne()))
+		case <-n.stepdown.ch:
+			newLeader := n.stepdown.popOne().(string)
 			n.switchToFollower(newLeader)
 			return
-		case ae := <-n.entryc:
-			n.processAppendEntry(ae, ae.sub)
+		case <-n.entry.ch:
+			n.processAppendEntries()
 		}
 	}
 }
@@ -1921,7 +1933,7 @@ func (n *raft) loadFirstEntry() (ae *appendEntry, err error) {
 	return n.loadEntry(state.FirstSeq)
 }
 
-func (n *raft) runCatchup(ar *appendEntryResponse, indexUpdatesC <-chan uint64) {
+func (n *raft) runCatchup(ar *appendEntryResponse, indexUpdatesQ *ipQueue /* of uint64 */) {
 	n.RLock()
 	s, reply := n.s, n.areply
 	peer, subj, last := ar.peer, ar.reply, n.pindex
@@ -1993,7 +2005,8 @@ func (n *raft) runCatchup(ar *appendEntryResponse, indexUpdatesC <-chan uint64)
 		case <-timeout.C:
 			n.debug("Catching up for %q stalled", peer)
 			return
-		case index := <-indexUpdatesC:
+		case <-indexUpdatesQ.ch:
+			index := indexUpdatesQ.popOne().(uint64)
 			// Update our activity timer.
 			timeout.Reset(activityInterval)
 			// Update outstanding total.
@@ -2016,7 +2029,7 @@ func (n *raft) sendSnapshotToFollower(subject string) (uint64, error) {
 	snap, err := n.loadLastSnapshot()
 	if err != nil {
 		// We need to stepdown here when this happens.
-		n.attemptStepDown(noLeader)
+		n.stepdown.push(noLeader)
 		return 0, err
 	}
 	// Go ahead and send the snapshot and peerstate here as first append entry to the catchup follower.
@@ -2040,15 +2053,11 @@ func (n *raft) catchupFollower(ar *appendEntryResponse) {
 	n.debug("Being asked to catch up follower: %q", ar.peer)
 	n.Lock()
 	if n.progress == nil {
-		n.progress = make(map[string]chan uint64)
-	} else if ch, ok := n.progress[ar.peer]; ok {
+		n.progress = make(map[string]*ipQueue)
+	} else if q, ok := n.progress[ar.peer]; ok {
 		n.debug("Will cancel existing entry for catching up %q", ar.peer)
 		delete(n.progress, ar.peer)
-		// Try to pop them out but make sure to not block.
-		select {
-		case ch <- n.pindex:
-		default:
-		}
+		q.push(n.pindex)
 	}
 	// Check to make sure we have this entry.
 	start := ar.index + 1
@@ -2085,18 +2094,10 @@ func (n *raft) catchupFollower(ar *appendEntryResponse) {
 	if ae.pindex != ar.index || ae.pterm != ar.term {
 		n.debug("Our first entry does not match request from follower")
 	}
-	// Create a chan for delivering updates from responses.
-	isz := 256
-	if ae.pindex > ar.index && ae.pindex-ar.index > uint64(isz) {
-		isz = int(ae.pindex - ar.index)
-	}
-	// Check if we already have one in place and its bigger.
-	if preCh, ok := n.progress[ar.peer]; ok && cap(preCh) > isz {
-		isz = cap(preCh)
-	}
-	indexUpdates := make(chan uint64, isz)
-
-	indexUpdates <- ae.pindex
+	// Create a queue for delivering updates from responses.
+	qname := fmt.Sprintf("RAFT [%s - %s] Index updates", n.id, n.group)
+	indexUpdates := newIPQueue(ipQueue_Logger(qname, n.s.ipqLog)) // of uint64
+	indexUpdates.push(ae.pindex)
 	n.progress[ar.peer] = indexUpdates
 	n.Unlock()
 
@@ -2213,7 +2214,7 @@ func (n *raft) applyCommit(index uint64) error {
 
 			// If this is us and we are the leader we should attempt to stepdown.
 			if peer == n.id && n.state == Leader {
-				n.attemptStepDown(n.selectNextLeader())
+				n.stepdown.push(n.selectNextLeader())
 			}
 
 			// Write out our new state.
@@ -2222,29 +2223,17 @@ func (n *raft) applyCommit(index uint64) error {
 			committed = append(committed, e)
 		}
 	}
-	var err error
 	// Pass to the upper layers if we have normal entries.
 	if len(committed) > 0 {
 		if fpae {
 			delete(n.pae, index)
 		}
-		// For entering and exiting the system, proposals and apply we
-		// will block.
-		closed := n.state == Closed
-		n.Unlock()
-		if !closed {
-			select {
-			case n.applyc <- &CommittedEntry{index, committed}:
-			default:
-				err = errChannelFull
-			}
-		}
-		n.Lock()
+		n.apply.push(&CommittedEntry{index, committed})
 	} else {
 		// If we processed inline update our applied index.
 		n.applied = index
 	}
-	return err
+	return nil
 }
 
 // Used to track a success response and apply entries.
@@ -2257,12 +2246,8 @@ func (n *raft) trackResponse(ar *appendEntryResponse) {
 	}
 
 	// If we are tracking this peer as a catchup follower, update that here.
-	if indexUpdateC := n.progress[ar.peer]; indexUpdateC != nil {
-		select {
-		case indexUpdateC <- ar.index:
-		default:
-			n.warn("TrackResponse failed to place progress update on internal channel")
-		}
+	if indexUpdateQ := n.progress[ar.peer]; indexUpdateQ != nil {
+		indexUpdateQ.push(ar.index)
 	}
 
 	// Ignore items already committed.
@@ -2283,7 +2268,7 @@ func (n *raft) trackResponse(ar *appendEntryResponse) {
 					break
 				}
 			}
-			sendHB = len(n.propc) == 0
+			sendHB = n.prop.len() == 0
 		}
 	}
 
@@ -2341,9 +2326,7 @@ func (n *raft) numActivePeers() int {
 func (n *raft) runAsCandidate() {
 	n.Lock()
 	// Drain old responses.
-	for len(n.votes) > 0 {
-		<-n.votes
-	}
+	n.votes.drain()
 	n.Unlock()
 
 	// Send out our request for votes.
@@ -2356,10 +2339,11 @@ func (n *raft) runAsCandidate() {
 	for {
 		elect := n.electTimer()
 		select {
-		case ae := <-n.entryc:
-			n.processAppendEntry(ae, ae.sub)
-		case <-n.respc:
+		case <-n.entry.ch:
+			n.processAppendEntries()
+		case <-n.resp.ch:
 			// Ignore
+			n.resp.popOne()
 		case <-n.s.quitCh:
 			n.shutdown(false)
 			return
@@ -2373,7 +2357,12 @@ func (n *raft) runAsCandidate() {
 				n.switchToCandidate()
 			}
 			return
-		case vresp := <-n.votes:
+		case <-n.votes.ch:
+			// Because of drain() it is possible that we get nil from popOne().
+			vresp := convertVoteResponse(n.votes.popOne())
+			if vresp == nil {
+				continue
+			}
 			n.RLock()
 			nterm, lxfer := n.term, n.lxfer
 			n.RUnlock()
@@ -2403,14 +2392,16 @@ func (n *raft) runAsCandidate() {
 				n.term = vresp.term
 				n.vote = noVote
 				n.writeTermVote()
-				n.Unlock()
 				n.debug("Stepping down from candidate, detected higher term: %d vs %d", vresp.term, n.term)
-				n.attemptStepDown(noLeader)
+				n.stepdown.push(noLeader)
+				n.Unlock()
 				return
 			}
-		case vreq := <-n.reqs:
-			n.processVoteRequest(vreq)
-		case newLeader := <-n.stepdown:
+		case <-n.reqs.ch:
+			// Because of drain() it is possible that we get nil from popOne().
+			n.processVoteRequest(convertVoteRequest(n.reqs.popOne()))
+		case <-n.stepdown.ch:
+			newLeader := n.stepdown.popOne().(string)
 			n.switchToFollower(newLeader)
 			return
 		}
@@ -2426,11 +2417,7 @@ func (n *raft) handleAppendEntry(sub *subscription, c *client, _ *Account, subje
 
 	msg = copyBytes(msg)
 	if ae, err := n.decodeAppendEntry(msg, sub, reply); err == nil {
-		select {
-		case n.entryc <- ae:
-		default:
-			n.warn("AppendEntry failed to be placed on internal channel")
-		}
+		n.entry.push(ae)
 	} else {
 		n.warn("AppendEntry failed to be placed on internal channel: corrupt entry")
 	}
@@ -2481,15 +2468,6 @@ func (n *raft) createCatchup(ae *appendEntry) string {
 	return inbox
 }
 
-// Attempt to stepdown, lock should be held.
-func (n *raft) attemptStepDown(newLeader string) {
-	select {
-	case n.stepdown <- newLeader:
-	default:
-		n.debug("Failed to place stepdown for new leader %q for %q", newLeader, n.group)
-	}
-}
-
 // Truncate our WAL and reset.
 func (n *raft) truncateWAL(pterm, pindex uint64) {
 	n.debug("Truncating and repairing WAL")
@@ -2528,7 +2506,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			n.vote = noVote
 			n.writeTermVote()
 			n.debug("Received append entry from another leader, stepping down to %q", ae.leader)
-			n.attemptStepDown(ae.leader)
+			n.stepdown.push(ae.leader)
 		} else {
 			// Let them know we are the leader.
 			ar := &appendEntryResponse{n.term, n.pindex, n.id, false, _EMPTY_}
@@ -2547,7 +2525,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			n.vote = noVote
 			n.writeTermVote()
 		}
-		n.attemptStepDown(ae.leader)
+		n.stepdown.push(ae.leader)
 	}
 
 	n.resetElectionTimeout()
@@ -2615,7 +2593,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		}
 		if n.state != Follower {
 			n.debug("Term higher than ours and we are not a follower: %v, stepping down to %q", n.state, ae.leader)
-			n.attemptStepDown(ae.leader)
+			n.stepdown.push(ae.leader)
 		}
 	}
 
@@ -2689,12 +2667,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			}
 
 			// Now send snapshot to upper levels. Only send the snapshot, not the peerstate entry.
-			select {
-			case n.applyc <- &CommittedEntry{n.commit, ae.entries[:1]}:
-			default:
-				n.warn("Failed to place snapshot entry onto our apply channel")
-				n.commit--
-			}
+			n.apply.push(&CommittedEntry{n.commit, ae.entries[:1]})
 			n.Unlock()
 			return
 
@@ -2726,7 +2699,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			}
 			// Save in memory for faster processing during applyCommit.
 			n.pae[n.pindex] = ae
-			if len(n.pae) > paeWarnThreshold {
+			if l := len(n.pae); l > paeWarnThreshold && l%1000 == 0 {
 				n.warn("%d append entries pending", len(n.pae))
 			}
 
@@ -2808,9 +2781,7 @@ func (n *raft) processAppendEntryResponse(ar *appendEntryResponse) {
 		n.term = ar.term
 		n.vote = noVote
 		n.writeTermVote()
-		n.Lock()
-		n.attemptStepDown(noLeader)
-		n.Unlock()
+		n.stepdown.push(noLeader)
 	} else if ar.reply != _EMPTY_ {
 		n.catchupFollower(ar)
 	}
@@ -2821,12 +2792,7 @@ func (n *raft) handleAppendEntryResponse(sub *subscription, c *client, _ *Accoun
 	msg = copyBytes(msg)
 	ar := n.decodeAppendEntryResponse(msg)
 	ar.reply = reply
-
-	select {
-	case n.respc <- ar:
-	default:
-		n.warn("AppendEntryResponse failed to be placed on internal channel")
-	}
+	n.resp.push(ar)
 }
 
 func (n *raft) buildAppendEntry(entries []*Entry) *appendEntry {
@@ -2856,7 +2822,7 @@ func (n *raft) storeToWAL(ae *appendEntry) error {
 			if ae, err := n.loadEntry(seq - 1); err == nil && ae != nil {
 				n.truncateWAL(ae.pterm, ae.pindex)
 				if n.state == Leader {
-					n.attemptStepDown(n.selectNextLeader())
+					n.stepdown.push(n.selectNextLeader())
 				}
 			} else {
 				panic(fmt.Sprintf("[%s | %s] Wrong index, ae is %+v, seq is %d, n.pindex is %d\n\n", n.s, n.group, ae, seq, n.pindex))
@@ -2898,7 +2864,7 @@ func (n *raft) sendAppendEntry(entries []*Entry) {
 
 		// Save in memory for faster processing during applyCommit.
 		n.pae[n.pindex] = ae
-		if len(n.pae) > paeWarnThreshold {
+		if l := len(n.pae); l > paeWarnThreshold && l%1000 == 0 {
 			n.warn("%d append entries pending", len(n.pae))
 		}
 	}
@@ -3210,15 +3176,20 @@ func (n *raft) handleVoteResponse(sub *subscription, c *client, _ *Account, _, r
 		return
 	}
 
-	select {
-	case n.votes <- vr:
-	default:
-		// FIXME(dlc)
-		n.error("Failed to place vote response on chan for %q", n.group)
+	if state := n.State(); state != Candidate && state != Leader {
+		n.debug("Ignoring old vote response, we have stepped down")
+		return
 	}
+
+	n.votes.push(vr)
 }
 
 func (n *raft) processVoteRequest(vr *voteRequest) error {
+	// To simplify calling code, we can possibly pass `nil` to this function.
+	// If that is the case, does not consider it an error.
+	if vr == nil {
+		return nil
+	}
 	n.debug("Received a voteRequest %+v", vr)
 
 	if err := n.trackPeer(vr.candidate); err != nil {
@@ -3242,7 +3213,7 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 	if vr.term > n.term {
 		if n.state != Follower {
 			n.debug("Stepping down from candidate, detected higher term: %d vs %d", vr.term, n.term)
-			n.attemptStepDown(noLeader)
+			n.stepdown.push(noLeader)
 		}
 		n.term = vr.term
 		n.vote = noVote
@@ -3269,11 +3240,7 @@ func (n *raft) handleVoteRequest(sub *subscription, c *client, _ *Account, subje
 		n.error("Received malformed vote request for %q", n.group)
 		return
 	}
-	select {
-	case n.reqs <- vr:
-	default:
-		n.error("Failed to place vote request on chan for %q", n.group)
-	}
+	n.reqs.push(vr)
 }
 
 func (n *raft) requestVote() {
@@ -3320,14 +3287,19 @@ func (n *raft) quorumNeeded() int {
 
 // Lock should be held.
 func (n *raft) updateLeadChange(isLeader bool) {
-	select {
-	case n.leadc <- isLeader:
-	case <-n.leadc:
-		// We had an old value not consumed.
+	// We don't care about values that have not been consumed (transitory states),
+	// so we dequeue any state that is pending and push the new one.
+	for {
 		select {
 		case n.leadc <- isLeader:
+			return
 		default:
-			n.error("Failed to post lead change to %v for %q", isLeader, n.group)
+			select {
+			case <-n.leadc:
+			default:
+				// May have been consumed by the "reader" go routine, so go back
+				// to the top of the loop and try to send again.
+			}
 		}
 	}
 }
@@ -3343,10 +3315,8 @@ func (n *raft) switchState(state RaftState) {
 
 	if n.state == Leader && state != Leader {
 		n.updateLeadChange(false)
-		// Drain our responses channel.
-		for len(n.respc) > 0 {
-			<-n.respc
-		}
+		// Drain the response queue.
+		n.resp.drain()
 	} else if state == Leader && n.state != Leader {
 		if len(n.pae) > 0 {
 			n.pae = make(map[uint64]*appendEntry)
diff --git a/server/sendq.go b/server/sendq.go
index 12413eac..35ff9456 100644
--- a/server/sendq.go
+++ b/server/sendq.go
@@ -23,26 +23,23 @@ type outMsg struct {
 	rply string
 	hdr  []byte
 	msg  []byte
-	next *outMsg
 }
 
 type sendq struct {
-	mu   sync.Mutex
-	mch  chan struct{}
-	head *outMsg
-	tail *outMsg
-	s    *Server
+	mu sync.Mutex
+	q  *ipQueue // of *outMsg
+	s  *Server
 }
 
 func (s *Server) newSendQ() *sendq {
-	sq := &sendq{s: s, mch: make(chan struct{}, 1)}
+	sq := &sendq{s: s, q: newIPQueue(ipQueue_Logger("Send", s.ipqLog))}
 	s.startGoRoutine(sq.internalLoop)
 	return sq
 }
 
 func (sq *sendq) internalLoop() {
 	sq.mu.Lock()
-	s, mch := sq.s, sq.mch
+	s, q := sq.s, sq.q
 	sq.mu.Unlock()
 
 	defer s.grWG.Done()
@@ -57,8 +54,10 @@ func (sq *sendq) internalLoop() {
 		select {
 		case <-s.quitCh:
 			return
-		case <-mch:
-			for pm := sq.pending(); pm != nil; {
+		case <-q.ch:
+			pms := q.pop()
+			for _, pmi := range pms {
+				pm := pmi.(*outMsg)
 				c.pa.subject = []byte(pm.subj)
 				c.pa.size = len(pm.msg) + len(pm.hdr)
 				c.pa.szb = []byte(strconv.Itoa(c.pa.size))
@@ -76,28 +75,16 @@ func (sq *sendq) internalLoop() {
 				}
 				c.processInboundClientMsg(msg)
 				c.pa.szb = nil
-				// Do this here to nil out below vs up in for loop.
-				next := pm.next
-				pm.next, pm.hdr, pm.msg = nil, nil, nil
-				if pm = next; pm == nil {
-					pm = sq.pending()
-				}
 			}
+			// TODO: should this be in the for-loop instead?
 			c.flushClients(0)
+			q.recycle(&pms)
 		}
 	}
 }
 
-func (sq *sendq) pending() *outMsg {
-	sq.mu.Lock()
-	head := sq.head
-	sq.head, sq.tail = nil, nil
-	sq.mu.Unlock()
-	return head
-}
-
 func (sq *sendq) send(subj, rply string, hdr, msg []byte) {
-	out := &outMsg{subj, rply, nil, nil, nil}
+	out := &outMsg{subj, rply, nil, nil}
 	// We will copy these for now.
 	if len(hdr) > 0 {
 		hdr = copyBytes(hdr)
@@ -107,22 +94,5 @@ func (sq *sendq) send(subj, rply string, hdr, msg []byte) {
 		msg = copyBytes(msg)
 		out.msg = msg
 	}
-
-	sq.mu.Lock()
-	var notify bool
-	if sq.head == nil {
-		sq.head = out
-		notify = true
-	} else {
-		sq.tail.next = out
-	}
-	sq.tail = out
-	sq.mu.Unlock()
-
-	if notify {
-		select {
-		case sq.mch <- struct{}{}:
-		default:
-		}
-	}
+	sq.q.push(out)
 }
diff --git a/server/server.go b/server/server.go
index 93d31c6e..1a44935f 100644
--- a/server/server.go
+++ b/server/server.go
@@ -268,6 +268,16 @@ type Server struct {
 
 	// How often user logon fails due to the issuer account not being pinned.
 	pinnedAccFail uint64
+
+	// This is a central logger for IPQueues when the number of pending
+	// messages reaches a certain thresold (per queue)
+	ipqLog *srvIPQueueLogger
+}
+
+type srvIPQueueLogger struct {
+	ch   chan string
+	done chan struct{}
+	s    *Server
 }
 
 // For tracking JS nodes.
@@ -1186,9 +1196,6 @@ func (s *Server) SetDefaultSystemAccount() error {
 	return s.SetSystemAccount(DEFAULT_SYSTEM_ACCOUNT)
 }
 
-// For internal sends.
-const internalSendQLen = 256 * 1024
-
 // Assign a system account. Should only be called once.
 // This sets up a server to send and receive messages from
 // inside the server itself.
@@ -1228,7 +1235,7 @@ func (s *Server) setSystemAccount(acc *Account) error {
 		sid:     1,
 		servers: make(map[string]*serverUpdate),
 		replies: make(map[string]msgHandler),
-		sendq:   make(chan *pubMsg, internalSendQLen),
+		sendq:   newIPQueue(ipQueue_Logger("System send", s.ipqLog)), // of *pubMsg
 		resetCh: make(chan struct{}),
 		sq:      s.newSendQ(),
 		statsz:  eventsHBInterval,
@@ -1598,6 +1605,8 @@ func (s *Server) Start() {
 	s.grRunning = true
 	s.grMu.Unlock()
 
+	s.startIPQLogger()
+
 	// Pprof http endpoint for the profiler.
 	if opts.ProfPort != 0 {
 		s.StartProfiler()
@@ -1966,6 +1975,11 @@ func (s *Server) Shutdown() {
 		doneExpected--
 	}
 
+	// Stop the IPQueue logger (before the grWG.Wait() call)
+	if s.ipqLog != nil {
+		s.ipqLog.stop()
+	}
+
 	// Wait for go routines to be done.
 	s.grWG.Wait()
 
@@ -3604,3 +3618,35 @@ func (s *Server) updateRemoteSubscription(acc *Account, sub *subscription, delta
 
 	s.updateLeafNodes(acc, sub, delta)
 }
+
+func (s *Server) startIPQLogger() {
+	s.ipqLog = &srvIPQueueLogger{
+		ch:   make(chan string, 128),
+		done: make(chan struct{}),
+		s:    s,
+	}
+	s.startGoRoutine(s.ipqLog.run)
+}
+
+func (l *srvIPQueueLogger) stop() {
+	close(l.done)
+}
+
+func (l *srvIPQueueLogger) log(name string, pending int) {
+	select {
+	case l.ch <- fmt.Sprintf("%s queue pending size: %v", name, pending):
+	default:
+	}
+}
+
+func (l *srvIPQueueLogger) run() {
+	defer l.s.grWG.Done()
+	for {
+		select {
+		case w := <-l.ch:
+			l.s.Warnf("%s", w)
+		case <-l.done:
+			return
+		}
+	}
+}
diff --git a/server/stream.go b/server/stream.go
index 0e302133..6e579dc0 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -164,9 +164,9 @@ type stream struct {
 	sid       int
 	pubAck    []byte
 	outq      *jsOutQ
-	msgs      *inbound
+	msgs      *ipQueue // of *inMsg
 	store     StreamStore
-	ackq      *ackMsgQueue
+	ackq      *ipQueue // of uint64
 	lseq      uint64
 	lmsgId    string
 	consumers map[string]*consumer
@@ -211,7 +211,7 @@ type sourceInfo struct {
 	iname string
 	cname string
 	sub   *subscription
-	msgs  *inbound
+	msgs  *ipQueue // of *inMsg
 	sseq  uint64
 	dseq  uint64
 	lag   uint64
@@ -368,6 +368,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 	c := s.createInternalJetStreamClient()
 	ic := s.createInternalJetStreamClient()
 
+	qname := fmt.Sprintf("Stream %s > %s messages", a.Name, config.Name)
 	mset := &stream{
 		acc:       a,
 		jsa:       jsa,
@@ -378,13 +379,13 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		sysc:      ic,
 		stype:     cfg.Storage,
 		consumers: make(map[string]*consumer),
-		msgs:      &inbound{mch: make(chan struct{}, 1)},
+		msgs:      newIPQueue(ipQueue_Logger(qname, s.ipqLog)), // of *inMsg
 		qch:       make(chan struct{}),
 	}
 
 	// For no-ack consumers when we are interest retention.
 	if cfg.Retention != LimitsPolicy {
-		mset.ackq = &ackMsgQueue{mch: make(chan struct{}, 1)}
+		mset.ackq = newIPQueue() // of uint64
 	}
 
 	jsa.streams[cfg.Name] = mset
@@ -1029,7 +1030,7 @@ func (mset *stream) update(config *StreamConfig) error {
 						mset.sources = make(map[string]*sourceInfo)
 					}
 					mset.cfg.Sources = append(mset.cfg.Sources, s)
-					si := &sourceInfo{name: s.Name, iname: s.iname, msgs: &inbound{mch: make(chan struct{}, 1)}}
+					si := &sourceInfo{name: s.Name, iname: s.iname, msgs: newIPQueue() /* of *inMsg */}
 					mset.sources[s.iname] = si
 					mset.setStartingSequenceForSource(s.iname)
 					mset.setSourceConsumer(s.iname, si.sseq+1)
@@ -1332,7 +1333,7 @@ func (mset *stream) processMirrorMsgs() {
 		mset.mu.Unlock()
 		return
 	}
-	msgs, mch, qch, siqch := mset.mirror.msgs, mset.mirror.msgs.mch, mset.qch, mset.mirror.qch
+	msgs, qch, siqch := mset.mirror.msgs, mset.qch, mset.mirror.qch
 	// Set the last seen as now so that we don't fail at the first check.
 	mset.mirror.last = time.Now()
 	mset.mu.Unlock()
@@ -1348,12 +1349,15 @@ func (mset *stream) processMirrorMsgs() {
 			return
 		case <-siqch:
 			return
-		case <-mch:
-			for im := mset.pending(msgs); im != nil; im = im.next {
+		case <-msgs.ch:
+			ims := msgs.pop()
+			for _, imi := range ims {
+				im := imi.(*inMsg)
 				if !mset.processInboundMirrorMsg(im) {
 					break
 				}
 			}
+			msgs.recycle(&ims)
 		case <-t.C:
 			mset.mu.RLock()
 			isLeader := mset.isLeader()
@@ -1549,6 +1553,8 @@ func (mset *stream) skipMsgs(start, end uint64) {
 			// So a single message does not get too big.
 			if len(entries) > 10_000 {
 				node.ProposeDirect(entries)
+				// We need to re-craete `entries` because there is a reference
+				// to it in the node's pae map.
 				entries = entries[:0]
 			}
 		} else {
@@ -1598,7 +1604,7 @@ func (mset *stream) setupMirrorConsumer() error {
 	}
 
 	if !isReset {
-		mset.mirror = &sourceInfo{name: mset.cfg.Mirror.Name, msgs: &inbound{mch: make(chan struct{}, 1)}}
+		mset.mirror = &sourceInfo{name: mset.cfg.Mirror.Name, msgs: newIPQueue() /* of *inMsg */}
 	}
 
 	if !mset.mirror.grr {
@@ -1672,7 +1678,7 @@ func (mset *stream) setupMirrorConsumer() error {
 		subject = strings.ReplaceAll(subject, "..", ".")
 	}
 
-	mset.outq.send(&jsPubMsg{subject, _EMPTY_, reply, nil, b, nil, 0, nil})
+	mset.outq.send(newJSPubMsg(subject, _EMPTY_, reply, nil, b, nil, 0))
 
 	go func() {
 		select {
@@ -1874,7 +1880,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64) {
 		subject = strings.ReplaceAll(subject, "..", ".")
 	}
 
-	mset.outq.send(&jsPubMsg{subject, _EMPTY_, reply, nil, b, nil, 0, nil})
+	mset.outq.send(newJSPubMsg(subject, _EMPTY_, reply, nil, b, nil, 0))
 
 	go func() {
 		select {
@@ -1937,7 +1943,7 @@ func (mset *stream) processSourceMsgs(si *sourceInfo) {
 
 	// Grab stream quit channel.
 	mset.mu.Lock()
-	msgs, mch, qch, siqch := si.msgs, si.msgs.mch, mset.qch, si.qch
+	msgs, qch, siqch := si.msgs, mset.qch, si.qch
 	// Set the last seen as now so that we don't fail at the first check.
 	si.last = time.Now()
 	mset.mu.Unlock()
@@ -1953,12 +1959,15 @@ func (mset *stream) processSourceMsgs(si *sourceInfo) {
 			return
 		case <-siqch:
 			return
-		case <-mch:
-			for im := mset.pending(msgs); im != nil; im = im.next {
+		case <-msgs.ch:
+			ims := msgs.pop()
+			for _, imi := range ims {
+				im := imi.(*inMsg)
 				if !mset.processInboundSourceMsg(si, im) {
 					break
 				}
 			}
+			msgs.recycle(&ims)
 		case <-t.C:
 			mset.mu.RLock()
 			iname, isLeader := si.iname, mset.isLeader()
@@ -2213,7 +2222,7 @@ func (mset *stream) startingSequenceForSources() {
 		if ssi.iname == _EMPTY_ {
 			ssi.setIndexName()
 		}
-		si := &sourceInfo{name: ssi.Name, iname: ssi.iname, msgs: &inbound{mch: make(chan struct{}, 1)}}
+		si := &sourceInfo{name: ssi.Name, iname: ssi.iname, msgs: newIPQueue() /* of *inMsg */}
 		mset.sources[ssi.iname] = si
 	}
 
@@ -2626,45 +2635,10 @@ type inMsg struct {
 	rply string
 	hdr  []byte
 	msg  []byte
-	next *inMsg
 }
 
-// Linked list for inbound messages.
-type inbound struct {
-	head *inMsg
-	tail *inMsg
-	mch  chan struct{}
-}
-
-func (mset *stream) pending(msgs *inbound) *inMsg {
-	mset.mu.Lock()
-	head := msgs.head
-	msgs.head, msgs.tail = nil, nil
-	mset.mu.Unlock()
-	return head
-}
-
-func (mset *stream) queueInbound(ib *inbound, subj, rply string, hdr, msg []byte) {
-	m := &inMsg{subj, rply, hdr, msg, nil}
-
-	mset.mu.Lock()
-	var notify bool
-	if ib.head == nil {
-		ib.head = m
-		notify = true
-	} else {
-		ib.tail.next = m
-	}
-	ib.tail = m
-	mch := ib.mch
-	mset.mu.Unlock()
-
-	if notify {
-		select {
-		case mch <- struct{}{}:
-		default:
-		}
-	}
+func (mset *stream) queueInbound(ib *ipQueue, subj, rply string, hdr, msg []byte) {
+	ib.push(&inMsg{subj, rply, hdr, msg})
 }
 
 func (mset *stream) queueInboundMsg(subj, rply string, hdr, msg []byte) {
@@ -3092,7 +3066,31 @@ type jsPubMsg struct {
 	msg   []byte
 	o     *consumer
 	seq   uint64
-	next  *jsPubMsg
+}
+
+var jsPubMsgPool sync.Pool
+
+func newJSPubMsg(subj, dsubj, reply string, hdr, msg []byte, o *consumer, seq uint64) *jsPubMsg {
+	var m *jsPubMsg
+	pm := jsPubMsgPool.Get()
+	if pm != nil {
+		m = pm.(*jsPubMsg)
+	} else {
+		m = &jsPubMsg{}
+	}
+	// 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{subj, dsubj, reply, hdr, msg, o, seq}
+	return m
+}
+
+func (pm *jsPubMsg) returnToPool() {
+	if pm == nil {
+		return
+	}
+	pm.subj, pm.dsubj, pm.reply, pm.hdr, pm.msg, pm.o = _EMPTY_, _EMPTY_, _EMPTY_, nil, nil, nil
+	jsPubMsgPool.Put(pm)
 }
 
 func (pm *jsPubMsg) size() int {
@@ -3102,28 +3100,14 @@ func (pm *jsPubMsg) size() int {
 	return len(pm.subj) + len(pm.reply) + len(pm.hdr) + len(pm.msg)
 }
 
-// Forms a linked list for sending internal system messages.
+// Queue of *jsPubMsg for sending internal system messages.
 type jsOutQ struct {
-	mu   sync.Mutex
-	mch  chan struct{}
-	head *jsPubMsg
-	tail *jsPubMsg
-}
-
-func (q *jsOutQ) pending() *jsPubMsg {
-	if q == nil {
-		return nil
-	}
-	q.mu.Lock()
-	head := q.head
-	q.head, q.tail = nil, nil
-	q.mu.Unlock()
-	return head
+	*ipQueue
 }
 
 func (q *jsOutQ) sendMsg(subj string, msg []byte) {
 	if q != nil {
-		q.send(&jsPubMsg{subj, _EMPTY_, _EMPTY_, nil, msg, nil, 0, nil})
+		q.send(newJSPubMsg(subj, _EMPTY_, _EMPTY_, nil, msg, nil, 0))
 	}
 }
 
@@ -3131,23 +3115,7 @@ func (q *jsOutQ) send(msg *jsPubMsg) {
 	if q == nil || msg == nil {
 		return
 	}
-	q.mu.Lock()
-	var notify bool
-	if q.head == nil {
-		q.head = msg
-		notify = true
-	} else {
-		q.tail.next = msg
-	}
-	q.tail = msg
-	q.mu.Unlock()
-
-	if notify {
-		select {
-		case q.mch <- struct{}{}:
-		default:
-		}
-	}
+	q.push(msg)
 }
 
 // StoredMsg is for raw access to messages in a stream.
@@ -3167,7 +3135,8 @@ func (mset *stream) setupSendCapabilities() {
 	if mset.outq != nil {
 		return
 	}
-	mset.outq = &jsOutQ{mch: make(chan struct{}, 1)}
+	qname := fmt.Sprintf("Stream %q send", mset.cfg.Name)
+	mset.outq = &jsOutQ{newIPQueue(ipQueue_Logger(qname, mset.srv.ipqLog))} // of *jsPubMsg
 	go mset.internalLoop()
 }
 
@@ -3191,65 +3160,22 @@ func (mset *stream) subjects() []string {
 	return copyStrings(mset.cfg.Subjects)
 }
 
-// Linked list for async ack of messages.
-// When we have a consumer to a stream that is interest based and the
-// consumer is R=1 and acknone. This is how mirrors and sources replicate.
-type ackMsgQueue struct {
-	sync.Mutex
-	mch  chan struct{}
-	seqs []uint64
-	back []uint64
-}
-
-// Push onto the queue.
-func (q *ackMsgQueue) push(seq uint64) {
-	q.Lock()
-	notify := len(q.seqs) == 0
-	q.seqs = append(q.seqs, seq)
-	q.Unlock()
-	if notify {
-		select {
-		case q.mch <- struct{}{}:
-		default:
-		}
-	}
-}
-
-// Pop all pending off.
-func (q *ackMsgQueue) pop() []uint64 {
-	q.Lock()
-	seqs := q.seqs
-	q.seqs, q.back = q.back, nil
-	q.Unlock()
-	return seqs
-}
-
-func (q *ackMsgQueue) recycle(seqs []uint64) {
-	const maxAckQueueReuse = 8 * 1024
-	if cap(seqs) > maxAckQueueReuse {
-		return
-	}
-	q.Lock()
-	q.back = seqs[:0]
-	q.Unlock()
-}
-
 func (mset *stream) internalLoop() {
 	mset.mu.RLock()
 	s := mset.srv
 	c := s.createInternalJetStreamClient()
 	c.registerWithAccount(mset.acc)
 	defer c.closeConnection(ClientClosed)
-	outq, qch, mch := mset.outq, mset.qch, mset.msgs.mch
+	outq, qch, msgs := mset.outq, mset.qch, mset.msgs
 	isClustered := mset.cfg.Replicas > 1
 
 	// For the ack msgs queue for interest retention.
 	var (
 		amch chan struct{}
-		ackq *ackMsgQueue
+		ackq *ipQueue // of uint64
 	)
 	if mset.ackq != nil {
-		ackq, amch = mset.ackq, mset.ackq.mch
+		ackq, amch = mset.ackq, mset.ackq.ch
 	}
 	mset.mu.RUnlock()
 
@@ -3258,8 +3184,10 @@ func (mset *stream) internalLoop() {
 
 	for {
 		select {
-		case <-outq.mch:
-			for pm := outq.pending(); pm != nil; pm = pm.next {
+		case <-outq.ch:
+			pms := outq.pop()
+			for _, pmi := range pms {
+				pm := pmi.(*jsPubMsg)
 				c.pa.subject = []byte(pm.subj)
 				c.pa.deliver = []byte(pm.dsubj)
 				c.pa.size = len(pm.msg) + len(pm.hdr)
@@ -3289,10 +3217,15 @@ func (mset *stream) internalLoop() {
 				if pm.o != nil && pm.seq > 0 && !didDeliver {
 					pm.o.didNotDeliver(pm.seq)
 				}
+				pm.returnToPool()
 			}
+			// TODO: Move in the for-loop?
 			c.flushClients(0)
-		case <-mch:
-			for im := mset.pending(mset.msgs); im != nil; im = im.next {
+			outq.recycle(&pms)
+		case <-msgs.ch:
+			ims := msgs.pop()
+			for _, imi := range ims {
+				im := imi.(*inMsg)
 				// If we are clustered we need to propose this message to the underlying raft group.
 				if isClustered {
 					mset.processClusteredInboundMsg(im.subj, im.rply, im.hdr, im.msg)
@@ -3300,12 +3233,13 @@ func (mset *stream) internalLoop() {
 					mset.processJetStreamMsg(im.subj, im.rply, im.hdr, im.msg, 0, 0)
 				}
 			}
+			msgs.recycle(&ims)
 		case <-amch:
 			seqs := ackq.pop()
 			for _, seq := range seqs {
-				mset.ackMsg(nil, seq)
+				mset.ackMsg(nil, seq.(uint64))
 			}
-			ackq.recycle(seqs)
+			ackq.recycle(&seqs)
 		case <-qch:
 			return
 		case <-s.quitCh:


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index 8b150e36..c648d0fe 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -3410,8 +3410,9 @@ func TestAccountLimitsServerConfig(t *testing.T) {
 
 	// Do quick test on connections, but if they are registered above should be good.
 	for i := 0; i < 5; i++ {
-		_, err := nats.Connect(s.ClientURL(), nats.UserInfo("derek", "foo"))
+		c, err := nats.Connect(s.ClientURL(), nats.UserInfo("derek", "foo"))
 		require_NoError(t, err)
+		defer c.Close()
 	}
 
 	// Should fail.
diff --git a/server/ipqueue_test.go b/server/ipqueue_test.go
new file mode 100644
index 00000000..d339536a
--- /dev/null
+++ b/server/ipqueue_test.go
@@ -0,0 +1,377 @@
+// Copyright 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
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package server
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestIPQueueBasic(t *testing.T) {
+	q := newIPQueue()
+	// Check that pool has been created
+	if q.pool == nil {
+		t.Fatal("Expected pool to have been created")
+	}
+	// Check for the default mrs
+	if q.mrs != ipQueueDefaultMaxRecycleSize {
+		t.Fatalf("Expected default max recycle size to be %v, got %v",
+			ipQueueDefaultMaxRecycleSize, q.mrs)
+	}
+	select {
+	case <-q.ch:
+		t.Fatalf("Should not have been notified")
+	default:
+		// OK!
+	}
+	if l := q.len(); l != 0 {
+		t.Fatalf("Expected len to be 0, got %v", l)
+	}
+
+	// Try to change the max recycle size
+	q = newIPQueue(ipQueue_MaxRecycleSize(10))
+	if q.mrs != 10 {
+		t.Fatalf("Expected max recycle size to be 10, got %v", q.mrs)
+	}
+}
+
+func TestIPQueuePush(t *testing.T) {
+	q := newIPQueue()
+	q.push(1)
+	if l := q.len(); l != 1 {
+		t.Fatalf("Expected len to be 1, got %v", l)
+	}
+	select {
+	case <-q.ch:
+		// OK
+	default:
+		t.Fatalf("Should have been notified of addition")
+	}
+	// Push a new element, we should not be notified.
+	q.push(2)
+	if l := q.len(); l != 2 {
+		t.Fatalf("Expected len to be 2, got %v", l)
+	}
+	select {
+	case <-q.ch:
+		t.Fatalf("Should not have been notified of addition")
+	default:
+		// OK
+	}
+}
+
+func TestIPQueuePop(t *testing.T) {
+	q := newIPQueue()
+	q.push(1)
+	<-q.ch
+	elts := q.pop()
+	if l := len(elts); l != 1 {
+		t.Fatalf("Expected 1 elt, got %v", l)
+	}
+	if l := q.len(); l != 0 {
+		t.Fatalf("Expected len to be 0, got %v", l)
+	}
+	// The channel notification should be empty
+	select {
+	case <-q.ch:
+		t.Fatalf("Should not have been notified of addition")
+	default:
+		// OK
+	}
+	// If we call pop() now, we should get an empty list.
+	if elts = q.pop(); elts != nil {
+		t.Fatalf("Expected nil, got %v", elts)
+	}
+}
+
+func TestIPQueuePopOne(t *testing.T) {
+	q := newIPQueue()
+	q.push(1)
+	<-q.ch
+	e := q.popOne()
+	if e == nil {
+		t.Fatal("Got nil")
+	}
+	if i := e.(int); i != 1 {
+		t.Fatalf("Expected 1, got %v", i)
+	}
+	if l := q.len(); l != 0 {
+		t.Fatalf("Expected len to be 0, got %v", l)
+	}
+	select {
+	case <-q.ch:
+		t.Fatalf("Should not have been notified of addition")
+	default:
+		// OK
+	}
+	q.push(2)
+	q.push(3)
+	e = q.popOne()
+	if e == nil {
+		t.Fatal("Got nil")
+	}
+	if i := e.(int); i != 2 {
+		t.Fatalf("Expected 2, got %v", i)
+	}
+	if l := q.len(); l != 1 {
+		t.Fatalf("Expected len to be 1, got %v", l)
+	}
+	select {
+	case <-q.ch:
+		// OK
+	default:
+		t.Fatalf("Should have been notified that there is more")
+	}
+	e = q.popOne()
+	if e == nil {
+		t.Fatal("Got nil")
+	}
+	if i := e.(int); i != 3 {
+		t.Fatalf("Expected 3, got %v", i)
+	}
+	if l := q.len(); l != 0 {
+		t.Fatalf("Expected len to be 0, got %v", l)
+	}
+	select {
+	case <-q.ch:
+		t.Fatalf("Should not have been notified that there is more")
+	default:
+		// OK
+	}
+	// Calling it again now that we know there is nothing, we
+	// should get nil.
+	if e = q.popOne(); e != nil {
+		t.Fatalf("Expected nil, got %v", e)
+	}
+
+	q = newIPQueue()
+	q.push(1)
+	q.push(2)
+	// Capture current capacity
+	q.RLock()
+	c := cap(q.elts)
+	q.RUnlock()
+	e = q.popOne()
+	if e == nil || e.(int) != 1 {
+		t.Fatalf("Invalid value: %v", e)
+	}
+	if l := q.len(); l != 1 {
+		t.Fatalf("Expected len to be 1, got %v", l)
+	}
+	values := q.pop()
+	if len(values) != 1 || values[0].(int) != 2 {
+		t.Fatalf("Unexpected values: %v", values)
+	}
+	if cap(values) != c-1 {
+		t.Fatalf("Unexpected capacity: %v vs %v", cap(values), c-1)
+	}
+	if l := q.len(); l != 0 {
+		t.Fatalf("Expected len to be 0, got %v", l)
+	}
+	// Just make sure that this is ok...
+	q.recycle(&values)
+}
+
+func TestIPQueueMultiProducers(t *testing.T) {
+	q := newIPQueue()
+
+	wg := sync.WaitGroup{}
+	wg.Add(3)
+	send := func(start, end int) {
+		defer wg.Done()
+
+		for i := start; i <= end; i++ {
+			q.push(i)
+		}
+	}
+	go send(1, 100)
+	go send(101, 200)
+	go send(201, 300)
+
+	tm := time.NewTimer(2 * time.Second)
+	m := make(map[int]struct{})
+	for done := false; !done; {
+		select {
+		case <-q.ch:
+			values := q.pop()
+			for _, v := range values {
+				m[v.(int)] = struct{}{}
+			}
+			q.recycle(&values)
+			done = len(m) == 300
+		case <-tm.C:
+			t.Fatalf("Did not receive all elements: %v", m)
+		}
+	}
+	wg.Wait()
+}
+
+func TestIPQueueRecycle(t *testing.T) {
+	q := newIPQueue()
+	total := 1000
+	for iter := 0; iter < 5; iter++ {
+		var sz int
+		for i := 0; i < total; i++ {
+			sz = q.push(i)
+		}
+		if sz != total {
+			t.Fatalf("Expected size to be %v, got %v", total, sz)
+		}
+		values := q.pop()
+		preRecycleCap := cap(values)
+		q.recycle(&values)
+		sz = q.push(1001)
+		if sz != 1 {
+			t.Fatalf("Expected size to be %v, got %v", 1, sz)
+		}
+		values = q.pop()
+		if l := len(values); l != 1 {
+			t.Fatalf("Len should be 1, got %v", l)
+		}
+		if c := cap(values); c == preRecycleCap {
+			break
+		} else if iter == 4 {
+			// We can't fail the test since there is no guarantee that the slice
+			// is still present in the pool when we do a Get(), but let's log that
+			// recycling did not occur even after all iterations..
+			t.Logf("Seem like the previous slice was not recycled, old cap=%v new cap=%v",
+				preRecycleCap, c)
+		}
+	}
+
+	q = newIPQueue(ipQueue_MaxRecycleSize(10))
+	for i := 0; i < 100; i++ {
+		q.push(i)
+	}
+	values := q.pop()
+	preRecycleCap := cap(values)
+	q.recycle(&values)
+	q.push(1001)
+	values = q.pop()
+	if l := len(values); l != 1 {
+		t.Fatalf("Len should be 1, got %v", l)
+	}
+	// This time, we should not have recycled it, so the new cap should
+	// be 1 for the new element added. In case Go creates a slice of
+	// cap more than 1 in some future release, just check that the
+	// cap is lower than the pre recycle cap.
+	if c := cap(values); c >= preRecycleCap {
+		t.Fatalf("The slice should not have been put back in the pool, got cap of %v", c)
+	}
+
+	// Also check that if we mistakenly pop a queue that was not
+	// notified (pop() will return nil), and we try to recycle,
+	// recycle() will ignore the call.
+	values = q.pop()
+	q.recycle(&values)
+	q.push(1002)
+	q.RLock()
+	recycled := &q.elts == &values
+	q.RUnlock()
+	if recycled {
+		t.Fatalf("Unexpected recycled slice")
+	}
+	// Check that we don't crash when recycling a nil or empty slice
+	values = q.pop()
+	q.recycle(&values)
+	q.recycle(nil)
+}
+
+func TestIPQueueDrain(t *testing.T) {
+	q := newIPQueue()
+	for iter, recycled := 0, false; iter < 5 && !recycled; iter++ {
+		for i := 0; i < 100; i++ {
+			q.push(i + 1)
+		}
+		q.drain()
+		// Try to get something from the pool right away
+		s := q.pool.Get()
+		recycled := s != nil
+		if !recycled {
+			// We can't fail the test, since we have no guarantee it will be recycled
+			// especially when running with `-race` flag...
+			if iter == 4 {
+				t.Log("nothing was recycled")
+			}
+		}
+		// Check that we have consumed the signal...
+		select {
+		case <-q.ch:
+			t.Fatal("Signal should have been consumed by drain")
+		default:
+			// OK!
+		}
+		// Check len
+		if l := q.len(); l != 0 {
+			t.Fatalf("Expected len to be 0, got %v", l)
+		}
+		if recycled {
+			break
+		}
+	}
+}
+
+type testIPQLog struct {
+	msgs []string
+}
+
+func (l *testIPQLog) log(name string, pending int) {
+	l.msgs = append(l.msgs, fmt.Sprintf("%s: %d pending", name, pending))
+}
+
+func TestIPQueueLogger(t *testing.T) {
+	l := &testIPQLog{}
+	q := newIPQueue(ipQueue_Logger("test_logger", l))
+	q.lt = 2
+	q.push(1)
+	q.push(2)
+	if len(l.msgs) != 1 {
+		t.Fatalf("Unexpected logging: %v", l.msgs)
+	}
+	if l.msgs[0] != "test_logger: 2 pending" {
+		t.Fatalf("Unexpected content: %v", l.msgs[0])
+	}
+	l.msgs = nil
+	q.push(3)
+	if len(l.msgs) != 1 {
+		t.Fatalf("Unexpected logging: %v", l.msgs)
+	}
+	if l.msgs[0] != "test_logger: 3 pending" {
+		t.Fatalf("Unexpected content: %v", l.msgs[0])
+	}
+	l.msgs = nil
+	q.popOne()
+	q.push(4)
+	if len(l.msgs) != 1 {
+		t.Fatalf("Unexpected logging: %v", l.msgs)
+	}
+	if l.msgs[0] != "test_logger: 3 pending" {
+		t.Fatalf("Unexpected content: %v", l.msgs[0])
+	}
+	l.msgs = nil
+	q.pop()
+	q.push(5)
+	if len(l.msgs) != 0 {
+		t.Fatalf("Unexpected logging: %v", l.msgs)
+	}
+	q.push(6)
+	if len(l.msgs) != 1 {
+		t.Fatalf("Unexpected logging: %v", l.msgs)
+	}
+	if l.msgs[0] != "test_logger: 2 pending" {
+		t.Fatalf("Unexpected content: %v", l.msgs[0])
+	}
+}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index bed3d90e..df7154f1 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -13792,7 +13792,8 @@ func TestJetStreamCrossAccountsDeliverSubjectInterest(t *testing.T) {
 		defer removeDir(t, config.StoreDir)
 	}
 
-	_, js := jsClientConnect(t, s, nats.UserInfo("a", "pwd"))
+	nc, js := jsClientConnect(t, s, nats.UserInfo("a", "pwd"))
+	defer nc.Close()
 
 	_, err := js.AddStream(&nats.StreamConfig{
 		Name:     "TEST",
@@ -13828,7 +13829,8 @@ func TestJetStreamCrossAccountsDeliverSubjectInterest(t *testing.T) {
 	}
 
 	// Now create interest in the delivery subject through the import on account B.
-	nc, _ := jsClientConnect(t, s, nats.UserInfo("b", "pwd"))
+	nc, _ = jsClientConnect(t, s, nats.UserInfo("b", "pwd"))
+	defer nc.Close()
 	sub, err := nc.SubscribeSync("foo")
 	require_NoError(t, err)
 	checkSubsPending(t, sub, toSend)
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index 4a9096b0..e6e4abdc 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -3131,7 +3131,7 @@ func TestMQTTClusterPlacement(t *testing.T) {
 }
 
 func TestMQTTLeafnodeWithoutJSToClusterWithJSNoSharedSysAcc(t *testing.T) {
-	test := func(resolution int) {
+	test := func(t *testing.T, resolution int) {
 		getClusterOpts := func(name string, i int) *Options {
 			o := testMQTTDefaultOptions()
 			o.ServerName = name
@@ -3214,15 +3214,17 @@ func TestMQTTLeafnodeWithoutJSToClusterWithJSNoSharedSysAcc(t *testing.T) {
 		ln := RunServer(lno)
 		defer testMQTTShutdownServer(ln)
 
+		checkLeafNodeConnected(t, ln)
+
 		// Now connect to leafnode and subscribe
-		mc, rc := testMQTTConnect(t, &mqttConnInfo{clientID: "sub", cleanSess: true}, lno.MQTT.Host, lno.MQTT.Port)
+		mc, rc := testMQTTConnectRetry(t, &mqttConnInfo{clientID: "sub", cleanSess: true}, lno.MQTT.Host, lno.MQTT.Port, 5)
 		defer mc.Close()
 		testMQTTCheckConnAck(t, rc, mqttConnAckRCConnectionAccepted, false)
 		testMQTTSub(t, 1, mc, rc, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
 		testMQTTFlush(t, mc, nil, rc)
 
 		connectAndPublish := func(o *Options) {
-			mp, rp := testMQTTConnect(t, &mqttConnInfo{cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+			mp, rp := testMQTTConnectRetry(t, &mqttConnInfo{cleanSess: true}, o.MQTT.Host, o.MQTT.Port, 5)
 			defer mp.Close()
 			testMQTTCheckConnAck(t, rp, mqttConnAckRCConnectionAccepted, false)
 
@@ -3237,7 +3239,7 @@ func TestMQTTLeafnodeWithoutJSToClusterWithJSNoSharedSysAcc(t *testing.T) {
 		testMQTTCheckPubMsg(t, mc, rc, "foo", mqttPubQos1, []byte("msg"))
 
 		// Connect from a server in the hub and subscribe
-		mc2, rc2 := testMQTTConnect(t, &mqttConnInfo{clientID: "sub2", cleanSess: true}, o2.MQTT.Host, o2.MQTT.Port)
+		mc2, rc2 := testMQTTConnectRetry(t, &mqttConnInfo{clientID: "sub2", cleanSess: true}, o2.MQTT.Host, o2.MQTT.Port, 5)
 		defer mc2.Close()
 		testMQTTCheckConnAck(t, rc2, mqttConnAckRCConnectionAccepted, false)
 		testMQTTSub(t, 1, mc2, rc2, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
@@ -3252,24 +3254,24 @@ func TestMQTTLeafnodeWithoutJSToClusterWithJSNoSharedSysAcc(t *testing.T) {
 		testMQTTCheckPubMsg(t, mc2, rc2, "foo", mqttPubQos1, []byte("msg"))
 	}
 	t.Run("backwards-compatibility-default-js-enabled-in-leaf", func(t *testing.T) {
-		test(0) // test with JsAccDefaultDomain set to default (pointing at hub) but jetstream enabled in leaf node too
+		test(t, 0) // test with JsAccDefaultDomain set to default (pointing at hub) but jetstream enabled in leaf node too
 	})
 	t.Run("backwards-compatibility-default-js-disabled-in-leaf", func(t *testing.T) {
 		// test with JsAccDefaultDomain set. Checks if it works with backwards compatibility code for empty domain
-		test(1)
+		test(t, 1)
 	})
 	t.Run("backwards-compatibility-domain-js-disabled-in-leaf", func(t *testing.T) {
 		// test with JsAccDefaultDomain set. Checks if it works with backwards compatibility code for domain set
-		test(2) // test with domain set in mqtt client
+		test(t, 2) // test with domain set in mqtt client
 	})
 	t.Run("mqtt-explicit-js-enabled-in-leaf", func(t *testing.T) {
-		test(3) // test with domain set in mqtt client (pointing at hub) but jetstream enabled in leaf node too
+		test(t, 3) // test with domain set in mqtt client (pointing at hub) but jetstream enabled in leaf node too
 	})
 	t.Run("mqtt-explicit-js-disabled-in-leaf", func(t *testing.T) {
-		test(4) // test with domain set in mqtt client
+		test(t, 4) // test with domain set in mqtt client
 	})
 	t.Run("backwards-compatibility-domain-js-enabled-in-leaf", func(t *testing.T) {
-		test(5) // test with JsAccDefaultDomain set to domain (pointing at hub) but jetstream enabled in leaf node too
+		test(t, 5) // test with JsAccDefaultDomain set to domain (pointing at hub) but jetstream enabled in leaf node too
 	})
 }
 
diff --git a/server/opts_test.go b/server/opts_test.go
index 8a579d1f..f5c912f9 100644
--- a/server/opts_test.go
+++ b/server/opts_test.go
@@ -3236,6 +3236,7 @@ func TestMaxSubTokens(t *testing.T) {
 
 	nc, err := nats.Connect(s.ClientURL())
 	require_NoError(t, err)
+	defer nc.Close()
 
 	errs := make(chan error, 1)
 
diff --git a/server/server_test.go b/server/server_test.go
index 630a305c..f1dfcd69 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -1279,6 +1279,9 @@ func TestServerShutdownDuringStart(t *testing.T) {
 	ch := make(chan struct{}, 1)
 	go func() {
 		s.Start()
+		// Since the server has already been shutdown and we don't want to leave
+		// the ipqLog run() routine running, stop it now.
+		s.ipqLog.stop()
 		close(ch)
 	}()
 	select {
@@ -1961,3 +1964,36 @@ func TestServerLogsConfigurationFile(t *testing.T) {
 		t.Fatalf("Config file location was not reported in log: %s", log)
 	}
 }
+
+func TestServerIPQueueLogger(t *testing.T) {
+	o := DefaultOptions()
+	s := RunServer(o)
+	defer s.Shutdown()
+
+	l := &captureWarnLogger{warn: make(chan string, 100)}
+	s.SetLogger(l, false, false)
+
+	q := newIPQueue(ipQueue_Logger("test", s.ipqLog))
+	// Normally, lt is immutable and set to ipQueueDefaultWarnThreshold, but
+	// for test, we set it to a low value.
+	q.lt = 2
+	q.push(1)
+	// This one should case a warning
+	q.push(2)
+
+	for {
+		select {
+		case w := <-l.warn:
+			// In case we get other warnings a runtime, just check that we
+			// get the one we expect and be done.
+			if strings.Contains(w, "test queue") {
+				if strings.Contains(w, "test queue pending size: 2") {
+					return
+				}
+				t.Fatalf("Invalid warning: %v", w)
+			}
+		case <-time.After(time.Second):
+			t.Fatal("Did not get warning")
+		}
+	}
+}