Repository: nats-io/nats-server
Pull request #1881 (2021-02-04 00:00:40)

[added] jsz nats and http monitoring endpoint for jetstream
Signed-off-by: Matthias Hanel <mh@synadia.com>

@ripienaar format and functionality what you had in mind? Currently I page on account level

sample:
```json 
{
  "server_id": "NBPTR6WDBM3YG5W36KJ6ER2YZC3MVHSAMXIYEXKUSBX5R2X67NJ5EPCU",
  "now": "2021-02-03T18:35:39.107669-05:00",
  "config": {
    "max_memory": 10485760,
    "max_storage": 10485760,
    "store_dir": "/var/folders/9h/6g_c9l6n6bb8gp331d_9y0_w0000gn/T/srv_7500046161537"
  },
  "memory": 0,
  "storage": 66,
  "api": {
    "total": 5,
    "errors": 0
  },
  "total_streams": 2,
  "total_consumers": 2,
  "total_messages": 1,
  "total_message_bytes": 33,
  "meta_cluster": {
    "name": "cluster_name",
    "leader": "server_5500",
    "replicas": [
      {
        "name": "server_5500",
        "current": true,
        "active": 2563000
      }
    ]
  },
  "total_streams_non_replicated": 1,
  "total_streams_in_quorum": 1,
  "total_consumers_non_replicated": 1,
  "total_consumers_out_quorum": 1,
  "account_details": [
    {
      "name": "ACC",
      "id": "ACC",
      "memory": 0,
      "storage": 66,
      "api": {
        "total": 5,
        "errors": 0
      },
      "stream_detail": [
        {
          "name": "my-stream-replicated",
          "cluster": {
            "name": "cluster_name",
            "leader": "server_7500",
            "replicas": [
              {
                "name": "server_5500",
                "current": true,
                "active": 718000
              }
            ]
          },
          "config": {
            "name": "my-stream-replicated",
            "subjects": [
              "foo",
              "bar"
            ],
            "retention": "limits",
            "max_consumers": -1,
            "max_msgs": -1,
            "max_bytes": -1,
            "discard": "old",
            "max_age": 0,
            "max_msg_size": -1,
            "storage": "file",
            "num_replicas": 2,
            "duplicate_window": 120000000000
          },
          "state": {
            "messages": 1,
            "bytes": 33,
            "first_seq": 1,
            "first_ts": "2021-02-03T23:35:39.106619Z",
            "last_seq": 1,
            "last_ts": "2021-02-03T23:35:39.106619Z",
            "consumer_count": 1
          },
          "consumer_detail": [
            {
              "stream_name": "my-stream-replicated",
              "name": "my-consumer-replicated",
              "created": "2021-02-03T18:35:38.929459-05:00",
              "config": {
                "durable_name": "my-consumer-replicated",
                "deliver_policy": "all",
                "ack_policy": "explicit",
                "ack_wait": 30000000000,
                "max_deliver": -1,
                "replay_policy": "instant",
                "max_waiting": 512
              },
              "delivered": {
                "consumer_seq": 0,
                "stream_seq": 0
              },
              "ack_floor": {
                "consumer_seq": 0,
                "stream_seq": 0
              },
              "num_ack_pending": 0,
              "num_redelivered": 0,
              "num_waiting": 0,
              "num_pending": 0,
              "cluster": {
                "name": "cluster_name",
                "leader": "server_5500",
                "replicas": [
                  {
                    "name": "server_5500",
                    "current": false,
                    "active": 1612395339107725000
                  }
                ]
              }
            }
          ]
        },
        {
          "name": "my-stream-non-replicated",
          "cluster": {
            "name": "cluster_name",
            "leader": "server_7500"
          },
          "config": {
            "name": "my-stream-non-replicated",
            "subjects": [
              "baz"
            ],
            "retention": "limits",
            "max_consumers": -1,
            "max_msgs": -1,
            "max_bytes": -1,
            "discard": "old",
            "max_age": 0,
            "max_msg_size": -1,
            "storage": "file",
            "num_replicas": 1,
            "duplicate_window": 120000000000
          },
          "state": {
            "messages": 0,
            "bytes": 0,
            "first_seq": 0,
            "first_ts": "0001-01-01T00:00:00Z",
            "last_seq": 0,
            "last_ts": "0001-01-01T00:00:00Z",
            "consumer_count": 1
          },
          "consumer_detail": [
            {
              "stream_name": "my-stream-non-replicated",
              "name": "my-consumer-non-replicated",
              "created": "2021-02-03T18:35:39.104882-05:00",
              "config": {
                "durable_name": "my-consumer-non-replicated",
                "deliver_policy": "all",
                "ack_policy": "explicit",
                "ack_wait": 30000000000,
                "max_deliver": -1,
                "replay_policy": "instant",
                "max_waiting": 512
              },
              "delivered": {
                "consumer_seq": 0,
                "stream_seq": 0
              },
              "ack_floor": {
                "consumer_seq": 0,
                "stream_seq": 0
              },
              "num_ack_pending": 0,
              "num_redelivered": 0,
              "num_waiting": 0,
              "num_pending": 0,
              "cluster": {
                "name": "cluster_name",
                "leader": "server_7500"
              }
            }
          ]
        }
      ]
    },
    {
      "name": "BCC_TO_HAVE_ONE_EXTRA",
      "id": "BCC_TO_HAVE_ONE_EXTRA",
      "memory": 0,
      "storage": 0,
      "api": {
        "total": 0,
        "errors": 0
      }
    }
  ]
}
```


Patch:
diff --git a/server/client.go b/server/client.go
index 2bd9dae9..62517f7f 100644
--- a/server/client.go
+++ b/server/client.go
@@ -4662,9 +4662,12 @@ func (c *client) getClientInfo(detailed bool) *ClientInfo {
 		return nil
 	}
 	// Server name. Defaults to server ID if not set explicitly.
-	var sn string
-	if detailed && c.kind != LEAF {
-		sn = c.srv.Name()
+	var cn, sn string
+	if detailed {
+		if c.kind != LEAF {
+			sn = c.srv.Name()
+		}
+		cn = c.srv.cachedClusterName()
 	}
 
 	c.mu.Lock()
@@ -4685,6 +4688,7 @@ func (c *client) getClientInfo(detailed bool) *ClientInfo {
 		ci.Lang = c.opts.Lang
 		ci.Version = c.opts.Version
 		ci.Server = sn
+		ci.Cluster = cn
 		ci.Jwt = c.opts.JWT
 		ci.IssuerKey = issuerForClient(c)
 		ci.NameTag = c.nameTag
diff --git a/server/const.go b/server/const.go
index aa2a9802..cc8614aa 100644
--- a/server/const.go
+++ b/server/const.go
@@ -40,7 +40,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.2.0-beta.68"
+	VERSION = "2.2.0-beta.70"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index 08455b9a..dd137ce6 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -32,17 +32,17 @@ import (
 )
 
 type ConsumerInfo struct {
-	Stream         string         `json:"stream_name"`
-	Name           string         `json:"name"`
-	Created        time.Time      `json:"created"`
-	Config         ConsumerConfig `json:"config"`
-	Delivered      SequencePair   `json:"delivered"`
-	AckFloor       SequencePair   `json:"ack_floor"`
-	NumAckPending  int            `json:"num_ack_pending"`
-	NumRedelivered int            `json:"num_redelivered"`
-	NumWaiting     int            `json:"num_waiting"`
-	NumPending     uint64         `json:"num_pending"`
-	Cluster        *ClusterInfo   `json:"cluster,omitempty"`
+	Stream         string          `json:"stream_name"`
+	Name           string          `json:"name"`
+	Created        time.Time       `json:"created"`
+	Config         *ConsumerConfig `json:"config,omitempty"`
+	Delivered      SequencePair    `json:"delivered"`
+	AckFloor       SequencePair    `json:"ack_floor"`
+	NumAckPending  int             `json:"num_ack_pending"`
+	NumRedelivered int             `json:"num_redelivered"`
+	NumWaiting     int             `json:"num_waiting"`
+	NumPending     uint64          `json:"num_pending"`
+	Cluster        *ClusterInfo    `json:"cluster,omitempty"`
 }
 
 type ConsumerConfig struct {
@@ -895,6 +895,10 @@ func (o *Consumer) updateDeliveryInterest(localInterest bool) {
 }
 
 func (o *Consumer) deleteNotActive() {
+	// Need to check again if there is not an interest now that the timer fires.
+	if !o.hasNoLocalInterest() {
+		return
+	}
 	o.mu.RLock()
 	if o.mset == nil {
 		o.mu.RUnlock()
@@ -1320,11 +1324,12 @@ func (o *Consumer) Info() *ConsumerInfo {
 	o.mu.RLock()
 	defer o.mu.RUnlock()
 
+	cfg := o.config
 	info := &ConsumerInfo{
 		Stream:  o.stream,
 		Name:    o.name,
 		Created: o.created,
-		Config:  o.config,
+		Config:  &cfg,
 		Delivered: SequencePair{
 			Consumer: o.dseq - 1,
 			Stream:   o.sseq - 1,
@@ -2369,9 +2374,9 @@ func (o *Consumer) Active() bool {
 
 // hasNoLocalInterest return true if we have no local interest.
 func (o *Consumer) hasNoLocalInterest() bool {
-	o.mu.Lock()
+	o.mu.RLock()
 	rr := o.acc.sl.Match(o.config.DeliverSubject)
-	o.mu.Unlock()
+	o.mu.RUnlock()
 	return len(rr.psubs)+len(rr.qsubs) == 0
 }
 
diff --git a/server/events.go b/server/events.go
index 0ebafde4..8722c8b8 100644
--- a/server/events.go
+++ b/server/events.go
@@ -173,6 +173,7 @@ type ClientInfo struct {
 	Version   string        `json:"ver,omitempty"`
 	RTT       time.Duration `json:"rtt,omitempty"`
 	Server    string        `json:"server,omitempty"`
+	Cluster   string        `json:"cluster,omitempty"`
 	Stop      *time.Time    `json:"stop,omitempty"`
 	Jwt       string        `json:"jwt,omitempty"`
 	IssuerKey string        `json:"issuer_key,omitempty"`
@@ -263,7 +264,7 @@ RESET:
 	host := s.info.Host
 	servername := s.info.Name
 	seqp := &s.sys.seq
-	js := s.js != nil
+	js := s.info.JetStream
 	cluster := s.info.Cluster
 	if s.gateway.enabled {
 		cluster = s.getGatewayName()
@@ -677,6 +678,10 @@ func (s *Server) initEventTracking() {
 			optz := &AccountzEventOptions{}
 			s.zReq(reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Accountz(&optz.AccountzOptions) })
 		},
+		"JSZ": func(sub *subscription, _ *client, subject, reply string, msg []byte) {
+			optz := &JszEventOptions{}
+			s.zReq(reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) { return s.Jsz(&optz.JSzOptions) })
+		},
 	}
 	for name, req := range monSrvc {
 		subject = fmt.Sprintf(serverDirectReqSubj, s.info.ID, name)
@@ -730,6 +735,17 @@ func (s *Server) initEventTracking() {
 				}
 			})
 		},
+		"JSZ": func(sub *subscription, _ *client, subject, reply string, msg []byte) {
+			optz := &JszEventOptions{}
+			s.zReq(reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
+				if acc, err := extractAccount(subject); err != nil {
+					return nil, err
+				} else {
+					optz.Account = acc
+					return s.JszAccount(&optz.JSzOptions)
+				}
+			})
+		},
 		"INFO": func(sub *subscription, _ *client, subject, reply string, msg []byte) {
 			optz := &AccInfoEventOptions{}
 			s.zReq(reply, msg, &optz.EventFilterOptions, optz, func() (interface{}, error) {
@@ -826,6 +842,7 @@ func (s *Server) remoteServerShutdown(sub *subscription, _ *client, subject, rep
 		s.Debugf("Received remote server shutdown on bad subject %q", subject)
 		return
 	}
+
 	sid := toks[serverSubjectIndex]
 	su := s.sys.servers[sid]
 	if su != nil {
@@ -860,7 +877,9 @@ func (s *Server) processNewServer(ms *ServerInfo) {
 	// connect update to make sure they switch this account to interest only mode.
 	s.ensureGWsInterestOnlyForLeafNodes()
 	// Add to our nodeToName
-	s.nodeToName[string(getHash(ms.Name))] = ms.Name
+	node := string(getHash(ms.Name))
+	s.nodeToName.Store(node, ms.Name)
+	s.nodeToCluster.Store(node, ms.Cluster)
 }
 
 // If GW is enabled on this server and there are any leaf node connections,
@@ -1032,6 +1051,12 @@ type AccountzEventOptions struct {
 	EventFilterOptions
 }
 
+// In the context of system events, JszEventOptions are options passed to Jsz
+type JszEventOptions struct {
+	JSzOptions
+	EventFilterOptions
+}
+
 // returns true if the request does NOT apply to this server and can be ignored.
 // DO NOT hold the server lock when
 func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 971313dd..ff01d073 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -1832,7 +1832,7 @@ func (s *Server) jsMsgDeleteRequest(sub *subscription, c *client, subject, reply
 	}
 
 	if s.JetStreamIsClustered() {
-		s.jsClusteredMsgDeleteRequest(ci, stream, subject, reply, req.Seq, rmsg)
+		s.jsClusteredMsgDeleteRequest(ci, mset, stream, subject, reply, &req, rmsg)
 		return
 	}
 
@@ -1863,7 +1863,56 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, subject, reply st
 		return
 	}
 
+	stream := tokenAt(subject, 6)
+
 	var resp = JSApiMsgGetResponse{ApiResponse: ApiResponse{Type: JSApiMsgGetResponseType}}
+
+	// If we are in clustered mode we need to be the stream leader to proceed.
+	if s.JetStreamIsClustered() {
+		// Check to make sure the stream is assigned.
+		js, cc := s.getJetStreamCluster()
+		if js == nil || cc == nil {
+			return
+		}
+
+		if cc.meta != nil && cc.meta.GroupLeader() == _EMPTY_ {
+			resp.Error = jsClusterNotAvailErr
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+
+		js.mu.RLock()
+		isLeader, sa := cc.isLeader(), js.streamAssignment(acc.Name, stream)
+		js.mu.RUnlock()
+
+		if isLeader && sa == nil {
+			// We can't find the stream, so mimic what would be the errors below.
+			if !acc.JetStreamEnabled() {
+				resp.Error = jsNotEnabledErr
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+				return
+			}
+			// No stream present.
+			resp.Error = jsNotFoundError(ErrJetStreamStreamNotFound)
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		} else if sa == nil {
+			return
+		}
+
+		// Check to see if we are a member of the group and if the group has no leader.
+		if js.isGroupLeaderless(sa.Group) {
+			resp.Error = jsClusterNotAvailErr
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+
+		// We have the stream assigned and a leader, so only the stream leader should answer.
+		if !acc.JetStreamIsStreamLeader(stream) {
+			return
+		}
+	}
+
 	if !acc.JetStreamEnabled() {
 		resp.Error = jsNotEnabledErr
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -1881,7 +1930,6 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, subject, reply st
 		return
 	}
 
-	stream := tokenAt(subject, 6)
 	mset, err := acc.LookupStream(stream)
 	if err != nil {
 		resp.Error = jsNotFoundError(err)
@@ -1984,7 +2032,7 @@ func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, subject, rep
 	}
 
 	if s.JetStreamIsClustered() {
-		s.jsClusteredStreamPurgeRequest(ci, stream, subject, reply, rmsg)
+		s.jsClusteredStreamPurgeRequest(ci, mset, stream, subject, reply, rmsg)
 		return
 	}
 
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 3094ec3f..e45d8219 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -46,6 +46,12 @@ type jetStreamCluster struct {
 	consumerResults *subscription
 }
 
+// Used to guide placement of streams in clustered JetStream.
+type Placement struct {
+	Cluster string   `json:"cluster"`
+	Tags    []string `json:"tags,omitempty"`
+}
+
 // Define types of the entry.
 type entryOp uint8
 
@@ -113,6 +119,7 @@ type consumerAssignment struct {
 type streamPurge struct {
 	Client  *ClientInfo `json:"client,omitempty"`
 	Stream  string      `json:"stream"`
+	LastSeq uint64      `json:"last_seq"`
 	Subject string      `json:"subject"`
 	Reply   string      `json:"reply"`
 }
@@ -122,6 +129,7 @@ type streamMsgDelete struct {
 	Client  *ClientInfo `json:"client,omitempty"`
 	Stream  string      `json:"stream"`
 	Seq     uint64      `json:"seq"`
+	NoErase bool        `json:"no_erase,omitempty"`
 	Subject string      `json:"subject"`
 	Reply   string      `json:"reply"`
 }
@@ -1165,6 +1173,8 @@ func (js *jetStream) monitorStream(mset *Stream, sa *streamAssignment) {
 	// we replace with the restore chan.
 	restoreDoneCh := make(<-chan error)
 
+	isRecovering := true
+
 	for {
 		select {
 		case err := <-restoreDoneCh:
@@ -1261,6 +1271,7 @@ func (js *jetStream) monitorStream(mset *Stream, sa *streamAssignment) {
 		case ce := <-ach:
 			// No special processing needed for when we are caught up on restart.
 			if ce == nil {
+				isRecovering = false
 				continue
 			}
 			if mset == nil && isRestore {
@@ -1272,7 +1283,7 @@ func (js *jetStream) monitorStream(mset *Stream, sa *streamAssignment) {
 				}
 			}
 			// Apply our entries.
-			if hadSnapshot, err := js.applyStreamEntries(mset, ce); err == nil {
+			if hadSnapshot, err := js.applyStreamEntries(mset, ce, isRecovering); err == nil {
 				n.Applied(ce.Index)
 				if hadSnapshot {
 					snapout = false
@@ -1303,11 +1314,13 @@ func (js *jetStream) monitorStream(mset *Stream, sa *streamAssignment) {
 	}
 }
 
-func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry) (bool, error) {
+func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry, isRecovering bool) (bool, error) {
 	var didSnap bool
 	for _, e := range ce.Entries {
 		if e.Type == EntrySnapshot {
-			mset.processSnapshot(e.Data)
+			if !isRecovering {
+				mset.processSnapshot(e.Data)
+			}
 			didSnap = true
 		} else if e.Type == EntryRemovePeer {
 			js.mu.RLock()
@@ -1332,7 +1345,9 @@ func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry) (bool,
 					continue
 				}
 				if err := mset.processJetStreamMsg(subject, reply, hdr, msg, lseq, ts); err != nil {
-					js.srv.Debugf("Got error processing JetStream msg: %v", err)
+					if err != errLastSeqMismatch || !isRecovering {
+						js.srv.Debugf("Got error processing JetStream msg: %v", err)
+					}
 				}
 			case deleteMsgOp:
 				md, err := decodeMsgDelete(buf[1:])
@@ -1340,14 +1355,22 @@ func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry) (bool,
 					panic(err.Error())
 				}
 				s, cc := js.server(), js.cluster
-				removed, err := mset.EraseMsg(md.Seq)
-				if err != nil {
-					s.Warnf("JetStream cluster failed to delete msg %d from stream %q for account %q: %v", md.Seq, md.Stream, md.Client.Account, err)
+
+				var removed bool
+				if md.NoErase {
+					removed, err = mset.RemoveMsg(md.Seq)
+				} else {
+					removed, err = mset.EraseMsg(md.Seq)
+				}
+				if err != nil && !isRecovering {
+					s.Debugf("JetStream cluster failed to delete msg %d from stream %q for account %q: %v", md.Seq, md.Stream, md.Client.Account, err)
 				}
+
 				js.mu.RLock()
 				isLeader := cc.isStreamLeader(md.Client.Account, md.Stream)
 				js.mu.RUnlock()
-				if isLeader {
+
+				if isLeader && !isRecovering {
 					var resp = JSApiMsgDeleteResponse{ApiResponse: ApiResponse{Type: JSApiMsgDeleteResponseType}}
 					if err != nil {
 						resp.Error = jsError(err)
@@ -1365,15 +1388,26 @@ func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry) (bool,
 				if err != nil {
 					panic(err.Error())
 				}
+				// Ignore if we are recovering and we have already processed.
+				if isRecovering {
+					if mset.State().FirstSeq <= sp.LastSeq {
+						// Make sure all messages from the purge are gone.
+						mset.store.Compact(sp.LastSeq + 1)
+					}
+					continue
+				}
+
 				s := js.server()
 				purged, err := mset.Purge()
 				if err != nil {
 					s.Warnf("JetStream cluster failed to purge stream %q for account %q: %v", sp.Stream, sp.Client.Account, err)
 				}
+
 				js.mu.RLock()
 				isLeader := js.cluster.isStreamLeader(sp.Client.Account, sp.Stream)
 				js.mu.RUnlock()
-				if isLeader {
+
+				if isLeader && !isRecovering {
 					var resp = JSApiStreamPurgeResponse{ApiResponse: ApiResponse{Type: JSApiStreamPurgeResponseType}}
 					if err != nil {
 						resp.Error = jsError(err)
@@ -1832,11 +1866,7 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 	if err != nil {
 		resp.Error = jsNotFoundError(err)
 	} else if mset != nil {
-		if mset.Config().internal {
-			err = errors.New("not allowed to delete internal stream")
-		} else {
-			err = mset.stop(true, wasLeader)
-		}
+		err = mset.stop(true, wasLeader)
 	}
 
 	if sa.Group.node != nil {
@@ -2076,9 +2106,7 @@ func (js *jetStream) processClusterDeleteConsumer(ca *consumerAssignment, isMemb
 	if err != nil {
 		resp.Error = jsNotFoundError(err)
 	} else if mset != nil {
-		if mset.Config().internal {
-			err = errors.New("not allowed to delete internal consumer")
-		} else if o := mset.LookupConsumer(ca.Name); o != nil {
+		if o := mset.LookupConsumer(ca.Name); o != nil {
 			err = o.stop(true, true, wasLeader)
 		} else {
 			resp.Error = jsNoConsumerErr
@@ -2564,40 +2592,61 @@ func (js *jetStream) processLeaderChange(isLeader bool) {
 // Lock should be held.
 func (cc *jetStreamCluster) remapStreamAssignment(sa *streamAssignment, removePeer string) bool {
 	// Need to select a replacement peer
+	s, now, cluster := cc.s, time.Now(), sa.Client.Cluster
+	if sa.Config.Placement != nil && sa.Config.Placement.Cluster != _EMPTY_ {
+		cluster = sa.Config.Placement.Cluster
+	}
 	for _, p := range cc.meta.Peers() {
-		if !sa.Group.isMember(p.ID) {
-			for i, peer := range sa.Group.Peers {
-				if peer == removePeer {
-					sa.Group.Peers[i] = p.ID
-					break
-				}
+		// Make sure they are active and current and not already part of our group.
+		current, lastSeen := p.Current, now.Sub(p.Last)
+		if !current || lastSeen > lostQuorumInterval || sa.Group.isMember(p.ID) {
+			continue
+		}
+		// Make sure the correct cluster.
+		if s.clusterNameForNode(p.ID) != cluster {
+			continue
+		}
+		// If we are here we have our candidate replacement, swap out the old one.
+		for i, peer := range sa.Group.Peers {
+			if peer == removePeer {
+				sa.Group.Peers[i] = p.ID
+				// Don't influence preferred leader.
+				sa.Group.Preferred = _EMPTY_
+				return true
 			}
-			// Don't influence preferred leader.
-			sa.Group.Preferred = _EMPTY_
-			return true
 		}
 	}
-	return true
+	return false
 }
 
 // selectPeerGroup will select a group of peers to start a raft group.
 // TODO(dlc) - For now randomly select. Can be way smarter.
-func (cc *jetStreamCluster) selectPeerGroup(r int) []string {
+func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string) []string {
 	var nodes []string
 	peers := cc.meta.Peers()
-	// Make sure they are active
-	s := cc.s
-	ourID := cc.meta.ID()
+	s, ourID := cc.s, cc.meta.ID()
+
+	now := time.Now()
 	for _, p := range peers {
-		// FIXME(dlc) - cluster scoped.
-		if p.ID == ourID || s.getRouteByHash([]byte(p.ID)) != nil {
-			nodes = append(nodes, p.ID)
+		// Make sure they are active and current.
+		current, lastSeen := p.Current, now.Sub(p.Last)
+		if current && lastSeen > lostQuorumInterval {
+			current = false
+		}
+		if p.ID == ourID || current {
+			if cluster != _EMPTY_ {
+				if s.clusterNameForNode(p.ID) == cluster {
+					nodes = append(nodes, p.ID)
+				}
+			} else {
+				nodes = append(nodes, p.ID)
+			}
 		}
 	}
 	if len(nodes) < r {
 		return nil
 	}
-	// Don't depend on range.
+	// Don't depend on range to randomize.
 	rand.Shuffle(len(nodes), func(i, j int) { nodes[i], nodes[j] = nodes[j], nodes[i] })
 	return nodes[:r]
 }
@@ -2622,15 +2671,18 @@ func groupName(prefix string, peers []string, storage StorageType) string {
 
 // createGroupForStream will create a group for assignment for the stream.
 // Lock should be held.
-func (cc *jetStreamCluster) createGroupForStream(cfg *StreamConfig) *raftGroup {
+func (cc *jetStreamCluster) createGroupForStream(ci *ClientInfo, cfg *StreamConfig) *raftGroup {
 	replicas := cfg.Replicas
 	if replicas == 0 {
 		replicas = 1
 	}
-
+	cluster := ci.Cluster
+	if cfg.Placement != nil && cfg.Placement.Cluster != _EMPTY_ {
+		cluster = cfg.Placement.Cluster
+	}
 	// Need to create a group here.
 	// TODO(dlc) - Can be way smarter here.
-	peers := cc.selectPeerGroup(replicas)
+	peers := cc.selectPeerGroup(replicas, cluster)
 	if len(peers) == 0 {
 		return nil
 	}
@@ -2682,7 +2734,7 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, subject, reply string,
 	}
 
 	// Raft group selection and placement.
-	rg := cc.createGroupForStream(cfg)
+	rg := cc.createGroupForStream(ci, cfg)
 	if rg == nil {
 		resp.Error = jsInsufficientErr
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
@@ -2724,7 +2776,7 @@ func (s *Server) jsClusteredStreamDeleteRequest(ci *ClientInfo, stream, subject,
 	cc.meta.Propose(encodeDeleteStreamAssignment(sa))
 }
 
-func (s *Server) jsClusteredStreamPurgeRequest(ci *ClientInfo, stream, subject, reply string, rmsg []byte) {
+func (s *Server) jsClusteredStreamPurgeRequest(ci *ClientInfo, mset *Stream, stream, subject, reply string, rmsg []byte) {
 	js, cc := s.getJetStreamCluster()
 	if js == nil || cc == nil {
 		return
@@ -2734,7 +2786,7 @@ func (s *Server) jsClusteredStreamPurgeRequest(ci *ClientInfo, stream, subject,
 	defer js.mu.Unlock()
 
 	sa := js.streamAssignment(ci.Account, stream)
-	if sa == nil || sa.Group == nil || sa.Group.node == nil {
+	if sa == nil {
 		resp := JSApiStreamPurgeResponse{ApiResponse: ApiResponse{Type: JSApiStreamPurgeResponseType}}
 		acc, err := s.LookupAccount(ci.Account)
 		if err != nil {
@@ -2746,9 +2798,20 @@ func (s *Server) jsClusteredStreamPurgeRequest(ci *ClientInfo, stream, subject,
 		return
 	}
 
-	n := sa.Group.node
-	sp := &streamPurge{Stream: stream, Subject: subject, Reply: reply, Client: ci}
-	n.Propose(encodeStreamPurge(sp))
+	if n := sa.Group.node; n != nil {
+		sp := &streamPurge{Stream: stream, LastSeq: mset.State().LastSeq, Subject: subject, Reply: reply, Client: ci}
+		n.Propose(encodeStreamPurge(sp))
+	} else if mset != nil {
+		var resp = JSApiStreamPurgeResponse{ApiResponse: ApiResponse{Type: JSApiStreamPurgeResponseType}}
+		purged, err := mset.Purge()
+		if err != nil {
+			resp.Error = jsError(err)
+		} else {
+			resp.Purged = purged
+			resp.Success = true
+		}
+		s.sendAPIResponse(ci, mset.account(), subject, reply, string(rmsg), s.jsonResponse(resp))
+	}
 }
 
 func (s *Server) jsClusteredStreamRestoreRequest(ci *ClientInfo, acc *Account, req *JSApiStreamRestoreRequest, stream, subject, reply string, rmsg []byte) {
@@ -2770,7 +2833,7 @@ func (s *Server) jsClusteredStreamRestoreRequest(ci *ClientInfo, acc *Account, r
 	}
 
 	// Raft group selection and placement.
-	rg := cc.createGroupForStream(cfg)
+	rg := cc.createGroupForStream(ci, cfg)
 	if rg == nil {
 		resp.Error = jsInsufficientErr
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
@@ -3067,7 +3130,7 @@ func decodeMsgDelete(buf []byte) (*streamMsgDelete, error) {
 	return &md, err
 }
 
-func (s *Server) jsClusteredMsgDeleteRequest(ci *ClientInfo, stream, subject, reply string, seq uint64, rmsg []byte) {
+func (s *Server) jsClusteredMsgDeleteRequest(ci *ClientInfo, mset *Stream, stream, subject, reply string, req *JSApiMsgDeleteRequest, rmsg []byte) {
 	js, cc := s.getJetStreamCluster()
 	if js == nil || cc == nil {
 		return
@@ -3077,13 +3140,32 @@ func (s *Server) jsClusteredMsgDeleteRequest(ci *ClientInfo, stream, subject, re
 	defer js.mu.Unlock()
 
 	sa := js.streamAssignment(ci.Account, stream)
-	if sa == nil || sa.Group == nil || sa.Group.node == nil {
-		// TODO(dlc) - Should respond? Log?
+	if sa == nil {
+		s.Debugf("Message delete failed, could not locate stream '%s > %s'", ci.Account, stream)
 		return
 	}
-	n := sa.Group.node
-	md := &streamMsgDelete{Seq: seq, Stream: stream, Subject: subject, Reply: reply, Client: ci}
-	n.Propose(encodeMsgDelete(md))
+	// Check for single replica items.
+	if n := sa.Group.node; n != nil {
+		md := &streamMsgDelete{Seq: req.Seq, NoErase: req.NoErase, Stream: stream, Subject: subject, Reply: reply, Client: ci}
+		n.Propose(encodeMsgDelete(md))
+	} else if mset != nil {
+		var err error
+		var removed bool
+		if req.NoErase {
+			removed, err = mset.RemoveMsg(req.Seq)
+		} else {
+			removed, err = mset.EraseMsg(req.Seq)
+		}
+		var resp = JSApiMsgDeleteResponse{ApiResponse: ApiResponse{Type: JSApiMsgDeleteResponseType}}
+		if err != nil {
+			resp.Error = jsError(err)
+		} else if !removed {
+			resp.Error = &ApiError{Code: 400, Description: fmt.Sprintf("sequence [%d] not found", req.Seq)}
+		} else {
+			resp.Success = true
+		}
+		s.sendAPIResponse(ci, mset.account(), subject, reply, string(rmsg), s.jsonResponse(resp))
+	}
 }
 
 func encodeAddStreamAssignment(sa *streamAssignment) []byte {
@@ -3170,7 +3252,7 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, subject, reply strin
 		}
 	}
 
-	ca := &consumerAssignment{Group: rg, Stream: stream, Name: oname, Config: cfg, Subject: subject, Reply: reply, Client: ci, Created: time.Now()}
+	ca := &consumerAssignment{Group: rg, Stream: stream, Name: oname, Config: cfg, Subject: subject, Reply: reply, Client: ci, Created: time.Now().UTC()}
 	cc.meta.Propose(encodeAddConsumerAssignment(ca))
 }
 
@@ -3494,8 +3576,27 @@ func (mset *Stream) processSnapshot(buf []byte) {
 
 	js := s.getJetStream()
 
+	var sub *subscription
+	var err error
+
+	const activityInterval = 5 * time.Second
+	notActive := time.NewTimer(activityInterval)
+	defer notActive.Stop()
+
+	defer func() {
+		if sub != nil {
+			s.sysUnsubscribe(sub)
+		}
+	}()
+
 RETRY:
 
+	// If we have a sub clear that here.
+	if sub != nil {
+		s.sysUnsubscribe(sub)
+		sub = nil
+	}
+
 	// Grab sync request again on failures.
 	if sreq == nil {
 		mset.mu.Lock()
@@ -3507,11 +3608,11 @@ RETRY:
 		}
 	}
 
-	msgsC := make(chan []byte, 1024)
+	msgsC := make(chan []byte, 8*1024)
 
 	// Send our catchup request here.
 	reply := syncReplySubject()
-	sub, err := s.sysSubscribe(reply, func(_ *subscription, _ *client, _, reply string, msg []byte) {
+	sub, err = s.sysSubscribe(reply, func(_ *subscription, _ *client, _, reply string, msg []byte) {
 		// Make copies - https://github.com/go101/go101/wiki
 		// TODO(dlc) - Since we are using a buffer from the inbound client/route.
 		if len(msg) > 0 {
@@ -3525,7 +3626,6 @@ RETRY:
 	if err != nil {
 		return
 	}
-	defer s.sysUnsubscribe(sub)
 
 	b, _ := json.Marshal(sreq)
 	s.sendInternalMsgLocked(subject, reply, nil, b)
@@ -3534,10 +3634,6 @@ RETRY:
 	last := sreq.LastSeq
 	sreq = nil
 
-	const activityInterval = 5 * time.Second
-	notActive := time.NewTimer(activityInterval)
-	defer notActive.Stop()
-
 	// Run our own select loop here.
 	for qch, lch := n.QuitC(), n.LeadChangeC(); ; {
 		select {
@@ -3547,7 +3643,6 @@ RETRY:
 			if len(msg) == 0 {
 				goto RETRY
 			}
-
 			if lseq, err := mset.processCatchupMsg(msg); err == nil {
 				if lseq >= last {
 					return
@@ -3557,6 +3652,7 @@ RETRY:
 			}
 		case <-notActive.C:
 			s.Warnf("Catchup for stream '%s > %s' stalled", mset.account(), mset.Name())
+			notActive.Reset(activityInterval)
 			goto RETRY
 		case <-s.quitCh:
 			return
diff --git a/server/monitor.go b/server/monitor.go
index fc17201b..6f8bf72c 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -2211,3 +2211,262 @@ func (s *Server) accountInfo(accName string) (*AccountInfo, error) {
 		responses,
 	}, nil
 }
+
+// LeafzOptions are options passed to Jsz
+type JSzOptions struct {
+	Account  string `json:"account,omitempty"`
+	Accounts bool   `json:"accounts,omitempty"`
+	Streams  bool   `json:"streams,omitempty"`
+	Consumer bool   `json:"consumer,omitempty"`
+	Config   bool   `json:"config,omitempty"`
+	Offset   int    `json:"offset,omitempty"`
+	Limit    int    `json:"limit,omitempty"`
+}
+
+type StreamDetail struct {
+	Name     string          `json:"name"`
+	Cluster  *ClusterInfo    `json:"cluster,omitempty"`
+	Config   *StreamConfig   `json:"config,omitempty"`
+	State    StreamState     `json:"state,omitempty"`
+	Consumer []*ConsumerInfo `json:"consumer_detail,omitempty"`
+}
+
+type AccountDetail struct {
+	Name string `json:"name"`
+	Id   string `json:"id"`
+	JetStreamStats
+	Streams []StreamDetail `json:"stream_detail,omitempty"`
+}
+
+// LeafInfo has detailed information on each remote leafnode connection.
+type JSInfo struct {
+	ID       string          `json:"server_id"`
+	Now      time.Time       `json:"now"`
+	Disabled bool            `json:"disabled,omitempty"`
+	Config   JetStreamConfig `json:"config,omitempty"`
+	JetStreamStats
+	StreamCnt    int          `json:"total_streams,omitempty"`
+	ConsumerCnt  int          `json:"total_consumers,omitempty"`
+	MessageCnt   uint64       `json:"total_messages,omitempty"`
+	MessageBytes uint64       `json:"total_message_bytes,omitempty"`
+	Meta         *ClusterInfo `json:"meta_cluster,omitempty"`
+	// aggregate raft info
+	AccountDetails []*AccountDetail `json:"account_details,omitempty"`
+}
+
+func (s *Server) accountDetail(jsa *jsAccount, optStreams, optConsumers, optCfg bool) *AccountDetail {
+	jsa.mu.RLock()
+	defer jsa.mu.RUnlock()
+	acc := jsa.account
+	name := acc.GetName()
+	id := name
+	if acc.nameTag != "" {
+		name = acc.nameTag
+	}
+	detail := AccountDetail{
+		Name: name,
+		Id:   id,
+		JetStreamStats: JetStreamStats{
+			Memory: uint64(jsa.memTotal),
+			Store:  uint64(jsa.storeTotal),
+			API: JetStreamAPIStats{
+				Total:  jsa.apiTotal,
+				Errors: jsa.apiErrors,
+			},
+		},
+		Streams: make([]StreamDetail, 0, len(jsa.streams)),
+	}
+	if optStreams {
+		for _, stream := range jsa.streams {
+			ci := s.js.clusterInfo(stream.raftGroup())
+			var cfg *StreamConfig
+			if optCfg {
+				c := stream.Config()
+				cfg = &c
+			}
+			sdet := StreamDetail{
+				Name:    stream.Name(),
+				State:   stream.State(),
+				Cluster: ci,
+				Config:  cfg}
+			if optConsumers {
+				for _, consumer := range stream.consumers {
+					cInfo := consumer.Info()
+					if !optCfg {
+						cInfo.Config = nil
+					}
+					sdet.Consumer = append(sdet.Consumer, cInfo)
+				}
+			}
+			detail.Streams = append(detail.Streams, sdet)
+		}
+	}
+	return &detail
+}
+
+func (s *Server) JszAccount(opts *JSzOptions) (*AccountDetail, error) {
+	if s.js == nil {
+		return nil, fmt.Errorf("jetstream not enabled")
+	}
+	acc := opts.Account
+	account, ok := s.accounts.Load(acc)
+	if !ok {
+		return nil, fmt.Errorf("account %q not found", acc)
+	}
+	s.js.mu.RLock()
+	jsa, ok := s.js.accounts[account.(*Account)]
+	s.js.mu.RUnlock()
+	if !ok {
+		return nil, fmt.Errorf("account %q not jetstream enabled", acc)
+	}
+	return s.accountDetail(jsa, opts.Streams, opts.Consumer, opts.Config), nil
+}
+
+// Leafz returns a Leafz structure containing information about leafnodes.
+func (s *Server) Jsz(opts *JSzOptions) (*JSInfo, error) {
+	// set option defaults
+	if opts == nil {
+		opts = &JSzOptions{}
+	}
+	if opts.Limit == 0 {
+		opts.Limit = 1024
+	}
+	if opts.Consumer {
+		opts.Streams = true
+	}
+	if opts.Streams {
+		opts.Accounts = true
+	}
+	// helper to get cluster info from node via dummy group
+	toClusterInfo := func(node RaftNode) *ClusterInfo {
+		peers := node.Peers()
+		peerList := make([]string, len(peers))
+		for i, p := range node.Peers() {
+			peerList[i] = p.ID
+		}
+		group := &raftGroup{
+			Name:  "",
+			Peers: peerList,
+			node:  node,
+		}
+		return s.js.clusterInfo(group)
+	}
+	jsi := &JSInfo{
+		ID:  s.ID(),
+		Now: time.Now(),
+	}
+	if !s.JetStreamEnabled() {
+		jsi.Disabled = true
+		return jsi, nil
+	}
+	accounts := []*jsAccount{}
+	s.js.mu.RLock()
+	jsi.Config = s.js.config
+	for _, info := range s.js.accounts {
+		accounts = append(accounts, info)
+	}
+	s.js.mu.RUnlock()
+
+	jsi.Meta = toClusterInfo(s.js.getMetaGroup())
+	filterIdx := -1
+	for i, jsa := range accounts {
+		jsa.mu.RLock()
+		if jsa.acc().GetName() == opts.Account {
+			filterIdx = i
+		}
+		jsi.StreamCnt += len(jsa.streams)
+		jsi.Memory += uint64(jsa.memTotal)
+		jsi.Store += uint64(jsa.storeTotal)
+		jsi.API.Total += jsa.apiTotal
+		jsi.API.Errors += jsa.apiErrors
+		for _, stream := range jsa.streams {
+			streamState := stream.State()
+			jsi.MessageCnt += streamState.Msgs
+			jsi.MessageBytes += streamState.Bytes
+			jsi.ConsumerCnt += streamState.Consumers
+		}
+		jsa.mu.RUnlock()
+	}
+	// filter logic
+	if filterIdx != -1 {
+		accounts = []*jsAccount{accounts[filterIdx]}
+	} else if opts.Accounts {
+		if opts.Offset != 0 {
+			sort.Slice(accounts, func(i, j int) bool {
+				return strings.Compare(accounts[i].acc().Name, accounts[j].acc().Name) < 0
+			})
+			if opts.Offset > len(accounts) {
+				accounts = []*jsAccount{}
+			} else {
+				accounts = accounts[opts.Offset:]
+			}
+		}
+		if opts.Limit != 0 {
+			if opts.Limit < len(accounts) {
+				accounts = accounts[:opts.Limit]
+			}
+		}
+	} else {
+		accounts = []*jsAccount{}
+	}
+	if len(accounts) > 0 {
+		jsi.AccountDetails = make([]*AccountDetail, 0, len(accounts))
+	}
+	// if wanted, obtain accounts/streams/consumer
+	for _, jsa := range accounts {
+		detail := s.accountDetail(jsa, opts.Streams, opts.Consumer, opts.Config)
+		jsi.AccountDetails = append(jsi.AccountDetails, detail)
+	}
+	return jsi, nil
+}
+
+// HandleJSz process HTTP requests for jetstream information.
+func (s *Server) HandleJsz(w http.ResponseWriter, r *http.Request) {
+	s.mu.Lock()
+	s.httpReqStats[LeafzPath]++
+	s.mu.Unlock()
+	accounts, err := decodeBool(w, r, "accounts")
+	if err != nil {
+		return
+	}
+	streams, err := decodeBool(w, r, "streams")
+	if err != nil {
+		return
+	}
+	consumers, err := decodeBool(w, r, "consumers")
+	if err != nil {
+		return
+	}
+	config, err := decodeBool(w, r, "config")
+	if err != nil {
+		return
+	}
+	offset, err := decodeInt(w, r, "offset")
+	if err != nil {
+		return
+	}
+	limit, err := decodeInt(w, r, "limit")
+	if err != nil {
+		return
+	}
+	l, err := s.Jsz(&JSzOptions{
+		r.URL.Query().Get("acc"),
+		accounts,
+		streams,
+		consumers,
+		config,
+		offset,
+		limit})
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		w.Write([]byte(err.Error()))
+		return
+	}
+	b, err := json.MarshalIndent(l, "", "  ")
+	if err != nil {
+		s.Errorf("Error marshaling response to /leafz request: %v", err)
+	}
+
+	// Handle response
+	ResponseHandler(w, r, b)
+}
diff --git a/server/raft.go b/server/raft.go
index 344856d8..d6054443 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"math/rand"
+	"net"
 	"os"
 	"path"
 	"sync"
@@ -228,11 +229,35 @@ func (s *Server) bootstrapRaftNode(cfg *RaftConfig, knownPeers []string, allPeer
 	expected := len(knownPeers)
 	// We need to adjust this is all peers are not known.
 	if !allPeersKnown {
+		s.Debugf("Determining expected peer size for JetStream metacontroller")
 		if expected < 2 {
 			expected = 2
 		}
-		if ncr := s.configuredRoutes(); expected < ncr {
-			expected = ncr
+		opts := s.getOpts()
+		nrs := len(opts.Routes)
+
+		cn := s.ClusterName()
+		ngwps := 0
+		for _, gw := range opts.Gateway.Gateways {
+			// Ignore our own cluster if specified.
+			if gw.Name == cn {
+				continue
+			}
+			for _, u := range gw.URLs {
+				host := u.Hostname()
+				// If this is an IP just add one.
+				if net.ParseIP(host) != nil {
+					ngwps++
+				} else {
+					addrs, _ := net.LookupHost(host)
+					ngwps += len(addrs)
+				}
+			}
+		}
+
+		if expected < nrs+ngwps {
+			expected = nrs + ngwps
+			s.Debugf("Adjusting expected peer set size to %d with %d known", expected, len(knownPeers))
 		}
 	}
 
@@ -261,6 +286,7 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 	if ps == nil || ps.clusterSize < 2 {
 		return nil, errors.New("raft: cluster too small")
 	}
+
 	n := &raft{
 		created:  time.Now(),
 		id:       hash[:idLen],
@@ -353,10 +379,18 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 
 // Maps node names back to server names.
 func (s *Server) serverNameForNode(node string) string {
-	s.mu.Lock()
-	sn := s.nodeToName[node]
-	s.mu.Unlock()
-	return sn
+	if sn, ok := s.nodeToName.Load(node); ok {
+		return sn.(string)
+	}
+	return _EMPTY_
+}
+
+// Maps node names back to cluster names.
+func (s *Server) clusterNameForNode(node string) string {
+	if cn, ok := s.nodeToCluster.Load(node); ok {
+		return cn.(string)
+	}
+	return _EMPTY_
 }
 
 // Server will track all raft nodes.
@@ -872,7 +906,7 @@ func (n *raft) shutdown(shouldDelete bool) {
 	}
 }
 
-func (n *raft) newInbox(cn string) string {
+func (n *raft) newInbox() string {
 	var b [replySuffixLen]byte
 	rn := rand.Int63()
 	for i, l := 0, rn; i < len(b); i++ {
@@ -883,8 +917,8 @@ func (n *raft) newInbox(cn string) string {
 }
 
 const (
-	raftVoteSubj       = "$NRG.V.%s.%s"
-	raftAppendSubj     = "$NRG.E.%s.%s"
+	raftVoteSubj       = "$NRG.V.%s"
+	raftAppendSubj     = "$NRG.AE.%s"
 	raftPropSubj       = "$NRG.P.%s"
 	raftRemovePeerSubj = "$NRG.RP.%s"
 	raftReplySubj      = "$NRG.R.%s"
@@ -897,9 +931,8 @@ func (n *raft) subscribe(subject string, cb msgHandler) (*subscription, error) {
 }
 
 func (n *raft) createInternalSubs() error {
-	cn := n.s.ClusterName()
-	n.vsubj, n.vreply = fmt.Sprintf(raftVoteSubj, cn, n.group), n.newInbox(cn)
-	n.asubj, n.areply = fmt.Sprintf(raftAppendSubj, cn, n.group), n.newInbox(cn)
+	n.vsubj, n.vreply = fmt.Sprintf(raftVoteSubj, n.group), n.newInbox()
+	n.asubj, n.areply = fmt.Sprintf(raftAppendSubj, n.group), n.newInbox()
 	n.psubj = fmt.Sprintf(raftPropSubj, n.group)
 	n.rpsubj = fmt.Sprintf(raftRemovePeerSubj, n.group)
 
@@ -1732,7 +1765,7 @@ func (n *raft) createCatchup(ae *appendEntry) string {
 		pterm:  n.pterm,
 		pindex: n.pindex,
 	}
-	inbox := n.newInbox(n.s.ClusterName())
+	inbox := n.newInbox()
 	sub, _ := n.subscribe(inbox, n.handleAppendEntry)
 	n.catchup.sub = sub
 	return inbox
@@ -1970,7 +2003,7 @@ func (n *raft) storeToWAL(ae *appendEntry) error {
 
 	// Sanity checking for now.
 	if ae.pindex != seq-1 {
-		panic(fmt.Sprintf("[%s] Placed an entry at the wrong index, ae is %+v, index is %d\n\n", n.s, ae, seq))
+		panic(fmt.Sprintf("[%s-%s] Placed an entry at the wrong index, ae is %+v, index is %d\n\n", n.s, n.group, ae, seq))
 	}
 
 	n.pterm = ae.term
diff --git a/server/route.go b/server/route.go
index cf36b862..92cda08d 100644
--- a/server/route.go
+++ b/server/route.go
@@ -1416,7 +1416,8 @@ func (s *Server) addRoute(c *client, info *Info) (bool, bool) {
 	if !exists {
 		s.routes[c.cid] = c
 		s.remotes[id] = c
-		s.nodeToName[c.route.hash] = c.route.remoteName
+		s.nodeToName.Store(c.route.hash, c.route.remoteName)
+		s.nodeToCluster.Store(c.route.hash, s.info.Cluster)
 		c.mu.Lock()
 		c.route.connectURLs = info.ClientConnectURLs
 		c.route.wsConnURLs = info.WSConnectURLs
diff --git a/server/server.go b/server/server.go
index 05a0cf3b..144b5b2d 100644
--- a/server/server.go
+++ b/server/server.go
@@ -238,9 +238,9 @@ type Server struct {
 	rnMu      sync.RWMutex
 	raftNodes map[string]RaftNode
 
-	// For mapping from a node name back to a server name.
-	// Normal server lock here.
-	nodeToName map[string]string
+	// For mapping from a node name back to a server name and cluster.
+	nodeToName    sync.Map
+	nodeToCluster sync.Map
 }
 
 // Make sure all are 64bits for atomic use
@@ -324,7 +324,6 @@ func NewServer(opts *Options) (*Server, error) {
 		httpBasePath: httpBasePath,
 		eventIds:     nuid.New(),
 		routesToSelf: make(map[string]struct{}),
-		nodeToName:   make(map[string]string),
 	}
 
 	// Trusted root operator keys.
@@ -332,11 +331,20 @@ func NewServer(opts *Options) (*Server, error) {
 		return nil, fmt.Errorf("Error processing trusted operator keys")
 	}
 
+	if opts.Cluster.Name != _EMPTY_ {
+		// Also place into mapping cn with cnMu lock.
+		s.cnMu.Lock()
+		s.cn = opts.Cluster.Name
+		s.cnMu.Unlock()
+	}
+
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	// Place ourselves.
-	s.nodeToName[string(getHash(serverName))] = serverName
+	// Place ourselves in some lookup maps.
+	ourId := string(getHash(serverName))
+	s.nodeToName.Store(ourId, serverName)
+	s.nodeToCluster.Store(ourId, opts.Cluster.Name)
 
 	s.routeResolver = opts.Cluster.resolver
 	if s.routeResolver == nil {
@@ -789,10 +797,11 @@ func (s *Server) activePeers() (peers []string) {
 	if s.sys == nil {
 		return nil
 	}
-	// FIXME(dlc) - When this spans supercluster need to adjust below.
-	for _, r := range s.routes {
-		peers = append(peers, r.route.hash)
-	}
+	s.nodeToName.Range(func(k, v interface{}) bool {
+		peers = append(peers, k.(string))
+		return true
+	})
+
 	return peers
 }
 
@@ -1994,6 +2003,7 @@ const (
 	SubszPath    = "/subsz"
 	StackszPath  = "/stacksz"
 	AccountzPath = "/accountz"
+	JszPath      = "/jsz"
 )
 
 func (s *Server) basePath(p string) string {
@@ -2073,7 +2083,8 @@ func (s *Server) startMonitoring(secure bool) error {
 	mux.HandleFunc(s.basePath(StackszPath), s.HandleStacksz)
 	// Accountz
 	mux.HandleFunc(s.basePath(AccountzPath), s.HandleAccountz)
-
+	// Jsz
+	mux.HandleFunc(s.basePath(JszPath), s.HandleJsz)
 	// Do not set a WriteTimeout because it could cause cURL/browser
 	// to return empty response or unable to display page if the
 	// server needs more time to build the response.
diff --git a/server/stream.go b/server/stream.go
index 4e12bbeb..5544c153 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -51,6 +51,7 @@ type StreamConfig struct {
 	NoAck        bool            `json:"no_ack,omitempty"`
 	Template     string          `json:"template_owner,omitempty"`
 	Duplicates   time.Duration   `json:"duplicate_window,omitempty"`
+	Placement    *Placement      `json:"placement,omitempty"`
 
 	// These are non public configuration options.
 	// If you add new options, check fileStreamInfoJSON in order for them to


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index 0b9874ff..a63852f1 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1232,12 +1232,15 @@ func TestAccountReqMonitoring(t *testing.T) {
 	defer s.Shutdown()
 	sacc, sakp := createAccount(s)
 	s.setSystemAccount(sacc)
+	s.EnableJetStream(nil)
 	acc, akp := createAccount(s)
 	if acc == nil {
 		t.Fatalf("did not create account")
 	}
+	acc.EnableJetStream(nil)
 	subsz := fmt.Sprintf(accReqSubj, acc.Name, "SUBSZ")
 	connz := fmt.Sprintf(accReqSubj, acc.Name, "CONNZ")
+	jsz := fmt.Sprintf(accReqSubj, acc.Name, "JSZ")
 	// Create system account connection to query
 	url := fmt.Sprintf("nats://%s:%d", opts.Host, opts.Port)
 	ncSys, err := nats.Connect(url, createUserCreds(t, s, sakp))
@@ -1254,7 +1257,7 @@ func TestAccountReqMonitoring(t *testing.T) {
 	// query SUBSZ for account
 	if resp, err := ncSys.Request(subsz, nil, time.Second); err != nil {
 		t.Fatalf("Error on request: %v", err)
-	} else if !strings.Contains(string(resp.Data), `"num_subscriptions": 0`) {
+	} else if !strings.Contains(string(resp.Data), `"num_subscriptions": 25,`) {
 		t.Fatalf("unexpected subs count (expected 0): %v", string(resp.Data))
 	}
 	// create a subscription
@@ -1267,19 +1270,27 @@ func TestAccountReqMonitoring(t *testing.T) {
 	// query SUBSZ for account
 	if resp, err := ncSys.Request(subsz, nil, time.Second); err != nil {
 		t.Fatalf("Error on request: %v", err)
-	} else if !strings.Contains(string(resp.Data), `"num_subscriptions": 1`) {
-		t.Fatalf("unexpected subs count (expected 1): %v", string(resp.Data))
+	} else if !strings.Contains(string(resp.Data), `"num_subscriptions": 26,`) {
+		t.Fatalf("unexpected subs count (expected 26): %v", string(resp.Data))
 	} else if !strings.Contains(string(resp.Data), `"subject": "foo"`) {
 		t.Fatalf("expected subscription foo: %v", string(resp.Data))
 	}
 	// query connections for account
 	if resp, err := ncSys.Request(connz, nil, time.Second); err != nil {
 		t.Fatalf("Error on request: %v", err)
-	} else if !strings.Contains(string(resp.Data), `"num_connections": 1`) {
+	} else if !strings.Contains(string(resp.Data), `"num_connections": 1,`) {
 		t.Fatalf("unexpected subs count (expected 1): %v", string(resp.Data))
-	} else if !strings.Contains(string(resp.Data), `"total": 2`) { // includes system acc connection
+	} else if !strings.Contains(string(resp.Data), `"total": 2,`) { // includes system acc connection
 		t.Fatalf("unexpected subs count (expected 1): %v", string(resp.Data))
 	}
+	// query connections for js account
+	if resp, err := ncSys.Request(jsz, nil, time.Second); err != nil {
+		t.Fatalf("Error on request: %v", err)
+	} else if !strings.Contains(string(resp.Data), `"memory": 0,`) {
+		t.Fatalf("jetstream should be enabled but empty: %v", string(resp.Data))
+	} else if !strings.Contains(string(resp.Data), `"storage": 0,`) {
+		t.Fatalf("jetstream should be enabled but empty: %v", string(resp.Data))
+	}
 }
 
 func TestAccountReqInfo(t *testing.T) {
@@ -1583,7 +1594,7 @@ func TestSystemAccountWithGateways(t *testing.T) {
 
 	// If this tests fails with wrong number after 10 seconds we may have
 	// added a new inititial subscription for the eventing system.
-	checkExpectedSubs(t, 33, sa)
+	checkExpectedSubs(t, 36, sa)
 
 	// Create a client on B and see if we receive the event
 	urlb := fmt.Sprintf("nats://%s:%d", ob.Host, ob.Port)
@@ -1972,6 +1983,8 @@ func TestServerEventsPingMonitorz(t *testing.T) {
 			[]string{"now", "routes"}},
 		{"ROUTEZ", json.RawMessage(`{"cluster":"TEST CLUSTER 22", "subscriptions":true}`), &Routez{},
 			[]string{"now", "routes"}},
+
+		{"JSZ", nil, &JSzOptions{}, []string{"now", "disabled"}},
 	}
 
 	for i, test := range tests {
diff --git a/server/monitor_test.go b/server/monitor_test.go
index a0512988..03b6af22 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -3789,7 +3789,7 @@ func TestMonitorAccountz(t *testing.T) {
 		t.Fatalf("Body missing value. Contains: %s", body)
 	} else if !strings.Contains(body, `"account_name": "$SYS",`) {
 		t.Fatalf("Body missing value. Contains: %s", body)
-	} else if !strings.Contains(body, `"subscriptions": 32,`) {
+	} else if !strings.Contains(body, `"subscriptions": 35,`) {
 		t.Fatalf("Body missing value. Contains: %s", body)
 	}
 }
@@ -3886,3 +3886,239 @@ func TestMonitorAuthorizedUsers(t *testing.T) {
 	// we should get the user's pubkey
 	checkAuthUser(upub)
 }
+
+// Helper function to check that a JS cluster is formed
+func checkForJSClusterUp(t *testing.T, servers ...*Server) {
+	t.Helper()
+	checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
+		for _, s := range servers {
+			if !s.JetStreamEnabled() {
+				return fmt.Errorf("jetstream not enabled")
+			}
+			if !s.JetStreamIsCurrent() {
+				return fmt.Errorf("jetstream not current")
+			}
+		}
+		return nil
+	})
+}
+
+func TestMonitorJsz(t *testing.T) {
+	readJsInfo := func(url string) *JSInfo {
+		t.Helper()
+		body := readBody(t, url)
+		info := &JSInfo{}
+		err := json.Unmarshal(body, info)
+		require_NoError(t, err)
+		return info
+	}
+	srvs := []*Server{}
+	for _, test := range []struct {
+		port   int
+		mport  int
+		cport  int
+		routed int
+	}{
+		{7500, 7501, 7502, 5502},
+		{5500, 5501, 5502, 7502},
+	} {
+		tmpDir := createDir(t, fmt.Sprintf("srv_%d", test.port))
+		defer os.RemoveAll(tmpDir)
+		cf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: 127.0.0.1:%d
+		http: 127.0.0.1:%d
+		system_account: SYS
+		accounts {
+			SYS {
+				users [{user: sys, password: pwd}]
+			}
+			ACC {
+				users [{user: usr, password: pwd}]
+				jetstream: enabled
+			}
+			BCC_TO_HAVE_ONE_EXTRA {
+				users [{user: usr2, password: pwd}]
+				jetstream: enabled
+			}
+		}
+		jetstream: {
+			max_mem_store: 10Mb
+			max_file_store: 10Mb
+			store_dir: %s
+		}
+		cluster {
+			name: cluster_name
+			listen: 127.0.0.1:%d
+			routes: [nats-route://127.0.0.1:%d]
+		}
+		server_name: server_%d `, test.port, test.mport, tmpDir, test.cport, test.routed, test.port)))
+		defer os.Remove(cf)
+
+		s, _ := RunServerWithConfig(cf)
+		defer s.Shutdown()
+		srvs = append(srvs, s)
+	}
+	checkClusterFormed(t, srvs...)
+	checkForJSClusterUp(t, srvs...)
+
+	nc := natsConnect(t, "nats://usr:pwd@127.0.0.1:7500")
+	defer nc.Close()
+	js, err := nc.JetStream(nats.MaxWait(5 * time.Second))
+	require_NoError(t, err)
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "my-stream-replicated",
+		Subjects: []string{"foo", "bar"},
+		Replicas: 2,
+	})
+	require_NoError(t, err)
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "my-stream-non-replicated",
+		Subjects: []string{"baz"},
+		Replicas: 1,
+	})
+	require_NoError(t, err)
+	_, err = js.AddConsumer("my-stream-replicated", &nats.ConsumerConfig{
+		Durable:   "my-consumer-replicated",
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+	_, err = js.AddConsumer("my-stream-non-replicated", &nats.ConsumerConfig{
+		Durable:   "my-consumer-non-replicated",
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+	nc.Flush()
+	_, err = js.Publish("foo", nil)
+	require_NoError(t, err)
+
+	monUrl1 := fmt.Sprintf("http://127.0.0.1:%d/jsz", 7501)
+	monUrl2 := fmt.Sprintf("http://127.0.0.1:%d/jsz", 5501)
+
+	t.Run("default", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url)
+			if len(info.AccountDetails) != 0 {
+				t.Fatalf("expected no account to be returned by %s but got %v", url, info)
+			}
+			if info.StreamCnt == 0 {
+				t.Fatalf("expected stream count to be 2 but got %d", info.StreamCnt)
+			}
+			if info.ConsumerCnt == 0 {
+				t.Fatalf("expected consumer count to be 2 but got %d", info.ConsumerCnt)
+			}
+			if info.MessageCnt != 1 {
+				t.Fatalf("expected one message but got %d", info.MessageCnt)
+			}
+		}
+	})
+	t.Run("accounts", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url + "?accounts=true")
+			if len(info.AccountDetails) != 2 {
+				t.Fatalf("expected both accounts to be returned by %s but got %v", url, info)
+			}
+		}
+	})
+	t.Run("offset-too-big", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url + "?accounts=true&offset=10")
+			if len(info.AccountDetails) != 0 {
+				t.Fatalf("expected no accounts to be returned by %s but got %v", url, info)
+			}
+		}
+	})
+	t.Run("limit", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url + "?accounts=true&limit=1")
+			if len(info.AccountDetails) != 1 {
+				t.Fatalf("expected one account to be returned by %s but got %v", url, info)
+			}
+			if info := readJsInfo(url + "?accounts=true&offset=1&limit=1"); len(info.AccountDetails) != 1 {
+				t.Fatalf("expected one account to be returned by %s but got %v", url, info)
+			}
+		}
+	})
+	t.Run("offset-stable", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info1 := readJsInfo(url + "?accounts=true&offset=1&limit=1")
+			if len(info1.AccountDetails) != 1 {
+				t.Fatalf("expected one account to be returned by %s but got %v", url, info1)
+			}
+			info2 := readJsInfo(url + "?accounts=true&offset=1&limit=1")
+			if len(info2.AccountDetails) != 1 {
+				t.Fatalf("expected one account to be returned by %s but got %v", url, info2)
+			}
+			if info1.AccountDetails[0].Name != info2.AccountDetails[0].Name {
+				t.Fatalf("absent changes, same offset should result in same account but gut: %v %v",
+					info1.AccountDetails[0].Name, info2.AccountDetails[0].Name)
+			}
+		}
+	})
+	t.Run("filter-account", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url + "?acc=ACC")
+			if len(info.AccountDetails) != 1 {
+				t.Fatalf("expected account ACC to be returned by %s but got %v", url, info)
+			}
+			if info.AccountDetails[0].Name != "ACC" {
+				t.Fatalf("expected account ACC to be returned by %s but got %v", url, info)
+			}
+			if len(info.AccountDetails[0].Streams) != 0 {
+				t.Fatalf("expected account ACC to be returned by %s but got %v", url, info)
+			}
+		}
+	})
+	t.Run("streams", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url + "?acc=ACC&streams=true")
+			if len(info.AccountDetails) != 1 {
+				t.Fatalf("expected account ACC to be returned by %s but got %v", url, info)
+			}
+			if len(info.AccountDetails[0].Streams) == 0 {
+				t.Fatalf("expected streams to be returned by %s but got %v", url, info)
+			}
+			if len(info.AccountDetails[0].Streams[0].Consumer) != 0 {
+				t.Fatalf("expected no consumers to be returned by %s but got %v", url, info)
+			}
+		}
+	})
+	t.Run("consumers", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url + "?acc=ACC&consumers=true")
+			if len(info.AccountDetails) != 1 {
+				t.Fatalf("expected account ACC to be returned by %s but got %v", url, info)
+			}
+			if len(info.AccountDetails[0].Streams[0].Consumer) == 0 {
+				t.Fatalf("expected consumers to be returned by %s but got %v", url, info)
+			}
+			if info.AccountDetails[0].Streams[0].Config != nil {
+				t.Fatal("Config expected to not be present")
+			}
+			if info.AccountDetails[0].Streams[0].Consumer[0].Config != nil {
+				t.Fatal("Config expected to not be present")
+			}
+		}
+	})
+	t.Run("config", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url + "?acc=ACC&consumers=true&config=true")
+			if len(info.AccountDetails) != 1 {
+				t.Fatalf("expected account ACC to be returned by %s but got %v", url, info)
+			}
+			if info.AccountDetails[0].Streams[0].Config == nil {
+				t.Fatal("Config expected to be present")
+			}
+			if info.AccountDetails[0].Streams[0].Consumer[0].Config == nil {
+				t.Fatal("Config expected to be present")
+			}
+		}
+	})
+	t.Run("account-non-existing", func(t *testing.T) {
+		for _, url := range []string{monUrl1, monUrl2} {
+			info := readJsInfo(url + "?acc=DOES_NOT_EXIT")
+			if len(info.AccountDetails) != 0 {
+				t.Fatalf("expected no account to be returned by %s but got %v", url, info)
+			}
+		}
+	})
+}
diff --git a/test/jetstream_cluster_test.go b/test/jetstream_cluster_test.go
index 969e4ac3..c74a3bda 100644
--- a/test/jetstream_cluster_test.go
+++ b/test/jetstream_cluster_test.go
@@ -322,7 +322,6 @@ func TestJetStreamClusterCompaction(t *testing.T) {
 			t.Fatalf("Unexpected error: %v", err)
 		}
 	}
-	nc.Flush()
 
 	if _, err := js.StreamInfo("TEST"); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -1277,9 +1276,8 @@ func TestJetStreamClusterStreamNormalCatchup(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
-	s := c.randomServer()
-
 	// Client based API
+	s := c.randomServer()
 	nc, js := jsClientConnect(t, s)
 	defer nc.Close()
 
@@ -1403,6 +1401,80 @@ func TestJetStreamClusterStreamSnapshotCatchup(t *testing.T) {
 	c.waitOnStreamCurrent(sl, "$G", "TEST")
 }
 
+func TestJetStreamClusterDeleteMsg(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	// Client based API
+	s := c.randomServer()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	// R=1 make sure delete works.
+	_, err := js.AddStream(&nats.StreamConfig{Name: "TEST"})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	toSend := 10
+	for i := 1; i <= toSend; i++ {
+		msg := []byte(fmt.Sprintf("HELLO JSC-%d", i))
+		if _, err = js.Publish("TEST", msg); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+
+	deleteMsg := func(seq uint64) {
+		if err := js.DeleteMsg("TEST", seq); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	}
+
+	deleteMsg(1)
+
+	// Also make sure purge of R=1 works too.
+	if err := js.PurgeStream("TEST"); err != nil {
+		t.Fatalf("Unexpected purge error: %v", err)
+	}
+}
+
+func TestJetStreamClusterDeleteMsgAndRestart(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	// Client based API
+	s := c.randomServer()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	// R=1 make sure delete works.
+	_, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Replicas: 2})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	toSend := 10
+	for i := 1; i <= toSend; i++ {
+		msg := []byte(fmt.Sprintf("HELLO JSC-%d", i))
+		if _, err = js.Publish("TEST", msg); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+
+	deleteMsg := func(seq uint64) {
+		if err := js.DeleteMsg("TEST", seq); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	}
+
+	deleteMsg(1)
+
+	c.stopAll()
+	c.restartAll()
+
+	c.waitOnStreamLeader("$G", "TEST")
+}
+
 func TestJetStreamClusterStreamSnapshotCatchupWithPurge(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R5S", 5)
 	defer c.shutdown()
@@ -3097,6 +3169,253 @@ func TestJetStreamClusterRemoveServer(t *testing.T) {
 	})
 }
 
+func TestJetStreamClusterPurgeReplayAfterRestart(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "P3F", 3)
+	defer c.shutdown()
+
+	// Client based API
+	s := c.randomNonLeader()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	if _, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Replicas: 3}); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	sendBatch := func(n int) {
+		t.Helper()
+		// Send a batch to a given subject.
+		for i := 0; i < n; i++ {
+			if _, err := js.Publish("TEST", []byte("OK")); err != nil {
+				t.Fatalf("Unexpected publish error: %v", err)
+			}
+		}
+	}
+
+	sendBatch(10)
+	if err := js.PurgeStream("TEST"); err != nil {
+		t.Fatalf("Unexpected purge error: %v", err)
+	}
+	sendBatch(10)
+
+	c.stopAll()
+	c.restartAll()
+
+	c.waitOnStreamLeader("$G", "TEST")
+
+	s = c.randomServer()
+	nc, js = jsClientConnect(t, s)
+	defer nc.Close()
+
+	si, err := js.StreamInfo("TEST")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if si.State.Msgs != 10 {
+		t.Fatalf("Expected 10 msgs after restart, got %d", si.State.Msgs)
+	}
+}
+
+func TestJetStreamClusterStreamGetMsg(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3F", 3)
+	defer c.shutdown()
+
+	// Client based API
+	s := c.randomServer()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	if _, err := js.AddStream(&nats.StreamConfig{Name: "TEST"}); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if _, err := js.Publish("TEST", []byte("OK")); err != nil {
+		t.Fatalf("Unexpected publish error: %v", err)
+	}
+
+	mreq := &server.JSApiMsgGetRequest{Seq: 1}
+	req, err := json.Marshal(mreq)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	rmsg, err := nc.Request(fmt.Sprintf(server.JSApiMsgGetT, "TEST"), req, time.Second)
+	if err != nil {
+		t.Fatalf("Could not retrieve stream message: %v", err)
+	}
+	if err != nil {
+		t.Fatalf("Could not retrieve stream message: %v", err)
+	}
+
+	var resp server.JSApiMsgGetResponse
+	err = json.Unmarshal(rmsg.Data, &resp)
+	if err != nil {
+		t.Fatalf("Could not parse stream message: %v", err)
+	}
+	if resp.Message == nil || resp.Error != nil {
+		t.Fatalf("Did not receive correct response: %+v", resp.Error)
+	}
+}
+
+func TestJetStreamClusterSuperClusterBasics(t *testing.T) {
+	sc := createJetStreamSuperCluster(t, 3, 3)
+	defer sc.shutdown()
+
+	// Client based API
+	s := sc.randomCluster().randomServer()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Replicas: 3})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Send in 10 messages.
+	msg, toSend := []byte("Hello JS Clustering"), 10
+	for i := 0; i < toSend; i++ {
+		if _, err = js.Publish("TEST", msg); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+	// Now grab info for this stream.
+	si, err := js.StreamInfo("TEST")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if si == nil || si.Config.Name != "TEST" {
+		t.Fatalf("StreamInfo is not correct %+v", si)
+	}
+	// Check active state as well, shows that the owner answered.
+	if si.State.Msgs != uint64(toSend) {
+		t.Fatalf("Expected %d msgs, got bad state: %+v", toSend, si.State)
+	}
+	// Check request origin placement.
+	if si.Cluster.Name != s.ClusterName() {
+		t.Fatalf("Expected stream to be placed in %q, but got %q", s.ClusterName(), si.Cluster.Name)
+	}
+
+	// Check consumers.
+	sub, err := js.SubscribeSync("TEST")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	checkSubsPending(t, sub, toSend)
+	ci, err := sub.ConsumerInfo()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if ci.Delivered.Consumer != uint64(toSend) || ci.NumAckPending != toSend {
+		t.Fatalf("ConsumerInfo is not correct: %+v", ci)
+	}
+
+	// Now check we can place a stream.
+	// Need to do this by hand for now until Go client catches up.
+	pcn := "C3"
+	cfg := server.StreamConfig{
+		Name:      "TEST2",
+		Storage:   server.FileStorage,
+		Placement: &server.Placement{Cluster: pcn},
+	}
+	req, err := json.Marshal(cfg)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	resp, _ := nc.Request(fmt.Sprintf(server.JSApiStreamCreateT, cfg.Name), req, time.Second)
+	var scResp server.JSApiStreamCreateResponse
+	if err := json.Unmarshal(resp.Data, &scResp); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if scResp.StreamInfo == nil || scResp.Error != nil {
+		t.Fatalf("Did not receive correct response: %+v", scResp.Error)
+	}
+
+	if scResp.StreamInfo.Cluster.Name != pcn {
+		t.Fatalf("Expected the stream to be placed in %q, got %q", pcn, scResp.StreamInfo.Cluster.Name)
+	}
+}
+
+func TestJetStreamClusterSuperClusterPeerReassign(t *testing.T) {
+	sc := createJetStreamSuperCluster(t, 3, 3)
+	defer sc.shutdown()
+
+	// Client based API
+	s := sc.randomCluster().randomServer()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	pcn := "C2"
+	cfg := server.StreamConfig{
+		Name:      "TEST",
+		Replicas:  3,
+		Storage:   server.FileStorage,
+		Placement: &server.Placement{Cluster: pcn},
+	}
+	req, err := json.Marshal(cfg)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	resp, err := nc.Request(fmt.Sprintf(server.JSApiStreamCreateT, cfg.Name), req, time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	var scResp server.JSApiStreamCreateResponse
+	if err := json.Unmarshal(resp.Data, &scResp); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if scResp.StreamInfo == nil || scResp.Error != nil {
+		t.Fatalf("Did not receive correct response: %+v", scResp.Error)
+	}
+
+	// Send in 10 messages.
+	msg, toSend := []byte("Hello JS Clustering"), 10
+	for i := 0; i < toSend; i++ {
+		if _, err = js.Publish("TEST", msg); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+	// Now grab info for this stream.
+	si, err := js.StreamInfo("TEST")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if si == nil || si.Config.Name != "TEST" {
+		t.Fatalf("StreamInfo is not correct %+v", si)
+	}
+	// Check active state as well, shows that the owner answered.
+	if si.State.Msgs != uint64(toSend) {
+		t.Fatalf("Expected %d msgs, got bad state: %+v", toSend, si.State)
+	}
+	// Check request origin placement.
+	if si.Cluster.Name != pcn {
+		t.Fatalf("Expected stream to be placed in %q, but got %q", s.ClusterName(), si.Cluster.Name)
+	}
+
+	// Now remove a peer that is assigned to the stream.
+	rc := sc.clusterForName(pcn)
+	rs := rc.randomNonStreamLeader("$G", "TEST")
+	rc.removeJetStream(rs)
+
+	// Check the stream info is eventually correct.
+	checkFor(t, 2*time.Second, 50*time.Millisecond, func() error {
+		si, err := js.StreamInfo("TEST")
+		if err != nil {
+			return fmt.Errorf("Could not fetch stream info: %v", err)
+		}
+		if len(si.Cluster.Replicas) != 2 {
+			return fmt.Errorf("Expected 2 replicas, got %d", len(si.Cluster.Replicas))
+		}
+		for _, peer := range si.Cluster.Replicas {
+			if !peer.Current {
+				return fmt.Errorf("Expected replica to be current: %+v", peer)
+			}
+			if !strings.HasPrefix(peer.Name, pcn) {
+				t.Fatalf("Stream peer reassigned to wrong cluster: %q", peer.Name)
+			}
+		}
+		return nil
+	})
+
+}
+
 func TestJetStreamClusterStreamPerf(t *testing.T) {
 	// Comment out to run, holding place for now.
 	skip(t)
@@ -3166,6 +3485,7 @@ var jsClusterTempl = `
 	listen: 127.0.0.1:-1
 	server_name: %s
 	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: "%s"}
+
 	cluster {
 		name: %s
 		listen: 127.0.0.1:%d
@@ -3173,6 +3493,130 @@ var jsClusterTempl = `
 	}
 `
 
+var jsSuperClusterTempl = `
+	%s
+	gateway {
+		name: %s
+		listen: 127.0.0.1:%d
+		gateways = [%s
+		]
+	}
+`
+
+var jsGWTempl = `%s{name: %s, urls: [%s]}`
+
+func createJetStreamSuperCluster(t *testing.T, numServersPer, numClusters int) *supercluster {
+	t.Helper()
+	if numServersPer < 1 {
+		t.Fatalf("Number of servers must be >= 1")
+	}
+	if numClusters <= 1 {
+		t.Fatalf("Number of clusters must be > 1")
+	}
+
+	const (
+		startClusterPort = 33222
+		startGWPort      = 11222
+	)
+
+	// Make the GWs form faster for the tests.
+	server.SetGatewaysSolicitDelay(10 * time.Millisecond)
+	defer server.ResetGatewaysSolicitDelay()
+
+	cp, gp := startClusterPort, startGWPort
+	var clusters []*cluster
+
+	var gws []string
+	// Build GWs first, will be same for all servers.
+	for i, port := 1, gp; i <= numClusters; i++ {
+		cn := fmt.Sprintf("C%d", i)
+		var urls []string
+		for n := 0; n < numServersPer; n++ {
+			urls = append(urls, fmt.Sprintf("nats-route://127.0.0.1:%d", port))
+			port++
+		}
+		gws = append(gws, fmt.Sprintf(jsGWTempl, "\n\t\t\t", cn, strings.Join(urls, ",")))
+	}
+	gwconf := strings.Join(gws, "")
+
+	for i := 1; i <= numClusters; i++ {
+		cn := fmt.Sprintf("C%d", i)
+		// Go ahead and build configurations.
+		c := &cluster{servers: make([]*server.Server, 0, numServersPer), opts: make([]*server.Options, 0, numServersPer), name: cn}
+
+		// Build out the routes that will be shared with all configs.
+		var routes []string
+		for port := cp; port < cp+numServersPer; port++ {
+			routes = append(routes, fmt.Sprintf("nats-route://127.0.0.1:%d", port))
+		}
+		routeConfig := strings.Join(routes, ",")
+
+		for i := 0; i < numServersPer; i++ {
+			storeDir, _ := ioutil.TempDir("", server.JetStreamStoreDir)
+			sn := fmt.Sprintf("%s-S%d", cn, i+1)
+			bconf := fmt.Sprintf(jsClusterTempl, sn, storeDir, cn, cp+i, routeConfig)
+			conf := fmt.Sprintf(jsSuperClusterTempl, bconf, cn, gp, gwconf)
+			gp++
+			s, o := RunServerWithConfig(createConfFile(t, []byte(conf)))
+			c.servers = append(c.servers, s)
+			c.opts = append(c.opts, o)
+		}
+		checkClusterFormed(t, c.servers...)
+		clusters = append(clusters, c)
+		cp += numServersPer
+		c.t = t
+	}
+
+	// Wait for the supercluster to be formed.
+	egws := numClusters - 1
+	for _, c := range clusters {
+		for _, s := range c.servers {
+			waitForOutboundGateways(t, s, egws, 2*time.Second)
+		}
+	}
+
+	sc := &supercluster{t, clusters}
+	sc.waitOnLeader()
+	sc.waitOnAllCurrent()
+	return sc
+}
+
+func (sc *supercluster) waitOnLeader() {
+	expires := time.Now().Add(5 * time.Second)
+	for time.Now().Before(expires) {
+		for _, c := range sc.clusters {
+			if leader := c.leader(); leader != nil {
+				time.Sleep(200 * time.Millisecond)
+				return
+			}
+		}
+		time.Sleep(25 * time.Millisecond)
+	}
+
+	sc.t.Fatalf("Expected a cluster leader, got none")
+}
+
+func (sc *supercluster) waitOnAllCurrent() {
+	for _, c := range sc.clusters {
+		c.waitOnAllCurrent()
+	}
+}
+
+func (sc *supercluster) clusterForName(name string) *cluster {
+	for _, c := range sc.clusters {
+		if c.name == name {
+			return c
+		}
+	}
+	return nil
+}
+
+func (sc *supercluster) randomCluster() *cluster {
+	clusters := append(sc.clusters[:0:0], sc.clusters...)
+	rand.Shuffle(len(clusters), func(i, j int) { clusters[i], clusters[j] = clusters[j], clusters[i] })
+	return clusters[0]
+}
+
 // This will create a cluster that is explicitly configured for the routes, etc.
 // and also has a defined clustername. All configs for routes and cluster name will be the same.
 func createJetStreamClusterExplicit(t *testing.T, clusterName string, numServers int) *cluster {
diff --git a/test/service_latency_test.go b/test/service_latency_test.go
index fb30e721..9083fbe9 100644
--- a/test/service_latency_test.go
+++ b/test/service_latency_test.go
@@ -35,10 +35,14 @@ import (
 
 // Used to setup superclusters for tests.
 type supercluster struct {
+	t        *testing.T
 	clusters []*cluster
 }
 
 func (sc *supercluster) shutdown() {
+	if sc == nil {
+		return
+	}
 	for _, c := range sc.clusters {
 		shutdownCluster(c)
 	}
@@ -66,7 +70,7 @@ func createSuperCluster(t *testing.T, numServersPer, numClusters int) *superclus
 		c := createClusterEx(t, true, 5*time.Millisecond, true, randClusterName(), numServersPer, clusters...)
 		clusters = append(clusters, c)
 	}
-	return &supercluster{clusters}
+	return &supercluster{t, clusters}
 }
 
 func (sc *supercluster) setResponseThreshold(t *testing.T, maxTime time.Duration) {