Repository: nats-io/nats-server
Pull request #5270 (2024-04-03 02:36:27)

[IMPROVED] Improvements to interest policy and workqueue clustered streams.
Specifically around interest and limited streams that are clustered. Since we operate consumers and streams independently, when a stream is removing msgs when acked and acks are coming into different servers at different times, etc. Messages can get stranded.

We introduce a better check of the interest state based on a collective low ack floor that will purge our copy of the stream to catch up. We run this periodically in monitorStream and during certain conditions.

Thanks to Wally for introducing the test that stresses this part of the system.

Signed-off-by: Derek Collison <derek@nats.io>




Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 0c01da94..ec98c8ae 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1392,6 +1392,10 @@ func (o *consumer) setLeader(isLeader bool) {
 		} else if o.srv.gateway.enabled {
 			stopAndClearTimer(&o.gwdtmr)
 		}
+		// If we were the leader make sure to drain queued up acks.
+		if wasLeader {
+			o.ackMsgs.drain()
+		}
 		o.mu.Unlock()
 
 		// Unregister as a leader with our parent stream.
@@ -2534,19 +2538,18 @@ func (o *consumer) applyState(state *ConsumerState) {
 		return
 	}
 
-	// If o.sseq is greater don't update. Don't go backwards on o.sseq.
-	if o.sseq <= state.Delivered.Stream {
+	// If o.sseq is greater don't update. Don't go backwards on o.sseq if leader.
+	if !o.isLeader() || o.sseq <= state.Delivered.Stream {
 		o.sseq = state.Delivered.Stream + 1
 	}
 	o.dseq = state.Delivered.Consumer + 1
-
 	o.adflr = state.AckFloor.Consumer
 	o.asflr = state.AckFloor.Stream
 	o.pending = state.Pending
 	o.rdc = state.Redelivered
 
 	// Setup tracking timer if we have restored pending.
-	if len(o.pending) > 0 {
+	if o.isLeader() && len(o.pending) > 0 {
 		// This is on startup or leader change. We want to check pending
 		// sooner in case there are inconsistencies etc. Pick between 500ms - 1.5s
 		delay := 500*time.Millisecond + time.Duration(rand.Int63n(1000))*time.Millisecond
@@ -2788,6 +2791,12 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 		return
 	}
 
+	mset := o.mset
+	if mset == nil || mset.closed.Load() {
+		o.mu.Unlock()
+		return
+	}
+
 	var sagap uint64
 	var needSignal bool
 
@@ -2803,17 +2812,20 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 			delete(o.pending, sseq)
 			// Use the original deliver sequence from our pending record.
 			dseq = p.Sequence
-		}
-		if len(o.pending) == 0 {
-			o.adflr, o.asflr = o.dseq-1, o.sseq-1
-		} else if dseq == o.adflr+1 {
-			o.adflr, o.asflr = dseq, sseq
-			for ss := sseq + 1; ss < o.sseq; ss++ {
-				if p, ok := o.pending[ss]; ok {
-					if p.Sequence > 0 {
-						o.adflr, o.asflr = p.Sequence-1, ss-1
+			// Only move floors if we matched an existing pending.
+			if dseq == o.adflr+1 {
+				o.adflr, o.asflr = dseq, sseq
+				for ss := sseq + 1; ss < o.sseq; ss++ {
+					if p, ok := o.pending[ss]; ok {
+						if p.Sequence > 0 {
+							o.adflr, o.asflr = p.Sequence-1, ss-1
+						}
+						break
 					}
-					break
+				}
+				// If nothing left set to current delivered.
+				if len(o.pending) == 0 {
+					o.adflr, o.asflr = o.dseq-1, o.sseq-1
 				}
 			}
 		}
@@ -2845,7 +2857,6 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 	// Update underlying store.
 	o.updateAcks(dseq, sseq)
 
-	mset := o.mset
 	clustered := o.node != nil
 
 	// In case retention changes for a stream, this ought to have been updated
@@ -3092,10 +3103,16 @@ func (wq *waitQueue) add(wr *waitingRequest) error {
 }
 
 func (wq *waitQueue) isFull() bool {
+	if wq == nil {
+		return false
+	}
 	return wq.n == wq.max
 }
 
 func (wq *waitQueue) isEmpty() bool {
+	if wq == nil {
+		return true
+	}
 	return wq.n == 0
 }
 
@@ -3627,7 +3644,6 @@ func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {
 				break
 			}
 		}
-
 	}
 
 	// Don't sort the o.subjf if it's only one entry
@@ -4034,8 +4050,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		pmsg, dc, err = o.getNextMsg()
 
 		// We can release the lock now under getNextMsg so need to check this condition again here.
-		// consumer is closed when mset is set to nil.
-		if o.mset == nil {
+		if o.closed || o.mset == nil {
 			o.mu.Unlock()
 			return
 		}
@@ -4286,50 +4301,55 @@ func (o *consumer) streamNumPending() uint64 {
 		o.npc, o.npf = 0, 0
 		return 0
 	}
+	npc, npf := o.calculateNumPending()
+	o.npc, o.npf = int64(npc), npf
+	return o.numPending()
+}
+
+// Will calculate num pending but only requires a read lock.
+// Depends on delivery policy, for last per subject we calculate differently.
+// At least RLock should be held.
+func (o *consumer) calculateNumPending() (npc, npf uint64) {
+	if o.mset == nil || o.mset.store == nil {
+		return 0, 0
+	}
 
 	isLastPerSubject := o.cfg.DeliverPolicy == DeliverLastPerSubject
 
 	// Deliver Last Per Subject calculates num pending differently.
 	if isLastPerSubject {
-		o.npc, o.npf = 0, 0
 		// Consumer without filters.
 		if o.subjf == nil {
-			npc, npf := o.mset.store.NumPending(o.sseq, _EMPTY_, isLastPerSubject)
-			o.npc, o.npf = int64(npc), npf
-			return o.numPending()
+			return o.mset.store.NumPending(o.sseq, _EMPTY_, isLastPerSubject)
 		}
 		// Consumer with filters.
 		for _, filter := range o.subjf {
-			npc, npf := o.mset.store.NumPending(o.sseq, filter.subject, isLastPerSubject)
-			o.npc += int64(npc)
-			if npf > o.npf {
-				o.npf = npf // Always last
+			lnpc, lnpf := o.mset.store.NumPending(o.sseq, filter.subject, isLastPerSubject)
+			npc += lnpc
+			if lnpf > npf {
+				npf = lnpf // Always last
 			}
 		}
-		return o.numPending()
+		return npc, npf
 	}
 	// Every other Delivery Policy is handled here.
 	// Consumer without filters.
 	if o.subjf == nil {
-		npc, npf := o.mset.store.NumPending(o.sseq, o.cfg.FilterSubject, isLastPerSubject)
-		o.npc, o.npf = int64(npc), npf
-		return o.numPending()
+		return o.mset.store.NumPending(o.sseq, _EMPTY_, false)
 	}
 	// Consumer with filters.
-	o.npc, o.npf = 0, 0
 	for _, filter := range o.subjf {
 		// We might loose state of o.subjf, so if we do recover from o.sseq
 		if filter.currentSeq < o.sseq {
 			filter.currentSeq = o.sseq
 		}
-		npc, npf := o.mset.store.NumPending(filter.currentSeq, filter.subject, isLastPerSubject)
-		o.npc += int64(npc)
-		if npf > o.npf {
-			o.npf = npf // Always last
+		lnpc, lnpf := o.mset.store.NumPending(filter.currentSeq, filter.subject, false)
+		npc += lnpc
+		if lnpf > npf {
+			npf = lnpf // Always last
 		}
 	}
-
-	return o.numPending()
+	return npc, npf
 }
 
 func convertToHeadersOnly(pmsg *jsPubMsg) {
@@ -4381,6 +4401,10 @@ func (o *consumer) deliverMsg(dsubj, ackReply string, pmsg *jsPubMsg, dc uint64,
 
 	// Cant touch pmsg after this sending so capture what we need.
 	seq, ts := pmsg.seq, pmsg.ts
+
+	// Update delivered first.
+	o.updateDelivered(dseq, seq, dc, ts)
+
 	// Send message.
 	o.outq.send(pmsg)
 
@@ -4401,9 +4425,6 @@ func (o *consumer) deliverMsg(dsubj, ackReply string, pmsg *jsPubMsg, dc uint64,
 		o.waiting.last = time.Now()
 	}
 
-	// FIXME(dlc) - Capture errors?
-	o.updateDelivered(dseq, seq, dc, ts)
-
 	// If we are ack none and mset is interest only we should make sure stream removes interest.
 	if ap == AckNone && rp != LimitsPolicy {
 		if o.node == nil || o.cfg.Direct {
@@ -4646,7 +4667,11 @@ func (o *consumer) checkPending() {
 	check := len(o.pending) > 1024
 	for seq, p := range o.pending {
 		if check && atomic.LoadInt64(&o.awl) > 0 {
-			o.ptmr.Reset(100 * time.Millisecond)
+			if o.ptmr == nil {
+				o.ptmr = time.AfterFunc(100*time.Millisecond, o.checkPending)
+			} else {
+				o.ptmr.Reset(100 * time.Millisecond)
+			}
 			return
 		}
 		// Check if these are no longer valid.
@@ -5391,7 +5416,7 @@ func (o *consumer) decStreamPending(sseq uint64, subj string) {
 
 	// If it was pending process it like an ack.
 	if wasPending {
-		// We could have lock for stream so do this in a go routine.
+		// We could have the lock for the stream so do this in a go routine.
 		// TODO(dlc) - We should do this with ipq vs naked go routines.
 		go o.processTerm(sseq, p.Sequence, rdc, ackTermUnackedLimitsReason)
 	}
@@ -5549,7 +5574,7 @@ func (o *consumer) checkStateForInterestStream() error {
 		return errAckFloorHigherThanLastSeq
 	}
 
-	for seq := ss.FirstSeq; seq <= asflr; seq++ {
+	for seq := ss.FirstSeq; asflr > 0 && seq <= asflr; seq++ {
 		mset.ackMsg(o, seq)
 	}
 
diff --git a/server/jetstream.go b/server/jetstream.go
index ab6debda..2b877ad8 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1410,10 +1410,6 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 			if !cfg.Created.IsZero() {
 				obs.setCreatedTime(cfg.Created)
 			}
-			lseq := e.mset.lastSeq()
-			obs.mu.Lock()
-			err = obs.readStoredState(lseq)
-			obs.mu.Unlock()
 			if err != nil {
 				s.Warnf("    Error restoring consumer %q state: %v", cfg.Name, err)
 			}
@@ -2290,8 +2286,8 @@ func (jsa *jsAccount) delete() {
 	jsa.templates = nil
 	jsa.mu.Unlock()
 
-	for _, ms := range streams {
-		ms.stop(false, false)
+	for _, mset := range streams {
+		mset.stop(false, false)
 	}
 
 	for _, t := range ts {
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 0b5d5634..9aaf90fc 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -1946,7 +1946,17 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
 			return
 		}
 	}
-	config := mset.config()
+
+	mset.mu.RLock()
+	config := mset.cfg
+	checkAcks := config.Retention != LimitsPolicy && mset.hasLimitsSet()
+	mset.mu.RUnlock()
+
+	// Check if we are a clustered interest retention stream with limits.
+	// If so, check ack floors against our state.
+	if checkAcks {
+		mset.checkInterestState()
+	}
 
 	resp.StreamInfo = &StreamInfo{
 		Created:    mset.createdTime(),
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index be8c35b0..3bb10d14 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -8471,6 +8471,7 @@ func (mset *stream) processClusterStreamInfoRequest(reply string) {
 	mset.mu.RLock()
 	sysc, js, sa, config := mset.sysc, mset.srv.js.Load(), mset.sa, mset.cfg
 	isLeader := mset.isLeader()
+	checkAcks := isLeader && config.Retention != LimitsPolicy && mset.isClustered() && mset.hasLimitsSet()
 	mset.mu.RUnlock()
 
 	// By design all members will receive this. Normally we only want the leader answering.
@@ -8484,6 +8485,12 @@ func (mset *stream) processClusterStreamInfoRequest(reply string) {
 		time.Sleep(500 * time.Millisecond)
 	}
 
+	// Check if we are a clustered interest retention stream with limits.
+	// If so, check ack floors against our state.
+	if checkAcks {
+		mset.checkInterestState()
+	}
+
 	si := &StreamInfo{
 		Created:   mset.createdTime(),
 		State:     mset.state(),
diff --git a/server/mqtt.go b/server/mqtt.go
index 2d90fc8f..df054d1b 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -190,18 +190,6 @@ const (
 	mqttRetainedTransferTimeout = 10 * time.Second
 )
 
-const (
-	sparkbNBIRTH = "NBIRTH"
-	sparkbDBIRTH = "DBIRTH"
-	sparkbNDEATH = "NDEATH"
-	sparkbDDEATH = "DDEATH"
-)
-
-var (
-	sparkbNamespaceTopicPrefix    = []byte("spBv1.0/")
-	sparkbCertificatesTopicPrefix = []byte("$sparkplug/certificates/")
-)
-
 var (
 	mqttPingResponse     = []byte{mqttPacketPingResp, 0x0}
 	mqttProtoName        = []byte("MQTT")
@@ -467,24 +455,14 @@ type mqttPublish struct {
 // When we submit a PUBREL for delivery, we add a "Nmqtt-PubRel" header that
 // contains the PI.
 const (
-	// NATS header that indicates that the message originated from MQTT and
-	// stores the published message QOS.
-	mqttNatsHeader = "Nmqtt-Pub"
-
-	// NATS headers to store retained message metadata (along with the original
-	// message as binary).
+	mqttNatsHeader                = "Nmqtt-Pub"
 	mqttNatsRetainedMessageTopic  = "Nmqtt-RTopic"
 	mqttNatsRetainedMessageOrigin = "Nmqtt-ROrigin"
 	mqttNatsRetainedMessageFlags  = "Nmqtt-RFlags"
 	mqttNatsRetainedMessageSource = "Nmqtt-RSource"
-
-	// NATS header that indicates that the message is an MQTT PubRel and stores
-	// the PI.
-	mqttNatsPubRelHeader = "Nmqtt-PubRel"
-
-	// NATS headers to store the original MQTT subject and the subject mapping.
-	mqttNatsHeaderSubject = "Nmqtt-Subject"
-	mqttNatsHeaderMapped  = "Nmqtt-Mapped"
+	mqttNatsPubRelHeader          = "Nmqtt-PubRel"
+	mqttNatsHeaderSubject         = "Nmqtt-Subject"
+	mqttNatsHeaderMapped          = "Nmqtt-Mapped"
 )
 
 type mqttParsedPublishNATSHeader struct {
@@ -1661,16 +1639,13 @@ func (jsa *mqttJSA) newRequestExMulti(kind, subject, cidHash string, hdrs []int,
 }
 
 func (jsa *mqttJSA) sendAck(ackSubject string) {
-	// 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.
-	jsa.sendMsg(ackSubject, nil)
-}
-
-func (jsa *mqttJSA) sendMsg(subj string, msg []byte) {
-	if subj == _EMPTY_ {
+	if ackSubject == _EMPTY_ {
 		return
 	}
-	jsa.sendq.push(&mqttJSPubMsg{subj: subj, msg: msg, hdr: -1})
+
+	// 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.
+	jsa.sendq.push(&mqttJSPubMsg{subj: ackSubject, hdr: -1})
 }
 
 func (jsa *mqttJSA) createEphemeralConsumer(cfg *CreateConsumerRequest) (*JSApiConsumerCreateResponse, error) {
@@ -3633,7 +3608,7 @@ func (c *client) mqttParseConnect(r *mqttReader, hasMappings bool) (byte, *mqttC
 				cp.will.mapped = c.pa.mapped
 				// We also now need to map the original MQTT topic to the new topic
 				// based on the new subject.
-				topic = natsSubjectToMQTTTopic(cp.will.subject)
+				topic = natsSubjectToMQTTTopic(string(cp.will.subject))
 			}
 			// Reset those now.
 			c.pa.subject, c.pa.mapped = nil, nil
@@ -3982,7 +3957,7 @@ func (c *client) mqttParsePub(r *mqttReader, pl int, pp *mqttPublish, hasMapping
 			pp.mapped = c.pa.mapped
 			// We also now need to map the original MQTT topic to the new topic
 			// based on the new subject.
-			pp.topic = natsSubjectToMQTTTopic(pp.subject)
+			pp.topic = natsSubjectToMQTTTopic(string(pp.subject))
 		}
 		// Reset those now.
 		c.pa.subject, c.pa.mapped = nil, nil
@@ -4038,7 +4013,7 @@ func mqttNewDeliverableMessage(pp *mqttPublish, encodePP bool) (natsMsg []byte,
 	size := len(hdrLine) +
 		len(mqttNatsHeader) + 2 + 2 + // 2 for ':<qos>', and 2 for CRLF
 		2 + // end-of-header CRLF
-		pp.sz
+		len(pp.msg)
 	if encodePP {
 		size += len(mqttNatsHeaderSubject) + 1 + // +1 for ':'
 			len(pp.subject) + 2 // 2 for CRLF
@@ -4246,7 +4221,7 @@ func (s *Server) mqttProcessPubRel(c *client, pi uint16, trace bool) error {
 	}
 
 	pp := &mqttPublish{
-		topic:   natsSubjectToMQTTTopic(h.subject),
+		topic:   natsSubjectToMQTTTopic(string(h.subject)),
 		subject: h.subject,
 		mapped:  h.mapped,
 		msg:     stored.Data,
@@ -4265,106 +4240,52 @@ func (s *Server) mqttProcessPubRel(c *client, pi uint16, trace bool) error {
 // Invoked from the MQTT publisher's readLoop. No client lock is held on entry.
 func (c *client) mqttHandlePubRetain() {
 	pp := c.mqtt.pp
-	retainMQTT := mqttIsRetained(pp.flags)
-	isBirth, _, isCertificate := sparkbParseBirthDeathTopic(pp.topic)
-	retainSparkbBirth := isBirth && !isCertificate
-
-	// [tck-id-topics-nbirth-mqtt] NBIRTH messages MUST be published with MQTT
-	// QoS equal to 0 and retain equal to false.
-	//
-	// [tck-id-conformance-mqtt-aware-nbirth-mqtt-retain] A Sparkplug Aware MQTT
-	// Server MUST make NBIRTH messages available on the topic:
-	// $sparkplug/certificates/namespace/group_id/NBIRTH/edge_node_id with the
-	// MQTT retain flag set to true.
-	if retainMQTT == retainSparkbBirth {
-		// (retainSparkbBirth && retainMQTT) : not valid, so ignore altogether.
-		// (!retainSparkbBirth && !retainMQTT) : nothing to do.
+	if !mqttIsRetained(pp.flags) {
 		return
 	}
-
-	asm := c.mqtt.asm
 	key := string(pp.subject)
-
-	// Always clear the retain flag to deliver a normal published message.
-	defer func() {
-		pp.flags &= ^mqttPubFlagRetain
-	}()
-
-	// Spec [MQTT-3.3.1-11]. Payload of size 0 removes the retained message, but
-	// should still be delivered as a normal message.
+	asm := c.mqtt.asm
+	// Spec [MQTT-3.3.1-11]. Payload of size 0 removes the retained message,
+	// but should still be delivered as a normal message.
 	if pp.sz == 0 {
 		if seqToRemove := asm.handleRetainedMsgDel(key, 0); seqToRemove > 0 {
 			asm.deleteRetainedMsg(seqToRemove)
 			asm.notifyRetainedMsgDeleted(key, seqToRemove)
 		}
-		return
-	}
-
-	rm := &mqttRetainedMsg{
-		Origin: asm.jsa.id,
-		Msg:    pp.msg, // will copy these bytes later as we process rm.
-		Flags:  pp.flags,
-		Source: c.opts.Username,
-	}
-
-	if retainSparkbBirth {
-		// [tck-id-conformance-mqtt-aware-store] A Sparkplug Aware MQTT Server
-		// MUST store NBIRTH and DBIRTH messages as they pass through the MQTT
-		// Server.
-		//
-		// [tck-id-conformance-mqtt-aware-nbirth-mqtt-topic]. A Sparkplug Aware
-		// MQTT Server MUST make NBIRTH messages available on a topic of the
-		// form: $sparkplug/certificates/namespace/group_id/NBIRTH/edge_node_id
-		//
-		// [tck-id-conformance-mqtt-aware-dbirth-mqtt-topic] A Sparkplug Aware
-		// MQTT Server MUST make DBIRTH messages available on a topic of the
-		// form:
-		// $sparkplug/certificates/namespace/group_id/DBIRTH/edge_node_id/device_id
-		topic := append(sparkbCertificatesTopicPrefix, pp.topic...)
-		subject, _ := mqttTopicToNATSPubSubject(topic)
-		rm.Topic = string(topic)
-		rm.Subject = string(subject)
-
-		// will use to save the retained message.
-		key = string(subject)
-
-		// Store the retained message with the RETAIN flag set.
-		rm.Flags |= mqttPubFlagRetain
-
-		// Copy the payload out of pp since we will be sending the message
-		// asynchronously.
-		msg := make([]byte, pp.sz)
-		copy(msg, pp.msg[:pp.sz])
-		asm.jsa.sendMsg(key, msg)
-
-	} else { // isRetained
+	} else {
 		// Spec [MQTT-3.3.1-5]. Store the retained message with its QoS.
 		//
 		// When coming from a publish protocol, `pp` is referencing a stack
 		// variable that itself possibly references the read buffer.
-		rm.Topic = string(pp.topic)
-	}
-
-	// Set the key to the subject of the message for retained, or the composed
-	// $sparkplug subject for sparkB.
-	rm.Subject = key
-	rmBytes, hdr := mqttEncodeRetainedMessage(rm) // will copy the payload bytes
-	smr, err := asm.jsa.storeMsg(mqttRetainedMsgsStreamSubject+key, hdr, rmBytes)
-	if err == nil {
-		// Update the new sequence.
-		rf := &mqttRetainedMsgRef{
-			sseq: smr.Sequence,
+		rm := &mqttRetainedMsg{
+			Origin:  asm.jsa.id,
+			Subject: key,
+			Topic:   string(pp.topic),
+			Msg:     pp.msg, // will copy these bytes later as we process rm.
+			Flags:   pp.flags,
+			Source:  c.opts.Username,
+		}
+		rmBytes, hdr := mqttEncodeRetainedMessage(rm) // will copy the payload bytes
+		smr, err := asm.jsa.storeMsg(mqttRetainedMsgsStreamSubject+key, hdr, rmBytes)
+		if err == nil {
+			// Update the new sequence.
+			rf := &mqttRetainedMsgRef{
+				sseq: smr.Sequence,
+			}
+			// Add/update the map. `true` to copy the payload bytes if needs to
+			// update rmsCache.
+			asm.handleRetainedMsg(key, rf, rm, true)
+		} else {
+			c.mu.Lock()
+			acc := c.acc
+			c.mu.Unlock()
+			c.Errorf("unable to store retained message for account %q, subject %q: %v",
+				acc.GetName(), key, err)
 		}
-		// Add/update the map. `true` to copy the payload bytes if needs to
-		// update rmsCache.
-		asm.handleRetainedMsg(key, rf, rm, true)
-	} else {
-		c.mu.Lock()
-		acc := c.acc
-		c.mu.Unlock()
-		c.Errorf("unable to store retained message for account %q, subject %q: %v",
-			acc.GetName(), key, err)
 	}
+
+	// Clear the retain flag for a normal published message.
+	pp.flags &= ^mqttPubFlagRetain
 }
 
 // After a config reload, it is possible that the source of a publish retained
@@ -4628,32 +4549,6 @@ func mqttIsRetained(flags byte) bool {
 	return flags&mqttPubFlagRetain != 0
 }
 
-func sparkbParseBirthDeathTopic(topic []byte) (isBirth, isDeath, isCertificate bool) {
-	if bytes.HasPrefix(topic, sparkbCertificatesTopicPrefix) {
-		isCertificate = true
-		topic = topic[len(sparkbCertificatesTopicPrefix):]
-	}
-	if !bytes.HasPrefix(topic, sparkbNamespaceTopicPrefix) {
-		return false, false, false
-	}
-	topic = topic[len(sparkbNamespaceTopicPrefix):]
-
-	parts := bytes.Split(topic, []byte{'/'})
-	if len(parts) < 3 || len(parts) > 4 {
-		return false, false, false
-	}
-	typ := bytesToString(parts[1])
-	switch typ {
-	case sparkbNBIRTH, sparkbDBIRTH:
-		isBirth = true
-	case sparkbNDEATH, sparkbDDEATH:
-		isDeath = true
-	default:
-		return false, false, false
-	}
-	return isBirth, isDeath, isCertificate
-}
-
 //////////////////////////////////////////////////////////////////////////////
 //
 // SUBSCRIBE related functions
@@ -4801,16 +4696,18 @@ func mqttDeliverMsgCbQoS0(sub *subscription, pc *client, _ *Account, subject, re
 		topic = pc.mqtt.pp.topic
 		// Check for service imports where subject mapping is in play.
 		if len(pc.pa.mapped) > 0 && len(pc.pa.psi) > 0 {
-			topic = natsSubjectStrToMQTTTopic(subject)
+			topic = natsSubjectToMQTTTopic(subject)
 		}
 
 	} else {
 		// Non MQTT client, could be NATS publisher, or ROUTER, etc..
 		h := mqttParsePublishNATSHeader(hdr)
 
-		// Check the subscription's QoS. If the message was published with a
-		// QoS>0 (in the header) and the sub has the QoS>0 then the message will
-		// be delivered by mqttDeliverMsgCbQoS12.
+		// If the message does not have the MQTT header, it is not a MQTT and
+		// should be delivered here, at QOS0. If it does have the header, we
+		// need to lock the session to check the sub QoS, and then ignore the
+		// message if the Sub wants higher QOS delivery. It will be delivered by
+		// mqttDeliverMsgCbQoS12.
 		if subQoS > 0 && h != nil && h.qos > 0 {
 			return
 		}
@@ -4820,7 +4717,7 @@ func mqttDeliverMsgCbQoS0(sub *subscription, pc *client, _ *Account, subject, re
 		if len(msg) > mqttMaxPayloadSize {
 			msg = msg[:mqttMaxPayloadSize]
 		}
-		topic = natsSubjectStrToMQTTTopic(subject)
+		topic = natsSubjectToMQTTTopic(subject)
 	}
 
 	// Message never has a packet identifier nor is marked as duplicate.
@@ -4882,7 +4779,7 @@ func mqttDeliverMsgCbQoS12(sub *subscription, pc *client, _ *Account, subject, r
 
 	// Check for reserved subject violation. If so, we will send the ack to
 	// remove the message, and do nothing else.
-	strippedSubj := subject[len(mqttStreamSubjectPrefix):]
+	strippedSubj := string(subject[len(mqttStreamSubjectPrefix):])
 	if mqttMustIgnoreForReservedSub(sub, strippedSubj) {
 		sess.mu.Unlock()
 		sess.jsa.sendAck(reply)
@@ -4899,7 +4796,7 @@ func mqttDeliverMsgCbQoS12(sub *subscription, pc *client, _ *Account, subject, r
 		return
 	}
 
-	originalTopic := natsSubjectStrToMQTTTopic(strippedSubj)
+	originalTopic := natsSubjectToMQTTTopic(strippedSubj)
 	pc.mqttEnqueuePublishMsgTo(cc, sub, pi, qos, dup, originalTopic, msg)
 }
 
@@ -4957,78 +4854,10 @@ func isMQTTReservedSubscription(subject string) bool {
 	return false
 }
 
-func sparkbReplaceDeathTimestamp(msg []byte) []byte {
-	const VARINT = 0
-	const TIMESTAMP = 1
-
-	orig := msg
-	buf := bytes.NewBuffer(make([]byte, 0, len(msg)+16)) // 16 bytes should be enough if we need to add a timestamp
-	writeDeathTimestamp := func() {
-		// [tck-id-conformance-mqtt-aware-ndeath-timestamp] A Sparkplug Aware
-		// MQTT Server MAY replace the timestamp of NDEATH messages. If it does,
-		// it MUST set the timestamp to the UTC time at which it attempts to
-		// deliver the NDEATH to subscribed clients
-		//
-		// sparkB spec: 6.4.1. Google Protocol Buffer Schema
-		//      optional uint64 timestamp = 1; // Timestamp at message sending time
-		//
-		// SparkplugB timestamps are milliseconds since epoch, represented as
-		// uint64 in go, transmitted as protobuf varint.
-		ts := uint64(time.Now().UnixMilli())
-		buf.Write(protoEncodeVarint(TIMESTAMP<<3 | VARINT))
-		buf.Write(protoEncodeVarint(ts))
-	}
-
-	for len(msg) > 0 {
-		fieldNumericID, fieldType, size, err := protoScanField(msg)
-		if err != nil {
-			return orig
-		}
-		if fieldType != VARINT || fieldNumericID != TIMESTAMP {
-			// Add the field as is
-			buf.Write(msg[:size])
-			msg = msg[size:]
-			continue
-		}
-
-		writeDeathTimestamp()
-
-		// Add the rest of the message as is, we are done
-		buf.Write(msg[size:])
-		return buf.Bytes()
-	}
-
-	// Add timestamp if we did not find one.
-	writeDeathTimestamp()
-
-	return buf.Bytes()
-}
-
 // Common function to mqtt delivery callbacks to serialize and send the message
 // to the `cc` client.
 func (c *client) mqttEnqueuePublishMsgTo(cc *client, sub *subscription, pi uint16, qos byte, dup bool, topic, msg []byte) {
-	// [tck-id-conformance-mqtt-aware-nbirth-mqtt-retain] A Sparkplug Aware
-	// MQTT Server MUST make NBIRTH messages available on the topic:
-	// $sparkplug/certificates/namespace/group_id/NBIRTH/edge_node_id with
-	// the MQTT retain flag set to true
-	//
-	// [tck-id-conformance-mqtt-aware-dbirth-mqtt-retain] A Sparkplug Aware
-	// MQTT Server MUST make DBIRTH messages available on the topic:
-	// $sparkplug/certificates/namespace/group_id/DBIRTH/edge_node_id/device_id
-	// with the MQTT retain flag set to true
-	//
-	// $sparkplug/certificates messages are sent as NATS messages, so we
-	// need to add the retain flag when sending them to MQTT ciients.
-
-	retain := false
-	isBirth, isDeath, isCertificate := sparkbParseBirthDeathTopic(topic)
-	if isBirth && qos == 0 {
-		retain = isCertificate
-	} else if isDeath && !isCertificate {
-		msg = sparkbReplaceDeathTimestamp(msg)
-	}
-
-	flags, headerBytes := mqttMakePublishHeader(pi, qos, dup, retain, topic, len(msg))
+	flags, headerBytes := mqttMakePublishHeader(pi, qos, dup, false, topic, len(msg))
 
 	cc.mu.Lock()
 	if sub.mqtt.prm != nil {
@@ -5587,12 +5416,8 @@ func mqttToNATSSubjectConversion(mt []byte, wcOk bool) ([]byte, error) {
 // Converts a NATS subject to MQTT topic. This is for publish
 // messages only, so there is no checking for wildcards.
 // Rules are reversed of mqttToNATSSubjectConversion.
-func natsSubjectStrToMQTTTopic(subject string) []byte {
-	return natsSubjectToMQTTTopic(stringToBytes(subject))
-}
-
-func natsSubjectToMQTTTopic(subject []byte) []byte {
-	topic := make([]byte, len(subject))
+func natsSubjectToMQTTTopic(subject string) []byte {
+	topic := []byte(subject)
 	end := len(subject) - 1
 	var j int
 	for i := 0; i < len(subject); i++ {
diff --git a/server/parser.go b/server/parser.go
index dabbe257..101e352d 100644
--- a/server/parser.go
+++ b/server/parser.go
@@ -495,10 +495,8 @@ func (c *client) parse(buf []byte) error {
 			// Check for mappings.
 			if (c.kind == CLIENT || c.kind == LEAF) && c.in.flags.isSet(hasMappings) {
 				changed := c.selectMappedSubject()
-				if changed {
-					if trace {
-						c.traceInOp("MAPPING", []byte(fmt.Sprintf("%s -> %s", c.pa.mapped, c.pa.subject)))
-					}
+				if (trace || mt != nil) && changed {
+					c.traceInOp("MAPPING", []byte(fmt.Sprintf("%s -> %s", c.pa.mapped, c.pa.subject)))
 					// c.pa.subject is the subject the original is now mapped to.
 					mt.addSubjectMappingEvent(c.pa.subject)
 				}
diff --git a/server/proto.go b/server/proto.go
deleted file mode 100644
index 9843fff2..00000000
--- a/server/proto.go
+++ /dev/null
@@ -1,269 +0,0 @@
-// Copyright 2024 The NATS Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// 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.
-
-// Inspired by https://github.com/protocolbuffers/protobuf-go/blob/master/encoding/protowire/wire.go
-
-package server
-
-import (
-	"errors"
-	"fmt"
-	"math"
-)
-
-var errProtoInsufficient = errors.New("insufficient data to read a value")
-var errProtoOverflow = errors.New("too much data for a value")
-var errProtoInvalidFieldNumber = errors.New("invalid field number")
-
-func protoScanField(b []byte) (num, typ, size int, err error) {
-	num, typ, sizeTag, err := protoScanTag(b)
-	if err != nil {
-		return 0, 0, 0, err
-	}
-	b = b[sizeTag:]
-
-	sizeValue, err := protoScanFieldValue(typ, b)
-	if err != nil {
-		return 0, 0, 0, err
-	}
-	return num, typ, sizeTag + sizeValue, nil
-}
-
-func protoScanTag(b []byte) (num, typ, size int, err error) {
-	tagint, size, err := protoScanVarint(b)
-	if err != nil {
-		return 0, 0, 0, err
-	}
-
-	// NOTE: MessageSet allows for larger field numbers than normal.
-	if (tagint >> 3) > uint64(math.MaxInt32) {
-		return 0, 0, 0, errProtoInvalidFieldNumber
-	}
-	num = int(tagint >> 3)
-	if num < 1 {
-		return 0, 0, 0, errProtoInvalidFieldNumber
-	}
-	typ = int(tagint & 7)
-
-	return num, typ, size, nil
-}
-
-func protoScanFieldValue(typ int, b []byte) (size int, err error) {
-	switch typ {
-	case 0:
-		_, size, err = protoScanVarint(b)
-	case 5: // fixed32
-		size = 4
-	case 1: // fixed64
-		size = 8
-	case 2: // length-delimited
-		size, err = protoScanBytes(b)
-	default:
-		return 0, fmt.Errorf("unsupported type: %d", typ)
-	}
-	return size, err
-}
-
-func protoScanVarint(b []byte) (v uint64, size int, err error) {
-	var y uint64
-	if len(b) <= 0 {
-		return 0, 0, errProtoInsufficient
-	}
-	v = uint64(b[0])
-	if v < 0x80 {
-		return v, 1, nil
-	}
-	v -= 0x80
-
-	if len(b) <= 1 {
-		return 0, 0, errProtoInsufficient
-	}
-	y = uint64(b[1])
-	v += y << 7
-	if y < 0x80 {
-		return v, 2, nil
-	}
-	v -= 0x80 << 7
-
-	if len(b) <= 2 {
-		return 0, 0, errProtoInsufficient
-	}
-	y = uint64(b[2])
-	v += y << 14
-	if y < 0x80 {
-		return v, 3, nil
-	}
-	v -= 0x80 << 14
-
-	if len(b) <= 3 {
-		return 0, 0, errProtoInsufficient
-	}
-	y = uint64(b[3])
-	v += y << 21
-	if y < 0x80 {
-		return v, 4, nil
-	}
-	v -= 0x80 << 21
-
-	if len(b) <= 4 {
-		return 0, 0, errProtoInsufficient
-	}
-	y = uint64(b[4])
-	v += y << 28
-	if y < 0x80 {
-		return v, 5, nil
-	}
-	v -= 0x80 << 28
-
-	if len(b) <= 5 {
-		return 0, 0, errProtoInsufficient
-	}
-	y = uint64(b[5])
-	v += y << 35
-	if y < 0x80 {
-		return v, 6, nil
-	}
-	v -= 0x80 << 35
-
-	if len(b) <= 6 {
-		return 0, 0, errProtoInsufficient
-	}
-	y = uint64(b[6])
-	v += y << 42
-	if y < 0x80 {
-		return v, 7, nil
-	}
-	v -= 0x80 << 42
-
-	if len(b) <= 7 {
-		return 0, 0, errProtoInsufficient
-	}
-	y = uint64(b[7])
-	v += y << 49
-	if y < 0x80 {
-		return v, 8, nil
-	}
-	v -= 0x80 << 49
-
-	if len(b) <= 8 {
-		return 0, 0, errProtoInsufficient
-	}
-	y = uint64(b[8])
-	v += y << 56
-	if y < 0x80 {
-		return v, 9, nil
-	}
-	v -= 0x80 << 56
-
-	if len(b) <= 9 {
-		return 0, 0, errProtoInsufficient
-	}
-	y = uint64(b[9])
-	v += y << 63
-	if y < 2 {
-		return v, 10, nil
-	}
-	return 0, 0, errProtoOverflow
-}
-
-func protoScanBytes(b []byte) (size int, err error) {
-	l, lenSize, err := protoScanVarint(b)
-	if err != nil {
-		return 0, err
-	}
-	if l > uint64(len(b[lenSize:])) {
-		return 0, errProtoInsufficient
-	}
-	return lenSize + int(l), nil
-}
-
-func protoEncodeVarint(v uint64) []byte {
-	b := make([]byte, 0, 10)
-	switch {
-	case v < 1<<7:
-		b = append(b, byte(v))
-	case v < 1<<14:
-		b = append(b,
-			byte((v>>0)&0x7f|0x80),
-			byte(v>>7))
-	case v < 1<<21:
-		b = append(b,
-			byte((v>>0)&0x7f|0x80),
-			byte((v>>7)&0x7f|0x80),
-			byte(v>>14))
-	case v < 1<<28:
-		b = append(b,
-			byte((v>>0)&0x7f|0x80),
-			byte((v>>7)&0x7f|0x80),
-			byte((v>>14)&0x7f|0x80),
-			byte(v>>21))
-	case v < 1<<35:
-		b = append(b,
-			byte((v>>0)&0x7f|0x80),
-			byte((v>>7)&0x7f|0x80),
-			byte((v>>14)&0x7f|0x80),
-			byte((v>>21)&0x7f|0x80),
-			byte(v>>28))
-	case v < 1<<42:
-		b = append(b,
-			byte((v>>0)&0x7f|0x80),
-			byte((v>>7)&0x7f|0x80),
-			byte((v>>14)&0x7f|0x80),
-			byte((v>>21)&0x7f|0x80),
-			byte((v>>28)&0x7f|0x80),
-			byte(v>>35))
-	case v < 1<<49:
-		b = append(b,
-			byte((v>>0)&0x7f|0x80),
-			byte((v>>7)&0x7f|0x80),
-			byte((v>>14)&0x7f|0x80),
-			byte((v>>21)&0x7f|0x80),
-			byte((v>>28)&0x7f|0x80),
-			byte((v>>35)&0x7f|0x80),
-			byte(v>>42))
-	case v < 1<<56:
-		b = append(b,
-			byte((v>>0)&0x7f|0x80),
-			byte((v>>7)&0x7f|0x80),
-			byte((v>>14)&0x7f|0x80),
-			byte((v>>21)&0x7f|0x80),
-			byte((v>>28)&0x7f|0x80),
-			byte((v>>35)&0x7f|0x80),
-			byte((v>>42)&0x7f|0x80),
-			byte(v>>49))
-	case v < 1<<63:
-		b = append(b,
-			byte((v>>0)&0x7f|0x80),
-			byte((v>>7)&0x7f|0x80),
-			byte((v>>14)&0x7f|0x80),
-			byte((v>>21)&0x7f|0x80),
-			byte((v>>28)&0x7f|0x80),
-			byte((v>>35)&0x7f|0x80),
-			byte((v>>42)&0x7f|0x80),
-			byte((v>>49)&0x7f|0x80),
-			byte(v>>56))
-	default:
-		b = append(b,
-			byte((v>>0)&0x7f|0x80),
-			byte((v>>7)&0x7f|0x80),
-			byte((v>>14)&0x7f|0x80),
-			byte((v>>21)&0x7f|0x80),
-			byte((v>>28)&0x7f|0x80),
-			byte((v>>35)&0x7f|0x80),
-			byte((v>>42)&0x7f|0x80),
-			byte((v>>49)&0x7f|0x80),
-			byte((v>>56)&0x7f|0x80),
-			1)
-	}
-	return b
-}
diff --git a/server/stream.go b/server/stream.go
index 93d8d0ae..0462a967 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -268,6 +268,7 @@ type stream struct {
 	sch   chan struct{}
 	sigq  *ipQueue[*cMsg]
 	csl   *Sublist // Consumer Sublist
+
 	// Leader will store seq/msgTrace in clustering mode. Used in applyStreamEntries
 	// to know if trace event should be sent after processing.
 	mt map[uint64]*msgTrace
@@ -859,6 +860,12 @@ func (mset *stream) setLeader(isLeader bool) error {
 		mset.leader = _EMPTY_
 	}
 	mset.mu.Unlock()
+
+	if mset.isInterestRetention() {
+		// If we are interest based make sure to check consumers.
+		// This is to make sure we process any outstanding acks.
+		mset.checkInterestState()
+	}
 	return nil
 }
 
@@ -3983,6 +3990,12 @@ func (mset *stream) isClustered() bool {
 	return mset.node != nil
 }
 
+// Return if any limits are set.
+// Lock should be held.
+func (mset *stream) hasLimitsSet() bool {
+	return mset.cfg.MaxMsgs > 0 || mset.cfg.MaxBytes > 0 || mset.cfg.MaxMsgsPer > 0 || mset.cfg.MaxAge > 0
+}
+
 // Used if we have to queue things internally to avoid the route/gw path.
 type inMsg struct {
 	subj string
@@ -5230,9 +5243,7 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 	accName := jsa.account.Name
 	jsa.mu.Unlock()
 
-	// Mark as closed, kick monitor and collect consumers first.
-	mset.closed.Store(true)
-
+	// Kick monitor and collect consumers first.
 	mset.mu.Lock()
 	// Signal to the monitor loop.
 	// Can't use qch here.
@@ -5294,6 +5305,9 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 		mset.sendDeleteAdvisoryLocked()
 	}
 
+	// Mark closed.
+	mset.closed.Store(true)
+
 	// Quit channel, do this after sending the delete advisory
 	if mset.qch != nil {
 		close(mset.qch)
@@ -5425,24 +5439,87 @@ func (mset *stream) getPublicConsumers() []*consumer {
 }
 
 // Will check for interest retention and make sure messages
-// that have been acked are processed.
+// that have been acked are processed and removed.
+// This will check the ack floors of all consumers, and adjust our first sequence accordingly.
 func (mset *stream) checkInterestState() {
 	if mset == nil {
 		return
 	}
-	mset.mu.RLock()
-	// If we are limits based nothing to do.
-	if mset.cfg.Retention == LimitsPolicy {
-		mset.mu.RUnlock()
+	if !mset.isInterestRetention() {
+		// If we are limits based nothing to do.
 		return
 	}
-	consumers := make([]*consumer, 0, len(mset.consumers))
-	for _, o := range mset.consumers {
-		consumers = append(consumers, o)
+
+	mset.clsMu.RLock()
+	var zeroAcks []*consumer
+	var lowAckFloor uint64 = math.MaxUint64
+	for _, o := range mset.cList {
+		o.mu.RLock()
+		if o.isLeader() {
+			// We need to account for consumers with ack floor of zero.
+			// We will collect them and see if we need to check pending below.
+			if o.asflr == 0 {
+				zeroAcks = append(zeroAcks, o)
+			} else if o.asflr < lowAckFloor {
+				lowAckFloor = o.asflr
+			}
+		} else {
+			// We are a follower so only have the store state, so read that in.
+			state, err := o.store.State()
+			if err != nil {
+				// On error we will not have enough information to process correctly so bail.
+				o.mu.RUnlock()
+				mset.clsMu.RUnlock()
+				return
+			}
+			// We need to account for consumers with ack floor of zero.
+			if state.AckFloor.Stream == 0 {
+				zeroAcks = append(zeroAcks, o)
+			} else if state.AckFloor.Stream < lowAckFloor {
+				lowAckFloor = state.AckFloor.Stream
+			}
+			// We are a follower here but if we detect a drift from when we were previous leader correct here.
+			if o.asflr > state.AckFloor.Stream || o.sseq > state.Delivered.Stream+1 {
+				o.applyState(state)
+			}
+		}
+		o.mu.RUnlock()
 	}
-	mset.mu.RUnlock()
-	for _, o := range consumers {
-		o.checkStateForInterestStream()
+	mset.clsMu.RUnlock()
+
+	// If nothing was set we can bail.
+	if lowAckFloor == math.MaxUint64 {
+		return
+	}
+
+	// Hold stream write lock in case we need to purge.
+	mset.mu.Lock()
+	defer mset.mu.Unlock()
+
+	// Capture our current state.
+	var state StreamState
+	mset.store.FastState(&state)
+
+	if lowAckFloor < math.MaxUint64 && lowAckFloor > state.FirstSeq {
+		// Check if we had any zeroAcks, we will need to check them.
+		for _, o := range zeroAcks {
+			var np uint64
+			o.mu.RLock()
+			if o.isLeader() {
+				np = uint64(o.numPending())
+			} else {
+				np, _ = o.calculateNumPending()
+			}
+			o.mu.RUnlock()
+			// This means we have pending and can not remove anything at this time.
+			if np > 0 {
+				return
+			}
+		}
+		// Purge the stream to lowest ack floor + 1
+		mset.store.PurgeEx(_EMPTY_, lowAckFloor+1, 0)
+		// Also make sure we clear any pending acks.
+		mset.clearAllPreAcksBelowFloor(lowAckFloor + 1)
 	}
 }
 
@@ -5805,19 +5882,22 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
 		return
 	}
 
+	store := mset.store
 	var state StreamState
-	mset.store.FastState(&state)
+	store.FastState(&state)
 
-	// Make sure this sequence is not below our first sequence.
-	if seq < state.FirstSeq {
-		mset.clearPreAck(o, seq)
+	// If this has arrived before we have processed the message itself.
+	if seq > state.LastSeq {
+		mset.registerPreAck(o, seq)
 		mset.mu.Unlock()
 		return
 	}
 
-	// If this has arrived before we have processed the message itself.
-	if seq > state.LastSeq {
-		mset.registerPreAck(o, seq)
+	// Always clear pre-ack if here.
+	mset.clearPreAck(o, seq)
+
+	// Make sure this sequence is not below our first sequence.
+	if seq < state.FirstSeq {
 		mset.mu.Unlock()
 		return
 	}
@@ -5839,7 +5919,7 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
 	}
 
 	// If we are here we should attempt to remove.
-	if _, err := mset.store.RemoveMsg(seq); err == ErrStoreEOF {
+	if _, err := store.RemoveMsg(seq); err == ErrStoreEOF {
 		// This should not happen, but being pedantic.
 		mset.registerPreAckLock(o, seq)
 	}
diff --git a/server/sysmem/sysctl.go b/server/sysmem/sysctl.go
index 82946f02..a7fa266a 100644
--- a/server/sysmem/sysctl.go
+++ b/server/sysmem/sysctl.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2024 The NATS Authors
+// Copyright 2019 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
@@ -26,9 +26,7 @@ func sysctlInt64(name string) int64 {
 	if err != nil {
 		return 0
 	}
-	// Make sure it's 8 bytes when we do the cast below.
-	// We were getting fatal error: checkptr: converted pointer straddles multiple allocations in go 1.22.1 on darwin.
-	var b [8]byte
-	copy(b[:], s)
+	// hack because the string conversion above drops a \0
+	b := []byte(s)
 	return *(*int64)(unsafe.Pointer(&b[0]))
 }


Test patch:
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 92492a58..7e16aa58 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -35,6 +35,7 @@ import (
 
 	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nuid"
 )
 
 func TestJetStreamClusterRemovePeerByID(t *testing.T) {
@@ -7037,3 +7038,347 @@ func TestJetStreamClusterConsumerPauseSurvivesRestart(t *testing.T) {
 	require_True(t, leader != nil)
 	checkTimer(leader)
 }
+
+func TestJetStreamClusterWorkQueueStreamOrphanIssue(t *testing.T) {
+	t.Run("R1F", func(t *testing.T) {
+		testJetStreamClusterWorkQueueStreamOrphanIssue(t, &nats.StreamConfig{
+			Name:        "OWQTEST_R1F",
+			Subjects:    []string{"MSGS.>"},
+			Replicas:    1,
+			MaxAge:      30 * time.Minute,
+			Duplicates:  5 * time.Minute,
+			Retention:   nats.WorkQueuePolicy,
+			Discard:     nats.DiscardOld,
+			AllowRollup: true,
+			Placement: &nats.Placement{
+				Tags: []string{"test"},
+			},
+		})
+	})
+	t.Run("R3M", func(t *testing.T) {
+		testJetStreamClusterWorkQueueStreamOrphanIssue(t, &nats.StreamConfig{
+			Name:        "OWQTEST_R3M",
+			Subjects:    []string{"MSGS.>"},
+			Replicas:    3,
+			MaxAge:      30 * time.Minute,
+			MaxMsgs:     100_000,
+			Duplicates:  5 * time.Minute,
+			Retention:   nats.WorkQueuePolicy,
+			Discard:     nats.DiscardNew,
+			AllowRollup: true,
+			Storage:     nats.MemoryStorage,
+			Placement: &nats.Placement{
+				Tags: []string{"test"},
+			},
+		})
+	})
+	t.Run("R3F_DN", func(t *testing.T) {
+		testJetStreamClusterWorkQueueStreamOrphanIssue(t, &nats.StreamConfig{
+			Name:        "OWQTEST_R3F_DN",
+			Subjects:    []string{"MSGS.>"},
+			Replicas:    3,
+			MaxAge:      30 * time.Minute,
+			MaxMsgs:     100_000,
+			Duplicates:  5 * time.Minute,
+			Retention:   nats.WorkQueuePolicy,
+			Discard:     nats.DiscardNew,
+			AllowRollup: true,
+			Placement: &nats.Placement{
+				Tags: []string{"test"},
+			},
+		})
+	})
+	t.Run("R3F_DO", func(t *testing.T) {
+		testJetStreamClusterWorkQueueStreamOrphanIssue(t, &nats.StreamConfig{
+			Name:        "OWQTEST_R3F_DO",
+			Subjects:    []string{"MSGS.>"},
+			Replicas:    3,
+			MaxAge:      30 * time.Minute,
+			MaxMsgs:     100_000,
+			Duplicates:  5 * time.Minute,
+			Retention:   nats.WorkQueuePolicy,
+			Discard:     nats.DiscardOld,
+			AllowRollup: true,
+			Placement: &nats.Placement{
+				Tags: []string{"test"},
+			},
+		})
+	})
+}
+
+func testJetStreamClusterWorkQueueStreamOrphanIssue(t *testing.T, sc *nats.StreamConfig) {
+	conf := `
+	listen: 127.0.0.1:-1
+	server_name: %s
+	jetstream: {
+		store_dir: '%s',
+	}
+	cluster {
+		name: %s
+		listen: 127.0.0.1:%d
+		routes = [%s]
+	}
+	server_tags: ["test"]
+	system_account: sys
+	no_auth_user: js
+	accounts {
+		sys { users = [ { user: sys, pass: sys } ] }
+		js {
+			jetstream = enabled
+			users = [ { user: js, pass: js } ]
+	    }
+	}`
+	c := createJetStreamClusterWithTemplate(t, conf, sc.Name, 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	cnc, cjs := jsClientConnect(t, c.randomServer())
+	defer cnc.Close()
+
+	_, err := js.AddStream(sc)
+	require_NoError(t, err)
+
+	pctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	// Start producers
+	var wg sync.WaitGroup
+
+	// First call is just to create the pull subscribers.
+	mp := nats.MaxAckPending(10000)
+	mw := nats.PullMaxWaiting(1000)
+	aw := nats.AckWait(5 * time.Second)
+
+	for i := 0; i < 10; i++ {
+		for _, partition := range []string{"EEEEE"} {
+			subject := fmt.Sprintf("MSGS.%s.*.H.100XY.*.*.WQ.00000000000%d", partition, i)
+			consumer := fmt.Sprintf("consumer:%s:%d", partition, i)
+			_, err := cjs.PullSubscribe(subject, consumer, mp, mw, aw)
+			require_NoError(t, err)
+		}
+	}
+
+	// Create a single consumer that does no activity.
+	// Make sure we still calculate low ack properly and cleanup etc.
+	_, err = cjs.PullSubscribe("MSGS.ZZ.>", "consumer:ZZ:0", mp, mw, aw)
+	require_NoError(t, err)
+
+	subjects := []string{
+		"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000000",
+		"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000001",
+		"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000002",
+		"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000003",
+		"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000004",
+		"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000005",
+		"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000006",
+		"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000007",
+		"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000008",
+		"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000009",
+	}
+	payload := []byte(strings.Repeat("A", 1024))
+
+	for i := 0; i < 50; i++ {
+		wg.Add(1)
+		go func() {
+			pnc, pjs := jsClientConnect(t, c.randomServer())
+			defer pnc.Close()
+
+			for i := 1; i < 200_000; i++ {
+				select {
+				case <-pctx.Done():
+					wg.Done()
+					return
+				default:
+				}
+				for _, subject := range subjects {
+					// Send each message a few times.
+					msgID := nats.MsgId(nuid.Next())
+					pjs.PublishAsync(subject, payload, msgID)
+					pjs.Publish(subject, payload, msgID, nats.AckWait(250*time.Millisecond))
+					pjs.Publish(subject, payload, msgID, nats.AckWait(250*time.Millisecond))
+				}
+			}
+			t.Logf("Stopped publishing.")
+		}()
+	}
+
+	// Rogue publisher that sends the same msg ID everytime.
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			pnc, pjs := jsClientConnect(t, c.randomServer())
+			defer pnc.Close()
+
+			msgID := nats.MsgId("1234567890")
+			for i := 1; ; i++ {
+				select {
+				case <-pctx.Done():
+					wg.Done()
+					return
+				default:
+				}
+				for _, subject := range subjects {
+					// Send each message a few times.
+					pjs.PublishAsync(subject, payload, msgID, nats.RetryAttempts(0), nats.RetryWait(0))
+					pjs.Publish(subject, payload, msgID, nats.AckWait(1*time.Millisecond), nats.RetryAttempts(0), nats.RetryWait(0))
+					pjs.Publish(subject, payload, msgID, nats.AckWait(1*time.Millisecond), nats.RetryAttempts(0), nats.RetryWait(0))
+				}
+			}
+		}()
+	}
+
+	// Let enough messages into the stream then start consumers.
+	time.Sleep(15 * time.Second)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
+	defer cancel()
+
+	for i := 0; i < 10; i++ {
+		subject := fmt.Sprintf("MSGS.EEEEE.*.H.100XY.*.*.WQ.00000000000%d", i)
+		consumer := fmt.Sprintf("consumer:EEEEE:%d", i)
+		for n := 0; n < 5; n++ {
+			cpnc, cpjs := jsClientConnect(t, c.randomServer())
+			defer cpnc.Close()
+
+			psub, err := cpjs.PullSubscribe(subject, consumer, mp)
+			require_NoError(t, err)
+
+			time.AfterFunc(15*time.Second, func() {
+				cpnc.Close()
+			})
+
+			wg.Add(1)
+			go func() {
+				tick := time.NewTicker(1 * time.Millisecond)
+				for {
+					if cpnc.IsClosed() {
+						wg.Done()
+						return
+					}
+					select {
+					case <-ctx.Done():
+						wg.Done()
+						return
+					case <-tick.C:
+						// Fetch 1 first, then if no errors Fetch 100.
+						msgs, err := psub.Fetch(1, nats.MaxWait(200*time.Millisecond))
+						if err != nil {
+							continue
+						}
+						for _, msg := range msgs {
+							msg.Ack()
+						}
+						msgs, err = psub.Fetch(100, nats.MaxWait(200*time.Millisecond))
+						if err != nil {
+							continue
+						}
+						for _, msg := range msgs {
+							msg.Ack()
+						}
+						msgs, err = psub.Fetch(1000, nats.MaxWait(200*time.Millisecond))
+						if err != nil {
+							continue
+						}
+						for _, msg := range msgs {
+							msg.Ack()
+						}
+					}
+				}
+			}()
+		}
+	}
+
+	for i := 0; i < 10; i++ {
+		subject := fmt.Sprintf("MSGS.EEEEE.*.H.100XY.*.*.WQ.00000000000%d", i)
+		consumer := fmt.Sprintf("consumer:EEEEE:%d", i)
+		for n := 0; n < 10; n++ {
+			cpnc, cpjs := jsClientConnect(t, c.randomServer())
+			defer cpnc.Close()
+
+			psub, err := cpjs.PullSubscribe(subject, consumer, mp)
+			if err != nil {
+				t.Logf("ERROR: %v", err)
+				continue
+			}
+
+			wg.Add(1)
+			go func() {
+				tick := time.NewTicker(1 * time.Millisecond)
+				for {
+					select {
+					case <-ctx.Done():
+						wg.Done()
+						return
+					case <-tick.C:
+						// Fetch 1 first, then if no errors Fetch 100.
+						msgs, err := psub.Fetch(1, nats.MaxWait(200*time.Millisecond))
+						if err != nil {
+							continue
+						}
+						for _, msg := range msgs {
+							msg.Ack()
+						}
+						msgs, err = psub.Fetch(100, nats.MaxWait(200*time.Millisecond))
+						if err != nil {
+							continue
+						}
+						for _, msg := range msgs {
+							msg.Ack()
+						}
+
+						msgs, err = psub.Fetch(1000, nats.MaxWait(200*time.Millisecond))
+						if err != nil {
+							continue
+						}
+						for _, msg := range msgs {
+							msg.Ack()
+						}
+					}
+				}
+			}()
+		}
+	}
+
+	time.AfterFunc(10*time.Second, func() {
+		if sc.Replicas == 1 {
+			// Find server leader of the stream and restart it.
+			leaderSrv := c.streamLeader("js", sc.Name)
+			leaderSrv.Shutdown()
+			leaderSrv.WaitForShutdown()
+			c.restartServer(leaderSrv)
+		} else {
+			// NOTE (wq): For R=3, not sure which server causes the issue here
+			// so this may be have flaky behavior.
+			s := c.servers[0]
+			s.optsMu.Lock()
+			s.opts.LameDuckDuration = 5 * time.Second
+			s.opts.LameDuckGracePeriod = -5 * time.Second
+			s.optsMu.Unlock()
+			s.lameDuckMode()
+			s.WaitForShutdown()
+			c.restartServer(s)
+			c.waitOnClusterReady()
+		}
+	})
+
+	// Wait until context is done then check state.
+	<-ctx.Done()
+
+	var consumerPending int
+	for i := 0; i < 10; i++ {
+		ci, err := js.ConsumerInfo(sc.Name, fmt.Sprintf("consumer:EEEEE:%d", i))
+		require_NoError(t, err)
+		consumerPending += int(ci.NumPending)
+	}
+
+	// Check state of streams and consumers.
+	si, err := js.StreamInfo(sc.Name)
+	require_NoError(t, err)
+
+	streamPending := int(si.State.Msgs)
+	if streamPending != consumerPending {
+		t.Fatalf("Unexpected number of pending messages, stream=%d, consumers=%d", streamPending, consumerPending)
+	}
+}
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index a92607cc..3c70ec1b 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -1820,7 +1820,7 @@ func TestMQTTTopicAndSubjectConversion(t *testing.T) {
 				t.Fatalf("Expected subject %q got %q", test.natsSubject, toNATS)
 			}
 
-			res = natsSubjectToMQTTTopic(res)
+			res = natsSubjectToMQTTTopic(string(res))
 			backToMQTT := string(res)
 			if backToMQTT != test.mqttTopic {
 				t.Fatalf("Expected topic %q got %q (NATS conversion was %q)", test.mqttTopic, backToMQTT, toNATS)
@@ -2047,11 +2047,11 @@ func testMQTTCheckPubMsgNoAck(t testing.TB, c net.Conn, r *mqttReader, topic str
 
 func testMQTTGetPubMsg(t testing.TB, c net.Conn, r *mqttReader, topic string, payload []byte) (byte, uint16) {
 	t.Helper()
-	flags, pi, _, _ := testMQTTGetPubMsgEx(t, c, r, topic, payload)
+	flags, pi, _ := testMQTTGetPubMsgEx(t, c, r, topic, payload)
 	return flags, pi
 }
 
-func testMQTTGetPubMsgEx(t testing.TB, _ net.Conn, r *mqttReader, topic string, payload []byte) (byte, uint16, string, []byte) {
+func testMQTTGetPubMsgEx(t testing.TB, _ net.Conn, r *mqttReader, topic string, payload []byte) (byte, uint16, string) {
 	t.Helper()
 	b, pl := testMQTTReadPacket(t, r)
 	if pt := b & mqttPacketMask; pt != mqttPacketPub {
@@ -2060,7 +2060,7 @@ func testMQTTGetPubMsgEx(t testing.TB, _ net.Conn, r *mqttReader, topic string,
 	return testMQTTGetPubMsgExEx(t, nil, r, b, pl, topic, payload)
 }
 
-func testMQTTGetPubMsgExEx(t testing.TB, _ net.Conn, r *mqttReader, b byte, pl int, topic string, payload []byte) (byte, uint16, string, []byte) {
+func testMQTTGetPubMsgExEx(t testing.TB, _ net.Conn, r *mqttReader, b byte, pl int, topic string, payload []byte) (byte, uint16, string) {
 	t.Helper()
 	pflags := b & mqttPacketFlagMask
 	qos := (pflags & mqttPubFlagQoS) >> 1
@@ -2089,7 +2089,7 @@ func testMQTTGetPubMsgExEx(t testing.TB, _ net.Conn, r *mqttReader, b byte, pl i
 		t.Fatalf("Expected payload %q, got %q", payload, ppayload)
 	}
 	r.pos += msgLen
-	return pflags, pi, ptopic, ppayload
+	return pflags, pi, ptopic
 }
 
 func testMQTTReadPubPacket(t testing.TB, r *mqttReader) (flags byte, pi uint16, topic string, payload []byte) {
@@ -3347,7 +3347,7 @@ func TestMQTTRetainedMsgMigration(t *testing.T) {
 	testMQTTSub(t, 1, mc, rc, []*mqttFilter{{filter: "+", qos: 0}}, []byte{0})
 	topics := map[string]struct{}{}
 	for i := 0; i < N; i++ {
-		_, _, topic, _ := testMQTTGetPubMsgEx(t, mc, rc, _EMPTY_, []byte("bar"))
+		_, _, topic := testMQTTGetPubMsgEx(t, mc, rc, _EMPTY_, []byte("bar"))
 		topics[topic] = struct{}{}
 	}
 	if len(topics) != N {
@@ -7455,6 +7455,7 @@ func TestMQTTJetStreamRepublishAndQoS0Subscribers(t *testing.T) {
 	testMQTTExpectNothing(t, r)
 }
 
+// Test for auto-cleanup of consumers.
 func TestMQTTDecodeRetainedMessage(t *testing.T) {
 	tdir := t.TempDir()
 	tmpl := `
@@ -7519,7 +7520,7 @@ func TestMQTTDecodeRetainedMessage(t *testing.T) {
 		if pt := b & mqttPacketMask; pt != mqttPacketPub {
 			t.Fatalf("Expected PUBLISH packet %x, got %x", mqttPacketPub, pt)
 		}
-		_, _, topic, _ := testMQTTGetPubMsgExEx(t, mc, r, mqttPubFlagRetain, pl, "", nil)
+		_, _, topic := testMQTTGetPubMsgExEx(t, mc, r, mqttPubFlagRetain, pl, "", nil)
 		if string(topic) != "foo/1" && string(topic) != "foo/2" {
 			t.Fatalf("Expected foo/1 or foo/2, got %q", topic)
 		}
@@ -7543,178 +7544,6 @@ func TestMQTTDecodeRetainedMessage(t *testing.T) {
 	testMQTTExpectNothing(t, r)
 }
 
-func TestMQTTSparkbDeathHandling(t *testing.T) {
-	o := testMQTTDefaultOptions()
-	s := testMQTTRunServer(t, o)
-	defer testMQTTShutdownServer(s)
-
-	// protoMetrics is protobuf-encoded test data, with no timestamp.
-	//
-	//	p := &sproto.Payload{
-	// 		Metrics: []*sproto.Payload_Metric{
-	// 			newSparkbMetric_Uint64(sparkbBDSeqMetric, uint64(sparkbBDSeq)),
-	// 			newSparkbMetric(sparkbDeviceMetric1, "value-at-death-1"),
-	// 			newSparkbMetric(sparkbDeviceMetric2, "value-at-death-2"),
-	// 		},
-	// 		Timestamp: nil,
-	// 	}
-	protoMetrics := []byte{0x12, 0x0b, 0x0a, 0x05, 0x62, 0x64, 0x53, 0x65, 0x71, 0x20, 0x04, 0x58, 0x02, 0x12, 0x24, 0x0a, 0x0e, 0x64, 0x65, 0x76, 0x69, 0x63, 0x65, 0x2d, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x31, 0x20, 0x0c, 0x7a, 0x10, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x2d, 0x61, 0x74, 0x2d, 0x64, 0x65, 0x61, 0x74, 0x68, 0x2d, 0x31, 0x12, 0x24, 0x0a, 0x0e, 0x64, 0x65, 0x76, 0x69, 0x63, 0x65, 0x2d, 0x6d, 0x65, 0x74, 0x72, 0x69, 0x63, 0x32, 0x20, 0x0c, 0x7a, 0x10, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x2d, 0x61, 0x74, 0x2d, 0x64, 0x65, 0x61, 0x74, 0x68, 0x2d, 0x32}
-
-	proto0Timestamp := []byte{0x08, 0x00}
-
-	protoDeadbeefTimestamp := []byte{0x08, 0xef, 0xfd, 0xb6, 0xf5, 0xfd, 0xde, 0xef, 0xd6, 0xde, 0x01}
-
-	mcPub, rPub := testMQTTConnect(t, &mqttConnInfo{cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
-	defer mcPub.Close()
-	testMQTTCheckConnAck(t, rPub, mqttConnAckRCConnectionAccepted, false)
-
-	mcSub, rSub := testMQTTConnect(t, &mqttConnInfo{cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
-	defer mcSub.Close()
-	testMQTTCheckConnAck(t, rSub, mqttConnAckRCConnectionAccepted, false)
-	testMQTTSub(t, 0, mcSub, rSub, []*mqttFilter{{filter: "spBv1.0/ggg/#", qos: 0}}, []byte{0})
-
-	for _, test := range []*struct {
-		name                  string
-		payload               []byte
-		expectTimestampOffset int
-		expectUnchanged       bool
-	}{
-		{"append", protoMetrics, len(protoMetrics), false},
-		{"replace at the end", append(protoMetrics, proto0Timestamp...), len(protoMetrics), false},
-		{"replace at the start", append(protoDeadbeefTimestamp, protoMetrics...), 0, false},
-		{"invalid", []byte{0xde, 0xad, 0xbe, 0xef}, 0, true}, // invalid payload
-	} {
-		var sendPI uint16
-		for _, topic := range []string{"NDEATH/nnn", "DDEATH/nnn/ddd"} {
-			sendPI++
-			t.Run(test.name+" "+topic, func(t *testing.T) {
-				testMQTTPublish(t, mcPub, rPub, 1, false, false, "spBv1.0/ggg/"+topic, sendPI, test.payload)
-
-				flags, pi, _, received := testMQTTGetPubMsgEx(t, mcSub, rSub, "", nil)
-				if mqttGetQoS(flags) == 1 {
-					testMQTTSendPIPacket(mqttPacketPubAck, t, mcSub, pi)
-				}
-
-				if test.expectUnchanged {
-					if !bytes.Equal(test.payload, received) {
-						t.Fatalf("Expected payload to be unchanged, got %q", received)
-					}
-					return
-				}
-
-				if test.expectTimestampOffset > len(received) {
-					t.Fatalf("Expected timestamp offset %v to be less than %v", test.expectTimestampOffset, len(received))
-				}
-				if test.expectTimestampOffset > 0 {
-					if !bytes.Equal(protoMetrics, received[:test.expectTimestampOffset]) {
-						t.Fatalf("Expected payload to be unchanged up to %v, got %q", test.expectTimestampOffset, received)
-					}
-				}
-
-				tsBytes := received[test.expectTimestampOffset:]
-				if tsBytes[0] != 0x08 {
-					t.Fatalf("Expected timestamp to start with 0x08, got %x", tsBytes)
-				}
-				v, _, err := protoScanVarint(tsBytes[1:])
-				if err != nil {
-					t.Fatalf("Error scanning varint: %v", err)
-				}
-
-				ts := time.UnixMilli(int64(v))
-				if time.Now().Before(ts) {
-					t.Fatalf("Expected timestamp to be in the past, got %v", ts)
-				}
-				if time.Since(ts) > time.Second {
-					t.Fatalf("Expected timestamp to be within a second, got %v", time.Since(ts))
-				}
-			})
-		}
-	}
-}
-
-func TestMQTTSparkbBirthHandling(t *testing.T) {
-	o := testMQTTDefaultOptions()
-	s := testMQTTRunServer(t, o)
-	defer testMQTTShutdownServer(s)
-
-	// Publish an NBIRTH message. Make sure it is received as both the original
-	// subject, and as a $sparkplug/certificates one.
-
-	const NBORN = "NODE BORN FAKE MESSAGE"
-	const DBORN = "DEVICE BORN FAKE MESSAGE"
-
-	tests := []*struct {
-		topic   string
-		payload string
-		flags   byte
-		mc      net.Conn
-		r       *mqttReader
-	}{
-		{
-			topic:   "spBv1.0/ggg/NBIRTH/nnn",
-			payload: NBORN,
-			flags:   0,
-		},
-		{
-			topic:   "spBv1.0/ggg/DBIRTH/nnn/ddd",
-			payload: DBORN,
-			flags:   0,
-		},
-		{
-			topic:   "$sparkplug/certificates/spBv1.0/ggg/NBIRTH/nnn",
-			payload: NBORN,
-			flags:   mqttPubFlagRetain,
-		},
-		{
-			topic:   "$sparkplug/certificates/spBv1.0/ggg/DBIRTH/nnn/ddd",
-			payload: DBORN,
-			flags:   mqttPubFlagRetain,
-		},
-	}
-
-	t.Run("$sparkplug messages delivered", func(t *testing.T) {
-		// Connect the subscribers, unique for each topic we monitor.
-		for i, test := range tests {
-			test.mc, test.r = testMQTTConnect(t, &mqttConnInfo{cleanSess: true, clientID: fmt.Sprintf("sub-%v", i)}, o.MQTT.Host, o.MQTT.Port)
-			defer test.mc.Close()
-			testMQTTCheckConnAck(t, test.r, mqttConnAckRCConnectionAccepted, false)
-
-			// Subscribne at QoS2 to make sure the messages are posted at QoS0 and
-			// not truncated to sub QoS.
-			testMQTTSub(t, 0, test.mc, test.r, []*mqttFilter{{filter: test.topic, qos: 2}}, []byte{2})
-		}
-
-		// connect the publisher
-		mc, r := testMQTTConnect(t, &mqttConnInfo{cleanSess: true, clientID: "pub"}, o.MQTT.Host, o.MQTT.Port)
-		defer mc.Close()
-		testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
-
-		// {Publish} [ND]BIRTH messages. Note that IRL these are protobuf payloads.
-		testMQTTPublish(t, mc, r, 0, false, false, "spBv1.0/ggg/NBIRTH/nnn", 0, []byte("NODE BORN FAKE MESSAGE"))
-		testMQTTPublish(t, mc, r, 0, false, false, "spBv1.0/ggg/DBIRTH/nnn/ddd", 0, []byte("DEVICE BORN FAKE MESSAGE"))
-
-		// Check that the subscribers got the messages.
-		for _, test := range tests {
-			testMQTTCheckPubMsg(t, test.mc, test.r, test.topic, test.flags, []byte(test.payload))
-		}
-	})
-
-	t.Run("$sparkplug messages retained", func(t *testing.T) {
-		// Connect/subscribe again, and make sure that only retained messages are there.
-		for i, test := range tests {
-			mc, r := testMQTTConnect(t, &mqttConnInfo{cleanSess: true, clientID: fmt.Sprintf("sub-%v", i+100)}, o.MQTT.Host, o.MQTT.Port)
-			defer mc.Close()
-			testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
-			testMQTTSub(t, 0, mc, r, []*mqttFilter{{filter: test.topic, qos: 2}}, []byte{2})
-			if test.flags&mqttPubFlagRetain != 0 {
-				testMQTTCheckPubMsg(t, mc, r, test.topic, test.flags, []byte(test.payload))
-			} else {
-				testMQTTExpectNothing(t, r)
-			}
-		}
-	})
-}
-
 //////////////////////////////////////////////////////////////////////////
 //
 // Benchmarks
diff --git a/server/test_test.go b/server/test_test.go
index 8f0d393a..704cc1e8 100644
--- a/server/test_test.go
+++ b/server/test_test.go
@@ -303,6 +303,7 @@ func (c *cluster) shutdown() {
 	for i, s := range c.servers {
 		sd := s.StoreDir()
 		s.Shutdown()
+		s.WaitForShutdown()
 		if cf := c.opts[i].ConfigFile; cf != _EMPTY_ {
 			os.Remove(cf)
 		}