Repository: nats-io/nats-server
Pull request #3217 (2022-06-25 00:38:38)

[Add] ability for operator to move streams (WIP)
    Also added:
    ability to reload tags
    special tag (!jetstream) to remove peer from peer placement
    $JS.API.SERVER.STREAM.MOVE subject to initiate move away from a server

    This changes a detail about regular stream move as well.
    Before differing cluster names where used to start/stop a transfer.
    Now only the peer list and it's size relative to configured replica
    matter.
    Once a transfer is considered completed, excess peers will be dropped
    from the beginning of the list.
    This allows transfers within the cluster as well.


Patch:
diff --git a/server/accounts.go b/server/accounts.go
index 1e415fd7..f958666d 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -1812,20 +1812,8 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 		rt = se.respType
 		lat = se.latency
 	}
-	s := dest.srv
 	dest.mu.RUnlock()
 
-	// Track if this maps us to the system account.
-	// We will always share information with them.
-	var isSysAcc bool
-	if s != nil {
-		s.mu.Lock()
-		if s.sys != nil && dest == s.sys.account {
-			isSysAcc = true
-		}
-		s.mu.Unlock()
-	}
-
 	a.mu.Lock()
 	if a.imports.services == nil {
 		a.imports.services = make(map[string]*serviceImport)
@@ -1862,8 +1850,7 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 			}
 		}
 	}
-	// Turn on sharing by default if importing from system services.
-	share := isSysAcc
+	var share bool
 	if claim != nil {
 		share = claim.Share
 	}
diff --git a/server/client.go b/server/client.go
index 37add98e..77e9891f 100644
--- a/server/client.go
+++ b/server/client.go
@@ -3848,12 +3848,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	// However the GetNext for consumers is a no-op and causes buildups of service imports,
 	// response service imports and rrMap entries which all will need to simply expire.
 	// TODO(dlc) - Come up with something better.
-	if checkJSGetNext && si.se != nil && si.se.acc == c.srv.SystemAccount() {
-		shouldReturn = true
-	}
-
-	// Check for short circuit return.
-	if shouldReturn {
+	if shouldReturn || (checkJSGetNext && si.se != nil && si.se.acc == c.srv.SystemAccount()) {
 		return
 	}
 
@@ -3920,6 +3915,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	// Place our client info for the request in the original message.
 	// This will survive going across routes, etc.
 	if !isResponse {
+		isSysImport := si.acc == c.srv.SystemAccount()
 		var ci *ClientInfo
 		if hadPrevSi && c.pa.hdr >= 0 {
 			var cis ClientInfo
@@ -3928,15 +3924,22 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 				ci.Service = acc.Name
 				// Check if we are moving into a share details account from a non-shared
 				// and add in server and cluster details.
-				if !share && si.share {
+				if !share && (si.share || isSysImport) {
 					c.addServerAndClusterInfo(ci)
 				}
 			}
 		} else if c.kind != LEAF || c.pa.hdr < 0 || len(getHeader(ClientInfoHdr, msg[:c.pa.hdr])) == 0 {
 			ci = c.getClientInfo(share)
-		} else if c.kind == LEAF && si.share {
+			// If we did not share but the imports destination is the system account add in the server and cluster info.
+			if !share && isSysImport {
+				c.addServerAndClusterInfo(ci)
+			}
+		} else if c.kind == LEAF && (si.share || isSysImport) {
 			// We have a leaf header here for ci, augment as above.
 			ci = c.getClientInfo(si.share)
+			if !si.share && isSysImport {
+				c.addServerAndClusterInfo(ci)
+			}
 		}
 		// Set clientInfo if present.
 		if ci != nil {
diff --git a/server/consumer.go b/server/consumer.go
index b801f199..3111e09e 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1963,6 +1963,7 @@ func (o *consumer) readStoredState(slseq uint64) error {
 }
 
 // Apply the consumer stored state.
+// Lock should be held.
 func (o *consumer) applyState(state *ConsumerState) {
 	if state == nil {
 		return
@@ -1993,6 +1994,7 @@ func (o *consumer) applyState(state *ConsumerState) {
 }
 
 // Sets our store state from another source. Used in clustered mode on snapshot restore.
+// Lock should be held.
 func (o *consumer) setStoreState(state *ConsumerState) error {
 	if state == nil || o.store == nil {
 		return nil
diff --git a/server/events.go b/server/events.go
index bec38180..8d9d91bd 100644
--- a/server/events.go
+++ b/server/events.go
@@ -296,9 +296,9 @@ func (s *Server) internalSendLoop(wg *sync.WaitGroup) {
 	defer wg.Done()
 
 RESET:
-	s.mu.Lock()
+	s.mu.RLock()
 	if s.sys == nil || s.sys.sendq == nil {
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		return
 	}
 	sysc := s.sys.client
@@ -314,7 +314,7 @@ RESET:
 	if s.gateway.enabled {
 		cluster = s.getGatewayName()
 	}
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	// Grab tags.
 	tags := s.getOpts().Tags
@@ -541,17 +541,17 @@ func (s *Server) eventsRunning() bool {
 	if s == nil {
 		return false
 	}
-	s.mu.Lock()
+	s.mu.RLock()
 	er := s.running && s.eventsEnabled()
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return er
 }
 
 // EventsEnabled will report if the server has internal events enabled via
 // a defined system account.
 func (s *Server) EventsEnabled() bool {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.eventsEnabled()
 }
 
@@ -564,8 +564,8 @@ func (s *Server) eventsEnabled() bool {
 // TrackedRemoteServers returns how many remote servers we are tracking
 // from a system events perspective.
 func (s *Server) TrackedRemoteServers() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if !s.running || !s.eventsEnabled() {
 		return -1
 	}
@@ -764,8 +764,8 @@ func getHash(name string) []byte {
 
 // Returns the node name for this server which is a hash of the server name.
 func (s *Server) Node() string {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.sys != nil {
 		return s.sys.shash
 	}
@@ -977,9 +977,9 @@ func (s *Server) initEventTracking() {
 func (s *Server) registerSystemImportsForExisting() {
 	var accounts []*Account
 
-	s.mu.Lock()
+	s.mu.RLock()
 	if s.sys == nil {
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		return
 	}
 	sacc := s.sys.account
@@ -990,7 +990,7 @@ func (s *Server) registerSystemImportsForExisting() {
 		}
 		return true
 	})
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	for _, a := range accounts {
 		s.registerSystemImports(a)
@@ -1290,9 +1290,9 @@ func (s *Server) leafNodeConnected(sub *subscription, _ *client, _ *Account, sub
 		return
 	}
 
-	s.mu.Lock()
+	s.mu.RLock()
 	na := m.Account == _EMPTY_ || !s.eventsEnabled() || !s.gateway.enabled
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	if na {
 		return
@@ -1382,10 +1382,7 @@ func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
 		return true
 	}
 	if fOpts.Cluster != _EMPTY_ {
-		s.mu.Lock()
-		cluster := s.info.Cluster
-		s.mu.Unlock()
-		if !strings.Contains(cluster, fOpts.Cluster) {
+		if !strings.Contains(s.ClusterName(), fOpts.Cluster) {
 			return true
 		}
 	}
@@ -1808,6 +1805,7 @@ func (s *Server) sendAuthErrorEvent(c *client) {
 	}
 	eid := s.nextEventID()
 	s.mu.Unlock()
+
 	now := time.Now().UTC()
 	c.mu.Lock()
 	m := DisconnectEventMsg{
@@ -1908,13 +1906,13 @@ func (s *Server) sysUnsubscribe(sub *subscription) {
 	if sub == nil {
 		return
 	}
-	s.mu.Lock()
+	s.mu.RLock()
 	if !s.eventsEnabled() {
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		return
 	}
 	c := sub.client
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	if c != nil {
 		c.processUnsub(sub.sid)
@@ -1996,13 +1994,13 @@ func (s *Server) remoteLatencyUpdate(sub *subscription, _ *client, _ *Account, s
 // This is used for all inbox replies so that we do not send supercluster wide interest
 // updates for every request. Same trick used in modern NATS clients.
 func (s *Server) inboxReply(sub *subscription, c *client, acc *Account, subject, reply string, msg []byte) {
-	s.mu.Lock()
+	s.mu.RLock()
 	if !s.eventsEnabled() || s.sys.replies == nil {
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		return
 	}
 	cb, ok := s.sys.replies[subject]
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	if ok && cb != nil {
 		cb(sub, c, acc, subject, reply, msg)
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 3d5bac47..ac719b61 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -187,6 +187,16 @@ const (
 	// Will return JSON response.
 	JSApiRemoveServer = "$JS.API.SERVER.REMOVE"
 
+	// JSApiServerStreamMove is the endpoint to move streams off a server
+	// Only works from system account.
+	// Will return JSON response.
+	JSApiServerStreamMove = "$JS.API.SERVER.STREAM.MOVE"
+
+	// JSApiServerInfo is the endpoint to abtain which stream a server contains
+	// Only works from system account.
+	// Will return JSON response.
+	//JSApiServerStreamInfo = "$JS.API.SERVER.STREAM.INFO"
+
 	// 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"
@@ -558,6 +568,36 @@ type JSApiMetaServerRemoveResponse struct {
 
 const JSApiMetaServerRemoveResponseType = "io.nats.jetstream.api.v1.meta_server_remove_response"
 
+/*
+// JSApiMetaServerStreamInfoRequest will provide which streams are located on a particular server
+type JSApiMetaServerStreamInfoRequest struct {
+	// Server name of the peer to be evacuated.
+	Server string `json:"peer"`
+}
+
+// JSApiMetaServerStreamInfoResponse is the response to a peer evacuation request in the meta group.
+type JSApiMetaServerStreamInfoResponse struct {
+	ApiResponse
+	Success bool                      `json:"success,omitempty"`
+	Content map[string][]StreamConfig `json:"content,omitempty"`
+}
+
+const JSApiMetaServerStreamInfoType = "io.nats.jetstream.api.v1.meta_server_stream_info"
+*/
+
+// JSApiMetaServerStreamMoveRequest will move a stream on a server to another
+// response to this will come as JSApiStreamUpdateResponse/JSApiStreamUpdateResponseType
+type JSApiMetaServerStreamMoveRequest struct {
+	// Server name of the peer to be evacuated.
+	Server string `json:"peer"`
+	// Account name the stream to move is in
+	Account string `json:"account"`
+	// Name of stream to move
+	Stream string `json:"string"`
+	// Ephemeral placement tags for the move
+	Tags []string `json:"tags,omitempty"`
+}
+
 // JSApiMsgGetRequest get a message request.
 type JSApiMsgGetRequest struct {
 	Seq     uint64 `json:"seq,omitempty"`
@@ -1360,9 +1400,9 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
 		// 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)
+			go s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg, nil)
 		} else {
-			s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg)
+			s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg, nil)
 		}
 		return
 	}
@@ -2156,6 +2196,262 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, _ *Ac
 	s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 }
 
+/*
+// Request to have the metaleader provide which streams a server contains
+func (s *Server) jsLeaderServerStreamInfoRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
+	if c == nil || !s.JetStreamEnabled() {
+		return
+	}
+
+	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	if err != nil {
+		s.Warnf(badAPIRequestT, msg)
+		return
+	}
+
+	js, cc := s.getJetStreamCluster()
+	if js == nil && (cc != nil && cc.meta == nil) {
+		return
+	}
+
+	// Extra checks here but only leader is listening.
+	js.mu.RLock()
+	isLeader := cc.isLeader()
+	js.mu.RUnlock()
+
+	if !isLeader {
+		return
+	}
+
+	var resp = JSApiMetaServerStreamInfoResponse{ApiResponse: ApiResponse{Type: JSApiMetaServerStreamInfoType}}
+
+	if isEmptyRequest(msg) {
+		resp.Error = NewJSBadRequestError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	var req JSApiMetaServerStreamInfoRequest
+	if err := json.Unmarshal(msg, &req); err != nil {
+		resp.Error = NewJSInvalidJSONError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	if cc == nil {
+		// single server logic
+		if req.Server != s.info.Name {
+			resp.Error = NewJSClusterServerNotMemberError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+
+		resp.Content = make(map[string][]StreamConfig)
+		resp.Success = true
+		js.mu.RLock()
+		for aName, jsa := range js.accounts {
+			jsa.mu.RLock()
+			for _, s := range jsa.streams {
+				resp.Content[aName] = append(resp.Content[aName], s.cfg)
+			}
+			jsa.mu.RUnlock()
+		}
+		js.mu.RUnlock()
+		return
+	}
+	// clustered code
+
+	var found string
+	js.mu.RLock()
+	for _, p := range cc.meta.Peers() {
+		si, ok := s.nodeToInfo.Load(p.ID)
+		if ok && si.(nodeInfo).name == req.Server {
+			found = p.ID
+			break
+		}
+	}
+	js.mu.RUnlock()
+
+	if found == _EMPTY_ {
+		resp.Error = NewJSClusterServerNotMemberError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	resp.Content = make(map[string][]StreamConfig)
+	js.mu.Lock()
+	for aName, a := range cc.streams {
+		for _, s := range a {
+			contains := false
+			for _, p := range s.Group.Peers {
+				if p == found {
+					contains = true
+					break
+				}
+			}
+			if contains {
+				resp.Content[aName] = append(resp.Content[aName], *s.Config)
+			}
+		}
+	}
+	js.mu.Unlock()
+
+	resp.Success = true
+	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+}
+*/
+
+// Request to have the metaleader move a stream on a peer to another
+func (s *Server) jsLeaderServerStreamMoveRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
+	if c == nil || !s.JetStreamEnabled() {
+		return
+	}
+
+	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	if err != nil {
+		s.Warnf(badAPIRequestT, msg)
+		return
+	}
+
+	js, cc := s.getJetStreamCluster()
+	if js == nil || cc == nil || cc.meta == nil {
+		return
+	}
+
+	// Extra checks here but only leader is listening.
+	js.mu.RLock()
+	isLeader := cc.isLeader()
+	js.mu.RUnlock()
+
+	if !isLeader {
+		return
+	}
+
+	var resp = JSApiStreamUpdateResponse{ApiResponse: ApiResponse{Type: JSApiStreamUpdateResponseType}}
+
+	if isEmptyRequest(msg) {
+		resp.Error = NewJSBadRequestError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	var req JSApiMetaServerStreamMoveRequest
+	if err := json.Unmarshal(msg, &req); err != nil {
+		resp.Error = NewJSInvalidJSONError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	var srcPeer string
+	js.mu.RLock()
+	for _, p := range cc.meta.Peers() {
+		si, ok := s.nodeToInfo.Load(p.ID)
+		if ok && si.(nodeInfo).name == req.Server {
+			srcPeer = p.ID
+			break
+		}
+	}
+	js.mu.RUnlock()
+
+	if srcPeer == _EMPTY_ {
+		resp.Error = NewJSClusterServerNotMemberError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	targetAcc, ok := s.accounts.Load(req.Account)
+	if !ok {
+		resp.Error = NewJSNoAccountError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	streamFound := false
+	cfg := StreamConfig{}
+	currPeers := []string{}
+	currCluster := _EMPTY_
+	js.mu.Lock()
+	streams, ok := cc.streams[req.Account]
+	if ok {
+		sa, ok := streams[req.Stream]
+		if ok {
+			cfg = *sa.Config
+			streamFound = true
+			currPeers = sa.Group.Peers
+			currCluster = sa.Group.Cluster
+		}
+	}
+	js.mu.Unlock()
+
+	// make sure src peer is first. Removal will drop peers from the left
+	for i := 0; i < len(currPeers); i++ {
+		if currPeers[i] == srcPeer {
+			currPeers[i] = currPeers[0]
+			currPeers[0] = srcPeer
+			break
+		}
+	}
+
+	if !streamFound {
+		resp.Error = NewJSStreamNotFoundError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	// make sure client is scoped to requested account
+	ciNew := *(ci)
+	ciNew.Account = req.Account
+
+	// backup placement such that peers can be looked up with modified tag list
+	var origPlacement *Placement
+	if cfg.Placement != nil {
+		tmp := *cfg.Placement
+		origPlacement = &tmp
+	}
+
+	if len(req.Tags) > 0 {
+		if cfg.Placement == nil {
+			cfg.Placement = &Placement{}
+		}
+		cfg.Placement.Tags = append(cfg.Placement.Tags, req.Tags...)
+	}
+
+	peers := cc.selectPeerGroup(cfg.Replicas+1, currCluster, &cfg, currPeers)
+	if len(peers) <= cfg.Replicas {
+		// since expanding in the same cluster did not yield a result, try in different cluster
+		peers = nil
+
+		clusters := map[string]struct{}{}
+		s.nodeToInfo.Range(func(_, ni interface{}) bool {
+			if currCluster != ni.(nodeInfo).cluster {
+				clusters[ni.(nodeInfo).cluster] = struct{}{}
+			}
+			return true
+		})
+		for cluster := range clusters {
+			newPeers := cc.selectPeerGroup(cfg.Replicas, cluster, &cfg, nil)
+			if len(newPeers) >= cfg.Replicas {
+				peers = append([]string{}, currPeers...)
+				peers = append(peers, newPeers[:cfg.Replicas]...)
+				break
+			}
+		}
+		if peers == nil {
+			resp.Error = NewJSClusterNoPeersError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+	}
+
+	cfg.Placement = origPlacement
+
+	if c.kind != ROUTER && c.kind != GATEWAY {
+		go s.jsClusteredStreamUpdateRequest(&ciNew, targetAcc.(*Account), subject, reply, rmsg, &cfg, peers)
+	} else {
+		s.jsClusteredStreamUpdateRequest(&ciNew, targetAcc.(*Account), subject, reply, rmsg, &cfg, peers)
+	}
+}
+
 // Request to have the meta leader stepdown.
 // These will only be received the the meta leaders, so less checking needed.
 func (s *Server) jsLeaderStepDownRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index d3c204a2..fa2791bb 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -52,6 +52,10 @@ type jetStreamCluster struct {
 	stepdown *subscription
 	// System level requests to remove a peer.
 	peerRemove *subscription
+	// System level request to evacuate a peer.
+	peerStreamMove *subscription
+	// System level request to obtain which streams a server
+	//peerStreamInfo *subscription
 }
 
 // Used to guide placement of streams and meta controllers in clustered JetStream.
@@ -158,6 +162,7 @@ const (
 	defaultStoreDirName  = "_js_"
 	defaultMetaGroupName = "_meta_"
 	defaultMetaFSBlkSize = 1024 * 1024
+	jsExcludePlacement   = "!jetstream"
 )
 
 // Returns information useful in mixed mode.
@@ -1316,7 +1321,7 @@ func (js *jetStream) processRemovePeer(peer string) {
 
 // Remove old peers after the new peers are caught up.
 // We are the old stream leader here.
-func (js *jetStream) removeOldPeers(mset *stream, newPreferred string) {
+func (js *jetStream) removeOldPeers(mset *stream, newPreferred string, newPeerSet []string) {
 	// Make sure still valid.
 	mset.mu.Lock()
 	isValid := mset.qch != nil
@@ -1327,7 +1332,6 @@ func (js *jetStream) removeOldPeers(mset *stream, newPreferred string) {
 	}
 
 	sa := mset.streamAssignment()
-	ci := js.clusterInfo(mset.raftGroup())
 
 	js.mu.Lock()
 	defer js.mu.Unlock()
@@ -1339,13 +1343,7 @@ func (js *jetStream) removeOldPeers(mset *stream, newPreferred string) {
 
 	cc, csa := js.cluster, sa.copyGroup()
 	numExpandedPeers := len(csa.Group.Peers)
-	csa.Group.Peers = csa.Group.Peers[:0]
-
-	for _, r := range ci.Replicas {
-		if r.cluster != ci.Name {
-			csa.Group.Peers = append(csa.Group.Peers, r.peer)
-		}
-	}
+	csa.Group.Peers = newPeerSet
 
 	// Now do consumers actually first here, followed by the owning stream.
 	for _, ca := range csa.consumers {
@@ -1360,6 +1358,9 @@ func (js *jetStream) removeOldPeers(mset *stream, newPreferred string) {
 		cc.meta.ForwardProposal(encodeAddConsumerAssignment(cca))
 	}
 
+	si, _ := js.srv.nodeToInfo.Load(newPreferred)
+	csa.Group.Cluster = si.(nodeInfo).cluster
+
 	csa.Group.Preferred = newPreferred
 	cc.meta.ForwardProposal(encodeUpdateStreamAssignment(csa))
 }
@@ -1797,7 +1798,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 
 			// We may receive a leader change after the stream assignment which would cancel us
 			// monitoring for this closely. So re-assess our state here as well.
-			migrating, peerGroup = mset.isMigrating()
+			migrating, peerGroup = mset.isMigrating(cc.meta.ID())
 
 			// Check for migrations here. We set the state on the stream assignment update below.
 			if isLeader && migrating {
@@ -1812,7 +1813,11 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		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 cc == nil || cc.meta == nil {
+				migrating = false
+			} else {
+				migrating, peerGroup = mset.isMigrating(cc.meta.ID())
+			}
 			// 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.
@@ -1829,26 +1834,50 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			// Check to see that we have someone caught up.
 			// TODO(dlc) - For now start checking after a second in order to give proper time to kick in any catchup logic needed.
 			// What we really need to do longer term is know if we need catchup and make sure that process has kicked off and/or completed.
-			ci := js.clusterInfo(mset.raftGroup())
+			rg := mset.raftGroup()
+			ci := js.clusterInfo(rg)
 			// The polling interval of one second allows this to be kicked in if needed.
 			if mset.hasCatchupPeers() {
 				mset.checkClusterInfo(ci)
 			}
 			// Track the new peers and check the ones that are current.
-			var newPeers []*PeerInfo
-			quorum := mset.cfg.Replicas/2 + 1
-			for _, r := range ci.Replicas {
-				if r.cluster != ci.Name {
-					if r.Current {
-						newPeers = append(newPeers, r)
+			if len(ci.Replicas)+1 != len(rg.Peers) {
+				continue
+			}
+			replicas := mset.cfg.Replicas
+			if len(rg.Peers) <= replicas {
+				stopMigrationMonitoring()
+				continue
+			}
+			//
+			toSkip := len(rg.Peers) - replicas
+			newPeerSet := rg.Peers[toSkip:]
+			currentCount := 0
+			firstPeer := _EMPTY_
+			for _, peer := range newPeerSet {
+				foundCurrent := peer == mset.leader
+				if !foundCurrent {
+					for _, p := range ci.Replicas {
+						if peer == string(getHash(p.Name)) {
+							if p.Current {
+								foundCurrent = true
+							}
+							break
+						}
+					}
+				}
+				if foundCurrent {
+					currentCount++
+					if firstPeer == _EMPTY_ {
+						firstPeer = peer
 					}
 				}
 			}
 			// If all are current we are good, or if we have some offline and we have a quorum.
-			if lnp := len(newPeers); lnp >= quorum {
+			if quorum := mset.cfg.Replicas/2 + 1; currentCount >= quorum {
 				stopMigrationMonitoring()
 				// Remove the old peers and transfer leadership.
-				time.AfterFunc(2*time.Second, func() { js.removeOldPeers(mset, newPeers[0].peer) })
+				time.AfterFunc(2*time.Second, func() { js.removeOldPeers(mset, firstPeer, newPeerSet) })
 			}
 		case err := <-restoreDoneCh:
 			// We have completed a restore from snapshot on this server. The stream assignment has
@@ -1960,9 +1989,9 @@ const (
 )
 
 // Determine if we are migrating and if so if we are part of the old or new set.
-func (mset *stream) isMigrating() (bool, peerMigrateType) {
+func (mset *stream) isMigrating(selfPeer string) (bool, peerMigrateType) {
 	mset.mu.RLock()
-	s, js, sa := mset.srv, mset.js, mset.sa
+	js, sa := mset.js, mset.sa
 	mset.mu.RUnlock()
 
 	js.mu.RLock()
@@ -1978,10 +2007,15 @@ func (mset *stream) isMigrating() (bool, peerMigrateType) {
 		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
+	for i, peer := range sa.Group.Peers {
+		if peer == selfPeer {
+			if i >= sa.Config.Replicas {
+				return true, newPeerGroup
+			}
+			break
+		}
 	}
+
 	return true, oldPeerGroup
 }
 
@@ -3246,7 +3280,9 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 
 	// If we have an initial state set apply that now.
 	if state != nil && o != nil {
+		o.mu.Lock()
 		err = o.setStoreState(state)
+		o.mu.Unlock()
 	}
 	if err != nil {
 		if IsNatsErr(err, JSConsumerStoreFailedErrF) {
@@ -4018,6 +4054,14 @@ func (js *jetStream) startUpdatesSub() {
 	if cc.peerRemove == nil {
 		cc.peerRemove, _ = s.systemSubscribe(JSApiRemoveServer, _EMPTY_, false, c, s.jsLeaderServerRemoveRequest)
 	}
+	/*
+		if cc.peerStreamInfo == nil {
+			cc.peerStreamInfo, _ = s.systemSubscribe(JSApiServerStreamInfo, _EMPTY_, false, c, s.jsLeaderServerStreamInfoRequest)
+		}
+	*/
+	if cc.peerStreamMove == nil {
+		cc.peerStreamMove, _ = s.systemSubscribe(JSApiServerStreamMove, _EMPTY_, false, c, s.jsLeaderServerStreamMoveRequest)
+	}
 }
 
 // Lock should be held.
@@ -4039,6 +4083,16 @@ func (js *jetStream) stopUpdatesSub() {
 		cc.s.sysUnsubscribe(cc.peerRemove)
 		cc.peerRemove = nil
 	}
+	/*
+		if cc.peerStreamInfo != nil {
+			cc.s.sysUnsubscribe(cc.peerStreamInfo)
+			cc.peerStreamInfo = nil
+		}
+	*/
+	if cc.peerStreamMove != nil {
+		cc.s.sysUnsubscribe(cc.peerStreamMove)
+		cc.peerStreamMove = nil
+	}
 }
 
 func (js *jetStream) processLeaderChange(isLeader bool) {
@@ -4211,6 +4265,10 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 			}
 		}
 
+		if ni.tags.Contains(jsExcludePlacement) {
+			continue
+		}
+
 		if len(tags) > 0 {
 			matched := true
 			for _, t := range tags {
@@ -4503,7 +4561,7 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 	cc.meta.Propose(encodeAddStreamAssignment(sa))
 }
 
-func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, cfg *StreamConfig) {
+func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, cfg *StreamConfig, peerSet []string) {
 	js, cc := s.getJetStreamCluster()
 	if js == nil || cc == nil {
 		return
@@ -4567,7 +4625,12 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 
 	// Check for a move update.
 	// TODO(dlc) - Should add a resolve from Tags to cluster and check that vs reflect.
-	isMoveRequest := newCfg.Placement != nil && !reflect.DeepEqual(osa.Config.Placement, newCfg.Placement)
+	var isMoveRequest bool
+	if len(peerSet) > 0 {
+		isMoveRequest = true
+	} else {
+		isMoveRequest = newCfg.Placement != nil && !reflect.DeepEqual(osa.Config.Placement, newCfg.Placement)
+	}
 
 	// Check for replica changes.
 	isReplicaChange := newCfg.Replicas != osa.Config.Replicas
@@ -4700,42 +4763,51 @@ 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]
+		if len(peerSet) == 0 {
+			nrg := js.createGroupForStream(ci, newCfg)
+			if nrg == nil {
+				resp.Error = NewJSInsufficientResourcesError()
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+				return
 			}
-			// 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])
+			// filter peers present in both sets
+			for _, peer := range rg.Peers {
+				found := false
+				for _, newPeer := range nrg.Peers {
+					if peer == newPeer {
+						found = true
+						break
+					}
 				}
-				// Make sure to set if not already set.
-				if cca.Group.Preferred == _EMPTY_ {
-					cca.Group.Preferred = cca.Group.Peers[0]
+				if !found {
+					peerSet = append(peerSet, peer)
 				}
-				// We can not propose here before the stream itself so we collect them.
-				consumers = append(consumers, cca)
 			}
+			peerSet = append(peerSet, nrg.Peers...)
+		}
+		origPeers := len(rg.Peers)
+		if origPeers == 1 {
+			rg.Preferred = peerSet[0]
+		}
+		rg.Peers = peerSet
+
+		for _, ca := range osa.consumers {
+			cca := ca.copyGroup()
+			// Ephemerals are R=1, so only auto-remap if consumer peer count == nrg peer count.
+			if len(ca.Group.Peers) == newCfg.Replicas {
+				cca.Group.Peers = peerSet
+			} else {
+				// This is an ephemeral, so R1. Just randomly pick a single peer from the new set.
+				randPeer := peerSet[len(peerSet)-newCfg.Replicas:][rand.Int31n(int32(newCfg.Replicas))]
+				cca.Group.Peers = append(cca.Group.Peers, randPeer)
+			}
+			// 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)
 		}
 	} else {
 		// All other updates make sure no preferred is set.
diff --git a/server/reload.go b/server/reload.go
index 2ae53172..1f7d3a64 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -293,6 +293,15 @@ func (a *authTimeoutOption) Apply(server *Server) {
 	server.Noticef("Reloaded: authorization timeout = %v", a.newValue)
 }
 
+// tagsOption implements the option interface for the `tags` setting.
+type tagsOption struct {
+	noopOption // Not authOption because this is a no-op; will be reloaded with options.
+}
+
+func (u *tagsOption) Apply(server *Server) {
+	server.Noticef("Reloaded: tags")
+}
+
 // usersOption implements the option interface for the authorization `users`
 // setting.
 type usersOption struct {
@@ -1000,6 +1009,8 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			diffOpts = append(diffOpts, &usernameOption{})
 		case "password":
 			diffOpts = append(diffOpts, &passwordOption{})
+		case "tags":
+			diffOpts = append(diffOpts, &tagsOption{})
 		case "authorization":
 			diffOpts = append(diffOpts, &authorizationOption{})
 		case "authtimeout":
diff --git a/server/server.go b/server/server.go
index 21e2b0f0..e9a3e879 100644
--- a/server/server.go
+++ b/server/server.go
@@ -112,7 +112,7 @@ type Server struct {
 	// How often user logon fails due to the issuer account not being pinned.
 	pinnedAccFail uint64
 	stats
-	mu                  sync.Mutex
+	mu                  sync.RWMutex
 	kp                  nkeys.KeyPair
 	prand               *rand.Rand
 	info                Info
@@ -571,9 +571,9 @@ func (s *Server) logRejectedTLSConns() {
 
 // clusterName returns our cluster name which could be dynamic.
 func (s *Server) ClusterName() string {
-	s.mu.Lock()
+	s.mu.RLock()
 	cn := s.info.Cluster
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return cn
 }
 
@@ -726,9 +726,9 @@ func (s *Server) setOpts(opts *Options) {
 }
 
 func (s *Server) globalAccount() *Account {
-	s.mu.Lock()
+	s.mu.RLock()
 	gacc := s.gacc
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return gacc
 }
 
@@ -936,7 +936,7 @@ func (s *Server) globalAccountOnly() bool {
 		return false
 	}
 
-	s.mu.Lock()
+	s.mu.RLock()
 	s.accounts.Range(func(k, v interface{}) bool {
 		acc := v.(*Account)
 		// Ignore global and system
@@ -946,7 +946,7 @@ func (s *Server) globalAccountOnly() bool {
 		hasOthers = true
 		return false
 	})
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	return !hasOthers
 }
@@ -976,10 +976,10 @@ func (s *Server) ActivePeers() (peers []string) {
 // isTrustedIssuer will check that the issuer is a trusted public key.
 // This is used to make sure an account was signed by a trusted operator.
 func (s *Server) isTrustedIssuer(issuer string) bool {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	// If we are not running in trusted mode and there is no issuer, that is ok.
-	if s.trustedKeys == nil && issuer == "" {
+	if s.trustedKeys == nil && issuer == _EMPTY_ {
 		return true
 	}
 	for _, tk := range s.trustedKeys {
@@ -1085,9 +1085,9 @@ func (s *Server) Running() bool {
 
 // Protected check on running state
 func (s *Server) isRunning() bool {
-	s.mu.Lock()
+	s.mu.RLock()
 	running := s.running
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return running
 }
 
@@ -1121,12 +1121,12 @@ func (s *Server) decActiveAccounts() {
 // range over all accounts in the sync.Map to count.
 func (s *Server) numAccounts() int {
 	count := 0
-	s.mu.Lock()
+	s.mu.RLock()
 	s.accounts.Range(func(k, v interface{}) bool {
 		count++
 		return true
 	})
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return count
 }
 
@@ -1187,19 +1187,19 @@ func (s *Server) SetSystemAccount(accName string) error {
 // SystemAccount returns the system account if set.
 func (s *Server) SystemAccount() *Account {
 	var sacc *Account
-	s.mu.Lock()
+	s.mu.RLock()
 	if s.sys != nil {
 		sacc = s.sys.account
 	}
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return sacc
 }
 
 // GlobalAccount returns the global account.
 // Default clients will use the global account.
 func (s *Server) GlobalAccount() *Account {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.gacc
 }
 
@@ -2801,30 +2801,30 @@ func (s *Server) addToTempClients(cid uint64, c *client) bool {
 
 // NumRoutes will report the number of registered routes.
 func (s *Server) NumRoutes() int {
-	s.mu.Lock()
+	s.mu.RLock()
 	nr := len(s.routes)
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return nr
 }
 
 // NumRemotes will report number of registered remotes.
 func (s *Server) NumRemotes() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return len(s.remotes)
 }
 
 // NumLeafNodes will report number of leaf node connections.
 func (s *Server) NumLeafNodes() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return len(s.leafs)
 }
 
 // NumClients will report the number of registered clients.
 func (s *Server) NumClients() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return len(s.clients)
 }
 
@@ -2835,22 +2835,22 @@ func (s *Server) GetClient(cid uint64) *client {
 
 // getClient will return the client associated with cid.
 func (s *Server) getClient(cid uint64) *client {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.clients[cid]
 }
 
 // GetLeafNode returns the leafnode associated with the cid.
 func (s *Server) GetLeafNode(cid uint64) *client {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.leafs[cid]
 }
 
 // NumSubscriptions will report how many subscriptions are active.
 func (s *Server) NumSubscriptions() uint32 {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.numSubscriptions()
 }
 
@@ -2875,15 +2875,15 @@ func (s *Server) NumSlowConsumers() int64 {
 
 // ConfigTime will report the last time the server configuration was loaded.
 func (s *Server) ConfigTime() time.Time {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.configTime
 }
 
 // Addr will return the net.Addr object for the current listener.
 func (s *Server) Addr() net.Addr {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.listener == nil {
 		return nil
 	}
@@ -2892,8 +2892,8 @@ func (s *Server) Addr() net.Addr {
 
 // MonitorAddr will return the net.Addr object for the monitoring listener.
 func (s *Server) MonitorAddr() *net.TCPAddr {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.http == nil {
 		return nil
 	}
@@ -2902,8 +2902,8 @@ func (s *Server) MonitorAddr() *net.TCPAddr {
 
 // ClusterAddr returns the net.Addr object for the route listener.
 func (s *Server) ClusterAddr() *net.TCPAddr {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.routeListener == nil {
 		return nil
 	}
@@ -2912,8 +2912,8 @@ func (s *Server) ClusterAddr() *net.TCPAddr {
 
 // ProfilerAddr returns the net.Addr object for the profiler listener.
 func (s *Server) ProfilerAddr() *net.TCPAddr {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.profiler == nil {
 		return nil
 	}
@@ -2932,14 +2932,14 @@ func (s *Server) readyForConnections(d time.Duration) error {
 
 	end := time.Now().Add(d)
 	for time.Now().Before(end) {
-		s.mu.Lock()
+		s.mu.RLock()
 		chk["server"] = info{ok: s.listener != nil, err: s.listenerErr}
 		chk["route"] = info{ok: (opts.Cluster.Port == 0 || s.routeListener != nil), err: s.routeListenerErr}
 		chk["gateway"] = info{ok: (opts.Gateway.Name == _EMPTY_ || s.gatewayListener != nil), err: s.gatewayListenerErr}
 		chk["leafNode"] = info{ok: (opts.LeafNode.Port == 0 || s.leafNodeListener != nil), err: s.leafNodeListenerErr}
 		chk["websocket"] = info{ok: (opts.Websocket.Port == 0 || s.websocket.listener != nil), err: s.websocket.listenerErr}
 		chk["mqtt"] = info{ok: (opts.MQTT.Port == 0 || s.mqtt.listener != nil), err: s.mqtt.listenerErr}
-		s.mu.Unlock()
+		s.mu.RUnlock()
 
 		var numOK int
 		for _, inf := range chk {
@@ -3021,20 +3021,20 @@ func (s *Server) startGoRoutine(f func()) bool {
 }
 
 func (s *Server) numClosedConns() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.closed.len()
 }
 
 func (s *Server) totalClosedConns() uint64 {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.closed.totalConns()
 }
 
 func (s *Server) closedClients() []*closedClient {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.closed.closedClients()
 }
 
@@ -3186,7 +3186,7 @@ func (s *Server) PortsInfo(maxWait time.Duration) *Ports {
 	if s.readyForListeners(maxWait) {
 		opts := s.getOpts()
 
-		s.mu.Lock()
+		s.mu.RLock()
 		tls := s.info.TLSRequired
 		listener := s.listener
 		httpListener := s.http
@@ -3194,7 +3194,7 @@ func (s *Server) PortsInfo(maxWait time.Duration) *Ports {
 		profileListener := s.profiler
 		wsListener := s.websocket.listener
 		wss := s.websocket.tls
-		s.mu.Unlock()
+		s.mu.RUnlock()
 
 		ports := Ports{}
 
@@ -3295,9 +3295,9 @@ func (s *Server) logPorts() {
 func (s *Server) readyForListeners(dur time.Duration) bool {
 	end := time.Now().Add(dur)
 	for time.Now().Before(end) {
-		s.mu.Lock()
+		s.mu.RLock()
 		listeners := s.serviceListeners()
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		if len(listeners) == 0 {
 			return false
 		}
@@ -3345,8 +3345,8 @@ func (s *Server) serviceListeners() []net.Listener {
 
 // Returns true if in lame duck mode.
 func (s *Server) isLameDuckMode() bool {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.ldm
 }
 


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 35dd9bf1..cd3b4b20 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -3599,6 +3599,253 @@ func TestJetStreamClusterPeerRemovalAndStreamReassignmentWithoutSpace(t *testing
 	streamCurrent(2)
 }
 
+func TestJetStreamClusterPeerExclusionTag(t *testing.T) {
+	c := createJetStreamClusterWithTemplateAndModHook(t, jsClusterTempl, "C", 3,
+		func(serverName, clusterName, storeDir, conf string) string {
+			switch serverName {
+			case "S-1":
+				return fmt.Sprintf("%s\nserver_tags: [server:%s, intersect, %s]", conf, serverName, jsExcludePlacement)
+			case "S-2":
+				return fmt.Sprintf("%s\nserver_tags: [server:%s, intersect]", conf, serverName)
+			default:
+				return fmt.Sprintf("%s\nserver_tags: [server:%s]", conf, serverName)
+			}
+		})
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	for i, c := range []nats.StreamConfig{
+		{Replicas: 1, Placement: &nats.Placement{Tags: []string{"server:S-1"}}},
+		{Replicas: 2, Placement: &nats.Placement{Tags: []string{"intersect"}}},
+		{Replicas: 3}, // not enough server without !jetstream
+	} {
+		c.Name = fmt.Sprintf("TEST%d", i)
+		c.Subjects = []string{c.Name}
+		_, err := js.AddStream(&c)
+		require_Error(t, err)
+		require_Contains(t, err.Error(), "insufficient resources")
+	}
+
+	// Test update failure
+	cfg := &nats.StreamConfig{Name: "TEST", Subjects: []string{"foo"}, Replicas: 2}
+	_, err := js.AddStream(cfg)
+	require_NoError(t, err)
+
+	cfg.Replicas = 3
+	_, err = js.UpdateStream(cfg)
+	require_Error(t, err)
+	require_Contains(t, err.Error(), "insufficient resources")
+
+	// Test tag reload removing !jetstream tag, and allowing placement again
+
+	srv := c.serverByName("S-1")
+
+	v, err := srv.Varz(nil)
+	require_NoError(t, err)
+	require_True(t, v.Tags.Contains(jsExcludePlacement))
+	content, err := os.ReadFile(srv.configFile)
+	require_NoError(t, err)
+	newContent := strings.ReplaceAll(string(content), fmt.Sprintf(", %s]", jsExcludePlacement), "]")
+	changeCurrentConfigContentWithNewContent(t, srv.configFile, []byte(newContent))
+
+	ncSys := natsConnect(t, c.randomServer().ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+	sub, err := ncSys.SubscribeSync(fmt.Sprintf("$SYS.SERVER.%s.STATSZ", srv.ID()))
+	require_NoError(t, err)
+
+	require_NoError(t, srv.Reload())
+	v, err = srv.Varz(nil)
+	require_NoError(t, err)
+	require_True(t, !v.Tags.Contains(jsExcludePlacement))
+
+	m, err := sub.NextMsg(time.Second)
+	require_NoError(t, err)
+	require_True(t, strings.Contains(string(m.Data), `"tags":["server:s-1","intersect"]`))
+
+	cfg.Replicas = 3
+	_, err = js.UpdateStream(cfg)
+	require_NoError(t, err)
+}
+
+func TestJetStreamClusterPeerEvacuationAndStreamReassignment(t *testing.T) {
+	s := createJetStreamSuperClusterWithTemplateAndModHook(t, jsClusterTempl, 4, 2,
+		func(serverName, clusterName, storeDir, conf string) string {
+			return fmt.Sprintf("%s\nserver_tags: [cluster:%s, server:%s]", conf, clusterName, serverName)
+		})
+	defer s.shutdown()
+
+	c := s.clusterForName("C1")
+
+	// Client based API
+	srv := c.randomNonLeader()
+	nc, js := jsClientConnect(t, srv)
+	defer nc.Close()
+
+	test := func(r int, moveTags []string, targetCluster string, testMigrateTo bool) {
+		si, err := js.AddStream(&nats.StreamConfig{
+			Name:     "TEST",
+			Subjects: []string{"foo"},
+			Replicas: r,
+		})
+		require_NoError(t, err)
+		defer js.DeleteStream("TEST")
+		toMoveFrom := si.Cluster.Leader
+
+		_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "DUR", AckPolicy: nats.AckExplicitPolicy})
+		require_NoError(t, err)
+
+		sub, err := js.SubscribeSync("foo")
+		require_NoError(t, err)
+
+		for i := 0; i < 100; i++ {
+			_, err = js.Publish("foo", nil)
+			require_NoError(t, err)
+		}
+
+		sEmpty := c.serverByName(toMoveFrom)
+		jszBefore, err := sEmpty.Jsz(nil)
+		require_NoError(t, err)
+		require_True(t, jszBefore.Streams == 1)
+		require_True(t, jszBefore.Consumers >= 1)
+		require_True(t, jszBefore.Store != 0)
+
+		migrateToServer := _EMPTY_
+		if testMigrateTo {
+			// find an empty server
+			for _, s := range c.servers {
+				name := s.Name()
+				found := si.Cluster.Leader == name
+				if !found {
+					for _, r := range si.Cluster.Replicas {
+						if r.Name == name {
+							found = true
+							break
+						}
+					}
+				}
+				if !found {
+					migrateToServer = name
+					break
+				}
+			}
+			jszAfter, err := c.serverByName(migrateToServer).Jsz(nil)
+			require_NoError(t, err)
+			require_True(t, jszAfter.Streams == 0)
+
+			moveTags = append(moveTags, fmt.Sprintf("server:%s", migrateToServer))
+		}
+
+		ncsys, err := nats.Connect(srv.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+		require_NoError(t, err)
+		defer ncsys.Close()
+		/*
+			infoReq, err := json.Marshal(&JSApiMetaServerStreamInfoRequest{Server: toMoveFrom})
+			require_NoError(t, err)
+			rmsg, err := ncsys.Request(JSApiServerStreamInfo, infoReq, time.Second)
+			require_NoError(t, err)
+			var infoResp JSApiMetaServerStreamInfoResponse
+			require_NoError(t, json.Unmarshal(rmsg.Data, &infoResp))
+			require_True(t, infoResp.Error == nil)
+
+			found := false
+			for _, sCfg := range infoResp.Content["$G"] {
+				if sCfg.Name == "TEST" {
+					found = true
+					break
+				}
+			}
+			require_True(t, found)
+		*/
+		moveReq, err := json.Marshal(&JSApiMetaServerStreamMoveRequest{
+			Server: toMoveFrom, Account: "$G", Stream: "TEST", Tags: moveTags})
+		require_NoError(t, err)
+		rmsg, err := ncsys.Request(JSApiServerStreamMove, moveReq, 100*time.Second)
+		require_NoError(t, err)
+		var moveResp JSApiStreamUpdateResponse
+		require_NoError(t, json.Unmarshal(rmsg.Data, &moveResp))
+		require_True(t, moveResp.Error == nil)
+
+		// test draining
+		checkFor(t, 20*time.Second, 1000*time.Millisecond, func() error {
+			jszAfter, err := sEmpty.Jsz(nil)
+			if err != nil {
+				return fmt.Errorf("could not fetch JS info for server: %v", err)
+			}
+			if jszAfter.Streams != 0 {
+				return fmt.Errorf("empty server still has %d streams", jszAfter.Streams)
+			}
+			if jszAfter.Consumers != 0 {
+				return fmt.Errorf("empty server still has %d consumers", jszAfter.Consumers)
+			}
+			if jszAfter.Store != 0 {
+				return fmt.Errorf("empty server still has %d storage", jszAfter.Store)
+			}
+			return nil
+		})
+		// test move to particular server
+		if testMigrateTo {
+			toSrv := c.serverByName(migrateToServer)
+			checkFor(t, 20*time.Second, 1000*time.Millisecond, func() error {
+				jszAfter, err := toSrv.Jsz(nil)
+				if err != nil {
+					return fmt.Errorf("could not fetch JS info for server: %v", err)
+				}
+				if jszAfter.Streams != 1 {
+					return fmt.Errorf("server expected to have one stream, has %d", jszAfter.Streams)
+				}
+				return nil
+			})
+		}
+		// Now wait until the stream is now current.
+		checkFor(t, 20*time.Second, 100*time.Millisecond, func() error {
+			si, err := js.StreamInfo("TEST", nats.MaxWait(time.Second))
+			if err != nil {
+				return fmt.Errorf("could not fetch stream info: %v", err)
+			}
+			if si.Cluster.Leader == toMoveFrom {
+				return fmt.Errorf("peer not removed yet: %+v", toMoveFrom)
+			}
+			if len(si.Cluster.Replicas) != r-1 {
+				return fmt.Errorf("not yet downsized replica should be empty has: %d", len(si.Cluster.Replicas))
+			}
+			if si.Config.Replicas != r {
+				return fmt.Errorf("bad replica count %d", si.Config.Replicas)
+			}
+			if si.Cluster.Name != targetCluster {
+				return fmt.Errorf("stream expected in %s but found in %s", si.Cluster.Name, targetCluster)
+			}
+			sNew := s.serverByName(si.Cluster.Leader)
+			if jszNew, err := sNew.Jsz(nil); err != nil {
+				return err
+			} else if jszNew.Streams != 1 {
+				return fmt.Errorf("new leader has %d streams, not one", jszNew.Streams)
+			} else if jszNew.Store != jszBefore.Store {
+				return fmt.Errorf("new leader has %d storage, should have %d", jszNew.Store, jszBefore.Store)
+			}
+			return nil
+		})
+		// consume messages from ephemeral consumer
+		for i := 0; i < 100; i++ {
+			_, err := sub.NextMsg(time.Second)
+			require_NoError(t, err)
+		}
+	}
+
+	for i := 1; i <= 3; i++ {
+		t.Run(fmt.Sprintf("r%d", i), func(t *testing.T) {
+			test(i, nil, "C1", false)
+		})
+		t.Run(fmt.Sprintf("r%d-explicit", i), func(t *testing.T) {
+			test(i, nil, "C1", true)
+		})
+	}
+
+	t.Run("r3-cluster-move", func(t *testing.T) {
+		test(3, []string{"cluster:C2"}, "C2", false)
+	})
+}
+
 func TestJetStreamClusterPeerOffline(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R5S", 5)
 	defer c.shutdown()