Repository: nats-io/nats-server
Pull request #1675 (2020-10-27 02:09:58)

Stability and performance updates for filestore.
The original design had a shared filestore write buffer and individual message blocks had a read cache.
This presented some performance and stability issues when lots of reads and deletes were happening to a
message block that was also being written to actively.

This change eliminates the shared write buffer and uses the message block's cache as a write through as
well as read cache and handles partials correctly.

This PR also introduces a dynamic autotuned block size for file based streams aiming at 25% overhead in most cases.
We also now dynamically discover the available disk space by storage directory.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core


derekcollison: I just noticed a performance regression, will need to address.
derekcollison: ok performance regression has been identified.

Patch:
diff --git a/conf/lex.go b/conf/lex.go
index 9358c9b3..b2357702 100644
--- a/conf/lex.go
+++ b/conf/lex.go
@@ -995,7 +995,7 @@ func lexNumberOrDateOrStringOrIP(lx *lexer) stateFn {
 func lexConvenientNumber(lx *lexer) stateFn {
 	r := lx.next()
 	switch {
-	case r == 'b' || r == 'B' || r == 'i' || r == 'I':
+	case r == 'b' || r == 'B':
 		return lexConvenientNumber
 	}
 	lx.backup()
diff --git a/conf/parse.go b/conf/parse.go
index afb6f6e8..a22b3e3d 100644
--- a/conf/parse.go
+++ b/conf/parse.go
@@ -253,27 +253,27 @@ func (p *parser) processItem(it item, fp string) error {
 			setValue(it, num)
 		case "k":
 			setValue(it, num*1000)
-		case "kb", "ki", "kib":
+		case "kb":
 			setValue(it, num*1024)
 		case "m":
 			setValue(it, num*1000*1000)
-		case "mb", "mi", "mib":
+		case "mb":
 			setValue(it, num*1024*1024)
 		case "g":
 			setValue(it, num*1000*1000*1000)
-		case "gb", "gi", "gib":
+		case "gb":
 			setValue(it, num*1024*1024*1024)
 		case "t":
 			setValue(it, num*1000*1000*1000*1000)
-		case "tb", "ti", "tib":
+		case "tb":
 			setValue(it, num*1024*1024*1024*1024)
 		case "p":
 			setValue(it, num*1000*1000*1000*1000*1000)
-		case "pb", "pi", "pib":
+		case "pb":
 			setValue(it, num*1024*1024*1024*1024*1024)
 		case "e":
 			setValue(it, num*1000*1000*1000*1000*1000*1000)
-		case "eb", "ei", "eib":
+		case "eb":
 			setValue(it, num*1024*1024*1024*1024*1024*1024)
 		}
 	case itemFloat:
diff --git a/server/consumer.go b/server/consumer.go
index 1bc7c0a4..fc5fad90 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -33,16 +33,16 @@ import (
 )
 
 type ConsumerInfo struct {
-	Stream         string         `json:"stream_name"`
-	Name           string         `json:"name"`
-	Created        time.Time      `json:"created"`
-	Config         ConsumerConfig `json:"config"`
-	Delivered      SequencePair   `json:"delivered"`
-	AckFloor       SequencePair   `json:"ack_floor"`
-	NumAckPending  int            `json:"num_ack_pending"`
-	NumRedelivered int            `json:"num_redelivered"`
-	NumWaiting     int            `json:"num_waiting"`
-	NumPending     uint64         `json:"num_pending"`
+	Stream           string         `json:"stream_name"`
+	Name             string         `json:"name"`
+	Created          time.Time      `json:"created"`
+	Config           ConsumerConfig `json:"config"`
+	Delivered        SequencePair   `json:"delivered"`
+	AckFloor         SequencePair   `json:"ack_floor"`
+	NumPending       int            `json:"num_pending"`
+	NumRedelivered   int            `json:"num_redelivered"`
+	NumWaiting       int            `json:"num_waiting"`
+	NumStreamPending uint64         `json:"num_stream_pending"`
 }
 
 type ConsumerConfig struct {
@@ -385,7 +385,7 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 		dsubj:   config.DeliverSubject,
 		active:  true,
 		qch:     make(chan struct{}),
-		fch:     make(chan struct{}),
+		fch:     make(chan struct{}, 1),
 		sfreq:   int32(sampleFreq),
 		maxdc:   uint64(config.MaxDeliver),
 		created: time.Now().UTC(),
@@ -846,7 +846,7 @@ func (o *Consumer) readStoredState() error {
 func (o *Consumer) writeState() {
 	o.mu.Lock()
 	if o.store != nil {
-		state := &ConsumerState{
+		state := ConsumerState{
 			Delivered: SequencePair{
 				ConsumerSeq: o.dseq,
 				StreamSeq:   o.sseq,
@@ -859,7 +859,7 @@ func (o *Consumer) writeState() {
 			Redelivered: o.rdc,
 		}
 		// FIXME(dlc) - Hold onto any errors.
-		o.store.Update(state)
+		o.store.Update(&state)
 	}
 	o.mu.Unlock()
 }
@@ -903,9 +903,9 @@ func (o *Consumer) Info() *ConsumerInfo {
 			ConsumerSeq: o.adflr,
 			StreamSeq:   o.asflr,
 		},
-		NumAckPending:  len(o.pending),
-		NumRedelivered: len(o.rdc),
-		NumPending:     o.spending,
+		NumPending:       len(o.pending),
+		NumRedelivered:   len(o.rdc),
+		NumStreamPending: o.spending,
 	}
 	// If we are a pull mode consumer, report on number of waiting requests.
 	if o.isPullMode() {
@@ -2090,8 +2090,7 @@ func (o *Consumer) decStreamPending(sseq uint64) {
 	if o.config.FilterSubject == _EMPTY_ {
 		o.spending--
 	} else if o.mset != nil {
-		subj, _, _, _, _ := o.mset.store.LoadMsg(sseq)
-		if o.isFilteredMatch(subj) {
+		if subj, _, _, _, _ := o.mset.store.LoadMsg(sseq); o.isFilteredMatch(subj) {
 			o.spending--
 		}
 	}
diff --git a/server/filestore.go b/server/filestore.go
index dd30cf85..3d05243d 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -32,7 +32,6 @@ import (
 	"os"
 	"path"
 	"sort"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -45,8 +44,8 @@ type FileStoreConfig struct {
 	StoreDir string
 	// BlockSize is the file block size. This also represents the maximum overhead size.
 	BlockSize uint64
-	// ReadCacheExpire is how long with no activity until we expire the read cache.
-	ReadCacheExpire time.Duration
+	// CacheExpire is how long with no activity until we expire the cache.
+	CacheExpire time.Duration
 	// SyncInterval is how often we sync to disk in the background.
 	SyncInterval time.Duration
 }
@@ -75,7 +74,6 @@ type fileStore struct {
 	lmb      *msgBlock
 	blks     []*msgBlock
 	hh       hash.Hash64
-	wmb      *bytes.Buffer
 	fch      chan struct{}
 	qch      chan struct{}
 	cfs      []*consumerFileStore
@@ -86,33 +84,41 @@ type fileStore struct {
 
 // 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
-	hh     hash.Hash64
-	cache  *cache
-	expire time.Duration
-	ctmr   *time.Timer
-	cgenid uint64
-	cloads uint64
-	dmap   map[uint64]struct{}
-	dch    chan struct{}
-	qch    chan struct{}
-	lchk   [8]byte
-}
-
+	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
+	lwts    int64
+	llts    int64
+	lrts    int64
+	hh      hash.Hash64
+	cache   *cache
+	cloads  uint64
+	cexp    time.Duration
+	ctmr    *time.Timer
+	loading bool
+	dmap    map[uint64]struct{}
+	dch     chan struct{}
+	qch     chan struct{}
+	lchk    [8]byte
+}
+
+// Write through caching layer that is also used on loading messages.
 type cache struct {
-	buf  []byte
-	idx  []uint32
-	fseq uint64
+	buf   []byte
+	off   int
+	wp    int
+	idx   []uint32
+	lrl   uint32
+	fseq  uint64
+	flush bool
 }
 
 type msgId struct {
@@ -126,6 +132,7 @@ type fileStoredMsg struct {
 	msg  []byte
 	seq  uint64
 	ts   int64 // nanoseconds
+	mb   *msgBlock
 	off  int64 // offset into block file
 }
 
@@ -154,22 +161,30 @@ const (
 	tmplsDir = "templates"
 	// Maximum size of a write buffer we may consider for re-use.
 	maxBufReuse = 2 * 1024 * 1024
-	// Default stream block size.
-	defaultStreamBlockSize = 64 * 1024 * 1024 // 64MB
-	// Default for workqueue or interest based.
-	defaultOtherBlockSize = 32 * 1024 * 1024 // 32MB
-	// max block size for now.
-	maxBlockSize = 2 * defaultStreamBlockSize
-	// default cache expiration
-	defaultCacheExpiration = 5 * time.Second
+	// default cache buffer expiration
+	defaultCacheBufferExpiration = 5 * time.Second
+	// cache idx expiration
+	defaultCacheIdxExpiration = 5 * time.Minute
 	// default sync interval
 	defaultSyncInterval = 10 * time.Second
 	// coalesceMinimum
-	coalesceMinimum = 64 * 1024
-
+	coalesceMinimum = 4 * 1024
+	// maxFlushWait is maximum we will wait to gather messages to flush.
+	maxFlushWait = 8 * time.Millisecond
 	// Metafiles for streams and consumers.
 	JetStreamMetaFile    = "meta.inf"
 	JetStreamMetaFileSum = "meta.sum"
+
+	// Default stream block size.
+	defaultStreamBlockSize = 64 * 1024 * 1024 // 64MB
+	// Default for workqueue or interest based.
+	defaultOtherBlockSize = 32 * 1024 * 1024 // 32MB
+	// max block size for now.
+	maxBlockSize = 2 * defaultStreamBlockSize
+	// FileStoreMinBlkSize is minimum size we will do for a blk size.
+	FileStoreMinBlkSize = 32 * 1000 // 32kib
+	// FileStoreMaxBlkSize is maximum size we will do for a blk size.
+	FileStoreMaxBlkSize = maxBlockSize
 )
 
 func newFileStore(fcfg FileStoreConfig, cfg StreamConfig) (*fileStore, error) {
@@ -190,8 +205,8 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	if fcfg.BlockSize > maxBlockSize {
 		return nil, fmt.Errorf("filestore max block size is %s", FriendlyBytes(maxBlockSize))
 	}
-	if fcfg.ReadCacheExpire == 0 {
-		fcfg.ReadCacheExpire = defaultCacheExpiration
+	if fcfg.CacheExpire == 0 {
+		fcfg.CacheExpire = defaultCacheBufferExpiration
 	}
 	if fcfg.SyncInterval == 0 {
 		fcfg.SyncInterval = defaultSyncInterval
@@ -214,8 +229,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	fs := &fileStore{
 		fcfg: fcfg,
 		cfg:  FileStreamInfo{Created: created, StreamConfig: cfg},
-		wmb:  &bytes.Buffer{},
-		fch:  make(chan struct{}),
+		fch:  make(chan struct{}, 1), // Make buffered on purpose.
 		qch:  make(chan struct{}),
 	}
 
@@ -249,7 +263,9 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		}
 	}
 
-	go fs.flushLoop(fs.fch, fs.qch)
+	sch := make(chan struct{})
+	go fs.flushLoop(fs.fch, fs.qch, sch)
+	<-sch // Will be closed when flush starts.
 
 	fs.syncTmr = time.AfterFunc(fs.fcfg.SyncInterval, fs.syncBlocks)
 
@@ -297,6 +313,21 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
 }
 
 func dynBlkSize(retention RetentionPolicy, maxBytes int64) uint64 {
+	if maxBytes > 0 {
+		blkSize := (maxBytes / 4) + 1 // (25% overhead)
+		// Round up to nearest 100
+		if m := blkSize % 100; m != 0 {
+			blkSize += 100 - m
+		}
+		if blkSize < FileStoreMinBlkSize {
+			blkSize = FileStoreMinBlkSize
+		}
+		if blkSize > FileStoreMaxBlkSize {
+			blkSize = FileStoreMaxBlkSize
+		}
+		return uint64(blkSize)
+	}
+
 	if retention == LimitsPolicy {
 		// TODO(dlc) - Make the blocksize relative to this if set.
 		return defaultStreamBlockSize
@@ -339,7 +370,7 @@ 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, expire: fs.fcfg.ReadCacheExpire}
+	mb := &msgBlock{index: index, cexp: fs.fcfg.CacheExpire}
 
 	mdir := path.Join(fs.fcfg.StoreDir, msgDir)
 	mb.mfn = path.Join(mdir, fi.Name())
@@ -374,6 +405,16 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint64) *msgBlock {
 		mb.first.seq = 0
 	}
 
+	addToDmap := func(seq uint64) {
+		if seq == 0 {
+			return
+		}
+		if mb.dmap == nil {
+			mb.dmap = make(map[uint64]struct{})
+		}
+		mb.dmap[seq] = struct{}{}
+	}
+
 	// Use data file itself to rebuild.
 	var hdr [msgHdrSize]byte
 	var offset int64
@@ -385,8 +426,15 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint64) *msgBlock {
 		}
 		rl := le.Uint32(hdr[0:])
 		seq := le.Uint64(hdr[4:])
-		// This is an erased message.
-		if seq == 0 {
+
+		// Can't recover with zero record length.
+		if rl == 0 {
+			return nil
+		}
+
+		// This is an old erased message, or a new one that we can track.
+		if seq == 0 || seq&ebit != 0 {
+			addToDmap(seq &^ ebit)
 			offset += int64(rl)
 			continue
 		}
@@ -422,7 +470,7 @@ func (fs *fileStore) recoverMsgs() error {
 	mdir := path.Join(fs.fcfg.StoreDir, msgDir)
 	fis, err := ioutil.ReadDir(mdir)
 	if err != nil {
-		return fmt.Errorf("storage directory not readable")
+		return errNotReadable
 	}
 
 	// Recover all of the msg blocks.
@@ -534,7 +582,7 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 		fs.closeLastMsgBlock(false)
 	}
 
-	mb := &msgBlock{index: index, expire: fs.fcfg.ReadCacheExpire}
+	mb := &msgBlock{index: index, cexp: fs.fcfg.CacheExpire}
 	fs.blks = append(fs.blks, mb)
 	fs.lmb = mb
 
@@ -595,7 +643,7 @@ func (fs *fileStore) StoreMsg(subj string, hdr, msg []byte) (uint64, int64, erro
 			fs.mu.Unlock()
 			return 0, 0, ErrMaxMsgs
 		}
-		if fs.cfg.MaxBytes > 0 && fs.state.Bytes+uint64(len(msg)) >= uint64(fs.cfg.MaxBytes) {
+		if fs.cfg.MaxBytes > 0 && fs.state.Bytes+uint64(len(msg)+len(hdr)) >= uint64(fs.cfg.MaxBytes) {
 			fs.mu.Unlock()
 			return 0, 0, ErrMaxBytes
 		}
@@ -641,6 +689,37 @@ func (fs *fileStore) StoreMsg(subj string, hdr, msg []byte) (uint64, int64, erro
 	return seq, ts, nil
 }
 
+// skipMsg will update this message block for a skipped message.
+// If we do not have any messages, just update the metadata, otherwise
+// we will place and empty record marking the sequence as used. The
+// sequence will be marked erased.
+// fs lock should be held.
+func (mb *msgBlock) skipMsg(seq uint64, now time.Time) {
+	if mb == nil {
+		return
+	}
+	// If we are empty can just do meta.
+	if mb.msgs == 0 {
+		mb.mu.Lock()
+		mb.last.seq = seq
+		mb.last.ts = now.UnixNano()
+		mb.first.seq = seq + 1
+		mb.first.ts = now.UnixNano()
+		mb.mu.Unlock()
+	} else {
+		mb.writeMsgRecord(emptyRecordLen, seq|ebit, _EMPTY_, nil, nil, now.UnixNano())
+		mb.mu.Lock()
+		if mb.dmap == nil {
+			mb.dmap = make(map[uint64]struct{})
+		}
+		mb.dmap[seq] = struct{}{}
+		mb.msgs--
+		mb.bytes -= emptyRecordLen
+		mb.mu.Unlock()
+	}
+	mb.kickWriteFlusher()
+}
+
 // SkipMsg will use the next sequence number but not store anything.
 func (fs *fileStore) SkipMsg() uint64 {
 	// Grab time.
@@ -657,6 +736,8 @@ func (fs *fileStore) SkipMsg() uint64 {
 		fs.state.FirstSeq = seq + 1
 		fs.state.FirstTime = now
 	}
+	fs.lmb.skipMsg(seq, now)
+	fs.kickFlusher()
 	fs.mu.Unlock()
 	return seq
 }
@@ -708,41 +789,173 @@ func (fs *fileStore) EraseMsg(seq uint64) (bool, error) {
 	return fs.removeMsg(seq, true)
 }
 
-func (fs *fileStore) isClosed() bool {
-	fs.mu.RLock()
-	closed := fs.closed
-	fs.mu.RUnlock()
-	return closed
-}
-
-func (fs *fileStore) isSnapshotting() bool {
-	fs.mu.RLock()
-	iss := fs.sips > 0
-	fs.mu.RUnlock()
-	return iss
-}
-
 // Remove a message, optionally rewriting the mb file.
 func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
-	if fs.isClosed() {
+	fs.mu.Lock()
+	if fs.closed {
+		fs.mu.Unlock()
 		return false, ErrStoreClosed
 	}
-	if fs.isSnapshotting() {
+	if fs.sips > 0 {
+		fs.mu.Unlock()
 		return false, ErrStoreSnapshotInProgress
 	}
+
 	mb := fs.selectMsgBlock(seq)
 	if mb == nil {
-		return false, nil
+		var err = ErrStoreEOF
+		if seq <= fs.state.LastSeq {
+			err = ErrStoreMsgNotFound
+		}
+		fs.mu.Unlock()
+		return false, err
 	}
-	sm, _ := mb.fetchMsg(seq)
-	// We might have the message here, so we can delete it.
-	found := sm != nil
-	if found {
-		if err := fs.deleteMsgFromBlock(mb, seq, sm, secure); err != nil {
+
+	mb.mu.Lock()
+
+	// Check cache. This will be very rare, will hold lock on this one.
+	if mb.cache == nil || mb.cache.idx == nil {
+		mb.mu.Unlock()
+		if err := mb.loadMsgs(); err != nil {
+			fs.mu.Unlock()
 			return false, err
 		}
+		mb.mu.Lock()
+	}
+
+	// See if the sequence numbers is still relevant. Check first and cache first.
+	if seq < mb.first.seq || seq < mb.cache.fseq || (seq-mb.cache.fseq) >= uint64(len(mb.cache.idx)) {
+		mb.mu.Unlock()
+		fs.mu.Unlock()
+		return false, nil
+	}
+
+	// Now check dmap if it is there.
+	if mb.dmap != nil {
+		if _, ok := mb.dmap[seq]; ok {
+			mb.mu.Unlock()
+			fs.mu.Unlock()
+			return false, nil
+		}
+	}
+
+	// Set cache timestamp for last remove.
+	mb.lrts = time.Now().UnixNano()
+
+	// Grab record length from idx.
+	slot := seq - mb.cache.fseq
+	ri, rl, _, _ := mb.slotInfo(int(slot))
+	msz := uint64(rl)
+
+	// Call this early so that we can look this message up if needed.
+	if cb := fs.scb; cb != nil {
+		mb.mu.Unlock()
+		fs.mu.Unlock()
+		delta := int64(msz)
+		cb(-1, -delta, seq)
+		fs.mu.Lock()
+		mb.mu.Lock()
+	}
+
+	// Global stats
+	fs.state.Msgs--
+	fs.state.Bytes -= msz
+
+	// Now local mb updates.
+	mb.msgs--
+	mb.bytes -= msz
+
+	var shouldWriteIndex bool
+	var firstSeqNeedsUpdate bool
+
+	if secure {
+		mb.eraseMsg(seq, int(ri), int(rl))
+	}
+
+	// Optimize for FIFO case.
+	if seq == mb.first.seq {
+		mb.selectNextFirst()
+		if mb.isEmpty() {
+			fs.removeMsgBlock(mb)
+			firstSeqNeedsUpdate = seq == fs.state.FirstSeq
+		} else {
+			shouldWriteIndex = true
+			if seq == fs.state.FirstSeq {
+				fs.state.FirstSeq = mb.first.seq // new one.
+				fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
+			}
+		}
+	} else {
+		// Out of order delete.
+		if mb.dmap == nil {
+			mb.dmap = make(map[uint64]struct{})
+		}
+		mb.dmap[seq] = struct{}{}
+		shouldWriteIndex = true
+	}
+
+	var dch chan struct{}
+
+	if shouldWriteIndex {
+		if mb.dch == nil {
+			// Spin up the write flusher.
+			mb.qch = make(chan struct{})
+			mb.dch = make(chan struct{})
+			go fs.flushWriteIndexLoop(mb, mb.dch, mb.qch)
+			// Do a blocking kick here to make sure loop is running.
+			mb.dch <- struct{}{}
+		} else {
+			dch = mb.dch
+		}
+	}
+	mb.mu.Unlock()
+
+	// Kick outside of lock.
+	if shouldWriteIndex && dch != nil {
+		select {
+		case dch <- struct{}{}:
+		default:
+		}
 	}
-	return found, nil
+
+	// If we emptied the current message block and the seq was state.First.Seq
+	// then we need to jump message blocks.
+	if firstSeqNeedsUpdate {
+		fs.selectNextFirst()
+	}
+	fs.mu.Unlock()
+
+	return true, nil
+}
+
+// Grab info from a slot.
+// Lock should be held.
+func (mb *msgBlock) slotInfo(slot int) (uint32, uint32, bool, error) {
+	if mb.cache == nil || mb.cache.idx == nil {
+		return 0, 0, false, errPartialCache
+	}
+	bi := mb.cache.idx[slot]
+	ri := (bi &^ hbit)
+	hashChecked := (bi & hbit) != 0
+	// Determine record length
+	var rl uint32
+	if len(mb.cache.idx) > slot+1 {
+		ni := mb.cache.idx[slot+1] &^ hbit
+		rl = ni - ri
+	} else {
+		rl = mb.cache.lrl
+	}
+	if rl < msgHdrSize {
+		return 0, 0, false, errBadMsg
+	}
+	return uint32(ri), rl, hashChecked, nil
+}
+
+func (fs *fileStore) isClosed() bool {
+	fs.mu.RLock()
+	closed := fs.closed
+	fs.mu.RUnlock()
+	return closed
 }
 
 // Loop on requests to write out our index file. This is used when calling
@@ -759,11 +972,66 @@ func (fs *fileStore) flushWriteIndexLoop(mb *msgBlock, dch, qch chan struct{}) {
 	}
 }
 
+// Lock should be held.
+func (mb *msgBlock) eraseMsg(seq uint64, ri, rl int) error {
+	var le = binary.LittleEndian
+	var hdr [msgHdrSize]byte
+
+	le.PutUint32(hdr[0:], uint32(rl))
+	le.PutUint64(hdr[4:], seq|ebit)
+	le.PutUint64(hdr[12:], 0)
+	le.PutUint16(hdr[20:], 0)
+
+	// Randomize record
+	data := make([]byte, rl-emptyRecordLen)
+	rand.Read(data)
+
+	// Now write to underlying buffer.
+	var b bytes.Buffer
+	b.Write(hdr[:])
+	b.Write(data)
+
+	// Calculate hash.
+	mb.hh.Reset()
+	mb.hh.Write(hdr[4:20])
+	mb.hh.Write(data)
+	checksum := mb.hh.Sum(nil)
+	// Write to msg record.
+	b.Write(checksum)
+
+	// Update both cache and disk.
+	nbytes := b.Bytes()
+
+	// Cache
+	if ri >= mb.cache.off {
+		li := ri - mb.cache.off
+		buf := mb.cache.buf[li : li+rl]
+		copy(buf, nbytes)
+	}
+	// Disk
+	if mb.cache.off+mb.cache.wp > ri {
+		mfd, err := os.OpenFile(mb.mfn, os.O_RDWR, 0644)
+		if err != nil {
+			return err
+		}
+		defer mfd.Close()
+		if _, err = mfd.WriteAt(nbytes, int64(ri)); err == nil {
+			mfd.Sync()
+		}
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (mb *msgBlock) kickWriteFlusher() {
+	mb.mu.RLock()
 	select {
 	case mb.dch <- struct{}{}:
 	default:
 	}
+	mb.mu.RUnlock()
 }
 
 // Lock should be held.
@@ -821,150 +1089,76 @@ func (fs *fileStore) selectNextFirst() {
 	}
 }
 
-func (fs *fileStore) deleteMsgFromBlock(mb *msgBlock, seq uint64, sm *fileStoredMsg, secure bool) error {
-	// Update global accounting.
-	msz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
-	delta := int64(msz)
-
-	fs.mu.RLock()
-	cb := fs.scb
-	fs.mu.RUnlock()
-
-	// Call this early so that we can look this message up if needed.
-	if cb != nil {
-		cb(-1, -delta, seq)
+// Lock should be held.
+func (mb *msgBlock) resetCacheExpireTimer(td time.Duration) {
+	if td == 0 {
+		td = mb.cexp
 	}
-
-	fs.mu.Lock()
-	mb.mu.Lock()
-
-	// Make sure the seq still exists.
-	if mb.cache == nil {
-		// Rare condition but can happen. We will load with lock held at this point.
-		buf, err := ioutil.ReadFile(mb.mfn)
-		if err == nil {
-			err = mb.indexCacheBuf(buf)
-		}
-		if err != nil {
-			mb.mu.Unlock()
-			fs.mu.Unlock()
-			return err
-		}
-		if len(buf) > 0 {
-			mb.cloads++
-			mb.startCacheExpireTimer()
-		}
+	if mb.ctmr == nil {
+		mb.ctmr = time.AfterFunc(td, mb.expireCache)
+	} else {
+		mb.ctmr.Reset(td)
 	}
+}
 
-	// See if the sequence numbers is still relevant. Check first and cache first.
-	if seq < mb.first.seq || seq < mb.cache.fseq || (seq-mb.cache.fseq) >= uint64(len(mb.cache.idx)) {
-		mb.mu.Unlock()
-		fs.mu.Unlock()
-		// Undo
-		if cb != nil {
-			cb(1, delta, seq)
-		}
-		return nil
-	}
+// Lock should be held.
+func (mb *msgBlock) startCacheExpireTimer() {
+	mb.resetCacheExpireTimer(0)
+}
 
-	// Now check dmap if it is there.
-	if mb.dmap != nil {
-		if _, ok := mb.dmap[seq]; ok {
-			mb.mu.Unlock()
-			fs.mu.Unlock()
-			// Undo
-			if cb != nil {
-				cb(1, delta, seq)
-			}
-			return nil
-		}
+// Lock should be held.
+func (mb *msgBlock) clearCache() {
+	if mb.ctmr != nil {
+		mb.ctmr.Stop()
+		mb.ctmr = nil
 	}
+	mb.cache = nil
+}
 
-	// Global stats
-	fs.state.Msgs--
-	fs.state.Bytes -= msz
-
-	// Now local mb updates.
-	mb.msgs--
-	mb.bytes -= msz
-	atomic.AddUint64(&mb.cgenid, 1)
-
-	var shouldWriteIndex bool
-	var firstSeqNeedsUpdate bool
+// Called to possibly expire a message block cache.
+func (mb *msgBlock) expireCache() {
+	mb.mu.Lock()
+	defer mb.mu.Unlock()
 
-	// Optimize for FIFO case.
-	if seq == mb.first.seq {
-		mb.selectNextFirst()
-		if mb.isEmpty() {
-			fs.removeMsgBlock(mb)
-			firstSeqNeedsUpdate = seq == fs.state.FirstSeq
-		} else {
-			shouldWriteIndex = true
-			if seq == fs.state.FirstSeq {
-				fs.state.FirstSeq = mb.first.seq // new one.
-				fs.state.FirstTime = time.Unix(0, mb.first.ts).UTC()
-			}
-		}
-	} else {
-		// Out of order delete.
-		if mb.dmap == nil {
-			mb.dmap = make(map[uint64]struct{})
+	if mb.cache == nil {
+		if mb.ctmr != nil {
+			mb.ctmr.Stop()
+			mb.ctmr = nil
 		}
-		mb.dmap[seq] = struct{}{}
-		shouldWriteIndex = true
+		return
 	}
 
-	if secure {
-		fs.eraseMsg(mb, sm)
+	// Can't expire if we are flushing or still have pending.
+	if mb.cache.flush || (len(mb.cache.buf)-mb.cache.wp > 0) {
+		mb.resetCacheExpireTimer(mb.cexp)
+		return
 	}
 
-	if shouldWriteIndex {
-		if mb.dch == nil {
-			// Spin up the write flusher.
-			mb.qch = make(chan struct{})
-			mb.dch = make(chan struct{})
-			go fs.flushWriteIndexLoop(mb, mb.dch, mb.qch)
-			// Do a blocking kick here.
-			mb.dch <- struct{}{}
-		} else {
-			mb.kickWriteFlusher()
-		}
-	}
-	mb.mu.Unlock()
+	// Grab timestamp to compare.
+	tns := time.Now().UnixNano()
 
-	// If we emptied the current message block and the seq was state.First.Seq
-	// then we need to jump message blocks.
-	if firstSeqNeedsUpdate {
-		fs.selectNextFirst()
+	// For the core buffer of messages, we care about reads and writes, but not removes.
+	bufts := mb.llts
+	if mb.lwts > bufts {
+		bufts = mb.lwts
 	}
 
-	fs.mu.Unlock()
-
-	return nil
-}
-
-// Lock should be held.
-func (mb *msgBlock) startCacheExpireTimer() {
-	genid := mb.cgenid
-	if mb.ctmr == nil {
-		mb.ctmr = time.AfterFunc(mb.expire, func() { mb.expireCache(genid) })
-	} else {
-		mb.ctmr.Reset(mb.expire)
+	// Check for the underlying buffer first.
+	if tns-bufts <= int64(mb.cexp) {
+		mb.resetCacheExpireTimer(mb.cexp - time.Duration(tns-bufts))
+		return
 	}
-}
 
-// Called to possibly expire a message block read cache.
-func (mb *msgBlock) expireCache(genid uint64) {
-	mb.mu.Lock()
-	defer mb.mu.Unlock()
+	// If we are here we will at least expire the core msg buffer.
+	mb.cache.buf = nil
+	mb.cache.wp = 0
 
-	cgenid := atomic.LoadUint64(&mb.cgenid)
-
-	if genid == cgenid {
-		mb.cache = nil
-		mb.ctmr = nil
+	// The idx is used in removes, and will have a longer timeframe.
+	// See if we should also remove the idx.
+	if tns-mb.lrts > int64(defaultCacheIdxExpiration) {
+		mb.clearCache()
 	} else {
-		mb.ctmr = time.AfterFunc(mb.expire, func() { mb.expireCache(cgenid) })
+		mb.resetCacheExpireTimer(mb.cexp)
 	}
 }
 
@@ -1100,17 +1294,23 @@ func (fs *fileStore) kickFlusher() {
 	}
 }
 
-func (fs *fileStore) pendingWriteSize() int {
-	var sz int
+func (fs *fileStore) lastMsgBlock() *msgBlock {
 	fs.mu.RLock()
-	if fs.wmb != nil {
-		sz = fs.wmb.Len()
-	}
+	mb := fs.lmb
 	fs.mu.RUnlock()
-	return sz
+	return mb
+}
+
+// Looks at active write message block (lmb).
+func (fs *fileStore) pendingWriteSize() int {
+	return fs.lastMsgBlock().numWriteBytesPending()
 }
 
-func (fs *fileStore) flushLoop(fch, qch chan struct{}) {
+// Main flush loop.
+func (fs *fileStore) flushLoop(fch, qch, sch chan struct{}) {
+	if sch != nil {
+		close(sch)
+	}
 	for {
 		select {
 		case <-fch:
@@ -1119,12 +1319,22 @@ func (fs *fileStore) flushLoop(fch, qch chan struct{}) {
 				continue
 			}
 			ts := 1 * time.Millisecond
+			var waited time.Duration
+
 			for waiting < coalesceMinimum {
 				time.Sleep(ts)
 				newWaiting := fs.pendingWriteSize()
 				if newWaiting <= waiting {
 					break
 				}
+				if waited = waited + ts; waited > maxFlushWait {
+					break
+				}
+				select {
+				case <-qch:
+					return
+				default:
+				}
 				waiting = newWaiting
 				ts *= 2
 			}
@@ -1135,54 +1345,23 @@ func (fs *fileStore) flushLoop(fch, qch chan struct{}) {
 	}
 }
 
-// Return the number of bytes in this message block.
-func (mb *msgBlock) numBytes() uint64 {
-	mb.mu.RLock()
-	nb := mb.bytes
-	mb.mu.RUnlock()
-	return nb
-}
-
-// Update accounting on a write msg.
-func (mb *msgBlock) updateAccounting(seq uint64, ts int64, rl uint64) {
+// 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) {
 	mb.mu.Lock()
-	if mb.first.seq == 0 {
-		mb.first.seq = seq
-		mb.first.ts = ts
-	}
-	// Need atomics here for selectMsgBlock speed.
-	atomic.StoreUint64(&mb.last.seq, seq)
-	mb.last.ts = ts
-	mb.bytes += rl
-	mb.msgs++
-	mb.mu.Unlock()
-}
-
-// Lock should be held.
-func (fs *fileStore) writeMsgRecord(seq uint64, subj string, mhdr, msg []byte) (uint64, int64, error) {
-	var err error
+	defer mb.mu.Unlock()
 
-	// Get size for this message.
-	rl := fileStoreMsgSize(subj, mhdr, msg)
-	if rl&hbit == 1 {
-		return 0, 0, ErrMsgTooLarge
-	}
-	// Grab our current last message block.
-	mb := fs.lmb
-	if mb == nil || mb.numBytes()+rl > fs.fcfg.BlockSize {
-		if mb, err = fs.newMsgBlockForWrite(); err != nil {
-			return 0, 0, err
-		}
+	// Make sure we have a cache setup.
+	if mb.cache == nil {
+		mb.cache = &cache{}
+		mb.startCacheExpireTimer()
 	}
 
-	// Make sure we have room.
-	fs.wmb.Grow(int(rl))
-
-	// Grab time
-	ts := time.Now().UnixNano()
-
-	// Update accounting.
-	mb.updateAccounting(seq, ts, rl)
+	// Indexing
+	index := len(mb.cache.buf)
+	if mb.cache.off > 0 {
+		index += mb.cache.off
+	}
 
 	// Formats
 	// Format with no header
@@ -1206,18 +1385,18 @@ func (fs *fileStore) writeMsgRecord(seq uint64, subj string, mhdr, msg []byte) (
 	le.PutUint16(hdr[20:], uint16(len(subj)))
 
 	// Now write to underlying buffer.
-	fs.wmb.Write(hdr[:])
-	fs.wmb.WriteString(subj)
+	mb.cache.buf = append(mb.cache.buf, hdr[:]...)
+	mb.cache.buf = append(mb.cache.buf, subj...)
+
 	if hasHeaders {
 		var hlen [4]byte
 		le.PutUint32(hlen[0:], uint32(len(mhdr)))
-		fs.wmb.Write(hlen[:])
-		fs.wmb.Write(mhdr)
+		mb.cache.buf = append(mb.cache.buf, hlen[:]...)
+		mb.cache.buf = append(mb.cache.buf, mhdr...)
 	}
-	fs.wmb.Write(msg)
+	mb.cache.buf = append(mb.cache.buf, msg...)
 
 	// Calculate hash.
-	mb.mu.Lock()
 	mb.hh.Reset()
 	mb.hh.Write(hdr[4:20])
 	mb.hh.Write([]byte(subj))
@@ -1228,76 +1407,118 @@ func (fs *fileStore) writeMsgRecord(seq uint64, subj string, mhdr, msg []byte) (
 	checksum := mb.hh.Sum(nil)
 	// Grab last checksum
 	copy(mb.lchk[0:], checksum)
-	mb.mu.Unlock()
 
+	// Update writethrough cache.
 	// Write to msg record.
-	fs.wmb.Write(checksum)
+	mb.cache.buf = append(mb.cache.buf, checksum...)
+	// Write index
+	mb.cache.idx = append(mb.cache.idx, uint32(index)|hbit)
+	mb.cache.lrl = uint32(rl)
+	if mb.cache.fseq == 0 {
+		mb.cache.fseq = seq
+	}
 
-	return rl, ts, nil
+	// Set cache timestamp for last store.
+	mb.lwts = ts
+
+	// Accounting
+	mb.updateAccounting(seq, ts, rl)
 }
 
-// Will rewrite the message in the underlying store.
-// Both fs and mb locks should be held.
-func (fs *fileStore) eraseMsg(mb *msgBlock, sm *fileStoredMsg) error {
-	if sm == nil || sm.off < 0 {
-		return fmt.Errorf("bad stored message")
-	}
-	// erase contents and rewrite with new hash.
-	if len(sm.hdr) > 0 {
-		sm.hdr = make([]byte, len(sm.hdr))
-		rand.Read(sm.hdr)
-	}
-	if len(sm.msg) > 0 {
-		sm.msg = make([]byte, len(sm.msg))
-		rand.Read(sm.msg)
+// How many bytes pending to be written for this message block.
+func (mb *msgBlock) numWriteBytesPending() int {
+	if mb == nil {
+		return 0
 	}
-	sm.seq, sm.ts = 0, 0
-	chars := []rune("ABCDEFGHIJKLMNOPQRSTUVWXYZ")
-	var b strings.Builder
-	for i := 0; i < len(sm.subj); i++ {
-		b.WriteRune(chars[rand.Intn(len(chars))])
+	var pending int
+	mb.mu.RLock()
+	if mb.mfd != nil && mb.cache != nil {
+		pending = len(mb.cache.buf) - mb.cache.wp
 	}
-	sm.subj = b.String()
-
-	var le = binary.LittleEndian
-	var hdr [msgHdrSize]byte
+	mb.mu.RUnlock()
 
-	rl := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
+	return pending
+}
 
-	le.PutUint32(hdr[0:], uint32(rl))
-	le.PutUint64(hdr[4:], 0)
-	le.PutUint64(hdr[12:], 0)
-	le.PutUint16(hdr[20:], uint16(len(sm.subj)))
+func (mb *msgBlock) clearFlushing() {
+	mb.mu.Lock()
+	if mb.cache != nil {
+		mb.cache.flush = false
+	}
+	mb.mu.Unlock()
+}
 
-	// Now write to underlying buffer.
-	var wmb bytes.Buffer
+// writeBytesPending returns the buffer to be used for writing to the underlying file.
+// This marks we are in flush and will return nil if asked again until cleared.
+func (mb *msgBlock) writeBytesPending() ([]byte, error) {
+	if mb == nil || mb.mfd == nil {
+		return nil, errNoPending
+	}
+	mb.mu.Lock()
+	if mb.cache == nil {
+		mb.mu.Unlock()
+		return nil, errNoCache
+	}
+	if mb.cache.flush {
+		mb.mu.Unlock()
+		return nil, errFlushRunning
+	}
+	buf := mb.cache.buf[mb.cache.wp:]
+	if len(buf) == 0 {
+		mb.mu.Unlock()
+		return nil, errNoPending
+	}
+	mb.cache.flush = true
+	mb.mu.Unlock()
+	return buf, nil
+}
 
-	wmb.Write(hdr[:])
-	wmb.WriteString(sm.subj)
-	wmb.Write(sm.msg)
+// Return the number of bytes in this message block.
+func (mb *msgBlock) numBytes() uint64 {
+	mb.mu.RLock()
+	nb := mb.bytes
+	mb.mu.RUnlock()
+	return nb
+}
 
-	// Calculate hash.
-	mb.hh.Reset()
-	mb.hh.Write(hdr[4:20])
-	mb.hh.Write([]byte(sm.subj))
-	if len(sm.hdr) > 0 {
-		mb.hh.Write(sm.hdr)
+// Update accounting on a write msg.
+// Lock should be held.
+func (mb *msgBlock) updateAccounting(seq uint64, ts int64, rl uint64) {
+	if mb.first.seq == 0 {
+		mb.first.seq = seq
+		mb.first.ts = ts
 	}
-	mb.hh.Write(sm.msg)
-	checksum := mb.hh.Sum(nil)
-	// Write to msg record.
-	wmb.Write(checksum)
+	// Need atomics here for selectMsgBlock speed.
+	atomic.StoreUint64(&mb.last.seq, seq)
+	mb.last.ts = ts
+	mb.bytes += rl
+	mb.msgs++
+}
 
-	mfd, err := os.OpenFile(mb.mfn, os.O_RDWR, 0644)
-	if err != nil {
-		return err
+// Lock should be held.
+func (fs *fileStore) writeMsgRecord(seq uint64, subj string, mhdr, msg []byte) (uint64, int64, error) {
+	var err error
+
+	// Get size for this message.
+	rl := fileStoreMsgSize(subj, mhdr, msg)
+	if rl&hbit != 0 {
+		return 0, 0, ErrMsgTooLarge
+	}
+	// Grab our current last message block.
+	mb := fs.lmb
+	if mb == nil || mb.numBytes()+rl > fs.fcfg.BlockSize {
+		if mb, err = fs.newMsgBlockForWrite(); err != nil {
+			return 0, 0, err
+		}
 	}
-	_, err = mfd.WriteAt(wmb.Bytes(), sm.off)
 
-	mfd.Sync()
-	mfd.Close()
+	// Grab time
+	ts := time.Now().UnixNano()
 
-	return err
+	// Ask msg block to store in write through cache.
+	mb.writeMsgRecord(rl, seq, subj, mhdr, msg, ts)
+
+	return rl, ts, nil
 }
 
 // Sync msg and index files as needed. This is called from a timer.
@@ -1337,37 +1558,22 @@ func (fs *fileStore) syncBlocks() {
 
 // Select the message block where this message should be found.
 // Return nil if not in the set.
+// Read lock should be held.
 func (fs *fileStore) selectMsgBlock(seq uint64) *msgBlock {
-	fs.mu.RLock()
 	// Check for out of range.
 	if seq < fs.state.FirstSeq || seq > fs.state.LastSeq {
-		fs.mu.RUnlock()
 		return nil
 	}
 
-	var smb *msgBlock
-	var needsFlush bool
-
 	// blks are sorted in ascending order.
 	// TODO(dlc) - Can be smarter here, when lots of blks maybe use binary search.
-	// For now this is cache friendly for small to medium num blks.
+	// For now this is cache friendly for small to medium numbers of blks.
 	for _, mb := range fs.blks {
 		if seq <= atomic.LoadUint64(&mb.last.seq) {
-			// This detects if what we may be looking for is staged in the write buffer.
-			if mb == fs.lmb {
-				needsFlush = true
-			}
-			smb = mb
-			break
+			return mb
 		}
 	}
-	fs.mu.RUnlock()
-
-	if needsFlush {
-		fs.flushPendingWritesUnlocked()
-	}
-
-	return smb
+	return nil
 }
 
 // Select the message block where this message should be found.
@@ -1395,21 +1601,6 @@ func (fs *fileStore) selectMsgBlockForStart(minTime time.Time) *msgBlock {
 	return nil
 }
 
-// This will update the cache for a block that is actively being written too.
-func (mb *msgBlock) checkUpdateCache(nbuf []byte) {
-	mb.mu.Lock()
-	defer mb.mu.Unlock()
-
-	// Just return if nothing to do.
-	if mb.cache == nil || len(nbuf) == 0 {
-		return
-	}
-
-	if err := mb.indexCacheBuf(nbuf); err != nil {
-		panic(fmt.Sprintf("Error indexing: %v", err))
-	}
-}
-
 // Index a raw msg buffer.
 // Lock should be held.
 func (mb *msgBlock) indexCacheBuf(buf []byte) error {
@@ -1423,9 +1614,13 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 		// Approximation, may adjust below.
 		fseq = mb.first.seq
 		idx = make([]uint32, 0, mb.msgs)
+		mb.cache = &cache{}
 	} else {
 		fseq = mb.cache.fseq
 		idx = mb.cache.idx
+		if len(idx) == 0 {
+			idx = make([]uint32, 0, mb.msgs)
+		}
 		index = uint32(len(mb.cache.buf))
 		buf = append(mb.cache.buf, buf...)
 	}
@@ -1456,27 +1651,102 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 		// We defer checksum checks to individual msg cache lookups to amortorize costs and
 		// not introduce latency for first message from a newly loaded block.
 		idx = append(idx, index)
-		index += uint32(rl)
-	}
-	if mb.cache == nil {
-		mb.cache = &cache{}
+		mb.cache.lrl = uint32(rl)
+		index += mb.cache.lrl
 	}
 	mb.cache.buf = buf
 	mb.cache.idx = idx
 	mb.cache.fseq = fseq
+	mb.cache.wp += len(buf)
+
+	return nil
+}
+
+// flushPendingWrites for this message block.
+func (mb *msgBlock) flushPendingWrites() error {
+	buf, err := mb.writeBytesPending()
+	if err != nil {
+		return err
+	}
+	defer mb.clearFlushing()
+
+	var lbb, tn int
+	// Append new data to the message block file.
+	for lbb = len(buf); lbb > 0; lbb = len(buf) {
+		n, err := mb.mfd.Write(buf)
+		if err != nil {
+			// FIXME(dlc) - What is the correct behavior here?
+			mb.mu.Lock()
+			mb.removeIndex()
+			mb.mu.Unlock()
+			return err
+		}
+		tn += n
+
+		// Success
+		if int(n) == lbb {
+			break
+		}
+		// Partial write..
+		buf = buf[n:]
+	}
+
+	mb.mu.Lock()
+	if mb.cache == nil {
+		mb.mu.Unlock()
+		return nil
+	}
+	mb.cache.flush = false
+
+	// We did a successful write. If we have active consumers (recent loads) hold onto the cache.
+	if ts := time.Now().UnixNano(); ts < mb.llts || (ts-mb.llts) <= int64(mb.cexp) {
+		mb.cache.wp += tn
+	} else {
+		if lbb <= maxBufReuse {
+			buf = buf[:0]
+		} else {
+			buf = nil
+		}
+		// Reset write pointer.
+		mb.cache.wp = 0
+		// Update our cache offset.
+		mb.cache.off += tn
+		// Place buffer back in the cache structure.
+		mb.cache.buf = buf
+	}
+	mb.mu.Unlock()
+
 	return nil
 }
 
+func (mb *msgBlock) clearLoading() {
+	mb.mu.Lock()
+	mb.loading = false
+	mb.mu.Unlock()
+}
+
 // Will load msgs from disk.
 func (mb *msgBlock) loadMsgs() error {
-	mb.mu.RLock()
+	mb.mu.Lock()
+	// Check to see if we are loading already.
+	if mb.loading {
+		mb.mu.Unlock()
+		return nil
+	}
+	// Check to see if we are full already.
+	if mb.cache != nil && len(mb.cache.idx) == int(mb.msgs) && mb.cache.off == 0 {
+		mb.mu.Unlock()
+		return nil
+	}
+	mb.loading = true
+	defer mb.clearLoading()
 	mfn := mb.mfn
-	hasCache := mb.cache != nil
-	mb.mu.RUnlock()
+	mb.llts = time.Now().UnixNano()
+	mb.mu.Unlock()
 
-	// Someone else may have filled this in by the time we get here.
-	if hasCache {
-		return nil
+	// FIXME(dlc) - We could be smarter here.
+	if mb.numWriteBytesPending() > 0 {
+		mb.flushPendingWrites()
 	}
 
 	// Load in the whole block.
@@ -1486,11 +1756,9 @@ func (mb *msgBlock) loadMsgs() error {
 	}
 
 	mb.mu.Lock()
-	// Someone else may have filled this in by the time we get here.
-	if mb.cache != nil {
-		mb.mu.Unlock()
-		return nil
-	}
+
+	// Make sure this is cleared in case we had a partial.
+	mb.clearCache()
 
 	if err := mb.indexCacheBuf(buf); err != nil {
 		mb.mu.Unlock()
@@ -1512,7 +1780,7 @@ func (mb *msgBlock) fetchMsg(seq uint64) (*fileStoredMsg, error) {
 	var sm *fileStoredMsg
 
 	sm, err := mb.cacheLookup(seq)
-	if err == nil || err != errNoCache {
+	if err == nil || (err != errNoCache && err != errPartialCache) {
 		return sm, err
 	}
 
@@ -1520,20 +1788,26 @@ func (mb *msgBlock) fetchMsg(seq uint64) (*fileStoredMsg, error) {
 	if err := mb.loadMsgs(); err != nil {
 		return nil, err
 	}
-
 	return mb.cacheLookup(seq)
 }
 
 var (
-	errNoCache    = errors.New("no message cache")
-	errBadMsg     = errors.New("malformed or corrupt msg")
-	errDeletedMsg = errors.New("deleted msg")
+	errNoCache      = errors.New("no message cache")
+	errBadMsg       = errors.New("malformed or corrupt message")
+	errDeletedMsg   = errors.New("deleted message")
+	errPartialCache = errors.New("partial cache")
+	errNoPending    = errors.New("message block does not have pending data")
+	errNotReadable  = errors.New("storage directory not readable")
+	errFlushRunning = errors.New("flush is already running")
 )
 
 // Used for marking messages that have had their checksums checked.
-// Also used to signal a message record with headers
+// Used to signal a message record with headers.
 const hbit = 1 << 31
 
+// Used for marking erased messages sequences.
+const ebit = 1 << 63
+
 // Will do a lookup from the cache.
 func (mb *msgBlock) cacheLookup(seq uint64) (*fileStoredMsg, error) {
 	// Currently grab the write lock for optional use of mb.hh. Prefer this for now
@@ -1550,7 +1824,7 @@ func (mb *msgBlock) cacheLookupLocked(seq uint64) (*fileStoredMsg, error) {
 		return nil, errNoCache
 	}
 
-	if seq < mb.cache.fseq || (seq-mb.cache.fseq) >= uint64(len(mb.cache.idx)) {
+	if seq < mb.first.seq || seq < mb.cache.fseq || (seq-mb.cache.fseq) >= uint64(len(mb.cache.idx)) {
 		return nil, ErrStoreMsgNotFound
 	}
 
@@ -1561,18 +1835,28 @@ func (mb *msgBlock) cacheLookupLocked(seq uint64) (*fileStoredMsg, error) {
 		}
 	}
 
-	bi := mb.cache.idx[seq-mb.cache.fseq]
+	// Update cache activity.
+	mb.llts = time.Now().UnixNano()
+
+	bi, _, hashChecked, _ := mb.slotInfo(int(seq - mb.cache.fseq))
 
 	// We use the high bit to denote we have already checked the checksum.
 	var hh hash.Hash64
-	if bi&hbit == 0 {
-		hh = mb.hh
+	if !hashChecked {
+		hh = mb.hh // This will force the hash check in msgFromBuf.
 		mb.cache.idx[seq-mb.cache.fseq] = (bi | hbit)
-	} else {
-		bi &^= hbit
 	}
 
-	buf := mb.cache.buf[bi:]
+	// Check if partial
+	if mb.cache.off > 0 && bi < uint32(mb.cache.off) {
+		buf := mb.cache.buf
+		mb.cache.buf = nil
+		mb.cache.buf = buf
+		return nil, errPartialCache
+	}
+	li := int(bi) - mb.cache.off
+	buf := mb.cache.buf[li:]
+
 	// Parse from the raw buffer.
 	subj, hdr, msg, mseq, ts, err := msgFromBuf(buf, hh)
 	if err != nil {
@@ -1587,9 +1871,9 @@ func (mb *msgBlock) cacheLookupLocked(seq uint64) (*fileStoredMsg, error) {
 		msg:  msg,
 		seq:  seq,
 		ts:   ts,
+		mb:   mb,
 		off:  int64(bi),
 	}
-	atomic.AddUint64(&mb.cgenid, 1)
 	return sm, nil
 }
 
@@ -1600,15 +1884,13 @@ func (fs *fileStore) msgForSeq(seq uint64) (*fileStoredMsg, error) {
 		fs.mu.RUnlock()
 		return nil, ErrStoreClosed
 	}
-	fseq := fs.state.FirstSeq
-	fs.mu.RUnlock()
-
 	// Indicates we want first msg.
 	if seq == 0 {
-		seq = fseq
+		seq = fs.state.FirstSeq
 	}
-
 	mb := fs.selectMsgBlock(seq)
+	fs.mu.RUnlock()
+
 	if mb == nil {
 		var err = ErrStoreEOF
 		fs.mu.RLock()
@@ -1638,7 +1920,7 @@ func msgFromBuf(buf []byte, hh hash.Hash64) (string, []byte, []byte, uint64, int
 	dlen := int(rl) - msgHdrSize
 	slen := int(le.Uint16(hdr[20:]))
 	// Simple sanity check.
-	if dlen < 0 || slen > dlen {
+	if dlen < 0 || slen > dlen || int(rl) > len(buf) {
 		return _EMPTY_, nil, nil, 0, 0, errBadMsg
 	}
 	data := buf[msgHdrSize : msgHdrSize+dlen]
@@ -1657,6 +1939,9 @@ func msgFromBuf(buf []byte, hh hash.Hash64) (string, []byte, []byte, uint64, int
 		}
 	}
 	seq := le.Uint64(hdr[4:])
+	if seq&ebit != 0 {
+		seq = 0
+	}
 	ts := int64(le.Uint64(hdr[12:]))
 	// FIXME(dlc) - We need to not allow appends to the underlying buffer, so we will
 	// fix the capacity. This will cause a copy though in stream:internalSendLoop when
@@ -1693,6 +1978,8 @@ func (fs *fileStore) State() StreamState {
 	return state
 }
 
+const emptyRecordLen = 22 + 8
+
 func fileStoreMsgSize(subj string, hdr, msg []byte) uint64 {
 	if len(hdr) == 0 {
 		// length of the message record (4bytes) + seq(8) + ts(8) + subj_len(2) + subj + msg + hash(8)
@@ -1702,6 +1989,10 @@ func fileStoreMsgSize(subj string, hdr, msg []byte) uint64 {
 	return uint64(22 + len(subj) + 4 + len(hdr) + len(msg) + 8)
 }
 
+func fileStoreMsgSizeEstimate(slen, maxPayload int) uint64 {
+	return uint64(emptyRecordLen + slen + 4 + maxPayload)
+}
+
 // Lock should not be held.
 func (fs *fileStore) flushPendingWritesUnlocked() error {
 	fs.mu.Lock()
@@ -1713,28 +2004,16 @@ func (fs *fileStore) flushPendingWritesUnlocked() error {
 func (fs *fileStore) flushPendingWrites() error {
 	mb := fs.lmb
 	if mb == nil || mb.mfd == nil {
-		return fmt.Errorf("filestore does not have last message block")
+		return errNoPending
 	}
-	// Append new data to the message block file.
-	for lbb := fs.wmb.Len(); lbb > 0; lbb = fs.wmb.Len() {
-		n, err := fs.wmb.WriteTo(mb.mfd)
-		if err != nil {
-			mb.mu.Lock()
-			mb.removeIndex()
-			mb.mu.Unlock()
+	err := mb.flushPendingWrites()
+	if err != nil {
+		switch err {
+		case errNoCache, errNoPending:
+			err = nil
+		default:
 			return err
 		}
-
-		// Update the cache if needed.
-		mb.checkUpdateCache(fs.wmb.Bytes()[:n])
-
-		if int(n) != lbb {
-			fs.wmb.Truncate(int(n))
-		} else if lbb <= maxBufReuse {
-			fs.wmb.Reset()
-		} else {
-			fs.wmb = &bytes.Buffer{}
-		}
 	}
 	// Now index info
 	return mb.writeIndexInfo()
@@ -1938,7 +2217,6 @@ func (fs *fileStore) Purge() uint64 {
 	}
 
 	fs.blks = nil
-	fs.wmb = &bytes.Buffer{}
 	fs.lmb = nil
 
 	// Move the msgs directory out of the way, will delete out of band.
@@ -2023,7 +2301,7 @@ func (mb *msgBlock) dirtyClose() {
 	}
 	mb.mu.Lock()
 	// Close cache
-	mb.cache = nil
+	mb.clearCache()
 	// Quit our loops.
 	if mb.qch != nil {
 		close(mb.qch)
@@ -2046,7 +2324,7 @@ func (mb *msgBlock) close(sync bool) {
 	}
 	mb.mu.Lock()
 	// Close cache
-	mb.cache = nil
+	mb.clearCache()
 	// Quit our loops.
 	if mb.qch != nil {
 		close(mb.qch)
@@ -2094,7 +2372,6 @@ func (fs *fileStore) Stop() error {
 	close(fs.qch)
 
 	err := fs.flushPendingWrites()
-	fs.wmb = &bytes.Buffer{}
 	fs.lmb = nil
 
 	fs.closeAllMsgBlocks(true)
@@ -2196,9 +2473,7 @@ func (fs *fileStore) streamSnapshot(w io.WriteCloser, blks []*msgBlock, includeC
 	}
 
 	// Now do messages themselves.
-	fs.mu.Lock()
-	lmb := fs.lmb
-	fs.mu.Unlock()
+	lmb := fs.lastMsgBlock()
 
 	// Can't use join path here, zip only recognizes relative paths with forward slashes.
 	msgPre := msgDir + "/"
@@ -2316,6 +2591,13 @@ func (fs *fileStore) Snapshot(deadline time.Duration, checkMsgs, includeConsumer
 	return &SnapshotResult{pr, blkSize, len(blks)}, nil
 }
 
+// Helper to return the config.
+func (fs *fileStore) fileStoreConfig() FileStoreConfig {
+	fs.mu.RLock()
+	defer fs.mu.RUnlock()
+	return fs.fcfg
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 // Consumers
 ////////////////////////////////////////////////////////////////////////////////
@@ -2330,8 +2612,6 @@ type consumerFileStore struct {
 	ifd    *os.File
 	lwsz   int64
 	hh     hash.Hash64
-	fch    chan struct{}
-	qch    chan struct{}
 	closed bool
 }
 
@@ -2356,8 +2636,6 @@ func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerSt
 		name: name,
 		odir: odir,
 		ifn:  path.Join(odir, consumerState),
-		fch:  make(chan struct{}),
-		qch:  make(chan struct{}),
 	}
 	key := sha256.Sum256([]byte(fs.cfg.Name + "/" + name))
 	hh, err := highwayhash.New64(key[:])
diff --git a/server/jetstream.go b/server/jetstream.go
index da3eeb1c..bc0eb12c 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -26,6 +26,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"syscall"
 
 	"github.com/minio/highwayhash"
 	"github.com/nats-io/nats-server/v2/server/sysmem"
@@ -101,13 +102,15 @@ func (s *Server) EnableJetStream(config *JetStreamConfig) error {
 	s.Noticef("Starting JetStream")
 	if config == nil || config.MaxMemory <= 0 || config.MaxStore <= 0 {
 		var storeDir string
-		s.Debugf("JetStream creating dynamic configuration - 75%% of system memory, %s disk", FriendlyBytes(JetStreamMaxStoreDefault))
+		var maxStore int64
 		if config != nil {
 			storeDir = config.StoreDir
+			maxStore = config.MaxStore
 		}
-		config = s.dynJetStreamConfig(storeDir)
+		config = s.dynJetStreamConfig(storeDir, maxStore)
+		s.Debugf("JetStream creating dynamic configuration - %s memory, %s disk", FriendlyBytes(config.MaxMemory), FriendlyBytes(config.MaxStore))
 	}
-	// Copy, don't change callers.
+	// Copy, don't change callers version.
 	cfg := *config
 	if cfg.StoreDir == "" {
 		cfg.StoreDir = filepath.Join(os.TempDir(), JetStreamStoreDir)
@@ -827,6 +830,13 @@ func (jsa *jsAccount) checkBytesLimits(addBytes int64, storage StorageType) erro
 	return nil
 }
 
+func (jsa *jsAccount) acc() *Account {
+	jsa.mu.RLock()
+	acc := jsa.account
+	jsa.mu.RUnlock()
+	return acc
+}
+
 // Delete the JetStream resources.
 func (jsa *jsAccount) delete() {
 	var streams []*Stream
@@ -922,7 +932,7 @@ const (
 )
 
 // Dynamically create a config with a tmp based directory (repeatable) and 75% of system memory.
-func (s *Server) dynJetStreamConfig(storeDir string) *JetStreamConfig {
+func (s *Server) dynJetStreamConfig(storeDir string, maxStore int64) *JetStreamConfig {
 	jsc := &JetStreamConfig{}
 	if storeDir != "" {
 		jsc.StoreDir = filepath.Join(storeDir, JetStreamStoreDir)
@@ -930,7 +940,22 @@ func (s *Server) dynJetStreamConfig(storeDir string) *JetStreamConfig {
 		tdir, _ := ioutil.TempDir(os.TempDir(), "nats-jetstream-storedir-")
 		jsc.StoreDir = filepath.Join(tdir, JetStreamStoreDir)
 	}
-	jsc.MaxStore = JetStreamMaxStoreDefault
+
+	if maxStore > 0 {
+		jsc.MaxStore = maxStore
+	} else {
+		if _, err := os.Stat(jsc.StoreDir); os.IsNotExist(err) {
+			os.MkdirAll(jsc.StoreDir, 0755)
+		}
+		var fs syscall.Statfs_t
+		if err := syscall.Statfs(jsc.StoreDir, &fs); err == nil {
+			// Estimate 75% of available storage.
+			jsc.MaxStore = int64(fs.Bavail * uint64(fs.Bsize) / 4 * 3)
+		} else {
+			// Used 1TB default as a guess if all else fails.
+			jsc.MaxStore = JetStreamMaxStoreDefault
+		}
+	}
 	// Estimate to 75% of total memory if we can determine system memory.
 	if sysMem := sysmem.Memory(); sysMem > 0 {
 		jsc.MaxMemory = sysMem / 4 * 3
diff --git a/server/memstore.go b/server/memstore.go
index 82da4d18..bcc845c1 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -317,7 +317,7 @@ func (ms *memStore) EraseMsg(seq uint64) (bool, error) {
 	return removed, nil
 }
 
-// Performs logic tp update first sequence number.
+// Performs logic to update first sequence number.
 // Lock should be held.
 func (ms *memStore) updateFirstSeq(seq uint64) {
 	if seq != ms.state.FirstSeq {
diff --git a/server/store.go b/server/store.go
index cf5556cc..06f7def6 100644
--- a/server/store.go
+++ b/server/store.go
@@ -50,6 +50,8 @@ var (
 	ErrStoreSnapshotInProgress = errors.New("snapshot in progress")
 	// ErrMsgTooBig is returned when a message is considered too large.
 	ErrMsgTooLarge = errors.New("message to large")
+	// ErrStoreWrongType is for when you access the wrong storage type.
+	ErrStoreWrongType = errors.New("wrong storage type")
 )
 
 type StreamStore interface {
diff --git a/server/stream.go b/server/stream.go
index 86594d31..de0a460a 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -169,6 +169,11 @@ func (a *Account) AddStreamWithStore(config *StreamConfig, fsConfig *FileStoreCo
 	fsCfg := fsConfig
 	if fsCfg == nil {
 		fsCfg = &FileStoreConfig{}
+		// If we are file based and not explicitly configured
+		// we may be able to auto-tune based on max msgs or bytes.
+		if cfg.Storage == FileStorage {
+			mset.autoTuneFileStorageBlockSize(fsCfg)
+		}
 	}
 	fsCfg.StoreDir = storeDir
 	if err := mset.setupStore(fsCfg); err != nil {
@@ -201,6 +206,71 @@ func (a *Account) AddStreamWithStore(config *StreamConfig, fsConfig *FileStoreCo
 	return mset, nil
 }
 
+// Helper to determine the max msg size for this stream if file based.
+func (mset *Stream) maxMsgSize() uint64 {
+	maxMsgSize := mset.config.MaxMsgSize
+	if maxMsgSize <= 0 {
+		// Pull from the account.
+		if mset.jsa != nil {
+			if acc := mset.jsa.acc(); acc != nil {
+				acc.mu.RLock()
+				maxMsgSize = acc.mpay
+				acc.mu.RUnlock()
+			}
+		}
+		// If all else fails use default.
+		if maxMsgSize <= 0 {
+			maxMsgSize = MAX_PAYLOAD_SIZE
+		}
+	}
+	// Now determine an estimation for the subjects etc.
+	maxSubject := -1
+	for _, subj := range mset.config.Subjects {
+		if subjectIsLiteral(subj) {
+			if len(subj) > maxSubject {
+				maxSubject = len(subj)
+			}
+		}
+	}
+	if maxSubject < 0 {
+		const defaultMaxSubject = 256
+		maxSubject = defaultMaxSubject
+	}
+	// filestore will add in estimates for record headers, etc.
+	return fileStoreMsgSizeEstimate(maxSubject, int(maxMsgSize))
+}
+
+// If we are file based and the file storage config was not explicitly set
+// we can autotune block sizes to better match. Our target will be to store 125%
+// of the theoretical limit. We will round up to nearest 100 bytes as well.
+func (mset *Stream) autoTuneFileStorageBlockSize(fsCfg *FileStoreConfig) {
+	var totalEstSize uint64
+
+	// MaxBytes will take precedence for now.
+	if mset.config.MaxBytes > 0 {
+		totalEstSize = uint64(mset.config.MaxBytes)
+	} else if mset.config.MaxMsgs > 0 {
+		// Determine max message size to estimate.
+		totalEstSize = mset.maxMsgSize() * uint64(mset.config.MaxMsgs)
+	} else {
+		// If nothing set will let underlying filestore determine blkSize.
+		return
+	}
+
+	blkSize := (totalEstSize / 4) + 1 // (25% overhead)
+	// Round up to nearest 100
+	if m := blkSize % 100; m != 0 {
+		blkSize += 100 - m
+	}
+	if blkSize < FileStoreMinBlkSize {
+		blkSize = FileStoreMinBlkSize
+	}
+	if blkSize > FileStoreMaxBlkSize {
+		blkSize = FileStoreMaxBlkSize
+	}
+	fsCfg.BlockSize = uint64(blkSize)
+}
+
 // rebuildDedupe will rebuild any dedupe structures needed after recovery of a stream.
 // Lock not needed, only called during initialization.
 // TODO(dlc) - Might be good to know if this should be checked at all for streams with no
@@ -407,6 +477,16 @@ func (mset *Stream) Config() StreamConfig {
 	return mset.config
 }
 
+func (mset *Stream) FileStoreConfig() (FileStoreConfig, error) {
+	mset.mu.Lock()
+	defer mset.mu.Unlock()
+	fs, ok := mset.store.(*fileStore)
+	if !ok {
+		return FileStoreConfig{}, ErrStoreWrongType
+	}
+	return fs.fileStoreConfig(), nil
+}
+
 // Delete deletes a stream from the owning account.
 func (mset *Stream) Delete() error {
 	mset.mu.Lock()


Test patch:
diff --git a/conf/parse_test.go b/conf/parse_test.go
index 72c5d8e9..68cba37a 100644
--- a/conf/parse_test.go
+++ b/conf/parse_test.go
@@ -142,44 +142,20 @@ func TestBcryptVariable(t *testing.T) {
 var easynum = `
 k = 8k
 kb = 4kb
-ki = 3ki
-kib = 4ki
 m = 1m
 mb = 2MB
-mi = 2Mi
-mib = 64MiB
 g = 2g
 gb = 22GB
-gi = 22Gi
-gib = 22GiB
-tb = 22TB
-ti = 22Ti
-tib = 22TiB
-pb = 22PB
-pi = 22Pi
-pib = 22PiB
 `
 
 func TestConvenientNumbers(t *testing.T) {
 	ex := map[string]interface{}{
 		"k":  int64(8 * 1000),
 		"kb": int64(4 * 1024),
-		"ki": int64(3 * 1024),
-		"kib": int64(4 * 1024),
 		"m":  int64(1000 * 1000),
 		"mb": int64(2 * 1024 * 1024),
-		"mi": int64(2 * 1024 * 1024),
-		"mib": int64(64 * 1024 * 1024),
 		"g":  int64(2 * 1000 * 1000 * 1000),
 		"gb": int64(22 * 1024 * 1024 * 1024),
-		"gi": int64(22 * 1024 * 1024 * 1024),
-		"gib": int64(22 * 1024 * 1024 * 1024),
-		"tb": int64(22 * 1024 * 1024 * 1024 * 1024),
-		"ti": int64(22 * 1024 * 1024 * 1024 * 1024),
-		"tib": int64(22 * 1024 * 1024 * 1024 * 1024),
-		"pb": int64(22 * 1024 * 1024 * 1024 * 1024 * 1024),
-		"pi": int64(22 * 1024 * 1024 * 1024 * 1024 * 1024),
-		"pib": int64(22 * 1024 * 1024 * 1024 * 1024 * 1024),
 	}
 	test(t, easynum, ex)
 }
diff --git a/server/filestore_test.go b/server/filestore_test.go
index ffd2ccdd..6e079b9e 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -60,6 +60,7 @@ func TestFileStoreBasics(t *testing.T) {
 	if state.Bytes != expectedSize {
 		t.Fatalf("Expected %d bytes, got %d", expectedSize, state.Bytes)
 	}
+
 	nsubj, _, nmsg, _, err := fs.LoadMsg(2)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
@@ -74,6 +75,27 @@ func TestFileStoreBasics(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
 	}
+
+	remove := func(seq, expectedMsgs uint64) {
+		t.Helper()
+		removed, err := fs.RemoveMsg(seq)
+		if err != nil {
+			t.Fatalf("Got an error on remove of %d: %v", seq, err)
+		}
+		if !removed {
+			t.Fatalf("Expected remove to return true for %d", seq)
+		}
+		if state := fs.State(); state.Msgs != expectedMsgs {
+			t.Fatalf("Expected %d msgs, got %d", expectedMsgs, state.Msgs)
+		}
+	}
+
+	// Remove first
+	remove(1, 4)
+	// Remove last
+	remove(5, 3)
+	// Remove a middle
+	remove(3, 2)
 }
 
 func TestFileStoreMsgHeaders(t *testing.T) {
@@ -279,6 +301,123 @@ func TestFileStoreSkipMsg(t *testing.T) {
 	if state.FirstSeq != uint64(numSkips+1) || state.LastSeq != uint64(numSkips+4) {
 		t.Fatalf("Expected first to be %d and last to be %d. got first %d and last %d", numSkips+1, numSkips+4, state.FirstSeq, state.LastSeq)
 	}
+
+	// Make sure we recover same state.
+	fs.Stop()
+
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	state = fs.State()
+	if state.Msgs != 2 {
+		t.Fatalf("Expected %d msgs, got %d", 2, state.Msgs)
+	}
+	if state.FirstSeq != uint64(numSkips+1) || state.LastSeq != uint64(numSkips+4) {
+		t.Fatalf("Expected first to be %d and last to be %d. got first %d and last %d", numSkips+1, numSkips+4, state.FirstSeq, state.LastSeq)
+	}
+
+	subj, _, msg, _, err := fs.LoadMsg(11)
+	if err != nil {
+		t.Fatalf("Unexpected error looking up seq 11: %v", err)
+	}
+	if subj != "zzz" || string(msg) != "Hello World!" {
+		t.Fatalf("Message did not match")
+	}
+
+	fs.SkipMsg()
+	nseq, _, err := fs.StoreMsg("AAA", nil, []byte("Skip?"))
+	if err != nil {
+		t.Fatalf("Unexpected error looking up seq 11: %v", err)
+	}
+	if nseq != 16 {
+		t.Fatalf("Expected seq of %d but got %d", 16, nseq)
+	}
+
+	// Make sure we recover same state.
+	fs.Stop()
+
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	subj, _, msg, _, err = fs.LoadMsg(nseq)
+	if err != nil {
+		fmt.Printf("fs is %+v\n", fs)
+		t.Fatalf("Unexpected error looking up seq %d: %v", nseq, err)
+	}
+	if subj != "AAA" || string(msg) != "Skip?" {
+		t.Fatalf("Message did not match")
+	}
+}
+
+func TestFileStoreWriteExpireWrite(t *testing.T) {
+	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	cexp := 10 * time.Millisecond
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 256, CacheExpire: cexp},
+		StreamConfig{Name: "zzz", Storage: FileStorage},
+	)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	toSend := 10
+	for i := 0; i < toSend; i++ {
+		fs.StoreMsg("zzz", nil, []byte("Hello World!"))
+	}
+
+	// Wait for write cache portion to go to zero.
+	checkFor(t, time.Second, 20*time.Millisecond, func() error {
+		if csz := fs.cacheSize(); csz != 0 {
+			return fmt.Errorf("cache size not 0, got %s", FriendlyBytes(int64(csz)))
+		}
+		return nil
+	})
+
+	for i := 0; i < toSend; i++ {
+		fs.StoreMsg("zzz", nil, []byte("Hello World! - 22"))
+	}
+
+	if state := fs.State(); state.Msgs != uint64(toSend*2) {
+		t.Fatalf("Expected %d msgs, got %d", toSend*2, state.Msgs)
+	}
+
+	// Make sure we recover same state.
+	fs.Stop()
+
+	fs, err = newFileStore(FileStoreConfig{StoreDir: storeDir, BlockSize: 256}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	if state := fs.State(); state.Msgs != uint64(toSend*2) {
+		t.Fatalf("Expected %d msgs, got %d", toSend*2, state.Msgs)
+	}
+
+	// Now load them in and check.
+	for i := 1; i <= toSend*2; i++ {
+		subj, _, msg, _, err := fs.LoadMsg(uint64(i))
+		if err != nil {
+			t.Fatalf("Unexpected error looking up seq 11: %v", err)
+		}
+		str := "Hello World!"
+		if i > toSend {
+			str = "Hello World! - 22"
+		}
+		if subj != "zzz" || string(msg) != str {
+			t.Fatalf("Message did not match")
+		}
+	}
 }
 
 func TestFileStoreMsgLimit(t *testing.T) {
@@ -437,7 +576,7 @@ func TestFileStoreTimeStamps(t *testing.T) {
 	for seq := uint64(1); seq <= 10; seq++ {
 		_, _, _, ts, err := fs.LoadMsg(seq)
 		if err != nil {
-			t.Fatalf("Unexpected error looking up msg: %v", err)
+			t.Fatalf("Unexpected error looking up msg [%d]: %v", seq, err)
 		}
 		// These should be different
 		if ts <= last {
@@ -685,7 +824,7 @@ func TestFileStoreAgeLimitRecovery(t *testing.T) {
 	defer os.RemoveAll(storeDir)
 
 	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir, ReadCacheExpire: 1 * time.Millisecond},
+		FileStoreConfig{StoreDir: storeDir, CacheExpire: 1 * time.Millisecond},
 		StreamConfig{Name: "zzz", Storage: FileStorage, MaxAge: maxAge},
 	)
 	if err != nil {
@@ -799,6 +938,7 @@ func TestFileStoreEraseMsg(t *testing.T) {
 
 	subj, msg := "foo", []byte("Hello World")
 	fs.StoreMsg(subj, nil, msg)
+	fs.StoreMsg(subj, nil, msg) // To keep block from being deleted.
 	_, _, smsg, _, err := fs.LoadMsg(1)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
@@ -808,22 +948,25 @@ func TestFileStoreEraseMsg(t *testing.T) {
 	}
 	// Hold for offset check later.
 	sm, _ := fs.msgForSeq(1)
-
 	if removed, _ := fs.EraseMsg(1); !removed {
 		t.Fatalf("Expected erase msg to return success")
 	}
 	if sm2, _ := fs.msgForSeq(1); sm2 != nil {
 		t.Fatalf("Expected msg to be erased")
 	}
+	if err := fs.flushPendingWritesUnlocked(); err != nil {
+		t.Fatalf("Unexpected error flushing: %v", err)
+	}
 
 	// Now look on disk as well.
 	rl := fileStoreMsgSize(subj, nil, msg)
 	buf := make([]byte, rl)
 	fp, err := os.Open(path.Join(storeDir, msgDir, fmt.Sprintf(blkScan, 1)))
 	if err != nil {
-		t.Fatalf("Error opening msgs file: %v", err)
+		t.Fatalf("Error opening msg block file: %v", err)
 	}
 	defer fp.Close()
+
 	fp.ReadAt(buf, sm.off)
 	nsubj, _, nmsg, seq, ts, err := msgFromBuf(buf, nil)
 	if err != nil {
@@ -832,7 +975,7 @@ func TestFileStoreEraseMsg(t *testing.T) {
 	if nsubj == subj {
 		t.Fatalf("Expected the subjects to be different")
 	}
-	if seq != 0 {
+	if seq != 0 && seq&ebit == 0 {
 		t.Fatalf("Expected seq to be 0, marking as deleted, got %d", seq)
 	}
 	if ts != 0 {
@@ -1142,7 +1285,7 @@ func TestFileStoreReadCache(t *testing.T) {
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
 
-	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir, ReadCacheExpire: 50 * 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)
 	}
@@ -1155,6 +1298,14 @@ func TestFileStoreReadCache(t *testing.T) {
 		fs.StoreMsg(subj, nil, msg)
 	}
 
+	// Wait for write cache portion to go to zero.
+	checkFor(t, time.Second, 10*time.Millisecond, func() error {
+		if csz := fs.cacheSize(); csz != 0 {
+			return fmt.Errorf("cache size not 0, got %s", FriendlyBytes(int64(csz)))
+		}
+		return nil
+	})
+
 	fs.LoadMsg(1)
 	if csz := fs.cacheSize(); csz != totalBytes {
 		t.Fatalf("Expected all messages to be cached, got %d vs %d", csz, totalBytes)
@@ -1670,6 +1821,58 @@ func TestFileStoreReadBackMsgPerf(t *testing.T) {
 	fmt.Printf("Time to load second msg [%d] = %v\n", index+2, time.Since(start))
 }
 
+// This test is testing an upper level stream with a message or byte limit.
+// Even though this is 1, any limit would trigger same behavior once the limit was reached
+// and we were adding and removing.
+func TestFileStoreStoreLimitRemovePerf(t *testing.T) {
+	// Comment out to run, holding place for now.
+	t.SkipNow()
+
+	subj, msg := "foo", make([]byte, 1024-33)
+	for i := 0; i < len(msg); i++ {
+		msg[i] = 'D'
+	}
+	storedMsgSize := fileStoreMsgSize(subj, nil, msg)
+
+	// 1GB
+	toStore := 1 * 1024 * 1024 * 1024 / storedMsgSize
+
+	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{Name: "zzz", Storage: FileStorage},
+	)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	fmt.Printf("storing and removing (limit 1) %d msgs of %s each, totalling %s\n",
+		toStore,
+		FriendlyBytes(int64(storedMsgSize)),
+		FriendlyBytes(int64(toStore*storedMsgSize)),
+	)
+
+	start := time.Now()
+	for i := 0; i < int(toStore); i++ {
+		seq, _, err := fs.StoreMsg(subj, nil, msg)
+		if err != nil {
+			t.Fatalf("Unexpected error storing message: %v", err)
+		}
+		if i > 0 {
+			fs.RemoveMsg(seq - 1)
+		}
+	}
+	fs.Stop()
+
+	tt := time.Since(start)
+	fmt.Printf("time to store and remove all messages is %v\n", tt)
+	fmt.Printf("%.0f msgs/sec\n", float64(toStore)/tt.Seconds())
+	fmt.Printf("%s per sec\n", FriendlyBytes(int64(float64(toStore*storedMsgSize)/tt.Seconds())))
+}
+
 func TestFileStoreConsumerPerf(t *testing.T) {
 	// Comment out to run, holding place for now.
 	t.SkipNow()
diff --git a/test/jetstream_test.go b/test/jetstream_test.go
index ee70a253..dbec48a7 100644
--- a/test/jetstream_test.go
+++ b/test/jetstream_test.go
@@ -303,6 +303,57 @@ func TestJetStreamAddStreamDiscardNew(t *testing.T) {
 	}
 }
 
+func TestJetStreamAutoTuneFSConfig(t *testing.T) {
+	s := RunRandClientPortServer()
+	defer s.Shutdown()
+
+	jsconfig := &server.JetStreamConfig{MaxMemory: -1, MaxStore: 128 * 1024 * 1024 * 1024 * 1024}
+	if err := s.EnableJetStream(jsconfig); err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer os.RemoveAll(config.StoreDir)
+	}
+
+	maxMsgSize := int32(512)
+	streamConfig := func(name string, maxMsgs, maxBytes int64) *server.StreamConfig {
+		t.Helper()
+		cfg := &server.StreamConfig{Name: name, MaxMsgSize: maxMsgSize, Storage: server.FileStorage}
+		if maxMsgs > 0 {
+			cfg.MaxMsgs = maxMsgs
+		}
+		if maxBytes > 0 {
+			cfg.MaxBytes = maxBytes
+		}
+		return cfg
+	}
+
+	acc := s.GlobalAccount()
+
+	testBlkSize := func(subject string, maxMsgs, maxBytes int64, expectedBlkSize uint64) {
+		t.Helper()
+		mset, err := acc.AddStream(streamConfig(subject, maxMsgs, maxBytes))
+		if err != nil {
+			t.Fatalf("Unexpected error adding stream: %v", err)
+		}
+		defer mset.Delete()
+		fsCfg, err := mset.FileStoreConfig()
+		if err != nil {
+			t.Fatalf("Unexpected error retrieving file store: %v", err)
+		}
+		if fsCfg.BlockSize != expectedBlkSize {
+			t.Fatalf("Expected auto tuned block size to be %d, got %d", expectedBlkSize, fsCfg.BlockSize)
+		}
+	}
+
+	testBlkSize("foo", 1, 0, server.FileStoreMinBlkSize)
+	testBlkSize("foo", 1, 512, server.FileStoreMinBlkSize)
+	testBlkSize("foo", 1, 1024*1024, 262200)
+	testBlkSize("foo", 1, 8*1024*1024, 2097200)
+	testBlkSize("foo_bar_baz", -1, 32*1024*1024*1024*1024, server.FileStoreMaxBlkSize)
+}
+
 func TestJetStreamPubAck(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer s.Shutdown()
@@ -2386,8 +2437,8 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 					t.Fatalf("Expected ack reply pending of %d, got %d", ep, pending)
 				}
 				// Now check consumer info.
-				if info := o.Info(); int(info.NumPending) != ep {
-					t.Fatalf("Expected consumer info pending of %d, got %d", ep, info.NumPending)
+				if info := o.Info(); int(info.NumStreamPending) != ep {
+					t.Fatalf("Expected consumer info pending of %d, got %d", ep, info.NumStreamPending)
 				}
 			}
 
@@ -2438,8 +2489,8 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 			for i := 0; i < toSend; i++ {
 				sendStreamMsg(t, nc, "foo.33", "Hello World!")
 			}
-			if info := o.Info(); info.NumPending != 0 {
-				t.Fatalf("Expected no pending, got %d", info.NumPending)
+			if info := o.Info(); info.NumStreamPending != 0 {
+				t.Fatalf("Expected no pending, got %d", info.NumStreamPending)
 			}
 			// Now send one message that will match us.
 			sendStreamMsg(t, nc, "foo.22", "Hello World!")
@@ -2519,8 +2570,8 @@ func TestJetStreamAckReplyStreamPendingWithAcks(t *testing.T) {
 			}
 			defer o.Delete()
 
-			if info := o.Info(); int(info.NumPending) != toSend {
-				t.Fatalf("Expected consumer info pending of %d, got %d", toSend, info.NumPending)
+			if info := o.Info(); int(info.NumStreamPending) != toSend {
+				t.Fatalf("Expected consumer info pending of %d, got %d", toSend, info.NumStreamPending)
 			}
 
 			sub, _ := nc.SubscribeSync(dsubj)
@@ -2533,10 +2584,10 @@ func TestJetStreamAckReplyStreamPendingWithAcks(t *testing.T) {
 			})
 
 			// Should be zero.
-			if info := o.Info(); int(info.NumPending) != 0 {
-				t.Fatalf("Expected consumer info pending of %d, got %d", 0, info.NumPending)
-			} else if info.NumAckPending != toSend {
-				t.Fatalf("Expected %d to be pending acks, got %d", toSend, info.NumAckPending)
+			if info := o.Info(); int(info.NumStreamPending) != 0 {
+				t.Fatalf("Expected consumer info pending of %d, got %d", 0, info.NumStreamPending)
+			} else if info.NumPending != toSend {
+				t.Fatalf("Expected %d to be pending acks, got %d", toSend, info.NumPending)
 			}
 		})
 	}
@@ -5341,8 +5392,8 @@ func TestJetStreamStreamPurgeWithConsumer(t *testing.T) {
 			if state.AckFloor.ConsumerSeq != 50 {
 				t.Fatalf("Expected ack floor of 50, got %d", state.AckFloor.ConsumerSeq)
 			}
-			if state.NumAckPending != 25 {
-				t.Fatalf("Expected len(pending) to be 25, got %d", state.NumAckPending)
+			if state.NumPending != 25 {
+				t.Fatalf("Expected len(pending) to be 25, got %d", state.NumPending)
 			}
 			// Now do purge.
 			mset.Purge()
@@ -5353,8 +5404,8 @@ func TestJetStreamStreamPurgeWithConsumer(t *testing.T) {
 			// Pending should be cleared, and stream sequences should have been set
 			// to the total messages before purge + 1.
 			state = o.Info()
-			if state.NumAckPending != 0 {
-				t.Fatalf("Expected no pending, got %d", state.NumAckPending)
+			if state.NumPending != 0 {
+				t.Fatalf("Expected no pending, got %d", state.NumPending)
 			}
 			if state.Delivered.StreamSeq != 100 {
 				t.Fatalf("Expected to have setseq now at next seq of 100, got %d", state.Delivered.StreamSeq)
@@ -5441,8 +5492,8 @@ func TestJetStreamStreamPurgeWithConsumerAndRedelivery(t *testing.T) {
 			// Pending should be cleared, and stream sequences should have been set
 			// to the total messages before purge + 1.
 			state := o.Info()
-			if state.NumAckPending != 0 {
-				t.Fatalf("Expected no pending, got %d", state.NumAckPending)
+			if state.NumPending != 0 {
+				t.Fatalf("Expected no pending, got %d", state.NumPending)
 			}
 			if state.Delivered.StreamSeq != 100 {
 				t.Fatalf("Expected to have setseq now at next seq of 100, got %d", state.Delivered.StreamSeq)
@@ -6437,9 +6488,14 @@ type info struct {
 func TestJetStreamSimpleFileStorageRecovery(t *testing.T) {
 	base := runtime.NumGoroutine()
 
-	s := RunBasicJetStreamServer()
+	s := RunRandClientPortServer()
 	defer s.Shutdown()
 
+	jsconfig := &server.JetStreamConfig{MaxMemory: 128 * 1024 * 1024, MaxStore: 32 * 1024 * 1024 * 1024}
+	if err := s.EnableJetStream(jsconfig); err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
 	if config := s.JetStreamConfig(); config != nil {
 		defer os.RemoveAll(config.StoreDir)
 	}
@@ -6503,11 +6559,6 @@ func TestJetStreamSimpleFileStorageRecovery(t *testing.T) {
 	pusage := acc.JetStreamUsage()
 
 	// Shutdown the server. Restart and make sure things come back.
-	// Capture port since it was dynamic.
-	u, _ := url.Parse(s.ClientURL())
-	port, _ := strconv.Atoi(u.Port())
-	sd := s.JetStreamConfig().StoreDir
-
 	s.Shutdown()
 
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
@@ -6518,9 +6569,13 @@ func TestJetStreamSimpleFileStorageRecovery(t *testing.T) {
 		return nil
 	})
 
-	s = RunJetStreamServerOnPort(port, sd)
+	s = RunRandClientPortServer()
 	defer s.Shutdown()
 
+	if err := s.EnableJetStream(jsconfig); err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+
 	acc = s.GlobalAccount()
 
 	nusage := acc.JetStreamUsage()
@@ -7688,7 +7743,7 @@ func TestJetStreamNextMsgNoInterest(t *testing.T) {
 			}
 			nc.Flush()
 			ostate := o.Info()
-			if ostate.AckFloor.StreamSeq != 11 || ostate.NumAckPending > 0 {
+			if ostate.AckFloor.StreamSeq != 11 || ostate.NumPending > 0 {
 				t.Fatalf("Inconsistent ack state: %+v", ostate)
 			}
 		})