Repository: nats-io/nats-server
Pull request #2776 (2022-01-12 00:36:08)

Pull Consumer Updates
This change cleans up alot of the inner workings of pull consumers to be more resilient, better at waiting request management, and the ability to determine and watch interest cross account.

We also introduce new consumer config items to limit next msg requests' batch size and maximum wait times as well as indicate inactive thresholds.

We also now allow pull consumers that are ephemeral (they will auto-cleanup) and can have ack-all, not just ack explicit.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/go.mod b/go.mod
index ae67a9a6..f6d29dd9 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/nats-io/nats-server/v2
 
-go 1.16
+go 1.17
 
 require (
 	github.com/golang/protobuf v1.4.2 // indirect
@@ -10,7 +10,7 @@ require (
 	github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
-	golang.org/x/crypto v0.0.0-20211202192323-5770296d904e
+	golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3
 	golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1
 	golang.org/x/time v0.0.0-20200416051211-89c76fbcd5d1
 )
diff --git a/go.sum b/go.sum
index cb6bbf29..58c1ab13 100644
--- a/go.sum
+++ b/go.sum
@@ -18,6 +18,8 @@ github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296 h1:vU9tpM3apjYlLL
 github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
 github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc h1:SHr4MUUZJ/fAC0uSm2OzWOJYsHpapmR86mpw7q1qPXU=
 github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nats.go v1.13.1-0.20220106142636-ebcdff697d3f h1:WtlVcCLq7NOOskKidJ1rRZoXwOTs6RIJSl0R/gIPBMg=
+github.com/nats-io/nats.go v1.13.1-0.20220106142636-ebcdff697d3f/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
@@ -25,6 +27,8 @@ github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OS
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20211202192323-5770296d904e h1:MUP6MR3rJ7Gk9LEia0LP2ytiH6MuCfs7qYz+47jGdD8=
 golang.org/x/crypto v0.0.0-20211202192323-5770296d904e/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3 h1:0es+/5331RGQPcXlMfP+WrnIIS6dNnNRe0WB02W0F4M=
+golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
diff --git a/server/accounts.go b/server/accounts.go
index e2a9ac22..d2102589 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -1,4 +1,4 @@
-// Copyright 2018-2021 The NATS Authors
+// Copyright 2018-2022 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
@@ -1242,9 +1242,7 @@ func (a *Account) sendReplyInterestLostTrackLatency(si *serviceImport) {
 		Error:  "Request Timeout",
 	}
 	a.mu.RLock()
-	rc := si.rc
-	share := si.share
-	ts := si.ts
+	rc, share, ts := si.rc, si.share, si.ts
 	sl.RequestHeader = si.trackingHdr
 	a.mu.RUnlock()
 	if rc != nil {
@@ -1257,9 +1255,7 @@ func (a *Account) sendReplyInterestLostTrackLatency(si *serviceImport) {
 func (a *Account) sendBackendErrorTrackingLatency(si *serviceImport, reason rsiReason) {
 	sl := &ServiceLatency{}
 	a.mu.RLock()
-	rc := si.rc
-	share := si.share
-	ts := si.ts
+	rc, share, ts := si.rc, si.share, si.ts
 	sl.RequestHeader = si.trackingHdr
 	a.mu.RUnlock()
 	if rc != nil {
@@ -1568,14 +1564,17 @@ func (a *Account) removeRespServiceImport(si *serviceImport, reason rsiReason) {
 	}
 
 	a.mu.Lock()
+	c := a.ic
 	delete(a.exports.responses, si.from)
-	dest := si.acc
-	to := si.to
-	tracking := si.tracking
-	rc := si.rc
+	dest, to, tracking, rc, didDeliver := si.acc, si.to, si.tracking, si.rc, si.didDeliver
 	a.mu.Unlock()
 
-	if tracking && rc != nil {
+	// If we have a sid make sure to unsub.
+	if len(si.sid) > 0 && c != nil {
+		c.processUnsub(si.sid)
+	}
+
+	if tracking && rc != nil && !didDeliver {
 		a.sendBackendErrorTrackingLatency(si, reason)
 	}
 
@@ -1714,12 +1713,17 @@ func (a *Account) checkForReverseEntry(reply string, si *serviceImport, checkInt
 			var trackingCleanup bool
 			var rsi *serviceImport
 			acc.mu.Lock()
+			c := acc.ic
 			if rsi = acc.exports.responses[sre.msub]; rsi != nil && !rsi.didDeliver {
 				delete(acc.exports.responses, rsi.from)
 				trackingCleanup = rsi.tracking && rsi.rc != nil
 			}
 			acc.mu.Unlock()
-
+			// If we are doing explicit subs for all responses (e.g. bound to leafnode)
+			// we will have a non-empty sid here.
+			if rsi != nil && len(rsi.sid) > 0 && c != nil {
+				c.processUnsub(rsi.sid)
+			}
 			if trackingCleanup {
 				acc.sendReplyInterestLostTrackLatency(rsi)
 			}
@@ -1842,10 +1846,18 @@ func (a *Account) internalClient() *client {
 
 // Internal account scoped subscriptions.
 func (a *Account) subscribeInternal(subject string, cb msgHandler) (*subscription, error) {
+	return a.subscribeInternalEx(subject, cb, false)
+}
+
+// Creates internal subscription for service import responses.
+func (a *Account) subscribeServiceImportResponse(subject string) (*subscription, error) {
+	return a.subscribeInternalEx(subject, a.processServiceImportResponse, true)
+}
+
+func (a *Account) subscribeInternalEx(subject string, cb msgHandler, ri bool) (*subscription, error) {
 	a.mu.Lock()
-	c := a.internalClient()
 	a.isid++
-	sid := strconv.FormatUint(a.isid, 10)
+	c, sid := a.internalClient(), strconv.FormatUint(a.isid, 10)
 	a.mu.Unlock()
 
 	// This will happen in parsing when the account has not been properly setup.
@@ -1853,7 +1865,7 @@ func (a *Account) subscribeInternal(subject string, cb msgHandler) (*subscriptio
 		return nil, fmt.Errorf("no internal account client")
 	}
 
-	return c.processSub([]byte(subject), nil, []byte(sid), cb, false)
+	return c.processSubEx([]byte(subject), nil, []byte(sid), cb, false, false, ri)
 }
 
 // This will add an account subscription that matches the "from" from a service import entry.
@@ -2062,11 +2074,11 @@ func (a *Account) processServiceImportResponse(sub *subscription, c *client, _ *
 	c.processServiceImport(si, a, msg)
 }
 
-// Will create a wildcard subscription to handle interest graph propagation for all
-// service replies.
-// Lock should not be held.
-func (a *Account) createRespWildcard() []byte {
-	a.mu.Lock()
+// Will create the response prefix for fast generation of responses.
+// A wildcard subscription may be used handle interest graph propagation
+// for all service replies, unless we are bound to a leafnode.
+// Lock should be held.
+func (a *Account) createRespWildcard() {
 	var b = [baseServerLen]byte{'_', 'R', '_', '.'}
 	rn := a.prand.Uint64()
 	for i, l := replyPrefixLen, rn; i < len(b); i++ {
@@ -2074,17 +2086,6 @@ func (a *Account) createRespWildcard() []byte {
 		l /= base
 	}
 	a.siReply = append(b[:], '.')
-	pre := a.siReply
-	wcsub := append(a.siReply, '>')
-	c := a.internalClient()
-	a.isid++
-	sid := strconv.FormatUint(a.isid, 10)
-	a.mu.Unlock()
-
-	// Create subscription and internal callback for all the wildcard response subjects.
-	c.processSubEx(wcsub, nil, []byte(sid), a.processServiceImportResponse, false, false, true)
-
-	return pre
 }
 
 // Test whether this is a tracked reply.
@@ -2097,17 +2098,27 @@ func isTrackedReply(reply []byte) bool {
 // FIXME(dlc) - probably do not have to use rand here. about 25ns per.
 func (a *Account) newServiceReply(tracking bool) []byte {
 	a.mu.Lock()
-	s, replyPre := a.srv, a.siReply
+	s := a.srv
 	if a.prand == nil {
 		var h maphash.Hash
 		h.WriteString(nuid.Next())
 		a.prand = rand.New(rand.NewSource(int64(h.Sum64())))
 	}
 	rn := a.prand.Uint64()
+
+	// Check if we need to create the reply here.
+	var createdSiReply bool
+	if a.siReply == nil {
+		a.createRespWildcard()
+		createdSiReply = true
+	}
+	replyPre, isBoundToLeafnode := a.siReply, a.lds != _EMPTY_
 	a.mu.Unlock()
 
-	if replyPre == nil {
-		replyPre = a.createRespWildcard()
+	// If we created the siReply and we are not bound to a leafnode
+	// we need to do the wildcard subscription.
+	if createdSiReply && !isBoundToLeafnode {
+		a.subscribeServiceImportResponse(string(append(replyPre, '>')))
 	}
 
 	var b [replyLen]byte
@@ -2127,6 +2138,7 @@ func (a *Account) newServiceReply(tracking bool) []byte {
 		reply = append(reply, s.sys.shash...)
 		reply = append(reply, '.', 'T')
 	}
+
 	return reply
 }
 
@@ -2256,12 +2268,19 @@ func (a *Account) addRespServiceImport(dest *Account, to string, osi *serviceImp
 		si.tracking = true
 		si.trackingHdr = header
 	}
+	isBoundToLeafnode := a.lds != _EMPTY_
 	a.mu.Unlock()
 
-	// We do not do individual subscriptions here like we do on configured imports.
+	// We might not do individual subscriptions here like we do on configured imports.
+	// If we are bound to a leafnode we do explicit subscriptions for these.
 	// We have an internal callback for all responses inbound to this account and
 	// will process appropriately there. This does not pollute the sublist and the caches.
 
+	if isBoundToLeafnode {
+		sub, _ := a.subscribeServiceImportResponse(nrr)
+		si.sid = sub.sid
+	}
+
 	// We do add in the reverse map such that we can detect loss of interest and do proper
 	// cleanup of this si as interest goes away.
 	dest.addReverseRespMapEntry(a, to, nrr)
diff --git a/server/consumer.go b/server/consumer.go
index 90dba309..f1675248 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -25,7 +25,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/nats-io/nuid"
@@ -50,8 +49,6 @@ type ConsumerInfo struct {
 type ConsumerConfig struct {
 	Durable         string        `json:"durable_name,omitempty"`
 	Description     string        `json:"description,omitempty"`
-	DeliverSubject  string        `json:"deliver_subject,omitempty"`
-	DeliverGroup    string        `json:"deliver_group,omitempty"`
 	DeliverPolicy   DeliverPolicy `json:"deliver_policy"`
 	OptStartSeq     uint64        `json:"opt_start_seq,omitempty"`
 	OptStartTime    *time.Time    `json:"opt_start_time,omitempty"`
@@ -68,6 +65,17 @@ type ConsumerConfig struct {
 	FlowControl     bool          `json:"flow_control,omitempty"`
 	HeadersOnly     bool          `json:"headers_only,omitempty"`
 
+	// Pull based options.
+	MaxRequestBatch   int           `json:"max_batch,omitempty"`
+	MaxRequestExpires time.Duration `json:"max_expires,omitempty"`
+
+	// Push based consumers.
+	DeliverSubject string `json:"deliver_subject,omitempty"`
+	DeliverGroup   string `json:"deliver_group,omitempty"`
+
+	// Ephemeral inactivity threshold.
+	InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
+
 	// Don't add to general clients.
 	Direct bool `json:"direct,omitempty"`
 }
@@ -199,6 +207,7 @@ type consumer struct {
 	adflr             uint64
 	asflr             uint64
 	sgap              uint64
+	lsgap             uint64
 	dsubj             string
 	qgroup            string
 	lss               *lastSeqSkipList
@@ -342,14 +351,9 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		}
 	} else {
 		// Pull mode / work queue mode require explicit ack.
-		if config.AckPolicy != AckExplicit {
+		if config.AckPolicy == AckNone {
 			return nil, NewJSConsumerPullRequiresAckError()
 		}
-		// They are also required to be durable since otherwise we will not know when to
-		// clean them up.
-		if config.Durable == _EMPTY_ {
-			return nil, NewJSConsumerPullNotDurableError()
-		}
 		if config.RateLimit > 0 {
 			return nil, NewJSConsumerPullWithRateLimitError()
 		}
@@ -362,6 +366,12 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		if config.FlowControl {
 			return nil, NewJSConsumerFCRequiresPushError()
 		}
+		if config.MaxRequestBatch < 0 {
+			return nil, NewJSConsumerMaxRequestBatchNegativeError()
+		}
+		if config.MaxRequestExpires != 0 && config.MaxRequestExpires < time.Millisecond {
+			return nil, NewJSConsumerMaxRequestExpiresToSmallError()
+		}
 	}
 
 	// Direct need to be non-mapped ephemerals.
@@ -416,15 +426,8 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		if config.OptStartTime != nil {
 			return nil, NewJSConsumerInvalidPolicyError(badStart("last per subject", "time"))
 		}
-		badConfig := config.FilterSubject == _EMPTY_
-		if !badConfig {
-			subjects, ext := mset.allSubjects()
-			if len(subjects) == 1 && !ext && subjects[0] == config.FilterSubject && subjectIsLiteral(subjects[0]) {
-				badConfig = true
-			}
-		}
-		if badConfig {
-			return nil, NewJSConsumerInvalidPolicyError(notSet("deliver last per subject", "filter subject"))
+		if config.FilterSubject == _EMPTY_ {
+			return nil, NewJSConsumerInvalidPolicyError(notSet("last per subject", "filter subject"))
 		}
 	case DeliverNew:
 		if config.OptStartSeq > 0 {
@@ -435,17 +438,17 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		}
 	case DeliverByStartSequence:
 		if config.OptStartSeq == 0 {
-			return nil, NewJSConsumerInvalidPolicyError(notSet("deliver by start sequence", "start sequence"))
+			return nil, NewJSConsumerInvalidPolicyError(notSet("by start sequence", "start sequence"))
 		}
 		if config.OptStartTime != nil {
-			return nil, NewJSConsumerInvalidPolicyError(badStart("deliver by start sequence", "time"))
+			return nil, NewJSConsumerInvalidPolicyError(badStart("by start sequence", "time"))
 		}
 	case DeliverByStartTime:
 		if config.OptStartTime == nil {
-			return nil, NewJSConsumerInvalidPolicyError(notSet("deliver by start time", "start time"))
+			return nil, NewJSConsumerInvalidPolicyError(notSet("by start time", "start time"))
 		}
 		if config.OptStartSeq != 0 {
-			return nil, NewJSConsumerInvalidPolicyError(badStart("deliver by start time", "start sequence"))
+			return nil, NewJSConsumerInvalidPolicyError(badStart("by start time", "start sequence"))
 		}
 	}
 
@@ -555,9 +558,6 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 			return nil, NewJSConsumerNameTooLongError(JSMaxNameLen)
 		}
 		o.name = config.Durable
-		if o.isPullMode() {
-			o.waiting = newWaitQueue(config.MaxWaiting)
-		}
 	} else if oname != _EMPTY_ {
 		o.name = oname
 	} else {
@@ -568,6 +568,10 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 			}
 		}
 	}
+	// Create our request waiting queue.
+	if o.isPullMode() {
+		o.waiting = newWaitQueue(config.MaxWaiting)
+	}
 
 	// Check if we have  filtered subject that is a wildcard.
 	if config.FilterSubject != _EMPTY_ && subjectHasWildcard(config.FilterSubject) {
@@ -634,9 +638,17 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	o.ackSubj = fmt.Sprintf("%s.*.*.*.*.*", pre)
 	o.nextMsgSubj = fmt.Sprintf(JSApiRequestNextT, mn, o.name)
 
+	// If not durable determine the inactive threshold.
+	if !o.isDurable() {
+		if o.cfg.InactiveThreshold != 0 {
+			o.dthresh = o.cfg.InactiveThreshold
+		} else {
+			// Add in 1 sec of jitter above and beyond the default of 5s.
+			o.dthresh = JsDeleteWaitTimeDefault + time.Duration(rand.Int63n(1000))*time.Millisecond
+		}
+	}
+
 	if o.isPushMode() {
-		// Add in 1 sec of jitter above and beyond the default of 5s.
-		o.dthresh = JsDeleteWaitTimeDefault + time.Duration(rand.Int63n(1000))*time.Millisecond
 		if !o.isDurable() {
 			// Check if we are not durable that the delivery subject has interest.
 			// Check in place here for interest. Will setup properly in setLeader.
@@ -698,7 +710,7 @@ func (o *consumer) setConsumerAssignment(ca *consumerAssignment) {
 	}
 }
 
-// checkInterest will check on our interest's queue group status.
+// checkQueueInterest will check on our interest's queue group status.
 // Lock should be held.
 func (o *consumer) checkQueueInterest() {
 	if !o.active || o.cfg.DeliverSubject == _EMPTY_ {
@@ -798,6 +810,12 @@ func (o *consumer) setLeader(isLeader bool) {
 				stopAndClearTimer(&o.gwdtmr)
 				o.gwdtmr = time.AfterFunc(time.Second, func() { o.watchGWinterest() })
 			}
+		} else if !o.isDurable() {
+			// Ephemeral pull consumer. We run the dtmr all the time for this one.
+			if o.dtmr != nil {
+				stopAndClearTimer(&o.dtmr)
+			}
+			o.dtmr = time.AfterFunc(o.dthresh, func() { o.deleteNotActive() })
 		}
 
 		// If we are not in ReplayInstant mode mark us as in replay state until resolved.
@@ -1027,26 +1045,41 @@ func (o *consumer) updateDeliveryInterest(localInterest bool) bool {
 }
 
 func (o *consumer) deleteNotActive() {
-	// If we have local interest simply return.
-	if o.hasLocalInterest() {
-		return
-	}
-	o.mu.RLock()
+	o.mu.Lock()
 	if o.mset == nil {
-		o.mu.RUnlock()
+		o.mu.Unlock()
 		return
 	}
+	// Push mode just look at active.
+	if o.isPushMode() {
+		// If we are active simply return.
+		if o.active {
+			o.mu.Unlock()
+			return
+		}
+	} else {
+		// These need to keep firing so reset first.
+		if o.dtmr != nil {
+			o.dtmr.Reset(o.dthresh)
+		}
+		// Check if we have had a request lately, or if we still have valid requests waiting.
+		if time.Since(o.waiting.last) <= o.dthresh || o.checkWaitingForInterest() {
+			o.mu.Unlock()
+			return
+		}
+	}
+
 	s, js := o.mset.srv, o.mset.srv.js
-	acc, stream, name := o.acc.Name, o.stream, o.name
-	o.mu.RUnlock()
+	acc, stream, name, isDirect := o.acc.Name, o.stream, o.name, o.cfg.Direct
+	o.mu.Unlock()
 
 	// If we are clustered, check if we still have this consumer assigned.
 	// If we do forward a proposal to delete ourselves to the metacontroller leader.
-	if s.JetStreamIsClustered() {
+	if !isDirect && s.JetStreamIsClustered() {
 		js.mu.RLock()
-		ca := js.consumerAssignment(acc, stream, name)
-		cc := js.cluster
+		ca, cc := js.consumerAssignment(acc, stream, name), js.cluster
 		js.mu.RUnlock()
+
 		if ca != nil && cc != nil {
 			cca := *ca
 			cca.Reply = _EMPTY_
@@ -1663,8 +1696,8 @@ func (o *consumer) info() *ConsumerInfo {
 
 	ci := js.clusterInfo(o.raftGroup())
 
-	o.mu.RLock()
-	defer o.mu.RUnlock()
+	o.mu.Lock()
+	defer o.mu.Unlock()
 
 	cfg := o.cfg
 	info := &ConsumerInfo{
@@ -1698,6 +1731,7 @@ func (o *consumer) info() *ConsumerInfo {
 
 	// If we are a pull mode consumer, report on number of waiting requests.
 	if o.isPullMode() {
+		o.expireWaiting()
 		info.NumWaiting = o.waiting.len()
 	}
 	return info
@@ -1932,16 +1966,42 @@ func nextReqFromMsg(msg []byte) (time.Time, int, bool, error) {
 
 // Represents a request that is on the internal waiting queue
 type waitingRequest struct {
-	client  *client
-	reply   string
-	n       int // For batching
-	expires time.Time
-	noWait  bool
+	acc      *Account
+	interest string
+	reply    string
+	n        int // For batching
+	expires  time.Time
+	received time.Time
+	noWait   bool
+}
+
+// sync.Pool for waiting requests.
+var wrPool = sync.Pool{
+	New: func() interface{} {
+		return new(waitingRequest)
+	},
+}
+
+// Recycle this request. This request can not be accessed after this call.
+func (wr *waitingRequest) recycleIfDone() {
+	if wr != nil && wr.n <= 0 {
+		wr.acc, wr.interest, wr.reply = nil, _EMPTY_, _EMPTY_
+		wrPool.Put(wr)
+	}
+}
+
+// Force a recycle.
+func (wr *waitingRequest) recycle() {
+	if wr != nil {
+		wr.acc, wr.interest, wr.reply = nil, _EMPTY_, _EMPTY_
+		wrPool.Put(wr)
+	}
 }
 
 // waiting queue for requests that are waiting for new messages to arrive.
 type waitQueue struct {
 	rp, wp int
+	last   time.Time
 	reqs   []*waitingRequest
 }
 
@@ -1956,14 +2016,14 @@ var (
 )
 
 // Adds in a new request.
-func (wq *waitQueue) add(req *waitingRequest) error {
+func (wq *waitQueue) add(wr *waitingRequest) error {
 	if wq == nil {
 		return errWaitQueueNil
 	}
 	if wq.isFull() {
 		return errWaitQueueFull
 	}
-	wq.reqs[wq.wp] = req
+	wq.reqs[wq.wp] = wr
 	// TODO(dlc) - Could make pow2 and get rid of mod.
 	wq.wp = (wq.wp + 1) % cap(wq.reqs)
 
@@ -1971,7 +2031,8 @@ func (wq *waitQueue) add(req *waitingRequest) error {
 	if wq.rp < 0 {
 		wq.rp = 0
 	}
-
+	// Track last active via when we receive a request.
+	wq.last = wr.received
 	return nil
 }
 
@@ -1979,6 +2040,10 @@ func (wq *waitQueue) isFull() bool {
 	return wq.rp == wq.wp
 }
 
+func (wq *waitQueue) isEmpty() bool {
+	return wq.len() == 0
+}
+
 func (wq *waitQueue) len() int {
 	if wq == nil || wq.rp < 0 {
 		return 0
@@ -2007,17 +2072,52 @@ func (wq *waitQueue) pop() *waitingRequest {
 	if wr != nil {
 		wr.n--
 		if wr.n <= 0 {
-			wq.reqs[wq.rp] = nil
-			wq.rp = (wq.rp + 1) % cap(wq.reqs)
-			// Check if we are empty.
-			if wq.rp == wq.wp {
-				wq.rp, wq.wp = -1, 0
-			}
+			wq.removeCurrent()
 		}
 	}
 	return wr
 }
 
+// removes the current read pointer (head FIFO) entry.
+func (wq *waitQueue) removeCurrent() {
+	if wq.rp < 0 {
+		return
+	}
+	wq.reqs[wq.rp] = nil
+	wq.rp = (wq.rp + 1) % cap(wq.reqs)
+	// Check if we are empty.
+	if wq.rp == wq.wp {
+		wq.rp, wq.wp = -1, 0
+	}
+}
+
+// Return next waiting request. This will check for expirations but not noWait or interest.
+// That will be handled by expireWaiting.
+// Lock should be held.
+func (o *consumer) nextWaiting() *waitingRequest {
+	if o.waiting == nil || o.waiting.isEmpty() {
+		return nil
+	}
+	for wr := o.waiting.peek(); !o.waiting.isEmpty(); wr = o.waiting.peek() {
+		if wr == nil || wr.expires.IsZero() || time.Now().Before(wr.expires) {
+			rr := wr.acc.sl.Match(wr.interest)
+			if len(rr.psubs)+len(rr.qsubs) > 0 {
+				return o.waiting.pop()
+			} else if o.srv.gateway.enabled {
+				if o.srv.hasGatewayInterest(wr.acc.Name, wr.interest) || time.Since(wr.received) < defaultGatewayRecentSubExpiration {
+					return o.waiting.pop()
+				}
+			}
+		}
+		hdr := []byte("NATS/1.0 408 Request Timeout\r\n\r\n")
+		o.outq.send(&jsPubMsg{wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0, nil})
+		// Remove the current one, no longer valid.
+		o.waiting.removeCurrent()
+		wr.recycle()
+	}
+	return nil
+}
+
 // 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.
@@ -2025,13 +2125,12 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 	if reply == _EMPTY_ {
 		return
 	}
-
 	_, msg = c.msgParts(msg)
 
 	o.mu.Lock()
 	defer o.mu.Unlock()
 
-	s, mset, js := o.srv, o.mset, o.js
+	mset := o.mset
 	if mset == nil {
 		return
 	}
@@ -2046,14 +2145,6 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 		return
 	}
 
-	if o.waiting.isFull() {
-		// Try to expire some of the requests.
-		if expired := o.expireWaiting(); expired == 0 {
-			// Force expiration if needed.
-			o.forceExpireFirstWaiting()
-		}
-	}
-
 	// Check payload here to see if they sent in batch size or a formal request.
 	expires, batchSize, noWait, err := nextReqFromMsg(msg)
 	if err != nil {
@@ -2061,63 +2152,52 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 		return
 	}
 
-	if o.maxp > 0 && batchSize > o.maxp {
-		sendErr(409, "Exceeded MaxAckPending")
+	// Check for request limits
+	if o.cfg.MaxRequestBatch > 0 && batchSize > o.cfg.MaxRequestBatch {
+		sendErr(409, fmt.Sprintf("Exceeded MaxRequestBatch of %d", o.cfg.MaxRequestBatch))
 		return
 	}
 
-	// In case we have to queue up this request.
-	wr := waitingRequest{client: c, reply: reply, n: batchSize, noWait: noWait, expires: expires}
-
-	// If we are in replay mode, defer to processReplay for delivery.
-	if o.replay {
-		o.waiting.add(&wr)
-		o.signalNewMessages()
+	if !expires.IsZero() && o.cfg.MaxRequestExpires > 0 && expires.After(time.Now().Add(o.cfg.MaxRequestExpires)) {
+		sendErr(409, fmt.Sprintf("Exceeded MaxRequestExpires of %v", o.cfg.MaxRequestExpires))
 		return
 	}
 
-	sendBatch := func(wr *waitingRequest) {
-		for i, batchSize := 0, wr.n; 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")
-					}
-				} else {
-					o.waiting.add(wr)
-				}
-				return
-			}
+	// If we have the max number of requests already pending try to expire.
+	if o.waiting.isFull() {
+		// Try to expire some of the requests.
+		if expired := o.expireWaiting(); expired == 0 {
+			// Force expiration if needed.
+			o.forceExpireFirstWaiting()
 		}
 	}
 
-	// If this is direct from a client can proceed inline.
-	if c.kind == CLIENT {
-		sendBatch(&wr)
-	} else {
-		// Check for API outstanding requests.
-		if apiOut := atomic.AddInt64(&js.apiInflight, 1); apiOut > maxJSApiOut {
-			atomic.AddInt64(&js.apiInflight, -1)
-			sendErr(503, "JetStream API limit exceeded")
-			s.Warnf("JetStream API limit exceeded: %d calls outstanding", apiOut)
-			return
+	// If the request is for noWait and we have pending requests already, error.
+	if noWait && o.checkWaitingForInterest() {
+		sendErr(408, "Requests Pending")
+		return
+	}
+
+	// If we receive this request though an account export, we need to track that interest subject and account.
+	acc, interest := o.acc, reply
+	for strings.HasPrefix(interest, replyPrefix) && acc.exports.responses != nil {
+		if si := acc.exports.responses[interest]; si != nil {
+			acc, interest = si.acc, si.to
+		} else {
+			break
 		}
-		// Dispatch the API call to its own Go routine.
-		go func() {
-			o.mu.Lock()
-			sendBatch(&wr)
-			o.mu.Unlock()
-			atomic.AddInt64(&js.apiInflight, -1)
-		}()
 	}
+
+	// In case we have to queue up this request.
+	wr := wrPool.Get().(*waitingRequest)
+	wr.acc, wr.interest, wr.reply, wr.n, wr.noWait, wr.expires = acc, interest, reply, batchSize, noWait, expires
+	wr.received = time.Now()
+
+	if err := o.waiting.add(wr); err != nil {
+		sendErr(409, "Exceeded MaxWaiting")
+		return
+	}
+	o.signalNewMessages()
 }
 
 // Increase the delivery count for this message.
@@ -2235,19 +2315,20 @@ func (o *consumer) getNextMsg() (subj string, hdr, msg []byte, seq uint64, dc ui
 
 // forceExpireFirstWaiting will force expire the first waiting.
 // Lock should be held.
-func (o *consumer) forceExpireFirstWaiting() *waitingRequest {
+func (o *consumer) forceExpireFirstWaiting() {
 	// FIXME(dlc) - Should we do advisory here as well?
-	wr := o.waiting.pop()
+	wr := o.waiting.peek()
 	if wr == nil {
-		return wr
+		return
 	}
 	// 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 {
+	if rr := wr.acc.sl.Match(wr.interest); len(rr.psubs)+len(rr.qsubs) > 0 && o.mset != nil {
 		// We still appear to have interest, so send alert as courtesy.
 		hdr := []byte("NATS/1.0 408 Request Timeout\r\n\r\n")
 		o.outq.send(&jsPubMsg{wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0, nil})
 	}
-	return wr
+	o.waiting.removeCurrent()
+	wr.recycle()
 }
 
 // Will check for expiration and lack of interest on waiting requests.
@@ -2255,24 +2336,38 @@ func (o *consumer) expireWaiting() int {
 	var expired int
 	now := time.Now()
 	for wr := o.waiting.peek(); wr != nil; wr = o.waiting.peek() {
-		if !wr.expires.IsZero() && now.After(wr.expires) {
-			wr.n = 0 // Force removal by setting requests left to 0.
-			o.waiting.pop()
+		if wr.noWait || !wr.expires.IsZero() && now.After(wr.expires) {
+			var hdr []byte
+			if wr.noWait {
+				hdr = []byte("NATS/1.0 404 No Messages\r\n\r\n")
+			} else {
+				hdr = []byte("NATS/1.0 408 Request Timeout\r\n\r\n")
+			}
+			o.outq.send(&jsPubMsg{wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0, nil})
+			o.waiting.removeCurrent()
+			wr.recycle()
 			expired++
 			continue
 		}
-		s, acc := o.acc.srv, o.acc
-		rr := acc.sl.Match(wr.reply)
-		if len(rr.psubs)+len(rr.qsubs) > 0 {
+
+		s, acc := o.srv, wr.acc
+		rr := acc.sl.Match(wr.interest)
+		// If we have local interest or no server break and do not remove.
+		if len(rr.psubs)+len(rr.qsubs) > 0 || s == nil {
 			break
 		}
-		// If we are here check on gateways.
-		if s != nil && s.hasGatewayInterest(acc.Name, wr.reply) {
-			break
+
+		// Check gateway interest.
+		if s.gateway.enabled {
+			// If we are here check on gateways.
+			// If we have interest or the request is too young break and do not expire.
+			if s.hasGatewayInterest(acc.Name, wr.interest) || time.Since(wr.received) < defaultGatewayRecentSubExpiration {
+				break
+			}
 		}
-		// No more interest so go ahead and remove this one from our list.
-		wr.n = 0 // Force removal by setting requests left to 0.
-		o.waiting.pop()
+		// No more interest here so go ahead and remove this one from our list.
+		o.waiting.removeCurrent()
+		wr.recycle()
 		expired++
 	}
 	return expired
@@ -2340,17 +2435,11 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 
 		// If we are in push mode and not active or under flowcontrol let's stop sending.
 		if o.isPushMode() {
-			if !o.active {
+			if !o.active || (o.maxpb > 0 && o.pbytes > o.maxpb) {
 				goto waitForMsgs
 			}
-			// Flowcontrol.
-			if o.maxpb > 0 && o.pbytes > o.maxpb {
-				goto waitForMsgs
-			}
-		}
-
-		// If we are in pull mode and no one is waiting already break and wait.
-		if o.isPullMode() && !o.checkWaitingForInterest() {
+		} else if o.waiting.isEmpty() {
+			// If we are in pull mode and no one is waiting already break and wait.
 			goto waitForMsgs
 		}
 
@@ -2361,16 +2450,20 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			if err == ErrStoreMsgNotFound || err == ErrStoreEOF || err == errMaxAckPending || err == errPartialCache {
 				goto waitForMsgs
 			} else {
-				o.mu.Unlock()
 				s.Errorf("Received an error looking up message for consumer: %v", err)
-				return
+				goto waitForMsgs
 			}
 		}
 
-		if wr := o.waiting.pop(); wr != nil {
+		if o.isPushMode() {
+			dsubj = o.dsubj
+		} else if wr := o.nextWaiting(); wr != nil {
 			dsubj = wr.reply
+			wr.recycleIfDone()
 		} else {
-			dsubj = o.dsubj
+			// We will redo this one.
+			o.sseq--
+			goto waitForMsgs
 		}
 
 		// If we are in a replay scenario and have not caught up check if we need to delay here.
@@ -2422,6 +2515,11 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			o.replay = false
 		}
 
+		// Make sure to process any expired requests that are pending.
+		if o.isPullMode() {
+			o.expireWaiting()
+		}
+
 		// We will wait here for new messages to arrive.
 		mch, outq, odsubj := o.mch, o.outq, o.cfg.DeliverSubject
 		o.mu.Unlock()
@@ -3019,11 +3117,6 @@ func (o *consumer) isActive() bool {
 	return active
 }
 
-// hasLocalInterest returns if we have local interest.
-func (o *consumer) hasLocalInterest() bool {
-	return !o.hasNoLocalInterest()
-}
-
 // hasNoLocalInterest return true if we have no local interest.
 func (o *consumer) hasNoLocalInterest() bool {
 	o.mu.RLock()
@@ -3255,11 +3348,8 @@ func (o *consumer) setInActiveDeleteThreshold(dthresh time.Duration) error {
 	o.mu.Lock()
 	defer o.mu.Unlock()
 
-	if o.isPullMode() {
-		return fmt.Errorf("consumer is not push-based")
-	}
 	if o.isDurable() {
-		return fmt.Errorf("consumer is not durable")
+		return fmt.Errorf("consumer is not ephemeral")
 	}
 	deleteWasRunning := o.dtmr != nil
 	stopAndClearTimer(&o.dtmr)
@@ -3277,6 +3367,15 @@ func (o *consumer) switchToEphemeral() {
 	o.cfg.Durable = _EMPTY_
 	store, ok := o.store.(*consumerFileStore)
 	rr := o.acc.sl.Match(o.cfg.DeliverSubject)
+	// Setup dthresh.
+	if o.dthresh == 0 {
+		if o.cfg.InactiveThreshold != 0 {
+			o.dthresh = o.cfg.InactiveThreshold
+		} else {
+			// Add in 1 sec of jitter above and beyond the default of 5s.
+			o.dthresh = JsDeleteWaitTimeDefault + time.Duration(rand.Int63n(1000))*time.Millisecond
+		}
+	}
 	o.mu.Unlock()
 
 	// Update interest
@@ -3317,6 +3416,7 @@ func (o *consumer) setInitialPendingAndStart() {
 		mset.store.FastState(&state)
 		if state.Msgs > 0 {
 			o.sgap = state.Msgs - (o.sseq - state.FirstSeq)
+			o.lsgap = state.LastSeq
 		}
 	} else {
 		// Here we are filtered.
@@ -3324,8 +3424,10 @@ func (o *consumer) setInitialPendingAndStart() {
 			ss := mset.store.FilteredState(o.lss.resume+1, o.cfg.FilterSubject)
 			o.sseq = o.lss.seqs[0]
 			o.sgap = ss.Msgs + uint64(len(o.lss.seqs))
+			o.lsgap = ss.Last
 		} else if ss := mset.store.FilteredState(o.sseq, o.cfg.FilterSubject); ss.Msgs > 0 {
 			o.sgap = ss.Msgs
+			o.lsgap = ss.Last
 			// See if we should update our starting sequence.
 			if dp == DeliverLast || dp == DeliverLastPerSubject {
 				o.sseq = ss.Last
diff --git a/server/errors.json b/server/errors.json
index a9e74658..7f114f99 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -823,7 +823,7 @@
     "constant": "JSConsumerPullRequiresAckErr",
     "code": 400,
     "error_code": 10084,
-    "description": "consumer in pull mode requires explicit ack policy",
+    "description": "consumer in pull mode requires ack policy",
     "comment": "",
     "help": "",
     "url": "",
@@ -1118,5 +1118,25 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSConsumerMaxRequestBatchNegativeErr",
+    "code": 400,
+    "error_code": 10114,
+    "description": "consumer max request batch needs to be \u003e 0",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
+  },
+  {
+    "constant": "JSConsumerMaxRequestExpiresToSmall",
+    "code": 400,
+    "error_code": 10115,
+    "description": "consumer max request expires needs to be \u003e= 1ms",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]
\ No newline at end of file
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index f078e231..1616d1fb 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -4496,14 +4496,7 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 
 	// Also short circuit if DeliverLastPerSubject is set with no FilterSubject.
 	if cfg.DeliverPolicy == DeliverLastPerSubject {
-		badConfig := cfg.FilterSubject == _EMPTY_
-		if !badConfig {
-			subjects := sa.Config.Subjects
-			if len(subjects) == 1 && subjects[0] == cfg.FilterSubject && subjectIsLiteral(subjects[0]) {
-				badConfig = true
-			}
-		}
-		if badConfig {
+		if cfg.FilterSubject == _EMPTY_ {
 			resp.Error = NewJSConsumerInvalidPolicyError(fmt.Errorf("consumer delivery policy is deliver last per subject, but FilterSubject is not set"))
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
 			return
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index 2b105f28..654b0624 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -107,6 +107,12 @@ const (
 	// JSConsumerMaxPendingAckPolicyRequiredErr consumer requires ack policy for max ack pending
 	JSConsumerMaxPendingAckPolicyRequiredErr ErrorIdentifier = 10082
 
+	// JSConsumerMaxRequestBatchNegativeErr consumer max request batch needs to be > 0
+	JSConsumerMaxRequestBatchNegativeErr ErrorIdentifier = 10114
+
+	// JSConsumerMaxRequestExpiresToSmall consumer max request expires needs to be >= 1ms
+	JSConsumerMaxRequestExpiresToSmall ErrorIdentifier = 10115
+
 	// JSConsumerMaxWaitingNegativeErr consumer max waiting needs to be positive
 	JSConsumerMaxWaitingNegativeErr ErrorIdentifier = 10087
 
@@ -125,7 +131,7 @@ const (
 	// JSConsumerPullNotDurableErr consumer in pull mode requires a durable name
 	JSConsumerPullNotDurableErr ErrorIdentifier = 10085
 
-	// JSConsumerPullRequiresAckErr consumer in pull mode requires explicit ack policy
+	// JSConsumerPullRequiresAckErr consumer in pull mode requires ack policy
 	JSConsumerPullRequiresAckErr ErrorIdentifier = 10084
 
 	// JSConsumerPullWithRateLimitErr consumer in pull mode can not have rate limit set
@@ -378,13 +384,15 @@ var (
 		JSConsumerInvalidPolicyErrF:                {Code: 400, ErrCode: 10094, Description: "{err}"},
 		JSConsumerInvalidSamplingErrF:              {Code: 400, ErrCode: 10095, Description: "failed to parse consumer sampling configuration: {err}"},
 		JSConsumerMaxPendingAckPolicyRequiredErr:   {Code: 400, ErrCode: 10082, Description: "consumer requires ack policy for max ack pending"},
+		JSConsumerMaxRequestBatchNegativeErr:       {Code: 400, ErrCode: 10114, Description: "consumer max request batch needs to be > 0"},
+		JSConsumerMaxRequestExpiresToSmall:         {Code: 400, ErrCode: 10115, Description: "consumer max request expires needs to be >= 1ms"},
 		JSConsumerMaxWaitingNegativeErr:            {Code: 400, ErrCode: 10087, Description: "consumer max waiting needs to be positive"},
 		JSConsumerNameExistErr:                     {Code: 400, ErrCode: 10013, Description: "consumer name already in use"},
 		JSConsumerNameTooLongErrF:                  {Code: 400, ErrCode: 10102, Description: "consumer name is too long, maximum allowed is {max}"},
 		JSConsumerNotFoundErr:                      {Code: 404, ErrCode: 10014, Description: "consumer not found"},
 		JSConsumerOnMappedErr:                      {Code: 400, ErrCode: 10092, Description: "consumer direct on a mapped consumer"},
 		JSConsumerPullNotDurableErr:                {Code: 400, ErrCode: 10085, Description: "consumer in pull mode requires a durable name"},
-		JSConsumerPullRequiresAckErr:               {Code: 400, ErrCode: 10084, Description: "consumer in pull mode requires explicit ack policy"},
+		JSConsumerPullRequiresAckErr:               {Code: 400, ErrCode: 10084, Description: "consumer in pull mode requires ack policy"},
 		JSConsumerPullWithRateLimitErr:             {Code: 400, ErrCode: 10086, Description: "consumer in pull mode can not have rate limit set"},
 		JSConsumerPushMaxWaitingErr:                {Code: 400, ErrCode: 10080, Description: "consumer in push mode can not set max waiting"},
 		JSConsumerReplacementWithDifferentNameErr:  {Code: 400, ErrCode: 10106, Description: "consumer replacement durable config not the same"},
@@ -845,6 +853,26 @@ func NewJSConsumerMaxPendingAckPolicyRequiredError(opts ...ErrorOption) *ApiErro
 	return ApiErrors[JSConsumerMaxPendingAckPolicyRequiredErr]
 }
 
+// NewJSConsumerMaxRequestBatchNegativeError creates a new JSConsumerMaxRequestBatchNegativeErr error: "consumer max request batch needs to be > 0"
+func NewJSConsumerMaxRequestBatchNegativeError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSConsumerMaxRequestBatchNegativeErr]
+}
+
+// NewJSConsumerMaxRequestExpiresToSmallError creates a new JSConsumerMaxRequestExpiresToSmall error: "consumer max request expires needs to be >= 1ms"
+func NewJSConsumerMaxRequestExpiresToSmallError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSConsumerMaxRequestExpiresToSmall]
+}
+
 // NewJSConsumerMaxWaitingNegativeError creates a new JSConsumerMaxWaitingNegativeErr error: "consumer max waiting needs to be positive"
 func NewJSConsumerMaxWaitingNegativeError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
@@ -911,7 +939,7 @@ func NewJSConsumerPullNotDurableError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSConsumerPullNotDurableErr]
 }
 
-// NewJSConsumerPullRequiresAckError creates a new JSConsumerPullRequiresAckErr error: "consumer in pull mode requires explicit ack policy"
+// NewJSConsumerPullRequiresAckError creates a new JSConsumerPullRequiresAckErr error: "consumer in pull mode requires ack policy"
 func NewJSConsumerPullRequiresAckError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
 	if ae, ok := eopts.err.(*ApiError); ok {
diff --git a/server/stream.go b/server/stream.go
index a1cc58ec..0e302133 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -3070,7 +3070,7 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 		for _, o := range mset.consumers {
 			o.mu.Lock()
 			if o.isLeader() {
-				if o.isFilteredMatch(subject) {
+				if seq > o.lsgap && o.isFilteredMatch(subject) {
 					o.sgap++
 				}
 				o.signalNewMessages()
diff --git a/vendor/github.com/minio/highwayhash/go.mod b/vendor/github.com/minio/highwayhash/go.mod
deleted file mode 100644
index 3015eba9..00000000
--- a/vendor/github.com/minio/highwayhash/go.mod
+++ /dev/null
@@ -1,5 +0,0 @@
-module github.com/minio/highwayhash
-
-go 1.15
-
-require golang.org/x/sys v0.0.0-20190130150945-aca44879d564
diff --git a/vendor/github.com/minio/highwayhash/go.sum b/vendor/github.com/minio/highwayhash/go.sum
deleted file mode 100644
index b45a8e12..00000000
--- a/vendor/github.com/minio/highwayhash/go.sum
+++ /dev/null
@@ -1,2 +0,0 @@
-golang.org/x/sys v0.0.0-20190130150945-aca44879d564 h1:o6ENHFwwr1TZ9CUPQcfo1HGvLP1OPsPOTB7xCIOPNmU=
-golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
diff --git a/vendor/github.com/nats-io/jwt/v2/go.mod b/vendor/github.com/nats-io/jwt/v2/go.mod
deleted file mode 100644
index 56a8ddb6..00000000
--- a/vendor/github.com/nats-io/jwt/v2/go.mod
+++ /dev/null
@@ -1,5 +0,0 @@
-module github.com/nats-io/jwt/v2
-
-go 1.16
-
-require github.com/nats-io/nkeys v0.3.0
diff --git a/vendor/github.com/nats-io/jwt/v2/go.sum b/vendor/github.com/nats-io/jwt/v2/go.sum
deleted file mode 100644
index e2eb2522..00000000
--- a/vendor/github.com/nats-io/jwt/v2/go.sum
+++ /dev/null
@@ -1,9 +0,0 @@
-github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
-github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
-golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b h1:wSOdpTq0/eI46Ez/LkDwIsAKA71YP2SRKBODiRWM0as=
-golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
diff --git a/vendor/github.com/nats-io/nats.go/go.mod b/vendor/github.com/nats-io/nats.go/go.mod
deleted file mode 100644
index 63faedf4..00000000
--- a/vendor/github.com/nats-io/nats.go/go.mod
+++ /dev/null
@@ -1,8 +0,0 @@
-module github.com/nats-io/nats.go
-
-go 1.16
-
-require (
-	github.com/nats-io/nkeys v0.3.0
-	github.com/nats-io/nuid v1.0.1
-)
diff --git a/vendor/github.com/nats-io/nats.go/go.sum b/vendor/github.com/nats-io/nats.go/go.sum
deleted file mode 100644
index 2138ffb5..00000000
--- a/vendor/github.com/nats-io/nats.go/go.sum
+++ /dev/null
@@ -1,11 +0,0 @@
-github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
-github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
-github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
-github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
-golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b h1:wSOdpTq0/eI46Ez/LkDwIsAKA71YP2SRKBODiRWM0as=
-golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
diff --git a/vendor/github.com/nats-io/nkeys/go.mod b/vendor/github.com/nats-io/nkeys/go.mod
deleted file mode 100644
index 2384db3f..00000000
--- a/vendor/github.com/nats-io/nkeys/go.mod
+++ /dev/null
@@ -1,5 +0,0 @@
-module github.com/nats-io/nkeys
-
-go 1.16
-
-require golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b
diff --git a/vendor/github.com/nats-io/nkeys/go.sum b/vendor/github.com/nats-io/nkeys/go.sum
deleted file mode 100644
index 42fec3ae..00000000
--- a/vendor/github.com/nats-io/nkeys/go.sum
+++ /dev/null
@@ -1,7 +0,0 @@
-golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b h1:wSOdpTq0/eI46Ez/LkDwIsAKA71YP2SRKBODiRWM0as=
-golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
-golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 5a208311..6cadcbda 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1,27 +1,27 @@
 # github.com/golang/protobuf v1.4.2
-## explicit
+## explicit; go 1.9
 # github.com/klauspost/compress v1.13.4
-## explicit
+## explicit; go 1.13
 github.com/klauspost/compress/s2
 # github.com/minio/highwayhash v1.0.1
-## explicit
+## explicit; go 1.15
 github.com/minio/highwayhash
 # github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296
-## explicit
+## explicit; go 1.16
 github.com/nats-io/jwt/v2
 # github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc
-## explicit
+## explicit; go 1.16
 github.com/nats-io/nats.go
 github.com/nats-io/nats.go/encoders/builtin
 github.com/nats-io/nats.go/util
 # github.com/nats-io/nkeys v0.3.0
-## explicit
+## explicit; go 1.16
 github.com/nats-io/nkeys
 # github.com/nats-io/nuid v1.0.1
 ## explicit
 github.com/nats-io/nuid
-# golang.org/x/crypto v0.0.0-20211202192323-5770296d904e
-## explicit
+# golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3
+## explicit; go 1.17
 golang.org/x/crypto/bcrypt
 golang.org/x/crypto/blowfish
 golang.org/x/crypto/chacha20
@@ -32,7 +32,7 @@ golang.org/x/crypto/internal/poly1305
 golang.org/x/crypto/internal/subtle
 golang.org/x/crypto/ocsp
 # golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1
-## explicit
+## explicit; go 1.17
 golang.org/x/sys/cpu
 golang.org/x/sys/internal/unsafeheader
 golang.org/x/sys/unix


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 1074b6ae..22f782b9 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -7232,7 +7232,7 @@ func TestJetStreamClusterSuperClusterPullConsumerAndHeaders(t *testing.T) {
 		return nil
 	})
 
-	// Now create a pull consumers for the sourced stream.
+	// Now create a pull consumer for the sourced stream.
 	_, err = js2.AddConsumer("S", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -9899,6 +9899,156 @@ func TestJetStreamClusterBalancedPlacement(t *testing.T) {
 	require_Error(t, err, NewJSInsufficientResourcesError(), NewJSStorageResourcesExceededError())
 }
 
+func TestJetStreamClusterConsumerPendingBug(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "JSC", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	nc2, js2 := jsClientConnect(t, c.randomServer())
+	defer nc2.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "foo", Replicas: 3})
+	require_NoError(t, err)
+
+	startCh, doneCh := make(chan bool), make(chan error)
+	go func() {
+		<-startCh
+		_, err := js2.AddConsumer("foo", &nats.ConsumerConfig{
+			Durable:        "dlc",
+			FilterSubject:  "foo",
+			DeliverSubject: "x",
+		})
+		doneCh <- err
+	}()
+
+	n := 10_000
+	for i := 0; i < n; i++ {
+		nc.Publish("foo", []byte("ok"))
+		if i == 222 {
+			startCh <- true
+		}
+	}
+	// Wait for them to all be there.
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		si, err := js.StreamInfo("foo")
+		require_NoError(t, err)
+		if si.State.Msgs != uint64(n) {
+			return fmt.Errorf("Not received all messages")
+		}
+		return nil
+	})
+
+	select {
+	case err := <-doneCh:
+		if err != nil {
+			t.Fatalf("Error creating consumer: %v", err)
+		}
+		ci, err := js.ConsumerInfo("foo", "dlc")
+		require_NoError(t, err)
+		if ci.NumPending != uint64(n) {
+			t.Fatalf("Expected NumPending to be %d, got %d", n, ci.NumPending)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Timed out?")
+	}
+}
+
+func TestJetStreamClusterPullPerf(t *testing.T) {
+	skip(t)
+
+	c := createJetStreamClusterExplicit(t, "JSC", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	js.AddStream(&nats.StreamConfig{Name: "f22"})
+	defer js.DeleteStream("f22")
+
+	n, msg := 1_000_000, []byte(strings.Repeat("A", 1000))
+	for i := 0; i < n; i++ {
+		js.PublishAsync("f22", msg)
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(10 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	si, err := js.StreamInfo("f22")
+	require_NoError(t, err)
+
+	fmt.Printf("msgs: %d, total_bytes: %v\n", si.State.Msgs, friendlyBytes(int64(si.State.Bytes)))
+
+	// OrderedConsumer - fastest push based.
+	start := time.Now()
+	received, done := 0, make(chan bool)
+	_, err = js.Subscribe("f22", func(m *nats.Msg) {
+		received++
+		if received >= n {
+			done <- true
+		}
+	}, nats.OrderedConsumer())
+	require_NoError(t, err)
+
+	// Wait to receive all messages.
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatalf("Did not receive all of our messages")
+	}
+
+	tt := time.Since(start)
+	fmt.Printf("Took %v to receive %d msgs\n", tt, n)
+	fmt.Printf("%.0f msgs/s\n", float64(n)/tt.Seconds())
+	fmt.Printf("%.0f mb/s\n\n", float64(si.State.Bytes/(1024*1024))/tt.Seconds())
+
+	// Now do pull based, this is custom for now.
+	// Current nats.PullSubscribe maxes at about 1/2 the performance even with large batches.
+	_, err = js.AddConsumer("f22", &nats.ConsumerConfig{
+		Durable:       "dlc",
+		AckPolicy:     nats.AckAllPolicy,
+		MaxAckPending: 1000,
+	})
+	require_NoError(t, err)
+
+	r := 0
+	_, err = nc.Subscribe("xx", func(m *nats.Msg) {
+		r++
+		if r >= n {
+			done <- true
+		}
+		if r%750 == 0 {
+			m.Ack()
+		}
+	})
+	require_NoError(t, err)
+
+	// Simulate an non-ending request.
+	req := &JSApiConsumerGetNextRequest{Batch: n, Expires: 60 * time.Second}
+	jreq, err := json.Marshal(req)
+	require_NoError(t, err)
+
+	start = time.Now()
+	rsubj := fmt.Sprintf(JSApiRequestNextT, "f22", "dlc")
+	err = nc.PublishRequest(rsubj, "xx", jreq)
+	require_NoError(t, err)
+
+	// Wait to receive all messages.
+	select {
+	case <-done:
+	case <-time.After(60 * time.Second):
+		t.Fatalf("Did not receive all of our messages")
+	}
+
+	tt = time.Since(start)
+	fmt.Printf("Took %v to receive %d msgs\n", tt, n)
+	fmt.Printf("%.0f msgs/s\n", float64(n)/tt.Seconds())
+	fmt.Printf("%.0f mb/s\n\n", float64(si.State.Bytes/(1024*1024))/tt.Seconds())
+}
+
 // Support functions
 
 // Used to setup superclusters for tests.
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 133136b1..bed3d90e 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -1263,12 +1263,6 @@ func TestJetStreamCreateConsumer(t *testing.T) {
 
 			// Non-Durables need to have subscription to delivery subject.
 			delivery := nats.NewInbox()
-			// Pull-based consumers are required to be durable since we do not know when they should
-			// be cleaned up.
-			if _, err := mset.addConsumer(&ConsumerConfig{AckPolicy: AckExplicit}); err == nil {
-				t.Fatalf("Expected an error on pull-based that is non-durable.")
-			}
-
 			nc := clientConnectToServer(t, s)
 			defer nc.Close()
 			sub, _ := nc.SubscribeSync(delivery)
@@ -1818,7 +1812,7 @@ func TestJetStreamWorkQueueRequest(t *testing.T) {
 			req.NoWait = true
 			jreq, _ = json.Marshal(req)
 
-			resp, err := nc.Request(getSubj, jreq, 50*time.Millisecond)
+			resp, err := nc.Request(getSubj, jreq, 100*time.Millisecond)
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
@@ -1833,6 +1827,7 @@ func TestJetStreamWorkQueueRequest(t *testing.T) {
 			req.Batch = toSend + 10
 			req.NoWait = true
 			jreq, _ = json.Marshal(req)
+
 			nc.PublishRequest(getSubj, reply, jreq)
 			// We should now have 2 * toSend + the 404 message.
 			checkSubPending(2*toSend + 1)
@@ -1856,7 +1851,15 @@ func TestJetStreamWorkQueueRequest(t *testing.T) {
 			sendStreamMsg(t, nc, "foo", "Hello World!")
 			sendStreamMsg(t, nc, "bar", "Hello World!")
 			time.Sleep(100 * time.Millisecond)
-			checkSubPending(0)
+
+			// Expect the request timed out message.
+			checkSubPending(1)
+			if resp, _ = sub.NextMsg(time.Millisecond); resp == nil {
+				t.Fatalf("Expected an expired status message")
+			}
+			if status := resp.Header.Get("Status"); !strings.HasPrefix(status, "408") {
+				t.Fatalf("Expected status code of 408")
+			}
 
 			// Send a new request, we should not get the 408 because our previous request
 			// should have expired.
@@ -3557,6 +3560,7 @@ func TestJetStreamPullConsumerRemoveInterest(t *testing.T) {
 
 	nc = clientConnectToServer(t, s)
 	defer nc.Close()
+
 	// Send a message
 	sendStreamMsg(t, nc, mname, "Hello World!")
 
@@ -4519,6 +4523,10 @@ func TestJetStreamPubAckPerf(t *testing.T) {
 	s := RunBasicJetStreamServer()
 	defer s.Shutdown()
 
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
 	nc, js := jsClientConnect(t, s)
 	defer nc.Close()
 
@@ -9278,23 +9286,6 @@ func TestJetStreamPushConsumersPullError(t *testing.T) {
 	if m.Header.Get("Status") != "409" {
 		t.Fatalf("Expected a 409 status code, got %q", m.Header.Get("Status"))
 	}
-
-	// Should not be possible to ask for more messages than MaxAckPending limit.
-	ci, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
-		Durable:       "test",
-		AckPolicy:     nats.AckExplicitPolicy,
-		MaxAckPending: 5,
-	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	m, err = nc.Request(fmt.Sprintf(JSApiRequestNextT, "TEST", ci.Name), []byte(`10`), 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"))
-	}
 }
 
 ////////////////////////////////////////
@@ -10293,16 +10284,6 @@ func TestJetStreamPullConsumerMaxAckPending(t *testing.T) {
 					toAck = append(toAck, m)
 				}
 			}
-			// This should fail.. But we do not want to queue up our request.
-			req := &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 {
@@ -10323,8 +10304,8 @@ func TestJetStreamPullConsumerMaxAckPending(t *testing.T) {
 				})
 			}
 
-			req = &JSApiConsumerGetNextRequest{Batch: maxAckPending}
-			jreq, _ = json.Marshal(req)
+			req := &JSApiConsumerGetNextRequest{Batch: maxAckPending}
+			jreq, _ := json.Marshal(req)
 			nc.PublishRequest(getSubj, sub.Subject, jreq)
 
 			checkSubPending(maxAckPending)
@@ -13861,6 +13842,486 @@ func TestJetStreamCrossAccountsDeliverSubjectInterest(t *testing.T) {
 	}
 }
 
+func TestJetStreamPullConsumerRequestCleanup(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "T", Storage: nats.MemoryStorage})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("T", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy})
+	require_NoError(t, err)
+
+	req := &JSApiConsumerGetNextRequest{Batch: 10, Expires: 100 * time.Millisecond}
+	jreq, err := json.Marshal(req)
+	require_NoError(t, err)
+
+	// Need interest otherwise the requests will be recycled based on that.
+	_, err = nc.SubscribeSync("xx")
+	require_NoError(t, err)
+
+	// Queue up 100 requests.
+	rsubj := fmt.Sprintf(JSApiRequestNextT, "T", "dlc")
+	for i := 0; i < 100; i++ {
+		err = nc.PublishRequest(rsubj, "xx", jreq)
+		require_NoError(t, err)
+	}
+	// Wait to expire
+	time.Sleep(200 * time.Millisecond)
+
+	ci, err := js.ConsumerInfo("T", "dlc")
+	require_NoError(t, err)
+
+	if ci.NumWaiting != 0 {
+		t.Fatalf("Expected to see no waiting requests, got %d", ci.NumWaiting)
+	}
+}
+
+func TestJetStreamPullConsumerRequestMaximums(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	nc, _ := jsClientConnect(t, s)
+	defer nc.Close()
+
+	// Need to do this via server for now.
+	acc := s.GlobalAccount()
+	mset, err := acc.addStream(&StreamConfig{Name: "TEST", Storage: MemoryStorage})
+	require_NoError(t, err)
+
+	_, err = mset.addConsumer(&ConsumerConfig{
+		Durable:           "dlc",
+		MaxRequestBatch:   10,
+		MaxRequestExpires: time.Second,
+		AckPolicy:         AckExplicit,
+	})
+	require_NoError(t, err)
+
+	genReq := func(b int, e time.Duration) []byte {
+		req := &JSApiConsumerGetNextRequest{Batch: b, Expires: e}
+		jreq, err := json.Marshal(req)
+		require_NoError(t, err)
+		return jreq
+	}
+
+	rsubj := fmt.Sprintf(JSApiRequestNextT, "TEST", "dlc")
+
+	// Exceeds max batch size.
+	resp, err := nc.Request(rsubj, genReq(11, 100*time.Millisecond), time.Second)
+	require_NoError(t, err)
+	if status := resp.Header.Get("Status"); status != "409" {
+		t.Fatalf("Expected a 409 status code, got %q", status)
+	}
+
+	// Exceeds max expires.
+	resp, err = nc.Request(rsubj, genReq(1, 10*time.Minute), time.Second)
+	require_NoError(t, err)
+	if status := resp.Header.Get("Status"); status != "409" {
+		t.Fatalf("Expected a 409 status code, got %q", status)
+	}
+}
+
+func TestJetStreamEphemeralPullConsumers(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "EC", Storage: nats.MemoryStorage})
+	require_NoError(t, err)
+
+	ci, err := js.AddConsumer("EC", &nats.ConsumerConfig{AckPolicy: nats.AckExplicitPolicy})
+	require_NoError(t, err)
+
+	mset, err := s.GlobalAccount().lookupStream("EC")
+	require_NoError(t, err)
+	o := mset.lookupConsumer(ci.Name)
+	if o == nil {
+		t.Fatalf("Error looking up consumer %q", ci.Name)
+	}
+	err = o.setInActiveDeleteThreshold(50 * time.Millisecond)
+	require_NoError(t, err)
+
+	time.Sleep(100 * time.Millisecond)
+	// Should no longer be around.
+	if o := mset.lookupConsumer(ci.Name); o != nil {
+		t.Fatalf("Expected consumer to be closed and removed")
+	}
+
+	// Make sure timer keeps firing etc. and does not delete until interest is gone.
+	ci, err = js.AddConsumer("EC", &nats.ConsumerConfig{AckPolicy: nats.AckExplicitPolicy})
+	require_NoError(t, err)
+	if o = mset.lookupConsumer(ci.Name); o == nil {
+		t.Fatalf("Error looking up consumer %q", ci.Name)
+	}
+	err = o.setInActiveDeleteThreshold(50 * time.Millisecond)
+	require_NoError(t, err)
+
+	// Need interest otherwise the requests will be recycled based on no real interest.
+	sub, err := nc.SubscribeSync("xx")
+	require_NoError(t, err)
+
+	req := &JSApiConsumerGetNextRequest{Batch: 10, Expires: 250 * time.Millisecond}
+	jreq, err := json.Marshal(req)
+	require_NoError(t, err)
+	rsubj := fmt.Sprintf(JSApiRequestNextT, "EC", ci.Name)
+	err = nc.PublishRequest(rsubj, "xx", jreq)
+	require_NoError(t, err)
+	nc.Flush()
+
+	time.Sleep(100 * time.Millisecond)
+	// Should still be alive here.
+	if o := mset.lookupConsumer(ci.Name); o == nil {
+		t.Fatalf("Expected consumer to still be active")
+	}
+	// Remove interest.
+	sub.Unsubscribe()
+	// Make sure this EPC goes away now.
+	checkFor(t, 5*time.Second, 10*time.Millisecond, func() error {
+		if o := mset.lookupConsumer(ci.Name); o != nil {
+			return fmt.Errorf("Consumer still present")
+		}
+		return nil
+	})
+}
+
+func TestJetStreamPullConsumerCrossAccountExpires(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: 127.0.0.1:-1
+		jetstream: {max_mem_store: 4GB, max_file_store: 1TB}
+		accounts: {
+			JS: {
+				jetstream: enabled
+				users: [ {user: dlc, password: foo} ]
+				exports [ { service: "$JS.API.CONSUMER.MSG.NEXT.>", response: stream } ]
+			},
+			IU: {
+				users: [ {user: mh, password: bar} ]
+				imports [ { service: { subject: "$JS.API.CONSUMER.MSG.NEXT.*.*", account: JS } }]
+				# Re-export for dasiy chain test.
+				exports [ { service: "$JS.API.CONSUMER.MSG.NEXT.>", response: stream } ]
+			},
+			IU2: {
+				users: [ {user: ik, password: bar} ]
+				imports [ { service: { subject: "$JS.API.CONSUMER.MSG.NEXT.*.*", account: IU } } ]
+			},
+		}
+	`))
+	defer removeFile(t, conf)
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	// Connect to JS account and create stream, put some messages into it.
+	nc, js := jsClientConnect(t, s, nats.UserInfo("dlc", "foo"))
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "PC", Subjects: []string{"foo"}})
+	require_NoError(t, err)
+
+	toSend := 50
+	for i := 0; i < toSend; i++ {
+		_, err := js.Publish("foo", []byte("OK"))
+		require_NoError(t, err)
+	}
+
+	// Now create pull consumer.
+	_, err = js.AddConsumer("PC", &nats.ConsumerConfig{Durable: "PC", AckPolicy: nats.AckExplicitPolicy})
+	require_NoError(t, err)
+
+	// Now access from the importing account.
+	nc2, _ := jsClientConnect(t, s, nats.UserInfo("mh", "bar"))
+	defer nc2.Close()
+
+	// Make sure batch request works properly with stream response.
+	req := &JSApiConsumerGetNextRequest{Batch: 10}
+	jreq, err := json.Marshal(req)
+	require_NoError(t, err)
+	rsubj := fmt.Sprintf(JSApiRequestNextT, "PC", "PC")
+	// Make sure we can get a batch correctly etc.
+	// This requires response stream above in the export definition.
+	sub, err := nc2.SubscribeSync("xx")
+	require_NoError(t, err)
+	err = nc2.PublishRequest(rsubj, "xx", jreq)
+	require_NoError(t, err)
+	checkSubsPending(t, sub, 10)
+
+	// Now let's queue up a bunch of requests and then delete interest to make sure the system
+	// removes those requests.
+
+	// Purge stream
+	err = js.PurgeStream("PC")
+	require_NoError(t, err)
+
+	// Queue up 10 requests
+	for i := 0; i < 10; i++ {
+		err = nc2.PublishRequest(rsubj, "xx", jreq)
+		require_NoError(t, err)
+	}
+	// Since using different connection, flush to make sure processed.
+	nc2.Flush()
+
+	ci, err := js.ConsumerInfo("PC", "PC")
+	require_NoError(t, err)
+	if ci.NumWaiting != 10 {
+		t.Fatalf("Expected to see 10 waiting requests, got %d", ci.NumWaiting)
+	}
+
+	// Now remove interest and make sure requests are removed.
+	sub.Unsubscribe()
+	checkFor(t, 5*time.Second, 10*time.Millisecond, func() error {
+		ci, err := js.ConsumerInfo("PC", "PC")
+		require_NoError(t, err)
+		if ci.NumWaiting != 0 {
+			return fmt.Errorf("Requests still present")
+		}
+		return nil
+	})
+
+	// Now let's test that ephemerals will go away as well when interest etc is no longer around.
+	ci, err = js.AddConsumer("PC", &nats.ConsumerConfig{AckPolicy: nats.AckExplicitPolicy})
+	require_NoError(t, err)
+
+	// Set the inactivity threshold by hand for now.
+	jsacc, err := s.LookupAccount("JS")
+	require_NoError(t, err)
+	mset, err := jsacc.lookupStream("PC")
+	require_NoError(t, err)
+	o := mset.lookupConsumer(ci.Name)
+	if o == nil {
+		t.Fatalf("Error looking up consumer %q", ci.Name)
+	}
+	err = o.setInActiveDeleteThreshold(50 * time.Millisecond)
+	require_NoError(t, err)
+
+	rsubj = fmt.Sprintf(JSApiRequestNextT, "PC", ci.Name)
+	sub, err = nc2.SubscribeSync("zz")
+	require_NoError(t, err)
+	err = nc2.PublishRequest(rsubj, "zz", jreq)
+	require_NoError(t, err)
+
+	// Wait past inactive threshold.
+	time.Sleep(100 * time.Millisecond)
+	// Make sure it is still there..
+	ci, err = js.ConsumerInfo("PC", ci.Name)
+	require_NoError(t, err)
+	if ci.NumWaiting != 1 {
+		t.Fatalf("Expected to see 1 waiting request, got %d", ci.NumWaiting)
+	}
+
+	// Now release interest.
+	sub.Unsubscribe()
+	checkFor(t, 5*time.Second, 10*time.Millisecond, func() error {
+		_, err := js.ConsumerInfo("PC", ci.Name)
+		if err == nil {
+			return fmt.Errorf("Consumer still present")
+		}
+		return nil
+	})
+
+	// Now test daisy chained.
+	toSend = 10
+	for i := 0; i < toSend; i++ {
+		_, err := js.Publish("foo", []byte("OK"))
+		require_NoError(t, err)
+	}
+
+	ci, err = js.AddConsumer("PC", &nats.ConsumerConfig{AckPolicy: nats.AckExplicitPolicy})
+	require_NoError(t, err)
+
+	// Set the inactivity threshold by hand for now.
+	o = mset.lookupConsumer(ci.Name)
+	if o == nil {
+		t.Fatalf("Error looking up consumer %q", ci.Name)
+	}
+	// Make this one longer so we test request purge and ephemerals in same test.
+	err = o.setInActiveDeleteThreshold(500 * time.Millisecond)
+	require_NoError(t, err)
+
+	// Now access from the importing account.
+	nc3, _ := jsClientConnect(t, s, nats.UserInfo("ik", "bar"))
+	defer nc3.Close()
+
+	sub, err = nc3.SubscribeSync("yy")
+	require_NoError(t, err)
+
+	rsubj = fmt.Sprintf(JSApiRequestNextT, "PC", ci.Name)
+	err = nc3.PublishRequest(rsubj, "yy", jreq)
+	require_NoError(t, err)
+	checkSubsPending(t, sub, 10)
+
+	// Purge stream
+	err = js.PurgeStream("PC")
+	require_NoError(t, err)
+
+	// Queue up 10 requests
+	for i := 0; i < 10; i++ {
+		err = nc3.PublishRequest(rsubj, "yy", jreq)
+		require_NoError(t, err)
+	}
+	// Since using different connection, flush to make sure processed.
+	nc3.Flush()
+
+	ci, err = js.ConsumerInfo("PC", ci.Name)
+	require_NoError(t, err)
+	if ci.NumWaiting != 10 {
+		t.Fatalf("Expected to see 10 waiting requests, got %d", ci.NumWaiting)
+	}
+
+	// Now remove interest and make sure requests are removed.
+	sub.Unsubscribe()
+	checkFor(t, 5*time.Second, 10*time.Millisecond, func() error {
+		ci, err := js.ConsumerInfo("PC", ci.Name)
+		require_NoError(t, err)
+		if ci.NumWaiting != 0 {
+			return fmt.Errorf("Requests still present")
+		}
+		return nil
+	})
+	// Now make sure the ephemeral goes away too.
+	checkFor(t, 5*time.Second, 10*time.Millisecond, func() error {
+		_, err := js.ConsumerInfo("PC", ci.Name)
+		if err == nil {
+			return fmt.Errorf("Consumer still present")
+		}
+		return nil
+	})
+}
+
+// This tests account export/import replies across a LN connection with account import/export
+// on both sides of the LN.
+func TestJetStreamPullConsumerCrossAccountsAndLeafNodes(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		server_name: SJS
+		listen: 127.0.0.1:-1
+		jetstream: {max_mem_store: 4GB, max_file_store: 1TB, domain: JSD }
+		accounts: {
+			JS: {
+				jetstream: enabled
+				users: [ {user: dlc, password: foo} ]
+				exports [ { service: "$JS.API.CONSUMER.MSG.NEXT.>", response: stream } ]
+			},
+			IU: {
+				users: [ {user: mh, password: bar} ]
+				imports [ { service: { subject: "$JS.API.CONSUMER.MSG.NEXT.*.*", account: JS } }]
+			},
+		}
+		leaf { listen: "127.0.0.1:-1" }
+	`))
+	defer removeFile(t, conf)
+
+	s, o := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	conf2 := createConfFile(t, []byte(fmt.Sprintf(`
+		server_name: SLN
+		listen: 127.0.0.1:-1
+		accounts: {
+			A: {
+				users: [ {user: l, password: p} ]
+				exports [ { service: "$JS.JSD.API.CONSUMER.MSG.NEXT.>", response: stream } ]
+			},
+			B: {
+				users: [ {user: m, password: p} ]
+				imports [ { service: { subject: "$JS.JSD.API.CONSUMER.MSG.NEXT.*.*", account: A } }]
+			},
+		}
+		# bind local A to IU account on other side of LN.
+		leaf { remotes [ { url: nats://mh:bar@127.0.0.1:%d; account: A } ] }
+	`, o.LeafNode.Port)))
+	defer removeFile(t, conf2)
+
+	s2, _ := RunServerWithConfig(conf2)
+	defer s2.Shutdown()
+
+	checkLeafNodeConnectedCount(t, s, 1)
+
+	// Connect to JS account, create stream and consumer and put in some messages.
+	nc, js := jsClientConnect(t, s, nats.UserInfo("dlc", "foo"))
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "PC", Subjects: []string{"foo"}})
+	require_NoError(t, err)
+
+	toSend := 10
+	for i := 0; i < toSend; i++ {
+		_, err := js.Publish("foo", []byte("OK"))
+		require_NoError(t, err)
+	}
+
+	// Now create durable pull consumer.
+	_, err = js.AddConsumer("PC", &nats.ConsumerConfig{Durable: "PC", AckPolicy: nats.AckExplicitPolicy})
+	require_NoError(t, err)
+
+	// Now access from the account on the leafnode, so importing on both sides and crossing a leafnode connection.
+	nc2, _ := jsClientConnect(t, s2, nats.UserInfo("m", "p"))
+	defer nc2.Close()
+
+	req := &JSApiConsumerGetNextRequest{Batch: toSend}
+	jreq, err := json.Marshal(req)
+	require_NoError(t, err)
+
+	// Make sure we can get a batch correctly etc.
+	// This requires response stream above in the export definition.
+	sub, err := nc2.SubscribeSync("xx")
+	require_NoError(t, err)
+
+	rsubj := "$JS.JSD.API.CONSUMER.MSG.NEXT.PC.PC"
+	err = nc2.PublishRequest(rsubj, "xx", jreq)
+	require_NoError(t, err)
+	checkSubsPending(t, sub, 10)
+
+	// Queue up a bunch of requests.
+	for i := 0; i < 10; i++ {
+		err = nc2.PublishRequest(rsubj, "xx", jreq)
+		require_NoError(t, err)
+	}
+	checkFor(t, 5*time.Second, 10*time.Millisecond, func() error {
+		ci, err := js.ConsumerInfo("PC", "PC")
+		require_NoError(t, err)
+		if ci.NumWaiting != 10 {
+			return fmt.Errorf("Expected to see 10 waiting requests, got %d", ci.NumWaiting)
+		}
+		return nil
+	})
+
+	// Remove interest.
+	sub.Unsubscribe()
+	// Make sure requests go away.
+	checkFor(t, 5*time.Second, 10*time.Millisecond, func() error {
+		ci, err := js.ConsumerInfo("PC", "PC")
+		require_NoError(t, err)
+		if ci.NumWaiting != 0 {
+			return fmt.Errorf("Expected to see no waiting requests, got %d", ci.NumWaiting)
+		}
+		return nil
+	})
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Simple JetStream Benchmarks
 ///////////////////////////////////////////////////////////////////////////