Repository: nats-io/nats-server
Pull request #1947 (2021-03-01 02:36:06)

Stability Fixes, bug fixes, RC features.
Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/.github/workflows/nightly.yaml b/.github/workflows/nightly.yaml
index d76d2b12..83011b90 100644
--- a/.github/workflows/nightly.yaml
+++ b/.github/workflows/nightly.yaml
@@ -23,7 +23,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.16
+          go-version: 1.15
 
       - name: goreleaser
         uses: goreleaser/goreleaser-action@v2
diff --git a/docker/Dockerfile.nightly b/docker/Dockerfile.nightly
index fc86f64f..333d848f 100644
--- a/docker/Dockerfile.nightly
+++ b/docker/Dockerfile.nightly
@@ -1,4 +1,4 @@
-FROM golang:1.16-alpine AS builder
+FROM golang:1.15-alpine AS builder
 
 ARG VERSION="nightly"
 
diff --git a/go.mod b/go.mod
index 370b430c..435847c0 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/klauspost/compress v1.11.7
 	github.com/minio/highwayhash v1.0.0
 	github.com/nats-io/jwt/v2 v2.0.0-20210208203759-ff814ca5f813
-	github.com/nats-io/nats.go v1.10.1-0.20210211000709-75ded9c77585
+	github.com/nats-io/nats.go v1.10.1-0.20210228004050-ed743748acac
 	github.com/nats-io/nkeys v0.2.0
 	github.com/nats-io/nuid v1.0.1
 	golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897
diff --git a/go.sum b/go.sum
index 816ce08d..76396bb5 100644
--- a/go.sum
+++ b/go.sum
@@ -27,6 +27,8 @@ github.com/nats-io/nats-server/v2 v2.1.8-0.20200601203034-f8d6dd992b71/go.mod h1
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200929001935-7f44d075f7ad/go.mod h1:TkHpUIDETmTI7mrHN40D1pzxfzHZuGmtMbtb83TGVQw=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20201129161730-ebe63db3e3ed/go.mod h1:XD0zHR/jTXdZvWaQfS5mQgsXj6x12kMjKLyAk/cOGgY=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20210205154825-f7ab27f7dad4/go.mod h1:kauGd7hB5517KeSqspW2U1Mz/jhPbTrE8eOXzUPk1m0=
+github.com/nats-io/nats-server/v2 v2.1.8-0.20210224195833-15fa5e21b858/go.mod h1:/QQ/dpqFavkNhVnjvMILSQ3cj5hlmhB66adlgNbjuoA=
+github.com/nats-io/nats-server/v2 v2.1.8-0.20210227190344-51550e242af8/go.mod h1:/QQ/dpqFavkNhVnjvMILSQ3cj5hlmhB66adlgNbjuoA=
 github.com/nats-io/nats.go v1.10.0/go.mod h1:AjGArbfyR50+afOUotNX2Xs5SYHf+CoOa5HH1eEl2HE=
 github.com/nats-io/nats.go v1.10.1-0.20200531124210-96f2130e4d55/go.mod h1:ARiFsjW9DVxk48WJbO3OSZ2DG8fjkMi7ecLmXoY/n9I=
 github.com/nats-io/nats.go v1.10.1-0.20200606002146-fc6fed82929a/go.mod h1:8eAIv96Mo9QW6Or40jUHejS7e4VwZ3VRYD6Sf0BTDp4=
@@ -35,6 +37,10 @@ github.com/nats-io/nats.go v1.10.1-0.20210127212649-5b4924938a9a h1:EjwBk6T/arS7
 github.com/nats-io/nats.go v1.10.1-0.20210127212649-5b4924938a9a/go.mod h1:Sa3kLIonafChP5IF0b55i9uvGR10I3hPETFbi4+9kOI=
 github.com/nats-io/nats.go v1.10.1-0.20210211000709-75ded9c77585 h1:xbs6PNOyQcxNFXII9qcFvodqBtQKec8hP7WzGHOdsz0=
 github.com/nats-io/nats.go v1.10.1-0.20210211000709-75ded9c77585/go.mod h1:uBWnCKg9luW1g7hgzPxUjHFRI40EuTSX7RCzgnc74Jk=
+github.com/nats-io/nats.go v1.10.1-0.20210225223043-fdeedcea71bf h1:IpesKsWjU6IyEbCY0Ubz8WWPdAhgbjSIx9WkzixBDrY=
+github.com/nats-io/nats.go v1.10.1-0.20210225223043-fdeedcea71bf/go.mod h1:vdUKYi2X5GRWRcHxLavN3LJRgf0AE2CPBum79YrQ3a0=
+github.com/nats-io/nats.go v1.10.1-0.20210228004050-ed743748acac h1:/cF7DEtxQBcwRDhpFZ3J0XU4TFpJa9KQF/xDirRNNI0=
+github.com/nats-io/nats.go v1.10.1-0.20210228004050-ed743748acac/go.mod h1:hxFvLNbNmT6UppX5B5Tr/r3g+XSwGjJzFn6mxPNJEHc=
 github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nkeys v0.1.4/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
 github.com/nats-io/nkeys v0.2.0 h1:WXKF7diOaPU9cJdLD7nuzwasQy9vT1tBqzXZZf3AMJM=
diff --git a/server/consumer.go b/server/consumer.go
index c397c7f3..450ca046 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -704,18 +704,10 @@ func (o *consumer) setLeader(isLeader bool) {
 }
 
 func (o *consumer) handleClusterConsumerInfoRequest(sub *subscription, c *client, subject, reply string, msg []byte) {
-	var s *Server
 	o.mu.RLock()
-	if o.client != nil {
-		s = o.client.srv
-	}
+	sysc := o.sysc
 	o.mu.RUnlock()
-	if s == nil {
-		return
-	}
-	ci := o.info()
-	b, _ := json.Marshal(ci)
-	s.sendInternalMsgLocked(reply, _EMPTY_, nil, b)
+	sysc.sendInternalMsg(reply, _EMPTY_, nil, o.info())
 }
 
 // Lock should be held.
@@ -1998,58 +1990,6 @@ func (o *consumer) ackReply(sseq, dseq, dc uint64, ts int64, pending uint64) str
 	return fmt.Sprintf(o.ackReplyT, dc, sseq, dseq, ts, pending)
 }
 
-// deliverCurrentMsg is the hot path to deliver a message that was just received.
-// Will return if the message was delivered or not.
-func (o *consumer) deliverCurrentMsg(subj string, hdr, msg []byte, seq uint64, ts int64) bool {
-	o.mu.Lock()
-	if seq != o.sseq {
-		o.mu.Unlock()
-		return false
-	}
-
-	// If we are in push mode and not active let's stop sending.
-	if o.isPushMode() && !o.active {
-		o.mu.Unlock()
-		return false
-	}
-
-	// If we are in pull mode and no one is waiting already break and wait.
-	if o.isPullMode() && !o.checkWaitingForInterest() {
-		o.mu.Unlock()
-		return false
-	}
-
-	// Since we short circuit the getNextMsg() call where we check for max pending
-	// we need to do that here as well.
-	if o.maxp > 0 && len(o.pending) >= o.maxp {
-		o.mu.Unlock()
-		return false
-	}
-
-	// Bump store sequence here.
-	o.sseq++
-
-	// If we are partitioned and we do not match, do not consider this a failure.
-	// Go ahead and return true.
-	if o.cfg.FilterSubject != _EMPTY_ && !o.isFilteredMatch(subj) {
-		o.updateSkipped()
-		o.mu.Unlock()
-		return true
-	}
-
-	var dsubj string
-	if wr := o.waiting.pop(); wr != nil {
-		dsubj = wr.reply
-	} else {
-		dsubj = o.dsubj
-	}
-
-	o.deliverMsg(dsubj, subj, hdr, msg, seq, 1, ts)
-	o.mu.Unlock()
-
-	return true
-}
-
 // Deliver a msg to the consumer.
 // Lock should be held and o.mset validated to be non-nil.
 func (o *consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint64, ts int64) {
@@ -2069,10 +2009,13 @@ func (o *consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint6
 	}
 
 	dseq := o.dseq
+	o.dseq++
+
 	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, dseq, dc, ts, o.sgap), hdr, msg, o, seq}
 	mset := o.mset
 	ap := o.cfg.AckPolicy
 	sendq := o.sendq
+
 	// This needs to be unlocked since the other side may need this lock on a failed delivery.
 	o.mu.Unlock()
 	// Send message.
@@ -2087,12 +2030,10 @@ func (o *consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint6
 	if ap == AckExplicit || ap == AckAll {
 		o.trackPending(seq, dseq)
 	} else if ap == AckNone {
-		o.adflr = o.dseq
+		o.adflr = dseq
 		o.asflr = seq
 	}
 
-	o.dseq++
-
 	// FIXME(dlc) - Capture errors?
 	o.updateDelivered(dseq, seq, dc, ts)
 }
@@ -2667,15 +2608,6 @@ func (o *consumer) setInitialPending() {
 	}
 }
 
-// addStreamPending will add to the stream pending.
-func (o *consumer) incStreamPending(sseq uint64, subj string) {
-	o.mu.Lock()
-	if o.isFilteredMatch(subj) {
-		o.sgap++
-	}
-	o.mu.Unlock()
-}
-
 func (o *consumer) decStreamPending(sseq uint64, subj string) {
 	o.mu.Lock()
 	// Ignore if we have already seen this one.
diff --git a/server/events.go b/server/events.go
index ff72266b..3ea3a63d 100644
--- a/server/events.go
+++ b/server/events.go
@@ -421,6 +421,26 @@ func (s *Server) sendInternalMsg(sub, rply string, si *ServerInfo, msg interface
 	s.mu.Lock()
 }
 
+// Used to send internal messages from other system clients to avoid no echo issues.
+func (c *client) sendInternalMsg(sub, rply string, si *ServerInfo, msg interface{}) {
+	if c == nil {
+		return
+	}
+	s := c.srv
+	if s == nil {
+		return
+	}
+	s.mu.Lock()
+	if s.sys == nil || s.sys.sendq == nil {
+		return
+	}
+	sendq := s.sys.sendq
+	// Don't hold lock while placing on the channel.
+	s.mu.Unlock()
+
+	sendq <- &pubMsg{c, sub, rply, si, msg, false}
+}
+
 // Locked version of checking if events system running. Also checks server.
 func (s *Server) eventsRunning() bool {
 	if s == nil {
diff --git a/server/filestore.go b/server/filestore.go
index 43559e0c..c5392890 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -515,6 +515,7 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 		rl := le.Uint32(hdr[0:])
 		slen := le.Uint16(hdr[20:])
 
+		hasHeaders := rl&hbit != 0
 		// Clear any headers bit that could be set.
 		rl &^= hbit
 		dlen := int(rl) - msgHdrSize
@@ -535,6 +536,7 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 		// If the first seq we read does not match our indexed first seq, reset.
 		if index == 0 && seq > mb.first.seq {
 			mb.first.seq = seq
+			mb.first.ts = ts
 		}
 
 		// This is an old erased message, or a new one that we can track.
@@ -558,7 +560,11 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 				hh.Reset()
 				hh.Write(hdr[4:20])
 				hh.Write(data[:slen])
-				hh.Write(data[slen : dlen-8])
+				if hasHeaders {
+					hh.Write(data[slen+4 : dlen-8])
+				} else {
+					hh.Write(data[slen : dlen-8])
+				}
 				checksum := hh.Sum(nil)
 				if !bytes.Equal(checksum, data[len(data)-8:]) {
 					truncate(index)
@@ -795,7 +801,10 @@ func (fs *fileStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts in
 
 	// Check sequence.
 	if seq != fs.state.LastSeq+1 {
-		return ErrSequenceMismatch
+		if seq > 0 {
+			return ErrSequenceMismatch
+		}
+		seq = fs.state.LastSeq + 1
 	}
 
 	// Write msg record.
@@ -1173,16 +1182,16 @@ func (fs *fileStore) isClosed() bool {
 // Will spin up our flush loop.
 func (mb *msgBlock) spinUpFlushLoop() {
 	mb.mu.Lock()
+	defer mb.mu.Unlock()
+
 	// Are we already running?
 	if mb.flusher {
-		mb.mu.Unlock()
 		return
 	}
 	mb.flusher = true
 	mb.fch = make(chan struct{}, 1)
 	mb.qch = make(chan struct{})
 	fch, qch := mb.fch, mb.qch
-	mb.mu.Unlock()
 
 	go mb.flushLoop(fch, qch)
 }
@@ -2659,22 +2668,10 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 		return fs.Purge()
 	}
 
-	// TODO(dlc) - We can be smarter for large compactions and drop whole msg blocks.
 	var purged uint64
-	if last := fs.lastSeq(); seq <= last {
-		for fseq := fs.firstSeq(); fseq < seq; fseq = fs.firstSeq() {
-			if found, err := fs.removeMsg(fseq, false); err != nil {
-				if err == ErrStoreMsgNotFound {
-					continue
-				} else if err == ErrStoreEOF {
-					err = nil
-				}
-				return purged, err
-			} else if found {
-				purged++
-			}
-		}
-	} else {
+	var bytes uint64
+
+	if last := fs.lastSeq(); seq > last {
 		// We are compacting past the end of our range. Do purge and set sequences correctly
 		// such that the next message placed will have seq.
 		var err error
@@ -2682,6 +2679,57 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 			return 0, err
 		}
 		fs.resetFirst(seq)
+	} else {
+		// We have to delete interior messages.
+		fs.mu.Lock()
+		smb := fs.selectMsgBlock(seq)
+		if smb == nil {
+			fs.mu.Unlock()
+			return 0, nil
+		}
+		// All msgblocks up to this one can be thrown away.
+		for i, mb := range fs.blks {
+			if mb == smb {
+				fs.blks = append(fs.blks[:0:0], fs.blks[i:]...)
+				break
+			}
+			mb.mu.Lock()
+			purged += mb.msgs
+			bytes += mb.bytes
+			mb.dirtyCloseWithRemove(true)
+			mb.mu.Unlock()
+		}
+		fs.mu.Unlock()
+
+		if err := smb.loadMsgs(); err != nil {
+			return purged, err
+		}
+
+		smb.mu.Lock()
+		for mseq := smb.first.seq; mseq < seq; mseq++ {
+			if _, rl, _, err := smb.slotInfo(int(mseq - smb.cache.fseq)); err != nil {
+				smb.bytes -= uint64(rl)
+			}
+			smb.msgs--
+			purged++
+		}
+		// Update first entry.
+		sm, _ := smb.cacheLookupWithLock(seq)
+		if sm != nil {
+			smb.first.seq = sm.seq
+			smb.first.ts = sm.ts
+		}
+		smb.mu.Unlock()
+
+		if sm != nil {
+			// Reset our version of first.
+			fs.mu.Lock()
+			fs.state.FirstSeq = sm.seq
+			fs.state.FirstTime = time.Unix(0, sm.ts).UTC()
+			fs.state.Msgs -= purged
+			fs.state.Bytes -= bytes
+			fs.mu.Unlock()
+		}
 	}
 
 	return purged, nil
@@ -2764,13 +2812,6 @@ func (fs *fileStore) resetFirst(newFirst uint64) {
 	fs.mu.Unlock()
 }
 
-func (fs *fileStore) firstSeq() uint64 {
-	fs.mu.RLock()
-	seq := fs.state.FirstSeq
-	fs.mu.RUnlock()
-	return seq
-}
-
 func (fs *fileStore) lastSeq() uint64 {
 	fs.mu.RLock()
 	seq := fs.state.LastSeq
@@ -2806,7 +2847,8 @@ func (fs *fileStore) removeMsgBlock(mb *msgBlock) {
 	// Remove from list.
 	for i, omb := range fs.blks {
 		if mb == omb {
-			fs.blks = append(fs.blks[:i], fs.blks[i+1:]...)
+			blks := append(fs.blks[:i], fs.blks[i+1:]...)
+			fs.blks = append(blks[:0:0], blks...)
 			break
 		}
 	}
@@ -2820,8 +2862,8 @@ func (fs *fileStore) removeMsgBlock(mb *msgBlock) {
 // Lock should not be held.
 func (mb *msgBlock) dirtyClose() {
 	mb.mu.Lock()
+	defer mb.mu.Unlock()
 	mb.dirtyCloseWithRemove(false)
-	mb.mu.Unlock()
 }
 
 // Should be called with lock held.
diff --git a/server/jetstream.go b/server/jetstream.go
index 89cd2adf..d259567f 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -456,6 +456,63 @@ func (s *Server) JetStreamEnabled() bool {
 	return enabled
 }
 
+// Will migrate off ephemerals if possible.
+// This means parent stream needs to be replicated.
+func (s *Server) migrateEphemerals() {
+	js, cc := s.getJetStreamCluster()
+	// Make sure JetStream is enabled and we are clustered.
+	if js == nil || cc == nil {
+		return
+	}
+
+	var consumers []*consumerAssignment
+
+	js.mu.Lock()
+	ourID := cc.meta.ID()
+	for _, asa := range cc.streams {
+		for _, sa := range asa {
+			if rg := sa.Group; rg != nil && len(rg.Peers) > 1 && rg.isMember(ourID) && len(sa.consumers) > 0 {
+				for _, ca := range sa.consumers {
+					if ca.Group != nil && len(ca.Group.Peers) == 1 && ca.Group.isMember(ourID) {
+						// Need to select possible new peer from parent stream.
+						for _, p := range rg.Peers {
+							if p != ourID {
+								ca.Group.Peers = []string{p}
+								ca.Group.Preferred = p
+								consumers = append(consumers, ca)
+								break
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+	js.mu.Unlock()
+
+	// Process the consumers.
+	for _, ca := range consumers {
+		// Locate the consumer itself.
+		if acc, err := s.LookupAccount(ca.Client.Account); err == nil && acc != nil {
+			if mset, err := acc.lookupStream(ca.Stream); err == nil && mset != nil {
+				if o := mset.lookupConsumer(ca.Name); o != nil {
+					state := o.readStoreState()
+					o.deleteWithoutAdvisory()
+					js.mu.Lock()
+					// Delete old one.
+					cc.meta.Propose(encodeDeleteConsumerAssignment(ca))
+					// Encode state and new name.
+					ca.State = state
+					ca.Name = createConsumerName()
+					addEntry := encodeAddConsumerAssignmentCompressed(ca)
+					cc.meta.ForwardProposal(addEntry)
+					js.mu.Unlock()
+				}
+			}
+		}
+	}
+}
+
 // Shutdown jetstream for this server.
 func (s *Server) shutdownJetStream() {
 	s.mu.Lock()
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index bc0ea0c8..9b8f6289 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -164,6 +164,11 @@ const (
 	// Will return JSON response.
 	JSApiLeaderStepDown = "$JS.API.META.LEADER.STEPDOWN"
 
+	// JSApiRemoveServer is the endpoint to remove a peer server from the cluster.
+	// Only works from system account.
+	// Will return JSON response.
+	JSApiRemoveServer = "$JS.API.SERVER.REMOVE"
+
 	// 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"
@@ -226,6 +231,9 @@ const (
 	// JSAdvisoryServerOutOfStorage notification that a server has no more storage.
 	JSAdvisoryServerOutOfStorage = "$JS.EVENT.ADVISORY.SERVER.OUT_OF_STORAGE"
 
+	// JSAdvisoryServerRemoved notification that a server has been removed from the system.
+	JSAdvisoryServerRemoved = "$JS.EVENT.ADVISORY.SERVER.REMOVED"
+
 	// JSAuditAdvisory is a notification about JetStream API access.
 	// FIXME - Add in details about who..
 	JSAuditAdvisory = "$JS.EVENT.ADVISORY.API"
@@ -440,6 +448,20 @@ type JSApiLeaderStepDownResponse struct {
 
 const JSApiLeaderStepDownResponseType = "io.nats.jetstream.api.v1.meta_leader_stepdown_response"
 
+// JSApiLeaderServerRemoveRequest will remove a peer from the system.
+type JSApiLeaderServerRemoveRequest struct {
+	// Server name of the peer to be removed.
+	Server string `json:"peer"`
+}
+
+// JSApiLeaderServerRemoveResponse is the response to a metaleader peer removal request.
+type JSApiLeaderServerRemoveResponse struct {
+	ApiResponse
+	Success bool `json:"success,omitempty"`
+}
+
+const JSApiLeaderServerRemovalResponseType = "io.nats.jetstream.api.v1.meta_leader_server_removal"
+
 // JSApiMsgGetRequest get a message request.
 type JSApiMsgGetRequest struct {
 	Seq uint64 `json:"seq"`
@@ -563,6 +585,7 @@ var (
 	jsClusterIncompleteErr = &ApiError{Code: 503, Description: "incomplete results"}
 	jsClusterTagsErr       = &ApiError{Code: 400, Description: "tags placement not supported for operation"}
 	jsClusterNoPeersErr    = &ApiError{Code: 400, Description: "no suitable peers for placement"}
+	jsServerNotMemberErr   = &ApiError{Code: 400, Description: "server is not a member of the cluster"}
 )
 
 // For easier handling of exports and imports.
@@ -1122,12 +1145,20 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, subject, re
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
-	var cfg StreamConfig
-	if err := json.Unmarshal(msg, &cfg); err != nil {
+	var ncfg StreamConfig
+	if err := json.Unmarshal(msg, &ncfg); err != nil {
 		resp.Error = jsInvalidJSONErr
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
+
+	cfg, err := checkStreamCfg(&ncfg)
+	if err != nil {
+		resp.Error = jsError(err)
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
 	streamName := streamNameFromSubject(subject)
 	if streamName != cfg.Name {
 		resp.Error = jsStreamMismatchErr
@@ -1742,6 +1773,73 @@ func (s *Server) jsStreamRemovePeerRequest(sub *subscription, c *client, subject
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
 }
 
+// Request to have the metaleader remove a peer from the system.
+func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, 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 = JSApiLeaderServerRemoveResponse{ApiResponse: ApiResponse{Type: JSApiLeaderServerRemovalResponseType}}
+
+	if isEmptyRequest(msg) {
+		resp.Error = jsBadRequestErr
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	var req JSApiLeaderServerRemoveRequest
+	if err := json.Unmarshal(msg, &req); err != nil {
+		resp.Error = jsInvalidJSONErr
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	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 = jsServerNotMemberErr
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	// So we have a valid peer.
+	js.mu.Lock()
+	cc.meta.ProposeRemovePeer(found)
+	js.mu.Unlock()
+
+	resp.Success = true
+	s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+}
+
 // 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, subject, reply string, rmsg []byte) {
@@ -2237,6 +2335,10 @@ func (s *Server) jsStreamRestoreRequest(sub *subscription, c *client, subject, r
 
 	stream := streamNameFromSubject(subject)
 
+	if stream != req.Config.Name && req.Config.Name == _EMPTY_ {
+		req.Config.Name = stream
+	}
+
 	if s.JetStreamIsClustered() {
 		s.jsClusteredStreamRestoreRequest(ci, acc, &req, stream, subject, reply, rmsg)
 		return
@@ -2248,10 +2350,10 @@ func (s *Server) jsStreamRestoreRequest(sub *subscription, c *client, subject, r
 		return
 	}
 
-	s.processStreamRestore(ci, acc, stream, subject, reply, string(msg))
+	s.processStreamRestore(ci, acc, &req.Config, subject, reply, string(msg))
 }
 
-func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, streamName, subject, reply, msg string) <-chan error {
+func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamConfig, subject, reply, msg string) <-chan error {
 	var resp = JSApiStreamRestoreResponse{ApiResponse: ApiResponse{Type: JSApiStreamRestoreResponseType}}
 
 	// FIXME(dlc) - Need to close these up if we fail for some reason.
@@ -2263,6 +2365,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, streamName,
 		return nil
 	}
 
+	streamName := cfg.Name
 	s.Noticef("Starting restore for stream '%s > %s'", acc.Name, streamName)
 
 	start := time.Now()
@@ -2289,6 +2392,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, streamName,
 	resultCh := make(chan result, 1)
 	activeCh := make(chan int, 32)
 
+	// FIXM(dlc) - Probably take out of network path eventually do to disk I/O?
 	processChunk := func(sub *subscription, c *client, subject, reply string, msg []byte) {
 		// We require reply subjects to communicate back failures, flow etc. If they do not have one log and cancel.
 		if reply == _EMPTY_ {
@@ -2369,7 +2473,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, streamName,
 				if err == nil {
 					s.Debugf("Finalizing restore for  stream '%s > %s'", acc.Name, streamName)
 					tfile.Seek(0, 0)
-					mset, err = acc.RestoreStream(streamName, tfile)
+					mset, err = acc.RestoreStream(cfg, tfile)
 				} else {
 					errStr := err.Error()
 					tmp := []rune(errStr)
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 44267335..44c0b2a4 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -45,8 +45,10 @@ type jetStreamCluster struct {
 	// Processing assignment results.
 	streamResults   *subscription
 	consumerResults *subscription
-	// For asking for leader to stepdown.
+	// System level request to have the leader stepdown.
 	stepdown *subscription
+	// System level requests to remove a peer.
+	peerRemove *subscription
 }
 
 // Used to guide placement of streams and meta controllers in clustered JetStream.
@@ -948,23 +950,41 @@ func (sa *streamAssignment) copyGroup() *streamAssignment {
 
 func (js *jetStream) processRemovePeer(peer string) {
 	js.mu.Lock()
-	defer js.mu.Unlock()
-	cc := js.cluster
+	s, cc := js.srv, js.cluster
+
+	// Only leader will do remappings for streams and consumers.
+	if cc.isLeader() {
+		// Grab our nodes.
+		// Need to search for this peer in our stream assignments for potential remapping.
+		for _, as := range cc.streams {
+			for _, sa := range as {
+				if sa.Group.isMember(peer) {
+					js.removePeerFromStream(sa, peer)
+				}
+			}
+		}
 
-	// Only leader should process and re-assign mappings.
-	if !cc.isLeader() {
-		return
 	}
 
-	// Grab our nodes.
-	// FIXME(dlc) - Make sure these are live.
-	// Need to search for this peer in our stream assignments for potential remapping.
-	for _, as := range cc.streams {
-		for _, sa := range as {
-			if sa.Group.isMember(peer) {
-				js.removePeerFromStream(sa, peer)
-			}
+	// All nodes will check if this is them.
+	isUs := cc.meta.ID() == peer
+	js.mu.Unlock()
+
+	if isUs {
+		s.Errorf("JetStream being DISABLED, our server was removed from the cluster")
+		adv := &JSServerRemovedAdvisory{
+			TypedEvent: TypedEvent{
+				Type: JSServerRemovedAdvisoryType,
+				ID:   nuid.Next(),
+				Time: time.Now().UTC(),
+			},
+			Server:   s.Name(),
+			ServerID: s.ID(),
+			Cluster:  s.cachedClusterName(),
 		}
+		s.publishAdvisory(nil, JSAdvisoryServerRemoved, adv)
+
+		go s.DisableJetStream()
 	}
 }
 
@@ -1271,7 +1291,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 			if isLeader {
 				if isRestore {
 					acc, _ := s.LookupAccount(sa.Client.serviceAccount())
-					restoreDoneCh = s.processStreamRestore(sa.Client, acc, sa.Config.Name, _EMPTY_, sa.Reply, _EMPTY_)
+					restoreDoneCh = s.processStreamRestore(sa.Client, acc, sa.Config, _EMPTY_, sa.Reply, _EMPTY_)
 					continue
 				} else {
 					doSnapshot()
@@ -1383,23 +1403,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 
 func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isRecovering bool) error {
 	for _, e := range ce.Entries {
-		if e.Type == EntrySnapshot {
-			if !isRecovering && mset != nil {
-				var snap streamSnapshot
-				if err := json.Unmarshal(e.Data, &snap); err != nil {
-					return err
-				}
-				mset.processSnapshot(&snap)
-			}
-		} else if e.Type == EntryRemovePeer {
-			js.mu.RLock()
-			ourID := js.cluster.meta.ID()
-			js.mu.RUnlock()
-			if peer := string(e.Data); peer == ourID {
-				mset.stop(true, false)
-			}
-			return nil
-		} else {
+		if e.Type == EntryNormal {
 			buf := e.Data
 			switch entryOp(buf[0]) {
 			case streamMsgOp:
@@ -1407,6 +1411,10 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 				if err != nil {
 					panic(err.Error())
 				}
+				// We can skip if we know this is less than what we already have.
+				if isRecovering && lseq <= mset.lastSeq() {
+					continue
+				}
 				// Skip by hand here since first msg special case.
 				// Reason is sequence is unsigned and for lseq being 0
 				// the lseq under stream would have be -1.
@@ -1497,6 +1505,22 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 			default:
 				panic("JetStream Cluster Unknown group entry op type!")
 			}
+		} else if e.Type == EntrySnapshot {
+			if !isRecovering && mset != nil {
+				var snap streamSnapshot
+				if err := json.Unmarshal(e.Data, &snap); err != nil {
+					return err
+				}
+				mset.processSnapshot(&snap)
+			}
+		} else if e.Type == EntryRemovePeer {
+			js.mu.RLock()
+			ourID := js.cluster.meta.ID()
+			js.mu.RUnlock()
+			if peer := string(e.Data); peer == ourID {
+				mset.stop(true, false)
+			}
+			return nil
 		}
 	}
 	return nil
@@ -1863,7 +1887,14 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, sa *streamAssignme
 		resp.Error = jsError(err)
 		s.sendAPIErrResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
 	} else {
-		resp.StreamInfo = &StreamInfo{Created: mset.createdTime(), State: mset.state(), Config: mset.config(), Cluster: js.clusterInfo(mset.raftGroup()), Mirror: mset.mirrorInfo(), Sources: mset.sourcesInfo()}
+		resp.StreamInfo = &StreamInfo{
+			Created: mset.createdTime(),
+			State:   mset.state(),
+			Config:  mset.config(),
+			Cluster: js.clusterInfo(mset.raftGroup()),
+			Mirror:  mset.mirrorInfo(),
+			Sources: mset.sourcesInfo(),
+		}
 		s.sendAPIResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
 	}
 }
@@ -1948,7 +1979,7 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 		// If we are restoring, process that first.
 		if sa.Restore != nil {
 			// We are restoring a stream here.
-			restoreDoneCh := s.processStreamRestore(sa.Client, acc, sa.Config.Name, _EMPTY_, sa.Reply, _EMPTY_)
+			restoreDoneCh := s.processStreamRestore(sa.Client, acc, sa.Config, _EMPTY_, sa.Reply, _EMPTY_)
 			s.startGoRoutine(func() {
 				defer s.grWG.Done()
 				select {
@@ -2837,6 +2868,9 @@ func (js *jetStream) startUpdatesSub() {
 	if cc.stepdown == nil {
 		cc.stepdown, _ = s.systemSubscribe(JSApiLeaderStepDown, _EMPTY_, false, c, s.jsLeaderStepDownRequest)
 	}
+	if cc.peerRemove == nil {
+		cc.peerRemove, _ = s.systemSubscribe(JSApiRemoveServer, _EMPTY_, false, c, s.jsLeaderServerRemoveRequest)
+	}
 }
 
 // Lock should be held.
@@ -2854,6 +2888,10 @@ func (js *jetStream) stopUpdatesSub() {
 		cc.s.sysUnsubscribe(cc.stepdown)
 		cc.stepdown = nil
 	}
+	if cc.peerRemove != nil {
+		cc.s.sysUnsubscribe(cc.peerRemove)
+		cc.peerRemove = nil
+	}
 }
 
 func (js *jetStream) processLeaderChange(isLeader bool) {
@@ -3058,6 +3096,7 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 	var resp = JSApiStreamUpdateResponse{ApiResponse: ApiResponse{Type: JSApiStreamUpdateResponseType}}
 
 	osa := js.streamAssignment(acc.Name, cfg.Name)
+
 	if osa == nil {
 		resp.Error = jsNotFoundError(ErrJetStreamStreamNotFound)
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
@@ -3077,6 +3116,17 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
 		return
 	}
+	// Check for cluster changes that we want to error on.
+	if newCfg.Replicas != len(osa.Group.Peers) {
+		resp.Error = &ApiError{Code: 400, Description: "Replicas configuration can not be updated"}
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+		return
+	}
+	if !reflect.DeepEqual(newCfg.Mirror, osa.Config.Mirror) {
+		resp.Error = &ApiError{Code: 400, Description: "Mirror configuration can not be updated"}
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+		return
+	}
 
 	sa := &streamAssignment{Group: osa.Group, Config: newCfg, Subject: subject, Reply: reply, Client: ci}
 	cc.meta.Propose(encodeUpdateStreamAssignment(sa))
@@ -3221,11 +3271,13 @@ func (s *Server) jsClusteredStreamListRequest(acc *Account, ci *ClientInfo, offs
 		return
 	}
 
-	// Create an inbox for our responses and send out requests.
-	inbox := infoReplySubject()
+	// Create an inbox for our responses and send out our requests.
+	s.mu.Lock()
+	inbox := s.newRespInbox()
 	rc := make(chan *StreamInfo, len(streams))
 
-	rsub, _ := s.systemSubscribe(inbox, _EMPTY_, false, cc.c, func(_ *subscription, _ *client, _, reply string, msg []byte) {
+	// Store our handler.
+	s.sys.replies[inbox] = func(sub *subscription, _ *client, subject, _ string, msg []byte) {
 		var si StreamInfo
 		if err := json.Unmarshal(msg, &si); err != nil {
 			s.Warnf("Error unmarshaling clustered stream info response:%v", err)
@@ -3236,8 +3288,17 @@ func (s *Server) jsClusteredStreamListRequest(acc *Account, ci *ClientInfo, offs
 		default:
 			s.Warnf("Failed placing remote stream info result on internal channel")
 		}
-	})
-	defer s.sysUnsubscribe(rsub)
+	}
+	s.mu.Unlock()
+
+	// Cleanup after.
+	defer func() {
+		s.mu.Lock()
+		if s.sys != nil && s.sys.replies != nil {
+			delete(s.sys.replies, inbox)
+		}
+		s.mu.Unlock()
+	}()
 
 	// Send out our requests here.
 	for _, sa := range streams {
@@ -3336,9 +3397,12 @@ func (s *Server) jsClusteredConsumerListRequest(acc *Account, ci *ClientInfo, of
 	}
 
 	// Create an inbox for our responses and send out requests.
-	inbox := infoReplySubject()
+	s.mu.Lock()
+	inbox := s.newRespInbox()
 	rc := make(chan *ConsumerInfo, len(consumers))
-	rsub, _ := s.systemSubscribe(inbox, _EMPTY_, false, cc.c, func(_ *subscription, _ *client, _, reply string, msg []byte) {
+
+	// Store our handler.
+	s.sys.replies[inbox] = func(sub *subscription, _ *client, subject, _ string, msg []byte) {
 		var ci ConsumerInfo
 		if err := json.Unmarshal(msg, &ci); err != nil {
 			s.Warnf("Error unmarshaling clustered consumer info response:%v", err)
@@ -3349,8 +3413,17 @@ func (s *Server) jsClusteredConsumerListRequest(acc *Account, ci *ClientInfo, of
 		default:
 			s.Warnf("Failed placing consumer info result on internal chan")
 		}
-	})
-	defer s.sysUnsubscribe(rsub)
+	}
+	s.mu.Unlock()
+
+	// Cleanup after.
+	defer func() {
+		s.mu.Lock()
+		if s.sys != nil && s.sys.replies != nil {
+			delete(s.sys.replies, inbox)
+		}
+		s.mu.Unlock()
+	}()
 
 	for _, ca := range consumers {
 		isubj := fmt.Sprintf(clusterConsumerInfoT, ca.Client.serviceAccount(), stream, ca.Name)
@@ -3557,6 +3630,10 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 	// We need to set the ephemeral here before replicating.
 	var oname string
 	if !isDurableConsumer(cfg) {
+		// We chose to have ephemerals be R=1.
+		rg.Peers = []string{rg.Preferred}
+		rg.Name = groupNameForConsumer(rg.Peers, rg.Storage)
+		// Make sure name is unique.
 		for {
 			oname = createConsumerName()
 			if sa.consumers != nil {
@@ -3944,11 +4021,13 @@ RETRY:
 		}
 	}
 
-	type mr struct {
+	// Used to transfer message from the wire to another Go routine internally.
+	type im struct {
 		msg   []byte
 		reply string
 	}
-	msgsC := make(chan *mr, 32768)
+
+	msgsC := make(chan *im, 32768)
 
 	// Send our catchup request here.
 	reply := syncReplySubject()
@@ -3956,7 +4035,7 @@ RETRY:
 		// Make copies - https://github.com/go101/go101/wiki
 		// TODO(dlc) - Since we are using a buffer from the inbound client/route.
 		select {
-		case msgsC <- &mr{msg: append(msg[:0:0], msg...), reply: reply}:
+		case msgsC <- &im{append(msg[:0:0], msg...), reply}:
 		default:
 			s.Warnf("Failed to place catchup message onto internal channel: %d pending", len(msgsC))
 			return
@@ -4094,16 +4173,18 @@ func (js *jetStream) clusterInfo(rg *raftGroup) *ClusterInfo {
 
 func (mset *stream) handleClusterStreamInfoRequest(sub *subscription, c *client, subject, reply string, msg []byte) {
 	mset.mu.RLock()
-	if mset.client == nil {
-		mset.mu.RUnlock()
-		return
-	}
-	s, js, config := mset.srv, mset.srv.js, mset.cfg
+	sysc, js, config := mset.sysc, mset.srv.js, mset.cfg
 	mset.mu.RUnlock()
 
-	si := &StreamInfo{Created: mset.createdTime(), State: mset.state(), Config: config, Cluster: js.clusterInfo(mset.raftGroup()), Sources: mset.sourcesInfo(), Mirror: mset.mirrorInfo()}
-	b, _ := json.Marshal(si)
-	s.sendInternalMsgLocked(reply, _EMPTY_, nil, b)
+	si := &StreamInfo{
+		Created: mset.createdTime(),
+		State:   mset.state(),
+		Config:  config,
+		Cluster: js.clusterInfo(mset.raftGroup()),
+		Sources: mset.sourcesInfo(),
+		Mirror:  mset.mirrorInfo(),
+	}
+	sysc.sendInternalMsg(reply, _EMPTY_, nil, si)
 }
 
 func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
diff --git a/server/jetstream_events.go b/server/jetstream_events.go
index 45ebb1c2..0bd95894 100644
--- a/server/jetstream_events.go
+++ b/server/jetstream_events.go
@@ -215,3 +215,14 @@ type JSServerOutOfSpaceAdvisory struct {
 	Stream   string `json:"stream,omitempty"`
 	Cluster  string `json:"cluster"`
 }
+
+// JSServerRemovedAdvisoryType is sent when the server has been removed and JS disabled.
+const JSServerRemovedAdvisoryType = "io.nats.jetstream.advisory.v1.server_removed"
+
+// JSServerOutOfSpaceAdvisory indicates that a stream has lost quorum and is stalled.
+type JSServerRemovedAdvisory struct {
+	TypedEvent
+	Server   string `json:"server"`
+	ServerID string `json:"server_id"`
+	Cluster  string `json:"cluster"`
+}
diff --git a/server/memstore.go b/server/memstore.go
index 66005115..3fefd197 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -98,7 +98,10 @@ func (ms *memStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts int
 	}
 
 	if seq != ms.state.LastSeq+1 {
-		return ErrSequenceMismatch
+		if seq > 0 {
+			return ErrSequenceMismatch
+		}
+		seq = ms.state.LastSeq + 1
 	}
 
 	// Adjust first if needed.
diff --git a/server/raft.go b/server/raft.go
index 043a8ab0..9c304d25 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -173,6 +173,7 @@ type raft struct {
 
 	// Channels
 	propc    chan *Entry
+	entryc   chan *appendEntry
 	applyc   chan *CommittedEntry
 	sendq    chan *pubMsg
 	quit     chan struct{}
@@ -338,8 +339,9 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 		wpsch:    make(chan struct{}, 1),
 		reqs:     make(chan *voteRequest, 8),
 		votes:    make(chan *voteResponse, 32),
-		propc:    make(chan *Entry, 256),
-		applyc:   make(chan *CommittedEntry, 512),
+		propc:    make(chan *Entry, 8192),
+		entryc:   make(chan *appendEntry, 8192),
+		applyc:   make(chan *CommittedEntry, 8192),
 		leadc:    make(chan bool, 8),
 		stepdown: make(chan string, 8),
 	}
@@ -782,13 +784,13 @@ func (n *raft) InstallSnapshot(data []byte) error {
 
 	// Remember our latest snapshot file.
 	n.snapfile = sfile
+	n.Unlock()
+
 	_, err := n.wal.Compact(snap.lastIndex)
 	if err != nil {
-		n.setWriteErrLocked(err)
-		n.Unlock()
+		n.setWriteErr(err)
 		return err
 	}
-	n.Unlock()
 
 	psnaps, _ := ioutil.ReadDir(snapDir)
 	// Remove any old snapshots.
@@ -1308,6 +1310,8 @@ func (n *raft) runAsFollower() {
 	for {
 		elect := n.electTimer()
 		select {
+		case ae := <-n.entryc:
+			n.processAppendEntry(ae, ae.sub)
 		case <-n.s.quitCh:
 			return
 		case <-n.quit:
@@ -1339,6 +1343,7 @@ type appendEntry struct {
 	entries []*Entry
 	// internal use only.
 	reply string
+	sub   *subscription
 	buf   []byte
 }
 
@@ -1412,7 +1417,7 @@ func (ae *appendEntry) encode() []byte {
 }
 
 // This can not be used post the wire level callback since we do not copy.
-func (n *raft) decodeAppendEntry(msg []byte, reply string) *appendEntry {
+func (n *raft) decodeAppendEntry(msg []byte, sub *subscription, reply string) *appendEntry {
 	if len(msg) < appendEntryBaseLen {
 		return nil
 	}
@@ -1435,6 +1440,7 @@ func (n *raft) decodeAppendEntry(msg []byte, reply string) *appendEntry {
 		ri += int(le)
 	}
 	ae.reply = reply
+	ae.sub = sub
 	ae.buf = msg
 	return ae
 }
@@ -1596,6 +1602,8 @@ func (n *raft) runAsLeader() {
 		case newLeader := <-n.stepdown:
 			n.switchToFollower(newLeader)
 			return
+		case ae := <-n.entryc:
+			n.processAppendEntry(ae, ae.sub)
 		}
 	}
 }
@@ -1816,7 +1824,7 @@ func (n *raft) loadEntry(index uint64) (*appendEntry, error) {
 	if err != nil {
 		return nil, err
 	}
-	return n.decodeAppendEntry(msg, _EMPTY_), nil
+	return n.decodeAppendEntry(msg, nil, _EMPTY_), nil
 }
 
 // applyCommit will update our commit index and apply the entry to the apply chan.
@@ -2027,6 +2035,8 @@ func (n *raft) runAsCandidate() {
 			}
 		case vreq := <-n.reqs:
 			n.processVoteRequest(vreq)
+		case ae := <-n.entryc:
+			n.processAppendEntry(ae, ae.sub)
 		case newLeader := <-n.stepdown:
 			n.switchToFollower(newLeader)
 			return
@@ -2034,14 +2044,16 @@ func (n *raft) runAsCandidate() {
 	}
 }
 
-// handleAppendEntry handles an append entry from the wire. We can't rely on msg being available
-// past this callback so will do a bunch of processing here to avoid copies, channels etc.
+// handleAppendEntry handles an append entry from the wire.
 func (n *raft) handleAppendEntry(sub *subscription, c *client, subject, reply string, msg []byte) {
-	ae := n.decodeAppendEntry(msg, reply)
-	if ae == nil {
-		return
+	msg = append(msg[:0:0], msg...)
+	if ae := n.decodeAppendEntry(msg, sub, reply); ae != nil {
+		select {
+		case n.entryc <- ae:
+		default:
+			n.warn("AppendEntry failed to be placed on internal channel")
+		}
 	}
-	n.processAppendEntry(ae, sub)
 }
 
 // Lock should be held.
@@ -2430,7 +2442,7 @@ func (n *raft) handleAppendEntryResponse(sub *subscription, c *client, subject,
 }
 
 func (n *raft) buildAppendEntry(entries []*Entry) *appendEntry {
-	return &appendEntry{n.id, n.term, n.commit, n.pterm, n.pindex, entries, _EMPTY_, nil}
+	return &appendEntry{n.id, n.term, n.commit, n.pterm, n.pindex, entries, _EMPTY_, nil, nil}
 }
 
 // Store our append entry to our WAL.
diff --git a/server/server.go b/server/server.go
index 9ba85932..4210c861 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1650,6 +1650,10 @@ func (s *Server) Shutdown() {
 	// Transfer off any raft nodes that we are a leader by shutting them all down.
 	s.shutdownRaftNodes()
 
+	// This is for clustered JetStream and ephemeral consumers.
+	// No-op if not clustered or not running JetStream.
+	s.migrateEphemerals()
+
 	// Shutdown the eventing system as needed.
 	// This is done first to send out any messages for
 	// account status. We will also clean up any
diff --git a/server/stream.go b/server/stream.go
index dc2b0417..140d7231 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -163,13 +163,14 @@ type stream struct {
 }
 
 type sourceInfo struct {
-	name string
-	sub  *subscription
-	sseq uint64
-	dseq uint64
-	lag  uint64
-	err  *ApiError
-	last time.Time
+	name  string
+	sub   *subscription
+	msgsC chan *imr
+	sseq  uint64
+	dseq  uint64
+	lag   uint64
+	err   *ApiError
+	last  time.Time
 }
 
 // Headers for published messages.
@@ -835,6 +836,32 @@ func (mset *stream) update(config *StreamConfig) error {
 			// Let it fire right away, it will adjust properly on purge.
 			mset.ddtmr.Reset(time.Microsecond)
 		}
+
+		// Check for Sources.
+		if len(cfg.Sources) > 0 || len(ocfg.Sources) > 0 {
+			current := make(map[string]struct{})
+			for _, s := range ocfg.Sources {
+				current[s.Name] = struct{}{}
+			}
+			for _, s := range cfg.Sources {
+				if _, ok := current[s.Name]; !ok {
+					if mset.sources == nil {
+						mset.sources = make(map[string]*sourceInfo)
+					}
+					mset.cfg.Sources = append(mset.cfg.Sources, s)
+					si := &sourceInfo{name: s.Name}
+					mset.sources[s.Name] = si
+					mset.setStartingSequenceForSource(s.Name)
+					mset.setSourceConsumer(s.Name, si.sseq+1)
+				}
+				delete(current, s.Name)
+			}
+			// What is left in current needs to be deleted.
+			for sname := range current {
+				mset.cancelSourceConsumer(sname)
+				delete(mset.sources, sname)
+			}
+		}
 	}
 
 	// Now update config and store's version of our config.
@@ -954,38 +981,75 @@ func (mset *stream) processClusteredMirrorMsg(subject string, hdr, msg []byte, s
 	return mset.node.Propose(encodeStreamMsg(subject, _EMPTY_, hdr, msg, seq, ts))
 }
 
+// Will run as a Go routine to process messages.
+func (mset *stream) processMirrorMsgs() {
+	s := mset.srv
+	defer s.grWG.Done()
+
+	// Grab stream quit channel.
+	mset.mu.RLock()
+	msgsC := mset.mirror.msgsC
+	qch := mset.qch
+	mset.mu.RUnlock()
+
+	for {
+		select {
+		case <-s.quitCh:
+			return
+		case <-qch:
+			return
+		case mr, ok := <-msgsC:
+			if !ok {
+				return
+			}
+			mset.processInboundMirrorMsg(mr)
+		}
+	}
+}
+
+// Used to transfer message from the wire to another Go routine internally.
+type imr struct {
+	subject string
+	reply   string
+	hdr     []byte
+	msg     []byte
+}
+
 // processInboundMirrorMsg handles processing messages bound for a stream.
-func (mset *stream) processInboundMirrorMsg(sub *subscription, c *client, subject, reply string, rmsg []byte) {
-	hdr, msg := c.msgParts(rmsg)
-	sseq, _, _, ts, pending := replyInfo(reply)
+func (mset *stream) processInboundMirrorMsg(mr *imr) {
+	sseq, _, _, ts, pending := replyInfo(mr.reply)
 
 	mset.mu.Lock()
-
-	// Ignore anything not current.
-	if mset.mirror == nil || sub != mset.mirror.sub {
+	if mset.mirror == nil {
 		mset.mu.Unlock()
 		return
 	}
 
 	// Mirror info tracking.
+	olag := mset.mirror.lag
 	mset.mirror.lag = pending
 	mset.mirror.last = time.Now()
-
 	isClustered := mset.node != nil
-	sendq := mset.sendq
 	mset.mu.Unlock()
 
 	s := mset.srv
 	var err error
 	if isClustered {
-		err = mset.processClusteredMirrorMsg(subject, hdr, msg, sseq-1, ts)
+		err = mset.processClusteredMirrorMsg(mr.subject, mr.hdr, mr.msg, sseq-1, ts)
 	} else {
-		err = mset.processJetStreamMsg(subject, reply, hdr, msg, sseq-1, ts)
+		err = mset.processJetStreamMsg(mr.subject, _EMPTY_, mr.hdr, mr.msg, sseq-1, ts)
 	}
 	if err != nil {
 		if err == errLastSeqMismatch {
 			// We may have missed messages, restart.
-			mset.resetMirrorConsumer()
+			if sseq <= mset.lastSeq() {
+				mset.mu.Lock()
+				mset.mirror.lag = olag
+				mset.mu.Unlock()
+				return
+			} else {
+				mset.resetMirrorConsumer()
+			}
 		} else {
 			s.Debugf("Got error processing JetStream mirror msg: %v", err)
 		}
@@ -993,8 +1057,6 @@ func (mset *stream) processInboundMirrorMsg(sub *subscription, c *client, subjec
 			s.Errorf("JetStream out of space, will be DISABLED")
 			s.DisableJetStream()
 		}
-	} else if sendq != nil && reply != _EMPTY_ {
-		sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, nil, nil, 0}
 	}
 }
 
@@ -1004,15 +1066,22 @@ func (mset *stream) setMirrorErr(err *ApiError) {
 	mset.mu.Unlock()
 }
 
-const mirrorMaxAckPending = 64
+const sourceMaxAckPending = 32768
 
 func (mset *stream) cancelMirrorConsumer() {
 	mset.mu.Lock()
 	defer mset.mu.Unlock()
-	if mset.mirror != nil && mset.mirror.sub != nil {
+	if mset.mirror == nil {
+		return
+	}
+	if mset.mirror.sub != nil {
 		mset.unsubscribe(mset.mirror.sub)
 		mset.mirror.sub = nil
 	}
+	if mset.mirror.msgsC != nil {
+		close(mset.mirror.msgsC)
+		mset.mirror.msgsC = nil
+	}
 }
 
 func (mset *stream) retryMirrorConsumer() {
@@ -1042,15 +1111,37 @@ func (mset *stream) setupMirrorConsumer() error {
 	}
 
 	// Reset
-	if mset.mirror != nil && mset.mirror.sub != nil {
-		mset.unsubscribe(mset.mirror.sub)
-		mset.mirror.sub = nil
+	if mset.mirror != nil {
+		if mset.mirror.sub != nil {
+			mset.unsubscribe(mset.mirror.sub)
+			mset.mirror.sub = nil
+		}
+		if mset.mirror.msgsC != nil {
+			close(mset.mirror.msgsC)
+			mset.mirror.msgsC = nil
+		}
 	}
-	sub, err := mset.subscribeInternal(syncSubject("$JS.M"), mset.processInboundMirrorMsg)
+
+	mset.mirror = &sourceInfo{name: mset.cfg.Mirror.Name, msgsC: make(chan *imr, sourceMaxAckPending)}
+	sub, err := mset.subscribeInternal(syncSubject("$JS.M"), func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+		mset.mu.RLock()
+		// Ignore anything not current.
+		if mset.mirror == nil || sub != mset.mirror.sub || mset.mirror.msgsC == nil {
+			mset.mu.RUnlock()
+			return
+		}
+		msgsC := mset.mirror.msgsC
+		mset.mu.RUnlock()
+
+		hdr, msg := c.msgParts(append(rmsg[:0:0], rmsg...)) // Need to copy.
+		msgsC <- &imr{subject, reply, hdr, msg}
+	})
 	if err != nil {
+		mset.mirror = nil
 		return err
 	}
-	mset.mirror = &sourceInfo{name: mset.cfg.Mirror.Name, sub: sub}
+	mset.mirror.sub = sub
+	mset.srv.startGoRoutine(func() { mset.processMirrorMsgs() })
 
 	// Now send off request to create/update our consumer. This will be all API based even in single server mode.
 	// We calculate durable names apriori so we do not need to save them off.
@@ -1069,10 +1160,10 @@ func (mset *stream) setupMirrorConsumer() error {
 			DeliverSubject: string(sub.subject),
 			DeliverPolicy:  DeliverByStartSequence,
 			OptStartSeq:    state.LastSeq,
-			AckPolicy:      AckExplicit,
+			AckPolicy:      AckNone,
 			AckWait:        48 * time.Hour,
 			MaxDeliver:     1,
-			MaxAckPending:  mirrorMaxAckPending,
+			RateLimit:      1 * 1024 * 1024 * 1024, // 1 Gbit
 		},
 	}
 	// Only use start optionals on first time.
@@ -1178,6 +1269,10 @@ func (mset *stream) setSourceConsumer(sname string, seq uint64) {
 	if si.sub != nil {
 		mset.unsubscribe(si.sub)
 	}
+	if si.msgsC != nil {
+		close(si.msgsC)
+		si.msgsC = nil
+	}
 	si.sseq, si.dseq = 0, 0
 
 	durable := mset.sourceDurable(sname)
@@ -1186,26 +1281,36 @@ func (mset *stream) setSourceConsumer(sname string, seq uint64) {
 	subject := fmt.Sprintf(JSApiConsumerDeleteT, sname, durable)
 	mset.sendq <- &jsPubMsg{subject, _EMPTY_, _EMPTY_, nil, nil, nil, 0}
 
-	dsubj := syncSubject("$JS.S")
-	sub, err := mset.subscribeInternal(dsubj, func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
-		mset.processInboundSourceMsg(si, sub, c, subject, reply, rmsg)
+	si.msgsC = make(chan *imr, sourceMaxAckPending)
+	sub, err := mset.subscribeInternal(syncSubject("$JS.S"), func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+		mset.mu.RLock()
+		defer mset.mu.RUnlock()
+		// Ignore anything not current.
+		if sub != si.sub || si.msgsC == nil {
+			return
+		}
+		hdr, msg := c.msgParts(append(rmsg[:0:0], rmsg...)) // Need to copy.
+		si.msgsC <- &imr{subject, reply, hdr, msg}
 	})
+
 	if err != nil {
 		si.err = jsError(err)
 		si.sub = nil
 		return
 	}
+
 	si.sub = sub
+	mset.srv.startGoRoutine(func() { mset.processSourceMsgs(si) })
 
 	req := &CreateConsumerRequest{
 		Stream: sname,
 		Config: ConsumerConfig{
 			Durable:        durable,
-			DeliverSubject: dsubj,
-			AckPolicy:      AckExplicit,
+			DeliverSubject: string(sub.subject),
+			AckPolicy:      AckNone,
 			AckWait:        48 * time.Hour,
 			MaxDeliver:     1,
-			MaxAckPending:  mirrorMaxAckPending,
+			RateLimit:      1 * 1024 * 1024 * 1024, // 1 Gbit
 		},
 	}
 	// If starting, check any configs.
@@ -1277,16 +1382,45 @@ func (mset *stream) setSourceConsumer(sname string, seq uint64) {
 	}()
 }
 
+func (mset *stream) processSourceMsgs(si *sourceInfo) {
+	s := mset.srv
+	defer s.grWG.Done()
+
+	if si == nil {
+		return
+	}
+
+	// Grab stream quit channel.
+	mset.mu.RLock()
+	msgsC := si.msgsC
+	qch := mset.qch
+	mset.mu.RUnlock()
+
+	for {
+		select {
+		case <-s.quitCh:
+			return
+		case <-qch:
+			return
+		case mr, ok := <-msgsC:
+			if !ok {
+				return
+			}
+			mset.processInboundSourceMsg(si, mr)
+		}
+	}
+}
+
 // processInboundSourceMsg handles processing other stream messages bound for this stream.
-func (mset *stream) processInboundSourceMsg(si *sourceInfo, sub *subscription, c *client, subject, reply string, rmsg []byte) {
-	sseq, dseq, dc, _, pending := replyInfo(reply)
+func (mset *stream) processInboundSourceMsg(si *sourceInfo, mr *imr) {
+	sseq, dseq, dc, _, pending := replyInfo(mr.reply)
 
-	// Tracking is done here.
-	mset.mu.Lock()
-	if si == nil || sub != si.sub || dc > 1 {
-		mset.mu.Unlock()
+	if dc > 1 {
 		return
 	}
+
+	// Tracking is done here.
+	mset.mu.Lock()
 	if dseq == si.dseq+1 {
 		si.dseq++
 		si.sseq++
@@ -1294,40 +1428,41 @@ func (mset *stream) processInboundSourceMsg(si *sourceInfo, sub *subscription, c
 			si.sseq = sseq
 		}
 	} else {
-		mset.setSourceConsumer(si.name, si.sseq+1)
+		if dseq > si.dseq {
+			mset.setSourceConsumer(si.name, si.sseq+1)
+		}
 		mset.mu.Unlock()
 		return
 	}
 
 	si.last = time.Now()
-	si.lag = pending
+	if pending == 0 {
+		si.lag = 0
+	} else {
+		si.lag = pending - 1
+	}
 	isClustered := mset.node != nil
-	sendq := mset.sendq
 	mset.mu.Unlock()
 
-	// Hold onto the origin reply which has all the metadata.
-	hdr, msg := c.msgParts(rmsg)
+	hdr, msg := mr.hdr, mr.msg
+
 	// If we are daisy chained here make sure to remove the original one.
 	if len(hdr) > 0 {
 		hdr = removeHeaderIfPresent(hdr, JSStreamSource)
 	}
-	hdr = genHeader(hdr, JSStreamSource, reply)
+	// Hold onto the origin reply which has all the metadata.
+	hdr = genHeader(hdr, JSStreamSource, mr.reply)
 
 	var err error
 	// If we are clustered we need to propose this message to the underlying raft group.
 	if isClustered {
-		err = mset.processClusteredInboundMsg(subject, _EMPTY_, hdr, msg)
+		err = mset.processClusteredInboundMsg(mr.subject, _EMPTY_, hdr, msg)
 	} else {
-		err = mset.processJetStreamMsg(subject, _EMPTY_, hdr, msg, 0, 0)
+		err = mset.processJetStreamMsg(mr.subject, _EMPTY_, hdr, msg, 0, 0)
 	}
-	if err != nil {
+	if err != nil && err != errLastSeqMismatch {
 		mset.srv.Errorf("JetStream got an error processing inbound source msg: %v", err)
 	}
-
-	// Ack
-	if sendq != nil && reply != _EMPTY_ {
-		sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, nil, nil, 0}
-	}
 }
 
 func streamAndSeq(subject string) (string, uint64) {
@@ -1906,7 +2041,7 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	}
 
 	// Grab timestamp if not already set.
-	if ts == 0 {
+	if ts == 0 && lseq > 0 {
 		ts = time.Now().UnixNano()
 	}
 
@@ -1940,12 +2075,10 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	mset.mu.Unlock()
 
 	// Store actual msg.
-	if seq == 0 && ts == 0 {
-		seq, _, err = store.StoreMsg(subject, hdr, msg)
+	if lseq == 0 && ts == 0 {
+		seq, ts, err = store.StoreMsg(subject, hdr, msg)
 	} else {
-		if seq == 0 {
-			seq = olseq + 1
-		}
+		seq = lseq + 1
 		err = store.StoreRawMsg(subject, hdr, msg, seq, ts)
 	}
 
@@ -1972,11 +2105,16 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 			resp.Error = &ApiError{Code: 400, Description: "resource limits exceeded for account"}
 			response, _ = json.Marshal(resp)
 		}
+		// If we did not succeed put those values back.
+		mset.mu.Lock()
+		mset.lseq = olseq
+		mset.lmsgId = olmsgId
+		mset.mu.Unlock()
 		store.RemoveMsg(seq)
 		seq = 0
 	} else {
 		// If we have a msgId make sure to save.
-		if msgId != "" {
+		if msgId != _EMPTY_ {
 			mset.storeMsgId(&ddentry{msgId, seq, ts})
 		}
 		if canRespond {
@@ -1991,26 +2129,18 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	}
 
 	if err == nil && seq > 0 && numConsumers > 0 {
-		var _obs [4]*consumer
-		obs := _obs[:0]
-
 		mset.mu.Lock()
 		for _, o := range mset.consumers {
-			o.mu.RLock()
-			isLeader := o.isLeader()
-			o.mu.RUnlock()
-			if isLeader {
-				obs = append(obs, o)
-			}
-		}
-		mset.mu.Unlock()
-
-		for _, o := range obs {
-			o.incStreamPending(seq, subject)
-			if !o.deliverCurrentMsg(subject, hdr, msg, seq, ts) {
+			o.mu.Lock()
+			if o.isLeader() {
+				if o.isFilteredMatch(subject) {
+					o.sgap++
+				}
 				o.signalNewMessages()
 			}
+			o.mu.Unlock()
 		}
+		mset.mu.Unlock()
 	}
 
 	return err
@@ -2037,7 +2167,7 @@ type StoredMsg struct {
 }
 
 // TODO(dlc) - Maybe look at onering instead of chan - https://github.com/pltr/onering
-const msetSendQSize = 4096
+const msetSendQSize = 65536
 
 // This is similar to system semantics but did not want to overload the single system sendq,
 // or require system account when doing simple setup with jetstream.
@@ -2359,7 +2489,16 @@ func (mset *stream) snapshot(deadline time.Duration, checkMsgs, includeConsumers
 const snapsDir = "__snapshots__"
 
 // RestoreStream will restore a stream from a snapshot.
-func (a *Account) RestoreStream(stream string, r io.Reader) (*stream, error) {
+func (a *Account) RestoreStream(ncfg *StreamConfig, r io.Reader) (*stream, error) {
+	if ncfg == nil {
+		return nil, errors.New("nil config on stream restore")
+	}
+
+	cfg, err := checkStreamCfg(ncfg)
+	if err != nil {
+		return nil, err
+	}
+
 	_, jsa, err := a.checkForJetStream()
 	if err != nil {
 		return nil, err
@@ -2406,19 +2545,17 @@ func (a *Account) RestoreStream(stream string, r io.Reader) (*stream, error) {
 		}
 	}
 
-	// Check metadata
-	var cfg FileStreamInfo
+	// Check metadata.
+	// The cfg passed in will be the new identity for the stream.
+
+	var fcfg FileStreamInfo
 	b, err := ioutil.ReadFile(path.Join(sdir, JetStreamMetaFile))
 	if err != nil {
 		return nil, err
 	}
-	if err := json.Unmarshal(b, &cfg); err != nil {
+	if err := json.Unmarshal(b, &fcfg); err != nil {
 		return nil, err
 	}
-	// See if names match
-	if cfg.Name != stream {
-		return nil, fmt.Errorf("stream name [%q] does not match snapshot stream [%q]", stream, cfg.Name)
-	}
 
 	// See if this stream already exists.
 	if _, err := a.lookupStream(cfg.Name); err == nil {
@@ -2434,12 +2571,12 @@ func (a *Account) RestoreStream(stream string, r io.Reader) (*stream, error) {
 			return nil, err
 		}
 	}
-	mset, err := a.addStream(&cfg.StreamConfig)
+	mset, err := a.addStream(&cfg)
 	if err != nil {
 		return nil, err
 	}
-	if !cfg.Created.IsZero() {
-		mset.setCreatedTime(cfg.Created)
+	if !fcfg.Created.IsZero() {
+		mset.setCreatedTime(fcfg.Created)
 	}
 
 	// Now do consumers.
diff --git a/vendor/github.com/nats-io/nats.go/go.mod b/vendor/github.com/nats-io/nats.go/go.mod
index 076b7571..b60123e7 100644
--- a/vendor/github.com/nats-io/nats.go/go.mod
+++ b/vendor/github.com/nats-io/nats.go/go.mod
@@ -4,7 +4,7 @@ go 1.15
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/nats-io/nats-server/v2 v2.1.8-0.20210205154825-f7ab27f7dad4
+	github.com/nats-io/nats-server/v2 v2.1.8-0.20210227190344-51550e242af8
 	github.com/nats-io/nkeys v0.2.0
 	github.com/nats-io/nuid v1.0.1
 	google.golang.org/protobuf v1.23.0
diff --git a/vendor/github.com/nats-io/nats.go/go.sum b/vendor/github.com/nats-io/nats.go/go.sum
index 388f1f2f..a2f355de 100644
--- a/vendor/github.com/nats-io/nats.go/go.sum
+++ b/vendor/github.com/nats-io/nats.go/go.sum
@@ -7,6 +7,7 @@ github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/klauspost/compress v1.11.7 h1:0hzRabrMN4tSTvMfnL3SCv1ZGeAP23ynzodBgaHeMeg=
 github.com/klauspost/compress v1.11.7/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
@@ -20,17 +21,22 @@ github.com/nats-io/jwt/v2 v2.0.0-20200916203241-1f8ce17dff02/go.mod h1:vs+ZEjP+X
 github.com/nats-io/jwt/v2 v2.0.0-20201015190852-e11ce317263c/go.mod h1:vs+ZEjP+XKy8szkBmQwCB7RjYdIlMaPsFPs4VdS4bTQ=
 github.com/nats-io/jwt/v2 v2.0.0-20210125223648-1c24d462becc h1:pu+s4XC+bYnI0iD2vDtOl83zjCYUau/q6c83pEvsGZc=
 github.com/nats-io/jwt/v2 v2.0.0-20210125223648-1c24d462becc/go.mod h1:PuO5FToRL31ecdFqVjc794vK0Bj0CwzveQEDvkb7MoQ=
+github.com/nats-io/jwt/v2 v2.0.0-20210208203759-ff814ca5f813 h1:km4lLzT86NyJRhO++VqfP/vn5cbfm+E05i2bGdqDbrY=
+github.com/nats-io/jwt/v2 v2.0.0-20210208203759-ff814ca5f813/go.mod h1:PuO5FToRL31ecdFqVjc794vK0Bj0CwzveQEDvkb7MoQ=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200524125952-51ebd92a9093/go.mod h1:rQnBf2Rv4P9adtAs/Ti6LfFmVtFG6HLhl/H7cVshcJU=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200601203034-f8d6dd992b71/go.mod h1:Nan/1L5Sa1JRW+Thm4HNYcIDcVRFc5zK9OpSZeI2kk4=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20200929001935-7f44d075f7ad/go.mod h1:TkHpUIDETmTI7mrHN40D1pzxfzHZuGmtMbtb83TGVQw=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20201129161730-ebe63db3e3ed/go.mod h1:XD0zHR/jTXdZvWaQfS5mQgsXj6x12kMjKLyAk/cOGgY=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20210205154825-f7ab27f7dad4 h1:GStuc0W1rK45FSlpt3+7UTLzmRys2/6WSDuJFyzZ6Xg=
 github.com/nats-io/nats-server/v2 v2.1.8-0.20210205154825-f7ab27f7dad4/go.mod h1:kauGd7hB5517KeSqspW2U1Mz/jhPbTrE8eOXzUPk1m0=
+github.com/nats-io/nats-server/v2 v2.1.8-0.20210227190344-51550e242af8 h1:jPZZofsCevE2oJl3YexVw3drWOFdo8H4AWMb/1WcVoc=
+github.com/nats-io/nats-server/v2 v2.1.8-0.20210227190344-51550e242af8/go.mod h1:/QQ/dpqFavkNhVnjvMILSQ3cj5hlmhB66adlgNbjuoA=
 github.com/nats-io/nats.go v1.10.0/go.mod h1:AjGArbfyR50+afOUotNX2Xs5SYHf+CoOa5HH1eEl2HE=
 github.com/nats-io/nats.go v1.10.1-0.20200531124210-96f2130e4d55/go.mod h1:ARiFsjW9DVxk48WJbO3OSZ2DG8fjkMi7ecLmXoY/n9I=
 github.com/nats-io/nats.go v1.10.1-0.20200606002146-fc6fed82929a/go.mod h1:8eAIv96Mo9QW6Or40jUHejS7e4VwZ3VRYD6Sf0BTDp4=
 github.com/nats-io/nats.go v1.10.1-0.20201021145452-94be476ad6e0/go.mod h1:VU2zERjp8xmF+Lw2NH4u2t5qWZxwc7jB3+7HVMWQXPI=
 github.com/nats-io/nats.go v1.10.1-0.20210127212649-5b4924938a9a/go.mod h1:Sa3kLIonafChP5IF0b55i9uvGR10I3hPETFbi4+9kOI=
+github.com/nats-io/nats.go v1.10.1-0.20210211000709-75ded9c77585/go.mod h1:uBWnCKg9luW1g7hgzPxUjHFRI40EuTSX7RCzgnc74Jk=
 github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nkeys v0.1.4/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
 github.com/nats-io/nkeys v0.2.0 h1:WXKF7diOaPU9cJdLD7nuzwasQy9vT1tBqzXZZf3AMJM=
@@ -52,6 +58,7 @@ golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1 h1:NusfzzA6yGQ+ua51ck7E3omNUX/JuqbFSaRGqU8CcLI=
 golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
diff --git a/vendor/github.com/nats-io/nats.go/js.go b/vendor/github.com/nats-io/nats.go/js.go
index 7f23b7de..429959d9 100644
--- a/vendor/github.com/nats-io/nats.go/js.go
+++ b/vendor/github.com/nats-io/nats.go/js.go
@@ -857,12 +857,12 @@ func (js *js) getConsumerInfo(stream, consumer string) (*ConsumerInfo, error) {
 }
 
 func (m *Msg) checkReply() (*js, bool, error) {
-	if m.Reply == "" {
-		return nil, false, ErrMsgNoReply
-	}
 	if m == nil || m.Sub == nil {
 		return nil, false, ErrMsgNotBound
 	}
+	if m.Reply == "" {
+		return nil, false, ErrMsgNoReply
+	}
 	sub := m.Sub
 	sub.mu.Lock()
 	if sub.jsi == nil {
diff --git a/vendor/github.com/nats-io/nats.go/jsm.go b/vendor/github.com/nats-io/nats.go/jsm.go
index 286e02ab..fb7851e4 100644
--- a/vendor/github.com/nats-io/nats.go/jsm.go
+++ b/vendor/github.com/nats-io/nats.go/jsm.go
@@ -24,16 +24,16 @@ import (
 
 // JetStreamManager is the public interface for managing JetStream streams & consumers.
 type JetStreamManager interface {
-	// Create a stream.
+	// AddStream creates a stream.
 	AddStream(cfg *StreamConfig) (*StreamInfo, error)
 
-	// Update a stream.
+	// UpdateStream updates a stream.
 	UpdateStream(cfg *StreamConfig) (*StreamInfo, error)
 
-	// Delete a stream.
+	// DeleteStream deletes a stream.
 	DeleteStream(name string) error
 
-	// Stream information.
+	// StreamInfo retrieves information from a stream.
 	StreamInfo(stream string) (*StreamInfo, error)
 
 	// Purge stream messages.
@@ -45,16 +45,16 @@ type JetStreamManager interface {
 	// GetMsg retrieves a raw stream message stored in JetStream by sequence number.
 	GetMsg(name string, seq uint64) (*RawStreamMsg, error)
 
-	// DeleteMsg erases a message from a Stream.
+	// DeleteMsg erases a message from a stream.
 	DeleteMsg(name string, seq uint64) error
 
-	// Create a consumer.
+	// AddConsumer adds a consumer to a stream.
 	AddConsumer(stream string, cfg *ConsumerConfig) (*ConsumerInfo, error)
 
-	// Delete a consumer.
+	// DeleteConsumer deletes a consumer.
 	DeleteConsumer(stream, consumer string) error
 
-	// Consumer information.
+	// ConsumerInfo retrieves consumer information.
 	ConsumerInfo(stream, name string) (*ConsumerInfo, error)
 
 	// NewConsumerLister is used to return pages of ConsumerInfo objects.
@@ -83,6 +83,8 @@ type StreamConfig struct {
 	Template     string          `json:"template_owner,omitempty"`
 	Duplicates   time.Duration   `json:"duplicate_window,omitempty"`
 	Placement    *Placement      `json:"placement,omitempty"`
+	Mirror       *StreamSource   `json:"mirror,omitempty"`
+	Sources      []*StreamSource `json:"sources,omitempty"`
 }
 
 // Placement is used to guide placement of streams in clustered JetStream.
@@ -91,6 +93,14 @@ type Placement struct {
 	Tags    []string `json:"tags,omitempty"`
 }
 
+// StreamSource dictates how streams can source from other streams.
+type StreamSource struct {
+	Name          string     `json:"name"`
+	OptStartSeq   uint64     `json:"opt_start_seq,omitempty"`
+	OptStartTime  *time.Time `json:"opt_start_time,omitempty"`
+	FilterSubject string     `json:"filter_subject,omitempty"`
+}
+
 // apiError is included in all API responses if there was an error.
 type apiError struct {
 	Code        int    `json:"code"`
@@ -380,13 +390,22 @@ func (js *js) StreamInfo(stream string) (*StreamInfo, error) {
 
 // StreamInfo shows config and current state for this stream.
 type StreamInfo struct {
-	Config  StreamConfig `json:"config"`
-	Created time.Time    `json:"created"`
-	State   StreamState  `json:"state"`
-	Cluster *ClusterInfo `json:"cluster,omitempty"`
+	Config  StreamConfig        `json:"config"`
+	Created time.Time           `json:"created"`
+	State   StreamState         `json:"state"`
+	Cluster *ClusterInfo        `json:"cluster,omitempty"`
+	Mirror  *StreamSourceInfo   `json:"mirror,omitempty"`
+	Sources []*StreamSourceInfo `json:"sources,omitempty"`
+}
+
+// StreamSourceInfo shows information about an upstream stream source.
+type StreamSourceInfo struct {
+	Name   string        `json:"name"`
+	Lag    uint64        `json:"lag"`
+	Active time.Duration `json:"active"`
 }
 
-// StreamStats is information about the given stream.
+// StreamState is information about the given stream.
 type StreamState struct {
 	Msgs      uint64    `json:"messages"`
 	Bytes     uint64    `json:"bytes"`
diff --git a/vendor/modules.txt b/vendor/modules.txt
index b8401183..2bce34f8 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -7,7 +7,7 @@ github.com/minio/highwayhash
 # github.com/nats-io/jwt/v2 v2.0.0-20210208203759-ff814ca5f813
 ## explicit
 github.com/nats-io/jwt/v2
-# github.com/nats-io/nats.go v1.10.1-0.20210211000709-75ded9c77585
+# github.com/nats-io/nats.go v1.10.1-0.20210228004050-ed743748acac
 ## explicit
 github.com/nats-io/nats.go
 github.com/nats-io/nats.go/encoders/builtin


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 9ac0bd8e..0c0c9f61 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -810,6 +810,45 @@ func TestFileStoreCompact(t *testing.T) {
 	}
 }
 
+func TestFileStoreCompactPerf(t *testing.T) {
+	t.SkipNow()
+
+	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 8192, AsyncFlush: true}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	subj, msg := "foo", []byte("Hello World")
+	for i := 0; i < 100_000; i++ {
+		fs.StoreMsg(subj, nil, msg)
+	}
+	if state := fs.State(); state.Msgs != 100_000 {
+		t.Fatalf("Expected 1000000 msgs, got %d", state.Msgs)
+	}
+	start := time.Now()
+	n, err := fs.Compact(90_001)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	t.Logf("Took %v to compact\n", time.Since(start))
+
+	if n != 90_000 {
+		t.Fatalf("Expected to have purged 90_000 msgs, got %d", n)
+	}
+	state := fs.State()
+	if state.Msgs != 10_000 {
+		t.Fatalf("Expected 10_000 msgs, got %d", state.Msgs)
+	}
+	if state.FirstSeq != 90_001 {
+		t.Fatalf("Expected first seq of 90_001, got %d", state.FirstSeq)
+	}
+}
+
 func TestFileStoreStreamTruncate(t *testing.T) {
 	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
 	os.MkdirAll(storeDir, 0755)
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 8c95d294..32a37c02 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -474,6 +474,47 @@ func TestJetStreamClusterStreamUpdateSubjects(t *testing.T) {
 	}
 }
 
+func TestJetStreamClusterConsumerRedeliveredInfo(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	// Client based API
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	cfg := &nats.StreamConfig{Name: "TEST"}
+	if _, err := js.AddStream(cfg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	if _, err := js.Publish("TEST", []byte("CI")); err != nil {
+		t.Fatalf("Unexpected publish error: %v", err)
+	}
+
+	sub, _ := nc.SubscribeSync("R")
+	sub.AutoUnsubscribe(2)
+
+	ci, err := js.AddConsumer("TEST", &nats.ConsumerConfig{
+		DeliverSubject: "R",
+		AckPolicy:      nats.AckExplicitPolicy,
+		AckWait:        100 * time.Millisecond,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	checkSubsPending(t, sub, 2)
+	sub.Unsubscribe()
+
+	ci, err = js.ConsumerInfo("TEST", ci.Name)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if ci.NumRedelivered != 1 {
+		t.Fatalf("Expected 1 redelivered, got %d", ci.NumRedelivered)
+	}
+}
+
 func TestJetStreamClusterConsumerState(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
@@ -736,7 +777,7 @@ func TestJetStreamClusterMetaSnapshotsMultiChange(t *testing.T) {
 }
 
 func TestJetStreamClusterStreamSynchedTimeStamps(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "R3S", 5)
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
 
 	// Client based API
@@ -744,7 +785,7 @@ func TestJetStreamClusterStreamSynchedTimeStamps(t *testing.T) {
 	nc, js := jsClientConnect(t, s)
 	defer nc.Close()
 
-	_, err := js.AddStream(&nats.StreamConfig{Name: "foo", Replicas: 3})
+	_, err := js.AddStream(&nats.StreamConfig{Name: "foo", Storage: nats.MemoryStorage, Replicas: 3})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -773,21 +814,15 @@ func TestJetStreamClusterStreamSynchedTimeStamps(t *testing.T) {
 	c.waitOnLeader()
 	c.waitOnStreamLeader("$G", "foo")
 
-	s = c.randomServer()
-	nc, js = jsClientConnect(t, s)
+	nc, js = jsClientConnect(t, c.leader())
 	defer nc.Close()
 
-	sub, err = js.SubscribeSync("foo")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	m, err = sub.NextMsg(time.Second)
+	sm, err := js.GetMsg("foo", 1)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	meta2, _ := m.MetaData()
-	if meta.Timestamp != meta2.Timestamp {
-		t.Fatalf("Expected same timestamps, got %v vs %v", meta.Timestamp, meta2.Timestamp)
+	if !sm.Time.Equal(meta.Timestamp) {
+		t.Fatalf("Expected same timestamps, got %v vs %v", sm.Time, meta.Timestamp)
 	}
 }
 
@@ -1248,6 +1283,53 @@ func TestJetStreamClusterStreamUpdate(t *testing.T) {
 	}
 }
 
+func TestJetStreamClusterStreamExtendedUpdates(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()
+
+	cfg := &nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	}
+	if _, err := js.AddStream(cfg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	updateStream := func() *nats.StreamInfo {
+		si, err := js.UpdateStream(cfg)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		return si
+	}
+
+	expectError := func() {
+		if _, err := js.UpdateStream(cfg); err == nil {
+			t.Fatalf("Expected error and got none")
+		}
+	}
+
+	// Subjects
+	cfg.Subjects = []string{"bar", "baz"}
+	if si := updateStream(); !reflect.DeepEqual(si.Config.Subjects, cfg.Subjects) {
+		t.Fatalf("Did not get expected stream info: %+v", si)
+	}
+	// Make sure these error for now.
+	// R factor changes
+	cfg.Replicas = 1
+	expectError()
+	// Mirror changes
+	cfg.Replicas = 3
+	cfg.Mirror = &nats.StreamSource{Name: "ORDERS"}
+	expectError()
+}
+
 func TestJetStreamClusterDoubleAdd(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R32", 2)
 	defer c.shutdown()
@@ -1849,13 +1931,13 @@ func TestJetStreamClusterInterestRetentionWithFilteredConsumers(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
-	fsub, err := js.SubscribeSync("foo")
+	fsub, err := js.SubscribeSync("foo", nats.Durable("d1"))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 	defer fsub.Unsubscribe()
 
-	bsub, err := js.SubscribeSync("bar")
+	bsub, err := js.SubscribeSync("bar", nats.Durable("d2"))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1988,6 +2070,55 @@ func TestJetStreamClusterEphemeralConsumerCleanup(t *testing.T) {
 	})
 }
 
+func TestJetStreamClusterEphemeralConsumersNotReplicated(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()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "foo", Replicas: 3})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	sub, err := js.SubscribeSync("foo")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	ci, _ := sub.ConsumerInfo()
+	if ci == nil {
+		t.Fatalf("Unexpected error: no consumer info")
+	}
+
+	if _, err = js.Publish("foo", []byte("OK")); err != nil {
+		t.Fatalf("Unexpected publish error: %v", err)
+	}
+	checkSubsPending(t, sub, 1)
+
+	if ci.Cluster == nil || len(ci.Cluster.Replicas) != 0 {
+		t.Fatalf("Expected ephemeral to be R=1, got %+v", ci.Cluster)
+	}
+	scl := c.serverByName(ci.Cluster.Leader)
+	if scl == nil {
+		t.Fatalf("Could not select server where ephemeral consumer is running")
+	}
+
+	// Test migrations. If we are also metadata leader will not work so skip.
+	if scl == c.leader() {
+		return
+	}
+
+	scl.Shutdown()
+	c.waitOnStreamLeader("$G", "foo")
+
+	if _, err = js.Publish("foo", []byte("OK")); err != nil {
+		t.Fatalf("Unexpected publish error: %v", err)
+	}
+	checkSubsPending(t, sub, 2)
+}
+
 func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
@@ -2255,6 +2386,162 @@ func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 	}
 }
 
+func TestJetStreamClusterUserSnapshotAndRestoreConfigChanges(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	// Client based API
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	// FIXME(dlc) - Do case with R=1
+	cfg := &nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 2,
+	}
+
+	if _, err := js.AddStream(cfg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	toSend := 10
+	for i := 0; i < toSend; i++ {
+		if _, err := js.Publish("foo", []byte("OK")); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+
+	getSnapshot := func() ([]byte, *StreamState) {
+		t.Helper()
+		sreq := &JSApiStreamSnapshotRequest{
+			DeliverSubject: nats.NewInbox(),
+			ChunkSize:      1024,
+		}
+
+		req, _ := json.Marshal(sreq)
+		rmsg, err := nc.Request(fmt.Sprintf(JSApiStreamSnapshotT, "TEST"), req, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error on snapshot request: %v", err)
+		}
+
+		var resp JSApiStreamSnapshotResponse
+		json.Unmarshal(rmsg.Data, &resp)
+		if resp.Error != nil {
+			t.Fatalf("Did not get correct error response: %+v", resp.Error)
+		}
+
+		var snapshot []byte
+		done := make(chan bool)
+
+		sub, _ := nc.Subscribe(sreq.DeliverSubject, func(m *nats.Msg) {
+			// EOF
+			if len(m.Data) == 0 {
+				done <- true
+				return
+			}
+			// Could be writing to a file here too.
+			snapshot = append(snapshot, m.Data...)
+			// Flow ack
+			m.Respond(nil)
+		})
+		defer sub.Unsubscribe()
+
+		// Wait to receive the snapshot.
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Fatalf("Did not receive our snapshot in time")
+		}
+		return snapshot, resp.State
+	}
+
+	restore := func(cfg *StreamConfig, state *StreamState, snap []byte) *nats.StreamInfo {
+		rreq := &JSApiStreamRestoreRequest{
+			Config: *cfg,
+			State:  *state,
+		}
+		req, err := json.Marshal(rreq)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		rmsg, err := nc.Request(fmt.Sprintf(JSApiStreamRestoreT, cfg.Name), req, 5*time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		var rresp JSApiStreamRestoreResponse
+		json.Unmarshal(rmsg.Data, &rresp)
+		if rresp.Error != nil {
+			t.Fatalf("Got an unexpected error response: %+v", rresp.Error)
+		}
+		// Send our snapshot back in to restore the stream.
+		// Can be any size message.
+		var chunk [1024]byte
+		for r := bytes.NewReader(snap); ; {
+			n, err := r.Read(chunk[:])
+			if err != nil {
+				break
+			}
+			nc.Request(rresp.DeliverSubject, chunk[:n], time.Second)
+		}
+		rmsg, err = nc.Request(rresp.DeliverSubject, nil, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		rresp.Error = nil
+		json.Unmarshal(rmsg.Data, &rresp)
+		if rresp.Error != nil {
+			t.Fatalf("Got an unexpected error response: %+v", rresp.Error)
+		}
+		si, err := js.StreamInfo(cfg.Name)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		return si
+	}
+
+	snap, state := getSnapshot()
+
+	if err := js.DeleteStream("TEST"); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Change name.
+	ncfg := &StreamConfig{
+		Name:     "TEST2",
+		Subjects: []string{"foo"},
+		Storage:  FileStorage,
+		Replicas: 2,
+	}
+	if si := restore(ncfg, state, snap); si.Config.Name != "TEST2" {
+		t.Fatalf("Did not get expected stream info: %+v", si)
+	}
+	if err := js.DeleteStream("TEST2"); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	// Now change subjects.
+	ncfg.Subjects = []string{"bar", "baz"}
+	if si := restore(ncfg, state, snap); !reflect.DeepEqual(si.Config.Subjects, ncfg.Subjects) {
+		t.Fatalf("Did not get expected stream info: %+v", si)
+	}
+	if err := js.DeleteStream("TEST2"); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	// Storage
+	ncfg.Storage = MemoryStorage
+	if si := restore(ncfg, state, snap); !reflect.DeepEqual(si.Config.Subjects, ncfg.Subjects) {
+		t.Fatalf("Did not get expected stream info: %+v", si)
+	}
+	if err := js.DeleteStream("TEST2"); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	// Now replicas
+	ncfg.Replicas = 3
+	if si := restore(ncfg, state, snap); !reflect.DeepEqual(si.Config.Subjects, ncfg.Subjects) {
+		t.Fatalf("Did not get expected stream info: %+v", si)
+	}
+}
+
 func TestJetStreamClusterAccountInfoAndLimits(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R5S", 5)
 	defer c.shutdown()
@@ -2612,6 +2899,78 @@ func TestJetStreamClusterExtendedAccountInfo(t *testing.T) {
 	}
 }
 
+func TestJetStreamClusterPeerRemovalAPI(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R5S", 5)
+	defer c.shutdown()
+
+	// Client based API
+	ml := c.leader()
+	nc, err := nats.Connect(ml.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+	if err != nil {
+		t.Fatalf("Failed to create system client: %v", err)
+	}
+	defer nc.Close()
+
+	// Expect error if unknown peer
+	req := &JSApiLeaderServerRemoveRequest{Server: "S-9"}
+	jsreq, err := json.Marshal(req)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	rmsg, err := nc.Request(JSApiRemoveServer, jsreq, time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	var resp JSApiLeaderServerRemoveResponse
+	if err := json.Unmarshal(rmsg.Data, &resp); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if resp.Error == nil {
+		t.Fatalf("Expected an error, got none")
+	}
+
+	sub, err := nc.SubscribeSync(JSAdvisoryServerRemoved)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	req = &JSApiLeaderServerRemoveRequest{Server: "S-2"}
+	jsreq, err = json.Marshal(req)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	rmsg, err = nc.Request(JSApiRemoveServer, jsreq, time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	resp.Error = nil
+	if err := json.Unmarshal(rmsg.Data, &resp); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if resp.Error != nil {
+		t.Fatalf("Unexpected error: %+v", resp.Error)
+	}
+	c.waitOnLeader()
+	ml = c.leader()
+
+	checkSubsPending(t, sub, 1)
+	madv, _ := sub.NextMsg(0)
+	var adv JSServerRemovedAdvisory
+	if err := json.Unmarshal(madv.Data, &adv); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if adv.Server != "S-2" {
+		t.Fatalf("Expected advisory about S-2 being removed, got %+v", adv)
+	}
+
+	for _, s := range ml.JetStreamClusterPeers() {
+		if s == "S-2" {
+			t.Fatalf("Still in the peer list")
+		}
+	}
+
+}
+
 func TestJetStreamClusterNoQuorumStepdown(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
@@ -2657,7 +3016,8 @@ func TestJetStreamClusterNoQuorumStepdown(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
-	sub, err := js.SubscribeSync("NO-Q")
+	// Make durable to have R match Stream.
+	sub, err := js.SubscribeSync("NO-Q", nats.Durable("rr"))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -4365,6 +4725,15 @@ func (c *cluster) waitOnAllCurrent() {
 	}
 }
 
+func (c *cluster) serverByName(sname string) *Server {
+	for _, s := range c.servers {
+		if s.Name() == sname {
+			return s
+		}
+	}
+	return nil
+}
+
 func (c *cluster) randomNonLeader() *Server {
 	// range should randomize.. but..
 	for _, s := range c.servers {
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index ddf410ed..7375183f 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -4061,7 +4061,7 @@ func TestJetStreamSnapshots(t *testing.T) {
 	}
 	// Try to restore from snapshot with current stream present, should error.
 	r := bytes.NewReader(snapshot)
-	if _, err := acc.RestoreStream(mname, r); err == nil {
+	if _, err := acc.RestoreStream(&info.cfg, r); err == nil {
 		t.Fatalf("Expected an error trying to restore existing stream")
 	} else if !strings.Contains(err.Error(), "name already in use") {
 		t.Fatalf("Incorrect error received: %v", err)
@@ -4071,13 +4071,7 @@ func TestJetStreamSnapshots(t *testing.T) {
 	mset.delete()
 	r.Reset(snapshot)
 
-	// Now send in wrong name
-	if _, err := acc.RestoreStream("foo", r); err == nil {
-		t.Fatalf("Expected an error trying to restore stream with wrong name")
-	}
-
-	r.Reset(snapshot)
-	mset, err = acc.RestoreStream(mname, r)
+	mset, err = acc.RestoreStream(&info.cfg, r)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -4114,7 +4108,7 @@ func TestJetStreamSnapshots(t *testing.T) {
 	}
 	acc = s2.GlobalAccount()
 	r.Reset(snapshot)
-	mset, err = acc.RestoreStream(mname, r)
+	mset, err = acc.RestoreStream(&info.cfg, r)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -10639,6 +10633,20 @@ func TestJetStreamSourceBasics(t *testing.T) {
 		return nil
 	})
 
+	// Test Source Updates
+	ncfg := &nats.StreamConfig{
+		Name: "MS",
+		Sources: []*nats.StreamSource{
+			// Keep foo, bar, remove baz, add dlc
+			&nats.StreamSource{Name: "foo"},
+			&nats.StreamSource{Name: "bar"},
+			&nats.StreamSource{Name: "dlc"},
+		},
+	}
+	if _, err := js.UpdateStream(ncfg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
 	// Test optional start times, filtered subjects etc.
 	if _, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Subjects: []string{"dlc", "rip"}}); err != nil {
 		t.Fatalf("Unexpected error: %v", err)