Repository: nats-io/nats-server
Pull request #1932 (2021-02-23 18:59:24)

Stream Mirrors, Sources and Updates Support
Also various fixes, stability improvements, etc.

/cc @nats-io/core



Patch:
diff --git a/.travis.yml b/.travis.yml
index 19a8c783..b9c1c3c9 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -3,8 +3,8 @@ dist: focal
 
 language: go
 go:
+- 1.16.x
 - 1.15.x
-- 1.14.x
 addons:
   apt:
     packages:
@@ -29,7 +29,7 @@ script:
 - set -e
 - go test -i ./...
 - go test -v -run=TestNoRace --failfast -p=1 ./...
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.14 ]]; then ./scripts/cov.sh TRAVIS; else go test -v -race -p=1 --failfast ./...; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.15 ]]; then ./scripts/cov.sh TRAVIS; else go test -v -race -p=1 --failfast ./...; fi
 - set +e
 
 deploy:
@@ -38,4 +38,4 @@ deploy:
   script: curl -sL http://git.io/goreleaser | bash
   on:
     tags: true
-    condition: $TRAVIS_GO_VERSION =~ 1.14
+    condition: $TRAVIS_GO_VERSION =~ 1.16
diff --git a/server/client.go b/server/client.go
index c1cc86e2..aab8306b 100644
--- a/server/client.go
+++ b/server/client.go
@@ -3476,7 +3476,12 @@ func (c *client) processInboundClientMsg(msg []byte) (bool, bool) {
 
 	// Now deal with gateways
 	if c.srv.gateway.enabled {
-		didDeliver = c.sendMsgToGateways(c.acc, msg, c.pa.subject, c.pa.reply, qnames) || didDeliver
+		reply := c.pa.reply
+		if len(c.pa.deliver) > 0 && c.kind == JETSTREAM && len(c.pa.reply) > 0 {
+			reply = append(reply, '@')
+			reply = append(reply, c.pa.deliver...)
+		}
+		didDeliver = c.sendMsgToGateways(c.acc, msg, c.pa.subject, reply, qnames) || didDeliver
 	}
 
 	// Check to see if we did not deliver to anyone and the client has a reply subject set
@@ -3585,6 +3590,21 @@ func removeHeaderIfPresent(hdr []byte, key string) []byte {
 	return append(hdr[:start], hdr[start+end+len(_CRLF_):]...)
 }
 
+// Generate a new header based on optional original header and key value.
+// More used in JetStream layers.
+func genHeader(hdr []byte, key, value string) []byte {
+	var bb *bytes.Buffer
+	if len(hdr) > LEN_CR_LF {
+		bb = bytes.NewBuffer(hdr[:len(hdr)-LEN_CR_LF])
+	} else {
+		bb = &bytes.Buffer{}
+		bb.WriteString(hdrLine)
+	}
+	http.Header{key: []string{value}}.Write(bb)
+	bb.WriteString(CR_LF)
+	return bb.Bytes()
+}
+
 // This will set a header for the message.
 // Lock does not need to be held but this should only be called
 // from the inbound go routine. We will update the pubArgs.
@@ -4742,7 +4762,7 @@ func (ci *ClientInfo) serviceAccount() string {
 
 // Grabs the information for this client.
 func (c *client) getClientInfo(detailed bool) *ClientInfo {
-	if c == nil || (c.kind != CLIENT && c.kind != LEAF) {
+	if c == nil || (c.kind != CLIENT && c.kind != LEAF && c.kind != JETSTREAM) {
 		return nil
 	}
 
diff --git a/server/const.go b/server/const.go
index 48eb7c96..6e437242 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.88"
+	VERSION = "2.2.0-beta.90"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index 8a967b40..b985aa18 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -2254,7 +2254,7 @@ func parseAckReplyNum(d string) (n int64) {
 const expectedNumReplyTokens = 9
 
 // Grab encoded information in the reply subject for a delivered message.
-func (o *consumer) replyInfo(subject string) (sseq, dseq, dc uint64, ts int64, pending uint64) {
+func replyInfo(subject string) (sseq, dseq, dc uint64, ts int64, pending uint64) {
 	tsa := [expectedNumReplyTokens]string{}
 	start, tokens := 0, tsa[:0]
 	for i := 0; i < len(subject); i++ {
diff --git a/server/events.go b/server/events.go
index c2549677..25c1e995 100644
--- a/server/events.go
+++ b/server/events.go
@@ -1557,11 +1557,10 @@ func (s *Server) sysUnsubscribe(sub *subscription) {
 		s.mu.Unlock()
 		return
 	}
-	acc := s.sys.account
 	c := s.sys.client
 	s.mu.Unlock()
 
-	c.unsubscribe(acc, sub, true, true)
+	c.processUnsub(sub.sid)
 }
 
 // This will generate the tracking subject for remote latency from the response subject.
diff --git a/server/gateway.go b/server/gateway.go
index aecdf2cd..cb2da1e6 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -401,14 +401,6 @@ func (g *srvGateway) updateRemotesTLSConfig(opts *Options) {
 	}
 }
 
-// Returns the Gateway's name of this server.
-func (g *srvGateway) getName() string {
-	g.RLock()
-	n := g.name
-	g.RUnlock()
-	return n
-}
-
 // Returns if this server rejects connections from gateways that are not
 // explicitly configured.
 func (g *srvGateway) rejectUnknown() bool {
@@ -843,6 +835,7 @@ func (s *Server) createGateway(cfg *gatewayCfg, url *url.URL, conn net.Conn) {
 }
 
 // Builds and sends the CONNECT protocol for a gateway.
+// Client lock held on entry.
 func (c *client) sendGatewayConnect(opts *Options) {
 	tlsRequired := c.gw.cfg.TLSConfig != nil
 	url := c.gw.connectURL
@@ -862,7 +855,7 @@ func (c *client) sendGatewayConnect(opts *Options) {
 		Pass:     pass,
 		TLS:      tlsRequired,
 		Name:     c.srv.info.ID,
-		Gateway:  c.srv.getGatewayName(),
+		Gateway:  c.srv.gateway.name,
 	}
 	b, err := json.Marshal(cinfo)
 	if err != nil {
@@ -1524,7 +1517,8 @@ func (s *Server) getGatewayURL() string {
 // Returns this server gateway name.
 // Same than calling s.gateway.getName()
 func (s *Server) getGatewayName() string {
-	return s.gateway.getName()
+	// This is immutable
+	return s.gateway.name
 }
 
 // All gateway connections (outbound and inbound) are put in the given map.
@@ -2340,6 +2334,11 @@ var subPool = &sync.Pool{
 // subject, etc..
 // <Invoked from any client connection's readLoop>
 func (c *client) sendMsgToGateways(acc *Account, msg, subject, reply []byte, qgroups [][]byte) bool {
+	// We had some times when we were sending across a GW with no subject, and the other side would break
+	// due to parser error. These need to be fixed upstream but also double check here.
+	if len(subject) == 0 {
+		return false
+	}
 	gwsa := [16]*client{}
 	gws := gwsa[:0]
 	// This is in fast path, so avoid calling functions when possible.
@@ -2952,7 +2951,7 @@ func (c *client) gatewaySwitchAccountToSendAllSubs(e *insie, accName string) {
 		// Use bare server info and simply set the
 		// gateway name and command
 		info := Info{
-			Gateway:           s.getGatewayName(),
+			Gateway:           s.gateway.name,
 			GatewayCmd:        cmd,
 			GatewayCmdPayload: []byte(accName),
 		}
diff --git a/server/jetstream.go b/server/jetstream.go
index 36c70889..f485d214 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -529,6 +529,7 @@ func (a *Account) EnableJetStream(limits *JetStreamAccountLimits) error {
 	a.mu.RLock()
 	s := a.srv
 	a.mu.RUnlock()
+
 	if s == nil {
 		return fmt.Errorf("jetstream account not registered")
 	}
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 280d765b..2f4922fb 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -549,7 +549,7 @@ var (
 	jsNotEnabledErr        = &ApiError{Code: 503, Description: "JetStream not enabled for account"}
 	jsBadRequestErr        = &ApiError{Code: 400, Description: "bad request"}
 	jsNotEmptyRequestErr   = &ApiError{Code: 400, Description: "expected an empty request payload"}
-	jsInvalidJSONErr       = &ApiError{Code: 400, Description: "invalid JSON request"}
+	jsInvalidJSONErr       = &ApiError{Code: 400, Description: "invalid JSON"}
 	jsInsufficientErr      = &ApiError{Code: 503, Description: "insufficient resources"}
 	jsNoConsumerErr        = &ApiError{Code: 404, Description: "consumer not found"}
 	jsStreamMismatchErr    = &ApiError{Code: 400, Description: "stream name in subject does not match request"}
@@ -633,13 +633,17 @@ func (s *Server) setJetStreamExportSubs() error {
 
 func (s *Server) sendAPIResponse(ci *ClientInfo, acc *Account, subject, reply, request, response string) {
 	acc.trackAPI()
-	s.sendInternalAccountMsg(nil, reply, response)
+	if reply != _EMPTY_ {
+		s.sendInternalAccountMsg(nil, reply, response)
+	}
 	s.sendJetStreamAPIAuditAdvisory(ci, acc, subject, request, response)
 }
 
 func (s *Server) sendAPIErrResponse(ci *ClientInfo, acc *Account, subject, reply, request, response string) {
 	acc.trackAPIErr()
-	s.sendInternalAccountMsg(nil, reply, response)
+	if reply != _EMPTY_ {
+		s.sendInternalAccountMsg(nil, reply, response)
+	}
 	s.sendJetStreamAPIAuditAdvisory(ci, acc, subject, request, response)
 }
 
@@ -1020,6 +1024,55 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, subject, re
 		return
 	}
 
+	// Do some pre-checking for mirror config to avoid cycles in clustered mode.
+	if cfg.Mirror != nil {
+		if len(cfg.Subjects) > 0 {
+			resp.Error = &ApiError{Code: 400, Description: "stream mirrors can not also contain subjects"}
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		if len(cfg.Sources) > 0 {
+			resp.Error = &ApiError{Code: 400, Description: "stream mirrors can not also contain other sources"}
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		if cfg.Mirror.FilterSubject != _EMPTY_ {
+			resp.Error = &ApiError{Code: 400, Description: "stream mirrors can not contain filtered subjects"}
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		if cfg.Mirror.OptStartSeq > 0 && cfg.Mirror.OptStartTime != nil {
+			resp.Error = &ApiError{Code: 400, Description: "stream mirrors can not have both start seq and start time configured"}
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+
+		// Now make sure the other stream exists.
+		var exists bool
+		var maxMsgSize int32
+
+		if s.JetStreamIsClustered() {
+			js, _ := s.getJetStreamCluster()
+			if sa := js.streamAssignment(acc.Name, cfg.Mirror.Name); sa != nil {
+				maxMsgSize = sa.Config.MaxMsgSize
+				exists = true
+			}
+		} else if mset, err := acc.lookupStream(cfg.Mirror.Name); err == nil {
+			maxMsgSize = mset.cfg.MaxMsgSize
+			exists = true
+		}
+		if !exists {
+			resp.Error = &ApiError{Code: 400, Description: "stream mirror source must exist"}
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+		if cfg.MaxMsgSize > 0 && maxMsgSize > 0 && cfg.MaxMsgSize < maxMsgSize {
+			resp.Error = &ApiError{Code: 400, Description: "stream mirror must have max message size >= source"}
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+	}
+
 	if s.JetStreamIsClustered() {
 		s.jsClusteredStreamRequest(ci, acc, subject, reply, rmsg, &cfg)
 		return
@@ -1083,6 +1136,12 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, subject, re
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
+
+	if s.JetStreamIsClustered() {
+		s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg)
+		return
+	}
+
 	mset, err := acc.lookupStream(streamName)
 	if err != nil {
 		resp.Error = jsNotFoundError(err)
@@ -1390,6 +1449,11 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, repl
 	js, _ := s.getJetStreamCluster()
 
 	resp.StreamInfo = &StreamInfo{Created: mset.createdTime(), State: mset.state(), Config: config, Cluster: js.clusterInfo(mset.raftGroup())}
+	if mset.isMirror() {
+		resp.StreamInfo.Mirror = mset.mirrorInfo()
+	} else if mset.hasSources() {
+		resp.StreamInfo.Sources = mset.sourcesInfo()
+	}
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
 }
 
@@ -1725,9 +1789,10 @@ func (s *Server) jsLeaderStepDownRequest(sub *subscription, c *client, subject,
 		}
 		cn := req.Placement.Cluster
 		var peers []string
+		ourID := cc.meta.ID()
 		for _, p := range cc.meta.Peers() {
 			if si, ok := s.nodeToInfo.Load(p.ID); ok && si != nil {
-				if ni := si.(*nodeInfo); ni.offline || ni.cluster != cn {
+				if ni := si.(*nodeInfo); ni.offline || ni.cluster != cn || p.ID == ourID {
 					continue
 				}
 				peers = append(peers, p.ID)
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 1680d9cb..7975faab 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"math/rand"
 	"path"
+	"reflect"
 	"sort"
 	"strings"
 	"sync/atomic"
@@ -74,6 +75,8 @@ const (
 	assignCompressedConsumerOp
 	// Filtered Consumer skip.
 	updateSkipOp
+	// Update Stream
+	updateStreamOp
 )
 
 // raftGroups are controlled by the metagroup controller.
@@ -116,6 +119,7 @@ type consumerAssignment struct {
 	State   *ConsumerState  `json:"state,omitempty"`
 	// Internal
 	responded bool
+	deleted   bool
 	err       error
 }
 
@@ -490,6 +494,7 @@ func (js *jetStream) setupMetaGroup() error {
 		s.Warnf("Could not start metadata controller: %v", err)
 		return err
 	}
+	n.Campaign()
 
 	c := s.createInternalJetStreamClient()
 	sacc := s.SystemAccount()
@@ -734,13 +739,13 @@ func (js *jetStream) monitorCluster() {
 				continue
 			}
 			// FIXME(dlc) - Deal with errors.
-			if hadSnapshot, didRemoval, err := js.applyMetaEntries(ce.Entries, isRecovering); err == nil {
+			if _, didRemoval, err := js.applyMetaEntries(ce.Entries, isRecovering); err == nil {
 				n.Applied(ce.Index)
-				if (hadSnapshot && !isRecovering) || didRemoval {
+				if didRemoval {
 					// Since we received one make sure we have our own since we do not store
 					// our meta state outside of raft.
 					doSnapshot()
-				} else if _, b := n.Size(); lastSnap != nil && b > uint64(len(lastSnap)*4) {
+				} else if _, b := n.Size(); b > uint64(len(lastSnap)*4) {
 					doSnapshot()
 				}
 			}
@@ -1030,6 +1035,16 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool) (bool
 				}
 				js.processConsumerRemoval(ca)
 				didRemove = true
+			case updateStreamOp:
+				sa, err := decodeStreamAssignment(buf[1:])
+				if err != nil {
+					js.srv.Errorf("JetStream cluster failed to decode stream assignment: %q", buf[1:])
+					return didSnap, didRemove, err
+				}
+				if isRecovering {
+					js.setStreamAssignmentResponded(sa)
+				}
+				js.processUpdateStreamAssignment(sa)
 			default:
 				panic("JetStream Cluster Unknown meta entry op type")
 			}
@@ -1215,7 +1230,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 				n.Applied(ce.Index)
 				ne := ce.Index - lastApplied
 				// If over our compact min and we have at least min entries to compact, go ahead and snapshot/compact.
-				if _, b := n.Size(); lastSnap == nil || (b > compactSizeMin && ne >= compactNumMin) {
+				if _, b := n.Size(); b > compactSizeMin && ne >= compactNumMin {
 					doSnapshot()
 				}
 			} else {
@@ -1230,9 +1245,6 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 					js.setStreamAssignmentResponded(sa)
 				}
 				js.processStreamLeaderChange(mset, isLeader)
-				if isLeader {
-					lastSnap = nil
-				}
 			}
 		case <-t.C:
 			if isLeader {
@@ -1369,14 +1381,14 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 					continue
 				}
 
+				s := js.srv
 				if err := mset.processJetStreamMsg(subject, reply, hdr, msg, lseq, ts); err != nil {
 					if err != errLastSeqMismatch || !isRecovering {
-						js.srv.Debugf("Got error processing JetStream msg: %v", err)
+						s.Debugf("Got error processing JetStream msg: %v", err)
 					}
 					if strings.Contains(err.Error(), "no space left") {
-						s := js.srv
 						s.Errorf("JetStream out of space, will be DISABLED")
-						js.srv.DisableJetStream()
+						s.DisableJetStream()
 						return err
 					}
 				}
@@ -1394,7 +1406,8 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 					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.serviceAccount(), err)
+					s.Debugf("JetStream cluster failed to delete msg %d from stream %q for account %q: %v",
+						md.Seq, md.Stream, md.Client.serviceAccount(), err)
 				}
 
 				js.mu.RLock()
@@ -1660,7 +1673,7 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) {
 	ourID := cc.meta.ID()
 
 	var isMember bool
-	if sa.Group != nil && cc.meta != nil {
+	if sa.Group != nil {
 		isMember = sa.Group.isMember(ourID)
 	}
 
@@ -1694,6 +1707,135 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) {
 	}
 }
 
+// processUpdateStreamAssignment is called when followers have replicated an updated assignment.
+func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
+	js.mu.RLock()
+	s, cc := js.srv, js.cluster
+	js.mu.RUnlock()
+	if s == nil || cc == nil {
+		// TODO(dlc) - debug at least
+		return
+	}
+
+	acc, err := s.LookupAccount(sa.Client.serviceAccount())
+	if err != nil {
+		// TODO(dlc) - log error
+		return
+	}
+	stream := sa.Config.Name
+
+	js.mu.Lock()
+	if cc.meta == nil {
+		js.mu.Unlock()
+		return
+	}
+	ourID := cc.meta.ID()
+
+	var isMember bool
+	if sa.Group != nil {
+		isMember = sa.Group.isMember(ourID)
+	}
+
+	accStreams := cc.streams[acc.Name]
+	if accStreams == nil {
+		js.mu.Unlock()
+		return
+	}
+	osa := accStreams[stream]
+	if osa == nil {
+		js.mu.Unlock()
+		return
+	}
+
+	// Copy over private existing state from former SA.
+	sa.Group.node = osa.Group.node
+	sa.consumers = osa.consumers
+	sa.err = osa.err
+
+	// Update our state.
+	accStreams[stream] = sa
+	cc.streams[acc.Name] = accStreams
+
+	js.mu.Unlock()
+
+	// Check if this is for us..
+	if isMember {
+		js.processClusterUpdateStream(acc, sa)
+	} else if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
+		// We have one here even though we are not a member. This can happen on re-assignment.
+		s.Debugf("JetStream removing stream '%s > %s' from this server, re-assigned", sa.Client.serviceAccount(), sa.Config.Name)
+		if node := mset.raftNode(); node != nil {
+			node.ProposeRemovePeer(ourID)
+		}
+		mset.stop(true, false)
+	}
+}
+
+// processClusterUpdateStream is called when we have a stream assignment that
+// has been updated for an existing assignment.
+func (js *jetStream) processClusterUpdateStream(acc *Account, sa *streamAssignment) {
+	if sa == nil {
+		return
+	}
+
+	js.mu.RLock()
+	s, rg := js.srv, sa.Group
+	client, subject, reply := sa.Client, sa.Subject, sa.Reply
+	alreadyRunning := rg.node != nil
+	hasResponded := sa.responded
+	sa.responded = true
+	js.mu.RUnlock()
+
+	mset, err := acc.lookupStream(sa.Config.Name)
+	if err == nil || mset != nil {
+		if rg.node != nil && !alreadyRunning {
+			s.startGoRoutine(func() { js.monitorStream(mset, sa) })
+		}
+		mset.setStreamAssignment(sa)
+		if err = mset.update(sa.Config); err != nil {
+			s.Warnf("JetStream cluster error updating stream %q for account %q: %v", sa.Config.Name, acc.Name, err)
+		}
+	}
+
+	if err != nil {
+		js.mu.Lock()
+		sa.err = err
+		sa.responded = true
+		if rg.node != nil {
+			rg.node.Delete()
+		}
+		result := &streamAssignmentResult{
+			Account:  sa.Client.serviceAccount(),
+			Stream:   sa.Config.Name,
+			Response: &JSApiStreamCreateResponse{ApiResponse: ApiResponse{Type: JSApiStreamCreateResponseType}},
+		}
+		result.Response.Error = jsError(err)
+		js.mu.Unlock()
+
+		// Send response to the metadata leader. They will forward to the user as needed.
+		s.sendInternalMsgLocked(streamAssignmentSubj, _EMPTY_, nil, result)
+		return
+	}
+
+	mset.mu.RLock()
+	isLeader := mset.isLeader()
+	mset.mu.RUnlock()
+
+	if !isLeader || hasResponded {
+		return
+	}
+
+	// Send our response.
+	var resp = JSApiStreamUpdateResponse{ApiResponse: ApiResponse{Type: JSApiStreamUpdateResponseType}}
+	if err != nil {
+		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())}
+		s.sendAPIResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
+	}
+}
+
 // processClusterCreateStream is called when we have a stream assignment that
 // has been committed and this server is a member of the peer group.
 func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignment) {
@@ -1759,8 +1901,7 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 		js.mu.Unlock()
 
 		// Send response to the metadata leader. They will forward to the user as needed.
-		b, _ := json.Marshal(result) // Avoids auto-processing and doing fancy json with newlines.
-		s.sendInternalMsgLocked(streamAssignmentSubj, _EMPTY_, nil, b)
+		s.sendInternalMsgLocked(streamAssignmentSubj, _EMPTY_, nil, result)
 		return
 	}
 
@@ -1897,6 +2038,7 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 	}
 	js.mu.RLock()
 	s := js.srv
+	hadLeader := sa.Group.node == nil || sa.Group.node.GroupLeader() != noLeader
 	js.mu.RUnlock()
 
 	acc, err := s.LookupAccount(sa.Client.serviceAccount())
@@ -1919,7 +2061,7 @@ func (js *jetStream) processClusterDeleteStream(sa *streamAssignment, isMember,
 		sa.Group.node.Delete()
 	}
 
-	if !isMember || !wasLeader && sa.Group.node != nil && sa.Group.node.GroupLeader() != noLeader {
+	if !isMember || !wasLeader && hadLeader {
 		return
 	}
 
@@ -2307,7 +2449,7 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 				n.Applied(ce.Index)
 				ne := ce.Index - lastApplied
 				// If over our compact min and we have at least min entries to compact, go ahead and snapshot/compact.
-				if _, b := n.Size(); lastSnap == nil || (b > compactSizeMin && ne > compactNumMin) {
+				if _, b := n.Size(); b > compactSizeMin && ne > compactNumMin {
 					doSnapshot()
 				}
 			} else {
@@ -2318,9 +2460,6 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 				js.setConsumerAssignmentResponded(ca)
 			}
 			js.processConsumerLeaderChange(o, isLeader)
-			if isLeader {
-				lastSnap = nil
-			}
 		case <-t.C:
 			if isLeader {
 				doSnapshot()
@@ -2806,7 +2945,7 @@ func (cc *jetStreamCluster) createGroupForStream(ci *ClientInfo, cfg *StreamConf
 	return &raftGroup{Name: groupNameForStream(peers, cfg.Storage), Storage: cfg.Storage, Peers: peers}
 }
 
-func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, cfg *StreamConfig) {
+func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, config *StreamConfig) {
 	js, cc := s.getJetStreamCluster()
 	if js == nil || cc == nil {
 		return
@@ -2814,10 +2953,6 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 
 	var resp = JSApiStreamCreateResponse{ApiResponse: ApiResponse{Type: JSApiStreamCreateResponseType}}
 
-	js.mu.RLock()
-	numStreams := len(cc.streams[acc.Name])
-	js.mu.RUnlock()
-
 	// Grab our jetstream account info.
 	acc.mu.RLock()
 	jsa := acc.js
@@ -2829,7 +2964,19 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 		return
 	}
 
-	// Check for stream limits here before proposing.
+	ccfg, err := checkStreamCfg(config)
+	if err != nil {
+		resp.Error = jsError(err)
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+		return
+	}
+	cfg := &ccfg
+
+	js.mu.RLock()
+	numStreams := len(cc.streams[acc.Name])
+	js.mu.RUnlock()
+
+	// Check for stream limits here before proposing. These need to be tracked from meta layer, not jsa.
 	jsa.mu.RLock()
 	exceeded := jsa.limits.MaxStreams > 0 && numStreams >= jsa.limits.MaxStreams
 	jsa.mu.RUnlock()
@@ -2840,6 +2987,13 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 		return
 	}
 
+	// Check for stream limits here before proposing.
+	if err := jsa.checkLimits(cfg); err != nil {
+		resp.Error = jsError(err)
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+		return
+	}
+
 	// Now process the request and proposal.
 	js.mu.Lock()
 	defer js.mu.Unlock()
@@ -2864,6 +3018,43 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 	cc.meta.Propose(encodeAddStreamAssignment(sa))
 }
 
+func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, cfg *StreamConfig) {
+	js, cc := s.getJetStreamCluster()
+	if js == nil || cc == nil {
+		return
+	}
+
+	// Now process the request and proposal.
+	js.mu.Lock()
+	defer js.mu.Unlock()
+
+	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))
+		return
+	}
+	var newCfg *StreamConfig
+	if jsa := js.accounts[acc]; jsa != nil {
+		if ncfg, err := jsa.configUpdateCheck(osa.Config, cfg); err != nil {
+			resp.Error = jsError(err)
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+			return
+		} else {
+			newCfg = ncfg
+		}
+	} else {
+		resp.Error = jsNotEnabledErr
+		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))
+}
+
 func (s *Server) jsClusteredStreamDeleteRequest(ci *ClientInfo, acc *Account, stream, subject, reply string, rmsg []byte) {
 	js, cc := s.getJetStreamCluster()
 	if js == nil || cc == nil {
@@ -3216,7 +3407,7 @@ func (s *Server) jsClusteredConsumerDeleteRequest(ci *ClientInfo, acc *Account,
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
 		return
 	}
-
+	oca.deleted = true
 	ca := &consumerAssignment{Group: oca.Group, Stream: stream, Name: consumer, Config: oca.Config, Subject: subject, Reply: reply, Client: ci}
 	cc.meta.Propose(encodeDeleteConsumerAssignment(ca))
 }
@@ -3279,6 +3470,13 @@ func encodeAddStreamAssignment(sa *streamAssignment) []byte {
 	return bb.Bytes()
 }
 
+func encodeUpdateStreamAssignment(sa *streamAssignment) []byte {
+	var bb bytes.Buffer
+	bb.WriteByte(byte(updateStreamOp))
+	json.NewEncoder(&bb).Encode(sa)
+	return bb.Bytes()
+}
+
 func encodeDeleteStreamAssignment(sa *streamAssignment) []byte {
 	var bb bytes.Buffer
 	bb.WriteByte(byte(removeStreamOp))
@@ -3343,10 +3541,13 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 		}
 	} else {
 		oname = cfg.Durable
-		if sa.consumers[oname] != nil {
-			resp.Error = jsError(ErrJetStreamConsumerAlreadyUsed)
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
-			return
+		if ca := sa.consumers[oname]; ca != nil && !ca.deleted {
+			// This can be ok if delivery subject update.
+			if !reflect.DeepEqual(cfg, ca.Config) && !configsEqualSansDelivery(*cfg, *ca.Config) {
+				resp.Error = jsError(ErrJetStreamConsumerAlreadyUsed)
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+				return
+			}
 		}
 	}
 
diff --git a/server/raft.go b/server/raft.go
index 0ec51220..839f4fd3 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -191,11 +191,11 @@ type lps struct {
 }
 
 const (
-	minElectionTimeout = 500 * time.Millisecond
+	minElectionTimeout = 1500 * time.Millisecond
 	maxElectionTimeout = 3 * minElectionTimeout
 	minCampaignTimeout = 50 * time.Millisecond
 	maxCampaignTimeout = 4 * minCampaignTimeout
-	hbInterval         = 200 * time.Millisecond
+	hbInterval         = 250 * time.Millisecond
 	lostQuorumInterval = hbInterval * 3
 )
 
@@ -1069,7 +1069,18 @@ func (n *raft) shutdown(shouldDelete bool) {
 		return
 	}
 	close(n.quit)
-	n.c.closeConnection(InternalClient)
+	if c := n.c; c != nil {
+		var subs []*subscription
+		c.mu.Lock()
+		for _, sub := range c.subs {
+			subs = append(subs, sub)
+		}
+		c.mu.Unlock()
+		for _, sub := range subs {
+			n.unsubscribe(sub)
+		}
+		c.closeConnection(InternalClient)
+	}
 	n.state = Closed
 	s, g, wal := n.s, n.group, n.wal
 
@@ -1120,6 +1131,13 @@ func (n *raft) subscribe(subject string, cb msgHandler) (*subscription, error) {
 	return n.s.systemSubscribe(subject, _EMPTY_, false, n.c, cb)
 }
 
+// Lock should be held.
+func (n *raft) unsubscribe(sub *subscription) {
+	if sub != nil {
+		n.c.processUnsub(sub.sid)
+	}
+}
+
 func (n *raft) createInternalSubs() error {
 	n.vsubj, n.vreply = fmt.Sprintf(raftVoteSubj, n.group), n.newInbox()
 	n.asubj, n.areply = fmt.Sprintf(raftAppendSubj, n.group), n.newInbox()
@@ -1431,28 +1449,30 @@ func (n *raft) handleForwardedProposal(sub *subscription, c *client, _, reply st
 
 func (n *raft) runAsLeader() {
 	n.RLock()
+	if n.state == Closed {
+		n.RUnlock()
+		return
+	}
 	psubj, rpsubj := n.psubj, n.rpsubj
 	n.RUnlock()
 
 	// For forwarded proposals, both normal and remove peer proposals.
 	fsub, err := n.subscribe(psubj, n.handleForwardedProposal)
 	if err != nil {
-		panic(fmt.Sprintf("Error subscribing to forwarded proposals: %v", err))
+		n.debug("Error subscribing to forwarded proposals: %v", err)
+		return
 	}
 	rpsub, err := n.subscribe(rpsubj, n.handleForwardedRemovePeerProposal)
 	if err != nil {
-		panic(fmt.Sprintf("Error subscribing to forwarded proposals: %v", err))
+		n.debug("Error subscribing to forwarded proposals: %v", err)
+		return
 	}
 
 	// Cleanup our subscription when we leave.
 	defer func() {
 		n.Lock()
-		if fsub != nil {
-			n.s.sysUnsubscribe(fsub)
-		}
-		if rpsub != nil {
-			n.s.sysUnsubscribe(rpsub)
-		}
+		n.unsubscribe(fsub)
+		n.unsubscribe(rpsub)
 		n.Unlock()
 	}()
 
@@ -1946,9 +1966,8 @@ func (n *raft) handleAppendEntry(sub *subscription, c *client, subject, reply st
 // Lock should be held.
 func (n *raft) cancelCatchup() {
 	n.debug("Canceling catchup subscription since we are now up to date")
-
 	if n.catchup != nil && n.catchup.sub != nil {
-		n.s.sysUnsubscribe(n.catchup.sub)
+		n.unsubscribe(n.catchup.sub)
 	}
 	n.catchup = nil
 }
@@ -1972,7 +1991,7 @@ func (n *raft) catchupStalled() bool {
 func (n *raft) createCatchup(ae *appendEntry) string {
 	// Cleanup any old ones.
 	if n.catchup != nil && n.catchup.sub != nil {
-		n.s.sysUnsubscribe(n.catchup.sub)
+		n.unsubscribe(n.catchup.sub)
 	}
 	// Snapshot term and index.
 	n.catchup = &catchupState{
diff --git a/server/stream.go b/server/stream.go
index 19783e16..e498e044 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -27,6 +27,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -52,6 +53,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"`
 }
 
 const JSApiPubAckResponseType = "io.nats.jetstream.api.v1.pub_ack_response"
@@ -72,10 +75,12 @@ type PubAck struct {
 
 // StreamInfo shows config and current state for this stream.
 type StreamInfo struct {
-	Config  StreamConfig `json:"config"`
-	Created time.Time    `json:"created"`
-	State   StreamState  `json:"state"`
-	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"`
 }
 
 // ClusterInfo shows information about the underlying set of servers
@@ -96,11 +101,28 @@ type PeerInfo struct {
 	Lag     uint64        `json:"lag,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"`
+	Error  *ApiError     `json:"error,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"`
+}
+
 // Stream is a jetstream stream of messages. When we receive a message internally destined
-// for a Stream we will direct link from the client to this Stream structure.
+// for a Stream we will direct link from the client to this structure.
 type stream struct {
 	mu        sync.RWMutex
 	jsa       *jsAccount
+	acc       *Account
 	srv       *Server
 	client    *client
 	sysc      *client
@@ -114,6 +136,7 @@ type stream struct {
 	numFilter int
 	cfg       StreamConfig
 	created   time.Time
+	stype     StorageType
 	ddmap     map[string]*ddentry
 	ddarr     []*ddentry
 	ddindex   int
@@ -121,6 +144,13 @@ type stream struct {
 	qch       chan struct{}
 	active    bool
 
+	// Mirror
+	mirror *sourceInfo
+
+	// Sources
+	sources map[string]*sourceInfo
+
+	// TODO(dlc) - Hide everything below behind two pointers.
 	// Clustered mode.
 	sa      *streamAssignment
 	node    RaftNode
@@ -132,12 +162,23 @@ type stream struct {
 	lqsent  time.Time
 }
 
+type sourceInfo struct {
+	name string
+	sub  *subscription
+	sseq uint64
+	dseq uint64
+	lag  uint64
+	err  *ApiError
+	last time.Time
+}
+
 // Headers for published messages.
 const (
 	JSMsgId             = "Nats-Msg-Id"
 	JSExpectedStream    = "Nats-Expected-Stream"
 	JSExpectedLastSeq   = "Nats-Expected-Last-Sequence"
 	JSExpectedLastMsgId = "Nats-Expected-Last-Msg-Id"
+	JSStreamSource      = "Nats-Stream-Source"
 )
 
 // Dedupe entry
@@ -209,6 +250,30 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		}
 	}
 
+	// Check for mirror designation.
+	if cfg.Mirror != nil {
+		// Can't have subjects.
+		if len(cfg.Subjects) > 0 {
+			jsa.mu.Unlock()
+			return nil, fmt.Errorf("stream mirrors can not also contain subjects")
+		}
+		if len(cfg.Sources) > 0 {
+			jsa.mu.Unlock()
+			return nil, fmt.Errorf("stream mirrors can not also contain other sources")
+		}
+		if cfg.Mirror.FilterSubject != _EMPTY_ {
+			jsa.mu.Unlock()
+			return nil, fmt.Errorf("stream mirrors can not contain filtered subjects")
+		}
+		if cfg.Mirror.OptStartSeq > 0 && cfg.Mirror.OptStartTime != nil {
+			jsa.mu.Unlock()
+			return nil, fmt.Errorf("stream mirrors can not have both start seq and start time configured")
+		}
+	} else if len(cfg.Subjects) == 0 && len(cfg.Sources) == 0 {
+		jsa.mu.Unlock()
+		return nil, fmt.Errorf("stream needs at least one configured subject or mirror")
+	}
+
 	// Check for overlapping subjects. These are not allowed for now.
 	if jsa.subjectsOverlap(cfg.Subjects) {
 		jsa.mu.Unlock()
@@ -219,7 +284,17 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 	c := s.createInternalJetStreamClient()
 	ic := s.createInternalJetStreamClient()
 
-	mset := &stream{jsa: jsa, cfg: cfg, srv: s, client: c, sysc: ic, consumers: make(map[string]*consumer), qch: make(chan struct{})}
+	mset := &stream{
+		acc:       a,
+		jsa:       jsa,
+		cfg:       cfg,
+		srv:       s,
+		client:    c,
+		sysc:      ic,
+		stype:     cfg.Storage,
+		consumers: make(map[string]*consumer),
+		qch:       make(chan struct{}),
+	}
 
 	jsa.streams[cfg.Name] = mset
 	storeDir := path.Join(jsa.storeDir, streamsDir, cfg.Name)
@@ -641,7 +716,9 @@ func checkStreamCfg(config *StreamConfig) (StreamConfig, error) {
 	}
 
 	if len(cfg.Subjects) == 0 {
-		cfg.Subjects = append(cfg.Subjects, cfg.Name)
+		if cfg.Mirror == nil && len(cfg.Sources) == 0 {
+			cfg.Subjects = append(cfg.Subjects, cfg.Name)
+		}
 	} else {
 		// We can allow overlaps, but don't allow direct duplicates.
 		dset := make(map[string]struct{}, len(cfg.Subjects))
@@ -677,101 +754,99 @@ func (mset *stream) fileStoreConfig() (FileStoreConfig, error) {
 	return fs.fileStoreConfig(), nil
 }
 
-// Update will allow certain configuration properties of an existing stream to be updated.
-func (mset *stream) update(config *StreamConfig) error {
-	cfg, err := checkStreamCfg(config)
+func (jsa *jsAccount) configUpdateCheck(old, new *StreamConfig) (*StreamConfig, error) {
+	cfg, err := checkStreamCfg(new)
 	if err != nil {
-		return err
+		return nil, err
 	}
-	o_cfg := mset.config()
 
 	// Name must match.
-	if cfg.Name != o_cfg.Name {
-		return fmt.Errorf("stream configuration name must match original")
+	if cfg.Name != old.Name {
+		return nil, fmt.Errorf("stream configuration name must match original")
 	}
 	// Can't change MaxConsumers for now.
-	if cfg.MaxConsumers != o_cfg.MaxConsumers {
-		return fmt.Errorf("stream configuration update can not change MaxConsumers")
+	if cfg.MaxConsumers != old.MaxConsumers {
+		return nil, fmt.Errorf("stream configuration update can not change MaxConsumers")
 	}
 	// Can't change storage types.
-	if cfg.Storage != o_cfg.Storage {
-		return fmt.Errorf("stream configuration update can not change storage type")
+	if cfg.Storage != old.Storage {
+		return nil, fmt.Errorf("stream configuration update can not change storage type")
 	}
 	// Can't change retention.
-	if cfg.Retention != o_cfg.Retention {
-		return fmt.Errorf("stream configuration update can not change retention policy")
+	if cfg.Retention != old.Retention {
+		return nil, fmt.Errorf("stream configuration update can not change retention policy")
 	}
 	// Can not have a template owner for now.
-	if o_cfg.Template != "" {
-		return fmt.Errorf("stream configuration update not allowed on template owned stream")
+	if old.Template != _EMPTY_ {
+		return nil, fmt.Errorf("stream configuration update not allowed on template owned stream")
 	}
-	if cfg.Template != "" {
-		return fmt.Errorf("stream configuration update can not be owned by a template")
+	if cfg.Template != _EMPTY_ {
+		return nil, fmt.Errorf("stream configuration update can not be owned by a template")
 	}
 
 	// Check limits.
-	mset.mu.Lock()
-	jsa := mset.jsa
-	mset.mu.Unlock()
-
-	jsa.mu.Lock()
-	if cfg.MaxConsumers > 0 && cfg.MaxConsumers > jsa.limits.MaxConsumers {
-		jsa.mu.Unlock()
-		return fmt.Errorf("stream configuration maximum consumers exceeds account limit")
-	}
-	if cfg.MaxBytes > 0 && cfg.MaxBytes > o_cfg.MaxBytes {
-		if err := jsa.checkBytesLimits(cfg.MaxBytes*int64(cfg.Replicas), cfg.Storage); err != nil {
-			jsa.mu.Unlock()
-			return err
-		}
+	if err := jsa.checkLimits(&cfg); err != nil {
+		return nil, err
 	}
-	jsa.mu.Unlock()
+	return &cfg, nil
+}
 
-	// Now check for subject interest differences.
-	current := make(map[string]struct{}, len(o_cfg.Subjects))
-	for _, s := range o_cfg.Subjects {
-		current[s] = struct{}{}
+// Update will allow certain configuration properties of an existing stream to be updated.
+func (mset *stream) update(config *StreamConfig) error {
+	ocfg := mset.config()
+	cfg, err := mset.jsa.configUpdateCheck(&ocfg, config)
+	if err != nil {
+		return err
 	}
-	// Update config with new values. The store update will enforce any stricter limits.
-	mset.mu.Lock()
 
-	// Now walk new subjects. All of these need to be added, but we will check
-	// the originals first, since if it is in there we can skip, already added.
-	for _, s := range cfg.Subjects {
-		if _, ok := current[s]; !ok {
-			if _, err := mset.subscribeInternal(s, mset.processInboundJetStreamMsg); err != nil {
+	mset.mu.Lock()
+	if mset.isLeader() {
+		// Now check for subject interest differences.
+		current := make(map[string]struct{}, len(ocfg.Subjects))
+		for _, s := range ocfg.Subjects {
+			current[s] = struct{}{}
+		}
+		// Update config with new values. The store update will enforce any stricter limits.
+
+		// Now walk new subjects. All of these need to be added, but we will check
+		// the originals first, since if it is in there we can skip, already added.
+		for _, s := range cfg.Subjects {
+			if _, ok := current[s]; !ok {
+				if _, err := mset.subscribeInternal(s, mset.processInboundJetStreamMsg); err != nil {
+					mset.mu.Unlock()
+					return err
+				}
+			}
+			delete(current, s)
+		}
+		// What is left in current needs to be deleted.
+		for s := range current {
+			if err := mset.unsubscribeInternal(s); err != nil {
 				mset.mu.Unlock()
 				return err
 			}
 		}
-		delete(current, s)
-	}
-	// What is left in current needs to be deleted.
-	for s := range current {
-		if err := mset.unsubscribeInternal(s); err != nil {
-			mset.mu.Unlock()
-			return err
+
+		// Check for the Duplicates
+		if cfg.Duplicates != ocfg.Duplicates && mset.ddtmr != nil {
+			// Let it fire right away, it will adjust properly on purge.
+			mset.ddtmr.Reset(time.Microsecond)
 		}
 	}
 
-	// Check for the Duplicates
-	if cfg.Duplicates != o_cfg.Duplicates && mset.ddtmr != nil {
-		// Let it fire right away, it will adjust properly on purge.
-		mset.ddtmr.Reset(time.Microsecond)
-	}
 	// Now update config and store's version of our config.
-	mset.cfg = cfg
+	mset.cfg = *cfg
 
 	var suppress bool
 	if mset.isClustered() && mset.sa != nil {
 		suppress = mset.sa.responded
 	}
-	if !suppress {
+	if mset.isLeader() && !suppress {
 		mset.sendUpdateAdvisoryLocked()
 	}
 	mset.mu.Unlock()
 
-	mset.store.UpdateConfig(&cfg)
+	mset.store.UpdateConfig(cfg)
 
 	return nil
 }
@@ -831,6 +906,478 @@ func (mset *stream) eraseMsg(seq uint64) (bool, error) {
 	return mset.store.EraseMsg(seq)
 }
 
+// Are we a mirror?
+func (mset *stream) isMirror() bool {
+	mset.mu.RLock()
+	defer mset.mu.RUnlock()
+	return mset.cfg.Mirror != nil
+}
+
+func (mset *stream) hasSources() bool {
+	mset.mu.RLock()
+	defer mset.mu.RUnlock()
+	return len(mset.sources) > 0
+}
+
+func (mset *stream) sourcesInfo() (sis []*StreamSourceInfo) {
+	mset.mu.RLock()
+	defer mset.mu.RUnlock()
+	for _, si := range mset.sources {
+		sis = append(sis, mset.sourceInfo(si))
+	}
+	return sis
+}
+
+// Lock should be held
+func (mset *stream) sourceInfo(si *sourceInfo) *StreamSourceInfo {
+	if si == nil {
+		return nil
+	}
+	return &StreamSourceInfo{Name: si.name, Lag: si.lag, Active: time.Since(si.last), Error: si.err}
+}
+
+// Return our source info for our mirror.
+func (mset *stream) mirrorInfo() *StreamSourceInfo {
+	mset.mu.RLock()
+	defer mset.mu.RUnlock()
+	return mset.sourceInfo(mset.mirror)
+}
+
+// processClusteredMirrorMsg will propose the inbound mirrored message to the underlying raft group.
+func (mset *stream) processClusteredMirrorMsg(subject string, hdr, msg []byte, seq uint64, ts int64) error {
+	mset.mu.Lock()
+	defer mset.mu.Unlock()
+	// Do proposal.
+	return mset.node.Propose(encodeStreamMsg(subject, _EMPTY_, hdr, msg, seq, ts))
+}
+
+// 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)
+
+	mset.mu.Lock()
+
+	// Ignore anything not current.
+	if mset.mirror == nil || sub != mset.mirror.sub {
+		mset.mu.Unlock()
+		return
+	}
+
+	// Mirror info tracking.
+	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)
+	} else {
+		err = mset.processJetStreamMsg(subject, reply, hdr, msg, sseq-1, ts)
+	}
+	if err != nil {
+		if err == errLastSeqMismatch {
+			// We may have missed messages, restart.
+			mset.resetMirrorConsumer()
+		} else {
+			s.Debugf("Got error processing JetStream mirror msg: %v", err)
+		}
+		if strings.Contains(err.Error(), "no space left") {
+			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}
+	}
+}
+
+func (mset *stream) setMirrorErr(err *ApiError) {
+	mset.mu.Lock()
+	mset.mirror.err = err
+	mset.mu.Unlock()
+}
+
+const mirrorMaxAckPending = 64
+
+func (mset *stream) cancelMirrorConsumer() {
+	mset.mu.Lock()
+	defer mset.mu.Unlock()
+	if mset.mirror != nil && mset.mirror.sub != nil {
+		mset.unsubscribe(mset.mirror.sub)
+		mset.mirror.sub = nil
+	}
+}
+
+func (mset *stream) resetMirrorConsumer() error {
+	mset.mu.Lock()
+	defer mset.mu.Unlock()
+	return mset.setupMirrorConsumer()
+}
+
+func (mset *stream) mirrorDurable() string {
+	return string(getHash(fmt.Sprintf("MIRROR:%s:%s", mset.cfg.Mirror.Name, mset.cfg.Name)))
+}
+
+// Setup our mirror consumer.
+// Lock should be held.
+func (mset *stream) setupMirrorConsumer() error {
+	if mset.sendq == nil {
+		return errors.New("sendq required")
+	}
+
+	// Reset
+	if mset.mirror != nil && mset.mirror.sub != nil {
+		mset.unsubscribe(mset.mirror.sub)
+		mset.mirror.sub = nil
+	}
+	sub, err := mset.subscribeInternal(syncSubject("$JS.M"), mset.processInboundMirrorMsg)
+	if err != nil {
+		return err
+	}
+	mset.mirror = &sourceInfo{name: mset.cfg.Mirror.Name, sub: sub}
+
+	// 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.
+
+	state := mset.store.State()
+	durable := mset.mirrorDurable()
+
+	// Make sure to delete any prior durable consumers.
+	subject := fmt.Sprintf(JSApiConsumerDeleteT, mset.cfg.Mirror.Name, durable)
+	mset.sendq <- &jsPubMsg{subject, _EMPTY_, _EMPTY_, nil, nil, nil, 0}
+
+	req := &CreateConsumerRequest{
+		Stream: mset.cfg.Mirror.Name,
+		Config: ConsumerConfig{
+			Durable:        durable,
+			DeliverSubject: string(sub.subject),
+			DeliverPolicy:  DeliverByStartSequence,
+			OptStartSeq:    state.LastSeq,
+			AckPolicy:      AckExplicit,
+			AckWait:        48 * time.Hour,
+			MaxDeliver:     1,
+			MaxAckPending:  mirrorMaxAckPending,
+		},
+	}
+	// Only use start optionals on first time.
+	if state.Msgs == 0 {
+		if mset.cfg.Mirror.OptStartSeq > 0 {
+			req.Config.OptStartSeq = mset.cfg.Mirror.OptStartSeq
+		} else if mset.cfg.Mirror.OptStartTime != nil {
+			req.Config.OptStartTime = mset.cfg.Mirror.OptStartTime
+			req.Config.DeliverPolicy = DeliverByStartTime
+		}
+	}
+	if req.Config.OptStartSeq == 0 && req.Config.OptStartTime == nil {
+		// If starting out and lastSeq is 0.
+		req.Config.DeliverPolicy = DeliverAll
+	}
+
+	respCh := make(chan *JSApiConsumerCreateResponse, 1)
+	reply := infoReplySubject()
+	mset.subscribeInternal(reply, func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+		mset.unsubscribe(sub)
+		_, msg := c.msgParts(rmsg)
+		var ccr JSApiConsumerCreateResponse
+		if err := json.Unmarshal(msg, &ccr); err != nil {
+			c.Warnf("JetStream bad mirror consumer create response: %q", msg)
+			mset.cancelMirrorConsumer()
+			mset.setMirrorErr(jsInvalidJSONErr)
+			return
+		}
+		respCh <- &ccr
+	})
+
+	b, _ := json.Marshal(req)
+	subject = fmt.Sprintf(JSApiDurableCreateT, mset.cfg.Mirror.Name, durable)
+	mset.sendq <- &jsPubMsg{subject, _EMPTY_, reply, nil, b, nil, 0}
+
+	go func() {
+		var shouldRetry bool
+		select {
+		case ccr := <-respCh:
+			if ccr.Error != nil {
+				mset.cancelMirrorConsumer()
+				shouldRetry = false
+			}
+			mset.setMirrorErr(ccr.Error)
+		case <-time.After(2 * time.Second):
+			shouldRetry = true
+		}
+		if shouldRetry {
+			mset.resetMirrorConsumer()
+		}
+	}()
+
+	return nil
+}
+
+func (mset *stream) sourceDurable(streamName string) string {
+	return string(getHash(fmt.Sprintf("SOURCE:%s:%s", streamName, mset.cfg.Name)))
+}
+
+func (mset *stream) streamSource(sname string) *StreamSource {
+	for _, ssi := range mset.cfg.Sources {
+		if ssi.Name == sname {
+			return ssi
+		}
+	}
+	return nil
+}
+
+// Lock should be held.
+func (mset *stream) setSourceConsumer(sname string, seq uint64) {
+	si := mset.sources[sname]
+	if si == nil {
+		return
+	}
+	if si.sub != nil {
+		mset.unsubscribe(si.sub)
+	}
+	si.sseq, si.dseq = 0, 0
+
+	durable := mset.sourceDurable(sname)
+
+	// Need to delete the old one.
+	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)
+	})
+	if err != nil {
+		si.err = jsError(err)
+		si.sub = nil
+		return
+	}
+	si.sub = sub
+
+	req := &CreateConsumerRequest{
+		Stream: sname,
+		Config: ConsumerConfig{
+			Durable:        durable,
+			DeliverSubject: dsubj,
+			AckPolicy:      AckExplicit,
+			AckWait:        48 * time.Hour,
+			MaxDeliver:     1,
+			MaxAckPending:  mirrorMaxAckPending,
+		},
+	}
+	// If starting, check any configs.
+	ssi := mset.streamSource(sname)
+	if seq <= 1 {
+		if ssi.OptStartSeq > 0 {
+			req.Config.OptStartSeq = ssi.OptStartSeq
+			req.Config.DeliverPolicy = DeliverByStartSequence
+		} else if ssi.OptStartTime != nil {
+			req.Config.OptStartTime = ssi.OptStartTime
+			req.Config.DeliverPolicy = DeliverByStartTime
+		}
+	} else {
+		req.Config.OptStartSeq = seq
+		req.Config.DeliverPolicy = DeliverByStartSequence
+	}
+	// Filters
+	if ssi.FilterSubject != _EMPTY_ {
+		req.Config.FilterSubject = ssi.FilterSubject
+	}
+
+	respCh := make(chan *JSApiConsumerCreateResponse, 1)
+	reply := infoReplySubject()
+	mset.subscribeInternal(reply, func(sub *subscription, c *client, subject, reply string, rmsg []byte) {
+		mset.unsubscribe(sub)
+		_, msg := c.msgParts(rmsg)
+		var ccr JSApiConsumerCreateResponse
+		if err := json.Unmarshal(msg, &ccr); err != nil {
+			c.Warnf("JetStream bad source consumer create response: %q", msg)
+			return
+		}
+		respCh <- &ccr
+	})
+
+	b, _ := json.Marshal(req)
+	subject = fmt.Sprintf(JSApiDurableCreateT, sname, durable)
+	mset.sendq <- &jsPubMsg{subject, _EMPTY_, reply, nil, b, nil, 0}
+
+	go func() {
+		var shouldRetry bool
+		select {
+		case ccr := <-respCh:
+			mset.mu.Lock()
+			if si := mset.sources[sname]; si != nil {
+				if ccr.Error != nil {
+					mset.srv.Warnf("JetStream error response for create source consumer: %+v", ccr.Error)
+					si.err = ccr.Error
+					shouldRetry = false
+				}
+			}
+			mset.mu.Unlock()
+		case <-time.After(2 * time.Second):
+			shouldRetry = true
+		}
+		if shouldRetry {
+			// Make sure things have not changed.
+			mset.mu.Lock()
+			if si := mset.sources[sname]; si != nil {
+				if si.sub != nil && si.sub == sub {
+					mset.setSourceConsumer(sname, seq)
+				}
+			}
+			mset.mu.Unlock()
+		}
+	}()
+}
+
+// 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)
+
+	// Tracking is done here.
+	mset.mu.Lock()
+	if si == nil || sub != si.sub || dc > 1 {
+		mset.mu.Unlock()
+		return
+	}
+	if dseq == si.dseq+1 {
+		si.dseq++
+		si.sseq++
+		if dseq == 1 {
+			si.sseq = sseq
+		}
+	} else {
+		mset.setSourceConsumer(si.name, si.sseq+1)
+		mset.mu.Unlock()
+		return
+	}
+
+	si.last = time.Now()
+	si.lag = pending
+	isClustered := mset.node != nil
+	sendq := mset.sendq
+	mset.mu.Unlock()
+
+	// Hold onto the origin reply which has all he metadata.
+	hdr, msg := c.msgParts(rmsg)
+	hdr = genHeader(hdr, JSStreamSource, 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)
+	} else {
+		err = mset.processJetStreamMsg(subject, _EMPTY_, hdr, msg, 0, 0)
+	}
+	if err != nil {
+		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) {
+	tsa := [expectedNumReplyTokens]string{}
+	start, tokens := 0, tsa[:0]
+	for i := 0; i < len(subject); i++ {
+		if subject[i] == btsep {
+			tokens = append(tokens, subject[start:i])
+			start = i + 1
+		}
+	}
+	tokens = append(tokens, subject[start:])
+	if len(tokens) != expectedNumReplyTokens || tokens[0] != "$JS" || tokens[1] != "ACK" {
+		return _EMPTY_, 0
+	}
+	return tokens[2], uint64(parseAckReplyNum(tokens[5]))
+}
+
+// Lock should be held.
+// This will do a reverse scan on startup or leader election
+// searching for the starting sequence number.
+// This can be slow in degenerative cases.
+// Lock should be held.
+func (mset *stream) startingSequenceForSources() {
+	mset.sources = make(map[string]*sourceInfo)
+	if len(mset.cfg.Sources) == 0 {
+		return
+	}
+	for _, ssi := range mset.cfg.Sources {
+		si := &sourceInfo{name: ssi.Name}
+		mset.sources[ssi.Name] = si
+	}
+
+	state := mset.store.State()
+	if state.Msgs == 0 {
+		return
+	}
+	// For short circuiting return.
+	expected := len(mset.cfg.Sources)
+	seqs := make(map[string]uint64)
+
+	// Stamp our si seq records on the way out.
+	defer func() {
+		for sname, seq := range seqs {
+			if si := mset.sources[sname]; si != nil {
+				si.sseq = seq
+				si.dseq = 1
+			}
+		}
+	}()
+
+	for seq := state.LastSeq; seq >= state.FirstSeq; seq-- {
+		_, hdr, _, _, err := mset.store.LoadMsg(seq)
+		if err != nil || len(hdr) == 0 {
+			continue
+		}
+		reply := getHeader(JSStreamSource, hdr)
+		if len(reply) == 0 {
+			continue
+		}
+		name, sseq := streamAndSeq(string(reply))
+		// Only update active in case we have older ones in here that got configured out.
+		if si := mset.sources[name]; si != nil {
+			if _, ok := seqs[name]; !ok {
+				seqs[name] = sseq
+				if len(seqs) == expected {
+					return
+				}
+			}
+		}
+	}
+}
+
+// Setup our source consumers.
+// Lock should be held.
+func (mset *stream) setupSourceConsumers() error {
+	if mset.sendq == nil {
+		return errors.New("sendq required")
+	}
+	// Reset if needed.
+	for _, si := range mset.sources {
+		if si.sub != nil {
+			mset.unsubscribe(si.sub)
+		}
+	}
+
+	mset.startingSequenceForSources()
+
+	// Setup our consumers at the proper starting position.
+	for _, ssi := range mset.cfg.Sources {
+		if si := mset.sources[ssi.Name]; si != nil {
+			mset.setSourceConsumer(ssi.Name, si.sseq+1)
+		}
+	}
+
+	return nil
+}
+
 // Will create internal subscriptions for the stream.
 // Lock should be held.
 func (mset *stream) subscribeToStream() error {
@@ -842,16 +1389,54 @@ func (mset *stream) subscribeToStream() error {
 			return err
 		}
 	}
+	// Check if we need to setup mirroring.
+	if mset.cfg.Mirror != nil {
+		if err := mset.setupMirrorConsumer(); err != nil {
+			return err
+		}
+	} else if len(mset.cfg.Sources) > 0 {
+		if err := mset.setupSourceConsumers(); err != nil {
+			return err
+		}
+	}
+
 	mset.active = true
 	return nil
 }
 
+// Stop our source consumers.
+// Lock should be held.
+func (mset *stream) stopSourceConsumers() {
+	for _, si := range mset.sources {
+		if si.sub != nil {
+			mset.unsubscribe(si.sub)
+		}
+		// Need to delete the old one.
+		subject := fmt.Sprintf(JSApiConsumerDeleteT, si.name, mset.sourceDurable(si.name))
+		mset.sendq <- &jsPubMsg{subject, _EMPTY_, _EMPTY_, nil, nil, nil, 0}
+	}
+}
+
 // Will unsubscribe from the stream.
 // Lock should be held.
 func (mset *stream) unsubscribeToStream() error {
 	for _, subject := range mset.cfg.Subjects {
 		mset.unsubscribeInternal(subject)
 	}
+	if mset.mirror != nil {
+		if mset.mirror.sub != nil {
+			mset.unsubscribe(mset.mirror.sub)
+		}
+		durable := mset.mirrorDurable()
+		subject := fmt.Sprintf(JSApiConsumerDeleteT, mset.cfg.Mirror.Name, durable)
+		mset.sendq <- &jsPubMsg{subject, _EMPTY_, _EMPTY_, nil, nil, nil, 0}
+		mset.mirror = nil
+	}
+
+	if len(mset.cfg.Sources) > 0 {
+		mset.stopSourceConsumers()
+	}
+
 	mset.active = false
 	return nil
 }
@@ -913,7 +1498,7 @@ func (mset *stream) unsubscribe(sub *subscription) {
 	if sub == nil || mset.client == nil {
 		return
 	}
-	mset.client.unsubscribe(mset.client.acc, sub, true, true)
+	mset.client.processUnsub(sub.sid)
 }
 
 func (mset *stream) unsubscribeUnlocked(sub *subscription) {
@@ -965,7 +1550,7 @@ func (mset *stream) storeUpdates(md, bd int64, seq uint64, subj string) {
 	}
 
 	if mset.jsa != nil {
-		mset.jsa.updateUsage(mset.cfg.Storage, bd)
+		mset.jsa.updateUsage(mset.stype, bd)
 	}
 }
 
@@ -1099,8 +1684,8 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	}
 
 	var accName string
-	if c != nil && c.acc != nil {
-		accName = c.acc.Name
+	if mset.acc != nil {
+		accName = mset.acc.Name
 	}
 
 	doAck := !mset.cfg.NoAck
@@ -1118,17 +1703,27 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	var resp = &JSPubAckResponse{}
 
 	// For clustering the lower layers will pass our expected lseq. If it is present check for that here.
-	// This is from the clustering layers so will not respond here.
 	if lseq > 0 && lseq != (mset.lseq+mset.clfs) {
-		sendq := mset.sendq
-		mset.mu.Unlock()
-		if canRespond && sendq != nil {
-			resp.PubAck = &PubAck{Stream: name}
-			resp.Error = &ApiError{Code: 503, Description: "expected stream sequence does not match"}
-			b, _ := json.Marshal(resp)
-			sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0}
+		isMisMatch := true
+		// If our first message for this mirror, see if we have to adjust our starting sequence.
+		if mset.cfg.Mirror != nil {
+			if state := mset.store.State(); state.FirstSeq == 0 {
+				mset.store.Compact(lseq + 1)
+				mset.lseq = lseq
+				isMisMatch = false
+			}
+		}
+		if isMisMatch {
+			sendq := mset.sendq
+			mset.mu.Unlock()
+			if canRespond && sendq != nil {
+				resp.PubAck = &PubAck{Stream: name}
+				resp.Error = &ApiError{Code: 503, Description: "expected stream sequence does not match"}
+				b, _ := json.Marshal(resp)
+				sendq <- &jsPubMsg{reply, _EMPTY_, _EMPTY_, nil, b, nil, 0}
+			}
+			return errLastSeqMismatch
 		}
-		return errLastSeqMismatch
 	}
 
 	// Process msg headers if present.
@@ -1254,28 +1849,29 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 	mset.lmsgId = msgId
 	mset.lseq++
 
-	// Set sequence if not set.
-	if seq == 0 {
-		seq = mset.lseq
-	}
-
 	// We hold the lock to this point to make sure nothing gets between us since we check for pre-conditions.
 	// Currently can not hold while calling store b/c we have inline storage update calls that may need the lock.
+	// Note that upstream that sets seq/ts should be serialized as much as possible.
 	mset.mu.Unlock()
 
 	// Store actual msg.
-	err = store.StoreRawMsg(subject, hdr, msg, seq, ts)
+	if seq == 0 && ts == 0 {
+		seq, _, err = store.StoreMsg(subject, hdr, msg)
+	} else {
+		if seq == 0 {
+			seq = olseq + 1
+		}
+		err = store.StoreRawMsg(subject, hdr, msg, seq, ts)
+	}
 
-	// If we did not succeed put those values back.
 	if err != nil {
+		// If we did not succeed put those values back.
 		// FIXME(dlc) - This most likely is asymmetric under clustered scenarios.
 		mset.mu.Lock()
 		mset.lseq = olseq
 		mset.lmsgId = olmsgId
 		mset.mu.Unlock()
-	}
 
-	if err != nil {
 		if err != ErrStoreClosed {
 			c.Errorf("JetStream failed to store a msg on stream '%s > %s' -  %v", accName, name, err)
 		}
@@ -1356,7 +1952,7 @@ type StoredMsg struct {
 }
 
 // TODO(dlc) - Maybe look at onering instead of chan - https://github.com/pltr/onering
-const msetSendQSize = 1024
+const msetSendQSize = 4096
 
 // 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.
@@ -1382,12 +1978,10 @@ func (mset *stream) name() string {
 
 func (mset *stream) internalSendLoop() {
 	mset.mu.RLock()
-	c := mset.client
-	if c == nil {
-		mset.mu.RUnlock()
-		return
-	}
-	s := c.srv
+	s := mset.srv
+	c := s.createInternalJetStreamClient()
+	c.registerWithAccount(mset.acc)
+	defer c.closeConnection(ClientClosed)
 	sendq := mset.sendq
 	name := mset.cfg.Name
 	mset.mu.RUnlock()
@@ -1478,6 +2072,11 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 
 	mset.mu.Lock()
 
+	// Stop responding to sync requests.
+	mset.stopClusterSubs()
+	// Unsubscribe from direct stream.
+	mset.unsubscribeToStream()
+
 	// Quit channel.
 	if mset.qch != nil {
 		close(mset.qch)
@@ -1491,7 +2090,13 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 		} else {
 			n.Stop()
 		}
-		mset.stopClusterSubs()
+	}
+
+	if deleteFlag {
+		for _, ssi := range mset.cfg.Sources {
+			subject := fmt.Sprintf(JSApiConsumerDeleteT, ssi.Name, mset.sourceDurable(ssi.Name))
+			mset.sendq <- &jsPubMsg{subject, _EMPTY_, _EMPTY_, nil, nil, nil, 0}
+		}
 	}
 
 	// Send stream delete advisory after the consumers.
diff --git a/server/sublist.go b/server/sublist.go
index d47e91ef..c9df0b60 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -247,13 +247,14 @@ func (s *Sublist) addRemoveNotify(subject string, notify chan<- bool) error {
 func (s *Sublist) addNotify(m map[string][]chan<- bool, subject string, notify chan<- bool) error {
 	chs := m[subject]
 	if len(chs) > 0 {
-		// Check to see if this chan is alredy registered.
+		// Check to see if this chan is already registered.
 		for _, ch := range chs {
 			if ch == notify {
 				return ErrAlreadyRegistered
 			}
 		}
 	}
+
 	m[subject] = append(chs, notify)
 	return nil
 }


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 34d2de57..03c9ec0e 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -294,51 +294,6 @@ func TestJetStreamClusterMemoryStore(t *testing.T) {
 	checkSubsPending(t, sub, toSend)
 }
 
-func TestJetStreamClusterCompaction(t *testing.T) {
-	// This test takes a long time to observe compactions.
-	// Once moved to server we can adjust and re-enable.
-	skip(t)
-
-	c := createJetStreamClusterExplicit(t, "JSC", 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", Replicas: 2}); err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-
-	toSend := 1000
-	payload := []byte("Hello JSC")
-	for i := 0; i < toSend; i++ {
-		if _, err := js.Publish("TEST", payload); err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-	}
-
-	if _, err := js.StreamInfo("TEST"); err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	sub, err := js.SubscribeSync("TEST")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	defer sub.Unsubscribe()
-
-	checkSubsPending(t, sub, toSend)
-
-	for i := 0; i < toSend; i++ {
-		m, err := sub.NextMsg(time.Second)
-		if err != nil {
-			t.Fatalf("Unexpected error getting msg %d: %v", i+1, err)
-		}
-		m.Ack()
-	}
-}
-
 func TestJetStreamClusterDelete(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "RNS", 3)
 	defer c.shutdown()
@@ -474,6 +429,52 @@ func TestJetStreamClusterStreamPurge(t *testing.T) {
 	}
 }
 
+func TestJetStreamClusterStreamUpdateSubjects(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", "bar"},
+		Replicas: 3,
+	}
+
+	if _, err := js.AddStream(cfg); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Make sure we can update subjects.
+	cfg.Subjects = []string{"bar", "baz"}
+
+	si, err := js.UpdateStream(cfg)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if si == nil {
+		t.Fatalf("Expected a stream info, got none")
+	}
+	if !reflect.DeepEqual(si.Config.Subjects, cfg.Subjects) {
+		t.Fatalf("Expected subjects to be updated: got %+v", si.Config.Subjects)
+	}
+	// Make sure it registered
+	js2, err := nc.JetStream(nats.MaxWait(50 * time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	if _, err = js2.Publish("foo", nil); err == nil {
+		t.Fatalf("Expected this to fail")
+	}
+	if _, err = js2.Publish("baz", nil); err != nil {
+		t.Fatalf("Unexpected publish error: %v", err)
+	}
+}
+
 func TestJetStreamClusterConsumerState(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
@@ -820,6 +821,7 @@ func TestJetStreamClusterRestoreSingleConsumer(t *testing.T) {
 
 	c.stopAll()
 	c.restartAll()
+	c.waitOnLeader()
 
 	s = c.randomServer()
 	nc, js = jsClientConnect(t, s)
@@ -1579,7 +1581,7 @@ func TestJetStreamClusterExtendedStreamInfo(t *testing.T) {
 	}
 
 	// Faster timeout since we loop below checking for condition.
-	js2, err := nc.JetStream(nats.MaxWait(10 * time.Millisecond))
+	js2, err := nc.JetStream(nats.MaxWait(50 * time.Millisecond))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1815,7 +1817,7 @@ func TestJetStreamClusterInterestRetention(t *testing.T) {
 	}
 	m.Ack()
 
-	js, err = nc.JetStream(nats.MaxWait(10 * time.Millisecond))
+	js, err = nc.JetStream(nats.MaxWait(50 * time.Millisecond))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1875,7 +1877,7 @@ func TestJetStreamClusterInterestRetentionWithFilteredConsumers(t *testing.T) {
 		m.Ack()
 	}
 
-	jsq, err := nc.JetStream(nats.MaxWait(10 * time.Millisecond))
+	jsq, err := nc.JetStream(nats.MaxWait(50 * time.Millisecond))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -2047,6 +2049,7 @@ func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 		}
 	}
 	nc.Flush()
+	time.Sleep(200 * time.Millisecond)
 
 	// Snapshot consumer info.
 	ci, err := jsub.ConsumerInfo()
@@ -2818,6 +2821,7 @@ func TestJetStreamClusterCreateResponseAdvisoriesHaveSubject(t *testing.T) {
 	}
 
 	checkSubsPending(t, sub, 6)
+
 	for m, err := sub.NextMsg(0); err == nil; m, err = sub.NextMsg(0) {
 		var audit JSAPIAudit
 		if err := json.Unmarshal(m.Data, &audit); err != nil {
@@ -2930,6 +2934,7 @@ func TestJetStreamClusterRestartAndRemoveAdvisories(t *testing.T) {
 	nc.Flush()
 
 	c.waitOnConsumerLeader("$G", "TEST-1", "DC")
+	c.waitOnLeader()
 
 	// Now check delete advisories as well.
 	if err := js.DeleteConsumer("TEST-1", "DC"); err != nil {
@@ -2947,7 +2952,7 @@ func TestJetStreamClusterRestartAndRemoveAdvisories(t *testing.T) {
 	}
 
 	checkSubsPending(t, dsub, 2) // Stream and the consumer underneath.
-	checkSubsPending(t, sub, 4)
+	checkSubsPending(t, sub, 2)
 }
 
 func TestJetStreamClusterNoDuplicateOnNodeRestart(t *testing.T) {
@@ -3123,7 +3128,7 @@ func TestJetStreamClusterRemovePeer(t *testing.T) {
 	}
 
 	// Grab shorter timeout jetstream context.
-	js, err = nc.JetStream(nats.MaxWait(100 * time.Millisecond))
+	js, err = nc.JetStream(nats.MaxWait(50 * time.Millisecond))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -3221,7 +3226,7 @@ func TestJetStreamClusterStreamLeaderStepDown(t *testing.T) {
 	}
 
 	// Grab shorter timeout jetstream context.
-	js, err = nc.JetStream(nats.MaxWait(100 * time.Millisecond))
+	js, err = nc.JetStream(nats.MaxWait(50 * time.Millisecond))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -3318,7 +3323,7 @@ func TestJetStreamClusterRemoveServer(t *testing.T) {
 	c.waitOnStreamLeader("$G", "TEST")
 
 	// Faster timeout since we loop below checking for condition.
-	js, err = nc.JetStream(nats.MaxWait(10 * time.Millisecond))
+	js, err = nc.JetStream(nats.MaxWait(50 * time.Millisecond))
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -4216,7 +4221,7 @@ func jsClientConnect(t *testing.T, s *Server) (*nats.Conn, nats.JetStreamContext
 
 func checkSubsPending(t *testing.T, sub *nats.Subscription, numExpected int) {
 	t.Helper()
-	checkFor(t, 3*time.Second, 10*time.Millisecond, func() error {
+	checkFor(t, 4*time.Second, 20*time.Millisecond, func() error {
 		if nmsgs, _, err := sub.Pending(); err != nil || nmsgs != numExpected {
 			return fmt.Errorf("Did not receive correct number of messages: %d vs %d", nmsgs, numExpected)
 		}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 4fd400b2..ff2a04e3 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -463,7 +463,7 @@ func TestJetStreamConsumerWithStartTime(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
-			sseq, dseq, _, _, _ := o.replyInfo(msg.Reply)
+			sseq, dseq, _, _, _ := replyInfo(msg.Reply)
 			if dseq != 1 {
 				t.Fatalf("Expected delivered seq of 1, got %d", dseq)
 			}
@@ -2507,7 +2507,7 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
-				_, _, _, _, pending := o.replyInfo(m.Reply)
+				_, _, _, _, pending := replyInfo(m.Reply)
 				if pending != uint64(ep) {
 					t.Fatalf("Expected ack reply pending of %d, got %d - reply: %q", ep, pending, m.Reply)
 				}
@@ -2748,7 +2748,7 @@ func TestJetStreamWorkQueueAckWaitRedelivery(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Unexpected error waiting for message[%d]: %v", i, err)
 				}
-				sseq, dseq, dcount, _, _ := o.replyInfo(m.Reply)
+				sseq, dseq, dcount, _, _ := replyInfo(m.Reply)
 				if sseq != uint64(i) {
 					t.Fatalf("Expected set sequence of %d , got %d", i, sseq)
 				}
@@ -2826,7 +2826,7 @@ func TestJetStreamWorkQueueNakRedelivery(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
-				rsseq, rdseq, _, _, _ := o.replyInfo(m.Reply)
+				rsseq, rdseq, _, _, _ := replyInfo(m.Reply)
 				if rdseq != uint64(dseq) {
 					t.Fatalf("Expected delivered sequence of %d , got %d", dseq, rdseq)
 				}
@@ -2906,7 +2906,7 @@ func TestJetStreamWorkQueueWorkingIndicator(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
-				rsseq, rdseq, _, _, _ := o.replyInfo(m.Reply)
+				rsseq, rdseq, _, _, _ := replyInfo(m.Reply)
 				if rdseq != uint64(dseq) {
 					t.Fatalf("Expected delivered sequence of %d , got %d", dseq, rdseq)
 				}
@@ -2993,7 +2993,7 @@ func TestJetStreamWorkQueueTerminateDelivery(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
-				rsseq, rdseq, _, _, _ := o.replyInfo(m.Reply)
+				rsseq, rdseq, _, _, _ := replyInfo(m.Reply)
 				if rdseq != uint64(dseq) {
 					t.Fatalf("Expected delivered sequence of %d , got %d", dseq, rdseq)
 				}
@@ -3504,7 +3504,7 @@ func TestJetStreamPullConsumerRemoveInterest(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	_, dseq, dc, _, _ := o.replyInfo(msg.Reply)
+	_, dseq, dc, _, _ := replyInfo(msg.Reply)
 	if dseq != 1 {
 		t.Fatalf("Expected consumer sequence of 1, got %d", dseq)
 	}
@@ -3530,7 +3530,7 @@ func TestJetStreamPullConsumerRemoveInterest(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	_, dseq, dc, _, _ = o.replyInfo(msg.Reply)
+	_, dseq, dc, _, _ = replyInfo(msg.Reply)
 	if dseq != 2 {
 		t.Fatalf("Expected consumer sequence of 2, got %d", dseq)
 	}
@@ -4980,7 +4980,7 @@ func TestJetStreamDurableConsumerReconnectWithOnlyPending(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
-				sseq, _, dc, _, _ := o.replyInfo(msg.Reply)
+				sseq, _, dc, _, _ := replyInfo(msg.Reply)
 				if sseq == 1 && dc == 1 {
 					t.Fatalf("Expected a redelivery count greater then 1 for sseq 1, got %d", dc)
 				}
@@ -5041,7 +5041,7 @@ func TestJetStreamDurableFilteredSubjectConsumerReconnect(t *testing.T) {
 			dsubj := nats.NewInbox()
 
 			// Now create an consumer for foo.AA, only requesting the last one.
-			o, err := mset.addConsumer(&ConsumerConfig{
+			_, err = mset.addConsumer(&ConsumerConfig{
 				Durable:        dname,
 				DeliverSubject: dsubj,
 				FilterSubject:  "foo.AA",
@@ -5065,7 +5065,7 @@ func TestJetStreamDurableFilteredSubjectConsumerReconnect(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
-				rsseq, roseq, dcount, _, _ := o.replyInfo(m.Reply)
+				rsseq, roseq, dcount, _, _ := replyInfo(m.Reply)
 				if roseq != uint64(seq) {
 					t.Fatalf("Expected consumer sequence of %d , got %d", seq, roseq)
 				}
@@ -5084,7 +5084,7 @@ func TestJetStreamDurableFilteredSubjectConsumerReconnect(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
-				_, roseq, dcount, _, _ := o.replyInfo(m.Reply)
+				_, roseq, dcount, _, _ := replyInfo(m.Reply)
 				if roseq != uint64(seq) {
 					t.Fatalf("Expected consumer sequence of %d , got %d", seq, roseq)
 				}
@@ -5250,7 +5250,7 @@ func TestJetStreamMetadata(t *testing.T) {
 					t.Fatalf("Unexpected error: %v", err)
 				}
 
-				sseq, dseq, dcount, ts, _ := o.replyInfo(m.Reply)
+				sseq, dseq, dcount, ts, _ := replyInfo(m.Reply)
 
 				mreq := &JSApiMsgGetRequest{Seq: sseq}
 				req, err := json.Marshal(mreq)
@@ -5356,7 +5356,7 @@ func TestJetStreamRedeliverCount(t *testing.T) {
 					t.Fatalf("Unexpected error: %v", err)
 				}
 
-				sseq, dseq, dcount, _, _ := o.replyInfo(m.Reply)
+				sseq, dseq, dcount, _, _ := replyInfo(m.Reply)
 
 				// Make sure we keep getting stream sequence #1
 				if sseq != 1 {
@@ -9483,7 +9483,7 @@ func TestJetStreamConsumerUpdateRedelivery(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Error getting message: %v", err)
 				}
-				seq, _, _, _, _ := o.replyInfo(m.Reply)
+				seq, _, _, _, _ := replyInfo(m.Reply)
 				// 4, 8, 12, 16, 20
 				if seq%4 == 0 {
 					m.Respond(nil)
@@ -9541,7 +9541,7 @@ func TestJetStreamConsumerUpdateRedelivery(t *testing.T) {
 				if eseq <= uint64(toSend) && eseq%4 == 0 {
 					eseq++
 				}
-				seq, _, dc, _, _ := o.replyInfo(m.Reply)
+				seq, _, dc, _, _ := replyInfo(m.Reply)
 				if seq != eseq {
 					t.Fatalf("Expected stream sequence of %d, got %d", eseq, seq)
 				}
@@ -9575,7 +9575,7 @@ func TestJetStreamConsumerUpdateRedelivery(t *testing.T) {
 				if eseq <= uint64(toSend) && eseq%4 == 0 {
 					eseq++
 				}
-				seq, _, dc, _, _ := o.replyInfo(m.Reply)
+				seq, _, dc, _, _ := replyInfo(m.Reply)
 				if seq != eseq {
 					t.Fatalf("Expected stream sequence of %d, got %d", eseq, seq)
 				}
@@ -9890,7 +9890,7 @@ func TestJetStreamPullConsumerMaxAckPendingRedeliveries(t *testing.T) {
 				if err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
-				sseq, dseq, dcount, _, pending := o.replyInfo(m.Reply)
+				sseq, dseq, dcount, _, pending := replyInfo(m.Reply)
 				if sseq != expSeq {
 					t.Fatalf("Expected stream sequence of %d, got %d", expSeq, sseq)
 				}
@@ -10390,6 +10390,332 @@ func TestJetStreamConfigReloadWithGlobalAccount(t *testing.T) {
 	checkJSAccount()
 }
 
+func TestJetStreamMirrorBasics(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer os.RemoveAll(config.StoreDir)
+	}
+
+	// Client for API requests.
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	createStream := func(cfg *StreamConfig) *JSApiStreamCreateResponse {
+		t.Helper()
+		req, err := json.Marshal(cfg)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		rm, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		var resp JSApiStreamCreateResponse
+		if err := json.Unmarshal(rm.Data, &resp); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		return &resp
+	}
+
+	createStreamOk := func(cfg *StreamConfig) {
+		t.Helper()
+		if scr := createStream(cfg); scr.Error != nil {
+			t.Fatalf("Expected error, got %+v", scr.Error)
+		}
+	}
+
+	// Test we get right config errors etc.
+	cfg := &StreamConfig{
+		Name:     "M1",
+		Storage:  FileStorage,
+		Subjects: []string{"foo", "bar", "baz"},
+		Mirror:   &StreamSource{Name: "S1"},
+	}
+
+	scr := createStream(cfg)
+	if scr.Error == nil || !strings.Contains(scr.Error.Description, "stream mirrors can not") {
+		t.Fatalf("Expected error, got %+v", scr.Error)
+	}
+
+	// Clear subjects.
+	cfg.Subjects = nil
+	scr = createStream(cfg)
+	if scr.Error == nil || !strings.Contains(scr.Error.Description, "stream mirror source must exist") {
+		t.Fatalf("Expected error, got %+v", scr.Error)
+	}
+
+	scfg := &StreamConfig{
+		Name:     "S1",
+		Storage:  FileStorage,
+		Subjects: []string{"foo", "bar", "baz"},
+	}
+	// Create source stream
+	createStreamOk(scfg)
+	// Now create our mirror stream.
+	createStreamOk(cfg)
+
+	// Send 100 messages.
+	for i := 0; i < 100; i++ {
+		if _, err := js.Publish("foo", []byte("OK")); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+
+	// Faster timeout since we loop below checking for condition.
+	js2, err := nc.JetStream(nats.MaxWait(10 * time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("M1")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 100 {
+			return fmt.Errorf("Expected 100 msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
+
+	// Purge the source stream.
+	if err := js.PurgeStream("S1"); err != nil {
+		t.Fatalf("Unexpected purge error: %v", err)
+	}
+	// Send 50 more msgs now.
+	for i := 0; i < 50; i++ {
+		if _, err := js.Publish("bar", []byte("OK")); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+
+	cfg = &StreamConfig{
+		Name:    "M2",
+		Storage: FileStorage,
+		Mirror:  &StreamSource{Name: "S1"},
+	}
+
+	// Now create our second mirror stream.
+	createStreamOk(cfg)
+
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("M2")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 50 {
+			return fmt.Errorf("Expected 50 msgs, got state: %+v", si.State)
+		}
+		if si.State.FirstSeq != 101 {
+			return fmt.Errorf("Expected start seq of 101, got state: %+v", si.State)
+		}
+		return nil
+	})
+
+	// Send 100 more msgs now. Should be 150 total, 101 first.
+	for i := 0; i < 100; i++ {
+		if _, err := js.Publish("baz", []byte("OK")); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+
+	cfg = &StreamConfig{
+		Name:    "M3",
+		Storage: FileStorage,
+		Mirror:  &StreamSource{Name: "S1", OptStartSeq: 150},
+	}
+
+	createStreamOk(cfg)
+
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("M3")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 101 {
+			return fmt.Errorf("Expected 101 msgs, got state: %+v", si.State)
+		}
+		if si.State.FirstSeq != 150 {
+			return fmt.Errorf("Expected start seq of 150, got state: %+v", si.State)
+		}
+		return nil
+	})
+
+	// Make sure setting time works ok.
+	start := time.Now().Add(-2 * time.Hour)
+	cfg = &StreamConfig{
+		Name:    "M4",
+		Storage: FileStorage,
+		Mirror:  &StreamSource{Name: "S1", OptStartTime: &start},
+	}
+	createStreamOk(cfg)
+
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("M4")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 150 {
+			return fmt.Errorf("Expected 150 msgs, got state: %+v", si.State)
+		}
+		if si.State.FirstSeq != 101 {
+			return fmt.Errorf("Expected start seq of 101, got state: %+v", si.State)
+		}
+		return nil
+	})
+}
+
+func TestJetStreamSourceBasics(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer os.RemoveAll(config.StoreDir)
+	}
+
+	// Client for API requests.
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	createStream := func(cfg *StreamConfig) {
+		t.Helper()
+		req, err := json.Marshal(cfg)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		rm, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		var resp JSApiStreamCreateResponse
+		if err := json.Unmarshal(rm.Data, &resp); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if resp.Error != nil {
+			t.Fatalf("Unexpected error: %+v", resp.Error)
+		}
+	}
+
+	for _, sname := range []string{"foo", "bar", "baz"} {
+		if _, err := js.AddStream(&nats.StreamConfig{Name: sname}); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	}
+	sendBatch := func(subject string, n int) {
+		for i := 0; i < n; i++ {
+			if _, err := js.Publish(subject, []byte("OK")); err != nil {
+				t.Fatalf("Unexpected publish error: %v", err)
+			}
+		}
+	}
+	// Populate each one.
+	sendBatch("foo", 10)
+	sendBatch("bar", 15)
+	sendBatch("baz", 25)
+
+	cfg := &StreamConfig{
+		Name:    "MS",
+		Storage: FileStorage,
+		Sources: []*StreamSource{
+			&StreamSource{Name: "foo"},
+			&StreamSource{Name: "bar"},
+			&StreamSource{Name: "baz"},
+		},
+	}
+
+	createStream(cfg)
+
+	// Faster timeout since we loop below checking for condition.
+	js2, err := nc.JetStream(nats.MaxWait(50 * time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("MS")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 50 {
+			return fmt.Errorf("Expected 50 msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
+
+	// 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)
+	}
+	sendBatch("dlc", 20)
+	sendBatch("rip", 20)
+	sendBatch("dlc", 10)
+
+	cfg = &StreamConfig{
+		Name:    "FMS",
+		Storage: FileStorage,
+		Sources: []*StreamSource{
+			&StreamSource{Name: "TEST", OptStartSeq: 26},
+		},
+	}
+	createStream(cfg)
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("FMS")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 25 {
+			return fmt.Errorf("Expected 25 msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
+	// Double check first starting.
+	m, err := js.GetMsg("FMS", 1)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if reply := m.Header.Get(JSStreamSource); reply == _EMPTY_ {
+		t.Fatalf("Expected a header, got none")
+	} else {
+		sseq, _, _, _, _ := replyInfo(reply)
+		if sseq != 26 {
+			t.Fatalf("Expected header sequence of 26, got %d", sseq)
+		}
+	}
+
+	// Test Filters
+	cfg = &StreamConfig{
+		Name:    "FMS2",
+		Storage: FileStorage,
+		Sources: []*StreamSource{
+			&StreamSource{Name: "TEST", OptStartSeq: 11, FilterSubject: "dlc"},
+		},
+	}
+	createStream(cfg)
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("FMS2")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 20 {
+			return fmt.Errorf("Expected 20 msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
+
+	// Double check first starting.
+	if m, err = js.GetMsg("FMS2", 1); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if reply := m.Header.Get(JSStreamSource); reply == _EMPTY_ {
+		t.Fatalf("Expected a header, got none")
+	} else {
+		sseq, _, _, _, _ := replyInfo(reply)
+		if sseq != 11 {
+			t.Fatalf("Expected header sequence of 11, got %d", sseq)
+		}
+	}
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Simple JetStream Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/norace_test.go b/server/norace_test.go
index 77cc0c78..d7ce8fcd 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -1436,3 +1436,300 @@ func TestJetStreamClusterStreamCreateAndLostQuorum(t *testing.T) {
 	c.waitOnStreamLeader("$G", "NO-LQ-START")
 	checkSubsPending(t, sub, 0)
 }
+
+func TestJetStreamClusterMirrors(t *testing.T) {
+	sc := createJetStreamSuperCluster(t, 3, 3)
+	defer sc.shutdown()
+
+	// Client based API
+	s := sc.clusterForName("C2").randomServer()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	// Create source stream.
+	_, err := js.AddStream(&nats.StreamConfig{Name: "S1", Subjects: []string{"foo", "bar"}, Replicas: 3, Placement: &nats.Placement{Cluster: "C2"}})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Needed while Go client does not have mirror support.
+	createStream := func(cfg *StreamConfig) {
+		t.Helper()
+		req, err := json.Marshal(cfg)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		rm, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		var resp JSApiStreamCreateResponse
+		if err := json.Unmarshal(rm.Data, &resp); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if resp.Error != nil {
+			t.Fatalf("Unexpected error: %+v", resp.Error)
+		}
+	}
+
+	// Send 100 messages.
+	for i := 0; i < 100; i++ {
+		if _, err := js.Publish("foo", []byte("MIRRORS!")); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+
+	createStream(&StreamConfig{
+		Name:      "M1",
+		Storage:   FileStorage,
+		Mirror:    &StreamSource{Name: "S1"},
+		Placement: &Placement{Cluster: "C1"},
+	})
+
+	// Faster timeout since we loop below checking for condition.
+	js2, err := nc.JetStream(nats.MaxWait(50 * time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("M1")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 100 {
+			return fmt.Errorf("Expected 100 msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
+
+	// Purge the source stream.
+	if err := js.PurgeStream("S1"); err != nil {
+		t.Fatalf("Unexpected purge error: %v", err)
+	}
+	// Send 50 more msgs now.
+	for i := 0; i < 50; i++ {
+		if _, err := js.Publish("bar", []byte("OK")); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+
+	createStream(&StreamConfig{
+		Name:      "M2",
+		Storage:   FileStorage,
+		Mirror:    &StreamSource{Name: "S1"},
+		Replicas:  3,
+		Placement: &Placement{Cluster: "C3"},
+	})
+
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("M2")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 50 {
+			return fmt.Errorf("Expected 50 msgs, got state: %+v", si.State)
+		}
+		if si.State.FirstSeq != 101 {
+			return fmt.Errorf("Expected start seq of 101, got state: %+v", si.State)
+		}
+		return nil
+	})
+
+	sl := sc.clusterForName("C3").streamLeader("$G", "M2")
+	doneCh := make(chan bool)
+
+	// Now test that if the mirror get's interrupted that it picks up where it left off etc.
+	go func() {
+		// Send 100 more messages.
+		for i := 0; i < 100; i++ {
+			if _, err := js.Publish("foo", []byte("MIRRORS!")); err != nil {
+				t.Errorf("Unexpected publish on %d error: %v", i, err)
+			}
+			time.Sleep(2 * time.Millisecond)
+		}
+		doneCh <- true
+	}()
+
+	time.Sleep(20 * time.Millisecond)
+	sl.Shutdown()
+
+	<-doneCh
+	sc.clusterForName("C3").waitOnStreamLeader("$G", "M2")
+
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("M2")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 150 {
+			return fmt.Errorf("Expected 150 msgs, got state: %+v", si.State)
+		}
+		if si.State.FirstSeq != 101 {
+			return fmt.Errorf("Expected start seq of 101, got state: %+v", si.State)
+		}
+		return nil
+	})
+}
+
+func TestJetStreamClusterSources(t *testing.T) {
+	sc := createJetStreamSuperCluster(t, 3, 3)
+	defer sc.shutdown()
+
+	// Client based API
+	s := sc.clusterForName("C1").randomServer()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	// Create our source streams.
+	for _, sname := range []string{"foo", "bar", "baz"} {
+		if _, err := js.AddStream(&nats.StreamConfig{Name: sname, Replicas: 1}); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+	}
+
+	sendBatch := func(subject string, n int) {
+		for i := 0; i < n; i++ {
+			msg := fmt.Sprintf("MSG-%d", i+1)
+			if _, err := js.Publish(subject, []byte(msg)); err != nil {
+				t.Fatalf("Unexpected publish error: %v", err)
+			}
+		}
+	}
+	// Populate each one.
+	sendBatch("foo", 10)
+	sendBatch("bar", 15)
+	sendBatch("baz", 25)
+
+	// Needed while Go client does not have mirror support for creating mirror or source streams.
+	createStream := func(cfg *StreamConfig) {
+		t.Helper()
+		req, err := json.Marshal(cfg)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		rm, err := nc.Request(fmt.Sprintf(JSApiStreamCreateT, cfg.Name), req, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		var resp JSApiStreamCreateResponse
+		if err := json.Unmarshal(rm.Data, &resp); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if resp.Error != nil {
+			t.Fatalf("Unexpected error: %+v", resp.Error)
+		}
+	}
+
+	cfg := &StreamConfig{
+		Name:    "MS",
+		Storage: FileStorage,
+		Sources: []*StreamSource{
+			&StreamSource{Name: "foo"},
+			&StreamSource{Name: "bar"},
+			&StreamSource{Name: "baz"},
+		},
+	}
+
+	createStream(cfg)
+	time.Sleep(time.Second)
+
+	// Faster timeout since we loop below checking for condition.
+	js2, err := nc.JetStream(nats.MaxWait(50 * time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("MS")
+		if err != nil {
+			return err
+		}
+		if si.State.Msgs != 50 {
+			return fmt.Errorf("Expected 50 msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
+
+	// Purge the source streams.
+	for _, sname := range []string{"foo", "bar", "baz"} {
+		if err := js.PurgeStream(sname); err != nil {
+			t.Fatalf("Unexpected purge error: %v", err)
+		}
+	}
+
+	if err := js.DeleteStream("MS"); err != nil {
+		t.Fatalf("Unexpected delete error: %v", err)
+	}
+
+	// Send more msgs now.
+	sendBatch("foo", 10)
+	sendBatch("bar", 15)
+	sendBatch("baz", 25)
+
+	cfg = &StreamConfig{
+		Name:    "MS2",
+		Storage: FileStorage,
+		Sources: []*StreamSource{
+			&StreamSource{Name: "foo"},
+			&StreamSource{Name: "bar"},
+			&StreamSource{Name: "baz"},
+		},
+		Replicas:  3,
+		Placement: &Placement{Cluster: "C3"},
+	}
+
+	createStream(cfg)
+
+	checkFor(t, 5*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("MS2")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 50 {
+			return fmt.Errorf("Expected 50 msgs, got state: %+v", si.State)
+		}
+		if si.State.FirstSeq != 1 {
+			return fmt.Errorf("Expected start seq of 1, got state: %+v", si.State)
+		}
+		return nil
+	})
+
+	sl := sc.clusterForName("C3").streamLeader("$G", "MS2")
+	doneCh := make(chan bool)
+
+	if sl == sc.leader() {
+		nc.Request(JSApiLeaderStepDown, nil, time.Second)
+		sc.waitOnLeader()
+	}
+
+	// Now test that if the mirror get's interrupted that it picks up where it left off etc.
+	go func() {
+		// Send 50 more messages each.
+		for i := 0; i < 50; i++ {
+			msg := fmt.Sprintf("R-MSG-%d", i+1)
+			for _, sname := range []string{"foo", "bar", "baz"} {
+				if _, err := js.Publish(sname, []byte(msg)); err != nil {
+					t.Errorf("Unexpected publish error: %v", err)
+				}
+			}
+			time.Sleep(2 * time.Millisecond)
+		}
+		doneCh <- true
+	}()
+
+	time.Sleep(20 * time.Millisecond)
+	sl.Shutdown()
+
+	sc.clusterForName("C3").waitOnStreamLeader("$G", "MS2")
+	<-doneCh
+
+	checkFor(t, 5*time.Second, 100*time.Millisecond, func() error {
+		si, err := js2.StreamInfo("MS2")
+		if err != nil {
+			return err
+		}
+		if si.State.Msgs != 200 {
+			return fmt.Errorf("Expected 200 msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
+}