Repository: nats-io/nats-server
Pull request #1892 (2021-02-07 12:48:17)

JetStream updates and improvments
More stability improvements to the raft core.
Made filestore better at handling write errors and propagating them.
Removed majority of MarshalIndent.

/cc @nats-io/core



Patch:
diff --git a/server/client.go b/server/client.go
index 62517f7f..527c9017 100644
--- a/server/client.go
+++ b/server/client.go
@@ -4013,14 +4013,8 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 				}
 			}
 
-			var rreply = reply
-			if rplyHasGWPrefix && sub.client.kind == CLIENT {
-				rreply = creply
-			}
-			// "rreply" will be stripped of the $GNR prefix (if present)
-			// for client connections only.
-			mh := c.msgHeader(dsubj, rreply, sub)
-			if c.deliverMsg(sub, subject, rreply, mh, msg, rplyHasGWPrefix) {
+			mh := c.msgHeader(dsubj, creply, sub)
+			if c.deliverMsg(sub, subject, creply, mh, msg, rplyHasGWPrefix) {
 				didDeliver = true
 				// Clear rsub
 				rsub = nil
diff --git a/server/consumer.go b/server/consumer.go
index dd137ce6..6f398a7f 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -785,7 +785,7 @@ func (o *Consumer) sendDeleteAdvisoryLocked() {
 		Action:   DeleteEvent,
 	}
 
-	j, err := json.MarshalIndent(e, "", "  ")
+	j, err := json.Marshal(e)
 	if err != nil {
 		return
 	}
@@ -809,7 +809,7 @@ func (o *Consumer) sendCreateAdvisory() {
 		Action:   CreateEvent,
 	}
 
-	j, err := json.MarshalIndent(e, "", "  ")
+	j, err := json.Marshal(e)
 	if err != nil {
 		return
 	}
@@ -1140,7 +1140,7 @@ func (o *Consumer) processTerm(sseq, dseq, dc uint64) {
 		Deliveries:  dc,
 	}
 
-	j, err := json.MarshalIndent(e, "", "  ")
+	j, err := json.Marshal(e)
 	if err != nil {
 		return
 	}
@@ -1395,7 +1395,7 @@ func (o *Consumer) sampleAck(sseq, dseq, dc uint64) {
 		Deliveries:  dc,
 	}
 
-	j, err := json.MarshalIndent(e, "", "  ")
+	j, err := json.Marshal(e)
 	if err != nil {
 		return
 	}
@@ -1720,7 +1720,7 @@ func (o *Consumer) notifyDeliveryExceeded(sseq, dc uint64) {
 		Deliveries: dc,
 	}
 
-	j, err := json.MarshalIndent(e, "", "  ")
+	j, err := json.Marshal(e)
 	if err != nil {
 		return
 	}
diff --git a/server/events.go b/server/events.go
index 8722c8b8..90522c14 100644
--- a/server/events.go
+++ b/server/events.go
@@ -303,7 +303,7 @@ RESET:
 				case []byte:
 					b = v
 				default:
-					b, _ = json.MarshalIndent(pm.msg, _EMPTY_, "  ")
+					b, _ = json.Marshal(pm.msg)
 				}
 			}
 
@@ -371,7 +371,8 @@ func (s *Server) sendShutdownEvent() {
 	s.sys.replies = nil
 	s.mu.Unlock()
 	// Send to the internal queue and mark as last.
-	sendq <- &pubMsg{nil, subj, _EMPTY_, nil, nil, true}
+	si := &ServerInfo{}
+	sendq <- &pubMsg{nil, subj, _EMPTY_, si, si, true}
 }
 
 // Used to send an internal message to an arbitrary account.
@@ -626,6 +627,11 @@ func (s *Server) initEventTracking() {
 	if _, err := s.sysSubscribe(accNumSubsReqSubj, s.nsubsRequest); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
 	}
+	// Listen for statsz from others.
+	subject = fmt.Sprintf(serverStatsSubj, "*")
+	if _, err := s.sysSubscribe(subject, s.remoteServerUpdate); err != nil {
+		s.Errorf("Error setting up internal tracking: %v", err)
+	}
 	// Listen for all server shutdowns.
 	subject = fmt.Sprintf(shutdownEventSubj, "*")
 	if _, err := s.sysSubscribe(subject, s.remoteServerShutdown); err != nil {
@@ -844,10 +850,41 @@ func (s *Server) remoteServerShutdown(sub *subscription, _ *client, subject, rep
 	}
 
 	sid := toks[serverSubjectIndex]
-	su := s.sys.servers[sid]
-	if su != nil {
+	if su := s.sys.servers[sid]; su != nil {
 		s.processRemoteServerShutdown(sid)
 	}
+
+	if len(msg) == 0 {
+		return
+	}
+
+	// We have an optional serverInfo here, remove from nodeToX lookups.
+	var si ServerInfo
+	if err := json.Unmarshal(msg, &si); err != nil {
+		s.Debugf("Received bad server info for remote server shutdown")
+		return
+	}
+	// Additional processing here.
+	node := string(getHash(si.Name))
+	s.nodeToInfo.Store(node, &nodeInfo{si.Name, si.Cluster, si.ID, true})
+}
+
+// remoteServerUpdate listens for statsz updates from other servers.
+func (s *Server) remoteServerUpdate(sub *subscription, _ *client, subject, reply string, msg []byte) {
+	var ssm ServerStatsMsg
+	if err := json.Unmarshal(msg, &ssm); err != nil {
+		s.Debugf("Received bad server info for remote server update")
+		return
+	}
+	si := ssm.Server
+	node := string(getHash(si.Name))
+	if _, ok := s.nodeToInfo.Load(node); !ok {
+		// Since we have not seen this one they probably have not seen us so send out our update.
+		s.mu.Lock()
+		s.sendStatsz(fmt.Sprintf(serverStatsSubj, s.info.ID))
+		s.mu.Unlock()
+	}
+	s.nodeToInfo.Store(node, &nodeInfo{si.Name, si.Cluster, si.ID, false})
 }
 
 // updateRemoteServer is called when we have an update from a remote server.
@@ -878,8 +915,7 @@ func (s *Server) processNewServer(ms *ServerInfo) {
 	s.ensureGWsInterestOnlyForLeafNodes()
 	// Add to our nodeToName
 	node := string(getHash(ms.Name))
-	s.nodeToName.Store(node, ms.Name)
-	s.nodeToCluster.Store(node, ms.Cluster)
+	s.nodeToInfo.Store(node, &nodeInfo{ms.Name, ms.Cluster, ms.ID, false})
 }
 
 // If GW is enabled on this server and there are any leaf node connections,
diff --git a/server/filestore.go b/server/filestore.go
index 403b1c48..f2127673 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -15,7 +15,6 @@ package server
 
 import (
 	"archive/tar"
-	"bufio"
 	"bytes"
 	"crypto/sha256"
 	"encoding/binary"
@@ -123,6 +122,7 @@ func (fci *FileConsumerInfo) UnmarshalJSON(b []byte) error {
 type fileStore struct {
 	mu       sync.RWMutex
 	state    StreamState
+	ld       *LostStreamData
 	scb      StorageUpdateHandler
 	ageChk   *time.Timer
 	syncTmr  *time.Timer
@@ -135,22 +135,26 @@ type fileStore struct {
 	cfs      []*consumerFileStore
 	closed   bool
 	expiring bool
+	fip      bool
 	sips     int
 }
 
 // Represents a message store block and its data.
 type msgBlock struct {
-	mu      sync.RWMutex
-	mfn     string
-	mfd     *os.File
-	ifn     string
-	ifd     *os.File
-	liwsz   int64
-	index   uint64
-	bytes   uint64
-	msgs    uint64
-	first   msgId
-	last    msgId
+	mu    sync.RWMutex
+	fs    *fileStore
+	mfn   string
+	mfd   *os.File
+	ifn   string
+	ifd   *os.File
+	liwsz int64
+	index uint64
+	bytes uint64
+	msgs  uint64
+	first msgId
+	last  msgId
+	lwits int64
+
 	lwts    int64
 	llts    int64
 	lrts    int64
@@ -159,6 +163,7 @@ type msgBlock struct {
 	cloads  uint64
 	cexp    time.Duration
 	ctmr    *time.Timer
+	werr    error
 	loading bool
 	flusher bool
 	dmap    map[uint64]struct{}
@@ -225,7 +230,7 @@ const (
 	// default sync interval
 	defaultSyncInterval = 10 * time.Second
 	// coalesceMinimum
-	coalesceMinimum = 4 * 1024
+	coalesceMinimum = 16 * 1024
 	// maxFlushWait is maximum we will wait to gather messages to flush.
 	maxFlushWait = 8 * time.Millisecond
 	// Metafiles for streams and consumers.
@@ -293,6 +298,11 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		qch:  make(chan struct{}),
 	}
 
+	// Although originally designed to be able to flush multiple writes to disk
+	// for now we will suppress this behavior and at least flush in place and call
+	// into the OS. Sync is still handled async.
+	fs.fip = true
+
 	// Check if this is a new setup.
 	mdir := path.Join(fcfg.StoreDir, msgDir)
 	odir := path.Join(fcfg.StoreDir, consumerDir)
@@ -400,7 +410,7 @@ func (fs *fileStore) writeStreamMeta() error {
 	if _, err := os.Stat(meta); err != nil && !os.IsNotExist(err) {
 		return err
 	}
-	b, err := json.MarshalIndent(fs.cfg, _EMPTY_, "  ")
+	b, err := json.Marshal(fs.cfg)
 	if err != nil {
 		return err
 	}
@@ -424,9 +434,7 @@ const checksumSize = 8
 const indexHdrSize = 7*binary.MaxVarintLen64 + hdrLen + checksumSize
 
 func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint64) *msgBlock {
-	var le = binary.LittleEndian
-
-	mb := &msgBlock{index: index, cexp: fs.fcfg.CacheExpire}
+	mb := &msgBlock{fs: fs, index: index, cexp: fs.fcfg.CacheExpire}
 
 	mdir := path.Join(fs.fcfg.StoreDir, msgDir)
 	mb.mfn = path.Join(mdir, fi.Name())
@@ -455,10 +463,71 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint64) *msgBlock {
 			fs.blks = append(fs.blks, mb)
 			return mb
 		}
-		// Fall back on the data file itself. We will keep the delete map if present.
-		mb.msgs = 0
-		mb.bytes = 0
-		mb.first.seq = 0
+	}
+
+	// Close here since we need to rebuild state.
+	file.Close()
+
+	// If we get an error rebuilding the message block state record that with the fs itself.
+	if ld, err := mb.rebuildState(); err != nil && ld != nil {
+		fs.rebuildState(ld)
+	}
+	// Rewrite this to make sure we are sync'd.
+	mb.writeIndexInfo()
+	fs.blks = append(fs.blks, mb)
+	fs.lmb = mb
+	return mb
+}
+
+func (fs *fileStore) lostData() *LostStreamData {
+	fs.mu.RLock()
+	defer fs.mu.RUnlock()
+	if fs.ld == nil {
+		return nil
+	}
+	nld := *fs.ld
+	return &nld
+}
+
+func (fs *fileStore) rebuildState(ld *LostStreamData) {
+	if fs.ld != nil {
+		fs.ld.Msgs = append(fs.ld.Msgs, ld.Msgs...)
+		msgs := fs.ld.Msgs
+		sort.Slice(msgs, func(i, j int) bool { return msgs[i] < msgs[j] })
+		fs.ld.Bytes += ld.Bytes
+	} else {
+		fs.ld = ld
+	}
+	fs.state.Msgs, fs.state.Bytes = 0, 0
+	fs.state.FirstSeq, fs.state.LastSeq = 0, 0
+
+	for _, mb := range fs.blks {
+		mb.mu.RLock()
+		fs.state.Msgs += mb.msgs
+		fs.state.Bytes += mb.bytes
+		if fs.state.FirstSeq == 0 || mb.first.seq < fs.state.FirstSeq {
+			fs.state.FirstSeq = mb.first.seq
+			fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
+		}
+		fs.state.LastSeq = mb.last.seq
+		fs.state.LastTime = time.Unix(0, mb.last.ts).UTC()
+		mb.mu.RUnlock()
+	}
+}
+
+func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
+	mb.mu.Lock()
+	defer mb.mu.Unlock()
+
+	startLastSeq := mb.last.seq
+
+	// Clear state we need to rebuild.
+	mb.msgs, mb.bytes = 0, 0
+	mb.last.seq, mb.last.ts = 0, 0
+
+	buf, err := ioutil.ReadFile(mb.mfn)
+	if err != nil {
+		return nil, err
 	}
 
 	addToDmap := func(seq uint64) {
@@ -471,47 +540,103 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint64) *msgBlock {
 		mb.dmap[seq] = struct{}{}
 	}
 
-	// Use data file itself to rebuild.
-	var hdr [msgHdrSize]byte
-	var offset int64
+	var le = binary.LittleEndian
 
-	for {
-		if _, err := file.ReadAt(hdr[:], offset); err != nil {
-			// FIXME(dlc) - If this is not EOF we probably should try to fix.
-			break
+	truncate := func(index uint32) {
+		var fd *os.File
+		if mb.mfd != nil {
+			fd = mb.mfd
+		} else {
+			fd, _ := os.Open(mb.mfn)
+			defer fd.Close()
+		}
+		if fd != nil {
+			fd.Truncate(int64(index))
+			fd.Sync()
+			// Update our checksum.
+			if index >= 8 {
+				var lchk [8]byte
+				mb.mfd.ReadAt(lchk[:], int64(index-8))
+				copy(mb.lchk[0:], lchk[:])
+			}
+		}
+	}
+
+	gatherLost := func(lb uint32) *LostStreamData {
+		var ld LostStreamData
+		for seq := mb.last.seq + 1; seq <= startLastSeq; seq++ {
+			ld.Msgs = append(ld.Msgs, seq)
 		}
+		ld.Bytes = uint64(lb)
+		return &ld
+	}
+
+	for index, lbuf := uint32(0), uint32(len(buf)); index < lbuf; {
+		hdr := buf[index : index+msgHdrSize]
 		rl := le.Uint32(hdr[0:])
-		seq := le.Uint64(hdr[4:])
+		slen := le.Uint16(hdr[20:])
 
-		// Can't recover with zero record length.
-		if rl == 0 {
-			return nil
+		// Clear any headers bit that could be set.
+		rl &^= hbit
+		dlen := int(rl) - msgHdrSize
+		// Do some quick sanity checks here.
+		if dlen < 0 || int(slen) > dlen || dlen > int(rl) {
+			truncate(index)
+			return gatherLost(lbuf - index), errBadMsg
 		}
 
+		if index+rl > lbuf {
+			truncate(index)
+			return gatherLost(lbuf - index), errBadMsg
+		}
+
+		seq := le.Uint64(hdr[4:])
+		ts := int64(le.Uint64(hdr[12:]))
+
 		// This is an old erased message, or a new one that we can track.
-		if seq == 0 || seq&ebit != 0 {
+		if seq == 0 || seq&ebit != 0 || seq < mb.first.seq {
 			seq = seq &^ ebit
 			addToDmap(seq)
-			offset += int64(rl)
+			index += rl
 			continue
 		}
-		ts := int64(le.Uint64(hdr[12:]))
-		if mb.first.seq == 0 {
-			mb.first.seq = seq
-			mb.first.ts = ts
+
+		var deleted bool
+		if mb.dmap != nil {
+			if _, ok := mb.dmap[seq]; ok {
+				deleted = true
+			}
+		}
+
+		if !deleted {
+			if hh := mb.hh; hh != nil {
+				data := buf[index+msgHdrSize : index+rl]
+				hh.Reset()
+				hh.Write(hdr[4:20])
+				hh.Write(data[:slen])
+				hh.Write(data[slen : dlen-8])
+				checksum := hh.Sum(nil)
+				if !bytes.Equal(checksum, data[len(data)-8:]) {
+					truncate(index)
+					return gatherLost(lbuf - index), errBadMsg
+				}
+			}
+
+			if mb.first.seq == 0 {
+				mb.first.seq = seq
+				mb.first.ts = ts
+			}
+			mb.last.seq = seq
+			mb.last.ts = ts
+
+			mb.msgs++
+			mb.bytes += uint64(rl)
 		}
-		mb.last.seq = seq
-		mb.last.ts = ts
 
-		mb.msgs++
-		mb.bytes += uint64(rl)
-		offset += int64(rl)
+		index += rl
 	}
-	// Rewrite this to make sure we are sync'd.
-	mb.writeIndexInfo()
-	fs.blks = append(fs.blks, mb)
-	fs.lmb = mb
-	return mb
+
+	return nil, nil
 }
 
 func (fs *fileStore) recoverMsgs() error {
@@ -636,7 +761,7 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 		index = fs.lmb.index + 1
 	}
 
-	mb := &msgBlock{index: index, cexp: fs.fcfg.CacheExpire}
+	mb := &msgBlock{fs: fs, index: index, cexp: fs.fcfg.CacheExpire}
 
 	// Now do local hash.
 	key := sha256.Sum256(fs.hashKeyForBlock(index))
@@ -671,8 +796,10 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 	mb.first.seq = fs.state.LastSeq + 1
 	mb.last.seq = fs.state.LastSeq
 
-	// We know we will need this so go ahead and spin up.
-	mb.spinUpFlushLoop()
+	// If we know we will need this so go ahead and spin up.
+	if !fs.fip {
+		mb.spinUpFlushLoop()
+	}
 
 	// Add to our list of blocks and mark as last.
 	fs.blks = append(fs.blks, mb)
@@ -697,8 +824,10 @@ func (fs *fileStore) enableLastMsgBlockForWriting() error {
 	}
 	mb.mfd = mfd
 
-	// Spin up our flusher loop.
-	mb.spinUpFlushLoop()
+	// Spin up our flusher loop if needed.
+	if !fs.fip {
+		mb.spinUpFlushLoop()
+	}
 
 	return nil
 }
@@ -806,7 +935,6 @@ func (mb *msgBlock) skipMsg(seq uint64, now time.Time) {
 		mb.last.ts = now.UnixNano()
 		mb.first.seq = seq + 1
 		mb.first.ts = now.UnixNano()
-
 	} else {
 		needsRecord = true
 		if mb.dmap == nil {
@@ -819,7 +947,7 @@ func (mb *msgBlock) skipMsg(seq uint64, now time.Time) {
 	mb.mu.Unlock()
 
 	if needsRecord {
-		mb.writeMsgRecord(emptyRecordLen, seq|ebit, _EMPTY_, nil, nil, now.UnixNano())
+		mb.writeMsgRecord(emptyRecordLen, seq|ebit, _EMPTY_, nil, nil, now.UnixNano(), true)
 	} else {
 		mb.kickFlusher()
 	}
@@ -1016,7 +1144,7 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 
 	// Kick outside of lock.
 	if shouldWriteIndex {
-		if qch == nil {
+		if qch == nil && !fs.fip {
 			mb.spinUpFlushLoop()
 		}
 		select {
@@ -1507,43 +1635,6 @@ func (fs *fileStore) expireMsgs() {
 	}
 }
 
-// Check all the checksums for a message block.
-func checkMsgBlockFile(fp *os.File, hh hash.Hash) []uint64 {
-	var le = binary.LittleEndian
-	var hdr [msgHdrSize]byte
-	var bad []uint64
-
-	r := bufio.NewReaderSize(fp, 64*1024*1024)
-
-	for {
-		if _, err := io.ReadFull(r, hdr[0:]); err != nil {
-			break
-		}
-		rl := le.Uint32(hdr[0:])
-		seq := le.Uint64(hdr[4:])
-		slen := le.Uint16(hdr[20:])
-		dlen := int(rl) - msgHdrSize
-		if dlen < 0 || int(slen) > dlen || dlen > int(rl) {
-			bad = append(bad, seq)
-			break
-		}
-		data := make([]byte, dlen)
-		if _, err := io.ReadFull(r, data); err != nil {
-			bad = append(bad, seq)
-			break
-		}
-		hh.Reset()
-		hh.Write(hdr[4:20])
-		hh.Write(data[:slen])
-		hh.Write(data[slen : dlen-8])
-		checksum := hh.Sum(nil)
-		if !bytes.Equal(checksum, data[len(data)-8:]) {
-			bad = append(bad, seq)
-		}
-	}
-	return bad
-}
-
 // Lock should be held.
 func (fs *fileStore) checkAndFlushAllBlocks() {
 	for _, mb := range fs.blks {
@@ -1555,53 +1646,44 @@ func (fs *fileStore) checkAndFlushAllBlocks() {
 }
 
 // This will check all the checksums on messages and report back any sequence numbers with errors.
-func (fs *fileStore) checkMsgs() []uint64 {
+func (fs *fileStore) checkMsgs() *LostStreamData {
 	fs.mu.Lock()
 	defer fs.mu.Unlock()
 
 	fs.checkAndFlushAllBlocks()
 
-	mdir := path.Join(fs.fcfg.StoreDir, msgDir)
-	fis, err := ioutil.ReadDir(mdir)
-	if err != nil {
-		return nil
-	}
-
-	var bad []uint64
-
-	// Check all of the msg blocks.
-	for _, fi := range fis {
-		var index uint64
-		if n, err := fmt.Sscanf(fi.Name(), blkScan, &index); err == nil && n == 1 {
-			if fp, err := os.Open(path.Join(mdir, fi.Name())); err != nil {
-				continue
-			} else {
-				key := sha256.Sum256(fs.hashKeyForBlock(index))
-				hh, _ := highwayhash.New64(key[:])
-				bad = append(bad, checkMsgBlockFile(fp, hh)...)
-				fp.Close()
-			}
+	for _, mb := range fs.blks {
+		if ld, err := mb.rebuildState(); err != nil && ld != nil {
+			// Rebuild fs state too.
+			mb.fs.rebuildState(ld)
 		}
 	}
-	return bad
+	return fs.ld
 }
 
 // Will write the message record to the underlying message block.
 // filestore lock will be held.
-func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte, ts int64) {
+func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte, ts int64, flush bool) error {
 	mb.mu.Lock()
 
 	// Make sure we have a cache setup.
 	if mb.cache == nil {
 		mb.cache = &cache{}
+		// Make sure we set the proper cache offset if we have existing data.
+		var fi os.FileInfo
+		if mb.mfd != nil {
+			fi, _ = mb.mfd.Stat()
+		} else {
+			fi, _ = os.Stat(mb.mfn)
+		}
+		if fi != nil {
+			mb.cache.off = int(fi.Size())
+		}
 		mb.startCacheExpireTimer()
 	}
 
 	// Indexing
-	index := len(mb.cache.buf)
-	if mb.cache.off > 0 {
-		index += int(mb.cache.off)
-	}
+	index := len(mb.cache.buf) + int(mb.cache.off)
 
 	// Formats
 	// Format with no header
@@ -1648,7 +1730,7 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
 	// Grab last checksum
 	copy(mb.lchk[0:], checksum)
 
-	// Update writethrough cache.
+	// Update write through cache.
 	// Write to msg record.
 	mb.cache.buf = append(mb.cache.buf, checksum...)
 	// Write index
@@ -1660,15 +1742,30 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
 
 	// Set cache timestamp for last store.
 	mb.lwts = ts
+	// Decide if we write index info if flushing in place.
+	writeIndex := ts-mb.lwits > int64(time.Second)
 
 	// Accounting
 	mb.updateAccounting(seq, ts, rl)
 
-	fch := mb.fch
+	fch, werr := mb.fch, mb.werr
 	mb.mu.Unlock()
 
-	// Flush the kicker here.
-	kickFlusher(fch)
+	// If we should be flushing in place do so here. We will also flip to flushing in place if we
+	// had a write error.
+	if flush || werr != nil {
+		if err := mb.flushPendingMsgs(); err != nil && err != errFlushRunning && err != errNoPending {
+			return err
+		}
+		if writeIndex {
+			mb.writeIndexInfo()
+		}
+	} else {
+		// Kick the flusher here.
+		kickFlusher(fch)
+	}
+
+	return nil
 }
 
 // How many bytes pending to be written for this message block.
@@ -1759,9 +1856,9 @@ func (fs *fileStore) writeMsgRecord(seq uint64, ts int64, subj string, hdr, msg
 	}
 
 	// Ask msg block to store in write through cache.
-	mb.writeMsgRecord(rl, seq, subj, hdr, msg, ts)
+	err = mb.writeMsgRecord(rl, seq, subj, hdr, msg, ts, fs.fip)
 
-	return rl, nil
+	return rl, err
 }
 
 // Sync msg and index files as needed. This is called from a timer.
@@ -1956,17 +2053,21 @@ func (mb *msgBlock) flushPendingMsgs() error {
 
 	var tn int
 
+	var n int
+
 	// Append new data to the message block file.
 	for lbb := lob; lbb > 0; lbb = len(buf) {
-		n, err := mfd.WriteAt(buf, woff)
+		n, err = mfd.WriteAt(buf, woff)
 		if err != nil {
-			// FIXME(dlc) - What is the correct behavior here?
 			mb.removeIndexFile()
-			mb.mu.Lock()
-			mb.clearFlushing()
-			mb.mu.Unlock()
-			return err
+			mb.dirtyClose()
+			if ld, err := mb.rebuildState(); err != nil && ld != nil {
+				// Rebuild fs state too.
+				mb.fs.rebuildState(ld)
+			}
+			break
 		}
+
 		woff += int64(n)
 		tn += n
 
@@ -1984,10 +2085,12 @@ func (mb *msgBlock) flushPendingMsgs() error {
 	defer mb.mu.Unlock()
 	// Clear on exit.
 	defer mb.clearFlushing()
+	// set write err to any error.
+	mb.werr = err
 
 	// Cache may be gone.
 	if mb.cache == nil || mb.mfd == nil {
-		return nil
+		return mb.werr
 	}
 
 	// Check for additional writes while we were writing to the disk.
@@ -2019,7 +2122,7 @@ func (mb *msgBlock) flushPendingMsgs() error {
 		mb.cache.buf = buf
 	}
 
-	return nil
+	return mb.werr
 }
 
 //  Lock should be held.
@@ -2054,9 +2157,11 @@ checkCache:
 	// FIXME(dlc) - We could be smarter here.
 	if mb.cache != nil && len(mb.cache.buf)-mb.cache.wp > 0 {
 		mb.mu.Unlock()
-		mb.flushPendingMsgsAndWait()
-		mb.writeIndexInfo()
+		err := mb.flushPendingMsgsAndWait()
 		mb.mu.Lock()
+		if err != nil && err != errFlushRunning {
+			return err
+		}
 		goto checkCache
 	}
 
@@ -2300,6 +2405,9 @@ func (fs *fileStore) State() StreamState {
 		mb.mu.Unlock()
 	}
 	fs.mu.RUnlock()
+
+	state.Lost = fs.lostData()
+
 	// Can not be guaranteed to be sorted.
 	if len(state.Deleted) > 0 {
 		sort.Slice(state.Deleted, func(i, j int) bool {
@@ -2359,8 +2467,13 @@ func (mb *msgBlock) writeIndexInfo() error {
 		mb.ifd = ifd
 	}
 
+	mb.lwits = time.Now().UnixNano()
+
 	if n, err = mb.ifd.WriteAt(buf, 0); err == nil {
 		mb.liwsz = int64(n)
+		mb.werr = nil
+	} else {
+		mb.werr = err
 	}
 
 	return err
@@ -3021,9 +3134,9 @@ func (fs *fileStore) Snapshot(deadline time.Duration, checkMsgs, includeConsumer
 	state := fs.State()
 
 	if checkMsgs {
-		bad := fs.checkMsgs()
-		if len(bad) > 0 {
-			return nil, fmt.Errorf("snapshot check detected %d bad messages", len(bad))
+		ld := fs.checkMsgs()
+		if ld != nil && len(ld.Msgs) > 0 {
+			return nil, fmt.Errorf("snapshot check detected %d bad messages", len(ld.Msgs))
 		}
 	}
 
@@ -3445,7 +3558,7 @@ func (cfs *consumerFileStore) writeConsumerMeta() error {
 	if _, err := os.Stat(meta); (err != nil && !os.IsNotExist(err)) || err == nil {
 		return err
 	}
-	b, err := json.MarshalIndent(cfs.cfg, _EMPTY_, "  ")
+	b, err := json.Marshal(cfs.cfg)
 	if err != nil {
 		return err
 	}
@@ -3782,7 +3895,7 @@ func (ts *templateFileStore) Store(t *StreamTemplate) error {
 		return err
 	}
 	t.mu.Lock()
-	b, err := json.MarshalIndent(t, _EMPTY_, "  ")
+	b, err := json.Marshal(t)
 	t.mu.Unlock()
 	if err != nil {
 		return err
diff --git a/server/gateway.go b/server/gateway.go
index d661aaa3..59f8c146 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -833,6 +833,13 @@ func (s *Server) createGateway(cfg *gatewayCfg, url *url.URL, conn net.Conn) {
 	s.setFirstPingTimer(c)
 
 	c.mu.Unlock()
+
+	// Announce ourselves again to new connections.
+	if solicit && s.EventsEnabled() {
+		s.mu.Lock()
+		s.sendStatsz(fmt.Sprintf(serverStatsSubj, s.info.ID))
+		s.mu.Unlock()
+	}
 }
 
 // Builds and sends the CONNECT protocol for a gateway.
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index ff01d073..821d7c2d 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -709,7 +709,7 @@ func (s *Server) jsAccountInfoRequest(sub *subscription, c *client, subject, rep
 		stats := acc.JetStreamUsage()
 		resp.JetStreamAccountStats = &stats
 	}
-	b, err := json.MarshalIndent(resp, "", "  ")
+	b, err := json.Marshal(resp)
 	if err != nil {
 		return
 	}
@@ -923,7 +923,7 @@ func (s *Server) jsTemplateDeleteRequest(sub *subscription, c *client, subject,
 }
 
 func (s *Server) jsonResponse(v interface{}) string {
-	b, err := json.MarshalIndent(v, "", "  ")
+	b, err := json.Marshal(v)
 	if err != nil {
 		s.Warnf("Problem marshaling JSON for JetStream API:", err)
 		return ""
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index e45d8219..99bb4eb3 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -294,6 +294,9 @@ func (s *Server) JetStreamClusterPeers() []string {
 	peers := cc.meta.Peers()
 	var nodes []string
 	for _, p := range peers {
+		if si, ok := s.nodeToInfo.Load(p.ID); !ok || si.(*nodeInfo).offline {
+			continue
+		}
 		nodes = append(nodes, p.ID)
 	}
 	return nodes
@@ -470,9 +473,11 @@ func (js *jetStream) setupMetaGroup() error {
 	if bootstrap {
 		s.Noticef("JetStream cluster bootstrapping")
 		// FIXME(dlc) - Make this real.
-		peers := s.activePeers()
+		peers := s.ActivePeers()
 		s.Debugf("JetStream cluster initial peers: %+v", peers)
-		s.bootstrapRaftNode(cfg, peers, false)
+		if err := s.bootstrapRaftNode(cfg, peers, false); err != nil {
+			return err
+		}
 	} else {
 		s.Noticef("JetStream cluster recovering state")
 	}
@@ -702,11 +707,16 @@ func (js *jetStream) monitorCluster() {
 				continue
 			}
 			// FIXME(dlc) - Deal with errors.
-			if hadSnapshot, err := js.applyMetaEntries(ce.Entries, isRecovering); err == nil {
+			if hadSnapshot, didRemoval, err := js.applyMetaEntries(ce.Entries, isRecovering); err == nil {
 				n.Applied(ce.Index)
 				if hadSnapshot {
 					snapout = false
-					n.Compact(ce.Index)
+					if !isRecovering {
+						n.Compact(ce.Index)
+					}
+				}
+				if didRemoval {
+					attemptSnapshot()
 				}
 			}
 			if isLeader && !snapout {
@@ -933,8 +943,8 @@ func (js *jetStream) removePeerFromStream(sa *streamAssignment, peer string) {
 	}
 }
 
-func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool) (bool, error) {
-	var didSnap bool
+func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool) (bool, bool, error) {
+	var didSnap, didRemove bool
 	for _, e := range entries {
 		if e.Type == EntrySnapshot {
 			js.applyMetaSnapshot(e.Data, isRecovering)
@@ -948,7 +958,7 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool) (bool
 				sa, err := decodeStreamAssignment(buf[1:])
 				if err != nil {
 					js.srv.Errorf("JetStream cluster failed to decode stream assignment: %q", buf[1:])
-					return didSnap, err
+					return didSnap, didRemove, err
 				}
 				if isRecovering {
 					js.setStreamAssignmentResponded(sa)
@@ -958,17 +968,18 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool) (bool
 				sa, err := decodeStreamAssignment(buf[1:])
 				if err != nil {
 					js.srv.Errorf("JetStream cluster failed to decode stream assignment: %q", buf[1:])
-					return didSnap, err
+					return didSnap, didRemove, err
 				}
 				if isRecovering {
 					js.setStreamAssignmentResponded(sa)
 				}
 				js.processStreamRemoval(sa)
+				didRemove = true
 			case assignConsumerOp:
 				ca, err := decodeConsumerAssignment(buf[1:])
 				if err != nil {
 					js.srv.Errorf("JetStream cluster failed to decode consumer assigment: %q", buf[1:])
-					return didSnap, err
+					return didSnap, didRemove, err
 				}
 				if isRecovering {
 					js.setConsumerAssignmentResponded(ca)
@@ -978,7 +989,7 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool) (bool
 				ca, err := decodeConsumerAssignmentCompressed(buf[1:])
 				if err != nil {
 					js.srv.Errorf("JetStream cluster failed to decode compressed consumer assigment: %q", buf[1:])
-					return didSnap, err
+					return didSnap, didRemove, err
 				}
 				if isRecovering {
 					js.setConsumerAssignmentResponded(ca)
@@ -988,18 +999,19 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, isRecovering bool) (bool
 				ca, err := decodeConsumerAssignment(buf[1:])
 				if err != nil {
 					js.srv.Errorf("JetStream cluster failed to decode consumer assigment: %q", buf[1:])
-					return didSnap, err
+					return didSnap, didRemove, err
 				}
 				if isRecovering {
 					js.setConsumerAssignmentResponded(ca)
 				}
 				js.processConsumerRemoval(ca)
+				didRemove = true
 			default:
 				panic("JetStream Cluster Unknown meta entry op type")
 			}
 		}
 	}
-	return didSnap, nil
+	return didSnap, didRemove, nil
 }
 
 func (rg *raftGroup) isMember(id string) bool {
@@ -1058,7 +1070,7 @@ func (js *jetStream) createRaftGroup(rg *raftGroup) error {
 	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},
+		StreamConfig{Name: rg.Name, Storage: FileStorage, Replicas: 2},
 	)
 	if err != nil {
 		s.Errorf("Error creating filestore: %v", err)
@@ -1283,11 +1295,14 @@ func (js *jetStream) monitorStream(mset *Stream, sa *streamAssignment) {
 				}
 			}
 			// Apply our entries.
-			if hadSnapshot, err := js.applyStreamEntries(mset, ce, isRecovering); err == nil {
+			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)
 			}
@@ -1314,8 +1329,8 @@ func (js *jetStream) monitorStream(mset *Stream, sa *streamAssignment) {
 	}
 }
 
-func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry, isRecovering bool) (bool, error) {
-	var didSnap bool
+func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry, isRecovering bool) (bool, bool, error) {
+	var didSnap, didRemove bool
 	for _, e := range ce.Entries {
 		if e.Type == EntrySnapshot {
 			if !isRecovering {
@@ -1329,7 +1344,7 @@ func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry, isReco
 			if peer := string(e.Data); peer == ourID {
 				mset.stop(true, false)
 			}
-			return false, nil
+			return false, false, nil
 		} else {
 			buf := e.Data
 			switch entryOp(buf[0]) {
@@ -1348,6 +1363,12 @@ func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry, isReco
 					if err != errLastSeqMismatch || !isRecovering {
 						js.srv.Debugf("Got error processing JetStream msg: %v", err)
 					}
+					if strings.Contains(err.Error(), "no space left") {
+						s := js.srv
+						s.Errorf("JetStream out of space, will be DISABLED")
+						js.srv.RemoveJetStream()
+						return didSnap, didRemove, err
+					}
 				}
 			case deleteMsgOp:
 				md, err := decodeMsgDelete(buf[1:])
@@ -1365,6 +1386,7 @@ 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)
@@ -1402,6 +1424,7 @@ 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)
@@ -1423,7 +1446,7 @@ func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry, isReco
 			}
 		}
 	}
-	return didSnap, nil
+	return didSnap, didRemove, nil
 }
 
 // Returns the PeerInfo for all replicas of a raft node. This is different than node.Peers()
@@ -1432,8 +1455,11 @@ func (s *Server) replicas(node RaftNode) []*PeerInfo {
 	now := time.Now()
 	var replicas []*PeerInfo
 	for _, rp := range node.Peers() {
-		pi := &PeerInfo{Name: s.serverNameForNode(rp.ID), Current: rp.Current, Active: now.Sub(rp.Last)}
-		replicas = append(replicas, pi)
+		if sir, ok := s.nodeToInfo.Load(rp.ID); ok && sir != nil {
+			si := sir.(*nodeInfo)
+			pi := &PeerInfo{Name: si.name, Current: rp.Current, Active: now.Sub(rp.Last), Offline: si.offline, Lag: rp.Lag}
+			replicas = append(replicas, pi)
+		}
 	}
 	return replicas
 }
@@ -1961,7 +1987,7 @@ func (js *jetStream) processConsumerAssignment(ca *consumerAssignment) {
 func (js *jetStream) processConsumerRemoval(ca *consumerAssignment) {
 	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
@@ -2597,6 +2623,10 @@ func (cc *jetStreamCluster) remapStreamAssignment(sa *streamAssignment, removePe
 		cluster = sa.Config.Placement.Cluster
 	}
 	for _, p := range cc.meta.Peers() {
+		// If it is not in our list it probably shutdown, so don't consider.
+		if si, ok := s.nodeToInfo.Load(p.ID); !ok || si.(*nodeInfo).offline {
+			continue
+		}
 		// Make sure they are active and current and not already part of our group.
 		current, lastSeen := p.Current, now.Sub(p.Last)
 		if !current || lastSeen > lostQuorumInterval || sa.Group.isMember(p.ID) {
@@ -2624,23 +2654,19 @@ func (cc *jetStreamCluster) remapStreamAssignment(sa *streamAssignment, removePe
 func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string) []string {
 	var nodes []string
 	peers := cc.meta.Peers()
-	s, ourID := cc.s, cc.meta.ID()
+	s := cc.s
 
-	now := time.Now()
 	for _, p := range peers {
-		// Make sure they are active and current.
-		current, lastSeen := p.Current, now.Sub(p.Last)
-		if current && lastSeen > lostQuorumInterval {
-			current = false
+		// If we know its offline or it is not in our list it probably shutdown, so don't consider.
+		if si, ok := s.nodeToInfo.Load(p.ID); !ok || si.(*nodeInfo).offline {
+			continue
 		}
-		if p.ID == ourID || current {
-			if cluster != _EMPTY_ {
-				if s.clusterNameForNode(p.ID) == cluster {
-					nodes = append(nodes, p.ID)
-				}
-			} else {
+		if cluster != _EMPTY_ {
+			if s.clusterNameForNode(p.ID) == cluster {
 				nodes = append(nodes, p.ID)
 			}
+		} else {
+			nodes = append(nodes, p.ID)
 		}
 	}
 	if len(nodes) < r {
@@ -3733,8 +3759,11 @@ func (js *jetStream) clusterInfo(rg *raftGroup) *ClusterInfo {
 			if current && lastSeen > lostQuorumInterval {
 				current = false
 			}
-			pi := &PeerInfo{Name: s.serverNameForNode(rp.ID), Current: current, Active: lastSeen}
-			ci.Replicas = append(ci.Replicas, pi)
+			if sir, ok := s.nodeToInfo.Load(rp.ID); ok && sir != nil {
+				si := sir.(*nodeInfo)
+				pi := &PeerInfo{Name: si.name, Current: current, Offline: si.offline, Active: lastSeen, Lag: rp.Lag}
+				ci.Replicas = append(ci.Replicas, pi)
+			}
 		}
 	}
 	return ci
diff --git a/server/jetstream_events.go b/server/jetstream_events.go
index 9f69e1be..8973c88f 100644
--- a/server/jetstream_events.go
+++ b/server/jetstream_events.go
@@ -6,7 +6,7 @@ import (
 )
 
 func (s *Server) publishAdvisory(acc *Account, subject string, adv interface{}) {
-	ej, err := json.MarshalIndent(adv, "", "  ")
+	ej, err := json.Marshal(adv)
 	if err == nil {
 		err = s.sendInternalAccountMsg(acc, subject, ej)
 		if err != nil {
diff --git a/server/raft.go b/server/raft.go
index d6054443..ae523a51 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -63,6 +63,7 @@ type WAL interface {
 	LoadMsg(index uint64) (subj string, hdr, msg []byte, ts int64, err error)
 	RemoveMsg(index uint64) (bool, error)
 	Compact(index uint64) (uint64, error)
+	Truncate(seq uint64) error
 	State() StreamState
 	Stop() error
 	Delete() error
@@ -77,7 +78,7 @@ type Peer struct {
 	ID      string
 	Current bool
 	Last    time.Time
-	Index   uint64
+	Lag     uint64
 }
 
 type RaftState uint8
@@ -185,7 +186,7 @@ type lps struct {
 }
 
 const (
-	minElectionTimeout = 300 * time.Millisecond
+	minElectionTimeout = 500 * time.Millisecond
 	maxElectionTimeout = 3 * minElectionTimeout
 	minCampaignTimeout = 50 * time.Millisecond
 	maxCampaignTimeout = 4 * minCampaignTimeout
@@ -379,16 +380,16 @@ func (s *Server) startRaftNode(cfg *RaftConfig) (RaftNode, error) {
 
 // Maps node names back to server names.
 func (s *Server) serverNameForNode(node string) string {
-	if sn, ok := s.nodeToName.Load(node); ok {
-		return sn.(string)
+	if si, ok := s.nodeToInfo.Load(node); ok && si != nil {
+		return si.(*nodeInfo).name
 	}
 	return _EMPTY_
 }
 
 // Maps node names back to cluster names.
 func (s *Server) clusterNameForNode(node string) string {
-	if cn, ok := s.nodeToCluster.Load(node); ok {
-		return cn.(string)
+	if si, ok := s.nodeToInfo.Load(node); ok && si != nil {
+		return si.(*nodeInfo).cluster
 	}
 	return _EMPTY_
 }
@@ -848,7 +849,12 @@ func (n *raft) Peers() []*Peer {
 
 	var peers []*Peer
 	for id, ps := range n.peers {
-		p := &Peer{ID: id, Current: id == n.leader || ps.li >= n.applied, Last: time.Unix(0, ps.ts)}
+		p := &Peer{
+			ID:      id,
+			Current: id == n.leader || ps.li >= n.applied,
+			Last:    time.Unix(0, ps.ts),
+			Lag:     n.commit - ps.li,
+		}
 		peers = append(peers, p)
 	}
 	return peers
@@ -1494,7 +1500,7 @@ func (n *raft) catchupFollower(ar *appendEntryResponse) {
 		return
 	}
 	if ae.pindex != ar.index || ae.pterm != ar.term {
-		n.debug("Our first entry does not match")
+		n.debug("Our first entry does not match request from follower")
 	}
 	// Create a chan for delivering updates from responses.
 	indexUpdates := make(chan uint64, 1024)
@@ -1792,9 +1798,11 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	// If we received an append entry as a candidate we should convert to a follower.
 	if n.state == Candidate {
 		n.debug("Received append entry in candidate state from %q, converting to follower", ae.leader)
-		n.term = ae.term
-		n.vote = noVote
-		n.writeTermVote()
+		if n.term < ae.term {
+			n.term = ae.term
+			n.vote = noVote
+			n.writeTermVote()
+		}
 		n.attemptStepDown(ae.leader)
 	}
 
@@ -1867,7 +1875,6 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	if isNew && n.leader != ae.leader && n.state == Follower {
 		n.debug("AppendEntry updating leader to %q", ae.leader)
 		n.leader = ae.leader
-		n.vote = noVote
 		n.writeTermVote()
 		if isNew {
 			n.resetElectionTimeout()
@@ -1875,25 +1882,51 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		}
 	}
 
-	// TODO(dlc) - Do both catchup and delete new behaviors from spec.
 	if ae.pterm != n.pterm || ae.pindex != n.pindex {
-		// Check if we are catching up and this is a snapshot, if so reset our wal's index.
-		// Snapshots will always be by themselves.
-		if catchingUp && len(ae.entries) > 0 && ae.entries[0].Type == EntrySnapshot {
-			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
+		// If this is a lower index that what we were expecting.
+		if ae.pindex < n.pindex {
+			var ar *appendEntryResponse
+			if eae, err := n.loadEntry(ae.pindex); err == nil && eae != nil {
+				// If terms mismatched, delete that entry and all others past it.
+				if eae.pterm != ae.pterm {
+					n.wal.Truncate(ae.pindex)
+					n.pindex = ae.pindex
+					n.pterm = ae.pterm
+					ar = &appendEntryResponse{n.pterm, n.pindex, n.id, false, _EMPTY_}
+				} else {
+					ar = &appendEntryResponse{ae.pterm, ae.pindex, n.id, true, _EMPTY_}
+				}
+			}
+			n.Unlock()
+			if ar != nil {
+				n.sendRPC(ae.reply, _EMPTY_, ar.encode())
+			}
+			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.
+		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
+			}
 		} else {
 			n.debug("AppendEntry did not match %d %d with %d %d", ae.pterm, ae.pindex, n.pterm, n.pindex)
 			// Reset our term.
 			n.term = n.pterm
-			// Setup our state for catching up.
-			inbox := n.createCatchup(ae)
-			ar := appendEntryResponse{n.pterm, n.pindex, n.id, false, _EMPTY_}
-			n.Unlock()
-			n.sendRPC(ae.reply, inbox, ar.encode())
-			return
+			if ae.pindex > n.pindex {
+				// Setup our state for catching up.
+				inbox := n.createCatchup(ae)
+				ar := appendEntryResponse{n.pterm, n.pindex, n.id, false, _EMPTY_}
+				n.Unlock()
+				n.sendRPC(ae.reply, inbox, ar.encode())
+				return
+			}
 		}
 	}
 
@@ -1966,9 +1999,14 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 func (n *raft) processPeerState(ps *peerState) {
 	// Update our version of peers to that of the leader.
 	n.csz = ps.clusterSize
+	old := n.peers
 	n.peers = make(map[string]*lps)
 	for _, peer := range ps.knownPeers {
-		n.peers[peer] = &lps{0, 0}
+		if lp := old[peer]; lp != nil {
+			n.peers[peer] = lp
+		} else {
+			n.peers[peer] = &lps{0, 0}
+		}
 	}
 	n.debug("Update peers from leader to %+v", n.peers)
 	writePeerState(n.sd, ps)
@@ -2003,6 +2041,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))
 	}
 
@@ -2257,6 +2297,7 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 	}
 
 	n.Lock()
+	n.resetElectionTimeout()
 
 	// Ignore if we are newer.
 	if vr.term < n.term {
@@ -2281,7 +2322,6 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 		vresp.granted = true
 		n.vote = vr.candidate
 		n.writeTermVote()
-		n.resetElectionTimeout()
 	}
 	n.Unlock()
 
@@ -2371,7 +2411,6 @@ func (n *raft) switchState(state RaftState) {
 	}
 
 	n.state = state
-	n.vote = noVote
 	n.writeTermVote()
 }
 
diff --git a/server/route.go b/server/route.go
index 92cda08d..2990ca4f 100644
--- a/server/route.go
+++ b/server/route.go
@@ -1416,8 +1416,7 @@ func (s *Server) addRoute(c *client, info *Info) (bool, bool) {
 	if !exists {
 		s.routes[c.cid] = c
 		s.remotes[id] = c
-		s.nodeToName.Store(c.route.hash, c.route.remoteName)
-		s.nodeToCluster.Store(c.route.hash, s.info.Cluster)
+		s.nodeToInfo.Store(c.route.hash, &nodeInfo{c.route.remoteName, s.info.Cluster, id, false})
 		c.mu.Lock()
 		c.route.connectURLs = info.ClientConnectURLs
 		c.route.wsConnURLs = info.WSConnectURLs
diff --git a/server/server.go b/server/server.go
index 144b5b2d..9128e1da 100644
--- a/server/server.go
+++ b/server/server.go
@@ -238,9 +238,15 @@ type Server struct {
 	rnMu      sync.RWMutex
 	raftNodes map[string]RaftNode
 
-	// For mapping from a node name back to a server name and cluster.
-	nodeToName    sync.Map
-	nodeToCluster sync.Map
+	// For mapping from a raft node name back to a server name and cluster.
+	nodeToInfo sync.Map
+}
+
+type nodeInfo struct {
+	name    string
+	cluster string
+	id      string
+	offline bool
 }
 
 // Make sure all are 64bits for atomic use
@@ -342,9 +348,8 @@ func NewServer(opts *Options) (*Server, error) {
 	defer s.mu.Unlock()
 
 	// Place ourselves in some lookup maps.
-	ourId := string(getHash(serverName))
-	s.nodeToName.Store(ourId, serverName)
-	s.nodeToCluster.Store(ourId, opts.Cluster.Name)
+	ourNode := string(getHash(serverName))
+	s.nodeToInfo.Store(ourNode, &nodeInfo{serverName, opts.Cluster.Name, info.ID, false})
 
 	s.routeResolver = opts.Cluster.resolver
 	if s.routeResolver == nil {
@@ -790,18 +795,14 @@ func (s *Server) configuredRoutes() int {
 }
 
 // activePeers is used in bootstrapping raft groups like the JetStream meta controller.
-func (s *Server) activePeers() (peers []string) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	if s.sys == nil {
-		return nil
-	}
-	s.nodeToName.Range(func(k, v interface{}) bool {
-		peers = append(peers, k.(string))
+func (s *Server) ActivePeers() (peers []string) {
+	s.nodeToInfo.Range(func(k, v interface{}) bool {
+		si := v.(*nodeInfo)
+		if !si.offline {
+			peers = append(peers, k.(string))
+		}
 		return true
 	})
-
 	return peers
 }
 
@@ -2676,20 +2677,21 @@ func (s *Server) supportsHeaders() bool {
 
 // ID returns the server's ID
 func (s *Server) ID() string {
-	s.mu.Lock()
-	defer s.mu.Unlock()
 	return s.info.ID
 }
 
+// NodeName returns the node name for this server.
+func (s *Server) NodeName() string {
+	return string(getHash(s.info.Name))
+}
+
 // Name returns the server's name. This will be the same as the ID if it was not set.
 func (s *Server) Name() string {
-	s.mu.Lock()
-	defer s.mu.Unlock()
 	return s.info.Name
 }
 
 func (s *Server) String() string {
-	return s.Name()
+	return s.info.Name
 }
 
 func (s *Server) startGoRoutine(f func()) bool {
diff --git a/server/store.go b/server/store.go
index b8662c39..980a46f3 100644
--- a/server/store.go
+++ b/server/store.go
@@ -110,14 +110,21 @@ const (
 
 // StreamState is information about the given stream.
 type StreamState struct {
-	Msgs      uint64    `json:"messages"`
-	Bytes     uint64    `json:"bytes"`
-	FirstSeq  uint64    `json:"first_seq"`
-	FirstTime time.Time `json:"first_ts"`
-	LastSeq   uint64    `json:"last_seq"`
-	LastTime  time.Time `json:"last_ts"`
-	Deleted   []uint64  `json:"deleted,omitempty"`
-	Consumers int       `json:"consumer_count"`
+	Msgs      uint64          `json:"messages"`
+	Bytes     uint64          `json:"bytes"`
+	FirstSeq  uint64          `json:"first_seq"`
+	FirstTime time.Time       `json:"first_ts"`
+	LastSeq   uint64          `json:"last_seq"`
+	LastTime  time.Time       `json:"last_ts"`
+	Deleted   []uint64        `json:"deleted,omitempty"`
+	Lost      *LostStreamData `json:"lost,omitempty"`
+	Consumers int             `json:"consumer_count"`
+}
+
+// LostStreamData indicates msgs that have been lost.
+type LostStreamData struct {
+	Msgs  []uint64 `json:"msgs"`
+	Bytes uint64   `json:"bytes"`
 }
 
 // SnapshotResult contains information about the snapshot.
diff --git a/server/stream.go b/server/stream.go
index 5544c153..e1afa678 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -97,7 +97,9 @@ type ClusterInfo struct {
 type PeerInfo struct {
 	Name    string        `json:"name"`
 	Current bool          `json:"current"`
+	Offline bool          `json:"offline,omitempty"`
 	Active  time.Duration `json:"active"`
+	Lag     uint64        `json:"lag,omitempty"`
 }
 
 // Stream is a jetstream stream of messages. When we receive a message internally destined
@@ -509,7 +511,7 @@ func (mset *Stream) sendCreateAdvisory() {
 		Template: template,
 	}
 
-	j, err := json.MarshalIndent(m, "", "  ")
+	j, err := json.Marshal(m)
 	if err != nil {
 		return
 	}
@@ -534,7 +536,7 @@ func (mset *Stream) sendDeleteAdvisoryLocked() {
 		Template: mset.config.Template,
 	}
 
-	j, err := json.MarshalIndent(m, "", "  ")
+	j, err := json.Marshal(m)
 	if err == nil {
 		subj := JSAdvisoryStreamDeletedPre + "." + mset.config.Name
 		mset.sendq <- &jsPubMsg{subj, subj, _EMPTY_, nil, j, nil, 0}
@@ -556,7 +558,7 @@ func (mset *Stream) sendUpdateAdvisoryLocked() {
 		Action: ModifyEvent,
 	}
 
-	j, err := json.MarshalIndent(m, "", "  ")
+	j, err := json.Marshal(m)
 	if err == nil {
 		subj := JSAdvisoryStreamUpdatedPre + "." + mset.config.Name
 		mset.sendq <- &jsPubMsg{subj, subj, _EMPTY_, nil, j, nil, 0}
@@ -1314,11 +1316,11 @@ func (mset *Stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 
 	if err != nil {
 		if err != ErrStoreClosed {
-			c.Errorf("JetStream failed to store a msg on account: %q stream: %q -  %v", accName, name, err)
+			c.Errorf("JetStream failed to store a msg on stream '%s > %s' -  %v", accName, name, err)
 		}
 		if canRespond {
 			resp.PubAck = &PubAck{Stream: name}
-			resp.Error = &ApiError{Code: 400, Description: err.Error()}
+			resp.Error = &ApiError{Code: 503, Description: err.Error()}
 			response, _ = json.Marshal(resp)
 		}
 	} else if jsa.limitsExceeded(stype) {
@@ -1366,7 +1368,7 @@ func (mset *Stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 		}
 	}
 
-	return nil
+	return err
 }
 
 // Internal message for use by jetstream subsystem.


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index a63852f1..8c6f2d2e 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -300,8 +300,8 @@ func TestSystemAccountNewConnection(t *testing.T) {
 	}
 	if !strings.HasPrefix(connsMsg.Subject, fmt.Sprintf("$SYS.ACCOUNT.%s.SERVER.CONNS", acc2.Name)) {
 		t.Fatalf("Expected subject to start with %q, got %q", "$SYS.ACCOUNT.<account>.CONNECT", msg.Subject)
-	} else if !strings.Contains(string(connsMsg.Data), `"total_conns": 0`) {
-		t.Fatalf("Expected event to reflect created connection, got: %s", string(msg.Data))
+	} else if !strings.Contains(string(connsMsg.Data), `"total_conns":0`) {
+		t.Fatalf("Expected event to reflect created connection, got: %s", string(connsMsg.Data))
 	}
 	conns = AccountNumConns{}
 	if err := json.Unmarshal(connsMsg.Data, &conns); err != nil {
@@ -1257,7 +1257,7 @@ func TestAccountReqMonitoring(t *testing.T) {
 	// query SUBSZ for account
 	if resp, err := ncSys.Request(subsz, nil, time.Second); err != nil {
 		t.Fatalf("Error on request: %v", err)
-	} else if !strings.Contains(string(resp.Data), `"num_subscriptions": 25,`) {
+	} else if !strings.Contains(string(resp.Data), `"num_subscriptions":25,`) {
 		t.Fatalf("unexpected subs count (expected 0): %v", string(resp.Data))
 	}
 	// create a subscription
@@ -1270,25 +1270,25 @@ func TestAccountReqMonitoring(t *testing.T) {
 	// query SUBSZ for account
 	if resp, err := ncSys.Request(subsz, nil, time.Second); err != nil {
 		t.Fatalf("Error on request: %v", err)
-	} else if !strings.Contains(string(resp.Data), `"num_subscriptions": 26,`) {
+	} else if !strings.Contains(string(resp.Data), `"num_subscriptions":26,`) {
 		t.Fatalf("unexpected subs count (expected 26): %v", string(resp.Data))
-	} else if !strings.Contains(string(resp.Data), `"subject": "foo"`) {
+	} else if !strings.Contains(string(resp.Data), `"subject":"foo"`) {
 		t.Fatalf("expected subscription foo: %v", string(resp.Data))
 	}
 	// query connections for account
 	if resp, err := ncSys.Request(connz, nil, time.Second); err != nil {
 		t.Fatalf("Error on request: %v", err)
-	} else if !strings.Contains(string(resp.Data), `"num_connections": 1,`) {
+	} else if !strings.Contains(string(resp.Data), `"num_connections":1,`) {
 		t.Fatalf("unexpected subs count (expected 1): %v", string(resp.Data))
-	} else if !strings.Contains(string(resp.Data), `"total": 2,`) { // includes system acc connection
+	} else if !strings.Contains(string(resp.Data), `"total":2,`) { // includes system acc connection
 		t.Fatalf("unexpected subs count (expected 1): %v", string(resp.Data))
 	}
 	// query connections for js account
 	if resp, err := ncSys.Request(jsz, nil, time.Second); err != nil {
 		t.Fatalf("Error on request: %v", err)
-	} else if !strings.Contains(string(resp.Data), `"memory": 0,`) {
+	} else if !strings.Contains(string(resp.Data), `"memory":0,`) {
 		t.Fatalf("jetstream should be enabled but empty: %v", string(resp.Data))
-	} else if !strings.Contains(string(resp.Data), `"storage": 0,`) {
+	} else if !strings.Contains(string(resp.Data), `"storage":0,`) {
 		t.Fatalf("jetstream should be enabled but empty: %v", string(resp.Data))
 	}
 }
@@ -1594,7 +1594,7 @@ func TestSystemAccountWithGateways(t *testing.T) {
 
 	// If this tests fails with wrong number after 10 seconds we may have
 	// added a new inititial subscription for the eventing system.
-	checkExpectedSubs(t, 36, sa)
+	checkExpectedSubs(t, 37, sa)
 
 	// Create a client on B and see if we receive the event
 	urlb := fmt.Sprintf("nats://%s:%d", ob.Host, ob.Port)
diff --git a/server/filestore_test.go b/server/filestore_test.go
index b36389a8..4c4dc142 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -1039,8 +1039,8 @@ func TestFileStoreBitRot(t *testing.T) {
 		t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 	}
 
-	if badSeqs := len(fs.checkMsgs()); badSeqs > 0 {
-		t.Fatalf("Expected to have no corrupt msgs, got %d", badSeqs)
+	if ld := fs.checkMsgs(); ld != nil && len(ld.Msgs) > 0 {
+		t.Fatalf("Expected to have no corrupt msgs, got %d", len(ld.Msgs))
 	}
 
 	// Now twiddle some bits.
@@ -1060,8 +1060,8 @@ func TestFileStoreBitRot(t *testing.T) {
 	ioutil.WriteFile(lmb.mfn, contents, 0644)
 	fs.mu.Unlock()
 
-	bseqs := fs.checkMsgs()
-	if badSeqs := len(bseqs); badSeqs == 0 {
+	ld := fs.checkMsgs()
+	if ld == nil || len(ld.Msgs) == 0 {
 		t.Fatalf("Expected to have corrupt msgs got none: changed [%d]", index)
 	}
 
@@ -1074,8 +1074,9 @@ func TestFileStoreBitRot(t *testing.T) {
 	}
 	defer fs.Stop()
 
-	if !reflect.DeepEqual(bseqs, fs.checkMsgs()) {
-		t.Fatalf("Different reporting on bad msgs: %+v vs %+v", bseqs, fs.checkMsgs())
+	// checkMsgs will repair the underlying store, so checkMsgs should be clean now.
+	if ld := fs.checkMsgs(); ld != nil {
+		t.Fatalf("Expected no errors restoring checked and fixed filestore, got %+v", ld)
 	}
 }
 
@@ -1620,6 +1621,7 @@ func TestFileStoreSnapshot(t *testing.T) {
 	// This will unzip the snapshot and create a new filestore that will recover the state.
 	// We will compare the states for this vs the original one.
 	verifySnapshot := func(snap []byte) {
+		t.Helper()
 		r := bytes.NewReader(snap)
 		tr := tar.NewReader(s2.NewReader(r))
 
@@ -1668,7 +1670,7 @@ func TestFileStoreSnapshot(t *testing.T) {
 		// work through that problem too. The test below will pass but if you try to extract a
 		// message that will most likely fail.
 		if !reflect.DeepEqual(rstate, state) {
-			t.Fatalf("Restored state does not match, %+v vs %+v", rstate, state)
+			t.Fatalf("Restored state does not match:\n%+v\n\n%+v", rstate, state)
 		}
 	}
 
@@ -1724,7 +1726,7 @@ func TestFileStoreSnapshot(t *testing.T) {
 	}
 
 	// Cause snapshot to timeout.
-	time.Sleep(30 * time.Millisecond)
+	time.Sleep(50 * time.Millisecond)
 	// Read should fail
 	var buf [32]byte
 	if _, err := sr.Reader.Read(buf[:]); err != io.EOF {
@@ -1849,6 +1851,93 @@ func TestFileStoreConsumer(t *testing.T) {
 	updateAndCheck()
 }
 
+func TestFileStoreWriteFailures(t *testing.T) {
+	// This test should be run inside an environment where this directory
+	// has a limited size.
+	// E.g. Docker
+	// docker run -ti --tmpfs /jswf_test:rw,size=32k --rm -v ~/Development/go/src:/go/src -w /go/src/github.com/nats-io/nats-server/ golang:1.15 /bin/bash
+	tdir := path.Join("/", "jswf_test")
+	if stat, err := os.Stat(tdir); err != nil || !stat.IsDir() {
+		t.SkipNow()
+	}
+
+	storeDir := path.Join(tdir, JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	subj, msg := "foo", []byte("Hello Write Failures!")
+	fs, _, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	var lseq uint64
+	// msz about will be ~54 bytes, so if limit is 32k trying to send 1000 will fail at some point.
+	for i := 1; i <= 1000; i++ {
+		if _, _, err := fs.StoreMsg(subj, nil, msg); err != nil {
+			lseq = uint64(i)
+			break
+		}
+	}
+	if lseq == 0 {
+		t.Fatalf("Expected to get a failure but did not")
+	}
+
+	state := fs.State()
+	if state.LastSeq != lseq-1 {
+		t.Fatalf("Expected last seq to be %d, got %d\n", lseq-1, state.LastSeq)
+	}
+	if state.Msgs != lseq-1 {
+		t.Fatalf("Expected total msgs to be %d, got %d\n", lseq-1, state.Msgs)
+	}
+	if _, _, _, _, err := fs.LoadMsg(lseq); err == nil {
+		t.Fatalf("Expected error loading seq that failed, got none")
+	}
+	// Loading should still work.
+	if _, _, _, _, err := fs.LoadMsg(2); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Make sure we recover same state.
+	fs.Stop()
+
+	fs, _, err = newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	state2 := fs.State()
+	// Ignore lost state.
+	state.Lost, state2.Lost = nil, nil
+	if !reflect.DeepEqual(state2, state) {
+		t.Fatalf("Expected recovered state to be the same, got %+v vs %+v\n", state2, state)
+	}
+
+	// We should still fail here.
+	if _, _, err = fs.StoreMsg(subj, nil, msg); err == nil {
+		t.Fatalf("Expected to get a failure but did not")
+	}
+
+	// Purge should help.
+	if _, err := fs.Purge(); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	// Wait for purge to complete its out of band processing.
+	time.Sleep(50 * time.Millisecond)
+
+	// Check we will fail again in same spot.
+	for i := 1; i <= 1000; i++ {
+		if _, _, err := fs.StoreMsg(subj, nil, msg); err != nil {
+			if i != int(lseq) {
+				t.Fatalf("Expected to fail after purge about the same spot, wanted %d got %d", lseq, i)
+			}
+			break
+		}
+	}
+}
+
 func TestFileStorePerf(t *testing.T) {
 	// Comment out to run, holding place for now.
 	t.SkipNow()
diff --git a/server/jwt_test.go b/server/jwt_test.go
index 87c8cc81..a78e7313 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -4426,7 +4426,7 @@ func TestJWTUserRevocation(t *testing.T) {
 	}
 	m := <-ncChan
 	require_Len(t, strings.Count(string(m.Data), apub), 2)
-	require_True(t, strings.Contains(string(m.Data), `"jwt": "eyJ0`))
+	require_True(t, strings.Contains(string(m.Data), `"jwt":"eyJ0`))
 	// try again with old credentials. Expected to fail
 	if nc1, err := nats.Connect(srv.ClientURL(), nats.UserCredentials(aCreds1)); err == nil {
 		nc1.Close()
@@ -4513,12 +4513,12 @@ func TestJWTAccountOps(t *testing.T) {
 			// delete nothing
 			resp, err = nc.Request(accDeleteReqSubj, generateRequest([]string{}), time.Second)
 			require_NoError(t, err)
-			require_True(t, strings.Contains(string(resp.Data), `"message": "deleted 0 accounts"`))
+			require_True(t, strings.Contains(string(resp.Data), `"message":"deleted 0 accounts"`))
 			// issue delete, twice to also delete a non existing account
 			for i := 0; i < 2; i++ {
 				resp, err = nc.Request(accDeleteReqSubj, generateRequest([]string{apub}), time.Second)
 				require_NoError(t, err)
-				require_True(t, strings.Contains(string(resp.Data), `"message": "deleted 1 accounts"`))
+				require_True(t, strings.Contains(string(resp.Data), `"message":"deleted 1 accounts"`))
 				resp, err = nc.Request(accListReqSubj, nil, time.Second)
 				require_False(t, strings.Contains(string(resp.Data), apub))
 				require_True(t, strings.Contains(string(resp.Data), syspub))
@@ -5373,7 +5373,7 @@ func TestJWTAccountProtectedImport(t *testing.T) {
 		require_NoError(t, err)
 		sysUsrCreds := newUserEx(t, sysAcc, false, sysPub)
 		defer os.Remove(sysUsrCreds)
-		cf := createConfFile(t, []byte(fmt.Sprintf(`		
+		cf := createConfFile(t, []byte(fmt.Sprintf(`
 		port: -1
 		operator = %s
 		system_account = %s
diff --git a/server/monitor_test.go b/server/monitor_test.go
index 03b6af22..d2c2932f 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -3789,7 +3789,7 @@ func TestMonitorAccountz(t *testing.T) {
 		t.Fatalf("Body missing value. Contains: %s", body)
 	} else if !strings.Contains(body, `"account_name": "$SYS",`) {
 		t.Fatalf("Body missing value. Contains: %s", body)
-	} else if !strings.Contains(body, `"subscriptions": 35,`) {
+	} else if !strings.Contains(body, `"subscriptions": 36,`) {
 		t.Fatalf("Body missing value. Contains: %s", body)
 	}
 }
diff --git a/test/jetstream_cluster_test.go b/test/jetstream_cluster_test.go
index c74a3bda..8ceb3972 100644
--- a/test/jetstream_cluster_test.go
+++ b/test/jetstream_cluster_test.go
@@ -68,7 +68,7 @@ func TestJetStreamClusterLeader(t *testing.T) {
 
 	// Kill our current leader and force an election.
 	c.leader().Shutdown()
-	c.waitOnClusterReady()
+	c.waitOnLeader()
 
 	// Now killing our current leader should leave us leaderless.
 	c.leader().Shutdown()
@@ -642,7 +642,10 @@ func TestJetStreamClusterMetaSnapshotsAndCatchup(t *testing.T) {
 
 	for i := 0; i < numStreams; i++ {
 		sn := fmt.Sprintf("T-%d", i+1)
-		resp, _ := nc.Request(fmt.Sprintf(server.JSApiStreamDeleteT, sn), nil, time.Second)
+		resp, err := nc.Request(fmt.Sprintf(server.JSApiStreamDeleteT, sn), nil, time.Second)
+		if err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
 		var dResp server.JSApiStreamDeleteResponse
 		if err := json.Unmarshal(resp.Data, &dResp); err != nil {
 			t.Fatalf("Unexpected error: %v", err)
@@ -687,10 +690,21 @@ func TestJetStreamClusterMetaSnapshotsMultiChange(t *testing.T) {
 	// Add in a new server to the group. This way we know we can delete the original streams and consumers.
 	rs := c.addInNewServer()
 	c.waitOnServerCurrent(rs)
+	rsnn := rs.NodeName()
 
 	// Shut it down.
 	rs.Shutdown()
 
+	// Wait for the peer to be removed.
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		for _, p := range s.JetStreamClusterPeers() {
+			if p == rsnn {
+				return fmt.Errorf("Old server still in peer set")
+			}
+		}
+		return nil
+	})
+
 	// We want to make changes here that test each delta scenario for the meta snapshots.
 	// Add new stream and consumer.
 	if _, err = js.AddStream(&nats.StreamConfig{Name: "S3"}); err != nil {
@@ -2537,7 +2551,7 @@ func TestJetStreamClusterNoQuorumStepdown(t *testing.T) {
 	}
 
 	// Make sure we received our lost quorum advisories.
-	adv, _ := ssub.NextMsg(2 * time.Second)
+	adv, _ := ssub.NextMsg(5 * time.Second)
 	if adv == nil {
 		t.Fatalf("Expected to receive a stream quorum lost advisory")
 	}
@@ -3479,6 +3493,82 @@ func TestJetStreamClusterStreamPerf(t *testing.T) {
 	fmt.Printf("%.0f msgs/sec\n\n", float64(toSend)/tt.Seconds())
 }
 
+// This test creates a queue consumer for the delivery subject,
+// and make sure it connects to the server that is not the leader
+// of the stream. A bug was not stripping the $JS.ACK reply subject
+// correctly, which means that ack sent on the reply subject was
+// droped by the routed server.
+func TestJetStreamClusterQueueSubConsumer(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R2S", 2)
+	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{"foo.>"},
+		Replicas: 1,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	inbox := nats.NewInbox()
+	ci, err := js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:        "ivan",
+		DeliverSubject: inbox,
+		AckPolicy:      nats.AckExplicitPolicy,
+		AckWait:        100 * time.Millisecond,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Now create a client that does NOT connect to the stream leader.
+	// Start with url from first server in the cluster.
+	u := c.servers[0].ClientURL()
+	// If leader is "S-1", then use S-2 to connect to, which is at servers[1].
+	if ci.Cluster.Leader == "S-1" {
+		u = c.servers[1].ClientURL()
+	}
+	qsubnc, err := nats.Connect(u)
+	if err != nil {
+		t.Fatalf("Error connecting: %v", err)
+	}
+	defer qsubnc.Close()
+
+	ch := make(chan struct{}, 2)
+	if _, err := qsubnc.QueueSubscribe(inbox, "queue", func(m *nats.Msg) {
+		m.Respond(nil)
+		ch <- struct{}{}
+	}); err != nil {
+		t.Fatalf("Error creating sub: %v", err)
+	}
+
+	// Use the other connection to publish a message
+	if _, err := js.Publish("foo.bar", []byte("hello")); err != nil {
+		t.Fatalf("Error on publish: %v", err)
+	}
+
+	// Wait that we receive the message first.
+	select {
+	case <-ch:
+	case <-time.After(time.Second):
+		t.Fatal("Did not receive message")
+	}
+
+	// Message should be ack'ed and not redelivered.
+	select {
+	case <-ch:
+		t.Fatal("Message redelivered!!!")
+	case <-time.After(250 * time.Millisecond):
+		// OK
+	}
+}
+
 // Support functions
 
 var jsClusterTempl = `
@@ -3551,10 +3641,10 @@ func createJetStreamSuperCluster(t *testing.T, numServersPer, numClusters int) *
 		}
 		routeConfig := strings.Join(routes, ",")
 
-		for i := 0; i < numServersPer; i++ {
+		for si := 0; si < numServersPer; si++ {
 			storeDir, _ := ioutil.TempDir("", server.JetStreamStoreDir)
-			sn := fmt.Sprintf("%s-S%d", cn, i+1)
-			bconf := fmt.Sprintf(jsClusterTempl, sn, storeDir, cn, cp+i, routeConfig)
+			sn := fmt.Sprintf("%s-S%d", cn, si+1)
+			bconf := fmt.Sprintf(jsClusterTempl, sn, storeDir, cn, cp+si, routeConfig)
 			conf := fmt.Sprintf(jsSuperClusterTempl, bconf, cn, gp, gwconf)
 			gp++
 			s, o := RunServerWithConfig(createConfFile(t, []byte(conf)))
@@ -3578,9 +3668,36 @@ func createJetStreamSuperCluster(t *testing.T, numServersPer, numClusters int) *
 	sc := &supercluster{t, clusters}
 	sc.waitOnLeader()
 	sc.waitOnAllCurrent()
+
+	// Wait for all the peer nodes to be registered.
+	checkFor(t, 5*time.Second, 100*time.Millisecond, func() error {
+		var peers []string
+		if ml := sc.leader(); ml != nil {
+			peers = ml.ActivePeers()
+			if len(peers) == numClusters*numServersPer {
+				return nil
+			}
+		}
+		fmt.Printf("AP is %+v\n", peers)
+		return fmt.Errorf("Not correct number of peers, expected %d, got %d", numClusters*numServersPer, len(peers))
+	})
+
+	if sc.leader() == nil {
+		sc.t.Fatalf("Expected a cluster leader, got none")
+	}
+
 	return sc
 }
 
+func (sc *supercluster) leader() *server.Server {
+	for _, c := range sc.clusters {
+		if leader := c.leader(); leader != nil {
+			return leader
+		}
+	}
+	return nil
+}
+
 func (sc *supercluster) waitOnLeader() {
 	expires := time.Now().Add(5 * time.Second)
 	for time.Now().Before(expires) {
diff --git a/test/norace_test.go b/test/norace_test.go
index 1092ac13..57e7d07d 100644
--- a/test/norace_test.go
+++ b/test/norace_test.go
@@ -756,11 +756,6 @@ func TestJetStreamClusterStreamCreateAndLostQuorum(t *testing.T) {
 	c.restartAll()
 
 	c.waitOnStreamLeader("$G", "NO-LQ-START")
-
-	for m, err := sub.NextMsg(0); err == nil; m, err = sub.NextMsg(0) {
-		fmt.Printf("m: %s\n", m.Data)
-	}
-
 	checkSubsPending(t, sub, 0)
 }