Repository: nats-io/nats-server
Pull request #3026 (2022-04-12 00:24:16)

[FIXED] subject renaming with hand crafted reply subject
Signed-off-by: Matthias Hanel <mh@synadia.com>

Please read comment in client.go. probably should make this illegal, @ or not.


Patch:
diff --git a/go.mod b/go.mod
index fe4a4f02..1bd4b74d 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/klauspost/compress v1.14.4
 	github.com/minio/highwayhash v1.0.2
 	github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a
-	github.com/nats-io/nats.go v1.13.1-0.20220318132711-e0e03e374228
+	github.com/nats-io/nats.go v1.14.0
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
 	golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce
diff --git a/go.sum b/go.sum
index 1f40540c..3be821b4 100644
--- a/go.sum
+++ b/go.sum
@@ -14,8 +14,8 @@ github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a h1:lem6QCvxR0Y28gth9P+wV2K/zYUUAkJ+55U8cpS0p5I=
 github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
-github.com/nats-io/nats.go v1.13.1-0.20220318132711-e0e03e374228 h1:czbQ9uYuV7dwLsh/0vpB+4rutgdLTYgoN5W5hf1S0eg=
-github.com/nats-io/nats.go v1.13.1-0.20220318132711-e0e03e374228/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nats.go v1.14.0 h1:/QLCss4vQ6wvDpbqXucsVRDi13tFIR6kTdau+nXzKJw=
+github.com/nats-io/nats.go v1.14.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
diff --git a/server/client.go b/server/client.go
index 495276fb..56d1787e 100644
--- a/server/client.go
+++ b/server/client.go
@@ -3493,8 +3493,11 @@ func isReservedReply(reply []byte) bool {
 	if isServiceReply(reply) {
 		return true
 	}
+	rLen := len(reply)
 	// Faster to check with string([:]) than byte-by-byte
-	if len(reply) > gwReplyPrefixLen && string(reply[:gwReplyPrefixLen]) == gwReplyPrefix {
+	if rLen > jsAckPreLen && string(reply[:jsAckPreLen]) == jsAckPre {
+		return true
+	} else if rLen > gwReplyPrefixLen && string(reply[:gwReplyPrefixLen]) == gwReplyPrefix {
 		return true
 	}
 	return false
@@ -4164,8 +4167,8 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 				dsubj = append(_dsubj[:0], sub.im.to...)
 			}
 
-			// Make sure deliver is set if inbound from a route.
-			if remapped && c.kind == ROUTER {
+			// Make sure deliver is set if inbound from a route. (leaf is fixed on send)
+			if remapped && (c.kind == GATEWAY || c.kind == ROUTER) {
 				deliver = subj
 			}
 			// If we are mapping for a deliver subject we will reverse roles.
diff --git a/server/const.go b/server/const.go
index 50e643cb..0986e3d2 100644
--- a/server/const.go
+++ b/server/const.go
@@ -41,7 +41,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.8.0-beta.12"
+	VERSION = "2.8.0-beta.14"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index dcf10e99..511103a3 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1905,35 +1905,6 @@ func (o *consumer) applyState(state *ConsumerState) {
 	}
 }
 
-func (o *consumer) readStoreState() *ConsumerState {
-	o.mu.RLock()
-	defer o.mu.RUnlock()
-	if o.store == nil {
-		return nil
-	}
-	if o.store.Type() == MemoryStorage {
-		state := o.liveState()
-		// Need to copy pending and rdc since we pass this up.
-		if len(state.Pending) > 0 {
-			pending := make(map[uint64]*Pending, len(state.Pending))
-			for seq, p := range state.Pending {
-				pending[seq] = &Pending{p.Sequence, p.Timestamp}
-			}
-			state.Pending = pending
-		}
-		if len(state.Redelivered) > 0 {
-			redelivered := make(map[uint64]uint64, len(state.Redelivered))
-			for seq, dc := range state.Redelivered {
-				redelivered[seq] = dc
-			}
-			state.Redelivered = redelivered
-		}
-		return state
-	}
-	state, _ := o.store.State()
-	return state
-}
-
 // Sets our store state from another source. Used in clustered mode on snapshot restore.
 func (o *consumer) setStoreState(state *ConsumerState) error {
 	if state == nil || o.store == nil {
@@ -1950,9 +1921,13 @@ func (o *consumer) writeStoreState() error {
 	return o.writeStoreStateUnlocked()
 }
 
+// Update our state to the store.
 // Lock should be held.
-func (o *consumer) liveState() *ConsumerState {
-	return &ConsumerState{
+func (o *consumer) writeStoreStateUnlocked() error {
+	if o.store == nil {
+		return nil
+	}
+	state := ConsumerState{
 		Delivered: SequencePair{
 			Consumer: o.dseq - 1,
 			Stream:   o.sseq - 1,
@@ -1964,15 +1939,7 @@ func (o *consumer) liveState() *ConsumerState {
 		Pending:     o.pending,
 		Redelivered: o.rdc,
 	}
-}
-
-// Update our state to the store.
-// Lock should be held.
-func (o *consumer) writeStoreStateUnlocked() error {
-	if o.store == nil {
-		return nil
-	}
-	return o.store.Update(o.liveState())
+	return o.store.Update(&state)
 }
 
 // Returns an initial info. Only applicable for non-clustered consumers.
@@ -3101,6 +3068,7 @@ func (o *consumer) deliverMsg(dsubj string, pmsg *jsPubMsg, dc uint64) {
 		pmsg.returnToPool()
 		return
 	}
+
 	// Update pending on first attempt. This can go upside down for a short bit, that is ok.
 	// See adjustedPending().
 	if dc == 1 {
diff --git a/server/errors.json b/server/errors.json
index 9c4175df..aa821c20 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1218,5 +1218,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSStreamMoveInProgress",
+    "code": 400,
+    "error_code": 10124,
+    "description": "stream move already in progress",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]
diff --git a/server/filestore.go b/server/filestore.go
index cd6f50b5..2677c084 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -5461,73 +5461,22 @@ const seqsHdrSize = 6*binary.MaxVarintLen64 + hdrLen
 
 // Encode our consumer state, version 2.
 // Lock should be held.
-func (o *consumerFileStore) encodeState() ([]byte, error) {
+
+func (o *consumerFileStore) EncodedState() ([]byte, error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
 	if o.closed {
 		return nil, ErrStoreClosed
 	}
 	return encodeConsumerState(&o.state), nil
 }
 
-func encodeConsumerState(state *ConsumerState) []byte {
-	var hdr [seqsHdrSize]byte
-	var buf []byte
-
-	maxSize := seqsHdrSize
-	if lp := len(state.Pending); lp > 0 {
-		maxSize += lp*(3*binary.MaxVarintLen64) + binary.MaxVarintLen64
-	}
-	if lr := len(state.Redelivered); lr > 0 {
-		maxSize += lr*(2*binary.MaxVarintLen64) + binary.MaxVarintLen64
-	}
-	if maxSize == seqsHdrSize {
-		buf = hdr[:seqsHdrSize]
-	} else {
-		buf = make([]byte, maxSize)
-	}
-
-	// Write header
-	buf[0] = magic
-	buf[1] = 2
-
-	n := hdrLen
-	n += binary.PutUvarint(buf[n:], state.AckFloor.Consumer)
-	n += binary.PutUvarint(buf[n:], state.AckFloor.Stream)
-	n += binary.PutUvarint(buf[n:], state.Delivered.Consumer)
-	n += binary.PutUvarint(buf[n:], state.Delivered.Stream)
-	n += binary.PutUvarint(buf[n:], uint64(len(state.Pending)))
-
-	asflr := state.AckFloor.Stream
-	adflr := state.AckFloor.Consumer
-
-	// These are optional, but always write len. This is to avoid a truncate inline.
-	if len(state.Pending) > 0 {
-		// To save space we will use now rounded to seconds to be base timestamp.
-		mints := time.Now().Round(time.Second).Unix()
-		// Write minimum timestamp we found from above.
-		n += binary.PutVarint(buf[n:], mints)
-
-		for k, v := range state.Pending {
-			n += binary.PutUvarint(buf[n:], k-asflr)
-			n += binary.PutUvarint(buf[n:], v.Sequence-adflr)
-			// Downsample to seconds to save on space.
-			// Subsecond resolution not needed for recovery etc.
-			ts := v.Timestamp / 1_000_000_000
-			n += binary.PutVarint(buf[n:], mints-ts)
-		}
-	}
-
-	// We always write the redelivered len.
-	n += binary.PutUvarint(buf[n:], uint64(len(state.Redelivered)))
-
-	// We expect these to be small.
-	if len(state.Redelivered) > 0 {
-		for k, v := range state.Redelivered {
-			n += binary.PutUvarint(buf[n:], k-asflr)
-			n += binary.PutUvarint(buf[n:], v)
-		}
+func (o *consumerFileStore) encodeState() ([]byte, error) {
+	if o.closed {
+		return nil, ErrStoreClosed
 	}
-
-	return buf[:n]
+	return encodeConsumerState(&o.state), nil
 }
 
 func (o *consumerFileStore) UpdateConfig(cfg *ConsumerConfig) error {
@@ -5757,6 +5706,10 @@ func (o *consumerFileStore) State() (*ConsumerState, error) {
 	o.mu.Lock()
 	defer o.mu.Unlock()
 
+	if o.closed {
+		return nil, ErrStoreClosed
+	}
+
 	state := &ConsumerState{}
 
 	// See if we have a running state or if we need to read in from disk.
diff --git a/server/jetstream.go b/server/jetstream.go
index 294fd82c..71a4be32 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -816,7 +816,7 @@ func (s *Server) migrateEphemerals() {
 						}
 						o.mu.Unlock()
 					}
-					state := o.readStoreState()
+					state, _ := o.store.State()
 					o.deleteWithoutAdvisory()
 					js.mu.Lock()
 					// Delete old one.
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index ae00a413..b8791b39 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -183,8 +183,9 @@ const (
 
 	// jsAckT is the template for the ack message stream coming back from a consumer
 	// when they ACK/NAK, etc a message.
-	jsAckT   = "$JS.ACK.%s.%s"
-	jsAckPre = "$JS.ACK."
+	jsAckT      = "$JS.ACK.%s.%s"
+	jsAckPre    = "$JS.ACK."
+	jsAckPreLen = len(jsAckPre)
 
 	// jsFlowControl is for flow control subjects.
 	jsFlowControlPre = "$JS.FC."
@@ -1388,7 +1389,12 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
-	resp.StreamInfo = &StreamInfo{Created: mset.createdTime(), State: mset.state(), Config: mset.config()}
+	resp.StreamInfo = &StreamInfo{
+		Created: mset.createdTime(),
+		State:   mset.state(),
+		Config:  mset.config(),
+		Domain:  s.getOpts().JetStreamDomain,
+	}
 	resp.DidCreate = true
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
 }
@@ -1488,9 +1494,14 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
-	js, _ := s.getJetStreamCluster()
-
-	resp.StreamInfo = &StreamInfo{Created: mset.createdTime(), State: mset.state(), Config: mset.config(), Cluster: js.clusterInfo(mset.raftGroup())}
+	resp.StreamInfo = &StreamInfo{
+		Created: mset.createdTime(),
+		State:   mset.state(),
+		Config:  mset.config(),
+		Domain:  s.getOpts().JetStreamDomain,
+		Mirror:  mset.mirrorInfo(),
+		Sources: mset.sourcesInfo(),
+	}
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
 }
 
@@ -1702,13 +1713,15 @@ func (s *Server) jsStreamListRequest(sub *subscription, c *client, _ *Account, s
 	}
 
 	for _, mset := range msets[offset:] {
+		config := mset.config()
 		resp.Streams = append(resp.Streams, &StreamInfo{
 			Created: mset.createdTime(),
 			State:   mset.state(),
-			Config:  mset.config(),
+			Config:  config,
+			Domain:  s.getOpts().JetStreamDomain,
 			Mirror:  mset.mirrorInfo(),
-			Sources: mset.sourcesInfo()},
-		)
+			Sources: mset.sourcesInfo(),
+		})
 		if len(resp.Streams) >= JSApiListLimit {
 			break
 		}
@@ -1845,11 +1858,14 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
 	js, _ := s.getJetStreamCluster()
 
 	resp.StreamInfo = &StreamInfo{
-		Created: mset.createdTime(),
-		State:   mset.stateWithDetail(details),
-		Config:  config,
-		Domain:  s.getOpts().JetStreamDomain,
-		Cluster: js.clusterInfo(mset.raftGroup()),
+		Created:    mset.createdTime(),
+		State:      mset.stateWithDetail(details),
+		Config:     config,
+		Domain:     s.getOpts().JetStreamDomain,
+		Cluster:    js.clusterInfo(mset.raftGroup()),
+		Mirror:     mset.mirrorInfo(),
+		Sources:    mset.sourcesInfo(),
+		Alternates: js.streamAlternates(ci, config.Name),
 	}
 	if clusterWideConsCount > 0 {
 		resp.StreamInfo.State.Consumers = clusterWideConsCount
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 1931dcc3..612ae6fd 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -1231,12 +1231,26 @@ func (js *jetStream) processRemovePeer(peer string) {
 // Remove old peers after the new leader has caught up and taken over.
 // We are the stream leader.
 func (js *jetStream) removeOldPeers(mset *stream) {
+	// Make sure still valid.
+	mset.mu.Lock()
+	isValid := mset.qch != nil
+	mset.mu.Unlock()
+
+	if !isValid {
+		return
+	}
+
 	sa := mset.streamAssignment()
 	ci := js.clusterInfo(mset.raftGroup())
 
 	js.mu.Lock()
 	defer js.mu.Unlock()
 
+	// Make sure still valid.
+	if js.srv == nil || !js.srv.isRunning() {
+		return
+	}
+
 	cc, csa := js.cluster, sa.copyGroup()
 	numExpandedPeers := len(csa.Group.Peers)
 	csa.Group.Peers = csa.Group.Peers[:0]
@@ -1599,8 +1613,10 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	var mmtc <-chan time.Time
 
 	startMigrationMonitoring := func() {
-		mmt = time.NewTicker(250 * time.Millisecond)
-		mmtc = mmt.C
+		if mmt == nil {
+			mmt = time.NewTicker(250 * time.Millisecond)
+			mmtc = mmt.C
+		}
 	}
 
 	stopMigrationMonitoring := func() {
@@ -1686,12 +1702,10 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 					startMigrationMonitoring()
 				} else {
 					stopMigrationMonitoring()
-					// We delay here since consumer migration, while in most cases is faster, is done after the stream itself.
-					// When we remove the stream from the old peers this will remove the consumers which might still be in flight.
-					time.AfterFunc(500*time.Millisecond, func() { js.removeOldPeers(mset) })
+					// Delay before removing the old peers after assuming leadership.
+					time.AfterFunc(2*time.Second, func() { js.removeOldPeers(mset) })
 				}
 			}
-
 		case <-t.C:
 			doSnapshot()
 		case <-uch:
@@ -1715,23 +1729,24 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			if mset.hasCatchupPeers() {
 				mset.checkClusterInfo(ci)
 			}
-			// Track the new peers and check to make sure at least 1 is current and up to date.
+			// Track the new peers and check the ones that are current.
 			var newPeers []*PeerInfo
+			quorum, offline := mset.cfg.Replicas/2+1, 0
 			for _, r := range ci.Replicas {
-				if r.cluster != ci.Name && r.Current {
-					newPeers = append(newPeers, r)
-				}
-			}
-			if len(newPeers) == mset.cfg.Replicas {
-				for _, r := range newPeers {
+				if r.cluster != ci.Name {
 					if r.Current {
-						stopMigrationMonitoring()
-						// Transfer leadership for stream.
-						mset.raftNode().StepDown(r.peer)
-						break
+						newPeers = append(newPeers, r)
+					} else if r.Offline {
+						offline++
 					}
 				}
 			}
+			// If all are current we are good, or if we have some offline and we have a quorum.
+			if lnp := len(newPeers); lnp == mset.cfg.Replicas || (offline > 0 && lnp >= quorum) {
+				stopMigrationMonitoring()
+				// Transfer leadership for stream.
+				mset.raftNode().StepDown(newPeers[0].peer)
+			}
 		case err := <-restoreDoneCh:
 			// We have completed a restore from snapshot on this server. The stream assignment has
 			// already been assigned but the replicas will need to catch up out of band. Consumers
@@ -1789,6 +1804,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 					rg.setPreferred()
 					name, cfg := o.String(), o.config()
 					// Place our initial state here as well for assignment distribution.
+					state, _ := o.store.State()
 					ca := &consumerAssignment{
 						Group:   rg,
 						Stream:  sa.Config.Name,
@@ -1796,7 +1812,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 						Config:  &cfg,
 						Client:  sa.Client,
 						Created: o.createdTime(),
-						State:   o.readStoreState(),
+						State:   state,
 					}
 
 					// We make these compressed in case state is complex.
@@ -1865,39 +1881,6 @@ func (mset *stream) isMigrating() (bool, peerMigrateType) {
 	return true, oldPeerGroup
 }
 
-// Determine if we are migrating and if so if we are part of the old or new set.
-func (o *consumer) isMigrating() (bool, peerMigrateType) {
-	o.mu.RLock()
-	s, js, ca, mset := o.srv, o.js, o.ca, o.mset
-	o.mu.RUnlock()
-
-	if mset == nil {
-		return false, oldPeerGroup
-	}
-
-	sa := mset.streamAssignment()
-
-	js.mu.RLock()
-	defer js.mu.RUnlock()
-
-	// During migration we will always be R>1, even when we start R1.
-	// So if we do not have a group or node we no we are not migrating.
-	if sa == nil || ca == nil || ca.Group == nil || ca.Group.node == nil {
-		return false, oldPeerGroup
-	}
-	// The sign of migration is if our group peer count != configured replica count.
-	// We inherit this from the stream.
-	if len(ca.Group.Peers) == 1 || sa.Config.Replicas == len(ca.Group.Peers) {
-		return false, oldPeerGroup
-	}
-	// So we believe we are migrating here, need to determine if we are the old set or new set.
-	// We can shor circuit this based on our group assigned cluster vs our own.
-	if ca.Group.Cluster == s.cachedClusterName() {
-		return true, newPeerGroup
-	}
-	return true, oldPeerGroup
-}
-
 // resetClusteredState is called when a clustered stream had a sequence mismatch and needs to be reset.
 func (mset *stream) resetClusteredState(err error) bool {
 	mset.mu.RLock()
@@ -2405,6 +2388,7 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) bool {
 		js.mu.Lock()
 		if node := sa.Group.node; node != nil {
 			if node.Leader() {
+				node.UpdateKnownPeers(sa.Group.Peers)
 				node.StepDown()
 			}
 			node.ProposeRemovePeer(ourID)
@@ -2871,7 +2855,7 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 	js.mu.RLock()
 	s, cc := js.srv, js.cluster
-	accName, stream, consumer := ca.Client.serviceAccount(), ca.Stream, ca.Name
+	accName, stream, consumerName := ca.Client.serviceAccount(), ca.Stream, ca.Name
 	noMeta := cc == nil || cc.meta == nil
 	var ourID string
 	if !noMeta {
@@ -2895,7 +2879,7 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 			result := &consumerAssignmentResult{
 				Account:  accName,
 				Stream:   stream,
-				Consumer: consumer,
+				Consumer: consumerName,
 				Response: &JSApiConsumerCreateResponse{ApiResponse: ApiResponse{Type: JSApiConsumerCreateResponseType}},
 			}
 			result.Response.Error = NewJSNoAccountError()
@@ -2913,6 +2897,9 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 		return
 	}
 
+	// Might need this below.
+	numReplicas := sa.Config.Replicas
+
 	// Track if this existed already.
 	var wasExisting bool
 
@@ -2942,28 +2929,60 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 	if isMember {
 		js.processClusterCreateConsumer(ca, state, wasExisting)
 	} else {
+		// We need to be removed here, we are no longer assigned.
+		// Grab consumer if we have it.
+		var o *consumer
+		if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
+			o = mset.lookupConsumer(ca.Name)
+		}
+
 		// Check if we have a raft node running, meaning we are no longer part of the group but were.
 		js.mu.Lock()
 		if node := ca.Group.node; node != nil {
 			// We have one here even though we are not a member. This can happen on re-assignment.
 			s.Debugf("JetStream removing consumer '%s > %s > %s' from this server", sa.Client.serviceAccount(), sa.Config.Name, ca.Name)
 			if node.Leader() {
-				var nl string
-				if peers := ca.Group.Peers; len(peers) > 0 {
-					nl = peers[rand.Int31n(int32(len(peers)))]
+				s.Debugf("JetStream consumer '%s > %s > %s' is being removed and was the leader, will perform stepdown",
+					sa.Client.serviceAccount(), sa.Config.Name, ca.Name)
+
+				peers, cn := node.Peers(), s.cachedClusterName()
+				migrating := numReplicas != len(peers)
+
+				// Select a new peer to transfer to. If we are a migrating make sure its from the new cluster.
+				var npeer string
+				for _, r := range peers {
+					if !r.Current {
+						continue
+					}
+					if !migrating {
+						npeer = r.ID
+						break
+					} else if sir, ok := s.nodeToInfo.Load(r.ID); ok && sir != nil {
+						si := sir.(nodeInfo)
+						if si.cluster != cn {
+							npeer = r.ID
+							break
+						}
+					}
+				}
+				// Clear the raftnode from our consumer so that a subsequent o.delete will not also issue a stepdown.
+				if o != nil {
+					o.clearRaftNode()
 				}
-				node.StepDown(nl)
+				// Manually handle the stepdown and deletion of the node.
+				node.UpdateKnownPeers(ca.Group.Peers)
+				node.StepDown(npeer)
+				node.Delete()
+			} else {
+				node.UpdateKnownPeers(ca.Group.Peers)
 			}
-			node.ProposeRemovePeer(ourID)
 		}
 		ca.Group.node = nil
 		ca.err = nil
 		js.mu.Unlock()
 
-		if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
-			if o := mset.lookupConsumer(ca.Name); o != nil {
-				o.deleteWithoutAdvisory()
-			}
+		if o != nil {
+			o.deleteWithoutAdvisory()
 		}
 	}
 }
@@ -3036,14 +3055,23 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 		return
 	}
 
+	// Check if we already have this consumer running.
+	o := mset.lookupConsumer(ca.Name)
+
 	if !alreadyRunning {
 		// Process the raft group and make sure its running if needed.
 		js.createRaftGroup(acc.GetName(), rg, mset.config().Storage)
+	} else {
+		// If we are clustered update the known peers.
+		js.mu.RLock()
+		if node := rg.node; node != nil {
+			node.UpdateKnownPeers(ca.Group.Peers)
+		}
+		js.mu.RUnlock()
 	}
 
 	// Check if we already have this consumer running.
 	var didCreate bool
-	o := mset.lookupConsumer(ca.Name)
 	if o == nil {
 		// Add in the consumer if needed.
 		o, err = mset.addConsumerWithAssignment(ca.Config, ca.Name, ca)
@@ -3065,15 +3093,32 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 		}
 		// Check if we already had a consumer assignment and its still pending.
 		cca, oca := ca, o.consumerAssignment()
+		o.mu.Lock()
+		leader := o.isLeader()
+		o.mu.Unlock()
+
+		var sendState bool
 		js.mu.Lock()
-		if oca != nil && !oca.responded {
-			// We can't over ride info for replying here otherwise leader once elected can not respond.
-			// So just update Config, leave off client and reply to the originals.
-			cac := *oca
-			cac.Config = ca.Config
-			cca = &cac
+		if oca != nil {
+			if !oca.responded {
+				// We can't override info for replying here otherwise leader once elected can not respond.
+				// So just update Config, leave off client and reply to the originals.
+				cac := *oca
+				cac.Config = ca.Config
+				cca = &cac
+			}
+			// If we look like we are scaling up, let's send our current state to the group.
+			sendState = len(ca.Group.Peers) > len(oca.Group.Peers) && leader
 		}
+		n := rg.node
 		js.mu.Unlock()
+
+		if sendState && n != nil {
+			if snap, err := o.store.EncodedState(); err == nil {
+				n.SendSnapshot(snap)
+			}
+		}
+
 		// Set CA for our consumer.
 		o.setConsumerAssignment(cca)
 		s.Debugf("JetStream cluster, consumer was already running")
@@ -3149,7 +3194,7 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 			js.processConsumerLeaderChange(o, true)
 		} else {
 			if !alreadyRunning {
-				s.startGoRoutine(func() { js.monitorConsumer(o, ca) })
+				s.startGoRoutine(func() { js.monitorConsumer(o, ca, !didCreate) })
 			}
 			// Process if existing.
 			if wasExisting && (o.isLeader() || (!didCreate && rg.node.GroupLeader() == _EMPTY_)) {
@@ -3286,6 +3331,15 @@ func (o *consumer) raftGroup() *raftGroup {
 	return o.ca.Group
 }
 
+func (o *consumer) clearRaftNode() {
+	if o == nil {
+		return
+	}
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.node = nil
+}
+
 func (o *consumer) raftNode() RaftNode {
 	if o == nil {
 		return nil
@@ -3295,7 +3349,7 @@ func (o *consumer) raftNode() RaftNode {
 	return o.node
 }
 
-func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
+func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment, sendSnapshot bool) {
 	s, n := js.server(), o.raftNode()
 	defer s.grWG.Done()
 
@@ -3338,12 +3392,7 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 			needSnap = nb > 0 && ne >= compactNumMin || nb > compactSizeMin
 		}
 
-		state := o.readStoreState()
-		if state == nil {
-			s.Warnf("Failed to get state for '%s > %s > %s' [%s]", o.acc.Name, ca.Stream, ca.Name, n.Group())
-		}
-
-		if snap := encodeConsumerState(state); !bytes.Equal(lastSnap, snap) || needSnap {
+		if snap, err := o.store.EncodedState(); err == nil && !bytes.Equal(lastSnap, snap) || needSnap {
 			if err := n.InstallSnapshot(snap); err == nil {
 				lastSnap, lastSnapTime = snap, time.Now()
 			} else {
@@ -3352,29 +3401,20 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 		}
 	}
 
+	// This is triggered during a scale up from 1 to clustered mode. We need the new followers to catchup,
+	// similar to how we trigger the catchup mechanism post a backup/restore. It's ok to do here and preferred
+	// over waiting to be elected, this just queues it up for the new members to see first and trigger the above
+	// RAFT layer catchup mechanism.
+	if sendSnapshot && o != nil {
+		if snap, err := o.store.EncodedState(); err == nil {
+			n.SendSnapshot(snap)
+		}
+	}
+
 	// Track if we are leader.
 	var isLeader bool
 	recovering := true
 
-	// For migration tracking.
-	var migrating bool
-	var peerGroup peerMigrateType
-	var mmt *time.Ticker
-	var mmtc <-chan time.Time
-
-	startMigrationMonitoring := func() {
-		mmt = time.NewTicker(250 * time.Millisecond)
-		mmtc = mmt.C
-	}
-
-	stopMigrationMonitoring := func() {
-		if mmt != nil {
-			mmt.Stop()
-			mmtc = nil
-		}
-	}
-	defer stopMigrationMonitoring()
-
 	for {
 		select {
 		case <-s.quitCh:
@@ -3411,41 +3451,7 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 			if err := js.processConsumerLeaderChange(o, isLeader); err == nil && isLeader {
 				doSnapshot(true)
 			}
-			// Check for migrations here. We set the state on the stream assignment update below.
-			if isLeader && migrating {
-				if peerGroup == oldPeerGroup {
-					startMigrationMonitoring()
-				} else {
-					stopMigrationMonitoring()
-				}
-			}
 		case <-uch:
-			// We get this when we have a new consumer assignment caused by an update. We want
-			// to know if we are migrating.
-			migrating, peerGroup = o.isMigrating()
-			// If we are migrating and in the old peer group and we are leader, monitor for the
-			// new peers to be caught up. We could not be leader yet, so we will do same check above
-			// on leadership change.
-			if isLeader && migrating && peerGroup == oldPeerGroup {
-				startMigrationMonitoring()
-			}
-		case <-mmtc:
-			if !isLeader {
-				// We are no longer leader, so not our job.
-				stopMigrationMonitoring()
-				continue
-			}
-			// Check to see that we have someone caught up.
-			ci := js.clusterInfo(o.raftGroup())
-			// Track the new peers and check to make sure at least 1 is current and up to date.
-			for _, r := range ci.Replicas {
-				if r.cluster != ci.Name && r.Current {
-					stopMigrationMonitoring()
-					// Transfer leadership for this consumer.
-					o.raftNode().StepDown(r.peer)
-					break
-				}
-			}
 		case <-t.C:
 			doSnapshot(false)
 		}
@@ -3661,17 +3667,6 @@ func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) err
 	// Tell consumer to switch leader status.
 	o.setLeader(isLeader)
 
-	// Synchronize others to our version of state.
-	if isLeader {
-		if n := o.raftNode(); n != nil {
-			if state, err := o.store.State(); err == nil && state != nil {
-				if snap := encodeConsumerState(state); len(snap) > 0 {
-					n.SendSnapshot(snap)
-				}
-			}
-		}
-	}
-
 	if !isLeader || hasResponded {
 		if isLeader {
 			o.clearInitialInfo()
@@ -4434,11 +4429,18 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 	isMoveRequest := !reflect.DeepEqual(osa.Config.Placement, newCfg.Placement)
 
 	// Check for replica changes.
-	isReplicaChange := newCfg.Replicas != len(rg.Peers)
+	isReplicaChange := newCfg.Replicas != osa.Config.Replicas
 
 	// We stage consumer updates and do them after the stream update.
 	var consumers []*consumerAssignment
 
+	// Check if this is a move request and we are already moving this stream.
+	if isMoveRequest && osa.Config.Replicas != len(rg.Peers) {
+		resp.Error = NewJSStreamMoveInProgressError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+		return
+	}
+
 	// Can not move and scale at same time.
 	if isMoveRequest && isReplicaChange {
 		resp.Error = NewJSStreamMoveAndScaleError()
@@ -6065,6 +6067,59 @@ func (mset *stream) checkClusterInfo(ci *ClusterInfo) {
 	}
 }
 
+// Return a list of alternates, ranked by preference order to the request, of stream mirrors.
+// This allows clients to select or get more information about read replicas that could be a
+// better option to connect to versus the original source.
+func (js *jetStream) streamAlternates(ci *ClientInfo, stream string) []StreamAlternate {
+	if js == nil {
+		return nil
+	}
+
+	js.mu.RLock()
+	defer js.mu.RUnlock()
+
+	s, cc := js.srv, js.cluster
+	// Track our domain.
+	domain := s.getOpts().JetStreamDomain
+
+	// No clustering just return nil.
+	if cc == nil {
+		return nil
+	}
+	acc, _ := s.LookupAccount(ci.serviceAccount())
+	if acc == nil {
+		return nil
+	}
+
+	// Collect our ordering first for clusters.
+	weights := make(map[string]int)
+	all := []string{ci.Cluster}
+	all = append(all, ci.Alternates...)
+
+	for i := 0; i < len(all); i++ {
+		weights[all[i]] = len(all) - i
+	}
+
+	var alts []StreamAlternate
+	for _, sa := range cc.streams[acc.Name] {
+		// Add in ourselves and any mirrors.
+		if sa.Config.Name == stream || (sa.Config.Mirror != nil && sa.Config.Mirror.Name == stream) {
+			alts = append(alts, StreamAlternate{Name: sa.Config.Name, Domain: domain, Cluster: sa.Group.Cluster})
+		}
+	}
+	// If just us don't fill in.
+	if len(alts) == 1 {
+		return nil
+	}
+
+	// Sort based on our weights that originate from the request itself.
+	sort.Slice(alts, func(i, j int) bool {
+		return weights[alts[i].Cluster] > weights[alts[j].Cluster]
+	})
+
+	return alts
+}
+
 func (mset *stream) handleClusterStreamInfoRequest(sub *subscription, c *client, _ *Account, subject, reply string, _ []byte) {
 	mset.mu.RLock()
 	sysc, js, sa, config := mset.sysc, mset.srv.js, mset.sa, mset.cfg
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index 1967beab..f91ea738 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -302,6 +302,9 @@ const (
 	// JSStreamMoveAndScaleErr can not move and scale a stream in a single update
 	JSStreamMoveAndScaleErr ErrorIdentifier = 10123
 
+	// JSStreamMoveInProgress can not move a stream already being migrated
+	JSStreamMoveInProgress ErrorIdentifier = 10124
+
 	// JSStreamMsgDeleteFailedF Generic message deletion failure error string ({err})
 	JSStreamMsgDeleteFailedF ErrorIdentifier = 10057
 
@@ -473,6 +476,7 @@ var (
 		JSStreamMirrorNotUpdatableErr:              {Code: 400, ErrCode: 10055, Description: "stream mirror configuration can not be updated"},
 		JSStreamMismatchErr:                        {Code: 400, ErrCode: 10056, Description: "stream name in subject does not match request"},
 		JSStreamMoveAndScaleErr:                    {Code: 400, ErrCode: 10123, Description: "can not move and scale a stream in a single update"},
+		JSStreamMoveInProgress:                     {Code: 400, ErrCode: 10124, Description: "can not move a stream already being migrated"},
 		JSStreamMsgDeleteFailedF:                   {Code: 500, ErrCode: 10057, Description: "{err}"},
 		JSStreamNameExistErr:                       {Code: 400, ErrCode: 10058, Description: "stream name already in use"},
 		JSStreamNotFoundErr:                        {Code: 404, ErrCode: 10059, Description: "stream not found"},
@@ -1631,6 +1635,16 @@ func NewJSStreamMoveAndScaleError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSStreamMoveAndScaleErr]
 }
 
+// NewJSStreamMoveInProgressError creates a new JSStreamMoveInProgress error: "can not move a stream already being migrated"
+func NewJSStreamMoveInProgressError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSStreamMoveInProgress]
+}
+
 // NewJSStreamMsgDeleteFailedError creates a new JSStreamMsgDeleteFailedF error: "{err}"
 func NewJSStreamMsgDeleteFailedError(err error, opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
diff --git a/server/memstore.go b/server/memstore.go
index e6010f7f..6b992fca 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -880,6 +880,12 @@ func (ms *memStore) Stop() error {
 	return nil
 }
 
+func (ms *memStore) isClosed() bool {
+	ms.mu.RLock()
+	defer ms.mu.RUnlock()
+	return ms.msgs == nil
+}
+
 func (ms *memStore) incConsumers() {
 	ms.mu.Lock()
 	ms.consumers++
@@ -895,40 +901,275 @@ func (ms *memStore) decConsumers() {
 }
 
 type consumerMemStore struct {
-	ms *memStore
+	mu     sync.Mutex
+	ms     *memStore
+	cfg    ConsumerConfig
+	state  ConsumerState
+	closed bool
 }
 
-func (ms *memStore) ConsumerStore(_ string, _ *ConsumerConfig) (ConsumerStore, error) {
+func (ms *memStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerStore, error) {
+	if ms == nil {
+		return nil, fmt.Errorf("memstore is nil")
+	}
+	if ms.isClosed() {
+		return nil, ErrStoreClosed
+	}
+	if cfg == nil || name == _EMPTY_ {
+		return nil, fmt.Errorf("bad consumer config")
+	}
 	ms.incConsumers()
-	return &consumerMemStore{ms}, nil
+	return &consumerMemStore{ms: ms, cfg: *cfg}, nil
 }
 
 func (ms *memStore) Snapshot(_ time.Duration, _, _ bool) (*SnapshotResult, error) {
 	return nil, fmt.Errorf("no impl")
 }
 
-// No-ops.
-func (os *consumerMemStore) Update(_ *ConsumerState) error                 { return nil }
-func (os *consumerMemStore) UpdateDelivered(_, _, _ uint64, _ int64) error { return nil }
-func (os *consumerMemStore) UpdateAcks(_, _ uint64) error                  { return nil }
-func (os *consumerMemStore) UpdateConfig(_ *ConsumerConfig) error          { return nil }
+func (o *consumerMemStore) Update(state *ConsumerState) error {
+	// Sanity checks.
+	if state.AckFloor.Consumer > state.Delivered.Consumer {
+		return fmt.Errorf("bad ack floor for consumer")
+	}
+	if state.AckFloor.Stream > state.Delivered.Stream {
+		return fmt.Errorf("bad ack floor for stream")
+	}
+
+	// Copy to our state.
+	var pending map[uint64]*Pending
+	var redelivered map[uint64]uint64
+	if len(state.Pending) > 0 {
+		pending = make(map[uint64]*Pending, len(state.Pending))
+		for seq, p := range state.Pending {
+			pending[seq] = &Pending{p.Sequence, p.Timestamp}
+		}
+		for seq := range pending {
+			if seq <= state.AckFloor.Stream || seq > state.Delivered.Stream {
+				return fmt.Errorf("bad pending entry, sequence [%d] out of range", seq)
+			}
+		}
+	}
+	if len(state.Redelivered) > 0 {
+		redelivered = make(map[uint64]uint64, len(state.Redelivered))
+		for seq, dc := range state.Redelivered {
+			redelivered[seq] = dc
+		}
+	}
+
+	// Replace our state.
+	o.mu.Lock()
+
+	// Check to see if this is an outdated update.
+	if state.Delivered.Consumer < o.state.Delivered.Consumer {
+		o.mu.Unlock()
+		return fmt.Errorf("old update ignored")
+	}
+
+	o.state.Delivered = state.Delivered
+	o.state.AckFloor = state.AckFloor
+	o.state.Pending = pending
+	o.state.Redelivered = redelivered
+	o.mu.Unlock()
 
-func (os *consumerMemStore) Stop() error {
-	os.ms.decConsumers()
 	return nil
 }
 
-func (os *consumerMemStore) Delete() error {
-	return os.Stop()
+func (o *consumerMemStore) UpdateDelivered(dseq, sseq, dc uint64, ts int64) error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	if dc != 1 && o.cfg.AckPolicy == AckNone {
+		return ErrNoAckPolicy
+	}
+
+	if dseq <= o.state.AckFloor.Consumer {
+		return nil
+	}
+
+	// See if we expect an ack for this.
+	if o.cfg.AckPolicy != AckNone {
+		// Need to create pending records here.
+		if o.state.Pending == nil {
+			o.state.Pending = make(map[uint64]*Pending)
+		}
+		var p *Pending
+		// Check for an update to a message already delivered.
+		if sseq <= o.state.Delivered.Stream {
+			if p = o.state.Pending[sseq]; p != nil {
+				p.Sequence, p.Timestamp = dseq, ts
+			}
+		} else {
+			// Add to pending.
+			o.state.Pending[sseq] = &Pending{dseq, ts}
+		}
+		// Update delivered as needed.
+		if dseq > o.state.Delivered.Consumer {
+			o.state.Delivered.Consumer = dseq
+		}
+		if sseq > o.state.Delivered.Stream {
+			o.state.Delivered.Stream = sseq
+		}
+
+		if dc > 1 {
+			if o.state.Redelivered == nil {
+				o.state.Redelivered = make(map[uint64]uint64)
+			}
+			o.state.Redelivered[sseq] = dc - 1
+		}
+	} else {
+		// For AckNone just update delivered and ackfloor at the same time.
+		o.state.Delivered.Consumer = dseq
+		o.state.Delivered.Stream = sseq
+		o.state.AckFloor.Consumer = dseq
+		o.state.AckFloor.Stream = sseq
+	}
+
+	return nil
 }
-func (os *consumerMemStore) StreamDelete() error {
-	return os.Stop()
+
+func (o *consumerMemStore) UpdateAcks(dseq, sseq uint64) error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	if o.cfg.AckPolicy == AckNone {
+		return ErrNoAckPolicy
+	}
+	if len(o.state.Pending) == 0 || o.state.Pending[sseq] == nil {
+		return ErrStoreMsgNotFound
+	}
+
+	// On restarts the old leader may get a replay from the raft logs that are old.
+	if dseq <= o.state.AckFloor.Consumer {
+		return nil
+	}
+
+	// Check for AckAll here.
+	if o.cfg.AckPolicy == AckAll {
+		sgap := sseq - o.state.AckFloor.Stream
+		o.state.AckFloor.Consumer = dseq
+		o.state.AckFloor.Stream = sseq
+		for seq := sseq; seq > sseq-sgap; seq-- {
+			delete(o.state.Pending, seq)
+			if len(o.state.Redelivered) > 0 {
+				delete(o.state.Redelivered, seq)
+			}
+		}
+		return nil
+	}
+
+	// AckExplicit
+
+	// First delete from our pending state.
+	if p, ok := o.state.Pending[sseq]; ok {
+		delete(o.state.Pending, sseq)
+		dseq = p.Sequence // Use the original.
+	}
+	// Now remove from redelivered.
+	if len(o.state.Redelivered) > 0 {
+		delete(o.state.Redelivered, sseq)
+	}
+
+	if len(o.state.Pending) == 0 {
+		o.state.AckFloor.Consumer = o.state.Delivered.Consumer
+		o.state.AckFloor.Stream = o.state.Delivered.Stream
+	} else if dseq == o.state.AckFloor.Consumer+1 {
+		first := o.state.AckFloor.Consumer == 0
+		o.state.AckFloor.Consumer = dseq
+		o.state.AckFloor.Stream = sseq
+
+		if !first && o.state.Delivered.Consumer > dseq {
+			for ss := sseq + 1; ss < o.state.Delivered.Stream; ss++ {
+				if p, ok := o.state.Pending[ss]; ok {
+					if p.Sequence > 0 {
+						o.state.AckFloor.Consumer = p.Sequence - 1
+						o.state.AckFloor.Stream = ss - 1
+					}
+					break
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func (o *consumerMemStore) UpdateConfig(cfg *ConsumerConfig) error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	// This is mostly unchecked here. We are assuming the upper layers have done sanity checking.
+	o.cfg = *cfg
+	return nil
+}
+
+func (o *consumerMemStore) Stop() error {
+	o.mu.Lock()
+	o.closed = true
+	ms := o.ms
+	o.mu.Unlock()
+	ms.decConsumers()
+	return nil
+}
+
+func (o *consumerMemStore) Delete() error {
+	return o.Stop()
 }
 
-func (os *consumerMemStore) State() (*ConsumerState, error) { return nil, nil }
+func (o *consumerMemStore) StreamDelete() error {
+	return o.Stop()
+}
+
+func (o *consumerMemStore) State() (*ConsumerState, error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	if o.closed {
+		return nil, ErrStoreClosed
+	}
+
+	state := &ConsumerState{}
+
+	state.Delivered = o.state.Delivered
+	state.AckFloor = o.state.AckFloor
+	if len(o.state.Pending) > 0 {
+		state.Pending = o.copyPending()
+	}
+	if len(o.state.Redelivered) > 0 {
+		state.Redelivered = o.copyRedelivered()
+	}
+	return state, nil
+}
+
+// EncodeState for this consumer store.
+func (o *consumerMemStore) EncodedState() ([]byte, error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	if o.closed {
+		return nil, ErrStoreClosed
+	}
+
+	return encodeConsumerState(&o.state), nil
+}
+
+func (o *consumerMemStore) copyPending() map[uint64]*Pending {
+	pending := make(map[uint64]*Pending, len(o.state.Pending))
+	for seq, p := range o.state.Pending {
+		pending[seq] = &Pending{p.Sequence, p.Timestamp}
+	}
+	return pending
+}
+
+func (o *consumerMemStore) copyRedelivered() map[uint64]uint64 {
+	redelivered := make(map[uint64]uint64, len(o.state.Redelivered))
+	for seq, dc := range o.state.Redelivered {
+		redelivered[seq] = dc
+	}
+	return redelivered
+}
 
 // Type returns the type of the underlying store.
-func (os *consumerMemStore) Type() StorageType { return MemoryStorage }
+func (o *consumerMemStore) Type() StorageType { return MemoryStorage }
 
 // Templates
 type templateMemStore struct{}
diff --git a/server/raft.go b/server/raft.go
index c33bf3c6..8a3e9737 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -56,6 +56,7 @@ type RaftNode interface {
 	ID() string
 	Group() string
 	Peers() []*Peer
+	UpdateKnownPeers(knownPeers []string)
 	ProposeAddPeer(peer string) error
 	ProposeRemovePeer(peer string) error
 	AdjustClusterSize(csz int) error
@@ -221,6 +222,7 @@ type catchupState struct {
 type lps struct {
 	ts int64
 	li uint64
+	kp bool // marks as known peer.
 }
 
 const (
@@ -255,14 +257,13 @@ var (
 	errNilCfg            = errors.New("raft: no config given")
 	errCorruptPeers      = errors.New("raft: corrupt peer state")
 	errEntryLoadFailed   = errors.New("raft: could not load entry from WAL")
-	errEntryStoreFailed  = errors.New("raft: could not storeentry to WAL")
+	errEntryStoreFailed  = errors.New("raft: could not store entry to WAL")
 	errNodeClosed        = errors.New("raft: node is closed")
 	errBadSnapName       = errors.New("raft: snapshot name could not be parsed")
 	errNoSnapAvailable   = errors.New("raft: no snapshot available")
 	errCatchupsRunning   = errors.New("raft: snapshot can not be installed while catchups running")
 	errSnapshotCorrupt   = errors.New("raft: snapshot corrupt")
 	errTooManyPrefs      = errors.New("raft: stepdown requires at most one preferred new leader")
-	errStepdownNoPeer    = errors.New("raft: stepdown failed, could not match new leader")
 	errNoPeerState       = errors.New("raft: no peerstate")
 	errAdjustBootCluster = errors.New("raft: can not adjust boot peer size on established group")
 	errLeaderLen         = fmt.Errorf("raft: leader should be exactly %d bytes", idLen)
@@ -284,7 +285,7 @@ func (s *Server) bootstrapRaftNode(cfg *RaftConfig, knownPeers []string, allPeer
 	expected := len(knownPeers)
 	// We need to adjust this is all peers are not known.
 	if !allPeersKnown {
-		s.Debugf("Determining expected peer size for JetStream metacontroller")
+		s.Debugf("Determining expected peer size for JetStream meta group")
 		if expected < 2 {
 			expected = 2
 		}
@@ -461,12 +462,12 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig) (RaftNode, error
 	n.apply.push(nil)
 
 	// Make sure to track ourselves.
-	n.trackPeer(n.id)
+	n.peers[n.id] = &lps{time.Now().UnixNano(), 0, true}
 	// Track known peers
 	for _, peer := range ps.knownPeers {
-		// Set these to 0 to start.
+		// Set these to 0 to start but mark as known peer.
 		if peer != n.id {
-			n.peers[peer] = &lps{0, 0}
+			n.peers[peer] = &lps{0, 0, true}
 		}
 	}
 
@@ -685,6 +686,17 @@ func (n *raft) ProposeAddPeer(peer string) error {
 	return nil
 }
 
+// As a leader if we are proposing to remove a peer assume its already gone.
+func (n *raft) doRemovePeerAsLeader(peer string) {
+	n.Lock()
+	if _, ok := n.peers[peer]; ok {
+		delete(n.peers, peer)
+		// We should decrease our cluster size since we are tracking this peer and the peer is most likely already gone.
+		n.adjustClusterSizeAndQuorum()
+	}
+	n.Unlock()
+}
+
 // ProposeRemovePeer is called to remove a peer from the group.
 func (n *raft) ProposeRemovePeer(peer string) error {
 	n.RLock()
@@ -700,6 +712,7 @@ func (n *raft) ProposeRemovePeer(peer string) error {
 
 	if isLeader {
 		prop.push(&Entry{EntryRemovePeer, []byte(peer)})
+		n.doRemovePeerAsLeader(peer)
 		return nil
 	}
 
@@ -1110,6 +1123,13 @@ func (n *raft) Leader() bool {
 	return isLeader
 }
 
+// Reports our catching up state.
+func (n *raft) isCatchingUp() bool {
+	n.RLock()
+	defer n.RUnlock()
+	return n.catchup != nil
+}
+
 // Lock should be held.
 func (n *raft) isCurrent() bool {
 	// First check if we match commit and applied.
@@ -1201,9 +1221,14 @@ func (n *raft) StepDown(preferred ...string) error {
 	// See if we have up to date followers.
 	nowts := time.Now().UnixNano()
 	maybeLeader := noLeader
-	if len(preferred) > 0 && preferred[0] != _EMPTY_ {
-		maybeLeader = preferred[0]
+	if len(preferred) > 0 {
+		if preferred[0] != _EMPTY_ {
+			maybeLeader = preferred[0]
+		} else {
+			preferred = nil
+		}
 	}
+
 	for peer, ps := range n.peers {
 		// If not us and alive and caughtup.
 		if peer != n.id && (nowts-ps.ts) < int64(hbInterval*3) {
@@ -1222,7 +1247,7 @@ func (n *raft) StepDown(preferred ...string) error {
 	n.Unlock()
 
 	if len(preferred) > 0 && maybeLeader == noLeader {
-		return errStepdownNoPeer
+		n.debug("Can not transfer to preferred peer %q", preferred[0])
 	}
 
 	if maybeLeader != noLeader {
@@ -1253,9 +1278,18 @@ func (n *raft) campaign() error {
 	if n.state == Leader {
 		return errAlreadyLeader
 	}
-	n.lxfer = true
 	n.resetElect(randCampaignTimeout())
+	return nil
+}
 
+// xferCampaign will have our node start an immediate leadership vote.
+// Lock should be held.
+func (n *raft) xferCampaign() error {
+	n.debug("Starting transfer campaign")
+	if n.state == Leader {
+		return errAlreadyLeader
+	}
+	n.resetElect(10 * time.Millisecond)
 	return nil
 }
 
@@ -1318,6 +1352,20 @@ func (n *raft) Peers() []*Peer {
 	return peers
 }
 
+// Update our known set of peers.
+func (n *raft) UpdateKnownPeers(knownPeers []string) {
+	n.Lock()
+	ps := &peerState{knownPeers, len(knownPeers), n.extSt}
+	n.processPeerState(ps)
+	isLeader := n.state == Leader
+	n.Unlock()
+
+	// If we are the leader send this update out as well.
+	if isLeader {
+		n.sendPeerState()
+	}
+}
+
 func (n *raft) ApplyQ() *ipQueue         { return n.apply } // queue of *CommittedEntry
 func (n *raft) LeadChangeC() <-chan bool { return n.leadc }
 func (n *raft) QuitC() <-chan struct{}   { return n.quit }
@@ -1574,6 +1622,8 @@ func (n *raft) runAsFollower() {
 			} else if n.isObserver() {
 				n.resetElect(48 * time.Hour)
 				n.debug("Not switching to candidate, observer only")
+			} else if n.isCatchingUp() {
+				n.debug("Not switching to candidate, catching up")
 			} else {
 				n.switchToCandidate()
 				return
@@ -1888,6 +1938,9 @@ func (n *raft) runAsLeader() {
 			sz := 0
 			for i, bi := range es {
 				b := bi.(*Entry)
+				if b.Type == EntryRemovePeer {
+					n.doRemovePeerAsLeader(string(b.Data))
+				}
 				entries = append(entries, b)
 				sz += len(b.Data) + 1
 				if i != len(es)-1 && sz < maxBatch && len(entries) < math.MaxUint16 {
@@ -2238,18 +2291,20 @@ func (n *raft) applyCommit(index uint64) error {
 				delete(n.removed, newPeer)
 			}
 
-			if _, ok := n.peers[newPeer]; !ok {
+			if lp, ok := n.peers[newPeer]; !ok {
 				// We are not tracking this one automatically so we need to bump cluster size.
-				n.peers[newPeer] = &lps{time.Now().UnixNano(), 0}
-				if n.csz < len(n.peers) {
-					n.debug("Expanding our clustersize: %d -> %d", n.csz, len(n.peers))
-					n.csz = len(n.peers)
-					n.qn = n.csz/2 + 1
-				}
+				n.peers[newPeer] = &lps{time.Now().UnixNano(), 0, true}
+			} else {
+				// Mark as added.
+				lp.kp = true
 			}
+			// Adjust cluster size and quorum if needed.
+			n.adjustClusterSizeAndQuorum()
+			// Write out our new state.
 			n.writePeerState(&peerState{n.peerNames(), n.csz, n.extSt})
 			// We pass these up as well.
 			committed = append(committed, e)
+
 		case EntryRemovePeer:
 			peer := string(e.Data)
 			n.debug("Removing peer %q", peer)
@@ -2261,13 +2316,11 @@ func (n *raft) applyCommit(index uint64) error {
 			n.removed[peer] = peer
 
 			if _, ok := n.peers[peer]; ok {
-				// We should decrease our cluster size since we are tracking this peer.
 				delete(n.peers, peer)
-				if n.csz != len(n.peers) {
-					n.debug("Decreasing our clustersize: %d -> %d", n.csz, len(n.peers))
-					n.csz = len(n.peers)
-					n.qn = n.csz/2 + 1
-				}
+				// We should decrease our cluster size since we are tracking this peer.
+				n.adjustClusterSizeAndQuorum()
+				// Write out our new state.
+				n.writePeerState(&peerState{n.peerNames(), n.csz, n.extSt})
 			}
 
 			// If this is us and we are the leader we should attempt to stepdown.
@@ -2275,8 +2328,6 @@ func (n *raft) applyCommit(index uint64) error {
 				n.stepdown.push(n.selectNextLeader())
 			}
 
-			// Write out our new state.
-			n.writePeerState(&peerState{n.peerNames(), n.csz, n.extSt})
 			// We pass these up as well.
 			committed = append(committed, e)
 		}
@@ -2323,13 +2374,13 @@ func (n *raft) trackResponse(ar *appendEntryResponse) {
 			// We have a quorum.
 			for index := n.commit + 1; index <= ar.index; index++ {
 				if err := n.applyCommit(index); err != nil {
+					n.error("Got an error apply commit for %d: %v", index, err)
 					break
 				}
 			}
 			sendHB = n.prop.len() == 0
 		}
 	}
-
 	n.Unlock()
 
 	if sendHB {
@@ -2337,50 +2388,54 @@ func (n *raft) trackResponse(ar *appendEntryResponse) {
 	}
 }
 
+// Used to adjust cluster size and peer count based on added official peers.
+// lock should be held.
+func (n *raft) adjustClusterSizeAndQuorum() {
+	pcsz, ncsz := n.csz, 0
+	for _, peer := range n.peers {
+		if peer.kp {
+			ncsz++
+		}
+	}
+	n.csz = ncsz
+	n.qn = n.csz/2 + 1
+
+	if ncsz > pcsz {
+		n.debug("Expanding our clustersize: %d -> %d", pcsz, ncsz)
+	} else if ncsz < pcsz {
+		n.debug("Decreasing our clustersize: %d -> %d", pcsz, ncsz)
+		if n.state == Leader {
+			go n.sendHeartbeat()
+		}
+	}
+}
+
 // Track interactions with this peer.
 func (n *raft) trackPeer(peer string) error {
 	n.Lock()
-	var needPeerUpdate, isRemoved bool
+	var needPeerAdd, isRemoved bool
 	if n.removed != nil {
 		_, isRemoved = n.removed[peer]
 	}
 	if n.state == Leader {
-		if _, ok := n.peers[peer]; !ok {
+		if lp, ok := n.peers[peer]; !ok || !lp.kp {
 			// Check if this peer had been removed previously.
-			if !isRemoved {
-				needPeerUpdate = true
-			}
+			needPeerAdd = !isRemoved
 		}
 	}
 	if ps := n.peers[peer]; ps != nil {
 		ps.ts = time.Now().UnixNano()
 	} else if !isRemoved {
-		n.peers[peer] = &lps{time.Now().UnixNano(), 0}
+		n.peers[peer] = &lps{time.Now().UnixNano(), 0, false}
 	}
 	n.Unlock()
 
-	if needPeerUpdate {
+	if needPeerAdd {
 		n.ProposeAddPeer(peer)
 	}
 	return nil
 }
 
-// Return the number of active peers for this group. We use this when we
-// are running as a candidate.
-// Lock should be held.
-func (n *raft) numActivePeers() int {
-	nap := 0
-	for id := range n.peers {
-		if sir, ok := n.s.nodeToInfo.Load(id); ok && sir != nil {
-			si := sir.(nodeInfo)
-			if !si.offline {
-				nap++
-			}
-		}
-	}
-	return nap
-}
-
 func (n *raft) runAsCandidate() {
 	n.Lock()
 	// Drain old responses.
@@ -2392,7 +2447,6 @@ func (n *raft) runAsCandidate() {
 
 	// We vote for ourselves.
 	votes := 1
-	won := false
 
 	for {
 		elect := n.electTimer()
@@ -2408,12 +2462,7 @@ func (n *raft) runAsCandidate() {
 		case <-n.quit:
 			return
 		case <-elect.C:
-			if won {
-				// we are here if we won the election but some server did not respond
-				n.switchToLeader()
-			} else {
-				n.switchToCandidate()
-			}
+			n.switchToCandidate()
 			return
 		case <-n.votes.ch:
 			// Because of drain() it is possible that we get nil from popOne().
@@ -2422,7 +2471,7 @@ func (n *raft) runAsCandidate() {
 				continue
 			}
 			n.RLock()
-			nterm, lxfer := n.term, n.lxfer
+			nterm := n.term
 			n.RUnlock()
 
 			if vresp.granted && nterm >= vresp.term {
@@ -2430,18 +2479,9 @@ func (n *raft) runAsCandidate() {
 				n.trackPeer(vresp.peer)
 				votes++
 				if n.wonElection(votes) {
-					if votes == n.numActivePeers() || lxfer {
-						// Become LEADER if we have won and gotten a quorum with everyone we should hear from.
-						n.switchToLeader()
-						return
-					} else {
-						// Not everyone is in this quorum, yet?
-						// Wait for the remaining responses and become leader once everyone did.
-						// Or Wait until after the election timeout and become leader then.
-						// In case another server responds with vresp.granted==false and vresp.term > n.term,
-						// we will start all over again.
-						won = true
-					}
+					// Become LEADER if we have won and gotten a quorum with everyone we should hear from.
+					n.switchToLeader()
+					return
 				}
 			} else if vresp.term > nterm {
 				// if we observe a bigger term, we should start over again or risk forming a quorum fully knowing
@@ -2452,6 +2492,7 @@ func (n *raft) runAsCandidate() {
 				n.writeTermVote()
 				n.debug("Stepping down from candidate, detected higher term: %d vs %d", vresp.term, n.term)
 				n.stepdown.push(noLeader)
+				n.lxfer = false
 				n.Unlock()
 				return
 			}
@@ -2598,7 +2639,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		if ps := n.peers[ae.leader]; ps != nil {
 			ps.ts = time.Now().UnixNano()
 		} else {
-			n.peers[ae.leader] = &lps{time.Now().UnixNano(), 0}
+			n.peers[ae.leader] = &lps{time.Now().UnixNano(), 0, true}
 		}
 	}
 
@@ -2774,16 +2815,17 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 				if isNew {
 					maybeLeader := string(e.Data)
 					if maybeLeader == n.id && !n.observer && !n.paused {
-						n.campaign()
+						n.lxfer = true
+						n.xferCampaign()
 					}
 				}
 			case EntryAddPeer:
 				if newPeer := string(e.Data); len(newPeer) == idLen {
-					// Track directly
+					// Track directly, but wait for commit to be official
 					if ps := n.peers[newPeer]; ps != nil {
 						ps.ts = time.Now().UnixNano()
 					} else {
-						n.peers[newPeer] = &lps{time.Now().UnixNano(), 0}
+						n.peers[newPeer] = &lps{time.Now().UnixNano(), 0, false}
 					}
 				}
 			}
@@ -2821,9 +2863,10 @@ func (n *raft) processPeerState(ps *peerState) {
 	n.peers = make(map[string]*lps)
 	for _, peer := range ps.knownPeers {
 		if lp := old[peer]; lp != nil {
+			lp.kp = true
 			n.peers[peer] = lp
 		} else {
-			n.peers[peer] = &lps{0, 0}
+			n.peers[peer] = &lps{0, 0, true}
 		}
 	}
 	n.debug("Update peers from leader to %+v", n.peers)
@@ -2837,7 +2880,7 @@ func (n *raft) processAppendEntryResponse(ar *appendEntryResponse) {
 	if ar.success {
 		n.trackResponse(ar)
 	} else if ar.term > n.term {
-		// False here, check to make sure they do not have a higher term.
+		// False here and they have a higher term.
 		n.term = ar.term
 		n.vote = noVote
 		n.writeTermVote()
@@ -2880,9 +2923,10 @@ func (n *raft) storeToWAL(ae *appendEntry) error {
 		if index > seq {
 			// Reset to last before this one.
 			if ae, err := n.loadEntry(seq - 1); err == nil && ae != nil {
+				nl := n.selectNextLeader()
 				n.truncateWAL(ae.pterm, ae.pindex)
 				if n.state == Leader {
-					n.stepdown.push(n.selectNextLeader())
+					n.stepdown.push(nl)
 				}
 			} else {
 				panic(fmt.Sprintf("[%s | %s] Wrong index, ae is %+v, seq is %d, n.pindex is %d\n\n", n.s, n.group, ae, seq, n.pindex))
@@ -2988,8 +3032,10 @@ func decodePeerState(buf []byte) (*peerState, error) {
 // Lock should be held.
 func (n *raft) peerNames() []string {
 	var peers []string
-	for peer := range n.peers {
-		peers = append(peers, peer)
+	for name, peer := range n.peers {
+		if peer.kp {
+			peers = append(peers, name)
+		}
 	}
 	return peers
 }
@@ -3006,6 +3052,7 @@ func (n *raft) sendPeerState() {
 	n.sendAppendEntry([]*Entry{{EntryPeerState, encodePeerState(n.currentPeerState())}})
 }
 
+// Send a heartbeat.
 func (n *raft) sendHeartbeat() {
 	n.sendAppendEntry(nil)
 }
@@ -3420,7 +3467,6 @@ func (n *raft) switchToCandidate() {
 	if n.state != Candidate {
 		n.debug("Switching to candidate")
 	} else {
-		n.lxfer = false
 		if n.lostQuorumLocked() && time.Since(n.llqrt) > 20*time.Second {
 			// We signal to the upper layers such that can alert on quorum lost.
 			n.updateLeadChange(false)
@@ -3441,7 +3487,8 @@ func (n *raft) switchToLeader() {
 		return
 	}
 	n.debug("Switching to leader")
-	n.updateLeader(n.id)
+
 	n.lxfer = false
+	n.updateLeader(n.id)
 	n.switchState(Leader)
 }
diff --git a/server/store.go b/server/store.go
index 5aa91949..10fc10a7 100644
--- a/server/store.go
+++ b/server/store.go
@@ -14,6 +14,7 @@
 package server
 
 import (
+	"encoding/binary"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -170,6 +171,7 @@ type ConsumerStore interface {
 	UpdateConfig(cfg *ConsumerConfig) error
 	Update(*ConsumerState) error
 	State() (*ConsumerState, error)
+	EncodedState() ([]byte, error)
 	Type() StorageType
 	Stop() error
 	Delete() error
@@ -196,6 +198,68 @@ type ConsumerState struct {
 	Redelivered map[uint64]uint64 `json:"redelivered,omitempty"`
 }
 
+func encodeConsumerState(state *ConsumerState) []byte {
+	var hdr [seqsHdrSize]byte
+	var buf []byte
+
+	maxSize := seqsHdrSize
+	if lp := len(state.Pending); lp > 0 {
+		maxSize += lp*(3*binary.MaxVarintLen64) + binary.MaxVarintLen64
+	}
+	if lr := len(state.Redelivered); lr > 0 {
+		maxSize += lr*(2*binary.MaxVarintLen64) + binary.MaxVarintLen64
+	}
+	if maxSize == seqsHdrSize {
+		buf = hdr[:seqsHdrSize]
+	} else {
+		buf = make([]byte, maxSize)
+	}
+
+	// Write header
+	buf[0] = magic
+	buf[1] = 2
+
+	n := hdrLen
+	n += binary.PutUvarint(buf[n:], state.AckFloor.Consumer)
+	n += binary.PutUvarint(buf[n:], state.AckFloor.Stream)
+	n += binary.PutUvarint(buf[n:], state.Delivered.Consumer)
+	n += binary.PutUvarint(buf[n:], state.Delivered.Stream)
+	n += binary.PutUvarint(buf[n:], uint64(len(state.Pending)))
+
+	asflr := state.AckFloor.Stream
+	adflr := state.AckFloor.Consumer
+
+	// These are optional, but always write len. This is to avoid a truncate inline.
+	if len(state.Pending) > 0 {
+		// To save space we will use now rounded to seconds to be base timestamp.
+		mints := time.Now().Round(time.Second).Unix()
+		// Write minimum timestamp we found from above.
+		n += binary.PutVarint(buf[n:], mints)
+
+		for k, v := range state.Pending {
+			n += binary.PutUvarint(buf[n:], k-asflr)
+			n += binary.PutUvarint(buf[n:], v.Sequence-adflr)
+			// Downsample to seconds to save on space.
+			// Subsecond resolution not needed for recovery etc.
+			ts := v.Timestamp / 1_000_000_000
+			n += binary.PutVarint(buf[n:], mints-ts)
+		}
+	}
+
+	// We always write the redelivered len.
+	n += binary.PutUvarint(buf[n:], uint64(len(state.Redelivered)))
+
+	// We expect these to be small.
+	if len(state.Redelivered) > 0 {
+		for k, v := range state.Redelivered {
+			n += binary.PutUvarint(buf[n:], k-asflr)
+			n += binary.PutUvarint(buf[n:], v)
+		}
+	}
+
+	return buf[:n]
+}
+
 // Represents a pending message for explicit ack or ack all.
 // Sequence is the original consumer sequence.
 type Pending struct {
diff --git a/server/stream.go b/server/stream.go
index 54ba8605..b0482884 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -96,13 +96,20 @@ type PubAck struct {
 
 // StreamInfo shows config and current state for this stream.
 type StreamInfo struct {
-	Config  StreamConfig        `json:"config"`
-	Created time.Time           `json:"created"`
-	State   StreamState         `json:"state"`
-	Domain  string              `json:"domain,omitempty"`
-	Cluster *ClusterInfo        `json:"cluster,omitempty"`
-	Mirror  *StreamSourceInfo   `json:"mirror,omitempty"`
-	Sources []*StreamSourceInfo `json:"sources,omitempty"`
+	Config     StreamConfig        `json:"config"`
+	Created    time.Time           `json:"created"`
+	State      StreamState         `json:"state"`
+	Domain     string              `json:"domain,omitempty"`
+	Cluster    *ClusterInfo        `json:"cluster,omitempty"`
+	Mirror     *StreamSourceInfo   `json:"mirror,omitempty"`
+	Sources    []*StreamSourceInfo `json:"sources,omitempty"`
+	Alternates []StreamAlternate   `json:"alternates,omitempty"`
+}
+
+type StreamAlternate struct {
+	Name    string `json:"name"`
+	Domain  string `json:"domain,omitempty"`
+	Cluster string `json:"cluster"`
 }
 
 // ClusterInfo shows information about the underlying set of servers


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 8697fc10..d871fcc9 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -2601,7 +2601,7 @@ func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 	}
 
 	// Create another consumer as well and give it a non-simplistic state.
-	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy, AckWait: 5 * time.Second})
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy, AckWait: 7500 * time.Millisecond})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -2791,7 +2791,7 @@ func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 
 	// Check that redelivered come in now..
 	redelivered := 50/3 + 1
-	fetchMsgs(t, jsub, redelivered, 5*time.Second)
+	fetchMsgs(t, jsub, redelivered, 10*time.Second)
 
 	// Now make sure the other server was properly caughtup.
 	// Need to call this by hand for now.
@@ -3407,11 +3407,14 @@ func TestJetStreamClusterPeerRemovalAPI(t *testing.T) {
 		t.Fatalf("Expected advisory about %s being removed, got %+v", rs.Name(), adv)
 	}
 
-	for _, s := range ml.JetStreamClusterPeers() {
-		if s == rs.Name() {
-			t.Fatalf("Still in the peer list")
+	checkFor(t, 2*time.Second, 250*time.Millisecond, func() error {
+		for _, s := range ml.JetStreamClusterPeers() {
+			if s == rs.Name() {
+				return fmt.Errorf("Still in the peer list")
+			}
 		}
-	}
+		return nil
+	})
 }
 
 func TestJetStreamClusterPeerRemovalAndStreamReassignment(t *testing.T) {
@@ -11788,7 +11791,6 @@ func TestJetStreamClusterMemoryConsumerCompactVsSnapshot(t *testing.T) {
 
 // This will test our ability to move streams and consumers between clusters.
 func TestJetStreamClusterMovingStreamsAndConsumers(t *testing.T) {
-
 	skip(t)
 
 	sc := createJetStreamTaggedSuperCluster(t)
@@ -11894,6 +11896,14 @@ func TestJetStreamClusterMovingStreamsAndConsumers(t *testing.T) {
 				t.Fatalf("Expected cluster of %q but got %q", "C1", si.Cluster.Name)
 			}
 
+			// Make sure we can not move an inflight stream and consumers, should error.
+			_, err = js.UpdateStream(&nats.StreamConfig{
+				Name:      "MOVE",
+				Replicas:  replicas,
+				Placement: &nats.Placement{Tags: []string{"cloud:aws"}},
+			})
+			require_Error(t, err, NewJSStreamMoveInProgressError())
+
 			checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
 				si, err := js.StreamInfo("MOVE")
 				if err != nil {
@@ -11926,7 +11936,7 @@ func TestJetStreamClusterMovingStreamsAndConsumers(t *testing.T) {
 
 			// Should see the cluster designation and leader switch to C2.
 			// We should also shrink back down to original replica count.
-			checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
+			checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
 				si, err := js.StreamInfo("MOVE")
 				if err != nil {
 					return err
@@ -11974,57 +11984,60 @@ func TestJetStreamClusterMovingStreamsAndConsumers(t *testing.T) {
 			// Now check consumers, make sure the state is correct and that they transferred state and reflect the new messages.
 			// We Ack'd 100 and sent another 100, so should be same.
 			checkConsumer := func(sub *nats.Subscription, isPull bool) {
-				if isPull {
-					checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
-						_, err := sub.ConsumerInfo()
+				t.Helper()
+				checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
+					ci, err := sub.ConsumerInfo()
+					if err != nil {
 						return err
-					})
-				} else {
-					// Make sure we re-established being push bound and our sub got the messages too.
-					checkSubsPending(t, sub, int(initialState.Msgs))
-				}
-				ci, err := sub.ConsumerInfo()
-				require_NoError(t, err)
-				var expectedMsgs uint64
-				if isPull {
-					expectedMsgs = expectedPullMsgs
-				} else {
-					expectedMsgs = expectedPushMsgs
-				}
-				if ci.Delivered.Consumer != expectedMsgs || ci.Delivered.Stream != expectedMsgs {
-					t.Fatalf("Delivered for %q is not correct: %+v", ci.Name, ci.Delivered)
-				}
-				if ci.AckFloor.Consumer != uint64(toAck) || ci.AckFloor.Stream != uint64(toAck) {
-					t.Fatalf("AckFloor for %q is not correct: %+v", ci.Name, ci.AckFloor)
-				}
-				if isPull {
-					if ci.NumAckPending != 0 {
-						t.Fatalf("NumAckPending for %q is not correct: %v", ci.Name, ci.NumAckPending)
 					}
-				} else {
-					if ci.NumAckPending != int(initialState.Msgs) {
-						t.Fatalf("NumAckPending for %q is not correct: %v", ci.Name, ci.NumAckPending)
+					var expectedMsgs uint64
+					if isPull {
+						expectedMsgs = expectedPullMsgs
+					} else {
+						expectedMsgs = expectedPushMsgs
 					}
-				}
-				// Make sure the replicas etc are back to what is expected.
-				si, err := js.StreamInfo("MOVE")
-				require_NoError(t, err)
-				numExpected := si.Config.Replicas
-				if ci.Config.Durable == _EMPTY_ {
-					numExpected = 1
-				}
-				numPeers := len(ci.Cluster.Replicas)
-				if ci.Cluster.Leader != _EMPTY_ {
-					numPeers++
-				}
-				if numPeers != numExpected {
-					t.Fatalf("Expected %d peers, got %d", numPeers, numExpected)
-				}
+
+					if ci.Delivered.Consumer != expectedMsgs || ci.Delivered.Stream != expectedMsgs {
+						return fmt.Errorf("Delivered for %q is not correct: %+v", ci.Name, ci.Delivered)
+					}
+					if ci.AckFloor.Consumer != uint64(toAck) || ci.AckFloor.Stream != uint64(toAck) {
+						return fmt.Errorf("AckFloor for %q is not correct: %+v", ci.Name, ci.AckFloor)
+					}
+					if isPull && ci.NumAckPending != 0 {
+						return fmt.Errorf("NumAckPending for %q is not correct: %v", ci.Name, ci.NumAckPending)
+					} else if !isPull && ci.NumAckPending != int(initialState.Msgs) {
+						return fmt.Errorf("NumAckPending for %q is not correct: %v", ci.Name, ci.NumAckPending)
+					}
+					// Make sure the replicas etc are back to what is expected.
+					si, err := js.StreamInfo("MOVE")
+					if err != nil {
+						return err
+					}
+					numExpected := si.Config.Replicas
+					if ci.Config.Durable == _EMPTY_ {
+						numExpected = 1
+					}
+					numPeers := len(ci.Cluster.Replicas)
+					if ci.Cluster.Leader != _EMPTY_ {
+						numPeers++
+					}
+					if numPeers != numExpected {
+						return fmt.Errorf("Expected %d peers, got %d", numExpected, numPeers)
+					}
+					// If we are push check sub pending.
+					if !isPull {
+						checkSubsPending(t, sub, int(expectedPushMsgs)-toAck)
+					}
+					return nil
+				})
 			}
+
 			checkPushConsumer := func(sub *nats.Subscription) {
+				t.Helper()
 				checkConsumer(sub, false)
 			}
 			checkPullConsumer := func(sub *nats.Subscription) {
+				t.Helper()
 				checkConsumer(sub, true)
 			}
 
@@ -12039,6 +12052,171 @@ func TestJetStreamClusterMovingStreamsAndConsumers(t *testing.T) {
 	}
 }
 
+func TestJetStreamClusterMovingStreamsWithMirror(t *testing.T) {
+	sc := createJetStreamTaggedSuperCluster(t)
+	defer sc.shutdown()
+
+	nc, js := jsClientConnect(t, sc.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "SOURCE",
+		Subjects:  []string{"foo", "bar"},
+		Replicas:  3,
+		Placement: &nats.Placement{Tags: []string{"cloud:aws"}},
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:      "MIRROR",
+		Replicas:  1,
+		Mirror:    &nats.StreamSource{Name: "SOURCE"},
+		Placement: &nats.Placement{Tags: []string{"cloud:gcp"}},
+	})
+	require_NoError(t, err)
+
+	done := make(chan struct{})
+	exited := make(chan struct{})
+	errors := make(chan error, 1)
+
+	numNoResp := uint64(0)
+
+	// We will run a separate routine and send at 100hz
+	go func() {
+		nc, js := jsClientConnect(t, sc.randomServer())
+		defer nc.Close()
+
+		defer close(exited)
+
+		for {
+			select {
+			case <-done:
+				return
+			case <-time.After(10 * time.Millisecond):
+				_, err := js.Publish("foo", []byte("100HZ"))
+				if err == nil {
+				} else if err == nats.ErrNoStreamResponse {
+					atomic.AddUint64(&numNoResp, 1)
+					continue
+				}
+				if err != nil {
+					errors <- err
+					return
+				}
+			}
+		}
+	}()
+
+	// Let it get going.
+	time.Sleep(500 * time.Millisecond)
+
+	// Now move the source to a new cluster.
+	_, err = js.UpdateStream(&nats.StreamConfig{
+		Name:      "SOURCE",
+		Subjects:  []string{"foo", "bar"},
+		Replicas:  3,
+		Placement: &nats.Placement{Tags: []string{"cloud:gcp"}},
+	})
+	require_NoError(t, err)
+
+	checkFor(t, 20*time.Second, 100*time.Millisecond, func() error {
+		si, err := js.StreamInfo("SOURCE")
+		if err != nil {
+			return err
+		}
+		if si.Cluster.Name != "C2" {
+			return fmt.Errorf("Wrong cluster: %q", si.Cluster.Name)
+		}
+		if si.Cluster.Leader == _EMPTY_ {
+			return fmt.Errorf("No leader yet")
+		} else if !strings.HasPrefix(si.Cluster.Leader, "C2-") {
+			return fmt.Errorf("Wrong leader: %q", si.Cluster.Leader)
+		}
+		// Now we want to see that we shrink back to original.
+		if len(si.Cluster.Replicas) != 2 {
+			return fmt.Errorf("Expected %d replicas, got %d", 2, len(si.Cluster.Replicas))
+		}
+		// Let's get to 50+ msgs.
+		if si.State.Msgs < 50 {
+			return fmt.Errorf("Only see %d msgs", si.State.Msgs)
+		}
+		return nil
+	})
+
+	close(done)
+	<-exited
+
+	if nnr := atomic.LoadUint64(&numNoResp); nnr > 0 {
+		t.Fatalf("Expected no failed message publishes, got %d", nnr)
+	}
+}
+
+func TestJetStreamClusterMemoryConsumerInterestRetention(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",
+		Storage:   nats.MemoryStorage,
+		Retention: nats.InterestPolicy,
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	sub, err := js.SubscribeSync("test", nats.Durable("dlc"))
+	require_NoError(t, err)
+
+	for i := 0; i < 1000; i++ {
+		_, err := js.PublishAsync("test", nil)
+		require_NoError(t, err)
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	toAck := 100
+	for i := 0; i < toAck; i++ {
+		m, err := sub.NextMsg(time.Second)
+		require_NoError(t, err)
+		m.AckSync()
+	}
+
+	si, err := js.StreamInfo("test")
+	require_NoError(t, err)
+
+	ci, err := sub.ConsumerInfo()
+	require_NoError(t, err)
+
+	// Make sure acks are not only replicated but processed in a way to remove messages from the replica streams.
+	_, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "test"), nil, time.Second)
+	require_NoError(t, err)
+	c.waitOnStreamLeader("$G", "test")
+
+	_, err = nc.Request(fmt.Sprintf(JSApiConsumerLeaderStepDownT, "test", "dlc"), nil, time.Second)
+	require_NoError(t, err)
+	c.waitOnConsumerLeader("$G", "test", "dlc")
+
+	nsi, err := js.StreamInfo("test")
+	require_NoError(t, err)
+	if nsi.State != si.State {
+		t.Fatalf("Stream states do not match: %+v vs %+v", si.State, nsi.State)
+	}
+
+	nci, err := sub.ConsumerInfo()
+	require_NoError(t, err)
+
+	// Last may be skewed a very small amount.
+	ci.AckFloor.Last, nci.AckFloor.Last = nil, nil
+	if nci.AckFloor != ci.AckFloor {
+		t.Fatalf("Consumer AckFloors are not the same: %+v vs %+v", ci.AckFloor, nci.AckFloor)
+	}
+}
+
 func TestJetStreamImportConsumerStreamSubjectRemap(t *testing.T) {
 	template := `
 	listen: 127.0.0.1:-1
@@ -12057,7 +12235,7 @@ func TestJetStreamImportConsumerStreamSubjectRemap(t *testing.T) {
 			users: [ {user: js, password: pwd} ]
 			exports [
 				# This is streaming to a delivery subject for a push based consumer.
-				{ stream: "deliver.ORDERS" }
+				{ stream: "deliver.ORDERS.*" }
 				# This is to ack received messages. This is a service to support sync ack.
 				{ service: "$JS.ACK.ORDERS.*.>" }
 				# To support ordered consumers, flow control.
@@ -12067,13 +12245,17 @@ func TestJetStreamImportConsumerStreamSubjectRemap(t *testing.T) {
 		IM: {
 			users: [ {user: im, password: pwd} ]
 			imports [
-				{ stream:  { account: JS, subject: "deliver.ORDERS" }}
+				{ stream:  { account: JS, subject: "deliver.ORDERS.*" }}
 				{ service: {account: JS, subject: "$JS.FC.>" }}
 			]
 		},
+		$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] },
+	}
+	leaf {
+		listen: 127.0.0.1:-1
 	}`
 
-	c := createJetStreamClusterWithTemplate(t, template, "SIMM", 3)
+	c := createJetStreamSuperClusterWithTemplate(t, template, 3, 2)
 	defer c.shutdown()
 
 	s := c.randomServer()
@@ -12081,37 +12263,185 @@ func TestJetStreamImportConsumerStreamSubjectRemap(t *testing.T) {
 	defer nc.Close()
 
 	_, err := js.AddStream(&nats.StreamConfig{
-		Name:     "ORDERS",
-		Subjects: []string{"foo"}, // The JS subject.
-		Replicas: 3,
+		Name:      "ORDERS",
+		Subjects:  []string{"foo"}, // The JS subject.
+		Replicas:  3,
+		Placement: &nats.Placement{Cluster: "C1"},
 	})
 	require_NoError(t, err)
 
 	_, err = js.Publish("foo", []byte("OK"))
 	require_NoError(t, err)
 
-	_, err = js.AddConsumer("ORDERS", &nats.ConsumerConfig{
-		Durable:        "oleg",
-		DeliverSubject: "deliver.ORDERS",
-		AckPolicy:      nats.AckExplicitPolicy,
+	for dur, deliver := range map[string]string{
+		"dur-route":   "deliver.ORDERS.route",
+		"dur-gateway": "deliver.ORDERS.gateway",
+		"dur-leaf-1":  "deliver.ORDERS.leaf1",
+		"dur-leaf-2":  "deliver.ORDERS.leaf2",
+	} {
+		_, err = js.AddConsumer("ORDERS", &nats.ConsumerConfig{
+			Durable:        dur,
+			DeliverSubject: deliver,
+			AckPolicy:      nats.AckExplicitPolicy,
+		})
+		require_NoError(t, err)
+	}
+
+	test := func(t *testing.T, s *Server, dSubj string) {
+		nc2, err := nats.Connect(s.ClientURL(), nats.UserInfo("im", "pwd"))
+		require_NoError(t, err)
+		defer nc2.Close()
+
+		sub, err := nc2.SubscribeSync(dSubj)
+		require_NoError(t, err)
+
+		m, err := sub.NextMsg(time.Second)
+		require_NoError(t, err)
+
+		if m.Subject != "foo" {
+			t.Fatalf("Subject not mapped correctly across account boundary, expected %q got %q", "foo", m.Subject)
+		}
+		require_False(t, strings.Contains(m.Reply, "@"))
+	}
+
+	t.Run("route", func(t *testing.T) {
+		// pick random non consumer leader so we receive via route
+		s := c.clusterForName("C1").randomNonConsumerLeader("JS", "ORDERS", "dur-route")
+		test(t, s, "deliver.ORDERS.route")
 	})
-	require_NoError(t, err)
+	t.Run("gateway", func(t *testing.T) {
+		// pick server with inbound gateway from consumer leader, so we receive from gateway and have no route in between
+		scl := c.clusterForName("C1").consumerLeader("JS", "ORDERS", "dur-gateway")
+		var sfound *Server
+		for _, s := range c.clusterForName("C2").servers {
+			s.mu.Lock()
+			for _, c := range s.gateway.in {
+				if c.GetName() == scl.info.ID {
+					sfound = s
+					break
+				}
+			}
+			s.mu.Unlock()
+			if sfound != nil {
+				break
+			}
+		}
+		test(t, sfound, "deliver.ORDERS.gateway")
+	})
+	t.Run("leaf-post-export", func(t *testing.T) {
+		// create leaf node server connected post export/import
+		scl := c.clusterForName("C1").consumerLeader("JS", "ORDERS", "dur-leaf-1")
+		cf := createConfFile(t, []byte(fmt.Sprintf(`
+			port: -1
+			leafnodes {
+				remotes [ { url: "nats://im:pwd@127.0.0.1:%d" } ]
+			}
+			authorization: {
+				user: im,
+				password: pwd
+			}
+		`, scl.getOpts().LeafNode.Port)))
+		defer removeFile(t, cf)
+		s, _ := RunServerWithConfig(cf)
+		defer s.Shutdown()
+		checkLeafNodeConnected(t, scl)
+		test(t, s, "deliver.ORDERS.leaf1")
+	})
+	t.Run("leaf-pre-export", func(t *testing.T) {
+		// create leaf node server connected pre export, perform export/import on leaf node server
+		scl := c.clusterForName("C1").consumerLeader("JS", "ORDERS", "dur-leaf-2")
+		cf := createConfFile(t, []byte(fmt.Sprintf(`
+			port: -1
+			leafnodes {
+				remotes [ { url: "nats://im:pwd@127.0.0.1:%d", account: JS2 } ]
+			}
+			accounts: {
+				JS2: {
+					users: [ {user: js, password: pwd} ]
+					exports [
+						# This is streaming to a delivery subject for a push based consumer.
+						{ stream: "deliver.ORDERS.*" }
+						# This is to ack received messages. This is a service to support sync ack.
+						{ service: "$JS.ACK.ORDERS.*.>" }
+						# To support ordered consumers, flow control.
+						{ service: "$JS.FC.>" }
+					]
+				},
+				IM2: {
+					users: [ {user: im, password: pwd} ]
+					imports [
+						{ stream:  { account: JS2, subject: "deliver.ORDERS.*" }}
+						{ service: {account: JS2, subject: "$JS.FC.>" }}
+					]
+				},
+			}
+		`, scl.getOpts().LeafNode.Port)))
+		defer removeFile(t, cf)
+		s, _ := RunServerWithConfig(cf)
+		defer s.Shutdown()
+		checkLeafNodeConnected(t, scl)
+		test(t, s, "deliver.ORDERS.leaf2")
+	})
+}
 
-	// This bug only happens when we go across a route, so pick another server other than the consumer leader.
-	s = c.randomNonConsumerLeader("JS", "ORDERS", "oleg")
+func TestJetStreamClusterStreamAlternates(t *testing.T) {
+	sc := createJetStreamTaggedSuperCluster(t)
+	defer sc.shutdown()
+
+	nc, js := jsClientConnect(t, sc.randomServer())
+	defer nc.Close()
 
-	nc2, err := nats.Connect(s.ClientURL(), nats.UserInfo("im", "pwd"))
+	// C1
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "SOURCE",
+		Subjects:  []string{"foo", "bar", "baz"},
+		Replicas:  3,
+		Placement: &nats.Placement{Tags: []string{"cloud:aws", "country:us"}},
+	})
 	require_NoError(t, err)
 
-	sub, err := nc2.SubscribeSync("deliver.ORDERS")
+	// C2
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:      "MIRROR-1",
+		Replicas:  1,
+		Mirror:    &nats.StreamSource{Name: "SOURCE"},
+		Placement: &nats.Placement{Tags: []string{"cloud:gcp", "country:uk"}},
+	})
 	require_NoError(t, err)
 
-	m, err := sub.NextMsg(time.Second)
+	// C3
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:      "MIRROR-2",
+		Replicas:  2,
+		Mirror:    &nats.StreamSource{Name: "SOURCE"},
+		Placement: &nats.Placement{Tags: []string{"cloud:az", "country:jp"}},
+	})
 	require_NoError(t, err)
 
-	if m.Subject != "foo" {
-		t.Fatalf("Subject not mapped correctly across account boundary, expected %q got %q", "foo", m.Subject)
-	}
+	// No client support yet, so do by hand.
+	getStreamInfo := func(nc *nats.Conn, expected string) {
+		t.Helper()
+		resp, err := nc.Request(fmt.Sprintf(JSApiStreamInfoT, "SOURCE"), nil, time.Second)
+		require_NoError(t, err)
+		var si StreamInfo
+		err = json.Unmarshal(resp.Data, &si)
+		require_NoError(t, err)
+		require_True(t, len(si.Alternates) == 3)
+		require_True(t, si.Alternates[0].Cluster == expected)
+		seen := make(map[string]struct{})
+		for _, alt := range si.Alternates {
+			seen[alt.Cluster] = struct{}{}
+		}
+		require_True(t, len(seen) == 3)
+	}
+
+	// Connect to different clusters to check ordering.
+	nc, _ = jsClientConnect(t, sc.clusterForName("C1").randomServer())
+	getStreamInfo(nc, "C1")
+	nc, _ = jsClientConnect(t, sc.clusterForName("C2").randomServer())
+	getStreamInfo(nc, "C2")
+	nc, _ = jsClientConnect(t, sc.clusterForName("C3").randomServer())
+	getStreamInfo(nc, "C3")
 }
 
 // Support functions
diff --git a/server/norace_test.go b/server/norace_test.go
index f77017af..1f06a6c1 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -1567,13 +1567,8 @@ func TestNoRaceJetStreamClusterSuperClusterMirrors(t *testing.T) {
 		Placement: &nats.Placement{Cluster: "C1"},
 	})
 
-	// Faster timeout since we loop below checking for condition.
-	js2, err := nc.JetStream(nats.MaxWait(50 * time.Millisecond))
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
-		si, err := js2.StreamInfo("M1")
+		si, err := js.StreamInfo("M1")
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -1602,7 +1597,7 @@ func TestNoRaceJetStreamClusterSuperClusterMirrors(t *testing.T) {
 	})
 
 	checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
-		si, err := js2.StreamInfo("M2")
+		si, err := js.StreamInfo("M2")
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -1636,8 +1631,8 @@ func TestNoRaceJetStreamClusterSuperClusterMirrors(t *testing.T) {
 	<-doneCh
 	sc.clusterForName("C3").waitOnStreamLeader("$G", "M2")
 
-	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
-		si, err := js2.StreamInfo("M2")
+	checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
+		si, err := js.StreamInfo("M2")
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -3333,7 +3328,7 @@ func TestNoRaceJetStreamClusterCorruptWAL(t *testing.T) {
 	// Check consumer consistency.
 	checkConsumerWith := func(delivered, ackFloor uint64, ackPending int) {
 		t.Helper()
-		nc, js = jsClientConnect(t, c.randomServer())
+		nc, js := jsClientConnect(t, c.randomServer())
 		defer nc.Close()
 
 		checkFor(t, 5*time.Second, 100*time.Millisecond, func() error {
@@ -3435,28 +3430,35 @@ func TestNoRaceJetStreamClusterCorruptWAL(t *testing.T) {
 
 	cl = c.consumerLeader("$G", "TEST", "dlc")
 	mset, err = cl.GlobalAccount().lookupStream("TEST")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 	o = mset.lookupConsumer("dlc")
-	if o == nil {
-		t.Fatalf("Error looking up consumer %q", "dlc")
-	}
+	require_NoError(t, err)
+
 	// Grab underlying raft node and the WAL (filestore) and truncate it.
 	// This will simulate the WAL losing state due to truncate and we want to make sure it recovers.
+
 	fs = o.raftNode().(*raft).wal.(*fileStore)
 	state = fs.State()
-	fs.Truncate(state.FirstSeq)
+	err = fs.Truncate(state.FirstSeq)
+	require_NoError(t, err)
+	state = fs.State()
 
 	sub, err = js.PullSubscribe("foo", "dlc")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(t, err)
 
 	// This will cause us to stepdown and truncate our WAL.
-	sub.Fetch(100)
+	msgs, _ := sub.Fetch(100)
 
-	checkFor(t, 20*time.Second, 500*time.Millisecond, func() error {
+	// In case we have to retry, we need to adjust the tests that follow.
+	newDelivered, ackPending, more := 200, 75, 0
+
+	if len(msgs) != 100 {
+		more = 100 - len(msgs)
+		_, err = sub.Fetch(more)
+		require_NoError(t, err)
+	}
+
+	checkFor(t, 30*time.Second, 500*time.Millisecond, func() error {
 		// Make sure we changed leaders.
 		if clnew := c.consumerLeader("$G", "TEST", "dlc"); clnew == nil || cl == clnew {
 			return fmt.Errorf("Expected leader to have moved")
@@ -3465,9 +3467,9 @@ func TestNoRaceJetStreamClusterCorruptWAL(t *testing.T) {
 	})
 
 	// First one triggered stepdown so no updates would be distributed, now make sure we are ok.
-	checkConsumer()
-	fetchMsgs(t, sub, 100, 5*time.Second)
-	checkConsumerWith(300, 50, 175)
+	checkConsumerWith(uint64(newDelivered+more), 50, ackPending+more)
+	fetchMsgs(t, sub, 100, 10*time.Second)
+	checkConsumerWith(uint64(300+more), 50, 175+more)
 }
 
 func TestNoRaceJetStreamClusterInterestRetentionDeadlock(t *testing.T) {
diff --git a/server/routes_test.go b/server/routes_test.go
index 13910f27..e68dcaa4 100644
--- a/server/routes_test.go
+++ b/server/routes_test.go
@@ -1509,3 +1509,20 @@ func testTLSRoutesCertificateImplicitAllow(t *testing.T, pass bool) {
 		})
 	}
 }
+
+func TestSubjectRenameViaJetStreamAck(t *testing.T) {
+	s := RunRandClientPortServer()
+	errChan := make(chan error)
+	defer close(errChan)
+	ncPub := natsConnect(t, s.ClientURL(), nats.UserInfo("client", "pwd"),
+		nats.ErrorHandler(func(conn *nats.Conn, s *nats.Subscription, err error) {
+			errChan <- err
+		}))
+	require_NoError(t, ncPub.PublishRequest("SVC.ALLOWED", "$JS.ACK.whatever@ADMIN", nil))
+	select {
+	case err := <-errChan:
+		require_Contains(t, err.Error(), "Permissions Violation for Publish with Reply of")
+	case <-time.After(time.Second):
+		t.Fatalf("Expected error")
+	}
+}