Repository: nats-io/nats-server
Pull request #1697 (2020-11-10 14:23:54)

Updates to consumer store to support delta updates.
In preparation for clustering we need to have the consumer filestore update state with deltas vs original design.

Not sure I will stick with this long term since we could do delta updates for clustering at the consumer level above the store implementation but this includes some fixes and stability improvements and moves us closer to clustering.

Signed-off-by: Derek Collison <derek@nats.io>

 
/cc @nats-io/core


kozlovic: @derekcollison I think you have have worked on a branch that got a bit stale. There is a conflict on filestore.go, but also, if I try to merge that with a branch (to test effects on mqtt branch), I get a compilation error with consumer.go where you recently added forceExpirePending() which does `off := now - o.pending[expired[0]]`. Should you try to rebase from master first?
derekcollison: I can rebase, looking at Phil's report from last night as well.

Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 5588da09..ca32c49a 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -171,14 +171,14 @@ type Consumer struct {
 	dseq              uint64
 	adflr             uint64
 	asflr             uint64
-	spending          uint64
+	sgap              uint64
 	dsubj             string
 	rlimit            *rate.Limiter
 	reqSub            *subscription
 	ackSub            *subscription
 	ackReplyT         string
 	nextMsgSubj       string
-	pending           map[uint64]int64
+	pending           map[uint64]*Pending
 	ptmr              *time.Timer
 	rdq               []uint64
 	rdc               map[uint64]uint64
@@ -191,7 +191,6 @@ type Consumer struct {
 	filterWC          bool
 	dtmr              *time.Timer
 	dthresh           time.Duration
-	fch               chan struct{}
 	qch               chan struct{}
 	inch              chan bool
 	sfreq             int32
@@ -385,7 +384,6 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 		dsubj:   config.DeliverSubject,
 		active:  true,
 		qch:     make(chan struct{}),
-		fch:     make(chan struct{}, 1),
 		sfreq:   int32(sampleFreq),
 		maxdc:   uint64(config.MaxDeliver),
 		created: time.Now().UTC(),
@@ -529,8 +527,9 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 
 	// Now start up Go routine to deliver msgs.
 	go o.loopAndDeliverMsgs(s, a)
-	// Startup our state update loop.
-	go o.updateStateLoop()
+	// Startup our interest update loop.
+	// FIXME(dlc) - combine with above once conditional signaling removed.
+	go o.updateInterestLoop()
 
 	o.sendCreateAdvisory()
 
@@ -693,9 +692,11 @@ func (o *Consumer) forceExpirePending() {
 		o.rdq = append(o.rdq, expired...)
 		// Now we should update the timestamp here since we are redelivering.
 		// We will use an incrementing time to preserve order for any other redelivery.
-		off := now - o.pending[expired[0]]
+		off := now - o.pending[expired[0]].Timestamp
 		for _, seq := range expired {
-			o.pending[seq] += off
+			if p, ok := o.pending[seq]; ok && p != nil {
+				p.Timestamp += off
+			}
 		}
 		o.ptmr.Reset(o.ackWait(0))
 	}
@@ -768,8 +769,10 @@ func (o *Consumer) processAck(_ *subscription, _ *client, subject, reply string,
 func (o *Consumer) progressUpdate(seq uint64) {
 	o.mu.Lock()
 	if len(o.pending) > 0 {
-		if _, ok := o.pending[seq]; ok {
-			o.pending[seq] = time.Now().UnixNano()
+		if p, ok := o.pending[seq]; ok {
+			p.Timestamp = time.Now().UnixNano()
+			// Update store system.
+			o.store.UpdateDelivered(p.Sequence, seq, 1, p.Timestamp)
 		}
 	}
 	o.mu.Unlock()
@@ -784,7 +787,7 @@ func (o *Consumer) processNak(sseq, dseq uint64) {
 		o.mu.Unlock()
 		return
 	}
-	// If we are explicit ack make sure this is still on pending list.
+	// If we are explicit ack make sure this is still on our pending list.
 	if len(o.pending) > 0 {
 		if _, ok := o.pending[sseq]; !ok {
 			o.mu.Unlock()
@@ -853,10 +856,10 @@ func (o *Consumer) readStoredState() error {
 	state, err := o.store.State()
 	if err == nil && state != nil {
 		// FIXME(dlc) - re-apply state.
-		o.dseq = state.Delivered.ConsumerSeq
-		o.sseq = state.Delivered.StreamSeq
-		o.adflr = state.AckFloor.ConsumerSeq
-		o.asflr = state.AckFloor.StreamSeq
+		o.dseq = state.Delivered.Consumer + 1
+		o.sseq = state.Delivered.Stream + 1
+		o.adflr = state.AckFloor.Consumer
+		o.asflr = state.AckFloor.Stream
 		o.pending = state.Pending
 		o.rdc = state.Redelivered
 	}
@@ -876,12 +879,12 @@ func (o *Consumer) writeState() {
 	if o.store != nil {
 		state := ConsumerState{
 			Delivered: SequencePair{
-				ConsumerSeq: o.dseq,
-				StreamSeq:   o.sseq,
+				Consumer: o.dseq - 1,
+				Stream:   o.sseq - 1,
 			},
 			AckFloor: SequencePair{
-				ConsumerSeq: o.adflr,
-				StreamSeq:   o.asflr,
+				Consumer: o.adflr,
+				Stream:   o.asflr,
 			},
 			Pending:     o.pending,
 			Redelivered: o.rdc,
@@ -892,9 +895,8 @@ func (o *Consumer) writeState() {
 	o.mu.Unlock()
 }
 
-func (o *Consumer) updateStateLoop() {
+func (o *Consumer) updateInterestLoop() {
 	o.mu.Lock()
-	fch := o.fch
 	qch := o.qch
 	inch := o.inch
 	o.mu.Unlock()
@@ -907,10 +909,6 @@ func (o *Consumer) updateStateLoop() {
 			// inch can be nil on pull-based, but then this will
 			// just block and not fire.
 			o.updateDeliveryInterest(interest)
-		case <-fch:
-			// FIXME(dlc) - Do better, check for fast changes at quick intervals.
-			time.Sleep(10 * time.Millisecond)
-			o.writeState()
 		}
 	}
 }
@@ -924,16 +922,16 @@ func (o *Consumer) Info() *ConsumerInfo {
 		Created: o.created,
 		Config:  o.config,
 		Delivered: SequencePair{
-			ConsumerSeq: o.dseq - 1,
-			StreamSeq:   o.sseq - 1,
+			Consumer: o.dseq - 1,
+			Stream:   o.sseq - 1,
 		},
 		AckFloor: SequencePair{
-			ConsumerSeq: o.adflr,
-			StreamSeq:   o.asflr,
+			Consumer: o.adflr,
+			Stream:   o.asflr,
 		},
 		NumAckPending:  len(o.pending),
 		NumRedelivered: len(o.rdc),
-		NumPending:     o.spending,
+		NumPending:     o.sgap,
 	}
 	// If we are a pull mode consumer, report on number of waiting requests.
 	if o.isPullMode() {
@@ -943,19 +941,6 @@ func (o *Consumer) Info() *ConsumerInfo {
 	return info
 }
 
-// Will update the underlying store.
-// Lock should be held.
-func (o *Consumer) updateStore() {
-	if o.store == nil {
-		return
-	}
-	// Kick our flusher
-	select {
-	case o.fch <- struct{}{}:
-	default:
-	}
-}
-
 // shouldSample lets us know if we are sampling metrics on acks.
 func (o *Consumer) shouldSample() bool {
 	switch {
@@ -988,7 +973,7 @@ func (o *Consumer) sampleAck(sseq, dseq, dcount uint64) {
 		Consumer:    o.name,
 		ConsumerSeq: dseq,
 		StreamSeq:   sseq,
-		Delay:       unow - o.pending[sseq],
+		Delay:       unow - o.pending[sseq].Timestamp,
 		Deliveries:  dcount,
 	}
 
@@ -1006,23 +991,36 @@ func (o *Consumer) ackMsg(sseq, dseq, dcount uint64) {
 }
 
 func (o *Consumer) processAckMsg(sseq, dseq, dcount uint64, doSample bool) {
+	o.mu.Lock()
 	var sagap uint64
 
-	o.mu.Lock()
 	switch o.config.AckPolicy {
 	case AckExplicit:
-		if _, ok := o.pending[sseq]; ok {
+		if p, ok := o.pending[sseq]; ok {
 			if doSample {
 				o.sampleAck(sseq, dseq, dcount)
 			}
 			delete(o.pending, sseq)
-			// Consumers sequence numbers can skip during redlivery since
+
+			// Use the original deliver sequence from our pending record.
+			dseq = p.Sequence
+
+			// Consumers sequence numbers can skip during re-delivery since
 			// they always increment. So if we do not have any pending treat
 			// as all scenario below. Otherwise check that we filled in a gap.
 			if len(o.pending) == 0 {
 				o.adflr, o.asflr = o.dseq-1, o.sseq-1
+				o.pending = nil
 			} else if dseq == o.adflr+1 {
 				o.adflr, o.asflr = dseq, sseq
+				for ss := sseq + 1; ss < o.sseq; ss++ {
+					if p, ok := o.pending[ss]; ok {
+						if p.Sequence > 0 {
+							o.adflr, o.asflr = p.Sequence-1, ss-1
+						}
+						break
+					}
+				}
 			}
 		}
 		// We do these regardless.
@@ -1046,7 +1044,9 @@ func (o *Consumer) processAckMsg(sseq, dseq, dcount uint64, doSample bool) {
 		o.mu.Unlock()
 		return
 	}
-	o.updateStore()
+
+	// Update underlying store.
+	o.store.UpdateAcks(dseq, sseq)
 
 	mset := o.mset
 	o.mu.Unlock()
@@ -1575,11 +1575,12 @@ func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount u
 		return
 	}
 	// Update pending on first attempt
-	if dcount == 1 && o.spending > 0 {
-		o.spending--
+	if dcount == 1 && o.sgap > 0 {
+		o.sgap--
 	}
 
-	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, o.dseq, dcount, ts, o.spending), hdr, msg, o, seq}
+	dseq := o.dseq
+	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, dseq, dcount, ts, o.sgap), hdr, msg, o, seq}
 	mset := o.mset
 	sendq := mset.sendq
 	ap := o.config.AckPolicy
@@ -1596,7 +1597,7 @@ func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount u
 	o.mu.Lock()
 
 	if ap == AckExplicit || ap == AckAll {
-		o.trackPending(seq)
+		o.trackPending(seq, dseq)
 	} else if ap == AckNone {
 		o.adflr = o.dseq
 		o.asflr = seq
@@ -1604,19 +1605,24 @@ func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount u
 
 	o.dseq++
 
-	o.updateStore()
+	// FIXME(dlc) - Capture errors?
+	o.store.UpdateDelivered(dseq, seq, dcount, ts)
 }
 
 // Tracks our outstanding pending acks. Only applicable to AckExplicit mode.
 // Lock should be held.
-func (o *Consumer) trackPending(seq uint64) {
+func (o *Consumer) trackPending(sseq, dseq uint64) {
 	if o.pending == nil {
-		o.pending = make(map[uint64]int64)
+		o.pending = make(map[uint64]*Pending)
 	}
 	if o.ptmr == nil {
 		o.ptmr = time.AfterFunc(o.ackWait(0), o.checkPending)
 	}
-	o.pending[seq] = time.Now().UnixNano()
+	if p, ok := o.pending[sseq]; ok {
+		p.Timestamp = time.Now().UnixNano()
+	} else {
+		o.pending[sseq] = &Pending{dseq, time.Now().UnixNano()}
+	}
 }
 
 // didNotDeliver is called when a delivery for a consumer message failed.
@@ -1689,10 +1695,9 @@ func (o *Consumer) checkPending() {
 
 	// Since we can update timestamps, we have to review all pending.
 	// We may want to unlock here or warn if list is big.
-	// We also need to sort after.
 	var expired []uint64
-	for seq, ts := range o.pending {
-		elapsed := now - ts
+	for seq, p := range o.pending {
+		elapsed := now - p.Timestamp
 		if elapsed >= ttl {
 			if !o.onRedeliverQueue(seq) {
 				expired = append(expired, seq)
@@ -1705,13 +1710,16 @@ func (o *Consumer) checkPending() {
 	}
 
 	if len(expired) > 0 {
+		// We need to sort.
 		sort.Slice(expired, func(i, j int) bool { return expired[i] < expired[j] })
 		o.rdq = append(o.rdq, expired...)
 		// Now we should update the timestamp here since we are redelivering.
 		// We will use an incrementing time to preserve order for any other redelivery.
-		off := now - o.pending[expired[0]]
+		off := now - o.pending[expired[0]].Timestamp
 		for _, seq := range expired {
-			o.pending[seq] += off
+			if p, ok := o.pending[seq]; ok {
+				p.Timestamp += off
+			}
 		}
 	}
 
@@ -1881,7 +1889,7 @@ func (o *Consumer) purge(sseq uint64) {
 	o.sseq = sseq
 	o.asflr = sseq - 1
 	o.adflr = o.dseq - 1
-	o.spending = 0
+	o.sgap = 0
 	if len(o.pending) > 0 {
 		o.pending = nil
 		if o.ptmr != nil {
@@ -1902,6 +1910,8 @@ func (o *Consumer) purge(sseq uint64) {
 		o.rdq = newRDQ
 	}
 	o.mu.Unlock()
+
+	o.writeState()
 }
 
 func stopAndClearTimer(tp **time.Timer) {
@@ -1995,11 +2005,6 @@ func (o *Consumer) stop(dflag, doSignal, advisory bool) error {
 		}
 	}
 
-	// Make sure we stamp our update state
-	if !dflag {
-		o.writeState()
-	}
-
 	var err error
 	if store != nil {
 		if dflag {
@@ -2084,7 +2089,7 @@ func (o *Consumer) setInitialPending() {
 	if o.config.FilterSubject == _EMPTY_ {
 		state := mset.store.State()
 		if state.Msgs > 0 {
-			o.spending = state.Msgs - (o.sseq - state.FirstSeq)
+			o.sgap = state.Msgs - (o.sseq - state.FirstSeq)
 		}
 	} else {
 		// Here we are filtered.
@@ -2096,7 +2101,7 @@ func (o *Consumer) setInitialPending() {
 			} else if err == ErrStoreEOF {
 				break
 			} else if err == nil && o.isFilteredMatch(subj) {
-				o.spending++
+				o.sgap++
 			}
 		}
 	}
@@ -2106,7 +2111,7 @@ func (o *Consumer) setInitialPending() {
 func (o *Consumer) incStreamPending(sseq uint64, subj string) {
 	o.mu.Lock()
 	if o.isFilteredMatch(subj) {
-		o.spending++
+		o.sgap++
 	}
 	o.mu.Unlock()
 }
@@ -2114,15 +2119,15 @@ func (o *Consumer) incStreamPending(sseq uint64, subj string) {
 func (o *Consumer) decStreamPending(sseq uint64) {
 	o.mu.Lock()
 	// Ignore if we have already sent this one.
-	if sseq < o.sseq || o.spending == 0 {
+	if sseq < o.sseq || o.sgap == 0 {
 		o.mu.Unlock()
 		return
 	}
 	if o.config.FilterSubject == _EMPTY_ {
-		o.spending--
+		o.sgap--
 	} else if o.mset != nil {
 		if subj, _, _, _, _ := o.mset.store.LoadMsg(sseq); o.isFilteredMatch(subj) {
-			o.spending--
+			o.sgap--
 		}
 	}
 	o.mu.Unlock()
diff --git a/server/filestore.go b/server/filestore.go
index e0645f1f..69be0be4 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -980,9 +980,7 @@ func (mb *msgBlock) spinUpFlushLoop() {
 	fch, qch := mb.fch, mb.qch
 	mb.mu.Unlock()
 
-	sch := make(chan struct{})
-	go mb.flushLoop(fch, qch, sch)
-	<-sch // Will be closed when flush loop starts.
+	go mb.flushLoop(fch, qch)
 }
 
 // Raw low level kicker for flush loops.
@@ -1015,15 +1013,10 @@ func (mb *msgBlock) clearInFlusher() {
 }
 
 // flushLoop watches for messages, index info, or recently closed msg block updates.
-func (mb *msgBlock) flushLoop(fch, qch, sch chan struct{}) {
+func (mb *msgBlock) flushLoop(fch, qch chan struct{}) {
 	mb.setInFlusher()
 	defer mb.clearInFlusher()
 
-	// Signal back that we are runnng.
-	if sch != nil {
-		close(sch)
-	}
-
 	// Will use to test if we have meta data updates.
 	var firstSeq, lastSeq uint64
 	var dmapLen int
@@ -1933,6 +1926,7 @@ var (
 	errNoPending    = errors.New("message block does not have pending data")
 	errNotReadable  = errors.New("storage directory not readable")
 	errFlushRunning = errors.New("flush is already running")
+	errCorruptState = errors.New("corrupt state file")
 )
 
 // Used for marking messages that have had their checksums checked.
@@ -2171,14 +2165,6 @@ func (mb *msgBlock) writeIndexInfo() error {
 	return err
 }
 
-// Make sure the header is correct.
-func checkHeader(hdr []byte) error {
-	if hdr == nil || len(hdr) < 2 || hdr[0] != magic || hdr[1] != version {
-		return fmt.Errorf("corrupt state file")
-	}
-	return nil
-}
-
 // readIndexInfo will read in the index information for the message block.
 func (mb *msgBlock) readIndexInfo() error {
 	buf, err := ioutil.ReadFile(mb.ifn)
@@ -2646,7 +2632,6 @@ func (fs *fileStore) streamSnapshot(w io.WriteCloser, blks []*msgBlock, includeC
 	fs.mu.Unlock()
 
 	for _, o := range cfs {
-		o.syncStateFile()
 		o.mu.Lock()
 		meta, err := ioutil.ReadFile(path.Join(o.odir, JetStreamMetaFile))
 		if err != nil {
@@ -2660,10 +2645,12 @@ func (fs *fileStore) streamSnapshot(w io.WriteCloser, blks []*msgBlock, includeC
 			writeErr(fmt.Sprintf("Could not read consumer checksum file for %q: %v", o.name, err))
 			return
 		}
-		state, err := ioutil.ReadFile(path.Join(o.odir, consumerState))
+
+		// We can have the running state directly encoded now.
+		state, err := o.encodeState()
 		if err != nil {
 			o.mu.Unlock()
-			writeErr(fmt.Sprintf("Could not read consumer state for %q: %v", o.name, err))
+			writeErr(fmt.Sprintf("Could not encode consumer state for %q: %v", o.name, err))
 			return
 		}
 		odirPre := consumerDir + "/" + o.name
@@ -2729,16 +2716,21 @@ func (fs *fileStore) fileStoreConfig() FileStoreConfig {
 ////////////////////////////////////////////////////////////////////////////////
 
 type consumerFileStore struct {
-	mu     sync.Mutex
-	fs     *fileStore
-	cfg    *FileConsumerInfo
-	name   string
-	odir   string
-	ifn    string
-	ifd    *os.File
-	lwsz   int64
-	hh     hash.Hash64
-	closed bool
+	mu      sync.Mutex
+	fs      *fileStore
+	cfg     *FileConsumerInfo
+	name    string
+	odir    string
+	ifn     string
+	ifd     *os.File
+	lwsz    int64
+	hh      hash.Hash64
+	state   ConsumerState
+	fch     chan struct{}
+	qch     chan struct{}
+	flusher bool
+	writing bool
+	closed  bool
 }
 
 func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerStore, error) {
@@ -2779,6 +2771,11 @@ func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerSt
 		}
 	}
 
+	// Create channels to control our flush go routine.
+	o.fch = make(chan struct{}, 1)
+	o.qch = make(chan struct{})
+	go o.flushLoop()
+
 	fs.mu.Lock()
 	fs.cfs = append(fs.cfs, o)
 	fs.mu.Unlock()
@@ -2786,91 +2783,302 @@ func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerSt
 	return o, nil
 }
 
-const seqsHdrSize = 6*binary.MaxVarintLen64 + hdrLen
+// Kick flusher for this consumer.
+// Lock should be held.
+func (o *consumerFileStore) kickFlusher() {
+	if o.fch != nil {
+		select {
+		case o.fch <- struct{}{}:
+		default:
+		}
+	}
+}
 
-func (o *consumerFileStore) Update(state *ConsumerState) error {
-	// Sanity checks.
-	if state.Delivered.ConsumerSeq < 1 || state.Delivered.StreamSeq < 1 {
-		return fmt.Errorf("bad delivered sequences")
+// Set in flusher status
+func (o *consumerFileStore) setInFlusher() {
+	o.mu.Lock()
+	o.flusher = true
+	o.mu.Unlock()
+}
+
+// Clear in flusher status
+func (o *consumerFileStore) clearInFlusher() {
+	o.mu.Lock()
+	o.flusher = false
+	o.mu.Unlock()
+}
+
+// Report in flusher status
+func (o *consumerFileStore) inFlusher() bool {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return o.flusher
+}
+
+// flushLoop watches for consumer updates and the quit channel.
+func (o *consumerFileStore) flushLoop() {
+	o.mu.Lock()
+	fch, qch := o.fch, o.qch
+	o.mu.Unlock()
+
+	o.setInFlusher()
+	defer o.clearInFlusher()
+
+	for {
+		select {
+		case <-fch:
+			time.Sleep(5 * time.Millisecond)
+			select {
+			case <-qch:
+				return
+			default:
+			}
+			o.mu.Lock()
+			if o.closed {
+				o.mu.Unlock()
+				return
+			}
+			buf, err := o.encodeState()
+			o.mu.Unlock()
+			if err != nil {
+				return
+			}
+			// TODO(dlc) - if we error should start failing upwards.
+			o.writeState(buf)
+		case <-qch:
+			return
+		}
 	}
-	if state.AckFloor.ConsumerSeq > state.Delivered.ConsumerSeq {
-		return fmt.Errorf("bad ack floor for consumer")
+}
+
+// UpdateDelivered is called whenever a new message has been delivered.
+func (o *consumerFileStore) UpdateDelivered(dseq, sseq, dc uint64, ts int64) error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	if dc != 1 && o.cfg.AckPolicy == AckNone {
+		return ErrNoAckPolicy
 	}
-	if state.AckFloor.StreamSeq > state.Delivered.StreamSeq {
-		return fmt.Errorf("bad ack floor for stream")
+
+	// See if we expect an ack for this.
+	if o.cfg.AckPolicy != AckNone {
+		// Need to create pending records here.
+		if o.state.Pending == nil {
+			o.state.Pending = make(map[uint64]*Pending)
+		}
+		var p *Pending
+		// Check for an update to a message already delivered.
+		if sseq <= o.state.Delivered.Stream {
+			if p = o.state.Pending[sseq]; p != nil {
+				p.Timestamp = ts
+			}
+		}
+		if p == nil {
+			// Move delivered if this is new.
+			o.state.Delivered.Consumer = dseq
+			o.state.Delivered.Stream = sseq
+			p = &Pending{dseq, ts}
+		} else if dc > 1 {
+			if o.state.Redelivered == nil {
+				o.state.Redelivered = make(map[uint64]uint64)
+			}
+			o.state.Redelivered[sseq] = dc
+		}
+		o.state.Pending[sseq] = &Pending{dseq, ts}
+	} else {
+		// For AckNone just update delivered and ackfloor at the same time.
+		o.state.Delivered.Consumer = dseq
+		o.state.Delivered.Stream = sseq
+		o.state.AckFloor.Consumer = dseq
+		o.state.AckFloor.Stream = sseq
+	}
+	// Make sure we flush to disk.
+	o.kickFlusher()
+
+	return nil
+}
+
+// UpdateAcks is called whenever a consumer with explicit ack or ack all acks a message.
+func (o *consumerFileStore) UpdateAcks(dseq, sseq uint64) error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	if o.cfg.AckPolicy == AckNone {
+		return ErrNoAckPolicy
+	}
+	if len(o.state.Pending) == 0 {
+		return ErrStoreMsgNotFound
+	}
+	p := o.state.Pending[sseq]
+	if p == nil {
+		return ErrStoreMsgNotFound
+	}
+	delete(o.state.Pending, sseq)
+
+	// TODO(dlc) - Check to see if we move ack floor.
+	if len(o.state.Pending) == 0 {
+		o.state.Pending = nil
+		o.state.AckFloor.Consumer = o.state.Delivered.Consumer
+		o.state.AckFloor.Stream = o.state.Delivered.Stream
+	} else if o.state.AckFloor.Consumer == 0 {
+		o.state.AckFloor.Consumer = dseq
+		o.state.AckFloor.Stream = sseq
+	} else if o.state.AckFloor.Consumer == dseq-1 {
+		o.state.AckFloor.Consumer = dseq
+		o.state.AckFloor.Stream = sseq
+		// Close gap if needed.
+		for ss := sseq + 1; ss < o.state.Delivered.Stream; ss++ {
+			if p, ok := o.state.Pending[ss]; ok {
+				if p.Sequence > 0 {
+					o.state.AckFloor.Consumer = p.Sequence - 1
+					o.state.AckFloor.Stream = ss - 1
+				}
+				break
+			}
+		}
+	}
+	o.kickFlusher()
+	return nil
+}
+
+const seqsHdrSize = 6*binary.MaxVarintLen64 + hdrLen
+
+// Encode our consumer state, version 2.
+// Lock should be held.
+func (o *consumerFileStore) encodeState() ([]byte, error) {
+	if o.closed {
+		return nil, ErrStoreClosed
 	}
 
 	var hdr [seqsHdrSize]byte
+	var buf []byte
+
+	maxSize := seqsHdrSize
+	if lp := len(o.state.Pending); lp > 0 {
+		maxSize += lp*(3*binary.MaxVarintLen64) + binary.MaxVarintLen64
+	}
+	if lr := len(o.state.Redelivered); lr > 0 {
+		maxSize += lr*(2*binary.MaxVarintLen64) + binary.MaxVarintLen64
+	}
+	if maxSize == seqsHdrSize {
+		buf = hdr[:seqsHdrSize]
+	} else {
+		buf = make([]byte, maxSize)
+	}
+
+	now := time.Now()
 
 	// Write header
-	hdr[0] = magic
-	hdr[1] = version
+	buf[0] = magic
+	buf[1] = 2
 
 	n := hdrLen
-	n += binary.PutUvarint(hdr[n:], state.AckFloor.ConsumerSeq)
-	n += binary.PutUvarint(hdr[n:], state.AckFloor.StreamSeq)
-	n += binary.PutUvarint(hdr[n:], state.Delivered.ConsumerSeq-state.AckFloor.ConsumerSeq)
-	n += binary.PutUvarint(hdr[n:], state.Delivered.StreamSeq-state.AckFloor.StreamSeq)
-	n += binary.PutUvarint(hdr[n:], uint64(len(state.Pending)))
-	buf := hdr[:n]
+	n += binary.PutUvarint(buf[n:], o.state.AckFloor.Consumer)
+	n += binary.PutUvarint(buf[n:], o.state.AckFloor.Stream)
+	n += binary.PutUvarint(buf[n:], o.state.Delivered.Consumer)
+	n += binary.PutUvarint(buf[n:], o.state.Delivered.Stream)
+	n += binary.PutUvarint(buf[n:], uint64(len(o.state.Pending)))
+
+	asflr := o.state.AckFloor.Stream
+	adflr := o.state.AckFloor.Consumer
 
 	// These are optional, but always write len. This is to avoid a truncate inline.
-	// If these get big might make more sense to do writes directly to the file.
-	if len(state.Pending) > 0 {
-		mbuf := make([]byte, len(state.Pending)*(2*binary.MaxVarintLen64)+binary.MaxVarintLen64)
-		aflr := state.AckFloor.StreamSeq
-		maxd := state.Delivered.StreamSeq
-
-		// To save space we select the smallest timestamp.
-		var mints int64
-		for k, v := range state.Pending {
-			if mints == 0 || v < mints {
-				mints = v
-			}
-			if k <= aflr || k > maxd {
-				return fmt.Errorf("bad pending entry, sequence [%d] out of range", k)
-			}
+	if len(o.state.Pending) > 0 {
+		// To save space we will use now rounded to seconds to be base timestamp.
+		mints := now.Round(time.Second).Unix()
+		// Write minimum timestamp we found from above.
+		n += binary.PutVarint(buf[n:], mints)
+
+		for k, v := range o.state.Pending {
+			n += binary.PutUvarint(buf[n:], k-asflr)
+			n += binary.PutUvarint(buf[n:], v.Sequence-adflr)
+			// Downsample to seconds to save on space.
+			// Subsecond resolution not needed for recovery etc.
+			ts := v.Timestamp / 1_000_000_000
+			n += binary.PutVarint(buf[n:], mints-ts)
 		}
+	}
 
-		// Downsample the minimum timestamp.
-		mints /= int64(time.Second)
-		var n int
-		// Write minimum timestamp we found from above.
-		n += binary.PutVarint(mbuf[n:], mints)
+	// We always write the redelivered len.
+	n += binary.PutUvarint(buf[n:], uint64(len(o.state.Redelivered)))
 
-		for k, v := range state.Pending {
-			n += binary.PutUvarint(mbuf[n:], k-aflr)
-			// Downsample to seconds to save on space. Subsecond resolution not
-			// needed for recovery etc.
-			n += binary.PutVarint(mbuf[n:], (v/int64(time.Second))-mints)
+	// We expect these to be small.
+	if len(o.state.Redelivered) > 0 {
+		for k, v := range o.state.Redelivered {
+			n += binary.PutUvarint(buf[n:], k-asflr)
+			n += binary.PutUvarint(buf[n:], v)
 		}
-		buf = append(buf, mbuf[:n]...)
 	}
+	return buf[:n], nil
+}
 
-	var lenbuf [binary.MaxVarintLen64]byte
-	n = binary.PutUvarint(lenbuf[0:], uint64(len(state.Redelivered)))
-	buf = append(buf, lenbuf[:n]...)
+func (o *consumerFileStore) Update(state *ConsumerState) error {
+	// Sanity checks.
+	if state.Delivered.Consumer < 1 || state.Delivered.Stream < 1 {
+		return fmt.Errorf("bad delivered sequences")
+	}
+	if state.AckFloor.Consumer > state.Delivered.Consumer {
+		return fmt.Errorf("bad ack floor for consumer")
+	}
+	if state.AckFloor.Stream > state.Delivered.Stream {
+		return fmt.Errorf("bad ack floor for stream")
+	}
 
-	// We expect these to be small so will not do anything too crazy here to
-	// keep the size small. Trick could be to offset sequence like above, but
-	// we would need to know low sequence number for redelivery, can't depend on ackfloor etc.
+	// Copy to our state.
+	var pending map[uint64]*Pending
+	var redelivered map[uint64]uint64
+	if len(state.Pending) > 0 {
+		pending = make(map[uint64]*Pending, len(state.Pending))
+		for seq, p := range state.Pending {
+			pending[seq] = &Pending{p.Sequence, p.Timestamp}
+		}
+		for seq := range pending {
+			if seq <= state.AckFloor.Stream || seq > state.Delivered.Stream {
+				return fmt.Errorf("bad pending entry, sequence [%d] out of range", seq)
+			}
+		}
+	}
 	if len(state.Redelivered) > 0 {
-		mbuf := make([]byte, len(state.Redelivered)*(2*binary.MaxVarintLen64))
-		var n int
-		for k, v := range state.Redelivered {
-			n += binary.PutUvarint(mbuf[n:], k)
-			n += binary.PutUvarint(mbuf[n:], v)
+		redelivered = make(map[uint64]uint64, len(state.Redelivered))
+		for seq, dc := range state.Redelivered {
+			redelivered[seq] = dc
 		}
-		buf = append(buf, mbuf[:n]...)
 	}
 
+	// Replace our state.
+	o.mu.Lock()
+	o.state.Delivered = state.Delivered
+	o.state.AckFloor = state.AckFloor
+	o.state.Pending = pending
+	o.state.Redelivered = redelivered
+	o.mu.Unlock()
+
+	o.kickFlusher()
+	return nil
+}
+
+func (o *consumerFileStore) writeState(buf []byte) error {
 	// Check if we have the index file open.
 	o.mu.Lock()
-	err := o.ensureStateFileOpen()
+	if o.writing || len(buf) == 0 {
+		o.mu.Unlock()
+		return nil
+	}
+	if err := o.ensureStateFileOpen(); err != nil {
+		o.mu.Unlock()
+		return err
+	}
+	o.writing = true
+	ifd := o.ifd
+	o.mu.Unlock()
+
+	n, err := ifd.WriteAt(buf, 0)
+
+	o.mu.Lock()
 	if err == nil {
-		n, err = o.ifd.WriteAt(buf, 0)
 		o.lwsz = int64(n)
 	}
+	o.writing = false
 	o.mu.Unlock()
 
 	return err
@@ -2930,24 +3138,76 @@ func (o *consumerFileStore) ensureStateFileOpen() error {
 	return nil
 }
 
+// Make sure the header is correct.
+func checkHeader(hdr []byte) error {
+	if hdr == nil || len(hdr) < 2 || hdr[0] != magic || hdr[1] != version {
+		return errCorruptState
+	}
+	return nil
+}
+
+func checkConsumerHeader(hdr []byte) (uint8, error) {
+	if hdr == nil || len(hdr) < 2 || hdr[0] != magic {
+		return 0, errCorruptState
+	}
+	version := hdr[1]
+	switch version {
+	case 1, 2:
+		return version, nil
+	}
+	return 0, fmt.Errorf("unsupported version: %d", version)
+}
+
+func (o *consumerFileStore) copyPending() map[uint64]*Pending {
+	pending := make(map[uint64]*Pending, len(o.state.Pending))
+	for seq, p := range o.state.Pending {
+		pending[seq] = &Pending{p.Sequence, p.Timestamp}
+	}
+	return pending
+}
+
+func (o *consumerFileStore) copyRedelivered() map[uint64]uint64 {
+	redelivered := make(map[uint64]uint64, len(o.state.Redelivered))
+	for seq, dc := range o.state.Redelivered {
+		redelivered[seq] = dc
+	}
+	return redelivered
+}
+
 // State retrieves the state from the state file.
 // This is not expected to be called in high performance code, only on startup.
 func (o *consumerFileStore) State() (*ConsumerState, error) {
 	o.mu.Lock()
 	defer o.mu.Unlock()
 
+	var state *ConsumerState
+
+	// See if we have a running state or if we need to read in from disk.
+	if o.state.Delivered.Consumer != 0 {
+		state = &ConsumerState{}
+		state.Delivered = o.state.Delivered
+		state.AckFloor = o.state.AckFloor
+		if len(o.state.Pending) > 0 {
+			state.Pending = o.copyPending()
+		}
+		if len(o.state.Redelivered) > 0 {
+			state.Redelivered = o.copyRedelivered()
+		}
+		return state, nil
+	}
+
+	// Read the state in here from disk..
 	buf, err := ioutil.ReadFile(o.ifn)
 	if err != nil && !os.IsNotExist(err) {
 		return nil, err
 	}
 
-	var state *ConsumerState
-
 	if len(buf) == 0 {
 		return state, nil
 	}
 
-	if err := checkHeader(buf); err != nil {
+	version, err := checkConsumerHeader(buf)
+	if err != nil {
 		return nil, err
 	}
 
@@ -2983,36 +3243,50 @@ func (o *consumerFileStore) State() (*ConsumerState, error) {
 	readCount := readSeq
 
 	state = &ConsumerState{}
-	state.AckFloor.ConsumerSeq = readSeq()
-	state.AckFloor.StreamSeq = readSeq()
-	state.Delivered.ConsumerSeq = readSeq()
-	state.Delivered.StreamSeq = readSeq()
+	state.AckFloor.Consumer = readSeq()
+	state.AckFloor.Stream = readSeq()
+	state.Delivered.Consumer = readSeq()
+	state.Delivered.Stream = readSeq()
 
 	if bi == -1 {
-		return nil, fmt.Errorf("corrupt state file")
+		return nil, errCorruptState
+	}
+	if version == 1 {
+		// Adjust back. Version 1 also stored delivered as next to be delivered,
+		// so adjust that back down here.
+		state.Delivered.Consumer += state.AckFloor.Consumer - 1
+		state.Delivered.Stream += state.AckFloor.Stream - 1
 	}
-	// Adjust back.
-	state.Delivered.ConsumerSeq += state.AckFloor.ConsumerSeq
-	state.Delivered.StreamSeq += state.AckFloor.StreamSeq
 
 	numPending := readLen()
 
 	// We have additional stuff.
 	if numPending > 0 {
 		mints := readTimeStamp()
-		state.Pending = make(map[uint64]int64, numPending)
+		state.Pending = make(map[uint64]*Pending, numPending)
 		for i := 0; i < int(numPending); i++ {
-			seq := readSeq()
+			sseq := readSeq()
+			var dseq uint64
+			if version == 2 {
+				dseq = readSeq()
+			}
 			ts := readTimeStamp()
-			if seq == 0 || ts == -1 {
-				return nil, fmt.Errorf("corrupt state file")
+			if sseq == 0 || ts == -1 {
+				return nil, errCorruptState
 			}
 			// Adjust seq back.
-			seq += state.AckFloor.StreamSeq
+			sseq += state.AckFloor.Stream
+			if version == 2 {
+				dseq += state.AckFloor.Consumer
+			}
 			// Adjust the timestamp back.
-			ts = (ts + mints) * int64(time.Second)
+			if version == 1 {
+				ts = (ts + mints) * int64(time.Second)
+			} else {
+				ts = (mints - ts) * int64(time.Second)
+			}
 			// Store in pending.
-			state.Pending[seq] = ts
+			state.Pending[sseq] = &Pending{dseq, ts}
 		}
 	}
 
@@ -3025,7 +3299,7 @@ func (o *consumerFileStore) State() (*ConsumerState, error) {
 			seq := readSeq()
 			n := readCount()
 			if seq == 0 || n == 0 {
-				return nil, fmt.Errorf("corrupt state file")
+				return nil, errCorruptState
 			}
 			state.Redelivered[seq] = n
 		}
@@ -3040,43 +3314,82 @@ func (o *consumerFileStore) Stop() error {
 		o.mu.Unlock()
 		return nil
 	}
-	o.closed = true
-	if o.ifd != nil {
-		o.ifd.Sync()
-		o.ifd.Close()
-		o.ifd = nil
+	if o.qch != nil {
+		close(o.qch)
+		o.qch = nil
 	}
+
+	err := o.ensureStateFileOpen()
+	ifd := o.ifd
+
+	if err == nil {
+		var buf []byte
+		// Make sure to write this out..
+		if buf, err = o.encodeState(); err == nil {
+			_, err = ifd.WriteAt(buf, 0)
+		}
+	}
+
+	o.ifd, o.odir = nil, _EMPTY_
 	fs := o.fs
+	o.closed = true
+	o.kickFlusher()
 	o.mu.Unlock()
+
+	if ifd != nil {
+		ifd.Sync()
+		ifd.Close()
+	}
+
 	fs.removeConsumer(o)
-	return nil
+	return err
 }
 
 // Delete the consumer.
 func (o *consumerFileStore) Delete() error {
-	// Call stop first. OK if already stopped.
-	o.Stop()
 	o.mu.Lock()
+
+	if o.closed {
+		o.mu.Unlock()
+		return nil
+	}
+	if o.qch != nil {
+		close(o.qch)
+		o.qch = nil
+	}
+
+	if o.ifd != nil {
+		o.ifd.Close()
+		o.ifd = nil
+	}
 	var err error
-	if o.odir != "" {
+	if o.odir != _EMPTY_ {
 		err = os.RemoveAll(o.odir)
 	}
+	o.ifd, o.odir = nil, _EMPTY_
+	o.closed = true
+	fs := o.fs
 	o.mu.Unlock()
+
+	fs.removeConsumer(o)
 	return err
 }
 
 func (fs *fileStore) removeConsumer(cfs *consumerFileStore) {
 	fs.mu.Lock()
+	defer fs.mu.Unlock()
 	for i, o := range fs.cfs {
 		if o == cfs {
 			fs.cfs = append(fs.cfs[:i], fs.cfs[i+1:]...)
 			break
 		}
 	}
-	fs.mu.Unlock()
 }
 
+////////////////////////////////////////////////////////////////////////////////
 // Templates
+////////////////////////////////////////////////////////////////////////////////
+
 type templateFileStore struct {
 	dir string
 	hh  hash.Hash64
diff --git a/server/jetstream.go b/server/jetstream.go
index 10b5dfd7..1462512a 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -533,7 +533,7 @@ func (a *Account) EnableJetStream(limits *JetStreamAccountLimits) error {
 		stats := mset.State()
 		s.Noticef("  Restored %s messages for Stream %q", comma(int64(stats.Msgs)), fi.Name())
 
-		// Now do Consumers.
+		// Now do the consumers.
 		odir := path.Join(sdir, fi.Name(), consumerDir)
 		ofis, _ := ioutil.ReadDir(odir)
 		if len(ofis) > 0 {
diff --git a/server/memstore.go b/server/memstore.go
index bcc845c1..43c866fe 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -435,9 +435,12 @@ func (ms *memStore) Snapshot(_ time.Duration, _, _ bool) (*SnapshotResult, error
 }
 
 // No-ops.
-func (os *consumerMemStore) Update(_ *ConsumerState) error {
-	return nil
-}
+func (os *consumerMemStore) Update(_ *ConsumerState) error { return nil }
+
+func (os *consumerMemStore) UpdateDelivered(_, _, _ uint64, _ int64) error { return nil }
+
+func (os *consumerMemStore) UpdateAcks(_, _ uint64) error { return nil }
+
 func (os *consumerMemStore) Stop() error {
 	os.ms.decConsumers()
 	return nil
diff --git a/server/store.go b/server/store.go
index 06f7def6..6818715d 100644
--- a/server/store.go
+++ b/server/store.go
@@ -52,6 +52,8 @@ var (
 	ErrMsgTooLarge = errors.New("message to large")
 	// ErrStoreWrongType is for when you access the wrong storage type.
 	ErrStoreWrongType = errors.New("wrong storage type")
+	// ErrNoAckPolicy is returned when trying to update a consumer's acks with no ack policy.
+	ErrNoAckPolicy = errors.New("ack policy is none")
 )
 
 type StreamStore interface {
@@ -115,16 +117,18 @@ type SnapshotResult struct {
 
 // ConsumerStore stores state on consumers for streams.
 type ConsumerStore interface {
-	State() (*ConsumerState, error)
+	UpdateDelivered(dseq, sseq, dc uint64, ts int64) error
+	UpdateAcks(dseq, sseq uint64) error
 	Update(*ConsumerState) error
+	State() (*ConsumerState, error)
 	Stop() error
 	Delete() error
 }
 
 // SequencePair has both the consumer and the stream sequence. They point to same message.
 type SequencePair struct {
-	ConsumerSeq uint64 `json:"consumer_seq"`
-	StreamSeq   uint64 `json:"stream_seq"`
+	Consumer uint64 `json:"consumer_seq"`
+	Stream   uint64 `json:"stream_seq"`
 }
 
 // ConsumerState represents a stored state for a consumer.
@@ -136,11 +140,18 @@ type ConsumerState struct {
 	// These are both in stream sequence context.
 	// Pending is for all messages pending and the timestamp for the delivered time.
 	// This will only be present when the AckPolicy is ExplicitAck.
-	Pending map[uint64]int64 `json:"pending"`
+	Pending map[uint64]*Pending `json:"pending"`
 	// This is for messages that have been redelivered, so count > 1.
 	Redelivered map[uint64]uint64 `json:"redelivered"`
 }
 
+// Represents a pending message for explicit ack or ack all.
+// Sequence is the original consumer sequence.
+type Pending struct {
+	Sequence  uint64
+	Timestamp int64
+}
+
 // TemplateStore stores templates.
 type TemplateStore interface {
 	Store(*StreamTemplate) error
diff --git a/server/stream.go b/server/stream.go
index ef85d831..743345ea 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -1335,17 +1335,8 @@ func (mset *Stream) Snapshot(deadline time.Duration, checkMsgs, includeConsumers
 		return nil, fmt.Errorf("invalid stream")
 	}
 	store := mset.store
-	var obs []*Consumer
-	for _, o := range mset.consumers {
-		obs = append(obs, o)
-	}
 	mset.mu.Unlock()
 
-	// Make sure to sync their state.
-	for _, o := range obs {
-		o.writeState()
-	}
-
 	return store.Snapshot(deadline, checkMsgs, includeConsumers)
 }
 


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index bd5885ec..8438a09e 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -348,7 +348,6 @@ func TestFileStoreSkipMsg(t *testing.T) {
 
 	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?" {
@@ -1437,16 +1436,16 @@ func TestFileStoreSnapshot(t *testing.T) {
 		t.Fatalf("Unexepected error: %v", err)
 	}
 	state := &ConsumerState{}
-	state.Delivered.ConsumerSeq = 100
-	state.Delivered.StreamSeq = 100
-	state.AckFloor.ConsumerSeq = 22
-	state.AckFloor.StreamSeq = 22
+	state.Delivered.Consumer = 100
+	state.Delivered.Stream = 100
+	state.AckFloor.Consumer = 22
+	state.AckFloor.Stream = 22
 
 	if err := o1.Update(state); err != nil {
 		t.Fatalf("Unexepected error updating state: %v", err)
 	}
-	state.AckFloor.ConsumerSeq = 33
-	state.AckFloor.StreamSeq = 33
+	state.AckFloor.Consumer = 33
+	state.AckFloor.Stream = 33
 
 	if err := o2.Update(state); err != nil {
 		t.Fatalf("Unexepected error updating state: %v", err)
@@ -1629,47 +1628,47 @@ func TestFileStoreConsumer(t *testing.T) {
 		}
 	}
 
-	state.Delivered.ConsumerSeq = 1
-	state.Delivered.StreamSeq = 22
+	state.Delivered.Consumer = 1
+	state.Delivered.Stream = 22
 	updateAndCheck()
 
-	state.Delivered.ConsumerSeq = 100
-	state.Delivered.StreamSeq = 122
-	state.AckFloor.ConsumerSeq = 50
-	state.AckFloor.StreamSeq = 123
+	state.Delivered.Consumer = 100
+	state.Delivered.Stream = 122
+	state.AckFloor.Consumer = 50
+	state.AckFloor.Stream = 123
 	// This should fail, bad state.
 	shouldFail()
 	// So should this.
-	state.AckFloor.ConsumerSeq = 200
-	state.AckFloor.StreamSeq = 100
+	state.AckFloor.Consumer = 200
+	state.AckFloor.Stream = 100
 	shouldFail()
 
 	// Should succeed
-	state.AckFloor.ConsumerSeq = 50
-	state.AckFloor.StreamSeq = 72
+	state.AckFloor.Consumer = 50
+	state.AckFloor.Stream = 72
 	updateAndCheck()
 
 	tn := time.Now().UnixNano()
 
 	// We should sanity check pending here as well, so will check if a pending value is below
 	// ack floor or above delivered.
-	state.Pending = map[uint64]int64{70: tn}
+	state.Pending = map[uint64]*Pending{70: &Pending{70, tn}}
 	shouldFail()
-	state.Pending = map[uint64]int64{140: tn}
+	state.Pending = map[uint64]*Pending{140: &Pending{140, tn}}
 	shouldFail()
-	state.Pending = map[uint64]int64{72: tn} // exact on floor should fail
+	state.Pending = map[uint64]*Pending{72: &Pending{72, tn}} // exact on floor should fail
 	shouldFail()
 
 	// Put timestamps a second apart.
 	// We will downsample to second resolution to save space. So setup our times
 	// to reflect that.
 	ago := time.Now().Add(-30 * time.Second).Truncate(time.Second)
-	nt := func() int64 {
+	nt := func() *Pending {
 		ago = ago.Add(time.Second)
-		return ago.UnixNano()
+		return &Pending{0, ago.UnixNano()}
 	}
 	// Should succeed.
-	state.Pending = map[uint64]int64{75: nt(), 80: nt(), 83: nt(), 90: nt(), 111: nt()}
+	state.Pending = map[uint64]*Pending{75: nt(), 80: nt(), 83: nt(), 90: nt(), 111: nt()}
 	updateAndCheck()
 
 	// Now do redlivery, but first with no pending.
@@ -1678,16 +1677,16 @@ func TestFileStoreConsumer(t *testing.T) {
 	updateAndCheck()
 
 	// All together.
-	state.Pending = map[uint64]int64{75: nt(), 80: nt(), 83: nt(), 90: nt(), 111: nt()}
+	state.Pending = map[uint64]*Pending{75: nt(), 80: nt(), 83: nt(), 90: nt(), 111: nt()}
 	updateAndCheck()
 
 	// Large one
-	state.Delivered.ConsumerSeq = 10000
-	state.Delivered.StreamSeq = 10000
-	state.AckFloor.ConsumerSeq = 100
-	state.AckFloor.StreamSeq = 100
+	state.Delivered.Consumer = 10000
+	state.Delivered.Stream = 10000
+	state.AckFloor.Consumer = 100
+	state.AckFloor.Stream = 100
 	// Generate 8k pending.
-	state.Pending = make(map[uint64]int64)
+	state.Pending = make(map[uint64]*Pending)
 	for len(state.Pending) < 8192 {
 		seq := uint64(rand.Intn(9890) + 101)
 		if _, ok := state.Pending[seq]; !ok {
@@ -1697,8 +1696,8 @@ func TestFileStoreConsumer(t *testing.T) {
 	updateAndCheck()
 
 	state.Pending = nil
-	state.AckFloor.ConsumerSeq = 10000
-	state.AckFloor.StreamSeq = 10000
+	state.AckFloor.Consumer = 10000
+	state.AckFloor.Stream = 10000
 	updateAndCheck()
 }
 
@@ -1990,6 +1989,216 @@ func TestFileStorePubPerfWithSmallBlkSize(t *testing.T) {
 	fmt.Printf("%s per sec\n", FriendlyBytes(int64(float64(toStore*storedMsgSize)/tt.Seconds())))
 }
 
+func TestFileStoreConsumerFlusher(t *testing.T) {
+	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	o, err := fs.ConsumerStore("o22", &ConsumerConfig{})
+	if err != nil {
+		t.Fatalf("Unexepected error: %v", err)
+	}
+	// Get the underlying impl.
+	oc := o.(*consumerFileStore)
+	// Wait for flusher to be running.
+	checkFor(t, time.Second, 20*time.Millisecond, func() error {
+		if !oc.inFlusher() {
+			return fmt.Errorf("Flusher not running")
+		}
+		return nil
+	})
+	// Stop and make sure the flusher goes away
+	o.Stop()
+	// Wait for flusher to stop.
+	checkFor(t, time.Second, 20*time.Millisecond, func() error {
+		if oc.inFlusher() {
+			return fmt.Errorf("Flusher still running")
+		}
+		return nil
+	})
+}
+
+func TestFileStoreConsumerDeliveredUpdates(t *testing.T) {
+	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	// Simple consumer, no ack policy configured.
+	o, err := fs.ConsumerStore("o22", &ConsumerConfig{})
+	if err != nil {
+		t.Fatalf("Unexepected error: %v", err)
+	}
+	defer o.Stop()
+
+	testDelivered := func(dseq, sseq uint64) {
+		t.Helper()
+		ts := time.Now().UnixNano()
+		if err := o.UpdateDelivered(dseq, sseq, 1, ts); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		state, err := o.State()
+		if err != nil {
+			t.Fatalf("Error getting state: %v", err)
+		}
+		if state == nil {
+			t.Fatalf("No state available")
+		}
+		expected := SequencePair{dseq, sseq}
+		if state.Delivered != expected {
+			t.Fatalf("Unexpected state, wanted %+v, got %+v", expected, state.Delivered)
+		}
+		if state.AckFloor != expected {
+			t.Fatalf("Unexpected ack floor state, wanted %+v, got %+v", expected, state.AckFloor)
+		}
+		if len(state.Pending) != 0 {
+			t.Fatalf("Did not expect any pending, got %d pending", len(state.Pending))
+		}
+	}
+
+	testDelivered(1, 100)
+	testDelivered(2, 110)
+	testDelivered(5, 130)
+
+	// If we try to do an ack this should err since we are not configured with ack policy.
+	if err := o.UpdateAcks(1, 100); err != ErrNoAckPolicy {
+		t.Fatalf("Expected a no ack policy error on update acks, got %v", err)
+	}
+	// Also if we do an update with a delivery count of anything but 1 here should also give same error.
+	ts := time.Now().UnixNano()
+	if err := o.UpdateDelivered(5, 130, 2, ts); err != ErrNoAckPolicy {
+		t.Fatalf("Expected a no ack policy error on update delivered with dc > 1, got %v", err)
+	}
+}
+
+func TestFileStoreConsumerDeliveredAndAckUpdates(t *testing.T) {
+	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
+	os.MkdirAll(storeDir, 0755)
+	defer os.RemoveAll(storeDir)
+
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	defer fs.Stop()
+
+	// Simple consumer, no ack policy configured.
+	o, err := fs.ConsumerStore("o22", &ConsumerConfig{AckPolicy: AckExplicit})
+	if err != nil {
+		t.Fatalf("Unexepected error: %v", err)
+	}
+	defer o.Stop()
+
+	// Track pending.
+	var pending int
+
+	testDelivered := func(dseq, sseq uint64) {
+		t.Helper()
+		ts := time.Now().Round(time.Second).UnixNano()
+		if err := o.UpdateDelivered(dseq, sseq, 1, ts); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		pending++
+		state, err := o.State()
+		if err != nil {
+			t.Fatalf("Error getting state: %v", err)
+		}
+		if state == nil {
+			t.Fatalf("No state available")
+		}
+		expected := SequencePair{dseq, sseq}
+		if state.Delivered != expected {
+			t.Fatalf("Unexpected delivered state, wanted %+v, got %+v", expected, state.Delivered)
+		}
+		if len(state.Pending) != pending {
+			t.Fatalf("Expected %d pending, got %d pending", pending, len(state.Pending))
+		}
+	}
+
+	testDelivered(1, 100)
+	testDelivered(2, 110)
+	testDelivered(3, 130)
+	testDelivered(4, 150)
+	testDelivered(5, 165)
+
+	testBadAck := func(dseq, sseq uint64) {
+		t.Helper()
+		if err := o.UpdateAcks(dseq, sseq); err == nil {
+			t.Fatalf("Expected error but got none")
+		}
+	}
+	testBadAck(3, 101)
+	testBadAck(1, 1)
+
+	testAck := func(dseq, sseq, dflr, sflr uint64) {
+		t.Helper()
+		if err := o.UpdateAcks(dseq, sseq); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
+		}
+		pending--
+		state, err := o.State()
+		if err != nil {
+			t.Fatalf("Error getting state: %v", err)
+		}
+		if state == nil {
+			t.Fatalf("No state available")
+		}
+		if len(state.Pending) != pending {
+			t.Fatalf("Expected %d pending, got %d pending", pending, len(state.Pending))
+		}
+		eflr := SequencePair{dflr, sflr}
+		if state.AckFloor != eflr {
+			t.Fatalf("Unexpected ack floor state, wanted %+v, got %+v", eflr, state.AckFloor)
+		}
+	}
+
+	testAck(1, 100, 1, 100)
+	testAck(3, 130, 1, 100)
+	testAck(2, 110, 3, 149) // We do not track explicit state on previous stream floors, so we take last known -1
+	testAck(5, 165, 3, 149)
+	testAck(4, 150, 5, 165)
+
+	testDelivered(6, 170)
+	testDelivered(7, 171)
+	testDelivered(8, 172)
+	testDelivered(9, 173)
+	testDelivered(10, 200)
+
+	testAck(7, 171, 5, 165)
+	testAck(8, 172, 5, 165)
+
+	state, err := o.State()
+	if err != nil {
+		t.Fatalf("Unexpected error getting state: %v", err)
+	}
+	o.Stop()
+
+	o, err = fs.ConsumerStore("o22", &ConsumerConfig{AckPolicy: AckExplicit})
+	if err != nil {
+		t.Fatalf("Unexepected error: %v", err)
+	}
+	defer o.Stop()
+
+	nstate, err := o.State()
+	if err != nil {
+		t.Fatalf("Unexpected error getting state: %v", err)
+	}
+	if !reflect.DeepEqual(nstate, state) {
+		t.Fatalf("States don't match!")
+	}
+}
+
 func TestFileStoreConsumerPerf(t *testing.T) {
 	// Comment out to run, holding place for now.
 	t.SkipNow()
@@ -1997,7 +2206,6 @@ func TestFileStoreConsumerPerf(t *testing.T) {
 	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
 	os.MkdirAll(storeDir, 0755)
 	defer os.RemoveAll(storeDir)
-	fmt.Printf("StoreDir is %q\n", storeDir)
 
 	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage})
 	if err != nil {
@@ -2005,53 +2213,61 @@ func TestFileStoreConsumerPerf(t *testing.T) {
 	}
 	defer fs.Stop()
 
-	// Test Consumers.
-	o, err := fs.ConsumerStore("obs22", &ConsumerConfig{})
+	o, err := fs.ConsumerStore("o22", &ConsumerConfig{AckPolicy: AckExplicit})
 	if err != nil {
 		t.Fatalf("Unexepected error: %v", err)
 	}
-	state := &ConsumerState{}
-	toStore := uint64(1000000)
-	fmt.Printf("consumer of %d msgs for ACK NONE\n", toStore)
+	// Get the underlying impl.
+	oc := o.(*consumerFileStore)
+	// Wait for flusher to br running
+	checkFor(t, time.Second, 20*time.Millisecond, func() error {
+		if !oc.inFlusher() {
+			return fmt.Errorf("not in flusher")
+		}
+		return nil
+	})
+
+	// Stop flusher for this benchmark since we will invoke directly.
+	oc.mu.Lock()
+	qch := oc.qch
+	oc.qch = nil
+	oc.mu.Unlock()
+	close(qch)
+
+	checkFor(t, time.Second, 20*time.Millisecond, func() error {
+		if oc.inFlusher() {
+			return fmt.Errorf("still in flusher")
+		}
+		return nil
+	})
+
+	toStore := uint64(1_000_000)
 
 	start := time.Now()
+
+	ts := start.UnixNano()
+
 	for i := uint64(1); i <= toStore; i++ {
-		state.Delivered.ConsumerSeq = i
-		state.Delivered.StreamSeq = i
-		state.AckFloor.ConsumerSeq = i
-		state.AckFloor.StreamSeq = i
-		if err := o.Update(state); err != nil {
-			t.Fatalf("Unexepected error updating state: %v", err)
+		if err := o.UpdateDelivered(i, i, 1, ts); err != nil {
+			t.Fatalf("Unexpected error: %v", err)
 		}
 	}
 	tt := time.Since(start)
-	fmt.Printf("time is %v\n", tt)
+	fmt.Printf("time to update %d is %v\n", toStore, tt)
 	fmt.Printf("%.0f updates/sec\n", float64(toStore)/tt.Seconds())
 
-	// We will lag behind with pending.
-	state.Pending = make(map[uint64]int64)
-	lag := uint64(100)
-	state.AckFloor.ConsumerSeq = 0
-	state.AckFloor.StreamSeq = 0
-
-	fmt.Printf("\nconsumer of %d msgs for ACK EXPLICIT with pending lag of %d\n", toStore, lag)
-
 	start = time.Now()
-	for i := uint64(1); i <= toStore; i++ {
-		state.Delivered.ConsumerSeq = i
-		state.Delivered.StreamSeq = i
-		state.Pending[i] = time.Now().UnixNano()
-		if i > lag {
-			ackseq := i - lag
-			state.AckFloor.ConsumerSeq = ackseq
-			state.AckFloor.StreamSeq = ackseq
-			delete(state.Pending, ackseq)
-		}
-		if err := o.Update(state); err != nil {
-			t.Fatalf("Unexepected error updating state: %v", err)
-		}
+	oc.mu.Lock()
+	buf, err := oc.encodeState()
+	oc.mu.Unlock()
+	if err != nil {
+		t.Fatalf("Error encoding state: %v", err)
 	}
-	tt = time.Since(start)
-	fmt.Printf("time is %v\n", tt)
-	fmt.Printf("%.0f updates/sec\n", float64(toStore)/tt.Seconds())
+	fmt.Printf("time to encode %d bytes is %v\n", len(buf), time.Since(start))
+	start = time.Now()
+	oc.writeState(buf)
+	fmt.Printf("time to write is %v\n", time.Since(start))
+	start = time.Now()
+	oc.syncStateFile()
+	fmt.Printf("time to sync is %v\n", time.Since(start))
 }
diff --git a/test/jetstream_test.go b/test/jetstream_test.go
index cc1bf640..ee80539a 100644
--- a/test/jetstream_test.go
+++ b/test/jetstream_test.go
@@ -638,6 +638,78 @@ func TestJetStreamPullConsumerDelayedFirstPullWithReplayOriginal(t *testing.T) {
 	}
 }
 
+func TestJetStreamConsumerAckFloorFill(t *testing.T) {
+	cases := []struct {
+		name    string
+		mconfig *server.StreamConfig
+	}{
+		{"MemoryStore", &server.StreamConfig{Name: "MQ", Storage: server.MemoryStorage}},
+		{"FileStore", &server.StreamConfig{Name: "MQ", Storage: server.FileStorage}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			s := RunBasicJetStreamServer()
+			defer s.Shutdown()
+
+			if config := s.JetStreamConfig(); config != nil {
+				defer os.RemoveAll(config.StoreDir)
+			}
+
+			mset, err := s.GlobalAccount().AddStream(c.mconfig)
+			if err != nil {
+				t.Fatalf("Unexpected error adding stream: %v", err)
+			}
+			defer mset.Delete()
+
+			nc := clientConnectToServer(t, s)
+			defer nc.Close()
+
+			for i := 1; i <= 4; i++ {
+				sendStreamMsg(t, nc, c.mconfig.Name, fmt.Sprintf("msg-%d", i))
+			}
+
+			sub, _ := nc.SubscribeSync(nats.NewInbox())
+			defer sub.Unsubscribe()
+			nc.Flush()
+
+			o, err := mset.AddConsumer(&server.ConsumerConfig{
+				Durable:        "d",
+				DeliverSubject: sub.Subject,
+				AckPolicy:      server.AckExplicit,
+			})
+			if err != nil {
+				t.Fatalf("Expected no error, got %v", err)
+			}
+			defer o.Delete()
+
+			var first *nats.Msg
+
+			for i := 1; i <= 3; i++ {
+				m, err := sub.NextMsg(time.Second)
+				if err != nil {
+					t.Fatalf("Error receiving message %d: %v", i, err)
+				}
+				// Don't ack 1 or 4.
+				if i == 1 {
+					first = m
+				} else if i == 2 || i == 3 {
+					m.Respond(nil)
+				}
+			}
+			nc.Flush()
+			if info := o.Info(); info.AckFloor.Consumer != 0 {
+				t.Fatalf("Expected the ack floor to be 0, got %d", info.AckFloor.Consumer)
+			}
+			// Now ack first, should move ack floor to 3.
+			first.Respond(nil)
+			nc.Flush()
+			if info := o.Info(); info.AckFloor.Consumer != 3 {
+				t.Fatalf("Expected the ack floor to be 3, got %d", info.AckFloor.Consumer)
+			}
+		})
+	}
+}
+
 func TestJetStreamNoPanicOnRaceBetweenShutdownAndConsumerDelete(t *testing.T) {
 	cases := []struct {
 		name    string
@@ -3081,7 +3153,7 @@ func TestJetStreamAckNext(t *testing.T) {
 	if len(q) != 0 {
 		t.Fatalf("Expected empty q got %d", len(q))
 	}
-	if o.Info().AckFloor.StreamSeq != 1 {
+	if o.Info().AckFloor.Stream != 1 {
 		t.Fatalf("First message was not acknowledged")
 	}
 	if !bytes.Equal(msg.Data, []byte("msg 1")) {
@@ -3124,7 +3196,7 @@ func TestJetStreamAckNext(t *testing.T) {
 
 	getMsgs(6, 10)
 
-	if o.Info().AckFloor.StreamSeq != 2 {
+	if o.Info().AckFloor.Stream != 2 {
 		t.Fatalf("second message was not acknowledged")
 	}
 }
@@ -3612,7 +3684,7 @@ func TestJetStreamConsumerMaxDeliveryAndServerRestart(t *testing.T) {
 		Durable:        "TO",
 		DeliverSubject: dsubj,
 		AckPolicy:      server.AckExplicit,
-		AckWait:        25 * time.Millisecond,
+		AckWait:        20 * time.Millisecond,
 		MaxDeliver:     max,
 	})
 	defer o.Delete()
@@ -3691,9 +3763,7 @@ func TestJetStreamConsumerMaxDeliveryAndServerRestart(t *testing.T) {
 	})
 
 	// Once we are here send third order.
-	// Send third
 	sendStreamMsg(t, nc, mname, "order-3")
-
 	checkNumMsgs(3)
 
 	// Restart.
@@ -3961,7 +4031,7 @@ func TestJetStreamSnapshots(t *testing.T) {
 	for _, oi := range info.obs {
 		if o := mset.LookupConsumer(oi.cfg.Durable); o != nil {
 			if uint64(oi.ack+1) != o.NextSeq() {
-				t.Fatalf("Consumer next seq is not correct: %d vs %d", oi.ack+1, o.NextSeq())
+				t.Fatalf("[%v] Consumer next seq is not correct: %d vs %d", o.Name(), oi.ack+1, o.NextSeq())
 			}
 		} else {
 			t.Fatalf("Expected to get an consumer")
@@ -5501,8 +5571,8 @@ func TestJetStreamStreamPurgeWithConsumer(t *testing.T) {
 				}
 			}
 			state := o.Info()
-			if state.AckFloor.ConsumerSeq != 50 {
-				t.Fatalf("Expected ack floor of 50, got %d", state.AckFloor.ConsumerSeq)
+			if state.AckFloor.Consumer != 50 {
+				t.Fatalf("Expected ack floor of 50, got %d", state.AckFloor.Consumer)
 			}
 			if state.NumAckPending != 25 {
 				t.Fatalf("Expected len(pending) to be 25, got %d", state.NumAckPending)
@@ -5519,15 +5589,15 @@ func TestJetStreamStreamPurgeWithConsumer(t *testing.T) {
 			if state.NumAckPending != 0 {
 				t.Fatalf("Expected no pending, got %d", state.NumAckPending)
 			}
-			if state.Delivered.StreamSeq != 100 {
-				t.Fatalf("Expected to have setseq now at next seq of 100, got %d", state.Delivered.StreamSeq)
+			if state.Delivered.Stream != 100 {
+				t.Fatalf("Expected to have setseq now at next seq of 100, got %d", state.Delivered.Stream)
 			}
 			// Check AckFloors which should have also been adjusted.
-			if state.AckFloor.StreamSeq != 100 {
-				t.Fatalf("Expected ackfloor for setseq to be 100, got %d", state.AckFloor.StreamSeq)
+			if state.AckFloor.Stream != 100 {
+				t.Fatalf("Expected ackfloor for setseq to be 100, got %d", state.AckFloor.Stream)
 			}
-			if state.AckFloor.ConsumerSeq != 75 {
-				t.Fatalf("Expected ackfloor for obsseq to be 75, got %d", state.AckFloor.ConsumerSeq)
+			if state.AckFloor.Consumer != 75 {
+				t.Fatalf("Expected ackfloor for obsseq to be 75, got %d", state.AckFloor.Consumer)
 			}
 			// Also make sure we can get new messages correctly.
 			nc.Request("DC", []byte("OK-22"), time.Second)
@@ -5607,15 +5677,15 @@ func TestJetStreamStreamPurgeWithConsumerAndRedelivery(t *testing.T) {
 			if state.NumAckPending != 0 {
 				t.Fatalf("Expected no pending, got %d", state.NumAckPending)
 			}
-			if state.Delivered.StreamSeq != 100 {
-				t.Fatalf("Expected to have setseq now at next seq of 100, got %d", state.Delivered.StreamSeq)
+			if state.Delivered.Stream != 100 {
+				t.Fatalf("Expected to have setseq now at next seq of 100, got %d", state.Delivered.Stream)
 			}
 			// Check AckFloors which should have also been adjusted.
-			if state.AckFloor.StreamSeq != 100 {
-				t.Fatalf("Expected ackfloor for setseq to be 100, got %d", state.AckFloor.StreamSeq)
+			if state.AckFloor.Stream != 100 {
+				t.Fatalf("Expected ackfloor for setseq to be 100, got %d", state.AckFloor.Stream)
 			}
-			if state.AckFloor.ConsumerSeq != 50 {
-				t.Fatalf("Expected ackfloor for obsseq to be 75, got %d", state.AckFloor.ConsumerSeq)
+			if state.AckFloor.Consumer != 50 {
+				t.Fatalf("Expected ackfloor for obsseq to be 75, got %d", state.AckFloor.Consumer)
 			}
 			// Also make sure we can get new messages correctly.
 			nc.Request("DC", []byte("OK-22"), time.Second)
@@ -7060,11 +7130,11 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	if oinfo.Config.DeliverSubject != delivery {
 		t.Fatalf("Expected to have delivery subject of %q, got %q", delivery, oinfo.Config.DeliverSubject)
 	}
-	if oinfo.Delivered.ConsumerSeq != 10 {
-		t.Fatalf("Expected consumer delivered sequence of 10, got %d", oinfo.Delivered.ConsumerSeq)
+	if oinfo.Delivered.Consumer != 10 {
+		t.Fatalf("Expected consumer delivered sequence of 10, got %d", oinfo.Delivered.Consumer)
 	}
-	if oinfo.AckFloor.ConsumerSeq != 10 {
-		t.Fatalf("Expected ack floor to be 10, got %d", oinfo.AckFloor.ConsumerSeq)
+	if oinfo.AckFloor.Consumer != 10 {
+		t.Fatalf("Expected ack floor to be 10, got %d", oinfo.AckFloor.Consumer)
 	}
 
 	// Now delete the consumer.
@@ -7920,7 +7990,7 @@ func TestJetStreamNextMsgNoInterest(t *testing.T) {
 			}
 			nc.Flush()
 			ostate := o.Info()
-			if ostate.AckFloor.StreamSeq != 11 || ostate.NumAckPending > 0 {
+			if ostate.AckFloor.Stream != 11 || ostate.NumAckPending > 0 {
 				t.Fatalf("Inconsistent ack state: %+v", ostate)
 			}
 		})
@@ -8642,6 +8712,77 @@ func TestJetStreamConsumerPerf(t *testing.T) {
 	fmt.Printf("%.0f msgs/sec\n", float64(toStore)/tt.Seconds())
 }
 
+func TestJetStreamConsumerAckFileStorePerf(t *testing.T) {
+	// Comment out to run, holding place for now.
+	t.SkipNow()
+
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer os.RemoveAll(config.StoreDir)
+	}
+
+	acc := s.GlobalAccount()
+
+	msetConfig := server.StreamConfig{
+		Name:     "sr22",
+		Storage:  server.FileStorage,
+		Subjects: []string{"foo"},
+	}
+
+	mset, err := acc.AddStream(&msetConfig)
+	if err != nil {
+		t.Fatalf("Unexpected error adding stream: %v", err)
+	}
+
+	nc := clientConnectToServer(t, s)
+	defer nc.Close()
+
+	payload := []byte("Hello World")
+
+	toStore := uint64(200000)
+	for i := uint64(0); i < toStore; i++ {
+		nc.Publish("foo", payload)
+	}
+	nc.Flush()
+
+	if msgs := mset.State().Msgs; msgs != uint64(toStore) {
+		t.Fatalf("Expected %d messages, got %d", toStore, msgs)
+	}
+
+	o, err := mset.AddConsumer(&server.ConsumerConfig{
+		Durable:        "d",
+		DeliverSubject: "d",
+		AckPolicy:      server.AckExplicit,
+		AckWait:        10 * time.Minute,
+	})
+	if err != nil {
+		t.Fatalf("Error creating consumer: %v", err)
+	}
+	defer o.Stop()
+
+	var received uint64
+	done := make(chan bool)
+
+	sub, _ := nc.Subscribe("d", func(m *nats.Msg) {
+		m.Respond(nil) // Ack
+		received++
+		if received >= toStore {
+			done <- true
+		}
+	})
+	sub.SetPendingLimits(-1, -1)
+
+	start := time.Now()
+	nc.Flush()
+
+	<-done
+	tt := time.Since(start)
+	fmt.Printf("time is %v\n", tt)
+	fmt.Printf("%.0f msgs/sec\n", float64(toStore)/tt.Seconds())
+}
+
 func TestJetStreamPubSubPerf(t *testing.T) {
 	// Comment out to run, holding place for now.
 	t.SkipNow()