Repository: nats-io/nats-server
Pull request #5987 (2024-10-10 16:59:32)

NRG (2.11): Start catchup from `n.commit` & fix AppendEntry is stored at `seq=ae.pindex+1`
This PR makes three complementary fixes to the way how catchup and truncating is handled.
Specifically:
- when doing `n.loadEntry(index)` we need to pass where the AppendEntry is in terms of stream sequence, this is equal to `ae.pindex+1` since the `ae.pindex` is the value before it's stored in the stream.
- start catchup from `n.commit`, we could have messages past our commit that have been invalidated and need to be truncated since there was a switch between leaders
- because we catchup from `n.commit`, we check if our local AppendEntry matches terms with the incoming AppendEntry, we only need to truncate if the terms don't match

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>

derekcollison: LMK when ready for review.

Patch:
diff --git a/.github/workflows/cov.yaml b/.github/workflows/cov.yaml
index e3324aa1..f372de48 100644
--- a/.github/workflows/cov.yaml
+++ b/.github/workflows/cov.yaml
@@ -35,8 +35,8 @@ jobs:
 
       - name: Convert coverage.out to coverage.lcov
         # Use commit hash here to avoid a re-tagging attack, as this is a third-party action
-        # Commit c680c0f7c7442485f1749eb2a13e54a686e76eb5 = tag v1.0.9
-        uses: jandelgado/gcov2lcov-action@c680c0f7c7442485f1749eb2a13e54a686e76eb5
+        # Commit 69ef3d59a24cc6e062516a73d8be123e85b15cc0 = tag v1.1.0
+        uses: jandelgado/gcov2lcov-action@69ef3d59a24cc6e062516a73d8be123e85b15cc0
         with:
           infile: acc.out
           working-directory: src/github.com/nats-io/nats-server
diff --git a/go.mod b/go.mod
index b2efd840..47be84c0 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ toolchain go1.22.8
 
 require (
 	github.com/google/go-tpm v0.9.0
-	github.com/klauspost/compress v1.17.10
+	github.com/klauspost/compress v1.17.11
 	github.com/minio/highwayhash v1.0.3
 	github.com/nats-io/jwt/v2 v2.7.2
 	github.com/nats-io/nats.go v1.36.0
diff --git a/go.sum b/go.sum
index 92e49feb..dc009b9f 100644
--- a/go.sum
+++ b/go.sum
@@ -2,8 +2,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/google/go-tpm v0.9.0 h1:sQF6YqWMi+SCXpsmS3fd21oPy/vSddwZry4JnmltHVk=
 github.com/google/go-tpm v0.9.0/go.mod h1:FkNVkc6C+IsvDI9Jw1OveJmxGZUUaKxtrpOS47QWKfU=
-github.com/klauspost/compress v1.17.10 h1:oXAz+Vh0PMUvJczoi+flxpnBEPxoER1IaAnU/NMPtT0=
-github.com/klauspost/compress v1.17.10/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
+github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
+github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
 github.com/minio/highwayhash v1.0.3 h1:kbnuUMoHYyVl7szWjSxJnxw11k2U709jqFPPmIUyD6Q=
 github.com/minio/highwayhash v1.0.3/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
 github.com/nats-io/jwt/v2 v2.7.2 h1:SCRjfDLJ2q8naXp8YlGJJS5/yj3wGSODFYVi4nnwVMw=
diff --git a/server/consumer.go b/server/consumer.go
index 8ede0b92..eb9d13ed 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -3736,9 +3736,7 @@ func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {
 	var pmsg = getJSPubMsgFromPool()
 
 	// Grab next message applicable to us.
-	// We will unlock here in case lots of contention, e.g. WQ.
 	filters, subjf, fseq := o.filters, o.subjf, o.sseq
-	o.mu.Unlock()
 	// Check if we are multi-filtered or not.
 	if filters != nil {
 		sm, sseq, err = store.LoadNextMsgMulti(filters, fseq, &pmsg.StoreMsg)
@@ -3753,8 +3751,6 @@ func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {
 		pmsg.returnToPool()
 		pmsg = nil
 	}
-	// Re-acquire lock.
-	o.mu.Lock()
 	// Check if we should move our o.sseq.
 	if sseq >= o.sseq {
 		// If we are moving step by step then sseq == o.sseq.
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index b3617563..93b26fc5 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -2833,40 +2833,42 @@ func (s *Server) jsLeaderStepDownRequest(sub *subscription, c *client, _ *Accoun
 	var preferredLeader string
 	var resp = JSApiLeaderStepDownResponse{ApiResponse: ApiResponse{Type: JSApiLeaderStepDownResponseType}}
 
-	if !isEmptyRequest(msg) {
+	if isJSONObjectOrArray(msg) {
 		var req JSApiLeaderStepdownRequest
 		if err := json.Unmarshal(msg, &req); err != nil {
 			resp.Error = NewJSInvalidJSONError(err)
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 			return
 		}
-		if len(req.Placement.Tags) > 0 {
-			// Tags currently not supported.
-			resp.Error = NewJSClusterTagsError()
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
-		cn := req.Placement.Cluster
-		var peers []string
-		ourID := cc.meta.ID()
-		for _, p := range cc.meta.Peers() {
-			if si, ok := s.nodeToInfo.Load(p.ID); ok && si != nil {
-				if ni := si.(nodeInfo); ni.offline || ni.cluster != cn || p.ID == ourID {
-					continue
+		if req.Placement != nil {
+			if len(req.Placement.Tags) > 0 {
+				// Tags currently not supported.
+				resp.Error = NewJSClusterTagsError()
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+				return
+			}
+			cn := req.Placement.Cluster
+			var peers []string
+			ourID := cc.meta.ID()
+			for _, p := range cc.meta.Peers() {
+				if si, ok := s.nodeToInfo.Load(p.ID); ok && si != nil {
+					if ni := si.(nodeInfo); ni.offline || ni.cluster != cn || p.ID == ourID {
+						continue
+					}
+					peers = append(peers, p.ID)
 				}
-				peers = append(peers, p.ID)
 			}
+			if len(peers) == 0 {
+				resp.Error = NewJSClusterNoPeersError(fmt.Errorf("no replacement peer connected"))
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+				return
+			}
+			// Randomize and select.
+			if len(peers) > 1 {
+				rand.Shuffle(len(peers), func(i, j int) { peers[i], peers[j] = peers[j], peers[i] })
+			}
+			preferredLeader = peers[0]
 		}
-		if len(peers) == 0 {
-			resp.Error = NewJSClusterNoPeersError(fmt.Errorf("no replacement peer connected"))
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
-		// Randomize and select.
-		if len(peers) > 1 {
-			rand.Shuffle(len(peers), func(i, j int) { peers[i], peers[j] = peers[j], peers[i] })
-		}
-		preferredLeader = peers[0]
 	}
 
 	// Call actual stepdown.
diff --git a/server/raft.go b/server/raft.go
index dedc4be5..6df4fcb6 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -343,8 +343,8 @@ func (s *Server) bootstrapRaftNode(cfg *RaftConfig, knownPeers []string, allPeer
 	return writePeerState(cfg.Store, &peerState{knownPeers, expected, extUndetermined})
 }
 
-// startRaftNode will start the raft node.
-func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabels) (RaftNode, error) {
+// initRaftNode will initialize the raft node, to be used by startRaftNode or when testing to not run the Go routine.
+func (s *Server) initRaftNode(accName string, cfg *RaftConfig, labels pprofLabels) (*raft, error) {
 	if cfg == nil {
 		return nil, errNilCfg
 	}
@@ -520,6 +520,16 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 	labels["group"] = n.group
 	s.registerRaftNode(n.group, n)
 
+	return n, nil
+}
+
+// startRaftNode will start the raft node.
+func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabels) (RaftNode, error) {
+	n, err := s.initRaftNode(accName, cfg, labels)
+	if err != nil {
+		return nil, err
+	}
+
 	// Start the run goroutine for the Raft state machine.
 	s.startGoRoutine(n.run, labels)
 
@@ -3141,10 +3151,10 @@ func (n *raft) catchupStalled() bool {
 	if n.catchup == nil {
 		return false
 	}
-	if n.catchup.pindex == n.pindex {
+	if n.catchup.pindex == n.commit {
 		return time.Since(n.catchup.active) > 2*time.Second
 	}
-	n.catchup.pindex = n.pindex
+	n.catchup.pindex = n.commit
 	n.catchup.active = time.Now()
 	return false
 }
@@ -3163,7 +3173,7 @@ func (n *raft) createCatchup(ae *appendEntry) string {
 		cterm:  ae.pterm,
 		cindex: ae.pindex,
 		pterm:  n.pterm,
-		pindex: n.pindex,
+		pindex: n.commit,
 		active: time.Now(),
 	}
 	inbox := n.newCatchupInbox()
@@ -3333,7 +3343,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			if n.catchupStalled() {
 				n.debug("Catchup may be stalled, will request again")
 				inbox = n.createCatchup(ae)
-				ar = newAppendEntryResponse(n.pterm, n.pindex, n.id, false)
+				ar = newAppendEntryResponse(n.pterm, n.commit, n.id, false)
 			}
 			n.Unlock()
 			if ar != nil {
@@ -3374,13 +3384,15 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	}
 
 	if (isNew && ae.pterm != n.pterm) || ae.pindex != n.pindex {
-		// Check if this is a lower or equal index than what we were expecting.
-		if ae.pindex <= n.pindex {
+		// Check if this is a lower index than what we were expecting.
+		if ae.pindex < n.pindex {
 			n.debug("AppendEntry detected pindex less than ours: %d:%d vs %d:%d", ae.pterm, ae.pindex, n.pterm, n.pindex)
 			var ar *appendEntryResponse
 
+			// An AppendEntry is stored at seq=ae.pindex+1. This can be checked when eae != nil, eae.pindex==ae.pindex.
+			seq := ae.pindex + 1
 			var success bool
-			if eae, _ := n.loadEntry(ae.pindex); eae == nil {
+			if eae, _ := n.loadEntry(seq); eae == nil {
 				// If terms are equal, and we are not catching up, we have simply already processed this message.
 				// So we will ACK back to the leader. This can happen on server restarts based on timings of snapshots.
 				if ae.pterm == n.pterm && !catchingUp {
@@ -3388,14 +3400,18 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 				} else {
 					n.resetWAL()
 				}
-			} else {
-				// If terms mismatched, or we got an error loading, delete that entry and all others past it.
+			} else if eae.term != ae.term {
+				// If terms mismatched, delete that entry and all others past it.
 				// Make sure to cancel any catchups in progress.
 				// Truncate will reset our pterm and pindex. Only do so if we have an entry.
 				n.truncateWAL(eae.pterm, eae.pindex)
+			} else {
+				success = true
+			}
+			// Cancel regardless if truncated/unsuccessful.
+			if !success {
+				n.cancelCatchup()
 			}
-			// Cancel regardless.
-			n.cancelCatchup()
 
 			// Create response.
 			ar = newAppendEntryResponse(ae.pterm, ae.pindex, n.id, success)
@@ -3469,11 +3485,11 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			return
 
 		} else {
-			n.debug("AppendEntry did not match %d %d with %d %d", ae.pterm, ae.pindex, n.pterm, n.pindex)
-			if ae.pindex > n.pindex {
+			n.debug("AppendEntry did not match %d %d with %d %d (commit %d)", ae.pterm, ae.pindex, n.pterm, n.pindex, n.commit)
+			if ae.pindex > n.commit {
 				// Setup our state for catching up.
 				inbox := n.createCatchup(ae)
-				ar := newAppendEntryResponse(n.pterm, n.pindex, n.id, false)
+				ar := newAppendEntryResponse(n.pterm, n.commit, n.id, false)
 				n.Unlock()
 				n.sendRPC(ae.reply, inbox, ar.encode(arbuf))
 				arPool.Put(ar)


Test patch:
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
index 51028fec..32f807df 100644
--- a/server/jetstream_cluster_4_test.go
+++ b/server/jetstream_cluster_4_test.go
@@ -4274,3 +4274,165 @@ func TestJetStreamClusterHardKillAfterStreamAdd(t *testing.T) {
 	_, err = js.StreamInfo("TEST")
 	require_NoError(t, err)
 }
+
+func TestJetStreamClusterDesyncAfterPublishToLeaderWithoutQuorum(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	si, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	streamLeader := si.Cluster.Leader
+	streamLeaderServer := c.serverByName(streamLeader)
+	nc.Close()
+	nc, js = jsClientConnect(t, streamLeaderServer)
+	defer nc.Close()
+
+	servers := slices.DeleteFunc([]string{"S-1", "S-2", "S-3"}, func(s string) bool {
+		return s == streamLeader
+	})
+
+	// Stop followers so further publishes will not have quorum.
+	followerName1 := servers[0]
+	followerName2 := servers[1]
+	followerServer1 := c.serverByName(followerName1)
+	followerServer2 := c.serverByName(followerName2)
+	followerServer1.Shutdown()
+	followerServer2.Shutdown()
+	followerServer1.WaitForShutdown()
+	followerServer2.WaitForShutdown()
+
+	// Although this request will time out, it will be added to the stream leader's WAL.
+	_, err = js.Publish("foo", []byte("first"), nats.AckWait(time.Second))
+	require_NotNil(t, err)
+	require_Equal(t, err, nats.ErrTimeout)
+
+	// Now shut down the leader as well.
+	nc.Close()
+	streamLeaderServer.Shutdown()
+	streamLeaderServer.WaitForShutdown()
+
+	// Only restart the (previous) followers.
+	followerServer1 = c.restartServer(followerServer1)
+	c.restartServer(followerServer2)
+	c.waitOnStreamLeader(globalAccountName, "TEST")
+
+	nc, js = jsClientConnect(t, followerServer1)
+	defer nc.Close()
+
+	// Publishing a message will now have quorum.
+	pubAck, err := js.Publish("foo", []byte("first, this is a retry"))
+	require_NoError(t, err)
+	require_Equal(t, pubAck.Sequence, 1)
+
+	// Bring up the previous stream leader.
+	c.restartServer(streamLeaderServer)
+	c.waitOnAllCurrent()
+	c.waitOnStreamLeader(globalAccountName, "TEST")
+
+	// Check all servers ended up with the last published message, which had quorum.
+	for _, s := range c.servers {
+		c.waitOnStreamCurrent(s, globalAccountName, "TEST")
+
+		acc, err := s.lookupAccount(globalAccountName)
+		require_NoError(t, err)
+		mset, err := acc.lookupStream("TEST")
+		require_NoError(t, err)
+		state := mset.state()
+		require_Equal(t, state.Msgs, 1)
+		require_Equal(t, state.Bytes, 55)
+	}
+}
+
+func TestJetStreamClusterPreserveWALDuringCatchupWithMatchingTerm(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo.>"},
+		Replicas: 3,
+	})
+	nc.Close()
+	require_NoError(t, err)
+
+	// Pick one server that will only store a part of the messages in its WAL.
+	rs := c.randomNonStreamLeader(globalAccountName, "TEST")
+	ts := time.Now().UnixNano()
+
+	// Manually add 3 append entries to each node's WAL, except for one node who is one behind.
+	var scratch [1024]byte
+	for _, s := range c.servers {
+		for _, n := range s.raftNodes {
+			rn := n.(*raft)
+			if rn.accName == globalAccountName {
+				for i := uint64(0); i < 3; i++ {
+					// One server will be one behind and need to catchup.
+					if s.Name() == rs.Name() && i >= 2 {
+						break
+					}
+
+					esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, i, ts, true)
+					entries := []*Entry{newEntry(EntryNormal, esm)}
+					rn.Lock()
+					ae := rn.buildAppendEntry(entries)
+					ae.buf, err = ae.encode(scratch[:])
+					require_NoError(t, err)
+					err = rn.storeToWAL(ae)
+					rn.Unlock()
+					require_NoError(t, err)
+				}
+			}
+		}
+	}
+
+	// Restart all.
+	c.stopAll()
+	c.restartAll()
+	c.waitOnAllCurrent()
+	c.waitOnStreamLeader(globalAccountName, "TEST")
+
+	rs = c.serverByName(rs.Name())
+
+	// Check all servers ended up with all published messages, which had quorum.
+	for _, s := range c.servers {
+		c.waitOnStreamCurrent(s, globalAccountName, "TEST")
+
+		acc, err := s.lookupAccount(globalAccountName)
+		require_NoError(t, err)
+		mset, err := acc.lookupStream("TEST")
+		require_NoError(t, err)
+		state := mset.state()
+		require_Equal(t, state.Msgs, 3)
+		require_Equal(t, state.Bytes, 99)
+	}
+
+	// Check that the first two published messages came from our WAL, and
+	// the last came from a catchup by another leader.
+	for _, n := range rs.raftNodes {
+		rn := n.(*raft)
+		if rn.accName == globalAccountName {
+			ae, err := rn.loadEntry(2)
+			require_NoError(t, err)
+			require_True(t, ae.leader == rn.ID())
+
+			ae, err = rn.loadEntry(3)
+			require_NoError(t, err)
+			require_True(t, ae.leader == rn.ID())
+
+			ae, err = rn.loadEntry(4)
+			require_NoError(t, err)
+			require_True(t, ae.leader != rn.ID())
+		}
+	}
+}
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index 611d5738..28b6dc3d 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -18,6 +18,7 @@ package server
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math/rand"
 	"slices"
@@ -1402,6 +1403,120 @@ func TestJetStreamConsumerPedanticMode(t *testing.T) {
 		}
 	}
 }
+
+func TestJetStreamConsumerStuckAckPending(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	type ActiveWorkItem struct {
+		ID     int
+		Expiry time.Time
+	}
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:              "TEST_ACTIVE_WORK_ITEMS",
+		Discard:           nats.DiscardOld,
+		MaxMsgsPerSubject: 1,
+		Subjects:          []string{"TEST_ACTIVE_WORK_ITEMS.>"},
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST_ACTIVE_WORK_ITEMS", &nats.ConsumerConfig{
+		Durable:       "testactiveworkitemsconsumer",
+		AckPolicy:     nats.AckExplicitPolicy,
+		MaxAckPending: -1,
+		MaxWaiting:    20000,
+		AckWait:       15 * time.Second,
+	})
+	require_NoError(t, err)
+
+	sub, err := js.PullSubscribe("TEST_ACTIVE_WORK_ITEMS.>", "testactiveworkitemsconsumer", nats.BindStream("TEST_ACTIVE_WORK_ITEMS"))
+	require_NoError(t, err)
+
+	errs := make(chan error)
+	go func() {
+		for {
+			msgs, err := sub.Fetch(200)
+			if err != nil {
+				// test is done. stop the loop.
+				if errors.Is(err, nats.ErrSubscriptionClosed) || errors.Is(err, nats.ErrConnectionClosed) {
+					return
+				}
+				if !errors.Is(err, nats.ErrTimeout) {
+					errs <- err
+					return
+				}
+				continue
+			}
+			for _, msg := range msgs {
+				msg := msg
+				var workItem ActiveWorkItem
+				if err := json.Unmarshal(msg.Data, &workItem); err != nil {
+					errs <- err
+					return
+				}
+
+				now := time.Now()
+				// If the work item has not expired, nak it with the respective delay.
+				if workItem.Expiry.After(now) {
+					msg.NakWithDelay(workItem.Expiry.Sub(now))
+				} else {
+					msg.Ack()
+				}
+			}
+		}
+	}()
+
+	for i := 0; i < 25_000; i++ {
+		// Publish item to TEST_ACTIVE_WORK_ITEMS stream with an expiry time.
+		workItem := ActiveWorkItem{ID: i, Expiry: time.Now().Add(30 * time.Second)}
+		data, err := json.Marshal(workItem)
+		require_NoError(t, err)
+
+		_, err = js.Publish(fmt.Sprintf("TEST_ACTIVE_WORK_ITEMS.%d", i), data)
+		require_NoError(t, err)
+
+		// Update expiry time and republish item to TEST_ACTIVE_WORK_ITEMS stream.
+		workItem.Expiry = time.Now().Add(3 * time.Second)
+		data, err = json.Marshal(workItem)
+		require_NoError(t, err)
+		_, err = js.Publish(fmt.Sprintf("TEST_ACTIVE_WORK_ITEMS.%d", i), data)
+		require_NoError(t, err)
+	}
+	noChange := false
+	lastNumAckPending := 0
+	checkFor(t, 60*time.Second, 3*time.Second, func() error {
+		select {
+		case err := <-errs:
+			t.Fatalf("consumer goroutine failed: %v", err)
+		default:
+		}
+		ci, err := js.ConsumerInfo("TEST_ACTIVE_WORK_ITEMS", "testactiveworkitemsconsumer")
+		require_NoError(t, err)
+
+		if lastNumAckPending != 0 && lastNumAckPending == ci.NumAckPending {
+			noChange = true
+		}
+		lastNumAckPending = ci.NumAckPending
+
+		// If we have no change since last check, we can fail the test before `totalWait` timeout.
+		if ci.NumAckPending > 0 && ci.NumPending == 0 {
+			if noChange {
+				_, err := sub.Fetch(1)
+				if err != nil && errors.Is(err, nats.ErrTimeout) {
+
+					t.Fatalf("num ack pending: %d\t num pending: %v\n", ci.NumAckPending, ci.NumPending)
+				}
+			}
+			return fmt.Errorf("num ack pending: %d\t num pending: %v\n", ci.NumAckPending, ci.NumPending)
+		}
+		return nil
+	})
+}
+
 func Benchmark____JetStreamConsumerIsFilteredMatch(b *testing.B) {
 	subject := "foo.bar.do.not.match.any.filter.subject"
 	for n := 1; n <= 1024; n *= 2 {
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 7991bfbc..6249a34d 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -6713,7 +6713,7 @@ func TestJetStreamSystemLimitsPlacement(t *testing.T) {
 			si.Config.MaxBytes, si.Config.Replicas)
 	}
 
-	t.Run("meta info placement in request", func(t *testing.T) {
+	t.Run("meta info placement in request - empty request", func(t *testing.T) {
 		nc, err = nats.Connect(largeSrv.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
 		require_NoError(t, err)
 		defer nc.Close()
@@ -6727,6 +6727,23 @@ func TestJetStreamSystemLimitsPlacement(t *testing.T) {
 		require_True(t, resp.Success)
 
 	})
+
+	t.Run("meta info placement in request - uninitialized fields", func(t *testing.T) {
+		nc, err = nats.Connect(largeSrv.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+		require_NoError(t, err)
+		defer nc.Close()
+
+		cluster.waitOnClusterReadyWithNumPeers(3)
+		var resp JSApiLeaderStepDownResponse
+		req, err := json.Marshal(JSApiLeaderStepdownRequest{Placement: nil})
+		require_NoError(t, err)
+		ncResp, err := nc.Request(JSApiLeaderStepDown, req, 10*time.Second)
+		require_NoError(t, err)
+		err = json.Unmarshal(ncResp.Data, &resp)
+		require_NoError(t, err)
+		require_True(t, resp.Error == nil)
+	})
+
 }
 
 func TestJetStreamStreamLimitUpdate(t *testing.T) {
diff --git a/server/raft_test.go b/server/raft_test.go
index 206dcc3b..1ec2193a 100644
--- a/server/raft_test.go
+++ b/server/raft_test.go
@@ -784,8 +784,8 @@ func TestNRGTermDoesntRollBackToPtermOnCatchup(t *testing.T) {
 		require_Equal(t, rn.term, 2)
 
 		if !rn.Leader() {
-			rn.truncateWAL(1, 6) // This will overwrite rn.term, so...
-			rn.term = 2          // ... we'll set it back manually.
+			rn.truncateWAL(1, 6)
+			require_Equal(t, rn.term, 2) // rn.term must stay the same
 			require_Equal(t, rn.pterm, 1)
 			require_Equal(t, rn.pindex, 6)
 		}
@@ -945,68 +945,94 @@ func TestNRGRemoveLeaderPeerDeadlockBug(t *testing.T) {
 }
 
 func TestNRGWALEntryWithoutQuorumMustTruncate(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
-	defer c.shutdown()
-
-	rg := c.createRaftGroup("TEST", 3, newStateAdder)
-	rg.waitOnLeader()
-
-	var err error
-	var scratch [1024]byte
-
-	// Simulate leader storing an AppendEntry in WAL but being hard killed before it can propose to its peers.
-	n := rg.leader().node().(*raft)
-	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
-	entries := []*Entry{newEntry(EntryNormal, esm)}
-	n.Lock()
-	ae := n.buildAppendEntry(entries)
-	ae.buf, err = ae.encode(scratch[:])
-	require_NoError(t, err)
-	err = n.storeToWAL(ae)
-	n.Unlock()
-	require_NoError(t, err)
-
-	// Stop the leader so it moves to another one.
-	n.shutdown(false)
-
-	// Wait for another leader to be picked
-	rg.waitOnLeader()
-
-	// Restart the previous leader that contains the stored AppendEntry without quorum.
-	for _, a := range rg {
-		if a.node().ID() == n.ID() {
-			sa := a.(*stateAdder)
-			sa.restart()
-			break
-		}
+	tests := []struct {
+		title  string
+		modify func(rg smGroup)
+	}{
+		{
+			// state equals, only need to remove the entry
+			title:  "equal",
+			modify: func(rg smGroup) {},
+		},
+		{
+			// state diverged, need to replace the entry
+			title: "diverged",
+			modify: func(rg smGroup) {
+				rg.leader().(*stateAdder).proposeDelta(11)
+			},
+		},
 	}
 
-	// The previous leader's WAL should truncate to remove the AppendEntry only it has.
-	// Eventually all WALs for all peers must match.
-	checkFor(t, 5*time.Second, 200*time.Millisecond, func() error {
-		var expected [][]byte
-		for _, a := range rg {
-			an := a.node().(*raft)
-			var state StreamState
-			an.wal.FastState(&state)
-			if len(expected) > 0 && int(state.LastSeq-state.FirstSeq+1) != len(expected) {
-				return fmt.Errorf("WAL is different: too many entries")
-			}
-			for index := state.FirstSeq; index <= state.LastSeq; index++ {
-				ae, err := an.loadEntry(index)
-				if err != nil {
-					return err
-				}
-				seq := int(index)
-				if len(expected) < seq {
-					expected = append(expected, ae.buf)
-				} else if !bytes.Equal(expected[seq-1], ae.buf) {
-					return fmt.Errorf("WAL is different: stored bytes differ")
+	for _, test := range tests {
+		t.Run(test.title, func(t *testing.T) {
+			c := createJetStreamClusterExplicit(t, "R3S", 3)
+			defer c.shutdown()
+
+			rg := c.createRaftGroup("TEST", 3, newStateAdder)
+			rg.waitOnLeader()
+
+			var err error
+			var scratch [1024]byte
+
+			// Simulate leader storing an AppendEntry in WAL but being hard killed before it can propose to its peers.
+			n := rg.leader().node().(*raft)
+			esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+			entries := []*Entry{newEntry(EntryNormal, esm)}
+			n.Lock()
+			ae := n.buildAppendEntry(entries)
+			ae.buf, err = ae.encode(scratch[:])
+			require_NoError(t, err)
+			err = n.storeToWAL(ae)
+			n.Unlock()
+			require_NoError(t, err)
+
+			// Stop the leader so it moves to another one.
+			n.shutdown(false)
+
+			// Wait for another leader to be picked
+			rg.waitOnLeader()
+
+			// Make a modification, specific to this test.
+			test.modify(rg)
+
+			// Restart the previous leader that contains the stored AppendEntry without quorum.
+			for _, a := range rg {
+				if a.node().ID() == n.ID() {
+					sa := a.(*stateAdder)
+					sa.restart()
+					break
 				}
 			}
-		}
-		return nil
-	})
+
+			// The previous leader's WAL should truncate to remove the AppendEntry only it has.
+			// Eventually all WALs for all peers must match.
+			checkFor(t, 5*time.Second, 200*time.Millisecond, func() error {
+				var expected [][]byte
+				for _, a := range rg {
+					an := a.node().(*raft)
+					var state StreamState
+					an.wal.FastState(&state)
+					if len(expected) > 0 && int(state.LastSeq-state.FirstSeq+1) != len(expected) {
+						return fmt.Errorf("WAL is different: too many entries")
+					}
+					// Loop over all entries in the WAL, checking if the contents for all RAFT nodes are equal.
+					for index := state.FirstSeq; index <= state.LastSeq; index++ {
+						ae, err := an.loadEntry(index)
+						if err != nil {
+							return err
+						}
+						seq := int(index)
+						if len(expected) < seq {
+							expected = append(expected, ae.buf)
+						} else if !bytes.Equal(expected[seq-1], ae.buf) {
+							return fmt.Errorf("WAL is different: stored bytes differ")
+						}
+					}
+				}
+				return nil
+			})
+		})
+	}
 }
 
 func TestNRGTermNoDecreaseAfterWALReset(t *testing.T) {
@@ -1051,3 +1077,132 @@ func TestNRGTermNoDecreaseAfterWALReset(t *testing.T) {
 		}
 	}
 }
+
+func TestNRGCatchupDoesNotTruncateUncommittedEntriesWithQuorum(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+	s := c.servers[0] // RunBasicJetStreamServer not available
+
+	ms, err := newMemStore(&StreamConfig{Name: "TEST", Storage: MemoryStorage})
+	require_NoError(t, err)
+	cfg := &RaftConfig{Name: "TEST", Store: t.TempDir(), Log: ms}
+
+	err = s.bootstrapRaftNode(cfg, nil, false)
+	require_NoError(t, err)
+	n, err := s.initRaftNode(globalAccountName, cfg, pprofLabels{})
+	require_NoError(t, err)
+
+	// An AppendEntry is encoded into a buffer and that's stored into the WAL.
+	// This is a helper function to generate that buffer.
+	encode := func(ae *appendEntry) *appendEntry {
+		buf, err := ae.encode(nil)
+		require_NoError(t, err)
+		ae.buf = buf
+		return ae
+	}
+
+	// Create a sample entry, the content doesn't matter, just that it's stored.
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	entries := []*Entry{newEntry(EntryNormal, esm)}
+
+	nats0 := "S1Nunr6R" // "nats-0"
+	nats1 := "yrzKKRBu" // "nats-1"
+
+	// Timeline, for first leader
+	aeInitial := encode(&appendEntry{leader: nats0, term: 1, commit: 0, pterm: 0, pindex: 0, entries: entries})
+	aeHeartbeat := encode(&appendEntry{leader: nats0, term: 1, commit: 1, pterm: 1, pindex: 1, entries: nil})
+	aeUncommitted := encode(&appendEntry{leader: nats0, term: 1, commit: 1, pterm: 1, pindex: 1, entries: entries})
+	aeNoQuorum := encode(&appendEntry{leader: nats0, term: 1, commit: 1, pterm: 1, pindex: 2, entries: entries})
+
+	// Timeline, after leader change
+	aeMissed := encode(&appendEntry{leader: nats1, term: 2, commit: 1, pterm: 1, pindex: 2, entries: entries})
+	aeCatchupTrigger := encode(&appendEntry{leader: nats1, term: 2, commit: 1, pterm: 2, pindex: 3, entries: entries})
+	aeHeartbeat2 := encode(&appendEntry{leader: nats1, term: 2, commit: 2, pterm: 2, pindex: 4, entries: nil})
+	aeHeartbeat3 := encode(&appendEntry{leader: nats1, term: 2, commit: 4, pterm: 2, pindex: 4, entries: nil})
+
+	// Initial case is simple, just store the entry.
+	n.processAppendEntry(aeInitial, n.aesub)
+	require_Equal(t, n.wal.State().Msgs, 1)
+	entry, err := n.loadEntry(1)
+	require_NoError(t, err)
+	require_Equal(t, entry.leader, nats0)
+
+	// Heartbeat, makes sure commit moves up.
+	n.processAppendEntry(aeHeartbeat, n.aesub)
+	require_Equal(t, n.commit, 1)
+
+	// We get one entry that has quorum (but we don't know that yet), so it stays uncommitted for a bit.
+	n.processAppendEntry(aeUncommitted, n.aesub)
+	require_Equal(t, n.wal.State().Msgs, 2)
+	entry, err = n.loadEntry(2)
+	require_NoError(t, err)
+	require_Equal(t, entry.leader, nats0)
+
+	// We get one entry that has NO quorum (but we don't know that yet).
+	n.processAppendEntry(aeNoQuorum, n.aesub)
+	require_Equal(t, n.wal.State().Msgs, 3)
+	entry, err = n.loadEntry(3)
+	require_NoError(t, err)
+	require_Equal(t, entry.leader, nats0)
+
+	// We've just had a leader election, and we missed one message from the previous leader, we should catchup.
+	n.processAppendEntry(aeCatchupTrigger, n.aesub)
+	require_Equal(t, n.wal.State().Msgs, 3)
+	require_True(t, n.catchup != nil)
+	require_Equal(t, n.catchup.pterm, 1)  // n.pterm
+	require_Equal(t, n.catchup.pindex, 1) // n.commit
+
+	// Make sure our WAL was not truncated, and we're still catching up.
+	aeUncommitted.leader = nats1
+	aeUncommitted = encode(aeUncommitted)
+	n.processAppendEntry(aeUncommitted, n.catchup.sub)
+	require_Equal(t, n.wal.State().Msgs, 3)
+	require_True(t, n.catchup != nil)
+	// Our entry should not be touched, so the 'leader' should've stayed the same.
+	entry, err = n.loadEntry(2)
+	require_NoError(t, err)
+	require_Equal(t, entry.leader, nats0)
+
+	// We now notice the leader indicated a different entry at the (no quorum) index, should truncate.
+	n.processAppendEntry(aeMissed, n.catchup.sub)
+	require_Equal(t, n.wal.State().Msgs, 2)
+	require_True(t, n.catchup == nil)
+
+	// We get a heartbeat that prompts us to catchup again.
+	n.processAppendEntry(aeHeartbeat2, n.aesub)
+	require_Equal(t, n.wal.State().Msgs, 2)
+	require_Equal(t, n.commit, 1) // Commit should not change, as we missed an item.
+	require_True(t, n.catchup != nil)
+	require_Equal(t, n.catchup.pterm, 1)  // n.pterm
+	require_Equal(t, n.catchup.pindex, 1) // n.commit
+
+	// We get the uncommitted entry again, it should stay the same.
+	n.processAppendEntry(aeUncommitted, n.catchup.sub)
+	require_Equal(t, n.wal.State().Msgs, 2)
+	require_True(t, n.catchup != nil)
+	// Our entry should still stay the same.
+	entry, err = n.loadEntry(2)
+	require_NoError(t, err)
+	require_Equal(t, entry.leader, nats0)
+
+	// We now get the missed append entry, store it.
+	n.processAppendEntry(aeMissed, n.catchup.sub)
+	require_Equal(t, n.wal.State().Msgs, 3)
+	require_True(t, n.catchup != nil)
+	entry, err = n.loadEntry(3)
+	require_NoError(t, err)
+	require_Equal(t, entry.leader, nats1)
+
+	// We now get the entry that initially triggered us to catchup again, it should be added.
+	n.processAppendEntry(aeCatchupTrigger, n.catchup.sub)
+	require_Equal(t, n.wal.State().Msgs, 4)
+	require_True(t, n.catchup != nil)
+	entry, err = n.loadEntry(4)
+	require_NoError(t, err)
+	require_Equal(t, entry.leader, nats1)
+
+	// Heartbeat, makes sure we commit (and reset catchup, as we're now up-to-date).
+	n.processAppendEntry(aeHeartbeat3, n.aesub)
+	require_Equal(t, n.commit, 4)
+	require_True(t, n.catchup == nil)
+}