Repository: nats-io/nats-server
Pull request #3001 (2022-04-04 23:07:19)

Allow streams and consumers to migrate between clusters.
The system will allow an update to a stream, and subsequently all attached consumers, to be placed in another cluster either directly or via tag placement. The meta layer will scale the underlying peerset appropriately to straddle the two clusters for both the stream and consumers, taking into account the consumer type. Control will then pass to the current leaders of the assets who will monitor the catch up status of the new peers (Note we can optimize this later to only traverse once across a GW for any given asset, but for now this is simpler). Once the original leaders have determined the assets are synched it will pass leadership to a member of the new peerset. Once the new leader has been elected, it will forward a request for the meta layer to shrink the peerset by removing the old peers.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core


derekcollison: FYI there may be two follow on PRs to have a time bound on the migration monitoring, and a test to check that we are not adversely affected by account limits here, meaning you can pop your usage for a short time over the account limits and not be blocked to move a stream. 

Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 68131624..98ccba03 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -264,12 +264,14 @@ type consumer struct {
 	closed            bool
 
 	// Clustered.
-	ca      *consumerAssignment
-	node    RaftNode
-	infoSub *subscription
-	lqsent  time.Time
-	prm     map[string]struct{}
-	prOk    bool
+	ca        *consumerAssignment
+	node      RaftNode
+	infoSub   *subscription
+	lqsent    time.Time
+	prm       map[string]struct{}
+	prOk      bool
+	uch       chan struct{}
+	retention RetentionPolicy
 
 	// R>1 proposals
 	pch   chan struct{}
@@ -485,6 +487,7 @@ func checkConsumerCfg(config *ConsumerConfig, srvLim *JSLimitOpts, cfg *StreamCo
 func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname string, ca *consumerAssignment) (*consumer, error) {
 	mset.mu.RLock()
 	s, jsa, tierName, cfg, acc := mset.srv, mset.jsa, mset.tier, mset.cfg, mset.acc
+	retention := cfg.Retention
 	mset.mu.RUnlock()
 
 	// If we do not have the consumer currently assigned to us in cluster mode we will proceed but warn.
@@ -590,22 +593,24 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 
 	// Set name, which will be durable name if set, otherwise we create one at random.
 	o := &consumer{
-		mset:    mset,
-		js:      s.getJetStream(),
-		acc:     a,
-		srv:     s,
-		client:  s.createInternalJetStreamClient(),
-		sysc:    s.createInternalJetStreamClient(),
-		cfg:     *config,
-		dsubj:   config.DeliverSubject,
-		outq:    mset.outq,
-		active:  true,
-		qch:     make(chan struct{}),
-		mch:     make(chan struct{}, 1),
-		sfreq:   int32(sampleFreq),
-		maxdc:   uint64(config.MaxDeliver),
-		maxp:    config.MaxAckPending,
-		created: time.Now().UTC(),
+		mset:      mset,
+		js:        s.getJetStream(),
+		acc:       a,
+		srv:       s,
+		client:    s.createInternalJetStreamClient(),
+		sysc:      s.createInternalJetStreamClient(),
+		cfg:       *config,
+		dsubj:     config.DeliverSubject,
+		outq:      mset.outq,
+		active:    true,
+		qch:       make(chan struct{}),
+		uch:       make(chan struct{}, 1),
+		mch:       make(chan struct{}, 1),
+		sfreq:     int32(sampleFreq),
+		maxdc:     uint64(config.MaxDeliver),
+		maxp:      config.MaxAckPending,
+		retention: retention,
+		created:   time.Now().UTC(),
 	}
 
 	// Bind internal client to the user account.
@@ -768,13 +773,27 @@ func (o *consumer) consumerAssignment() *consumerAssignment {
 func (o *consumer) setConsumerAssignment(ca *consumerAssignment) {
 	o.mu.Lock()
 	defer o.mu.Unlock()
+
 	o.ca = ca
+	if ca == nil {
+		return
+	}
 	// Set our node.
-	if ca != nil {
-		o.node = ca.Group.node
+	o.node = ca.Group.node
+
+	// Trigger update chan.
+	select {
+	case o.uch <- struct{}{}:
+	default:
 	}
 }
 
+func (o *consumer) updateC() <-chan struct{} {
+	o.mu.RLock()
+	defer o.mu.RUnlock()
+	return o.uch
+}
+
 // checkQueueInterest will check on our interest's queue group status.
 // Lock should be held.
 func (o *consumer) checkQueueInterest() {
diff --git a/server/errors.json b/server/errors.json
index 132b8ac3..9c4175df 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -51,7 +51,7 @@
   },
   {
     "constant": "JSStreamSubjectOverlapErr",
-    "code": 500,
+    "code": 400,
     "error_code": 10065,
     "description": "subjects overlap with an existing stream",
     "comment": "",
@@ -293,7 +293,7 @@
     "constant": "JSMirrorWithSubjectsErr",
     "code": 400,
     "error_code": 10034,
-    "description": "stream mirrors can not also contain subjects",
+    "description": "stream mirrors can not contain subjects",
     "comment": "",
     "help": "",
     "url": "",
@@ -333,7 +333,7 @@
     "constant": "JSStreamMirrorNotUpdatableErr",
     "code": 400,
     "error_code": 10055,
-    "description": "Mirror configuration can not be updated",
+    "description": "stream mirror configuration can not be updated",
     "comment": "",
     "help": "",
     "url": "",
@@ -404,7 +404,7 @@
     "code": 500,
     "error_code": 10029,
     "description": "{err}",
-    "comment": "Generic mirror consumer setup failure string",
+    "comment": "generic mirror consumer setup failure string",
     "help": "",
     "url": "",
     "deprecates": ""
@@ -1208,5 +1208,15 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSStreamMoveAndScaleErr",
+    "code": 400,
+    "error_code": 10123,
+    "description": "can not move and scale a stream in a single update",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
-]
\ No newline at end of file
+]
diff --git a/server/events.go b/server/events.go
index 8955812d..63670b9b 100644
--- a/server/events.go
+++ b/server/events.go
@@ -691,6 +691,9 @@ func (s *Server) sendStatsz(subj string) {
 			ni := v.(nodeInfo)
 			ni.stats = jStat.Stats
 			ni.cfg = jStat.Config
+			s.optsMu.RLock()
+			ni.tags = copyStrings(s.opts.Tags)
+			s.optsMu.RUnlock()
 			s.nodeToInfo.Store(ourNode, ni)
 		}
 		// Metagroup info.
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 26897c93..333d9397 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -1468,8 +1468,15 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
+	// Handle clustered version here.
 	if s.JetStreamIsClustered() {
-		s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg)
+		// If we are inline with client, we still may need to do a callout for stream info
+		// during this call, so place in Go routine to not block client.
+		if c.kind != ROUTER && c.kind != GATEWAY {
+			go s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg)
+		} else {
+			s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg)
+		}
 		return
 	}
 
@@ -1877,7 +1884,7 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
 	}
 	// Check for out of band catchups.
 	if mset.hasCatchupPeers() {
-		mset.checkClusterInfo(resp.StreamInfo)
+		mset.checkClusterInfo(resp.StreamInfo.Cluster)
 	}
 
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
@@ -3655,14 +3662,15 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
 				s.sendDelayedAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp), ca.Group)
 				return
 			}
-			if ca == nil {
-				return
-			}
 			// We have a consumer assignment.
 			js.mu.RLock()
 			var node RaftNode
+			var leaderNotPartOfGroup bool
 			if rg := ca.Group; rg != nil && rg.node != nil && rg.isMember(ourID) {
 				node = rg.node
+				if gl := node.GroupLeader(); gl != _EMPTY_ && !rg.isMember(gl) {
+					leaderNotPartOfGroup = true
+				}
 			}
 			js.mu.RUnlock()
 			// Check if we should ignore all together.
@@ -3680,7 +3688,12 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
 				}
 				return
 			}
+			// If we are a member and we have a group leader or we had a previous leader consider bailing out.
 			if node != nil && (node.GroupLeader() != _EMPTY_ || node.HadPreviousLeader()) {
+				if leaderNotPartOfGroup {
+					resp.Error = NewJSConsumerOfflineError()
+					s.sendDelayedAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp), nil)
+				}
 				return
 			}
 			// If we are here we are a member and this is just a new consumer that does not have a leader yet.
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 494a8053..3ac85eb3 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -1228,6 +1228,43 @@ 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) {
+	sa := mset.streamAssignment()
+	ci := js.clusterInfo(mset.raftGroup())
+
+	js.mu.Lock()
+	defer js.mu.Unlock()
+
+	cc, csa := js.cluster, sa.copyGroup()
+	numExpandedPeers := len(csa.Group.Peers)
+	csa.Group.Peers = csa.Group.Peers[:0]
+
+	if ci.Leader != _EMPTY_ {
+		csa.Group.Peers = append(csa.Group.Peers, string(getHash(ci.Leader)))
+	}
+	for _, r := range ci.Replicas {
+		if r.cluster == ci.Name {
+			csa.Group.Peers = append(csa.Group.Peers, r.peer)
+		}
+	}
+
+	// Now do consumers actually first here, followed by the owning stream.
+	for _, ca := range csa.consumers {
+		cca := ca.copyGroup()
+		numPeers := len(cca.Group.Peers)
+		if numPeers == numExpandedPeers {
+			cca.Group.Peers = csa.Group.Peers
+		} else {
+			cca.Group.Peers = cca.Group.Peers[len(cca.Group.Peers)-1:]
+		}
+		cc.meta.ForwardProposal(encodeAddConsumerAssignment(cca))
+	}
+
+	cc.meta.ForwardProposal(encodeUpdateStreamAssignment(csa))
+}
+
 // Assumes all checks have already been done.
 func (js *jetStream) removePeerFromStream(sa *streamAssignment, peer string) bool {
 	js.mu.Lock()
@@ -1497,7 +1534,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		return
 	}
 
-	qch, lch, aq := n.QuitC(), n.LeadChangeC(), n.ApplyQ()
+	qch, lch, aq, uch := n.QuitC(), n.LeadChangeC(), n.ApplyQ(), mset.updateC()
 
 	s.Debugf("Starting stream monitor for '%s > %s' [%s]", sa.Client.serviceAccount(), sa.Config.Name, n.Group())
 	defer s.Debugf("Exiting stream monitor for '%s > %s' [%s]", sa.Client.serviceAccount(), sa.Config.Name, n.Group())
@@ -1555,6 +1592,25 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	restoreDoneCh := make(<-chan error)
 	isRecovering := 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()
+
 	// 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
@@ -1621,9 +1677,61 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 				js.setStreamAssignmentRecovering(sa)
 			}
 
+			// Process our leader change.
 			js.processStreamLeaderChange(mset, isLeader)
+
+			// Check for migrations here. We set the state on the stream assignment update below.
+			if isLeader && migrating {
+				if peerGroup == oldPeerGroup {
+					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) })
+				}
+			}
+
 		case <-t.C:
 			doSnapshot()
+		case <-uch:
+			// We get this when we have a new stream assignment caused by an update. We want
+			// to know if we are migrating.
+			migrating, peerGroup = mset.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 below
+			// 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(mset.raftGroup())
+			if mset.hasCatchupPeers() {
+				mset.checkClusterInfo(ci)
+			}
+			// Track the new peers and check to make sure at least 1 is current and up to date.
+			var newPeers []*PeerInfo
+			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.Current {
+						stopMigrationMonitoring()
+						// Transfer leadership for stream.
+						mset.raftNode().StepDown(r.peer)
+						break
+					}
+				}
+			}
 		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
@@ -1638,6 +1746,8 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			if err == nil {
 				if mset, err = acc.lookupStream(sa.Config.Name); mset != nil {
 					mset.setStreamAssignment(sa)
+					// Make sure to update our updateC which would have been nil.
+					uch = mset.updateC()
 				}
 			}
 			if err != nil {
@@ -1719,6 +1829,75 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	}
 }
 
+// When we are migration denotes if we ourselves are part of the old peer set or the new one.
+// Both types will be running at the same time as we scale up to extend into the new cluster.
+// Once detected we will us our type to dictate our behavior.
+type peerMigrateType int8
+
+const (
+	oldPeerGroup = peerMigrateType(iota)
+	newPeerGroup
+)
+
+// Determine if we are migrating and if so if we are part of the old or new set.
+func (mset *stream) isMigrating() (bool, peerMigrateType) {
+	mset.mu.RLock()
+	s, js, sa := mset.srv, mset.js, mset.sa
+	mset.mu.RUnlock()
+
+	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 || sa.Group == nil || sa.Group.node == nil {
+		return false, oldPeerGroup
+	}
+	// The sign of migration is if our group peer count != configured replica count.
+	if sa.Config.Replicas == len(sa.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 sa.Group.Cluster == s.cachedClusterName() {
+		return true, newPeerGroup
+	}
+	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()
@@ -2305,7 +2484,6 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
 	if isMember {
 		sa.responded = false
 	}
-
 	js.mu.Unlock()
 
 	// Check if this is for us..
@@ -2313,8 +2491,12 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
 		js.processClusterUpdateStream(acc, osa, sa)
 	} else if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
 		// We have one here even though we are not a member. This can happen on re-assignment.
-		s.Debugf("JetStream removing stream '%s > %s' from this server, re-assigned", sa.Client.serviceAccount(), sa.Config.Name)
+		s.Debugf("JetStream removing stream '%s > %s' from this server", sa.Client.serviceAccount(), sa.Config.Name)
+
 		if node := mset.raftNode(); node != nil {
+			if node.Leader() {
+				node.StepDown()
+			}
 			node.ProposeRemovePeer(ourID)
 		}
 		mset.stop(true, false)
@@ -2331,14 +2513,14 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
 	js.mu.Lock()
 	s, rg := js.srv, sa.Group
 	client, subject, reply := sa.Client, sa.Subject, sa.Reply
-	alreadyRunning, numReplicas := osa.Group.node != nil, sa.Config.Replicas
+	alreadyRunning, numReplicas := osa.Group.node != nil, len(rg.Peers)
 	needsNode := rg.node == nil
-	storage := sa.Config.Storage
+	storage, cfg := sa.Config.Storage, sa.Config
 	hasResponded := sa.responded
 	sa.responded = true
 	js.mu.Unlock()
 
-	mset, err := acc.lookupStream(sa.Config.Name)
+	mset, err := acc.lookupStream(cfg.Name)
 	if err == nil && mset != nil {
 		var needsSetLeader bool
 		if !alreadyRunning && numReplicas > 1 {
@@ -2351,17 +2533,19 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
 			mset.removeNode()
 			// Make sure we are leader now that we are R1.
 			needsSetLeader = true
-			// In case we nned to shutdown the cluster specific subs, etc.
+			// In case we need to shutdown the cluster specific subs, etc.
 			mset.setLeader(false)
 			js.mu.Lock()
 			rg.node = nil
 			js.mu.Unlock()
 		}
+
 		mset.setStreamAssignment(sa)
-		if err = mset.update(sa.Config); err != nil {
-			s.Warnf("JetStream cluster error updating stream %q for account %q: %v", sa.Config.Name, acc.Name, err)
+		if err = mset.update(cfg); err != nil {
+			s.Warnf("JetStream cluster error updating stream %q for account %q: %v", cfg.Name, acc.Name, err)
 			mset.setStreamAssignment(osa)
 		}
+
 		// Make sure we are the leader now that we are R1.
 		if needsSetLeader {
 			mset.setLeader(true)
@@ -2703,7 +2887,8 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 		return
 	}
 
-	if _, err := s.LookupAccount(accName); err != nil {
+	acc, err := s.LookupAccount(accName)
+	if err != nil {
 		ll := fmt.Sprintf("Account [%s] lookup for consumer create failed: %v", accName, err)
 		if isMember {
 			// If we can not lookup the account and we are a member, send this result back to the metacontroller leader.
@@ -2760,14 +2945,26 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 		// 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() {
-				node.StepDown()
+				var nl string
+				if peers := ca.Group.Peers; len(peers) > 0 {
+					nl = peers[rand.Int31n(int32(len(peers)))]
+				}
+				node.StepDown(nl)
 			}
 			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()
+			}
+		}
 	}
 }
 
@@ -3107,7 +3304,7 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 		return
 	}
 
-	qch, lch, aq := n.QuitC(), n.LeadChangeC(), n.ApplyQ()
+	qch, lch, aq, uch := n.QuitC(), n.LeadChangeC(), n.ApplyQ(), o.updateC()
 
 	s.Debugf("Starting consumer monitor for '%s > %s > %s' [%s]", o.acc.Name, ca.Stream, ca.Name, n.Group())
 	defer s.Debugf("Exiting consumer monitor for '%s > %s > %s' [%s]", o.acc.Name, ca.Stream, ca.Name, n.Group())
@@ -3159,6 +3356,25 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 	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:
@@ -3195,6 +3411,41 @@ 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)
 		}
@@ -3302,7 +3553,7 @@ func (o *consumer) processReplicatedAck(dseq, sseq uint64) {
 	o.store.UpdateAcks(dseq, sseq)
 
 	mset := o.mset
-	if mset == nil || mset.cfg.Retention == LimitsPolicy {
+	if mset == nil || o.retention == LimitsPolicy {
 		o.mu.Unlock()
 		return
 	}
@@ -3899,12 +4150,7 @@ func groupNameForConsumer(peers []string, storage StorageType) string {
 }
 
 func groupName(prefix string, peers []string, storage StorageType) string {
-	var gns string
-	if len(peers) == 1 {
-		gns = peers[0]
-	} else {
-		gns = string(getHash(nuid.Next()))
-	}
+	gns := string(getHash(nuid.Next()))
 	return fmt.Sprintf("%s-R%d%s-%s", prefix, len(peers), storage.String()[:1], gns)
 }
 
@@ -4127,7 +4373,7 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
 		return
 	}
-	// Check for mirrot changes which are not allowed.
+	// Check for mirror changes which are not allowed.
 	if !reflect.DeepEqual(newCfg.Mirror, osa.Config.Mirror) {
 		resp.Error = NewJSStreamMirrorNotUpdatableError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
@@ -4150,11 +4396,27 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 		}
 	}
 
+	// Make copy so to not change original.
+	rg := osa.copyGroup().Group
+
+	// Check for a move update.
+	// TODO(dlc) - Should add a resolve from Tags to cluster and check that vs reflect.
+	isMoveRequest := !reflect.DeepEqual(osa.Config.Placement, newCfg.Placement)
+
 	// Check for replica changes.
-	rg := osa.Group
+	isReplicaChange := newCfg.Replicas != len(rg.Peers)
+
+	// We stage consumer updates and do them after the stream update.
 	var consumers []*consumerAssignment
 
-	if newCfg.Replicas != len(rg.Peers) {
+	// Can not move and scale at same time.
+	if isMoveRequest && isReplicaChange {
+		resp.Error = NewJSStreamMoveAndScaleError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+		return
+	}
+
+	if isReplicaChange {
 		// We are adding new peers here.
 		if newCfg.Replicas > len(rg.Peers) {
 			peers := cc.selectPeerGroup(newCfg.Replicas, rg.Cluster, newCfg, rg.Peers)
@@ -4172,8 +4434,79 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 			}
 			rg.Peers = peers
 		} else {
-			// We are deleting nodes here.
-			rg.Peers = rg.Peers[:newCfg.Replicas]
+			// We are deleting nodes here. We want to do our best to preserve the current leader.
+			// We have support now from above that guarantees we are in our own Go routine, so can
+			// ask for stream info from the stream leader to make sure we keep the leader in the new list.
+			var curLeader string
+			if !s.allPeersOffline(rg) {
+				// Need to release js lock.
+				js.mu.Unlock()
+				s.mu.Lock()
+				inbox := s.newRespInbox()
+				results := make(chan *StreamInfo, 1)
+				// Store our handler.
+				s.sys.replies[inbox] = func(sub *subscription, _ *client, _ *Account, subject, _ string, msg []byte) {
+					var si StreamInfo
+					if err := json.Unmarshal(msg, &si); err != nil {
+						s.Warnf("Error unmarshaling clustered stream info response:%v", err)
+						return
+					}
+					select {
+					case results <- &si:
+					default:
+						s.Warnf("Failed placing remote stream info result on internal channel")
+					}
+				}
+				s.mu.Unlock()
+
+				isubj := fmt.Sprintf(clusterStreamInfoT, ci.serviceAccount(), cfg.Name)
+				s.sendInternalMsgLocked(isubj, inbox, nil, nil)
+
+				const timeout = 2 * time.Second
+				notActive := time.NewTimer(timeout)
+				defer notActive.Stop()
+
+				select {
+				case <-s.quitCh:
+					break
+				case <-notActive.C:
+					s.Warnf("Did not receive stream info results for '%s > %s'", acc, cfg.Name)
+				case si := <-results:
+					if si.Cluster != nil {
+						// The leader here is the server name, but need to convert to internal name.
+						curLeader = string(getHash(si.Cluster.Leader))
+					}
+				}
+				// Clean up here.
+				s.mu.Lock()
+				if s.sys != nil && s.sys.replies != nil {
+					delete(s.sys.replies, inbox)
+				}
+				s.mu.Unlock()
+				// Re-acquire here.
+				js.mu.Lock()
+			}
+			// If we identified a leader make sure its part of the new group.
+			selected := make([]string, 0, newCfg.Replicas)
+
+			if curLeader != _EMPTY_ {
+				selected = append(selected, curLeader)
+			}
+			for _, peer := range rg.Peers {
+				if len(selected) == newCfg.Replicas {
+					break
+				}
+				if peer == curLeader {
+					continue
+				}
+				if si, ok := s.nodeToInfo.Load(peer); ok && si != nil {
+					if si.(nodeInfo).offline {
+						continue
+					}
+					selected = append(selected, peer)
+				}
+			}
+			rg.Peers = selected
 		}
 
 		// Need to remap any consumers.
@@ -4194,15 +4527,53 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 				consumers = append(consumers, cca)
 			}
 		}
+
+	} else if isMoveRequest {
+		nrg := js.createGroupForStream(ci, newCfg)
+		if nrg == nil {
+			resp.Error = NewJSInsufficientResourcesError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+			return
+		}
+		// Only change if resolved clusters are different.
+		if rg.Cluster != nrg.Cluster {
+			// If we are R1, make sure original is leader during scale up for move.
+			if len(rg.Peers) == 1 {
+				rg.Preferred = rg.Peers[0]
+			}
+			// Add in new peers since we will extend the peer group to straddle both clusters.
+			rg.Peers = append(rg.Peers, nrg.Peers...)
+			rg.Cluster = nrg.Cluster
+
+			for _, ca := range osa.consumers {
+				cca := ca.copyGroup()
+				// Ephemerals are R=1, so only auto-remap if consumer peer count == nrg peer count.
+				numPeers := len(ca.Group.Peers)
+				if numPeers == len(nrg.Peers) {
+					cca.Group.Peers = append(cca.Group.Peers, nrg.Peers...)
+				} else {
+					// This is an ephemeral, so R1. Just randomly pick a single peer from the new set.
+					pi := rand.Int31n(int32(len(nrg.Peers)))
+					cca.Group.Peers = append(cca.Group.Peers, nrg.Peers[pi])
+				}
+				// Make sure to set if not already set.
+				if cca.Group.Preferred == _EMPTY_ {
+					cca.Group.Preferred = cca.Group.Peers[0]
+				}
+				// We can not propose here before the stream itself so we collect them.
+				consumers = append(consumers, cca)
+			}
+		}
 	}
 
-	sa := &streamAssignment{Group: rg, Sync: osa.Sync, Config: newCfg, Subject: subject, Reply: reply, Client: ci}
+	sa := &streamAssignment{Group: rg, Sync: osa.Sync, Created: osa.Created, Config: newCfg, Subject: subject, Reply: reply, Client: ci}
 	cc.meta.Propose(encodeUpdateStreamAssignment(sa))
 
 	// Process any staged consumers.
 	for _, ca := range consumers {
 		cc.meta.Propose(encodeAddConsumerAssignment(ca))
 	}
+
 }
 
 func (s *Server) jsClusteredStreamDeleteRequest(ci *ClientInfo, acc *Account, stream, subject, reply string, rmsg []byte) {
@@ -4317,6 +4688,7 @@ func (s *Server) jsClusteredStreamRestoreRequest(
 	cc.meta.Propose(encodeAddStreamAssignment(sa))
 }
 
+// Determine if all peers for this group are offline.
 func (s *Server) allPeersOffline(rg *raftGroup) bool {
 	if rg == nil {
 		return false
@@ -4883,6 +5255,9 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 		// Pick a preferred leader.
 		rg.setPreferred()
 
+		// Inherit cluster from stream.
+		rg.Cluster = sa.Group.Cluster
+
 		// We need to set the ephemeral here before replicating.
 		if !isDurableConsumer(cfg) {
 			// We chose to have ephemerals be R=1 unless stream is interest or workqueue.
@@ -5634,7 +6009,15 @@ func (js *jetStream) clusterInfo(rg *raftGroup) *ClusterInfo {
 			}
 			if sir, ok := s.nodeToInfo.Load(rp.ID); ok && sir != nil {
 				si := sir.(nodeInfo)
-				pi := &PeerInfo{Name: si.name, Current: current, Offline: si.offline, Active: lastSeen, Lag: rp.Lag}
+				pi := &PeerInfo{
+					Name:    si.name,
+					Current: current,
+					Offline: si.offline,
+					Active:  lastSeen,
+					Lag:     rp.Lag,
+					cluster: si.cluster,
+					peer:    rp.ID,
+				}
 				ci.Replicas = append(ci.Replicas, pi)
 			}
 		}
@@ -5642,8 +6025,8 @@ func (js *jetStream) clusterInfo(rg *raftGroup) *ClusterInfo {
 	return ci
 }
 
-func (mset *stream) checkClusterInfo(si *StreamInfo) {
-	for _, r := range si.Cluster.Replicas {
+func (mset *stream) checkClusterInfo(ci *ClusterInfo) {
+	for _, r := range ci.Replicas {
 		peer := string(getHash(r.Name))
 		if lag := mset.lagForCatchupPeer(peer); lag > 0 {
 			r.Current = false
@@ -5681,7 +6064,7 @@ func (mset *stream) handleClusterStreamInfoRequest(sub *subscription, c *client,
 
 	// Check for out of band catchups.
 	if mset.hasCatchupPeers() {
-		mset.checkClusterInfo(si)
+		mset.checkClusterInfo(si.Cluster)
 	}
 
 	sysc.sendInternalMsg(reply, _EMPTY_, nil, si)
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index aa84eb7f..1967beab 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -188,7 +188,7 @@ const (
 	// JSMemoryResourcesExceededErr insufficient memory resources available
 	JSMemoryResourcesExceededErr ErrorIdentifier = 10028
 
-	// JSMirrorConsumerSetupFailedErrF Generic mirror consumer setup failure string ({err})
+	// JSMirrorConsumerSetupFailedErrF generic mirror consumer setup failure string ({err})
 	JSMirrorConsumerSetupFailedErrF ErrorIdentifier = 10029
 
 	// JSMirrorMaxMessageSizeTooBigErr stream mirror must have max message size >= source
@@ -203,7 +203,7 @@ const (
 	// JSMirrorWithSubjectFiltersErr stream mirrors can not contain filtered subjects
 	JSMirrorWithSubjectFiltersErr ErrorIdentifier = 10033
 
-	// JSMirrorWithSubjectsErr stream mirrors can not also contain subjects
+	// JSMirrorWithSubjectsErr stream mirrors can not contain subjects
 	JSMirrorWithSubjectsErr ErrorIdentifier = 10034
 
 	// JSNoAccountErr account not found
@@ -293,12 +293,15 @@ const (
 	// JSStreamMessageExceedsMaximumErr message size exceeds maximum allowed
 	JSStreamMessageExceedsMaximumErr ErrorIdentifier = 10054
 
-	// JSStreamMirrorNotUpdatableErr Mirror configuration can not be updated
+	// JSStreamMirrorNotUpdatableErr stream mirror configuration can not be updated
 	JSStreamMirrorNotUpdatableErr ErrorIdentifier = 10055
 
 	// JSStreamMismatchErr stream name in subject does not match request
 	JSStreamMismatchErr ErrorIdentifier = 10056
 
+	// JSStreamMoveAndScaleErr can not move and scale a stream in a single update
+	JSStreamMoveAndScaleErr ErrorIdentifier = 10123
+
 	// JSStreamMsgDeleteFailedF Generic message deletion failure error string ({err})
 	JSStreamMsgDeleteFailedF ErrorIdentifier = 10057
 
@@ -437,7 +440,7 @@ var (
 		JSMirrorWithSourcesErr:                     {Code: 400, ErrCode: 10031, Description: "stream mirrors can not also contain other sources"},
 		JSMirrorWithStartSeqAndTimeErr:             {Code: 400, ErrCode: 10032, Description: "stream mirrors can not have both start seq and start time configured"},
 		JSMirrorWithSubjectFiltersErr:              {Code: 400, ErrCode: 10033, Description: "stream mirrors can not contain filtered subjects"},
-		JSMirrorWithSubjectsErr:                    {Code: 400, ErrCode: 10034, Description: "stream mirrors can not also contain subjects"},
+		JSMirrorWithSubjectsErr:                    {Code: 400, ErrCode: 10034, Description: "stream mirrors can not contain subjects"},
 		JSNoAccountErr:                             {Code: 503, ErrCode: 10035, Description: "account not found"},
 		JSNoLimitsErr:                              {Code: 400, ErrCode: 10120, Description: "no JetStream default or applicable tiered limit present"},
 		JSNoMessageFoundErr:                        {Code: 404, ErrCode: 10037, Description: "no message found"},
@@ -467,8 +470,9 @@ var (
 		JSStreamMaxBytesRequired:                   {Code: 400, ErrCode: 10113, Description: "account requires a stream config to have max bytes set"},
 		JSStreamMaxStreamBytesExceeded:             {Code: 400, ErrCode: 10122, Description: "stream max bytes exceeds account limit max stream bytes"},
 		JSStreamMessageExceedsMaximumErr:           {Code: 400, ErrCode: 10054, Description: "message size exceeds maximum allowed"},
-		JSStreamMirrorNotUpdatableErr:              {Code: 400, ErrCode: 10055, Description: "Mirror configuration can not be updated"},
+		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"},
 		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"},
@@ -483,7 +487,7 @@ var (
 		JSStreamSequenceNotMatchErr:                {Code: 503, ErrCode: 10063, Description: "expected stream sequence does not match"},
 		JSStreamSnapshotErrF:                       {Code: 500, ErrCode: 10064, Description: "snapshot failed: {err}"},
 		JSStreamStoreFailedF:                       {Code: 503, ErrCode: 10077, Description: "{err}"},
-		JSStreamSubjectOverlapErr:                  {Code: 500, ErrCode: 10065, Description: "subjects overlap with an existing stream"},
+		JSStreamSubjectOverlapErr:                  {Code: 400, ErrCode: 10065, Description: "subjects overlap with an existing stream"},
 		JSStreamTemplateCreateErrF:                 {Code: 500, ErrCode: 10066, Description: "{err}"},
 		JSStreamTemplateDeleteErrF:                 {Code: 500, ErrCode: 10067, Description: "{err}"},
 		JSStreamTemplateNotFoundErr:                {Code: 404, ErrCode: 10068, Description: "template not found"},
@@ -1219,7 +1223,7 @@ func NewJSMirrorWithSubjectFiltersError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSMirrorWithSubjectFiltersErr]
 }
 
-// NewJSMirrorWithSubjectsError creates a new JSMirrorWithSubjectsErr error: "stream mirrors can not also contain subjects"
+// NewJSMirrorWithSubjectsError creates a new JSMirrorWithSubjectsErr error: "stream mirrors can not contain subjects"
 func NewJSMirrorWithSubjectsError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
 	if ae, ok := eopts.err.(*ApiError); ok {
@@ -1597,7 +1601,7 @@ func NewJSStreamMessageExceedsMaximumError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSStreamMessageExceedsMaximumErr]
 }
 
-// NewJSStreamMirrorNotUpdatableError creates a new JSStreamMirrorNotUpdatableErr error: "Mirror configuration can not be updated"
+// NewJSStreamMirrorNotUpdatableError creates a new JSStreamMirrorNotUpdatableErr error: "stream mirror configuration can not be updated"
 func NewJSStreamMirrorNotUpdatableError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
 	if ae, ok := eopts.err.(*ApiError); ok {
@@ -1617,6 +1621,16 @@ func NewJSStreamMismatchError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSStreamMismatchErr]
 }
 
+// NewJSStreamMoveAndScaleError creates a new JSStreamMoveAndScaleErr error: "can not move and scale a stream in a single update"
+func NewJSStreamMoveAndScaleError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSStreamMoveAndScaleErr]
+}
+
 // NewJSStreamMsgDeleteFailedError creates a new JSStreamMsgDeleteFailedF error: "{err}"
 func NewJSStreamMsgDeleteFailedError(err error, opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
diff --git a/server/raft.go b/server/raft.go
index 13682dda..c33bf3c6 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -749,6 +749,7 @@ func (n *raft) AdjustClusterSize(csz int) error {
 	if csz < 2 {
 		csz = 2
 	}
+
 	// Adjust.
 	n.csz = csz
 	n.qn = n.csz/2 + 1
@@ -1200,7 +1201,7 @@ 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 {
+	if len(preferred) > 0 && preferred[0] != _EMPTY_ {
 		maybeLeader = preferred[0]
 	}
 	for peer, ps := range n.peers {
@@ -2814,6 +2815,8 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 func (n *raft) processPeerState(ps *peerState) {
 	// Update our version of peers to that of the leader.
 	n.csz = ps.clusterSize
+	n.qn = n.csz/2 + 1
+
 	old := n.peers
 	n.peers = make(map[string]*lps)
 	for _, peer := range ps.knownPeers {
diff --git a/server/stream.go b/server/stream.go
index fe245cc1..54ba8605 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -121,6 +121,9 @@ type PeerInfo struct {
 	Offline bool          `json:"offline,omitempty"`
 	Active  time.Duration `json:"active"`
 	Lag     uint64        `json:"lag,omitempty"`
+	// For migrations.
+	cluster string
+	peer    string
 }
 
 // StreamSourceInfo shows information about an upstream stream source.
@@ -204,6 +207,7 @@ type stream struct {
 	leader   string
 	lqsent   time.Time
 	catchups map[string]uint64
+	uch      chan struct{}
 }
 
 type sourceInfo struct {
@@ -396,6 +400,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		consumers: make(map[string]*consumer),
 		msgs:      s.newIPQueue(qpfx + "messages"), // of *inMsg
 		qch:       make(chan struct{}),
+		uch:       make(chan struct{}, 1),
 	}
 
 	// For no-ack consumers when we are interest retention.
@@ -512,6 +517,7 @@ func (mset *stream) streamAssignment() *streamAssignment {
 func (mset *stream) setStreamAssignment(sa *streamAssignment) {
 	mset.mu.Lock()
 	defer mset.mu.Unlock()
+
 	mset.sa = sa
 	if sa == nil {
 		return
@@ -526,6 +532,21 @@ func (mset *stream) setStreamAssignment(sa *streamAssignment) {
 		// Note below the way we subscribe here is so that we can send requests to ourselves.
 		mset.infoSub, _ = mset.srv.systemSubscribe(isubj, _EMPTY_, false, mset.sysc, mset.handleClusterStreamInfoRequest)
 	}
+
+	// Trigger update chan.
+	select {
+	case mset.uch <- struct{}{}:
+	default:
+	}
+}
+
+func (mset *stream) updateC() <-chan struct{} {
+	if mset == nil {
+		return nil
+	}
+	mset.mu.RLock()
+	defer mset.mu.RUnlock()
+	return mset.uch
 }
 
 // IsLeader will return if we are the current leader.
@@ -1151,7 +1172,7 @@ func (mset *stream) update(config *StreamConfig) error {
 	// Now update config and store's version of our config.
 	mset.cfg = *cfg
 
-	// If we are the leader never suppres update advisory, simply send.
+	// If we are the leader never suppress update advisory, simply send.
 	if mset.isLeader() {
 		mset.sendUpdateAdvisoryLocked()
 	}


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index f300aded..ce268952 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -10667,7 +10667,7 @@ func TestJetStreamClusterStreamReplicaUpdates(t *testing.T) {
 	updateReplicas(1)
 }
 
-func TestJetStreamClusterStreamReplicaUpdateFunctionCheck(t *testing.T) {
+func TestJetStreamClusterStreamAndConsumerScaleUpAndDown(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
@@ -10725,9 +10725,16 @@ func TestJetStreamClusterStreamReplicaUpdateFunctionCheck(t *testing.T) {
 		}
 	}
 
+	// Capture leader, we want to make sure when we scale down this does not change.
+	sl := c.streamLeader("$G", "TEST")
+
 	// Scale down to 1.
 	updateReplicas(1)
 
+	if sl != c.streamLeader("$G", "TEST") {
+		t.Fatalf("Expected same leader, but it changed")
+	}
+
 	// Make sure we can still send to the stream.
 	for i := 0; i < numMsgs; i++ {
 		_, err := js.Publish("foo", []byte("HELLO WORLD"))
@@ -10791,48 +10798,9 @@ func TestJetStreamClusterStreamReplicaUpdateFunctionCheck(t *testing.T) {
 }
 
 func TestJetStreamClusterStreamTagPlacement(t *testing.T) {
-	sc := createJetStreamSuperCluster(t, 3, 4)
+	sc := createJetStreamTaggedSuperCluster(t)
 	defer sc.shutdown()
 
-	reset := func(s *Server) {
-		s.mu.Lock()
-		rch := s.sys.resetCh
-		s.mu.Unlock()
-		if rch != nil {
-			rch <- struct{}{}
-		}
-		s.sendStatszUpdate()
-	}
-
-	// Make first cluster AWS, US country code.
-	for _, s := range sc.clusterForName("C1").servers {
-		opts := s.getOpts()
-		opts.Tags.Add("cloud:aws")
-		opts.Tags.Add("country:us")
-		reset(s)
-	}
-	// Make second cluster GCP, UK country code.
-	for _, s := range sc.clusterForName("C2").servers {
-		opts := s.getOpts()
-		opts.Tags.Add("cloud:gcp")
-		opts.Tags.Add("country:uk")
-		reset(s)
-	}
-	// Make third cluster AZ, JP country code.
-	for _, s := range sc.clusterForName("C3").servers {
-		opts := s.getOpts()
-		opts.Tags.Add("cloud:az")
-		opts.Tags.Add("country:jp")
-		reset(s)
-	}
-	// Make fourth cluster GCP, and SG country code.
-	for _, s := range sc.clusterForName("C4").servers {
-		opts := s.getOpts()
-		opts.Tags.Add("cloud:gcp")
-		opts.Tags.Add("country:sg")
-		reset(s)
-	}
-
 	placeOK := func(connectCluster string, tags []string, expectedCluster string) {
 		t.Helper()
 		nc, js := jsClientConnect(t, sc.clusterForName(connectCluster).randomServer())
@@ -10852,11 +10820,10 @@ func TestJetStreamClusterStreamTagPlacement(t *testing.T) {
 	placeOK("C2", []string{"cloud:aws"}, "C1")
 	placeOK("C2", []string{"country:jp"}, "C3")
 	placeOK("C1", []string{"cloud:gcp", "country:uk"}, "C2")
-	placeOK("C2", []string{"cloud:gcp", "country:sg"}, "C4")
 
 	// Case shoud not matter.
 	placeOK("C1", []string{"cloud:GCP", "country:UK"}, "C2")
-	placeOK("C2", []string{"Cloud:Gcp", "Country:Sg"}, "C4")
+	placeOK("C2", []string{"Cloud:AwS", "Country:uS"}, "C1")
 
 	placeErr := func(connectCluster string, tags []string) {
 		t.Helper()
@@ -11656,7 +11623,256 @@ func TestJetStreamClusterMemoryConsumerCompactVsSnapshot(t *testing.T) {
 		}
 		return nil
 	})
+}
+
+// This will test our ability to move streams and consumers between clusters.
+func TestJetStreamClusterMovingStreamsAndConsumers(t *testing.T) {
+	sc := createJetStreamTaggedSuperCluster(t)
+	defer sc.shutdown()
+
+	nc, js := jsClientConnect(t, sc.randomServer())
+	defer nc.Close()
+
+	for _, test := range []struct {
+		name     string
+		replicas int
+	}{
+		{"R1", 1},
+		{"R3", 3},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			replicas := test.replicas
+
+			si, err := js.AddStream(&nats.StreamConfig{
+				Name:      "MOVE",
+				Replicas:  replicas,
+				Placement: &nats.Placement{Tags: []string{"cloud:aws"}},
+			})
+			require_NoError(t, err)
+			defer js.DeleteStream("MOVE")
+
+			if si.Cluster.Name != "C1" {
+				t.Fatalf("Failed to place properly in %q, got %q", "C1", si.Cluster.Name)
+			}
+
+			for i := 0; i < 1000; i++ {
+				_, err := js.PublishAsync("MOVE", []byte("Moving on up"))
+				require_NoError(t, err)
+			}
+			select {
+			case <-js.PublishAsyncComplete():
+			case <-time.After(5 * time.Second):
+				t.Fatalf("Did not receive completion signal")
+			}
+
+			// Durable Push Consumer, so same R.
+			dpushSub, err := js.SubscribeSync("MOVE", nats.Durable("dlc"))
+			require_NoError(t, err)
+			defer dpushSub.Unsubscribe()
+
+			// Ephemeral Push Consumer, R1.
+			epushSub, err := js.SubscribeSync("MOVE")
+			require_NoError(t, err)
+			defer epushSub.Unsubscribe()
+
+			// Durable Pull Consumer, so same R.
+			dpullSub, err := js.PullSubscribe("MOVE", "dlc-pull")
+			require_NoError(t, err)
+			defer dpullSub.Unsubscribe()
+
+			// TODO(dlc) - Server supports ephemeral pulls but Go client does not yet.
+
+			si, err = js.StreamInfo("MOVE")
+			require_NoError(t, err)
+			if si.State.Consumers != 3 {
+				t.Fatalf("Expected 3 attached consumers, got %d", si.State.Consumers)
+			}
+
+			initialState := si.State
+
+			checkSubsPending(t, dpushSub, int(initialState.Msgs))
+			checkSubsPending(t, epushSub, int(initialState.Msgs))
+
+			// Ack 100
+			toAck := 100
+			for i := 0; i < toAck; i++ {
+				m, err := dpushSub.NextMsg(time.Second)
+				require_NoError(t, err)
+				m.AckSync()
+				// Ephemeral
+				m, err = epushSub.NextMsg(time.Second)
+				require_NoError(t, err)
+				m.AckSync()
+			}
+
+			// Do same with pull subscriber.
+			for _, m := range fetchMsgs(t, dpullSub, toAck, 5*time.Second) {
+				m.AckSync()
+			}
+
+			// First make sure we disallow move and replica changes in same update.
+			_, err = js.UpdateStream(&nats.StreamConfig{
+				Name:      "MOVE",
+				Placement: &nats.Placement{Tags: []string{"cloud:gcp"}},
+				Replicas:  replicas + 1,
+			})
+			require_Error(t, err, NewJSStreamMoveAndScaleError())
+
+			// Now move to new cluster.
+			si, err = js.UpdateStream(&nats.StreamConfig{
+				Name:      "MOVE",
+				Replicas:  replicas,
+				Placement: &nats.Placement{Tags: []string{"cloud:gcp"}},
+			})
+			require_NoError(t, err)
+
+			if si.Cluster.Name != "C1" {
+				t.Fatalf("Expected cluster of %q but got %q", "C1", si.Cluster.Name)
+			}
+
+			checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
+				si, err := js.StreamInfo("MOVE")
+				if err != nil {
+					return err
+				}
+				// We should see 2X peers.
+				numPeers := len(si.Cluster.Replicas)
+				if si.Cluster.Leader != _EMPTY_ {
+					numPeers++
+				}
+				if numPeers != 2*replicas {
+					return fmt.Errorf("Expected to see %d replicas, got %d", 2*replicas, numPeers)
+				}
+				return nil
+			})
+
+			// Expect a new leader to emerge and replicas to drop as a leader is elected.
+			// We have to check fast or it might complete and we will not see intermediate steps.
+			sc.waitOnStreamLeader("$G", "MOVE")
+			checkFor(t, 10*time.Second, 10*time.Millisecond, func() error {
+				si, err := js.StreamInfo("MOVE")
+				if err != nil {
+					return err
+				}
+				if len(si.Cluster.Replicas) >= 2*replicas {
+					return fmt.Errorf("Expected <%d replicas, got %d", 2*replicas, len(si.Cluster.Replicas))
+				}
+				return nil
+			})
+
+			// 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 {
+				si, err := js.StreamInfo("MOVE")
+				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) != replicas-1 {
+					return fmt.Errorf("Expected %d replicas, got %d", replicas-1, len(si.Cluster.Replicas))
+				}
+				return nil
+			})
+
+			// Check moved state is same as initial state.
+			si, err = js.StreamInfo("MOVE")
+			require_NoError(t, err)
+
+			if si.State != initialState {
+				t.Fatalf("States do not match after migration:\n%+v\nvs\n%+v", si.State, initialState)
+			}
+
+			// Make sure we can still send messages.
+			addN := toAck
+			for i := 0; i < addN; i++ {
+				_, err := js.Publish("MOVE", []byte("Done Moved"))
+				require_NoError(t, err)
+			}
+
+			si, err = js.StreamInfo("MOVE")
+			require_NoError(t, err)
+
+			expectedPushMsgs := initialState.Msgs + uint64(addN)
+			expectedPullMsgs := uint64(addN)
+
+			if si.State.Msgs != expectedPushMsgs {
+				t.Fatalf("Expected to be able to send new messages")
+			}
 
+			// 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()
+						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)
+					}
+				}
+				// 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)
+				}
+			}
+			checkPushConsumer := func(sub *nats.Subscription) {
+				checkConsumer(sub, false)
+			}
+			checkPullConsumer := func(sub *nats.Subscription) {
+				checkConsumer(sub, true)
+			}
+
+			checkPushConsumer(dpushSub)
+			checkPushConsumer(epushSub)
+			checkPullConsumer(dpullSub)
+
+			// Cleanup
+			err = js.DeleteStream("MOVE")
+			require_NoError(t, err)
+		})
+	}
 }
 
 func TestJetStreamClusterSendBadRequestResponseOnInvalidAPIRequest(t *testing.T) {
@@ -11709,6 +11925,30 @@ func (sc *supercluster) randomServer() *Server {
 	return sc.randomCluster().randomServer()
 }
 
+func (sc *supercluster) serverByName(sname string) *Server {
+	for _, c := range sc.clusters {
+		if s := c.serverByName(sname); s != nil {
+			return s
+		}
+	}
+	return nil
+}
+
+func (sc *supercluster) waitOnStreamLeader(account, stream string) {
+	sc.t.Helper()
+	expires := time.Now().Add(30 * time.Second)
+	for time.Now().Before(expires) {
+		for _, c := range sc.clusters {
+			if leader := c.streamLeader(account, stream); leader != nil {
+				time.Sleep(200 * time.Millisecond)
+				return
+			}
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
+	sc.t.Fatalf("Expected a stream leader for %q %q, got none", account, stream)
+}
+
 var jsClusterAccountsTempl = `
 	listen: 127.0.0.1:-1
 
@@ -11864,6 +12104,74 @@ var jsMixedModeGlobalAccountTempl = `
 
 var jsGWTempl = `%s{name: %s, urls: [%s]}`
 
+func createJetStreamTaggedSuperCluster(t *testing.T) *supercluster {
+	sc := createJetStreamSuperCluster(t, 3, 3)
+	sc.waitOnPeerCount(9)
+
+	reset := func(s *Server) {
+		s.mu.Lock()
+		rch := s.sys.resetCh
+		s.mu.Unlock()
+		if rch != nil {
+			rch <- struct{}{}
+		}
+		s.sendStatszUpdate()
+	}
+
+	// Make first cluster AWS, US country code.
+	for _, s := range sc.clusterForName("C1").servers {
+		s.optsMu.Lock()
+		s.opts.Tags.Add("cloud:aws")
+		s.opts.Tags.Add("country:us")
+		s.optsMu.Unlock()
+		reset(s)
+	}
+	// Make second cluster GCP, UK country code.
+	for _, s := range sc.clusterForName("C2").servers {
+		s.optsMu.Lock()
+		s.opts.Tags.Add("cloud:gcp")
+		s.opts.Tags.Add("country:uk")
+		s.optsMu.Unlock()
+		reset(s)
+	}
+	// Make third cluster AZ, JP country code.
+	for _, s := range sc.clusterForName("C3").servers {
+		s.optsMu.Lock()
+		s.opts.Tags.Add("cloud:az")
+		s.opts.Tags.Add("country:jp")
+		s.optsMu.Unlock()
+		reset(s)
+	}
+
+	ml := sc.leader()
+	js := ml.getJetStream()
+	require_True(t, js != nil)
+	js.mu.RLock()
+	defer js.mu.RUnlock()
+	cc := js.cluster
+	require_True(t, cc != nil)
+
+	// Walk and make sure all tags are registered.
+	expires := time.Now().Add(10 * time.Second)
+	for time.Now().Before(expires) {
+		allOK := true
+		for _, p := range cc.meta.Peers() {
+			si, ok := ml.nodeToInfo.Load(p.ID)
+			require_True(t, ok)
+			ni := si.(nodeInfo)
+			if len(ni.tags) == 0 {
+				allOK = false
+				reset(sc.serverByName(ni.name))
+			}
+		}
+		if allOK {
+			break
+		}
+	}
+
+	return sc
+}
+
 func createJetStreamSuperCluster(t *testing.T, numServersPer, numClusters int) *supercluster {
 	return createJetStreamSuperClusterWithTemplate(t, jsClusterTempl, numServersPer, numClusters)
 }
diff --git a/server/raft_test.go b/server/raft_test.go
index 89c983b9..a9241c72 100644
--- a/server/raft_test.go
+++ b/server/raft_test.go
@@ -82,9 +82,8 @@ func TestNRGAppendEntryDecode(t *testing.T) {
 
 	for i := 0; i < 100; i++ {
 		b := copyBytes(buf)
-		// modifying the header (idx < 42) will not result in an error by decodeAppendEntry
-		bi := 42 + rand.Intn(len(b)-42)
-		if b[bi] != 0 && bi != 40 {
+		bi := rand.Intn(len(b))
+		if b[bi] != 0 {
 			b[bi] = 0
 			_, err = node.decodeAppendEntry(b, nil, _EMPTY_)
 			require_Error(t, err, errBadAppendEntry)