Repository: nats-io/nats-server
Pull request #2813 (2022-01-24 02:30:51)

Updates to Pull Consumers
This change fixes a break in behavior for one-shot pull batch requests and formalizes pull requests and their associated behaviors. We also now can expire out of order and we will expire on a leader change for the consumer as that state is shared for pending requests between leaders and followers, only if >= 2.7.1

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/go.mod b/go.mod
index 6f19fc07..16600d65 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/klauspost/compress v1.13.4
 	github.com/minio/highwayhash v1.0.1
 	github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296
-	github.com/nats-io/nats.go v1.13.1-0.20211122170419-d7c1d78a50fc
+	github.com/nats-io/nats.go v1.13.1-0.20220121202836-972a071d373d
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
 	golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce
diff --git a/go.sum b/go.sum
index 00b8f80d..d24030fd 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.20220121202836-972a071d373d h1:GRSmEJutHkdoxKsRypP575IIdoXe7Bm6yHQF6GcDBnA=
+github.com/nats-io/nats.go v1.13.1-0.20220121202836-972a071d373d/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=
diff --git a/server/auth.go b/server/auth.go
index e7b4d56a..ca3fe5b9 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -171,19 +171,16 @@ func (p *Permissions) clone() *Permissions {
 // Lock is assumed held.
 func (s *Server) checkAuthforWarnings() {
 	warn := false
-	if s.opts.Password != _EMPTY_ && !isBcrypt(s.opts.Password) {
+	if s.opts.Password != "" && !isBcrypt(s.opts.Password) {
 		warn = true
 	}
 	for _, u := range s.users {
 		// Skip warn if using TLS certs based auth
 		// unless a password has been left in the config.
-		if u.Password == _EMPTY_ && s.opts.TLSMap {
-			continue
-		}
-		// Check if this is our internal sys client created on the fly.
-		if s.sysAccOnlyNoAuthUser != _EMPTY_ && u.Username == s.sysAccOnlyNoAuthUser {
+		if u.Password == "" && s.opts.TLSMap {
 			continue
 		}
+
 		if !isBcrypt(u.Password) {
 			warn = true
 			break
@@ -1145,7 +1142,7 @@ func validateAllowedConnectionTypes(m map[string]struct{}) error {
 }
 
 func validateNoAuthUser(o *Options, noAuthUser string) error {
-	if noAuthUser == _EMPTY_ {
+	if noAuthUser == "" {
 		return nil
 	}
 	if len(o.TrustedOperators) > 0 {
diff --git a/server/const.go b/server/const.go
index e178413f..c6e1b6db 100644
--- a/server/const.go
+++ b/server/const.go
@@ -41,7 +41,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.7.1-beta02"
+	VERSION = "2.7.1-beta01"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index ecca76e4..960ddc06 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -256,6 +256,8 @@ type consumer struct {
 	node    RaftNode
 	infoSub *subscription
 	lqsent  time.Time
+	prm     map[string]struct{}
+	prOk    bool
 
 	// R>1 proposals
 	pch   chan struct{}
@@ -837,6 +839,10 @@ func (o *consumer) setLeader(isLeader bool) {
 
 		// If we are R>1 spin up our proposal loop.
 		if node != nil {
+			// Determine if we can send pending requests info to the group.
+			// They must be on server versions >= 2.7.1
+			o.checkAndSetPendingRequestsOk()
+			o.checkPendingRequests()
 			go o.loopAndForwardProposals(qch)
 		}
 
@@ -857,6 +863,10 @@ func (o *consumer) setLeader(isLeader bool) {
 			close(o.qch)
 			o.qch = nil
 		}
+		// Reset waiting if we are in pull mode.
+		if o.isPullMode() {
+			o.waiting = newWaitQueue(o.cfg.MaxWaiting)
+		}
 		o.mu.Unlock()
 	}
 }
@@ -1505,6 +1515,82 @@ func (o *consumer) updateAcks(dseq, sseq uint64) {
 	o.lat = time.Now()
 }
 
+// Communicate to the cluster an addition of a pending request.
+// Lock should be held.
+func (o *consumer) addClusterPendingRequest(reply string) {
+	if o.node == nil || !o.pendingRequestsOk() {
+		return
+	}
+	b := make([]byte, len(reply)+1)
+	b[0] = byte(addPendingRequest)
+	copy(b[1:], reply)
+	o.propose(b)
+}
+
+// Communicate to the cluster a removal of a pending request.
+// Lock should be held.
+func (o *consumer) removeClusterPendingRequest(reply string) {
+	if o.node == nil || !o.pendingRequestsOk() {
+		return
+	}
+	b := make([]byte, len(reply)+1)
+	b[0] = byte(removePendingRequest)
+	copy(b[1:], reply)
+	o.propose(b)
+}
+
+// Set whether or not we can send pending requests to followers.
+func (o *consumer) setPendingRequestsOk(ok bool) {
+	o.mu.Lock()
+	o.prOk = ok
+	o.mu.Unlock()
+}
+
+// Lock should be held.
+func (o *consumer) pendingRequestsOk() bool {
+	return o.prOk
+}
+
+// Set whether or not we can send info about pending pull requests to our group.
+// Will require all peers have a minimum version.
+func (o *consumer) checkAndSetPendingRequestsOk() {
+	o.mu.RLock()
+	s, isValid := o.srv, o.mset != nil
+	o.mu.RUnlock()
+	if !isValid {
+		return
+	}
+
+	if ca := o.consumerAssignment(); ca != nil && len(ca.Group.Peers) > 1 {
+		for _, pn := range ca.Group.Peers {
+			if si, ok := s.nodeToInfo.Load(pn); ok {
+				if !versionAtLeast(si.(nodeInfo).version, 2, 7, 1) {
+					// We expect all of our peers to eventually be up to date.
+					// So check again in awhile.
+					time.AfterFunc(eventsHBInterval, func() { o.checkAndSetPendingRequestsOk() })
+					o.setPendingRequestsOk(false)
+					return
+				}
+			}
+		}
+	}
+	o.setPendingRequestsOk(true)
+}
+
+// On leadership change make sure we alert the pending requests that they are no longer valid.
+func (o *consumer) checkPendingRequests() {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if o.mset == nil || o.outq == nil {
+		return
+	}
+	hdr := []byte("NATS/1.0 409 Leadership Change\r\n\r\n")
+	for reply := range o.prm {
+		o.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+	}
+	o.prm = nil
+}
+
 // Process a NAK.
 func (o *consumer) processNak(sseq, dseq uint64) {
 	o.mu.Lock()
@@ -1972,6 +2058,7 @@ type waitingRequest struct {
 	interest string
 	reply    string
 	n        int // For batching
+	d        int
 	expires  time.Time
 	received time.Time
 	noWait   bool
@@ -1985,11 +2072,13 @@ var wrPool = sync.Pool{
 }
 
 // Recycle this request. This request can not be accessed after this call.
-func (wr *waitingRequest) recycleIfDone() {
+func (wr *waitingRequest) recycleIfDone() bool {
 	if wr != nil && wr.n <= 0 {
 		wr.acc, wr.interest, wr.reply = nil, _EMPTY_, _EMPTY_
 		wrPool.Put(wr)
+		return true
 	}
+	return false
 }
 
 // Force a recycle.
@@ -2004,6 +2093,7 @@ func (wr *waitingRequest) recycle() {
 type waitQueue struct {
 	rp, wp int
 	last   time.Time
+	fexp   time.Time
 	reqs   []*waitingRequest
 }
 
@@ -2035,6 +2125,10 @@ func (wq *waitQueue) add(wr *waitingRequest) error {
 	}
 	// Track last active via when we receive a request.
 	wq.last = wr.received
+	// Track next to expire for all pending requests.
+	if !wr.expires.IsZero() && (wq.fexp.IsZero() || wr.expires.Before(wq.fexp)) {
+		wq.fexp = wr.expires
+	}
 	return nil
 }
 
@@ -2072,6 +2166,7 @@ func (wq *waitQueue) peek() *waitingRequest {
 func (wq *waitQueue) pop() *waitingRequest {
 	wr := wq.peek()
 	if wr != nil {
+		wr.d++
 		wr.n--
 		if wr.n <= 0 {
 			wq.removeCurrent()
@@ -2080,7 +2175,7 @@ func (wq *waitQueue) pop() *waitingRequest {
 	return wr
 }
 
-// removes the current read pointer (head FIFO) entry.
+// Removes the current read pointer (head FIFO) entry.
 func (wq *waitQueue) removeCurrent() {
 	if wq.rp < 0 {
 		return
@@ -2093,6 +2188,43 @@ func (wq *waitQueue) removeCurrent() {
 	}
 }
 
+// Will compact when we have interior deletes.
+func (wq *waitQueue) compact() {
+	if wq.isEmpty() {
+		return
+	}
+	nreqs, i := make([]*waitingRequest, cap(wq.reqs)), 0
+	for rp := wq.rp; rp != wq.wp; rp = (rp + 1) % cap(wq.reqs) {
+		if wr := wq.reqs[rp]; wr != nil {
+			nreqs[i] = wr
+			i++
+		}
+	}
+	// Reset here.
+	wq.rp, wq.wp, wq.reqs = 0, i, nreqs
+}
+
+// Return the replies for our pending requests.
+// No-op if push consumer or invalid etc.
+func (o *consumer) pendingRequestReplies() []string {
+	o.mu.RLock()
+	defer o.mu.RUnlock()
+	if o.waiting == nil {
+		return nil
+	}
+	wq, m := o.waiting, make(map[string]struct{})
+	for rp := o.waiting.rp; o.waiting.rp >= 0 && rp != wq.wp; rp = (rp + 1) % cap(wq.reqs) {
+		if wr := wq.reqs[rp]; wr != nil {
+			m[wr.reply] = struct{}{}
+		}
+	}
+	var replies []string
+	for reply := range m {
+		replies = append(replies, reply)
+	}
+	return replies
+}
+
 // Return next waiting request. This will check for expirations but not noWait or interest.
 // That will be handled by expireWaiting.
 // Lock should be held.
@@ -2115,6 +2247,9 @@ func (o *consumer) nextWaiting() *waitingRequest {
 		o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 		// Remove the current one, no longer valid.
 		o.waiting.removeCurrent()
+		if o.node != nil {
+			o.removeClusterPendingRequest(wr.reply)
+		}
 		wr.recycle()
 	}
 	return nil
@@ -2168,16 +2303,30 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 	// 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 {
+		if expired, _, _ := o.expireWaiting(); expired == 0 {
 			// Force expiration if needed.
 			o.forceExpireFirstWaiting()
 		}
 	}
 
-	// If the request is for noWait and we have pending requests already, error.
-	if noWait && o.checkWaitingForInterest() {
-		sendErr(408, "Requests Pending")
-		return
+	// If the request is for noWait and we have pending requests already, check if we have room.
+	if noWait {
+		msgsPending := o.adjustedPending()
+		// If no pending at all, decide what to do with request.
+		// If no expires was set then fail.
+		if msgsPending == 0 && expires.IsZero() {
+			sendErr(404, "No Messages")
+			return
+		}
+		if msgsPending > 0 {
+			_, _, batchPending := o.expireWaiting()
+			if msgsPending < uint64(batchPending) {
+				sendErr(408, "Requests Pending")
+				return
+			}
+		}
+		// If we are here this should be considered a one-shot situation.
+		// We will wait for expires but will return as soon as we have any messages.
 	}
 
 	// If we receive this request though an account export, we need to track that interest subject and account.
@@ -2192,7 +2341,7 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 
 	// 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.acc, wr.interest, wr.reply, wr.n, wr.d, wr.noWait, wr.expires = acc, interest, reply, batchSize, 0, noWait, expires
 	wr.received = time.Now()
 
 	if err := o.waiting.add(wr); err != nil {
@@ -2200,6 +2349,10 @@ func (o *consumer) processNextMsgReq(_ *subscription, c *client, _ *Account, _,
 		return
 	}
 	o.signalNewMessages()
+	// If we are clustered update our followers about this request.
+	if o.node != nil {
+		o.addClusterPendingRequest(wr.reply)
+	}
 }
 
 // Increase the delivery count for this message.
@@ -2326,53 +2479,81 @@ func (o *consumer) forceExpireFirstWaiting() {
 	// If we are expiring this and we think there is still interest, alert.
 	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")
+		hdr := []byte("NATS/1.0 408 Request Canceled\r\n\r\n")
 		o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 	}
 	o.waiting.removeCurrent()
+	if o.node != nil {
+		o.removeClusterPendingRequest(wr.reply)
+	}
 	wr.recycle()
 }
 
 // Will check for expiration and lack of interest on waiting requests.
-func (o *consumer) expireWaiting() int {
-	var expired int
-	now := time.Now()
-	for wr := o.waiting.peek(); wr != nil; wr = o.waiting.peek() {
-		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(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+func (o *consumer) expireWaiting() (int, int, int) {
+	if o.srv == nil || o.waiting.isEmpty() {
+		return 0, 0, 0
+	}
+
+	var expired, brp int
+	s, now := o.srv, time.Now()
+
+	// Signals interior deletes, which we will compact if needed.
+	var hid bool
+	remove := func(wr *waitingRequest, i int) {
+		if i == o.waiting.rp {
 			o.waiting.removeCurrent()
-			wr.recycle()
-			expired++
-			continue
+		} else {
+			o.waiting.reqs[i] = nil
+			hid = true
 		}
-
-		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 o.node != nil {
+			o.removeClusterPendingRequest(wr.reply)
 		}
+		expired++
+		wr.recycle()
+	}
+
+	wq := o.waiting
+	wq.fexp = time.Time{}
 
-		// Check gateway interest.
-		if s.gateway.enabled {
+	for rp := o.waiting.rp; o.waiting.rp >= 0 && rp != wq.wp; rp = (rp + 1) % cap(wq.reqs) {
+		wr := wq.reqs[rp]
+		// Check expiration.
+		if (wr.noWait && wr.d > 0) || (!wr.expires.IsZero() && now.After(wr.expires)) {
+			hdr := []byte("NATS/1.0 408 Request Timeout\r\n\r\n")
+			o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+			remove(wr, rp)
+			continue
+		}
+		// Now check interest.
+		rr := wr.acc.sl.Match(wr.interest)
+		interest := len(rr.psubs)+len(rr.qsubs) > 0
+		if !interest && 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
+			if s.hasGatewayInterest(wr.acc.Name, wr.interest) || time.Since(wr.received) < defaultGatewayRecentSubExpiration {
+				interest = true
 			}
 		}
+		// If interest, update batch pending requests counter and update fexp timer.
+		if interest {
+			brp += wr.n
+			if !wr.expires.IsZero() && (wq.fexp.IsZero() || wr.expires.Before(wq.fexp)) {
+				wq.fexp = wr.expires
+			}
+			continue
+		}
 		// No more interest here so go ahead and remove this one from our list.
-		o.waiting.removeCurrent()
-		wr.recycle()
-		expired++
+		remove(wr, rp)
+	}
+
+	// If we have interior deletes from out of order invalidation, compact the waiting queue.
+	if hid {
+		o.waiting.compact()
 	}
-	return expired
+
+	return expired, o.waiting.len(), brp
 }
 
 // Will check to make sure those waiting still have registered interest.
@@ -2461,7 +2642,9 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			dsubj = o.dsubj
 		} else if wr := o.nextWaiting(); wr != nil {
 			dsubj = wr.reply
-			wr.recycleIfDone()
+			if wr.recycleIfDone() && o.node != nil {
+				o.removeClusterPendingRequest(dsubj)
+			}
 		} else {
 			// We will redo this one.
 			o.sseq--
@@ -2518,8 +2701,16 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		}
 
 		// Make sure to process any expired requests that are pending.
+		var wrExp <-chan time.Time
 		if o.isPullMode() {
 			o.expireWaiting()
+			if o.waiting.len() > 0 && !o.waiting.fexp.IsZero() {
+				expires := time.Until(o.waiting.fexp)
+				if expires <= 0 {
+					expires = time.Millisecond
+				}
+				wrExp = time.NewTimer(expires).C
+			}
 		}
 
 		// We will wait here for new messages to arrive.
@@ -2535,6 +2726,10 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			return
 		case <-mch:
 			// Messages are waiting.
+		case <-wrExp:
+			o.mu.Lock()
+			o.expireWaiting()
+			o.mu.Unlock()
 		case <-hbc:
 			if o.isActive() {
 				const t = "NATS/1.0 100 Idle Heartbeat\r\n%s: %d\r\n%s: %d\r\n\r\n"
diff --git a/server/events.go b/server/events.go
index d097c47a..8fba068a 100644
--- a/server/events.go
+++ b/server/events.go
@@ -1135,6 +1135,7 @@ func (s *Server) remoteServerUpdate(sub *subscription, c *client, _ *Account, su
 	node := string(getHash(si.Name))
 	s.nodeToInfo.Store(node, nodeInfo{
 		si.Name,
+		si.Version,
 		si.Cluster,
 		si.Domain,
 		si.ID,
@@ -1176,7 +1177,7 @@ func (s *Server) processNewServer(si *ServerInfo) {
 		node := string(getHash(si.Name))
 		// Only update if non-existent
 		if _, ok := s.nodeToInfo.Load(node); !ok {
-			s.nodeToInfo.Store(node, nodeInfo{si.Name, si.Cluster, si.Domain, si.ID, nil, nil, false, si.JetStream})
+			s.nodeToInfo.Store(node, nodeInfo{si.Name, si.Version, si.Cluster, si.Domain, si.ID, nil, nil, false, si.JetStream})
 		}
 	}
 	// Announce ourselves..
diff --git a/server/jetstream.go b/server/jetstream.go
index ce032c38..63d3a5bc 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -785,12 +785,22 @@ func (s *Server) migrateEphemerals() {
 	}
 	js.mu.Unlock()
 
+	// In case we have stale pending requests.
+	hdr := []byte("NATS/1.0 409 Consumer Migration\r\n\r\n")
+
 	// Process the consumers.
 	for _, ca := range consumers {
 		// Locate the consumer itself.
 		if acc, err := s.LookupAccount(ca.Client.Account); err == nil && acc != nil {
 			if mset, err := acc.lookupStream(ca.Stream); err == nil && mset != nil {
 				if o := mset.lookupConsumer(ca.Name); o != nil {
+					if pr := o.pendingRequestReplies(); len(pr) > 0 {
+						o.mu.Lock()
+						for _, reply := range pr {
+							o.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+						}
+						o.mu.Unlock()
+					}
 					state := o.readStoreState()
 					o.deleteWithoutAdvisory()
 					js.mu.Lock()
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index bb345d39..f68b9314 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -63,6 +63,7 @@ type Placement struct {
 // Define types of the entry.
 type entryOp uint8
 
+// ONLY ADD TO THE END, DO NOT INSERT IN BETWEEN WILL BREAK SERVER INTEROP.
 const (
 	// Meta ops.
 	assignStreamOp entryOp = iota
@@ -73,15 +74,18 @@ const (
 	streamMsgOp
 	purgeStreamOp
 	deleteMsgOp
-	// Consumer ops
+	// Consumer ops.
 	updateDeliveredOp
 	updateAcksOp
 	// Compressed consumer assignments.
 	assignCompressedConsumerOp
 	// Filtered Consumer skip.
 	updateSkipOp
-	// Update Stream
+	// Update Stream.
 	updateStreamOp
+	// For updating information on pending pull requests.
+	addPendingRequest
+	removePendingRequest
 )
 
 // raftGroups are controlled by the metagroup controller.
@@ -3142,6 +3146,23 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 					o.sseq = le.Uint64(buf[1:])
 				}
 				o.mu.Unlock()
+			case addPendingRequest:
+				if !o.isLeader() {
+					o.mu.Lock()
+					if o.prm == nil {
+						o.prm = make(map[string]struct{})
+					}
+					o.prm[string(buf[1:])] = struct{}{}
+					o.mu.Unlock()
+				}
+			case removePendingRequest:
+				if !o.isLeader() {
+					o.mu.Lock()
+					if o.prm != nil {
+						delete(o.prm, string(buf[1:]))
+					}
+					o.mu.Unlock()
+				}
 			default:
 				panic(fmt.Sprintf("JetStream Cluster Unknown group entry op type! %v", entryOp(buf[0])))
 			}
diff --git a/server/opts.go b/server/opts.go
index fba65710..dc170786 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -663,7 +663,7 @@ func configureSystemAccount(o *Options, m map[string]interface{}) (retErr error)
 // or was present but set to false.
 func (o *Options) ProcessConfigFile(configFile string) error {
 	o.ConfigFile = configFile
-	if configFile == _EMPTY_ {
+	if configFile == "" {
 		return nil
 	}
 	m, err := conf.ParseFileWithChecks(configFile)
@@ -1018,7 +1018,7 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 				}
 			}
 			// In case "system_account" is defined as well, it takes precedence
-			if o.SystemAccount == _EMPTY_ {
+			if o.SystemAccount == "" {
 				o.SystemAccount = o.TrustedOperators[0].SystemAccount
 			}
 		}
diff --git a/server/parser.go b/server/parser.go
index d5c9b495..a9c3533a 100644
--- a/server/parser.go
+++ b/server/parser.go
@@ -147,7 +147,8 @@ func (c *client) parse(buf []byte) error {
 	// proper CONNECT if needed.
 	authSet := c.awaitingAuth()
 	// Snapshot max control line as well.
-	s, mcl, trace := c.srv, c.mcl, c.trace
+	mcl := c.mcl
+	trace := c.trace
 	c.mu.Unlock()
 
 	// Move to loop instead of range syntax to allow jumping of i
@@ -159,28 +160,7 @@ func (c *client) parse(buf []byte) error {
 			c.op = b
 			if b != 'C' && b != 'c' {
 				if authSet {
-					if s == nil {
-						goto authErr
-					}
-					var ok bool
-					// Check here for NoAuthUser. If this is set allow non CONNECT protos as our first.
-					// E.g. telnet proto demos.
-					if noAuthUser := s.getOpts().NoAuthUser; noAuthUser != _EMPTY_ {
-						s.mu.Lock()
-						user, exists := s.users[noAuthUser]
-						s.mu.Unlock()
-						if exists {
-							c.RegisterUser(user)
-							c.mu.Lock()
-							c.clearAuthTimer()
-							c.flags.set(connectReceived)
-							c.mu.Unlock()
-							authSet, ok = false, true
-						}
-					}
-					if !ok {
-						goto authErr
-					}
+					goto authErr
 				}
 				// If the connection is a gateway connection, make sure that
 				// if this is an inbound, it starts with a CONNECT.
diff --git a/server/route.go b/server/route.go
index 2942912d..da8a74dd 100644
--- a/server/route.go
+++ b/server/route.go
@@ -1422,7 +1422,8 @@ func (s *Server) addRoute(c *client, info *Info) (bool, bool) {
 		s.remotes[id] = c
 		// check to be consistent and future proof. but will be same domain
 		if s.sameDomain(info.Domain) {
-			s.nodeToInfo.Store(c.route.hash, nodeInfo{c.route.remoteName, s.info.Cluster, info.Domain, id, nil, nil, false, info.JetStream})
+			s.nodeToInfo.Store(c.route.hash,
+				nodeInfo{c.route.remoteName, s.info.Version, s.info.Cluster, info.Domain, id, nil, nil, false, info.JetStream})
 		}
 		c.mu.Lock()
 		c.route.connectURLs = info.ClientConnectURLs
diff --git a/server/server.go b/server/server.go
index 785460e7..42df6091 100644
--- a/server/server.go
+++ b/server/server.go
@@ -283,6 +283,7 @@ type srvIPQueueLogger struct {
 // For tracking JS nodes.
 type nodeInfo struct {
 	name    string
+	version string
 	cluster string
 	domain  string
 	id      string
@@ -407,6 +408,7 @@ func NewServer(opts *Options) (*Server, error) {
 		ourNode := string(getHash(serverName))
 		s.nodeToInfo.Store(ourNode, nodeInfo{
 			serverName,
+			VERSION,
 			opts.Cluster.Name,
 			opts.JetStreamDomain,
 			info.ID,
@@ -529,6 +531,39 @@ func NewServer(opts *Options) (*Server, error) {
 	return s, nil
 }
 
+var semVerRe = regexp.MustCompile(`\Av?([0-9]+)\.?([0-9]+)?\.?([0-9]+)?`)
+
+func versionComponents(version string) (major, minor, patch int, err error) {
+	m := semVerRe.FindStringSubmatch(version)
+	if m == nil {
+		return 0, 0, 0, errors.New("invalid semver")
+	}
+	major, err = strconv.Atoi(m[1])
+	if err != nil {
+		return -1, -1, -1, err
+	}
+	minor, err = strconv.Atoi(m[2])
+	if err != nil {
+		return -1, -1, -1, err
+	}
+	patch, err = strconv.Atoi(m[3])
+	if err != nil {
+		return -1, -1, -1, err
+	}
+	return major, minor, patch, err
+}
+
+func versionAtLeast(version string, emajor, eminor, epatch int) bool {
+	major, minor, patch, err := versionComponents(version)
+	if err != nil {
+		return false
+	}
+	if major < emajor || minor < eminor || patch < epatch {
+		return false
+	}
+	return true
+}
+
 func (s *Server) logRejectedTLSConns() {
 	defer s.grWG.Done()
 	t := time.NewTicker(time.Second)
@@ -2362,6 +2397,7 @@ func (s *Server) startMonitoring(secure bool) error {
 			}
 		}
 		srv.Close()
+		srv.Handler = nil
 		s.mu.Lock()
 		s.httpHandler = nil
 		s.mu.Unlock()


Test patch:
diff --git a/server/auth_test.go b/server/auth_test.go
index dcd87649..65cc6fcb 100644
--- a/server/auth_test.go
+++ b/server/auth_test.go
@@ -15,13 +15,11 @@ package server
 
 import (
 	"fmt"
-	"net"
 	"net/url"
 	"os"
 	"reflect"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
@@ -275,46 +273,3 @@ func TestNoAuthUser(t *testing.T) {
 		})
 	}
 }
-
-func TestNoAuthUserNoConnectProto(t *testing.T) {
-	conf := createConfFile(t, []byte(`
-		listen: "127.0.0.1:-1"
-		accounts {
-			A { users [{user: "foo", password: "pwd"}] }
-		}
-		authorization { timeout: 1 }
-		no_auth_user: "foo"
-	`))
-	defer os.Remove(conf)
-	s, o := RunServerWithConfig(conf)
-	defer s.Shutdown()
-
-	checkClients := func(n int) {
-		t.Helper()
-		time.Sleep(100 * time.Millisecond)
-		if nc := s.NumClients(); nc != n {
-			t.Fatalf("Expected %d clients, got %d", n, nc)
-		}
-	}
-
-	conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", o.Host, o.Port))
-	require_NoError(t, err)
-	defer conn.Close()
-	checkClientsCount(t, s, 1)
-
-	// With no auth user we should not require a CONNECT.
-	// Make sure we are good on not sending CONN first.
-	_, err = conn.Write([]byte("PUB foo 2\r\nok\r\n"))
-	require_NoError(t, err)
-	checkClients(1)
-	conn.Close()
-
-	// Now make sure we still do get timed out though.
-	conn, err = net.Dial("tcp", fmt.Sprintf("%s:%d", o.Host, o.Port))
-	require_NoError(t, err)
-	defer conn.Close()
-	checkClientsCount(t, s, 1)
-
-	time.Sleep(1200 * time.Millisecond)
-	checkClientsCount(t, s, 0)
-}
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 22f782b9..4637198a 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -10049,6 +10049,146 @@ func TestJetStreamClusterPullPerf(t *testing.T) {
 	fmt.Printf("%.0f mb/s\n\n", float64(si.State.Bytes/(1024*1024))/tt.Seconds())
 }
 
+// Test that we get the right signaling when a consumer leader change occurs for any pending requests.
+func TestJetStreamClusterPullConsumerLeaderChange(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "JSC", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Replicas: 3,
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:       "dlc",
+		AckPolicy:     nats.AckExplicitPolicy,
+		FilterSubject: "foo",
+	})
+	require_NoError(t, err)
+
+	rsubj := fmt.Sprintf(JSApiRequestNextT, "TEST", "dlc")
+	sub, err := nc.SubscribeSync("reply")
+	require_NoError(t, err)
+	defer sub.Unsubscribe()
+
+	drainSub := func() {
+		t.Helper()
+		for _, err := sub.NextMsg(0); err == nil; _, err = sub.NextMsg(0) {
+		}
+		checkSubsPending(t, sub, 0)
+	}
+
+	// Queue up a request that can live for a bit.
+	req := &JSApiConsumerGetNextRequest{Expires: 2 * time.Second}
+	jreq, err := json.Marshal(req)
+	require_NoError(t, err)
+	err = nc.PublishRequest(rsubj, "reply", jreq)
+	require_NoError(t, err)
+	// Make sure request is recorded and replicated.
+	time.Sleep(100 * time.Millisecond)
+	checkSubsPending(t, sub, 0)
+
+	// Now have consumer leader change and make sure we get signaled that our request is not valid.
+	_, err = nc.Request(fmt.Sprintf(JSApiConsumerLeaderStepDownT, "TEST", "dlc"), nil, time.Second)
+	require_NoError(t, err)
+	c.waitOnConsumerLeader("$G", "TEST", "dlc")
+	checkSubsPending(t, sub, 1)
+	m, err := sub.NextMsg(0)
+	require_NoError(t, err)
+	// Make sure this is an alert that tells us our request is no longer valid.
+	if m.Header.Get("Status") != "409" {
+		t.Fatalf("Expected a 409 status code, got %q", m.Header.Get("Status"))
+	}
+	checkSubsPending(t, sub, 0)
+
+	// Add a few messages to the stream to fulfill a request.
+	for i := 0; i < 10; i++ {
+		_, err := js.Publish("foo", []byte("HELLO"))
+		require_NoError(t, err)
+	}
+	req = &JSApiConsumerGetNextRequest{Batch: 10, Expires: 10 * time.Second}
+	jreq, err = json.Marshal(req)
+	require_NoError(t, err)
+	err = nc.PublishRequest(rsubj, "reply", jreq)
+	require_NoError(t, err)
+	checkSubsPending(t, sub, 10)
+	drainSub()
+
+	// Now do a leader change again, make sure we do not get anything about that request.
+	_, err = nc.Request(fmt.Sprintf(JSApiConsumerLeaderStepDownT, "TEST", "dlc"), nil, time.Second)
+	require_NoError(t, err)
+	c.waitOnConsumerLeader("$G", "TEST", "dlc")
+	time.Sleep(100 * time.Millisecond)
+	checkSubsPending(t, sub, 0)
+
+	// Make sure we do not get anything if we expire, etc.
+	req = &JSApiConsumerGetNextRequest{Batch: 10, Expires: 250 * time.Millisecond}
+	jreq, err = json.Marshal(req)
+	require_NoError(t, err)
+	err = nc.PublishRequest(rsubj, "reply", jreq)
+	require_NoError(t, err)
+	// Let it expire.
+	time.Sleep(350 * time.Millisecond)
+	checkSubsPending(t, sub, 1)
+
+	// Now do a leader change again, make sure we do not get anything about that request.
+	_, err = nc.Request(fmt.Sprintf(JSApiConsumerLeaderStepDownT, "TEST", "dlc"), nil, time.Second)
+	require_NoError(t, err)
+	c.waitOnConsumerLeader("$G", "TEST", "dlc")
+	checkSubsPending(t, sub, 1)
+}
+
+func TestJetStreamClusterEphemeralPullConsumerServerShutdown(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "JSC", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Replicas: 2,
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	ci, err := js.AddConsumer("TEST", &nats.ConsumerConfig{
+		AckPolicy:     nats.AckExplicitPolicy,
+		FilterSubject: "foo",
+	})
+	require_NoError(t, err)
+
+	rsubj := fmt.Sprintf(JSApiRequestNextT, "TEST", ci.Name)
+	sub, err := nc.SubscribeSync("reply")
+	require_NoError(t, err)
+	defer sub.Unsubscribe()
+
+	// Queue up a request that can live for a bit.
+	req := &JSApiConsumerGetNextRequest{Expires: 2 * time.Second}
+	jreq, err := json.Marshal(req)
+	require_NoError(t, err)
+	err = nc.PublishRequest(rsubj, "reply", jreq)
+	require_NoError(t, err)
+	// Make sure request is recorded and replicated.
+	time.Sleep(100 * time.Millisecond)
+	checkSubsPending(t, sub, 0)
+
+	// Now shutdown the server where this ephemeral lives.
+	c.consumerLeader("$G", "TEST", ci.Name).Shutdown()
+	checkSubsPending(t, sub, 1)
+	m, err := sub.NextMsg(0)
+	require_NoError(t, err)
+	// Make sure this is an alert that tells us our request is no longer valid.
+	if m.Header.Get("Status") != "409" {
+		t.Fatalf("Expected a 409 status code, got %q", m.Header.Get("Status"))
+	}
+}
+
 // Support functions
 
 // Used to setup superclusters for tests.
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index df7154f1..d8bd142d 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -1656,7 +1656,7 @@ func TestJetStreamWorkQueueMaxWaiting(t *testing.T) {
 			}
 			sendStreamMsg(t, nc, "foo", "Hello World!")
 			sendStreamMsg(t, nc, "bar", "Hello World!")
-			expectWaiting(JSWaitQueueDefaultMax - 2)
+			expectWaiting(JSWaitQueueDefaultMax - 3)
 		})
 	}
 }
@@ -14324,6 +14324,183 @@ func TestJetStreamPullConsumerCrossAccountsAndLeafNodes(t *testing.T) {
 	})
 }
 
+// This test is to explicitly test for all combinations of pull consumer behavior.
+// 1. Long poll, will be used to emulate push. A request is only invalidated when batch is filled, it expires, or we lose interest.
+// 2. Batch 1, will return no messages or a message. Works today.
+// 3. Conditional wait, or one shot. This is what the clients do when the do a fetch().
+//    They expect to wait up to a given time for any messages but will return once they have any to deliver, so parital fills.
+// 4. Try, which never waits at all ever.
+func TestJetStreamPullConsumersOneShotBehavior(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	// Client for API requests.
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:       "dlc",
+		AckPolicy:     nats.AckExplicitPolicy,
+		FilterSubject: "foo",
+	})
+	require_NoError(t, err)
+
+	// We will do low level requests by hand for this test as to not depend on any client impl.
+	rsubj := fmt.Sprintf(JSApiRequestNextT, "TEST", "dlc")
+
+	getNext := func(batch int, expires time.Duration, noWait bool) (numMsgs int, elapsed time.Duration, hdr *nats.Header) {
+		t.Helper()
+		req := &JSApiConsumerGetNextRequest{Batch: batch, Expires: expires, NoWait: noWait}
+		jreq, err := json.Marshal(req)
+		require_NoError(t, err)
+		// Create listener.
+		reply, msgs := nats.NewInbox(), make(chan *nats.Msg, batch)
+		sub, err := nc.ChanSubscribe(reply, msgs)
+		require_NoError(t, err)
+		defer sub.Unsubscribe()
+
+		// Send request.
+		start := time.Now()
+		err = nc.PublishRequest(rsubj, reply, jreq)
+		require_NoError(t, err)
+
+		for {
+			select {
+			case m := <-msgs:
+				if len(m.Data) == 0 && m.Header != nil {
+					return numMsgs, time.Since(start), &m.Header
+				}
+				numMsgs++
+				if numMsgs >= batch {
+					return numMsgs, time.Since(start), nil
+				}
+			case <-time.After(expires + 250*time.Millisecond):
+				t.Fatalf("Did not receive all the msgs in time")
+			}
+		}
+	}
+
+	expect := func(batch int, expires time.Duration, noWait bool, ne int, he *nats.Header, lt time.Duration, gt time.Duration) {
+		t.Helper()
+		n, e, h := getNext(batch, expires, noWait)
+		if n != ne {
+			t.Fatalf("Expected %d msgs, got %d", ne, n)
+		}
+		if !reflect.DeepEqual(h, he) {
+			t.Fatalf("Expected %+v hdr, got %+v", he, h)
+		}
+		if lt > 0 && e > lt {
+			t.Fatalf("Expected elapsed of %v to be less than %v", e, lt)
+		}
+		if gt > 0 && e < gt {
+			t.Fatalf("Expected elapsed of %v to be greater than %v", e, gt)
+		}
+	}
+	expectAfter := func(batch int, expires time.Duration, noWait bool, ne int, he *nats.Header, gt time.Duration) {
+		t.Helper()
+		expect(batch, expires, noWait, ne, he, 0, gt)
+	}
+	expectInstant := func(batch int, expires time.Duration, noWait bool, ne int, he *nats.Header) {
+		t.Helper()
+		expect(batch, expires, noWait, ne, he, 2*time.Millisecond, 0)
+	}
+	expectOK := func(batch int, expires time.Duration, noWait bool, ne int) {
+		t.Helper()
+		expectInstant(batch, expires, noWait, ne, nil)
+	}
+
+	noMsgs := &nats.Header{"Status": []string{"404"}, "Description": []string{"No Messages"}}
+	reqTimeout := &nats.Header{"Status": []string{"408"}, "Description": []string{"Request Timeout"}}
+
+	// We are empty here, meaning no messages available.
+	// Do not wait, should get noMsgs.
+	expectInstant(1, 0, true, 0, noMsgs)
+	// We should wait here for the full second.
+	expectAfter(1, 250*time.Millisecond, false, 0, reqTimeout, 250*time.Millisecond)
+	// This should also wait since no messages are available. This is the one shot scenario, or wait for at least a message if none are there.
+	expectAfter(1, 500*time.Millisecond, true, 0, reqTimeout, 500*time.Millisecond)
+
+	// Now let's put some messages into the system.
+	for i := 0; i < 20; i++ {
+		_, err := js.Publish("foo", []byte("HELLO"))
+		require_NoError(t, err)
+	}
+
+	// Now run same 3 scenarios.
+	expectOK(1, 0, true, 1)
+	expectOK(5, 500*time.Millisecond, false, 5)
+	expectOK(5, 500*time.Millisecond, true, 5)
+}
+
+func TestJetStreamPullConsumersMultipleRequestsExpireOutOfOrder(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+
+	// Client for API requests.
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:       "dlc",
+		AckPolicy:     nats.AckExplicitPolicy,
+		FilterSubject: "foo",
+	})
+	require_NoError(t, err)
+
+	// We will now queue up 4 requests. All should expire but they will do so out of order.
+	// We want to make sure we get them in correct order.
+	rsubj := fmt.Sprintf(JSApiRequestNextT, "TEST", "dlc")
+	sub, err := nc.SubscribeSync("i.*")
+	require_NoError(t, err)
+	defer sub.Unsubscribe()
+
+	for _, expires := range []time.Duration{200, 100, 25, 75} {
+		reply := fmt.Sprintf("i.%d", expires)
+		req := &JSApiConsumerGetNextRequest{Expires: expires * time.Millisecond}
+		jreq, err := json.Marshal(req)
+		require_NoError(t, err)
+		err = nc.PublishRequest(rsubj, reply, jreq)
+		require_NoError(t, err)
+	}
+	start := time.Now()
+	checkSubsPending(t, sub, 4)
+	elapsed := time.Since(start)
+
+	if elapsed < 200*time.Millisecond || elapsed > 500*time.Millisecond {
+		t.Fatalf("Expected elapsed to be close to %v, but got %v", 200*time.Millisecond, elapsed)
+	}
+
+	var rs []string
+	for i := 0; i < 4; i++ {
+		m, err := sub.NextMsg(0)
+		require_NoError(t, err)
+		rs = append(rs, m.Subject)
+	}
+	if expected := []string{"i.25", "i.75", "i.100", "i.200"}; !reflect.DeepEqual(rs, expected) {
+		t.Fatalf("Received in wrong order, wanted %+v, got %+v", expected, rs)
+	}
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Simple JetStream Benchmarks
 ///////////////////////////////////////////////////////////////////////////