Repository: nats-io/nats-server
Pull request #1906 (2021-02-12 03:27:09)

Snapshot and Compaction Improvements
Previously we did a snapshot and compaction strategy that would require pauses to the proposals from clients which was not ideal. This changes that to a raft node maintained external snapshots that do not require pauses. We also moved to memory based WALs for streams and consumers.

Various other bug fixes, data race fixes and stability improvements.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/.gitignore b/.gitignore
index f5876149..781a1495 100644
--- a/.gitignore
+++ b/.gitignore
@@ -42,6 +42,7 @@ _testmain.go
 # bin
 nats-server
 gnatsd
+/util/mkpasswd
 
 # coverage
 coverage.out
diff --git a/server/auth.go b/server/auth.go
index 8c520051..8879f447 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -500,13 +500,15 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) boo
 			// but we set it here to be able to identify it in the logs.
 			c.opts.Username = user.Username
 		} else {
-			if c.kind == CLIENT && c.opts.Username == "" && noAuthUser != "" {
+			if c.kind == CLIENT && c.opts.Username == _EMPTY_ && noAuthUser != _EMPTY_ {
 				if u, exists := s.users[noAuthUser]; exists {
+					c.mu.Lock()
 					c.opts.Username = u.Username
 					c.opts.Password = u.Password
+					c.mu.Unlock()
 				}
 			}
-			if c.opts.Username != "" {
+			if c.opts.Username != _EMPTY_ {
 				user, ok = s.users[c.opts.Username]
 				if !ok || !c.connectionTypeAllowed(user.AllowedConnectionTypes) {
 					s.mu.Unlock()
diff --git a/server/client.go b/server/client.go
index 6a7b4f7a..c926eebf 100644
--- a/server/client.go
+++ b/server/client.go
@@ -4576,7 +4576,12 @@ func (c *client) getAccAndResultFromCache() (*Account, *SublistResult) {
 	// Check our cache.
 	if pac, ok = c.in.pacache[string(c.pa.pacache)]; ok {
 		// Check the genid to see if it's still valid.
-		if genid := atomic.LoadUint64(&pac.acc.sl.genid); genid != pac.genid {
+		// sl could be swapped out on reload so need to lock.
+		pac.acc.mu.RLock()
+		sl := pac.acc.sl
+		pac.acc.mu.RUnlock()
+
+		if genid := atomic.LoadUint64(&sl.genid); genid != pac.genid {
 			ok = false
 			delete(c.in.pacache, string(c.pa.pacache))
 		} else {
@@ -4591,11 +4596,16 @@ func (c *client) getAccAndResultFromCache() (*Account, *SublistResult) {
 			return nil, nil
 		}
 
+		// sl could be swapped out on reload so need to lock.
+		acc.mu.RLock()
+		sl := acc.sl
+		acc.mu.RUnlock()
+
 		// Match against the account sublist.
-		r = acc.sl.Match(string(c.pa.subject))
+		r = sl.Match(string(c.pa.subject))
 
 		// Store in our cache
-		c.in.pacache[string(c.pa.pacache)] = &perAccountCache{acc, r, atomic.LoadUint64(&acc.sl.genid)}
+		c.in.pacache[string(c.pa.pacache)] = &perAccountCache{acc, r, atomic.LoadUint64(&sl.genid)}
 
 		// Check if we need to prune.
 		if len(c.in.pacache) > maxPerAccountCacheSize {
diff --git a/server/consumer.go b/server/consumer.go
index 87d88b93..8cc44e73 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -295,7 +295,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	}
 
 	// Make sure any partition subject is also a literal.
-	if config.FilterSubject != "" {
+	if config.FilterSubject != _EMPTY_ {
 		var checkSubject bool
 
 		mset.mu.RLock()
@@ -665,7 +665,7 @@ func (o *consumer) setLeader(isLeader bool) {
 			return
 		}
 		// Setup the internal sub for next message requests.
-		if !o.isPushMode() {
+		if o.isPullMode() {
 			if o.reqSub, err = o.subscribeInternal(o.nextMsgSubj, o.processNextMsgReq); err != nil {
 				o.mu.Unlock()
 				o.deleteWithoutAdvisory()
@@ -1062,6 +1062,19 @@ func (o *consumer) progressUpdate(seq uint64) {
 	o.mu.Unlock()
 }
 
+// Lock should be held.
+func (o *consumer) updateSkipped() {
+	// Clustered mode and R>1 only.
+	if o.node == nil || !o.isLeader() {
+		return
+	}
+	var b [1 + 8]byte
+	b[0] = byte(updateSkipOp)
+	var le = binary.LittleEndian
+	le.PutUint64(b[1:], o.sseq)
+	o.node.Propose(b[:])
+}
+
 // Lock should be held.
 func (o *consumer) updateDelivered(dseq, sseq, dc uint64, ts int64) {
 	// Clustered mode and R>1.
@@ -1472,10 +1485,13 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 	o.updateAcks(dseq, sseq)
 
 	mset := o.mset
+	clustered := o.node != nil
 	o.mu.Unlock()
 
 	// Let the owning stream know if we are interest or workqueue retention based.
-	if mset != nil && mset.cfg.Retention != LimitsPolicy {
+	// If this consumer is clustered this will be handled by processReplicatedAck
+	// after the ack has propagated.
+	if !clustered && mset != nil && mset.cfg.Retention != LimitsPolicy {
 		if sagap > 1 {
 			// FIXME(dlc) - This is very inefficient, will need to fix.
 			for seq := sseq; seq > sseq-sagap; seq-- {
@@ -1496,20 +1512,38 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 // This is called for interest based retention streams to remove messages.
 func (o *consumer) needAck(sseq uint64) bool {
 	var needAck bool
-	o.mu.Lock()
+	var asflr, osseq uint64
+	var pending map[uint64]*Pending
+	o.mu.RLock()
+	if o.isLeader() {
+		asflr, osseq = o.asflr, o.sseq
+		pending = o.pending
+	} else {
+		state, err := o.store.State()
+		if err != nil || state == nil {
+			o.mu.RUnlock()
+			return false
+		}
+		asflr, osseq = state.AckFloor.Stream, o.sseq
+		pending = state.Pending
+	}
 	switch o.cfg.AckPolicy {
 	case AckNone, AckAll:
-		needAck = sseq > o.asflr
+		needAck = sseq > asflr
 	case AckExplicit:
-		if sseq > o.asflr {
+		if sseq > asflr {
 			// Generally this means we need an ack, but just double check pending acks.
 			needAck = true
-			if len(o.pending) > 0 && sseq < o.sseq {
-				_, needAck = o.pending[sseq]
+			if sseq < osseq {
+				if len(pending) == 0 {
+					needAck = false
+				} else {
+					_, needAck = pending[sseq]
+				}
 			}
 		}
 	}
-	o.mu.Unlock()
+	o.mu.RUnlock()
 	return needAck
 }
 
@@ -1779,6 +1813,7 @@ func (o *consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dc ui
 			if dc == 1 { // First delivery.
 				o.sseq++
 				if o.cfg.FilterSubject != _EMPTY_ && !o.isFilteredMatch(subj) {
+					o.updateSkipped()
 					continue
 				}
 			}
@@ -1998,6 +2033,7 @@ func (o *consumer) deliverCurrentMsg(subj string, hdr, msg []byte, seq uint64, t
 	// If we are partitioned and we do not match, do not consider this a failure.
 	// Go ahead and return true.
 	if o.cfg.FilterSubject != _EMPTY_ && !o.isFilteredMatch(subj) {
+		o.updateSkipped()
 		o.mu.Unlock()
 		return true
 	}
@@ -2275,6 +2311,7 @@ func (o *consumer) selectSubjectLast() {
 		}
 		if o.isFilteredMatch(subj) {
 			o.sseq = seq
+			o.updateSkipped()
 			return
 		}
 	}
diff --git a/server/events.go b/server/events.go
index 74559a2e..9eba3c75 100644
--- a/server/events.go
+++ b/server/events.go
@@ -1155,8 +1155,6 @@ func (s *Server) statszReq(sub *subscription, _ *client, subject, reply string,
 	s.mu.Unlock()
 }
 
-var errSkipZreq = errors.New("filtered response")
-
 func (s *Server) zReq(reply string, msg []byte, fOpts *EventFilterOptions, optz interface{}, respf func() (interface{}, error)) {
 	if !s.EventsEnabled() || reply == _EMPTY_ {
 		return
@@ -1174,9 +1172,6 @@ func (s *Server) zReq(reply string, msg []byte, fOpts *EventFilterOptions, optz
 	}
 	if err == nil {
 		response["data"], err = respf()
-		if errors.Is(err, errSkipZreq) {
-			return
-		}
 		status = http.StatusInternalServerError
 	}
 	if err != nil {
diff --git a/server/filestore.go b/server/filestore.go
index df61cbe6..0ddf3a9e 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -249,23 +249,23 @@ const (
 	FileStoreMaxBlkSize = maxBlockSize
 )
 
-func newFileStore(fcfg FileStoreConfig, cfg StreamConfig) (*fileStore, bool, error) {
+func newFileStore(fcfg FileStoreConfig, cfg StreamConfig) (*fileStore, error) {
 	return newFileStoreWithCreated(fcfg, cfg, time.Now())
 }
 
-func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created time.Time) (*fileStore, bool, error) {
+func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created time.Time) (*fileStore, error) {
 	if cfg.Name == "" {
-		return nil, false, fmt.Errorf("name required")
+		return nil, fmt.Errorf("name required")
 	}
 	if cfg.Storage != FileStorage {
-		return nil, false, fmt.Errorf("fileStore requires file storage type in config")
+		return nil, fmt.Errorf("fileStore requires file storage type in config")
 	}
 	// Default values.
 	if fcfg.BlockSize == 0 {
 		fcfg.BlockSize = dynBlkSize(cfg.Retention, cfg.MaxBytes)
 	}
 	if fcfg.BlockSize > maxBlockSize {
-		return nil, false, fmt.Errorf("filestore max block size is %s", friendlyBytes(maxBlockSize))
+		return nil, fmt.Errorf("filestore max block size is %s", friendlyBytes(maxBlockSize))
 	}
 	if fcfg.CacheExpire == 0 {
 		fcfg.CacheExpire = defaultCacheBufferExpiration
@@ -274,21 +274,17 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		fcfg.SyncInterval = defaultSyncInterval
 	}
 
-	// Track if we created this vs restored.
-	var bootstrap bool
-
 	// Check the directory
 	if stat, err := os.Stat(fcfg.StoreDir); os.IsNotExist(err) {
-		bootstrap = true
 		if err := os.MkdirAll(fcfg.StoreDir, 0755); err != nil {
-			return nil, bootstrap, fmt.Errorf("could not create storage directory - %v", err)
+			return nil, fmt.Errorf("could not create storage directory - %v", err)
 		}
 	} else if stat == nil || !stat.IsDir() {
-		return nil, bootstrap, fmt.Errorf("store directory is not a directory")
+		return nil, fmt.Errorf("storage directory is not a directory")
 	}
 	tmpfile, err := ioutil.TempFile(fcfg.StoreDir, "_test_")
 	if err != nil {
-		return nil, bootstrap, fmt.Errorf("storage directory is not writable")
+		return nil, fmt.Errorf("storage directory is not writable")
 	}
 	os.Remove(tmpfile.Name())
 
@@ -307,35 +303,35 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	mdir := path.Join(fcfg.StoreDir, msgDir)
 	odir := path.Join(fcfg.StoreDir, consumerDir)
 	if err := os.MkdirAll(mdir, 0755); err != nil {
-		return nil, bootstrap, fmt.Errorf("could not create message storage directory - %v", err)
+		return nil, fmt.Errorf("could not create message storage directory - %v", err)
 	}
 	if err := os.MkdirAll(odir, 0755); err != nil {
-		return nil, bootstrap, fmt.Errorf("could not create message storage directory - %v", err)
+		return nil, fmt.Errorf("could not create message storage directory - %v", err)
 	}
 
 	// Create highway hash for message blocks. Use sha256 of directory as key.
 	key := sha256.Sum256([]byte(cfg.Name))
 	fs.hh, err = highwayhash.New64(key[:])
 	if err != nil {
-		return nil, bootstrap, fmt.Errorf("could not create hash: %v", err)
+		return nil, fmt.Errorf("could not create hash: %v", err)
 	}
 
 	// Recover our state.
 	if err := fs.recoverMsgs(); err != nil {
-		return nil, bootstrap, err
+		return nil, err
 	}
 
 	// Write our meta data iff does not exist.
 	meta := path.Join(fcfg.StoreDir, JetStreamMetaFile)
 	if _, err := os.Stat(meta); err != nil && os.IsNotExist(err) {
 		if err := fs.writeStreamMeta(); err != nil {
-			return nil, bootstrap, err
+			return nil, err
 		}
 	}
 
 	fs.syncTmr = time.AfterFunc(fs.fcfg.SyncInterval, fs.syncBlocks)
 
-	return fs, bootstrap, nil
+	return fs, nil
 }
 
 func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
@@ -593,6 +589,11 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 		seq := le.Uint64(hdr[4:])
 		ts := int64(le.Uint64(hdr[12:]))
 
+		// If the first seq we read does not match our indexed first seq, reset.
+		if index == 0 && seq > mb.first.seq {
+			mb.first.seq = seq
+		}
+
 		// This is an old erased message, or a new one that we can track.
 		if seq == 0 || seq&ebit != 0 || seq < mb.first.seq {
 			seq = seq &^ ebit
@@ -976,6 +977,19 @@ func (fs *fileStore) SkipMsg() uint64 {
 	return seq
 }
 
+// Lock should be held.
+func (fs *fileStore) rebuildFirst() {
+	if len(fs.blks) == 0 {
+		return
+	}
+	if fmb := fs.blks[0]; fmb != nil {
+		fmb.removeIndexFile()
+		fmb.rebuildState()
+		fmb.writeIndexInfo()
+		fs.selectNextFirst()
+	}
+}
+
 // Will check the msg limit and drop firstSeq msg if needed.
 // Lock should be held.
 func (fs *fileStore) enforceMsgLimit() {
@@ -983,7 +997,10 @@ func (fs *fileStore) enforceMsgLimit() {
 		return
 	}
 	for nmsgs := fs.state.Msgs; nmsgs > uint64(fs.cfg.MaxMsgs); nmsgs = fs.state.Msgs {
-		fs.deleteFirstMsgLocked()
+		if removed, err := fs.deleteFirstMsgLocked(); err != nil || !removed {
+			fs.rebuildFirst()
+			return
+		}
 	}
 }
 
@@ -994,7 +1011,10 @@ func (fs *fileStore) enforceBytesLimit() {
 		return
 	}
 	for bs := fs.state.Bytes; bs > uint64(fs.cfg.MaxBytes); bs = fs.state.Bytes {
-		fs.deleteFirstMsgLocked()
+		if removed, err := fs.deleteFirstMsgLocked(); err != nil || !removed {
+			fs.rebuildFirst()
+			return
+		}
 	}
 }
 
@@ -1059,7 +1079,7 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 	mb.mu.Lock()
 
 	// Check cache. This should be very rare.
-	if mb.cache == nil || mb.cache.idx == nil {
+	if mb.cache == nil || mb.cache.idx == nil || seq < mb.cache.fseq && mb.cache.off > 0 {
 		mb.mu.Unlock()
 		fs.mu.Unlock()
 		if err := mb.loadMsgs(); err != nil {
@@ -1144,12 +1164,16 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 
 	// Kick outside of lock.
 	if shouldWriteIndex {
-		if qch == nil && !fs.fip {
-			mb.spinUpFlushLoop()
-		}
-		select {
-		case fch <- struct{}{}:
-		default:
+		if !fs.fip {
+			if qch == nil {
+				mb.spinUpFlushLoop()
+			}
+			select {
+			case fch <- struct{}{}:
+			default:
+			}
+		} else {
+			mb.writeIndexInfo()
 		}
 	}
 
@@ -3473,9 +3497,6 @@ func encodeConsumerState(state *ConsumerState) []byte {
 
 func (o *consumerFileStore) Update(state *ConsumerState) error {
 	// Sanity checks.
-	if state.Delivered.Consumer < 1 || state.Delivered.Stream < 1 {
-		return fmt.Errorf("bad delivered sequences")
-	}
 	if state.AckFloor.Consumer > state.Delivered.Consumer {
 		return fmt.Errorf("bad ack floor for consumer")
 	}
diff --git a/server/jetstream.go b/server/jetstream.go
index 54a1ff5b..fc942096 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -444,18 +444,26 @@ func (s *Server) shutdownJetStream() {
 
 	js.mu.Lock()
 	js.accounts = nil
+	var n RaftNode
+
 	if cc := js.cluster; cc != nil {
 		js.stopUpdatesSub()
-		if cc.meta != nil {
-			cc.meta.Stop()
-			cc.meta = nil
-		}
 		if cc.c != nil {
 			cc.c.closeConnection(ClientClosed)
 			cc.c = nil
 		}
+		n = cc.meta
+		cc.meta = nil
 	}
 	js.mu.Unlock()
+
+	// If we still have our raft group, do a snapshot on exit.
+	if n != nil {
+		if snap := js.metaSnapshot(); len(snap) > 0 {
+			n.InstallSnapshot(snap)
+		}
+		n.Stop()
+	}
 }
 
 // JetStreamConfig will return the current config. Useful if the system
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 59df417d..ac63f3c0 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -1305,7 +1305,7 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, repl
 		return
 	}
 
-	name := streamNameFromSubject(subject)
+	streamName := streamNameFromSubject(subject)
 
 	var resp = JSApiStreamInfoResponse{ApiResponse: ApiResponse{Type: JSApiStreamInfoResponseType}}
 
@@ -1317,14 +1317,8 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, repl
 			return
 		}
 
-		if cc.meta != nil && cc.meta.GroupLeader() == _EMPTY_ {
-			resp.Error = jsClusterNotAvailErr
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
-
 		js.mu.RLock()
-		isLeader, sa := cc.isLeader(), js.streamAssignment(acc.Name, name)
+		isLeader, sa := cc.isLeader(), js.streamAssignment(acc.Name, streamName)
 		js.mu.RUnlock()
 
 		if isLeader && sa == nil {
@@ -1339,6 +1333,10 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, repl
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 			return
 		} else if sa == nil {
+			if js.isLeaderless() {
+				resp.Error = jsClusterNotAvailErr
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			}
 			return
 		}
 
@@ -1350,7 +1348,11 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, repl
 		}
 
 		// We have the stream assigned and a leader, so only the stream leader should answer.
-		if !acc.JetStreamIsStreamLeader(name) {
+		if !acc.JetStreamIsStreamLeader(streamName) {
+			if js.isLeaderless() {
+				resp.Error = jsClusterNotAvailErr
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			}
 			return
 		}
 	}
@@ -1366,7 +1368,7 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, repl
 		return
 	}
 
-	mset, err := acc.lookupStream(name)
+	mset, err := acc.lookupStream(streamName)
 	if err != nil {
 		resp.Error = jsNotFoundError(err)
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
@@ -2031,12 +2033,6 @@ func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, subject, rep
 			return
 		}
 
-		if cc.meta != nil && cc.meta.GroupLeader() == _EMPTY_ {
-			resp.Error = jsClusterNotAvailErr
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
-
 		js.mu.RLock()
 		isLeader, sa := cc.isLeader(), js.streamAssignment(acc.Name, stream)
 		js.mu.RUnlock()
@@ -2053,6 +2049,10 @@ func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, subject, rep
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 			return
 		} else if sa == nil {
+			if js.isLeaderless() {
+				resp.Error = jsClusterNotAvailErr
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			}
 			return
 		}
 
@@ -2065,6 +2065,10 @@ func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, subject, rep
 
 		// We have the stream assigned and a leader, so only the stream leader should answer.
 		if !acc.JetStreamIsStreamLeader(stream) {
+			if js.isLeaderless() {
+				resp.Error = jsClusterNotAvailErr
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			}
 			return
 		}
 	}
@@ -2841,8 +2845,8 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, subject, re
 		return
 	}
 
-	stream := streamNameFromSubject(subject)
-	consumer := consumerNameFromSubject(subject)
+	streamName := streamNameFromSubject(subject)
+	consumerName := consumerNameFromSubject(subject)
 
 	var resp = JSApiConsumerInfoResponse{ApiResponse: ApiResponse{Type: JSApiConsumerInfoResponseType}}
 
@@ -2854,14 +2858,8 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, subject, re
 			return
 		}
 
-		if cc.meta != nil && cc.meta.GroupLeader() == _EMPTY_ {
-			resp.Error = jsClusterNotAvailErr
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
-
 		js.mu.RLock()
-		isLeader, sa, ca := cc.isLeader(), js.streamAssignment(acc.Name, stream), js.consumerAssignment(acc.Name, stream, consumer)
+		isLeader, sa, ca := cc.isLeader(), js.streamAssignment(acc.Name, streamName), js.consumerAssignment(acc.Name, streamName, consumerName)
 		js.mu.RUnlock()
 
 		if isLeader && ca == nil {
@@ -2881,6 +2879,10 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, subject, re
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 			return
 		} else if ca == nil {
+			if js.isLeaderless() {
+				resp.Error = jsClusterNotAvailErr
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			}
 			return
 		}
 
@@ -2892,7 +2894,11 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, subject, re
 		}
 
 		// We have the consumer assigned and a leader, so only the consumer leader should answer.
-		if !acc.JetStreamIsConsumerLeader(stream, consumer) {
+		if !acc.JetStreamIsConsumerLeader(streamName, consumerName) {
+			if js.isLeaderless() {
+				resp.Error = jsClusterNotAvailErr
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			}
 			return
 		}
 	}
@@ -2908,14 +2914,14 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, subject, re
 		return
 	}
 
-	mset, err := acc.lookupStream(stream)
+	mset, err := acc.lookupStream(streamName)
 	if err != nil {
 		resp.Error = jsNotFoundError(err)
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
 
-	obs := mset.lookupConsumer(consumer)
+	obs := mset.lookupConsumer(consumerName)
 	if obs == nil {
 		resp.Error = jsNoConsumerErr
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 66475f9a..0f01dc4d 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -72,6 +72,8 @@ const (
 	updateAcksOp
 	// Compressed consumer assignments.
 	assignCompressedConsumerOp
+	// Filtered Consumer skip.
+	updateSkipOp
 )
 
 // raftGroups are controlled by the metagroup controller.
@@ -218,7 +220,8 @@ func (s *Server) JetStreamSnapshotMeta() error {
 	if !cc.isLeader() {
 		return errNotLeader
 	}
-	return cc.meta.Snapshot(js.metaSnapshot())
+
+	return cc.meta.InstallSnapshot(js.metaSnapshot())
 }
 
 func (s *Server) JetStreamStepdownStream(account, stream string) error {
@@ -274,11 +277,7 @@ func (s *Server) JetStreamSnapshotStream(account, stream string) error {
 	n := mset.node
 	mset.mu.RUnlock()
 
-	n.PausePropose()
-	err = n.Snapshot(mset.stateSnapshot())
-	n.ResumePropose()
-
-	return err
+	return n.InstallSnapshot(mset.stateSnapshot())
 }
 
 func (s *Server) JetStreamClusterPeers() []string {
@@ -460,9 +459,10 @@ func (js *jetStream) setupMetaGroup() error {
 
 	// Setup our WAL for the metagroup.
 	sysAcc := s.SystemAccount()
-	stateDir := path.Join(js.config.StoreDir, sysAcc.Name, defaultStoreDirName, defaultMetaGroupName)
-	fs, bootstrap, err := newFileStore(
-		FileStoreConfig{StoreDir: stateDir, BlockSize: defaultMetaFSBlkSize},
+	storeDir := path.Join(js.config.StoreDir, sysAcc.Name, defaultStoreDirName, defaultMetaGroupName)
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: defaultMetaFSBlkSize},
 		StreamConfig{Name: defaultMetaGroupName, Storage: FileStorage},
 	)
 	if err != nil {
@@ -470,11 +470,10 @@ func (js *jetStream) setupMetaGroup() error {
 		return err
 	}
 
-	cfg := &RaftConfig{Name: defaultMetaGroupName, Store: stateDir, Log: fs}
+	cfg := &RaftConfig{Name: defaultMetaGroupName, Store: storeDir, Log: fs}
 
-	if bootstrap {
+	if _, err := readPeerState(storeDir); err != nil {
 		s.Noticef("JetStream cluster bootstrapping")
-		// FIXME(dlc) - Make this real.
 		peers := s.ActivePeers()
 		s.Debugf("JetStream cluster initial peers: %+v", peers)
 		if err := s.bootstrapRaftNode(cfg, peers, false); err != nil {
@@ -483,6 +482,7 @@ func (js *jetStream) setupMetaGroup() error {
 	} else {
 		s.Noticef("JetStream cluster recovering state")
 	}
+
 	// Start up our meta node.
 	n, err := s.startRaftNode(cfg)
 	if err != nil {
@@ -523,6 +523,26 @@ func (js *jetStream) server() *Server {
 	return s
 }
 
+// Will respond if we do not think we have a metacontroller leader.
+func (js *jetStream) isLeaderless() bool {
+	js.mu.RLock()
+	defer js.mu.RUnlock()
+
+	cc := js.cluster
+	if cc == nil {
+		return false
+	}
+
+	// If we don't have a leader.
+	if cc.meta.GroupLeader() == _EMPTY_ {
+		// Make sure we have been running for enough time.
+		if time.Since(cc.meta.Created()) > lostQuorumInterval {
+			return true
+		}
+	}
+	return false
+}
+
 // Will respond iff we are a member and we know we have no leader.
 func (js *jetStream) isGroupLeaderless(rg *raftGroup) bool {
 	if rg == nil {
@@ -657,14 +677,16 @@ func (cc *jetStreamCluster) isConsumerLeader(account, stream, consumer string) b
 		return false
 	}
 	// Check if we are the leader of this raftGroup assigned to this consumer.
+	ca := sa.consumers[consumer]
+	if ca == nil {
+		return false
+	}
+	rg := ca.Group
 	ourID := cc.meta.ID()
-	for _, ca := range sa.consumers {
-		rg := ca.Group
-		for _, peer := range rg.Peers {
-			if peer == ourID {
-				if len(rg.Peers) == 1 || (rg.node != nil && rg.node.Leader()) {
-					return true
-				}
+	for _, peer := range rg.Peers {
+		if peer == ourID {
+			if len(rg.Peers) == 1 || (rg.node != nil && rg.node.Leader()) {
+				return true
 			}
 		}
 	}
@@ -672,12 +694,7 @@ func (cc *jetStreamCluster) isConsumerLeader(account, stream, consumer string) b
 }
 
 func (js *jetStream) monitorCluster() {
-	const (
-		compactInterval  = 5 * time.Minute
-		compactSizeLimit = 64 * 1024
-	)
-
-	s, cc, n := js.server(), js.cluster, js.getMetaGroup()
+	s, n := js.server(), js.getMetaGroup()
 	qch, lch, ach := n.QuitC(), n.LeadChangeC(), n.ApplyC()
 
 	defer s.grWG.Done()
@@ -685,25 +702,17 @@ func (js *jetStream) monitorCluster() {
 	s.Debugf("Starting metadata monitor")
 	defer s.Debugf("Exiting metadata monitor")
 
+	const compactInterval = 2 * time.Minute
 	t := time.NewTicker(compactInterval)
 	defer t.Stop()
 
-	isLeader := cc.isLeader()
-
+	var isLeader bool
 	var lastSnap []byte
-	var snapout bool
 
-	// Only to be called from leader.
-	attemptSnapshot := func() {
-		if snapout {
-			return
-		}
-		n.PausePropose()
-		defer n.ResumePropose()
+	doSnapshot := func() {
 		if snap := js.metaSnapshot(); !bytes.Equal(lastSnap, snap) {
-			if err := n.Snapshot(snap); err == nil {
+			if err := n.InstallSnapshot(snap); err == nil {
 				lastSnap = snap
-				snapout = true
 			}
 		}
 	}
@@ -726,28 +735,22 @@ func (js *jetStream) monitorCluster() {
 			// FIXME(dlc) - Deal with errors.
 			if hadSnapshot, didRemoval, err := js.applyMetaEntries(ce.Entries, isRecovering); err == nil {
 				n.Applied(ce.Index)
-				if hadSnapshot {
-					snapout = false
-					if !isRecovering {
-						n.Compact(ce.Index)
-					}
-				}
-				if didRemoval {
-					attemptSnapshot()
+				if (hadSnapshot && !isRecovering) || didRemoval {
+					// Since we received one make sure we have our own since we do not store
+					// our meta state outside of raft.
+					doSnapshot()
 				}
 			}
-			if isLeader && !snapout {
-				_, b := n.Size()
-				if b > compactSizeLimit {
-					attemptSnapshot()
+			// See if we could save some memory here.
+			if lastSnap != nil {
+				if _, b := n.Size(); b > uint64(len(lastSnap)*4) {
+					doSnapshot()
 				}
 			}
 		case isLeader = <-lch:
 			js.processLeaderChange(isLeader)
 		case <-t.C:
-			if isLeader && !snapout {
-				attemptSnapshot()
-			}
+			doSnapshot()
 		}
 	}
 }
@@ -795,6 +798,9 @@ func (js *jetStream) metaSnapshot() []byte {
 }
 
 func (js *jetStream) applyMetaSnapshot(buf []byte, isRecovering bool) error {
+	if len(buf) == 0 {
+		return nil
+	}
 	jse, err := s2.Decode(nil, buf)
 	if err != nil {
 		return err
@@ -803,6 +809,7 @@ func (js *jetStream) applyMetaSnapshot(buf []byte, isRecovering bool) error {
 	if err = json.Unmarshal(jse, &wsas); err != nil {
 		return err
 	}
+
 	// Build our new version here outside of js.
 	streams := make(map[string]map[string]*streamAssignment)
 	for _, wsa := range wsas {
@@ -1087,21 +1094,18 @@ func (js *jetStream) createRaftGroup(rg *raftGroup) error {
 		return errors.New("shutting down")
 	}
 
-	stateDir := path.Join(js.config.StoreDir, sysAcc.Name, defaultStoreDirName, rg.Name)
-	fs, bootstrap, err := newFileStore(
-		FileStoreConfig{StoreDir: stateDir, BlockSize: 32 * 1024 * 1024},
-		StreamConfig{Name: rg.Name, Storage: FileStorage, Replicas: 2},
-	)
+	storeDir := path.Join(js.config.StoreDir, sysAcc.Name, defaultStoreDirName, rg.Name)
+	ms, err := newMemStore(&StreamConfig{Name: rg.Name, Storage: MemoryStorage})
 	if err != nil {
-		s.Errorf("Error creating filestore: %v", err)
+		s.Errorf("Error creating memstore: %v", err)
 		return err
 	}
+	cfg := &RaftConfig{Name: rg.Name, Store: storeDir, Log: ms}
 
-	cfg := &RaftConfig{Name: rg.Name, Store: stateDir, Log: fs}
-
-	if bootstrap {
+	if _, err := readPeerState(storeDir); err != nil {
 		s.bootstrapRaftNode(cfg, rg.Peers, true)
 	}
+
 	n, err := s.startRaftNode(cfg)
 	if err != nil {
 		s.Debugf("Error creating raft group: %v", err)
@@ -1150,15 +1154,15 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 
 	qch, lch, ach := n.QuitC(), n.LeadChangeC(), n.ApplyC()
 
-	const (
-		compactInterval  = 10 * time.Minute
-		compactSizeLimit = 64 * 1024 * 1024
-		compactMinWait   = 5 * time.Second
-	)
-
 	s.Debugf("Starting stream monitor for '%s > %s'", sa.Client.Account, sa.Config.Name)
 	defer s.Debugf("Exiting stream monitor for '%s > %s'", sa.Client.Account, sa.Config.Name)
 
+	const (
+		compactInterval  = 2 * time.Minute
+		compactSizeLimit = 32 * 1024 * 1024
+		compactNumLimit  = 4096
+	)
+
 	t := time.NewTicker(compactInterval)
 	defer t.Stop()
 
@@ -1173,30 +1177,16 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 		return
 	}
 
-	var (
-		lastSnap   []byte
-		snapout    bool
-		lastFailed time.Time
-	)
+	var lastSnap []byte
 
-	// Only to be called from leader.
-	attemptSnapshot := func() {
-		if mset == nil || isRestore || snapout {
+	// Should only to be called from leader.
+	doSnapshot := func() {
+		if mset == nil || isRestore {
 			return
 		}
-		n.PausePropose()
-		defer n.ResumePropose()
 		if snap := mset.stateSnapshot(); !bytes.Equal(lastSnap, snap) {
-			if !lastFailed.IsZero() && time.Since(lastFailed) <= compactMinWait {
-				s.Debugf("Stream compaction delayed")
-				return
-			}
-			if err := n.Snapshot(snap); err != nil {
-				lastFailed = time.Now()
-			} else {
+			if err := n.InstallSnapshot(snap); err == nil {
 				lastSnap = snap
-				snapout = true
-				lastFailed = time.Time{}
 			}
 		}
 	}
@@ -1204,11 +1194,52 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 	// We will establish a restoreDoneCh no matter what. Will never be triggered unless
 	// we replace with the restore chan.
 	restoreDoneCh := make(<-chan error)
-
 	isRecovering := true
 
 	for {
 		select {
+		case <-s.quitCh:
+			return
+		case <-qch:
+			return
+		case ce := <-ach:
+			// No special processing needed for when we are caught up on restart.
+			if ce == nil {
+				isRecovering = false
+				continue
+			}
+			// Apply our entries.
+			if err := js.applyStreamEntries(mset, ce, isRecovering); err == nil {
+				n.Applied(ce.Index)
+				if !isLeader {
+					// If over 32MB go ahead and compact if not the leader.
+					if m, b := n.Size(); m > compactNumLimit || b > compactSizeLimit {
+						n.Compact(ce.Index)
+					}
+				}
+			} else {
+				s.Warnf("Error applying entries to '%s > %s'", sa.Client.Account, sa.Config.Name)
+			}
+			if isLeader && lastSnap != nil {
+				// If over 32MB go ahead and compact if not the leader.
+				if m, b := n.Size(); m > compactNumLimit || b > compactSizeLimit {
+					doSnapshot()
+				}
+			}
+		case isLeader = <-lch:
+			if isLeader && isRestore {
+				acc, _ := s.LookupAccount(sa.Client.Account)
+				restoreDoneCh = s.processStreamRestore(sa.Client, acc, sa.Config.Name, _EMPTY_, sa.Reply, _EMPTY_)
+			} else {
+				if !isLeader && n.GroupLeader() != noLeader {
+					js.setStreamAssignmentResponded(sa)
+				}
+				js.processStreamLeaderChange(mset, isLeader)
+			}
+		case <-t.C:
+			if isLeader {
+				doSnapshot()
+			}
 		case err := <-restoreDoneCh:
 			// We have completed a restore from snapshot on this server. The stream assignment has
 			// already been assigned but the replicas will need to catch up out of band. Consumers
@@ -1252,8 +1283,11 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 			if !isLeader {
 				panic("Finished restore but not leader")
 			}
+			// Trigger the stream followers to catchup.
+			if n := mset.raftNode(); n != nil {
+				n.SendSnapshot(mset.stateSnapshot())
+			}
 			js.processStreamLeaderChange(mset, isLeader)
-			attemptSnapshot()
 
 			// Check to see if we have restored consumers here.
 			// These are not currently assigned so we will need to do so here.
@@ -1288,7 +1322,11 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 							js.mu.RUnlock()
 							if ca == nil {
 								s.Warnf("Consumer assignment has not been assigned, retrying")
-								meta.ForwardProposal(addEntry)
+								if meta != nil {
+									meta.ForwardProposal(addEntry)
+								} else {
+									return
+								}
 							} else {
 								return
 							}
@@ -1296,67 +1334,20 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 					}()
 				}
 			}
-		case <-s.quitCh:
-			return
-		case <-qch:
-			return
-		case ce := <-ach:
-			// No special processing needed for when we are caught up on restart.
-			if ce == nil {
-				isRecovering = false
-				continue
-			}
-			if mset == nil && isRestore {
-				isRestore = false
-				sa.Restore = nil
-				if mset, err = acc.addStreamWithAssignment(sa.Config, nil, sa); err != nil {
-					s.Warnf("Could not add stream after restore '%s > %s': %v", sa.Client.Account, sa.Config.Name, err)
-					return
-				}
-			}
-			// Apply our entries.
-			if hadSnapshot, didRemove, err := js.applyStreamEntries(mset, ce, isRecovering); err == nil {
-				n.Applied(ce.Index)
-				if hadSnapshot {
-					snapout = false
-				}
-				if isLeader && didRemove {
-					attemptSnapshot()
-				}
-			} else {
-				s.Warnf("Error applying entries to '%s > %s'", sa.Client.Account, sa.Config.Name)
-			}
-			if isLeader && !snapout {
-				if _, b := n.Size(); b > compactSizeLimit {
-					attemptSnapshot()
-				}
-			}
-		case isLeader = <-lch:
-			if isLeader && isRestore {
-				acc, _ := s.LookupAccount(sa.Client.Account)
-				restoreDoneCh = s.processStreamRestore(sa.Client, acc, sa.Config.Name, _EMPTY_, sa.Reply, _EMPTY_)
-			} else {
-				if !isLeader && n.GroupLeader() != noLeader {
-					js.setStreamAssignmentResponded(sa)
-				}
-				js.processStreamLeaderChange(mset, isLeader)
-			}
-		case <-t.C:
-			if isLeader {
-				attemptSnapshot()
-			}
 		}
 	}
 }
 
-func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isRecovering bool) (bool, bool, error) {
-	var didSnap, didRemove bool
+func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isRecovering bool) error {
 	for _, e := range ce.Entries {
 		if e.Type == EntrySnapshot {
 			if !isRecovering {
-				mset.processSnapshot(e.Data)
+				var snap streamSnapshot
+				if err := json.Unmarshal(e.Data, &snap); err != nil {
+					return err
+				}
+				mset.processSnapshot(&snap)
 			}
-			didSnap = true
 		} else if e.Type == EntryRemovePeer {
 			js.mu.RLock()
 			ourID := js.cluster.meta.ID()
@@ -1364,7 +1355,7 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 			if peer := string(e.Data); peer == ourID {
 				mset.stop(true, false)
 			}
-			return false, false, nil
+			return nil
 		} else {
 			buf := e.Data
 			switch entryOp(buf[0]) {
@@ -1387,7 +1378,7 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 						s := js.srv
 						s.Errorf("JetStream out of space, will be DISABLED")
 						js.srv.DisableJetStream()
-						return didSnap, didRemove, err
+						return err
 					}
 				}
 			case deleteMsgOp:
@@ -1406,7 +1397,6 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 				if err != nil && !isRecovering {
 					s.Debugf("JetStream cluster failed to delete msg %d from stream %q for account %q: %v", md.Seq, md.Stream, md.Client.Account, err)
 				}
-				didRemove = removed
 
 				js.mu.RLock()
 				isLeader := cc.isStreamLeader(md.Client.Account, md.Stream)
@@ -1444,7 +1434,6 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 				if err != nil {
 					s.Warnf("JetStream cluster failed to purge stream %q for account %q: %v", sp.Stream, sp.Client.Account, err)
 				}
-				didRemove = purged > 0
 
 				js.mu.RLock()
 				isLeader := js.cluster.isStreamLeader(sp.Client.Account, sp.Stream)
@@ -1466,7 +1455,7 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 			}
 		}
 	}
-	return didSnap, didRemove, nil
+	return nil
 }
 
 // Returns the PeerInfo for all replicas of a raft node. This is different than node.Peers()
@@ -1501,6 +1490,9 @@ func (js *jetStream) checkPeers(rg *raftGroup) {
 }
 
 func (js *jetStream) processStreamLeaderChange(mset *stream, isLeader bool) {
+	if mset == nil {
+		return
+	}
 	sa := mset.streamAssignment()
 	if sa == nil {
 		return
@@ -1720,6 +1712,8 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 	if sa.Restore != nil {
 		if len(rg.Peers) == 1 || rg.node != nil && rg.node.ID() == rg.Preferred {
 			shouldCreate = false
+		} else {
+			sa.Restore = nil
 		}
 	}
 
@@ -1843,7 +1837,11 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 									js.mu.RUnlock()
 									if ca == nil {
 										s.Warnf("Consumer assignment has not been assigned, retrying")
-										meta.ForwardProposal(addEntry)
+										if meta != nil {
+											meta.ForwardProposal(addEntry)
+										} else {
+											return
+										}
 									} else {
 										return
 									}
@@ -1851,7 +1849,6 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 							}()
 						}
 					}
-
 				case <-s.quitCh:
 					return
 				}
@@ -1866,7 +1863,7 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
 func (js *jetStream) processStreamRemoval(sa *streamAssignment) {
 	js.mu.Lock()
 	s, cc := js.srv, js.cluster
-	if s == nil || cc == nil {
+	if s == nil || cc == nil || cc.meta == nil {
 		// TODO(dlc) - debug at least
 		js.mu.Unlock()
 		return
@@ -2101,7 +2098,7 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment) {
 	}
 
 	if err != nil {
-		js.srv.Debugf("Consumer create failed for '%s > %s > %s': %v\n", ca.Client.Account, ca.Stream, ca.Name, err)
+		s.Debugf("Consumer create failed for '%s > %s > %s': %v\n", ca.Client.Account, ca.Stream, ca.Name, err)
 		ca.err = err
 		if rg.node != nil {
 			rg.node.Delete()
@@ -2251,7 +2248,7 @@ func (o *consumer) raftNode() RaftNode {
 }
 
 func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
-	s, n := js.server(), o.raftNode()
+	s, cc, n := js.server(), js.cluster, o.raftNode()
 	defer s.grWG.Done()
 
 	if n == nil {
@@ -2261,19 +2258,34 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 
 	qch, lch, ach := n.QuitC(), n.LeadChangeC(), n.ApplyC()
 
-	const (
-		compactInterval  = 1 * time.Minute
-		compactSizeLimit = 8 * 1024 * 1024
-	)
-
 	s.Debugf("Starting consumer monitor for '%s > %s > %s", o.acc.Name, ca.Stream, ca.Name)
 	defer s.Debugf("Exiting consumer monitor for '%s > %s > %s'", o.acc.Name, ca.Stream, ca.Name)
 
+	js.mu.RLock()
+	isLeader := cc.isConsumerLeader(ca.Client.Account, ca.Stream, ca.Name)
+	js.mu.RUnlock()
+
+	const (
+		compactInterval  = 2 * time.Minute
+		compactSizeLimit = 4 * 1024 * 1024
+		compactNumLimit  = 4096
+	)
+
 	t := time.NewTicker(compactInterval)
 	defer t.Stop()
 
-	// Our last applied.
-	last := uint64(0)
+	var lastSnap []byte
+
+	// Should only to be called from leader.
+	doSnapshot := func() {
+		if state, err := o.store.State(); err == nil && state != nil {
+			if snap := encodeConsumerState(state); !bytes.Equal(lastSnap, snap) {
+				if err := n.InstallSnapshot(snap); err == nil {
+					lastSnap = snap
+				}
+			}
+		}
+	}
 
 	for {
 		select {
@@ -2286,11 +2298,18 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 			if ce == nil {
 				continue
 			}
-			if _, err := js.applyConsumerEntries(o, ce); err == nil {
+			if err := js.applyConsumerEntries(o, ce); err == nil {
 				n.Applied(ce.Index)
-				last = ce.Index
-				if _, b := n.Size(); b > compactSizeLimit {
-					n.Compact(last)
+				// If over 4MB go ahead and compact if not the leader.
+				if !isLeader {
+					if m, b := n.Size(); m > compactNumLimit || b > compactSizeLimit {
+						n.Compact(ce.Index)
+					}
+				}
+			}
+			if isLeader && lastSnap != nil {
+				if _, b := n.Size(); b > uint64(len(lastSnap)*4) {
+					doSnapshot()
 				}
 			}
 		case isLeader := <-lch:
@@ -2299,16 +2318,14 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 			}
 			js.processConsumerLeaderChange(o, isLeader)
 		case <-t.C:
-			// TODO(dlc) - We should have this delayed a bit to not race the invariants.
-			if last != 0 {
-				n.Compact(last)
+			if isLeader {
+				doSnapshot()
 			}
 		}
 	}
 }
 
-func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry) (bool, error) {
-	var didSnap bool
+func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry) error {
 	for _, e := range ce.Entries {
 		if e.Type == EntrySnapshot {
 			// No-op needed?
@@ -2317,7 +2334,6 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry) (bool
 				panic(err.Error())
 			}
 			o.store.Update(state)
-			didSnap = true
 		} else if e.Type == EntryRemovePeer {
 			js.mu.RLock()
 			ourID := js.cluster.meta.ID()
@@ -2325,7 +2341,7 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry) (bool
 			if peer := string(e.Data); peer == ourID {
 				o.stopWithFlags(true, false, false)
 			}
-			return false, nil
+			return nil
 		} else {
 			buf := e.Data
 			switch entryOp(buf[0]) {
@@ -2342,13 +2358,57 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry) (bool
 				if err != nil {
 					panic(err.Error())
 				}
-				o.store.UpdateAcks(dseq, sseq)
+				o.processReplicatedAck(dseq, sseq)
+			case updateSkipOp:
+				o.mu.Lock()
+				if !o.isLeader() {
+					var le = binary.LittleEndian
+					o.sseq = le.Uint64(buf[1:])
+				}
+				o.mu.Unlock()
 			default:
 				panic(fmt.Sprintf("JetStream Cluster Unknown group entry op type! %v", entryOp(buf[0])))
 			}
 		}
 	}
-	return didSnap, nil
+	return nil
+}
+
+func (o *consumer) processReplicatedAck(dseq, sseq uint64) {
+	o.store.UpdateAcks(dseq, sseq)
+
+	o.mu.RLock()
+
+	mset := o.mset
+	if mset == nil || mset.cfg.Retention != InterestPolicy {
+		o.mu.RUnlock()
+		return
+	}
+
+	var sagap uint64
+	if o.cfg.AckPolicy == AckAll {
+		if o.isLeader() {
+			sagap = sseq - o.asflr
+		} else {
+			// We are a follower so only have the store state, so read that in.
+			state, err := o.store.State()
+			if err != nil {
+				o.mu.RUnlock()
+				return
+			}
+			sagap = sseq - state.AckFloor.Stream
+		}
+	}
+	o.mu.RUnlock()
+
+	if sagap > 1 {
+		// FIXME(dlc) - This is very inefficient, will need to fix.
+		for seq := sseq; seq > sseq-sagap; seq-- {
+			mset.ackMsg(o, seq)
+		}
+	} else {
+		mset.ackMsg(o, sseq)
+	}
 }
 
 var errBadAckUpdate = errors.New("jetstream cluster bad replicated ack update")
@@ -3598,19 +3658,13 @@ func (mset *stream) isCatchingUp() bool {
 }
 
 // Process a stream snapshot.
-func (mset *stream) processSnapshot(buf []byte) {
-	var snap streamSnapshot
-	if err := json.Unmarshal(buf, &snap); err != nil {
-		// Log error.
-		return
-	}
-
+func (mset *stream) processSnapshot(snap *streamSnapshot) {
 	// Update any deletes, etc.
-	mset.processSnapshotDeletes(&snap)
+	mset.processSnapshotDeletes(snap)
 
 	mset.mu.Lock()
 	state := mset.store.State()
-	sreq := mset.calculateSyncRequest(&state, &snap)
+	sreq := mset.calculateSyncRequest(&state, snap)
 	s, subject, n := mset.srv, mset.sa.Sync, mset.node
 	mset.mu.Unlock()
 
@@ -3654,7 +3708,8 @@ RETRY:
 	if sreq == nil {
 		mset.mu.Lock()
 		state := mset.store.State()
-		sreq = mset.calculateSyncRequest(&state, &snap)
+		sreq = mset.calculateSyncRequest(&state, snap)
+
 		mset.mu.Unlock()
 		if sreq == nil {
 			return
diff --git a/server/memstore.go b/server/memstore.go
index 53790dec..66005115 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -83,6 +83,10 @@ func (ms *memStore) UpdateConfig(cfg *StreamConfig) error {
 // Stores a raw message with expected sequence number and timestamp.
 // Lock should be held.
 func (ms *memStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts int64) error {
+	if ms.msgs == nil {
+		return ErrStoreClosed
+	}
+
 	// Check if we are discarding new messages when we reach the limit.
 	if ms.cfg.Discard == DiscardNew {
 		if ms.cfg.MaxMsgs > 0 && ms.state.Msgs >= uint64(ms.cfg.MaxMsgs) {
diff --git a/server/monitor.go b/server/monitor.go
index fb298254..755ecfe0 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -2357,7 +2357,7 @@ func (s *Server) Jsz(opts *JSzOptions) (*JSInfo, error) {
 		js, cc := s.getJetStreamCluster()
 		if js == nil {
 			// Ignore
-			return nil, fmt.Errorf("%w: no cluster", errSkipZreq)
+			return nil, ErrJetStreamNotEnabled
 		}
 		// So if we have JS but no clustering, we are the leader so allow.
 		if cc != nil {
@@ -2365,7 +2365,7 @@ func (s *Server) Jsz(opts *JSzOptions) (*JSInfo, error) {
 			isLeader := cc.isLeader()
 			js.mu.RUnlock()
 			if !isLeader {
-				return nil, fmt.Errorf("%w: not leader", errSkipZreq)
+				return nil, errNotLeader
 			}
 		}
 	}
@@ -2411,10 +2411,10 @@ func (s *Server) Jsz(opts *JSzOptions) (*JSInfo, error) {
 			filterIdx = i
 		}
 		jsi.StreamCnt += len(jsa.streams)
-		jsi.Memory += uint64(jsa.usage.mem)
-		jsi.Store += uint64(jsa.usage.store)
-		jsi.API.Total += jsa.usage.api
-		jsi.API.Errors += jsa.usage.err
+		jsi.Memory += uint64(jsa.memTotal)
+		jsi.Store += uint64(jsa.storeTotal)
+		jsi.API.Total += jsa.apiTotal
+		jsi.API.Errors += jsa.apiErrors
 		for _, stream := range jsa.streams {
 			streamState := stream.state()
 			jsi.MessageCnt += streamState.Msgs
diff --git a/server/raft.go b/server/raft.go
index ae523a51..6de5cfbb 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -14,25 +14,30 @@
 package server
 
 import (
+	"bytes"
+	"crypto/sha256"
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"hash"
 	"io/ioutil"
 	"math/rand"
 	"net"
 	"os"
 	"path"
+	"path/filepath"
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/minio/highwayhash"
 )
 
 type RaftNode interface {
 	Propose(entry []byte) error
-	PausePropose()
-	ResumePropose()
 	ForwardProposal(entry []byte) error
-	Snapshot(snap []byte) error
+	InstallSnapshot(snap []byte) error
+	SendSnapshot(snap []byte) error
 	Applied(index uint64)
 	Compact(index uint64) error
 	State() RaftState
@@ -110,30 +115,31 @@ func (state RaftState) String() string {
 
 type raft struct {
 	sync.RWMutex
-	created time.Time
-	group   string
-	sd      string
-	id      string
-	wal     WAL
-	state   RaftState
-	csz     int
-	qn      int
-	peers   map[string]*lps
-	acks    map[uint64]map[string]struct{}
-	elect   *time.Timer
-	active  time.Time
-	term    uint64
-	pterm   uint64
-	pindex  uint64
-	sindex  uint64
-	commit  uint64
-	applied uint64
-	leader  string
-	vote    string
-	hash    string
-	s       *Server
-	c       *client
-	dflag   bool
+	created  time.Time
+	group    string
+	sd       string
+	id       string
+	wal      WAL
+	state    RaftState
+	hh       hash.Hash64
+	snapfile string
+	csz      int
+	qn       int
+	peers    map[string]*lps
+	acks     map[uint64]map[string]struct{}
+	elect    *time.Timer
+	active   time.Time
+	term     uint64
+	pterm    uint64
+	pindex   uint64
+	commit   uint64
+	applied  uint64
+	leader   string
+	vote     string
+	hash     string
+	s        *Server
+	c        *client
+	dflag    bool
 
 	// Subjects for votes, updates, replays.
 	psubj  string
@@ -157,7 +163,6 @@ type raft struct {
 
 	// Channels
 	propc    chan *Entry
-	pausec   chan struct{}
 	applyc   chan *CommittedEntry
 	sendq    chan *pubMsg
 	quit     chan struct{}
@@ -202,18 +207,18 @@ type RaftConfig struct {
 
 var (
 	errProposalFailed  = errors.New("raft: proposal failed")
-	errProposalsPaused = errors.New("raft: proposals paused")
 	errNotLeader       = errors.New("raft: not leader")
 	errAlreadyLeader   = errors.New("raft: already leader")
-	errNotCurrent      = errors.New("raft: not current")
 	errNilCfg          = errors.New("raft: no config given")
 	errUnknownPeer     = errors.New("raft: unknown peer")
 	errCorruptPeers    = errors.New("raft: corrupt peer state")
 	errStepdownFailed  = errors.New("raft: stepdown failed")
-	errPeersNotCurrent = errors.New("raft: all peers are not current")
 	errFailedToApply   = errors.New("raft: could not place apply entry")
 	errEntryLoadFailed = errors.New("raft: could not load entry from WAL")
 	errNodeClosed      = errors.New("raft: node is closed")
+	errBadSnapName     = errors.New("raft: snapshot name could not be parsed")
+	errNoSnapAvailable = errors.New("raft: no snapshot available")
+	errSnapshotCorrupt = errors.New("raft: snapshot corrupt")
 )
 
 // This will bootstrap a raftNode by writing its config into the store directory.
@@ -262,6 +267,20 @@ func (s *Server) bootstrapRaftNode(cfg *RaftConfig, knownPeers []string, allPeer
 		}
 	}
 
+	// Check the store directory. If we have a memory based WAL we need to make sure the directory is setup.
+	if stat, err := os.Stat(cfg.Store); os.IsNotExist(err) {
+		if err := os.MkdirAll(cfg.Store, 0755); err != nil {
+			return fmt.Errorf("raft: could not create storage directory - %v", err)
+		}
+	} else if stat == nil || !stat.IsDir() {
+		return fmt.Errorf("raft: storage directory is not a directory")
+	}
+	tmpfile, err := ioutil.TempFile(cfg.Store, "_test_")
+	if err != nil {
+		return fmt.Errorf("raft: storage directory is not writable")
+	}
+	os.Remove(tmpfile.Name())
+
 	return writePeerState(cfg.Store, &peerState{knownPeers, expected})
 }
 
@@ -280,6 +299,10 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 	hash := s.sys.shash
 	s.mu.Unlock()
 
+	if err := os.MkdirAll(path.Join(cfg.Store, snapshotsDir), 0755); err != nil {
+		return nil, fmt.Errorf("could not create snapshots directory - %v", err)
+	}
+
 	ps, err := readPeerState(cfg.Store)
 	if err != nil {
 		return nil, err
@@ -318,11 +341,17 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 		n.dflag = true
 	}
 
+	key := sha256.Sum256([]byte(n.group))
+	n.hh, _ = highwayhash.New64(key[:])
+
 	if term, vote, err := n.readTermVote(); err != nil && term > 0 {
 		n.term = term
 		n.vote = vote
 	}
 
+	// See if we have any snapshots and if so load and process on startup.
+	n.setupLastSnapshot()
+
 	if state := n.wal.State(); state.Msgs > 0 {
 		// TODO(dlc) - Recover our state here.
 		if first, err := n.loadFirstEntry(); err == nil {
@@ -461,26 +490,6 @@ func (s *Server) transferRaftLeaders() bool {
 	return didTransfer
 }
 
-func (s *Server) shutdownRaftNodes() {
-	if s == nil {
-		return
-	}
-
-	var nodes []RaftNode
-	s.rnMu.RLock()
-	for _, n := range s.raftNodes {
-		nodes = append(nodes, n)
-	}
-	s.rnMu.RUnlock()
-
-	for _, node := range nodes {
-		if node.Leader() {
-			node.StepDown()
-		}
-		node.Stop()
-	}
-}
-
 // Formal API
 
 // Propose will propose a new entry to the group.
@@ -492,24 +501,13 @@ func (n *raft) Propose(data []byte) error {
 		n.debug("Proposal ignored, not leader")
 		return errNotLeader
 	}
-	propc, paused, quit := n.propc, n.pausec, n.quit
+	propc := n.propc
 	n.RUnlock()
 
-	if paused != nil {
-		n.debug("Proposals paused, will wait")
-		select {
-		case <-paused:
-		case <-quit:
-			return errProposalFailed
-		case <-time.After(422 * time.Millisecond):
-			return errProposalsPaused
-		}
-	}
-
 	select {
 	case propc <- &Entry{EntryNormal, data}:
 	default:
-		n.debug("Propose failed!")
+		n.warn("Propose failed to be placed on internal channel")
 		return errProposalFailed
 	}
 	return nil
@@ -531,27 +529,6 @@ func (n *raft) ForwardProposal(entry []byte) error {
 	return nil
 }
 
-// PausePropose will pause new proposals.
-func (n *raft) PausePropose() {
-	n.Lock()
-	if n.pausec == nil {
-		n.pausec = make(chan struct{})
-	}
-	n.Unlock()
-}
-
-// ResumePropose will resum new proposals.
-func (n *raft) ResumePropose() {
-	n.Lock()
-	paused := n.pausec
-	n.pausec = nil
-	n.Unlock()
-
-	if paused != nil {
-		close(paused)
-	}
-}
-
 // ProposeAddPeer is called to add a peer to the group.
 func (n *raft) ProposeAddPeer(peer string) error {
 	n.RLock()
@@ -621,68 +598,238 @@ func (n *raft) ResumeApply() {
 	n.hcommit = 0
 }
 
-// Compact will compact our WAL. If this node is a leader we will want
-// all our peers to be at least to the same index. Non-leaders just compact
-// directly. This is for when we know we have our state on stable storage.
-// E.g JS Consumers.
+// Compact will compact our WAL.
+// This is for when we know we have our state on stable storage.
+// E.g. snapshots.
 func (n *raft) Compact(index uint64) error {
 	n.Lock()
-	defer n.Unlock()
-	// If we are not the leader compact at will.
-	if n.state != Leader {
-		_, err := n.wal.Compact(index)
-		return err
-	}
-	// We are the leader so we need to make sure all peers are at least up to this index.
-	for peer, ps := range n.peers {
-		if peer != n.id && ps.li < index {
-			return errPeersNotCurrent
-		}
-	}
-	return nil
+	_, err := n.wal.Compact(index)
+	n.Unlock()
+	return err
 }
 
 // Applied is to be called when the FSM has applied the committed entries.
 func (n *raft) Applied(index uint64) {
 	n.Lock()
-	defer n.Unlock()
-
 	// Ignore if already applied.
-	if index <= n.applied {
-		return
+	if index > n.applied {
+		n.applied = index
 	}
+	n.Unlock()
+}
+
+// For capturing data needed by snapshot.
+type snapshot struct {
+	lastTerm  uint64
+	lastIndex uint64
+	peerstate []byte
+	data      []byte
+}
+
+const minSnapshotLen = 28
 
-	// FIXME(dlc) - Check spec on error conditions, storage
-	n.applied = index
-	if index == n.sindex {
-		n.debug("Found snapshot entry: compacting log to index %d", index)
-		n.wal.Compact(index)
+// Encodes a snapshot into a buffer for storage.
+// Lock should be held.
+func (n *raft) encodeSnapshot(snap *snapshot) []byte {
+	if snap == nil {
+		return nil
 	}
+	var le = binary.LittleEndian
+	buf := make([]byte, minSnapshotLen+len(snap.peerstate)+len(snap.data))
+	le.PutUint64(buf[0:], snap.lastTerm)
+	le.PutUint64(buf[8:], snap.lastIndex)
+	// Peer state
+	le.PutUint32(buf[16:], uint32(len(snap.peerstate)))
+	wi := 20
+	copy(buf[wi:], snap.peerstate)
+	wi += len(snap.peerstate)
+	// data itself.
+	copy(buf[wi:], snap.data)
+	wi += len(snap.data)
+
+	// Now do the hash for the end.
+	n.hh.Reset()
+	n.hh.Write(buf[:wi])
+	checksum := n.hh.Sum(nil)
+	copy(buf[wi:], checksum)
+	wi += len(checksum)
+	return buf[:wi]
 }
 
-// Snapshot is used to snapshot the fsm. This can only be called from a leader.
-// For now these are assumed to be small and will be placed into the log itself.
-// TODO(dlc) - For meta and consumers this is straightforward, and for streams sans the messages this is as well.
-func (n *raft) Snapshot(snap []byte) error {
+// SendSnapshot will send the latest snapshot as a normal AE.
+// Should only be used when the upper layers know this is most recent.
+// Used when restoring streams etc.
+func (n *raft) SendSnapshot(data []byte) error {
+	n.sendAppendEntry([]*Entry{&Entry{EntrySnapshot, data}})
+	return nil
+}
+
+// Used to install a snapshot for the given term and applied index. This will release
+// all of the log entries up to and including index. This should not be called with
+// entries that have been applied to the FSM but have not been applied to the raft state.
+func (n *raft) InstallSnapshot(data []byte) error {
 	n.Lock()
-	defer n.Unlock()
+	if n.state == Closed {
+		n.Unlock()
+		return errNodeClosed
+	}
 
-	n.debug("Snapshot called with %d bytes, applied is %d", len(snap), n.applied)
+	ae, err := n.loadEntry(n.applied)
+	if err != nil {
+		n.Unlock()
+		return err
+	}
 
-	if n.state != Leader {
-		return errNotLeader
+	snap := &snapshot{
+		lastTerm:  ae.term,
+		lastIndex: n.applied,
+		peerstate: encodePeerState(&peerState{n.peerNames(), n.csz}),
+		data:      data,
 	}
-	if !n.isCurrent() {
-		return errNotCurrent
+
+	snapDir := path.Join(n.sd, snapshotsDir)
+	sn := fmt.Sprintf(snapFileT, snap.lastTerm, snap.lastIndex)
+	sfile := path.Join(snapDir, sn)
+
+	if err := ioutil.WriteFile(sfile, n.encodeSnapshot(snap), 0644); err != nil {
+		n.Unlock()
+		return err
 	}
 
-	select {
-	case n.propc <- &Entry{EntrySnapshot, snap}:
-	default:
-		return errProposalFailed
+	// Remember our latest snapshot file.
+	n.snapfile = sfile
+	_, err = n.wal.Compact(snap.lastIndex)
+	n.Unlock()
+
+	psnaps, _ := ioutil.ReadDir(snapDir)
+	// Remove any old snapshots.
+	for _, fi := range psnaps {
+		pn := fi.Name()
+		if pn != sn {
+			os.Remove(path.Join(snapDir, pn))
+		}
 	}
 
-	return nil
+	return err
+}
+
+const (
+	snapshotsDir = "snapshots"
+	snapFileT    = "snap.%d.%d"
+)
+
+func termAndIndexFromSnapFile(sn string) (term, index uint64, err error) {
+	if sn == _EMPTY_ {
+		return 0, 0, errBadSnapName
+	}
+	fn := filepath.Base(sn)
+	if n, err := fmt.Sscanf(fn, snapFileT, &term, &index); err != nil || n != 2 {
+		return 0, 0, errBadSnapName
+	}
+	return term, index, nil
+}
+
+func (n *raft) setupLastSnapshot() {
+	snapDir := path.Join(n.sd, snapshotsDir)
+	psnaps, err := ioutil.ReadDir(snapDir)
+	if err != nil {
+		return
+	}
+
+	var lterm, lindex uint64
+	var latest string
+	for _, sf := range psnaps {
+		sfile := path.Join(snapDir, sf.Name())
+		var term, index uint64
+		term, index, err := termAndIndexFromSnapFile(sf.Name())
+		if err == nil {
+			if term > lterm {
+				lterm, lindex = term, index
+				latest = sfile
+			} else if term == lterm && index > lindex {
+				lindex = index
+				latest = sfile
+			}
+		} else {
+			// Clean this up, can't parse the name.
+			// TODO(dlc) - We could read in and check actual contents.
+			n.debug("Removing snapshot, can't parse name: %q", sf.Name())
+			os.Remove(sfile)
+		}
+	}
+
+	// Now cleanup any old entries
+	for _, sf := range psnaps {
+		sfile := path.Join(snapDir, sf.Name())
+		if sfile != latest {
+			n.debug("Removing old snapshot: %q", sfile)
+			os.Remove(sfile)
+		}
+	}
+
+	if latest == _EMPTY_ {
+		return
+	}
+
+	// Set latest snapshot we have.
+	n.Lock()
+	n.snapfile = latest
+	snap, err := n.loadLastSnapshot()
+	if err != nil {
+		os.Remove(n.snapfile)
+		n.snapfile = _EMPTY_
+	} else {
+		n.pindex = snap.lastIndex
+		n.pterm = snap.lastTerm
+		n.commit = snap.lastIndex
+		n.applyc <- &CommittedEntry{n.commit, []*Entry{&Entry{EntrySnapshot, snap.data}}}
+		n.wal.Compact(snap.lastIndex)
+	}
+	n.Unlock()
+}
+
+// loadLastSnapshot will load and return our last snapshot.
+// Lock should be held.
+func (n *raft) loadLastSnapshot() (*snapshot, error) {
+	if n.snapfile == _EMPTY_ {
+		return nil, errNoSnapAvailable
+	}
+	buf, err := ioutil.ReadFile(n.snapfile)
+	if err != nil {
+		n.warn("Error reading snapshot: %v", err)
+		os.Remove(n.snapfile)
+		n.snapfile = _EMPTY_
+		return nil, err
+	}
+	if len(buf) < minSnapshotLen {
+		n.warn("Snapshot corrupt, too short")
+		os.Remove(n.snapfile)
+		n.snapfile = _EMPTY_
+		return nil, errSnapshotCorrupt
+	}
+
+	// Check to make sure hash is consistent.
+	hoff := len(buf) - 8
+	lchk := buf[hoff:]
+	n.hh.Reset()
+	n.hh.Write(buf[:hoff])
+	if !bytes.Equal(lchk[:], n.hh.Sum(nil)) {
+		n.warn("Snapshot corrupt, checksums did not match")
+		os.Remove(n.snapfile)
+		n.snapfile = _EMPTY_
+		return nil, errSnapshotCorrupt
+	}
+
+	var le = binary.LittleEndian
+	lps := le.Uint32(buf[16:])
+	snap := &snapshot{
+		lastTerm:  le.Uint64(buf[0:]),
+		lastIndex: le.Uint64(buf[8:]),
+		peerstate: buf[20 : 20+lps],
+		data:      buf[20+lps : hoff],
+	}
+
+	return snap, nil
 }
 
 // Leader returns if we are the leader for our group.
@@ -860,6 +1007,10 @@ func (n *raft) Peers() []*Peer {
 	return peers
 }
 
+func (n *raft) ApplyC() <-chan *CommittedEntry { return n.applyc }
+func (n *raft) LeadChangeC() <-chan bool       { return n.leadc }
+func (n *raft) QuitC() <-chan struct{}         { return n.quit }
+
 func (n *raft) Created() time.Time {
 	n.RLock()
 	defer n.RUnlock()
@@ -874,10 +1025,6 @@ func (n *raft) Delete() {
 	n.shutdown(true)
 }
 
-func (n *raft) ApplyC() <-chan *CommittedEntry { return n.applyc }
-func (n *raft) LeadChangeC() <-chan bool       { return n.leadc }
-func (n *raft) QuitC() <-chan struct{}         { return n.quit }
-
 func (n *raft) shutdown(shouldDelete bool) {
 	n.Lock()
 	if n.state == Closed {
@@ -889,12 +1036,12 @@ func (n *raft) shutdown(shouldDelete bool) {
 	n.state = Closed
 	s, g, wal := n.s, n.group, n.wal
 
-	// Delete our peer state and vote state.
+	// Delete our peer state and vote state and any snapshots.
 	if shouldDelete {
 		os.Remove(path.Join(n.sd, peerStateFile))
 		os.Remove(path.Join(n.sd, termVoteFile))
+		os.RemoveAll(path.Join(n.sd, snapshotsDir))
 	}
-
 	n.Unlock()
 
 	s.unregisterRaftNode(g)
@@ -1076,19 +1223,20 @@ type EntryType uint8
 
 const (
 	EntryNormal EntryType = iota
-	EntrySnapshot
+	EntryOldSnapshot
 	EntryPeerState
 	EntryAddPeer
 	EntryRemovePeer
 	EntryLeaderTransfer
+	EntrySnapshot
 )
 
 func (t EntryType) String() string {
 	switch t {
 	case EntryNormal:
 		return "Normal"
-	case EntrySnapshot:
-		return "Snapshot"
+	case EntryOldSnapshot:
+		return "OldSnapshot"
 	case EntryPeerState:
 		return "PeerState"
 	case EntryAddPeer:
@@ -1097,6 +1245,8 @@ func (t EntryType) String() string {
 		return "RemovePeer"
 	case EntryLeaderTransfer:
 		return "LeaderTransfer"
+	case EntrySnapshot:
+		return "Snapshot"
 	}
 	return fmt.Sprintf("Unknown [%d]", uint8(t))
 }
@@ -1479,6 +1629,19 @@ func (n *raft) runCatchup(peer, subj string, indexUpdatesC <-chan uint64) {
 	}
 }
 
+// Lock should be held.
+func (n *raft) sendSnapshotToFollower(subject string) (uint64, error) {
+	snap, err := n.loadLastSnapshot()
+	if err != nil {
+		return 0, err
+	}
+	// Go ahead and send the snapshot and peerstate here as first append entry to the catchup follower.
+	ae := n.buildAppendEntry([]*Entry{&Entry{EntrySnapshot, snap.data}, &Entry{EntryPeerState, snap.peerstate}})
+	ae.pterm, ae.pindex = snap.lastTerm, snap.lastIndex
+	n.sendRPC(subject, n.areply, ae.encode())
+	return snap.lastIndex, nil
+}
+
 func (n *raft) catchupFollower(ar *appendEntryResponse) {
 	n.debug("Being asked to catch up follower: %q", ar.peer)
 	n.Lock()
@@ -1490,7 +1653,22 @@ func (n *raft) catchupFollower(ar *appendEntryResponse) {
 		delete(n.progress, ar.peer)
 		ch <- n.pindex
 	}
-	ae, err := n.loadEntry(ar.index + 1)
+	// Check to make sure we have this entry.
+	start := ar.index + 1
+	state := n.wal.State()
+
+	if start < state.FirstSeq {
+		n.debug("Need to send snapshot to follower")
+		var err error
+		if _, err = n.sendSnapshotToFollower(ar.reply); err != nil {
+			n.error("Error sending snapshot to followers: %v", err)
+			n.attemptStepDown(noLeader)
+			n.Unlock()
+			return
+		}
+	}
+
+	ae, err := n.loadEntry(start)
 	if err != nil {
 		ae, err = n.loadFirstEntry()
 	}
@@ -1552,6 +1730,9 @@ func (n *raft) applyCommit(index uint64) error {
 		switch e.Type {
 		case EntryNormal:
 			committed = append(committed, e)
+		case EntryOldSnapshot:
+			// For old snapshots in our WAL.
+			committed = append(committed, &Entry{EntrySnapshot, e.Data})
 		case EntrySnapshot:
 			committed = append(committed, e)
 		case EntryPeerState:
@@ -1813,14 +1994,12 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	// Is this a new entry?
 	isNew := sub == n.aesub
 
-	if isNew {
-		// Track leader directly
-		if ae.leader != noLeader {
-			if ps := n.peers[ae.leader]; ps != nil {
-				ps.ts = time.Now().UnixNano()
-			} else {
-				n.peers[ae.leader] = &lps{time.Now().UnixNano(), 0}
-			}
+	// Track leader directly
+	if isNew && ae.leader != noLeader {
+		if ps := n.peers[ae.leader]; ps != nil {
+			ps.ts = time.Now().UnixNano()
+		} else {
+			n.peers[ae.leader] = &lps{time.Now().UnixNano(), 0}
 		}
 	}
 
@@ -1832,18 +2011,18 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	}
 
 	// If we are catching up ignore old catchup subs.
-	// This can happen when we stall a cacthup.
-	if catchingUp && !isNew && sub != n.catchup.sub {
+	// This could happen when we stall or cancel a catchup.
+	if !isNew && sub != n.catchup.sub {
 		return
 	}
 
 	// Check state if we are catching up.
-	if catchingUp && isNew {
+	if catchingUp {
 		if cs := n.catchup; cs != nil && n.pterm >= cs.cterm && n.pindex >= cs.cindex {
 			// If we are here we are good, so if we have a catchup pending we can cancel.
 			n.cancelCatchup()
 			catchingUp = false
-		} else {
+		} else if isNew {
 			var ar *appendEntryResponse
 			var inbox string
 			// Check to see if we are stalled. If so recreate our catchup state and resend response.
@@ -1883,7 +2062,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	}
 
 	if ae.pterm != n.pterm || ae.pindex != n.pindex {
-		// If this is a lower index that what we were expecting.
+		// If this is a lower index than what we were expecting.
 		if ae.pindex < n.pindex {
 			var ar *appendEntryResponse
 			if eae, err := n.loadEntry(ae.pindex); err == nil && eae != nil {
@@ -1904,17 +2083,42 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			return
 		}
 
-		// Check if we are catching up. If we are here we know this is a catchup messages since we
-		// ignore new ones above while catching up. We also know it is not what we expected, so we
-		// could need to adjust our starting index since the leader does not have entries prior, or
-		// we missed some catchup messages.
+		// Check if we are catching up. If we are here we know the leader did not have all of the entries
+		// so make sure this is a snapshot entry. If it is not start the catchup process again since it
+		// means we missed additional messages.
 		if catchingUp {
-			if len(ae.entries) > 0 {
-				n.debug("Should reset index for wal to %d", ae.pindex+1)
-				n.wal.Compact(ae.pindex + 1)
-				n.pindex = ae.pindex
-				n.commit = ae.pindex
+			// Snapshots and peerstate will always be together when a leader is catching us up.
+			if len(ae.entries) != 2 || ae.entries[0].Type != EntrySnapshot || ae.entries[1].Type != EntryPeerState {
+				n.warn("Expected first catchup entry to be a snapshot and peerstate, will retry")
+				n.cancelCatchup()
+				n.Unlock()
+				return
+			}
+			if ps, err := decodePeerState(ae.entries[1].Data); err == nil {
+				n.processPeerState(ps)
+				// Also need to copy from client's buffer.
+				ae.entries[0].Data = append(ae.entries[0].Data[:0:0], ae.entries[0].Data...)
+			} else {
+				n.warn("Could not parse snapshot peerstate correctly")
+				n.cancelCatchup()
+				n.Unlock()
+				return
+			}
+			n.pindex = ae.pindex
+			n.pterm = ae.pterm
+			n.commit = ae.commit
+			n.wal.Compact(n.pindex + 1)
+
+			// Now send snapshot to upper levels. Only send the snapshot, not the peerstate entry.
+			select {
+			case n.applyc <- &CommittedEntry{n.commit, ae.entries[:1]}:
+			default:
+				n.debug("Failed to place snapshot entry onto our apply channel")
+				n.commit--
 			}
+			n.Unlock()
+			return
+
 		} else {
 			n.debug("AppendEntry did not match %d %d with %d %d", ae.pterm, ae.pindex, n.pterm, n.pindex)
 			// Reset our term.
@@ -1940,6 +2144,9 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 					return
 				}
 				n.debug("Error storing to WAL: %v", err)
+
+				//FIXME(dlc)!!, WARN AT LEAST, RESPOND FALSE, return etc!
+
 			}
 		} else {
 			// This is a replay on startup so just take the appendEntry version.
@@ -1966,10 +2173,6 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 						n.peers[newPeer] = &lps{time.Now().UnixNano(), 0}
 					}
 				}
-			case EntrySnapshot:
-				if ae.pindex+1 > n.sindex {
-					n.sindex = ae.pindex + 1
-				}
 			}
 		}
 	}
@@ -2042,8 +2245,8 @@ func (n *raft) storeToWAL(ae *appendEntry) error {
 	// Sanity checking for now.
 	if ae.pindex != seq-1 {
 		fmt.Printf("[%s] n is %+v\n\n", n.s, n)
-		fmt.Printf("[%s] n.cactchup is %+v\n", n.s, n.catchup)
-		panic(fmt.Sprintf("[%s-%s] Placed an entry at the wrong index, ae is %+v, index is %d\n\n", n.s, n.group, ae, seq))
+		fmt.Printf("[%s] n.catchup is %+v\n", n.s, n.catchup)
+		panic(fmt.Sprintf("[%s-%s] Placed an entry at the wrong index, ae is %+v, seq is %d, n.pindex is %d\n\n", n.s, n.group, ae, seq, n.pindex))
 	}
 
 	n.pterm = ae.term
@@ -2066,12 +2269,6 @@ func (n *raft) sendAppendEntry(entries []*Entry) {
 		}
 		// We count ourselves.
 		n.acks[n.pindex] = map[string]struct{}{n.id: struct{}{}}
-		// Check for snapshot
-		for _, e := range entries {
-			if e.Type == EntrySnapshot {
-				n.sindex = n.pindex
-			}
-		}
 		n.active = time.Now()
 	}
 	n.sendRPC(n.asubj, n.areply, ae.buf)
@@ -2082,9 +2279,13 @@ type peerState struct {
 	clusterSize int
 }
 
+func peerStateBufSize(ps *peerState) int {
+	return 4 + 4 + (8 * len(ps.knownPeers))
+}
+
 func encodePeerState(ps *peerState) []byte {
 	var le = binary.LittleEndian
-	buf := make([]byte, 4+4+(8*len(ps.knownPeers)))
+	buf := make([]byte, peerStateBufSize(ps))
 	le.PutUint32(buf[0:], uint32(ps.clusterSize))
 	le.PutUint32(buf[4:], uint32(len(ps.knownPeers)))
 	wi := 8
diff --git a/server/server.go b/server/server.go
index aab4446b..7c15ea3b 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1647,8 +1647,9 @@ func (s *Server) Start() {
 // Shutdown will shutdown the server instance by kicking out the AcceptLoop
 // and closing all associated clients.
 func (s *Server) Shutdown() {
-	// Shutdown our raftnodes. If we are the leader we will attempt to transfer.
-	s.shutdownRaftNodes()
+	// Transfer off any raft nodes that we are a leader.
+	s.transferRaftLeaders()
+
 	// Shutdown the eventing system as needed.
 	// This is done first to send out any messages for
 	// account status. We will also clean up any
diff --git a/server/stream.go b/server/stream.go
index 38c4cf14..e95dc573 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -943,7 +943,7 @@ func (mset *stream) setupStore(fsCfg *FileStoreConfig) error {
 		}
 		mset.store = ms
 	case FileStorage:
-		fs, _, err := newFileStoreWithCreated(*fsCfg, mset.cfg, mset.created)
+		fs, err := newFileStoreWithCreated(*fsCfg, mset.cfg, mset.created)
 		if err != nil {
 			mset.mu.Unlock()
 			return err
@@ -1347,7 +1347,10 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 
 		mset.mu.Lock()
 		for _, o := range mset.consumers {
-			if o.isLeader() {
+			o.mu.RLock()
+			isLeader := o.isLeader()
+			o.mu.RUnlock()
+			if isLeader {
 				obs = append(obs, o)
 			}
 		}
diff --git a/util/mkpasswd/mkpasswd.go b/util/mkpasswd/mkpasswd.go
new file mode 100644
index 00000000..76e0800c
--- /dev/null
+++ b/util/mkpasswd/mkpasswd.go
@@ -0,0 +1,95 @@
+// Copyright 2015-2018 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build ignore
+
+package main
+
+import (
+	"bytes"
+	"crypto/rand"
+	"flag"
+	"fmt"
+	"log"
+	"math/big"
+	"syscall"
+
+	"golang.org/x/crypto/bcrypt"
+	"golang.org/x/crypto/ssh/terminal"
+)
+
+func usage() {
+	fmt.Printf("Usage: mkpasswd [-p <stdin password>] [-c COST] \n")
+	flag.PrintDefaults()
+}
+
+const (
+	// Make sure the password is reasonably long to generate enough entropy.
+	PasswordLength = 22
+	// Common advice from the past couple of years suggests that 10 should be sufficient.
+	// Up that a little, to 11. Feel free to raise this higher if this value from 2015 is
+	// no longer appropriate. Min is bcrypt.MinCost, Max is bcrypt.MaxCost.
+	DefaultCost = 11
+)
+
+func main() {
+	var pw = flag.Bool("p", false, "Input password via stdin")
+	var cost = flag.Int("c", DefaultCost, fmt.Sprintf("The cost weight, range of %d-%d", bcrypt.MinCost, bcrypt.MaxCost))
+
+	log.SetFlags(0)
+	flag.Usage = usage
+	flag.Parse()
+
+	var password string
+
+	if *pw {
+		fmt.Printf("Enter Password: ")
+		bytePassword, err := terminal.ReadPassword(int(syscall.Stdin))
+		if err != nil {
+			log.Fatalf("Error reading password: %v\n", err)
+		}
+		fmt.Printf("\nReenter Password: ")
+		bytePassword2, err := terminal.ReadPassword(int(syscall.Stdin))
+		if err != nil {
+			log.Fatalf("Error reading password: %v\n", err)
+		}
+		if !bytes.Equal(bytePassword, bytePassword2) {
+			log.Fatalf("Error, passwords do not match\n")
+		}
+		password = string(bytePassword)
+		fmt.Printf("\n")
+	} else {
+		password = genPassword()
+		fmt.Printf("pass: %s\n", password)
+	}
+
+	cb, err := bcrypt.GenerateFromPassword([]byte(password), *cost)
+	if err != nil {
+		log.Fatalf("Error producing bcrypt hash: %v\n", err)
+	}
+	fmt.Printf("bcrypt hash: %s\n", cb)
+}
+
+func genPassword() string {
+	var ch = []byte("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!@$#%^&*()")
+	b := make([]byte, PasswordLength)
+	max := big.NewInt(int64(len(ch)))
+	for i := range b {
+		ri, err := rand.Int(rand.Reader, max)
+		if err != nil {
+			log.Fatalf("Error producing random integer: %v\n", err)
+		}
+		b[i] = ch[int(ri.Int64())]
+	}
+	return string(b)
+}


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 3f9352d2..f5718413 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -37,7 +37,7 @@ import (
 
 func TestFileStoreBasics(t *testing.T) {
 	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -103,7 +103,7 @@ func TestFileStoreBasics(t *testing.T) {
 
 func TestFileStoreMsgHeaders(t *testing.T) {
 	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -137,14 +137,14 @@ func TestFileStoreBasicWriteMsgsAndRestore(t *testing.T) {
 
 	fcfg := FileStoreConfig{StoreDir: storeDir}
 
-	if _, _, err := newFileStore(fcfg, StreamConfig{Storage: MemoryStorage}); err == nil {
+	if _, err := newFileStore(fcfg, StreamConfig{Storage: MemoryStorage}); err == nil {
 		t.Fatalf("Expected an error with wrong type")
 	}
-	if _, _, err := newFileStore(fcfg, StreamConfig{Storage: FileStorage}); err == nil {
+	if _, err := newFileStore(fcfg, StreamConfig{Storage: FileStorage}); err == nil {
 		t.Fatalf("Expected an error with no name")
 	}
 
-	fs, _, err := newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
+	fs, err := newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -181,7 +181,7 @@ func TestFileStoreBasicWriteMsgsAndRestore(t *testing.T) {
 	}
 
 	// Restart
-	fs, _, err = newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
+	fs, err = newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -214,7 +214,7 @@ func TestFileStoreBasicWriteMsgsAndRestore(t *testing.T) {
 	fs.Stop()
 
 	// Restart
-	fs, _, err = newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
+	fs, err = newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -231,7 +231,7 @@ func TestFileStoreBasicWriteMsgsAndRestore(t *testing.T) {
 	fs.Purge()
 	fs.Stop()
 
-	fs, _, err = newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
+	fs, err = newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -252,7 +252,7 @@ func TestFileStoreBasicWriteMsgsAndRestore(t *testing.T) {
 	fs.RemoveMsg(seq)
 
 	fs.Stop()
-	fs, _, err = newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
+	fs, err = newFileStore(fcfg, StreamConfig{Name: "dlc", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -269,7 +269,7 @@ func TestFileStoreSelectNextFirst(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -309,7 +309,7 @@ func TestFileStoreSkipMsg(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -344,7 +344,7 @@ func TestFileStoreSkipMsg(t *testing.T) {
 	// Make sure we recover same state.
 	fs.Stop()
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -378,7 +378,7 @@ func TestFileStoreSkipMsg(t *testing.T) {
 	// Make sure we recover same state.
 	fs.Stop()
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -399,7 +399,7 @@ func TestFileStoreWriteExpireWrite(t *testing.T) {
 	defer os.RemoveAll(storeDir)
 
 	cexp := 10 * time.Millisecond
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: 256, CacheExpire: cexp},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -432,7 +432,7 @@ func TestFileStoreWriteExpireWrite(t *testing.T) {
 	// Make sure we recover same state.
 	fs.Stop()
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -463,7 +463,7 @@ func TestFileStoreMsgLimit(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxMsgs: 10})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxMsgs: 10})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -496,6 +496,29 @@ func TestFileStoreMsgLimit(t *testing.T) {
 	}
 }
 
+func TestFileStoreMsgLimitBug(t *testing.T) {
+	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxMsgs: 1})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	subj, msg := "foo", []byte("Hello World")
+	fs.StoreMsg(subj, nil, msg)
+	fs.StoreMsg(subj, nil, msg)
+	fs.Stop()
+
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxMsgs: 1})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	fs.StoreMsg(subj, nil, msg)
+}
+
 func TestFileStoreBytesLimit(t *testing.T) {
 	subj, msg := "foo", make([]byte, 512)
 	storedMsgSize := fileStoreMsgSize(subj, nil, msg)
@@ -507,7 +530,7 @@ func TestFileStoreBytesLimit(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxBytes: int64(maxBytes)})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxBytes: int64(maxBytes)})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -552,7 +575,7 @@ func TestFileStoreAgeLimit(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxAge: maxAge})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxAge: maxAge})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -599,7 +622,7 @@ func TestFileStoreTimeStamps(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -630,7 +653,7 @@ func TestFileStorePurge(t *testing.T) {
 	defer os.RemoveAll(storeDir)
 
 	blkSize := uint64(64 * 1024)
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: blkSize}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: blkSize}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -683,7 +706,7 @@ func TestFileStorePurge(t *testing.T) {
 	// Make sure we recover same state.
 	fs.Stop()
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: blkSize}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: blkSize}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -722,7 +745,7 @@ func TestFileStorePurge(t *testing.T) {
 	purgeDir := path.Join(fs.fcfg.StoreDir, purgeDir)
 	os.Rename(pdir, purgeDir)
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: blkSize}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: blkSize}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -747,7 +770,7 @@ func TestFileStoreCompact(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -792,7 +815,7 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: 350},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -846,7 +869,7 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 
 	// Make sure we can recover same state.
 	fs.Stop()
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -862,7 +885,7 @@ func TestFileStoreRemovePartialRecovery(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -891,7 +914,7 @@ func TestFileStoreRemovePartialRecovery(t *testing.T) {
 	// Make sure we recover same state.
 	fs.Stop()
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -908,7 +931,7 @@ func TestFileStoreRemoveOutOfOrderRecovery(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -948,7 +971,7 @@ func TestFileStoreRemoveOutOfOrderRecovery(t *testing.T) {
 	// Make sure we recover same state.
 	fs.Stop()
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -976,7 +999,7 @@ func TestFileStoreAgeLimitRecovery(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir, CacheExpire: 1 * time.Millisecond},
 		StreamConfig{Name: "zzz", Storage: FileStorage, MaxAge: maxAge},
 	)
@@ -997,7 +1020,7 @@ func TestFileStoreAgeLimitRecovery(t *testing.T) {
 	}
 	fs.Stop()
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxAge: maxAge})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxAge: maxAge})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1022,7 +1045,7 @@ func TestFileStoreBitRot(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1068,7 +1091,7 @@ func TestFileStoreBitRot(t *testing.T) {
 	// Make sure we can restore.
 	fs.Stop()
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1085,7 +1108,7 @@ func TestFileStoreEraseMsg(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1144,7 +1167,7 @@ func TestFileStoreEraseAndNoIndexRecovery(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1176,7 +1199,7 @@ func TestFileStoreEraseAndNoIndexRecovery(t *testing.T) {
 	ifn := path.Join(storeDir, msgDir, fmt.Sprintf(indexScan, 1))
 	os.Remove(ifn)
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1201,7 +1224,7 @@ func TestFileStoreMeta(t *testing.T) {
 
 	mconfig := StreamConfig{Name: "ZZ-22-33", Storage: FileStorage, Subjects: []string{"foo.*"}, Replicas: 22}
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, mconfig)
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, mconfig)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1295,7 +1318,7 @@ func TestFileStoreWriteAndReadSameBlock(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -1322,7 +1345,7 @@ func TestFileStoreAndRetrieveMultiBlock(t *testing.T) {
 	subj, msg := "foo", []byte("Hello World!")
 	storedMsgSize := fileStoreMsgSize(subj, nil, msg)
 
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: 4 * storedMsgSize},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -1339,7 +1362,7 @@ func TestFileStoreAndRetrieveMultiBlock(t *testing.T) {
 	}
 	fs.Stop()
 
-	fs, _, err = newFileStore(
+	fs, err = newFileStore(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: 4 * storedMsgSize},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -1363,7 +1386,7 @@ func TestFileStoreCollapseDmap(t *testing.T) {
 	subj, msg := "foo", []byte("Hello World!")
 	storedMsgSize := fileStoreMsgSize(subj, nil, msg)
 
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: 4 * storedMsgSize},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -1438,7 +1461,7 @@ func TestFileStoreReadCache(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir, CacheExpire: 100 * time.Millisecond}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, CacheExpire: 100 * time.Millisecond}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1492,7 +1515,7 @@ func TestFileStorePartialCacheExpiration(t *testing.T) {
 
 	cexp := 10 * time.Millisecond
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir, CacheExpire: cexp}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, CacheExpire: cexp}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1518,7 +1541,7 @@ func TestFileStorePartialIndexes(t *testing.T) {
 
 	cexp := 10 * time.Millisecond
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir, CacheExpire: cexp}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, CacheExpire: cexp}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1566,7 +1589,7 @@ func TestFileStoreSnapshot(t *testing.T) {
 
 	subj, msg := "foo", []byte("Hello Snappy!")
 
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -1649,7 +1672,7 @@ func TestFileStoreSnapshot(t *testing.T) {
 			fd.Close()
 		}
 
-		fsr, _, err := newFileStore(
+		fsr, err := newFileStore(
 			FileStoreConfig{StoreDir: rstoreDir},
 			StreamConfig{Name: "zzz", Storage: FileStorage},
 		)
@@ -1739,7 +1762,7 @@ func TestFileStoreConsumer(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1752,10 +1775,8 @@ func TestFileStoreConsumer(t *testing.T) {
 	if state, err := o.State(); state != nil || err != nil {
 		t.Fatalf("Unexpected state or error: %v", err)
 	}
+
 	state := &ConsumerState{}
-	if err := o.Update(state); err == nil {
-		t.Fatalf("Expected an error and got none")
-	}
 
 	updateAndCheck := func() {
 		t.Helper()
@@ -1866,7 +1887,7 @@ func TestFileStoreWriteFailures(t *testing.T) {
 	defer os.RemoveAll(storeDir)
 
 	subj, msg := "foo", []byte("Hello Write Failures!")
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1902,7 +1923,7 @@ func TestFileStoreWriteFailures(t *testing.T) {
 	// Make sure we recover same state.
 	fs.Stop()
 
-	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1961,7 +1982,7 @@ func TestFileStorePerf(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -1985,7 +2006,7 @@ func TestFileStorePerf(t *testing.T) {
 
 	fmt.Printf("Restoring..\n")
 	start = time.Now()
-	fs, _, err = newFileStore(
+	fs, err = newFileStore(
 		FileStoreConfig{StoreDir: storeDir},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -2033,7 +2054,7 @@ func TestFileStorePerf(t *testing.T) {
 
 	fs.Stop()
 
-	fs, _, err = newFileStore(
+	fs, err = newFileStore(
 		FileStoreConfig{StoreDir: storeDir},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -2060,7 +2081,7 @@ func TestFileStorePerf(t *testing.T) {
 	fmt.Printf("%.0f msgs/sec\n", float64(toStore)/tt.Seconds())
 	fmt.Printf("%s per sec\n", friendlyBytes(int64(float64(toStore*storedMsgSize)/tt.Seconds())))
 
-	fs, _, err = newFileStore(
+	fs, err = newFileStore(
 		FileStoreConfig{StoreDir: storeDir},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -2101,7 +2122,7 @@ func TestFileStoreReadBackMsgPerf(t *testing.T) {
 	defer os.RemoveAll(storeDir)
 	fmt.Printf("StoreDir is %q\n", storeDir)
 
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -2151,7 +2172,7 @@ func TestFileStoreStoreLimitRemovePerf(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -2208,7 +2229,7 @@ func TestFileStorePubPerfWithSmallBlkSize(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(
+	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: FileStoreMinBlkSize},
 		StreamConfig{Name: "zzz", Storage: FileStorage},
 	)
@@ -2234,7 +2255,7 @@ func TestFileStoreConsumerRedeliveredLost(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -2301,7 +2322,7 @@ func TestFileStoreConsumerFlusher(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -2336,7 +2357,7 @@ func TestFileStoreConsumerDeliveredUpdates(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -2394,7 +2415,7 @@ func TestFileStoreConsumerDeliveredAndAckUpdates(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -2511,7 +2532,7 @@ func TestFileStoreStreamStateDeleted(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -2563,7 +2584,7 @@ func TestFileStoreConsumerPerf(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -2644,7 +2665,7 @@ func TestFileStoreStreamIndexBug(t *testing.T) {
 // Reported by Ivan.
 func TestFileStoreStreamDeleteCacheBug(t *testing.T) {
 	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
-	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir, CacheExpire: 50 * time.Millisecond}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, CacheExpire: 50 * time.Millisecond}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 153115a0..40cf1045 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -172,9 +172,8 @@ func TestJetStreamClusterSingleReplicaStreams(t *testing.T) {
 	sl := c.streamLeader("$G", "TEST")
 	sl.Shutdown()
 	c.restartServer(sl)
-	c.waitOnStreamLeader("$G", "TEST")
-	time.Sleep(500 * time.Millisecond)
 
+	c.waitOnStreamLeader("$G", "TEST")
 	si, err = js.StreamInfo("TEST")
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -184,8 +183,6 @@ func TestJetStreamClusterSingleReplicaStreams(t *testing.T) {
 	}
 	// Now durable consumer.
 	c.waitOnConsumerLeader("$G", "TEST", "dlc")
-	time.Sleep(500 * time.Millisecond)
-
 	if _, err = js.ConsumerInfo("TEST", "dlc"); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -610,8 +607,8 @@ func TestJetStreamClusterMetaSnapshotsAndCatchup(t *testing.T) {
 	// Shut one down.
 	rs := c.randomServer()
 	rs.Shutdown()
-	c.waitOnLeader()
 
+	c.waitOnLeader()
 	s := c.leader()
 
 	// Client based API
@@ -630,7 +627,6 @@ func TestJetStreamClusterMetaSnapshotsAndCatchup(t *testing.T) {
 	}
 
 	c.leader().JetStreamSnapshotMeta()
-	time.Sleep(250 * time.Millisecond)
 
 	rs = c.restartServer(rs)
 	c.checkClusterFormed()
@@ -641,17 +637,10 @@ func TestJetStreamClusterMetaSnapshotsAndCatchup(t *testing.T) {
 
 	for i := 0; i < numStreams; i++ {
 		sn := fmt.Sprintf("T-%d", i+1)
-		resp, err := nc.Request(fmt.Sprintf(JSApiStreamDeleteT, sn), nil, time.Second)
+		err := js.DeleteStream(sn)
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
-		var dResp JSApiStreamDeleteResponse
-		if err := json.Unmarshal(resp.Data, &dResp); err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		if !dResp.Success || dResp.Error != nil {
-			t.Fatalf("Got a bad response %+v", dResp.Error)
-		}
 	}
 
 	rs = c.restartServer(rs)
@@ -1589,11 +1578,17 @@ func TestJetStreamClusterExtendedStreamInfo(t *testing.T) {
 		t.Fatalf("Expected %d replicas, got %d", 2, len(si.Cluster.Replicas))
 	}
 
+	// 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)
+	}
+
 	// We may need to wait a bit for peers to catch up.
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		for _, peer := range si.Cluster.Replicas {
 			if !peer.Current {
-				if si, err = js.StreamInfo("TEST"); err != nil {
+				if si, err = js2.StreamInfo("TEST"); err != nil {
 					t.Fatalf("Could not retrieve stream info")
 				}
 				return fmt.Errorf("Expected replica to be current: %+v", peer)
@@ -1658,11 +1653,12 @@ func TestJetStreamClusterExtendedStreamInfo(t *testing.T) {
 	if len(si.Cluster.Replicas) != 2 {
 		t.Fatalf("Expected %d replicas, got %d", 2, len(si.Cluster.Replicas))
 	}
+
 	// We may need to wait a bit for peers to catch up.
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		for _, peer := range si.Cluster.Replicas {
 			if !peer.Current {
-				if si, err = js.StreamInfo("TEST"); err != nil {
+				if si, err = js2.StreamInfo("TEST"); err != nil {
 					t.Fatalf("Could not retrieve stream info")
 				}
 				return fmt.Errorf("Expected replica to be current: %+v", peer)
@@ -1781,6 +1777,136 @@ func TestJetStreamClusterExtendedStreamInfoSingleReplica(t *testing.T) {
 
 }
 
+func TestJetStreamClusterInterestRetention(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	// Client based API
+	s := c.randomServer()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "foo", Retention: nats.InterestPolicy, Replicas: 3})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	sub, err := js.SubscribeSync("foo", nats.Durable("dlc"))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	sl := c.streamLeader("$G", "foo")
+	cl := c.consumerLeader("$G", "foo", "dlc")
+	if sl == cl {
+		_, err := nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "foo"), nil, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		c.waitOnStreamLeader("$G", "foo")
+	}
+
+	if _, err = js.Publish("foo", []byte("OK")); err != nil {
+		t.Fatalf("Unexpected publish error: %v", err)
+	}
+
+	m, err := sub.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error getting msg: %v", err)
+	}
+	m.Ack()
+
+	js, 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 := js.StreamInfo("foo")
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		if si.State.Msgs != 0 {
+			return fmt.Errorf("Expected 0 msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
+}
+
+func TestJetStreamClusterInterestRetentionWithFilteredConsumers(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	// Client based API
+	s := c.randomServer()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Subjects: []string{"*"}, Retention: nats.InterestPolicy, Replicas: 3})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	fsub, err := js.SubscribeSync("foo")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fsub.Unsubscribe()
+
+	bsub, err := js.SubscribeSync("bar")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer bsub.Unsubscribe()
+
+	msg := []byte("FILTERED")
+	sendMsg := func(subj string) {
+		t.Helper()
+		if _, err = js.Publish(subj, msg); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+
+	getAndAck := func(sub *nats.Subscription) {
+		t.Helper()
+		m, err := sub.NextMsg(time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error getting msg: %v", err)
+		}
+		m.Ack()
+	}
+
+	jsq, err := nc.JetStream(nats.MaxWait(10 * time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	checkState := func(expected uint64) {
+		t.Helper()
+		checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+			si, err := jsq.StreamInfo("TEST")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if si.State.Msgs != expected {
+				return fmt.Errorf("Expected %d msgs, got %d", expected, si.State.Msgs)
+			}
+			return nil
+		})
+	}
+
+	sendMsg("foo")
+	checkState(1)
+	getAndAck(fsub)
+	checkState(0)
+	sendMsg("bar")
+	sendMsg("foo")
+	checkState(2)
+	getAndAck(bsub)
+	checkState(1)
+	getAndAck(fsub)
+	checkState(0)
+}
+
 func TestJetStreamClusterEphemeralConsumerCleanup(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
@@ -1885,7 +2011,13 @@ func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 		}
 	}
 
-	// Create a consumer as well and give it a non-simplistic state.
+	// Create consumer with no state.
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "rip", AckPolicy: nats.AckExplicitPolicy, AckWait: time.Second})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Create another consumer as well and give it a non-simplistic state.
 	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy, AckWait: time.Second})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -1914,9 +2046,9 @@ func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 			m.Ack()
 		}
 	}
+	nc.Flush()
 
 	// Snapshot consumer info.
-	time.Sleep(100 * time.Millisecond)
 	ci, err := jsub.ConsumerInfo()
 	if err != nil {
 		t.Fatalf("Unexpected error getting consumer info: %v", err)
@@ -2069,9 +2201,9 @@ func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 	start := 101
 	end := toSend
 	for i := start; i <= end; i++ {
-		m, err := jsub.NextMsg(time.Second)
+		m, err := jsub.NextMsg(2 * time.Second)
 		if err != nil {
-			t.Fatalf("Unexpected error getting msg: %v", err)
+			t.Fatalf("Unexpected error getting msg [%d]: %v", i, err)
 		}
 		meta, err := m.MetaData()
 		if err != nil {
@@ -2086,6 +2218,37 @@ func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 	// Check that redelivered come in now..
 	redelivered := 50/3 + 1
 	checkSubsPending(t, jsub, redelivered)
+
+	// Now make sure the other server was properly caughtup.
+	// Need to call this by hand for now.
+	rmsg, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "TEST"), nil, time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	var sdResp JSApiStreamLeaderStepDownResponse
+	if err := json.Unmarshal(rmsg.Data, &sdResp); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if sdResp.Error != nil {
+		t.Fatalf("Unexpected error: %+v", sdResp.Error)
+	}
+
+	c.waitOnStreamLeader("$G", "TEST")
+	si, err = js.StreamInfo("TEST")
+	if err != nil {
+		t.Fatalf("Unexpected error: %+v", err)
+	}
+	if si.State.Msgs != uint64(toSend) {
+		t.Fatalf("Unexpected stream info: %+v", si)
+	}
+
+	// Check idle consumer
+	c.waitOnConsumerLeader("$G", "TEST", "rip")
+
+	// Now check for the consumer being recreated.
+	if _, err := js.ConsumerInfo("TEST", "rip"); err != nil {
+		t.Fatalf("Unexpected error: %+v", err)
+	}
 }
 
 func TestJetStreamClusterAccountInfoAndLimits(t *testing.T) {
@@ -2741,6 +2904,7 @@ func TestJetStreamClusterRestartAndRemoveAdvisories(t *testing.T) {
 	defer usub.Unsubscribe()
 	nc.Flush()
 
+	checkSubsPending(t, csub, 0)
 	checkSubsPending(t, sub, 0)
 	checkSubsPending(t, usub, 0)
 
@@ -3013,7 +3177,7 @@ func TestJetStreamClusterRemovePeer(t *testing.T) {
 	})
 }
 
-func TestJetStreamClusterStepDown(t *testing.T) {
+func TestJetStreamClusterStreamLeaderStepDown(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "RNS", 3)
 	defer c.shutdown()
 
@@ -3141,12 +3305,23 @@ func TestJetStreamClusterRemoveServer(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 	checkSubsPending(t, sub, toSend)
+	ci, err := sub.ConsumerInfo()
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	cname := ci.Name
 
 	sl := c.streamLeader("$G", "TEST")
 	c.removeJetStream(sl)
 
 	c.waitOnStreamLeader("$G", "TEST")
 
+	// Faster timeout since we loop below checking for condition.
+	js, err = nc.JetStream(nats.MaxWait(10 * time.Millisecond))
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
 	// Check the stream info is eventually correct.
 	checkFor(t, 2*time.Second, 50*time.Millisecond, func() error {
 		si, err := js.StreamInfo("TEST")
@@ -3165,8 +3340,9 @@ func TestJetStreamClusterRemoveServer(t *testing.T) {
 	})
 
 	// Now do consumer.
+	c.waitOnConsumerLeader("$G", "TEST", cname)
 	checkFor(t, 5*time.Second, 50*time.Millisecond, func() error {
-		ci, err := sub.ConsumerInfo()
+		ci, err := js.ConsumerInfo("TEST", cname)
 		if err != nil {
 			return fmt.Errorf("Could not fetch consumer info: %v", err)
 		}
@@ -3450,7 +3626,7 @@ func TestJetStreamClusterStreamPerf(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
-	numConnections := 2
+	numConnections := 4
 	var conns []nats.JetStream
 	for i := 0; i < numConnections; i++ {
 		s := c.randomServer()
@@ -3459,7 +3635,7 @@ func TestJetStreamClusterStreamPerf(t *testing.T) {
 	}
 
 	toSend := 100000
-	numProducers := 10
+	numProducers := 8
 
 	payload := []byte("Hello JSC")
 
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 912847ee..08341023 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -5941,6 +5941,87 @@ func TestJetStreamInterestRetentionStream(t *testing.T) {
 	}
 }
 
+func TestJetStreamInterestRetentionStreamWithFilteredConsumers(t *testing.T) {
+	cases := []struct {
+		name    string
+		mconfig *StreamConfig
+	}{
+		{"MemoryStore", &StreamConfig{Name: "DC", Subjects: []string{"*"}, Storage: MemoryStorage, Retention: InterestPolicy}},
+		{"FileStore", &StreamConfig{Name: "DC", Subjects: []string{"*"}, Storage: FileStorage, Retention: InterestPolicy}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			s := RunBasicJetStreamServer()
+			defer s.Shutdown()
+
+			if config := s.JetStreamConfig(); config != nil {
+				defer os.RemoveAll(config.StoreDir)
+			}
+
+			mset, err := s.GlobalAccount().addStream(c.mconfig)
+			if err != nil {
+				t.Fatalf("Unexpected error adding stream: %v", err)
+			}
+			defer mset.delete()
+
+			nc, js := jsClientConnect(t, s)
+			defer nc.Close()
+
+			fsub, err := js.SubscribeSync("foo")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			defer fsub.Unsubscribe()
+
+			bsub, err := js.SubscribeSync("bar")
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			defer bsub.Unsubscribe()
+
+			msg := []byte("FILTERED")
+			sendMsg := func(subj string) {
+				t.Helper()
+				if _, err = js.Publish(subj, msg); err != nil {
+					t.Fatalf("Unexpected publish error: %v", err)
+				}
+			}
+
+			getAndAck := func(sub *nats.Subscription) {
+				t.Helper()
+				m, err := sub.NextMsg(time.Second)
+				if err != nil {
+					t.Fatalf("Unexpected error getting msg: %v", err)
+				}
+				m.Ack()
+			}
+
+			checkState := func(expected uint64) {
+				t.Helper()
+				si, err := js.StreamInfo("DC")
+				if err != nil {
+					t.Fatalf("Unexpected error: %v", err)
+				}
+				if si.State.Msgs != expected {
+					t.Fatalf("Expected %d msgs, got %d", expected, si.State.Msgs)
+				}
+			}
+
+			sendMsg("foo")
+			checkState(1)
+			getAndAck(fsub)
+			checkState(0)
+			sendMsg("bar")
+			sendMsg("foo")
+			checkState(2)
+			getAndAck(bsub)
+			checkState(1)
+			getAndAck(fsub)
+			checkState(0)
+		})
+	}
+}
+
 func TestJetStreamInterestRetentionWithWildcardsAndFilteredConsumers(t *testing.T) {
 	msc := StreamConfig{
 		Name:      "DCWC",
diff --git a/test/auth_test.go b/test/auth_test.go
index 232ae717..4ffaa39f 100644
--- a/test/auth_test.go
+++ b/test/auth_test.go
@@ -169,7 +169,7 @@ func TestPasswordClientGoodConnect(t *testing.T) {
 // The bcrypt username/password version
 ////////////////////////////////////////////////////////////
 
-// Generated with nats server passwd (Cost 4 because of cost of --race, default is 11)
+// Generated with util/mkpasswd (Cost 4 because of cost of --race, default is 11)
 const BCRYPT_AUTH_PASS = "IW@$6v(y1(t@fhPDvf!5^%"
 const BCRYPT_AUTH_HASH = "$2a$04$Q.CgCP2Sl9pkcTXEZHazaeMwPaAkSHk7AI51HkyMt5iJQQyUA4qxq"