Repository: nats-io/nats-server
Pull request #1718 (2020-11-17 18:59:40)

Updated fuzzer
The updates in the fuzzers fix the broken oss-fuzz build locally. Once merged I will run the oss-fuzz integration tests again.


Patch:
diff --git a/GOVERNANCE.md b/GOVERNANCE.md
index 152e14af..186d3d4f 100644
--- a/GOVERNANCE.md
+++ b/GOVERNANCE.md
@@ -1,3 +1,3 @@
 # NATS Server Governance
 
-NATS Server is part of the NATS project and is subject to the [NATS Governance](https://github.com/nats-io/nats-general/blob/master/GOVERNANCE.md).
+NATS Server (gnastd) is part of the NATS project and is subject to the [NATS Governance](https://github.com/nats-io/nats-general/blob/master/GOVERNANCE.md).
\ No newline at end of file
diff --git a/README.md b/README.md
index 75a174c9..d9b60244 100644
--- a/README.md
+++ b/README.md
@@ -29,8 +29,8 @@ If you are interested in contributing to NATS, read about our...
 [Fossa-Image]: https://app.fossa.io/api/projects/git%2Bgithub.com%2Fnats-io%2Fgnatsd.svg?type=shield
 [Build-Status-Url]: https://travis-ci.com/github/nats-io/nats-server
 [Build-Status-Image]: https://travis-ci.com/nats-io/nats-server.svg?branch=master
-[Release-Url]: https://github.com/nats-io/nats-server/releases/tag/v2.1.9
-[Release-image]: https://img.shields.io/badge/release-v2.1.9-1eb0fc.svg
+[Release-Url]: https://github.com/nats-io/nats-server/releases/tag/v2.1.8
+[Release-image]: https://img.shields.io/badge/release-v2.1.8-1eb0fc.svg
 [Coverage-Url]: https://coveralls.io/r/nats-io/nats-server?branch=master
 [Coverage-image]: https://coveralls.io/repos/github/nats-io/nats-server/badge.svg?branch=master
 [ReportCard-Url]: https://goreportcard.com/report/nats-io/nats-server
diff --git a/doc/adr/0004-nats-headers.md b/doc/adr/0004-nats-headers.md
index 134fbc75..ba8d5e9b 100644
--- a/doc/adr/0004-nats-headers.md
+++ b/doc/adr/0004-nats-headers.md
@@ -57,16 +57,16 @@ Messages that include a header have a `HPUB` protocol:
 ```
 HPUB SUBJECT REPLY 23 30␍␊NATS/1.0␍␊Header: X␍␊␍␊PAYLOAD␍␊
 HPUB SUBJECT REPLY 23 23␍␊NATS/1.0␍␊Header: X␍␊␍␊␍␊
-HPUB SUBJECT REPLY 48 55␍␊NATS/1.0␍␊Header1: X␍␊Header1: Y␍␊Header2: Z␍␊␍␊PAYLOAD␍␊
-HPUB SUBJECT REPLY 48 48␍␊NATS/1.0␍␊Header1: X␍␊Header1: Y␍␊Header2: Z␍␊␍␊␍␊
 
 HPUB <SUBJ> [REPLY] <HDR_LEN> <TOT_LEN>
 <HEADER><PAYLOAD>
+
 ```
 
 #### NOTES:
-- `HDR_LEN` includes the entire serialized header, from the start of the version string up to and including the ␍␊ before the payload
-- `TOT_LEN` the payload length plus the HDR_LEN
+
+- `HDR_LEN` includes the entire serialized header
+- `TOT_LEN` includes the entire payload length including the header
 
 ### MSG with Headers
 
@@ -74,16 +74,15 @@ Clients will see `HMSG` protocol lines for `MSG`s that contain headers
 
 ```
 HMSG SUBJECT 1 REPLY 23 30␍␊NATS/1.0␍␊Header: X␍␊␍␊PAYLOAD␍␊
-HMSG SUBJECT 1 REPLY 23 23␍␊NATS/1.0␍␊Header: X␍␊␍␊␍␊
-HMSG SUBJECT 1 REPLY 48 55␍␊NATS/1.0␍␊Header1: X␍␊Header1: Y␍␊Header2: Z␍␊␍␊PAYLOAD␍␊
-HMSG SUBJECT 1 REPLY 48 48␍␊NATS/1.0␍␊Header1: X␍␊Header1: Y␍␊Header2: Z␍␊␍␊␍␊
+HPUB SUBJECT 1 REPLY 23 23␍␊NATS/1.0␍␊Header: X␍␊␍␊␍␊
 
 HMSG <SUBJECT> <SID> [REPLY] <HDR_LEN> <TOT_LEN>
 <PAYLOAD>
 ```
 
-- `HDR_LEN` includes the entire serialized header, from the start of the version string up to and including the ␍␊ before the payload
-- `TOT_LEN` the payload length plus the HDR_LEN
+- `HDR_LEN` includes the entire header
+- `TOT_LEN` includes the entire payload length including the header
+
 
 ## Decision
 
diff --git a/server/client.go b/server/client.go
index 015f1562..fc3dcac8 100644
--- a/server/client.go
+++ b/server/client.go
@@ -3952,30 +3952,18 @@ func (c *client) processPingTimer() {
 
 	c.Debugf("%s Ping Timer", c.typeString())
 
-	var sendPing bool
-
 	// If we have had activity within the PingInterval then
 	// there is no need to send a ping. This can be client data
 	// or if we received a ping from the other side.
 	pingInterval := c.srv.getOpts().PingInterval
-	if c.kind == GATEWAY {
-		pingInterval = adjustPingIntervalForGateway(pingInterval)
-		sendPing = true
-	}
 	now := time.Now()
 	needRTT := c.rtt == 0 || now.Sub(c.rttStart) > DEFAULT_RTT_MEASUREMENT_INTERVAL
 
-	// Do not delay PINGs for GATEWAY connections.
-	if c.kind != GATEWAY {
-		if delta := now.Sub(c.last); delta < pingInterval && !needRTT {
-			c.Debugf("Delaying PING due to client activity %v ago", delta.Round(time.Second))
-		} else if delta := now.Sub(c.ping.last); delta < pingInterval && !needRTT {
-			c.Debugf("Delaying PING due to remote ping %v ago", delta.Round(time.Second))
-		} else {
-			sendPing = true
-		}
-	}
-	if sendPing {
+	if delta := now.Sub(c.last); delta < pingInterval && !needRTT {
+		c.Debugf("Delaying PING due to client activity %v ago", delta.Round(time.Second))
+	} else if delta := now.Sub(c.ping.last); delta < pingInterval && !needRTT {
+		c.Debugf("Delaying PING due to remote ping %v ago", delta.Round(time.Second))
+	} else {
 		// Check for violation
 		if c.ping.out+1 > c.srv.getOpts().MaxPingsOut {
 			c.Debugf("Stale Client Connection - Closing")
@@ -3993,24 +3981,12 @@ func (c *client) processPingTimer() {
 	c.mu.Unlock()
 }
 
-// Returns the smallest value between the given `d` and `gatewayMaxPingInterval` durations.
-// Invoked for connections known to be of GATEWAY type.
-func adjustPingIntervalForGateway(d time.Duration) time.Duration {
-	if d > gatewayMaxPingInterval {
-		return gatewayMaxPingInterval
-	}
-	return d
-}
-
 // Lock should be held
 func (c *client) setPingTimer() {
 	if c.srv == nil {
 		return
 	}
 	d := c.srv.getOpts().PingInterval
-	if c.kind == GATEWAY {
-		d = adjustPingIntervalForGateway(d)
-	}
 	c.ping.tmr = time.AfterFunc(d, c.processPingTimer)
 }
 
diff --git a/server/const.go b/server/const.go
index 4636eca1..073d648d 100644
--- a/server/const.go
+++ b/server/const.go
@@ -40,7 +40,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.2.0-beta.32"
+	VERSION = "2.2.0-beta.27"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index afc522ce..32c95f83 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -59,7 +59,6 @@ type ConsumerConfig struct {
 	RateLimit       uint64        `json:"rate_limit_bps,omitempty"` // Bits per sec
 	SampleFrequency string        `json:"sample_freq,omitempty"`
 	MaxWaiting      int           `json:"max_waiting,omitempty"`
-	MaxAckPending   int           `json:"max_ack_pending,omitempty"`
 }
 
 type CreateConsumerRequest struct {
@@ -159,6 +158,8 @@ var (
 	AckNext = []byte("+NXT")
 	// Terminate delivery of the message.
 	AckTerm = []byte("+TERM")
+
+	ackNextCnt = "+NXT %d+"
 )
 
 // Consumer is a jetstream consumer.
@@ -172,16 +173,14 @@ type Consumer struct {
 	dseq              uint64
 	adflr             uint64
 	asflr             uint64
-	sgap              uint64
+	spending          uint64
 	dsubj             string
 	rlimit            *rate.Limiter
 	reqSub            *subscription
 	ackSub            *subscription
 	ackReplyT         string
 	nextMsgSubj       string
-	maxp              int
-	sendq             chan *jsPubMsg
-	pending           map[uint64]*Pending
+	pending           map[uint64]int64
 	ptmr              *time.Timer
 	rdq               []uint64
 	rdc               map[uint64]uint64
@@ -194,7 +193,7 @@ type Consumer struct {
 	filterWC          bool
 	dtmr              *time.Timer
 	dthresh           time.Duration
-	mch               chan struct{}
+	fch               chan struct{}
 	qch               chan struct{}
 	inch              chan bool
 	sfreq             int32
@@ -229,9 +228,6 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 		if config.MaxWaiting != 0 {
 			return nil, fmt.Errorf("consumer in push mode can not set max waiting")
 		}
-		if config.MaxAckPending > 0 && config.AckPolicy == AckNone {
-			return nil, fmt.Errorf("consumer requires ack policy for max ack pending")
-		}
 	} else {
 		// Pull mode / work queue mode require explicit ack.
 		if config.AckPolicy != AckExplicit {
@@ -391,10 +387,9 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 		dsubj:   config.DeliverSubject,
 		active:  true,
 		qch:     make(chan struct{}),
-		mch:     make(chan struct{}, 1),
+		fch:     make(chan struct{}, 1),
 		sfreq:   int32(sampleFreq),
 		maxdc:   uint64(config.MaxDeliver),
-		maxp:    config.MaxAckPending,
 		created: time.Now().UTC(),
 	}
 	if isDurableConsumer(config) {
@@ -519,7 +514,7 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 	// If push mode, register for notifications on interest.
 	if o.isPushMode() {
 		o.dthresh = JsDeleteWaitTimeDefault
-		o.inch = make(chan bool, 8)
+		o.inch = make(chan bool, 4)
 		a.sl.RegisterNotification(config.DeliverSubject, o.inch)
 		o.active = o.hasDeliveryInterest(<-o.inch)
 		// Check if we are not durable that the delivery subject has interest.
@@ -534,13 +529,10 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 		o.replay = true
 	}
 
-	// Create internal sendq
-	o.sendq = make(chan *jsPubMsg, msetSendQSize)
-
 	// Now start up Go routine to deliver msgs.
-	go o.loopAndGatherMsgs(s, a)
-	// Startup our deliver loop.
-	go o.loopAndDeliverMsgs()
+	go o.loopAndDeliverMsgs(s, a)
+	// Startup our state update loop.
+	go o.updateStateLoop()
 
 	o.sendCreateAdvisory()
 
@@ -551,12 +543,12 @@ func (mset *Stream) AddConsumer(config *ConsumerConfig) (*Consumer, error) {
 // Do all advisory sends here.
 // Lock should be held on entry but will be released.
 func (o *Consumer) sendAdvisory(subj string, msg []byte) {
-	sendq := o.sendq
-	o.mu.Unlock()
-	if sendq != nil {
+	if o.mset != nil && o.mset.sendq != nil {
+		sendq := o.mset.sendq
+		o.mu.Unlock()
 		sendq <- &jsPubMsg{subj, subj, _EMPTY_, nil, msg, nil, 0}
+		o.mu.Lock()
 	}
-	o.mu.Lock()
 }
 
 func (o *Consumer) sendDeleteAdvisoryLocked() {
@@ -658,16 +650,12 @@ func (o *Consumer) updateDeliveryInterest(localInterest bool) {
 	interest := o.hasDeliveryInterest(localInterest)
 
 	o.mu.Lock()
-	defer o.mu.Unlock()
-
 	mset := o.mset
 	if mset == nil || o.isPullMode() {
+		o.mu.Unlock()
 		return
 	}
-
-	if interest && !o.active {
-		o.signalNewMessages()
-	}
+	shouldSignal := interest && !o.active
 	o.active = interest
 
 	// Stop and clear the delete timer always.
@@ -678,6 +666,11 @@ func (o *Consumer) updateDeliveryInterest(localInterest bool) {
 	if !o.isDurable() && !interest {
 		o.dtmr = time.AfterFunc(o.dthresh, func() { o.Delete() })
 	}
+	o.mu.Unlock()
+
+	if shouldSignal {
+		mset.signalConsumers()
+	}
 }
 
 // Config returns the consumer's configuration.
@@ -687,32 +680,6 @@ func (o *Consumer) Config() ConsumerConfig {
 	return o.config
 }
 
-// Force expiration of all pending.
-// Lock should be held.
-func (o *Consumer) forceExpirePending() {
-	now := time.Now().UnixNano()
-	var expired []uint64
-	for seq := range o.pending {
-		if !o.onRedeliverQueue(seq) {
-			expired = append(expired, seq)
-		}
-	}
-	if len(expired) > 0 {
-		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]].Timestamp
-		for _, seq := range expired {
-			if p, ok := o.pending[seq]; ok && p != nil {
-				p.Timestamp += off
-			}
-		}
-		o.ptmr.Reset(o.ackWait(0))
-	}
-	o.signalNewMessages()
-}
-
 // This is a config change for the delivery subject for a
 // push based consumer.
 func (o *Consumer) updateDeliverSubject(newDeliver string) {
@@ -720,15 +687,10 @@ func (o *Consumer) updateDeliverSubject(newDeliver string) {
 	o.mu.Lock()
 	defer o.mu.Unlock()
 
-	if o.closed || o.isPullMode() || o.config.DeliverSubject == newDeliver {
+	if o.closed || o.isPullMode() {
 		return
 	}
 
-	// Force redeliver of all pending on change of delivery subject.
-	if len(o.pending) > 0 {
-		o.forceExpirePending()
-	}
-
 	o.acc.sl.ClearNotification(o.dsubj, o.inch)
 	o.dsubj, o.config.DeliverSubject = newDeliver, newDeliver
 	// When we register new one it will deliver to update state loop.
@@ -760,7 +722,7 @@ func (o *Consumer) processAck(_ *subscription, _ *client, subject, reply string,
 		o.ackMsg(sseq, dseq, dcount)
 	case bytes.HasPrefix(msg, AckNext):
 		o.ackMsg(sseq, dseq, dcount)
-		o.processNextMsgReq(nil, nil, subject, reply, msg[len(AckNext):])
+		o.processNextMsgReq(nil, nil, subject, reply, msg)
 		skipAckReply = true
 	case bytes.Equal(msg, AckNak):
 		o.processNak(sseq, dseq)
@@ -780,10 +742,8 @@ func (o *Consumer) processAck(_ *subscription, _ *client, subject, reply string,
 func (o *Consumer) progressUpdate(seq uint64) {
 	o.mu.Lock()
 	if len(o.pending) > 0 {
-		if p, ok := o.pending[seq]; ok {
-			p.Timestamp = time.Now().UnixNano()
-			// Update store system.
-			o.store.UpdateDelivered(p.Sequence, seq, 1, p.Timestamp)
+		if _, ok := o.pending[seq]; ok {
+			o.pending[seq] = time.Now().UnixNano()
 		}
 	}
 	o.mu.Unlock()
@@ -791,25 +751,29 @@ func (o *Consumer) progressUpdate(seq uint64) {
 
 // Process a NAK.
 func (o *Consumer) processNak(sseq, dseq uint64) {
+	var mset *Stream
 	o.mu.Lock()
-	defer o.mu.Unlock()
-
 	// Check for out of range.
 	if dseq <= o.adflr || dseq > o.dseq {
+		o.mu.Unlock()
 		return
 	}
-	// If we are explicit ack make sure this is still on our pending list.
+	// If we are explicit ack make sure this is still on pending list.
 	if len(o.pending) > 0 {
 		if _, ok := o.pending[sseq]; !ok {
+			o.mu.Unlock()
 			return
 		}
 	}
 	// If already queued up also ignore.
 	if !o.onRedeliverQueue(sseq) {
 		o.rdq = append(o.rdq, sseq)
+		mset = o.mset
+	}
+	o.mu.Unlock()
+	if mset != nil {
+		mset.signalConsumers()
 	}
-
-	o.signalNewMessages()
 }
 
 // Process a TERM
@@ -863,10 +827,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.Consumer + 1
-		o.sseq = state.Delivered.Stream + 1
-		o.adflr = state.AckFloor.Consumer
-		o.asflr = state.AckFloor.Stream
+		o.dseq = state.Delivered.ConsumerSeq
+		o.sseq = state.Delivered.StreamSeq
+		o.adflr = state.AckFloor.ConsumerSeq
+		o.asflr = state.AckFloor.StreamSeq
 		o.pending = state.Pending
 		o.rdc = state.Redelivered
 	}
@@ -886,12 +850,12 @@ func (o *Consumer) writeState() {
 	if o.store != nil {
 		state := ConsumerState{
 			Delivered: SequencePair{
-				Consumer: o.dseq - 1,
-				Stream:   o.sseq - 1,
+				ConsumerSeq: o.dseq,
+				StreamSeq:   o.sseq,
 			},
 			AckFloor: SequencePair{
-				Consumer: o.adflr,
-				Stream:   o.asflr,
+				ConsumerSeq: o.adflr,
+				StreamSeq:   o.asflr,
 			},
 			Pending:     o.pending,
 			Redelivered: o.rdc,
@@ -902,31 +866,14 @@ func (o *Consumer) writeState() {
 	o.mu.Unlock()
 }
 
-// loopAndDeliverMsgs() will loop and deliver messages and watch for interest changes.
-func (o *Consumer) loopAndDeliverMsgs() {
+func (o *Consumer) updateStateLoop() {
 	o.mu.Lock()
-	qch, inch, sendq := o.qch, o.inch, o.sendq
-	s, acc := o.acc.srv, o.acc
+	fch := o.fch
+	qch := o.qch
+	inch := o.inch
 	o.mu.Unlock()
 
-	// Create our client used to send messages.
-	c := s.createInternalJetStreamClient()
-	// Bind to the account.
-	c.registerWithAccount(acc)
-	// Clean up on exit.
-	defer c.closeConnection(ClientClosed)
-
-	// Warn when internal send queue is backed up past 75%
-	warnThresh := cap(sendq) * 3 / 4
-	warnFreq := time.Second
-	last := time.Now().Add(-warnFreq)
-
 	for {
-		if len(sendq) > warnThresh && time.Since(last) >= warnFreq {
-			s.Warnf("Jetstream internal consumer send queue > 75%% for account: %q consumer: %q", acc.Name, o.Name())
-			last = time.Now()
-		}
-
 		select {
 		case <-qch:
 			return
@@ -934,37 +881,10 @@ func (o *Consumer) loopAndDeliverMsgs() {
 			// inch can be nil on pull-based, but then this will
 			// just block and not fire.
 			o.updateDeliveryInterest(interest)
-		case pm := <-sendq:
-			if pm == nil {
-				return
-			}
-			c.pa.subject = []byte(pm.subj)
-			c.pa.deliver = []byte(pm.dsubj)
-			c.pa.size = len(pm.msg) + len(pm.hdr)
-			c.pa.szb = []byte(strconv.Itoa(c.pa.size))
-			c.pa.reply = []byte(pm.reply)
-
-			var msg []byte
-			if len(pm.hdr) > 0 {
-				c.pa.hdr = len(pm.hdr)
-				c.pa.hdb = []byte(strconv.Itoa(c.pa.hdr))
-				msg = append(pm.hdr, pm.msg...)
-				msg = append(msg, _CRLF_...)
-			} else {
-				c.pa.hdr = -1
-				c.pa.hdb = nil
-				msg = append(pm.msg, _CRLF_...)
-			}
-
-			didDeliver := c.processInboundClientMsg(msg)
-			c.pa.szb = nil
-			c.flushClients(0)
-
-			// Check to see if this is a delivery for an observable and
-			// we failed to deliver the message. If so alert the observable.
-			if !didDeliver && pm.o != nil && pm.seq > 0 {
-				pm.o.didNotDeliver(pm.seq)
-			}
+		case <-fch:
+			// FIXME(dlc) - Do better, check for fast changes at quick intervals.
+			time.Sleep(10 * time.Millisecond)
+			o.writeState()
 		}
 	}
 }
@@ -978,16 +898,16 @@ func (o *Consumer) Info() *ConsumerInfo {
 		Created: o.created,
 		Config:  o.config,
 		Delivered: SequencePair{
-			Consumer: o.dseq - 1,
-			Stream:   o.sseq - 1,
+			ConsumerSeq: o.dseq - 1,
+			StreamSeq:   o.sseq - 1,
 		},
 		AckFloor: SequencePair{
-			Consumer: o.adflr,
-			Stream:   o.asflr,
+			ConsumerSeq: o.adflr,
+			StreamSeq:   o.asflr,
 		},
 		NumAckPending:  len(o.pending),
 		NumRedelivered: len(o.rdc),
-		NumPending:     o.sgap,
+		NumPending:     o.spending,
 	}
 	// If we are a pull mode consumer, report on number of waiting requests.
 	if o.isPullMode() {
@@ -997,11 +917,15 @@ func (o *Consumer) Info() *ConsumerInfo {
 	return info
 }
 
-// Will signal us that new messages are available. Will break out of waiting.
-func (o *Consumer) signalNewMessages() {
-	// Kick our new message channel
+// Will update the underlying store.
+// Lock should be held.
+func (o *Consumer) updateStore() {
+	if o.store == nil {
+		return
+	}
+	// Kick our flusher
 	select {
-	case o.mch <- struct{}{}:
+	case o.fch <- struct{}{}:
 	default:
 	}
 }
@@ -1038,7 +962,7 @@ func (o *Consumer) sampleAck(sseq, dseq, dcount uint64) {
 		Consumer:    o.name,
 		ConsumerSeq: dseq,
 		StreamSeq:   sseq,
-		Delay:       unow - o.pending[sseq].Timestamp,
+		Delay:       unow - o.pending[sseq],
 		Deliveries:  dcount,
 	}
 
@@ -1056,40 +980,24 @@ func (o *Consumer) ackMsg(sseq, dseq, dcount uint64) {
 }
 
 func (o *Consumer) processAckMsg(sseq, dseq, dcount uint64, doSample bool) {
-	o.mu.Lock()
 	var sagap uint64
-	var needSignal bool
 
+	o.mu.Lock()
 	switch o.config.AckPolicy {
 	case AckExplicit:
-		if p, ok := o.pending[sseq]; ok {
+		if _, ok := o.pending[sseq]; ok {
 			if doSample {
 				o.sampleAck(sseq, dseq, dcount)
 			}
-			if o.maxp > 0 && len(o.pending) >= o.maxp {
-				needSignal = true
-			}
 			delete(o.pending, sseq)
-			// Use the original deliver sequence from our pending record.
-			dseq = p.Sequence
-			// Consumers sequence numbers can skip during re-delivery since
+			// Consumers sequence numbers can skip during redlivery 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.
 		delete(o.rdc, sseq)
@@ -1100,9 +1008,6 @@ func (o *Consumer) processAckMsg(sseq, dseq, dcount uint64, doSample bool) {
 			o.mu.Unlock()
 			return
 		}
-		if o.maxp > 0 && len(o.pending) >= o.maxp {
-			needSignal = true
-		}
 		sagap = sseq - o.asflr
 		o.adflr, o.asflr = dseq, sseq
 		for seq := sseq; seq > sseq-sagap; seq-- {
@@ -1115,9 +1020,7 @@ func (o *Consumer) processAckMsg(sseq, dseq, dcount uint64, doSample bool) {
 		o.mu.Unlock()
 		return
 	}
-
-	// Update underlying store.
-	o.store.UpdateAcks(dseq, sseq)
+	o.updateStore()
 
 	mset := o.mset
 	o.mu.Unlock()
@@ -1133,11 +1036,6 @@ func (o *Consumer) processAckMsg(sseq, dseq, dcount uint64, doSample bool) {
 			mset.ackMsg(o, sseq)
 		}
 	}
-
-	// If we had max ack pending set and were at limit we need to unblock folks.
-	if needSignal {
-		o.signalNewMessages()
-	}
 }
 
 // Check if we need an ack for this store seq.
@@ -1163,26 +1061,32 @@ func (o *Consumer) needAck(sseq uint64) bool {
 
 // Helper for the next message requests.
 func nextReqFromMsg(msg []byte) (time.Time, int, bool, error) {
-	req := bytes.TrimSpace(msg)
-
-	switch {
-	case len(req) == 0:
+	req := strings.TrimSpace(string(msg))
+	if len(req) == 0 {
 		return time.Time{}, 1, false, nil
-
-	case req[0] == '{':
+	}
+	if req[0] == '{' {
 		var cr JSApiConsumerGetNextRequest
-		if err := json.Unmarshal(req, &cr); err != nil {
+		if err := json.Unmarshal(msg, &cr); err != nil {
 			return time.Time{}, -1, false, err
 		}
 		return cr.Expires, cr.Batch, cr.NoWait, nil
+	}
 
+	bs := 1
+	// Naked batch size here for backward compatibility.
+	switch {
+	case strings.HasPrefix(req, string(AckNext)):
+		if n, _ := fmt.Sscanf(req, ackNextCnt, &bs); n == 0 {
+			bs = 1
+		}
 	default:
-		if n, err := strconv.Atoi(string(req)); err == nil {
-			return time.Time{}, n, false, nil
+		if n, err := strconv.Atoi(req); err == nil {
+			bs = n
 		}
 	}
 
-	return time.Time{}, 1, false, nil
+	return time.Time{}, bs, false, nil
 }
 
 // Represents a request that is on the internal waiting queue
@@ -1274,8 +1178,8 @@ func (wq *waitQueue) pop() *waitingRequest {
 }
 
 // processNextMsgReq will process a request for the next message available. A nil message payload means deliver
-// a single message. If the payload is a formal request or a number parseable with Atoi(), then we will send a
-// batch of messages without requiring another request to this endpoint, or an ACK.
+// a single message. If the payload is a number parseable with Atoi(), then we will send a batch of messages without
+// requiring another request to this endpoint, or an ACK.
 func (o *Consumer) processNextMsgReq(_ *subscription, c *client, _, reply string, msg []byte) {
 	o.mu.Lock()
 	mset := o.mset
@@ -1285,10 +1189,11 @@ func (o *Consumer) processNextMsgReq(_ *subscription, c *client, _, reply string
 	}
 
 	sendErr := func(status int, description string) {
+		sendq := mset.sendq
 		o.mu.Unlock()
 		hdr := []byte(fmt.Sprintf("NATS/1.0 %d %s\r\n\r\n", status, description))
 		pmsg := &jsPubMsg{reply, reply, _EMPTY_, hdr, nil, nil, 0}
-		o.sendq <- pmsg // Send message.
+		sendq <- pmsg // Send message.
 	}
 
 	if o.waiting.isFull() {
@@ -1313,24 +1218,18 @@ func (o *Consumer) processNextMsgReq(_ *subscription, c *client, _, reply string
 	if o.replay {
 		o.waiting.add(&wr)
 		o.mu.Unlock()
-		o.signalNewMessages()
+		mset.signalConsumers()
 		return
 	}
 
 	for i := 0; i < batchSize; i++ {
-		// See if we have more messages available.
 		if subj, hdr, msg, seq, dc, ts, err := o.getNextMsg(); err == nil {
 			o.deliverMsg(reply, subj, hdr, msg, seq, dc, ts)
 			// Need to discount this from the total n for the request.
 			wr.n--
 		} else {
 			if wr.noWait {
-				switch err {
-				case errMaxAckPending:
-					sendErr(409, "Exceeded MaxAckPending")
-				default:
-					sendErr(404, "No Messages")
-				}
+				sendErr(404, "No Messages")
 				return
 			}
 			o.waiting.add(&wr)
@@ -1388,15 +1287,12 @@ func (o *Consumer) isFilteredMatch(subj string) bool {
 	return subjectIsSubsetMatch(subj, o.config.FilterSubject)
 }
 
-var errMaxAckPending = errors.New("max ack pending reached")
-var errBadConsumer = errors.New("consumer not valid")
-
 // Get next available message from underlying store.
 // Is partition aware and redeliver aware.
 // Lock should be held.
 func (o *Consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dcount uint64, ts int64, err error) {
 	if o.mset == nil {
-		return _EMPTY_, nil, nil, 0, 0, 0, errBadConsumer
+		return _EMPTY_, nil, nil, 0, 0, 0, fmt.Errorf("consumer not valid")
 	}
 	for {
 		seq, dcount := o.sseq, uint64(1)
@@ -1413,12 +1309,7 @@ func (o *Consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dcoun
 				delete(o.pending, seq)
 				continue
 			}
-		} else if o.maxp > 0 && len(o.pending) >= o.maxp {
-			// maxp only set when ack policy != AckNone and user set MaxAckPending
-			// Stall if we have hit max pending.
-			return _EMPTY_, nil, nil, 0, 0, 0, errMaxAckPending
 		}
-
 		subj, hdr, msg, ts, err := o.mset.store.LoadMsg(seq)
 		if err == nil {
 			if dcount == 1 { // First delivery.
@@ -1451,9 +1342,12 @@ func (o *Consumer) forceExpireFirstWaiting() *waitingRequest {
 	// If we are expiring this and we think there is still interest, alert.
 	if rr := o.acc.sl.Match(wr.reply); len(rr.psubs)+len(rr.qsubs) > 0 && o.mset != nil {
 		// We still appear to have interest, so send alert as courtesy.
+		sendq := o.mset.sendq
+		o.mu.Unlock()
 		hdr := []byte("NATS/1.0 408 Request Timeout\r\n\r\n")
 		pmsg := &jsPubMsg{wr.reply, wr.reply, _EMPTY_, hdr, nil, nil, 0}
-		o.sendq <- pmsg // Send message.
+		sendq <- pmsg // Send message.
+		o.mu.Lock()
 	}
 	return wr
 }
@@ -1485,7 +1379,7 @@ func (o *Consumer) checkWaitingForInterest() bool {
 	return o.waiting.len() > 0
 }
 
-func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
+func (o *Consumer) loopAndDeliverMsgs(s *Server, a *Account) {
 	// On startup check to see if we are in a a reply situtation where replay policy is not instant.
 	var (
 		lts  int64 // last time stamp seen, used for replay.
@@ -1506,7 +1400,8 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 	// Deliver all the msgs we have now, once done or on a condition, we wait for new ones.
 	for {
 		var (
-			seq, dcount uint64
+			mset        *Stream
+			seq, dcnt   uint64
 			subj, dsubj string
 			hdr         []byte
 			msg         []byte
@@ -1521,6 +1416,7 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 			o.mu.Unlock()
 			return
 		}
+		mset = o.mset
 
 		// If we are in push mode and not active let's stop sending.
 		if o.isPushMode() && !o.active {
@@ -1532,11 +1428,11 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 			goto waitForMsgs
 		}
 
-		subj, hdr, msg, seq, dcount, ts, err = o.getNextMsg()
+		subj, hdr, msg, seq, dcnt, ts, err = o.getNextMsg()
 
 		// On error either wait or return.
 		if err != nil {
-			if err == ErrStoreMsgNotFound || err == ErrStoreEOF || err == errMaxAckPending {
+			if err == ErrStoreMsgNotFound || err == ErrStoreEOF {
 				goto waitForMsgs
 			} else {
 				o.mu.Unlock()
@@ -1584,7 +1480,7 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 			}
 		}
 
-		o.deliverMsg(dsubj, subj, hdr, msg, seq, dcount, ts)
+		o.deliverMsg(dsubj, subj, hdr, msg, seq, dcnt, ts)
 
 		o.mu.Unlock()
 		continue
@@ -1594,17 +1490,9 @@ func (o *Consumer) loopAndGatherMsgs(s *Server, a *Account) {
 		if o.replay && o.sseq > lseq {
 			o.replay = false
 		}
-
 		// We will wait here for new messages to arrive.
-		mch := o.mch
-		qch := o.qch
 		o.mu.Unlock()
-
-		select {
-		case <-qch:
-			return
-		case <-mch:
-		}
+		mset.waitForMsgs()
 	}
 }
 
@@ -1633,13 +1521,6 @@ func (o *Consumer) deliverCurrentMsg(subj string, hdr, msg []byte, seq uint64, t
 		return false
 	}
 
-	// Since we short circuit the getNextMsg() call where we check for max pending
-	// we need to do that here as well.
-	if o.maxp > 0 && len(o.pending) >= o.maxp {
-		o.mu.Unlock()
-		return false
-	}
-
 	// Bump store sequence here.
 	o.sseq++
 
@@ -1657,6 +1538,10 @@ func (o *Consumer) deliverCurrentMsg(subj string, hdr, msg []byte, seq uint64, t
 		dsubj = o.dsubj
 	}
 
+	if len(msg) > 0 {
+		msg = append(msg[:0:0], msg...)
+	}
+
 	o.deliverMsg(dsubj, subj, hdr, msg, seq, 1, ts)
 	o.mu.Unlock()
 
@@ -1670,35 +1555,28 @@ func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount u
 		return
 	}
 	// Update pending on first attempt
-	if dcount == 1 && o.sgap > 0 {
-		o.sgap--
-	}
-
-	if len(hdr) > 0 {
-		hdr = append(hdr[:0:0], hdr...)
-	}
-	if len(msg) > 0 {
-		msg = append(msg[:0:0], msg...)
+	if dcount == 1 && o.spending > 0 {
+		o.spending--
 	}
 
-	dseq := o.dseq
-	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, dseq, dcount, ts, o.sgap), hdr, msg, o, seq}
+	pmsg := &jsPubMsg{dsubj, subj, o.ackReply(seq, o.dseq, dcount, ts, o.spending), hdr, msg, o, seq}
 	mset := o.mset
+	sendq := mset.sendq
 	ap := o.config.AckPolicy
 
 	// This needs to be unlocked since the other side may need this lock on a failed delivery.
 	o.mu.Unlock()
 	// Send message.
-	o.sendq <- pmsg
+	sendq <- pmsg
+
 	// If we are ack none and mset is interest only we should make sure stream removes interest.
 	if ap == AckNone && mset.config.Retention == InterestPolicy && !mset.checkInterest(seq, o) {
 		mset.store.RemoveMsg(seq)
 	}
-	// Re-acquire lock.
 	o.mu.Lock()
 
 	if ap == AckExplicit || ap == AckAll {
-		o.trackPending(seq, dseq)
+		o.trackPending(seq)
 	} else if ap == AckNone {
 		o.adflr = o.dseq
 		o.asflr = seq
@@ -1706,24 +1584,19 @@ func (o *Consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dcount u
 
 	o.dseq++
 
-	// FIXME(dlc) - Capture errors?
-	o.store.UpdateDelivered(dseq, seq, dcount, ts)
+	o.updateStore()
 }
 
 // Tracks our outstanding pending acks. Only applicable to AckExplicit mode.
 // Lock should be held.
-func (o *Consumer) trackPending(sseq, dseq uint64) {
+func (o *Consumer) trackPending(seq uint64) {
 	if o.pending == nil {
-		o.pending = make(map[uint64]*Pending)
+		o.pending = make(map[uint64]int64)
 	}
 	if o.ptmr == nil {
 		o.ptmr = time.AfterFunc(o.ackWait(0), o.checkPending)
 	}
-	if p, ok := o.pending[sseq]; ok {
-		p.Timestamp = time.Now().UnixNano()
-	} else {
-		o.pending[sseq] = &Pending{dseq, time.Now().UnixNano()}
-	}
+	o.pending[seq] = time.Now().UnixNano()
 }
 
 // didNotDeliver is called when a delivery for a consumer message failed.
@@ -1736,21 +1609,25 @@ func (o *Consumer) didNotDeliver(seq uint64) {
 		return
 	}
 
+	shouldSignal := false
 	if o.isPushMode() {
 		o.active = false
 	} else if o.pending != nil {
-		// pull mode and we have pending.
+		// push mode and we have pending.
 		if _, ok := o.pending[seq]; ok {
 			// We found this messsage on pending, we need
 			// to queue it up for immediate redelivery since
 			// we know it was not delivered.
 			if !o.onRedeliverQueue(seq) {
 				o.rdq = append(o.rdq, seq)
-				o.signalNewMessages()
+				shouldSignal = true
 			}
 		}
 	}
 	o.mu.Unlock()
+	if shouldSignal {
+		mset.signalConsumers()
+	}
 }
 
 // This checks if we already have this sequence queued for redelivery.
@@ -1780,25 +1657,26 @@ func (o *Consumer) removeFromRedeliverQueue(seq uint64) bool {
 // Checks the pending messages.
 func (o *Consumer) checkPending() {
 	o.mu.Lock()
-	defer o.mu.Unlock()
-
 	mset := o.mset
 	if mset == nil {
+		o.mu.Unlock()
 		return
 	}
 	ttl := int64(o.config.AckWait)
 	next := int64(o.ackWait(0))
 	now := time.Now().UnixNano()
+	shouldSignal := false
 
 	// 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, p := range o.pending {
-		elapsed := now - p.Timestamp
+	for seq, ts := range o.pending {
+		elapsed := now - ts
 		if elapsed >= ttl {
 			if !o.onRedeliverQueue(seq) {
 				expired = append(expired, seq)
-				o.signalNewMessages()
+				shouldSignal = true
 			}
 		} else if ttl-elapsed < next {
 			// Update when we should fire next.
@@ -1807,16 +1685,13 @@ 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]].Timestamp
+		off := now - o.pending[expired[0]]
 		for _, seq := range expired {
-			if p, ok := o.pending[seq]; ok {
-				p.Timestamp += off
-			}
+			o.pending[seq] += off
 		}
 	}
 
@@ -1826,6 +1701,11 @@ func (o *Consumer) checkPending() {
 		o.ptmr.Stop()
 		o.ptmr = nil
 	}
+	o.mu.Unlock()
+
+	if shouldSignal {
+		mset.signalConsumers()
+	}
 }
 
 // SeqFromReply will extract a sequence number from a reply subject.
@@ -1981,7 +1861,7 @@ func (o *Consumer) purge(sseq uint64) {
 	o.sseq = sseq
 	o.asflr = sseq - 1
 	o.adflr = o.dseq - 1
-	o.sgap = 0
+	o.spending = 0
 	if len(o.pending) > 0 {
 		o.pending = nil
 		if o.ptmr != nil {
@@ -2002,8 +1882,6 @@ func (o *Consumer) purge(sseq uint64) {
 		o.rdq = newRDQ
 	}
 	o.mu.Unlock()
-
-	o.writeState()
 }
 
 func stopAndClearTimer(tp **time.Timer) {
@@ -2057,10 +1935,6 @@ func (o *Consumer) stop(dflag, doSignal, advisory bool) error {
 	stopAndClearTimer(&o.dtmr)
 	delivery := o.config.DeliverSubject
 	o.waiting = nil
-	// Break us out of the readLoop.
-	if doSignal {
-		o.signalNewMessages()
-	}
 	o.mu.Unlock()
 
 	if delivery != "" {
@@ -2068,6 +1942,13 @@ func (o *Consumer) stop(dflag, doSignal, advisory bool) error {
 	}
 
 	mset.mu.Lock()
+	// Break us out of the readLoop.
+	// TODO(dlc) - Should not be bad for small amounts of observables, maybe
+	// even into thousands. Above that should check what this might do
+	// performance wise.
+	if doSignal {
+		mset.sg.Broadcast()
+	}
 	mset.unsubscribe(ackSub)
 	mset.unsubscribe(reqSub)
 	mset.deleteConsumer(o)
@@ -2094,6 +1975,11 @@ 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 {
@@ -2178,7 +2064,7 @@ func (o *Consumer) setInitialPending() {
 	if o.config.FilterSubject == _EMPTY_ {
 		state := mset.store.State()
 		if state.Msgs > 0 {
-			o.sgap = state.Msgs - (o.sseq - state.FirstSeq)
+			o.spending = state.Msgs - (o.sseq - state.FirstSeq)
 		}
 	} else {
 		// Here we are filtered.
@@ -2190,7 +2076,7 @@ func (o *Consumer) setInitialPending() {
 			} else if err == ErrStoreEOF {
 				break
 			} else if err == nil && o.isFilteredMatch(subj) {
-				o.sgap++
+				o.spending++
 			}
 		}
 	}
@@ -2200,20 +2086,24 @@ func (o *Consumer) setInitialPending() {
 func (o *Consumer) incStreamPending(sseq uint64, subj string) {
 	o.mu.Lock()
 	if o.isFilteredMatch(subj) {
-		o.sgap++
+		o.spending++
 	}
 	o.mu.Unlock()
 }
 
-func (o *Consumer) decStreamPending(sseq uint64, subj string) {
+func (o *Consumer) decStreamPending(sseq uint64) {
 	o.mu.Lock()
-	defer o.mu.Unlock()
-
-	// Ignore if we have already seen this one.
-	if sseq < o.sseq || o.sgap == 0 {
+	// Ignore if we have already sent this one.
+	if sseq < o.sseq || o.spending == 0 {
+		o.mu.Unlock()
 		return
 	}
-	if o.isFilteredMatch(subj) {
-		o.sgap--
+	if o.config.FilterSubject == _EMPTY_ {
+		o.spending--
+	} else if o.mset != nil {
+		if subj, _, _, _, _ := o.mset.store.LoadMsg(sseq); o.isFilteredMatch(subj) {
+			o.spending--
+		}
 	}
+	o.mu.Unlock()
 }
diff --git a/server/disk_avail.go b/server/disk_avail.go
deleted file mode 100644
index 90157fe7..00000000
--- a/server/disk_avail.go
+++ /dev/null
@@ -1,37 +0,0 @@
-// Copyright 2020 The NATS Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// +build !windows
-
-package server
-
-import (
-	"os"
-	"syscall"
-)
-
-func diskAvailable(storeDir string) int64 {
-	var ba int64
-	if _, err := os.Stat(storeDir); os.IsNotExist(err) {
-		os.MkdirAll(storeDir, 0755)
-	}
-	var fs syscall.Statfs_t
-	if err := syscall.Statfs(storeDir, &fs); err == nil {
-		// Estimate 75% of available storage.
-		ba = int64(fs.Bavail * uint64(fs.Bsize) / 4 * 3)
-	} else {
-		// Used 1TB default as a guess if all else fails.
-		ba = JetStreamMaxStoreDefault
-	}
-	return ba
-}
diff --git a/server/disk_avail_windows.go b/server/disk_avail_windows.go
deleted file mode 100644
index 0e057087..00000000
--- a/server/disk_avail_windows.go
+++ /dev/null
@@ -1,21 +0,0 @@
-// Copyright 2020 The NATS Authors
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// +build windows
-
-package server
-
-// TODO(dlc) - See if there is a version of this for windows.
-func diskAvailable(storeDir string) int64 {
-	return JetStreamMaxStoreDefault
-}
diff --git a/server/errors.go b/server/errors.go
index ef085acf..89e50d42 100644
--- a/server/errors.go
+++ b/server/errors.go
@@ -169,15 +169,6 @@ var (
 
 	// ErrNoTransforms signals no subject transforms are available to map this subject.
 	ErrNoTransforms = errors.New("no matching transforms available")
-
-	// ErrJetStreamNotEnabled is returned when JetStream is not enabled.
-	ErrJetStreamNotEnabled = errors.New("jetstream not enabled")
-
-	// ErrJetStreamStreamNotFound is returned when a stream can not be found.
-	ErrJetStreamStreamNotFound = errors.New("stream not found")
-
-	// ErrJetStreamNotEnabledForAccount is returned JetStream is not enabled for this account.
-	ErrJetStreamNotEnabledForAccount = errors.New("jetstream not enabled for account")
 )
 
 // configErr is a configuration error.
diff --git a/server/filestore.go b/server/filestore.go
index f5e91100..afa4f25c 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -66,7 +66,7 @@ type FileConsumerInfo struct {
 type fileStore struct {
 	mu       sync.RWMutex
 	state    StreamState
-	scb      StorageUpdateHandler
+	scb      func(int64, int64, uint64)
 	ageChk   *time.Timer
 	syncTmr  *time.Timer
 	cfg      FileStreamInfo
@@ -74,6 +74,7 @@ type fileStore struct {
 	lmb      *msgBlock
 	blks     []*msgBlock
 	hh       hash.Hash64
+	fch      chan struct{}
 	qch      chan struct{}
 	cfs      []*consumerFileStore
 	closed   bool
@@ -103,9 +104,8 @@ type msgBlock struct {
 	cexp    time.Duration
 	ctmr    *time.Timer
 	loading bool
-	flusher bool
 	dmap    map[uint64]struct{}
-	fch     chan struct{}
+	dch     chan struct{}
 	qch     chan struct{}
 	lchk    [8]byte
 }
@@ -229,6 +229,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	fs := &fileStore{
 		fcfg: fcfg,
 		cfg:  FileStreamInfo{Created: created, StreamConfig: cfg},
+		fch:  make(chan struct{}, 1), // Make buffered on purpose.
 		qch:  make(chan struct{}),
 	}
 
@@ -262,6 +263,10 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		}
 	}
 
+	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)
 
 	return fs, nil
@@ -551,13 +556,13 @@ func (fs *fileStore) GetSeqFromTime(t time.Time) uint64 {
 // RegisterStorageUpdates registers a callback for updates to storage changes.
 // It will present number of messages and bytes as a signed integer and an
 // optional sequence number of the message if a single.
-func (fs *fileStore) RegisterStorageUpdates(cb StorageUpdateHandler) {
+func (fs *fileStore) RegisterStorageUpdates(cb func(int64, int64, uint64)) {
 	fs.mu.Lock()
 	fs.scb = cb
 	bsz := fs.state.Bytes
 	fs.mu.Unlock()
 	if cb != nil && bsz > 0 {
-		cb(0, int64(bsz), 0, _EMPTY_)
+		cb(0, int64(bsz), 0)
 	}
 }
 
@@ -571,25 +576,21 @@ func (fs *fileStore) hashKeyForBlock(index uint64) []byte {
 // Lock should be held.
 func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 	index := uint64(1)
+
 	if fs.lmb != nil {
 		index = fs.lmb.index + 1
+		fs.flushPendingWrites()
+		fs.closeLastMsgBlock(false)
 	}
 
 	mb := &msgBlock{index: index, cexp: fs.fcfg.CacheExpire}
-
-	// Now do local hash.
-	key := sha256.Sum256(fs.hashKeyForBlock(index))
-	hh, err := highwayhash.New64(key[:])
-	if err != nil {
-		return nil, fmt.Errorf("could not create hash: %v", err)
-	}
-	mb.hh = hh
+	fs.blks = append(fs.blks, mb)
+	fs.lmb = mb
 
 	mdir := path.Join(fs.fcfg.StoreDir, msgDir)
 	mb.mfn = path.Join(mdir, fmt.Sprintf(blkScan, mb.index))
-	mfd, err := os.OpenFile(mb.mfn, os.O_CREATE|os.O_RDWR, 0644)
+	mfd, err := os.OpenFile(mb.mfn, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0644)
 	if err != nil {
-		mb.dirtyCloseWithRemove(true)
 		return nil, fmt.Errorf("Error creating msg block file [%q]: %v", mb.mfn, err)
 	}
 	mb.mfd = mfd
@@ -597,21 +598,16 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 	mb.ifn = path.Join(mdir, fmt.Sprintf(indexScan, mb.index))
 	ifd, err := os.OpenFile(mb.ifn, os.O_CREATE|os.O_RDWR, 0644)
 	if err != nil {
-		mb.dirtyCloseWithRemove(true)
 		return nil, fmt.Errorf("Error creating msg index file [%q]: %v", mb.mfn, err)
 	}
 	mb.ifd = ifd
 
-	// Set cache time to creation time to start.
-	ts := time.Now().UnixNano()
-	mb.llts, mb.lrts, mb.lwts = ts, ts, ts
-
-	// We know we will need this so go ahead and spin up.
-	mb.spinUpFlushLoop()
-
-	// Add to our list of blocks and mark as last.
-	fs.blks = append(fs.blks, mb)
-	fs.lmb = mb
+	// Now do local hash.
+	key := sha256.Sum256(fs.hashKeyForBlock(index))
+	mb.hh, err = highwayhash.New64(key[:])
+	if err != nil {
+		return nil, fmt.Errorf("could not create hash: %v", err)
+	}
 
 	return mb, nil
 }
@@ -626,7 +622,7 @@ func (fs *fileStore) enableLastMsgBlockForWriting() error {
 	if mb.mfd != nil {
 		return nil
 	}
-	mfd, err := os.OpenFile(mb.mfn, os.O_CREATE|os.O_RDWR, 0644)
+	mfd, err := os.OpenFile(mb.mfn, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0644)
 	if err != nil {
 		return fmt.Errorf("error opening msg block file [%q]: %v", mb.mfn, err)
 	}
@@ -634,10 +630,9 @@ func (fs *fileStore) enableLastMsgBlockForWriting() error {
 	return nil
 }
 
-// Store stores a message. We hold the main filestore lock for any write operation.
+// Store stores a message.
 func (fs *fileStore) StoreMsg(subj string, hdr, msg []byte) (uint64, int64, error) {
 	fs.mu.Lock()
-
 	if fs.closed {
 		fs.mu.Unlock()
 		return 0, 0, ErrStoreClosed
@@ -662,6 +657,7 @@ func (fs *fileStore) StoreMsg(subj string, hdr, msg []byte) (uint64, int64, erro
 		fs.mu.Unlock()
 		return 0, 0, err
 	}
+	fs.kickFlusher()
 
 	if fs.state.Msgs == 0 {
 		fs.state.FirstSeq = seq
@@ -683,12 +679,12 @@ func (fs *fileStore) StoreMsg(subj string, hdr, msg []byte) (uint64, int64, erro
 	if fs.ageChk == nil && fs.cfg.MaxAge != 0 {
 		fs.startAgeChk()
 	}
+
 	cb := fs.scb
 	fs.mu.Unlock()
 
-	// Update the upper layers regarding storage used.
 	if cb != nil {
-		cb(1, int64(n), seq, subj)
+		cb(1, int64(n), seq)
 	}
 
 	return seq, ts, nil
@@ -703,41 +699,33 @@ func (mb *msgBlock) skipMsg(seq uint64, now time.Time) {
 	if mb == nil {
 		return
 	}
-	var needsRecord bool
-
-	mb.mu.Lock()
 	// 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 {
-		needsRecord = true
+		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.mu.Unlock()
-
-	if needsRecord {
-		mb.writeMsgRecord(emptyRecordLen, seq|ebit, _EMPTY_, nil, nil, now.UnixNano())
-	} else {
-		mb.kickFlusher()
-	}
+	mb.kickWriteFlusher()
 }
 
 // SkipMsg will use the next sequence number but not store anything.
 func (fs *fileStore) SkipMsg() uint64 {
-	fs.mu.Lock()
-	defer fs.mu.Unlock()
-
 	// Grab time.
 	now := time.Now().UTC()
+	fs.mu.Lock()
 	seq := fs.state.LastSeq + 1
 	fs.state.LastSeq = seq
 	fs.state.LastTime = now
@@ -750,7 +738,8 @@ func (fs *fileStore) SkipMsg() uint64 {
 		fs.state.FirstTime = now
 	}
 	fs.lmb.skipMsg(seq, now)
-
+	fs.kickFlusher()
+	fs.mu.Unlock()
 	return seq
 }
 
@@ -804,7 +793,6 @@ func (fs *fileStore) EraseMsg(seq uint64) (bool, error) {
 // Remove a message, optionally rewriting the mb file.
 func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 	fs.mu.Lock()
-
 	if fs.closed {
 		fs.mu.Unlock()
 		return false, ErrStoreClosed
@@ -824,26 +812,15 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 		return false, err
 	}
 
-	// If we have a callback grab the message since we need the subject.
-	// TODO(dlc) - This will cause whole buffer to be loaded which I was trying
-	// to avoid. Maybe use side cache for subjects or understand when we really need them.
-	// Meaning if the stream above is only a single subject no need to store, this is just
-	// for updating stream pending for consumers.
-	var sm *fileStoredMsg
-	if fs.scb != nil {
-		sm, _ = mb.fetchMsg(seq)
-	}
-
 	mb.mu.Lock()
 
-	// Check cache. This should be very rare.
+	// Check cache. This will be very rare, will hold lock on this one.
 	if mb.cache == nil || mb.cache.idx == nil {
 		mb.mu.Unlock()
-		fs.mu.Unlock()
 		if err := mb.loadMsgs(); err != nil {
+			fs.mu.Unlock()
 			return false, err
 		}
-		fs.mu.Lock()
 		mb.mu.Lock()
 	}
 
@@ -871,6 +848,16 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 	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
@@ -908,21 +895,26 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 		shouldWriteIndex = true
 	}
 
-	var qch, fch chan struct{}
+	var dch chan struct{}
+
 	if shouldWriteIndex {
-		qch = mb.qch
-		fch = mb.fch
+		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
+		}
 	}
-	cb := fs.scb
 	mb.mu.Unlock()
 
 	// Kick outside of lock.
-	if shouldWriteIndex {
-		if qch == nil {
-			mb.spinUpFlushLoop()
-		}
+	if shouldWriteIndex && dch != nil {
 		select {
-		case fch <- struct{}{}:
+		case dch <- struct{}{}:
 		default:
 		}
 	}
@@ -934,23 +926,13 @@ func (fs *fileStore) removeMsg(seq uint64, secure bool) (bool, error) {
 	}
 	fs.mu.Unlock()
 
-	// Storage updates.
-	if cb != nil {
-		subj := _EMPTY_
-		if sm != nil {
-			subj = sm.subj
-		}
-		delta := int64(msz)
-		cb(-1, -delta, seq, subj)
-	}
-
 	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 || slot >= len(mb.cache.idx) {
+	if mb.cache == nil || mb.cache.idx == nil {
 		return 0, 0, false, errPartialCache
 	}
 	bi := mb.cache.idx[slot]
@@ -977,100 +959,14 @@ func (fs *fileStore) isClosed() bool {
 	return closed
 }
 
-// Will spin up our flush loop.
-func (mb *msgBlock) spinUpFlushLoop() {
-	mb.mu.Lock()
-	// Are we already running?
-	if mb.flusher {
-		mb.mu.Unlock()
-		return
-	}
-	mb.flusher = true
-	mb.fch = make(chan struct{}, 1)
-	mb.qch = make(chan struct{})
-	fch, qch := mb.fch, mb.qch
-	mb.mu.Unlock()
-
-	go mb.flushLoop(fch, qch)
-}
-
-// Raw low level kicker for flush loops.
-func kickFlusher(fch chan struct{}) {
-	if fch != nil {
-		select {
-		case fch <- struct{}{}:
-		default:
-		}
-	}
-}
-
-// Kick flusher for this message block.
-func (mb *msgBlock) kickFlusher() {
-	mb.mu.RLock()
-	defer mb.mu.RUnlock()
-	kickFlusher(mb.fch)
-}
-
-func (mb *msgBlock) setInFlusher() {
-	mb.mu.Lock()
-	mb.flusher = true
-	mb.mu.Unlock()
-}
-
-func (mb *msgBlock) clearInFlusher() {
-	mb.mu.Lock()
-	mb.flusher = false
-	mb.mu.Unlock()
-}
-
-// flushLoop watches for messages, index info, or recently closed msg block updates.
-func (mb *msgBlock) flushLoop(fch, qch chan struct{}) {
-	mb.setInFlusher()
-	defer mb.clearInFlusher()
-
-	// Will use to test if we have meta data updates.
-	var firstSeq, lastSeq uint64
-	var dmapLen int
-
-	infoChanged := func() bool {
-		mb.mu.RLock()
-		defer mb.mu.RUnlock()
-		var changed bool
-		if firstSeq != mb.first.seq || lastSeq != mb.last.seq || dmapLen != len(mb.dmap) {
-			changed = true
-			firstSeq, lastSeq = mb.first.seq, mb.last.seq
-			dmapLen = len(mb.dmap)
-		}
-		return changed
-	}
-
+// Loop on requests to write out our index file. This is used when calling
+// remove for a message. Updates to the last.seq etc are handled by main
+// flush loop when storing messages.
+func (fs *fileStore) flushWriteIndexLoop(mb *msgBlock, dch, qch chan struct{}) {
 	for {
 		select {
-		case <-fch:
-			// If we have pending messages process them first.
-			if waiting := mb.pendingWriteSize(); waiting != 0 {
-				ts := 1 * time.Millisecond
-				var waited time.Duration
-
-				for waiting < coalesceMinimum {
-					time.Sleep(ts)
-					select {
-					case <-qch:
-						return
-					default:
-					}
-					newWaiting := mb.pendingWriteSize()
-					if waited = waited + ts; waited > maxFlushWait || newWaiting <= waiting {
-						break
-					}
-					waiting = newWaiting
-					ts *= 2
-				}
-				mb.flushPendingMsgs()
-			}
-			if infoChanged() {
-				mb.writeIndexInfo()
-			}
+		case <-dch:
+			mb.writeIndexInfo()
 		case <-qch:
 			return
 		}
@@ -1130,6 +1026,15 @@ func (mb *msgBlock) eraseMsg(seq uint64, ri, rl int) error {
 	return nil
 }
 
+func (mb *msgBlock) kickWriteFlusher() {
+	mb.mu.RLock()
+	select {
+	case mb.dch <- struct{}{}:
+	default:
+	}
+	mb.mu.RUnlock()
+}
+
 // Lock should be held.
 func (mb *msgBlock) isEmpty() bool {
 	return mb.first.seq > mb.last.seq
@@ -1156,7 +1061,7 @@ func (mb *msgBlock) selectNextFirst() {
 
 	// Need to get the timestamp.
 	// We will try the cache direct and fallback if needed.
-	sm, _ := mb.cacheLookupWithLock(seq)
+	sm, _ := mb.cacheLookupLocked(seq)
 	if sm == nil {
 		// Slow path, need to unlock.
 		mb.mu.Unlock()
@@ -1202,34 +1107,13 @@ func (mb *msgBlock) startCacheExpireTimer() {
 	mb.resetCacheExpireTimer(0)
 }
 
-// Used when we load in a message block.
-// Lock should be held.
-func (mb *msgBlock) clearCacheAndOffset() {
-	if mb.cache != nil {
-		mb.cache.off = 0
-		mb.cache.wp = 0
-	}
-	mb.clearCache()
-}
-
 // Lock should be held.
 func (mb *msgBlock) clearCache() {
 	if mb.ctmr != nil {
 		mb.ctmr.Stop()
 		mb.ctmr = nil
 	}
-	if mb.cache == nil {
-		return
-	}
-
-	if mb.cache.off == 0 {
-		mb.cache = nil
-	} else {
-		// Clear msgs and index.
-		mb.cache.buf = nil
-		mb.cache.idx = nil
-		mb.cache.wp = 0
-	}
+	mb.cache = nil
 }
 
 // Called to possibly expire a message block cache.
@@ -1246,7 +1130,7 @@ func (mb *msgBlock) expireCache() {
 	}
 
 	// Can't expire if we are flushing or still have pending.
-	if mb.cache.flush || (len(mb.cache.buf)-int(mb.cache.wp) > 0) {
+	if mb.cache.flush || (len(mb.cache.buf)-mb.cache.wp > 0) {
 		mb.resetCacheExpireTimer(mb.cexp)
 		return
 	}
@@ -1260,15 +1144,13 @@ func (mb *msgBlock) expireCache() {
 		bufts = mb.lwts
 	}
 
-	// Check for activity on the cache that would prevent us from expiring.
+	// Check for the underlying buffer first.
 	if tns-bufts <= int64(mb.cexp) {
 		mb.resetCacheExpireTimer(mb.cexp - time.Duration(tns-bufts))
 		return
 	}
 
 	// If we are here we will at least expire the core msg buffer.
-	// We need to capture offset in case we do a write next before a full load.
-	mb.cache.off += len(mb.cache.buf)
 	mb.cache.buf = nil
 	mb.cache.wp = 0
 
@@ -1376,22 +1258,9 @@ func checkMsgBlockFile(fp *os.File, hh hash.Hash) []uint64 {
 	return bad
 }
 
-// Lock should be held.
-func (fs *fileStore) checkAndFlushAllBlocks() {
-	for _, mb := range fs.blks {
-		if mb.pendingWriteSize() > 0 {
-			mb.flushPendingMsgsAndWait()
-			mb.writeIndexInfo()
-		}
-	}
-}
-
 // This will check all the checksums on messages and report back any sequence numbers with errors.
 func (fs *fileStore) checkMsgs() []uint64 {
-	fs.mu.Lock()
-	defer fs.mu.Unlock()
-
-	fs.checkAndFlushAllBlocks()
+	fs.flushPendingWritesUnlocked()
 
 	mdir := path.Join(fs.fcfg.StoreDir, msgDir)
 	fis, err := ioutil.ReadDir(mdir)
@@ -1418,10 +1287,70 @@ func (fs *fileStore) checkMsgs() []uint64 {
 	return bad
 }
 
+// This will kick out our flush routine if its waiting.
+func (fs *fileStore) kickFlusher() {
+	select {
+	case fs.fch <- struct{}{}:
+	default:
+	}
+}
+
+func (fs *fileStore) lastMsgBlock() *msgBlock {
+	fs.mu.RLock()
+	mb := fs.lmb
+	fs.mu.RUnlock()
+	return mb
+}
+
+// Looks at active write message block (lmb).
+func (fs *fileStore) pendingWriteSize() int {
+	return fs.lastMsgBlock().numWriteBytesPending()
+}
+
+// Main flush loop.
+func (fs *fileStore) flushLoop(fch, qch, sch chan struct{}) {
+	if sch != nil {
+		close(sch)
+	}
+	for {
+		select {
+		case <-fch:
+			waiting := fs.pendingWriteSize()
+			if waiting == 0 {
+				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
+			}
+			fs.flushPendingWritesUnlocked()
+		case <-qch:
+			return
+		}
+	}
+}
+
 // 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()
+	defer mb.mu.Unlock()
 
 	// Make sure we have a cache setup.
 	if mb.cache == nil {
@@ -1432,7 +1361,7 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
 	// Indexing
 	index := len(mb.cache.buf)
 	if mb.cache.off > 0 {
-		index += int(mb.cache.off)
+		index += mb.cache.off
 	}
 
 	// Formats
@@ -1495,59 +1424,53 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
 
 	// Accounting
 	mb.updateAccounting(seq, ts, rl)
-
-	fch := mb.fch
-	mb.mu.Unlock()
-
-	// Flush the kicker here.
-	kickFlusher(fch)
 }
 
 // How many bytes pending to be written for this message block.
-func (mb *msgBlock) pendingWriteSize() int {
+func (mb *msgBlock) numWriteBytesPending() int {
 	if mb == nil {
 		return 0
 	}
 	var pending int
 	mb.mu.RLock()
 	if mb.mfd != nil && mb.cache != nil {
-		pending = len(mb.cache.buf) - int(mb.cache.wp)
+		pending = len(mb.cache.buf) - mb.cache.wp
 	}
 	mb.mu.RUnlock()
+
 	return pending
 }
 
-// Lock should be held.
 func (mb *msgBlock) clearFlushing() {
+	mb.mu.Lock()
 	if mb.cache != nil {
 		mb.cache.flush = false
 	}
+	mb.mu.Unlock()
 }
 
-// Lock should be held.
-func (mb *msgBlock) setFlushing() {
-	if mb.cache != nil {
-		mb.cache.flush = true
-	}
-}
-
-// bytesPending returns the buffer to be used for writing to the underlying file.
+// 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.
-// Lock should be held.
-func (mb *msgBlock) bytesPending() ([]byte, error) {
+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
 }
 
@@ -1602,41 +1525,36 @@ func (fs *fileStore) writeMsgRecord(seq uint64, subj string, mhdr, msg []byte) (
 // Sync msg and index files as needed. This is called from a timer.
 func (fs *fileStore) syncBlocks() {
 	fs.mu.RLock()
-	if fs.closed {
-		fs.mu.RUnlock()
-		return
-	}
-	blks := append([]*msgBlock(nil), fs.blks...)
+	closed := fs.closed
+	blks := fs.blks
 	fs.mu.RUnlock()
 
+	if closed {
+		return
+	}
 	for _, mb := range blks {
 		mb.mu.RLock()
-		mfd := mb.mfd
-		ifd := mb.ifd
-		liwsz := mb.liwsz
-		mb.mu.RUnlock()
-
-		if mfd != nil {
-			mfd.Sync()
+		if mb.mfd != nil {
+			mb.mfd.Sync()
 		}
-		if ifd != nil {
-			ifd.Sync()
-			ifd.Truncate(liwsz)
+		if mb.ifd != nil {
+			mb.ifd.Sync()
+			mb.ifd.Truncate(mb.liwsz)
 		}
+		mb.mu.RUnlock()
 	}
 
-	fs.mu.RLock()
-	cfs := append([]*consumerFileStore(nil), fs.cfs...)
-	fs.mu.RUnlock()
+	var _cfs [256]*consumerFileStore
+
+	fs.mu.Lock()
+	cfs := append(_cfs[:0], fs.cfs...)
+	fs.syncTmr = time.AfterFunc(fs.fcfg.SyncInterval, fs.syncBlocks)
+	fs.mu.Unlock()
 
 	// Do consumers.
 	for _, o := range cfs {
 		o.syncStateFile()
 	}
-
-	fs.mu.Lock()
-	fs.syncTmr = time.AfterFunc(fs.fcfg.SyncInterval, fs.syncBlocks)
-	fs.mu.Unlock()
 }
 
 // Select the message block where this message should be found.
@@ -1647,6 +1565,7 @@ func (fs *fileStore) selectMsgBlock(seq uint64) *msgBlock {
 	if seq < fs.state.FirstSeq || seq > fs.state.LastSeq {
 		return nil
 	}
+
 	// 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 numbers of blks.
@@ -1662,14 +1581,21 @@ func (fs *fileStore) selectMsgBlock(seq uint64) *msgBlock {
 // Return nil if not in the set.
 func (fs *fileStore) selectMsgBlockForStart(minTime time.Time) *msgBlock {
 	fs.mu.RLock()
-	defer fs.mu.RUnlock()
+	blks := fs.blks
+	lmb := fs.lmb
+	fs.mu.RUnlock()
 
 	t := minTime.UnixNano()
-	for _, mb := range fs.blks {
+	for _, mb := range blks {
 		mb.mu.RLock()
 		found := t <= mb.last.ts
 		mb.mu.RUnlock()
+
 		if found {
+			// This detects if what we may be looking for is staged in the write buffer.
+			if mb == lmb {
+				fs.flushPendingWritesUnlocked()
+			}
 			return mb
 		}
 	}
@@ -1732,179 +1658,111 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 	mb.cache.buf = buf
 	mb.cache.idx = idx
 	mb.cache.fseq = fseq
-	mb.cache.wp += int(lbuf)
+	mb.cache.wp += len(buf)
 
 	return nil
 }
 
-func (mb *msgBlock) quitChan() chan struct{} {
-	mb.mu.RLock()
-	defer mb.mu.RUnlock()
-	return mb.qch
-}
-
-// When called directly, flushPending could be busy already and return errFlushRunning.
-// This function is called for in place flushing so we need to wait.
-func (mb *msgBlock) flushPendingMsgsAndWait() error {
-	var err error
-
-	// If we are in flush wait for that to clear.
-	for err = mb.flushPendingMsgs(); err == errFlushRunning; err = mb.flushPendingMsgs() {
-		qch := mb.quitChan()
-		select {
-		case <-qch:
-			return nil
-		case <-time.After(time.Millisecond):
-		}
-	}
-	return err
-}
-
-// flushPendingMsgs writes out any messages for this message block.
-func (mb *msgBlock) flushPendingMsgs() error {
-	// We will not hold the lock across I/O so we can add more messages
-	// in parallel but we allow only one flush to be running.
-	mb.mu.Lock()
-	if mb.cache == nil || mb.mfd == nil {
-		mb.mu.Unlock()
-		return nil
-	}
-
-	// bytesPending will return with errFlushRunning
-	// if we are already flushing this message block.
-	buf, err := mb.bytesPending()
-	// If we got an error back return here.
+// flushPendingWrites for this message block.
+func (mb *msgBlock) flushPendingWrites() error {
+	buf, err := mb.writeBytesPending()
 	if err != nil {
-		mb.mu.Unlock()
 		return err
 	}
+	defer mb.clearFlushing()
 
-	woff := int64(mb.cache.off + mb.cache.wp)
-	lob := len(buf)
-
-	// Only one can be flushing at a time.
-	mb.setFlushing()
-	mfd := mb.mfd
-	mb.mu.Unlock()
-
-	var tn int
-
+	var lbb, tn int
 	// Append new data to the message block file.
-	for lbb := lob; lbb > 0; lbb = len(buf) {
-		n, err := mfd.WriteAt(buf, woff)
+	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.removeIndexFile()
 			mb.mu.Lock()
-			mb.clearFlushing()
+			mb.removeIndex()
 			mb.mu.Unlock()
 			return err
 		}
-		woff += int64(n)
 		tn += n
 
 		// Success
-		if n == lbb {
+		if int(n) == lbb {
 			break
 		}
 		// Partial write..
 		buf = buf[n:]
 	}
 
-	// We did a successful write.
-	// Re-acquire lock to update.
 	mb.mu.Lock()
-	defer mb.mu.Unlock()
-	// Clear on exit.
-	defer mb.clearFlushing()
-
-	// Cache may be gone.
-	if mb.cache == nil || mb.mfd == nil {
+	if mb.cache == nil {
+		mb.mu.Unlock()
 		return nil
 	}
+	mb.cache.flush = false
 
-	// Check for additional writes while we were writing to the disk.
-	moreBytes := len(mb.cache.buf) - mb.cache.wp - lob
-
-	// Decide what we want to do with the buffer in hand. If we have load interest
-	// we will hold onto the whole thing, otherwise empty the buffer, possibly reusing it.
+	// 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 cap(buf) <= maxBufReuse {
+		if lbb <= maxBufReuse {
 			buf = buf[:0]
 		} else {
 			buf = nil
 		}
-		if moreBytes > 0 {
-			nbuf := mb.cache.buf[len(mb.cache.buf)-moreBytes:]
-			if moreBytes > (len(mb.cache.buf)/4*3) && cap(nbuf) <= maxBufReuse {
-				buf = nbuf
-			} else {
-				buf = append(buf, nbuf...)
-			}
-		}
-		// Update our cache offset.
-		mb.cache.off = int(woff)
 		// 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
 }
 
-//  Lock should be held.
 func (mb *msgBlock) clearLoading() {
+	mb.mu.Lock()
 	mb.loading = false
+	mb.mu.Unlock()
 }
 
 // Will load msgs from disk.
 func (mb *msgBlock) loadMsgs() error {
-	// We hold the lock here the whole time by design.
 	mb.mu.Lock()
-	defer mb.mu.Unlock()
-
 	// Check to see if we are loading already.
 	if mb.loading {
+		mb.mu.Unlock()
 		return nil
 	}
-
-	// Set loading status.
-	mb.loading = true
-	defer mb.clearLoading()
-
-checkCache:
-	// Check to see if we have a full cache.
-	if mb.cache != nil && len(mb.cache.idx) == int(mb.msgs) && mb.cache.off == 0 && len(mb.cache.buf) > 0 {
+	// 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
 	mb.llts = time.Now().UnixNano()
+	mb.mu.Unlock()
 
 	// FIXME(dlc) - We could be smarter here.
-	if mb.cache != nil && len(mb.cache.buf)-mb.cache.wp > 0 {
-		mb.mu.Unlock()
-		mb.flushPendingMsgsAndWait()
-		mb.writeIndexInfo()
-		mb.mu.Lock()
-		goto checkCache
+	if mb.numWriteBytesPending() > 0 {
+		mb.flushPendingWrites()
 	}
 
-	// Load in the whole block. We want to hold the mb lock here to avoid any changes to
-	// state.
+	// Load in the whole block.
 	buf, err := ioutil.ReadFile(mfn)
 	if err != nil {
 		return err
 	}
 
-	// Reset the cache since we just read everything in.
-	// Make sure this is cleared in case we had a partial when we started.
-	mb.clearCacheAndOffset()
+	mb.mu.Lock()
+
+	// Make sure this is cleared in case we had a partial.
+	mb.clearCache()
 
 	if err := mb.indexCacheBuf(buf); err != nil {
+		mb.mu.Unlock()
 		return err
 	}
 
@@ -1912,6 +1770,7 @@ checkCache:
 		mb.cloads++
 		mb.startCacheExpireTimer()
 	}
+	mb.mu.Unlock()
 
 	return nil
 }
@@ -1941,7 +1800,6 @@ 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.
@@ -1957,16 +1815,17 @@ func (mb *msgBlock) cacheLookup(seq uint64) (*fileStoredMsg, error) {
 	// vs read lock and promote. Also defer based on 1.14 performance.
 	mb.mu.Lock()
 	defer mb.mu.Unlock()
-	return mb.cacheLookupWithLock(seq)
+	return mb.cacheLookupLocked(seq)
 }
 
-// Will do a lookup from cache assuming lock is held.
-func (mb *msgBlock) cacheLookupWithLock(seq uint64) (*fileStoredMsg, error) {
-	if mb.cache == nil || len(mb.cache.idx) == 0 {
+// Will do a lookup from cache.
+// lock should be held.
+func (mb *msgBlock) cacheLookupLocked(seq uint64) (*fileStoredMsg, error) {
+	if mb.cache == nil {
 		return nil, errNoCache
 	}
 
-	if seq < mb.first.seq || seq < mb.cache.fseq || seq > mb.last.seq {
+	if seq < mb.first.seq || seq < mb.cache.fseq || (seq-mb.cache.fseq) >= uint64(len(mb.cache.idx)) {
 		return nil, ErrStoreMsgNotFound
 	}
 
@@ -1977,18 +1836,11 @@ func (mb *msgBlock) cacheLookupWithLock(seq uint64) (*fileStoredMsg, error) {
 		}
 	}
 
-	if mb.cache.off > 0 {
-		return nil, errPartialCache
-	}
-
-	bi, _, hashChecked, err := mb.slotInfo(int(seq - mb.cache.fseq))
-	if err != nil {
-		return nil, errPartialCache
-	}
-
 	// 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 !hashChecked {
@@ -1996,6 +1848,13 @@ func (mb *msgBlock) cacheLookupWithLock(seq uint64) (*fileStoredMsg, error) {
 		mb.cache.idx[seq-mb.cache.fseq] = (bi | hbit)
 	}
 
+	// 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:]
 
@@ -2021,8 +1880,6 @@ func (mb *msgBlock) cacheLookupWithLock(seq uint64) (*fileStoredMsg, error) {
 
 // Will return message for the given sequence number.
 func (fs *fileStore) msgForSeq(seq uint64) (*fileStoredMsg, error) {
-	// TODO(dlc) - Since Store, Remove, Skip all hold the write lock on fs this will
-	// be stalled. Need another lock if want to happen in parallel.
 	fs.mu.RLock()
 	if fs.closed {
 		fs.mu.RUnlock()
@@ -2032,16 +1889,16 @@ func (fs *fileStore) msgForSeq(seq uint64) (*fileStoredMsg, error) {
 	if seq == 0 {
 		seq = fs.state.FirstSeq
 	}
-	// Make sure to snapshot here.
-	lseq := fs.state.LastSeq
 	mb := fs.selectMsgBlock(seq)
 	fs.mu.RUnlock()
 
 	if mb == nil {
 		var err = ErrStoreEOF
-		if seq <= lseq {
+		fs.mu.RLock()
+		if seq <= fs.state.LastSeq {
 			err = ErrStoreMsgNotFound
 		}
+		fs.mu.RUnlock()
 		return nil, err
 	}
 	// TODO(dlc) - older design had a check to prefetch when we knew we were
@@ -2137,6 +1994,32 @@ 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()
+	defer fs.mu.Unlock()
+	return fs.flushPendingWrites()
+}
+
+// Lock should be held.
+func (fs *fileStore) flushPendingWrites() error {
+	mb := fs.lmb
+	if mb == nil || mb.mfd == nil {
+		return errNoPending
+	}
+	err := mb.flushPendingWrites()
+	if err != nil {
+		switch err {
+		case errNoCache, errNoPending:
+			err = nil
+		default:
+			return err
+		}
+	}
+	// Now index info
+	return mb.writeIndexInfo()
+}
+
 // Write index info to the appropriate file.
 func (mb *msgBlock) writeIndexInfo() error {
 	// HEADER: magic version msgs bytes fseq fts lseq lts checksum
@@ -2147,8 +2030,6 @@ func (mb *msgBlock) writeIndexInfo() error {
 	hdr[1] = version
 
 	mb.mu.Lock()
-	defer mb.mu.Unlock()
-
 	n := hdrLen
 	n += binary.PutUvarint(hdr[n:], mb.msgs)
 	n += binary.PutUvarint(hdr[n:], mb.bytes)
@@ -2167,19 +2048,21 @@ func (mb *msgBlock) writeIndexInfo() error {
 	if mb.ifd == nil {
 		ifd, err := os.OpenFile(mb.ifn, os.O_CREATE|os.O_RDWR, 0644)
 		if err != nil {
+			mb.mu.Unlock()
 			return err
 		}
 		mb.ifd = ifd
 	}
-
-	if n, err = mb.ifd.WriteAt(buf, 0); err == nil {
+	// TODO(dlc) - don't hold lock here.
+	n, err = mb.ifd.WriteAt(buf, 0)
+	if err == nil {
 		mb.liwsz = int64(n)
 	}
+	mb.mu.Unlock()
 
 	return err
 }
 
-// readIndexInfo will read in the index information for the message block.
 func (mb *msgBlock) readIndexInfo() error {
 	buf, err := ioutil.ReadFile(mb.ifn)
 	if err != nil {
@@ -2362,7 +2245,7 @@ func (fs *fileStore) Purge() uint64 {
 	fs.mu.Unlock()
 
 	if cb != nil {
-		cb(-int64(purged), -rbytes, 0, _EMPTY_)
+		cb(-int64(purged), -rbytes, 0)
 	}
 
 	return purged
@@ -2375,25 +2258,25 @@ func (fs *fileStore) numMsgBlocks() int {
 	return len(fs.blks)
 }
 
-// Will remove our index file.
-func (mb *msgBlock) removeIndexFile() {
-	mb.mu.RLock()
-	defer mb.mu.RUnlock()
+// Lock should be held.
+func (mb *msgBlock) removeIndex() {
 	if mb.ifd != nil {
 		mb.ifd.Close()
 		mb.ifd = nil
 	}
-	if mb.ifn != _EMPTY_ {
-		os.Remove(mb.ifn)
-	}
+	os.Remove(mb.ifn)
 }
 
 // Removes the msgBlock
 // Both locks should be held.
 func (fs *fileStore) removeMsgBlock(mb *msgBlock) {
-	mb.dirtyCloseWithRemove(true)
+	mb.removeIndex()
+	if mb.mfd != nil {
+		mb.mfd.Close()
+		mb.mfd = nil
+	}
+	os.Remove(mb.mfn)
 
-	// Remove from list.
 	for i, omb := range fs.blks {
 		if mb == omb {
 			fs.blks = append(fs.blks[:i], fs.blks[i+1:]...)
@@ -2402,25 +2285,24 @@ func (fs *fileStore) removeMsgBlock(mb *msgBlock) {
 	}
 	// Check for us being last message block
 	if mb == fs.lmb {
+		fs.lmb = nil
 		fs.newMsgBlockForWrite()
+		fs.lmb.first = mb.first
+		fs.lmb.last = mb.last
+		fs.lmb.writeIndexInfo()
 	}
+	go mb.close(true)
 }
 
 // Called by purge to simply get rid of the cache and close and fds.
-// Lock should not be held.
+// FIXME(dlc) - Merge with below func.
 func (mb *msgBlock) dirtyClose() {
-	mb.mu.Lock()
-	mb.dirtyCloseWithRemove(false)
-	mb.mu.Unlock()
-}
-
-// Should be called with lock held.
-func (mb *msgBlock) dirtyCloseWithRemove(remove bool) {
-	if mb == nil || mb.qch == nil {
+	if mb == nil {
 		return
 	}
+	mb.mu.Lock()
 	// Close cache
-	mb.clearCacheAndOffset()
+	mb.clearCache()
 	// Quit our loops.
 	if mb.qch != nil {
 		close(mb.qch)
@@ -2434,16 +2316,7 @@ func (mb *msgBlock) dirtyCloseWithRemove(remove bool) {
 		mb.ifd.Close()
 		mb.ifd = nil
 	}
-	if remove {
-		if mb.ifn != _EMPTY_ {
-			os.Remove(mb.ifn)
-			mb.ifn = _EMPTY_
-		}
-		if mb.mfn != _EMPTY_ {
-			os.Remove(mb.mfn)
-			mb.mfn = _EMPTY_
-		}
-	}
+	mb.mu.Unlock()
 }
 
 func (mb *msgBlock) close(sync bool) {
@@ -2451,14 +2324,8 @@ func (mb *msgBlock) close(sync bool) {
 		return
 	}
 	mb.mu.Lock()
-	defer mb.mu.Unlock()
-
-	if mb.qch == nil {
-		return
-	}
-
 	// Close cache
-	mb.clearCacheAndOffset()
+	mb.clearCache()
 	// Quit our loops.
 	if mb.qch != nil {
 		close(mb.qch)
@@ -2471,6 +2338,7 @@ func (mb *msgBlock) close(sync bool) {
 	}
 	mb.mfd = nil
 	mb.ifd = nil
+	mb.mu.Unlock()
 }
 
 func (fs *fileStore) closeAllMsgBlocks(sync bool) {
@@ -2479,6 +2347,10 @@ func (fs *fileStore) closeAllMsgBlocks(sync bool) {
 	}
 }
 
+func (fs *fileStore) closeLastMsgBlock(sync bool) {
+	fs.lmb.close(sync)
+}
+
 func (fs *fileStore) Delete() error {
 	if fs.isClosed() {
 		return ErrStoreClosed
@@ -2498,9 +2370,11 @@ func (fs *fileStore) Stop() error {
 		return ErrStoreClosed
 	}
 	fs.closed = true
+	close(fs.qch)
+
+	err := fs.flushPendingWrites()
 	fs.lmb = nil
 
-	fs.checkAndFlushAllBlocks()
 	fs.closeAllMsgBlocks(true)
 
 	if fs.syncTmr != nil {
@@ -2521,7 +2395,7 @@ func (fs *fileStore) Stop() error {
 		o.Stop()
 	}
 
-	return nil
+	return err
 }
 
 const errFile = "errors.txt"
@@ -2599,14 +2473,15 @@ func (fs *fileStore) streamSnapshot(w io.WriteCloser, blks []*msgBlock, includeC
 		return
 	}
 
+	// Now do messages themselves.
+	lmb := fs.lastMsgBlock()
+
 	// Can't use join path here, zip only recognizes relative paths with forward slashes.
 	msgPre := msgDir + "/"
 
-	// Now do messages themselves.
 	for _, mb := range blks {
-		if mb.pendingWriteSize() > 0 {
-			mb.flushPendingMsgsAndWait()
-			mb.writeIndexInfo()
+		if mb == lmb {
+			fs.flushPendingWritesUnlocked()
 		}
 		mb.mu.Lock()
 		buf, err := ioutil.ReadFile(mb.ifn)
@@ -2646,6 +2521,7 @@ 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 {
@@ -2659,12 +2535,10 @@ 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
 		}
-
-		// We can have the running state directly encoded now.
-		state, err := o.encodeState()
+		state, err := ioutil.ReadFile(path.Join(o.odir, consumerState))
 		if err != nil {
 			o.mu.Unlock()
-			writeErr(fmt.Sprintf("Could not encode consumer state for %q: %v", o.name, err))
+			writeErr(fmt.Sprintf("Could not read consumer state for %q: %v", o.name, err))
 			return
 		}
 		odirPre := consumerDir + "/" + o.name
@@ -2730,21 +2604,16 @@ 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
-	state   ConsumerState
-	fch     chan struct{}
-	qch     chan struct{}
-	flusher bool
-	writing bool
-	closed  bool
+	mu     sync.Mutex
+	fs     *fileStore
+	cfg    *FileConsumerInfo
+	name   string
+	odir   string
+	ifn    string
+	ifd    *os.File
+	lwsz   int64
+	hh     hash.Hash64
+	closed bool
 }
 
 func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerStore, error) {
@@ -2785,11 +2654,6 @@ 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()
@@ -2797,302 +2661,93 @@ func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerSt
 	return o, nil
 }
 
-// Kick flusher for this consumer.
-// Lock should be held.
-func (o *consumerFileStore) kickFlusher() {
-	if o.fch != nil {
-		select {
-		case o.fch <- struct{}{}:
-		default:
-		}
-	}
-}
-
-// 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
-		}
-	}
-}
-
-// 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
-	}
-
-	// 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
+func (o *consumerFileStore) Update(state *ConsumerState) error {
+	// Sanity checks.
+	if state.Delivered.ConsumerSeq < 1 || state.Delivered.StreamSeq < 1 {
+		return fmt.Errorf("bad delivered sequences")
 	}
-	if lr := len(o.state.Redelivered); lr > 0 {
-		maxSize += lr*(2*binary.MaxVarintLen64) + binary.MaxVarintLen64
+	if state.AckFloor.ConsumerSeq > state.Delivered.ConsumerSeq {
+		return fmt.Errorf("bad ack floor for consumer")
 	}
-	if maxSize == seqsHdrSize {
-		buf = hdr[:seqsHdrSize]
-	} else {
-		buf = make([]byte, maxSize)
+	if state.AckFloor.StreamSeq > state.Delivered.StreamSeq {
+		return fmt.Errorf("bad ack floor for stream")
 	}
 
-	now := time.Now()
+	var hdr [seqsHdrSize]byte
 
 	// Write header
-	buf[0] = magic
-	buf[1] = 2
+	hdr[0] = magic
+	hdr[1] = version
 
 	n := hdrLen
-	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 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)
+	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]
 
-		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)
+	// These are optional, but always write len. This is to avoid 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)
+			}
 		}
-	}
 
-	// We always write the redelivered len.
-	n += binary.PutUvarint(buf[n:], uint64(len(o.state.Redelivered)))
+		// 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 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)
+		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)
 		}
+		buf = append(buf, mbuf[:n]...)
 	}
-	return buf[:n], nil
-}
 
-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")
-	}
+	var lenbuf [binary.MaxVarintLen64]byte
+	n = binary.PutUvarint(lenbuf[0:], uint64(len(state.Redelivered)))
+	buf = append(buf, lenbuf[:n]...)
 
-	// 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)
-			}
-		}
-	}
+	// 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.
 	if len(state.Redelivered) > 0 {
-		redelivered = make(map[uint64]uint64, len(state.Redelivered))
-		for seq, dc := range state.Redelivered {
-			redelivered[seq] = dc
+		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)
 		}
+		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()
-	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()
+	err := o.ensureStateFileOpen()
 	if err == nil {
+		n, err = o.ifd.WriteAt(buf, 0)
 		o.lwsz = int64(n)
 	}
-	o.writing = false
 	o.mu.Unlock()
 
 	return err
@@ -3152,76 +2807,31 @@ 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 fmt.Errorf("corrupt state file")
 	}
 	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
 	}
 
-	version, err := checkConsumerHeader(buf)
-	if err != nil {
+	if err := checkHeader(buf); err != nil {
 		return nil, err
 	}
 
@@ -3257,50 +2867,36 @@ func (o *consumerFileStore) State() (*ConsumerState, error) {
 	readCount := readSeq
 
 	state = &ConsumerState{}
-	state.AckFloor.Consumer = readSeq()
-	state.AckFloor.Stream = readSeq()
-	state.Delivered.Consumer = readSeq()
-	state.Delivered.Stream = readSeq()
+	state.AckFloor.ConsumerSeq = readSeq()
+	state.AckFloor.StreamSeq = readSeq()
+	state.Delivered.ConsumerSeq = readSeq()
+	state.Delivered.StreamSeq = readSeq()
 
 	if bi == -1 {
-		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
+		return nil, fmt.Errorf("corrupt state file")
 	}
+	// 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]*Pending, numPending)
+		state.Pending = make(map[uint64]int64, numPending)
 		for i := 0; i < int(numPending); i++ {
-			sseq := readSeq()
-			var dseq uint64
-			if version == 2 {
-				dseq = readSeq()
-			}
+			seq := readSeq()
 			ts := readTimeStamp()
-			if sseq == 0 || ts == -1 {
-				return nil, errCorruptState
+			if seq == 0 || ts == -1 {
+				return nil, fmt.Errorf("corrupt state file")
 			}
 			// Adjust seq back.
-			sseq += state.AckFloor.Stream
-			if version == 2 {
-				dseq += state.AckFloor.Consumer
-			}
+			seq += state.AckFloor.StreamSeq
 			// Adjust the timestamp back.
-			if version == 1 {
-				ts = (ts + mints) * int64(time.Second)
-			} else {
-				ts = (mints - ts) * int64(time.Second)
-			}
+			ts = (ts + mints) * int64(time.Second)
 			// Store in pending.
-			state.Pending[sseq] = &Pending{dseq, ts}
+			state.Pending[seq] = ts
 		}
 	}
 
@@ -3313,7 +2909,7 @@ func (o *consumerFileStore) State() (*ConsumerState, error) {
 			seq := readSeq()
 			n := readCount()
 			if seq == 0 || n == 0 {
-				return nil, errCorruptState
+				return nil, fmt.Errorf("corrupt state file")
 			}
 			state.Redelivered[seq] = n
 		}
@@ -3328,82 +2924,43 @@ func (o *consumerFileStore) Stop() error {
 		o.mu.Unlock()
 		return 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.closed = true
+	if o.ifd != nil {
+		o.ifd.Sync()
+		o.ifd.Close()
+		o.ifd = nil
 	}
-
-	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 err
+	return nil
 }
 
 // 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 != _EMPTY_ {
+	if o.odir != "" {
 		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/fuzz.go b/server/fuzz.go
index e4aa7898..1f626983 100644
--- a/server/fuzz.go
+++ b/server/fuzz.go
@@ -15,6 +15,19 @@
 
 package server
 
+var defaultServerOptions = Options{
+	Host:                  "127.0.0.1",
+	Trace:                 true,
+	Debug:                 true,
+	DisableShortFirstPing: true,
+	NoLog:                 true,
+	NoSigs:                true,
+}
+
+func dummyClient() *client {
+	return &client{srv: New(&defaultServerOptions), msubs: -1, mpay: -1, mcl: MAX_CONTROL_LINE_SIZE}
+}
+
 func FuzzClient(data []byte) int {
 	if len(data) < 100 {
 		return -1
diff --git a/server/gateway.go b/server/gateway.go
index 5f7a2903..6b0aff05 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -49,10 +49,6 @@ const (
 	gwClusterOffset  = gwReplyPrefixLen
 	gwServerOffset   = gwClusterOffset + gwHashLen + 1
 	gwSubjectOffset  = gwServerOffset + gwHashLen + 1
-
-	// Gateway connections send PINGs regardless of traffic. The interval is
-	// either Options.PingInterval or this value, whichever is the smallest.
-	gwMaxPingInterval = 15 * time.Second
 )
 
 var (
@@ -60,7 +56,6 @@ var (
 	gatewayReconnectDelay        = defaultGatewayReconnectDelay
 	gatewayMaxRUnsubBeforeSwitch = defaultGatewayMaxRUnsubBeforeSwitch
 	gatewaySolicitDelay          = int64(defaultSolicitGatewaysDelay)
-	gatewayMaxPingInterval       = gwMaxPingInterval
 )
 
 // Warning when user configures gateway TLS insecure
diff --git a/server/jetstream.go b/server/jetstream.go
index 16959c85..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"
@@ -343,7 +344,7 @@ func (s *Server) JetStreamNumAccounts() int {
 func (s *Server) JetStreamReservedResources() (int64, int64, error) {
 	js := s.getJetStream()
 	if js == nil {
-		return -1, -1, ErrJetStreamNotEnabled
+		return -1, -1, fmt.Errorf("jetstream not enabled")
 	}
 	js.mu.RLock()
 	defer js.mu.RUnlock()
@@ -369,7 +370,7 @@ func (a *Account) EnableJetStream(limits *JetStreamAccountLimits) error {
 	// FIXME(dlc) - cluster mode
 	js := s.getJetStream()
 	if js == nil {
-		return ErrJetStreamNotEnabled
+		return fmt.Errorf("jetstream not enabled")
 	}
 	if s.SystemAccount() == a {
 		return fmt.Errorf("jetstream can not be enabled on the system account")
@@ -533,7 +534,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 the consumers.
+		// Now do Consumers.
 		odir := path.Join(sdir, fi.Name(), consumerDir)
 		ofis, _ := ioutil.ReadDir(odir)
 		if len(ofis) > 0 {
@@ -622,51 +623,23 @@ func (a *Account) Streams() []*Stream {
 	return msets
 }
 
-// LookupStream will lookup a stream by name.
 func (a *Account) LookupStream(name string) (*Stream, error) {
 	a.mu.RLock()
 	jsa := a.js
 	a.mu.RUnlock()
 
 	if jsa == nil {
-		return nil, ErrJetStreamNotEnabled
+		return nil, fmt.Errorf("jetstream not enabled")
 	}
 	jsa.mu.Lock()
 	mset, ok := jsa.streams[name]
 	jsa.mu.Unlock()
 	if !ok {
-		return nil, ErrJetStreamStreamNotFound
+		return nil, fmt.Errorf("stream not found")
 	}
 	return mset, nil
 }
 
-// LookupStreamBySubject will lookup a stream by a subject.
-// The subject needs to be an exact match or a subset. Supersets will not match.
-func (a *Account) LookupStreamBySubject(subj string) (*Stream, bool, error) {
-	a.mu.RLock()
-	jsa := a.js
-	a.mu.RUnlock()
-
-	if jsa == nil {
-		return nil, false, ErrJetStreamNotEnabled
-	}
-	jsa.mu.Lock()
-	defer jsa.mu.Unlock()
-
-	for _, mset := range jsa.streams {
-		for _, tsubj := range mset.config.Subjects {
-			if subj == tsubj {
-				return mset, len(mset.config.Subjects) > 1, nil
-			}
-			if subjectIsSubsetMatch(subj, tsubj) {
-				return mset, true, nil
-			}
-		}
-	}
-
-	return nil, false, ErrJetStreamStreamNotFound
-}
-
 // UpdateJetStreamLimits will update the account limits for a JetStream enabled account.
 func (a *Account) UpdateJetStreamLimits(limits *JetStreamAccountLimits) error {
 	a.mu.RLock()
@@ -679,10 +652,10 @@ func (a *Account) UpdateJetStreamLimits(limits *JetStreamAccountLimits) error {
 	}
 	js := s.getJetStream()
 	if js == nil {
-		return ErrJetStreamNotEnabled
+		return fmt.Errorf("jetstream not enabled")
 	}
 	if jsa == nil {
-		return ErrJetStreamNotEnabledForAccount
+		return fmt.Errorf("jetstream not enabled for account")
 	}
 
 	if limits == nil {
@@ -752,7 +725,7 @@ func (a *Account) DisableJetStream() error {
 
 	js := s.getJetStream()
 	if js == nil {
-		return ErrJetStreamNotEnabled
+		return fmt.Errorf("jetstream not enabled")
 	}
 
 	// Remove service imports.
@@ -766,7 +739,7 @@ func (a *Account) DisableJetStream() error {
 // Disable JetStream for the account.
 func (js *jetStream) disableJetStream(jsa *jsAccount) error {
 	if jsa == nil {
-		return ErrJetStreamNotEnabledForAccount
+		return fmt.Errorf("jetstream not enabled for account")
 	}
 
 	js.mu.Lock()
@@ -971,7 +944,17 @@ func (s *Server) dynJetStreamConfig(storeDir string, maxStore int64) *JetStreamC
 	if maxStore > 0 {
 		jsc.MaxStore = maxStore
 	} else {
-		jsc.MaxStore = diskAvailable(jsc.StoreDir)
+		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 {
@@ -994,7 +977,7 @@ func (a *Account) checkForJetStream() (*Server, *jsAccount, error) {
 	}
 
 	if jsa == nil {
-		return nil, nil, ErrJetStreamNotEnabledForAccount
+		return nil, nil, fmt.Errorf("jetstream not enabled for account")
 	}
 
 	return s, jsa, nil
@@ -1205,7 +1188,7 @@ func (t *StreamTemplate) Delete() error {
 	t.mu.Unlock()
 
 	if jsa == nil {
-		return ErrJetStreamNotEnabled
+		return fmt.Errorf("jetstream not enabled")
 	}
 
 	jsa.mu.Lock()
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index ba12c443..43329894 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -75,10 +75,6 @@ const (
 	JSApiStreamInfo  = "$JS.API.STREAM.INFO.*"
 	JSApiStreamInfoT = "$JS.API.STREAM.INFO.%s"
 
-	// JSApiStreamLookup is for obtaining a stream by a target subject.
-	// Will return JSON response.
-	JSApiStreamLookup = "$JS.API.STREAM.LOOKUP"
-
 	// JSApiStreamDelete is the endpoint to delete streams.
 	// Will return JSON response.
 	JSApiStreamDelete  = "$JS.API.STREAM.DELETE.*"
@@ -116,7 +112,7 @@ const (
 	JSApiConsumerCreate  = "$JS.API.CONSUMER.CREATE.*"
 	JSApiConsumerCreateT = "$JS.API.CONSUMER.CREATE.%s"
 
-	// JSApiDurableCreate is the endpoint to create durable consumers for streams.
+	// JSApiDurableCreate is the endpoint to create ephemeral consumers for streams.
 	// You need to include the stream and consumer name in the subject.
 	JSApiDurableCreate  = "$JS.API.CONSUMER.DURABLE.CREATE.*.*"
 	JSApiDurableCreateT = "$JS.API.CONSUMER.DURABLE.CREATE.%s.%s"
@@ -266,20 +262,6 @@ type JSApiStreamInfoResponse struct {
 
 const JSApiStreamInfoResponseType = "io.nats.jetstream.api.v1.stream_info_response"
 
-type JSApiStreamLookupRequest struct {
-	// Subject finds any stream that matches this subject including those where wildcards intercepts
-	Subject string `json:"subject"`
-}
-
-// JSApiStreamLookupResponse.
-type JSApiStreamLookupResponse struct {
-	ApiResponse
-	Stream   string `json:"stream"`
-	Filtered bool   `json:"is_filtered"`
-}
-
-const JSApiStreamLookupResponseType = "io.nats.jetstream.api.v1.stream_lookup_response"
-
 // Maximum entries we will return for streams or consumers lists.
 // TODO(dlc) - with header or request support could request chunked response.
 const JSApiNamesLimit = 1024
@@ -498,7 +480,6 @@ var allJsExports = []string{
 	JSApiStreams,
 	JSApiStreamList,
 	JSApiStreamInfo,
-	JSApiStreamLookup,
 	JSApiStreamDelete,
 	JSApiStreamPurge,
 	JSApiStreamSnapshot,
@@ -528,7 +509,6 @@ func (s *Server) setJetStreamExportSubs() error {
 		{JSApiStreams, s.jsStreamNamesRequest},
 		{JSApiStreamList, s.jsStreamListRequest},
 		{JSApiStreamInfo, s.jsStreamInfoRequest},
-		{JSApiStreamLookup, s.jsStreamLookupRequest},
 		{JSApiStreamDelete, s.jsStreamDeleteRequest},
 		{JSApiStreamPurge, s.jsStreamPurgeRequest},
 		{JSApiStreamSnapshot, s.jsStreamSnapshotRequest},
@@ -973,57 +953,6 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, subject, repl
 	s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(resp))
 }
 
-// Request to lookup a stream by target subject.
-func (s *Server) jsStreamLookupRequest(sub *subscription, c *client, subject, reply string, msg []byte) {
-	if c == nil || c.acc == nil {
-		return
-	}
-
-	var resp = JSApiStreamLookupResponse{ApiResponse: ApiResponse{Type: JSApiStreamLookupResponseType}}
-	if !c.acc.JetStreamEnabled() {
-		resp.Error = jsNotEnabledErr
-		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
-		return
-	}
-
-	if isEmptyRequest(msg) {
-		resp.Error = &ApiError{Code: 400, Description: "subject required"}
-		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
-		return
-	}
-
-	subj := string(msg)
-
-	if bytes.HasPrefix(bytes.TrimSpace(msg), []byte("{")) {
-		var req JSApiStreamLookupRequest
-		err := json.Unmarshal(msg, &req)
-		// happy path we set it, otherwise we try whatever was in the
-		// msg as if its a subject this is because '{' is a valid subject
-		if err == nil {
-			subj = req.Subject
-		}
-	}
-
-	if !IsValidSubject(subj) {
-		resp.Error = &ApiError{Code: 400, Description: "subject argument is not a valid subject or request"}
-		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
-		return
-	}
-
-	// Lookup our stream.
-	mset, filtered, err := c.acc.LookupStreamBySubject(subj)
-	if err != nil {
-		resp.Error = jsNotFoundError(err)
-		s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(&resp))
-		return
-	}
-
-	resp.Stream = mset.Name()
-	resp.Filtered = filtered
-
-	s.sendAPIResponse(c, subject, reply, string(msg), s.jsonResponse(resp))
-}
-
 func isEmptyRequest(req []byte) bool {
 	if len(req) == 0 {
 		return true
diff --git a/server/memstore.go b/server/memstore.go
index 971ed45b..bcc845c1 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -27,7 +27,7 @@ type memStore struct {
 	cfg       StreamConfig
 	state     StreamState
 	msgs      map[uint64]*storedMsg
-	scb       StorageUpdateHandler
+	scb       func(int64, int64, uint64)
 	ageChk    *time.Timer
 	consumers int
 }
@@ -134,7 +134,7 @@ func (ms *memStore) StoreMsg(subj string, hdr, msg []byte) (uint64, int64, error
 	ms.mu.Unlock()
 
 	if cb != nil {
-		cb(1, stopBytes-startBytes, seq, subj)
+		cb(1, stopBytes-startBytes, seq)
 	}
 
 	return seq, ts, nil
@@ -161,7 +161,7 @@ func (ms *memStore) SkipMsg() uint64 {
 // RegisterStorageUpdates registers a callback for updates to storage changes.
 // It will present number of messages and bytes as a signed integer and an
 // optional sequence number of the message if a single.
-func (ms *memStore) RegisterStorageUpdates(cb StorageUpdateHandler) {
+func (ms *memStore) RegisterStorageUpdates(cb func(int64, int64, uint64)) {
 	ms.mu.Lock()
 	ms.scb = cb
 	ms.mu.Unlock()
@@ -267,7 +267,7 @@ func (ms *memStore) Purge() uint64 {
 	ms.mu.Unlock()
 
 	if cb != nil {
-		cb(-int64(purged), -bytes, 0, _EMPTY_)
+		cb(-int64(purged), -bytes, 0)
 	}
 
 	return purged
@@ -351,6 +351,13 @@ func (ms *memStore) removeMsg(seq uint64, secure bool) bool {
 
 	ss = memStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 
+	if ms.scb != nil {
+		delta := int64(ss)
+		ms.mu.Unlock()
+		ms.scb(-1, -delta, seq)
+		ms.mu.Lock()
+	}
+
 	delete(ms.msgs, seq)
 	ms.state.Msgs--
 	ms.state.Bytes -= ss
@@ -368,13 +375,6 @@ func (ms *memStore) removeMsg(seq uint64, secure bool) bool {
 		sm.seq, sm.ts = 0, 0
 	}
 
-	if ms.scb != nil {
-		ms.mu.Unlock()
-		delta := int64(ss)
-		ms.scb(-1, -delta, seq, sm.subj)
-		ms.mu.Lock()
-	}
-
 	return ok
 }
 
@@ -435,12 +435,9 @@ func (ms *memStore) Snapshot(_ time.Duration, _, _ bool) (*SnapshotResult, error
 }
 
 // No-ops.
-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) Update(_ *ConsumerState) error {
+	return nil
+}
 func (os *consumerMemStore) Stop() error {
 	os.ms.decConsumers()
 	return nil
diff --git a/server/server.go b/server/server.go
index 59abcb83..ef3a0eb0 100644
--- a/server/server.go
+++ b/server/server.go
@@ -3255,9 +3255,6 @@ func (s *Server) setFirstPingTimer(c *client) {
 			if d > firstPingInterval {
 				d = firstPingInterval
 			}
-			if c.kind == GATEWAY {
-				d = adjustPingIntervalForGateway(d)
-			}
 		} else if d > firstClientPingInterval {
 			d = firstClientPingInterval
 		}
diff --git a/server/store.go b/server/store.go
index bfa41bc4..06f7def6 100644
--- a/server/store.go
+++ b/server/store.go
@@ -52,14 +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")
 )
 
-// Used to call back into the upper layers to report on changes in storage resources.
-// For the cases where its a single message we will also supply sequence number and subject.
-type StorageUpdateHandler func(msgs, bytes int64, seq uint64, subj string)
-
 type StreamStore interface {
 	StoreMsg(subj string, hdr, msg []byte) (uint64, int64, error)
 	SkipMsg() uint64
@@ -69,7 +63,7 @@ type StreamStore interface {
 	Purge() uint64
 	GetSeqFromTime(t time.Time) uint64
 	State() StreamState
-	RegisterStorageUpdates(StorageUpdateHandler)
+	RegisterStorageUpdates(func(int64, int64, uint64))
 	UpdateConfig(cfg *StreamConfig) error
 	Delete() error
 	Stop() error
@@ -121,18 +115,16 @@ type SnapshotResult struct {
 
 // ConsumerStore stores state on consumers for streams.
 type ConsumerStore interface {
-	UpdateDelivered(dseq, sseq, dc uint64, ts int64) error
-	UpdateAcks(dseq, sseq uint64) error
-	Update(*ConsumerState) error
 	State() (*ConsumerState, error)
+	Update(*ConsumerState) error
 	Stop() error
 	Delete() error
 }
 
 // SequencePair has both the consumer and the stream sequence. They point to same message.
 type SequencePair struct {
-	Consumer uint64 `json:"consumer_seq"`
-	Stream   uint64 `json:"stream_seq"`
+	ConsumerSeq uint64 `json:"consumer_seq"`
+	StreamSeq   uint64 `json:"stream_seq"`
 }
 
 // ConsumerState represents a stored state for a consumer.
@@ -144,18 +136,11 @@ 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]*Pending `json:"pending"`
+	Pending map[uint64]int64 `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 770fb012..ef85d831 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -71,6 +71,8 @@ type StreamInfo struct {
 // for a Stream we will direct link from the client to this Stream structure.
 type Stream struct {
 	mu        sync.RWMutex
+	sg        *sync.Cond
+	sgw       int
 	jsa       *jsAccount
 	client    *client
 	sid       int
@@ -155,6 +157,7 @@ func (a *Account) AddStreamWithStore(config *StreamConfig, fsConfig *FileStoreCo
 	// Setup the internal client.
 	c := s.createInternalJetStreamClient()
 	mset := &Stream{jsa: jsa, config: cfg, client: c, consumers: make(map[string]*Consumer)}
+	mset.sg = sync.NewCond(&mset.mu)
 
 	jsa.streams[cfg.Name] = mset
 	storeDir := path.Join(jsa.storeDir, streamsDir, cfg.Name)
@@ -491,7 +494,7 @@ func (mset *Stream) Delete() error {
 	jsa := mset.jsa
 	mset.mu.Unlock()
 	if jsa == nil {
-		return ErrJetStreamNotEnabledForAccount
+		return fmt.Errorf("jetstream not enabled for account")
 	}
 	jsa.mu.Lock()
 	delete(jsa.streams, mset.config.Name)
@@ -759,17 +762,23 @@ func (mset *Stream) setupStore(fsCfg *FileStoreConfig) error {
 // jetstream account. We do local processing for stream pending for consumers, but only
 // for removals.
 // Lock should not ne held.
-func (mset *Stream) storeUpdates(md, bd int64, seq uint64, subj string) {
+func (mset *Stream) storeUpdates(md, bd int64, seq uint64) {
+	var _obs [4]*Consumer
+	obs := _obs[:0]
+
 	// If we have a single negative update then we will process our consumers for stream pending.
 	// Purge and Store handled separately inside individual calls.
 	if md == -1 {
-		mset.mu.RLock()
+		mset.mu.Lock()
 		for _, o := range mset.consumers {
-			o.decStreamPending(seq, subj)
+			obs = append(obs, o)
 		}
-		mset.mu.RUnlock()
+		mset.mu.Unlock()
 	}
 
+	for _, o := range obs {
+		o.decStreamPending(seq)
+	}
 	if mset.jsa != nil {
 		mset.jsa.updateUsage(mset.config.Storage, bd)
 	}
@@ -991,6 +1000,7 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 	}
 
 	if err == nil && seq > 0 && numConsumers > 0 {
+		var needSignal bool
 		var _obs [4]*Consumer
 		obs := _obs[:0]
 
@@ -1003,10 +1013,23 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 		for _, o := range obs {
 			o.incStreamPending(seq, subject)
 			if !o.deliverCurrentMsg(subject, hdr, msg, seq, ts) {
-				o.signalNewMessages()
+				needSignal = true
 			}
 		}
+
+		if needSignal {
+			mset.signalConsumers()
+		}
+	}
+}
+
+// Will signal all waiting consumers.
+func (mset *Stream) signalConsumers() {
+	mset.mu.Lock()
+	if mset.sgw > 0 {
+		mset.sg.Broadcast()
 	}
+	mset.mu.Unlock()
 }
 
 // Internal message for use by jetstream subsystem.
@@ -1133,7 +1156,9 @@ func (mset *Stream) stop(delete bool) error {
 		o.stop(delete, false, delete)
 	}
 
+	// Make sure we release all consumers here at once.
 	mset.mu.Lock()
+	mset.sg.Broadcast()
 
 	// Send stream delete advisory after the consumers.
 	if delete {
@@ -1245,6 +1270,22 @@ func (mset *Stream) State() StreamState {
 	return mset.store.State()
 }
 
+// waitForMsgs will have the stream wait for the arrival of new messages.
+func (mset *Stream) waitForMsgs() {
+	mset.mu.Lock()
+
+	if mset.client == nil {
+		mset.mu.Unlock()
+		return
+	}
+
+	mset.sgw++
+	mset.sg.Wait()
+	mset.sgw--
+
+	mset.mu.Unlock()
+}
+
 // Determines if the new proposed partition is unique amongst all observables.
 // Lock should be held.
 func (mset *Stream) partitionUnique(partition string) bool {
@@ -1294,8 +1335,17 @@ 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)
 }
 
diff --git a/server/sublist.go b/server/sublist.go
index 37ea486e..7ae9a3b8 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -138,10 +138,9 @@ func (s *Sublist) CacheEnabled() bool {
 // RegisterNotification will register for notifications when interest for the given
 // subject changes. The subject must be a literal publish type subject. The
 // notification is true for when the first interest for a subject is inserted,
-// and false when all interest in the subject is removed. Note that this interest
-// needs to be exact and that wildcards will not trigger the notifications. The sublist
-// will not block when trying to send the notification. Its up to the caller to make
-// sure the channel send will not block.
+// and false when all interest in the subject is removed. The sublist will not
+// block when trying to send the notification. Its up to the caller to make sure
+// the channel send will not block.
 func (s *Sublist) RegisterNotification(subject string, notify chan<- bool) error {
 	if subjectHasWildcard(subject) {
 		return ErrInvalidSubject
@@ -150,24 +149,8 @@ func (s *Sublist) RegisterNotification(subject string, notify chan<- bool) error
 		return ErrNilChan
 	}
 
-	var hasInterest bool
 	r := s.Match(subject)
-
-	if len(r.psubs)+len(r.qsubs) > 0 {
-		for _, sub := range r.psubs {
-			if string(sub.subject) == subject {
-				hasInterest = true
-				break
-			}
-		}
-		for _, qsub := range r.qsubs {
-			qs := qsub[0]
-			if string(qs.subject) == subject {
-				hasInterest = true
-				break
-			}
-		}
-	}
+	hasInterest := len(r.psubs)+len(r.qsubs) > 0
 
 	s.Lock()
 	if s.notify == nil {
@@ -260,48 +243,52 @@ func (s *Sublist) addNotify(m map[string][]chan<- bool, subject string, notify c
 
 // chkForInsertNotification will check to see if we need to notify on this subject.
 // Write lock should be held.
-func (s *Sublist) chkForInsertNotification(subject string) {
-	// All notify subjects are also literal so just do a hash lookup here.
-	if chs := s.notify.insert[subject]; len(chs) > 0 {
-		for _, ch := range chs {
-			sendNotification(ch, true)
+func (s *Sublist) chkForInsertNotification(subject string, isLiteral bool) {
+	// If we are a literal, all notify subjects are also literal so just do a
+	// hash lookup here.
+	if isLiteral {
+		chs := s.notify.insert[subject]
+		if len(chs) > 0 {
+			for _, ch := range chs {
+				sendNotification(ch, true)
+			}
+			// Move from the insert map to the remove map.
+			s.notify.remove[subject] = append(s.notify.remove[subject], chs...)
+			delete(s.notify.insert, subject)
+		}
+		return
+	}
+
+	// We are not a literal, so we may match any subject that we want.
+	// Note we could be smarter here and try to make the list smaller, but probably not worth it TBH.
+	for target, chs := range s.notify.insert {
+		r := s.matchNoLock(target)
+		if len(r.psubs)+len(r.qsubs) > 0 {
+			for _, ch := range chs {
+				sendNotification(ch, true)
+			}
+			// Move from the insert map to the remove map.
+			s.notify.remove[target] = append(s.notify.remove[target], chs...)
+			delete(s.notify.insert, target)
+			break
 		}
-		// Move from the insert map to the remove map.
-		s.notify.remove[subject] = append(s.notify.remove[subject], chs...)
-		delete(s.notify.insert, subject)
 	}
 }
 
 // chkForRemoveNotification will check to see if we need to notify on this subject.
 // Write lock should be held.
-func (s *Sublist) chkForRemoveNotification(subject string) {
-	if chs := s.notify.remove[subject]; len(chs) > 0 {
+func (s *Sublist) chkForRemoveNotification(subject string, isLiteral bool) {
+	for target, chs := range s.notify.remove {
 		// We need to always check that we have no interest anymore.
-		var hasInterest bool
-		r := s.matchNoLock(subject)
-
-		if len(r.psubs)+len(r.qsubs) > 0 {
-			for _, sub := range r.psubs {
-				if string(sub.subject) == subject {
-					hasInterest = true
-					break
-				}
-			}
-			for _, qsub := range r.qsubs {
-				qs := qsub[0]
-				if string(qs.subject) == subject {
-					hasInterest = true
-					break
-				}
-			}
-		}
-		if !hasInterest {
+		r := s.matchNoLock(target)
+		if len(r.psubs)+len(r.qsubs) == 0 {
 			for _, ch := range chs {
 				sendNotification(ch, false)
 			}
 			// Move from the remove map to the insert map.
-			s.notify.insert[subject] = append(s.notify.insert[subject], chs...)
-			delete(s.notify.remove, subject)
+			s.notify.insert[target] = append(s.notify.insert[target], chs...)
+			delete(s.notify.remove, target)
+			break
 		}
 	}
 }
@@ -400,8 +387,8 @@ func (s *Sublist) Insert(sub *subscription) error {
 	s.addToCache(subject, sub)
 	atomic.AddUint64(&s.genid, 1)
 
-	if s.notify != nil && isnew && !haswc && len(s.notify.insert) > 0 {
-		s.chkForInsertNotification(subject)
+	if s.notify != nil && isnew && len(s.notify.insert) > 0 {
+		s.chkForInsertNotification(subject, !haswc)
 	}
 	s.Unlock()
 
@@ -741,8 +728,8 @@ func (s *Sublist) remove(sub *subscription, shouldLock bool, doCacheUpdates bool
 		atomic.AddUint64(&s.genid, 1)
 	}
 
-	if s.notify != nil && last && !haswc && len(s.notify.remove) > 0 {
-		s.chkForRemoveNotification(subject)
+	if s.notify != nil && last && len(s.notify.remove) > 0 {
+		s.chkForRemoveNotification(subject, !haswc)
 	}
 
 	return nil


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index ff5570b5..97e05772 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -348,6 +348,7 @@ 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?" {
@@ -852,7 +853,6 @@ func TestFileStoreAgeLimitRecovery(t *testing.T) {
 
 	// Make sure they expire.
 	checkFor(t, time.Second, 2*maxAge, func() error {
-		t.Helper()
 		state = fs.State()
 		if state.Msgs != 0 {
 			return fmt.Errorf("Expected no msgs, got %d", state.Msgs)
@@ -955,7 +955,9 @@ func TestFileStoreEraseMsg(t *testing.T) {
 	if sm2, _ := fs.msgForSeq(1); sm2 != nil {
 		t.Fatalf("Expected msg to be erased")
 	}
-	fs.checkAndFlushAllBlocks()
+	if err := fs.flushPendingWritesUnlocked(); err != nil {
+		t.Fatalf("Unexpected error flushing: %v", err)
+	}
 
 	// Now look on disk as well.
 	rl := fileStoreMsgSize(subj, nil, msg)
@@ -1297,7 +1299,7 @@ func TestFileStoreReadCache(t *testing.T) {
 		fs.StoreMsg(subj, nil, msg)
 	}
 
-	// Wait for cache to go to zero.
+	// 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)))
@@ -1331,80 +1333,6 @@ func TestFileStoreReadCache(t *testing.T) {
 	}
 }
 
-func TestFileStorePartialCacheExpiration(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, CacheExpire: cexp}, StreamConfig{Name: "zzz", Storage: FileStorage})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	defer fs.Stop()
-
-	fs.StoreMsg("foo", nil, []byte("msg1"))
-
-	// Should expire and be removed.
-	time.Sleep(2 * cexp)
-	fs.StoreMsg("bar", nil, []byte("msg2"))
-
-	// Again wait for cache to expire.
-	time.Sleep(2 * cexp)
-	if _, _, _, _, err := fs.LoadMsg(1); err != nil {
-		t.Fatalf("Error loading message 1: %v", err)
-	}
-}
-
-func TestFileStorePartialIndexes(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, CacheExpire: cexp}, StreamConfig{Name: "zzz", Storage: FileStorage})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	defer fs.Stop()
-
-	toSend := 5
-	for i := 0; i < toSend; i++ {
-		fs.StoreMsg("foo", nil, []byte("ok-1"))
-	}
-
-	// Now wait til the cache expires, including the index.
-	fs.mu.Lock()
-	mb := fs.blks[0]
-	fs.mu.Unlock()
-
-	// Force idx to expire by resetting last remove ts.
-	mb.mu.Lock()
-	mb.lrts = mb.lrts - int64(defaultCacheIdxExpiration*2)
-	mb.mu.Unlock()
-
-	checkFor(t, time.Second, 10*time.Millisecond, func() error {
-		mb.mu.Lock()
-		defer mb.mu.Unlock()
-		if mb.cache == nil || len(mb.cache.idx) == 0 {
-			return nil
-		}
-		return fmt.Errorf("Index not empty")
-	})
-
-	// Create a partial cache by adding more msgs.
-	for i := 0; i < toSend; i++ {
-		fs.StoreMsg("foo", nil, []byte("ok-2"))
-	}
-	// If we now load in a message in second half if we do not
-	// detect idx is a partial correctly this will panic.
-	if _, _, _, _, err := fs.LoadMsg(8); err != nil {
-		t.Fatalf("Error loading %d: %v", 1, err)
-	}
-}
-
 func TestFileStoreSnapshot(t *testing.T) {
 	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
 	os.MkdirAll(storeDir, 0755)
@@ -1436,16 +1364,16 @@ func TestFileStoreSnapshot(t *testing.T) {
 		t.Fatalf("Unexepected error: %v", err)
 	}
 	state := &ConsumerState{}
-	state.Delivered.Consumer = 100
-	state.Delivered.Stream = 100
-	state.AckFloor.Consumer = 22
-	state.AckFloor.Stream = 22
+	state.Delivered.ConsumerSeq = 100
+	state.Delivered.StreamSeq = 100
+	state.AckFloor.ConsumerSeq = 22
+	state.AckFloor.StreamSeq = 22
 
 	if err := o1.Update(state); err != nil {
 		t.Fatalf("Unexepected error updating state: %v", err)
 	}
-	state.AckFloor.Consumer = 33
-	state.AckFloor.Stream = 33
+	state.AckFloor.ConsumerSeq = 33
+	state.AckFloor.StreamSeq = 33
 
 	if err := o2.Update(state); err != nil {
 		t.Fatalf("Unexepected error updating state: %v", err)
@@ -1628,47 +1556,47 @@ func TestFileStoreConsumer(t *testing.T) {
 		}
 	}
 
-	state.Delivered.Consumer = 1
-	state.Delivered.Stream = 22
+	state.Delivered.ConsumerSeq = 1
+	state.Delivered.StreamSeq = 22
 	updateAndCheck()
 
-	state.Delivered.Consumer = 100
-	state.Delivered.Stream = 122
-	state.AckFloor.Consumer = 50
-	state.AckFloor.Stream = 123
+	state.Delivered.ConsumerSeq = 100
+	state.Delivered.StreamSeq = 122
+	state.AckFloor.ConsumerSeq = 50
+	state.AckFloor.StreamSeq = 123
 	// This should fail, bad state.
 	shouldFail()
 	// So should this.
-	state.AckFloor.Consumer = 200
-	state.AckFloor.Stream = 100
+	state.AckFloor.ConsumerSeq = 200
+	state.AckFloor.StreamSeq = 100
 	shouldFail()
 
 	// Should succeed
-	state.AckFloor.Consumer = 50
-	state.AckFloor.Stream = 72
+	state.AckFloor.ConsumerSeq = 50
+	state.AckFloor.StreamSeq = 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]*Pending{70: &Pending{70, tn}}
+	state.Pending = map[uint64]int64{70: tn}
 	shouldFail()
-	state.Pending = map[uint64]*Pending{140: &Pending{140, tn}}
+	state.Pending = map[uint64]int64{140: tn}
 	shouldFail()
-	state.Pending = map[uint64]*Pending{72: &Pending{72, tn}} // exact on floor should fail
+	state.Pending = map[uint64]int64{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() *Pending {
+	nt := func() int64 {
 		ago = ago.Add(time.Second)
-		return &Pending{0, ago.UnixNano()}
+		return ago.UnixNano()
 	}
 	// Should succeed.
-	state.Pending = map[uint64]*Pending{75: nt(), 80: nt(), 83: nt(), 90: nt(), 111: nt()}
+	state.Pending = map[uint64]int64{75: nt(), 80: nt(), 83: nt(), 90: nt(), 111: nt()}
 	updateAndCheck()
 
 	// Now do redlivery, but first with no pending.
@@ -1677,16 +1605,16 @@ func TestFileStoreConsumer(t *testing.T) {
 	updateAndCheck()
 
 	// All together.
-	state.Pending = map[uint64]*Pending{75: nt(), 80: nt(), 83: nt(), 90: nt(), 111: nt()}
+	state.Pending = map[uint64]int64{75: nt(), 80: nt(), 83: nt(), 90: nt(), 111: nt()}
 	updateAndCheck()
 
 	// Large one
-	state.Delivered.Consumer = 10000
-	state.Delivered.Stream = 10000
-	state.AckFloor.Consumer = 100
-	state.AckFloor.Stream = 100
+	state.Delivered.ConsumerSeq = 10000
+	state.Delivered.StreamSeq = 10000
+	state.AckFloor.ConsumerSeq = 100
+	state.AckFloor.StreamSeq = 100
 	// Generate 8k pending.
-	state.Pending = make(map[uint64]*Pending)
+	state.Pending = make(map[uint64]int64)
 	for len(state.Pending) < 8192 {
 		seq := uint64(rand.Intn(9890) + 101)
 		if _, ok := state.Pending[seq]; !ok {
@@ -1696,8 +1624,8 @@ func TestFileStoreConsumer(t *testing.T) {
 	updateAndCheck()
 
 	state.Pending = nil
-	state.AckFloor.Consumer = 10000
-	state.AckFloor.Stream = 10000
+	state.AckFloor.ConsumerSeq = 10000
+	state.AckFloor.StreamSeq = 10000
 	updateAndCheck()
 }
 
@@ -1922,8 +1850,6 @@ func TestFileStoreStoreLimitRemovePerf(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
-	fs.RegisterStorageUpdates(func(md, bd int64, seq uint64, subj string) {})
-
 	fmt.Printf("storing and removing (limit 1) %d msgs of %s each, totalling %s\n",
 		toStore,
 		FriendlyBytes(int64(storedMsgSize)),
@@ -1948,259 +1874,6 @@ func TestFileStoreStoreLimitRemovePerf(t *testing.T) {
 	fmt.Printf("%s per sec\n", FriendlyBytes(int64(float64(toStore*storedMsgSize)/tt.Seconds())))
 }
 
-func TestFileStorePubPerfWithSmallBlkSize(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
-
-	fmt.Printf("storing %d msgs of %s each, totalling %s\n",
-		toStore,
-		FriendlyBytes(int64(storedMsgSize)),
-		FriendlyBytes(int64(toStore*storedMsgSize)),
-	)
-
-	storeDir, _ := ioutil.TempDir("", JetStreamStoreDir)
-	os.MkdirAll(storeDir, 0755)
-	defer os.RemoveAll(storeDir)
-
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir, BlockSize: FileStoreMinBlkSize},
-		StreamConfig{Name: "zzz", Storage: FileStorage},
-	)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-
-	start := time.Now()
-	for i := 0; i < int(toStore); i++ {
-		fs.StoreMsg(subj, nil, msg)
-	}
-	fs.Stop()
-
-	tt := time.Since(start)
-	fmt.Printf("time to store 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 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()
@@ -2208,6 +1881,7 @@ 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 {
@@ -2215,61 +1889,53 @@ func TestFileStoreConsumerPerf(t *testing.T) {
 	}
 	defer fs.Stop()
 
-	o, err := fs.ConsumerStore("o22", &ConsumerConfig{AckPolicy: AckExplicit})
+	// Test Consumers.
+	o, err := fs.ConsumerStore("obs22", &ConsumerConfig{})
 	if err != nil {
 		t.Fatalf("Unexepected error: %v", err)
 	}
-	// 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)
+	state := &ConsumerState{}
+	toStore := uint64(1000000)
+	fmt.Printf("consumer of %d msgs for ACK NONE\n", toStore)
 
 	start := time.Now()
-
-	ts := start.UnixNano()
-
 	for i := uint64(1); i <= toStore; i++ {
-		if err := o.UpdateDelivered(i, i, 1, ts); err != nil {
-			t.Fatalf("Unexpected error: %v", err)
+		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)
 		}
 	}
 	tt := time.Since(start)
-	fmt.Printf("time to update %d is %v\n", toStore, tt)
+	fmt.Printf("time is %v\n", 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()
-	oc.mu.Lock()
-	buf, err := oc.encodeState()
-	oc.mu.Unlock()
-	if err != nil {
-		t.Fatalf("Error encoding state: %v", err)
+	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)
+		}
 	}
-	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))
+	tt = time.Since(start)
+	fmt.Printf("time is %v\n", tt)
+	fmt.Printf("%.0f updates/sec\n", float64(toStore)/tt.Seconds())
 }
diff --git a/server/gateway_test.go b/server/gateway_test.go
index 44d4adc6..d4de012f 100644
--- a/server/gateway_test.go
+++ b/server/gateway_test.go
@@ -6201,50 +6201,3 @@ func TestGatewayUpdateURLsFromRemoteCluster(t *testing.T) {
 	expected[fmt.Sprintf("127.0.0.1:%d", ob3.Gateway.Port)] = "B3"
 	checkURLs(expected)
 }
-
-type capturePingConn struct {
-	net.Conn
-	ch chan struct{}
-}
-
-func (c *capturePingConn) Write(b []byte) (int, error) {
-	if bytes.Contains(b, []byte(pingProto)) {
-		select {
-		case c.ch <- struct{}{}:
-		default:
-		}
-	}
-	return c.Conn.Write(b)
-}
-
-func TestGatewayPings(t *testing.T) {
-	gatewayMaxPingInterval = 50 * time.Millisecond
-	defer func() { gatewayMaxPingInterval = gwMaxPingInterval }()
-
-	ob := testDefaultOptionsForGateway("B")
-	sb := RunServer(ob)
-	defer sb.Shutdown()
-
-	oa := testGatewayOptionsFromToWithServers(t, "A", "B", sb)
-	sa := RunServer(oa)
-	defer sa.Shutdown()
-
-	waitForInboundGateways(t, sa, 1, 2*time.Second)
-	waitForOutboundGateways(t, sa, 1, 2*time.Second)
-	waitForInboundGateways(t, sb, 1, 2*time.Second)
-	waitForOutboundGateways(t, sb, 1, 2*time.Second)
-
-	c := sa.getOutboundGatewayConnection("B")
-	ch := make(chan struct{}, 1)
-	c.mu.Lock()
-	c.nc = &capturePingConn{c.nc, ch}
-	c.mu.Unlock()
-
-	for i := 0; i < 5; i++ {
-		select {
-		case <-ch:
-		case <-time.After(250 * time.Millisecond):
-			t.Fatalf("Did not send PING")
-		}
-	}
-}
diff --git a/server/sublist_test.go b/server/sublist_test.go
index 43a2ae84..4920fc80 100644
--- a/server/sublist_test.go
+++ b/server/sublist_test.go
@@ -1085,7 +1085,7 @@ func TestSublistRegisterInterestNotification(t *testing.T) {
 		t.Fatalf("Expected to return false on non-existent notification entry")
 	}
 
-	// This should work properly.
+	// This should work.
 	if err := s.RegisterNotification("foo", ch); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1176,28 +1176,27 @@ func TestSublistRegisterInterestNotification(t *testing.T) {
 	}
 
 	// Let's do some wildcard checks.
-	// Wildcards will not trigger interest.
-	subpwc := newSub("*")
-	s.Insert(subpwc)
-	expectNone()
-
 	if err := s.RegisterNotification("foo", ch); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 	expectFalse()
 
-	s.Insert(sub)
+	subpwc := newSub("*")
+	s.Insert(subpwc)
 	expectTrue()
 
+	s.Insert(sub)
+	expectNone()
+
 	s.Remove(sub)
-	expectFalse()
+	expectNone()
 
 	s.Remove(subpwc)
-	expectNone()
+	expectFalse()
 
 	subfwc := newSub(">")
 	s.Insert(subfwc)
-	expectNone()
+	expectTrue()
 
 	s.Insert(subpwc)
 	expectNone()
@@ -1206,7 +1205,7 @@ func TestSublistRegisterInterestNotification(t *testing.T) {
 	expectNone()
 
 	s.Remove(subfwc)
-	expectNone()
+	expectFalse()
 
 	// Test batch
 	subs := []*subscription{sub, sub2, sub3, sub4, subpwc, subfwc}
@@ -1219,29 +1218,6 @@ func TestSublistRegisterInterestNotification(t *testing.T) {
 	expectOne()
 	expectFalse()
 
-	// Test queue subs
-	qsub := newQSub("foo.bar.baz", "1")
-	s.Insert(qsub)
-	expectNone()
-
-	if err := s.RegisterNotification("foo.bar.baz", ch); err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	expectTrue()
-
-	wcqsub := newQSub("foo.bar.>", "1")
-	s.Insert(wcqsub)
-	expectNone()
-
-	s.Remove(qsub)
-	expectFalse()
-
-	s.Remove(wcqsub)
-	expectNone()
-
-	s.Insert(wcqsub)
-	expectNone()
-
 	// Test non-blocking notifications.
 	if err := s.RegisterNotification("bar", ch); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
diff --git a/test/jetstream_test.go b/test/jetstream_test.go
index f974558f..5ec8371d 100644
--- a/test/jetstream_test.go
+++ b/test/jetstream_test.go
@@ -404,109 +404,6 @@ func TestJetStreamPubAck(t *testing.T) {
 	}
 }
 
-func TestJetStreamLookupStreamBySubject(t *testing.T) {
-	s := RunBasicJetStreamServer()
-	defer s.Shutdown()
-
-	if config := s.JetStreamConfig(); config != nil {
-		defer os.RemoveAll(config.StoreDir)
-	}
-	acc := s.GlobalAccount()
-
-	if _, err := acc.AddStream(&server.StreamConfig{Name: "1", Subjects: []string{"foo"}}); err != nil {
-		t.Fatalf("Unexpected error adding stream: %v", err)
-	}
-	if _, err := acc.AddStream(&server.StreamConfig{Name: "2", Subjects: []string{"bar", "baz", "boo"}}); err != nil {
-		t.Fatalf("Unexpected error adding stream: %v", err)
-	}
-	if _, err := acc.AddStream(&server.StreamConfig{Name: "3", Subjects: []string{"foo.*", "bar.*"}}); err != nil {
-		t.Fatalf("Unexpected error adding stream: %v", err)
-	}
-	if _, err := acc.AddStream(&server.StreamConfig{Name: "4", Subjects: []string{"baz.*.*.>"}}); err != nil {
-		t.Fatalf("Unexpected error adding stream: %v", err)
-	}
-	if _, err := acc.AddStream(&server.StreamConfig{Name: "5", Subjects: []string{"{test"}}); err != nil {
-		t.Fatalf("Unexpected error adding stream: %v", err)
-	}
-
-	// Check some errors first.
-	checkError := func(subj string) {
-		t.Helper()
-		if _, _, err := acc.LookupStreamBySubject(subj); err != server.ErrJetStreamStreamNotFound {
-			t.Fatalf("Expected to get a stream not found error, got %v", err)
-		}
-	}
-
-	checkError("zzz")
-	checkError("*")
-	checkError("baz.>")
-
-	checkLookup := func(subj, stream string, filtered bool) {
-		t.Helper()
-		s, f, err := acc.LookupStreamBySubject(subj)
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		if s.Name() != stream {
-			t.Fatalf("Expected stream name of %q, got %q", stream, s.Name())
-		}
-		if f != filtered {
-			t.Fatalf("Expected filtered to be %v, got %v", filtered, f)
-		}
-	}
-
-	checkLookup("foo", "1", false)
-	checkLookup("boo", "2", true)
-	checkLookup("foo.*", "3", true)
-	checkLookup("foo.1", "3", true)
-	checkLookup("baz.*.*.>", "4", false)
-	checkLookup("baz.2.*.>", "4", true)
-
-	// Now test API
-	nc := clientConnectToServer(t, s)
-	defer nc.Close()
-
-	checkAPILookup := func(subj, stream string, filtered bool) {
-		t.Helper()
-		resp, err := nc.Request(server.JSApiStreamLookup, []byte(subj), time.Second)
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		var lresp server.JSApiStreamLookupResponse
-		if err := json.Unmarshal(resp.Data, &lresp); err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		if lresp.Error != nil {
-			t.Fatalf("Got an API error: %+v", lresp.Error)
-		}
-		if lresp.Stream != stream {
-			t.Fatalf("Expected stream name of %q, got %q", stream, lresp.Stream)
-		}
-		if lresp.Filtered != filtered {
-			t.Fatalf("Expected filtered to be %v, got %v", filtered, lresp.Filtered)
-		}
-	}
-
-	cases := []struct {
-		subj     string
-		stream   string
-		filtered bool
-	}{
-		{"foo", "1", false},
-		{"boo", "2", true},
-		{"foo.*", "3", true},
-		{"foo.1", "3", true},
-		{"baz.*.*.>", "4", false},
-		{"baz.2.*.>", "4", true},
-	}
-	for _, c := range cases {
-		checkAPILookup(c.subj, c.stream, c.filtered)
-		checkAPILookup(fmt.Sprintf(`{"subject":%q}`, c.subj), c.stream, c.filtered)
-	}
-
-	checkAPILookup("{test", "5", false)
-}
-
 func TestJetStreamConsumerWithStartTime(t *testing.T) {
 	subj := "my_stream"
 	cases := []struct {
@@ -741,78 +638,6 @@ 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
@@ -2557,7 +2382,7 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 		Name:      "MY_WQ",
 		Subjects:  []string{"foo.*"},
 		Storage:   server.MemoryStorage,
-		MaxAge:    250 * time.Millisecond,
+		MaxAge:    100 * time.Millisecond,
 		Retention: server.WorkQueuePolicy,
 	}
 	fsc := msc
@@ -2593,8 +2418,6 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 			for i := 0; i < toSend; i++ {
 				sendStreamMsg(t, nc, "foo.1", "Hello World!")
 			}
-			nc.Flush()
-
 			state := mset.State()
 			if state.Msgs != uint64(toSend) {
 				t.Fatalf("Expected %d messages, got %d", toSend, state.Msgs)
@@ -2608,20 +2431,17 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 
 			expectPending := func(ep int) {
 				t.Helper()
-				// Now check consumer info.
-				checkFor(t, time.Second, 10*time.Millisecond, func() error {
-					if info, pep := o.Info(), ep+1; int(info.NumPending) != pep {
-						return fmt.Errorf("Expected consumer info pending of %d, got %d", pep, info.NumPending)
-					}
-					return nil
-				})
 				m, err := nc.Request(o.RequestNextMsgSubject(), nil, time.Second)
 				if err != nil {
 					t.Fatalf("Unexpected error: %v", err)
 				}
 				_, _, _, _, pending := o.ReplyInfo(m.Reply)
 				if pending != uint64(ep) {
-					t.Fatalf("Expected ack reply pending of %d, got %d - reply: %q", ep, pending, m.Reply)
+					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)
 				}
 			}
 
@@ -2630,13 +2450,9 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 			for i := 0; i < toSend; i++ {
 				sendStreamMsg(t, nc, "foo.1", "Hello World!")
 			}
-			nc.Flush()
-
 			expectPending(toSend*2 - 2)
 			// Purge and send a new one.
 			mset.Purge()
-			nc.Flush()
-
 			sendStreamMsg(t, nc, "foo.1", "Hello World!")
 			expectPending(0)
 			for i := 0; i < toSend; i++ {
@@ -2655,8 +2471,6 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 			for i := 0; i < toSend; i++ {
 				sendStreamMsg(t, nc, "foo.1", "Hello World!")
 			}
-			nc.Flush()
-
 			// Wait for expiration to kick in.
 			checkFor(t, time.Second, 10*time.Millisecond, func() error {
 				if state := mset.State(); state.Msgs != 0 {
@@ -2678,8 +2492,6 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 			for i := 0; i < toSend; i++ {
 				sendStreamMsg(t, nc, "foo.33", "Hello World!")
 			}
-			nc.Flush()
-
 			if info := o.Info(); info.NumPending != 0 {
 				t.Fatalf("Expected no pending, got %d", info.NumPending)
 			}
@@ -2696,7 +2508,6 @@ func TestJetStreamAckReplyStreamPending(t *testing.T) {
 			for i := 0; i < toSend; i++ {
 				sendStreamMsg(t, nc, "foo.22", "Hello World!")
 			}
-			nc.Flush()
 			expectPending(100)
 			mset.Purge()
 			sendStreamMsg(t, nc, "foo.22", "Hello World!")
@@ -2768,7 +2579,6 @@ func TestJetStreamAckReplyStreamPendingWithAcks(t *testing.T) {
 
 			sub, _ := nc.SubscribeSync(dsubj)
 			defer sub.Unsubscribe()
-
 			checkFor(t, 500*time.Millisecond, 10*time.Millisecond, func() error {
 				if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != toSend {
 					return fmt.Errorf("Did not receive correct number of messages: %d vs %d", nmsgs, toSend)
@@ -3233,7 +3043,7 @@ func TestJetStreamAckNext(t *testing.T) {
 	nc := clientConnectToServer(t, s)
 	defer nc.Close()
 
-	for i := 0; i < 12; i++ {
+	for i := 0; i < 10; i++ {
 		sendStreamMsg(t, nc, mname, fmt.Sprintf("msg %d", i))
 	}
 
@@ -3257,50 +3067,37 @@ func TestJetStreamAckNext(t *testing.T) {
 	if len(q) != 0 {
 		t.Fatalf("Expected empty q got %d", len(q))
 	}
-	if o.Info().AckFloor.Stream != 1 {
+	if o.Info().AckFloor.StreamSeq != 1 {
 		t.Fatalf("First message was not acknowledged")
 	}
 	if !bytes.Equal(msg.Data, []byte("msg 1")) {
 		t.Fatalf("wrong message received, expected: msg 1 got %q", msg.Data)
 	}
 
-	// now ack and request 5 more using a naked number
+	// now ack and request 5 more
 	err = msg.RespondMsg(&nats.Msg{Reply: sub.Subject, Subject: msg.Reply, Data: append(server.AckNext, []byte(" 5")...)})
 	if err != nil {
 		t.Fatalf("RespondMsg failed: %s", err)
 	}
 
-	getMsgs := func(start, count int) {
-		t.Helper()
-
-		ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
-		defer cancel()
+	// check next+cnt worked and got the right amount of messages, use ctx to avoid
+	// sleeps when all is working
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
 
-		for i := start; i < count+1; i++ {
-			select {
-			case msg := <-q:
-				expect := fmt.Sprintf("msg %d", i+1)
-				if !bytes.Equal(msg.Data, []byte(expect)) {
-					t.Fatalf("wrong message received, expected: %s got %#v", expect, msg)
-				}
-			case <-ctx.Done():
-				t.Fatalf("did not receive all messages")
+	for i := 1; i < 6; i++ {
+		select {
+		case msg := <-q:
+			expect := fmt.Sprintf("msg %d", i+1)
+			if !bytes.Equal(msg.Data, []byte(expect)) {
+				t.Fatalf("wrong message received, expected: %s", expect)
 			}
+		case <-ctx.Done():
+			t.Fatalf("did not receive all messages")
 		}
-
-	}
-
-	getMsgs(1, 5)
-
-	// now ack and request 5 more using the full request
-	err = msg.RespondMsg(&nats.Msg{Reply: sub.Subject, Subject: msg.Reply, Data: append(server.AckNext, []byte(`{"batch": 5}`)...)})
-	if err != nil {
-		t.Fatalf("RespondMsg failed: %s", err)
 	}
 
-	getMsgs(6, 10)
-
-	if o.Info().AckFloor.Stream != 2 {
+	if o.Info().AckFloor.StreamSeq != 2 {
 		t.Fatalf("second message was not acknowledged")
 	}
 }
@@ -3788,7 +3585,7 @@ func TestJetStreamConsumerMaxDeliveryAndServerRestart(t *testing.T) {
 		Durable:        "TO",
 		DeliverSubject: dsubj,
 		AckPolicy:      server.AckExplicit,
-		AckWait:        20 * time.Millisecond,
+		AckWait:        25 * time.Millisecond,
 		MaxDeliver:     max,
 	})
 	defer o.Delete()
@@ -3867,7 +3664,9 @@ func TestJetStreamConsumerMaxDeliveryAndServerRestart(t *testing.T) {
 	})
 
 	// Once we are here send third order.
+	// Send third
 	sendStreamMsg(t, nc, mname, "order-3")
+
 	checkNumMsgs(3)
 
 	// Restart.
@@ -4135,7 +3934,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("[%v] Consumer next seq is not correct: %d vs %d", o.Name(), oi.ack+1, o.NextSeq())
+				t.Fatalf("Consumer next seq is not correct: %d vs %d", oi.ack+1, o.NextSeq())
 			}
 		} else {
 			t.Fatalf("Expected to get an consumer")
@@ -4758,7 +4557,6 @@ func TestJetStreamConsumerReconnect(t *testing.T) {
 			sub.Unsubscribe()
 			// Re-establish new sub on same subject.
 			sub, _ = nc.SubscribeSync(delivery)
-			nc.Flush()
 
 			// We should be getting 2 here.
 			sendMsg() // 2
@@ -5676,8 +5474,8 @@ func TestJetStreamStreamPurgeWithConsumer(t *testing.T) {
 				}
 			}
 			state := o.Info()
-			if state.AckFloor.Consumer != 50 {
-				t.Fatalf("Expected ack floor of 50, got %d", state.AckFloor.Consumer)
+			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)
@@ -5694,15 +5492,15 @@ func TestJetStreamStreamPurgeWithConsumer(t *testing.T) {
 			if state.NumAckPending != 0 {
 				t.Fatalf("Expected no pending, got %d", state.NumAckPending)
 			}
-			if state.Delivered.Stream != 100 {
-				t.Fatalf("Expected to have setseq now at next seq of 100, got %d", state.Delivered.Stream)
+			if state.Delivered.StreamSeq != 100 {
+				t.Fatalf("Expected to have setseq now at next seq of 100, got %d", state.Delivered.StreamSeq)
 			}
 			// Check AckFloors which should have also been adjusted.
-			if state.AckFloor.Stream != 100 {
-				t.Fatalf("Expected ackfloor for setseq to be 100, got %d", state.AckFloor.Stream)
+			if state.AckFloor.StreamSeq != 100 {
+				t.Fatalf("Expected ackfloor for setseq to be 100, got %d", state.AckFloor.StreamSeq)
 			}
-			if state.AckFloor.Consumer != 75 {
-				t.Fatalf("Expected ackfloor for obsseq to be 75, got %d", state.AckFloor.Consumer)
+			if state.AckFloor.ConsumerSeq != 75 {
+				t.Fatalf("Expected ackfloor for obsseq to be 75, got %d", state.AckFloor.ConsumerSeq)
 			}
 			// Also make sure we can get new messages correctly.
 			nc.Request("DC", []byte("OK-22"), time.Second)
@@ -5782,15 +5580,15 @@ func TestJetStreamStreamPurgeWithConsumerAndRedelivery(t *testing.T) {
 			if state.NumAckPending != 0 {
 				t.Fatalf("Expected no pending, got %d", state.NumAckPending)
 			}
-			if state.Delivered.Stream != 100 {
-				t.Fatalf("Expected to have setseq now at next seq of 100, got %d", state.Delivered.Stream)
+			if state.Delivered.StreamSeq != 100 {
+				t.Fatalf("Expected to have setseq now at next seq of 100, got %d", state.Delivered.StreamSeq)
 			}
 			// Check AckFloors which should have also been adjusted.
-			if state.AckFloor.Stream != 100 {
-				t.Fatalf("Expected ackfloor for setseq to be 100, got %d", state.AckFloor.Stream)
+			if state.AckFloor.StreamSeq != 100 {
+				t.Fatalf("Expected ackfloor for setseq to be 100, got %d", state.AckFloor.StreamSeq)
 			}
-			if state.AckFloor.Consumer != 50 {
-				t.Fatalf("Expected ackfloor for obsseq to be 75, got %d", state.AckFloor.Consumer)
+			if state.AckFloor.ConsumerSeq != 50 {
+				t.Fatalf("Expected ackfloor for obsseq to be 75, got %d", state.AckFloor.ConsumerSeq)
 			}
 			// Also make sure we can get new messages correctly.
 			nc.Request("DC", []byte("OK-22"), time.Second)
@@ -7142,7 +6940,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	}
 	checkNotFound(bResp.Error, "stream not found")
 
-	// Now create a consumer.
+	// Now create an consumer.
 	delivery := nats.NewInbox()
 	obsReq := server.CreateConsumerRequest{
 		Stream: msetCfg.Name,
@@ -7235,11 +7033,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.Consumer != 10 {
-		t.Fatalf("Expected consumer delivered sequence of 10, got %d", oinfo.Delivered.Consumer)
+	if oinfo.Delivered.ConsumerSeq != 10 {
+		t.Fatalf("Expected consumer delivered sequence of 10, got %d", oinfo.Delivered.ConsumerSeq)
 	}
-	if oinfo.AckFloor.Consumer != 10 {
-		t.Fatalf("Expected ack floor to be 10, got %d", oinfo.AckFloor.Consumer)
+	if oinfo.AckFloor.ConsumerSeq != 10 {
+		t.Fatalf("Expected ack floor to be 10, got %d", oinfo.AckFloor.ConsumerSeq)
 	}
 
 	// Now delete the consumer.
@@ -7583,14 +7381,13 @@ func TestJetStreamAPIConsumerListPaging(t *testing.T) {
 	nc := clientConnectToServer(t, s)
 	defer nc.Close()
 
+	sub, _ := nc.SubscribeSync("d.*")
+	defer sub.Unsubscribe()
+	nc.Flush()
+
 	consumersNum := server.JSApiNamesLimit
 	for i := 1; i <= consumersNum; i++ {
-		dsubj := fmt.Sprintf("d.%d", i)
-		sub, _ := nc.SubscribeSync(dsubj)
-		defer sub.Unsubscribe()
-		nc.Flush()
-
-		_, err := mset.AddConsumer(&server.ConsumerConfig{DeliverSubject: dsubj})
+		_, err := mset.AddConsumer(&server.ConsumerConfig{DeliverSubject: fmt.Sprintf("d.%d", i)})
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -8096,7 +7893,7 @@ func TestJetStreamNextMsgNoInterest(t *testing.T) {
 			}
 			nc.Flush()
 			ostate := o.Info()
-			if ostate.AckFloor.Stream != 11 || ostate.NumAckPending > 0 {
+			if ostate.AckFloor.StreamSeq != 11 || ostate.NumAckPending > 0 {
 				t.Fatalf("Inconsistent ack state: %+v", ostate)
 			}
 		})
@@ -8756,139 +8553,6 @@ func TestJetStreamPubPerf(t *testing.T) {
 	fmt.Printf("%.0f msgs/sec\n", float64(toSend)/tt.Seconds())
 }
 
-func TestJetStreamConsumerPerf(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.MemoryStorage,
-		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 := 2000000
-	for i := 0; i < toStore; i++ {
-		nc.Publish("foo", payload)
-	}
-	nc.Flush()
-
-	_, err = mset.AddConsumer(&server.ConsumerConfig{
-		Durable:        "d",
-		DeliverSubject: "d",
-		AckPolicy:      server.AckNone,
-	})
-	if err != nil {
-		t.Fatalf("Error creating consumer: %v", err)
-	}
-
-	var received int
-	done := make(chan bool)
-
-	nc.Subscribe("d", func(m *nats.Msg) {
-		received++
-		if received >= toStore {
-			done <- true
-		}
-	})
-	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 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()
@@ -9009,14 +8673,13 @@ func TestJetStreamAckExplicitMsgRemoval(t *testing.T) {
 
 			sub2, _ := nc2.SubscribeSync(nats.NewInbox())
 			defer sub2.Unsubscribe()
-			nc2.Flush()
 
 			o2, err := mset.AddConsumer(&server.ConsumerConfig{
 				Durable:        "dur2",
 				DeliverSubject: sub2.Subject,
 				FilterSubject:  "foo.bar",
 				AckPolicy:      server.AckExplicit,
-				AckWait:        100 * time.Millisecond,
+				AckWait:        25 * time.Millisecond,
 			})
 			if err != nil {
 				t.Fatalf("Unexpected error adding consumer: %v", err)
@@ -9045,16 +8708,11 @@ func TestJetStreamAckExplicitMsgRemoval(t *testing.T) {
 				}
 			}
 			// To make sure acks are processed for checking state after sending new ones.
-			checkFor(t, time.Second, 25*time.Millisecond, func() error {
-				if state = mset.State(); state.Msgs != 0 {
-					return fmt.Errorf("Stream still has messages")
-				}
-				return nil
-			})
+			nc1.Flush()
+			nc2.Flush()
 
 			// Now close the 2nd subscription...
 			sub2.Unsubscribe()
-			nc2.Flush()
 
 			// Send 2 more new messages
 			for i := 0; i < toSend; i++ {
@@ -9069,7 +8727,7 @@ func TestJetStreamAckExplicitMsgRemoval(t *testing.T) {
 			for i := 0; i < toSend; i++ {
 				m, err := sub1.NextMsg(time.Second)
 				if err != nil {
-					t.Fatalf("Error getting message to ack: %v", err)
+					t.Fatalf("Error acking message: %v", err)
 				}
 				m.Respond(nil)
 			}
@@ -9085,7 +8743,7 @@ func TestJetStreamAckExplicitMsgRemoval(t *testing.T) {
 				DeliverSubject: sub2.Subject,
 				FilterSubject:  "foo.bar",
 				AckPolicy:      server.AckExplicit,
-				AckWait:        100 * time.Millisecond,
+				AckWait:        25 * time.Millisecond,
 			})
 			if err != nil {
 				t.Fatalf("Unexpected error adding consumer: %v", err)
@@ -9110,589 +8768,3 @@ func TestJetStreamAckExplicitMsgRemoval(t *testing.T) {
 		})
 	}
 }
-
-func TestJetStreamStoredMsgsDontDisappearAfterCacheExpiration(t *testing.T) {
-	sc := &server.StreamConfig{
-		Name:      "MY_STREAM",
-		Storage:   server.FileStorage,
-		Subjects:  []string{"foo.>"},
-		Retention: server.InterestPolicy,
-	}
-
-	s := RunBasicJetStreamServer()
-	defer s.Shutdown()
-
-	if config := s.JetStreamConfig(); config != nil {
-		defer os.RemoveAll(config.StoreDir)
-	}
-
-	mset, err := s.GlobalAccount().AddStreamWithStore(sc, &server.FileStoreConfig{BlockSize: 128, CacheExpire: 15 * time.Millisecond})
-	if err != nil {
-		t.Fatalf("Unexpected error adding stream: %v", err)
-	}
-	defer mset.Delete()
-
-	nc1 := clientConnectWithOldRequest(t, s)
-	defer nc1.Close()
-
-	// Create a durable consumers
-	sub, _ := nc1.SubscribeSync(nats.NewInbox())
-	defer sub.Unsubscribe()
-	nc1.Flush()
-
-	o, err := mset.AddConsumer(&server.ConsumerConfig{
-		Durable:        "dur",
-		DeliverSubject: sub.Subject,
-		FilterSubject:  "foo.bar",
-		DeliverPolicy:  server.DeliverNew,
-		AckPolicy:      server.AckExplicit,
-	})
-	if err != nil {
-		t.Fatalf("Unexpected error adding consumer: %v", err)
-	}
-	defer o.Delete()
-
-	nc2 := clientConnectWithOldRequest(t, s)
-	defer nc2.Close()
-
-	sendStreamMsg(t, nc2, "foo.bar", "msg1")
-
-	msg, err := sub.NextMsg(time.Second)
-	if err != nil {
-		t.Fatalf("Did not get message: %v", err)
-	}
-	if string(msg.Data) != "msg1" {
-		t.Fatalf("Unexpected message: %q", msg.Data)
-	}
-
-	nc1.Close()
-
-	// Get the message from the stream
-	getMsgSeq := func(seq uint64) {
-		t.Helper()
-		mreq := &server.JSApiMsgGetRequest{Seq: seq}
-		req, err := json.Marshal(mreq)
-		if err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-		smsgj, err := nc2.Request(fmt.Sprintf(server.JSApiMsgGetT, sc.Name), req, time.Second)
-		if err != nil {
-			t.Fatalf("Could not retrieve stream message: %v", err)
-		}
-		if strings.Contains(string(smsgj.Data), "code") {
-			t.Fatalf("Error: %q", smsgj.Data)
-		}
-	}
-
-	getMsgSeq(1)
-
-	time.Sleep(time.Second)
-
-	sendStreamMsg(t, nc2, "foo.bar", "msg2")
-	sendStreamMsg(t, nc2, "foo.bar", "msg3")
-
-	getMsgSeq(1)
-	getMsgSeq(2)
-	getMsgSeq(3)
-}
-
-func TestJetStreamConsumerUpdateRedelivery(t *testing.T) {
-	cases := []struct {
-		name    string
-		mconfig *server.StreamConfig
-	}{
-		{"MemoryStore", &server.StreamConfig{
-			Name:      "MY_STREAM",
-			Storage:   server.MemoryStorage,
-			Subjects:  []string{"foo.>"},
-			Retention: server.InterestPolicy,
-		}},
-		{"FileStore", &server.StreamConfig{
-			Name:      "MY_STREAM",
-			Storage:   server.FileStorage,
-			Subjects:  []string{"foo.>"},
-			Retention: server.InterestPolicy,
-		}},
-	}
-	for _, c := range cases {
-		t.Run(c.name, func(t *testing.T) {
-			s := RunBasicJetStreamServer()
-			defer s.Shutdown()
-
-			if config := s.JetStreamConfig(); config != nil {
-				defer os.RemoveAll(config.StoreDir)
-			}
-
-			mset, err := s.GlobalAccount().AddStream(c.mconfig)
-			if err != nil {
-				t.Fatalf("Unexpected error adding stream: %v", err)
-			}
-			defer mset.Delete()
-
-			nc := clientConnectToServer(t, s)
-			defer nc.Close()
-
-			// Create a durable consumer.
-			sub, _ := nc.SubscribeSync(nats.NewInbox())
-			defer sub.Unsubscribe()
-
-			o, err := mset.AddConsumer(&server.ConsumerConfig{
-				Durable:        "dur22",
-				DeliverSubject: sub.Subject,
-				FilterSubject:  "foo.bar",
-				AckPolicy:      server.AckExplicit,
-				AckWait:        100 * time.Millisecond,
-				MaxDeliver:     3,
-			})
-			if err != nil {
-				t.Fatalf("Unexpected error adding consumer: %v", err)
-			}
-			defer o.Delete()
-
-			// Send 20 messages
-			toSend := 20
-			for i := 1; i <= toSend; i++ {
-				sendStreamMsg(t, nc, "foo.bar", fmt.Sprintf("msg-%v", i))
-			}
-			state := mset.State()
-			if state.Msgs != uint64(toSend) {
-				t.Fatalf("Expected %v messages, got %d", toSend, state.Msgs)
-			}
-
-			// Receive the messages and ack only every 4th
-			for i := 0; i < toSend; i++ {
-				m, err := sub.NextMsg(time.Second)
-				if err != nil {
-					t.Fatalf("Error getting message: %v", err)
-				}
-				seq, _, _, _, _ := o.ReplyInfo(m.Reply)
-				// 4, 8, 12, 16, 20
-				if seq%4 == 0 {
-					m.Respond(nil)
-				}
-			}
-
-			// Now close the sub and open a new one and update the consumer.
-			sub.Unsubscribe()
-
-			// Wait for it to become inactive
-			checkFor(t, 200*time.Millisecond, 10*time.Millisecond, func() error {
-				if o.Active() {
-					return fmt.Errorf("Consumer still active")
-				}
-				return nil
-			})
-
-			// Send 20 more messages.
-			for i := toSend; i < toSend*2; i++ {
-				sendStreamMsg(t, nc, "foo.bar", fmt.Sprintf("msg-%v", i))
-			}
-
-			// Create new subscription.
-			sub, _ = nc.SubscribeSync(nats.NewInbox())
-			defer sub.Unsubscribe()
-			nc.Flush()
-
-			o, err = mset.AddConsumer(&server.ConsumerConfig{
-				Durable:        "dur22",
-				DeliverSubject: sub.Subject,
-				FilterSubject:  "foo.bar",
-				AckPolicy:      server.AckExplicit,
-				AckWait:        100 * time.Millisecond,
-				MaxDeliver:     3,
-			})
-			if err != nil {
-				t.Fatalf("Unexpected error adding consumer: %v", err)
-			}
-			defer o.Delete()
-
-			expect := toSend + toSend - 5 // mod 4 acks
-			checkFor(t, time.Second, 5*time.Millisecond, func() error {
-				if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != expect {
-					return fmt.Errorf("Did not receive correct number of messages: %d vs %d", nmsgs, expect)
-				}
-				return nil
-			})
-
-			for i, eseq := 0, uint64(1); i < expect; i++ {
-				m, err := sub.NextMsg(time.Second)
-				if err != nil {
-					t.Fatalf("Error getting message: %v", err)
-				}
-				// Skip the ones we ack'd from above. We should not get them back here.
-				if eseq <= uint64(toSend) && eseq%4 == 0 {
-					eseq++
-				}
-				seq, _, dc, _, _ := o.ReplyInfo(m.Reply)
-				if seq != eseq {
-					t.Fatalf("Expected stream sequence of %d, got %d", eseq, seq)
-				}
-				if seq <= uint64(toSend) && dc != 2 {
-					t.Fatalf("Expected delivery count of 2 for sequence of %d, got %d", seq, dc)
-				}
-				if seq > uint64(toSend) && dc != 1 {
-					t.Fatalf("Expected delivery count of 1 for sequence of %d, got %d", seq, dc)
-				}
-				if seq > uint64(toSend) {
-					m.Respond(nil) // Ack
-				}
-				eseq++
-			}
-
-			// We should get the second half back since we did not ack those from above.
-			expect = toSend - 5
-			checkFor(t, time.Second, 5*time.Millisecond, func() error {
-				if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != expect {
-					return fmt.Errorf("Did not receive correct number of messages: %d vs %d", nmsgs, expect)
-				}
-				return nil
-			})
-
-			for i, eseq := 0, uint64(1); i < expect; i++ {
-				m, err := sub.NextMsg(time.Second)
-				if err != nil {
-					t.Fatalf("Error getting message: %v", err)
-				}
-				// Skip the ones we ack'd from above. We should not get them back here.
-				if eseq <= uint64(toSend) && eseq%4 == 0 {
-					eseq++
-				}
-				seq, _, dc, _, _ := o.ReplyInfo(m.Reply)
-				if seq != eseq {
-					t.Fatalf("Expected stream sequence of %d, got %d", eseq, seq)
-				}
-				if dc != 3 {
-					t.Fatalf("Expected delivery count of 3 for sequence of %d, got %d", seq, dc)
-				}
-				eseq++
-			}
-		})
-	}
-}
-
-func TestJetStreamConsumerMaxAckPending(t *testing.T) {
-	cases := []struct {
-		name    string
-		mconfig *server.StreamConfig
-	}{
-		{"MemoryStore", &server.StreamConfig{
-			Name:     "MY_STREAM",
-			Storage:  server.MemoryStorage,
-			Subjects: []string{"foo.*"},
-		}},
-		{"FileStore", &server.StreamConfig{
-			Name:     "MY_STREAM",
-			Storage:  server.FileStorage,
-			Subjects: []string{"foo.*"},
-		}},
-	}
-	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()
-
-			// Do error scenarios.
-			_, err = mset.AddConsumer(&server.ConsumerConfig{
-				Durable:        "d22",
-				DeliverSubject: nats.NewInbox(),
-				AckPolicy:      server.AckNone,
-				MaxAckPending:  1,
-			})
-			if err == nil {
-				t.Fatalf("Expected error, MaxAckPending only applicable to ack != AckNone")
-			}
-
-			// Queue up 100 messages.
-			toSend := 100
-			for i := 0; i < toSend; i++ {
-				sendStreamMsg(t, nc, "foo.bar", fmt.Sprintf("MSG: %d", i+1))
-			}
-
-			// Limit to 33
-			maxAckPending := 33
-
-			o, err := mset.AddConsumer(&server.ConsumerConfig{
-				Durable:        "d22",
-				DeliverSubject: nats.NewInbox(),
-				AckPolicy:      server.AckExplicit,
-				MaxAckPending:  maxAckPending,
-			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
-			defer o.Delete()
-
-			sub, _ := nc.SubscribeSync(o.Info().Config.DeliverSubject)
-			defer sub.Unsubscribe()
-
-			checkSubPending := func(numExpected int) {
-				t.Helper()
-				checkFor(t, 200*time.Millisecond, 10*time.Millisecond, func() error {
-					if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != numExpected {
-						return fmt.Errorf("Did not receive correct number of messages: %d vs %d", nmsgs, numExpected)
-					}
-					return nil
-				})
-			}
-
-			checkSubPending(maxAckPending)
-			// We hit the limit, double check we stayed there.
-			if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != maxAckPending {
-				t.Fatalf("Too many messages received: %d vs %d", nmsgs, maxAckPending)
-			}
-
-			// Now ack them all.
-			for i := 0; i < maxAckPending; i++ {
-				m, err := sub.NextMsg(time.Second)
-				if err != nil {
-					t.Fatalf("Error receiving message %d: %v", i, err)
-				}
-				m.Respond(nil)
-			}
-			checkSubPending(maxAckPending)
-
-			o.Stop()
-			mset.Purge()
-
-			// Now test a consumer that is live while we publish messages to the stream.
-			o, err = mset.AddConsumer(&server.ConsumerConfig{
-				Durable:        "d22",
-				DeliverSubject: nats.NewInbox(),
-				AckPolicy:      server.AckExplicit,
-				MaxAckPending:  maxAckPending,
-			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
-			defer o.Delete()
-
-			sub, _ = nc.SubscribeSync(o.Info().Config.DeliverSubject)
-			defer sub.Unsubscribe()
-			nc.Flush()
-
-			checkSubPending(0)
-
-			// Now stream more then maxAckPending.
-			for i := 0; i < toSend; i++ {
-				sendStreamMsg(t, nc, "foo.baz", fmt.Sprintf("MSG: %d", i+1))
-			}
-			checkSubPending(maxAckPending)
-			// We hit the limit, double check we stayed there.
-			if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != maxAckPending {
-				t.Fatalf("Too many messages received: %d vs %d", nmsgs, maxAckPending)
-			}
-		})
-	}
-}
-
-func TestJetStreamPullConsumerMaxAckPending(t *testing.T) {
-	cases := []struct {
-		name    string
-		mconfig *server.StreamConfig
-	}{
-		{"MemoryStore", &server.StreamConfig{
-			Name:     "MY_STREAM",
-			Storage:  server.MemoryStorage,
-			Subjects: []string{"foo.*"},
-		}},
-		{"FileStore", &server.StreamConfig{
-			Name:     "MY_STREAM",
-			Storage:  server.FileStorage,
-			Subjects: []string{"foo.*"},
-		}},
-	}
-	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()
-
-			// Queue up 100 messages.
-			toSend := 100
-			for i := 0; i < toSend; i++ {
-				sendStreamMsg(t, nc, "foo.bar", fmt.Sprintf("MSG: %d", i+1))
-			}
-
-			// Limit to 33
-			maxAckPending := 33
-
-			o, err := mset.AddConsumer(&server.ConsumerConfig{
-				Durable:       "d22",
-				AckPolicy:     server.AckExplicit,
-				MaxAckPending: maxAckPending,
-			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
-			defer o.Delete()
-
-			getSubj := o.RequestNextMsgSubject()
-
-			var toAck []*nats.Msg
-
-			for i := 0; i < maxAckPending; i++ {
-				if m, err := nc.Request(getSubj, nil, time.Second); err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				} else {
-					toAck = append(toAck, m)
-				}
-			}
-			// This should fail.. But we do not want to queue up our request.
-			req := &server.JSApiConsumerGetNextRequest{Batch: 1, NoWait: true}
-			jreq, _ := json.Marshal(req)
-			m, err := nc.Request(getSubj, jreq, time.Second)
-			if err != nil {
-				t.Fatalf("Unexpected error: %v", err)
-			}
-			if m.Header.Get("Status") != "409" {
-				t.Fatalf("Expected a 409 status code, got %q", m.Header.Get("Status"))
-			}
-
-			// Now ack them all.
-			for _, m := range toAck {
-				m.Respond(nil)
-			}
-
-			// Now do batch above the max.
-			sub, _ := nc.SubscribeSync(nats.NewInbox())
-			defer sub.Unsubscribe()
-
-			checkSubPending := func(numExpected int) {
-				t.Helper()
-				checkFor(t, time.Second, 10*time.Millisecond, func() error {
-					if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != numExpected {
-						return fmt.Errorf("Did not receive correct number of messages: %d vs %d", nmsgs, numExpected)
-					}
-					return nil
-				})
-			}
-
-			req = &server.JSApiConsumerGetNextRequest{Batch: toSend}
-			jreq, _ = json.Marshal(req)
-			nc.PublishRequest(getSubj, sub.Subject, jreq)
-
-			checkSubPending(maxAckPending)
-			// We hit the limit, double check we stayed there.
-			if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != maxAckPending {
-				t.Fatalf("Too many messages received: %d vs %d", nmsgs, maxAckPending)
-			}
-		})
-	}
-}
-
-func TestJetStreamPullConsumerMaxAckPendingRedeliveries(t *testing.T) {
-	cases := []struct {
-		name    string
-		mconfig *server.StreamConfig
-	}{
-		{"MemoryStore", &server.StreamConfig{
-			Name:     "MY_STREAM",
-			Storage:  server.MemoryStorage,
-			Subjects: []string{"foo.*"},
-		}},
-		{"FileStore", &server.StreamConfig{
-			Name:     "MY_STREAM",
-			Storage:  server.FileStorage,
-			Subjects: []string{"foo.*"},
-		}},
-	}
-	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()
-
-			// Queue up 10 messages.
-			toSend := 10
-			for i := 0; i < toSend; i++ {
-				sendStreamMsg(t, nc, "foo.bar", fmt.Sprintf("MSG: %d", i+1))
-			}
-
-			// Limit to 1
-			maxAckPending := 1
-			ackWait := 20 * time.Millisecond
-			expSeq := uint64(4)
-
-			o, err := mset.AddConsumer(&server.ConsumerConfig{
-				Durable:       "d22",
-				DeliverPolicy: server.DeliverByStartSequence,
-				OptStartSeq:   expSeq,
-				AckPolicy:     server.AckExplicit,
-				AckWait:       ackWait,
-				MaxAckPending: maxAckPending,
-			})
-			if err != nil {
-				t.Fatalf("Expected no error, got %v", err)
-			}
-			defer o.Delete()
-
-			getSubj := o.RequestNextMsgSubject()
-			delivery := uint64(1)
-
-			getNext := func() {
-				t.Helper()
-				m, err := nc.Request(getSubj, nil, time.Second)
-				if err != nil {
-					t.Fatalf("Unexpected error: %v", err)
-				}
-				sseq, dseq, dcount, _, pending := o.ReplyInfo(m.Reply)
-				if sseq != expSeq {
-					t.Fatalf("Expected stream sequence of %d, got %d", expSeq, sseq)
-				}
-				if dseq != delivery {
-					t.Fatalf("Expected consumer sequence of %d, got %d", delivery, dseq)
-				}
-				if dcount != delivery {
-					t.Fatalf("Expected delivery count of %d, got %d", delivery, dcount)
-				}
-				if pending != uint64(toSend)-expSeq {
-					t.Fatalf("Expected pending to be %d, got %d", uint64(toSend)-expSeq, pending)
-				}
-				delivery++
-			}
-
-			getNext()
-			getNext()
-			getNext()
-			getNext()
-			getNext()
-		})
-	}
-}