Repository: nats-io/nats-server
Pull request #5111 (2024-02-21 08:03:25)

Fix sending `Consumer Deleted` on peer remove
There was a bug that caused sending `Consumer Deleted` sending to pull consumers whenever a Peer was removed.

This is a draft solution, as I don't like how the code looks right now:
When peer is removed, the `delete flag` is set to `true`, and thats fine. After all we need to remove the stream from the peer.
However, we do not differentiate in any way if we are removing the RAFT asset as a whole, or just it from the given peer.
That's why I added additional flag, which allows more informed choices by the codepaths for stopping streams and consumers.

Signed-off-by: Tomasz Pietrek <tomasz@nats.io>


Jarema: @derekcollison I think the biggest thing here is that the  one `delete` flag does not distinguish `remove consumer from a peer` and `remove consumer RAFT resource as a whole` explicitly.
Because of that the consumer function calls seems to have too little information.
I agree that it does not look clean, hence the draft PR, that just confirms source of the bug. Now time to clean it up without breaking the test :).

Thanks for the hints! That might allow removing the ugly new bool flag.
derekcollison: You could check for a consumer assignment on a delete flag to determine the difference.
derekcollison: Let me know when you want a review.
Jarema: @derekcollison After some tests, I realized that the consumer assignment deleted flag is not always set. 

Chatted with @neilalexander and we came to the conclusion that the cleanest thing is to change the current flag into one that clearly lets us know if we're just deleting the consumer/stream on this node, or we're removing the whole Raft asset.

I would like to write few more tests before merging this one.
derekcollison: What I meant was if you can lookup up the assignment, its not deleted, just being deleted from this node.

Patch:
diff --git a/.travis.yml b/.travis.yml
index b687f3ad..1ee604e7 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -8,8 +8,8 @@ language: go
 go:
   # This should be quoted or use .x, but should not be unquoted.
   # Remember that a YAML bare float drops trailing zeroes.
+  - "1.22.0"
   - "1.21.7"
-  - "1.20.13"
 
 go_import_path: github.com/nats-io/nats-server
 
@@ -52,4 +52,4 @@ deploy:
   script: curl -sL http://git.io/goreleaser | bash
   on:
     tags: true
-    condition: ($TRAVIS_GO_VERSION =~ 1.21) && ($TEST_SUITE = "compile")
+    condition: ($TRAVIS_GO_VERSION =~ 1.22) && ($TEST_SUITE = "compile")
diff --git a/server/client.go b/server/client.go
index a25ac6c2..b8d5e8c8 100644
--- a/server/client.go
+++ b/server/client.go
@@ -4175,27 +4175,27 @@ func getHeader(key string, hdr []byte) []byte {
 		return nil
 	}
 	index := bytes.Index(hdr, []byte(key))
-	if index < 0 {
-		return nil
-	}
-	// Make sure this key does not have additional prefix.
-	if index < 2 || hdr[index-1] != '\n' || hdr[index-2] != '\r' {
+	hdrLen := len(hdr)
+	// Check that we have enough characters, this will handle the -1 case of the key not
+	// being found and will also handle not having enough characters for trailing CRLF.
+	if index < 2 {
 		return nil
 	}
-	index += len(key)
-	if index >= len(hdr) {
+	// There should be a terminating CRLF.
+	if index >= hdrLen-1 || hdr[index-1] != '\n' || hdr[index-2] != '\r' {
 		return nil
 	}
-	if hdr[index] != ':' {
+	// The key should be immediately followed by a : separator.
+	index += len(key) + 1
+	if index >= hdrLen || hdr[index-1] != ':' {
 		return nil
 	}
-	index++
-
-	var value []byte
-	hdrLen := len(hdr)
-	for hdr[index] == ' ' && index < hdrLen {
+	// Skip over whitespace before the value.
+	for index < hdrLen && hdr[index] == ' ' {
 		index++
 	}
+	// Collect together the rest of the value until we hit a CRLF.
+	var value []byte
 	for index < hdrLen {
 		if hdr[index] == '\r' && index < hdrLen-1 && hdr[index+1] == '\n' {
 			break
diff --git a/server/const.go b/server/const.go
index 1e97fa63..82e968f5 100644
--- a/server/const.go
+++ b/server/const.go
@@ -14,6 +14,7 @@
 package server
 
 import (
+	"regexp"
 	"time"
 )
 
@@ -37,6 +38,8 @@ var (
 	gitCommit string
 	// trustedKeys is a whitespace separated array of trusted operator's public nkeys.
 	trustedKeys string
+	// SemVer regexp to validate the VERSION.
+	semVerRe = regexp.MustCompile(`^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$`)
 )
 
 const (
@@ -218,9 +221,3 @@ const (
 	// DEFAULT_FETCH_TIMEOUT is the default time that the system will wait for an account fetch to return.
 	DEFAULT_ACCOUNT_FETCH_TIMEOUT = 1900 * time.Millisecond
 )
-
-// For direct get batch requests.
-const (
-	dg  = "NATS/1.0\r\nNats-Stream: %s\r\nNats-Subject: %s\r\nNats-Sequence: %d\r\nNats-Time-Stamp: %s\r\n\r\n"
-	dgb = "NATS/1.0\r\nNats-Stream: %s\r\nNats-Subject: %s\r\nNats-Sequence: %d\r\nNats-Time-Stamp: %s\r\nNats-Num-Pending: %d\r\nNats-Last-Sequence: %d\r\n\r\n"
-)
diff --git a/server/consumer.go b/server/consumer.go
index 563ae5e9..e4f58015 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -2333,16 +2333,21 @@ func (o *consumer) checkPendingRequests() {
 // This will release any pending pull requests if applicable.
 // Should be called only by the leader being deleted or stopped.
 // Lock should be held.
-func (o *consumer) releaseAnyPendingRequests() {
+func (o *consumer) releaseAnyPendingRequests(isConsumerAssigned bool) {
 	if o.mset == nil || o.outq == nil || o.waiting.len() == 0 {
 		return
 	}
-	hdr := []byte("NATS/1.0 409 Consumer Deleted\r\n\r\n")
+	var hdr []byte
+	if !isConsumerAssigned {
+		hdr = []byte("NATS/1.0 409 Consumer Deleted\r\n\r\n")
+	}
 	wq := o.waiting
 	o.waiting = nil
 	for i, rp := 0, wq.rp; i < wq.n; i++ {
 		if wr := wq.reqs[rp]; wr != nil {
-			o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+			if hdr != nil {
+				o.outq.send(newJSPubMsg(wr.reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+			}
 			wr.recycle()
 		}
 		rp = (rp + 1) % cap(wq.reqs)
@@ -5145,18 +5150,27 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 	}
 	o.closed = true
 
-	// Check if we are the leader and are being deleted.
+	// Check if we are the leader and are being deleted (as a node).
 	if dflag && o.isLeader() {
 		// If we are clustered and node leader (probable from above), stepdown.
 		if node := o.node; node != nil && node.Leader() {
 			node.StepDown()
 		}
-		if advisory {
+
+		// dflag does not mean that the whole consumer is being deleted,
+		// just that the consumer node is being removed, so we send delete
+		// advisories only if the consumer assignment is gone, which means
+		// actual consumer removal happened.
+		isConsumerAssigned := true
+		if o.acc == nil || o.acc.js == nil || !o.acc.js.consumerAssigned(o.stream, o.name) {
+			isConsumerAssigned = false
+		}
+		if !isConsumerAssigned && advisory {
 			o.sendDeleteAdvisoryLocked()
 		}
 		if o.isPullMode() {
 			// Release any pending.
-			o.releaseAnyPendingRequests()
+			o.releaseAnyPendingRequests(isConsumerAssigned)
 		}
 	}
 
diff --git a/server/filestore.go b/server/filestore.go
index 7f43d8fd..8e5f6ba0 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -2620,6 +2620,119 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 	return fss
 }
 
+// MultiLastSeqs will return a sorted list of sequences that match all subjects presented in filters.
+// We will not exceed the maxSeq, which if 0 becomes the store's last sequence.
+func (fs *fileStore) MultiLastSeqs(filters []string, maxSeq uint64, maxAllowed int) ([]uint64, error) {
+	fs.mu.RLock()
+	defer fs.mu.RUnlock()
+
+	if fs.state.Msgs == 0 || fs.noTrackSubjects() {
+		return nil, nil
+	}
+
+	lastBlkIndex := len(fs.blks) - 1
+	lastMB := fs.blks[lastBlkIndex]
+
+	// Implied last sequence.
+	if maxSeq == 0 {
+		maxSeq = fs.state.LastSeq
+	} else {
+		// Udate last mb index if not last seq.
+		lastBlkIndex, lastMB = fs.selectMsgBlockWithIndex(maxSeq)
+	}
+	//Make sure non-nil
+	if lastMB == nil {
+		return nil, nil
+	}
+
+	// Grab our last mb index (not same as blk index).
+	lastMB.mu.RLock()
+	lastMBIndex := lastMB.index
+	lastMB.mu.RUnlock()
+
+	subs := make(map[string]*psi)
+	ltSeen := make(map[string]uint32)
+	for _, filter := range filters {
+		fs.psim.Match(stringToBytes(filter), func(subj []byte, psi *psi) {
+			s := string(subj)
+			subs[s] = psi
+			if psi.lblk < lastMBIndex {
+				ltSeen[s] = psi.lblk
+			}
+		})
+	}
+
+	// If all subjects have a lower last index, select the largest for our walk backwards.
+	if len(ltSeen) == len(subs) {
+		max := uint32(0)
+		for _, mbi := range ltSeen {
+			if mbi > max {
+				max = mbi
+			}
+		}
+		lastMB = fs.bim[max]
+	}
+
+	// Collect all sequences needed.
+	seqs := make([]uint64, 0, len(subs))
+	for i, lnf := lastBlkIndex, false; i >= 0; i-- {
+		if len(subs) == 0 {
+			break
+		}
+		mb := fs.blks[i]
+		if !lnf {
+			if mb != lastMB {
+				continue
+			}
+			lnf = true
+		}
+		// We can start properly looking here.
+		mb.mu.Lock()
+		mb.ensurePerSubjectInfoLoaded()
+		for subj, psi := range subs {
+			if ss := mb.fss[subj]; ss != nil {
+				if ss.Last <= maxSeq {
+					seqs = append(seqs, ss.Last)
+					delete(subs, subj)
+				} else {
+					// Need to search for it since last is > maxSeq.
+					if mb.cacheNotLoaded() {
+						mb.loadMsgsWithLock()
+					}
+					var smv StoreMsg
+					fseq := atomic.LoadUint64(&mb.first.seq)
+					for seq := maxSeq; seq >= fseq; seq-- {
+						sm, _ := mb.cacheLookup(seq, &smv)
+						if sm == nil || sm.subj != subj {
+							continue
+						}
+						seqs = append(seqs, sm.seq)
+						delete(subs, subj)
+						break
+					}
+				}
+			} else if mb.index <= psi.fblk {
+				// Track which subs are no longer applicable, meaning we will not find a valid msg at this point.
+				delete(subs, subj)
+			}
+			// TODO(dlc) we could track lblk like above in case some subs are very far apart.
+			// Not too bad if fss loaded since we will skip over quickly with it loaded, but might be worth it.
+		}
+		mb.mu.Unlock()
+
+		// If maxAllowed was sepcified check that we will not exceed that.
+		if maxAllowed > 0 && len(seqs) > maxAllowed {
+			return nil, ErrTooManyResults
+		}
+
+	}
+	if len(seqs) == 0 {
+		return nil, nil
+	}
+	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
+	return seqs, nil
+}
+
 // NumPending will return the number of pending messages matching the filter subject starting at sequence.
 // Optimized for stream num pending calculations for consumers.
 func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool) (total, validThrough uint64) {
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 8dcf6bad..7523138b 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -649,6 +649,13 @@ type JSApiMsgGetRequest struct {
 	// This will make sure we limit how much data we blast out. If not set we will
 	// inherit the slow consumer default max setting of the server. Default is MAX_PENDING_SIZE.
 	MaxBytes int `json:"max_bytes,omitempty"`
+
+	// Multiple response support. Will get the last msgs matching the subjects. These can include wildcards.
+	MultiLastFor []string `json:"multi_last,omitempty"`
+	// Only return messages up to this sequence. If not set, will be last sequence for the stream.
+	UpToSeq uint64 `json:"up_to_seq,omitempty"`
+	// Only return messages up to this time.
+	UpToTime *time.Time `json:"up_to_time,omitempty"`
 }
 
 type JSApiMsgGetResponse struct {
diff --git a/server/jwt.go b/server/jwt.go
index b9003265..feda5130 100644
--- a/server/jwt.go
+++ b/server/jwt.go
@@ -106,8 +106,8 @@ func validateTrustedOperators(o *Options) error {
 			return fmt.Errorf("using nats based account resolver - the system account needs to be specified in configuration or the operator jwt")
 		}
 	}
-	ver := strings.Split(strings.Split(strings.Split(VERSION, "-")[0], ".RC")[0], ".beta")[0]
-	srvMajor, srvMinor, srvUpdate, _ := jwt.ParseServerVersion(ver)
+
+	srvMajor, srvMinor, srvUpdate, _ := versionComponents(VERSION)
 	for _, opc := range o.TrustedOperators {
 		if major, minor, update, err := jwt.ParseServerVersion(opc.AssertServerVersion); err != nil {
 			return fmt.Errorf("operator %s expects version %s got error instead: %s",
diff --git a/server/memstore.go b/server/memstore.go
index 87fa5b82..0a046421 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -546,6 +546,54 @@ func (ms *memStore) SubjectsState(subject string) map[string]SimpleState {
 	return fss
 }
 
+func (ms *memStore) MultiLastSeqs(filters []string, maxSeq uint64, maxAllowed int) ([]uint64, error) {
+	ms.mu.RLock()
+	defer ms.mu.RUnlock()
+
+	if len(ms.msgs) == 0 {
+		return nil, nil
+	}
+
+	// Implied last sequence.
+	if maxSeq == 0 {
+		maxSeq = ms.state.LastSeq
+	}
+
+	//subs := make(map[string]*SimpleState)
+	seqs := make([]uint64, 0, 64)
+	seen := make(map[uint64]struct{})
+
+	addIfNotDupe := func(seq uint64) {
+		if _, ok := seen[seq]; !ok {
+			seqs = append(seqs, seq)
+			seen[seq] = struct{}{}
+		}
+	}
+
+	for _, filter := range filters {
+		ms.fss.Match(stringToBytes(filter), func(subj []byte, ss *SimpleState) {
+			if ss.Last <= maxSeq {
+				addIfNotDupe(ss.Last)
+			} else if ss.Msgs > 1 {
+				// The last is greater than maxSeq.
+				s := bytesToString(subj)
+				for seq := maxSeq; seq > 0; seq-- {
+					if sm, ok := ms.msgs[seq]; ok && sm.subj == s {
+						addIfNotDupe(seq)
+						break
+					}
+				}
+			}
+		})
+		// If maxAllowed was sepcified check that we will not exceed that.
+		if maxAllowed > 0 && len(seqs) > maxAllowed {
+			return nil, ErrTooManyResults
+		}
+	}
+	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
+	return seqs, nil
+}
+
 // SubjectsTotal return message totals per subject.
 func (ms *memStore) SubjectsTotals(filterSubject string) map[string]uint64 {
 	ms.mu.RLock()
diff --git a/server/store.go b/server/store.go
index 5c486b3e..e013ac1a 100644
--- a/server/store.go
+++ b/server/store.go
@@ -65,6 +65,8 @@ var (
 	ErrSequenceMismatch = errors.New("expected sequence does not match store")
 	// ErrCorruptStreamState
 	ErrCorruptStreamState = errors.New("stream state snapshot is corrupt")
+	// ErrTooManyResults
+	ErrTooManyResults = errors.New("too many matching results for request")
 )
 
 // StoreMsg is the stored message format for messages that are retained by the Store layer.
@@ -99,6 +101,7 @@ type StreamStore interface {
 	FilteredState(seq uint64, subject string) SimpleState
 	SubjectsState(filterSubject string) map[string]SimpleState
 	SubjectsTotals(filterSubject string) map[string]uint64
+	MultiLastSeqs(filters []string, maxSeq uint64, maxAllowed int) ([]uint64, error)
 	NumPending(sseq uint64, filter string, lastPerSubject bool) (total, validThrough uint64)
 	State() StreamState
 	FastState(*StreamState)
diff --git a/server/stream.go b/server/stream.go
index 405ee958..e5196979 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -355,6 +355,7 @@ const (
 	JSSubject      = "Nats-Subject"
 	JSLastSequence = "Nats-Last-Sequence"
 	JSNumPending   = "Nats-Num-Pending"
+	JSUpToSequence = "Nats-UpTo-Sequence"
 )
 
 // Rollups, can be subject only or all messages.
@@ -4027,7 +4028,7 @@ func (mset *stream) processDirectGetRequest(_ *subscription, c *client, _ *Accou
 		return
 	}
 	// Check if nothing set.
-	if req.Seq == 0 && req.LastFor == _EMPTY_ && req.NextFor == _EMPTY_ {
+	if req.Seq == 0 && req.LastFor == _EMPTY_ && req.NextFor == _EMPTY_ && len(req.MultiLastFor) == 0 {
 		hdr := []byte("NATS/1.0 408 Empty Request\r\n\r\n")
 		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 		return
@@ -4036,7 +4037,10 @@ func (mset *stream) processDirectGetRequest(_ *subscription, c *client, _ *Accou
 	// We do not allow batch mode for lastFor requests.
 	if (req.Seq > 0 && req.LastFor != _EMPTY_) ||
 		(req.LastFor != _EMPTY_ && req.NextFor != _EMPTY_) ||
-		(req.LastFor != _EMPTY_ && req.Batch > 0) {
+		(req.LastFor != _EMPTY_ && req.Batch > 0) ||
+		(req.LastFor != _EMPTY_ && len(req.MultiLastFor) > 0) ||
+		(req.NextFor != _EMPTY_ && len(req.MultiLastFor) > 0) ||
+		(req.UpToSeq > 0 && req.UpToTime != nil) {
 		hdr := []byte("NATS/1.0 408 Bad Request\r\n\r\n")
 		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 		return
@@ -4095,9 +4099,127 @@ func (mset *stream) processDirectGetLastBySubjectRequest(_ *subscription, c *cli
 	}
 }
 
+// For direct get batch and multi requests.
+const (
+	dg   = "NATS/1.0\r\nNats-Stream: %s\r\nNats-Subject: %s\r\nNats-Sequence: %d\r\nNats-Time-Stamp: %s\r\n\r\n"
+	dgb  = "NATS/1.0\r\nNats-Stream: %s\r\nNats-Subject: %s\r\nNats-Sequence: %d\r\nNats-Time-Stamp: %s\r\nNats-Num-Pending: %d\r\nNats-Last-Sequence: %d\r\n\r\n"
+	eob  = "NATS/1.0 204 EOB\r\nNats-Num-Pending: %d\r\nNats-Last-Sequence: %d\r\n\r\n"
+	eobm = "NATS/1.0 204 EOB\r\nNats-Num-Pending: %d\r\nNats-Last-Sequence: %d\r\nNats-UpTo-Sequence: %d\r\n\r\n"
+)
+
+// Handle a multi request.
+func (mset *stream) getDirectMulti(req *JSApiMsgGetRequest, reply string) {
+	// TODO(dlc) - Make configurable?
+	const maxAllowedResponses = 1024
+
+	// We hold the lock here to try to avoid changes out from underneath of us.
+	mset.mu.RLock()
+	defer mset.mu.RUnlock()
+	// Grab store and name.
+	store, name, s := mset.store, mset.cfg.Name, mset.srv
+
+	// Grab MaxBytes
+	mb := req.MaxBytes
+	if mb == 0 && s != nil {
+		// Fill in with the server's MaxPending.
+		mb = int(s.opts.MaxPending)
+	}
+
+	upToSeq := req.UpToSeq
+	// If we have UpToTime set get the proper sequence.
+	if req.UpToTime != nil {
+		upToSeq = store.GetSeqFromTime((*req.UpToTime).UTC())
+		// We need to back off one since this is used to determine start sequence normally,
+		// were as here we want it to be the ceiling.
+		upToSeq--
+	}
+	// If not set, set to the last sequence and remember that for EOB.
+	if upToSeq == 0 {
+		var state StreamState
+		mset.store.FastState(&state)
+		upToSeq = state.LastSeq
+	}
+
+	seqs, err := store.MultiLastSeqs(req.MultiLastFor, upToSeq, maxAllowedResponses)
+	if err != nil {
+		var hdr []byte
+		if err == ErrTooManyResults {
+			hdr = []byte("NATS/1.0 413 Too Many Results\r\n\r\n")
+		} else {
+			hdr = []byte(fmt.Sprintf("NATS/1.0 500 %v\r\n\r\n", err))
+		}
+		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+		return
+	}
+	if len(seqs) == 0 {
+		hdr := []byte("NATS/1.0 404 No Results\r\n\r\n")
+		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+		return
+	}
+
+	np, lseq, sentBytes, sent := uint64(len(seqs)-1), uint64(0), 0, 0
+	for _, seq := range seqs {
+		if seq < req.Seq {
+			if np > 0 {
+				np--
+			}
+			continue
+		}
+		var svp StoreMsg
+		sm, err := store.LoadMsg(seq, &svp)
+		if err != nil {
+			hdr := []byte("NATS/1.0 404 Message Not Found\r\n\r\n")
+			mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+			return
+		}
+
+		hdr := sm.hdr
+		ts := time.Unix(0, sm.ts).UTC()
+
+		if len(hdr) == 0 {
+			hdr = fmt.Appendf(nil, dgb, name, sm.subj, sm.seq, ts.Format(time.RFC3339Nano), np, lseq)
+		} else {
+			hdr = copyBytes(hdr)
+			hdr = genHeader(hdr, JSStream, name)
+			hdr = genHeader(hdr, JSSubject, sm.subj)
+			hdr = genHeader(hdr, JSSequence, strconv.FormatUint(sm.seq, 10))
+			hdr = genHeader(hdr, JSTimeStamp, ts.Format(time.RFC3339Nano))
+			hdr = genHeader(hdr, JSNumPending, strconv.FormatUint(np, 10))
+			hdr = genHeader(hdr, JSLastSequence, strconv.FormatUint(lseq, 10))
+		}
+		// Decrement num pending. This is optimization and we do not continue to look it up for these operations.
+		if np > 0 {
+			np--
+		}
+		// Track our lseq
+		lseq = sm.seq
+		// Send out our message.
+		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, sm.msg, nil, 0))
+		// Check if we have exceeded max bytes.
+		sentBytes += len(sm.subj) + len(sm.hdr) + len(sm.msg)
+		if sentBytes >= mb {
+			break
+		}
+		sent++
+		if req.Batch > 0 && sent >= req.Batch {
+			break
+		}
+	}
+
+	// Send out EOB
+	hdr := fmt.Appendf(nil, eobm, np, lseq, upToSeq)
+	mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
+}
+
 // Do actual work on a direct msg request.
 // This could be called in a Go routine if we are inline for a non-client connection.
 func (mset *stream) getDirectRequest(req *JSApiMsgGetRequest, reply string) {
+	// Handle multi in separate function.
+	if len(req.MultiLastFor) > 0 {
+		mset.getDirectMulti(req, reply)
+		return
+	}
+
 	mset.mu.RLock()
 	store, name, s := mset.store, mset.cfg.Name, mset.srv
 	mset.mu.RUnlock()
@@ -4204,7 +4326,7 @@ func (mset *stream) getDirectRequest(req *JSApiMsgGetRequest, reply string) {
 		if mset.lastSeq() > validThrough {
 			np, _ = store.NumPending(seq, req.NextFor, false)
 		}
-		hdr := fmt.Appendf(nil, "NATS/1.0 204 EOB\r\nNats-Num-Pending: %d\r\nNats-Last-Sequence: %d\r\n\r\n", np, lseq)
+		hdr := fmt.Appendf(nil, eob, np, lseq)
 		mset.outq.send(newJSPubMsg(reply, _EMPTY_, _EMPTY_, hdr, nil, nil, 0))
 	}
 }
diff --git a/server/util.go b/server/util.go
index aea3dcf1..250fdb34 100644
--- a/server/util.go
+++ b/server/util.go
@@ -23,7 +23,6 @@ import (
 	"net"
 	"net/url"
 	"reflect"
-	"regexp"
 	"strconv"
 	"strings"
 	"time"
@@ -39,11 +38,9 @@ const (
 	asciiNine = 57
 )
 
-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 {
+	if len(m) == 0 {
 		return 0, 0, 0, errors.New("invalid semver")
 	}
 	major, err = strconv.Atoi(m[1])


Test patch:
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
new file mode 100644
index 00000000..35b85b60
--- /dev/null
+++ b/.github/workflows/tests.yaml
@@ -0,0 +1,246 @@
+name: NATS Server Tests
+
+on:
+  # push:
+  # pull_request:
+  schedule:
+    - cron: "15 * * * *" # Hourly, 15 mins past the hour
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  lint:
+    name: Lint
+    runs-on: ${{ vars.GHA_WORKER_SMALL }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run golangci-lint
+        uses: golangci/golangci-lint-action@v4
+        with:
+          skip-cache: true
+          skip-pkg-cache: true
+          skip-build-cache: true
+          args: --timeout=5m --config=.golangci.yml
+
+  build-latest:
+    name: Build (Latest Go)
+    runs-on: ${{ vars.GHA_WORKER_SMALL }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Build NATS Server
+        run: go build
+
+  build-supported:
+    name: Build (Minimum Go)
+    runs-on: ${{ vars.GHA_WORKER_SMALL }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version-file: "go.mod"
+
+      - name: Build NATS Server
+        run: go build
+
+  # Using GitHub-supplied workers for Windows for now.
+  # Note that the below testing steps depend on the Linux build
+  # only, as the Windows builds take a fair bit longer to set up.
+  build-windows:
+    name: Build (Minimum Go, ${{ matrix.os }})
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [windows-2019, windows-2022]
+    runs-on: ${{ matrix.os }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version-file: "go.mod"
+
+      - name: Build NATS Server
+        run: go build
+
+  js-no-cluster:
+    name: JetStream tests
+    needs: [build-latest, build-supported, lint]
+    runs-on: ${{ vars.GHA_WORKER_LARGE }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run unit tests
+        run: go test -race -v -run=TestJetStream ./server/... -tags=skip_js_cluster_tests,skip_js_cluster_tests_2,skip_js_cluster_tests_3,skip_js_super_cluster_tests -count=1 -vet=off -timeout=30m -failfast
+
+  js-cluster-1:
+    name: JetStream Cluster tests (1)
+    needs: [build-latest, build-supported, lint]
+    runs-on: ${{ vars.GHA_WORKER_LARGE }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run unit tests
+        run: go test -race -v -run=TestJetStreamCluster ./server/... -tags=skip_js_cluster_tests_2,skip_js_cluster_tests_3 -count=1 -vet=off -timeout=30m -failfast
+
+  js-cluster-2:
+    name: JetStream Cluster tests (2)
+    needs: [build-latest, build-supported, lint]
+    runs-on: ${{ vars.GHA_WORKER_LARGE }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run unit tests
+        run: go test -race -v -run=TestJetStreamCluster ./server/... -tags=skip_js_cluster_tests,skip_js_cluster_tests_3 -count=1 -vet=off -timeout=30m -failfast
+
+  js-cluster-3:
+    name: JetStream Cluster tests (3)
+    needs: [build-latest, build-supported, lint]
+    runs-on: ${{ vars.GHA_WORKER_LARGE }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run unit tests
+        run: go test -race -v -run=TestJetStreamCluster ./server/... -tags=skip_js_cluster_tests,skip_js_cluster_tests_2 -count=1 -vet=off -timeout=30m -failfast
+
+  js-supercluster:
+    name: JetStream Supercluster tests
+    needs: [build-latest, build-supported, lint]
+    runs-on: ${{ vars.GHA_WORKER_LARGE }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run unit tests
+        run: go test -race -v -run=TestJetStreamSuperCluster ./server/... -count=1 -vet=off -timeout=30m -failfast
+
+  no-race:
+    name: No-race tests
+    needs: [build-latest, build-supported, lint]
+    runs-on: ${{ vars.GHA_WORKER_LARGE }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run unit tests
+        run: go test -v -p=1 -run=TestNoRace ./... -count=1 -vet=off -timeout=30m -failfast
+
+  js-chaos:
+    name: JetStream chaos tests
+    if: ${{ false }} # Don't run for now
+    needs: [build-latest, build-supported, lint]
+    runs-on: ${{ vars.GHA_WORKER_LARGE }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run unit tests
+        run: go test -race -v -p=1 -run=TestJetStreamChaos ./server/... -tags=js_chaos_tests -count=1 -vet=off -timeout=30m -failfast
+
+  mqtt:
+    name: MQTT tests
+    needs: [build-latest, build-supported, lint]
+    runs-on: ${{ vars.GHA_WORKER_MEDIUM }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run unit tests
+        run: go test -race -v -run=TestMQTT ./server/... -count=1 -vet=off -timeout=30m -failfast
+
+  server-pkg-non-js:
+    name: Non-JetStream/MQTT tests
+    needs: [build-latest, build-supported, lint]
+    runs-on: ${{ vars.GHA_WORKER_LARGE }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run unit tests
+        run: go test -race -v -p=1 ./server/... -tags=skip_js_tests,skip_mqtt_tests -count=1 -vet=off -timeout=30m -failfast
+        timeout-minutes: 15
+
+  non-server-pkg:
+    name: Tests from all other packages
+    needs: [build-latest, build-supported, lint]
+    runs-on: ${{ vars.GHA_WORKER_MEDIUM }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+
+      - name: Run unit tests
+        run: go test -race -v -p=1 ./conf/... ./internal/... ./logger/... ./test/... -count=1 -vet=off -timeout=30m -failfast
diff --git a/scripts/runTestsOnTravis.sh b/scripts/runTestsOnTravis.sh
index 53961a5b..d7c47983 100755
--- a/scripts/runTestsOnTravis.sh
+++ b/scripts/runTestsOnTravis.sh
@@ -7,7 +7,7 @@ if [ "$1" = "compile" ]; then
     go build;
 
     # Now run the linters.
-    go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.53.3;
+    go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.56.1;
     golangci-lint run;
     if [ "$TRAVIS_TAG" != "" ]; then
         go test -race -v -run=TestVersionMatchesTag ./server -count=1 -vet=off
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 8d1b06c5..12cd08aa 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -6617,6 +6617,107 @@ func TestFileStoreEraseMsgWithAllTrailingDbitSlots(t *testing.T) {
 	require_True(t, removed)
 }
 
+func TestFileStoreMultiLastSeqs(t *testing.T) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: t.TempDir(), BlockSize: 256}, // Make block size small to test multiblock selections with maxSeq
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := []byte("abc")
+	for i := 0; i < 33; i++ {
+		fs.StoreMsg("foo.foo", nil, msg)
+		fs.StoreMsg("foo.bar", nil, msg)
+		fs.StoreMsg("foo.baz", nil, msg)
+	}
+	for i := 0; i < 33; i++ {
+		fs.StoreMsg("bar.foo", nil, msg)
+		fs.StoreMsg("bar.bar", nil, msg)
+		fs.StoreMsg("bar.baz", nil, msg)
+	}
+
+	checkResults := func(seqs, expected []uint64) {
+		t.Helper()
+		if len(seqs) != len(expected) {
+			t.Fatalf("Expected %+v got %+v", expected, seqs)
+		}
+		for i := range seqs {
+			if seqs[i] != expected[i] {
+				t.Fatalf("Expected %+v got %+v", expected, seqs)
+			}
+		}
+	}
+
+	// UpTo sequence 3. Tests block split.
+	seqs, err := fs.MultiLastSeqs([]string{"foo.*"}, 3, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{1, 2, 3})
+	// Up to last sequence of the stream.
+	seqs, err = fs.MultiLastSeqs([]string{"foo.*"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{97, 98, 99})
+	// Check for bar.* at the end.
+	seqs, err = fs.MultiLastSeqs([]string{"bar.*"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{196, 197, 198})
+	// This should find nothing.
+	seqs, err = fs.MultiLastSeqs([]string{"bar.*"}, 99, -1)
+	require_NoError(t, err)
+	checkResults(seqs, nil)
+
+	// Do multiple subjects explicitly.
+	seqs, err = fs.MultiLastSeqs([]string{"foo.foo", "foo.bar", "foo.baz"}, 3, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{1, 2, 3})
+	seqs, err = fs.MultiLastSeqs([]string{"foo.foo", "foo.bar", "foo.baz"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{97, 98, 99})
+	seqs, err = fs.MultiLastSeqs([]string{"bar.foo", "bar.bar", "bar.baz"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{196, 197, 198})
+	seqs, err = fs.MultiLastSeqs([]string{"bar.foo", "bar.bar", "bar.baz"}, 99, -1)
+	require_NoError(t, err)
+	checkResults(seqs, nil)
+
+	// Check single works
+	seqs, err = fs.MultiLastSeqs([]string{"foo.foo"}, 3, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{1})
+
+	// Now test that we properly de-duplicate between filters.
+	seqs, err = fs.MultiLastSeqs([]string{"foo.*", "foo.bar"}, 3, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{1, 2, 3})
+	seqs, err = fs.MultiLastSeqs([]string{"bar.>", "bar.bar", "bar.baz"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{196, 197, 198})
+
+	// All
+	seqs, err = fs.MultiLastSeqs([]string{">"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{97, 98, 99, 196, 197, 198})
+	seqs, err = fs.MultiLastSeqs([]string{">"}, 99, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{97, 98, 99})
+}
+
+func TestFileStoreMultiLastSeqsMaxAllowed(t *testing.T) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: t.TempDir()},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := []byte("abc")
+	for i := 1; i <= 100; i++ {
+		fs.StoreMsg(fmt.Sprintf("foo.%d", i), nil, msg)
+	}
+	// Test that if we specify maxAllowed that we get the correct error.
+	seqs, err := fs.MultiLastSeqs([]string{"foo.*"}, 0, 10)
+	require_True(t, seqs == nil)
+	require_Error(t, err, ErrTooManyResults)
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index 0c2b0552..98cfed81 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -4632,6 +4632,30 @@ func TestJetStreamClusterStreamRemovePeer(t *testing.T) {
 	}
 	checkSubsPending(t, esub, toSend)
 
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "pull", AckPolicy: nats.AckNonePolicy})
+	require_NoError(t, err)
+
+	pullSub, err := js.PullSubscribe("TEST", "pull")
+	require_NoError(t, err)
+
+	// First fetch the messages that are already there.
+	msgs, err := pullSub.Fetch(toSend, nats.MaxWait(500*time.Millisecond))
+	require_NoError(t, err)
+	require_Equal(t, toSend, len(msgs))
+
+	// Now prepare a check to see if we get unwated `Consumer Deleted` error on peer remove.
+	pullResults := make(chan error, 1)
+	go func() {
+		_, err := pullSub.Fetch(1, nats.MaxWait(30*time.Second))
+		// Let's check if we get unwted `Consumer Deleted` error on peer remove.
+		// Everything else is fine (Leader Changed, Timeout, etc.)
+		if err != nats.ErrConsumerDeleted {
+			close(pullResults)
+		} else {
+			pullResults <- err
+		}
+	}()
+
 	ci, err := esub.ConsumerInfo()
 	if err != nil {
 		t.Fatalf("Could not fetch consumer info: %v", err)
@@ -4716,6 +4740,7 @@ func TestJetStreamClusterStreamRemovePeer(t *testing.T) {
 	})
 
 	c.waitOnConsumerLeader("$G", "TEST", "cat")
+	c.waitOnConsumerLeader("$G", "TEST", "pull")
 
 	// Now check consumer info as well.
 	checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
@@ -4742,6 +4767,15 @@ func TestJetStreamClusterStreamRemovePeer(t *testing.T) {
 		return nil
 	})
 
+	// Check if we got the `Consumer Deleted` error on the pull consumer.
+	select {
+	case err := <-pullResults:
+		if err != nil {
+			t.Fatalf("Expected timeout error or nil, got %v", err)
+		}
+	default:
+	}
+
 	// Now check ephemeral consumer info.
 	// Make sure we did not stamp same new group into the ephemeral where R=1.
 	ci, err = esub.ConsumerInfo()
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 032e49cd..8c6730cf 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -6592,6 +6592,7 @@ func TestJetStreamClusterConsumerPauseViaConfig(t *testing.T) {
 
 	// Now we're going to set the deadline.
 	deadline := jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", time.Now().Add(time.Second*3))
+	c.waitOnAllCurrent()
 
 	// It will now take longer than 3 seconds.
 	publish(time.Second * 5)
@@ -6608,6 +6609,7 @@ func TestJetStreamClusterConsumerPauseViaConfig(t *testing.T) {
 	// Now we're going to do an update and then immediately kick the
 	// leader. The pause should still be in effect afterwards.
 	deadline = jsTestPause_PauseConsumer(t, nc, "TEST", "my_consumer", time.Now().Add(time.Second*3))
+	c.waitOnAllCurrent()
 	publish(time.Second * 5)
 	require_True(t, time.Now().After(deadline))
 
@@ -6651,6 +6653,7 @@ func TestJetStreamClusterConsumerPauseViaEndpoint(t *testing.T) {
 		// Now we'll pause the consumer for 3 seconds.
 		deadline := time.Now().Add(time.Second * 3)
 		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "pull_consumer", deadline).Equal(deadline))
+		c.waitOnAllCurrent()
 
 		// This should fail as we'll wait for only half of the deadline.
 		for i := 0; i < 10; i++ {
@@ -6678,6 +6681,7 @@ func TestJetStreamClusterConsumerPauseViaEndpoint(t *testing.T) {
 		require_Equal(t, len(msgs), 10)
 
 		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "pull_consumer", time.Time{}).Equal(time.Time{}))
+		c.waitOnAllCurrent()
 
 		// This should succeed as there's no pause, so it definitely
 		// shouldn't take more than a second.
@@ -6709,6 +6713,7 @@ func TestJetStreamClusterConsumerPauseViaEndpoint(t *testing.T) {
 		// Now we'll pause the consumer for 3 seconds.
 		deadline := time.Now().Add(time.Second * 3)
 		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "push_consumer", deadline).Equal(deadline))
+		c.waitOnAllCurrent()
 
 		// This should succeed after a short wait, and when we're done,
 		// we should be after the deadline.
@@ -6734,6 +6739,7 @@ func TestJetStreamClusterConsumerPauseViaEndpoint(t *testing.T) {
 		}
 
 		require_True(t, jsTestPause_PauseConsumer(t, nc, "TEST", "push_consumer", time.Time{}).Equal(time.Time{}))
+		c.waitOnAllCurrent()
 
 		// This should succeed as there's no pause, so it definitely
 		// shouldn't take more than a second.
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 89dbedec..8b5d068e 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2023 The NATS Authors
+// Copyright 2019-2024 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
@@ -22740,3 +22740,341 @@ func TestJetStreamConsumerSurvivesRestart(t *testing.T) {
 	consumer.mu.RUnlock()
 	require_True(t, timer != nil)
 }
+
+func TestJetStreamDirectGetMulti(t *testing.T) {
+	cases := []struct {
+		name string
+		cfg  *nats.StreamConfig
+	}{
+		{name: "MemoryStore",
+			cfg: &nats.StreamConfig{
+				Name:        "TEST",
+				Subjects:    []string{"foo.*"},
+				AllowDirect: true,
+				Storage:     nats.MemoryStorage,
+			}},
+		{name: "FileStore",
+			cfg: &nats.StreamConfig{
+				Name:        "TEST",
+				Subjects:    []string{"foo.*"},
+				AllowDirect: true,
+			}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+
+			s := RunBasicJetStreamServer(t)
+			defer s.Shutdown()
+
+			nc, js := jsClientConnect(t, s)
+			defer nc.Close()
+
+			_, err := js.AddStream(c.cfg)
+			require_NoError(t, err)
+
+			// Add in messages
+			for i := 0; i < 33; i++ {
+				js.PublishAsync("foo.foo", []byte(fmt.Sprintf("HELLO-%d", i)))
+				js.PublishAsync("foo.bar", []byte(fmt.Sprintf("WORLD-%d", i)))
+				js.PublishAsync("foo.baz", []byte(fmt.Sprintf("AGAIN-%d", i)))
+			}
+			select {
+			case <-js.PublishAsyncComplete():
+			case <-time.After(5 * time.Second):
+				t.Fatalf("Did not receive completion signal")
+			}
+
+			// Direct subjects.
+			sendRequest := func(mreq *JSApiMsgGetRequest) *nats.Subscription {
+				t.Helper()
+				req, _ := json.Marshal(mreq)
+				// We will get multiple responses so can't do normal request.
+				reply := nats.NewInbox()
+				sub, err := nc.SubscribeSync(reply)
+				require_NoError(t, err)
+				err = nc.PublishRequest("$JS.API.DIRECT.GET.TEST", reply, req)
+				require_NoError(t, err)
+				return sub
+			}
+
+			// Subject / Sequence pair
+			type p struct {
+				subj string
+				seq  int
+			}
+			var eob p
+
+			// Multi-Get will have a nil message as the end marker regardless.
+			checkResponses := func(sub *nats.Subscription, numPendingStart int, expected ...p) {
+				t.Helper()
+				defer sub.Unsubscribe()
+				checkSubsPending(t, sub, len(expected))
+				np := numPendingStart
+				for i := 0; i < len(expected); i++ {
+					msg, err := sub.NextMsg(10 * time.Millisecond)
+					require_NoError(t, err)
+					// If expected is _EMPTY_ that signals we expect a EOB marker.
+					if subj := expected[i].subj; subj != _EMPTY_ {
+						// Make sure subject is correct.
+						require_Equal(t, subj, msg.Header.Get(JSSubject))
+						// Make sure sequence is correct.
+						require_Equal(t, strconv.Itoa(expected[i].seq), msg.Header.Get(JSSequence))
+						// Should have Data field non-zero
+						require_True(t, len(msg.Data) > 0)
+						// Check we have NumPending and its correct.
+						require_Equal(t, strconv.Itoa(np), msg.Header.Get(JSNumPending))
+						if np > 0 {
+							np--
+						}
+					} else {
+						// Check for properly formatted EOB marker.
+						// Should have no body.
+						require_Equal(t, len(msg.Data), 0)
+						// We mark status as 204 - No Content
+						require_Equal(t, msg.Header.Get("Status"), "204")
+						// Check description is EOB
+						require_Equal(t, msg.Header.Get("Description"), "EOB")
+						// Check we have NumPending and its correct.
+						require_Equal(t, strconv.Itoa(np), msg.Header.Get(JSNumPending))
+					}
+				}
+			}
+
+			sub := sendRequest(&JSApiMsgGetRequest{MultiLastFor: []string{"foo.*"}})
+			checkResponses(sub, 2, p{"foo.foo", 97}, p{"foo.bar", 98}, p{"foo.baz", 99}, eob)
+			// Check with UpToSeq
+			sub = sendRequest(&JSApiMsgGetRequest{MultiLastFor: []string{"foo.*"}, UpToSeq: 3})
+			checkResponses(sub, 2, p{"foo.foo", 1}, p{"foo.bar", 2}, p{"foo.baz", 3}, eob)
+
+			// Test No Results.
+			sub = sendRequest(&JSApiMsgGetRequest{MultiLastFor: []string{"bar.*"}})
+			checkSubsPending(t, sub, 1)
+			msg, err := sub.NextMsg(10 * time.Millisecond)
+			require_NoError(t, err)
+			// Check for properly formatted No Results.
+			// Should have no body.
+			require_Equal(t, len(msg.Data), 0)
+			// We mark status as 204 - No Content
+			require_Equal(t, msg.Header.Get("Status"), "404")
+			// Check description is No Results
+			require_Equal(t, msg.Header.Get("Description"), "No Results")
+		})
+	}
+}
+
+func TestJetStreamDirectGetMultiUpToTime(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:        "TEST",
+		Subjects:    []string{"foo.*"},
+		AllowDirect: true,
+	})
+	require_NoError(t, err)
+
+	js.Publish("foo.foo", []byte("1"))
+	js.Publish("foo.bar", []byte("1"))
+	js.Publish("foo.baz", []byte("1"))
+	start := time.Now()
+	time.Sleep(time.Second)
+	js.Publish("foo.foo", []byte("2"))
+	js.Publish("foo.bar", []byte("2"))
+	js.Publish("foo.baz", []byte("2"))
+	mid := time.Now()
+	time.Sleep(time.Second)
+	js.Publish("foo.foo", []byte("3"))
+	js.Publish("foo.bar", []byte("3"))
+	js.Publish("foo.baz", []byte("3"))
+	end := time.Now()
+
+	// Direct subjects.
+	sendRequest := func(mreq *JSApiMsgGetRequest) *nats.Subscription {
+		t.Helper()
+		req, _ := json.Marshal(mreq)
+		// We will get multiple responses so can't do normal request.
+		reply := nats.NewInbox()
+		sub, err := nc.SubscribeSync(reply)
+		require_NoError(t, err)
+		err = nc.PublishRequest("$JS.API.DIRECT.GET.TEST", reply, req)
+		require_NoError(t, err)
+		return sub
+	}
+
+	checkResponses := func(sub *nats.Subscription, val string, expected ...string) {
+		t.Helper()
+		defer sub.Unsubscribe()
+		checkSubsPending(t, sub, len(expected))
+		for i := 0; i < len(expected); i++ {
+			msg, err := sub.NextMsg(10 * time.Millisecond)
+			require_NoError(t, err)
+			// If expected is _EMPTY_ that signals we expect a EOB marker.
+			if subj := expected[i]; subj != _EMPTY_ {
+				// Make sure subject is correct.
+				require_Equal(t, subj, msg.Header.Get(JSSubject))
+				// Should have Data field non-zero
+				require_True(t, len(msg.Data) > 0)
+				// Make sure the value matches.
+				require_Equal(t, string(msg.Data), val)
+			}
+		}
+	}
+
+	// Make sure you can't set both.
+	sub := sendRequest(&JSApiMsgGetRequest{Seq: 1, MultiLastFor: []string{"foo.*"}, UpToSeq: 3, UpToTime: &start})
+	checkSubsPending(t, sub, 1)
+	msg, err := sub.NextMsg(10 * time.Millisecond)
+	require_NoError(t, err)
+	// Check for properly formatted No Results.
+	// Should have no body.
+	require_Equal(t, len(msg.Data), 0)
+	// We mark status as 204 - No Content
+	require_Equal(t, msg.Header.Get("Status"), "408")
+	// Check description is No Results
+	require_Equal(t, msg.Header.Get("Description"), "Bad Request")
+
+	// Valid responses.
+	sub = sendRequest(&JSApiMsgGetRequest{Seq: 1, MultiLastFor: []string{"foo.*"}, UpToTime: &start})
+	checkResponses(sub, "1", "foo.foo", "foo.bar", "foo.baz", _EMPTY_)
+
+	sub = sendRequest(&JSApiMsgGetRequest{Seq: 1, MultiLastFor: []string{"foo.*"}, UpToTime: &mid})
+	checkResponses(sub, "2", "foo.foo", "foo.bar", "foo.baz", _EMPTY_)
+
+	sub = sendRequest(&JSApiMsgGetRequest{Seq: 1, MultiLastFor: []string{"foo.*"}, UpToTime: &end})
+	checkResponses(sub, "3", "foo.foo", "foo.bar", "foo.baz", _EMPTY_)
+}
+
+func TestJetStreamDirectGetMultiMaxAllowed(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:        "TEST",
+		Subjects:    []string{"foo.*"},
+		AllowDirect: true,
+	})
+	require_NoError(t, err)
+
+	// from stream.go - const maxAllowedResponses = 1024, so max sure > 1024
+	// Add in messages
+	for i := 1; i <= 1025; i++ {
+		js.PublishAsync(fmt.Sprintf("foo.%d", i), []byte("OK"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	req, _ := json.Marshal(&JSApiMsgGetRequest{Seq: 1, MultiLastFor: []string{"foo.*"}})
+	msg, err := nc.Request("$JS.API.DIRECT.GET.TEST", req, time.Second)
+	require_NoError(t, err)
+
+	// Check for properly formatted Too Many Results error.
+	// Should have no body.
+	require_Equal(t, len(msg.Data), 0)
+	// We mark status as 413 - Too Many Results
+	require_Equal(t, msg.Header.Get("Status"), "413")
+	// Check description is No Results
+	require_Equal(t, msg.Header.Get("Description"), "Too Many Results")
+}
+
+func TestJetStreamDirectGetMultiPaging(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:        "TEST",
+		Subjects:    []string{"foo.*"},
+		AllowDirect: true,
+	})
+	require_NoError(t, err)
+
+	// We will queue up 500 messages, each 512k big and request them for a multi-get.
+	// This will not hit the max allowed limit of 1024, but will bump up against max bytes and only return partial results.
+	// We want to make sure we can pick up where we left off.
+
+	// Add in messages
+	data, sent := bytes.Repeat([]byte("Z"), 512*1024), 500
+	for i := 1; i <= sent; i++ {
+		js.PublishAsync(fmt.Sprintf("foo.%d", i), data)
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+	// Wait for all replicas to be correct.
+	time.Sleep(time.Second)
+
+	// Direct subjects.
+	sendRequest := func(mreq *JSApiMsgGetRequest) *nats.Subscription {
+		t.Helper()
+		req, _ := json.Marshal(mreq)
+		// We will get multiple responses so can't do normal request.
+		reply := nats.NewInbox()
+		sub, err := nc.SubscribeSync(reply)
+		require_NoError(t, err)
+		err = nc.PublishRequest("$JS.API.DIRECT.GET.TEST", reply, req)
+		require_NoError(t, err)
+		return sub
+	}
+
+	// Setup variables that control procesPartial
+	start, seq, np, b, bsz := 1, 1, sent-1, 0, 128
+
+	processPartial := func(expected int) {
+		t.Helper()
+		sub := sendRequest(&JSApiMsgGetRequest{Seq: uint64(start), Batch: b, MultiLastFor: []string{"foo.*"}})
+		checkSubsPending(t, sub, expected)
+		// Check partial.
+		// We should receive seqs seq-(seq+bsz-1)
+		for ; seq < start+(expected-1); seq++ {
+			msg, err := sub.NextMsg(10 * time.Millisecond)
+			require_NoError(t, err)
+			// Make sure sequence is correct.
+			require_Equal(t, strconv.Itoa(int(seq)), msg.Header.Get(JSSequence))
+			// Check we have NumPending and its correct.
+			require_Equal(t, strconv.Itoa(int(np)), msg.Header.Get(JSNumPending))
+			if np > 0 {
+				np--
+			}
+		}
+		// Now check EOB
+		msg, err := sub.NextMsg(10 * time.Millisecond)
+		require_NoError(t, err)
+		// We mark status as 204 - No Content
+		require_Equal(t, msg.Header.Get("Status"), "204")
+		// Check description is EOB
+		require_Equal(t, msg.Header.Get("Description"), "EOB")
+		// Check we have NumPending and its correct.
+		require_Equal(t, strconv.Itoa(np), msg.Header.Get(JSNumPending))
+		// Check we have LastSequence and its correct.
+		require_Equal(t, strconv.Itoa(seq-1), msg.Header.Get(JSLastSequence))
+		// Check we have UpToSequence and its correct.
+		require_Equal(t, strconv.Itoa(sent), msg.Header.Get(JSUpToSequence))
+		// Update start
+		start = seq
+	}
+
+	processPartial(bsz + 1) // 128 + EOB
+	processPartial(bsz + 1) // 128 + EOB
+	processPartial(bsz + 1) // 128 + EOB
+	// Last one will be a partial block.
+	processPartial(116 + 1)
+
+	// Now reset and test that batch is honored as well.
+	start, seq, np, b = 1, 1, sent-1, 100
+	for i := 0; i < 5; i++ {
+		processPartial(b + 1) // 100 + EOB
+	}
+}
diff --git a/server/memstore_test.go b/server/memstore_test.go
index 34dae8db..e6e6e1d3 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -837,6 +837,7 @@ func TestMemStoreGetSeqFromTimeWithLastDeleted(t *testing.T) {
 	}
 	ms, err := newMemStore(cfg)
 	require_NoError(t, err)
+	defer ms.Stop()
 
 	// Put in 1000 msgs.
 	total := 1000
@@ -868,6 +869,7 @@ func TestMemStoreSkipMsgs(t *testing.T) {
 	}
 	ms, err := newMemStore(cfg)
 	require_NoError(t, err)
+	defer ms.Stop()
 
 	// Test on empty FS first.
 	// Make sure wrong starting sequence fails.
@@ -910,3 +912,110 @@ func TestMemStoreSkipMsgs(t *testing.T) {
 	require_Equal(t, state.Msgs, 1)
 	require_Equal(t, state.NumDeleted, 10)
 }
+
+func TestMemStoreMultiLastSeqs(t *testing.T) {
+	cfg := &StreamConfig{
+		Name:     "zzz",
+		Subjects: []string{"foo.*"},
+		Storage:  MemoryStorage,
+	}
+	ms, err := newMemStore(cfg)
+	require_NoError(t, err)
+	defer ms.Stop()
+
+	msg := []byte("abc")
+	for i := 0; i < 33; i++ {
+		ms.StoreMsg("foo.foo", nil, msg)
+		ms.StoreMsg("foo.bar", nil, msg)
+		ms.StoreMsg("foo.baz", nil, msg)
+	}
+	for i := 0; i < 33; i++ {
+		ms.StoreMsg("bar.foo", nil, msg)
+		ms.StoreMsg("bar.bar", nil, msg)
+		ms.StoreMsg("bar.baz", nil, msg)
+	}
+
+	checkResults := func(seqs, expected []uint64) {
+		t.Helper()
+		if len(seqs) != len(expected) {
+			t.Fatalf("Expected %+v got %+v", expected, seqs)
+		}
+		for i := range seqs {
+			if seqs[i] != expected[i] {
+				t.Fatalf("Expected %+v got %+v", expected, seqs)
+			}
+		}
+	}
+
+	// UpTo sequence 3. Tests block split.
+	seqs, err := ms.MultiLastSeqs([]string{"foo.*"}, 3, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{1, 2, 3})
+	// Up to last sequence of the stream.
+	seqs, err = ms.MultiLastSeqs([]string{"foo.*"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{97, 98, 99})
+	// Check for bar.* at the end.
+	seqs, err = ms.MultiLastSeqs([]string{"bar.*"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{196, 197, 198})
+	// This should find nothing.
+	seqs, err = ms.MultiLastSeqs([]string{"bar.*"}, 99, -1)
+	require_NoError(t, err)
+	checkResults(seqs, nil)
+
+	// Do multiple subjects explicitly.
+	seqs, err = ms.MultiLastSeqs([]string{"foo.foo", "foo.bar", "foo.baz"}, 3, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{1, 2, 3})
+	seqs, err = ms.MultiLastSeqs([]string{"foo.foo", "foo.bar", "foo.baz"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{97, 98, 99})
+	seqs, err = ms.MultiLastSeqs([]string{"bar.foo", "bar.bar", "bar.baz"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{196, 197, 198})
+	seqs, err = ms.MultiLastSeqs([]string{"bar.foo", "bar.bar", "bar.baz"}, 99, -1)
+	require_NoError(t, err)
+	checkResults(seqs, nil)
+
+	// Check single works
+	seqs, err = ms.MultiLastSeqs([]string{"foo.foo"}, 3, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{1})
+
+	// Now test that we properly de-duplicate between filters.
+	seqs, err = ms.MultiLastSeqs([]string{"foo.*", "foo.bar"}, 3, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{1, 2, 3})
+	seqs, err = ms.MultiLastSeqs([]string{"bar.>", "bar.bar", "bar.baz"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{196, 197, 198})
+
+	// All
+	seqs, err = ms.MultiLastSeqs([]string{">"}, 0, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{97, 98, 99, 196, 197, 198})
+	seqs, err = ms.MultiLastSeqs([]string{">"}, 99, -1)
+	require_NoError(t, err)
+	checkResults(seqs, []uint64{97, 98, 99})
+}
+
+func TestMemStoreMultiLastSeqsMaxAllowed(t *testing.T) {
+	cfg := &StreamConfig{
+		Name:     "zzz",
+		Subjects: []string{"foo.*"},
+		Storage:  MemoryStorage,
+	}
+	ms, err := newMemStore(cfg)
+	require_NoError(t, err)
+	defer ms.Stop()
+
+	msg := []byte("abc")
+	for i := 1; i <= 100; i++ {
+		ms.StoreMsg(fmt.Sprintf("foo.%d", i), nil, msg)
+	}
+	// Test that if we specify maxAllowed that we get the correct error.
+	seqs, err := ms.MultiLastSeqs([]string{"foo.*"}, 0, 10)
+	require_True(t, seqs == nil)
+	require_Error(t, err, ErrTooManyResults)
+}
diff --git a/server/server_test.go b/server/server_test.go
index 42b650ac..e2227f9d 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -113,6 +113,12 @@ func RunServerWithConfig(configFile string) (srv *Server, opts *Options) {
 	return
 }
 
+func TestSemanticVersion(t *testing.T) {
+	if !semVerRe.MatchString(VERSION) {
+		t.Fatalf("Version (%s) is not a valid SemVer string", VERSION)
+	}
+}
+
 func TestVersionMatchesTag(t *testing.T) {
 	tag := os.Getenv("TRAVIS_TAG")
 	// Travis started to return '' when no tag is set. Support both now.